feat: Consolidate SSE for logs into events to reduce connection pool usage for HTTP/1.1

src/backend/common/infrastructure/MSBackendEventMap.ts

@@ -1,5 +1,5 @@
 import type { Dayjs } from "dayjs";
-import type { EmittedMSEvent, PlayObject, SourcePlayerObj, SourceType } from "../../../core/Atomic.ts";
+import type { EmittedMSEvent, LogEvent, PlayObject, SourcePlayerObj, SourceType } from "../../../core/Atomic.ts";
 import type { MSCoreEvents } from "../../../core/MSCoreEventMap.ts";
 import type { WebhookPayload } from "./config/health/webhooks.ts";
 
@@ -19,4 +19,5 @@ export interface MSBackendEventMap extends Omit<MSCoreEvents, 'playerUpdate'> {
     playerUpdate: [EmittedMSEvent<SourcePlayerObj<Dayjs> & { options: { scrobbleTo: string[], wasMonitored?: boolean } },{},SourceType>]
     configError: [Error | string]
     instantiateError: [Error]
+    log: [LogEvent]
 }

src/backend/server/api.ts

@@ -1,11 +1,11 @@
 import {loggerTest, type LogDataPretty, type Logger, type LogLevel} from "@foxxmd/logging";
 import type {Express, Request, Response} from 'express';
-import bsseDef from 'better-sse';
+import bsseDef, { type Session } from 'better-sse';
 import bodyParser from "body-parser";
 import { FixedSizeList } from 'fixed-size-list';
 import { PassThrough } from "node:stream";
