// ПРОБЛЕМНЫЙ КОД
val policy = savePolicy() // Транзакция коммитится
sendEventToRabbit()       // Может упасть - событие потеряется

Бизнес-сервис → [Транзакция: Данные + Событие в Outbox] → Processor → RabbitMQ
                                ↓
                    [Optimistic Locking между инстансами]

CREATE TABLE outbox_events (
    id BIGSERIAL PRIMARY KEY,
    event_id VARCHAR(255) NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    aggregate_id VARCHAR(255) NOT NULL,
    aggregate_type VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    status VARCHAR(20) DEFAULT 'PENDING',
    version INTEGER DEFAULT 0,  -- для Optimistic Locking
    processing_instance VARCHAR(255) NULL,
    locked_at TIMESTAMP NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    processed_at TIMESTAMP NULL,
    next_retry_at TIMESTAMP NULL,
    retry_count INTEGER DEFAULT 0,
    error_message TEXT NULL,
    
    INDEX idx_status_next_retry (status, next_retry_at),
    INDEX idx_pending_ready (status, next_retry_at) 
    WHERE status = 'PENDING' AND (next_retry_at IS NULL OR next_retry_at <= NOW())
);

@Entity
@Table(name = "outbox_events")
data class OutboxEvent(
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    val id: Long? = null,
    
    @Column(name = "event_id", nullable = false)
    val eventId: String,
    
    @Column(name = "event_type", nullable = false)
    val eventType: String,
    
    @Column(name = "aggregate_id", nullable = false)
    val aggregateId: String,
    
    @Column(name = "aggregate_type", nullable = false)
    val aggregateType: String,
    
    @Column(name = "payload", nullable = false)
    val payload: String,
    
    @Column(name = "status", nullable = false)
    @Enumerated(EnumType.STRING)
    var status: OutboxEventStatus = OutboxEventStatus.PENDING,
    
    @Version
    var version: Int = 0,  // JPA автоматически управляет версией
    
    @Column(name = "processing_instance")
    var processingInstance: String? = null,
    
    @Column(name = "locked_at")
    var lockedAt: LocalDateTime? = null,
    
    @Column(name = "created_at", nullable = false)
    val createdAt: LocalDateTime = LocalDateTime.now(),
    
    @Column(name = "processed_at")
    var processedAt: LocalDateTime? = null,
    
    @Column(name = "next_retry_at")
    var nextRetryAt: LocalDateTime? = null,
    
    @Column(name = "retry_count", nullable = false)
    var retryCount: Int = 0,
    
    @Column(name = "error_message")
    var errorMessage: String? = null
)

enum class OutboxEventStatus {
    PENDING, PROCESSING, COMPLETED, FAILED
}

@Repository
interface OutboxEventRepository : JpaRepository<OutboxEvent, Long> {
    
    @Query("""
        SELECT e FROM OutboxEvent e 
        WHERE e.status = 'PENDING' 
        AND (e.nextRetryAt IS NULL OR e.nextRetryAt <= :now)
        AND e.processingInstance IS NULL
        ORDER BY e.createdAt ASC
        LIMIT :batchSize
    """)
    fun findPendingEvents(now: LocalDateTime, batchSize: Int = 10): List<OutboxEvent>
    
    @Modifying
    @Query("""
        UPDATE OutboxEvent 
        SET processingInstance = :instanceId, lockedAt = :lockedAt, 
            status = 'PROCESSING', version = version + 1
        WHERE id = :id AND version = :currentVersion AND processingInstance IS NULL
    """)
    fun claimEvent(
        id: Long, 
        currentVersion: Int, 
        instanceId: String, 
        lockedAt: LocalDateTime
    ): Int
    
    @Modifying
    @Query("""
        UPDATE OutboxEvent 
        SET status = 'COMPLETED', processedAt = :processedAt, version = version + 1
        WHERE id = :id AND version = :currentVersion
    """)
    fun markAsCompleted(id: Long, currentVersion: Int, processedAt: LocalDateTime): Int
    
    @Modifying
    @Query("""
        UPDATE OutboxEvent 
        SET status = 'PENDING', processingInstance = NULL, lockedAt = NULL,
            retryCount = :retryCount, nextRetryAt = :nextRetryAt, 
            errorMessage = :errorMessage, version = version + 1
        WHERE id = :id AND version = :currentVersion
    """)
    fun scheduleRetry(
        id: Long, 
        currentVersion: Int,
        retryCount: Int, 
        nextRetryAt: LocalDateTime,
        errorMessage: String
    ): Int
    
    @Modifying
    @Query("""
        UPDATE OutboxEvent 
        SET status = 'FAILED', retryCount = :retryCount, 
            errorMessage = :errorMessage, version = version + 1
        WHERE id = :id AND version = :currentVersion
    """)
    fun markAsFailed(id: Long, currentVersion: Int, errorMessage: String, retryCount: Int): Int
    
    @Modifying
    @Query("""
        UPDATE OutboxEvent 
        SET processingInstance = NULL, lockedAt = NULL, status = 'PENDING', version = version + 1
        WHERE status = 'PROCESSING' AND lockedAt < :staleBefore
    """)
    fun releaseStaleEvents(staleBefore: LocalDateTime): Int
}

@Service
@Transactional
class OutboxService(
    private val outboxRepository: OutboxEventRepository,
    private val objectMapper: ObjectMapper
) {
    
    suspend fun saveEvent(
        eventId: String,
        eventType: String,
        aggregateId: String,
        aggregateType: String,
        payload: Any
    ) {
        val outboxEvent = OutboxEvent(
            eventId = eventId,
            eventType = eventType,
            aggregateId = aggregateId,
            aggregateType = aggregateType,
            payload = objectMapper.writeValueAsString(payload)
        )
        outboxRepository.save(outboxEvent)
    }
    
    suspend fun getPendingEvents(): List<OutboxEvent> {
        return outboxRepository.findPendingEvents(LocalDateTime.now(), 10)
    }
    
    suspend fun claimEvent(event: OutboxEvent, instanceId: String): Boolean {
        val updated = outboxRepository.claimEvent(
            event.id!!, 
            event.version,
            instanceId,
            LocalDateTime.now()
        )
        return updated > 0
    }
    
    suspend fun markAsCompleted(event: OutboxEvent): Boolean {
        val updated = outboxRepository.markAsCompleted(
            event.id!!, 
            event.version + 1, // версия увеличилась при claim
            LocalDateTime.now()
        )
        return updated > 0
    }
    
    suspend fun scheduleRetry(
        event: OutboxEvent, 
        retryCount: Int, 
        nextRetryAt: LocalDateTime,
        errorMessage: String
    ): Boolean {
        val updated = outboxRepository.scheduleRetry(
            event.id!!,
            event.version + 1,
            retryCount,
            nextRetryAt,
            errorMessage
        )
        return updated > 0
    }
    
    suspend fun markAsFailed(
        event: OutboxEvent, 
        errorMessage: String, 
        retryCount: Int
    ): Boolean {
        val updated = outboxRepository.markAsFailed(
            event.id!!, 
            event.version + 1,
            errorMessage, 
            retryCount
        )
        return updated > 0
    }
}

@Component
class OutboxProcessor(
    private val outboxService: OutboxService,
    private val rabbitService: RabbitService,
    private val objectMapper: ObjectMapper
) {
    private val logger = logger { }
    private val instanceId = InetAddress.getLocalHost().hostName + "-" + UUID.randomUUID().toString().take(8)
    
    @Scheduled(fixedDelay = 5000) // каждые 5 секунд
    @Transactional
    suspend fun processEvents() {
        val events = outboxService.getPendingEvents()
        
        events.forEach { event ->
            processEventSafely(event)
        }
    }
    
    private suspend fun processEventSafely(event: OutboxEvent) {
        try {
            // Пытаемся захватить событие с Optimistic Locking
            val claimed = outboxService.claimEvent(event, instanceId)
            if (!claimed) {
                logger.debug("Event ${event.eventId} already claimed by another instance")
                return
            }
            
            // Отправляем в RabbitMQ
            val statusEvent = objectMapper.readValue(event.payload, EventUpdateStatusRequest::class.java)
            rabbitService.unpaidStatusSend(statusEvent)
            
            // Помечаем как успешно обработанное
            val completed = outboxService.markAsCompleted(event)
            if (completed) {
                logger.info("Event ${event.eventId} processed successfully by $instanceId")
            }
            
        } catch (e: OptimisticLockingFailureException) {
            logger.debug("Event ${event.eventId} processed by another instance")
        } catch (e: Exception) {
            handleProcessingError(event, e)
        }
    }
    
    private suspend fun handleProcessingError(event: OutboxEvent, error: Exception) {
        val newRetryCount = event.retryCount + 1
        val maxRetries = 5
        
        if (newRetryCount >= maxRetries) {
            // Превышен лимит - помечаем как окончательно FAILED
            outboxService.markAsFailed(
                event, 
                error.message ?: "Unknown error",
                newRetryCount
            )
            logger.error("Event ${event.eventId} failed after $maxRetries attempts", error)
        } else {
            // Планируем retry с экспоненциальным backoff
            val delaySeconds = calculateBackoffDelay(newRetryCount)
            val nextRetryAt = LocalDateTime.now().plusSeconds(delaySeconds)
            
            outboxService.scheduleRetry(
                event,
                newRetryCount, 
                nextRetryAt,
                error.message ?: "Retry scheduled"
            )
            logger.warn("Event ${event.eventId} scheduled for retry #$newRetryCount in ${delaySeconds}s", error)
        }
    }
    
    // Экспоненциальный backoff: 2^retry_count секунд (макс 5 минут)
    private fun calculateBackoffDelay(retryCount: Int): Long {
        return minOf(2.0.pow(retryCount).toLong(), 300)
    }
    
    // Cleanup механизм для заблокированных событий
    @Scheduled(fixedDelay = 60000) // каждую минуту
    @Transactional
    suspend fun cleanupStaleEvents() {
        val updated = outboxRepository.releaseStaleEvents(
            LocalDateTime.now().minusMinutes(5)
        )
        
        if (updated > 0) {
            logger.warn("Released $updated stale events locked more than 5 minutes ago")
        }
    }
}

@Service
class SavePolicyExecutorImpl(
    private val storageClient: StorageClient,
    private val outboxService: OutboxService
) : SavePolicyExecutor() {

    @Transactional
    suspend fun savePolicyWithEvent(
        policy: InsurancePolicy,
        eventId: String?,
        task: ExternalTask
    ): UUID {
        // 1. Сохранение полиса
        val storageResponse = storageClient.savePolicy(policy)
        val policyId = storageResponse?.policyId
        
        // 2. Сохранение события в outbox (в той же транзакции)
        if (eventId != null) {
            val statusEvent = EventUpdateStatusRequest(
                eventId = eventId,
                statusCode = RabbitProcessStatus.CREATE_POLICY.statusCode,
                comment = null,
                date = LocalDateTime.now()
            )
            
            outboxService.saveEvent(
                eventId = eventId,
                eventType = "STATUS_UPDATE",
                aggregateId = policyId.toString(),
                aggregateType = "POLICY",
                payload = statusEvent
            )
        }
        
        return policyId
    }
}

@Component
class OutboxMetrics {
    
    @Counter(
        name = "outbox_events_total",
        description = "Total number of outbox events by status"
    )
    private lateinit var eventsTotal: Counter
    
    @Timer(
        name = "outbox_processing_duration_seconds",
        description = "Time taken to process outbox events"
    )
    private lateinit var processingDuration: Timer
    
    @Gauge(
        name = "outbox_queue_size",
        description = "Number of pending events in outbox"
    )
    private lateinit var queueSize: Gauge
    
