Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/nperez0111/bookhive. Track your books, share your shelves, see what others are reading bookhive.buzz
atproto bluesky books bookshelf goodreads management-system
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720import 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 sessionexport 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 }}