[Reporting] Fix job notifications poller

packages/kbn-reporting/public/job_completion_notifications.test.ts

@@ -0,0 +1,33 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { jobCompletionNotifications } from './job_completion_notifications';
+
+describe('Job completion notifications', () => {
+  const { setPendingJobIds, getPendingJobIds, addPendingJobId } = jobCompletionNotifications();
+
+  afterEach(async () => {
+    await setPendingJobIds([]);
+  });
+
+  it('initially contains not job IDs', async () => {
+    expect(await getPendingJobIds()).toEqual([]);
+  });
+
+  it('handles multiple job ID additions', async () => {
+    await addPendingJobId('job-123');
+    await addPendingJobId('job-456');
+    await addPendingJobId('job-789');
+    expect(await getPendingJobIds()).toEqual(['job-123', 'job-456', 'job-789']);
+  });
+
+  it('handles setting a total of amount of job ID', async () => {
+    await setPendingJobIds(['job-abc', 'job-def', 'job-ghi']);
+    expect(await getPendingJobIds()).toEqual(['job-abc', 'job-def', 'job-ghi']);
+  });
+});

packages/kbn-reporting/public/job_completion_notifications.ts

@@ -9,29 +9,63 @@
 import { JOB_COMPLETION_NOTIFICATIONS_SESSION_KEY } from '@kbn/reporting-common';
 import { JobId } from '@kbn/reporting-common/types';
 
-const set = (jobs: string[]) => {
-  sessionStorage.setItem(JOB_COMPLETION_NOTIFICATIONS_SESSION_KEY, JSON.stringify(jobs));
-};
-
-const getAll = (): string[] => {
-  const sessionValue = sessionStorage.getItem(JOB_COMPLETION_NOTIFICATIONS_SESSION_KEY);
-  return sessionValue ? JSON.parse(sessionValue) : [];
-};
+export function jobCompletionNotifications() {
+  // Reading and writing to the local storage must be atomic,
+  // i.e. performed in a single operation. This storage queue
+  // Operations on the localStorage key can happen from various
+  // parts of code. Using a queue to manage async operations allows
+  // operations to process one at a time
+  let operationQueue = Promise.resolve();
+  async function addToQueue(func: (error: Error | null) => void) {
+    operationQueue = operationQueue.then(() => func(null)).catch(func);
+    await operationQueue;
+  }
 
-export const add = (jobId: JobId) => {
-  const jobs = getAll();
-  jobs.push(jobId);
-  set(jobs);
-};
+  async function getPendingJobIds(): Promise<JobId[]> {
+    let jobs: JobId[] = [];
+    await addToQueue(async () => {
+      // get the current jobs
+      const jobsData = localStorage.getItem(JOB_COMPLETION_NOTIFICATIONS_SESSION_KEY);
+      jobs = jobsData ? JSON.parse(jobsData) : [];
+    });
+    return jobs;
+  }
 
-export const remove = (jobId: JobId) => {
-  const jobs = getAll();
-  const index = jobs.indexOf(jobId);
+  async function addPendingJobId(jobId: JobId) {
+    await addToQueue(async (error: Error | null) => {
+      return new Promise((resolve, reject) => {
+        if (error) {
+          window.console.error(error);
+          reject(error);
+        }
+        // get the current jobs synchronously
+        const jobsData = localStorage.getItem(JOB_COMPLETION_NOTIFICATIONS_SESSION_KEY);
+        const jobs: JobId[] = jobsData ? JSON.parse(jobsData) : [];
+        // add the new job
+        jobs.push(jobId);
+        // write back to local storage
+        localStorage.setItem(JOB_COMPLETION_NOTIFICATIONS_SESSION_KEY, JSON.stringify(jobs));
+        resolve();
+      });
+    });
+  }
 
-  if (!index) {
-    throw new Error('Unable to find job to remove it');
+  async function setPendingJobIds(jobIds: JobId[]) {
+    await addToQueue(async (error: Error | null) => {
+      return new Promise((resolve, reject) => {
+        if (error) {
+          reject(error);
+        }
+        // write update jobs back to local storage
+        localStorage.setItem(JOB_COMPLETION_NOTIFICATIONS_SESSION_KEY, JSON.stringify(jobIds));
+        resolve();
+      });
+    });
   }
 
-  jobs.splice(index, 1);
-  set(jobs);
-};
+  return {
+    getPendingJobIds,
+    addPendingJobId,
+    setPendingJobIds,
+  };
+}

packages/kbn-reporting/public/reporting_api_client.ts

@@ -11,17 +11,18 @@ import type { HttpFetchQuery } from '@kbn/core/public';
 import { HttpSetup, IUiSettingsClient } from '@kbn/core/public';
 import { i18n } from '@kbn/i18n';
 import {
+  INTERNAL_ROUTES,
+  PUBLIC_ROUTES,
   REPORTING_MANAGEMENT_HOME,
   buildKibanaPath,
   getRedirectAppPath,
-  INTERNAL_ROUTES,
-  PUBLIC_ROUTES,
 } from '@kbn/reporting-common';
 import { BaseParams, JobId, ManagementLinkFn, ReportApiJSON } from '@kbn/reporting-common/types';
 import rison from '@kbn/rison';
 import moment from 'moment';
 import { stringify } from 'query-string';
-import { Job, add } from '.';
+import { Job } from '.';
+import { jobCompletionNotifications } from './job_completion_notifications';
 
 /*
  * For convenience, apps do not have to provide the browserTimezone and Kibana version.
@@ -66,6 +67,7 @@ interface IReportingAPI {
  */
 export class ReportingAPIClient implements IReportingAPI {
   private http: HttpSetup;
+  private addPendingJobId = jobCompletionNotifications().addPendingJobId;
 
   constructor(
     http: HttpSetup,
@@ -182,7 +184,7 @@ export class ReportingAPIClient implements IReportingAPI {
         body: JSON.stringify({ jobParams: jobParamsRison }),
       }
     );
-    add(resp.job.id);
+    await this.addPendingJobId(resp.job.id);
     return new Job(resp.job);
   }