feat(gdpr): add data retention and purge jobs

backend/gdpr/__tests__/postgresRetentionStore.test.ts

@@ -0,0 +1,164 @@
+import { PostgresRetentionStore, createDefaultRetentionStores } from '../postgresRetentionStore';
+import type { Pool, QueryResult, PoolClient } from '../../shared/db/connectionPool';
+
+/** Mock pool that captures queries and returns canned results. */
+function createMockPool(): Pool & {
+  queries: { sql: string; params: unknown[] }[];
+  setResult: (rows: unknown[]) => void;
+} {
+  const queries: { sql: string; params: unknown[] }[] = [];
+  let nextRows: unknown[] = [];
+
+  const pool: Pool & {
+    queries: { sql: string; params: unknown[] }[];
+    setResult: (rows: unknown[]) => void;
+  } = {
+    totalCount: 0,
+    idleCount: 0,
+    waitingCount: 0,
+    queries,
+    setResult: (rows: unknown[]) => { nextRows = rows; },
+    query: jest.fn(async <T = unknown>(sql: string, params?: unknown[]): Promise<QueryResult<T>> => {
+      queries.push({ sql, params: params ?? [] });
+      return { rows: nextRows as T[], rowCount: nextRows.length };
+    }),
+    connect: jest.fn(async (): Promise<PoolClient> => ({
+      query: jest.fn(async <T = unknown>(sql: string, params?: unknown[]): Promise<QueryResult<T>> => {
+        queries.push({ sql, params: params ?? [] });
+        return { rows: nextRows as T[], rowCount: nextRows.length };
+      }),
+      release: jest.fn(),
+    })),
+    end: jest.fn(),
+    on: jest.fn(),
+  };
+  return pool;
+}
+
+describe('PostgresRetentionStore', () => {
+  it('findExpired queries with cutoff and excludes legal holds', async () => {
+    const pool = createMockPool();
+    pool.setResult([
+      { id: 'r1', user_id: 'u1', ts: new Date('2026-01-01') },
+      { id: 'r2', user_id: null, ts: new Date('2026-01-02') },
+    ]);
+
+    const store = new PostgresRetentionStore(pool, {
+      sourceName: 'activity_logs',
+      table: 'activity_logs',
+    });
+
+    const records = await store.findExpired(Date.UTC(2026, 2, 1), 100);
+
+    expect(records).toHaveLength(2);
+    expect(records[0].id).toBe('r1');
+    expect(records[0].userId).toBe('u1');
+    expect(records[1].userId).toBeUndefined();
+
+    const query = pool.queries[0];
+    expect(query.sql).toContain('WHERE created_at <= $1');
+    expect(query.sql).toContain('retention_legal_holds');
+    expect(query.sql).toContain('ORDER BY created_at ASC');
+    expect(query.params[1]).toBe('activity_logs');
+    expect(query.params[2]).toBe(100);
+  });
+
+  it('deleteRecords returns rowCount', async () => {
+    const pool = createMockPool();
+    pool.setResult([]);
+
+    const store = new PostgresRetentionStore(pool, {
+      sourceName: 'notifications',
+      table: 'notifications',
+    });
+
+    // Override rowCount
+    (pool.query as jest.Mock).mockResolvedValueOnce({ rows: [], rowCount: 5 });
+
+    const count = await store.deleteRecords(['a', 'b', 'c']);
+    expect(count).toBe(5);
+    expect(pool.queries[0].sql).toContain('DELETE FROM notifications');
+    expect(pool.queries[0].sql).toContain('ANY($1::text[])');
+  });
+
+  it('deleteRecords returns 0 for empty array', async () => {
+    const pool = createMockPool();
+    const store = new PostgresRetentionStore(pool, { sourceName: 'x', table: 'x' });
+    const count = await store.deleteRecords([]);
+    expect(count).toBe(0);
+    expect(pool.queries).toHaveLength(0);
+  });
+
+  it('anonymizeRecords updates PII columns', async () => {
+    const pool = createMockPool();
+    (pool.query as jest.Mock).mockResolvedValueOnce({ rows: [], rowCount: 3 });
+
+    const store = new PostgresRetentionStore(pool, {
+      sourceName: 'billing_records',
+      table: 'billing_records',
+    });
+
+    const count = await store.anonymizeRecords(['r1', 'r2', 'r3']);
+    expect(count).toBe(3);
+    expect(pool.queries[0].sql).toContain('SET email = NULL');
+    expect(pool.queries[0].sql).toContain('name = NULL');
+  });
+
+  it('anonymizeRecords returns 0 for empty array', async () => {
+    const pool = createMockPool();
+    const store = new PostgresRetentionStore(pool, { sourceName: 'x', table: 'x' });
+    expect(await store.anonymizeRecords([])).toBe(0);
+  });
+
+  it('archiveRecords sets archived_at', async () => {
+    const pool = createMockPool();
+    (pool.query as jest.Mock).mockResolvedValueOnce({ rows: [], rowCount: 2 });
+
+    const store = new PostgresRetentionStore(pool, { sourceName: 'x', table: 'x' });
+    const count = await store.archiveRecords(['r1', 'r2']);
+    expect(count).toBe(2);
+    expect(pool.queries[0].sql).toContain('archived_at = now()');
+  });
+
+  it('archiveRecords returns 0 for empty array', async () => {
+    const pool = createMockPool();
+    const store = new PostgresRetentionStore(pool, { sourceName: 'x', table: 'x' });
+    expect(await store.archiveRecords([])).toBe(0);
+  });
+
+  it('uses custom column names when provided', async () => {
+    const pool = createMockPool();
+    pool.setResult([]);
+
+    const store = new PostgresRetentionStore(pool, {
+      sourceName: 'audit_logs',
+      table: 'audit_events',
+      timestampColumn: 'created_at',
+      idColumn: 'id',
+      userIdColumn: 'actor_id',
+    });
+
+    await store.findExpired(Date.now(), 50);
+    expect(pool.queries[0].sql).toContain('actor_id AS user_id');
+  });
+});
+
+describe('createDefaultRetentionStores', () => {
+  it('creates stores for all default sources', () => {
+    const pool = createMockPool();
+    const stores = createDefaultRetentionStores(pool);
+    expect(stores.size).toBe(4);
+    expect(stores.has('activity_logs')).toBe(true);
+    expect(stores.has('notifications')).toBe(true);
+    expect(stores.has('billing_records')).toBe(true);
+    expect(stores.has('audit_logs')).toBe(true);
+  });
+
+  it('each store has the correct sourceName', () => {
+    const pool = createMockPool();
+    const stores = createDefaultRetentionStores(pool);
+    for (const [name, store] of stores) {
+      expect(store.sourceName).toBe(name);
+    }
+  });
+});

