From 3c5e7d9aee8855713dfae518b0931aac68a0dc0b Mon Sep 17 00:00:00 2001
From: ntr <ntr@smokestack.io>
Date: Fri, 3 May 2024 00:24:01 +1000
Subject: [PATCH] alternate between new and old

---
 src/attachments.ts |  2 +-
 src/scraper.ts     | 64 +++++++++++++++++++++++++++++-----------------
 2 files changed, 42 insertions(+), 24 deletions(-)

diff --git a/src/attachments.ts b/src/attachments.ts
index 56f5537..72e5583 100644
--- a/src/attachments.ts
+++ b/src/attachments.ts
@@ -87,7 +87,7 @@ export default async function attachmentService(log: Logger, minio: Minio.Client
 
             // 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(_.omit(a, ['vision']), 'failed to process attachment')
             }
diff --git a/src/scraper.ts b/src/scraper.ts
index 89eb823..624eacb 100644
--- a/src/scraper.ts
+++ b/src/scraper.ts
@@ -1,3 +1,4 @@
+import { setTimeout } from 'timers/promises'
 import axios from 'axios'
 
 import _ from 'lodash';
@@ -104,7 +105,7 @@ export default async function scraper(
                 .flatMap(m => m.attachments.map(a => [a, m]))
                 .filter(([a, m]) => acceptedContentTypes.includes(a.content_type));
 
-            log.info({ validAttachments: validAttachments.length }, 'attachments found in messages')
+            log.debug({ validAttachments: validAttachments.length }, 'attachments found in messages')
 
             return Promise.all(validAttachments
                 .map(async ([a, m]) => {
@@ -138,27 +139,21 @@ export default async function scraper(
             // the api call returns them most recent first
             const messages = (await fetchChannelMessagesAfter(channelId, latestMessageId)).reverse()
             if (messages.length === 0) {
-                log.info('no new messages')
-                return
+                log.info('reached latest message')
             }
 
-            log.info({ messages: messages.length, latestMessageId }, 'fetched new messages')
+            log.debug({ messages: messages.length, latestMessageId }, 'fetched new messages')
             yield* await filterAttachments(messages)
-            const cursor = _.last(messages).id
-            yield* fetchNew(channelId, log, cursor)
         }
 
         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('reached beginning of history')
             }
 
-            log.info({ messages: messages.length, earliestMessageId }, 'fetched older messages')
+            log.debug({ messages: messages.length, earliestMessageId }, 'fetched older messages')
             yield* await filterAttachments(messages)
-            const cursor = _.last(messages).id
-            yield* fetchOld(channelId, log, cursor)
         }
 
         const ensureHistory = async () => {
@@ -173,20 +168,43 @@ export default async function scraper(
 
         const history = 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)
+        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()
         }
 
-        // 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)
-        }
+        await processLoop();
     }
 
     return {