import { Cron } from "croner"; import { and, eq, inArray, isNotNull, lt, or } from "drizzle-orm"; import { db } from "@/lib/db/client"; import { availabilityOffers, cronRuns, seasons, titleCast, titles, userTitleStatus, } from "@/lib/db/schema"; import { createLogger } from "@/lib/logger"; import { refreshAvailability } from "@/lib/services/availability"; import { createBackup, ensureBackupDir, pruneBackups, } from "@/lib/services/backup"; import { refreshCredits } from "@/lib/services/credits"; import { cacheEpisodeStills, cacheImagesForTitle, cacheProfilePhotos, cacheProviderLogos, imageCacheEnabled, } from "@/lib/services/image-cache"; import { refreshRecommendations, refreshTitle, refreshTvChildren, } from "@/lib/services/metadata"; import { getSetting } from "@/lib/services/settings"; import { performUpdateCheck } from "@/lib/services/update-check"; import { getTvDetails } from "@/lib/tmdb/client"; export type BackupFrequency = "6h" | "12h" | "1d" | "7d"; const log = createLogger("cron"); const DAY = 24 * 60 * 60 * 1000; const RATE_LIMIT_MS = 300; const globalForJobs = globalThis as unknown as { _jobs: Map | undefined; }; if (!globalForJobs._jobs) { globalForJobs._jobs = new Map(); } const jobs = globalForJobs._jobs; function schedule(name: string, cron: string, handler: () => Promise) { jobs.set( name, new Cron(cron, { name, protect: true }, async () => { log.info(`Running job: ${name}`); const startMs = performance.now(); const run = db .insert(cronRuns) .values({ jobName: name, status: "running", startedAt: new Date() }) .returning() .get(); try { await handler(); const durationMs = Math.round(performance.now() - startMs); db.update(cronRuns) .set({ status: "success", finishedAt: new Date(), durationMs }) .where(eq(cronRuns.id, run.id)) .run(); log.info(`Completed job: ${name} (${durationMs}ms)`); } catch (err) { const durationMs = Math.round(performance.now() - startMs); db.update(cronRuns) .set({ status: "error", finishedAt: new Date(), durationMs, errorMessage: err instanceof Error ? err.message : String(err), }) .where(eq(cronRuns.id, run.id)) .run(); log.error(`Job ${name} failed:`, err); } }), ); } /** Get schedule metadata for all registered jobs */ export function getJobSchedules(): { jobName: string; pattern: string; nextRunAt: string | null; }[] { return Array.from(jobs.entries()).map(([name, cron]) => ({ jobName: name, pattern: cron.getPattern() ?? "", nextRunAt: cron.nextRun()?.toISOString() ?? null, })); } /** Manually trigger a job by name. Returns false if job not found. */ export async function triggerJob(name: string): Promise { const job = jobs.get(name); if (!job) return false; await job.trigger(); return true; } function getLibraryTitleIds(): string[] { const rows = db .select({ titleId: userTitleStatus.titleId }) .from(userTitleStatus) .groupBy(userTitleStatus.titleId) .all(); return rows.map((r) => r.titleId); } // Refresh titles where lastFetchedAt is stale async function nightlyRefreshLibrary() { const libraryIds = getLibraryTitleIds(); log.debug(`Checking ${libraryIds.length} library titles for staleness`); const libraryStale = new Date(Date.now() - 7 * DAY); const nonLibraryStale = new Date(Date.now() - 30 * DAY); // Library titles: 7 days const staleLibrary = db .select({ id: titles.id }) .from(titles) .where( and( inArray(titles.id, libraryIds), lt(titles.lastFetchedAt, libraryStale), ), ) .all(); for (const { id } of staleLibrary) { await refreshTitle(id); await Bun.sleep(RATE_LIMIT_MS); } // Non-library titles: 30 days const nonLibrary = db .select() .from(titles) .where( and( isNotNull(titles.lastFetchedAt), lt(titles.lastFetchedAt, nonLibraryStale), ), ) .limit(50) .all(); for (const t of nonLibrary) { if (!libraryIds.includes(t.id)) { await refreshTitle(t.id); await Bun.sleep(RATE_LIMIT_MS); } } } // Refresh availability for library titles where stale async function refreshAvailabilityJob() { const libraryIds = getLibraryTitleIds(); log.debug(`Checking availability for ${libraryIds.length} library titles`); const stale = new Date(Date.now() - DAY); // Batch: find titles with any offers, and titles with stale offers const titlesWithOffers = new Set( db .select({ titleId: availabilityOffers.titleId }) .from(availabilityOffers) .where(inArray(availabilityOffers.titleId, libraryIds)) .groupBy(availabilityOffers.titleId) .all() .map((r) => r.titleId), ); const titlesWithStaleOffers = new Set( db .select({ titleId: availabilityOffers.titleId }) .from(availabilityOffers) .where( and( inArray(availabilityOffers.titleId, libraryIds), lt(availabilityOffers.lastFetchedAt, stale), ), ) .groupBy(availabilityOffers.titleId) .all() .map((r) => r.titleId), ); for (const titleId of libraryIds) { if (titlesWithStaleOffers.has(titleId) || !titlesWithOffers.has(titleId)) { await refreshAvailability(titleId); await Bun.sleep(RATE_LIMIT_MS); } } } // Refresh recommendations for recently active titles async function refreshRecommendationsJob() { const libraryIds = getLibraryTitleIds(); log.debug( `Refreshing recommendations for ${libraryIds.length} library titles`, ); for (const titleId of libraryIds) { await refreshRecommendations(titleId); await Bun.sleep(RATE_LIMIT_MS); } } // Refresh TV episodes for returning shows async function refreshTvChildrenJob() { const returningStatuses = ["Returning Series", "In Production"]; const stale = new Date(Date.now() - 7 * DAY); const tvShows = db .select() .from(titles) .where( and( eq(titles.type, "tv"), isNotNull(titles.lastFetchedAt), or(...returningStatuses.map((s) => eq(titles.status, s))), ), ) .all(); log.debug(`Checking ${tvShows.length} returning TV shows for stale episodes`); // Batch: find shows with at least one stale season const tvIds = tvShows.map((s) => s.id); const titlesWithStaleSeasons = new Set( tvIds.length > 0 ? db .select({ titleId: seasons.titleId }) .from(seasons) .where( and( inArray(seasons.titleId, tvIds), lt(seasons.lastFetchedAt, stale), ), ) .groupBy(seasons.titleId) .all() .map((r) => r.titleId) : [], ); for (const show of tvShows) { if (titlesWithStaleSeasons.has(show.id)) { const details = await getTvDetails(show.tmdbId); await refreshTvChildren(show.id, show.tmdbId, details.number_of_seasons); await Bun.sleep(RATE_LIMIT_MS); } } } // Cache images for all library titles (posters, backdrops, stills, logos) async function cacheImagesJob() { if (!imageCacheEnabled()) return; const libraryIds = getLibraryTitleIds(); log.debug(`Caching images for ${libraryIds.length} library titles`); for (const titleId of libraryIds) { try { await Promise.all([ cacheImagesForTitle(titleId), cacheEpisodeStills(titleId), cacheProviderLogos(titleId), cacheProfilePhotos(titleId), ]); } catch (err) { log.warn(`Failed to cache images for title ${titleId}:`, err); } await Bun.sleep(RATE_LIMIT_MS); } } // Refresh credits for library titles where cast is stale or missing async function refreshCreditsJob() { const libraryIds = getLibraryTitleIds(); log.debug(`Checking credits for ${libraryIds.length} library titles`); const stale = new Date(Date.now() - 30 * DAY); for (const titleId of libraryIds) { const castEntry = db .select() .from(titleCast) .where(eq(titleCast.titleId, titleId)) .limit(1) .get(); const needsRefresh = !castEntry || (castEntry.lastFetchedAt && castEntry.lastFetchedAt < stale); if (needsRefresh) { await refreshCredits(titleId); await Bun.sleep(RATE_LIMIT_MS); } } } async function scheduledBackupJob() { const enabled = getSetting("scheduledBackups"); if (enabled !== "true") { log.debug("Scheduled backups disabled, skipping"); return; } await ensureBackupDir(); await createBackup("sofa-scheduled"); const maxStr = getSetting("maxBackupRetention"); const max = maxStr ? Number.parseInt(maxStr, 10) : 7; await pruneBackups(max); } export function buildBackupCron( frequency: BackupFrequency = "1d", time = "02:00", dayOfWeek = 0, ): string { const [hour, minute] = time.split(":").map(Number); const h = Number.isFinite(hour) ? hour : 2; const m = Number.isFinite(minute) ? minute : 0; const dow = dayOfWeek >= 0 && dayOfWeek <= 6 ? dayOfWeek : 0; switch (frequency) { case "6h": return `${m} */6 * * *`; case "12h": return `${m} ${h},${(h + 12) % 24} * * *`; case "7d": return `${m} ${h} * * ${dow}`; default: return `${m} ${h} * * *`; } } function getBackupCronFromSettings(): string { const frequency = (getSetting("backupScheduleFrequency") ?? "1d") as BackupFrequency; const time = getSetting("backupScheduleTime") ?? "02:00"; const dayOfWeek = Number.parseInt(getSetting("backupScheduleDow") ?? "0", 10); return buildBackupCron(frequency, time, dayOfWeek); } export function rescheduleBackup() { const existing = jobs.get("scheduledBackup"); if (existing) { existing.stop(); jobs.delete("scheduledBackup"); } const cron = getBackupCronFromSettings(); log.info(`Rescheduling backup job with cron: ${cron}`); schedule("scheduledBackup", cron, scheduledBackupJob); } export function startJobs() { if (jobs.size > 0) return; schedule("scheduledBackup", getBackupCronFromSettings(), scheduledBackupJob); schedule("nightlyRefreshLibrary", "0 3 * * *", nightlyRefreshLibrary); schedule("refreshAvailability", "0 */6 * * *", refreshAvailabilityJob); schedule("refreshRecommendations", "0 */12 * * *", refreshRecommendationsJob); schedule("refreshTvChildren", "30 */12 * * *", refreshTvChildrenJob); schedule("cacheImages", "0 1,13 * * *", cacheImagesJob); schedule("refreshCredits", "0 2 * * *", refreshCreditsJob); schedule("updateCheck", "0 */6 * * *", async () => { await performUpdateCheck(); }); log.info(`Started ${jobs.size} jobs`); } export function stopJobs() { for (const job of jobs.values()) { job.stop(); } }