[PM-18903] Desktop sync issues (#13681)

* [PM-18707] Use different BroadcasterSubscriptionId in base view component to avoid collision with desktop view component

* [PM-18707] Use userId instead of payloadUserId for cipher notification syncs

* [PM-19032] Live Sync on Desktop (#13851)

* migrate the vault-items to an observables rather than async/promises

- this helps keep data in sync with the service state and avoids race conditions

* migrate the view component to an observables rather than async/promises

- this helps keep data in sync with the service state and avoids race conditions

* decrypt saved cipher from server

* bump timeout for upserting ciphers

* mark `go` as async in desktop vault

- previously it was a floating promise

* Revert "mark `go` as async in desktop vault"

This reverts commit fd28f40b18.

* Revert "bump timeout for upserting ciphers"

This reverts commit e963acc377.

* move vault utilities to `common` rather than `lib` to avoid circular dependencies

* use `perUserCache$` for `cipherViews$` to avoid new subscriptions from being created

* use userId from observable rather than locally set to be the most up to date

* [PM-18707] Add clearBuffer$ input to perUserCache$ helper so that  the internal share replay buffers can be cleared

* [PM-18707] Rework forceCipherViews$ to clearBuffer$ refactor

- Add dependency for cipherDecryptionKeys$ for the cipherViews so that decryption is never attempted without keys

* [PM-18707] Add overload to perUserCache to satisfy type checker

* [PM-18707] Fix overloads

* [PM-18707] Add check for empty failed to decrypt ciphers

* [PM-18707] Mark vault component for check after observable emits.

The cipherViews$ observable now persists between subscriptions, meaning that updates via the sync push notifications can occur outside the AngularZone causing delays in updating the view.

---------

Co-authored-by: Nick Krantz <125900171+nick-livefront@users.noreply.github.com>
Co-authored-by: Nick Krantz <nick@livefront.com>

libs/common/src/platform/notifications/internal/default-notifications.service.ts

@@ -153,14 +153,14 @@ export class DefaultNotificationsService implements NotificationsServiceAbstract
         await this.syncService.syncUpsertCipher(
           notification.payload as SyncCipherNotification,
           notification.type === NotificationType.SyncCipherUpdate,
-          payloadUserId,
+          userId,
         );
         break;
       case NotificationType.SyncCipherDelete:
       case NotificationType.SyncLoginDelete:
         await this.syncService.syncDeleteCipher(
           notification.payload as SyncCipherNotification,
-          payloadUserId,
+          userId,
         );
         break;
       case NotificationType.SyncFolderCreate:

libs/common/src/vault/services/cipher.service.ts

@@ -1,17 +1,6 @@
 // FIXME: Update this file to be type safe and remove this and next line
 // @ts-strict-ignore
-import {
-  combineLatest,
-  filter,
-  firstValueFrom,
-  map,
-  merge,
-  Observable,
-  of,
-  shareReplay,
-  Subject,
-  switchMap,
-} from "rxjs";
+import { combineLatest, filter, firstValueFrom, map, Observable, Subject, switchMap } from "rxjs";
 import { SemVer } from "semver";
 
 import { LogService } from "@bitwarden/common/platform/abstractions/log.service";
@@ -40,6 +29,7 @@ import { SymmetricCryptoKey } from "../../platform/models/domain/symmetric-crypt
 import { StateProvider } from "../../platform/state";
 import { CipherId, CollectionId, OrganizationId, UserId } from "../../types/guid";
 import { OrgKey, UserKey } from "../../types/key";
+import { perUserCache$ } from "../../vault/utils/observable-utilities";
 import { CipherService as CipherServiceAbstraction } from "../abstractions/cipher.service";
 import { CipherFileUploadService } from "../abstractions/file-upload/cipher-file-upload.service";
 import { FieldType } from "../enums";
@@ -91,11 +81,12 @@ export class CipherService implements CipherServiceAbstraction {
     this.sortCiphersByLastUsed,
   );
   /**
-   * Observable that forces the `cipherViews$` observable to re-emit with the provided value.
-   * Used to let subscribers of `cipherViews$` know that the decrypted ciphers have been cleared for the active user.
+   * Observable that forces the `cipherViews$` observable for the given user to emit a null value.
+   * Used to let subscribers of `cipherViews$` know that the decrypted ciphers have been cleared for the user and to
+   * clear them from the shareReplay buffer created in perUserCache$().
    * @private
    */
-  private forceCipherViews$: Subject<CipherView[]> = new Subject<CipherView[]>();
+  private clearCipherViewsForUser$: Subject<UserId> = new Subject<UserId>();
 
   constructor(
     private keyService: KeyService,
@@ -132,13 +123,16 @@ export class CipherService implements CipherServiceAbstraction {
    * A `null` value indicates that the latest encrypted ciphers have not been decrypted yet and that
    * decryption is in progress. The latest decrypted ciphers will be emitted once decryption is complete.
    */
-  cipherViews$(userId: UserId): Observable<CipherView[] | null> {
-    return combineLatest([this.encryptedCiphersState(userId).state$, this.localData$(userId)]).pipe(
-      filter(([ciphers]) => ciphers != null), // Skip if ciphers haven't been loaded yor synced yet
-      switchMap(() => merge(this.forceCipherViews$, this.getAllDecrypted(userId))),
-      shareReplay({ bufferSize: 1, refCount: true }),
+  cipherViews$ = perUserCache$((userId: UserId): Observable<CipherView[] | null> => {
+    return combineLatest([
+      this.encryptedCiphersState(userId).state$,
+      this.localData$(userId),
+      this.keyService.cipherDecryptionKeys$(userId, true),
+    ]).pipe(
+      filter(([ciphers, keys]) => ciphers != null && keys != null), // Skip if ciphers haven't been loaded yor synced yet
+      switchMap(() => this.getAllDecrypted(userId)),
     );
-  }
+  }, this.clearCipherViewsForUser$);
 
   addEditCipherInfo$(userId: UserId): Observable<AddEditCipherInfo> {
     return this.addEditCipherInfoState(userId).state$;
@@ -149,13 +143,11 @@ export class CipherService implements CipherServiceAbstraction {
    *
    * An empty array indicates that all ciphers were successfully decrypted.
    */
-  failedToDecryptCiphers$(userId: UserId): Observable<CipherView[]> {
+  failedToDecryptCiphers$ = perUserCache$((userId: UserId): Observable<CipherView[]> => {
     return this.failedToDecryptCiphersState(userId).state$.pipe(
       filter((ciphers) => ciphers != null),
-      switchMap((ciphers) => merge(this.forceCipherViews$, of(ciphers))),
-      shareReplay({ bufferSize: 1, refCount: true }),
     );
-  }
+  }, this.clearCipherViewsForUser$);
 
   async setDecryptedCipherCache(value: CipherView[], userId: UserId) {
     // Sometimes we might prematurely decrypt the vault and that will result in no ciphers
@@ -190,10 +182,8 @@ export class CipherService implements CipherServiceAbstraction {
     userId ??= activeUserId;
     await this.clearDecryptedCiphersState(userId);
 
-    // Force the cipherView$ observable (which always tracks the active user) to re-emit
-    if (userId == activeUserId) {
-      this.forceCipherViews$.next(null);
-    }
+    // Force the cached cipherView$ observable(s) to emit a null value
+    this.clearCipherViewsForUser$.next(userId);
   }
 
   async encrypt(
@@ -402,10 +392,14 @@ export class CipherService implements CipherServiceAbstraction {
       return await this.getDecryptedCiphers(userId);
     }
 
-    const [newDecCiphers, failedCiphers] = await this.decryptCiphers(
-      await this.getAll(userId),
-      userId,
-    );
+    const decrypted = await this.decryptCiphers(await this.getAll(userId), userId);
+
+    // We failed to decrypt, return empty array but do not cache
+    if (decrypted == null) {
+      return [];
+    }
+
+    const [newDecCiphers, failedCiphers] = decrypted;
 
     await this.setDecryptedCipherCache(newDecCiphers, userId);
     await this.setFailedDecryptedCiphers(failedCiphers, userId);
@@ -429,12 +423,12 @@ export class CipherService implements CipherServiceAbstraction {
   private async decryptCiphers(
     ciphers: Cipher[],
     userId: UserId,
-  ): Promise<[CipherView[], CipherView[]]> {
+  ): Promise<[CipherView[], CipherView[]] | null> {
     const keys = await firstValueFrom(this.keyService.cipherDecryptionKeys$(userId, true));
 
     if (keys == null || (keys.userKey == null && Object.keys(keys.orgKeys).length === 0)) {
       // return early if there are no keys to decrypt with
-      return [[], []];
+      return null;
     }
 
     // Group ciphers by orgId or under 'null' for the user's ciphers

libs/common/src/vault/tasks/services/default-task.service.ts

@@ -12,8 +12,11 @@ import { MessageListener } from "@bitwarden/common/platform/messaging";
 import { NotificationsService } from "@bitwarden/common/platform/notifications";
 import { StateProvider } from "@bitwarden/common/platform/state";
 import { SecurityTaskId, UserId } from "@bitwarden/common/types/guid";
+import {
+  filterOutNullish,
+  perUserCache$,
+} from "@bitwarden/common/vault/utils/observable-utilities";
 
-import { filterOutNullish, perUserCache$ } from "../../utils/observable-utilities";
 import { TaskService } from "../abstractions/task.service";
 import { SecurityTaskStatus } from "../enums";
 import { SecurityTask, SecurityTaskData, SecurityTaskResponse } from "../models";

libs/common/src/vault/utils/observable-utilities.ts

@@ -1,20 +1,38 @@
-import { filter, Observable, OperatorFunction, shareReplay } from "rxjs";
+import { EMPTY, filter, map, merge, Observable, OperatorFunction, shareReplay } from "rxjs";
 
 import { UserId } from "@bitwarden/common/types/guid";
 
 /**
  * Builds an observable once per userId and caches it for future requests.
  * The built observables are shared among subscribers with a replay buffer size of 1.
+ *
+ * Optionally, a clearBuffer$ observable can be provided to clear the replay buffer for a specific or all userIds.
  * @param create - A function that creates an observable for a given userId.
+ * @param clearBuffer$ - An observable that, when emitted, clears the buffer for the emitted userId. When null is emitted, all caches are cleared.
  */
 export function perUserCache$<TValue>(
   create: (userId: UserId) => Observable<TValue>,
-): (userId: UserId) => Observable<TValue> {
-  const cache = new Map<UserId, Observable<TValue>>();
+  clearBuffer$: Observable<UserId | null>,
+): (userId: UserId) => Observable<TValue | null>;
+export function perUserCache$<TValue>(
+  create: (userId: UserId) => Observable<TValue>,
+): (userId: UserId) => Observable<TValue>;
+export function perUserCache$<TValue>(
+  create: (userId: UserId) => Observable<TValue>,
+  clearBuffer$: Observable<UserId | null> | undefined = undefined,
+): (userId: UserId) => Observable<TValue | null> {
+  const cache = new Map<UserId, Observable<TValue | null>>();
   return (userId: UserId) => {
     let observable = cache.get(userId);
     if (!observable) {
-      observable = create(userId).pipe(shareReplay({ bufferSize: 1, refCount: false }));
+      clearBuffer$ ??= EMPTY;
+      observable = merge(
+        create(userId),
+        clearBuffer$.pipe(
+          filter((clearId) => clearId === userId || clearId === null),
+          map(() => null),
+        ),
+      ).pipe(shareReplay({ bufferSize: 1, refCount: false }));
       cache.set(userId, observable);
     }
     return observable;