From ac73c483098d31fb05eefeb977fe96213db138e2 Mon Sep 17 00:00:00 2001 From: clippy Date: Fri, 31 Jul 2026 14:11:26 -0700 Subject: [PATCH 1/7] feat(embed): add the Go router (net/http) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the Go port of the /embed router — embed.Router(embed.Config{...}) on the standard library net/http, zero third-party dependencies (module github.com/monad-inc/embed/routers/go). Same contract, idiomatic Go. Generalizes the conformance CI job to a ts + go matrix; the Go leg boots `go run ./cmd/conformance` and runs ROUTER=go against the same suite, proving contract-parity with the TypeScript router. Local: go vet + go test pass; conformance green for ROUTER=go and ROUTER=ts. Layer 4 (PRO-403), stacked on the TS router + conformance (PRO-402). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_011ASKA2VrufrNLGGrcFjY7n --- .github/workflows/ci.yml | 25 +- routers/go/README.md | 76 +++++ routers/go/client.go | 432 ++++++++++++++++++++++++++++ routers/go/cmd/conformance/main.go | 36 +++ routers/go/embed.go | 448 +++++++++++++++++++++++++++++ routers/go/embed_test.go | 159 ++++++++++ routers/go/go.mod | 3 + 7 files changed, 1173 insertions(+), 6 deletions(-) create mode 100644 routers/go/README.md create mode 100644 routers/go/client.go create mode 100644 routers/go/cmd/conformance/main.go create mode 100644 routers/go/embed.go create mode 100644 routers/go/embed_test.go create mode 100644 routers/go/go.mod diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 360c92d..7569514 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -46,22 +46,35 @@ jobs: - run: pnpm run build conformance: - name: Conformance (ts) + name: Conformance (${{ matrix.router }}) runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + router: [ts, go] steps: - uses: actions/checkout@v4 - uses: pnpm/action-setup@v4 + if: matrix.router == 'ts' - uses: actions/setup-node@v4 + if: matrix.router == 'ts' with: node-version: 22 cache: pnpm - - run: pnpm install --frozen-lockfile - - name: Build the TS router - run: pnpm -C routers/typescript build + if: matrix.router == 'ts' + run: | + pnpm install --frozen-lockfile + pnpm -C routers/typescript build + + - name: Set up Go + if: matrix.router == 'go' + uses: actions/setup-go@v5 + with: + go-version: 'stable' - uses: astral-sh/setup-uv@v5 @@ -71,8 +84,8 @@ jobs: uv venv --python 3.10 .venv uv pip install --python .venv/bin/python schemathesis pytest httpx jsonschema - - name: Run conformance against the TS router + - name: Run conformance (${{ matrix.router }}) working-directory: conformance env: - ROUTER: ts + ROUTER: ${{ matrix.router }} run: .venv/bin/python -m pytest -q diff --git a/routers/go/README.md b/routers/go/README.md new file mode 100644 index 0000000..8da0b2e --- /dev/null +++ b/routers/go/README.md @@ -0,0 +1,76 @@ +# Monad Embed — Go router + +A mountable `/embed` backend router for a Monad embed integration, on the +standard `net/http`. Implements the [`/embed` route contract](../../packages/embed/openapi/embed.openapi.yaml). + +```go +import embed "github.com/monad-inc/embed/routers/go" + +h := embed.Router(embed.Config{ + APIKey: os.Getenv("MONAD_API_KEY"), + // your auth → the tenant's Monad team id + GetCustomerOrgID: func(r *http.Request) (string, error) { return sessionOrg(r), nil }, + // server-side lookup of a tenant's pre-provisioned components (which + // destination an input feeds, etc.) + GetProvisionedComponents: func(org string) embed.Provision { + return embed.Provision{DestinationOutputID: stores[org]} + }, +}) + +mux.Handle("/embed/", http.StripPrefix("/embed", h)) +``` + +Mount under `/embed` (via `http.StripPrefix`) so the router sees `/session`, +`/pipelines/ingress`, etc. The browser holds only a session token; this router +holds the API key and is the seam to the Monad API. + +## Mounting on common frameworks + +`embed.Router(cfg)` is a standard `http.Handler` that expects root-relative +paths, so mount it under `/embed` and strip the prefix. Assume `h := embed.Router(cfg)`. + +**net/http** — standard library (Go 1.22+) + +```go +mux := http.NewServeMux() +mux.Handle("/embed/", http.StripPrefix("/embed", h)) +``` + +**chi** — [go-chi/chi](https://github.com/go-chi/chi) + +```go +r := chi.NewRouter() +r.Mount("/embed", http.StripPrefix("/embed", h)) +``` + +**gorilla/mux** — [gorilla/mux](https://github.com/gorilla/mux) + +```go +r := mux.NewRouter() +r.PathPrefix("/embed/").Handler(http.StripPrefix("/embed", h)) +``` + +**Echo** — [labstack/echo](https://github.com/labstack/echo) + +```go +e.Any("/embed/*", echo.WrapHandler(http.StripPrefix("/embed", h))) +``` + +**Gin** — [gin-gonic/gin](https://github.com/gin-gonic/gin) + +```go +r.Any("/embed/*any", gin.WrapH(http.StripPrefix("/embed", h))) +``` + +## Config + +| Field | Purpose | +| -------------------------- | -------------------------------------------------------------------------------------------------------------- | +| `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). | +| `GetProvisionedComponents` | Per-tenant `Provision{ DestinationOutputID, SourceInputID }`. Nil → ingress uses dev/null, egress unavailable. | +| `CatalogAllow` | Restrict the catalog to these connector type ids. Nil → all. | + +Zero external dependencies. `go test ./...`, `go vet ./...`, `gofmt` clean. diff --git a/routers/go/client.go b/routers/go/client.go new file mode 100644 index 0000000..4bd2266 --- /dev/null +++ b/routers/go/client.go @@ -0,0 +1,432 @@ +package embed + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "time" +) + +// client is the hand-written Monad API client — the Go equivalent of the TS +// kit. It sequences Monad's /v1 + /v2 + /v3 calls; the router layer (embed.go) +// turns its results into the /embed contract's responses. +type client struct { + apiKey string + apiBase string + hc *http.Client + pollAttempts int + pollInterval time.Duration +} + +func newClient(cfg Config) *client { + return &client{ + apiKey: cfg.APIKey, + apiBase: cfg.APIBase, + hc: &http.Client{Timeout: 30 * time.Second}, + pollAttempts: 15, + pollInterval: 2 * time.Second, + } +} + +// 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 + +// 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, +// otherwise 502 upstream_error). +type upstreamError struct { + status int + detail string +} + +func (e *upstreamError) Error() string { return e.detail } + +// do performs an authenticated request and returns the raw JSON body. A >=400 +// response becomes an error, which the router maps to 502 upstream_error. +func (c *client) do(ctx context.Context, method, path string, body any) ([]byte, error) { + var rdr io.Reader + if body != nil { + b, err := json.Marshal(body) + if err != nil { + return nil, err + } + rdr = bytes.NewReader(b) + } + req, err := http.NewRequestWithContext(ctx, method, c.apiBase+path, rdr) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "ApiKey "+c.apiKey) + req.Header.Set("Content-Type", "application/json") + + resp, err := c.hc.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + data, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + if resp.StatusCode >= 400 { + msg := string(data) + if len(msg) > 300 { + msg = msg[:300] + } + return nil, &upstreamError{status: resp.StatusCode, detail: fmt.Sprintf("%d %s: %s", resp.StatusCode, path, msg)} + } + return data, nil +} + +func (c *client) mintSession(ctx context.Context, org string) (Session, error) { + data, err := c.do(ctx, "POST", "/v3/sessions", map[string]any{ + "ttl_seconds": 1800, "organization_id": org, + }) + if err != nil { + return Session{}, err + } + var out struct { + SessionToken string `json:"session_token"` + ExpiresAt string `json:"expires_at"` + } + if err := json.Unmarshal(data, &out); err != nil { + return Session{}, err + } + 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) + if err != nil { + return nil, err + } + var raw []struct { + TypeID string `json:"type_id"` + Name string `json:"name"` + } + if err := json.Unmarshal(data, &raw); err != nil { + return nil, err + } + var allowSet map[string]bool + if len(allow) > 0 { + allowSet = make(map[string]bool, len(allow)) + for _, a := range allow { + allowSet[a] = true + } + } + out := []CatalogType{} + for _, t := range raw { + if allowSet != nil && !allowSet[t.TypeID] { + continue + } + out = append(out, CatalogType{TypeID: t.TypeID, Name: t.Name}) + } + 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 { + return nil, err + } + } + 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 { + Slug string `json:"slug"` + ComponentID string `json:"component_id"` + ComponentType string `json:"component_type"` + Enabled bool `json:"enabled"` +} + +// wirePipeline creates an input→output pipeline and polls until it reports +// Running (or Erroring). The shared primitive behind ingress and egress. +func (c *client) wirePipeline(ctx context.Context, org, inputID, outputID, name string) (BuiltPipeline, error) { + body := map[string]any{ + "name": name, + "description": "Created when the connector was configured via embed", + "enabled": true, + "nodes": []wireNode{ + {Slug: "in", ComponentID: inputID, ComponentType: "input", Enabled: true}, + {Slug: "out", ComponentID: outputID, ComponentType: "output", Enabled: true}, + }, + "edges": []map[string]any{{ + "from_node_instance_id": "in", + "to_node_instance_id": "out", + "description": "all records", + "conditions": map[string]any{"operator": "always"}, + }}, + } + data, err := c.do(ctx, "POST", "/v2/"+esc(org)+"/pipelines/", body) + if err != nil { + return BuiltPipeline{}, err + } + var created struct { + ID string `json:"id"` + } + if err := json.Unmarshal(data, &created); err != nil { + return BuiltPipeline{}, err + } + + status := "Pending" + for i := 0; i < c.pollAttempts; i++ { + sd, err := c.do(ctx, "GET", "/v2/"+esc(org)+"/pipelines/"+esc(created.ID)+"/status", nil) + if err != nil { + return BuiltPipeline{}, err + } + var s struct { + Status string `json:"status"` + } + if err := json.Unmarshal(sd, &s); err == nil && s.Status != "" { + status = s.Status + } + if status == "Running" || status == "Erroring" { + break + } + select { + case <-ctx.Done(): + return BuiltPipeline{}, ctx.Err() + case <-time.After(c.pollInterval): + } + } + return BuiltPipeline{PipelineID: created.ID, OutputID: outputID, Status: status, Active: status == "Running"}, nil +} + +// 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", + "name": name + " → /dev/null", + "description": "Auto-created sink for embed pipeline", + "promise_id": "", + "config": map[string]any{"settings": map[string]any{}, "secrets": map[string]any{}}, + }) + if err != nil { + return BuiltPipeline{}, err + } + var out struct { + ID string `json:"id"` + } + if err := json.Unmarshal(data, &out); err != nil { + return BuiltPipeline{}, err + } + 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"` + 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) + if err != nil { + return nil, err + } + var arr []pipeSummary + if json.Unmarshal(data, &arr) == nil && len(arr) > 0 { + return arr, nil + } + var obj struct { + Pipelines []pipeSummary `json:"pipelines"` + Data []pipeSummary `json:"data"` + } + if err := json.Unmarshal(data, &obj); err != nil { + return nil, err + } + if len(obj.Pipelines) > 0 { + return obj.Pipelines, nil + } + return obj.Data, 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) + if err != nil { + return nil, err + } + var d pipeDetail + if err := json.Unmarshal(data, &d); err != nil { + return nil, err + } + if d.Config != nil { + return d.Config, nil + } + 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 { + 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 + } + return PipelineStatus{HasPipeline: false}, nil +} + +// 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) + if 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 == "output" && n.ComponentID == outputID { + outNode = n + } + if n.ComponentType == "input" { + inNode = n + } + } + if outNode == nil { + continue + } + ps := PipelineStatus{HasPipeline: true, Enabled: body.Enabled, PipelineID: s.ID} + if inNode != nil { + ps.InputID = inNode.ComponentID + } + return ps, nil + } + return PipelineStatus{HasPipeline: false}, 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. +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, + }) + return err +} + +// del removes a resource. Order matters at the call site: the pipeline +// references its input/output, so it must be deleted first. +func (c *client) del(ctx context.Context, path string) error { + _, err := c.do(ctx, "DELETE", path, nil) + return err +} diff --git a/routers/go/cmd/conformance/main.go b/routers/go/cmd/conformance/main.go new file mode 100644 index 0000000..892a63c --- /dev/null +++ b/routers/go/cmd/conformance/main.go @@ -0,0 +1,36 @@ +// Command conformance boots the Go /embed router for the conformance harness, +// pointed at the mock Monad API via MONAD_API_BASE and listening on PORT. +package main + +import ( + "net/http" + "os" + + embed "github.com/monad-inc/embed/routers/go" +) + +func main() { + cfg := embed.Config{ + APIKey: "conf-key", + APIBase: os.Getenv("MONAD_API_BASE"), + FrameOrigin: "https://app.monad.com/embed", + GetCustomerOrgID: func(r *http.Request) (string, error) { + return "org_conf", nil + }, + GetProvisionedComponents: func(org string) embed.Provision { + return embed.Provision{DestinationOutputID: "out_store", SourceInputID: "in_source"} + }, + CatalogAllow: []string{"aws-cloudtrail", "okta-systemlog"}, + } + + mux := http.NewServeMux() + mux.Handle("/embed/", http.StripPrefix("/embed", embed.Router(cfg))) + + port := os.Getenv("PORT") + if port == "" { + port = "8791" + } + if err := http.ListenAndServe("127.0.0.1:"+port, mux); err != nil { + panic(err) + } +} diff --git a/routers/go/embed.go b/routers/go/embed.go new file mode 100644 index 0000000..30adfa9 --- /dev/null +++ b/routers/go/embed.go @@ -0,0 +1,448 @@ +// Package embed is a mountable /embed backend router for a Monad embed +// integration. It implements the /embed route contract +// (packages/embed/openapi/embed.openapi.yaml) on the standard net/http, so it +// mounts under any router: +// +// h := embed.Router(embed.Config{ +// APIKey: os.Getenv("MONAD_API_KEY"), +// APIBase: "https://app.monad.com/api", +// FrameOrigin: "https://app.monad.com/embed", +// GetCustomerOrgID: func(r *http.Request) (string, error) { return sessionOrg(r), nil }, +// }) +// mux.Handle("/embed/", http.StripPrefix("/embed", h)) +// +// The browser holds only a session token; this router (holding the API key) is +// the seam to the Monad API. +package embed + +import ( + "encoding/json" + "errors" + "log" + "net/http" +) + +// Provision is what the host pre-provisions per tenant, resolved server-side — +// never sent by the browser. +type Provision struct { + // DestinationOutputID is the ingress target. Empty → ingress goes to a dev/null sink. + DestinationOutputID string + // SourceInputID is the egress source. Empty → egress is unavailable. + SourceInputID string +} + +// Config is the uniform router configuration shared across every language. +type Config struct { + // APIKey is the long-lived Monad API key. Server-side only. + APIKey string + // APIBase is the Monad API base. Defaults to "https://app.monad.com/api" + // (production); set only for non-prod. + APIBase string + // FrameOrigin is returned by GET /embed/config. Defaults to + // "https://app.monad.com/embed" (production); set only for non-prod. + FrameOrigin string + // GetCustomerOrgID maps the authenticated request to the caller's Monad team id. + // The one seam only the host can fill. Return a non-nil error (or "") to + // reject an unauthenticated caller — the router responds 401. + GetCustomerOrgID func(r *http.Request) (string, error) + // GetProvisionedComponents returns a tenant's pre-provisioned resources. Nil → + // ingress uses dev/null and egress is unavailable. + GetProvisionedComponents func(org string) Provision + // CatalogAllow restricts the catalog to these connector type ids. Nil → all. + CatalogAllow []string +} + +// ---- response shapes (camelCase per the /embed contract) ---- + +type configResponse struct { + FrameOrigin string `json:"frameOrigin"` + APIBase string `json:"apiBase"` +} + +// Session is a minted embed session, shaped to drop straight into the iframe mount. +type Session struct { + SessionToken string `json:"sessionToken"` + OrganizationID string `json:"organizationId"` + ExpiresAt string `json:"expiresAt"` +} + +// CatalogType is a connector type available to embed. +type CatalogType struct { + TypeID string `json:"typeId"` + Name string `json:"name"` +} + +// ConfiguredConnector is a connector a tenant has already configured. +type ConfiguredConnector struct { + ID string `json:"id"` + TypeID string `json:"typeId"` + Name string `json:"name"` +} + +// BuiltPipeline is the result of standing up a pipeline. +type BuiltPipeline struct { + PipelineID string `json:"pipelineId"` + OutputID string `json:"outputId"` + Status string `json:"status"` + Active bool `json:"active"` +} + +// PipelineStatus is a configured connector's resolved pipeline state. +type PipelineStatus struct { + HasPipeline bool `json:"hasPipeline"` + Enabled bool `json:"enabled"` + PipelineID string `json:"pipelineId,omitempty"` + InputID string `json:"inputId,omitempty"` + OutputID string `json:"outputId,omitempty"` +} + +type errorBody struct { + Code string `json:"code"` + Message string `json:"message"` +} + +// ---- router ---- + +type handler struct { + cfg Config + c *client +} + +// Router builds the /embed router. Mount it under "/embed" (e.g. via +// http.StripPrefix) so it sees paths like "/session" and "/pipelines/ingress". +func Router(cfg Config) http.Handler { + if cfg.GetCustomerOrgID == nil { + panic("embed: Config.GetCustomerOrgID is required") + } + // Production defaults — set APIBase/FrameOrigin only for non-prod. + if cfg.APIBase == "" { + cfg.APIBase = "https://app.monad.com/api" + } + if cfg.FrameOrigin == "" { + cfg.FrameOrigin = "https://app.monad.com/embed" + } + h := &handler{cfg: cfg, c: newClient(cfg)} + mux := http.NewServeMux() + mux.HandleFunc("GET /config", h.getConfig) + mux.HandleFunc("POST /session", h.session) + mux.HandleFunc("GET /catalog", h.catalog) + mux.HandleFunc("GET /connectors", h.connectors) + mux.HandleFunc("POST /pipelines/ingress", h.ingress) + mux.HandleFunc("POST /pipelines/egress", h.egress) + mux.HandleFunc("GET /pipelines", h.status) + mux.HandleFunc("POST /pipelines/state", h.state) + mux.HandleFunc("POST /pipelines/remove", h.remove) + mux.HandleFunc("/", h.notFound) + return mux +} + +// ---- handlers ---- + +func (h *handler) getConfig(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, configResponse{FrameOrigin: h.cfg.FrameOrigin, APIBase: h.cfg.APIBase}) +} + +func (h *handler) session(w http.ResponseWriter, r *http.Request) { + org, ok := h.tenant(w, r) + if !ok { + return + } + s, err := h.c.mintSession(r.Context(), org) + if err != nil { + upstream(w, err) + return + } + writeJSON(w, http.StatusOK, s) +} + +func (h *handler) catalog(w http.ResponseWriter, r *http.Request) { + if _, ok := h.tenant(w, r); !ok { + return + } + kind, ok := kindQuery(w, r.URL.Query().Get("kind")) + if !ok { + return + } + types, err := h.c.listCatalog(r.Context(), kind, h.cfg.CatalogAllow) + if err != nil { + upstream(w, err) + return + } + writeJSON(w, http.StatusOK, types) +} + +func (h *handler) connectors(w http.ResponseWriter, r *http.Request) { + org, ok := h.tenant(w, r) + if !ok { + return + } + kind, ok := kindQuery(w, r.URL.Query().Get("kind")) + if !ok { + return + } + rows, err := h.c.listConnectors(r.Context(), org, kind) + if err != nil { + upstream(w, err) + return + } + writeJSON(w, http.StatusOK, rows) +} + +func (h *handler) ingress(w http.ResponseWriter, r *http.Request) { + org, ok := h.tenant(w, r) + if !ok { + return + } + var body struct { + InputID string `json:"inputId"` + Name string `json:"name"` + } + if !decode(w, r, &body) { + return + } + if !requireStr(w, body.InputID, "inputId") || !requireStr(w, body.Name, "name") { + return + } + prov := h.prov(org) + var built BuiltPipeline + var err error + if prov.DestinationOutputID != "" { + built, err = h.c.wirePipeline(r.Context(), org, body.InputID, prov.DestinationOutputID, body.Name) + } else { + built, err = h.c.buildDevNull(r.Context(), org, body.InputID, body.Name) + } + if err != nil { + upstream(w, err) + return + } + writeJSON(w, http.StatusCreated, built) +} + +func (h *handler) egress(w http.ResponseWriter, r *http.Request) { + org, ok := h.tenant(w, r) + if !ok { + return + } + var body struct { + OutputID string `json:"outputId"` + Name string `json:"name"` + } + if !decode(w, r, &body) { + return + } + if !requireStr(w, body.OutputID, "outputId") || !requireStr(w, body.Name, "name") { + return + } + prov := h.prov(org) + if prov.SourceInputID == "" { + writeErr(w, http.StatusInternalServerError, "internal_error", + "No source input is provisioned for this tenant; egress cannot be built.") + return + } + built, err := h.c.wirePipeline(r.Context(), org, prov.SourceInputID, body.OutputID, body.Name) + if err != nil { + upstream(w, err) + return + } + writeJSON(w, http.StatusCreated, built) +} + +func (h *handler) status(w http.ResponseWriter, r *http.Request) { + org, ok := h.tenant(w, r) + if !ok { + return + } + connectorID := r.URL.Query().Get("connectorId") + if !requireStr(w, connectorID, "connectorId") { + return + } + kind, ok := kindQuery(w, r.URL.Query().Get("kind")) + 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) + } + if err != nil { + upstream(w, err) + return + } + writeJSON(w, http.StatusOK, ps) +} + +func (h *handler) state(w http.ResponseWriter, r *http.Request) { + org, ok := h.tenant(w, r) + if !ok { + return + } + var body struct { + PipelineID string `json:"pipelineId"` + Enabled *bool `json:"enabled"` + } + if !decode(w, r, &body) { + return + } + if !requireStr(w, body.PipelineID, "pipelineId") { + return + } + if body.Enabled == nil { + writeErr(w, http.StatusBadRequest, "invalid_request", "Field 'enabled' must be a boolean.") + return + } + if err := h.c.setEnabled(r.Context(), org, body.PipelineID, *body.Enabled); err != nil { + upstream(w, err) + return + } + w.WriteHeader(http.StatusNoContent) +} + +func (h *handler) remove(w http.ResponseWriter, r *http.Request) { + org, ok := h.tenant(w, r) + if !ok { + return + } + var body struct { + ConnectorID string `json:"connectorId"` + Kind string `json:"kind"` + } + if !decode(w, r, &body) { + return + } + if !requireStr(w, body.ConnectorID, "connectorId") { + return + } + kind, ok := kindQuery(w, body.Kind) + if !ok { + return + } + ctx := r.Context() + + if kind == "input" { + ps, err := h.c.statusByInput(ctx, org, body.ConnectorID) + if 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 { + upstream(w, err) + return + } + if ps.OutputID != "" && !keepStore { + if err := h.c.del(ctx, "/v1/"+esc(org)+"/outputs/"+esc(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 { + upstream(w, err) + return + } + } + w.WriteHeader(http.StatusNoContent) +} + +func (h *handler) notFound(w http.ResponseWriter, r *http.Request) { + writeErr(w, http.StatusNotFound, "not_found", "No route for "+r.Method+" "+r.URL.Path+".") +} + +// ---- helpers ---- + +func (h *handler) tenant(w http.ResponseWriter, r *http.Request) (string, bool) { + org, err := h.cfg.GetCustomerOrgID(r) + if err != nil || org == "" { + msg := "Could not resolve a tenant for the request." + if err != nil { + msg = err.Error() + } + writeErr(w, http.StatusUnauthorized, "unauthenticated", msg) + return "", false + } + return org, true +} + +func (h *handler) prov(org string) Provision { + if h.cfg.GetProvisionedComponents != nil { + return h.cfg.GetProvisionedComponents(org) + } + 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 + } + return value, true +} + +func requireStr(w http.ResponseWriter, value, field string) bool { + if value == "" { + writeErr(w, http.StatusBadRequest, "invalid_request", "Field '"+field+"' is required.") + return false + } + return true +} + +func decode(w http.ResponseWriter, r *http.Request, v any) bool { + if err := json.NewDecoder(r.Body).Decode(v); err != nil { + writeErr(w, http.StatusBadRequest, "invalid_request", "Request body must be valid JSON.") + return false + } + return true +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} + +func writeErr(w http.ResponseWriter, status int, code, message string) { + writeJSON(w, status, errorBody{Code: code, Message: message}) +} + +// upstream maps a failed Monad call to the contract's 502 upstream_error. +func upstream(w http.ResponseWriter, err error) { + // Translate Monad's own status codes into the contract's error model. + var ue *upstreamError + if errors.As(err, &ue) { + switch ue.status { + case http.StatusNotFound: + writeErr(w, http.StatusNotFound, "not_found", + "The referenced connector or pipeline does not exist.") + return + case http.StatusConflict: + writeErr(w, http.StatusConflict, "conflict", + "The request conflicts with existing state.") + return + } + } + // Anything else: log the detail server-side, return a generic 502. + log.Printf("[embed] upstream Monad API call failed: %v", err) + writeErr(w, http.StatusBadGateway, "upstream_error", "The upstream Monad API request failed.") +} diff --git a/routers/go/embed_test.go b/routers/go/embed_test.go new file mode 100644 index 0000000..4c56454 --- /dev/null +++ b/routers/go/embed_test.go @@ -0,0 +1,159 @@ +package embed_test + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "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 { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + 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 == "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"}`)) + default: + t.Errorf("unexpected Monad call: %s %s", r.Method, r.URL.Path) + http.Error(w, "unmocked", http.StatusInternalServerError) + } + })) +} + +func newHandler(t *testing.T, monadURL string, over func(*embed.Config)) http.Handler { + cfg := embed.Config{ + APIKey: "k", + APIBase: monadURL, + FrameOrigin: "https://app.monad.com/embed", + GetCustomerOrgID: func(r *http.Request) (string, error) { return "org_1", nil }, + } + if over != nil { + over(&cfg) + } + return embed.Router(cfg) +} + +func do(h http.Handler, method, path, body string) *httptest.ResponseRecorder { + var r *http.Request + if body != "" { + r = httptest.NewRequest(method, path, strings.NewReader(body)) + } else { + r = httptest.NewRequest(method, path, nil) + } + rec := httptest.NewRecorder() + h.ServeHTTP(rec, r) + return rec +} + +func TestGetConfig(t *testing.T) { + h := newHandler(t, "http://unused", nil) + rec := do(h, "GET", "/config", "") + if rec.Code != 200 { + t.Fatalf("status = %d", rec.Code) + } + var got map[string]string + _ = json.Unmarshal(rec.Body.Bytes(), &got) + if got["frameOrigin"] != "https://app.monad.com/embed" { + t.Fatalf("frameOrigin = %q", got["frameOrigin"]) + } +} + +func TestConfigDefaults(t *testing.T) { + // APIBase + FrameOrigin omitted → production defaults. + h := embed.Router(embed.Config{ + APIKey: "k", + GetCustomerOrgID: func(r *http.Request) (string, error) { return "org_1", nil }, + }) + rec := do(h, "GET", "/config", "") + var got map[string]string + _ = json.Unmarshal(rec.Body.Bytes(), &got) + if got["apiBase"] != "https://app.monad.com/api" || got["frameOrigin"] != "https://app.monad.com/embed" { + t.Fatalf("defaults not applied: %v", got) + } +} + +func TestSession(t *testing.T) { + monad := mockMonad(t) + defer monad.Close() + rec := do(newHandler(t, monad.URL, nil), "POST", "/session", "") + if rec.Code != 200 { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + var got map[string]string + _ = json.Unmarshal(rec.Body.Bytes(), &got) + if got["sessionToken"] != "tok" || got["organizationId"] != "org_1" { + t.Fatalf("session = %v", got) + } +} + +func TestCatalogAllowList(t *testing.T) { + monad := mockMonad(t) + defer monad.Close() + h := newHandler(t, monad.URL, func(c *embed.Config) { c.CatalogAllow = []string{"aws-cloudtrail"} }) + rec := do(h, "GET", "/catalog?kind=input", "") + if rec.Code != 200 { + t.Fatalf("status = %d", rec.Code) + } + var got []map[string]string + _ = json.Unmarshal(rec.Body.Bytes(), &got) + if len(got) != 1 || got[0]["typeId"] != "aws-cloudtrail" { + t.Fatalf("catalog = %v", got) + } +} + +func TestIngressDevNull(t *testing.T) { + monad := mockMonad(t) + defer monad.Close() + rec := do(newHandler(t, monad.URL, nil), "POST", "/pipelines/ingress", `{"inputId":"in_1","name":"CT"}`) + if rec.Code != 201 { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + var got map[string]any + _ = json.Unmarshal(rec.Body.Bytes(), &got) + if got["outputId"] != "out_devnull" || got["active"] != true { + t.Fatalf("built = %v", got) + } +} + +func TestUnauthenticated(t *testing.T) { + h := newHandler(t, "http://unused", func(c *embed.Config) { + c.GetCustomerOrgID = func(r *http.Request) (string, error) { return "", nil } + }) + rec := do(h, "POST", "/session", "") + if rec.Code != 401 { + t.Fatalf("status = %d", rec.Code) + } + var got map[string]string + _ = json.Unmarshal(rec.Body.Bytes(), &got) + if got["code"] != "unauthenticated" { + t.Fatalf("code = %q", got["code"]) + } +} + +func TestInvalidKind(t *testing.T) { + h := newHandler(t, "http://unused", nil) + rec := do(h, "GET", "/catalog?kind=nope", "") + if rec.Code != 400 { + t.Fatalf("status = %d", rec.Code) + } +} + +func TestNotFound(t *testing.T) { + rec := do(newHandler(t, "http://unused", nil), "GET", "/nope", "") + if rec.Code != 404 { + t.Fatalf("status = %d", rec.Code) + } +} diff --git a/routers/go/go.mod b/routers/go/go.mod new file mode 100644 index 0000000..ebd5da2 --- /dev/null +++ b/routers/go/go.mod @@ -0,0 +1,3 @@ +module github.com/monad-inc/embed/routers/go + +go 1.23 From 0e66ed710c65e83335c7bd5cfed2509516ca9178 Mon Sep 17 00:00:00 2001 From: clippy Date: Fri, 31 Jul 2026 15:42:33 -0700 Subject: [PATCH 2/7] test(conformance): drive the Go router in live mode Parameterizes the Go conformance server (cmd/conformance) from env with the same knobs as the TS/Python servers, defaulting to the mock fixtures. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_011ASKA2VrufrNLGGrcFjY7n --- routers/go/cmd/conformance/main.go | 42 +++++++++++++++++++++++++----- 1 file changed, 35 insertions(+), 7 deletions(-) diff --git a/routers/go/cmd/conformance/main.go b/routers/go/cmd/conformance/main.go index 892a63c..309c9c1 100644 --- a/routers/go/cmd/conformance/main.go +++ b/routers/go/cmd/conformance/main.go @@ -1,26 +1,54 @@ -// Command conformance boots the Go /embed router for the conformance harness, -// pointed at the mock Monad API via MONAD_API_BASE and listening on PORT. +// Command conformance boots the Go /embed router for the conformance harness. +// +// In the default (mock) mode the harness points it at the in-memory mock via +// MONAD_API_BASE; in live mode the same knobs are set to real Monad staging +// values. Every value falls back to the mock fixture default, so the existing +// hermetic run is unchanged. package main import ( "net/http" "os" + "strings" embed "github.com/monad-inc/embed/routers/go" ) +// envOr returns the env var value, or fallback when unset. An explicitly-empty +// value ("") is returned as-is — a live tenant may have no provisioned store. +func envOr(name, fallback string) string { + if v, ok := os.LookupEnv(name); ok { + return v + } + return fallback +} + +// splitAllow parses a comma-separated allow-list; empty → nil (expose all). +func splitAllow(s string) []string { + var out []string + for _, part := range strings.Split(s, ",") { + if p := strings.TrimSpace(part); p != "" { + out = append(out, p) + } + } + return out +} + func main() { + store := envOr("MONAD_STORE_ID", "out_store") + source := envOr("MONAD_SOURCE_ID", "in_source") + cfg := embed.Config{ - APIKey: "conf-key", + APIKey: envOr("MONAD_API_KEY", "conf-key"), APIBase: os.Getenv("MONAD_API_BASE"), - FrameOrigin: "https://app.monad.com/embed", + FrameOrigin: envOr("MONAD_FRAME_ORIGIN", "https://app.monad.com/embed"), GetCustomerOrgID: func(r *http.Request) (string, error) { - return "org_conf", nil + return envOr("MONAD_ORG_ID", "org_conf"), nil }, GetProvisionedComponents: func(org string) embed.Provision { - return embed.Provision{DestinationOutputID: "out_store", SourceInputID: "in_source"} + return embed.Provision{DestinationOutputID: store, SourceInputID: source} }, - CatalogAllow: []string{"aws-cloudtrail", "okta-systemlog"}, + CatalogAllow: splitAllow(envOr("MONAD_CATALOG_ALLOW", "aws-cloudtrail,okta-systemlog")), } mux := http.NewServeMux() From 6e88f9bf020a095115dbd367f84556a81cf77df7 Mon Sep 17 00:00:00 2001 From: curtis Date: Tue, 4 Aug 2026 11:56:20 -0400 Subject: [PATCH 3/7] 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 0c8a42f347334da509a5e81c2658c8746e929592 Mon Sep 17 00:00:00 2001 From: curtis Date: Tue, 4 Aug 2026 12:27:10 -0400 Subject: [PATCH 4/7] 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 a2921a0c29d8583e93b31fd873d991e966adf9a1 Mon Sep 17 00:00:00 2001 From: clippy Date: Tue, 8 Sep 2026 13:47:40 -0700 Subject: [PATCH 5/7] fix(embed): harden Go router pagination + body limit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - listConnectors: drain until a short page; drop the pagination.total early-break (truncates when Monad under-reports total); add maxPages cap against an upstream that never returns a short page. - decode: wrap the request body in http.MaxBytesReader (1 MiB) → 400 invalid_request when exceeded, guarding the shared process from an oversized-body OOM. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_011ASKA2VrufrNLGGrcFjY7n --- routers/go/client.go | 25 ++++++++++++++----------- routers/go/embed.go | 8 +++++++- 2 files changed, 21 insertions(+), 12 deletions(-) diff --git a/routers/go/client.go b/routers/go/client.go index 1bdc7b0..9c802f4 100644 --- a/routers/go/client.go +++ b/routers/go/client.go @@ -55,6 +55,11 @@ func (k componentKind) collection() string { // response-size trade-off only: correctness comes from draining every page. const pageSize = 200 +// maxPages bounds draining a single list, so an upstream that never returns a +// short page (e.g. offset-clamping) can't spin forever / exhaust memory. +// maxPages * pageSize = 200k rows — far beyond any real tenant. +const maxPages = 1000 + // 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, // otherwise 502 upstream_error). @@ -158,7 +163,14 @@ func (c *client) listCatalog(ctx context.Context, kind componentKind, allow []st func (c *client) listConnectors(ctx context.Context, org string, kind componentKind) ([]ConfiguredConnector, error) { key := kind.collection() out := []ConfiguredConnector{} - for offset := 0; ; offset += pageSize { + // A short page is the ONLY safe stop: trusting pagination.total to break + // early truncates the list when Monad under-reports it. maxPages bounds a + // pathological upstream that never returns a short page. + for pageIndex := 0; ; pageIndex++ { + if pageIndex >= maxPages { + return nil, fmt.Errorf("pagination exceeded %d pages draining %s", maxPages, key) + } + offset := pageIndex * 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 { @@ -186,19 +198,10 @@ func (c *client) listConnectors(ctx context.Context, org string, kind componentK // 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. + // A short page is the last page. 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 - } - } } } diff --git a/routers/go/embed.go b/routers/go/embed.go index e3de74e..d2c4f7a 100644 --- a/routers/go/embed.go +++ b/routers/go/embed.go @@ -398,9 +398,15 @@ func requireStr(w http.ResponseWriter, value, field string) bool { return true } +// maxBodyBytes caps a request body. Embed requests carry only ids + names, so +// 1 MiB is generous; the cap stops a hostile/buggy client (behind host auth) +// from driving the shared process to OOM with an oversized body. +const maxBodyBytes = 1 << 20 + func decode(w http.ResponseWriter, r *http.Request, v any) bool { + r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes) if err := json.NewDecoder(r.Body).Decode(v); err != nil { - writeErr(w, http.StatusBadRequest, "invalid_request", "Request body must be valid JSON.") + writeErr(w, http.StatusBadRequest, "invalid_request", "Request body must be valid JSON or is too large.") return false } return true From 049f6c0c71d8db3ab86f153371db7a90e1e39fc6 Mon Sep 17 00:00:00 2001 From: clippy Date: Tue, 8 Sep 2026 14:23:47 -0700 Subject: [PATCH 6/7] fix(embed): Go router orphan-safety + redaction (D/E/F/H) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - wirePipeline: a transient status-poll failure or a cancelled request no longer fails the request — the pipeline exists, so return it as built (last-known status) rather than orphaning it and inviting a retry duplicate. (D) - buildDevNull: best-effort delete the created sink if wiring fails. (E) - pipelineFor: document the one-pipeline-per-connector invariant behind component_of[0]. (F) - tenant: redact the host GetCustomerOrgID error (log server-side, generic 401). (H) Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_011ASKA2VrufrNLGGrcFjY7n --- routers/go/client.go | 25 ++++++++++++++++++++++--- routers/go/embed.go | 8 +++++--- 2 files changed, 27 insertions(+), 6 deletions(-) diff --git a/routers/go/client.go b/routers/go/client.go index 9c802f4..df3a80d 100644 --- a/routers/go/client.go +++ b/routers/go/client.go @@ -241,11 +241,16 @@ func (c *client) wirePipeline(ctx context.Context, org, inputID, outputID, name return BuiltPipeline{}, err } + // The pipeline now EXISTS. Polling for Running is a courtesy — a transient + // status-poll failure (or a cancelled request) must NOT fail the request, or + // the caller treats a live pipeline as un-built and retries into a duplicate. + // On a poll error or cancellation we return the pipeline with its last-known + // status instead. status := "Pending" for i := 0; i < c.pollAttempts; i++ { sd, err := c.do(ctx, "GET", "/v2/"+url.PathEscape(org)+"/pipelines/"+url.PathEscape(created.ID)+"/status", nil) if err != nil { - return BuiltPipeline{}, err + break } var s struct { Status string `json:"status"` @@ -258,7 +263,7 @@ func (c *client) wirePipeline(ctx context.Context, org, inputID, outputID, name } select { case <-ctx.Done(): - return BuiltPipeline{}, ctx.Err() + return BuiltPipeline{PipelineID: created.ID, OutputID: outputID, Status: status, Active: false}, nil case <-time.After(c.pollInterval): } } @@ -284,7 +289,17 @@ func (c *client) buildDevNull(ctx context.Context, org, inputID, name string) (B if err := json.Unmarshal(data, &out); err != nil { return BuiltPipeline{}, err } - return c.wirePipeline(ctx, org, inputID, out.ID, name) + built, err := c.wirePipeline(ctx, org, inputID, out.ID, name) + if err != nil { + // The sink was created but wiring failed. Best-effort delete it so a + // retry doesn't accumulate dev/null outputs; if the cleanup itself fails, + // leave it for orphan monitoring (docs/embed-security-and-operations.md) + // and surface the original error. Create/delete are non-atomic — Monad + // offers no transaction. + _, _ = c.do(ctx, "DELETE", "/v1/"+url.PathEscape(org)+"/outputs/"+url.PathEscape(out.ID), nil) + return BuiltPipeline{}, err + } + return built, nil } type pipeNode struct { @@ -333,6 +348,10 @@ func (c *client) pipelineFor(ctx context.Context, org string, kind componentKind if err := json.Unmarshal(data, &connector); err != nil { return PipelineStatus{}, err } + // The embed flow wires each connector into exactly one pipeline (Monad + // guarantees a pipeline has a single input, and embed never adds a connector + // to a second pipeline), so component_of has at most one entry and taking the + // first is unambiguous. if len(connector.ComponentOf) == 0 || connector.ComponentOf[0].ID == "" { return PipelineStatus{HasPipeline: false}, nil } diff --git a/routers/go/embed.go b/routers/go/embed.go index d2c4f7a..44ef073 100644 --- a/routers/go/embed.go +++ b/routers/go/embed.go @@ -362,11 +362,13 @@ func (h *handler) notFound(w http.ResponseWriter, r *http.Request) { func (h *handler) tenant(w http.ResponseWriter, r *http.Request) (string, bool) { org, err := h.cfg.GetCustomerOrgID(r) if err != nil || org == "" { - msg := "Could not resolve a tenant for the request." + // Redact the host callback's error detail — it can carry internal + // information (DB errors, internal URLs). Log it server-side; return a + // generic message, like the upstream-Monad path does. if err != nil { - msg = err.Error() + log.Printf("[embed] GetCustomerOrgID failed: %v", err) } - writeErr(w, http.StatusUnauthorized, "unauthenticated", msg) + writeErr(w, http.StatusUnauthorized, "unauthenticated", "Could not resolve a tenant for the request.") return "", false } return org, true From 8011a974fc118198066282683da1f6f5221c3246 Mon Sep 17 00:00:00 2001 From: clippy Date: Fri, 11 Sep 2026 17:54:57 -0700 Subject: [PATCH 7/7] =?UTF-8?q?fix(embed):=20address=20Go=20PR=20review=20?= =?UTF-8?q?(Brent/Curtis)=20=E2=80=94=20go=20version,=20error=20detail,=20?= =?UTF-8?q?catalog=20filter,=20README?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - go.mod: bump go 1.23 → 1.25 (Brent: use a current Go). (B1) - client.go do(): keep the FULL upstream body in the (server-side-only) error detail instead of truncating at 300 chars, and drop the now-redundant status prefix (it's a struct field). Truncation was dropping validation details from logs. (Brent B2/B3) - client.go listCatalog: replace the allow-set map with slices.Contains — the allow-list is a tiny operator-configured set. (Brent B4) - routers/go/README.md: cross-link the GetCustomerOrgID security-boundary section to docs/embed-security-and-operations.md (dedicated-org model + orphan monitoring). (Curtis C10) Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_011ASKA2VrufrNLGGrcFjY7n --- routers/go/README.md | 4 ++++ routers/go/client.go | 22 +++++++++------------- routers/go/go.mod | 2 +- 3 files changed, 14 insertions(+), 14 deletions(-) diff --git a/routers/go/README.md b/routers/go/README.md index cc9da25..ba70e77 100644 --- a/routers/go/README.md +++ b/routers/go/README.md @@ -102,6 +102,10 @@ 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`. +For the full trust model — why each embed tenant needs its **own dedicated +Monad org**, and the recommendation to monitor for orphaned resources — see +[`docs/embed-security-and-operations.md`](../../docs/embed-security-and-operations.md). + 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.) diff --git a/routers/go/client.go b/routers/go/client.go index df3a80d..750cd8a 100644 --- a/routers/go/client.go +++ b/routers/go/client.go @@ -8,6 +8,7 @@ import ( "io" "net/http" "net/url" + "slices" "time" ) @@ -99,11 +100,11 @@ func (c *client) do(ctx context.Context, method, path string, body any) ([]byte, return nil, err } if resp.StatusCode >= 400 { - msg := string(data) - if len(msg) > 300 { - msg = msg[:300] - } - return nil, &upstreamError{status: resp.StatusCode, detail: fmt.Sprintf("%d %s: %s", resp.StatusCode, path, msg)} + // Keep the FULL upstream body in the detail: it is logged server-side + // only (never returned to the browser), and truncating it can drop the + // validation details that make a failure diagnosable. The status is a + // struct field, so it isn't repeated in the string. + return nil, &upstreamError{status: resp.StatusCode, detail: fmt.Sprintf("%s: %s", path, string(data))} } return data, nil } @@ -137,16 +138,11 @@ func (c *client) listCatalog(ctx context.Context, kind componentKind, allow []st if err := json.Unmarshal(data, &raw); err != nil { return nil, err } - var allowSet map[string]bool - if len(allow) > 0 { - allowSet = make(map[string]bool, len(allow)) - for _, a := range allow { - allowSet[a] = true - } - } out := []CatalogType{} for _, t := range raw { - if allowSet != nil && !allowSet[t.TypeID] { + // allow is a short operator-configured set of connector types, so a + // linear check is simpler than building a set and the list is tiny. + if len(allow) > 0 && !slices.Contains(allow, t.TypeID) { continue } out = append(out, CatalogType{TypeID: t.TypeID, Name: t.Name}) diff --git a/routers/go/go.mod b/routers/go/go.mod index ebd5da2..02287dc 100644 --- a/routers/go/go.mod +++ b/routers/go/go.mod @@ -1,3 +1,3 @@ module github.com/monad-inc/embed/routers/go -go 1.23 +go 1.25