From f05f73918429bd3b5f57b121836beda470965e8c Mon Sep 17 00:00:00 2001 From: olavurellefsen Date: Sun, 1 Nov 2026 15:47:17 +0000 Subject: [PATCH] fix(pump): separate read routing from tenant webhook writes --- README.md | 21 +++++++++++++++ src/pathways/builder.ts | 9 +++++-- tests/pump-read-routing.test.ts | 48 +++++++++++++++++++++++++++++++++ 3 files changed, 76 insertions(+), 2 deletions(-) create mode 100644 tests/pump-read-routing.test.ts diff --git a/README.md b/README.md index cdaf373..3aa4999 100644 --- a/README.md +++ b/README.md @@ -1364,3 +1364,24 @@ await uploadDocument("./invoice.pdf", "invoice", "finance") ## API Reference For a complete API reference, please see the [API documentation](https://jsr.io/@flowcore/pathways). + +### Separate webhook writes from pump reads + +For a tenant-specific hosted webhook URL, retain `baseUrl` for ingestion and +set `pumpBaseUrl: null` so pump reads use the SDK commands’ dedicated service +hosts. A webhook endpoint does not serve the pump’s catalog or event-source API. + +```typescript +const pathways = new PathwaysBuilder({ + baseUrl: "https://tenant.webhook.flowcore.io", + pumpBaseUrl: null, + tenant: "your-tenant", + dataCore: "your-data-core", + apiKey: process.env.FLOWCORE_API_KEY!, +}) +``` + +Omitting `pumpBaseUrl` preserves the existing single-host routing through +`baseUrl`, including local test fixtures and self-hosted proxies. A string +selects a separate pump-read proxy. This option does not change webhook writes, +provisioning, event identity, state tables or saved checkpoints. diff --git a/src/pathways/builder.ts b/src/pathways/builder.ts index b7d6d59..67fcf5a 100644 --- a/src/pathways/builder.ts +++ b/src/pathways/builder.ts @@ -237,6 +237,8 @@ export interface ManagedPathwayConfig { export interface PathwaysBuilderConfig { baseUrl: string + /** Pump read URL: omit to reuse baseUrl, null for SDK service hosts, or a separate proxy URL. */ + pumpBaseUrl?: string | null tenant: string dataCore: string apiKey: string @@ -479,6 +481,7 @@ export class PathwaysBuilder< // Configuration values needed for cloning private readonly baseUrl: string + private readonly pumpBaseUrl?: string | null private readonly tenant: string private readonly dataCore: string private readonly apiKey: string @@ -536,6 +539,7 @@ export class PathwaysBuilder< */ constructor({ baseUrl, + pumpBaseUrl, tenant, dataCore, apiKey, @@ -581,6 +585,7 @@ export class PathwaysBuilder< // Store configuration values for cloning this.baseUrl = baseUrl + this.pumpBaseUrl = pumpBaseUrl this.tenant = tenant this.dataCore = dataCore this.apiKey = apiKey @@ -2651,7 +2656,7 @@ export class PathwaysBuilder< tenant: this.tenant, dataCore: this.dataCore, apiKey: this.apiKey, - baseUrl: this.baseUrl, + baseUrl: this.pumpBaseUrl === null ? "" : this.pumpBaseUrl ?? this.baseUrl, processEvent: async (pathway: string, event: FlowcoreEvent) => { await this.process(pathway as keyof TPathway, event) }, @@ -2664,7 +2669,7 @@ export class PathwaysBuilder< tenant: this.tenant, dataCore: this.dataCore, apiKey: this.apiKey, - baseUrl: this.baseUrl, + baseUrl: this.pumpBaseUrl === null ? "" : this.pumpBaseUrl ?? this.baseUrl, processEvent: async (pathway: string, event: FlowcoreEvent) => { await this.process(pathway as keyof TPathway, event) }, diff --git a/tests/pump-read-routing.test.ts b/tests/pump-read-routing.test.ts new file mode 100644 index 0000000..e7372df --- /dev/null +++ b/tests/pump-read-routing.test.ts @@ -0,0 +1,48 @@ +import { assertEquals } from "https://deno.land/std@0.224.0/assert/mod.ts" +import { stub } from "https://deno.land/std@0.224.0/testing/mock.ts" +import { FlowcoreDataPump } from "@flowcore/data-pump" +import { z } from "zod" +import { PathwaysBuilder } from "../src/pathways/builder.ts" + +Deno.test("pump read routing preserves the explicit webhook write URL", async (t) => { + for ( + const [name, config, expected] of [ + ["omitted keeps single-host compatibility", {}, "https://tenant.webhook.example"], + ["null uses dedicated SDK read hosts", { pumpBaseUrl: null }, ""], + ["a string selects a read proxy", { pumpBaseUrl: "https://read-proxy.example" }, "https://read-proxy.example"], + ] as const + ) { + await t.step(name, async () => { + let readUrl: string | undefined + let writeUrl = "" + const createStub = stub(FlowcoreDataPump, "create", (options) => { + readUrl = options.baseUrlOverride + return { start: async () => {}, stop: async () => {} } as unknown as ReturnType< + typeof FlowcoreDataPump.create + > + }) + const fetchStub = stub(globalThis, "fetch", async (input) => { + writeUrl = String(input) + return Response.json({ eventId: crypto.randomUUID() }) + }) + const builder = new PathwaysBuilder({ + baseUrl: "https://tenant.webhook.example", + tenant: "synthetic", + dataCore: "synthetic", + apiKey: "fc_synthetic_synthetic", + runtimeEnv: "test", + ...config, + }).register({ flowType: "sample.0", eventType: "created.0", schema: z.object({ id: z.string() }) }) + try { + await builder.write("sample.0/created.0", { data: { id: "synthetic" }, options: { fireAndForget: true } }) + assertEquals(writeUrl, "https://tenant.webhook.example/event/synthetic/synthetic/sample.0/created.0") + await builder.startPump({ stateManagerFactory: () => ({ getState: () => null, setState: () => {} }) }) + assertEquals(readUrl, expected) + } finally { + await builder.stopPump() + fetchStub.restore() + createStub.restore() + } + }) + } +})