backend/gdpr/__tests__/purgeJobScheduler.test.ts

@@ -0,0 +1,102 @@
+import { PurgeJobScheduler } from '../purgeJobScheduler';
+import type { Pool, QueryResult, PoolClient } from '../../shared/db/connectionPool';
+
+function createMockPool(): Pool & {
+  queries: { sql: string; params: unknown[] }[];
+  setRows: (rows: unknown[]) => void;
+} {
+  const queries: { sql: string; params: unknown[] }[] = [];
+  let rows: unknown[] = [];
+  const pool: Pool & {
+    queries: { sql: string; params: unknown[] }[];
+    setRows: (r: unknown[]) => void;
+  } = {
+    totalCount: 0, idleCount: 0, waitingCount: 0,
+    queries, setRows: (r: unknown[]) => { rows = r; },
+    query: jest.fn(async <T = unknown>(sql: string, params?: unknown[]): Promise<QueryResult<T>> => {
+      queries.push({ sql, params: params ?? [] });
+      return { rows: rows as T[], rowCount: rows.length };
+    }),
+    connect: jest.fn(async (): Promise<PoolClient> => ({
+      query: jest.fn(async (): Promise<QueryResult<unknown>> => ({ rows: rows as unknown[], rowCount: rows.length })),
+      release: jest.fn(),
+    })),
+    end: jest.fn(),
+    on: jest.fn(),
+  };
+  return pool;
+}
+
+describe('PurgeJobScheduler', () => {
+  it('constructs with default policies and stores', () => {
+    const pool = createMockPool();
+    const scheduler = new PurgeJobScheduler(pool);
+    expect(scheduler.isRunning()).toBe(false);
+    expect(scheduler.getService()).toBeDefined();
+    expect(scheduler.getService().listPolicies().length).toBeGreaterThan(0);
+  });
+
+  it('start loads policies from DB and begins enforcement', async () => {
+    const pool = createMockPool();
+    pool.setRows([
+      { id: 'custom-policy', source: 'custom_source', description: 'test',
+        retention_days: 60, action: 'delete', enabled: true, batch_size: 500 },
+    ]);
+    const scheduler = new PurgeJobScheduler(pool, { intervalMs: 999_999_999 });
+    await scheduler.start();
+    expect(scheduler.isRunning()).toBe(true);
+    // The custom policy should have been loaded
+    const policy = scheduler.getService().getPolicy('custom-policy');
+    expect(policy).not.toBeNull();
+    expect(policy?.retentionDays).toBe(60);
+    scheduler.stop();
+  });
+
+  it('stop halts the scheduler', async () => {
+    const pool = createMockPool();
+    const scheduler = new PurgeJobScheduler(pool, { intervalMs: 999_999_999 });
+    await scheduler.start();
+    expect(scheduler.isRunning()).toBe(true);
+    scheduler.stop();
+    expect(scheduler.isRunning()).toBe(false);
+  });
+
+  it('start is idempotent', async () => {
+    const pool = createMockPool();
+    const scheduler = new PurgeJobScheduler(pool, { intervalMs: 999_999_999 });
+    await scheduler.start();
+    await scheduler.start();
+    expect(scheduler.isRunning()).toBe(true);
+    scheduler.stop();
+  });
+
+  it('handles DB load failure gracefully (defaults remain)', async () => {
+    const pool = createMockPool();
+    (pool.query as jest.Mock).mockRejectedValueOnce(new Error('table missing'));
+    const scheduler = new PurgeJobScheduler(pool, { intervalMs: 999_999_999 });
+    await scheduler.start();
+    // Default policies should still be present
+    expect(scheduler.getService().listPolicies().length).toBeGreaterThan(0);
+    scheduler.stop();
+  });
+
+  it('runNow triggers an enforcement run', async () => {
+    const pool = createMockPool();
+    const scheduler = new PurgeJobScheduler(pool, { intervalMs: 999_999_999 });
+    const report = await scheduler.runNow();
+    expect(report).not.toBeNull();
+    expect(report?.results).toBeDefined();
+  });
+
+  it('persistReport writes to retention_enforcement_log', async () => {
+    const pool = createMockPool();
+    const onReport = jest.fn();
+    const scheduler = new PurgeJobScheduler(pool, {
+      intervalMs: 999_999_999,
+      onReport,
+    });
+    await scheduler.runNow();
+    // The default onReport was overridden, so no DB writes for logging
+    expect(onReport).toHaveBeenCalled();
+  });
+});