    @Counter(
        name = "outbox_retry_attempts_total",
        description = "Total retry attempts by retry count"
    )
    private lateinit var retryAttempts: Counter
    
    @Counter(
        name = "outbox_optimistic_lock_failures_total",
        description = "Total optimistic locking failures"
    )
    private lateinit var lockFailures: Counter
}

# Критичные алерты
- alert: OutboxEventsStuck
  expr: outbox_queue_size > 1000
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: "Too many pending events in outbox"
  
- alert: OutboxHighFailureRate  
  expr: rate(outbox_events_total{status="FAILED"}[5m]) > 0.1
  for: 2m
  labels:
    severity: warning
  annotations:
    summary: "High failure rate in outbox processing"

- alert: OutboxProcessingLag
  expr: outbox_processing_duration_seconds{quantile="0.95"} > 300
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "Outbox processing lag too high"

@Service
class OrderService(
    private val orderRepository: OrderRepository,
    private val kafkaTemplate: KafkaTemplate<String, OrderEvent>
) {
    
    @Transactional
    fun processOrder(order: Order) {
        // 1. Сохраняем заказ в БД
        val savedOrder = orderRepository.save(order)
        
        // 2. Отправляем событие в Kafka (ВНЕ транзакции!)
        kafkaTemplate.send("order-events", OrderCreatedEvent(savedOrder.id))
        
        // 3. Обновляем статус
        orderRepository.updateStatus(savedOrder.id, OrderStatus.PROCESSED)
    }
}

• Пиковая нагрузка: 10,000 заказов/минуту
• Потеря событий: 5-7% в часы пик
• Дублирующие события: ~15% от объема
• P95 latency обработки заказа: 2.5 секунды
• Availability: 99.2%

[Order Service] → [DB Transaction: Order + Outbox] → [Outbox Processor] → [Kafka]
       ↓                    ↓                              ↓
   Sync Fast           Atomic Write                 Async Reliable

CREATE TABLE kafka_outbox (
    id BIGSERIAL PRIMARY KEY,
    aggregate_id VARCHAR(255) NOT NULL,
    aggregate_type VARCHAR(100) NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    kafka_topic VARCHAR(255) NOT NULL,
    kafka_key VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    headers JSONB NULL,
    
    -- Статус и обработка
    status VARCHAR(20) DEFAULT 'PENDING',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    processed_at TIMESTAMP NULL,
    
    -- Retry логика
    retry_count INTEGER DEFAULT 0,
    next_retry_at TIMESTAMP NULL,
    error_message TEXT NULL,
    
    -- Deduplication
    idempotency_key VARCHAR(255) UNIQUE NOT NULL,
    
    -- Партиционирование для производительности
    PARTITION BY RANGE (created_at),
    
    -- Индексы для быстрого поиска
    INDEX idx_status_created (status, created_at),
    INDEX idx_retry_ready (status, next_retry_at) WHERE status = 'PENDING',
    INDEX idx_idempotency (idempotency_key)
);

-- Партиции по дням для легкой очистки
CREATE TABLE kafka_outbox_2024_01 PARTITION OF kafka_outbox 
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');

@Service
class OrderService(
    private val orderRepository: OrderRepository,
    private val outboxService: KafkaOutboxService
) {
    
    @Transactional
    fun processOrder(order: Order) {
        // 1. Сохраняем заказ
        val savedOrder = orderRepository.save(order)
        
        // 2. Сохраняем событие в outbox (в той же транзакции!)
        outboxService.publishEvent(
            aggregateId = savedOrder.id.toString(),
            aggregateType = "Order",
            eventType = "OrderCreated",
            topic = "order-events",
            key = savedOrder.id.toString(),
            payload = OrderCreatedEvent(
                orderId = savedOrder.id,
                customerId = savedOrder.customerId,
                amount = savedOrder.totalAmount,
                timestamp = Instant.now()
            ),
            idempotencyKey = "order-created-${savedOrder.id}"
        )
        
        // 3. Обновляем статус (все в одной транзакции)
        orderRepository.updateStatus(savedOrder.id, OrderStatus.PUBLISHED)
    }
}

@Service
@Transactional
class KafkaOutboxService(
    private val outboxRepository: KafkaOutboxRepository,
    private val objectMapper: ObjectMapper
) {
    
    fun publishEvent(
        aggregateId: String,
        aggregateType: String,
        eventType: String,
        topic: String,
        key: String,
        payload: Any,
        headers: Map<String, String> = emptyMap(),
        idempotencyKey: String
    ) {
        val outboxEvent = KafkaOutboxEvent(
            aggregateId = aggregateId,
            aggregateType = aggregateType,
            eventType = eventType,
            kafkaTopic = topic,
            kafkaKey = key,
            payload = objectMapper.writeValueAsString(payload),
            headers = if (headers.isEmpty()) null else objectMapper.writeValueAsString(headers),
            idempotencyKey = idempotencyKey
        )
        
        try {
            outboxRepository.save(outboxEvent)
        } catch (e: DataIntegrityViolationException) {
            // Идемпотентность - событие уже существует
            logger.debug("Event with idempotency key $idempotencyKey already exists")
        }
    }
}

@Component
class KafkaOutboxProcessor(
    private val outboxRepository: KafkaOutboxRepository,
    private val kafkaTemplate: KafkaTemplate<String, String>,
    private val meterRegistry: MeterRegistry
) {
    private val logger = LoggerFactory.getLogger(javaClass)
    private val instanceId = InetAddress.getLocalHost().hostName
    
    // Параллельная обработка для высокой производительности
    private val executor = Executors.newFixedThreadPool(10)
    
    @Scheduled(fixedDelay = 1000) // каждую секунду
    fun processPendingEvents() {
        val startTime = System.currentTimeMillis()
        
        // Batch processing для производительности
        val events = outboxRepository.findPendingEventsBatch(
            limit = 100,
            instanceId = instanceId
        )
        
        if (events.isEmpty()) return
        
        logger.info("Processing ${events.size} outbox events")
        
        // Параллельная обработка batch'а
        val futures = events.map { event ->
            CompletableFuture.supplyAsync({ processEvent(event) }, executor)
        }
        
        // Ждем завершения всех событий в batch'е
        CompletableFuture.allOf(*futures.toTypedArray()).join()
        
        val processingTime = System.currentTimeMillis() - startTime
        
        // Метрики производительности  
        meterRegistry.timer("outbox.batch.processing.time")
            .record(processingTime, TimeUnit.MILLISECONDS)
        meterRegistry.counter("outbox.events.processed")
            .increment(events.size.toDouble())
        
        logger.info("Processed ${events.size} events in ${processingTime}ms")
    }
    
    private fun processEvent(event: KafkaOutboxEvent): Boolean {
        return try {
            // Отправляем в Kafka с настроенными retry
            val record = ProducerRecord<String, String>(
                event.kafkaTopic,
                event.kafkaKey,
                event.payload
            )
            
            // Добавляем headers если есть
            event.headers?.let { headersJson ->
                val headers: Map<String, String> = objectMapper.readValue(headersJson)
                headers.forEach { (key, value) ->
                    record.headers().add(key, value.toByteArray())
                }
            }
            
            // Синхронная отправка для гарантии доставки
            val result = kafkaTemplate.send(record).get(5, TimeUnit.SECONDS)
            
            // Помечаем как успешно обработанное
            outboxRepository.markAsCompleted(event.id!!, Instant.now())
            
            logger.debug("Successfully sent event ${event.id} to topic ${event.kafkaTopic}")
            
            meterRegistry.counter("outbox.events.sent.success",
                "topic", event.kafkaTopic,
                "event_type", event.eventType
            ).increment()
            
            true
            
        } catch (e: Exception) {
            handleEventError(event, e)
            false
        }
    }
    
    private fun handleEventError(event: KafkaOutboxEvent, error: Exception) {
        val newRetryCount = event.retryCount + 1
        val maxRetries = 5
        
        logger.error("Failed to process event ${event.id}, attempt $newRetryCount", error)
        
        meterRegistry.counter("outbox.events.sent.failure",
            "topic", event.kafkaTopic,
            "event_type", event.eventType,
            "error_type", error.javaClass.simpleName
        ).increment()
        
        if (newRetryCount >= maxRetries) {
            // Dead Letter после исчерпания retry
            outboxRepository.markAsFailed(
                event.id!!, 
                error.message ?: "Unknown error",
                newRetryCount
            )
            
            // Отправляем в DLQ для ручной обработки
            sendToDeadLetterQueue(event, error)
            
        } else {
            // Экспоненциальный backoff
            val delaySeconds = minOf(2.0.pow(newRetryCount).toLong(), 300)
            val nextRetryAt = Instant.now().plusSeconds(delaySeconds)
            
            outboxRepository.scheduleRetry(
                event.id!!,
                newRetryCount,
                nextRetryAt,
                error.message ?: "Retry scheduled"
            )
        }
    }
    
    private fun sendToDeadLetterQueue(event: KafkaOutboxEvent, error: Exception) {
        try {
            val dlqEvent = mapOf(
                "original_event" to event,
                "error" to error.message,
                "failed_at" to Instant.now(),
                "retry_count" to event.retryCount
            )
            
            kafkaTemplate.send("outbox-dlq", event.kafkaKey, objectMapper.writeValueAsString(dlqEvent))
            logger.error("Event ${event.id} sent to DLQ after ${event.retryCount} retries")
            
        } catch (dlqError: Exception) {
            logger.error("Failed to send event ${event.id} to DLQ", dlqError)
        }
    }
}

@Repository
interface KafkaOutboxRepository : JpaRepository<KafkaOutboxEvent, Long> {
    
    @Query("""
        SELECT e FROM KafkaOutboxEvent e 
        WHERE e.status = 'PENDING' 
        AND (e.nextRetryAt IS NULL OR e.nextRetryAt <= :now)
        ORDER BY e.createdAt ASC
        LIMIT :limit
        FOR UPDATE SKIP LOCKED
    """, nativeQuery = true)
    fun findPendingEventsBatch(
        @Param("now") now: Instant = Instant.now(),
        @Param("limit") limit: Int,
        @Param("instanceId") instanceId: String
    ): List<KafkaOutboxEvent>
    
    @Modifying
    @Query("""
        UPDATE kafka_outbox 
        SET status = 'COMPLETED', processed_at = :processedAt 
        WHERE id = :id
    """, nativeQuery = true)
    fun markAsCompleted(@Param("id") id: Long, @Param("processedAt") processedAt: Instant)
    
    @Modifying
    @Query("""
        UPDATE kafka_outbox 
        SET status = 'FAILED', retry_count = :retryCount, error_message = :errorMessage
        WHERE id = :id
    """, nativeQuery = true)
    fun markAsFailed(
        @Param("id") id: Long, 
        @Param("errorMessage") errorMessage: String,
        @Param("retryCount") retryCount: Int
    )
    
    @Modifying
    @Query("""
        UPDATE kafka_outbox 
        SET retry_count = :retryCount, next_retry_at = :nextRetryAt, 
            error_message = :errorMessage, status = 'PENDING'
        WHERE id = :id
    """, nativeQuery = true)
    fun scheduleRetry(
        @Param("id") id: Long,
        @Param("retryCount") retryCount: Int,
        @Param("nextRetryAt") nextRetryAt: Instant,
        @Param("errorMessage") errorMessage: String
    )
    
    // Очистка старых событий для производительности
    @Modifying
    @Query("""
        DELETE FROM kafka_outbox 
        WHERE status = 'COMPLETED' 
        AND processed_at < :before
    """, nativeQuery = true)
    fun cleanupOldEvents(@Param("before") before: Instant): Int
}

