combination WIP, add client

archiver/src/scraper.ts

@@ -0,0 +1,213 @@
+import { setTimeout } from 'timers/promises'
+import axios from 'axios'
+
+import _ from 'lodash';
+
+import type { PrismaClient, Attachment } from '@prisma/client';
+import type { ImageAnnotatorClient } from '@google-cloud/vision';
+import type { Logger } from 'pino';
+import type Minio from 'minio'
+
+import sharp from 'sharp';
+
+import type { AttachmentService } from './attachments'
+import { IMAGE_TYPES, VIDEO_TYPES } from './attachments'
+
+// 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(
+    discordAuth,
+    vision: ImageAnnotatorClient,
+    minio: Minio.Client,
+    prisma: PrismaClient,
+    baseLog: Logger,
+    attachments: AttachmentService
+) {
+    const headers = {
+        'Authorization': discordAuth,
+    }
+
+    async function fetchChannelMessages(channelId) {
+        const url = `https://discord.com/api/v9/channels/${channelId}/messages&limit=100`
+        const res = await axios.get(url, { headers })
+        return res.data
+    }
+
+    // returns msgs in desc id order
+    async function fetchChannelMessagesAfter(channelId, lastMessageId) {
+        const url = `https://discord.com/api/v9/channels/${channelId}/messages?after=${lastMessageId}&limit=100`
+        const res = await axios.get(url, { headers })
+        return res.data
+    }
+
+    async function fetchChannelMessagesBefore(channelId, earliestMessageId) {
+        const url = `https://discord.com/api/v9/channels/${channelId}/messages?before=${earliestMessageId}&limit=100`
+        const res = await axios.get(url, { headers })
+        return res.data
+    }
+
+    async function fetchGuildDetails(guildId) {
+        const url = `https://discord.com/api/v9/guilds/${guildId}`
+        const res = await axios.get(url, { headers })
+        return _.omit(res.data, ['emojis', 'stickers', 'roles', 'features'])
+    }
+
+    async function fetchChannel(channelId) {
+        const url = `https://discord.com/api/v9/channels/${channelId}`
+        const res = await axios.get(url, { headers })
+        return res.data
+    }
+
+    async function detectText(attachment: Attachment) {
+        // bail out of anything not an image
+        if (!IMAGE_TYPES.includes(attachment.contentType)) return {}
+
+        const annotateRequest = {
+            image: {
+                source: {
+                    imageUri: attachment.url
+                }
+            },
+            features: [{ type: 'TEXT_DETECTION' }],
+        }
+        const res = await vision.annotateImage(annotateRequest)
+        // comes back as an array
+        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)
+        const log = baseLog.child({ guild: guild.name, channel: channel.name, channelId })
+
+        async function filterAttachments(messages): Promise<Attachment[]> {
+            const acceptedContentTypes = [
+                ...IMAGE_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));
+
+            log.debug({ validAttachments: validAttachments.length }, 'attachments found in messages')
+
+            return Promise.all(validAttachments
+                .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(),
+
+                        vision: {}, // quick hack so that detectText can receive a full attachment
+                        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
+                    }
+
+                    // actually assign vision here
+                    return { ...attachment, vision: await detectText(attachment) }
+                }))
+        }
+
+        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()
+            if (messages.length === 0) {
+                log.info('reached latest message')
+            }
+
+            log.debug({ messages: messages.length, latestMessageId }, 'fetched new messages')
+            yield* await filterAttachments(messages)
+        }
+
+        async function* fetchOld(channelId, log, earliestMessageId): AsyncIterable<Attachment> {
+            const messages = await fetchChannelMessagesBefore(channelId, earliestMessageId)
+            if (messages.length === 0) {
+                log.info('reached beginning of history')
+            }
+
+            log.debug({ messages: messages.length, earliestMessageId }, 'fetched older messages')
+            yield* await filterAttachments(messages)
+        }
+
+        const ensureHistory = async () => {
+            const stored = await prisma.channelHistory.findFirst({ where: { id: channelId }})
+            if (stored) return stored
+
+            const lastMessageId = channel.last_message_id
+            return await prisma.channelHistory.create({ data:
+                { latestMessageId: lastMessageId, earliestMessageId: lastMessageId, id: channelId }
+            })
+        }
+
+        const history = await ensureHistory()
+
+        async function processLoop() {
+            const history = await ensureHistory()
+            const newAttachments = fetchNew(channelId, log, history.latestMessageId)
+            const oldAttachments = fetchOld(channelId, log, history.earliestMessageId)
+
+            let oldCount = 0;
+            let newCount = 0;
+
+            for await (const attachment of newAttachments) {
+                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)
+                // pretty oldschool, the async iterable from the fetch fn does not have a count
+                // so we just use a plain old count to see if we should wait
+                oldCount++
+            }
+
+            for await (const attachment of oldAttachments) {
+                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)
+                newCount++
+            }
+
+            log.info({ oldCount, newCount }, 'processed attachments')
+
+            const shouldWait = newCount === 0 && oldCount === 0
+            const PROCESS_DELAY_MS = 1000 * 60 // 1 minute
+            if (shouldWait) {
+                log.info({ PROCESS_DELAY_MS }, 'waiting for new messages')
+                await setTimeout(PROCESS_DELAY_MS)
+            }
+
+            return processLoop()
+        }
+
+        await processLoop();
+    }
+
+    return {
+        processChannel,
+    }
+}