Add 0pt Iceberg write-path ownership: dataset<->table mapping + source rate policy
Real, tested implementation of docs/11-lakehouse-write-plan.md's items 2-8 (the "future implementation session" that doc named): 0pt is now the canonical writer of Iceberg tables through ST0RE's real, shipped /v1/st0re/lakehouse/* table registry (sax3l/st0re commit 9ea0abb), never persisting lakehouse table/catalog/snapshot state itself. - src/types + src/services for 0pt.dataset_iceberg_mapping.schema.v1 (dataset -> ST0RE tableId reference) and 0pt.source_rate_policy.schema.v1 (the sourceRatePolicy ownership repair, moved out of N0D per ADR-0030). - src/clients/st0re-lakehouse-client.ts: new client matching ST0RE's actual camelCase route contract, with the same mock/HTTP boundary convention as this repo's existing cl0ud/n0d/aud0/st0re clients. Honest caveat in its module doc comment: not verified against a live ST0RE instance in this pass (no reachable ST0RE_URL); every test exercises the deterministic mock. - src/services/iceberg-write-service.ts: registers tables/snapshots through ST0RE, requests N0D jobs only for heavy writes (never executes them in-process), emits all four AUD0 event names (dataset_iceberg_table_mapped / iceberg_write_requested / iceberg_snapshot_created / source_rate_policy_created). - sql/004, packages/schemas/*.json, app/api/0pt/lakehouse + sources/rate-policies routes, tests/p0/*.test.ts (154 passed + 1 honest it.todo for the live-ST0RE case, up from 142 baseline, 0 failures). - Reconciled a real collision: an earlier untracked draft schema/test pair (tests/p0/enterprise-data-os-expansion.test.ts) asserted a different, now-superseded source_rate_policy shape; updated that one test block only. - docs/ICEBERG_WRITE_BOUNDARY.md states the ownership boundary plainly and is explicit throughout about what's real-and-tested vs. contract-only. docs/11-lakehouse-write-plan.md gets a status-update section, not a rewrite. - Data quality rules (quality-check-service.ts, quality-report-service.ts) were not touched -- confirmed still 0pt-owned. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
@@ -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 });
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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 });
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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);
|
||||||
|
}
|
||||||
@@ -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 });
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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);
|
||||||
|
}
|
||||||
@@ -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
|
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.
|
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
|
## To-do list
|
||||||
|
|
||||||
1. **Use Apache Iceberg as the canonical write format** for datasets 0pt produces, per
|
1. **Use Apache Iceberg as the canonical write format** for datasets 0pt produces, per
|
||||||
|
|||||||
@@ -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.
|
||||||
@@ -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" }
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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" }
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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);
|
||||||
@@ -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<St0reLakehouseNamespace>;
|
||||||
|
registerTable(input: St0reIcebergTableInput): Promise<St0reIcebergTable>;
|
||||||
|
getTable(tableId: string): Promise<St0reIcebergTable>;
|
||||||
|
createSnapshot(tableId: string, input: St0reTableSnapshotInput): Promise<St0reTableSnapshot>;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function createSt0reLakehouseClient(baseUrl?: string): St0reLakehouseClient {
|
||||||
|
if (!baseUrl) return new MockSt0reLakehouseClient();
|
||||||
|
return new HttpSt0reLakehouseClient(baseUrl);
|
||||||
|
}
|
||||||
|
|
||||||
|
class MockSt0reLakehouseClient implements St0reLakehouseClient {
|
||||||
|
private tables: Map<string, St0reIcebergTable> = new Map();
|
||||||
|
|
||||||
|
async registerNamespace(input: St0reLakehouseNamespaceInput): Promise<St0reLakehouseNamespace> {
|
||||||
|
return { ...input, namespaceId: `mock_ns_${Date.now()}`, createdAt: Date.now() };
|
||||||
|
}
|
||||||
|
|
||||||
|
async registerTable(input: St0reIcebergTableInput): Promise<St0reIcebergTable> {
|
||||||
|
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<St0reIcebergTable> {
|
||||||
|
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<St0reTableSnapshot> {
|
||||||
|
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<St0reLakehouseNamespace> {
|
||||||
|
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<St0reIcebergTable> {
|
||||||
|
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<St0reIcebergTable> {
|
||||||
|
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<St0reTableSnapshot> {
|
||||||
|
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();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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<string, DatasetIcebergMappingRecord> = 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<DatasetIcebergMapping> {
|
||||||
|
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<DatasetIcebergMapping> {
|
||||||
|
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<DatasetIcebergMapping | undefined> {
|
||||||
|
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<DatasetIcebergMapping[]> {
|
||||||
|
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,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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<MapDatasetToIcebergTableResult> {
|
||||||
|
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<RequestIcebergWriteResult> {
|
||||||
|
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,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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<string, SourceRatePolicyRecord> = 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<SourceRatePolicy> {
|
||||||
|
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<SourceRatePolicy> {
|
||||||
|
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<SourceRatePolicy[]> {
|
||||||
|
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,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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;
|
||||||
|
}
|
||||||
@@ -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;
|
||||||
|
}
|
||||||
@@ -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");
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -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");
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -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");
|
||||||
|
});
|
||||||
@@ -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");
|
||||||
|
});
|
||||||
|
});
|
||||||
Reference in New Issue
Block a user