import type { HiveId, ProfileViewDetailed } from "./types";
import type { ActorIdentifier } from "@atcute/lexicons/syntax";
import type { Did } from "@atcute/lexicons";
import type { OAuthClient } from "@atcute/oauth-node-client";
import type { Ingester } from "./bsky/ingester";
import type { Context } from "hono";
import { Hono } from "hono";
import { endTime, startTime, type TimingVariables } from "hono/timing";
import { getIronSession, type IronSession } from "iron-session";
import { sql } from "kysely";
import type { Logger } from "pino";
import { createStorage, type Storage } from "unstorage";
import lruCacheDriver from "unstorage/drivers/lru-cache";

import {
  createOAuthClient,
  sessionClientFromOAuthSession,
  type SessionClient,
} from "./auth/client";
import { createCrossProcessLock } from "./auth/refresh-lock";
import { guardedRestore, isSessionTerminatingError } from "./auth/restore-guard";
import { getStoredSessionIssuerHost } from "./auth/storage";
import { createServiceAccountAgent } from "./services/catalogBookService";
import {
  evictCachedSessionClient,
  getCachedSessionClient,
  getSessionConfig,
  markSessionSaved,
  setCachedSessionClient,
} from "./auth/session";
import {
  createBaseIdResolver,
  createBidirectionalResolverAtcute,
  createCachingBaseIdResolver,
  createCachingBidirectionalResolver,
  getDidDocumentResolver,
} from "./bsky/id-resolver";
import type { BidirectionalResolver } from "./bsky/id-resolver";
import { ServiceJwtVerifier } from "@atcute/xrpc-server/auth";
import type { AtprotoAudience } from "@atcute/lexicons/syntax";
import { BOOKHIVE_DID } from "./constants";
import { sweepStaleUploads } from "./services/uploadPersonalBook";
import type { Database } from "./db";
import { createDb, migrateToLatest } from "./db";
import { env } from "./env";
import { getLogger } from "./logger/index.ts";
import { PAGE_CACHE_TTL_MS } from "./middleware/anon-page-cache";
import sqliteKv, {
  createSharedKvDb,
  incrementalVacuumKv,
  readPragma,
  vacuumKvIfBloated,
  VACUUM_FREELIST_RATIO,
} from "./sqlite-kv.ts";
import { startEnrichmentDrain } from "./data/enrichQueue";
import { lazy } from "./lib/lazy";
import { readThroughCache } from "./lib/readThroughCache";
import { updateBookRecord } from "./services/getBook";
import type { PendingWrite } from "./data/syncBridge";
import { errorMessage } from "./lib/errors";

/** Add business context to the single wide event emitted at request end. Prefer this over logger.info in handlers. */
export type AddWideEventContext = (context: Record<string, unknown>) => void;

/** Minimal context needed by shared book utilities (getBook, ensureBookCataloged, etc.). */
export type BookUtilContext = {
  db: Database;
  kv: Storage;
  serviceAccountAgent: SessionClient | null;
  addWideEventContext: AddWideEventContext;
};

// Application state passed to the router and elsewhere. No logger – request observability is via addWideEventContext + wide-event middleware.
export type AppContext = {
  db: Database;
  kv: Storage;
  ingester: Ingester;
  oauthClient: OAuthClient;
  resolver: BidirectionalResolver;
  baseIdResolver: ReturnType<typeof createBaseIdResolver>;
  /** Cheap DID lookup from iron-session cookie — no OAuth restore, no network calls. */
  getSessionDid: () => Promise<string | null>;
  getSessionAgent: () => Promise<SessionClient | null>;
  getProfile: () => Promise<ProfileViewDetailed | null>;
  /** Service account agent for @bookhive.buzz ATProto writes. Null if env vars not set. */
  serviceAccountAgent: SessionClient | null;
  /** Verifies atproto service-auth JWTs on /xrpc/*. Null when disabled. */
  serviceJwtVerifier: ServiceJwtVerifier | null;
  /** Add fields to the one wide event logged per request (observability). */
  addWideEventContext: AddWideEventContext;
};

