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
|
||||
rules) and this doc just confirms 0pt intends to keep them that way.
|
||||
|
||||
## 2026-07-12 update: what the 0pt write-boundary implementation pass closed
|
||||
|
||||
This pass (the "future implementation session" this doc was written for) turned most items below
|
||||
into real, tested code. Full detail in `docs/ICEBERG_WRITE_BOUNDARY.md`; short version:
|
||||
|
||||
- Item 2 (`0pt.dataset_iceberg_mapping`): CLOSED as real code -- `src/types/dataset-iceberg-mapping.ts`,
|
||||
`src/services/dataset-iceberg-mapping-service.ts`, `packages/schemas/0pt.dataset_iceberg_mapping.schema.json`,
|
||||
`sql/004_iceberg_write_boundary.sql` (`opt_dataset_iceberg_mappings`), tested in
|
||||
`tests/p0/dataset-iceberg-mapping.test.ts`. Field names differ slightly from the "suggested shape"
|
||||
above (`dataset_iceberg_mapping_id` as PK, `iceberg_table_id` instead of `iceberg_table_ref`,
|
||||
`mapping_status` instead of bare `entity_type`/`market`) -- reconciled against ST0RE's REAL
|
||||
shipped `IcebergTable.tableId` shape (sax3l/st0re, not the "future" registry this doc assumed) and
|
||||
against this session's own task spec; see the boundary doc for the full reconciliation note.
|
||||
- Item 3 (first test dataset write): CLOSED against ST0RE's now-real (but not live-verified from
|
||||
this environment) `/v1/st0re/lakehouse/*` API -- `src/clients/st0re-lakehouse-client.ts`,
|
||||
`src/services/iceberg-write-service.ts`, proven in `tests/p0/iceberg-write-service.test.ts`
|
||||
against that client's mock implementation (same mock/HTTP convention as `cl0ud-client.ts` etc).
|
||||
NOT verified against a running ST0RE deployment in this pass -- see the boundary doc's
|
||||
"Verification status" section and the `it.todo` in that test file.
|
||||
- Item 4 (quality rules stay in 0pt): CONFIRMED, untouched -- `src/services/quality-check-service.ts`
|
||||
and `src/services/quality-report-service.ts` were not modified by this pass.
|
||||
- Item 5 (`0pt.source_rate_policy`): CLOSED as real code -- `src/types/source-rate-policy.ts`,
|
||||
`src/services/source-rate-policy-service.ts`, `packages/schemas/0pt.source_rate_policy.schema.json`,
|
||||
`sql/004_iceberg_write_boundary.sql` (`opt_source_rate_policies`), tested in
|
||||
`tests/p0/source-rate-policy.test.ts`. Kept `requests_per_second`-style fields (as `domain`/
|
||||
`robots_ok`/`access`) as OPTIONAL alongside the task-spec's primary `rate_limit_per_minute`/
|
||||
`crawl_budget`/`compliance_notes` fields -- both specs for this same contract are satisfied on one
|
||||
row, not two competing shapes.
|
||||
- Item 6 (four AUD0 events): CLOSED for all four names -- `0pt.dataset_iceberg_table_mapped` and
|
||||
`0pt.iceberg_write_requested`/`0pt.iceberg_snapshot_created` emit from
|
||||
`src/services/iceberg-write-service.ts` and `dataset-iceberg-mapping-service.ts`;
|
||||
`0pt.source_rate_policy_created` emits from `src/services/source-rate-policy-service.ts`. All via
|
||||
the existing `Aud0Client` (mock/no-op by default, same as every other AUD0 emission in this repo --
|
||||
AUD0 itself still has no live ingestion endpoint).
|
||||
- Item 7 (N0D heavy-write requests): CLOSED as real code -- `IcebergWriteService.requestIcebergWrite`
|
||||
requests an N0D job (`0pt.iceberg_write_job`) only when `heavy: true`, never executes in-process;
|
||||
tested in `tests/p0/iceberg-write-service.test.ts`.
|
||||
- Item 8 (no lakehouse state outside ST0RE): CLOSED as a design property, not just a statement --
|
||||
`IcebergWriteService` and `DatasetIcebergMappingService` hold zero table/snapshot fields; the only
|
||||
thing 0pt persists is the mapping's `iceberg_table_id` reference. See the boundary doc.
|
||||
- Item 1 (Iceberg canonical, Delta rejected): CONFIRMED, unchanged from `docs/10` --
|
||||
`IcebergWriteService.mapDatasetToIcebergTable` defaults `table_format` to `"iceberg"`.
|
||||
|
||||
## To-do list
|
||||
|
||||
1. **Use Apache Iceberg as the canonical write format** for datasets 0pt produces, per
|
||||
|
||||
@@ -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