mirror of
https://github.com/jakejarvis/sofa.git
synced 2026-08-29 00:25:38 -04:00
refactor: optimize cron scheduling, DB maintenance, and TMDB reliability
- Skip recommendations refresh for titles fetched within the last 7 days via `getTitlesWithFreshRecommendations`; only stale titles are re-fetched - Replace `Promise.allSettled` in `refreshTvChildren` with a `mapWithConcurrency` helper (concurrency = 5) to stay within TMDB's rate limit - Add 30s request timeout middleware to TMDB client to prevent indefinitely hung fetches - Expand `optimizeDb` weekly job to prune 30-day-old cron runs, integration events, expired sessions, and expired verifications - Add `wal_checkpoint(TRUNCATE)` to `optimizeDatabase` so the WAL is flushed before `PRAGMA optimize` - Add `recoverStaleImportJobs` called on server startup to mark any `pending`/`running` import jobs as errored after a crash/restart - Return `false` from `triggerJob` immediately if the job is already running
This commit is contained in:
+27
-3
@@ -15,6 +15,7 @@ import {
|
||||
getThumbhashBackfillTitleIds,
|
||||
getTitleByIdForCron,
|
||||
getTitleIdsWithStaleSeasons,
|
||||
getTitlesWithFreshRecommendations,
|
||||
startCronRun,
|
||||
} from "@sofa/core/cron";
|
||||
import {
|
||||
@@ -87,10 +88,11 @@ export function getJobSchedules(): {
|
||||
}));
|
||||
}
|
||||
|
||||
/** Manually trigger a job by name. Returns false if job not found. */
|
||||
/** Manually trigger a job by name. Returns false if job not found or already running. */
|
||||
export async function triggerJob(name: string): Promise<boolean> {
|
||||
const job = jobs.get(name);
|
||||
if (!job) return false;
|
||||
if (job.isBusy()) return false;
|
||||
await job.trigger();
|
||||
return true;
|
||||
}
|
||||
@@ -139,9 +141,14 @@ async function refreshAvailabilityJob() {
|
||||
|
||||
async function refreshRecommendationsJob() {
|
||||
const libraryIds = getLibraryTitleIds();
|
||||
log.debug(`Refreshing recommendations for ${libraryIds.length} library titles`);
|
||||
const stale = new Date(Date.now() - 7 * DAY);
|
||||
const fresh = getTitlesWithFreshRecommendations(libraryIds, stale);
|
||||
const staleIds = libraryIds.filter((id) => !fresh.has(id));
|
||||
log.debug(
|
||||
`Refreshing recommendations for ${staleIds.length} of ${libraryIds.length} library titles`,
|
||||
);
|
||||
|
||||
for (const titleId of libraryIds) {
|
||||
for (const titleId of staleIds) {
|
||||
await refreshRecommendations(titleId);
|
||||
await Bun.sleep(RATE_LIMIT_MS);
|
||||
}
|
||||
@@ -300,6 +307,23 @@ export function startJobs() {
|
||||
});
|
||||
schedule("optimizeDb", "0 4 * * 0", async () => {
|
||||
const { optimizeDatabase } = await import("@sofa/db/client");
|
||||
const { deleteOldCronRuns } = await import("@sofa/db/queries/cron");
|
||||
const { deleteOldIntegrationEvents } = await import("@sofa/db/queries/webhooks");
|
||||
const { deleteExpiredSessions, deleteExpiredVerifications } =
|
||||
await import("@sofa/db/queries/auth-cleanup");
|
||||
|
||||
const retentionDate = new Date(Date.now() - 30 * DAY);
|
||||
const cronDeleted = deleteOldCronRuns(retentionDate);
|
||||
const eventsDeleted = deleteOldIntegrationEvents(retentionDate);
|
||||
const sessionsDeleted = deleteExpiredSessions();
|
||||
const verificationsDeleted = deleteExpiredVerifications();
|
||||
|
||||
if (cronDeleted + eventsDeleted + sessionsDeleted + verificationsDeleted > 0) {
|
||||
log.info(
|
||||
`Pruned ${cronDeleted} cron runs, ${eventsDeleted} integration events, ${sessionsDeleted} sessions, ${verificationsDeleted} verifications`,
|
||||
);
|
||||
}
|
||||
|
||||
optimizeDatabase();
|
||||
});
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ import { ensureImageDirs, imageCacheEnabled } from "@sofa/core/image-cache";
|
||||
import { registerJobScheduleProvider } from "@sofa/core/system-health";
|
||||
import { closeDatabase, isDatabaseAccessBlocked } from "@sofa/db/client";
|
||||
import { runMigrations } from "@sofa/db/migrate";
|
||||
import { recoverStaleImportJobs } from "@sofa/db/queries/imports";
|
||||
import { createLogger } from "@sofa/logger";
|
||||
|
||||
import { getJobSchedules, startJobs, stopJobs } from "./cron";
|
||||
@@ -35,6 +36,12 @@ await ensureBackupDir();
|
||||
// Run database migrations
|
||||
runMigrations();
|
||||
|
||||
// Recover import jobs left in running/pending state from a previous crash
|
||||
const recoveredJobs = recoverStaleImportJobs();
|
||||
if (recoveredJobs > 0) {
|
||||
log.info(`Recovered ${recoveredJobs} stale import job(s) from previous shutdown`);
|
||||
}
|
||||
|
||||
// Wire up job schedule provider for system health
|
||||
registerJobScheduleProvider(getJobSchedules);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user