import type { BundleAssetUrls } from "./lib/manifest";

declare module "hono" {
  interface ContextVariableMap {
    ctx: AppContext;
    assetUrls: BundleAssetUrls | null;
    /** Request ID (UUID); set by wide-event middleware for the rest of the request. */
    requestId: string;
    /** Mutable bag for wide-event context; merged into the single request log. */
    wideEventBag: Record<string, unknown>;
    /** App logger; only wide-event middleware should call it for request-scoped logs. */
    appLogger: Logger;
    /** Set by error-capture middleware (thrown) or by handlers (caught then return 5xx); included in wide-event log. */
    requestError?: unknown;
  }
}

export type AppEnv = {
  Variables: TimingVariables & {
    ctx: AppContext;
    assetUrls: BundleAssetUrls | null;
    requestId: string;
    appLogger: Logger;
    requestError?: unknown;
  };
};

export type HonoServer = Hono<AppEnv>;

export type Session = { did: string };

/** Long-lived dependencies created at server startup (no request-scoped helpers). */
export type AppDeps = {
  db: Database;
  kv: Storage;
  logger: Logger;
  oauthClient: OAuthClient;
  baseIdResolver: ReturnType<typeof createBaseIdResolver>;
  ingester: Ingester;
  resolver: BidirectionalResolver;
  serviceAccountAgent: SessionClient | null;
  /** Verifies atproto service-auth JWTs on /xrpc/*. Null when disabled. */
  serviceJwtVerifier: ServiceJwtVerifier | null;
  /** Stops the primary worker's enrichment drain loop; no-op elsewhere. */
  stopEnrichmentDrain: () => void;
};

