Merge pull request #198 from Demilade01/main

feat(jobs): add ownership snapshot cleanup job scaffold

Author: Chucks1093

docs/indexer/ownership-snapshot-cleanup.md

@@ -0,0 +1,61 @@
+# Ownership Snapshot Cleanup Scaffold
+
+This scaffold adds a scheduled cleanup job that targets expired ownership snapshot rows.
+
+The implementation is intentionally conservative:
+
+- it runs only when explicitly enabled
+- it supports dry-run mode by default
+- it skips cleanup if the target table does not exist yet
+
+## Environment variables
+
+Use the following variables to control behavior:
+
+- `OWNERSHIP_SNAPSHOT_CLEANUP_ENABLED` (default: `false`)
+- `OWNERSHIP_SNAPSHOT_CLEANUP_INTERVAL_MINUTES` (default: `60`)
+- `OWNERSHIP_SNAPSHOT_RETENTION_DAYS` (default: `30`)
+- `OWNERSHIP_SNAPSHOT_CLEANUP_DRY_RUN` (default: `true`)
+- `OWNERSHIP_SNAPSHOT_TABLE_NAME` (default: `creator_ownership_snapshots`)
+
+## How cleanup works
+
+On each run, the job computes a cutoff timestamp as:
+
+`now - OWNERSHIP_SNAPSHOT_RETENTION_DAYS`
+
+Then it checks whether the configured table exists.
+
+- In dry-run mode, it only counts rows where `expiresAt < cutoff`.
+- In non-dry-run mode, it deletes rows where `expiresAt < cutoff`.
+
+## Local validation steps
+
+1. Configure `.env`:
+
+```bash
+OWNERSHIP_SNAPSHOT_CLEANUP_ENABLED=true
+OWNERSHIP_SNAPSHOT_CLEANUP_INTERVAL_MINUTES=1
+OWNERSHIP_SNAPSHOT_RETENTION_DAYS=30
+OWNERSHIP_SNAPSHOT_CLEANUP_DRY_RUN=true
+OWNERSHIP_SNAPSHOT_TABLE_NAME=creator_ownership_snapshots
+```
+
+2. Start the server:
+
+```bash
+pnpm dev
+```
+
+3. Confirm logs show one of the following:
+
+- `Ownership snapshot cleanup dry-run completed`
+- `Ownership snapshot cleanup skipped because target table was not found`
+
+4. Run tests for the scaffold:
+
+```bash
+pnpm exec jest src/jobs/ownership-snapshot-cleanup.job.test.ts
+```
+
+5. Optional delete validation (only when safe): set `OWNERSHIP_SNAPSHOT_CLEANUP_DRY_RUN=false`, run locally against non-production data, and verify delete counts in logs.

src/config.schema.ts

@@ -116,6 +116,16 @@ export const envSchema = z
             'STELLAR_SOROBAN_RPC_URL must be a valid URL (e.g. https://soroban-testnet.stellar.org)'
          )
          .default('https://soroban-testnet.stellar.org'),
