From 62ae4ec9426fe1205364df08366ff94caa3cb95e Mon Sep 17 00:00:00 2001 From: curtis Date: Tue, 4 Aug 2026 11:56:20 -0400 Subject: [PATCH 1/2] fix(conformance): make the mock reproduce Monad's paging + component_of [PRO-455] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mock returned every row regardless of `limit` and had no `GET /v1/{org}/{kind}s/{id}` route at all, so it could not catch two bugs the routers actually have: they never send `limit` (Monad defaults it to 10) and they walk the whole pipeline list to answer a question `component_of` answers in one call. - page every list at Monad's default limit of 10, with no maximum - serve the connector detail route, including `component_of` — pinned to the same narrow projection the datastore emits (no `status`, no `nodes`) - accept `type` on output create, `output_type` staying a deprecated alias - track connectors statefully so a delete is observable - add the >10-pipelines scenario; it fails against the pre-fix routers --- conformance/mock_monad.py | 196 +++++++++++++++++++++++++----- conformance/monad_schemas.py | 36 ++++++ conformance/test_mock_fidelity.py | 52 +++++++- conformance/test_scenarios.py | 58 ++++++++- 4 files changed, 305 insertions(+), 37 deletions(-) diff --git a/conformance/mock_monad.py b/conformance/mock_monad.py index 2fec298..e89e5ef 100644 --- a/conformance/mock_monad.py +++ b/conformance/mock_monad.py @@ -11,12 +11,20 @@ - ``GET /v1/{kind}s`` (catalog) → a bare array of connector types. - ``GET /v1/{org}/{kind}s`` → ``{ "s": [...] | null, "pagination": {...} }`` (the list is **null**, not ``[]``, when empty). +- ``GET /v1/{org}/{kind}s/{id}`` → the connector record + ``component_of``, + the pipelines it is a node of. - ``POST /v2/{org}/pipelines`` → **201** with the full pipeline object. - ``GET /v2/{org}/pipelines`` → ``{ "pipelines": [...], "pagination": {...} }``. - ``GET /v2/{org}/pipelines/{id}`` → the full pipeline object at top level (``nodes``/``edges``/``enabled`` — no ``config`` wrapper). - ``POST /v2/{org}/outputs`` → the created output object. - ``POST /v3/sessions`` → ``{ "session_token", "expires_at" }``. + +**Pagination is enforced, and the default limit is 10** — matching every list +handler in Monad (``api/pkg/routes/v2/routes/pipelines.go:727`` and friends). +A router that omits ``limit`` gets 10 rows here exactly as it would in +production, which is what turns the "only ever sees 10" class of bug into a +test failure instead of a live incident. """ from __future__ import annotations @@ -24,13 +32,71 @@ import json import re import threading +import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +# Monad's list handlers all default to 10 and enforce no maximum. +DEFAULT_LIMIT = 10 + +# Enough configured inputs that a router which does not page sees a truncated +# list. `test_connectors_pagination_is_exhaustive` pins the full count. +_SEEDED_INPUTS = 12 + +# Inputs the bulk-pipeline scenario wires, one pipeline each, to push the +# pipeline list past its first page. +_BULK_INPUTS = 12 + + +def _seed_connectors() -> dict[str, dict[str, dict]]: + """The tenant's configured connectors, keyed by id. + + `outputs` is deliberately empty so `GET /v1/{org}/outputs` returns **null** + — the real-Monad behavior its own spec gets wrong, and the reason live mode + exists. `out_store`/`out_conf` are the host's pre-provisioned components: + resolvable by id, but not part of the tenant's own list. + """ + inputs = { + f"cfg_{i}": { + "id": f"cfg_{i}", + "type": "aws-cloudtrail" if i % 2 else "okta-systemlog", + "name": f"Configured {i}", + "organization_id": "org_conf", + "config": {"settings": {}, "secrets": {}}, + } + for i in range(1, _SEEDED_INPUTS + 1) + } + # Ids the lifecycle scenarios wire, resolvable but outside the listed page + # so they cannot perturb the pagination counts. `bulk_*` exists to push the + # pipeline list past its first page. + extras = ["in_conf", "in_dup_conflict"] + [f"bulk_{i}" for i in range(1, _BULK_INPUTS + 1)] + for extra in extras: + inputs[extra] = { + "id": extra, + "type": "aws-cloudtrail", + "name": extra, + "organization_id": "org_conf", + "config": {"settings": {}, "secrets": {}}, + "listed": False, + } + outputs = { + oid: { + "id": oid, + "type": "dev-null", + "name": oid, + "organization_id": "org_conf", + "config": {"settings": {}, "secrets": {}}, + "listed": False, + } + for oid in ("out_store", "out_conf") + } + return {"input": inputs, "output": outputs} + class _State: def __init__(self) -> None: self.lock = threading.Lock() self.pipelines: dict[str, dict] = {} + self.connectors = _seed_connectors() self._counter = 0 def new_id(self, prefix: str) -> str: @@ -39,6 +105,16 @@ def new_id(self, prefix: str) -> str: return f"{prefix}_{self._counter}" +def _paginate(rows: list, limit: int, offset: int) -> tuple[list | None, dict]: + """Slice `rows` the way Monad does, and build the `pagination` sibling. + + An empty page comes back as ``None``, not ``[]`` — see the module note. + `total` is the full count, not the page length. + """ + page = rows[offset : offset + limit] + return (page or None), {"limit": limit, "offset": offset, "total": len(rows)} + + def _component_ids(pipeline: dict) -> frozenset: """The set of component ids a pipeline wires — used to detect a duplicate connection (Monad rejects connecting the same components twice → 409).""" @@ -84,11 +160,14 @@ def _pipeline_summary(p: dict) -> dict: def _output_view(oid: str, body: dict) -> dict: + # `type` is the canonical field; `output_type` is the deprecated alias the + # API still accepts, with `type` winning when both are sent + # (api/pkg/routes/v2/routes/organization_outputs.go:188). return { "id": oid, "name": body.get("name", ""), "description": body.get("description", ""), - "type": body.get("output_type", "dev-null"), + "type": body.get("type") or body.get("output_type") or "dev-null", "organization_id": "org_conf", "managed_by": "", "config": {"settings": {}, "secrets": {}}, @@ -97,6 +176,36 @@ def _output_view(oid: str, body: dict) -> dict: } +def _connector_view(record: dict) -> dict: + """A configured connector as the list/detail endpoints return it.""" + return {k: v for k, v in record.items() if k != "listed"} + + +def _component_of(pipelines: dict[str, dict], component_id: str) -> list[dict]: + """The pipelines a component is a node of — Monad's `component_of`. + + Deliberately the same narrow projection the datastore emits + (core/pkg/datastore/postgres/pipelines.go:292): **no `status`, no `nodes`**. + A router that needs the peer component has to fetch the pipeline detail. + """ + return [ + { + "id": p.get("id"), + "organization_id": "org_conf", + "name": p.get("name", ""), + "description": p.get("description", ""), + "enabled": bool(p.get("enabled")), + "component_tier": 0, + "input_id": "", + "managed_by": "", + "created_at": "2026-01-01T00:00:00Z", + "updated_at": "2026-01-01T00:00:00Z", + } + for p in pipelines.values() + if component_id in _component_ids(p) + ] + + def _make_handler(state: _State): class Handler(BaseHTTPRequestHandler): def log_message(self, *_args) -> None: # silence per-request logging @@ -133,6 +242,20 @@ def do_PATCH(self): # noqa: N802 def do_DELETE(self): # noqa: N802 self._route("DELETE") + def _limit_offset(self) -> tuple[int, int]: + """Parse `limit`/`offset` exactly as Monad's handlers do: anything + unparseable or out of range falls back to the default.""" + query = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query) + + def _int(name: str, default: int, floor: int) -> int: + try: + value = int(query.get(name, [""])[0]) + except (TypeError, ValueError): + return default + return value if value >= floor else default + + return _int("limit", DEFAULT_LIMIT, 1), _int("offset", 0, 0) + def _route(self, method: str) -> None: path = self.path.split("?", 1)[0] body = self._read() if method in ("POST", "PATCH") else {} @@ -166,35 +289,49 @@ def _route(self, method: str) -> None: ) # A tenant's configured connectors — wrapped alongside `pagination`, - # and returned as null (not []) when the tenant has none. Exercise - # both shapes: a populated `inputs`, an empty (null) `outputs`. + # paged with Monad's default limit of 10, and returned as null (not + # []) when the page is empty. Exercise both shapes: a populated + # `inputs` that spans two pages, an empty (null) `outputs`. m = re.fullmatch(r"/v1/([^/]+)/(inputs|outputs)", path) if method == "GET" and m: kind = m.group(2) - rows = ( - None - if kind == "outputs" - else [ - { - "id": "cfg_1", - "type": "aws-cloudtrail", - "name": "Configured", - "organization_id": "org_conf", - "config": {"settings": {}, "secrets": {}}, - } + limit, offset = self._limit_offset() + with state.lock: + all_rows = [ + _connector_view(c) + for c in state.connectors[kind[:-1]].values() + if c.get("listed", True) ] - ) - return self._send( - 200, - { - kind: rows, - "pagination": {"limit": 1000, "offset": 0, "total": 0 if rows is None else 1}, - }, - ) + rows, pagination = _paginate(all_rows, limit, offset) + return self._send(200, {kind: rows, "pagination": pagination}) + + # A single configured connector + `component_of`: the pipelines it is + # a node of. This is what lets a router resolve a connector's + # pipeline in one call instead of walking every pipeline in the org. + m = re.fullmatch(r"/v1/([^/]+)/(inputs|outputs)/([^/]+)", path) + if m and method in ("GET", "DELETE"): + kind, cid = m.group(2)[:-1], m.group(3) + with state.lock: + record = state.connectors[kind].get(cid) + if record is None: + return self._send(404, {"error": f"{kind} not found"}) + if method == "DELETE": + del state.connectors[kind][cid] + return self._send(204) + view = { + **_connector_view(record), + "component_of": _component_of(state.pipelines, cid), + } + return self._send(200, view) # Create an output (e.g. the dev/null sink) — the created record. + # It joins the tenant's connectors so it is resolvable by id and + # deletable, like any other output. if method == "POST" and re.fullmatch(r"/v2/([^/]+)/outputs", path): - return self._send(200, _output_view(state.new_id("out"), body)) + view = _output_view(state.new_id("out"), body) + with state.lock: + state.connectors["output"][view["id"]] = {**view, "listed": False} + return self._send(200, view) # pipelines collection (create / list) — check before status/detail if re.fullmatch(r"/v2/([^/]+)/pipelines/?", path): @@ -214,15 +351,11 @@ def _route(self, method: str) -> None: # Real Monad returns 201 with the full pipeline record. return self._send(201, view) if method == "GET": + limit, offset = self._limit_offset() with state.lock: items = [_pipeline_summary(p) for p in state.pipelines.values()] - return self._send( - 200, - { - "pipelines": items, - "pagination": {"limit": 1000, "offset": 0, "total": len(items)}, - }, - ) + page, pagination = _paginate(items, limit, offset) + return self._send(200, {"pipelines": page or [], "pagination": pagination}) if method == "GET" and re.fullmatch(r"/v2/([^/]+)/pipelines/([^/]+)/status", path): pid = path.rsplit("/", 2)[-2] @@ -259,9 +392,6 @@ def _route(self, method: str) -> None: state.pipelines.pop(pid, None) return self._send(204) - if method == "DELETE" and re.fullmatch(r"/v1/([^/]+)/(inputs|outputs)/([^/]+)", path): - return self._send(204) - return self._send(404, {"error": f"mock unhandled {method} {path}"}) return Handler diff --git a/conformance/monad_schemas.py b/conformance/monad_schemas.py index 5494bf1..af4cba1 100644 --- a/conformance/monad_schemas.py +++ b/conformance/monad_schemas.py @@ -18,6 +18,11 @@ ``null`` (not ``[]``) for a tenant with none, though its spec types it ``array``. Live conformance is what proves this, but pinning it here stops the mock from regressing to a non-null-emitting shape. +- ``component_of`` entries are pinned *without* ``status`` or ``nodes``: the + datastore projection behind them fills only a subset of ``models.Pipeline`` + (core/pkg/datastore/postgres/pipelines.go:292), so a router that tried to read + a pipeline's wiring straight off ``component_of`` would work against a + too-generous mock and fail live. """ # POST /v3/sessions — the embed session mint (swagger leaves the body untyped; @@ -69,6 +74,37 @@ def connectors_list(kind: str) -> dict: } +def connector_detail(kind: str) -> dict: + """GET /v1/{org}/{kind}s/{id} — the connector record plus ``component_of``, + the pipelines it is a node of. This is the one call that answers "which + pipeline is this connector wired into", so ``component_of`` is required + even when empty.""" + return { + "type": "object", + "required": ["id", "type", "name", "component_of"], + "properties": { + "id": {"type": "string"}, + "type": {"type": "string"}, + "name": {"type": "string"}, + "component_of": { + "type": "array", + "items": { + "type": "object", + "required": ["id", "enabled"], + "properties": { + "id": {"type": "string"}, + "name": {"type": "string"}, + "enabled": {"type": "boolean"}, + # Pinned absent — see the module note. + "status": False, + "nodes": False, + }, + }, + }, + }, + } + + # A pipeline node as the routers read it (match on component_type/component_id). _NODE = { "type": "object", diff --git a/conformance/test_mock_fidelity.py b/conformance/test_mock_fidelity.py index 7fd4aeb..211d229 100644 --- a/conformance/test_mock_fidelity.py +++ b/conformance/test_mock_fidelity.py @@ -64,10 +64,58 @@ def test_connectors_list_shape(http, kind): _valid(S.connectors_list(kind), r.json()) -def test_output_create_shape(http): - r = http.post(f"/v2/{_ORG}/outputs", json={"output_type": "dev-null", "name": "sink"}) +def test_connectors_list_defaults_to_ten(http): + # Monad's list handlers default `limit` to 10 with no maximum. A router that + # omits it gets a truncated page — the mock must reproduce that, not paper + # over it, or the "only ever sees 10" bug stays invisible until production. + unpaged = http.get(f"/v1/{_ORG}/inputs").json() + assert len(unpaged["inputs"]) == 10 + assert unpaged["pagination"] == {"limit": 10, "offset": 0, "total": mock_monad._SEEDED_INPUTS} + + rest = http.get(f"/v1/{_ORG}/inputs?limit=10&offset=10").json() + assert len(rest["inputs"]) == mock_monad._SEEDED_INPUTS - 10 + # Past the end the list is null, not [] — the same shape as "tenant has none". + assert http.get(f"/v1/{_ORG}/inputs?limit=10&offset=99").json()["inputs"] is None + + +@pytest.mark.parametrize("kind", ["input", "output"]) +def test_connector_detail_shape(http, kind): + cid = "in_conf" if kind == "input" else "out_store" + r = http.get(f"/v1/{_ORG}/{kind}s/{cid}") + assert r.status_code == 200, r.text + _valid(S.connector_detail(kind), r.json()) + # Unconnected component → an empty `component_of`, not a missing key. + assert r.json()["component_of"] == [] + assert http.get(f"/v1/{_ORG}/{kind}s/nope_404").status_code == 404 + + +def test_connector_detail_reports_component_of(http): + body = { + "name": "component-of", + "enabled": True, + "nodes": [ + {"slug": "in", "component_id": "cfg_3", "component_type": "input", "enabled": True}, + {"slug": "out", "component_id": "out_store", "component_type": "output", "enabled": True}, + ], + "edges": [], + } + pid = http.post(f"/v2/{_ORG}/pipelines/", json=body).json()["id"] + for cid in ("cfg_3", "out_store"): + kind = "inputs" if cid.startswith("cfg") else "outputs" + entry = http.get(f"/v1/{_ORG}/{kind}/{cid}").json()["component_of"] + assert [p["id"] for p in entry] == [pid] + assert entry[0]["enabled"] is True + http.delete(f"/v2/{_ORG}/pipelines/{pid}") + + +@pytest.mark.parametrize("field", ["type", "output_type"]) +def test_output_create_shape(http, field): + # `type` is canonical; `output_type` is the deprecated alias the API still + # accepts. Both must round-trip to the same created record. + r = http.post(f"/v2/{_ORG}/outputs", json={field: "dev-null", "name": "sink"}) assert r.status_code == 200 _valid(S.OUTPUT, r.json()) + assert r.json()["type"] == "dev-null" def test_pipeline_lifecycle_shapes(http): diff --git a/conformance/test_scenarios.py b/conformance/test_scenarios.py index 6e135b1..4c0a549 100644 --- a/conformance/test_scenarios.py +++ b/conformance/test_scenarios.py @@ -22,6 +22,8 @@ import httpx import pytest +import mock_monad + _BASE_URL = os.environ.get("ROUTER_BASE_URL", "http://127.0.0.1:8791") LIVE = os.environ.get("MONAD_LIVE") == "1" @@ -46,6 +48,13 @@ reason="mutating live scenario — set MONAD_LIVE_MUTATE=1 with throwaway CONF_INPUT_ID/CONF_OUTPUT_ID", ) +# Scenarios that assert against the mock's seeded fixtures (exact row counts, +# bulk pipeline creation). They pin router logic, not Monad behavior, so they +# run hermetically only — a real tenant has its own arbitrary inventory. +mock_only = pytest.mark.skipif( + LIVE, reason="asserts against the mock's seeded fixture counts; a live tenant has its own" +) + @pytest.fixture() def client(): @@ -82,6 +91,47 @@ def test_connectors_list_parses(client): assert {"id", "typeId", "name"} <= row.keys() +@mock_only +def test_connectors_pagination_is_exhaustive(client): + # Monad pages every list at limit=10 by default and the /embed contract + # returns a bare array, so the router owns draining the pages. A guard, not + # a gate: it fails a router that sends no `limit` at all, but it cannot + # distinguish exhaustive paging from a large hardcoded limit — the mock, + # like Monad, enforces no maximum. Correctness past a fixed ceiling is what + # live mode is for. + rows = client.get("/embed/connectors", params={"kind": "input"}).json() + assert len(rows) == mock_monad._SEEDED_INPUTS, ( + f"expected all {mock_monad._SEEDED_INPUTS} configured inputs, got {len(rows)} " + "— the router is not draining pages" + ) + assert len({r["id"] for r in rows}) == len(rows), "paging returned duplicates" + + +@mock_only +def test_status_resolves_past_the_first_page_of_pipelines(client): + # The regression gate for the "only ever sees 10 pipelines" bug: resolving a + # connector's pipeline must not depend on where that pipeline falls in the + # org's list. A router that scans `GET /v2/{org}/pipelines/` sees only the + # first page and reports hasPipeline:false for everything after it. + ids = [f"bulk_{i}" for i in range(1, 13)] + try: + for cid in ids: + r = client.post("/embed/pipelines/ingress", json={"inputId": cid, "name": cid}) + assert r.status_code == 201, r.text + + # The last one is well past page 1 (limit=10). + last = ids[-1] + r = client.get("/embed/pipelines", params={"connectorId": last, "kind": "input"}) + assert r.status_code == 200, r.text + assert r.json()["hasPipeline"] is True, ( + f"{last}'s pipeline was created but did not resolve — the router is " + "only seeing the first page of pipelines" + ) + finally: + for cid in ids: + client.post("/embed/pipelines/remove", json={"connectorId": cid, "kind": "input"}) + + def test_catalog_is_allow_listed(client): r = client.get("/embed/catalog", params={"kind": "input"}) assert r.status_code == 200 @@ -127,12 +177,16 @@ def test_ingress_lifecycle(client): r = client.get("/embed/pipelines", params={"connectorId": INPUT_ID, "kind": "input"}) assert r.json()["enabled"] is False - # 5) remove — the provisioned store is kept, the pipeline + input are gone + # 5) remove — the provisioned store is kept, the pipeline + input are gone. + # The input no longer exists upstream, so asking for its pipeline is a 404, + # not a 200 saying "no pipeline": the router resolves status by fetching the + # connector, and a deleted connector is genuinely not found. r = client.post("/embed/pipelines/remove", json={"connectorId": INPUT_ID, "kind": "input"}) assert r.status_code == 204, r.text r = client.get("/embed/pipelines", params={"connectorId": INPUT_ID, "kind": "input"}) - assert r.json()["hasPipeline"] is False + assert r.status_code == 404, r.text + assert r.json()["code"] == "not_found" @skip_mutation From 9dbe1af15ea5214925d3d8065a8638cbe14ed469 Mon Sep 17 00:00:00 2001 From: curtis Date: Tue, 4 Aug 2026 11:56:20 -0400 Subject: [PATCH 2/2] fix(embed): correct the TypeScript router against the real Monad API [PRO-455] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of the Go port (#15) surfaced defects the TypeScript reference has too. Validated against the API source rather than the published spec, which is stale. - resolve a connector's pipeline via `GET /v1/{org}/{kind}s/{id}`'s `component_of` instead of scanning every pipeline. The scan sent no `limit`, so past the tenant's tenth pipeline it reported "not connected" for pipelines that exist. - `setEnabled` now PATCHes `{enabled}` alone. PATCH became a true partial update in api#2163; the read-modify-write it replaces silently dropped node `config_overrides` and edge `schema_detection_spec` on every toggle. - drain the connector list instead of one `limit=1000` request - send `type` on output create; `output_type` is deprecated (api#2156) - drop the speculative `data` / `config` response-shape fallbacks — Monad returns neither - map `kind` to its path segment explicitly rather than appending "s" --- routers/typescript/src/core.ts | 20 ++-- routers/typescript/src/monad.ts | 159 ++++++++++++++++---------------- 2 files changed, 84 insertions(+), 95 deletions(-) diff --git a/routers/typescript/src/core.ts b/routers/typescript/src/core.ts index 9170968..1bc7699 100644 --- a/routers/typescript/src/core.ts +++ b/routers/typescript/src/core.ts @@ -214,16 +214,7 @@ export function createEmbedHandler( throw new EmbedError(400, 'invalid_request', "Query 'connectorId' is required."); } const kind = requireKind(req.query.kind, "Query 'kind'"); - if (kind === 'input') { - return json(200, await upstream(monad.pipelineStatus(org, connectorId))); - } - const r = await upstream(monad.findPipelineByOutput(org, connectorId)); - return json(200, { - hasPipeline: r !== null, - enabled: r?.enabled ?? false, - pipelineId: r?.pipelineId, - inputId: r?.inputId - }); + return json(200, await upstream(monad.pipelineFor(org, kind, connectorId))); } if (method === 'POST' && path === '/pipelines/state') { @@ -241,8 +232,8 @@ export function createEmbedHandler( (req.body as Record | undefined)?.kind as string | undefined, "Field 'kind'" ); + const status = await upstream(monad.pipelineFor(org, kind, connectorId)); if (kind === 'input') { - const status = await upstream(monad.pipelineStatus(org, connectorId)); const prov = await provision(org); const keepStore = Boolean( prov.destinationOutputId && @@ -257,10 +248,13 @@ export function createEmbedHandler( ) ); } else { - const r = await upstream(monad.findPipelineByOutput(org, connectorId)); // Keep the shared source input; remove only the pipeline + output. await upstream( - monad.remove(org, { pipelineId: r?.pipelineId, outputId: connectorId }, { input: false }) + monad.remove( + org, + { pipelineId: status.pipelineId, outputId: connectorId }, + { input: false } + ) ); } return noContent(); diff --git a/routers/typescript/src/monad.ts b/routers/typescript/src/monad.ts index e1bb381..c84e7b3 100644 --- a/routers/typescript/src/monad.ts +++ b/routers/typescript/src/monad.ts @@ -51,10 +51,19 @@ export interface CleanupPolicy { const POLL_ATTEMPTS = 15; const POLL_INTERVAL_MS = 2000; +/** Rows per request when draining a Monad list. Monad defaults `limit` to 10 + * and enforces no maximum, so this is purely a round-trip/response-size + * trade-off — correctness comes from draining, not from the size. */ +const PAGE_SIZE = 200; + /** URL-encode a single path segment so browser-supplied ids can't inject query * params or traverse the path (`/`, `?`, `#` are neutralised). */ const seg = (s: string): string => encodeURIComponent(s); +/** The contract's singular `kind` → Monad's collection path segment. Explicit + * rather than `${kind}s`, so the mapping is a fact you can read. */ +const collection = (kind: ComponentKind): string => (kind === 'input' ? 'inputs' : 'outputs'); + /** * Thrown when a Monad API call fails. Carries the status + internal detail for * server-side logging; the router surfaces only the generic super() message to @@ -108,20 +117,37 @@ export class MonadApi { } async listCatalog(kind: ComponentKind, allow?: string[]): Promise { - const types = (await this.req(`/v1/${kind}s`)) as { type_id: string; name: string }[]; - const list = types.map((t) => ({ typeId: t.type_id, name: t.name })); + const types = (await this.req(`/v1/${collection(kind)}`)) as { + type_id: string; + name: string; + }[]; + const list = (types ?? []).map((t) => ({ typeId: t.type_id, name: t.name })); if (!allow || allow.length === 0) return list; const set = new Set(allow); return list.filter((t) => set.has(t.typeId)); } + /** + * Every configured connector of a kind. Monad pages this at `limit=10` by + * default with no maximum, and the `/embed` contract returns a bare array + * with no pagination — so the router owns draining the pages. Sending one + * large `limit` instead would silently truncate whichever tenant outgrows it. + */ async listConnectors(org: string, kind: ComponentKind): Promise { - const page = (await this.req(`/v1/${seg(org)}/${kind}s?limit=1000&offset=0`)) as Record< - string, - { id: string; type: string; name: string }[] - >; - const rows = page?.[`${kind}s`] ?? []; - return rows.map((r) => ({ id: r.id, typeId: r.type, name: r.name })); + const key = collection(kind); + const rows: ConfiguredConnector[] = []; + for (let offset = 0; ; offset += PAGE_SIZE) { + const page = await this.req(`/v1/${seg(org)}/${key}?limit=${PAGE_SIZE}&offset=${offset}`); + // Monad returns `null`, not `[]`, for a page with no rows. + const items = (page?.[key] ?? []) as { id: string; type: string; name: string }[]; + rows.push(...items.map((r) => ({ id: r.id, typeId: r.type, name: r.name }))); + // A short page is the last page; `total` just lets us stop one round + // trip earlier when the count divides evenly. + if (items.length < PAGE_SIZE) break; + const total = page?.pagination?.total; + if (typeof total === 'number' && rows.length >= total) break; + } + return rows; } private async wire( @@ -167,7 +193,8 @@ export class MonadApi { const output = (await this.req(`/v2/${seg(org)}/outputs`, { method: 'POST', body: JSON.stringify({ - output_type: 'dev-null', + // `type` is the canonical field; `output_type` is a deprecated alias. + type: 'dev-null', name: `${name} → /dev/null`, description: 'Auto-created sink for embed pipeline', promise_id: '', @@ -194,88 +221,56 @@ export class MonadApi { return this.wire(org, opts.fromInputId, opts.outputId, opts.name); } - private async pipelines(org: string): Promise { - const listed = await this.req(`/v2/${seg(org)}/pipelines/`); - return Array.isArray(listed) ? listed : (listed?.pipelines ?? listed?.data ?? []); - } - private async detail(org: string, id: string): Promise { - const d = await this.req(`/v2/${seg(org)}/pipelines/${seg(id)}`); - return d?.config ?? d ?? {}; + return (await this.req(`/v2/${seg(org)}/pipelines/${seg(id)}`)) ?? {}; } - async pipelineStatus(org: string, inputId: string): Promise { - for (const summary of await this.pipelines(org)) { - if (!summary?.id) continue; - let p: any; - try { - p = await this.detail(org, summary.id); - } catch { - continue; - } - const nodes: any[] = p.nodes ?? []; - const inNode = nodes.find((n) => n.component_type === 'input' && n.component_id === inputId); - if (!inNode) continue; - const outNode = nodes.find((n) => n.component_type === 'output'); - return { - hasPipeline: true, - enabled: Boolean(p.enabled), - pipelineId: summary.id, - outputId: outNode?.component_id - }; - } - return { hasPipeline: false, enabled: false }; - } + /** + * The pipeline a configured connector is wired into. + * + * Monad answers this directly: `GET /v1/{org}/{kind}s/{id}` returns + * `component_of`, the pipelines the component is a node of. Walking the + * org's pipeline list instead would be both O(n) and wrong — that list pages + * at 10, so any pipeline past the first page would read as "not connected". + * + * `component_of` carries no wiring (the datastore fills only a summary + * projection), so resolving the peer connector costs one further fetch. + * Throws `UpstreamError(404)` when the connector itself does not exist. + */ + async pipelineFor( + org: string, + kind: ComponentKind, + connectorId: string + ): Promise { + const connector = await this.req(`/v1/${seg(org)}/${collection(kind)}/${seg(connectorId)}`); + const [pipeline] = (connector?.component_of ?? []) as any[]; + if (!pipeline?.id) return { hasPipeline: false, enabled: false }; - async findPipelineByOutput(org: string, outputId: string): Promise { - for (const summary of await this.pipelines(org)) { - if (!summary?.id) continue; - let p: any; - try { - p = await this.detail(org, summary.id); - } catch { - continue; - } - const nodes: any[] = p.nodes ?? []; - const outNode = nodes.find( - (n) => n.component_type === 'output' && n.component_id === outputId - ); - if (!outNode) continue; - const inNode = nodes.find((n) => n.component_type === 'input'); - return { - hasPipeline: true, - enabled: Boolean(p.enabled), - pipelineId: summary.id, - inputId: inNode?.component_id - }; - } - return null; + const peerType = kind === 'input' ? 'output' : 'input'; + const nodes: any[] = (await this.detail(org, pipeline.id)).nodes ?? []; + const peer = nodes.find((n) => n.component_type === peerType)?.component_id; + + return { + hasPipeline: true, + enabled: Boolean(pipeline.enabled), + pipelineId: pipeline.id, + ...(kind === 'input' ? { outputId: peer } : { inputId: peer }) + }; } + /** + * Flip a pipeline's enabled flag. + * + * `PATCH` is a true partial update: omitted fields keep their stored value + * and the node/edge graph is preserved untouched. Reading the pipeline and + * sending it back would replace the graph with whatever subset of fields the + * round trip happened to reproduce — silently dropping node + * `config_overrides` and edge `schema_detection_spec`. + */ async setEnabled(org: string, pipelineId: string, enabled: boolean): Promise { - const p = await this.detail(org, pipelineId); await this.req(`/v2/${seg(org)}/pipelines/${seg(pipelineId)}`, { method: 'PATCH', - body: JSON.stringify({ - name: p.name, - description: p.description ?? '', - enabled, - nodes: (p.nodes ?? []).map((n: any) => ({ - id: n.id, - slug: n.slug, - component_id: n.component_id, - component_type: n.component_type, - enabled: n.enabled ?? true - })), - edges: (p.edges ?? []).map((e: any) => ({ - name: e.name, - description: e.description ?? '', - from_node_instance_id: e.from_node_instance_id, - to_node_instance_id: e.to_node_instance_id, - disabled: e.disabled ?? false, - conditions: e.conditions - })) - }) + body: JSON.stringify({ enabled }) }); }