Mark /api/events endpoint as uncachable

The content of the events stream is generated live in response to
changes done in the system as they occur.  It therefore can't be
meaningfully cached or stored in any way.  Mark the response as
uncacheable to prevent potential problems with caches.

app.vue

@@ -11,7 +11,10 @@
 import "~/assets/global.css";
 const event = useRequestEvent();
 const sessionStore = useSessionStore();
-await callOnce("fetch-session", async () => {
+const eventsStore = useEventsStore();
+const nuxtApp = useNuxtApp();
+await callOnce("fetch-globals", async () => {
 	await sessionStore.fetch(event);
+	await nuxtApp.runWithContext(eventsStore.fetchLastEventId);
 })
 </script>

components/CardEvent.vue

@@ -3,7 +3,10 @@
 	SPDX-License-Identifier: AGPL-3.0-or-later
 -->
 <template>
-	<section class="event">
+	<section
+		class="event"
+		:class="{ cancelled: event.cancelled }"
+	>
 		<h3>{{ event.name }}</h3>
 		<p v-if=event.host>
 			Host: {{ event.host }}
@@ -36,7 +39,12 @@
 
 		<h4>Timeslots</h4>
 		<ul>
-			<li v-for="slot in event.slots.values()" :key="slot.id">
+			<li
+				v-for="slot in event.slots.values()"
+				:key="slot.id"
+				class="slot"
+				:class="{ cancelled: slot.cancelled }"
+			>
 				{{ formatTime(slot.start) }} - {{ formatTime(slot.end) }}
 				<button
 					v-if="accountStore.valid && event.slots.size > 1"
@@ -91,6 +99,10 @@ async function toggle(type: "event" | "slot", id: number, slotIds?: number[]) {
 	padding: 0.5rem;
 	border-radius: 0.5rem;
 }
+.event.cancelled>*, .slot.cancelled {
+	text-decoration: line-through;
+	color: grey;
+}
 .event h3 {
 	margin: 0;
 }
@@ -98,7 +110,9 @@ async function toggle(type: "event" | "slot", id: number, slotIds?: number[]) {
 	margin-block-start: 0.5rem;
 }
 
-.notice {
+.event .notice {
+	text-decoration: none;
+	color: CanvasText;
 	display: flex;
 	width: fit-content;
 	gap: 0.5rem;

components/CardEventSlot.vue

@@ -3,7 +3,10 @@
 	SPDX-License-Identifier: AGPL-3.0-or-later
 -->
 <template>
-	<section class="eventSlot">
+	<section
+		class="eventSlot"
+		:class="{ cancelled: slot.cancelled || event?.cancelled }"
+	>
 		<hgroup>
 			<h3>{{ event?.name }}</h3>
 			<p>
@@ -75,6 +78,10 @@ function formatTime(time: DateTime) {
 	padding: 0.5rem;
 	border-radius: 0.5rem;
 }
+.eventSlot.cancelled>* {
+	text-decoration: line-through;
+	color: grey;
+}
 .eventSlot h3 {
 	margin: 0;
 }
@@ -82,7 +89,9 @@ function formatTime(time: DateTime) {
 	margin-block-start: 0.5rem;
 }
 
-.notice {
+.eventSlot .notice {
+	text-decoration: none;
+	color: CanvasText;
 	display: flex;
 	width: fit-content;
 	gap: 0.5rem;

components/DiffSchedule.vue

@@ -61,7 +61,7 @@ const shifts = computed(() => {
 });
 </script>
 
-<style>
+<style scoped>
 h2 {
 	margin-block-start: 0.2rem;
 }

components/DiffScheduleEvent.vue

@@ -29,6 +29,12 @@
 			:after='event.crew ? "No" : "Yes"'
 			:state
 		/>
+		<DiffFieldString
+			title="Cancelled"
+			:before='event.serverCancelled ? "Yes" : "No"'
+			:after='event.cancelled ? "Yes" : "No"'
+			:state
+		/>
 		<DiffFieldString
 			title="Notice"
 			:before="event.serverNotice"

components/DiffScheduleEventSlot.vue

@@ -24,6 +24,12 @@
 			:entities="schedule.locations"
 			:state
 		/>
+		<DiffFieldString
+			title="Cancelled"
+			:before='slot.serverCancelled ? "Yes" : "No"'
+			:after='slot.cancelled ? "Yes" : "No"'
+			:state
+		/>
 		<DiffFieldSetEntityId
 			title="Assigned"
 			:before="slot.serverAssigned"

components/LogInDemo.vue

@@ -35,7 +35,3 @@ async function logIn() {
 	}
 }
 </script>
-
-<style>
-
-</style>

components/TableScheduleEventSlots.vue

@@ -12,6 +12,7 @@
 					<th>duration</th>
 					<th>event</th>
 					<th>location</th>
+					<th title="cancelled">c</th>
 					<th>assigned</th>
 					<th v-if="edit"></th>
 				</tr>
@@ -50,6 +51,7 @@
 								/>
 							</td>
 							<td></td>
+							<td></td>
 							<td>
 								Add at
 								<button
@@ -97,6 +99,11 @@
 									v-model="es.slot.locationIds"
 								/>
 							</td>
+							<td>
+								<input
+									type="checkbox"
+									v-model="es.slot.cancelled">
+							</td>
 							<td>
 								<SelectMultiEntity
 									:entities="usersStore.users"
@@ -149,6 +156,7 @@
 							/>
 						</td>
 						<td></td>
+						<td></td>
 						<td colspan="2">
 							<button
 								type="button"
@@ -177,6 +185,7 @@
 							<td>{{ es.end.diff(es.start).toFormat('hh:mm') }}</td>
 							<td>{{ es.event?.name }}</td>
 							<td></td>
+							<td>{{ es.slot.cancelled ? "x" : undefined }}</td>
 							<td><AssignedCrew :modelValue="es.slot.assigned" :edit="false" /></td>
 						</template>
 					</tr>

components/TableScheduleEvents.vue

@@ -14,6 +14,7 @@
 					<th>notice</th>
 					<th>description</th>
 					<th>p</th>
+					<th title="cancelled">c</th>
 					<th>s</th>
 					<th v-if="edit"></th>
 				</tr>
@@ -69,6 +70,13 @@
 								@change="event.crew = !event.crew"
 							>
 						</td>
+						<td>
+							<input
+								type="checkbox"
+								:disabled="!canEdit(event)"
+								v-model="event.cancelled"
+							>
+						</td>
 						<td>{{ event.slots.size ? event.slots.size : "" }}</td>
 						<td>
 							<button
@@ -123,6 +131,7 @@
 							>
 						</td>
 						<td></td>
+						<td></td>
 						<td>
 							<button
 								v-if="eventExists(newEventName)"

components/TableUsers.vue

@@ -61,7 +61,3 @@
 useEventSource();
 const usersStore = useUsersStore();
 </script>
-
-<style>
-
-</style>

components/Tabs.vue

@@ -55,7 +55,6 @@ nav {
 }
 .tab {
 	display: flex;
-	flex-wrap: wrap;
 }
 .tab.active {
 	padding-block-start: 1px;
@@ -65,6 +64,7 @@ nav {
 }
 .tab .spacer {
 	flex: 1 0 0.75rem;
+	width: 0.75rem;
 }
 .tab .flap,
 .tab .spacer {

components/Timetable.vue

@@ -102,7 +102,11 @@
 							v-for="cell, index in row"
 							:key="index"
 							:colSpan="cell.span"
-							:class='{"event": cell.slot, "crew": cell.event?.crew }'
+							:class="{
+								event: cell.slot,
+								crew: cell.event?.crew,
+								cancelled: cell.event?.cancelled || cell.slot?.cancelled,
+							}"
 							:title="cell.event?.name"
 						>
 							{{ cell.event?.notice ? "⚠️" : undefined }}
@@ -791,4 +795,9 @@ tr.hours>th + th.dayShift div {
 .event.crew {
 	background-color: color-mix(in oklab, var(--background), rgb(127, 127, 127) 60%);
 }
+
+.event.cancelled {
+	color: color-mix(in oklab, CanvasText, rgb(127, 127, 127) 80%);
+	text-decoration: line-through;
+}
 </style>

composables/event-source.ts

@@ -2,12 +2,12 @@
 	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
 	SPDX-License-Identifier: AGPL-3.0-or-later
 */
-import type { ApiEvent } from "~/shared/types/api";
+import type { ApiEvent, ApiEventStreamMessage } from "~/shared/types/api";
 
 interface AppEventMap {
 	"open": Event,
-	"message": MessageEvent<string>,
+	"message": MessageEvent<ApiEventStreamMessage>,
-	"update": MessageEvent<ApiEvent>,
+	"event": MessageEvent<ApiEvent>,
 	"error": Event,
 	"close": Event,
 }
@@ -18,12 +18,11 @@ class AppEventSource extends EventTarget {
 
 	#forwardEvent(type: string) {
 		this.#source!.addEventListener(type, event => {
-			if (type === "open" || type === "message" || type === "error") {
 			console.log("AppEventSource", event.type, event.data);
+			if (type === "open" || type === "error") {
 				this.dispatchEvent(new Event(event.type));
-			} else {
+			} else if (type === "message") {
-				const data = event.data ? JSON.parse(event.data) as ApiEvent : undefined;
+				const data = event.data ? JSON.parse(event.data) as ApiEventStreamMessage : undefined;
-				console.log("AppEventSource", event.type, data);
 				if (data?.type === "connected") {
 					this.#sourceSessionId = data.session?.id;
 				}
@@ -34,17 +33,27 @@ class AppEventSource extends EventTarget {
 					source: event.source,
 					ports: [...event.ports],
 				}));
+			} else {
+				const data = event.data ? JSON.parse(event.data) as ApiEvent : undefined;
+				this.dispatchEvent(new MessageEvent(event.type, {
+					data,
+					origin: event.origin,
+					lastEventId: event.lastEventId,
+					source: event.source,
+					ports: [...event.ports],
+				}));
 			}
 		});
 	}
 
-	open(sessionId: number | undefined) {
+	open(sessionId: number | undefined, lastEventId: number) {
 		console.log("Opening event source sid:", sessionId);
 		this.#sourceSessionId = sessionId;
-		this.#source = new EventSource("/api/events");
+		const query = new URLSearchParams({ lastEventId: String(lastEventId) });
+		this.#source = new EventSource(`/api/events?${query}`);
 		this.#forwardEvent("open");
 		this.#forwardEvent("message");
-		this.#forwardEvent("update");
+		this.#forwardEvent("event");
 		this.#forwardEvent("error");
 	}
 
@@ -58,20 +67,20 @@ class AppEventSource extends EventTarget {
 	}
 
 	#connectRefs = 0;
-	connect(sessionId: number | undefined) {
+	connect(sessionId: number | undefined, lastEventId: number) {
 		this.#connectRefs += 1;
 		if (this.#source && this.#sourceSessionId !== sessionId) {
 			this.close();
 		}
 		if (!this.#source) {
-			this.open(sessionId);
+			this.open(sessionId, lastEventId);
 		}
 	}
 
-	reconnect(sessionId: number | undefined) {
+	reconnect(sessionId: number | undefined, lastEventId: number) {
 		if (this.#source && this.#sourceSessionId !== sessionId) {
 			this.close();
-			this.open(sessionId);
+			this.open(sessionId, lastEventId);
 		}
 	}
 
@@ -113,14 +122,15 @@ export const appEventSource = import.meta.client ? new AppEventSource() : null;
 
 export function useEventSource() {
 	const sessionStore = useSessionStore();
+	const eventsStore = useEventsStore();
 	onMounted(() => {
 		console.log("useEventSource onMounted", sessionStore.id);
-		appEventSource!.connect(sessionStore.id);
+		appEventSource!.connect(sessionStore.id, eventsStore.lastEventId);
 	})
 
 	watch(() => sessionStore.id, () => {
 		console.log("useEventSource sessionStore.id change", sessionStore.id);
-		appEventSource!.reconnect(sessionStore.id);
+		appEventSource!.reconnect(sessionStore.id, eventsStore.lastEventId);
 	})
 
 	onUnmounted(() => {

docs/dev/server-sent-events.md

@@ -11,3 +11,9 @@ To update in real time this application sends a `text/event-source` stream using
 Upon connecting a `"connect"` event is emitted with the session the connection was made under. This is the primary mechanism a user agent discovers its own session having been rotated into a new one, which also happens when the access level of the account associated with the session changes.
 
 After the `"connect"` event the user agent will start to receive updates to resources it has access to that has changed. There is no filtering for what resoucres the user agent receives updates for at the moment as there's not enough events to justify the complexity of server-side subscriptions and filtering.
+
+## Id and order
+
+Events are guaranteed to be delivered in order, and to maintain consistency the server provides the following guarantee: Any entities fetched after receiving a response from `/api/last-event-id` will include updates from all events up to and including the `id` received from the response.
+
+This means that a client can fetch an up to date and live representation of any API entity by first fetching the last event from `/api/last-event-id`, and then in parallel fetch any entities as well as opening the `/api/events` stream with the `lastEventId` query param set to the value received from the `/api/last-event-id` endpoint.

pages/admin/index.vue

@@ -109,7 +109,3 @@ const tabs = [
 	{ id: "database", title: "Database" },
 ];
 </script>
-
-<style>
-
-</style>

pages/admin/users/[id].vue

@@ -84,7 +84,7 @@ const { pending, data, error } = await useFetch(() => `/api/users/${id.value}/de
 const userDetails = data as Ref<ApiUserDetails | ApiTombstone>;
 </script>
 
-<style>
+<style scoped>
 dl {
 	display: grid;
 	grid-template-columns: auto 1fr;

server/api/admin/database-export.post.ts

@@ -9,11 +9,11 @@ export default defineEventHandler(async (event) => {
 	setHeader(event, "Content-Disposition", 'attachment; filename="database-dump.json"');
 	setHeader(event, "Content-Type", "application/json; charset=utf-8");
 	return {
-		nextUserId: await readNextUserId(),
+		nextUserId: readNextUserId(),
-		users: await readUsers(),
+		users: readUsers(),
-		nextSessionId: await readNextSessionId(),
+		nextSessionId: readNextSessionId(),
-		sessions: await readSessions(),
+		sessions: readSessions(),
-		subscriptions: await readSubscriptions(),
+		subscriptions: readSubscriptions(),
-		schedule: await readSchedule(),
+		schedule: readSchedule(),
 	};
 })

server/api/admin/database-import-demo.post.ts

@@ -7,14 +7,14 @@ import { generateDemoSchedule, generateDemoAccounts } from "~/server/generate-de
 export default defineEventHandler(async (event) => {
 	await requireServerSessionWithAdmin(event);
 	const accounts = generateDemoAccounts();
-	await writeUsers(accounts);
+	writeUsers(accounts);
-	await writeSchedule(generateDemoSchedule());
+	writeSchedule(generateDemoSchedule());
-	await writeAuthenticationMethods(accounts.map((user, index) => ({
+	writeAuthenticationMethods(accounts.map((user, index) => ({
 		id: index,
 		userId: user.id,
 		provider: "demo",
 		slug: user.name!,
 		name: user.name!,
 	})));
-	await writeNextAuthenticationMethodId(Math.max(await nextAuthenticationMethodId(), accounts.length));
+	writeNextAuthenticationMethodId(Math.max(nextAuthenticationMethodId(), accounts.length));
 })

server/api/admin/database-import.post.ts

@@ -29,20 +29,20 @@ export default defineEventHandler(async (event) => {
 		});
 	}
 
-	const currentNextUserId = await readNextUserId();
+	const currentNextUserId = readNextUserId();
-	await writeNextUserId(Math.max(currentNextUserId, snapshot.nextUserId));
+	writeNextUserId(Math.max(currentNextUserId, snapshot.nextUserId));
-	await writeUsers(snapshot.users);
+	writeUsers(snapshot.users);
-	const currentNextSessionId = await readNextSessionId();
+	const currentNextSessionId = readNextSessionId();
-	await writeNextSessionId(Math.max(currentNextSessionId, snapshot.nextSessionId));
+	writeNextSessionId(Math.max(currentNextSessionId, snapshot.nextSessionId));
-	const currentSessions = new Map((await readSessions()).map(session => [session.id, session]));
+	const currentSessions = new Map((readSessions()).map(session => [session.id, session]));
-	await writeSessions(snapshot.sessions.filter(session => {
+	writeSessions(snapshot.sessions.filter(session => {
 		const current = currentSessions.get(session.id);
 		// Only keep sessions that match the account id in both sets to avoid
 		// resurrecting deleted sessions. This will still cause session cross
 		// pollution if a snapshot from another instance is loaded here.
 		return current?.accountId !== undefined && current.accountId === session.accountId;
 	}));
-	await writeSubscriptions(snapshot.subscriptions);
+	writeSubscriptions(snapshot.subscriptions);
-	await writeSchedule(snapshot.schedule);
+	writeSchedule(snapshot.schedule);
 	await sendRedirect(event, "/");
 })

server/api/admin/delete-database.post.ts

@@ -6,5 +6,5 @@ import { deleteDatabase } from "~/server/database";
 
 export default defineEventHandler(async (event) => {
 	await requireServerSessionWithAdmin(event);
-	await deleteDatabase();
+	deleteDatabase();
 })

server/api/admin/user.patch.ts

@@ -2,7 +2,7 @@
 	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
 	SPDX-License-Identifier: AGPL-3.0-or-later
 */
-import { readSessions, readUsers, writeSessions, writeUsers } from "~/server/database";
+import { nextEventId, readSessions, readUsers, writeSessions, writeUsers } from "~/server/database";
 import { apiUserPatchSchema } from "~/shared/types/api";
 import { z } from "zod/v4-mini";
 import { broadcastEvent } from "~/server/streams";
@@ -18,7 +18,7 @@ export default defineEventHandler(async (event) => {
 		});
 	}
 
-	const users = await readUsers();
+	const users = readUsers();
 	const user = users.find(user => user.id === patch.id);
 	if (!user || user.deleted) {
 		throw createError({
@@ -52,26 +52,28 @@ export default defineEventHandler(async (event) => {
 		user.name = patch.name;
 	}
 	user.updatedAt = new Date().toISOString();
-	await writeUsers(users);
+	writeUsers(users);
 	broadcastEvent({
+		id: nextEventId(),
 		type: "user-update",
 		data: serverUserToApi(user),
 	});
 
 	// Rotate sessions with the user in it if the access changed
 	if (accessChanged) {
-		const sessions = await readSessions();
+		const sessions = readSessions();
 		const nowMs = Date.now();
 		for (const session of sessions) {
 			if (session.accountId === user.id) {
 				session.rotatesAtMs = nowMs;
 				broadcastEvent({
+					id: nextEventId(),
 					type: "session-expired",
 					sessionId: session.id,
 				});
 			}
 		}
-		await writeSessions(sessions);
+		writeSessions(sessions);
 	}
 
 	// Update Schedule counts.

server/api/auth/account.delete.ts

@@ -5,16 +5,17 @@
 import {
 	readUsers, readSessions, readSubscriptions,
 	writeUsers, writeSessions, writeSubscriptions,
+	nextEventId,
 } from "~/server/database";
 import { broadcastEvent, cancelAccountStreams } from "~/server/streams";
 
 export default defineEventHandler(async (event) => {
 	const serverSession = await requireServerSessionWithUser(event);
-	let users = await readUsers();
+	let users = readUsers();
 
 	// Expire sessions for this user
 	const expiredSessionIds = new Set<number>();
-	let sessions = await readSessions();
+	let sessions = readSessions();
 	const nowMs = Date.now();
 	for (const session of sessions) {
 		if (
@@ -24,6 +25,7 @@ export default defineEventHandler(async (event) => {
 		) {
 			session.expiresAtMs = nowMs;
 			broadcastEvent({
+				id: nextEventId(),
 				type: "session-expired",
 				sessionId: session.id,
 			});
@@ -31,23 +33,24 @@ export default defineEventHandler(async (event) => {
 		}
 	}
 	cancelAccountStreams(serverSession.accountId);
-	await writeSessions(sessions);
+	writeSessions(sessions);
 	await deleteCookie(event, "session");
 
 	// Remove subscriptions for this user
-	let subscriptions = await readSubscriptions();
+	let subscriptions = readSubscriptions();
 	subscriptions = subscriptions.filter(
 		subscription => !expiredSessionIds.has(subscription.sessionId)
 	);
-	await writeSubscriptions(subscriptions);
+	writeSubscriptions(subscriptions);
 
 	// Remove the user
 	const account = users.find(user => user.id === serverSession.accountId)!;
 	const now = new Date(nowMs).toISOString();
 	account.deleted = true;
 	account.updatedAt = now;
-	await writeUsers(users);
+	writeUsers(users);
 	await broadcastEvent({
+		id: nextEventId(),
 		type: "user-update",
 		data: {
 			id: account.id,

server/api/auth/account.patch.ts

@@ -38,7 +38,7 @@ export default defineEventHandler(async (event) => {
 		}
 	}
 
-	const users = await readUsers();
+	const users = readUsers();
 	const account = users.find(user => user.id === session.accountId);
 	if (!account) {
 		throw Error("Account does not exist");
@@ -70,7 +70,7 @@ export default defineEventHandler(async (event) => {
 		else
 			delete account.locale;
 	}
-	await writeUsers(users);
+	writeUsers(users);
 
 	// Update Schedule counts.
 	await updateScheduleInterestedCounts(users);

server/api/auth/account.post.ts

@@ -2,7 +2,7 @@
 	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
 	SPDX-License-Identifier: AGPL-3.0-or-later
 */
-import { readUsers, writeUsers, nextUserId, type ServerUser, readAuthenticationMethods, nextAuthenticationMethodId, writeAuthenticationMethods } from "~/server/database";
+import { readUsers, writeUsers, nextUserId, type ServerUser, readAuthenticationMethods, nextAuthenticationMethodId, writeAuthenticationMethods, nextEventId } from "~/server/database";
 import { broadcastEvent } from "~/server/streams";
 import type { ApiSession } from "~/shared/types/api";
 
@@ -18,7 +18,7 @@ export default defineEventHandler(async (event): Promise<ApiSession> => {
 	const body = await readBody(event);
 	const name = body?.name;
 
-	const users = await readUsers();
+	const users = readUsers();
 	let user: ServerUser;
 	if (typeof name === "string") {
 		if (name === "") {
@@ -36,7 +36,7 @@ export default defineEventHandler(async (event): Promise<ApiSession> => {
 
 		const firstUser = users.every(user => user.type === "anonymous");
 		user = {
-			id: await nextUserId(),
+			id: nextUserId(),
 			updatedAt: new Date().toISOString(),
 			type: firstUser ? "admin" : "regular",
 			name,
@@ -44,7 +44,7 @@ export default defineEventHandler(async (event): Promise<ApiSession> => {
 
 	} else if (name === undefined) {
 		user = {
-			id: await nextUserId(),
+			id: nextUserId(),
 			updatedAt: new Date().toISOString(),
 			type: "anonymous",
 		};
@@ -76,18 +76,19 @@ export default defineEventHandler(async (event): Promise<ApiSession> => {
 			});
 		}
 		authMethods.push({
-			id: await nextAuthenticationMethodId(),
+			id: nextAuthenticationMethodId(),
 			userId: user.id,
 			provider: session.authenticationProvider,
 			slug: session.authenticationSlug!,
 			name: session.authenticationName!,
 		})
-		await writeAuthenticationMethods(authMethods);
+		writeAuthenticationMethods(authMethods);
 	}
 
 	users.push(user);
-	await writeUsers(users);
+	writeUsers(users);
 	await broadcastEvent({
+		id: nextEventId(),
 		type: "user-update",
 		data: user,
 	});

server/api/auth/ap/demo-login.post.ts

@@ -24,11 +24,11 @@ export default defineEventHandler(async (event) => {
 		});
 	}
 
-	const authMethods = await readAuthenticationMethods();
+	const authMethods = readAuthenticationMethods();
 	const method = authMethods.find(method => method.provider === "demo" && method.slug === slug);
 	let session;
 	if (method) {
-		const users = await readUsers();
+		const users = readUsers();
 		const account = users.find(user => !user.deleted && user.id === method.userId);
 		session = await setServerSession(event, account);
 	} else {

server/api/auth/ap/telegram-login.post.ts

@@ -84,11 +84,11 @@ export default defineEventHandler(async (event): Promise<ApiSession> => {
 	}
 
 	const slug = String(data.authData.id);
-	const authMethods = await readAuthenticationMethods();
+	const authMethods = readAuthenticationMethods();
 	const method = authMethods.find(method => method.provider === "telegram" && method.slug === slug);
 	let session;
 	if (method) {
-		const users = await readUsers();
+		const users = readUsers();
 		const account = users.find(user => !user.deleted && user.id === method.userId);
 		session = await setServerSession(event, account);
 	} else {

server/api/auth/session.delete.ts

@@ -8,7 +8,7 @@ import { cancelSessionStreams } from "~/server/streams";
 export default defineEventHandler(async (event) => {
 	const session = await getServerSession(event, true);
 	if (session) {
-		const users = await readUsers();
+		const users = readUsers();
 		const account = users.find(user => user.id === session.accountId);
 		if (account?.type === "anonymous") {
 			throw createError({

server/api/events.ts

@@ -3,33 +3,47 @@
 	SPDX-License-Identifier: AGPL-3.0-or-later
 */
 import { pipeline } from "node:stream";
-import { addStream, deleteStream } from "~/server/streams";
+import { createEventStream } from "~/server/streams";
 
 export default defineEventHandler(async (event) => {
 	const session = await getServerSession(event, false);
 
-	const encoder = new TextEncoder();
+	let lastEventId: number | undefined;
-	const source = event.headers.get("x-forwarded-for");
+	const lastEventIdHeader = event.headers.get("Last-Event-ID");
-	console.log(`starting event stream for ${source}`)
+	const lastEventIdQuery = getQuery(event)["lastEventId"];
-	const stream = new TransformStream<string, Uint8Array>({
+	if (lastEventIdHeader) {
-		transform(chunk, controller) {
+		if (!/^[0-9]{1,15}$/.test(lastEventIdHeader)) {
-			controller.enqueue(encoder.encode(chunk));
+			throw createError({
-		},
+				statusCode: 400,
-		flush(controller) {
+				statusMessage: "Bad Request",
-			console.log(`finished event stream for ${source}`);
+				message: "Malformed Last-Event-ID header",
-			deleteStream(stream.writable);
-		},
-		// @ts-expect-error experimental API
-		cancel(reason) {
-			console.log(`cancelled event stream for ${source}`);
-			deleteStream(stream.writable);
-		}
 			});
-	addStream(event, stream.writable, session);
+		}
+		lastEventId = Number.parseInt(lastEventIdHeader, 10);
+	} else if (lastEventIdQuery) {
+		if (typeof lastEventIdQuery !== "string" || !/^[0-9]{1,15}$/.test(lastEventIdQuery)) {
+			throw createError({
+				statusCode: 400,
+				statusMessage: "Bad Request",
+				message: "Malformed lastEventId",
+			});
+		}
+		lastEventId = Number.parseInt(lastEventIdQuery, 10);
+	} else {
+		throw createError({
+			statusCode: 400,
+			statusMessage: "Bad Request",
+			message: "lastEventId is required",
+		});
+	}
+
+	const source = event.headers.get("x-forwarded-for") ?? "";
+	const stream = await createEventStream(event, source, lastEventId, session);
 
 	// Workaround to properly handle stream errors. See https://github.com/unjs/h3/issues/986
 	setHeader(event, "Access-Control-Allow-Origin", "*");
 	setHeader(event, "Content-Type", "text/event-stream");
-	pipeline(stream.readable as unknown as NodeJS.ReadableStream, event.node.res, (err) => { /* ignore */ });
+	setHeader(event, "Cache-Control", "no-store");
+	pipeline(stream as unknown as NodeJS.ReadableStream, event.node.res, (err) => { /* ignore */ });
 	event._handled = true;
 });

server/api/last-event-id.ts

@@ -0,0 +1,11 @@
+/*
+	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
+	SPDX-License-Identifier: AGPL-3.0-or-later
+*/
+
+import { readEvents } from "../database";
+
+export default defineEventHandler(async (event) => {
+	const events = readEvents();
+	return events[events.length - 1]?. id ?? 0;
+});

server/api/schedule.patch.ts

@@ -3,7 +3,7 @@
 	SPDX-License-Identifier: AGPL-3.0-or-later
 */
 import { z } from "zod/v4-mini";
-import { readSchedule, writeSchedule } from "~/server/database";
+import { nextEventId, readSchedule, writeSchedule } from "~/server/database";
 import { broadcastEvent } from "~/server/streams";
 import { apiScheduleSchema } from "~/shared/types/api";
 import { applyUpdatesToArray } from "~/shared/utils/update";
@@ -35,7 +35,7 @@ export default defineEventHandler(async (event) => {
 		});
 	}
 
-	const schedule = await readSchedule();
+	const schedule = readSchedule();
 
 	if (schedule.deleted) {
 		throw createError({
@@ -85,8 +85,9 @@ export default defineEventHandler(async (event) => {
 		applyUpdatesToArray(update.shifts, schedule.shifts = schedule.shifts ?? []);
 	}
 
-	await writeSchedule(schedule);
+	writeSchedule(schedule);
 	await broadcastEvent({
+		id: nextEventId(),
 		type: "schedule-update",
 		updatedFrom,
 		data: update,

server/api/schedule.ts

@@ -6,6 +6,6 @@ import { readSchedule } from "~/server/database";
 
 export default defineEventHandler(async (event) => {
 	const session = await getServerSession(event, false);
-	const schedule = await readSchedule();
+	const schedule = readSchedule();
 	return canSeeCrew(session?.access) ? schedule : filterSchedule(schedule);
 });

server/api/subscribe.post.ts

@@ -20,7 +20,7 @@ export default defineEventHandler(async (event) => {
 			message: z.prettifyError(error),
 		});
 	}
-	const subscriptions = await readSubscriptions();
+	const subscriptions = readSubscriptions();
 	const existingIndex = subscriptions.findIndex(
 		sub => sub.type === "push" && sub.sessionId === session.id
 	);
@@ -34,7 +34,7 @@ export default defineEventHandler(async (event) => {
 	} else {
 		subscriptions.push(subscription);
 	}
-	await writeSubscriptions(subscriptions);
+	writeSubscriptions(subscriptions);
 	if (existingIndex !== -1) {
 		return { message: "Existing subscription refreshed."};
 	}

server/api/unsubscribe.post.ts

@@ -6,7 +6,7 @@ import { readSubscriptions, writeSubscriptions } from "~/server/database";
 
 export default defineEventHandler(async (event) => {
 	const session = await requireServerSessionWithUser(event);
-	const subscriptions = await readSubscriptions();
+	const subscriptions = readSubscriptions();
 	const existingIndex = subscriptions.findIndex(
 		sub => sub.type === "push" && sub.sessionId === session.id
 	);
@@ -15,6 +15,6 @@ export default defineEventHandler(async (event) => {
 	} else {
 		return { message: "No subscription registered."};
 	}
-	await writeSubscriptions(subscriptions);
+	writeSubscriptions(subscriptions);
 	return { message: "Existing subscription removed."};
 });

server/api/users/[id]/details.get.ts

@@ -13,7 +13,7 @@ const detailsSchema = z.object({
 
 export default defineEventHandler(async (event) => {
 	await requireServerSessionWithAdmin(event);
-	const users = await readUsers();
+	const users = readUsers();
 	const { success, error, data: params } = detailsSchema.safeParse(getRouterParams(event));
 	if (!success) {
 		throw createError({

server/api/users/index.get.ts

@@ -6,7 +6,7 @@ import { readUsers } from "~/server/database"
 
 export default defineEventHandler(async (event) => {
 	const session = await requireServerSessionWithUser(event);
-	const users = await readUsers();
+	const users = readUsers();
 
 	if (session.access === "admin") {
 		return users.map(serverUserToApi);

server/database.ts

@@ -2,8 +2,8 @@
 	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
 	SPDX-License-Identifier: AGPL-3.0-or-later
 */
-import { readFile, unlink, writeFile } from "node:fs/promises";
+import { readFileSync, writeFileSync, unlinkSync } from "node:fs";
-import type { ApiAuthenticationProvider, ApiSchedule, ApiSubscription, ApiUserType } from "~/shared/types/api";
+import type { ApiAuthenticationProvider, ApiEvent, ApiSchedule, ApiSubscription, ApiUserType } from "~/shared/types/api";
 import type { Id } from "~/shared/types/common";
 
 export interface ServerSession {
@@ -50,10 +50,12 @@ const sessionsPath = "data/sessions.json";
 const nextSessionIdPath = "data/next-session-id.json";
 const authMethodPath = "data/auth-method.json";
 const nextAuthenticationMethodIdPath = "data/auth-method-id.json"
+const nextEventIdPath = "data/next-event-id.json";
+const eventsPath = "data/events.json";
 
-async function remove(path: string) {
+function remove(path: string) {
 	try {
-		await unlink(path);
+		unlinkSync(path);
 	} catch (err: any) {
 		if (err.code !== "ENOENT") {
 			throw err;
@@ -61,17 +63,17 @@ async function remove(path: string) {
 	}
 }
 
-export async function deleteDatabase() {
+export function deleteDatabase() {
-	await remove(schedulePath);
+	remove(schedulePath);
-	await remove(subscriptionsPath);
+	remove(subscriptionsPath);
-	await remove(usersPath);
+	remove(usersPath);
-	await remove(sessionsPath);
+	remove(sessionsPath);
 }
 
-async function readJson<T>(filePath: string, fallback: T) {
+function readJson<T>(filePath: string, fallback: T) {
 	let data: T extends () => infer R ? R : T;
 	try {
-		data = JSON.parse(await readFile(filePath, "utf-8"));
+		data = JSON.parse(readFileSync(filePath, "utf-8"));
 	} catch (err: any) {
 		if (err.code !== "ENOENT")
 			throw err;
@@ -80,19 +82,19 @@ async function readJson<T>(filePath: string, fallback: T) {
 	return data;
 }
 
-export async function readSchedule() {
+export function readSchedule() {
 	return readJson(schedulePath, (): ApiSchedule => ({
 		id: 111,
 		updatedAt: new Date().toISOString(),
 	}));
 }
 
-export async function writeSchedule(schedule: ApiSchedule) {
+export function writeSchedule(schedule: ApiSchedule) {
-	await writeFile(schedulePath, JSON.stringify(schedule, undefined, "\t") + "\n", "utf-8");
+	writeFileSync(schedulePath, JSON.stringify(schedule, undefined, "\t") + "\n", "utf-8");
 }
 
-export async function readSubscriptions() {
+export function readSubscriptions() {
-	let subscriptions = await readJson<ApiSubscription[]>(subscriptionsPath, []);
+	let subscriptions = readJson<ApiSubscription[]>(subscriptionsPath, []);
 	if (subscriptions.length && "keys" in subscriptions[0]) {
 		// Discard old format
 		subscriptions = [];
@@ -100,71 +102,89 @@ export async function readSubscriptions() {
 	return subscriptions;
 }
 
-export async function writeSubscriptions(subscriptions: ApiSubscription[]) {
+export function writeSubscriptions(subscriptions: ApiSubscription[]) {
-	await writeFile(subscriptionsPath, JSON.stringify(subscriptions, undefined, "\t") + "\n", "utf-8");
+	writeFileSync(subscriptionsPath, JSON.stringify(subscriptions, undefined, "\t") + "\n", "utf-8");
 }
 
-export async function readNextUserId() {
+export function readNextUserId() {
-	return await readJson(nextUserIdPath, 0);
+	return readJson(nextUserIdPath, 0);
 }
 
-export async function writeNextUserId(nextId: number) {
+export function writeNextUserId(nextId: number) {
-	await writeFile(nextUserIdPath, String(nextId), "utf-8");
+	writeFileSync(nextUserIdPath, String(nextId), "utf-8");
 }
 
-export async function nextUserId() {
+export function nextUserId() {
-	let nextId = await readJson(nextUserIdPath, 0);
+	let nextId = readJson(nextUserIdPath, 0);
 	if (nextId === 0) {
-		nextId = Math.max(...(await readUsers()).map(user => user.id), -1) + 1;
+		nextId = Math.max(...(readUsers()).map(user => user.id), -1) + 1;
 	}
-	await writeFile(nextUserIdPath, String(nextId + 1), "utf-8");
+	writeFileSync(nextUserIdPath, String(nextId + 1), "utf-8");
 	return nextId;
 }
 
-export async function readUsers() {
+export function readUsers() {
-	return await readJson(usersPath, (): ServerUser[] => []);
+	return readJson(usersPath, (): ServerUser[] => []);
 }
 
-export async function writeUsers(users: ServerUser[]) {
+export function writeUsers(users: ServerUser[]) {
-	await writeFile(usersPath, JSON.stringify(users, undefined, "\t") + "\n", "utf-8");
+	writeFileSync(usersPath, JSON.stringify(users, undefined, "\t") + "\n", "utf-8");
 }
 
-export async function readNextSessionId() {
+export function readNextSessionId() {
-	return await readJson(nextSessionIdPath, 0);
+	return readJson(nextSessionIdPath, 0);
 }
 
-export async function writeNextSessionId(nextId: number) {
+export function writeNextSessionId(nextId: number) {
-	await writeFile(nextSessionIdPath, String(nextId), "utf-8");
+	writeFileSync(nextSessionIdPath, String(nextId), "utf-8");
 }
 
-export async function nextSessionId() {
+export function nextSessionId() {
-	const nextId = await readJson(nextSessionIdPath, 0);
+	const nextId = readJson(nextSessionIdPath, 0);
-	await writeFile(nextSessionIdPath, String(nextId + 1), "utf-8");
+	writeFileSync(nextSessionIdPath, String(nextId + 1), "utf-8");
 	return nextId;
 }
 
-export async function readSessions() {
+export function readSessions() {
 	return readJson<ServerSession[]>(sessionsPath, [])
 }
 
-export async function writeSessions(sessions: ServerSession[]) {
+export function writeSessions(sessions: ServerSession[]) {
-	await writeFile(sessionsPath, JSON.stringify(sessions, undefined, "\t") + "\n", "utf-8");
+	writeFileSync(sessionsPath, JSON.stringify(sessions, undefined, "\t") + "\n", "utf-8");
 }
 
-export async function nextAuthenticationMethodId() {
+export function nextAuthenticationMethodId() {
-	const nextId = await readJson(nextAuthenticationMethodIdPath, 0);
+	const nextId = readJson(nextAuthenticationMethodIdPath, 0);
-	await writeFile(nextAuthenticationMethodIdPath, String(nextId + 1), "utf-8");
+	writeFileSync(nextAuthenticationMethodIdPath, String(nextId + 1), "utf-8");
 	return nextId;
 }
 
-export async function writeNextAuthenticationMethodId(nextId: number) {
+export function writeNextAuthenticationMethodId(nextId: number) {
-	await writeFile(nextAuthenticationMethodIdPath, String(nextId), "utf-8");
+	writeFileSync(nextAuthenticationMethodIdPath, String(nextId), "utf-8");
 }
 
-export async function readAuthenticationMethods() {
+export function readAuthenticationMethods() {
 	return readJson<ServerAuthenticationMethod[]>(authMethodPath, [])
 }
 
-export async function writeAuthenticationMethods(authMethods: ServerAuthenticationMethod[]) {
+export function writeAuthenticationMethods(authMethods: ServerAuthenticationMethod[]) {
-	await writeFile(authMethodPath, JSON.stringify(authMethods, undefined, "\t") + "\n", "utf-8");
+	writeFileSync(authMethodPath, JSON.stringify(authMethods, undefined, "\t") + "\n", "utf-8");
+}
+
+export function nextEventId() {
+	const nextId = readJson(nextEventIdPath, 0);
+	writeFileSync(nextEventIdPath, String(nextId + 1), "utf-8");
+	return nextId;
+}
+
+export function writeNextEventId(nextId: number) {
+	writeFileSync(nextEventIdPath, String(nextId), "utf-8");
+}
+
+export function readEvents() {
+	return readJson<ApiEvent[]>(eventsPath, [])
+}
+
+export function writeEvents(events: ApiEvent[]) {
+	writeFileSync(eventsPath, JSON.stringify(events, undefined, "\t") + "\n", "utf-8");
 }

server/streams.ts

@@ -2,82 +2,134 @@
 	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
 	SPDX-License-Identifier: AGPL-3.0-or-later
 */
-import { readUsers, type ServerSession } from "~/server/database";
+import { readEvents, writeEvents, readUsers, type ServerSession } from "~/server/database";
-import type { ApiAccount, ApiEvent } from "~/shared/types/api";
+import type { ApiAccount, ApiDisconnected, ApiEvent, ApiEventStreamMessage, ApiUserType } from "~/shared/types/api";
 import { serverSessionToApi } from "./utils/session";
 import { H3Event } from "h3";
 
-function sendMessage(
+const keepaliveTimeoutMs = 45e3;
-	stream: WritableStream<string>,
+const eventUpdateTimeMs = 1e3;
-	message: string,
+
-) {
+class EventStream {
-	const writer = stream.getWriter();
+	write!: (data: string) => void;
-	writer.ready
+	close!: (reason?: string) => void;
-		.then(() => writer.write(message))
+
-		.catch(console.error)
+	constructor(
-		.finally(() => writer.releaseLock())
+		public sessionId: number | undefined,
-	;
+		public accountId: number | undefined,
+		public userType: ApiUserType | undefined,
+		public rotatesAtMs: number ,
+		public lastKeepAliveMs: number,
+		public lastEventId: number,
+	) {
+	}
 }
 
-function sendMessageAndClose(
+export async function createEventStream(
-	stream: WritableStream<string>,
-	message: string,
-) {
-	const writer = stream.getWriter();
-	writer.ready
-		.then(() => {
-			writer.write(message);
-			writer.close();
-		}).catch(console.error)
-		.finally(() => writer.releaseLock())
-	;
-}
-
-const streams = new Map<WritableStream<string>, { sessionId?: number, accountId?: number, rotatesAtMs: number }>();
-
-let keepaliveInterval: ReturnType<typeof setInterval> | null = null
-export async function addStream(
 	event: H3Event,
-	stream: WritableStream<string>,
+	source: string,
+	lastEventId: number,
 	session?: ServerSession,
 ) {
-	if (streams.size === 0) {
-		console.log("Starting keepalive")
-		keepaliveInterval = setInterval(sendKeepalive, 4000)
-	}
 	const runtimeConfig = useRuntimeConfig(event);
-	streams.set(stream, {
+	const now = Date.now();
-		sessionId: session?.id,
+	const events = (readEvents()).filter(e => e.id > lastEventId);
-		accountId: session?.accountId,
+	const users = readUsers();
-		rotatesAtMs: session?.rotatesAtMs ?? Date.now() + runtimeConfig.sessionRotatesTimeout * 1000,
+	const apiSession = session ? await serverSessionToApi(event, session) : undefined;
+	let userType: ApiAccount["type"] | undefined;
+	if (session?.accountId !== undefined) {
+		userType = users.find(a => !a.deleted && a.id === session.accountId)?.type
+	}
+	const stream = new EventStream(
+		session?.id,
+		session?.accountId,
+		userType,
+		session?.rotatesAtMs ?? now + runtimeConfig.sessionRotatesTimeout * 1000,
+		now,
+		events[events.length - 1]?.id ?? lastEventId,
+	);
+
+	const readableStream = new ReadableStream<Uint8Array>({
+		start(controller) {
+			const encoder = new TextEncoder();
+			stream.write = (data: string) => {
+				controller.enqueue(encoder.encode(data));
+			}
+			stream.close = (reason?: string) => {
+				const data: ApiDisconnected = {
+					type: "disconnect",
+					reason,
+				};
+				stream.write(`data: ${JSON.stringify(data)}\n\n`);
+				controller.close();
+				deleteStream(stream);
+			},
+			console.log(`Starting event stream for ${source}`)
+			addStream(stream);
+		},
+		cancel(reason) {
+			console.log(`Cancelled event stream for ${source}:`, reason);
+			deleteStream(stream);
+		}
 	});
 	// Produce a response immediately to avoid the reply waiting for content.
-	const update: ApiEvent = {
+	const update: ApiEventStreamMessage = {
 		type: "connected",
-		session: session ? await serverSessionToApi(event, session) : undefined,
+		session: apiSession,
 	};
-	sendMessage(stream, `event: update\ndata: ${JSON.stringify(update)}\n\n`);
+	stream.write(`data: ${JSON.stringify(update)}\n\n`);
+
+	/*
+		Send events since the provided lastEventId
+
+		Warning: This have to happen either before addStream(stream) is
+		called, or as done here synchronously after it. Otherwise there's a
+		possibility of events being delivered out of order, which will break
+		the assumption made by the schedule updating logic.
+	*/
+	if (events.length)
+		console.log(`Sending ${events.length} event(s) to ${source}`);
+	for (const event of events) {
+		if (!sendEventToStream(stream, event)) {
+			break;
+		}
+	}
+
+	return readableStream;
 }
-export function deleteStream(stream: WritableStream<string>) {
+
+let updateInterval: ReturnType<typeof setInterval> | null = null
+const streams = new Set<EventStream>();
+function addStream(
+	stream: EventStream,
+) {
+	if (streams.size === 0) {
+		console.log("Starting event updates")
+		updateInterval = setInterval(sendEventUpdates, eventUpdateTimeMs)
+	}
+	streams.add(stream);
+}
+function deleteStream(stream: EventStream) {
 	streams.delete(stream);
 	if (streams.size === 0) {
-		console.log("Ending keepalive")
+		console.log("Ending event updates")
-		clearInterval(keepaliveInterval!);
+		clearInterval(updateInterval!);
+		updateInterval = null;
 	}
 }
 
 export function cancelAccountStreams(accountId: number) {
-	for (const [stream, data] of streams) {
+	for (const stream of streams.values()) {
-		if (data.accountId === accountId) {
+		if (stream.accountId === accountId) {
-			sendMessageAndClose(stream, `data: cancelled\n\n`);
+			stream.close("cancelled");
 		}
 	}
 }
 
 export function cancelSessionStreams(sessionId: number) {
-	for (const [stream, data] of streams) {
+	for (const stream of streams.values()) {
-		if (data.sessionId === sessionId) {
+		if (stream.sessionId === sessionId) {
-			sendMessageAndClose(stream, `data: cancelled\n\n`);
+			stream.close("cancelled");
 		}
 	}
 }
@@ -94,6 +146,7 @@ function encodeEvent(event: ApiEvent, userType: ApiAccount["type"] | undefined)
 	if (event.type === "schedule-update") {
 		if (!canSeeCrew(userType)) {
 			event = {
+				id: event.id,
 				type: event.type,
 				updatedFrom: event.updatedFrom,
 				data: filterSchedule(event.data),
@@ -106,6 +159,7 @@ function encodeEvent(event: ApiEvent, userType: ApiAccount["type"] | undefined)
 			|| !event.data.deleted && event.data.type === "anonymous" && !canSeeAnonymous(userType)
 		) {
 			event = {
+				id: event.id,
 				type: event.type,
 				data: {
 					id: event.data.id,
@@ -128,44 +182,68 @@ function encodeEvent(event: ApiEvent, userType: ApiAccount["type"] | undefined)
 }
 
 export async function broadcastEvent(event: ApiEvent) {
-	const id = Date.now();
+	const events = readEvents();
-	console.log(`broadcasting update to ${streams.size} clients`);
+	events.push(event);
-	if (!streams.size) {
+	writeEvents(events);
-		return;
+}
-	}
 
+function sendEventToStream(stream: EventStream, event: ApiEvent) {
 	// Session expiry events cause the streams belonging to that session to be terminated
 	if (event.type === "session-expired") {
-		cancelSessionStreams(event.sessionId);
+		if (stream.sessionId === event.sessionId) {
-		return;
+			stream.close("session expired");
+			return false;
+		}
+		return true;
 	}
 
-	const users = await readUsers();
-	for (const [stream, streamData] of streams) {
 	// Account events are specially handled and only sent to the user they belong to.
 	if (event.type === "account-update") {
-			if (streamData.accountId === event.data.id) {
+		if (stream.accountId === event.data.id) {
-				sendMessage(stream, `id: ${id}\nevent: update\ndata: ${JSON.stringify(event)}\n\n`);
+			stream.write(`id: ${event.id}\nevent: event\ndata: ${JSON.stringify(event)}\n\n`);
+		}
+		return true;
 	}
 
-		} else {
+	// All other events are encoded according to the user access level seeing it.
-			let userType: ApiAccount["type"] | undefined;
+	const data = encodeEvent(event, stream.userType)
-			if (streamData.accountId !== undefined) {
+	stream.write(`id: ${event.id}\nevent: event\ndata: ${data}\n\n`);
-				userType = users.find(a => !a.deleted && a.id === streamData.accountId)?.type
+	return true;
-			}
-			const data = encodeEvent(event, userType)
-			sendMessage(stream, `id: ${id}\nevent: update\ndata: ${data}\n\n`);
-		}
-	}
 }
 
-function sendKeepalive() {
+async function sendEventUpdates() {
+	// Cancel streams that need to be rotated.
 	const now = Date.now();
-	for (const [stream, streamData] of streams) {
+	for (const stream of streams.values()) {
-		if (streamData.rotatesAtMs > now) {
+		if (stream.rotatesAtMs < now) {
-			sendMessage(stream, ": keepalive\n");
+			stream.close("session rotation");
-		} else {
+			continue;
-			sendMessageAndClose(stream, `data: cancelled\n\n`);
+		}
+	}
+
+	// Send events.
+	const skipEventId = Math.min(...[...streams.values()].map(s => s.lastEventId));
+	const events = (readEvents()).filter(e => e.id > skipEventId);
+	if (events.length)
+		console.log(`broadcasting ${events.length} event(s) to ${streams.size} client(s)`);
+	for (const stream of streams.values()) {
+		for (const event of events) {
+			if (event.id > stream.lastEventId) {
+				stream.lastEventId = event.id;
+				stream.lastKeepAliveMs = now;
+
+				if (!sendEventToStream(stream, event)) {
+					break;
+				}
+			}
+		}
+	}
+
+	// Send Keepalives to streams with no activity.
+	for (const stream of streams.values()) {
+		if (stream.lastKeepAliveMs + keepaliveTimeoutMs < now) {
+			stream.write(": keepalive\n");
+			stream.lastKeepAliveMs = now;
 		}
 	}
 }

server/utils/schedule.ts

@@ -2,7 +2,7 @@
 	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
 	SPDX-License-Identifier: AGPL-3.0-or-later
 */
-import { readSchedule, type ServerUser, writeSchedule } from '~/server/database';
+import { nextEventId, readSchedule, type ServerUser, writeSchedule } from '~/server/database';
 import { broadcastEvent } from '~/server/streams';
 import type { ApiSchedule, ApiTombstone } from '~/shared/types/api';
 
@@ -20,7 +20,7 @@ export async function updateScheduleInterestedCounts(users: ServerUser[]) {
 				eventSlotCounts.set(id, (eventSlotCounts.get(id) ?? 0) + 1);
 	}
 
-	const schedule = await readSchedule();
+	const schedule = readSchedule();
 	if (schedule.deleted) {
 		throw new Error("Deleted schedule not implemented");
 	}
@@ -58,6 +58,7 @@ export async function updateScheduleInterestedCounts(users: ServerUser[]) {
 	schedule.updatedAt = updatedFrom;
 	await writeSchedule(schedule);
 	await broadcastEvent({
+		id: nextEventId(),
 		type: "schedule-update",
 		updatedFrom,
 		data: update,

server/utils/session.ts

@@ -4,6 +4,7 @@
 */
 import type { H3Event } from "h3";
 import {
+	nextEventId,
 	nextSessionId,
 	readSessions,
 	readSubscriptions,
@@ -18,11 +19,11 @@ import type { ApiAuthenticationProvider, ApiSession } from "~/shared/types/api";
 import { serverUserToApiAccount } from "./user";
 
 async function removeSessionSubscription(sessionId: number) {
-	const subscriptions = await readSubscriptions();
+	const subscriptions = readSubscriptions();
 	const index = subscriptions.findIndex(subscription => subscription.sessionId === sessionId);
 	if (index !== -1) {
 		subscriptions.splice(index, 1);
-		await writeSubscriptions(subscriptions);
+		writeSubscriptions(subscriptions);
 	}
 }
 
@@ -34,6 +35,7 @@ async function clearServerSessionInternal(event: H3Event, sessions: ServerSessio
 		if (session) {
 			session.expiresAtMs = Date.now();
 			broadcastEvent({
+				id: nextEventId(),
 				type: "session-expired",
 				sessionId,
 			});
@@ -45,9 +47,9 @@ async function clearServerSessionInternal(event: H3Event, sessions: ServerSessio
 }
 
 export async function clearServerSession(event: H3Event) {
-	const sessions = await readSessions();
+	const sessions = readSessions();
 	if (await clearServerSessionInternal(event, sessions)) {
-		await writeSessions(sessions);
+		writeSessions(sessions);
 	}
 	deleteCookie(event, "session");
 }
@@ -59,7 +61,7 @@ export async function setServerSession(
 	authenticationSlug?: string,
 	authenticationName?: string,
 ) {
-	const sessions = await readSessions();
+	const sessions = readSessions();
 	const runtimeConfig = useRuntimeConfig(event);
 	await clearServerSessionInternal(event, sessions);
 
@@ -76,14 +78,14 @@ export async function setServerSession(
 	};
 
 	sessions.push(newSession);
-	await writeSessions(sessions);
+	writeSessions(sessions);
 	await setSignedCookie(event, "session", String(newSession.id), runtimeConfig.sessionDiscardTimeout)
 	return newSession;
 }
 
 async function rotateSession(event: H3Event, sessions: ServerSession[], session: ServerSession) {
 	const runtimeConfig = useRuntimeConfig(event);
-	const users = await readUsers();
+	const users = readUsers();
 	const account = users.find(user => !user.deleted && user.id === session.accountId);
 	const now = Date.now();
 	const newSession: ServerSession = {
@@ -92,12 +94,12 @@ async function rotateSession(event: H3Event, sessions: ServerSession[], session:
 		// Authentication provider is removed to avoid possibility of an infinite delay before using it.
 		rotatesAtMs: now + runtimeConfig.sessionRotatesTimeout * 1000,
 		discardAtMs: now + runtimeConfig.sessionDiscardTimeout * 1000,
-		id: await nextSessionId(),
+		id: nextSessionId(),
 	};
 	session.successor = newSession.id;
 	session.expiresAtMs = Date.now() + 10 * 1000;
 	sessions.push(newSession);
-	await writeSessions(sessions);
+	writeSessions(sessions);
 	await setSignedCookie(event, "session", String(newSession.id), runtimeConfig.sessionDiscardTimeout)
 	return newSession;
 }
@@ -106,7 +108,7 @@ export async function getServerSession(event: H3Event, ignoreTaken: boolean) {
 	const sessionCookie = await getSignedCookie(event, "session");
 	if (sessionCookie) {
 		const sessionId = parseInt(sessionCookie, 10);
-		const sessions = await readSessions();
+		const sessions = readSessions();
 		const session = sessions.find(session => session.id === sessionId);
 		if (session) {
 			const nowMs = Date.now();
@@ -146,7 +148,7 @@ export async function requireServerSession(event: H3Event, message: string) {
 export async function requireServerSessionWithUser(event: H3Event) {
 	const message = "User session required";
 	const session = await requireServerSession(event, message);
-	const users = await readUsers();
+	const users = readUsers();
 	const account = users.find(user => user.id === session.accountId);
 	if (session.accountId === undefined || !account || account.deleted)
 		throw createError({
@@ -161,7 +163,7 @@ export async function requireServerSessionWithUser(event: H3Event) {
 export async function requireServerSessionWithAdmin(event: H3Event) {
 	const message = "Admin session required";
 	const session = await requireServerSession(event, message);
-	const users = await readUsers();
+	const users = readUsers();
 	const account = users.find(user => user.id === session.accountId);
 	if (session.access !== "admin" || account?.type !== "admin") {
 		throw createError({
@@ -174,9 +176,9 @@ export async function requireServerSessionWithAdmin(event: H3Event) {
 }
 
 export async function serverSessionToApi(event: H3Event, session: ServerSession): Promise<ApiSession> {
-	const users = await readUsers();
+	const users = readUsers();
 	const account = users.find(user => !user.deleted && user.id === session.accountId);
-	const subscriptions = await readSubscriptions();
+	const subscriptions = readSubscriptions();
 	const push = Boolean(
 		subscriptions.find(sub => sub.type === "push" && sub.sessionId === session.id)
 	);

server/web-push.ts

@@ -33,7 +33,7 @@ async function useVapidDetails(event: H3Event) {
 export async function sendPush(event: H3Event, title: string, body: string) {
 	const vapidDetails = await useVapidDetails(event);
 	const payload = JSON.stringify({ title, body });
-	const subscriptions = await readSubscriptions();
+	const subscriptions = readSubscriptions();
 	console.log(`Sending "${payload}" to ${subscriptions.length} subscribers`);
 	const removeIndexes = [];
 	for (let index = 0; index < subscriptions.length; index += 1) {
@@ -65,7 +65,7 @@ export async function sendPush(event: H3Event, title: string, body: string) {
 		for (const index of removeIndexes) {
 			subscriptions.splice(index, 1);
 		}
-		await writeSubscriptions(subscriptions);
+		writeSubscriptions(subscriptions);
 	}
 	console.log("Push notices sent");
 }

shared/types/api.ts

@@ -155,27 +155,26 @@ export interface ApiUserDetails {
 }
 
 export interface ApiAccountUpdate {
+	id: Id,
 	type: "account-update",
 	data: ApiAccount,
 }
 
-export interface ApiConnected {
-	type: "connected",
-	session?: ApiSession,
-}
-
 export interface ApiScheduleUpdate {
+	id: Id,
 	type: "schedule-update",
 	updatedFrom?: string,
 	data: ApiSchedule | ApiTombstone,
 }
 
 export interface ApiSessionExpired {
+	id: Id,
 	type: "session-expired",
 	sessionId: Id,
 }
 
 export interface ApiUserUpdate {
+	id: Id,
 	type: "user-update",
 	updatedFrom?: string,
 	data: ApiUser | ApiTombstone,
@@ -183,8 +182,22 @@ export interface ApiUserUpdate {
 
 export type ApiEvent =
 	| ApiAccountUpdate
-	| ApiConnected
 	| ApiScheduleUpdate
 	| ApiSessionExpired
 	| ApiUserUpdate
 ;
+
+export interface ApiConnected {
+	type: "connected",
+	session?: ApiSession,
+}
+
+export interface ApiDisconnected {
+	type: "disconnect",
+	reason?: string,
+}
+
+export type ApiEventStreamMessage =
+	| ApiConnected
+	| ApiDisconnected
+;

stores/events.ts

@@ -0,0 +1,31 @@
+/*
+	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
+	SPDX-License-Identifier: AGPL-3.0-or-later
+*/
+
+export const useEventsStore = defineStore("events", () => {
+	const state = {
+		lastEventId: ref(0),
+	};
+	const getters = {
+	}
+	const actions = {
+		async fetchLastEventId() {
+			const requestFetch = useRequestFetch();
+			state.lastEventId.value = await requestFetch("/api/last-event-id");
+		}
+	}
+
+	appEventSource?.addEventListener("event", (event) => {
+		if (event.data.id !== undefined) {
+			state.lastEventId.value = event.data.id
+			return;
+		}
+	});
+
+	return {
+		...state,
+		...getters,
+		...actions,
+	};
+});

stores/schedules.ts

@@ -96,7 +96,7 @@ export const useSchedulesStore = defineStore("schedules", () => {
 		}
 	})
 
-	appEventSource?.addEventListener("update", (event) => {
+	appEventSource?.addEventListener("event", (event) => {
 		if (event.data.type !== "schedule-update") {
 			return;
 		}

stores/session.ts

@@ -57,7 +57,7 @@ export const useSessionStore = defineStore("session", () => {
 		},
 	};
 
-	appEventSource?.addEventListener("update", (event) => {
+	appEventSource?.addEventListener("message", (event) => {
 		if (event.data.type !== "connected") {
 			return;
 		}

stores/users.ts

@@ -75,7 +75,7 @@ export const useUsersStore = defineStore("users", () => {
 		},
 	}
 
-	appEventSource?.addEventListener("update", (event) => {
+	appEventSource?.addEventListener("event", (event) => {
 		if (event.data.type !== "user-update") {
 			return;
 		}