export async function createAppDeps(): Promise<AppDeps> {
  const logger = getLogger({
    name: "server",
    redact: {
      paths: ["req.headers.cookie"],
      censor: "***REDACTED***",
    },
  });

  // WORKER_INDEX is set by server/cluster.ts when running multiple processes.
  // Absent (dev, tests, bare `bun run .output/server/index.mjs`) => primary, for
  // back-compat. Only the primary runs migrations/VACUUM and the Jetstream
  // ingester; the supervisor starts the primary alone and waits for its
  // healthcheck before spawning siblings, so migrations are done before any
  // non-primary worker opens the DB.
  const isPrimaryWorker = !env.WORKER_INDEX || env.WORKER_INDEX === "0";
  logger.info({ workerIndex: env.WORKER_INDEX || "solo" }, "worker starting");

  const { db, sqlite } = createDb(env.DB_PATH);
  if (isPrimaryWorker) {
    logger.info("starting DB migrations");
    const migrationStart = Date.now();
    const migrationResults = await migrateToLatest(db, sqlite);
    logger.info({ durationMs: Date.now() - migrationStart }, "db migrations completed");
    if (migrationResults.length > 0) {
      logger.info(
        { migrations: migrationResults.map((r: { migrationName: string }) => r.migrationName) },
        "migrations applied",
      );
      // VACUUM only when there is actually something to reclaim — the main DB
      // is essentially append-only, so an unconditional VACUUM here costs
      // startup time on every migrating deploy for nothing to reclaim. Gated
      // on this file's own freelist ratio (same threshold `vacuumKvIfBloated`
      // uses), so a migration that does free a lot still gets cleaned up.
      const pageCount = readPragma(sqlite, "page_count");
      const freelist = readPragma(sqlite, "freelist_count");
      const ratio = pageCount > 0 ? freelist / pageCount : 0;
      if (ratio < VACUUM_FREELIST_RATIO) {
        logger.info({ pageCount, freelist, ratio }, "db VACUUM skipped — nothing to reclaim");
      } else {
        const vacuumStart = Date.now();
        sqlite.exec("VACUUM");
        // `hive_book_fts` is external-content, keyed by `hive_book`'s *implicit*
        // rowid (`id` is TEXT, so not an INTEGER PRIMARY KEY alias). SQLite
        // documents that VACUUM "may change the ROWIDs of entries in any tables
        // that do not have an explicit INTEGER PRIMARY KEY"; if it ever does,
        // every search result silently points at the wrong book, and FTS5's
        // 'integrity-check' does not detect this class of desync — so rebuild
        // unconditionally here rather than trust it.
        const rebuildStart = Date.now();
        try {
          sqlite.exec(`INSERT INTO hive_book_fts(hive_book_fts) VALUES('rebuild')`);
        } catch (err) {
          logger.error({ err }, "hive_book_fts rebuild after VACUUM failed");
        }
        logger.info(
          {
            durationMs: Date.now() - vacuumStart,
            ftsRebuildMs: Date.now() - rebuildStart,
            freelist,
          },
          "db VACUUM complete",
        );
      }
    }
  }

  // Single shared connection for all KV tables on KV_DB_PATH.
  const { db: kvDb, sqlite: kvSqlite } = createSharedKvDb(env.KV_DB_PATH);
  if (isPrimaryWorker) {
    // The KV is delete-heavy (the page cache, auth state, the sweep below), so
    // unlike the main DB it does bloat. Runs before the siblings spawn, and
    // only when the file is actually bloated.
    vacuumKvIfBloated(kvSqlite, (fields, msg) => logger.info(fields, msg));
  }
  const kv = createStorage({
    driver: sqliteKv({ table: "kv", db: kvDb }),
  });

  if (env.isProd) {
    kv.mount("search:", lruCacheDriver({ max: 1000 }));
  }
  kv.mount("profile:", sqliteKv({ table: "profile", db: kvDb }));
  kv.mount("identity:", sqliteKv({ table: "identity", db: kvDb }));
  kv.mount("follows_sync:", sqliteKv({ table: "follows_sync", db: kvDb }));

  // Auth tables: in development use a separate file; in production share the main KV connection.
  const authKvDb = env.isDevelopment ? createSharedKvDb("./auth.sqlite").db : kvDb;
  kv.mount("auth_session:", sqliteKv({ table: "auth_sessions", db: authKvDb }));
  kv.mount("auth_state:", sqliteKv({ table: "auth_state", db: authKvDb }));
  // Shared (not in-memory) so the main process and the ingester/import
  // workers see the same per-DID book locks.
  kv.mount("book_lock:", sqliteKv({ table: "book_lock", db: kvDb }));
  kv.mount("sync_pending:", sqliteKv({ table: "sync_pending", db: kvDb }));
  // Per-user KOSync token rotation counter (see src/middleware/sync-auth.ts).
  kv.mount("sync_token:", sqliteKv({ table: "sync_token", db: kvDb }));
  // Anonymous full-page HTML cache (see src/middleware/anon-page-cache.ts).
  kv.mount("page:", sqliteKv({ table: "page_cache", db: kvDb }));
  if (isPrimaryWorker) {
    // Expire old cached pages so a bot sweep of the long tail can't grow the
    // KV file unboundedly. 2x TTL keeps recently-stale rows around for cheap
    // overwrite instead of insert.
    const pageCacheTimer = setInterval(
      () => {
        const cutoff = new Date(Date.now() - 2 * PAGE_CACHE_TTL_MS).toISOString();
        void sql`DELETE FROM page_cache WHERE updated_at < ${cutoff}`
          .execute(kvDb)
          .then(() => {
            // Awaited: this hands back the pages *that DELETE just freed*, and
            // firing it alongside an un-awaited DELETE only ever reclaimed the
            // previous cycle's. Bounded, so it can never become the
            // multi-second stall a full VACUUM would be on this timer. The
            // logger is passed so a SQLITE_BUSY or a full disk here is visible
            // rather than swallowed on a 15-minute timer nobody watches.
            incrementalVacuumKv(kvSqlite, 1000, (fields, msg) => logger.warn(fields, msg));
          })
          .catch((e: any) => {
            if (String(e?.message).includes("no such table")) return;
            logger.error({ err: e }, "page_cache cleanup failed");
          });
      },
      15 * 60 * 1000,
    );
    pageCacheTimer.unref();
  }

  const requestLock = createCrossProcessLock(authKvDb);
  const oauthClient = await createOAuthClient(kv, { requestLock });
  const baseIdResolver = createCachingBaseIdResolver(kv, createBaseIdResolver());
  const resolver = createCachingBidirectionalResolver(kv, createBidirectionalResolverAtcute());

  const serviceAccountAgent =
    env.BOOKHIVE_SERVICE_HANDLE && env.BOOKHIVE_APP_PASSWORD
      ? await createServiceAccountAgent(env.BOOKHIVE_SERVICE_HANDLE, env.BOOKHIVE_APP_PASSWORD)
      : null;

  // Only the primary worker runs the Jetstream ingester — N copies would mean
  // N firehose subscriptions racing on the same DB rows and KV cursor.
  let ingester: Ingester;
  if (isPrimaryWorker) {
    // When running the Nitro bundle (.output/server/index.mjs), load the pre-built worker.
    // In dev, Bun runs the .ts source directly.
    const isBundled = import.meta.url.includes(".output/");
    const workerUrl = isBundled
      ? new URL("./workers/ingester-worker.js", import.meta.url).href
      : new URL("./workers/ingester-worker.ts", import.meta.url).href;
    const ingesterWorker = new Worker(workerUrl);
    ingesterWorker.onmessage = (event: MessageEvent) => {
      if (event.data.type === "wideEvent") {
        logger.info(event.data.payload);
      } else if (event.data.type === "ready") {
        logger.info("ingester worker ready");
      }
    };
    ingesterWorker.onerror = (event) => {
      logger.error({ error: event.message }, "ingester worker error");
    };
    ingester = {
      start() {},
      destroy() {
        ingesterWorker.terminate();
        return Promise.resolve();
      },
    };
  } else {
    ingester = {
      start() {},
      destroy: () => Promise.resolve(),
    };
  }

  // Goodreads enrichment is queued by every process but drained only here — one
  // WAF token cache and one writer, instead of N processes each fanning out a
  // scrape per search result.
  const stopEnrichmentDrain = isPrimaryWorker ? startEnrichmentDrain({ db, logger }) : () => {};

  // Accepts atproto inter-service auth on /xrpc/*, which is what lets a client
  // that is not a browser (a script, an e-reader, another app) use the personal
  // library. Always on; the cookie path is unaffected either way. No replay
  // store — a token is already scoped to one lxm and one audience, so within its
  // short lifetime "authed is authed".
  const serviceJwtVerifier: ServiceJwtVerifier = new ServiceJwtVerifier({
    // atcute compares these with exact string equality, so a bare DID does
    // *not* match a `#fragment` audience — both spellings have to be listed.
    // The fragment form is here in advance of a PLC operation adding a
    // `#bookhive_appview` service entry to the DID document; once that lands,
    // clients that switch to PDS proxying keep working with no server change.
    acceptAudiences: serviceAuthAudiences(),
    resolver: getDidDocumentResolver(),
    maxAge: SERVICE_AUTH_MAX_AGE_SECONDS,
  });

  if (isPrimaryWorker) {
    // Clear `.part` files left by a process that died between an upload's write
    // and its rename. Nothing else removes them, and each is up to 100 MB.
    void sweepStaleUploads().then(
      (removed) => {
        if (removed > 0) logger.info({ removed }, "swept stale upload temp files");
      },
      (err: unknown) => logger.warn({ err }, "stale upload sweep failed"),
    );
  }

  return {
    db,
    kv,
    logger,
    oauthClient,
    baseIdResolver,
    ingester,
    resolver,
    serviceAccountAgent,
    serviceJwtVerifier,
    stopEnrichmentDrain,
  };
}

