feat: restore progress (#281)
* feat: restore progress * feat: keep restore progress on reload * refactor: centralize sse event types * refactor(sse): generic handler factory
This commit is contained in:
Nico
GitHub
parent
1017f1a38b
commit
bad944a232
11 changed files with 530 additions and 419 deletions
|
|
@ -1,16 +1,26 @@
|
|||
import { useCallback, useState } from "react";
|
||||
import { useCallback, useEffect, useRef, useState } from "react";
|
||||
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
|
||||
import { toast } from "sonner";
|
||||
import { ChevronDown, FileIcon, FolderOpen, RotateCcw } from "lucide-react";
|
||||
import { Button } from "~/client/components/ui/button";
|
||||
import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "~/client/components/ui/card";
|
||||
import { Input } from "~/client/components/ui/input";
|
||||
import { Label } from "~/client/components/ui/label";
|
||||
import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "~/client/components/ui/select";
|
||||
import {
|
||||
AlertDialog,
|
||||
AlertDialogAction,
|
||||
AlertDialogContent,
|
||||
AlertDialogDescription,
|
||||
AlertDialogFooter,
|
||||
AlertDialogHeader,
|
||||
AlertDialogTitle,
|
||||
} from "~/client/components/ui/alert-dialog";
|
||||
import { PathSelector } from "~/client/components/path-selector";
|
||||
import { FileTree } from "~/client/components/file-tree";
|
||||
import { RestoreProgress } from "~/client/components/restore-progress";
|
||||
import { listSnapshotFilesOptions, restoreSnapshotMutation } from "~/client/api-client/@tanstack/react-query.gen";
|
||||
import { useFileBrowser } from "~/client/hooks/use-file-browser";
|
||||
import { type RestoreCompletedEvent, useServerEvents } from "~/client/hooks/use-server-events";
|
||||
import { OVERWRITE_MODES, type OverwriteMode } from "~/schemas/restic";
|
||||
import type { Repository, Snapshot } from "~/client/lib/types";
|
||||
import { handleRepositoryError } from "~/client/lib/errors";
|
||||
|
|
@ -27,6 +37,7 @@ interface RestoreFormProps {
|
|||
|
||||
export function RestoreForm({ snapshot, repository, snapshotId, returnPath }: RestoreFormProps) {
|
||||
const navigate = useNavigate();
|
||||
const { addEventListener } = useServerEvents();
|
||||
const queryClient = useQueryClient();
|
||||
|
||||
const volumeBasePath = snapshot.paths[0]?.match(/^(.*?_data)(\/|$)/)?.[1] || "/";
|
||||
|
|
@ -36,9 +47,48 @@ export function RestoreForm({ snapshot, repository, snapshotId, returnPath }: Re
|
|||
const [overwriteMode, setOverwriteMode] = useState<OverwriteMode>("always");
|
||||
const [showAdvanced, setShowAdvanced] = useState(false);
|
||||
const [excludeXattr, setExcludeXattr] = useState("");
|
||||
const [isRestoreActive, setIsRestoreActive] = useState(false);
|
||||
const [restoreResult, setRestoreResult] = useState<RestoreCompletedEvent | null>(null);
|
||||
const [showRestoreResultAlert, setShowRestoreResultAlert] = useState(false);
|
||||
const restoreCompletedRef = useRef(false);
|
||||
|
||||
const [selectedPaths, setSelectedPaths] = useState<Set<string>>(new Set());
|
||||
|
||||
useEffect(() => {
|
||||
const unsubscribeStarted = addEventListener("restore:started", (startedData) => {
|
||||
if (startedData.repositoryId === repository.id && startedData.snapshotId === snapshotId) {
|
||||
restoreCompletedRef.current = false;
|
||||
setIsRestoreActive(true);
|
||||
setRestoreResult(null);
|
||||
setShowRestoreResultAlert(false);
|
||||
}
|
||||
});
|
||||
|
||||
const unsubscribeProgress = addEventListener("restore:progress", (progressData) => {
|
||||
if (progressData.repositoryId === repository.id && progressData.snapshotId === snapshotId) {
|
||||
if (restoreCompletedRef.current) {
|
||||
return;
|
||||
}
|
||||
setIsRestoreActive(true);
|
||||
}
|
||||
});
|
||||
|
||||
const unsubscribeCompleted = addEventListener("restore:completed", (completedData) => {
|
||||
if (completedData.repositoryId === repository.id && completedData.snapshotId === snapshotId) {
|
||||
restoreCompletedRef.current = true;
|
||||
setIsRestoreActive(false);
|
||||
setRestoreResult(completedData);
|
||||
setShowRestoreResultAlert(true);
|
||||
}
|
||||
});
|
||||
|
||||
return () => {
|
||||
unsubscribeStarted();
|
||||
unsubscribeProgress();
|
||||
unsubscribeCompleted();
|
||||
};
|
||||
}, [addEventListener, repository.id, snapshotId]);
|
||||
|
||||
const { data: filesData, isLoading: filesLoading } = useQuery({
|
||||
...listSnapshotFilesOptions({
|
||||
path: { id: repository.id, snapshotId },
|
||||
|
|
@ -97,11 +147,9 @@ export function RestoreForm({ snapshot, repository, snapshotId, returnPath }: Re
|
|||
|
||||
const { mutate: restoreSnapshot, isPending: isRestoring } = useMutation({
|
||||
...restoreSnapshotMutation(),
|
||||
onSuccess: () => {
|
||||
toast.success("Restore completed");
|
||||
void navigate({ to: returnPath });
|
||||
},
|
||||
onError: (error) => {
|
||||
restoreCompletedRef.current = true;
|
||||
setIsRestoreActive(false);
|
||||
handleRepositoryError("Restore failed", error, repository.id);
|
||||
},
|
||||
});
|
||||
|
|
@ -118,6 +166,11 @@ export function RestoreForm({ snapshot, repository, snapshotId, returnPath }: Re
|
|||
const pathsArray = Array.from(selectedPaths);
|
||||
const includePaths = pathsArray.map((path) => addBasePath(path));
|
||||
|
||||
restoreCompletedRef.current = false;
|
||||
setIsRestoreActive(true);
|
||||
setRestoreResult(null);
|
||||
setShowRestoreResultAlert(false);
|
||||
|
||||
restoreSnapshot({
|
||||
path: { id: repository.id },
|
||||
body: {
|
||||
|
|
@ -140,7 +193,19 @@ export function RestoreForm({ snapshot, repository, snapshotId, returnPath }: Re
|
|||
restoreSnapshot,
|
||||
]);
|
||||
|
||||
const acknowledgeRestoreResult = useCallback(() => {
|
||||
setShowRestoreResultAlert(false);
|
||||
setRestoreResult(null);
|
||||
}, []);
|
||||
|
||||
const handleResultAlertOpenChange = useCallback((open: boolean) => {
|
||||
if (open) {
|
||||
setShowRestoreResultAlert(true);
|
||||
}
|
||||
}, []);
|
||||
|
||||
const canRestore = restoreLocation === "original" || customTargetPath.trim();
|
||||
const isRestoreRunning = isRestoring || isRestoreActive;
|
||||
|
||||
return (
|
||||
<div className="space-y-6">
|
||||
|
|
@ -155,9 +220,9 @@ export function RestoreForm({ snapshot, repository, snapshotId, returnPath }: Re
|
|||
<Button variant="outline" onClick={() => navigate({ to: returnPath })}>
|
||||
Cancel
|
||||
</Button>
|
||||
<Button variant="primary" onClick={handleRestore} disabled={isRestoring || !canRestore}>
|
||||
<Button variant="primary" onClick={handleRestore} disabled={isRestoreRunning || !canRestore}>
|
||||
<RotateCcw className="h-4 w-4 mr-2" />
|
||||
{isRestoring
|
||||
{isRestoreRunning
|
||||
? "Restoring..."
|
||||
: selectedPaths.size > 0
|
||||
? `Restore ${selectedPaths.size} ${selectedPaths.size === 1 ? "item" : "items"}`
|
||||
|
|
@ -168,6 +233,8 @@ export function RestoreForm({ snapshot, repository, snapshotId, returnPath }: Re
|
|||
|
||||
<div className="grid grid-cols-1 lg:grid-cols-3 gap-6">
|
||||
<div className="space-y-6">
|
||||
{isRestoreRunning && <RestoreProgress repositoryId={repository.id} snapshotId={snapshotId} />}
|
||||
|
||||
<Card>
|
||||
<CardHeader>
|
||||
<CardTitle>Restore Location</CardTitle>
|
||||
|
|
@ -305,6 +372,24 @@ export function RestoreForm({ snapshot, repository, snapshotId, returnPath }: Re
|
|||
</CardContent>
|
||||
</Card>
|
||||
</div>
|
||||
|
||||
<AlertDialog open={showRestoreResultAlert} onOpenChange={handleResultAlertOpenChange}>
|
||||
<AlertDialogContent>
|
||||
<AlertDialogHeader>
|
||||
<AlertDialogTitle>
|
||||
{restoreResult?.status === "success" ? "Restore completed" : "Restore failed"}
|
||||
</AlertDialogTitle>
|
||||
<AlertDialogDescription>
|
||||
{restoreResult?.status === "success"
|
||||
? `Snapshot ${snapshotId} was restored successfully.`
|
||||
: restoreResult?.error || `Snapshot ${snapshotId} could not be restored.`}
|
||||
</AlertDialogDescription>
|
||||
</AlertDialogHeader>
|
||||
<AlertDialogFooter>
|
||||
<AlertDialogAction onClick={acknowledgeRestoreResult}>OK</AlertDialogAction>
|
||||
</AlertDialogFooter>
|
||||
</AlertDialogContent>
|
||||
</AlertDialog>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
|
|
|||
91
app/client/components/restore-progress.tsx
Normal file
91
app/client/components/restore-progress.tsx
Normal file
|
|
@ -0,0 +1,91 @@
|
|||
import { useEffect, useState } from "react";
|
||||
import { ByteSize } from "~/client/components/bytes-size";
|
||||
import { Card } from "~/client/components/ui/card";
|
||||
import { Progress } from "~/client/components/ui/progress";
|
||||
import { type RestoreProgressEvent, useServerEvents } from "~/client/hooks/use-server-events";
|
||||
import { formatBytes } from "~/utils/format-bytes";
|
||||
import { formatDuration } from "~/utils/utils";
|
||||
|
||||
type Props = {
|
||||
repositoryId: string;
|
||||
snapshotId: string;
|
||||
};
|
||||
|
||||
export const RestoreProgress = ({ repositoryId, snapshotId }: Props) => {
|
||||
const { addEventListener } = useServerEvents();
|
||||
const [progress, setProgress] = useState<RestoreProgressEvent | null>(null);
|
||||
|
||||
useEffect(() => {
|
||||
const unsubscribe = addEventListener("restore:progress", (progressData) => {
|
||||
if (progressData.repositoryId === repositoryId && progressData.snapshotId === snapshotId) {
|
||||
setProgress(progressData);
|
||||
}
|
||||
});
|
||||
|
||||
const unsubscribeComplete = addEventListener("restore:completed", (completedData) => {
|
||||
if (completedData.repositoryId === repositoryId && completedData.snapshotId === snapshotId) {
|
||||
setProgress(null);
|
||||
}
|
||||
});
|
||||
|
||||
return () => {
|
||||
unsubscribe();
|
||||
unsubscribeComplete();
|
||||
};
|
||||
}, [addEventListener, repositoryId, snapshotId]);
|
||||
|
||||
if (!progress) {
|
||||
return (
|
||||
<Card className="p-4">
|
||||
<div className="flex items-center gap-2">
|
||||
<div className="h-4 w-4 animate-spin rounded-full border-2 border-primary border-t-transparent" />
|
||||
<span className="font-medium">Restore in progress</span>
|
||||
</div>
|
||||
</Card>
|
||||
);
|
||||
}
|
||||
|
||||
const percentDone = Math.round(progress.percent_done * 100);
|
||||
const speed = progress.seconds_elapsed > 0 ? formatBytes(progress.bytes_restored / progress.seconds_elapsed) : null;
|
||||
|
||||
return (
|
||||
<Card className="p-4">
|
||||
<div className="flex items-center justify-between mb-4">
|
||||
<div className="flex items-center gap-2">
|
||||
<div className="h-4 w-4 animate-spin rounded-full border-2 border-primary border-t-transparent" />
|
||||
<span className="font-medium">Restore in progress</span>
|
||||
</div>
|
||||
<span className="text-sm font-medium text-primary">{percentDone}%</span>
|
||||
</div>
|
||||
|
||||
<Progress value={percentDone} className="h-2 mb-4" />
|
||||
|
||||
<div className="grid grid-cols-2 gap-4 text-sm">
|
||||
<div>
|
||||
<p className="text-xs uppercase text-muted-foreground">Files</p>
|
||||
<p className="font-medium">
|
||||
{progress.files_restored.toLocaleString()}
|
||||
/
|
||||
{progress.total_files.toLocaleString()}
|
||||
</p>
|
||||
</div>
|
||||
<div>
|
||||
<p className="text-xs uppercase text-muted-foreground">Data</p>
|
||||
<p className="font-medium">
|
||||
<ByteSize bytes={progress.bytes_restored} base={1024} />
|
||||
/
|
||||
<ByteSize bytes={progress.total_bytes} base={1024} />
|
||||
</p>
|
||||
</div>
|
||||
<div>
|
||||
<p className="text-xs uppercase text-muted-foreground">Elapsed</p>
|
||||
<p className="font-medium">{formatDuration(progress.seconds_elapsed)}</p>
|
||||
</div>
|
||||
<div>
|
||||
<p className="text-xs uppercase text-muted-foreground">Speed</p>
|
||||
<p className="font-medium">{speed ? `${speed.text} ${speed.unit}/s` : "Calculating..."}</p>
|
||||
</div>
|
||||
</div>
|
||||
</Card>
|
||||
);
|
||||
};
|
||||
|
|
@ -1,56 +1,38 @@
|
|||
import { useCallback, useEffect, useRef } from "react";
|
||||
import { useQueryClient } from "@tanstack/react-query";
|
||||
import type {
|
||||
BackupCompletedEventDto,
|
||||
BackupProgressEventDto,
|
||||
BackupStartedEventDto,
|
||||
} from "~/schemas/events-dto";
|
||||
import { useCallback, useEffect, useRef } from "react";
|
||||
import { serverEventNames, type ServerEventPayloadMap } from "~/schemas/server-events";
|
||||
|
||||
type ServerEventType =
|
||||
| "connected"
|
||||
| "heartbeat"
|
||||
| "backup:started"
|
||||
| "backup:progress"
|
||||
| "backup:completed"
|
||||
| "volume:mounted"
|
||||
| "volume:unmounted"
|
||||
| "volume:updated"
|
||||
| "mirror:started"
|
||||
| "mirror:completed"
|
||||
| "doctor:started"
|
||||
| "doctor:completed"
|
||||
| "doctor:cancelled";
|
||||
type LifecycleEventPayloadMap = {
|
||||
connected: { type: "connected"; timestamp: number };
|
||||
heartbeat: { timestamp: number };
|
||||
};
|
||||
|
||||
export interface VolumeEvent {
|
||||
volumeName: string;
|
||||
}
|
||||
type ServerEventsPayloadMap = LifecycleEventPayloadMap & ServerEventPayloadMap;
|
||||
type ServerEventType = keyof ServerEventsPayloadMap;
|
||||
|
||||
export interface MirrorEvent {
|
||||
scheduleId: number;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
status?: "success" | "error" | "in_progress";
|
||||
error?: string;
|
||||
}
|
||||
type EventHandler<T extends ServerEventType> = (data: ServerEventsPayloadMap[T]) => void;
|
||||
type EventHandlerSet<T extends ServerEventType> = Set<EventHandler<T>>;
|
||||
type EventHandlerMap = {
|
||||
[K in ServerEventType]?: EventHandlerSet<K>;
|
||||
};
|
||||
|
||||
export interface DoctorEvent {
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
error?: string;
|
||||
}
|
||||
const invalidatingEvents = new Set<ServerEventType>([
|
||||
"backup:completed",
|
||||
"volume:updated",
|
||||
"volume:status_changed",
|
||||
"mirror:completed",
|
||||
"doctor:started",
|
||||
"doctor:completed",
|
||||
"doctor:cancelled",
|
||||
]);
|
||||
|
||||
export interface DoctorCompletedEvent extends DoctorEvent {
|
||||
success: boolean;
|
||||
completedAt: number;
|
||||
steps: Array<{
|
||||
step: string;
|
||||
success: boolean;
|
||||
output: string | null;
|
||||
error: string | null;
|
||||
}>;
|
||||
}
|
||||
export type RestoreEvent = ServerEventsPayloadMap["restore:started"] | ServerEventsPayloadMap["restore:completed"];
|
||||
export type RestoreProgressEvent = ServerEventsPayloadMap["restore:progress"];
|
||||
export type RestoreCompletedEvent = ServerEventsPayloadMap["restore:completed"];
|
||||
export type BackupProgressEvent = ServerEventsPayloadMap["backup:progress"];
|
||||
|
||||
type EventHandler = (data: unknown) => void;
|
||||
const parseEventData = <T extends ServerEventType>(event: Event): ServerEventsPayloadMap[T] =>
|
||||
JSON.parse((event as MessageEvent<string>).data) as ServerEventsPayloadMap[T];
|
||||
|
||||
/**
|
||||
* Hook to listen to Server-Sent Events (SSE) from the backend
|
||||
|
|
@ -59,140 +41,51 @@ type EventHandler = (data: unknown) => void;
|
|||
export function useServerEvents() {
|
||||
const queryClient = useQueryClient();
|
||||
const eventSourceRef = useRef<EventSource | null>(null);
|
||||
const handlersRef = useRef<Map<ServerEventType, Set<EventHandler>>>(new Map());
|
||||
const handlersRef = useRef<EventHandlerMap>({});
|
||||
const emit = useCallback(<T extends ServerEventType>(eventName: T, data: ServerEventsPayloadMap[T]) => {
|
||||
const handlers = handlersRef.current[eventName] as EventHandlerSet<T> | undefined;
|
||||
handlers?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
}, []);
|
||||
|
||||
useEffect(() => {
|
||||
const eventSource = new EventSource("/api/v1/events");
|
||||
eventSourceRef.current = eventSource;
|
||||
|
||||
eventSource.addEventListener("connected", () => {
|
||||
eventSource.addEventListener("connected", (event) => {
|
||||
const data = parseEventData<"connected">(event);
|
||||
console.info("[SSE] Connected to server events");
|
||||
emit("connected", data);
|
||||
});
|
||||
|
||||
eventSource.addEventListener("heartbeat", () => {});
|
||||
eventSource.addEventListener("heartbeat", (event) => {
|
||||
emit("heartbeat", parseEventData<"heartbeat">(event));
|
||||
});
|
||||
|
||||
eventSource.addEventListener("backup:started", (e) => {
|
||||
const data = JSON.parse(e.data) as BackupStartedEventDto;
|
||||
console.info("[SSE] Backup started:", data);
|
||||
for (const eventName of serverEventNames) {
|
||||
eventSource.addEventListener(eventName, (event) => {
|
||||
const data = parseEventData<typeof eventName>(event);
|
||||
console.info(`[SSE] ${eventName}:`, data);
|
||||
|
||||
handlersRef.current.get("backup:started")?.forEach((handler) => {
|
||||
handler(data);
|
||||
if (invalidatingEvents.has(eventName)) {
|
||||
void queryClient.invalidateQueries();
|
||||
}
|
||||
|
||||
if (eventName === "backup:completed") {
|
||||
void queryClient.refetchQueries();
|
||||
}
|
||||
|
||||
if (eventName === "volume:status_changed") {
|
||||
const statusData = data as ServerEventsPayloadMap["volume:status_changed"];
|
||||
emit("volume:status_changed", statusData);
|
||||
emit("volume:updated", statusData);
|
||||
return;
|
||||
}
|
||||
|
||||
emit(eventName, data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("backup:progress", (e) => {
|
||||
const data = JSON.parse(e.data) as BackupProgressEventDto;
|
||||
|
||||
handlersRef.current.get("backup:progress")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("backup:completed", (e) => {
|
||||
const data = JSON.parse(e.data) as BackupCompletedEventDto;
|
||||
console.info("[SSE] Backup completed:", data);
|
||||
|
||||
void queryClient.invalidateQueries();
|
||||
void queryClient.refetchQueries();
|
||||
|
||||
handlersRef.current.get("backup:completed")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("volume:mounted", (e) => {
|
||||
const data = JSON.parse(e.data) as VolumeEvent;
|
||||
console.info("[SSE] Volume mounted:", data);
|
||||
|
||||
handlersRef.current.get("volume:mounted")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("volume:unmounted", (e) => {
|
||||
const data = JSON.parse(e.data) as VolumeEvent;
|
||||
console.info("[SSE] Volume unmounted:", data);
|
||||
|
||||
handlersRef.current.get("volume:unmounted")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("volume:updated", (e) => {
|
||||
const data = JSON.parse(e.data) as VolumeEvent;
|
||||
console.info("[SSE] Volume updated:", data);
|
||||
|
||||
void queryClient.invalidateQueries();
|
||||
|
||||
handlersRef.current.get("volume:updated")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("volume:status_changed", (e) => {
|
||||
const data = JSON.parse(e.data) as VolumeEvent;
|
||||
console.info("[SSE] Volume status updated:", data);
|
||||
|
||||
void queryClient.invalidateQueries();
|
||||
|
||||
handlersRef.current.get("volume:updated")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("mirror:started", (e) => {
|
||||
const data = JSON.parse(e.data) as MirrorEvent;
|
||||
console.info("[SSE] Mirror copy started:", data);
|
||||
|
||||
handlersRef.current.get("mirror:started")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("mirror:completed", (e) => {
|
||||
const data = JSON.parse(e.data) as MirrorEvent;
|
||||
console.info("[SSE] Mirror copy completed:", data);
|
||||
|
||||
// Invalidate queries to refresh mirror status in the UI
|
||||
void queryClient.invalidateQueries();
|
||||
|
||||
handlersRef.current.get("mirror:completed")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("doctor:started", (e) => {
|
||||
const data = JSON.parse(e.data) as DoctorEvent;
|
||||
console.info("[SSE] Doctor started:", data);
|
||||
|
||||
void queryClient.invalidateQueries();
|
||||
|
||||
handlersRef.current.get("doctor:started")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("doctor:completed", (e) => {
|
||||
const data = JSON.parse(e.data) as DoctorCompletedEvent;
|
||||
console.info("[SSE] Doctor completed:", data);
|
||||
|
||||
void queryClient.invalidateQueries();
|
||||
|
||||
handlersRef.current.get("doctor:completed")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("doctor:cancelled", (e) => {
|
||||
const data = JSON.parse(e.data) as DoctorEvent;
|
||||
console.info("[SSE] Doctor cancelled:", data);
|
||||
|
||||
void queryClient.invalidateQueries();
|
||||
|
||||
handlersRef.current.get("doctor:cancelled")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
eventSource.onerror = (error) => {
|
||||
console.error("[SSE] Connection error:", error);
|
||||
|
|
@ -203,16 +96,20 @@ export function useServerEvents() {
|
|||
eventSource.close();
|
||||
eventSourceRef.current = null;
|
||||
};
|
||||
}, [queryClient]);
|
||||
}, [emit, queryClient]);
|
||||
|
||||
const addEventListener = useCallback((event: ServerEventType, handler: EventHandler) => {
|
||||
if (!handlersRef.current.has(event)) {
|
||||
handlersRef.current.set(event, new Set());
|
||||
}
|
||||
handlersRef.current.get(event)?.add(handler);
|
||||
const addEventListener = useCallback(<T extends ServerEventType>(eventName: T, handler: EventHandler<T>) => {
|
||||
const existingHandlers = handlersRef.current[eventName] as EventHandlerSet<T> | undefined;
|
||||
const eventHandlers = existingHandlers ?? new Set<EventHandler<T>>();
|
||||
eventHandlers.add(handler);
|
||||
handlersRef.current[eventName] = eventHandlers as EventHandlerMap[T];
|
||||
|
||||
return () => {
|
||||
handlersRef.current.get(event)?.delete(handler);
|
||||
const handlers = handlersRef.current[eventName] as EventHandlerSet<T> | undefined;
|
||||
handlers?.delete(handler);
|
||||
if (handlers && handlers.size === 0) {
|
||||
delete handlersRef.current[eventName];
|
||||
}
|
||||
};
|
||||
}, []);
|
||||
|
||||
|
|
|
|||
|
|
@ -2,8 +2,7 @@ import { useEffect, useState } from "react";
|
|||
import { ByteSize } from "~/client/components/bytes-size";
|
||||
import { Card } from "~/client/components/ui/card";
|
||||
import { Progress } from "~/client/components/ui/progress";
|
||||
import { useServerEvents } from "~/client/hooks/use-server-events";
|
||||
import type { BackupProgressEventDto } from "~/schemas/events-dto";
|
||||
import { type BackupProgressEvent, useServerEvents } from "~/client/hooks/use-server-events";
|
||||
import { formatDuration } from "~/utils/utils";
|
||||
import { formatBytes } from "~/utils/format-bytes";
|
||||
|
||||
|
|
@ -13,18 +12,16 @@ type Props = {
|
|||
|
||||
export const BackupProgressCard = ({ scheduleId }: Props) => {
|
||||
const { addEventListener } = useServerEvents();
|
||||
const [progress, setProgress] = useState<BackupProgressEventDto | null>(null);
|
||||
const [progress, setProgress] = useState<BackupProgressEvent | null>(null);
|
||||
|
||||
useEffect(() => {
|
||||
const unsubscribe = addEventListener("backup:progress", (data) => {
|
||||
const progressData = data as BackupProgressEventDto;
|
||||
const unsubscribe = addEventListener("backup:progress", (progressData) => {
|
||||
if (progressData.scheduleId === scheduleId) {
|
||||
setProgress(progressData);
|
||||
}
|
||||
});
|
||||
|
||||
const unsubscribeComplete = addEventListener("backup:completed", (data) => {
|
||||
const completedData = data as { scheduleId: number };
|
||||
const unsubscribeComplete = addEventListener("backup:completed", (completedData) => {
|
||||
if (completedData.scheduleId === scheduleId) {
|
||||
setProgress(null);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -99,8 +99,7 @@ export const ScheduleMirrorsConfig = ({ scheduleId, primaryRepositoryId, reposit
|
|||
}, [compatibility]);
|
||||
|
||||
useEffect(() => {
|
||||
const unsubscribeStarted = addEventListener("mirror:started", (data) => {
|
||||
const event = data as { scheduleId: number; repositoryId: string };
|
||||
const unsubscribeStarted = addEventListener("mirror:started", (event) => {
|
||||
if (event.scheduleId !== scheduleId) return;
|
||||
setAssignments((prev) => {
|
||||
const next = new Map(prev);
|
||||
|
|
@ -111,8 +110,7 @@ export const ScheduleMirrorsConfig = ({ scheduleId, primaryRepositoryId, reposit
|
|||
});
|
||||
});
|
||||
|
||||
const unsubscribeCompleted = addEventListener("mirror:completed", (data) => {
|
||||
const event = data as { scheduleId: number; repositoryId: string; status?: "success" | "error"; error?: string };
|
||||
const unsubscribeCompleted = addEventListener("mirror:completed", (event) => {
|
||||
if (event.scheduleId !== scheduleId) return;
|
||||
setAssignments((prev) => {
|
||||
const next = new Map(prev);
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ import { type } from "arktype";
|
|||
import { resticBackupProgressMetricsSchema, resticBackupRunSummarySchema } from "~/schemas/restic-dto";
|
||||
|
||||
export const backupEventStatusSchema = type("'success' | 'error' | 'stopped' | 'warning'");
|
||||
export const restoreEventStatusSchema = type("'success' | 'error'");
|
||||
|
||||
const backupEventBaseSchema = type({
|
||||
scheduleId: "number",
|
||||
|
|
@ -13,6 +14,20 @@ const organizationScopedSchema = type({
|
|||
organizationId: "string",
|
||||
});
|
||||
|
||||
const restoreEventBaseSchema = type({
|
||||
repositoryId: "string",
|
||||
snapshotId: "string",
|
||||
});
|
||||
|
||||
const restoreProgressMetricsSchema = type({
|
||||
seconds_elapsed: "number",
|
||||
percent_done: "number",
|
||||
total_files: "number",
|
||||
files_restored: "number",
|
||||
total_bytes: "number",
|
||||
bytes_restored: "number",
|
||||
});
|
||||
|
||||
export const backupStartedEventSchema = backupEventBaseSchema;
|
||||
|
||||
export const backupProgressEventSchema = backupEventBaseSchema.and(resticBackupProgressMetricsSchema);
|
||||
|
|
@ -24,16 +39,36 @@ export const backupCompletedEventSchema = backupEventBaseSchema.and(
|
|||
}),
|
||||
);
|
||||
|
||||
export const restoreStartedEventSchema = restoreEventBaseSchema;
|
||||
|
||||
export const restoreProgressEventSchema = restoreEventBaseSchema.and(restoreProgressMetricsSchema);
|
||||
|
||||
export const restoreCompletedEventSchema = restoreEventBaseSchema.and(
|
||||
type({ status: restoreEventStatusSchema, error: "string?" }),
|
||||
);
|
||||
|
||||
export const serverBackupStartedEventSchema = organizationScopedSchema.and(backupStartedEventSchema);
|
||||
|
||||
export const serverBackupProgressEventSchema = organizationScopedSchema.and(backupProgressEventSchema);
|
||||
|
||||
export const serverBackupCompletedEventSchema = organizationScopedSchema.and(backupCompletedEventSchema);
|
||||
|
||||
export const serverRestoreStartedEventSchema = organizationScopedSchema.and(restoreStartedEventSchema);
|
||||
|
||||
export const serverRestoreProgressEventSchema = organizationScopedSchema.and(restoreProgressEventSchema);
|
||||
|
||||
export const serverRestoreCompletedEventSchema = organizationScopedSchema.and(restoreCompletedEventSchema);
|
||||
|
||||
export type BackupEventStatusDto = typeof backupEventStatusSchema.infer;
|
||||
export type BackupStartedEventDto = typeof backupStartedEventSchema.infer;
|
||||
export type BackupProgressEventDto = typeof backupProgressEventSchema.infer;
|
||||
export type BackupCompletedEventDto = typeof backupCompletedEventSchema.infer;
|
||||
export type RestoreStartedEventDto = typeof restoreStartedEventSchema.infer;
|
||||
export type RestoreProgressEventDto = typeof restoreProgressEventSchema.infer;
|
||||
export type RestoreCompletedEventDto = typeof restoreCompletedEventSchema.infer;
|
||||
export type ServerBackupStartedEventDto = typeof serverBackupStartedEventSchema.infer;
|
||||
export type ServerBackupProgressEventDto = typeof serverBackupProgressEventSchema.infer;
|
||||
export type ServerBackupCompletedEventDto = typeof serverBackupCompletedEventSchema.infer;
|
||||
export type ServerRestoreStartedEventDto = typeof serverRestoreStartedEventSchema.infer;
|
||||
export type ServerRestoreProgressEventDto = typeof serverRestoreProgressEventSchema.infer;
|
||||
export type ServerRestoreCompletedEventDto = typeof serverRestoreCompletedEventSchema.infer;
|
||||
|
|
|
|||
64
app/schemas/server-events.ts
Normal file
64
app/schemas/server-events.ts
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
import type {
|
||||
ServerBackupCompletedEventDto,
|
||||
ServerBackupProgressEventDto,
|
||||
ServerBackupStartedEventDto,
|
||||
ServerRestoreCompletedEventDto,
|
||||
ServerRestoreProgressEventDto,
|
||||
ServerRestoreStartedEventDto,
|
||||
} from "~/schemas/events-dto";
|
||||
import type { DoctorResult } from "~/schemas/restic";
|
||||
|
||||
const payload = <T>() => undefined as unknown as T;
|
||||
|
||||
/**
|
||||
* Single runtime registry for all broadcastable server events.
|
||||
* Used as source-of-truth for both event names and payload typing.
|
||||
*/
|
||||
export const serverEventPayloads = {
|
||||
"backup:started": payload<ServerBackupStartedEventDto>(),
|
||||
"backup:progress": payload<ServerBackupProgressEventDto>(),
|
||||
"backup:completed": payload<ServerBackupCompletedEventDto>(),
|
||||
"restore:started": payload<ServerRestoreStartedEventDto>(),
|
||||
"restore:progress": payload<ServerRestoreProgressEventDto>(),
|
||||
"restore:completed": payload<ServerRestoreCompletedEventDto>(),
|
||||
"mirror:started": payload<{
|
||||
organizationId: string;
|
||||
scheduleId: number;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
}>(),
|
||||
"mirror:completed": payload<{
|
||||
organizationId: string;
|
||||
scheduleId: number;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
status: "success" | "error";
|
||||
error?: string;
|
||||
}>(),
|
||||
"volume:mounted": payload<{ organizationId: string; volumeName: string }>(),
|
||||
"volume:unmounted": payload<{ organizationId: string; volumeName: string }>(),
|
||||
"volume:updated": payload<{ organizationId: string; volumeName: string }>(),
|
||||
"volume:status_changed": payload<{ organizationId: string; volumeName: string; status: string }>(),
|
||||
"doctor:started": payload<{ organizationId: string; repositoryId: string; repositoryName: string }>(),
|
||||
"doctor:completed": payload<
|
||||
{
|
||||
organizationId: string;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
} & DoctorResult
|
||||
>(),
|
||||
"doctor:cancelled": payload<{
|
||||
organizationId: string;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
error?: string;
|
||||
}>(),
|
||||
} as const;
|
||||
|
||||
export type ServerEventPayloadMap = typeof serverEventPayloads;
|
||||
|
||||
export type ServerEventHandlers = {
|
||||
[K in keyof ServerEventPayloadMap]: (data: ServerEventPayloadMap[K]) => void;
|
||||
};
|
||||
|
||||
export const serverEventNames = Object.keys(serverEventPayloads) as Array<keyof ServerEventPayloadMap>;
|
||||
|
|
@ -1,55 +1,10 @@
|
|||
import { EventEmitter } from "node:events";
|
||||
import type { TypedEmitter } from "tiny-typed-emitter";
|
||||
import type { DoctorResult } from "~/schemas/restic";
|
||||
import type {
|
||||
ServerBackupCompletedEventDto,
|
||||
ServerBackupProgressEventDto,
|
||||
ServerBackupStartedEventDto,
|
||||
} from "~/schemas/events-dto";
|
||||
|
||||
/**
|
||||
* Event payloads for the SSE system
|
||||
*/
|
||||
interface ServerEvents {
|
||||
"backup:started": (data: ServerBackupStartedEventDto) => void;
|
||||
"backup:progress": (data: ServerBackupProgressEventDto) => void;
|
||||
"backup:completed": (data: ServerBackupCompletedEventDto) => void;
|
||||
"mirror:started": (data: {
|
||||
organizationId: string;
|
||||
scheduleId: number;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
}) => void;
|
||||
"mirror:completed": (data: {
|
||||
organizationId: string;
|
||||
scheduleId: number;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
status: "success" | "error";
|
||||
error?: string;
|
||||
}) => void;
|
||||
"volume:mounted": (data: { organizationId: string; volumeName: string }) => void;
|
||||
"volume:unmounted": (data: { organizationId: string; volumeName: string }) => void;
|
||||
"volume:updated": (data: { organizationId: string; volumeName: string }) => void;
|
||||
"volume:status_changed": (data: { organizationId: string; volumeName: string; status: string }) => void;
|
||||
"doctor:started": (data: { organizationId: string; repositoryId: string; repositoryName: string }) => void;
|
||||
"doctor:completed": (
|
||||
data: {
|
||||
organizationId: string;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
} & DoctorResult,
|
||||
) => void;
|
||||
"doctor:cancelled": (data: {
|
||||
organizationId: string;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
error?: string;
|
||||
}) => void;
|
||||
}
|
||||
import type { ServerEventHandlers } from "~/schemas/server-events";
|
||||
export type { ServerEventHandlers, ServerEventPayloadMap } from "~/schemas/server-events";
|
||||
|
||||
/**
|
||||
* Global event emitter for server-side events
|
||||
* Use this to emit events that should be broadcasted to connected clients via SSE
|
||||
*/
|
||||
export const serverEvents = new EventEmitter() as TypedEmitter<ServerEvents>;
|
||||
export const serverEvents = new EventEmitter() as TypedEmitter<ServerEventHandlers>;
|
||||
|
|
|
|||
|
|
@ -1,14 +1,36 @@
|
|||
import { Hono } from "hono";
|
||||
import { streamSSE } from "hono/streaming";
|
||||
import { logger } from "../../utils/logger";
|
||||
import { serverEvents } from "../../core/events";
|
||||
import { logger } from "../../utils/logger";
|
||||
import { requireAuth } from "../auth/auth.middleware";
|
||||
import type { DoctorResult } from "~/schemas/restic";
|
||||
import type {
|
||||
ServerBackupCompletedEventDto,
|
||||
ServerBackupProgressEventDto,
|
||||
ServerBackupStartedEventDto,
|
||||
} from "~/schemas/events-dto";
|
||||
import type { ServerEventPayloadMap } from "~/schemas/server-events";
|
||||
|
||||
type OrganizationScopedEvent = {
|
||||
[EventName in keyof ServerEventPayloadMap]: ServerEventPayloadMap[EventName] extends {
|
||||
organizationId: string;
|
||||
}
|
||||
? EventName
|
||||
: never;
|
||||
}[keyof ServerEventPayloadMap];
|
||||
|
||||
const broadcastEvents = [
|
||||
"backup:started",
|
||||
"backup:progress",
|
||||
"backup:completed",
|
||||
"volume:mounted",
|
||||
"volume:unmounted",
|
||||
"volume:updated",
|
||||
"mirror:started",
|
||||
"mirror:completed",
|
||||
"restore:started",
|
||||
"restore:progress",
|
||||
"restore:completed",
|
||||
"doctor:started",
|
||||
"doctor:completed",
|
||||
"doctor:cancelled",
|
||||
] as const satisfies OrganizationScopedEvent[];
|
||||
|
||||
type BroadcastEvent = (typeof broadcastEvents)[number];
|
||||
|
||||
export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
|
||||
logger.info("Client connected to SSE endpoint");
|
||||
|
|
@ -20,128 +42,27 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
|
|||
event: "connected",
|
||||
});
|
||||
|
||||
const onBackupStarted = async (data: ServerBackupStartedEventDto) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event: "backup:started",
|
||||
});
|
||||
const createOrganizationEventHandler = <EventName extends BroadcastEvent>(event: EventName) => {
|
||||
return async (data: ServerEventPayloadMap[EventName]) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event,
|
||||
});
|
||||
};
|
||||
};
|
||||
|
||||
const onBackupProgress = async (data: ServerBackupProgressEventDto) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event: "backup:progress",
|
||||
});
|
||||
};
|
||||
const eventHandlers = broadcastEvents.reduce(
|
||||
(handlers, event) => {
|
||||
handlers[event] = createOrganizationEventHandler(event);
|
||||
return handlers;
|
||||
},
|
||||
{} as { [EventName in BroadcastEvent]: (data: ServerEventPayloadMap[EventName]) => Promise<void> },
|
||||
);
|
||||
|
||||
const onBackupCompleted = async (data: ServerBackupCompletedEventDto) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event: "backup:completed",
|
||||
});
|
||||
};
|
||||
|
||||
const onVolumeMounted = async (data: { organizationId: string; volumeName: string }) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event: "volume:mounted",
|
||||
});
|
||||
};
|
||||
|
||||
const onVolumeUnmounted = async (data: { organizationId: string; volumeName: string }) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event: "volume:unmounted",
|
||||
});
|
||||
};
|
||||
|
||||
const onVolumeUpdated = async (data: { organizationId: string; volumeName: string }) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event: "volume:updated",
|
||||
});
|
||||
};
|
||||
|
||||
const onMirrorStarted = async (data: {
|
||||
organizationId: string;
|
||||
scheduleId: number;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
}) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event: "mirror:started",
|
||||
});
|
||||
};
|
||||
|
||||
const onMirrorCompleted = async (data: {
|
||||
organizationId: string;
|
||||
scheduleId: number;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
status: "success" | "error";
|
||||
error?: string;
|
||||
}) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event: "mirror:completed",
|
||||
});
|
||||
};
|
||||
|
||||
const onDoctorStarted = async (data: { organizationId: string; repositoryId: string; repositoryName: string }) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event: "doctor:started",
|
||||
});
|
||||
};
|
||||
|
||||
const onDoctorCompleted = async (
|
||||
data: {
|
||||
organizationId: string;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
} & DoctorResult,
|
||||
) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event: "doctor:completed",
|
||||
});
|
||||
};
|
||||
|
||||
const onDoctorCancelled = async (data: {
|
||||
organizationId: string;
|
||||
repositoryId: string;
|
||||
repositoryName: string;
|
||||
error?: string;
|
||||
}) => {
|
||||
if (data.organizationId !== organizationId) return;
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify(data),
|
||||
event: "doctor:cancelled",
|
||||
});
|
||||
};
|
||||
|
||||
serverEvents.on("backup:started", onBackupStarted);
|
||||
serverEvents.on("backup:progress", onBackupProgress);
|
||||
serverEvents.on("backup:completed", onBackupCompleted);
|
||||
serverEvents.on("volume:mounted", onVolumeMounted);
|
||||
serverEvents.on("volume:unmounted", onVolumeUnmounted);
|
||||
serverEvents.on("volume:updated", onVolumeUpdated);
|
||||
serverEvents.on("mirror:started", onMirrorStarted);
|
||||
serverEvents.on("mirror:completed", onMirrorCompleted);
|
||||
serverEvents.on("doctor:started", onDoctorStarted);
|
||||
serverEvents.on("doctor:completed", onDoctorCompleted);
|
||||
serverEvents.on("doctor:cancelled", onDoctorCancelled);
|
||||
for (const event of broadcastEvents) {
|
||||
serverEvents.on(event, eventHandlers[event]);
|
||||
}
|
||||
|
||||
let keepAlive = true;
|
||||
let cleanedUp = false;
|
||||
|
|
@ -151,17 +72,10 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
|
|||
cleanedUp = true;
|
||||
|
||||
c.req.raw.signal.removeEventListener("abort", onRequestAbort);
|
||||
serverEvents.off("backup:started", onBackupStarted);
|
||||
serverEvents.off("backup:progress", onBackupProgress);
|
||||
serverEvents.off("backup:completed", onBackupCompleted);
|
||||
serverEvents.off("volume:mounted", onVolumeMounted);
|
||||
serverEvents.off("volume:unmounted", onVolumeUnmounted);
|
||||
serverEvents.off("volume:updated", onVolumeUpdated);
|
||||
serverEvents.off("mirror:started", onMirrorStarted);
|
||||
serverEvents.off("mirror:completed", onMirrorCompleted);
|
||||
serverEvents.off("doctor:started", onDoctorStarted);
|
||||
serverEvents.off("doctor:completed", onDoctorCompleted);
|
||||
serverEvents.off("doctor:cancelled", onDoctorCancelled);
|
||||
|
||||
for (const event of broadcastEvents) {
|
||||
serverEvents.off(event, eventHandlers[event]);
|
||||
}
|
||||
}
|
||||
|
||||
function handleDisconnect() {
|
||||
|
|
@ -181,10 +95,7 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
|
|||
|
||||
try {
|
||||
while (keepAlive && !c.req.raw.signal.aborted && !stream.aborted) {
|
||||
await stream.writeSSE({
|
||||
data: JSON.stringify({ timestamp: Date.now() }),
|
||||
event: "heartbeat",
|
||||
});
|
||||
await stream.writeSSE({ data: JSON.stringify({ timestamp: Date.now() }), event: "heartbeat" });
|
||||
await stream.sleep(5000);
|
||||
}
|
||||
} finally {
|
||||
|
|
|
|||
|
|
@ -1,29 +1,34 @@
|
|||
import crypto from "node:crypto";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { BadRequestError, ConflictError, InternalServerError, NotFoundError } from "http-errors-enhanced";
|
||||
import { db } from "../../db/db";
|
||||
import { repositoriesTable } from "../../db/schema";
|
||||
import { toMessage } from "../../utils/errors";
|
||||
import { generateShortId } from "../../utils/id";
|
||||
import { restic, buildEnv, buildRepoUrl, addCommonArgs, cleanupTemporaryKeys } from "../../utils/restic";
|
||||
import { safeSpawn } from "../../utils/spawn";
|
||||
import { cryptoUtils } from "../../utils/crypto";
|
||||
import { cache } from "../../utils/cache";
|
||||
import { repoMutex } from "../../core/repository-mutex";
|
||||
import { type } from "arktype";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import {
|
||||
BadRequestError,
|
||||
ConflictError,
|
||||
InternalServerError,
|
||||
NotFoundError,
|
||||
} from "http-errors-enhanced";
|
||||
import {
|
||||
repositoryConfigSchema,
|
||||
type CompressionMode,
|
||||
type OverwriteMode,
|
||||
type RepositoryConfig,
|
||||
repositoryConfigSchema,
|
||||
} from "~/schemas/restic";
|
||||
import { getOrganizationId } from "~/server/core/request-context";
|
||||
import { serverEvents } from "~/server/core/events";
|
||||
import { executeDoctor } from "./doctor";
|
||||
import { getOrganizationId } from "~/server/core/request-context";
|
||||
import { logger } from "~/server/utils/logger";
|
||||
import { parseRetentionCategories, type RetentionCategory } from "~/server/utils/retention-categories";
|
||||
import { repoMutex } from "../../core/repository-mutex";
|
||||
import { db } from "../../db/db";
|
||||
import { repositoriesTable } from "../../db/schema";
|
||||
import { cache } from "../../utils/cache";
|
||||
import { cryptoUtils } from "../../utils/crypto";
|
||||
import { toMessage } from "../../utils/errors";
|
||||
import { generateShortId } from "../../utils/id";
|
||||
import { addCommonArgs, buildEnv, buildRepoUrl, cleanupTemporaryKeys, restic } from "../../utils/restic";
|
||||
import { safeSpawn } from "../../utils/spawn";
|
||||
import { backupsService } from "../backups/backups.service";
|
||||
import type { UpdateRepositoryBody } from "./repositories.dto";
|
||||
import { executeDoctor } from "./doctor";
|
||||
|
||||
const runningDoctors = new Map<string, AbortController>();
|
||||
|
||||
|
|
@ -334,7 +339,31 @@ const restoreSnapshot = async (
|
|||
|
||||
const releaseLock = await repoMutex.acquireShared(repository.id, `restore:${snapshotId}`);
|
||||
try {
|
||||
const result = await restic.restore(repository.config, snapshotId, target, { ...options, organizationId });
|
||||
serverEvents.emit("restore:started", {
|
||||
organizationId,
|
||||
repositoryId: repository.id,
|
||||
snapshotId,
|
||||
});
|
||||
|
||||
const result = await restic.restore(repository.config, snapshotId, target, {
|
||||
...options,
|
||||
organizationId,
|
||||
onProgress: (progress) => {
|
||||
serverEvents.emit("restore:progress", {
|
||||
organizationId,
|
||||
repositoryId: repository.id,
|
||||
snapshotId,
|
||||
...progress,
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
serverEvents.emit("restore:completed", {
|
||||
organizationId,
|
||||
repositoryId: repository.id,
|
||||
snapshotId,
|
||||
status: "success",
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
|
|
@ -342,6 +371,15 @@ const restoreSnapshot = async (
|
|||
filesRestored: result.files_restored,
|
||||
filesSkipped: result.files_skipped,
|
||||
};
|
||||
} catch (error) {
|
||||
serverEvents.emit("restore:completed", {
|
||||
organizationId,
|
||||
repositoryId: repository.id,
|
||||
snapshotId,
|
||||
status: "error",
|
||||
error: toMessage(error),
|
||||
});
|
||||
throw error;
|
||||
} finally {
|
||||
releaseLock();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,28 +1,28 @@
|
|||
import crypto from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import os from "node:os";
|
||||
import { throttle } from "es-toolkit";
|
||||
import path from "node:path";
|
||||
import { type } from "arktype";
|
||||
import { RESTIC_PASS_FILE, DEFAULT_EXCLUDES, RESTIC_CACHE_DIR } from "../core/constants";
|
||||
import { config as appConfig } from "../core/config";
|
||||
import { logger } from "./logger";
|
||||
import { cryptoUtils } from "./crypto";
|
||||
import type { RetentionPolicy } from "../modules/backups/backups.dto";
|
||||
import { safeSpawn, exec } from "./spawn";
|
||||
import type { CompressionMode, RepositoryConfig, OverwriteMode, BandwidthLimit } from "~/schemas/restic";
|
||||
import { throttle } from "es-toolkit";
|
||||
import type { BandwidthLimit, CompressionMode, OverwriteMode, RepositoryConfig } from "~/schemas/restic";
|
||||
import {
|
||||
type ResticBackupProgressDto,
|
||||
type ResticRestoreOutputDto,
|
||||
type ResticSnapshotSummaryDto,
|
||||
resticBackupOutputSchema,
|
||||
resticBackupProgressSchema,
|
||||
resticRestoreOutputSchema,
|
||||
resticSnapshotSummarySchema,
|
||||
type ResticBackupProgressDto,
|
||||
type ResticRestoreOutputDto,
|
||||
type ResticSnapshotSummaryDto,
|
||||
} from "~/schemas/restic-dto";
|
||||
import { ResticError } from "./errors";
|
||||
import { config as appConfig } from "../core/config";
|
||||
import { DEFAULT_EXCLUDES, RESTIC_CACHE_DIR, RESTIC_PASS_FILE } from "../core/constants";
|
||||
import { db } from "../db/db";
|
||||
import type { RetentionPolicy } from "../modules/backups/backups.dto";
|
||||
import { cryptoUtils } from "./crypto";
|
||||
import { ResticError } from "./errors";
|
||||
import { safeJsonParse } from "./json";
|
||||
import { logger } from "./logger";
|
||||
import { exec, safeSpawn } from "./spawn";
|
||||
|
||||
const snapshotInfoSchema = type({
|
||||
gid: "number?",
|
||||
|
|
@ -317,6 +317,8 @@ const backup = async (
|
|||
const progress = resticBackupProgressSchema(jsonData);
|
||||
if (!(progress instanceof type.errors)) {
|
||||
options.onProgress(progress);
|
||||
} else {
|
||||
logger.error(progress.summary);
|
||||
}
|
||||
} catch (_) {
|
||||
// Ignore JSON parse errors for non-JSON lines
|
||||
|
|
@ -379,6 +381,17 @@ const backup = async (
|
|||
return { result, exitCode: res.exitCode };
|
||||
};
|
||||
|
||||
const restoreProgressSchema = type({
|
||||
message_type: "'status' | 'summary'",
|
||||
seconds_elapsed: "number",
|
||||
percent_done: "number = 0",
|
||||
total_files: "number",
|
||||
files_restored: "number = 0",
|
||||
total_bytes: "number = 0",
|
||||
bytes_restored: "number = 0",
|
||||
});
|
||||
|
||||
export type RestoreProgress = typeof restoreProgressSchema.infer;
|
||||
const restore = async (
|
||||
config: RepositoryConfig,
|
||||
snapshotId: string,
|
||||
|
|
@ -390,6 +403,8 @@ const restore = async (
|
|||
excludeXattr?: string[];
|
||||
delete?: boolean;
|
||||
overwrite?: OverwriteMode;
|
||||
onProgress?: (progress: RestoreProgress) => void;
|
||||
signal?: AbortSignal;
|
||||
},
|
||||
) => {
|
||||
const repoUrl = buildRepoUrl(config);
|
||||
|
|
@ -421,8 +436,34 @@ const restore = async (
|
|||
|
||||
addCommonArgs(args, env, config);
|
||||
|
||||
const streamProgress = throttle((data: string) => {
|
||||
if (options?.onProgress) {
|
||||
try {
|
||||
const jsonData = JSON.parse(data);
|
||||
const progress = restoreProgressSchema(jsonData);
|
||||
if (!(progress instanceof type.errors)) {
|
||||
options.onProgress(progress);
|
||||
} else {
|
||||
logger.error(progress.summary);
|
||||
}
|
||||
} catch (_) {
|
||||
// Ignore JSON parse errors for non-JSON lines
|
||||
}
|
||||
}
|
||||
}, 1000);
|
||||
|
||||
logger.debug(`Executing: restic ${args.join(" ")}`);
|
||||
const res = await safeSpawn({ command: "restic", args, env });
|
||||
const res = await safeSpawn({
|
||||
command: "restic",
|
||||
args,
|
||||
env,
|
||||
signal: options?.signal,
|
||||
onStdout: (data) => {
|
||||
if (options?.onProgress) {
|
||||
streamProgress(data);
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
await cleanupTemporaryKeys(env);
|
||||
|
||||
|
|
@ -432,16 +473,15 @@ const restore = async (
|
|||
}
|
||||
|
||||
const lastLine = res.summary.trim();
|
||||
let summaryLine = "";
|
||||
let summaryLine: unknown = {};
|
||||
try {
|
||||
const resSummary = JSON.parse(lastLine ?? "{}");
|
||||
summaryLine = resSummary;
|
||||
summaryLine = JSON.parse(lastLine ?? "{}");
|
||||
} catch (_) {
|
||||
logger.warn("Failed to parse restic backup output JSON summary.", lastLine);
|
||||
summaryLine = "{}";
|
||||
logger.warn("Failed to parse restic restore output JSON summary.", lastLine);
|
||||
summaryLine = {};
|
||||
}
|
||||
|
||||
logger.debug(`Restic restore output last line: ${summaryLine}`);
|
||||
logger.debug(`Restic restore output last line: ${JSON.stringify(summaryLine)}`);
|
||||
const result = resticRestoreOutputSchema(summaryLine);
|
||||
|
||||
if (result instanceof type.errors) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue