danvics/zerobyte
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor: simplify singleton pattern

Browse source
This commit is contained in:
Nicolas Meienberger 2026-04-01 23:11:51 +02:00
parent 23a2a168be
commit af4ac1c39c
12 changed files with 266 additions and 270 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.dockerignore
View file

@ -11,6 +11,7 @@
!**/components.json
!app/**
!apps/agent/**
!packages/**
!public/**

4
Dockerfile
View file

@ -64,6 +64,8 @@ COPY --from=deps /deps/shoutrrr /usr/local/bin/shoutrrr
COPY ./package.json ./bun.lock ./
COPY ./packages/core/package.json ./packages/core/package.json
COPY ./packages/contracts/package.json ./packages/contracts/package.json
COPY ./apps/agent/package.json ./apps/agent/package.json
RUN bun install --frozen-lockfile --ignore-scripts
@ -86,6 +88,8 @@ WORKDIR /app
COPY ./package.json ./bun.lock ./
COPY ./packages/core/package.json ./packages/core/package.json
COPY ./packages/contracts/package.json ./packages/contracts/package.json
COPY ./apps/agent/package.json ./apps/agent/package.json
RUN bun install --frozen-lockfile
COPY . .

33
app/server/modules/agents/__tests__/agents-manager.test.ts
View file

@ -1,33 +0,0 @@
import { afterEach, describe, expect, mock, test } from "bun:test";
const loadAgentsManagerModule = async () => {
const moduleUrl = new URL("../agents-manager.ts", import.meta.url);
moduleUrl.searchParams.set("test", crypto.randomUUID());
return import(moduleUrl.href);
};
afterEach(() => {
delete (globalThis as Record<string, unknown>).__agentManager;
mock.restore();
});
describe("agents-manager module", () => {
test("reuses the existing global agent manager without stopping it", async () => {
const existingManager = {
start: mock(() => {}),
sendBackup: mock(() => false),
cancelBackup: mock(() => false),
setBackupEventHandlers: mock(() => {}),
getBackupEventHandlers: mock(() => ({})),
stop: mock(() => {}),
};
(globalThis as Record<string, unknown>).__agentManager = existingManager;
const { agentManager } = await loadAgentsManagerModule();
expect(agentManager).toBe(existingManager);
expect(existingManager.stop).not.toHaveBeenCalled();
expect((globalThis as Record<string, unknown>).__agentManager).toBe(existingManager);
});
});

94
app/server/modules/agents/agents-manager.ts
View file

@ -14,14 +14,11 @@ import type {
} from "@zerobyte/contracts/agent-protocol";
import { config } from "../../core/config";
import { validateAgentToken, deriveLocalAgentToken } from "./agent-tokens";
import { createControllerAgentSession, type ControllerAgentSession } from "./controller-agent-session";
type AgentConnectionData = {
id: string;
agentId: string;
organizationId: string | null;
agentName: string;
};
import {
createControllerAgentSession,
type AgentConnectionData,
type ControllerAgentSession,
} from "./controller-agent-session";
type AgentBackupEventContext = {
agentId: string;
@ -42,12 +39,38 @@ export type AgentBackupEventHandlers = {
onBackupCancelled?: (context: AgentBackupEventContext & { payload: BackupCancelledPayload }) => void;
};
export const spawnLocalAgent = async () => {
const previousAgent = (globalThis as Record<string, unknown>).__localAgent as ChildProcess | undefined;
if (previousAgent) {
previousAgent.kill();
type AgentManagerRuntime = ReturnType<typeof createAgentManagerRuntime>;
type AgentRuntimeState = {
agentManager: AgentManagerRuntime;
localAgent: ChildProcess | null;
};
type ProcessWithAgentRuntime = NodeJS.Process & {
__zerobyteAgentRuntime?: AgentRuntimeState;
};
const getAgentRuntimeState = () => {
const runtimeProcess = process as ProcessWithAgentRuntime;
const existingRuntime = runtimeProcess.__zerobyteAgentRuntime;
if (existingRuntime) {
return existingRuntime;
}
const runtime = {
agentManager: createAgentManagerRuntime(),
localAgent: null,
};
runtimeProcess.__zerobyteAgentRuntime = runtime;
return runtime;
};
const getAgentManagerRuntime = () => getAgentRuntimeState().agentManager;
export const spawnLocalAgent = async () => {
await stopLocalAgent();
const sourceEntryPoint = path.join(process.cwd(), "apps", "agent", "src", "index.ts");
const productionEntryPoint = path.join(process.cwd(), ".output", "agent", "index.mjs");
@ -59,7 +82,8 @@ export const spawnLocalAgent = async () => {
const agentToken = await deriveLocalAgentToken();
const args = config.__prod__ ? ["run", agentEntryPoint] : ["run", "--watch", agentEntryPoint];
const localAgent = spawn("bun", args, {
const runtime = getAgentRuntimeState();
const agentProcess = spawn("bun", args, {
env: {
...process.env,
ZEROBYTE_CONTROLLER_URL: "ws://localhost:3001",
@ -68,23 +92,49 @@ export const spawnLocalAgent = async () => {
stdio: ["ignore", "pipe", "pipe"],
});
(globalThis as Record<string, unknown>).__localAgent = localAgent;
runtime.localAgent = agentProcess;
localAgent.stdout?.on("data", (data: Buffer) => {
agentProcess.stdout?.on("data", (data: Buffer) => {
const line = data.toString().trim();
if (line) logger.info(`[agent] ${line}`);
});
localAgent.stderr?.on("data", (data: Buffer) => {
agentProcess.stderr?.on("data", (data: Buffer) => {
const line = data.toString().trim();
if (line) logger.error(`[agent] ${line}`);
});
localAgent.on("exit", (code, signal) => {
agentProcess.on("exit", (code, signal) => {
if (runtime.localAgent === agentProcess) {
runtime.localAgent = null;
}
logger.info(`Agent process exited with code ${code} and signal ${signal}`);
});
};
export const stopLocalAgent = async () => {
const runtime = getAgentRuntimeState();
if (!runtime.localAgent) {
return;
}
const agentProcess = runtime.localAgent;
runtime.localAgent = null;
if (agentProcess.exitCode !== null || agentProcess.signalCode !== null) {
return;
}
const exited = new Promise<void>((resolve) => {
agentProcess.once("exit", () => {
resolve();
});
});
agentProcess.kill();
await exited;
};
const createAgentManagerRuntime = () => {
const sessionsRef = Effect.runSync(Ref.make<Map<string, ControllerAgentSession>>(new Map()));
const backupHandlersRef = Effect.runSync(Ref.make<AgentBackupEventHandlers>({}));
@ -290,9 +340,9 @@ const createAgentManagerRuntime = () => {
};
};
const previous = (globalThis as Record<string, unknown>).__agentManager as
| ReturnType<typeof createAgentManagerRuntime>
| undefined;
export const agentManager = getAgentManagerRuntime();
export const agentManager = previous ?? createAgentManagerRuntime();
(globalThis as Record<string, unknown>).__agentManager = agentManager;
export const stopAgentRuntime = async () => {
getAgentManagerRuntime().stop();
await stopLocalAgent();
};

29
app/server/modules/agents/controller-agent-session.ts
View file

@ -15,7 +15,7 @@ import {
import { logger } from "@zerobyte/core/node";
import { toMessage } from "@zerobyte/core/utils";
type AgentConnectionData = {
export type AgentConnectionData = {
id: string;
agentId: string;
organizationId: string | null;
@ -123,14 +123,15 @@ export const createControllerAgentSession = (
);
const handleAgentMessage = (message: AgentMessage) => {
updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
switch (message.type) {
case "agent.ready": {
updateState((current) => ({ ...current, isReady: true, lastSeenAt: Date.now() }));
updateState((current) => ({ ...current, isReady: true }));
logger.info(`Agent "${socket.data.agentName}" (${socket.data.agentId}) is ready`);
break;
}
case "backup.started": {
updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
setActiveBackupJob(message.payload.jobId, message.payload.scheduleId);
logger.info(
`Backup ${message.payload.jobId} started on agent ${socket.data.agentId} for schedule ${message.payload.scheduleId}`,
@ -139,34 +140,26 @@ export const createControllerAgentSession = (
break;
}
case "backup.progress": {
updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
handlers.onBackupProgress?.(message.payload);
break;
}
case "backup.completed": {
updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
deleteActiveBackupJob(message.payload.jobId);
handlers.onBackupCompleted?.(message.payload);
break;
}
case "backup.failed": {
updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
deleteActiveBackupJob(message.payload.jobId);
handlers.onBackupFailed?.(message.payload);
break;
}
case "backup.cancelled": {
updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
deleteActiveBackupJob(message.payload.jobId);
handlers.onBackupCancelled?.(message.payload);
break;
}
case "heartbeat.pong": {
updateState((current) => ({
...current,
lastSeenAt: Date.now(),
lastPongAt: message.payload.sentAt,
}));
updateState((current) => ({ ...current, lastPongAt: message.payload.sentAt }));
break;
}
}
@ -190,19 +183,11 @@ export const createControllerAgentSession = (
handleAgentMessage(parsed.data);
},
sendBackup: (payload) => {
offerOutbound(
createControllerMessage("backup.run", {
...payload,
}),
);
offerOutbound(createControllerMessage("backup.run", payload));
return payload.jobId;
},
sendBackupCancel: (payload) => {
offerOutbound(
createControllerMessage("backup.cancel", {
...payload,
}),
);
offerOutbound(createControllerMessage("backup.cancel", payload));
},
isReady: () => Effect.runSync(Ref.get(state)).isReady,
close: () => {

91
app/server/modules/backups/__tests__/backups.execution.test.ts
View file

@ -16,7 +16,7 @@ import { NotFoundError, BadRequestError } from "http-errors-enhanced";
import { repositoriesService } from "~/server/modules/repositories/repositories.service";
import { repoMutex } from "~/server/core/repository-mutex";
import { agentManager } from "~/server/modules/agents/agents-manager";
import { fromAny } from "@total-typescript/shoehorn";
import { createAgentBackupMocks } from "~/test/helpers/agent-mock";
const setup = () => {
const resticBackupMock = vi.fn((_: SafeSpawnParams) =>
@ -24,94 +24,7 @@ const setup = () => {
);
const resticForgetMock = vi.fn(() => Promise.resolve({ success: true, data: null }));
const resticCopyMock = vi.fn(() => Promise.resolve({ success: true, output: "" }));
const runningJobs = new Map<string, { scheduleId: string; cancelled: boolean }>();
const sendBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
const handlers = agentManager.getBackupEventHandlers();
runningJobs.set(payload.jobId, { scheduleId: payload.scheduleId, cancelled: false });
handlers.onBackupStarted?.({
agentId: "local",
agentName: "local",
payload: { jobId: payload.jobId, scheduleId: payload.scheduleId },
});
void (async () => {
const stderrLines: string[] = [];
const result = await resticBackupMock(
fromAny({
onStderr: (line: string) => {
stderrLines.push(line);
},
} satisfies Partial<SafeSpawnParams>),
);
const running = runningJobs.get(payload.jobId);
if (!running || running.cancelled) {
return;
}
if (result.exitCode === 0 || result.exitCode === 3) {
let parsedResult: Record<string, unknown> | null = null;
if (result.summary) {
try {
parsedResult = JSON.parse(result.summary) as Record<string, unknown>;
} catch {
parsedResult = null;
}
}
handlers.onBackupCompleted?.({
agentId: "local",
agentName: "local",
payload: {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
exitCode: result.exitCode,
result: fromAny(parsedResult),
warningDetails: stderrLines.join("\n") || undefined,
},
});
} else {
const resultWithStderr = result as typeof result & { stderr?: string };
const errorDetails = stderrLines.join("\n") || resultWithStderr.stderr || result.error;
handlers.onBackupFailed?.({
agentId: "local",
agentName: "local",
payload: {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
error: result.error || `Backup failed with code ${result.exitCode}`,
errorDetails,
},
});
}
runningJobs.delete(payload.jobId);
})().catch(() => {});
return true;
});
const cancelBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
const running = runningJobs.get(payload.jobId);
if (!running) {
return false;
}
running.cancelled = true;
const handlers = agentManager.getBackupEventHandlers();
handlers.onBackupCancelled?.({
agentId: "local",
agentName: "local",
payload: {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
message: "Backup was stopped by user",
},
});
runningJobs.delete(payload.jobId);
return true;
});
const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock);
const refreshStatsMock = vi.fn(() =>
Promise.resolve({
total_size: 0,

91
app/server/modules/backups/__tests__/backups.service.test.ts
View file

@ -15,98 +15,11 @@ import * as context from "~/server/core/request-context";
import { backupsExecutionService } from "../backups.execution";
import { repositoriesService } from "~/server/modules/repositories/repositories.service";
import { agentManager } from "~/server/modules/agents/agents-manager";
import { fromAny } from "@total-typescript/shoehorn";
import { createAgentBackupMocks } from "~/test/helpers/agent-mock";
const setup = () => {
const resticBackupMock = vi.fn((_: unknown) => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
const runningJobs = new Map<string, { scheduleId: string; cancelled: boolean }>();
const sendBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
const handlers = agentManager.getBackupEventHandlers();
runningJobs.set(payload.jobId, { scheduleId: payload.scheduleId, cancelled: false });
handlers.onBackupStarted?.({
agentId: "local",
agentName: "local",
payload: { jobId: payload.jobId, scheduleId: payload.scheduleId },
});
void (async () => {
const stderrLines: string[] = [];
const result = await resticBackupMock(
fromAny({
onStderr: (line: string) => {
stderrLines.push(line);
},
}),
);
const running = runningJobs.get(payload.jobId);
if (!running || running.cancelled) {
return;
}
if (result.exitCode === 0 || result.exitCode === 3) {
let parsedResult: Record<string, unknown> | null = null;
if (result.summary) {
try {
parsedResult = JSON.parse(result.summary) as Record<string, unknown>;
} catch {
parsedResult = null;
}
}
handlers.onBackupCompleted?.({
agentId: "local",
agentName: "local",
payload: {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
exitCode: result.exitCode,
result: fromAny(parsedResult),
warningDetails: stderrLines.join("\n") || undefined,
},
});
} else {
const resultWithStderr = result as typeof result & { stderr?: string };
const errorDetails = stderrLines.join("\n") || resultWithStderr.stderr || result.error;
handlers.onBackupFailed?.({
agentId: "local",
agentName: "local",
payload: {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
error: result.error || `Backup failed with code ${result.exitCode}`,
errorDetails,
},
});
}
runningJobs.delete(payload.jobId);
})().catch(() => {});
return true;
});
const cancelBackupMock = vi.fn((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
const running = runningJobs.get(payload.jobId);
if (!running) {
return false;
}
running.cancelled = true;
const handlers = agentManager.getBackupEventHandlers();
handlers.onBackupCancelled?.({
agentId: "local",
agentName: "local",
payload: {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
message: "Backup was stopped by user",
},
});
runningJobs.delete(payload.jobId);
return true;
});
const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock);
const refreshStatsMock = vi.fn(() =>
Promise.resolve({
total_size: 0,

58
app/server/modules/lifecycle/__tests__/shutdown.test.ts Normal file
View file

@ -0,0 +1,58 @@
import { afterEach, describe, expect, mock, spyOn, test } from "bun:test";
import { Scheduler } from "../../../core/scheduler";
import * as agentsManagerModule from "../../agents/agents-manager";
import * as backendModule from "../../backends/backend";
import type { VolumeBackend } from "../../backends/backend";
import { createTestVolume } from "~/test/helpers/volume";
const loadShutdownModule = async () => {
const moduleUrl = new URL("../shutdown.ts", import.meta.url);
moduleUrl.searchParams.set("test", crypto.randomUUID());
return import(moduleUrl.href);
};
afterEach(() => {
mock.restore();
});
describe("shutdown", () => {
test("stops the agent runtime before unmounting mounted volumes", async () => {
const events: string[] = [];
const stopScheduler = mock(async () => {
events.push("scheduler.stop");
});
const stopAgentRuntime = mock(async () => {
events.push("agents.stop");
});
const unmountVolume = mock(async () => {
events.push("backend.unmount");
return { status: "unmounted" as const };
});
await createTestVolume({
name: "Shutdown test volume",
config: {
backend: "directory",
path: "/Applications",
},
status: "mounted",
});
spyOn(Scheduler, "stop").mockImplementation(stopScheduler);
spyOn(agentsManagerModule, "stopAgentRuntime").mockImplementation(stopAgentRuntime);
spyOn(backendModule, "createVolumeBackend").mockImplementation(
() =>
({
mount: async () => ({ status: "mounted" as const }),
unmount: unmountVolume,
checkHealth: async () => ({ status: "mounted" as const }),
}) satisfies VolumeBackend,
);
const { shutdown } = await loadShutdownModule();
await shutdown();
expect(events).toEqual(["scheduler.stop", "agents.stop", "backend.unmount"]);
});
});

2
app/server/modules/lifecycle/shutdown.ts
View file

@ -2,9 +2,11 @@ import { Scheduler } from "../../core/scheduler";
import { db } from "../../db/db";
import { logger } from "@zerobyte/core/node";
import { createVolumeBackend } from "../backends/backend";
import { stopAgentRuntime } from "../agents/agents-manager";
export const shutdown = async () => {
await Scheduler.stop();
await stopAgentRuntime();
const volumes = await db.query.volumesTable.findMany({
where: { status: "mounted" },

14
app/server/plugins/bootstrap.ts
View file

@ -2,8 +2,20 @@ import { definePlugin } from "nitro";
import { bootstrapApplication } from "../modules/lifecycle/bootstrap";
import { logger } from "@zerobyte/core/node";
import { toMessage } from "../utils/errors";
import { stopAgentRuntime } from "../modules/agents/agents-manager";
type ProcessWithAgentCloseHook = NodeJS.Process & {
__zerobyteAgentRuntimeCloseHookRegistered?: boolean;
};
export default definePlugin(async (nitroApp) => {
const runtimeProcess = process as ProcessWithAgentCloseHook;
if (!runtimeProcess.__zerobyteAgentRuntimeCloseHookRegistered) {
nitroApp.hooks.hook("close", stopAgentRuntime);
runtimeProcess.__zerobyteAgentRuntimeCloseHookRegistered = true;
}
export default definePlugin(async () => {
await bootstrapApplication().catch((err) => {
logger.error(`Bootstrap failed: ${toMessage(err)}`);
process.exit(1);

105
app/test/helpers/agent-mock.ts Normal file
View file

@ -0,0 +1,105 @@
import { mock } from "bun:test";
import { fromAny } from "@total-typescript/shoehorn";
import { agentManager } from "~/server/modules/agents/agents-manager";
export const createAgentBackupMocks = (
resticBackupMock: (params: never) => Promise<{
exitCode: number;
summary: string;
error: string;
stderr?: string;
}>,
) => {
const runningJobs = new Map<string, { scheduleId: string; cancelled: boolean }>();
const sendBackupMock = mock((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
const handlers = agentManager.getBackupEventHandlers();
runningJobs.set(payload.jobId, { scheduleId: payload.scheduleId, cancelled: false });
handlers.onBackupStarted?.({
agentId: "local",
agentName: "local",
payload: { jobId: payload.jobId, scheduleId: payload.scheduleId },
});
void (async () => {
const stderrLines: string[] = [];
const result = await resticBackupMock(
fromAny({
onStderr: (line: string) => {
stderrLines.push(line);
},
}),
);
const running = runningJobs.get(payload.jobId);
if (!running || running.cancelled) {
return;
}
if (result.exitCode === 0 || result.exitCode === 3) {
let parsedResult: Record<string, unknown> | null = null;
if (result.summary) {
try {
parsedResult = JSON.parse(result.summary) as Record<string, unknown>;
} catch {
parsedResult = null;
}
}
handlers.onBackupCompleted?.({
agentId: "local",
agentName: "local",
payload: {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
exitCode: result.exitCode,
result: fromAny(parsedResult),
warningDetails: stderrLines.join("\n") || undefined,
},
});
} else {
const resultWithStderr = result as typeof result & { stderr?: string };
const errorDetails = stderrLines.join("\n") || resultWithStderr.stderr || result.error;
handlers.onBackupFailed?.({
agentId: "local",
agentName: "local",
payload: {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
error: result.error || `Backup failed with code ${result.exitCode}`,
errorDetails,
},
});
}
runningJobs.delete(payload.jobId);
})().catch(() => {});
return true;
});
const cancelBackupMock = mock((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
const running = runningJobs.get(payload.jobId);
if (!running) {
return false;
}
running.cancelled = true;
const handlers = agentManager.getBackupEventHandlers();
handlers.onBackupCancelled?.({
agentId: "local",
agentName: "local",
payload: {
jobId: payload.jobId,
scheduleId: payload.scheduleId,
message: "Backup was stopped by user",
},
});
runningJobs.delete(payload.jobId);
return true;
});
return { sendBackupMock, cancelBackupMock };
};

14
packages/contracts/src/agent-protocol.ts
View file

@ -168,10 +168,6 @@ 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">;
@ -221,13 +217,3 @@ export const createAgentMessage = <TType extends AgentMessage["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>;