🦋 POS pool synching between devices

SSE-based real-time synchronization for POS pools and session cart:
- New SSE endpoint for orderTabs change stream (watches all pools for sidebar updates)
- Shared pool-sse.ts utility with debounced invalidate (1s) to reduce redundant requests
- Tab page: reactive SSE reconnection on pool switch via switchPoolSSE()
- Split page: SSE connection for real-time payment/item updates
- Session SSE: simplified cart events to trigger-only (client reloads via load()), added initial cart trigger on connect for reconnect consistency
- Removed sending full cart data over SSE (was using formatCart), now uses invalidate() pattern matching the pool SSE approach

Fixes issue #2364

Author: gntw

src/lib/utils/pool-sse.ts

@@ -0,0 +1,22 @@
+import { fetchEventSource } from '@microsoft/fetch-event-source';
+import { invalidate } from '$app/navigation';
+import { UrlDependency } from '$lib/types/UrlDependency';
+import { debounce } from '$lib/utils/debounce';
+
+const SSE_DEBOUNCE_MS = 1000;
+
+export function connectPoolSSE(slug: string, signal: AbortSignal): void {
+	const debouncedInvalidate = debounce(
+		() => invalidate(UrlDependency.orderTab(slug)),
+		SSE_DEBOUNCE_MS
+	);
+	fetchEventSource(`/pos/touch/tab/${slug}/sse`, {
+		signal,
+		onmessage() {
+			debouncedInvalidate();
+		},
+		onerror(err) {
+			console.error(`Pool SSE error for tab ${slug}:`, err);
+		}
+	});
+}

src/routes/(app)/pos/session/+page.server.ts

@@ -3,11 +3,13 @@ import { redirect } from '@sveltejs/kit';
 import { formatCart, formatOrder } from './formatCartOrder.js';
 import { runtimeConfig } from '$lib/server/runtime-config.js';
 
-export const load = async ({ locals }) => {
+export const load = async ({ locals, depends }) => {
 	if (!locals.user) {
 		throw redirect(303, '/admin/login');
 	}
 
+	depends('data:pos-session-cart');
+
 	const cart = await collections.carts.findOne(
 		{ 'user.userId': locals.user._id },
 		{ sort: { createdAt: -1 } }

src/routes/(app)/pos/session/+page.svelte

@@ -9,6 +9,7 @@
 	import { orderRemainingToPay } from '$lib/types/Order.js';
 	import Trans from '$lib/components/Trans.svelte';
 	import { computePriceInfo } from '$lib/cart.js';
+	import { invalidate } from '$app/navigation';
 
 	interface CustomEventSource {
 		onerror?: ((this: CustomEventSource, ev: Event) => unknown) | null;
@@ -22,7 +23,7 @@
 		: 5_000;
 
 	let eventSourceInstance: CustomEventSource | void | null = null;
-	let formattedCart = data.formattedCart;
+	$: formattedCart = data.formattedCart;
 	let order = data.order;
 
 	$: view =
@@ -39,9 +40,9 @@
 			onmessage(ev) {
 				if (ev.data) {
 					try {
-						const { eventType, cart: sseCart, order: sseOrder } = JSON.parse(ev.data);
+						const { eventType, order: sseOrder } = JSON.parse(ev.data);
 						if (eventType === 'cart') {
-							formattedCart = sseCart;
+							invalidate('data:pos-session-cart');
 						} else if (eventType === 'order') {
 							order = sseOrder;
 							clearTimeout(currentTimeout);

src/routes/(app)/pos/session/sse/+server.ts

@@ -3,7 +3,7 @@ import type { Cart } from '$lib/types/Cart.js';
 import type { Order } from '$lib/types/Order.js';
 import { error } from '@sveltejs/kit';
 import type { ChangeStream, ChangeStreamDocument, ObjectId } from 'mongodb';
-import { formatCart, formatOrder } from '../formatCartOrder.js';
+import { formatOrder } from '../formatCartOrder.js';
 
 export async function GET({ locals }) {
 	const userId = locals.user?._id;
@@ -68,13 +68,9 @@ export async function GET({ locals }) {
 			ownCartId = ownCart._id;
 		}
 		try {
-			const formattedCart = await formatCart(ownCart || null, locals);
 			await writer?.ready;
-			await writer?.write(
-				`data: ${JSON.stringify({ eventType: 'cart', cart: formattedCart })}\n\n`
-			);
-		} catch (error) {
-			// Error writing to the client, assume it has disconnected
+			await writer?.write(`data: ${JSON.stringify({ eventType: 'cart' })}\n\n`);
+		} catch {
 			cleanup();
 		}
 	});
@@ -125,17 +121,16 @@ export async function GET({ locals }) {
 
 	writer?.ready
 		.then(async () => {
+			// Cart: just get ID for delete tracking. Client already has data from load()
 			const cart = await collections.carts.findOne(
 				{ 'user.userId': userId },
-				{ sort: { createdAt: -1 } }
+				{ sort: { createdAt: -1 }, projection: { _id: 1 } }
 			);
-			if (cart) {
-				ownCartId = cart._id;
-			}
-			const formattedCart = await formatCart(cart, locals);
+			ownCartId = cart?._id ?? null;
 
-			writer?.write(`data: ${JSON.stringify({ eventType: 'cart', cart: formattedCart })}\n\n`);
+			await writer?.write(`data: ${JSON.stringify({ eventType: 'cart' })}\n\n`);
 
+			// Order: keep initial push (client displays order directly from SSE data)
 			const order = await collections.orders.findOne(
 				{ 'user.userId': userId },
 				{ sort: { createdAt: -1 } }

src/routes/(app)/pos/touch/tab/[orderTabSlug]/+page.svelte

@@ -23,9 +23,29 @@
 	import type { PrintHistoryEntry } from '$lib/types/PrintHistoryEntry';
 	import MoveItemsModal from '$lib/components/MoveItemsModal.svelte';
 	import { z } from 'zod';
+	import { connectPoolSSE } from '$lib/utils/pool-sse';
+	import { browser } from '$app/environment';
 
 	export let data;
 	$: tabSlug = data.tabSlug;
+
+	let sseAbort: AbortController | null = null;
+	let sseCurrentSlug: string | null = null;
+
+	function switchPoolSSE(slug: string) {
+		if (sseCurrentSlug === slug) {
+			return;
+		}
+		sseAbort?.abort();
+		sseCurrentSlug = slug;
+		sseAbort = new AbortController();
+		connectPoolSSE(slug, sseAbort.signal);
+	}
+
+	$: if (browser) {
+		switchPoolSSE(tabSlug);
+	}
+
 	$: next = Number($page.url.searchParams.get('skip')) || 0;
 	$: picturesByProduct = groupBy(
 		data.pictures.filter(
@@ -160,6 +180,8 @@
 		window.addEventListener('resize', updatePaginationLimit);
 
 		return () => {
+			sseAbort?.abort();
+			sseCurrentSlug = null;
 			window.removeEventListener('resize', updatePaginationLimit);
 			window.removeEventListener('resize', checkMobileView);
 		};

src/routes/(app)/pos/touch/tab/[orderTabSlug]/split/+page.svelte

@@ -16,6 +16,7 @@
 	import PosPaymentMethodSelector, {
 		type PaymentOption
 	} from '$lib/components/PosPaymentMethodSelector.svelte';
+	import { connectPoolSSE } from '$lib/utils/pool-sse';
 
 	const { t } = useI18n();
 
@@ -170,10 +171,19 @@
 		}
 	}
 
+	let sseAbort: AbortController | null = null;
+
 	onMount(() => {
 		checkMobileView();
 		window.addEventListener('resize', checkMobileView);
-		return () => window.removeEventListener('resize', checkMobileView);
+
+		sseAbort = new AbortController();
+		connectPoolSSE(tabSlug, sseAbort.signal);
+
+		return () => {
+			sseAbort?.abort();
+			window.removeEventListener('resize', checkMobileView);
+		};
 	});
 
 	let fromCashInAll = false;

src/routes/(app)/pos/touch/tab/[orderTabSlug]/sse/+server.ts

@@ -0,0 +1,48 @@
+import { collections } from '$lib/server/database.js';
+import type { OrderTab } from '$lib/types/OrderTab.js';
+import { error } from '@sveltejs/kit';
+import type { ChangeStream, ChangeStreamDocument } from 'mongodb';
+
+export async function GET({ locals }) {
+	if (!locals.user) {
+		throw error(401, 'Unauthorized');
+	}
+
+	const { readable, writable } = new TransformStream();
+	let writer: WritableStreamDefaultWriter<unknown> | null = writable.getWriter();
+
+	function cleanup() {
+		writer?.close();
+		writer = null;
+		changeStream?.close().catch(console.error);
+		changeStream = null;
+	}
+
+	let changeStream: ChangeStream<OrderTab, ChangeStreamDocument<OrderTab>> | null =
+		collections.orderTabs.watch([]);
+
+	changeStream
+		.on('change', async () => {
+			if (!writer) {
+				return;
+			}
+			try {
+				await writer?.ready;
+				await writer?.write(`data: {}\n\n`);
+			} catch {
+				cleanup();
+			}
+		})
+		.on('error', () => {
+			cleanup();
+		});
+
+	return new Response(readable, {
+		headers: {
+			'Content-Type': 'text/event-stream',
+			'Cache-Control': 'no-cache',
+			Connection: 'keep-alive',
+			'X-Accel-Buffering': 'no'
+		}
+	});
+}