feat(scim-sync): read the shared PostgreSQL through Hyperdrive

.github/workflows/ci.yml

@@ -100,6 +100,7 @@ jobs:
               - 'packages/trpc/**'
               - 'packages/web-shared/**'
               - 'packages/worker-utils/**'
+              - 'services/scim-sync/**'
               - 'services/session-ingest/**'
               - 'tools/oxlint/**'
               - 'package.json'
@@ -145,6 +146,7 @@ jobs:
               - 'packages/trpc/**'
               - 'packages/web-shared/**'
               - 'packages/worker-utils/**'
+              - 'services/scim-sync/**'
               - 'services/session-ingest/**'
               - 'tools/oxlint/**'
               - 'package.json'
@@ -695,6 +697,9 @@ jobs:
       - name: Run Session Ingest Postgres tests
         run: NODE_ENV=development pnpm --filter cloudflare-session-ingest test:postgres
 
+      - name: Run SCIM Sync Postgres tests
+        run: NODE_ENV=development pnpm --filter cloudflare-scim-sync test:postgres
+
   # In its own job, madge runs beside the script tests instead of before them.
   dependency-cycle-check:
     needs: changes

services/scim-sync/README.md

@@ -44,7 +44,7 @@ the value can be copied forward by hand.
 | Route | Purpose |
 |---|---|
 | `GET /health` | Worker liveness. Does not touch the Durable Object. |
-| `GET /status` | Interval, cursor, next alarm time, and the last poll result. |
+| `GET /status` | Interval, cursor, next alarm time, the last poll result, and PostgreSQL reachability. |
 
 Both routes are unauthenticated, so the Worker has no *HTTP* ingress: no
 `workers_dev` hostname, no `routes`, and no service binding. Neither route is
@@ -85,6 +85,61 @@ npx wrangler secrets-store secret create 342a86d9e3a94da698e82d0c6e2a36f0 \
   --name WORKOS_API_KEY --scopes workers
 ```
 
+## Database
+
+Reads the shared PostgreSQL through the `HYPERDRIVE` binding — the same
+instance and Hyperdrive config as every other service in this repo. Workers
+cannot hold long-lived connections, so Hyperdrive owns the regional pool and
+`getPgDb` opens a `max: 1` pool per call against the string it hands out.
+
+**Read-only for now.** The query modules under `src/db/` only select; nothing
+writes. The poll loop does not use them yet, so `/status` is their only caller.
+
+The WorkOS directory hierarchy maps onto the organization hierarchy:
+
+| WorkOS | Kilo | Binding |
+|---|---|---|
+| directory | parent organization | `directory_id` |
+| directory group | **sub-organization** | `directory_group_id`, `directory_idp_id` |
+| directory user | user | `kilocode_users.external_id` |
+| group membership | membership on that sub-organization | — |
+
+The binding columns live on every `organizations` row, parent or child, which is
+what makes a sub-organization the thing a directory group resolves to. So
+`findOrganizationByDirectoryGroupId` is the group-resolution query and
+`findChildOrganizations` lists a directory's groups.
+
+`getPgDb` must be called per operation and never cached in module scope. A
+cached client outlives the request or Durable Object context that created it,
+and its I/O then fails in whatever context picks it up next. Calling per use is
+also why the probe recovers on its own after a database restart.
+
+Four schema facts the query modules encode, each easy to get wrong:
+
+- **`organizations.deleted_at IS NULL` on every org query.** The unique indexes
+  on `directory_group_id` and `external_org_id` are *partial*, scoped to live
+  rows, so a soft-deleted organization can hold the same binding as a live one.
+  Omit the filter and those lookups stop being deterministic.
+- **`organization_memberships` has no foreign keys** on either column, so a row
+  does not prove the organization or user exists, and neither side is cleaned
+  up when the other goes away.
+- **`kilocode_users.external_id` is not unique**, so `findUsersByExternalId`
+  returns an array. A caller needing exactly one must decide what a duplicate
+  means rather than taking the first row.
+- **User IDs are `text`, not `uuid`** — OAuth IDs look like `oauth/…`. Never
+  validate, parse, or cast one as a UUID.
+
+Kilo's `organization_groups` table is unrelated and is not read: it holds
+capability policy sets (`read`, `edit`, `browser`, `command`, `mcp`), and
+nothing populates it from directory sync. Worth knowing anyway, because
+deleting a membership cascades a user's assignments in that table away through a
+composite foreign key — so a membership write has authorization consequences no
+directory event names.
+
+Timestamps come back as strings shaped `2026-04-29 01:16:12.945+00`, which
+strict ISO validators such as `z.string().datetime()` reject. Normalize before
+sending them across a strict boundary.
+
 ## Commands
 
 ```sh
