From 6e88f9bf020a095115dbd367f84556a81cf77df7 Mon Sep 17 00:00:00 2001 From: curtis Date: Tue, 4 Aug 2026 11:56:20 -0400 Subject: [PATCH 1/5] 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 | 158 +++++++++++++++----------------- 2 files changed, 83 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..8d5edf0 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,34 @@ 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 +190,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 +218,58 @@ 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 }) }); } From 62ae4ec9426fe1205364df08366ff94caa3cb95e Mon Sep 17 00:00:00 2001 From: curtis Date: Tue, 4 Aug 2026 11:56:20 -0400 Subject: [PATCH 2/5] 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 0c8a42f347334da509a5e81c2658c8746e929592 Mon Sep 17 00:00:00 2001 From: curtis Date: Tue, 4 Aug 2026 12:27:10 -0400 Subject: [PATCH 3/5] fix(embed): correct the Go router against the real Monad API [PRO-456] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the review on #15. Each item was validated against the API source rather than the published spec, which is ~15 months 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`, and Monad defaults it to 10 — past a tenant's tenth pipeline the router reported "not connected" for pipelines that exist. Also turns an O(n) request fan-out into two calls. - `setEnabled` now PATCHes `{enabled}` alone. PATCH became a true partial update in api#2163; the read-modify-write it replaces rebuilt the whole graph from a partial field list, silently dropping node `config_overrides` and edge `schema_detection_spec` on every toggle. - drain the connector list rather than one `limit=1000` request - send `type` on output create; `output_type` is deprecated (api#2156) - drop the speculative `data` / `config` response-shape fallbacks — the list returns only `{pipelines, pagination}` and the detail is flat - `componentKind` type with an explicit path mapping, replacing `kind + "s"` - inline `url.PathEscape` instead of the one-letter `esc` alias - README: document `GetCustomerOrgID` as the tenant-isolation boundary --- routers/go/README.md | 36 ++++- routers/go/client.go | 317 ++++++++++++++++----------------------- routers/go/embed.go | 64 ++++---- routers/go/embed_test.go | 285 ++++++++++++++++++++++++++++++++++- 4 files changed, 473 insertions(+), 229 deletions(-) diff --git a/routers/go/README.md b/routers/go/README.md index 8da0b2e..e7fc8e9 100644 --- a/routers/go/README.md +++ b/routers/go/README.md @@ -69,8 +69,42 @@ r.Any("/embed/*any", gin.WrapH(http.StripPrefix("/embed", h))) | `APIKey` | Long-lived Monad API key (server-side only). | | `APIBase` | Monad API base. Optional — defaults to `https://app.monad.com/api` (production). | | `FrameOrigin` | Iframe origin returned by `GET /embed/config`. Optional — defaults to production. | -| `GetCustomerOrgID` | Map the authenticated request → the tenant's Monad team id. Return `("", nil)` or an error to reject (→ 401). | +| `GetCustomerOrgID` | **The tenant-isolation boundary** — see below. Return `("", nil)` or an error to reject (→ 401). | | `GetProvisionedComponents` | Per-tenant `Provision{ DestinationOutputID, SourceInputID }`. Nil → ingress uses dev/null, egress unavailable. | | `CatalogAllow` | Restrict the catalog to these connector type ids. Nil → all. | +### `GetCustomerOrgID` is a security boundary + +This callback is the only thing standing between one of your tenants and +another's data. The router holds a Monad API key with access to every +organization your key can reach; whatever org id this returns is the org the +request then reads, writes and deletes in. Get it wrong and you serve tenant A +the pipelines of tenant B. + +So derive the org **from your own authenticated session** — the thing your auth +middleware already verified: + +```go +GetCustomerOrgID: func(r *http.Request) (string, error) { + user, ok := auth.FromContext(r.Context()) // set by your auth middleware + if !ok { + return "", errors.New("not signed in") + } + return user.MonadOrgID, nil // your tenant → Monad org mapping +} +``` + +Never take it from anything the caller controls — a header, a query param, a +request body field, or a client-supplied JWT claim you have not verified. Those +are all attacker-chosen values, and the router will use them verbatim. + +Mount the router behind your auth middleware; it performs no authentication of +its own beyond calling this hook, and treats `("", nil)` or a non-nil error as +`401`. + +You own this mapping because only you can know it: Monad has no endpoint that +turns your product's bearer token into an org id. (`GET /v1/organizations` maps +a *Monad API key* to the orgs it can reach — not your user to their tenant.) +Store the tenant → org id association when you provision the tenant. + Zero external dependencies. `go test ./...`, `go vet ./...`, `gofmt` clean. diff --git a/routers/go/client.go b/routers/go/client.go index 4bd2266..1bdc7b0 100644 --- a/routers/go/client.go +++ b/routers/go/client.go @@ -32,9 +32,28 @@ func newClient(cfg Config) *client { } } -// esc URL-encodes a single path segment so browser-supplied ids can't inject -// query params or traverse the path (`/`, `?`, `#` are neutralised). -var esc = url.PathEscape +// componentKind is the contract's `kind` — the only two values any /embed route +// accepts. Mirrors the ComponentKind enum in embed.openapi.yaml. +type componentKind string + +const ( + kindInput componentKind = "input" + kindOutput componentKind = "output" +) + +// collection is the Monad path segment for a kind. Spelled out rather than +// appending "s", so the mapping is a fact rather than a coincidence of English. +func (k componentKind) collection() string { + if k == kindInput { + return "inputs" + } + return "outputs" +} + +// pageSize is how many rows to request when draining a Monad list. Monad +// defaults `limit` to 10 and enforces no maximum, so this is a round-trip / +// response-size trade-off only: correctness comes from draining every page. +const pageSize = 200 // upstreamError carries a failed Monad call's status + detail. The router maps // the status onto the contract's error model (404→not_found, 409→conflict, @@ -101,8 +120,8 @@ func (c *client) mintSession(ctx context.Context, org string) (Session, error) { return Session{SessionToken: out.SessionToken, OrganizationID: org, ExpiresAt: out.ExpiresAt}, nil } -func (c *client) listCatalog(ctx context.Context, kind string, allow []string) ([]CatalogType, error) { - data, err := c.do(ctx, "GET", "/v1/"+kind+"s", nil) +func (c *client) listCatalog(ctx context.Context, kind componentKind, allow []string) ([]CatalogType, error) { + data, err := c.do(ctx, "GET", "/v1/"+kind.collection(), nil) if err != nil { return nil, err } @@ -130,34 +149,57 @@ func (c *client) listCatalog(ctx context.Context, kind string, allow []string) ( return out, nil } -func (c *client) listConnectors(ctx context.Context, org, kind string) ([]ConfiguredConnector, error) { - data, err := c.do(ctx, "GET", "/v1/"+esc(org)+"/"+kind+"s?limit=1000&offset=0", nil) - if err != nil { - return nil, err - } - // Rows come wrapped as { inputs: [...] } / { outputs: [...] } alongside other - // keys (e.g. a `pagination` object). Decode only the connector array for this - // kind so sibling keys don't force a type mismatch; carry the type slug as - // `type`, normalized to typeId for the /embed contract. - var page map[string]json.RawMessage - if err := json.Unmarshal(data, &page); err != nil { - return nil, err - } - var rows []struct { - ID string `json:"id"` - Type string `json:"type"` - Name string `json:"name"` - } - if raw, ok := page[kind+"s"]; ok { - if err := json.Unmarshal(raw, &rows); err != nil { +// listConnectors returns every connector of a kind the tenant has configured. +// +// Monad pages this at limit=10 by default with no maximum, and the /embed +// contract returns a bare array with no pagination — so draining the pages is +// the router's job. A single large limit instead would silently truncate +// whichever tenant outgrows it. +func (c *client) listConnectors(ctx context.Context, org string, kind componentKind) ([]ConfiguredConnector, error) { + key := kind.collection() + out := []ConfiguredConnector{} + for offset := 0; ; offset += pageSize { + path := fmt.Sprintf("/v1/%s/%s?limit=%d&offset=%d", url.PathEscape(org), key, pageSize, offset) + data, err := c.do(ctx, "GET", path, nil) + if err != nil { return nil, err } + // Rows come wrapped as { inputs: [...] } / { outputs: [...] } alongside a + // `pagination` object. Decode only the connector array for this kind so + // the sibling can't force a type mismatch; the array is `null`, not `[]`, + // when the page is empty. + var page map[string]json.RawMessage + if err := json.Unmarshal(data, &page); err != nil { + return nil, err + } + var rows []struct { + ID string `json:"id"` + Type string `json:"type"` + Name string `json:"name"` + } + if raw, ok := page[key]; ok { + if err := json.Unmarshal(raw, &rows); err != nil { + return nil, err + } + } + for _, r := range rows { + // Monad names the type slug `type` here; the contract calls it typeId. + out = append(out, ConfiguredConnector{ID: r.ID, TypeID: r.Type, Name: r.Name}) + } + // A short page is the last page. `total` only lets us stop one round trip + // earlier when the count happens to divide evenly. + if len(rows) < pageSize { + return out, nil + } + var pg struct { + Total int `json:"total"` + } + if raw, ok := page["pagination"]; ok { + if err := json.Unmarshal(raw, &pg); err == nil && pg.Total > 0 && len(out) >= pg.Total { + return out, nil + } + } } - out := []ConfiguredConnector{} - for _, r := range rows { - out = append(out, ConfiguredConnector{ID: r.ID, TypeID: r.Type, Name: r.Name}) - } - return out, nil } type wireNode struct { @@ -185,7 +227,7 @@ func (c *client) wirePipeline(ctx context.Context, org, inputID, outputID, name "conditions": map[string]any{"operator": "always"}, }}, } - data, err := c.do(ctx, "POST", "/v2/"+esc(org)+"/pipelines/", body) + data, err := c.do(ctx, "POST", "/v2/"+url.PathEscape(org)+"/pipelines/", body) if err != nil { return BuiltPipeline{}, err } @@ -198,7 +240,7 @@ func (c *client) wirePipeline(ctx context.Context, org, inputID, outputID, name status := "Pending" for i := 0; i < c.pollAttempts; i++ { - sd, err := c.do(ctx, "GET", "/v2/"+esc(org)+"/pipelines/"+esc(created.ID)+"/status", nil) + sd, err := c.do(ctx, "GET", "/v2/"+url.PathEscape(org)+"/pipelines/"+url.PathEscape(created.ID)+"/status", nil) if err != nil { return BuiltPipeline{}, err } @@ -222,8 +264,9 @@ func (c *client) wirePipeline(ctx context.Context, org, inputID, outputID, name // buildDevNull creates a throwaway dev/null output, then wires the input to it. func (c *client) buildDevNull(ctx context.Context, org, inputID, name string) (BuiltPipeline, error) { - data, err := c.do(ctx, "POST", "/v2/"+esc(org)+"/outputs", map[string]any{ - "output_type": "dev-null", + data, err := c.do(ctx, "POST", "/v2/"+url.PathEscape(org)+"/outputs", map[string]any{ + // `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": "", @@ -241,186 +284,90 @@ func (c *client) buildDevNull(ctx context.Context, org, inputID, name string) (B return c.wirePipeline(ctx, org, inputID, out.ID, name) } -type pipeSummary struct { - ID string `json:"id"` -} - -type nodeFull struct { - ID string `json:"id"` - Slug string `json:"slug"` +type pipeNode struct { ComponentID string `json:"component_id"` ComponentType string `json:"component_type"` - Enabled *bool `json:"enabled"` -} - -type edgeFull struct { - Name string `json:"name"` - Description string `json:"description"` - From string `json:"from_node_instance_id"` - To string `json:"to_node_instance_id"` - Disabled *bool `json:"disabled"` - Conditions json.RawMessage `json:"conditions"` -} - -type pipeBody struct { - Name string `json:"name"` - Description string `json:"description"` - Enabled bool `json:"enabled"` - Nodes []nodeFull `json:"nodes"` - Edges []edgeFull `json:"edges"` -} - -// pipeDetail handles both the wrapped ({ config: {...} }) and flat detail shapes. -type pipeDetail struct { - Config *pipeBody `json:"config"` - pipeBody } -func (c *client) listPipelines(ctx context.Context, org string) ([]pipeSummary, error) { - data, err := c.do(ctx, "GET", "/v2/"+esc(org)+"/pipelines/", nil) +// pipelineNodes reads a pipeline's wiring. The detail response is flat — the +// nodes sit at the top level, not under a `config` envelope. +func (c *client) pipelineNodes(ctx context.Context, org, pipelineID string) ([]pipeNode, error) { + data, err := c.do(ctx, "GET", "/v2/"+url.PathEscape(org)+"/pipelines/"+url.PathEscape(pipelineID), nil) if err != nil { return nil, err } - var arr []pipeSummary - if json.Unmarshal(data, &arr) == nil && len(arr) > 0 { - return arr, nil + var detail struct { + Nodes []pipeNode `json:"nodes"` } - var obj struct { - Pipelines []pipeSummary `json:"pipelines"` - Data []pipeSummary `json:"data"` - } - if err := json.Unmarshal(data, &obj); err != nil { + if err := json.Unmarshal(data, &detail); err != nil { return nil, err } - if len(obj.Pipelines) > 0 { - return obj.Pipelines, nil - } - return obj.Data, nil + return detail.Nodes, nil } -func (c *client) getPipeline(ctx context.Context, org, id string) (*pipeBody, error) { - data, err := c.do(ctx, "GET", "/v2/"+esc(org)+"/pipelines/"+esc(id), nil) +// pipelineFor resolves 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. +// Returns an upstreamError with status 404 when the connector does not exist. +func (c *client) pipelineFor(ctx context.Context, org string, kind componentKind, connectorID string) (PipelineStatus, error) { + data, err := c.do(ctx, "GET", + "/v1/"+url.PathEscape(org)+"/"+kind.collection()+"/"+url.PathEscape(connectorID), nil) if err != nil { - return nil, err - } - var d pipeDetail - if err := json.Unmarshal(data, &d); err != nil { - return nil, err + return PipelineStatus{}, err } - if d.Config != nil { - return d.Config, nil + var connector struct { + ComponentOf []struct { + ID string `json:"id"` + Enabled bool `json:"enabled"` + } `json:"component_of"` } - b := d.pipeBody - return &b, nil -} - -// statusByInput resolves the pipeline an input feeds, knowing only the input id. -func (c *client) statusByInput(ctx context.Context, org, inputID string) (PipelineStatus, error) { - sums, err := c.listPipelines(ctx, org) - if err != nil { + if err := json.Unmarshal(data, &connector); err != nil { return PipelineStatus{}, err } - for _, s := range sums { - if s.ID == "" { - continue - } - body, err := c.getPipeline(ctx, org, s.ID) - if err != nil { - continue - } - var inNode, outNode *nodeFull - for i := range body.Nodes { - n := &body.Nodes[i] - if n.ComponentType == "input" && n.ComponentID == inputID { - inNode = n - } - if n.ComponentType == "output" { - outNode = n - } - } - if inNode == nil { - continue - } - ps := PipelineStatus{HasPipeline: true, Enabled: body.Enabled, PipelineID: s.ID} - if outNode != nil { - ps.OutputID = outNode.ComponentID - } - return ps, nil + if len(connector.ComponentOf) == 0 || connector.ComponentOf[0].ID == "" { + return PipelineStatus{HasPipeline: false}, nil } - return PipelineStatus{HasPipeline: false}, nil -} + p := connector.ComponentOf[0] + ps := PipelineStatus{HasPipeline: true, Enabled: p.Enabled, PipelineID: p.ID} -// findByOutput resolves the pipeline feeding an output — the egress counterpart. -func (c *client) findByOutput(ctx context.Context, org, outputID string) (PipelineStatus, error) { - sums, err := c.listPipelines(ctx, org) + peer := kindOutput + if kind == kindOutput { + peer = kindInput + } + nodes, err := c.pipelineNodes(ctx, org, p.ID) if err != nil { return PipelineStatus{}, err } - for _, s := range sums { - if s.ID == "" { + for _, n := range nodes { + if n.ComponentType != string(peer) { continue } - body, err := c.getPipeline(ctx, org, s.ID) - if err != nil { - continue - } - var inNode, outNode *nodeFull - for i := range body.Nodes { - n := &body.Nodes[i] - if n.ComponentType == "output" && n.ComponentID == outputID { - outNode = n - } - if n.ComponentType == "input" { - inNode = n - } + if kind == kindInput { + ps.OutputID = n.ComponentID + } else { + ps.InputID = n.ComponentID } - if outNode == nil { - continue - } - ps := PipelineStatus{HasPipeline: true, Enabled: body.Enabled, PipelineID: s.ID} - if inNode != nil { - ps.InputID = inNode.ComponentID - } - return ps, nil + break } - return PipelineStatus{HasPipeline: false}, nil + return ps, nil } -// setEnabled flips the pipeline's enabled flag. The PATCH endpoint replaces the -// whole config, so this reads the current pipeline and sends it back unchanged -// except for the flag. +// setEnabled flips the 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`. func (c *client) setEnabled(ctx context.Context, org, pipelineID string, enabled bool) error { - body, err := c.getPipeline(ctx, org, pipelineID) - if err != nil { - return err - } - nodes := make([]map[string]any, 0, len(body.Nodes)) - for _, n := range body.Nodes { - en := true - if n.Enabled != nil { - en = *n.Enabled - } - nodes = append(nodes, map[string]any{ - "id": n.ID, "slug": n.Slug, "component_id": n.ComponentID, - "component_type": n.ComponentType, "enabled": en, - }) - } - edges := make([]map[string]any, 0, len(body.Edges)) - for _, e := range body.Edges { - dis := false - if e.Disabled != nil { - dis = *e.Disabled - } - edges = append(edges, map[string]any{ - "name": e.Name, "description": e.Description, - "from_node_instance_id": e.From, "to_node_instance_id": e.To, - "disabled": dis, "conditions": e.Conditions, - }) - } - _, err = c.do(ctx, "PATCH", "/v2/"+esc(org)+"/pipelines/"+esc(pipelineID), map[string]any{ - "name": body.Name, "description": body.Description, "enabled": enabled, - "nodes": nodes, "edges": edges, - }) + _, err := c.do(ctx, "PATCH", "/v2/"+url.PathEscape(org)+"/pipelines/"+url.PathEscape(pipelineID), + map[string]any{"enabled": enabled}) return err } diff --git a/routers/go/embed.go b/routers/go/embed.go index 30adfa9..e3de74e 100644 --- a/routers/go/embed.go +++ b/routers/go/embed.go @@ -20,6 +20,7 @@ import ( "errors" "log" "net/http" + "net/url" ) // Provision is what the host pre-provisions per tenant, resolved server-side — @@ -260,13 +261,7 @@ func (h *handler) status(w http.ResponseWriter, r *http.Request) { if !ok { return } - var ps PipelineStatus - var err error - if kind == "input" { - ps, err = h.c.statusByInput(r.Context(), org, connectorID) - } else { - ps, err = h.c.findByOutput(r.Context(), org, connectorID) - } + ps, err := h.c.pipelineFor(r.Context(), org, kind, connectorID) if err != nil { upstream(w, err) return @@ -321,44 +316,36 @@ func (h *handler) remove(w http.ResponseWriter, r *http.Request) { } ctx := r.Context() - if kind == "input" { - ps, err := h.c.statusByInput(ctx, org, body.ConnectorID) - if err != nil { + ps, err := h.c.pipelineFor(ctx, org, kind, body.ConnectorID) + if err != nil { + upstream(w, err) + return + } + if ps.PipelineID != "" { + if err := h.c.del(ctx, "/v2/"+url.PathEscape(org)+"/pipelines/"+url.PathEscape(ps.PipelineID)); err != nil { upstream(w, err) return } - prov := h.prov(org) - keepStore := prov.DestinationOutputID != "" && ps.OutputID != "" && prov.DestinationOutputID == ps.OutputID - if ps.PipelineID != "" { - if err := h.c.del(ctx, "/v2/"+esc(org)+"/pipelines/"+esc(ps.PipelineID)); err != nil { - upstream(w, err) - return - } - } - if err := h.c.del(ctx, "/v1/"+esc(org)+"/inputs/"+esc(body.ConnectorID)); err != nil { + } + + if kind == kindInput { + if err := h.c.del(ctx, "/v1/"+url.PathEscape(org)+"/inputs/"+url.PathEscape(body.ConnectorID)); err != nil { upstream(w, err) return } + // The tenant's provisioned store is shared; only a sink this pipeline + // created on the fly gets torn down with it. + prov := h.prov(org) + keepStore := prov.DestinationOutputID != "" && prov.DestinationOutputID == ps.OutputID if ps.OutputID != "" && !keepStore { - if err := h.c.del(ctx, "/v1/"+esc(org)+"/outputs/"+esc(ps.OutputID)); err != nil { + if err := h.c.del(ctx, "/v1/"+url.PathEscape(org)+"/outputs/"+url.PathEscape(ps.OutputID)); err != nil { upstream(w, err) return } } } else { - ps, err := h.c.findByOutput(ctx, org, body.ConnectorID) - if err != nil { - upstream(w, err) - return - } - if ps.PipelineID != "" { - if err := h.c.del(ctx, "/v2/"+esc(org)+"/pipelines/"+esc(ps.PipelineID)); err != nil { - upstream(w, err) - return - } - } // Keep the shared source input; remove only the user's output. - if err := h.c.del(ctx, "/v1/"+esc(org)+"/outputs/"+esc(body.ConnectorID)); err != nil { + if err := h.c.del(ctx, "/v1/"+url.PathEscape(org)+"/outputs/"+url.PathEscape(body.ConnectorID)); err != nil { upstream(w, err) return } @@ -392,12 +379,15 @@ func (h *handler) prov(org string) Provision { return Provision{} } -func kindQuery(w http.ResponseWriter, value string) (string, bool) { - if value != "input" && value != "output" { - writeErr(w, http.StatusBadRequest, "invalid_request", "'kind' must be 'input' or 'output'.") - return "", false +// kindQuery is the single place a request's `kind` becomes a componentKind — +// anything else is rejected before it can reach a Monad path. +func kindQuery(w http.ResponseWriter, value string) (componentKind, bool) { + switch componentKind(value) { + case kindInput, kindOutput: + return componentKind(value), true } - return value, true + writeErr(w, http.StatusBadRequest, "invalid_request", "'kind' must be 'input' or 'output'.") + return "", false } func requireStr(w http.ResponseWriter, value, field string) bool { diff --git a/routers/go/embed_test.go b/routers/go/embed_test.go index 4c56454..63c8d47 100644 --- a/routers/go/embed_test.go +++ b/routers/go/embed_test.go @@ -2,35 +2,128 @@ package embed_test import ( "encoding/json" + "fmt" + "io" "net/http" "net/http/httptest" + "strconv" "strings" + "sync" "testing" embed "github.com/monad-inc/embed/routers/go" ) -// mockMonad stands in for the Monad API with just the endpoints these tests hit. -func mockMonad(t *testing.T) *httptest.Server { +// recorder captures what the router actually sent upstream, so a test can +// assert on the request rather than only on the response. +type recorder struct { + mu sync.Mutex + calls []string // "METHOD /path?query" + body map[string]string +} + +func (rec *recorder) record(r *http.Request) { + rec.mu.Lock() + defer rec.mu.Unlock() + call := r.Method + " " + r.URL.Path + rec.calls = append(rec.calls, call+"?"+r.URL.RawQuery) + if r.Body != nil { + b, _ := io.ReadAll(r.Body) + rec.body[call] = string(b) + } +} + +func (rec *recorder) count(prefix string) int { + rec.mu.Lock() + defer rec.mu.Unlock() + n := 0 + for _, c := range rec.calls { + if strings.HasPrefix(c, prefix) { + n++ + } + } + return n +} + +// connectorsPage renders one page of `total` configured inputs the way Monad +// does: the rows under the `s` key, a `pagination` sibling, and null +// rather than [] once the page is empty. +func connectorsPage(total int, q map[string][]string) string { + limit, _ := strconv.Atoi(first(q["limit"], "10")) + offset, _ := strconv.Atoi(first(q["offset"], "0")) + rows := []string{} + for i := offset; i < offset+limit && i < total; i++ { + rows = append(rows, fmt.Sprintf(`{"id":"cfg_%d","type":"aws-cloudtrail","name":"C%d"}`, i, i)) + } + list := "null" + if len(rows) > 0 { + list = "[" + strings.Join(rows, ",") + "]" + } + return fmt.Sprintf(`{"inputs":%s,"pagination":{"limit":%d,"offset":%d,"total":%d}}`, + list, limit, offset, total) +} + +func first(values []string, fallback string) string { + if len(values) == 0 { + return fallback + } + return values[0] +} + +// mockMonad stands in for the Monad API with just the endpoints these tests +// hit. `connectorTotal` sizes the configured-inputs list so paging is testable. +func mockMonad(t *testing.T, connectorTotal int) (*httptest.Server, *recorder) { t.Helper() - return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rec := &recorder{body: map[string]string{}} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rec.record(r) w.Header().Set("Content-Type", "application/json") switch { case r.Method == "POST" && r.URL.Path == "/v3/sessions": _, _ = w.Write([]byte(`{"session_token":"tok","expires_at":"2026-01-01T00:00:00Z"}`)) case r.Method == "GET" && r.URL.Path == "/v1/inputs": _, _ = w.Write([]byte(`[{"type_id":"aws-cloudtrail","name":"AWS"},{"type_id":"secret","name":"Hidden"}]`)) + case r.Method == "GET" && r.URL.Path == "/v1/org_1/inputs": + _, _ = w.Write([]byte(connectorsPage(connectorTotal, r.URL.Query()))) case r.Method == "POST" && r.URL.Path == "/v2/org_1/outputs": _, _ = w.Write([]byte(`{"id":"out_devnull"}`)) case r.Method == "POST" && r.URL.Path == "/v2/org_1/pipelines/": _, _ = w.Write([]byte(`{"id":"pipe_1"}`)) case r.Method == "GET" && r.URL.Path == "/v2/org_1/pipelines/pipe_1/status": _, _ = w.Write([]byte(`{"status":"Running"}`)) + + // A connector wired into pipe_1, and one wired into nothing. + case r.Method == "GET" && r.URL.Path == "/v1/org_1/inputs/in_wired": + _, _ = w.Write([]byte(`{"id":"in_wired","type":"aws-cloudtrail","name":"Wired", + "component_of":[{"id":"pipe_1","name":"p","enabled":true}]}`)) + case r.Method == "GET" && r.URL.Path == "/v1/org_1/outputs/out_wired": + _, _ = w.Write([]byte(`{"id":"out_wired","type":"s3","name":"Wired", + "component_of":[{"id":"pipe_1","name":"p","enabled":false}]}`)) + case r.Method == "GET" && r.URL.Path == "/v1/org_1/inputs/in_bare": + _, _ = w.Write([]byte(`{"id":"in_bare","type":"aws-cloudtrail","name":"Bare","component_of":[]}`)) + case r.Method == "GET" && r.URL.Path == "/v1/org_1/inputs/in_gone": + http.Error(w, `{"error":"not found"}`, http.StatusNotFound) + + // The pipeline detail: flat, nodes at the top level. + case r.Method == "GET" && r.URL.Path == "/v2/org_1/pipelines/pipe_1": + _, _ = w.Write([]byte(`{"id":"pipe_1","name":"p","enabled":true,"nodes":[ + {"id":"n1","slug":"in","component_id":"in_wired","component_type":"input", + "config_overrides":{"keep":"me"}}, + {"id":"n2","slug":"out","component_id":"out_wired","component_type":"output"}], + "edges":[{"from_node_instance_id":"in","to_node_instance_id":"out", + "schema_detection_spec":{"enabled":true}}]}`)) + case r.Method == "PATCH" && r.URL.Path == "/v2/org_1/pipelines/pipe_1": + _, _ = w.Write([]byte(`{"id":"pipe_1"}`)) + + case r.Method == "DELETE" && strings.HasPrefix(r.URL.Path, "/v1/org_1/"), + r.Method == "DELETE" && strings.HasPrefix(r.URL.Path, "/v2/org_1/"): + w.WriteHeader(http.StatusNoContent) default: t.Errorf("unexpected Monad call: %s %s", r.Method, r.URL.Path) http.Error(w, "unmocked", http.StatusInternalServerError) } })) + return srv, rec } func newHandler(t *testing.T, monadURL string, over func(*embed.Config)) http.Handler { @@ -86,7 +179,7 @@ func TestConfigDefaults(t *testing.T) { } func TestSession(t *testing.T) { - monad := mockMonad(t) + monad, _ := mockMonad(t, 0) defer monad.Close() rec := do(newHandler(t, monad.URL, nil), "POST", "/session", "") if rec.Code != 200 { @@ -100,7 +193,7 @@ func TestSession(t *testing.T) { } func TestCatalogAllowList(t *testing.T) { - monad := mockMonad(t) + monad, _ := mockMonad(t, 0) defer monad.Close() h := newHandler(t, monad.URL, func(c *embed.Config) { c.CatalogAllow = []string{"aws-cloudtrail"} }) rec := do(h, "GET", "/catalog?kind=input", "") @@ -115,7 +208,7 @@ func TestCatalogAllowList(t *testing.T) { } func TestIngressDevNull(t *testing.T) { - monad := mockMonad(t) + monad, _ := mockMonad(t, 0) defer monad.Close() rec := do(newHandler(t, monad.URL, nil), "POST", "/pipelines/ingress", `{"inputId":"in_1","name":"CT"}`) if rec.Code != 201 { @@ -157,3 +250,183 @@ func TestNotFound(t *testing.T) { t.Fatalf("status = %d", rec.Code) } } + +// A connector's pipeline is resolved from `component_of` plus one detail fetch +// for the peer — never by scanning the org's pipeline list, which pages at 10. +func TestPipelineStatusUsesComponentOf(t *testing.T) { + for _, tc := range []struct { + name, query string + wantPeerField string + wantPeer string + wantEnabled bool + }{ + {"input", "connectorId=in_wired&kind=input", "outputId", "out_wired", true}, + {"output", "connectorId=out_wired&kind=output", "inputId", "in_wired", false}, + } { + t.Run(tc.name, func(t *testing.T) { + monad, rec := mockMonad(t, 0) + defer monad.Close() + res := do(newHandler(t, monad.URL, nil), "GET", "/pipelines?"+tc.query, "") + if res.Code != 200 { + t.Fatalf("status = %d body=%s", res.Code, res.Body.String()) + } + var got map[string]any + _ = json.Unmarshal(res.Body.Bytes(), &got) + if got["hasPipeline"] != true || got["pipelineId"] != "pipe_1" { + t.Fatalf("status = %v", got) + } + if got["enabled"] != tc.wantEnabled { + t.Fatalf("enabled = %v, want %v", got["enabled"], tc.wantEnabled) + } + if got[tc.wantPeerField] != tc.wantPeer { + t.Fatalf("%s = %v, want %q", tc.wantPeerField, got[tc.wantPeerField], tc.wantPeer) + } + if n := rec.count("GET /v2/org_1/pipelines?"); n != 0 { + t.Fatalf("listed pipelines %d times; component_of makes the scan unnecessary", n) + } + }) + } +} + +func TestPipelineStatusWithoutPipeline(t *testing.T) { + monad, _ := mockMonad(t, 0) + defer monad.Close() + res := do(newHandler(t, monad.URL, nil), "GET", "/pipelines?connectorId=in_bare&kind=input", "") + if res.Code != 200 { + t.Fatalf("status = %d", res.Code) + } + var got map[string]any + _ = json.Unmarshal(res.Body.Bytes(), &got) + if got["hasPipeline"] != false || got["enabled"] != false { + t.Fatalf("status = %v", got) + } +} + +// A connector that does not exist is the contract's 404, not a generic 502. +func TestPipelineStatusUnknownConnector(t *testing.T) { + monad, _ := mockMonad(t, 0) + defer monad.Close() + res := do(newHandler(t, monad.URL, nil), "GET", "/pipelines?connectorId=in_gone&kind=input", "") + if res.Code != 404 { + t.Fatalf("status = %d body=%s", res.Code, res.Body.String()) + } + var got map[string]string + _ = json.Unmarshal(res.Body.Bytes(), &got) + if got["code"] != "not_found" { + t.Fatalf("code = %q", got["code"]) + } +} + +// PATCH is a true partial update, so the body carries the flag and nothing +// else. Sending a rebuilt graph would drop whatever fields the round trip +// failed to reproduce (config_overrides, schema_detection_spec, …). +func TestSetEnabledSendsOnlyTheFlag(t *testing.T) { + monad, rec := mockMonad(t, 0) + defer monad.Close() + res := do(newHandler(t, monad.URL, nil), "POST", "/pipelines/state", + `{"pipelineId":"pipe_1","enabled":false}`) + if res.Code != 204 { + t.Fatalf("status = %d body=%s", res.Code, res.Body.String()) + } + if got := rec.body["PATCH /v2/org_1/pipelines/pipe_1"]; got != `{"enabled":false}` { + t.Fatalf("PATCH body = %s, want only the enabled flag", got) + } + if n := rec.count("GET /v2/org_1/pipelines/pipe_1?"); n != 0 { + t.Fatalf("read the pipeline %d times; a partial PATCH needs no read-modify-write", n) + } +} + +// `type` is the canonical field on output create; `output_type` is deprecated. +func TestBuildDevNullSendsType(t *testing.T) { + monad, rec := mockMonad(t, 0) + defer monad.Close() + if res := do(newHandler(t, monad.URL, nil), "POST", "/pipelines/ingress", + `{"inputId":"in_1","name":"CT"}`); res.Code != 201 { + t.Fatalf("status = %d body=%s", res.Code, res.Body.String()) + } + var body map[string]any + _ = json.Unmarshal([]byte(rec.body["POST /v2/org_1/outputs"]), &body) + if body["type"] != "dev-null" { + t.Fatalf("output create sent %v, want type=dev-null", body) + } + if _, legacy := body["output_type"]; legacy { + t.Fatalf("output create still sends the deprecated output_type: %v", body) + } +} + +// Monad pages every list; the contract returns a bare array. The router has to +// drain the pages or it truncates the tenant's inventory. +func TestListConnectorsDrainsEveryPage(t *testing.T) { + const total = 450 // spans three pages at the client's page size + monad, rec := mockMonad(t, total) + defer monad.Close() + res := do(newHandler(t, monad.URL, nil), "GET", "/connectors?kind=input", "") + if res.Code != 200 { + t.Fatalf("status = %d body=%s", res.Code, res.Body.String()) + } + var got []map[string]string + _ = json.Unmarshal(res.Body.Bytes(), &got) + if len(got) != total { + t.Fatalf("got %d connectors, want %d — pages were not drained", len(got), total) + } + seen := map[string]bool{} + for _, c := range got { + if seen[c["id"]] { + t.Fatalf("duplicate connector %q across pages", c["id"]) + } + seen[c["id"]] = true + } + if n := rec.count("GET /v1/org_1/inputs?"); n < 2 { + t.Fatalf("made %d list calls for %d rows; expected several pages", n, total) + } +} + +// An empty list comes back as null, not [], and must still yield an empty +// array rather than a decode error or a JSON `null` body. +func TestListConnectorsHandlesNullPage(t *testing.T) { + monad, _ := mockMonad(t, 0) + defer monad.Close() + res := do(newHandler(t, monad.URL, nil), "GET", "/connectors?kind=input", "") + if res.Code != 200 { + t.Fatalf("status = %d", res.Code) + } + if body := strings.TrimSpace(res.Body.String()); body != "[]" { + t.Fatalf("body = %s, want []", body) + } +} + +// Removing an input tears down the pipeline and the input, and the throwaway +// sink with them — but never the tenant's provisioned store. +func TestRemoveKeepsProvisionedStore(t *testing.T) { + for _, tc := range []struct { + name string + provisioned string + wantDeleted bool + }{ + {"throwaway sink is removed", "", true}, + {"provisioned store is kept", "out_wired", false}, + } { + t.Run(tc.name, func(t *testing.T) { + monad, rec := mockMonad(t, 0) + defer monad.Close() + h := newHandler(t, monad.URL, func(c *embed.Config) { + c.GetProvisionedComponents = func(string) embed.Provision { + return embed.Provision{DestinationOutputID: tc.provisioned} + } + }) + if res := do(h, "POST", "/pipelines/remove", + `{"connectorId":"in_wired","kind":"input"}`); res.Code != 204 { + t.Fatalf("status = %d body=%s", res.Code, res.Body.String()) + } + if rec.count("DELETE /v2/org_1/pipelines/pipe_1?") != 1 { + t.Fatal("the pipeline must be deleted first — it references the connectors") + } + if rec.count("DELETE /v1/org_1/inputs/in_wired?") != 1 { + t.Fatal("the input was not deleted") + } + if got := rec.count("DELETE /v1/org_1/outputs/out_wired?") == 1; got != tc.wantDeleted { + t.Fatalf("output deleted = %v, want %v", got, tc.wantDeleted) + } + }) + } +} From dcfca8b888d91f5872c4cccb3582402379c73e26 Mon Sep 17 00:00:00 2001 From: curtis Date: Tue, 4 Aug 2026 12:32:12 -0400 Subject: [PATCH 4/5] fix(embed): correct the Python router against the real Monad API [PRO-457] Ports the fixes from the Go and TypeScript routers so all three stay contract-identical on the wire. - resolve a connector's pipeline via `component_of` instead of scanning every pipeline; the scan sent no `limit` and Monad defaults it to 10 - `set_enabled` PATCHes `{enabled}` alone now that PATCH is a true partial update, instead of rebuilding and replacing the node/edge graph - drain the connector list rather than one `limit=1000` request - send `type` on output create; `output_type` is deprecated - drop the speculative `data` / `config` response-shape fallbacks - map `kind` to its path segment explicitly rather than appending "s" --- routers/python/monad_embed/client.py | 187 +++++++++++++-------------- routers/python/monad_embed/router.py | 21 ++- 2 files changed, 97 insertions(+), 111 deletions(-) diff --git a/routers/python/monad_embed/client.py b/routers/python/monad_embed/client.py index ad82810..c58c2f4 100644 --- a/routers/python/monad_embed/client.py +++ b/routers/python/monad_embed/client.py @@ -25,6 +25,11 @@ _POLL_ATTEMPTS = 15 _POLL_INTERVAL = 2.0 +# Rows per request when draining a Monad list. Monad defaults ``limit`` to 10 +# and enforces no maximum, so this is a round-trip/response-size trade-off only: +# correctness comes from draining every page. +_PAGE_SIZE = 200 + def _seg(value: Any) -> str: """URL-encode a single path segment so browser-supplied ids can't inject @@ -32,6 +37,12 @@ def _seg(value: Any) -> str: return quote(str(value), safe="") +def _collection(kind: str) -> str: + """The contract's singular ``kind`` → Monad's collection path segment. + Spelled out rather than appending "s", so the mapping is a fact.""" + return "inputs" if kind == "input" else "outputs" + + class MonadClient: """Talks to the Monad API with the host's long-lived key.""" @@ -79,7 +90,7 @@ async def list_catalog( self, kind: str, allow: Optional[list[str]] ) -> list[CatalogType]: async with self._open() as c: - data = await self._do(c, "GET", f"/v1/{kind}s") + data = await self._do(c, "GET", f"/v1/{_collection(kind)}") allow_set = set(allow) if allow else None out: list[CatalogType] = [] for t in data or []: @@ -89,14 +100,36 @@ async def list_catalog( return out async def list_connectors(self, org: str, kind: str) -> list[ConfiguredConnector]: + """Every connector of a kind the tenant has configured. + + Monad pages this at ``limit=10`` by default with no maximum, and the + ``/embed`` contract returns a bare array with no pagination — so + draining the pages is the router's job. One large ``limit`` instead + would silently truncate whichever tenant outgrows it. + """ + key = _collection(kind) + out: list[ConfiguredConnector] = [] async with self._open() as c: - page = await self._do(c, "GET", f"/v1/{_seg(org)}/{kind}s?limit=1000&offset=0") - # Monad returns the list as null (not []) when a tenant has none, so - # coalesce with `or []` — a bare .get(key, []) would return that null. - rows = (page or {}).get(f"{kind}s") or [] - return [ - ConfiguredConnector(id=r["id"], typeId=r["type"], name=r["name"]) for r in rows - ] + offset = 0 + while True: + page = await self._do( + c, "GET", f"/v1/{_seg(org)}/{key}?limit={_PAGE_SIZE}&offset={offset}" + ) + # Monad returns the list as null (not []) for an empty page, so + # coalesce with `or []` — a bare .get(key, []) would return null. + rows = (page or {}).get(key) or [] + out.extend( + ConfiguredConnector(id=r["id"], typeId=r["type"], name=r["name"]) + for r in rows + ) + # A short page is the last page; `total` only lets us stop one + # round trip earlier when the count divides evenly. + if len(rows) < _PAGE_SIZE: + return out + total = ((page or {}).get("pagination") or {}).get("total") + if isinstance(total, int) and len(out) >= total: + return out + offset += _PAGE_SIZE async def wire_pipeline( self, org: str, input_id: str, output_id: str, name: str @@ -139,7 +172,8 @@ async def build_dev_null(self, org: str, input_id: str, name: str) -> BuiltPipel "POST", f"/v2/{_seg(org)}/outputs", { - "output_type": "dev-null", + # `type` is canonical; `output_type` is a deprecated alias. + "type": "dev-null", "name": f"{name} → /dev/null", "description": "Auto-created sink for embed pipeline", "promise_id": "", @@ -148,107 +182,62 @@ async def build_dev_null(self, org: str, input_id: str, name: str) -> BuiltPipel ) return await self.wire_pipeline(org, input_id, out["id"], name) - async def _list_pipelines(self, c: httpx.AsyncClient, org: str) -> list[dict]: - data = await self._do(c, "GET", f"/v2/{_seg(org)}/pipelines/") - if isinstance(data, list): - return data - if isinstance(data, dict): - return data.get("pipelines") or data.get("data") or [] - return [] - - async def _get_pipeline(self, c: httpx.AsyncClient, org: str, pid: str) -> dict: + async def _pipeline_nodes(self, c: httpx.AsyncClient, org: str, pid: str) -> list[dict]: + """A pipeline's wiring. The detail response is flat — the nodes sit at + the top level, not under a ``config`` envelope.""" data = await self._do(c, "GET", f"/v2/{_seg(org)}/pipelines/{_seg(pid)}") - if isinstance(data, dict): - return data.get("config") or data - return {} + return (data or {}).get("nodes") or [] - async def status_by_input(self, org: str, input_id: str) -> PipelineStatus: - async with self._open() as c: - for summary in await self._list_pipelines(c, org): - pid = summary.get("id") - if not pid: - continue - try: - p = await self._get_pipeline(c, org, pid) - except MonadError: - continue - nodes = p.get("nodes") or [] - in_node = next( - (n for n in nodes if n.get("component_type") == "input" and n.get("component_id") == input_id), - None, - ) - if in_node is None: - continue - out_node = next((n for n in nodes if n.get("component_type") == "output"), None) - return PipelineStatus( - hasPipeline=True, - enabled=bool(p.get("enabled")), - pipelineId=pid, - outputId=out_node.get("component_id") if out_node else None, - ) - return PipelineStatus(hasPipeline=False, enabled=False) + async def pipeline_for(self, org: str, kind: str, connector_id: str) -> PipelineStatus: + """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". - async def find_by_output(self, org: str, output_id: str) -> PipelineStatus: + ``component_of`` carries no wiring (the datastore fills only a summary + projection), so resolving the peer connector costs one further fetch. + Raises :class:`MonadError` with status 404 when the connector is gone. + """ async with self._open() as c: - for summary in await self._list_pipelines(c, org): - pid = summary.get("id") - if not pid: - continue - try: - p = await self._get_pipeline(c, org, pid) - except MonadError: - continue - nodes = p.get("nodes") or [] - out_node = next( - (n for n in nodes if n.get("component_type") == "output" and n.get("component_id") == output_id), - None, - ) - if out_node is None: - continue - in_node = next((n for n in nodes if n.get("component_type") == "input"), None) - return PipelineStatus( - hasPipeline=True, - enabled=bool(p.get("enabled")), - pipelineId=pid, - inputId=in_node.get("component_id") if in_node else None, - ) - return PipelineStatus(hasPipeline=False, enabled=False) + connector = await self._do( + c, "GET", f"/v1/{_seg(org)}/{_collection(kind)}/{_seg(connector_id)}" + ) + pipelines = (connector or {}).get("component_of") or [] + if not pipelines or not pipelines[0].get("id"): + return PipelineStatus(hasPipeline=False, enabled=False) + + pipeline = pipelines[0] + peer_type = "output" if kind == "input" else "input" + nodes = await self._pipeline_nodes(c, org, pipeline["id"]) + peer = next((n for n in nodes if n.get("component_type") == peer_type), None) + peer_id = peer.get("component_id") if peer else None + + return PipelineStatus( + hasPipeline=True, + enabled=bool(pipeline.get("enabled")), + pipelineId=pipeline["id"], + outputId=peer_id if kind == "input" else None, + inputId=peer_id if kind == "output" else None, + ) async def set_enabled(self, org: str, pipeline_id: str, enabled: bool) -> None: + """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 with self._open() as c: - p = await self._get_pipeline(c, org, pipeline_id) - nodes = [ - { - "id": n.get("id"), - "slug": n.get("slug"), - "component_id": n.get("component_id"), - "component_type": n.get("component_type"), - "enabled": n.get("enabled", True), - } - for n in (p.get("nodes") or []) - ] - edges = [ - { - "name": e.get("name"), - "description": e.get("description", ""), - "from_node_instance_id": e.get("from_node_instance_id"), - "to_node_instance_id": e.get("to_node_instance_id"), - "disabled": e.get("disabled", False), - "conditions": e.get("conditions"), - } - for e in (p.get("edges") or []) - ] await self._do( c, "PATCH", f"/v2/{_seg(org)}/pipelines/{_seg(pipeline_id)}", - { - "name": p.get("name"), - "description": p.get("description", ""), - "enabled": enabled, - "nodes": nodes, - "edges": edges, - }, + {"enabled": enabled}, ) async def delete(self, path: str) -> None: diff --git a/routers/python/monad_embed/router.py b/routers/python/monad_embed/router.py index 7265d32..e9684f7 100644 --- a/routers/python/monad_embed/router.py +++ b/routers/python/monad_embed/router.py @@ -164,10 +164,7 @@ async def get_pipeline_status(request: Request) -> Response: if not connector_id: raise EmbedError(400, "invalid_request", "Query 'connectorId' is required.") kind = _require_kind(request.query_params.get("kind")) - if kind == "input": - status = await client.status_by_input(org, connector_id) - else: - status = await client.find_by_output(org, connector_id) + status = await client.pipeline_for(org, kind, connector_id) return JSONResponse(status.model_dump(exclude_none=True)) async def set_pipeline_state(request: Request) -> Response: @@ -186,23 +183,23 @@ async def remove_integration(request: Request) -> Response: connector_id = _require_str(body, "connectorId") kind = _require_kind(body.get("kind") if isinstance(body, dict) else None) + status = await client.pipeline_for(org, kind, connector_id) + # The pipeline references its connectors, so it goes first. + if status.pipelineId: + await client.delete(f"/v2/{_seg(org)}/pipelines/{_seg(status.pipelineId)}") + if kind == "input": - status = await client.status_by_input(org, connector_id) + await client.delete(f"/v1/{_seg(org)}/inputs/{_seg(connector_id)}") + # The tenant's provisioned store is shared; only a sink this + # pipeline created on the fly gets torn down with it. prov = await _provision(config, org) keep_store = bool( prov.destination_output_id - and status.outputId and prov.destination_output_id == status.outputId ) - if status.pipelineId: - await client.delete(f"/v2/{_seg(org)}/pipelines/{_seg(status.pipelineId)}") - await client.delete(f"/v1/{_seg(org)}/inputs/{_seg(connector_id)}") if status.outputId and not keep_store: await client.delete(f"/v1/{_seg(org)}/outputs/{_seg(status.outputId)}") else: - status = await client.find_by_output(org, connector_id) - if status.pipelineId: - await client.delete(f"/v2/{_seg(org)}/pipelines/{_seg(status.pipelineId)}") # Keep the shared source input; remove only the user's output. await client.delete(f"/v1/{_seg(org)}/outputs/{_seg(connector_id)}") return Response(status_code=204) From 9dbe1af15ea5214925d3d8065a8638cbe14ed469 Mon Sep 17 00:00:00 2001 From: curtis Date: Tue, 4 Aug 2026 11:56:20 -0400 Subject: [PATCH 5/5] 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 }) }); }