refactor(agent): create dedicated jobs for recurring tasks

app/server/modules/volumes/volume.service.ts

@@ -23,7 +23,6 @@ import {
 	getStatFs,
 	getVolumePath,
 	type AgentVolume,
-	type BackendConfig as HostBackendConfig,
 } from "../../../../apps/agent/src/volume-host";
 import {
 	browseFilesystem as browseHostFilesystem,
@@ -72,7 +71,7 @@ const volumeForAgent = async (volume: Volume): Promise<Volume> => ({
 const volumeForHost = async (volume: Volume): Promise<AgentVolume> => ({
 	...volume,
 	shortId: volume.shortId,
-	config: (await decryptVolumeConfig(volume.config)) as HostBackendConfig,
+	config: await decryptVolumeConfig(volume.config),
 	provisioningId: volume.provisioningId ?? null,
 });
 
@@ -290,7 +289,7 @@ const updateVolume = async (shortId: ShortId, volumeData: UpdateVolumeBody) => {
 
 const testConnection = async (backendConfig: BackendConfig) => {
 	if (!config.flags.enableLocalAgent) {
-		return Effect.runPromise(testVolumeConnection(backendConfig as HostBackendConfig));
+		return Effect.runPromise(testVolumeConnection(backendConfig));
 	}
 
 	const command = await runVolumeCommand(LOCAL_AGENT_ID, { name: "volume.testConnection", backendConfig });

apps/agent/src/index.ts

@@ -1,31 +1,24 @@
 import { logger } from "@zerobyte/core/node";
+import { Fiber } from "effect";
 import { createControllerSession, type ControllerSession } from "./controller-session";
-import { cleanupDanglingVolumeMountDirectories } from "./volume-host/cleanup";
+import { startAgentJobs } from "./jobs";
 
 const controllerUrl = process.env.ZEROBYTE_CONTROLLER_URL;
 const agentToken = process.env.ZEROBYTE_AGENT_TOKEN;
 const RECONNECT_DELAY_MS = 1000;
-const VOLUME_CLEANUP_INTERVAL_MS = 60 * 60 * 1000;
 
 export class Agent {
 	private ws: WebSocket | null = null;
 	private controllerSession: ControllerSession | null = null;
 	private reconnectTimeout: ReturnType<typeof setTimeout> | null = null;
-	private volumeCleanupInterval: ReturnType<typeof setInterval> | null = null;
+	private jobFibers: Fiber.RuntimeFiber<never, never>[] | null = null;
 
-	private runVolumeCleanup() {
+	private startJobs() {
-		void cleanupDanglingVolumeMountDirectories().catch((error) => {
+		if (this.jobFibers) {
-			logger.warn(`Agent volume cleanup failed: ${error instanceof Error ? error.message : String(error)}`);
-		});
-	}
-
-	private startVolumeCleanup() {
-		if (this.volumeCleanupInterval) {
 			return;
 		}
 
-		this.runVolumeCleanup();
+		this.jobFibers = startAgentJobs();
-		this.volumeCleanupInterval = setInterval(() => this.runVolumeCleanup(), VOLUME_CLEANUP_INTERVAL_MS);
 	}
 
 	private scheduleReconnect() {
@@ -40,7 +33,7 @@ export class Agent {
 	}
 
 	connect() {
-		this.startVolumeCleanup();
+		this.startJobs();
 
 		if (this.reconnectTimeout) {
 			clearTimeout(this.reconnectTimeout);

apps/agent/src/jobs/cleanup-dangling.ts

@@ -0,0 +1,21 @@
+import { logger } from "@zerobyte/core/node";
+import { toMessage } from "@zerobyte/core/utils";
+import { Effect } from "effect";
+import { cleanupDanglingVolumeMountDirectories } from "../volume-host/cleanup";
+import type { AgentJob } from "./types";
+
+const VOLUME_CLEANUP_INTERVAL_MS = 60 * 60 * 1000;
+
+export const CleanupDanglingVolumeMountsJob: AgentJob = {
+	name: "cleanup-dangling-volume-mounts",
+	intervalMs: VOLUME_CLEANUP_INTERVAL_MS,
+	run: () =>
+		Effect.tryPromise({
+			try: cleanupDanglingVolumeMountDirectories,
+			catch: (error) => error,
+		}).pipe(
+			Effect.catchAll((error) =>
+				Effect.sync(() => logger.warn(`Agent volume cleanup failed: ${toMessage(error)}`)),
+			),
+		),
+};

apps/agent/src/jobs/index.ts

@@ -0,0 +1,23 @@
+import { logger } from "@zerobyte/core/node";
+import { Effect, Fiber } from "effect";
+import { CleanupDanglingVolumeMountsJob } from "./cleanup-dangling";
+import type { AgentJob } from "./types";
+
+const agentJobs = [CleanupDanglingVolumeMountsJob];
+
+export const startAgentJobs = (jobs: readonly AgentJob[] = agentJobs) => {
+	return jobs.map((job) =>
+		Effect.runFork(
+			Effect.forever(
+				Effect.gen(function* () {
+					yield* job.run();
+					yield* Effect.sleep(job.intervalMs);
+				}),
+			).pipe(Effect.ensuring(logger.effect.debug(`Agent job stopped: ${job.name}`))),
+		),
+	);
+};
+
+export const stopAgentJobs = (fibers: readonly Fiber.RuntimeFiber<never, never>[]) => {
+	return Effect.forEach(fibers, (fiber) => Fiber.interrupt(fiber), { discard: true });
+};

apps/agent/src/jobs/types.ts

@@ -0,0 +1,7 @@
+import type { Effect } from "effect";
+
+export type AgentJob = {
+	name: string;
+	intervalMs: number;
+	run: () => Effect.Effect<void, never, never>;
+};