refactor: make sse events isolated per org

app/client/hooks/use-server-events.ts

@@ -144,7 +144,7 @@ export function useServerEvents() {
 			});
 		});
 
-		eventSource.addEventListener("volume:status_updated", (e) => {
+		eventSource.addEventListener("volume:status_changed", (e) => {
 			const data = JSON.parse(e.data) as VolumeEvent;
 			console.info("[SSE] Volume status updated:", data);
 

app/server/__tests__/isolation.test.ts

@@ -2,13 +2,60 @@ import { test, describe, expect } from "bun:test";
 import { createApp } from "~/server/app";
 import { createTestSession } from "~/test/helpers/auth";
 import { db } from "~/server/db/db";
-import { repositoriesTable, volumesTable, backupSchedulesTable } from "~/server/db/schema";
+import {
+	repositoriesTable,
+	volumesTable,
+	backupSchedulesTable,
+	sessionsTable,
+	organization,
+	notificationDestinationsTable,
+	backupScheduleNotificationsTable,
+} from "~/server/db/schema";
 import crypto from "node:crypto";
 import { generateShortId } from "~/server/utils/id";
+import { eq } from "drizzle-orm";
 
 const app = createApp();
 
 describe("multi-organization isolation", () => {
+	test("should reject requests when session active organization is not a membership", async () => {
+		const session = await createTestSession();
+
+		// Create a different organization the user is NOT a member of
+		const foreignOrgId = crypto.randomUUID();
+		await db.insert(organization).values({
+			id: foreignOrgId,
+			name: `Org ${foreignOrgId}`,
+			slug: `test-org-${foreignOrgId}`,
+			createdAt: new Date(),
+		});
+		await db.insert(repositoriesTable).values({
+			id: crypto.randomUUID(),
+			shortId: generateShortId(),
+			name: "Foreign Repo",
+			type: "local",
+			config: { backend: "local", name: "foreign", path: "/tmp/repo-foreign" },
+			organizationId: foreignOrgId,
+		});
+
+		// Force the session to point at the foreign organization
+		const rawSessionToken = decodeURIComponent(session.token).split(".")[0];
+		await db
+			.update(sessionsTable)
+			.set({ activeOrganizationId: foreignOrgId })
+			.where(eq(sessionsTable.id, rawSessionToken));
+
+		const res = await app.request("/api/v1/repositories", {
+			headers: {
+				Cookie: `better-auth.session_token=${session.token}`,
+			},
+		});
+
+		expect(res.status).toBe(403);
+		const body = await res.json();
+		expect(body.message).toBe("Invalid organization context");
+	});
+
 	test("should not be able to access repositories from another organization", async () => {
 		const session1 = await createTestSession();
 		const session2 = await createTestSession();
@@ -204,4 +251,89 @@ describe("multi-organization isolation", () => {
 		});
 		expect(resOk.status).toBe(200);
 	});
+
+	test("should not be able to access or modify notifications for another organization's schedule", async () => {
+		const session1 = await createTestSession();
+		const session2 = await createTestSession();
+
+		const volId = Math.floor(Math.random() * 1000000);
+		await db.insert(volumesTable).values({
+			id: volId,
+			shortId: generateShortId(),
+			name: "Org 1 Volume",
+			type: "directory",
+			config: { backend: "directory", path: "/tmp/vol1" },
+			organizationId: session1.organizationId,
+			status: "unmounted",
+		});
+
+		const repoId = crypto.randomUUID();
+		await db.insert(repositoriesTable).values({
+			id: repoId,
+			shortId: generateShortId(),
+			name: "Org 1 Repo",
+			type: "local",
+			config: { backend: "local", name: "org1repo", path: "/tmp/repo1" },
+			organizationId: session1.organizationId,
+		});
+
+		const [schedule] = await db
+			.insert(backupSchedulesTable)
+			.values({
+				shortId: generateShortId(),
+				name: "Org 1 Schedule",
+				volumeId: volId,
+				repositoryId: repoId,
+				cronExpression: "0 0 * * *",
+				organizationId: session1.organizationId,
+			})
+			.returning();
+
+		const [destination] = await db
+			.insert(notificationDestinationsTable)
+			.values({
+				name: "Org 1 Destination",
+				enabled: true,
+				type: "discord",
+				config: { type: "discord", url: "https://example.com/webhook" },
+				organizationId: session1.organizationId,
+			})
+			.returning();
+
+		await db.insert(backupScheduleNotificationsTable).values({
+			scheduleId: schedule.id,
+			destinationId: destination.id,
+			notifyOnStart: true,
+			notifyOnSuccess: true,
+			notifyOnWarning: true,
+			notifyOnFailure: true,
+		});
+
+		const resGet = await app.request(`/api/v1/backups/${schedule.id}/notifications`, {
+			headers: {
+				Cookie: `better-auth.session_token=${session2.token}`,
+			},
+		});
+		expect(resGet.status).toBe(404);
+
+		const resPut = await app.request(`/api/v1/backups/${schedule.id}/notifications`, {
+			method: "PUT",
+			headers: {
+				Cookie: `better-auth.session_token=${session2.token}`,
+				"Content-Type": "application/json",
+			},
+			body: JSON.stringify({
+				assignments: [
+					{
+						destinationId: destination.id,
+						notifyOnStart: false,
+						notifyOnSuccess: false,
+						notifyOnWarning: false,
+						notifyOnFailure: false,
+					},
+				],
+			}),
+		});
+		expect(resPut.status).toBe(404);
+	});
 });

app/server/core/events.ts

@@ -6,8 +6,9 @@ import type { DoctorResult } from "~/schemas/restic";
  * Event payloads for the SSE system
  */
 interface ServerEvents {
-	"backup:started": (data: { scheduleId: number; volumeName: string; repositoryName: string }) => void;
+	"backup:started": (data: { organizationId: string; scheduleId: number; volumeName: string; repositoryName: string }) => void;
 	"backup:progress": (data: {
+		organizationId: string;
 		scheduleId: number;
 		volumeName: string;
 		repositoryName: string;
@@ -20,31 +21,44 @@ interface ServerEvents {
 		current_files: string[];
 	}) => void;
 	"backup:completed": (data: {
+		organizationId: string;
 		scheduleId: number;
 		volumeName: string;
 		repositoryName: string;
 		status: "success" | "error" | "stopped" | "warning";
 	}) => void;
-	"mirror:started": (data: { scheduleId: number; repositoryId: string; repositoryName: string }) => void;
+	"mirror:started": (data: {
+		organizationId: string;
+		scheduleId: number;
+		repositoryId: string;
+		repositoryName: string;
+	}) => void;
 	"mirror:completed": (data: {
+		organizationId: string;
 		scheduleId: number;
 		repositoryId: string;
 		repositoryName: string;
 		status: "success" | "error";
 		error?: string;
 	}) => void;
-	"volume:mounted": (data: { volumeName: string }) => void;
+	"volume:mounted": (data: { organizationId: string; volumeName: string }) => void;
-	"volume:unmounted": (data: { volumeName: string }) => void;
+	"volume:unmounted": (data: { organizationId: string; volumeName: string }) => void;
-	"volume:updated": (data: { volumeName: string }) => void;
+	"volume:updated": (data: { organizationId: string; volumeName: string }) => void;
-	"volume:status_changed": (data: { volumeName: string; status: string }) => void;
+	"volume:status_changed": (data: { organizationId: string; volumeName: string; status: string }) => void;
-	"doctor:started": (data: { repositoryId: string; repositoryName: string }) => void;
+	"doctor:started": (data: { organizationId: string; repositoryId: string; repositoryName: string }) => void;
 	"doctor:completed": (
 		data: {
+			organizationId: string;
 			repositoryId: string;
 			repositoryName: string;
 		} & DoctorResult,
 	) => void;
-	"doctor:cancelled": (data: { repositoryId: string; repositoryName: string; error?: string }) => void;
+	"doctor:cancelled": (data: {
+		organizationId: string;
+		repositoryId: string;
+		repositoryName: string;
+		error?: string;
+	}) => void;
 }
 
 /**

app/server/modules/auth/auth.middleware.ts

@@ -33,6 +33,14 @@ export const requireAuth = createMiddleware(async (c, next) => {
 		return c.json<unknown>({ message: "Invalid or expired session" }, 401);
 	}
 
+	const membership = await db.query.member.findFirst({
+		where: and(eq(member.userId, user.id), eq(member.organizationId, activeOrganizationId)),
+	});
+
+	if (!membership) {
+		return c.json({ message: "Invalid organization context" }, 403);
+	}
+
 	c.set("user", user);
 	c.set("organizationId", activeOrganizationId);
 

app/server/modules/backups/backups.service.ts

@@ -251,6 +251,7 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 	logger.info(`Starting backup ${schedule.name} for volume ${volume.name} to repository ${repository.name}`);
 
 	serverEvents.emit("backup:started", {
+		organizationId,
 		scheduleId,
 		volumeName: volume.name,
 		repositoryName: repository.name,
@@ -318,6 +319,7 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 				organizationId,
 				onProgress: (progress) => {
 					serverEvents.emit("backup:progress", {
+						organizationId,
 						scheduleId,
 						volumeName: volume.name,
 						repositoryName: repository.name,
@@ -367,6 +369,7 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 		}
 
 		serverEvents.emit("backup:completed", {
+			organizationId,
 			scheduleId,
 			volumeName: volume.name,
 			repositoryName: repository.name,
@@ -398,6 +401,7 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 			.where(and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId)));
 
 		serverEvents.emit("backup:completed", {
+			organizationId,
 			scheduleId,
 			volumeName: volume.name,
 			repositoryName: repository.name,
@@ -633,6 +637,7 @@ const copyToMirrors = async (
 			logger.info(`[Background] Copying to mirror repository: ${mirror.repository.name}`);
 
 			serverEvents.emit("mirror:started", {
+				organizationId,
 				scheduleId,
 				repositoryId: mirror.repositoryId,
 				repositoryName: mirror.repository.name,
@@ -665,6 +670,7 @@ const copyToMirrors = async (
 			logger.info(`[Background] Successfully copied to mirror repository: ${mirror.repository.name}`);
 
 			serverEvents.emit("mirror:completed", {
+				organizationId,
 				scheduleId,
 				repositoryId: mirror.repositoryId,
 				repositoryName: mirror.repository.name,
@@ -680,6 +686,7 @@ const copyToMirrors = async (
 				.where(eq(backupScheduleMirrorsTable.id, mirror.id));
 
 			serverEvents.emit("mirror:completed", {
+				organizationId,
 				scheduleId,
 				repositoryId: mirror.repositoryId,
 				repositoryName: mirror.repository.name,

app/server/modules/events/events.controller.ts

@@ -7,6 +7,7 @@ import type { DoctorResult } from "~/schemas/restic";
 
 export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
 	logger.info("Client connected to SSE endpoint");
+	const organizationId = c.get("organizationId");
 
 	return streamSSE(c, async (stream) => {
 		await stream.writeSSE({
@@ -14,7 +15,13 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
 			event: "connected",
 		});
 
-		const onBackupStarted = async (data: { scheduleId: number; volumeName: string; repositoryName: string }) => {
+		const onBackupStarted = async (data: {
+			organizationId: string;
+			scheduleId: number;
+			volumeName: string;
+			repositoryName: string;
+		}) => {
+			if (data.organizationId !== organizationId) return;
 			await stream.writeSSE({
 				data: JSON.stringify(data),
 				event: "backup:started",
@@ -22,6 +29,7 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
 		};
 
 		const onBackupProgress = async (data: {
+			organizationId: string;
 			scheduleId: number;
 			volumeName: string;
 			repositoryName: string;
@@ -33,6 +41,7 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
 			bytes_done: number;
 			current_files: string[];
 		}) => {
+			if (data.organizationId !== organizationId) return;
 			await stream.writeSSE({
 				data: JSON.stringify(data),
 				event: "backup:progress",
@@ -40,39 +49,50 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
 		};
 
 		const onBackupCompleted = async (data: {
+			organizationId: string;
 			scheduleId: number;
 			volumeName: string;
 			repositoryName: string;
 			status: "success" | "error" | "stopped" | "warning";
 		}) => {
+			if (data.organizationId !== organizationId) return;
 			await stream.writeSSE({
 				data: JSON.stringify(data),
 				event: "backup:completed",
 			});
 		};
 
-		const onVolumeMounted = async (data: { volumeName: string }) => {
+		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: { volumeName: string }) => {
+		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: { volumeName: string }) => {
+		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: { scheduleId: number; repositoryId: string; repositoryName: string }) => {
+		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",
@@ -80,19 +100,22 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
 		};
 
 		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 onDoctorStarted = async (data: { repositoryId: string; repositoryName: string }) => {
+		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",
@@ -101,17 +124,25 @@ export const eventsController = new Hono().use(requireAuth).get("/", (c) => {
 
 		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: { repositoryId: string; repositoryName: string; error?: string }) => {
+		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",

app/server/modules/notifications/notifications.service.ts

@@ -1,10 +1,11 @@
-import { eq, and, ne } from "drizzle-orm";
+import { eq, and, ne, inArray } from "drizzle-orm";
 import { ConflictError, InternalServerError, NotFoundError } from "http-errors-enhanced";
 import slugify from "slugify";
 import { db } from "../../db/db";
 import {
 	notificationDestinationsTable,
 	backupScheduleNotificationsTable,
+	backupSchedulesTable,
 	type NotificationDestination,
 } from "../../db/schema";
 import { cryptoUtils } from "../../utils/crypto";
@@ -256,6 +257,15 @@ const testDestination = async (id: number) => {
 };
 
 const getScheduleNotifications = async (scheduleId: number) => {
+	const organizationId = getOrganizationId();
+	const schedule = await db.query.backupSchedulesTable.findFirst({
+		where: and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId)),
+	});
+
+	if (!schedule) {
+		throw new NotFoundError("Backup schedule not found");
+	}
+
 	const assignments = await db.query.backupScheduleNotificationsTable.findMany({
 		where: eq(backupScheduleNotificationsTable.scheduleId, scheduleId),
 		with: {
@@ -263,7 +273,7 @@ const getScheduleNotifications = async (scheduleId: number) => {
 		},
 	});
 
-	return assignments;
+	return assignments.filter((a) => a.destination.organizationId === organizationId);
 };
 
 const updateScheduleNotifications = async (
@@ -276,6 +286,29 @@ const updateScheduleNotifications = async (
 		notifyOnFailure: boolean;
 	}>,
 ) => {
+	const organizationId = getOrganizationId();
+	const schedule = await db.query.backupSchedulesTable.findFirst({
+		where: and(eq(backupSchedulesTable.id, scheduleId), eq(backupSchedulesTable.organizationId, organizationId)),
+	});
+
+	if (!schedule) {
+		throw new NotFoundError("Backup schedule not found");
+	}
+
+	const destinationIds = [...new Set(assignments.map((a) => a.destinationId))];
+	if (destinationIds.length > 0) {
+		const destinations = await db.query.notificationDestinationsTable.findMany({
+			where: and(
+				inArray(notificationDestinationsTable.id, destinationIds),
+				eq(notificationDestinationsTable.organizationId, organizationId),
+			),
+		});
+
+		if (destinations.length !== destinationIds.length) {
+			throw new NotFoundError("One or more notification destinations were not found");
+		}
+	}
+
 	await db.delete(backupScheduleNotificationsTable).where(eq(backupScheduleNotificationsTable.scheduleId, scheduleId));
 
 	if (assignments.length > 0) {
@@ -305,6 +338,8 @@ const sendBackupNotification = async (
 	},
 ) => {
 	try {
+		const organizationId = getOrganizationId();
+
 		const assignments = await db.query.backupScheduleNotificationsTable.findMany({
 			where: eq(backupScheduleNotificationsTable.scheduleId, scheduleId),
 			with: {
@@ -313,6 +348,7 @@ const sendBackupNotification = async (
 		});
 
 		const relevantAssignments = assignments.filter((assignment) => {
+			if (assignment.destination.organizationId !== organizationId) return false;
 			if (!assignment.destination.enabled) return false;
 
 			switch (event) {

app/server/modules/repositories/doctor.ts

@@ -121,6 +121,7 @@ export const executeDoctor = async (
 	signal: AbortSignal,
 ) => {
 	const steps: DoctorStep[] = [];
+	const organizationId = getOrganizationId();
 
 	try {
 		const releaseLock = await repoMutex.acquireExclusive(repositoryId, "doctor", signal);
@@ -151,6 +152,7 @@ export const executeDoctor = async (
 		await saveDoctorResults(repositoryId, steps, finalStatus);
 
 		serverEvents.emit("doctor:completed", {
+			organizationId,
 			repositoryId,
 			repositoryName,
 			success: steps.every((s) => s.success),
@@ -176,6 +178,7 @@ export const executeDoctor = async (
 				.where(eq(repositoriesTable.id, repositoryId));
 
 			serverEvents.emit("doctor:cancelled", {
+				organizationId,
 				repositoryId,
 				repositoryName,
 				error: toMessage(error),
@@ -188,6 +191,7 @@ export const executeDoctor = async (
 
 			steps.push({ step: "doctor", success: false, output: null, error: toMessage(error) });
 			serverEvents.emit("doctor:completed", {
+				organizationId,
 				repositoryId,
 				repositoryName,
 				success: false,

app/server/modules/repositories/repositories.service.ts

@@ -366,6 +366,7 @@ const startDoctor = async (id: string) => {
 		await db.update(repositoriesTable).set({ status: "doctor" }).where(eq(repositoriesTable.id, repository.id));
 
 		serverEvents.emit("doctor:started", {
+			organizationId: repository.organizationId,
 			repositoryId: repository.id,
 			repositoryName: repository.name,
 		});
@@ -405,6 +406,7 @@ const cancelDoctor = async (id: string) => {
 	await db.update(repositoriesTable).set({ status: "unknown" }).where(eq(repositoriesTable.id, repository.id));
 
 	serverEvents.emit("doctor:cancelled", {
+		organizationId: repository.organizationId,
 		repositoryId: repository.id,
 		repositoryName: repository.name,
 	});

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

@@ -136,7 +136,7 @@ const mountVolume = async (idOrShortId: string | number) => {
 		.where(and(eq(volumesTable.id, volume.id), eq(volumesTable.organizationId, organizationId)));
 
 	if (status === "mounted") {
-		serverEvents.emit("volume:mounted", { volumeName: volume.name });
+		serverEvents.emit("volume:mounted", { organizationId, volumeName: volume.name });
 	}
 
 	return { error, status };
@@ -159,7 +159,7 @@ const unmountVolume = async (idOrShortId: string | number) => {
 		.where(and(eq(volumesTable.id, volume.id), eq(volumesTable.organizationId, organizationId)));
 
 	if (status === "unmounted") {
-		serverEvents.emit("volume:unmounted", { volumeName: volume.name });
+		serverEvents.emit("volume:unmounted", { organizationId, volumeName: volume.name });
 	}
 
 	return { error, status };
@@ -250,7 +250,7 @@ const updateVolume = async (idOrShortId: string | number, volumeData: UpdateVolu
 			.set({ status, lastError: error ?? null, lastHealthCheck: Date.now() })
 			.where(and(eq(volumesTable.id, existing.id), eq(volumesTable.organizationId, organizationId)));
 
-		serverEvents.emit("volume:updated", { volumeName: updated.name });
+		serverEvents.emit("volume:updated", { organizationId, volumeName: updated.name });
 	}
 
 	return { volume: updated };
@@ -301,7 +301,7 @@ const checkHealth = async (idOrShortId: string | number) => {
 	const { error, status } = await backend.checkHealth();
 
 	if (status !== volume.status) {
-		serverEvents.emit("volume:status_changed", { volumeName: volume.name, status });
+		serverEvents.emit("volume:status_changed", { organizationId, volumeName: volume.name, status });
 	}
 
 	await db