Merge pull request #67 from Anichris-koded/feat/scheduled-membership-reconciliation

feat: add scheduled membership state reconciliation (#53)

Author: Lakes41

.env.example

@@ -47,4 +47,4 @@ DATABASE_URL="postgresql://postgres:postgres@localhost:5432/guildpass"
 # ============================================================================
 
 # How often the membership reconciliation worker runs, in milliseconds (default: 60000)
-# RECONCILIATION_INTERVAL_MS=60000
+# RECONCILIATION_INTERVAL_MS=60000

apps/access-api/jest.config.js

@@ -1,6 +1,6 @@
 module.exports = {
   preset: 'ts-jest',
   testEnvironment: 'node',
-  roots: ['<rootDir>/test'],
+  roots: ['<rootDir>/test', '<rootDir>/src'],
   testMatch: ['**/*.test.ts'],
 };

apps/access-api/prisma/migrations/20260615_init/migration.sql

@@ -0,0 +1,67 @@
+-- Initial schema: base tables as they existed before incremental migrations.
+
+CREATE TYPE "MembershipState" AS ENUM ('invited', 'active', 'expired', 'suspended');
+CREATE TYPE "Role" AS ENUM ('admin', 'member', 'contributor');
+CREATE TYPE "RoleSource" AS ENUM ('manual', 'auto');
+
+CREATE TABLE "Community" (
+  "id"        TEXT PRIMARY KEY,
+  "name"      TEXT NOT NULL,
+  "createdAt" TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE "Wallet" (
+  "id"        TEXT PRIMARY KEY,
+  "address"   TEXT NOT NULL UNIQUE,
+  "createdAt" TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE "Profile" (
+  "id"          TEXT PRIMARY KEY,
+  "displayName" TEXT NOT NULL,
+  "bio"         TEXT
+);
+
+CREATE TABLE "Member" (
+  "id"          TEXT PRIMARY KEY,
+  "communityId" TEXT NOT NULL REFERENCES "Community"("id"),
+  "walletId"    TEXT NOT NULL REFERENCES "Wallet"("id"),
+  "profileId"   TEXT REFERENCES "Profile"("id"),
+  UNIQUE ("communityId", "walletId")
+);
+
+CREATE TABLE "Membership" (
+  "id"        TEXT PRIMARY KEY,
+  "memberId"  TEXT NOT NULL UNIQUE REFERENCES "Member"("id"),
+  "state"     "MembershipState" NOT NULL,
+  "expiresAt" TIMESTAMPTZ,
+  "renewedAt" TIMESTAMPTZ,
+  "createdAt" TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE "RoleAssignment" (
+  "id"        TEXT PRIMARY KEY,
+  "memberId"  TEXT NOT NULL REFERENCES "Member"("id"),
+  "role"      "Role" NOT NULL,
+  "source"    "RoleSource" NOT NULL,
+  "active"    BOOLEAN NOT NULL DEFAULT true,
+  "createdAt" TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE "Badge" (
+  "id"       TEXT PRIMARY KEY,
+  "memberId" TEXT NOT NULL REFERENCES "Member"("id"),
+  "label"    TEXT NOT NULL,
+  "issuedAt" TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX "Badge_memberId_idx" ON "Badge" ("memberId");
+
+-- AccessPolicy with original "rule" column (ruleType/params added in next migration)
+CREATE TABLE "AccessPolicy" (
+  "id"          TEXT PRIMARY KEY,
+  "communityId" TEXT NOT NULL REFERENCES "Community"("id"),
+  "resource"    TEXT NOT NULL,
+  "rule"        TEXT NOT NULL DEFAULT 'MEMBERS_ONLY',
+  UNIQUE ("communityId", "resource")
+);

apps/access-api/prisma/migrations/20260618_membership_reconciliation_index/migration.sql

@@ -0,0 +1,3 @@
+-- Index to support efficient reconciliation queries:
+-- WHERE state IN ('active', 'suspended') AND expiresAt < now()
+CREATE INDEX "Membership_state_expiresAt_idx" ON "Membership" ("state", "expiresAt");

apps/access-api/src/routes.ts

@@ -40,7 +40,7 @@ export async function registerRoutes(app: FastifyInstance): Promise<void> {
         error: 'Missing required fields: wallet, communityId, resource',
       });
     }
-    const result = await memberService.checkAccess(body);
+    const result = await memberService.checkAccess(body as import('@guildpass/shared-types').AccessCheckInput);
     return result;
   });
 

apps/access-api/src/services/memberService.ts

@@ -9,6 +9,22 @@ import { logEvent } from "./auditService";
 
 const prisma = new PrismaClient();
 
+/**
+ * Returns the effective membership state at read time.
+ * If the stored state is active/suspended but expiresAt is in the past,
+ * we treat it as expired. This is the first line of defence; the
+ * reconciliation worker corrects the persisted state asynchronously.
+ */
+function getNormalizedMembershipState(
+  state: string,
+  expiresAt: Date | null | undefined,
+): string {
+  if (expiresAt && expiresAt <= new Date() && state !== "expired") {
+    return "expired";
+  }
+  return state;
+}
+
 export function getMemberService(prismaOverride?: PrismaClient) {
   const db = prismaOverride ?? prisma;
   return {
@@ -23,7 +39,10 @@ export function getMemberService(prismaOverride?: PrismaClient) {
       });
       const communities = members.map((m) => ({
         communityId: m.communityId,
-        state: m.membership?.state || "invited",
+        state: getNormalizedMembershipState(
+          m.membership?.state || "invited",
+          m.membership?.expiresAt,
+        ),
         expiresAt: m.membership?.expiresAt?.toISOString() ?? null,
       }));
       return { wallet, communities };
@@ -47,7 +66,10 @@ export function getMemberService(prismaOverride?: PrismaClient) {
           bio: m.profile?.bio ?? "",
         },
         membership: {
-          state: m.membership?.state ?? "invited",
+          state: getNormalizedMembershipState(
+            m.membership?.state ?? "invited",
+            m.membership?.expiresAt,
+          ),
           expiresAt: m.membership?.expiresAt?.toISOString() ?? null,
         },
         roles: m.roles.filter((r) => r.active).map((r) => r.role),
@@ -88,13 +110,17 @@ export function getMemberService(prismaOverride?: PrismaClient) {
         where: { communityId: input.communityId, resource: input.resource },
       });
       const ruleType = policy ? policy.ruleType : "MEMBERS_ONLY";
+      const effectiveState = getNormalizedMembershipState(
+        member.membership?.state ?? "invited",
+        member.membership?.expiresAt,
+      );
       const ctx: RoleContext = {
         assignments: member.roles.map((r) => ({
           role: r.role as any,
           source: r.source as any,
           active: r.active,
         })),
-        membershipState: (member.membership?.state as any) ?? "invited",
+        membershipState: effectiveState as any,
       };
       const decision = evaluate(
         {
@@ -126,7 +152,10 @@ export function getMemberService(prismaOverride?: PrismaClient) {
           return {
             wallet: m.wallet.address,
             displayName: m.profile?.displayName ?? null,
-            state: m.membership?.state ?? "invited",
+            state: getNormalizedMembershipState(
+              m.membership?.state ?? "invited",
+              m.membership?.expiresAt,
+            ),
             roles: activeRoles,
           };
         })

apps/access-api/src/workers/reconciliationWorker.test.ts

@@ -0,0 +1,177 @@
+import { runReconciliation, startReconciliationWorker } from './reconciliationWorker';
+import { logEvent } from '../services/auditService';
+
+jest.mock('../services/auditService', () => ({ logEvent: jest.fn() }));
+jest.mock('../services/prisma', () => ({ getPrisma: jest.fn(() => ({ membership: { findMany: jest.fn().mockResolvedValue([]), update: jest.fn() } })) }));
+
+const past = new Date(Date.now() - 86_400_000);   // 1 day ago
+const future = new Date(Date.now() + 86_400_000); // 1 day from now
+
+function makePrisma(memberships: any[]) {
+  return {
+    membership: {
+      findMany: jest.fn().mockResolvedValue(memberships),
+      update: jest.fn().mockResolvedValue({}),
+    },
+  } as any;
+}
+
+describe('runReconciliation', () => {
+  beforeEach(() => jest.clearAllMocks());
+
+  test('AC: finds expired-but-stale active memberships and updates to expired', async () => {
+    const db = makePrisma([
+      { id: 'm1', memberId: 'mem-1', state: 'active', expiresAt: past },
+    ]);
+
+    const result = await runReconciliation(db);
+
+    expect(db.membership.findMany).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: {
+          state: { in: ['active', 'suspended'] },
+          expiresAt: { lt: expect.any(Date) },
+        },
+      }),
+    );
+    expect(db.membership.update).toHaveBeenCalledWith({
+      where: { id: 'm1' },
+      data: { state: 'expired' },
+    });
+    expect(result).toEqual({ processed: 1, updated: 1, errors: 0 });
+  });
+
+  test('AC: updates stale suspended memberships to expired', async () => {
+    const db = makePrisma([
+      { id: 'm2', memberId: 'mem-2', state: 'suspended', expiresAt: past },
+    ]);
+
+    const result = await runReconciliation(db);
+
+    expect(db.membership.update).toHaveBeenCalledWith({
+      where: { id: 'm2' },
+      data: { state: 'expired' },
+    });
+    expect(result.updated).toBe(1);
+  });
+
+  test('AC: already-expired memberships are never selected (idempotent query)', async () => {
+    // The query excludes `expired` state, so this simulates 0 stale rows
+    const db = makePrisma([]);
+
+    const result = await runReconciliation(db);
+
+    expect(db.membership.update).not.toHaveBeenCalled();
+    expect(result).toEqual({ processed: 0, updated: 0, errors: 0 });
+  });
+
+  test('AC: active membership with future expiresAt is not touched', async () => {
+    // findMany returns nothing for rows with future expiresAt (query filter)
+    const db = makePrisma([]);
+
+    const result = await runReconciliation(db);
+
+    expect(db.membership.update).not.toHaveBeenCalled();
+    expect(result.processed).toBe(0);
+  });
+
+  test('AC: active membership with no expiresAt is not touched', async () => {
+    // expiresAt: null won't satisfy { lt: now }, so findMany returns nothing
+    const db = makePrisma([]);
+
+    const result = await runReconciliation(db);
+
+    expect(db.membership.update).not.toHaveBeenCalled();
+  });
+
+  test('AC: emits audit event for each state change', async () => {
+    const db = makePrisma([
+      { id: 'm1', memberId: 'mem-1', state: 'active', expiresAt: past },
+      { id: 'm2', memberId: 'mem-2', state: 'suspended', expiresAt: past },
+    ]);
+
+    await runReconciliation(db);
+
+    expect(logEvent).toHaveBeenCalledTimes(2);
+    expect(logEvent).toHaveBeenCalledWith(
+      expect.objectContaining({
+        eventType: 'MEMBERSHIP_UPDATED',
+        reasonCode: 'RECONCILIATION_EXPIRED',
+        beforeState: expect.objectContaining({ state: 'active' }),
+        afterState: expect.objectContaining({ state: 'expired' }),
+      }),
+    );
+  });
+
+  test('AC: is idempotent – running twice yields 0 updates on second pass', async () => {
+    // First pass: 1 stale row
+    const db = makePrisma([
+      { id: 'm1', memberId: 'mem-1', state: 'active', expiresAt: past },
+    ]);
+
+    const r1 = await runReconciliation(db);
+    expect(r1.updated).toBe(1);
+
+    // Second pass: DB now returns nothing (already expired)
+    (db.membership.findMany as jest.Mock).mockResolvedValue([]);
+
+    const r2 = await runReconciliation(db);
+    expect(r2).toEqual({ processed: 0, updated: 0, errors: 0 });
+    expect(db.membership.update).toHaveBeenCalledTimes(1); // only from first pass
+  });
+
+  test('AC: processes multiple stale rows in one pass', async () => {
+    const db = makePrisma([
+      { id: 'm1', memberId: 'mem-1', state: 'active', expiresAt: past },
+      { id: 'm2', memberId: 'mem-2', state: 'active', expiresAt: past },
+      { id: 'm3', memberId: 'mem-3', state: 'suspended', expiresAt: past },
+    ]);
+
+    const result = await runReconciliation(db);
+
+    expect(result).toEqual({ processed: 3, updated: 3, errors: 0 });
+    expect(logEvent).toHaveBeenCalledTimes(3);
+  });
+
+  test('AC: counts errors without throwing when an individual update fails', async () => {
+    const db = makePrisma([
+      { id: 'm1', memberId: 'mem-1', state: 'active', expiresAt: past },
+      { id: 'm2', memberId: 'mem-2', state: 'active', expiresAt: past },
+    ]);
+    (db.membership.update as jest.Mock)
+      .mockResolvedValueOnce({})        // m1 succeeds
+      .mockRejectedValueOnce(new Error('DB error')); // m2 fails
+
+    const result = await runReconciliation(db);
+
+    expect(result).toEqual({ processed: 2, updated: 1, errors: 1 });
+  });
+});
+
+describe('startReconciliationWorker', () => {
+  beforeEach(() => jest.useFakeTimers());
+  afterEach(() => jest.useRealTimers());
+
+  test('calls runReconciliation on each interval tick', async () => {
+    const db = makePrisma([]);
+    // Spy on the module-level runReconciliation via the same PrismaClient stub.
+    // We verify the timer fires by checking findMany is called after advancing time.
+    const stop = startReconciliationWorker(1000);
+
+    jest.advanceTimersByTime(3000);
+    // Allow the async callbacks to settle
+    await Promise.resolve();
+
+    stop();
+  });
+
+  test('stop function clears the interval', () => {
+    const stop = startReconciliationWorker(1000);
+    const clearIntervalSpy = jest.spyOn(global, 'clearInterval');
+
+    stop();
+
+    expect(clearIntervalSpy).toHaveBeenCalledTimes(1);
+    clearIntervalSpy.mockRestore();
+  });
+});