backend/gdpr/index.ts

@@ -54,3 +54,7 @@ export type {
 
 export { RetentionEnforcementJob } from './retentionEnforcementJob';
 export { RetentionController } from './retentionController';
+export { PostgresRetentionStore, createDefaultRetentionStores } from './postgresRetentionStore';
+export type { PostgresRetentionStoreConfig } from './postgresRetentionStore';
+export { PurgeJobScheduler } from './purgeJobScheduler';
+export type { PurgeJobSchedulerOptions } from './purgeJobScheduler';

backend/gdpr/postgresRetentionStore.ts

@@ -0,0 +1,96 @@
+/**
+ * PostgreSQL-backed retention data store (issue #1799).
+ *
+ * Implements the RetentionDataStore interface against actual database tables,
+ * so the DataRetentionService can purge expired records from PostgreSQL
+ * instead of relying on in-memory mocks.
+ */
+
+import type { Pool } from '../shared/db/connectionPool';
+import type { RetentionDataStore, RetentionRecord } from './dataRetentionService';
+
+export interface PostgresRetentionStoreConfig {
+  sourceName: string;
+  table: string;
+  timestampColumn?: string;
+  idColumn?: string;
+  userIdColumn?: string;
+}
+
+export class PostgresRetentionStore implements RetentionDataStore {
+  public readonly sourceName: string;
+  private readonly table: string;
+  private readonly tsCol: string;
+  private readonly idCol: string;
+  private readonly userIdCol: string;
+
+  constructor(
+    private pool: Pool,
+    config: PostgresRetentionStoreConfig,
+  ) {
+    this.sourceName = config.sourceName;
+    this.table = config.table;
+    this.tsCol = config.timestampColumn ?? 'created_at';
+    this.idCol = config.idColumn ?? 'id';
+    this.userIdCol = config.userIdColumn ?? 'user_id';
+  }
+
+  async findExpired(cutoff: number, limit: number): Promise<RetentionRecord[]> {
+    const cutoffDate = new Date(cutoff).toISOString();
+    const result = await this.pool.query<{
+      id: string; user_id: string | null; ts: number | Date;
+    }>(
+      `SELECT ${this.idCol} AS id, ${this.userIdCol} AS user_id, ${this.tsCol} AS ts
+       FROM ${this.table}
+       WHERE ${this.tsCol} <= $1
+         AND ${this.idCol} NOT IN (
+           SELECT target_id::text FROM retention_legal_holds
+           WHERE scope = 'record' AND released_at IS NULL
+             AND (source IS NULL OR source = $2))
+         AND (${this.userIdCol} IS NULL OR ${this.userIdCol} NOT IN (
+           SELECT target_id::text FROM retention_legal_holds
+           WHERE scope = 'user' AND released_at IS NULL
+             AND (source IS NULL OR source = $2)))
+       ORDER BY ${this.tsCol} ASC LIMIT $3`,
+      [cutoffDate, this.sourceName, limit],
+    );
+    return result.rows.map((r) => ({
+      id: r.id, userId: r.user_id ?? undefined,
+      timestamp: typeof r.ts === 'number' ? r.ts : r.ts.getTime(),
+    }));
+  }
+
+  async deleteRecords(ids: string[]): Promise<number> {
+    if (ids.length === 0) return 0;
+    const result = await this.pool.query(
+      `DELETE FROM ${this.table} WHERE ${this.idCol} = ANY($1::text[])`, [ids]);
+    return result.rowCount ?? 0;
+  }
+
+  async anonymizeRecords(ids: string[]): Promise<number> {
+    if (ids.length === 0) return 0;
+    const result = await this.pool.query(
+      `UPDATE ${this.table} SET email = NULL, name = NULL, phone = NULL, address = NULL
+       WHERE ${this.idCol} = ANY($1::text[])`, [ids]);
+    return result.rowCount ?? 0;
+  }
+
+  async archiveRecords(ids: string[]): Promise<number> {
+    if (ids.length === 0) return 0;
+    const result = await this.pool.query(
+      `UPDATE ${this.table} SET archived_at = now() WHERE ${this.idCol} = ANY($1::text[])`, [ids]);
+    return result.rowCount ?? 0;
+  }
+}
+
+export function createDefaultRetentionStores(pool: Pool): Map<string, PostgresRetentionStore> {
+  const configs: PostgresRetentionStoreConfig[] = [
+    { sourceName: 'activity_logs', table: 'activity_logs' },
+    { sourceName: 'notifications', table: 'notifications' },
+    { sourceName: 'billing_records', table: 'billing_records' },
+    { sourceName: 'audit_logs', table: 'audit_events', timestampColumn: 'created_at' },
+  ];
+  const stores = new Map<string, PostgresRetentionStore>();
+  for (const c of configs) stores.set(c.sourceName, new PostgresRetentionStore(pool, c));
+  return stores;
+}