// ---------------------------------------------------------------------------// The membership mirror's WRITE store — the Drizzle queries behind// `WorkOsMirror`, plus the converters from WorkOS SDK payloads to mirror rows.//// Kept free of `cloudflare:workers` (no `DbService`, no `env`) so the one-off// backfill (`scripts/backfill-workos-mirror.ts`) can run the SAME upserts over// a plain postgres.js connection under bun. The request-scoped service that// wraps this store is `workos-mirror.ts`.//// Every write is idempotent and out-of-order safe. Both upserts carry the// WorkOS `updatedAt` of their payload and refuse to overwrite a row whose// stored `workos_updated_at` is newer, so a replayed or late-arriving event// can never regress the mirror. A delete never drops the row: it TOMBSTONES// it (`status = 'inactive'`, `deleted_at` set) — and INSERTS the tombstone// when the row is not there yet — so a feeder that fetched the membership// before the deletion and writes it after (a login, the backfill) finds the// tombstone instead of reinstating access. The tombstone is protected by// IDENTITY, not by time: WorkOS never reuses a deleted `om_…` id, so a// payload naming the tombstoned id is refused whatever it is stamped — a// role change issued before the removal and delivered after it is stamped// NEWER than the row and would beat any timestamp guard — while a payload// naming a DIFFERENT id for the same (account, organization) is the member// re-added in WorkOS, a replacement, and takes the row over under the usual// `updatedAt` rule (WorkOS creates it after the old one is deleted, so it is// always the newer). A tombstone keeps the stamp the row holds, the deleted// membership's last reported state — never a local clock: read after WorkOS// answered, it could post-date a replacement created meanwhile and refuse// it for good. `deleted_at` is the deletion instant when the caller holds// one (an event's `createdAt`, a scan's listing time) and `now()` otherwise;// it records WHEN, it orders nothing. A membership WorkOS merely// deactivated (`status = 'inactive'`, no `deleted_at`) is not a tombstone:// WorkOS can reactivate it under the same id, and the `updatedAt` rule// orders that as any other update. The row tombstone can only speak for the// id the row happens to hold, so every delete ALSO records the deleted id in// `membership_tombstones`, a ledger keyed by the WorkOS id alone, and every// membership write consults it first: that covers the case the row cannot —// membership A replaced by B in WorkOS before the mirror saw either, B then// deleted. The delete finds a row holding A (not B's to tombstone) and would// otherwise leave nothing behind; a later scan that still lists B would then// insert it live, stamped after A and under another id, exactly what the row// guard lets through. With the ledger the delete is recorded whatever the// row holds, and B's payload is refused by identity. A scan records the ids// it tombstones the same way. A deleted USER is protected by identity// too: `deleteUser` leaves the account row behind as a tombstone (profile// cleared, stamped with the deletion), and no membership naming that account// is written again, however the payload is stamped — WorkOS never reuses a// user id, and a membership the mirror had not seen has no row of its own// for a guard to refuse the insert against. The reconciler applies a page// of events and advances the cursor in ONE transaction that// compare-and-sets the cursor first (`applyPage`), so a run that has lost// the stream to another run writes nothing — the `updatedAt` guard alone// cannot stop it re-applying a page the leading run has moved past.//// A backfill SCAN of one organization (its full membership listing, taken// at one instant) is applied the same way: ONE transaction that first// compare-and-sets the organization's `backfilled_at` to the listing's// instant (`applyOrganizationScan`), so the row stays locked until commit,// two overlapping scans serialize on it, and the one whose listing is older// than the recorded one writes NOTHING. The `updatedAt` guard alone cannot// order two scans: a scan that listed a membership, stalled, and resumed// after a later scan had found it gone would insert it live — the later scan// left no tombstone, because the row was not there to tombstone. For the// same reason every OTHER membership write is ordered against the scan too:// a payload stamped before the organization's `backfilled_at` is refused// unless the row already holds it — judged inside the write's own// transaction, reading the organization row `FOR SHARE`, so a write that// races a scan waits for the scan's commit and sees the mark it set, and// never inserts a membership the scan has just proved revoked. A completed// scan is the full listing as// of that instant, so a membership it did not contain but an older payload// still names (a login whose list was fetched before the revocation and// written after the scan) was revoked before the scan — and that revocation// predates the events replay boundary, so nothing would ever tombstone the// reinstated row. The scan's own writes are the one exception: they are the// listing that sets the mark.//// The events replay boundary (`workos_sync.range_start`) is written ONCE, by// the first backfill run BEFORE its first listing, and never advanced: a// scan refreshes memberships only, so an organization rename or user// deletion after that instant — between two runs, or between a run that// failed part-way and its retry — is covered by the events stream alone,// and moving the boundary past it would skip it for good.// --------------------------------------------------------------------------- import { and, eq, isNotNull, isNull, lt, ne, notInArray, or, sql } from "drizzle-orm";import type { AnyPgColumn } from "drizzle-orm/pg-core";import { Data, Effect, Option } from "effect"; import type { MemberStatus } from "@executor-js/api/server"; import { accounts, membershipTombstones, memberships, organizations, workosSync,} from "../db/schema";import type { DrizzleDb } from "../db/db";import { insertOrganization, organizationAcceptsName } from "./user-store";import { WorkOsMirrorError, tryPromiseService, userStoreReasonFromCause, withServiceLogging,} from "./errors"; /** A WorkOS user, as the mirror stores it. `updatedAt` is WorkOS's own. */export interface WorkOsMirrorUser { readonly id: string; readonly email: string; readonly firstName: string | null; readonly lastName: string | null; readonly avatarUrl: string | null; readonly lastSignInAt: Date | null; readonly updatedAt: Date;} /** * A WorkOS organization membership, as the mirror stores it. `id` is the * WorkOS `om_…`; `accountId` the WorkOS user id; `updatedAt` is WorkOS's own. * The organization row must already be mirrored (`upsertOrganization`) — a * membership of an unknown org is a `query` failure, not a silent skip. */export interface WorkOsMirrorMembership { readonly id: string; readonly accountId: string; readonly organizationId: string; readonly role: string; readonly status: MemberStatus; readonly updatedAt: Date;} /** * What identifies a membership to a delete: the WorkOS `om_…` id AND the * (account, organization) pair it belongs to. The pair is the row's key, * and a delete must be able to mint the row as a tombstone when the mirror * has not seen the membership yet — an id alone cannot. */export type WorkOsMirrorMembershipRef = Pick< WorkOsMirrorMembership, "id" | "accountId" | "organizationId">; /** One member as a backfill scan lists it: the membership and its user. */export interface WorkOsScannedMember { readonly user: WorkOsMirrorUser; readonly membership: WorkOsMirrorMembership;} /** * One organization's full membership listing, as `applyOrganizationScan` * applies it. `listedAt` is the instant the listing was taken — BEFORE the * WorkOS read, so no change can fall between the instant and the listing — * and is the tombstone time for what the listing no longer contains, the * cut-off for what may be tombstoned at all, and the organization's new * `backfilled_at`. */export interface WorkOsOrganizationScan { readonly organizationId: string; readonly listedAt: Date; readonly members: readonly WorkOsScannedMember[];} /** What an applied scan wrote: the upserts the `updatedAt` guard let through, and the tombstones. */export interface WorkOsOrganizationScanWrites { readonly usersWritten: number; readonly membershipsWritten: number; readonly membershipsTombstoned: number;} /** * One write of a reconciler page, applied by `applyPage` inside the page's * transaction. The reconciler plans a page into these BEFORE the transaction * opens, so every WorkOS read (resolving an organization the mirror has never * seen) is done by then: the transaction holds the mirror's single connection * and must not wait on the network. */export type WorkOsMirrorWrite = Data.TaggedEnum<{ readonly UpsertUser: { readonly user: WorkOsMirrorUser }; readonly UpsertMembership: { readonly membership: WorkOsMirrorMembership }; /** * A membership together with its member's profile, read from WorkOS at * plan time because the mirror held no profile for the account * (`workos-events-sync.ts`): the user is upserted first, under the usual * guard, then the membership. The outcome is the membership's. */ readonly UpsertMember: { readonly user: WorkOsMirrorUser; readonly membership: WorkOsMirrorMembership; }; /** Tombstone a membership as of `deletedAt` (the event's `createdAt`). */ readonly DeleteMembership: { readonly membership: WorkOsMirrorMembershipRef; readonly deletedAt: Date; }; /** Tombstone a user and their memberships as of `deletedAt`. */ readonly DeleteUser: { readonly accountId: string; readonly deletedAt: Date }; /** * Rename an organization the mirror already holds — never inserts one — * from a payload stamped `updatedAt` (the WorkOS organization's own), under * the same name guard every feeder applies (`organizationAcceptsName`). */ readonly RenameOrganization: { readonly organizationId: string; readonly name: string; readonly updatedAt: Date; }; /** * Mark an organization deleted as of `deletedAt` (the event's * `createdAt`) — minting the row as a TOMBSTONE, named `name`, when the * mirror has never seen the organization, so a feeder still holding a * membership of it (a login that stalled across the deletion) finds the * tombstone and cannot mint the organization live. Never purges tenant * data (that is cloud's own flow, `db/org-deletion.ts`). An earlier mark * stands (`absent`). */ readonly MarkOrganizationDeleted: { readonly organizationId: string; readonly name: string; readonly deletedAt: Date; };}>;export const WorkOsMirrorWrite = Data.taggedEnum<WorkOsMirrorWrite>(); /** * What one write did: a row was written, renamed, or tombstoned (`applied`); * the `updatedAt` guard refused an older payload (`stale`); or a delete found * its row already tombstoned or superseded by a newer membership (a * replayed delete), or a rename found no live organization row to change — * the mirror has never seen it, or it is marked deleted — or a deletion * mark found the organization already marked (`absent`). */export type WorkOsMirrorWriteOutcome = "applied" | "stale" | "absent"; export interface WorkOsMirrorShape { /** * Insert or refresh a user row. `false` when the payload was refused and * the row left untouched: the stored row is newer than `updatedAt`, or is * a deletion tombstone stamped at `updatedAt` or later. */ readonly upsertUser: (user: WorkOsMirrorUser) => Effect.Effect<boolean, WorkOsMirrorError>; /** * Insert or refresh a membership row, minting the bare account row first so * the foreign key holds when the membership arrives before its user. `false` * when the payload was refused: the stored row is newer than `updatedAt`, * or is a deletion tombstone of THIS membership id, or THIS membership id * is in the deletion ledger (`membership_tombstones`, written by every * delete and every scan tombstone, whatever row it found) — a deleted * WorkOS id never returns, however the payload is stamped; a payload under * another id is a replacement and is ordered by `updatedAt` against the * row's stamp — or the account is a deletion tombstone (`deleteUser`), whatever * the payload is stamped: WorkOS never reuses a user id, so a deleted user * has no memberships to mirror — or the organization is marked deleted * (`organizations.deleted_at`), in which case nothing is written at all: * no membership of a deleted organization is ever mirrored — or the * payload is stamped BEFORE the organization's last full scan * (`organizations.backfilled_at`): the scan listed everything WorkOS held * at that instant, so a membership it wrote already carries a stamp at * least this new (the write would change nothing) and one it did not * write was gone by then and must not come back from a list fetched * earlier. Only the scan itself writes past that mark * (`applyOrganizationScan`). The organization checks and the write run in * ONE transaction that holds the organization row `FOR SHARE`, so a scan * claiming the row at the same time is waited for and its mark seen — and * the account row `FOR SHARE` too, so a `deleteUser` tombstoning it at the * same time is waited for and its tombstone seen. */ readonly upsertMembership: ( membership: WorkOsMirrorMembership, ) => Effect.Effect<boolean, WorkOsMirrorError>; /** * Tombstone the membership as deleted: the row stays (or is minted, when * the mirror has not seen the membership yet), `inactive`, carrying the * deleted WorkOS id with `deleted_at` set. The tombstone is protected by * that id: `upsertMembership` refuses every later payload naming it, * however stamped, and accepts only a REPLACEMENT under another id. The * row's `workos_updated_at` is left as it is — the deleted membership's * last reported state, which a replacement is always stamped after; a * local clock read after WorkOS answered could post-date a replacement * created meanwhile and must never become the row's stamp. `deletedAt` * is the deletion instant when the caller holds one (a deletion event's * `createdAt`, a scan's listing time) or `null` when it holds none * (WorkOS answers a delete with no time): `deleted_at` then takes the * current time. It records when the membership was deleted; it orders * nothing. Matches the row by IDENTITY: `false` when the row is already * tombstoned (a replayed delete), or holds ANOTHER membership id — the * member re-added in WorkOS under a new id, which stands whether the * replacement was mirrored before or after this delete. In EVERY case the * deleted id is recorded in the deletion ledger (`membership_tombstones`), * so a membership the mirror never held under its own id (replaced and * deleted in WorkOS before the mirror saw it) cannot be inserted live by a * scan that listed it before the deletion: `true` when the row was * tombstoned OR the id was newly recorded. The organization row must * already be mirrored, as for `upsertMembership`. */ readonly deleteMembership: ( membership: WorkOsMirrorMembershipRef, deletedAt: Date | null, ) => Effect.Effect<boolean, WorkOsMirrorError>; /** * Tombstone a deleted WorkOS user: every membership of the account is * tombstoned as by `deleteMembership` (as of `deletedAt`; a deleted user's * membership ids never return), and the account row is kept (it anchors * foreign keys) — or minted, when the mirror has not seen the user yet — * with its profile cleared and stamped `deletedAt`, so a stale user * payload cannot restore it. `false` when the account already carries * this tombstone or a later one (a replayed delete). ONE transaction that * locks the account row first (`FOR NO KEY UPDATE`) and holds it until * the memberships are tombstoned: `upsertMembership` reads that row `FOR * SHARE` before it inserts, so a membership write racing the deletion * waits for it and sees the tombstone, or committed first and is * tombstoned here — never a live membership left behind for a deleted * user. */ readonly deleteUser: ( accountId: string, deletedAt: Date, ) => Effect.Effect<boolean, WorkOsMirrorError>; /** The id of the last WorkOS event applied, or `null` before the first run. */ readonly getCursor: () => Effect.Effect<string | null, WorkOsMirrorError>; /** * Apply one reconciler page atomically: in a single transaction, * compare-and-set the cursor from `prev` (`null` = no cursor yet) to * `next`, and only if that succeeded apply `writes` in order. The cursor * row stays locked until commit, so two runs applying pages serialize on * it and the one whose `prev` is stale sees the moved cursor and writes * nothing: `None` means another run owns the stream and the caller must * stop. `Some` carries one outcome per write, in order. An empty `writes` * is a bare cursor advance. */ readonly applyPage: ( prev: string | null, next: string, writes: readonly WorkOsMirrorWrite[], ) => Effect.Effect<Option.Option<readonly WorkOsMirrorWriteOutcome[]>, WorkOsMirrorError>; /** * Apply one organization's backfill scan — the memberships (with their * users) a WorkOS listing taken at `listedAt` contained — atomically: in a * single transaction, compare-and-set the organization's `backfilled_at` * forward to `listedAt`, and only if that succeeded upsert every listed * user and membership and tombstone (as `deleteMembership` does, at * `listedAt`) every membership of the organization the listing did NOT * contain: the rows whose WorkOS id is not among the listed ones AND whose * stamp is older than the listing. A row stamped at or after `listedAt` * was written after the listing was taken (it could not be in it) and is * left alone — its own event lands through the reconciler; a tombstone * here would beat that event. Rows already tombstoned are left alone. * * The organization row stays locked until commit, so two overlapping * scans serialize on it and the one whose listing is older than (or the * same instant as) the recorded one writes nothing: `None` — the mirror * already holds the organization as of a later listing, and a membership * that listing no longer contained must not be inserted from this one. * `None` too for an organization the mirror does not hold or has marked * deleted: there is nothing to scan into. `Some` carries what was written. */ readonly applyOrganizationScan: ( scan: WorkOsOrganizationScan, ) => Effect.Effect<Option.Option<WorkOsOrganizationScanWrites>, WorkOsMirrorError>; /** * The Events API replay boundary: the instant the FIRST one-off backfill * run began reading WorkOS, or `null` if none has started. The * reconciler's first run (no cursor yet) reads the stream from here — the * backfill covers everything before it — and without a boundary it must * not guess. */ readonly replayBoundary: () => Effect.Effect<Date | null, WorkOsMirrorError>; /** * Record the replay boundary, ONCE: `at` is the instant a backfill run * began reading WorkOS, and it is kept only when no boundary is recorded * yet — `true` when this call recorded it. A later run never moves it: the * backfill refreshes memberships and tombstones only, not organization * names or deleted users' profiles, so a change after the first boundary * is covered only by the events stream, which must still be read from * there. Written BEFORE the run's first listing, so a run that fails * part-way leaves the boundary standing and its retry keeps it — a * `user.deleted` between the attempts stays inside the replay. Never * touches the cursor: a stream already being followed keeps its position, * and the boundary is then unused. */ readonly setReplayBoundary: (at: Date) => Effect.Effect<boolean, WorkOsMirrorError>; /** * When a backfill run first wrote EVERY live organization * (`workos_sync.backfill_completed_at`), or `null` while none has. The * first half of the mirror's READINESS for authorization: until a run has * covered every organization, the mirror may lack members who have not * signed in since it shipped, and a membership check read from it would * deny them. The other half is the events reconciler being caught up, * which its cursor row reports. */ readonly backfillCompletedAt: () => Effect.Effect<Date | null, WorkOsMirrorError>; /** * Record that a backfill run has written every live organization, as of * `at` — ONCE: a later completed run keeps the first mark (`false`), so * readiness never flips back. Written only after the last organization * was applied (or refused in favour of a later listing), so a run that * fails part-way records nothing here. Mints the events row when absent, * as `setReplayBoundary` does, and touches neither the cursor nor the * boundary. */ readonly markBackfillCompleted: (at: Date) => Effect.Effect<boolean, WorkOsMirrorError>; /** * When a reconciler run last read the events stream to its end * (`workos_sync.drained_at`), or `null` if none has. The second half of * the mirror's readiness for authorization: a mirror whose reconciler * has not caught up within the lag budget may still hold a membership * WorkOS has since revoked. */ readonly drainedAt: () => Effect.Effect<Date | null, WorkOsMirrorError>; /** * Record that a reconciler run read the stream to its end at `at`. Moves * the mark forward only — a run that finished after a later one keeps the * later mark — and only on the row a run already owns: the events row is * minted by the boundary or the first cursor advance, so a missing row * means nothing was drained and nothing is written (`false`). */ readonly markDrained: (at: Date) => Effect.Effect<boolean, WorkOsMirrorError>; /** * When the organization's membership list was last FULLY scanned from * WorkOS (`backfillOrganization` in workos-mirror-backfill.ts), or `null` * if it never was — or the organization is not mirrored. Until it has been, * the mirror may hold only the members login and write-through happened to * record, so a member count read from it is PARTIAL: every seat gate reads * this first and scans the organization when it is `null`. Per * organization, never database-wide, so an organization mirrored after a * backfill ran (lazily, or by a sign-in) is never mistaken for a scanned * one. */ readonly organizationBackfilledAt: ( organizationId: string, ) => Effect.Effect<Date | null, WorkOsMirrorError>;} // ---------------------------------------------------------------------------// SDK payload → mirror row. The feeders (login callback, write-through, the// backfill, the Events reconciler) all hand the mirror WorkOS objects; this is// the one place their field names and ISO timestamps are translated. Typed// structurally (the fields actually read) so the SDK's `User` /// `OrganizationMembership`, an event payload, and a test fixture all fit.// --------------------------------------------------------------------------- /** The WorkOS user fields the mirror reads. `User` from the SDK satisfies it. */export interface WorkOsUserPayload { readonly id: string; readonly email: string; readonly firstName: string | null; readonly lastName: string | null; readonly profilePictureUrl: string | null; readonly lastSignInAt: string | null; readonly updatedAt: string;} /** * The WorkOS membership fields the mirror reads. `OrganizationMembership` from * the SDK satisfies it; its `status` is exactly the mirror's `MemberStatus`. */export interface WorkOsMembershipPayload { readonly id: string; readonly userId: string; readonly organizationId: string; readonly role: { readonly slug: string }; readonly status: MemberStatus; readonly updatedAt: string;} /** Translate a WorkOS user payload to the row `upsertUser` stores. */export const mirrorUserFromWorkOs = (user: WorkOsUserPayload): WorkOsMirrorUser => ({ id: user.id, email: user.email, firstName: user.firstName, lastName: user.lastName, avatarUrl: user.profilePictureUrl, lastSignInAt: user.lastSignInAt === null ? null : new Date(user.lastSignInAt), updatedAt: new Date(user.updatedAt),}); /** Translate a WorkOS membership payload to the row `upsertMembership` stores. */export const mirrorMembershipFromWorkOs = ( membership: WorkOsMembershipPayload,): WorkOsMirrorMembership => ({ id: membership.id, accountId: membership.userId, organizationId: membership.organizationId, role: membership.role.slug, status: membership.status, updatedAt: new Date(membership.updatedAt),}); /** * The `workos_sync` row of the one WorkOS events stream the reconciler * follows. A row id rather than a singleton table so a second stream * (another WorkOS environment, a replay) can be added without a schema * change; the same row carries the backfill's replay boundary and * completion mark. Not a `db/schema.ts` export: that module's exports are * enumerated as tables by the purge-coverage test. */export const WORKOS_EVENTS_STREAM_ID = "events"; // A raw `sql` fragment binds a Date without the column's driver mapping, so// every instant below is passed as ISO text and cast.Lines 1–500 of 1232. Load more
// --------------------------------------------------------------------------- // The membership mirror's WRITE store — the Drizzle queries behind // `WorkOsMirror`, plus the converters from WorkOS SDK payloads to mirror rows. // // Kept free of `cloudflare:workers` (no `DbService`, no `env`) so the one-off // backfill (`scripts/backfill-workos-mirror.ts`) can run the SAME upserts over // a plain postgres.js connection under bun. The request-scoped service that // wraps this store is `workos-mirror.ts`. // // Every write is idempotent and out-of-order safe. Both upserts carry the // WorkOS `updatedAt` of their payload and refuse to overwrite a row whose // stored `workos_updated_at` is newer, so a replayed or late-arriving event // can never regress the mirror. A delete never drops the row: it TOMBSTONES // it (`status = 'inactive'`, `deleted_at` set) — and INSERTS the tombstone // when the row is not there yet — so a feeder that fetched the membership // before the deletion and writes it after (a login, the backfill) finds the // tombstone instead of reinstating access. The tombstone is protected by // IDENTITY, not by time: WorkOS never reuses a deleted `om_…` id, so a // payload naming the tombstoned id is refused whatever it is stamped — a // role change issued before the removal and delivered after it is stamped // NEWER than the row and would beat any timestamp guard — while a payload // naming a DIFFERENT id for the same (account, organization) is the member // re-added in WorkOS, a replacement, and takes the row over under the usual // `updatedAt` rule (WorkOS creates it after the old one is deleted, so it is // always the newer). A tombstone keeps the stamp the row holds, the deleted // membership's last reported state — never a local clock: read after WorkOS // answered, it could post-date a replacement created meanwhile and refuse // it for good. `deleted_at` is the deletion instant when the caller holds // one (an event's `createdAt`, a scan's listing time) and `now()` otherwise; // it records WHEN, it orders nothing. A membership WorkOS merely // deactivated (`status = 'inactive'`, no `deleted_at`) is not a tombstone: // WorkOS can reactivate it under the same id, and the `updatedAt` rule // orders that as any other update. The row tombstone can only speak for the // id the row happens to hold, so every delete ALSO records the deleted id in // `membership_tombstones`, a ledger keyed by the WorkOS id alone, and every // membership write consults it first: that covers the case the row cannot — // membership A replaced by B in WorkOS before the mirror saw either, B then // deleted. The delete finds a row holding A (not B's to tombstone) and would // otherwise leave nothing behind; a later scan that still lists B would then // insert it live, stamped after A and under another id, exactly what the row // guard lets through. With the ledger the delete is recorded whatever the // row holds, and B's payload is refused by identity. A scan records the ids // it tombstones the same way. A deleted USER is protected by identity // too: `deleteUser` leaves the account row behind as a tombstone (profile // cleared, stamped with the deletion), and no membership naming that account // is written again, however the payload is stamped — WorkOS never reuses a // user id, and a membership the mirror had not seen has no row of its own // for a guard to refuse the insert against. The reconciler applies a page // of events and advances the cursor in ONE transaction that // compare-and-sets the cursor first (`applyPage`), so a run that has lost // the stream to another run writes nothing — the `updatedAt` guard alone // cannot stop it re-applying a page the leading run has moved past. // // A backfill SCAN of one organization (its full membership listing, taken // at one instant) is applied the same way: ONE transaction that first // compare-and-sets the organization's `backfilled_at` to the listing's // instant (`applyOrganizationScan`), so the row stays locked until commit, // two overlapping scans serialize on it, and the one whose listing is older // than the recorded one writes NOTHING. The `updatedAt` guard alone cannot // order two scans: a scan that listed a membership, stalled, and resumed // after a later scan had found it gone would insert it live — the later scan // left no tombstone, because the row was not there to tombstone. For the // same reason every OTHER membership write is ordered against the scan too: // a payload stamped before the organization's `backfilled_at` is refused // unless the row already holds it — judged inside the write's own // transaction, reading the organization row `FOR SHARE`, so a write that // races a scan waits for the scan's commit and sees the mark it set, and // never inserts a membership the scan has just proved revoked. A completed // scan is the full listing as // of that instant, so a membership it did not contain but an older payload // still names (a login whose list was fetched before the revocation and // written after the scan) was revoked before the scan — and that revocation // predates the events replay boundary, so nothing would ever tombstone the // reinstated row. The scan's own writes are the one exception: they are the // listing that sets the mark. // // The events replay boundary (`workos_sync.range_start`) is written ONCE, by // the first backfill run BEFORE its first listing, and never advanced: a // scan refreshes memberships only, so an organization rename or user // deletion after that instant — between two runs, or between a run that // failed part-way and its retry — is covered by the events stream alone, // and moving the boundary past it would skip it for good. // --------------------------------------------------------------------------- import { and, eq, isNotNull, isNull, lt, ne, notInArray, or, sql } from "drizzle-orm"; import type { AnyPgColumn } from "drizzle-orm/pg-core"; import { Data, Effect, Option } from "effect"; import type { MemberStatus } from "@executor-js/api/server"; import { accounts, membershipTombstones, memberships, organizations, workosSync, } from "../db/schema"; import type { DrizzleDb } from "../db/db"; import { insertOrganization, organizationAcceptsName } from "./user-store"; import { WorkOsMirrorError, tryPromiseService, userStoreReasonFromCause, withServiceLogging, } from "./errors"; /** A WorkOS user, as the mirror stores it. `updatedAt` is WorkOS's own. */ export interface WorkOsMirrorUser { readonly id: string; readonly email: string; readonly firstName: string | null; readonly lastName: string | null; readonly avatarUrl: string | null; readonly lastSignInAt: Date | null; readonly updatedAt: Date; } /** * A WorkOS organization membership, as the mirror stores it. `id` is the * WorkOS `om_…`; `accountId` the WorkOS user id; `updatedAt` is WorkOS's own. * The organization row must already be mirrored (`upsertOrganization`) — a * membership of an unknown org is a `query` failure, not a silent skip. */ export interface WorkOsMirrorMembership { readonly id: string; readonly accountId: string; readonly organizationId: string; readonly role: string; readonly status: MemberStatus; readonly updatedAt: Date; } /** * What identifies a membership to a delete: the WorkOS `om_…` id AND the * (account, organization) pair it belongs to. The pair is the row's key, * and a delete must be able to mint the row as a tombstone when the mirror * has not seen the membership yet — an id alone cannot. */ export type WorkOsMirrorMembershipRef = Pick< WorkOsMirrorMembership, "id" | "accountId" | "organizationId" >; /** One member as a backfill scan lists it: the membership and its user. */ export interface WorkOsScannedMember { readonly user: WorkOsMirrorUser; readonly membership: WorkOsMirrorMembership; } /** * One organization's full membership listing, as `applyOrganizationScan` * applies it. `listedAt` is the instant the listing was taken — BEFORE the * WorkOS read, so no change can fall between the instant and the listing — * and is the tombstone time for what the listing no longer contains, the * cut-off for what may be tombstoned at all, and the organization's new * `backfilled_at`. */ export interface WorkOsOrganizationScan { readonly organizationId: string; readonly listedAt: Date; readonly members: readonly WorkOsScannedMember[]; } /** What an applied scan wrote: the upserts the `updatedAt` guard let through, and the tombstones. */ export interface WorkOsOrganizationScanWrites { readonly usersWritten: number; readonly membershipsWritten: number; readonly membershipsTombstoned: number; } /** * One write of a reconciler page, applied by `applyPage` inside the page's * transaction. The reconciler plans a page into these BEFORE the transaction * opens, so every WorkOS read (resolving an organization the mirror has never * seen) is done by then: the transaction holds the mirror's single connection * and must not wait on the network. */ export type WorkOsMirrorWrite = Data.TaggedEnum<{ readonly UpsertUser: { readonly user: WorkOsMirrorUser }; readonly UpsertMembership: { readonly membership: WorkOsMirrorMembership }; /** * A membership together with its member's profile, read from WorkOS at * plan time because the mirror held no profile for the account * (`workos-events-sync.ts`): the user is upserted first, under the usual * guard, then the membership. The outcome is the membership's. */ readonly UpsertMember: { readonly user: WorkOsMirrorUser; readonly membership: WorkOsMirrorMembership; }; /** Tombstone a membership as of `deletedAt` (the event's `createdAt`). */ readonly DeleteMembership: { readonly membership: WorkOsMirrorMembershipRef; readonly deletedAt: Date; }; /** Tombstone a user and their memberships as of `deletedAt`. */ readonly DeleteUser: { readonly accountId: string; readonly deletedAt: Date }; /** * Rename an organization the mirror already holds — never inserts one — * from a payload stamped `updatedAt` (the WorkOS organization's own), under * the same name guard every feeder applies (`organizationAcceptsName`). */ readonly RenameOrganization: { readonly organizationId: string; readonly name: string; readonly updatedAt: Date; }; /** * Mark an organization deleted as of `deletedAt` (the event's * `createdAt`) — minting the row as a TOMBSTONE, named `name`, when the * mirror has never seen the organization, so a feeder still holding a * membership of it (a login that stalled across the deletion) finds the * tombstone and cannot mint the organization live. Never purges tenant * data (that is cloud's own flow, `db/org-deletion.ts`). An earlier mark * stands (`absent`). */ readonly MarkOrganizationDeleted: { readonly organizationId: string; readonly name: string; readonly deletedAt: Date; }; }>; export const WorkOsMirrorWrite = Data.taggedEnum<WorkOsMirrorWrite>(); /** * What one write did: a row was written, renamed, or tombstoned (`applied`); * the `updatedAt` guard refused an older payload (`stale`); or a delete found * its row already tombstoned or superseded by a newer membership (a * replayed delete), or a rename found no live organization row to change — * the mirror has never seen it, or it is marked deleted — or a deletion * mark found the organization already marked (`absent`). */ export type WorkOsMirrorWriteOutcome = "applied" | "stale" | "absent"; export interface WorkOsMirrorShape { /** * Insert or refresh a user row. `false` when the payload was refused and * the row left untouched: the stored row is newer than `updatedAt`, or is * a deletion tombstone stamped at `updatedAt` or later. */ readonly upsertUser: (user: WorkOsMirrorUser) => Effect.Effect<boolean, WorkOsMirrorError>; /** * Insert or refresh a membership row, minting the bare account row first so * the foreign key holds when the membership arrives before its user. `false` * when the payload was refused: the stored row is newer than `updatedAt`, * or is a deletion tombstone of THIS membership id, or THIS membership id * is in the deletion ledger (`membership_tombstones`, written by every * delete and every scan tombstone, whatever row it found) — a deleted * WorkOS id never returns, however the payload is stamped; a payload under * another id is a replacement and is ordered by `updatedAt` against the * row's stamp — or the account is a deletion tombstone (`deleteUser`), whatever * the payload is stamped: WorkOS never reuses a user id, so a deleted user * has no memberships to mirror — or the organization is marked deleted * (`organizations.deleted_at`), in which case nothing is written at all: * no membership of a deleted organization is ever mirrored — or the * payload is stamped BEFORE the organization's last full scan * (`organizations.backfilled_at`): the scan listed everything WorkOS held * at that instant, so a membership it wrote already carries a stamp at * least this new (the write would change nothing) and one it did not * write was gone by then and must not come back from a list fetched * earlier. Only the scan itself writes past that mark * (`applyOrganizationScan`). The organization checks and the write run in * ONE transaction that holds the organization row `FOR SHARE`, so a scan * claiming the row at the same time is waited for and its mark seen — and * the account row `FOR SHARE` too, so a `deleteUser` tombstoning it at the * same time is waited for and its tombstone seen. */ readonly upsertMembership: ( membership: WorkOsMirrorMembership, ) => Effect.Effect<boolean, WorkOsMirrorError>; /** * Tombstone the membership as deleted: the row stays (or is minted, when * the mirror has not seen the membership yet), `inactive`, carrying the * deleted WorkOS id with `deleted_at` set. The tombstone is protected by * that id: `upsertMembership` refuses every later payload naming it, * however stamped, and accepts only a REPLACEMENT under another id. The * row's `workos_updated_at` is left as it is — the deleted membership's * last reported state, which a replacement is always stamped after; a * local clock read after WorkOS answered could post-date a replacement * created meanwhile and must never become the row's stamp. `deletedAt` * is the deletion instant when the caller holds one (a deletion event's * `createdAt`, a scan's listing time) or `null` when it holds none * (WorkOS answers a delete with no time): `deleted_at` then takes the * current time. It records when the membership was deleted; it orders * nothing. Matches the row by IDENTITY: `false` when the row is already * tombstoned (a replayed delete), or holds ANOTHER membership id — the * member re-added in WorkOS under a new id, which stands whether the * replacement was mirrored before or after this delete. In EVERY case the * deleted id is recorded in the deletion ledger (`membership_tombstones`), * so a membership the mirror never held under its own id (replaced and * deleted in WorkOS before the mirror saw it) cannot be inserted live by a * scan that listed it before the deletion: `true` when the row was * tombstoned OR the id was newly recorded. The organization row must * already be mirrored, as for `upsertMembership`. */ readonly deleteMembership: ( membership: WorkOsMirrorMembershipRef, deletedAt: Date | null, ) => Effect.Effect<boolean, WorkOsMirrorError>; /** * Tombstone a deleted WorkOS user: every membership of the account is * tombstoned as by `deleteMembership` (as of `deletedAt`; a deleted user's * membership ids never return), and the account row is kept (it anchors * foreign keys) — or minted, when the mirror has not seen the user yet — * with its profile cleared and stamped `deletedAt`, so a stale user * payload cannot restore it. `false` when the account already carries * this tombstone or a later one (a replayed delete). ONE transaction that * locks the account row first (`FOR NO KEY UPDATE`) and holds it until * the memberships are tombstoned: `upsertMembership` reads that row `FOR * SHARE` before it inserts, so a membership write racing the deletion * waits for it and sees the tombstone, or committed first and is * tombstoned here — never a live membership left behind for a deleted * user. */ readonly deleteUser: ( accountId: string, deletedAt: Date, ) => Effect.Effect<boolean, WorkOsMirrorError>; /** The id of the last WorkOS event applied, or `null` before the first run. */ readonly getCursor: () => Effect.Effect<string | null, WorkOsMirrorError>; /** * Apply one reconciler page atomically: in a single transaction, * compare-and-set the cursor from `prev` (`null` = no cursor yet) to * `next`, and only if that succeeded apply `writes` in order. The cursor * row stays locked until commit, so two runs applying pages serialize on * it and the one whose `prev` is stale sees the moved cursor and writes * nothing: `None` means another run owns the stream and the caller must * stop. `Some` carries one outcome per write, in order. An empty `writes` * is a bare cursor advance. */ readonly applyPage: ( prev: string | null, next: string, writes: readonly WorkOsMirrorWrite[], ) => Effect.Effect<Option.Option<readonly WorkOsMirrorWriteOutcome[]>, WorkOsMirrorError>; /** * Apply one organization's backfill scan — the memberships (with their * users) a WorkOS listing taken at `listedAt` contained — atomically: in a * single transaction, compare-and-set the organization's `backfilled_at` * forward to `listedAt`, and only if that succeeded upsert every listed * user and membership and tombstone (as `deleteMembership` does, at * `listedAt`) every membership of the organization the listing did NOT * contain: the rows whose WorkOS id is not among the listed ones AND whose * stamp is older than the listing. A row stamped at or after `listedAt` * was written after the listing was taken (it could not be in it) and is * left alone — its own event lands through the reconciler; a tombstone * here would beat that event. Rows already tombstoned are left alone. * * The organization row stays locked until commit, so two overlapping * scans serialize on it and the one whose listing is older than (or the * same instant as) the recorded one writes nothing: `None` — the mirror * already holds the organization as of a later listing, and a membership * that listing no longer contained must not be inserted from this one. * `None` too for an organization the mirror does not hold or has marked * deleted: there is nothing to scan into. `Some` carries what was written. */ readonly applyOrganizationScan: ( scan: WorkOsOrganizationScan, ) => Effect.Effect<Option.Option<WorkOsOrganizationScanWrites>, WorkOsMirrorError>; /** * The Events API replay boundary: the instant the FIRST one-off backfill * run began reading WorkOS, or `null` if none has started. The * reconciler's first run (no cursor yet) reads the stream from here — the * backfill covers everything before it — and without a boundary it must * not guess. */ readonly replayBoundary: () => Effect.Effect<Date | null, WorkOsMirrorError>; /** * Record the replay boundary, ONCE: `at` is the instant a backfill run * began reading WorkOS, and it is kept only when no boundary is recorded * yet — `true` when this call recorded it. A later run never moves it: the * backfill refreshes memberships and tombstones only, not organization * names or deleted users' profiles, so a change after the first boundary * is covered only by the events stream, which must still be read from * there. Written BEFORE the run's first listing, so a run that fails * part-way leaves the boundary standing and its retry keeps it — a * `user.deleted` between the attempts stays inside the replay. Never * touches the cursor: a stream already being followed keeps its position, * and the boundary is then unused. */ readonly setReplayBoundary: (at: Date) => Effect.Effect<boolean, WorkOsMirrorError>; /** * When a backfill run first wrote EVERY live organization * (`workos_sync.backfill_completed_at`), or `null` while none has. The * first half of the mirror's READINESS for authorization: until a run has * covered every organization, the mirror may lack members who have not * signed in since it shipped, and a membership check read from it would * deny them. The other half is the events reconciler being caught up, * which its cursor row reports. */ readonly backfillCompletedAt: () => Effect.Effect<Date | null, WorkOsMirrorError>; /** * Record that a backfill run has written every live organization, as of * `at` — ONCE: a later completed run keeps the first mark (`false`), so * readiness never flips back. Written only after the last organization * was applied (or refused in favour of a later listing), so a run that * fails part-way records nothing here. Mints the events row when absent, * as `setReplayBoundary` does, and touches neither the cursor nor the * boundary. */ readonly markBackfillCompleted: (at: Date) => Effect.Effect<boolean, WorkOsMirrorError>; /** * When a reconciler run last read the events stream to its end * (`workos_sync.drained_at`), or `null` if none has. The second half of * the mirror's readiness for authorization: a mirror whose reconciler * has not caught up within the lag budget may still hold a membership * WorkOS has since revoked. */ readonly drainedAt: () => Effect.Effect<Date | null, WorkOsMirrorError>; /** * Record that a reconciler run read the stream to its end at `at`. Moves * the mark forward only — a run that finished after a later one keeps the * later mark — and only on the row a run already owns: the events row is * minted by the boundary or the first cursor advance, so a missing row * means nothing was drained and nothing is written (`false`). */ readonly markDrained: (at: Date) => Effect.Effect<boolean, WorkOsMirrorError>; /** * When the organization's membership list was last FULLY scanned from * WorkOS (`backfillOrganization` in workos-mirror-backfill.ts), or `null` * if it never was — or the organization is not mirrored. Until it has been, * the mirror may hold only the members login and write-through happened to * record, so a member count read from it is PARTIAL: every seat gate reads * this first and scans the organization when it is `null`. Per * organization, never database-wide, so an organization mirrored after a * backfill ran (lazily, or by a sign-in) is never mistaken for a scanned * one. */ readonly organizationBackfilledAt: ( organizationId: string, ) => Effect.Effect<Date | null, WorkOsMirrorError>; } // --------------------------------------------------------------------------- // SDK payload → mirror row. The feeders (login callback, write-through, the // backfill, the Events reconciler) all hand the mirror WorkOS objects; this is // the one place their field names and ISO timestamps are translated. Typed // structurally (the fields actually read) so the SDK's `User` / // `OrganizationMembership`, an event payload, and a test fixture all fit. // --------------------------------------------------------------------------- /** The WorkOS user fields the mirror reads. `User` from the SDK satisfies it. */ export interface WorkOsUserPayload { readonly id: string; readonly email: string; readonly firstName: string | null; readonly lastName: string | null; readonly profilePictureUrl: string | null; readonly lastSignInAt: string | null; readonly updatedAt: string; } /** * The WorkOS membership fields the mirror reads. `OrganizationMembership` from * the SDK satisfies it; its `status` is exactly the mirror's `MemberStatus`. */ export interface WorkOsMembershipPayload { readonly id: string; readonly userId: string; readonly organizationId: string; readonly role: { readonly slug: string }; readonly status: MemberStatus; readonly updatedAt: string; } /** Translate a WorkOS user payload to the row `upsertUser` stores. */ export const mirrorUserFromWorkOs = (user: WorkOsUserPayload): WorkOsMirrorUser => ({ id: user.id, email: user.email, firstName: user.firstName, lastName: user.lastName, avatarUrl: user.profilePictureUrl, lastSignInAt: user.lastSignInAt === null ? null : new Date(user.lastSignInAt), updatedAt: new Date(user.updatedAt), }); /** Translate a WorkOS membership payload to the row `upsertMembership` stores. */ export const mirrorMembershipFromWorkOs = ( membership: WorkOsMembershipPayload, ): WorkOsMirrorMembership => ({ id: membership.id, accountId: membership.userId, organizationId: membership.organizationId, role: membership.role.slug, status: membership.status, updatedAt: new Date(membership.updatedAt), }); /** * The `workos_sync` row of the one WorkOS events stream the reconciler * follows. A row id rather than a singleton table so a second stream * (another WorkOS environment, a replay) can be added without a schema * change; the same row carries the backfill's replay boundary and * completion mark. Not a `db/schema.ts` export: that module's exports are * enumerated as tables by the purge-coverage test. */ export const WORKOS_EVENTS_STREAM_ID = "events"; // A raw `sql` fragment binds a Date without the column's driver mapping, so // every instant below is passed as ISO text and cast.