add content-type to uploads

docker-compose.yaml

@@ -12,15 +12,6 @@ services:
     volumes:
       - "./data/pg:/var/lib/postgresql/data"
 
-  redis:
-    image: redis:latest
-    ports:
-      - '6379:6379'
-    command: ['redis-server', '/usr/local/etc/redis/redis.conf']
-    volumes:
-      - "./data/redis:/data"
-      - "./redis.conf:/usr/local/etc/redis/redis.conf"
-
   meili:
     image: getmeili/meilisearch:v1.7
     ports:
@@ -41,18 +32,3 @@ services:
     environment:
       MINIO_ROOT_USER: '${MINIO_ROOT_USER}'
       MINIO_ROOT_PASSWORD: '${MINIO_ROOT_PASSWORD}'
-
-  # meilisync:
-  #   platform: linux/x86_64
-  #   image: long2ice/meilisync
-  #   volumes:
-  #     - ./config.yml:/meilisync/config.yml
-
-  # nginx:
-  #   image: nginx:latest
-  #   depends_on:
-  #     - 'api'
-  #   volumes:
-  #     - './nginx/localhost.nginx.conf:/etc/nginx/conf.d/default.conf'
-  #   ports:
-  #     - '80:80'

index.ts

@@ -37,16 +37,18 @@ 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, memes)
+const scraper = await discordScraper(process.env.DISCORD_AUTH, vision, minio, prisma, log, attachments)
 
 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()

package.json

