diff --git a/src/services/dataset-service.ts b/src/services/dataset-service.ts index d373ca6..b706463 100644 --- a/src/services/dataset-service.ts +++ b/src/services/dataset-service.ts @@ -1,116 +1,72 @@ /** - * src/services/dataset-service.ts - * ================================ - * Owns 0pt.dataset.schema.v1 — the registry of 0pt datasets (raw / structured - * / enriched / curated / feature / derived outputs). This is the thing - * `DatasetIcebergMappingService` (src/services/dataset-iceberg-mapping-service.ts) - * and `IcebergWriteService` (src/services/iceberg-write-service.ts) look a - * `dataset_id` up against before pointing an Iceberg table mapping or a write - * at it. + * Owns 0pt.dataset.schema.v1 — the canonical registry of 0pt datasets. * - * Deliberately small: it is a reference/registry entry, NOT dataset content, - * schema, or quality state — see docs/ICEBERG_WRITE_BOUNDARY.md for that - * boundary. In-memory reference implementation (same convention as the other - * 0pt services in this repo; a real deployment backs this with `opt_datasets`). - * - * In-memory scope: each `new DatasetService()` owns its own registry, the same - * pre-existing limitation source-service.ts and every other P0 registry - * service documents (docs/P0_QUERY_SOURCE_FLOW.md's "Known gaps"). Callers - * that need a shared view across API routes pass an existing instance in via - * the `datasetService` constructor option — see - * src/services/lakehouse-mapping-registry.ts, which does exactly that for the - * whole app/api/0pt/lakehouse/mappings route group. - * - * Registering a dataset emits a best-effort AUD0 `0pt.dataset_registered` - * event; with no AUD0_URL configured that emission is a no-op (see - * aud0-client.ts), so this stays hermetic in tests and local dev. + * Storage is Postgres-backed through DatasetRepository whenever DATABASE_URL + * is configured. Non-production tests/local development may use the in-memory + * repository; production without a persistent store fails at construction via + * resolveStoreMode. Dataset content remains outside this registry. */ import { createAud0Client, type Aud0Client } from "../clients/aud0-client"; +import { + createDatasetRepository, + type DatasetListFilter, + type DatasetRepository, +} from "../repositories/dataset-repository"; import type { Dataset, DatasetStatus, DatasetType, CreateDatasetInput } from "../types/dataset"; -// Re-exported so the historical `import { Dataset, DatasetType, ... } from -// "./dataset-service"` call shape keeps working alongside the canonical -// `../types/dataset` module. export type { Dataset, DatasetStatus, DatasetType, CreateDatasetInput }; -/** Alias so both historical input-type names resolve to the same shape. */ export type RegisterDatasetInput = CreateDatasetInput; -/** - * Storage shape: nullable columns, exactly as the eventual `opt_datasets` - * table would hold them. `toDataset()` converts back to the `Dataset` DTO. - */ -interface DatasetRecord { - dataset_id: string; - tenant_id: string; - project_id: string; - name: string; - dataset_type: string; - status: string; - description: string | null; - correlation_id: string | null; - created_at: string; - updated_at: string | null; - created_by: string | null; +export interface DatasetServiceOptions { + aud0BaseUrl?: string; + databaseUrl?: string; + repository?: DatasetRepository; +} + +function normalizeOptions(input?: string | DatasetServiceOptions): DatasetServiceOptions { + return typeof input === "string" ? { aud0BaseUrl: input } : input ?? {}; } export class DatasetService { - private datasets: Map = new Map(); - private aud0: Aud0Client; + private readonly repo: DatasetRepository; + private readonly aud0: Aud0Client; - constructor(aud0BaseUrl?: string) { - this.aud0 = createAud0Client(aud0BaseUrl); + /** + * A string argument remains supported as the historical aud0BaseUrl form; + * new callers should use the options object so repository/database injection + * is explicit and testable. + */ + constructor(input?: string | DatasetServiceOptions) { + const opts = normalizeOptions(input); + this.aud0 = createAud0Client(opts.aud0BaseUrl); + this.repo = opts.repository ?? createDatasetRepository(opts.databaseUrl); } async registerDataset(input: CreateDatasetInput): Promise { - if (this.datasets.has(input.dataset_id)) { - throw new Error(`Dataset ${input.dataset_id} already exists`); - } - if (!input.name) { - throw new Error("name is required"); - } + const dataset = await this.repo.create(input); - const now = new Date().toISOString(); - const record: DatasetRecord = { - dataset_id: input.dataset_id, - tenant_id: input.tenant_id, - project_id: input.project_id, - name: input.name, - dataset_type: input.dataset_type, - status: input.status ?? "active", - description: input.description ?? null, - correlation_id: input.correlation_id ?? null, - created_at: now, - updated_at: null, - created_by: input.created_by ?? null, - }; - this.datasets.set(record.dataset_id, record); - - // 0pt.dataset_registered -- same "no mutation without an evented trail" - // doctrine every other P0 registry service in this repo follows - // (SourceService.createSource, SourceRatePolicyService.createSourceRatePolicy). await this.aud0.ingestEvent({ - event_id: `evt_0pt_dataset_registered_${record.dataset_id}`, + event_id: `evt_0pt_dataset_registered_${dataset.dataset_id}`, event_type: "0pt.dataset_registered", - tenant_id: record.tenant_id, - project_id: record.project_id, - actor_id: record.created_by ?? "system", + tenant_id: dataset.tenant_id, + project_id: dataset.project_id, + actor_id: dataset.created_by ?? "system", app_id: "0pt", capability_id: "0pt.register_dataset", resource_type: "dataset", - resource_id: record.dataset_id, + resource_id: dataset.dataset_id, risk_level: "L2", - payload: { name: record.name, dataset_type: record.dataset_type }, - created_at: now, - correlation_id: record.correlation_id ?? undefined, + payload: { name: dataset.name, dataset_type: dataset.dataset_type }, + created_at: dataset.created_at, + correlation_id: dataset.correlation_id, + idempotency_key: `dataset-register:${dataset.tenant_id}:${dataset.dataset_id}`, }); - return this.toDataset(record); + return dataset; } async getDataset(datasetId: string): Promise { - const record = this.datasets.get(datasetId); - if (!record) throw new Error(`Dataset ${datasetId} not found`); - return this.toDataset(record); + return this.repo.get(datasetId); } async listDatasets(filter?: { @@ -119,27 +75,6 @@ export class DatasetService { dataset_type?: DatasetType; status?: DatasetStatus; }): Promise { - let results = Array.from(this.datasets.values()); - if (filter?.tenant_id) results = results.filter((r) => r.tenant_id === filter.tenant_id); - if (filter?.project_id) results = results.filter((r) => r.project_id === filter.project_id); - if (filter?.dataset_type) results = results.filter((r) => r.dataset_type === filter.dataset_type); - if (filter?.status) results = results.filter((r) => r.status === filter.status); - return results.map((r) => this.toDataset(r)); - } - - private toDataset(r: DatasetRecord): Dataset { - return { - dataset_id: r.dataset_id, - tenant_id: r.tenant_id, - project_id: r.project_id, - name: r.name, - dataset_type: r.dataset_type as DatasetType, - status: r.status as DatasetStatus, - description: r.description ?? undefined, - correlation_id: r.correlation_id ?? undefined, - created_at: r.created_at, - updated_at: r.updated_at ?? undefined, - created_by: r.created_by ?? undefined, - }; + return this.repo.list(filter as DatasetListFilter | undefined); } }