diff --git a/app/server/jobs/backup-execution.ts b/app/server/jobs/backup-execution.ts index 41a37594..7328fd05 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 { backupsService } from "../modules/backups/backups.service"; +import { backupsExecutionService } from "../modules/backups/backups.execution"; import { logger } from "../utils/logger"; 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 backupsService.getSchedulesToExecute(); + const scheduleIds = await backupsExecutionService.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) { - backupsService.executeBackup(scheduleId).catch((err) => { + backupsExecutionService.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 new file mode 100644 index 00000000..a0e03207 --- /dev/null +++ b/app/server/modules/backups/__tests__/backups.execution.test.ts @@ -0,0 +1,380 @@ +import waitForExpect from "wait-for-expect"; +import { test, describe, mock, expect, beforeEach, afterEach, spyOn } from "bun:test"; +import { backupsService } from "../backups.service"; +import { backupsExecutionService } from "../backups.execution"; +import { createTestVolume } from "~/test/helpers/volume"; +import { createTestBackupSchedule } from "~/test/helpers/backup"; +import { createTestRepository } from "~/test/helpers/repository"; +import { createTestBackupScheduleMirror } from "~/test/helpers/backup-mirror"; +import { generateBackupOutput } from "~/test/helpers/restic"; +import { TEST_ORG_ID } from "~/test/helpers/organization"; +import * as context from "~/server/core/request-context"; +import * as spawnModule from "~/server/utils/spawn"; +import { restic } from "~/server/utils/restic"; +import { NotFoundError, BadRequestError } from "http-errors-enhanced"; + +const resticBackupMock = mock(() => Promise.resolve({ exitCode: 0, summary: generateBackupOutput(), error: "" })); +const resticForgetMock = mock(() => Promise.resolve({ success: true })); +const resticCopyMock = mock(() => Promise.resolve({ success: true, output: "" })); + +beforeEach(() => { + resticBackupMock.mockClear(); + resticForgetMock.mockClear(); + resticCopyMock.mockClear(); + spyOn(spawnModule, "safeSpawn").mockImplementation(resticBackupMock); + spyOn(restic, "forget").mockImplementation(resticForgetMock); + spyOn(restic, "copy").mockImplementation(resticCopyMock); + spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID); +}); + +afterEach(() => { + mock.restore(); +}); + +describe("backup execution - validation failures", () => { + test("should fail backup when volume is not mounted", async () => { + // arrange + const volume = await createTestVolume({ status: "unmounted" }); + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + }); + + // act + const result = await backupsExecutionService.validateBackupExecution(schedule.id); + + // assert + expect(result.type).toBe("failure"); + if (result.type === "failure") { + expect(result.error).toBeInstanceOf(BadRequestError); + expect(result.error.message).toBe("Volume is not mounted"); + } + expect(resticBackupMock).not.toHaveBeenCalled(); + }); + + test("should fail backup when volume does not exist", async () => { + // arrange + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: 99999, + repositoryId: repository.id, + }); + + // act + const result = await backupsExecutionService.validateBackupExecution(schedule.id); + + // assert + expect(result.type).toBe("failure"); + if (result.type === "failure") { + expect(result.error).toBeInstanceOf(NotFoundError); + expect(result.error.message).toBe("Volume not found"); + expect(result.partialContext?.schedule).toBeDefined(); + } + }); + + test("should fail backup when repository does not exist", async () => { + // arrange + const volume = await createTestVolume(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: "non-existent-repo", + }); + + // act + const result = await backupsExecutionService.validateBackupExecution(schedule.id); + + // assert + expect(result.type).toBe("failure"); + if (result.type === "failure") { + expect(result.error).toBeInstanceOf(NotFoundError); + expect(result.error.message).toBe("Repository not found"); + expect(result.partialContext?.schedule).toBeDefined(); + expect(result.partialContext?.volume).toBeDefined(); + } + }); + + test("should fail backup when schedule does not exist", async () => { + // act + const result = await backupsExecutionService.validateBackupExecution(99999); + + // assert + expect(result.type).toBe("failure"); + if (result.type === "failure") { + expect(result.error).toBeInstanceOf(NotFoundError); + expect(result.error.message).toBe("Backup schedule not found"); + } + }); +}); + +describe("stop backup", () => { + test("should stop a running backup", async () => { + // arrange + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + }); + + resticBackupMock.mockImplementation(async () => { + await new Promise((resolve) => setTimeout(resolve, 500)); + return { exitCode: 0, summary: generateBackupOutput(), error: "" }; + }); + + void backupsExecutionService.executeBackup(schedule.id); + + await waitForExpect(async () => { + const runningSchedule = await backupsService.getScheduleById(schedule.id); + expect(runningSchedule.lastBackupStatus).toBe("in_progress"); + }); + + // act + await backupsExecutionService.stopBackup(schedule.id); + + // assert + const updatedSchedule = await backupsService.getScheduleById(schedule.id); + expect(updatedSchedule.lastBackupStatus).toBe("warning"); + expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by user"); + }); + + test("should throw ConflictError when trying to stop non-running backup", async () => { + // arrange + const volume = await createTestVolume(); + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + }); + + // act & assert + expect(backupsExecutionService.stopBackup(schedule.id)).rejects.toThrow( + "No backup is currently running for this schedule", + ); + }); + + test("should throw NotFoundError when schedule does not exist", async () => { + // act & assert + expect(backupsExecutionService.stopBackup(99999)).rejects.toThrow("Backup schedule not found"); + }); +}); + +describe("retention policy - runForget", () => { + test("should execute forget with retention policy", async () => { + // arrange + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + repositoryId: repository.id, + retentionPolicy: { + keepHourly: 24, + keepDaily: 7, + keepWeekly: 4, + keepMonthly: 12, + keepYearly: 3, + }, + }); + + // act + await backupsExecutionService.runForget(schedule.id); + + // assert + expect(resticForgetMock).toHaveBeenCalledWith( + repository.config, + expect.objectContaining({ + keepHourly: 24, + keepDaily: 7, + keepWeekly: 4, + keepMonthly: 12, + keepYearly: 3, + }), + expect.objectContaining({ + tag: schedule.shortId, + organizationId: TEST_ORG_ID, + }), + ); + }); + + test("should throw BadRequestError if no retention policy configured", async () => { + // arrange + const repository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + repositoryId: repository.id, + retentionPolicy: undefined, + }); + + // act & assert + expect(backupsExecutionService.runForget(schedule.id)).rejects.toThrow( + "No retention policy configured for this schedule", + ); + }); + + test("should throw NotFoundError when schedule does not exist", async () => { + // act & assert + expect(backupsExecutionService.runForget(99999)).rejects.toThrow("Backup schedule not found"); + }); + + test("should throw NotFoundError when repository does not exist", async () => { + // arrange + const schedule = await createTestBackupSchedule({ + repositoryId: "non-existent-repo", + retentionPolicy: { + keepHourly: 24, + }, + }); + + // act & assert + expect(backupsExecutionService.runForget(schedule.id)).rejects.toThrow("Repository not found"); + }); +}); + +describe("mirror operations", () => { + test("should copy snapshots to mirror repositories", async () => { + // arrange + const volume = await createTestVolume(); + const sourceRepository = await createTestRepository(); + const mirrorRepository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: sourceRepository.id, + }); + + await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id); + + // act + await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null); + + // assert + expect(resticCopyMock).toHaveBeenCalledWith( + sourceRepository.config, + mirrorRepository.config, + expect.objectContaining({ + tag: schedule.shortId, + organizationId: TEST_ORG_ID, + }), + ); + }); + + test("should skip disabled mirrors", async () => { + // arrange + const volume = await createTestVolume(); + const sourceRepository = await createTestRepository(); + const mirrorRepository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: sourceRepository.id, + }); + + await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id, { enabled: false }); + + // act + await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null); + + // assert + expect(resticCopyMock).not.toHaveBeenCalled(); + }); + + test("should update mirror status on success", async () => { + // arrange + const volume = await createTestVolume(); + const sourceRepository = await createTestRepository(); + const mirrorRepository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: sourceRepository.id, + }); + + const mirror = await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id); + + // act + await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null); + + // assert + const mirrors = await backupsService.getMirrors(schedule.id); + const updatedMirror = mirrors.find((m) => m.id === mirror.id); + expect(updatedMirror?.lastCopyStatus).toBe("success"); + expect(updatedMirror?.lastCopyError).toBeNull(); + expect(updatedMirror?.lastCopyAt).not.toBeNull(); + }); + + test("should update mirror status on failure", async () => { + // arrange + const volume = await createTestVolume(); + const sourceRepository = await createTestRepository(); + const mirrorRepository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: sourceRepository.id, + }); + + const mirror = await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id); + + resticCopyMock.mockImplementationOnce(() => Promise.reject(new Error("Copy failed"))); + + // act + await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null); + + // assert + const mirrors = await backupsService.getMirrors(schedule.id); + const updatedMirror = mirrors.find((m) => m.id === mirror.id); + expect(updatedMirror?.lastCopyStatus).toBe("error"); + expect(updatedMirror?.lastCopyError).toBe("Copy failed"); + expect(updatedMirror?.lastCopyAt).not.toBeNull(); + }); + + test("should run forget on mirror after successful copy when retention policy exists", async () => { + // arrange + const volume = await createTestVolume(); + const sourceRepository = await createTestRepository(); + const mirrorRepository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: sourceRepository.id, + retentionPolicy: { keepHourly: 24, keepDaily: 7 }, + }); + + await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id); + + resticCopyMock.mockClear(); + resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" })); + + // act + await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy); + + await waitForExpect(() => { + expect(resticCopyMock).toHaveBeenCalled(); + }); + + // assert + expect(resticForgetMock).toHaveBeenCalledWith( + mirrorRepository.config, + expect.objectContaining({ keepHourly: 24, keepDaily: 7 }), + expect.objectContaining({ tag: schedule.shortId, organizationId: TEST_ORG_ID }), + ); + }); + + test("should not run forget on mirror when no retention policy", async () => { + // arrange + const volume = await createTestVolume(); + const sourceRepository = await createTestRepository(); + const mirrorRepository = await createTestRepository(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: sourceRepository.id, + retentionPolicy: undefined, + }); + + await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id); + + resticForgetMock.mockClear(); + + // act + await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy); + + await waitForExpect(() => { + expect(resticCopyMock).toHaveBeenCalled(); + }); + + // assert + expect(resticForgetMock).not.toHaveBeenCalled(); + }); +}); diff --git a/app/server/modules/backups/__tests__/backups.patterns.test.ts b/app/server/modules/backups/__tests__/backups.patterns.test.ts index 811daae2..841a71f6 100644 --- a/app/server/modules/backups/__tests__/backups.patterns.test.ts +++ b/app/server/modules/backups/__tests__/backups.patterns.test.ts @@ -1,5 +1,4 @@ import { test, describe, mock, expect, beforeEach, afterEach, spyOn } from "bun:test"; -import { backupsService } from "../backups.service"; import { createTestVolume } from "~/test/helpers/volume"; import { createTestBackupSchedule } from "~/test/helpers/backup"; import { createTestRepository } from "~/test/helpers/repository"; @@ -9,6 +8,7 @@ import { restic } from "~/server/utils/restic"; import path from "node:path"; import { TEST_ORG_ID } from "~/test/helpers/organization"; import * as context from "~/server/core/request-context"; +import { backupsExecutionService } from "../backups.execution"; const backupMock = mock(() => Promise.resolve({ exitCode: 0, result: JSON.parse(generateBackupOutput()) })); @@ -39,7 +39,7 @@ describe("executeBackup - include / exclude patterns", () => { }); // act - await backupsService.executeBackup(schedule.id); + await backupsExecutionService.executeBackup(schedule.id); // assert expect(backupMock).toHaveBeenCalledWith( @@ -70,7 +70,7 @@ describe("executeBackup - include / exclude patterns", () => { }); // act - await backupsService.executeBackup(schedule.id); + await backupsExecutionService.executeBackup(schedule.id); // assert expect(backupMock).toHaveBeenCalledWith( @@ -100,7 +100,7 @@ describe("executeBackup - include / exclude patterns", () => { }); // act - await backupsService.executeBackup(schedule.id); + await backupsExecutionService.executeBackup(schedule.id); // assert expect(backupMock).toHaveBeenCalledWith( @@ -124,15 +124,15 @@ describe("executeBackup - include / exclude patterns", () => { }); // act - await backupsService.executeBackup(schedule.id); + await backupsExecutionService.executeBackup(schedule.id); // assert expect(backupMock).toHaveBeenCalledWith( expect.anything(), getVolumePath(volume), - expect.not.objectContaining({ - include: expect.anything(), - exclude: expect.anything(), + expect.objectContaining({ + include: [], + exclude: [], }), ); }); diff --git a/app/server/modules/backups/__tests__/backups.queries.test.ts b/app/server/modules/backups/__tests__/backups.queries.test.ts new file mode 100644 index 00000000..2427cc3a --- /dev/null +++ b/app/server/modules/backups/__tests__/backups.queries.test.ts @@ -0,0 +1,178 @@ +import { test, describe, expect, beforeEach } from "bun:test"; +import { scheduleQueries } from "../backups.queries"; +import { createTestBackupSchedule } from "~/test/helpers/backup"; +import { createTestVolume } from "~/test/helpers/volume"; +import { createTestRepository } from "~/test/helpers/repository"; +import { TEST_ORG_ID } from "~/test/helpers/organization"; +import { faker } from "@faker-js/faker"; + +describe("scheduleQueries.findExecutable", () => { + let volume: { id: number }; + let repository: { id: string }; + + beforeEach(async () => { + volume = await createTestVolume(); + repository = await createTestRepository(); + }); + + test("should return enabled schedules with null nextBackupAt", async () => { + // arrange + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + enabled: true, + nextBackupAt: null, + lastBackupStatus: null, + }); + + // act + const result = await scheduleQueries.findExecutable(TEST_ORG_ID); + + // assert + expect(result).toContain(schedule.id); + }); + + test("should return enabled schedules with past nextBackupAt", async () => { + // arrange + const pastTime = faker.date.past().getTime(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + enabled: true, + nextBackupAt: pastTime, + lastBackupStatus: null, + }); + + // act + const result = await scheduleQueries.findExecutable(TEST_ORG_ID); + + // assert + expect(result).toContain(schedule.id); + }); + + test("should not return schedules with future nextBackupAt", async () => { + // arrange + const futureTime = faker.date.future().getTime(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + enabled: true, + nextBackupAt: futureTime, + lastBackupStatus: null, + }); + + // act + const result = await scheduleQueries.findExecutable(TEST_ORG_ID); + + // assert + expect(result).not.toContain(schedule.id); + }); + + test("should not return disabled schedules", async () => { + // arrange + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + enabled: false, + nextBackupAt: null, + lastBackupStatus: null, + }); + + // act + const result = await scheduleQueries.findExecutable(TEST_ORG_ID); + + // assert + expect(result).not.toContain(schedule.id); + }); + + test("should not return schedules with in_progress status", async () => { + // arrange + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + enabled: true, + nextBackupAt: null, + lastBackupStatus: "in_progress", + }); + + // act + const result = await scheduleQueries.findExecutable(TEST_ORG_ID); + + // assert + expect(result).not.toContain(schedule.id); + }); + + test("should return schedules with success status and past nextBackupAt", async () => { + // arrange + const pastTime = faker.date.past().getTime(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + enabled: true, + nextBackupAt: pastTime, + lastBackupStatus: "success", + }); + + // act + const result = await scheduleQueries.findExecutable(TEST_ORG_ID); + + // assert + expect(result).toContain(schedule.id); + }); + + test("should return schedules with error status and null nextBackupAt", async () => { + // arrange + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + enabled: true, + nextBackupAt: null, + lastBackupStatus: "error", + }); + + // act + const result = await scheduleQueries.findExecutable(TEST_ORG_ID); + + // assert + expect(result).toContain(schedule.id); + }); + + test("should not return schedules from other organizations", async () => { + // arrange + const otherOrgId = faker.string.uuid(); + const schedule = await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + enabled: true, + nextBackupAt: null, + lastBackupStatus: null, + organizationId: otherOrgId, + }); + + // act + const result = await scheduleQueries.findExecutable(TEST_ORG_ID); + + // assert + expect(result).not.toContain(schedule.id); + }); + + test("should only return schedule IDs", async () => { + // arrange + await createTestBackupSchedule({ + volumeId: volume.id, + repositoryId: repository.id, + enabled: true, + nextBackupAt: null, + lastBackupStatus: null, + }); + + // act + const result = await scheduleQueries.findExecutable(TEST_ORG_ID); + + // assert + expect(result.length).toBeGreaterThan(0); + for (const id of result) { + expect(typeof id).toBe("number"); + } + }); +}); diff --git a/app/server/modules/backups/__tests__/backups.service.test.ts b/app/server/modules/backups/__tests__/backups.service.test.ts index ecfbc939..5570a6ef 100644 --- a/app/server/modules/backups/__tests__/backups.service.test.ts +++ b/app/server/modules/backups/__tests__/backups.service.test.ts @@ -1,3 +1,4 @@ +import waitForExpect from "wait-for-expect"; import { test, describe, mock, expect, beforeEach, afterEach, spyOn } from "bun:test"; import { backupsService } from "../backups.service"; import { createTestVolume } from "~/test/helpers/volume"; @@ -8,6 +9,7 @@ import { faker } from "@faker-js/faker"; import * as spawnModule from "~/server/utils/spawn"; import { TEST_ORG_ID } from "~/test/helpers/organization"; import * as context from "~/server/core/request-context"; +import { backupsExecutionService } from "../backups.execution"; const resticBackupMock = mock(() => Promise.resolve({ exitCode: 0, summary: "", error: "" })); @@ -38,7 +40,7 @@ describe("execute backup", () => { ); // act - await backupsService.executeBackup(schedule.id); + await backupsExecutionService.executeBackup(schedule.id); // assert const updatedSchedule = await backupsService.getScheduleById(schedule.id); @@ -62,7 +64,7 @@ describe("execute backup", () => { }); // act - await backupsService.executeBackup(schedule.id); + await backupsExecutionService.executeBackup(schedule.id); // assert expect(resticBackupMock).not.toHaveBeenCalled(); @@ -83,7 +85,7 @@ describe("execute backup", () => { ); // act - await backupsService.executeBackup(schedule.id, true); + await backupsExecutionService.executeBackup(schedule.id, true); // assert expect(resticBackupMock).toHaveBeenCalled(); @@ -104,9 +106,13 @@ describe("execute backup", () => { }); // act - void backupsService.executeBackup(schedule.id); - await new Promise((resolve) => setTimeout(resolve, 10)); - await backupsService.executeBackup(schedule.id); + void backupsExecutionService.executeBackup(schedule.id); + + await waitForExpect(() => { + expect(resticBackupMock).toHaveBeenCalledTimes(1); + }); + + await backupsExecutionService.executeBackup(schedule.id); // assert expect(resticBackupMock).toHaveBeenCalledTimes(1); @@ -126,7 +132,7 @@ describe("execute backup", () => { ); // act - await backupsService.executeBackup(schedule.id); + await backupsExecutionService.executeBackup(schedule.id); // assert const updatedSchedule = await backupsService.getScheduleById(schedule.id); @@ -147,7 +153,7 @@ describe("execute backup", () => { ); // act - await backupsService.executeBackup(schedule.id); + await backupsExecutionService.executeBackup(schedule.id); // assert const updatedSchedule = await backupsService.getScheduleById(schedule.id); @@ -171,7 +177,7 @@ describe("getSchedulesToExecute", () => { }); // act - const schedulesToExecute = await backupsService.getSchedulesToExecute(); + const schedulesToExecute = await backupsExecutionService.getSchedulesToExecute(); // assert expect(schedulesToExecute).toContain(schedule.id); diff --git a/app/server/modules/backups/backup.helpers.ts b/app/server/modules/backups/backup.helpers.ts new file mode 100644 index 00000000..ed3d5ca5 --- /dev/null +++ b/app/server/modules/backups/backup.helpers.ts @@ -0,0 +1,44 @@ +import CronExpressionParser from "cron-parser"; +import path from "node:path"; +import type { BackupSchedule } from "~/server/db/schema"; +import { toMessage } from "~/server/utils/errors"; +import { logger } from "~/server/utils/logger"; + +export const calculateNextRun = (cronExpression: string) => { + try { + const interval = CronExpressionParser.parse(cronExpression, { + currentDate: new Date(), + tz: Intl.DateTimeFormat().resolvedOptions().timeZone, + }); + return interval.next().getTime(); + } catch (error) { + logger.error(`Failed to parse cron expression "${cronExpression}": ${toMessage(error)}`); + const fallback = new Date(); + fallback.setMinutes(fallback.getMinutes() + 1); + return fallback.getTime(); + } +}; + +export const processPattern = (pattern: string, volumePath: string) => { + const isNegated = pattern.startsWith("!"); + const p = isNegated ? pattern.slice(1) : pattern; + + const relative = path.relative(volumePath, p); + const isInsideVolume = !relative.startsWith("..") && !path.isAbsolute(relative); + + if (isInsideVolume || !p.startsWith("/")) { + return pattern; + } + + const processed = path.join(volumePath, p.slice(1)); + return isNegated ? `!${processed}` : processed; +}; + +export const createBackupOptions = (schedule: BackupSchedule, volumePath: string, signal: AbortSignal) => ({ + tags: [schedule.shortId], + oneFileSystem: schedule.oneFileSystem, + signal, + exclude: schedule.excludePatterns ? schedule.excludePatterns.map((p) => processPattern(p, volumePath)) : undefined, + excludeIfPresent: schedule.excludeIfPresent ?? undefined, + include: schedule.includePatterns ? schedule.includePatterns.map((p) => processPattern(p, volumePath)) : undefined, +}); diff --git a/app/server/modules/backups/backups.controller.ts b/app/server/modules/backups/backups.controller.ts index d99481de..84473978 100644 --- a/app/server/modules/backups/backups.controller.ts +++ b/app/server/modules/backups/backups.controller.ts @@ -42,6 +42,8 @@ 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 "~/server/utils/logger"; export const backupScheduleController = new Hono() .use(requireAuth) @@ -83,21 +85,31 @@ export const backupScheduleController = new Hono() }) .post("/:scheduleId/run", runBackupNowDto, async (c) => { const scheduleId = c.req.param("scheduleId"); - backupsService.executeBackup(Number(scheduleId), true).catch((err) => { - console.error(`Error executing manual backup for schedule ${scheduleId}:`, err); + const result = await backupsExecutionService.validateBackupExecution(Number(scheduleId), true); + + if (result.type === "failure") { + throw result.error; + } + + if (result.type === "skipped") { + return c.json({ success: true }, 200); + } + + backupsExecutionService.executeBackup(Number(scheduleId), true).catch((err) => { + logger.error(`Error executing manual backup for schedule ${scheduleId}:`, err); }); return c.json({ success: true }, 200); }) .post("/:scheduleId/stop", stopBackupDto, async (c) => { const scheduleId = c.req.param("scheduleId"); - await backupsService.stopBackup(Number(scheduleId)); + await backupsExecutionService.stopBackup(Number(scheduleId)); return c.json({ success: true }, 200); }) .post("/:scheduleId/forget", runForgetDto, async (c) => { const scheduleId = c.req.param("scheduleId"); - await backupsService.runForget(Number(scheduleId)); + await backupsExecutionService.runForget(Number(scheduleId)); return c.json({ success: true }, 200); }) diff --git a/app/server/modules/backups/backups.execution.ts b/app/server/modules/backups/backups.execution.ts new file mode 100644 index 00000000..93947fab --- /dev/null +++ b/app/server/modules/backups/backups.execution.ts @@ -0,0 +1,439 @@ +import { NotFoundError, BadRequestError, ConflictError } from "http-errors-enhanced"; +import type { BackupSchedule, Volume, Repository } from "../../db/schema"; +import { restic } from "../../utils/restic"; +import { logger } from "../../utils/logger"; +import { cache } from "../../utils/cache"; +import { getVolumePath } from "../volumes/helpers"; +import { toMessage } from "../../utils/errors"; +import { serverEvents } from "../../core/events"; +import { notificationsService } from "../notifications/notifications.service"; +import { repoMutex } from "../../core/repository-mutex"; +import { getOrganizationId } from "~/server/core/request-context"; +import { scheduleQueries, mirrorQueries, repositoryQueries } from "./backups.queries"; +import { calculateNextRun, createBackupOptions } from "./backup.helpers"; + +const runningBackups = new Map(); + +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, + 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) => { + serverEvents.emit("backup:progress", { + organizationId: ctx.organizationId, + scheduleId: ctx.schedule.id, + volumeName: ctx.volume.name, + repositoryName: ctx.repository.name, + ...progress, + }); + }, + }); + return result.exitCode; + } finally { + releaseBackupLock(); + } +}; + +const finalizeSuccessfulBackup = async (ctx: BackupContext, scheduleId: number, exitCode: number) => { + 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(`snapshots:${ctx.repository.id}:`); + + const nextBackupAt = calculateNextRun(ctx.schedule.cronExpression); + await scheduleQueries.updateStatus(scheduleId, ctx.organizationId, { + lastBackupAt: Date.now(), + lastBackupStatus: finalStatus, + lastBackupError: 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, + volumeName: ctx.volume.name, + repositoryName: ctx.repository.name, + status: finalStatus, + }); + + notificationsService + .sendBackupNotification(scheduleId, finalStatus, { + volumeName: ctx.volume.name, + repositoryName: ctx.repository.name, + scheduleName: ctx.schedule.name, + }) + .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); + + await scheduleQueries.updateStatus(scheduleId, organizationId, { + lastBackupAt: Date.now(), + lastBackupStatus: "error", + lastBackupError: errorMessage, + }); + + 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, + 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: errorMessage, + }) + .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; + emitBackupStarted(ctx, scheduleId); + + const nextBackupAt = calculateNextRun(ctx.schedule.cronExpression); + + await scheduleQueries.updateStatus(scheduleId, ctx.organizationId, { + lastBackupStatus: "in_progress", + lastBackupError: null, + nextBackupAt, + }); + + const abortController = new AbortController(); + runningBackups.set(scheduleId, abortController); + + try { + const exitCode = await runBackupOperation(ctx, abortController.signal); + await finalizeSuccessfulBackup(ctx, scheduleId, exitCode); + } catch (error) { + await handleBackupFailure(scheduleId, ctx.organizationId, error, ctx); + } finally { + runningBackups.delete(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 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(`snapshots:${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: { + id: number; + 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, + repositoryId: mirror.repositoryId, + repositoryName: mirror.repository.name, + }); + + 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(`snapshots:${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(mirror.id, { + lastCopyAt: Date.now(), + lastCopyStatus: "success", + lastCopyError: null, + }); + + logger.info(`[Background] Successfully copied to mirror repository: ${mirror.repository.name}`); + + serverEvents.emit("mirror:completed", { + organizationId, + scheduleId, + repositoryId: mirror.repositoryId, + 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(mirror.id, { + lastCopyAt: Date.now(), + lastCopyStatus: "error", + lastCopyError: errorMessage, + }); + + serverEvents.emit("mirror:completed", { + organizationId, + scheduleId, + repositoryId: mirror.repositoryId, + repositoryName: mirror.repository.name, + status: "error", + error: errorMessage, + }); + } +}; + +export const backupsExecutionService = { + executeBackup, + validateBackupExecution, + getSchedulesToExecute, + stopBackup, + runForget, + copyToMirrors, +}; diff --git a/app/server/modules/backups/backups.queries.ts b/app/server/modules/backups/backups.queries.ts new file mode 100644 index 00000000..fcd7fcec --- /dev/null +++ b/app/server/modules/backups/backups.queries.ts @@ -0,0 +1,74 @@ +import { and, eq } from "drizzle-orm"; +import { db } from "../../db/db"; +import { backupSchedulesTable, backupScheduleMirrorsTable } from "../../db/schema"; + +export type BackupStatusType = "in_progress" | "success" | "warning" | "error"; +export type MirrorStatusType = "success" | "error"; + +export const scheduleQueries = { + findById: async (scheduleId: number, organizationId: string) => { + return db.query.backupSchedulesTable.findFirst({ + where: { AND: [{ id: scheduleId }, { organizationId }] }, + with: { volume: true, repository: true }, + }); + }, + + findExecutable: async (organizationId: string) => { + const now = Date.now(); + const schedules = await db.query.backupSchedulesTable.findMany({ + where: { + AND: [ + { enabled: true }, + { OR: [{ lastBackupStatus: { ne: "in_progress" } }, { lastBackupStatus: { isNull: true } }] }, + { organizationId }, + ], + }, + }); + + return schedules.filter((s) => !s.nextBackupAt || s.nextBackupAt <= now).map((s) => s.id); + }, + + updateStatus: async ( + scheduleId: number, + organizationId: string, + status: { + lastBackupStatus?: BackupStatusType; + lastBackupAt?: number; + lastBackupError?: string | null; + nextBackupAt?: number; + }, + ) => { + return db + .update(backupSchedulesTable) + .set({ ...status, updatedAt: Date.now() }) + .where(and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId))); + }, +}; + +export const mirrorQueries = { + findEnabledBySchedule: async (scheduleId: number) => { + return db.query.backupScheduleMirrorsTable.findMany({ + where: { scheduleId, enabled: true }, + with: { repository: true }, + }); + }, + + updateStatus: async ( + mirrorId: number, + status: { + lastCopyAt: number; + lastCopyStatus: MirrorStatusType; + lastCopyError: string | null; + }, + ) => { + return db.update(backupScheduleMirrorsTable).set(status).where(eq(backupScheduleMirrorsTable.id, mirrorId)); + }, +}; + +export const repositoryQueries = { + findById: async (repositoryId: string, organizationId: string) => { + return db.query.repositoriesTable.findFirst({ + where: { AND: [{ id: repositoryId }, { organizationId }] }, + }); + }, +}; diff --git a/app/server/modules/backups/backups.service.ts b/app/server/modules/backups/backups.service.ts index ab1825a9..26a88707 100644 --- a/app/server/modules/backups/backups.service.ts +++ b/app/server/modules/backups/backups.service.ts @@ -1,61 +1,14 @@ import { and, eq } from "drizzle-orm"; import cron from "node-cron"; -import { CronExpressionParser } from "cron-parser"; import { NotFoundError, BadRequestError, ConflictError } from "http-errors-enhanced"; import { db } from "../../db/db"; -import { backupSchedulesTable, backupScheduleMirrorsTable, repositoriesTable } from "../../db/schema"; -import { restic } from "../../utils/restic"; -import { logger } from "../../utils/logger"; -import { cache } from "../../utils/cache"; -import { getVolumePath } from "../volumes/helpers"; +import { backupSchedulesTable, backupScheduleMirrorsTable } from "../../db/schema"; import type { CreateBackupScheduleBody, UpdateBackupScheduleBody, UpdateScheduleMirrorsBody } from "./backups.dto"; -import { toMessage } from "../../utils/errors"; -import { serverEvents } from "../../core/events"; -import { notificationsService } from "../notifications/notifications.service"; -import { repoMutex } from "../../core/repository-mutex"; + import { checkMirrorCompatibility, getIncompatibleMirrorError } from "~/server/utils/backend-compatibility"; -import path from "node:path"; import { generateShortId } from "~/server/utils/id"; import { getOrganizationId } from "~/server/core/request-context"; - -const runningBackups = new Map(); - -const calculateNextRun = (cronExpression: string) => { - try { - const interval = CronExpressionParser.parse(cronExpression, { - currentDate: new Date(), - tz: Intl.DateTimeFormat().resolvedOptions().timeZone, - }); - - return interval.next().getTime(); - } catch (error) { - logger.error(`Failed to parse cron expression "${cronExpression}": ${toMessage(error)}`); - const fallback = new Date(); - fallback.setMinutes(fallback.getMinutes() + 1); - return fallback.getTime(); - } -}; - -const processPattern = (pattern: string, volumePath: string) => { - let isNegated = false; - let p = pattern; - - if (p.startsWith("!")) { - isNegated = true; - p = p.slice(1); - } - - if (p.startsWith(volumePath)) { - return pattern; - } - - if (p.startsWith("/")) { - const processed = path.join(volumePath, p.slice(1)); - return isNegated ? `!${processed}` : processed; - } - - return pattern; -}; +import { calculateNextRun } from "./backup.helpers"; const listSchedules = async () => { const organizationId = getOrganizationId(); @@ -230,250 +183,6 @@ const deleteSchedule = async (scheduleId: number) => { .where(and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId))); }; -const executeBackup = async (scheduleId: number, manual = false) => { - const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { AND: [{ id: scheduleId }, { organizationId }] }, - }); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } - - if (!schedule.enabled && !manual) { - logger.info(`Backup schedule ${scheduleId} is disabled. Skipping execution.`); - return; - } - - if (schedule.lastBackupStatus === "in_progress") { - logger.info(`Backup schedule ${scheduleId} is already in progress. Skipping execution.`); - return; - } - - const volume = await db.query.volumesTable.findFirst({ - where: { AND: [{ id: schedule.volumeId }, { organizationId }] }, - }); - - if (!volume) { - throw new NotFoundError("Volume not found"); - } - - const repository = await db.query.repositoriesTable.findFirst({ - where: { AND: [{ id: schedule.repositoryId }, { organizationId }] }, - }); - - if (!repository) { - throw new NotFoundError("Repository not found"); - } - - if (volume.status !== "mounted") { - throw new BadRequestError("Volume is not mounted"); - } - - logger.info(`Starting backup ${schedule.name} for volume ${volume.name} to repository ${repository.name}`); - - serverEvents.emit("backup:started", { - organizationId, - scheduleId, - volumeName: volume.name, - repositoryName: repository.name, - }); - - notificationsService - .sendBackupNotification(scheduleId, "start", { - volumeName: volume.name, - repositoryName: repository.name, - scheduleName: schedule.name, - }) - .catch((error) => { - logger.error(`Failed to send backup start notification: ${toMessage(error)}`); - }); - - const nextBackupAt = calculateNextRun(schedule.cronExpression); - - await db - .update(backupSchedulesTable) - .set({ - lastBackupStatus: "in_progress", - updatedAt: Date.now(), - lastBackupError: null, - nextBackupAt, - }) - .where(and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId))); - - const abortController = new AbortController(); - runningBackups.set(scheduleId, abortController); - - try { - const volumePath = getVolumePath(volume); - - const backupOptions: { - exclude?: string[]; - excludeIfPresent?: string[]; - include?: string[]; - tags?: string[]; - oneFileSystem?: boolean; - signal?: AbortSignal; - } = { - tags: [schedule.shortId], - oneFileSystem: schedule.oneFileSystem, - signal: abortController.signal, - }; - - if (schedule.excludePatterns && schedule.excludePatterns.length > 0) { - backupOptions.exclude = schedule.excludePatterns.map((p) => processPattern(p, volumePath)); - } - - if (schedule.excludeIfPresent && schedule.excludeIfPresent.length > 0) { - backupOptions.excludeIfPresent = schedule.excludeIfPresent; - } - - if (schedule.includePatterns && schedule.includePatterns.length > 0) { - backupOptions.include = schedule.includePatterns.map((p) => processPattern(p, volumePath)); - } - - const releaseBackupLock = await repoMutex.acquireShared( - repository.id, - `backup:${volume.name}`, - abortController.signal, - ); - let exitCode: number; - try { - const result = await restic.backup(repository.config, volumePath, { - ...backupOptions, - compressionMode: repository.compressionMode ?? "auto", - organizationId, - onProgress: (progress) => { - const progressData = { - organizationId, - scheduleId, - volumeName: volume.name, - repositoryName: repository.name, - ...progress, - }; - serverEvents.emit("backup:progress", progressData); - }, - }); - exitCode = result.exitCode; - } finally { - releaseBackupLock(); - } - - if (schedule.retentionPolicy) { - void runForget(schedule.id).catch((error) => { - logger.error(`Failed to run retention policy for schedule ${scheduleId}: ${toMessage(error)}`); - }); - } - - void copyToMirrors(scheduleId, repository, schedule.retentionPolicy).catch((error) => { - logger.error(`Background mirror copy failed for schedule ${scheduleId}: ${toMessage(error)}`); - }); - - const finalStatus = exitCode === 0 ? "success" : "warning"; - - cache.delByPrefix(`snapshots:${repository.id}:`); - - const nextBackupAt = calculateNextRun(schedule.cronExpression); - await db - .update(backupSchedulesTable) - .set({ - lastBackupAt: Date.now(), - lastBackupStatus: finalStatus, - lastBackupError: null, - nextBackupAt: nextBackupAt, - updatedAt: Date.now(), - }) - .where(and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId))); - - if (finalStatus === "warning") { - logger.warn( - `Backup ${schedule.name} completed with warnings for volume ${volume.name} to repository ${repository.name}`, - ); - } else { - logger.info( - `Backup ${schedule.name} completed successfully for volume ${volume.name} to repository ${repository.name}`, - ); - } - - serverEvents.emit("backup:completed", { - organizationId, - scheduleId, - volumeName: volume.name, - repositoryName: repository.name, - status: finalStatus, - }); - - notificationsService - .sendBackupNotification(scheduleId, finalStatus === "success" ? "success" : "warning", { - volumeName: volume.name, - repositoryName: repository.name, - scheduleName: schedule.name, - }) - .catch((error) => { - logger.error(`Failed to send backup success notification: ${toMessage(error)}`); - }); - } catch (error) { - logger.error( - `Backup ${schedule.name} failed for volume ${volume.name} to repository ${repository.name}: ${toMessage(error)}`, - ); - - await db - .update(backupSchedulesTable) - .set({ - lastBackupAt: Date.now(), - lastBackupStatus: "error", - lastBackupError: toMessage(error), - updatedAt: Date.now(), - }) - .where(and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId))); - - serverEvents.emit("backup:completed", { - organizationId, - scheduleId, - volumeName: volume.name, - repositoryName: repository.name, - status: "error", - }); - - notificationsService - .sendBackupNotification(scheduleId, "failure", { - volumeName: volume.name, - repositoryName: repository.name, - scheduleName: schedule.name, - error: toMessage(error), - }) - .catch((notifError) => { - logger.error(`Failed to send backup failure notification: ${toMessage(notifError)}`); - }); - } finally { - runningBackups.delete(scheduleId); - } -}; - -const getSchedulesToExecute = async () => { - const organizationId = getOrganizationId(); - const now = Date.now(); - const schedules = await db.query.backupSchedulesTable.findMany({ - where: { - AND: [ - { enabled: true }, - { OR: [{ lastBackupStatus: { NOT: "in_progress" } }, { lastBackupStatus: { isNull: true } }] }, - { organizationId }, - ], - }, - }); - - const schedulesToRun: number[] = []; - - for (const schedule of schedules) { - if (!schedule.nextBackupAt || schedule.nextBackupAt <= now) { - schedulesToRun.push(schedule.id); - } - } - - return schedulesToRun; -}; - const getScheduleForVolume = async (volumeId: number) => { const organizationId = getOrganizationId(); const schedule = await db.query.backupSchedulesTable.findFirst({ @@ -484,75 +193,6 @@ const getScheduleForVolume = async (volumeId: number) => { return schedule ?? null; }; -const stopBackup = async (scheduleId: number) => { - const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { AND: [{ id: 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 db - .update(backupSchedulesTable) - .set({ - lastBackupStatus: "warning", - lastBackupError: "Backup was stopped by user", - updatedAt: Date.now(), - }) - .where(and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId))); - } -}; - -const runForget = async (scheduleId: number, repositoryId?: string) => { - const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { - AND: [{ id: 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 db.query.repositoriesTable.findFirst({ - where: { - AND: [{ id: 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(`snapshots:${repository.id}:`); - } finally { - releaseLock(); - } - - logger.info(`Retention policy applied successfully for schedule ${scheduleId}`); -}; - const getMirrors = async (scheduleId: number) => { const organizationId = getOrganizationId(); const schedule = await db.query.backupSchedulesTable.findFirst({ @@ -640,100 +280,6 @@ const updateMirrors = async (scheduleId: number, data: UpdateScheduleMirrorsBody return getMirrors(scheduleId); }; -const copyToMirrors = async ( - scheduleId: number, - sourceRepository: { id: string; config: (typeof repositoriesTable.$inferSelect)["config"] }, - retentionPolicy: (typeof backupSchedulesTable.$inferSelect)["retentionPolicy"], -) => { - const organizationId = getOrganizationId(); - const schedule = await db.query.backupSchedulesTable.findFirst({ - where: { AND: [{ id: scheduleId }, { organizationId }] }, - }); - - if (!schedule) { - throw new NotFoundError("Backup schedule not found"); - } - - const mirrors = await db.query.backupScheduleMirrorsTable.findMany({ - where: { scheduleId }, - with: { repository: true }, - }); - - const enabledMirrors = mirrors.filter((m) => m.enabled); - - if (enabledMirrors.length === 0) { - return; - } - - logger.info( - `[Background] Copying snapshots to ${enabledMirrors.length} mirror repositories for schedule ${scheduleId}`, - ); - - for (const mirror of enabledMirrors) { - try { - logger.info(`[Background] Copying to mirror repository: ${mirror.repository.name}`); - - serverEvents.emit("mirror:started", { - organizationId, - scheduleId, - repositoryId: mirror.repositoryId, - repositoryName: mirror.repository.name, - }); - - 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(`snapshots:${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 db - .update(backupScheduleMirrorsTable) - .set({ lastCopyAt: Date.now(), lastCopyStatus: "success", lastCopyError: null }) - .where(eq(backupScheduleMirrorsTable.id, mirror.id)); - - logger.info(`[Background] Successfully copied to mirror repository: ${mirror.repository.name}`); - - serverEvents.emit("mirror:completed", { - organizationId, - scheduleId, - repositoryId: mirror.repositoryId, - 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 db - .update(backupScheduleMirrorsTable) - .set({ lastCopyAt: Date.now(), lastCopyStatus: "error", lastCopyError: errorMessage }) - .where(eq(backupScheduleMirrorsTable.id, mirror.id)); - - serverEvents.emit("mirror:completed", { - organizationId, - scheduleId, - repositoryId: mirror.repositoryId, - repositoryName: mirror.repository.name, - status: "error", - error: errorMessage, - }); - } - } -}; - const getMirrorCompatibility = async (scheduleId: number) => { const organizationId = getOrganizationId(); const schedule = await db.query.backupSchedulesTable.findFirst({ @@ -793,11 +339,7 @@ export const backupsService = { createSchedule, updateSchedule, deleteSchedule, - executeBackup, - getSchedulesToExecute, getScheduleForVolume, - stopBackup, - runForget, getMirrors, updateMirrors, getMirrorCompatibility, diff --git a/app/test/helpers/backup-mirror.ts b/app/test/helpers/backup-mirror.ts new file mode 100644 index 00000000..f6a4aa60 --- /dev/null +++ b/app/test/helpers/backup-mirror.ts @@ -0,0 +1,26 @@ +import { db } from "~/server/db/db"; +import { backupScheduleMirrorsTable, type BackupScheduleMirror } from "~/server/db/schema"; +import { ensureTestOrganization } from "./organization"; + +type BackupScheduleMirrorInsert = Omit; + +export const createTestBackupScheduleMirror = async ( + scheduleId: number, + repositoryId: string, + overrides: Partial = {}, +) => { + await ensureTestOrganization(); + + const mirror = { + scheduleId, + repositoryId, + enabled: true, + lastCopyAt: null, + lastCopyStatus: null, + lastCopyError: null, + ...overrides, + }; + + const data = await db.insert(backupScheduleMirrorsTable).values(mirror).returning(); + return data[0]; +}; diff --git a/bun.lock b/bun.lock index 7e866dbd..b48f585e 100644 --- a/bun.lock +++ b/bun.lock @@ -98,6 +98,7 @@ "vite-bundle-analyzer": "^1.2.3", "vite-plugin-babel": "^1.4.1", "vite-tsconfig-paths": "^6.0.5", + "wait-for-expect": "^4.0.0", }, }, }, @@ -1655,6 +1656,8 @@ "vite-tsconfig-paths": ["vite-tsconfig-paths@6.0.5", "", { "dependencies": { "debug": "^4.1.1", "globrex": "^0.1.2", "tsconfck": "^3.0.3" }, "peerDependencies": { "vite": "*" } }, "sha512-f/WvY6ekHykUF1rWJUAbCU7iS/5QYDIugwpqJA+ttwKbxSbzNlqlE8vZSrsnxNQciUW+z6lvhlXMaEyZn9MSig=="], + "wait-for-expect": ["wait-for-expect@4.0.0", "", {}, "sha512-mcH2HYUUHhdFGHVJkgwkBxRihZO4VSuPyh6xhYHz7LEnYkcaLbTAEEsTpYiFw4UY45XdTZYYIaquuMucw9wWMw=="], + "web-streams-polyfill": ["web-streams-polyfill@3.3.3", "", {}, "sha512-d2JWLCivmZYTSIoge9MsgFCZrt571BikcWGYkjC1khllbTeDlGqZ2D8vD8E/lJa8WGWbb7Plm8/XJYV7IJHZZw=="], "webidl-conversions": ["webidl-conversions@3.0.1", "", {}, "sha512-2JAn3z8AR6rjK8Sm8orRC0h/bcl/DqL7tRPdGZ4I1CjdF+EaMLmYxBHyXuKL849eucPFhvBoxMsflfOb8kxaeQ=="], diff --git a/package.json b/package.json index cc554cb0..fdf8ae34 100644 --- a/package.json +++ b/package.json @@ -117,7 +117,8 @@ "vite": "^7.3.1", "vite-bundle-analyzer": "^1.2.3", "vite-plugin-babel": "^1.4.1", - "vite-tsconfig-paths": "^6.0.5" + "vite-tsconfig-paths": "^6.0.5", + "wait-for-expect": "^4.0.0" }, "overrides": { "esbuild": "^0.27.2"