diff --git a/.changeset/adr-0057-lifecycle-p1-p4.md b/.changeset/adr-0057-lifecycle-p1-p4.md new file mode 100644 index 0000000000..0a348ade6e --- /dev/null +++ b/.changeset/adr-0057-lifecycle-p1-p4.md @@ -0,0 +1,20 @@ +--- +'@objectstack/spec': minor +'@objectstack/objectql': minor +'@objectstack/driver-sql': minor +'@objectstack/driver-sqlite-wasm': minor +'@objectstack/platform-objects': minor +'@objectstack/metadata-core': minor +'@objectstack/service-messaging': minor +'@objectstack/service-automation': minor +'@objectstack/plugin-audit': minor +--- + +ADR-0057 data lifecycle P1–P4 (#2786): platform-generated data is now bounded by construction. + +- **P1 — contract**: new `lifecycle` object property (`class: record | audit | telemetry | transient | event` + `retention` / `ttl` / `storage(rotation)` / `archive` / `reclaim`), enforced by the platform-owned **LifecycleService** registered by `ObjectQLPlugin` (default-on; disable via `OS_LIFECYCLE_DISABLED=1` or plugin `lifecycle.enabled=false`). The Reaper batch-deletes rows past `retention.maxAge` / `ttl` under a system context and reclaims space (`SqlDriver.reclaimSpace()` → SQLite `PRAGMA incremental_vacuum`). Non-`record` classes must declare a bounding policy (parse-time invariant + spec-liveness gate + dogfood storage-growth gate). +- **P2 — rotation**: `storage: { strategy: 'rotation', shards, unit }` physically time-shards the table on SQLite — writes land in the current shard, reads go through a UNION-ALL view under the base name, expiry is an O(1) `DROP` of shards past the window. A legacy table is adopted as the first shard on upgrade. Other dialects fall back to an equivalent age-based reap. +- **P3 — separation + Archiver**: registering a datasource named `telemetry` routes telemetry/event/audit objects to it (opt-in by existence; `transient` deliberately stays on the primary). Audit objects with `archive` declared get retain → archive → delete once the archive datasource exists; without it rows are retained, never dropped unarchived. +- **P4 — governance**: new `lifecycle` settings namespace — runtime enable switch, per-object retention overrides (tenant-scoped: regulated tenants set years, dev sets days), per-object/per-class row quotas and growth alerts (observe-and-alert only). + +**Behavior change**: 11 platform objects now carry lifecycle declarations and their telemetry is bounded by default — `sys_activity` 14d (rotated), `sys_audit_log` 90d hot → archive (retained forever until an `archive` datasource is registered), `sys_metadata_audit` 365d → archive, `sys_job_run` / `sys_automation_run` / `sys_http_delivery` 30d, notification pipeline (`sys_notification`, delivery, receipt, inbox) 90d, `sys_device_code` expires_at + 1d. Extend windows per environment/tenant via the `lifecycle.retention_overrides` setting. diff --git a/docs/adr/0057-system-data-lifecycle-and-retention.md b/docs/adr/0057-system-data-lifecycle-and-retention.md index 7313275fbf..2654fc9252 100644 --- a/docs/adr/0057-system-data-lifecycle-and-retention.md +++ b/docs/adr/0057-system-data-lifecycle-and-retention.md @@ -1,6 +1,6 @@ # ADR-0057: System data has a lifecycle — declarative retention, rotation, and reclamation for platform-generated objects -**Status**: Proposed (2026-06-20) +**Status**: Accepted — **P0–P4 implemented** (P0 shipped with this ADR, [#2079](https://github.com/objectstack-ai/framework/pull/2079); P1–P4 in [#2791](https://github.com/objectstack-ai/framework/pull/2791), tracked in [#2786](https://github.com/objectstack-ai/framework/issues/2786)) (proposed 2026-06-20 · implemented 2026-07-10) **Deciders**: ObjectStack Protocol Architects **Builds on**: [ADR-0052](./0052-audit-is-not-the-activity-feed.md) (decomposed audit / activity / collaboration into bounded contexts — this ADR adds the *orthogonal* axis those contexts never specified: **how long each one lives and how its space is reclaimed**), [ADR-0049](./0049-no-unenforced-security-properties.md) (enforce-or-remove — a declared `retention` that drives no sweeper is dead surface; this ADR wires it to a runtime consumer), [ADR-0030](./0030-notification-platform-convergence.md) (notification objects are messaging-owned with their own lifecycle), [ADR-0021](./0021-analytics-dataset-semantic-layer.md) (precedent for moving event-shaped data off the primary OLTP store) **Consumers**: `@objectstack/spec` (`lifecycle` object property + `LifecycleClass`), `@objectstack/objectql` (LifecycleService: Reaper / Rotator / Archiver), `@objectstack/plugin-audit` (audit/activity exclusion of telemetry objects — shipped P0), `@objectstack/driver-sql` (`auto_vacuum=INCREMENTAL` default + incremental-vacuum hook — shipped P0), `@objectstack/dogfood` (storage-growth regression gate), platform spec-liveness gate @@ -164,6 +164,12 @@ activity-logged** (otherwise cleanup re-feeds the tables it is draining — the same self-audit trap ADR-0052 already guards). Aggregate one summary row at most. +Second hard rule (implementation): **an object that declares `archive` is +never hot-deleted before the archive copy succeeded.** No archive datasource +registered ⇒ rows are retained (today's behavior), reported as +`archive-pending` — a compliance ledger cannot be destroyed by declaring a +lifecycle. + ### 3.4 Reclaim — driver space hygiene SQLite driver defaults to `auto_vacuum=INCREMENTAL` (shipped P0); the Reaper @@ -192,15 +198,25 @@ the store can later be swapped for an append-only log / time-series / object store. This is the ADR-0021 dataset-migration pattern applied to event-shaped system data. +*Implemented as*: registering a datasource named **`telemetry`** routes every +`telemetry`/`event`/`audit`-classed object to it (engine `getDriver` step 3 — +after explicit `datasource`/`datasourceMapping`, before the manifest +`defaultDatasource`). Opt-in purely by the datasource's existence. +`transient` deliberately stays on the primary: those objects are user-session +data and some (better-auth's `sys_device_code`) are accessed outside the +engine — splitting their storage would split their brain. + ## 4. Rollout +P1–P4 are tracked in [#2786](https://github.com/objectstack-ai/framework/issues/2786). + | Phase | Scope | Status | |---|---|---| -| **P0 — stop the bleed** | (a) exclude operational/plumbing objects from the audit+activity writer (`plugin-audit` `SKIP_OBJECTS`); (b) SQLite `auto_vacuum=INCREMENTAL` driver default; (c) showcase digest interval 20s→60s, flagged demo-only | **this ADR ships P0** | -| **P1 — contract** | `lifecycleClass` + `lifecycle` spec; LifecycleService Reaper; spec-liveness enforcement; dogfood growth gate | proposed | -| **P2 — rotation/TTL** | Rotator (shard + DROP) for high-freq telemetry; transient TTL expiry | proposed | -| **P3 — separation** | telemetry/audit on a dedicated datasource; Archiver cold-store | proposed | -| **P4 — governance** | per-table storage quotas, growth alerts, tenant-level retention overrides | proposed | +| **P0 — stop the bleed** | (a) exclude operational/plumbing objects from the audit+activity writer (`plugin-audit` `SKIP_OBJECTS`); (b) SQLite `auto_vacuum=INCREMENTAL` driver default; (c) showcase digest interval 20s→60s, flagged demo-only | **shipped with this ADR ([#2079](https://github.com/objectstack-ai/framework/pull/2079))** | +| **P1 — contract** | `lifecycleClass` + `lifecycle` spec; LifecycleService Reaper; spec-liveness enforcement; dogfood growth gate | **implemented ([#2791](https://github.com/objectstack-ai/framework/pull/2791))** | +| **P2 — rotation/TTL** | Rotator (shard + DROP) for high-freq telemetry; transient TTL expiry | **implemented ([#2791](https://github.com/objectstack-ai/framework/pull/2791))** | +| **P3 — separation** | telemetry/audit on a dedicated datasource; Archiver cold-store | **implemented ([#2791](https://github.com/objectstack-ai/framework/pull/2791))** | +| **P4 — governance** | per-table storage quotas, growth alerts, tenant-level retention overrides | **implemented ([#2791](https://github.com/objectstack-ai/framework/pull/2791))** | P0 alone removes ~76 % of the row growth (audit/activity exclusion) and a further 3× (interval), and lets space be reclaimed — without any schema change diff --git a/packages/dogfood/test/storage-growth.dogfood.test.ts b/packages/dogfood/test/storage-growth.dogfood.test.ts new file mode 100644 index 0000000000..cc973f2c11 --- /dev/null +++ b/packages/dogfood/test/storage-growth.dogfood.test.ts @@ -0,0 +1,264 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// STORAGE GROWTH — ADR-0057 data lifecycle, exercised end-to-end through a real +// showcase boot (#2786 P1). The 260 MB dev.db regression happened because +// platform-generated rows had NO retention contract: every append-only table +// grew forever. This gate makes that class of regression revert-provable: +// +// @proof: adr0057-lifecycle-bounded-growth +// +// • CONTRACT — every registered object declaring a non-`record` lifecycle +// class carries a bounding policy (retention / ttl / rotation), and the +// shipped sys_* declarations are present in a real boot (deleting one +// turns this red). +// • REAPER — rows older than the declared window are deleted by +// `lifecycle.sweep()`; fresh rows and record-class/business data are +// untouched; space reclaim runs on the touched datasource. +// • ARCHIVE SAFETY — an audit-class object with `archive` declared is NEVER +// hot-deleted before the Archiver has copied it out (no archive +// datasource registered ⇒ rows are retained, not dropped). +// +// The runaway writer is simulated with direct driver inserts (backdated +// `created_at`) rather than waiting on scheduled flow ticks — deterministic, +// and it exercises the exact path the Reaper sweeps. + +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { bootStack, type VerifyStack } from '@objectstack/verify'; +import showcaseStack from '@objectstack/example-showcase'; + +const DAY_MS = 86_400_000; + +interface EngineLike { + registry: { + getAllObjects(): Array<{ name: string; lifecycle?: Record }>; + registerObject(obj: Record): void; + getObject(name: string): { name: string; lifecycle?: Record } | undefined; + }; + syncObjectSchema(name: string): Promise; + getDriverForObject(name: string): DriverLike | undefined; +} + +interface DriverLike { + create(object: string, data: Record): Promise>; + count(object: string, query?: Record): Promise; + reclaimSpace?(): Promise; +} + +interface LifecycleLike { + sweep(): Promise<{ + swept: Array<{ object: string; policy: string; deleted?: number }>; + skipped: Array<{ object: string; reason: string }>; + errors: Array<{ object: string; error: string }>; + reclaimed: string[]; + }>; +} + +describe('objectstack verify LIFECYCLE (ADR-0057): declared policies bound growth (#adr0057-lifecycle-bounded-growth)', () => { + let stack: VerifyStack; + let engine: EngineLike; + let lifecycle: LifecycleLike; + let driver: DriverLike; + + // Platform storage convention: `created_at` is registry-injected as a + // `datetime` field, stored as epoch-ms INTEGER on SQLite (a JS Date through + // the driver) — and filter cutoffs are coerced to match. Backdate with Date + // objects, exactly like the real write paths, or the rows land as TEXT and + // no temporal filter ever matches them. + const backdated = (days: number) => new Date(Date.now() - days * DAY_MS); + + beforeAll(async () => { + stack = await bootStack(showcaseStack); + engine = stack.kernel.getService('objectql') as unknown as EngineLike; + lifecycle = stack.kernel.getService('lifecycle') as unknown as LifecycleLike; + expect(lifecycle?.sweep, 'the ObjectQLPlugin must register the ADR-0057 lifecycle service').toBeTruthy(); + + // Fixture objects: a telemetry stream, a record-class sibling, and an + // audit ledger with a declared (but unresolvable) archive target. + engine.registry.registerObject({ + name: 'growth_probe_event', + label: 'Growth Probe Event', + lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } }, + fields: { payload: { type: 'text', label: 'Payload' } }, + }); + engine.registry.registerObject({ + name: 'growth_probe_record', + label: 'Growth Probe Record', + lifecycle: { class: 'record' }, + fields: { payload: { type: 'text', label: 'Payload' } }, + }); + engine.registry.registerObject({ + name: 'growth_probe_ledger', + label: 'Growth Probe Ledger', + lifecycle: { + class: 'audit', + retention: { maxAge: '90d' }, + archive: { after: '90d', to: 'archive_missing' }, + }, + fields: { payload: { type: 'text', label: 'Payload' } }, + }); + await engine.syncObjectSchema('growth_probe_event'); + await engine.syncObjectSchema('growth_probe_record'); + await engine.syncObjectSchema('growth_probe_ledger'); + + driver = engine.getDriverForObject('growth_probe_event') as DriverLike; + expect(driver?.create, 'a driver must back the probe objects').toBeTruthy(); + }, 60_000); + + afterAll(async () => { + await stack?.stop(); + }); + + it('CONTRACT: every non-record lifecycle declaration in a real boot carries a bounding policy', () => { + const declared = engine.registry + .getAllObjects() + .filter((o) => o.lifecycle && o.lifecycle.class !== 'record'); + + // A platform boot with ZERO lifecycle-declared objects would mean the + // shipped sys_* annotations were dropped — exactly the regression this + // gate exists to catch. + expect(declared.length, 'no lifecycle-declared objects registered — sys_* annotations were dropped').toBeGreaterThan(0); + + for (const obj of declared) { + const lc = obj.lifecycle!; + expect( + Boolean(lc.retention || lc.ttl || lc.storage), + `${obj.name} declares lifecycle.class='${lc.class}' with NO bounding policy (retention/ttl/storage) — ADR-0057 §3.5`, + ).toBe(true); + } + }); + + it('CONTRACT: the shipped sys_metadata_audit declaration survives (audit + archive-then-delete)', () => { + const meta = engine.registry.getObject('sys_metadata_audit'); + expect(meta, 'sys_metadata_audit must be registered in a platform boot').toBeTruthy(); + expect(meta!.lifecycle?.class).toBe('audit'); + expect(meta!.lifecycle?.retention?.maxAge).toBe('365d'); + expect(meta!.lifecycle?.archive?.to).toBe('archive'); + }); + + it('REAPER: past-window telemetry is reaped, fresh telemetry and record-class data survive, space is reclaimed', async () => { + // Simulate a runaway writer: 40 telemetry rows spread over ~200 days, + // 10 of them inside the 30d window; plus record-class rows older than + // any window (must never be touched). + for (let i = 0; i < 40; i++) { + await driver.create('growth_probe_event', { + payload: `tick-${i}`, + created_at: backdated(5 + i * 5), // 5d … 200d + }); + } + for (let i = 0; i < 5; i++) { + await driver.create('growth_probe_record', { + payload: `business-${i}`, + created_at: backdated(400), + }); + } + + const before = await driver.count('growth_probe_event', { object: 'growth_probe_event' }); + expect(before).toBe(40); + + const report = await lifecycle.sweep(); + + // Errors first — a failed delete otherwise shows up as a confusing + // "expected 40 to be 5" count mismatch. + expect(report.errors, `sweep reported errors: ${JSON.stringify(report.errors)}`).toEqual([]); + expect( + report.swept.map((e) => e.object), + `sweep applied no policy — report: ${JSON.stringify(report)}`, + ).toContain('growth_probe_event'); + + // The telemetry table is now bounded by its declared 30d window: + // rows at 5,10,15,20,25d survive (5 rows), everything older is gone. + const after = await driver.count('growth_probe_event', { object: 'growth_probe_event' }); + expect(after, 'telemetry rows past retention.maxAge must be reaped').toBe(5); + + // Record-class/business data is sacrosanct — same age, still alive. + const records = await driver.count('growth_probe_record', { object: 'growth_probe_record' }); + expect(records, 'record-class rows must NEVER be reaped').toBe(5); + + // The sweep reported the reap and reclaimed the datasource. + const entry = report.swept.find((e) => e.object === 'growth_probe_event'); + expect(entry?.policy).toBe('retention'); + expect(report.errors).toEqual([]); + expect(report.reclaimed.length, 'reclaimSpace must run on the touched datasource').toBeGreaterThan(0); + }); + + it('ROTATOR (P2): a rotation-declared stream is physically sharded through the real engine → driver path', async () => { + engine.registry.registerObject({ + name: 'growth_probe_stream', + label: 'Growth Probe Stream', + lifecycle: { + class: 'telemetry', + storage: { strategy: 'rotation', shards: 3, unit: 'day' }, + }, + fields: { payload: { type: 'text', label: 'Payload' } }, + }); + await engine.syncObjectSchema('growth_probe_stream'); + + // The base name is now a read view; writes land in the current shard. + const streamDriver = engine.getDriverForObject('growth_probe_stream') as DriverLike & { + execute(sql: string): Promise; + }; + const master = (await streamDriver.execute( + "SELECT name, type FROM sqlite_master WHERE name LIKE 'growth_probe_stream%' AND name NOT LIKE '%autoindex%'", + )) as Array<{ name: string; type: string }>; + const types = Object.fromEntries(master.map((r) => [r.name, r.type])); + expect(types['growth_probe_stream'], 'rotation must turn the base name into a view').toBe('view'); + const shardNames = master.filter((r) => /__r\d{6,8}$/.test(r.name)).map((r) => r.name); + expect(shardNames.length, 'a current shard table must exist').toBeGreaterThan(0); + + // Round-trip: write through the driver, read through the view, and a + // sweep applies the 'rotation' policy without touching the rows inside + // the window. + await streamDriver.create('growth_probe_stream', { payload: 'tick' }); + expect(await streamDriver.count('growth_probe_stream', { object: 'growth_probe_stream' })).toBe(1); + + const report = await lifecycle.sweep(); + const entry = report.swept.find((e) => e.object === 'growth_probe_stream'); + expect(entry?.policy).toBe('rotation'); + expect(await streamDriver.count('growth_probe_stream', { object: 'growth_probe_stream' })).toBe(1); + }); + + it('ARCHIVE SAFETY: an audit ledger with a declared archive is never hot-deleted unarchived', async () => { + for (let i = 0; i < 3; i++) { + await driver.create('growth_probe_ledger', { + payload: `ledger-${i}`, + created_at: backdated(365), // far past the 90d hot window + }); + } + + const report = await lifecycle.sweep(); + + // No archive datasource named 'archive_missing' exists ⇒ the rows are + // RETAINED (today's behavior), not dropped. Compliance data cannot be + // destroyed by declaring a lifecycle. + const ledger = await driver.count('growth_probe_ledger', { object: 'growth_probe_ledger' }); + expect(ledger, 'archive-declared audit rows must be retained until archived').toBe(3); + expect(report.skipped).toContainEqual({ object: 'growth_probe_ledger', reason: 'archive-pending' }); + }); + + it('ARCHIVER (P3): once the archive datasource exists, cold rows move there and leave the hot store', async () => { + // Provision a real second SQL store under the datasource name the ledger + // declares, then re-run the sweep: retain → archive → delete. + const { SqlDriver } = await import('@objectstack/driver-sql'); + const cold = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: ':memory:' }, + useNullAsDefault: true, + } as any); + Object.defineProperty(cold, 'name', { value: 'archive_missing' }); + await cold.connect(); + (engine as unknown as { registerDriver(d: unknown): void }).registerDriver(cold); + + const report = await lifecycle.sweep(); + + const entry = report.swept.find((e) => e.object === 'growth_probe_ledger'); + expect(entry?.policy).toBe('archive'); + expect((entry as { archived?: number })?.archived).toBe(3); + + // Hot store drained, cold store holds the ledger. + const hot = await driver.count('growth_probe_ledger', { object: 'growth_probe_ledger' }); + expect(hot, 'archived rows must leave the hot store').toBe(0); + const coldRows = await cold.count('growth_probe_ledger', { object: 'growth_probe_ledger' }); + expect(coldRows, 'archived rows must land in the cold store').toBe(3); + await cold.disconnect(); + }); +}); diff --git a/packages/metadata-core/src/objects/sys-metadata-audit.object.ts b/packages/metadata-core/src/objects/sys-metadata-audit.object.ts index a35f3e3820..115ec732be 100644 --- a/packages/metadata-core/src/objects/sys-metadata-audit.object.ts +++ b/packages/metadata-core/src/objects/sys-metadata-audit.object.ts @@ -37,6 +37,14 @@ export const SysMetadataAuditObject = ObjectSchema.create({ icon: 'shield-check', isSystem: true, managedBy: 'append-only', + // ADR-0057: compliance ledger for metadata changes — hot 365d, then + // archive-then-delete. Without an 'archive' datasource rows are retained + // forever (the Reaper never hot-deletes archive-declared objects). + lifecycle: { + class: 'audit', + retention: { maxAge: '365d' }, + archive: { after: '365d', to: 'archive' }, + }, description: 'Append-only audit trail of metadata write decisions (ADR-0010).', fields: { diff --git a/packages/objectql/src/engine-lifecycle-datasource.test.ts b/packages/objectql/src/engine-lifecycle-datasource.test.ts new file mode 100644 index 0000000000..de9a6eebfe --- /dev/null +++ b/packages/objectql/src/engine-lifecycle-datasource.test.ts @@ -0,0 +1,123 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * ADR-0057 §3.6 (P3) — lifecycle-class datasource separation. + * + * When a datasource named 'telemetry' is registered, objects whose + * `lifecycle.class` is telemetry / event / audit route to it — so + * platform-generated growth can never again pollute the business DB. The + * routing is opt-in by the datasource's existence: without it, resolution is + * exactly what it was before. `transient` and `record` classes always stay on + * their normal resolution (transient objects are user-session data, and some + * are accessed outside the engine — splitting their storage would split + * their brain). + */ + +import { describe, it, expect, beforeEach } from 'vitest'; +import { ObjectQL } from './engine.js'; + +function stubDriver(name: string) { + return { + name, + version: '0.0.0', + supports: {}, + async connect() {}, + async disconnect() {}, + async checkHealth() { + return true; + }, + async execute() { + return null; + }, + async find() { + return []; + }, + findStream() { + throw new Error('ns'); + }, + async findOne() { + return null; + }, + async create(_o: string, d: Record) { + return d; + }, + async update() { + return {}; + }, + async upsert() { + return {}; + }, + async delete() { + return true; + }, + async count() { + return 0; + }, + async bulkCreate() { + return []; + }, + async bulkUpdate() { + return []; + }, + async bulkDelete() {}, + async beginTransaction() { + return {}; + }, + async commit() {}, + async rollback() {}, + } as any; +} + +const OBJECTS = [ + { name: 'biz_account', fields: {} }, + { name: 'probe_activity', lifecycle: { class: 'telemetry', retention: { maxAge: '14d' } }, fields: {} }, + { name: 'probe_bus_event', lifecycle: { class: 'event', ttl: { field: 'created_at', expireAfter: '6h' } }, fields: {} }, + { name: 'probe_ledger', lifecycle: { class: 'audit', retention: { maxAge: '90d' } }, fields: {} }, + { name: 'probe_receipt', lifecycle: { class: 'transient', ttl: { field: 'created_at', expireAfter: '7d' } }, fields: {} }, +]; + +describe('lifecycle-class datasource separation (ADR-0057 §3.6)', () => { + let engine: ObjectQL; + let primary: any; + + beforeEach(async () => { + engine = new ObjectQL(); + primary = stubDriver('memory'); + engine.registerDriver(primary, true); + await engine.init(); + for (const o of OBJECTS) engine.registry.registerObject(o as any); + }); + + it('without a telemetry datasource, every object resolves exactly as before', () => { + for (const o of OBJECTS) { + expect(engine.getDriverForObject(o.name)).toBe(primary); + } + }); + + it('with a telemetry datasource, telemetry/event/audit route to it — record/transient stay put', () => { + const telemetry = stubDriver(ObjectQL.LIFECYCLE_DATASOURCE); + engine.registerDriver(telemetry); + + expect(engine.getDriverForObject('probe_activity')).toBe(telemetry); + expect(engine.getDriverForObject('probe_bus_event')).toBe(telemetry); + expect(engine.getDriverForObject('probe_ledger')).toBe(telemetry); + + expect(engine.getDriverForObject('biz_account')).toBe(primary); + expect(engine.getDriverForObject('probe_receipt')).toBe(primary); + }); + + it("an object's explicit datasource still wins over lifecycle routing", () => { + const telemetry = stubDriver(ObjectQL.LIFECYCLE_DATASOURCE); + const special = stubDriver('special'); + engine.registerDriver(telemetry); + engine.registerDriver(special); + engine.registry.registerObject({ + name: 'probe_pinned', + datasource: 'special', + lifecycle: { class: 'telemetry', retention: { maxAge: '14d' } }, + fields: {}, + } as any); + + expect(engine.getDriverForObject('probe_pinned')).toBe(special); + }); +}); diff --git a/packages/objectql/src/engine.ts b/packages/objectql/src/engine.ts index 904951b79f..d68239a851 100644 --- a/packages/objectql/src/engine.ts +++ b/packages/objectql/src/engine.ts @@ -1494,14 +1494,24 @@ export class ObjectQL implements IDataEngine { return StorageNameMapping.resolveTableName({ name }); } + /** + * Name of the dedicated datasource lifecycle-classed system data prefers + * when one is registered (ADR-0057 §3.6 / P3). Purely opt-in by the + * datasource's existence — no `telemetry` driver registered ⇒ resolution + * is exactly what it was before. + */ + static readonly LIFECYCLE_DATASOURCE = 'telemetry'; + /** * Helper to get the target driver * * Resolution priority (first match wins): * 1. Object's explicit `datasource` field (if not 'default') * 2. DatasourceMapping rules (namespace/package/pattern matching) - * 3. Package's `defaultDatasource` from manifest - * 4. Global default driver + * 3. Lifecycle-class separation (ADR-0057 §3.6): telemetry/event/audit + * objects route to the dedicated 'telemetry' datasource when registered + * 4. Package's `defaultDatasource` from manifest + * 5. Global default driver */ private getDriver(objectName: string): IDataDriver { const object = this._registry.getObject(objectName); @@ -1524,7 +1534,22 @@ export class ObjectQL implements IDataEngine { return this.drivers.get(mappedDatasource)!; } - // 3. Check package's defaultDatasource + // 3. Lifecycle-class separation (ADR-0057 §3.6): high-frequency + // platform-generated data (telemetry / event) and the audit ledger move + // to a dedicated 'telemetry' datasource when one is registered, so their + // growth can never again pollute the business DB. `transient` stays on + // the primary deliberately: those objects are user-session data, and + // some (e.g. better-auth's sys_device_code) are also accessed outside + // the engine — splitting their storage would split their brain. + const lifecycleClass = (object as { lifecycle?: { class?: string } } | undefined)?.lifecycle?.class; + if ( + (lifecycleClass === 'telemetry' || lifecycleClass === 'event' || lifecycleClass === 'audit') && + this.drivers.has(ObjectQL.LIFECYCLE_DATASOURCE) + ) { + return this.drivers.get(ObjectQL.LIFECYCLE_DATASOURCE)!; + } + + // 4. Check package's defaultDatasource // Use the object's FQN name (from getObject) for ownership lookup const fqn = object?.name || objectName; const owner = this._registry.getObjectOwner(fqn); @@ -1542,7 +1567,7 @@ export class ObjectQL implements IDataEngine { } } - // 4. Fallback to global default driver + // 5. Fallback to global default driver if (this.defaultDriver && this.drivers.has(this.defaultDriver)) { return this.drivers.get(this.defaultDriver)!; } diff --git a/packages/objectql/src/index.ts b/packages/objectql/src/index.ts index 23fcfbbb41..55aed4b61a 100644 --- a/packages/objectql/src/index.ts +++ b/packages/objectql/src/index.ts @@ -49,6 +49,24 @@ export type { HookSkipReason, } from './hook-metrics.js'; +// Export LifecycleService (ADR-0057 — declarative retention/rotation/archival) +export { + LifecycleService, + DEFAULT_LIFECYCLE_SWEEP_MS, + DEFAULT_LIFECYCLE_INITIAL_DELAY_MS, +} from './lifecycle/lifecycle-service.js'; +export type { + LifecycleServiceOptions, + LifecycleSweepReport, + LifecycleSweepEntry, + LifecycleEngineLike, + LifecycleObjectLike, + LifecycleSettingsLike, + LifecycleGovernanceAlert, +} from './lifecycle/lifecycle-service.js'; +export { parseLifecycleDuration } from './lifecycle/duration.js'; +export { lifecycleSettingsManifest } from './lifecycle/lifecycle-settings.js'; + // Export MetadataFacade export { MetadataFacade } from './metadata-facade.js'; diff --git a/packages/objectql/src/lifecycle/duration.ts b/packages/objectql/src/lifecycle/duration.ts new file mode 100644 index 0000000000..69037e2f96 --- /dev/null +++ b/packages/objectql/src/lifecycle/duration.ts @@ -0,0 +1,34 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * Duration literal parsing for ADR-0057 lifecycle policies. + * + * The spec (`LifecycleSchema`, `@objectstack/spec`) validates the literal + * shape (`/^\d+(h|d|w|y)$/`); this is the single runtime consumer that turns + * it into milliseconds. Calendar-exact arithmetic is deliberately avoided: + * retention windows are coarse operational bounds, so `y` is 365 days. + */ +const UNIT_MS: Record = { + h: 3_600_000, + d: 86_400_000, + w: 7 * 86_400_000, + y: 365 * 86_400_000, +}; + +export const LIFECYCLE_DURATION_REGEX = /^(\d+)(h|d|w|y)$/; + +/** + * Parse a lifecycle duration literal (`'6h'`, `'14d'`, `'12w'`, `'7y'`) to + * milliseconds. Throws on malformed input — declarations reach the runtime + * already validated by the spec, so a parse failure here is a programming + * error, not user input. + */ +export function parseLifecycleDuration(literal: string): number { + const m = LIFECYCLE_DURATION_REGEX.exec(literal); + if (!m) { + throw new Error( + `[lifecycle] invalid duration literal '${literal}' — expected with unit h|d|w|y (e.g. '14d')`, + ); + } + return Number(m[1]) * UNIT_MS[m[2]]; +} diff --git a/packages/objectql/src/lifecycle/lifecycle-service.test.ts b/packages/objectql/src/lifecycle/lifecycle-service.test.ts new file mode 100644 index 0000000000..b7a26c79ca --- /dev/null +++ b/packages/objectql/src/lifecycle/lifecycle-service.test.ts @@ -0,0 +1,559 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import { describe, it, expect, vi } from 'vitest'; +import { LifecycleService, type LifecycleObjectLike } from './lifecycle-service.js'; +import { parseLifecycleDuration } from './duration.js'; + +const FIXED_NOW = 1_700_000_000_000; // fixed clock for deterministic cutoffs + +function silentLogger() { + return { info: () => {}, warn: () => {}, debug: () => {} }; +} + +/** Fake engine capturing every bulk delete, with a declarable object set. */ +function captureEngine( + objects: LifecycleObjectLike[], + opts: { + deleteImpl?: (object: string, options: any) => any; + driver?: Record; + datasources?: Record; + } = {}, +) { + const deletes: Array<{ object: string; where: any; multi: any; context: any }> = []; + const engine = { + registry: { getAllObjects: () => objects }, + async delete(object: string, options: any) { + deletes.push({ object, where: options?.where, multi: options?.multi, context: options?.context }); + return opts.deleteImpl ? opts.deleteImpl(object, options) : { deletedCount: 3 }; + }, + getDriverForObject: () => opts.driver, + datasource(name: string) { + const ds = opts.datasources?.[name]; + if (!ds) throw new Error(`[ObjectQL] Datasource '${name}' not found`); + return ds; + }, + }; + return { engine, deletes }; +} + +function service(engine: any, extra: Partial[0]> = {}) { + return new LifecycleService({ + getEngine: () => engine, + logger: silentLogger(), + now: () => FIXED_NOW, + initialDelayMs: 1, + sweepIntervalMs: 10, + ...extra, + }); +} + +const isoCutoff = (literal: string) => new Date(FIXED_NOW - parseLifecycleDuration(literal)).toISOString(); + +describe('parseLifecycleDuration', () => { + it('parses the ADR unit set', () => { + expect(parseLifecycleDuration('6h')).toBe(6 * 3_600_000); + expect(parseLifecycleDuration('14d')).toBe(14 * 86_400_000); + expect(parseLifecycleDuration('2w')).toBe(14 * 86_400_000); + expect(parseLifecycleDuration('7y')).toBe(7 * 365 * 86_400_000); + }); + + it('throws on malformed literals', () => { + for (const bad of ['', '14', 'd', '14 days', '2mo', '1.5d']) { + expect(() => parseLifecycleDuration(bad)).toThrow(); + } + }); +}); + +describe('LifecycleService.sweep — Reaper', () => { + it('reaps telemetry by retention.maxAge with an ISO cutoff on created_at, multi + system context', async () => { + const { engine, deletes } = captureEngine([ + { name: 'sys_job_run', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } }, + ]); + + const report = await service(engine).sweep(); + + expect(deletes).toHaveLength(1); + expect(deletes[0].object).toBe('sys_job_run'); + expect(deletes[0].multi).toBe(true); + expect(deletes[0].context).toEqual({ isSystem: true, positions: [], permissions: [] }); + expect(deletes[0].where).toEqual({ created_at: { $lt: isoCutoff('30d') } }); + // ISO-8601 string, never a bare epoch-ms number (Postgres timestamp columns). + expect(deletes[0].where.created_at.$lt).toMatch(/^\d{4}-\d{2}-\d{2}T/); + + expect(report.swept).toEqual([ + { object: 'sys_job_run', class: 'telemetry', policy: 'retention', cutoff: isoCutoff('30d'), deleted: 3 }, + ]); + expect(report.errors).toEqual([]); + }); + + it('reaps transient rows by ttl on the declared field', async () => { + const { engine, deletes } = captureEngine([ + { name: 'sys_device_code', lifecycle: { class: 'transient', ttl: { field: 'expires_at', expireAfter: '1d' } } }, + ]); + + const report = await service(engine).sweep(); + + expect(deletes).toHaveLength(1); + expect(deletes[0].where).toEqual({ expires_at: { $lt: isoCutoff('1d') } }); + expect(report.swept[0].policy).toBe('ttl'); + }); + + it('never touches record-class or undeclared objects', async () => { + const { engine, deletes } = captureEngine([ + { name: 'crm_account' }, + { name: 'crm_invoice', lifecycle: { class: 'record' } }, + ]); + + const report = await service(engine).sweep(); + + expect(deletes).toHaveLength(0); + expect(report.swept).toEqual([]); + }); + + it('skips hot deletion entirely while an archive is declared (retain → archive → delete)', async () => { + const { engine, deletes } = captureEngine([ + { + name: 'sys_audit_log', + lifecycle: { + class: 'audit', + retention: { maxAge: '90d' }, + archive: { after: '90d', to: 'archive', keep: '7y' }, + }, + }, + ]); + + const report = await service(engine).sweep(); + + // A compliance ledger must never be dropped unarchived: with `archive` + // declared and no Archiver run, the Reaper must not delete a single row. + expect(deletes).toHaveLength(0); + expect(report.skipped).toEqual([{ object: 'sys_audit_log', reason: 'archive-pending' }]); + }); + + it('reaps audit-class rows when only retention is declared (explicit delete-after-window)', async () => { + const { engine, deletes } = captureEngine([ + { name: 'sys_metadata_audit', lifecycle: { class: 'audit', retention: { maxAge: '365d' } } }, + ]); + + await service(engine).sweep(); + + expect(deletes).toHaveLength(1); + expect(deletes[0].where).toEqual({ created_at: { $lt: isoCutoff('365d') } }); + }); + + it('bounds rotation-declared objects by shards × unit until the Rotator shards physically', async () => { + const { engine, deletes } = captureEngine([ + { + name: 'sys_activity', + lifecycle: { class: 'telemetry', storage: { strategy: 'rotation', shards: 14, unit: 'day' } }, + }, + ]); + + const report = await service(engine).sweep(); + + expect(deletes).toHaveLength(1); + expect(deletes[0].where).toEqual({ created_at: { $lt: isoCutoff('14d') } }); + expect(report.swept[0].policy).toBe('rotation-fallback'); + }); + + it('rotates physically when the driver supports it — no fallback age reap, reclaim on dropped shards', async () => { + const rotateShards = vi.fn(async () => ({ + object: 'sys_activity', + current: 'sys_activity__r20260710', + shards: ['sys_activity__r20260710'], + dropped: ['sys_activity__r20260626'], + })); + const reclaimSpace = vi.fn(async () => {}); + const driver = { name: 'default', supportsRotation: true, rotateShards, reclaimSpace }; + const obj = { + name: 'sys_activity', + lifecycle: { class: 'telemetry' as const, storage: { strategy: 'rotation' as const, shards: 14, unit: 'day' as const } }, + }; + const { engine, deletes } = captureEngine([obj], { driver }); + + const report = await service(engine).sweep(); + + expect(rotateShards).toHaveBeenCalledWith(obj, FIXED_NOW); + // Rotation replaces the fallback age reap entirely (no retention declared). + expect(deletes).toHaveLength(0); + expect(report.swept).toEqual([ + { + object: 'sys_activity', + class: 'telemetry', + policy: 'rotation', + cutoff: isoCutoff('14d'), + droppedShards: 1, + }, + ]); + // A dropped shard freed pages — the datasource gets an incremental vacuum. + expect(reclaimSpace).toHaveBeenCalledTimes(1); + }); + + it('an explicit retention still trims inside the live shards after rotation', async () => { + const rotateShards = vi.fn(async () => ({ + object: 'sys_activity', + current: 'sys_activity__r20260710', + shards: ['sys_activity__r20260710'], + dropped: [], + })); + const driver = { name: 'default', supportsRotation: true, rotateShards }; + const { engine, deletes } = captureEngine( + [ + { + name: 'sys_activity', + lifecycle: { + class: 'telemetry', + retention: { maxAge: '14d' }, + storage: { strategy: 'rotation', shards: 14, unit: 'day' }, + }, + }, + ], + { driver }, + ); + + const report = await service(engine).sweep(); + + expect(rotateShards).toHaveBeenCalledTimes(1); + expect(deletes).toHaveLength(1); + expect(deletes[0].where).toEqual({ created_at: { $lt: isoCutoff('14d') } }); + expect(report.swept.map((e) => e.policy)).toEqual(['rotation', 'retention']); + }); + + it('prefers explicit retention over the rotation fallback window', async () => { + const { engine, deletes } = captureEngine([ + { + name: 'sys_activity', + lifecycle: { + class: 'telemetry', + retention: { maxAge: '10d' }, + storage: { strategy: 'rotation', shards: 14, unit: 'day' }, + }, + }, + ]); + + const report = await service(engine).sweep(); + + expect(deletes).toHaveLength(1); + expect(deletes[0].where).toEqual({ created_at: { $lt: isoCutoff('10d') } }); + expect(report.swept[0].policy).toBe('retention'); + }); + + it('isolates a failing object — other policies still run, error lands in the report', async () => { + const { engine, deletes } = captureEngine( + [ + { name: 'bad_object', lifecycle: { class: 'telemetry', retention: { maxAge: '7d' } } }, + { name: 'good_object', lifecycle: { class: 'telemetry', retention: { maxAge: '7d' } } }, + ], + { + deleteImpl: (object) => { + if (object === 'bad_object') throw new Error('no such table'); + return { deletedCount: 2 }; + }, + }, + ); + + const report = await service(engine).sweep(); + + expect(deletes.map((d) => d.object)).toEqual(['bad_object', 'good_object']); + expect(report.errors).toEqual([{ object: 'bad_object', error: 'no such table' }]); + expect(report.swept.map((e) => e.object)).toEqual(['good_object']); + }); + + it('no-ops without an engine', async () => { + const svc = new LifecycleService({ getEngine: () => undefined, logger: silentLogger() }); + const report = await svc.sweep(); + expect(report.swept).toEqual([]); + }); + + it('no-ops when disabled via option or OS_LIFECYCLE_DISABLED', async () => { + const { engine, deletes } = captureEngine([ + { name: 'sys_job_run', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } }, + ]); + + const disabled = service(engine, { enabled: false }); + await disabled.sweep(); + expect(deletes).toHaveLength(0); + + process.env.OS_LIFECYCLE_DISABLED = '1'; + try { + await service(engine).sweep(); + expect(deletes).toHaveLength(0); + } finally { + delete process.env.OS_LIFECYCLE_DISABLED; + } + }); +}); + +describe('LifecycleService.sweep — Archiver (P3)', () => { + const AUDIT_OBJ: LifecycleObjectLike = { + name: 'sys_audit_log', + lifecycle: { + class: 'audit', + retention: { maxAge: '90d' }, + archive: { after: '90d', to: 'archive', keep: '7y' }, + } as any, + }; + + function coldStore() { + const upserts: Array> = []; + const coldDeletes: any[] = []; + return { + upserts, + coldDeletes, + driver: { + name: 'archive', + syncSchema: vi.fn(async () => {}), + find: async () => [], + upsert: async (_object: string, row: Record) => { + upserts.push(row); + return row; + }, + bulkDelete: async () => {}, + deleteMany: async (_object: string, query: any) => { + coldDeletes.push(query); + return 0; + }, + }, + }; + } + + function hotStore(rows: Array>) { + const bulkDeleted: Array> = []; + let remaining = [...rows]; + return { + bulkDeleted, + remaining: () => remaining, + driver: { + name: 'default', + find: async (_object: string, query: any) => remaining.slice(0, query.limit ?? remaining.length), + upsert: async () => ({}), + bulkDelete: async (_object: string, ids: Array) => { + bulkDeleted.push(ids); + remaining = remaining.filter((r) => !ids.includes(r.id as string)); + }, + deleteMany: async () => 0, + }, + }; + } + + it('copies past-window rows to the cold store, then hot-deletes exactly the copied ids', async () => { + const cold = coldStore(); + const hot = hotStore([ + { id: 'a', created_at: '2020-01-01T00:00:00.000Z' }, + { id: 'b', created_at: '2020-06-01T00:00:00.000Z' }, + ]); + const { engine } = captureEngine([AUDIT_OBJ], { + driver: hot.driver, + datasources: { archive: cold.driver }, + }); + + const report = await service(engine).sweep(); + + expect(cold.driver.syncSchema).toHaveBeenCalledWith('sys_audit_log', AUDIT_OBJ); + expect(cold.upserts.map((r) => r.id)).toEqual(['a', 'b']); + expect(hot.bulkDeleted).toEqual([['a', 'b']]); + expect(hot.remaining()).toEqual([]); + + const entry = report.swept.find((e) => e.policy === 'archive'); + expect(entry?.archived).toBe(2); + expect(entry?.cutoff).toBe(isoCutoff('90d')); + // keep: '7y' → the archive itself is pruned past the keep window. + expect(cold.coldDeletes).toEqual([{ where: { created_at: { $lt: isoCutoff('7y') } } }]); + expect(report.skipped).toEqual([]); + }); + + it('retains everything and reports archive-pending when the archive datasource is missing', async () => { + const hot = hotStore([{ id: 'a', created_at: '2020-01-01T00:00:00.000Z' }]); + const { engine, deletes } = captureEngine([AUDIT_OBJ], { driver: hot.driver }); + + const report = await service(engine).sweep(); + + expect(deletes).toHaveLength(0); + expect(hot.bulkDeleted).toEqual([]); + expect(report.skipped).toEqual([{ object: 'sys_audit_log', reason: 'archive-pending' }]); + }); +}); + +describe('LifecycleService.sweep — space reclaim', () => { + it('reclaims once per driver after deletions', async () => { + const reclaimSpace = vi.fn(async () => {}); + const driver = { name: 'default', reclaimSpace }; + const { engine } = captureEngine( + [ + { name: 'sys_job_run', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } }, + { name: 'sys_http_delivery', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } }, + ], + { driver }, + ); + + const report = await service(engine).sweep(); + + // Two objects share one datasource — a single incremental_vacuum suffices. + expect(reclaimSpace).toHaveBeenCalledTimes(1); + expect(report.reclaimed).toEqual(['default']); + }); + + it('honors reclaim:false and skips reclaim when nothing was deleted', async () => { + const reclaimSpace = vi.fn(async () => {}); + const driver = { name: 'default', reclaimSpace }; + + const optedOut = captureEngine( + [{ name: 'sys_job_run', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' }, reclaim: false } }], + { driver }, + ); + await service(optedOut.engine).sweep(); + expect(reclaimSpace).not.toHaveBeenCalled(); + + const nothingDeleted = captureEngine( + [{ name: 'sys_job_run', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } }], + { driver, deleteImpl: () => ({ deletedCount: 0 }) }, + ); + await service(nothingDeleted.engine).sweep(); + expect(reclaimSpace).not.toHaveBeenCalled(); + }); + + it('a reclaim failure is logged, not thrown', async () => { + const driver = { name: 'default', reclaimSpace: vi.fn(async () => { throw new Error('locked'); }) }; + const { engine } = captureEngine( + [{ name: 'sys_job_run', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } }], + { driver }, + ); + + const report = await service(engine).sweep(); + expect(report.reclaimed).toEqual([]); + expect(report.swept).toHaveLength(1); + }); +}); + +describe('LifecycleService.sweep — governance (P4)', () => { + const TELEMETRY_OBJ: LifecycleObjectLike = { + name: 'sys_job_run', + lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } as any, + }; + + /** Fake settings service backed by a value map, with per-tenant values. */ + function fakeSettings(values: Record, tenantValues: Record> = {}) { + return { + async get(_ns: string, key: string, ctx?: Record) { + const tenantId = ctx?.tenantId as string | undefined; + if (tenantId && tenantValues[tenantId] && key in tenantValues[tenantId]) { + return { value: tenantValues[tenantId][key], source: 'tenant' }; + } + if (key in values) return { value: values[key], source: 'global' }; + return { value: undefined, source: 'default' }; + }, + }; + } + + it('a global retention override beats the declared window', async () => { + const { engine, deletes } = captureEngine([TELEMETRY_OBJ]); + const settings = fakeSettings({ retention_overrides: { sys_job_run: { maxAge: '90d' } } }); + + await service(engine, { getSettings: () => settings }).sweep(); + + expect(deletes).toHaveLength(1); + expect(deletes[0].where).toEqual({ created_at: { $lt: isoCutoff('90d') } }); + }); + + it('an invalid override keeps the declared window (never fails open)', async () => { + const { engine, deletes } = captureEngine([TELEMETRY_OBJ]); + const settings = fakeSettings({ retention_overrides: { sys_job_run: { maxAge: 'forever' } } }); + + await service(engine, { getSettings: () => settings }).sweep(); + + expect(deletes[0].where).toEqual({ created_at: { $lt: isoCutoff('30d') } }); + }); + + it('settings enabled=false disables the sweep at runtime', async () => { + const { engine, deletes } = captureEngine([TELEMETRY_OBJ]); + const settings = fakeSettings({ enabled: false }); + + const report = await service(engine, { getSettings: () => settings }).sweep(); + + expect(deletes).toHaveLength(0); + expect(report.swept).toEqual([]); + }); + + it('tenant-scoped overrides sweep each tenant on its own window and everyone else globally', async () => { + const { engine, deletes } = captureEngine([TELEMETRY_OBJ]); + (engine as any).find = async (object: string) => + object === 'sys_organization' ? [{ id: 'org_reg' }, { id: 'org_plain' }] : []; + const settings = fakeSettings( + {}, + { org_reg: { retention_overrides: { sys_job_run: { maxAge: '2y' } } } }, + ); + + await service(engine, { getSettings: () => settings }).sweep(); + + // One tenant-scoped delete on the regulated tenant's 2y window… + expect(deletes[0].where).toEqual({ + created_at: { $lt: isoCutoff('2y') }, + organization_id: 'org_reg', + }); + // …then the global 30d pass excluding it but INCLUDING NULL-org rows. + expect(deletes[1].where).toEqual({ + created_at: { $lt: isoCutoff('30d') }, + $or: [{ organization_id: { $nin: ['org_reg'] } }, { organization_id: null }], + }); + expect(deletes).toHaveLength(2); + }); + + it('raises quota and growth alerts (observe-only — no extra deletes)', async () => { + const onAlert = vi.fn(); + const count = vi.fn(async () => 1_500); + const driver = { name: 'default', count }; + const { engine, deletes } = captureEngine([TELEMETRY_OBJ], { driver }); + const settings = fakeSettings({ quotas: { sys_job_run: 1_000 }, growth_alert_rows: 100 }); + const svc = service(engine, { getSettings: () => settings, onAlert }); + + const first = await svc.sweep(); + expect(first.alerts).toEqual([{ type: 'quota-exceeded', object: 'sys_job_run', rowCount: 1_500, quota: 1_000 }]); + + // Second sweep: +500 rows since the baseline → growth alert too. + count.mockResolvedValue(2_000); + const second = await svc.sweep(); + expect(second.alerts).toContainEqual({ type: 'quota-exceeded', object: 'sys_job_run', rowCount: 2_000, quota: 1_000 }); + expect(second.alerts).toContainEqual({ type: 'growth', object: 'sys_job_run', rowCount: 2_000, delta: 500 }); + expect(onAlert).toHaveBeenCalledTimes(3); + + // Governance only ever ALERTS — the reap count is untouched (one per sweep). + expect(deletes).toHaveLength(2); + }); + + it('quota defaults by class apply when no per-object quota is set', async () => { + const driver = { name: 'default', count: async () => 50 }; + const { engine } = captureEngine([TELEMETRY_OBJ], { driver }); + const settings = fakeSettings({ quota_defaults: { telemetry: 10 } }); + + const report = await service(engine, { getSettings: () => settings }).sweep(); + + expect(report.alerts).toEqual([{ type: 'quota-exceeded', object: 'sys_job_run', rowCount: 50, quota: 10 }]); + }); +}); + +describe('LifecycleService timers', () => { + it('start() sweeps after the initial delay and then on the interval; stop() disarms', async () => { + vi.useFakeTimers(); + try { + const { engine, deletes } = captureEngine([ + { name: 'sys_job_run', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } }, + ]); + const svc = service(engine, { initialDelayMs: 1_000, sweepIntervalMs: 5_000 }); + + svc.start(); + expect(deletes).toHaveLength(0); + + await vi.advanceTimersByTimeAsync(1_000); + expect(deletes).toHaveLength(1); + + await vi.advanceTimersByTimeAsync(5_000); + expect(deletes).toHaveLength(2); + + svc.stop(); + await vi.advanceTimersByTimeAsync(20_000); + expect(deletes).toHaveLength(2); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/packages/objectql/src/lifecycle/lifecycle-service.ts b/packages/objectql/src/lifecycle/lifecycle-service.ts new file mode 100644 index 0000000000..2302983f5b --- /dev/null +++ b/packages/objectql/src/lifecycle/lifecycle-service.ts @@ -0,0 +1,663 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import type { Lifecycle } from '@objectstack/spec/data'; +import { parseLifecycleDuration } from './duration.js'; + +/** + * LifecycleService — the single platform-owned enforcer of ADR-0057 + * `lifecycle` declarations. Scans every registered object carrying a + * `lifecycle` block and applies its policy: + * + * - **Reaper** (P1): batch-deletes rows past `retention.maxAge` (by + * `created_at`) or past `ttl.field + ttl.expireAfter`, then asks each + * touched driver to reclaim free space (SQLite `incremental_vacuum`). + * - **Rotator** (P2): time-shards high-frequency telemetry and DROPs the + * oldest shard. Until a driver advertises rotation support, declared + * rotation falls back to an age-based reap bounded by `shards × unit`. + * - **Archiver** (P3): copies audit-class cold rows to the declared archive + * datasource, then deletes them from the hot store. **Safety rule:** an + * object that declares `archive` is never hot-deleted unless the archive + * copy succeeded — a compliance ledger must not be dropped unarchived. + * + * Design constraints (ADR-0057 §3.3): + * - One implementation, owned here — not N per-plugin sweepers. + * - Sweeps run under a system context (cross-tenant operator policy) and + * use bulk `multi: true` deletes, so at most ONE afterDelete hook fires + * per object per sweep — audit sees an aggregate, never per-row noise + * (telemetry-class sys_* objects are additionally in the audit writer's + * SKIP_OBJECTS, so they produce no audit rows at all). + * - A sweep failure is logged and isolated; it never throws into the + * scheduler and never blocks other objects' policies. + */ + +/** Cross-tenant operator context — lifecycle is a system policy, not a user + * action (mirrors the existing retention sweepers). */ +const SYSTEM_CTX: LifecycleSweepContext = { isSystem: true, positions: [], permissions: [] }; + +export interface LifecycleSweepContext { + isSystem: boolean; + positions: string[]; + permissions: string[]; +} + +/** Width of one rotation shard. Months are the operational 30d, matching the + * coarse-bound posture of {@link parseLifecycleDuration}. */ +const SHARD_UNIT_MS: Record<'day' | 'week' | 'month', number> = { + day: 86_400_000, + week: 7 * 86_400_000, + month: 30 * 86_400_000, +}; + +/** Default cadence between sweeps. Lifecycle windows are hours-to-years, so + * hourly enforcement is ample and keeps the sweep invisible in profiles. */ +export const DEFAULT_LIFECYCLE_SWEEP_MS = 3_600_000; + +/** Delay before the first sweep after boot — lets seeding/migrations finish + * and keeps short-lived test kernels from ever sweeping. */ +export const DEFAULT_LIFECYCLE_INITIAL_DELAY_MS = 60_000; + +/** Minimal engine surface the service needs — duck-typed for tests. */ +export interface LifecycleEngineLike { + registry: { getAllObjects(): LifecycleObjectLike[] }; + delete( + object: string, + options: { where: Record; multi: true; context: LifecycleSweepContext }, + ): Promise; + getDriverForObject(objectName: string): unknown; + /** Datasource lookup by name; throws/absent when not registered. */ + datasource?(name: string): unknown; + /** Row reads for governance (tenant enumeration); optional. */ + find?(object: string, options: Record): Promise>>; +} + +export interface LifecycleObjectLike { + name: string; + lifecycle?: Lifecycle; + fields?: Record; +} + +export interface LifecycleLoggerLike { + info(msg: string, meta?: unknown): void; + warn(msg: string, meta?: unknown): void; + debug?(msg: string, meta?: unknown): void; +} + +/** Duck-typed SettingsService surface (avoids a package dependency). */ +export interface LifecycleSettingsLike { + get( + namespace: string, + key: string, + ctx?: Record, + ): Promise<{ value: unknown; source?: string }>; +} + +/** Governance alert (ADR-0057 P4) — quotas/growth never delete data beyond + * the declared policy; they alert so an operator decides. */ +export interface LifecycleGovernanceAlert { + type: 'quota-exceeded' | 'growth'; + object: string; + rowCount: number; + quota?: number; + delta?: number; +} + +export interface LifecycleServiceOptions { + /** Resolve the data engine; `undefined` ⇒ sweep is a no-op. */ + getEngine(): LifecycleEngineLike | undefined; + logger: LifecycleLoggerLike; + /** Master switch. Defaults to true; `OS_LIFECYCLE_DISABLED=1` also wins. */ + enabled?: boolean; + /** Cadence between sweeps. Default {@link DEFAULT_LIFECYCLE_SWEEP_MS}. */ + sweepIntervalMs?: number; + /** Delay before the first sweep. Default {@link DEFAULT_LIFECYCLE_INITIAL_DELAY_MS}. */ + initialDelayMs?: number; + /** Clock injection for deterministic tests. Defaults to `Date.now()`. */ + now?(): number; + /** Resolve the settings service for governance (P4); absent ⇒ declared + * policies apply unmodified and quotas/alerts are off. */ + getSettings?(): LifecycleSettingsLike | undefined; + /** Governance alert sink. Defaults to a logger warning. */ + onAlert?(alert: LifecycleGovernanceAlert): void; +} + +/** Per-sweep governance snapshot resolved from the `lifecycle` namespace. */ +interface GovernanceSnapshot { + enabled: boolean; + /** Global-resolved per-object window overrides. */ + overrides: Record; + /** object → tenant-specific windows (only tenants whose override is + * genuinely tenant-scoped, not inherited). */ + tenantOverrides: Map>; + quotas: Record; + quotaDefaults: Record; + growthAlertRows: number; +} + +const DEFAULT_GOVERNANCE: GovernanceSnapshot = { + enabled: true, + overrides: {}, + tenantOverrides: new Map(), + quotas: {}, + quotaDefaults: {}, + growthAlertRows: 0, +}; + +/** Cap on tenants scanned for per-tenant overrides each sweep. */ +const TENANT_SCAN_LIMIT = 200; + +export interface LifecycleSweepEntry { + object: string; + class: string; + policy: 'ttl' | 'retention' | 'rotation' | 'rotation-fallback' | 'archive'; + cutoff: string; + /** `undefined` when the driver doesn't report a count. */ + deleted?: number; + /** Rotation only: expired shard tables DROPped this sweep (O(1) reclaim). */ + droppedShards?: number; + /** Archive only: rows copied to the cold store (then hot-deleted). */ + archived?: number; +} + +export interface LifecycleSweepReport { + at: string; + /** Policies applied, one entry per (object, policy). */ + swept: LifecycleSweepEntry[]; + /** Objects intentionally not swept, with the reason. */ + skipped: Array<{ object: string; reason: string }>; + /** Isolated per-object failures — the sweep itself never throws. */ + errors: Array<{ object: string; error: string }>; + /** Datasources whose driver reclaimed space after this sweep. */ + reclaimed: string[]; + /** Governance alerts raised this sweep (quota breaches, growth spikes). */ + alerts: LifecycleGovernanceAlert[]; +} + +interface ReclaimCapableDriver { + name?: string; + reclaimSpace?(): Promise; +} + +interface RotationCapableDriver extends ReclaimCapableDriver { + supportsRotation?: boolean; + rotateShards?( + objectDef: LifecycleObjectLike, + nowMs?: number, + ): Promise<{ object: string; current: string; shards: string[]; dropped: string[] }>; +} + +/** Driver surface the Archiver uses on both the hot and the cold store. */ +interface ArchiveCapableDriver { + name?: string; + find(object: string, query: Record, options?: unknown): Promise>>; + upsert(object: string, data: Record, conflictKeys?: string[], options?: unknown): Promise; + bulkDelete(object: string, ids: Array, options?: unknown): Promise; + deleteMany?(object: string, query: Record, options?: unknown): Promise; + syncSchema?(object: string, schema: unknown, options?: unknown): Promise; +} + +/** Max rows the Archiver moves per object per sweep — bounds sweep latency; + * the backlog drains across consecutive sweeps. */ +const ARCHIVE_BATCH_SIZE = 500; +const ARCHIVE_MAX_BATCHES_PER_SWEEP = 20; + +export class LifecycleService { + private readonly now: () => number; + private timer: ReturnType | undefined; + private initialTimer: ReturnType | undefined; + private sweeping = false; + /** Row counts from the previous sweep — baseline for growth alerts. */ + private lastCounts = new Map(); + /** Governance snapshot for the sweep in flight. */ + private governance: GovernanceSnapshot = DEFAULT_GOVERNANCE; + + constructor(private readonly opts: LifecycleServiceOptions) { + this.now = opts.now ?? (() => Date.now()); + } + + get enabled(): boolean { + if (process.env.OS_LIFECYCLE_DISABLED === '1') return false; + return this.opts.enabled !== false; + } + + /** Arm the periodic sweep. Idempotent; timers are unref'ed so a kernel + * shutdown is never held open by the lifecycle schedule. */ + start(): void { + if (!this.enabled || this.timer || this.initialTimer) return; + const interval = this.opts.sweepIntervalMs ?? DEFAULT_LIFECYCLE_SWEEP_MS; + const initial = this.opts.initialDelayMs ?? DEFAULT_LIFECYCLE_INITIAL_DELAY_MS; + this.initialTimer = setTimeout(() => { + this.initialTimer = undefined; + void this.sweep(); + this.timer = setInterval(() => void this.sweep(), interval); + this.timer.unref?.(); + }, initial); + this.initialTimer.unref?.(); + } + + stop(): void { + if (this.initialTimer) clearTimeout(this.initialTimer); + if (this.timer) clearInterval(this.timer); + this.initialTimer = undefined; + this.timer = undefined; + } + + /** + * Apply every declared lifecycle policy once. Safe to call directly (the + * dogfood growth gate and `db:clean`-style tooling do); re-entrant calls + * while a sweep is running resolve to an empty report. + */ + async sweep(): Promise { + const report: LifecycleSweepReport = { + at: new Date(this.now()).toISOString(), + swept: [], + skipped: [], + errors: [], + reclaimed: [], + alerts: [], + }; + if (this.sweeping || !this.enabled) return report; + const engine = this.opts.getEngine(); + if (!engine || typeof engine.delete !== 'function' || !engine.registry) { + this.opts.logger.debug?.('[lifecycle] no data engine available; sweep skipped'); + return report; + } + + this.sweeping = true; + try { + const declared = engine.registry + .getAllObjects() + .filter((o) => o?.lifecycle && o.lifecycle.class !== 'record'); + + // Governance snapshot (P4): settings-driven overrides / quotas. + this.governance = await this.loadGovernance(engine, declared); + if (!this.governance.enabled) { + this.opts.logger.debug?.('[lifecycle] disabled via settings; sweep skipped'); + return report; + } + + // Drivers that should reclaim space after this sweep (deduped by + // instance — several objects usually share one datasource). + const reclaimable = new Set(); + + for (const obj of declared) { + const lc = obj.lifecycle as Lifecycle; + try { + const outcomes = await this.reapObject(engine, obj, lc, report); + const deletedSomething = outcomes.some((n) => n === undefined || n > 0); + if (deletedSomething && lc.reclaim !== false) { + const driver = engine.getDriverForObject(obj.name) as ReclaimCapableDriver | undefined; + if (driver && typeof driver.reclaimSpace === 'function') reclaimable.add(driver); + } + } catch (err) { + const msg = (err as Error)?.message ?? String(err); + report.errors.push({ object: obj.name, error: msg }); + this.opts.logger.warn(`[lifecycle] sweep of ${obj.name} failed (${msg})`); + } + } + + for (const driver of reclaimable) { + try { + await driver.reclaimSpace!(); + report.reclaimed.push(driver.name ?? 'default'); + } catch (err) { + this.opts.logger.warn( + `[lifecycle] space reclaim on datasource '${driver.name ?? 'default'}' failed (${(err as Error)?.message ?? err})`, + ); + } + } + + // Governance (P4): quotas + growth alerts — observe-and-alert only, + // never a delete beyond the declared policy. + await this.checkGovernance(engine, declared, report); + + if (report.swept.length > 0 || report.errors.length > 0 || report.alerts.length > 0) { + // ADR-0057 §3.3: cleanup must not re-feed the tables it drains — one + // aggregate log line per sweep is the entire trace it leaves. + const total = report.swept.reduce((sum, e) => sum + (e.deleted ?? 0), 0); + this.opts.logger.info( + `[lifecycle] sweep: ${report.swept.length} policy(ies) applied, ~${total} rows reaped, ` + + `${report.reclaimed.length} datasource(s) reclaimed, ${report.errors.length} error(s), ` + + `${report.alerts.length} alert(s)`, + ); + } + return report; + } finally { + this.sweeping = false; + } + } + + /** Resolve the `lifecycle` settings namespace into a per-sweep snapshot. + * Every read is best-effort: no settings service / unregistered namespace + * ⇒ declared policies apply unmodified. */ + private async loadGovernance( + engine: LifecycleEngineLike, + declared: LifecycleObjectLike[], + ): Promise { + const settings = this.opts.getSettings?.(); + if (!settings || typeof settings.get !== 'function') return DEFAULT_GOVERNANCE; + + const read = async (key: string, fallback: T, ctx?: Record): Promise<{ value: T; source?: string }> => { + try { + const r = await settings.get('lifecycle', key, ctx); + return { value: (r?.value ?? fallback) as T, source: r?.source }; + } catch { + return { value: fallback }; + } + }; + + const snapshot: GovernanceSnapshot = { + enabled: (await read('enabled', true)).value !== false, + overrides: (await read>('retention_overrides', {})).value ?? {}, + tenantOverrides: new Map(), + quotas: (await read>('quotas', {})).value ?? {}, + quotaDefaults: (await read>('quota_defaults', {})).value ?? {}, + growthAlertRows: Number((await read('growth_alert_rows', 0)).value) || 0, + }; + + // Tenant-level windows (ADR-0057 §3.2): only overrides genuinely stored + // at TENANT scope count — inherited global values would otherwise turn + // every tenant into a "tenant override" and break the global pass. + if (typeof engine.find === 'function' && declared.length > 0) { + try { + const orgs = await engine.find('sys_organization', { + limit: TENANT_SCAN_LIMIT, + context: { ...SYSTEM_CTX }, + }); + for (const org of orgs ?? []) { + const tenantId = org?.id as string | undefined; + if (!tenantId) continue; + const r = await read>( + 'retention_overrides', + {}, + { tenantId }, + ); + if (r.source !== 'tenant') continue; + for (const [objectName, windows] of Object.entries(r.value ?? {})) { + if (!windows || typeof windows !== 'object') continue; + const list = snapshot.tenantOverrides.get(objectName) ?? []; + list.push({ tenantId, maxAge: windows.maxAge, expireAfter: windows.expireAfter }); + snapshot.tenantOverrides.set(objectName, list); + } + } + } catch { + // No sys_organization (single-tenant kernel) — tenant overrides n/a. + } + } + + return snapshot; + } + + /** Quota + growth checks (P4). Alerts only — an operator decides. */ + private async checkGovernance( + engine: LifecycleEngineLike, + declared: LifecycleObjectLike[], + report: LifecycleSweepReport, + ): Promise { + const gov = this.governance; + const nextCounts = new Map(); + for (const obj of declared) { + const driver = engine.getDriverForObject(obj.name) as + | { count?(object: string, query?: Record): Promise } + | undefined; + if (!driver || typeof driver.count !== 'function') continue; + let rowCount: number; + try { + rowCount = await driver.count(obj.name, { object: obj.name }); + } catch { + continue; + } + nextCounts.set(obj.name, rowCount); + + const quota = gov.quotas[obj.name] ?? gov.quotaDefaults[obj.lifecycle!.class]; + if (typeof quota === 'number' && quota > 0 && rowCount > quota) { + this.alert(report, { type: 'quota-exceeded', object: obj.name, rowCount, quota }); + } + + const last = this.lastCounts.get(obj.name); + if (gov.growthAlertRows > 0 && last !== undefined && rowCount - last > gov.growthAlertRows) { + this.alert(report, { type: 'growth', object: obj.name, rowCount, delta: rowCount - last }); + } + } + this.lastCounts = nextCounts; + } + + private alert(report: LifecycleSweepReport, alert: LifecycleGovernanceAlert): void { + report.alerts.push(alert); + if (this.opts.onAlert) { + try { + this.opts.onAlert(alert); + } catch { + /* alert sinks must never break the sweep */ + } + } else { + this.opts.logger.warn( + `[lifecycle] governance alert: ${alert.type} on ${alert.object} ` + + `(rows=${alert.rowCount}${alert.quota != null ? `, quota=${alert.quota}` : ''}${alert.delta != null ? `, delta=+${alert.delta}` : ''})`, + ); + } + } + + /** Apply the policies declared on one object (Rotator first, then the + * Reaper). Returns per-policy outcomes so the caller can decide on + * reclaim: numbers are deleted-row counts; `undefined` means "work was + * done but the driver reports no count" (also used for dropped shards). */ + private async reapObject( + engine: LifecycleEngineLike, + obj: LifecycleObjectLike, + lc: Lifecycle, + report: LifecycleSweepReport, + ): Promise> { + const object = obj.name; + + // Safety rule: declared `archive` means retain → archive → delete. Hot + // deletion happens ONLY for rows the Archiver has copied to the cold + // store; when the archive datasource isn't registered, rows are retained + // (never dropped unarchived) and the object is reported as skipped. + if (lc.archive) { + return this.archiveObject(engine, obj, lc, report); + } + + const outcomes: Array = []; + // Governance overrides (P4): a configured window beats the declared one. + const ov = this.governance.overrides[object] ?? {}; + + if (lc.ttl) { + const windowMs = this.effectiveWindowMs(ov.expireAfter, parseLifecycleDuration(lc.ttl.expireAfter), object); + outcomes.push(await this.reap(engine, object, lc, 'ttl', lc.ttl.field, windowMs, report)); + } + + // Rotation (P2): physical time-sharding when the driver supports it — + // the window bound comes from DROPping expired shards (O(1) reclaim). + // Drivers without rotation fall through to an equivalent age-based reap, + // so the declared bound holds on every dialect. + let rotated = false; + if (lc.storage?.strategy === 'rotation') { + const driver = engine.getDriverForObject(object) as RotationCapableDriver | undefined; + if (driver && typeof driver.rotateShards === 'function' && driver.supportsRotation !== false) { + const windowMs = lc.storage.shards * SHARD_UNIT_MS[lc.storage.unit]; + const res = await driver.rotateShards(obj, this.now()); + report.swept.push({ + object, + class: lc.class, + policy: 'rotation', + cutoff: new Date(this.now() - windowMs).toISOString(), + droppedShards: res.dropped.length, + }); + // Dropped shards freed pages — signal the reclaim pass. + outcomes.push(res.dropped.length > 0 ? undefined : 0); + rotated = true; + } + } + + if (lc.retention) { + // Runs even when rotation is active: rotation reclaims at SHARD + // granularity, an explicit retention.maxAge trims to the day inside the + // live shards — and immediately bounds a legacy table the Rotator just + // adopted whole into its first shard. + const windowMs = this.effectiveWindowMs(ov.maxAge, parseLifecycleDuration(lc.retention.maxAge), object); + outcomes.push(await this.reap(engine, object, lc, 'retention', 'created_at', windowMs, report)); + } else if (lc.storage?.strategy === 'rotation' && !rotated && !lc.ttl) { + // Rotation declared but the driver can't shard physically: the shard + // window IS the bound — enforce the same window with an age-based reap + // so the declaration is never inert. + const windowMs = this.effectiveWindowMs(ov.maxAge, lc.storage.shards * SHARD_UNIT_MS[lc.storage.unit], object); + outcomes.push(await this.reap(engine, object, lc, 'rotation-fallback', 'created_at', windowMs, report)); + } + + return outcomes; + } + + /** A governance override window beats the declared one — unless it fails to + * parse, in which case the declared window stands (never fail open into + * "no bound at all"). */ + private effectiveWindowMs(override: string | undefined, declaredMs: number, object: string): number { + if (!override) return declaredMs; + try { + return parseLifecycleDuration(override); + } catch { + this.opts.logger.warn(`[lifecycle] invalid override window '${override}' for ${object}; keeping the declared window`); + return declaredMs; + } + } + + /** + * Archiver (ADR-0057 §3.3 / P3): copy rows past `archive.after` from the + * hot store to the archive datasource, then delete the copied rows hot. + * Batched (500 × 20 per sweep) so a large backlog drains across sweeps + * without one long-locking pass. Copies are per-row idempotent upserts, so + * a sweep interrupted between copy and hot-delete re-converges. When + * `archive.keep` is set, cold rows past it are pruned from the archive. + */ + private async archiveObject( + engine: LifecycleEngineLike, + obj: LifecycleObjectLike, + lc: Lifecycle, + report: LifecycleSweepReport, + ): Promise> { + const object = obj.name; + const archive = lc.archive!; + + let cold: ArchiveCapableDriver | undefined; + try { + cold = engine.datasource?.(archive.to) as ArchiveCapableDriver | undefined; + } catch { + cold = undefined; + } + const hot = engine.getDriverForObject(object) as ArchiveCapableDriver | undefined; + if (!cold || !hot || typeof hot.find !== 'function' || typeof cold.upsert !== 'function') { + // No archive target ⇒ retain everything. A compliance ledger cannot be + // destroyed by declaring a lifecycle — this is the safe default state + // for deployments that never provision cold storage. + report.skipped.push({ object, reason: 'archive-pending' }); + return []; + } + + // The cold store mirrors the hot schema (idempotent DDL). + if (typeof cold.syncSchema === 'function') { + await cold.syncSchema(object, obj); + } + + const cutoff = new Date(this.now() - parseLifecycleDuration(archive.after)).toISOString(); + let archived = 0; + for (let batch = 0; batch < ARCHIVE_MAX_BATCHES_PER_SWEEP; batch++) { + const rows = await hot.find(object, { + where: { created_at: { $lt: cutoff } }, + limit: ARCHIVE_BATCH_SIZE, + }); + if (!rows.length) break; + for (const row of rows) { + await cold.upsert(object, row, ['id']); + } + await hot.bulkDelete(object, rows.map((r) => r.id as string)); + archived += rows.length; + if (rows.length < ARCHIVE_BATCH_SIZE) break; + } + + // Cold-side retention: `keep` bounds the archive itself. + if (archive.keep && typeof cold.deleteMany === 'function') { + const keepCutoff = new Date(this.now() - parseLifecycleDuration(archive.keep)).toISOString(); + await cold.deleteMany(object, { where: { created_at: { $lt: keepCutoff } } }); + } + + report.swept.push({ object, class: lc.class, policy: 'archive', cutoff, archived }); + return [archived]; + } + + private async reap( + engine: LifecycleEngineLike, + object: string, + lc: Lifecycle, + policy: LifecycleSweepEntry['policy'], + field: string, + windowMs: number, + report: LifecycleSweepReport, + ): Promise { + const cutoff = new Date(this.now() - windowMs).toISOString(); + const overrideKey = policy === 'ttl' ? 'expireAfter' : 'maxAge'; + const tenantWindows = (this.governance.tenantOverrides.get(object) ?? []).filter( + (t) => typeof t[overrideKey] === 'string', + ); + + let total: number | undefined = 0; + const accumulate = (res: unknown) => { + const n = countDeleted(res); + if (n === undefined) total = undefined; + else if (total !== undefined) total += n; + }; + + if (tenantWindows.length === 0) { + accumulate( + await engine.delete(object, { + where: { [field]: { $lt: cutoff } }, + multi: true, + context: { ...SYSTEM_CTX }, + }), + ); + } else { + // Tenant-level windows (P4): each overriding tenant gets its own + // cutoff on its own rows… + for (const t of tenantWindows) { + const tMs = this.effectiveWindowMs(t[overrideKey], windowMs, `${object} (tenant ${t.tenantId})`); + const tCutoff = new Date(this.now() - tMs).toISOString(); + accumulate( + await engine.delete(object, { + where: { [field]: { $lt: tCutoff }, organization_id: t.tenantId }, + multi: true, + context: { ...SYSTEM_CTX }, + }), + ); + } + // …and the global pass covers everyone else, INCLUDING rows with no + // organization (a bare `$nin` would silently skip NULL-org rows). + accumulate( + await engine.delete(object, { + where: { + [field]: { $lt: cutoff }, + $or: [ + { organization_id: { $nin: tenantWindows.map((t) => t.tenantId) } }, + { organization_id: null }, + ], + }, + multi: true, + context: { ...SYSTEM_CTX }, + }), + ); + } + + report.swept.push({ object, class: lc.class, policy, cutoff, deleted: total }); + return total; + } +} + +/** Best-effort row-count extraction from a driver's delete result. */ +function countDeleted(res: unknown): number | undefined { + if (typeof res === 'number') return res; + if (Array.isArray(res)) return res.length; + if (res && typeof res === 'object') { + const r = res as Record; + for (const k of ['deletedCount', 'deleted', 'count', 'affected', 'affectedRows']) { + if (typeof r[k] === 'number') return r[k] as number; + } + } + return undefined; +} diff --git a/packages/objectql/src/lifecycle/lifecycle-settings.ts b/packages/objectql/src/lifecycle/lifecycle-settings.ts new file mode 100644 index 0000000000..ecc33c0f27 --- /dev/null +++ b/packages/objectql/src/lifecycle/lifecycle-settings.ts @@ -0,0 +1,76 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +import type { SettingsManifest } from '@objectstack/spec/system'; + +/** + * `lifecycle` settings namespace (ADR-0057 P4 — governance). + * + * Registered by ObjectQLPlugin at kernel:ready when a SettingsService is + * present. Every value follows the standard cascade (OS_LIFECYCLE_* env > + * global > tenant > default), which is exactly ADR-0057 §3.2's "regulated + * tenants set years; dev sets days — the same knob at different settings": + * `retention_overrides` is tenant-scoped, so one deployment can carry both. + */ +export const lifecycleSettingsManifest = { + namespace: 'lifecycle', + version: 1, + label: 'Data Lifecycle', + icon: 'Timer', + description: + 'Governance for ADR-0057 data lifecycle enforcement: retention window ' + + 'overrides, per-table row quotas, and growth alerts for the platform ' + + 'LifecycleService (Reaper / Rotator / Archiver).', + scope: 'global', + readPermission: 'manage_platform_settings', + writePermission: 'manage_platform_settings', + category: 'Infrastructure', + order: 40, + specifiers: [ + { + type: 'toggle', + key: 'enabled', + label: 'Enforce lifecycle policies', + required: false, + default: true, + description: + 'Master switch for the periodic sweep. Off = declared retention/ttl/rotation policies stop being enforced (data grows unbounded again).', + }, + { + type: 'json', + key: 'retention_overrides', + label: 'Retention overrides', + required: false, + scope: 'tenant', + default: {}, + description: + 'Per-object window overrides: { "": { "maxAge": "1y", "expireAfter": "30d" } }. ' + + 'Duration literals: h/d/w/y. Tenant-scoped — a regulated tenant sets years while dev keeps days (ADR-0057 §3.2).', + }, + { + type: 'json', + key: 'quotas', + label: 'Row quotas', + required: false, + default: {}, + description: + 'Per-object max row counts: { "": 1000000 }. A breach raises a governance alert — quotas never delete beyond the declared policy.', + }, + { + type: 'json', + key: 'quota_defaults', + label: 'Per-class quota defaults', + required: false, + default: {}, + description: 'Fallback quotas by lifecycle class: { "telemetry": 1000000, "transient": 500000, "event": 100000, "audit": 5000000 }.', + }, + { + type: 'number', + key: 'growth_alert_rows', + label: 'Growth alert threshold (rows per sweep)', + required: false, + default: 0, + description: + 'Alert when a lifecycle-declared table grows by more than this many rows between two sweeps. 0 disables growth alerts.', + }, + ], +} as unknown as SettingsManifest; diff --git a/packages/objectql/src/plugin.ts b/packages/objectql/src/plugin.ts index a6f4fa5123..9a05666afe 100644 --- a/packages/objectql/src/plugin.ts +++ b/packages/objectql/src/plugin.ts @@ -4,6 +4,8 @@ import { ObjectQL } from './engine.js'; import { ObjectStackProtocolImplementation } from '@objectstack/metadata-protocol'; import { Plugin, PluginContext } from '@objectstack/core'; import { StorageNameMapping } from '@objectstack/spec/system'; +import { LifecycleService } from './lifecycle/lifecycle-service.js'; +import { lifecycleSettingsManifest } from './lifecycle/lifecycle-settings.js'; import { SysMetadataObject, SysMetadataHistoryObject, @@ -98,6 +100,19 @@ export interface ObjectQLPluginOptions { * Safe to leave unset: hydration tolerates a missing table. */ hydrateMetadataFromDb?: boolean; + /** + * ADR-0057 LifecycleService tuning. Lifecycle enforcement is a platform + * primitive and defaults ON — objects without a `lifecycle` declaration are + * never touched, so a kernel with no declarations sees zero deletes. Set + * `enabled: false` (or env `OS_LIFECYCLE_DISABLED=1`) to disable the + * periodic sweep entirely; the `lifecycle` service stays registered so + * tooling can still run `sweep()` explicitly. + */ + lifecycle?: { + enabled?: boolean; + sweepIntervalMs?: number; + initialDelayMs?: number; + }; } export class ObjectQLPlugin implements Plugin { @@ -118,6 +133,9 @@ export class ObjectQLPlugin implements Plugin { private hydrateMetadataFromDb = false; /** Unsubscribe handles for metadata-event subscriptions (ADR-0008 PR-7). */ private metadataUnsubscribes: Array<() => void> = []; + /** ADR-0057 lifecycle enforcement (Reaper/Rotator/Archiver). */ + private lifecycleService: LifecycleService | undefined; + private lifecycleOptions: ObjectQLPluginOptions['lifecycle']; constructor(qlOrOptions?: ObjectQL | ObjectQLPluginOptions, hostContext?: Record) { // Back-compat: legacy callers passed `(ObjectQL, hostContext)` positionally. @@ -141,6 +159,7 @@ export class ObjectQLPlugin implements Plugin { ? opts.skipSchemaSync : process.env.OS_SKIP_SCHEMA_SYNC === '1'; this.hydrateMetadataFromDb = opts.hydrateMetadataFromDb === true; + this.lifecycleOptions = opts.lifecycle; } init = async (ctx: PluginContext) => { @@ -295,6 +314,27 @@ export class ObjectQLPlugin implements Plugin { message: 'Analytics SQL generation not implemented by ObjectQL adapter', }), }); + + // ADR-0057: the platform-owned LifecycleService. Registered from the + // engine plugin (not an opt-in capability) so every kernel that has data + // also has lifecycle enforcement — a declared retention that drives no + // sweeper is dead surface (ADR-0049). + this.lifecycleService = new LifecycleService({ + getEngine: () => this.ql, + logger: ctx.logger, + // P4 governance: overrides/quotas resolve from the 'lifecycle' + // settings namespace when a SettingsService is present. Lazy per + // sweep — plugin registration order doesn't matter. + getSettings: () => { + try { + return ctx.getService('settings') as never; + } catch { + return undefined; + } + }, + ...this.lifecycleOptions, + }); + ctx.registerService('lifecycle', this.lifecycleService); } start = async (ctx: PluginContext) => { @@ -335,6 +375,14 @@ export class ObjectQLPlugin implements Plugin { ctx.hook('kernel:ready', async () => { await this.resyncAuthoredHooks(ctx); await this.resyncAuthoredActions(ctx); + // ADR-0057 P4: surface the lifecycle governance namespace in Settings + // (overrides / quotas / growth alerts) when a SettingsService exists. + try { + const settings = ctx.getService('settings') as { registerManifest?: (m: unknown) => void }; + settings?.registerManifest?.(lifecycleSettingsManifest); + } catch { + // No settings service — governance stays at declared defaults. + } }); ctx.hook('metadata:reloaded', async () => { await this.resyncAuthoredHooks(ctx); @@ -449,9 +497,14 @@ export class ObjectQLPlugin implements Plugin { driversRegistered: this.ql?.['drivers']?.size || 0, objectsRegistered: this.ql?.registry?.getAllObjects?.()?.length || 0 }); + + // ADR-0057: arm the periodic lifecycle sweep once the engine is live. + this.lifecycleService?.start(); } stop = async (ctx: PluginContext) => { + // ADR-0057: disarm the lifecycle sweep timers. + this.lifecycleService?.stop(); // ADR-0008 PR-7: tear down metadata subscriptions on plugin stop so // tests don't leak watchers and reloaded plugins don't double-subscribe. for (const unsub of this.metadataUnsubscribes) { diff --git a/packages/platform-objects/src/audit/sys-job-run.object.ts b/packages/platform-objects/src/audit/sys-job-run.object.ts index 75f5af78d9..2239ff3845 100644 --- a/packages/platform-objects/src/audit/sys-job-run.object.ts +++ b/packages/platform-objects/src/audit/sys-job-run.object.ts @@ -22,6 +22,12 @@ export const SysJobRun = ObjectSchema.create({ icon: 'play', isSystem: true, managedBy: 'append-only', + // ADR-0057: mirrors the JobRunRetention default (30d) so the platform + // LifecycleService and the plugin sweeper enforce one window. + lifecycle: { + class: 'telemetry', + retention: { maxAge: '30d' }, + }, description: 'Background job execution audit trail', displayNameField: 'job_name', nameField: 'job_name', // [ADR-0079] canonical primary-title pointer (mirrors deprecated displayNameField) diff --git a/packages/platform-objects/src/audit/sys-notification.object.ts b/packages/platform-objects/src/audit/sys-notification.object.ts index 1033070611..8ae8a7aab0 100644 --- a/packages/platform-objects/src/audit/sys-notification.object.ts +++ b/packages/platform-objects/src/audit/sys-notification.object.ts @@ -32,6 +32,12 @@ export const SysNotification = ObjectSchema.create({ icon: 'bell', isSystem: true, managedBy: 'system', + // ADR-0057: one 90d window across the whole notification pipeline + // (event → delivery → receipt/inbox), matching NotificationRetention. + lifecycle: { + class: 'telemetry', + retention: { maxAge: '90d' }, + }, description: 'Notification events — one row per emit() (ADR-0030 Layer 2 ingress)', displayNameField: 'topic', nameField: 'topic', // [ADR-0079] canonical primary-title pointer (mirrors deprecated displayNameField) diff --git a/packages/platform-objects/src/identity/sys-device-code.object.ts b/packages/platform-objects/src/identity/sys-device-code.object.ts index 9c659b777f..debba67c3e 100644 --- a/packages/platform-objects/src/identity/sys-device-code.object.ts +++ b/packages/platform-objects/src/identity/sys-device-code.object.ts @@ -26,6 +26,12 @@ export const SysDeviceCode = ObjectSchema.create({ icon: 'key-round', isSystem: true, managedBy: 'better-auth', + // ADR-0057: device codes are dead the moment `expires_at` passes — keep a + // 1d grace for post-mortem, then reap. + lifecycle: { + class: 'transient', + ttl: { field: 'expires_at', expireAfter: '1d' }, + }, // [ADR-0066 D2/④] Secure-by-default: rows are LIVE pending device-grant // codes — reading `user_code`/`device_code` lets an attacker hijack a // pending CLI login. Not covered by the wildcard `'*'` grant; admins retain diff --git a/packages/plugins/driver-sql/src/sql-driver-rotation.test.ts b/packages/plugins/driver-sql/src/sql-driver-rotation.test.ts new file mode 100644 index 0000000000..7791d77e1a --- /dev/null +++ b/packages/plugins/driver-sql/src/sql-driver-rotation.test.ts @@ -0,0 +1,156 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * ADR-0057 P2 — physical table rotation for high-frequency telemetry. + * + * Rotation-declared objects are time-sharded: writes land in the current + * shard (`__r`), reads go through a UNION ALL view under the base + * name, and expiry DROPs shards past the `shards × unit` window (O(1) + * reclaim). These tests drive the full shard lifecycle on SQLite. + * + * Every test boots through `initObjects` first — that's what registers the + * per-field read/filter bookkeeping (datetime coercion etc.) the shards + * alias; `rotateShards` alone is the sweep-time entry point, not a boot path. + * The pinned clock T0 is far in the future so real-`Date.now()` boot shards + * are deterministically outside the test window. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { SqlDriver } from './index.js'; + +const DAY_MS = 86_400_000; +const T0 = Date.parse('2036-08-01T12:00:00.000Z'); + +const ROTATED_OBJECT = { + name: 'rot_event', + fields: { + payload: { type: 'text' }, + created_at: { type: 'datetime' }, + }, + lifecycle: { + class: 'telemetry', + storage: { strategy: 'rotation', shards: 3, unit: 'day' }, + }, +}; + +async function tableNames(driver: SqlDriver): Promise> { + const rows = (await driver.execute( + "SELECT name, type FROM sqlite_master WHERE name LIKE 'rot_event%' AND name NOT LIKE '%autoindex%'", + )) as Array<{ name: string; type: string }>; + return Object.fromEntries(rows.map((r) => [r.name, r.type])); +} + +describe('SqlDriver rotation (ADR-0057 P2)', () => { + let driver: SqlDriver; + + beforeEach(async () => { + driver = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: ':memory:' }, + useNullAsDefault: true, + }); + }); + + afterEach(async () => { + await driver.disconnect(); + }); + + it('initObjects shards a rotation-declared object: writes hit the current shard, reads the view', async () => { + // Boot path — initObjects rotates on the REAL clock, so derive the + // expected shard key from Date.now() rather than pinning one. + await driver.initObjects([ROTATED_OBJECT]); + const todayKey = new Date().toISOString().slice(0, 10).replace(/-/g, ''); + const currentShard = `rot_event__r${todayKey}`; + + const names = await tableNames(driver); + expect(names['rot_event']).toBe('view'); + expect(names[currentShard]).toBe('table'); + + await driver.create('rot_event', { id: 'a', payload: 'x' }, { bypassTenantAudit: true }); + + // The row is physically in the shard and visible through the view. + const inShard = (await driver.execute(`SELECT id FROM "${currentShard}"`)) as any[]; + expect(inShard.map((r) => r.id)).toEqual(['a']); + expect(await driver.count('rot_event', { object: 'rot_event' })).toBe(1); + const found = await driver.findOne('rot_event', { object: 'rot_event', where: { id: 'a' } }); + expect(found?.payload).toBe('x'); + }); + + it('rotating across days creates new shards, unions all live rows, and DROPs shards past the window', async () => { + await driver.initObjects([ROTATED_OBJECT]); + await driver.rotateShards(ROTATED_OBJECT, T0); // drops the empty boot shard (outside the T0 window) + await driver.create('rot_event', { id: 'd0', payload: 'day0', created_at: new Date(T0) }, { bypassTenantAudit: true }); + + // Day 1: new shard becomes the write target; day-0 row still readable. + const day1 = await driver.rotateShards(ROTATED_OBJECT, T0 + 1 * DAY_MS); + expect(day1.current).toBe('rot_event__r20360802'); + expect(day1.dropped).toEqual([]); + await driver.create('rot_event', { id: 'd1', payload: 'day1', created_at: new Date(T0 + DAY_MS) }, { bypassTenantAudit: true }); + expect(await driver.count('rot_event', { object: 'rot_event' })).toBe(2); + + // Day 3 (shards=3, window = [day1 .. day3]): the day-0 shard falls out — + // one O(1) DROP, its rows gone from the view, newer rows intact. + const day3 = await driver.rotateShards(ROTATED_OBJECT, T0 + 3 * DAY_MS); + expect(day3.dropped).toEqual(['rot_event__r20360801']); + expect(day3.shards).toEqual(['rot_event__r20360804', 'rot_event__r20360802']); + + const names = await tableNames(driver); + expect(names['rot_event__r20360801']).toBeUndefined(); + const remaining = await driver.find('rot_event', { object: 'rot_event' }); + expect(remaining.map((r: any) => r.id).sort()).toEqual(['d1']); + }); + + it('adopts a legacy pre-rotation table as the first shard (no data loss)', async () => { + // Boot WITHOUT rotation: plain table with history. + const legacyDef = { name: 'rot_event', fields: ROTATED_OBJECT.fields }; + await driver.initObjects([legacyDef]); + await driver.create('rot_event', { id: 'legacy', payload: 'old-world', created_at: new Date(T0 - 5 * DAY_MS) }, { bypassTenantAudit: true }); + + // Upgrade: same object now declares rotation. + const res = await driver.rotateShards(ROTATED_OBJECT, T0); + expect(res.current).toBe('rot_event__r20360801'); + + const names = await tableNames(driver); + expect(names['rot_event']).toBe('view'); + const rows = await driver.find('rot_event', { object: 'rot_event' }); + expect(rows.map((r: any) => r.id)).toEqual(['legacy']); + }); + + it('by-id update/delete and bulk deleteMany fan out across shards', async () => { + await driver.initObjects([ROTATED_OBJECT]); + await driver.rotateShards(ROTATED_OBJECT, T0); + await driver.create('rot_event', { id: 'old', payload: 'p0', created_at: new Date(T0) }, { bypassTenantAudit: true }); + await driver.rotateShards(ROTATED_OBJECT, T0 + DAY_MS); + await driver.create('rot_event', { id: 'new', payload: 'p1', created_at: new Date(T0 + DAY_MS) }, { bypassTenantAudit: true }); + + // Update a row living in the OLDER shard (not the write target). + const updated = await driver.update('rot_event', 'old', { payload: 'patched' }, { bypassTenantAudit: true }); + expect(updated?.payload).toBe('patched'); + + // Delete by id probes shards until the hit. + expect(await driver.delete('rot_event', 'old', { bypassTenantAudit: true })).toBe(true); + expect(await driver.delete('rot_event', 'old', { bypassTenantAudit: true })).toBe(false); + + // deleteMany with a temporal cutoff spans every live shard (the Reaper's + // retention trim on a rotated object). + await driver.create('rot_event', { id: 'old2', payload: 'p2', created_at: new Date(T0 - 10 * DAY_MS) }, { bypassTenantAudit: true }); + const deleted = await driver.deleteMany( + 'rot_event', + { object: 'rot_event', where: { created_at: { $lt: new Date(T0).toISOString() } } }, + { bypassTenantAudit: true }, + ); + expect(deleted).toBe(1); + const rest = await driver.find('rot_event', { object: 'rot_event' }); + expect(rest.map((r: any) => r.id)).toEqual(['new']); + }); + + it('rotateShards is idempotent within the same period', async () => { + await driver.initObjects([ROTATED_OBJECT]); + const first = await driver.rotateShards(ROTATED_OBJECT, T0); + await driver.create('rot_event', { id: 'a', payload: 'x', created_at: new Date(T0) }, { bypassTenantAudit: true }); + const second = await driver.rotateShards(ROTATED_OBJECT, T0 + 3_600_000); // +1h, same day + expect(second.current).toBe(first.current); + expect(second.dropped).toEqual([]); + expect(await driver.count('rot_event', { object: 'rot_event' })).toBe(1); + }); +}); diff --git a/packages/plugins/driver-sql/src/sql-driver.ts b/packages/plugins/driver-sql/src/sql-driver.ts index 80a1b397e1..26f5d0d368 100644 --- a/packages/plugins/driver-sql/src/sql-driver.ts +++ b/packages/plugins/driver-sql/src/sql-driver.ts @@ -731,7 +731,9 @@ export class SqlDriver implements IDataDriver { this.injectTenantOnInsert(object, toInsert, options); await this.fillAutoNumberFields(object, toInsert, options); - const builder = this.getBuilder(object, options); + // Rotation (ADR-0057 P2): the base name is a read-only view — new rows + // land in the current shard. + const builder = this.getBuilder(this.rotationWriteTarget(object) ?? object, options); const formatted = this.applyWriteColumnMap(object, this.formatInput(object, toInsert)); this.stampInsertTimestamps(object, formatted); @@ -1109,6 +1111,8 @@ export class SqlDriver implements IDataDriver { async update(object: string, id: string | number, data: Record, options?: DriverOptions): Promise { this.auditMissingTenant(object, 'update', options); + const rotationShards = this.rotationShardsOf(object); + if (rotationShards) return this.rotatedUpdateById(object, rotationShards, id, data, options); const builder = this.getBuilder(object, options).where('id', id); this.applyTenantScope(builder, object, options); const formatted = this.applyWriteColumnMap(object, this.formatInput(object, data)); @@ -1151,7 +1155,10 @@ export class SqlDriver implements IDataDriver { this.stampInsertTimestamps(object, formatted); const mergeKeys = conflictKeys && conflictKeys.length > 0 ? conflictKeys : ['id']; - const builder = this.getBuilder(object, options); + // Rotation: conflict-merge is scoped to the CURRENT shard (telemetry is + // effectively append-only; a cross-shard upsert would need a probe-first + // strategy nothing on the platform requires today). + const builder = this.getBuilder(this.rotationWriteTarget(object) ?? object, options); // `created_at` is insert-only — never overwrite it when an existing row is // merged on conflict (the stamped/seeded value belongs to the original // insert). Everything else (incl. `updated_at`) merges as before, so an @@ -1168,6 +1175,17 @@ export class SqlDriver implements IDataDriver { async delete(object: string, id: string | number, options?: DriverOptions): Promise { this.auditMissingTenant(object, 'delete', options); + const rotationShards = this.rotationShardsOf(object); + if (rotationShards) { + // The row lives in exactly one shard — probe newest-first. + for (const shard of rotationShards) { + const builder = this.getBuilder(shard, options).where('id', id); + this.applyTenantScope(builder, object, options); + const count = await builder.delete(); + if (count > 0) return true; + } + return false; + } const builder = this.getBuilder(object, options).where('id', id); this.applyTenantScope(builder, object, options); const count = await builder.delete(); @@ -1216,7 +1234,7 @@ export class SqlDriver implements IDataDriver { this.stampInsertTimestamps(object, formatted); return formatted; }); - const builder = this.getBuilder(object, options); + const builder = this.getBuilder(this.rotationWriteTarget(object) ?? object, options); const result = await builder.insert(formattedRows).returning('*'); // Read-back parity with create(): JSON columns come back as their stored // strings from `returning('*')` — decode them so batch callers see the @@ -1242,27 +1260,62 @@ export class SqlDriver implements IDataDriver { async bulkDelete(object: string, ids: Array, options?: DriverOptions): Promise { this.auditMissingTenant(object, 'bulkDelete', options); - const builder = this.getBuilder(object, options).whereIn('id', ids); - this.applyTenantScope(builder, object, options); - await builder.delete(); + for (const target of this.rotationShardsOf(object) ?? [object]) { + const builder = this.getBuilder(target, options).whereIn('id', ids); + this.applyTenantScope(builder, object, options); + await builder.delete(); + } } async updateMany(object: string, query: QueryAST, data: any, options?: DriverOptions): Promise { this.auditMissingTenant(object, 'updateMany', options); - const builder = this.getBuilder(object, options); - this.applyTenantScope(builder, object, options); - if (query.where) this.applyFilters(builder, query.where); - const count = await builder.update(data); - return count || 0; + let total = 0; + for (const target of this.rotationShardsOf(object) ?? [object]) { + const builder = this.getBuilder(target, options); + this.applyTenantScope(builder, object, options); + if (query.where) this.applyFilters(builder, query.where); + total += (await builder.update(data)) || 0; + } + return total; } async deleteMany(object: string, query: QueryAST, options?: DriverOptions): Promise { this.auditMissingTenant(object, 'deleteMany', options); - const builder = this.getBuilder(object, options); - this.applyTenantScope(builder, object, options); - if (query.where) this.applyFilters(builder, query.where); - const count = await builder.delete(); - return count || 0; + let total = 0; + for (const target of this.rotationShardsOf(object) ?? [object]) { + const builder = this.getBuilder(target, options); + this.applyTenantScope(builder, object, options); + if (query.where) this.applyFilters(builder, query.where); + total += (await builder.delete()) || 0; + } + return total; + } + + /** By-id update for a rotation-managed object: the row lives in exactly one + * shard — probe newest-first, mirroring the un-rotated {@link update}. */ + protected async rotatedUpdateById( + object: string, + shards: string[], + id: string | number, + data: Record, + options?: DriverOptions, + ): Promise { + const formatted = this.applyWriteColumnMap(object, this.formatInput(object, data)); + if (this.tablesWithTimestamps.has(object)) { + formatted.updated_at = this.isSqlite ? new Date().toISOString() : this.knex.fn.now(); + } + for (const shard of shards) { + const builder = this.getBuilder(shard, options).where('id', id); + this.applyTenantScope(builder, object, options); + const count = await builder.update(formatted); + if (count > 0) { + const readback = this.getBuilder(shard, options).where('id', id); + this.applyTenantScope(readback, object, options); + const updated = await readback.first(); + return this.formatOutput(object, updated) || null; + } + } + return null; } async count(object: string, query?: QueryAST, options?: DriverOptions): Promise { @@ -1582,6 +1635,232 @@ export class SqlDriver implements IDataDriver { await this.knex.schema.dropTableIfExists(object); } + /** + * Reclaim free pages after bulk deletions (ADR-0057 §3.4). On SQLite this + * issues `PRAGMA incremental_vacuum`, returning freelist pages to the OS — + * it pairs with the `auto_vacuum=INCREMENTAL` default set in {@link connect} + * (files created before that default need one full `VACUUM` to adopt it). + * Postgres/MySQL manage space via their own vacuum/purge machinery, so this + * is a no-op there. + */ + async reclaimSpace(_options?: DriverOptions): Promise { + if (!this.isSqlite) return; + await this.knex.raw('PRAGMA incremental_vacuum'); + } + + // ── Data-lifecycle rotation (ADR-0057 P2) ───────────────────────────────── + // + // High-frequency telemetry declared with `lifecycle.storage.strategy = + // 'rotation'` is physically time-sharded: writes land in the CURRENT shard + // table (`
__r`), reads go through a UNION ALL view named after + // the base table (so every query path is unchanged), and expiry is an O(1) + // `DROP TABLE` of the oldest shard — real space reclamation with no + // row-by-row delete (the ServiceNow Table Rotation model, ADR-0057 §3.3). + // + // The view is READ-ONLY by design (SQLite views reject writes and don't + // support RETURNING): the driver redirects every write path shard-wise + // instead — inserts to the current shard; by-id updates/deletes probe each + // live shard; bulk updates/deletes fan out and sum. SQLite-only + // ({@link supportsRotation}); on other dialects the LifecycleService falls + // back to an age-based reap, so the declared bound holds everywhere — only + // the reclamation mechanics differ. + + /** table → live rotation state (shard names newest-first + write target). */ + protected rotationStateByTable = new Map(); + + get supportsRotation(): boolean { + return this.isSqlite; + } + + /** Live shard set (newest first) when `object` is rotation-managed. */ + protected rotationShardsOf(object: string): string[] | undefined { + return this.rotationStateByTable.get(object)?.shards; + } + + /** The shard new rows land in when `object` is rotation-managed. */ + protected rotationWriteTarget(object: string): string | undefined { + return this.rotationStateByTable.get(object)?.current; + } + + /** + * Shard key for an instant: `day` → UTC `YYYYMMDD`, `week` → the UTC + * Monday's `YYYYMMDD`, `month` → `YYYYMM`. Keys of one unit sort + * lexicographically = chronologically, which `ensureRotation` relies on. + */ + protected rotationShardKey(nowMs: number, unit: 'day' | 'week' | 'month'): string { + const ymd = (ms: number) => new Date(ms).toISOString().slice(0, 10).replace(/-/g, ''); + if (unit === 'month') return ymd(nowMs).slice(0, 6); + if (unit === 'week') { + const dow = (new Date(nowMs).getUTCDay() + 6) % 7; // Monday = 0 + return ymd(nowMs - dow * 86_400_000); + } + return ymd(nowMs); + } + + /** + * Public Rotator entry point (called by the LifecycleService each sweep and + * by {@link initObjects} at boot). Idempotent: ensures the current shard + + * read view exist, adopts a legacy pre-rotation base table as the current + * shard, column-syncs every retained shard so the UNION stays uniform, and + * drops shards past the `shards × unit` window. + */ + async rotateShards( + objectDef: { name: string; fields?: Record; lifecycle?: any }, + nowMs: number = Date.now(), + ): Promise<{ object: string; current: string; shards: string[]; dropped: string[] }> { + this.assertSchemaMutable('rotateShards'); + const policy = objectDef.lifecycle?.storage; + if (!policy || policy.strategy !== 'rotation') { + throw new Error(`[sql-driver] rotateShards: '${objectDef.name}' declares no lifecycle.storage rotation policy`); + } + if (!this.supportsRotation) { + throw new Error(`[sql-driver] rotateShards: rotation is not supported on dialect '${this.dialectName}'`); + } + const tableName = StorageNameMapping.resolveTableName(objectDef as any); + return this.ensureRotation(tableName, objectDef, policy, nowMs); + } + + protected async ensureRotation( + tableName: string, + obj: { name: string; fields?: Record }, + policy: { shards: number; unit: 'day' | 'week' | 'month' }, + nowMs: number = Date.now(), + ): Promise<{ object: string; current: string; shards: string[]; dropped: string[] }> { + const current = `${tableName}__r${this.rotationShardKey(nowMs, policy.unit)}`; + + // Physical inventory: the base name (table before adoption, view after) + // and every existing shard. + const esc = tableName.replace(/[\\%_]/g, '\\$&'); + const raw: any = await this.knex.raw( + `SELECT name, type FROM sqlite_master WHERE name = ? OR (name LIKE ? ESCAPE '\\')`, + [tableName, `${esc}\\_\\_r%`], + ); + const rows: Array<{ name: string; type: string }> = Array.isArray(raw) ? raw : raw?.rows ?? []; + const baseType = rows.find((r) => r.name === tableName)?.type as 'table' | 'view' | undefined; + const shardNames = new Set( + rows.map((r) => r.name).filter((n) => n !== tableName && /__r\d{6,8}$/.test(n)), + ); + + // Adopt a legacy pre-rotation table: its whole history becomes the + // current shard (coarse, but safe — it then ages out of the window). + if (baseType === 'table') { + if (!shardNames.has(current)) { + await this.knex.schema.renameTable(tableName, current); + } else { + // Partial-failure recovery: both exist — merge, then drop the base. + await this.knex.raw(`INSERT INTO "${current}" SELECT * FROM "${tableName}"`); + await this.knex.schema.dropTable(tableName); + } + shardNames.add(current); + } + shardNames.add(current); + + // Time-based window: retain shards whose period falls inside the last + // `shards × unit` (ends at the current period); everything older is the + // O(1) reclaim. Count-based retention would silently stretch the window + // when rotation cadence has gaps. + const unitMs = { day: 86_400_000, week: 7 * 86_400_000, month: 30 * 86_400_000 }[policy.unit]; + const oldestRetainedKey = this.rotationShardKey(nowMs - (Math.max(1, policy.shards) - 1) * unitMs, policy.unit); + const parsed = [...shardNames].sort((a, b) => b.localeCompare(a)); + const keyOf = (n: string) => n.slice(tableName.length + 3); + const retained = parsed.filter((n) => keyOf(n) >= oldestRetainedKey); + const dropped = parsed.filter((n) => keyOf(n) < oldestRetainedKey); + + // Column-sync every retained shard (creates the current one; adds any + // newly declared columns to older shards so the UNION stays uniform). + for (const shard of retained) { + await this.ensureShardTable(shard, obj); + this.aliasShardBookkeeping(tableName, shard); + } + + for (const d of dropped) { + await this.knex.schema.dropTableIfExists(d); + } + + // Rebuild the read view over the retained set, newest shard first. An + // explicit column list (not `*`) keeps the view stable when old shards + // carry orphaned columns. + const cols = this.rotationColumnList(obj); + await this.knex.raw(`DROP VIEW IF EXISTS "${tableName}"`); + await this.knex.raw( + `CREATE VIEW "${tableName}" AS ` + retained.map((s) => `SELECT ${cols} FROM "${s}"`).join(' UNION ALL '), + ); + + this.rotationStateByTable.set(tableName, { shards: retained, current }); + if (dropped.length > 0) { + (this.logger as { info?: (msg: string) => void }).info?.( + `[sql-driver] rotated ${tableName}: ${retained.length} shard(s) live, dropped ${dropped.join(', ')}`, + ); + } + return { object: tableName, current, shards: retained, dropped }; + } + + /** Create/column-sync one physical shard table (mirrors the managed-table + * branch of {@link initObjects}, scoped to a shard). */ + protected async ensureShardTable(shardName: string, obj: { fields?: Record }): Promise { + const builtinColumns = new Set(['id', 'created_at', 'updated_at']); + const exists = await this.knex.schema.hasTable(shardName); + if (!exists) { + await this.knex.schema.createTable(shardName, (table) => { + table.string('id').primary(); + table.timestamp('created_at').defaultTo(this.knex.fn.now()); + table.timestamp('updated_at').defaultTo(this.knex.fn.now()); + for (const [name, field] of Object.entries(obj.fields ?? {})) { + if (builtinColumns.has(name)) continue; + this.createColumn(table, name, field); + } + }); + } else { + const columnInfo = await this.knex(shardName).columnInfo(); + const existingColumns = Object.keys(columnInfo); + await this.knex.schema.alterTable(shardName, (table) => { + for (const [name, field] of Object.entries(obj.fields ?? {})) { + if (!existingColumns.includes(name)) { + this.createColumn(table, name, field); + } + } + }); + } + + // Declared indexes per shard. Auto-derived names already embed the shard + // name; explicit names get a shard prefix so they can't collide across + // shards in the same database. + const declared = (obj as any).indexes; + if (Array.isArray(declared) && declared.length > 0) { + const colInfo = await this.knex(shardName).columnInfo(); + const perShard = declared.map((idx: any) => ({ + ...idx, + name: typeof idx?.name === 'string' && idx.name.trim() ? `${shardName}__${idx.name.trim()}` : undefined, + })); + await this.syncDeclaredIndexes(shardName, perShard, new Set(Object.keys(colInfo))); + } + } + + /** Quoted, deterministic column list for the rotation view. */ + protected rotationColumnList(obj: { fields?: Record }): string { + const builtin = ['id', 'created_at', 'updated_at']; + const declared = Object.keys(obj.fields ?? {}).filter((f) => !builtin.includes(f)); + return [...builtin, ...declared].map((c) => `"${c}"`).join(', '); + } + + /** + * Point every per-table bookkeeping map (read coercion, JSON/boolean + * columns, tenant scope, timestamp stamping) for a shard at the base + * table's entries, so a builder targeting a shard behaves exactly like one + * targeting the view. + */ + protected aliasShardBookkeeping(base: string, shard: string): void { + this.jsonFields[shard] = this.jsonFields[base] ?? []; + this.booleanFields[shard] = this.booleanFields[base] ?? []; + this.numericFields[shard] = this.numericFields[base] ?? []; + this.autoNumberFields[shard] = this.autoNumberFields[base] ?? []; + if (this.dateFields[base]) this.dateFields[shard] = this.dateFields[base]; + if (this.datetimeFields[base]) this.datetimeFields[shard] = this.datetimeFields[base]; + if (this.timeFields[base]) this.timeFields[shard] = this.timeFields[base]; + this.tenantFieldByTable[shard] = this.tenantFieldByTable[base] ?? null; + this.tablesWithTimestamps.add(shard); + } + /** * Resolve the per-table tenant-isolation column for a schema, honoring an * explicit tenancy opt-out. Single source of truth for both {@link initObjects} @@ -1778,6 +2057,16 @@ export class SqlDriver implements IDataDriver { this.autoNumberFields[tableName] = autoNumberCols; this.tenantFieldByTable[tableName] = tenantField; + // ADR-0057 P2: rotation-declared telemetry is physically time-sharded — + // the Rotator owns its DDL (shard tables + a read view under the base + // name); the plain create/alter path below would collide with the view. + const rotationPolicy = (obj as any).lifecycle?.storage; + if (rotationPolicy?.strategy === 'rotation' && this.supportsRotation) { + this.tablesWithTimestamps.add(tableName); + await this.ensureRotation(tableName, obj, rotationPolicy); + continue; + } + let exists = await this.knex.schema.hasTable(tableName); if (exists) { diff --git a/packages/plugins/plugin-audit/src/objects/sys-activity.object.ts b/packages/plugins/plugin-audit/src/objects/sys-activity.object.ts index 85923ae51f..fb956fa8bd 100644 --- a/packages/plugins/plugin-audit/src/objects/sys-activity.object.ts +++ b/packages/plugins/plugin-audit/src/objects/sys-activity.object.ts @@ -23,6 +23,15 @@ export const SysActivity = ObjectSchema.create({ icon: 'activity', isSystem: true, managedBy: 'append-only', + // ADR-0057: the highest-frequency telemetry table on the platform (the + // 260 MB dev.db regression was ~50% this table). 14 day-shards once the + // Rotator lands; the same 14d window is age-reaped until then. + lifecycle: { + class: 'telemetry', + retention: { maxAge: '14d' }, + storage: { strategy: 'rotation', shards: 14, unit: 'day' }, + reclaim: true, + }, description: 'Recent activity stream entries (lightweight, denormalized)', displayNameField: 'summary', nameField: 'summary', // [ADR-0079] canonical primary-title pointer (mirrors deprecated displayNameField) diff --git a/packages/plugins/plugin-audit/src/objects/sys-audit-log.object.ts b/packages/plugins/plugin-audit/src/objects/sys-audit-log.object.ts index 9820717d83..2f5fc536b1 100644 --- a/packages/plugins/plugin-audit/src/objects/sys-audit-log.object.ts +++ b/packages/plugins/plugin-audit/src/objects/sys-audit-log.object.ts @@ -20,6 +20,15 @@ export const SysAuditLog = ObjectSchema.create({ icon: 'scroll-text', isSystem: true, managedBy: 'append-only', + // ADR-0057: compliance ledger — retain hot 90d, then archive-then-delete. + // The LifecycleService NEVER hot-deletes rows with `archive` declared until + // the archive copy succeeded; deployments without an 'archive' datasource + // simply retain everything (today's behavior). + lifecycle: { + class: 'audit', + retention: { maxAge: '90d' }, + archive: { after: '90d', to: 'archive', keep: '7y' }, + }, description: 'Immutable audit trail for platform events', displayNameField: 'action', nameField: 'action', // [ADR-0079] canonical primary-title pointer (mirrors deprecated displayNameField) diff --git a/packages/services/service-automation/src/sys-automation-run.object.ts b/packages/services/service-automation/src/sys-automation-run.object.ts index 08aced1afa..ec02c0a77b 100644 --- a/packages/services/service-automation/src/sys-automation-run.object.ts +++ b/packages/services/service-automation/src/sys-automation-run.object.ts @@ -40,6 +40,13 @@ export const SysAutomationRun = ObjectSchema.create({ icon: 'pause-circle', isSystem: true, managedBy: 'system', + // ADR-0057 (#2786 "why now"): terminal run history is append-only + // telemetry — bounded like sys_job_run. Suspended runs are recent by + // definition (SLA-bounded), so a 30d age reap cannot strand a live run. + lifecycle: { + class: 'telemetry', + retention: { maxAge: '30d' }, + }, description: 'Durable automation run state: live suspended runs (resumable, ADR-0019) and terminal run history (completed / failed, for observability).', displayNameField: 'id', nameField: 'id', // [ADR-0079] canonical primary-title pointer (mirrors deprecated displayNameField) diff --git a/packages/services/service-messaging/src/objects/http-delivery.object.ts b/packages/services/service-messaging/src/objects/http-delivery.object.ts index 648c4b1dd4..6afd5ff610 100644 --- a/packages/services/service-messaging/src/objects/http-delivery.object.ts +++ b/packages/services/service-messaging/src/objects/http-delivery.object.ts @@ -32,6 +32,12 @@ export const HttpDelivery = ObjectSchema.create({ icon: 'globe', isSystem: true, managedBy: 'system', + // ADR-0057: webhook delivery attempts are debugging telemetry — 30d is + // ample for retry forensics. + lifecycle: { + class: 'telemetry', + retention: { maxAge: '30d' }, + }, userActions: { create: false, edit: false, delete: false, import: false }, description: 'Durable outbox row for one outbound-HTTP attempt (ADR-0018). Managed by @objectstack/service-messaging; do not write directly.', diff --git a/packages/services/service-messaging/src/objects/inbox-message.object.ts b/packages/services/service-messaging/src/objects/inbox-message.object.ts index dc780be9f2..68d5577a48 100644 --- a/packages/services/service-messaging/src/objects/inbox-message.object.ts +++ b/packages/services/service-messaging/src/objects/inbox-message.object.ts @@ -22,6 +22,12 @@ export const InboxMessage = ObjectSchema.create({ label: 'Inbox Message', pluralLabel: 'Inbox Messages', icon: 'inbox', + // ADR-0057: user-facing but ephemeral — expires with the pipeline's 90d + // window (the same bound NotificationRetention applies when enabled). + lifecycle: { + class: 'transient', + ttl: { field: 'created_at', expireAfter: '90d' }, + }, description: 'User-facing in-app notification rows materialized by the inbox messaging channel.', nameField: 'title', // [ADR-0079] canonical primary-title pointer (single-field titleFormat) titleFormat: '{title}', diff --git a/packages/services/service-messaging/src/objects/notification-delivery.object.ts b/packages/services/service-messaging/src/objects/notification-delivery.object.ts index 99b072a201..a63781f722 100644 --- a/packages/services/service-messaging/src/objects/notification-delivery.object.ts +++ b/packages/services/service-messaging/src/objects/notification-delivery.object.ts @@ -24,6 +24,11 @@ export const NotificationDelivery = ObjectSchema.create({ icon: 'send', isSystem: true, managedBy: 'system', + // ADR-0057: pipeline telemetry — same 90d window as sys_notification. + lifecycle: { + class: 'telemetry', + retention: { maxAge: '90d' }, + }, description: 'Durable per-recipient × channel delivery outbox (ADR-0030 Layer 4).', titleFormat: '{channel} → {recipient_id}', highlightFields: ['notification_id', 'recipient_id', 'channel', 'status', 'attempts'], diff --git a/packages/services/service-messaging/src/objects/notification-receipt.object.ts b/packages/services/service-messaging/src/objects/notification-receipt.object.ts index 37418dd635..0d9135cde6 100644 --- a/packages/services/service-messaging/src/objects/notification-receipt.object.ts +++ b/packages/services/service-messaging/src/objects/notification-receipt.object.ts @@ -26,6 +26,13 @@ export const NotificationReceipt = ObjectSchema.create({ icon: 'check-check', isSystem: true, managedBy: 'system', + // ADR-0057: read-state rows expire with the pipeline's 90d window — NOT + // shorter, or read notifications would resurface as unread while their + // inbox rows are still alive. + lifecycle: { + class: 'transient', + ttl: { field: 'created_at', expireAfter: '90d' }, + }, description: 'Per-recipient × channel receipt; the source of truth for notification read-state.', titleFormat: '{state}', highlightFields: ['notification_id', 'user_id', 'channel', 'state', 'at'], diff --git a/packages/spec/api-surface.json b/packages/spec/api-surface.json index 831dcd8995..426124a222 100644 --- a/packages/spec/api-surface.json +++ b/packages/spec/api-surface.json @@ -306,7 +306,12 @@ "JoinNodeSchema (const)", "JoinStrategy (const)", "JoinType (const)", + "LIFECYCLE_DURATION_REGEX (const)", "LOGICAL_OPERATORS (const)", + "Lifecycle (type)", + "LifecycleClass (type)", + "LifecycleClassSchema (const)", + "LifecycleSchema (const)", "LocationCoordinates (type)", "LocationCoordinatesSchema (const)", "LogicalOperatorKey (type)", diff --git a/packages/spec/liveness/object.json b/packages/spec/liveness/object.json index 4f37ff7f1f..99f8e9ca11 100644 --- a/packages/spec/liveness/object.json +++ b/packages/spec/liveness/object.json @@ -61,6 +61,12 @@ "evidence": "packages/objectql/src/engine.ts:1147", "note": "driver routing." }, + "lifecycle": { + "status": "live", + "evidence": "packages/objectql/src/lifecycle/lifecycle-service.ts", + "proof": "packages/dogfood/test/storage-growth.dogfood.test.ts#adr0057-lifecycle-bounded-growth", + "note": "ADR-0057 data lifecycle contract. The platform LifecycleService (registered by ObjectQLPlugin, default-on) reaps rows past retention.maxAge / ttl.expireAfter under a system context, bounds rotation-declared telemetry by shards×unit, never hot-deletes archive-declared audit ledgers unarchived, and reclaims driver space (SQLite incremental_vacuum) after each sweep. Non-record classes without a bounding policy are rejected at parse time (LifecycleSchema superRefine)." + }, "external": { "status": "live", "evidence": "packages/runtime/src/external-validation-plugin.ts", diff --git a/packages/spec/scripts/liveness/proof-registry.mts b/packages/spec/scripts/liveness/proof-registry.mts index 24ebc04da4..04aaaeb54f 100644 --- a/packages/spec/scripts/liveness/proof-registry.mts +++ b/packages/spec/scripts/liveness/proof-registry.mts @@ -118,6 +118,18 @@ export const HIGH_RISK_CLASSES: HighRiskClass[] = [ // the triggering user — the security property the proof guards in both directions. ledgerBindings: [{ type: 'flow', path: 'runAs' }], }, + { + id: 'data-lifecycle', + label: 'Data lifecycle (ADR-0057)', + summary: + 'declared lifecycle policies actually bound growth at runtime — the Reaper deletes past-window rows, record-class/business data stays untouched, archive-declared audit ledgers are never hot-deleted unarchived.', + proofId: 'adr0057-lifecycle-bounded-growth', + proofRef: 'packages/dogfood/test/storage-growth.dogfood.test.ts#adr0057-lifecycle-bounded-growth', + bound: true, + // `lifecycle` declares the retention contract the LifecycleService enforces — + // a declaration whose sweeper stopped running is dead surface (ADR-0049). + ledgerBindings: [{ type: 'object', path: 'lifecycle' }], + }, { id: 'form-widget', label: 'Form layout / section / widget', diff --git a/packages/spec/scripts/liveness/proof-registry.test.ts b/packages/spec/scripts/liveness/proof-registry.test.ts index 1f6688a171..3fd5735d42 100644 --- a/packages/spec/scripts/liveness/proof-registry.test.ts +++ b/packages/spec/scripts/liveness/proof-registry.test.ts @@ -109,6 +109,7 @@ describe('registry invariants', () => { 'permission/rowLevelSecurity.using', 'dataset/dimensions.dateGranularity', 'object/sharingModel', + 'object/lifecycle', ].sort(), ); }); diff --git a/packages/spec/src/contracts/data-driver.ts b/packages/spec/src/contracts/data-driver.ts index 7b1d5c685f..73380a443b 100644 --- a/packages/spec/src/contracts/data-driver.ts +++ b/packages/spec/src/contracts/data-driver.ts @@ -178,6 +178,15 @@ export interface IDataDriver { /** Drop the underlying table or collection (destructive) */ dropTable(object: string, options?: DriverOptions): Promise; + /** + * Reclaim free space after bulk deletions (ADR-0057 §3.4). SQLite issues + * `PRAGMA incremental_vacuum` (pairs with the `auto_vacuum=INCREMENTAL` + * connect-time default); engines with their own background reclamation + * (Postgres autovacuum) may no-op. Optional — the LifecycleService calls + * it best-effort after every sweep that deleted rows. + */ + reclaimSpace?(options?: DriverOptions): Promise; + /** * Analyze query performance. * Returns execution plan without executing the query (optional). diff --git a/packages/spec/src/data/object.test.ts b/packages/spec/src/data/object.test.ts index b3734a6839..77629bc3d0 100644 --- a/packages/spec/src/data/object.test.ts +++ b/packages/spec/src/data/object.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect } from 'vitest'; -import { ObjectSchema, ObjectCapabilities, IndexSchema, ObjectFieldGroupSchema, ObjectExternalBindingSchema, ObjectAccessConfigSchema, type ServiceObject } from './object.zod'; +import { ObjectSchema, ObjectCapabilities, IndexSchema, ObjectFieldGroupSchema, ObjectExternalBindingSchema, ObjectAccessConfigSchema, LifecycleSchema, type ServiceObject } from './object.zod'; describe('ObjectCapabilities', () => { it('should apply default values correctly', () => { @@ -36,6 +36,91 @@ describe('ObjectCapabilities', () => { }); }); +describe('LifecycleSchema (ADR-0057)', () => { + it('accepts the ADR §3.2 telemetry rotation shape', () => { + const result = LifecycleSchema.safeParse({ + class: 'telemetry', + retention: { maxAge: '14d' }, + storage: { strategy: 'rotation', shards: 14, unit: 'day' }, + reclaim: true, + }); + expect(result.success).toBe(true); + }); + + it('accepts the ADR §3.2 audit archive-then-delete shape', () => { + const result = LifecycleSchema.safeParse({ + class: 'audit', + retention: { maxAge: '90d' }, + archive: { after: '90d', to: 'datalake', keep: '7y' }, + }); + expect(result.success).toBe(true); + }); + + it('accepts the ADR §3.2 transient ttl shape', () => { + const result = LifecycleSchema.safeParse({ + class: 'transient', + ttl: { field: 'created_at', expireAfter: '7d' }, + }); + expect(result.success).toBe(true); + }); + + it('accepts a bare record class (permanent, no policies)', () => { + expect(LifecycleSchema.safeParse({ class: 'record' }).success).toBe(true); + }); + + it('rejects a non-record class with no bounding policy (§3.5 enforce-or-remove)', () => { + for (const cls of ['audit', 'telemetry', 'transient', 'event'] as const) { + const result = LifecycleSchema.safeParse({ class: cls }); + expect(result.success).toBe(false); + } + }); + + it('rejects retention/ttl/storage/archive on a record class', () => { + const result = LifecycleSchema.safeParse({ + class: 'record', + retention: { maxAge: '30d' }, + }); + expect(result.success).toBe(false); + }); + + it('rejects an archive window that does not start where the hot window ends', () => { + const result = LifecycleSchema.safeParse({ + class: 'audit', + retention: { maxAge: '90d' }, + archive: { after: '30d', to: 'datalake' }, + }); + expect(result.success).toBe(false); + }); + + it('rejects malformed duration literals', () => { + for (const bad of ['14', 'd14', '14 days', '2mo', '-3d', '1.5d']) { + const result = LifecycleSchema.safeParse({ + class: 'telemetry', + retention: { maxAge: bad }, + }); + expect(result.success).toBe(false); + } + }); + + it('is accepted as an object-level property by ObjectSchema.create', () => { + const obj = ObjectSchema.create({ + name: 'my_trace', + fields: {}, + lifecycle: { + class: 'telemetry', + retention: { maxAge: '14d' }, + }, + }); + expect(obj.lifecycle?.class).toBe('telemetry'); + expect(obj.lifecycle?.retention?.maxAge).toBe('14d'); + }); + + it('objects without a lifecycle block stay back-compatible (undefined = record semantics)', () => { + const obj = ObjectSchema.create({ name: 'plain_object', fields: {} }); + expect(obj.lifecycle).toBeUndefined(); + }); +}); + describe('IndexSchema', () => { it('should accept basic index definition', () => { const index = { diff --git a/packages/spec/src/data/object.zod.ts b/packages/spec/src/data/object.zod.ts index 9f2b8a9347..5b9838edb0 100644 --- a/packages/spec/src/data/object.zod.ts +++ b/packages/spec/src/data/object.zod.ts @@ -263,6 +263,88 @@ export const VersioningConfigSchema = lazySchema(() => z.object({ versionField: z.string().default('version').describe('Field name for version number/timestamp'), })); +/** + * Data Lifecycle (ADR-0057) + * + * Declares how long an object's data lives and how its space is reclaimed — + * the axis validation/permissions never covered. Enforced at runtime by the + * platform-owned LifecycleService (`@objectstack/objectql`): Reaper (TTL/age + * batch delete), Rotator (time-shard + DROP oldest), Archiver (cold-store + * copy then delete). A declared policy with no runtime consumer is a spec + * defect (ADR-0049 enforce-or-remove); the liveness gate requires every + * non-`record` class to declare `retention`, `ttl`, or rotation `storage`. + */ + +/** + * Lifecycle class — what persistence contract the object's data carries. + * + * | class | contract | + * |-------------|-------------------------------------------------| + * | `record` | business truth — permanent, recoverable | + * | `audit` | compliance ledger — retain → archive → delete | + * | `telemetry` | high-frequency log — rotation, short retention | + * | `transient` | ephemeral state — TTL auto-expire | + * | `event` | event-bus messages — very short TTL | + * + * `record` is the back-compat default: an object with no `lifecycle` block + * behaves exactly as today (immortal data). + */ +export const LifecycleClassSchema = z.enum(['record', 'audit', 'telemetry', 'transient', 'event']); + +/** + * Duration literal: `` where unit is h(ours), d(ays), w(eeks) or + * y(ears) — e.g. `'6h'`, `'14d'`, `'12w'`, `'7y'`. Parsed by + * `@objectstack/objectql` `parseLifecycleDuration`. + */ +export const LIFECYCLE_DURATION_REGEX = /^\d+(h|d|w|y)$/; +const lifecycleDuration = (what: string) => + z.string().regex(LIFECYCLE_DURATION_REGEX, `${what} must be a duration literal like '6h', '14d', '12w' or '7y'`); + +export const LifecycleSchema = lazySchema(() => z.object({ + class: LifecycleClassSchema.describe( + 'Persistence contract: record (business truth, permanent) | audit (compliance ledger) | telemetry (high-freq log) | transient (ephemeral state) | event (bus messages).', + ), + retention: z.object({ + maxAge: lifecycleDuration('retention.maxAge').describe('Rows older than this (by created_at) are deleted by the Reaper — or archived first when `archive` is set.'), + }).optional().describe('Age-based retention window enforced by the LifecycleService Reaper.'), + ttl: z.object({ + field: z.string().describe('Timestamp field the TTL is measured from (e.g. created_at, expires_at).'), + expireAfter: lifecycleDuration('ttl.expireAfter').describe('Rows expire this long after `field` and are deleted by the Reaper.'), + }).optional().describe('Per-row TTL auto-expiry (transient/event classes).'), + storage: z.object({ + strategy: z.literal('rotation').describe('Time-shard the table; rotate by DROPping the oldest shard (O(1) reclaim).'), + shards: z.number().int().min(2).describe('Number of shards retained; total window = shards × unit.'), + unit: z.enum(['day', 'week', 'month']).describe('Time width of one shard.'), + }).optional().describe('Physical storage strategy for high-frequency telemetry (LifecycleService Rotator).'), + archive: z.object({ + after: lifecycleDuration('archive.after').describe('Rows older than this are copied to the archive datasource before hot deletion.'), + to: z.string().describe('Target datasource name for cold storage. When it is not registered, the Archiver skips (audit rows are then retained, never dropped unarchived).'), + keep: lifecycleDuration('archive.keep').optional().describe('How long archived rows are kept in cold storage (undefined = forever).'), + }).optional().describe('Cold-store archival (LifecycleService Archiver) — audit-class hot→cold hand-off.'), + reclaim: z.boolean().optional().describe('Run driver space reclamation (SQLite incremental_vacuum) after sweeping this object. Default true for non-record classes.'), +}).superRefine((lc, ctx) => { + // ADR-0057 §3.5: a non-`record` lifecycle class with no bounding policy is a + // false surface — the object would still grow forever. Enforce-or-remove. + if (lc.class !== 'record' && !lc.retention && !lc.ttl && !lc.storage) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: `lifecycle.class '${lc.class}' requires at least one bounding policy: retention, ttl, or storage (rotation) — ADR-0057 §3.5`, + }); + } + if (lc.class === 'record' && (lc.retention || lc.ttl || lc.storage || lc.archive)) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: `lifecycle.class 'record' is permanent business truth — retention/ttl/storage/archive policies are not allowed on it (ADR-0057 §3.1)`, + }); + } + if (lc.archive && lc.retention && lc.archive.after !== lc.retention.maxAge) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: `lifecycle.archive.after ('${lc.archive.after}') must equal retention.maxAge ('${lc.retention.maxAge}') — the hot window ends where the archive begins`, + }); + } +})); + /** * Object Field Group Schema — MVP (data-layer protocol) @@ -607,7 +689,11 @@ const ObjectSchemaBase = z.object({ // Versioning configuration versioning: VersioningConfigSchema.optional().describe('Record versioning and history tracking configuration'), - + + // Data lifecycle (ADR-0057) — retention / rotation / archival contract, + // enforced by the LifecycleService. Absent = `record` (today's behavior). + lifecycle: LifecycleSchema.optional().describe('Data lifecycle contract (ADR-0057): class + retention/ttl/rotation/archive policies enforced by the platform LifecycleService.'), + // Partitioning strategy /** @@ -1112,6 +1198,8 @@ export type TenancyConfig = z.infer; export type ObjectAccessConfig = z.infer; export type SoftDeleteConfig = z.infer; export type VersioningConfig = z.infer; +export type LifecycleClass = z.infer; +export type Lifecycle = z.infer; /** * Resolved CRUD affordance matrix for an object — what generic