AI Implementation feature(876): Authenticated Shell, Quick Tabs and Change Password 1.01 (#16)

frontend/src/app/core/services/authenticated-event-source.service.ts

@@ -0,0 +1,127 @@
+import { Injectable, inject } from '@angular/core';
+import { AuthService } from './auth.service';
+import { EventSourceLike } from './event-status.pure';
+
+interface SseMessage {
+  event: string;
+  data: string;
+}
+
+@Injectable({ providedIn: 'root' })
+export class AuthenticatedEventSourceService {
+  private readonly auth = inject(AuthService);
+
+  open(url: string): EventSourceLike {
+    const token = this.auth.getAccessToken();
+    const headers: Record<string, string> = { Accept: 'text/event-stream' };
+    if (token) headers['Authorization'] = `Bearer ${token}`;
+
+    const listeners = new Map<string, Array<(ev: MessageEvent | Event) => void>>();
+    let controller: AbortController | null = new AbortController();
+    let closed = false;
+    let buffered: SseMessage[] = [];
+
+    const dispatch = (type: string, ev: MessageEvent | Event) => {
+      const arr = listeners.get(type);
+      if (!arr) return;
+      for (const fn of arr) {
+        try {
+          fn(ev);
+        } catch {
+          // ignore listener errors
+        }
+      }
+    };
+
+    const flushBuffered = () => {
+      if (closed) return;
+      const items = buffered;
+      buffered = [];
+      for (const m of items) {
+        const me = new MessageEvent(m.event, { data: m.data });
+        dispatch('message', me);
+      }
+    };
+
+    const fireError = () => {
+      if (closed) return;
+      dispatch('error', new Event('error'));
+    };
+
+    const start = async () => {
+      try {
+        const res = await fetch(url, {
+          method: 'GET',
+          headers,
+          credentials: 'include',
+          signal: controller!.signal,
+        });
+        if (!res.ok || !res.body) {
+          fireError();
+          return;
+        }
+        const reader = res.body.getReader();
+        const decoder = new TextDecoder();
+        let buf = '';
+        for (;;) {
+          const { value, done } = await reader.read();
+          if (done) break;
+          buf += decoder.decode(value, { stream: true });
+          let idx: number;
+          while ((idx = buf.indexOf('\n\n')) >= 0) {
+            const raw = buf.slice(0, idx);
+            buf = buf.slice(idx + 2);
+            parseAndDispatch(raw);
+          }
+        }
+        if (buf.trim().length > 0) parseAndDispatch(buf);
+      } catch {
+        if (!closed) fireError();
+      }
+    };
+
+    const parseAndDispatch = (raw: string) => {
+      let event = 'message';
+      const dataLines: string[] = [];
+      for (const line of raw.split('\n')) {
+        if (line.startsWith(':')) continue;
+        if (line.startsWith('event:')) event = line.slice(6).trim();
+        else if (line.startsWith('data:')) dataLines.push(line.slice(5).trim());
+      }
+      if (dataLines.length === 0) return;
+      const payload = dataLines.join('\n');
+      if (listeners.has('message')) {
+        dispatch('message', new MessageEvent(event, { data: payload }));
+      } else {
+        buffered.push({ event, data: payload });
+      }
+    };
+
+    void start();
+
+    return {
+      addEventListener(type, listener) {
+        const arr = listeners.get(type) ?? [];
+        arr.push(listener);
+        listeners.set(type, arr);
+        if (type === 'message' && buffered.length > 0) {
+          queueMicrotask(flushBuffered);
+        }
+      },
+      close() {
+        if (closed) return;
+        closed = true;
+        if (controller) {
+          try {
+            controller.abort();
+          } catch {
+            // ignore
+          }
+          controller = null;
+        }
+        listeners.clear();
+        buffered = [];
+      },
+    };
+  }
+}

frontend/src/app/features/home/home.component.ts

@@ -14,6 +14,7 @@ import { filter } from 'rxjs/operators';
 import { BootstrapService } from '../../core/services/bootstrap.service';
 import { AuthService } from '../../core/services/auth.service';
 import { EventStatusStore } from '../../core/services/event-status.store';
+import { AuthenticatedEventSourceService } from '../../core/services/authenticated-event-source.service';
 import { UserStore } from '../../core/services/user.store';
 import { ShellHeaderComponent } from '../shell/header/shell-header.component';
 import {
@@ -88,6 +89,7 @@ export class HomeComponent implements OnInit, OnDestroy {
   readonly bootstrap = inject(BootstrapService);
   readonly auth = inject(AuthService);
   readonly eventStatus = inject(EventStatusStore);
+  private readonly authenticatedEventSource = inject(AuthenticatedEventSourceService);
   readonly userStore = inject(UserStore);
   private readonly router = inject(Router);
   private readonly destroyRef = inject(DestroyRef);
@@ -127,12 +129,7 @@ export class HomeComponent implements OnInit, OnDestroy {
   ngOnInit(): void {
     this.userStore.hydrateFromAuth();
     void this.userStore.loadMe();
-    if (typeof window !== 'undefined' && typeof (window as any).EventSource !== 'undefined') {
-      const Ctor = (window as any).EventSource as new (url: string, init?: any) => any;
-      this.eventStatus.start(() => {
-        return new Ctor('/api/v1/events/status', { withCredentials: true }) as any;
-      });
-    }
+    this.eventStatus.start(() => this.authenticatedEventSource.open('/api/v1/events/status'));
   }
 
   ngOnDestroy(): void {