Feat/projection versioned

.env.example

@@ -44,7 +44,8 @@ REDIS_PORT=6379
 REDIS_PASSWORD=
 REDIS_DB=0
 REDIS_TLS=false  # Set to 'true' for production with TLS
-CACHE_CLAIMS_TTL=3600  # Claim cache TTL in seconds
+CACHE_CLAIMS_TTL=300  # Claim cache TTL in seconds (reduced to 5 minutes for safety)
+CACHE_VERSION=v1       # Cache version for versioned keys - increment when cache schema changes
 
 # ==============================================
 # Blockchain Configuration
@@ -139,4 +140,4 @@ SMTP_FROM=noreply@truthbounty.com
 # Notification queue configuration
 NOTIFICATION_QUEUE_DELAY=0
 NOTIFICATION_MAX_RETRIES=5
-NOTIFICATION_RETRY_DELAY=2000
+NOTIFICATION_RETRY_DELAY=2000

src/blockchain/blockchain-indexer.service.ts

@@ -5,6 +5,7 @@ import { ProcessedEvent } from './entities/processed-event.entity';
 import { TokenBalance } from './entities/token-balance.entity';
 import { IndexerCheckpoint } from './entities/indexer-checkpoint.entity';
 import { BlockchainEvent, TransferEventData } from './interfaces/blockchain-event.interface';
+import { ClaimsCache } from '../cache/claims.cache';
 import { SequentialQueue } from './utils/sequential-queue';
 
 @Injectable()
