diff --git a/app/api/0pt/lakehouse/mappings/[mappingId]/route.ts b/app/api/0pt/lakehouse/mappings/[mappingId]/route.ts new file mode 100644 index 0000000..5e76ca2 --- /dev/null +++ b/app/api/0pt/lakehouse/mappings/[mappingId]/route.ts @@ -0,0 +1,15 @@ +import { NextResponse } from "next/server"; +import { DatasetIcebergMappingService } from "@/src/services/dataset-iceberg-mapping-service"; + +const mappingService = new DatasetIcebergMappingService(); + +export async function GET(_request: Request, { params }: { params: Promise<{ mappingId: string }> }) { + try { + const { mappingId } = await params; + const mapping = await mappingService.getMapping(mappingId); + return NextResponse.json(mapping); + } catch (error: unknown) { + const message = error instanceof Error ? error.message : "Unknown error"; + return NextResponse.json({ error: message }, { status: 404 }); + } +} diff --git a/app/api/0pt/lakehouse/mappings/[mappingId]/writes/route.ts b/app/api/0pt/lakehouse/mappings/[mappingId]/writes/route.ts new file mode 100644 index 0000000..52fccdd --- /dev/null +++ b/app/api/0pt/lakehouse/mappings/[mappingId]/writes/route.ts @@ -0,0 +1,28 @@ +import { NextRequest, NextResponse } from "next/server"; +import { IcebergWriteService } from "@/src/services/iceberg-write-service"; + +const writeService = new IcebergWriteService(); + +// Registers a new Iceberg snapshot (a write) through ST0RE's table registry +// for an already-mapped dataset. See docs/ICEBERG_WRITE_BOUNDARY.md. +export async function POST(request: NextRequest, { params }: { params: Promise<{ mappingId: string }> }) { + try { + const { mappingId } = await params; + const body = await request.json(); + const result = await writeService.requestIcebergWrite({ + dataset_iceberg_mapping_id: mappingId, + actor_id: body.actor_id, + correlation_id: body.correlation_id, + snapshot_type: body.snapshot_type, + schema_version: body.schema_version, + manifest_uri: body.manifest_uri, + record_count: body.record_count, + size_bytes: body.size_bytes, + heavy: body.heavy, + }); + return NextResponse.json(result, { status: 202 }); + } catch (error: unknown) { + const message = error instanceof Error ? error.message : "Unknown error"; + return NextResponse.json({ error: message }, { status: 400 }); + } +} diff --git a/app/api/0pt/lakehouse/mappings/route.ts b/app/api/0pt/lakehouse/mappings/route.ts new file mode 100644 index 0000000..a843b54 --- /dev/null +++ b/app/api/0pt/lakehouse/mappings/route.ts @@ -0,0 +1,48 @@ +import { NextRequest, NextResponse } from "next/server"; +import { IcebergWriteService } from "@/src/services/iceberg-write-service"; +import { DatasetIcebergMappingService } from "@/src/services/dataset-iceberg-mapping-service"; + +// Shared instances so a mapping created via POST is visible to a later GET +// within the same server process, matching the in-memory-per-module +// convention already used by app/api/0pt/datasets and app/api/0pt/quality +// (Postgres persistence exists in sql/ but is not wired into route handlers +// in this pass -- same limitation those routes already have). +const mappingService = new DatasetIcebergMappingService(); +const writeService = new IcebergWriteService({ mappingService }); + +// POST registers a NEW Iceberg table in ST0RE's table registry (via +// IcebergWriteService.mapDatasetToIcebergTable) and records the +// dataset_iceberg_mapping row that points at it -- this is 0pt's write-path +// entrypoint, see docs/ICEBERG_WRITE_BOUNDARY.md. +export async function POST(request: NextRequest) { + try { + const body = await request.json(); + const result = await writeService.mapDatasetToIcebergTable({ + tenant_id: body.tenant_id, + project_id: body.project_id, + dataset_id: body.dataset_id, + dataset_iceberg_mapping_id: body.dataset_iceberg_mapping_id, + namespace: body.namespace, + table_name: body.table_name, + storage_uri: body.storage_uri, + catalog_ref: body.catalog_ref, + table_format: body.table_format, + actor_id: body.actor_id, + correlation_id: body.correlation_id, + }); + return NextResponse.json(result, { status: 201 }); + } catch (error: unknown) { + const message = error instanceof Error ? error.message : "Unknown error"; + return NextResponse.json({ error: message }, { status: 400 }); + } +} + +export async function GET(request: NextRequest) { + const { searchParams } = new URL(request.url); + const mappings = await mappingService.listMappings({ + tenant_id: searchParams.get("tenant_id") ?? undefined, + project_id: searchParams.get("project_id") ?? undefined, + dataset_id: searchParams.get("dataset_id") ?? undefined, + }); + return NextResponse.json(mappings); +} diff --git a/app/api/0pt/sources/rate-policies/[sourceRatePolicyId]/route.ts b/app/api/0pt/sources/rate-policies/[sourceRatePolicyId]/route.ts new file mode 100644 index 0000000..245038a --- /dev/null +++ b/app/api/0pt/sources/rate-policies/[sourceRatePolicyId]/route.ts @@ -0,0 +1,15 @@ +import { NextResponse } from "next/server"; +import { SourceRatePolicyService } from "@/src/services/source-rate-policy-service"; + +const service = new SourceRatePolicyService(); + +export async function GET(_request: Request, { params }: { params: Promise<{ sourceRatePolicyId: string }> }) { + try { + const { sourceRatePolicyId } = await params; + const policy = await service.getSourceRatePolicy(sourceRatePolicyId); + return NextResponse.json(policy); + } catch (error: unknown) { + const message = error instanceof Error ? error.message : "Unknown error"; + return NextResponse.json({ error: message }, { status: 404 }); + } +} diff --git a/app/api/0pt/sources/rate-policies/route.ts b/app/api/0pt/sources/rate-policies/route.ts new file mode 100644 index 0000000..cce357c --- /dev/null +++ b/app/api/0pt/sources/rate-policies/route.ts @@ -0,0 +1,25 @@ +import { NextRequest, NextResponse } from "next/server"; +import { SourceRatePolicyService } from "@/src/services/source-rate-policy-service"; + +const service = new SourceRatePolicyService(); + +export async function POST(request: NextRequest) { + try { + const body = await request.json(); + const policy = await service.createSourceRatePolicy(body); + return NextResponse.json(policy, { status: 201 }); + } catch (error: unknown) { + const message = error instanceof Error ? error.message : "Unknown error"; + return NextResponse.json({ error: message }, { status: 400 }); + } +} + +export async function GET(request: NextRequest) { + const { searchParams } = new URL(request.url); + const policies = await service.listSourceRatePolicies({ + tenant_id: searchParams.get("tenant_id") ?? undefined, + project_id: searchParams.get("project_id") ?? undefined, + source_id: searchParams.get("source_id") ?? undefined, + }); + return NextResponse.json(policies); +} diff --git a/docs/11-lakehouse-write-plan.md b/docs/11-lakehouse-write-plan.md index 4fe598f..3713db7 100644 --- a/docs/11-lakehouse-write-plan.md +++ b/docs/11-lakehouse-write-plan.md @@ -5,6 +5,49 @@ implementation session — it names the schemas/events precisely so that session guess, but it does not build the full pipeline today. Some items are already true today (quality rules) and this doc just confirms 0pt intends to keep them that way. +## 2026-07-12 update: what the 0pt write-boundary implementation pass closed + +This pass (the "future implementation session" this doc was written for) turned most items below +into real, tested code. Full detail in `docs/ICEBERG_WRITE_BOUNDARY.md`; short version: + +- Item 2 (`0pt.dataset_iceberg_mapping`): CLOSED as real code -- `src/types/dataset-iceberg-mapping.ts`, + `src/services/dataset-iceberg-mapping-service.ts`, `packages/schemas/0pt.dataset_iceberg_mapping.schema.json`, + `sql/004_iceberg_write_boundary.sql` (`opt_dataset_iceberg_mappings`), tested in + `tests/p0/dataset-iceberg-mapping.test.ts`. Field names differ slightly from the "suggested shape" + above (`dataset_iceberg_mapping_id` as PK, `iceberg_table_id` instead of `iceberg_table_ref`, + `mapping_status` instead of bare `entity_type`/`market`) -- reconciled against ST0RE's REAL + shipped `IcebergTable.tableId` shape (sax3l/st0re, not the "future" registry this doc assumed) and + against this session's own task spec; see the boundary doc for the full reconciliation note. +- Item 3 (first test dataset write): CLOSED against ST0RE's now-real (but not live-verified from + this environment) `/v1/st0re/lakehouse/*` API -- `src/clients/st0re-lakehouse-client.ts`, + `src/services/iceberg-write-service.ts`, proven in `tests/p0/iceberg-write-service.test.ts` + against that client's mock implementation (same mock/HTTP convention as `cl0ud-client.ts` etc). + NOT verified against a running ST0RE deployment in this pass -- see the boundary doc's + "Verification status" section and the `it.todo` in that test file. +- Item 4 (quality rules stay in 0pt): CONFIRMED, untouched -- `src/services/quality-check-service.ts` + and `src/services/quality-report-service.ts` were not modified by this pass. +- Item 5 (`0pt.source_rate_policy`): CLOSED as real code -- `src/types/source-rate-policy.ts`, + `src/services/source-rate-policy-service.ts`, `packages/schemas/0pt.source_rate_policy.schema.json`, + `sql/004_iceberg_write_boundary.sql` (`opt_source_rate_policies`), tested in + `tests/p0/source-rate-policy.test.ts`. Kept `requests_per_second`-style fields (as `domain`/ + `robots_ok`/`access`) as OPTIONAL alongside the task-spec's primary `rate_limit_per_minute`/ + `crawl_budget`/`compliance_notes` fields -- both specs for this same contract are satisfied on one + row, not two competing shapes. +- Item 6 (four AUD0 events): CLOSED for all four names -- `0pt.dataset_iceberg_table_mapped` and + `0pt.iceberg_write_requested`/`0pt.iceberg_snapshot_created` emit from + `src/services/iceberg-write-service.ts` and `dataset-iceberg-mapping-service.ts`; + `0pt.source_rate_policy_created` emits from `src/services/source-rate-policy-service.ts`. All via + the existing `Aud0Client` (mock/no-op by default, same as every other AUD0 emission in this repo -- + AUD0 itself still has no live ingestion endpoint). +- Item 7 (N0D heavy-write requests): CLOSED as real code -- `IcebergWriteService.requestIcebergWrite` + requests an N0D job (`0pt.iceberg_write_job`) only when `heavy: true`, never executes in-process; + tested in `tests/p0/iceberg-write-service.test.ts`. +- Item 8 (no lakehouse state outside ST0RE): CLOSED as a design property, not just a statement -- + `IcebergWriteService` and `DatasetIcebergMappingService` hold zero table/snapshot fields; the only + thing 0pt persists is the mapping's `iceberg_table_id` reference. See the boundary doc. +- Item 1 (Iceberg canonical, Delta rejected): CONFIRMED, unchanged from `docs/10` -- + `IcebergWriteService.mapDatasetToIcebergTable` defaults `table_format` to `"iceberg"`. + ## To-do list 1. **Use Apache Iceberg as the canonical write format** for datasets 0pt produces, per diff --git a/docs/ICEBERG_WRITE_BOUNDARY.md b/docs/ICEBERG_WRITE_BOUNDARY.md new file mode 100644 index 0000000..f58c45b --- /dev/null +++ b/docs/ICEBERG_WRITE_BOUNDARY.md @@ -0,0 +1,125 @@ +# Iceberg write-path ownership boundary (0pt) + +Read `docs/10-lakehouse-decision.md` and `docs/11-lakehouse-write-plan.md` first -- this doc is +the implementation record for the plan those wrote. It states the ownership boundary plainly and +says exactly what in this pass is real code vs. contract-only, matching the honesty convention +already used across this SIAX effort (e.g. sax3l/act0's own client docs, sax3l/st0re's +PRODUCTION_READINESS.md). + +## The boundary, plainly + +- **ST0RE owns table/catalog/artifact metadata for the lakehouse.** The Iceberg catalog, table + registry, namespace registry, and snapshot registry live in ST0RE (sax3l/st0re, + `src/api/routes/lakehouse.ts`, `/v1/st0re/lakehouse/*`). 0pt does not run a catalog, does not + keep a second copy of table/snapshot rows, and does not talk to MinIO or an Iceberg REST catalog + directly. +- **0pt owns dataset production, quality, data contracts, and writes.** 0pt runs sourcing -> + extraction -> structuring -> enrichment -> QA (`docs/06-data-engine.md`) as it always has, and is + the canonical writer of the Iceberg tables those datasets produce -- but it writes THROUGH + ST0RE's table registry API, never around it. +- **0pt owns source-rate/crawl-compliance policy.** This is the genuine ownership repair this pass + makes concrete: `0pt.source_rate_policy` (per-source rate limits, crawl budgets, compliance + notes) is 0pt's own contract. It does not exist in, and must not be reintroduced into, N0D's + `ComputePolicy.sourceRatePolicy` (deprecated there per the ADR-0030 audit), CL0UD, or any other + product's policy defaults. +- **N0D executes.** 0pt may request an N0D job for a heavy write or compaction + (`IcebergWriteService.requestIcebergWrite({ heavy: true })`), but N0D never owns the schema, the + catalog state, or the decision of what to write -- and 0pt never persists the result of that job + anywhere except by registering the resulting snapshot through ST0RE. +- **0pt never persists lakehouse table/catalog/snapshot state outside ST0RE.** Not in Postgres + (`opt_*` tables), not in memory beyond a single dataset -> table-id pointer + (`opt_dataset_iceberg_mappings.iceberg_table_id`, a REFERENCE, never a copy of ST0RE's row). + Standing up a parallel catalog in 0pt would repeat the exact mistake this whole cross-repo effort + is repairing (N0D's own bootstrap catalog, built because nothing else had one yet -- see + sax3l/st0re `docs/LAKEHOUSE_OWNERSHIP_PLAN.md`). + +## What this pass built + +| Concern | File(s) | Status | +|---|---|---| +| `0pt.dataset_iceberg_mapping.schema.v1` type | `src/types/dataset-iceberg-mapping.ts` | Real | +| `0pt.source_rate_policy.schema.v1` type | `src/types/source-rate-policy.ts` | Real | +| Dataset<->Iceberg-table mapping service | `src/services/dataset-iceberg-mapping-service.ts` | Real, tested | +| Source rate policy service | `src/services/source-rate-policy-service.ts` | Real, tested | +| ST0RE lakehouse HTTP/mock client | `src/clients/st0re-lakehouse-client.ts` | Real code, contract matches ST0RE's shipped API; **not live-verified** (see below) | +| Write orchestration (map table, request write) | `src/services/iceberg-write-service.ts` | Real, tested against the mock client | +| JSON Schemas | `packages/schemas/0pt.dataset_iceberg_mapping.schema.json`, `packages/schemas/0pt.source_rate_policy.schema.json` | Real | +| Postgres migration | `sql/004_iceberg_write_boundary.sql` | Real (not applied to a live DB by this pass -- same as `sql/002`/`sql/003`, which this repo's own P0 docs already note are schema-only until `DATABASE_URL` wiring lands) | +| API routes | `app/api/0pt/lakehouse/mappings/route.ts` (+`[mappingId]`, `+writes`), `app/api/0pt/sources/rate-policies/route.ts` (+`[sourceRatePolicyId]`) | Real, follow the existing `app/api/0pt/*` convention | +| Tests | `tests/p0/dataset-iceberg-mapping.test.ts`, `tests/p0/source-rate-policy.test.ts`, `tests/p0/iceberg-write-service.test.ts` | Real, all passing (`pnpm test`); one explicit `it.todo` for the live-ST0RE case | + +## Verification status (real vs. contract-only -- read this before trusting any claim above) + +- **Real and tested**: every TypeScript file above typechecks and its behavior is exercised by a + passing vitest test in `tests/p0/`. The full suite (`pnpm test`) passed at 154 tests + 1 todo + after this pass, up from a 142-test baseline before it, with 0 failures. +- **Real code, not live-verified**: `src/clients/st0re-lakehouse-client.ts`'s HTTP path is written + against ST0RE's actual, committed, shipped route contract (sax3l/st0re commit `9ea0abb`, + `src/api/routes/lakehouse.ts` + `src/types.ts` -- read directly from that repo's source during + this pass, not guessed). ST0RE's own `STATUS.md` documents it as deployed via Coolify on port + 8090. This pass did not have a confirmed, reachable `ST0RE_URL` to call, so the HTTP path has + **not been executed against a running ST0RE instance**. Every test in this pass exercises the + client's deterministic in-memory mock instead -- the exact same convention `cl0ud-client.ts`, + `n0d-client.ts`, `aud0-client.ts`, and `st0re-client.ts` already use in this repo for the same + reason (no live CL0UD/N0D/AUD0/ST0RE endpoint is reachable from this repo's test environment + either). This is not a downgrade from the rest of P0 -- it is the same maturity level as + everything else in `src/services/quality-check-service.ts`. +- **Contract-only / explicit TODO**: a live-ST0RE integration test is named but not built -- + `tests/p0/iceberg-write-service.test.ts`'s `it.todo(...)` line. A future session with a reachable + `ST0RE_URL` should fill it in by constructing `IcebergWriteService({ st0reBaseUrl: + process.env.ST0RE_URL })` and asserting the same flow against the real HTTP API. +- **Not built at all in this pass**: actual Iceberg data-file writes (Parquet bytes, manifest + generation). `IcebergWriteService.requestIcebergWrite` registers a snapshot POINTER through + ST0RE (`manifestUri`, `recordCount`, `sizeBytes` supplied by the caller) -- it does not produce + those bytes itself. This mirrors ST0RE's own documented stance on its snapshot route ("ST0RE + doesn't itself write the Iceberg metadata.json... this registry tracks the pointer/version + history for whatever the caller's own catalog write produced" -- `src/api/routes/lakehouse.ts` in + sax3l/st0re). Producing the actual bytes is a distinct, larger piece of work for a future session + once a real Iceberg writer library and MinIO credentials are wired into 0pt's engine + (`services/api`). + +## A test collision this pass found and fixed + +`tests/p0/enterprise-data-os-expansion.test.ts` (untracked, pre-existing, from an earlier "PROMPT +2" pass) had one test asserting `packages/schemas/0pt.source_rate_policy.schema.json`'s `required` +array contained `requests_per_second`/`burst_limit` and its `priority` enum contained `critical` -- +the OLD draft shape for this same schema file. This pass's canonical fields +(`rate_limit_per_minute`/`crawl_budget`/`compliance_notes`, per this session's explicit task spec) +replaced that draft, which would have silently broken that one pre-existing test. Fixed by updating +that single test block (not the rest of the file) to assert the new canonical fields instead -- +`pnpm test` is 154 passed + 1 todo, 0 failures, after this pass (up from a 142-test baseline before +it). See that test file's inline comment for the same note. + +## A pre-existing artifact this pass did not touch + +`packages/schemas/0pt.iceberg_table.schema.json` already existed (untracked, un-pretty-printed) in +this repo's working tree before this pass started -- it appears to be an earlier, different attempt +at this same concern that mirrors ST0RE's full `IcebergTable` row shape (`storage_uri`, +`catalog_ref`, `current_snapshot_id`, etc.) rather than a lightweight reference. If it is ever +adopted, it must be documented as a non-authoritative READ CACHE only -- 0pt must never treat it as +the source of truth or write to it independently of ST0RE. This pass's actual, owned contract is +`0pt.dataset_iceberg_mapping.schema.v1` above; it was left in place, unmodified, rather than +deleted or silently rewritten, since it wasn't part of this task's scope and this repo has other, +concurrent work in flight (`STATUS.md`, the wider uncommitted P0 data-platform build) that this +pass did not want to disturb outside its own files. + +## Event names (all four wired, docs/11 item 6) + +| Event | Emitted by | When | +|---|---|---| +| `0pt.dataset_iceberg_table_mapped` | `DatasetIcebergMappingService.createMapping` | A dataset -> Iceberg-table mapping is created | +| `0pt.iceberg_write_requested` | `IcebergWriteService.requestIcebergWrite` | Before 0pt calls ST0RE to register a snapshot | +| `0pt.iceberg_snapshot_created` | `IcebergWriteService.requestIcebergWrite` | After ST0RE confirms the new snapshot | +| `0pt.source_rate_policy_created` | `SourceRatePolicyService.createSourceRatePolicy` | A new source rate policy is created | + +All four go through the existing `Aud0Client` (`src/clients/aud0-client.ts`) -- no-op by default, +same as every other AUD0 emission already in this repo (AUD0 has no live ingestion endpoint yet). + +## Data quality stays in 0pt (confirmed, not moved) + +This pass did not modify `src/services/quality-check-service.ts`, `src/services/quality-report-service.ts`, +`src/types/quality-check.ts`, or `src/types/quality-report.ts`. 0pt's QA stage +(`docs/06-data-engine.md`) remains entirely 0pt-owned; nothing in this pass routes quality logic +through N0D, CL0UD, or ST0RE. N0D may still be asked to EXECUTE a quality-check job +(`quality-check-service.ts`'s existing `n0d.createJob` call, unrelated to this pass), but the rules +being executed are always 0pt's own. diff --git a/packages/schemas/0pt.dataset_iceberg_mapping.schema.json b/packages/schemas/0pt.dataset_iceberg_mapping.schema.json new file mode 100644 index 0000000..12a4a6c --- /dev/null +++ b/packages/schemas/0pt.dataset_iceberg_mapping.schema.json @@ -0,0 +1,30 @@ +{ + "$id": "0pt.dataset_iceberg_mapping.schema.v1", + "type": "object", + "required": [ + "dataset_iceberg_mapping_id", + "tenant_id", + "project_id", + "dataset_id", + "iceberg_table_id", + "mapping_status", + "created_at" + ], + "properties": { + "dataset_iceberg_mapping_id": { "type": "string" }, + "tenant_id": { "type": "string" }, + "project_id": { "type": "string" }, + "dataset_id": { "type": "string", "description": "References 0pt.dataset.schema.v1's dataset_id." }, + "iceberg_table_id": { + "type": "string", + "description": "Reference into ST0RE's table registry (IcebergTable.tableId, st0re_iceberg_tables). Never a copy of ST0RE's table row." + }, + "namespace": { "type": "string", "description": "ST0RE lakehouse namespace, carried for convenience/filtering only." }, + "table_name": { "type": "string" }, + "mapping_status": { "type": "string", "enum": ["pending", "active", "deprecated", "archived"] }, + "mapped_by": { "type": "string" }, + "correlation_id": { "type": "string" }, + "created_at": { "type": "string" }, + "updated_at": { "type": "string" } + } +} diff --git a/packages/schemas/0pt.source_rate_policy.schema.json b/packages/schemas/0pt.source_rate_policy.schema.json new file mode 100644 index 0000000..ca59df4 --- /dev/null +++ b/packages/schemas/0pt.source_rate_policy.schema.json @@ -0,0 +1,31 @@ +{ + "$id": "0pt.source_rate_policy.schema.v1", + "type": "object", + "required": [ + "source_rate_policy_id", + "tenant_id", + "project_id", + "source_id", + "rate_limit_per_minute", + "crawl_budget", + "status", + "created_at" + ], + "properties": { + "source_rate_policy_id": { "type": "string" }, + "tenant_id": { "type": "string" }, + "project_id": { "type": "string" }, + "source_id": { "type": "string", "description": "References 0pt.source.schema.v1's source_id." }, + "rate_limit_per_minute": { "type": "number" }, + "crawl_budget": { "type": "number", "description": "Max requests/pages/records this policy permits per crawl run or per day; unit documented in compliance_notes when non-obvious." }, + "compliance_notes": { "type": "string" }, + "domain": { "type": "string", "description": "Crawl-target domain, e.g. allabolag.se." }, + "robots_ok": { "type": "boolean" }, + "access": { "type": "string", "enum": ["api", "bulk", "authorized_web", "licensed"] }, + "status": { "type": "string", "enum": ["active", "paused", "deprecated"] }, + "correlation_id": { "type": "string" }, + "created_at": { "type": "string" }, + "updated_at": { "type": "string" }, + "created_by": { "type": "string" } + } +} diff --git a/sql/004_iceberg_write_boundary.sql b/sql/004_iceberg_write_boundary.sql new file mode 100644 index 0000000..3a27665 --- /dev/null +++ b/sql/004_iceberg_write_boundary.sql @@ -0,0 +1,50 @@ +-- 0pt Iceberg write-path ownership repair (docs/ICEBERG_WRITE_BOUNDARY.md) +-- Two tables: a REFERENCE from a 0pt dataset to its Iceberg table in ST0RE's +-- table registry, and 0pt's own source-rate/crawl-compliance policy (moved +-- out of N0D per ADR-0030's audit -- see docs/10-lakehouse-decision.md). +-- +-- opt_dataset_iceberg_mappings.iceberg_table_id is intentionally NOT a +-- foreign key -- it references ST0RE's st0re_iceberg_tables.table_id, a +-- different service's database. 0pt holds only the reference, never a copy +-- of ST0RE's table/catalog/snapshot state (see the module doc comment on +-- src/clients/st0re-lakehouse-client.ts). + +CREATE TABLE IF NOT EXISTS opt_dataset_iceberg_mappings ( + dataset_iceberg_mapping_id TEXT PRIMARY KEY, + tenant_id TEXT NOT NULL, + project_id TEXT NOT NULL, + dataset_id TEXT NOT NULL REFERENCES opt_datasets (dataset_id), + iceberg_table_id TEXT NOT NULL, + namespace TEXT, + table_name TEXT, + mapping_status TEXT NOT NULL DEFAULT 'active', + mapped_by TEXT, + correlation_id TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS opt_dataset_iceberg_mappings_dataset_idx ON opt_dataset_iceberg_mappings (dataset_id); +CREATE INDEX IF NOT EXISTS opt_dataset_iceberg_mappings_tenant_project_idx ON opt_dataset_iceberg_mappings (tenant_id, project_id); +CREATE INDEX IF NOT EXISTS opt_dataset_iceberg_mappings_table_idx ON opt_dataset_iceberg_mappings (iceberg_table_id); + +CREATE TABLE IF NOT EXISTS opt_source_rate_policies ( + source_rate_policy_id TEXT PRIMARY KEY, + tenant_id TEXT NOT NULL, + project_id TEXT NOT NULL, + source_id TEXT NOT NULL REFERENCES opt_sources (source_id), + rate_limit_per_minute NUMERIC NOT NULL, + crawl_budget NUMERIC NOT NULL, + compliance_notes TEXT, + domain TEXT, + robots_ok BOOLEAN, + access TEXT, + status TEXT NOT NULL DEFAULT 'active', + correlation_id TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ, + created_by TEXT +); + +CREATE INDEX IF NOT EXISTS opt_source_rate_policies_source_idx ON opt_source_rate_policies (source_id); +CREATE INDEX IF NOT EXISTS opt_source_rate_policies_tenant_project_idx ON opt_source_rate_policies (tenant_id, project_id); diff --git a/src/clients/st0re-lakehouse-client.ts b/src/clients/st0re-lakehouse-client.ts new file mode 100644 index 0000000..ea04244 --- /dev/null +++ b/src/clients/st0re-lakehouse-client.ts @@ -0,0 +1,185 @@ +// ST0RE lakehouse client -- 0pt's ONLY path for writing Iceberg table/ +// snapshot state. 0pt must never persist lakehouse table/catalog/snapshot +// state anywhere else (see docs/ICEBERG_WRITE_BOUNDARY.md). +// +// Distinct from `./st0re-client.ts` (the generic Artifact registry client, +// snake_case request shape) because it targets a different, newer route +// family on the SAME ST0RE service: `/v1/st0re/lakehouse/*` +// (sax3l/st0re `src/api/routes/lakehouse.ts`, commit 9ea0abb). That route +// family's Fastify handlers parse camelCase JSON bodies matching ST0RE's own +// `IcebergTableInput`/`TableSnapshotInput`/`LakehouseNamespaceInput` types +// (sax3l/st0re `src/types.ts`) -- this client's request/response shapes are +// camelCase deliberately, to match ST0RE's real contract byte-for-byte +// rather than 0pt's own snake_case domain convention. The service layer +// (`src/services/dataset-iceberg-mapping-service.ts`, +// `src/services/iceberg-write-service.ts`) is what translates between 0pt's +// snake_case types and this client's camelCase wire shape. +// +// HONESTY NOTE (not live-verified): ST0RE's lakehouse routes are real, +// committed code in sax3l/st0re (verified by reading that repo's source in +// this same pass) and are documented as deployed via Coolify at port 8090 +// (sax3l/st0re STATUS.md), but this pass did not confirm the deployment is +// currently reachable from this environment (no ST0RE_URL was available to +// probe). Exactly like `cl0ud-client.ts`/`aud0-client.ts`/`n0d-client.ts`/ +// `st0re-client.ts` already do in this repo, this client falls back to a +// deterministic in-memory mock when no base URL is configured, so callers +// (and this pass's tests) exercise the real request/response shape and +// control flow without requiring a live ST0RE instance. Set ST0RE_URL (or +// pass a baseUrl) to point this at a real deployment; the HTTP path is +// written against the documented contract above but has not been executed +// against a running ST0RE in this pass. + +export type IcebergTableFormat = "iceberg" | "delta" | "hudi"; +export type LakehouseObjectStatus = "active" | "deprecated" | "archived"; +export type TableSnapshotType = "append" | "overwrite" | "delete" | "replace" | "schema_change" | "compaction"; + +export interface St0reLakehouseNamespaceInput { + tenantId: string; + projectId: string; + name: string; + ownerProduct: string; +} + +export interface St0reLakehouseNamespace extends St0reLakehouseNamespaceInput { + namespaceId: string; + createdAt: number; +} + +export interface St0reIcebergTableInput { + tenantId: string; + projectId: string; + namespace: string; + tableName: string; + tableFormat: IcebergTableFormat; + storageUri: string; + /** Opaque reference into the external Iceberg catalog -- ST0RE does not resolve it. */ + catalogRef: string; + datasetId?: string; + ownerProduct: string; + schemaId?: string; + partitionSpec?: readonly string[]; + retentionPolicy?: string; + dataResidency?: string; + piiLevel?: "none" | "low" | "medium" | "high"; + createdBy: string; +} + +export interface St0reIcebergTable extends St0reIcebergTableInput { + tableId: string; + currentSnapshotId?: string; + status: LakehouseObjectStatus; + createdAt: number; + artifactId?: string; +} + +export interface St0reTableSnapshotInput { + snapshotType: TableSnapshotType; + schemaVersion: string; + /** URI of the Iceberg manifest list/file for this snapshot -- 0pt/N0D writes the actual bytes; ST0RE only tracks the pointer. */ + manifestUri: string; + createdByJobId?: string; + createdByCapability?: string; + recordCount?: number; + sizeBytes?: number; + aud0EventId?: string; +} + +export interface St0reTableSnapshot extends St0reTableSnapshotInput { + snapshotId: string; + tableId: string; + tenantId: string; + projectId: string; + createdAt: number; +} + +export interface St0reLakehouseClient { + registerNamespace(input: St0reLakehouseNamespaceInput): Promise; + registerTable(input: St0reIcebergTableInput): Promise; + getTable(tableId: string): Promise; + createSnapshot(tableId: string, input: St0reTableSnapshotInput): Promise; +} + +export function createSt0reLakehouseClient(baseUrl?: string): St0reLakehouseClient { + if (!baseUrl) return new MockSt0reLakehouseClient(); + return new HttpSt0reLakehouseClient(baseUrl); +} + +class MockSt0reLakehouseClient implements St0reLakehouseClient { + private tables: Map = new Map(); + + async registerNamespace(input: St0reLakehouseNamespaceInput): Promise { + return { ...input, namespaceId: `mock_ns_${Date.now()}`, createdAt: Date.now() }; + } + + async registerTable(input: St0reIcebergTableInput): Promise { + const table: St0reIcebergTable = { + ...input, + tableId: `mock_table_${Date.now()}`, + status: "active", + piiLevel: input.piiLevel ?? "none", + createdAt: Date.now(), + }; + this.tables.set(table.tableId, table); + return table; + } + + async getTable(tableId: string): Promise { + const table = this.tables.get(tableId); + if (!table) throw new Error(`ST0RE mock: table ${tableId} not found`); + return table; + } + + async createSnapshot(tableId: string, input: St0reTableSnapshotInput): Promise { + const table = this.tables.get(tableId); + const snapshotId = `mock_snapshot_${Date.now()}`; + if (table) table.currentSnapshotId = snapshotId; + return { + ...input, + snapshotId, + tableId, + tenantId: table?.tenantId ?? "unknown", + projectId: table?.projectId ?? "unknown", + createdAt: Date.now(), + }; + } +} + +class HttpSt0reLakehouseClient implements St0reLakehouseClient { + constructor(private baseUrl: string) {} + + async registerNamespace(input: St0reLakehouseNamespaceInput): Promise { + const res = await fetch(`${this.baseUrl}/v1/st0re/lakehouse/namespaces`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(input), + }); + if (!res.ok) throw new Error(`ST0RE lakehouse namespace error: ${res.status}`); + return res.json(); + } + + async registerTable(input: St0reIcebergTableInput): Promise { + const res = await fetch(`${this.baseUrl}/v1/st0re/lakehouse/tables`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(input), + }); + if (!res.ok) throw new Error(`ST0RE lakehouse table error: ${res.status}`); + return res.json(); + } + + async getTable(tableId: string): Promise { + const res = await fetch(`${this.baseUrl}/v1/st0re/lakehouse/tables/${tableId}`); + if (!res.ok) throw new Error(`ST0RE lakehouse get-table error: ${res.status}`); + return res.json(); + } + + async createSnapshot(tableId: string, input: St0reTableSnapshotInput): Promise { + const res = await fetch(`${this.baseUrl}/v1/st0re/lakehouse/tables/${tableId}/snapshots`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(input), + }); + if (!res.ok) throw new Error(`ST0RE lakehouse snapshot error: ${res.status}`); + return res.json(); + } +} diff --git a/src/services/dataset-iceberg-mapping-service.ts b/src/services/dataset-iceberg-mapping-service.ts new file mode 100644 index 0000000..acd65f8 --- /dev/null +++ b/src/services/dataset-iceberg-mapping-service.ts @@ -0,0 +1,164 @@ +import type { + DatasetIcebergMapping, + DatasetIcebergMappingStatus, + CreateDatasetIcebergMappingInput, +} from "../types/dataset-iceberg-mapping"; +import { createAud0Client, type Aud0Client } from "../clients/aud0-client"; +import { DatasetService } from "./dataset-service"; + +interface DatasetIcebergMappingRecord { + dataset_iceberg_mapping_id: string; + tenant_id: string; + project_id: string; + dataset_id: string; + iceberg_table_id: string; + namespace: string | null; + table_name: string | null; + mapping_status: string; + mapped_by: string | null; + correlation_id: string | null; + created_at: string; + updated_at: string | null; +} + +/** + * Owns 0pt.dataset_iceberg_mapping.schema.v1 -- the pointer from a 0pt + * dataset to the Iceberg table it writes to. `iceberg_table_id` is always a + * REFERENCE into ST0RE's table registry (never a copy of ST0RE's table row); + * this service holds no table/catalog/snapshot state itself. See + * docs/ICEBERG_WRITE_BOUNDARY.md. + * + * This service does not call ST0RE directly -- it records a mapping to an + * `iceberg_table_id` that already exists in ST0RE's registry. + * `IcebergWriteService.mapDatasetToIcebergTable` (src/services/ + * iceberg-write-service.ts) is the higher-level flow that registers a NEW + * table in ST0RE first and then calls this service to record the mapping -- + * kept as two composable steps so a mapping can also be created against an + * already-registered table (e.g. one N0D or another 0pt job already set up). + */ +export class DatasetIcebergMappingService { + private mappings: Map = new Map(); + private aud0: Aud0Client; + private datasetService: DatasetService; + + constructor(opts?: { aud0BaseUrl?: string; datasetService?: DatasetService }) { + this.aud0 = createAud0Client(opts?.aud0BaseUrl); + this.datasetService = opts?.datasetService ?? new DatasetService(opts?.aud0BaseUrl); + } + + async createMapping(input: CreateDatasetIcebergMappingInput): Promise { + if (this.mappings.has(input.dataset_iceberg_mapping_id)) { + throw new Error(`Dataset Iceberg mapping ${input.dataset_iceberg_mapping_id} already exists`); + } + if (!input.iceberg_table_id) { + throw new Error("iceberg_table_id is required (reference into ST0RE's table registry)"); + } + // Verify the dataset exists -- a mapping with no dataset is meaningless. + try { + await this.datasetService.getDataset(input.dataset_id); + } catch { + throw new Error(`Dataset ${input.dataset_id} not found`); + } + + const now = new Date().toISOString(); + const record: DatasetIcebergMappingRecord = { + dataset_iceberg_mapping_id: input.dataset_iceberg_mapping_id, + tenant_id: input.tenant_id, + project_id: input.project_id, + dataset_id: input.dataset_id, + iceberg_table_id: input.iceberg_table_id, + namespace: input.namespace ?? null, + table_name: input.table_name ?? null, + mapping_status: input.mapping_status ?? "active", + mapped_by: input.mapped_by ?? null, + correlation_id: input.correlation_id ?? null, + created_at: now, + updated_at: null, + }; + this.mappings.set(input.dataset_iceberg_mapping_id, record); + + // 0pt.dataset_iceberg_table_mapped -- docs/11-lakehouse-write-plan.md event list, + // docs/ICEBERG_WRITE_BOUNDARY.md. Fired when the mapping is created/updated. + await this.aud0.ingestEvent({ + event_id: `evt_0pt_dataset_iceberg_table_mapped_${input.dataset_iceberg_mapping_id}`, + event_type: "0pt.dataset_iceberg_table_mapped", + tenant_id: input.tenant_id, + project_id: input.project_id, + actor_id: input.mapped_by ?? "system", + app_id: "0pt", + capability_id: "0pt.map_dataset_iceberg_table", + resource_type: "dataset_iceberg_mapping", + resource_id: input.dataset_iceberg_mapping_id, + risk_level: "L2", + payload: { + dataset_id: input.dataset_id, + iceberg_table_id: input.iceberg_table_id, + namespace: input.namespace, + table_name: input.table_name, + }, + created_at: now, + correlation_id: input.correlation_id ?? "", + }); + + // Lineage: dataset -> iceberg table. + await this.aud0.ingestLineageEvent({ + lineage_event_id: `lineage_0pt_dataset_iceberg_${input.dataset_iceberg_mapping_id}`, + tenant_id: input.tenant_id, + project_id: input.project_id, + app_id: "0pt", + capability_id: "0pt.map_dataset_iceberg_table", + source_type: "dataset", + source_id: input.dataset_id, + target_type: "iceberg_table", + target_id: input.iceberg_table_id, + lineage_type: "stored_as", + metadata: { dataset_iceberg_mapping_id: input.dataset_iceberg_mapping_id }, + created_at: now, + correlation_id: input.correlation_id ?? "", + }); + + return this.toMapping(record); + } + + async getMapping(mappingId: string): Promise { + const record = this.mappings.get(mappingId); + if (!record) throw new Error(`Dataset Iceberg mapping ${mappingId} not found`); + return this.toMapping(record); + } + + async getMappingForDataset(datasetId: string): Promise { + const record = Array.from(this.mappings.values()).find((r) => r.dataset_id === datasetId && r.mapping_status === "active"); + return record ? this.toMapping(record) : undefined; + } + + async listMappings(filter?: { + tenant_id?: string; + project_id?: string; + dataset_id?: string; + mapping_status?: DatasetIcebergMappingStatus; + }): Promise { + let results = Array.from(this.mappings.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_id) results = results.filter((r) => r.dataset_id === filter.dataset_id); + if (filter?.mapping_status) results = results.filter((r) => r.mapping_status === filter.mapping_status); + return results.map((r) => this.toMapping(r)); + } + + private toMapping(r: DatasetIcebergMappingRecord): DatasetIcebergMapping { + return { + dataset_iceberg_mapping_id: r.dataset_iceberg_mapping_id, + tenant_id: r.tenant_id, + project_id: r.project_id, + dataset_id: r.dataset_id, + iceberg_table_id: r.iceberg_table_id, + namespace: r.namespace ?? undefined, + table_name: r.table_name ?? undefined, + mapping_status: r.mapping_status as DatasetIcebergMappingStatus, + mapped_by: r.mapped_by ?? undefined, + correlation_id: r.correlation_id ?? undefined, + created_at: r.created_at, + updated_at: r.updated_at ?? undefined, + }; + } +} diff --git a/src/services/iceberg-write-service.ts b/src/services/iceberg-write-service.ts new file mode 100644 index 0000000..001582c --- /dev/null +++ b/src/services/iceberg-write-service.ts @@ -0,0 +1,284 @@ +import type { DatasetIcebergMapping } from "../types/dataset-iceberg-mapping"; +import { createCl0udClient, type Cl0udClient } from "../clients/cl0ud-client"; +import { createN0dClient, type N0dClient } from "../clients/n0d-client"; +import { createAud0Client, type Aud0Client } from "../clients/aud0-client"; +import { + createSt0reLakehouseClient, + type St0reLakehouseClient, + type IcebergTableFormat, + type TableSnapshotType, +} from "../clients/st0re-lakehouse-client"; +import { DatasetService } from "./dataset-service"; +import { DatasetIcebergMappingService } from "./dataset-iceberg-mapping-service"; + +/** + * IcebergWriteService -- 0pt's write-path entrypoint for Iceberg tables. + * + * This is the concrete implementation of docs/ICEBERG_WRITE_BOUNDARY.md / + * docs/11-lakehouse-write-plan.md items 1, 3, 6, 7, 8: 0pt is the canonical + * writer of Iceberg tables, and it writes THROUGH ST0RE's table registry + * (`St0reLakehouseClient`) rather than persisting any lakehouse table, + * catalog, or snapshot state itself. This service holds NO table/snapshot + * state of its own -- only `DatasetIcebergMappingService`'s lightweight + * dataset-to-table-id pointer, which is a reference, not a copy. + * + * Two entry points: + * - `mapDatasetToIcebergTable` -- registers a NEW Iceberg table in ST0RE for + * a dataset that does not have one yet, then records the mapping. This is + * the "first test dataset written to Iceberg through ST0RE's table + * registry" proof path (see tests/p0/iceberg-write-service.test.ts). + * - `requestIcebergWrite` -- registers a new snapshot (an actual write) for + * an already-mapped dataset, optionally routed through an N0D job when the + * write is heavy (item 7: 0pt REQUESTS N0D jobs for heavy writes/ + * compactions, it does not execute them itself, and it never owns the + * result of that job beyond the ST0RE-registered snapshot ref). + * + * Byte-level writes (actually producing Parquet/Iceberg data files) are OUT + * OF SCOPE for this pass, same honest boundary ST0RE itself documents for + * its own snapshot routes ("ST0RE doesn't itself write the Iceberg + * metadata.json... this registry tracks the pointer/version history for + * whatever the caller's own catalog write produced" -- + * sax3l/st0re src/api/routes/lakehouse.ts). What this service proves is the + * OWNERSHIP path -- that 0pt requests the write and registers its result + * through ST0RE, never anywhere else. + */ +export interface MapDatasetToIcebergTableInput { + tenant_id: string; + project_id: string; + dataset_id: string; + dataset_iceberg_mapping_id: string; + namespace: string; + table_name: string; + storage_uri: string; + catalog_ref: string; + table_format?: IcebergTableFormat; + actor_id: string; + correlation_id: string; +} + +export interface MapDatasetToIcebergTableResult { + mapping: DatasetIcebergMapping; + iceberg_table_id: string; + policy_decision_id: string; +} + +export interface RequestIcebergWriteInput { + dataset_iceberg_mapping_id: string; + actor_id: string; + correlation_id: string; + snapshot_type?: TableSnapshotType; + schema_version: string; + manifest_uri: string; + record_count?: number; + size_bytes?: number; + /** True for large backfills/table maintenance -- routes through an N0D job request rather than 0pt executing in-process (docs/11-lakehouse-write-plan.md item 7). */ + heavy?: boolean; +} + +export interface RequestIcebergWriteResult { + snapshot_id: string; + iceberg_table_id: string; + policy_decision_id: string; + n0d_job_id?: string; +} + +export class IcebergWriteService { + private cl0ud: Cl0udClient; + private n0d: N0dClient; + private st0reLakehouse: St0reLakehouseClient; + private aud0: Aud0Client; + private datasetService: DatasetService; + private mappingService: DatasetIcebergMappingService; + + constructor(opts?: { + cl0udBaseUrl?: string; + n0dBaseUrl?: string; + st0reBaseUrl?: string; + aud0BaseUrl?: string; + datasetService?: DatasetService; + mappingService?: DatasetIcebergMappingService; + }) { + this.cl0ud = createCl0udClient(opts?.cl0udBaseUrl); + this.n0d = createN0dClient(opts?.n0dBaseUrl); + this.st0reLakehouse = createSt0reLakehouseClient(opts?.st0reBaseUrl); + this.aud0 = createAud0Client(opts?.aud0BaseUrl); + this.datasetService = opts?.datasetService ?? new DatasetService(opts?.aud0BaseUrl); + this.mappingService = opts?.mappingService ?? new DatasetIcebergMappingService({ aud0BaseUrl: opts?.aud0BaseUrl, datasetService: this.datasetService }); + } + + async mapDatasetToIcebergTable(input: MapDatasetToIcebergTableInput): Promise { + const dataset = await this.datasetService.getDataset(input.dataset_id); // throws if missing + + // CL0UD policy boundary -- same L3 gate ST0RE's own table-registration + // route applies (sax3l/st0re src/api/routes/lakehouse.ts, + // 'st0re.lakehouse.write_table'), checked here too since 0pt is the + // caller initiating the write, not just relying on ST0RE to gate it. + const policy = await this.cl0ud.evaluatePolicy({ + tenant_id: input.tenant_id, + project_id: input.project_id, + actor_id: input.actor_id, + capability_id: "0pt.write_iceberg_table", + resource_type: "dataset", + resource_id: input.dataset_id, + risk_level: "L3", + correlation_id: input.correlation_id, + }); + if (policy.decision === "deny") throw new Error(`Policy denied: ${policy.reason}`); + if (policy.decision === "approval_required") throw new Error(`Policy requires approval: ${policy.reason}`); + + // Register the table THROUGH ST0RE's table registry -- 0pt does not + // create or track lakehouse table state on its own. + const table = await this.st0reLakehouse.registerTable({ + tenantId: input.tenant_id, + projectId: input.project_id, + namespace: input.namespace, + tableName: input.table_name, + tableFormat: input.table_format ?? "iceberg", // ADR-0030: Iceberg is canonical; Delta/Hudi accepted only for bridge/import scenarios, never chosen by default here. + storageUri: input.storage_uri, + catalogRef: input.catalog_ref, + datasetId: input.dataset_id, + ownerProduct: "0pt", + createdBy: input.actor_id, + }); + + const mapping = await this.mappingService.createMapping({ + dataset_iceberg_mapping_id: input.dataset_iceberg_mapping_id, + tenant_id: input.tenant_id, + project_id: input.project_id, + dataset_id: input.dataset_id, + iceberg_table_id: table.tableId, + namespace: input.namespace, + table_name: input.table_name, + mapping_status: "active", + mapped_by: input.actor_id, + correlation_id: input.correlation_id, + }); + + return { mapping, iceberg_table_id: table.tableId, policy_decision_id: policy.decision_id }; + } + + async requestIcebergWrite(input: RequestIcebergWriteInput): Promise { + const mapping = await this.mappingService.getMapping(input.dataset_iceberg_mapping_id); + if (mapping.mapping_status !== "active") { + throw new Error(`Dataset Iceberg mapping ${input.dataset_iceberg_mapping_id} is not active (status=${mapping.mapping_status})`); + } + + const policy = await this.cl0ud.evaluatePolicy({ + tenant_id: mapping.tenant_id, + project_id: mapping.project_id, + actor_id: input.actor_id, + capability_id: "0pt.write_iceberg_table", + resource_type: "dataset", + resource_id: mapping.dataset_id, + risk_level: "L3", + correlation_id: input.correlation_id, + }); + if (policy.decision === "deny") throw new Error(`Policy denied: ${policy.reason}`); + if (policy.decision === "approval_required") throw new Error(`Policy requires approval: ${policy.reason}`); + + // 0pt.iceberg_write_requested -- fired when 0pt requests the write, + // before the write lands (docs/11-lakehouse-write-plan.md event list). + await this.aud0.ingestEvent({ + event_id: `evt_0pt_iceberg_write_requested_${input.dataset_iceberg_mapping_id}_${Date.now()}`, + event_type: "0pt.iceberg_write_requested", + tenant_id: mapping.tenant_id, + project_id: mapping.project_id, + actor_id: input.actor_id, + app_id: "0pt", + capability_id: "0pt.write_iceberg_table", + resource_type: "iceberg_table", + resource_id: mapping.iceberg_table_id, + risk_level: "L3", + payload: { dataset_id: mapping.dataset_id, dataset_iceberg_mapping_id: mapping.dataset_iceberg_mapping_id, heavy: input.heavy ?? false }, + created_at: new Date().toISOString(), + correlation_id: input.correlation_id, + }); + + // Heavy writes/compactions are REQUESTED from N0D, never executed + // in-process by 0pt (docs/11-lakehouse-write-plan.md item 7). N0D + // executes; it does not own the schema or the decision of what to + // write -- the snapshot registration below still goes through ST0RE + // regardless of who performed the write. + let n0dJobId: string | undefined; + if (input.heavy) { + const job = await this.n0d.createJob({ + tenant_id: mapping.tenant_id, + project_id: mapping.project_id, + job_type: "0pt.iceberg_write_job", + capability_id: "0pt.write_iceberg_table", + requested_by_app_id: "0pt", + requested_by_actor_id: input.actor_id, + input_dataset_ids: [mapping.dataset_id], + output_artifact_type: "iceberg_table_snapshot", + requires_backbone: true, + requires_gpu: false, + required_node_roles: ["compute_general"], + resource_profile: "cpu_io_large", + risk_level: "L3", + correlation_id: input.correlation_id, + }); + n0dJobId = job.job_id; + } + + // Register the snapshot THROUGH ST0RE's table registry -- the one and + // only place 0pt persists that a write landed. + const snapshot = await this.st0reLakehouse.createSnapshot(mapping.iceberg_table_id, { + snapshotType: input.snapshot_type ?? "append", + schemaVersion: input.schema_version, + manifestUri: input.manifest_uri, + createdByJobId: n0dJobId, + createdByCapability: "0pt.write_iceberg_table", + recordCount: input.record_count, + sizeBytes: input.size_bytes, + }); + + // 0pt.iceberg_snapshot_created -- fired once the write lands and ST0RE + // has registered a new snapshot, carrying table ref + snapshot id + row + // count + dataset id for AUD0 to build source -> dataset -> snapshot + // lineage. + await this.aud0.ingestEvent({ + event_id: `evt_0pt_iceberg_snapshot_created_${snapshot.snapshotId}`, + event_type: "0pt.iceberg_snapshot_created", + tenant_id: mapping.tenant_id, + project_id: mapping.project_id, + actor_id: input.actor_id, + app_id: "0pt", + capability_id: "0pt.write_iceberg_table", + resource_type: "iceberg_snapshot", + resource_id: snapshot.snapshotId, + risk_level: "L2", + payload: { + dataset_id: mapping.dataset_id, + iceberg_table_id: mapping.iceberg_table_id, + snapshot_id: snapshot.snapshotId, + record_count: input.record_count, + n0d_job_id: n0dJobId, + }, + created_at: new Date().toISOString(), + correlation_id: input.correlation_id, + }); + + await this.aud0.ingestLineageEvent({ + lineage_event_id: `lineage_0pt_iceberg_snapshot_${snapshot.snapshotId}`, + tenant_id: mapping.tenant_id, + project_id: mapping.project_id, + app_id: "0pt", + capability_id: "0pt.write_iceberg_table", + source_type: "dataset", + source_id: mapping.dataset_id, + target_type: "iceberg_snapshot", + target_id: snapshot.snapshotId, + lineage_type: "produced", + metadata: { iceberg_table_id: mapping.iceberg_table_id, n0d_job_id: n0dJobId }, + created_at: new Date().toISOString(), + correlation_id: input.correlation_id, + }); + + return { + snapshot_id: snapshot.snapshotId, + iceberg_table_id: mapping.iceberg_table_id, + policy_decision_id: policy.decision_id, + n0d_job_id: n0dJobId, + }; + } +} diff --git a/src/services/source-rate-policy-service.ts b/src/services/source-rate-policy-service.ts new file mode 100644 index 0000000..088a1bd --- /dev/null +++ b/src/services/source-rate-policy-service.ts @@ -0,0 +1,141 @@ +import type { SourceRatePolicy, SourceRatePolicyStatus, CreateSourceRatePolicyInput } from "../types/source-rate-policy"; +import { createAud0Client, type Aud0Client } from "../clients/aud0-client"; +import { SourceService } from "./source-service"; + +interface SourceRatePolicyRecord { + source_rate_policy_id: string; + tenant_id: string; + project_id: string; + source_id: string; + rate_limit_per_minute: number; + crawl_budget: number; + compliance_notes: string | null; + domain: string | null; + robots_ok: boolean | null; + access: string | null; + status: string; + correlation_id: string | null; + created_at: string; + updated_at: string | null; + created_by: string | null; +} + +/** + * Owns 0pt.source_rate_policy.schema.v1 -- the genuine ownership repair: + * source/crawl-domain rate limiting is 0pt's own compliance policy, moved + * OUT of N0D's ComputePolicy.sourceRatePolicy (deprecated there per + * ADR-0030's audit). See docs/ICEBERG_WRITE_BOUNDARY.md. + */ +export class SourceRatePolicyService { + private policies: Map = new Map(); + private aud0: Aud0Client; + private sourceService: SourceService; + + constructor(opts?: { aud0BaseUrl?: string; sourceService?: SourceService }) { + this.aud0 = createAud0Client(opts?.aud0BaseUrl); + this.sourceService = opts?.sourceService ?? new SourceService(opts?.aud0BaseUrl); + } + + async createSourceRatePolicy(input: CreateSourceRatePolicyInput): Promise { + if (this.policies.has(input.source_rate_policy_id)) { + throw new Error(`Source rate policy ${input.source_rate_policy_id} already exists`); + } + if (input.rate_limit_per_minute <= 0) { + throw new Error("rate_limit_per_minute must be > 0"); + } + if (input.crawl_budget <= 0) { + throw new Error("crawl_budget must be > 0"); + } + // Verify the source exists -- a rate policy with no source is meaningless. + try { + await this.sourceService.getSource(input.source_id); + } catch { + throw new Error(`Source ${input.source_id} not found`); + } + + const now = new Date().toISOString(); + const record: SourceRatePolicyRecord = { + source_rate_policy_id: input.source_rate_policy_id, + tenant_id: input.tenant_id, + project_id: input.project_id, + source_id: input.source_id, + rate_limit_per_minute: input.rate_limit_per_minute, + crawl_budget: input.crawl_budget, + compliance_notes: input.compliance_notes ?? null, + domain: input.domain ?? null, + robots_ok: input.robots_ok ?? null, + access: input.access ?? null, + status: input.status ?? "active", + correlation_id: input.correlation_id ?? null, + created_at: now, + updated_at: null, + created_by: input.created_by ?? null, + }; + this.policies.set(input.source_rate_policy_id, record); + + // 0pt.source_rate_policy_created -- docs/11-lakehouse-write-plan.md event list, + // docs/ICEBERG_WRITE_BOUNDARY.md. Gives AUD0/CL0UD an evented trail of 0pt's + // crawl-compliance posture without polling 0pt's DB. + await this.aud0.ingestEvent({ + event_id: `evt_0pt_source_rate_policy_created_${input.source_rate_policy_id}`, + event_type: "0pt.source_rate_policy_created", + tenant_id: input.tenant_id, + project_id: input.project_id, + actor_id: input.created_by ?? "system", + app_id: "0pt", + capability_id: "0pt.create_source_rate_policy", + resource_type: "source_rate_policy", + resource_id: input.source_rate_policy_id, + risk_level: "L2", + payload: { + source_id: input.source_id, + rate_limit_per_minute: input.rate_limit_per_minute, + crawl_budget: input.crawl_budget, + }, + created_at: now, + correlation_id: input.correlation_id ?? "", + }); + + return this.toPolicy(record); + } + + async getSourceRatePolicy(sourceRatePolicyId: string): Promise { + const record = this.policies.get(sourceRatePolicyId); + if (!record) throw new Error(`Source rate policy ${sourceRatePolicyId} not found`); + return this.toPolicy(record); + } + + async listSourceRatePolicies(filter?: { + tenant_id?: string; + project_id?: string; + source_id?: string; + status?: SourceRatePolicyStatus; + }): Promise { + let results = Array.from(this.policies.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?.source_id) results = results.filter((r) => r.source_id === filter.source_id); + if (filter?.status) results = results.filter((r) => r.status === filter.status); + return results.map((r) => this.toPolicy(r)); + } + + private toPolicy(r: SourceRatePolicyRecord): SourceRatePolicy { + return { + source_rate_policy_id: r.source_rate_policy_id, + tenant_id: r.tenant_id, + project_id: r.project_id, + source_id: r.source_id, + rate_limit_per_minute: r.rate_limit_per_minute, + crawl_budget: r.crawl_budget, + compliance_notes: r.compliance_notes ?? undefined, + domain: r.domain ?? undefined, + robots_ok: r.robots_ok ?? undefined, + access: (r.access as SourceRatePolicy["access"]) ?? undefined, + status: r.status as SourceRatePolicyStatus, + correlation_id: r.correlation_id ?? undefined, + created_at: r.created_at, + updated_at: r.updated_at ?? undefined, + created_by: r.created_by ?? undefined, + }; + } +} diff --git a/src/types/dataset-iceberg-mapping.ts b/src/types/dataset-iceberg-mapping.ts new file mode 100644 index 0000000..3b202c2 --- /dev/null +++ b/src/types/dataset-iceberg-mapping.ts @@ -0,0 +1,41 @@ +// 0pt.dataset_iceberg_mapping.schema.v1 +// +// Maps a 0pt dataset to the Iceberg table that carries its canonical lakehouse +// write path. `iceberg_table_id` is a REFERENCE into ST0RE's table registry +// (IcebergTable.tableId, st0re_iceberg_tables) -- it is not a copy of ST0RE's +// table metadata. 0pt does not persist table/catalog/snapshot state of its +// own; this row exists only to answer "which Iceberg table does this dataset +// write to", per docs/10-lakehouse-decision.md and +// docs/ICEBERG_WRITE_BOUNDARY.md. + +export type DatasetIcebergMappingStatus = "pending" | "active" | "deprecated" | "archived"; + +export interface DatasetIcebergMapping { + dataset_iceberg_mapping_id: string; + tenant_id: string; + project_id: string; + dataset_id: string; + /** Reference into ST0RE's table registry (IcebergTable.tableId). Not resolved or duplicated locally. */ + iceberg_table_id: string; + /** ST0RE lakehouse namespace the table lives in, carried for convenience/filtering only -- ST0RE remains the source of truth. */ + namespace?: string; + table_name?: string; + mapping_status: DatasetIcebergMappingStatus; + mapped_by?: string; + correlation_id?: string; + created_at: string; + updated_at?: string; +} + +export interface CreateDatasetIcebergMappingInput { + dataset_iceberg_mapping_id: string; + tenant_id: string; + project_id: string; + dataset_id: string; + iceberg_table_id: string; + namespace?: string; + table_name?: string; + mapping_status?: DatasetIcebergMappingStatus; + mapped_by?: string; + correlation_id?: string; +} diff --git a/src/types/source-rate-policy.ts b/src/types/source-rate-policy.ts new file mode 100644 index 0000000..c63f4da --- /dev/null +++ b/src/types/source-rate-policy.ts @@ -0,0 +1,57 @@ +// 0pt.source_rate_policy.schema.v1 +// +// The genuine ownership repair this contract encodes: crawl-domain rate +// limiting and compliance posture belongs to 0pt, not N0D. N0D's +// `ComputePolicy.sourceRatePolicy` field (per-domain crawl rate limits) is +// deprecated/removed from N0D's seed data per ADR-0030's audit -- 0pt already +// owns the sourcing chain (docs/06-data-engine.md, `sources.yaml`) and is the +// only product that should hold this policy. +// +// Field names reconcile two prior specs written for this same contract: +// docs/11-lakehouse-write-plan.md item 5 (domain/requests_per_second/ +// robots_ok/access) and this pass's task spec (rate_limit_per_minute/ +// crawl_budget/compliance_notes). Both are kept: the task-spec fields are +// primary, the docs/11 fields are additional optional detail for the same +// row, not a competing shape. + +export type SourceRatePolicyStatus = "active" | "paused" | "deprecated"; + +export interface SourceRatePolicy { + source_rate_policy_id: string; + tenant_id: string; + project_id: string; + /** References source.source_id (src/types/source.ts). */ + source_id: string; + rate_limit_per_minute: number; + /** Max requests/pages/records this policy permits per crawl run or per day (unit documented in compliance_notes when non-obvious). */ + crawl_budget: number; + /** Free-text compliance rationale: robots.txt posture, licence terms, contact/ToS reference. */ + compliance_notes?: string; + /** Crawl-target domain, e.g. "allabolag.se" -- optional convenience field mirroring docs/06-data-engine.md's sources.yaml rate_limit entries. */ + domain?: string; + /** Whether robots.txt permits the access pattern this policy governs. */ + robots_ok?: boolean; + /** Mirrors dim_source.access (docs/03-data-model.md): api | bulk | authorized_web | licensed. */ + access?: "api" | "bulk" | "authorized_web" | "licensed"; + status: SourceRatePolicyStatus; + correlation_id?: string; + created_at: string; + updated_at?: string; + created_by?: string; +} + +export interface CreateSourceRatePolicyInput { + source_rate_policy_id: string; + tenant_id: string; + project_id: string; + source_id: string; + rate_limit_per_minute: number; + crawl_budget: number; + compliance_notes?: string; + domain?: string; + robots_ok?: boolean; + access?: "api" | "bulk" | "authorized_web" | "licensed"; + status?: SourceRatePolicyStatus; + correlation_id?: string; + created_by?: string; +} diff --git a/tests/p0/dataset-iceberg-mapping.test.ts b/tests/p0/dataset-iceberg-mapping.test.ts new file mode 100644 index 0000000..a5acdb1 --- /dev/null +++ b/tests/p0/dataset-iceberg-mapping.test.ts @@ -0,0 +1,87 @@ +import { describe, it, expect } from "vitest"; +import { DatasetService } from "@/src/services/dataset-service"; +import { DatasetIcebergMappingService } from "@/src/services/dataset-iceberg-mapping-service"; + +describe("DatasetIcebergMappingService", () => { + it("createMapping persists a reference into ST0RE's table registry", async () => { + const datasetService = new DatasetService(); + const service = new DatasetIcebergMappingService({ datasetService }); + await datasetService.registerDataset({ + dataset_id: "ds_iceberg_map_1", + tenant_id: "t1", + project_id: "p1", + name: "Iceberg Mapping Dataset", + dataset_type: "raw", + }); + + const mapping = await service.createMapping({ + dataset_iceberg_mapping_id: "dim_test_001", + tenant_id: "t1", + project_id: "p1", + dataset_id: "ds_iceberg_map_1", + iceberg_table_id: "st0re_table_abc123", + namespace: "0pt_default", + table_name: "dim_company_se", + }); + + expect(mapping.dataset_iceberg_mapping_id).toBe("dim_test_001"); + expect(mapping.iceberg_table_id).toBe("st0re_table_abc123"); + expect(mapping.mapping_status).toBe("active"); + expect(mapping.created_at).toBeDefined(); + }); + + it("createMapping rejects an unknown dataset_id", async () => { + const service = new DatasetIcebergMappingService(); + await expect( + service.createMapping({ + dataset_iceberg_mapping_id: "dim_test_bad", + tenant_id: "t1", + project_id: "p1", + dataset_id: "ds_does_not_exist", + iceberg_table_id: "st0re_table_xyz", + }) + ).rejects.toThrow("not found"); + }); + + it("createMapping rejects a missing iceberg_table_id", async () => { + const datasetService = new DatasetService(); + const service = new DatasetIcebergMappingService({ datasetService }); + await datasetService.registerDataset({ + dataset_id: "ds_iceberg_map_2", + tenant_id: "t1", + project_id: "p1", + name: "No Table Ref", + dataset_type: "raw", + }); + await expect( + service.createMapping({ + dataset_iceberg_mapping_id: "dim_test_no_ref", + tenant_id: "t1", + project_id: "p1", + dataset_id: "ds_iceberg_map_2", + iceberg_table_id: "", + }) + ).rejects.toThrow("iceberg_table_id is required"); + }); + + it("getMappingForDataset returns the active mapping for a dataset", async () => { + const datasetService = new DatasetService(); + const service = new DatasetIcebergMappingService({ datasetService }); + await datasetService.registerDataset({ + dataset_id: "ds_iceberg_map_3", + tenant_id: "t1", + project_id: "p1", + name: "Lookup Dataset", + dataset_type: "raw", + }); + await service.createMapping({ + dataset_iceberg_mapping_id: "dim_test_lookup", + tenant_id: "t1", + project_id: "p1", + dataset_id: "ds_iceberg_map_3", + iceberg_table_id: "st0re_table_lookup", + }); + const found = await service.getMappingForDataset("ds_iceberg_map_3"); + expect(found?.iceberg_table_id).toBe("st0re_table_lookup"); + }); +}); diff --git a/tests/p0/enterprise-data-os-expansion.test.ts b/tests/p0/enterprise-data-os-expansion.test.ts new file mode 100644 index 0000000..572d362 --- /dev/null +++ b/tests/p0/enterprise-data-os-expansion.test.ts @@ -0,0 +1,65 @@ +import {describe,it,expect} from "vitest"; + +describe("0pt Enterprise Data OS Expansion (PROMPT 2)",()=>{ + it("iceberg_table schema has required fields",()=>{ + const schema=require("../../packages/schemas/0pt.iceberg_table.schema.json"); + expect(schema.required).toContain("table_id"); + expect(schema.required).toContain("table_name"); + expect(schema.required).toContain("storage_uri"); + expect(schema.properties.table_format.enum).toContain("iceberg"); + }); + + it("signal schema has entity and strength",()=>{ + const schema=require("../../packages/schemas/0pt.signal.schema.json"); + expect(schema.required).toContain("signal_id"); + expect(schema.required).toContain("signal_type"); + expect(schema.required).toContain("signal_strength"); + expect(schema.required).toContain("entity_type"); + }); + + it("lineage_event schema has input/output assets",()=>{ + const schema=require("../../packages/schemas/0pt.lineage_event.schema.json"); + expect(schema.required).toContain("input_assets"); + expect(schema.required).toContain("output_assets"); + expect(schema.required).toContain("producer"); + }); + + // Updated 2026-07-12 (0pt Iceberg write-boundary pass, docs/ICEBERG_WRITE_BOUNDARY.md): + // 0pt.source_rate_policy.schema.v1's canonical fields are rate_limit_per_minute / + // crawl_budget / compliance_notes per this session's explicit task spec (the genuine + // sourceRatePolicy ownership repair, moving this out of N0D) -- superseding the + // requests_per_second/burst_limit/priority draft this test previously checked, which + // predated that spec. See src/types/source-rate-policy.ts and + // src/services/source-rate-policy-service.ts for the real, tested implementation. + it("source_rate_policy schema has rate limits",()=>{ + const schema=require("../../packages/schemas/0pt.source_rate_policy.schema.json"); + expect(schema.required).toContain("rate_limit_per_minute"); + expect(schema.required).toContain("crawl_budget"); + expect(schema.properties.access.enum).toContain("licensed"); + }); + + it("dataset contract includes Iceberg storage_format",()=>{ + const dataset={dataset_id:"ds_001",storage_format:"iceberg",quality_score:0.94,pii_level:"medium",status:"active"}; + expect(dataset.storage_format).toBe("iceberg"); + expect(dataset.quality_score).toBeGreaterThan(0.9); + }); + + it("feature_view includes offline/online store refs",()=>{ + const fv={feature_view_id:"fv_001",entity:"company",offline_store:"iceberg",online_store:"redis",features:[{name:"hiring_growth_30d",type:"float"}]}; + expect(fv.offline_store).toBe("iceberg"); + expect(fv.online_store).toBe("redis"); + expect(fv.features[0].name).toBe("hiring_growth_30d"); + }); + + it("N0D job request for feature materialization",()=>{ + const jobRequest={job_type:"0pt.feature_materialization_job",capability_id:"0pt.materialize_features",output_artifact_type:"feature_snapshot",requires_backbone:true,resource_profile:"cpu_memory_high",risk_level:"L3"}; + expect(jobRequest.job_type).toBe("0pt.feature_materialization_job"); + expect(jobRequest.risk_level).toBe("L3"); + }); + + it("0pt emits AUD0 lineage event",()=>{ + const lineageEvent={event_type:"0pt.lineage_emitted",app_id:"0pt",capability_id:"0pt.enrich_entities",input_assets:[{asset_type:"dataset",asset_id:"ds_raw"}],output_assets:[{asset_type:"iceberg_table",asset_id:"table_enriched"}],st0re_artifacts:["artifact_123"],quality_report_id:"qr_123"}; + expect(lineageEvent.event_type).toBe("0pt.lineage_emitted"); + expect(lineageEvent.output_assets[0].asset_type).toBe("iceberg_table"); + }); +}); diff --git a/tests/p0/iceberg-write-service.test.ts b/tests/p0/iceberg-write-service.test.ts new file mode 100644 index 0000000..e567ce7 --- /dev/null +++ b/tests/p0/iceberg-write-service.test.ts @@ -0,0 +1,177 @@ +import { describe, it, expect } from "vitest"; +import { DatasetService } from "@/src/services/dataset-service"; +import { DatasetIcebergMappingService } from "@/src/services/dataset-iceberg-mapping-service"; +import { IcebergWriteService } from "@/src/services/iceberg-write-service"; + +/** + * "Write first test dataset to Iceberg through ST0RE's table registry" proof + * (task item 2). This exercises the REAL request/response shapes and control + * flow of IcebergWriteService -> St0reLakehouseClient against that client's + * deterministic MOCK implementation, the same convention this repo already + * uses for every other cross-product boundary (see + * tests/p0/cross-product-boundary.test.ts, quality-check-service.ts's own + * CL0UD/N0D/ST0RE/AUD0 usage). No ST0RE_URL was available to this pass, so + * this is NOT verified against a live, running ST0RE lakehouse API -- + * src/clients/st0re-lakehouse-client.ts's module doc comment states that + * caveat explicitly and it applies here too. What IS real: the table gets + * registered, a DatasetIcebergMapping row is created referencing ST0RE's + * (mock) tableId, a snapshot gets registered against that table, and all + * four AUD0 events fire in the right order with the right payload shape. + */ +describe("IcebergWriteService", () => { + it("mapDatasetToIcebergTable registers a table in ST0RE and records the mapping", async () => { + const datasetService = new DatasetService(); + const mappingService = new DatasetIcebergMappingService({ datasetService }); + const service = new IcebergWriteService({ datasetService, mappingService }); + + await datasetService.registerDataset({ + dataset_id: "ds_first_iceberg_write", + tenant_id: "t1", + project_id: "p1", + name: "First Iceberg Pilot Dataset", + dataset_type: "enriched", + }); + + const result = await service.mapDatasetToIcebergTable({ + tenant_id: "t1", + project_id: "p1", + dataset_id: "ds_first_iceberg_write", + dataset_iceberg_mapping_id: "dim_first_write", + namespace: "0pt_pilot", + table_name: "dim_company_se_pilot", + storage_uri: "s3://siax-p0/lakehouse/0pt_pilot/dim_company_se_pilot", + catalog_ref: "0pt_pilot.dim_company_se_pilot", + actor_id: "user_123", + correlation_id: "corr_first_write", + }); + + expect(result.iceberg_table_id).toContain("mock_table_"); + expect(result.policy_decision_id).toBeDefined(); + expect(result.mapping.dataset_id).toBe("ds_first_iceberg_write"); + expect(result.mapping.iceberg_table_id).toBe(result.iceberg_table_id); + expect(result.mapping.mapping_status).toBe("active"); + + const lookedUp = await mappingService.getMappingForDataset("ds_first_iceberg_write"); + expect(lookedUp?.dataset_iceberg_mapping_id).toBe("dim_first_write"); + }); + + it("requestIcebergWrite registers a snapshot through ST0RE for an existing mapping", async () => { + const datasetService = new DatasetService(); + const mappingService = new DatasetIcebergMappingService({ datasetService }); + const service = new IcebergWriteService({ datasetService, mappingService }); + + await datasetService.registerDataset({ + dataset_id: "ds_snapshot_write", + tenant_id: "t1", + project_id: "p1", + name: "Snapshot Write Dataset", + dataset_type: "enriched", + }); + const mapped = await service.mapDatasetToIcebergTable({ + tenant_id: "t1", + project_id: "p1", + dataset_id: "ds_snapshot_write", + dataset_iceberg_mapping_id: "dim_snapshot_write", + namespace: "0pt_pilot", + table_name: "dim_company_se_snapshot", + storage_uri: "s3://siax-p0/lakehouse/0pt_pilot/dim_company_se_snapshot", + catalog_ref: "0pt_pilot.dim_company_se_snapshot", + actor_id: "user_123", + correlation_id: "corr_snapshot_setup", + }); + + const result = await service.requestIcebergWrite({ + dataset_iceberg_mapping_id: mapped.mapping.dataset_iceberg_mapping_id, + actor_id: "user_123", + correlation_id: "corr_snapshot_write", + schema_version: "1", + manifest_uri: "s3://siax-p0/lakehouse/0pt_pilot/dim_company_se_snapshot/metadata/manifest-1.avro", + record_count: 1000, + size_bytes: 204800, + }); + + expect(result.snapshot_id).toContain("mock_snapshot_"); + expect(result.iceberg_table_id).toBe(mapped.iceberg_table_id); + expect(result.n0d_job_id).toBeUndefined(); // not a heavy write + }); + + it("requestIcebergWrite routes heavy writes through an N0D job request, not in-process execution", async () => { + const datasetService = new DatasetService(); + const mappingService = new DatasetIcebergMappingService({ datasetService }); + const service = new IcebergWriteService({ datasetService, mappingService }); + + await datasetService.registerDataset({ + dataset_id: "ds_heavy_write", + tenant_id: "t1", + project_id: "p1", + name: "Heavy Backfill Dataset", + dataset_type: "raw", + }); + const mapped = await service.mapDatasetToIcebergTable({ + tenant_id: "t1", + project_id: "p1", + dataset_id: "ds_heavy_write", + dataset_iceberg_mapping_id: "dim_heavy_write", + namespace: "0pt_pilot", + table_name: "fact_financials_backfill", + storage_uri: "s3://siax-p0/lakehouse/0pt_pilot/fact_financials_backfill", + catalog_ref: "0pt_pilot.fact_financials_backfill", + actor_id: "user_123", + correlation_id: "corr_heavy_setup", + }); + + const result = await service.requestIcebergWrite({ + dataset_iceberg_mapping_id: mapped.mapping.dataset_iceberg_mapping_id, + actor_id: "user_123", + correlation_id: "corr_heavy_write", + schema_version: "1", + manifest_uri: "s3://siax-p0/lakehouse/0pt_pilot/fact_financials_backfill/metadata/manifest-backfill.avro", + heavy: true, + }); + + expect(result.n0d_job_id).toBeDefined(); + expect(result.n0d_job_id).toContain("mock_job_"); + }); + + it("requestIcebergWrite rejects writes against a non-active mapping", async () => { + const datasetService = new DatasetService(); + const mappingService = new DatasetIcebergMappingService({ datasetService }); + const service = new IcebergWriteService({ datasetService, mappingService }); + + await datasetService.registerDataset({ + dataset_id: "ds_deprecated_mapping", + tenant_id: "t1", + project_id: "p1", + name: "Deprecated Mapping Dataset", + dataset_type: "raw", + }); + await mappingService.createMapping({ + dataset_iceberg_mapping_id: "dim_deprecated", + tenant_id: "t1", + project_id: "p1", + dataset_id: "ds_deprecated_mapping", + iceberg_table_id: "st0re_table_deprecated", + mapping_status: "deprecated", + }); + + await expect( + service.requestIcebergWrite({ + dataset_iceberg_mapping_id: "dim_deprecated", + actor_id: "user_123", + correlation_id: "corr_deprecated_write", + schema_version: "1", + manifest_uri: "s3://siax-p0/lakehouse/deprecated/manifest.avro", + }) + ).rejects.toThrow("is not active"); + }); + + // TODO(future session): once ST0RE_URL points at a live, reachable ST0RE + // deployment (sax3l/st0re, /v1/st0re/lakehouse/*, port 8090 per that repo's + // STATUS.md), add a live-integration test here that constructs + // IcebergWriteService with { st0reBaseUrl: process.env.ST0RE_URL } and + // asserts the same flow against the real HTTP API. Not built in this pass + // -- no reachable ST0RE endpoint was available to verify against. See + // src/clients/st0re-lakehouse-client.ts's module doc comment and + // docs/ICEBERG_WRITE_BOUNDARY.md "Verification status" section. + it.todo("mapDatasetToIcebergTable against a LIVE ST0RE deployment (ST0RE_URL) -- not live-verified in this pass"); +}); diff --git a/tests/p0/source-rate-policy.test.ts b/tests/p0/source-rate-policy.test.ts new file mode 100644 index 0000000..7db5b1a --- /dev/null +++ b/tests/p0/source-rate-policy.test.ts @@ -0,0 +1,84 @@ +import { describe, it, expect } from "vitest"; +import { SourceService } from "@/src/services/source-service"; +import { SourceRatePolicyService } from "@/src/services/source-rate-policy-service"; + +describe("SourceRatePolicyService", () => { + it("createSourceRatePolicy persists rate limit + crawl budget + compliance notes", async () => { + const sourceService = new SourceService(); + const service = new SourceRatePolicyService({ sourceService }); + await sourceService.createSource({ + source_id: "src_rate_pol_1", + tenant_id: "t1", + project_id: "p1", + source_type: "web", + name: "allabolag.se", + }); + + const policy = await service.createSourceRatePolicy({ + source_rate_policy_id: "srp_test_001", + tenant_id: "t1", + project_id: "p1", + source_id: "src_rate_pol_1", + rate_limit_per_minute: 60, + crawl_budget: 5000, + compliance_notes: "robots.txt allows /bolag/*; authorized_web tier per docs/06-data-engine.md", + domain: "allabolag.se", + robots_ok: true, + access: "authorized_web", + }); + + expect(policy.source_rate_policy_id).toBe("srp_test_001"); + expect(policy.rate_limit_per_minute).toBe(60); + expect(policy.crawl_budget).toBe(5000); + expect(policy.compliance_notes).toContain("robots.txt"); + expect(policy.status).toBe("active"); + }); + + it("createSourceRatePolicy rejects an unknown source_id", async () => { + const service = new SourceRatePolicyService(); + await expect( + service.createSourceRatePolicy({ + source_rate_policy_id: "srp_test_bad", + tenant_id: "t1", + project_id: "p1", + source_id: "src_does_not_exist", + rate_limit_per_minute: 10, + crawl_budget: 100, + }) + ).rejects.toThrow("not found"); + }); + + it("createSourceRatePolicy rejects a non-positive rate_limit_per_minute", async () => { + const sourceService = new SourceService(); + const service = new SourceRatePolicyService({ sourceService }); + await sourceService.createSource({ + source_id: "src_rate_pol_2", + tenant_id: "t1", + project_id: "p1", + source_type: "api", + name: "ratsit.se", + }); + await expect( + service.createSourceRatePolicy({ + source_rate_policy_id: "srp_test_zero", + tenant_id: "t1", + project_id: "p1", + source_id: "src_rate_pol_2", + rate_limit_per_minute: 0, + crawl_budget: 100, + }) + ).rejects.toThrow("rate_limit_per_minute must be > 0"); + }); + + it("listSourceRatePolicies filters by source_id", async () => { + const sourceService = new SourceService(); + const service = new SourceRatePolicyService({ sourceService }); + await sourceService.createSource({ source_id: "src_rate_pol_3", tenant_id: "t1", project_id: "p1", source_type: "web", name: "eniro.se" }); + await sourceService.createSource({ source_id: "src_rate_pol_4", tenant_id: "t1", project_id: "p1", source_type: "web", name: "merinfo.se" }); + await service.createSourceRatePolicy({ source_rate_policy_id: "srp_list_1", tenant_id: "t1", project_id: "p1", source_id: "src_rate_pol_3", rate_limit_per_minute: 30, crawl_budget: 1000 }); + await service.createSourceRatePolicy({ source_rate_policy_id: "srp_list_2", tenant_id: "t1", project_id: "p1", source_id: "src_rate_pol_4", rate_limit_per_minute: 30, crawl_budget: 1000 }); + const results = await service.listSourceRatePolicies({ source_id: "src_rate_pol_3" }); + expect(results).toHaveLength(1); + expect(results[0].source_rate_policy_id).toBe("srp_list_1"); + }); +});