Feature/soft delete retention purge

docs/multi-tenancy.md

@@ -22,6 +22,49 @@ org-scoped endpoint must uphold.
    `req.params`. Client-supplied `orgId` values in query strings or request bodies are
    never trusted for data access decisions.
 
+### Org-scoped retention and hard purge
+
+Retention purge jobs are also scoped to the organization whose ID appears in the job
+payload. A per-org retention pass only hard-deletes soft-deleted vaults and related
+milestones for that organization.
+
+Retention configuration supports the following environment variables:
+
+- `RETENTION_PURGE_AGE_MS` — minimum age before a soft-deleted vault is eligible for hard purge.
+  If unset, the per-org retention window is consulted (see below).
+- `RETENTION_PURGE_INTERVAL_MS` — frequency at which the retention scheduler enqueues per-org jobs.
+- `RETENTION_PURGE_BATCH_SIZE` — maximum number of vaults deleted per org on each retention run.
+
+Every retention purge also writes an audit log entry scoped to the target organization with a
+summary of deleted vault and milestone counts.
+
+### Per-org retention window
+
+Each organization can override the global retention window by setting
+`retention_purge_age_ms` in its `metadata` JSONB column:
+
+```sql
+UPDATE organizations
+SET metadata = jsonb_set(
+  COALESCE(metadata, '{}'::jsonb),
+  '{retention_purge_age_ms}',
+  '604800000'   -- 7 days in milliseconds
+)
+WHERE id = '<org-uuid>';
+```
+
+Resolution order (first match wins):
+
+1. Explicit `retentionAgeMs` argument passed to `purgeSoftDeletedVaults` (testing).
+2. `RETENTION_PURGE_AGE_MS` environment variable (global override).
+3. `organizations.metadata.retention_purge_age_ms` (per-org override).
+4. Built-in default of 30 days (`2_592_000_000` ms).
+
+If the per-org value is missing, unset, or not a valid non-negative number, the system
+falls through to the next level. This allows a cluster-wide default via `RETENTION_PURGE_AGE_MS`
+while letting individual organizations opt into a shorter (or longer) window through their
+metadata.
+
 5. **No cross-org leakage** — pagination, sorting, and filtering are applied *after* the
    org-scope filter. A filter that matches zero records in the target org returns an empty
    result set, not records from another org.

src/jobs/handlers.ts

@@ -13,6 +13,8 @@ import {
   processDeferredReminders,
 } from '../services/vaultExpiry.service.js'
 import { cleanupExpiredSessions } from '../services/session.js'
+import { purgeSoftDeletedVaults } from '../services/retention.js'
+import { createAuditLog } from '../lib/audit-logs.js'
 import { relayOutboxBatch } from '../services/outboxRelay.js'
 import { runReindexBatches } from '../services/evidenceReindex.js'
 import { renderOrgAnalyticsSnapshot } from '../services/analytics.service.js'
@@ -185,6 +187,25 @@ export const createDefaultJobHandlers = (
       `deleted=${deleted} batchSize=${batchSize} attempt=${context.attempt}`,
     )
   },
