State divergence i reconciliation w systemach real-time telefonii: kiedy Redis mówi jedno, a rzeczywistość drugie

async function reconcileActiveCalls() {
  // 1. Get our view of active calls from Redis
  const redisCallIds = await redis.smembers("active_calls");
  // 2. Get authoritative view from PostgreSQL
  const dbActiveCalls = await db.campaignCalls.findMany({
    status: "in_progress",
    updatedAt: { gte: DateTime.now().minus({ minutes: 10 }) }
  });
  const dbCallIds = new Set(dbActiveCalls.map(c => c.callId));
  // 3. Get actual state from Telnyx
  const telnyxCalls = await telnyx.calls.list({ 
    status: "active",
    connectionId: config.telnyxConnectionId 
  });
  const telnyxCallIds = new Set(telnyxCalls.map(c => c.call_control_id));
  // 4. Find divergences
  const ghostCallsInRedis = redisCallIds.filter(id => 
    !telnyxCallIds.has(id) && !dbCallIds.has(id)
  );
  const missingInRedis = [...dbCallIds].filter(id => 
    !redisCallIds.includes(id)
  );
  const stalledInDB = dbActiveCalls.filter(call =>
    !telnyxCallIds.has(call.callId) &&
    DateTime.fromJSDate(call.updatedAt) < DateTime.now().minus({ minutes: 5 })
  );
  // 5. Fix divergences
  for (const callId of ghostCallsInRedis) {
    await redis.srem("active_calls", callId);
    metrics.increment("reconciliation.ghost_call_removed");
    logger.warn("Reconciliation: removed ghost call ${callId} from Redis");
  }
  for (const callId of missingInRedis) {
    await redis.sadd("active_calls", callId);
    metrics.increment("reconciliation.call_restored_to_redis");
  }
  for (const call of stalledInDB) {
    await markCallEnded(call.callId, "stalled_detected_in_reconciliation");
    metrics.increment("reconciliation.stalled_call_cleaned");
    logger.error("Reconciliation: force-ended stalled call ${call.callId}");
  }
}