commit 7efd435222d0e5a35d3319ddbbf5314c402830f8
parent a72a556539edb00e773e559d6283c5fc719375e2
Author: Florian Dold <florian@dold.me>
Date: Wed, 4 Jun 2025 22:32:18 +0200
wallet-core: refactor transaction/balance notifications
Instead of threading the notifications explicitly through the code, the
database handle now supports caching notifications and emitting them iff
the transaction has commited successfully.
Diffstat:
11 files changed, 251 insertions(+), 200 deletions(-)
diff --git a/packages/taler-wallet-core/src/db.ts b/packages/taler-wallet-core/src/db.ts
@@ -3823,14 +3823,6 @@ export async function openTalerDatabase(
onTalerDbUpgradeNeeded,
);
- const mainDbAccess = new DbAccessImpl(
- mainDbHandle,
- WalletStoresV1,
- {},
- CancellationToken.CONTINUE,
- );
- await applyFixups(mainDbAccess);
-
return mainDbHandle;
}
diff --git a/packages/taler-wallet-core/src/deposits.ts b/packages/taler-wallet-core/src/deposits.ts
@@ -135,7 +135,7 @@ import {
} from "./refresh.js";
import {
BalanceEffect,
- TransitionInfo,
+ applyNotifyTransition,
constructTransactionIdentifier,
isUnsuccessfulTransaction,
notifyTransition,
@@ -362,7 +362,7 @@ export class DepositTransactionContext implements TransactionContext {
async suspendTransaction(): Promise<void> {
const { wex, depositGroupId, transactionId, taskId: retryTag } = this;
- const transitionInfo = await wex.db.runReadWriteTx(
+ await wex.db.runReadWriteTx(
{ storeNames: ["depositGroups", "transactionsMeta"] },
async (tx) => {
const dg = await tx.depositGroups.get(depositGroupId);
@@ -417,20 +417,19 @@ export class DepositTransactionContext implements TransactionContext {
dg.operationStatus = newOpStatus;
await tx.depositGroups.put(dg);
await this.updateTransactionMeta(tx);
- return {
+ applyNotifyTransition(tx.notify, transactionId, {
oldTxState: oldState,
newTxState: computeDepositTransactionStatus(dg),
balanceEffect: BalanceEffect.None,
- } satisfies TransitionInfo;
+ });
},
);
wex.taskScheduler.stopShepherdTask(retryTag);
- notifyTransition(wex, transactionId, transitionInfo);
}
async abortTransaction(reason?: TalerErrorDetail): Promise<void> {
const { wex, depositGroupId, transactionId, taskId: retryTag } = this;
- const transitionInfo = await wex.db.runReadWriteTx(
+ await wex.db.runReadWriteTx(
{ storeNames: ["depositGroups", "transactionsMeta"] },
async (tx) => {
const dg = await tx.depositGroups.get(depositGroupId);
@@ -452,11 +451,12 @@ export class DepositTransactionContext implements TransactionContext {
dg.abortReason = reason;
await tx.depositGroups.put(dg);
await this.updateTransactionMeta(tx);
- return {
+ applyNotifyTransition(tx.notify, transactionId, {
oldTxState: oldState,
newTxState: computeDepositTransactionStatus(dg),
balanceEffect: BalanceEffect.Any,
- };
+ });
+ return;
}
case DepositOperationStatus.FinalizingTrack:
case DepositOperationStatus.SuspendedFinalizingTrack:
@@ -478,7 +478,6 @@ export class DepositTransactionContext implements TransactionContext {
},
);
wex.taskScheduler.stopShepherdTask(retryTag);
- notifyTransition(wex, transactionId, transitionInfo);
wex.taskScheduler.startShepherdTask(retryTag);
}
diff --git a/packages/taler-wallet-core/src/exchanges.ts b/packages/taler-wallet-core/src/exchanges.ts
@@ -181,8 +181,8 @@ import { RecoupTransactionContext, createRecoupGroup } from "./recoup.js";
import { RefreshTransactionContext, createRefreshGroup } from "./refresh.js";
import {
BalanceEffect,
+ applyNotifyTransition,
constructTransactionIdentifier,
- notifyTransition,
rematerializeTransactions,
} from "./transactions.js";
import { WALLET_EXCHANGE_PROTOCOL_VERSION } from "./versions.js";
@@ -2408,25 +2408,24 @@ export class DenomLossTransactionContext implements TransactionContext {
}
async deleteTransaction(): Promise<void> {
- const transitionInfo = await this.wex.db.runReadWriteTx(
+ await this.wex.db.runReadWriteTx(
{ storeNames: ["denomLossEvents"] },
async (tx) => {
const rec = await tx.denomLossEvents.get(this.denomLossEventId);
- if (rec) {
- const oldTxState = computeDenomLossTransactionStatus(rec);
- await tx.denomLossEvents.delete(this.denomLossEventId);
- return {
- oldTxState,
- newTxState: {
- major: TransactionMajorState.Deleted,
- },
- balanceEffect: BalanceEffect.Any,
- };
+ if (!rec) {
+ return;
}
- return undefined;
+ const oldTxState = computeDenomLossTransactionStatus(rec);
+ await tx.denomLossEvents.delete(this.denomLossEventId);
+ applyNotifyTransition(tx.notify, this.transactionId, {
+ oldTxState,
+ newTxState: {
+ major: TransactionMajorState.Deleted,
+ },
+ balanceEffect: BalanceEffect.Any,
+ });
},
);
- notifyTransition(this.wex, this.transactionId, transitionInfo);
}
async lookupFullTransaction(
@@ -2728,14 +2727,12 @@ export async function listExchanges(
export async function markExchangeUsed(
tx: WalletDbReadWriteTransaction<["exchanges"]>,
exchangeBaseUrl: string,
-): Promise<{ notif: WalletNotification | undefined }> {
+): Promise<void> {
logger.trace(`marking exchange ${exchangeBaseUrl} as used`);
const exch = await tx.exchanges.get(exchangeBaseUrl);
if (!exch) {
logger.info(`exchange ${exchangeBaseUrl} NOT found`);
- return {
- notif: undefined,
- };
+ return;
}
const oldExchangeState = getExchangeState(exch);
@@ -2745,19 +2742,16 @@ export async function markExchangeUsed(
exch.entryStatus = ExchangeEntryDbRecordStatus.Used;
await tx.exchanges.put(exch);
const newExchangeState = getExchangeState(exch);
- return {
- notif: {
- type: NotificationType.ExchangeStateTransition,
- exchangeBaseUrl,
- newExchangeState: newExchangeState,
- oldExchangeState: oldExchangeState,
- } satisfies WalletNotification,
- };
+ tx.notify({
+ type: NotificationType.ExchangeStateTransition,
+ exchangeBaseUrl,
+ newExchangeState: newExchangeState,
+ oldExchangeState: oldExchangeState,
+ });
+ return;
}
default:
- return {
- notif: undefined,
- };
+ return;
}
}
diff --git a/packages/taler-wallet-core/src/pay-peer-common.ts b/packages/taler-wallet-core/src/pay-peer-common.ts
@@ -44,11 +44,7 @@ import {
WalletStoresV1,
} from "./db.js";
import { getTotalRefreshCost } from "./refresh.js";
-import {
- BalanceEffect,
- TransitionInfo,
- notifyTransition,
-} from "./transactions.js";
+import { BalanceEffect, applyNotifyTransition } from "./transactions.js";
import {
WalletExecutionContext,
getDenomInfo,
@@ -281,10 +277,13 @@ export async function recordCreate<
await (tx[ctx.store] as any).add(rec);
await tx.transactionsMeta.put(ctx.recordMeta(rec));
const newTxState = ctx.recordState(rec);
- return { oldTxState, newTxState, balanceEffect: BalanceEffect.Any };
+ applyNotifyTransition(tx.notify, ctx.transactionId, {
+ oldTxState,
+ newTxState,
+ balanceEffect: BalanceEffect.Any,
+ });
},
);
- notifyTransition(ctx.wex, ctx.transactionId, transitionInfo);
}
/**
@@ -303,7 +302,7 @@ export async function recordTransition<
[Store, "transactionsMeta", ...ExtraStores]
>,
) => Promise<TransitionResultType.Stay | TransitionResultType.Transition>,
-): Promise<TransitionInfo | undefined> {
+): Promise<void> {
const baseStore = [ctx.store, "transactionsMeta" as const];
const storeNames = opts.extraStores
? [...baseStore, ...opts.extraStores]
@@ -323,19 +322,18 @@ export async function recordTransition<
await tx[ctx.store].put(rec);
await tx.transactionsMeta.put(ctx.recordMeta(rec));
const newTxState = ctx.recordState(rec);
- return {
+ applyNotifyTransition(tx.notify, ctx.transactionId, {
oldTxState,
newTxState,
balanceEffect: BalanceEffect.Any,
- };
+ });
+ return;
}
case TransitionResultType.Stay:
return;
}
},
);
- notifyTransition(ctx.wex, ctx.transactionId, transitionInfo);
- return transitionInfo;
}
/** Extract the stored type status if any */
@@ -345,13 +343,13 @@ type StoreTypeStatus<Store extends WalletDbStoresName> =
/**
* Optionally update an existing record status from a state to another, ignore if missing.
* If a transition occurs, update its metadata and notify.
- **/
+ */
export async function recordTransitionStatus<Store extends WalletDbStoresName>(
ctx: RecordCtx<Store>,
from: StoreTypeStatus<Store>,
to: StoreTypeStatus<Store>,
-): Promise<TransitionInfo | undefined> {
- return recordTransition(ctx, {}, async (rec, _) => {
+): Promise<void> {
+ await recordTransition(ctx, {}, async (rec, _) => {
const it = rec as { status: StoreTypeStatus<Store> };
if (it.status !== from) {
return TransitionResultType.Stay;
diff --git a/packages/taler-wallet-core/src/pay-peer-pull-credit.ts b/packages/taler-wallet-core/src/pay-peer-pull-credit.ts
@@ -95,6 +95,8 @@ import {
waitForKycCompletion,
} from "./pay-peer-common.js";
import {
+ BalanceEffect,
+ applyNotifyBalanceEffect,
constructTransactionIdentifier,
isUnsuccessfulTransaction,
} from "./transactions.js";
@@ -590,7 +592,8 @@ async function handlePeerPullCreditWithdrawing(
await waitWithdrawalFinal(wex, pullIni.withdrawalGroupId);
const ctx = new PeerPullCreditTransactionContext(wex, pullIni.pursePub);
const wgId = pullIni.withdrawalGroupId;
- const info = await recordTransition(
+ let newTxState: TransactionState | undefined;
+ await recordTransition(
ctx,
{
extraStores: ["withdrawalGroups"],
@@ -610,10 +613,11 @@ async function handlePeerPullCreditWithdrawing(
break;
// FIXME: Also handle other final states!
}
+ newTxState = computePeerPullCreditTransactionState(rec);
return TransitionResultType.Transition;
},
);
- if (info?.newTxState.major != TransactionMajorState.Pending) {
+ if (newTxState && newTxState.major != TransactionMajorState.Pending) {
return TaskRunResult.finished();
} else {
// FIXME: Return indicator that we depend on the other operation!
@@ -904,13 +908,18 @@ async function processPeerPullCreditKycRequired(
return TaskRunResult.finished();
case HttpStatusCode.Accepted: {
logger.info(`kyc status: ${j2s(res.body)}`);
- await recordTransition(ctx, {}, async (rec) => {
+ await recordTransition(ctx, {}, async (rec, tx) => {
rec.kycPaytoHash = kycPaytoHash;
logger.info(
`setting peer-pull-credit kyc payto hash to ${kycPaytoHash}`,
);
rec.kycAccessToken = res.body.access_token;
rec.status = PeerPullPaymentCreditStatus.PendingMergeKycRequired;
+ applyNotifyBalanceEffect(
+ tx.notify,
+ ctx.transactionId,
+ BalanceEffect.Flags,
+ );
return TransitionResultType.Transition;
});
return TaskRunResult.progress();
diff --git a/packages/taler-wallet-core/src/pay-peer-push-credit.ts b/packages/taler-wallet-core/src/pay-peer-push-credit.ts
@@ -98,9 +98,9 @@ import {
} from "./pay-peer-common.js";
import {
BalanceEffect,
+ applyNotifyTransition,
constructTransactionIdentifier,
isUnsuccessfulTransaction,
- notifyTransition,
parseTransactionIdentifier,
} from "./transactions.js";
import { WalletExecutionContext, walletExchangeClient } from "./wallet.js";
@@ -669,15 +669,12 @@ async function processPeerPushCreditKycRequired(
return TaskRunResult.finished();
case HttpStatusCode.Accepted:
logger.info(`kyc-check response body: ${j2s(resp.body)}`);
- const { transitionInfo, result } = await wex.db.runReadWriteTx(
+ return await wex.db.runReadWriteTx(
{ storeNames: ["peerPushCredit", "transactionsMeta"] },
async (tx) => {
const peerInc = await tx.peerPushCredit.get(ctx.peerPushCreditId);
if (!peerInc) {
- return {
- transitionInfo: undefined,
- result: TaskRunResult.finished(),
- };
+ return TaskRunResult.finished();
}
const oldTxState = computePeerPushCreditTransactionState(peerInc);
peerInc.kycPaytoHash = kycPending.h_payto;
@@ -686,18 +683,14 @@ async function processPeerPushCreditKycRequired(
const newTxState = computePeerPushCreditTransactionState(peerInc);
await tx.peerPushCredit.put(peerInc);
await ctx.updateTransactionMeta(tx);
- return {
- transitionInfo: {
- oldTxState,
- newTxState,
- balanceEffect: BalanceEffect.Any,
- },
- result: TaskRunResult.progress(),
- };
+ applyNotifyTransition(tx.notify, ctx.transactionId, {
+ oldTxState,
+ newTxState,
+ balanceEffect: BalanceEffect.Flags,
+ });
+ return TaskRunResult.progress();
},
);
- notifyTransition(wex, ctx.transactionId, transitionInfo);
- return result;
case HttpStatusCode.Conflict:
case HttpStatusCode.Forbidden:
case HttpStatusCode.NotFound:
@@ -825,7 +818,7 @@ async function handlePendingMerge(
},
});
- const txRes = await wex.db.runReadWriteTx(
+ await wex.db.runReadWriteTx(
{
storeNames: [
"contractTerms",
@@ -862,27 +855,13 @@ async function handlePendingMerge(
await tx.peerPushCredit.put(peerInc);
await ctx.updateTransactionMeta(tx);
const newTxState = computePeerPushCreditTransactionState(peerInc);
- return {
- peerPushCreditTransition: {
- oldTxState,
- newTxState,
- balanceEffect: BalanceEffect.Any,
- },
- wgCreateRes,
- };
+ applyNotifyTransition(tx.notify, ctx.transactionId, {
+ oldTxState,
+ newTxState,
+ balanceEffect: BalanceEffect.Any,
+ });
},
);
- // Transaction was committed, now we can emit notifications.
- if (txRes?.wgCreateRes?.exchangeNotif) {
- wex.ws.notify(txRes.wgCreateRes.exchangeNotif);
- }
- notifyTransition(
- wex,
- withdrawalGroupPrep.transactionId,
- txRes?.wgCreateRes?.transitionInfo,
- );
- notifyTransition(wex, ctx.transactionId, txRes?.peerPushCreditTransition);
-
return TaskRunResult.backoff();
}
@@ -900,7 +879,7 @@ async function handlePendingWithdrawing(
);
const wgId = peerInc.withdrawalGroupId;
let finished: boolean = false;
- const transitionInfo = await wex.db.runReadWriteTx(
+ await wex.db.runReadWriteTx(
{ storeNames: ["peerPushCredit", "withdrawalGroups", "transactionsMeta"] },
async (tx) => {
const ppi = await tx.peerPushCredit.get(peerInc.peerPushCreditId);
@@ -916,7 +895,7 @@ async function handlePendingWithdrawing(
const wg = await tx.withdrawalGroups.get(wgId);
if (!wg) {
// FIXME: Fail the operation instead?
- return undefined;
+ return;
}
switch (wg.status) {
case WithdrawalGroupStatus.Done:
@@ -928,14 +907,14 @@ async function handlePendingWithdrawing(
await tx.peerPushCredit.put(ppi);
await ctx.updateTransactionMeta(tx);
const newTxState = computePeerPushCreditTransactionState(ppi);
- return {
+ applyNotifyTransition(tx.notify, ctx.transactionId, {
oldTxState,
newTxState,
balanceEffect: BalanceEffect.Any,
- };
+ });
+ return;
},
);
- notifyTransition(wex, ctx.transactionId, transitionInfo);
if (finished) {
return TaskRunResult.finished();
} else {
diff --git a/packages/taler-wallet-core/src/query.ts b/packages/taler-wallet-core/src/query.ts
@@ -43,6 +43,7 @@ import {
Logger,
openPromise,
safeStringifyException,
+ WalletNotification,
} from "@gnu-taler/taler-util";
const logger = new Logger("query.ts");
@@ -526,8 +527,8 @@ type DerefKeyPath<T, P> = P extends `${infer PX extends keyof T &
KeyPathComponents}`
? T[PX]
: P extends `${infer P0 extends keyof T & KeyPathComponents}.${infer Rest}`
- ? DerefKeyPath<T[P0], Rest>
- : unknown;
+ ? DerefKeyPath<T[P0], Rest>
+ : unknown;
/**
* Return a path if it is a valid dot-separate path to an object.
@@ -537,8 +538,8 @@ type ValidateKeyPath<T, P> = P extends `${infer PX extends keyof T &
KeyPathComponents}`
? PX
: P extends `${infer P0 extends keyof T & KeyPathComponents}.${infer Rest}`
- ? `${P0}.${ValidateKeyPath<T[P0], Rest>}`
- : never;
+ ? `${P0}.${ValidateKeyPath<T[P0], Rest>}`
+ : never;
// function foo<T, P>(
// x: T,
@@ -547,20 +548,31 @@ type ValidateKeyPath<T, P> = P extends `${infer PX extends keyof T &
// foo({x: [0,1,2]}, "x.0");
-export type StoreMap = { [Store: string]: StoreWithIndexes<any, any, any> }
-export type StoreNames<Stores extends StoreMap> = keyof Stores
+export type StoreMap = { [Store: string]: StoreWithIndexes<any, any, any> };
+export type StoreNames<Stores extends StoreMap> = keyof Stores;
export type DbReadWriteTransaction<
Stores extends StoreMap,
StoresArr extends Array<StoreNames<Stores>>,
> = {
- [X in StoresArr[number]]: StoreReadWriteAccessor<Stores[X]['store']['_dummy'], Stores[X]['indexMap']>
- }
+ [X in StoresArr[number]]: StoreReadWriteAccessor<
+ Stores[X]["store"]["_dummy"],
+ Stores[X]["indexMap"]
+ >;
+} & {
+ notify: (w: WalletNotification) => void;
+};
+
export type DbReadOnlyTransaction<
Stores extends StoreMap,
StoresArr extends Array<StoreNames<Stores>>,
> = {
- [X in StoresArr[number]]: StoreReadOnlyAccessor<Stores[X]['store']['_dummy'], Stores[X]['indexMap']>
- }
+ [X in StoresArr[number]]: StoreReadOnlyAccessor<
+ Stores[X]["store"]["_dummy"],
+ Stores[X]["indexMap"]
+ >;
+} & {
+ notify: (w: WalletNotification) => void;
+};
/**
* Convert the type of an array to a union of the contents.
@@ -623,6 +635,13 @@ function runTx<Arg, Res>(
resolve(funResult);
}
internalContext.handleAfterCommit();
+ // Notify here.
+ if (
+ internalContext.notifications.length > 0 &&
+ internalContext.applyNotifications
+ ) {
+ internalContext.applyNotifications(internalContext.notifications);
+ }
unregisterOnCancelled();
};
tx.onerror = () => {
@@ -701,12 +720,20 @@ function runTx<Arg, Res>(
* Create a transaction handle that will be passed
* to the main handler for the transaction.
*/
-function makeTxContext(
+function makeTxClientContext(
tx: IDBTransaction,
storePick: { [n: string]: StoreWithIndexes<any, any, any> },
internalContext: InternalTransactionContext,
): any {
- const ctx: { [s: string]: StoreReadWriteAccessor<any, any> } = {};
+ const ctx: {
+ [s: string]:
+ | StoreReadWriteAccessor<any, any>
+ | ((notif: WalletNotification) => void);
+ } = {
+ notify(notif: WalletNotification): void {
+ internalContext.scheduleNotification(notif);
+ },
+ };
for (const storeAlias in storePick) {
const indexes: { [s: string]: IndexReadWriteAccessor<any> } = {};
const swi = storePick[storeAlias];
@@ -926,17 +953,23 @@ class InternalTransactionContext {
storesModified: Set<string> = new Set();
allowWrite: boolean;
abortExn: TransactionAbortedError | undefined;
+ notifications: WalletNotification[] = [];
constructor(
- private triggerSpec: TriggerSpec,
- private mode: IDBTransactionMode,
- scope: string[],
- public cancellationToken: CancellationToken,
+ private readonly triggerSpec: TriggerSpec,
+ private readonly mode: IDBTransactionMode,
+ readonly scope: string[],
+ public readonly cancellationToken: CancellationToken,
+ public readonly applyNotifications?: (notifs: WalletNotification[]) => void,
) {
this.storesScope = new Set(scope);
this.allowWrite = mode === "readwrite" || mode === "versionchange";
}
+ scheduleNotification(notif: WalletNotification): void {
+ this.notifications.push(notif);
+ }
+
handleAfterCommit() {
if (this.triggerSpec.afterCommit) {
this.triggerSpec.afterCommit({
@@ -967,7 +1000,8 @@ export class DbAccessImpl<Stores extends StoreMap> implements DbAccess<Stores> {
private stores: Stores,
private triggers: TriggerSpec = {},
private cancellationToken: CancellationToken,
- ) { }
+ private applyNotifications?: (notifs: WalletNotification[]) => void,
+ ) {}
idbHandle(): IDBDatabase {
return this.db;
@@ -996,9 +1030,14 @@ export class DbAccessImpl<Stores extends StoreMap> implements DbAccess<Stores> {
mode,
strStoreNames,
this.cancellationToken,
+ this.applyNotifications,
);
const tx = this.db.transaction(strStoreNames, mode);
- const writeContext = makeTxContext(tx, accessibleStores, triggerContext);
+ const writeContext = makeTxClientContext(
+ tx,
+ accessibleStores,
+ triggerContext,
+ );
return await runTx(tx, writeContext, txf, triggerContext);
}
@@ -1020,15 +1059,20 @@ export class DbAccessImpl<Stores extends StoreMap> implements DbAccess<Stores> {
accessibleStores[swi.storeName] = swi;
}
const mode = "readonly";
- const triggerContext = new InternalTransactionContext(
+ const internalContext = new InternalTransactionContext(
this.triggers,
mode,
strStoreNames,
this.cancellationToken,
+ this.applyNotifications,
);
const tx = this.db.transaction(strStoreNames, mode);
- const writeContext = makeTxContext(tx, accessibleStores, triggerContext);
- const res = await runTx(tx, writeContext, txf, triggerContext);
+ const writeContext = makeTxClientContext(
+ tx,
+ accessibleStores,
+ internalContext,
+ );
+ const res = await runTx(tx, writeContext, txf, internalContext);
return res;
}
@@ -1053,9 +1097,14 @@ export class DbAccessImpl<Stores extends StoreMap> implements DbAccess<Stores> {
mode,
strStoreNames,
this.cancellationToken,
+ this.applyNotifications,
);
const tx = this.db.transaction(strStoreNames, mode);
- const writeContext = makeTxContext(tx, accessibleStores, triggerContext);
+ const writeContext = makeTxClientContext(
+ tx,
+ accessibleStores,
+ triggerContext,
+ );
const res = await runTx(tx, writeContext, txf, triggerContext);
return res;
}
@@ -1081,9 +1130,14 @@ export class DbAccessImpl<Stores extends StoreMap> implements DbAccess<Stores> {
mode,
strStoreNames,
this.cancellationToken,
+ this.applyNotifications,
);
const tx = this.db.transaction(strStoreNames, mode);
- const readContext = makeTxContext(tx, accessibleStores, triggerContext);
+ const readContext = makeTxClientContext(
+ tx,
+ accessibleStores,
+ triggerContext,
+ );
const res = await runTx(tx, readContext, txf, triggerContext);
return res;
}
diff --git a/packages/taler-wallet-core/src/refresh.ts b/packages/taler-wallet-core/src/refresh.ts
@@ -120,11 +120,10 @@ import {
import { selectWithdrawalDenominations } from "./denomSelection.js";
import { fetchFreshExchange, getScopeForAllExchanges } from "./exchanges.js";
import {
+ applyNotifyTransition,
BalanceEffect,
constructTransactionIdentifier,
isUnsuccessfulTransaction,
- notifyTransition,
- TransitionInfo,
} from "./transactions.js";
import {
EXCHANGE_COINS_LOCK,
@@ -245,7 +244,7 @@ export class RefreshTransactionContext implements TransactionContext {
]
>,
) => Promise<TransitionResult<RefreshGroupRecord>>,
- ): Promise<TransitionInfo | undefined> {
+ ): Promise<boolean> {
const baseStores = [
"refreshGroups" as const,
"transactionsMeta" as const,
@@ -256,7 +255,7 @@ export class RefreshTransactionContext implements TransactionContext {
let stores = opts.extraStores
? [...baseStores, ...opts.extraStores]
: baseStores;
- const transitionInfo = await this.wex.db.runReadWriteTx(
+ return await this.wex.db.runReadWriteTx(
{ storeNames: stores },
async (tx) => {
const wgRec = await tx.refreshGroups.get(this.refreshGroupId);
@@ -274,30 +273,30 @@ export class RefreshTransactionContext implements TransactionContext {
await tx.refreshGroups.put(res.rec);
await this.updateTransactionMeta(tx);
const newTxState = computeRefreshTransactionState(res.rec);
- return {
+ applyNotifyTransition(tx.notify, this.transactionId, {
oldTxState,
newTxState,
balanceEffect: BalanceEffect.PreserveUserVisible,
- } satisfies TransitionInfo;
+ });
+ return true;
}
case TransitionResultType.Delete:
await tx.refreshGroups.delete(this.refreshGroupId);
await this.updateTransactionMeta(tx);
- return {
+ applyNotifyTransition(tx.notify, this.transactionId, {
oldTxState,
newTxState: {
major: TransactionMajorState.None,
},
// Deletion will affect balance
balanceEffect: BalanceEffect.Any,
- } satisfies TransitionInfo;
+ });
+ return true;
default:
- return undefined;
+ return false;
}
},
);
- notifyTransition(this.wex, this.transactionId, transitionInfo);
- return transitionInfo;
}
async deleteTransaction(): Promise<void> {
@@ -1747,7 +1746,7 @@ export async function processRefreshGroup(
// We've processed all refresh session and can now update the
// status of the whole refresh group.
- const transitionInfo = await wex.db.runReadWriteTx(
+ const didTransition: boolean = await wex.db.runReadWriteTx(
{
storeNames: [
"coins",
@@ -1759,13 +1758,13 @@ export async function processRefreshGroup(
async (tx) => {
const rg = await tx.refreshGroups.get(refreshGroupId);
if (!rg) {
- return;
+ return false;
}
switch (rg.operationStatus) {
case RefreshOperationStatus.Pending:
break;
default:
- return undefined;
+ return false;
}
const oldTxState = computeRefreshTransactionState(rg);
const allFinal = fnutil.all(
@@ -1793,21 +1792,21 @@ export async function processRefreshGroup(
await tx.refreshGroups.put(rg);
await ctx.updateTransactionMeta(tx);
const newTxState = computeRefreshTransactionState(rg);
- return {
+ applyNotifyTransition(tx.notify, ctx.transactionId, {
oldTxState,
newTxState,
balanceEffect:
rg.operationStatus === RefreshOperationStatus.Failed
? BalanceEffect.Any
: BalanceEffect.PreserveUserVisible,
- } satisfies TransitionInfo;
+ });
+ return true;
}
- return undefined;
+ return false;
},
);
- if (transitionInfo) {
- notifyTransition(wex, ctx.transactionId, transitionInfo);
+ if (didTransition) {
return TaskRunResult.progress();
}
diff --git a/packages/taler-wallet-core/src/transactions.ts b/packages/taler-wallet-core/src/transactions.ts
@@ -41,6 +41,7 @@ import {
TransactionsResponse,
TransactionState,
TransactionType,
+ WalletNotification,
} from "@gnu-taler/taler-util";
import {
constructTaskIdentifier,
@@ -942,10 +943,28 @@ export enum BalanceEffect {
}
/**
- * Notify of a state transition if necessary.
+ * Call the given notification function with a
+ * balance effect notification, if necessary.
*/
-export function notifyTransition(
- wex: WalletExecutionContext,
+export function applyNotifyBalanceEffect(
+ notify: (n: WalletNotification) => void,
+ transactionId: string,
+ balanceEffect: BalanceEffect,
+): void {
+ if (balanceEffect > BalanceEffect.None) {
+ notify({
+ type: NotificationType.BalanceChange,
+ hintTransactionId: transactionId,
+ isInternal: balanceEffect <= BalanceEffect.PreserveUserVisible,
+ });
+ }
+}
+
+/**
+ * Notify of a state transition and balance change if necessary.
+ */
+export function applyNotifyTransition(
+ notify: (n: WalletNotification) => void,
transactionId: string,
transitionInfo: TransitionInfo | undefined,
experimentalUserData: any = undefined,
@@ -957,7 +976,7 @@ export function notifyTransition(
transitionInfo.oldTxState.minor === transitionInfo.newTxState.minor
)
) {
- wex.ws.notify({
+ notify({
type: NotificationType.TransactionStateTransition,
oldTxState: transitionInfo.oldTxState,
newTxState: transitionInfo.newTxState,
@@ -965,13 +984,29 @@ export function notifyTransition(
experimentalUserData,
});
- if (transitionInfo.balanceEffect > BalanceEffect.None) {
- wex.ws.notify({
- type: NotificationType.BalanceChange,
- hintTransactionId: transactionId,
- isInternal:
- transitionInfo.balanceEffect <= BalanceEffect.PreserveUserVisible,
- });
- }
+ applyNotifyBalanceEffect(
+ notify,
+ transactionId,
+ transitionInfo.balanceEffect,
+ );
}
}
+
+/**
+ * Notify of a state transition if necessary.
+ *
+ * @deprecated use applyNotifyTransition inside transaction
+ */
+export function notifyTransition(
+ wex: WalletExecutionContext,
+ transactionId: string,
+ transitionInfo: TransitionInfo | undefined,
+ experimentalUserData: any = undefined,
+): void {
+ applyNotifyTransition(
+ (wn: WalletNotification) => wex.ws.notify(wn),
+ transactionId,
+ transitionInfo,
+ experimentalUserData,
+ );
+}
diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
@@ -277,6 +277,7 @@ import {
WalletDbHelpers,
WalletDbReadOnlyTransaction,
WalletStoresV1,
+ applyFixups,
clearDatabase,
exportDb,
importDb,
@@ -2825,11 +2826,17 @@ export class InternalWalletState {
if (!this._indexedDbHandle) {
throw Error("db not initialized");
}
+ const iws = this;
return new DbAccessImpl(
this._indexedDbHandle,
WalletStoresV1,
new WalletDbTriggerSpec(this),
cancellationToken,
+ (notifs: WalletNotification[]): void => {
+ for (const notif of notifs) {
+ iws.notify(notif);
+ }
+ },
);
}
@@ -2868,6 +2875,8 @@ export class InternalWalletState {
try {
const myDb = await openTalerDatabase(this.idbFactory, myVersionChange);
this._indexedDbHandle = myDb;
+ const dbAccess = this.createDbAccessHandle(CancellationToken.CONTINUE);
+ await applyFixups(dbAccess);
} catch (e) {
logger.error("error writing to database during initialization");
throw TalerError.fromDetail(TalerErrorCode.WALLET_DB_UNAVAILABLE, {
diff --git a/packages/taler-wallet-core/src/withdraw.ts b/packages/taler-wallet-core/src/withdraw.ts
@@ -184,10 +184,9 @@ import {
import { DbAccess } from "./query.js";
import {
BalanceEffect,
- TransitionInfo,
+ applyNotifyTransition,
constructTransactionIdentifier,
isUnsuccessfulTransaction,
- notifyTransition,
parseTransactionIdentifier,
} from "./transactions.js";
import { WALLET_EXCHANGE_PROTOCOL_VERSION } from "./versions.js";
@@ -498,7 +497,7 @@ export class WithdrawTransactionContext implements TransactionContext {
]
>,
) => Promise<TransitionResult<WithdrawalGroupRecord>>,
- ): Promise<TransitionInfo | undefined> {
+ ): Promise<boolean> {
const baseStores = [
"withdrawalGroups" as const,
"transactionsMeta" as const,
@@ -511,7 +510,7 @@ export class WithdrawTransactionContext implements TransactionContext {
: baseStores;
let errorThrown: Error | undefined;
- const transitionInfo = await this.wex.db.runReadWriteTx(
+ const didTransition: boolean = await this.wex.db.runReadWriteTx(
{ storeNames: stores },
async (tx) => {
const wgRec = await tx.withdrawalGroups.get(this.withdrawalGroupId);
@@ -530,7 +529,7 @@ export class WithdrawTransactionContext implements TransactionContext {
if (error instanceof Error) {
errorThrown = error;
}
- return undefined;
+ return false;
}
switch (res.type) {
@@ -538,32 +537,33 @@ export class WithdrawTransactionContext implements TransactionContext {
await tx.withdrawalGroups.put(res.rec);
await this.updateTransactionMeta(tx);
const newTxState = computeWithdrawalTransactionStatus(res.rec);
- return {
+ applyNotifyTransition(tx.notify, this.transactionId, {
oldTxState,
newTxState,
balanceEffect: res.balanceEffect,
- };
+ });
+ return true;
}
case TransitionResultType.Delete:
await tx.withdrawalGroups.delete(this.withdrawalGroupId);
await this.updateTransactionMeta(tx);
- return {
+ applyNotifyTransition(tx.notify, this.transactionId, {
oldTxState,
newTxState: {
major: TransactionMajorState.None,
},
balanceEffect: BalanceEffect.Any,
- };
+ });
+ return true;
default:
- return undefined;
+ return false;
}
},
);
if (errorThrown) {
throw errorThrown;
}
- notifyTransition(this.wex, this.transactionId, transitionInfo);
- return transitionInfo;
+ return didTransition;
}
async deleteTransaction(): Promise<void> {
@@ -3552,14 +3552,6 @@ export async function internalPrepareCreateWithdrawalGroup(
export interface PerformCreateWithdrawalGroupResult {
withdrawalGroup: WithdrawalGroupRecord;
- transitionInfo: TransitionInfo | undefined;
-
- /**
- * Notification for the exchange state transition.
- *
- * Should be emitted after the transaction has succeeded.
- */
- exchangeNotif: WalletNotification | undefined;
}
export async function internalPerformCreateWithdrawalGroup(
@@ -3576,8 +3568,6 @@ export async function internalPerformCreateWithdrawalGroup(
if (existingWg) {
return {
withdrawalGroup: existingWg,
- transitionInfo: undefined,
- exchangeNotif: undefined,
};
}
await tx.withdrawalGroups.add(withdrawalGroup);
@@ -3589,8 +3579,6 @@ export async function internalPerformCreateWithdrawalGroup(
if (!prep.creationInfo) {
return {
withdrawalGroup,
- transitionInfo: undefined,
- exchangeNotif: undefined,
};
}
return internalPerformExchangeWasUsed(
@@ -3624,7 +3612,7 @@ async function internalPerformExchangeWasUsed(
balanceEffect: BalanceEffect.Any,
};
- const exchangeUsedRes = await markExchangeUsed(tx, canonExchange);
+ await markExchangeUsed(tx, canonExchange);
const ctx = new WithdrawTransactionContext(
wex,
@@ -3633,10 +3621,10 @@ async function internalPerformExchangeWasUsed(
wex.taskScheduler.startShepherdTask(ctx.taskId);
+ applyNotifyTransition(tx.notify, ctx.transactionId, transitionInfo);
+
return {
withdrawalGroup,
- transitionInfo,
- exchangeNotif: exchangeUsedRes.notif,
};
}
@@ -3688,10 +3676,6 @@ export async function internalCreateWithdrawalGroup(
return res;
},
);
- if (res.exchangeNotif) {
- wex.ws.notify(res.exchangeNotif);
- }
- notifyTransition(wex, transactionId, res.transitionInfo);
return res.withdrawalGroup;
}
@@ -4019,17 +4003,16 @@ export async function confirmWithdrawal(
await wex.taskScheduler.resetTaskRetries(ctx.taskId);
// FIXME: Merge with transaction above!
- const res = await wex.db.runReadWriteTx({ storeNames: ["exchanges"] }, (tx) =>
- internalPerformExchangeWasUsed(
- wex,
- tx,
- exchange.exchangeBaseUrl,
- withdrawalGroup,
- ),
+ await wex.db.runReadWriteTx(
+ { storeNames: ["exchanges"] },
+ async (tx) =>
+ await internalPerformExchangeWasUsed(
+ wex,
+ tx,
+ exchange.exchangeBaseUrl,
+ withdrawalGroup,
+ ),
);
- if (res.exchangeNotif) {
- wex.ws.notify(res.exchangeNotif);
- }
return {
transactionId: req.transactionId as TransactionIdStr,