@@ -99,9 +154,21 @@ pnpm types       # regenerate worker-configuration.d.ts after config changes
 curl http://127.0.0.1:8819/cdn-cgi/local/scheduled
 ```
 
-There are no tests. The poll has one branch worth asserting — the interval
-clamp — and everything else is the SDK's behavior or WorkOS's; tests arrive with
-the cursor persistence that has real invariants.
+```sh
+pnpm test:postgres   # query modules against a real database
+```
+
+Needs a running local PostgreSQL (`dev/docker-compose.yml`) with migrations
+applied. If a query fails on a missing column, the local database is behind the
+schema:
+
+```sh
+POSTGRES_URL='postgres://postgres:postgres@localhost:5432/postgres' \
+  NODE_ENV=development pnpm drizzle migrate
+```
+
+The poll loop itself has no tests: its one branch worth asserting is the
+interval clamp, and everything else is the SDK's behavior or WorkOS's.
 
 ## Deployment
 

services/scim-sync/package.json

@@ -14,9 +14,11 @@
     "lint": "pnpm -w exec oxlint --config .oxlintrc.json services/scim-sync/src",
     "format": "oxfmt src",
     "format:check": "oxfmt --list-different src",
-    "typecheck": "tsgo --noEmit"
+    "typecheck": "tsgo --noEmit",
+    "test:postgres": "vitest run --config vitest.postgres.config.ts"
   },
   "dependencies": {
+    "@kilocode/db": "workspace:*",
     "@workos-inc/node": "catalog:",
     "workers-tagged-logger": "catalog:"
   },
@@ -25,6 +27,7 @@
     "@types/node": "catalog:",
     "@typescript/native-preview": "catalog:",
     "typescript": "catalog:",
+    "vitest": "catalog:",
     "wrangler": "catalog:"
   }
 }

services/scim-sync/src/db/memberships.ts

@@ -0,0 +1,78 @@
+import { organization_membership_removals, organization_memberships } from '@kilocode/db/schema';
+import { and, eq } from 'drizzle-orm';
+
+import type { WorkerDb } from './pg';
+
+export type MembershipRow = typeof organization_memberships.$inferSelect;
+export type MembershipRemovalRow = typeof organization_membership_removals.$inferSelect;
+
+/**
+ * `organization_memberships` declares no foreign keys on either
+ * `organization_id` or `kilo_user_id`, so a row here does not prove the
+ * organization or the user exists, and neither is cleaned up when the other
+ * side goes away. Callers that need the referenced row must look it up.
+ *
+ * `(organization_id, kilo_user_id)` is unique, so this returns at most one row.
+ */
+export async function findMembership(
+  db: WorkerDb,
+  organizationId: string,
+  kiloUserId: string
+): Promise<MembershipRow | null> {
+  const rows = await db
+    .select()
+    .from(organization_memberships)
+    .where(
+      and(
+        eq(organization_memberships.organization_id, organizationId),
+        eq(organization_memberships.kilo_user_id, kiloUserId)
+      )
+    )
+    .limit(1);
+  return rows.at(0) ?? null;
+}
+
+export async function listOrganizationMemberships(
+  db: WorkerDb,
+  organizationId: string
+): Promise<MembershipRow[]> {
+  return db
+    .select()
+    .from(organization_memberships)
+    .where(eq(organization_memberships.organization_id, organizationId));
+}
+
+export async function listUserMemberships(
+  db: WorkerDb,
+  kiloUserId: string
+): Promise<MembershipRow[]> {
+  return db
+    .select()
+    .from(organization_memberships)
+    .where(eq(organization_memberships.kilo_user_id, kiloUserId));
+}
+
+/**
+ * Removal tombstone. A removed membership is hard-deleted and a row is written
+ * here instead, so the absence of a membership plus the presence of a tombstone
+ * is what distinguishes "deliberately removed" from "never joined". A tombstone
+ * is not cleared when the user rejoins, so its presence does not imply the
+ * membership is currently absent — check both.
+ */
+export async function findMembershipRemoval(
+  db: WorkerDb,
+  organizationId: string,
+  kiloUserId: string
+): Promise<MembershipRemovalRow | null> {
+  const rows = await db
+    .select()
+    .from(organization_membership_removals)
+    .where(
+      and(
+        eq(organization_membership_removals.organization_id, organizationId),
+        eq(organization_membership_removals.kilo_user_id, kiloUserId)
+      )
+    )
+    .limit(1);
+  return rows.at(0) ?? null;
+}

services/scim-sync/src/db/organizations.ts

@@ -0,0 +1,115 @@
+import { organizations } from '@kilocode/db/schema';
+import { and, count, eq, isNull } from 'drizzle-orm';
+
+import type { WorkerDb } from './pg';
+
+/**
+ * The WorkOS directory hierarchy maps onto the organization hierarchy:
+ *
+ * | WorkOS | Kilo | Binding |
+ * |---|---|---|
+ * | directory | parent organization | `directory_id` |
+ * | directory group | **sub-organization** | `directory_group_id`, `directory_idp_id` |
+ * | group membership | membership on that sub-organization | — |
+ *
+ * The binding columns live on every `organizations` row, parent or child, which
+ * is what makes a sub-organization the thing a directory group resolves to.
+ * Kilo's `organization_groups` table is unrelated: it holds capability policy
+ * sets and is not populated by directory sync.
+ */
+
+export type OrganizationRow = typeof organizations.$inferSelect;
+
+/**
+ * Every query here filters `deleted_at IS NULL`.
+ *
+ * That is not cosmetic. The unique indexes on `directory_group_id` and
+ * `external_org_id` are partial, scoped to live rows, so a soft-deleted
+ * organization can hold the same binding as a live one. Dropping the filter
+ * makes those lookups non-deterministic.
+ */
+const isLive = isNull(organizations.deleted_at);
+
+export async function findOrganizationById(
+  db: WorkerDb,
+  organizationId: string
+): Promise<OrganizationRow | null> {
+  const rows = await db
+    .select()
+    .from(organizations)
+    .where(and(eq(organizations.id, organizationId), isLive))
+    .limit(1);
+  return rows.at(0) ?? null;
+}
+
+/**
+ * Resolve the sub-organization a directory group is bound to. This is the
+ * central lookup for a `dsync.group.*` event.
+ */
+export async function findOrganizationByDirectoryGroupId(
+  db: WorkerDb,
+  directoryGroupId: string
+): Promise<OrganizationRow | null> {
+  const rows = await db
+    .select()
+    .from(organizations)
+    .where(and(eq(organizations.directory_group_id, directoryGroupId), isLive))
+    .limit(1);
+  return rows.at(0) ?? null;
+}
+
+/**
+ * Resolve by identity-provider group ID, which survives a directory
+ * disconnect/reconnect that mints a new `directory_group_id`.
+ *
+ * Returns every match rather than one: this index is deliberately non-unique,
+ * because a reconnect can leave two live sub-organizations briefly sharing an
+ * IdP group while the new `directory_group_id` is adopted.
+ */
+export async function findOrganizationsByDirectoryIdpId(
+  db: WorkerDb,
+  directoryIdpId: string
+): Promise<OrganizationRow[]> {
+  return db
+    .select()
+    .from(organizations)
+    .where(and(eq(organizations.directory_idp_id, directoryIdpId), isLive));
+}
+
+/** The organization's identifier in the customer's own system. */
+export async function findOrganizationByExternalOrgId(
+  db: WorkerDb,
+  externalOrgId: string
+): Promise<OrganizationRow | null> {
+  const rows = await db
+    .select()
+    .from(organizations)
+    .where(and(eq(organizations.external_org_id, externalOrgId), isLive))
+    .limit(1);
+  return rows.at(0) ?? null;
+}
+
+/**
+ * The sub-organizations under a parent — which is also the set of directory
+ * groups bound to that directory, since a directory group maps to a
+ * sub-organization.
+ *
+ * The hierarchy is one level deep (a child cannot itself be a parent), so this
+ * is the whole subtree rather than just the first layer. Depth is enforced in
+ * application code, not by the schema.
+ */
+export async function findChildOrganizations(
+  db: WorkerDb,
+  parentOrganizationId: string
+): Promise<OrganizationRow[]> {
+  return db
+    .select()
+    .from(organizations)
+    .where(and(eq(organizations.parent_organization_id, parentOrganizationId), isLive));
+}
+
+/** Reachability probe. Counts in SQL rather than materializing every row. */
+export async function countOrganizations(db: WorkerDb): Promise<number> {
+  const rows = await db.select({ total: count() }).from(organizations).where(isLive);
+  return rows.at(0)?.total ?? 0;
+}

services/scim-sync/src/db/pg.ts

@@ -0,0 +1,31 @@
+import { getWorkerDb, type WorkerDb } from '@kilocode/db/client';
+
+/**
+ * Bounds both the statement and the connection.
+ *
+ * `statement_timeout` is sent in the PostgreSQL startup packet, so it cannot
+ * fire until a connection exists. A database or Hyperdrive that refuses or
+ * blackholes new connections would otherwise wait on `pg.Pool`'s default, which
+ * has no client-side connect timeout — the exact outage `/status` exists to
+ * report. Same reasoning as `services/ai-attribution/src/util/auth.ts`.
+ */
+const TIMEOUT_MS = 10_000;
+
+/**
+ * Return a Drizzle client for the shared PostgreSQL reachable through the
+ * `HYPERDRIVE` binding.
+ *
+ * `getWorkerDb` opens a fresh `max: 1` pool per call — Hyperdrive owns the real
+ * pooling — so callers must invoke this per operation and never cache the
+ * result in module scope. A cached client outlives the request or Durable
+ * Object context that created it, and its I/O then fails in whatever context
+ * picks it up next.
+ */
+export function getPgDb(env: Pick<Env, 'HYPERDRIVE'>): WorkerDb {
+  return getWorkerDb(env.HYPERDRIVE.connectionString, {
+    connectionTimeoutMillis: TIMEOUT_MS,
+    statement_timeout: TIMEOUT_MS,
+  });
+}
+
+export type { WorkerDb };