Batch the admin joined connection read and key the email filter

.changeset/admin-users-batched-connection-read.md

@@ -0,0 +1,20 @@
+---
+"@executor-js/sdk": patch
+---
+
+**Fix: the admin joined user view no longer issues one connection query per subject**
+
+`admin.listSubjectsWithConnections` read a page of subjects and then queried
+connections once per subject, sequentially. A default page therefore cost 100
+round trips inside a single request, which on a per-request socket dominated
+the response. It now reads the page and then batches every subject's
+connections into one query, so the cost is two queries regardless of page size.
+A subject with no connections still reports an empty array rather than dropping
+out of the page, and the batched read carries the same `owner: "user"` and
+tenant scoping the per-subject read did.
+
+The `?email=` filter on the admin users endpoints is also applied before the
+read rather than after it: the address resolves to a principal id and that id is
+read directly, instead of paging the tenant and keeping the row that matched.
+Paging still applies to a filtered response, but to the selected row — one row
+at `offset: 0`, empty beyond it.

packages/core/api/src/admin/admin-users.test.ts

@@ -3,7 +3,15 @@ import { HttpRouter, HttpServer } from "effect/unstable/http";
 import { describe, expect, it } from "@effect/vitest";
 import { Context, Data, Effect, Layer, type Scope } from "effect";
 
-import { Subject, Tenant, collectTables, createExecutor, type Executor } from "@executor-js/sdk";
+import {
+  Subject,
+  Tenant,
+  collectTables,
+  createExecutor,
+  type AdminSubject,
+  type Executor,
+  type ExecutorAdmin,
+} from "@executor-js/sdk";
 import { createSqliteTestFumaDb, type SqliteTestFumaDb } from "@executor-js/sdk/testing";
 import { resetSubjectTouchCache, touchSubject } from "@executor-js/sdk/host-internal";
 
@@ -1068,3 +1076,119 @@ describe("admin users API", () => {
     ),
   );
 });
+
+// ---------------------------------------------------------------------------
+// `?email=` is a KEYED read, not a filtered scan.
+//
+// These drive the reads directly over a recording `ExecutorAdmin`, because the
+// property under test is WHICH storage call the filter makes — invisible from
+// the HTTP edge, where a page-then-filter and a keyed read return the same
+// body. That equivalence is exactly what let the joined view read 100 subjects
+// (and their connections) to answer with one row.
+// ---------------------------------------------------------------------------
+
+const A_SUBJECT: AdminSubject = {
+  externalId: USER_A1,
+  createdAt: new Date(0),
+  lastSeenAt: null,
+  status: null,
+};
+
+/** An `ExecutorAdmin` that answers everything and records the reads it was
+ *  asked for, so a test can assert the call the filter chose. */
+const recordingAdmin = (calls: string[]): ExecutorAdmin => ({
+  listSubjects: () => {
+    calls.push("listSubjects");
+    return Effect.succeed([A_SUBJECT]);
+  },
+  getSubject: () => {
+    calls.push("getSubject");
+    return Effect.succeed(A_SUBJECT);
+  },
+  listSubjectConnections: () => {
+    calls.push("listSubjectConnections");
+    return Effect.succeed([]);
+  },
+  listSubjectsWithConnections: () => {
+    calls.push("listSubjectsWithConnections");
+    return Effect.succeed([{ ...A_SUBJECT, connections: [] }]);
+  },
+  getSubjectWithConnections: () => {
+    calls.push("getSubjectWithConnections");
+    return Effect.succeed({ ...A_SUBJECT, connections: [] });
+  },
+});
+
+describe("admin users reads — the ?email= filter is applied before the read", () => {
+  it.effect("resolves the joined view through the keyed read, never a page scan", () =>
+    Effect.gen(function* () {
+      const calls: string[] = [];
+      const body = yield* listUsersWithConnections(
+        recordingAdmin(calls),
+        { email: A1_EMAIL },
+        stubUserDirectory({}),
+      );
+
+      expect(calls).toEqual(["getSubjectWithConnections"]);
+      expect(body.users.map((user) => user.externalId)).toEqual([USER_A1]);
+    }),
+  );
+
+  it.effect("resolves the plain list through the keyed read too", () =>
+    Effect.gen(function* () {
+      const calls: string[] = [];
+      const body = yield* listUsers(
+        recordingAdmin(calls),
+        { email: A1_EMAIL },
+        stubUserDirectory({}),
+      );
+
+      expect(calls).toEqual(["getSubject"]);
+      expect(body.users.map((user) => user.externalId)).toEqual([USER_A1]);
+    }),
+  );
+
+  it.effect("issues NO storage read at all for an email the directory cannot name", () =>
+    Effect.gen(function* () {
+      const calls: string[] = [];
+      const body = yield* listUsersWithConnections(
+        recordingAdmin(calls),
+        { email: "nobody@users.test" },
+        stubUserDirectory({}),
+      );
+
+      expect(calls).toEqual([]);
+      expect(body.users).toEqual([]);
+    }),
+  );
+
+  it.effect("still pages the filtered result: offset past the only row is empty", () =>
+    Effect.gen(function* () {
+      const calls: string[] = [];
+      const admin = recordingAdmin(calls);
+
+      const first = yield* listUsersWithConnections(
+        admin,
+        { email: A1_EMAIL, offset: 0, limit: 10 },
+        stubUserDirectory({}),
+      );
+      const past = yield* listUsersWithConnections(
+        admin,
+        { email: A1_EMAIL, offset: 1, limit: 10 },
+        stubUserDirectory({}),
+      );
+
+      expect(first.users.map((user) => user.externalId)).toEqual([USER_A1]);
+      expect(past.users).toEqual([]);
+    }),
+  );
+
+  it.effect("leaves the unfiltered list on the paged read", () =>
+    Effect.gen(function* () {
+      const calls: string[] = [];
+      yield* listUsersWithConnections(recordingAdmin(calls), { limit: 50 }, stubUserDirectory({}));
+
+      expect(calls).toEqual(["listSubjectsWithConnections"]);
+    }),
+  );
+});

