feat(wallet): add signed transactions observable

Author: Lucas Araujo

packages/wallet/src/PersonalWallet/PersonalWallet.ts

@@ -199,8 +199,10 @@ export class PersonalWallet implements ObservableWallet {
   #newTransactions = {
     failedToSubmit$: new Subject<FailedTx>(),
     pending$: new Subject<OutgoingTx>(),
+    signed$: new Subject<SignedTx>(),
     submitting$: new Subject<OutgoingTx>()
   };
+  #removeSignedTx$: Subject<Cardano.TransactionId>;
   #reemitSubscriptions: Subscription;
   #failedFromReemitter$: Subject<FailedTx>;
   #trackedTxSubmitProvider: TrackedTxSubmitProvider;
@@ -403,6 +405,7 @@ export class PersonalWallet implements ObservableWallet {
       map((addresses) => addresses.map((groupedAddress) => groupedAddress.address))
     );
     this.#failedFromReemitter$ = new Subject<FailedTx>();
+    this.#removeSignedTx$ = new Subject<Cardano.TransactionId>();
     this.transactions = createTransactionsTracker({
       addresses$,
       chainHistoryProvider: this.chainHistoryProvider,
@@ -411,6 +414,7 @@ export class PersonalWallet implements ObservableWallet {
       logger: contextLogger(this.#logger, 'transactions'),
       newTransactions: this.#newTransactions,
       onFatalError,
+      removeSignedTx$: this.#removeSignedTx$,
       retryBackoffConfig,
       tip$: this.tip$,
       transactionsHistoryStore: stores.transactions
@@ -444,6 +448,7 @@ export class PersonalWallet implements ObservableWallet {
       logger: contextLogger(this.#logger, 'utxo'),
       onFatalError,
       retryBackoffConfig,
+      signedTransactions$: this.transactions.outgoing.signed$,
       stores,
       tipBlockHeight$,
       transactionsInFlight$: this.transactions.outgoing.inFlight$,
@@ -535,7 +540,7 @@ export class PersonalWallet implements ObservableWallet {
 
   async finalizeTx({ tx, sender, ...rest }: FinalizeTxProps, stubSign = false): Promise<Cardano.Tx> {
     const knownAddresses = await firstValueFrom(this.addresses$);
-    const { tx: signedTx } = await finalizeTx(
+    const result = await finalizeTx(
       tx,
       {
         ...rest,
@@ -548,7 +553,8 @@ export class PersonalWallet implements ObservableWallet {
       { bip32Account: this.bip32Account, witnesser: this.witnesser },
       stubSign
     );
-    return signedTx;
+    this.#newTransactions.signed$.next(result);
+    return result.tx;
   }
 
   private initializeHandles(handlePolicyIds$: Observable<Cardano.PolicyId[]>): Observable<HandleInfo[]> {
@@ -748,4 +754,8 @@ export class PersonalWallet implements ObservableWallet {
     await firstValueFrom(this.#addressTracker.addAddresses(newAddresses));
     return firstValueFrom(this.addresses$);
   }
+
+  removeSignedTx(id: Cardano.TransactionId): void {
+    this.#removeSignedTx$.next(id);
+  }
 }

packages/wallet/src/services/TransactionsTracker.ts

@@ -7,6 +7,7 @@ import {
   Subject,
   combineLatest,
   concat,
+  concatMap,
   defaultIfEmpty,
   exhaustMap,
   filter,
@@ -33,8 +34,9 @@ import { Logger } from 'ts-log';
 import { Range, Shutdown, contextLogger } from '@cardano-sdk/util';
 import { RetryBackoffConfig } from 'backoff-rxjs';
 import { TrackerSubject, coldObservableProvider } from '@cardano-sdk/util-rxjs';
-import { distinctBlock, transactionsEquals } from './util';
+import { distinctBlock, transactionsEquals, txInEquals } from './util';
 
+import { SignedTx } from '@cardano-sdk/tx-construction';
 import chunk from 'lodash/chunk';
 import intersectionBy from 'lodash/intersectionBy';
 import sortBy from 'lodash/sortBy';
@@ -51,9 +53,11 @@ export interface TransactionsTrackerProps {
     submitting$: Observable<OutgoingTx>;
     pending$: Observable<OutgoingTx>;
     failedToSubmit$: Observable<FailedTx>;
+    signed$: Observable<SignedTx>;
   };
   failedFromReemitter$?: Observable<FailedTx>;
   logger: Logger;
+  removeSignedTx$: Subject<Cardano.TransactionId>;
   onFatalError?: (value: unknown) => void;
 }
 
@@ -217,12 +221,13 @@ export const createTransactionsTracker = (
     tip$,
     chainHistoryProvider,
     addresses$,
-    newTransactions: { submitting$: newSubmitting$, pending$, failedToSubmit$ },
+    newTransactions: { submitting$: newSubmitting$, pending$, signed$: newSigned$, failedToSubmit$ },
     retryBackoffConfig,
     transactionsHistoryStore: transactionsStore,
     inFlightTransactionsStore: newTransactionsStore,
     logger,
     failedFromReemitter$,
+    removeSignedTx$,
     onFatalError
   }: TransactionsTrackerProps,
   { transactionsSource$: txSource$, rollback$ }: TransactionsTrackerInternals = createAddressTransactionsProvider({
@@ -382,6 +387,53 @@ export const createTransactionsTracker = (
     )
   );
 
+  const signed$ = new TrackerSubject<SignedTx[]>(
+    merge(
+      newSigned$.pipe(map((signedTx) => ({ op: 'add' as const, signedTx }))),
+      removeSignedTx$.pipe(map((id) => ({ id, op: 'remove' as const }))),
+      from(inFlight$).pipe(
+        concatMap((txs) => from(txs)),
+        map((tx) => ({ id: tx.id, op: 'remove' as const }))
+      ),
+      historicalTransactions$.pipe(
+        map((txs) => ({
+          inputs: txs.flatMap((tx) => tx.body.inputs.map((inputs) => inputs)),
+          op: 'check_inputs' as const
+        }))
+      ),
+      tip$.pipe(map(({ slot }) => ({ op: 'check_interval' as const, slot })))
+    ).pipe(
+      scan((signed, action) => {
+        if (action.op === 'add') {
+          return [...signed, action.signedTx];
+        }
+        if (action.op === 'remove') {
+          return signed.filter(({ tx }) => tx.id !== action.id);
+        }
+        if (action.op === 'check_interval') {
+          return signed.filter(
+            ({
+              tx: {
+                body: { validityInterval: { invalidHereafter } = {} }
+              }
+            }) => invalidHereafter && invalidHereafter > action.slot
+          );
+        }
+        if (action.op === 'check_inputs') {
+          return signed.filter(({ tx }) => {
+            const anyUtxoIsUsed = tx.body.inputs.some((signedTxInput) =>
+              action.inputs.some((historicalInput) => txInEquals(signedTxInput, historicalInput))
+            );
+
+            return !anyUtxoIsUsed;
+          });
+        }
+        return signed;
+      }, [] as SignedTx[]),
+      startWith([])
+    )
+  );
+
   const onChain$ = new Subject<OutgoingOnChainTx>();
   const onChainSubscription = submittingOrPreviouslySubmitted$
     .pipe(mergeMap((group$) => group$.pipe(switchMap((tx) => txOnChain$(tx).pipe(takeUntil(txFailed$(tx)))))))
@@ -394,6 +446,7 @@ export const createTransactionsTracker = (
       inFlight$,
       onChain$,
       pending$,
+      signed$,
       submitting$
     },
     rollback$,

packages/wallet/src/services/UtxoTracker.ts

@@ -3,6 +3,7 @@ import { Logger } from 'ts-log';
 import { NEVER, Observable, combineLatest, concat, distinctUntilChanged, map, of, switchMap } from 'rxjs';
 import { PersistentCollectionTrackerSubject, txInEquals, utxoEquals } from './util';
 import { RetryBackoffConfig } from 'backoff-rxjs';
+import { SignedTx } from '@cardano-sdk/tx-construction';
 import { TxInFlight, UtxoTracker } from './types';
 import { WalletStores } from '../persistence';
 import { coldObservableProvider } from '@cardano-sdk/util-rxjs';
@@ -17,6 +18,7 @@ export interface UtxoTrackerProps {
   addresses$: Observable<Cardano.PaymentAddress[]>;
   stores: Pick<WalletStores, 'utxo' | 'unspendableUtxo'>;
   transactionsInFlight$: Observable<TxInFlight[]>;
+  signedTransactions$: Observable<SignedTx[]>;
   tipBlockHeight$: Observable<Cardano.BlockNo>;
   retryBackoffConfig: RetryBackoffConfig;
   logger: Logger;
@@ -65,6 +67,7 @@ export const createUtxoTracker = (
     transactionsInFlight$,
     retryBackoffConfig,
     tipBlockHeight$,
+    signedTransactions$,
     logger,
     onFatalError
   }: UtxoTrackerProps,
@@ -79,15 +82,24 @@ export const createUtxoTracker = (
     )
   }: UtxoTrackerInternals = {}
 ): UtxoTracker => {
-  const total$ = combineLatest([utxoSource$, transactionsInFlight$, addresses$]).pipe(
-    map(([onChainUtxo, transactionsInFlight, ownAddresses]) => [
+  const total$ = combineLatest([utxoSource$, transactionsInFlight$, signedTransactions$, addresses$]).pipe(
+    map(([onChainUtxo, transactionsInFlight, signedTransactions, ownAddresses]) => [
       ...onChainUtxo.filter(([utxoTxIn]) => {
         const utxoIsUsedInFlight = transactionsInFlight.some(({ body: { inputs } }) =>
-          inputs.some((input) => input.txId === utxoTxIn.txId && input.index === utxoTxIn.index)
+          inputs.some((input) => txInEquals(input, utxoTxIn))
         );
-        utxoIsUsedInFlight &&
-          logger.debug('OnChain UTXO is already used in in-flight transaction. Excluding from total$.', utxoTxIn);
-        return !utxoIsUsedInFlight;
+        const utxoIsUsedInSigned = signedTransactions.some(({ tx }) =>
+          tx.body.inputs.some((input) => txInEquals(input, utxoTxIn))
+        );
+        (utxoIsUsedInSigned || utxoIsUsedInFlight) &&
+          logger.debug(
+            `OnChain UTXO is already used in ${
+              utxoIsUsedInFlight ? 'in-flight' : 'signed'
+            } transaction. Excluding from total$.`,
+            utxoTxIn
+          );
+
+        return !utxoIsUsedInFlight && !utxoIsUsedInSigned;
       }),
       ...transactionsInFlight.flatMap(({ body: { outputs }, id }, txInFlightIndex) =>
         outputs
@@ -109,6 +121,31 @@ export const createUtxoTracker = (
             logger.debug('New UTXO available from in-flight transactions. Including in total$.', txIn);
             return [txIn, txOut];
           })
+      ),
+      ...signedTransactions.flatMap(({ tx: signedTx }, signedTxIndex) =>
+        signedTx.body.outputs
+          .filter(({ address }, outputIndex) => {
+            if (!ownAddresses.includes(address)) {
+              return false;
+            }
+
+            const alreadyConsumed = signedTransactions.some(
+              ({ tx: { body } }, i) =>
+                signedTxIndex !== i &&
+                body.inputs.some((input) => txInEquals(input, { index: outputIndex, txId: signedTx.id }))
+            );
+
+            return !alreadyConsumed;
+          })
+          .map((txOut): Cardano.Utxo => {
+            const txIn: Cardano.HydratedTxIn = {
+              address: txOut.address, // not necessarily correct in multi-address wallet
+              index: signedTx.body.outputs.indexOf(txOut),
+              txId: signedTx.id
+            };
+            logger.debug('New UTXO available from signed transactions. Including in total$.', txIn);
+            return [txIn, txOut];
+          })
       )
     ]),
     map((utxo) => {

packages/wallet/src/services/types.ts

@@ -83,6 +83,7 @@ export interface TransactionsTracker {
     readonly inFlight$: Observable<TxInFlight[]>;
     readonly submitting$: Observable<OutgoingTx>;
     readonly pending$: Observable<OutgoingTx>;
+    readonly signed$: Observable<SignedTx[]>;
     readonly failed$: Observable<FailedTx>;
     readonly onChain$: Observable<OutgoingOnChainTx>;
   };

packages/wallet/test/integration/cip30mapping.test.ts

@@ -28,11 +28,11 @@ import {
 } from '@cardano-sdk/core';
 import { HexBlob, ManagedFreeableScope } from '@cardano-sdk/util';
 import { InMemoryUnspendableUtxoStore, createInMemoryWalletStores } from '../../src/persistence';
-import { InitializeTxProps, InitializeTxResult } from '@cardano-sdk/tx-construction';
-import { PersonalWallet, cip30 } from '../../src';
+import { InitializeTxProps, InitializeTxResult, finalizeTx } from '@cardano-sdk/tx-construction';
+import { Observable, firstValueFrom, of } from 'rxjs';
+import { PersonalWallet, WalletUtil, cip30 } from '../../src';
 import { Providers, createWallet } from './util';
 import { buildDRepIDFromDRepKey, waitForWalletStateSettle } from '../util';
-import { firstValueFrom, of } from 'rxjs';
 import { dummyLogger as logger } from 'ts-log';
 import { stakeKeyDerivationPath, testAsyncKeyAgent } from '../../../key-management/test/mocks';
 import uniq from 'lodash/uniq';
@@ -83,6 +83,35 @@ const createWalletAndApiWithStores = async (
   return { api, confirmationCallback, wallet };
 };
 
+const signTx = async ({
+  tx,
+  addresses$,
+  walletUtil
+}: {
+  tx: Cardano.TxBodyWithHash;
+  addresses$: Observable<GroupedAddress[]>;
+  walletUtil: WalletUtil;
+}): Promise<Cardano.Tx> => {
+  const keyAgent = await testAsyncKeyAgent();
+  const knownAddresses = await firstValueFrom(addresses$);
+
+  const signed = await finalizeTx(
+    tx,
+    {
+      signingContext: {
+        knownAddresses,
+        txInKeyPathMap: await util.createTxInKeyPathMap(tx.body, knownAddresses, walletUtil)
+      }
+    },
+    {
+      bip32Account: await Bip32Account.fromAsyncKeyAgent(keyAgent),
+      witnesser: util.createBip32Ed25519Witnesser(keyAgent)
+    }
+  );
+
+  return signed.tx;
+};
+
 describe('cip30', () => {
   const context: SenderContext = { sender: { url: 'https://lace.io' } };
   let wallet: PersonalWallet;
@@ -489,7 +518,11 @@ describe('cip30', () => {
 
         beforeEach(async () => {
           const txInternals: InitializeTxResult = await wallet.initializeTx(simpleTxProps);
-          finalizedTx = await wallet.finalizeTx({ tx: txInternals });
+          finalizedTx = await signTx({
+            addresses$: wallet.addresses$,
+            tx: txInternals,
+            walletUtil: wallet.util
+          });
           hexTx = Serialization.Transaction.fromCore(finalizedTx).toCbor();
         });
 
@@ -663,7 +696,7 @@ describe('cip30', () => {
         let hexTx: string;
         beforeAll(async () => {
           const txInternals = await wallet.initializeTx(simpleTxProps);
-          const finalizedTx = await wallet.finalizeTx({ tx: txInternals });
+          const finalizedTx = await signTx({ addresses$: wallet.addresses$, tx: txInternals, walletUtil: wallet.util });
           hexTx = Serialization.Transaction.fromCore(finalizedTx).toCbor();
         });
 
@@ -789,7 +822,11 @@ describe('cip30', () => {
           ])
         };
 
-        tx = await mockWallet.finalizeTx({ tx: await mockWallet.initializeTx(props) });
+        tx = await signTx({
+          addresses$: mockWallet.addresses$,
+          tx: await mockWallet.initializeTx(props),
+          walletUtil: mockWallet.util
+        });
       });
 
       afterEach(() => {