.changeset/finite-analytics-retention.md

@@ -0,0 +1,7 @@
+---
+'@goto/web': minor
+'@goto/cli': minor
+'@goto/worker': minor
+---
+
+Replace analytics enabled/retention-days settings with finite calendar-month retention. Apply live retention changes through the worker service's jittered hourly OpenWorkflow schedule and purge analytics data when retention is zero. Keep Piscina ingestion-only, serialize SQLite migrations, and fence queued writes across collection changes.

.github/workflows/worker.yml

@@ -11,6 +11,7 @@ on:
       - .woodpecker/**
       - apps/worker/**
       - packages/config/**
+      - packages/db/**
       - tools/**
   pull_request:
     branches: [main]
@@ -21,6 +22,7 @@ on:
       - .woodpecker/**
       - apps/worker/**
       - packages/config/**
+      - packages/db/**
       - tools/**
 
 permissions:
@@ -53,6 +55,11 @@ jobs:
         run: pnpm --filter @goto/worker check
       - name: Lint worker
         run: pnpm --filter @goto/worker lint
+      - name: Check database dependency
+        run: |
+          pnpm --filter @goto/db check
+          pnpm --filter @goto/db lint
+          pnpm --filter @goto/db test:coverage
       - name: Unit tests
         run: pnpm --filter @goto/worker test:coverage
       - name: Compiled service smoke test

.woodpecker/worker.yml

@@ -13,6 +13,7 @@ when:
         - .woodpecker/**
         - apps/worker/**
         - packages/config/**
+        - packages/db/**
         - tools/**
   - event: pull_request
     path:
@@ -23,6 +24,7 @@ when:
         - .woodpecker/**
         - apps/worker/**
         - packages/config/**
+        - packages/db/**
         - tools/**
   - event: tag
     ref: 'refs/tags/@goto/worker@*'
@@ -48,6 +50,8 @@ steps:
     commands:
       - npm install --global corepack@0.36.0
       - corepack enable
+      - pnpm --filter @goto/db check
+      - pnpm --filter @goto/db lint
       - pnpm --filter @goto/worker check
       - pnpm --filter @goto/worker lint
 
@@ -58,6 +62,7 @@ steps:
     commands:
       - npm install --global corepack@0.36.0
       - corepack enable
+      - pnpm --filter @goto/db test:coverage
       - pnpm --filter @goto/worker test:coverage
       - pnpm --filter @goto/worker test:smoke
 

apps/web/docker-compose.dev.yml

@@ -63,6 +63,7 @@ services:
       - ./data:/m/apps/worker/data
       - ../worker/src:/m/apps/worker/src:ro
       - ../../packages/config/src:/m/packages/config/src:ro
+      - ../../packages/db/src:/m/packages/db/src:ro
     restart: unless-stopped
     healthcheck:
       test:

apps/web/server/adapter.ts

@@ -5,7 +5,7 @@ import { DevelopmentServer } from './development.ts';
 import { configureVite } from './plugins.ts';
 
 const entry = fileURLToPath(new URL('../src/lib/server/entry.ts', import.meta.url));
-const applicationFiles = ['application.ts', 'visits/recorder.ts'].map((file) =>
+const applicationFiles = ['application.ts', 'visits/recorder.ts', 'visits/pool.ts'].map((file) =>
 	fileURLToPath(new URL(`../src/lib/server/${file}`, import.meta.url))
 );
 

apps/web/src/lib/server/application.test.ts

@@ -25,6 +25,7 @@ let reload: ReloadOptions;
 const stop = vi.fn();
 const stopRecorder = vi.fn();
 const recordVisit = vi.fn();
+const setEnabled = vi.fn();
 
 function prepared(
 	destination = 'https://example.org/first',
@@ -45,7 +46,7 @@ beforeEach(() => {
 	vi.clearAllMocks();
 	vi.spyOn(console, 'warn').mockImplementation(() => {});
 	vi.mocked(loadConfigFile).mockResolvedValue(prepared());
-	vi.mocked(createVisitRecorder).mockReturnValue({ recordVisit, stop: stopRecorder });
+	vi.mocked(createVisitRecorder).mockReturnValue({ recordVisit, setEnabled, stop: stopRecorder });
 	vi.mocked(startConfigReloading).mockImplementation(async (options) => {
 		reload = options as ReloadOptions;
 		return stop;
@@ -68,7 +69,7 @@ test('uses a fixed path/environment and stops monitoring with the application',
 	expect(application.currentSnapshot().configuration.auth).not.toHaveProperty('secret');
 	await application.close?.();
 	expect(stop).toHaveBeenCalledOnce();
-	expect(createVisitRecorder).toHaveBeenCalledExactlyOnceWith('/deployment/data/goto.sqlite');
+	expect(createVisitRecorder).toHaveBeenCalledExactlyOnceWith('/deployment/data/goto.sqlite', true);
 	expect(application.recordVisit).toBe(recordVisit);
 	expect(stopRecorder).toHaveBeenCalledOnce();
 });
@@ -98,7 +99,7 @@ test('publishes valid changes, ignores equivalent data, and retains last good da
 test.each([
 	{ site: { url: 'https://different-admin.example.test' } },
 	{ auth: { credentials: { username: 'changed', password: 'private-changed-password' } } },
-	{ database: { path: './data/different.sqlite' } }
+	{ database: { path: './data/different.sqlite' }, analytics: { retention_months: 0 } }
 ])('rejects the whole candidate when startup settings change: %j', async (changes) => {
 	const application = await createApplication(options);
 	const initial = application.currentSnapshot();
@@ -106,12 +107,29 @@ test.each([
 	expect(reload.onChange(prepared('https://example.org/blocked', changes), prepared())).toBe(false);
 
 	expect(application.currentSnapshot()).toBe(initial);
+	expect(setEnabled).not.toHaveBeenCalled();
 	expect(console.warn).toHaveBeenCalledExactlyOnceWith(
 		'Configuration change requires restart; keeping current configuration'
 	);
 	await application.close?.();
 });
 
+test('disables live collection without replacing the recorder', async () => {
+	const application = await createApplication(options);
+	const next = prepared('https://example.org/first', { analytics: { retention_months: 0 } });
+	expect(reload.onChange(next, prepared())).toBe(true);
+	expect(setEnabled).toHaveBeenCalledExactlyOnceWith(false);
+	expect(application.currentSnapshot().configuration.analytics.retention_months).toBe(0);
+	expect(createVisitRecorder).toHaveBeenCalledOnce();
+	await application.close?.();
+});
+
+test('closes the recorder if configuration monitoring cannot start', async () => {
+	vi.mocked(startConfigReloading).mockRejectedValueOnce(new Error('Monitor startup failed'));
+	await expect(createApplication(options)).rejects.toThrow('Monitor startup failed');
+	expect(stopRecorder).toHaveBeenCalledOnce();
+});
+
 test('invalid startup fails before registering a monitor', async () => {
 	vi.mocked(loadConfigFile).mockRejectedValue(new ConfigError());
 

apps/web/src/lib/server/application.ts

@@ -42,28 +42,58 @@ export async function createApplication(options: ConfigOptions = {}): Promise<Ap
 	};
 	const path = resolveConfigPath(captured);
 	const initial = await loadConfigFile(path, captured);
-	const startup = structuredClone(startupSettings(initial));
+	const initialSettings = startupSettings(initial);
+	const startup = structuredClone(initialSettings);
 	let snapshot = prepareSnapshot(initial);
 	const authentication = createAuthentication(snapshot.configuration);
-	const stopReloading = await startConfigReloading({
+	const databasePath = resolve(captured.cwd, startup.databasePath);
+	const collectionEnabled = snapshot.configuration.analytics.retention_months !== 0;
+	const recorder = createVisitRecorder(databasePath, collectionEnabled);
+
+	function loadConfiguration(path: string): Promise<PreparedConfiguration> {
+		return loadConfigFile(path, captured);
+	}
+
+	function applyConfiguration(
+		next: PreparedConfiguration,
+		previous: PreparedConfiguration
+	): boolean {
+		const nextSettings = startupSettings(next);
+		if (!isDeepStrictEqual(nextSettings, startup)) {
+			reportReloadIssue('Configuration change requires restart; keeping current configuration');
+			return false;
+		}
+		if (!isDeepStrictEqual(next, previous)) {
+			const candidate = prepareSnapshot(next);
+			const collectionEnabled = candidate.configuration.analytics.retention_months !== 0;
+			recorder.setEnabled(collectionEnabled);
+			snapshot = candidate;
+		}
+		return true;
+	}
+
+	function reportConfigurationFailure(): void {
+		reportReloadIssue('Invalid configuration update; keeping current configuration');
+	}
+
+	const reloadOptions = {
 		path,
 		initial,
-		load: (path) => loadConfigFile(path, captured),
-		onChange(next, previous) {
-			if (!isDeepStrictEqual(startupSettings(next), startup)) {
-				reportReloadIssue('Configuration change requires restart; keeping current configuration');
-				return false;
-			}
-			if (!isDeepStrictEqual(next, previous)) snapshot = prepareSnapshot(next);
-			return true;
-		},
-		onError() {
-			reportReloadIssue('Invalid configuration update; keeping current configuration');
-		}
-	});
-	const recorder = createVisitRecorder(resolve(captured.cwd, startup.databasePath));
+		load: loadConfiguration,
+		onChange: applyConfiguration,
+		onError: reportConfigurationFailure
+	};
+	let stopReloading: () => Promise<void>;
+	try {
+		stopReloading = await startConfigReloading(reloadOptions);
+	} catch (error) {
+		await recorder.stop();
+		throw error;
+	}
 	const application: Application = {
-		currentSnapshot: () => snapshot,
+		currentSnapshot() {
+			return snapshot;
+		},
 		authentication,
 		recordVisit: recorder.recordVisit,
 		async close() {

apps/web/src/lib/server/dispatch.test.ts

@@ -145,7 +145,7 @@ test('does not record HEAD, fallback redirects, or visits when analytics is disa
 		...snapshot,
 		configuration: {
 			...snapshot.configuration,
-			analytics: { ...snapshot.configuration.analytics, enabled: false }
+			analytics: { ...snapshot.configuration.analytics, retention_months: 0 as const }
 		}
 	};
 	expect(

apps/web/src/lib/server/dispatch.ts

@@ -25,7 +25,7 @@ export function handleIngress(
 			request.url!,
 			request.method ?? 'GET'
 		);
-		if (snapshot.configuration.analytics.enabled) {
+		if (snapshot.configuration.analytics.retention_months !== 0) {
 			recordRedirectVisit(request, redirect.visit, application.recordVisit);
 		}
 		return redirect.response;

apps/web/src/lib/server/visits/pool.ts

@@ -0,0 +1,25 @@
+import { Piscina } from 'piscina';
+
+export const MAX_OUTSTANDING = 1000;
+
+export function createVisitPool(databasePath: string): Piscina {
+	return new Piscina({
+		filename: import.meta.resolve('@goto/db/visit-worker'),
+		name: 'recordVisit',
+		workerData: { databasePath },
+		minThreads: 1,
+		maxThreads: 1,
+		concurrentTasksPerWorker: 1,
+		stricterFIFO: true,
+		maxQueue: MAX_OUTSTANDING,
+		closeTimeout: 5000
+	});
+}
+
+export async function closeVisitPool(pool: Piscina): Promise<void> {
+	const databaseClosed = pool.run(null, { name: 'close' });
+	const poolClosed = pool.close();
+	const results = await Promise.allSettled([databaseClosed, poolClosed]);
+	const failed = results.some((result) => result.status === 'rejected');
+	if (failed) throw new Error('Visit pool shutdown failed');
+}

apps/web/src/lib/server/visits/recorder.test.ts

@@ -44,8 +44,8 @@ afterEach(() => {
 	vi.restoreAllMocks();
 });
 
-test('bounds all unsettled visits and releases capacity after a failed insert', async () => {
-	const recorder = createVisitRecorder('/data/goto.sqlite');
+test('bounds unsettled visits and releases capacity after a failed insert', async () => {
+	const recorder = createVisitRecorder('/data/goto.sqlite', true);
 	recorder.recordVisit(visit);
 	expect(writes).toHaveLength(0);
 	initialized.resolve();
@@ -63,8 +63,8 @@ test('bounds all unsettled visits and releases capacity after a failed insert',
 	await recorder.stop();
 });
 
-test('drops oversized events without submitting or changing the payload', async () => {
-	const recorder = createVisitRecorder('/data/goto.sqlite');
+test('drops oversized events without changing the payload', async () => {
+	const recorder = createVisitRecorder('/data/goto.sqlite', true);
 	initialized.resolve();
 	await initialized.promise;
 	const oversized = { ...visit, user_agent: 'x'.repeat(32 * 1024) };
@@ -74,17 +74,41 @@ test('drops oversized events without submitting or changing the payload', async
 	await recorder.stop();
 });
 
-test('stops admission and waits for both cooperative database close and pool closure', async () => {
+test('disabled collection starts no worker and can later enable and pause admission', async () => {
+	const recorder = createVisitRecorder('/data/goto.sqlite', false);
+	recorder.recordVisit(visit);
+	expect(Piscina).not.toHaveBeenCalled();
+	recorder.setEnabled(true);
+	initialized.resolve();
+	await initialized.promise;
+	recorder.recordVisit(visit);
+	expect(writes).toHaveLength(1);
+	recorder.setEnabled(false);
+	recorder.recordVisit(visit);
+	expect(writes).toHaveLength(1);
+	recorder.setEnabled(true);
+	recorder.recordVisit(visit);
+	expect(writes).toHaveLength(2);
+	expect(Piscina).toHaveBeenCalledOnce();
+	await recorder.stop();
+});
+
+test('stops admission and waits for database and pool closure', async () => {
 	const closed = Promise.withResolvers<void>();
 	const poolClosed = Promise.withResolvers<void>();
-	const recorder = createVisitRecorder('/data/goto.sqlite');
+	const recorder = createVisitRecorder('/data/goto.sqlite', true);
 	initialized.resolve();
 	await initialized.promise;
 	pool.run.mockImplementation(() => closed.promise);
 	pool.close.mockReturnValue(poolClosed.promise);
 	const stop = recorder.stop();
 	expect(Piscina).toHaveBeenCalledWith(
-		expect.objectContaining({ maxThreads: 1, concurrentTasksPerWorker: 1, stricterFIFO: true })
+		expect.objectContaining({
+			maxThreads: 1,
+			concurrentTasksPerWorker: 1,
+			stricterFIFO: true,
+			maxQueue: 1000
+		})
 	);
 	expect(recorder.stop()).toBe(stop);
 	recorder.recordVisit(visit);
@@ -101,8 +125,8 @@ test('stops admission and waits for both cooperative database close and pool clo
 	await stop;
 });
 
-test('pool initialization failures disable capture without propagating private errors', async () => {
-	const recorder = createVisitRecorder('/data/goto.sqlite');
+test('initialization failures disable capture without propagating private errors', async () => {
+	const recorder = createVisitRecorder('/data/goto.sqlite', true);
 	initialized.reject(new Error('private database path'));
 	await Promise.resolve();
 	recorder.recordVisit(visit);

apps/web/src/lib/server/visits/recorder.ts

@@ -1,15 +1,19 @@
-import { Piscina } from 'piscina';
+import type { Piscina } from 'piscina';
 import type { NewVisit } from '@goto/db';
+import { closeVisitPool, createVisitPool, MAX_OUTSTANDING } from './pool.js';
 
-const MAX_OUTSTANDING = 1000;
 const MAX_VISIT_BYTES = 32 * 1024;
-const DRAIN_TIMEOUT = 5000;
 
 export interface VisitRecorder {
 	recordVisit(visit: NewVisit): void;
+	setEnabled(enabled: boolean): void;
 	stop(): Promise<void>;
 }
 
+export function createVisitRecorder(databasePath: string, enabled: boolean): VisitRecorder {
+	return new Recorder(databasePath, enabled);
+}
+
 function visitBytes(visit: NewVisit): number {
 	return [visit.domain, visit.path, visit.ip, visit.user_agent, visit.accept_language].reduce(
 		(bytes, value) => bytes + Buffer.byteLength(value ?? ''),
@@ -17,86 +21,93 @@ function visitBytes(visit: NewVisit): number {
 	);
 }
 
-export function createVisitRecorder(databasePath: string): VisitRecorder {
-	let reported = false;
-	function report(): void {
-		if (reported) return;
-		reported = true;
+class Recorder implements VisitRecorder {
+	private readonly databasePath: string;
+	private enabled: boolean;
+	private reported = false;
+	private pool: Piscina | undefined;
+	private ready = false;
+	private outstanding = 0;
+	private stopping: Promise<void> | undefined;
+
+	constructor(databasePath: string, enabled: boolean) {
+		this.databasePath = databasePath;
+		this.enabled = enabled;
+		if (enabled) this.startPool();
+	}
+
+	readonly recordVisit = (visit: NewVisit): void => {
+		const pool = this.pool;
+		if (!this.enabled || !this.ready || !pool || this.stopping) return;
+		const bytes = visitBytes(visit);
+		if (this.outstanding >= MAX_OUTSTANDING || bytes > MAX_VISIT_BYTES) {
+			this.report();
+			return;
+		}
+		this.submitVisit(pool, visit);
+	};
+
+	readonly setEnabled = (enabled: boolean): void => {
+		if (this.stopping) return;
+		this.enabled = enabled;
+		if (enabled) this.startPool();
+	};
+
+	readonly stop = (): Promise<void> => {
+		if (!this.stopping) {
+			this.stopping = this.pool ? closeVisitPool(this.pool).catch(this.report) : Promise.resolve();
+		}
+		return this.stopping;
+	};
+
+	private submitVisit(pool: Piscina, visit: NewVisit): void {
+		this.outstanding += 1;
+		try {
+			const recorded = pool.run(visit);
+			void recorded.then(this.onRecorded, this.onRecordFailure);
+		} catch {
+			this.onRecordFailure();
+		}
+	}
+
+	private startPool(): void {
+		if (this.pool || this.stopping) return;
+		try {
+			const pool = createVisitPool(this.databasePath);
+			this.pool = pool;
+			pool.on('error', this.onPoolFailure);
+			const initialized = pool.run(null, { name: 'ready' });
+			void initialized.then(this.onReady, this.onPoolFailure);
+		} catch {
+			this.onPoolFailure();
+		}
+	}
+
+	private readonly onRecorded = (): void => {
+		this.outstanding -= 1;
+	};
+
+	private readonly onRecordFailure = (): void => {
+		this.onRecorded();
+		this.report();
+	};
+
+	private readonly onReady = (): void => {
+		if (!this.stopping) this.ready = true;
+	};
+
+	private readonly onPoolFailure = (): void => {
+		this.report();
+		void this.stop();
+	};
+
+	private readonly report = (): void => {
+		if (this.reported) return;
+		this.reported = true;
 		try {
 			console.warn('Visit tracking unavailable; some visits may not be recorded');
 		} catch {
 			return;
 		}
-	}
-
-	let pool: Piscina;
-	try {
-		pool = new Piscina({
-			filename: import.meta.resolve('@goto/db/visit-worker'),
-			name: 'recordVisit',
-			workerData: { databasePath },
-			minThreads: 1,
-			maxThreads: 1,
-			concurrentTasksPerWorker: 1,
-			stricterFIFO: true,
-			maxQueue: MAX_OUTSTANDING,
-			closeTimeout: DRAIN_TIMEOUT
-		});
-	} catch {
-		report();
-		return { recordVisit() {}, async stop() {} };
-	}
-
-	let accepting = false;
-	let outstanding = 0;
-	let stopping: Promise<void> | undefined;
-
-	function stop(): Promise<void> {
-		if (stopping) return stopping;
-		accepting = false;
-		const closed = pool.run(null, { name: 'close' });
-		stopping = Promise.allSettled([closed, pool.close()]).then((results) => {
-			if (results.some((result) => result.status === 'rejected')) report();
-		});
-		return stopping;
-	}
-
-	pool.on('error', () => {
-		report();
-		void stop();
-	});
-	void pool.run(null, { name: 'ready' }).then(
-		() => {
-			if (!stopping) accepting = true;
-		},
-		() => {
-			report();
-			void stop();
-		}
-	);
-
-	function recordVisit(visit: NewVisit): void {
-		if (!accepting) return;
-		if (outstanding >= MAX_OUTSTANDING || visitBytes(visit) > MAX_VISIT_BYTES) {
-			report();
-			return;
-		}
-		outstanding += 1;
-		try {
-			void pool.run(visit).then(
-				() => {
-					outstanding -= 1;
-				},
-				() => {
-					outstanding -= 1;
-					report();
-				}
-			);
-		} catch {
-			outstanding -= 1;
-			report();
-		}
-	}
-
-	return { recordVisit, stop };
+	};
 }

apps/worker/e2e/startup.js

@@ -7,6 +7,52 @@ import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { fileURLToPath } from 'node:url';
 import { DatabaseSync } from 'node:sqlite';
+import { setTimeout as delay } from 'node:timers/promises';
+import { insertVisit, openDb, runMigrations } from '@goto/db';
+
+const visit = {
+	domain: 'go.example.test',
+	path: '/docs',
+	visited_at: Date.now(),
+	ip: '127.0.0.1',
+	user_agent: null,
+	accept_language: null,
+	browser: 'Other',
+	os: 'Other'
+};
+
+async function seedHistory(directory) {
+	const db = await openDb(join(directory, 'data/goto.sqlite'));
+	try {
+		await runMigrations(db);
+		await insertVisit(db, { ...visit, visited_at: Date.UTC(2020, 0, 1) });
+		await insertVisit(db, visit);
+	} finally {
+		await db.destroy();
+	}
+}
+
+function visitState(directory) {
+	const db = new DatabaseSync(join(directory, 'data/goto.sqlite'), { readOnly: true });
+	try {
+		return {
+			count: db.prepare('SELECT count(*) AS count FROM visits').get().count,
+			collectAfter: db.prepare('SELECT collect_after FROM visit_collection WHERE id = 1').get()
+				.collect_after
+		};
+	} finally {
+		db.close();
+	}
+}
+
+async function saveConfiguration(path, configuration) {
+	await writeFile(`${path}.next`, JSON.stringify(configuration));
+	await rename(`${path}.next`, path);
+}
+
+async function waitUntil(check, signal) {
+	while (!check()) await delay(25, undefined, { signal });
+}
 
 async function checkStartup(mode, port = 0) {
 	const directory = await mkdtemp(join(tmpdir(), 'goto-worker-startup-'));
@@ -24,9 +70,11 @@ async function checkStartup(mode, port = 0) {
 			JSON.stringify({
 				site: { url: 'http://admin.example.test' },
 				auth: { credentials: { username: 'smoke', password: 'private-worker-smoke-password' } },
+				analytics: { retention_months: mode === 'retention-change' ? 3 : 0 },
 				domains: {}
 			})
 		);
+		if (mode === 'retention-change') await seedHistory(directory);
 		const source = mode === 'source';
 		const entry = new URL(source ? '../src/index.ts' : '../dist/index.js', import.meta.url);
 		const register = fileURLToPath(new URL('../../../tools/register.mjs', import.meta.url));
@@ -93,15 +141,16 @@ async function checkStartup(mode, port = 0) {
 			});
 			assert.equal(response.status, 200);
 			assert.deepEqual(await response.json(), { status: 'ok' });
-			assert(!existsSync(join(directory, 'data/goto.sqlite')), 'Visit database was opened');
+			const visits = visitState(directory);
+			assert.equal(visits.count, mode === 'retention-change' ? 1 : 0);
+			if (mode !== 'retention-change') assert.equal(visits.collectAfter, null);
 			const database = new DatabaseSync(join(directory, 'data/worker.sqlite'), {
 				readOnly: true
 			});
 			try {
-				assert.equal(
-					database.prepare('SELECT count(*) AS count FROM workflow_runs').get().count,
-					0
-				);
+				const runs = database.prepare('SELECT count(*) AS count FROM workflow_runs').get().count;
+				if (mode === 'retention-change') assert(runs >= 1);
+				else assert.equal(runs, 0);
 			} finally {
 				database.close();
 			}
@@ -109,8 +158,23 @@ async function checkStartup(mode, port = 0) {
 				const path = join(directory, 'config/goto.yaml');
 				const configuration = JSON.parse(await readFile(path, 'utf8'));
 				configuration.database = { worker_path: './data/restarted-worker.sqlite' };
-				await writeFile(`${path}.next`, JSON.stringify(configuration));
-				await rename(`${path}.next`, path);
+				await saveConfiguration(path, configuration);
+			} else if (mode === 'retention-change') {
+				const stopped = new AbortController();
+				void finished.then(
+					() => stopped.abort(),
+					() => stopped.abort()
+				);
+				const path = join(directory, 'config/goto.yaml');
+				const configuration = JSON.parse(await readFile(path, 'utf8'));
+				configuration.analytics.retention_months = 0;
+				await saveConfiguration(path, configuration);
+				await waitUntil(() => visitState(directory).count === 0, stopped.signal);
+				assert.equal(visitState(directory).collectAfter, null);
+				configuration.analytics.retention_months = 9;
+				await saveConfiguration(path, configuration);
+				await waitUntil(() => visitState(directory).collectAfter !== null, stopped.signal);
+				child.kill('SIGTERM');
 			} else {
 				child.kill('SIGTERM');
 			}
@@ -137,6 +201,7 @@ async function checkStartup(mode, port = 0) {
 await checkStartup('compiled');
 await checkStartup('source');
 await checkStartup('storage-change');
+await checkStartup('retention-change');
 const occupied = createServer();
 await new Promise((resolve, reject) => {
 	occupied.once('error', reject);
@@ -150,5 +215,5 @@ try {
 	await new Promise((resolve) => occupied.close(resolve));
 }
 console.log(
-	'Compiled/source startup, config restart, empty queue, health, partial-startup cleanup, and shutdown passed'
+	'Compiled/source startup, config restart, idle zero policy, live retention, health, partial-startup cleanup, and shutdown passed'
 );

apps/worker/package.json

@@ -15,10 +15,11 @@
 		"format": "prettier --ignore-path ../../.prettierignore --write .",
 		"test": "vitest run",
 		"test:coverage": "vitest run --coverage",
-		"test:smoke": "node e2e/startup.mjs"
+		"test:smoke": "node e2e/startup.js"
 	},
 	"dependencies": {
 		"@goto/config": "workspace:*",
+		"@goto/db": "workspace:*",
 		"openworkflow": "catalog:",
 		"tslog": "catalog:"
 	},

apps/worker/src/services/config-reload.test.ts

@@ -1,54 +1,125 @@
 import { startConfigReloading, validateConfig } from '@goto/config';
-import { afterEach, expect, test, vi } from 'vitest';
+import type { OpenWorkflow } from 'openworkflow';
+import { afterEach, beforeEach, expect, test, vi } from 'vitest';
 import { applyConfig, watchConfiguration } from './config-reload.js';
+import type { WorkerContext } from './context.js';
 import type { ShutdownResources } from './lifecycle.js';
+import { configureRetention, requestRetention } from './retention.js';
 
 vi.mock('./logger.js', () => ({ logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() } }));
+vi.mock('./retention.js', () => ({ configureRetention: vi.fn(), requestRetention: vi.fn() }));
 vi.mock('@goto/config', async (importOriginal) => ({
 	...(await importOriginal<typeof import('@goto/config')>()),
 	startConfigReloading: vi.fn()
 }));
 
-function configuration(worker_path = './data/worker.sqlite') {
+function configuration(
+	worker_path = './data/worker.sqlite',
+	retention_months = 3,
+	path = './data/goto.sqlite'
+) {
 	return validateConfig(
 		{
 			auth: { credentials: { username: 'test', password: 'config-test-password' } },
-			database: { worker_path },
+			database: { worker_path, path },
+			analytics: { retention_months },
 			domains: {}
 		},
 		{ env: {} }
 	);
 }
 
+const engine = {} as OpenWorkflow;
+function context(): WorkerContext {
+	return { config: configuration().configuration, db: {} as WorkerContext['db'] };
+}
+
+beforeEach(() => {
+	vi.resetAllMocks();
+});
+
 afterEach(() => {
 	vi.restoreAllMocks();
 	vi.useRealTimers();
 });
 
-test('accepts unchanged worker storage and requests restart for a new storage path', () => {
+test('requests restart for either database path and rejects the whole candidate', async () => {
 	const initial = configuration();
 	const restart = vi.fn();
-	expect(applyConfig(configuration(), initial, restart)).toBe(true);
-	expect(restart).not.toHaveBeenCalled();
-	expect(applyConfig(configuration('./data/other.sqlite'), initial, restart)).toBe(false);
-	expect(restart).toHaveBeenCalledOnce();
+	const ctx = context();
+	const accepted = ctx.config;
+	expect(await applyConfig(configuration(), initial, restart, ctx, engine)).toBe(true);
+	ctx.config = accepted;
+	expect(await applyConfig(configuration('./other.sqlite', 0), initial, restart, ctx, engine)).toBe(
+		false
+	);
+	expect(
+		await applyConfig(configuration(undefined, 0, './other.sqlite'), initial, restart, ctx, engine)
+	).toBe(false);
+	expect(restart).toHaveBeenCalledTimes(2);
+	expect(ctx.config).toBe(accepted);
+	expect(configureRetention).not.toHaveBeenCalled();
+});
+
+test('waits for zero-policy purge and reconciles a later re-enable', async () => {
+	const ctx = context();
+	const purged = Promise.withResolvers<void>();
+	vi.mocked(requestRetention).mockReturnValueOnce(purged.promise);
+	const disabled = configuration(undefined, 0);
+	const applying = applyConfig(disabled, configuration(), vi.fn(), ctx, engine);
+	await Promise.resolve();
+	expect(ctx.config.analytics.retention_months).toBe(0);
+	expect(requestRetention).toHaveBeenCalledExactlyOnceWith(ctx, engine);
+	let accepted = false;
+	void applying.then(() => {
+		accepted = true;
+	});
+	await Promise.resolve();
+	expect(accepted).toBe(false);
+	purged.resolve();
+	expect(await applying).toBe(true);
+	expect(await applyConfig(configuration(undefined, 9), disabled, vi.fn(), ctx, engine)).toBe(true);
+	expect(ctx.config.analytics.retention_months).toBe(9);
+	expect(requestRetention).toHaveBeenCalledTimes(2);
+});
+
+test('reconciles positive policy changes immediately', async () => {
+	const ctx = context();
+	await applyConfig(configuration(), configuration(undefined, 9), vi.fn(), ctx, engine);
+	expect(requestRetention).toHaveBeenCalledOnce();
+	await applyConfig(configuration(undefined, 12), configuration(), vi.fn(), ctx, engine);
+	expect(configureRetention).toHaveBeenCalledTimes(2);
+	expect(requestRetention).toHaveBeenCalledTimes(2);
+});
+
+test('a failed policy reconciliation keeps zero effective and requests shutdown', async () => {
+	const ctx = context();
+	const initial = configuration();
+	const failure = new Error('Private storage detail');
+	vi.mocked(requestRetention).mockRejectedValueOnce(failure);
+	const stop = vi.fn();
+	await expect(applyConfig(configuration(undefined, 0), initial, stop, ctx, engine)).rejects.toBe(
+		failure
+	);
+	expect(ctx.config.analytics.retention_months).toBe(0);
+	expect(stop).toHaveBeenCalledOnce();
 });
 
 test('initial reread restart waits until the monitor stop handle is owned', async () => {
 	vi.useFakeTimers();
 	const initial = configuration();
-	const stopWatch = vi.fn();
+	const stopWatch = vi.fn(async () => {});
 	const resources: ShutdownResources = {};
 	const shutdown = vi.fn(async () => {
-		resources.stopWatch?.();
+		await resources.stopWatch?.();
 	});
-	const monitor = Promise.withResolvers<() => void>();
+	const monitor = Promise.withResolvers<() => Promise<void>>();
 	vi.mocked(startConfigReloading).mockImplementationOnce(async (options) => {
-		options.onChange(configuration('./data/new.sqlite'), initial);
+		await options.onChange(configuration('./data/new.sqlite'), initial);
 		return monitor.promise;
 	});
 	const source = { path: '/config/goto.yaml', initial, load: async () => initial };
-	const watching = watchConfiguration(source, resources, shutdown);
+	const watching = watchConfiguration(source, resources, shutdown, context(), engine);
 	await vi.runAllTimersAsync();
 	expect(shutdown).not.toHaveBeenCalled();
 	monitor.resolve(stopWatch);

apps/worker/src/services/config-reload.ts

@@ -1,6 +1,9 @@
 import { startConfigReloading, type PreparedConfiguration } from '@goto/config';
 import { logger } from './logger.js';
 import type { ShutdownResources } from './lifecycle.js';
+import type { OpenWorkflow } from 'openworkflow';
+import type { WorkerContext } from './context.js';
+import { configureRetention, requestRetention } from './retention.js';
 
 interface ConfigurationSource {
 	path: string;
@@ -8,16 +11,37 @@ interface ConfigurationSource {
 	load(path: string): Promise<PreparedConfiguration>;
 }
 
-export function applyConfig(
+export async function applyConfig(
 	next: PreparedConfiguration,
 	previous: PreparedConfiguration,
-	onRestart: () => void
-): boolean {
-	if (next.configuration.database.worker_path !== previous.configuration.database.worker_path) {
+	onRestart: () => void,
+	context: WorkerContext,
+	engine: OpenWorkflow
+): Promise<boolean> {
+	const nextDatabase = next.configuration.database;
+	const previousDatabase = previous.configuration.database;
+	if (
+		nextDatabase.worker_path !== previousDatabase.worker_path ||
+		nextDatabase.path !== previousDatabase.path
+	) {
 		logger.warn('Worker configuration change requires restart');
 		onRestart();
 		return false;
 	}
+	const currentMonths = context.config.analytics.retention_months;
+	context.config = next.configuration;
+	const months = next.configuration.analytics.retention_months;
+	const previousMonths = previous.configuration.analytics.retention_months;
+	try {
+		if (months !== previousMonths || months !== currentMonths) {
+			await configureRetention(context, engine);
+			await requestRetention(context, engine);
+		}
+	} catch (error) {
+		logger.error('Unable to apply worker retention policy; stopping service');
+		onRestart();
+		throw error;
+	}
 	logger.info('Worker configuration reloaded');
 	return true;
 }
@@ -25,7 +49,9 @@ export function applyConfig(
 export async function watchConfiguration(
 	source: ConfigurationSource,
 	resources: ShutdownResources,
-	shutdown: () => Promise<void>
+	shutdown: () => Promise<void>,
+	context: WorkerContext,
+	engine: OpenWorkflow
 ): Promise<void> {
 	let watchOwned = false;
 	let restartRequested = false;
@@ -42,9 +68,9 @@ export async function watchConfiguration(
 	}
 	resources.stopWatch = await startConfigReloading({
 		...source,
-		onChange: (next, previous) => applyConfig(next, previous, requestRestart),
+		onChange: (next, previous) => applyConfig(next, previous, requestRestart, context, engine),
 		onError: () =>
-			logger.error('Invalid worker configuration update; keeping current configuration')
+			logger.error('Unable to apply worker configuration update; keeping current configuration')
 	});
 	watchOwned = true;
 	if (restartRequested) shutdownForRestart();

apps/worker/src/services/context.ts

@@ -0,0 +1,7 @@
+import type { GotoConfiguration } from '@goto/config';
+import type { openDb } from '@goto/db';
+
+export interface WorkerContext {
+	config: GotoConfiguration;
+	db: Awaited<ReturnType<typeof openDb>>;
+}

apps/worker/src/services/lifecycle.test.ts

@@ -5,19 +5,26 @@ test('shutdown is idempotent and waits for the worker before closing its backend
 	const workerStopped = Promise.withResolvers<void>();
 	const workflowWorker = { stop: vi.fn(() => workerStopped.promise) };
 	const backend = { stop: vi.fn(async () => {}) };
-	const stopWatch = vi.fn();
-	const stop = createShutdown({ workflowWorker, backend, stopWatch });
+	const db = { destroy: vi.fn(async () => {}) };
+	const monitorStopped = Promise.withResolvers<void>();
+	const stopWatch = vi.fn(() => monitorStopped.promise);
+	const stop = createShutdown({ workflowWorker, backend, stopWatch, db });
 	const stopping = stop();
 	expect(stop()).toBe(stopping);
-	expect(workflowWorker.stop).toHaveBeenCalledOnce();
+	expect(workflowWorker.stop).not.toHaveBeenCalled();
 	expect(stopWatch).toHaveBeenCalledOnce();
+	monitorStopped.resolve();
+	await Promise.resolve();
+	expect(workflowWorker.stop).toHaveBeenCalledOnce();
 	expect(stopWatch.mock.invocationCallOrder[0]).toBeLessThan(
 		workflowWorker.stop.mock.invocationCallOrder[0]!
 	);
 	expect(backend.stop).not.toHaveBeenCalled();
+	expect(db.destroy).not.toHaveBeenCalled();
 	workerStopped.resolve();
 	await stopping;
 	expect(backend.stop).toHaveBeenCalledOnce();
+	expect(db.destroy).toHaveBeenCalledOnce();
 });
 
 test('a failed monitor stop still stops the worker and closes the backend', async () => {
@@ -34,7 +41,7 @@ test('a failed monitor stop still stops the worker and closes the backend', asyn
 	expect(backend.stop).toHaveBeenCalledOnce();
 });
 
-test('a failed worker stop still closes the workflow backend', async () => {
+test('a failed worker stop still closes both databases', async () => {
 	const failure = new Error('Worker stop failed');
 	const workflowWorker = {
 		stop: vi.fn(async () => {
@@ -42,8 +49,10 @@ test('a failed worker stop still closes the workflow backend', async () => {
 		})
 	};
 	const backend = { stop: vi.fn(async () => {}) };
-	const stop = createShutdown({ workflowWorker, backend });
+	const db = { destroy: vi.fn(async () => {}) };
+	const stop = createShutdown({ workflowWorker, backend, db });
 
 	await expect(stop()).rejects.toBe(failure);
 	expect(backend.stop).toHaveBeenCalledOnce();
+	expect(db.destroy).toHaveBeenCalledOnce();
 });

apps/worker/src/services/lifecycle.ts

@@ -1,4 +1,5 @@
 import type { Server } from 'node:http';
+import type { openDb } from '@goto/db';
 import type { Worker as WorkflowWorker } from 'openworkflow';
 import type { BackendSqlite } from 'openworkflow/sqlite';
 
@@ -6,7 +7,8 @@ export interface ShutdownResources {
 	workflowWorker?: Pick<WorkflowWorker, 'stop'>;
 	server?: Server;
 	backend?: Pick<BackendSqlite, 'stop'>;
-	stopWatch?: () => void;
+	stopWatch?: () => void | Promise<void>;
+	db?: Pick<Awaited<ReturnType<typeof openDb>>, 'destroy'>;
 }
 
 export function createShutdown(resources: ShutdownResources): () => Promise<void> {
@@ -17,7 +19,7 @@ export function createShutdown(resources: ShutdownResources): () => Promise<void
 async function stopResources(resources: ShutdownResources): Promise<void> {
 	const failures: unknown[] = [];
 	try {
-		resources.stopWatch?.();
+		await resources.stopWatch?.();
 	} catch (error) {
 		failures.push(error);
 	}
@@ -41,5 +43,10 @@ async function stopResources(resources: ShutdownResources): Promise<void> {
 	} catch (error) {
 		failures.push(error);
 	}
+	try {
+		await resources.db?.destroy();
+	} catch (error) {
+		failures.push(error);
+	}
 	if (failures.length > 0) throw failures[0];
 }

apps/worker/src/services/retention.test.ts

@@ -0,0 +1,179 @@
+import { validateConfig } from '@goto/config';
+import { openDb, runMigrations, setVisitCollection } from '@goto/db';
+import type { OpenWorkflow } from 'openworkflow';
+import { expect, test, vi } from 'vitest';
+import {
+	applyRetention,
+	configureRetention,
+	requestRetention,
+	retentionCutoff
+} from './retention.js';
+import type { WorkerContext } from './context.js';
+import { retainVisits, retentionSchedule } from '../workflows/specs.js';
+
+test('uses UTC calendar months, clamping the day while preserving the time', () => {
+	const now = Date.parse('2026-05-31T12:34:56.789Z');
+	expect(new Date(retentionCutoff(3, now)).toISOString()).toBe('2026-02-28T12:34:56.789Z');
+	expect(new Date(retentionCutoff(9, now)).toISOString()).toBe('2025-08-31T12:34:56.789Z');
+	expect(new Date(retentionCutoff(12, now)).toISOString()).toBe('2025-05-31T12:34:56.789Z');
+});
+
+async function context(): Promise<WorkerContext> {
+	const db = await openDb(':memory:');
+	await runMigrations(db);
+	const { configuration: config } = validateConfig(
+		{
+			auth: { credentials: { username: 'test', password: 'retention-test-password' } },
+			domains: {}
+		},
+		{ env: {} }
+	);
+	return { db, config };
+}
+
+function client() {
+	const result = vi.fn(async () => 'completed');
+	const engine = {
+		scheduleWorkflow: vi.fn(async () => {}),
+		schedules: { pause: vi.fn(async () => {}), resume: vi.fn(async () => {}) },
+		runWorkflow: vi.fn(async () => ({ result }))
+	};
+	return { engine, result };
+}
+
+const visit = {
+	domain: 'go.example.test',
+	path: '/docs',
+	visited_at: 1,
+	ip: '127.0.0.1',
+	user_agent: null,
+	accept_language: null,
+	browser: 'Other',
+	os: 'Other'
+};
+
+test('zero pauses the native schedule and queues one purge only when data exists', async () => {
+	const ctx = await context();
+	const { engine, result } = client();
+	const sdk = engine as unknown as OpenWorkflow;
+	try {
+		ctx.config.analytics.retention_months = 0;
+		await configureRetention(ctx, sdk);
+		expect(engine.scheduleWorkflow).toHaveBeenCalledExactlyOnceWith(
+			retainVisits,
+			retentionSchedule.input,
+			retentionSchedule
+		);
+		expect(engine.schedules.pause).toHaveBeenCalledExactlyOnceWith(retentionSchedule.id);
+		expect(engine.schedules.resume).not.toHaveBeenCalled();
+		await requestRetention(ctx, sdk);
+		expect(engine.runWorkflow).not.toHaveBeenCalled();
+		await ctx.db.insertInto('visits').values(visit).execute();
+		const completed = Promise.withResolvers<string>();
+		const requested = Promise.withResolvers<void>();
+		result.mockImplementationOnce(() => {
+			requested.resolve();
+			return completed.promise;
+		});
+		const requesting = requestRetention(ctx, sdk);
+		await requested.promise;
+		let finished = false;
+		void requesting.then(() => {
+			finished = true;
+		});
+		await Promise.resolve();
+		expect(finished).toBe(false);
+		completed.resolve('completed');
+		await requesting;
+		expect(engine.runWorkflow).toHaveBeenCalledExactlyOnceWith(retainVisits, {});
+		await applyRetention(ctx);
+		expect(await ctx.db.selectFrom('visits').selectAll().execute()).toEqual([]);
+		expect(
+			await ctx.db.selectFrom('visit_collection').select('collect_after').executeTakeFirstOrThrow()
+		).toEqual({ collect_after: null });
+	} finally {
+		await ctx.db.destroy();
+	}
+});
+
+test('positive retention resumes the schedule and prunes every expired batch using one cutoff', async () => {
+	const ctx = await context();
+	const { engine } = client();
+	try {
+		const now = Date.parse('2026-05-31T12:34:56.789Z');
+		const cutoff = retentionCutoff(3, now);
+		await ctx.db
+			.insertInto('visits')
+			.values([
+				...Array.from({ length: 1001 }, () => ({ ...visit, visited_at: cutoff - 1 })),
+				{ ...visit, visited_at: cutoff }
+			])
+			.execute();
+		await configureRetention(ctx, engine as unknown as OpenWorkflow);
+		expect(engine.schedules.resume).toHaveBeenCalledExactlyOnceWith(retentionSchedule.id);
+		expect(engine.schedules.pause).not.toHaveBeenCalled();
+		await applyRetention(ctx, now);
+		expect(await ctx.db.selectFrom('visits').select('visited_at').execute()).toEqual([
+			{ visited_at: cutoff }
+		]);
+	} finally {
+		await ctx.db.destroy();
+	}
+});
+
+test('finishes an interrupted zero-policy purge before reopening collection', async () => {
+	const ctx = await context();
+	try {
+		const now = Date.now();
+		await setVisitCollection(ctx.db, false);
+		await ctx.db
+			.insertInto('visits')
+			.values({ ...visit, visited_at: now })
+			.execute();
+		ctx.config.analytics.retention_months = 9;
+		await applyRetention(ctx, now);
+		expect(await ctx.db.selectFrom('visits').selectAll().execute()).toEqual([]);
+		const state = await ctx.db
+			.selectFrom('visit_collection')
+			.select('collect_after')
+			.executeTakeFirstOrThrow();
+		expect(state.collect_after).toBeTypeOf('number');
+	} finally {
+		await ctx.db.destroy();
+	}
+});
+
+test('a positive job waiting for its connection cannot reopen collection after zero', async () => {
+	const ctx = await context();
+	const acquired = Promise.withResolvers<void>();
+	const release = Promise.withResolvers<void>();
+	try {
+		await setVisitCollection(ctx.db, false);
+		const holding = ctx.db.transaction().execute(async () => {
+			acquired.resolve();
+			await release.promise;
+		});
+		await Promise.race([acquired.promise, holding]);
+		const applying = applyRetention(ctx);
+		ctx.config.analytics.retention_months = 0;
+		release.resolve();
+		await Promise.all([holding, applying]);
+		const state = await ctx.db
+			.selectFrom('visit_collection')
+			.select('collect_after')
+			.executeTakeFirstOrThrow();
+		expect(state.collect_after).toBeNull();
+	} finally {
+		release.resolve();
+		await ctx.db.destroy();
+	}
+});
+
+test('handles leap-day boundaries without overflowing into March', () => {
+	expect(new Date(retentionCutoff(3, Date.parse('2024-05-31T01:00:00Z'))).toISOString()).toBe(
+		'2024-02-29T01:00:00.000Z'
+	);
+	expect(new Date(retentionCutoff(12, Date.parse('2024-02-29T01:00:00Z'))).toISOString()).toBe(
+		'2023-02-28T01:00:00.000Z'
+	);
+});

apps/worker/src/services/retention.ts

@@ -0,0 +1,62 @@
+import { setImmediate } from 'node:timers/promises';
+import type { GotoConfiguration } from '@goto/config';
+import { pruneVisits, purgeVisits, purgeVisitsIfDisabled, setVisitCollection } from '@goto/db';
+import type { OpenWorkflow } from 'openworkflow';
+import { retainVisits, retentionSchedule } from '../workflows/specs.js';
+import type { WorkerContext } from './context.js';
+
+export type RetentionMonths = GotoConfiguration['analytics']['retention_months'];
+
+export async function configureRetention(
+	context: WorkerContext,
+	engine: OpenWorkflow
+): Promise<void> {
+	const enabled = context.config.analytics.retention_months !== 0;
+	if (!enabled) await setVisitCollection(context.db, false);
+	await engine.scheduleWorkflow(retainVisits, retentionSchedule.input, retentionSchedule);
+	if (enabled) await engine.schedules.resume(retentionSchedule.id);
+	else await engine.schedules.pause(retentionSchedule.id);
+}
+
+export async function requestRetention(
+	context: WorkerContext,
+	engine: OpenWorkflow
+): Promise<void> {
+	if (context.config.analytics.retention_months === 0) {
+		const visit = await context.db.selectFrom('visits').select('id').limit(1).executeTakeFirst();
+		if (!visit) return;
+	}
+	const run = await engine.runWorkflow(retainVisits, {});
+	await run.result();
+}
+
+export async function applyRetention(context: WorkerContext, now = Date.now()): Promise<void> {
+	const months = context.config.analytics.retention_months;
+	if (months === 0) {
+		await setVisitCollection(context.db, false);
+		await purgeVisits(context.db);
+		return;
+	}
+	await context.db.transaction().execute(async (transaction) => {
+		await purgeVisitsIfDisabled(transaction);
+		if (context.config.analytics.retention_months !== 0) {
+			await setVisitCollection(transaction, true);
+		}
+	});
+	const before = retentionCutoff(months, now);
+	while (await pruneVisits(context.db, before)) {
+		await setImmediate();
+	}
+}
+
+export function retentionCutoff(months: Exclude<RetentionMonths, 0>, now = Date.now()): number {
+	const cutoff = new Date(now);
+	const day = cutoff.getUTCDate();
+	cutoff.setUTCDate(1);
+	cutoff.setUTCMonth(cutoff.getUTCMonth() - months);
+	const lastDay = new Date(
+		Date.UTC(cutoff.getUTCFullYear(), cutoff.getUTCMonth() + 1, 0)
+	).getUTCDate();
+	cutoff.setUTCDate(Math.min(day, lastDay));
+	return cutoff.getTime();
+}

apps/worker/src/services/startup.test.ts

@@ -2,6 +2,8 @@ import { existsSync } from 'node:fs';
 import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
+import { DatabaseSync } from 'node:sqlite';
+import { insertVisit, openDb, runMigrations } from '@goto/db';
 import { Worker } from 'openworkflow';
 import { afterEach, beforeEach, expect, test, vi } from 'vitest';
 import { startWorkerService } from './startup.js';
@@ -18,21 +20,23 @@ afterEach(async () => {
 	await rm(directory, { recursive: true, force: true });
 });
 
-test('starts an empty workflow service and health server without opening visit storage', async () => {
-	await writeFile(
-		join(directory, 'config/goto.yaml'),
-		JSON.stringify({
-			site: { url: 'http://admin.example.test' },
-			auth: { credentials: { username: 'test', password: 'worker-test-only-password' } },
-			database: { path: './data/visits.sqlite', worker_path: './state/worker.sqlite' },
-			domains: {}
-		})
-	);
+async function writeConfiguration(worker_path = './state/worker.sqlite'): Promise<void> {
+	const configuration = {
+		site: { url: 'http://admin.example.test' },
+		auth: { credentials: { username: 'test', password: 'worker-test-only-password' } },
+		database: { path: './data/visits.sqlite', worker_path },
+		analytics: { retention_months: 0 },
+		domains: {}
+	};
+	await writeFile(join(directory, 'config/goto.yaml'), JSON.stringify(configuration));
+}
+
+const env = { HOST: '127.0.0.1', PORT: '0' };
+
+test('starts an idle zero-policy service, migrates visit storage, and exposes health', async () => {
+	await writeConfiguration();
 	const start = vi.spyOn(Worker.prototype, 'start');
-	const service = await startWorkerService({
-		cwd: directory,
-		env: { HOST: '127.0.0.1', PORT: '0' }
-	});
+	const service = await startWorkerService({ cwd: directory, env });
 	try {
 		expect(start).toHaveBeenCalledOnce();
 		const response = await fetch(`http://127.0.0.1:${service.port}/healthz`, {
@@ -40,14 +44,67 @@ test('starts an empty workflow service and health server without opening visit s
 		});
 		expect(response.status).toBe(200);
 		expect(await response.json()).toEqual({ status: 'ok' });
-		expect(existsSync(join(directory, 'state/worker.sqlite'))).toBe(true);
-		expect(existsSync(join(directory, 'data/visits.sqlite'))).toBe(false);
+		const state = new DatabaseSync(join(directory, 'state/worker.sqlite'), { readOnly: true });
+		try {
+			expect(state.prepare('SELECT count(*) AS count FROM workflow_runs').get()?.count).toBe(0);
+		} finally {
+			state.close();
+		}
+		const db = await openDb(join(directory, 'data/visits.sqlite'), { readOnly: true });
+		try {
+			expect(await db.selectFrom('visits').selectAll().execute()).toEqual([]);
+			expect(
+				await db.selectFrom('visit_collection').select('collect_after').executeTakeFirstOrThrow()
+			).toEqual({ collect_after: null });
+		} finally {
+			await db.destroy();
+		}
 	} finally {
 		await service.stop();
 	}
 });
 
-test('invalid configuration fails before creating workflow storage', async () => {
+test('an existing zero-policy database is purged by a real startup workflow', async () => {
+	await writeConfiguration();
+	const path = join(directory, 'data/visits.sqlite');
+	const db = await openDb(path);
+	try {
+		await runMigrations(db);
+		await insertVisit(db, {
+			domain: 'go.example.test',
+			path: '/docs',
+			visited_at: Date.now(),
+			ip: '127.0.0.1',
+			user_agent: null,
+			accept_language: null,
+			browser: 'Other',
+			os: 'Other'
+		});
+	} finally {
+		await db.destroy();
+	}
+	const service = await startWorkerService({ cwd: directory, env });
+	try {
+		const reader = await openDb(path, { readOnly: true });
+		try {
+			expect(await reader.selectFrom('visits').selectAll().execute()).toEqual([]);
+		} finally {
+			await reader.destroy();
+		}
+	} finally {
+		await service.stop();
+	}
+});
+
+test('rejects a shared worker/visit file before opening either connection', async () => {
+	await writeConfiguration('./data/visits.sqlite');
+	await expect(startWorkerService({ cwd: directory, env })).rejects.toThrow(
+		'Worker and visit database paths must differ'
+	);
+	expect(existsSync(join(directory, 'data'))).toBe(false);
+});
+
+test('invalid configuration fails before creating database storage', async () => {
 	await writeFile(join(directory, 'config/goto.yaml'), 'auth: {}\ndomains: {}\n');
 	await expect(startWorkerService({ cwd: directory, env: {} })).rejects.toThrow(
 		'Invalid configuration'

apps/worker/src/services/startup.ts

@@ -2,6 +2,7 @@ import { mkdir } from 'node:fs/promises';
 import { once } from 'node:events';
 import { dirname } from 'node:path';
 import { loadConfigFile, resolveConfigPath, type ConfigOptions } from '@goto/config';
+import { openDb, runMigrations } from '@goto/db';
 import { OpenWorkflow } from 'openworkflow';
 import { BackendSqlite } from 'openworkflow/sqlite';
 import { createEngineLogger } from './engine-logger.js';
@@ -9,6 +10,9 @@ import { createHealthServer } from './health.js';
 import { createShutdown, type ShutdownResources } from './lifecycle.js';
 import { watchConfiguration } from './config-reload.js';
 import { logger } from './logger.js';
+import type { WorkerContext } from './context.js';
+import { configureRetention, requestRetention } from './retention.js';
+import { registerRetentionWorkflow } from '../workflows/retain-visits.js';
 
 export interface WorkerService {
 	port: number;
@@ -32,15 +36,21 @@ export async function startWorkerService(
 	const loadConfiguration = () => loadConfigFile(path, captured);
 	const prepared = await loadConfiguration();
 	const databasePath = prepared.configuration.database.worker_path;
+	const visitDatabasePath = prepared.configuration.database.path;
+	if (databasePath === visitDatabasePath) {
+		throw new Error('Worker and visit database paths must differ');
+	}
 	const resources: ShutdownResources = {};
 	const stop = createShutdown(resources);
 	try {
-		await startWorkflowWorker(databasePath, resources);
+		const db = await openVisitDatabase(visitDatabasePath, resources);
+		const context: WorkerContext = { config: prepared.configuration, db };
+		const engine = await startWorkflowWorker(databasePath, context, resources);
 		const host = captured.env.HOST ?? '0.0.0.0';
 		const healthPort = await startHealthServer(host, port, resources);
 		const source = { path, initial: prepared, load: loadConfiguration };
 		const requestRestart = onRestartRequired ?? stop;
-		await watchConfiguration(source, resources, requestRestart);
+		await watchConfiguration(source, resources, requestRestart, context, engine);
 		return { port: healthPort, stop };
 	} catch (error) {
 		await stop();
@@ -48,20 +58,35 @@ export async function startWorkerService(
 	}
 }
 
+async function openVisitDatabase(
+	path: string,
+	resources: ShutdownResources
+): Promise<WorkerContext['db']> {
+	const db = await openDb(path);
+	resources.db = db;
+	await runMigrations(db);
+	return db;
+}
+
 async function startWorkflowWorker(
 	databasePath: string,
+	context: WorkerContext,
 	resources: ShutdownResources
-): Promise<void> {
+): Promise<OpenWorkflow> {
 	const directory = dirname(databasePath);
 	await mkdir(directory, { recursive: true });
 	const backend = BackendSqlite.connect(databasePath);
 	resources.backend = backend;
 	const engineLogger = createEngineLogger(logger);
 	const engine = new OpenWorkflow({ backend, logger: engineLogger });
-	const workflowWorker = engine.newWorker();
+	registerRetentionWorkflow(engine, context);
+	await configureRetention(context, engine);
+	const workflowWorker = engine.newWorker({ concurrency: 1 });
 	resources.workflowWorker = workflowWorker;
 	await workflowWorker.start();
+	await requestRetention(context, engine);
 	logger.info('Workflow worker started');
+	return engine;
 }
 
 async function startHealthServer(

apps/worker/src/workflows/retain-visits.test.ts

@@ -0,0 +1,73 @@
+import { validateConfig } from '@goto/config';
+import { afterEach, expect, test, vi } from 'vitest';
+import type { WorkerContext } from '../services/context.js';
+import { applyRetention } from '../services/retention.js';
+import { runRetentionWorkflow, type RetentionStep } from './retain-visits.js';
+
+vi.mock('../services/retention.js', () => ({ applyRetention: vi.fn(async () => {}) }));
+
+afterEach(() => vi.clearAllMocks());
+
+function context(): WorkerContext {
+	const prepared = validateConfig(
+		{
+			auth: { credentials: { username: 'test', password: 'jitter-test-password' } },
+			domains: {}
+		},
+		{ env: {} }
+	);
+	return { config: prepared.configuration, db: {} as WorkerContext['db'] };
+}
+
+function steps() {
+	const names: string[] = [];
+	const sleep = vi
+		.fn<(name: string, duration: string) => Promise<void>>()
+		.mockResolvedValue(undefined);
+	const step: RetentionStep = {
+		async run(config, fn) {
+			names.push(config.name);
+			return fn();
+		},
+		sleep
+	};
+	return { step, names, sleep };
+}
+
+test('scheduled cleanup uses durable jitter; startup and policy runs do not', async () => {
+	const ctx = context();
+	const { step, names, sleep } = steps();
+	await runRetentionWorkflow(ctx, { scheduled: true }, step);
+	expect(names).toEqual(['choose-jitter', 'retain-visits']);
+	expect(sleep).toHaveBeenCalledOnce();
+	const [name, duration] = sleep.mock.calls[0]!;
+	const seconds = Number(duration.slice(0, -1));
+	expect(name).toBe('retention-jitter');
+	expect(duration.endsWith('s')).toBe(true);
+	expect(seconds).toBeGreaterThanOrEqual(1);
+	expect(seconds).toBeLessThanOrEqual(60);
+	expect(applyRetention).toHaveBeenCalledExactlyOnceWith(ctx);
+	vi.clearAllMocks();
+	names.length = 0;
+	await runRetentionWorkflow(ctx, {}, step);
+	expect(names).toEqual(['retain-visits']);
+	expect(sleep).not.toHaveBeenCalled();
+	expect(applyRetention).toHaveBeenCalledExactlyOnceWith(ctx);
+});
+
+test('a scheduled run does no cleanup if collection is disabled during its delay', async () => {
+	const ctx = context();
+	const { step, sleep } = steps();
+	const sleeping = Promise.withResolvers<void>();
+	const resumed = Promise.withResolvers<void>();
+	sleep.mockImplementationOnce(async () => {
+		sleeping.resolve();
+		await resumed.promise;
+	});
+	const running = runRetentionWorkflow(ctx, { scheduled: true }, step);
+	await sleeping.promise;
+	ctx.config.analytics.retention_months = 0;
+	resumed.resolve();
+	await running;
+	expect(applyRetention).not.toHaveBeenCalled();
+});

apps/worker/src/workflows/retain-visits.ts

@@ -0,0 +1,47 @@
+import { randomInt } from 'node:crypto';
+import type { OpenWorkflow } from 'openworkflow';
+import type { WorkerContext } from '../services/context.js';
+import { applyRetention } from '../services/retention.js';
+import { retainVisits, type RetainVisitsInput } from './specs.js';
+
+const MAX_JITTER_SECONDS = 60;
+
+export interface RetentionStep {
+	run<T>(config: { name: string }, fn: () => Promise<T>): Promise<T>;
+	sleep(name: string, duration: `${number}s`): Promise<void>;
+}
+
+function collectionDisabled(context: WorkerContext): boolean {
+	return context.config.analytics.retention_months === 0;
+}
+
+async function chooseJitter(): Promise<number> {
+	const upperBound = MAX_JITTER_SECONDS + 1;
+	return randomInt(1, upperBound);
+}
+
+export async function runRetentionWorkflow(
+	context: WorkerContext,
+	input: RetainVisitsInput,
+	step: RetentionStep
+): Promise<'completed'> {
+	if (input.scheduled) {
+		if (collectionDisabled(context)) return 'completed';
+		const seconds = await step.run({ name: 'choose-jitter' }, chooseJitter);
+		const duration: `${number}s` = `${seconds}s`;
+		await step.sleep('retention-jitter', duration);
+		if (collectionDisabled(context)) return 'completed';
+	}
+	function applyPolicy(): Promise<void> {
+		return applyRetention(context);
+	}
+	await step.run({ name: 'retain-visits' }, applyPolicy);
+	return 'completed';
+}
+
+export function registerRetentionWorkflow(engine: OpenWorkflow, context: WorkerContext): void {
+	function execute({ input, step }: { input: RetainVisitsInput; step: RetentionStep }) {
+		return runRetentionWorkflow(context, input, step);
+	}
+	engine.implementWorkflow(retainVisits, execute);
+}

apps/worker/src/workflows/specs.ts

@@ -0,0 +1,18 @@
+import { defineWorkflowSpec, type Workflow } from 'openworkflow';
+
+export interface RetainVisitsInput {
+	scheduled?: boolean;
+}
+
+export const retentionSchedule = {
+	id: 'goto:retain-visits',
+	cron: '0 * * * *',
+	timezone: 'UTC',
+	input: { scheduled: true }
+};
+
+export const retainVisits: Workflow<RetainVisitsInput, 'completed', RetainVisitsInput>['spec'] =
+	defineWorkflowSpec<RetainVisitsInput, 'completed'>({
+		name: 'retain-visits',
+		schedules: [retentionSchedule]
+	});

examples/goto.yaml

@@ -21,8 +21,7 @@ database:
   worker_path: ./data/worker.sqlite
 
 analytics:
-  enabled: true
-  retention_days: 90
+  retention_months: 3
 
 defaults:
   status_code: 302

packages/config/src/config.schema.json

@@ -288,8 +288,16 @@
 			"additionalProperties": false,
 			"type": "object",
 			"properties": {
-				"enabled": { "default": true, "type": "boolean" },
-				"retention_days": { "minimum": 1, "default": 90, "type": "integer" }
+				"retention_months": {
+					"default": 3,
+					"description": "Calendar months of analytics history. Zero disables collection and purges retained visits.",
+					"anyOf": [
+						{ "const": 0, "type": "number" },
+						{ "const": 3, "type": "number" },
+						{ "const": 9, "type": "number" },
+						{ "const": 12, "type": "number" }
+					]
+				}
 			}
 		},
 		"Defaults": {

packages/config/src/load.test.ts

@@ -23,12 +23,40 @@ describe('configuration loading', () => {
 			version: 1,
 			site: { url: 'http://localhost:3000', title: 'Goto' },
 			database: { path: './data/goto.sqlite', worker_path: './data/worker.sqlite' },
-			analytics: { enabled: true, retention_days: 90 },
+			analytics: { retention_months: 3 },
 			defaults: { status_code: 302, forward_query: false, fallback: 404 }
 		});
 		expect(input).toEqual(original);
 	});
 
+	it.each([0, 3, 9, 12])('accepts the finite analytics retention %s', (retention_months) => {
+		const input = { ...configurationInput(), analytics: { retention_months } };
+		expect(validateConfig(input, options).configuration.analytics.retention_months).toBe(
+			retention_months
+		);
+	});
+
+	it.each([1, 6, Infinity, '3'])(
+		'rejects unsupported analytics retention %j',
+		(retention_months) => {
+			expect(() =>
+				validateConfig({ ...configurationInput(), analytics: { retention_months } }, options)
+			).toThrow(ConfigError);
+		}
+	);
+
+	it('rejects the replaced analytics switches instead of silently ignoring them', () => {
+		expect(() =>
+			validateConfig(
+				{
+					...configurationInput(),
+					analytics: { enabled: false, retention_days: 90 }
+				},
+				options
+			)
+		).toThrow(ConfigError);
+	});
+
 	it.each([
 		[{}, 'http://localhost:3000'],
 		[{ NODE_ENV: 'development' }, 'http://localhost:5173'],
@@ -230,7 +258,7 @@ describe('configuration loading', () => {
 				GOTO_OIDC_CLIENT_SECRET: 'example-client-secret'
 			}
 		});
-		expect(prepared.configuration.analytics.retention_days).toBe(90);
+		expect(prepared.configuration.analytics.retention_months).toBe(3);
 		expect(prepared.domains['go.example.org']?.links['/docs']).toMatchObject({
 			forward_query: false,
 			provenance: { forward_query: 'link' }

packages/config/src/monitor.test.ts

@@ -14,7 +14,9 @@ function input() {
 		path: '/config/goto.yaml',
 		initial: 'initial',
 		load: vi.fn<(path: string) => Promise<string>>().mockResolvedValue('reconciled'),
-		onChange: vi.fn<(next: string, previous: string) => boolean>().mockReturnValue(true),
+		onChange: vi
+			.fn<(next: string, previous: string) => boolean | Promise<boolean>>()
+			.mockReturnValue(true),
 		onError: vi.fn<(error: unknown) => void>()
 	};
 }
@@ -122,6 +124,31 @@ test.each(['resolve', 'reject'])('stop suppresses late %s and pending reloads',
 	expect(options.onError).not.toHaveBeenCalled();
 });
 
+test('awaits policy application before accepting another candidate and drains it on stop', async () => {
+	const options = input();
+	const stop = await startConfigReloading(options);
+	const applied = Promise.withResolvers<boolean>();
+	options.load.mockResolvedValueOnce('zero');
+	options.onChange.mockReturnValueOnce(applied.promise);
+	const applying = change();
+	await Promise.resolve();
+	await Promise.resolve();
+	expect(options.onChange).toHaveBeenLastCalledWith('zero', 'reconciled');
+	void change();
+	const stopping = stop();
+	let stopped = false;
+	void stopping.then(() => {
+		stopped = true;
+	});
+	await Promise.resolve();
+	expect(stopped).toBe(false);
+	expect(options.load).toHaveBeenCalledTimes(2);
+	applied.resolve(true);
+	await applying;
+	await stopping;
+	expect(options.load).toHaveBeenCalledTimes(2);
+});
+
 test('stop prevents a queued read from starting', async () => {
 	const options = input();
 	const stop = await startConfigReloading(options);

packages/config/src/monitor.ts

@@ -10,10 +10,10 @@ export async function startConfigReloading<T>({
 	path: string;
 	initial: T;
 	load: (path: string) => Promise<T>;
-	onChange: (next: T, previous: T) => boolean;
+	onChange: (next: T, previous: T) => boolean | Promise<boolean>;
 	/** Receives raw failures; the caller is responsible for safe diagnostics. */
 	onError: (error: unknown) => void;
