fix flaw

src/lib/db.ts

@@ -9,6 +9,8 @@ const BASE_DELAY_MS = 50;
 const IS_VERCEL = process.env.VERCEL || false;
 const HEALTH_CHECK_INTERVAL = 60 * 1000; // Check primary health every minute
 const REPLICA_PROMOTION_THRESHOLD = 5; // After 5 failed primary health checks, promote a replica to primary
+const SYNC_BUFFER_SIZE = 100; // Maximum number of operations to buffer before syncing
+const SYNC_INTERVAL = 10 * 1000; // Sync replicas every 10 seconds
 
 // Redis URLs from environment variables
 const PRIMARY_REDIS_URL =
@@ -31,6 +33,21 @@ let primaryFailedChecks = 0;
 let replicaClients: Map<string, Redis> = new Map();
 let activeReplicaUrl: string | null = null;
 
+// Track which replica was used when primary was down
+let lastUsedReplica: string | null = null;
+
+// Operation buffer for sync
+interface RedisOperation {
+  key: string;
+  value: string;
+  ttl?: number;
+  type: 'set' | 'del';
+  timestamp: number;
+}
+
+// Buffer to store operations that need to be synced
+const operationBuffer: RedisOperation[] = [];
+
 // Redis instance naming for logging
 const redisInstanceNames = new Map<string, string>();
 
