From ad4b6e1f70ad0f72f32247b603b1348ffa070919 Mon Sep 17 00:00:00 2001 From: Nathan Rashleigh Date: Fri, 12 Apr 2024 18:06:14 +1000 Subject: [PATCH] 100 batch failing cause of tx pool i think --- index.ts | 2 +- src/attachments.ts | 15 ++---- src/memes.ts | 6 +-- src/scraper.ts | 120 ++++++++++++++++++++++++++------------------- 4 files changed, 79 insertions(+), 64 deletions(-) diff --git a/index.ts b/index.ts index ab25bf0..620e164 100644 --- a/index.ts +++ b/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 diff --git a/src/attachments.ts b/src/attachments.ts index 49a5e8e..244d052 100644 --- a/src/attachments.ts +++ b/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) { + 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() diff --git a/src/memes.ts b/src/memes.ts index bc3ddd1..486bd35 100644 --- a/src/memes.ts +++ b/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, } } diff --git a/src/scraper.ts b/src/scraper.ts index 4be23bb..3587364 100644 --- a/src/scraper.ts +++ b/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 { + async function filterAttachments(messages): Promise { 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 { + 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 { + 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) } }