+  'retention.purge': async (payload, context) => {
+    const batchSize = payload.batchSize ?? 1000
+    const result = await purgeSoftDeletedVaults(payload.organizationId, batchSize)
+    const message = `organization=${payload.organizationId} deletedVaults=${result.deletedVaults} deletedMilestones=${result.deletedMilestones} batchSize=${batchSize}`
+    logJob('retention.purge', `${message} attempt=${context.attempt}`)
+
+    await createAuditLog({
+      actor_user_id: 'system',
+      organization_id: payload.organizationId,
+      action: 'retention.purge',
+      target_type: 'organization',
+      target_id: payload.organizationId,
+      metadata: {
+        deleted_vaults: result.deletedVaults,
+        deleted_milestones: result.deletedMilestones,
+        batch_size: batchSize,
+      },
+    })
+  },
   'outbox.relay': async (payload, context) => {
     const count = await relayOutboxBatch()
     logJob(

src/jobs/system.ts

@@ -8,6 +8,7 @@ import {
 } from './queue.js'
 import { type EnqueueOptions, type JobPayloadByType, type JobType } from './types.js'
 import { recoverPendingExportJobs } from '../services/exportQueue.js'
+import { listOrganizations } from '../services/organization.js'
 import {
   createNotificationService,
   type NotificationService,
@@ -197,6 +198,7 @@ export class BackgroundJobSystem {
     this.queue.registerHandler('analytics.report.generate', handlers['analytics.report.generate'])
     this.queue.registerHandler('export.generate', handlers['export.generate'])
     this.queue.registerHandler('sessions.cleanup', handlers['sessions.cleanup'])
+    this.queue.registerHandler('retention.purge', handlers['retention.purge'])
     this.queue.registerHandler('outbox.relay', handlers['outbox.relay'])
     this.queue.registerHandler('embeddings.reindex', handlers['embeddings.reindex'])
     this.queue.registerHandler('saved-search.evaluate', handlers['saved-search.evaluate'])
@@ -339,6 +341,36 @@ export class BackgroundJobSystem {
       },
     })
 
+    const retentionPurgeIntervalMs = parsePositiveInteger(
+      process.env.RETENTION_PURGE_INTERVAL_MS,
+      86_400_000,
+    )
+    const retentionPurgeBatchSize = parsePositiveInteger(
+      process.env.RETENTION_PURGE_BATCH_SIZE,
+      500,
+    )
+
+    this.schedulerRegistry.registerJob({
+      name: 'retention.purge',
+      intervalMs: retentionPurgeIntervalMs,
+      immediate: true,
+      initialDelayMs: 20_000,
+      execute: async () => {
+        try {
+          const organizations = await listOrganizations()
+          for (const org of organizations) {
+            this.enqueue('retention.purge', {
+              organizationId: org.id,
+              batchSize: retentionPurgeBatchSize,
+            })
+          }
+        } catch (error) {
+          const message = error instanceof Error ? error.message : String(error)
+          console.error(`[jobs:retention.purge] failed to enqueue jobs: ${message}`)
+        }
+      },
+    })
+
     this.schedulerRegistry.registerJob({
       name: 'outbox.relay',
       intervalMs: outboxRelayIntervalMs,

src/jobs/types.ts

@@ -10,6 +10,7 @@ export const JOB_TYPES = [
   'export.generate',
   'vault.reconcile',
   'sessions.cleanup',
+  'retention.purge',
   'outbox.relay',
   'embeddings.reindex',
   'saved-search.evaluate',
@@ -72,6 +73,11 @@ export interface SessionsCleanupJobPayload {
   batchSize?: number
 }
 
+export interface RetentionPurgeJobPayload {
+  organizationId: string
+  batchSize?: number
+}
+
 export interface OutboxRelayJobPayload {
   batchSize?: number
 }
@@ -84,7 +90,6 @@ export interface EmbeddingsReindexJobPayload {
 export interface SavedSearchEvaluateJobPayload {
   searchId?: string
 }
-
 export interface JobPayloadByType {
   'notification.send': NotificationJobPayload
   'deadline.check': DeadlineCheckJobPayload
@@ -97,6 +102,7 @@ export interface JobPayloadByType {
   'export.generate': ExportGenerateJobPayload
   'vault.reconcile': VaultReconcileJobPayload
   'sessions.cleanup': SessionsCleanupJobPayload
+  'retention.purge': RetentionPurgeJobPayload
   'outbox.relay': OutboxRelayJobPayload
   'embeddings.reindex': EmbeddingsReindexJobPayload
   'saved-search.evaluate': SavedSearchEvaluateJobPayload
@@ -117,7 +123,7 @@ export interface EnqueueOptions {
   maxAttempts?: number
 }
 
-const isRecord = (value: unknown): value is Record<string, unknown> => {
+export const isRecord = (value: unknown): value is Record<string, unknown> => {
   return typeof value === 'object' && value !== null
 }
 
@@ -193,6 +199,11 @@ export const isPayloadForJobType = (
       )
     case 'sessions.cleanup':
       return payload.batchSize === undefined || (typeof payload.batchSize === 'number' && payload.batchSize > 0)
+    case 'retention.purge':
+      return (
+        isNonEmptyString(payload.organizationId) &&
+        (payload.batchSize === undefined || (typeof payload.batchSize === 'number' && payload.batchSize > 0))
+      )
     case 'outbox.relay':
       return true
     case 'embeddings.reindex':

src/lib/validation.ts

@@ -84,6 +84,11 @@ export const analyticsRecomputePayloadSchema = z.object({
   reason: z.string().optional(),
 })
 
+export const retentionPurgePayloadSchema = z.object({
+  organizationId: nonEmptyString,
+  batchSize: z.number().int().min(1).optional(),
+})
+
 export const enqueueJobSchema = z.discriminatedUnion('type', [
   z.object({
     type: z.literal('notification.send'),
@@ -109,6 +114,12 @@ export const enqueueJobSchema = z.discriminatedUnion('type', [
     maxAttempts: z.number().int().min(1).max(10).optional(),
     delayMs: z.number().int().min(0).max(60000).optional(),
   }),
+  z.object({
+    type: z.literal('retention.purge'),
+    payload: retentionPurgePayloadSchema,
+    maxAttempts: z.number().int().min(1).max(10).optional(),
+    delayMs: z.number().int().min(0).max(60000).optional(),
+  }),
 ])
 
 export interface ValidationErrorField {

src/routes/jobs.ts

@@ -1,11 +1,13 @@
 import { Router, type RequestHandler } from 'express'
-import { z } from 'zod'
 import { UserRole } from '../types/user.js'
 import type { BackgroundJobSystem } from '../jobs/system.js'
 import {
   type EnqueueOptions,
   type JobPayloadByType,
   type JobType,
+  isJobType,
+  isPayloadForJobType,
+  isRecord,
 } from '../jobs/types.js'
 import { parseEnqueueOptions } from '../jobs/enqueueOptions.js'
 import { authenticate, authorize } from '../middleware/auth.js'
@@ -48,6 +50,8 @@ const enqueueTypedJob = (
       return jobSystem.enqueue(type, payload, options)
     case 'analytics.recompute':
       return jobSystem.enqueue(type, payload, options)
+    case 'retention.purge':
+      return jobSystem.enqueue(type, payload, options)
     default:
       throw new Error('Unsupported job type')
   }
@@ -211,7 +215,7 @@ export const createJobsRouter = (jobSystem: BackgroundJobSystem, options: JobsRo
         if (!isJobType(type)) {
           res.status(400).json({
             error:
-              'Invalid or missing job type. Supported types: notification.send, deadline.check, oracle.call, analytics.recompute',
+              'Invalid or missing job type. Supported types: notification.send, deadline.check, oracle.call, analytics.recompute, retention.purge',
           })
           return
         }

src/services/retention.ts

@@ -0,0 +1,99 @@
+import db from '../db/index.js'
+import type { Knex } from 'knex'
+
+export type PurgeSoftDeletedVaultsResult = {
+  deletedVaults: number
+  deletedMilestones: number
+}
+
+const DEFAULT_RETENTION_AGE_MS = 30 * 24 * 60 * 60 * 1000 // 30 days
+
+const validateOrganizationId = (organizationId: string): void => {
+  if (typeof organizationId !== 'string' || organizationId.trim() === '') {
+    throw new Error('organizationId is required')
+  }
+}
+
+const getGlobalRetentionAgeMs = (): number | undefined => {
+  const rawValue = process.env.RETENTION_PURGE_AGE_MS
+  if (rawValue === undefined || rawValue.trim() === '') {
+    return undefined
+  }
+
+  const parsed = Number(rawValue)
+  if (!Number.isFinite(parsed) || parsed < 0) {
+    throw new Error('RETENTION_PURGE_AGE_MS must be a non-negative integer')
+  }
+
+  return Math.floor(parsed)
+}
+
+const resolvePerOrgRetentionAgeMs = async (
+  organizationId: string,
+  knexInstance: typeof db | Knex,
+): Promise<number> => {
+  const org = await knexInstance('organizations')
+    .select('metadata')
+    .where('id', organizationId)
+    .first()
+
+  const orgRetention = org?.metadata?.retention_purge_age_ms
+  if (typeof orgRetention === 'number' && Number.isFinite(orgRetention) && orgRetention >= 0) {
+    return Math.floor(orgRetention)
+  }
+
+  return DEFAULT_RETENTION_AGE_MS
+}
+
+export const purgeSoftDeletedVaults = async (
+  organizationId: string,
+  batchSize = 500,
+  knexInstance: typeof db | Knex = db,
+  retentionAgeMs?: number,
+): Promise<PurgeSoftDeletedVaultsResult> => {
+  validateOrganizationId(organizationId)
+
+  if (!Number.isFinite(batchSize) || batchSize <= 0) {
+    throw new Error('batchSize must be a positive integer')
+  }
+
+  const effectiveRetentionAgeMs =
+    retentionAgeMs !== undefined
+      ? retentionAgeMs
+      : getGlobalRetentionAgeMs() ?? (await resolvePerOrgRetentionAgeMs(organizationId, knexInstance))
+
+  return await knexInstance.transaction(async (trx) => {
+    const cutoffDate =
+      effectiveRetentionAgeMs > 0
+        ? new Date(Date.now() - effectiveRetentionAgeMs)
+        : undefined
+
+    let query = trx('vaults')
+      .where({ organization_id: organizationId })
+      .whereNotNull('deleted_at')
+
+    if (cutoffDate) {
+      query = query.where('deleted_at', '<=', cutoffDate)
+    }
+
+    const vaultIds = await query.limit(batchSize).pluck('id')
+
+    if (vaultIds.length === 0) {
+      return { deletedVaults: 0, deletedMilestones: 0 }
+    }
+
+    const milestonesCountRow = await trx('milestones')
+      .whereIn('vault_id', vaultIds)
+      .count<{ total: string }>('id as total')
+      .first()
+
+    const deletedMilestones = Number(milestonesCountRow?.total ?? 0)
+
+    const deletedVaults = await trx('vaults').whereIn('id', vaultIds).delete()
+
+    return {
+      deletedVaults,
+      deletedMilestones,
+    }
+  })
+}