Skip to content

Commit 812ee18

Browse files
pablopunkpablopunk
committed
feat: add per-job queue log streaming
1 parent 00021d3 commit 812ee18

10 files changed

Lines changed: 632 additions & 63 deletions

File tree

‎src/components/queue/JobDetailLive.tsx‎

Lines changed: 161 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { useEffect, useState } from "react";
1+
import { useEffect, useRef, useState } from "react";
22
import { toast } from "sonner";
33
import type { QueueJob } from "@/server/db/schema";
44
import { ConfirmQueueActionDialog } from "./ConfirmQueueActionDialog";
@@ -14,13 +14,47 @@ interface JobDetailLiveProps {
1414
initialJob: QueueJob;
1515
}
1616

17+
interface JobLogChunkEvent {
18+
jobId: string;
19+
offset: number;
20+
nextOffset: number;
21+
text: string;
22+
truncated: boolean;
23+
}
24+
25+
interface ParsedJobLogLine {
26+
timestamp: string;
27+
level: string;
28+
message: string;
29+
}
30+
31+
function parseJobLogChunk(text: string): ParsedJobLogLine[] {
32+
return text
33+
.split("\n")
34+
.filter((line) => line.trim().length > 0)
35+
.map((line) => {
36+
const [timestamp = "", level = "info", ...messageParts] =
37+
line.split("\t");
38+
const message = messageParts.join("\t").trim();
39+
return {
40+
timestamp,
41+
level,
42+
message: message || line,
43+
};
44+
});
45+
}
46+
1747
export function JobDetailLive({ initialJob }: JobDetailLiveProps) {
1848
const [job, setJob] = useState<QueueJob>(initialJob);
1949
const [dialogOpen, setDialogOpen] = useState(false);
2050
const [pendingAction, setPendingAction] = useState<
2151
"cancel" | "forceUnlock" | null
2252
>(null);
2353
const [isLoading, setIsLoading] = useState(false);
54+
const [jobLogs, setJobLogs] = useState<ParsedJobLogLine[]>([]);
55+
const [logsTruncated, setLogsTruncated] = useState(false);
56+
const [logsConnected, setLogsConnected] = useState(false);
57+
const logsContainerRef = useRef<HTMLDivElement>(null);
2458

2559
useEffect(() => {
2660
const eventSource = new EventSource(
@@ -46,6 +80,81 @@ export function JobDetailLive({ initialJob }: JobDetailLiveProps) {
4680
};
4781
}, [initialJob.id]);
4882

83+
useEffect(() => {
84+
let eventSource: EventSource | null = null;
85+
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
86+
let isUnmounted = false;
87+
let nextOffset = 0;
88+
89+
setJobLogs([]);
90+
setLogsTruncated(false);
91+
92+
const connect = () => {
93+
if (isUnmounted) return;
94+
95+
const params = new URLSearchParams({ jobId: initialJob.id });
96+
if (nextOffset > 0) {
97+
params.set("offset", String(nextOffset));
98+
}
99+
100+
eventSource = new EventSource(
101+
`/api/queue/job-logs-stream?${params.toString()}`,
102+
);
103+
eventSource.addEventListener("open", () => {
104+
setLogsConnected(true);
105+
});
106+
107+
eventSource.addEventListener("log.chunk", (event) => {
108+
try {
109+
const data = JSON.parse(event.data) as JobLogChunkEvent;
110+
nextOffset = data.nextOffset;
111+
112+
if (data.truncated) {
113+
setLogsTruncated(true);
114+
}
115+
116+
if (!data.text) {
117+
return;
118+
}
119+
120+
const newLines = parseJobLogChunk(data.text);
121+
setJobLogs((prev) => {
122+
const merged = [...prev, ...newLines];
123+
const MAX_LOG_LINES = 1_000;
124+
return merged.length > MAX_LOG_LINES
125+
? merged.slice(-MAX_LOG_LINES)
126+
: merged;
127+
});
128+
requestAnimationFrame(() => {
129+
if (!logsContainerRef.current) return;
130+
logsContainerRef.current.scrollTop =
131+
logsContainerRef.current.scrollHeight;
132+
});
133+
} catch {
134+
toast.error("Failed to parse job log stream data");
135+
}
136+
});
137+
138+
eventSource.addEventListener("error", () => {
139+
setLogsConnected(false);
140+
eventSource?.close();
141+
eventSource = null;
142+
if (!isUnmounted) {
143+
reconnectTimer = setTimeout(connect, 1_000);
144+
}
145+
});
146+
};
147+
148+
connect();
149+
150+
return () => {
151+
isUnmounted = true;
152+
setLogsConnected(false);
153+
if (reconnectTimer) clearTimeout(reconnectTimer);
154+
eventSource?.close();
155+
};
156+
}, [initialJob.id]);
157+
49158
const canCancel = job.state === "queued" || job.state === "running";
50159
const canRunNow = job.state === "queued";
51160
const canForceUnlock = job.state === "running";
@@ -138,6 +247,7 @@ export function JobDetailLive({ initialJob }: JobDetailLiveProps) {
138247
</a>
139248
{canCancel && (
140249
<button
250+
type="button"
141251
onClick={() => handleAction("cancel")}
142252
className="rounded-md border px-3 py-2 text-sm hover:bg-muted"
143253
>
@@ -146,6 +256,7 @@ export function JobDetailLive({ initialJob }: JobDetailLiveProps) {
146256
)}
147257
{canRunNow && (
148258
<button
259+
type="button"
149260
onClick={() => handleAction("runNow")}
150261
className="rounded-md border px-3 py-2 text-sm hover:bg-muted"
151262
>
@@ -154,6 +265,7 @@ export function JobDetailLive({ initialJob }: JobDetailLiveProps) {
154265
)}
155266
{canRetry && (
156267
<button
268+
type="button"
157269
onClick={() => handleAction("retry")}
158270
className="rounded-md border px-3 py-2 text-sm hover:bg-muted"
159271
>
@@ -162,6 +274,7 @@ export function JobDetailLive({ initialJob }: JobDetailLiveProps) {
162274
)}
163275
{canForceUnlock && (
164276
<button
277+
type="button"
165278
onClick={() => handleAction("forceUnlock")}
166279
className="rounded-md border px-3 py-2 text-sm hover:bg-muted"
167280
>
@@ -282,6 +395,53 @@ export function JobDetailLive({ initialJob }: JobDetailLiveProps) {
282395
error={job.lastError}
283396
onRetry={() => handleAction("retry")}
284397
/>
398+
399+
<div className="mt-6 rounded-lg border p-4">
400+
<div className="mb-2 flex items-center justify-between gap-3">
401+
<h2 className="font-semibold">Logs</h2>
402+
<div className="text-xs text-muted-foreground">
403+
{logsConnected ? "Live" : "Reconnecting..."}
404+
</div>
405+
</div>
406+
{logsTruncated && (
407+
<div className="mb-2 text-xs text-muted-foreground">
408+
Showing the latest log tail.
409+
</div>
410+
)}
411+
<div
412+
ref={logsContainerRef}
413+
className="max-h-96 overflow-y-auto rounded-md border bg-black p-3 font-mono text-xs"
414+
>
415+
{jobLogs.length === 0 ? (
416+
<div className="text-gray-400">No logs yet...</div>
417+
) : (
418+
jobLogs.map((line, index) => {
419+
const level = line.level.toLowerCase();
420+
const isError = level === "error" || level === "fatal";
421+
const colorClass = isError
422+
? "text-status-error"
423+
: level === "warn"
424+
? "text-yellow-300"
425+
: "text-status-success";
426+
427+
return (
428+
<div
429+
key={`${line.timestamp}-${index}`}
430+
className={colorClass}
431+
>
432+
{line.timestamp && (
433+
<span className="text-gray-500">[{line.timestamp}] </span>
434+
)}
435+
<span className="uppercase text-gray-400">
436+
{line.level}
437+
</span>{" "}
438+
{line.message}
439+
</div>
440+
);
441+
})
442+
)}
443+
</div>
444+
</div>
285445
</main>
286446

287447
{pendingAction && (

‎src/pages/api/queue/job-logs-stream.ts‎

Lines changed: 160 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,160 @@
1+
import type { APIRoute } from "astro";
2+
import { logger } from "@/server/logger";
3+
import { canUserAccessQueueJob } from "@/server/queue/access";
4+
import { readJobLogFromOffset, readJobLogTail } from "@/server/queue/job-logs";
5+
import { getJobById } from "@/server/queue/queue.model";
6+
7+
const KEEP_ALIVE_INTERVAL_MS = 15_000;
8+
const POLL_INTERVAL_MS = 1_000;
9+
10+
export const GET: APIRoute = async ({ request, url, locals }) => {
11+
const user = locals.user;
12+
if (!user) {
13+
return new Response("Unauthorized", { status: 401 });
14+
}
15+
16+
const jobId = url.searchParams.get("jobId");
17+
if (!jobId) {
18+
return new Response("Job ID required", { status: 400 });
19+
}
20+
21+
const job = await getJobById(jobId);
22+
if (!job) {
23+
return new Response("Job not found", { status: 404 });
24+
}
25+
26+
const canAccessJob = await canUserAccessQueueJob(user.id, job);
27+
if (!canAccessJob) {
28+
return new Response("Not found", { status: 404 });
29+
}
30+
31+
const offsetParam = url.searchParams.get("offset");
32+
const requestedOffset = offsetParam ? Number.parseInt(offsetParam, 10) : null;
33+
if (requestedOffset !== null && Number.isNaN(requestedOffset)) {
34+
return new Response("Invalid offset", { status: 400 });
35+
}
36+
37+
const encoder = new TextEncoder();
38+
let isClosed = false;
39+
let keepAliveTimer: ReturnType<typeof setInterval> | null = null;
40+
let pollTimer: ReturnType<typeof setInterval> | null = null;
41+
let lastOffset = requestedOffset ?? 0;
42+
43+
const stream = new ReadableStream({
44+
async start(controller) {
45+
const sendEvent = (data: object) => {
46+
if (isClosed) return;
47+
try {
48+
controller.enqueue(
49+
encoder.encode(
50+
`event: log.chunk\ndata: ${JSON.stringify(data)}\n\n`,
51+
),
52+
);
53+
} catch {
54+
isClosed = true;
55+
}
56+
};
57+
58+
const sendKeepAlive = () => {
59+
if (isClosed) return;
60+
try {
61+
controller.enqueue(encoder.encode(": keep-alive\n\n"));
62+
} catch {
63+
isClosed = true;
64+
}
65+
};
66+
67+
const cleanup = () => {
68+
isClosed = true;
69+
if (keepAliveTimer) clearInterval(keepAliveTimer);
70+
if (pollTimer) clearInterval(pollTimer);
71+
};
72+
73+
try {
74+
if (requestedOffset === null) {
75+
const { content, offset, truncated } = await readJobLogTail(jobId);
76+
lastOffset = offset;
77+
sendEvent({
78+
jobId,
79+
offset: 0,
80+
nextOffset: offset,
81+
text: content,
82+
truncated,
83+
});
84+
} else if (requestedOffset >= 0) {
85+
const { content, nextOffset } = await readJobLogFromOffset(
86+
jobId,
87+
requestedOffset,
88+
);
89+
lastOffset = nextOffset;
90+
if (content) {
91+
sendEvent({
92+
jobId,
93+
offset: requestedOffset,
94+
nextOffset,
95+
text: content,
96+
truncated: false,
97+
});
98+
}
99+
}
100+
101+
pollTimer = setInterval(async () => {
102+
if (isClosed) return;
103+
try {
104+
const { content, nextOffset } = await readJobLogFromOffset(
105+
jobId,
106+
lastOffset,
107+
);
108+
if (!content) return;
109+
110+
sendEvent({
111+
jobId,
112+
offset: lastOffset,
113+
nextOffset,
114+
text: content,
115+
truncated: false,
116+
});
117+
lastOffset = nextOffset;
118+
} catch (error) {
119+
logger.warn(
120+
{ error, jobId },
121+
"Failed reading job log stream chunk",
122+
);
123+
}
124+
}, POLL_INTERVAL_MS);
125+
126+
keepAliveTimer = setInterval(sendKeepAlive, KEEP_ALIVE_INTERVAL_MS);
127+
128+
request.signal?.addEventListener("abort", () => {
129+
cleanup();
130+
try {
131+
controller.close();
132+
} catch (error) {
133+
logger.debug(
134+
{ error, jobId },
135+
"Queue job logs stream already closed",
136+
);
137+
}
138+
});
139+
} catch (error) {
140+
cleanup();
141+
logger.error({ error, jobId }, "Error in queue job logs stream");
142+
controller.error(error);
143+
}
144+
},
145+
146+
cancel() {
147+
isClosed = true;
148+
if (keepAliveTimer) clearInterval(keepAliveTimer);
149+
if (pollTimer) clearInterval(pollTimer);
150+
},
151+
});
152+
153+
return new Response(stream, {
154+
headers: {
155+
"Content-Type": "text/event-stream",
156+
"Cache-Control": "no-cache",
157+
Connection: "keep-alive",
158+
},
159+
});
160+
};

0 commit comments

Comments
 (0)