Feed the cloud membership mirror from login and member writes

Author: RhysSullivan

apps/cloud/package.json

@@ -32,6 +32,8 @@
     "db:backfill-org-slugs:dev": "op run --env-file=.env.op -- bun run scripts/backfill-org-slugs.ts",
     "db:backfill-subjects:prod": "op run --env-file=.env.production -- bun run scripts/backfill-subjects.ts",
     "db:backfill-subjects:dev": "op run --env-file=.env.op -- bun run scripts/backfill-subjects.ts",
+    "db:backfill-workos-mirror:prod": "op run --env-file=.env.production -- bun run scripts/backfill-workos-mirror.ts",
+    "db:backfill-workos-mirror:dev": "op run --env-file=.env.op -- bun run scripts/backfill-workos-mirror.ts",
     "routes:gen": "bun scripts/gen-routes.ts",
     "vendor-wasm": "bun run scripts/vendor-quickjs-wasm.ts"
   },

apps/cloud/scripts/backfill-workos-mirror.ts

@@ -0,0 +1,79 @@
+// ---------------------------------------------------------------------------
+// One-off data backfill: fill the membership mirror (`accounts` profile
+// columns + `memberships` rows, migration 0018) from WorkOS for every
+// organization the mirror already knows.
+//
+//   bun run db:backfill-workos-mirror:prod   # op run --env-file=.env.production
+//   bun run db:backfill-workos-mirror:dev    # against the local PGlite dev db
+//
+// For each row in `organizations`: list its active + pending memberships,
+// fetch each member's user (concurrency 5), upsert user + membership through
+// the same guarded store the request path uses (`auth/workos-mirror-store.ts`).
+// Idempotent — the upserts refuse anything older than the stored WorkOS
+// `updatedAt`, so re-running is safe and never rewinds a fresher row.
+// Pass --dry-run to read and count without writing.
+// ---------------------------------------------------------------------------
+
+import { asc } from "drizzle-orm";
+import { drizzle } from "drizzle-orm/postgres-js";
+import { Effect } from "effect";
+import postgres from "postgres";
+import { WorkOS } from "@workos-inc/node";
+
+import { backfillWorkOsMirror } from "../src/auth/workos-mirror-backfill";
+import { makeWorkOsMirrorStore } from "../src/auth/workos-mirror-store";
+import { organizations } from "../src/db/schema";
+
+const dryRun = process.argv.includes("--dry-run");
+
+const connectionString = process.env.DATABASE_URL;
+if (!connectionString) {
+  console.error("DATABASE_URL is not set");
+  process.exit(1);
+}
+const apiKey = process.env.WORKOS_API_KEY;
+if (!apiKey) {
+  console.error("WORKOS_API_KEY is not set");
+  process.exit(1);
+}
+
+const usesLocalDatabase =
+  connectionString.includes("127.0.0.1") || connectionString.includes("localhost");
+
+const sql = postgres(connectionString, {
+  max: 1,
+  prepare: false,
+  ...(usesLocalDatabase ? {} : { ssl: "require" as const }),
+});
+const db = drizzle(sql);
+const workos = new WorkOS(apiKey);
+
+// The script boundary: raw SDK / driver promises lifted once, here.
+const fromPromise = <A>(fn: () => Promise<A>) =>
+  Effect.tryPromise({ try: fn, catch: (cause) => cause });
+
+await Effect.runPromise(
+  backfillWorkOsMirror(
+    {
+      listOrganizationIds: () =>
+        fromPromise(async () => {
+          const rows = await db
+            .select({ id: organizations.id })
+            .from(organizations)
+            .orderBy(asc(organizations.createdAt));
+          return rows.map((row) => row.id);
+        }),
+      listOrgMembers: (organizationId) =>
+        fromPromise(async () => {
+          const page = await workos.userManagement.listOrganizationMemberships({
+            organizationId,
+            statuses: ["active", "pending"],
+          });
+          return page.listMetadata.after ? page.autoPagination() : page.data;
+        }),
+      getUser: (userId) => fromPromise(() => workos.userManagement.getUser(userId)),
+    },
+    makeWorkOsMirrorStore(db),
+    { dryRun, log: (line) => console.log(line) },
+  ).pipe(Effect.ensuring(Effect.promise(() => sql.end({ timeout: 5 })))),
+);