@Component
class OutboxMetrics(private val meterRegistry: MeterRegistry) {
    
    @EventListener
    fun onEventPublished(event: OutboxEventPublished) {
        meterRegistry.counter("outbox.events.published",
            "aggregate_type", event.aggregateType,
            "event_type", event.eventType
        ).increment()
    }
    
    @Scheduled(fixedDelay = 30000)
    fun recordQueueSize() {
        val pendingCount = outboxRepository.countPendingEvents()
        meterRegistry.gauge("outbox.queue.size", pendingCount.toDouble())
        
        val failedCount = outboxRepository.countFailedEvents()  
        meterRegistry.gauge("outbox.failed.size", failedCount.toDouble())
    }
    
    @Scheduled(fixedDelay = 60000)
    fun recordLatencyMetrics() {
        val avgLatency = outboxRepository.calculateAverageProcessingLatency()
        meterRegistry.gauge("outbox.processing.latency.avg", avgLatency)
        
        val p95Latency = outboxRepository.calculateP95ProcessingLatency()
        meterRegistry.gauge("outbox.processing.latency.p95", p95Latency)
    }
}

@Configuration
class KafkaProducerConfig {
    
    @Bean
    fun kafkaTemplate(): KafkaTemplate<String, String> {
        val props = mapOf(
            // === БАЗОВЫЕ НАСТРОЙКИ ===
            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to "kafka1:9092,kafka2:9092,kafka3:9092",
            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java,
            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java,
            ProducerConfig.CLIENT_ID_CONFIG to "outbox-producer-${InetAddress.getLocalHost().hostName}",
            
            // === ИДЕМПОТЕНТНОСТЬ (ключевое для исключения дублей) ===
            ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG to true,
            ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION to 5, // <= 5 для idempotence
            
            // === НАДЕЖНОСТЬ ДОСТАВКИ ===
            ProducerConfig.ACKS_CONFIG to "all", // ждем подтверждения от всех ISR replicas
            ProducerConfig.RETRIES_CONFIG to Integer.MAX_VALUE, // бесконечные retry
            ProducerConfig.RETRY_BACKOFF_MS_CONFIG to 1000, // 1 секунда между retry
            ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG to 30000, // 30 секунд timeout
            ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG to 120000, // 2 минуты общий timeout
            
            // === ПРОИЗВОДИТЕЛЬНОСТЬ ===
            ProducerConfig.BATCH_SIZE_CONFIG to 65536, // 64KB batches для throughput
            ProducerConfig.LINGER_MS_CONFIG to 20, // ждем 20ms для наполнения batch
            ProducerConfig.BUFFER_MEMORY_CONFIG to 134217728, // 128MB buffer
            ProducerConfig.COMPRESSION_TYPE_CONFIG to "snappy", // быстрое сжатие
            
            // === БЕЗОПАСНОСТЬ ===
            "security.protocol" to "SASL_SSL",
            "sasl.mechanism" to "SCRAM-SHA-512", 
            "sasl.jaas.config" to "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"outbox-producer\" password=\"${getPassword()}\";",
            "ssl.truststore.location" to "/app/certs/kafka.truststore.jks",
            "ssl.truststore.password" to "${getTruststorePassword()}",
            
            // === МОНИТОРИНГ ===
            ProducerConfig.INTERCEPTOR_CLASSES_CONFIG to listOf(
                "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
            ),
            "confluent.monitoring.interceptor.bootstrap.servers" to "kafka1:9092,kafka2:9092,kafka3:9092"
        )
        
        val producerFactory = DefaultKafkaProducerFactory<String, String>(props)
        
        return KafkaTemplate(producerFactory).apply {
            // Кастомный error handler
            setProducerExceptionHandler { record, exception ->
                logger.error("Failed to send record to topic ${record.topic()}", exception)
                // Не останавливаем процесс - ошибка будет обработана в outbox retry
            }
        }
    }
}

# === БАЗОВЫЕ НАСТРОЙКИ ===
broker.id=1
listeners=PLAINTEXT://0.0.0.0:9092,SASL_SSL://0.0.0.0:9093
advertised.listeners=PLAINTEXT://kafka1:9092,SASL_SSL://kafka1:9093
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181

# === НАДЕЖНОСТЬ И DURABILITY ===
# Минимальное количество in-sync replicas для записи
min.insync.replicas=2
# Фактор репликации по умолчанию  
default.replication.factor=3
# Количество партиций по умолчанию
num.partitions=12

# === ПРОИЗВОДИТЕЛЬНОСТЬ ===
# Количество I/O потоков
num.io.threads=16
# Количество сетевых потоков  
num.network.threads=8
# Размер socket buffers
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

# === LOG SETTINGS ===
# Время хранения сообщений (7 дней)
log.retention.hours=168
# Максимальный размер лог-файла
log.segment.bytes=1073741824
# Интервал проверки на удаление старых сегментов
log.retention.check.interval.ms=300000
# Компрессия логов
log.compression.type=snappy

# === PERFORMANCE TUNING ===
# Размер batch для репликации
replica.fetch.max.bytes=10485760
# Максимальное время ожидания fetch запроса
replica.fetch.wait.max.ms=500
# Размер буфера для записи на диск
log.flush.interval.messages=10000
log.flush.interval.ms=1000

# === МОНИТОРИНГ ===
# JMX настройки
jmx.enabled=true
kafka.metrics.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter
confluent.metrics.reporter.bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092

# === БЕЗОПАСНОСТЬ ===
security.inter.broker.protocol=SASL_SSL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
ssl.keystore.location=/opt/kafka/certs/kafka.keystore.jks
ssl.keystore.password=kafka-secret
ssl.key.password=kafka-secret
ssl.truststore.location=/opt/kafka/certs/kafka.truststore.jks
ssl.truststore.password=kafka-secret

# Создание основного топика для событий
kafka-topics --create \
  --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 \
  --topic order-events \
  --partitions 12 \
  --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config cleanup.policy=delete \
  --config retention.ms=604800000 \
  --config compression.type=snappy \
  --config segment.ms=86400000 \
  --config max.message.bytes=10485760

# Dead Letter Queue топик
kafka-topics --create \
  --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 \
  --topic outbox-dlq \
  --partitions 3 \
  --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config cleanup.policy=delete \
  --config retention.ms=2592000000 \
  --config compression.type=snappy

# Топик для мониторинга метрик
kafka-topics --create \
  --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 \
  --topic _confluent-metrics \
  --partitions 6 \
  --replication-factor 3 \
  --config cleanup.policy=delete \
  --config retention.ms=259200000

@Configuration  
class KafkaConsumerConfig {

    @Bean
    fun consumerFactory(): ConsumerFactory<String, String> {
        val props = mapOf(
            // === БАЗОВЫЕ НАСТРОЙКИ ===
            ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to "kafka1:9092,kafka2:9092,kafka3:9092",
            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
            ConsumerConfig.GROUP_ID_CONFIG to "order-processing-service",
            ConsumerConfig.CLIENT_ID_CONFIG to "order-consumer-${InetAddress.getLocalHost().hostName}",
            
            // === EXACTLY-ONCE PROCESSING ===
            ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false, // ручной commit для exactly-once
            ConsumerConfig.ISOLATION_LEVEL_CONFIG to "read_committed", // читаем только committed записи
            
            // === ПРОИЗВОДИТЕЛЬНОСТЬ ===
            ConsumerConfig.FETCH_MIN_BYTES_CONFIG to 50000, // минимум 50KB для fetch
            ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG to 200, // максимум 200ms ожидания
            ConsumerConfig.MAX_POLL_RECORDS_CONFIG to 100, // обрабатываем по 100 записей
            ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG to 300000, // 5 минут на обработку batch
            
            // === OFFSET MANAGEMENT ===
            ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest", // начинаем с начала при отсутствии offset
            ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG to 30000, // 30 секунд session timeout
            ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG to 10000, // heartbeat каждые 10 секунд
            
            // === БЕЗОПАСНОСТЬ ===
            "security.protocol" to "SASL_SSL",
            "sasl.mechanism" to "SCRAM-SHA-512",
            "sasl.jaas.config" to "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"order-consumer\" password=\"${getPassword()}\";",
            "ssl.truststore.location" to "/app/certs/kafka.truststore.jks",
            "ssl.truststore.password" to "${getTruststorePassword()}"
        )
        
        return DefaultKafkaConsumerFactory(props)
    }

    @Bean
    fun kafkaListenerContainerFactory(): ConcurrentKafkaListenerContainerFactory<String, String> {
        val factory = ConcurrentKafkaListenerContainerFactory<String, String>()
        factory.consumerFactory = consumerFactory()
        
        // === CONCURRENCY ===
        factory.setConcurrency(3) // 3 consumer thread на partition
        
        // === ERROR HANDLING ===
        factory.setCommonErrorHandler(DefaultErrorHandler(
            DeadLetterPublishingRecoverer(kafkaTemplate()),
            FixedBackOff(1000L, 3) // 3 retry с задержкой 1 сек
        ))
        
        // === ACKS ===
        factory.containerProperties.ackMode = ContainerProperties.AckMode.MANUAL_IMMEDIATE
        
        return factory
    }
}

# Kafka Broker JVM settings
export KAFKA_HEAP_OPTS="-Xmx8G -Xms8G"
export KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true -Dsun.net.useExclusiveBind=false"

# Producer/Consumer JVM settings  
export JAVA_OPTS="-Xmx2G -Xms2G -XX:+UseG1GC -XX:MaxGCPauseMillis=100"

@Configuration
class KafkaMetricsConfig {
    
    @Bean
    fun kafkaProducerMetrics(meterRegistry: MeterRegistry): MeterBinder {
        return KafkaProducerMetrics(mapOf(
            "client.id" to "outbox-producer"
        ))
    }
    
    @Bean  
    fun kafkaConsumerMetrics(meterRegistry: MeterRegistry): MeterBinder {
        return KafkaConsumerMetrics(mapOf(
            "client.id" to "order-consumer"
        ))
    }
}

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:

      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on:

      - zookeeper
    ports:

      - "9092:9092"
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_DEFAULT_REPLICATION_FACTOR: 1
      KAFKA_MIN_INSYNC_REPLICAS: 1
      KAFKA_NUM_PARTITIONS: 12
      KAFKA_COMPRESSION_TYPE: snappy
      KAFKA_LOG_RETENTION_HOURS: 168
      KAFKA_LOG_SEGMENT_BYTES: 1073741824
      
  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    depends_on:

      - kafka
    ports:

      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092

• Потеря событий: 5-7% в часы пик
• Дублирующие события: ~15% от общего объема  
• P95 latency: 2.5 секунды
• Availability: 99.2%
• Throughput: 8,000 заказов/минуту (макс)

• Потеря событий: 0% (guaranteed delivery)
• Дублирующие события: ~3% (сокращение на 80%)
• P95 latency: 180ms (улучшение в 14 раз)
• Availability: 99.9%  
• Throughput: 15,000 заказов/минуту (в 1.9 раза больше)

# Grafana Dashboard
- Events Published Rate: 167/sec (average)
- Events Processing Rate: 170/sec (average)  
- Queue Size: 15 events (average)
- Failed Events: 0.1% of total
- Processing Latency P95: 180ms
- Duplicate Rate: 3% (down from 15%)

Окно 1: [0-60с]  -> 10 запросов разрешено
Окно 2: [60-120с] -> 10 запросов разрешено

Ведро: [🪙🪙🪙🪙🪙] (5 токенов)
Пополнение: +1 токен каждые 200мс
Запрос: -1 токен

