100 batch failing cause of tx pool i think

index.ts

@@ -37,18 +37,16 @@ const meili = new MeiliSearch({
 
 const memes = await memeService(log, prisma, meili)
 const attachments = await attachmentService(log, minio, prisma, memes)
-const scraper = await 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
-    367936467592413194n,  // tpm screenshots
-    1028511952277864501n,  // tpm sludge
-    725835129326665768n, // machinegirl memes
     1095060717117374687n, // aella memes
 ]
 
 async function main() {
     await Promise.all(CHANNEL_IDS.map(id => scraper.processChannel(id)))
+    await memes.updateMeili();
 }
 
 await main()

prisma/migrations/20240414022802_add_timestamp_and_postedat/migration.sql

@@ -1,25 +0,0 @@
-/*
-  Warnings:
-
-  - Added the required column `postedAt` to the `attachments` table without a default value. This is not possible if the table is not empty.
-  - Added the required column `timestamp` to the `memes` table without a default value. This is not possible if the table is not empty.
-
-*/
--- AlterTable
-ALTER TABLE "attachments"
-  ADD COLUMN     "postedAt"
-  TIMESTAMP(3);
-
-UPDATE "attachments"
-  set "postedAt" = TIMESTAMP 'epoch' + ((id >> 22) + 1420070400000) * INTERVAL '1 millisecond';
-
--- AlterTable
-ALTER TABLE "memes"
-  ADD COLUMN     "postedAt"
-  TIMESTAMP(3);
-
-UPDATE "memes"
-  set "postedAt" = TIMESTAMP 'epoch' + (("attachmentId" >> 22) + 1420070400000) * INTERVAL '1 millisecond';
-
-ALTER TABLE "attachments" ALTER COLUMN "postedAt" SET NOT NULL;
-ALTER TABLE "memes" ALTER COLUMN "postedAt" SET NOT NULL;

prisma/schema.prisma

@@ -27,7 +27,6 @@ model Attachment {
 
     createdAt  DateTime @default(now())
     updatedAt  DateTime @default(now())
-    postedAt   DateTime
 
     @@index([channelId])
     @@map("attachments")
@@ -39,12 +38,6 @@ model Meme {
     text       String
     tags       Json
 
-    // need an timestamp for giving to meili to sort by
-    // i think a usize would actually work but can't
-    // convince the meili-js lib to serialize it properly to json
-    // this one actually has to be .valueOf()'d before we send it
-    // cause js numbers are bigger than postgres int4
-    postedAt   DateTime
     createdAt  DateTime @default(now())
     updatedAt  DateTime @default(now())
 

search.html

@@ -0,0 +1,40 @@
+<!DOCTYPE html>
+<html lang="en">
+  <head>
+    <meta charset="utf-8" />
+    <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/@meilisearch/instant-meilisearch/templates/basic_search.css" />
+  </head>
+  <body>
+      <div id="app" class="wrapper"></div>
+  </body>
+  <script src="https://unpkg.com/react@18/umd/react.development.js" crossorigin></script>
+  <script src="https://unpkg.com/react-dom@18/umd/react-dom.development.js" crossorigin></script>
+  <script src="https://cdn.jsdelivr.net/npm/react-instantsearch-dom@6.7.0/dist/umd/ReactInstantSearchDOM.js"></script>
+  <script src="https://cdn.jsdelivr.net/npm/@meilisearch/instant-meilisearch/dist/instant-meilisearch.umd.min.js"></script>
+  <script>
+    const { InstantSearch, SearchBox, Hits, Highlight, Configure } = ReactInstantSearchDOM;
+    const searchClient = instantMeiliSearch(
+      "http://localhost:7700",
+      '33e714590199031d076954182c55a47b09ae7546e3dbf6f35862eb99cc950e1e'
+    );
+    const App = () => (
+      React.createElement(InstantSearch, {
+        indexName: "memes",
+        searchClient: searchClient
+      }, [
+        React.createElement(SearchBox, { key: 1 }),
+        React.createElement(Hits, { hitComponent: Hit, key: 2 }),
+        React.createElement(Configure, { hitsPerPage: 10 })]
+      )
+    );
+
+    const Hit = (props) => (
+      React.createElement(Highlight, {
+        attribute: "id",
+        hit: props.hit
+      })
+    );
+    ReactDOM.createRoot(document.querySelector('#app')).render(App());
+    console.log(searchClient)
+  </script>
+</html>

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,16 +71,13 @@ 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)
-
-                log.info(dbMeme, 'new meme')
                 return { meme: dbMeme, upload, attachment: dbA }
             }, { timeout: 60000 })
         } catch (e) {
 
             // this happens pretty regularly it seems
             // when ppl reply to an attachment it is included in their message
-            if (e.meta?.modelName !== 'Meme' && e.meta?.target[0] !== 'attachmentId') {
+            if (e.meta.modelName !== 'Meme' && e.meta.target[0] !== 'attachmentId') {
                 log.error(e, 'failed to process attachment')
                 log.error(a, 'failed to process attachment')
             }
@@ -88,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

@@ -2,8 +2,6 @@ import type { PrismaClient, Attachment, Meme } from '@prisma/client';
 import type { Meilisearch } from 'meilisearch';
 import type { Logger } from 'pino';
 
-import _ from 'lodash'
-
 import * as attachments from './attachments'
 
 const STORAGE_BUCKET = process.env.STORAGE_BUCKET
@@ -31,7 +29,6 @@ export default async function memeService(baseLog: Logger, prisma: PrismaClient,
             text,
             tags: tags,
             attachmentId: a.id,
-            postedAt: a.postedAt,
             createdAt: undefined,
             updatedAt: undefined
         }
@@ -40,44 +37,36 @@ export default async function memeService(baseLog: Logger, prisma: PrismaClient,
     }
 
     async function ensureMeiliIndex() {
+        const delTask = await meili.deleteIndex('memes')
+        const delResult = await meili.waitForTask(delTask.taskUid)
         const task = await meili.createIndex('memes', { primaryKey: 'id' })
         const result = await meili.waitForTask(task.taskUid)
         log.debug(result);
-        const index = await meili.index('memes')
-        await index.updateSortableAttributes(['postedAt'])
-        return index;
+        return meili.index('memes')
     }
 
-    // BLOWS IT UP
-    async function resetMeili() {
-        const delTask = await meili.deleteIndex('memes')
-        const delResult = await meili.waitForTask(delTask.taskUid)
-        const index = await ensureMeiliIndex()
-        const memes = (await prisma.meme.findMany())
-            .map(m => ({ ...m, postedAt: m.postedAt.valueOf() }));
-
-        for await (const batch of _.chunk(memes, 1000)) {
-            const task = await index.updateDocuments(batch)
-            const res = await meili.waitForTask(task.taskUid)
-            log.info({ res }, 'added meme batch')
-        }
-    }
-
-    async function persist(m: Meme) {
+    async function updateMeili() {
+        await ensureMeiliIndex();
         const index = meili.index('memes')
-        return index.updateDocuments([m])
+        const memes = await prisma.meme.findMany();
+        const task = await index.updateDocuments(memes)
+        const res = await meili.waitForTask(task.taskUid)
+        log.debug(res)
+    }
+
+    async function addToIndex(memes: Meme[]) {
+        const index = meili.index('memes')
+        return index.updateDocuments(memes)
     }
 
     await ensureMeiliIndex();
-
-    // BAM
-    // await resetMeili();
+    await updateMeili();
 
     return {
         ensureMeiliIndex,
-        resetMeili,
+        updateMeili,
         fromAttachment,
-        persist,
+        addToIndex,
     }
 }
 

src/scraper.test.ts

@@ -0,0 +1,44 @@
+import "dotenv/config"
+
+import fs from 'fs/promises'
+
+import pino from 'pino'
+
+import * as Minio from 'minio'
+import Vision from '@google-cloud/vision'
+import { PrismaClient } from '@prisma/client'
+import { MeiliSearch } from 'meilisearch'
+
+import discordScraper from './scraper'
+import attachmentService from "./attachments"
+
+// @ts-ignore - ts don't know wtf it talking about
+BigInt.prototype.toJSON = function bigIntStringify() {
+    return this.toString();
+};
+
+const VISION_CREDENTIALS = (await import(process.env.GOOGLE_APPLICATION_CREDENTIALS, { assert: { type: "json" } })).default
+const STORAGE_BUCKET = process.env.STORAGE_BUCKET
+
+const log = pino({ base: {} });
+const vision = new Vision.ImageAnnotatorClient({ credentials: VISION_CREDENTIALS })
+const prisma = new PrismaClient()
+const minio = new Minio.Client({
+    endPoint: process.env.STORAGE_URL,
+    accessKey: process.env.STORAGE_ACCESS_KEY,
+    secretKey: process.env.STORAGE_SECRET_KEY,
+})
+const meili = new MeiliSearch({
+  host: process.env.MEILI_URL,
+  apiKey: process.env.MEILI_MASTER_KEY,
+})
+
+const attachments = attachmentService(log, minio, prisma)
+const scraper = discordScraper(process.env.DISCORD_AUTH, vision, minio, prisma, log)
+
+const CHANNEL_IDS = [
+    367936426962190336n,  // TPM shitposts
+    1095060717117374687n, // aella memes
+]
+
+await Promise.all(CHANNEL_IDS.map(id => scraper.processChannel(id)))

src/scraper.ts

@@ -11,23 +11,19 @@ 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
 
-function convertSnowflakeToDate(snowflake) {
-    const DISCORD_EPOCH = 1420070400000
-    const milliseconds = global.BigInt(snowflake) >> 22n
-    return new Date(Number(milliseconds) + DISCORD_EPOCH)
-}
-
-export default async function scraper(
+export default function scraper(
     discordAuth,
     vision: ImageAnnotatorClient,
     minio: Minio.Client,
     prisma: PrismaClient,
     baseLog: Logger,
-    attachments: AttachmentService
+    attachments: AttachmentService,
+    memes: MemeService,
 ) {
     const headers = {
         'Authorization': discordAuth,
@@ -83,12 +79,6 @@ export default async function scraper(
         return res[0] || {}
     }
 
-    async function fetchMissingVision() {
-        const attachments = await prisma.attachment
-            .findMany({ where: { vision: {}, contentType: { in: IMAGE_TYPES } }})
-        baseLog.info({attachments: attachments.length }, 'missing vision attachments')
-    }
-
     async function processChannel(channelId) {
         const channel = await fetchChannel(channelId)
         const guild = await fetchGuildDetails(channel.guild_id)
@@ -97,7 +87,7 @@ export default async function scraper(
         async function filterAttachments(messages): Promise<Attachment[]> {
             const acceptedContentTypes = [
                 ...IMAGE_TYPES,
-                // ...VIDEO_TYPES,
+                ...VIDEO_TYPES,
             ]
 
             const validAttachments = messages
@@ -126,15 +116,17 @@ export default async function scraper(
                         guild,
                         channel,
                         message: m,
-                        postedAt: convertSnowflakeToDate(a.id),
+
                         createdAt: undefined,
                         updatedAt: undefined
                     }
-                    return { ...attachment, vision: await detectText(attachment) }
+
+                    const visionData = await detectText(attachment)
+                    return { ...attachment, vision: visionData }
                 }))
         }
 
-        async function* fetchNew(channelId, log, latestMessageId): AsyncIterable<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()
@@ -144,21 +136,21 @@ export default async 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): AsyncIterable<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)
         }
 
@@ -172,21 +164,42 @@ export default async 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)
         }
     }