+
+      // Ownership snapshot cleanup job
+      OWNERSHIP_SNAPSHOT_TABLE_NAME: z
+         .string()
+         .min(1)
+         .default('creator_ownership_snapshots'),
+      OWNERSHIP_SNAPSHOT_CLEANUP_DRY_RUN: z.coerce.boolean().default(true),
+      OWNERSHIP_SNAPSHOT_RETENTION_DAYS: z.coerce.number().int().positive().default(30),
+      OWNERSHIP_SNAPSHOT_CLEANUP_ENABLED: z.coerce.boolean().default(false),
+      OWNERSHIP_SNAPSHOT_CLEANUP_INTERVAL_MINUTES: z.coerce.number().int().positive().default(60),
    })
    .superRefine((data, ctx) => {
       if (

src/jobs/ownership-snapshot-cleanup.job.test.ts

@@ -0,0 +1,127 @@
+jest.mock('../config', () => ({
+   envConfig: {
+      OWNERSHIP_SNAPSHOT_CLEANUP_DRY_RUN: true,
+      OWNERSHIP_SNAPSHOT_RETENTION_DAYS: 30,
+      OWNERSHIP_SNAPSHOT_TABLE_NAME: 'creator_ownership_snapshots',
+      OWNERSHIP_SNAPSHOT_CLEANUP_ENABLED: false,
+      OWNERSHIP_SNAPSHOT_CLEANUP_INTERVAL_MINUTES: 60,
+   },
+}));
+
+jest.mock('../utils/prisma.utils', () => ({
+   prisma: {
+      $queryRawUnsafe: jest.fn(),
+      $executeRawUnsafe: jest.fn(),
+   },
+}));
+
+jest.mock('../utils/logger.utils', () => ({
+   logger: {
+      info: jest.fn(),
+      warn: jest.fn(),
+      error: jest.fn(),
+   },
+}));
+
+import { envConfig } from '../config';
+import { cleanupExpiredOwnershipSnapshots } from './ownership-snapshot-cleanup.job';
+
+describe('ownership-snapshot-cleanup.job', () => {
+   beforeEach(() => {
+      jest.clearAllMocks();
+      envConfig.OWNERSHIP_SNAPSHOT_CLEANUP_DRY_RUN = true;
+      envConfig.OWNERSHIP_SNAPSHOT_RETENTION_DAYS = 30;
+      envConfig.OWNERSHIP_SNAPSHOT_TABLE_NAME = 'creator_ownership_snapshots';
+   });
+
+   it('returns skipped when the snapshot table does not exist', async () => {
+      const queryRawUnsafe = jest
+         .fn()
+         .mockResolvedValueOnce([{ regclass: null }]);
+
+      const result = await cleanupExpiredOwnershipSnapshots({
+         queryRawUnsafe,
+         now: () => new Date('2026-01-31T00:00:00.000Z'),
+      });
+
+      expect(result).toMatchObject({
+         skipped: true,
+         reason: 'table_not_found',
+         dryRun: true,
+         affectedRows: 0,
+         tableName: 'creator_ownership_snapshots',
+      });
+      expect(queryRawUnsafe).toHaveBeenCalledTimes(1);
+   });
+
+   it('counts rows in dry-run mode', async () => {
+      const queryRawUnsafe = jest
+         .fn()
+         .mockResolvedValueOnce([{ regclass: 'creator_ownership_snapshots' }])
+         .mockResolvedValueOnce([{ count: 12 }]);
+      const executeRawUnsafe = jest.fn();
+
+      const result = await cleanupExpiredOwnershipSnapshots({
+         queryRawUnsafe,
+         executeRawUnsafe,
+         now: () => new Date('2026-01-31T00:00:00.000Z'),
+      });
+
+      expect(result).toMatchObject({
+         skipped: false,
+         dryRun: true,
+         affectedRows: 12,
+         tableName: 'creator_ownership_snapshots',
+      });
+      expect(queryRawUnsafe).toHaveBeenCalledTimes(2);
+      expect(executeRawUnsafe).not.toHaveBeenCalled();
+   });
+
+   it('deletes rows when dry-run is disabled', async () => {
+      envConfig.OWNERSHIP_SNAPSHOT_CLEANUP_DRY_RUN = false;
+
+      const queryRawUnsafe = jest
+         .fn()
+         .mockResolvedValueOnce([{ regclass: 'creator_ownership_snapshots' }]);
+      const executeRawUnsafe = jest.fn().mockResolvedValueOnce(5);
+
+      const result = await cleanupExpiredOwnershipSnapshots({
+         queryRawUnsafe,
+         executeRawUnsafe,
+         now: () => new Date('2026-01-31T00:00:00.000Z'),
+      });
+
+      expect(result).toMatchObject({
+         skipped: false,
+         dryRun: false,
+         affectedRows: 5,
+         tableName: 'creator_ownership_snapshots',
+      });
+      expect(executeRawUnsafe).toHaveBeenCalledTimes(1);
+   });
+
+   it('applies retention days when calculating cutoff timestamp', async () => {
+      envConfig.OWNERSHIP_SNAPSHOT_RETENTION_DAYS = 10;
+
+      const queryRawUnsafe = jest
+         .fn()
+         .mockResolvedValueOnce([{ regclass: 'creator_ownership_snapshots' }])
+         .mockResolvedValueOnce([{ count: 0 }]);
+
+      const now = new Date('2026-02-15T00:00:00.000Z');
+      const expectedCutoff = new Date('2026-02-05T00:00:00.000Z');
+
+      const result = await cleanupExpiredOwnershipSnapshots({
+         queryRawUnsafe,
+         now: () => now,
+      });
+
+      expect(result.cutoffTimestamp.toISOString()).toBe(
+         expectedCutoff.toISOString()
+      );
+      expect(queryRawUnsafe).toHaveBeenLastCalledWith(
+         expect.stringContaining('COUNT(*)::int'),
+         expectedCutoff
+      );
+   });
+});

src/jobs/ownership-snapshot-cleanup.job.ts

@@ -0,0 +1,187 @@
+import { envConfig } from '../config';
+import { logger } from '../utils/logger.utils';
+import { prisma } from '../utils/prisma.utils';
+
+const VALID_IDENTIFIER = /^[A-Za-z_][A-Za-z0-9_]*$/;
+
+export type OwnershipSnapshotCleanupResult = {
+   skipped: boolean;
+   dryRun: boolean;
+   cutoffTimestamp: Date;
+   affectedRows: number;
+   tableName: string;
+   reason?: string;
+};
+
+type OwnershipSnapshotCleanupDeps = {
+   queryRawUnsafe: typeof prisma.$queryRawUnsafe;
+   executeRawUnsafe: typeof prisma.$executeRawUnsafe;
+   now: () => Date;
+};
+
+function getCutoffTimestamp(now: Date, retentionDays: number): Date {
+   const cutoffMs = retentionDays * 24 * 60 * 60 * 1000;
+   return new Date(now.getTime() - cutoffMs);
+}
+
+function assertSafeTableName(tableName: string) {
+   if (!VALID_IDENTIFIER.test(tableName)) {
+      throw new Error(
+         `Invalid OWNERSHIP_SNAPSHOT_TABLE_NAME: ${tableName}. Expected a simple SQL identifier.`
+      );
+   }
+}
+
+async function tableExists(
+   queryRawUnsafe: typeof prisma.$queryRawUnsafe,
+   tableName: string
+): Promise<boolean> {
+   const rows = await queryRawUnsafe<Array<{ regclass: string | null }>>(
+      'SELECT to_regclass($1) AS regclass',
+      `public.${tableName}`
+   );
+
+   return Boolean(rows[0]?.regclass);
+}
+
+export async function cleanupExpiredOwnershipSnapshots(
+   deps?: Partial<OwnershipSnapshotCleanupDeps>
+): Promise<OwnershipSnapshotCleanupResult> {
+   const queryRawUnsafe = deps?.queryRawUnsafe ?? prisma.$queryRawUnsafe.bind(prisma);
+   const executeRawUnsafe =
+      deps?.executeRawUnsafe ?? prisma.$executeRawUnsafe.bind(prisma);
+   const now = deps?.now ?? (() => new Date());
+
+   const tableName = envConfig.OWNERSHIP_SNAPSHOT_TABLE_NAME;
+   const dryRun = envConfig.OWNERSHIP_SNAPSHOT_CLEANUP_DRY_RUN;
+   const cutoffTimestamp = getCutoffTimestamp(
+      now(),
+      envConfig.OWNERSHIP_SNAPSHOT_RETENTION_DAYS
+   );
+
+   assertSafeTableName(tableName);
+
+   const exists = await tableExists(queryRawUnsafe, tableName);
+   if (!exists) {
+      logger.warn(
+         {
+            tableName,
+            cutoffTimestamp: cutoffTimestamp.toISOString(),
+            dryRun,
+         },
+         'Ownership snapshot cleanup skipped because target table was not found'
+      );
+
+      return {
+         skipped: true,
+         reason: 'table_not_found',
+         dryRun,
+         cutoffTimestamp,
+         affectedRows: 0,
+         tableName,
+      };
+   }
+
+   if (dryRun) {
+      const rows = await queryRawUnsafe<Array<{ count: number }>>(
+         `SELECT COUNT(*)::int AS count FROM "${tableName}" WHERE "expiresAt" < $1`,
+         cutoffTimestamp
+      );
+      const count = rows[0]?.count ?? 0;
+
+      logger.info(
+         {
+            tableName,
+            cutoffTimestamp: cutoffTimestamp.toISOString(),
+            retainedDays: envConfig.OWNERSHIP_SNAPSHOT_RETENTION_DAYS,
+            wouldDeleteCount: count,
+            dryRun,
+         },
+         'Ownership snapshot cleanup dry-run completed'
+      );
+
+      return {
+         skipped: false,
+         dryRun,
+         cutoffTimestamp,
+         affectedRows: count,
+         tableName,
+      };
+   }
+
+   const deletedCount = await executeRawUnsafe(
+      `DELETE FROM "${tableName}" WHERE "expiresAt" < $1`,
+      cutoffTimestamp
+   );
+
+   logger.info(
+      {
+         tableName,
+         cutoffTimestamp: cutoffTimestamp.toISOString(),
+         retainedDays: envConfig.OWNERSHIP_SNAPSHOT_RETENTION_DAYS,
+         deletedCount,
+         dryRun,
+      },
+      'Ownership snapshot cleanup completed'
+   );
+
+   return {
+      skipped: false,
+      dryRun,
+      cutoffTimestamp,
+      affectedRows: deletedCount,
+      tableName,
+   };
+}
+
+let cleanupTimer: NodeJS.Timeout | null = null;
+
+export function startOwnershipSnapshotCleanupJob() {
+   if (!envConfig.OWNERSHIP_SNAPSHOT_CLEANUP_ENABLED) {
+      logger.info('Ownership snapshot cleanup job is disabled');
+      return;
+   }
+
+   const intervalMs =
+      envConfig.OWNERSHIP_SNAPSHOT_CLEANUP_INTERVAL_MINUTES * 60 * 1000;
+
+   const run = async () => {
+      try {
+         await cleanupExpiredOwnershipSnapshots();
+      } catch (error) {
+         logger.error(
+            { err: error },
+            'Ownership snapshot cleanup failed with an unexpected error'
+         );
+      }
+   };
+
+   void run();
+   cleanupTimer = setInterval(() => {
+      void run();
+   }, intervalMs);
+
+   if (typeof cleanupTimer.unref === 'function') {
+      cleanupTimer.unref();
+   }
+
+   logger.info(
+      {
+         intervalMinutes: envConfig.OWNERSHIP_SNAPSHOT_CLEANUP_INTERVAL_MINUTES,
+         retentionDays: envConfig.OWNERSHIP_SNAPSHOT_RETENTION_DAYS,
+         dryRun: envConfig.OWNERSHIP_SNAPSHOT_CLEANUP_DRY_RUN,
+         tableName: envConfig.OWNERSHIP_SNAPSHOT_TABLE_NAME,
+      },
+      'Ownership snapshot cleanup job started'
+   );
+}
+
+export function stopOwnershipSnapshotCleanupJob() {
+   if (!cleanupTimer) {
+      return;
+   }
+
+   clearInterval(cleanupTimer);
+   cleanupTimer = null;
+   logger.info('Ownership snapshot cleanup job stopped');
+}