refactor: effect

chore: contracts package

feat: agent package

app/drizzle/20260311132215_fair_rogue/migration.sql

@@ -0,0 +1,29 @@
+CREATE TABLE `agent_tokens` (
+	`id` text PRIMARY KEY,
+	`name` text NOT NULL,
+	`token_hash` text NOT NULL,
+	`token_prefix` text NOT NULL,
+	`agent_id` text NOT NULL,
+	`created_by` text NOT NULL,
+	`last_used_at` integer,
+	`revoked_at` integer,
+	`created_at` integer DEFAULT (unixepoch() * 1000) NOT NULL,
+	CONSTRAINT `fk_agent_tokens_agent_id_agents_id_fk` FOREIGN KEY (`agent_id`) REFERENCES `agents`(`id`) ON DELETE CASCADE,
+	CONSTRAINT `fk_agent_tokens_created_by_users_table_id_fk` FOREIGN KEY (`created_by`) REFERENCES `users_table`(`id`) ON DELETE CASCADE
+);
+--> statement-breakpoint
+CREATE TABLE `agents` (
+	`id` text PRIMARY KEY,
+	`name` text NOT NULL,
+	`organization_id` text NOT NULL,
+	`created_by` text NOT NULL,
+	`last_seen_at` integer,
+	`created_at` integer DEFAULT (unixepoch() * 1000) NOT NULL,
+	CONSTRAINT `fk_agents_organization_id_organization_id_fk` FOREIGN KEY (`organization_id`) REFERENCES `organization`(`id`) ON DELETE CASCADE,
+	CONSTRAINT `fk_agents_created_by_users_table_id_fk` FOREIGN KEY (`created_by`) REFERENCES `users_table`(`id`) ON DELETE CASCADE,
+	CONSTRAINT `agents_name_org_uidx` UNIQUE(`name`,`organization_id`)
+);
+--> statement-breakpoint
+CREATE UNIQUE INDEX `agent_tokens_token_hash_uidx` ON `agent_tokens` (`token_hash`);--> statement-breakpoint
+CREATE INDEX `agent_tokens_agent_id_idx` ON `agent_tokens` (`agent_id`);--> statement-breakpoint
+CREATE INDEX `agents_organization_id_idx` ON `agents` (`organization_id`);

app/server/db/relations.ts

@@ -95,6 +95,29 @@ export const relations = defineRelations(schema, (r) => ({
 			to: r.usersTable.id,
 		}),
 	},
