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

src/app/admin/schema/frontend.config.jsonforms.json

@@ -24,6 +24,7 @@
       "accessTokenPrefix": { "type": "string" },
       "addDatasetEnabled": { "type": "boolean" },
       "archiveWorkflowEnabled": { "type": "boolean" },
+      "realTimeUpdatesEnabled": { "type": "boolean" },
       "datasetReduceEnabled": { "type": "boolean" },
       "datasetRelationshipsEnabled": { "type": "boolean" },
       "datasetJsonScientificMetadata": { "type": "boolean" },
@@ -520,7 +521,8 @@
           { "type": "Control", "scope": "#/properties/logbookEnabled" },
           { "type": "Control", "scope": "#/properties/loginFormEnabled" },
           { "type": "Control", "scope": "#/properties/metadataPreviewEnabled" },
-          { "type": "Control", "scope": "#/properties/autoApplyFilters" }
+          { "type": "Control", "scope": "#/properties/autoApplyFilters" },
+          { "type": "Control", "scope": "#/properties/realTimeUpdatesEnabled" }
         ]
       },
       {

src/app/app-config.service.ts

@@ -209,6 +209,7 @@ export interface AppConfigInterface {
   aboutSettings?: AboutSettings;
   batchActionsEnabled?: boolean;
   batchActions?: ActionConfig[];
+  realTimeUpdatesEnabled?: boolean;
 }
 
 function isMainPageConfiguration(obj: any): obj is MainPageConfiguration {

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

@@ -55,6 +55,7 @@ export class DashboardComponent implements OnInit, OnDestroy {
   private readyToFetch$ = this.store
     .select(selectHasPrefilledFilters)
     .pipe(filter((has) => has));
+
   loggedIn$ = this.store.select(selectIsLoggedIn);
   selectColumns$ = this.store.select(selectColumns);
   selectHasFetchedSettings$ = this.store.select(selectHasFetchedSettings);

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,45 +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,
-        isLoading: data.isLoading,
-      };
-
-      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(
@@ -192,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) {
@@ -270,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,76 @@
+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 {
+  BehaviorSubject,
+  distinctUntilChanged,
+  EMPTY,
+  map,
+  Observable,
+  Subject,
+  Subscription,
+  switchMap,
+} from "rxjs";
+
+@Injectable({ providedIn: "root" })
+export class EventsService {
+  private connectionSub: Subscription | null = null;
+  private messageSubject = new Subject<Record<string, unknown>>();
+  private connectionErrorSubject = new BehaviorSubject<boolean>(false);
+
+  connectionError$ = this.connectionErrorSubject.asObservable();
+
+  message$ = this.messageSubject.asObservable();
+
+  latestUpdatedId$ = this.messageSubject.pipe(
+    map((m) => (m["data"] as { _id: string })._id),
+  );
+
+  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.onopen = () =>
+        this.ngZone.run(() => this.connectionErrorSubject.next(false));
+
+      es.onmessage = (event) => {
+        this.ngZone.run(() => observer.next(JSON.parse(event.data)));
+      };
+
+      es.onerror = () => {
+        es.close();
+        this.ngZone.run(() => this.connectionErrorSubject.next(true));
+      };
+
+      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) => {
+        return this.messageSubject.next(msg);
+      });
+  }
+
+  disconnect() {
+    this.connectionSub?.unsubscribe();
+    this.connectionSub = 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,29 @@
   class="table-header-controls"
   [class.with-side-filter]="sideFilterCollapsed"
 >
+  <div
+    class="real-time-toggle-wrapper"
+    *ngIf="showRealTimeToggle && appConfig?.realTimeUpdatesEnabled"
+  >
+    <mat-slide-toggle
+      [checked]="realTimeEnabled"
+      (change)="realTimeEnabledChange.emit($event.checked)"
+      data-cy="real-time-toggle"
+    >
+      Live updates
+    </mat-slide-toggle>
+
+    <mat-icon
+      *ngIf="realTimeEnabled && eventsService.connectionError$ | async"
+      class="connection-error-icon"
+      matTooltip="Live updates are currently unavailable. Please check with your administrator."
+      matTooltipPosition="above"
+      aria-label="Live updates are currently unavailable. Please check with your administrator."
+      data-cy="live-updates-error"
+    >
+      cloud_off
+    </mat-icon>
+  </div>
   <div class="global-search-wrapper" *ngIf="showGlobalTextSearch">
     <mat-form-field>
       <mat-icon matPrefix style="color: gray" *ngIf="!globalTextSearch"
@@ -429,6 +452,7 @@
       [class.row-selection]="
         rowSelectionModel ? rowSelectionModel.isSelected(row) : false
       "
+      [class.row-highlight]="highlighted.has(row._id)"
       (contextmenu)="onContextMenu($event, null, row)"
     >
     </mat-row>