/**
 * Maximum accepted service-JWT lifetime window, in seconds. atcute defaults to
 * 300, but a PDS mints up to 3600 when `lxm` is set and most client SDKs don't
 * expose `exp`, so those tokens would be refused as JwtTooOld. The token's own
 * `exp` is still enforced separately, so widening this only accepts tokens the
 * issuing PDS already considered valid.
 */
const SERVICE_AUTH_MAX_AGE_SECONDS = 3600;

/**
 * The `aud` values this deployment answers for. Overridable so a staging
 * deployment can run under its own DID without a code change.
 */
function serviceAuthAudiences(): (Did | AtprotoAudience)[] {
  const override = env.XRPC_SERVICE_AUTH_AUDIENCES.split(",")
    .map((s) => s.trim())
    .filter(Boolean);
  if (override.length > 0) return override as (Did | AtprotoAudience)[];
  return [BOOKHIVE_DID, `${BOOKHIVE_DID}#bookhive_appview`] as (Did | AtprotoAudience)[];
}

/** Optional timing callbacks for server-timing breakdown (session_iron, session_restore, session_save). */
export type SessionTiming = {
  start: (name: string) => void;
  end: (name: string) => void;
};

/**
 * A cached SessionClient refreshes its token lazily, mid-request. When another
 * cluster process rotated that session first, the refresh throws
 * "session was deleted by another process" and the user got a 500 on an
 * otherwise fine page load.
 *
 * Treat it as retryable: drop the stale cache entry, restore the session once,
 * and replay the call against the fresh client. If that fails too, surface the
 * original error to the caller (which will handle it as an auth failure).
 */
