// ✅ Core messaging metrics — export these from every consumer
import { metrics } from '@opentelemetry/api'

const meter = metrics.getMeter('messaging-consumer', '1.0.0')

const consumerLag        = meter.createObservableGauge('messaging.consumer.lag')
const dlqDepth           = meter.createObservableGauge('messaging.dlq.depth')
const processingDuration = meter.createHistogram('messaging.message.processing.duration.ms')
const errorRate          = meter.createCounter('messaging.message.errors.total')

// Observe lag per partition per consumer group
consumerLag.addCallback(async (result) => {
  const offsets = await admin.fetchOffsets({ groupId: 'fulfillment-service', topics: ['orders.created'] })
  for (const topic of offsets) {
    for (const partition of topic.partitions) {
      const end = await admin.fetchTopicOffsetsByTimestamp(topic.topic, Date.now())
      const lag = Number(end[partition.partition].offset) - Number(partition.offset)
      result.observe(lag, {
        'messaging.consumer.group': 'fulfillment-service',
        'messaging.topic':          topic.topic,
        'messaging.partition':      partition.partition.toString(),
      })
    }
  }
})

// Observe DLQ depth — alert when non-zero
dlqDepth.addCallback(async (result) => {
  const dlqOffset = await admin.fetchTopicOffsets('orders.dlq')
  for (const partition of dlqOffset) {
    result.observe(Number(partition.offset), {
      'messaging.dlq.topic': 'orders.dlq',
      'messaging.partition': partition.partition.toString(),
    })
  }
})

// ❌ No trace propagation — producer and consumer are invisible to each other
await producer.send({
  topic:    'orders.created',
  messages: [{ key: order.id, value: JSON.stringify(order) }]
  // No trace context — consumer's spans appear as disconnected roots in Jaeger/Grafana
})

// ✅ Producer: inject trace context into message headers
import { context, propagation, trace } from '@opentelemetry/api'

const tracer = trace.getTracer('order-service')

async function publishOrderCreated(order: Order): Promise<void> {
  const span = tracer.startSpan('messaging.produce', {
    kind: SpanKind.PRODUCER,
    attributes: {
      'messaging.system':          'kafka',
      'messaging.destination':     'orders.created',
      'messaging.destination.kind':'topic',
      'messaging.message.id':       order.id,
    }
  })

  await context.with(trace.setSpan(context.active(), span), async () => {
    // Inject current trace context into headers
    const headers: Record<string, string> = {}
    propagation.inject(context.active(), headers)
    // headers now contains: { traceparent: '00-abc...', tracestate: '...' }

    await producer.send({
      topic:    'orders.created',
      messages: [{
        key:     order.id,
        value:   JSON.stringify(order),
        headers: {
          traceparent:         headers['traceparent'],
          tracestate:          headers['tracestate'] ?? '',
          'x-correlation-id':  order.id,          // business-level ID for log correlation
          'x-causation-id':    span.spanContext().spanId,
        }
      }]
    })
    span.end()
  })
}

// ✅ Consumer: extract trace context and link to parent span
await consumer.run({
  eachMessage: async ({ message, topic, partition }) => {
    // Extract trace context from headers
    const extractedContext = propagation.extract(context.active(), {
      traceparent: message.headers?.['traceparent']?.toString(),
      tracestate:  message.headers?.['tracestate']?.toString(),
    })

    const span = tracer.startSpan('messaging.consume', {
      kind: SpanKind.CONSUMER,
      attributes: {
        'messaging.system':            'kafka',
        'messaging.source':            topic,
        'messaging.kafka.partition':   partition,
        'messaging.kafka.offset':      message.offset,
        'messaging.message.id':        message.headers?.['x-correlation-id']?.toString(),
      }
    }, extractedContext)   // links to the producer's span via parent context

    const start = Date.now()
    try {
      await context.with(trace.setSpan(extractedContext, span), async () => {
        await processOrder(JSON.parse(message.value!.toString()))
      })
      processingDuration.record(Date.now() - start, { status: 'success' })
    } catch (err) {
      span.recordException(err as Error)
      span.setStatus({ code: SpanStatusCode.ERROR })
      errorRate.add(1, { 'error.type': (err as Error).constructor.name })
      processingDuration.record(Date.now() - start, { status: 'error' })
      throw err
    } finally {
      span.end()
    }
  }
})

// ✅ Kafka: DLQ with retry counter and dead-lettering
const MAX_RETRIES = 5

