feat(followers): implement CRDT-based follower count to eliminate los…

prisma/schema/follower.prisma

@@ -0,0 +1,31 @@
+// prisma/schema/follower.prisma
+
+enum FollowDirection {
+  FOLLOW
+  UNFOLLOW
+}
+
+model FollowerCounterShard {
+  id            String   @id @default(cuid())
+  creatorWallet String
+  nodeId        String
+  increments    BigInt   @default(0)
+  decrements    BigInt   @default(0)
+  updatedAt     DateTime @updatedAt
+
+  @@unique([creatorWallet, nodeId])
+  @@index([creatorWallet])
+  @@map("follower_counter_shards")
+}
+
+model FollowEvent {
+  id             String          @id @default(cuid())
+  followerWallet String
+  creatorWallet  String
+  direction      FollowDirection
+  createdAt      DateTime        @default(now())
+
+  @@unique([followerWallet, creatorWallet])
+  @@index([creatorWallet])
+  @@map("follow_events")
+}

src/config.schema.ts

@@ -46,6 +46,7 @@ export const envSchema = z
       DATABASE_URL: z
          .string()
          .min(1, 'DATABASE_URL is required in the environment variables'),
+      NODE_ID: z.string().default('node-local'),
 
       GMAIL_USER: z.string(),
       GMAIL_APP_PASSWORD: z.string(),

src/jobs/follower-shard-compaction.job.ts

@@ -0,0 +1,29 @@
+import { prisma } from '../utils/prisma.utils';
+import { compactShardsForCreator } from '../modules/followers/follower.service';
+import { logger } from '../utils/logger.utils';
+
+export async function runNightlyFollowerShardCompaction(): Promise<{
+   creatorsCompacted: number;
+}> {
+   logger.info('Starting nightly follower shard compaction job');
+
+   const creators = await prisma.followerCounterShard.findMany({
+      distinct: ['creatorWallet'],
+      select: { creatorWallet: true },
+   });
+
+   let count = 0;
+   for (const { creatorWallet } of creators) {
+      const compacted = await compactShardsForCreator(creatorWallet);
+      if (compacted) {
+         count++;
+      }
+   }
+
+   logger.info(
+      { creators_compacted: count, total_creators_evaluated: creators.length },
+      'Nightly follower shard compaction completed'
+   );
+
+   return { creatorsCompacted: count };
+}

src/modules/followers/follower.controllers.ts

@@ -0,0 +1,61 @@
+import { AsyncController } from '../../types/auth.types';
+import {
+   follow,
+   unfollow,
+   getFollowerCount,
+} from './follower.service';
+import { sendSuccess, sendValidationError } from '../../utils/api-response.utils';
+
+function getParamString(val: string | string[] | undefined): string | undefined {
+   if (!val) return undefined;
+   return Array.isArray(val) ? val[0] : val;
+}
+
+export const httpFollow: AsyncController = async (req, res, next) => {
+   try {
+      const creatorWallet = getParamString(req.params.creatorWallet);
+      const followerWallet = req.body?.followerWallet || req.jwtPayload?.walletAddress;
+
+      if (!creatorWallet || !followerWallet) {
+         sendValidationError(res, 'creatorWallet param and followerWallet body/token are required');
+         return;
+      }
+
+      const result = await follow(followerWallet, creatorWallet);
+      sendSuccess(res, result);
+   } catch (err) {
+      next(err);
+   }
+};
+
+export const httpUnfollow: AsyncController = async (req, res, next) => {
+   try {
+      const creatorWallet = getParamString(req.params.creatorWallet);
+      const followerWallet = req.body?.followerWallet || req.jwtPayload?.walletAddress;
+
+      if (!creatorWallet || !followerWallet) {
+         sendValidationError(res, 'creatorWallet param and followerWallet body/token are required');
+         return;
+      }
+
+      const result = await unfollow(followerWallet, creatorWallet);
+      sendSuccess(res, result);
+   } catch (err) {
+      next(err);
+   }
+};
+
+export const httpGetFollowerCount: AsyncController = async (req, res, next) => {
+   try {
+      const creatorWallet = getParamString(req.params.creatorWallet);
+      if (!creatorWallet) {
+         sendValidationError(res, 'creatorWallet parameter is required');
+         return;
+      }
+
+      const count = await getFollowerCount(creatorWallet);
+      sendSuccess(res, { creatorWallet, count });
+   } catch (err) {
+      next(err);
+   }
+};

src/modules/followers/follower.routes.ts