@@ -27,6 +28,7 @@ export class BlockchainIndexerService {
     @InjectRepository(IndexerCheckpoint)
     private checkpointRepo: Repository<IndexerCheckpoint>,
     private dataSource: DataSource,
+    private claimsCache: ClaimsCache,
   ) {}
 
   /**
@@ -80,6 +82,11 @@ export class BlockchainIndexerService {
 
       await queryRunner.commitTransaction();
       this.logger.log(`Processed event: ${eventType} at block ${blockNumber}`);
+
+      // Invalidate cache after successfully committing a projection change
+      // In a production system, you'd track which claim IDs are affected by this event
+      // For safety, we invalidate all claims cache to ensure no stale data is served
+      await this.claimsCache.invalidateForProjectionUpdate();
     } catch (error) {
       await queryRunner.rollbackTransaction();
       this.logger.error(`Failed to process event: ${error.message}`, error.stack);
@@ -141,6 +148,10 @@ export class BlockchainIndexerService {
       this.logger.log(
         `Rolled back ${orphaned.length} event(s); checkpoint rewound to block ${rewoundTo}`,
       );
+
+      // Invalidate all cache after rolling back events during a reorg
+      // This is critical to ensure we never serve stale data based on orphaned chain state
+      await this.claimsCache.invalidateAllForReorg();
     } catch (error) {
       await queryRunner.rollbackTransaction();
       this.logger.error(`Failed to roll back from block ${startBlock}: ${error.message}`, error.stack);
@@ -179,4 +190,4 @@ export class BlockchainIndexerService {
     const checkpoint = await this.checkpointRepo.findOne({ where: { id: 1 } });
     return checkpoint ? checkpoint.lastBlock : null;
   }
-}
+}

src/blockchain/reorg-detector.service.ts

@@ -1,6 +1,7 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { BlockchainStateService } from './state.service';
+import { ClaimsCache } from '../cache/claims.cache';
 import { BlockInfo, ReorgEvent, PendingEvent } from './types';
 
 /**
@@ -19,6 +20,7 @@ export class ReorgDetectorService {
   constructor(
     private stateService: BlockchainStateService,
     private configService: ConfigService,
+    private claimsCache: ClaimsCache,
   ) {
     this.confirmationDepth = this.configService.get<number>(
       'BLOCKCHAIN_CONFIRMATION_DEPTH',
@@ -82,6 +84,10 @@ export class ReorgDetectorService {
         `Orphaned events: ${orphanedEventIds.length}`,
     );
 
+    // Invalidate all cached claims since blockchain state has changed
+    // This maintains the invariant that smart contracts are always the source of truth
+    await this.claimsCache.invalidateBlockRange(affectedBlockStart, affectedBlockEnd);
+
     return reorg;
   }
 

src/cache/claims.cache.spec.ts

@@ -0,0 +1,155 @@
+import { ConfigService } from '@nestjs/config';
+import { ClaimsCache } from './claims.cache';
+import { RedisService } from '../redis/redis.service';
+
+describe('ClaimsCache', () => {
+  let cache: ClaimsCache;
+  let redisService: jest.Mocked<
+    Pick<RedisService, 'get' | 'set' | 'del' | 'getClient'>
+  >;
+  let mockSadd: jest.Mock;
+  let mockSmembers: jest.Mock;
+  let mockDel: jest.Mock;
+  let mockSrem: jest.Mock;
+  let mockExpire: jest.Mock;
+
+  beforeEach(() => {
+    mockSadd = jest.fn().mockResolvedValue(1);
+    mockSmembers = jest.fn().mockResolvedValue(['v1:claim:123', 'v1:claims:latest']);
+    mockDel = jest.fn().mockResolvedValue(2);
+    mockSrem = jest.fn().mockResolvedValue(1);
+    mockExpire = jest.fn().mockResolvedValue(true);
+
+    const mockRedisClient = {
+      sadd: mockSadd,
+      smembers: mockSmembers,
+      del: mockDel,
+      srem: mockSrem,
+      expire: mockExpire,
+    };
+
+    redisService = {
+      get: jest.fn(),
+      set: jest.fn(),
+      del: jest.fn(),
+      getClient: jest.fn().mockReturnValue(mockRedisClient),
+    };
+
+    const configService = {
+      get: jest.fn((key: string) => {
+        if (key === 'CACHE_CLAIMS_TTL') return 300;
+        if (key === 'CACHE_VERSION') return 'v1';
+        return null;
+      }),
+    } as unknown as ConfigService;
+
+    cache = new ClaimsCache(
+      redisService as unknown as RedisService,
+      configService,
+    );
+  });
+
+  describe('versioned keys', () => {
+    it('uses versioned cache keys', async () => {
+      const claim = { id: '123', title: 'Test Claim' };
+      redisService.get.mockResolvedValue(JSON.stringify(claim));
+
+      await cache.setClaim('123', claim);
+      expect(redisService.set).toHaveBeenCalledWith(
+        'v1:claim:123',
+        JSON.stringify(claim),
+        300,
+      );
+
+      const retrieved = await cache.getClaim('123');
+      expect(retrieved).toEqual(claim);
+      expect(redisService.get).toHaveBeenCalledWith('v1:claim:123');
+    });
+
+    it('generates correct versioned keys for user claims', async () => {
+      const wallet = '0x1234567890123456789012345678901234567890';
+      const claims = [{ id: '123', title: 'Test' }];
+
+      await cache.setUserClaims(wallet, claims);
+      expect(redisService.set).toHaveBeenCalledWith(
+        'v1:claims:user:0x1234567890123456789012345678901234567890',
+        JSON.stringify(claims),
+        300,
+      );
+    });
+  });
+
+  describe('key tracking', () => {
+    it('tracks cache keys in the index set', async () => {
+      const claim = { id: '123', title: 'Test' };
+      await cache.setClaim('123', claim);
+
+      expect(mockSadd).toHaveBeenCalledWith(
+        'claims:cache:keys',
+        'v1:claim:123'
+      );
+      expect(mockExpire).toHaveBeenCalledWith('claims:cache:keys', 600); // 2*TTL
+    });
+  });
+
+  describe('cache invalidation', () => {
+    it('invalidates a specific claim and related lists', async () => {
+      await cache.invalidateClaim('123', '0x1234567890');
+
+      expect(redisService.del).toHaveBeenCalledTimes(3);
+      expect(redisService.del).toHaveBeenCalledWith('v1:claim:123');
+      expect(redisService.del).toHaveBeenCalledWith('v1:claims:latest');
+      expect(redisService.del).toHaveBeenCalledWith('v1:claims:user:0x1234567890123456789012345678901234567890');
+      expect(mockSrem).toHaveBeenCalled();
+    });
+  });
+
+  describe('reorg handling', () => {
+    it('invalidates ALL cache during a chain reorg', async () => {
+      await cache.invalidateAllForReorg();
+
+      expect(mockSmembers).toHaveBeenCalledWith('claims:cache:keys');
+      expect(mockDel).toHaveBeenCalledWith('v1:claim:123', 'v1:claims:latest');
+    });
+
+    it('invalidates cache for a block range', async () => {
+      await cache.invalidateBlockRange(1000, 1050);
+
+      expect(mockSmembers).toHaveBeenCalled();
+      expect(mockDel).toHaveBeenCalled();
+    });
+  });
+
+  describe('projection updates', () => {
+    it('invalidates specific affected claims', async () => {
+      const invalidateSpy = jest.spyOn(cache, 'invalidateClaim');
+      await cache.invalidateForProjectionUpdate(['123', '456']);
+
+      expect(invalidateSpy).toHaveBeenCalledTimes(2);
+    });
+
+    it('invalidates all if no specific claims provided', async () => {
+      const invalidateAllSpy = jest.spyOn(cache, 'invalidateAllForReorg');
+      await cache.invalidateForProjectionUpdate();
+
+      expect(invalidateAllSpy).toHaveBeenCalled();
+    });
+  });
+
+  describe('graceful degradation', () => {
+    it('returns null when Redis is unavailable', async () => {
+      redisService.getClient.mockReturnValue(null);
+      redisService.get.mockResolvedValue(null);
+
+      const result = await cache.getClaim('123');
+      expect(result).toBeNull();
+    });
+
+    it('handles JSON parsing errors gracefully', async () => {
+      redisService.get.mockResolvedValue('invalid json');
+
+      const result = await cache.getClaim('123');
+      expect(result).toBeNull();
+    });
+  });
+});