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
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637/** * Import processing logic — runs in a Bun Worker thread. * * Design: parse CSV, then process each book sequentially — search, DB lookup, * accumulate matched books, flush PDS writes every BATCH_SIZE. Every step emits * SSE progress so the user always sees what's happening. */import type { SessionClient } from "../../auth/client";import type { BookIdentifiers, HiveId } from "../../types";import { Book as BookRecord } from "../../bsky/lexicon";import { getGoodreadsCsvParser, getStorygraphCsvParser, type GoodreadsBook, type StorygraphBook,} from "../../utils/csv";import { getUserRepoRecords, updateBookRecords, updateBookRecord } from "../../utils/getBook";import { normalizeStr, mapGoodreadsStatus, mapStorygraphStatus, mergeGoodreadsIdentifiers, mergeStorygraphIdentifiers, buildGoodreadsBookRecord, buildStorygraphBookRecord, deduplicateUnmatchedWithDetails,} from "../../utils/importBook";// Note: Worker threads get isolated metric registries, so metrics here won't appear// at the main /metrics endpoint. The main thread (routes/import.ts) tracks import// duration and active operations. Per-book counts are conveyed via SSE events.import { searchBooks } from "../../routes/lib";import type { ImportContext } from "./types";
/** Serialize an SSE payload with an auto-stamped ISO timestamp. */function sseJSON(payload: Record<string, unknown>): string { return JSON.stringify({ ...payload, ts: new Date().toISOString() });}
const BATCH_SIZE = 10;const SEARCH_CONCURRENCY = 3;
/** Read all rows from a CSV ReadableStream into an array. */async function drainStream<T>(stream: ReadableStream<T>): Promise<T[]> { const items: T[] = []; const reader = stream.getReader(); while (true) { const { done, value } = await reader.read(); if (done) break; items.push(value); } return items;}
type BookUpdate = Partial<BookRecord.Record> & { coverImage?: string; alreadyExists?: boolean;};
/** Flush a batch of matched books to the user's PDS. */async function flushBatch({ batch, ctx, agent, bookRecords, onSSE, id, matchedBooks, uploadedBooks, unmatchedBooks, totalBooks, makeFallbackBook,}: { batch: Map<HiveId, BookUpdate>; ctx: ImportContext; agent: SessionClient; bookRecords: Promise<{ books: Map<string, BookRecord.Record> }>; onSSE: (data: string) => void | Promise<void>; id: { value: number }; matchedBooks: { value: number }; uploadedBooks: { value: number }; unmatchedBooks: Array<{ book: any; reason: string }>; totalBooks: number; makeFallbackBook: (bookUpdate: BookUpdate) => any;}): Promise<void> { if (batch.size === 0) return;
await onSSE( sseJSON({ event: "batch-save", stage: "uploading", stageProgress: { current: matchedBooks.value, total: totalBooks, message: `Saving ${batch.size} books to your library…`, }, id: id.value++, }), );
try { await updateBookRecords({ ctx, agent, updates: batch, bookRecords }); } catch (error) { ctx.addWideEventContext({ import_batch_update: "failed", error: error instanceof Error ? error.message : String(error), book_count: batch.size, }); let individualSuccesses = 0; let individualFailures = 0; for (const [hiveId, bookUpdate] of batch.entries()) { try { await updateBookRecord({ ctx, agent, hiveId, updates: bookUpdate }); individualSuccesses++; } catch (individualError) { individualFailures++; ctx.addWideEventContext({ import_individual_book: "failed", error: individualError instanceof Error ? individualError.message : String(individualError), hiveId, }); unmatchedBooks.push({ book: makeFallbackBook(bookUpdate), reason: "update_error" }); } } if (individualFailures > 0) { await onSSE( sseJSON({ event: "import-error", stage: "uploading", stageProgress: { current: matchedBooks.value, total: totalBooks, message: `Individual save completed: ${individualSuccesses} succeeded, ${individualFailures} failed`, }, error: `Individual save: ${individualSuccesses} succeeded, ${individualFailures} failed`, id: id.value++, }), ); } }
// Emit per-book upload events after the PDS write confirms const startProcessed = matchedBooks.value; let idx = 0; for (const book of batch.values()) { const processed = startProcessed + ++idx; const alreadyExists = !!(book as { alreadyExists?: boolean })["alreadyExists"]; if (!alreadyExists) { uploadedBooks.value++; } await onSSE( sseJSON({ title: book?.["title"], author: book?.["authors"], uploaded: 1, processed, failed: unmatchedBooks.length, total: totalBooks, event: "book-upload", stage: "uploading", stageProgress: { current: processed, total: totalBooks, message: `Uploading books to your library (${processed}/${totalBooks})`, }, book: book ? { hiveId: book["hiveId"], title: book["title"], authors: book["authors"], coverImage: book["coverImage"], status: book["status"], finishedAt: book["finishedAt"], stars: book["stars"], review: book["review"], alreadyExists: book["alreadyExists"], } : undefined, id: id.value++, }), ); } matchedBooks.value += batch.size;}
// ─── Goodreads ───────────────────────────────────────────────────────────────
export async function processGoodreadsImport({ csvData, ctx, agent, onSSE,}: { csvData: ArrayBuffer; ctx: ImportContext; agent: SessionClient; onSSE: (data: string) => void | Promise<void>;}): Promise<void> { const id = { value: 0 }; const matchedBooks = { value: 0 }; const uploadedBooks = { value: 0 }; const unmatchedBooks: Array<{ book: GoodreadsBook; reason: string }> = []; const unmatchedSet = new Set<string>();
await onSSE( sseJSON({ event: "import-start", stage: "initializing", stageProgress: { message: "Reading CSV file..." }, id: id.value++, }), );
// Phase 1: Parse entire CSV const allBooks = await drainStream( new Blob([csvData]).stream().pipeThrough(getGoodreadsCsvParser()), ); const totalBooks = allBooks.length;
// Start fetching user's existing PDS records in the background const bookRecords = getUserRepoRecords({ ctx, agent }); const existingHiveIdsPromise = bookRecords.then( (br) => new Set(Array.from(br.books.values()).map((b) => b.hiveId)), );
// Phase 2: Search in groups of SEARCH_CONCURRENCY, process results in order let currentBatch = new Map<HiveId, BookUpdate>(); const grFallback = (bu: BookUpdate) => ({ bookId: "", title: bu.title || "Unknown", author: bu.authors || "Unknown", authorLastFirst: "", additionalAuthors: [], isbn: "", isbn13: "", myRating: bu.stars ? bu.stars / 2 : 0, averageRating: 0, publisher: "", binding: "", numberOfPages: 0, yearPublished: 0, originalPublicationYear: 0, dateRead: bu.finishedAt ? new Date(bu.finishedAt) : null, dateAdded: new Date(), bookshelves: [], bookshelvesWithPositions: "", exclusiveShelf: "", myReview: bu.review || "", spoiler: false, privateNotes: "", readCount: 0, ownedCopies: 0, }) as GoodreadsBook;
for (let i = 0; i < allBooks.length; i += SEARCH_CONCURRENCY) { const chunk = allBooks.slice(i, i + SEARCH_CONCURRENCY);
// Fire all searches in this chunk (non-blocking) const searches = chunk.map((book) => searchBooks({ query: book.title, ctx }));
// Process each book in order, awaiting its search for (let j = 0; j < chunk.length; j++) { const book = chunk[j]!; const bookIdx = i + j;
await onSSE( sseJSON({ title: book.title, author: book.author, processed: matchedBooks.value, failed: unmatchedBooks.length, total: totalBooks, event: "book-load", stage: "searching", stageProgress: { current: bookIdx + 1, total: totalBooks, message: `Looking up "${book.title}"…`, }, id: id.value++, }), );
// Await this book's search (others in the chunk are already in-flight) await searches[j];
const hiveBook = await ctx.db .selectFrom("hive_book") .select(["id", "title", "cover", "identifiers"]) .where("hive_book.rawTitle", "=", book.title) .where("authors", "=", book.author) .executeTakeFirst();
if (!hiveBook) { const key = `${normalizeStr(book.title)}::${normalizeStr(book.author)}`; if (!unmatchedSet.has(key)) { unmatchedSet.add(key); unmatchedBooks.push({ book, reason: "no_match" }); } continue; }
const existingIdentifiers: BookIdentifiers = hiveBook.identifiers ? JSON.parse(hiveBook.identifiers) : {}; const { identifiers: newIdentifiers, changed } = mergeGoodreadsIdentifiers({ bookId: book.bookId, isbn: book.isbn, isbn13: book.isbn13, existingIdentifiers, hiveBookId: hiveBook.id, }); if (changed) { const updatedAt = new Date().toISOString(); await ctx.db .updateTable("hive_book") .set({ identifiers: JSON.stringify(newIdentifiers), updatedAt, }) .where("id", "=", hiveBook.id) .execute(); // Keep book_id_map in sync so findBookIdentifiersByLookup sees the merged IDs await ctx.db .insertInto("book_id_map") .values({ hiveId: hiveBook.id as HiveId, isbn: newIdentifiers.isbn10 ?? null, isbn13: newIdentifiers.isbn13 ?? null, goodreadsId: newIdentifiers.goodreadsId ?? null, updatedAt, }) .onConflict((oc) => oc.column("hiveId").doUpdateSet((eb) => ({ isbn: eb.ref("excluded.isbn"), isbn13: eb.ref("excluded.isbn13"), goodreadsId: eb.ref("excluded.goodreadsId"), updatedAt: eb.ref("excluded.updatedAt"), })), ) .execute(); }
const existingHiveIds = await existingHiveIdsPromise; currentBatch.set( hiveBook.id as HiveId, buildGoodreadsBookRecord({ book, hiveBook, existingHiveIds }), );
if (currentBatch.size >= BATCH_SIZE) { await flushBatch({ batch: currentBatch, ctx, agent, bookRecords, onSSE, id, matchedBooks, uploadedBooks, unmatchedBooks, totalBooks, makeFallbackBook: grFallback, }); currentBatch = new Map(); } } // end inner for (j) } // end outer for (i += SEARCH_CONCURRENCY)
// Flush remaining await flushBatch({ batch: currentBatch, ctx, agent, bookRecords, onSSE, id, matchedBooks, uploadedBooks, unmatchedBooks, totalBooks, makeFallbackBook: grFallback, });
await onSSE( sseJSON({ event: "import-complete", stage: "complete", stageProgress: { current: matchedBooks.value, total: totalBooks, message: `Import complete! Successfully imported ${uploadedBooks.value} books${unmatchedBooks.length > 0 ? ` (${unmatchedBooks.length} failed)` : ""}`, }, ...deduplicateUnmatchedWithDetails( unmatchedBooks, (b) => b.title, (b) => b.author, (b) => ({ title: b.book.title, author: b.book.author, isbn10: b.book.isbn || undefined, isbn13: b.book.isbn13 || undefined, stars: b.book.myRating ? b.book.myRating * 2 : undefined, review: b.book.myReview || undefined, finishedAt: b.book.dateRead ? b.book.dateRead.toISOString() : undefined, status: mapGoodreadsStatus(b.book), reason: b.reason, }), ), id: id.value++, }), );}
// ─── StoryGraph ──────────────────────────────────────────────────────────────
export async function processStorygraphImport({ csvData, ctx, agent, onSSE,}: { csvData: ArrayBuffer; ctx: ImportContext; agent: SessionClient; onSSE: (data: string) => void | Promise<void>;}): Promise<void> { const id = { value: 0 }; const matchedBooks = { value: 0 }; const uploadedBooks = { value: 0 }; const unmatchedBooks: Array<{ book: StorygraphBook; reason: string }> = []; const unmatchedSet = new Set<string>();
await onSSE( sseJSON({ event: "import-start", stage: "initializing", stageProgress: { message: "Reading CSV file..." }, id: id.value++, }), );
const allBooks = await drainStream( new Blob([csvData]).stream().pipeThrough(getStorygraphCsvParser()), ); const totalBooks = allBooks.length;
const bookRecords = getUserRepoRecords({ ctx, agent }); const existingHiveIdsPromise = bookRecords.then( (br) => new Set(Array.from(br.books.values()).map((b) => b.hiveId)), );
let currentBatch = new Map<HiveId, BookUpdate>(); const sgFallback = (bu: BookUpdate) => ({ title: bu.title || "Unknown", authors: bu.authors || "Unknown", contributors: "", isbn: "", format: "", readStatus: "", dateAdded: null, lastDateRead: bu.finishedAt ? new Date(bu.finishedAt) : null, datesRead: "", readCount: 0, moods: "", pace: "", characterOrPlot: "", strongCharacterDevelopment: "", loveableCharacters: "", diverseCharacters: "", flawedCharacters: "", starRating: bu.stars ? bu.stars / 2 : 0, review: bu.review || "", contentWarnings: "", contentWarningDescription: "", tags: "", owned: false, }) as StorygraphBook;
for (let i = 0; i < allBooks.length; i += SEARCH_CONCURRENCY) { const chunk = allBooks.slice(i, i + SEARCH_CONCURRENCY);
// Fire all searches in this chunk (non-blocking) const searches = chunk.map((book) => searchBooks({ query: book.title, ctx }));
for (let j = 0; j < chunk.length; j++) { const book = chunk[j]!; const bookIdx = i + j;
await onSSE( sseJSON({ title: book.title, author: book.authors, processed: matchedBooks.value, failed: unmatchedBooks.length, total: totalBooks, event: "book-load", stage: "searching", stageProgress: { current: bookIdx + 1, total: totalBooks, message: `Looking up "${book.title}"…`, }, id: id.value++, }), );
await searches[j];
const hiveBook = await ctx.db .selectFrom("hive_book") .select(["id", "title", "cover", "identifiers"]) .where("hive_book.rawTitle", "=", book.title) .where("authors", "=", book.authors) .executeTakeFirst();
if (!hiveBook) { const key = `${normalizeStr(book.title)}::${normalizeStr(book.authors)}`; if (!unmatchedSet.has(key)) { unmatchedSet.add(key); unmatchedBooks.push({ book, reason: "no_match" }); } continue; }
const existingIdentifiers: BookIdentifiers = hiveBook.identifiers ? JSON.parse(hiveBook.identifiers) : {}; const { identifiers: newIdentifiers, changed } = mergeStorygraphIdentifiers({ isbn: book.isbn, existingIdentifiers, hiveBookId: hiveBook.id, }); if (changed) { const updatedAt = new Date().toISOString(); await ctx.db .updateTable("hive_book") .set({ identifiers: JSON.stringify(newIdentifiers), updatedAt, }) .where("id", "=", hiveBook.id) .execute(); // Keep book_id_map in sync so findBookIdentifiersByLookup sees the merged IDs await ctx.db .insertInto("book_id_map") .values({ hiveId: hiveBook.id as HiveId, isbn: newIdentifiers.isbn10 ?? null, isbn13: newIdentifiers.isbn13 ?? null, goodreadsId: newIdentifiers.goodreadsId ?? null, updatedAt, }) .onConflict((oc) => oc.column("hiveId").doUpdateSet((eb) => ({ isbn: eb.ref("excluded.isbn"), isbn13: eb.ref("excluded.isbn13"), goodreadsId: eb.ref("excluded.goodreadsId"), updatedAt: eb.ref("excluded.updatedAt"), })), ) .execute(); }
const existingHiveIds = await existingHiveIdsPromise; currentBatch.set( hiveBook.id as HiveId, buildStorygraphBookRecord({ book, hiveBook, existingHiveIds }), );
if (currentBatch.size >= BATCH_SIZE) { await flushBatch({ batch: currentBatch, ctx, agent, bookRecords, onSSE, id, matchedBooks, uploadedBooks, unmatchedBooks, totalBooks, makeFallbackBook: sgFallback, }); currentBatch = new Map(); } } }
await flushBatch({ batch: currentBatch, ctx, agent, bookRecords, onSSE, id, matchedBooks, uploadedBooks, unmatchedBooks, totalBooks, makeFallbackBook: sgFallback, });
await onSSE( sseJSON({ event: "import-complete", stage: "complete", stageProgress: { current: matchedBooks.value, total: totalBooks, message: `Import complete! Successfully imported ${uploadedBooks.value} books${unmatchedBooks.length > 0 ? ` (${unmatchedBooks.length} failed)` : ""}`, }, ...deduplicateUnmatchedWithDetails( unmatchedBooks, (b) => b.title, (b) => b.authors, (b) => { const cleanIsbn = b.book.isbn?.replace(/[-\s]/g, "") || ""; return { title: b.book.title, author: b.book.authors, isbn10: cleanIsbn.length === 10 ? cleanIsbn : undefined, isbn13: cleanIsbn.length === 13 ? cleanIsbn : undefined, stars: b.book.starRating ? b.book.starRating * 2 : undefined, review: b.book.review || undefined, finishedAt: b.book.lastDateRead ? b.book.lastDateRead.toISOString() : undefined, status: mapStorygraphStatus(b.book), reason: b.reason, }; }, ), id: id.value++, }), );}