feat(app): stream live server logs to the management UI via SSE (#45)

Closes #43

Adds a real-time log tail to the Logs panel:

- LogsService.streamLogs(game, signal) – AsyncGenerator that polls
  FilterLogEvents every 2 s, de-duplicates by eventId, advances
  startTime on each tick, and exits cleanly when the AbortSignal fires.
  Queries the whole log group so task stop+start (new stream) is handled
  automatically.

- GET /api/logs/:game/stream – @sse() endpoint wrapping the generator
  in an RxJS Observable. Teardown aborts the generator when the client
  disconnects.

- ApiTokenGuard – falls back to ?token= query param when the
  Authorization header is absent, which is necessary because the
  browser's native EventSource cannot set custom headers.

- LogsPanel.tsx – fetches a snapshot on game change then opens an
  EventSource stream that appends new lines (capped at 1 000). Pause
  buffers incoming lines; Resume flushes the buffer. Replaces the old
  manual Refresh button.

- LogsService.test.ts – seven new streamLogs cases covering clean
  termination, multi-poll delivery, de-duplication, log-group naming,
  error recovery, and missing message fields.

- docs/components/management-app.md – updated endpoint table, Live Logs
  description, auth notes, and LogsService blurb.

IAM: logs:* already covers FilterLogEvents (docs/setup.md unchanged).

https://claude.ai/code/session_01THkoGDYjJhojhtnUEYi4w9

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: CoderCoco

CLAUDE.md

@@ -154,6 +154,7 @@ Any time you add or remove Terraform variables, update **all four** of these in
 - **Always use `/pr` to create pull requests.** The `.claude/commands/pr.md` skill validates the title format before calling the API. Never call `mcp__github__create_pull_request` directly without running this check first.
 - **PR titles MUST use Conventional Commits.** We squash-merge, so the PR title becomes the commit subject on `main` verbatim — a badly-formed title produces a badly-formed commit that can't be fixed after merge. Format: `<type>(<optional-scope>): <imperative summary>`, where `<type>` is one of `feat`, `fix`, `refactor`, `docs`, `test`, `chore`, `perf`, `build`, `ci`, `style`. Keep the subject under ~70 characters; put details in the PR body. Examples: `refactor(app): migrate server from Express+tsyringe to Nest.js`, `docs: reflect Nest.js migration in CLAUDE.md`, `fix(watchdog): stop leaking tags on failed runs`, `chore: add ESLint flat config`.
 - **Pre-flight check (mandatory):** before any `create_pull_request` call, verify the title matches `^(feat|fix|refactor|docs|test|chore|perf|build|ci|style)(\([^)]+\))?: .+$`. If it doesn't, fix it first. `Add ESLint configuration` fails (no type prefix); `chore: add ESLint configuration` passes.
+- **Always include `Closes #N`** in the PR body when the PR resolves a GitHub issue. Place it as the first line so GitHub auto-closes the issue on merge.
 
 ## PR Review Workflow
 

app/packages/server/src/controllers/logs.controller.ts

@@ -1,4 +1,5 @@
-import { Controller, Get, Param, Query } from '@nestjs/common';
+import { Controller, Get, MessageEvent, Param, Query, Sse } from '@nestjs/common';
+import { Observable } from 'rxjs';
 import { LogsService } from '../services/LogsService.js';
 
 /** Tails CloudWatch logs from the `/ecs/{game}-server` log group. */
@@ -13,4 +14,35 @@ export class LogsController {
     const lines = await this.logs.getRecentLogs(game, limit);
     return { game, lines };
   }
+
+  /**
+   * SSE stream of new log lines for a game, delivered as they arrive from
+   * `FilterLogEvents`. The client receives `{ data: { line: "..." } }` events.
+   * Auth: `Authorization: Bearer` header OR `?token=` query param (the latter
+   * is required because the browser's native `EventSource` cannot set headers).
+   */
+  @Sse(':game/stream')
+  streamLogs(@Param('game') game: string): Observable<MessageEvent> {
+    const ac = new AbortController();
+
+    return new Observable<MessageEvent>((subscriber) => {
+      const run = async () => {
+        try {
+          for await (const line of this.logs.streamLogs(game, ac.signal)) {
+            subscriber.next({ data: { line } } as MessageEvent);
+          }
+          subscriber.complete();
+        } catch (err) {
+          if ((err as Error).name === 'AbortError') {
+            subscriber.complete();
+          } else {
+            subscriber.error(err);
+          }
+        }
+      };
+      void run();
+
+      return () => ac.abort();
+    });
+  }
 }

app/packages/server/src/guards/api-token.guard.ts

@@ -55,11 +55,22 @@ export class ApiTokenGuard implements CanActivate {
       return true;
     }
 
+    // Prefer the Authorization header; fall back to ?token= query param for
+    // SSE endpoints where the browser's native EventSource cannot set headers.
     const header = req.headers['authorization'];
