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.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360import { Client, CredentialManager } from "@atcute/client";import type { ActorIdentifier } from "@atcute/lexicons/syntax";import type { Logger } from "pino";import type { SessionClient } from "../auth/client";import type { AppContext } from "../context";import { createActorResolver } from "../bsky/id-resolver";import { ids } from "../bsky/lexicon/ids";import type { BlobRef, HiveBook, HiveId } from "../types";import { loadGenresForHiveBook, loadGenresMapForHiveBooks } from "./hiveBookGenres.js";import { uploadImageBlob } from "./uploadImageBlob";
export async function createServiceAccountAgent( handle: string, appPassword: string,): Promise<SessionClient | null> { if (!handle || !appPassword) return null; try { const actor = await createActorResolver().resolve(handle as ActorIdentifier); const manager = new CredentialManager({ service: actor.pds }); await manager.login({ identifier: handle, password: appPassword }); const client = new Client({ handler: manager }); return { get did() { if (!manager.session?.did) { throw new Error("[catalogBookService] No active session on service account agent"); } return manager.session.did; }, get: client.get.bind(client) as SessionClient["get"], post: client.post.bind(client) as SessionClient["post"], }; } catch (err) { console.error("[catalogBookService] Failed to create service account agent:", err); return null; }}
type CatalogCtx = Pick<AppContext, "db"> & { serviceAccountAgent: AppContext["serviceAccountAgent"] | undefined; logger?: Logger;};
interface CatalogBlobs { thumbnailBlob?: BlobRef; coverBlob?: BlobRef;}
function safeJsonParse<T>(json: string, fallback: T, context: string): T { try { return JSON.parse(json) as T; } catch { console.warn(`[catalogBookService] JSON.parse failed for ${context}:`, json); return fallback; }}
function buildCatalogBookValue(book: HiveBook, blobs?: CatalogBlobs, catalogGenres?: string[]) { return { $type: ids.BuzzBookhiveCatalogBook, id: book.id, title: book.title, authors: book.authors, thumbnail: book.thumbnail, createdAt: book.createdAt, updatedAt: book.updatedAt, ...(blobs?.thumbnailBlob ? { thumbnailBlob: blobs.thumbnailBlob } : {}), ...(book.description ? { description: book.description.slice(0, 5000) } : {}), ...(book.cover ? { cover: book.cover } : {}), ...(blobs?.coverBlob ? { coverBlob: blobs.coverBlob } : {}), ...(book.source ? { source: book.source } : {}), ...(book.sourceUrl ? { sourceUrl: book.sourceUrl } : {}), ...(book.sourceId ? { sourceId: book.sourceId } : {}), ...(book.rating !== null && book.rating !== undefined ? { rating: Math.round(book.rating) } : {}), ...(book.ratingsCount !== null && book.ratingsCount !== undefined ? { ratingsCount: book.ratingsCount } : {}), ...(catalogGenres && catalogGenres.length > 0 ? { genres: catalogGenres } : {}), ...(book.series ? { series: book.series } : {}), ...(book.identifiers ? { identifiers: safeJsonParse<Record<string, string>>( book.identifiers, {}, `book ${book.id} identifiers`, ), } : {}), };}
/** * Upserts a single book to the @bookhive.buzz ATProto repo as a catalogBook record. * Skips if not yet enriched, or if already synced and updatedAt hasn't changed since. * Safe to call fire-and-forget. */export async function writeCatalogBookIfNeeded(ctx: CatalogCtx, bookId: HiveId): Promise<void> { if (!ctx.serviceAccountAgent) return;
// Fetch the full book fresh from DB to avoid races where enrichment runs // between when the caller triggered this and when we actually write the record. // Also gates on enrichedAt so we never write a partial record. const freshBook = await ctx.db .selectFrom("hive_book") .selectAll() .where("id", "=", bookId) .where("enrichedAt", "is not", null) .where((eb) => eb.or([ eb("hiveBookAtUri", "is", null), eb("hiveBookCatalogUpdatedAt", "is", null), eb(eb.ref("hiveBookCatalogUpdatedAt"), "<", eb.ref("updatedAt")), ]), ) .executeTakeFirst();
if (!freshBook) return;
const agent = ctx.serviceAccountAgent;
const [thumbnailBlob, coverBlob, catalogGenres] = await Promise.all([ uploadImageBlob(freshBook.thumbnail, agent), uploadImageBlob(freshBook.cover, agent), loadGenresForHiveBook(ctx.db, freshBook.id), ]);
const response = await agent.post("com.atproto.repo.putRecord", { input: { repo: agent.did, collection: ids.BuzzBookhiveCatalogBook, rkey: freshBook.id, record: buildCatalogBookValue(freshBook, { thumbnailBlob, coverBlob }, catalogGenres), }, });
if (!response.ok) return;
const uri = (response.data as { uri: string }).uri; if (!uri) return;
await ctx.db .updateTable("hive_book") .set({ hiveBookAtUri: uri, hiveBookCatalogUpdatedAt: freshBook.updatedAt }) .where("id", "=", freshBook.id) .execute();}
/** * Writes a batch of books to @bookhive.buzz using individual putRecord calls. * putRecord is idempotent (create-or-update), so this is safe to retry if the * process crashes between the ATProto write and the local DB update. * * Throws RateLimitError if the PDS rate limit is hit, so callers can back off. */export async function writeCatalogBooksBatch(ctx: CatalogCtx, books: HiveBook[]): Promise<void> { if (!ctx.serviceAccountAgent || books.length === 0) return;
const agent = ctx.serviceAccountAgent;
// Upload blobs for all books in parallel const blobResults = await Promise.all( books.map(async (book) => { const [thumbnailBlob, coverBlob] = await Promise.all([ uploadImageBlob(book.thumbnail, agent), uploadImageBlob(book.cover, agent), ]); return { thumbnailBlob, coverBlob }; }), ); const blobMap = new Map(books.map((book, i) => [book.id, blobResults[i]!])); const genresMap = await loadGenresMapForHiveBooks( ctx.db, books.map((b) => b.id), );
// Use putRecord (idempotent) for each book, limited to 5 concurrent to avoid rate limits const CONCURRENCY = 5; for (let i = 0; i < books.length; i += CONCURRENCY) { await Promise.all( books.slice(i, i + CONCURRENCY).map(async (book) => { const response = await agent.post("com.atproto.repo.putRecord", { input: { repo: agent.did, collection: ids.BuzzBookhiveCatalogBook, rkey: book.id, record: buildCatalogBookValue(book, blobMap.get(book.id), genresMap.get(book.id)), }, });
if (!response.ok) { const errData = response.data as { error?: string; message?: string } | undefined; if (errData?.error === "RateLimitExceeded") { throw new RateLimitError(); } ctx.logger?.error({ job: "catalog_book_put_record", outcome: "error", bookId: book.id, error: response.data, }); return; }
const uri = (response.data as { uri: string }).uri; if (!uri || uri === book.hiveBookAtUri) return;
await ctx.db .updateTable("hive_book") .set({ hiveBookAtUri: uri, hiveBookCatalogUpdatedAt: book.updatedAt }) .where("id", "=", book.id) .execute(); }), ); }}
class RateLimitError extends Error { constructor() { super("RateLimitExceeded"); this.name = "RateLimitError"; }}
export interface BackfillProgress { status: "idle" | "running" | "completed" | "failed"; startedAt: string | null; completedAt: string | null; written: number; batches: number; totalPending: number | null; lastBatchAt: string | null; error: string | null;}
let backfillProgress: BackfillProgress = { status: "idle", startedAt: null, completedAt: null, written: 0, batches: 0, totalPending: null, lastBatchAt: null, error: null,};
export function getBackfillProgress(): BackfillProgress { return { ...backfillProgress };}
/** * Iterates all enriched hive_book rows without hiveBookAtUri, writing them to @bookhive.buzz * in batches of 25. * * Rate limit target: 60% of the Bluesky relay limit (1,500 events/hr) to leave headroom * for real user activity → 900 events/hr → 36 batches/hr of 25 → 100s between batches. * * If a RateLimitExceeded error is returned mid-backfill (e.g. due to other activity on the * account), we pause for RATE_LIMIT_BACKOFF_MS before resuming. */export async function backfillCatalogBooks( ctx: CatalogCtx,): Promise<{ written: number; batches: number }> { if (!ctx.serviceAccountAgent) return { written: 0, batches: 0 };
const BATCH_SIZE = 25; // 100s between batches → 36 batches/hr → 900 relay events/hr (60% of relay 1,500/hr limit) const BATCH_DELAY_MS = 100_000; // If we unexpectedly hit a rate limit, wait 65 minutes before resuming const RATE_LIMIT_BACKOFF_MS = 65 * 60 * 1000;
let lastId = ""; let written = 0; let batches = 0; const startTime = Date.now();
const wideEvent: Record<string, unknown> = { job: "backfill_catalog_books", };
const totalRow = await ctx.db .selectFrom("hive_book") .where("hiveBookAtUri", "is", null) .where("enrichedAt", "is not", null) .select((eb) => eb.fn.countAll<number>().as("count")) .executeTakeFirst();
backfillProgress = { status: "running", startedAt: new Date().toISOString(), completedAt: null, written: 0, batches: 0, totalPending: totalRow ? Number(totalRow.count) : null, lastBatchAt: null, error: null, };
try { while (true) { const batch = await ctx.db .selectFrom("hive_book") .selectAll() .where("hiveBookAtUri", "is", null) .where("enrichedAt", "is not", null) .where("id", ">", lastId as HiveId) .orderBy("id") .limit(BATCH_SIZE) .execute();
if (batch.length === 0) break;
try { await writeCatalogBooksBatch(ctx, batch); } catch (err) { if (err instanceof RateLimitError) { ctx.logger?.warn({ job: "backfill_catalog_books", outcome: "rate_limited", written, batches, backoff_ms: RATE_LIMIT_BACKOFF_MS, }); await new Promise((r) => setTimeout(r, RATE_LIMIT_BACKOFF_MS)); // Retry the same batch (lastId not advanced) continue; } throw err; }
lastId = batch[batch.length - 1]!.id; written += batch.length; batches++; backfillProgress.written = written; backfillProgress.batches = batches; backfillProgress.lastBatchAt = new Date().toISOString();
await new Promise((r) => setTimeout(r, BATCH_DELAY_MS)); }
backfillProgress.status = "completed"; backfillProgress.completedAt = new Date().toISOString(); wideEvent["outcome"] = "success"; return { written, batches }; } catch (err) { backfillProgress.status = "failed"; backfillProgress.completedAt = new Date().toISOString(); backfillProgress.error = err instanceof Error ? err.message : String(err); wideEvent["outcome"] = "error"; wideEvent["error"] = err instanceof Error ? { message: err.message, type: err.name } : String(err); throw err; } finally { wideEvent["written"] = written; wideEvent["batches"] = batches; wideEvent["duration_ms"] = Date.now() - startTime; ctx.logger?.info(wideEvent); }}