@@ -15,7 +15,6 @@
   "dependencies": {
     "@google-cloud/vision": "^4.2.0",
     "@prisma/client": "^5.12.1",
-    "@theinternetfolks/snowflake": "^1.3.0",
     "axios": "^1.6.8",
     "dotenv": "^16.4.1",
     "lodash": "^4.17.21",

prisma/migrations/20240414022802_add_timestamp_and_postedat/migration.sql

@@ -0,0 +1,25 @@
+/*
+  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,6 +27,7 @@ model Attachment {
 
     createdAt  DateTime @default(now())
     updatedAt  DateTime @default(now())
+    postedAt   DateTime
 
     @@index([channelId])
     @@map("attachments")
@@ -38,6 +39,12 @@ 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())
 

redis.conf

@@ -1 +0,0 @@
-appendfsync everysec

search.html

@@ -1,40 +0,0 @@
-<!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

@@ -2,6 +2,8 @@ import sharp from 'sharp'
 import axios from 'axios'
 import mime from 'mime-types'
 
+import _ from 'lodash';
+
 import type { Logger } from 'pino'
 
 import type Minio from 'minio'
@@ -53,16 +55,21 @@ export default async function attachmentService(log: Logger, minio: Minio.Client
             return download.data
         }
 
-        return minio.putObject(STORAGE_BUCKET, objectName(a), stream())
+        return minio.putObject(STORAGE_BUCKET,
+            objectName(a),
+            stream(),
+            null,
+            // hardcode this for now, we upload everything as webp
+            { 'Content-Type': 'image/webp' })
     }
 
     async function deleteAttachment(a: Attachment) {
         return minio.removeObject(STORAGE_BUCKET, objectName(a))
     }
 
-    async function process(a: Attachment) {
+    async function process(a: Attachment, history: Partial<ChannelHistory>) {
         try {
-            return prisma.$transaction(async prisma => {
+            const results = await 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,24 +78,30 @@ 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')
+                log.error(_.omit(a, ['vision']), 'failed to process attachment')
             }
 
             // deletes the upload if we got that far
             // (probably not)
             await deleteAttachment(a)
-
-            // throw again cause we gather with allSettled
-            throw e
         }
+
+        // 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
     }
 
     await ensureStorageBucketExists()

src/memes.ts

@@ -2,6 +2,8 @@ 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
@@ -29,6 +31,7 @@ export default async function memeService(baseLog: Logger, prisma: PrismaClient,
             text,
             tags: tags,
             attachmentId: a.id,
+            postedAt: a.postedAt,
             createdAt: undefined,
             updatedAt: undefined
         }
@@ -37,36 +40,45 @@ 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);
-        return meili.index('memes')
+        const index = await meili.index('memes')
+        await index.updateSortableAttributes(['postedAt'])
+        return index;
     }
 
-    async function updateMeili() {
+    // BLOWS IT UP
-        await ensureMeiliIndex();
+    async function resetMeili() {
-        const index = meili.index('memes')
+        const delTask = await meili.deleteIndex('memes')
-        const memes = await prisma.meme.findMany();
+        const delResult = await meili.waitForTask(delTask.taskUid)
-        const task = await index.updateDocuments(memes)
+        const index = await ensureMeiliIndex()
-        const res = await meili.waitForTask(task.taskUid)
+        const memes = (await prisma.meme.findMany())
-        log.debug(res)
+            .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 addToIndex(memes: Meme[]) {
+    async function persist(m: Meme) {
         const index = meili.index('memes')
-        return index.updateDocuments(memes)
+        const doc = ({ ...m, postedAt: m.postedAt.valueOf() })
+        return index.updateDocuments([doc])
     }
 
     await ensureMeiliIndex();
-    await updateMeili();
+
+    // BAM
+    // await resetMeili();
 
     return {
         ensureMeiliIndex,
-        updateMeili,
+        resetMeili,
         fromAttachment,
-        addToIndex,
+        persist,
     }
 }
 

src/scraper.test.ts

@@ -1,44 +0,0 @@
-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,19 +11,23 @@ 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
 
-export default function scraper(
+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(
     discordAuth,
     vision: ImageAnnotatorClient,
     minio: Minio.Client,
     prisma: PrismaClient,
     baseLog: Logger,
-    attachments: AttachmentService,
+    attachments: AttachmentService
-    memes: MemeService,
 ) {
     const headers = {
         'Authorization': discordAuth,
@@ -51,9 +55,7 @@ export default function scraper(
     async function fetchGuildDetails(guildId) {
         const url = `https://discord.com/api/v9/guilds/${guildId}`
         const res = await axios.get(url, { headers })
-        const keys_to_remove = ['emojis', 'stickers', 'roles', 'features']
+        return _.omit(res.data, ['emojis', 'stickers', 'roles', 'features'])
-        keys_to_remove.forEach(k => delete res.data[k])
-        return res.data
     }
 
     async function fetchChannel(channelId) {
@@ -79,6 +81,12 @@ export default 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)
@@ -87,19 +95,18 @@ export default function scraper(
         async function filterAttachments(messages): Promise<Attachment[]> {
             const acceptedContentTypes = [
                 ...IMAGE_TYPES,
-                ...VIDEO_TYPES,
+                // ...VIDEO_TYPES,
             ]
 
             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))
+                .filter(([a, m]) => acceptedContentTypes.includes(a.content_type));
 
-            const newAttachments = validAttachments.filter(([a, m]) => ![].includes(a.id))
+            log.info({ validAttachments: validAttachments.length }, 'attachments found in messages')
-            log.info({ newAttachments: newAttachments.length, validAttachments: validAttachments.length })
 
-            return Promise.all(newAttachments
+            return Promise.all(validAttachments
                 .map(async ([a, m]) => {
                     const attachment: Attachment = {
                         id: BigInt(a.id),
@@ -111,22 +118,22 @@ export default function scraper(
                         // create this ahead of time to associate with our copy of the media
                         memeId: crypto.randomUUID(),
 
-                        // @ts-ignore it's fine
+                        vision: {}, // quick hack so that detectText can receive a full attachment
-                        vision: {},
                         guild,
                         channel,
                         message: m,
-
+                        postedAt: convertSnowflakeToDate(a.id),
+                        // ts asks for these to be included but they don't need to be set
                         createdAt: undefined,
                         updatedAt: undefined
                     }
 
-                    const visionData = await detectText(attachment)
+                    // actually assign vision here
-                    return { ...attachment, vision: visionData }
+                    return { ...attachment, vision: await detectText(attachment) }
                 }))
         }
 
-        async function* fetchNew(channelId, log, latestMessageId) {
+        async function* fetchNew(channelId, log, latestMessageId): AsyncIterable<Attachment> {
             // 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()
@@ -136,21 +143,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) {
+        async function* fetchOld(channelId, log, earliestMessageId): AsyncIterable<Attachment> {
             const messages = await fetchChannelMessagesBefore(channelId, earliestMessageId)
             if (messages.length === 0) {
                 log.info('no new messages')
                 return
             }
 
-            log.info({ messages: messages.length, earliestMessageId }, 'fetched old messages')
+            log.info({ messages: messages.length, earliestMessageId }, 'fetched older messages')
+            yield* await filterAttachments(messages)
             const cursor = _.last(messages).id
-            yield [await filterAttachments(messages), cursor]
             yield* fetchOld(channelId, log, cursor)
         }
 
@@ -164,42 +171,21 @@ export default function scraper(
             })
         }
 
-        const startingHistory = await ensureHistory()
+        const history = await ensureHistory()
 
-        const updateHistory = async update => prisma.channelHistory
+        // we can do this better if we yield a list of attachments instead of one at a time
-            .update({ data: update, where: { id: startingHistory.id } })
+        // will have to only update the history if they all succeed, probably better anyway
-
+        for await (const attachment of fetchNew(channelId, log, history.latestMessageId)) {
-        async function processBatch([batch, cursor], genHistory) {
+            log.debug({ msg: 'scraped new attachment', attachment: _.omit(attachment, ['guild', 'message', 'channel', 'vision']) })
-            log.debug({ msg: 'scraped new attachments', attachments: batch.length })
+            const newHistory = { id: history.id, latestMessageId: attachment.msgId }
-            const results = await Promise.allSettled(batch.map(async a => attachments.process(a)))
+            await attachments.process(attachment, newHistory)
-            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
         }
 
-        for await (const batch of fetchNew(channelId, log, startingHistory.latestMessageId)) {
+        // assuming history of earliestId doesn't change...
-            const genHistory = cursor => ({ latestMessageId: cursor })
+        for await (const attachment of fetchOld(channelId, log, history.earliestMessageId)) {
-            await processBatch(batch, genHistory)
+            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 fetchOld(channelId, log, startingHistory.latestMessageId)) {
-            const genHistory = cursor => ({ earliestMessageId: cursor })
-            await processBatch(batch, genHistory)
         }
     }