packages/core/api/src/admin/api.ts

@@ -247,8 +247,15 @@ const AdminUserIdentifierParams = { identifier: Schema.String };
 // `email` is an exact-match FILTER on the fixed list shape (the response is
 // still a `users` array, empty when nothing matches), which is how every other
 // list here uses query params. It selects a specific principal rather than
-// narrowing a scan, so paging over a filtered result is not meaningful — it is
-// still honored so the endpoint has one set of semantics, not two.
+// narrowing a scan, so the read resolves the address to an id and reads THAT
+// id — it does not page the tenant and keep the row that matches.
+//
+// FILTER, THEN PAGE, in that order. The window therefore applies to the
+// selected row rather than to the scan it would once have been found in: with
+// a filter present a response is one row at `offset: 0` and empty beyond it.
+// Paging is applied rather than ignored so the endpoint keeps one set of
+// semantics, not two — but on a filter that names a single principal it can
+// only ever include or exclude that principal.
 //
 // Carried as a plain string: the trim + lower-case normalization lives at the
 // handler seam (`normalizeEmail`), which is also where the single-user path

packages/core/api/src/admin/reads.ts

@@ -248,14 +248,49 @@ const resolveEmailFilter = (
   );
 };
 
-/** Keep only the subject naming `externalId`. The filter is applied AFTER the
- *  page read, so it narrows the requested page rather than scanning past it —
- *  see the contract's note on paging a filtered result. */
-const filterToId = <T extends { readonly externalId: string }>(
-  subjects: readonly T[],
-  externalId: string | null,
-): readonly T[] =>
-  externalId === null ? [] : subjects.filter((subject) => subject.externalId === externalId);
+/**
+ * Apply the caller's paging window to rows the email filter ALREADY selected.
+ *
+ * An `?email=` read resolves to at most one subject, so the window can only
+ * keep that row (`offset: 0`) or drop it. It is applied rather than ignored so
+ * the endpoint keeps ONE set of paging semantics, instead of growing a second
+ * set that appears only when a filter is present.
+ */
+const pageOf = <T>(rows: readonly T[], options: AdminUsersListOptions): readonly T[] => {
+  const offset = Math.max(Math.floor(options.offset ?? 0), 0);
+  if (options.limit === undefined) return rows.slice(offset);
+  return rows.slice(offset, offset + Math.max(Math.floor(options.limit), 1));
+};
+
+/**
+ * The `?email=` read, as a KEYED lookup.
+ *
+ * The filter names ONE principal, so it resolves to an id and reads that id
+ * directly. Reading a page and discarding the rest — which is what this did
+ * before — made the joined view pull a full default page (100 subjects AND
+ * their connections) to answer with a single row.
+ *
+ * ORDER OF OPERATIONS IS THE FIX, and it also settles what paging means on a
+ * filtered result: the window applies to the selected row rather than to the
+ * scan the row was found in. That is how any "filter, then page" read behaves,
+ * and the only reading that survives the scan going away.
+ *
+ * A `read` that answers `null` is a resolved id with no subject row — the same
+ * "absent" the single-user path reports, not a storage fault.
+ */
+const selectByEmail = <T>(
+  directory: AdminUserDirectory,
+  email: string,
+  options: AdminUsersListOptions,
+  read: (externalId: string) => Effect.Effect<T | null, AdminUsersError>,
+): Effect.Effect<readonly T[], AdminUsersError> =>
+  Effect.gen(function* () {
+    // An email no directory can resolve matches nothing, and costs no read.
+    const wanted = yield* resolveEmailFilter(directory, email);
+    if (wanted === null) return [];
+    const row = yield* read(wanted);
+    return row === null ? [] : pageOf([row], options);
+  });
 
 export const listUsers = (
   admin: ExecutorAdmin,
@@ -264,10 +299,12 @@ export const listUsers = (
 ): Effect.Effect<typeof AdminUsersResponse.Type, AdminUsersError> =>
   Effect.gen(function* () {
     const dir = asDirectory(directory);
-    const wanted =
-      options.email === undefined ? undefined : yield* resolveEmailFilter(dir, options.email);
-    const all = yield* admin.listSubjects(options).pipe(Effect.mapError(readFailed("users")));
-    const subjects = wanted === undefined ? all : filterToId(all, wanted);
+    const subjects =
+      options.email === undefined
+        ? yield* admin.listSubjects(options).pipe(Effect.mapError(readFailed("users")))
+        : yield* selectByEmail(dir, options.email, options, (externalId) =>
+            admin.getSubject(externalId).pipe(Effect.mapError(readFailed("users"))),
+          );
     // One directory read for the page that was actually returned, joined in
     // memory — never a lookup per user.
     const identities = yield* resolveIdentities(
@@ -284,12 +321,14 @@ export const listUsersWithConnections = (
 ): Effect.Effect<typeof AdminUsersWithConnectionsResponse.Type, AdminUsersError> =>
   Effect.gen(function* () {
     const dir = asDirectory(directory);
-    const wanted =
-      options.email === undefined ? undefined : yield* resolveEmailFilter(dir, options.email);
-    const all = yield* admin
-      .listSubjectsWithConnections(options)
-      .pipe(Effect.mapError(readFailed("users")));
-    const subjects = wanted === undefined ? all : filterToId(all, wanted);
+    const subjects =
+      options.email === undefined
+        ? yield* admin
+            .listSubjectsWithConnections(options)
+            .pipe(Effect.mapError(readFailed("users")))
+        : yield* selectByEmail(dir, options.email, options, (externalId) =>
+            admin.getSubjectWithConnections(externalId).pipe(Effect.mapError(readFailed("users"))),
+          );
     const identities = yield* resolveIdentities(
       dir.identities,
       subjects.map((subject) => subject.externalId),

packages/core/sdk/src/executor.ts

@@ -492,15 +492,14 @@ export interface AdminListSubjectsOptions {
 /**
  * Page size applied when a caller names none. Every admin list is BOUNDED:
  * `listSubjects()` with no arguments is the obvious call, and unbounded it
- * returns every subject in the tenant — which `listSubjectsWithConnections`
- * then turns into one sequential connection query PER SUBJECT, inside a single
- * request, over a per-request socket on cloud. A default is what keeps the
- * no-args call honest; a caller who wants more asks for more, up to
- * {@link ADMIN_MAX_PAGE_SIZE}.
+ * returns every subject in the tenant — an unbounded row count to build,
+ * serialize, and ship, and an unbounded `in` predicate for the joined read to
+ * carry. A default is what keeps the no-args call honest; a caller who wants
+ * more asks for more, up to {@link ADMIN_MAX_PAGE_SIZE}.
  *
  * 100 rather than the maximum: large enough that no realistic operator UI pages
- * twice for a first screen, small enough that the joined read's fan-out stays a
- * bounded cost even at its worst.
+ * twice for a first screen, small enough that one response stays a bounded
+ * amount of work even at its worst.
  */
 export const ADMIN_DEFAULT_PAGE_SIZE = 100;
 
@@ -557,12 +556,12 @@ export interface ExecutorAdmin {
   readonly listSubjectConnections: (
     externalId: string,
   ) => Effect.Effect<readonly AdminConnection[], StorageFailure>;
-  /** `listSubjects` joined with each subject's connections — one connection
-   *  query per subject IN THE PAGE, sequentially. The paging bound is what
-   *  makes that fan-out affordable: it is capped at
-   *  {@link ADMIN_DEFAULT_PAGE_SIZE} round trips by default and
-   *  {@link ADMIN_MAX_PAGE_SIZE} at worst, never "every subject in the
-   *  tenant". */
+  /** `listSubjects` joined with each subject's connections in TWO queries —
+   *  the page of subjects, then one batched connection read over that page.
+   *  The cost does not scale with page size, so {@link ADMIN_DEFAULT_PAGE_SIZE}
+   *  and {@link ADMIN_MAX_PAGE_SIZE} bound the ROWS returned rather than the
+   *  round trips taken. A subject with no connections reports an empty array;
+   *  it is never dropped from the page. */
   readonly listSubjectsWithConnections: (
     options?: AdminListSubjectsOptions,
   ) => Effect.Effect<readonly AdminSubjectWithConnections[], StorageFailure>;
@@ -4835,16 +4834,57 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
           })
           .pipe(Effect.map((rows) => rows.map(rowToAdminConnection)));
 
+      // ONE connection query for the whole page, not one per subject. The
+      // per-subject form was an N+1: a default page issued 100 sequential
+      // `findMany`s over a per-request socket, which on cloud cost ~1.4s of a
+      // ~2.4s response. Cost is now two queries regardless of page size.
+      //
+      // The `in` predicate carries the SAME `owner: "user"` clause the keyed
+      // read does, so org rows (whose `subject` is the empty-string sentinel)
+      // stay excluded, and the tenant policy scopes both reads identically.
+      //
+      // Ordering is preserved WITHOUT a per-subject sort: the query orders by
+      // `(integration, name)` across the page, and grouping walks those rows
+      // in order, so each subject's bucket comes out in the same order the
+      // per-subject query produced. Subjects with no connections still report
+      // an empty array rather than dropping out of the page.
       const listSubjectsWithConnections = (
         options?: AdminListSubjectsOptions,
       ): Effect.Effect<readonly AdminSubjectWithConnections[], StorageFailure> =>
         Effect.gen(function* () {
           const subjects = yield* listSubjects(options);
-          return yield* Effect.forEach(subjects, (entry) =>
-            listSubjectConnections(entry.externalId).pipe(
-              Effect.map((connections) => ({ ...entry, connections })),
-            ),
-          );
+          // No page, no connection query — `in ([])` is a query that cannot
+          // match, so issuing it would be pure latency.
+          if (subjects.length === 0) return [];
+
+          const rows = yield* platformCore.findMany("connection", {
+            where: (b: AnyCb) =>
+              b.and(
+                b("owner", "=", "user"),
+                b(
+                  "subject",
+                  "in",
+                  subjects.map((entry) => entry.externalId),
+                ),
+              ),
+            orderBy: [
+              ["integration", "asc"],
+              ["name", "asc"],
+            ],
+          });
+
+          const bySubject = new Map<string, AdminConnection[]>();
+          for (const row of rows) {
+            const connection = rowToAdminConnection(row);
+            const bucket = bySubject.get(row.subject);
+            if (bucket) bucket.push(connection);
+            else bySubject.set(row.subject, [connection]);
+          }
+
+          return subjects.map((entry) => ({
+            ...entry,
+            connections: bySubject.get(entry.externalId) ?? [],
+          }));
         });
 
       // Absent subject short-circuits: no connection query is issued for a