// ❌ Broken pattern: Cache-Aside with Write-Update under concurrent requests
export async function updateUserProfile(userId: string, updates: Partial<User>) {
  // Thread 1 (Setting name = 'Alice') and Thread 2 (Setting name = 'Bob') execute concurrently
  await db.user.update({ where: { id: userId }, data: updates });

  // 💥 RACE CONDITION:
  // 1. Thread 1 updates DB to 'Alice'
  // 2. Thread 2 updates DB to 'Bob'
  // 3. Thread 2 sets Redis to 'Bob'
  // 4. Thread 1 (delayed by GC pause/network jitter) sets Redis to 'Alice'
  // RESULT: Database holds 'Bob', but Cache permanently holds 'Alice'!
  await redis.set(`user:${userId}`, JSON.stringify(updates));
}

// ✅ Resilient pattern: Cache-Aside with Write-Invalidate (Post-Commit Hook)
export async function updateUserProfileResilient(userId: string, updates: Partial<User>) {
  // 1. Execute DB mutation inside an ACID transaction
  await db.$transaction(async (tx) => {
    await tx.user.update({ where: { id: userId }, data: updates });
    
    // 2. Schedule cache deletion ONLY after the transaction successfully commits
    tx.afterCommit(async () => {
      // Invalidate the cache — next read will safely pull latest committed state from DB
      await redis.del(`user:${userId}`);
    });
  });
}

// ❌ Broken pattern: Cache-aside reading from an asynchronous read replica
export async function getUser(userId: string) {
  const cached = await redis.get(`user:${userId}`);
  if (cached) return JSON.parse(cached);

  // 💥 DANGER: dbReplica is lagging by 40ms.
  // If an update just committed on dbPrimary and deleted Redis, this read fetches
  // the pre-update stale record from dbReplica and caches it in Redis for 1 hour!
  const user = await dbReplica.user.findUnique({ where: { id: userId } });
  if (user) {
    await redis.set(`user:${userId}`, JSON.stringify(user), 'EX', 3600);
  }
  return user;
}

// ✅ Resilient pattern: Delayed Double Deletion & Primary-Pinned Re-reads
export async function invalidateWithDelayedDoubleDelete(userId: string, estimatedLagMs = 500) {
  const key = `user:${userId}`;

  // 1. Immediate invalidation post-commit
  await redis.del(key);

  // 2. Schedule a second deletion after the maximum replication lag window has passed
  setTimeout(async () => {
    try {
      await redis.del(key);
    } catch (err) {
      console.error(`Failed secondary delayed cache invalidation for ${key}:`, err);
    }
  }, estimatedLagMs);
}

// ✅ Production Write-Back Buffer with Persistent WAL / Stream Protection
import Redis from 'ioredis';

export class WriteBackBuffer {
  private redis: Redis;
  private flushBatchSize = 1000;

  constructor(redisClient: Redis) {
    this.redis = redisClient;
  }

  // 1. Write to Redis Stream (WAL) AND Memory Hash atomically via Pipeline
  async recordMetric(metricId: string, delta: number) {
    const pipeline = this.redis.pipeline();
    
    // In-memory counter for instant reads
    pipeline.hincrby('metrics:live', metricId, delta);
    
    // Append-only Redis Stream as durable Write-Ahead Log (WAL)
    pipeline.xadd('metrics:wal', '*', 'id', metricId, 'delta', delta.toString());
    
    await pipeline.exec();
  }

  // 2. Background consumer flushes batched writes to Postgres with transaction safety
  async flushDirtyMetricsToDatabase() {
    // Read batch of entries from Redis Stream
    const entries = await this.redis.xread('COUNT', this.flushBatchSize, 'STREAMS', 'metrics:wal', '0');
    if (!entries || entries.length === 0) return;

    const [streamName, streamRows] = entries[0];
    const updates = new Map<string, number>();
    const ackIds: string[] = [];

    // Write coalescing: collapse 10,000 stream events into aggregated row sums
    for (const [id, fields] of streamRows) {
      const metricId = fields[1];
      const delta = parseInt(fields[3], 10);
      updates.set(metricId, (updates.get(metricId) || 0) + delta);
      ackIds.push(id);
    }

    // Persist coalesced batch to PostgreSQL inside single transaction
    await db.$transaction(async (tx) => {
      for (const [metricId, totalDelta] of updates.entries()) {
        await tx.$executeRaw`
          INSERT INTO metrics (id, count) VALUES (${metricId}, ${totalDelta})
          ON CONFLICT (id) DO UPDATE SET count = metrics.count + ${totalDelta}
        `;
      }
    });

    // Acknowledge and trim stream only after DB commit succeeds
    await this.redis.xdel('metrics:wal', ...ackIds);
  }
}

// ❌ Fatal: Blocking KEYS command
const keys = await redis.keys(`tenant:${tenantId}:*`);
if (keys.length > 0) await redis.del(...keys);

// ✅ Resilient Pattern 1: Non-blocking SCAN iterator
export async function purgeNamespaceWithScan(pattern: string) {
  let cursor = '0';
  do {
    // Scan in chunks of 500 keys without blocking event loop
    const [nextCursor, keys] = await redis.scan(cursor, 'MATCH', pattern, 'COUNT', 500);
    cursor = nextCursor;
    if (keys.length > 0) {
      // UNLINK is asynchronous: reclaims memory in a background thread
      await redis.unlink(...keys);
    }
  } while (cursor !== '0');
}

// ✅ Resilient Pattern 2: Explicit Reverse-Index Sets (Zero Scanning Overhead)
export async function invalidateTenantExplicit(tenantId: string) {
  const indexKey = `tenant:${tenantId}:keys_index`;
  
  // Fetch all registered entity keys in O(1)
  const keys = await redis.smembers(indexKey);
  if (keys.length > 0) {
    const pipeline = redis.pipeline();
    pipeline.unlink(...keys);
    pipeline.unlink(indexKey);
    await pipeline.exec();
  }
}