Вход: запросы -> [🌊🌊🌊] -> Выход: постоянная скорость
                 Ведро       (например, 1 req/100ms)

Время:  [--10с--][--10с--][--10с--]
Окно 1:      [=========10с=========]  (3+2=5 запросов)
Окно 2:           [=========10с=========]  (2+4=6 запросов)

// Создание лимитера через реестр
val rateLimiter = rateLimiterRegistry.rateLimiter("api-service")
// При повторном вызове с тем же именем вернется тот же экземпляр
val sameRateLimiter = rateLimiterRegistry.rateLimiter("api-service")
// rateLimiter === sameRateLimiter (true)

// Глобальная конфигурация по умолчанию
val globalConfig = RateLimiterConfig.custom()
    .limitForPeriod(10)
    .limitRefreshPeriod(Duration.ofSeconds(1))
    .build()

// Кастомная конфигурация для конкретного сервиса
val customConfig = RateLimiterConfig.custom()
    .limitForPeriod(5)
    .limitRefreshPeriod(Duration.ofSeconds(1))
    .build()

val registry = RateLimiterRegistry.of(globalConfig)
registry.addConfiguration("slow-service", customConfig)

val now = System.currentTimeMillis()
val windowStart = now - window.toMillis()

redis.multi()
redis.zremrangebyscore(key, 0, windowStart)
val currentCount = redis.zcard(key)

if (currentCount >= limit) {
    redis.discard()
    throw RequestNotPermitted()
}

redis.zadd(key, now.toDouble(), now.toString())
redis.expire(key, window.seconds)
redis.exec()

HTTP/1.1 429 Too Many Requests
Content-Type: application/json
Retry-After: 60

{
  "error": "rate_limit_exceeded",
  "message": "API rate limit exceeded. Try again in 60 seconds",
  "limit": 6,
  "remaining": 0,
  "reset_time": "2025-07-25T10:31:00Z"
}

@Service
class ExternalApiClient(
    private val webClient: WebClient
) {
    suspend fun createApplication(request: AddApplicationRequest): ApplicationDto {
        return try {
            webClient.post()
                .uri("/api/applications")
                .bodyValue(request)
                .retrieve()
                .awaitBody<ApplicationDto>()
                
        } catch (e: WebClientResponseException) {
            when (e.statusCode.value()) {
                429 -> {
                    // Внешний API вернул Rate Limit exceeded
                    val retryAfter = e.headers.getFirst("Retry-After")?.toLongOrNull() ?: 60
                    throw ExternalApiRateLimitException(
                        "External API rate limit exceeded. Retry after $retryAfter seconds",
                        retryAfter,
                        e
                    )
                }
                503 -> {
                    throw ExternalApiUnavailableException("External API temporarily unavailable", e)
                }
                else -> throw e
            }
        }
    }
}

// Кастомные исключения для Rate Limit
class ExternalApiRateLimitException(
    message: String,
    val retryAfterSeconds: Long,
    cause: Throwable? = null
) : RuntimeException(message, cause)

class InternalRateLimitException(
    message: String,
    val rateLimiterName: String,
    cause: Throwable? = null
) : RuntimeException(message, cause)

@Service
class ExternalApiClient(
    private val webClient: WebClient
) {
    suspend fun createApplication(request: AddApplicationRequest): ApplicationDto {
        return webClient.post()
            .uri("/api/applications")
            .bodyValue(request)
            .retrieve()
            .onStatus(HttpStatus::is4xxClientError) { response ->
                when (response.statusCode()) {
                    HttpStatus.TOO_MANY_REQUESTS -> {
                        response.toEntity(ErrorResponse::class.java)
                            .map { entity ->
                                val retryAfter = response.headers().getFirst("Retry-After")?.toLongOrNull() ?: 60
                                val errorBody = entity.body
                                ExternalApiRateLimitException(
                                    "External API rate limit: ${errorBody?.message ?: "Rate limit exceeded"}",
                                    retryAfter
                                )
                            }
                    }
                    else -> {
                        Mono.error(WebClientResponseException.create(
                            response.statusCode().value(),
                            "Client error",
                            response.headers().asHttpHeaders(),
                            ByteArray(0),
                            null
                        ))
                    }
                }
            }
            .onStatus(HttpStatus::is5xxServerError) { response ->
                when (response.statusCode()) {
                    HttpStatus.SERVICE_UNAVAILABLE -> {
                        Mono.error(ExternalApiUnavailableException("External API unavailable"))
                    }
                    else -> {
                        Mono.error(WebClientResponseException.create(
                            response.statusCode().value(),
                            "Server error", 
                            response.headers().asHttpHeaders(),
                            ByteArray(0),
                            null
                        ))
                    }
                }
            }
            .awaitBody<ApplicationDto>()
    }
}

data class ErrorResponse(
    val error: String,
    val message: String,
    val limit: Int? = null,
    val remaining: Int? = null,
    val resetTime: String? = null
)

@RestController
class ApplicationController(
    private val applicationService: ApplicationService
) {
    
    @PostMapping("/applications")
    suspend fun createApplication(@RequestBody request: AddApplicationRequest): ResponseEntity<ApplicationDto> {
        return try {
            val result = applicationService.addApplicationWithRateLimit(request)
            ResponseEntity.ok(result)
            
        } catch (e: RequestNotPermitted) {
            // Наш внутренний Rate Limiter отклонил запрос
            ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS)
                .header("Retry-After", "1") // Попробовать через 1 секунду
                .body(ErrorResponse(
                    error = "internal_rate_limit_exceeded",
                    message = "Internal rate limit exceeded. Please retry after 1 second"
                ))
                
        } catch (e: ExternalApiRateLimitException) {
            // Внешний API отклонил запрос
            ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS)
                .header("Retry-After", e.retryAfterSeconds.toString())
                .body(ErrorResponse(
                    error = "external_rate_limit_exceeded", 
                    message = e.message ?: "External API rate limit exceeded",
                    retryAfterSeconds = e.retryAfterSeconds
                ))
                
        } catch (e: Exception) {
            ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body(ErrorResponse(
                    error = "internal_server_error",
                    message = "An unexpected error occurred"
                ))
        }
    }
}

@Component
class RateLimitErrorHandler {
    
    private val logger = LoggerFactory.getLogger(RateLimitErrorHandler::class.java)
    
    @EventListener
    fun handleInternalRateLimit(event: RequestNotPermitted) {
        logger.warn(
            "Internal rate limit exceeded for rate limiter: ${event.rateLimiterName}, " +
            "available permissions: ${event.rateLimiter.availablePermissions}"
        )
        
        // Отправка метрики
        Metrics.counter("rate_limiter.internal_exceeded", 
            "limiter_name", event.rateLimiterName
        ).increment()
    }
    
    @EventListener 
    fun handleExternalRateLimit(event: ExternalApiRateLimitException) {
        logger.warn(
            "External API rate limit exceeded: ${event.message}, " +
            "retry after: ${event.retryAfterSeconds} seconds"
        )
        
        // Отправка метрики
        Metrics.counter("rate_limiter.external_exceeded",
            "retry_after", event.retryAfterSeconds.toString()
        ).increment()
    }
}

suspend fun <T> executeWithRetry(
    maxRetries: Int = 3,
    initialDelay: Duration = Duration.ofMillis(100),
    maxDelay: Duration = Duration.ofSeconds(5),
    block: suspend () -> T
): T {
    var currentDelay = initialDelay
    repeat(maxRetries + 1) { attempt ->
        try {
            return block()
        } catch (e: RequestNotPermitted) {
            if (attempt < maxRetries) {
                delay(currentDelay.toMillis())
                currentDelay = minOf(currentDelay * 2, maxDelay)
            } else {
                throw RateLimitExceededException("Превышен лимит запросов после $maxRetries попыток", e)
            }
        }
    }
    throw RuntimeException("Неожиданная ошибка")
}

# Метрики за неделю до внедрения
external_api.errors_5xx.total: 2,847 errors
external_api.errors_429.total: 156 rate limit errors
external_api.calls_total: 45,230 calls
error_rate: 6.6% (2,847 + 156) / 45,230
average_response_time: 2,340ms
sla_breaches: 12 incidents

# Метрики за неделю после внедрения
external_api.errors_5xx.total: 1,547 errors
external_api.errors_429.total: 23 rate limit errors
external_api.calls_total: 44,890 calls
error_rate: 3.5% (1,547 + 23) / 44,890
average_response_time: 1,980ms
sla_breaches: 2 incidents
rate_limiter.exceeded.total: 3,420 prevented calls
rate_limiter.fallback_used.total: 45 fallback events

@Configuration
class MetricsConfiguration {

    @Bean
    fun rateLimiterMetrics(
        meterRegistry: MeterRegistry,
        rateLimiterRegistry: RateLimiterRegistry
    ): RateLimiterMetrics {
        // Автоматическая регистрация всех метрик Rate Limiter в Prometheus
        return RateLimiterMetrics.ofRateLimiterRegistry(rateLimiterRegistry)
            .bindTo(meterRegistry)
    }
}

@Aspect
@Component
class RateLimiterMetricsAspect(
    private val meterRegistry: MeterRegistry
) {

    @Around("@annotation(resilientRateLimit)")
    fun aroundRateLimit(joinPoint: ProceedingJoinPoint, resilientRateLimit: ResilientApplicationRateLimit): Any? {
        val rateLimiterName = resilientRateLimit.rateLimiterName
        val timer = Timer.start(meterRegistry)

        return try {
            // Попытка выполнения с Redis Rate Limiter
            val rateLimiter = redisRateLimiterRegistry.rateLimiter(rateLimiterName)
            val result = runBlocking {
                rateLimiter.executeSuspendFunction { joinPoint.proceed() }
            }

            // ✅ МЕТРИКА: Успешное выполнение
            Counter.builder("rate_limiter.success")
                .tag("name", rateLimiterName)
                .tag("type", "redis")
                .register(meterRegistry)
                .increment()

            result

        } catch (e: RequestNotPermitted) {
            // ✅ МЕТРИКА: Превышение лимита Redis
            Counter.builder("rate_limiter.exceeded")
                .tag("name", rateLimiterName)
                .tag("type", "redis")
                .register(meterRegistry)
                .increment()

            if (resilientRateLimit.fallbackEnabled) {
                // Переход на локальный fallback
                try {
                    val localRateLimiter = localRateLimiterRegistry.rateLimiter(rateLimiterName)
                    val result = runBlocking {
                        localRateLimiter.executeSuspendFunction { joinPoint.proceed() }
                    }

                    // ✅ МЕТРИКА: Использование fallback
                    Counter.builder("rate_limiter.fallback_used")
                        .tag("name", rateLimiterName)
                        .tag("reason", "redis_rate_limit_exceeded")
                        .register(meterRegistry)
                        .increment()

                    result
                } catch (localException: RequestNotPermitted) {
                    // ✅ МЕТРИКА: Окончательный отказ
                    Counter.builder("rate_limiter.failed")
                        .tag("name", rateLimiterName)
                        .tag("type", "local_fallback")
                        .register(meterRegistry)
                        .increment()

                    throw localException
                }
            } else {
                throw e
            }

        } catch (e: Exception) {
            // Redis недоступен
            if (resilientRateLimit.fallbackEnabled) {
                // ✅ МЕТРИКА: Fallback из-за недоступности Redis
                Counter.builder("rate_limiter.fallback_used")
                    .tag("name", rateLimiterName)
                    .tag("reason", "redis_unavailable")
                    .register(meterRegistry)
                    .increment()

                val localRateLimiter = localRateLimiterRegistry.rateLimiter(rateLimiterName)
                runBlocking { localRateLimiter.executeSuspendFunction { joinPoint.proceed() } }
            } else {
                throw e
            }
        } finally {
            // ✅ МЕТРИКА: Время выполнения
            timer.stop(Timer.builder("rate_limiter.execution_time")
                .tag("name", rateLimiterName)
                .register(meterRegistry))
        }
    }
}