@@ -87,7 +104,15 @@ async function createRedisClient(url: string): Promise<Redis> {
 
   // Setup event listeners
   client
-    .on("ready", () => console.log(`${redisName} ready`))
+    .on("ready", () => {
+      console.log(`${redisName} ready`);
+      
+      // If primary recovers and we had used a replica, initiate sync from replica to primary
+      if (url === PRIMARY_REDIS_URL && lastUsedReplica) {
+        syncFromReplicaToPrimary(lastUsedReplica).catch(err => 
+          console.error(`Failed to sync from replica to primary: ${err}`));
+      }
+    })
     .on("error", (err) => {
       console.error(`${redisName} error:`, err);
       if (url === PRIMARY_REDIS_URL) {
@@ -153,6 +178,7 @@ export async function getRedisClient(): Promise<Redis> {
     const activeReplica = replicaClients.get(activeReplicaUrl)!;
     if (activeReplica.status === "ready" && await checkConnection(activeReplica, getRedisName(activeReplicaUrl))) {
       console.log(`Using active replica: ${getRedisName(activeReplicaUrl)}`);
+      lastUsedReplica = activeReplicaUrl; // Track that we used this replica
       return activeReplica;
     }
   }
@@ -169,6 +195,7 @@ export async function getRedisClient(): Promise<Redis> {
       if (await checkConnection(client, getRedisName(url))) {
         console.log(`Switching to replica: ${getRedisName(url)}`);
         activeReplicaUrl = url;
+        lastUsedReplica = url; // Track that we used this replica
         
         // If primary has been down for a while, promote this replica to be the new primary
         if (primaryFailedChecks >= REPLICA_PROMOTION_THRESHOLD) {
@@ -191,12 +218,164 @@ export async function getRedisClient(): Promise<Redis> {
   return primaryClient;
 }
 
+/**
+ * Add an operation to the buffer for replication
+ */
+function bufferOperation(operation: RedisOperation): void {
+  operationBuffer.push(operation);
+  
+  // If buffer exceeds size limit, trigger a sync
+  if (operationBuffer.length >= SYNC_BUFFER_SIZE) {
+    syncReplicas().catch(err => console.error("Failed to sync replicas:", err));
+  }
+}
+
+/**
+ * Synchronize data from a replica to the primary
+ * Used when primary recovers after being down
+ */
+async function syncFromReplicaToPrimary(replicaUrl: string): Promise<void> {
+  if (!primaryClient || primaryClient.status !== "ready" || !replicaClients.has(replicaUrl)) {
+    console.log("Cannot sync: primary or replica not available");
+    return;
+  }
+  
+  const replicaClient = replicaClients.get(replicaUrl)!;
+  console.log(`Starting sync from ${getRedisName(replicaUrl)} to Primary...`);
+  
+  try {
+    // Get all feed keys using scan to prevent blocking
+    const feedKeys: string[] = [];
+    let cursor = '0';
+    
+    do {
+      const [nextCursor, keys] = await replicaClient.scan(
+        cursor, 
+        'MATCH', 
+        '[a-f0-9]*', // Match feed IDs (hex format)
+        'COUNT',
+        '100'
+      );
+      
+      cursor = nextCursor;
+      feedKeys.push(...keys);
+    } while (cursor !== '0');
+    
+    // Get chapter cache keys
+    const chapterKeys: string[] = [];
+    cursor = '0';
+    
+    do {
+      const [nextCursor, keys] = await replicaClient.scan(
+        cursor,
+        'MATCH',
+        'chapters:*',
+        'COUNT',
+        '100'
+      );
+      
+      cursor = nextCursor;
+      chapterKeys.push(...keys);
+    } while (cursor !== '0');
+    
+    console.log(`Found ${feedKeys.length} feed keys and ${chapterKeys.length} chapter keys to sync`);
+    
+    // Sync feed mappings
+    if (feedKeys.length > 0) {
+      const feedPipeline = primaryClient.pipeline();
+      
+      for (const key of feedKeys) {
+        const value = await replicaClient.get(key);
+        const ttl = await replicaClient.ttl(key);
+        
+        if (value && ttl > 0) {
+          feedPipeline.set(key, value, 'EX', ttl);
+        }
+      }
+      
+      await feedPipeline.exec();
+      console.log(`Synced ${feedKeys.length} feed mappings from replica to primary`);
+    }
+    
+    // Sync chapter cache
+    if (chapterKeys.length > 0) {
+      const chapterPipeline = primaryClient.pipeline();
+      
+      for (const key of chapterKeys) {
+        const value = await replicaClient.get(key);
+        const ttl = await replicaClient.ttl(key);
+        
+        if (value && ttl > 0) {
+          chapterPipeline.set(key, value, 'EX', ttl);
+        }
+      }
+      
+      await chapterPipeline.exec();
+      console.log(`Synced ${chapterKeys.length} chapter caches from replica to primary`);
+    }
+    
+    lastUsedReplica = null; // Reset after successful sync
+    console.log("Replica to primary sync completed successfully");
+  } catch (error) {
+    console.error("Error during replica to primary sync:", error);
+  }
+}
+
+/**
+ * Synchronize operations from buffer to all replicas
+ */
+async function syncReplicas(): Promise<void> {
+  if (operationBuffer.length === 0) return;
+  
+  // Get operations to sync and clear buffer
+  const operations = [...operationBuffer];
+  operationBuffer.length = 0;
+  
+  // Skip if no operations or no replicas
+  if (operations.length === 0 || replicaClients.size === 0) return;
+  
+  console.log(`Syncing ${operations.length} operations to ${replicaClients.size} replicas...`);
+  
+  // Process each replica
+  const syncPromises = Array.from(replicaClients.entries()).map(async ([url, client]) => {
+    try {
+      if (client.status !== "ready" || !(await checkConnection(client, getRedisName(url)))) {
+        return;
+      }
+      
+      const pipeline = client.pipeline();
+      
+      // Add all operations to pipeline
+      for (const op of operations) {
+        if (op.type === 'set') {
+          if (op.ttl) {
+            pipeline.set(op.key, op.value, 'EX', op.ttl);
+          } else {
+            pipeline.set(op.key, op.value);
+          }
+        } else if (op.type === 'del') {
+          pipeline.del(op.key);
+        }
+      }
+      
+      // Execute pipeline
+      await pipeline.exec();
+      console.log(`Successfully synced ${operations.length} operations to ${getRedisName(url)}`);
+    } catch (error) {
+      console.error(`Error syncing to ${getRedisName(url)}:`, error);
+    }
+  });
+  
+  await Promise.allSettled(syncPromises);
+}
+
 /**
  * Schedule periodic health checks to detect when primary recovers
  */
 function scheduleHealthChecks() {
   if (typeof setInterval === 'undefined') return; // Skip in environments without setInterval
 
+  // Check primary health
   setInterval(async () => {
     if (!primaryHealthy && primaryClient) {
       try {
@@ -205,6 +384,12 @@ function scheduleHealthChecks() {
           console.log("Primary Redis recovered");
           primaryHealthy = true;
           primaryFailedChecks = 0;
+          
+          // If we used a replica when primary was down, sync data back
+          if (lastUsedReplica) {
+            syncFromReplicaToPrimary(lastUsedReplica).catch(err => 
+              console.error(`Failed to sync from replica to primary: ${err}`));
+          }
         } else {
           primaryFailedChecks++;
           console.log(`Primary still unhealthy, failed checks: ${primaryFailedChecks}`);
@@ -215,15 +400,20 @@ function scheduleHealthChecks() {
       }
     }
   }, HEALTH_CHECK_INTERVAL);
+  
+  // Schedule regular replica sync
+  setInterval(async () => {
+    await syncReplicas().catch(err => console.error("Failed to sync replicas:", err));
+  }, SYNC_INTERVAL);
 }
 
-// Start health checks if not in a build context
+// Start health checks and sync if not in a build context
 if (!import.meta.env.SSR) {
   scheduleHealthChecks();
 }
 
 /**
- * Store a feed mapping in Redis with automatic failover
+ * Store a feed mapping in Redis with automatic failover and replication
  */
 export async function storeFeedMapping(
   feedId: string,
@@ -236,10 +426,28 @@ export async function storeFeedMapping(
     created_at: new Date().toISOString() 
   };
   
+  const stringValue = JSON.stringify(mapping);
+  
   try {
+    // First write to primary/active client
     const redis = await getRedisClient(); // This already handles failover
-    await redis.set(feedId, JSON.stringify(mapping), "EX", DEFAULT_CACHE_TTL);
+    await redis.set(feedId, stringValue, "EX", DEFAULT_CACHE_TTL);
     console.log(`Successfully stored mapping for feed ${feedId}`);
+    
+    // Buffer operation for replication to other instances
+    bufferOperation({
+      key: feedId,
+      value: stringValue,
+      ttl: DEFAULT_CACHE_TTL,
+      type: 'set',
+      timestamp: Date.now()
+    });
+    
+    // If we're writing to a replica because primary is down, force immediate sync
+    // to other replicas to ensure consistent data
+    if (!primaryHealthy && lastUsedReplica) {
+      await syncReplicas().catch(err => console.error("Failed to sync replicas after write:", err));
+    }
   } catch (error) {
     console.error(`Error storing feed mapping for ${feedId}:`, error);
     throw error;
@@ -261,11 +469,25 @@ export async function getFeedMapping(
     // Reset expiration on access
     await redis.expire(feedId, DEFAULT_CACHE_TTL);
     
-    const data = JSON.parse(mapping);
+    // Buffer this TTL-reset operation for replication
-    return { 
+    bufferOperation({
-      slugs: data.slugs, 
+      key: feedId,
-      lang: data.lang 
+      value: mapping,
-    };
+      ttl: DEFAULT_CACHE_TTL,
+      type: 'set',
+      timestamp: Date.now()
+    });
+    
+    try {
+      const data = JSON.parse(mapping);
+      return { 
+        slugs: data.slugs, 
+        lang: data.lang 
+      };
+    } catch (e) {
+      console.error(`Error parsing feed mapping for ${feedId}:`, e);
+      return null;
+    }
   } catch (error) {
     console.error(`Error retrieving feed mapping for ${feedId}:`, error);
     return null;