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

angular.json

@@ -61,7 +61,7 @@
                 {
                   "type": "anyComponentStyle",
                   "maximumWarning": "8kb",
-                  "maximumError": "10kb"
+                  "maximumError": "20kb"
                 }
               ],
               "fileReplacements": [

src/app/datasets/dashboard/dashboard.component.ts

@@ -30,7 +30,6 @@ import {
 import { distinctUntilChanged, filter, map, take } from "rxjs/operators";
 import { MatDialog } from "@angular/material/dialog";
 import { MatSidenav } from "@angular/material/sidenav";
-import { AddDatasetDialogComponent } from "datasets/add-dataset-dialog/add-dataset-dialog.component";
 import { combineLatest, Subscription, lastValueFrom } from "rxjs";
 import {
   selectProfile,
@@ -58,6 +57,7 @@ export class DashboardComponent implements OnInit, OnDestroy {
   private readyToFetch$ = this.store
     .select(selectHasPrefilledFilters)
     .pipe(filter((has) => has));
+
   loggedIn$ = this.store.select(selectIsLoggedIn);
   selectedSets$ = this.store.select(selectSelectedDatasets);
   selectColumns$ = this.store.select(selectColumns);

src/app/proposals/proposal-datasets/proposal-datasets.component.html

@@ -22,5 +22,9 @@
   class="mat-elevation-z2 proposal-dataset-table"
   [emptyMessage]="'No datasets available'"
   [emptyIcon]="'folder'"
+  [showRealTimeToggle]="isLoggedIn$ | async"
+  [latestUpdatedId]="latestUpdatedId$ | async"
+  [realTimeEnabled]="realTimeEnabled"
+  (realTimeEnabledChange)="onRealTimeToggle($event)"
 >
 </dynamic-mat-table>

src/app/proposals/proposal-datasets/proposal-datasets.component.ts

@@ -6,7 +6,13 @@ import { ActivatedRoute, Router } from "@angular/router";
 import { Store } from "@ngrx/store";
 import { OutputDatasetObsoleteDto } from "@scicatproject/scicat-sdk-ts-angular";
 import { AppConfigService } from "app-config.service";
-import { BehaviorSubject, lastValueFrom, Subscription, take } from "rxjs";
+import {
+  BehaviorSubject,
+  filter,
+  lastValueFrom,
+  Subscription,
+  take,
+} from "rxjs";
 import { PrintConfig } from "shared/modules/dynamic-material-table/models/print-config.model";
 import { TableField } from "shared/modules/dynamic-material-table/models/table-field.model";
 import {
@@ -28,7 +34,11 @@ import { DatasetsListService } from "shared/services/datasets-list.service";
 import { TableConfigService } from "shared/services/table-config.service";
 import { fetchProposalDatasetsAction } from "state-management/actions/proposals.actions";
 import { selectViewProposalPageViewModel } from "state-management/selectors/proposals.selectors";
-import { selectColumnsWithHasFetchedSettings } from "state-management/selectors/user.selectors";
+import {
+  selectColumnsWithHasFetchedSettings,
+  selectIsLoggedIn,
+} from "state-management/selectors/user.selectors";
+import { EventsService } from "shared/events.service";
 
 export interface TableData {
   pid: string;
@@ -47,16 +57,20 @@ export interface TableData {
   standalone: false,
 })
 export class ProposalDatasetsComponent implements OnInit, OnDestroy {
+  isLoggedIn$ = this.store.select(selectIsLoggedIn);
   proposalDatasets$ = this.store.select(selectViewProposalPageViewModel);
+  latestUpdatedId$ = this.eventsService.latestUpdatedId$;
 
-  subscription: Subscription;
+  subscriptions: Subscription[] = [];
   @Input() proposalId: string;
 
   appConfig = this.appConfigService.getConfig();
   selectColumnsWithFetchedSettings$ = this.store.select(
     selectColumnsWithHasFetchedSettings,
   );
 
+  realTimeEnabled = false;
+
   tableName = "proposalDatasetsTable";
 
   columns: TableField<any>[];
@@ -75,9 +89,6 @@ export class ProposalDatasetsComponent implements OnInit, OnDestroy {
 
   showNoData = true;
 
-  //dataSource: BehaviorSubject<TableData[]> = new BehaviorSubject<TableData[]>(
-  //  [],
-  //);
   dataSource: BehaviorSubject<OutputDatasetObsoleteDto[]> = new BehaviorSubject<
     OutputDatasetObsoleteDto[]
   >([]);
@@ -127,9 +138,31 @@ export class ProposalDatasetsComponent implements OnInit, OnDestroy {
     private store: Store,
     private tableConfigService: TableConfigService,
     private datasetsListService: DatasetsListService,
+    private eventsService: EventsService,
   ) {}
 
   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({
+              proposalId: this.proposalId,
+              skip: 0,
+              limit: this.defaultPageSize,
+              sortColumn: "creationTime",
+              sortDirection: "desc",
+            }),
+          );
+        }),
+    );
+
     this.store.dispatch(
       fetchProposalDatasetsAction({
         proposalId: this.proposalId,
@@ -138,44 +171,46 @@ export class ProposalDatasetsComponent implements OnInit, OnDestroy {
       }),
     );
 
-    this.subscription = this.proposalDatasets$.subscribe(async (data) => {
-      this.dataSource.next(data.datasets);
-      this.pending = false;
+    this.subscriptions.push(
+      this.proposalDatasets$.subscribe(async (data) => {
+        this.dataSource.next(data.datasets);
+        this.pending = false;
 
-      const defaultTableColumns = await lastValueFrom(
-        this.selectColumnsWithFetchedSettings$.pipe(take(1)),
-      );
-
-      const defaultConfigColumns =
-        this.appConfig?.defaultDatasetsListSettings?.columns;
-
-      const userTableConfigColumns =
-        this.datasetsListService.convertSavedDatasetColumns(
-          defaultTableColumns.columns,
+        const defaultTableColumns = await lastValueFrom(
+          this.selectColumnsWithFetchedSettings$.pipe(take(1)),
         );
 
-      this.tableDefaultSettingsConfig.settingList[0].columnSetting =
-        this.datasetsListService.convertSavedDatasetColumns(
-          defaultConfigColumns as TableColumn[],
-        );
-
-      const tableSettingsConfig =
-        this.tableConfigService.getTableSettingsConfig(
-          this.tableName,
-          this.tableDefaultSettingsConfig,
-          userTableConfigColumns,
-        );
-      const paginationConfig = {
-        pageSizeOptions: [5, 10, 25, 100],
-        pageIndex: data.currentPage || 0,
-        pageSize: data.datasetsPerPage || this.defaultPageSize,
-        length: data.datasetCount,
-      };
-
-      if (tableSettingsConfig?.settingList.length) {
-        this.initTable(tableSettingsConfig, paginationConfig);
-      }
-    });
+        const defaultConfigColumns =
+          this.appConfig?.defaultDatasetsListSettings?.columns;
+
+        const userTableConfigColumns =
+          this.datasetsListService.convertSavedDatasetColumns(
+            defaultTableColumns.columns,
+          );
+
+        this.tableDefaultSettingsConfig.settingList[0].columnSetting =
+          this.datasetsListService.convertSavedDatasetColumns(
+            defaultConfigColumns as TableColumn[],
+          );
+
+        const tableSettingsConfig =
+          this.tableConfigService.getTableSettingsConfig(
+            this.tableName,
+            this.tableDefaultSettingsConfig,
+            userTableConfigColumns,
+          );
+        const paginationConfig = {
+          pageSizeOptions: [5, 10, 25, 100],
+          pageIndex: data.currentPage || 0,
+          pageSize: data.datasetsPerPage || this.defaultPageSize,
+          length: data.datasetCount,
+        };
+
+        if (tableSettingsConfig?.settingList.length) {
+          this.initTable(tableSettingsConfig, paginationConfig);
+        }
+      }),
+    );
   }
 
   initTable(
@@ -191,6 +226,24 @@ export class ProposalDatasetsComponent implements OnInit, OnDestroy {
     this.pagination = paginationConfig;
   }
 
+  onRealTimeToggle = (enabled: boolean) => {
+    this.realTimeEnabled = enabled;
+    if (enabled) {
+      this.eventsService.connect();
+      this.store.dispatch(
+        fetchProposalDatasetsAction({
+          proposalId: this.proposalId,
+          skip: 0,
+          limit: this.defaultPageSize,
+          sortColumn: "creationTime",
+          sortDirection: "desc",
+        }),
+      );
+    } else {
+      this.eventsService.disconnect();
+    }
+  };
+
   formatTableData(datasets: OutputDatasetObsoleteDto[]): TableData[] {
     let tableData: TableData[] = [];
     if (datasets) {
@@ -269,6 +322,7 @@ export class ProposalDatasetsComponent implements OnInit, OnDestroy {
   }
 
   ngOnDestroy() {
-    this.subscription.unsubscribe();
+    this.eventsService.disconnect();
+    this.subscriptions.forEach((subscription) => subscription.unsubscribe());
   }
 }

src/app/shared/events.service.ts

@@ -0,0 +1,69 @@
+import { Injectable, NgZone } from "@angular/core";
+import { Store } from "@ngrx/store";
+import { fetchScicatTokenAction } from "state-management/actions/user.actions";
+import { selectScicatToken } from "state-management/selectors/user.selectors";
+import {
+  distinctUntilChanged,
+  map,
+  startWith,
+  Subject,
+  Subscription,
+  switchMap,
+  timer,
+} from "rxjs";
+
+@Injectable({ providedIn: "root" })
+export class EventsService {
+  private eventSource: EventSource | null = null;
+  private connectionSub: Subscription | null = null;
+  private messageSubject = new Subject<Record<string, unknown>>();
+
+  message$ = this.messageSubject.asObservable();
+
+  latestUpdatedId$ = this.messageSubject.pipe(
+    map((m) => (m["data"] as { _id: string })._id),
+  );
+
+  constructor(
+    private ngZone: NgZone,
+    private store: Store,
+  ) {}
+
+  connect() {
+    if (this.connectionSub) return;
+
+    this.store.dispatch(fetchScicatTokenAction());
+
+    this.connectionSub = this.store
+      .select(selectScicatToken)
+      .pipe(distinctUntilChanged())
+      .subscribe((token) => {
+        if (token) {
+          this.openConnection(token);
+        } else {
+          this.closeConnection();
+        }
+      });
+  }
+
+  disconnect() {
+    this.closeConnection();
+    this.connectionSub?.unsubscribe();
+    this.connectionSub = null;
+  }
+
+  private openConnection(token: string) {
+    this.closeConnection();
+
+    this.eventSource = new EventSource(`/api/v3/events/stream?token=${token}`);
+    this.eventSource.onmessage = (event) => {
+      this.ngZone.run(() => this.messageSubject.next(JSON.parse(event.data)));
+    };
+    this.eventSource.onerror = () => this.closeConnection();
+  }
+
+  private closeConnection() {
+    this.eventSource?.close();
+    this.eventSource = null;
+  }
+}

src/app/shared/modules/dynamic-material-table/cores/table.core.directive.ts

@@ -75,6 +75,9 @@ export class TableCoreDirective<T extends TableRow> {
   @Input() globalTextSearchPlaceholder = "Search...";
   @Input() selectionIds = [];
   @Input() disableBorder: boolean;
+  @Input() showRealTimeToggle = false;
+  @Input() realTimeEnabled = false;
+  @Output() realTimeEnabledChange = new EventEmitter<boolean>();
   // eslint-disable-next-line @angular-eslint/no-output-on-prefix
   @Output() onTableEvent: EventEmitter<ITableEvent> = new EventEmitter();
   // eslint-disable-next-line @angular-eslint/no-output-on-prefix

src/app/shared/modules/dynamic-material-table/table/dynamic-mat-table.component.html

@@ -2,6 +2,17 @@
   class="table-header-controls"
   [class.with-side-filter]="sideFilterCollapsed"
 >
+  <div class="real-time-toggle-wrapper" *ngIf="showRealTimeToggle">
+    <mat-slide-toggle
+      [checked]="realTimeEnabled"
+      (change)="realTimeEnabledChange.emit($event.checked)"
+      matTooltip="When on, the table watches for new datasets and shows a notice when updates arrive."
+      matTooltipPosition="above"
+      data-cy="real-time-toggle"
+    >
+      Live updates
+    </mat-slide-toggle>
+  </div>
   <div class="global-search-wrapper" *ngIf="showGlobalTextSearch">
     <mat-form-field>
       <mat-icon matPrefix style="color: gray" *ngIf="!globalTextSearch"
@@ -428,6 +439,7 @@
       [class.row-selection]="
         rowSelectionModel ? rowSelectionModel.isSelected(row) : false
       "
+      [class.row-highlight]="highlighted.has(row._id)"
       (contextmenu)="onContextMenu($event, null, row)"
     >
     </mat-row>