+	agentsTable: {
+		organization: r.one.organization({
+			from: r.agentsTable.organizationId,
+			to: r.organization.id,
+			optional: false,
+		}),
+		createdByUser: r.one.usersTable({
+			from: r.agentsTable.createdBy,
+			to: r.usersTable.id,
+		}),
+		tokens: r.many.agentTokensTable(),
+	},
+	agentTokensTable: {
+		agent: r.one.agentsTable({
+			from: r.agentTokensTable.agentId,
+			to: r.agentsTable.id,
+			optional: false,
+		}),
+		createdByUser: r.one.usersTable({
+			from: r.agentTokensTable.createdBy,
+			to: r.usersTable.id,
+		}),
+	},
 	organization: {
 		users: r.many.usersTable({
 			alias: "usersTable_id_organization_id_via_member",
@@ -106,6 +129,7 @@ export const relations = defineRelations(schema, (r) => ({
 		members: r.many.member(),
 		invitations: r.many.invitation(),
 		ssoProviders: r.many.ssoProvider(),
+		agents: r.many.agentsTable(),
 	},
 	ssoProvider: {
 		user: r.one.usersTable({

app/server/db/schema.ts

@@ -410,6 +410,61 @@ export const appMetadataTable = sqliteTable("app_metadata", {
 });
 export type AppMetadata = typeof appMetadataTable.$inferSelect;
 
+/**
+ * Agents Table
+ */
+export const agentsTable = sqliteTable(
+	"agents",
+	{
+		id: text("id").primaryKey(),
+		name: text("name").notNull(),
+		organizationId: text("organization_id")
+			.notNull()
+			.references(() => organization.id, { onDelete: "cascade" }),
+		createdBy: text("created_by")
+			.notNull()
+			.references(() => usersTable.id, { onDelete: "cascade" }),
+		lastSeenAt: int("last_seen_at", { mode: "number" }),
+		createdAt: int("created_at", { mode: "number" })
+			.notNull()
+			.default(sql`(unixepoch() * 1000)`),
+	},
+	(table) => [
+		index("agents_organization_id_idx").on(table.organizationId),
+		unique("agents_name_org_uidx").on(table.name, table.organizationId),
+	],
+);
+export type Agent = typeof agentsTable.$inferSelect;
+
+/**
+ * Agent Tokens Table
+ */
+export const agentTokensTable = sqliteTable(
+	"agent_tokens",
+	{
+		id: text("id").primaryKey(),
+		name: text("name").notNull(),
+		tokenHash: text("token_hash").notNull(),
+		tokenPrefix: text("token_prefix").notNull(),
+		agentId: text("agent_id")
+			.notNull()
+			.references(() => agentsTable.id, { onDelete: "cascade" }),
+		createdBy: text("created_by")
+			.notNull()
+			.references(() => usersTable.id, { onDelete: "cascade" }),
+		lastUsedAt: int("last_used_at", { mode: "number" }),
+		revokedAt: int("revoked_at", { mode: "number" }),
+		createdAt: int("created_at", { mode: "number" })
+			.notNull()
+			.default(sql`(unixepoch() * 1000)`),
+	},
+	(table) => [
+		uniqueIndex("agent_tokens_token_hash_uidx").on(table.tokenHash),
+		index("agent_tokens_agent_id_idx").on(table.agentId),
+	],
+);
+export type AgentToken = typeof agentTokensTable.$inferSelect;
+
 export const twoFactor = sqliteTable(
 	"two_factor",
 	{

app/server/modules/agents/agent-tokens.ts

@@ -0,0 +1,87 @@
+import crypto from "node:crypto";
+import { eq } from "drizzle-orm";
+import { db } from "~/server/db/db";
+import { agentTokensTable } from "~/server/db/schema";
+import { cryptoUtils } from "~/server/utils/crypto";
+
+export const generateToken = () => {
+	return `zbk_${crypto.randomBytes(32).toString("hex")}`;
+};
+
+export const hashToken = (token: string) => {
+	return crypto.createHash("sha256").update(token).digest("hex");
+};
+
+export const deriveLocalAgentToken = async () => {
+	const derived = await cryptoUtils.deriveSecret("zerobyte:local-agent-token");
+	return `zbk_${derived}`;
+};
+
+export const createAgentToken = async ({
+	name,
+	agentId,
+	createdBy,
+}: {
+	name: string;
+	agentId: string;
+	createdBy: string;
+}) => {
+	const plaintext = generateToken();
+	const tokenHash = hashToken(plaintext);
+	const tokenPrefix = plaintext.slice(0, 12);
+
+	const id = Bun.randomUUIDv7();
+	await db.insert(agentTokensTable).values({
+		id,
+		name,
+		tokenHash,
+		tokenPrefix,
+		agentId,
+		createdBy,
+	});
+
+	return { id, name, tokenPrefix, plaintext };
+};
+
+export const validateAgentToken = async (token: string) => {
+	const localToken = await deriveLocalAgentToken();
+	if (token === localToken) {
+		return { agentId: "local", organizationId: null, agentName: "local" };
+	}
+
+	const tokenHash = hashToken(token);
+
+	const record = await db.query.agentTokensTable.findFirst({
+		where: { tokenHash, revokedAt: { isNull: true } },
+		with: { agent: true },
+	});
+
+	if (!record) return null;
+
+	await db.update(agentTokensTable).set({ lastUsedAt: Date.now() }).where(eq(agentTokensTable.id, record.id));
+
+	return {
+		agentId: record.agentId,
+		organizationId: record.agent.organizationId,
+		agentName: record.name,
+	};
+};
+
+export const revokeAgentToken = async (tokenId: string, agentId: string) => {
+	const token = await db.query.agentTokensTable.findFirst({
+		where: { id: tokenId, agentId, revokedAt: { isNull: true } },
+	});
+
+	if (!token) return false;
+
+	await db.update(agentTokensTable).set({ revokedAt: Date.now() }).where(eq(agentTokensTable.id, tokenId));
+
+	return true;
+};
+
+export const listAgentTokens = async (agentId: string) => {
+	return db.query.agentTokensTable.findMany({
+		where: { agentId },
+		columns: { id: true, name: true, tokenPrefix: true, lastUsedAt: true, revokedAt: true, createdAt: true },
+	});
+};

app/server/modules/agents/agents-manager.ts

@@ -0,0 +1,213 @@
+import { type ChildProcess, spawn } from "node:child_process";
+import path from "node:path";
+import { Effect, Exit, Ref, Scope } from "effect";
+import { logger } from "@zerobyte/core/node";
+import { config } from "../../core/config";
+import { validateAgentToken, deriveLocalAgentToken } from "./agent-tokens";
+import {
+	createControllerAgentSession,
+	type BackupDispatchPayload,
+	type ControllerAgentSession,
+} from "./controller-agent-session";
+
+type AgentConnectionData = {
+	id: string;
+	agentId: string;
+	organizationId: string | null;
+	agentName: string;
+};
+
+export const spawnLocalAgent = async () => {
+	const previousAgent = (globalThis as Record<string, unknown>).__localAgent as ChildProcess | undefined;
+	if (previousAgent) {
+		previousAgent.kill();
+	}
+
+	const agentEntryPoint = path.join(process.cwd(), "apps", "agent", "src", "index.ts");
+	const agentToken = await deriveLocalAgentToken();
+	const args = config.__prod__ ? ["run", agentEntryPoint] : ["run", "--watch", agentEntryPoint];
+
+	const localAgent = spawn("bun", args, {
+		env: {
+			PATH: process.env.PATH,
+			ZEROBYTE_CONTROLLER_URL: "ws://localhost:3001",
+			ZEROBYTE_AGENT_TOKEN: agentToken,
+		},
+		stdio: ["ignore", "pipe", "pipe"],
+	});
+
+	(globalThis as Record<string, unknown>).__localAgent = localAgent;
+
+	localAgent.stdout?.on("data", (data: Buffer) => {
+		const line = data.toString().trim();
+		if (line) logger.info(`[agent] ${line}`);
+	});
+
+	localAgent.stderr?.on("data", (data: Buffer) => {
+		const line = data.toString().trim();
+		if (line) logger.error(`[agent] ${line}`);
+	});
+
+	localAgent.on("exit", (code, signal) => {
+		logger.info(`Agent process exited with code ${code} and signal ${signal}`);
+	});
+};
+
+const createAgentManagerRuntime = () => {
+	const sessionsRef = Effect.runSync(Ref.make<Map<string, ControllerAgentSession>>(new Map()));
+	let runtimeScope: Scope.CloseableScope | null = null;
+
+	const getSessions = () => Effect.runSync(Ref.get(sessionsRef));
+	const setSessions = (sessions: Map<string, ControllerAgentSession>) => {
+		Effect.runSync(Ref.set(sessionsRef, sessions));
+	};
+
+	const closeAllSessions = () => {
+		const sessions = getSessions();
+		for (const session of sessions.values()) {
+			session.close();
+		}
+		setSessions(new Map());
+	};
+
+	const getSession = (agentId: string) => getSessions().get(agentId);
+
+	const setSession = (agentId: string, session: ControllerAgentSession) => {
+		const existingSession = getSession(agentId);
+		if (existingSession) {
+			existingSession.close();
+		}
+
+		const nextSessions = new Map(getSessions());
+		nextSessions.set(agentId, session);
+		setSessions(nextSessions);
+	};
+
+	const removeSession = (agentId: string, connectionId: string) => {
+		const session = getSession(agentId);
+		if (!session || session.connectionId !== connectionId) {
+			return;
+		}
+
+		session.close();
+		const nextSessions = new Map(getSessions());
+		nextSessions.delete(agentId);
+		setSessions(nextSessions);
+	};
+
+	const acquireServer = Effect.acquireRelease(
+		Effect.sync(() =>
+			Bun.serve<AgentConnectionData>({
+				port: 3001,
+				async fetch(req, srv) {
+					const url = new URL(req.url);
+					const token = url.searchParams.get("token");
+
+					if (!token) {
+						return new Response("Missing token", { status: 401 });
+					}
+
+					const result = await validateAgentToken(token);
+					if (!result) {
+						return new Response("Invalid or revoked token", { status: 401 });
+					}
+
+					const upgraded = srv.upgrade(req, {
+						data: {
+							id: Bun.randomUUIDv7(),
+							agentId: result.agentId,
+							organizationId: result.organizationId,
+							agentName: result.agentName,
+						},
+					});
+					if (upgraded) return undefined;
+					return new Response("WebSocket upgrade failed", { status: 400 });
+				},
+				websocket: {
+					open: (ws) => {
+						setSession(ws.data.agentId, createControllerAgentSession(ws));
+						logger.info(`Agent "${ws.data.agentName}" (${ws.data.agentId}) connected on ${ws.data.id}`);
+					},
+					message: (ws, data) => {
+						if (typeof data !== "string") {
+							logger.warn(`Ignoring non-text message from agent ${ws.data.agentId}`);
+							return;
+						}
+
+						const session = getSession(ws.data.agentId);
+						if (!session || session.connectionId !== ws.data.id) {
+							logger.warn(`No active session for agent ${ws.data.agentId} on ${ws.data.id}`);
+							return;
+						}
+
+						session.handleMessage(data);
+					},
+					close: (ws) => {
+						removeSession(ws.data.agentId, ws.data.id);
+						logger.info(`Agent "${ws.data.agentName}" (${ws.data.agentId}) disconnected`);
+					},
+				},
+			}),
+		),
+		(server) =>
+			Effect.sync(() => {
+				closeAllSessions();
+				server.stop(true);
+			}),
+	);
+
+	const stop = () => {
+		if (!runtimeScope) {
+			return;
+		}
+
+		logger.info("Stopping Agent Manager...");
+		const scope = runtimeScope;
+		runtimeScope = null;
+		Effect.runSync(Scope.close(scope, Exit.succeed(undefined)));
+	};
+
+	const start = () => {
+		if (runtimeScope) {
+			stop();
+		}
+
+		logger.info("Starting Agent Manager...");
+		const scope = Effect.runSync(Scope.make());
+
+		try {
+			const server = Effect.runSync(Scope.extend(acquireServer, scope));
+			runtimeScope = scope;
+			logger.info(`Agent Manager listening on port ${server.port}`);
+		} catch (error) {
+			Effect.runSync(Scope.close(scope, Exit.fail(error)));
+			throw error;
+		}
+	};
+
+	return {
+		start,
+		sendBackup: (agentId: string, payload: BackupDispatchPayload) => {
+			const session = getSession(agentId);
+
+			if (!session) {
+				logger.warn(`Cannot send backup command. Agent ${agentId} is not connected.`);
+				return false;
+			}
+
+			const jobId = session.sendBackup(payload);
+			logger.info(`Sent backup command ${jobId} to agent ${agentId} for schedule ${payload.scheduleId}`);
+			return true;
+		},
+		stop,
+	};
+};
+
+const previous = (globalThis as Record<string, unknown>).__agentManager as
+	| ReturnType<typeof createAgentManagerRuntime>
+	| undefined;
+
+previous?.stop();
+
+export const agentManager = createAgentManagerRuntime();
+(globalThis as Record<string, unknown>).__agentManager = agentManager;

app/server/modules/agents/controller-agent-session.ts

@@ -0,0 +1,157 @@
+import { Effect, Fiber, Queue, Ref } from "effect";
+import {
+	createControllerMessage,
+	parseAgentMessage,
+	type AgentMessage,
+	type ControllerWireMessage,
+} from "@zerobyte/contracts/agent-protocol";
+import { logger } from "@zerobyte/core/node";
+
+type AgentConnectionData = {
+	id: string;
+	agentId: string;
+	organizationId: string | null;
+	agentName: string;
+};
+
+type AgentSocket = Bun.ServerWebSocket<AgentConnectionData>;
+
+type SessionState = {
+	isReady: boolean;
+	lastSeenAt: number | null;
+	lastPongAt: number | null;
+};
+
+const toMessage = (error: unknown) => {
+	if (error instanceof Error) {
+		return error.message;
+	}
+
+	return String(error);
+};
+
+export type BackupDispatchPayload = {
+	scheduleId: string;
+	jobId?: string;
+};
+
+export type ControllerAgentSession = {
+	readonly connectionId: string;
+	handleMessage: (data: string) => void;
+	sendBackup: (payload: BackupDispatchPayload) => string;
+	isReady: () => boolean;
+	close: () => void;
+};
+
+export const createControllerAgentSession = (socket: AgentSocket): ControllerAgentSession => {
+	const outboundQueue = Effect.runSync(Queue.bounded<ControllerWireMessage>(64));
+	const state = Effect.runSync(
+		Ref.make<SessionState>({
+			isReady: false,
+			lastSeenAt: null,
+			lastPongAt: null,
+		}),
+	);
+
+	const offerOutbound = (message: ControllerWireMessage) => {
+		void Effect.runPromise(Queue.offer(outboundQueue, message)).catch((error) => {
+			logger.error(`Failed to queue outbound message for agent ${socket.data.agentId}: ${toMessage(error)}`);
+		});
+	};
+
+	const updateState = (update: (current: SessionState) => SessionState) => {
+		Effect.runSync(Ref.update(state, update));
+	};
+
+	const writerFiber = Effect.runFork(
+		Effect.forever(
+			Effect.gen(function* () {
+				const message = yield* Queue.take(outboundQueue);
+				yield* Effect.sync(() => {
+					try {
+						socket.send(message);
+					} catch (error) {
+						logger.error(
+							`Failed to send message to agent ${socket.data.agentId} on ${socket.data.id}: ${toMessage(error)}`,
+						);
+					}
+				});
+			}),
+		),
+	);
+
+	const heartbeatFiber = Effect.runFork(
+		Effect.forever(
+			Effect.gen(function* () {
+				yield* Effect.sleep("15 seconds");
+				yield* Queue.offer(
+					outboundQueue,
+					createControllerMessage("heartbeat.ping", {
+						sentAt: Date.now(),
+					}),
+				);
+			}),
+		),
+	);
+
+	const handleAgentMessage = (message: AgentMessage) => {
+		switch (message.type) {
+			case "agent.ready": {
+				updateState((current) => ({ ...current, isReady: true, lastSeenAt: Date.now() }));
+				logger.info(`Agent "${socket.data.agentName}" (${socket.data.agentId}) is ready`);
+				break;
+			}
+			case "backup.started": {
+				updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
+				logger.info(
+					`Backup ${message.payload.jobId} started on agent ${socket.data.agentId} for schedule ${message.payload.scheduleId}`,
+				);
+				break;
+			}
+			case "heartbeat.pong": {
+				updateState((current) => ({
+					...current,
+					lastSeenAt: Date.now(),
+					lastPongAt: message.payload.sentAt,
+				}));
+				break;
+			}
+		}
+	};
+
+	return {
+		connectionId: socket.data.id,
+		handleMessage: (data: string) => {
+			const parsed = parseAgentMessage(data);
+
+			if (parsed === null) {
+				logger.warn(`Invalid JSON from agent ${socket.data.agentId}`);
+				return;
+			}
+
+			if (!parsed.success) {
+				logger.warn(`Invalid agent message from ${socket.data.agentId}: ${parsed.error.message}`);
+				return;
+			}
+
+			handleAgentMessage(parsed.data);
+		},
+		sendBackup: (payload) => {
+			const jobId = payload.jobId ?? Bun.randomUUIDv7();
+			offerOutbound(
+				createControllerMessage("backup.run", {
+					jobId,
+					scheduleId: payload.scheduleId,
+				}),
+			);
+			return jobId;
+		},
+		isReady: () => Effect.runSync(Ref.get(state)).isReady,
+		close: () => {
+			updateState((current) => ({ ...current, isReady: false }));
+			void Effect.runPromise(Fiber.interrupt(writerFiber)).catch(() => {});
+			void Effect.runPromise(Fiber.interrupt(heartbeatFiber)).catch(() => {});
+			void Effect.runPromise(Queue.shutdown(outboundQueue)).catch(() => {});
+		},
+	};
+};

app/server/modules/lifecycle/bootstrap.ts

@@ -1,4 +1,5 @@
 import { runDbMigrations } from "../../db/db";
+import { agentManager, spawnLocalAgent } from "../agents/agents-manager";
 import { runMigrations } from "./migrations";
 import { startup } from "./startup";
 
@@ -8,6 +9,8 @@ const runBootstrap = async () => {
 	await runDbMigrations();
 	await runMigrations();
 	await startup();
+	agentManager.start();
+	await spawnLocalAgent();
 };
 
 export const bootstrapApplication = async () => {

apps/agent/.gitignore

@@ -0,0 +1,34 @@
+# dependencies (bun install)
+node_modules
+
+# output
+out
+dist
+*.tgz
+
+# code coverage
+coverage
+*.lcov
+
+# logs
+logs
+_.log
+report.[0-9]_.[0-9]_.[0-9]_.[0-9]_.json
+
+# dotenv environment variable files
+.env
+.env.development.local
+.env.test.local
+.env.production.local
+.env.local
+
+# caches
+.eslintcache
+.cache
+*.tsbuildinfo
+
+# IntelliJ based IDEs
+.idea
+
+# Finder (MacOS) folder config
+.DS_Store

apps/agent/package.json

@@ -0,0 +1,17 @@
+{
+	"name": "agent",
+	"private": true,
+	"type": "module",
+	"module": "index.ts",
+	"dependencies": {
+		"@zerobyte/contracts": "workspace:*",
+		"@zerobyte/core": "workspace:*",
+		"effect": "^3.18.4"
+	},
+	"devDependencies": {
+		"@types/bun": "latest"
+	},
+	"peerDependencies": {
+		"typescript": "^5"
+	}
+}

apps/agent/src/controller-session.ts

@@ -0,0 +1,116 @@
+import { Effect, Fiber, Queue } from "effect";
+import {
+	createAgentMessage,
+	parseControllerMessage,
+	type AgentWireMessage,
+	type ControllerWireMessage,
+} from "@zerobyte/contracts/agent-protocol";
+import { logger } from "@zerobyte/core/node";
+
+const toMessage = (error: unknown) => {
+	if (error instanceof Error) {
+		return error.message;
+	}
+
+	return String(error);
+};
+
+export type ControllerSession = {
+	onOpen: () => void;
+	onMessage: (data: unknown) => void;
+	close: () => void;
+};
+
+export const createControllerSession = (ws: WebSocket): ControllerSession => {
+	const outboundQueue = Effect.runSync(Queue.bounded<AgentWireMessage>(64));
+	const inboundQueue = Effect.runSync(Queue.bounded<ControllerWireMessage>(64));
+
+	const offerOutbound = (message: AgentWireMessage) => {
+		void Effect.runPromise(Queue.offer(outboundQueue, message)).catch((error) => {
+			logger.error(`Failed to queue outbound controller message: ${toMessage(error)}`);
+		});
+	};
+
+	const offerInbound = (message: ControllerWireMessage) => {
+		void Effect.runPromise(Queue.offer(inboundQueue, message)).catch((error) => {
+			logger.error(`Failed to queue inbound controller message: ${toMessage(error)}`);
+		});
+	};
+
+	const writerFiber = Effect.runFork(
+		Effect.forever(
+			Effect.gen(function* () {
+				const message = yield* Queue.take(outboundQueue);
+				yield* Effect.sync(() => {
+					try {
+						ws.send(message);
+					} catch (error) {
+						logger.error(`Failed to send controller message: ${toMessage(error)}`);
+					}
+				});
+			}),
+		),
+	);
+
+	const processorFiber = Effect.runFork(
+		Effect.forever(
+			Effect.gen(function* () {
+				const data = yield* Queue.take(inboundQueue);
+				const parsed = parseControllerMessage(data);
+
+				if (parsed === null) {
+					logger.warn("Agent received invalid JSON");
+					return;
+				}
+
+				if (!parsed.success) {
+					logger.warn(`Agent received an invalid message: ${parsed.error.message}`);
+					return;
+				}
+
+				switch (parsed.data.type) {
+					case "backup.run": {
+						logger.info(`Starting backup ${parsed.data.payload.jobId} for schedule ${parsed.data.payload.scheduleId}`);
+						yield* Queue.offer(
+							outboundQueue,
+							createAgentMessage("backup.started", {
+								jobId: parsed.data.payload.jobId,
+								scheduleId: parsed.data.payload.scheduleId,
+							}),
+						);
+						break;
+					}
+					case "heartbeat.ping": {
+						yield* Queue.offer(
+							outboundQueue,
+							createAgentMessage("heartbeat.pong", {
+								sentAt: parsed.data.payload.sentAt,
+							}),
+						);
+						break;
+					}
+				}
+			}),
+		),
+	);
+
+	return {
+		onOpen: () => {
+			offerOutbound(createAgentMessage("agent.ready", { agentId: "" }));
+		},
+		onMessage: (data) => {
+			if (typeof data !== "string") {
+				logger.warn("Agent received a non-text message");
+				return;
+			}
+
+			offerInbound(data as ControllerWireMessage);
+		},
+		close: () => {
+			void Effect.runPromise(Fiber.interrupt(writerFiber)).catch(() => {});
+			void Effect.runPromise(Fiber.interrupt(processorFiber)).catch(() => {});
+			void Effect.runPromise(Queue.shutdown(outboundQueue)).catch(() => {});
+			void Effect.runPromise(Queue.shutdown(inboundQueue)).catch(() => {});
+		},
+	};
+};

apps/agent/src/index.ts

@@ -0,0 +1,47 @@
+import { logger } from "@zerobyte/core/node";
+import { createControllerSession, type ControllerSession } from "./controller-session";
+
+const controllerUrl = process.env.ZEROBYTE_CONTROLLER_URL;
+const agentToken = process.env.ZEROBYTE_AGENT_TOKEN;
+
+class Agent {
+	private ws: WebSocket | null = null;
+	private controllerSession: ControllerSession | null = null;
+
+	connect() {
+		if (!controllerUrl) {
+			throw new Error("Env variable ZEROBYTE_CONTROLLER_URL is not set");
+		}
+
+		if (!agentToken) {
+			throw new Error("Env variable ZEROBYTE_AGENT_TOKEN is not set");
+		}
+
+		const url = new URL(controllerUrl);
+		url.searchParams.set("token", agentToken);
+
+		this.ws = new WebSocket(url.toString());
+		this.controllerSession = createControllerSession(this.ws);
+
+		this.ws.onopen = () => {
+			logger.info("Agent connected to controller");
+			this.controllerSession?.onOpen();
+		};
+
+		this.ws.onmessage = (event) => {
+			this.controllerSession?.onMessage(event.data);
+		};
+		this.ws.onclose = () => {
+			this.controllerSession?.close();
+			this.controllerSession = null;
+			this.ws = null;
+			logger.info("Agent disconnected from controller");
+		};
+		this.ws.onerror = (error) => {
+			logger.error("Agent encountered an error:", error);
+		};
+	}
+}
+
+const agent = new Agent();
+agent.connect();

apps/agent/tsconfig.json

@@ -0,0 +1,29 @@
+{
+	"compilerOptions": {
+		// Environment setup & latest features
+		"lib": ["ESNext"],
+		"target": "ESNext",
+		"module": "Preserve",
+		"moduleDetection": "force",
+		"jsx": "react-jsx",
+		"allowJs": true,
+
+		// Bundler mode
+		"moduleResolution": "bundler",
+		"allowImportingTsExtensions": true,
+		"verbatimModuleSyntax": true,
+		"noEmit": true,
+
+		// Best practices
+		"strict": true,
+		"skipLibCheck": true,
+		"noFallthroughCasesInSwitch": true,
+		"noUncheckedIndexedAccess": true,
+		"noImplicitOverride": true,
+
+		// Some stricter flags (disabled by default)
+		"noUnusedLocals": false,
+		"noUnusedParameters": false,
+		"noPropertyAccessFromIndexSignature": false
+	}
+}

bun.lock

@@ -34,6 +34,7 @@
         "@tanstack/react-router": "^1.166.8",
         "@tanstack/react-router-ssr-query": "^1.166.7",
         "@tanstack/react-start": "^1.166.9",
+        "@zerobyte/contracts": "workspace:*",
         "@zerobyte/core": "workspace:*",
         "better-auth": "^1.5.5",
         "class-variance-authority": "^0.7.1",
@@ -46,6 +47,7 @@
         "dither-plugin": "^1.1.1",
         "dotenv": "^17.3.1",
         "drizzle-orm": "^1.0.0-beta.16-ea816b6",
+        "effect": "^3.18.4",
         "es-toolkit": "^1.45.1",
         "hono": "^4.12.7",
         "hono-openapi": "^1.3.0",
@@ -110,6 +112,32 @@
         "wait-for-expect": "^4.0.0",
       },
     },
+    "apps/agent": {
+      "name": "agent",
+      "dependencies": {
+        "@zerobyte/contracts": "workspace:*",
+        "@zerobyte/core": "workspace:*",
+        "effect": "^3.18.4",
+      },
+      "devDependencies": {
+        "@types/bun": "latest",
+      },
+      "peerDependencies": {
+        "typescript": "^5",
+      },
+    },
+    "packages/contracts": {
+      "name": "@zerobyte/contracts",
+      "dependencies": {
+        "@zerobyte/core": "workspace:*",
+      },
+      "devDependencies": {
+        "@types/bun": "latest",
+      },
+      "peerDependencies": {
+        "typescript": "^5",
+      },
+    },
     "packages/core": {
       "name": "@zerobyte/core",
       "devDependencies": {
@@ -968,12 +996,16 @@
 
     "@xmldom/xmldom": ["@xmldom/xmldom@0.8.11", "", {}, "sha512-cQzWCtO6C8TQiYl1ruKNn2U6Ao4o4WBBcbL61yJl84x+j5sOWWFU9X7DpND8XZG3daDppSsigMdfAIl2upQBRw=="],
 
+    "@zerobyte/contracts": ["@zerobyte/contracts@workspace:packages/contracts"],
+
     "@zerobyte/core": ["@zerobyte/core@workspace:packages/core"],
 
     "abort-controller": ["abort-controller@3.0.0", "", { "dependencies": { "event-target-shim": "^5.0.0" } }, "sha512-h8lQ8tacZYnR3vNQTgibj+tODHI5/+l06Au2Pcriv/Gmet0eaj4TwWH41sO9wnHDiQsEj19q0drzdWdeAHtweg=="],
 
     "acorn": ["acorn@8.16.0", "", { "bin": { "acorn": "bin/acorn" } }, "sha512-UVJyE9MttOsBQIDKw1skb9nAwQuR5wuGD3+82K6JgJlm/Y+KI92oNsMNGZCYdDsVtRHSak0pcV5Dno5+4jh9sw=="],
 
+    "agent": ["agent@workspace:apps/agent"],
+
     "agent-base": ["agent-base@7.1.4", "", {}, "sha512-MnA+YT8fwfJPgBx3m60MNqakm30XOkyIoH1y6huTQvC0PwZG7ki8NacLBcrPbNoo8vEZy7Jpuk7+jMO+CUovTQ=="],
 
     "ansi-colors": ["ansi-colors@4.1.3", "", {}, "sha512-/6w/C21Pm1A7aZitlI5Ni/2J6FFQN8i1Cvz3kHABAAbw93v/NlvKdVOqz7CCWz/3iv/JplRSEEZ83XION15ovw=="],

package.json

@@ -59,6 +59,7 @@
 		"@tanstack/react-router": "^1.166.8",
 		"@tanstack/react-router-ssr-query": "^1.166.7",
 		"@tanstack/react-start": "^1.166.9",
+		"@zerobyte/contracts": "workspace:*",
 		"@zerobyte/core": "workspace:*",
 		"better-auth": "^1.5.5",
 		"class-variance-authority": "^0.7.1",
@@ -71,6 +72,7 @@
 		"dither-plugin": "^1.1.1",
 		"dotenv": "^17.3.1",
 		"drizzle-orm": "^1.0.0-beta.16-ea816b6",
+		"effect": "^3.18.4",
 		"es-toolkit": "^1.45.1",
 		"hono": "^4.12.7",
 		"hono-openapi": "^1.3.0",

packages/contracts/.gitignore

@@ -0,0 +1,34 @@
+# dependencies (bun install)
+node_modules
+
+# output
+out
+dist
+*.tgz
+
+# code coverage
+coverage
+*.lcov
+
+# logs
+logs
+_.log
+report.[0-9]_.[0-9]_.[0-9]_.[0-9]_.json
+
+# dotenv environment variable files
+.env
+.env.development.local
+.env.test.local
+.env.production.local
+.env.local
+
+# caches
+.eslintcache
+.cache
+*.tsbuildinfo
+
+# IntelliJ based IDEs
+.idea
+
+# Finder (MacOS) folder config
+.DS_Store

packages/contracts/README.md

@@ -0,0 +1,15 @@
+# contracts
+
+To install dependencies:
+
+```bash
+bun install
+```
+
+To run:
+
+```bash
+bun run index.ts
+```
+
+This project was created using `bun init` in bun v1.3.10. [Bun](https://bun.com) is a fast all-in-one JavaScript runtime.

packages/contracts/package.json

@@ -0,0 +1,24 @@
+{
+	"name": "@zerobyte/contracts",
+	"private": true,
+	"type": "module",
+	"exports": {
+		"./agent-protocol": {
+			"types": "./src/agent-protocol.ts",
+			"import": "./src/agent-protocol.ts",
+			"default": "./src/agent-protocol.ts"
+		}
+	},
+	"scripts": {
+		"tsc": "tsc --noEmit"
+	},
+	"dependencies": {
+		"@zerobyte/core": "workspace:*"
+	},
+	"devDependencies": {
+		"@types/bun": "latest"
+	},
+	"peerDependencies": {
+		"typescript": "^5"
+	}
+}

packages/contracts/src/agent-protocol.ts

@@ -0,0 +1,112 @@
+import { z } from "zod";
+import { safeJsonParse } from "@zerobyte/core/utils";
+
+const backupCommandSchema = z
+	.object({
+		type: z.literal("backup.run"),
+		payload: z.object({ jobId: z.string(), scheduleId: z.string() }),
+	})
+	.strict();
+
+const heartbeatPingSchema = z
+	.object({
+		type: z.literal("heartbeat.ping"),
+		payload: z.object({ sentAt: z.number() }),
+	})
+	.strict();
+
+const agentReadySchema = z
+	.object({
+		type: z.literal("agent.ready"),
+		payload: z.object({ agentId: z.string() }),
+	})
+	.strict();
+
+const backupStartedSchema = z
+	.object({
+		type: z.literal("backup.started"),
+		payload: z.object({ jobId: z.string(), scheduleId: z.string() }),
+	})
+	.strict();
+
+const heartbeatPongSchema = z
+	.object({
+		type: z.literal("heartbeat.pong"),
+		payload: z.object({ sentAt: z.number() }),
+	})
+	.strict();
+
+const controllerMessageSchema = z.discriminatedUnion("type", [backupCommandSchema, heartbeatPingSchema]);
+const agentMessageSchema = z.discriminatedUnion("type", [agentReadySchema, backupStartedSchema, heartbeatPongSchema]);
+
+export type BackupCommandPayload = z.infer<typeof backupCommandSchema>["payload"];
+export type ControllerMessage = z.infer<typeof controllerMessageSchema>;
+export type AgentMessage = z.infer<typeof agentMessageSchema>;
+
+type Brand<TValue, TBrand extends string> = TValue & {
+	readonly __brand: TBrand;
+};
+
+type MessageSender = {
+	send(message: string): unknown;
+};
+
+export type ControllerWireMessage = Brand<string, "ControllerWireMessage">;
+export type AgentWireMessage = Brand<string, "AgentWireMessage">;
+
+type PayloadForMessage<TMessage extends { type: string; payload: unknown }, TType extends TMessage["type"]> = Extract<
+	TMessage,
+	{ type: TType }
+>["payload"];
+
+const parseJsonMessage = (data: string) => safeJsonParse<unknown>(data);
+
+export const parseControllerMessage = (data: ControllerWireMessage) => {
+	const parsed = parseJsonMessage(data);
+	if (parsed === null) {
+		return null;
+	}
+
+	return controllerMessageSchema.safeParse(parsed);
+};
+
+export const parseAgentMessage = (data: string) => {
+	const parsed = parseJsonMessage(data);
+	if (parsed === null) {
+		return null;
+	}
+
+	return agentMessageSchema.safeParse(parsed);
+};
+
+export const createControllerMessage = <TType extends ControllerMessage["type"]>(
+	type: TType,
+	payload: PayloadForMessage<ControllerMessage, TType>,
+) =>
+	JSON.stringify(
+		controllerMessageSchema.parse({
+			type,
+			payload,
+		}),
+	) as ControllerWireMessage;
+
+export const createAgentMessage = <TType extends AgentMessage["type"]>(
+	type: TType,
+	payload: PayloadForMessage<AgentMessage, TType>,
+) =>
+	JSON.stringify(
+		agentMessageSchema.parse({
+			type,
+			payload,
+		}),
+	) as AgentWireMessage;
+
+export const sendControllerMessage = (target: MessageSender, message: ControllerWireMessage) => {
+	target.send(message);
+};
+
+export const sendAgentMessage = (target: MessageSender, message: AgentWireMessage) => {
+	target.send(message);
+};
+
+export type ControllerData = MessageEvent<ControllerWireMessage>;

packages/contracts/tsconfig.json

@@ -0,0 +1,29 @@
+{
+	"compilerOptions": {
+		// Environment setup & latest features
+		"lib": ["ESNext"],
+		"target": "ESNext",
+		"module": "Preserve",
+		"moduleDetection": "force",
+		"jsx": "react-jsx",
+		"allowJs": true,
+
+		// Bundler mode
+		"moduleResolution": "bundler",
+		"allowImportingTsExtensions": true,
+		"verbatimModuleSyntax": true,
+		"noEmit": true,
+
+		// Best practices
+		"strict": true,
+		"skipLibCheck": true,
+		"noFallthroughCasesInSwitch": true,
+		"noUncheckedIndexedAccess": true,
+		"noImplicitOverride": true,
+
+		// Some stricter flags (disabled by default)
+		"noUnusedLocals": false,
+		"noUnusedParameters": false,
+		"noPropertyAccessFromIndexSignature": false
+	}
+}