refactor: split backups module
This commit is contained in:
Nicolas Meienberger
parent
cb9186adc0
commit
b0fd6fff80
7 changed files with 589 additions and 480 deletions
|
|
@ -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: [],
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -8,6 +8,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 +39,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 +63,7 @@ describe("execute backup", () => {
|
|||
});
|
||||
|
||||
// act
|
||||
await backupsService.executeBackup(schedule.id);
|
||||
await backupsExecutionService.executeBackup(schedule.id);
|
||||
|
||||
// assert
|
||||
expect(resticBackupMock).not.toHaveBeenCalled();
|
||||
|
|
@ -83,7 +84,7 @@ describe("execute backup", () => {
|
|||
);
|
||||
|
||||
// act
|
||||
await backupsService.executeBackup(schedule.id, true);
|
||||
await backupsExecutionService.executeBackup(schedule.id, true);
|
||||
|
||||
// assert
|
||||
expect(resticBackupMock).toHaveBeenCalled();
|
||||
|
|
@ -104,9 +105,9 @@ describe("execute backup", () => {
|
|||
});
|
||||
|
||||
// act
|
||||
void backupsService.executeBackup(schedule.id);
|
||||
void backupsExecutionService.executeBackup(schedule.id);
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
await backupsService.executeBackup(schedule.id);
|
||||
await backupsExecutionService.executeBackup(schedule.id);
|
||||
|
||||
// assert
|
||||
expect(resticBackupMock).toHaveBeenCalledTimes(1);
|
||||
|
|
@ -126,7 +127,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 +148,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 +172,7 @@ describe("getSchedulesToExecute", () => {
|
|||
});
|
||||
|
||||
// act
|
||||
const schedulesToExecute = await backupsService.getSchedulesToExecute();
|
||||
const schedulesToExecute = await backupsExecutionService.getSchedulesToExecute();
|
||||
|
||||
// assert
|
||||
expect(schedulesToExecute).toContain(schedule.id);
|
||||
|
|
|
|||
41
app/server/modules/backups/backup.helpers.ts
Normal file
41
app/server/modules/backups/backup.helpers.ts
Normal file
|
|
@ -0,0 +1,41 @@
|
|||
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;
|
||||
|
||||
if (p.startsWith(volumePath) || !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,
|
||||
});
|
||||
|
|
@ -42,6 +42,7 @@ import {
|
|||
} from "../notifications/notifications.dto";
|
||||
import { notificationsService } from "../notifications/notifications.service";
|
||||
import { requireAuth } from "../auth/auth.middleware";
|
||||
import { backupsExecutionService } from "./backups.execution";
|
||||
|
||||
export const backupScheduleController = new Hono()
|
||||
.use(requireAuth)
|
||||
|
|
@ -83,7 +84,17 @@ export const backupScheduleController = new Hono()
|
|||
})
|
||||
.post("/:scheduleId/run", runBackupNowDto, async (c) => {
|
||||
const scheduleId = c.req.param("scheduleId");
|
||||
backupsService.executeBackup(Number(scheduleId), true).catch((err) => {
|
||||
const result = await backupsExecutionService.validateBackupExecution(Number(scheduleId), true);
|
||||
|
||||
if (result.type === "failure") {
|
||||
throw result.error;
|
||||
}
|
||||
|
||||
if (result.type === "skipped") {
|
||||
return c.json<RunBackupNowDto>({ success: true }, 200);
|
||||
}
|
||||
|
||||
backupsExecutionService.executeBackup(Number(scheduleId), true).catch((err) => {
|
||||
console.error(`Error executing manual backup for schedule ${scheduleId}:`, err);
|
||||
});
|
||||
|
||||
|
|
@ -91,13 +102,13 @@ export const backupScheduleController = new Hono()
|
|||
})
|
||||
.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<StopBackupDto>({ 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<RunForgetDto>({ success: true }, 200);
|
||||
})
|
||||
|
|
|
|||
439
app/server/modules/backups/backups.execution.ts
Normal file
439
app/server/modules/backups/backups.execution.ts
Normal file
|
|
@ -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<number, AbortController>();
|
||||
|
||||
interface BackupContext {
|
||||
schedule: BackupSchedule;
|
||||
volume: Volume;
|
||||
repository: Repository;
|
||||
organizationId: string;
|
||||
}
|
||||
type ValidationSuccess = {
|
||||
type: "success";
|
||||
context: BackupContext;
|
||||
};
|
||||
type ValidationFailure = {
|
||||
type: "failure";
|
||||
error: Error;
|
||||
partialContext?: Partial<BackupContext>;
|
||||
};
|
||||
type ValidationSkipped = {
|
||||
type: "skipped";
|
||||
reason: string;
|
||||
};
|
||||
type ValidationResult = ValidationSuccess | ValidationFailure | ValidationSkipped;
|
||||
|
||||
const validateBackupExecution = async (scheduleId: number, manual = false): Promise<ValidationResult> => {
|
||||
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<BackupContext>,
|
||||
): Promise<void> => {
|
||||
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<void> => {
|
||||
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,
|
||||
};
|
||||
75
app/server/modules/backups/backups.queries.ts
Normal file
75
app/server/modules/backups/backups.queries.ts
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
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: { NOT: "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) => {
|
||||
const mirrors = await db.query.backupScheduleMirrorsTable.findMany({
|
||||
where: { scheduleId },
|
||||
with: { repository: true },
|
||||
});
|
||||
return mirrors.filter((m) => m.enabled);
|
||||
},
|
||||
|
||||
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 }] },
|
||||
});
|
||||
},
|
||||
};
|
||||
|
|
@ -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<number, AbortController>();
|
||||
|
||||
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,
|
||||
|
|
|
|||
Loading…
Reference in a new issue