diff --git a/app/server/jobs/backup-execution.ts b/app/server/jobs/backup-execution.ts index ff72bd08..29811af1 100644 --- a/app/server/jobs/backup-execution.ts +++ b/app/server/jobs/backup-execution.ts @@ -1,5 +1,5 @@ import { Job } from "../core/scheduler"; -import { backupsExecutionService } from "../modules/backups/backups.execution"; +import { backupsService } from "../modules/backups/backups.service"; import { logger } from "@zerobyte/core/node"; import { db } from "../db/db"; import { withContext } from "../core/request-context"; @@ -14,7 +14,7 @@ export class BackupExecutionJob extends Job { for (const org of organizations) { await withContext({ organizationId: org.id }, async () => { - const scheduleIds = await backupsExecutionService.getSchedulesToExecute(); + const scheduleIds = await backupsService.getSchedulesToExecute(); if (scheduleIds.length === 0) { return; @@ -23,7 +23,7 @@ export class BackupExecutionJob extends Job { logger.info(`Found ${scheduleIds.length} backup schedule(s) to execute for organization ${org.name}`); for (const scheduleId of scheduleIds) { - backupsExecutionService.executeBackup(scheduleId).catch((err: Error) => { + backupsService.executeBackup(scheduleId).catch((err: Error) => { logger.error(`Error executing backup for schedule ${scheduleId}:`, err); }); } diff --git a/app/server/modules/backups/__tests__/backups.execution.test.ts b/app/server/modules/backups/__tests__/backups.execution.test.ts index 28980273..cd65a5e3 100644 --- a/app/server/modules/backups/__tests__/backups.execution.test.ts +++ b/app/server/modules/backups/__tests__/backups.execution.test.ts @@ -16,6 +16,7 @@ import { NotFoundError, BadRequestError } from "http-errors-enhanced"; import { scheduleQueries } from "../backups.queries"; import { fromAny } from "@total-typescript/shoehorn"; import { repositoriesService } from "~/server/modules/repositories/repositories.service"; +import { repoMutex } from "~/server/core/repository-mutex"; const setup = () => { const resticBackupMock = mock((_: SafeSpawnParams) => @@ -238,9 +239,8 @@ describe("stop backup", () => { expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user"); }); - test("should throw ConflictError when trying to stop non-running backup", async () => { - // arrange - setup(); + test("should stop a queued backup before it acquires the repository lock", async () => { + const { resticBackupMock } = setup(); const volume = await createTestVolume(); const repository = await createTestRepository(); const schedule = await createTestBackupSchedule({ @@ -248,10 +248,54 @@ describe("stop backup", () => { repositoryId: repository.id, }); + const releaseLock = await repoMutex.acquireExclusive(repository.id, "test"); + + try { + const executePromise = backupsExecutionService.executeBackup(schedule.id); + + await waitForExpect(async () => { + const queuedSchedule = await backupsService.getScheduleById(schedule.id); + expect(queuedSchedule.lastBackupStatus).toBe("in_progress"); + }); + + expect(resticBackupMock).not.toHaveBeenCalled(); + + await backupsExecutionService.stopBackup(schedule.id); + releaseLock(); + + await executePromise; + + const updatedSchedule = await backupsService.getScheduleById(schedule.id); + expect(updatedSchedule.lastBackupStatus).toBe("warning"); + expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user"); + expect(resticBackupMock).not.toHaveBeenCalled(); + } finally { + releaseLock(); + } + }); + + test("should throw ConflictError when trying to stop non-running backup", async () => { + // arrange + setup(); + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const previousLastBackupAt = 1_700_000_000_000; + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + lastBackupAt: previousLastBackupAt, + lastBackupStatus: "in_progress", + }); + // act & assert await expect(backupsExecutionService.stopBackup(schedule.id)).rejects.toThrow( "No backup is currently running for this schedule", ); + + const updatedSchedule = await backupsService.getScheduleById(schedule.id); + expect(updatedSchedule.lastBackupAt).toBe(previousLastBackupAt); + expect(updatedSchedule.lastBackupStatus).toBe("warning"); + expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user"); }); test("should throw NotFoundError when schedule does not exist", async () => { diff --git a/app/server/modules/backups/backup-executor.ts b/app/server/modules/backups/backup-executor.ts new file mode 100644 index 00000000..fa45a472 --- /dev/null +++ b/app/server/modules/backups/backup-executor.ts @@ -0,0 +1,87 @@ +import { restic } from "../../core/restic"; +import type { BackupSchedule, Repository, Volume } from "../../db/schema"; +import type { ResticBackupOutputDto, ResticBackupProgressDto } from "@zerobyte/core/restic"; +import { createBackupOptions } from "./backup.helpers"; +import { getVolumePath } from "../volumes/helpers"; + +type BackupExecutionRequest = { + scheduleId: number; + schedule: BackupSchedule; + volume: Volume; + repository: Repository; + organizationId: string; + signal: AbortSignal; + onProgress: (progress: BackupExecutionProgress) => void; +}; + +export type BackupExecutionProgress = ResticBackupProgressDto; + +export type BackupExecutionResult = + | { + status: "unavailable"; + error: Error; + } + | { + status: "completed"; + exitCode: number; + result: ResticBackupOutputDto | null; + warningDetails: string | null; + } + | { + status: "failed"; + error: unknown; + } + | { + status: "cancelled"; + message?: string; + }; + +const activeControllersByScheduleId = new Map(); + +export const backupExecutor = { + track: (scheduleId: number) => { + const abortController = new AbortController(); + activeControllersByScheduleId.set(scheduleId, abortController); + return abortController; + }, + untrack: (scheduleId: number, abortController: AbortController) => { + if (activeControllersByScheduleId.get(scheduleId) === abortController) { + activeControllersByScheduleId.delete(scheduleId); + } + }, + execute: async (params: BackupExecutionRequest): Promise => { + const { schedule, volume, repository, organizationId, signal, onProgress } = params; + try { + const volumePath = getVolumePath(volume); + const backupOptions = createBackupOptions(schedule, volumePath, signal); + + const result = await restic.backup(repository.config, volumePath, { + ...backupOptions, + compressionMode: repository.compressionMode ?? "auto", + organizationId, + onProgress, + }); + + return { + status: "completed", + exitCode: result.exitCode, + result: result.result, + warningDetails: result.warningDetails, + } satisfies BackupExecutionResult; + } catch (error) { + return { + status: "failed", + error, + } satisfies BackupExecutionResult; + } + }, + cancel: (scheduleId: number) => { + const abortController = activeControllersByScheduleId.get(scheduleId); + if (!abortController) { + return false; + } + + abortController.abort(); + return true; + }, +}; diff --git a/app/server/modules/backups/backup.helpers.ts b/app/server/modules/backups/backup.helpers.ts index cc99dfbb..f7f68ac4 100644 --- a/app/server/modules/backups/backup.helpers.ts +++ b/app/server/modules/backups/backup.helpers.ts @@ -19,6 +19,15 @@ export const calculateNextRun = (cronExpression: string) => { } }; +export const isValidCron = (expression: string) => { + try { + CronExpressionParser.parse(expression); + return true; + } catch { + return false; + } +}; + export const processPattern = (pattern: string, volumePath: string, relative = false) => { const isNegated = pattern.startsWith("!"); const p = isNegated ? pattern.slice(1) : pattern; @@ -47,7 +56,7 @@ export const processPattern = (pattern: string, volumePath: string, relative = f return isNegated ? `!${processed}` : processed; }; -export const createBackupOptions = (schedule: BackupSchedule, volumePath: string, signal: AbortSignal) => ({ +export const createBackupOptions = (schedule: BackupSchedule, volumePath: string, signal?: AbortSignal) => ({ tags: [schedule.shortId], oneFileSystem: schedule.oneFileSystem, signal, diff --git a/app/server/modules/backups/backups.controller.ts b/app/server/modules/backups/backups.controller.ts index 41c54352..9a47a756 100644 --- a/app/server/modules/backups/backups.controller.ts +++ b/app/server/modules/backups/backups.controller.ts @@ -44,10 +44,10 @@ import { } from "../notifications/notifications.dto"; import { notificationsService } from "../notifications/notifications.service"; import { requireAuth } from "../auth/auth.middleware"; -import { backupsExecutionService } from "./backups.execution"; import { logger } from "@zerobyte/core/node"; import { asShortId } from "~/server/utils/branded"; import { cache, cacheKeys } from "~/server/utils/cache"; +import { getScheduleByIdOrShortId } from "./helpers/backup-schedule-lookups"; export const backupScheduleController = new Hono() .use(requireAuth) @@ -58,7 +58,7 @@ export const backupScheduleController = new Hono() }) .get("/:shortId", getBackupScheduleDto, async (c) => { const shortId = asShortId(c.req.param("shortId")); - const schedule = await backupsService.getScheduleByShortId(shortId); + const schedule = await getScheduleByIdOrShortId(shortId); return c.json(schedule, 200); }) @@ -89,8 +89,8 @@ export const backupScheduleController = new Hono() }) .post("/:shortId/run", runBackupNowDto, async (c) => { const shortId = asShortId(c.req.param("shortId")); - const schedule = await backupsService.getScheduleByShortId(shortId); - const result = await backupsExecutionService.validateBackupExecution(schedule.id, true); + const schedule = await getScheduleByIdOrShortId(shortId); + const result = await backupsService.validateBackupExecution(schedule.id, true); if (result.type === "failure") { throw result.error; @@ -100,7 +100,7 @@ export const backupScheduleController = new Hono() return c.json({ success: true }, 200); } - backupsExecutionService.executeBackup(schedule.id, true).catch((err) => { + backupsService.executeBackup(schedule.id, true).catch((err) => { logger.error(`Error executing manual backup for schedule ${shortId}:`, err); }); @@ -108,21 +108,21 @@ export const backupScheduleController = new Hono() }) .post("/:shortId/stop", stopBackupDto, async (c) => { const shortId = asShortId(c.req.param("shortId")); - const schedule = await backupsService.getScheduleByShortId(shortId); - await backupsExecutionService.stopBackup(schedule.id); + const schedule = await getScheduleByIdOrShortId(shortId); + await backupsService.stopBackup(schedule.id); return c.json({ success: true }, 200); }) .post("/:shortId/forget", runForgetDto, async (c) => { const shortId = asShortId(c.req.param("shortId")); - const schedule = await backupsService.getScheduleByShortId(shortId); - await backupsExecutionService.runForget(schedule.id); + const schedule = await getScheduleByIdOrShortId(shortId); + await backupsService.runForget(schedule.id); return c.json({ success: true }, 200); }) .get("/:shortId/notifications", getScheduleNotificationsDto, async (c) => { const shortId = asShortId(c.req.param("shortId")); - const schedule = await backupsService.getScheduleByShortId(shortId); + const schedule = await getScheduleByIdOrShortId(shortId); const assignments = await notificationsService.getScheduleNotifications(schedule.id); return c.json(assignments, 200); @@ -133,7 +133,7 @@ export const backupScheduleController = new Hono() validator("json", updateScheduleNotificationsBody), async (c) => { const shortId = asShortId(c.req.param("shortId")); - const schedule = await backupsService.getScheduleByShortId(shortId); + const schedule = await getScheduleByIdOrShortId(shortId); const body = c.req.valid("json"); const assignments = await notificationsService.updateScheduleNotifications(schedule.id, body.assignments); @@ -167,12 +167,12 @@ export const backupScheduleController = new Hono() }) .get("/:shortId/progress", getBackupProgressDto, async (c) => { const shortId = asShortId(c.req.param("shortId")); - const schedule = await backupsService.getScheduleByShortId(shortId); + const schedule = await getScheduleByIdOrShortId(shortId); if (schedule.lastBackupStatus !== "in_progress") { cache.del(cacheKeys.backup.progress(schedule.id)); return c.json(null, 200); } - const progress = backupsExecutionService.getBackupProgress(schedule.id); + const progress = backupsService.getBackupProgress(schedule.id); return c.json(progress ?? null, 200); }); diff --git a/app/server/modules/backups/backups.execution.ts b/app/server/modules/backups/backups.execution.ts index ba864f85..f85388bf 100644 --- a/app/server/modules/backups/backups.execution.ts +++ b/app/server/modules/backups/backups.execution.ts @@ -1,477 +1,11 @@ -import { NotFoundError, BadRequestError, ConflictError } from "http-errors-enhanced"; -import type { BackupSchedule, Volume, Repository } from "../../db/schema"; -import { restic } from "../../core/restic"; -import { logger } from "@zerobyte/core/node"; -import { cache, cacheKeys } from "../../utils/cache"; -import { getVolumePath } from "../volumes/helpers"; -import { toErrorDetails, toMessage } from "../../utils/errors"; -import { serverEvents } from "../../core/events"; -import { notificationsService } from "../notifications/notifications.service"; -import { repoMutex } from "../../core/repository-mutex"; -import { repositoriesService } from "../repositories/repositories.service"; -import { getOrganizationId } from "~/server/core/request-context"; -import { scheduleQueries, mirrorQueries, repositoryQueries } from "./backups.queries"; -import { calculateNextRun, createBackupOptions } from "./backup.helpers"; -import type { ResticBackupOutputDto } from "@zerobyte/core/restic"; -import type { BackupProgressEventDto } from "~/schemas/events-dto"; - -const runningBackups = new Map(); - -export const getBackupProgress = (scheduleId: number): BackupProgressEventDto | undefined => - cache.get(cacheKeys.backup.progress(scheduleId)); - -interface BackupContext { - schedule: BackupSchedule; - volume: Volume; - repository: Repository; - organizationId: string; -} -type ValidationSuccess = { - type: "success"; - context: BackupContext; -}; -type ValidationFailure = { - type: "failure"; - error: Error; - partialContext?: Partial; -}; -type ValidationSkipped = { - type: "skipped"; - reason: string; -}; -type ValidationResult = ValidationSuccess | ValidationFailure | ValidationSkipped; - -const validateBackupExecution = async (scheduleId: number, manual = false): Promise => { - const organizationId = getOrganizationId(); - const result = await scheduleQueries.findById(scheduleId, organizationId); - - if (!result) { - return { type: "failure", error: new NotFoundError("Backup schedule not found") }; - } - - const { volume, repository, ...schedule } = result; - - if (!schedule) { - return { type: "failure", error: new NotFoundError("Backup schedule not found") }; - } - - if (!schedule.enabled && !manual) { - logger.info(`Backup schedule ${scheduleId} is disabled. Skipping execution.`); - return { type: "skipped", reason: "Backup schedule is disabled" }; - } - - if (schedule.lastBackupStatus === "in_progress") { - logger.info(`Backup schedule ${scheduleId} is already in progress. Skipping execution.`); - return { type: "skipped", reason: "Backup is already in progress" }; - } - - if (!volume) { - return { type: "failure", error: new NotFoundError("Volume not found"), partialContext: { schedule } }; - } - - if (!repository) { - return { type: "failure", error: new NotFoundError("Repository not found"), partialContext: { schedule, volume } }; - } - - if (volume.status !== "mounted") { - return { - type: "failure", - error: new BadRequestError("Volume is not mounted"), - partialContext: { schedule, volume, repository }, - }; - } - - return { - type: "success", - context: { schedule, volume, repository, organizationId }, - }; -}; - -const emitBackupStarted = (ctx: BackupContext, scheduleId: number) => { - logger.info( - `Starting backup ${ctx.schedule.name} for volume ${ctx.volume.name} to repository ${ctx.repository.name}`, - ); - - serverEvents.emit("backup:started", { - organizationId: ctx.organizationId, - scheduleId: ctx.schedule.shortId, - volumeName: ctx.volume.name, - repositoryName: ctx.repository.name, - }); - - notificationsService - .sendBackupNotification(scheduleId, "start", { - volumeName: ctx.volume.name, - repositoryName: ctx.repository.name, - scheduleName: ctx.schedule.name, - }) - .catch((error) => { - logger.error(`Failed to send backup start notification: ${toMessage(error)}`); - }); -}; - -const runBackupOperation = async (ctx: BackupContext, signal: AbortSignal) => { - const volumePath = getVolumePath(ctx.volume); - const backupOptions = createBackupOptions(ctx.schedule, volumePath, signal); - - const releaseBackupLock = await repoMutex.acquireShared(ctx.repository.id, `backup:${ctx.volume.name}`, signal); - - try { - const result = await restic.backup(ctx.repository.config, volumePath, { - ...backupOptions, - compressionMode: ctx.repository.compressionMode ?? "auto", - organizationId: ctx.organizationId, - onProgress: (progress) => { - const progressEvent = { - scheduleId: ctx.schedule.shortId, - volumeName: ctx.volume.name, - repositoryName: ctx.repository.name, - ...progress, - }; - cache.set(cacheKeys.backup.progress(ctx.schedule.id), progressEvent, 60 * 60); - serverEvents.emit("backup:progress", { - organizationId: ctx.organizationId, - ...progressEvent, - }); - }, - }); - return result; - } finally { - releaseBackupLock(); - } -}; - -const finalizeSuccessfulBackup = async ( - ctx: BackupContext, - scheduleId: number, - exitCode: number, - result: ResticBackupOutputDto | null, - warningDetails: string | null, -) => { - const finalStatus = exitCode === 0 ? "success" : "warning"; - - if (ctx.schedule.retentionPolicy) { - void runForget(scheduleId).catch((error) => { - logger.error(`Failed to run retention policy for schedule ${scheduleId}: ${toMessage(error)}`); - }); - } - - void copyToMirrors(scheduleId, ctx.repository, ctx.schedule.retentionPolicy).catch((error) => { - logger.error(`Background mirror copy failed for schedule ${scheduleId}: ${toMessage(error)}`); - }); - - cache.delByPrefix(cacheKeys.repository.all(ctx.repository.id)); - - void repositoriesService.refreshRepositoryStats(ctx.repository.shortId).catch((error) => { - logger.error( - `Background repository stats refresh failed for schedule ${scheduleId} (${ctx.repository.shortId}): ${toMessage(error)}`, - ); - }); - - const nextBackupAt = ctx.schedule.cronExpression ? calculateNextRun(ctx.schedule.cronExpression) : null; - await scheduleQueries.updateStatus(scheduleId, ctx.organizationId, { - lastBackupAt: Date.now(), - lastBackupStatus: finalStatus, - lastBackupError: finalStatus === "warning" ? warningDetails : null, - nextBackupAt, - }); - - if (finalStatus === "warning") { - logger.warn( - `Backup ${ctx.schedule.name} completed with warnings for volume ${ctx.volume.name} to repository ${ctx.repository.name}`, - ); - } else { - logger.info( - `Backup ${ctx.schedule.name} completed successfully for volume ${ctx.volume.name} to repository ${ctx.repository.name}`, - ); - } - - serverEvents.emit("backup:completed", { - organizationId: ctx.organizationId, - scheduleId: ctx.schedule.shortId, - volumeName: ctx.volume.name, - repositoryName: ctx.repository.name, - status: finalStatus, - summary: result ?? undefined, - }); - - notificationsService - .sendBackupNotification(scheduleId, finalStatus, { - volumeName: ctx.volume.name, - repositoryName: ctx.repository.name, - scheduleName: ctx.schedule.name, - summary: result ?? undefined, - }) - .catch((error) => { - logger.error(`Failed to send backup success notification: ${toMessage(error)}`); - }); -}; - -const handleValidationResult = async (scheduleId: number, result: ValidationFailure | ValidationSkipped) => { - const organizationId = getOrganizationId(); - - if (result.type === "skipped") { - logger.info(`Backup execution for schedule ${scheduleId} was skipped: ${result.reason}`); - return; - } - - await handleBackupFailure(scheduleId, organizationId, result.error, result.partialContext); -}; - -const handleBackupFailure = async ( - scheduleId: number, - organizationId: string, - error: unknown, - partialContext?: Partial, -): Promise => { - const errorMessage = toMessage(error); - const errorDetails = toErrorDetails(error); - - await scheduleQueries.updateStatus(scheduleId, organizationId, { - lastBackupAt: Date.now(), - lastBackupStatus: "error", - lastBackupError: errorDetails, - }); - - if (partialContext?.schedule && partialContext?.volume && partialContext?.repository) { - const ctx = partialContext as BackupContext; - - logger.error( - `Backup ${ctx.schedule.name} failed for volume ${ctx.volume.name} to repository ${ctx.repository.name}: ${errorMessage}`, - ); - - serverEvents.emit("backup:completed", { - organizationId, - scheduleId: ctx.schedule.shortId, - volumeName: ctx.volume.name, - repositoryName: ctx.repository.name, - status: "error", - }); - - notificationsService - .sendBackupNotification(scheduleId, "failure", { - volumeName: ctx.volume.name, - repositoryName: ctx.repository.name, - scheduleName: ctx.schedule.name, - error: errorDetails, - }) - .catch((notifError) => { - logger.error(`Failed to send backup failure notification: ${toMessage(notifError)}`); - }); - } -}; - -const executeBackup = async (scheduleId: number, manual = false): Promise => { - const result = await validateBackupExecution(scheduleId, manual); - - if (result.type !== "success") { - return handleValidationResult(scheduleId, result); - } - - const { context: ctx } = result; - cache.del(cacheKeys.backup.progress(scheduleId)); - emitBackupStarted(ctx, scheduleId); - - const nextBackupAt = ctx.schedule.cronExpression ? calculateNextRun(ctx.schedule.cronExpression) : null; - - await scheduleQueries.updateStatus(scheduleId, ctx.organizationId, { - lastBackupStatus: "in_progress", - lastBackupError: null, - nextBackupAt, - }); - - const abortController = new AbortController(); - runningBackups.set(scheduleId, abortController); - - try { - const backupResult = await runBackupOperation(ctx, abortController.signal); - await finalizeSuccessfulBackup( - ctx, - scheduleId, - backupResult.exitCode, - backupResult.result, - backupResult.warningDetails, - ); - } catch (error) { - await handleBackupFailure(scheduleId, ctx.organizationId, error, ctx); - } finally { - runningBackups.delete(scheduleId); - cache.del(cacheKeys.backup.progress(scheduleId)); - } -}; - -const getSchedulesToExecute = async () => { - const organizationId = getOrganizationId(); - return scheduleQueries.findExecutable(organizationId); -}; - -const stopBackup = async (scheduleId: number) => { - const organizationId = getOrganizationId(); - const schedule = await scheduleQueries.findById(scheduleId, organizationId); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } - - try { - const abortController = runningBackups.get(scheduleId); - if (!abortController) { - throw new ConflictError("No backup is currently running for this schedule"); - } - - logger.info(`Stopping backup for schedule ${scheduleId}`); - abortController.abort(); - } finally { - await scheduleQueries.updateStatus(scheduleId, organizationId, { - lastBackupStatus: "warning", - lastBackupError: "Backup was stopped by the user", - }); - } -}; - -const runForget = async (scheduleId: number, repositoryId?: string) => { - const organizationId = getOrganizationId(); - const schedule = await scheduleQueries.findById(scheduleId, organizationId); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } - - if (!schedule.retentionPolicy) { - throw new BadRequestError("No retention policy configured for this schedule"); - } - - const repository = await repositoryQueries.findById(repositoryId ?? schedule.repositoryId, organizationId); - - if (!repository) { - throw new NotFoundError("Repository not found"); - } - - logger.info(`running retention policy (forget) for schedule ${scheduleId}`); - const releaseLock = await repoMutex.acquireExclusive(repository.id, `forget:${scheduleId}`); - - try { - await restic.forget(repository.config, schedule.retentionPolicy, { tag: schedule.shortId, organizationId }); - cache.delByPrefix(cacheKeys.repository.all(repository.id)); - } finally { - releaseLock(); - } - - logger.info(`Retention policy applied successfully for schedule ${scheduleId}`); -}; - -const copyToMirrors = async ( - scheduleId: number, - sourceRepository: Repository, - retentionPolicy: BackupSchedule["retentionPolicy"], -) => { - const organizationId = getOrganizationId(); - const schedule = await scheduleQueries.findById(scheduleId, organizationId); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } - - const mirrors = await mirrorQueries.findEnabledBySchedule(scheduleId); - - if (mirrors.length === 0) { - return; - } - - logger.info(`[Background] Copying snapshots to ${mirrors.length} mirror repositories for schedule ${scheduleId}`); - - for (const mirror of mirrors) { - await copyToSingleMirror(scheduleId, schedule, sourceRepository, mirror, retentionPolicy, organizationId); - } -}; - -const copyToSingleMirror = async ( - scheduleId: number, - schedule: BackupSchedule, - sourceRepository: Repository, - mirror: { - repositoryId: string; - repository: Repository; - }, - retentionPolicy: BackupSchedule["retentionPolicy"], - organizationId: string, -) => { - try { - logger.info(`[Background] Copying to mirror repository: ${mirror.repository.name}`); - - serverEvents.emit("mirror:started", { - organizationId, - scheduleId: schedule.shortId, - repositoryId: mirror.repository.shortId, - repositoryName: mirror.repository.name, - }); - - await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, { - lastCopyStatus: "in_progress", - lastCopyError: null, - }); - - const releaseSource = await repoMutex.acquireShared(sourceRepository.id, `mirror_source:${scheduleId}`); - const releaseMirror = await repoMutex.acquireShared(mirror.repository.id, `mirror:${scheduleId}`); - - try { - await restic.copy(sourceRepository.config, mirror.repository.config, { tag: schedule.shortId, organizationId }); - cache.delByPrefix(cacheKeys.repository.all(mirror.repository.id)); - } finally { - releaseSource(); - releaseMirror(); - } - - if (retentionPolicy) { - void runForget(scheduleId, mirror.repository.id).catch((error) => { - logger.error( - `Failed to run retention policy for mirror repository ${mirror.repository.name}: ${toMessage(error)}`, - ); - }); - } - - await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, { - lastCopyAt: Date.now(), - lastCopyStatus: "success", - lastCopyError: null, - }); - - logger.info(`[Background] Successfully copied to mirror repository: ${mirror.repository.name}`); - - serverEvents.emit("mirror:completed", { - organizationId, - scheduleId: schedule.shortId, - repositoryId: mirror.repository.shortId, - repositoryName: mirror.repository.name, - status: "success", - }); - } catch (error) { - const errorMessage = toMessage(error); - logger.error(`[Background] Failed to copy to mirror repository ${mirror.repository.name}: ${errorMessage}`); - - await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, { - lastCopyAt: Date.now(), - lastCopyStatus: "error", - lastCopyError: errorMessage, - }); - - serverEvents.emit("mirror:completed", { - organizationId, - scheduleId: schedule.shortId, - repositoryId: mirror.repository.shortId, - repositoryName: mirror.repository.name, - status: "error", - error: errorMessage, - }); - } -}; +import { backupsService } from "./backups.service"; export const backupsExecutionService = { - executeBackup, - validateBackupExecution, - getSchedulesToExecute, - stopBackup, - runForget, - copyToMirrors, - getBackupProgress, + executeBackup: backupsService.executeBackup, + validateBackupExecution: backupsService.validateBackupExecution, + getSchedulesToExecute: backupsService.getSchedulesToExecute, + stopBackup: backupsService.stopBackup, + runForget: backupsService.runForget, + copyToMirrors: backupsService.copyToMirrors, + getBackupProgress: backupsService.getBackupProgress, }; diff --git a/app/server/modules/backups/backups.service.ts b/app/server/modules/backups/backups.service.ts index 02609981..3b9ff793 100644 --- a/app/server/modules/backups/backups.service.ts +++ b/app/server/modules/backups/backups.service.ts @@ -1,25 +1,31 @@ import { and, eq, inArray } from "drizzle-orm"; -import { CronExpressionParser } from "cron-parser"; import { NotFoundError, BadRequestError, ConflictError } from "http-errors-enhanced"; - -const isValidCron = (expression: string): boolean => { - try { - CronExpressionParser.parse(expression); - return true; - } catch { - return false; - } -}; -import { db } from "../../db/db"; -import { backupScheduleMirrorsTable, backupScheduleNotificationsTable, backupSchedulesTable } from "../../db/schema"; -import type { CreateBackupScheduleBody, UpdateBackupScheduleBody, UpdateScheduleMirrorsBody } from "./backups.dto"; - +import { logger } from "@zerobyte/core/node"; import { checkMirrorCompatibility, getIncompatibleMirrorError } from "~/server/utils/backend-compatibility"; import { generateShortId } from "~/server/utils/id"; import { getOrganizationId } from "~/server/core/request-context"; -import { calculateNextRun } from "./backup.helpers"; import { asShortId, type ShortId } from "~/server/utils/branded"; import { validateCustomResticParams } from "@zerobyte/core/restic/server"; +import { db } from "../../db/db"; +import { backupScheduleMirrorsTable, backupScheduleNotificationsTable, backupSchedulesTable } from "../../db/schema"; +import { cache, cacheKeys } from "../../utils/cache"; +import { repoMutex } from "../../core/repository-mutex"; +import { backupExecutor } from "./backup-executor"; +import { calculateNextRun, isValidCron } from "./backup.helpers"; +import { scheduleQueries } from "./backups.queries"; +import type { CreateBackupScheduleBody, UpdateBackupScheduleBody, UpdateScheduleMirrorsBody } from "./backups.dto"; +import { + emitBackupStarted, + finalizeSuccessfulBackup, + getBackupProgress, + handleBackupCancellation, + handleBackupFailure, + handleValidationResult, + updateBackupProgress, + validateBackupExecution, +} from "./helpers/backup-lifecycle"; +import { getScheduleByIdOrShortId } from "./helpers/backup-schedule-lookups"; +import { copyToMirrors, runForget } from "./helpers/backup-maintenance"; const listSchedules = async () => { const organizationId = getOrganizationId(); @@ -32,62 +38,11 @@ const listSchedules = async () => { }; const getScheduleById = async (scheduleId: number) => { - const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { AND: [{ id: scheduleId }, { organizationId }] }, - with: { volume: true, repository: true }, - }); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } - - if (!schedule.volume || !schedule.repository) { - throw new NotFoundError("Backup schedule not found"); - } - - return schedule; + return getScheduleByIdOrShortId(scheduleId); }; const getScheduleByShortId = async (shortId: ShortId) => { - const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { AND: [{ shortId: { eq: shortId } }, { organizationId }] }, - with: { volume: true, repository: true }, - }); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } - - if (!schedule.volume || !schedule.repository) { - throw new NotFoundError("Backup schedule not found"); - } - - return schedule; -}; - -const getScheduleByIdOrShortId = async (idOrShortId: string | number) => { - const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { - AND: [ - { OR: [{ id: Number(idOrShortId) }, { shortId: { eq: asShortId(String(idOrShortId)) } }] }, - { organizationId }, - ], - }, - with: { volume: true, repository: true }, - }); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } - - if (!schedule.volume || !schedule.repository) { - throw new NotFoundError("Backup schedule not found"); - } - - return schedule; + return getScheduleByIdOrShortId(shortId); }; const createSchedule = async (data: CreateBackupScheduleBody) => { @@ -169,18 +124,7 @@ const createSchedule = async (data: CreateBackupScheduleBody) => { const updateSchedule = async (scheduleIdOrShortId: number | string, data: UpdateBackupScheduleBody) => { const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { - AND: [ - { OR: [{ id: Number(scheduleIdOrShortId) }, { shortId: { eq: asShortId(String(scheduleIdOrShortId)) } }] }, - { organizationId }, - ], - }, - }); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } + const schedule = await getScheduleByIdOrShortId(scheduleIdOrShortId); if (data.cronExpression && !isValidCron(data.cronExpression)) { throw new BadRequestError("Invalid cron expression"); @@ -235,18 +179,7 @@ const updateSchedule = async (scheduleIdOrShortId: number | string, data: Update const deleteSchedule = async (scheduleIdOrShortId: number | string) => { const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { - AND: [ - { OR: [{ id: Number(scheduleIdOrShortId) }, { shortId: { eq: asShortId(String(scheduleIdOrShortId)) } }] }, - { organizationId }, - ], - }, - }); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } + const schedule = await getScheduleByIdOrShortId(scheduleIdOrShortId); await db .delete(backupSchedulesTable) @@ -284,19 +217,7 @@ const getScheduleForVolume = async (volumeIdOrShortId: number | string) => { }; const getMirrors = async (scheduleIdOrShortId: number | string) => { - const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { - AND: [ - { OR: [{ id: Number(scheduleIdOrShortId) }, { shortId: { eq: asShortId(String(scheduleIdOrShortId)) } }] }, - { organizationId }, - ], - }, - }); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } + const schedule = await getScheduleByIdOrShortId(scheduleIdOrShortId); const mirrors = await db.query.backupScheduleMirrorsTable.findMany({ where: { @@ -320,19 +241,7 @@ const getMirrors = async (scheduleIdOrShortId: number | string) => { const updateMirrors = async (scheduleIdOrShortId: number | string, data: UpdateScheduleMirrorsBody) => { const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { - AND: [ - { OR: [{ id: Number(scheduleIdOrShortId) }, { shortId: { eq: asShortId(String(scheduleIdOrShortId)) } }] }, - { organizationId }, - ], - }, - with: { repository: true }, - }); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } + const schedule = await getScheduleByIdOrShortId(scheduleIdOrShortId); const normalizedMirrors = await Promise.all( data.mirrors.map(async (mirror) => { @@ -399,19 +308,7 @@ const updateMirrors = async (scheduleIdOrShortId: number | string, data: UpdateS const getMirrorCompatibility = async (scheduleIdOrShortId: number | string) => { const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { - AND: [ - { OR: [{ id: Number(scheduleIdOrShortId) }, { shortId: { eq: asShortId(String(scheduleIdOrShortId)) } }] }, - { organizationId }, - ], - }, - with: { repository: true }, - }); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } + const schedule = await getScheduleByIdOrShortId(scheduleIdOrShortId); const allRepositories = await db.query.repositoriesTable.findMany({ where: { organizationId } }); const repos = allRepositories.filter((repo) => repo.id !== schedule.repositoryId); @@ -485,10 +382,103 @@ const cleanupOrphanedSchedules = async () => { return { deletedSchedules: orphanScheduleIds.length }; }; +const executeBackup = async (scheduleId: number, manual = false) => { + const result = await validateBackupExecution(scheduleId, manual); + + if (result.type !== "success") { + return handleValidationResult(scheduleId, result); + } + + const { context: ctx } = result; + cache.del(cacheKeys.backup.progress(scheduleId)); + emitBackupStarted(ctx, scheduleId); + + await scheduleQueries.updateStatus(scheduleId, ctx.organizationId, { + lastBackupStatus: "in_progress", + lastBackupError: null, + ...(ctx.schedule.cronExpression ? { nextBackupAt: calculateNextRun(ctx.schedule.cronExpression) } : {}), + }); + + const abortController = backupExecutor.track(scheduleId); + + try { + const releaseLock = await repoMutex.acquireShared( + ctx.repository.id, + `backup:${ctx.volume.name}`, + abortController.signal, + ); + + try { + const executionResult = await backupExecutor.execute({ + scheduleId, + schedule: ctx.schedule, + volume: ctx.volume, + repository: ctx.repository, + organizationId: ctx.organizationId, + signal: abortController.signal, + onProgress: (progress) => { + updateBackupProgress(ctx, progress); + }, + }); + + switch (executionResult.status) { + case "unavailable": + return handleBackupFailure(scheduleId, ctx.organizationId, executionResult.error, ctx); + case "completed": + return finalizeSuccessfulBackup( + ctx, + executionResult.exitCode, + executionResult.result, + executionResult.warningDetails, + ); + case "failed": + return handleBackupFailure(scheduleId, ctx.organizationId, executionResult.error, ctx); + case "cancelled": + return handleBackupCancellation(scheduleId, ctx.organizationId, executionResult.message); + } + } finally { + releaseLock(); + } + } catch (error) { + if (abortController.signal.aborted) { + return; + } + + return handleBackupFailure(scheduleId, ctx.organizationId, error, ctx); + } finally { + backupExecutor.untrack(scheduleId, abortController); + cache.del(cacheKeys.backup.progress(scheduleId)); + } +}; + +const getSchedulesToExecute = async () => { + const organizationId = getOrganizationId(); + return scheduleQueries.findExecutable(organizationId); +}; + +const stopBackup = async (scheduleId: number) => { + const organizationId = getOrganizationId(); + const schedule = await scheduleQueries.findById(scheduleId, organizationId); + + if (!schedule) { + throw new NotFoundError("Backup schedule not found"); + } + + try { + if (!backupExecutor.cancel(scheduleId)) { + throw new ConflictError("No backup is currently running for this schedule"); + } + + logger.info(`Stopping backup for schedule ${scheduleId}`); + } finally { + await handleBackupCancellation(scheduleId, organizationId, undefined, false); + } +}; export const backupsService = { listSchedules, getScheduleById, + getScheduleByShortId, getScheduleByIdOrShortId, createSchedule, updateSchedule, @@ -498,6 +488,12 @@ export const backupsService = { updateMirrors, getMirrorCompatibility, reorderSchedules, - getScheduleByShortId, cleanupOrphanedSchedules, + getBackupProgress, + validateBackupExecution, + executeBackup, + getSchedulesToExecute, + stopBackup, + runForget, + copyToMirrors, }; diff --git a/app/server/modules/backups/helpers/backup-lifecycle.ts b/app/server/modules/backups/helpers/backup-lifecycle.ts new file mode 100644 index 00000000..9bc5653d --- /dev/null +++ b/app/server/modules/backups/helpers/backup-lifecycle.ts @@ -0,0 +1,259 @@ +import { BadRequestError, NotFoundError } from "http-errors-enhanced"; +import { logger } from "@zerobyte/core/node"; +import type { ResticBackupOutputDto } from "@zerobyte/core/restic"; +import type { BackupSchedule, Repository, Volume } from "../../../db/schema"; +import { serverEvents } from "../../../core/events"; +import { cache, cacheKeys } from "../../../utils/cache"; +import { toErrorDetails, toMessage } from "../../../utils/errors"; +import { notificationsService } from "../../notifications/notifications.service"; +import { getOrganizationId } from "~/server/core/request-context"; +import type { BackupProgressEventDto } from "~/schemas/events-dto"; +import { calculateNextRun } from "../backup.helpers"; +import { scheduleQueries } from "../backups.queries"; +import type { BackupExecutionProgress } from "../backup-executor"; +import { repositoriesService } from "../../repositories/repositories.service"; +import { copyToMirrors, runForget } from "./backup-maintenance"; + +export interface BackupContext { + schedule: BackupSchedule; + volume: Volume; + repository: Repository; + organizationId: string; +} + +type ValidationSuccess = { + type: "success"; + context: BackupContext; +}; + +type ValidationFailure = { + type: "failure"; + error: Error; + partialContext?: Partial; +}; + +type ValidationSkipped = { + type: "skipped"; + reason: string; +}; + +export type ValidationResult = ValidationSuccess | ValidationFailure | ValidationSkipped; + +export function getBackupProgress(scheduleId: number): BackupProgressEventDto | undefined { + return cache.get(cacheKeys.backup.progress(scheduleId)); +} + +export async function validateBackupExecution(scheduleId: number, manual = false): Promise { + const organizationId = getOrganizationId(); + const result = await scheduleQueries.findById(scheduleId, organizationId); + + if (!result) { + return { type: "failure", error: new NotFoundError("Backup schedule not found") }; + } + + const { volume, repository, ...schedule } = result; + + if (!schedule.enabled && !manual) { + logger.info(`Backup schedule ${scheduleId} is disabled. Skipping execution.`); + return { type: "skipped", reason: "Backup schedule is disabled" }; + } + + if (schedule.lastBackupStatus === "in_progress") { + logger.info(`Backup schedule ${scheduleId} is already in progress. Skipping execution.`); + return { type: "skipped", reason: "Backup is already in progress" }; + } + + if (!volume) { + return { type: "failure", error: new NotFoundError("Volume not found"), partialContext: { schedule } }; + } + + if (!repository) { + return { type: "failure", error: new NotFoundError("Repository not found"), partialContext: { schedule, volume } }; + } + + if (volume.status !== "mounted") { + return { + type: "failure", + error: new BadRequestError("Volume is not mounted"), + partialContext: { schedule, volume, repository }, + }; + } + + return { + type: "success", + context: { schedule, volume, repository, organizationId }, + }; +} + +export async function handleValidationResult(scheduleId: number, result: ValidationFailure | ValidationSkipped) { + const organizationId = getOrganizationId(); + + if (result.type === "skipped") { + logger.info(`Backup execution for schedule ${scheduleId} was skipped: ${result.reason}`); + return; + } + + await handleBackupFailure(scheduleId, organizationId, result.error, result.partialContext); +} + +export function emitBackupStarted(ctx: BackupContext, scheduleId: number) { + logger.info( + `Starting backup ${ctx.schedule.name} for volume ${ctx.volume.name} to repository ${ctx.repository.name}`, + ); + + serverEvents.emit("backup:started", { + organizationId: ctx.organizationId, + scheduleId: ctx.schedule.shortId, + volumeName: ctx.volume.name, + repositoryName: ctx.repository.name, + }); + + notificationsService + .sendBackupNotification(scheduleId, "start", { + volumeName: ctx.volume.name, + repositoryName: ctx.repository.name, + scheduleName: ctx.schedule.name, + }) + .catch((error) => { + logger.error(`Failed to send backup start notification: ${toMessage(error)}`); + }); +} + +export function updateBackupProgress(ctx: BackupContext, progress: BackupExecutionProgress) { + const progressEvent = { + scheduleId: ctx.schedule.shortId, + volumeName: ctx.volume.name, + repositoryName: ctx.repository.name, + ...progress, + }; + + cache.set(cacheKeys.backup.progress(ctx.schedule.id), progressEvent, 60 * 60); + + serverEvents.emit("backup:progress", { + organizationId: ctx.organizationId, + ...progressEvent, + }); +} + +export async function finalizeSuccessfulBackup( + ctx: BackupContext, + exitCode: number, + result: ResticBackupOutputDto | null, + warningDetails: string | null, +) { + const scheduleId = ctx.schedule.id; + const finalStatus = exitCode === 0 ? "success" : "warning"; + + if (ctx.schedule.retentionPolicy) { + void runForget(scheduleId, undefined, ctx.organizationId).catch((error) => { + logger.error(`Failed to run retention policy for schedule ${scheduleId}: ${toMessage(error)}`); + }); + } + + void copyToMirrors(scheduleId, ctx.repository, ctx.schedule.retentionPolicy, ctx.organizationId).catch((error) => { + logger.error(`Background mirror copy failed for schedule ${scheduleId}: ${toMessage(error)}`); + }); + + cache.delByPrefix(cacheKeys.repository.all(ctx.repository.id)); + + void repositoriesService.refreshRepositoryStats(ctx.repository.shortId).catch((error) => { + logger.error( + `Background repository stats refresh failed for schedule ${scheduleId} (${ctx.repository.shortId}): ${toMessage(error)}`, + ); + }); + + await scheduleQueries.updateStatus(scheduleId, ctx.organizationId, { + lastBackupAt: Date.now(), + lastBackupStatus: finalStatus, + lastBackupError: finalStatus === "warning" ? warningDetails : null, + nextBackupAt: ctx.schedule.cronExpression ? calculateNextRun(ctx.schedule.cronExpression) : null, + }); + + if (finalStatus === "warning") { + logger.warn( + `Backup ${ctx.schedule.name} completed with warnings for volume ${ctx.volume.name} to repository ${ctx.repository.name}`, + ); + } else { + logger.info( + `Backup ${ctx.schedule.name} completed successfully for volume ${ctx.volume.name} to repository ${ctx.repository.name}`, + ); + } + + serverEvents.emit("backup:completed", { + organizationId: ctx.organizationId, + scheduleId: ctx.schedule.shortId, + volumeName: ctx.volume.name, + repositoryName: ctx.repository.name, + status: finalStatus, + summary: result ?? undefined, + }); + + notificationsService + .sendBackupNotification(scheduleId, finalStatus, { + volumeName: ctx.volume.name, + repositoryName: ctx.repository.name, + scheduleName: ctx.schedule.name, + summary: result ?? undefined, + }) + .catch((error) => { + logger.error(`Failed to send backup success notification: ${toMessage(error)}`); + }); +} + +export async function handleBackupFailure( + scheduleId: number, + organizationId: string, + error: unknown, + partialContext?: Partial, +) { + const errorMessage = toMessage(error); + const errorDetails = toErrorDetails(error); + + await scheduleQueries.updateStatus(scheduleId, organizationId, { + lastBackupAt: Date.now(), + lastBackupStatus: "error", + lastBackupError: errorDetails, + }); + + if (!partialContext?.schedule || !partialContext?.volume || !partialContext?.repository) { + return; + } + + const { schedule, volume, repository } = partialContext; + + logger.error( + `Backup ${schedule.name} failed for volume ${volume.name} to repository ${repository.name}: ${errorMessage}`, + ); + + serverEvents.emit("backup:completed", { + organizationId, + scheduleId: schedule.shortId, + volumeName: volume.name, + repositoryName: repository.name, + status: "error", + }); + + notificationsService + .sendBackupNotification(scheduleId, "failure", { + volumeName: volume.name, + repositoryName: repository.name, + scheduleName: schedule.name, + error: errorDetails, + }) + .catch((notifError) => { + logger.error(`Failed to send backup failure notification: ${toMessage(notifError)}`); + }); +} + +export async function handleBackupCancellation( + scheduleId: number, + organizationId: string, + message?: string, + shouldSetLastBackupAt = true, +) { + await scheduleQueries.updateStatus(scheduleId, organizationId, { + lastBackupAt: shouldSetLastBackupAt ? Date.now() : undefined, + lastBackupStatus: "warning", + lastBackupError: message ?? "Backup was stopped by the user", + }); +} diff --git a/app/server/modules/backups/helpers/backup-maintenance.ts b/app/server/modules/backups/helpers/backup-maintenance.ts new file mode 100644 index 00000000..77f6c797 --- /dev/null +++ b/app/server/modules/backups/helpers/backup-maintenance.ts @@ -0,0 +1,148 @@ +import { BadRequestError, NotFoundError } from "http-errors-enhanced"; +import { logger } from "@zerobyte/core/node"; +import type { BackupSchedule, Repository } from "../../../db/schema"; +import { restic } from "../../../core/restic"; +import { repoMutex } from "../../../core/repository-mutex"; +import { serverEvents } from "../../../core/events"; +import { cache, cacheKeys } from "../../../utils/cache"; +import { toMessage } from "../../../utils/errors"; +import { getOrganizationId } from "~/server/core/request-context"; +import { mirrorQueries, repositoryQueries, scheduleQueries } from "../backups.queries"; + +export async function runForget(scheduleId: number, repositoryId?: string, organizationIdOverride?: string) { + const organizationId = organizationIdOverride ?? getOrganizationId(); + const schedule = await scheduleQueries.findById(scheduleId, organizationId); + + if (!schedule) { + throw new NotFoundError("Backup schedule not found"); + } + + if (!schedule.retentionPolicy) { + throw new BadRequestError("No retention policy configured for this schedule"); + } + + const repository = await repositoryQueries.findById(repositoryId ?? schedule.repositoryId, organizationId); + + if (!repository) { + throw new NotFoundError("Repository not found"); + } + + logger.info(`running retention policy (forget) for schedule ${scheduleId}`); + const releaseLock = await repoMutex.acquireExclusive(repository.id, `forget:${scheduleId}`); + + try { + await restic.forget(repository.config, schedule.retentionPolicy, { tag: schedule.shortId, organizationId }); + cache.delByPrefix(cacheKeys.repository.all(repository.id)); + } finally { + releaseLock(); + } + + logger.info(`Retention policy applied successfully for schedule ${scheduleId}`); +} + +export async function copyToMirrors( + scheduleId: number, + sourceRepository: Repository, + retentionPolicy: BackupSchedule["retentionPolicy"], + organizationIdOverride?: string, +) { + const organizationId = organizationIdOverride ?? getOrganizationId(); + const schedule = await scheduleQueries.findById(scheduleId, organizationId); + + if (!schedule) { + throw new NotFoundError("Backup schedule not found"); + } + + const mirrors = await mirrorQueries.findEnabledBySchedule(scheduleId); + + if (mirrors.length === 0) { + return; + } + + logger.info(`[Background] Copying snapshots to ${mirrors.length} mirror repositories for schedule ${scheduleId}`); + + for (const mirror of mirrors) { + await copyToSingleMirror(scheduleId, schedule, sourceRepository, mirror, retentionPolicy, organizationId); + } +} + +async function copyToSingleMirror( + scheduleId: number, + schedule: BackupSchedule, + sourceRepository: Repository, + mirror: { + repositoryId: string; + repository: Repository; + }, + retentionPolicy: BackupSchedule["retentionPolicy"], + organizationId: string, +) { + try { + logger.info(`[Background] Copying to mirror repository: ${mirror.repository.name}`); + + serverEvents.emit("mirror:started", { + organizationId, + scheduleId: schedule.shortId, + repositoryId: mirror.repository.shortId, + repositoryName: mirror.repository.name, + }); + + await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, { + lastCopyStatus: "in_progress", + lastCopyError: null, + }); + + const releaseSource = await repoMutex.acquireShared(sourceRepository.id, `mirror_source:${scheduleId}`); + const releaseMirror = await repoMutex.acquireShared(mirror.repository.id, `mirror:${scheduleId}`); + + try { + await restic.copy(sourceRepository.config, mirror.repository.config, { tag: schedule.shortId, organizationId }); + cache.delByPrefix(cacheKeys.repository.all(mirror.repository.id)); + } finally { + releaseSource(); + releaseMirror(); + } + + if (retentionPolicy) { + void runForget(scheduleId, mirror.repository.id, organizationId).catch((error) => { + logger.error( + `Failed to run retention policy for mirror repository ${mirror.repository.name}: ${toMessage(error)}`, + ); + }); + } + + await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, { + lastCopyAt: Date.now(), + lastCopyStatus: "success", + lastCopyError: null, + }); + + logger.info(`[Background] Successfully copied to mirror repository: ${mirror.repository.name}`); + + serverEvents.emit("mirror:completed", { + organizationId, + scheduleId: schedule.shortId, + repositoryId: mirror.repository.shortId, + repositoryName: mirror.repository.name, + status: "success", + }); + } catch (error) { + const errorMessage = toMessage(error); + logger.error(`[Background] Failed to copy to mirror repository ${mirror.repository.name}: ${errorMessage}`); + + await mirrorQueries.updateStatus(scheduleId, mirror.repositoryId, { + lastCopyAt: Date.now(), + lastCopyStatus: "error", + lastCopyError: errorMessage, + }); + + serverEvents.emit("mirror:completed", { + organizationId, + scheduleId: schedule.shortId, + repositoryId: mirror.repository.shortId, + repositoryName: mirror.repository.name, + status: "error", + error: errorMessage, + }); + } +} diff --git a/app/server/modules/backups/helpers/backup-schedule-lookups.ts b/app/server/modules/backups/helpers/backup-schedule-lookups.ts new file mode 100644 index 00000000..32b705b1 --- /dev/null +++ b/app/server/modules/backups/helpers/backup-schedule-lookups.ts @@ -0,0 +1,23 @@ +import { NotFoundError } from "http-errors-enhanced"; +import { db } from "../../../db/db"; +import { getOrganizationId } from "~/server/core/request-context"; +import { asShortId } from "~/server/utils/branded"; + +export async function getScheduleByIdOrShortId(idOrShortId: number | string) { + const organizationId = getOrganizationId(); + const schedule = await db.query.backupSchedulesTable.findFirst({ + where: { + AND: [ + { OR: [{ id: Number(idOrShortId) }, { shortId: { eq: asShortId(String(idOrShortId)) } }] }, + { organizationId }, + ], + }, + with: { volume: true, repository: true }, + }); + + if (!schedule || !schedule.volume || !schedule.repository) { + throw new NotFoundError("Backup schedule not found"); + } + + return schedule; +} diff --git a/app/server/modules/repositories/repositories.service.ts b/app/server/modules/repositories/repositories.service.ts index 0827d5d0..1acd63ca 100644 --- a/app/server/modules/repositories/repositories.service.ts +++ b/app/server/modules/repositories/repositories.service.ts @@ -25,13 +25,13 @@ import { generateShortId } from "../../utils/id"; import { addCommonArgs, buildEnv, buildRepoUrl, cleanupTemporaryKeys } from "@zerobyte/core/restic/server"; import { restic, resticDeps } from "../../core/restic"; import { safeSpawn } from "@zerobyte/core/node"; -import { backupsService } from "../backups/backups.service"; import type { DumpPathKind, UpdateRepositoryBody } from "./repositories.dto"; import { findCommonAncestor } from "@zerobyte/core/utils"; import { prepareSnapshotDump } from "./helpers/dump"; import { executeDoctor } from "./helpers/doctor"; import type { ShortId } from "~/server/utils/branded"; import { decryptRepositoryConfig, encryptRepositoryConfig } from "./repository-config-secrets"; +import { getScheduleByIdOrShortId } from "../backups/helpers/backup-schedule-lookups"; const runningDoctors = new Map(); @@ -810,7 +810,7 @@ const getRetentionCategories = async (repositoryId: ShortId, scheduleId?: ShortI return new Map(Object.entries(cached)); } - const schedule = await backupsService.getScheduleByShortId(scheduleId); + const schedule = await getScheduleByIdOrShortId(scheduleId); if (!schedule?.retentionPolicy) { return new Map(); diff --git a/app/server/utils/errors.ts b/app/server/utils/errors.ts index d1adf687..1282dbf7 100644 --- a/app/server/utils/errors.ts +++ b/app/server/utils/errors.ts @@ -1,6 +1,7 @@ import { HttpError } from "http-errors-enhanced"; import { sanitizeSensitiveData } from "@zerobyte/core/node"; import { ResticError } from "@zerobyte/core/restic"; +import { toErrorDetails as getErrorDetails, toMessage as getMessage } from "@zerobyte/core/utils"; export const handleServiceError = (error: unknown) => { if (error instanceof HttpError) { @@ -19,14 +20,9 @@ export const handleServiceError = (error: unknown) => { }; export const toMessage = (err: unknown): string => { - const message = err instanceof Error ? err.message : String(err); - return sanitizeSensitiveData(message); + return sanitizeSensitiveData(getMessage(err)); }; export const toErrorDetails = (err: unknown): string => { - if (err instanceof ResticError) { - return sanitizeSensitiveData(err.details || err.summary); - } - - return toMessage(err); + return sanitizeSensitiveData(getErrorDetails(err)); }; diff --git a/packages/core/src/utils/errors.ts b/packages/core/src/utils/errors.ts new file mode 100644 index 00000000..bd3b6c7e --- /dev/null +++ b/packages/core/src/utils/errors.ts @@ -0,0 +1,17 @@ +import { ResticError } from "../restic/error.js"; + +export const toMessage = (error: unknown) => { + if (error instanceof Error) { + return error.message; + } + + return String(error); +}; + +export const toErrorDetails = (error: unknown) => { + if (error instanceof ResticError) { + return error.details || error.summary; + } + + return toMessage(error); +}; diff --git a/packages/core/src/utils/index.ts b/packages/core/src/utils/index.ts index cb464efb..dd2d8be6 100644 --- a/packages/core/src/utils/index.ts +++ b/packages/core/src/utils/index.ts @@ -1,4 +1,5 @@ export { safeJsonParse } from "./json.js"; +export { toErrorDetails, toMessage } from "./errors.js"; export { isPathWithin, normalizeAbsolutePath } from "./path.js"; export { findCommonAncestor } from "./common-ancestor.js"; export { FILE_MODES, writeFileWithMode } from "./fs.js";