// ---------------------------------------------------------------------------// Out-of-band reconciler run: replay the WorkOS Events API into the// membership mirror from the persisted cursor until the stream is drained,// over a plain postgres.js connection under bun — the SAME replay the// Worker's every-minute cron runs (`src/auth/workos-events-replay.ts`).//// bun run db:drain-workos-events:prod # op run --env-file=.env.production//// Exists for the deploy gate (`scripts/ensure-workos-mirror-ready.ts`): the// build that authorizes from the mirror trusts it only once the reconciler// has drained the stream recently, and the gate must be able to MAKE that// true itself rather than wait for a cron that may not be deployed yet —// otherwise the reconciler build could only ever ship ahead of the gated// one, by hand. Safe to run beside a live cron: a page is applied under the// cursor's compare-and-set, so whichever run loses the stream writes// nothing and stops. Runs until the stream is drained or another run owns// it; a page budget bounds one pass, so a long backlog takes several. Exits// 0 on a drain, 1 otherwise, with the reason.// ---------------------------------------------------------------------------import { drizzle } from "drizzle-orm/postgres-js";import { Effect, Option } from "effect";import postgres from "postgres";import { WorkOS } from "@workos-inc/node";import { makeUserStore } from "../src/auth/user-store";import { replayWorkOsEvents, type WorkOsEventsSyncReport } from "../src/auth/workos-events-replay";import { makeWorkOsMirrorStore } from "../src/auth/workos-mirror-store";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);const users = makeUserStore(db);// The script boundary: raw SDK / driver promises lifted once, here. Only a// 404 is the deterministic "gone" the replay acts on; every other failure// fails the pass, as in the Worker (`src/auth/workos-events-sync.ts`).const fromPromise = <A>(fn: () => Promise<A>) => Effect.tryPromise({ try: fn, catch: (cause) => cause });const isNotFound = (cause: unknown): boolean => typeof cause === "object" && cause !== null && "status" in cause && (cause as { readonly status: unknown }).status === 404;const noneWhenGone = <A>(fn: () => Promise<A>) => Effect.tryPromise({ try: fn, catch: (cause) => cause }).pipe( Effect.map(Option.some), Effect.catch((cause) => isNotFound(cause) ? Effect.succeed(Option.none<A>()) : Effect.fail(cause), ), );// One pass is bounded by the replay's page budget; loop until the stream// is drained, another run owns it, or the backfill has not run.const MAX_PASSES = 50;const drain = Effect.gen(function* () { const deps = { source: { listEvents: (options: Parameters<typeof workos.events.listEvents>[0]) => fromPromise(async () => { const page = await workos.events.listEvents({ ...options, events: [...options.events], }); return { data: page.data, after: page.listMetadata.after ?? null }; }), getOrganization: (organizationId: string) => noneWhenGone(() => workos.organizations.getOrganization(organizationId)), getUser: (userId: string) => noneWhenGone(() => workos.userManagement.getUser(userId)), }, store: { getOrganization: (organizationId: string) => fromPromise(() => users.getOrganization(organizationId)), upsertOrganization: (organization: Parameters<typeof users.upsertOrganization>[0]) => fromPromise(() => users.upsertOrganization(organization)), getAccount: (accountId: string) => fromPromise(() => users.getAccount(accountId)), }, mirror: makeWorkOsMirrorStore(db), }; let last: WorkOsEventsSyncReport | null = null; for (let pass = 0; pass < MAX_PASSES; pass++) { const report = yield* replayWorkOsEvents(deps); console.log( `[drain-events] pass ${pass + 1}: ${report.pages} page(s), ${report.events} event(s), ` + `${report.applied} applied, ${report.stale} stale, ${report.absent} absent — ${report.stopped}`, ); last = report; if (report.stopped !== "page_budget") break; } return last;});const report = await Effect.runPromise( drain.pipe(Effect.ensuring(Effect.promise(() => sql.end({ timeout: 5 })))),);if (report === null || report.stopped !== "drained") { console.error( `[drain-events] the events stream was not drained: ${report?.stopped ?? "no pass ran"}` + (report?.stopped === "awaiting_backfill" ? " (run scripts/backfill-workos-mirror.ts first)" : report?.stopped === "cursor_contended" ? " (another run owns the stream; rerun once it finishes)" : ""), ); process.exit(1);}
// ---------------------------------------------------------------------------
// Out-of-band reconciler run: replay the WorkOS Events API into the
// membership mirror from the persisted cursor until the stream is drained,
// over a plain postgres.js connection under bun — the SAME replay the
// Worker's every-minute cron runs (`src/auth/workos-events-replay.ts`).
//
// bun run db:drain-workos-events:prod # op run --env-file=.env.production
//
// Exists for the deploy gate (`scripts/ensure-workos-mirror-ready.ts`): the
// build that authorizes from the mirror trusts it only once the reconciler
// has drained the stream recently, and the gate must be able to MAKE that
// true itself rather than wait for a cron that may not be deployed yet —
// otherwise the reconciler build could only ever ship ahead of the gated
// one, by hand. Safe to run beside a live cron: a page is applied under the
// cursor's compare-and-set, so whichever run loses the stream writes
// nothing and stops. Runs until the stream is drained or another run owns
// it; a page budget bounds one pass, so a long backlog takes several. Exits
// 0 on a drain, 1 otherwise, with the reason.
// ---------------------------------------------------------------------------
import { drizzle } from "drizzle-orm/postgres-js";
import { Effect, Option } from "effect";
import postgres from "postgres";
import { WorkOS } from "@workos-inc/node";
import { makeUserStore } from "../src/auth/user-store";
import { replayWorkOsEvents, type WorkOsEventsSyncReport } from "../src/auth/workos-events-replay";
import { makeWorkOsMirrorStore } from "../src/auth/workos-mirror-store";
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);
const users = makeUserStore(db);
// The script boundary: raw SDK / driver promises lifted once, here. Only a
// 404 is the deterministic "gone" the replay acts on; every other failure
// fails the pass, as in the Worker (`src/auth/workos-events-sync.ts`).
const fromPromise = <A>(fn: () => Promise<A>) =>
Effect.tryPromise({ try: fn, catch: (cause) => cause });
const isNotFound = (cause: unknown): boolean =>
typeof cause === "object" &&
cause !== null &&
"status" in cause &&
(cause as { readonly status: unknown }).status === 404;
const noneWhenGone = <A>(fn: () => Promise<A>) =>
Effect.tryPromise({ try: fn, catch: (cause) => cause }).pipe(
Effect.map(Option.some),
Effect.catch((cause) =>
isNotFound(cause) ? Effect.succeed(Option.none<A>()) : Effect.fail(cause),
),
);
// One pass is bounded by the replay's page budget; loop until the stream
// is drained, another run owns it, or the backfill has not run.
const MAX_PASSES = 50;
const drain = Effect.gen(function* () {
const deps = {
source: {
listEvents: (options: Parameters<typeof workos.events.listEvents>[0]) =>
fromPromise(async () => {
const page = await workos.events.listEvents({
...options,
events: [...options.events],
});
return { data: page.data, after: page.listMetadata.after ?? null };
}),
getOrganization: (organizationId: string) =>
noneWhenGone(() => workos.organizations.getOrganization(organizationId)),
getUser: (userId: string) => noneWhenGone(() => workos.userManagement.getUser(userId)),
},
store: {
getOrganization: (organizationId: string) =>
fromPromise(() => users.getOrganization(organizationId)),
upsertOrganization: (organization: Parameters<typeof users.upsertOrganization>[0]) =>
fromPromise(() => users.upsertOrganization(organization)),
getAccount: (accountId: string) => fromPromise(() => users.getAccount(accountId)),
},
mirror: makeWorkOsMirrorStore(db),
};
let last: WorkOsEventsSyncReport | null = null;
for (let pass = 0; pass < MAX_PASSES; pass++) {
const report = yield* replayWorkOsEvents(deps);
console.log(
`[drain-events] pass ${pass + 1}: ${report.pages} page(s), ${report.events} event(s), ` +
`${report.applied} applied, ${report.stale} stale, ${report.absent} absent — ${report.stopped}`,
);
last = report;
if (report.stopped !== "page_budget") break;
}
return last;
});
const report = await Effect.runPromise(
drain.pipe(Effect.ensuring(Effect.promise(() => sql.end({ timeout: 5 })))),
);
if (report === null || report.stopped !== "drained") {
console.error(
`[drain-events] the events stream was not drained: ${report?.stopped ?? "no pass ran"}` +
(report?.stopped === "awaiting_backfill"
? " (run scripts/backfill-workos-mirror.ts first)"
: report?.stopped === "cursor_contended"
? " (another run owns the stream; rerun once it finishes)"
: ""),
);
process.exit(1);
}