@Service
class ApplicationService(
    private val externalApiClient: ExternalApiClient
) {

    @ResilientApplicationRateLimit("application-service", fallbackEnabled = true)
    @Timed(
        name = "external_api.response_time",
        description = "External API response time",
        extraTags = ["service", "application"]
    )
    @Counted(
        name = "external_api.calls",
        description = "External API calls count",
        extraTags = ["service", "application"]
    )
    suspend fun addApplicationWithRateLimit(request: AddApplicationRequest): ApplicationDto {
        return try {
            externalApiClient.createApplication(request)
        } catch (e: HttpServerErrorException) {
            // Можно добавить кастомную метрику для ошибок через аспект
            throw e
        }
    }
}

@Service
class ApplicationService(
    private val externalApiClient: ExternalApiClient
) {

    @ResilientApplicationRateLimit("application-service", fallbackEnabled = true)
    @Timed(name = "external_api.response_time", extraTags = ["service", "application"])
    @Counted(name = "external_api.calls", extraTags = ["service", "application"])
    @NewSpan("application-service-call") // Для трассировки
    suspend fun addApplicationWithRateLimit(request: AddApplicationRequest): ApplicationDto {
        return externalApiClient.createApplication(request)
    }

    @Timed(name = "external_api.response_time", extraTags = ["service", "application", "operation", "update"])
    @Counted(name = "external_api.calls", extraTags = ["service", "application", "operation", "update"])
    suspend fun updateApplicationWithRateLimit(id: String, request: UpdateApplicationRequest): ApplicationDto {
        return externalApiClient.updateApplication(id, request)
    }
}

@Configuration
@EnableTimedAspect  // Включает @Timed
@EnableCountedAspect // Включает @Counted
class MicrometerConfiguration {

    @Bean
    fun timedAspect(meterRegistry: MeterRegistry): TimedAspect {
        return TimedAspect(meterRegistry)
    }

    @Bean
    fun countedAspect(meterRegistry: MeterRegistry): CountedAspect {
        return CountedAspect(meterRegistry)
    }
}

@Aspect
@Component
class ExternalApiErrorMetricsAspect(
    private val meterRegistry: MeterRegistry
) {

    @AfterThrowing(
        pointcut = "@annotation(timed) && @annotation(counted)",
        throwing = "exception"
    )
    fun recordApiErrors(
        joinPoint: JoinPoint,
        timed: Timed,
        counted: Counted,
        exception: Exception
    ) {
        when (exception) {
            is HttpServerErrorException -> {
                // ✅ МЕТРИКА: Ошибки 5xx через аннотации
                Counter.builder("external_api.errors")
                    .tags(timed.extraTags.toList().chunked(2) { Tags.of(it[0], it[1]) }.flatten())
                    .tag("status_code", exception.statusCode.value().toString())
                    .tag("error_type", "5xx")
                    .register(meterRegistry)
                    .increment()
            }
            is HttpClientErrorException -> {
                if (exception.statusCode.value() == 429) {
                    // ✅ МЕТРИКА: Rate limit ошибки внешнего API
                    Counter.builder("external_api.rate_limit_errors")
                        .tags(timed.extraTags.toList().chunked(2) { Tags.of(it[0], it[1]) }.flatten())
                        .register(meterRegistry)
                        .increment()
                }
            }
        }
    }
}

# От @Timed аннотации
external_api_response_time_seconds{service="application",quantile="0.95"} 1.98
external_api_response_time_seconds_count{service="application"} 44890
external_api_response_time_seconds_sum{service="application"} 88843.2

# От @Counted аннотации  
external_api_calls_total{service="application"} 44890

# От кастомного аспекта для ошибок
external_api_errors_total{service="application",status_code="500",error_type="5xx"} 1547
external_api_rate_limit_errors_total{service="application"} 23

@Component
class RedisHealthIndicator(
    private val redisConnection: StatefulRedisConnection<String, String>,
    private val meterRegistry: MeterRegistry
) : HealthIndicator {

    override fun health(): Health {
        return try {
            val start = System.currentTimeMillis()
            redisConnection.sync().ping()
            val responseTime = System.currentTimeMillis() - start

            // ✅ МЕТРИКА: Redis доступен
            Gauge.builder("redis.connection.status")
                .description("Redis connection status (1=UP, 0=DOWN)")
                .register(meterRegistry) { 1.0 }

            // ✅ МЕТРИКА: Время ответа Redis
            Timer.builder("redis.response_time")
                .register(meterRegistry)
                .record(Duration.ofMillis(responseTime))

            Health.up()
                .withDetail("redis.response_time", "${responseTime}ms")
                .build()

        } catch (e: Exception) {
            // ✅ МЕТРИКА: Redis недоступен
            Gauge.builder("redis.connection.status")
                .register(meterRegistry) { 0.0 }

            Health.down()
                .withDetail("redis.error", e.message)
                .build()
        }
    }
}

management:
  endpoints:
    web:
      exposure:
        include: health,info,prometheus,metrics
  endpoint:
    prometheus:
      enabled: true
    health:
      show-details: always
  metrics:
    export:
      prometheus:
        enabled: true
    tags:
      application: insurance-bpm-service
      environment: production

# Автоматические метрики Resilience4j
resilience4j_ratelimiter_calls_total{name="application-service",kind="successful"} 1547
resilience4j_ratelimiter_calls_total{name="application-service",kind="failed"} 23

# Кастомные метрики Rate Limiter
rate_limiter_exceeded_total{name="application-service",type="redis"} 3420
rate_limiter_fallback_used_total{name="application-service",reason="redis_rate_limit_exceeded"} 45
rate_limiter_success_total{name="application-service",type="redis"} 41470

# Метрики внешнего API
external_api_calls_total{service="application",status="success"} 44890
external_api_errors_total{service="application",status_code="500"} 1547
external_api_response_time_seconds{service="application",quantile="0.95"} 1.98

# Redis Health метрики  
redis_connection_status{} 1.0
redis_response_time_seconds{quantile="0.99"} 0.003

@Configuration
class RedisRateLimiterConfiguration {

    @Bean
    fun redisClient(): RedisClient = RedisClient.create("redis://localhost:6379")

    @Bean
    fun redisConnection(redisClient: RedisClient): StatefulRedisConnection<String, String> =
        redisClient.connect()

    @Bean
    @Primary
    fun rateLimiterRegistry(redisConnection: StatefulRedisConnection<String, String>): RateLimiterRegistry {
        val config = RateLimiterConfig.custom()
            .limitForPeriod(6)
            .limitRefreshPeriod(Duration.ofSeconds(1))
            .timeoutDuration(Duration.ofSeconds(5))
            .build()
        return RedisRateLimiterRegistry.of(config, redisConnection)
    }
}

@Bean
fun localRateLimiterRegistry(): RateLimiterRegistry {
    val config = RateLimiterConfig.custom()
        .limitForPeriod(1)
        .limitRefreshPeriod(Duration.ofSeconds(1))
        .timeoutDuration(Duration.ofSeconds(5))
        .build()
    return RateLimiterRegistry.of(config)
}

@Target(AnnotationTarget.FUNCTION)
@Retention(AnnotationRetention.RUNTIME)
annotation class ResilientApplicationRateLimit(
    val rateLimiterName: String = "application-service",
    val fallbackEnabled: Boolean = true
)

@Aspect
@Component
class ResilientRateLimitAspect(
    @Qualifier("redisRateLimiterRegistry") private val redisRateLimiterRegistry: RateLimiterRegistry,
    @Qualifier("localRateLimiterRegistry") private val localRateLimiterRegistry: RateLimiterRegistry
) {
    @Around("@annotation(resilientRateLimit)")
    fun around(joinPoint: ProceedingJoinPoint, resilientRateLimit: ResilientApplicationRateLimit): Any? {
        return try {
            val rateLimiter = redisRateLimiterRegistry.rateLimiter(resilientRateLimit.rateLimiterName)
            runBlocking { rateLimiter.executeSuspendFunction { joinPoint.proceed() } }
        } catch (e: Exception) {
            if (resilientRateLimit.fallbackEnabled) {
                val rateLimiter = localRateLimiterRegistry.rateLimiter(resilientRateLimit.rateLimiterName)
                runBlocking { rateLimiter.executeSuspendFunction { joinPoint.proceed() } }
            } else {
                throw e
            }
        }
    }
}

@Service
class ApplicationService(
    private val resilientRateLimiterService: ResilientRateLimiterService
) {

    @ResilientApplicationRateLimit("application-service", fallbackEnabled = true)
    suspend fun addApplicationWithRateLimit(request: AddApplicationRequest): ApplicationDto {
        // ... вызов внешнего сервиса ...
    }
}

spring:
  redis:
    host: localhost
    port: 6379
    timeout: 2000ms
    lettuce:
      pool:
        max-active: 8
        max-idle: 8
        min-idle: 0

vtb:
  insurance:
    bpm:
      executor:
        rate-limiter:
          application:
            limit-for-period: 6
            limit-refresh-period: 1s
            timeout-duration: 5s
            fallback:
              enabled: true
              local-limit-for-period: 1
              local-limit-refresh-period: 1s

enum class UnpaidBpmProcessingStage(val order: Int) {
    INIT(order = 1),                           // Инициализация
    CHECK_INSURANCE(order = 2),                // Проверка активной страховки
    CHECK_COMPLEX_INSURANCE(order = 3),        // Комплексная проверка
    GET_INSURANCE_PROGRAM(order = 4),          // Получение программы
    GET_CLIENT_INFO(order = 5),                // Получение данных клиента
    CREATE_APPLICATION(order = 6),             // Создание заявки
    INSURED_OBJECT_CONFIRM(order = 7),         // Подтверждение объекта
    GET_TEMPLATES_PROGRAM(order = 8),          // Получение шаблонов
    GENERATE_POLICY(order = 9),                // Генерация полиса
    AGREEMENTS_GENERATION(order = 10),         // Генерация соглашений
    CART_RESPONSE_PROCESSING(order = 11),      // Обработка корзины
    APPLICATION_CONFIRMED(order = 12),         // Подтверждение заявки
    INSURED_OBJECT_SAVING(order = 13),         // Сохранение объекта
    FINALIZATION_SUCCESSFULLY(order = 14),     // Успешная финализация
    FINALIZATION_FAILURE(order = 15)           // Ошибка финализации
}

enum class RabbitProcessStatus(val statusCode: Int) {
    NEW(10),                           // Новая заявка
    SENDED_TO_RABBIT(20),              // Отправлено в RabbitMQ
    START_INSTANCE(30),                // Запущен экземпляр
    CHECK_INSTANCE(31),                // Проверка экземпляра
    IN_WORK_INSTANCE(32),              // Экземпляр в работе
    DECLINED_INSTANCE(39),             // Отклонен на уровне экземпляра
    ALLOW_INSTANCE(40),                // Разрешено на уровне экземпляра
    DECLINED_APP(49),                  // Отклонена заявка
    CREATE_APP(50),                    // Создана заявка
    DECLINED_POLICY(59),               // Отклонен полис
    CREATE_POLICY(60),                 // Создан полис
    DONE(70),                          // Завершено успешно
    TERMINATION(65),                   // Расторжение
    DECLINED_TERMINATION(57)           // Отклонено расторжение
}

