339 lines
12 KiB
TypeScript
339 lines
12 KiB
TypeScript
import React, { useMemo, useState, useEffect, useRef } from "react";
|
|
import { useParams, useNavigate } from "react-router-dom";
|
|
import {
|
|
Box,
|
|
Container,
|
|
Paper,
|
|
Typography,
|
|
Button,
|
|
Chip,
|
|
CircularProgress,
|
|
Alert,
|
|
IconButton,
|
|
Snackbar,
|
|
} from "@mui/material";
|
|
import ArrowBackIcon from "@mui/icons-material/ArrowBack";
|
|
import ReplayIcon from "@mui/icons-material/Replay";
|
|
import CheckCircleIcon from "@mui/icons-material/CheckCircle";
|
|
import ErrorIcon from "@mui/icons-material/Error";
|
|
import WarningAmberIcon from "@mui/icons-material/WarningAmber";
|
|
import PlayArrowIcon from "@mui/icons-material/PlayArrow";
|
|
import RemoveCircleOutlineIcon from "@mui/icons-material/RemoveCircleOutline";
|
|
import FiberManualRecordIcon from "@mui/icons-material/FiberManualRecord";
|
|
import { useResource, useItemSse, DetailFieldRenderer, applyDisplayFormat } from "../../react-openapi";
|
|
import { useQuery, useMutation } from "@tanstack/react-query";
|
|
import { RETRY_MAX, formatApiError } from "../features/fetch-requests";
|
|
import type { FetchRequestStatus, SSEEvent, ProgressMessage } from "../features/fetch-requests";
|
|
import { PipelineStepper } from "./components/PipelineStepper";
|
|
import { AmbiguityResolver } from "./components/AmbiguityResolver";
|
|
|
|
const statusColors: Record<FetchRequestStatus, "default" | "primary" | "warning" | "info" | "success" | "error"> = {
|
|
pending: "default",
|
|
processing: "info",
|
|
paused: "warning",
|
|
raw_expenses_done: "primary",
|
|
enriched_done: "warning",
|
|
completed: "success",
|
|
failed: "error",
|
|
};
|
|
|
|
const statusIcons: Record<FetchRequestStatus, React.ReactNode> = {
|
|
pending: <PlayArrowIcon sx={{ fontSize: 16 }} />,
|
|
processing: <CircularProgress size={14} />,
|
|
paused: <WarningAmberIcon sx={{ fontSize: 16 }} />,
|
|
raw_expenses_done: <CheckCircleIcon sx={{ fontSize: 16 }} />,
|
|
enriched_done: <CheckCircleIcon sx={{ fontSize: 16 }} />,
|
|
completed: <CheckCircleIcon sx={{ fontSize: 16 }} />,
|
|
failed: <ErrorIcon sx={{ fontSize: 16 }} />,
|
|
};
|
|
|
|
function formatProgressMessage(msg: ProgressMessage): string {
|
|
if (msg.lines !== undefined) return `${msg.lines} lines`;
|
|
if (msg.blocks !== undefined) return `${msg.blocks} blocks`;
|
|
if (msg.count !== undefined && msg.unit) return `${msg.count} ${msg.unit}`;
|
|
if (msg.count !== undefined) return `${msg.count} items`;
|
|
if (msg.raw_ocr_line) return `"${msg.raw_ocr_line.slice(0, 60)}${msg.raw_ocr_line.length > 60 ? "…" : ""}"`;
|
|
if (msg.error) return msg.error.slice(0, 80);
|
|
return "";
|
|
}
|
|
|
|
function sseIcon(status: SSEEvent["status"]) {
|
|
switch (status) {
|
|
case "started": return <CircularProgress size={14} />;
|
|
case "completed": return <CheckCircleIcon sx={{ fontSize: 16, color: "success.main" }} />;
|
|
case "failed": return <ErrorIcon sx={{ fontSize: 16, color: "error.main" }} />;
|
|
case "skipped": return <RemoveCircleOutlineIcon sx={{ fontSize: 16, color: "text.disabled" }} />;
|
|
case "paused": return <WarningAmberIcon sx={{ fontSize: 16, color: "warning.main" }} />;
|
|
case "progress": return <FiberManualRecordIcon sx={{ fontSize: 14, color: "info.main" }} />;
|
|
}
|
|
}
|
|
|
|
export default function FetchRequestDetail() {
|
|
const { id } = useParams<{ id: string }>();
|
|
const navigate = useNavigate();
|
|
const { get, update, resource } = useResource("fetch-requests");
|
|
const [stepStats, setStepStats] = useState<Record<string, number>>({});
|
|
const [liveParsedCount, setLiveParsedCount] = useState<number | undefined>(undefined);
|
|
const [failNotif, setFailNotif] = useState<string | null>(null);
|
|
const feedRef = useRef<HTMLDivElement>(null);
|
|
|
|
const { data: fetchRequest, isLoading, error: fetchError, refetch: refetchRequest } = useQuery({
|
|
queryKey: ["fetch-requests", "detail", id],
|
|
queryFn: () => get(id!),
|
|
enabled: !!id,
|
|
});
|
|
|
|
const updateMutation = useMutation({
|
|
mutationFn: ({ id: rid, data }: { id: string; data: any }) => update(rid, data),
|
|
});
|
|
|
|
const sseUrl = id ? `/fetch-requests/${id}/events` : null;
|
|
const { connected: sseConnected, events: sseEvents } = useItemSse(sseUrl, {
|
|
onEvent: (parsed: SSEEvent) => {
|
|
if (parsed.status === "progress" && parsed.message.count !== undefined) {
|
|
if (parsed.step === "txn_dicts") setLiveParsedCount(parsed.message.count);
|
|
if (parsed.step === "enrich") setStepStats((prev) => ({ ...prev, enrich_count: parsed.message.count! }));
|
|
if (parsed.step === "save_expenses") setStepStats((prev) => ({ ...prev, save_count: parsed.message.count! }));
|
|
}
|
|
|
|
if (parsed.status === "completed" && parsed.message.count !== undefined) {
|
|
const stats: Record<string, number> = {};
|
|
if (parsed.step === "raw_lines" && parsed.message.lines !== undefined) stats.raw_lines = parsed.message.lines;
|
|
if (parsed.step === "txn_blocks" && parsed.message.blocks !== undefined) stats.txn_blocks = parsed.message.blocks;
|
|
if (parsed.step === "txn_dicts") stats.txn_dicts = parsed.message.count;
|
|
if (parsed.step === "enrich") stats.enrich_count = parsed.message.count;
|
|
if (parsed.step === "save_expenses") stats.save_count = parsed.message.count;
|
|
if (Object.keys(stats).length) {
|
|
setStepStats((prev) => ({ ...prev, ...stats }));
|
|
}
|
|
}
|
|
|
|
if (parsed.status === "paused") {
|
|
refetchRequest();
|
|
}
|
|
if (parsed.status === "failed") {
|
|
setFailNotif(parsed.message.error || "Fetch request failed");
|
|
refetchRequest();
|
|
}
|
|
if (parsed.status === "completed" || parsed.step === "resume_extract") {
|
|
refetchRequest();
|
|
}
|
|
},
|
|
});
|
|
|
|
useEffect(() => {
|
|
if (feedRef.current) {
|
|
feedRef.current.scrollTop = feedRef.current.scrollHeight;
|
|
}
|
|
}, [sseEvents]);
|
|
|
|
const displayEvents = useMemo(() => {
|
|
const progressSteps = new Set(["txn_dicts", "enrich", "save_expenses"]);
|
|
const lastProgressIdx: Record<string, number> = {};
|
|
for (let i = sseEvents.length - 1; i >= 0; i--) {
|
|
const e = sseEvents[i];
|
|
if (progressSteps.has(e.step) && e.status === "progress" && lastProgressIdx[e.step] === undefined) {
|
|
lastProgressIdx[e.step] = i;
|
|
}
|
|
}
|
|
|
|
const terminalStatuses = new Set(["completed", "skipped", "paused", "failed"]);
|
|
return sseEvents.filter((e, i) => {
|
|
if (progressSteps.has(e.step) && e.status === "progress") return i === lastProgressIdx[e.step];
|
|
if (e.status === "started") {
|
|
return !sseEvents.slice(i + 1).some(
|
|
(later) => later.step === e.step && terminalStatuses.has(later.status),
|
|
);
|
|
}
|
|
return true;
|
|
});
|
|
}, [sseEvents]);
|
|
|
|
const handleRetry = async () => {
|
|
if (!id) return;
|
|
try {
|
|
await updateMutation.mutateAsync({ id, data: { status: "pending" } });
|
|
refetchRequest();
|
|
} catch (err: any) {
|
|
setFailNotif(formatApiError(err));
|
|
}
|
|
};
|
|
|
|
if (isLoading) {
|
|
return (
|
|
<Box sx={{ display: "flex", justifyContent: "center", p: 8 }}>
|
|
<CircularProgress />
|
|
</Box>
|
|
);
|
|
}
|
|
|
|
if (fetchError || !fetchRequest) {
|
|
return (
|
|
<Container sx={{ mt: 4 }}>
|
|
<Button startIcon={<ArrowBackIcon />} onClick={() => navigate("/fetch-requests")} sx={{ mb: 2 }}>
|
|
Back
|
|
</Button>
|
|
<Alert severity="error">Failed to load fetch request</Alert>
|
|
</Container>
|
|
);
|
|
}
|
|
|
|
const req = fetchRequest as any;
|
|
const retryCount = req.retry_count ?? 0;
|
|
const isRetryExhausted = retryCount >= RETRY_MAX;
|
|
const status = req.status as FetchRequestStatus;
|
|
|
|
const detailFields = (resource?.orderedFields ?? []).filter(
|
|
(f) => f.name !== "source",
|
|
);
|
|
|
|
return (
|
|
<Container sx={{ mt: 4, mb: 4 }}>
|
|
<Button startIcon={<ArrowBackIcon />} onClick={() => navigate("/fetch-requests")} sx={{ mb: 2 }}>
|
|
Back to Fetch Requests
|
|
</Button>
|
|
|
|
<Paper sx={{ p: 3, borderRadius: 4, mb: 3 }} variant="outlined">
|
|
<Box sx={{ display: "flex", alignItems: "center", gap: 2, mb: 2, flexWrap: "wrap" }}>
|
|
<Chip
|
|
icon={statusIcons[status] as any}
|
|
label={status.replace(/_/g, " ")}
|
|
color={statusColors[status]}
|
|
/>
|
|
<Typography variant="h6" fontWeight={600}>{req.account_name}</Typography>
|
|
<Chip
|
|
label={"path" in (req.source ?? {}) ? "File" : "Email"}
|
|
size="small"
|
|
variant="outlined"
|
|
color={"path" in (req.source ?? {}) ? "primary" : "secondary"}
|
|
/>
|
|
</Box>
|
|
|
|
<Box sx={{ display: "flex", gap: 4, flexWrap: "wrap", mb: 2 }}>
|
|
{detailFields.map((field) => (
|
|
<DetailFieldRenderer
|
|
key={field.name}
|
|
field={field}
|
|
value={req[field.name]}
|
|
displayFormat={resource?.displayFormat}
|
|
/>
|
|
))}
|
|
</Box>
|
|
|
|
<Box sx={{ display: "flex", alignItems: "center", gap: 2 }}>
|
|
<Box sx={{ flex: 1, maxWidth: 300 }}>
|
|
<Typography variant="caption" color="text.secondary">
|
|
Retries: {retryCount}/{RETRY_MAX}
|
|
</Typography>
|
|
</Box>
|
|
{status === "failed" && !isRetryExhausted && (
|
|
<Button
|
|
variant="outlined"
|
|
size="small"
|
|
startIcon={<ReplayIcon />}
|
|
onClick={handleRetry}
|
|
disabled={updateMutation.isPending}
|
|
>
|
|
Retry
|
|
</Button>
|
|
)}
|
|
</Box>
|
|
</Paper>
|
|
|
|
{status === "failed" && req.error_message && (
|
|
<Alert severity="error" sx={{ mb: 3, borderRadius: 2 }}>
|
|
{req.error_message}
|
|
</Alert>
|
|
)}
|
|
{isRetryExhausted && status === "failed" && (
|
|
<Alert severity="info" sx={{ mb: 3, borderRadius: 2 }}>
|
|
Max retries reached — no further retry attempts will be made.
|
|
</Alert>
|
|
)}
|
|
|
|
<PipelineStepper
|
|
fetchRequest={req}
|
|
sseEvents={sseEvents}
|
|
stepStats={stepStats}
|
|
liveParsedCount={liveParsedCount ?? 0}
|
|
/>
|
|
|
|
<Paper sx={{ borderRadius: 4, mb: 3 }} variant="outlined">
|
|
<Box sx={{ display: "flex", alignItems: "center", gap: 1, p: 2, pb: 0 }}>
|
|
<Typography variant="subtitle1" fontWeight={600} sx={{ flex: 1 }}>
|
|
Progress Events
|
|
</Typography>
|
|
<Box
|
|
sx={{
|
|
width: 10,
|
|
height: 10,
|
|
borderRadius: "50%",
|
|
bgcolor: sseConnected ? "success.main" : "error.main",
|
|
flexShrink: 0,
|
|
}}
|
|
/>
|
|
<Typography variant="caption" color="text.secondary">
|
|
{sseConnected ? "Connected" : "Disconnected"}
|
|
</Typography>
|
|
</Box>
|
|
<Box
|
|
ref={feedRef}
|
|
sx={{
|
|
maxHeight: 300,
|
|
overflowY: "auto",
|
|
p: 2,
|
|
display: "flex",
|
|
flexDirection: "column",
|
|
gap: 1,
|
|
}}
|
|
>
|
|
{displayEvents.length === 0 ? (
|
|
<Typography variant="body2" color="text.disabled" sx={{ textAlign: "center", py: 2 }}>
|
|
Waiting for events...
|
|
</Typography>
|
|
) : (
|
|
displayEvents.map((evt, i) => (
|
|
<Box
|
|
key={i}
|
|
sx={{
|
|
display: "flex",
|
|
alignItems: "center",
|
|
gap: 1.5,
|
|
p: 1,
|
|
borderRadius: 2,
|
|
bgcolor: "action.hover",
|
|
}}
|
|
>
|
|
{sseIcon(evt.status)}
|
|
<Box sx={{ flex: 1 }}>
|
|
<Typography variant="body2" fontWeight={600}>
|
|
{evt.step.replace(/_/g, " ")}
|
|
</Typography>
|
|
{evt.message && formatProgressMessage(evt.message) && (
|
|
<Typography variant="caption" color="text.secondary">
|
|
{formatProgressMessage(evt.message)}
|
|
</Typography>
|
|
)}
|
|
</Box>
|
|
</Box>
|
|
))
|
|
)}
|
|
</Box>
|
|
</Paper>
|
|
|
|
<AmbiguityResolver fetchRequestId={id!} />
|
|
|
|
<Snackbar
|
|
open={!!failNotif}
|
|
autoHideDuration={6000}
|
|
onClose={() => setFailNotif(null)}
|
|
anchorOrigin={{ vertical: "bottom", horizontal: "center" }}
|
|
>
|
|
<Alert severity="error" onClose={() => setFailNotif(null)} sx={{ borderRadius: 2 }}>
|
|
{failNotif}
|
|
</Alert>
|
|
</Snackbar>
|
|
</Container>
|
|
);
|
|
}
|