refactor: persist DatasetService through repository boundary

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<string, DatasetRecord> = 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<Dataset> {
-    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<Dataset> {
-    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<Dataset[]> {
-    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);
   }
 }