// Метрики RabbitMQ обработки
const val INS_QUEUE_UNPAID_BPM_ADD_OK = "ins.queue.unpaid.bpm.add.ok"
const val INS_QUEUE_UNPAID_BPM_ADD_ERROR = "ins.queue.unpaid.bpm.add.error"
const val INS_QUEUE_UNPAID_BPM_CANCEL_OK = "ins.queue.unpaid.bpm.cancel.ok"
const val INS_QUEUE_UNPAID_BPM_CANCEL_ERROR = "ins.queue.unpaid.bpm.cancel.error"
const val INS_QUEUE_UNPAID_BPM_SOM_OK = "ins.queue.unpaid.bpm.som.ok"
const val INS_QUEUE_UNPAID_BPM_SOM_ERROR = "ins.queue.unpaid.bpm.som.error"

management:
  endpoints:
    web:
      exposure:
        include: health,info,prometheus,metrics,loggers,httpexchanges
  endpoint:
    loggers:
      enabled: false
    probes:
      enabled: false
  httpexchanges:
    recording:
      enabled: false

// Конверсия подключения страховок
unpaid_conversion_rate_total{product_code="MEDCARE"}
unpaid_conversion_rate_total{product_code="SOMLT"}
unpaid_conversion_rate_total{product_code="TRVL10"}

// Время обработки заявки
unpaid_processing_duration_seconds{product_code="MEDCARE"}
unpaid_processing_duration_seconds{product_code="SOMLT"}

// Количество активных заявок
unpaid_active_applications_total{stage="INIT"}
unpaid_active_applications_total{stage="CREATE_APPLICATION"}
unpaid_active_applications_total{stage="GENERATE_POLICY"}

// Подключения по каналам
unpaid_connections_total{channel="front"}
unpaid_connections_total{channel="back"}

// Успешность по каналам
unpaid_channel_success_rate{channel="front"}
unpaid_channel_success_rate{channel="back"}

// Экземпляры процессов
camunda_process_instances_total{process_definition_id="bpmUnpaidProcessId"}
camunda_process_instances_active{process_definition_id="bpmUnpaidProcessId"}
camunda_process_instances_completed{process_definition_id="bpmUnpaidProcessId"}

// Время выполнения задач
camunda_task_duration_seconds{task_id="getInsuranceProgram"}
camunda_task_duration_seconds{task_id="createApplication"}
camunda_task_duration_seconds{task_id="generatePolicy"}
camunda_task_duration_seconds{task_id="signOperation"}

// Ошибки задач
camunda_task_failures_total{task_id="getInsuranceProgram"}
camunda_task_failures_total{task_id="createApplication"}

// Время в стадиях
unpaid_stage_duration_seconds{stage="INIT"}
unpaid_stage_duration_seconds{stage="CHECK_INSURANCE"}
unpaid_stage_duration_seconds{stage="CREATE_APPLICATION"}
unpaid_stage_duration_seconds{stage="GENERATE_POLICY"}

// Количество заявок в стадиях
unpaid_stage_count{stage="INIT"}
unpaid_stage_count{stage="CHECK_INSURANCE"}
unpaid_stage_count{stage="CREATE_APPLICATION"}

// Время ответа
http_client_duration_seconds{service="insurance-application"}
http_client_duration_seconds{service="mdm"}
http_client_duration_seconds{service="catalogue"}
http_client_duration_seconds{service="content"}
http_client_duration_seconds{service="digital-sign"}

// Ошибки интеграций
http_client_errors_total{service="insurance-application"}
http_client_errors_total{service="mdm"}
http_client_errors_total{service="catalogue"}

// Доступность сервисов
http_client_up{service="insurance-application"}
http_client_up{service="mdm"}
http_client_up{service="catalogue"}

// Обработка сообщений
rabbit_message_processed_total{status="NEW"}
rabbit_message_processed_total{status="START_INSTANCE"}
rabbit_message_processed_total{status="CREATE_APP"}
rabbit_message_processed_total{status="DONE"}

// Ошибки обработки
rabbit_message_errors_total{status="DECLINED_INSTANCE"}
rabbit_message_errors_total{status="DECLINED_APP"}
rabbit_message_errors_total{status="DECLINED_POLICY"}

// Память
jvm_memory_used_bytes{area="heap"}
jvm_memory_used_bytes{area="nonheap"}

// Сборщик мусора
jvm_gc_duration_seconds
jvm_gc_collection_count

// Потоки
jvm_threads_current
jvm_threads_daemon

// Соединения
hikaricp_connections_active
hikaricp_connections_idle
hikaricp_connections_max

// Производительность
hikaricp_connection_acquire_seconds
hikaricp_connection_usage_seconds

// Hit/Miss ratio
cache_hits_total{cache="clientCache"}
cache_misses_total{cache="clientCache"}
cache_size{cache="clientCache"}

// Ошибки по типам
unpaid_errors_total{error_type="MDM_ERROR"}
unpaid_errors_total{error_type="CATALOGUE_ERROR"}
unpaid_errors_total{error_type="APPLICATION_ERROR"}
unpaid_errors_total{error_type="TIMEOUT"}
unpaid_errors_total{error_type="DIGITAL_SIGN_ERROR"}

// Ошибки по продуктам
unpaid_errors_total{product_code="MEDCARE"}
unpaid_errors_total{product_code="SOMLT"}
unpaid_errors_total{product_code="TRVL10"}

// Статусы
unpaid_application_status_total{status="CREATED"}
unpaid_application_status_total{status="CONFIRMED"}
unpaid_application_status_total{status="DONE"}
unpaid_application_status_total{status="REJECTED"}

// Статусы подписания
unpaid_digital_sign_total{status="SUCCESS"}
unpaid_digital_sign_total{status="FAILED"}
unpaid_digital_sign_total{status="TIMEOUT"}

// Время подписания
unpaid_digital_sign_duration_seconds{status="SUCCESS"}
unpaid_digital_sign_duration_seconds{status="FAILED"}

// SLA-метрики времени обработки (гистограммы)
unpaid_processing_duration_seconds_bucket{product_code="MEDCARE", le="900"} // 15 мин SLA
unpaid_processing_duration_seconds_bucket{product_code="MEDCARE", le="1800"} // 30 мин критично
unpaid_processing_duration_seconds_bucket{product_code="MEDCARE", le="3600"} // 1 час критично

// P95/P99 для критичных операций
http_client_duration_seconds{service="mdm", quantile="0.95"}
http_client_duration_seconds{service="mdm", quantile="0.99"}
http_client_duration_seconds{service="insurance-application", quantile="0.95"}

// Время выполнения ключевых задач с SLA
camunda_task_duration_seconds_bucket{task_id="createApplication", le="60"} // 1 мин SLA
camunda_task_duration_seconds_bucket{task_id="generatePolicy", le="300"} // 5 мин SLA
camunda_task_duration_seconds_bucket{task_id="getClientInfo", le="30"} // 30 сек SLA

// Процент заявок, обработанных в рамках SLA (15 минут)
unpaid_sla_processing_compliance_ratio = 
  histogram_quantile(0.95, unpaid_processing_duration_seconds_bucket) <= 900

// Среднее время обработки за период
unpaid_avg_processing_time = 
  rate(unpaid_processing_duration_seconds_sum[24h]) / 
  rate(unpaid_processing_duration_seconds_count[24h])

// Общий SLA по доступности (99.9% uptime)
unpaid_sla_availability_ratio = 
  (rate(unpaid_requests_total[24h]) - rate(unpaid_errors_total[24h])) / 
  rate(unpaid_requests_total[24h])

// Error rate по стадиям процесса
unpaid_error_rate_by_stage{stage="CREATE_APPLICATION"} = 
  rate(unpaid_errors_total{stage="CREATE_APPLICATION"}[5m]) / 
  rate(unpaid_requests_total{stage="CREATE_APPLICATION"}[5m])

unpaid_error_rate_by_stage{stage="GENERATE_POLICY"} = 
  rate(unpaid_errors_total{stage="GENERATE_POLICY"}[5m]) / 
  rate(unpaid_requests_total{stage="GENERATE_POLICY"}[5m])

// Критичные vs некритичные ошибки
unpaid_critical_error_rate = 
  rate(unpaid_errors_total{severity="critical"}[5m]) / 
  rate(unpaid_requests_total[5m])

unpaid_business_error_rate = 
  rate(unpaid_errors_total{type="business"}[5m]) / 
  rate(unpaid_requests_total[5m])

// Error rate по интеграциям
integration_error_rate{service="mdm"} = 
  rate(http_client_errors_total{service="mdm"}[5m]) / 
  rate(http_client_requests_total{service="mdm"}[5m])

// Пропускная способность по продуктам
unpaid_throughput_per_minute{product_code="MEDCARE"} = 
  rate(unpaid_connections_total{product_code="MEDCARE"}[1m]) * 60

unpaid_throughput_per_hour{product_code="MEDCARE"} = 
  rate(unpaid_connections_total{product_code="MEDCARE"}[1h]) * 3600

// SLA по объему обработки
unpaid_daily_volume_sla{product_code="MEDCARE"} // целевой объем в день: 10000
unpaid_current_daily_volume{product_code="MEDCARE"} = 
  increase(unpaid_connections_total{product_code="MEDCARE"}[24h])

// Пиковая нагрузка vs целевая
unpaid_peak_throughput_sla = 100 // заявок в минуту
unpaid_current_peak_throughput = 
  max_over_time(rate(unpaid_connections_total[1m])[1h]) * 60

// Capacity utilization
unpaid_capacity_utilization = 
  rate(unpaid_connections_total[5m]) * 60 / unpaid_max_capacity_per_minute

// Composite SLA Score (0-100%)
unpaid_composite_sla_score = (
  (unpaid_sla_availability_ratio * 0.4) +           // 40% вес доступности
  (unpaid_sla_processing_compliance_ratio * 0.3) +  // 30% вес времени обработки
  (unpaid_throughput_compliance_ratio * 0.2) +      // 20% вес пропускной способности
  (unpaid_quality_score * 0.1)                      // 10% вес качества
) * 100

// Детализация по продуктам
unpaid_product_sla_score{product_code="MEDCARE"}
unpaid_product_sla_score{product_code="SOMLT"}
unpaid_product_sla_score{product_code="TRVL10"}

@Configuration
class MetricsConfiguration {
    
    @Bean
    fun meterRegistry(): MeterRegistry {
        return PrometheusMeterRegistry(PrometheusConfig.DEFAULT)
    }
    
    @Bean
    fun unpaidMetrics(meterRegistry: MeterRegistry): UnpaidMetrics {
        return UnpaidMetrics(meterRegistry)
    }
    
    @Bean
    fun slaMetrics(meterRegistry: MeterRegistry): SLAMetrics {
        return SLAMetrics(meterRegistry)
    }
}

@Component
class UnpaidMetrics(private val meterRegistry: MeterRegistry) {
    
    // Основные счетчики
    private val conversionCounter = Counter.builder("unpaid_conversion_total")
        .description("Количество успешных подключений")
        .tag("product_code", "unknown")
        .register(meterRegistry)
    
    private val errorCounter = Counter.builder("unpaid_errors_total")
        .description("Количество ошибок")
        .tag("error_type", "unknown")
        .tag("product_code", "unknown")
        .tag("severity", "unknown")
        .register(meterRegistry)
    
    private val requestCounter = Counter.builder("unpaid_requests_total")
        .description("Общее количество запросов")
        .tag("product_code", "unknown")
        .tag("stage", "unknown")
        .register(meterRegistry)
    
