/** * Domain data form (`ctx.storage.domain`): schema-validated, change-emitting * KV domains over storage backends. The single implementation of the domain * layer — consumers depend on this package and never touch backends directly. * Plugin `Config` is schemastery; record schemas inside domain specs are zod * (see `src/spec.ts` for the split rationale). * @module @deepseek-ai/dsh-storage-domain */ import type { Context } from '@deepseek-ai/cordis' import z from '@deepseek-ai/schemastery' import { storageBackendServiceKey } from '@deepseek-ai/dsh-storage' import { DomainError } from './error.ts' import { descriptorOf } from './spec.ts' import type { DomainSpec } from './spec.ts' import { DomainImpl } from './domain.ts' import type { Domain } from './domain.ts' export { DomainError } from './error.ts' export type { DomainErrorCode, DomainErrorOptions, InvalidRecordDetail } from './error.ts' export { defineDomain, domainTable, descriptorOf } from './spec.ts' export type { DomainSpec, DomainGlobalSpec, DomainTableSpec, TableKeyOf, TableValueOf, GlobalValueOf, } from './spec.ts' export type { DomainChanged } from './events.ts' export type { Domain, DomainGlobal, DomainGlobalHandleOf, KvTable } from './domain.ts' declare module '@deepseek-ai/dsh-storage' { interface StorageForms { domain: DomainFacility } } declare module '@deepseek-ai/cordis' { interface Context { storageDomain: DomainFacility } } /** Cordis plugin name. */ export const name = 'storage-domain' /** The storage hub must be present before the form can mount. */ export const inject = ['storage'] /** * Plugin config. Which backend serves which domain is decided here, not * globally on the hub: `backend` is the default route and `routes` overrides * it per domain name. A route naming an unregistered backend fails loud at * `open` with `backend-not-found`. */ export interface Config { /** Default backend name for every domain without an explicit route. Required: there is no universally correct medium. */ backend: string /** Per-domain overrides: domain name → backend name. */ routes?: Record } export const Config: z = z.object({ backend: z.string().required(), routes: z.dict(z.string()).default({}), }) /** * The mounted domain facility. Opens declared domains over routed backends; * one facility instance owns the open-domain table and enforces single-open * per domain name. */ export class DomainFacility { private readonly domains = new Map() /** Names reserved by an in-flight or completed open, so concurrent opens of one name fail loud. */ private readonly reserved = new Set() /** * @param ctx - Context of the domain plugin; open-domain effects and change * events attach here. * @param config - Validated plugin config. */ constructor( private readonly ctx: Context, private readonly config: Config, ) {} /** * Open one declared domain. Steps, each failing the whole call: reject a * name that is already open (`already-open`); resolve the backend route * (`backend-not-found` passes through from the hub); require its `kv` facet * (`facet-unsupported`); open the unit projected from the spec (backend * `version-mismatch`/`malformed-medium` pass through); load and validate * every stored record against the spec's zod schemas (`invalid-record` * with the offending table and key); construct the domain. * * Lifecycle: the CALLER owns the returned handle and closes it via * `Domain.close()` (typically as its own `ctx.effect` disposer) — the * facility does not tie the domain to any consumer fiber. Domains still * open when the facility unmounts are closed by the plugin disposer. * @param spec - The domain declaration, typically from `defineDomain`. * @returns the opened domain handle, typed by the spec. */ async open(spec: S): Promise> { if (this.reserved.has(spec.name)) { throw new DomainError('already-open', `domain '${spec.name}' is already open`) } this.reserved.add(spec.name) try { const backendName = this.config.routes?.[spec.name] ?? this.config.backend const backend = this.ctx.storage.backend.get(backendName) if (!backend.kv) { throw new DomainError( 'facet-unsupported', `backend '${backendName}' routed for domain '${spec.name}' has no kv facet`, ) } const unit = await backend.kv.open(descriptorOf(spec)) try { const snapshot = await unit.loadAll() const tables = new Map>() for (const [table, tableSpec] of Object.entries(spec.tables)) { const records = new Map() for (const [key, raw] of Object.entries(snapshot.tables[table] ?? {})) { records.set(key, parseRecord(spec.name, table, key, () => tableSpec.valueSchema.parse(raw))) } tables.set(table, records) } // A null stored global means "never written": serve `initial` without // materializing it — the first `set` writes. const globalSpec = spec.global const globalValue = globalSpec === undefined ? undefined : snapshot.global === null ? globalSpec.initial : parseRecord(spec.name, '', '', () => globalSpec.schema.parse(snapshot.global)) // The onClosed hook runs strictly after teardown completes: writes // landing during the drain still emit domain/changed, and the domain // stays resolvable (the package invariant cross-checks each event) // until fully closed — only then does the name free up for reopening. const domain: DomainImpl = new DomainImpl(this.ctx, spec, unit, tables, globalValue, () => { this.domains.delete(spec.name) this.reserved.delete(spec.name) }) this.domains.set(spec.name, domain) // The single type-erasure point: DomainImpl is the untyped runtime, // Domain the spec-typed view; the unknown hop is required because // S's conditional global-handle type stays unresolved here. return domain as unknown as Domain } catch (error) { await unit.close() throw error } } catch (error) { // Any failure means the domain never registered (nothing can throw // after it), so releasing the name reservation is unconditional. this.reserved.delete(spec.name) throw error } } /** * Look up an open domain by name, untyped. Diagnostic surface (the package * invariant cross-checks change events against live domain state); typed * consumers hold the handle returned by {@link open}. * @param name - Domain name. * @returns the open domain runtime, or `undefined` when not open. */ get(name: string): DomainImpl | undefined { return this.domains.get(name) } /** * Close every domain still open on this facility. The unmount path for * consumers that never called `Domain.close()` themselves; closing is * idempotent, so double-closing an already-closed domain is harmless. * @returns resolution after every unit is released. */ async closeAll(): Promise { await Promise.all([...this.domains.values()].map(domain => domain.close())) } } /** Run one zod parse, translating failure to `invalid-record` with its location. */ function parseRecord(domain: string, table: string, key: string, parse: () => T): T { try { return parse() } catch (error) { const slot = table === '' ? 'global' : `record '${key}' in table '${table}'` throw new DomainError( 'invalid-record', `domain '${domain}': stored ${slot} does not match its schema`, { detail: { table, key }, cause: error }, ) } } /** * Mount the domain data form on the storage hub. * @param ctx - Plugin context. * @param config - Validated plugin config. * @returns resolution after an already-available backend set activates the form. */ export function apply(ctx: Context, config: Config): Promise { const backendServices = [...new Set([ config.backend, ...Object.values(config.routes ?? {}), ])].map(storageBackendServiceKey) const fiber = ctx.inject(backendServices, (domainCtx) => { const facility = new DomainFacility(domainCtx, config) domainCtx.effect(() => { const unmount = domainCtx.storage.mount('domain', facility) return async () => { // Close leftovers before unmounting: draining writes still emit // domain/changed, whose invariant resolves the facility through the hub. await facility.closeAll() unmount() } }) domainCtx.provide('storageDomain', facility) }) return Promise.resolve(fiber).then(() => {}) }