diff --git a/.changeset/finite-analytics-retention.md b/.changeset/finite-analytics-retention.md new file mode 100644 index 0000000..49e072b --- /dev/null +++ b/.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. diff --git a/.github/workflows/worker.yml b/.github/workflows/worker.yml index c8b6fc3..364bbb2 100644 --- a/.github/workflows/worker.yml +++ b/.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 diff --git a/.woodpecker/worker.yml b/.woodpecker/worker.yml index 53e24f9..8fb138f 100644 --- a/.woodpecker/worker.yml +++ b/.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 diff --git a/apps/web/docker-compose.dev.yml b/apps/web/docker-compose.dev.yml index 01fa898..6111137 100644 --- a/apps/web/docker-compose.dev.yml +++ b/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: diff --git a/apps/web/server/adapter.ts b/apps/web/server/adapter.ts index 4be7831..faacc99 100644 --- a/apps/web/server/adapter.ts +++ b/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)) ); diff --git a/apps/web/src/lib/server/application.test.ts b/apps/web/src/lib/server/application.test.ts index 4984d84..1beb0ac 100644 --- a/apps/web/src/lib/server/application.test.ts +++ b/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()); diff --git a/apps/web/src/lib/server/application.ts b/apps/web/src/lib/server/application.ts index 777c2e2..7f966b8 100644 --- a/apps/web/src/lib/server/application.ts +++ b/apps/web/src/lib/server/application.ts @@ -42,28 +42,58 @@ export async function createApplication(options: ConfigOptions = {}): Promise { + 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; + 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() { diff --git a/apps/web/src/lib/server/dispatch.test.ts b/apps/web/src/lib/server/dispatch.test.ts index d96885e..96f84d7 100644 --- a/apps/web/src/lib/server/dispatch.test.ts +++ b/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( diff --git a/apps/web/src/lib/server/dispatch.ts b/apps/web/src/lib/server/dispatch.ts index f90cd78..df72d34 100644 --- a/apps/web/src/lib/server/dispatch.ts +++ b/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; diff --git a/apps/web/src/lib/server/visits/pool.ts b/apps/web/src/lib/server/visits/pool.ts new file mode 100644 index 0000000..43852dd --- /dev/null +++ b/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 { + 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'); +} diff --git a/apps/web/src/lib/server/visits/recorder.test.ts b/apps/web/src/lib/server/visits/recorder.test.ts index 835897c..f68bbc6 100644 --- a/apps/web/src/lib/server/visits/recorder.test.ts +++ b/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(); const poolClosed = Promise.withResolvers(); - 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); diff --git a/apps/web/src/lib/server/visits/recorder.ts b/apps/web/src/lib/server/visits/recorder.ts index 4e74644..a549ffe 100644 --- a/apps/web/src/lib/server/visits/recorder.ts +++ b/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; } +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 | 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 => { + 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 | undefined; - - function stop(): Promise { - 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 }; + }; } diff --git a/apps/worker/e2e/startup.mjs b/apps/worker/e2e/startup.js similarity index 63% rename from apps/worker/e2e/startup.mjs rename to apps/worker/e2e/startup.js index c6d06f1..215797c 100644 --- a/apps/worker/e2e/startup.mjs +++ b/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' ); diff --git a/apps/worker/package.json b/apps/worker/package.json index 8cf83e4..2c06bd2 100644 --- a/apps/worker/package.json +++ b/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:" }, diff --git a/apps/worker/src/services/config-reload.test.ts b/apps/worker/src/services/config-reload.test.ts index 1ad72c4..c6c6334 100644 --- a/apps/worker/src/services/config-reload.test.ts +++ b/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()), 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(); + 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>(); 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); diff --git a/apps/worker/src/services/config-reload.ts b/apps/worker/src/services/config-reload.ts index 816f0c1..7d6b227 100644 --- a/apps/worker/src/services/config-reload.ts +++ b/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; } -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 { + 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 + shutdown: () => Promise, + context: WorkerContext, + engine: OpenWorkflow ): Promise { 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(); diff --git a/apps/worker/src/services/context.ts b/apps/worker/src/services/context.ts new file mode 100644 index 0000000..6a79b32 --- /dev/null +++ b/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>; +} diff --git a/apps/worker/src/services/lifecycle.test.ts b/apps/worker/src/services/lifecycle.test.ts index 1dd518a..18694f5 100644 --- a/apps/worker/src/services/lifecycle.test.ts +++ b/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(); 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(); + 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(); }); diff --git a/apps/worker/src/services/lifecycle.ts b/apps/worker/src/services/lifecycle.ts index 31b9033..59e973b 100644 --- a/apps/worker/src/services/lifecycle.ts +++ b/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; server?: Server; backend?: Pick; - stopWatch?: () => void; + stopWatch?: () => void | Promise; + db?: Pick>, 'destroy'>; } export function createShutdown(resources: ShutdownResources): () => Promise { @@ -17,7 +19,7 @@ export function createShutdown(resources: ShutdownResources): () => Promise { const failures: unknown[] = []; try { - resources.stopWatch?.(); + await resources.stopWatch?.(); } catch (error) { failures.push(error); } @@ -41,5 +43,10 @@ async function stopResources(resources: ShutdownResources): Promise { } catch (error) { failures.push(error); } + try { + await resources.db?.destroy(); + } catch (error) { + failures.push(error); + } if (failures.length > 0) throw failures[0]; } diff --git a/apps/worker/src/services/retention.test.ts b/apps/worker/src/services/retention.test.ts new file mode 100644 index 0000000..cd077fa --- /dev/null +++ b/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 { + 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(); + const requested = Promise.withResolvers(); + 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(); + const release = Promise.withResolvers(); + 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' + ); +}); diff --git a/apps/worker/src/services/retention.ts b/apps/worker/src/services/retention.ts new file mode 100644 index 0000000..f285aea --- /dev/null +++ b/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 { + 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 { + 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 { + 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, 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(); +} diff --git a/apps/worker/src/services/startup.test.ts b/apps/worker/src/services/startup.test.ts index 834c387..1472c29 100644 --- a/apps/worker/src/services/startup.test.ts +++ b/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 { + 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' diff --git a/apps/worker/src/services/startup.ts b/apps/worker/src/services/startup.ts index 23c7167..005cce1 100644 --- a/apps/worker/src/services/startup.ts +++ b/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 { + const db = await openDb(path); + resources.db = db; + await runMigrations(db); + return db; +} + async function startWorkflowWorker( databasePath: string, + context: WorkerContext, resources: ShutdownResources -): Promise { +): Promise { 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( diff --git a/apps/worker/src/workflows/retain-visits.test.ts b/apps/worker/src/workflows/retain-visits.test.ts new file mode 100644 index 0000000..277fc1c --- /dev/null +++ b/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>() + .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(); + const resumed = Promise.withResolvers(); + 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(); +}); diff --git a/apps/worker/src/workflows/retain-visits.ts b/apps/worker/src/workflows/retain-visits.ts new file mode 100644 index 0000000..8a5b642 --- /dev/null +++ b/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(config: { name: string }, fn: () => Promise): Promise; + sleep(name: string, duration: `${number}s`): Promise; +} + +function collectionDisabled(context: WorkerContext): boolean { + return context.config.analytics.retention_months === 0; +} + +async function chooseJitter(): Promise { + 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 { + 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); +} diff --git a/apps/worker/src/workflows/specs.ts b/apps/worker/src/workflows/specs.ts new file mode 100644 index 0000000..29ceb9b --- /dev/null +++ b/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['spec'] = + defineWorkflowSpec({ + name: 'retain-visits', + schedules: [retentionSchedule] + }); diff --git a/examples/goto.yaml b/examples/goto.yaml index 0c3a80e..68e9bf9 100644 --- a/examples/goto.yaml +++ b/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 diff --git a/packages/config/src/config.schema.json b/packages/config/src/config.schema.json index 8767d30..e38a1de 100644 --- a/packages/config/src/config.schema.json +++ b/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": { diff --git a/packages/config/src/load.test.ts b/packages/config/src/load.test.ts index c19f16a..a08c51c 100644 --- a/packages/config/src/load.test.ts +++ b/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' } diff --git a/packages/config/src/monitor.test.ts b/packages/config/src/monitor.test.ts index 6d25c49..60dc665 100644 --- a/packages/config/src/monitor.test.ts +++ b/packages/config/src/monitor.test.ts @@ -14,7 +14,9 @@ function input() { path: '/config/goto.yaml', initial: 'initial', load: vi.fn<(path: string) => Promise>().mockResolvedValue('reconciled'), - onChange: vi.fn<(next: string, previous: string) => boolean>().mockReturnValue(true), + onChange: vi + .fn<(next: string, previous: string) => boolean | Promise>() + .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(); + 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); diff --git a/packages/config/src/monitor.ts b/packages/config/src/monitor.ts index 1ef5435..1bc3954 100644 --- a/packages/config/src/monitor.ts +++ b/packages/config/src/monitor.ts @@ -10,10 +10,10 @@ export async function startConfigReloading({ path: string; initial: T; load: (path: string) => Promise; - onChange: (next: T, previous: T) => boolean; + onChange: (next: T, previous: T) => boolean | Promise; /** Receives raw failures; the caller is responsible for safe diagnostics. */ onError: (error: unknown) => void; -}): Promise<() => void> { +}): Promise<() => Promise> { let accepted = initial; let stopped = false; let pending = false; @@ -26,7 +26,8 @@ export async function startConfigReloading({ 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({ return reload(); } - function stop(): void { - if (stopped) return; - stopped = true; - pending = false; - unwatchFile(path, listener); + async function stop(): Promise { + if (!stopped) { + stopped = true; + pending = false; + unwatchFile(path, listener); + } + await active; } // Poll the path so atomic saves and deletion/recreation remain observable. diff --git a/packages/config/src/schema.ts b/packages/config/src/schema.ts index 9ba05a8..73fb769 100644 --- a/packages/config/src/schema.ts +++ b/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 } ); diff --git a/packages/db/package.json b/packages/db/package.json index 8a1c884..b9afce7 100644 --- a/packages/db/package.json +++ b/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" diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 8e54bb8..15bc07b 100644 --- a/packages/db/src/index.ts +++ b/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({ dialect: new NodeSqliteDialect({ database }) }); await db.connection().execute(async () => undefined); return db; diff --git a/packages/db/src/migrations/0003_visit_retention.ts b/packages/db/src/migrations/0003_visit_retention.ts new file mode 100644 index 0000000..fbb43b5 --- /dev/null +++ b/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): Promise { + 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): Promise { + await db.schema.dropTable('visit_collection').execute(); + await db.schema.dropIndex('visits_visited_at').execute(); + } +}; diff --git a/packages/db/src/migrations/index.ts b/packages/db/src/migrations/index.ts index a3ad836..22d8cf9 100644 --- a/packages/db/src/migrations/index.ts +++ b/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 }; diff --git a/packages/db/src/node-sqlite-dialect.ts b/packages/db/src/node-sqlite-dialect.ts index ff74391..31d6358 100644 --- a/packages/db/src/node-sqlite-dialect.ts +++ b/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, + options: MigrationLockOptions + ): Promise { + 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 { diff --git a/packages/db/src/storage.test.ts b/packages/db/src/storage.test.ts index 8826a0b..95e6e14 100644 --- a/packages/db/src/storage.test.ts +++ b/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 { diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index bb66cf9..b94c5e4 100644 --- a/packages/db/src/types.ts +++ b/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, 'id'>; diff --git a/packages/db/src/visits.ts b/packages/db/src/visits.ts index 5002173..ecfa43d 100644 --- a/packages/db/src/visits.ts +++ b/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, visit: ClassifiedVisit): Promise { - const row = await db.insertInto('visits').values(visit).returning('id').executeTakeFirstOrThrow(); - return row.id; +export async function insertVisit( + db: Kysely, + visit: ClassifiedVisit +): Promise { + 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, + enabled: boolean, + now = Date.now() +): Promise { + const collectAfter = enabled ? sql`coalesce(collect_after, ${now})` : null; + await db + .updateTable('visit_collection') + .set({ collect_after: collectAfter }) + .where('id', '=', 1) + .execute(); +} + +export async function pruneVisits(db: Kysely, before: number): Promise { + 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): Promise { + 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): Promise { + await db.deleteFrom('visits').execute(); } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b28d248..fc12213 100644 --- a/pnpm-lock.yaml +++ b/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