import { CronJob } from 'cron' import type { Subscription } from '../alert.js' import { getCronExpression } from '../alert.js' import { Digest } from '../digest.js' import * as db from '../database.js' const jobsById = new Map() export const runJob = async (sub: Subscription) => { try { if (!sub.confirmed_at || sub.unsubscribed_at) { console.log('worker: job skipped', sub.id, 'not confirmed or unsubscribed') return false } console.log('worker: job starting', sub.id) const start = Date.now() const since = sub.last_digest_at || sub.created_at const events = await db.getEventsForSubscription(sub.id, since) if (events.length === 0) { console.log('worker: job skipped', sub.id, 'ok', 'no data received') return false } const digest = new Digest(sub) await digest.sendFromStoredEvents(events) // Clean up processed events await db.deleteEventsForSubscription(sub.id, since) await db.updateLastDigestAt(sub.id, Math.floor(Date.now() / 1000)) console.log('worker: job completed', sub.id, 'in', Date.now() - start, 'ms') return true } catch (e) { console.log('worker: job failed', sub.id, e) return false } } const createJob = (sub: Subscription) => { const cron = getCronExpression(sub.frequency) const run = async () => { await runJob(sub) } return CronJob.from({ cronTime: cron, onTick: run, start: true, timeZone: 'UTC', }) } export const addJob = (sub: Subscription) => { jobsById.get(sub.id)?.stop() jobsById.set(sub.id, createJob(sub)) } export const removeJob = (sub: Subscription) => { jobsById.get(sub.id)?.stop() jobsById.delete(sub.id) } // Daily purge of events older than 7 days CronJob.from({ cronTime: '0 0 3 * * *', // 3am UTC daily onTick: async () => { const weekAgo = Math.floor(Date.now() / 1000) - 7 * 24 * 3600 await db.purgeEventsOlderThan(weekAgo) console.log('worker: purged events older than 7 days') }, start: true, timeZone: 'UTC', })