From cb0657077a63bc7452fbcbbe558e9d83be89cdd5 Mon Sep 17 00:00:00 2001 From: nissinanlung <269563815+nissinanlung@users.noreply.github.com> Date: Sat, 17 Oct 2026 10:40:44 +0000 Subject: [PATCH] feat(gdpr): add data retention and purge jobs (#1799) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add SQL-backed retention stores, a purge job scheduler, and database tables for persistent policy storage, enforcement audit logging, and legal holds. Changes: - Create db/migrations/010_data_retention_purge_jobs.sql: retention_policies, retention_enforcement_log, and retention_legal_holds tables with seed data for default policies. Includes -- @down rollback section. - Create backend/gdpr/postgresRetentionStore.ts: implements RetentionDataStore against PostgreSQL with findExpired (excluding legal holds), deleteRecords, anonymizeRecords, and archiveRecords. Factory for default SubTrackr tables. - Create backend/gdpr/purgeJobScheduler.ts: coordinates DataRetentionService, PostgresRetentionStores, and RetentionEnforcementJob. Loads persistent policies from DB on startup, logs enforcement results to audit table. - Create backend/gdpr/__tests__/postgresRetentionStore.test.ts: 11 test cases covering findExpired, delete, anonymize, archive, empty arrays, custom columns, and factory. - Create backend/gdpr/__tests__/purgeJobScheduler.test.ts: 7 test cases covering construction, start/stop, idempotency, DB failure handling, manual enforcement, and report persistence. - Export new types from backend/gdpr/index.ts - Create docs/DATA_RETENTION.md: architecture, policies, legal holds, scheduling, REST API, and configuration reference Acceptance criteria: - [x] Policies documented — docs/DATA_RETENTION.md with 4 default policies, actions, and configuration - [x] Jobs scheduled — PurgeJobScheduler runs RetentionEnforcementJob every 24h with configurable interval - [x] Tested — 18 new test cases covering store operations and scheduler --- .../__tests__/postgresRetentionStore.test.ts | 164 ++++++++++++++++++ .../gdpr/__tests__/purgeJobScheduler.test.ts | 102 +++++++++++ backend/gdpr/index.ts | 4 + backend/gdpr/postgresRetentionStore.ts | 96 ++++++++++ backend/gdpr/purgeJobScheduler.ts | 155 +++++++++++++++++ .../010_data_retention_purge_jobs.sql | 108 ++++++++++++ docs/DATA_RETENTION.md | 116 +++++++++++++ 7 files changed, 745 insertions(+) create mode 100644 backend/gdpr/__tests__/postgresRetentionStore.test.ts create mode 100644 backend/gdpr/__tests__/purgeJobScheduler.test.ts create mode 100644 backend/gdpr/postgresRetentionStore.ts create mode 100644 backend/gdpr/purgeJobScheduler.ts create mode 100644 db/migrations/010_data_retention_purge_jobs.sql create mode 100644 docs/DATA_RETENTION.md diff --git a/backend/gdpr/__tests__/postgresRetentionStore.test.ts b/backend/gdpr/__tests__/postgresRetentionStore.test.ts new file mode 100644 index 00000000..a381c517 --- /dev/null +++ b/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 (sql: string, params?: unknown[]): Promise> => { + queries.push({ sql, params: params ?? [] }); + return { rows: nextRows as T[], rowCount: nextRows.length }; + }), + connect: jest.fn(async (): Promise => ({ + query: jest.fn(async (sql: string, params?: unknown[]): Promise> => { + 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); + } + }); +}); diff --git a/backend/gdpr/__tests__/purgeJobScheduler.test.ts b/backend/gdpr/__tests__/purgeJobScheduler.test.ts new file mode 100644 index 00000000..c2769799 --- /dev/null +++ b/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 (sql: string, params?: unknown[]): Promise> => { + queries.push({ sql, params: params ?? [] }); + return { rows: rows as T[], rowCount: rows.length }; + }), + connect: jest.fn(async (): Promise => ({ + query: jest.fn(async (): Promise> => ({ 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(); + }); +}); diff --git a/backend/gdpr/index.ts b/backend/gdpr/index.ts index 292ed853..4f48bd91 100644 --- a/backend/gdpr/index.ts +++ b/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'; diff --git a/backend/gdpr/postgresRetentionStore.ts b/backend/gdpr/postgresRetentionStore.ts new file mode 100644 index 00000000..8e9acfb1 --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + 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 { + 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(); + for (const c of configs) stores.set(c.sourceName, new PostgresRetentionStore(pool, c)); + return stores; +} diff --git a/backend/gdpr/purgeJobScheduler.ts b/backend/gdpr/purgeJobScheduler.ts new file mode 100644 index 00000000..04fb2f54 --- /dev/null +++ b/backend/gdpr/purgeJobScheduler.ts @@ -0,0 +1,155 @@ +/** + * Purge job scheduler (issue #1799). + * + * Wires the DataRetentionService with PostgresRetentionStores and the + * RetentionEnforcementJob into a single coordinator that: + * 1. Loads persistent policies from the database on startup + * 2. Registers SQL-backed stores for each data source + * 3. Starts the periodic enforcement job + * 4. Logs enforcement results to retention_enforcement_log + * + * Usage: + * const scheduler = new PurgeJobScheduler(pool); + * await scheduler.start(); // begins periodic enforcement + * await scheduler.stop(); // stops the scheduler + */ + +import type { Pool } from '../shared/db/connectionPool'; +import { + DataRetentionService, + DEFAULT_RETENTION_POLICIES, + type DataRetentionPolicy, + type RetentionEnforcementReport, +} from './dataRetentionService'; +import { RetentionEnforcementJob } from './retentionEnforcementJob'; +import { + PostgresRetentionStore, + createDefaultRetentionStores, +} from './postgresRetentionStore'; + +export interface PurgeJobSchedulerOptions { + /** Enforcement interval in ms. Default: 24h */ + intervalMs?: number; + /** Called after each enforcement run with the report */ + onReport?: (report: RetentionEnforcementReport) => void; +} + +const DAY_MS = 24 * 60 * 60 * 1000; + +export class PurgeJobScheduler { + private service: DataRetentionService; + private job: RetentionEnforcementJob; + private stores: Map; + private started = false; + + constructor( + private pool: Pool, + opts: PurgeJobSchedulerOptions = {}, + ) { + this.service = new DataRetentionService(DEFAULT_RETENTION_POLICIES); + this.stores = createDefaultRetentionStores(pool); + + // Register all stores + for (const store of this.stores.values()) { + this.service.registerStore(store); + } + + const onReport = opts.onReport ?? ((report: RetentionEnforcementReport) => { + void this.persistReport(report); + }); + + this.job = new RetentionEnforcementJob(this.service, { + intervalMs: opts.intervalMs ?? DAY_MS, + onReport, + }); + } + + /** Load persistent policies from DB, register stores, and start the job. */ + async start(): Promise { + if (this.started) return; + await this.loadPoliciesFromDb(); + this.job.start(); + this.started = true; + } + + /** Stop the periodic enforcement job. */ + stop(): void { + this.job.stop(); + this.started = false; + } + + /** True when the scheduler is actively running. */ + isRunning(): boolean { + return this.started && this.job.isScheduled(); + } + + /** Return the underlying service for manual enforcement / API access. */ + getService(): DataRetentionService { + return this.service; + } + + /** Trigger an immediate enforcement run (outside the schedule). */ + async runNow(): Promise { + return this.job.run(); + } + + /** Load persistent retention policies from the database. */ + private async loadPoliciesFromDb(): Promise { + try { + const result = await this.pool.query<{ + id: string; source: string; description: string | null; + retention_days: number; action: string; enabled: boolean; batch_size: number; + }>(`SELECT id, source, description, retention_days, action, enabled, batch_size + FROM retention_policies`); + + for (const row of result.rows) { + const policy: DataRetentionPolicy = { + id: row.id, + source: row.source, + description: row.description ?? undefined, + retentionDays: row.retention_days, + action: row.action as 'delete' | 'anonymize' | 'archive', + enabled: row.enabled, + batchSize: row.batch_size, + }; + try { + this.service.upsertPolicy(policy); + } catch { + // Skip invalid policies — defaults are already loaded + } + } + } catch { + // Table may not exist yet — defaults are already loaded + } + } + + /** Persist an enforcement report to the audit log table. */ + private async persistReport(report: RetentionEnforcementReport): Promise { + for (const result of report.results) { + try { + await this.pool.query( + `INSERT INTO retention_enforcement_log + (run_id, policy_id, source, action, cutoff, expired_found, + skipped_legal_hold, processed, error, dry_run, started_at, completed_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`, + [ + report.runId, + result.policyId, + result.source, + result.action, + new Date(result.cutoff).toISOString(), + result.expiredFound, + result.skippedLegalHold, + result.processed, + result.error ?? null, + report.dryRun, + new Date(report.startedAt).toISOString(), + new Date(report.completedAt).toISOString(), + ], + ); + } catch { + // Logging is best-effort — don't fail enforcement + } + } + } +} diff --git a/db/migrations/010_data_retention_purge_jobs.sql b/db/migrations/010_data_retention_purge_jobs.sql new file mode 100644 index 00000000..2f72801b --- /dev/null +++ b/db/migrations/010_data_retention_purge_jobs.sql @@ -0,0 +1,108 @@ +-- ── Migration 010: Data retention and purge jobs (issue #1799) ──────────────── +-- +-- Creates tables for persistent retention policy storage and an enforcement +-- audit log. The DataRetentionService and RetentionEnforcementJob already exist +-- in backend/gdpr/; this migration adds the database backing store so policies +-- survive restarts and enforcement runs are auditable. +-- +-- Tables: +-- retention_policies — persistent policy definitions (override defaults) +-- retention_enforcement_log — audit trail of every enforcement run +-- retention_legal_holds — persistent legal holds +-- +-- Run with: psql $DATABASE_URL -f 010_data_retention_purge_jobs.sql + +-- ── retention_policies ─────────────────────────────────────────────────────── +-- Persistent retention policy definitions. The application loads these on +-- startup and merges with DEFAULT_RETENTION_POLICIES. + +CREATE TABLE IF NOT EXISTS retention_policies ( + id TEXT PRIMARY KEY, + source TEXT NOT NULL, + description TEXT, + retention_days INTEGER NOT NULL CHECK (retention_days >= 1 AND retention_days <= 3650), + action TEXT NOT NULL CHECK (action IN ('delete', 'anonymize', 'archive')), + enabled BOOLEAN NOT NULL DEFAULT true, + batch_size INTEGER NOT NULL DEFAULT 1000 CHECK (batch_size >= 1), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_retention_policies_source + ON retention_policies (source) WHERE enabled = true; + +CREATE INDEX IF NOT EXISTS idx_retention_policies_enabled + ON retention_policies (enabled); + +-- ── retention_enforcement_log ───────────────────────────────────────────────── +-- Append-only audit trail of every enforcement run. One row per policy per run. + +CREATE TABLE IF NOT EXISTS retention_enforcement_log ( + id BIGSERIAL PRIMARY KEY, + run_id UUID NOT NULL, + policy_id TEXT NOT NULL, + source TEXT NOT NULL, + action TEXT NOT NULL, + cutoff TIMESTAMPTZ NOT NULL, + expired_found INTEGER NOT NULL DEFAULT 0, + skipped_legal_hold INTEGER NOT NULL DEFAULT 0, + processed INTEGER NOT NULL DEFAULT 0, + error TEXT, + dry_run BOOLEAN NOT NULL DEFAULT false, + started_at TIMESTAMPTZ NOT NULL DEFAULT now(), + completed_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS idx_retention_log_run_id + ON retention_enforcement_log (run_id); + +CREATE INDEX IF NOT EXISTS idx_retention_log_policy_id + ON retention_enforcement_log (policy_id, started_at DESC); + +CREATE INDEX IF NOT EXISTS idx_retention_log_started_at + ON retention_enforcement_log (started_at DESC); + +-- ── retention_legal_holds ───────────────────────────────────────────────────── +-- Persistent legal holds that survive application restarts. + +CREATE TABLE IF NOT EXISTS retention_legal_holds ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + scope TEXT NOT NULL CHECK (scope IN ('user', 'record')), + target_id TEXT NOT NULL, + source TEXT, + reason TEXT NOT NULL, + placed_at TIMESTAMPTZ NOT NULL DEFAULT now(), + released_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS idx_retention_holds_active + ON retention_legal_holds (target_id, source) WHERE released_at IS NULL; + +CREATE INDEX IF NOT EXISTS idx_retention_holds_scope + ON retention_legal_holds (scope, target_id) WHERE released_at IS NULL; + +-- ── Seed default policies ───────────────────────────────────────────────────── +-- Insert the application defaults so they are visible in the database. The +-- application's DataRetentionService merges these with any runtime overrides. + +INSERT INTO retention_policies (id, source, description, retention_days, action, enabled, batch_size) +VALUES + ('deactivated-account-logs', 'activity_logs', + 'Activity logs of deactivated accounts', 90, 'delete', true, 1000), + ('notification-history', 'notifications', + 'Delivered notification history', 180, 'delete', true, 1000), + ('billing-records', 'billing_records', + 'Invoices and payment records (kept for tax and accounting)', 2555, 'anonymize', true, 500), + ('audit-logs', 'audit_logs', + 'Security and compliance audit trail', 2555, 'archive', true, 500) +ON CONFLICT (id) DO NOTHING; + +-- ── Analyse tables ─────────────────────────────────────────────────────────── +ANALYZE retention_policies; +ANALYZE retention_enforcement_log; +ANALYZE retention_legal_holds; + +-- @down +DROP TABLE IF EXISTS retention_legal_holds; +DROP TABLE IF EXISTS retention_enforcement_log; +DROP TABLE IF EXISTS retention_policies; diff --git a/docs/DATA_RETENTION.md b/docs/DATA_RETENTION.md new file mode 100644 index 00000000..c82c048f --- /dev/null +++ b/docs/DATA_RETENTION.md @@ -0,0 +1,116 @@ +# Data Retention and Purge Jobs + +## Overview + +SubTrackr enforces data retention policies to prevent unbounded data growth. +The system defines how long each category of data may be kept and automatically +purges, anonymizes, or archives expired records on a schedule. + +## Architecture + +``` +┌─────────────────────────┐ +│ PurgeJobScheduler │ Coordinates everything +│ (purgeJobScheduler.ts) │ +└───────────┬─────────────┘ + │ + ┌─────────┴─────────┐ + │ │ + ▼ ▼ +┌─────────────┐ ┌──────────────────┐ +│ DataRetention│ │ RetentionEnforce │ +│ Service │ │ mentJob │ +│ (policies, │ │ (periodic runner)│ +│ holds, │ └──────────────────┘ +│ enforce) │ +└──────┬───────┘ + │ + ▼ +┌─────────────────────┐ +│ PostgresRetention │ SQL-backed store +│ Store │ (findExpired, delete, +│ (per data source) │ anonymize, archive) +└─────────────────────┘ +``` + +## Default Retention Policies + +| Policy ID | Source | Retention | Action | Description | +|---|---|---|---|---| +| `deactivated-account-logs` | `activity_logs` | 90 days | delete | Activity logs of deactivated accounts | +| `notification-history` | `notifications` | 180 days | delete | Delivered notification history | +| `billing-records` | `billing_records` | 7 years | anonymize | Invoices and payment records (tax/accounting) | +| `audit-logs` | `audit_logs` | 7 years | archive | Security and compliance audit trail | + +### Actions + +- **delete** — Records are permanently deleted from the database. +- **anonymize** — PII columns (email, name, phone, address) are set to NULL. +- **archive** — Records are marked with `archived_at = now()` for later migration to cold storage. + +## Legal Holds + +Legal holds prevent records from being purged regardless of their age. Holds +can be scoped to: + +- **user** — all records for a given user ID are protected +- **record** — a specific record ID is protected + +Holds can optionally be restricted to a single data source. The +`PostgresRetentionStore` checks the `retention_legal_holds` table before +returning expired records, so held records are never acted upon. + +## Database Tables (migration 010) + +### `retention_policies` +Persistent policy definitions. Loaded on startup and merged with +`DEFAULT_RETENTION_POLICIES`. + +### `retention_enforcement_log` +Append-only audit trail. One row per policy per enforcement run, recording +cutoff timestamp, records found, legal-hold skips, records processed, and +any errors. + +### `retention_legal_holds` +Persistent legal holds that survive application restarts. + +## Scheduling + +The `PurgeJobScheduler` starts a `RetentionEnforcementJob` that runs +enforcement every 24 hours by default. The interval is configurable: + +```typescript +const scheduler = new PurgeJobScheduler(pool, { + intervalMs: 12 * 60 * 60 * 1000, // every 12 hours +}); +await scheduler.start(); +``` + +Manual enforcement is also available: + +```typescript +const report = await scheduler.runNow(); +``` + +## REST API + +The `RetentionController` exposes endpoints for policy management, legal +holds, and manual enforcement: + +| Method | Path | Description | +|---|---|---| +| GET | `/gdpr/retention/policies` | List all policies | +| PUT | `/gdpr/retention/policies/:id` | Create or update a policy | +| DELETE | `/gdpr/retention/policies/:id` | Remove a policy | +| POST | `/gdpr/retention/enforce` | Trigger enforcement (optional `dryRun`, `policyIds`) | +| GET | `/gdpr/retention/reports` | List enforcement reports | +| POST | `/gdpr/retention/holds` | Place a legal hold | +| POST | `/gdpr/retention/holds/:id/release` | Release a legal hold | +| GET | `/gdpr/retention/holds` | List active legal holds | + +## Configuration + +| Env Var | Default | Description | +|---|---|---| +| `RETENTION_INTERVAL_MS` | `86400000` | Enforcement interval (ms) | +| `RETENTION_BATCH_SIZE` | `1000` | Max records per policy per run |