// ---------------------------------------------------------------------------// Invoke-path batching: DataLoader semantics via Effect Request/RequestResolver.//// The invoke hot path loads four things per call: the tool row, the active// policy rule set, the connection row, and the integration row. Called once,// that's four point queries, which is fine. Called N times concurrently (code// mode fanning out `await Promise.all(items.map(x => tools.foo.bar(x)))`, an// MCP host serving parallel tool calls), it's 4×N queries against the same// handful of rows: the classic N+1.//// Each lookup below is an Effect `Request` plus a `RequestResolver`. The// Effect runtime collects every request issued within the same microtask// window (resolver delay is `Effect.yieldNow`) and hands the resolver the// whole batch, which it serves with ONE query per table. Sequential callers// pay nothing: a batch of one is exactly the point query this replaced. There// is no cross-batch caching; every batch re-reads storage, so invalidation// semantics are unchanged.//// Requests are keyed by plain data (owner/slug/name strings), never rows, so// equality is meaningful and nothing heavy is retained.// --------------------------------------------------------------------------- import { Effect, Exit, Request, RequestResolver } from "effect"; import type { ConnectionRow, IntegrationRow, ToolInvocationRow } from "./core-schema";import { activeFumaDbRef, type StorageFailure } from "./fuma-runtime";import type { ConnectionName, IntegrationSlug, Owner, ToolName } from "./ids"; // ---------------------------------------------------------------------------// Request types// --------------------------------------------------------------------------- interface GetToolRow extends Request.Request<ToolInvocationRow | null, StorageFailure> { readonly _tag: "GetToolRow"; readonly owner: Owner; readonly integration: IntegrationSlug; readonly connection: ConnectionName; readonly tool: ToolName;}const GetToolRow = Request.tagged<GetToolRow>("GetToolRow"); interface GetConnectionRow extends Request.Request<ConnectionRow | null, StorageFailure> { readonly _tag: "GetConnectionRow"; readonly owner: Owner; readonly integration: IntegrationSlug; readonly name: ConnectionName;}const GetConnectionRow = Request.tagged<GetConnectionRow>("GetConnectionRow"); interface GetIntegrationRow extends Request.Request<IntegrationRow | null, StorageFailure> { readonly _tag: "GetIntegrationRow"; readonly slug: IntegrationSlug;}const GetIntegrationRow = Request.tagged<GetIntegrationRow>("GetIntegrationRow"); // The policy rule set is one shared snapshot per batch window: every request// is identical, so the resolver runs the underlying load once and fans the// result out to all awaiting fibers.interface GetPolicyRuleSet<A> extends Request.Request<A, StorageFailure> { readonly _tag: "GetPolicyRuleSet";} // ---------------------------------------------------------------------------// Wiring: the executor passes its storage closures in; we hand back// point-lookup functions with identical signatures to the ones they replace.// --------------------------------------------------------------------------- export interface InvokeBatchingDeps<TPolicyRuleSet> { /** Batched tool-row load: one query for all (owner, integration, connection, * tool) tuples in the window, projected to the invocation columns. */ readonly loadToolRows: ( requests: readonly { readonly owner: Owner; readonly integration: IntegrationSlug; readonly connection: ConnectionName; readonly tool: ToolName; }[], ) => Effect.Effect<readonly ToolInvocationRow[], StorageFailure>; readonly loadConnectionRows: ( requests: readonly { readonly owner: Owner; readonly integration: IntegrationSlug; readonly name: ConnectionName; }[], ) => Effect.Effect<readonly ConnectionRow[], StorageFailure>; readonly loadIntegrationRows: ( slugs: readonly IntegrationSlug[], ) => Effect.Effect<readonly IntegrationRow[], StorageFailure>; readonly loadPolicyRuleSet: () => Effect.Effect<TPolicyRuleSet, StorageFailure>;} export interface InvokeBatching<TPolicyRuleSet> { readonly getToolRow: (input: { readonly owner: Owner; readonly integration: IntegrationSlug; readonly connection: ConnectionName; readonly tool: ToolName; }) => Effect.Effect<ToolInvocationRow | null, StorageFailure>; readonly getConnectionRow: (input: { readonly owner: Owner; readonly integration: IntegrationSlug; readonly name: ConnectionName; }) => Effect.Effect<ConnectionRow | null, StorageFailure>; readonly getIntegrationRow: ( slug: IntegrationSlug, ) => Effect.Effect<IntegrationRow | null, StorageFailure>; readonly getPolicyRuleSet: () => Effect.Effect<TPolicyRuleSet, StorageFailure>;} // Batches from different fibers execute the `runAll` effect on a fiber forked// with the FIRST caller's context. Inside `fuma.transaction` the storage// handle rides on `activeFumaDbRef` in fiber context, so joining a shared// batch would read through the wrong (or no) transaction. Transactional// callers bypass the batch window and run the load directly on their own// fiber — same query, exact transaction semantics, no cross-fiber sharing.const inTransaction: Effect.Effect<boolean> = Effect.map( Effect.service(activeFumaDbRef), (active) => active !== null,); /** Complete every entry in a batch from a keyed index of loaded rows. A * request whose key has no row completes with `null`: absence is a value * (the invoke path turns it into its own typed not-found error). A load * failure fails every entry in the batch, mirroring what each point query * would have done. */const completeFromIndex = <A extends Request.Request<unknown, StorageFailure>>( entries: readonly Request.Entry<A>[], load: Effect.Effect<ReadonlyMap<string, Request.Success<A>>, StorageFailure>, keyOf: (request: A) => string,): Effect.Effect<void, StorageFailure> => load.pipe( Effect.matchEffect({ onSuccess: (index) => Effect.sync(() => { for (const entry of entries) { entry.completeUnsafe( Exit.succeed((index.get(keyOf(entry.request)) ?? null) as Request.Success<A>), ); } }), onFailure: (error) => Effect.sync(() => { for (const entry of entries) { // The constraint pins A's error channel to StorageFailure, but the // conditional `Request.Error<A>` doesn't reduce over an unresolved // generic — assert the equivalence once here. entry.completeUnsafe(Exit.fail(error) as Exit.Exit<never, Request.Error<A>>); } }), }), ); /** The runtime collects entries per window without deduping equal requests; * two fibers asking for the same connection contribute two entries. Loaders * receive each distinct key once (DataLoader's contract) and both entries * complete from the shared index. */const uniqueBy = <T>(items: readonly T[], key: (item: T) => string): readonly T[] => { const seen = new Set<string>(); const out: T[] = []; for (const item of items) { const k = key(item); if (seen.has(k)) continue; seen.add(k); out.push(item); } return out;}; const toolKey = (r: { readonly owner: string; readonly integration: string; readonly connection: string; readonly tool: string;}): string => `${r.owner} ${r.integration} ${r.connection} ${r.tool}`; const connectionKey = (r: { readonly owner: string; readonly integration: string; readonly name: string;}): string => `${r.owner} ${r.integration} ${r.name}`; export const makeInvokeBatching = <TPolicyRuleSet>( deps: InvokeBatchingDeps<TPolicyRuleSet>,): InvokeBatching<TPolicyRuleSet> => { const toolResolver = RequestResolver.make<GetToolRow>((entries) => completeFromIndex( entries, deps .loadToolRows( uniqueBy( entries.map((entry) => entry.request), toolKey, ), ) .pipe( Effect.map( (rows) => new Map( rows.map((row) => [ toolKey({ owner: row.owner, integration: row.integration, connection: row.connection, tool: row.name, }), row, ]), ), ), ), toolKey, ), ); const connectionResolver = RequestResolver.make<GetConnectionRow>((entries) => completeFromIndex( entries, deps .loadConnectionRows( uniqueBy( entries.map((entry) => entry.request), connectionKey, ), ) .pipe( Effect.map( (rows) => new Map( rows.map((row) => [ connectionKey({ owner: row.owner, integration: row.integration, name: row.name, }), row, ]), ), ), ), connectionKey, ), ); const integrationResolver = RequestResolver.make<GetIntegrationRow>((entries) => completeFromIndex( entries, deps .loadIntegrationRows( uniqueBy( entries.map((entry) => entry.request.slug), String, ), ) .pipe(Effect.map((rows) => new Map(rows.map((row) => [String(row.slug), row])))), (request) => String(request.slug), ), ); const policyResolver = RequestResolver.make<GetPolicyRuleSet<TPolicyRuleSet>>((entries) => deps.loadPolicyRuleSet().pipe( Effect.matchEffect({ onSuccess: (ruleSet) => Effect.sync(() => { for (const entry of entries) entry.completeUnsafe(Exit.succeed(ruleSet)); }), onFailure: (error) => Effect.sync(() => { for (const entry of entries) entry.completeUnsafe(Exit.fail(error)); }), }), ), ); const policyRequest = Request.of<GetPolicyRuleSet<TPolicyRuleSet>>()({ _tag: "GetPolicyRuleSet", }); const first = <A>(rows: readonly A[]): A | null => rows[0] ?? null; return { getToolRow: (input) => Effect.flatMap(inTransaction, (transactional) => transactional ? Effect.map(deps.loadToolRows([input]), first) : Effect.request(GetToolRow(input), toolResolver), ), getConnectionRow: (input) => Effect.flatMap(inTransaction, (transactional) => transactional ? Effect.map(deps.loadConnectionRows([input]), first) : Effect.request(GetConnectionRow(input), connectionResolver), ), getIntegrationRow: (slug) => Effect.flatMap(inTransaction, (transactional) => transactional ? Effect.map(deps.loadIntegrationRows([slug]), first) : Effect.request(GetIntegrationRow({ slug }), integrationResolver), ), getPolicyRuleSet: () => Effect.flatMap(inTransaction, (transactional) => transactional ? deps.loadPolicyRuleSet() : Effect.request(policyRequest, policyResolver), ), };};
// ---------------------------------------------------------------------------
// Invoke-path batching: DataLoader semantics via Effect Request/RequestResolver.
//
// The invoke hot path loads four things per call: the tool row, the active
// policy rule set, the connection row, and the integration row. Called once,
// that's four point queries, which is fine. Called N times concurrently (code
// mode fanning out `await Promise.all(items.map(x => tools.foo.bar(x)))`, an
// MCP host serving parallel tool calls), it's 4×N queries against the same
// handful of rows: the classic N+1.
//
// Each lookup below is an Effect `Request` plus a `RequestResolver`. The
// Effect runtime collects every request issued within the same microtask
// window (resolver delay is `Effect.yieldNow`) and hands the resolver the
// whole batch, which it serves with ONE query per table. Sequential callers
// pay nothing: a batch of one is exactly the point query this replaced. There
// is no cross-batch caching; every batch re-reads storage, so invalidation
// semantics are unchanged.
//
// Requests are keyed by plain data (owner/slug/name strings), never rows, so
// equality is meaningful and nothing heavy is retained.
// ---------------------------------------------------------------------------
import { Effect, Exit, Request, RequestResolver } from "effect";
import type { ConnectionRow, IntegrationRow, ToolInvocationRow } from "./core-schema";
import { activeFumaDbRef, type StorageFailure } from "./fuma-runtime";
import type { ConnectionName, IntegrationSlug, Owner, ToolName } from "./ids";
// ---------------------------------------------------------------------------
// Request types
// ---------------------------------------------------------------------------
interface GetToolRow extends Request.Request<ToolInvocationRow | null, StorageFailure> {
readonly _tag: "GetToolRow";
readonly owner: Owner;
readonly integration: IntegrationSlug;
readonly connection: ConnectionName;
readonly tool: ToolName;
}
const GetToolRow = Request.tagged<GetToolRow>("GetToolRow");
interface GetConnectionRow extends Request.Request<ConnectionRow | null, StorageFailure> {
readonly _tag: "GetConnectionRow";
readonly owner: Owner;
readonly integration: IntegrationSlug;
readonly name: ConnectionName;
}
const GetConnectionRow = Request.tagged<GetConnectionRow>("GetConnectionRow");
interface GetIntegrationRow extends Request.Request<IntegrationRow | null, StorageFailure> {
readonly _tag: "GetIntegrationRow";
readonly slug: IntegrationSlug;
}
const GetIntegrationRow = Request.tagged<GetIntegrationRow>("GetIntegrationRow");
// The policy rule set is one shared snapshot per batch window: every request
// is identical, so the resolver runs the underlying load once and fans the
// result out to all awaiting fibers.
interface GetPolicyRuleSet<A> extends Request.Request<A, StorageFailure> {
readonly _tag: "GetPolicyRuleSet";
}
// ---------------------------------------------------------------------------
// Wiring: the executor passes its storage closures in; we hand back
// point-lookup functions with identical signatures to the ones they replace.
// ---------------------------------------------------------------------------
export interface InvokeBatchingDeps<TPolicyRuleSet> {
/** Batched tool-row load: one query for all (owner, integration, connection,
* tool) tuples in the window, projected to the invocation columns. */
readonly loadToolRows: (
requests: readonly {
readonly owner: Owner;
readonly integration: IntegrationSlug;
readonly connection: ConnectionName;
readonly tool: ToolName;
}[],
) => Effect.Effect<readonly ToolInvocationRow[], StorageFailure>;
readonly loadConnectionRows: (
requests: readonly {
readonly owner: Owner;
readonly integration: IntegrationSlug;
readonly name: ConnectionName;
}[],
) => Effect.Effect<readonly ConnectionRow[], StorageFailure>;
readonly loadIntegrationRows: (
slugs: readonly IntegrationSlug[],
) => Effect.Effect<readonly IntegrationRow[], StorageFailure>;
readonly loadPolicyRuleSet: () => Effect.Effect<TPolicyRuleSet, StorageFailure>;
}
export interface InvokeBatching<TPolicyRuleSet> {
readonly getToolRow: (input: {
readonly owner: Owner;
readonly integration: IntegrationSlug;
readonly connection: ConnectionName;
readonly tool: ToolName;
}) => Effect.Effect<ToolInvocationRow | null, StorageFailure>;
readonly getConnectionRow: (input: {
readonly owner: Owner;
readonly integration: IntegrationSlug;
readonly name: ConnectionName;
}) => Effect.Effect<ConnectionRow | null, StorageFailure>;
readonly getIntegrationRow: (
slug: IntegrationSlug,
) => Effect.Effect<IntegrationRow | null, StorageFailure>;
readonly getPolicyRuleSet: () => Effect.Effect<TPolicyRuleSet, StorageFailure>;
}
// Batches from different fibers execute the `runAll` effect on a fiber forked
// with the FIRST caller's context. Inside `fuma.transaction` the storage
// handle rides on `activeFumaDbRef` in fiber context, so joining a shared
// batch would read through the wrong (or no) transaction. Transactional
// callers bypass the batch window and run the load directly on their own
// fiber — same query, exact transaction semantics, no cross-fiber sharing.
const inTransaction: Effect.Effect<boolean> = Effect.map(
Effect.service(activeFumaDbRef),
(active) => active !== null,
);
/** Complete every entry in a batch from a keyed index of loaded rows. A
* request whose key has no row completes with `null`: absence is a value
* (the invoke path turns it into its own typed not-found error). A load
* failure fails every entry in the batch, mirroring what each point query
* would have done. */
const completeFromIndex = <A extends Request.Request<unknown, StorageFailure>>(
entries: readonly Request.Entry<A>[],
load: Effect.Effect<ReadonlyMap<string, Request.Success<A>>, StorageFailure>,
keyOf: (request: A) => string,
): Effect.Effect<void, StorageFailure> =>
load.pipe(
Effect.matchEffect({
onSuccess: (index) =>
Effect.sync(() => {
for (const entry of entries) {
entry.completeUnsafe(
Exit.succeed((index.get(keyOf(entry.request)) ?? null) as Request.Success<A>),
);
}
}),
onFailure: (error) =>
Effect.sync(() => {
for (const entry of entries) {
// The constraint pins A's error channel to StorageFailure, but the
// conditional `Request.Error<A>` doesn't reduce over an unresolved
// generic — assert the equivalence once here.
entry.completeUnsafe(Exit.fail(error) as Exit.Exit<never, Request.Error<A>>);
}
}),
}),
);
/** The runtime collects entries per window without deduping equal requests;
* two fibers asking for the same connection contribute two entries. Loaders
* receive each distinct key once (DataLoader's contract) and both entries
* complete from the shared index. */
const uniqueBy = <T>(items: readonly T[], key: (item: T) => string): readonly T[] => {
const seen = new Set<string>();
const out: T[] = [];
for (const item of items) {
const k = key(item);
if (seen.has(k)) continue;
seen.add(k);
out.push(item);
}
return out;
};
const toolKey = (r: {
readonly owner: string;
readonly integration: string;
readonly connection: string;
readonly tool: string;
}): string => `${r.owner} ${r.integration} ${r.connection} ${r.tool}`;
const connectionKey = (r: {
readonly owner: string;
readonly integration: string;
readonly name: string;
}): string => `${r.owner} ${r.integration} ${r.name}`;
export const makeInvokeBatching = <TPolicyRuleSet>(
deps: InvokeBatchingDeps<TPolicyRuleSet>,
): InvokeBatching<TPolicyRuleSet> => {
const toolResolver = RequestResolver.make<GetToolRow>((entries) =>
completeFromIndex(
entries,
deps
.loadToolRows(
uniqueBy(
entries.map((entry) => entry.request),
toolKey,
),
)
.pipe(
Effect.map(
(rows) =>
new Map(
rows.map((row) => [
toolKey({
owner: row.owner,
integration: row.integration,
connection: row.connection,
tool: row.name,
}),
row,
]),
),
),
),
toolKey,
),
);
const connectionResolver = RequestResolver.make<GetConnectionRow>((entries) =>
completeFromIndex(
entries,
deps
.loadConnectionRows(
uniqueBy(
entries.map((entry) => entry.request),
connectionKey,
),
)
.pipe(
Effect.map(
(rows) =>
new Map(
rows.map((row) => [
connectionKey({
owner: row.owner,
integration: row.integration,
name: row.name,
}),
row,
]),
),
),
),
connectionKey,
),
);
const integrationResolver = RequestResolver.make<GetIntegrationRow>((entries) =>
completeFromIndex(
entries,
deps
.loadIntegrationRows(
uniqueBy(
entries.map((entry) => entry.request.slug),
String,
),
)
.pipe(Effect.map((rows) => new Map(rows.map((row) => [String(row.slug), row])))),
(request) => String(request.slug),
),
);
const policyResolver = RequestResolver.make<GetPolicyRuleSet<TPolicyRuleSet>>((entries) =>
deps.loadPolicyRuleSet().pipe(
Effect.matchEffect({
onSuccess: (ruleSet) =>
Effect.sync(() => {
for (const entry of entries) entry.completeUnsafe(Exit.succeed(ruleSet));
}),
onFailure: (error) =>
Effect.sync(() => {
for (const entry of entries) entry.completeUnsafe(Exit.fail(error));
}),
}),
),
);
const policyRequest = Request.of<GetPolicyRuleSet<TPolicyRuleSet>>()({
_tag: "GetPolicyRuleSet",
});
const first = <A>(rows: readonly A[]): A | null => rows[0] ?? null;
return {
getToolRow: (input) =>
Effect.flatMap(inTransaction, (transactional) =>
transactional
? Effect.map(deps.loadToolRows([input]), first)
: Effect.request(GetToolRow(input), toolResolver),
),
getConnectionRow: (input) =>
Effect.flatMap(inTransaction, (transactional) =>
transactional
? Effect.map(deps.loadConnectionRows([input]), first)
: Effect.request(GetConnectionRow(input), connectionResolver),
),
getIntegrationRow: (slug) =>
Effect.flatMap(inTransaction, (transactional) =>
transactional
? Effect.map(deps.loadIntegrationRows([slug]), first)
: Effect.request(GetIntegrationRow({ slug }), integrationResolver),
),
getPolicyRuleSet: () =>
Effect.flatMap(inTransaction, (transactional) =>
transactional ? deps.loadPolicyRuleSet() : Effect.request(policyRequest, policyResolver),
),
};
};