refactor: centralize sse event types
This commit is contained in:
Nicolas Meienberger
parent
a91f994966
commit
00ddfe7f30
11 changed files with 284 additions and 746 deletions
|
|
@ -1,4 +1,4 @@
|
|||
import { useCallback, useEffect, useState } from "react";
|
||||
import { useCallback, useEffect, useRef, useState } from "react";
|
||||
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
|
||||
import { ChevronDown, FileIcon, FolderOpen, RotateCcw } from "lucide-react";
|
||||
import { Button } from "~/client/components/ui/button";
|
||||
|
|
@ -20,9 +20,8 @@ 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 { useServerEvents } from "~/client/hooks/use-server-events";
|
||||
import { type RestoreCompletedEvent, useServerEvents } from "~/client/hooks/use-server-events";
|
||||
import { OVERWRITE_MODES, type OverwriteMode } from "~/schemas/restic";
|
||||
import type { RestoreCompletedEventDto, RestoreProgressEventDto } from "~/schemas/events-dto";
|
||||
import type { Repository, Snapshot } from "~/client/lib/types";
|
||||
import { handleRepositoryError } from "~/client/lib/errors";
|
||||
import { useNavigate } from "@tanstack/react-router";
|
||||
|
|
@ -49,34 +48,34 @@ export function RestoreForm({ snapshot, repository, snapshotId, returnPath }: Re
|
|||
const [showAdvanced, setShowAdvanced] = useState(false);
|
||||
const [excludeXattr, setExcludeXattr] = useState("");
|
||||
const [isRestoreActive, setIsRestoreActive] = useState(false);
|
||||
const [restoreResult, setRestoreResult] = useState<RestoreCompletedEventDto | null>(null);
|
||||
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", (data) => {
|
||||
const startedData = data as {
|
||||
repositoryId: string;
|
||||
snapshotId: string;
|
||||
};
|
||||
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", (data) => {
|
||||
const progressData = data as RestoreProgressEventDto;
|
||||
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", (data) => {
|
||||
const completedData = data as RestoreCompletedEventDto;
|
||||
const unsubscribeCompleted = addEventListener("restore:completed", (completedData) => {
|
||||
if (completedData.repositoryId === repository.id && completedData.snapshotId === snapshotId) {
|
||||
restoreCompletedRef.current = true;
|
||||
setIsRestoreActive(false);
|
||||
setRestoreResult(completedData);
|
||||
setShowRestoreResultAlert(true);
|
||||
|
|
@ -149,6 +148,7 @@ export function RestoreForm({ snapshot, repository, snapshotId, returnPath }: Re
|
|||
const { mutate: restoreSnapshot, isPending: isRestoring } = useMutation({
|
||||
...restoreSnapshotMutation(),
|
||||
onError: (error) => {
|
||||
restoreCompletedRef.current = true;
|
||||
setIsRestoreActive(false);
|
||||
handleRepositoryError("Restore failed", error, repository.id);
|
||||
},
|
||||
|
|
@ -166,6 +166,7 @@ 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);
|
||||
|
|
|
|||
|
|
@ -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 { RestoreCompletedEventDto, RestoreProgressEventDto } from "~/schemas/events-dto";
|
||||
import { type RestoreProgressEvent, useServerEvents } from "~/client/hooks/use-server-events";
|
||||
import { formatBytes } from "~/utils/format-bytes";
|
||||
import { formatDuration } from "~/utils/utils";
|
||||
|
||||
|
|
@ -14,18 +13,16 @@ type Props = {
|
|||
|
||||
export const RestoreProgress = ({ repositoryId, snapshotId }: Props) => {
|
||||
const { addEventListener } = useServerEvents();
|
||||
const [progress, setProgress] = useState<RestoreProgressEventDto | null>(null);
|
||||
const [progress, setProgress] = useState<RestoreProgressEvent | null>(null);
|
||||
|
||||
useEffect(() => {
|
||||
const unsubscribe = addEventListener("restore:progress", (data) => {
|
||||
const progressData = data as RestoreProgressEventDto;
|
||||
const unsubscribe = addEventListener("restore:progress", (progressData) => {
|
||||
if (progressData.repositoryId === repositoryId && progressData.snapshotId === snapshotId) {
|
||||
setProgress(progressData);
|
||||
}
|
||||
});
|
||||
|
||||
const unsubscribeComplete = addEventListener("restore:completed", (data) => {
|
||||
const completedData = data as RestoreCompletedEventDto;
|
||||
const unsubscribeComplete = addEventListener("restore:completed", (completedData) => {
|
||||
if (completedData.repositoryId === repositoryId && completedData.snapshotId === snapshotId) {
|
||||
setProgress(null);
|
||||
}
|
||||
|
|
@ -49,7 +46,7 @@ export const RestoreProgress = ({ repositoryId, snapshotId }: Props) => {
|
|||
}
|
||||
|
||||
const percentDone = Math.round(progress.percent_done * 100);
|
||||
const speed = progress.seconds_elapsed > 0 ? formatBytes(progress.bytes_done / progress.seconds_elapsed) : null;
|
||||
const speed = progress.seconds_elapsed > 0 ? formatBytes(progress.bytes_restored / progress.seconds_elapsed) : null;
|
||||
|
||||
return (
|
||||
<Card className="p-4">
|
||||
|
|
@ -67,13 +64,17 @@ export const RestoreProgress = ({ repositoryId, snapshotId }: Props) => {
|
|||
<div>
|
||||
<p className="text-xs uppercase text-muted-foreground">Files</p>
|
||||
<p className="font-medium">
|
||||
{progress.files_done.toLocaleString()} / {progress.total_files.toLocaleString()}
|
||||
{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_done} /> / <ByteSize bytes={progress.total_bytes} />
|
||||
<ByteSize bytes={progress.bytes_restored} base={1024} />
|
||||
/
|
||||
<ByteSize bytes={progress.total_bytes} base={1024} />
|
||||
</p>
|
||||
</div>
|
||||
<div>
|
||||
|
|
|
|||
|
|
@ -1,62 +1,38 @@
|
|||
import { useQueryClient } from "@tanstack/react-query";
|
||||
import { useCallback, useEffect, useRef } from "react";
|
||||
import type {
|
||||
BackupCompletedEventDto,
|
||||
BackupProgressEventDto,
|
||||
BackupStartedEventDto,
|
||||
RestoreCompletedEventDto,
|
||||
RestoreProgressEventDto,
|
||||
RestoreStartedEventDto,
|
||||
} from "~/schemas/events-dto";
|
||||
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"
|
||||
| "restore:started"
|
||||
| "restore:progress"
|
||||
| "restore: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
|
||||
|
|
@ -65,166 +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("restore:started", (e) => {
|
||||
const data = JSON.parse(e.data) as RestoreStartedEventDto;
|
||||
console.info("[SSE] Restore started:", data);
|
||||
|
||||
handlersRef.current.get("restore:started")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("restore:progress", (e) => {
|
||||
const data = JSON.parse(e.data) as RestoreProgressEventDto;
|
||||
|
||||
handlersRef.current.get("restore:progress")?.forEach((handler) => {
|
||||
handler(data);
|
||||
});
|
||||
});
|
||||
|
||||
eventSource.addEventListener("restore:completed", (e) => {
|
||||
const data = JSON.parse(e.data) as RestoreCompletedEventDto;
|
||||
console.info("[SSE] Restore completed:", data);
|
||||
|
||||
handlersRef.current.get("restore: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);
|
||||
|
|
@ -235,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);
|
||||
|
|
|
|||
|
|
@ -23,9 +23,9 @@ const restoreProgressMetricsSchema = type({
|
|||
seconds_elapsed: "number",
|
||||
percent_done: "number",
|
||||
total_files: "number",
|
||||
files_done: "number",
|
||||
files_restored: "number",
|
||||
total_bytes: "number",
|
||||
bytes_done: "number",
|
||||
bytes_restored: "number",
|
||||
});
|
||||
|
||||
export const backupStartedEventSchema = backupEventBaseSchema;
|
||||
|
|
@ -44,10 +44,7 @@ export const restoreStartedEventSchema = restoreEventBaseSchema;
|
|||
export const restoreProgressEventSchema = restoreEventBaseSchema.and(restoreProgressMetricsSchema);
|
||||
|
||||
export const restoreCompletedEventSchema = restoreEventBaseSchema.and(
|
||||
type({
|
||||
status: restoreEventStatusSchema,
|
||||
error: "string?",
|
||||
}),
|
||||
type({ status: restoreEventStatusSchema, error: "string?" }),
|
||||
);
|
||||
|
||||
export const serverBackupStartedEventSchema = organizationScopedSchema.and(backupStartedEventSchema);
|
||||
|
|
|
|||
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,61 +1,10 @@
|
|||
import { EventEmitter } from "node:events";
|
||||
import type { TypedEmitter } from "tiny-typed-emitter";
|
||||
import type {
|
||||
ServerBackupCompletedEventDto,
|
||||
ServerBackupProgressEventDto,
|
||||
ServerBackupStartedEventDto,
|
||||
ServerRestoreCompletedEventDto,
|
||||
ServerRestoreProgressEventDto,
|
||||
ServerRestoreStartedEventDto,
|
||||
} from "~/schemas/events-dto";
|
||||
import type { DoctorResult } from "~/schemas/restic";
|
||||
|
||||
/**
|
||||
* Event payloads for the SSE system
|
||||
*/
|
||||
interface ServerEvents {
|
||||
"backup:started": (data: ServerBackupStartedEventDto) => void;
|
||||
"backup:progress": (data: ServerBackupProgressEventDto) => void;
|
||||
"backup:completed": (data: ServerBackupCompletedEventDto) => void;
|
||||
"restore:started": (data: ServerRestoreStartedEventDto) => void;
|
||||
"restore:progress": (data: ServerRestoreProgressEventDto) => void;
|
||||
"restore:completed": (data: ServerRestoreCompletedEventDto) => 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>;
|
||||
|
|
|
|||
|
|
@ -143,8 +143,6 @@ export const repositoriesController = new Hono()
|
|||
summary: snapshot.summary,
|
||||
};
|
||||
|
||||
c.header("Cache-Control", "max-age=300, stale-while-revalidate=600");
|
||||
|
||||
return c.json<GetSnapshotDetailsDto>(response, 200);
|
||||
})
|
||||
.get(
|
||||
|
|
|
|||
|
|
@ -16,10 +16,7 @@ import {
|
|||
import { serverEvents } from "~/server/core/events";
|
||||
import { getOrganizationId } from "~/server/core/request-context";
|
||||
import { logger } from "~/server/utils/logger";
|
||||
import {
|
||||
parseRetentionCategories,
|
||||
type RetentionCategory,
|
||||
} from "~/server/utils/retention-categories";
|
||||
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";
|
||||
|
|
@ -27,13 +24,7 @@ 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 { 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";
|
||||
|
|
@ -45,31 +36,22 @@ const findRepository = async (idOrShortId: string) => {
|
|||
const organizationId = getOrganizationId();
|
||||
return await db.query.repositoriesTable.findFirst({
|
||||
where: {
|
||||
AND: [
|
||||
{ OR: [{ id: idOrShortId }, { shortId: idOrShortId }] },
|
||||
{ organizationId },
|
||||
],
|
||||
AND: [{ OR: [{ id: idOrShortId }, { shortId: idOrShortId }] }, { organizationId }],
|
||||
},
|
||||
});
|
||||
};
|
||||
|
||||
const listRepositories = async () => {
|
||||
const organizationId = getOrganizationId();
|
||||
const repositories = await db.query.repositoriesTable.findMany({
|
||||
where: { organizationId },
|
||||
});
|
||||
const repositories = await db.query.repositoriesTable.findMany({ where: { organizationId } });
|
||||
return repositories;
|
||||
};
|
||||
|
||||
const encryptConfig = async (
|
||||
config: RepositoryConfig,
|
||||
): Promise<RepositoryConfig> => {
|
||||
const encryptConfig = async (config: RepositoryConfig): Promise<RepositoryConfig> => {
|
||||
const encryptedConfig: Record<string, unknown> = { ...config };
|
||||
|
||||
if (config.customPassword) {
|
||||
encryptedConfig.customPassword = await cryptoUtils.sealSecret(
|
||||
config.customPassword,
|
||||
);
|
||||
encryptedConfig.customPassword = await cryptoUtils.sealSecret(config.customPassword);
|
||||
}
|
||||
|
||||
if (config.cacert) {
|
||||
|
|
@ -79,39 +61,25 @@ const encryptConfig = async (
|
|||
switch (config.backend) {
|
||||
case "s3":
|
||||
case "r2":
|
||||
encryptedConfig.accessKeyId = await cryptoUtils.sealSecret(
|
||||
config.accessKeyId,
|
||||
);
|
||||
encryptedConfig.secretAccessKey = await cryptoUtils.sealSecret(
|
||||
config.secretAccessKey,
|
||||
);
|
||||
encryptedConfig.accessKeyId = await cryptoUtils.sealSecret(config.accessKeyId);
|
||||
encryptedConfig.secretAccessKey = await cryptoUtils.sealSecret(config.secretAccessKey);
|
||||
break;
|
||||
case "gcs":
|
||||
encryptedConfig.credentialsJson = await cryptoUtils.sealSecret(
|
||||
config.credentialsJson,
|
||||
);
|
||||
encryptedConfig.credentialsJson = await cryptoUtils.sealSecret(config.credentialsJson);
|
||||
break;
|
||||
case "azure":
|
||||
encryptedConfig.accountKey = await cryptoUtils.sealSecret(
|
||||
config.accountKey,
|
||||
);
|
||||
encryptedConfig.accountKey = await cryptoUtils.sealSecret(config.accountKey);
|
||||
break;
|
||||
case "rest":
|
||||
if (config.username) {
|
||||
encryptedConfig.username = await cryptoUtils.sealSecret(
|
||||
config.username,
|
||||
);
|
||||
encryptedConfig.username = await cryptoUtils.sealSecret(config.username);
|
||||
}
|
||||
if (config.password) {
|
||||
encryptedConfig.password = await cryptoUtils.sealSecret(
|
||||
config.password,
|
||||
);
|
||||
encryptedConfig.password = await cryptoUtils.sealSecret(config.password);
|
||||
}
|
||||
break;
|
||||
case "sftp":
|
||||
encryptedConfig.privateKey = await cryptoUtils.sealSecret(
|
||||
config.privateKey,
|
||||
);
|
||||
encryptedConfig.privateKey = await cryptoUtils.sealSecret(config.privateKey);
|
||||
break;
|
||||
}
|
||||
|
||||
|
|
@ -193,9 +161,7 @@ const createRepository = async (name: string, config: RepositoryConfig, compress
|
|||
|
||||
error = result.error;
|
||||
} else {
|
||||
const initResult = await restic.init(encryptedConfig, organizationId, {
|
||||
timeoutMs: 20000,
|
||||
});
|
||||
const initResult = await restic.init(encryptedConfig, organizationId, { timeoutMs: 20000 });
|
||||
error = initResult.error;
|
||||
}
|
||||
|
||||
|
|
@ -203,12 +169,7 @@ const createRepository = async (name: string, config: RepositoryConfig, compress
|
|||
await db
|
||||
.update(repositoriesTable)
|
||||
.set({ status: "healthy", lastChecked: Date.now(), lastError: null })
|
||||
.where(
|
||||
and(
|
||||
eq(repositoriesTable.id, id),
|
||||
eq(repositoriesTable.organizationId, organizationId),
|
||||
),
|
||||
);
|
||||
.where(and(eq(repositoriesTable.id, id), eq(repositoriesTable.organizationId, organizationId)));
|
||||
|
||||
return { repository: created, status: 201 };
|
||||
}
|
||||
|
|
@ -216,16 +177,9 @@ const createRepository = async (name: string, config: RepositoryConfig, compress
|
|||
const errorMessage = toMessage(error);
|
||||
await db
|
||||
.delete(repositoriesTable)
|
||||
.where(
|
||||
and(
|
||||
eq(repositoriesTable.id, id),
|
||||
eq(repositoriesTable.organizationId, organizationId),
|
||||
),
|
||||
);
|
||||
.where(and(eq(repositoriesTable.id, id), eq(repositoriesTable.organizationId, organizationId)));
|
||||
|
||||
throw new InternalServerError(
|
||||
`Failed to initialize repository: ${errorMessage}`,
|
||||
);
|
||||
throw new InternalServerError(`Failed to initialize repository: ${errorMessage}`);
|
||||
};
|
||||
|
||||
const getRepository = async (id: string) => {
|
||||
|
|
@ -250,10 +204,7 @@ const deleteRepository = async (id: string) => {
|
|||
await db
|
||||
.delete(repositoriesTable)
|
||||
.where(
|
||||
and(
|
||||
eq(repositoriesTable.id, repository.id),
|
||||
eq(repositoriesTable.organizationId, repository.organizationId),
|
||||
),
|
||||
and(eq(repositoriesTable.id, repository.id), eq(repositoriesTable.organizationId, repository.organizationId)),
|
||||
);
|
||||
|
||||
cache.delByPrefix(`snapshots:${repository.id}:`);
|
||||
|
|
@ -277,8 +228,7 @@ const listSnapshots = async (id: string, backupId?: string) => {
|
|||
}
|
||||
|
||||
const cacheKey = `snapshots:${repository.id}:${backupId || "all"}`;
|
||||
const cached =
|
||||
cache.get<Awaited<ReturnType<typeof restic.snapshots>>>(cacheKey);
|
||||
const cached = cache.get<Awaited<ReturnType<typeof restic.snapshots>>>(cacheKey);
|
||||
if (cached) {
|
||||
return cached;
|
||||
}
|
||||
|
|
@ -288,10 +238,7 @@ const listSnapshots = async (id: string, backupId?: string) => {
|
|||
let snapshots = [];
|
||||
|
||||
if (backupId) {
|
||||
snapshots = await restic.snapshots(repository.config, {
|
||||
tags: [backupId],
|
||||
organizationId,
|
||||
});
|
||||
snapshots = await restic.snapshots(repository.config, { tags: [backupId], organizationId });
|
||||
} else {
|
||||
snapshots = await restic.snapshots(repository.config, { organizationId });
|
||||
}
|
||||
|
|
@ -322,26 +269,9 @@ const listSnapshotFiles = async (
|
|||
|
||||
const cacheKey = `ls:${repository.id}:${snapshotId}:${path || "root"}:${offset}:${limit}`;
|
||||
type LsResult = {
|
||||
snapshot: {
|
||||
id: string;
|
||||
short_id: string;
|
||||
time: string;
|
||||
hostname: string;
|
||||
paths: string[];
|
||||
} | null;
|
||||
nodes: {
|
||||
name: string;
|
||||
type: string;
|
||||
path: string;
|
||||
size?: number;
|
||||
mode?: number;
|
||||
}[];
|
||||
pagination: {
|
||||
offset: number;
|
||||
limit: number;
|
||||
total: number;
|
||||
hasMore: boolean;
|
||||
};
|
||||
snapshot: { id: string; short_id: string; time: string; hostname: string; paths: string[] } | null;
|
||||
nodes: { name: string; type: string; path: string; size?: number; mode?: number }[];
|
||||
pagination: { offset: number; limit: number; total: number; hasMore: boolean };
|
||||
};
|
||||
const cached = cache.get<LsResult>(cacheKey);
|
||||
if (cached?.snapshot) {
|
||||
|
|
@ -355,18 +285,9 @@ const listSnapshotFiles = async (
|
|||
};
|
||||
}
|
||||
|
||||
const releaseLock = await repoMutex.acquireShared(
|
||||
repository.id,
|
||||
`ls:${snapshotId}`,
|
||||
);
|
||||
const releaseLock = await repoMutex.acquireShared(repository.id, `ls:${snapshotId}`);
|
||||
try {
|
||||
const result = await restic.ls(
|
||||
repository.config,
|
||||
snapshotId,
|
||||
organizationId,
|
||||
path,
|
||||
{ offset, limit },
|
||||
);
|
||||
const result = await restic.ls(repository.config, snapshotId, organizationId, path, { offset, limit });
|
||||
|
||||
if (!result.snapshot) {
|
||||
throw new NotFoundError("Snapshot not found or empty");
|
||||
|
|
@ -416,10 +337,7 @@ const restoreSnapshot = async (
|
|||
|
||||
const target = options?.targetPath || "/";
|
||||
|
||||
const releaseLock = await repoMutex.acquireShared(
|
||||
repository.id,
|
||||
`restore:${snapshotId}`,
|
||||
);
|
||||
const releaseLock = await repoMutex.acquireShared(repository.id, `restore:${snapshotId}`);
|
||||
try {
|
||||
serverEvents.emit("restore:started", {
|
||||
organizationId,
|
||||
|
|
@ -431,25 +349,11 @@ const restoreSnapshot = async (
|
|||
...options,
|
||||
organizationId,
|
||||
onProgress: (progress) => {
|
||||
const {
|
||||
seconds_elapsed,
|
||||
percent_done,
|
||||
total_files,
|
||||
files_done,
|
||||
total_bytes,
|
||||
bytes_done,
|
||||
} = progress;
|
||||
|
||||
serverEvents.emit("restore:progress", {
|
||||
organizationId,
|
||||
repositoryId: repository.id,
|
||||
snapshotId,
|
||||
seconds_elapsed,
|
||||
percent_done,
|
||||
total_files,
|
||||
files_done,
|
||||
total_bytes,
|
||||
bytes_done,
|
||||
...progress,
|
||||
});
|
||||
},
|
||||
});
|
||||
|
|
@ -490,14 +394,10 @@ const getSnapshotDetails = async (id: string, snapshotId: string) => {
|
|||
}
|
||||
|
||||
const cacheKey = `snapshots:${repository.id}:all`;
|
||||
let snapshots =
|
||||
cache.get<Awaited<ReturnType<typeof restic.snapshots>>>(cacheKey);
|
||||
let snapshots = cache.get<Awaited<ReturnType<typeof restic.snapshots>>>(cacheKey);
|
||||
|
||||
if (!snapshots) {
|
||||
const releaseLock = await repoMutex.acquireShared(
|
||||
repository.id,
|
||||
`snapshot_details:${snapshotId}`,
|
||||
);
|
||||
const releaseLock = await repoMutex.acquireShared(repository.id, `snapshot_details:${snapshotId}`);
|
||||
try {
|
||||
snapshots = await restic.snapshots(repository.config, { organizationId });
|
||||
cache.set(cacheKey, snapshots);
|
||||
|
|
@ -506,9 +406,7 @@ const getSnapshotDetails = async (id: string, snapshotId: string) => {
|
|||
}
|
||||
}
|
||||
|
||||
const snapshot = snapshots.find(
|
||||
(snap) => snap.id === snapshotId || snap.short_id === snapshotId,
|
||||
);
|
||||
const snapshot = snapshots.find((snap) => snap.id === snapshotId || snap.short_id === snapshotId);
|
||||
|
||||
if (!snapshot) {
|
||||
void refreshSnapshots(id).catch(() => {});
|
||||
|
|
@ -529,9 +427,7 @@ const checkHealth = async (repositoryId: string) => {
|
|||
|
||||
const releaseLock = await repoMutex.acquireExclusive(repository.id, "check");
|
||||
try {
|
||||
const { hasErrors, error } = await restic.check(repository.config, {
|
||||
organizationId,
|
||||
});
|
||||
const { hasErrors, error } = await restic.check(repository.config, { organizationId });
|
||||
|
||||
await db
|
||||
.update(repositoriesTable)
|
||||
|
|
@ -541,10 +437,7 @@ const checkHealth = async (repositoryId: string) => {
|
|||
lastError: error,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(repositoriesTable.id, repository.id),
|
||||
eq(repositoriesTable.organizationId, repository.organizationId),
|
||||
),
|
||||
and(eq(repositoriesTable.id, repository.id), eq(repositoriesTable.organizationId, repository.organizationId)),
|
||||
);
|
||||
|
||||
return { lastError: error };
|
||||
|
|
@ -567,10 +460,7 @@ const startDoctor = async (id: string) => {
|
|||
const abortController = new AbortController();
|
||||
|
||||
try {
|
||||
await db
|
||||
.update(repositoriesTable)
|
||||
.set({ status: "doctor" })
|
||||
.where(eq(repositoriesTable.id, repository.id));
|
||||
await db.update(repositoriesTable).set({ status: "doctor" }).where(eq(repositoriesTable.id, repository.id));
|
||||
|
||||
serverEvents.emit("doctor:started", {
|
||||
organizationId: repository.organizationId,
|
||||
|
|
@ -584,12 +474,7 @@ const startDoctor = async (id: string) => {
|
|||
throw error;
|
||||
}
|
||||
|
||||
executeDoctor(
|
||||
repository.id,
|
||||
repository.config,
|
||||
repository.name,
|
||||
abortController.signal,
|
||||
)
|
||||
executeDoctor(repository.id, repository.config, repository.name, abortController.signal)
|
||||
.catch((error) => {
|
||||
logger.error(`Doctor background task failed: ${toMessage(error)}`);
|
||||
})
|
||||
|
|
@ -609,20 +494,14 @@ const cancelDoctor = async (id: string) => {
|
|||
|
||||
const abortController = runningDoctors.get(repository.id);
|
||||
if (!abortController) {
|
||||
await db
|
||||
.update(repositoriesTable)
|
||||
.set({ status: "unknown" })
|
||||
.where(eq(repositoriesTable.id, repository.id));
|
||||
await db.update(repositoriesTable).set({ status: "unknown" }).where(eq(repositoriesTable.id, repository.id));
|
||||
throw new ConflictError("No doctor operation is currently running");
|
||||
}
|
||||
|
||||
abortController.abort();
|
||||
runningDoctors.delete(repository.id);
|
||||
|
||||
await db
|
||||
.update(repositoriesTable)
|
||||
.set({ status: "unknown" })
|
||||
.where(eq(repositoriesTable.id, repository.id));
|
||||
await db.update(repositoriesTable).set({ status: "unknown" }).where(eq(repositoriesTable.id, repository.id));
|
||||
|
||||
serverEvents.emit("doctor:cancelled", {
|
||||
organizationId: repository.organizationId,
|
||||
|
|
@ -641,10 +520,7 @@ const deleteSnapshot = async (id: string, snapshotId: string) => {
|
|||
throw new NotFoundError("Repository not found");
|
||||
}
|
||||
|
||||
const releaseLock = await repoMutex.acquireExclusive(
|
||||
repository.id,
|
||||
`delete:${snapshotId}`,
|
||||
);
|
||||
const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:${snapshotId}`);
|
||||
try {
|
||||
await restic.deleteSnapshot(repository.config, snapshotId, organizationId);
|
||||
cache.delByPrefix(`snapshots:${repository.id}:`);
|
||||
|
|
@ -662,16 +538,9 @@ const deleteSnapshots = async (id: string, snapshotIds: string[]) => {
|
|||
throw new NotFoundError("Repository not found");
|
||||
}
|
||||
|
||||
const releaseLock = await repoMutex.acquireExclusive(
|
||||
repository.id,
|
||||
`delete:bulk`,
|
||||
);
|
||||
const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:bulk`);
|
||||
try {
|
||||
await restic.deleteSnapshots(
|
||||
repository.config,
|
||||
snapshotIds,
|
||||
organizationId,
|
||||
);
|
||||
await restic.deleteSnapshots(repository.config, snapshotIds, organizationId);
|
||||
cache.delByPrefix(`snapshots:${repository.id}:`);
|
||||
for (const snapshotId of snapshotIds) {
|
||||
cache.delByPrefix(`ls:${repository.id}:${snapshotId}:`);
|
||||
|
|
@ -693,17 +562,9 @@ const tagSnapshots = async (
|
|||
throw new NotFoundError("Repository not found");
|
||||
}
|
||||
|
||||
const releaseLock = await repoMutex.acquireExclusive(
|
||||
repository.id,
|
||||
`tag:bulk`,
|
||||
);
|
||||
const releaseLock = await repoMutex.acquireExclusive(repository.id, `tag:bulk`);
|
||||
try {
|
||||
await restic.tagSnapshots(
|
||||
repository.config,
|
||||
snapshotIds,
|
||||
tags,
|
||||
organizationId,
|
||||
);
|
||||
await restic.tagSnapshots(repository.config, snapshotIds, tags, organizationId);
|
||||
cache.delByPrefix(`snapshots:${repository.id}:`);
|
||||
for (const snapshotId of snapshotIds) {
|
||||
cache.delByPrefix(`ls:${repository.id}:${snapshotId}:`);
|
||||
|
|
@ -726,9 +587,7 @@ const refreshSnapshots = async (id: string) => {
|
|||
|
||||
const releaseLock = await repoMutex.acquireShared(repository.id, "refresh");
|
||||
try {
|
||||
const snapshots = await restic.snapshots(repository.config, {
|
||||
organizationId,
|
||||
});
|
||||
const snapshots = await restic.snapshots(repository.config, { organizationId });
|
||||
const cacheKey = `snapshots:${repository.id}:all`;
|
||||
cache.set(cacheKey, snapshots);
|
||||
|
||||
|
|
@ -855,24 +714,14 @@ const execResticCommand = async (
|
|||
}
|
||||
addCommonArgs(resticArgs, env, repository.config);
|
||||
|
||||
const result = await safeSpawn({
|
||||
command: "restic",
|
||||
args: resticArgs,
|
||||
env,
|
||||
signal,
|
||||
onStdout,
|
||||
onStderr,
|
||||
});
|
||||
const result = await safeSpawn({ command: "restic", args: resticArgs, env, signal, onStdout, onStderr });
|
||||
|
||||
await cleanupTemporaryKeys(env);
|
||||
|
||||
return { exitCode: result.exitCode };
|
||||
};
|
||||
|
||||
const getRetentionCategories = async (
|
||||
repositoryId: string,
|
||||
scheduleId?: string,
|
||||
) => {
|
||||
const getRetentionCategories = async (repositoryId: string, scheduleId?: string) => {
|
||||
if (!scheduleId) {
|
||||
return new Map<string, RetentionCategory[]>();
|
||||
}
|
||||
|
|
@ -896,15 +745,11 @@ const getRetentionCategories = async (
|
|||
|
||||
const { repository } = repositoryResult;
|
||||
|
||||
const dryRunResults = await restic.forget(
|
||||
repository.config,
|
||||
schedule.retentionPolicy,
|
||||
{
|
||||
tag: scheduleId,
|
||||
organizationId: getOrganizationId(),
|
||||
dryRun: true,
|
||||
},
|
||||
);
|
||||
const dryRunResults = await restic.forget(repository.config, schedule.retentionPolicy, {
|
||||
tag: scheduleId,
|
||||
organizationId: getOrganizationId(),
|
||||
dryRun: true,
|
||||
});
|
||||
|
||||
if (!dryRunResults.data) {
|
||||
return new Map<string, RetentionCategory[]>();
|
||||
|
|
|
|||
|
|
@ -4,12 +4,7 @@ import os from "node:os";
|
|||
import path from "node:path";
|
||||
import { type } from "arktype";
|
||||
import { throttle } from "es-toolkit";
|
||||
import type {
|
||||
BandwidthLimit,
|
||||
CompressionMode,
|
||||
OverwriteMode,
|
||||
RepositoryConfig,
|
||||
} from "~/schemas/restic";
|
||||
import type { BandwidthLimit, CompressionMode, OverwriteMode, RepositoryConfig } from "~/schemas/restic";
|
||||
import {
|
||||
type ResticBackupProgressDto,
|
||||
type ResticRestoreOutputDto,
|
||||
|
|
@ -20,12 +15,7 @@ import {
|
|||
resticSnapshotSummarySchema,
|
||||
} from "~/schemas/restic-dto";
|
||||
import { config as appConfig } from "../core/config";
|
||||
import {
|
||||
DEFAULT_EXCLUDES,
|
||||
REPOSITORY_BASE,
|
||||
RESTIC_CACHE_DIR,
|
||||
RESTIC_PASS_FILE,
|
||||
} from "../core/constants";
|
||||
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";
|
||||
|
|
@ -72,37 +62,25 @@ export const buildRepoUrl = (config: RepositoryConfig): string => {
|
|||
case "sftp":
|
||||
return `sftp:${config.user}@${config.host}:${config.path}`;
|
||||
default: {
|
||||
throw new Error(
|
||||
`Unsupported repository backend: ${JSON.stringify(config)}`,
|
||||
);
|
||||
throw new Error(`Unsupported repository backend: ${JSON.stringify(config)}`);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
export const buildEnv = async (
|
||||
config: RepositoryConfig,
|
||||
organizationId: string,
|
||||
) => {
|
||||
export const buildEnv = async (config: RepositoryConfig, organizationId: string) => {
|
||||
const env: Record<string, string> = {
|
||||
RESTIC_CACHE_DIR,
|
||||
PATH: process.env.PATH || "/usr/local/bin:/usr/bin:/bin",
|
||||
};
|
||||
|
||||
if (config.isExistingRepository && config.customPassword) {
|
||||
const decryptedPassword = await cryptoUtils.resolveSecret(
|
||||
config.customPassword,
|
||||
);
|
||||
const passwordFilePath = path.join(
|
||||
"/tmp",
|
||||
`zerobyte-pass-${crypto.randomBytes(8).toString("hex")}.txt`,
|
||||
);
|
||||
const decryptedPassword = await cryptoUtils.resolveSecret(config.customPassword);
|
||||
const passwordFilePath = path.join("/tmp", `zerobyte-pass-${crypto.randomBytes(8).toString("hex")}.txt`);
|
||||
|
||||
await fs.writeFile(passwordFilePath, decryptedPassword, { mode: 0o600 });
|
||||
env.RESTIC_PASSWORD_FILE = passwordFilePath;
|
||||
} else {
|
||||
const org = await db.query.organization.findFirst({
|
||||
where: { id: organizationId },
|
||||
});
|
||||
const org = await db.query.organization.findFirst({ where: { id: organizationId } });
|
||||
|
||||
if (!org) {
|
||||
throw new Error(`Organization ${organizationId} not found`);
|
||||
|
|
@ -110,17 +88,10 @@ export const buildEnv = async (
|
|||
|
||||
const metadata = org.metadata;
|
||||
if (!metadata?.resticPassword) {
|
||||
throw new Error(
|
||||
`Restic password not configured for organization ${organizationId}`,
|
||||
);
|
||||
throw new Error(`Restic password not configured for organization ${organizationId}`);
|
||||
} else {
|
||||
const decryptedPassword = await cryptoUtils.resolveSecret(
|
||||
metadata.resticPassword,
|
||||
);
|
||||
const passwordFilePath = path.join(
|
||||
"/tmp",
|
||||
`zerobyte-pass-${crypto.randomBytes(8).toString("hex")}.txt`,
|
||||
);
|
||||
const decryptedPassword = await cryptoUtils.resolveSecret(metadata.resticPassword);
|
||||
const passwordFilePath = path.join("/tmp", `zerobyte-pass-${crypto.randomBytes(8).toString("hex")}.txt`);
|
||||
await fs.writeFile(passwordFilePath, decryptedPassword, { mode: 0o600 });
|
||||
env.RESTIC_PASSWORD_FILE = passwordFilePath;
|
||||
}
|
||||
|
|
@ -128,12 +99,8 @@ export const buildEnv = async (
|
|||
|
||||
switch (config.backend) {
|
||||
case "s3":
|
||||
env.AWS_ACCESS_KEY_ID = await cryptoUtils.resolveSecret(
|
||||
config.accessKeyId,
|
||||
);
|
||||
env.AWS_SECRET_ACCESS_KEY = await cryptoUtils.resolveSecret(
|
||||
config.secretAccessKey,
|
||||
);
|
||||
env.AWS_ACCESS_KEY_ID = await cryptoUtils.resolveSecret(config.accessKeyId);
|
||||
env.AWS_SECRET_ACCESS_KEY = await cryptoUtils.resolveSecret(config.secretAccessKey);
|
||||
|
||||
// Huawei OBS requires virtual-hosted style access
|
||||
if (config.endpoint.includes("myhuaweicloud")) {
|
||||
|
|
@ -141,35 +108,22 @@ export const buildEnv = async (
|
|||
}
|
||||
break;
|
||||
case "r2":
|
||||
env.AWS_ACCESS_KEY_ID = await cryptoUtils.resolveSecret(
|
||||
config.accessKeyId,
|
||||
);
|
||||
env.AWS_SECRET_ACCESS_KEY = await cryptoUtils.resolveSecret(
|
||||
config.secretAccessKey,
|
||||
);
|
||||
env.AWS_ACCESS_KEY_ID = await cryptoUtils.resolveSecret(config.accessKeyId);
|
||||
env.AWS_SECRET_ACCESS_KEY = await cryptoUtils.resolveSecret(config.secretAccessKey);
|
||||
env.AWS_REGION = "auto";
|
||||
env.AWS_S3_FORCE_PATH_STYLE = "true";
|
||||
break;
|
||||
case "gcs": {
|
||||
const decryptedCredentials = await cryptoUtils.resolveSecret(
|
||||
config.credentialsJson,
|
||||
);
|
||||
const credentialsPath = path.join(
|
||||
"/tmp",
|
||||
`zerobyte-gcs-${crypto.randomBytes(8).toString("hex")}.json`,
|
||||
);
|
||||
await fs.writeFile(credentialsPath, decryptedCredentials, {
|
||||
mode: 0o600,
|
||||
});
|
||||
const decryptedCredentials = await cryptoUtils.resolveSecret(config.credentialsJson);
|
||||
const credentialsPath = path.join("/tmp", `zerobyte-gcs-${crypto.randomBytes(8).toString("hex")}.json`);
|
||||
await fs.writeFile(credentialsPath, decryptedCredentials, { mode: 0o600 });
|
||||
env.GOOGLE_PROJECT_ID = config.projectId;
|
||||
env.GOOGLE_APPLICATION_CREDENTIALS = credentialsPath;
|
||||
break;
|
||||
}
|
||||
case "azure": {
|
||||
env.AZURE_ACCOUNT_NAME = config.accountName;
|
||||
env.AZURE_ACCOUNT_KEY = await cryptoUtils.resolveSecret(
|
||||
config.accountKey,
|
||||
);
|
||||
env.AZURE_ACCOUNT_KEY = await cryptoUtils.resolveSecret(config.accountKey);
|
||||
if (config.endpointSuffix) {
|
||||
env.AZURE_ENDPOINT_SUFFIX = config.endpointSuffix;
|
||||
}
|
||||
|
|
@ -177,23 +131,16 @@ export const buildEnv = async (
|
|||
}
|
||||
case "rest": {
|
||||
if (config.username) {
|
||||
env.RESTIC_REST_USERNAME = await cryptoUtils.resolveSecret(
|
||||
config.username,
|
||||
);
|
||||
env.RESTIC_REST_USERNAME = await cryptoUtils.resolveSecret(config.username);
|
||||
}
|
||||
if (config.password) {
|
||||
env.RESTIC_REST_PASSWORD = await cryptoUtils.resolveSecret(
|
||||
config.password,
|
||||
);
|
||||
env.RESTIC_REST_PASSWORD = await cryptoUtils.resolveSecret(config.password);
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "sftp": {
|
||||
const decryptedKey = await cryptoUtils.resolveSecret(config.privateKey);
|
||||
const keyPath = path.join(
|
||||
"/tmp",
|
||||
`zerobyte-ssh-${crypto.randomBytes(8).toString("hex")}`,
|
||||
);
|
||||
const keyPath = path.join("/tmp", `zerobyte-ssh-${crypto.randomBytes(8).toString("hex")}`);
|
||||
|
||||
let normalizedKey = decryptedKey.replace(/\r\n/g, "\n");
|
||||
if (!normalizedKey.endsWith("\n")) {
|
||||
|
|
@ -201,12 +148,8 @@ export const buildEnv = async (
|
|||
}
|
||||
|
||||
if (normalizedKey.includes("ENCRYPTED")) {
|
||||
logger.error(
|
||||
"SFTP: Private key appears to be passphrase-protected. Please use an unencrypted key.",
|
||||
);
|
||||
throw new Error(
|
||||
"Passphrase-protected SSH keys are not supported. Please provide an unencrypted private key.",
|
||||
);
|
||||
logger.error("SFTP: Private key appears to be passphrase-protected. Please use an unencrypted key.");
|
||||
throw new Error("Passphrase-protected SSH keys are not supported. Please provide an unencrypted private key.");
|
||||
}
|
||||
|
||||
await fs.writeFile(keyPath, normalizedKey, { mode: 0o600 });
|
||||
|
|
@ -241,25 +184,12 @@ export const buildEnv = async (
|
|||
];
|
||||
|
||||
if (config.skipHostKeyCheck || !config.knownHosts) {
|
||||
sshArgs.push(
|
||||
"-o",
|
||||
"StrictHostKeyChecking=no",
|
||||
"-o",
|
||||
"UserKnownHostsFile=/dev/null",
|
||||
);
|
||||
sshArgs.push("-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null");
|
||||
} else if (config.knownHosts) {
|
||||
const knownHostsPath = path.join(
|
||||
"/tmp",
|
||||
`zerobyte-known-hosts-${crypto.randomBytes(8).toString("hex")}`,
|
||||
);
|
||||
const knownHostsPath = path.join("/tmp", `zerobyte-known-hosts-${crypto.randomBytes(8).toString("hex")}`);
|
||||
await fs.writeFile(knownHostsPath, config.knownHosts, { mode: 0o600 });
|
||||
env._SFTP_KNOWN_HOSTS_PATH = knownHostsPath;
|
||||
sshArgs.push(
|
||||
"-o",
|
||||
"StrictHostKeyChecking=yes",
|
||||
"-o",
|
||||
`UserKnownHostsFile=${knownHostsPath}`,
|
||||
);
|
||||
sshArgs.push("-o", "StrictHostKeyChecking=yes", "-o", `UserKnownHostsFile=${knownHostsPath}`);
|
||||
}
|
||||
|
||||
if (config.port && config.port !== 22) {
|
||||
|
|
@ -274,10 +204,7 @@ export const buildEnv = async (
|
|||
|
||||
if (config.cacert) {
|
||||
const decryptedCert = await cryptoUtils.resolveSecret(config.cacert);
|
||||
const certPath = path.join(
|
||||
"/tmp",
|
||||
`zerobyte-cacert-${crypto.randomBytes(8).toString("hex")}.pem`,
|
||||
);
|
||||
const certPath = path.join("/tmp", `zerobyte-cacert-${crypto.randomBytes(8).toString("hex")}.pem`);
|
||||
await fs.writeFile(certPath, decryptedCert, { mode: 0o600 });
|
||||
env.RESTIC_CACERT = certPath;
|
||||
}
|
||||
|
|
@ -289,11 +216,7 @@ export const buildEnv = async (
|
|||
return env;
|
||||
};
|
||||
|
||||
const init = async (
|
||||
config: RepositoryConfig,
|
||||
organizationId: string,
|
||||
options?: { timeoutMs?: number },
|
||||
) => {
|
||||
const init = async (config: RepositoryConfig, organizationId: string, options?: { timeoutMs?: number }) => {
|
||||
const repoUrl = buildRepoUrl(config);
|
||||
|
||||
logger.info(`Initializing restic repository at ${repoUrl}...`);
|
||||
|
|
@ -303,12 +226,7 @@ const init = async (
|
|||
const args = ["init", "--repo", repoUrl];
|
||||
addCommonArgs(args, env, config);
|
||||
|
||||
const res = await exec({
|
||||
command: "restic",
|
||||
args,
|
||||
env,
|
||||
timeout: options?.timeoutMs ?? 20000,
|
||||
});
|
||||
const res = await exec({ command: "restic", args, env, timeout: options?.timeoutMs ?? 20000 });
|
||||
await cleanupTemporaryKeys(env);
|
||||
|
||||
if (res.exitCode !== 0) {
|
||||
|
|
@ -338,13 +256,7 @@ const backup = async (
|
|||
const repoUrl = buildRepoUrl(config);
|
||||
const env = await buildEnv(config, options?.organizationId);
|
||||
|
||||
const args: string[] = [
|
||||
"--repo",
|
||||
repoUrl,
|
||||
"backup",
|
||||
"--compression",
|
||||
options?.compressionMode ?? "auto",
|
||||
];
|
||||
const args: string[] = ["--repo", repoUrl, "backup", "--compression", options?.compressionMode ?? "auto"];
|
||||
|
||||
if (options?.oneFileSystem) {
|
||||
args.push("--one-file-system");
|
||||
|
|
@ -362,9 +274,7 @@ const backup = async (
|
|||
|
||||
let includeFile: string | null = null;
|
||||
if (options?.include && options.include.length > 0) {
|
||||
const tmp = await fs.mkdtemp(
|
||||
path.join(os.tmpdir(), "zerobyte-restic-include-"),
|
||||
);
|
||||
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-include-"));
|
||||
includeFile = path.join(tmp, `include.txt`);
|
||||
|
||||
await fs.writeFile(includeFile, options.include.join("\n"), "utf-8");
|
||||
|
|
@ -380,9 +290,7 @@ const backup = async (
|
|||
|
||||
let excludeFile: string | null = null;
|
||||
if (options?.exclude && options.exclude.length > 0) {
|
||||
const tmp = await fs.mkdtemp(
|
||||
path.join(os.tmpdir(), "zerobyte-restic-exclude-"),
|
||||
);
|
||||
const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-restic-exclude-"));
|
||||
excludeFile = path.join(tmp, `exclude.txt`);
|
||||
|
||||
await fs.writeFile(excludeFile, options.exclude.join("\n"), "utf-8");
|
||||
|
|
@ -409,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
|
||||
|
|
@ -472,13 +382,13 @@ const backup = async (
|
|||
};
|
||||
|
||||
const restoreProgressSchema = type({
|
||||
message_type: "'status'",
|
||||
message_type: "'status' | 'summary'",
|
||||
seconds_elapsed: "number",
|
||||
percent_done: "number",
|
||||
percent_done: "number = 0",
|
||||
total_files: "number",
|
||||
files_done: "number",
|
||||
total_bytes: "number",
|
||||
bytes_done: "number",
|
||||
files_restored: "number = 0",
|
||||
total_bytes: "number = 0",
|
||||
bytes_restored: "number = 0",
|
||||
});
|
||||
|
||||
export type RestoreProgress = typeof restoreProgressSchema.infer;
|
||||
|
|
@ -500,23 +410,12 @@ const restore = async (
|
|||
const repoUrl = buildRepoUrl(config);
|
||||
const env = await buildEnv(config, options.organizationId);
|
||||
|
||||
const args: string[] = [
|
||||
"--repo",
|
||||
repoUrl,
|
||||
"restore",
|
||||
snapshotId,
|
||||
"--target",
|
||||
target,
|
||||
];
|
||||
const args: string[] = ["--repo", repoUrl, "restore", snapshotId, "--target", target];
|
||||
|
||||
if (options?.overwrite) {
|
||||
args.push("--overwrite", options.overwrite);
|
||||
}
|
||||
|
||||
if (options?.delete) {
|
||||
args.push("--delete");
|
||||
}
|
||||
|
||||
if (options?.include?.length) {
|
||||
for (const pattern of options.include) {
|
||||
args.push("--include", pattern);
|
||||
|
|
@ -544,6 +443,8 @@ const restore = async (
|
|||
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
|
||||
|
|
@ -576,23 +477,16 @@ const restore = async (
|
|||
try {
|
||||
summaryLine = JSON.parse(lastLine ?? "{}");
|
||||
} catch (_) {
|
||||
logger.warn(
|
||||
"Failed to parse restic restore output JSON summary.",
|
||||
lastLine,
|
||||
);
|
||||
logger.warn("Failed to parse restic restore output JSON summary.", lastLine);
|
||||
summaryLine = {};
|
||||
}
|
||||
|
||||
logger.debug(
|
||||
`Restic restore output last line: ${JSON.stringify(summaryLine)}`,
|
||||
);
|
||||
logger.debug(`Restic restore output last line: ${JSON.stringify(summaryLine)}`);
|
||||
const result = resticRestoreOutputSchema(summaryLine);
|
||||
|
||||
if (result instanceof type.errors) {
|
||||
logger.warn(`Restic restore output validation failed: ${result.summary}`);
|
||||
logger.info(
|
||||
`Restic restore completed for snapshot ${snapshotId} to target ${target}`,
|
||||
);
|
||||
logger.info(`Restic restore completed for snapshot ${snapshotId} to target ${target}`);
|
||||
const fallback: ResticRestoreOutputDto = {
|
||||
message_type: "summary" as const,
|
||||
total_files: 0,
|
||||
|
|
@ -611,10 +505,7 @@ const restore = async (
|
|||
return result;
|
||||
};
|
||||
|
||||
const snapshots = async (
|
||||
config: RepositoryConfig,
|
||||
options: { tags?: string[]; organizationId: string },
|
||||
) => {
|
||||
const snapshots = async (config: RepositoryConfig, options: { tags?: string[]; organizationId: string }) => {
|
||||
const { tags, organizationId } = options;
|
||||
|
||||
const repoUrl = buildRepoUrl(config);
|
||||
|
|
@ -641,12 +532,8 @@ const snapshots = async (
|
|||
const result = snapshotInfoSchema.array()(JSON.parse(res.stdout));
|
||||
|
||||
if (result instanceof type.errors) {
|
||||
logger.error(
|
||||
`Restic snapshots output validation failed: ${result.summary}`,
|
||||
);
|
||||
throw new Error(
|
||||
`Restic snapshots output validation failed: ${result.summary}`,
|
||||
);
|
||||
logger.error(`Restic snapshots output validation failed: ${result.summary}`);
|
||||
throw new Error(`Restic snapshots output validation failed: ${result.summary}`);
|
||||
}
|
||||
|
||||
return result;
|
||||
|
|
@ -693,15 +580,7 @@ const forget = async (
|
|||
const repoUrl = buildRepoUrl(config);
|
||||
const env = await buildEnv(config, extra.organizationId);
|
||||
|
||||
const args: string[] = [
|
||||
"--repo",
|
||||
repoUrl,
|
||||
"forget",
|
||||
"--group-by",
|
||||
"tags",
|
||||
"--tag",
|
||||
extra.tag,
|
||||
];
|
||||
const args: string[] = ["--repo", repoUrl, "forget", "--group-by", "tags", "--tag", extra.tag];
|
||||
|
||||
if (extra.dryRun) {
|
||||
args.push("--dry-run", "--no-lock");
|
||||
|
|
@ -744,18 +623,12 @@ const forget = async (
|
|||
}
|
||||
|
||||
const lines = res.stdout.split("\n").filter((line) => line.trim());
|
||||
const result = extra.dryRun
|
||||
? safeJsonParse<ResticForgetResponse>(lines.at(-1) ?? "[]")
|
||||
: null;
|
||||
const result = extra.dryRun ? safeJsonParse<ResticForgetResponse>(lines.at(-1) ?? "[]") : null;
|
||||
|
||||
return { success: true, data: result };
|
||||
};
|
||||
|
||||
const deleteSnapshots = async (
|
||||
config: RepositoryConfig,
|
||||
snapshotIds: string[],
|
||||
organizationId: string,
|
||||
) => {
|
||||
const deleteSnapshots = async (config: RepositoryConfig, snapshotIds: string[], organizationId: string) => {
|
||||
const repoUrl = buildRepoUrl(config);
|
||||
const env = await buildEnv(config, organizationId);
|
||||
|
||||
|
|
@ -763,13 +636,7 @@ const deleteSnapshots = async (
|
|||
throw new Error("No snapshot IDs provided for deletion.");
|
||||
}
|
||||
|
||||
const args: string[] = [
|
||||
"--repo",
|
||||
repoUrl,
|
||||
"forget",
|
||||
...snapshotIds,
|
||||
"--prune",
|
||||
];
|
||||
const args: string[] = ["--repo", repoUrl, "forget", ...snapshotIds, "--prune"];
|
||||
addCommonArgs(args, env, config);
|
||||
|
||||
const res = await exec({ command: "restic", args, env });
|
||||
|
|
@ -783,11 +650,7 @@ const deleteSnapshots = async (
|
|||
return { success: true };
|
||||
};
|
||||
|
||||
const deleteSnapshot = async (
|
||||
config: RepositoryConfig,
|
||||
snapshotId: string,
|
||||
organizationId: string,
|
||||
) => {
|
||||
const deleteSnapshot = async (config: RepositoryConfig, snapshotId: string, organizationId: string) => {
|
||||
return deleteSnapshots(config, [snapshotId], organizationId);
|
||||
};
|
||||
|
||||
|
|
@ -954,22 +817,14 @@ const ls = async (
|
|||
};
|
||||
};
|
||||
|
||||
const unlock = async (
|
||||
config: RepositoryConfig,
|
||||
options: { signal?: AbortSignal; organizationId: string },
|
||||
) => {
|
||||
const unlock = async (config: RepositoryConfig, options: { signal?: AbortSignal; organizationId: string }) => {
|
||||
const repoUrl = buildRepoUrl(config);
|
||||
const env = await buildEnv(config, options.organizationId);
|
||||
|
||||
const args = ["unlock", "--repo", repoUrl, "--remove-all"];
|
||||
addCommonArgs(args, env, config);
|
||||
|
||||
const res = await exec({
|
||||
command: "restic",
|
||||
args,
|
||||
env,
|
||||
signal: options?.signal,
|
||||
});
|
||||
const res = await exec({ command: "restic", args, env, signal: options?.signal });
|
||||
await cleanupTemporaryKeys(env);
|
||||
|
||||
if (options?.signal?.aborted) {
|
||||
|
|
@ -1001,22 +856,12 @@ const check = async (
|
|||
|
||||
addCommonArgs(args, env, config);
|
||||
|
||||
const res = await exec({
|
||||
command: "restic",
|
||||
args,
|
||||
env,
|
||||
signal: options?.signal,
|
||||
});
|
||||
const res = await exec({ command: "restic", args, env, signal: options?.signal });
|
||||
await cleanupTemporaryKeys(env);
|
||||
|
||||
if (options?.signal?.aborted) {
|
||||
logger.warn("Restic check was aborted by signal.");
|
||||
return {
|
||||
success: false,
|
||||
hasErrors: true,
|
||||
output: "",
|
||||
error: "Operation aborted",
|
||||
};
|
||||
return { success: false, hasErrors: true, output: "", error: "Operation aborted" };
|
||||
}
|
||||
|
||||
const { stdout, stderr } = res;
|
||||
|
|
@ -1042,22 +887,14 @@ const check = async (
|
|||
};
|
||||
};
|
||||
|
||||
const repairIndex = async (
|
||||
config: RepositoryConfig,
|
||||
options: { signal?: AbortSignal; organizationId: string },
|
||||
) => {
|
||||
const repairIndex = async (config: RepositoryConfig, options: { signal?: AbortSignal; organizationId: string }) => {
|
||||
const repoUrl = buildRepoUrl(config);
|
||||
const env = await buildEnv(config, options.organizationId);
|
||||
|
||||
const args = ["repair", "index", "--repo", repoUrl];
|
||||
addCommonArgs(args, env, config);
|
||||
|
||||
const res = await exec({
|
||||
command: "restic",
|
||||
args,
|
||||
env,
|
||||
signal: options?.signal,
|
||||
});
|
||||
const res = await exec({ command: "restic", args, env, signal: options?.signal });
|
||||
await cleanupTemporaryKeys(env);
|
||||
|
||||
if (options?.signal?.aborted) {
|
||||
|
|
@ -1097,13 +934,7 @@ const copy = async (
|
|||
RESTIC_FROM_PASSWORD_FILE: sourceEnv.RESTIC_PASSWORD_FILE,
|
||||
};
|
||||
|
||||
const args: string[] = [
|
||||
"--repo",
|
||||
destRepoUrl,
|
||||
"copy",
|
||||
"--from-repo",
|
||||
sourceRepoUrl,
|
||||
];
|
||||
const args: string[] = ["--repo", destRepoUrl, "copy", "--from-repo", sourceRepoUrl];
|
||||
|
||||
if (options.tag) {
|
||||
args.push("--tag", options.tag);
|
||||
|
|
@ -1232,12 +1063,7 @@ export const restic = {
|
|||
};
|
||||
|
||||
export const cleanupTemporaryKeys = async (env: Record<string, string>) => {
|
||||
const keysToClean = [
|
||||
"_SFTP_KEY_PATH",
|
||||
"_SFTP_KNOWN_HOSTS_PATH",
|
||||
"RESTIC_CACERT",
|
||||
"GOOGLE_APPLICATION_CREDENTIALS",
|
||||
];
|
||||
const keysToClean = ["_SFTP_KEY_PATH", "_SFTP_KNOWN_HOSTS_PATH", "RESTIC_CACERT", "GOOGLE_APPLICATION_CREDENTIALS"];
|
||||
|
||||
for (const key of keysToClean) {
|
||||
if (env[key]) {
|
||||
|
|
@ -1246,10 +1072,7 @@ export const cleanupTemporaryKeys = async (env: Record<string, string>) => {
|
|||
}
|
||||
|
||||
// Clean up custom password files
|
||||
if (
|
||||
env.RESTIC_PASSWORD_FILE &&
|
||||
env.RESTIC_PASSWORD_FILE !== RESTIC_PASS_FILE
|
||||
) {
|
||||
if (env.RESTIC_PASSWORD_FILE && env.RESTIC_PASSWORD_FILE !== RESTIC_PASS_FILE) {
|
||||
await fs.unlink(env.RESTIC_PASSWORD_FILE).catch(() => {});
|
||||
}
|
||||
};
|
||||
|
|
|
|||
Loading…
Reference in a new issue