feat(proposal-datasets): live dataset table updates with SSE

[Junjiequan]

Description

Adds real-time dataset updates to the proposal-datasets table via Server-Sent Events (SSE). When enabled, newly created datasets appear in the table automatically and the new row flashes briefly to draw attention, without a manual reload.

Fixes:

###Render fix (key change)
Switched the shared table's trackBy from index-based to _id-based identity tracking with index as fallback, and removed the empty-then-refill (data = []) on reload. Previously every reload remounted all rows, causing the whole table to re-animate. Now a reload that adds one dataset reuses existing row DOM nodes and only mounts the new row, so only it animates.

Changes:

Real-time toggle

Adds a Live updates toggle to the shared DynamicMatTableComponent, gated to logged-in users. The parent owns the SSE connection; the table only renders the toggle and emits its state.

###SSE wiring
EventsService manages the EventSource lifecycle (connect on toggle-on, disconnect on toggle-off and on destroy) and exposes a message$ stream.
On Dataset.created, the proposal-datasets page reloads the list newest-first so the new dataset appears at the top.

Per-row highlight

New rows flash for a few seconds, each on its own timer, so multiple arrivals highlight independently rather than cancelling each other.
Highlight is applied via a separate visual layer (box-shadow overlay) so it no longer conflicts with the existing zebra-stripe background-color.

Tests included

• Included for each change/fix?
• Passing? (Merge will not be approved unless this is checked)

Documentation

• swagger documentation updated [required]
• official documentation updated [nice-to-have]

official documentation info

If you have updated the official documentation, please provide PR # and URL of the pages where the updates are included

Backend version

• Does it require a specific version of the backend
• which version of the backend is required:

Summary by Sourcery

Add optional real-time dataset updates to the proposal datasets table using a shared SSE events service and visual cues for live activity.

New Features:

• Introduce a shared EventsService that connects to the backend SSE stream using the user token and exposes incoming messages and transient latest-updated IDs.
• Add an optional live-updates slide toggle and animated border/highlight states to the dynamic material table for indicating real-time activity.
• Enable live dataset updates in the proposal datasets view, refreshing and highlighting rows when new datasets are created for logged-in users.

Enhancements:

• Refactor proposal datasets subscriptions to support multiple concurrent subscriptions and ensure proper cleanup on destroy.
• Wire table components and core directive to accept real-time related inputs and emit toggle changes without impacting existing table behavior.

Summary by Sourcery

Integrate optional real-time dataset updates into the proposal datasets table using a shared SSE events service and visual highlighting for new rows.

New Features:

• Add an EventsService that manages an authenticated SSE connection and exposes incoming event messages and latest-updated dataset IDs.
• Expose an optional Live updates toggle and real-time visual states (border and per-row flash) in the shared dynamic material table component.
• Enable live updates for the proposal datasets view so newly created datasets for the active proposal appear automatically and are highlighted when SSE is enabled.

Enhancements:

• Refactor proposal datasets subscriptions to support multiple concurrent subscriptions with proper cleanup on destroy.
• Change table row tracking to use dataset _id instead of index and avoid fully resetting the data source on updates to prevent unnecessary remounting and animations.

Build:

• Increase the global anyComponentStyle maximumError size limit in angular.json to accommodate the additional table styles.

[sourcery-ai]

Hey - I've found 3 issues, and left some high level feedback:

• EventsService.latestUpdatedId$ assumes every SSE message has a data._id field; consider narrowing this stream to the event types you expect (e.g. Dataset.created) and adding guards so malformed or different message shapes don’t cause runtime errors.
• Because EventsService is provided in root and connect/disconnect are global, toggling live updates off in one component will tear down the SSE connection for all consumers; consider ref-counting active subscribers or tracking per-feature enablement so one view doesn’t inadvertently disable updates elsewhere.
• The DynamicMatTableComponent latestUpdatedId @input will receive null/undefined values from the async pipe when there’s no event yet; it would be safer to type this as string | null and short‑circuit in the setter to avoid adding invalid ids to the highlighted set.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- EventsService.latestUpdatedId$ assumes every SSE message has a data._id field; consider narrowing this stream to the event types you expect (e.g. Dataset.created) and adding guards so malformed or different message shapes don’t cause runtime errors.
- Because EventsService is provided in root and connect/disconnect are global, toggling live updates off in one component will tear down the SSE connection for all consumers; consider ref-counting active subscribers or tracking per-feature enablement so one view doesn’t inadvertently disable updates elsewhere.
- The DynamicMatTableComponent latestUpdatedId @Input will receive null/undefined values from the async pipe when there’s no event yet; it would be safer to type this as string | null and short‑circuit in the setter to avoid adding invalid ids to the highlighted set.

## Individual Comments

