feat(agents): create agent registry and service
This commit is contained in:
parent
e981211a2d
commit
4e17daf6fe
13 changed files with 2800 additions and 53 deletions
16
app/drizzle/20260414064347_common_skin/migration.sql
Normal file
16
app/drizzle/20260414064347_common_skin/migration.sql
Normal file
|
|
@ -0,0 +1,16 @@
|
||||||
|
CREATE TABLE `agents_table` (
|
||||||
|
`id` text PRIMARY KEY,
|
||||||
|
`organization_id` text,
|
||||||
|
`name` text NOT NULL,
|
||||||
|
`kind` text NOT NULL,
|
||||||
|
`status` text DEFAULT 'offline' NOT NULL,
|
||||||
|
`capabilities` text DEFAULT '{}' NOT NULL,
|
||||||
|
`last_seen_at` integer,
|
||||||
|
`last_ready_at` integer,
|
||||||
|
`created_at` integer DEFAULT (unixepoch() * 1000) NOT NULL,
|
||||||
|
`updated_at` integer DEFAULT (unixepoch() * 1000) NOT NULL,
|
||||||
|
CONSTRAINT `fk_agents_table_organization_id_organization_id_fk` FOREIGN KEY (`organization_id`) REFERENCES `organization`(`id`) ON DELETE CASCADE
|
||||||
|
);
|
||||||
|
--> statement-breakpoint
|
||||||
|
CREATE INDEX `agents_table_organization_id_idx` ON `agents_table` (`organization_id`);--> statement-breakpoint
|
||||||
|
CREATE INDEX `agents_table_status_idx` ON `agents_table` (`status`);
|
||||||
2481
app/drizzle/20260414064347_common_skin/snapshot.json
Normal file
2481
app/drizzle/20260414064347_common_skin/snapshot.json
Normal file
File diff suppressed because it is too large
Load diff
|
|
@ -99,6 +99,7 @@ export const relations = defineRelations(schema, (r) => ({
|
||||||
users: r.many.usersTable({
|
users: r.many.usersTable({
|
||||||
alias: "usersTable_id_organization_id_via_member",
|
alias: "usersTable_id_organization_id_via_member",
|
||||||
}),
|
}),
|
||||||
|
agents: r.many.agentsTable(),
|
||||||
backupSchedules: r.many.backupSchedulesTable(),
|
backupSchedules: r.many.backupSchedulesTable(),
|
||||||
notificationDestinations: r.many.notificationDestinationsTable(),
|
notificationDestinations: r.many.notificationDestinationsTable(),
|
||||||
repositories: r.many.repositoriesTable(),
|
repositories: r.many.repositoriesTable(),
|
||||||
|
|
@ -119,6 +120,13 @@ export const relations = defineRelations(schema, (r) => ({
|
||||||
optional: false,
|
optional: false,
|
||||||
}),
|
}),
|
||||||
},
|
},
|
||||||
|
agentsTable: {
|
||||||
|
organization: r.one.organization({
|
||||||
|
from: r.agentsTable.organizationId,
|
||||||
|
to: r.organization.id,
|
||||||
|
optional: true,
|
||||||
|
}),
|
||||||
|
},
|
||||||
volumesTable: {
|
volumesTable: {
|
||||||
backupSchedules: r.many.backupSchedulesTable(),
|
backupSchedules: r.many.backupSchedulesTable(),
|
||||||
organization: r.one.organization({
|
organization: r.one.organization({
|
||||||
|
|
|
||||||
|
|
@ -198,6 +198,36 @@ export const ssoProvider = sqliteTable("sso_provider", {
|
||||||
.default(sql`(unixepoch() * 1000)`),
|
.default(sql`(unixepoch() * 1000)`),
|
||||||
});
|
});
|
||||||
|
|
||||||
|
export type AgentKind = "local" | "remote";
|
||||||
|
export type AgentStatus = "offline" | "connecting" | "online" | "degraded";
|
||||||
|
export type AgentCapabilities = Record<string, unknown>;
|
||||||
|
|
||||||
|
export const agentsTable = sqliteTable(
|
||||||
|
"agents_table",
|
||||||
|
{
|
||||||
|
id: text("id").primaryKey(),
|
||||||
|
organizationId: text("organization_id").references(() => organization.id, { onDelete: "cascade" }),
|
||||||
|
name: text("name").notNull(),
|
||||||
|
kind: text("kind").$type<AgentKind>().notNull(),
|
||||||
|
status: text("status").$type<AgentStatus>().notNull().default("offline"),
|
||||||
|
capabilities: text("capabilities", { mode: "json" }).$type<AgentCapabilities>().notNull().default({}),
|
||||||
|
lastSeenAt: int("last_seen_at", { mode: "number" }),
|
||||||
|
lastReadyAt: int("last_ready_at", { mode: "number" }),
|
||||||
|
createdAt: int("created_at", { mode: "number" })
|
||||||
|
.notNull()
|
||||||
|
.default(sql`(unixepoch() * 1000)`),
|
||||||
|
updatedAt: int("updated_at", { mode: "number" })
|
||||||
|
.notNull()
|
||||||
|
.$onUpdate(() => Date.now())
|
||||||
|
.default(sql`(unixepoch() * 1000)`),
|
||||||
|
},
|
||||||
|
(table) => [
|
||||||
|
index("agents_table_organization_id_idx").on(table.organizationId),
|
||||||
|
index("agents_table_status_idx").on(table.status),
|
||||||
|
],
|
||||||
|
);
|
||||||
|
export type Agent = typeof agentsTable.$inferSelect;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Volumes Table
|
* Volumes Table
|
||||||
*/
|
*/
|
||||||
|
|
|
||||||
17
app/server/modules/agents/__tests__/agents.service.test.ts
Normal file
17
app/server/modules/agents/__tests__/agents.service.test.ts
Normal file
|
|
@ -0,0 +1,17 @@
|
||||||
|
import { beforeEach, expect, test } from "vitest";
|
||||||
|
import { db } from "~/server/db/db";
|
||||||
|
import { agentsTable } from "~/server/db/schema";
|
||||||
|
import { agentsService } from "../agents.service";
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
await db.delete(agentsTable);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("ensureLocalAgent seeds the built-in local agent once", async () => {
|
||||||
|
await agentsService.ensureLocalAgent();
|
||||||
|
await agentsService.ensureLocalAgent();
|
||||||
|
|
||||||
|
const agents = await agentsService.listAgents();
|
||||||
|
|
||||||
|
expect(agents).toHaveLength(1);
|
||||||
|
});
|
||||||
|
|
@ -3,22 +3,44 @@ import { expect, test, vi } from "vitest";
|
||||||
import waitForExpect from "wait-for-expect";
|
import waitForExpect from "wait-for-expect";
|
||||||
import { fromPartial } from "@total-typescript/shoehorn";
|
import { fromPartial } from "@total-typescript/shoehorn";
|
||||||
import { createAgentMessage } from "@zerobyte/contracts/agent-protocol";
|
import { createAgentMessage } from "@zerobyte/contracts/agent-protocol";
|
||||||
|
import { LOCAL_AGENT_ID, LOCAL_AGENT_KIND, LOCAL_AGENT_NAME } from "../constants";
|
||||||
import { createControllerAgentSession } from "../controller/session";
|
import { createControllerAgentSession } from "../controller/session";
|
||||||
|
|
||||||
const createSocket = (overrides: Partial<Parameters<typeof createControllerAgentSession>[0]> = {}) => {
|
const createSocket = (overrides: Partial<Parameters<typeof createControllerAgentSession>[0]> = {}) => {
|
||||||
return {
|
return {
|
||||||
data: { id: "connection-1", agentId: "local", organizationId: null, agentName: "Local Agent" },
|
data: {
|
||||||
|
id: "connection-1",
|
||||||
|
agentId: LOCAL_AGENT_ID,
|
||||||
|
organizationId: null,
|
||||||
|
agentName: LOCAL_AGENT_NAME,
|
||||||
|
agentKind: LOCAL_AGENT_KIND,
|
||||||
|
},
|
||||||
send: vi.fn(() => 1),
|
send: vi.fn(() => 1),
|
||||||
close: vi.fn(),
|
close: vi.fn(),
|
||||||
...overrides,
|
...overrides,
|
||||||
};
|
};
|
||||||
};
|
};
|
||||||
|
|
||||||
const createSession = (handlers: Parameters<typeof createControllerAgentSession>[1] = {}, socket = createSocket()) => {
|
const createSession = (
|
||||||
|
handlers: Partial<Parameters<typeof createControllerAgentSession>[1]> = {},
|
||||||
|
socket = createSocket(),
|
||||||
|
) => {
|
||||||
const scope = Effect.runSync(Scope.make());
|
const scope = Effect.runSync(Scope.make());
|
||||||
|
const sessionHandlers: Parameters<typeof createControllerAgentSession>[1] = {
|
||||||
|
onReady: () => Effect.void,
|
||||||
|
onHeartbeatPong: () => Effect.void,
|
||||||
|
onBackupStarted: () => Effect.void,
|
||||||
|
onBackupProgress: () => Effect.void,
|
||||||
|
onBackupCompleted: () => Effect.void,
|
||||||
|
onBackupFailed: () => Effect.void,
|
||||||
|
onBackupCancelled: () => Effect.void,
|
||||||
|
...handlers,
|
||||||
|
};
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const session = Effect.runSync(Scope.extend(createControllerAgentSession(fromPartial(socket), handlers), scope));
|
const session = Effect.runSync(
|
||||||
|
Scope.extend(createControllerAgentSession(fromPartial(socket), sessionHandlers), scope),
|
||||||
|
);
|
||||||
|
|
||||||
return {
|
return {
|
||||||
session,
|
session,
|
||||||
|
|
@ -40,9 +62,10 @@ const createSession = (handlers: Parameters<typeof createControllerAgentSession>
|
||||||
};
|
};
|
||||||
|
|
||||||
test("close emits a synthetic backup.cancelled for a started backup", () => {
|
test("close emits a synthetic backup.cancelled for a started backup", () => {
|
||||||
const onBackupCancelled = vi.fn();
|
const onBackupCancelled = vi.fn(() => Effect.void);
|
||||||
const { session, close } = createSession({
|
const { session, close } = createSession({
|
||||||
onBackupCancelled,
|
onBackupCancelled,
|
||||||
|
onBackupStarted: vi.fn(() => Effect.void),
|
||||||
});
|
});
|
||||||
|
|
||||||
Effect.runSync(
|
Effect.runSync(
|
||||||
|
|
@ -101,10 +124,8 @@ test.each([
|
||||||
expectedCancelledCalls: 1,
|
expectedCancelledCalls: 1,
|
||||||
},
|
},
|
||||||
])("close does not emit an extra synthetic backup.cancelled after $name", (testCase) => {
|
])("close does not emit an extra synthetic backup.cancelled after $name", (testCase) => {
|
||||||
const onBackupCancelled = vi.fn();
|
const onBackupCancelled = vi.fn(() => Effect.void);
|
||||||
const { session, close } = createSession({
|
const { session, close } = createSession({ onBackupCancelled });
|
||||||
onBackupCancelled,
|
|
||||||
});
|
|
||||||
|
|
||||||
Effect.runSync(
|
Effect.runSync(
|
||||||
session.handleMessage(
|
session.handleMessage(
|
||||||
|
|
@ -121,10 +142,8 @@ test.each([
|
||||||
});
|
});
|
||||||
|
|
||||||
test("close emits a synthetic backup.cancelled for a queued backup", () => {
|
test("close emits a synthetic backup.cancelled for a queued backup", () => {
|
||||||
const onBackupCancelled = vi.fn();
|
const onBackupCancelled = vi.fn(() => Effect.void);
|
||||||
const { session, close } = createSession({
|
const { session, close } = createSession({ onBackupCancelled });
|
||||||
onBackupCancelled,
|
|
||||||
});
|
|
||||||
|
|
||||||
Effect.runSync(
|
Effect.runSync(
|
||||||
session.sendBackup({
|
session.sendBackup({
|
||||||
|
|
@ -164,7 +183,7 @@ test("close emits a synthetic backup.cancelled for a queued backup", () => {
|
||||||
test("a dropped backup.cancel closes the session and emits a synthetic backup.cancelled", async () => {
|
test("a dropped backup.cancel closes the session and emits a synthetic backup.cancelled", async () => {
|
||||||
const send = vi.fn(() => 0);
|
const send = vi.fn(() => 0);
|
||||||
const socket = createSocket({ send, close: vi.fn() });
|
const socket = createSocket({ send, close: vi.fn() });
|
||||||
const onBackupCancelled = vi.fn();
|
const onBackupCancelled = vi.fn(() => Effect.void);
|
||||||
const { session, run, closeAsync } = createSession({ onBackupCancelled }, socket);
|
const { session, run, closeAsync } = createSession({ onBackupCancelled }, socket);
|
||||||
|
|
||||||
try {
|
try {
|
||||||
|
|
|
||||||
129
app/server/modules/agents/agents.service.ts
Normal file
129
app/server/modules/agents/agents.service.ts
Normal file
|
|
@ -0,0 +1,129 @@
|
||||||
|
import { eq } from "drizzle-orm";
|
||||||
|
import { db } from "../../db/db";
|
||||||
|
import { agentsTable, type Agent, type AgentCapabilities, type AgentKind } from "../../db/schema";
|
||||||
|
import { LOCAL_AGENT_CAPABILITIES, LOCAL_AGENT_ID, LOCAL_AGENT_KIND, LOCAL_AGENT_NAME } from "./constants";
|
||||||
|
|
||||||
|
type AgentConnectionRegistration = {
|
||||||
|
agentId: string;
|
||||||
|
organizationId: string | null;
|
||||||
|
agentName: string;
|
||||||
|
agentKind: AgentKind;
|
||||||
|
capabilities?: AgentCapabilities;
|
||||||
|
connectedAt?: number;
|
||||||
|
};
|
||||||
|
|
||||||
|
const listAgents = async (organizationId?: string | null) => {
|
||||||
|
if (organizationId === undefined) {
|
||||||
|
return db.query.agentsTable.findMany({ orderBy: { createdAt: "asc" } });
|
||||||
|
}
|
||||||
|
|
||||||
|
if (organizationId === null) {
|
||||||
|
return db.query.agentsTable.findMany({
|
||||||
|
where: { organizationId: { isNull: true } },
|
||||||
|
orderBy: { createdAt: "asc" },
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
return db.query.agentsTable.findMany({
|
||||||
|
where: { organizationId },
|
||||||
|
orderBy: { createdAt: "asc" },
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const getAgent = async (agentId: string) => {
|
||||||
|
return db.query.agentsTable.findFirst({ where: { id: agentId } });
|
||||||
|
};
|
||||||
|
|
||||||
|
const ensureLocalAgent = async () => {
|
||||||
|
const existing = await getAgent(LOCAL_AGENT_ID);
|
||||||
|
|
||||||
|
if (existing) {
|
||||||
|
return existing;
|
||||||
|
}
|
||||||
|
|
||||||
|
await db.insert(agentsTable).values({
|
||||||
|
id: LOCAL_AGENT_ID,
|
||||||
|
organizationId: null,
|
||||||
|
name: LOCAL_AGENT_NAME,
|
||||||
|
kind: LOCAL_AGENT_KIND,
|
||||||
|
status: "offline",
|
||||||
|
capabilities: LOCAL_AGENT_CAPABILITIES,
|
||||||
|
updatedAt: Date.now(),
|
||||||
|
});
|
||||||
|
|
||||||
|
return getAgent(LOCAL_AGENT_ID);
|
||||||
|
};
|
||||||
|
|
||||||
|
const markAgentConnecting = async (params: AgentConnectionRegistration) => {
|
||||||
|
const { agentId, organizationId, agentName, agentKind, capabilities, connectedAt = Date.now() } = params;
|
||||||
|
|
||||||
|
await db
|
||||||
|
.insert(agentsTable)
|
||||||
|
.values({
|
||||||
|
id: agentId,
|
||||||
|
organizationId,
|
||||||
|
name: agentName,
|
||||||
|
kind: agentKind,
|
||||||
|
status: "connecting",
|
||||||
|
capabilities: capabilities ?? {},
|
||||||
|
lastSeenAt: connectedAt,
|
||||||
|
updatedAt: connectedAt,
|
||||||
|
})
|
||||||
|
.onConflictDoUpdate({
|
||||||
|
target: agentsTable.id,
|
||||||
|
set: {
|
||||||
|
organizationId,
|
||||||
|
name: agentName,
|
||||||
|
kind: agentKind,
|
||||||
|
status: "connecting",
|
||||||
|
lastSeenAt: connectedAt,
|
||||||
|
updatedAt: connectedAt,
|
||||||
|
capabilities: capabilities ?? {},
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
return getAgent(agentId);
|
||||||
|
};
|
||||||
|
|
||||||
|
const updateAgentRuntime = async (agentId: string, values: Partial<Agent>) => {
|
||||||
|
const [updatedAgent] = await db.update(agentsTable).set(values).where(eq(agentsTable.id, agentId)).returning();
|
||||||
|
|
||||||
|
if (!updatedAgent) {
|
||||||
|
throw new Error(`Agent ${agentId} not found`);
|
||||||
|
}
|
||||||
|
|
||||||
|
return updatedAgent;
|
||||||
|
};
|
||||||
|
|
||||||
|
const markAgentOnline = async (agentId: string, readyAt = Date.now()) => {
|
||||||
|
return updateAgentRuntime(agentId, {
|
||||||
|
status: "online",
|
||||||
|
lastSeenAt: readyAt,
|
||||||
|
lastReadyAt: readyAt,
|
||||||
|
updatedAt: readyAt,
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const markAgentSeen = async (agentId: string, seenAt = Date.now()) => {
|
||||||
|
return updateAgentRuntime(agentId, {
|
||||||
|
lastSeenAt: seenAt,
|
||||||
|
updatedAt: seenAt,
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const markAgentOffline = async (agentId: string, disconnectedAt = Date.now()) => {
|
||||||
|
return updateAgentRuntime(agentId, {
|
||||||
|
status: "offline",
|
||||||
|
updatedAt: disconnectedAt,
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
export const agentsService = {
|
||||||
|
listAgents,
|
||||||
|
getAgent,
|
||||||
|
ensureLocalAgent,
|
||||||
|
markAgentConnecting,
|
||||||
|
markAgentOnline,
|
||||||
|
markAgentSeen,
|
||||||
|
markAgentOffline,
|
||||||
|
};
|
||||||
6
app/server/modules/agents/constants.ts
Normal file
6
app/server/modules/agents/constants.ts
Normal file
|
|
@ -0,0 +1,6 @@
|
||||||
|
import type { AgentCapabilities, AgentKind } from "../../db/schema";
|
||||||
|
|
||||||
|
export const LOCAL_AGENT_ID = "local";
|
||||||
|
export const LOCAL_AGENT_NAME = "Local Agent";
|
||||||
|
export const LOCAL_AGENT_KIND: AgentKind = "local";
|
||||||
|
export const LOCAL_AGENT_CAPABILITIES: AgentCapabilities = {};
|
||||||
|
|
@ -11,6 +11,7 @@ import type {
|
||||||
BackupStartedPayload,
|
BackupStartedPayload,
|
||||||
} from "@zerobyte/contracts/agent-protocol";
|
} from "@zerobyte/contracts/agent-protocol";
|
||||||
import { createControllerAgentSession, type AgentConnectionData, type ControllerAgentSession } from "./session";
|
import { createControllerAgentSession, type AgentConnectionData, type ControllerAgentSession } from "./session";
|
||||||
|
import { agentsService } from "../agents.service";
|
||||||
import { validateAgentToken } from "../helpers/tokens";
|
import { validateAgentToken } from "../helpers/tokens";
|
||||||
|
|
||||||
type AgentBackupEventContext = {
|
type AgentBackupEventContext = {
|
||||||
|
|
@ -70,20 +71,26 @@ export function createAgentManagerRuntime() {
|
||||||
const agentName = ws.data.agentName;
|
const agentName = ws.data.agentName;
|
||||||
|
|
||||||
return {
|
return {
|
||||||
|
onReady: ({ at }: { at: number }) => {
|
||||||
|
return Effect.promise(() => agentsService.markAgentOnline(agentId, at));
|
||||||
|
},
|
||||||
|
onHeartbeatPong: ({ at }: { at: number }) => {
|
||||||
|
return Effect.promise(() => agentsService.markAgentSeen(agentId, at));
|
||||||
|
},
|
||||||
onBackupStarted: (payload: BackupStartedPayload) => {
|
onBackupStarted: (payload: BackupStartedPayload) => {
|
||||||
backupHandlers.onBackupStarted?.({ agentId, agentName, payload });
|
return Effect.sync(() => backupHandlers.onBackupStarted?.({ agentId, agentName, payload }));
|
||||||
},
|
},
|
||||||
onBackupProgress: (payload: BackupProgressPayload) => {
|
onBackupProgress: (payload: BackupProgressPayload) => {
|
||||||
backupHandlers.onBackupProgress?.({ agentId, agentName, payload });
|
return Effect.sync(() => backupHandlers.onBackupProgress?.({ agentId, agentName, payload }));
|
||||||
},
|
},
|
||||||
onBackupCompleted: (payload: BackupCompletedPayload) => {
|
onBackupCompleted: (payload: BackupCompletedPayload) => {
|
||||||
backupHandlers.onBackupCompleted?.({ agentId, agentName, payload });
|
return Effect.sync(() => backupHandlers.onBackupCompleted?.({ agentId, agentName, payload }));
|
||||||
},
|
},
|
||||||
onBackupFailed: (payload: BackupFailedPayload) => {
|
onBackupFailed: (payload: BackupFailedPayload) => {
|
||||||
backupHandlers.onBackupFailed?.({ agentId, agentName, payload });
|
return Effect.sync(() => backupHandlers.onBackupFailed?.({ agentId, agentName, payload }));
|
||||||
},
|
},
|
||||||
onBackupCancelled: (payload: BackupCancelledPayload) => {
|
onBackupCancelled: (payload: BackupCancelledPayload) => {
|
||||||
backupHandlers.onBackupCancelled?.({ agentId, agentName, payload });
|
return Effect.sync(() => backupHandlers.onBackupCancelled?.({ agentId, agentName, payload }));
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
};
|
};
|
||||||
|
|
@ -117,13 +124,14 @@ export function createAgentManagerRuntime() {
|
||||||
const removeSession = (agentId: string, connectionId: string) => {
|
const removeSession = (agentId: string, connectionId: string) => {
|
||||||
const sessionHandle = getSessionHandle(agentId);
|
const sessionHandle = getSessionHandle(agentId);
|
||||||
if (!sessionHandle || sessionHandle.session.connectionId !== connectionId) {
|
if (!sessionHandle || sessionHandle.session.connectionId !== connectionId) {
|
||||||
return;
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
sessions.delete(agentId);
|
sessions.delete(agentId);
|
||||||
void Effect.runPromise(closeSession(sessionHandle)).catch((error) => {
|
void Effect.runPromise(closeSession(sessionHandle)).catch((error) => {
|
||||||
logger.error(`Failed to close agent session for ${agentId}: ${toMessage(error)}`);
|
logger.error(`Failed to close agent session for ${agentId}: ${toMessage(error)}`);
|
||||||
});
|
});
|
||||||
|
return true;
|
||||||
};
|
};
|
||||||
|
|
||||||
const acquireServer = Effect.acquireRelease(
|
const acquireServer = Effect.acquireRelease(
|
||||||
|
|
@ -149,6 +157,7 @@ export function createAgentManagerRuntime() {
|
||||||
agentId: result.agentId,
|
agentId: result.agentId,
|
||||||
organizationId: result.organizationId,
|
organizationId: result.organizationId,
|
||||||
agentName: result.agentName,
|
agentName: result.agentName,
|
||||||
|
agentKind: result.agentKind,
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
if (upgraded) return undefined;
|
if (upgraded) return undefined;
|
||||||
|
|
@ -157,6 +166,16 @@ export function createAgentManagerRuntime() {
|
||||||
websocket: {
|
websocket: {
|
||||||
open: (ws) => {
|
open: (ws) => {
|
||||||
setSession(ws.data.agentId, createSession(ws));
|
setSession(ws.data.agentId, createSession(ws));
|
||||||
|
void agentsService
|
||||||
|
.markAgentConnecting({
|
||||||
|
agentId: ws.data.agentId,
|
||||||
|
organizationId: ws.data.organizationId,
|
||||||
|
agentName: ws.data.agentName,
|
||||||
|
agentKind: ws.data.agentKind,
|
||||||
|
})
|
||||||
|
.catch((error) => {
|
||||||
|
logger.error(`Failed to mark agent ${ws.data.agentId} as connecting: ${toMessage(error)}`);
|
||||||
|
});
|
||||||
logger.info(`Agent "${ws.data.agentName}" (${ws.data.agentId}) connected on ${ws.data.id}`);
|
logger.info(`Agent "${ws.data.agentName}" (${ws.data.agentId}) connected on ${ws.data.id}`);
|
||||||
},
|
},
|
||||||
message: (ws, data) => {
|
message: (ws, data) => {
|
||||||
|
|
@ -179,6 +198,9 @@ export function createAgentManagerRuntime() {
|
||||||
},
|
},
|
||||||
close: (ws) => {
|
close: (ws) => {
|
||||||
removeSession(ws.data.agentId, ws.data.id);
|
removeSession(ws.data.agentId, ws.data.id);
|
||||||
|
void agentsService.markAgentOffline(ws.data.agentId).catch((error) => {
|
||||||
|
logger.error(`Failed to mark agent ${ws.data.agentId} as offline: ${toMessage(error)}`);
|
||||||
|
});
|
||||||
logger.info(`Agent "${ws.data.agentName}" (${ws.data.agentId}) disconnected`);
|
logger.info(`Agent "${ws.data.agentName}" (${ws.data.agentId}) disconnected`);
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,5 @@
|
||||||
import { Effect, Queue, Ref, type Scope } from "effect";
|
import { Effect, Queue, Ref, type Scope } from "effect";
|
||||||
|
import type { AgentKind } from "../../../db/schema";
|
||||||
import {
|
import {
|
||||||
createControllerMessage,
|
createControllerMessage,
|
||||||
parseAgentMessage,
|
parseAgentMessage,
|
||||||
|
|
@ -20,6 +21,7 @@ export type AgentConnectionData = {
|
||||||
agentId: string;
|
agentId: string;
|
||||||
organizationId: string | null;
|
organizationId: string | null;
|
||||||
agentName: string;
|
agentName: string;
|
||||||
|
agentKind: AgentKind;
|
||||||
};
|
};
|
||||||
|
|
||||||
type AgentSocket = Bun.ServerWebSocket<AgentConnectionData>;
|
type AgentSocket = Bun.ServerWebSocket<AgentConnectionData>;
|
||||||
|
|
@ -35,12 +37,22 @@ type TrackedBackupJob = {
|
||||||
state: "pending" | "active";
|
state: "pending" | "active";
|
||||||
};
|
};
|
||||||
|
|
||||||
|
type AgentRuntimeEventPayload = {
|
||||||
|
agentId: string;
|
||||||
|
agentName: string;
|
||||||
|
organizationId: string | null;
|
||||||
|
agentKind: AgentKind;
|
||||||
|
at: number;
|
||||||
|
};
|
||||||
|
|
||||||
type ControllerAgentSessionHandlers = {
|
type ControllerAgentSessionHandlers = {
|
||||||
onBackupStarted?: (payload: BackupStartedPayload) => void;
|
onReady: (payload: AgentRuntimeEventPayload) => Effect.Effect<void>;
|
||||||
onBackupProgress?: (payload: BackupProgressPayload) => void;
|
onHeartbeatPong: (payload: AgentRuntimeEventPayload) => Effect.Effect<void>;
|
||||||
onBackupCompleted?: (payload: BackupCompletedPayload) => void;
|
onBackupStarted: (payload: BackupStartedPayload) => Effect.Effect<void>;
|
||||||
onBackupFailed?: (payload: BackupFailedPayload) => void;
|
onBackupProgress: (payload: BackupProgressPayload) => Effect.Effect<void>;
|
||||||
onBackupCancelled?: (payload: BackupCancelledPayload) => void;
|
onBackupCompleted: (payload: BackupCompletedPayload) => Effect.Effect<void>;
|
||||||
|
onBackupFailed: (payload: BackupFailedPayload) => Effect.Effect<void>;
|
||||||
|
onBackupCancelled: (payload: BackupCancelledPayload) => Effect.Effect<void>;
|
||||||
};
|
};
|
||||||
|
|
||||||
export type ControllerAgentSession = {
|
export type ControllerAgentSession = {
|
||||||
|
|
@ -54,7 +66,7 @@ export type ControllerAgentSession = {
|
||||||
|
|
||||||
export const createControllerAgentSession = (
|
export const createControllerAgentSession = (
|
||||||
socket: AgentSocket,
|
socket: AgentSocket,
|
||||||
handlers: ControllerAgentSessionHandlers = {},
|
handlers: ControllerAgentSessionHandlers,
|
||||||
): Effect.Effect<ControllerAgentSession, never, Scope.Scope> =>
|
): Effect.Effect<ControllerAgentSession, never, Scope.Scope> =>
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
let isClosed = false;
|
let isClosed = false;
|
||||||
|
|
@ -105,9 +117,7 @@ export const createControllerAgentSession = (
|
||||||
for (const [jobId, trackedJob] of trackedJobs) {
|
for (const [jobId, trackedJob] of trackedJobs) {
|
||||||
const message = "The connection to the backup agent was lost. Restart the backup to ensure it completes.";
|
const message = "The connection to the backup agent was lost. Restart the backup to ensure it completes.";
|
||||||
|
|
||||||
yield* Effect.sync(() => {
|
yield* handlers.onBackupCancelled({ jobId, scheduleId: trackedJob.scheduleId, message });
|
||||||
handlers.onBackupCancelled?.({ jobId, scheduleId: trackedJob.scheduleId, message });
|
|
||||||
});
|
|
||||||
}
|
}
|
||||||
|
|
||||||
yield* Queue.shutdown(outboundQueue);
|
yield* Queue.shutdown(outboundQueue);
|
||||||
|
|
@ -177,11 +187,19 @@ export const createControllerAgentSession = (
|
||||||
|
|
||||||
const handleAgentMessage = (message: AgentMessage) =>
|
const handleAgentMessage = (message: AgentMessage) =>
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
yield* updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
|
|
||||||
|
|
||||||
switch (message.type) {
|
switch (message.type) {
|
||||||
case "agent.ready": {
|
case "agent.ready": {
|
||||||
yield* updateState((current) => ({ ...current, isReady: true }));
|
const readyAt = Date.now();
|
||||||
|
yield* updateState((current) => ({ ...current, isReady: true, lastSeenAt: readyAt }));
|
||||||
|
|
||||||
|
yield* handlers.onReady({
|
||||||
|
agentId: socket.data.agentId,
|
||||||
|
agentName: socket.data.agentName,
|
||||||
|
organizationId: socket.data.organizationId,
|
||||||
|
agentKind: socket.data.agentKind,
|
||||||
|
at: readyAt,
|
||||||
|
});
|
||||||
|
|
||||||
yield* Effect.sync(() => {
|
yield* Effect.sync(() => {
|
||||||
logger.info(`Agent "${socket.data.agentName}" (${socket.data.agentId}) is ready`);
|
logger.info(`Agent "${socket.data.agentName}" (${socket.data.agentId}) is ready`);
|
||||||
});
|
});
|
||||||
|
|
@ -192,43 +210,42 @@ export const createControllerAgentSession = (
|
||||||
scheduleId: message.payload.scheduleId,
|
scheduleId: message.payload.scheduleId,
|
||||||
state: "active",
|
state: "active",
|
||||||
});
|
});
|
||||||
yield* Effect.sync(() => {
|
logger.info(
|
||||||
logger.info(
|
`Backup ${message.payload.jobId} started on agent ${socket.data.agentId} for schedule ${message.payload.scheduleId}`,
|
||||||
`Backup ${message.payload.jobId} started on agent ${socket.data.agentId} for schedule ${message.payload.scheduleId}`,
|
);
|
||||||
);
|
yield* handlers.onBackupStarted(message.payload);
|
||||||
handlers.onBackupStarted?.(message.payload);
|
|
||||||
});
|
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
case "backup.progress": {
|
case "backup.progress": {
|
||||||
yield* Effect.sync(() => {
|
yield* handlers.onBackupProgress(message.payload);
|
||||||
handlers.onBackupProgress?.(message.payload);
|
|
||||||
});
|
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
case "backup.completed": {
|
case "backup.completed": {
|
||||||
yield* deleteTrackedBackupJob(message.payload.jobId);
|
yield* deleteTrackedBackupJob(message.payload.jobId);
|
||||||
yield* Effect.sync(() => {
|
yield* handlers.onBackupCompleted(message.payload);
|
||||||
handlers.onBackupCompleted?.(message.payload);
|
|
||||||
});
|
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
case "backup.failed": {
|
case "backup.failed": {
|
||||||
yield* deleteTrackedBackupJob(message.payload.jobId);
|
yield* deleteTrackedBackupJob(message.payload.jobId);
|
||||||
yield* Effect.sync(() => {
|
yield* handlers.onBackupFailed(message.payload);
|
||||||
handlers.onBackupFailed?.(message.payload);
|
|
||||||
});
|
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
case "backup.cancelled": {
|
case "backup.cancelled": {
|
||||||
yield* deleteTrackedBackupJob(message.payload.jobId);
|
yield* deleteTrackedBackupJob(message.payload.jobId);
|
||||||
yield* Effect.sync(() => {
|
yield* handlers.onBackupCancelled(message.payload);
|
||||||
handlers.onBackupCancelled?.(message.payload);
|
|
||||||
});
|
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
case "heartbeat.pong": {
|
case "heartbeat.pong": {
|
||||||
yield* updateState((current) => ({ ...current, lastPongAt: message.payload.sentAt }));
|
const seenAt = Date.now();
|
||||||
|
yield* updateState((current) => ({ ...current, lastSeenAt: seenAt, lastPongAt: message.payload.sentAt }));
|
||||||
|
|
||||||
|
yield* handlers.onHeartbeatPong({
|
||||||
|
agentId: socket.data.agentId,
|
||||||
|
agentName: socket.data.agentName,
|
||||||
|
organizationId: socket.data.organizationId,
|
||||||
|
agentKind: socket.data.agentKind,
|
||||||
|
at: seenAt,
|
||||||
|
});
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,5 @@
|
||||||
import { cryptoUtils } from "~/server/utils/crypto";
|
import { cryptoUtils } from "~/server/utils/crypto";
|
||||||
|
import { LOCAL_AGENT_ID, LOCAL_AGENT_KIND, LOCAL_AGENT_NAME } from "../constants";
|
||||||
|
|
||||||
export const deriveLocalAgentToken = async () => {
|
export const deriveLocalAgentToken = async () => {
|
||||||
return cryptoUtils.deriveSecret("zerobyte:local-agent-token");
|
return cryptoUtils.deriveSecret("zerobyte:local-agent-token");
|
||||||
|
|
@ -7,6 +8,6 @@ export const deriveLocalAgentToken = async () => {
|
||||||
export const validateAgentToken = async (token: string) => {
|
export const validateAgentToken = async (token: string) => {
|
||||||
const localToken = await deriveLocalAgentToken();
|
const localToken = await deriveLocalAgentToken();
|
||||||
if (token === localToken) {
|
if (token === localToken) {
|
||||||
return { agentId: "local", organizationId: null, agentName: "local" };
|
return { agentId: LOCAL_AGENT_ID, organizationId: null, agentName: LOCAL_AGENT_NAME, agentKind: LOCAL_AGENT_KIND };
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
|
||||||
|
|
@ -5,15 +5,14 @@ import { config } from "../../core/config";
|
||||||
import { restic, resticDeps } from "../../core/restic";
|
import { restic, resticDeps } from "../../core/restic";
|
||||||
import type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
|
import type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
|
||||||
import { agentManager, type BackupExecutionProgress } from "../agents/agents-manager";
|
import { agentManager, type BackupExecutionProgress } from "../agents/agents-manager";
|
||||||
|
import { LOCAL_AGENT_ID } from "../agents/constants";
|
||||||
import { getVolumePath } from "../volumes/helpers";
|
import { getVolumePath } from "../volumes/helpers";
|
||||||
import { decryptRepositoryConfig } from "../repositories/repository-config-secrets";
|
import { decryptRepositoryConfig } from "../repositories/repository-config-secrets";
|
||||||
import { createBackupOptions } from "./backup.helpers";
|
import { createBackupOptions } from "./backup.helpers";
|
||||||
import { toErrorDetails } from "../../utils/errors";
|
import { toErrorDetails } from "../../utils/errors";
|
||||||
|
|
||||||
const LOCAL_AGENT_ID = "local";
|
|
||||||
const FUSE_VOLUME_BACKENDS = new Set<Volume["type"]>(["rclone", "sftp", "webdav"]);
|
const FUSE_VOLUME_BACKENDS = new Set<Volume["type"]>(["rclone", "sftp", "webdav"]);
|
||||||
const IGNORE_INODE_FLAG = "--ignore-inode";
|
const IGNORE_INODE_FLAG = "--ignore-inode";
|
||||||
|
|
||||||
type BackupExecutionRequest = {
|
type BackupExecutionRequest = {
|
||||||
scheduleId: number;
|
scheduleId: number;
|
||||||
schedule: BackupSchedule;
|
schedule: BackupSchedule;
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,7 @@
|
||||||
import { runDbMigrations } from "../../db/db";
|
import { runDbMigrations } from "../../db/db";
|
||||||
import { config } from "../../core/config";
|
import { config } from "../../core/config";
|
||||||
import { startAgentController, startLocalAgent, stopAgentController, stopLocalAgent } from "../agents/agents-manager";
|
import { startAgentController, startLocalAgent, stopAgentController, stopLocalAgent } from "../agents/agents-manager";
|
||||||
|
import { agentsService } from "../agents/agents.service";
|
||||||
import { runMigrations } from "./migrations";
|
import { runMigrations } from "./migrations";
|
||||||
import { startup } from "./startup";
|
import { startup } from "./startup";
|
||||||
|
|
||||||
|
|
@ -9,6 +10,7 @@ let bootstrapPromise: Promise<void> | undefined;
|
||||||
const runBootstrap = async () => {
|
const runBootstrap = async () => {
|
||||||
await runDbMigrations();
|
await runDbMigrations();
|
||||||
await runMigrations();
|
await runMigrations();
|
||||||
|
await agentsService.ensureLocalAgent();
|
||||||
|
|
||||||
try {
|
try {
|
||||||
await startAgentController();
|
await startAgentController();
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue