Skip to content

Commit c40fbf6

Browse files
committed
Populate subject table from request seam and connection create
1 parent 8212a02 commit c40fbf6

8 files changed

Lines changed: 548 additions & 1 deletion

File tree

‎apps/cloud/src/db/org-deletion.test.ts‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import {
2828
oauth_client,
2929
oauth_session,
3030
plugin_storage,
31+
subject,
3132
tool,
3233
tool_policy,
3334
} from "./executor-schema";
@@ -131,6 +132,13 @@ const seedTenant = async (db: DrizzleDb, tenant: string, tag: string) => {
131132
subject: "s",
132133
});
133134

135+
await db.insert(subject).values({
136+
external_id: `acct-${tag}`,
137+
created_at: now,
138+
last_seen_at: BigInt(now.getTime()),
139+
tenant,
140+
});
141+
134142
const orgNs = `o:${tenant}/plugin`;
135143
const userNs = `u:${tenant}:subject/plugin`;
136144
await db.insert(blob).values({
@@ -156,6 +164,7 @@ const TENANT_TABLES = [
156164
definition,
157165
tool_policy,
158166
plugin_storage,
167+
subject,
159168
] as const;
160169

161170
const countTenantRows = async (db: DrizzleDb, tenant: string): Promise<number> => {

‎apps/cloud/src/db/org-deletion.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import {
2323
oauth_client,
2424
oauth_session,
2525
plugin_storage,
26+
subject,
2627
tool,
2728
tool_policy,
2829
} from "./executor-schema";
@@ -48,6 +49,7 @@ export const purgeOrganizationData = (db: DrizzleDb, organizationId: string): Pr
4849
await tx.delete(oauth_session).where(eq(oauth_session.tenant, organizationId));
4950
await tx.delete(tool_policy).where(eq(tool_policy.tenant, organizationId));
5051
await tx.delete(plugin_storage).where(eq(plugin_storage.tenant, organizationId));
52+
await tx.delete(subject).where(eq(subject.tenant, organizationId));
5153

5254
// Secrets, OAuth tokens, and cached specs live in `blob`, namespaced by
5355
// owner: `o:<org>/<plugin>` (org scope) and `u:<org>:<subject>/<plugin>`
Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
import { mkdtempSync } from "node:fs";
2+
import { tmpdir } from "node:os";
3+
import { join } from "node:path";
4+
5+
import { Effect, Layer } from "effect";
6+
import { afterAll, beforeAll, expect, test } from "@effect/vitest";
7+
import { withQueryContext } from "@executor-js/fumadb/query";
8+
9+
import { makeScopedExecutor } from "@executor-js/api/server";
10+
11+
import { createSelfHostDb, SelfHostDb } from "./db/self-host-db";
12+
import { SelfHostScopedExecutorSeams } from "./execution";
13+
import type { SelfHostPlugins } from "./plugins";
14+
15+
// The `subject` table is populated at the request seam: `makeScopedExecutor` is
16+
// what every HTTP request and MCP session on every host passes through, so
17+
// building a scoped executor IS a sighting. Exercised here against a real host's
18+
// seams (long-lived SQLite DbProvider + real plugins) rather than a stub, since
19+
// the point is that the production wiring reaches the writer at all.
20+
21+
const createScopedExecutor = (accountId: string, organizationId: string) =>
22+
makeScopedExecutor<SelfHostPlugins>(accountId, organizationId, "Default").pipe(
23+
Effect.provide(SelfHostScopedExecutorSeams),
24+
);
25+
26+
const dataDir = mkdtempSync(join(tmpdir(), "eh-subj-"));
27+
process.env.EXECUTOR_DATA_DIR = dataDir;
28+
29+
let dbLayer!: Layer.Layer<SelfHostDb>;
30+
let dbHandle: Awaited<ReturnType<typeof createSelfHostDb>> | undefined;
31+
32+
beforeAll(async () => {
33+
dbHandle = await createSelfHostDb({
34+
path: join(dataDir, "data.db"),
35+
namespace: "executor_selfhost",
36+
version: "1.0.0",
37+
});
38+
dbLayer = Layer.succeed(SelfHostDb)(dbHandle);
39+
});
40+
afterAll(async () => {
41+
await dbHandle?.close();
42+
});
43+
44+
const subjectsOf = async (tenant: string): Promise<readonly string[]> => {
45+
const rows = await withQueryContext(dbHandle!.db, {
46+
tenant,
47+
subject: null,
48+
}).findMany("subject", {
49+
orderBy: ["external_id", "asc"],
50+
});
51+
return rows.map((row) => String(row.external_id));
52+
};
53+
54+
test("building a scoped executor records the acting principal", async () => {
55+
const before = Date.now();
56+
await Effect.runPromise(
57+
createScopedExecutor("alice", "sightings-org").pipe(Effect.provide(dbLayer), Effect.asVoid),
58+
);
59+
60+
expect(await subjectsOf("sightings-org")).toEqual(["alice"]);
61+
62+
const [row] = await withQueryContext(dbHandle!.db, {
63+
tenant: "sightings-org",
64+
subject: null,
65+
}).findMany("subject", {});
66+
expect(Number(row?.last_seen_at)).toBeGreaterThanOrEqual(before);
67+
});
68+
69+
test("repeat requests from several members converge on one row each", async () => {
70+
const org = "sightings-org-2";
71+
await Effect.runPromise(
72+
Effect.forEach(
73+
["alice", "alice", "bob", "alice"],
74+
(accountId) => createScopedExecutor(accountId, org).pipe(Effect.asVoid),
75+
{ discard: true },
76+
).pipe(Effect.provide(dbLayer)),
77+
);
78+
79+
// Four requests, two principals, two rows: the sighting is an upsert, and
80+
// the repeat sightings are absorbed by the throttle rather than duplicating.
81+
expect(await subjectsOf(org)).toEqual(["alice", "bob"]);
82+
});

‎packages/core/api/src/server/scoped-executor.ts‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,11 @@ import {
4242
type Executor,
4343
type StorageFailure,
4444
} from "@executor-js/sdk";
45-
import { makeHostedFetch, makeHostedHttpClientLayer } from "@executor-js/sdk/host-internal";
45+
import {
46+
makeHostedFetch,
47+
makeHostedHttpClientLayer,
48+
touchSubject,
49+
} from "@executor-js/sdk/host-internal";
4650

4751
import { DbProvider } from "./executor-fuma-db";
4852

@@ -281,6 +285,16 @@ export const makeScopedExecutor = <
281285
includeProviders: config.exposeCredentialProviders ?? true,
282286
},
283287
}).pipe(Effect.withSpan("executor.stack.create_executor"));
288+
// Record the sighting. THIS is the seam every HTTP request and MCP session
289+
// on every host passes through, so it is where the `subject` table gets
290+
// populated: a principal earns a row the first time it authenticates,
291+
// whether or not it ever connects anything. Throttled (`touchSubject` only
292+
// rewrites `last_seen_at` past a coarse interval) and non-fatal by
293+
// construction — it returns `Effect<void>`, so a lost sighting logs and the
294+
// request proceeds. `accountId` is passed verbatim, sentinels ("local")
295+
// included.
296+
yield* touchSubject(db, { tenant: organizationId, externalId: accountId });
297+
284298
// The seam erases the plugin tuple type; the caller re-narrows via the
285299
// `TPlugins` phantom. Runtime shape is identical to a typed
286300
// `createExecutor({ plugins })` call.

