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,
@@ -82,46 +84,49 @@ export default function scraper(
         const guild = await fetchGuildDetails(channel.guild_id)
         const log = baseLog.child({ guild: guild.name, channel: channel.name, channelId })
 
-        async function* filterAttachments(message): AsyncGenerator<Attachment> {
+        async function filterAttachments(messages): Promise<Attachment[]> {
             const acceptedContentTypes = [
                 ...IMAGE_TYPES,
                 ...VIDEO_TYPES,
             ]
 
-            if (!message.attachments) return null
+            const validAttachments = messages
+                .filter(m => m.attachments)
+                // bundle message and attachment together
+                .flatMap(m => m.attachments.map(a => [a, m]))
+                .filter(([a, m]) => acceptedContentTypes.includes(a.content_type))
 
-            const validAttachments = message.attachments
-                .filter(a => acceptedContentTypes.includes(a.content_type))
+            const newAttachments = validAttachments.filter(([a, m]) => ![].includes(a.id))
+            log.info({ newAttachments: newAttachments.length, validAttachments: validAttachments.length })
 
-            for (const a of validAttachments) {
-                const attachment: Attachment = {
-                    id: BigInt(a.id),
-                    msgId: BigInt(message.id),
-                    channelId: BigInt(message.channel_id),
-                    url: a.url,
-                    contentType: a.content_type,
+            return Promise.all(newAttachments
+                .map(async ([a, m]) => {
+                    const attachment: Attachment = {
+                        id: BigInt(a.id),
+                        msgId: BigInt(m.id),
+                        channelId: BigInt(m.channel_id),
+                        url: a.url,
+                        contentType: a.content_type,
 
-                    // create this ahead of time to associate with our copy of the media
-                    memeId: crypto.randomUUID(),
+                        // create this ahead of time to associate with our copy of the media
+                        memeId: crypto.randomUUID(),
 
-                    // @ts-ignore it's fine
-                    vision: {},
-                    guild,
-                    channel,
-                    message,
+                        // @ts-ignore it's fine
+                        vision: {},
+                        guild,
+                        channel,
+                        message: m,
 
-                    createdAt: undefined,
-                    updatedAt: undefined
-                }
+                        createdAt: undefined,
+                        updatedAt: undefined
+                    }
 
-                log.info('fetching vision')
-
-                const visionData = await detectText(attachment)
-                yield { ...attachment, vision: visionData }
-            }
+                    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()
@@ -131,27 +136,21 @@ export default function scraper(
             }
 
             log.info({ messages: messages.length, latestMessageId }, 'fetched new messages')
-            for (const msg of messages) {
-                yield* filterAttachments(msg)
-            }
-
             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')
-            for (const msg of messages) {
-                yield* filterAttachments(msg)
-            }
-
+            log.info({ messages: messages.length, earliestMessageId }, 'fetched old messages')
             const cursor = _.last(messages).id
+            yield [await filterAttachments(messages), cursor]
             yield* fetchOld(channelId, log, cursor)
         }
 
@@ -165,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)
         }
     }