@@ -0,0 +1,14 @@
+import { Router } from 'express';
+import {
+   httpFollow,
+   httpUnfollow,
+   httpGetFollowerCount,
+} from './follower.controllers';
+
+const followerRouter = Router();
+
+followerRouter.post('/:creatorWallet/follow', httpFollow);
+followerRouter.post('/:creatorWallet/unfollow', httpUnfollow);
+followerRouter.get('/:creatorWallet/count', httpGetFollowerCount);
+
+export default followerRouter;

src/modules/followers/follower.service.test.ts

@@ -0,0 +1,228 @@
+import {
+   follow,
+   unfollow,
+   getFollowerCount,
+   compactShardsForCreator,
+} from './follower.service';
+import { prisma } from '../../utils/prisma.utils';
+
+jest.mock('../../utils/logger.utils', () => ({
+   logger: {
+      debug: jest.fn(),
+      info: jest.fn(),
+      warn: jest.fn(),
+      error: jest.fn(),
+   },
+}));
+
+// In-memory mock for Prisma to test CRDT logic isolated from real DB
+jest.mock('../../utils/prisma.utils', () => {
+   const followEvents = new Map<string, { followerWallet: string; creatorWallet: string; direction: string }>();
+   const shards = new Map<string, { id: string; creatorWallet: string; nodeId: string; increments: bigint; decrements: bigint; updatedAt: Date }>();
+
+   return {
+      prisma: {
+         followEvent: {
+            findUnique: jest.fn().mockImplementation(async ({ where }) => {
+               const key = `${where.followerWallet_creatorWallet.followerWallet}:${where.followerWallet_creatorWallet.creatorWallet}`;
+               return followEvents.get(key) || null;
+            }),
+            upsert: jest.fn().mockImplementation(async ({ where, update, create }) => {
+               const key = `${where.followerWallet_creatorWallet.followerWallet}:${where.followerWallet_creatorWallet.creatorWallet}`;
+               const existing = followEvents.get(key);
+               const direction = existing ? update.direction : create.direction;
+               const record = {
+                  followerWallet: where.followerWallet_creatorWallet.followerWallet,
+                  creatorWallet: where.followerWallet_creatorWallet.creatorWallet,
+                  direction,
+               };
+               followEvents.set(key, record);
+               return record;
+            }),
+            update: jest.fn().mockImplementation(async ({ where, data }) => {
+               const key = `${where.followerWallet_creatorWallet.followerWallet}:${where.followerWallet_creatorWallet.creatorWallet}`;
+               const existing = followEvents.get(key);
+               if (existing) {
+                  existing.direction = data.direction;
+               }
+               return existing;
+            }),
+         },
+         followerCounterShard: {
+            findUnique: jest.fn().mockImplementation(async ({ where }) => {
+               const key = `${where.creatorWallet_nodeId.creatorWallet}:${where.creatorWallet_nodeId.nodeId}`;
+               return shards.get(key) || null;
+            }),
+            findMany: jest.fn().mockImplementation(async ({ where }) => {
+               const list = Array.from(shards.values()).filter(
+                  s => s.creatorWallet === where.creatorWallet
+               );
+               if (where.updatedAt?.gte) {
+                  return list.filter(s => s.updatedAt >= where.updatedAt.gte);
+               }
+               return list;
+            }),
+            findFirst: jest.fn().mockImplementation(async ({ where }) => {
+               const list = Array.from(shards.values()).filter(
+                  s => s.creatorWallet === where.creatorWallet
+               );
+               if (where.updatedAt?.gte) {
+                  return list.find(s => s.updatedAt >= where.updatedAt.gte) || null;
+               }
+               return list[0] || null;
+            }),
+            create: jest.fn().mockImplementation(async ({ data }) => {
+               const key = `${data.creatorWallet}:${data.nodeId}`;
+               let record = shards.get(key);
+               if (record) {
+                  record.increments += BigInt(data.increments ?? 0);
+                  record.decrements += BigInt(data.decrements ?? 0);
+               } else {
+                  record = {
+                     id: key,
+                     creatorWallet: data.creatorWallet,
+                     nodeId: data.nodeId,
+                     increments: BigInt(data.increments ?? 0),
+                     decrements: BigInt(data.decrements ?? 0),
+                     updatedAt: new Date(),
+                  };
+                  shards.set(key, record);
+               }
+               return record;
+            }),
+            update: jest.fn().mockImplementation(async ({ where, data }) => {
+               const key = `${where.creatorWallet_nodeId.creatorWallet}:${where.creatorWallet_nodeId.nodeId}`;
+               let record = shards.get(key);
+               if (!record) {
+                  record = {
+                     id: key,
+                     creatorWallet: where.creatorWallet_nodeId.creatorWallet,
+                     nodeId: where.creatorWallet_nodeId.nodeId,
+                     increments: 0n,
+                     decrements: 0n,
+                     updatedAt: new Date(),
+                  };
+                  shards.set(key, record);
+               }
+               if (data.increments?.increment) {
+                  record.increments += BigInt(data.increments.increment);
+               }
+               if (data.decrements?.increment) {
+                  record.decrements += BigInt(data.decrements.increment);
+               }
+               record.updatedAt = new Date();
+               return record;
+            }),
+            deleteMany: jest.fn().mockImplementation(async ({ where }) => {
+               for (const [key, shard] of shards.entries()) {
+                  if (shard.creatorWallet === where.creatorWallet) {
+                     shards.delete(key);
+                  }
+               }
+               return { count: 1 };
+            }),
+         },
+         $executeRaw: jest.fn().mockRejectedValue(new Error('raw query fallback')),
+         $transaction: jest.fn().mockImplementation(async (actions) => Promise.all(actions)),
+         _reset: () => {
+            followEvents.clear();
+            shards.clear();
+         },
+      },
+   };
+});
+
+jest.mock('../../utils/redis.utils', () => {
+   const redisStore = new Map<string, string>();
+   return {
+      getRedis: () => ({
+         get: jest.fn().mockImplementation(async (key: string) => redisStore.get(key) ?? null),
+         set: jest.fn().mockImplementation(async (key: string, val: string) => {
+            redisStore.set(key, val);
+            return 'OK';
+         }),
+         del: jest.fn().mockImplementation(async (key: string) => {
+            redisStore.delete(key);
+            return 1;
+         }),
+      }),
+   };
+});
+
+describe('CRDT Follower Counter (#757)', () => {
+   const creatorWallet = 'GCREATOR_CRDT_TEST';
+
+   beforeEach(() => {
+      (prisma as any)._reset();
+      jest.clearAllMocks();
+   });
+
+   it('100 concurrent follows from different wallets produces count of exactly 100', async () => {
+      const promises = Array.from({ length: 100 }, (_, i) =>
+         follow(`follower-wallet-${i}`, creatorWallet, 'node-1')
+      );
+
+      await Promise.all(promises);
+
+      const count = await getFollowerCount(creatorWallet);
+      expect(count).toBe(100);
+   });
+
+   it('50 concurrent follows and 30 concurrent unfollows produces net count of 20', async () => {
+      // First 50 follow
+      const followPromises = Array.from({ length: 50 }, (_, i) =>
+         follow(`follower-${i}`, creatorWallet, 'node-1')
+      );
+      await Promise.all(followPromises);
+
+      // 30 of them unfollow
+      const unfollowPromises = Array.from({ length: 30 }, (_, i) =>
+         unfollow(`follower-${i}`, creatorWallet, 'node-1')
+      );
+      await Promise.all(unfollowPromises);
+
+      const count = await getFollowerCount(creatorWallet);
+      expect(count).toBe(20);
+   });
+
+   it('double-follow from the same wallet is idempotent and increments count only once', async () => {
+      const first = await follow('wallet-alice', creatorWallet, 'node-1');
+      const second = await follow('wallet-alice', creatorWallet, 'node-1');
+
+      expect(first.followed).toBe(true);
+      expect(second.followed).toBe(false);
+
+      const count = await getFollowerCount(creatorWallet);
+      expect(count).toBe(1);
+   });
+
+   it('sums counts across multiple node shards correctly', async () => {
+      await follow('wallet-1', creatorWallet, 'node-alpha');
+      await follow('wallet-2', creatorWallet, 'node-alpha');
+      await follow('wallet-3', creatorWallet, 'node-beta');
+
+      const count = await getFollowerCount(creatorWallet);
+      expect(count).toBe(3);
+   });
+
+   it('nightly compaction merges shards atomically while maintaining identical resolved count', async () => {
+      await follow('wallet-1', creatorWallet, 'node-1');
+      await follow('wallet-2', creatorWallet, 'node-2');
+      await follow('wallet-3', creatorWallet, 'node-3');
+
+      const countBefore = await getFollowerCount(creatorWallet);
+      expect(countBefore).toBe(3);
+
+      // Mock Date.now to simulate 6 minutes passing so compaction is allowed
+      const realNow = Date.now;
+      jest.spyOn(Date, 'now').mockReturnValue(realNow() + 6 * 60 * 1000);
+
+      const compacted = await compactShardsForCreator(creatorWallet);
+      expect(compacted).toBe(true);
+
+      const countAfter = await getFollowerCount(creatorWallet);
+      expect(countAfter).toBe(3);
+
+      jest.restoreAllMocks();
+   });
+});