100 batch failing cause of tx pool i think

index.ts

@@ -37,7 +37,7 @@ const meili = new MeiliSearch({
 
 const memes = await memeService(log, prisma, meili)
 const attachments = await attachmentService(log, minio, prisma, memes)
-const scraper = discordScraper(process.env.DISCORD_AUTH, vision, minio, prisma, log, attachments)
+const scraper = discordScraper(process.env.DISCORD_AUTH, vision, minio, prisma, log, attachments, memes)
 
 const CHANNEL_IDS = [
     367936426962190336n,  // TPM shitposts

src/attachments.ts

@@ -60,9 +60,9 @@ export default async function attachmentService(log: Logger, minio: Minio.Client
         return minio.removeObject(STORAGE_BUCKET, objectName(a))
     }
 
-    async function process(a: Attachment, history: Partial<ChannelHistory>) {
+    async function process(a: Attachment) {
         try {
-            const results = await prisma.$transaction(async prisma => {
+            return prisma.$transaction(async prisma => {
                 const meme = memes.fromAttachment(a)
                 const dbMeme = await prisma.meme.upsert({ create: meme, update: meme, where: { id: meme.id } })
                 const dbA = await prisma.attachment.upsert({ create: a, update: a, where: { id: a.id } })
@@ -71,8 +71,6 @@ export default async function attachmentService(log: Logger, minio: Minio.Client
                 const upload = await storeAttachment(a)
                 log.debug({ msg: 'successfully uploaded', upload })
 
-                await memes.persist(dbMeme)
-
                 return { meme: dbMeme, upload, attachment: dbA }
             }, { timeout: 60000 })
         } catch (e) {
@@ -87,13 +85,10 @@ export default async function attachmentService(log: Logger, minio: Minio.Client
             // deletes the upload if we got that far
             // (probably not)
             await deleteAttachment(a)
-        }
 
-        // we assume here that a failure in processing was caused by a duplicate
-        // so we update pg to say we've processed it
-        const newHistory = await prisma.channelHistory.update({ data: history, where: { id: history.id }})
-        log.debug({ newHistory }, 'updated channel history')
-        return true
+            // throw again cause we gather with allSettled
+            throw e
+        }
     }
 
     await ensureStorageBucketExists()

src/memes.ts

@@ -54,9 +54,9 @@ export default async function memeService(baseLog: Logger, prisma: PrismaClient,
         log.debug(res)
     }
 
-    async function persist(m: Meme) {
+    async function addToIndex(memes: Meme[]) {
         const index = meili.index('memes')
-        return index.updateDocuments([m])
+        return index.updateDocuments(memes)
     }
 
     await ensureMeiliIndex();
@@ -66,7 +66,7 @@ export default async function memeService(baseLog: Logger, prisma: PrismaClient,
         ensureMeiliIndex,
         updateMeili,
         fromAttachment,
-        persist,
+        addToIndex,
     }
 }
 

src/scraper.ts

@@ -11,6 +11,7 @@ import sharp from 'sharp';
 
 import type { AttachmentService } from './attachments'
 import { IMAGE_TYPES, VIDEO_TYPES } from './attachments'
+import { MemeService } from './memes';
 
 // TODO
 // also fetch embeds and their descriptions
@@ -21,7 +22,8 @@ export default function scraper(
     minio: Minio.Client,
     prisma: PrismaClient,
     baseLog: Logger,
-    attachments: AttachmentService
+    attachments: AttachmentService,
+    memes: MemeService,
 ) {
     const headers = {
         'Authorization': discordAuth,
@@ -110,7 +112,7 @@ export default function scraper(
                         memeId: crypto.randomUUID(),
 
                         // @ts-ignore it's fine
-                        vision: {}, // await detectText(attachment),
+                        vision: {},
                         guild,
                         channel,
                         message: m,
@@ -118,11 +120,13 @@ export default function scraper(
                         createdAt: undefined,
                         updatedAt: undefined
                     }
-                    return attachment
+
+                    const visionData = await detectText(attachment)
+                    return { ...attachment, vision: visionData }
                 }))
         }
 
-        async function* fetchNew(channelId, log, latestMessageId): AsyncGenerator<Attachment> {
+        async function* fetchNew(channelId, log, latestMessageId) {
             // reverse the messages because we want to yield them in chronological order
             // the api call returns them most recent first
             const messages = (await fetchChannelMessagesAfter(channelId, latestMessageId)).reverse()
@@ -132,21 +136,21 @@ export default function scraper(
             }
 
             log.info({ messages: messages.length, latestMessageId }, 'fetched new messages')
-            yield* await filterAttachments(messages)
             const cursor = _.last(messages).id
+            yield [await filterAttachments(messages), cursor]
             yield* fetchNew(channelId, log, cursor)
         }
 
-        async function* fetchOld(channelId, log, earliestMessageId): AsyncGenerator<Attachment, void, Attachment> {
+        async function* fetchOld(channelId, log, earliestMessageId) {
             const messages = await fetchChannelMessagesBefore(channelId, earliestMessageId)
             if (messages.length === 0) {
                 log.info('no new messages')
                 return
             }
 
-            log.info({ messages: messages.length, earliestMessageId }, 'fetched older messages')
-            yield* await filterAttachments(messages)
+            log.info({ messages: messages.length, earliestMessageId }, 'fetched old messages')
             const cursor = _.last(messages).id
+            yield [await filterAttachments(messages), cursor]
             yield* fetchOld(channelId, log, cursor)
         }
 
@@ -160,21 +164,42 @@ export default function scraper(
             })
         }
 
-        const history = await ensureHistory()
+        const startingHistory = await ensureHistory()
 
-        // we can do this better if we yield a list of attachments instead of one at a time
-        // will have to only update the history if they all succeed, probably better anyway
-        for await (const attachment of fetchNew(channelId, log, history.latestMessageId)) {
-            log.debug({ msg: 'scraped new attachment', attachment: _.omit(attachment, ['guild', 'message', 'channel', 'vision']) })
-            const newHistory = { id: history.id, latestMessageId: attachment.msgId }
-            await attachments.process(attachment, newHistory)
+        const updateHistory = async update => prisma.channelHistory
+            .update({ data: update, where: { id: startingHistory.id } })
+
+        async function processBatch([batch, cursor], genHistory) {
+            log.debug({ msg: 'scraped new attachments', attachments: batch.length })
+            const results = await Promise.allSettled(batch.map(async a => attachments.process(a)))
+            const inserts = results
+                .filter(r => r.value)
+                .map(r => r.value)
+
+            const failures = results
+                .filter(r => r.reason)
+                .map(r => r.reason)
+
+            if (failures.length) log.error({ failures }, 'batch failures')
+            log.info({ inserts: inserts.length }, 'batch success')
+
+            await memes.addToIndex(inserts.map(i => i.meme))
+
+            // yeah cool whatever we processed the whole lot... sure
+            const newHistory = await updateHistory(genHistory())
+            log.debug({ newHistory }, 'updated channel history')
+
+            return inserts
         }
 
-        // assuming history of earliestId doesn't change...
-        for await (const attachment of fetchOld(channelId, log, history.earliestMessageId)) {
-            log.debug({ msg: 'scraped old attachment', attachment: _.omit(attachment, ['guild', 'message', 'channel', 'vision']) })
-            const newHistory = { id: history.id, earliestMessageId: attachment.msgId }
-            await attachments.process(attachment, newHistory)
+        for await (const batch of fetchNew(channelId, log, startingHistory.latestMessageId)) {
+            const genHistory = cursor => ({ latestMessageId: cursor })
+            await processBatch(batch, genHistory)
+        }
+
+        for await (const batch of fetchOld(channelId, log, startingHistory.latestMessageId)) {
+            const genHistory = cursor => ({ earliestMessageId: cursor })
+            await processBatch(batch, genHistory)
         }
     }