    // SLA Таймеры с гистограммами
    private val processingTimer = Timer.builder("unpaid_processing_duration_seconds")
        .description("Время обработки заявки")
        .tag("product_code", "unknown")
        .serviceLevelObjectives(
            Duration.ofMinutes(5),   // быстрая обработка
            Duration.ofMinutes(15),  // SLA целевое время
            Duration.ofMinutes(30),  // критичное время
            Duration.ofHours(1)      // максимальное время
        )
        .register(meterRegistry)
    
    private val taskTimer = Timer.builder("camunda_task_duration_seconds")
        .description("Время выполнения задач Camunda")
        .tag("task_id", "unknown")
        .serviceLevelObjectives(
            Duration.ofSeconds(30),  // быстрая задача
            Duration.ofMinutes(1),   // нормальная задача
            Duration.ofMinutes(5),   // медленная задача
            Duration.ofMinutes(10)   // критично медленная
        )
        .register(meterRegistry)
    
    // Гейджи для активных состояний
    private val activeApplicationsGauge = Gauge.builder("unpaid_active_applications_total")
        .description("Количество активных заявок")
        .tag("stage", "unknown")
        .register(meterRegistry, AtomicLong(0))
    
    // SLA Гейджи
    private val slaComplianceGauge = Gauge.builder("unpaid_sla_compliance_ratio")
        .description("Соответствие SLA (0-1)")
        .tag("sla_type", "unknown")
        .tag("product_code", "unknown")
        .register(meterRegistry, AtomicDouble(0.0))
    
    // Методы для записи метрик
    fun recordConversion(productCode: String) {
        conversionCounter.increment(Tags.of("product_code", productCode))
        recordRequest(productCode, "FINALIZATION_SUCCESSFULLY")
    }
    
    fun recordError(errorType: String, productCode: String, severity: String = "normal", stage: String = "unknown") {
        errorCounter.increment(Tags.of(
            "error_type", errorType, 
            "product_code", productCode,
            "severity", severity
        ))
        recordRequest(productCode, stage)
    }
    
    fun recordRequest(productCode: String, stage: String) {
        requestCounter.increment(Tags.of("product_code", productCode, "stage", stage))
    }
    
    fun recordProcessingTime(productCode: String, duration: Duration) {
        processingTimer.record(duration, Tags.of("product_code", productCode))
    }
    
    fun recordTaskDuration(taskId: String, duration: Duration) {
        taskTimer.record(duration, Tags.of("task_id", taskId))
    }
    
    fun setActiveApplications(stage: String, count: Long) {
        activeApplicationsGauge.set(count.toDouble(), Tags.of("stage", stage))
    }
    
    fun updateSLACompliance(slaType: String, productCode: String, ratio: Double) {
        slaComplianceGauge.set(ratio, Tags.of("sla_type", slaType, "product_code", productCode))
    }
}

@Component
class SLAMetrics(private val meterRegistry: MeterRegistry) {
    
    private val slaViolationCounter = Counter.builder("unpaid_sla_violations_total")
        .description("Количество нарушений SLA")
        .tag("sla_type", "unknown")
        .tag("severity", "unknown")
        .register(meterRegistry)
    
    private val compositeScoreGauge = Gauge.builder("unpaid_composite_sla_score")
        .description("Общий SLA Score (0-100)")
        .tag("product_code", "unknown")
        .register(meterRegistry, AtomicDouble(100.0))
    
    // Методы для SLA
    fun recordSLAViolation(slaType: String, severity: String = "warning") {
        slaViolationCounter.increment(Tags.of("sla_type", slaType, "severity", severity))
    }
    
    fun updateCompositeScore(productCode: String, score: Double) {
        compositeScoreGauge.set(score, Tags.of("product_code", productCode))
    }
    
    fun calculateAvailabilitySLA(): Double {
        // Расчет SLA доступности на основе метрик
        val totalRequests = meterRegistry.counter("unpaid_requests_total").count()
        val totalErrors = meterRegistry.counter("unpaid_errors_total").count()
        
        return if (totalRequests > 0) {
            (totalRequests - totalErrors) / totalRequests
        } else 1.0
    }
    
    fun calculateProcessingTimeSLA(): Double {
        // Расчет SLA времени обработки
        val timer = meterRegistry.timer("unpaid_processing_duration_seconds")
        val snapshot = timer.takeSnapshot()
        val p95 = snapshot.percentileValue(0.95)
        
        return if (p95 <= 900.0) 1.0 else 900.0 / p95 // 15 минут = 900 секунд
    }
}

@Component
class CamundaMetricsListener(
    private val unpaidMetrics: UnpaidMetrics,
    private val slaMetrics: SLAMetrics
) {
    
    @EventListener
    fun onProcessInstanceEvent(event: ProcessInstanceEvent) {
        when (event) {
            is ProcessInstanceStartedEvent -> {
                unpaidMetrics.recordRequest(
                    extractProductCode(event.variables), 
                    "INIT"
                )
            }
            is ProcessInstanceCompletedEvent -> {
                val duration = Duration.between(event.startTime, event.endTime)
                val productCode = extractProductCode(event.variables)
                
                unpaidMetrics.recordProcessingTime(productCode, duration)
                
                // Проверка SLA нарушений
                if (duration.toSeconds() > 900) { // 15 минут
                    slaMetrics.recordSLAViolation("processing_time", "warning")
                }
                if (duration.toSeconds() > 1800) { // 30 минут
                    slaMetrics.recordSLAViolation("processing_time", "critical")
                }
                
                unpaidMetrics.recordConversion(productCode)
            }
        }
    }
    
    @EventListener
    fun onTaskEvent(event: TaskEvent) {
        when (event) {
            is TaskCompletedEvent -> {
                val duration = Duration.between(event.task.createTime, event.task.endTime)
                unpaidMetrics.recordTaskDuration(event.task.taskDefinitionKey, duration)
                
                // SLA проверки для критичных задач
                checkTaskSLA(event.task.taskDefinitionKey, duration)
            }
            is TaskFailedEvent -> {
                unpaidMetrics.recordError(
                    "TASK_FAILURE", 
                    extractProductCode(event.variables),
                    "critical",
                    event.task.taskDefinitionKey
                )
                slaMetrics.recordSLAViolation("task_execution", "critical")
            }
        }
    }
    
    private fun checkTaskSLA(taskId: String, duration: Duration) {
        val slaLimits = mapOf(
            "createApplication" to Duration.ofMinutes(1),
            "generatePolicy" to Duration.ofMinutes(5),
            "getClientInfo" to Duration.ofSeconds(30),
            "signOperation" to Duration.ofMinutes(2)
        )
        
        val slaLimit = slaLimits[taskId]
        if (slaLimit != null && duration > slaLimit) {
            slaMetrics.recordSLAViolation("task_sla_$taskId", "warning")
        }
    }
}

@Component
class HttpClientMetricsInterceptor(
    private val unpaidMetrics: UnpaidMetrics,
    private val slaMetrics: SLAMetrics
) : ClientHttpRequestInterceptor {
    
    override fun intercept(
        request: HttpRequest, 
        body: ByteArray, 
        execution: ClientHttpRequestExecution
    ): ClientHttpResponse {
        val startTime = System.currentTimeMillis()
        val serviceName = extractServiceName(request.uri)
        
        return try {
            val response = execution.execute(request, body)
            val duration = Duration.ofMillis(System.currentTimeMillis() - startTime)
            
            // Запись успешного запроса
            recordHttpRequest(serviceName, duration, true, response.statusCode.value())
            
            // SLA проверки для внешних сервисов
            checkIntegrationSLA(serviceName, duration)
            
            response
        } catch (e: Exception) {
            val duration = Duration.ofMillis(System.currentTimeMillis() - startTime)
            
            // Запись неуспешного запроса
            recordHttpRequest(serviceName, duration, false, 0)
            
            // Критичное нарушение SLA при ошибках интеграции
            slaMetrics.recordSLAViolation("integration_failure", "critical")
            
            throw e
        }
    }
    
    private fun recordHttpRequest(service: String, duration: Duration, success: Boolean, statusCode: Int) {
        // Метрики времени ответа
        Timer.builder("http_client_duration_seconds")
            .tag("service", service)
            .tag("success", success.toString())
            .register(meterRegistry)
            .record(duration)
            
        // Счетчики запросов
        Counter.builder("http_client_requests_total")
            .tag("service", service)
            .tag("status_code", statusCode.toString())
            .register(meterRegistry)
            .increment()
            
        if (!success) {
            Counter.builder("http_client_errors_total")
                .tag("service", service)
                .register(meterRegistry)
                .increment()
        }
    }
    
    private fun checkIntegrationSLA(service: String, duration: Duration) {
        val slaLimits = mapOf(
            "mdm" to Duration.ofSeconds(5),
            "insurance-application" to Duration.ofSeconds(10),
            "catalogue" to Duration.ofSeconds(3),
            "digital-sign" to Duration.ofSeconds(15)
        )
        
        val slaLimit = slaLimits[service]
        if (slaLimit != null && duration > slaLimit) {
            slaMetrics.recordSLAViolation("integration_sla_$service", "warning")
        }
    }
}

@Component
class SLACalculatorService(
    private val unpaidMetrics: UnpaidMetrics,
    private val slaMetrics: SLAMetrics,
    private val meterRegistry: MeterRegistry
) {
    
    @Scheduled(fixedRate = 60000) // каждую минуту
    fun calculateRealTimeSLA() {
        val productCodes = listOf("MEDCARE", "SOMLT", "TRVL10")
        
        productCodes.forEach { productCode ->
            val compositeScore = calculateCompositeScore(productCode)
            slaMetrics.updateCompositeScore(productCode, compositeScore)
            
            // Обновление отдельных SLA показателей
            updateIndividualSLAs(productCode)
        }
    }
    
    private fun calculateCompositeScore(productCode: String): Double {
        val availabilityScore = calculateAvailabilityScore(productCode)
        val latencyScore = calculateLatencyScore(productCode)
        val throughputScore = calculateThroughputScore(productCode)
        val qualityScore = calculateQualityScore(productCode)
        
        return (availabilityScore * 0.4 + latencyScore * 0.3 + throughputScore * 0.2 + qualityScore * 0.1) * 100
    }
    
    private fun calculateAvailabilityScore(productCode: String): Double {
        val totalRequests = getCounterValue("unpaid_requests_total", "product_code", productCode)
        val totalErrors = getCounterValue("unpaid_errors_total", "product_code", productCode)
        
        return if (totalRequests > 0) {
            (totalRequests - totalErrors) / totalRequests
        } else 1.0
    }
    
    private fun calculateLatencyScore(productCode: String): Double {
        val timer = getTimer("unpaid_processing_duration_seconds", "product_code", productCode)
        val p95 = timer?.takeSnapshot()?.percentileValue(0.95) ?: 0.0
        
        return when {
            p95 <= 300.0 -> 1.0      // отлично: <= 5 минут
            p95 <= 900.0 -> 0.8      // хорошо: <= 15 минут (SLA)
            p95 <= 1800.0 -> 0.5     // удовлетворительно: <= 30 минут
            else -> 0.2              // плохо: > 30 минут
        }
    }
    
    private fun calculateThroughputScore(productCode: String): Double {
        val currentThroughput = getCurrentThroughput(productCode)
        val targetThroughput = getTargetThroughput(productCode)
        
        return minOf(1.0, currentThroughput / targetThroughput)
    }
    
    private fun calculateQualityScore(productCode: String): Double {
        val businessErrors = getCounterValue("unpaid_errors_total", 
            mapOf("product_code" to productCode, "type" to "business"))
        val totalRequests = getCounterValue("unpaid_requests_total", "product_code", productCode)
        
        return if (totalRequests > 0) {
            1.0 - (businessErrors / totalRequests)
        } else 1.0
    }
}