apps/cloud/src/account/account-api.ts

@@ -9,6 +9,7 @@ import {
 
 import { ApiKeyService } from "../auth/api-keys";
 import { UserStoreService } from "../auth/context";
+import { WorkOsMirror } from "../auth/workos-mirror";
 import { sessionFromSealed, type Session } from "../auth/middleware";
 import { WorkOSClient } from "../auth/workos";
 import { AutumnService } from "../extensions/billing/service";
@@ -95,10 +96,13 @@ const AccountProviderMiddleware = HttpRouter.middleware<{ provides: AccountProvi
  * account service closes over the per-request postgres socket). `AutumnService`
  * (the seat-gate) stays a residual requirement, satisfied by the app `boot`.
  */
-export const workosAccountMiddleware = (rsLive: Layer.Layer<DbService | UserStoreService>) =>
-  AccountProviderMiddleware.combine(requestScopedMiddleware(rsLive)).layer;
+export const workosAccountMiddleware = (
+  rsLive: Layer.Layer<DbService | UserStoreService | WorkOsMirror>,
+) => AccountProviderMiddleware.combine(requestScopedMiddleware(rsLive)).layer;
 
-export const makeAccountApiLive = (rsLive: Layer.Layer<DbService | UserStoreService>) => {
+export const makeAccountApiLive = (
+  rsLive: Layer.Layer<DbService | UserStoreService | WorkOsMirror>,
+) => {
   // Cloud builds the WorkOS `AccountProvider` INSIDE the request body (so it
   // closes over the per-request postgres socket), so it can't be a self-
   // contained `Layer<AccountProvider>` — it combines its own middleware with

apps/cloud/src/account/org-api-key-revoke.node.test.ts

@@ -8,6 +8,7 @@ import { ApiKeyService, OrgApiKeyNotFound } from "../auth/api-keys";
 import { UserStoreService } from "../auth/context";
 import { ORG_SELECTOR_HEADER } from "../auth/organization";
 import { WorkOSClient, type WorkOSClientService } from "../auth/workos";
+import { WorkOsMirror } from "../auth/workos-mirror";
 import { AutumnService } from "../extensions/billing/service";
 import { AccountCaller, workosAccountProvider } from "./workos-account-service";
 
@@ -119,6 +120,16 @@ const stubUsers = Layer.succeed(UserStoreService)({
     ),
 });
 
+// Revoke changes no membership, so the mirror is never written.
+const stubMirror = Layer.succeed(WorkOsMirror)({
+  upsertUser: () => Effect.die("revoke does not write the membership mirror"),
+  upsertMembership: () => Effect.die("revoke does not write the membership mirror"),
+  deleteMembership: () => Effect.die("revoke does not write the membership mirror"),
+  deleteUser: () => Effect.die("revoke does not write the membership mirror"),
+  getCursor: () => Effect.die("revoke does not read the events cursor"),
+  setCursor: () => Effect.die("revoke does not move the events cursor"),
+});
+
 const stubAutumn = Layer.succeed(AutumnService)({
   use: () => Effect.die("revoke does not touch billing"),
   ensureCustomer: () => Effect.die("revoke does not touch billing"),
@@ -156,6 +167,7 @@ const providerWith = (accountId: string) => {
           Layer.mergeAll(
             stubWorkOS,
             stubUsers,
+            stubMirror,
             stubApiKeys,
             stubAutumn,
             Layer.succeed(AccountCaller)({ session: session(accountId) }),

apps/cloud/src/account/workos-account-service.ts

@@ -12,6 +12,7 @@ import { ApiKeyService } from "../auth/api-keys";
 import { UserStoreService } from "../auth/context";
 import type { Session } from "../auth/middleware";
 import { WorkOSClient } from "../auth/workos";
+import { WorkOsMirror, mirrorMembershipFromWorkOs } from "../auth/workos-mirror";
 import { ORG_SELECTOR_HEADER, authorizeOrganizationSelector } from "../auth/organization";
 import { AutumnService } from "../extensions/billing/service";
 import { forkReportMemberSeats } from "../extensions/billing/member-seats";
@@ -50,7 +51,7 @@ export class AccountCaller extends Context.Service<
 // (me / API keys) and `org/handlers.ts` (members / roles / invite / role /
 // name). Native WorkOS / store failures are mapped at this boundary onto the
 // neutral account errors so the shared UI sees one shape:
-//   WorkOSError | UserStoreError | ApiKeyManagementError → AccountError
+//   WorkOSError | UserStoreError | ApiKeyManagementError | WorkOsMirrorError → AccountError
 //   no organization in session                           → AccountNoOrganization
 //   not-an-admin / over-seat-limit / not-allowed         → AccountForbidden
 // ---------------------------------------------------------------------------
@@ -65,13 +66,17 @@ const toAccountError = () => Effect.fail(new AccountError({ message: "Account re
 export const workosAccountProvider: Layer.Layer<
   AccountProvider,
   never,
-  WorkOSClient | UserStoreService | ApiKeyService | AutumnService | AccountCaller
+  WorkOSClient | UserStoreService | WorkOsMirror | ApiKeyService | AutumnService | AccountCaller
 > = Layer.effect(AccountProvider)(
   Effect.gen(function* () {
     const workos = yield* WorkOSClient;
     const apiKeys = yield* ApiKeyService;
     const autumn = yield* AutumnService;
     const users = yield* UserStoreService;
+    // Membership writes below go to WorkOS FIRST (the authority), then are
+    // written through to the local mirror so the member list and the seat
+    // count read the change without waiting for the Events reconciler.
+    const mirror = yield* WorkOsMirror;
 
     // The caller, resolved once per request by the cookie-only session
     // middleware (account-api.ts) — the same credential `SessionAuthLive`
@@ -365,6 +370,9 @@ export const workosAccountProvider: Layer.Layer<
           yield* workos
             .deleteOrgMembership(membershipId)
             .pipe(Effect.catchTag("WorkOSError", toAccountError));
+          yield* mirror
+            .deleteMembership(membershipId)
+            .pipe(Effect.catchTag("WorkOsMirrorError", toAccountError));
           yield* forkReportMemberSeats(org.id).pipe(Effect.provideContext(ctx));
           return { success: true };
         }),
@@ -374,9 +382,12 @@ export const workosAccountProvider: Layer.Layer<
           const { session, org } = yield* requireOrganization(headers);
           yield* requireAdmin(session.accountId, org.id);
           yield* assertMembershipInOrg(org.id, membershipId);
-          yield* workos
+          const updated = yield* workos
             .updateOrgMembershipRole(membershipId, roleSlug)
             .pipe(Effect.catchTag("WorkOSError", toAccountError));
+          yield* mirror
+            .upsertMembership(mirrorMembershipFromWorkOs(updated))
+            .pipe(Effect.catchTag("WorkOsMirrorError", toAccountError));
           return { success: true };
         }),
 

apps/cloud/src/api/layers.ts

@@ -56,7 +56,9 @@ export const BootSharedServices = Layer.mergeAll(
 // `AutumnService.Default` is provided HERE because the `createOrganization`
 // handler reads it for the free-organizations-per-user limit gate — one of the
 // few app-only billing touchpoints. (It is NOT on the neutral boot core.)
-export const makeNonProtectedApiLive = (rsLive: Layer.Layer<DbService | UserStoreService>) =>
+export const makeNonProtectedApiLive = (
+  rsLive: Layer.Layer<DbService | UserStoreService | WorkOsMirror>,
+) =>
   HttpApiBuilder.layer(NonProtectedApi).pipe(
     Layer.provide(Layer.mergeAll(CloudAuthPublicHandlers, CloudSessionAuthHandlers)),
     Layer.provide(requestScopedMiddleware(rsLive).layer),

apps/cloud/src/api/router.ts

@@ -4,6 +4,7 @@ import { HttpRouter } from "effect/unstable/http";
 import { RouterConfigLive, requestScopedMiddleware } from "@executor-js/api/server";
 
 import { UserStoreService } from "../auth/context";
+import { WorkOsMirror } from "../auth/workos-mirror";
 import { DbService } from "../db/db";
 import { makeAccountApiLive } from "../account/account-api";
 
@@ -29,7 +30,9 @@ import { makeProtectedApiLive } from "./protected";
 // so tests can substitute a counting fake for `DbService.Live` and
 // assert per-request semantics — see
 // `apps/cloud/src/api.request-scope.node.test.ts`.
-export const makeApiLive = (requestScopedLive: Layer.Layer<DbService | UserStoreService>) => {
+export const makeApiLive = (
+  requestScopedLive: Layer.Layer<DbService | UserStoreService | WorkOsMirror>,
+) => {
   const BillingRoutesLive = AutumnRoutesLive.pipe(
     Layer.provide(requestScopedMiddleware(requestScopedLive).layer),
   );

apps/cloud/src/auth/api.ts

@@ -1,6 +1,6 @@
 import { HttpApiEndpoint, HttpApiGroup } from "effect/unstable/httpapi";
 import { Schema } from "effect";
-import { UserStoreError, WorkOSError } from "./errors";
+import { UserStoreError, WorkOSError, WorkOsMirrorError } from "./errors";
 import { NoOrganization } from "@executor-js/api/server";
 import { SessionAuth } from "./middleware";
 
@@ -171,7 +171,9 @@ export const AUTH_PATHS = {
   callback: "/api/auth/callback",
 } as const;
 
-const AuthErrors = [UserStoreError, WorkOSError] as const;
+// The login callback and the org handlers feed the membership mirror, so a
+// mirror write failure is one of their wire errors (same 500 as a store failure).
+const AuthErrors = [UserStoreError, WorkOSError, WorkOsMirrorError] as const;
 const McpApprovalErrors = [
   NoOrganization,
   McpExecutionNotFoundError,

apps/cloud/src/auth/errors.ts

@@ -38,6 +38,27 @@ export class UserStoreError extends Schema.TaggedErrorClass<UserStoreError>()(
   }
 }
 
+/**
+ * The public failure of every cloud membership-mirror write (`WorkOsMirror`).
+ * Same two diagnosable fields as `UserStoreError` — which mirror call failed,
+ * and how — classified from the driver cause the same way. Declared here,
+ * beside `UserStoreError`, because the auth API (`auth/api.ts`, in the SPA
+ * bundle) names it on the wire for the login and org handlers that feed the
+ * mirror; the service itself lives in `workos-mirror.ts`.
+ */
+export class WorkOsMirrorError extends Schema.TaggedErrorClass<WorkOsMirrorError>()(
+  "WorkOsMirrorError",
+  {
+    operation: Schema.String,
+    reason: Schema.Literals(USER_STORE_FAILURE_REASONS),
+  },
+  { httpApiStatus: 500 },
+) {
+  override get message(): string {
+    return `workos mirror ${this.operation} failed: ${this.reason}`;
+  }
+}
+
 /** Reasons a retry can plausibly clear: the query never reached a healthy
  *  server. A `query` failure is deterministic and must not be retried. */
 export const isTransientUserStoreReason = (reason: UserStoreFailureReason): boolean =>