// ---------------------------------------------------------------------------
// Runs one reconciler pass (`syncWorkOsEvents`) from a Worker entry that is
// not an HTTP request handled by the Effect app: the every-minute cron
// (`scheduled` in server.ts) and the webhook poke (`workos-webhook.ts`,
// detached past the response with `waitUntil`).
//
// Both entries build the request-scoped services FRESH for the run — the
// same reason `mcp/auth.ts` does: a postgres socket belongs to one Workers
// invocation, and the webhook route's own per-request layer is closed the
// moment its response is returned, so a detached run cannot borrow it. The
// run is its own scope; the socket is released when it ends.
//
// A failing run is captured (Sentry + structured log) and swallowed here:
// neither entry has a caller to report to, and the run is retried by the
// next cron tick from the last committed cursor.
//
// A run that does not fail outright can still leave the mirror stale: one
// that stops short of `"drained"` (page budget, cursor contention) means the
// reconciler did not catch up this tick, and one stuck on
// `"awaiting_backfill"` past the budget means the boundary row the backfill
// was supposed to hand off was lost — both are read from `drainedAt` after
// the run and, past `MIRROR_RECONCILER_LAG_BUDGET`, reported the same way a
// thrown failure is: a structured error log plus a Sentry capture. This is
// the ONLY place a stale reconciler surfaces now; the request path
// (`auth/organization.ts`) no longer checks readiness or falls back.
// ---------------------------------------------------------------------------

import { Clock, Data, Duration, Effect, Layer } from "effect";

import { captureCauseEffect } from "../observability";
import { WorkerTelemetryLive } from "../observability/telemetry";
import { makeDbLayer } from "../db/db";
import { makeUserStoreLayer } from "./context";
import { MIRROR_RECONCILER_LAG_BUDGET } from "./mirror-readiness-store";
import { CoreSharedServices } from "./workos";
import { syncWorkOsEvents, type WorkOsEventsSyncReport } from "./workos-events-sync";
import { makeWorkOsMirrorLayer, WorkOsMirror } from "./workos-mirror";

const makeSyncServices = () => {
  const dbLive = makeDbLayer();
  return Layer.mergeAll(
    makeUserStoreLayer().pipe(Layer.provide(dbLive)),
    makeWorkOsMirrorLayer().pipe(Layer.provide(dbLive)),
    CoreSharedServices,
  );
};

const LAG_BUDGET_MS = Duration.toMillis(MIRROR_RECONCILER_LAG_BUDGET);

/**
 * The mirror has not drained the WorkOS events stream inside its lag budget.
 * Its own tagged error so Sentry groups every occurrence under one issue,
 * with the run's outcome and the heartbeat's age as the fields to read.
 */
export class WorkOsReconcilerStale extends Data.TaggedError("WorkOsReconcilerStale")<{
  readonly stopped: WorkOsEventsSyncReport["stopped"];
  readonly drainedAt: string | null;
  readonly ageMs: number | null;
}> {}

/**
 * After a run that did not end `"drained"`, check whether the mirror has
 * fallen behind its lag budget and, if so, report it the way a failed run is
 * reported: a structured error log plus a Sentry capture. A run ending
 * `"drained"` just wrote `markDrained` and is healthy by definition, so it is
 * never checked. Cheap by design: one `drainedAt` read per run (about once a
 * minute).
 */
export const alertOnStaleReconciler = Effect.fn("workos_events.alert_on_stale_reconciler")(
  function* (report: WorkOsEventsSyncReport) {
    if (report.stopped === "drained") return;

    const mirror = yield* WorkOsMirror;
    const drainedAt = yield* mirror.drainedAt();
    const now = yield* Clock.currentTimeMillis;
    const ageMs = drainedAt === null ? null : now - drainedAt.getTime();
    if (ageMs !== null && ageMs <= LAG_BUDGET_MS) return;

    const stale = new WorkOsReconcilerStale({
      stopped: report.stopped,
      drainedAt: drainedAt === null ? null : drainedAt.toISOString(),
      ageMs,
    });
    yield* Effect.logError("workos_events: reconciler stale", stale);
    yield* captureCauseEffect(stale);
  },
);

/**
 * One reconciler pass over fresh request-scoped services. Resolves when the
 * pass ends, whether it drained the stream, stopped at the page budget,
 * yielded to another run, or failed (a failure is reported, never thrown).
 * A run that ends short of `"drained"` and leaves `drainedAt` past the lag
 * budget is ALSO reported (see {@link alertOnStaleReconciler}) — a stalled
 * reconciler is now an operational alert, not a per-request fallback.
 */
export const runWorkOsEventsSync = (): Promise<void> =>
  Effect.runPromise(
    syncWorkOsEvents().pipe(
      Effect.tap((report) => alertOnStaleReconciler(report)),
      Effect.asVoid,
      Effect.provide(makeSyncServices()),
      Effect.scoped,
      Effect.catchCause((cause) =>
        Effect.gen(function* () {
          yield* Effect.logError("workos_events: sync run failed", cause);
          yield* captureCauseEffect(cause);
        }),
      ),
      Effect.provide(WorkerTelemetryLive),
    ),
  );