Advanced Service Communication and Resilience Patterns in Cipher Horizon

@Injectable()
class CircuitBreaker {
    private state: 'CLOSED' | 'OPEN' | 'HALF_OPEN' = 'CLOSED';
    private failureCount: number = 0;
    private lastFailureTime?: Date;
    private readonly metrics: CircuitMetrics;

    constructor(
        private readonly config: CircuitBreakerConfig,
        private readonly logger: Logger
    ) {
        this.metrics = new CircuitMetrics();
    }

    async execute<T>(
        operation: () => Promise<T>,
        fallback?: () => Promise<T>
    ): Promise<T> {
        if (this.isOpen()) {
            return this.handleOpenCircuit(fallback);
        }

        try {
            const result = await this.executeWithTimeout(operation);
            this.onSuccess();
            return result;
        } catch (error) {
            return this.handleFailure(error, fallback);
        }
    }
}

@Injectable()
class MessageQueue {
    constructor(
        private readonly redis: Redis,
        private readonly config: QueueConfig,
        private readonly metrics: QueueMetrics
    ) {}

    async publish<T>(
        topic: string,
        message: T,
        options: PublishOptions = {}
    ): Promise<void> {
        const messageId = uuid();
        const envelope = this.createEnvelope(messageId, message, options);

        await this.storeAndTrack(topic, envelope);
        this.metrics.recordPublish(topic);
    }

    private async storeAndTrack(
        topic: string,
        envelope: MessageEnvelope
    ): Promise<void> {
        const multi = this.redis.multi();

        multi.zadd(
            this.getQueueKey(topic),
            Date.now(),
            JSON.stringify(envelope)
        );

        multi.hset(
            this.getProcessingKey(topic),
            envelope.id,
            JSON.stringify({
                attempts: 0,
                firstAttempt: Date.now()
            })
        );

        await multi.exec();
    }
}

const queueConfig = {
    retryStrategy: {
        maxRetries: 3,
        baseDelay: 1000,  // 1 second
        maxDelay: 30000,  // 30 seconds
        jitterFactor: 0.1
    },
    monitoring: {
        metricsInterval: 60000,  // 1 minute
        alertThresholds: {
            errorRate: 0.05,     // 5%
            processingTime: 5000  // 5 seconds
        }
    }
};

@Injectable()
class DistributedLock {
    async acquireLock(
        resource: string,
        options: LockOptions = {}
    ): Promise<Lock | null> {
        const lockId = uuid();
        const acquired = await this.redis.set(
            this.getLockKey(resource),
            lockId,
            'NX',
            'PX',
            options.ttl || this.config.defaultTTL
        );

        if (!acquired) {
            return null;
        }

        return this.createLockObject(resource, lockId, options);
    }

    private async extendLock(
        resource: string,
        lockId: string
    ): Promise<boolean> {
        const result = await this.redis.eval(
            `
            if redis.call("get", KEYS[1]) == ARGV[1] then
                return redis.call("pexpire", KEYS[1], ARGV[2])
            else
                return 0
            end
            `,
            1,
            this.getLockKey(resource),
            lockId,
            this.config.defaultTTL
        );

        return result === 1;
    }
}

const circuitBreakerConfig = {
    failureThreshold: 5,    // Number of failures before opening
    resetTimeout: 30000,    // 30 seconds cool-down period
    monitorWindow: 60000,   // 1 minute rolling window
    healthCheckInterval: 5000 // 5 seconds between health checks
};

const reliabilityConfig = {
    persistence: true,
    acknowledgment: 'explicit',
    deadLetterExchange: 'dlx.cipher',
    messageExpiration: 86400000, // 24 hours
    queuePrefetch: 10
};