refactor: wrap backup error in a tagged effect error

app/server/modules/backups/backup-executor.ts

@@ -56,26 +56,28 @@ export const backupExecutor = {
 			const volumePath = getVolumePath(volume);
 			const backupOptions = createBackupOptions(schedule, volumePath, signal);
 
-			const result = await Effect.runPromise(
+			const execution = await Effect.runPromise(
-				restic.backup(repository.config, volumePath, {
+				restic
-					...backupOptions,
+					.backup(repository.config, volumePath, {
-					compressionMode: repository.compressionMode ?? "auto",
+						...backupOptions,
-					organizationId,
+						compressionMode: repository.compressionMode ?? "auto",
-					onProgress,
+						organizationId,
-				}),
+						onProgress,
+					})
+					.pipe(
+						Effect.map((result) => ({ success: true as const, result })),
+						Effect.catchAll((error) => Effect.succeed({ success: false as const, error })),
+					),
 			);
 
-			return {
+			if (!execution.success) {
-				status: "completed",
+				throw execution.error;
-				exitCode: result.exitCode,
+			}
-				result: result.result,
+
-				warningDetails: result.warningDetails,
+			const { exitCode, result, warningDetails } = execution.result;
-			} satisfies BackupExecutionResult;
+			return { status: "completed", exitCode, result, warningDetails };
 		} catch (error) {
-			return {
+			return { status: "failed", error };
-				status: "failed",
-				error,
-			} satisfies BackupExecutionResult;
 		}
 	},
 	cancel: (scheduleId: number) => {

apps/agent/package.json

@@ -2,7 +2,7 @@
 	"name": "agent",
 	"private": true,
 	"type": "module",
-	"module": "index.ts",
+	"module": "src/index.ts",
 	"scripts": {
 		"tsc": "tsc --noEmit"
 	},
@@ -15,6 +15,6 @@
 		"@types/bun": "latest"
 	},
 	"peerDependencies": {
-		"typescript": "^5"
+		"typescript": "^6"
 	}
 }

apps/agent/src/commands/backup-cancel.ts

@@ -3,8 +3,8 @@ import { type BackupCancelPayload } from "@zerobyte/contracts/agent-protocol";
 import { logger } from "@zerobyte/core/node";
 import type { ControllerCommandContext } from "../context";
 
-export const handleBackupCancelCommand = (context: ControllerCommandContext, payload: BackupCancelPayload) =>
+export const handleBackupCancelCommand = (context: ControllerCommandContext, payload: BackupCancelPayload) => {
-	Effect.gen(function* () {
+	return Effect.gen(function* () {
 		const running = yield* context.getRunningJob(payload.jobId);
 		if (!running) {
 			logger.warn(`Backup ${payload.jobId} is not running`);
@@ -18,3 +18,4 @@ export const handleBackupCancelCommand = (context: ControllerCommandContext, pay
 
 		running.abortController.abort();
 	});
+};

apps/agent/src/commands/backup-run.ts

@@ -6,8 +6,8 @@ import { createRestic } from "@zerobyte/core/restic/server";
 import { toErrorDetails, toMessage } from "@zerobyte/core/utils";
 import type { ControllerCommandContext } from "../context";
 
-export const handleBackupRunCommand = (context: ControllerCommandContext, payload: BackupRunPayload) =>
+export const handleBackupRunCommand = (context: ControllerCommandContext, payload: BackupRunPayload) => {
-	Effect.fork(
+	return Effect.fork(
 		Effect.gen(function* () {
 			const existing = yield* context.getRunningJob(payload.jobId);
 			if (existing) {
@@ -110,3 +110,4 @@ export const handleBackupRunCommand = (context: ControllerCommandContext, payloa
 				);
 		}),
 	).pipe(Effect.asVoid);
+};

apps/agent/src/commands/heartbeat-ping.ts

@@ -3,9 +3,10 @@ import type { ControllerCommandContext } from "../context";
 
 type HeartbeatPingPayload = Extract<ControllerMessage, { type: "heartbeat.ping" }>["payload"];
 
-export const handleHeartbeatPingCommand = (context: ControllerCommandContext, payload: HeartbeatPingPayload) =>
+export const handleHeartbeatPingCommand = (context: ControllerCommandContext, payload: HeartbeatPingPayload) => {
-	context.offerOutbound(
+	return context.offerOutbound(
 		createAgentMessage("heartbeat.pong", {
 			sentAt: payload.sentAt,
 		}),
 	);
+};

apps/agent/src/controller-session.ts

@@ -106,11 +106,6 @@ export const createControllerSession = (ws: WebSocket): ControllerSession => {
 			});
 		},
 		onMessage: (data) => {
-			if (typeof data !== "string") {
-				logger.warn("Agent received a non-text message");
-				return;
-			}
-
 			void Effect.runPromise(offerInbound(data as ControllerWireMessage)).catch((error) => {
 				logger.error(`Failed to queue inbound message: ${toMessage(error)}`);
 			});

apps/agent/src/index.ts

@@ -3,12 +3,29 @@ import { createControllerSession, type ControllerSession } from "./controller-se
 
 const controllerUrl = process.env.ZEROBYTE_CONTROLLER_URL;
 const agentToken = process.env.ZEROBYTE_AGENT_TOKEN;
+const RECONNECT_DELAY_MS = 1000;
 
 class Agent {
 	private ws: WebSocket | null = null;
 	private controllerSession: ControllerSession | null = null;
+	private reconnectTimeout: ReturnType<typeof setTimeout> | null = null;
+
+	private scheduleReconnect() {
+		if (this.reconnectTimeout) {
+			return;
+		}
+
+		this.reconnectTimeout = setTimeout(() => {
+			this.reconnectTimeout = null;
+			this.connect();
+		}, RECONNECT_DELAY_MS);
+	}
 
 	connect() {
+		if (this.ws) {
+			return;
+		}
+
 		if (!controllerUrl) {
 			throw new Error("Env variable ZEROBYTE_CONTROLLER_URL is not set");
 		}
@@ -36,6 +53,7 @@ class Agent {
 			this.controllerSession = null;
 			this.ws = null;
 			logger.info("Agent disconnected from controller");
+			this.scheduleReconnect();
 		};
 		this.ws.onerror = (error) => {
 			logger.error("Agent encountered an error:", error);
@@ -43,5 +61,7 @@ class Agent {
 	}
 }
 
-const agent = new Agent();
+if (import.meta.main) {
-agent.connect();
+	const agent = new Agent();
+	agent.connect();
+}

bun.lock

@@ -128,7 +128,7 @@
         "@types/bun": "latest",
       },
       "peerDependencies": {
-        "typescript": "^5",
+        "typescript": "^6",
       },
     },
     "packages/contracts": {

packages/core/src/restic/commands/backup.ts

@@ -1,7 +1,7 @@
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
-import { Effect } from "effect";
+import { Data, Effect } from "effect";
 import { throttle } from "es-toolkit";
 import type { CompressionMode, RepositoryConfig } from "../schemas";
 import { type ResticBackupProgressDto, resticBackupOutputSchema, resticBackupProgressSchema } from "../restic-dto";
@@ -13,6 +13,12 @@ import { validateCustomResticParams } from "../helpers/validate-custom-params";
 import { ResticError } from "../error";
 import { logger, safeSpawn } from "../../node";
 import type { ResticDeps } from "../types";
+import { toMessage } from "../../utils";
+
+class ResticBackupCommandError extends Data.TaggedError("ResticBackupCommandError")<{
+	cause: unknown;
+	message: string;
+}> {}
 
 export const backup = (
 	config: RepositoryConfig,
@@ -31,8 +37,8 @@ export const backup = (
 		customResticParams?: string[];
 	},
 	deps: ResticDeps,
-) =>
+) => {
-	Effect.tryPromise({
+	return Effect.tryPromise({
 		try: async () => {
 			const repoUrl = buildRepoUrl(config);
 			const env = await buildEnv(config, options.organizationId, deps);
@@ -214,5 +220,15 @@ export const backup = (
 				warningDetails: stderrLines.length > 0 ? stderrLines.join("\n") : null,
 			};
 		},
-		catch: (error) => (error instanceof Error ? error : new Error(String(error))),
+		catch: (error) => {
+			if (error instanceof ResticError) {
+				return error;
+			}
+
+			return new ResticBackupCommandError({
+				cause: error,
+				message: toMessage(error),
+			});
+		},
 	});
+};