diff --git a/src/backend/common/infrastructure/MSBackendEventMap.ts b/src/backend/common/infrastructure/MSBackendEventMap.ts index 8a407a13..d9891d9e 100644 --- a/src/backend/common/infrastructure/MSBackendEventMap.ts +++ b/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 { playerUpdate: [EmittedMSEvent & { options: { scrobbleTo: string[], wasMonitored?: boolean } },{},SourceType>] configError: [Error | string] instantiateError: [Error] + log: [LogEvent] } \ No newline at end of file diff --git a/src/backend/server/api.ts b/src/backend/server/api.ts index aec78f36..422973a3 100644 --- a/src/backend/server/api.ts +++ b/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(); 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); diff --git a/src/backend/tests/server/events.test.ts b/src/backend/tests/server/events.test.ts new file mode 100644 index 00000000..c1e78310 --- /dev/null +++ b/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) { + 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(); + } + }); +}); diff --git a/src/client/App.tsx b/src/client/App.tsx index ec01d9d4..e57255bc 100644 --- a/src/client/App.tsx +++ b/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() { diff --git a/src/client/components/LogsNext.tsx b/src/client/components/LogsNext.tsx index 1e5d0cfb..14eb3be1 100644 --- a/src/client/components/LogsNext.tsx +++ b/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 = 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, ref?: React.Ref}) => { return {props.logs.map((x, index) => )} @@ -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(); + 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)); } }); diff --git a/src/client/sseTab.ts b/src/client/sseTab.ts new file mode 100644 index 00000000..36670d7c --- /dev/null +++ b/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); diff --git a/src/core/Api.ts b/src/core/Api.ts index 5213e649..e22aca3f 100644 --- a/src/core/Api.ts +++ b/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> = { componentUpdate: MsSseEventPayload playInsert: MsSseEventPayload playUpdate: MsSseEventPayload> + log: LogEvent } export type SortPlaysBy = 'played' | 'seen'; diff --git a/src/core/Atomic.ts b/src/core/Atomic.ts index 3b76bd07..9f0ecd29 100644 --- a/src/core/Atomic.ts +++ b/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 {