diff --git a/client/src/components/app/jobs/JobOutputModal.tsx b/client/src/components/app/jobs/JobOutputModal.tsx new file mode 100644 index 0000000..609d606 --- /dev/null +++ b/client/src/components/app/jobs/JobOutputModal.tsx @@ -0,0 +1,65 @@ +import { + Dialog, + DialogContent, + DialogDescription, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog"; +import { useEffect, useRef, type ReactNode } from "react"; + +interface JobOutputModalProps { + open: boolean; + onOpenChange: (open: boolean) => void; + logs: string[]; + isRunning: boolean; + renderLine?: (line: string) => ReactNode; +} + +export default function JobOutputModal({ + open, + onOpenChange, + logs, + isRunning, + renderLine, +}: JobOutputModalProps) { + const outputRef = useRef(null); + const logText = logs.join("\n"); + + useEffect(() => { + if (!open || !isRunning || !outputRef.current) { + return; + } + outputRef.current.scrollTop = outputRef.current.scrollHeight; + }, [open, isRunning, logText]); + + return ( + + + + Job output + + Public output from the running job. + + +
+ {logs.length === 0 ? ( +

+ Waiting for output… +

+ ) : ( +
+ {logs.map((line, index) => ( +

+ {renderLine ? renderLine(line) : line} +

+ ))} +
+ )} +
+ + + ); +} diff --git a/client/src/components/app/jobs/useJobWatcher.ts b/client/src/components/app/jobs/useJobWatcher.ts new file mode 100644 index 0000000..b902511 --- /dev/null +++ b/client/src/components/app/jobs/useJobWatcher.ts @@ -0,0 +1,86 @@ +import { trpc } from "@/utils"; +import { useEffect, useRef, useState } from "react"; + +const TERMINAL_JOB_STATES = new Set(["completed", "failed"]); + +export function isTerminalJobState(state: string | undefined): boolean { + return !!state && TERMINAL_JOB_STATES.has(state); +} + +/** + * Watches a job identified by `watchKey` (queue name and job id) using polling. + * + * @param watchKey Queue and job id, or null/undefined to stop watching. + */ +export function useJobWatcher(watchKey: string | null | undefined) { + const utils = trpc.useContext(); + const [logs, setLogs] = useState([]); + const consumedUpdatedAt = useRef(0); + + useEffect(() => { + setLogs([]); + consumedUpdatedAt.current = 0; + }, [watchKey]); + + const statusQuery = trpc.jobWatching.status.useQuery( + { watchKey: watchKey ?? "" }, + { + enabled: !!watchKey, + refetchInterval: (data) => + isTerminalJobState(data?.state) ? false : 2000, + retry: false, + } + ); + + const isTerminal = isTerminalJobState(statusQuery.data?.state); + + const logsQuery = trpc.jobWatching.logs.useQuery( + { watchKey: watchKey ?? "", startLineIndex: logs.length }, + { + enabled: !!watchKey && !statusQuery.isError, + refetchInterval: isTerminal || statusQuery.isError ? false : 2000, + retry: false, + } + ); + + useEffect(() => { + if (!watchKey || !logsQuery.data) { + return; + } + if (logsQuery.dataUpdatedAt === consumedUpdatedAt.current) { + return; + } + consumedUpdatedAt.current = logsQuery.dataUpdatedAt; + if (logsQuery.data.logs.length === 0) { + return; + } + setLogs((prev) => [...prev, ...logsQuery.data.logs]); + }, [watchKey, logsQuery.data, logsQuery.dataUpdatedAt]); + + const refetchLogs = logsQuery.refetch; + useEffect(() => { + if (!isTerminal || !watchKey) { + return; + } + void refetchLogs(); + }, [isTerminal, watchKey, refetchLogs]); + + return { + watchKey: watchKey ?? null, + status: statusQuery.data ?? null, + logs, + logText: logs.join("\n"), + isTerminal, + startedAt: statusQuery.data?.startedAt ?? null, + isError: statusQuery.isError || logsQuery.isError, + resetLogs: () => { + setLogs([]); + consumedUpdatedAt.current = 0; + if (!watchKey) { + return; + } + void utils.jobWatching.logs.invalidate(); + void utils.jobWatching.status.invalidate({ watchKey }); + }, + }; +} diff --git a/client/src/components/app/recipes/RecipeJobStatus.tsx b/client/src/components/app/recipes/RecipeJobStatus.tsx new file mode 100644 index 0000000..1ca3b54 --- /dev/null +++ b/client/src/components/app/recipes/RecipeJobStatus.tsx @@ -0,0 +1,246 @@ +import JobOutputModal from "@/components/app/jobs/JobOutputModal"; +import { useJobWatcher } from "@/components/app/jobs/useJobWatcher"; +import { Button } from "@/components/ui/button"; +import { + Card, + CardContent, + CardDescription, + CardHeader, +} from "@/components/ui/card"; +import { useToast } from "@/components/ui/use-toast"; +import { TimeElapsedText } from "@/useTimeElapsed"; +import { RecipeDetectionStatus, trpc } from "@/utils"; +import { Check, LoaderIcon, RotateCcw, ScrollText } from "lucide-react"; +import { useState } from "react"; + +const STATUS_PREFIX = ""; +const STATUS_SUFFIX = ""; + +function parseStatusLogMessage(line: string): string | null { + if (!line.startsWith(STATUS_PREFIX)) { + return null; + } + let message = line.slice(STATUS_PREFIX.length); + if (message.endsWith(STATUS_SUFFIX)) { + message = message.slice(0, -STATUS_SUFFIX.length); + } + message = message.trim(); + return message || null; +} + +function renderRecipeJobLogLine(line: string) { + const statusMessage = parseStatusLogMessage(line); + if (statusMessage) { + return ( + <> + Status changed: {statusMessage} + + ); + } + return line; +} + +function latestStatusFromLogs(logs: string[]): string | null { + for (let i = logs.length - 1; i >= 0; i--) { + const statusMessage = parseStatusLogMessage(logs[i]); + if (statusMessage) { + return statusMessage; + } + } + return null; +} + +export default function RecipeJobStatus({ recipeId }: { recipeId: number }) { + const [jobOutputOpen, setJobOutputOpen] = useState(false); + const { toast } = useToast(); + const recipeQuery = trpc.recipes.detail.useQuery( + { id: recipeId }, + { + refetchInterval: (data) => + data?.status === RecipeDetectionStatus.SUCCESS ? false : 2000, + } + ); + const recipeStatus = recipeQuery.data?.status; + const jobStatusQuery = trpc.recipes.configurationJobStatus.useQuery( + { recipeId }, + { + refetchInterval: (data) => { + if (data?.watchKey) { + return false; + } + if ( + recipeStatus === RecipeDetectionStatus.WAITING || + recipeStatus === RecipeDetectionStatus.IN_PROGRESS + ) { + return 2000; + } + return false; + }, + refetchOnWindowFocus: false, + refetchOnReconnect: false, + } + ); + const jobWatcher = useJobWatcher(jobStatusQuery.data?.watchKey ?? null); + const reconfigureRecipe = trpc.recipes.reconfigure.useMutation(); + const reconfigureAgenticRecipe = trpc.recipes.reconfigureAgentic.useMutation(); + + const recipe = recipeQuery.data; + const isAgenticJob = jobStatusQuery.data?.kind === "agentic"; + if (!recipe || (recipe.status === RecipeDetectionStatus.SUCCESS && !isAgenticJob)) { + return null; + } + + const isAgenticComplete = + isAgenticJob && recipe.status === RecipeDetectionStatus.SUCCESS; + const showOutputButton = + !isAgenticComplete || jobWatcher.logs.length > 0; + const lastStatus = latestStatusFromLogs(jobWatcher.logs); + const elapsedTickMs = jobWatcher.isTerminal ? 0 : 1000; + const elapsed = ( + + ); + + async function onReconfigure() { + if (!recipe) { + return; + } + try { + if (isAgenticJob) { + await reconfigureAgenticRecipe.mutateAsync({ id: recipe.id }); + } else { + await reconfigureRecipe.mutateAsync({ id: recipe.id }); + } + jobWatcher.resetLogs(); + await recipeQuery.refetch(); + } catch (err) { + toast({ + title: isAgenticJob + ? "Could not retry agentic configuration" + : "Could not redetect configuration", + description: (err as Error).message, + variant: "destructive", + }); + } + } + + return ( + <> + {isAgenticJob && + recipe.status !== RecipeDetectionStatus.ERROR ? ( +
+ + + Agentic Configuration + + + {isAgenticComplete ? ( +
+ + Agent finished configuring this recipe. +
+ ) : ( +
+ + Agent is configuring this recipe… +
+ )} + {lastStatus ? ( +

+ {lastStatus} + {jobWatcher.startedAt ? <> · {elapsed} : null} +

+ ) : jobWatcher.startedAt && !isAgenticComplete ? ( +

Running for {elapsed}

+ ) : null} + {showOutputButton ? ( + + ) : null} + + +
+ ) : null} + {recipe.status == RecipeDetectionStatus.WAITING && !isAgenticJob ? ( +
+ + + Configuration Pending + + +

+ Configuration detection hasn't started for this recipe. Please + check back later. +

+ + +
+ ) : null} + {recipe.status == RecipeDetectionStatus.ERROR ? ( +
+ + + Configuration Error + + +

+ {isAgenticJob + ? "The agentic configuration workflow failed for this recipe." + : "CTDL xTRA failed to detect a valid configuration for this recipe."} +

+

You can adjust the URL, or try again.

+

Failure reason:

+ 
+                {recipe.detectionFailureReason}
+
+ {jobWatcher.logs.length > 0 ? ( + + ) : null} + + + +
+ ) : null} + + + ); +} diff --git a/client/src/components/app/recipes/edit.tsx b/client/src/components/app/recipes/edit.tsx index 963a852..c21cf14 100644 --- a/client/src/components/app/recipes/edit.tsx +++ b/client/src/components/app/recipes/edit.tsx @@ -45,14 +45,14 @@ import { Check, LoaderIcon, Pickaxe, - RotateCcw, Star, } from "lucide-react"; -import { useEffect, useRef, useState } from "react"; +import { useEffect, useState } from "react"; import { Link } from "wouter"; import { displayRecipeDetails } from "./util"; import TestLinkRegex from "./TestLinkRegex"; import { RecipeConfigurationEditor } from "./RecipeConfigurationEditor"; +import RecipeJobStatus from "./RecipeJobStatus"; import { CatalogueType, PageType, @@ -177,14 +177,13 @@ export default function EditRecipe() { const jobStatusQuery = trpc.recipes.configurationJobStatus.useQuery( { recipeId: parseInt(recipeId || "") }, { - enabled: !!recipeId && recipeQuery.data?.status !== RecipeDetectionStatus.SUCCESS, - refetchInterval: - recipeQuery.data?.status !== RecipeDetectionStatus.SUCCESS ? 2000 : false, + enabled: + !!recipeId && + recipeQuery.data?.status !== RecipeDetectionStatus.SUCCESS, + refetchInterval: false, } ); const updateRecipe = trpc.recipes.update.useMutation(); - const reconfigureRecipe = trpc.recipes.reconfigure.useMutation(); - const reconfigureAgenticRecipe = trpc.recipes.reconfigureAgentic.useMutation(); const setDefaultRecipe = trpc.recipes.setDefault.useMutation(); const destroyRecipe = trpc.recipes.destroy.useMutation(); const { toast } = useToast(); @@ -194,60 +193,31 @@ export default function EditRecipe() { url: "", }, }); - const intervalRef = useRef(null); - useEffect(() => { - const pollQuery = () => { - if (recipeQuery.data?.status == RecipeDetectionStatus.SUCCESS) { - if (intervalRef.current) { - window.clearInterval(intervalRef.current); - intervalRef.current = null; - } - return; - } - recipeQuery.refetch(); - jobStatusQuery.refetch(); - }; + const recipeIdNum = recipeQuery.data?.id; + const recipeStatus = recipeQuery.data?.status; - if (!recipeQuery.data) { + useEffect(() => { + const recipe = recipeQuery.data; + if (!recipe) { return; } - - if (recipeQuery.data.status != RecipeDetectionStatus.SUCCESS) { - if (!intervalRef.current) { - intervalRef.current = window.setInterval(pollQuery, 2000); - } - } else if (intervalRef.current) { - window.clearInterval(intervalRef.current); - intervalRef.current = null; - } - form.reset({ - ...recipeQuery.data, - name: recipeQuery.data.name ?? undefined, - description: recipeQuery.data.description ?? undefined, - configuration: recipeQuery.data.configuration + ...recipe, + name: recipe.name ?? undefined, + description: recipe.description ?? undefined, + configuration: recipe.configuration ? withDefaultPageSetupAtEachLevel( - recipeQuery.data.configuration as RecipeConfiguration + recipe.configuration as RecipeConfiguration ) : undefined, }); - - // Update JSON text when recipe data loads - if (recipeQuery.data.configuration) { - setJsonText(JSON.stringify(recipeQuery.data.configuration, null, 2)); + if (recipe.configuration) { + setJsonText(JSON.stringify(recipe.configuration, null, 2)); } - - return () => { - if (intervalRef.current) { - window.clearInterval(intervalRef.current); - intervalRef.current = null; - } - }; - }, [recipeQuery.data]); + }, [recipeIdNum, recipeStatus, form.reset]); const isAgenticJob = jobStatusQuery.data?.kind === "agentic"; - const agenticProgress = isAgenticJob ? jobStatusQuery.data?.progress : null; if (!catalogueQuery.data || !recipeQuery.data) { return null; @@ -270,26 +240,6 @@ export default function EditRecipe() { recipeQuery.refetch(); } - async function onReconfigure() { - try { - if (isAgenticJob) { - await reconfigureAgenticRecipe.mutateAsync({ id: recipe.id }); - } else { - await reconfigureRecipe.mutateAsync({ id: recipe.id }); - } - await recipeQuery.refetch(); - await jobStatusQuery.refetch(); - } catch (err) { - toast({ - title: isAgenticJob - ? "Could not retry agentic configuration" - : "Could not redetect configuration", - description: (err as Error).message, - variant: "destructive", - }); - } - } - async function onSetDefault(e: React.MouseEvent) { e.preventDefault(); await setDefaultRecipe.mutateAsync({ @@ -526,90 +476,7 @@ export default function EditRecipe() { pageSetup={recipe.configuration?.pageSetup} /> )} - {isAgenticJob && - (recipe.status == RecipeDetectionStatus.WAITING || - recipe.status == RecipeDetectionStatus.IN_PROGRESS) ? ( -
- - - Agentic Configuration - - -
- - Agent is configuring this recipe… -
- {agenticProgress?.message ? ( -

- {agenticProgress.message} - {"elapsedSeconds" in agenticProgress && - agenticProgress.elapsedSeconds != null - ? ` (${agenticProgress.elapsedSeconds}s)` - : null} -

- ) : null} - - -
- ) : null} - {recipe.status == RecipeDetectionStatus.WAITING ? ( -
- - - Configuration Pending - - -
-

- Configuration detection hasn't started for this - recipe. Please check back later. -

-
- - -
- ) : null} - {recipe.status == RecipeDetectionStatus.ERROR ? ( -
- - - Configuration Error - - -
-

- {isAgenticJob - ? "The agentic configuration workflow failed for this recipe." - : "CTDL xTRA failed to detect a valid configuration for this recipe."} -

-

- You can adjust the URL, or try again. -

-

Failure reason:

- 
-                          {recipe.detectionFailureReason}
-
- -
- - -
- ) : null} + {recipe.status != RecipeDetectionStatus.IN_PROGRESS ? (
@@ -848,8 +715,6 @@ export default function EditRecipe() { disabled={ recipeQuery.isLoading || updateRecipe.isLoading || - reconfigureRecipe.isLoading || - reconfigureAgenticRecipe.isLoading || recipe.status == RecipeDetectionStatus.IN_PROGRESS || !form.formState.isDirty } diff --git a/client/src/useTimeElapsed.ts b/client/src/useTimeElapsed.ts new file mode 100644 index 0000000..f6a9099 --- /dev/null +++ b/client/src/useTimeElapsed.ts @@ -0,0 +1,69 @@ +import { useEffect, useState } from "react"; + +function toStartMs( + startTimestamp: string | number | Date | null | undefined +): number | null { + if (startTimestamp == null || startTimestamp === "") { + return null; + } + const startMs = + startTimestamp instanceof Date + ? startTimestamp.getTime() + : new Date(startTimestamp).getTime(); + return Number.isFinite(startMs) ? startMs : null; +} + +export function formatElapsedMs(elapsedMs: number) { + const clamped = Math.max(0, elapsedMs); + const elapsedHours = Math.floor(clamped / (1000 * 60 * 60)); + const elapsedMinutes = Math.floor((clamped % (1000 * 60 * 60)) / (1000 * 60)); + const elapsedSeconds = Math.floor((clamped % (1000 * 60)) / 1000); + + if (elapsedHours > 0) { + return `${elapsedHours}hr${elapsedMinutes}mins`; + } + if (elapsedMinutes > 0) { + return `${elapsedMinutes}mins`; + } + return `${elapsedSeconds}s`; +} + +/** + * Live elapsed time from `startTimestamp`, updated every `tickIntervalMs`. + * Pass a non-positive interval to freeze. Returns null when there is no start. + */ +export function useTimeElapsed( + startTimestamp: string | number | Date | null | undefined, + tickIntervalMs: number +): string | null { + const [nowMs, setNowMs] = useState(() => Date.now()); + const startMs = toStartMs(startTimestamp); + + useEffect(() => { + if (startMs == null || tickIntervalMs <= 0) { + return; + } + setNowMs(Date.now()); + const interval = window.setInterval( + () => setNowMs(Date.now()), + tickIntervalMs + ); + return () => window.clearInterval(interval); + }, [startMs, tickIntervalMs]); + + if (startMs == null) { + return null; + } + return formatElapsedMs(nowMs - startMs); +} + +/** Isolates the 1s clock so only this node re-renders while time elapses. */ +export function TimeElapsedText({ + startTimestamp, + tickIntervalMs, +}: { + startTimestamp: string | number | Date | null | undefined; + tickIntervalMs: number; +}) { + return useTimeElapsed(startTimestamp, tickIntervalMs); +} diff --git a/server/src/appRouter.ts b/server/src/appRouter.ts index 1486208..627b6fc 100644 --- a/server/src/appRouter.ts +++ b/server/src/appRouter.ts @@ -8,6 +8,7 @@ import { cataloguesRouter } from "./routers/catalogues"; import { datasetsRouter } from "./routers/datasets"; import { extractionsRouter } from "./routers/extractions"; import { institutionsRouter } from "./routers/institutions"; +import { jobWatchingRouter } from "./routers/jobWatching"; import { recipesRouter } from "./routers/recipes"; import { settingsRouter } from "./routers/settings"; import { usersRouter } from "./routers/users"; @@ -21,6 +22,7 @@ const appRouter = router({ recipes: recipesRouter, extractions: extractionsRouter, users: usersRouter, + jobWatching: jobWatchingRouter, }); export { appRouter }; diff --git a/server/src/extraction/submitAgenticRecipeDetection.ts b/server/src/extraction/submitAgenticRecipeDetection.ts index c8df5fc..d53acf3 100644 --- a/server/src/extraction/submitAgenticRecipeDetection.ts +++ b/server/src/extraction/submitAgenticRecipeDetection.ts @@ -1,6 +1,7 @@ import { PageType } from "../../../common/types"; import { findCatalogueById } from "../data/catalogues"; import { startRecipe } from "../data/recipes"; +import { toWatchRef } from "../jobWatching"; import getLogger from "../logging"; import { Queues, submitJob } from "../workers"; @@ -23,14 +24,16 @@ export async function submitAgenticRecipeDetection( const result = await startRecipe(catalogueId, url, pageType); logger.info(`Created recipe ${result.id}`); const id = result.id; + const jobId = `agenticRecipeConfig.${id}`; await submitJob( Queues.AgenticRecipeConfig, { recipeId: id, triggeredByUserId: triggeredByUserId ?? null }, - `agenticRecipeConfig.${id}` + jobId ); return { id, pageType, message: null, + ...toWatchRef(Queues.AgenticRecipeConfig, jobId), }; } diff --git a/server/src/extraction/submitRecipeDetection.ts b/server/src/extraction/submitRecipeDetection.ts index b0623c0..8a03139 100644 --- a/server/src/extraction/submitRecipeDetection.ts +++ b/server/src/extraction/submitRecipeDetection.ts @@ -1,6 +1,7 @@ import { CatalogueType, PageType } from "../../../common/types"; import { findCatalogueById } from "../data/catalogues"; import { startRecipe } from "../data/recipes"; +import { toWatchRef } from "../jobWatching"; import getLogger from "../logging"; import { bestOutOf } from "../utils"; import { Queues, submitJob } from "../workers"; @@ -44,14 +45,16 @@ export async function submitRecipeDetection( const result = await startRecipe(catalogueId, url, pageType); logger.info(`Created recipe ${result.id}`); const id = result.id; + const jobId = `detectConfiguration.${id}`; await submitJob( Queues.DetectConfiguration, { recipeId: id, triggeredByUserId: triggeredByUserId ?? null }, - `detectConfiguration.${id}` + jobId ); return { id, pageType, message, + ...toWatchRef(Queues.DetectConfiguration, jobId), }; } diff --git a/server/src/jobWatching/getJobWatchLogs.ts b/server/src/jobWatching/getJobWatchLogs.ts new file mode 100644 index 0000000..ce7c8dd --- /dev/null +++ b/server/src/jobWatching/getJobWatchLogs.ts @@ -0,0 +1,21 @@ +import { resolveWatchKey } from "./resolveWatchKey"; + +export type JobWatchLogs = { + logs: string[]; + logCount: number; +}; + +/** + * Incremental read of Bull's ephemeral `{jobKey}:logs` list. Durable log + * retention is intentionally omitted — use an off-the-shelf logging stack + * rather than persisting transcripts here. + */ + +export async function getJobWatchLogs( + watchKey: string, + startLineIndex = 0 +): Promise { + const { queue, jobId } = await resolveWatchKey(watchKey); + const { logs, count } = await queue.getJobLogs(jobId, startLineIndex, -1); + return { logs, logCount: count }; +} diff --git a/server/src/jobWatching/getJobWatchStatus.ts b/server/src/jobWatching/getJobWatchStatus.ts new file mode 100644 index 0000000..21dac7b --- /dev/null +++ b/server/src/jobWatching/getJobWatchStatus.ts @@ -0,0 +1,38 @@ +import { resolveWatchKey } from "./resolveWatchKey"; + +export type JobWatchStatus = { + watchKey: string; + queueName: string; + jobId: string; + state: string; + progress: unknown; + startedAt: string | null; +}; + +export const TERMINAL_JOB_STATES = new Set(["completed", "failed"]); + +export function isTerminalJobState(state: string): boolean { + return TERMINAL_JOB_STATES.has(state); +} + +/** ISO time the worker started the job; null while still waiting. */ +export function jobStartedAtIso( + processedOn: number | null | undefined +): string | null { + return processedOn ? new Date(processedOn).toISOString() : null; +} + +export async function getJobWatchStatus( + watchKey: string +): Promise { + const { queueName, jobId, job } = await resolveWatchKey(watchKey); + const state = await job.getState(); + return { + watchKey, + queueName, + jobId, + state, + progress: job.progress ?? null, + startedAt: jobStartedAtIso(job.processedOn), + }; +} diff --git a/server/src/jobWatching/index.ts b/server/src/jobWatching/index.ts new file mode 100644 index 0000000..f48b8e9 --- /dev/null +++ b/server/src/jobWatching/index.ts @@ -0,0 +1,15 @@ +export { getJobWatchLogs } from "./getJobWatchLogs"; +export type { JobWatchLogs } from "./getJobWatchLogs"; +export { + getJobWatchStatus, + isTerminalJobState, + jobStartedAtIso, + TERMINAL_JOB_STATES, +} from "./getJobWatchStatus"; +export type { JobWatchStatus } from "./getJobWatchStatus"; +export { mergeJobProgress } from "./mergeJobProgress"; +export { parseWatchKey, toWatchKey, toWatchRef } from "./parseWatchKey"; +export type { ParsedWatchKey } from "./parseWatchKey"; +export { publicLog } from "./publicLog"; +export { resolveWatchKey } from "./resolveWatchKey"; +export type { ResolvedWatchKey } from "./resolveWatchKey"; diff --git a/server/src/jobWatching/mergeJobProgress.ts b/server/src/jobWatching/mergeJobProgress.ts new file mode 100644 index 0000000..34f8910 --- /dev/null +++ b/server/src/jobWatching/mergeJobProgress.ts @@ -0,0 +1,21 @@ +export type ProgressJob = { + progress: unknown; + updateProgress: (value: object) => Promise; +}; + +/** + * Merge a small replaceable snapshot onto `job.progress`. Do not store log + * text here — use `publicLog` for append-only user-visible output. + */ +export async function mergeJobProgress( + job: ProgressJob, + patch: Record +): Promise { + const current = + job.progress && + typeof job.progress === "object" && + !Array.isArray(job.progress) + ? (job.progress as Record) + : {}; + await job.updateProgress({ ...current, ...patch }); +} diff --git a/server/src/jobWatching/parseWatchKey.ts b/server/src/jobWatching/parseWatchKey.ts new file mode 100644 index 0000000..969edb1 --- /dev/null +++ b/server/src/jobWatching/parseWatchKey.ts @@ -0,0 +1,53 @@ +import { AppError, AppErrors } from "../appErrors"; +import { Queues } from "../workers"; + +export type ParsedWatchKey = { + queueName: string; + jobId: string; +}; + +export function toWatchKey(queueName: string, jobId: string): string { + return `${queueName}.${jobId}`; +} + +export function toWatchRef(queue: { name: string }, jobId: string) { + return { + queueName: queue.name, + jobId, + watchKey: toWatchKey(queue.name, jobId), + }; +} + +function queueNames(): string[] { + return Object.values(Queues) + .map((queue) => queue.name) + .sort((a, b) => b.length - a.length); +} + +/** + * Split `{queueName}.{jobId}` using a longest-prefix match against existing + * `Queues` names. `jobId` may itself contain dots. + */ +export function parseWatchKey(watchKey: string): ParsedWatchKey { + const trimmed = watchKey.trim(); + const names = queueNames(); + if ( + names.some( + (queueName) => trimmed === queueName || trimmed === `${queueName}.` + ) + ) { + throw new AppError("Watch key is missing a job id", AppErrors.NOT_FOUND); + } + for (const queueName of names) { + const prefix = `${queueName}.`; + if (!trimmed.startsWith(prefix)) { + continue; + } + const jobId = trimmed.slice(prefix.length); + if (!jobId) { + throw new AppError("Watch key is missing a job id", AppErrors.NOT_FOUND); + } + return { queueName, jobId }; + } + throw new AppError("Unknown job queue in watch key", AppErrors.NOT_FOUND); +} diff --git a/server/src/jobWatching/publicLog.ts b/server/src/jobWatching/publicLog.ts new file mode 100644 index 0000000..053c163 --- /dev/null +++ b/server/src/jobWatching/publicLog.ts @@ -0,0 +1,24 @@ +/** + * ⚠️ WARNING — `publicLog` output is user-visible. Anything passed here is + * stored in Bull job logs and returned to authenticated users via + * `jobWatching.logs`. Never log secrets, tokens, raw HTML with PII, internal + * URLs/credentials, etc. Use private `logger` calls for operator-only detail. + */ + +export type PublicLoggableJob = { + id?: string; + log: (row: string) => Promise | void; +}; + +export type PublicLogLogger = { + info: (obj: object, msg?: string) => void; +}; + +export async function publicLog( + job: PublicLoggableJob, + logger: PublicLogLogger, + message: string +): Promise { + await job.log(message); + logger.info({ jobId: job.id }, message); +} diff --git a/server/src/jobWatching/resolveWatchKey.ts b/server/src/jobWatching/resolveWatchKey.ts new file mode 100644 index 0000000..31df086 --- /dev/null +++ b/server/src/jobWatching/resolveWatchKey.ts @@ -0,0 +1,30 @@ +import { Job, Queue } from "bullmq"; +import { AppError, AppErrors } from "../appErrors"; +import { Queues } from "../workers"; +import { parseWatchKey } from "./parseWatchKey"; + +export type ResolvedWatchKey = { + queue: Queue; + queueName: string; + jobId: string; + job: Job; +}; + +function findQueue(queueName: string): Queue | undefined { + return Object.values(Queues).find((queue) => queue.name === queueName); +} + +export async function resolveWatchKey( + watchKey: string +): Promise { + const { queueName, jobId } = parseWatchKey(watchKey); + const queue = findQueue(queueName); + if (!queue) { + throw new AppError("Unknown job queue in watch key", AppErrors.NOT_FOUND); + } + const job = await queue.getJob(jobId); + if (!job) { + throw new AppError("Job not found", AppErrors.NOT_FOUND); + } + return { queue, queueName, jobId, job }; +} diff --git a/server/src/routers/index.ts b/server/src/routers/index.ts index 9fde229..b1104d4 100644 --- a/server/src/routers/index.ts +++ b/server/src/routers/index.ts @@ -1,5 +1,33 @@ -import { initTRPC } from "@trpc/server"; +import { initTRPC, TRPCError } from "@trpc/server"; +import { AppError, AppErrors } from "../appErrors"; import { Context } from "../trpcContext"; + const t = initTRPC.context().create(); + +function trpcCodeForAppError(error: AppError): TRPCError["code"] { + return error.code === AppErrors.NOT_FOUND ? "NOT_FOUND" : "BAD_REQUEST"; +} + +const mapAppError = t.middleware(async ({ next }) => { + try { + const result = await next(); + if (!result.ok && result.error.cause instanceof AppError) { + throw new TRPCError({ + code: trpcCodeForAppError(result.error.cause), + message: result.error.cause.message, + }); + } + return result; + } catch (err) { + if (err instanceof AppError) { + throw new TRPCError({ + code: trpcCodeForAppError(err), + message: err.message, + }); + } + throw err; + } +}); + export const router = t.router; -export const publicProcedure = t.procedure; +export const publicProcedure = t.procedure.use(mapAppError); diff --git a/server/src/routers/jobWatching.ts b/server/src/routers/jobWatching.ts new file mode 100644 index 0000000..44c9f01 --- /dev/null +++ b/server/src/routers/jobWatching.ts @@ -0,0 +1,28 @@ +import { z } from "zod"; +import { publicProcedure, router } from "."; +import { getJobWatchLogs, getJobWatchStatus } from "../jobWatching"; + +export const jobWatchingRouter = router({ + status: publicProcedure + .input( + z.object({ + watchKey: z.string().min(1), + }) + ) + .query(async (opts) => { + return getJobWatchStatus(opts.input.watchKey); + }), + logs: publicProcedure + .input( + z.object({ + watchKey: z.string().min(1), + startLineIndex: z.number().int().min(0).optional(), + }) + ) + .query(async (opts) => { + return getJobWatchLogs( + opts.input.watchKey, + opts.input.startLineIndex ?? 0 + ); + }), +}); diff --git a/server/src/routers/recipes.ts b/server/src/routers/recipes.ts index e6c7ee4..d3cf681 100644 --- a/server/src/routers/recipes.ts +++ b/server/src/routers/recipes.ts @@ -26,6 +26,7 @@ import { } from "../extraction/robotsParser"; import { submitRecipeDetection } from "../extraction/submitRecipeDetection"; import { submitAgenticRecipeDetection } from "../extraction/submitAgenticRecipeDetection"; +import { toWatchRef } from "../jobWatching"; import getLogger from "../logging"; import { SimplifiedMarkdown } from "../types"; import { bestOutOf, exponentialRetry, normalizeUrl } from "../utils"; @@ -125,6 +126,9 @@ export const recipesRouter = router({ ): Promise<{ id: number; context?: { pageType: PageType; message?: string }; + queueName?: string; + jobId?: string; + watchKey?: string; }> => { try { const robotsTxt = await fetchAndParseRobotsTxt(opts.input.url); @@ -202,6 +206,9 @@ export const recipesRouter = router({ pageType: detection.pageType, message: detection.message ?? undefined, }, + queueName: detection.queueName, + jobId: detection.jobId, + watchKey: detection.watchKey, }; } } catch (error) { @@ -227,15 +234,16 @@ export const recipesRouter = router({ if (!recipe) { throw new AppError("Recipe not found", AppErrors.NOT_FOUND); } + const jobId = `detectConfiguration.${recipe.id}`; await submitJob( Queues.DetectConfiguration, { recipeId: opts.input.id, triggeredByUserId: opts.ctx.user?.id ?? null, }, - `detectConfiguration.${recipe.id}` + jobId ); - return; + return toWatchRef(Queues.DetectConfiguration, jobId); }), configurationJobStatus: publicProcedure .input( @@ -253,10 +261,12 @@ export const recipesRouter = router({ const progress = agenticJob.progress as | import("../workers").AgenticRecipeConfigProgress | undefined; + const watch = toWatchRef(Queues.AgenticRecipeConfig, agenticJobId); return { kind: "agentic" as const, state, progress: progress ?? null, + ...watch, }; } @@ -266,10 +276,12 @@ export const recipesRouter = router({ const progress = detectJob.progress as | import("../workers").DetectConfigurationProgress | undefined; + const watch = toWatchRef(Queues.DetectConfiguration, detectJobId); return { kind: "detect" as const, state, progress: progress ?? null, + ...watch, }; } @@ -292,7 +304,7 @@ export const recipesRouter = router({ jobId ); if (existing === "active") { - return; + return toWatchRef(Queues.AgenticRecipeConfig, jobId); } await updateRecipe(recipe.id, { status: RecipeDetectionStatus.WAITING, @@ -306,7 +318,7 @@ export const recipesRouter = router({ }, jobId ); - return; + return toWatchRef(Queues.AgenticRecipeConfig, jobId); }), detectPagination: publicProcedure .input( diff --git a/server/src/server.ts b/server/src/server.ts index ac453af..b5e31ba 100644 --- a/server/src/server.ts +++ b/server/src/server.ts @@ -85,7 +85,13 @@ server.register(async (instance) => { router: appRouter, createContext, onError(opts) { - logger.error( + const isClientError = + opts.error.code === "NOT_FOUND" || + opts.error.code === "BAD_REQUEST" || + opts.error.code === "UNAUTHORIZED" || + opts.error.code === "FORBIDDEN"; + const log = isClientError ? logger.warn : logger.error; + log( { path: opts.path, type: opts.error.name, @@ -95,7 +101,7 @@ server.register(async (instance) => { "Error in tRPC request" ); - if (!airbrake) { + if (!airbrake || isClientError) { return; } diff --git a/server/src/workers/agenticRecipeConfig.ts b/server/src/workers/agenticRecipeConfig.ts index 9ba1ff1..67c866a 100644 --- a/server/src/workers/agenticRecipeConfig.ts +++ b/server/src/workers/agenticRecipeConfig.ts @@ -5,6 +5,7 @@ import { } from "."; import { RecipeDetectionStatus } from "../../../common/types"; import { findRecipeById, updateRecipe } from "../data/recipes"; +import { mergeJobProgress, publicLog } from "../jobWatching"; import getLogger from "../logging"; const logger = getLogger("workers.agenticRecipeConfig"); @@ -24,20 +25,29 @@ export default createProcessor< } logger.info(`${logPrefix} Starting job ${job.id}`); + await publicLog(job, logger, "Starting agentic configuration"); await updateRecipe(recipe.id, { status: RecipeDetectionStatus.IN_PROGRESS, }); + await mergeJobProgress(job, { + message: "Starting agentic configuration", + status: "info", + }); for (let step = 1; step <= 6; step++) { await sleep(10_000); const elapsedSeconds = step * 10; - await job.updateProgress({ - message: `Bogus agent working… (${elapsedSeconds}s)`, + await publicLog( + job, + logger, + `Bogus agent working… (${elapsedSeconds}s)` + ); + await mergeJobProgress(job, { + message: `Step ${step}/6`, status: "info", elapsedSeconds, step, }); - logger.info(`${logPrefix} Job ${job.id} progress step ${step}/6`); } const succeeded = Math.random() < 0.5; @@ -52,7 +62,8 @@ export default createProcessor< status: RecipeDetectionStatus.SUCCESS, detectionFailureReason: null, }); - await job.updateProgress({ + await publicLog(job, logger, "Bogus agent completed"); + await mergeJobProgress(job, { status: "success", message: "Bogus agent completed", }); @@ -65,7 +76,8 @@ export default createProcessor< status: RecipeDetectionStatus.ERROR, detectionFailureReason: failureMessage, }); - await job.updateProgress({ + await publicLog(job, logger, failureMessage); + await mergeJobProgress(job, { status: "failure", message: failureMessage, }); diff --git a/server/src/workers/index.ts b/server/src/workers/index.ts index 20ba6de..d31e1b0 100644 --- a/server/src/workers/index.ts +++ b/server/src/workers/index.ts @@ -226,9 +226,16 @@ const defaultJobOptions: DefaultJobOptions = { const agenticRecipeConfigJobOptions: DefaultJobOptions = { attempts: 1, - removeOnComplete: true, + // BullMQ `age` is seconds. keepLogs caps the Redis list while the job exists; + // removeOnComplete.age keeps a finished job long enough for the UI to observe + // `completed` and drain logs. Durable log retention is intentionally omitted — + // use an off-the-shelf logging stack rather than persisting Bull transcripts here. + keepLogs: 500, + removeOnComplete: { + age: 60 * 60, // 1 hour + }, removeOnFail: { - age: 1000 * 60 * 60 * 24 * 5, // 5 days + age: 60 * 60 * 24 * 5, // 5 days }, }; diff --git a/server/tests/jobWatching/getJobWatchLogs.test.ts b/server/tests/jobWatching/getJobWatchLogs.test.ts new file mode 100644 index 0000000..7b22ca8 --- /dev/null +++ b/server/tests/jobWatching/getJobWatchLogs.test.ts @@ -0,0 +1,56 @@ +import { beforeEach, describe, expect, test, vi } from "vitest"; +import { getJobWatchLogs } from "../../src/jobWatching/getJobWatchLogs"; +import { resolveWatchKey } from "../../src/jobWatching/resolveWatchKey"; + +vi.mock("../../src/jobWatching/resolveWatchKey", () => ({ + resolveWatchKey: vi.fn(), +})); + +const mockedResolveWatchKey = vi.mocked(resolveWatchKey); +const getJobLogs = vi.fn(); +const ALL_LOGS = ["line0", "line1", "line2"]; + +describe("getJobWatchLogs", () => { + beforeEach(() => { + getJobLogs.mockReset(); + mockedResolveWatchKey.mockReset(); + mockedResolveWatchKey.mockResolvedValue({ + queue: { getJobLogs }, + queueName: "recipes.agenticRecipeConfig", + jobId: "agenticRecipeConfig.42", + job: {}, + } as unknown as Awaited>); + getJobLogs.mockImplementation(async (_jobId: string, start: number) => { + return { + logs: start >= ALL_LOGS.length ? [] : ALL_LOGS.slice(start), + count: ALL_LOGS.length, + }; + }); + }); + + test("returns all lines from the start", async () => { + await expect(getJobWatchLogs("watch-key", 0)).resolves.toEqual({ + logs: ["line0", "line1", "line2"], + logCount: 3, + }); + expect(getJobLogs).toHaveBeenCalledWith( + "agenticRecipeConfig.42", + 0, + -1 + ); + }); + + test("returns new lines from a cursor", async () => { + await expect(getJobWatchLogs("watch-key", 2)).resolves.toEqual({ + logs: ["line2"], + logCount: 3, + }); + }); + + test("returns an empty slice when the cursor is at the end", async () => { + await expect(getJobWatchLogs("watch-key", 3)).resolves.toEqual({ + logs: [], + logCount: 3, + }); + }); +}); diff --git a/server/tests/jobWatching/getJobWatchStatus.test.ts b/server/tests/jobWatching/getJobWatchStatus.test.ts new file mode 100644 index 0000000..74ef86a --- /dev/null +++ b/server/tests/jobWatching/getJobWatchStatus.test.ts @@ -0,0 +1,14 @@ +import { describe, expect, test } from "vitest"; +import { jobStartedAtIso } from "../../src/jobWatching"; + +describe("jobStartedAtIso", () => { + test("returns null while the job has not been processed", () => { + expect(jobStartedAtIso(undefined)).toBeNull(); + expect(jobStartedAtIso(null)).toBeNull(); + }); + + test("returns the process-start time, not enqueue time", () => { + const processedOn = Date.UTC(2026, 7, 27, 12, 0, 0); + expect(jobStartedAtIso(processedOn)).toBe("2026-08-27T12:00:00.000Z"); + }); +}); diff --git a/server/tests/jobWatching/mergeJobProgress.test.ts b/server/tests/jobWatching/mergeJobProgress.test.ts new file mode 100644 index 0000000..a225677 --- /dev/null +++ b/server/tests/jobWatching/mergeJobProgress.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, test } from "vitest"; +import { mergeJobProgress } from "../../src/jobWatching"; + +describe("mergeJobProgress", () => { + test("merges a patch onto an object snapshot", async () => { + const job = { + progress: { message: "old", status: "info" } as Record, + updateProgress: async (value: object) => { + job.progress = value as Record; + }, + }; + + await mergeJobProgress(job, { message: "new", step: 2 }); + + expect(job.progress).toEqual({ + message: "new", + status: "info", + step: 2, + }); + }); + + test("replaces a numeric progress value", async () => { + const job = { + progress: 50 as unknown, + updateProgress: async (value: object) => { + job.progress = value; + }, + }; + + await mergeJobProgress(job, { message: "started" }); + + expect(job.progress).toEqual({ message: "started" }); + }); +}); diff --git a/server/tests/jobWatching/parseWatchKey.test.ts b/server/tests/jobWatching/parseWatchKey.test.ts new file mode 100644 index 0000000..7c1ed74 --- /dev/null +++ b/server/tests/jobWatching/parseWatchKey.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, test } from "vitest"; +import { AppError, AppErrors } from "../../src/appErrors"; +import { parseWatchKey, toWatchKey } from "../../src/jobWatching"; +import { Queues } from "../../src/workers"; + +describe("parseWatchKey", () => { + test("splits an agentic recipe watch key whose jobId contains dots", () => { + const jobId = "agenticRecipeConfig.42"; + const watchKey = toWatchKey(Queues.AgenticRecipeConfig.name, jobId); + expect(watchKey).toBe( + "recipes.agenticRecipeConfig.agenticRecipeConfig.42" + ); + expect(parseWatchKey(watchKey)).toEqual({ + queueName: Queues.AgenticRecipeConfig.name, + jobId, + }); + }); + + test("splits a detect-configuration watch key", () => { + const jobId = "detectConfiguration.7"; + expect( + parseWatchKey(toWatchKey(Queues.DetectConfiguration.name, jobId)) + ).toEqual({ + queueName: Queues.DetectConfiguration.name, + jobId, + }); + }); + + test("does not match a shorter queue name prefix of another queue", () => { + const jobId = "page-1"; + const watchKey = toWatchKey(Queues.ExtractDataWithAPI.name, jobId); + expect(parseWatchKey(watchKey)).toEqual({ + queueName: Queues.ExtractDataWithAPI.name, + jobId, + }); + }); + + test("uses the longest matching queue name", () => { + const names = Object.values(Queues).map((queue) => queue.name); + const longest = [...names].sort((a, b) => b.length - a.length)[0]; + const jobId = "job.with.dots"; + expect(parseWatchKey(`${longest}.${jobId}`)).toEqual({ + queueName: longest, + jobId, + }); + }); + + test("rejects an unknown queue name", () => { + try { + parseWatchKey("not.a.queue.job-1"); + throw new Error("expected parseWatchKey to throw"); + } catch (error) { + expect(error).toBeInstanceOf(AppError); + expect((error as AppError).code).toBe(AppErrors.NOT_FOUND); + } + }); + + test("rejects a queue name with an empty job id", () => { + try { + parseWatchKey(`${Queues.AgenticRecipeConfig.name}.`); + throw new Error("expected parseWatchKey to throw"); + } catch (error) { + expect(error).toBeInstanceOf(AppError); + expect((error as AppError).code).toBe(AppErrors.NOT_FOUND); + } + }); + + test("rejects a longer queue name that would otherwise match a shorter prefix", () => { + try { + parseWatchKey(Queues.ExtractDataWithAPI.name); + throw new Error("expected parseWatchKey to throw"); + } catch (error) { + expect(error).toBeInstanceOf(AppError); + expect((error as AppError).code).toBe(AppErrors.NOT_FOUND); + } + }); +});