diff --git a/server/src/modules/migration/executor/migration-import.repository.test.ts b/server/src/modules/migration/executor/migration-import.repository.test.ts index 48502834d..512a47f12 100644 --- a/server/src/modules/migration/executor/migration-import.repository.test.ts +++ b/server/src/modules/migration/executor/migration-import.repository.test.ts @@ -111,6 +111,19 @@ describe('MigrationImportRepository', () => { expect(where).toHaveBeenCalledTimes(1); }); + it('chunks user/book clear queries for large migration runs', async () => { + const where = vi.fn().mockResolvedValue(undefined); + const deleteFn = vi.fn().mockReturnValue({ where }); + const repo = new MigrationImportRepository({ delete: deleteFn } as never); + const userIds = Array.from({ length: 501 }, (_, index) => index + 1); + const bookIds = Array.from({ length: 1001 }, (_, index) => index + 1); + + await repo.clearUserBookStatuses(userIds, bookIds); + + expect(deleteFn).toHaveBeenCalledTimes(6); + expect(where).toHaveBeenCalledTimes(6); + }); + it('batchUpsertAuthors returns name-to-id mapping from inserted rows', async () => { const returning = vi.fn().mockResolvedValue([ { id: 10, name: 'Frank Herbert' }, @@ -161,6 +174,37 @@ describe('MigrationImportRepository', () => { expect(result.audiobookPrimaryFilesByBookId).toEqual(new Map([[2, 202]])); }); + it('chunks primary file lookups for large matched book sets', async () => { + const where = vi + .fn() + .mockResolvedValueOnce([{ bookId: 1, primaryFileId: 101, primaryFileFormat: 'epub' }]) + .mockResolvedValueOnce([{ bookId: 501, primaryFileId: 50101, primaryFileFormat: 'm4b' }]) + .mockResolvedValueOnce([{ bookId: 1001, primaryFileId: 100101, primaryFileFormat: 'mp3' }]); + const leftJoin = vi.fn().mockReturnValue({ where }); + const from = vi.fn().mockReturnValue({ leftJoin }); + const select = vi.fn().mockReturnValue({ from }); + const repo = new MigrationImportRepository({ select } as never); + const bookIds = Array.from({ length: 1001 }, (_, index) => index + 1); + bookIds.push(1, Number.NaN); + + const result = await repo.fetchTargetBookPrimaryFiles(bookIds); + + expect(where).toHaveBeenCalledTimes(3); + expect(result.primaryFilesByBookId).toEqual( + new Map([ + [1, 101], + [501, 50101], + [1001, 100101], + ]), + ); + expect(result.audiobookPrimaryFilesByBookId).toEqual( + new Map([ + [501, 50101], + [1001, 100101], + ]), + ); + }); + it('groups target files by book id and supports empty input short-circuit', async () => { const where = vi.fn().mockResolvedValue([ { id: 11, bookId: 1, hash: 'h1', absolutePath: '/books/1.epub' }, @@ -187,6 +231,26 @@ describe('MigrationImportRepository', () => { ); }); + it('chunks target file lookups for large matched book sets', async () => { + const where = vi + .fn() + .mockResolvedValueOnce([{ id: 11, bookId: 1, hash: 'h1', absolutePath: '/books/1.epub', format: 'epub' }]) + .mockResolvedValueOnce([{ id: 5011, bookId: 501, hash: 'h501', absolutePath: '/books/501.epub', format: 'epub' }]) + .mockResolvedValueOnce([{ id: 10011, bookId: 1001, hash: 'h1001', absolutePath: '/books/1001.epub', format: 'epub' }]); + const from = vi.fn().mockReturnValue({ where }); + const select = vi.fn().mockReturnValue({ from }); + const repo = new MigrationImportRepository({ select } as never); + const bookIds = Array.from({ length: 1001 }, (_, index) => index + 1); + bookIds.push(1); + + const result = await repo.fetchTargetBookFiles(bookIds); + + expect(where).toHaveBeenCalledTimes(3); + expect(result.get(1)).toEqual([{ id: 11, hash: 'h1', absolutePath: '/books/1.epub', format: 'epub' }]); + expect(result.get(501)).toEqual([{ id: 5011, hash: 'h501', absolutePath: '/books/501.epub', format: 'epub' }]); + expect(result.get(1001)).toEqual([{ id: 10011, hash: 'h1001', absolutePath: '/books/1001.epub', format: 'epub' }]); + }); + it('returns library-id map for book ids and skips query for empty input', async () => { const where = vi.fn().mockResolvedValue([ { id: 1, libraryId: 10 }, @@ -206,6 +270,28 @@ describe('MigrationImportRepository', () => { ); }); + it('chunks library-id lookups for large matched book sets', async () => { + const where = vi + .fn() + .mockResolvedValueOnce([{ id: 1, libraryId: 10 }]) + .mockResolvedValueOnce([{ id: 501, libraryId: 50 }]) + .mockResolvedValueOnce([{ id: 1001, libraryId: 100 }]); + const from = vi.fn().mockReturnValue({ where }); + const select = vi.fn().mockReturnValue({ from }); + const repo = new MigrationImportRepository({ select } as never); + const bookIds = Array.from({ length: 1001 }, (_, index) => index + 1); + bookIds.push(1); + + await expect(repo.fetchLibraryIdsByBookIds(bookIds)).resolves.toEqual( + new Map([ + [1, 10], + [501, 50], + [1001, 100], + ]), + ); + expect(where).toHaveBeenCalledTimes(3); + }); + it('upsertCollectionBook reflects whether insert produced a row', async () => { const returning = vi .fn() diff --git a/server/src/modules/migration/executor/migration-import.repository.ts b/server/src/modules/migration/executor/migration-import.repository.ts index d209c6c27..ac207b803 100644 --- a/server/src/modules/migration/executor/migration-import.repository.ts +++ b/server/src/modules/migration/executor/migration-import.repository.ts @@ -201,9 +201,13 @@ export class MigrationImportRepository { const targetUserIds = uniqueNumbers(userIds); const targetBookIds = uniqueNumbers(bookIds); if (targetUserIds.length === 0 || targetBookIds.length === 0) return; - await this.db - .delete(schema.userBookStatus) - .where(and(inArray(schema.userBookStatus.userId, targetUserIds), inArray(schema.userBookStatus.bookId, targetBookIds))); + for (const userBatch of chunk(targetUserIds, BATCH_CHUNK_SIZE)) { + for (const bookBatch of chunk(targetBookIds, BATCH_CHUNK_SIZE)) { + await this.db + .delete(schema.userBookStatus) + .where(and(inArray(schema.userBookStatus.userId, userBatch), inArray(schema.userBookStatus.bookId, bookBatch))); + } + } } async upsertUserBookStatus(values: typeof schema.userBookStatus.$inferInsert): Promise { @@ -220,9 +224,13 @@ export class MigrationImportRepository { const targetUserIds = uniqueNumbers(userIds); const targetBookIds = uniqueNumbers(bookIds); if (targetUserIds.length === 0 || targetBookIds.length === 0) return; - await this.db - .delete(schema.userBookRatings) - .where(and(inArray(schema.userBookRatings.userId, targetUserIds), inArray(schema.userBookRatings.bookId, targetBookIds))); + for (const userBatch of chunk(targetUserIds, BATCH_CHUNK_SIZE)) { + for (const bookBatch of chunk(targetBookIds, BATCH_CHUNK_SIZE)) { + await this.db + .delete(schema.userBookRatings) + .where(and(inArray(schema.userBookRatings.userId, userBatch), inArray(schema.userBookRatings.bookId, bookBatch))); + } + } } // --- Reading progress --- @@ -231,9 +239,13 @@ export class MigrationImportRepository { const targetUserIds = uniqueNumbers(userIds); const targetFileIds = uniqueNumbers(fileIds); if (targetUserIds.length === 0 || targetFileIds.length === 0) return; - await this.db - .delete(schema.readingProgress) - .where(and(inArray(schema.readingProgress.userId, targetUserIds), inArray(schema.readingProgress.bookFileId, targetFileIds))); + for (const userBatch of chunk(targetUserIds, BATCH_CHUNK_SIZE)) { + for (const fileBatch of chunk(targetFileIds, BATCH_CHUNK_SIZE)) { + await this.db + .delete(schema.readingProgress) + .where(and(inArray(schema.readingProgress.userId, userBatch), inArray(schema.readingProgress.bookFileId, fileBatch))); + } + } } async upsertReadingProgress(values: typeof schema.readingProgress.$inferInsert): Promise { @@ -252,9 +264,13 @@ export class MigrationImportRepository { const targetUserIds = uniqueNumbers(userIds); const targetBookIds = uniqueNumbers(bookIds); if (targetUserIds.length === 0 || targetBookIds.length === 0) return; - await this.db - .delete(schema.audiobookProgress) - .where(and(inArray(schema.audiobookProgress.userId, targetUserIds), inArray(schema.audiobookProgress.bookId, targetBookIds))); + for (const userBatch of chunk(targetUserIds, BATCH_CHUNK_SIZE)) { + for (const bookBatch of chunk(targetBookIds, BATCH_CHUNK_SIZE)) { + await this.db + .delete(schema.audiobookProgress) + .where(and(inArray(schema.audiobookProgress.userId, userBatch), inArray(schema.audiobookProgress.bookId, bookBatch))); + } + } } async upsertAudiobookProgress(values: typeof schema.audiobookProgress.$inferInsert): Promise { @@ -273,9 +289,11 @@ export class MigrationImportRepository { const targetUserIds = uniqueNumbers(userIds); const targetBookIds = uniqueNumbers(bookIds); if (targetUserIds.length === 0 || targetBookIds.length === 0) return; - await this.db - .delete(schema.bookmarks) - .where(and(inArray(schema.bookmarks.userId, targetUserIds), inArray(schema.bookmarks.bookId, targetBookIds))); + for (const userBatch of chunk(targetUserIds, BATCH_CHUNK_SIZE)) { + for (const bookBatch of chunk(targetBookIds, BATCH_CHUNK_SIZE)) { + await this.db.delete(schema.bookmarks).where(and(inArray(schema.bookmarks.userId, userBatch), inArray(schema.bookmarks.bookId, bookBatch))); + } + } } async insertBookmark(values: typeof schema.bookmarks.$inferInsert): Promise { @@ -289,15 +307,15 @@ export class MigrationImportRepository { const targetBookIds = uniqueNumbers(bookIds); if (targetUserIds.length === 0 || targetBookIds.length === 0) return; // Only web-origin rows are replaced by an import; device-synced annotations stay. - await this.db - .delete(schema.annotations) - .where( - and( - inArray(schema.annotations.userId, targetUserIds), - inArray(schema.annotations.bookId, targetBookIds), - eq(schema.annotations.origin, 'web'), - ), - ); + for (const userBatch of chunk(targetUserIds, BATCH_CHUNK_SIZE)) { + for (const bookBatch of chunk(targetBookIds, BATCH_CHUNK_SIZE)) { + await this.db + .delete(schema.annotations) + .where( + and(inArray(schema.annotations.userId, userBatch), inArray(schema.annotations.bookId, bookBatch), eq(schema.annotations.origin, 'web')), + ); + } + } } async insertAnnotation(values: typeof schema.annotations.$inferInsert & { cfi: string }): Promise { @@ -312,8 +330,13 @@ export class MigrationImportRepository { // --- Collections --- async fetchExistingCollections(userIds: number[]): Promise> { - if (userIds.length === 0) return []; - return this.db.select().from(schema.collections).where(inArray(schema.collections.userId, userIds)); + const targetUserIds = uniqueNumbers(userIds); + if (targetUserIds.length === 0) return []; + const collections: Array = []; + for (const userBatch of chunk(targetUserIds, BATCH_CHUNK_SIZE)) { + collections.push(...(await this.db.select().from(schema.collections).where(inArray(schema.collections.userId, userBatch)))); + } + return collections; } async insertCollection(values: typeof schema.collections.$inferInsert): Promise<{ id: number }> { @@ -560,21 +583,36 @@ export class MigrationImportRepository { const targetBookIds = uniqueNumbers(params.bookIds); if (targetUserIds.length === 0 || targetBookIds.length === 0) return; - const existing = await this.db - .select({ - id: schema.readingSessions.id, - userId: schema.readingSessions.userId, - bookId: schema.readingSessions.bookId, - sessionId: schema.readingSessions.sessionId, - startedAt: schema.readingSessions.startedAt, - endedAt: schema.readingSessions.endedAt, - durationSeconds: schema.readingSessions.durationSeconds, - progressDelta: schema.readingSessions.progressDelta, - libraryId: schema.books.libraryId, - }) - .from(schema.readingSessions) - .innerJoin(schema.books, eq(schema.books.id, schema.readingSessions.bookId)) - .where(and(inArray(schema.readingSessions.userId, targetUserIds), like(schema.readingSessions.sessionId, `${params.sessionIdPrefix}%`))); + const existing: Array<{ + id: number; + userId: number; + bookId: number; + sessionId: string; + startedAt: Date; + endedAt: Date; + durationSeconds: number; + progressDelta: number | null; + libraryId: number; + }> = []; + for (const userBatch of chunk(targetUserIds, BATCH_CHUNK_SIZE)) { + existing.push( + ...(await this.db + .select({ + id: schema.readingSessions.id, + userId: schema.readingSessions.userId, + bookId: schema.readingSessions.bookId, + sessionId: schema.readingSessions.sessionId, + startedAt: schema.readingSessions.startedAt, + endedAt: schema.readingSessions.endedAt, + durationSeconds: schema.readingSessions.durationSeconds, + progressDelta: schema.readingSessions.progressDelta, + libraryId: schema.books.libraryId, + }) + .from(schema.readingSessions) + .innerJoin(schema.books, eq(schema.books.id, schema.readingSessions.bookId)) + .where(and(inArray(schema.readingSessions.userId, userBatch), like(schema.readingSessions.sessionId, `${params.sessionIdPrefix}%`)))), + ); + } const desiredKeys = new Set(params.items.map((item) => `${item.userId}:${item.sessionId}`)); const targetBookIdSet = new Set(targetBookIds); @@ -711,21 +749,23 @@ export class MigrationImportRepository { const { isAudioFormat } = await import('@bookorbit/types'); - const rows = await this.db - .select({ - bookId: schema.books.id, - primaryFileId: schema.books.primaryFileId, - primaryFileFormat: schema.bookFiles.format, - }) - .from(schema.books) - .leftJoin(schema.bookFiles, eq(schema.bookFiles.id, schema.books.primaryFileId)) - .where(inArray(schema.books.id, bookIds)); + for (const batch of chunk(uniqueNumbers(bookIds), BATCH_CHUNK_SIZE)) { + const rows = await this.db + .select({ + bookId: schema.books.id, + primaryFileId: schema.books.primaryFileId, + primaryFileFormat: schema.bookFiles.format, + }) + .from(schema.books) + .leftJoin(schema.bookFiles, eq(schema.bookFiles.id, schema.books.primaryFileId)) + .where(inArray(schema.books.id, batch)); - for (const row of rows) { - if (!row.primaryFileId) continue; - primaryFilesByBookId.set(row.bookId, row.primaryFileId); - if (row.primaryFileFormat && isAudioFormat(row.primaryFileFormat)) { - audiobookPrimaryFilesByBookId.set(row.bookId, row.primaryFileId); + for (const row of rows) { + if (!row.primaryFileId) continue; + primaryFilesByBookId.set(row.bookId, row.primaryFileId); + if (row.primaryFileFormat && isAudioFormat(row.primaryFileFormat)) { + audiobookPrimaryFilesByBookId.set(row.bookId, row.primaryFileId); + } } } @@ -736,23 +776,26 @@ export class MigrationImportRepository { bookIds: number[], ): Promise>> { const result = new Map>(); - if (bookIds.length === 0) return result; + const targetBookIds = uniqueNumbers(bookIds); + if (targetBookIds.length === 0) return result; - const rows = await this.db - .select({ - id: schema.bookFiles.id, - bookId: schema.bookFiles.bookId, - hash: schema.bookFiles.fileHash, - absolutePath: schema.bookFiles.absolutePath, - format: schema.bookFiles.format, - }) - .from(schema.bookFiles) - .where(inArray(schema.bookFiles.bookId, bookIds)); + for (const batch of chunk(targetBookIds, BATCH_CHUNK_SIZE)) { + const rows = await this.db + .select({ + id: schema.bookFiles.id, + bookId: schema.bookFiles.bookId, + hash: schema.bookFiles.fileHash, + absolutePath: schema.bookFiles.absolutePath, + format: schema.bookFiles.format, + }) + .from(schema.bookFiles) + .where(inArray(schema.bookFiles.bookId, batch)); - for (const row of rows) { - const files = result.get(row.bookId) ?? []; - files.push({ id: row.id, hash: row.hash, absolutePath: row.absolutePath, format: row.format }); - result.set(row.bookId, files); + for (const row of rows) { + const files = result.get(row.bookId) ?? []; + files.push({ id: row.id, hash: row.hash, absolutePath: row.absolutePath, format: row.format }); + result.set(row.bookId, files); + } } return result; @@ -760,15 +803,18 @@ export class MigrationImportRepository { async fetchLibraryIdsByBookIds(bookIds: number[]): Promise> { const result = new Map(); - if (bookIds.length === 0) return result; + const targetBookIds = uniqueNumbers(bookIds); + if (targetBookIds.length === 0) return result; - const rows = await this.db - .select({ id: schema.books.id, libraryId: schema.books.libraryId }) - .from(schema.books) - .where(inArray(schema.books.id, bookIds)); + for (const batch of chunk(targetBookIds, BATCH_CHUNK_SIZE)) { + const rows = await this.db + .select({ id: schema.books.id, libraryId: schema.books.libraryId }) + .from(schema.books) + .where(inArray(schema.books.id, batch)); - for (const row of rows) { - result.set(row.id, row.libraryId); + for (const row of rows) { + result.set(row.id, row.libraryId); + } } return result; @@ -779,13 +825,15 @@ export class MigrationImportRepository { const targetUserIds = uniqueNumbers(userIds); if (targetUserIds.length === 0) return result; - const rows = await this.db - .select({ id: schema.users.id, settings: schema.users.settings }) - .from(schema.users) - .where(inArray(schema.users.id, targetUserIds)); + for (const userBatch of chunk(targetUserIds, BATCH_CHUNK_SIZE)) { + const rows = await this.db + .select({ id: schema.users.id, settings: schema.users.settings }) + .from(schema.users) + .where(inArray(schema.users.id, userBatch)); - for (const row of rows) { - result.set(row.id, resolveTimeZone((row.settings as { timezone?: unknown } | undefined)?.timezone, 'UTC')); + for (const row of rows) { + result.set(row.id, resolveTimeZone((row.settings as { timezone?: unknown } | undefined)?.timezone, 'UTC')); + } } return result; } @@ -796,15 +844,17 @@ export class MigrationImportRepository { await this.lockReadingDailyStats(userId, libraryId); - await this.db - .delete(schema.userReadingDailyStats) - .where( - and( - eq(schema.userReadingDailyStats.userId, userId), - eq(schema.userReadingDailyStats.libraryId, libraryId), - inArray(schema.userReadingDailyStats.day, affectedDays), - ), - ); + for (const dayBatch of chunk(affectedDays, BATCH_CHUNK_SIZE)) { + await this.db + .delete(schema.userReadingDailyStats) + .where( + and( + eq(schema.userReadingDailyStats.userId, userId), + eq(schema.userReadingDailyStats.libraryId, libraryId), + inArray(schema.userReadingDailyStats.day, dayBatch), + ), + ); + } const range = getDayRangeForDateKeys(affectedDays, timeZone); if (!range) return; @@ -848,27 +898,29 @@ export class MigrationImportRepository { if (segments.length === 0) return; const now = new Date(); - await this.db - .insert(schema.userReadingDailyStats) - .values( - segments.map((segment) => ({ - userId, - libraryId, - day: segment.day, - readingSeconds: segment.readingSeconds, - progressDelta: segment.progressDelta, - sessionsCount: segment.sessionsCount, - updatedAt: now, - })), - ) - .onConflictDoUpdate({ - target: [schema.userReadingDailyStats.userId, schema.userReadingDailyStats.libraryId, schema.userReadingDailyStats.day], - set: { - readingSeconds: sql`excluded.reading_seconds`, - progressDelta: sql`excluded.progress_delta`, - sessionsCount: sql`excluded.sessions_count`, - updatedAt: now, - }, - }); + for (const batch of chunk(segments, BATCH_CHUNK_SIZE)) { + await this.db + .insert(schema.userReadingDailyStats) + .values( + batch.map((segment) => ({ + userId, + libraryId, + day: segment.day, + readingSeconds: segment.readingSeconds, + progressDelta: segment.progressDelta, + sessionsCount: segment.sessionsCount, + updatedAt: now, + })), + ) + .onConflictDoUpdate({ + target: [schema.userReadingDailyStats.userId, schema.userReadingDailyStats.libraryId, schema.userReadingDailyStats.day], + set: { + readingSeconds: sql`excluded.reading_seconds`, + progressDelta: sql`excluded.progress_delta`, + sessionsCount: sql`excluded.sessions_count`, + updatedAt: now, + }, + }); + } } } diff --git a/server/src/modules/migration/migration.repository.test.ts b/server/src/modules/migration/migration.repository.test.ts index e011cdb82..72cf63b76 100644 --- a/server/src/modules/migration/migration.repository.test.ts +++ b/server/src/modules/migration/migration.repository.test.ts @@ -29,6 +29,29 @@ describe('MigrationRepository', () => { expect(select).not.toHaveBeenCalled(); }); + it('deduplicates title lookups and queries in chunks of 500', async () => { + const where = vi + .fn() + .mockResolvedValueOnce([{ bookId: 1, title: 'Dune' }]) + .mockResolvedValueOnce([{ bookId: 501, title: 'Messiah' }]) + .mockResolvedValueOnce([{ bookId: 1001, title: null }]); + const from = vi.fn().mockReturnValue({ where }); + const select = vi.fn().mockReturnValue({ from }); + + const repo = new MigrationRepository({ select } as never); + const ids = Array.from({ length: 1001 }, (_, index) => index + 1); + ids.push(1, Number.NaN); + + await expect(repo.findBookTitlesByIds(ids)).resolves.toEqual( + new Map([ + [1, 'Dune'], + [501, 'Messiah'], + [1001, null], + ]), + ); + expect(where).toHaveBeenCalledTimes(3); + }); + it('purges run state and removes all plan artifacts when there are no completed runs', async () => { const deleteWhere = vi.fn().mockResolvedValue(undefined); const deleteFn = vi.fn().mockReturnValue({ where: deleteWhere }); diff --git a/server/src/modules/migration/migration.repository.ts b/server/src/modules/migration/migration.repository.ts index 6bdc10d77..824e2a1e9 100644 --- a/server/src/modules/migration/migration.repository.ts +++ b/server/src/modules/migration/migration.repository.ts @@ -8,6 +8,15 @@ import * as schema from '../../db/schema'; type Db = NodePgDatabase; type MigrationRunState = 'draft' | 'preflight_failed' | 'dry_run_ready' | 'running' | 'failed' | 'completed'; const ACTIVE_RUN_STATES: MigrationRunState[] = ['running']; +const BATCH_CHUNK_SIZE = 500; + +function chunk(items: T[], size: number): T[][] { + const chunks: T[][] = []; + for (let i = 0; i < items.length; i += size) { + chunks.push(items.slice(i, i + size)); + } + return chunks; +} @Injectable() export class MigrationRepository { @@ -73,14 +82,12 @@ export class MigrationRepository { if (uniquePaths.length === 0) return new Set(); const out = new Set(); - const chunkSize = 500; - for (let index = 0; index < uniquePaths.length; index += chunkSize) { - const chunk = uniquePaths.slice(index, index + chunkSize); + for (const batch of chunk(uniquePaths, BATCH_CHUNK_SIZE)) { const rows = await this.db .select({ absolutePath: schema.bookFiles.absolutePath }) .from(schema.bookFiles) - .where(inArray(schema.bookFiles.absolutePath, chunk)); + .where(inArray(schema.bookFiles.absolutePath, batch)); for (const row of rows) out.add(row.absolutePath); } @@ -183,12 +190,19 @@ export class MigrationRepository { } async findBookTitlesByIds(ids: number[]): Promise> { - if (ids.length === 0) return new Map(); - const rows = await this.db - .select({ bookId: schema.bookMetadata.bookId, title: schema.bookMetadata.title }) - .from(schema.bookMetadata) - .where(inArray(schema.bookMetadata.bookId, ids)); - return new Map(rows.map((row) => [row.bookId, row.title ?? null])); + const targetIds = [...new Set(ids.filter((id) => Number.isFinite(id)))]; + const titles = new Map(); + if (targetIds.length === 0) return titles; + + for (const batch of chunk(targetIds, BATCH_CHUNK_SIZE)) { + const rows = await this.db + .select({ bookId: schema.bookMetadata.bookId, title: schema.bookMetadata.title }) + .from(schema.bookMetadata) + .where(inArray(schema.bookMetadata.bookId, batch)); + for (const row of rows) titles.set(row.bookId, row.title ?? null); + } + + return titles; } async createRunWithLock(