{
  "dashboard": {
    "title": "Insurance Unpaid SLA Dashboard",
    "panels": [
      {
        "title": "SLA Compliance Overview",
        "type": "stat",
        "targets": [
          {
            "expr": "unpaid_composite_sla_score",
            "legendFormat": "{{product_code}} SLA Score",
            "refId": "A"
          }
        ],
        "fieldConfig": {
          "defaults": {
            "thresholds": {
              "steps": [
                {"color": "red", "value": 0},
                {"color": "yellow", "value": 90},
                {"color": "green", "value": 95}
              ]
            },
            "unit": "percent"
          }
        }
      },
      {
        "title": "Real-time SLA Status",
        "type": "table",
        "targets": [
          {
            "expr": "unpaid_sla_compliance_ratio * 100",
            "legendFormat": "{{sla_type}} - {{product_code}}",
            "format": "table"
          }
        ]
      },
      {
        "title": "SLA Violations (Last 24h)",
        "type": "bargauge",
        "targets": [
          {
            "expr": "increase(unpaid_sla_violations_total[24h])",
            "legendFormat": "{{sla_type}} {{severity}}"
          }
        ]
      }
    ]
  }
}

{
  "panels": [
    {
      "title": "Processing Time SLA Compliance",
      "type": "graph",
      "targets": [
        {
          "expr": "histogram_quantile(0.95, unpaid_processing_duration_seconds_bucket)",
          "legendFormat": "P95 Processing Time"
        },
        {
          "expr": "900",
          "legendFormat": "SLA Threshold (15 min)"
        }
      ]
    },
    {
      "title": "Integration SLA Status",
      "type": "heatmap",
      "targets": [
        {
          "expr": "rate(http_client_duration_seconds_bucket[5m])",
          "legendFormat": "{{service}}"
        }
      ]
    },
    {
      "title": "Task Execution SLA",
      "type": "table",
      "targets": [
        {
          "expr": "histogram_quantile(0.95, camunda_task_duration_seconds_bucket) by (task_id)",
          "format": "table"
        }
      ]
    }
  ]
}

# Критичное нарушение SLA времени обработки
- alert: ProcessingSLACriticalViolation
  expr: histogram_quantile(0.95, unpaid_processing_duration_seconds_bucket) > 1800
  for: 1m
  labels:
    severity: critical
    sla: "processing_time"
    impact: "high"
  annotations:
    summary: "КРИТИЧНО: Время обработки превышает 30 минут"
    description: "P95 времени обработки: {{ $value | humanizeDuration }}"
    runbook_url: "https://wiki.vtb.ru/runbooks/unpaid-processing-sla"

# Нарушение SLA времени обработки
- alert: ProcessingSLAViolation
  expr: histogram_quantile(0.95, unpaid_processing_duration_seconds_bucket) > 900
  for: 2m
  labels:
    severity: warning
    sla: "processing_time"
    impact: "medium"
  annotations:
    summary: "Нарушение SLA: время обработки превышает 15 минут"
    description: "P95 времени обработки: {{ $value | humanizeDuration }}"

# Критичный error rate > 5%
- alert: ErrorRateSLACriticalViolation
  expr: |
    (
      rate(unpaid_errors_total[5m]) / 
      rate(unpaid_requests_total[5m])
    ) > 0.05
  for: 1m
  labels:
    severity: critical
    sla: "error_rate"
    impact: "high"
  annotations:
    summary: "КРИТИЧНО: Error rate превышает 5%"
    description: "Текущий error rate: {{ $value | humanizePercentage }}"

# Нарушение SLA error rate > 1%
- alert: ErrorRateSLAViolation
  expr: |
    (
      rate(unpaid_errors_total[5m]) / 
      rate(unpaid_requests_total[5m])
    ) > 0.01
  for: 2m
  labels:
    severity: warning
    sla: "error_rate"
    impact: "medium"
  annotations:
    summary: "Нарушение SLA: error rate превышает 1%"
    description: "Текущий error rate: {{ $value | humanizePercentage }}"

# Нарушение SLA доступности < 99.9%
- alert: AvailabilitySLAViolation
  expr: |
    (
      (rate(unpaid_requests_total[1h]) - rate(unpaid_errors_total{severity="critical"}[1h])) / 
      rate(unpaid_requests_total[1h])
    ) < 0.999
  for: 5m
  labels:
    severity: critical
    sla: "availability"
    impact: "high"
  annotations:
    summary: "Нарушение SLA доступности"
    description: "Доступность: {{ $value | humanizePercentage }}"

# Падение throughput ниже SLA
- alert: ThroughputSLAViolation
  expr: rate(unpaid_connections_total[5m]) * 60 < 100
  for: 3m
  labels:
    severity: warning
    sla: "throughput"
    impact: "medium"
  annotations:
    summary: "Нарушение SLA пропускной способности"
    description: "Текущий throughput: {{ $value }} заявок/мин (SLA: 100)"

# Composite SLA Score критично низкий
- alert: CompositeSLACritical
  expr: unpaid_composite_sla_score < 90
  for: 5m
  labels:
    severity: critical
    sla: "composite"
    impact: "high"
  annotations:
    summary: "Критично низкий общий SLA Score"
    description: "SLA Score для {{ $labels.product_code }}: {{ $value }}%"

# Приближение к нарушению SLA обработки (80% от лимита)
- alert: ProcessingSLAWarning
  expr: histogram_quantile(0.95, unpaid_processing_duration_seconds_bucket) > 720
  for: 5m
  labels:
    severity: warning
    sla: "processing_time_warning"
    impact: "low"
  annotations:
    summary: "Приближение к нарушению SLA времени обработки"
    description: "P95 времени: {{ $value | humanizeDuration }} (80% от SLA)"

# Тренд роста error rate
- alert: ErrorRateTrendWarning
  expr: |
    (
      rate(unpaid_errors_total[5m]) / 
      rate(unpaid_requests_total[5m])
    ) > 0.005 and
    (
      rate(unpaid_errors_total[10m]) - 
      rate(unpaid_errors_total[10m] offset 10m)
    ) > 0
  for: 2m
  labels:
    severity: warning
    sla: "error_rate_trend"
    impact: "low"
  annotations:
    summary: "Растущий тренд error rate"
    description: "Error rate: {{ $value | humanizePercentage }} и растет"

# Медленные интеграции приближаются к SLA
- alert: IntegrationLatencyWarning
  expr: histogram_quantile(0.95, http_client_duration_seconds_bucket) > 8
  for: 3m
  labels:
    severity: warning
    sla: "integration_latency"
    impact: "medium"
  annotations:
    summary: "Медленный ответ интеграции {{ $labels.service }}"
    description: "P95 latency: {{ $value }}s"

# Снижение throughput (предупреждение)
- alert: ThroughputDeclineWarning
  expr: |
    rate(unpaid_connections_total[5m]) * 60 < 120 and
    rate(unpaid_connections_total[5m]) * 60 > 100
  for: 10m
  labels:
    severity: warning
    sla: "throughput_decline"
    impact: "low"
  annotations:
    summary: "Снижение пропускной способности"
    description: "Throughput: {{ $value }} заявок/мин"

# Восстановление SLA после нарушения
- alert: SLARecovered
  expr: |
    (
      histogram_quantile(0.95, unpaid_processing_duration_seconds_bucket) <= 900
    ) and
    (
      ALERTS{alertname="ProcessingSLAViolation"} == 1
    )
  for: 2m
  labels:
    severity: info
    sla: "processing_time_recovery"
  annotations:
    summary: "SLA времени обработки восстановлено"
    description: "P95 времени: {{ $value | humanizeDuration }}"

- alert: ErrorRateSLARecovered
  expr: |
    (
      rate(unpaid_errors_total[5m]) / 
      rate(unpaid_requests_total[5m])
    ) <= 0.01 and
    (
      ALERTS{alertname="ErrorRateSLAViolation"} == 1
    )
  for: 5m
  labels:
    severity: info
    sla: "error_rate_recovery"
  annotations:
    summary: "SLA error rate восстановлено"
    description: "Error rate: {{ $value | humanizePercentage }}"

# Конфигурация AlertManager для SLA
route:
  group_by: ['sla', 'severity']
  group_wait: 10s
  group_interval: 5m
  repeat_interval: 30m
  routes:
  # Критичные SLA нарушения - немедленно
  - match:
      severity: critical
      sla: processing_time|error_rate|availability
    receiver: 'sla-critical'
    group_wait: 0s
    repeat_interval: 15m
  
  # Предупреждения SLA - с задержкой
  - match:
      severity: warning
    receiver: 'sla-warnings'
    group_wait: 2m
    repeat_interval: 1h
  
  # Восстановление SLA
  - match:
      severity: info
    receiver: 'sla-recovery'

receivers:
# Критичные SLA нарушения
- name: 'sla-critical'
  slack_configs:

  - api_url: 'https://hooks.slack.com/services/...'
    channel: '#sla-critical'
    title: '🚨 КРИТИЧНОЕ НАРУШЕНИЕ SLA: {{ .GroupLabels.sla }}'
    text: |
      {{ range .Alerts }}
      **Alert:** {{ .Annotations.summary }}
      **Description:** {{ .Annotations.description }}
      **Impact:** {{ .Labels.impact }}
      **Runbook:** {{ .Annotations.runbook_url }}
      {{ end }}
    send_resolved: true
  
  pagerduty_configs:

  - service_key: 'sla-critical-key'
    severity: 'critical'
    description: '{{ .GroupLabels.sla }} SLA нарушен'

# Предупреждения SLA
- name: 'sla-warnings'
  slack_configs:

  - api_url: 'https://hooks.slack.com/services/...'
    channel: '#sla-monitoring'
    title: '⚠️ SLA Warning: {{ .GroupLabels.sla }}'
    text: |
      {{ range .Alerts }}
      **Alert:** {{ .Annotations.summary }}
      **Description:** {{ .Annotations.description }}
      {{ end }}
    send_resolved: true

# Восстановление SLA
- name: 'sla-recovery'
  slack_configs:

  - api_url: 'https://hooks.slack.com/services/...'
    channel: '#sla-monitoring'
    title: '✅ SLA Восстановлено: {{ .GroupLabels.sla }}'
    text: |
      {{ range .Alerts }}
      **Recovered:** {{ .Annotations.summary }}
      {{ end }}

// Мгновенное обнаружение нарушений SLA
- alert: ProcessingSLACritical
  expr: histogram_quantile(0.95, unpaid_processing_duration_seconds_bucket) > 1800
  for: 1m  // обнаружение через 1 минуту вместо часов

// Предупреждение до нарушения SLA
- alert: ProcessingSLAWarning
  expr: histogram_quantile(0.95, unpaid_processing_duration_seconds_bucket) > 720
  for: 5m  // предупреждение за 8 минут до нарушения SLA

// Выявление связанных проблем
- alert: CascadingFailure
  expr: |
    (http_client_errors_total{service="mdm"} > 10) and
    (unpaid_errors_total{stage="GET_CLIENT_INFO"} > 5)
  for: 1m  // обнаружение цепочки проблем

// Время от возникновения до обнаружения
incident_detection_time_seconds{severity="critical"}
incident_detection_time_seconds{severity="warning"}

// Время от обнаружения до реагирования
incident_response_time_seconds{team="ops"}
incident_response_time_seconds{team="dev"}

// Процент инцидентов, обнаруженных автоматически
automated_detection_ratio{period="24h"}