[PM-10381] Fix waiting for last sync - Browser Refresh (#10438)

* [PM-10381] Add activeUserLastSync$ to SyncService

* [PM-10381] Introduce waitUtil operator

* [PM-10381] Use new activeUserLastSync$ observable to wait until a sync completes before attempting to get decrypted ciphers

* [PM-10381] Fix failing test

---------

Co-authored-by: bnagawiecki <[email protected]>

apps/browser/src/vault/popup/services/vault-popup-items.service.spec.ts

@@ -30,6 +30,7 @@ describe("VaultPopupItemsService", () => {
 
   let mockOrg: Organization;
   let mockCollections: CollectionView[];
+  let activeUserLastSync$: BehaviorSubject<Date>;
 
   const cipherServiceMock = mock<CipherService>();
   const vaultSettingsServiceMock = mock<VaultSettingsService>();
@@ -92,7 +93,8 @@ describe("VaultPopupItemsService", () => {
     organizationServiceMock.organizations$ = new BehaviorSubject([mockOrg]);
     collectionService.decryptedCollections$ = new BehaviorSubject(mockCollections);
 
-    syncServiceMock.getLastSync.mockResolvedValue(new Date());
+    activeUserLastSync$ = new BehaviorSubject(new Date());
+    syncServiceMock.activeUserLastSync$.mockReturnValue(activeUserLastSync$);
 
     testBed = TestBed.configureTestingModule({
       providers: [
@@ -161,7 +163,7 @@ describe("VaultPopupItemsService", () => {
   });
 
   it("should not emit cipher list if syncService.getLastSync returns null", async () => {
-    syncServiceMock.getLastSync.mockResolvedValue(null);
+    activeUserLastSync$.next(null);
 
     const obs$ = service.autoFillCiphers$.pipe(timeout(50));
 

apps/browser/src/vault/popup/services/vault-popup-items.service.ts

@@ -9,6 +9,7 @@ import {
   from,
   map,
   merge,
+  MonoTypeOperatorFunction,
   Observable,
   of,
   shareReplay,
@@ -31,6 +32,7 @@ import { CipherType } from "@bitwarden/common/vault/enums";
 import { CipherView } from "@bitwarden/common/vault/models/view/cipher.view";
 
 import { runInsideAngular } from "../../../platform/browser/run-inside-angular.operator";
+import { waitUntil } from "../../util";
 import { PopupCipherView } from "../views/popup-cipher.view";
 
 import { VaultPopupAutofillService } from "./vault-popup-autofill.service";
@@ -80,8 +82,7 @@ export class VaultPopupItemsService {
   ).pipe(
     runInsideAngular(inject(NgZone)), // Workaround to ensure cipher$ state provider emissions are run inside Angular
     tap(() => this._ciphersLoading$.next()),
-    switchMap(() => Utils.asyncToObservable(() => this.syncService.getLastSync())),
-    filter((lastSync) => lastSync !== null), // Only attempt to load ciphers if we performed a sync
+    waitUntilSync(this.syncService),
     switchMap(() => Utils.asyncToObservable(() => this.cipherService.getAllDecrypted())),
     switchMap((ciphers) =>
       combineLatest([
@@ -270,3 +271,11 @@ export class VaultPopupItemsService {
     return this.cipherService.sortCiphersByLastUsedThenName(a, b);
   }
 }
+
+/**
+ * Operator that waits until the active account has synced at least once before allowing the source to continue emission.
+ * @param syncService
+ */
+const waitUntilSync = <T>(syncService: SyncService): MonoTypeOperatorFunction<T> => {
+  return waitUntil(syncService.activeUserLastSync$().pipe(filter((lastSync) => lastSync != null)));
+};

apps/browser/src/vault/util.ts

@@ -0,0 +1,30 @@
+import {
+  merge,
+  MonoTypeOperatorFunction,
+  Observable,
+  ObservableInput,
+  sample,
+  share,
+  skipUntil,
+  take,
+} from "rxjs";
+
+/**
+ * Operator that waits until the trigger observable emits before allowing the source to continue emission.
+ * @param trigger$ The observable that will trigger the source to continue emission.
+ *
+ * ```
+ * source$  a-----b-----c-----d-----e
+ * trigger$ ---------------X---------
+ * output$  ---------------c--d-----e
+ * ```
+ */
+export const waitUntil = <T>(trigger$: ObservableInput<any>): MonoTypeOperatorFunction<T> => {
+  return (source: Observable<T>) => {
+    const sharedSource$ = source.pipe(share());
+    return merge(
+      sharedSource$.pipe(sample(trigger$), take(1)),
+      sharedSource$.pipe(skipUntil(trigger$)),
+    );
+  };
+};

libs/common/src/platform/sync/core-sync.service.ts

@@ -1,4 +1,4 @@
-import { firstValueFrom, map, of, switchMap } from "rxjs";
+import { firstValueFrom, map, Observable, of, switchMap } from "rxjs";
 
 import { ApiService } from "../../abstractions/api.service";
 import { AccountService } from "../../auth/abstractions/account.service";
@@ -67,6 +67,17 @@ export abstract class CoreSyncService implements SyncService {
     return this.stateProvider.getUser(userId, LAST_SYNC_DATE).state$;
   }
 
+  activeUserLastSync$(): Observable<Date | null> {
+    return this.accountService.activeAccount$.pipe(
+      switchMap((a) => {
+        if (a == null) {
+          return of(null);
+        }
+        return this.lastSync$(a.id);
+      }),
+    );
+  }
+
   async setLastSync(date: Date, userId: UserId): Promise<void> {
     await this.stateProvider.getUser(userId, LAST_SYNC_DATE).update(() => date);
   }

libs/common/src/platform/sync/sync.service.ts

@@ -34,6 +34,12 @@ export abstract class SyncService {
    */
   abstract lastSync$(userId: UserId): Observable<Date | null>;
 
+  /**
+   * Retrieves a stream of the currently active user's last sync date.
+   * Or null if there is no current active user or the active user has not synced before.
+   */
+  abstract activeUserLastSync$(): Observable<Date | null>;
+
   /**
    * Optionally does a full sync operation including going to the server to gather the source
    * of truth and set that data to state.