### Comment 1
<location path="src/app/proposals/proposal-datasets/proposal-datasets.component.ts" line_range="146-153" />
<code_context>

   ngOnInit(): void {
+    this.subscriptions.push(
+      this.eventsService.message$
+        .pipe(
+          filter((payload) => {
+            return payload.type === "Dataset.created";
+          }),
+        )
+        .subscribe((payload: Record<string, any>) => {
+          if (!payload.data.proposalIds.includes(this.proposalId)) return;
+          this.store.dispatch(
+            fetchProposalDatasetsAction({
</code_context>
<issue_to_address>
**issue (bug_risk):** The Dataset.created handler assumes `payload.data.proposalIds` always exists, which can throw for unexpected payloads.

Because only `payload.type` is checked, a malformed or changed event shape (e.g. missing `data` or `data.proposalIds`) will throw at `.includes()`. Please guard access defensively, e.g.:

```ts
const proposalIds = payload.data?.proposalIds as string[] | undefined;
if (!proposalIds?.includes(this.proposalId)) return;
```

This avoids runtime errors from unexpected payloads.
</issue_to_address>

### Comment 2
<location path="src/app/shared/events.service.ts" line_range="23-24" />
<code_context>
+
+  message$ = this.messageSubject.asObservable();
+
+  latestUpdatedId$ = this.messageSubject.pipe(
+    map((m) => (m["data"] as { _id: string })._id),
+  );
+
</code_context>
<issue_to_address>
**issue (bug_risk):** latestUpdatedId$ assumes every event has `data._id`, which may cause runtime errors if other event types are emitted.

If `messageSubject` ever emits an event without `data._id`, this cast `(m["data"] as { _id: string })._id` will throw at runtime. Consider filtering to only the relevant event types before mapping, or making this defensive (e.g. guard checks / optional chaining and handling an undefined ID at the call site).
</issue_to_address>

### Comment 3
<location path="src/app/shared/events.service.ts" line_range="16" />
<code_context>
+} from "rxjs";
+
+@Injectable({ providedIn: "root" })
+export class EventsService {
+  private eventSource: EventSource | null = null;
+  private connectionSub: Subscription | null = null;
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the EventsService so RxJS manages the EventSource lifecycle and domain-specific projections live outside this generic transport service.

You can simplify this service by letting RxJS own the EventSource lifecycle instead of tracking `eventSource` and `open/closeConnection` imperatively, while keeping the `connect`/`disconnect` API and behavior intact.

### 1. Wrap `EventSource` in an observable and remove `eventSource` state

```ts
import { EMPTY, Observable, Subject, Subscription, distinctUntilChanged, switchMap } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class EventsService {
  private connectionSub: Subscription | null = null;
  private messageSubject = new Subject<Record<string, unknown>>();

  message$ = this.messageSubject.asObservable();

  constructor(
    private ngZone: NgZone,
    private store: Store,
  ) {}

  private createEventStream(token: string): Observable<Record<string, unknown>> {
    return new Observable<Record<string, unknown>>((observer) => {
      const es = new EventSource(`/api/v3/events/stream?token=${token}`);

      es.onmessage = (event) => {
        this.ngZone.run(() => observer.next(JSON.parse(event.data)));
      };

      es.onerror = () => {
        es.close();
        observer.complete(); // same effect as closeConnection(): no more events
      };

      return () => es.close();
    });
  }

  connect() {
    if (this.connectionSub) return;

    this.store.dispatch(fetchScicatTokenAction());

    this.connectionSub = this.store.select(selectScicatToken).pipe(
      distinctUntilChanged(),
      switchMap((token) => (token ? this.createEventStream(token) : EMPTY)),
    ).subscribe((msg) => this.messageSubject.next(msg));
  }

  disconnect() {
    this.connectionSub?.unsubscribe();
    this.connectionSub = null;
  }
}
```

This:

- Removes `eventSource`, `openConnection`, and `closeConnection` as mutable state and imperative wiring.
- Keeps the current `connect`/`disconnect` contract and token-driven behavior.
- Centralizes lifecycle in a single Rx pipeline (`selectScicatToken` → `switchMap` → `EventSource`).

### 2. Consider moving `latestUpdatedId$` closer to the domain

If possible, keep this service transport-focused and define domain-specific projections where they are used:

```ts
// In feature/domain layer, not in EventsService
const latestUpdatedId$ = eventsService.message$.pipe(
  map((m) => (m['data'] as { _id: string })._id),
);
```

This avoids baking a specific message shape into the generic event service and keeps it simpler to extend when more event types are introduced.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[Copilot]

Pull request overview

Adds optional Server-Sent Events (SSE) support to show real-time dataset creations in the proposal datasets table, including a UI toggle, a live-border indicator, and per-row highlight behavior. This is implemented via a new shared EventsService and updates to the shared dynamic material table component.

Changes:

• Introduces a new EventsService that manages an EventSource connection and exposes message$ / latestUpdatedId$.
• Extends DynamicMatTableComponent with a “Live updates” toggle, live-border styling, row flash styling, and improves trackBy to prefer _id.
• Wires ProposalDatasetsComponent to toggle SSE on/off, reload on Dataset.created, and pass the latest-updated id down for row highlighting.

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
src/app/shared/modules/dynamic-material-table/table/dynamic-mat-table.module.ts	Imports MatSlideToggleModule for the new live-updates toggle UI.
src/app/shared/modules/dynamic-material-table/table/dynamic-mat-table.component.ts	Adds highlight tracking, latestUpdatedId input handling, and changes trackBy to prefer _id.
src/app/shared/modules/dynamic-material-table/table/dynamic-mat-table.component.scss	Adds live-border and row flash animations plus minor layout styling.
src/app/shared/modules/dynamic-material-table/table/dynamic-mat-table.component.html	Renders the live-updates toggle and applies row highlight class.
src/app/shared/modules/dynamic-material-table/cores/table.core.directive.ts	Adds inputs/outputs for the live-updates toggle state plumbing.
src/app/shared/events.service.ts	New singleton service for SSE connection lifecycle and message streams.
src/app/proposals/proposal-datasets/proposal-datasets.component.ts	Subscribes to SSE messages, reloads on Dataset.created, toggles SSE connection.
src/app/proposals/proposal-datasets/proposal-datasets.component.html	Wires live-updates inputs/outputs and latest-updated id into the shared table.
src/app/datasets/dashboard/dashboard.component.ts	Removes an unused import.
angular.json	Increases the anyComponentStyle budget maximum error threshold.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.