-    if (typeof header !== 'string' || !header.startsWith('Bearer ')) {
-      throw new UnauthorizedException({ error: 'missing bearer token' });
+    let presented: string;
+    if (typeof header === 'string' && header.startsWith('Bearer ')) {
+      presented = header.slice('Bearer '.length).trim();
+    } else {
+      const queryToken = (req.query as Record<string, unknown> | undefined)?.['token'];
+      if (typeof queryToken !== 'string') {
+        throw new UnauthorizedException({ error: 'missing bearer token' });
+      }
+      // Remove the token from req.query so RequestLoggerMiddleware doesn't
+      // log it — the finish handler fires after this guard runs.
+      delete (req.query as Record<string, unknown>)['token'];
+      presented = queryToken;
     }
-    const presented = header.slice('Bearer '.length).trim();
     if (presented !== configured) {
       logger.warn('Rejected /api request with bad bearer token', {
         path: req.path,

app/packages/server/src/services/LogsService.test.ts

@@ -4,6 +4,7 @@ import { mockClient } from 'aws-sdk-client-mock';
 import {
   CloudWatchLogsClient,
   DescribeLogStreamsCommand,
+  FilterLogEventsCommand,
   GetLogEventsCommand,
 } from '@aws-sdk/client-cloudwatch-logs';
 
@@ -104,3 +105,137 @@ describe('LogsService', () => {
     expect(lines[0]).toContain('denied');
   });
 });
+
+describe('LogsService.streamLogs', () => {
+  /** Service under test, freshly constructed per test. */
+  let service: LogsService;
+
+  beforeEach(() => {
+    cwMock.reset();
+    service = new LogsService(makeConfig());
+  });
+
+  it('should terminate immediately when signal is already aborted before the first poll', async () => {
+    const ac = new AbortController();
+    ac.abort();
+
+    const lines: string[] = [];
+    for await (const line of service.streamLogs('minecraft', ac.signal, 0)) {
+      lines.push(line);
+    }
+
+    expect(lines).toEqual([]);
+    expect(cwMock.commandCalls(FilterLogEventsCommand)).toHaveLength(0);
+  });
+
+  it('should yield log lines from the first poll and terminate on abort', async () => {
+    cwMock.on(FilterLogEventsCommand).resolves({
+      events: [
+        { eventId: 'e1', timestamp: 1000, message: 'line1' },
+        { eventId: 'e2', timestamp: 2000, message: 'line2' },
+      ],
+    });
+
+    const ac = new AbortController();
+    const gen = service.streamLogs('minecraft', ac.signal, 0);
+
+    const { value: l1 } = await gen.next();
+    const { value: l2 } = await gen.next();
+    ac.abort();
+    const { done } = await gen.next();
+
+    expect(l1).toBe('line1');
+    expect(l2).toBe('line2');
+    expect(done).toBe(true);
+  });
+
+  it('should yield new events from successive polls', async () => {
+    cwMock
+      .on(FilterLogEventsCommand)
+      .resolvesOnce({ events: [{ eventId: 'e1', timestamp: 1000, message: 'first' }] })
+      .resolves({ events: [{ eventId: 'e2', timestamp: 2000, message: 'second' }] });
+
+    const ac = new AbortController();
+    const gen = service.streamLogs('minecraft', ac.signal, 0);
+
+    const { value: l1 } = await gen.next();
+    const { value: l2 } = await gen.next();
+    ac.abort();
+    await gen.return(undefined);
+
+    expect(l1).toBe('first');
+    expect(l2).toBe('second');
+  });
+
+  it('should de-duplicate events with the same eventId across polls', async () => {
+    cwMock
+      .on(FilterLogEventsCommand)
+      .resolvesOnce({ events: [{ eventId: 'e1', timestamp: 1000, message: 'line1' }] })
+      .resolvesOnce({
+        events: [
+          { eventId: 'e1', timestamp: 1000, message: 'line1' }, // already seen
+          { eventId: 'e2', timestamp: 2000, message: 'line2' }, // new
+        ],
+      });
+
+    const ac = new AbortController();
+    const gen = service.streamLogs('minecraft', ac.signal, 0);
+
+    const { value: l1 } = await gen.next(); // first poll yields 'line1'
+    const { value: l2 } = await gen.next(); // second poll skips duplicate, yields 'line2'
+    ac.abort();
+    await gen.return(undefined);
+
+    expect(l1).toBe('line1');
+    expect(l2).toBe('line2');
+  });
+
+  it('should query the /ecs/{game}-server log group', async () => {
+    cwMock.on(FilterLogEventsCommand).resolves({
+      events: [{ eventId: 'e1', timestamp: 1000, message: 'hello' }],
+    });
+
+    const ac = new AbortController();
+    const gen = service.streamLogs('valheim', ac.signal, 0);
+
+    await gen.next(); // first poll runs and yields 'hello'
+    ac.abort();
+    await gen.return(undefined);
+
+    const calls = cwMock.commandCalls(FilterLogEventsCommand);
+    expect(calls[0]!.args[0].input.logGroupName).toBe('/ecs/valheim-server');
+  });
+
+  it('should yield a stream-error sentinel and continue when a poll throws', async () => {
+    cwMock
+      .on(FilterLogEventsCommand)
+      .rejectsOnce(new Error('throttled'))
+      .resolves({ events: [{ eventId: 'e1', timestamp: 1000, message: 'recovered' }] });
+
+    const ac = new AbortController();
+    const gen = service.streamLogs('minecraft', ac.signal, 0);
+
+    const { value: errLine } = await gen.next(); // first poll throws
+    const { value: okLine } = await gen.next(); // second poll succeeds
+    ac.abort();
+    await gen.return(undefined);
+
+    expect(errLine).toMatch(/\[stream error\].*throttled/);
+    expect(okLine).toBe('recovered');
+  });
+
+  it('should map a missing event.message to an empty string', async () => {
+    cwMock.on(FilterLogEventsCommand).resolves({
+      events: [{ eventId: 'e1', timestamp: 1000 }], // no message field
+    });
+
+    const ac = new AbortController();
+    const gen = service.streamLogs('minecraft', ac.signal, 0);
+
+    const { value: line } = await gen.next();
+    ac.abort();
+    await gen.return(undefined);
+
+    expect(line).toBe('');
+  });
+});

app/packages/server/src/services/LogsService.ts

@@ -2,11 +2,35 @@ import { Injectable } from '@nestjs/common';
 import {
   CloudWatchLogsClient,
   DescribeLogStreamsCommand,
+  FilterLogEventsCommand,
   GetLogEventsCommand,
 } from '@aws-sdk/client-cloudwatch-logs';
 import { logger } from '../logger.js';
 import { ConfigService } from './ConfigService.js';
 
+/**
+ * Sleep for `ms` milliseconds, but reject immediately if `signal` is aborted.
+ * Used by `streamLogs` so the poll loop exits promptly when the SSE client disconnects.
+ */
+function sleepInterruptible(ms: number, signal: AbortSignal): Promise<void> {
+  return new Promise((resolve, reject) => {
+    if (signal.aborted) {
+      reject(new DOMException('Aborted', 'AbortError'));
+      return;
+    }
+    const onAbort = () => {
+      clearTimeout(timer);
+      signal.removeEventListener('abort', onAbort);
+      reject(new DOMException('Aborted', 'AbortError'));
+    };
+    const timer = setTimeout(() => {
+      signal.removeEventListener('abort', onAbort);
+      resolve();
+    }, ms);
+    signal.addEventListener('abort', onAbort);
+  });
+}
+
 /**
  * Fetches recent CloudWatch Logs lines for a game's ECS task so the UI can
  * render a tail. Assumes the Terraform-provisioned log group naming
@@ -25,6 +49,55 @@ export class LogsService {
     return this.client;
   }
 
+  /**
+   * Async generator that polls `FilterLogEvents` every `pollInterval` ms and
+   * yields new log lines as they arrive. De-duplicates by `eventId` so lines
+   * are never emitted twice even when `startTime` windows overlap at the
+   * boundary. The generator exits cleanly when `signal` is aborted (i.e.
+   * when the SSE client disconnects).
+   *
+   * Queries the whole log group so that a stop+start of the ECS task
+   * (which creates a new stream) is handled automatically without reconnecting.
+   */
+  async *streamLogs(
+    game: string,
+    signal: AbortSignal,
+    pollInterval = 2000,
+  ): AsyncGenerator<string> {
+    const logGroup = `/ecs/${game}-server`;
+    let startTime = Date.now();
+    const seen = new Set<string>();
+
+    while (!signal.aborted) {
+      try {
+        const resp = await this.getClient().send(
+          new FilterLogEventsCommand({ logGroupName: logGroup, startTime, limit: 100 }),
+          { abortSignal: signal },
+        );
+        for (const e of resp.events ?? []) {
+          const id = e.eventId ?? `${e.timestamp}-${e.message}`;
+          if (!seen.has(id)) {
+            seen.add(id);
+            yield e.message ?? '';
+          }
+          if ((e.timestamp ?? 0) >= startTime) {
+            startTime = (e.timestamp ?? startTime) + 1;
+          }
+        }
+      } catch (err) {
+        if ((err as Error).name === 'AbortError') break;
+        logger.error('Log stream poll error', { err, game, logGroup });
+        yield `[stream error] ${String(err)}`;
+      }
+
+      try {
+        await sleepInterruptible(pollInterval, signal);
+      } catch {
+        break;
+      }
+    }
+  }
+
   /**
    * Return up to `limit` recent messages from the most recently written log
    * stream in `/ecs/{game}-server`. Errors are folded into a single-element