const SESSION_DELETED_PATTERN = /session was deleted by another process/i;

/** DIDs currently inside a refresh retry — stops a replayed call from
 *  recursing when the restored client hits the same race again. */
const sessionRetryInFlight = new Set<string>();

function withSessionRefreshRetry(
  did: string,
  client: SessionClient,
  restore: () => Promise<SessionClient | null>,
): SessionClient {
  const wrap = <M extends "get" | "post">(method: M): SessionClient[M] =>
    (async (name: string, opts?: Record<string, unknown>) => {
      try {
        return await client[method](name, opts);
      } catch (err) {
        const message = errorMessage(err);
        if (!SESSION_DELETED_PATTERN.test(message)) throw err;
        if (sessionRetryInFlight.has(did)) throw err;

        evictCachedSessionClient(did);
        sessionRetryInFlight.add(did);
        try {
          const fresh = await restore().catch(() => null);
          if (!fresh) throw err;
          return await fresh[method](name, opts);
        } finally {
          sessionRetryInFlight.delete(did);
        }
      }
    }) as SessionClient[M];

  return {
    get did() {
      return client.did;
    },
    get: wrap("get"),
    post: wrap("post"),
  };
}

/**
 * Breaker key for a DID: the authorization-server host when we have a stored
 * session, else the DID itself. Falling back to the DID keeps a first-ever
 * restore guarded instead of unguarded.
 */
async function restoreGuardKey(ctx: { kv: Storage }, did: string): Promise<string> {
  return (await getStoredSessionIssuerHost(ctx.kv, did)) ?? did;
}

/** Restore a session client for a DID, bypassing the process cache. */
async function restoreSessionClient(ctx: AppContext, did: string): Promise<SessionClient | null> {
  try {
    const key = await restoreGuardKey(ctx, did);
    const oauthSession = await guardedRestore(key, () =>
      ctx.oauthClient.restore(did as Did, { refresh: "auto" }),
    );
    const tokenInfo = await oauthSession.getTokenInfo(false);
    // Same factory as the primary path, so a client that gets cached here can
    // still recover from a later rotation by another process.
    const client = withSessionRefreshRetry(did, sessionClientFromOAuthSession(oauthSession), () =>
      restoreSessionClient(ctx, did),
    );
    setCachedSessionClient(did, client, tokenInfo.expiresAt?.getTime());
    return client;
  } catch {
    return null;
  }
}

/**
 * Expire the `sid` cookie on the response. Used when the cookie can't be
 * decoded (see `readIronSession`) so the browser stops resending a value that
 * will only ever 500 — without this, a tampered or stale-secret cookie is a
 * sticky hard error on every page for that browser until the user clears it by
 * hand. Cookie name + options mirror `getSessionConfig`.
 */
function clearSidCookie(res: Response): void {
  const parts = ["sid=", "Path=/", "Max-Age=0", "HttpOnly", "SameSite=Lax"];
  if (env.NODE_ENV === "production") parts.push("Secure");
  res.headers.append("set-cookie", parts.join("; "));
}

