import { ImapFlow } from 'imapflow' import { simpleParser, type ParsedMail } from 'mailparser' import { db } from '@/utils/database' import redis from '@/utils/redis' import { addContext, createRequest } from '@/utils/archiveRequests' import { addColdContext } from '@/utils/coldRequests' import { extractBareIds, extractUrlIds } from '@/utils/youtubeIds' import type { ArchiveRequest, ColdRequest } from '@/types' const POLL_INTERVAL_MS = 2 * 60000 const INGESTED_FOLDER = 'ingested' // bump the suffix to run the catch-up scan again const BACKFILL_KEY = 'inbox:backfill:cold:v1' let running = false function bodyOf(mail: ParsedMail): string { const body = mail.text || (typeof mail.html === 'string' ? mail.html.replace(/<[^>]+>/g, ' ').replace(/\s+/g, ' ').trim() : '') // postgres text columns reject NUL bytes return body.replaceAll('\u0000', '') } // bounces, out-of-office, newsletters: never turn these into requests or replies function isAutomated(mail: ParsedMail, address: string): boolean { const autoSubmitted = String(mail.headers.get('auto-submitted') || '').toLowerCase() const precedence = String(mail.headers.get('precedence') || '').toLowerCase() if (autoSubmitted && autoSubmitted !== 'no') return true if (['bulk', 'junk', 'list'].includes(precedence)) return true return /^(mailer-daemon|postmaster|no-?reply|donotreply)@/i.test(address) } function referencedIds(mail: ParsedMail): string[] { const refs = Array.isArray(mail.references) ? mail.references : mail.references ? [mail.references] : [] return [...refs, ...(mail.inReplyTo ? [mail.inReplyTo] : [])] } // ingested: became (or already is) a request or answered one, moves to the ingested folder // seen: nothing to do with it, just flag it read // skip: leave it in the inbox for a human type MailResult = 'ingested' | 'seen' | 'skip' interface InboxMail { uid: number fromEmail: string fromName: string | null subject: string } async function processMail(mail: ParsedMail, fallbackId: string, backfill: boolean): Promise { const from = mail.from?.value[0] if (!from?.address) return 'seen' const address = from.address.toLowerCase() const own = [process.env.IMAP_USER, process.env.SMTP_FROM] .map(v => v?.match(/[^<\s]+@[^>\s]+/)?.[0]?.toLowerCase()) .filter(Boolean) if (own.includes(address)) return 'seen' // leave automated mail unread for a human to glance at if (isAutomated(mail, address)) return 'skip' const body = bodyOf(mail) const raw = `${mail.text || ''}\n${typeof mail.html === 'string' ? mail.html : ''}` const videoIds = extractUrlIds(raw) const bareIds = extractBareIds(mail.text || body, new Set(videoIds)) // answer to our "need more context" mail: match by thread headers, else by sender // (some smtp providers rewrite Message-ID, which breaks header matching) const refs = referencedIds(mail) const isReply = Boolean(mail.inReplyTo) || /^(re|aw|sv|fw):/i.test(mail.subject || '') const awaitingArchive = await findAwaiting('archive_requests', address, refs, isReply) if (awaitingArchive) { await addContext(awaitingArchive, body, videoIds) return 'ingested' } const awaitingCold = await findAwaiting('cold_requests', address, refs, isReply) if (awaitingCold) { await addColdContext(awaitingCold, body, videoIds) return 'ingested' } // no youtube id at all = not something we can act on, leave it in the inbox for a human if (videoIds.length === 0 && bareIds.length === 0) return 'skip' const result = await createRequest({ messageId: mail.messageId || fallbackId, fromEmail: from.address, fromName: from.name || null, subject: mail.subject || '(no subject)', body, videoIds, bareIds, backfill }) // not a request (removal, abuse, spam...): leave in the inbox for a human, redis remembers it return result === 'ignored' ? 'skip' : 'ingested' } async function findAwaiting(table: 'archive_requests', address: string, refs: string[], isReply: boolean): Promise async function findAwaiting(table: 'cold_requests', address: string, refs: string[], isReply: boolean): Promise async function findAwaiting(table: 'archive_requests' | 'cold_requests', address: string, refs: string[], isReply: boolean): Promise { const byRefs = refs.length ? await db.selectFrom(table) .selectAll() .where('status', '=', 'awaiting_context') .where('context_message_id', 'in', refs) .executeTakeFirst() : undefined if (byRefs) return byRefs const bySender = await db.selectFrom(table) .selectAll() .where('status', '=', 'awaiting_context') .where('from_email', '=', address) .orderBy('updated_at', 'desc') .execute() // only when unambiguous, otherwise it is a new request from the same person return bySender.length === 1 && isReply ? bySender[0] : undefined } async function pollInbox(backfill = false) { if (running) return running = true const port = Number(process.env.IMAP_PORT || 993) const client = new ImapFlow({ host: process.env.IMAP_HOST!, port, secure: port === 993, auth: { user: process.env.IMAP_USER!, pass: process.env.IMAP_PASS! }, logger: false }) // socket errors are emitted as events, an unhandled one would crash the process client.on('error', (error: Error) => console.log(`[inbox] connection error: ${error.message}`)) try { await client.connect() // fails when it already exists, which is fine await client.mailboxCreate(INGESTED_FOLDER).catch(() => {}) const lock = await client.getMailboxLock('INBOX') try { // the catch-up scan looks at everything: old mail may already be read or was ignored earlier const uids = await client.search(backfill ? { all: true } : { seen: false }, { uid: true }) || [] if (backfill) console.log(`[inbox] catch-up scan over ${uids.length} mails`) let failures = 0 for (const uid of uids) { try { const msg = await client.fetchOne(String(uid), { source: true }, { uid: true }) if (!msg || !msg.source) continue const mail = await simpleParser(msg.source) const result = await processMail(mail, ``, backfill) if (result === 'skip') continue await client.messageFlagsAdd(String(uid), ['\\Seen'], { uid: true }) if (result === 'ingested') { try { await client.messageMove(String(uid), INGESTED_FOLDER, { uid: true }) } catch (error: unknown) { // unread again so the next poll retries the move (the request itself is deduped) console.log(`[inbox] failed to move uid ${uid}: ${(error as Error).message}`) if (!backfill) await client.messageFlagsRemove(String(uid), ['\\Seen'], { uid: true }) } } } catch (error: unknown) { // leave unseen so the next poll retries it console.log(`[inbox] failed to process uid ${uid}: ${(error as Error).message}`) failures++ } } // any failure (e.g. AI down) means the scan reruns on next start; finished mails dedupe if (backfill && failures === 0) { await redis.set(BACKFILL_KEY, '1') console.log('[inbox] catch-up scan done') } else if (backfill) { console.log(`[inbox] catch-up scan had ${failures} failures, will rerun on next start`) } } finally { lock.release() } } catch (error: unknown) { console.log(`[inbox] poll failed: ${(error as Error).message}`) } finally { running = false await client.logout().catch(() => {}) } } async function listInboxMail(): Promise { if (!process.env.IMAP_HOST || !process.env.IMAP_USER || !process.env.IMAP_PASS) return [] const port = Number(process.env.IMAP_PORT || 993) const client = new ImapFlow({ host: process.env.IMAP_HOST, port, secure: port === 993, auth: { user: process.env.IMAP_USER, pass: process.env.IMAP_PASS }, logger: false }) client.on('error', (error: Error) => console.log(`[inbox] connection error: ${error.message}`)) try { await client.connect() const lock = await client.getMailboxLock('INBOX') try { const uids = await client.search({ all: true }, { uid: true }) || [] const mails: InboxMail[] = [] for (const uid of uids.slice(-20).reverse()) { const msg = await client.fetchOne(String(uid), { source: true }, { uid: true }) if (!msg || !msg.source) continue const mail = await simpleParser(msg.source) mails.push({ uid, fromEmail: mail.from?.value[0]?.address || '(unknown sender)', fromName: mail.from?.value[0]?.name || null, subject: mail.subject || '(no subject)' }) } return mails } finally { lock.release() } } finally { await client.logout().catch(() => {}) } } async function ingestInboxMail(uid: number): Promise<'created' | 'duplicate' | 'ignored' | 'missing'> { if (!process.env.IMAP_HOST || !process.env.IMAP_USER || !process.env.IMAP_PASS) return 'missing' const port = Number(process.env.IMAP_PORT || 993) const client = new ImapFlow({ host: process.env.IMAP_HOST, port, secure: port === 993, auth: { user: process.env.IMAP_USER, pass: process.env.IMAP_PASS }, logger: false }) client.on('error', (error: Error) => console.log(`[inbox] connection error: ${error.message}`)) try { await client.connect() await client.mailboxCreate(INGESTED_FOLDER).catch(() => {}) const lock = await client.getMailboxLock('INBOX') try { const msg = await client.fetchOne(String(uid), { source: true }, { uid: true }) if (!msg || !msg.source) return 'missing' const mail = await simpleParser(msg.source) const from = mail.from?.value[0] if (!from?.address) return 'ignored' const body = bodyOf(mail) const raw = `${mail.text || ''}\n${typeof mail.html === 'string' ? mail.html : ''}` const videoIds = extractUrlIds(raw) const result = await createRequest({ messageId: mail.messageId || ``, fromEmail: from.address, fromName: from.name || null, subject: mail.subject || '(no subject)', body, videoIds, bareIds: extractBareIds(mail.text || body, new Set(videoIds)), forceArchive: true }) if (result !== 'ignored') { await client.messageFlagsAdd(String(uid), ['\\Seen'], { uid: true }) await client.messageMove(String(uid), INGESTED_FOLDER, { uid: true }) } return result } finally { lock.release() } } finally { await client.logout().catch(() => {}) } } async function startInboxPoller() { if (!process.env.IMAP_HOST || !process.env.IMAP_USER || !process.env.IMAP_PASS) { console.log('inbox poller disabled (IMAP_HOST / IMAP_USER / IMAP_PASS not set)') return } // an archive job dies with the process, don't leave it stuck as "archiving" forever await db.updateTable('archive_requests') .set({ status: 'failed', error_message: 'Interrupted by server restart, retry.', updated_at: new Date() }) .where('status', 'in', ['archiving']) .execute() await db.updateTable('cold_requests') .set({ status: 'failed', error_message: 'Interrupted by server restart, retry.', updated_at: new Date() }) .where('status', '=', 'approved') .execute() // first run after this feature shipped: catch up old mail once pollInbox(!(await redis.get(BACKFILL_KEY))) setInterval(() => pollInbox(), POLL_INTERVAL_MS).unref() console.log('inbox poller started (this server is primary)') } export { startInboxPoller, pollInbox, listInboxMail, ingestInboxMail }