// ---------------------------------------------------------------------------// Backfill of the membership mirror from WorkOS, one organization at a time:// list EVERY membership WorkOS holds for it — active, pending, and inactive// alike — fetch each member's user, write both through the mirror's guarded// upserts, tombstone whatever the mirror still holds that WorkOS no longer// lists, and mark the organization BACKFILLED as of the listing. Inactive// memberships are listed on purpose: the scan tombstones every mirrored// membership its listing lacks, and a tombstone is keyed to the membership// id for good (`membershipAcceptsPayload`), so a listing that skipped the// inactive ones would tombstone a membership WorkOS merely deactivated and// refuse its reactivation — under the same id — forever. Listed with its// real status it stays an ordinary `inactive` row that a newer payload// reactivates. The core is a pure function over a `source`// (WorkOS reads) and the mirror store, so the one-off script// (`scripts/backfill-workos-mirror.ts`) can wire real clients, the request// path can wire `WorkOSClient` (an organization whose mark is missing is// scanned on demand before its seats are counted), and the test can wire// fakes against the test database.//// Idempotent and repairing: the upserts are guarded on WorkOS `updatedAt`,// so a re-scan over unchanged data writes nothing new, and every membership// the mirror holds for the org that WorkOS no longer lists is TOMBSTONED// (`inactive`, `deleted_at` = the time the listing was taken) — so a re-scan// repairs a stale row instead of leaving it granting access. `dryRun` reads// everything and writes nothing, so the printed counts are the plan.//// One scan is applied in ONE transaction (`WorkOsMirror.applyOrganizationScan`)// that first moves the organization's `backfilled_at` forward to the// listing's instant and writes nothing if a LATER listing already did. Two// scans of the same organization can overlap (the one-off script and an// on-demand scan from a request, or a stalled script run and its retry), and// the `updatedAt` guard alone cannot order them: a scan that listed a// membership, stalled, and resumed after a later scan had found it gone// would insert it live — the later scan tombstoned nothing, because the row// was not there to tombstone. Refusing the older listing whole is what keeps// a membership revoked between the two listings revoked.//// Completeness is tracked PER ORGANIZATION (`organizations.backfilled_at`),// never database-wide: the mark is written only by a scan that listed that// organization's memberships in full, so an organization mirrored after a// backfill ran (lazily by a request, or by a sign-in that records only the// caller's own membership) starts unmarked and is scanned before any count// read from the mirror is trusted. A run over every organization// (`backfillWorkOsMirror`) additionally records the Events API replay// boundary — the instant it began reading WorkOS — BEFORE anything is// listed, and only if no boundary is recorded yet. Recording it first, not// on completion, is what makes a failed run safe to retry: a run that// fails part-way has already fixed the boundary at its start, and its// retry reads that boundary back instead of taking a fresh, later one. A// scan refreshes memberships and tombstones, not organization names or// deleted users' profiles, so an `organization.updated` or `user.deleted`// that lands between the attempts (or between two runs) is covered only by// the events stream — a boundary taken by the retry would fall after it and// skip it for good, leaving the deleted user's profile in the mirror. The// reconciler's own cursor takes over from the boundary after its first// page, so the boundary's only job is to name where that first page starts.//// The mark also orders every OTHER membership write against the scan: the// mirror refuses a membership payload stamped before the organization's// `backfilled_at` (`upsertMembership`). A login whose membership list was// fetched before a revocation and written after the scan would otherwise// reinstate the revoked membership — and that revocation predates the// events replay boundary, so no event would ever tombstone it again.// --------------------------------------------------------------------------- import { Clock, Effect, Option } from "effect"; import {  mirrorMembershipFromWorkOs,  mirrorUserFromWorkOs,  type WorkOsMembershipPayload,  type WorkOsMirrorShape,  type WorkOsOrganizationScanWrites,  type WorkOsUserPayload,} from "./workos-mirror-store"; /** The WorkOS reads one organization's scan performs, over whatever client the caller wires. */export interface WorkOsOrganizationScanSource<E> {  /**   * EVERY membership of one organization, all pages and all statuses   * (active, pending, inactive). A source that filtered by status would   * have the scan tombstone what it filtered out — see the header.   */  readonly listOrgMembers: (    organizationId: string,  ) => Effect.Effect<readonly WorkOsMembershipPayload[], E>;  readonly getUser: (userId: string) => Effect.Effect<WorkOsUserPayload, E>;} /** The reads the full backfill performs: every organization, then each one's scan. */export interface WorkOsMirrorBackfillSource<E> extends WorkOsOrganizationScanSource<E> {  /** Every organization id the mirror knows (FK target of `memberships`). */  readonly listOrganizationIds: () => Effect.Effect<readonly string[], E>;} export interface WorkOsMirrorBackfillOptions {  readonly dryRun: boolean;  /** One line per organization and one summary line; never a user's data. */  readonly log: (line: string) => void;} /** What one organization's scan did. */export interface WorkOsOrganizationScanCounts extends WorkOsOrganizationScanWrites {  /** Memberships WorkOS reported for the organization. */  readonly memberships: number;  /**   * Whether the listing was written to the mirror. `false` on a dry run, and   * when the mirror refused the listing whole: a later listing of the   * organization had already been applied (an overlapping scan finished   * first), or the organization is marked deleted or not mirrored. Every   * write count is 0 then.   */  readonly applied: boolean;} const NOTHING_WRITTEN: WorkOsOrganizationScanWrites = {  usersWritten: 0,  membershipsWritten: 0,  membershipsTombstoned: 0,}; export interface WorkOsMirrorBackfillCounts extends WorkOsOrganizationScanWrites {  readonly organizations: number;  /** Memberships WorkOS reported across every organization. */  readonly memberships: number;} // Bounded fan-out for the per-member `getUser` calls: enough to overlap the// WorkOS round-trips, low enough to stay clear of its rate limit.const USER_FETCH_CONCURRENCY = 5; const now = () => Effect.map(Clock.currentTimeMillis, (millis) => new Date(millis)); /** * Scan one organization: list every membership WorkOS holds for it, fetch * each member's user, and apply the listing to the mirror in one transaction * — the listed users and memberships upserted, the rest tombstoned, the * organization marked backfilled as of the listing — unless a later listing * was applied first, in which case nothing is written (`applied: false`). * The organization row must already be mirrored. Fails on the first source * or mirror failure and writes nothing then — a failed scan is safe to * repeat, so surfacing the failure beats a silent skip. A dry run reads * everything, writes nothing, and marks nothing. */export const backfillOrganization = <E>(  source: WorkOsOrganizationScanSource<E>,  mirror: WorkOsMirrorShape,  organizationId: string,  options: { readonly dryRun: boolean },) =>  Effect.gen(function* () {    // The listing's own instant, taken BEFORE the read so no change can fall    // between them: the tombstone time for whatever the listing no longer    // contains, the cut-off for what may be tombstoned at all (a row stamped    // at or after it was written after the listing and is not missing from    // it), and the organization's new `backfilled_at`.    const listedAt = yield* now();    const listed = yield* source.listOrgMembers(organizationId);    const members = yield* Effect.forEach(      listed,      (membership) =>        Effect.map(source.getUser(membership.userId), (user) => ({          user: mirrorUserFromWorkOs(user),          membership: mirrorMembershipFromWorkOs(membership),        })),      { concurrency: USER_FETCH_CONCURRENCY },    );    if (options.dryRun) {      const counts: WorkOsOrganizationScanCounts = {        ...NOTHING_WRITTEN,        memberships: members.length,        applied: false,      };      return counts;    }    const written = yield* mirror.applyOrganizationScan({      organizationId,      listedAt,      members,    });    if (Option.isNone(written)) {      yield* Effect.logWarning(        "workos_mirror: organization scan not applied — a later listing was already applied, or the organization is deleted or not mirrored",        { organizationId, listedAt: listedAt.toISOString() },      );    }    const counts: WorkOsOrganizationScanCounts = {      ...Option.getOrElse(written, () => NOTHING_WRITTEN),      memberships: members.length,      applied: Option.isSome(written),    };    return counts;  }).pipe(    Effect.withSpan("workos_mirror.backfillOrganization", {      attributes: { organizationId },    }),  ); /** * Run the full backfill: record the Events API replay boundary if none is * recorded yet, then scan every organization the mirror knows. Fails on the * first source or mirror failure — the organizations scanned so far stay * marked (each was covered in full), the boundary recorded at the start * stands, and the run is safe to repeat: the retry keeps that boundary, so * every change since the first attempt began is the reconciler's to replay. */export const backfillWorkOsMirror = <E>(  source: WorkOsMirrorBackfillSource<E>,  mirror: WorkOsMirrorShape,  options: WorkOsMirrorBackfillOptions,) =>  Effect.gen(function* () {    // The replay boundary: taken AND recorded before anything is listed, so    // every change from this instant on is the events stream's to apply —    // one that lands while this run is still listing, or between this run    // failing part-way and its retry. Kept only when none is recorded yet    // (`setReplayBoundary`): a retry or a later run reads the first one    // back instead of moving it. A dry run records nothing.    const boundary = yield* now();    if (!options.dryRun) {      const recorded = yield* mirror.setReplayBoundary(boundary);      options.log(        recorded          ? `events replay boundary set to ${boundary.toISOString()}`          : "events replay boundary already recorded by an earlier run; kept (the events reconciler replays every change since it)",      );    }     const organizationIds = yield* source.listOrganizationIds();    let memberships = 0;    let usersWritten = 0;    let membershipsWritten = 0;    let membershipsTombstoned = 0;     for (const organizationId of organizationIds) {      const scanned = yield* backfillOrganization(source, mirror, organizationId, options);      memberships += scanned.memberships;      usersWritten += scanned.usersWritten;      membershipsWritten += scanned.membershipsWritten;      membershipsTombstoned += scanned.membershipsTombstoned;      options.log(        `${organizationId}  ${scanned.memberships} membership(s)` +          (options.dryRun            ? ""            : scanned.applied              ? `  wrote ${scanned.usersWritten} user(s), ${scanned.membershipsWritten} membership(s), tombstoned ${scanned.membershipsTombstoned}`              : "  not applied (a later listing was already applied, or the organization is deleted)"),      );    }     const counts: WorkOsMirrorBackfillCounts = {      organizations: organizationIds.length,      memberships,      usersWritten,      membershipsWritten,      membershipsTombstoned,    };    options.log(      options.dryRun        ? `dry run — ${counts.organizations} organization(s), ${counts.memberships} membership(s) would be mirrored`        : `${counts.organizations} organization(s), ${counts.memberships} membership(s): wrote ${counts.usersWritten} user(s), ${counts.membershipsWritten} membership(s), tombstoned ${counts.membershipsTombstoned}`,    );    if (!options.dryRun) {      // Every live organization is now covered (a failure above fails the      // whole run): the mirror is complete enough to authorize from, once      // the reconciler has caught up too — the first half of the readiness      // the authorization path checks. Once: a re-run keeps the first      // completion.      const completedAt = yield* now();      const marked = yield* mirror.markBackfillCompleted(completedAt);      options.log(        marked          ? `backfill completion recorded at ${completedAt.toISOString()}`          : "backfill completion already recorded by an earlier run; kept",      );    }    return counts;  });