diff --git a/app/server/modules/backups/backup-executor.ts b/app/server/modules/backups/backup-executor.ts index 1332c96d..6c7ef75c 100644 --- a/app/server/modules/backups/backup-executor.ts +++ b/app/server/modules/backups/backup-executor.ts @@ -1,4 +1,3 @@ -import { Effect } from "effect"; import { runBackupLifecycle } from "@zerobyte/core/backup-hooks"; import type { BackupSchedule, Volume, Repository } from "../../db/schema"; import { config } from "../../core/config"; @@ -10,7 +9,7 @@ import { getVolumePath } from "../volumes/helpers"; import { decryptVolumeConfig } from "../volumes/volume-config-secrets"; import { decryptRepositoryConfig } from "../repositories/repository-config-secrets"; import { createBackupOptions } from "./backup.helpers"; -import { toErrorDetails } from "../../utils/errors"; +import { runEffectPromise, toErrorDetails } from "../../utils/errors"; import { BadRequestError } from "http-errors-enhanced"; const FUSE_VOLUME_BACKENDS = new Set(["rclone", "sftp", "webdav"]); @@ -90,7 +89,7 @@ const executeBackupWithoutAgent = async ( compressionMode: payload.options.compressionMode, }; - return Effect.runPromise( + return runEffectPromise( runBackupLifecycle({ restic, repositoryConfig: payload.repositoryConfig, diff --git a/app/server/modules/backups/backups.service.ts b/app/server/modules/backups/backups.service.ts index 31b862f4..a27fb711 100644 --- a/app/server/modules/backups/backups.service.ts +++ b/app/server/modules/backups/backups.service.ts @@ -28,7 +28,7 @@ import { getScheduleByIdOrShortId } from "./helpers/backup-schedule-lookups"; import { copyToMirrors, runForget, syncSnapshotsToMirror } from "./helpers/backup-maintenance"; import { restic } from "../../core/restic"; import { mirrorQueries } from "./backups.queries"; -import { toMessage } from "../../utils/errors"; +import { runEffectPromise, toMessage } from "../../utils/errors"; import { Effect } from "effect"; const listSchedules = async () => { const organizationId = getOrganizationId(); @@ -514,7 +514,7 @@ const getMirrorSyncStatus = async (scheduleIdOrShortId: number | string, mirrorS throw new NotFoundError("Mirror not found for this schedule"); } - const [sourceSnapshots, mirrorSnapshots] = await Effect.runPromise( + const [sourceSnapshots, mirrorSnapshots] = await runEffectPromise( Effect.all( [ restic.snapshots(schedule.repository.config, { tags: [schedule.shortId], organizationId }), diff --git a/app/server/modules/backups/helpers/backup-maintenance.ts b/app/server/modules/backups/helpers/backup-maintenance.ts index 4e519d9c..1416cc9b 100644 --- a/app/server/modules/backups/helpers/backup-maintenance.ts +++ b/app/server/modules/backups/helpers/backup-maintenance.ts @@ -5,10 +5,9 @@ import { restic } from "../../../core/restic"; import { repoMutex } from "../../../core/repository-mutex"; import { serverEvents } from "../../../core/events"; import { cache, cacheKeys } from "../../../utils/cache"; -import { toMessage } from "../../../utils/errors"; +import { runEffectPromise, toMessage } from "../../../utils/errors"; import { getOrganizationId } from "~/server/core/request-context"; import { mirrorQueries, repositoryQueries, scheduleQueries } from "../backups.queries"; -import { Effect } from "effect"; export async function runForget(scheduleId: number, repositoryId?: string, organizationIdOverride?: string) { const organizationId = organizationIdOverride ?? getOrganizationId(); @@ -32,7 +31,7 @@ export async function runForget(scheduleId: number, repositoryId?: string, organ const releaseLock = await repoMutex.acquireExclusive(repository.id, `forget:${scheduleId}`); try { - await Effect.runPromise( + await runEffectPromise( restic.forget(repository.config, schedule.retentionPolicy, { tag: schedule.shortId, organizationId }), ); cache.delByPrefix(cacheKeys.repository.all(repository.id)); @@ -121,7 +120,7 @@ export async function syncSnapshotsToMirror( ]); try { - await Effect.runPromise( + await runEffectPromise( restic.copy(sourceRepository.config, mirrorRepository.config, { tag: schedule.shortId, organizationId, @@ -211,7 +210,7 @@ async function copyToSingleMirror( ]); try { - await Effect.runPromise( + await runEffectPromise( restic.copy(sourceRepository.config, mirror.repository.config, { tag: schedule.shortId, organizationId, diff --git a/app/server/modules/provisioning/provisioning.ts b/app/server/modules/provisioning/provisioning.ts index 4fba565a..a801d943 100644 --- a/app/server/modules/provisioning/provisioning.ts +++ b/app/server/modules/provisioning/provisioning.ts @@ -18,9 +18,8 @@ import { mapRepositoryConfigSecrets } from "~/server/modules/repositories/reposi import { mapVolumeConfigSecrets } from "~/server/modules/volumes/volume-config-secrets"; import { BACKEND_TYPES, volumeConfigSchema, type BackendConfig } from "@zerobyte/contracts/volumes"; import { cryptoUtils } from "~/server/utils/crypto"; -import { toMessage } from "~/server/utils/errors"; +import { runEffectPromise, toMessage } from "~/server/utils/errors"; import { generateShortId } from "~/server/utils/id"; -import { Effect } from "effect"; const envSecretPrefix = "env://"; const fileSecretPrefix = "file://"; @@ -172,7 +171,7 @@ const syncProvisionedRepositories = async (repositories: ProvisionedRepository[] }); if (!repository.config.isExistingRepository) { - const result = await Effect.runPromise( + const result = await runEffectPromise( restic.init(encryptedConfig, { organizationId: repository.organizationId, timeoutMs: appConfig.serverIdleTimeout * 1000, diff --git a/app/server/modules/repositories/__tests__/repositories.controller.test.ts b/app/server/modules/repositories/__tests__/repositories.controller.test.ts index c1786fdd..7de1322a 100644 --- a/app/server/modules/repositories/__tests__/repositories.controller.test.ts +++ b/app/server/modules/repositories/__tests__/repositories.controller.test.ts @@ -328,17 +328,17 @@ describe("repositories updates", () => { }); describe("delete snapshot", () => { - test("should return 500 when restic deleteSnapshot throws ResticError", async () => { + test("should return ResticError details when restic deleteSnapshot fails", async () => { const repository = await createRepositoryRecord(session.organizationId); const { restic } = await import("~/server/core/restic"); const { ResticError } = await import("@zerobyte/core/restic"); - const deleteSnapshotSpy = vi.spyOn(restic, "deleteSnapshot").mockImplementation(() => - Effect.sync(() => { - throw new ResticError(1, "Fatal: unexpected HTTP response (403): 403 Forbidden"); - }), - ); + const deleteSnapshotSpy = vi + .spyOn(restic, "deleteSnapshot") + .mockImplementation(() => + Effect.fail(new ResticError(1, "Fatal: unexpected HTTP response (403): 403 Forbidden")), + ); try { const res = await app.request(`/api/v1/repositories/${repository.shortId}/snapshots/snap123`, { @@ -348,9 +348,10 @@ describe("repositories updates", () => { expect(res.status).toBe(500); const body = await res.json(); - expect(body.message).toBe( - "Command failed: An error occurred while executing the command.\nFatal: unexpected HTTP response (403): 403 Forbidden", - ); + expect(body).toEqual({ + message: "Command failed: An error occurred while executing the command.", + details: "Fatal: unexpected HTTP response (403): 403 Forbidden", + }); } finally { deleteSnapshotSpy.mockRestore(); } diff --git a/app/server/modules/repositories/helpers/doctor.ts b/app/server/modules/repositories/helpers/doctor.ts index aacf98f4..3fe56d8a 100644 --- a/app/server/modules/repositories/helpers/doctor.ts +++ b/app/server/modules/repositories/helpers/doctor.ts @@ -3,14 +3,13 @@ import { type DoctorStep, type DoctorResult, type RepositoryConfig } from "@zero import { z } from "zod"; import { getOrganizationId } from "~/server/core/request-context"; import { restic } from "~/server/core/restic"; -import { toMessage } from "~/server/utils/errors"; +import { runEffectPromise, toMessage } from "~/server/utils/errors"; import { safeJsonParse } from "@zerobyte/core/utils"; import { logger } from "@zerobyte/core/node"; import { db } from "~/server/db/db"; import { repositoriesTable } from "~/server/db/schema"; import { repoMutex } from "~/server/core/repository-mutex"; import { serverEvents } from "~/server/core/events"; -import { Effect } from "effect"; class AbortError extends Error { name = "AbortError"; @@ -18,7 +17,7 @@ class AbortError extends Error { const runUnlockStep = async (config: RepositoryConfig, signal?: AbortSignal) => { const orgId = getOrganizationId(); - const result = await Effect.runPromise(restic.unlock(config, { signal, organizationId: orgId })).then( + const result = await runEffectPromise(restic.unlock(config, { signal, organizationId: orgId })).then( (result) => ({ success: true, message: result.message, error: null }), (error) => ({ success: false, message: null, error: toMessage(error) }), ); @@ -33,7 +32,7 @@ const runUnlockStep = async (config: RepositoryConfig, signal?: AbortSignal) => const runCheckStep = async (config: RepositoryConfig, signal: AbortSignal) => { const orgId = getOrganizationId(); - const result = await Effect.runPromise( + const result = await runEffectPromise( restic.check(config, { readData: false, signal, organizationId: orgId }), ).then( (result) => result, @@ -50,7 +49,7 @@ const runCheckStep = async (config: RepositoryConfig, signal: AbortSignal) => { const runRepairIndexStep = async (config: RepositoryConfig, signal: AbortSignal) => { const orgId = getOrganizationId(); - const result = await Effect.runPromise(restic.repairIndex(config, { signal, organizationId: orgId })).then( + const result = await runEffectPromise(restic.repairIndex(config, { signal, organizationId: orgId })).then( (result) => ({ success: true, output: result.output, error: null }), (error) => ({ success: false, output: null, error: toMessage(error) }), ); diff --git a/app/server/modules/repositories/repositories.service.ts b/app/server/modules/repositories/repositories.service.ts index 9b5d6ee5..61eeaeb4 100644 --- a/app/server/modules/repositories/repositories.service.ts +++ b/app/server/modules/repositories/repositories.service.ts @@ -21,7 +21,7 @@ import { repoMutex } from "../../core/repository-mutex"; import { db } from "../../db/db"; import { repositoriesTable, type Repository } from "../../db/schema"; import { cache, cacheKeys } from "../../utils/cache"; -import { toMessage } from "../../utils/errors"; +import { runEffectPromise, toMessage } from "../../utils/errors"; import { generateShortId } from "../../utils/id"; import { addCommonArgs, buildEnv, buildRepoUrl, cleanupTemporaryKeys } from "@zerobyte/core/restic/server"; import { restic, resticDeps } from "../../core/restic"; @@ -102,13 +102,13 @@ const createRepository = async (name: string, config: RepositoryConfig, compress let error: string | null = null; if (config.isExistingRepository) { - const result = await Effect.runPromise(restic.snapshots(encryptedConfig, { organizationId })) + const result = await runEffectPromise(restic.snapshots(encryptedConfig, { organizationId })) .then(() => ({ error: null })) .catch((error) => ({ error })); error = result.error; } else { - const initResult = await Effect.runPromise( + const initResult = await runEffectPromise( restic.init(encryptedConfig, { organizationId, timeoutMs: appConfig.serverIdleTimeout * 1000, @@ -147,7 +147,7 @@ const getRepository = async (shortId: ShortId) => { const runAndStoreRepositoryStats = async (repository: Repository): Promise => { const releaseLock = await repoMutex.acquireShared(repository.id, "stats"); try { - const stats = await Effect.runPromise( + const stats = await runEffectPromise( restic.stats(repository.config, { organizationId: repository.organizationId }), ); @@ -235,11 +235,11 @@ const listSnapshots = async (shortId: ShortId, backupId?: ShortId) => { let snapshots = []; if (backupId) { - snapshots = await Effect.runPromise( + snapshots = await runEffectPromise( restic.snapshots(repository.config, { tags: [backupId], organizationId }), ); } else { - snapshots = await Effect.runPromise(restic.snapshots(repository.config, { organizationId })); + snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId })); } cache.set(cacheKey, snapshots); @@ -286,7 +286,7 @@ const listSnapshotFiles = async ( const releaseLock = await repoMutex.acquireShared(repository.id, `ls:${snapshotId}`); try { - const result = await Effect.runPromise( + const result = await runEffectPromise( restic.ls(repository.config, snapshotId, path, { organizationId, offset, limit }), ); @@ -368,7 +368,7 @@ const restoreSnapshot = async ( snapshotId, }); - const result = await Effect.runPromise( + const result = await runEffectPromise( restic.restore(repository.config, snapshotId, target, { basePath, ...options, @@ -448,7 +448,7 @@ const dumpSnapshot = async (shortId: ShortId, snapshotId: string, path?: string, } } - dumpStream = await Effect.runPromise(restic.dump(repository.config, preparedDump.snapshotRef, dumpOptions)); + dumpStream = await runEffectPromise(restic.dump(repository.config, preparedDump.snapshotRef, dumpOptions)); serverEvents.emit("dump:started", { organizationId, @@ -485,7 +485,7 @@ const getSnapshotDetails = async (shortId: ShortId, snapshotId: string) => { if (!snapshots) { const releaseLock = await repoMutex.acquireShared(repository.id, `snapshot_details:${snapshotId}`); try { - snapshots = await Effect.runPromise(restic.snapshots(repository.config, { organizationId })); + snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId })); cache.set(cacheKey, snapshots); } finally { releaseLock(); @@ -513,7 +513,7 @@ const checkHealth = async (shortId: ShortId) => { const releaseLock = await repoMutex.acquireExclusive(repository.id, "check"); try { - const { hasErrors, error } = await Effect.runPromise(restic.check(repository.config, { organizationId })); + const { hasErrors, error } = await runEffectPromise(restic.check(repository.config, { organizationId })); await db .update(repositoriesTable) @@ -611,7 +611,7 @@ const deleteSnapshot = async (shortId: ShortId, snapshotId: string) => { const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:${snapshotId}`); try { - await Effect.runPromise(restic.deleteSnapshot(repository.config, snapshotId, { organizationId })); + await runEffectPromise(restic.deleteSnapshot(repository.config, snapshotId, { organizationId })); cache.delByPrefix(cacheKeys.repository.all(repository.id)); void runAndStoreRepositoryStats(repository).catch((error) => { logger.error( @@ -634,7 +634,7 @@ const deleteSnapshots = async (shortId: ShortId, snapshotIds: string[]) => { let shouldRefreshStats = false; const releaseLock = await repoMutex.acquireExclusive(repository.id, `delete:bulk`); try { - await Effect.runPromise(restic.deleteSnapshots(repository.config, snapshotIds, { organizationId })); + await runEffectPromise(restic.deleteSnapshots(repository.config, snapshotIds, { organizationId })); cache.delByPrefix(cacheKeys.repository.all(repository.id)); shouldRefreshStats = true; } finally { @@ -666,7 +666,7 @@ const tagSnapshots = async ( const releaseLock = await repoMutex.acquireExclusive(repository.id, `tag:bulk`); try { - await Effect.runPromise(restic.tagSnapshots(repository.config, snapshotIds, tags, { organizationId })); + await runEffectPromise(restic.tagSnapshots(repository.config, snapshotIds, tags, { organizationId })); cache.delByPrefix(cacheKeys.repository.all(repository.id)); } finally { releaseLock(); @@ -685,7 +685,7 @@ const refreshSnapshots = async (shortId: ShortId) => { const releaseLock = await repoMutex.acquireShared(repository.id, "refresh"); try { - const snapshots = await Effect.runPromise(restic.snapshots(repository.config, { organizationId })); + const snapshots = await runEffectPromise(restic.snapshots(repository.config, { organizationId })); const cacheKey = cacheKeys.repository.snapshots(repository.id); cache.set(cacheKey, snapshots); @@ -784,7 +784,7 @@ const unlockRepository = async (shortId: ShortId) => { const releaseLock = await repoMutex.acquireExclusive(repository.id, "unlock"); try { - const result = await Effect.runPromise(restic.unlock(repository.config, { organizationId })); + const result = await runEffectPromise(restic.unlock(repository.config, { organizationId })); return result; } finally { releaseLock(); @@ -847,7 +847,7 @@ const getRetentionCategories = async (repositoryId: ShortId, scheduleId?: ShortI return new Map(); } - const dryRunResults = await Effect.runPromise( + const dryRunResults = await runEffectPromise( restic.forget(repository.config, schedule.retentionPolicy, { tag: scheduleId, organizationId: getOrganizationId(), diff --git a/app/server/utils/errors.ts b/app/server/utils/errors.ts index 03d6742e..a424ea7a 100644 --- a/app/server/utils/errors.ts +++ b/app/server/utils/errors.ts @@ -2,6 +2,7 @@ import { HttpError } from "http-errors-enhanced"; import { sanitizeSensitiveData } from "@zerobyte/core/node"; import { ResticError } from "@zerobyte/core/restic"; import { toErrorDetails as getErrorDetails, toMessage as getMessage } from "@zerobyte/core/utils"; +import { Cause, Effect, Exit, Option } from "effect"; const formatAllowedHostsMessage = (message: string) => { const referencesAllowedHosts = /\ballowed\s+hosts?\b|\ballowedHosts\b/i.test(message); @@ -48,3 +49,13 @@ export const toMessage = (err: unknown): string => { export const toErrorDetails = (err: unknown): string => { return sanitizeSensitiveData(getErrorDetails(err)); }; + +export const runEffectPromise = (effect: Effect.Effect) => { + return Effect.runPromiseExit(effect).then((exit) => { + if (Exit.isSuccess(exit)) { + return exit.value; + } + + throw Option.getOrUndefined(Cause.failureOption(exit.cause)) ?? Cause.squash(exit.cause); + }); +};