‎packages/core/sdk/src/executor.ts‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,7 @@ import {
107107
type UpdateToolPolicyInput,
108108
} from "./policies";
109109
import type { CredentialProvider, ProviderEntry } from "./provider";
110+
import { touchSubject } from "./subject-registry";
110111
import type {
111112
AnyPlugin,
112113
Elicit,
@@ -2428,6 +2429,16 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
24282429
}),
24292430
);
24302431

2432+
// Record the sighting. The request seam (`makeScopedExecutor`) already
2433+
// does this for every hosted call, so this is the belt for direct
2434+
// SDK/CLI callers that never pass through it — a connecting principal
2435+
// must always have a subject row. Outside
2436+
// the transaction above: bookkeeping must not roll back the
2437+
// connection, and `touchSubject` cannot fail. No-ops on a pure-org
2438+
// executor (no principal to record), including for `owner: "org"`
2439+
// connections created by a bound member.
2440+
yield* touchSubject(rootDbUntyped, { tenant, externalId: subject });
2441+
24312442
const ref: ConnectionRef = {
24322443
owner: input.owner,
24332444
integration: input.integration,

‎packages/core/sdk/src/host-internal.ts‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,9 @@
2020
// endpoint SSRF guard + default fetch timeout. The SDK's own OAuth flow uses
2121
// them; the host's connection-identity handler reuses the same guard/timeout
2222
// when fetching OIDC userinfo, so they surface here (not the plugin barrel).
23+
// - `touchSubject`: the sole writer of the `subject` table. The SDK writes a
24+
// sighting at connection-create; the host layer writes one per request in
25+
// `makeScopedExecutor`. Not a plugin-author concern.
2326
// ---------------------------------------------------------------------------
2427

2528
export {
@@ -31,6 +34,12 @@ export {
3134

3235
export { OAUTH2_DEFAULT_TIMEOUT_MS, assertSupportedOAuthEndpointUrl } from "./oauth-helpers";
3336

37+
export {
38+
DEFAULT_SUBJECT_LAST_SEEN_THROTTLE_MS,
39+
touchSubject,
40+
type TouchSubjectInput,
41+
} from "./subject-registry";
42+
3443
export {
3544
createExecutorFumaDb,
3645
type CreateExecutorFumaDbOptions,

0 commit comments

Comments
 (0)