-}): Promise<() => void> {
+}): Promise<() => Promise<void>> {
 	let accepted = initial;
 	let stopped = false;
 	let pending = false;
@@ -26,7 +26,8 @@ export async function startConfigReloading<T>({
 				try {
 					const next = await load(path);
 					if (stopped) return;
-					if (onChange(next, accepted) && !stopped) accepted = next;
+					const applied = await onChange(next, accepted);
+					if (applied && !stopped) accepted = next;
 				} catch (error) {
 					if (stopped) return;
 					try {
@@ -62,11 +63,13 @@ export async function startConfigReloading<T>({
 		return reload();
 	}
 
-	function stop(): void {
-		if (stopped) return;
-		stopped = true;
-		pending = false;
-		unwatchFile(path, listener);
+	async function stop(): Promise<void> {
+		if (!stopped) {
+			stopped = true;
+			pending = false;
+			unwatchFile(path, listener);
+		}
+		await active;
 	}
 
 	// Poll the path so atomic saves and deletion/recreation remain observable.

packages/config/src/schema.ts

@@ -186,8 +186,14 @@ const DatabaseSchema = Type.Object(
 
 const AnalyticsSchema = Type.Object(
 	{
-		enabled: Type.Boolean({ default: true }),
-		retention_days: Type.Integer({ minimum: 1, default: 90 })
+		retention_months: Type.Union(
+			[Type.Literal(0), Type.Literal(3), Type.Literal(9), Type.Literal(12)],
+			{
+				default: 3,
+				description:
+					'Calendar months of analytics history. Zero disables collection and purges retained visits.'
+			}
+		)
 	},
 	{ $id: 'Analytics', additionalProperties: false }
 );

packages/db/package.json

@@ -6,6 +6,7 @@
 	"exports": {
 		".": {
 			"types": "./dist/index.d.ts",
+			"development": "./src/index.ts",
 			"import": "./dist/index.js"
 		},
 		"./visit-worker": "./dist/visit-worker.js"

packages/db/src/index.ts

@@ -6,7 +6,13 @@ import { NodeSqliteDialect } from './node-sqlite-dialect.js';
 import type { Database } from './types.js';
 
 export { runMigrations } from './migrate.js';
-export { insertVisit } from './visits.js';
+export {
+	insertVisit,
+	pruneVisits,
+	purgeVisits,
+	purgeVisitsIfDisabled,
+	setVisitCollection
+} from './visits.js';
 export type { ClassifiedVisit, Database, NewVisit, Visit } from './types.js';
 
 export async function openDb(
@@ -16,12 +22,12 @@ export async function openDb(
 	if (!readOnly && path !== ':memory:') mkdirSync(dirname(path), { recursive: true });
 	const database = new DatabaseSync(path, { readOnly });
 	try {
+		database.exec('PRAGMA busy_timeout = 5000');
 		if (!readOnly) {
 			database.exec('PRAGMA journal_mode = WAL');
 			database.exec('PRAGMA synchronous = NORMAL');
 		}
 		database.exec('PRAGMA foreign_keys = ON');
-		database.exec('PRAGMA busy_timeout = 5000');
 		const db = new Kysely<Database>({ dialect: new NodeSqliteDialect({ database }) });
 		await db.connection().execute(async () => undefined);
 		return db;

packages/db/src/migrations/0003_visit_retention.ts

@@ -0,0 +1,17 @@
+import { sql, type Kysely, type Migration } from 'kysely';
+
+export const visitRetention: Migration = {
+	async up(db: Kysely<unknown>): Promise<void> {
+		await db.schema.createIndex('visits_visited_at').on('visits').column('visited_at').execute();
+		await db.schema
+			.createTable('visit_collection')
+			.addColumn('id', 'integer', (column) => column.primaryKey())
+			.addColumn('collect_after', 'integer')
+			.execute();
+		await sql`INSERT INTO visit_collection (id, collect_after) VALUES (1, 0)`.execute(db);
+	},
+	async down(db: Kysely<unknown>): Promise<void> {
+		await db.schema.dropTable('visit_collection').execute();
+		await db.schema.dropIndex('visits_visited_at').execute();
+	}
+};

packages/db/src/migrations/index.ts

@@ -1,7 +1,9 @@
 import { initial } from './0001_visits.js';
 import { addVisitClassification } from './0002_visit_classification.js';
+import { visitRetention } from './0003_visit_retention.js';
 
 export const migrations = {
 	'0001_visits': initial,
-	'0002_visit_classification': addVisitClassification
+	'0002_visit_classification': addVisitClassification,
+	'0003_visit_retention': visitRetention
 };

packages/db/src/node-sqlite-dialect.ts

@@ -1,4 +1,12 @@
-import { SqliteAdapter, SqliteDialect, type SqliteDatabase, type SqliteStatement } from 'kysely';
+import {
+	sql,
+	SqliteAdapter,
+	SqliteDialect,
+	type Kysely,
+	type MigrationLockOptions,
+	type SqliteDatabase,
+	type SqliteStatement
+} from 'kysely';
 import type { DatabaseSync, SQLInputValue } from 'node:sqlite';
 
 export interface NodeSqliteDialectConfig {
@@ -24,6 +32,19 @@ class NodeSqliteAdapter extends SqliteAdapter {
 	override get supportsTransactionalDdl(): boolean {
 		return true;
 	}
+
+	override async acquireMigrationLock(
+		db: Kysely<unknown>,
+		options: MigrationLockOptions
+	): Promise<void> {
+		const name = options.lockTableSchema
+			? `${options.lockTableSchema}.${options.lockTable}`
+			: options.lockTable;
+		const table = sql.table(name);
+		// Reserve the SQLite writer before Kysely reads migration history.
+		const lock = sql`UPDATE ${table} SET is_locked = 0 WHERE id = ${options.lockRowId}`;
+		await lock.execute(db);
+	}
 }
 
 export class NodeSqliteDialect extends SqliteDialect {

packages/db/src/storage.test.ts

@@ -5,7 +5,15 @@ import { DatabaseSync } from 'node:sqlite';
 import { dirname, join } from 'node:path';
 import { Migrator, sql, type Migration } from 'kysely';
 import { afterEach, beforeEach, expect, test, vi } from 'vitest';
-import { insertVisit, openDb, runMigrations, type ClassifiedVisit } from './index.js';
+import {
+	insertVisit,
+	openDb,
+	pruneVisits,
+	purgeVisits,
+	runMigrations,
+	setVisitCollection,
+	type ClassifiedVisit
+} from './index.js';
 import { migrations } from './migrations/index.js';
 
 let directory: string;
@@ -41,7 +49,11 @@ test('opens a real WAL database and applies versioned migrations once', async ()
 		expect((await sql<{ timeout: number }>`PRAGMA busy_timeout`.execute(db)).rows[0]?.timeout).toBe(
 			5000
 		);
-		expect(await runMigrations(db)).toEqual(['0001_visits', '0002_visit_classification']);
+		expect(await runMigrations(db)).toEqual([
+			'0001_visits',
+			'0002_visit_classification',
+			'0003_visit_retention'
+		]);
 		expect(await runMigrations(db)).toEqual([]);
 	} finally {
 		await db.destroy();
@@ -51,7 +63,7 @@ test('opens a real WAL database and applies versioned migrations once', async ()
 test('persists classified visits for the same logical path on different domains', async () => {
 	const path = join(directory, 'visits.sqlite');
 	const writer = await openDb(path);
-	const ids: number[] = [];
+	const ids: (number | undefined)[] = [];
 	try {
 		await runMigrations(writer);
 		ids.push(await insertVisit(writer, visit));
@@ -107,7 +119,7 @@ test('adds classification columns without changing existing visit data', async (
 			VALUES (${visit.domain}, ${visit.path}, ${visit.visited_at}, ${visit.ip}, ${visit.user_agent}, ${visit.accept_language})
 		`.execute(db);
 
-		expect(await runMigrations(db)).toEqual(['0002_visit_classification']);
+		expect(await runMigrations(db)).toEqual(['0002_visit_classification', '0003_visit_retention']);
 		expect(await db.selectFrom('visits').selectAll().executeTakeFirstOrThrow()).toEqual({
 			...visit,
 			id: 1,
@@ -139,13 +151,69 @@ test('a failed migration rolls back schema and bookkeeping so startup can retry'
 		});
 
 		expect((await migrator.migrateToLatest()).error).toBe(failure);
-		expect(await runMigrations(db)).toEqual(['0001_visits', '0002_visit_classification']);
+		expect(await runMigrations(db)).toEqual([
+			'0001_visits',
+			'0002_visit_classification',
+			'0003_visit_retention'
+		]);
 		expect(await runMigrations(db)).toEqual([]);
 	} finally {
 		await db.destroy();
 	}
 });
 
+test('prunes expired visits in bounded passes and zero-retention purge removes the remainder', async () => {
+	const db = await openDb(join(directory, 'retention.sqlite'));
+	try {
+		await runMigrations(db);
+		const cutoff = visit.visited_at;
+		await db
+			.insertInto('visits')
+			.values([
+				...Array.from({ length: 1001 }, () => ({ ...visit, visited_at: cutoff - 1 })),
+				{ ...visit, visited_at: cutoff },
+				{ ...visit, visited_at: cutoff + 1 }
+			])
+			.execute();
+
+		expect(await pruneVisits(db, cutoff)).toBe(true);
+		expect(await db.selectFrom('visits').select('id').execute()).toHaveLength(3);
+		expect(await pruneVisits(db, cutoff)).toBe(false);
+		expect(
+			await db.selectFrom('visits').select('visited_at').orderBy('visited_at').execute()
+		).toEqual([{ visited_at: cutoff }, { visited_at: cutoff + 1 }]);
+		await purgeVisits(db);
+		expect(await db.selectFrom('visits').selectAll().execute()).toEqual([]);
+	} finally {
+		await db.destroy();
+	}
+});
+
+test('a collection fence rejects queued writes after purge and after re-enabling', async () => {
+	const path = join(directory, 'collection.sqlite');
+	const writer = await openDb(path);
+	await runMigrations(writer);
+	const maintenance = await openDb(path);
+	try {
+		await insertVisit(writer, visit);
+		await setVisitCollection(maintenance, false);
+		await purgeVisits(maintenance);
+		expect(await insertVisit(writer, visit)).toBeUndefined();
+		const resumedAt = visit.visited_at + 10;
+		await setVisitCollection(maintenance, true, resumedAt);
+		expect(await insertVisit(writer, visit)).toBeUndefined();
+		expect(await insertVisit(writer, { ...visit, visited_at: resumedAt })).toBeUndefined();
+		await setVisitCollection(maintenance, true, resumedAt + 100);
+		const resumed = { ...visit, visited_at: resumedAt + 1 };
+		expect(await insertVisit(writer, resumed)).toBeTypeOf('number');
+		const rows = await writer.selectFrom('visits').select('visited_at').execute();
+		expect(rows).toEqual([{ visited_at: resumed.visited_at }]);
+	} finally {
+		await maintenance.destroy();
+		await writer.destroy();
+	}
+});
+
 test('destroy closes an opened database without an intervening query', async () => {
 	const close = vi.spyOn(DatabaseSync.prototype, 'close');
 	try {

packages/db/src/types.ts

@@ -12,8 +12,14 @@ export interface VisitTable {
 	os: string;
 }
 
+export interface VisitCollectionTable {
+	id: number;
+	collect_after: number | null;
+}
+
 export interface Database {
 	visits: VisitTable;
+	visit_collection: VisitCollectionTable;
 }
 
 export type ClassifiedVisit = Omit<Insertable<VisitTable>, 'id'>;

packages/db/src/visits.ts

@@ -1,7 +1,62 @@
-import type { Kysely } from 'kysely';
+import { sql, type Kysely } from 'kysely';
 import type { ClassifiedVisit, Database } from './types.js';
 
-export async function insertVisit(db: Kysely<Database>, visit: ClassifiedVisit): Promise<number> {
-	const row = await db.insertInto('visits').values(visit).returning('id').executeTakeFirstOrThrow();
-	return row.id;
+export async function insertVisit(
+	db: Kysely<Database>,
+	visit: ClassifiedVisit
+): Promise<number | undefined> {
+	const insert = sql<{ id: number }>`
+		INSERT INTO visits (domain, path, visited_at, ip, user_agent, accept_language, browser, os)
+		SELECT ${visit.domain}, ${visit.path}, ${visit.visited_at}, ${visit.ip},
+			${visit.user_agent}, ${visit.accept_language}, ${visit.browser}, ${visit.os}
+		FROM visit_collection
+		WHERE id = 1 AND collect_after < ${visit.visited_at}
+		RETURNING id
+	`;
+	const result = await insert.execute(db);
+	return result.rows[0]?.id;
+}
+
+export async function setVisitCollection(
+	db: Kysely<Database>,
+	enabled: boolean,
+	now = Date.now()
+): Promise<void> {
+	const collectAfter = enabled ? sql<number>`coalesce(collect_after, ${now})` : null;
+	await db
+		.updateTable('visit_collection')
+		.set({ collect_after: collectAfter })
+		.where('id', '=', 1)
+		.execute();
+}
+
+export async function pruneVisits(db: Kysely<Database>, before: number): Promise<boolean> {
+	const batchSize = 1000;
+	const expired = db
+		.selectFrom('visits')
+		.select('id')
+		.where('visited_at', '<', before)
+		.orderBy('visited_at')
+		.orderBy('id')
+		.limit(batchSize);
+	const result = await db.deleteFrom('visits').where('id', 'in', expired).executeTakeFirstOrThrow();
+	return result.numDeletedRows === BigInt(batchSize);
+}
+
+export async function purgeVisitsIfDisabled(db: Kysely<Database>): Promise<void> {
+	await db
+		.deleteFrom('visits')
+		.where((expression) => {
+			const disabled = expression
+				.selectFrom('visit_collection')
+				.select('id')
+				.where('id', '=', 1)
+				.where('collect_after', 'is', null);
+			return expression.exists(disabled);
+		})
+		.execute();
+}
+
+export async function purgeVisits(db: Kysely<Database>): Promise<void> {
+	await db.deleteFrom('visits').execute();
 }

pnpm-lock.yaml

@@ -218,6 +218,9 @@ importers:
       '@goto/config':
         specifier: workspace:*
         version: link:../../packages/config
+      '@goto/db':
+        specifier: workspace:*
+        version: link:../../packages/db
       openworkflow:
         specifier: 'catalog:'
         version: 0.11.0