-import { Transform } from "stream";
 import {
+    type LogEvent,
     type LogOutputConfig,
     queueContextSchema,
     logLevelStandaloneSchema,
@@ -114,25 +114,12 @@ export const setupApi = (args: ApiArgs, opts: ApiOptions = {}) => {
         limit: 50,
     }
 
-    let logObjectStream: Transform;
-    try {
-        logObjectStream = new Transform({
-            transform: (chunk, e, cb) => {
-                cb(null, chunk)
-            },
-            objectMode: true,
-            allowHalfOpen: true
-        })
-    } catch (e) {
-        console.log(e);
-    }
+    appLoggerStream.on('data', addToLogBuffer);
+    // one data listener per open /api/events connection
+    appLoggerStream.setMaxListeners(0);
 
-    appLoggerStream.on('data', (log: LogDataPretty) => {
-        addToLogBuffer(log);
-        if(log.level >= logger.levels.values[logConfig.level]) {
-            logObjectStream.write({message: log.line, level: log.level, levelLabel: logger.levels.labels[log.level]});
-        }
-    });
+    // keyed by client-generated per-tab id so logs are only pushed to tabs actually rendering them
+    const eventSessions = new Map<string, {session: Session, logs: boolean}>();
 
     const componentAwareMiddle = makeComponentMiddle(scrobbleSources, scrobbleClients);
 
@@ -150,11 +137,6 @@ export const setupApi = (args: ApiArgs, opts: ApiOptions = {}) => {
         next();
     }
 
-    router.get('/api/logs/stream', {middleware: [setLogWebSettings], tags: ['Events'], summary: 'SSE Logs'}, async (req, res) => {
-        const session = await bsseDef.createSession(req, res);
-        await session.stream(logObjectStream);
-    });
-
     router.get('/api/logs', {middleware: [setLogWebSettings], tags: ['Events'], summary: 'Get Logs'}, async (req, res) => {
         const slicedLog = getLogs(logger.levels.values[logConfig.level], logConfig.limit + 1, logConfig.sort === 'ascending' ? 'asc' : 'desc');
         return res.json({data: slicedLog, settings: logConfig});
@@ -180,18 +162,29 @@ export const setupApi = (args: ApiArgs, opts: ApiOptions = {}) => {
     });
 
     router.get('/api/events', {querySchema: z.object({
-        next: z.literal('true').optional().meta({description: 'When used events are sent in new ui format'})
-    }), tags: ['Events'], summary: 'SSE Events'}, async (req, res) => {
+        next: z.literal('true').optional().meta({description: 'When used events are sent in new ui format'}),
+        tab: z.string().optional().meta({description: 'Client-generated connection id. Required to opt in to log events via PUT /api/events/:tab/logs'})
+    }), middleware: [setLogWebSettings], tags: ['Events'], summary: 'SSE Events'}, async (req, res) => {
         const {
             query: {
-                next: nextQs
+                next: nextQs,
+                tab
             }
         } = req;
 
         const isNextapi = nextQs === 'true';
 
         const session = await bsseDef.createSession(req, res);
-        scrobbleSources.emitter.onAny((eventName: string, payload: any) => {
+        if(tab !== undefined) {
+            eventSessions.set(tab, {session, logs: false});
+        }
+        const onLog = (log: LogDataPretty) => {
+            if(tab !== undefined && eventSessions.get(tab)?.logs && log.level >= logger.levels.values[logConfig.level]) {
+                session.push({message: log.line, level: log.level, levelLabel: logger.levels.labels[log.level]} satisfies LogEvent, 'log');
+            }
+        };
+        appLoggerStream.on('data', onLog);
+        const unsubSources = scrobbleSources.emitter.onAny((eventName: string, payload: any) => {
             if(payload !== undefined && payload.from !== undefined) {
                 if(isNextapi) {
                     session.push({event: eventName, ...payload}, eventName);
@@ -200,7 +193,7 @@ export const setupApi = (args: ApiArgs, opts: ApiOptions = {}) => {
                 }
             }
         });
-        scrobbleClients.emitter.onAny((eventName: string, payload: any) => {
+        const unsubClients = scrobbleClients.emitter.onAny((eventName: string, payload: any) => {
             if(payload !== undefined && payload.from !== undefined) {
                 if(isNextapi) {
                     session.push({event: eventName, ...payload}, eventName);
@@ -209,6 +202,32 @@ export const setupApi = (args: ApiArgs, opts: ApiOptions = {}) => {
                 }
             }
         });
+        session.once('disconnected', () => {
+            appLoggerStream.off('data', onLog);
+            unsubSources();
+            unsubClients();
+            // a stale connection may close after the same tab has already reconnected
+            if(tab !== undefined && eventSessions.get(tab)?.session === session) {
+                eventSessions.delete(tab);
+            }
+        });
+    });
+
+    router.put('/api/events/:tab/logs', {
+        middleware: [jsonParser],
+        bodySchema: z.object({
+            enabled: z.boolean()
+        }),
+        tags: ['Events'],
+        summary: 'Toggle Log Events',
+        description: 'Enable or disable log events for the /api/events connection opened with this tab id'
+    }, async (req, res) => {
+        const entry = eventSessions.get(req.params.tab);
+        if(entry === undefined) {
+            return res.sendStatus(404);
+        }
+        entry.logs = req.body.enabled;
+        return res.sendStatus(200);
     });
 
     setupDeezerRoutes(app, logger, scrobbleSources);

src/backend/tests/server/events.test.ts

@@ -0,0 +1,81 @@
+import { expect } from 'chai';
+import { describe, it } from 'mocha';
+import { loggerTest } from '@foxxmd/logging';
+import { PassThrough } from 'node:stream';
+import type { AddressInfo } from 'node:net';
+import http from 'node:http';
+import ScrobbleSources from '../../sources/ScrobbleSources.ts';
+import ScrobbleClients from '../../scrobblers/ScrobbleClients.ts';
+import { WildcardEmitter } from '../../common/WildcardEmitter.ts';
+import { initServer } from '../../server/index.ts';
+
+const internalConfig = { localUrl: new URL('https://example.com'), configDir: 'fake', logger: loggerTest, version: 'test' };
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+describe('SSE Events Endpoint', function () {
+
+    it('only pushes log events to a connection while logs are enabled for its tab, and cleans up on disconnect', async function () {
+        const appLoggerStream = new PassThrough({ objectMode: true });
+        const sources = new ScrobbleSources(new WildcardEmitter(), internalConfig, loggerTest);
+        const clients = new ScrobbleClients(new WildcardEmitter(), new WildcardEmitter(), internalConfig, loggerTest);
+        const [app] = await initServer({ sources, clients }, { testMode: true, appLoggerStream });
+        const server = http.createServer(app).listen(0);
+        const base = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
+        const baseListeners = appLoggerStream.listenerCount('data');
+
+        const writeLog = (line: string) => appLoggerStream.write({ level: 30, line, time: Date.now() });
+        const setLogs = (enabled: boolean) => fetch(`${base}/api/events/tab1/logs`, {
+            method: 'PUT',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ enabled })
+        });
+
+        try {
+            await fetch(`${base}/api/logs`, {
+                method: 'PUT',
+                headers: { 'Content-Type': 'application/json' },
+                body: JSON.stringify({ level: 'trace' })
+            });
+
+            expect((await setLogs(true)).status, 'no connection for tab yet').to.eq(404);
+
+            const abort = new AbortController();
+            const res = await fetch(`${base}/api/events?next=true&tab=tab1`, { signal: abort.signal });
+            let received = '';
+            const reading = (async () => {
+                const decoder = new TextDecoder();
+                try {
+                    for await (const chunk of res.body as unknown as AsyncIterable<Uint8Array>) {
+                        received += decoder.decode(chunk);
+                    }
+                } catch {
+                    // aborted
+                }
+            })();
+            await sleep(50);
+            expect(appLoggerStream.listenerCount('data')).to.eq(baseListeners + 1);
+
+            writeLog('before enable');
+            expect((await setLogs(true)).status).to.eq(200);
+            writeLog('while enabled');
+            expect((await setLogs(false)).status).to.eq(200);
+            writeLog('after disable');
+            await sleep(50);
+
+            expect(received).to.include('event:log');
+            expect(received).to.include('while enabled');
+            expect(received).to.not.include('before enable');
+            expect(received).to.not.include('after disable');
+
+            abort.abort();
+            await reading;
+            await sleep(50);
+            expect(appLoggerStream.listenerCount('data'), 'log listener removed on disconnect').to.eq(baseListeners);
+            expect((await setLogs(true)).status, 'tab forgotten on disconnect').to.eq(404);
+        } finally {
+            server.closeAllConnections();
+            server.close();
+        }
+    });
+});

src/client/App.tsx

@@ -7,6 +7,7 @@ import {
     RouterProvider, useLocation
 } from "react-router-dom";
 import type {MsSseEvent} from '../core/Api';
+import { sseTabId } from './sseTab';
 import { AppHeader } from './components/AppHeader';
 import { MSErrorBoundary } from './components/ErrorBoundary';
 import { ComponentDetailedRoutable } from './components/msComponent/MSComponentDetailed';
@@ -104,7 +105,7 @@ const router = createBrowserRouter(routesNested, { basename: window.__MS_RUNTIME
 
 export const sseProviderOptions = {
     key: ['events'],
-    url: 'api/events?next=true'
+    url: `api/events?next=true&tab=${sseTabId}`
 }
 
 function App() {

src/client/components/LogsNext.tsx

@@ -2,13 +2,15 @@ import { Box, HStack, SegmentGroup, Separator, Span, Stack, Text, FloatingPanel,
 import {
     useWindowSize,
 } from '@react-hook/window-size';
-import { useSSE } from "@flamefrontend/sse-runtime-react";
+import { useSSEContext, useSSEEvent, useSSEStatus } from "@flamefrontend/sse-runtime-react";
 import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
 import * as AnsiImport from "ansi-to-react";
 import { FixedSizeList } from "fixed-size-list";
 import ky from 'ky';
 import React, { useCallback, useEffect, useRef, useState, type ComponentProps } from 'react';
 import type {LogLevelStandalone, LogOutputConfig} from '../../core/Atomic';
+import type { MsSseEvent } from '../../core/Api';
+import { sseTabId } from '../sseTab';
 import { tanQueries } from '../queries';
 import { ChakraClipDynamic } from './ChakraClipboard';
 import { TerminalButton, XButton } from './icons/ChakraIcons';
@@ -43,6 +45,14 @@ const createFixedList = (size: number, initialList: MinLogInfo[] = []): FixedSiz
 
 let list = createFixedList(50);
 
+// number of mounted, streaming LogsFetchable in this tab. Logs are only pushed by the server while > 0
+let logViewers = 0;
+// chained so a fast unmount/remount can't have its enable/disable arrive out of order
+let logEventsQueue: Promise<unknown> = Promise.resolve();
+const setLogEvents = (enabled: boolean) => {
+    logEventsQueue = logEventsQueue.then(() => ky.put(`api/events/${sseTabId}/logs`, {json: {enabled}})).catch(() => undefined);
+};
+
 export const Logs = (props: {logs: Readonly<LogLineProps[]>, ref?: React.Ref<HTMLDivElement>}) => {
     return <Box ref={props.ref} fontFamily="source-code-pro, Menlo, Monaco, Consolas,'Courier New',monospace;">
         {props.logs.map((x, index) => <LogLine key={index} message={x.message}/>)}
@@ -69,17 +79,28 @@ export const LogsFetchable = (props: {settings?: LogOutputConfig, streamable?: b
         staleTime: Infinity
     });
 
-    const connection = useSSE<{stream: MinLogInfo}>({
-        key: ['stream'],
-        url: "api/logs/stream",
-        enabled: streamable,
-        events: {
-            stream: (data) => {
-                if(!isPending) {
-                    list.add(data);
-                    setLogList(Array.from(list.data));
-                }
+    const sse = useSSEContext<MsSseEvent>();
+    const { status: sseStatus } = useSSEStatus(sse);
+
+    // re-runs on reconnect since server forgets opt-in when the connection closes
+    useEffect(() => {
+        if(!streamable || sseStatus !== 'open') {
+            return;
+        }
+        logViewers++;
+        setLogEvents(true);
+        return () => {
+            logViewers--;
+            if(logViewers === 0) {
+                setLogEvents(false);
             }
+        };
+    }, [streamable, sseStatus]);
+
+    useSSEEvent(sse, 'log', (data) => {
+        if(streamable && !isPending) {
+            list.add({...data, id: ''});
+            setLogList(Array.from(list.data));
         }
     });
 

src/client/sseTab.ts

@@ -0,0 +1,3 @@
+// per-page-load id sent with the shared /api/events connection so the server can opt this tab in/out of log events
+// not crypto.randomUUID because it is undefined on insecure (plain http LAN) origins
+export const sseTabId = Math.random().toString(36).slice(2);

src/core/Api.ts

@@ -1,6 +1,6 @@
 import type { CompareOpKey, ComponentMinimalSelect } from "../backend/common/database/drizzle/drizzleTypes.ts"
 import { creditSchema, playTrackDataSchema, type ClientType, type ComponentAuthType, type MonitoringStatus, type QueueContext, type MusicServices, type Replace, creditBaseSchema, trackSearchHash, albumSearchHash, artistSearchHash } from "./Atomic.ts"
-import type { SourceType } from "./Atomic.ts"
+import type { LogEvent, SourceType } from "./Atomic.ts"
 import type { ComponentType, DateLike, ErrorLike, JsonPlayObject, PlayState, QueueName, SOURCE_SOT_TYPES, SourcePlayerJson } from "./Atomic.ts"
 import type { Dayjs } from "dayjs"
 import type { ErrorIsh } from "./ErrorUtils.ts"
@@ -193,6 +193,7 @@ export type MsSseEvent<T extends object = Record<string, any>> = {
     componentUpdate: MsSseEventPayload<ComponentCommonApiJson>
     playInsert: MsSseEventPayload<PlayApiCommonDetailed>
     playUpdate: MsSseEventPayload<Partial<PlayApiCommonDetailed>>
+    log: LogEvent
 }
 
 export type SortPlaysBy = 'played' | 'seen';

src/core/Atomic.ts

@@ -524,6 +524,12 @@ export interface LogOutputConfig {
     limit: number
 }
 
+export interface LogEvent {
+    message: string
+    level: number
+    levelLabel: string
+}
+
 export type PlayPlatformIdStr = string;
 
 export interface SourcePlayerObj<D extends DateLike = Dayjs> {