async function processWithDLQ(
  message: KafkaMessage,
  topic:   string,
  partition: number,
): Promise<void> {
  const retryCount = Number(message.headers?.['x-retry-count'] ?? 0)

  try {
    await processOrder(JSON.parse(message.value!.toString()))
  } catch (err) {
    const isRetryable = !(err instanceof NonRetryableError)

    if (isRetryable && retryCount < MAX_RETRIES) {
      // Exponential backoff: re-publish with incremented counter
      const delayMs = Math.min(1000 * 2 ** retryCount, 60_000)  // cap at 60s
      await new Promise(resolve => setTimeout(resolve, delayMs))

      await producer.send({
        topic:    topic,
        messages: [{
          key:     message.key,
          value:   message.value,
          headers: {
            ...message.headers,
            'x-retry-count':    (retryCount + 1).toString(),
            'x-original-topic': topic,
            'x-error-message':  (err as Error).message.slice(0, 256),
            traceparent:        message.headers?.['traceparent'],
          }
        }]
      })
    } else {
      // Max retries exceeded or non-retryable — send to DLQ
      await producer.send({
        topic:    `${topic}.dlq`,
        messages: [{
          key:     message.key,
          value:   message.value,
          headers: {
            ...message.headers,
            'x-dlq-reason':       retryCount >= MAX_RETRIES ? 'max-retries-exceeded' : 'non-retryable',
            'x-dlq-at':           new Date().toISOString(),
            'x-error-message':    (err as Error).message.slice(0, 256),
            'x-retry-count':      retryCount.toString(),
            'x-original-topic':   topic,
            'x-original-partition': partition.toString(),
            traceparent:          message.headers?.['traceparent'],  // preserve trace linkage
          }
        }]
      })
      // Increment the DLQ depth metric
      errorRate.add(1, { 'error.type': 'dlq', 'dlq.reason': 'max-retries-exceeded' })
    }
  }
}

// ✅ Poison message quarantine — detect non-retryable errors, skip and isolate
class NonRetryableError extends Error {
  constructor(message: string, public readonly reason: string) {
    super(message)
    this.name = 'NonRetryableError'
  }
}

async function deserializeMessage(rawValue: Buffer): Promise<OrderEvent> {
  try {
    return await registry.decode(rawValue)
  } catch (err) {
    // Schema mismatch, corrupt payload — no amount of retrying will fix this
    throw new NonRetryableError(
      `Deserialization failed: ${(err as Error).message}`,
      'schema-mismatch'
    )
  }
}

// Consumer: detect and quarantine immediately
await consumer.run({
  eachMessage: async ({ message, topic, partition }) => {
    try {
      const event = await deserializeMessage(message.value!)
      await processOrder(event)
    } catch (err) {
      if (err instanceof NonRetryableError) {
        // Skip retry loop — goes directly to DLQ + quarantine
        await quarantineMessage(message, topic, partition, err.reason)
        return   // ack the message — do NOT rethrow (would trigger infinite retry)
      }
      throw err   // retryable — propagate to retry logic
    }
  }
})

# Diagnose: check consumer lag trend + DB write latency correlation
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group fulfillment-service

# Check DB write latency (PostgreSQL)
SELECT query, mean_exec_time, calls
FROM pg_stat_statements
WHERE query ILIKE '%INSERT%' OR query ILIKE '%UPDATE%'
ORDER BY mean_exec_time DESC LIMIT 10;

// Fix: increase max.poll.interval.ms to match actual processing time
const consumer = kafka.consumer({
  groupId:             'fulfillment-service',
  sessionTimeout:      60_000,   // 60s
  heartbeatInterval:   10_000,   // 10s (< sessionTimeout / 3)
  // In Java client: maxPollIntervalMs: 300_000 (5 min)
  // In kafkajs: handled by eachBatch heartbeat() calls
})

# Pause a consumer group by setting all offsets to current position
# (consumers will not advance — new messages accumulate but are not processed)
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group fulfillment-service --reset-offsets --to-current \
  --topic orders.created --execute

// ✅ Handle OffsetOutOfRangeError gracefully
consumer.on(consumer.events.FETCH_START, async () => {
  // kafkajs emits fetch errors as events — handle gracefully
})

// Configure: on out-of-range, jump to earliest available offset (accept data gap)
const consumer = kafka.consumer({
  groupId:             'analytics-service',
  // On OffsetOutOfRangeError: reset to 'earliest' available offset
  // This means: accept that data was lost during the offline window
  // Alternative: reset to 'latest' — start from now, larger data gap
})

// Emit an alert: offset reset = data gap = must audit downstream read models
errorRate.add(1, { 'error.type': 'offset-out-of-range', 'action': 'reset-to-earliest' })