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

refactor(sse): generic handler factory

Browse source
This commit is contained in:
Nicolas Meienberger 2026-02-14 12:21:39 +01:00
parent 00ddfe7f30
commit d528c5f518
1 changed files with 51 additions and 173 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

224
app/server/modules/events/events.controller.ts
View file

@ -1,17 +1,36 @@
import { Hono } from "hono"; import { Hono } from "hono";
import { streamSSE } from "hono/streaming"; import { streamSSE } from "hono/streaming";
import type {
ServerBackupCompletedEventDto,
ServerBackupProgressEventDto,
ServerBackupStartedEventDto,
ServerRestoreCompletedEventDto,
ServerRestoreProgressEventDto,
ServerRestoreStartedEventDto,
} from "~/schemas/events-dto";
import type { DoctorResult } from "~/schemas/restic";
import { serverEvents } from "../../core/events"; import { serverEvents } from "../../core/events";
import { logger } from "../../utils/logger"; import { logger } from "../../utils/logger";
import { requireAuth } from "../auth/auth.middleware"; import { requireAuth } from "../auth/auth.middleware";
import type { ServerEventPayloadMap } from "~/schemas/server-events";
type OrganizationScopedEvent = {
[EventName in keyof ServerEventPayloadMap]: ServerEventPayloadMap[EventName] extends {
organizationId: string;
}
? EventName
: never;
}[keyof ServerEventPayloadMap];
const broadcastEvents = [
"backup:started",
"backup:progress",
"backup:completed",
"volume:mounted",
"volume:unmounted",
"volume:updated",
"mirror:started",
"mirror:completed",
"restore:started",
"restore:progress",
"restore:completed",
"doctor:started",
"doctor:completed",
"doctor:cancelled",
] as const satisfies OrganizationScopedEvent[];
type BroadcastEvent = (typeof broadcastEvents)[number];
export const eventsController = new Hono().use(requireAuth).get("/", (c) => { export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
logger.info("Client connected to SSE endpoint"); logger.info("Client connected to SSE endpoint");
@ -23,155 +42,27 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
event: "connected", event: "connected",
}); });
const onBackupStarted = async (data: ServerBackupStartedEventDto) => { const createOrganizationEventHandler = <EventName extends BroadcastEvent>(event: EventName) => {
if (data.organizationId !== organizationId) return; return async (data: ServerEventPayloadMap[EventName]) => {
await stream.writeSSE({ if (data.organizationId !== organizationId) return;
data: JSON.stringify(data), await stream.writeSSE({
event: "backup:started", data: JSON.stringify(data),
}); event,
});
};
}; };
const onBackupProgress = async (data: ServerBackupProgressEventDto) => { const eventHandlers = broadcastEvents.reduce(
if (data.organizationId !== organizationId) return; (handlers, event) => {
await stream.writeSSE({ handlers[event] = createOrganizationEventHandler(event);
data: JSON.stringify(data), return handlers;
event: "backup:progress", },
}); {} as { [EventName in BroadcastEvent]: (data: ServerEventPayloadMap[EventName]) => Promise<void> },
}; );
const onBackupCompleted = async (data: ServerBackupCompletedEventDto) => { for (const event of broadcastEvents) {
if (data.organizationId !== organizationId) return; serverEvents.on(event, eventHandlers[event]);
await stream.writeSSE({ }
data: JSON.stringify(data),
event: "backup:completed",
});
};
const onVolumeMounted = async (data: { organizationId: string; volumeName: string }) => {
if (data.organizationId !== organizationId) return;
await stream.writeSSE({
data: JSON.stringify(data),
event: "volume:mounted",
});
};
const onVolumeUnmounted = async (data: { organizationId: string; volumeName: string }) => {
if (data.organizationId !== organizationId) return;
await stream.writeSSE({
data: JSON.stringify(data),
event: "volume:unmounted",
});
};
const onVolumeUpdated = async (data: { organizationId: string; volumeName: string }) => {
if (data.organizationId !== organizationId) return;
await stream.writeSSE({
data: JSON.stringify(data),
event: "volume:updated",
});
};
const onMirrorStarted = async (data: {
organizationId: string;
scheduleId: number;
repositoryId: string;
repositoryName: string;
}) => {
if (data.organizationId !== organizationId) return;
await stream.writeSSE({
data: JSON.stringify(data),
event: "mirror:started",
});
};
const onMirrorCompleted = async (data: {
organizationId: string;
scheduleId: number;
repositoryId: string;
repositoryName: string;
status: "success" | "error";
error?: string;
}) => {
if (data.organizationId !== organizationId) return;
await stream.writeSSE({
data: JSON.stringify(data),
event: "mirror:completed",
});
};
const onRestoreStarted = async (data: ServerRestoreStartedEventDto) => {
if (data.organizationId !== organizationId) return;
await stream.writeSSE({
data: JSON.stringify(data),
event: "restore:started",
});
};
const onRestoreProgress = async (data: ServerRestoreProgressEventDto) => {
if (data.organizationId !== organizationId) return;
await stream.writeSSE({
data: JSON.stringify(data),
event: "restore:progress",
});
};
const onRestoreCompleted = async (data: ServerRestoreCompletedEventDto) => {
if (data.organizationId !== organizationId) return;
await stream.writeSSE({
data: JSON.stringify(data),
event: "restore:completed",
});
};
const onDoctorStarted = async (data: { organizationId: string; repositoryId: string; repositoryName: string }) => {
if (data.organizationId !== organizationId) return;
await stream.writeSSE({
data: JSON.stringify(data),
event: "doctor:started",
});
};
const onDoctorCompleted = async (
data: {
organizationId: string;
repositoryId: string;
repositoryName: string;
} & DoctorResult,
) => {
if (data.organizationId !== organizationId) return;
await stream.writeSSE({
data: JSON.stringify(data),
event: "doctor:completed",
});
};
const onDoctorCancelled = async (data: {
organizationId: string;
repositoryId: string;
repositoryName: string;
error?: string;
}) => {
if (data.organizationId !== organizationId) return;
await stream.writeSSE({
data: JSON.stringify(data),
event: "doctor:cancelled",
});
};
serverEvents.on("backup:started", onBackupStarted);
serverEvents.on("backup:progress", onBackupProgress);
serverEvents.on("backup:completed", onBackupCompleted);
serverEvents.on("volume:mounted", onVolumeMounted);
serverEvents.on("volume:unmounted", onVolumeUnmounted);
serverEvents.on("volume:updated", onVolumeUpdated);
serverEvents.on("mirror:started", onMirrorStarted);
serverEvents.on("mirror:completed", onMirrorCompleted);
serverEvents.on("restore:started", onRestoreStarted);
serverEvents.on("restore:progress", onRestoreProgress);
serverEvents.on("restore:completed", onRestoreCompleted);
serverEvents.on("doctor:started", onDoctorStarted);
serverEvents.on("doctor:completed", onDoctorCompleted);
serverEvents.on("doctor:cancelled", onDoctorCancelled);
let keepAlive = true; let keepAlive = true;
let cleanedUp = false; let cleanedUp = false;
@ -181,20 +72,10 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
cleanedUp = true; cleanedUp = true;
c.req.raw.signal.removeEventListener("abort", onRequestAbort); c.req.raw.signal.removeEventListener("abort", onRequestAbort);
serverEvents.off("backup:started", onBackupStarted);
serverEvents.off("backup:progress", onBackupProgress); for (const event of broadcastEvents) {
serverEvents.off("backup:completed", onBackupCompleted); serverEvents.off(event, eventHandlers[event]);
serverEvents.off("volume:mounted", onVolumeMounted); }
serverEvents.off("volume:unmounted", onVolumeUnmounted);
serverEvents.off("volume:updated", onVolumeUpdated);
serverEvents.off("mirror:started", onMirrorStarted);
serverEvents.off("mirror:completed", onMirrorCompleted);
serverEvents.off("restore:started", onRestoreStarted);
serverEvents.off("restore:progress", onRestoreProgress);
serverEvents.off("restore:completed", onRestoreCompleted);
serverEvents.off("doctor:started", onDoctorStarted);
serverEvents.off("doctor:completed", onDoctorCompleted);
serverEvents.off("doctor:cancelled", onDoctorCancelled);
} }
function handleDisconnect() { function handleDisconnect() {
@ -214,10 +95,7 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
try { try {
while (keepAlive && !c.req.raw.signal.aborted && !stream.aborted) { while (keepAlive && !c.req.raw.signal.aborted && !stream.aborted) {
await stream.writeSSE({ await stream.writeSSE({ data: JSON.stringify({ timestamp: Date.now() }), event: "heartbeat" });
data: JSON.stringify({ timestamp: Date.now() }),
event: "heartbeat",
});
await stream.sleep(5000); await stream.sleep(5000);
} }
} finally { } finally {