/**
 * Decode the iron-session cookie, tolerating a corrupt/tampered/rotated-secret
 * value instead of throwing. iron-session throws `Wrong mac prefix` (and
 * similar) when the HMAC doesn't verify; left uncaught that surfaced as a 500
 * on every route for anyone whose `sid` cookie was modified or was sealed with
 * a previous `COOKIE_SECRET`. A cookie we can't verify is, for our purposes,
 * no session at all — so treat it as logged out and clear it.
 */
async function readIronSession(req: Request, res: Response): Promise<IronSession<Session> | null> {
  try {
    return await getIronSession<Session>(req, res, getSessionConfig());
  } catch {
    clearSidCookie(res);
    return null;
  }
}

// Helper function to get the session client for the active session
export async function getSessionAgent(
  req: Request,
  res: Response,
  ctx: AppContext,
  timing?: SessionTiming,
): Promise<SessionClient | null> {
  timing?.start("session_iron");
  const session = await readIronSession(req, res);
  timing?.end("session_iron");
  if (!session) {
    return null;
  }

  if (!session.did) {
    return null;
  }

  const cached = getCachedSessionClient(session.did);
  if (cached) {
    ctx.addWideEventContext({ session_cache: "hit" });
    // Periodically re-save the iron-session cookie to keep its 180-day TTL rolling.
    if (cached.needsSave) {
      session.updateConfig(getSessionConfig());
      await session.save();
      markSessionSaved(session.did);
    }
    return cached.client;
  }

  const guardKey = await restoreGuardKey(ctx, session.did);

  try {
    timing?.start("session_restore");
    const oauthSession = await guardedRestore(
      guardKey,
      () => ctx.oauthClient.restore(session.did as Did, { refresh: "auto" }),
      (outcome) =>
        ctx.addWideEventContext({
          pds_host: outcome.key,
          pds_breaker: outcome.state,
          oauth_restore_ms: outcome.durationMs,
        }),
    );
    timing?.end("session_restore");

    // Get token expiration so we can cache until just before it expires.
    // This is cheap: reads from the session store with allowStale (no network calls).
    const tokenInfo = await oauthSession.getTokenInfo(false);
    const tokenExpiresAt = tokenInfo.expiresAt?.getTime();

    ctx.addWideEventContext({
      session_cache: "miss",
      token_expires_in_ms: tokenExpiresAt ? tokenExpiresAt - Date.now() : undefined,
    });

    timing?.start("session_save");
    session.updateConfig(getSessionConfig());
    await session.save();
    timing?.end("session_save");

    const did = session.did;
    const client = withSessionRefreshRetry(did, sessionClientFromOAuthSession(oauthSession), () =>
      restoreSessionClient(ctx, did),
    );
    setCachedSessionClient(did, client, tokenExpiresAt);
    return client;
  } catch (err) {
    // Only tear the session down when the PDS has actually rejected our
    // credentials. A timeout or an unreachable host says nothing about whether
    // the user is still logged in, and destroying on those would silently sign
    // people out for the duration of their PDS's downtime.
    const terminal = isSessionTerminatingError(err);
    ctx.addWideEventContext({
      oauth_restore: "failed",
      oauth_restore_terminal: terminal,
      pds_host: guardKey,
      error: errorMessage(err),
    });
    if (terminal) session.destroy();
    return null;
  }
}

/** Middleware that sets request-scoped AppContext (deps + getSessionAgent, getProfile). Use on any Hono app that needs c.get("ctx"). */
export function createContextMiddleware(deps: AppDeps) {
  return async (c: Context<AppEnv>, next: () => Promise<void>) => {
    c.set("wideEventBag", {});
    c.set("appLogger", deps.logger);

    const { logger: _logger, ...restDeps } = deps;
    const ctx: AppContext = {
      ...restDeps,
      addWideEventContext(context: Record<string, unknown>) {
        Object.assign(c.get("wideEventBag"), context);
      },
      getSessionDid(): Promise<string | null> {
        return didLazy.value;
      },
      getSessionAgent(): Promise<SessionClient | null> {
        return sessionLazy.value;
      },
      async getProfile(): Promise<ProfileViewDetailed | null> {
        return profileLazy.value as Promise<ProfileViewDetailed | null>;
      },
    };
    const sessionTiming: SessionTiming = {
      start: (name) => startTime(c, name),
      end: (name) => endTime(c, name),
    };

    // Fast path: read DID from iron-session cookie only (no OAuth restore).
    const didLazy = lazy(async () => {
      startTime(c, "session_iron");
      const session = await readIronSession(c.req.raw, c.res);
      endTime(c, "session_iron");
      return session?.did || null;
    });

    const sessionLazy = lazy(() =>
      getSessionAgent(c.req.raw, c.res, ctx, sessionTiming).then((client) => {
        if (client) {
          ctx.addWideEventContext({ userDid: client.did });
          void flushPendingSyncWrites(client, ctx);
        }
        return client;
      }),
    );

    // getProfile uses the fast DID path for cache lookups.
    // The expensive OAuth restore only happens on profile cache miss.
    const profileLazy = lazy(async () => {
      startTime(c, "get_profile_did");
      const did = await didLazy.value;
      endTime(c, "get_profile_did");
      if (!did) {
        return null;
      }
      ctx.addWideEventContext({ userDid: did });

      startTime(c, "get_profile_cache");
      const result = await readThroughCache<ProfileViewDetailed | null>(
        deps.kv,
        "profile:" + did,
        async () => {
          // Cache miss — need a full session to call the Bluesky AppView via PDS proxy.
          startTime(c, "get_profile_session");
          const client = await sessionLazy.value;
          endTime(c, "get_profile_session");
          if (!client) throw new Error("session_unavailable");
          startTime(c, "get_profile_network");
          try {
            const res = await client.get("app.bsky.actor.getProfile", {
              params: { actor: client.did as ActorIdentifier },
              headers: { "atproto-proxy": "did:web:api.bsky.app#bsky_appview" },
            });
            if (!res.ok) return null;
            return res.data as ProfileViewDetailed | null;
          } finally {
            endTime(c, "get_profile_network");
          }
        },
        undefined,
        { revalidateAfter: 24 * 60 * 60 * 1000, ttl: 30 * 24 * 60 * 60 * 1000 },
      );
      endTime(c, "get_profile_cache");

      // Fire off session restore in the background so it's warm for subsequent API calls,
      // and to keep the iron-session cookie's rolling TTL fresh even on cache hits.
      const cached = getCachedSessionClient(did);
      if (!cached || cached.needsSave) {
        sessionLazy.value.catch(() => {});
      }

      return result;
    });
    c.set("ctx", ctx);
    await next();
  };
}

/**
 * Flush KOSync progress that was written optimistically to `user_book` but
 * could not yet be persisted to the user's PDS (KOSync requests carry no OAuth
 * session). Runs opportunistically whenever we do have a session agent, and
 * routes through the canonical `updateBookRecord` writer so PDS + DB stay
 * consistent with every other book update.
 */
async function flushPendingSyncWrites(client: SessionClient, ctx: BookUtilContext): Promise<void> {
  const key = `sync_pending:${client.did}`;
  try {
    const pending = await ctx.kv.getItem<PendingWrite[]>(key);
    if (!pending?.length) return;

    const failed: PendingWrite[] = [];
    for (const entry of pending) {
      try {
        // Only bridge books the user still tracks; never auto-create from a sync.
        const userBook = await ctx.db
          .selectFrom("user_book")
          .select(["uri"])
          .where("userDid", "=", client.did)
          .where("hiveId", "=", entry.hiveId as HiveId)
          .executeTakeFirst();
        if (!userBook) continue;

        await updateBookRecord({
          ctx,
          agent: client,
          hiveId: entry.hiveId as HiveId,
          updates: { bookProgress: JSON.parse(entry.bookProgress) },
        });
      } catch {
        failed.push(entry);
      }
    }

    if (failed.length > 0) {
      await ctx.kv.setItem(key, failed);
    } else {
      await ctx.kv.removeItem(key);
    }
  } catch {
    // fire-and-forget — don't break the login flow
  }
}