mirror of
https://github.com/jakejarvis/sofa.git
synced 2026-08-29 01:35:39 -04:00
fix: misc bug fixes and reliability improvements
- Fix `refreshCreditsJob` to treat `null` `lastFetchedAt` as needing refresh (not just missing cast entry) - Fix `buildBackupCron` 6h expression to correctly fire every 6 hours using explicit hour list instead of `*/6` - Fix `vacuumDatabase` to pass the path directly as a bound parameter (the manual escape was redundant and incorrect) - Clean up orphaned `.tmp` files when image cache writes fail in both `downloadAndCacheImage` and `fetchAndMaybeCache` - Surface import job payload parse/validation errors as job-level `error` status instead of throwing unhandled exceptions; check for cancellation before the processing loop starts - Replace bare `Error` throws in `readImportJob` with `ORPCError` (`NOT_FOUND` / `FORBIDDEN`) so oRPC maps them to correct HTTP status codes - Auto-promote title status to `in_progress` when `logEpisodeWatchBatch` is called for titles that have no status or are only on the watchlist - Refresh seasons from TMDB when a webhook references a season not yet in the DB (handles newly-released seasons) - Upgrade fire-and-forget enrichment failures and webhook processing errors from `log.debug` → `log.warn` so they surface in production logs
This commit is contained in:
@@ -218,7 +218,7 @@ async function refreshCreditsJob() {
|
|||||||
for (const titleId of libraryIds) {
|
for (const titleId of libraryIds) {
|
||||||
const castEntry = getCastEntryForTitle(titleId);
|
const castEntry = getCastEntryForTitle(titleId);
|
||||||
|
|
||||||
const needsRefresh = !castEntry || (castEntry.lastFetchedAt && castEntry.lastFetchedAt < stale);
|
const needsRefresh = !castEntry || !castEntry.lastFetchedAt || castEntry.lastFetchedAt < stale;
|
||||||
|
|
||||||
if (needsRefresh) {
|
if (needsRefresh) {
|
||||||
await refreshCredits(titleId);
|
await refreshCredits(titleId);
|
||||||
@@ -254,7 +254,7 @@ export function buildBackupCron(
|
|||||||
|
|
||||||
switch (frequency) {
|
switch (frequency) {
|
||||||
case "6h":
|
case "6h":
|
||||||
return `${m} */6 * * *`;
|
return `${m} ${h},${(h + 6) % 24},${(h + 12) % 24},${(h + 18) % 24} * * *`;
|
||||||
case "12h":
|
case "12h":
|
||||||
return `${m} ${h},${(h + 12) % 24} * * *`;
|
return `${m} ${h},${(h + 12) % 24} * * *`;
|
||||||
case "7d":
|
case "7d":
|
||||||
|
|||||||
@@ -171,7 +171,7 @@ export const createJob = os.imports.createJob.use(authed).handler(async ({ input
|
|||||||
log.error(`Import job ${job.id} failed:`, err);
|
log.error(`Import job ${job.id} failed:`, err);
|
||||||
});
|
});
|
||||||
|
|
||||||
return readImportJob(job.id);
|
return readImportJob(job.id, context.user.id);
|
||||||
});
|
});
|
||||||
|
|
||||||
export const getJob = os.imports.getJob.use(authed).handler(({ input, context }) => {
|
export const getJob = os.imports.getJob.use(authed).handler(({ input, context }) => {
|
||||||
@@ -187,7 +187,7 @@ export const cancelJob = os.imports.cancelJob.use(authed).handler(({ input, cont
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
updateImportJobProgress(input.id, { status: "cancelled" });
|
updateImportJobProgress(input.id, { status: "cancelled" });
|
||||||
return readImportJob(input.id);
|
return readImportJob(input.id, context.user.id);
|
||||||
});
|
});
|
||||||
|
|
||||||
export const jobEvents = os.imports.jobEvents.use(authed).handler(async function* ({
|
export const jobEvents = os.imports.jobEvents.use(authed).handler(async function* ({
|
||||||
|
|||||||
@@ -13,10 +13,13 @@ import {
|
|||||||
removeTitleStatus,
|
removeTitleStatus,
|
||||||
setTitleStatus,
|
setTitleStatus,
|
||||||
} from "@sofa/core/tracking";
|
} from "@sofa/core/tracking";
|
||||||
|
import { createLogger } from "@sofa/logger";
|
||||||
|
|
||||||
import { os } from "../context";
|
import { os } from "../context";
|
||||||
import { authed } from "../middleware";
|
import { authed } from "../middleware";
|
||||||
|
|
||||||
|
const log = createLogger("titles");
|
||||||
|
|
||||||
export const detail = os.titles.detail.use(authed).handler(async ({ input }) => {
|
export const detail = os.titles.detail.use(authed).handler(async ({ input }) => {
|
||||||
const result = await getOrFetchTitle(input.id);
|
const result = await getOrFetchTitle(input.id);
|
||||||
if (!result)
|
if (!result)
|
||||||
@@ -77,7 +80,9 @@ export const quickAdd = os.titles.quickAdd.use(authed).handler(async ({ input, c
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Trigger full TMDB import if still a shell (fire-and-forget)
|
// Trigger full TMDB import if still a shell (fire-and-forget)
|
||||||
getOrFetchTitleByTmdbId(result.tmdbId, result.type as "movie" | "tv").catch(() => {});
|
getOrFetchTitleByTmdbId(result.tmdbId, result.type as "movie" | "tv").catch((err) => {
|
||||||
|
log.warn(`Failed to import ${result.type} TMDB ${result.tmdbId}:`, err);
|
||||||
|
});
|
||||||
|
|
||||||
return { id: result.id, alreadyAdded: result.alreadyAdded };
|
return { id: result.id, alreadyAdded: result.alreadyAdded };
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -53,7 +53,7 @@ app.post("/:token", async (c) => {
|
|||||||
await processWebhook(connection.id, connection.userId, provider, event);
|
await processWebhook(connection.id, connection.userId, provider, event);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
// Swallow errors — never return non-200 to media servers
|
// Swallow errors — never return non-200 to media servers
|
||||||
log.debug("Webhook processing failed:", err);
|
log.warn("Webhook processing failed:", err);
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.json({ ok: true });
|
return c.json({ ok: true });
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
import { mkdir, rename } from "node:fs/promises";
|
import { mkdir, rename, unlink } from "node:fs/promises";
|
||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
|
|
||||||
import { CACHE_DIR, TMDB_IMAGE_BASE_URL } from "@sofa/config";
|
import { CACHE_DIR, TMDB_IMAGE_BASE_URL } from "@sofa/config";
|
||||||
@@ -101,6 +101,7 @@ export async function downloadAndCacheImage(
|
|||||||
log.debug(`Cached ${category}/${filename} (${buffer.length} bytes)`);
|
log.debug(`Cached ${category}/${filename} (${buffer.length} bytes)`);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
log.warn(`Failed to write cached image ${filename}:`, err);
|
log.warn(`Failed to write cached image ${filename}:`, err);
|
||||||
|
unlink(tmpPath).catch(() => {});
|
||||||
}
|
}
|
||||||
|
|
||||||
return buffer;
|
return buffer;
|
||||||
@@ -132,7 +133,10 @@ export async function fetchAndMaybeCache(
|
|||||||
const tmpPath = `${finalPath}.tmp.${Date.now()}`;
|
const tmpPath = `${finalPath}.tmp.${Date.now()}`;
|
||||||
Bun.write(tmpPath, buffer)
|
Bun.write(tmpPath, buffer)
|
||||||
.then(() => rename(tmpPath, finalPath))
|
.then(() => rename(tmpPath, finalPath))
|
||||||
.catch((err) => log.warn(`Failed to cache ${category}/${filename}:`, err));
|
.catch((err) => {
|
||||||
|
log.warn(`Failed to cache ${category}/${filename}:`, err);
|
||||||
|
unlink(tmpPath).catch(() => {});
|
||||||
|
});
|
||||||
|
|
||||||
return { buffer, contentType };
|
return { buffer, contentType };
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,3 +1,5 @@
|
|||||||
|
import { ORPCError } from "@orpc/server";
|
||||||
|
|
||||||
import { type ImportJob, NormalizedImportSchema } from "@sofa/api/schemas";
|
import { type ImportJob, NormalizedImportSchema } from "@sofa/api/schemas";
|
||||||
import {
|
import {
|
||||||
getImportJob,
|
getImportJob,
|
||||||
@@ -254,11 +256,11 @@ export function readImportJob(jobId: string, userId?: string): ImportJob {
|
|||||||
const row = getImportJob(jobId);
|
const row = getImportJob(jobId);
|
||||||
|
|
||||||
if (!row) {
|
if (!row) {
|
||||||
throw new Error(`Import job ${jobId} not found`);
|
throw new ORPCError("NOT_FOUND", { message: `Import job ${jobId} not found` });
|
||||||
}
|
}
|
||||||
|
|
||||||
if (userId && row.userId !== userId) {
|
if (userId && row.userId !== userId) {
|
||||||
throw new Error("Not authorized");
|
throw new ORPCError("FORBIDDEN", { message: "Not authorized" });
|
||||||
}
|
}
|
||||||
|
|
||||||
return {
|
return {
|
||||||
@@ -285,18 +287,33 @@ export async function processImportJob(jobId: string): Promise<void> {
|
|||||||
const row = getImportJob(jobId);
|
const row = getImportJob(jobId);
|
||||||
|
|
||||||
if (!row) {
|
if (!row) {
|
||||||
throw new Error(`Import job ${jobId} not found`);
|
log.error(`Import job ${jobId} not found`);
|
||||||
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
let rawPayload: unknown;
|
let rawPayload: unknown;
|
||||||
try {
|
try {
|
||||||
rawPayload = JSON.parse(row.payload);
|
rawPayload = JSON.parse(row.payload);
|
||||||
} catch {
|
} catch {
|
||||||
throw new Error(`Import job ${jobId} has invalid JSON payload`);
|
updateImportJobProgress(jobId, {
|
||||||
|
status: "error",
|
||||||
|
finishedAt: new Date(),
|
||||||
|
errors: JSON.stringify([`Import job ${jobId} has invalid JSON payload`]),
|
||||||
|
currentMessage: "Import failed",
|
||||||
|
});
|
||||||
|
log.error(`Import job ${jobId} has invalid JSON payload`);
|
||||||
|
return;
|
||||||
}
|
}
|
||||||
const parsed = NormalizedImportSchema.safeParse(rawPayload);
|
const parsed = NormalizedImportSchema.safeParse(rawPayload);
|
||||||
if (!parsed.success) {
|
if (!parsed.success) {
|
||||||
throw new Error(`Import job ${jobId} has malformed payload: ${parsed.error.message}`);
|
updateImportJobProgress(jobId, {
|
||||||
|
status: "error",
|
||||||
|
finishedAt: new Date(),
|
||||||
|
errors: JSON.stringify([`Malformed payload: ${parsed.error.message}`]),
|
||||||
|
currentMessage: "Import failed",
|
||||||
|
});
|
||||||
|
log.error(`Import job ${jobId} has malformed payload: ${parsed.error.message}`);
|
||||||
|
return;
|
||||||
}
|
}
|
||||||
const data = parsed.data;
|
const data = parsed.data;
|
||||||
const result: ImportResult = {
|
const result: ImportResult = {
|
||||||
@@ -343,6 +360,18 @@ export async function processImportJob(jobId: string): Promise<void> {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Check if the job was cancelled while we were preparing
|
||||||
|
const preStartStatus = getImportJobStatus(jobId);
|
||||||
|
if (preStartStatus?.status === "cancelled") {
|
||||||
|
updateImportJobProgress(jobId, {
|
||||||
|
finishedAt: new Date(),
|
||||||
|
totalItems: total,
|
||||||
|
currentMessage: "Import cancelled",
|
||||||
|
});
|
||||||
|
log.info(`Import job ${jobId} was cancelled before processing started`);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
// Set status to running + totalItems atomically
|
// Set status to running + totalItems atomically
|
||||||
updateImportJobProgress(jobId, {
|
updateImportJobProgress(jobId, {
|
||||||
status: "running",
|
status: "running",
|
||||||
|
|||||||
@@ -118,15 +118,15 @@ function fireAndForgetEnrichment(
|
|||||||
backdropPath: string | null | undefined,
|
backdropPath: string | null | undefined,
|
||||||
type: "movie" | "tv",
|
type: "movie" | "tv",
|
||||||
) {
|
) {
|
||||||
refreshAvailability(titleId).catch((err) => log.debug("Availability enrichment failed:", err));
|
refreshAvailability(titleId).catch((err) => log.warn("Availability enrichment failed:", err));
|
||||||
refreshRecommendations(titleId).catch((err) =>
|
refreshRecommendations(titleId).catch((err) =>
|
||||||
log.debug("Recommendations enrichment failed:", err),
|
log.warn("Recommendations enrichment failed:", err),
|
||||||
);
|
);
|
||||||
syncTitleArt(titleId, posterPath, backdropPath, type).catch((err) =>
|
syncTitleArt(titleId, posterPath, backdropPath, type).catch((err) =>
|
||||||
log.debug("Cache/thumbhash failed:", err),
|
log.warn("Cache/thumbhash failed:", err),
|
||||||
);
|
);
|
||||||
refreshCredits(titleId).catch((err) => log.debug("Credits enrichment failed:", err));
|
refreshCredits(titleId).catch((err) => log.warn("Credits enrichment failed:", err));
|
||||||
refreshTrailer(titleId).catch((err) => log.debug("Trailer enrichment failed:", err));
|
refreshTrailer(titleId).catch((err) => log.warn("Trailer enrichment failed:", err));
|
||||||
}
|
}
|
||||||
|
|
||||||
type ImportResult = ReturnType<typeof _getOrFetchTitleByTmdbId>;
|
type ImportResult = ReturnType<typeof _getOrFetchTitleByTmdbId>;
|
||||||
|
|||||||
@@ -88,6 +88,19 @@ export function logEpisodeWatchBatch(
|
|||||||
if (episodeIds.length === 0) return;
|
if (episodeIds.length === 0) return;
|
||||||
|
|
||||||
batchInsertEpisodeWatchesTransaction(userId, episodeIds, source, watchedAt);
|
batchInsertEpisodeWatchesTransaction(userId, episodeIds, source, watchedAt);
|
||||||
|
|
||||||
|
// Auto-set title status to in_progress for affected titles
|
||||||
|
const titleIds = new Set<string>();
|
||||||
|
for (const episodeId of episodeIds) {
|
||||||
|
const titleId = getEpisodeTitleId(episodeId);
|
||||||
|
if (titleId) titleIds.add(titleId);
|
||||||
|
}
|
||||||
|
for (const titleId of titleIds) {
|
||||||
|
const existing = getTitleStatus(userId, titleId);
|
||||||
|
if (!existing || existing.status === "watchlist") {
|
||||||
|
setTitleStatus(userId, titleId, "in_progress", source);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
export function markAllEpisodesWatched(
|
export function markAllEpisodesWatched(
|
||||||
|
|||||||
@@ -6,9 +6,10 @@ import {
|
|||||||
updateIntegrationLastEvent,
|
updateIntegrationLastEvent,
|
||||||
} from "@sofa/db/queries/webhooks";
|
} from "@sofa/db/queries/webhooks";
|
||||||
import { createLogger } from "@sofa/logger";
|
import { createLogger } from "@sofa/logger";
|
||||||
|
import { getTvDetails } from "@sofa/tmdb/client";
|
||||||
|
|
||||||
import { resolveMovieTmdbId, resolveShowTmdbId } from "./imports/resolve";
|
import { resolveMovieTmdbId, resolveShowTmdbId } from "./imports/resolve";
|
||||||
import { getOrFetchTitleByTmdbId } from "./metadata";
|
import { getOrFetchTitleByTmdbId, refreshTvChildren } from "./metadata";
|
||||||
import { logEpisodeWatch, logMovieWatch } from "./tracking";
|
import { logEpisodeWatch, logMovieWatch } from "./tracking";
|
||||||
|
|
||||||
const log = createLogger("webhooks");
|
const log = createLogger("webhooks");
|
||||||
@@ -265,7 +266,18 @@ export async function processWebhook(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Find the episode in our DB
|
// Find the episode in our DB
|
||||||
const season = findSeasonByTitleAndNumber(title.id, resolved.seasonNumber);
|
let season = findSeasonByTitleAndNumber(title.id, resolved.seasonNumber);
|
||||||
|
|
||||||
|
if (!season) {
|
||||||
|
// Season might be newly released — try refreshing from TMDB
|
||||||
|
try {
|
||||||
|
const show = await getTvDetails(resolved.showTmdbId);
|
||||||
|
await refreshTvChildren(title.id, resolved.showTmdbId, show.number_of_seasons);
|
||||||
|
season = findSeasonByTitleAndNumber(title.id, resolved.seasonNumber);
|
||||||
|
} catch (err) {
|
||||||
|
log.warn(`Failed to refresh seasons for TMDB ${resolved.showTmdbId}:`, err);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if (!season) {
|
if (!season) {
|
||||||
logEvent(connectionId, event, "error", `Season ${resolved.seasonNumber} not found`);
|
logEvent(connectionId, event, "error", `Season ${resolved.seasonNumber} not found`);
|
||||||
|
|||||||
@@ -89,7 +89,7 @@ export function optimizeDatabase() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
export function vacuumDatabase(into: string): void {
|
export function vacuumDatabase(into: string): void {
|
||||||
getClient().run("VACUUM INTO ?", [into.replace(/'/g, "''")]);
|
getClient().run("VACUUM INTO ?", [into]);
|
||||||
}
|
}
|
||||||
|
|
||||||
export function isDatabaseAccessBlocked(): boolean {
|
export function isDatabaseAccessBlocked(): boolean {
|
||||||
|
|||||||
Reference in New Issue
Block a user