fix: delay InMemoryCollectionStore observeAll emission after setAll

observeAll implementation may not be subscribed to updates$ yet

also update WalletRepository methods to take(1) because setAll makes
source emit again

Author: mkazlauskas

packages/wallet/src/persistence/inMemoryStores/InMemoryCollectionStore.ts

@@ -1,6 +1,6 @@
 /* eslint-disable brace-style */
 import { CollectionStore } from '../types';
-import { EMPTY, Observable, Subject, of } from 'rxjs';
+import { EMPTY, Observable, Subject, delay, of, tap } from 'rxjs';
 import { InMemoryStore } from './InMemoryStore';
 import { observeAll } from '../util';
 
@@ -22,8 +22,10 @@ export class InMemoryCollectionStore<T> extends InMemoryStore implements Collect
   setAll(docs: T[]): Observable<void> {
     if (!this.destroyed) {
       this.docs = docs;
-      this.#updates$.next(this.docs);
-      return of(void 0);
+      return of(void 0).pipe(
+        delay(1),
+        tap(() => this.#updates$.next(this.docs))
+      );
     }
     return EMPTY;
   }

packages/wallet/test/persistence/inMemoryStores.test.ts

@@ -18,7 +18,7 @@ import {
   InMemoryUtxoStore
 } from '../../src/persistence';
 import { assertCompletesWithoutEmitting } from './util';
-import { firstValueFrom, share, take, toArray } from 'rxjs';
+import { firstValueFrom, mergeMap, share, shareReplay, take, toArray } from 'rxjs';
 
 describe('inMemoryStores', () => {
   describe('InMemoryDocumentStore', () => {
@@ -75,6 +75,12 @@ describe('inMemoryStores', () => {
         await firstValueFrom(store.setAll(updatedDocs));
         await expect(twoEmissions).resolves.toEqual([docs, updatedDocs]);
       });
+      it('observeAll followed by immediate setAll does not skip 2nd emission', async () => {
+        const store = new InMemoryCollectionStore();
+        const items$ = store.observeAll().pipe(shareReplay(1));
+        await firstValueFrom(items$.pipe(mergeMap(() => store.setAll(docs))));
+        await expect(firstValueFrom(items$)).resolves.toEqual(docs);
+      });
     });
   });
 

packages/wallet/test/persistence/pouchDbStores.test.ts

@@ -1,7 +1,7 @@
 /* eslint-disable unicorn/consistent-function-scoping */
 import { PouchDbCollectionStore, PouchDbDocumentStore, PouchDbKeyValueStore } from '../../src/persistence';
 import { assertCompletesWithoutEmitting } from './util';
-import { combineLatest, firstValueFrom, mergeMap, share, take, timer, toArray } from 'rxjs';
+import { combineLatest, firstValueFrom, mergeMap, share, shareReplay, take, timer, toArray } from 'rxjs';
 import { dummyLogger as logger } from 'ts-log';
 import PouchDB from 'pouchdb';
 
@@ -115,6 +115,11 @@ describe('pouchDbStores', () => {
         await firstValueFrom(store1.setAll([doc1, doc2]));
         await expect(twoEmissions).resolves.toEqual([[doc1], [doc1, doc2]]);
       });
+      it('observeAll followed by immediate setAll does not skip 2nd emission', async () => {
+        const items$ = store1.observeAll().pipe(shareReplay(1));
+        await firstValueFrom(items$.pipe(mergeMap(() => store1.setAll([doc1]))));
+        await expect(firstValueFrom(items$)).resolves.toEqual([doc1]);
+      });
     });
   });
 

packages/web-extension/src/walletManager/WalletRepository/WalletRepository.ts

@@ -2,7 +2,7 @@ import { AddAccountProps, AddWalletProps, RemoveAccountProps, UpdateMetadataProp
 import { AnyWallet, ScriptWallet, WalletId, WalletType } from '../types';
 import { Bip32PublicKey, Hash28ByteBase16 } from '@cardano-sdk/crypto';
 import { Logger } from 'ts-log';
-import { Observable, defer, firstValueFrom, map, shareReplay, switchMap } from 'rxjs';
+import { Observable, defer, firstValueFrom, map, shareReplay, switchMap, take } from 'rxjs';
 import { Serialization } from '@cardano-sdk/core';
 import { WalletConflictError } from '../errors';
 import { contextLogger } from '@cardano-sdk/util';
@@ -48,14 +48,22 @@ export class WalletRepository<AccountMetadata extends {}> implements WalletRepos
     this.wallets$ = defer(() => store.observeAll()).pipe(shareReplay(1));
   }
 
+  #getWallets() {
+    return this.wallets$.pipe(
+      // `setAll` makes the store.observeAll source emit
+      // so the pipes are triggered twice otherwise
+      take(1)
+    );
+  }
+
   async addWallet(props: AddWalletProps<AccountMetadata>): Promise<WalletId> {
     this.#logger.debug('addWallet', props.type);
     const walletId =
       props.type === WalletType.Script
         ? Serialization.Script.fromCore(props.script).hash()
         : Hash28ByteBase16(await Bip32PublicKey.fromHex(props.extendedAccountPublicKey).hash());
     return firstValueFrom(
-      this.wallets$.pipe(
+      this.#getWallets().pipe(
         switchMap((wallets) => {
           if (wallets.some((wallet) => wallet.walletId === walletId)) {
             throw new WalletConflictError(`Wallet '${walletId}' already exists`);
@@ -90,7 +98,7 @@ export class WalletRepository<AccountMetadata extends {}> implements WalletRepos
     const { walletId, accountIndex, metadata } = props;
     this.#logger.debug('addAccount', walletId, accountIndex, metadata);
     return firstValueFrom(
-      this.wallets$.pipe(
+      this.#getWallets().pipe(
         switchMap((wallets) => {
           const walletIndex = wallets.findIndex((w) => w.walletId === walletId);
           if (walletIndex < 0) {
@@ -126,7 +134,7 @@ export class WalletRepository<AccountMetadata extends {}> implements WalletRepos
     const { walletId, accountIndex, metadata } = props;
     this.#logger.debug('updateMetadata', walletId, accountIndex, metadata);
     return firstValueFrom(
-      this.wallets$.pipe(
+      this.#getWallets().pipe(
         switchMap((wallets) => {
           if (typeof accountIndex !== 'undefined') {
             const bip32Account = findAccount(wallets, walletId, accountIndex);
@@ -165,7 +173,7 @@ export class WalletRepository<AccountMetadata extends {}> implements WalletRepos
     const { walletId, accountIndex } = props;
     this.#logger.debug('removeAccount', walletId, accountIndex);
     return firstValueFrom(
-      this.wallets$.pipe(
+      this.#getWallets().pipe(
         switchMap((wallets) => {
           const bip32Account = findAccount(wallets, walletId, accountIndex);
           if (!bip32Account) {
@@ -196,7 +204,8 @@ export class WalletRepository<AccountMetadata extends {}> implements WalletRepos
   removeWallet(walletId: WalletId): Promise<WalletId> {
     this.#logger.debug('removeWallet', walletId);
     return firstValueFrom(
-      this.wallets$.pipe(
+      this.#getWallets().pipe(
+        take(1),
         switchMap((wallets) => {
           const walletIndex = wallets.findIndex((w) => w.walletId === walletId);
           if (walletIndex < 0) {