commit f61b66937c046a75728b974ce0f8ac906b3face4
parent e700111ac1d156834478d480d05bc6e651f48bcf
Author: Florian Dold <florian@dold.me>
Date: Tue, 30 Jul 2024 22:10:34 +0200
wallet-core: use materialized transaction, migrate
Diffstat:
5 files changed, 157 insertions(+), 495 deletions(-)
diff --git a/packages/taler-wallet-core/src/db.ts b/packages/taler-wallet-core/src/db.ts
@@ -1353,6 +1353,7 @@ export enum ConfigRecordKey {
// Only for testing, do not use!
TestLoopTx = "testTxLoop",
LastInitInfo = "lastInitInfo",
+ MaterializedTransactionsVersion = "materializedTransactionsVersion",
}
/**
@@ -1366,7 +1367,8 @@ export type ConfigRecord =
}
| { key: ConfigRecordKey.CurrencyDefaultsApplied; value: boolean }
| { key: ConfigRecordKey.TestLoopTx; value: number }
- | { key: ConfigRecordKey.LastInitInfo; value: DbProtocolTimestamp };
+ | { key: ConfigRecordKey.LastInitInfo; value: DbProtocolTimestamp }
+ | { key: ConfigRecordKey.MaterializedTransactionsVersion; value: number };
export interface WalletBackupConfState {
deviceId: string;
diff --git a/packages/taler-wallet-core/src/dev-experiments.ts b/packages/taler-wallet-core/src/dev-experiments.ts
@@ -47,6 +47,8 @@ import {
RefreshOperationStatus,
timestampPreciseToDb,
} from "./db.js";
+import { DenomLossTransactionContext } from "./exchanges.js";
+import { RefreshTransactionContext } from "./refresh.js";
import { WalletExecutionContext } from "./wallet.js";
const logger = new Logger("dev-experiments.ts");
@@ -80,7 +82,7 @@ export async function applyDevExperiment(
case "insert-pending-refresh": {
const refreshGroupId = encodeCrock(getRandomBytes(32));
await wex.db.runReadWriteTx(
- { storeNames: ["refreshGroups"] },
+ { storeNames: ["refreshGroups", "transactionsMeta"] },
async (tx) => {
const newRg: RefreshGroupRecord = {
currency: "TESTKUDOS",
@@ -97,6 +99,8 @@ export async function applyDevExperiment(
infoPerExchange: {},
};
await tx.refreshGroups.put(newRg);
+ const ctx = new RefreshTransactionContext(wex, refreshGroupId);
+ await ctx.updateTransactionMeta(tx);
},
);
wex.taskScheduler.startShepherdTask(
@@ -109,7 +113,7 @@ export async function applyDevExperiment(
}
case "insert-denom-loss": {
await wex.db.runReadWriteTx(
- { storeNames: ["denomLossEvents"] },
+ { storeNames: ["denomLossEvents", "transactionsMeta"] },
async (tx) => {
const eventId = encodeCrock(getRandomBytes(32));
const newRg: DenomLossEventRecord = {
@@ -126,6 +130,11 @@ export async function applyDevExperiment(
timestampCreated: timestampPreciseToDb(TalerPreciseTimestamp.now()),
};
await tx.denomLossEvents.put(newRg);
+ const ctx = new DenomLossTransactionContext(
+ wex,
+ newRg.denomLossEventId,
+ );
+ await ctx.updateTransactionMeta(tx);
},
);
return;
diff --git a/packages/taler-wallet-core/src/pay-merchant.ts b/packages/taler-wallet-core/src/pay-merchant.ts
@@ -265,14 +265,12 @@ export class PayMerchantTransactionContext implements TransactionContext {
}));
const timestamp = purchaseRec.timestampAccept;
- checkDbInvariant(
- !!timestamp,
- `purchase ${purchaseRec.orderId} without accepted time`,
- );
- checkDbInvariant(
- !!purchaseRec.payInfo,
- `purchase ${purchaseRec.orderId} without payinfo`,
- );
+ if (!timestamp) {
+ return undefined;
+ }
+ if (!purchaseRec.payInfo) {
+ return undefined;
+ }
const txState = computePayMerchantTransactionState(purchaseRec);
return {
diff --git a/packages/taler-wallet-core/src/transactions.ts b/packages/taler-wallet-core/src/transactions.ts
@@ -17,7 +17,7 @@
/**
* Imports.
*/
-import { GlobalIDB } from "@gnu-taler/idb-bridge";
+import { GlobalIDB, IDBKeyRange } from "@gnu-taler/idb-bridge";
import {
AbsoluteTime,
Amounts,
@@ -30,7 +30,6 @@ import {
TransactionByIdRequest,
TransactionIdStr,
TransactionMajorState,
- TransactionRecordFilter,
TransactionsRequest,
TransactionsResponse,
TransactionState,
@@ -39,28 +38,13 @@ import {
import {
constructTaskIdentifier,
PendingTaskType,
- TaskIdentifiers,
TaskIdStr,
TransactionContext,
} from "./common.js";
import {
- DenomLossEventRecord,
- DepositGroupRecord,
OPERATION_STATUS_NONFINAL_FIRST,
OPERATION_STATUS_NONFINAL_LAST,
- PeerPullCreditRecord,
- PeerPullDebitRecordStatus,
- PeerPullPaymentIncomingRecord,
- PeerPushCreditStatus,
- PeerPushDebitRecord,
- PeerPushPaymentIncomingRecord,
- PurchaseRecord,
- RefreshGroupRecord,
- RefreshOperationStatus,
- RefundGroupRecord,
- WalletDbReadOnlyTransaction,
- WithdrawalGroupRecord,
- WithdrawalRecordType,
+ WalletDbAllStoresReadWriteTransaction,
} from "./db.js";
import { DepositTransactionContext } from "./deposits.js";
import { DenomLossTransactionContext } from "./exchanges.js";
@@ -116,22 +100,6 @@ function shouldSkipCurrency(
return false;
}
-function shouldSkipSearch(
- transactionsRequest: TransactionsRequest | undefined,
- fields: string[],
-): boolean {
- if (!transactionsRequest?.search) {
- return false;
- }
- const needle = transactionsRequest.search.trim();
- for (const f of fields) {
- if (f.indexOf(needle) >= 0) {
- return false;
- }
- }
- return true;
-}
-
/**
* Fallback order of transactions that have the same timestamp.
*/
@@ -204,288 +172,44 @@ export async function getTransactions(
): Promise<TransactionsResponse> {
const transactions: Transaction[] = [];
- const filter: TransactionRecordFilter = {};
- if (transactionsRequest?.filterByState) {
- filter.onlyState = transactionsRequest.filterByState;
+ let keyRange: IDBKeyRange | undefined = undefined;
+
+ if (transactionsRequest?.filterByState === "nonfinal") {
+ keyRange = GlobalIDB.KeyRange.bound(
+ OPERATION_STATUS_NONFINAL_FIRST,
+ OPERATION_STATUS_NONFINAL_LAST,
+ );
}
await wex.db.runAllStoresReadOnlyTx({}, async (tx) => {
- await iterRecordsForPeerPushDebit(tx, filter, async (pi) => {
- const amount = Amounts.parseOrThrow(pi.amount);
- const exchangesInTx = [pi.exchangeBaseUrl];
- if (
- shouldSkipCurrency(transactionsRequest, amount.currency, exchangesInTx)
- ) {
- return;
- }
- if (shouldSkipSearch(transactionsRequest, [])) {
- return;
- }
- const ctx = new PeerPushDebitTransactionContext(wex, pi.pursePub);
- const txDetails = await ctx.lookupFullTransaction(tx);
- if (txDetails) {
- transactions.push(txDetails);
- }
- });
-
- await iterRecordsForPeerPullDebit(tx, filter, async (pi) => {
- const amount = Amounts.parseOrThrow(pi.amount);
- const exchangesInTx = [pi.exchangeBaseUrl];
- if (
- shouldSkipCurrency(transactionsRequest, amount.currency, exchangesInTx)
- ) {
- return;
- }
- if (shouldSkipSearch(transactionsRequest, [])) {
- return;
- }
- if (
- pi.status !== PeerPullDebitRecordStatus.PendingDeposit &&
- pi.status !== PeerPullDebitRecordStatus.Done
- ) {
- // FIXME: Why?!
- return;
- }
-
- const ctx = new PeerPullDebitTransactionContext(wex, pi.peerPullDebitId);
- const txDetails = await ctx.lookupFullTransaction(tx);
- if (txDetails) {
- transactions.push(txDetails);
- }
- });
-
- await iterRecordsForPeerPushCredit(tx, filter, async (pi) => {
- if (!pi.currency) {
- // Legacy transaction
- return;
- }
- const exchangesInTx = [pi.exchangeBaseUrl];
- if (shouldSkipCurrency(transactionsRequest, pi.currency, exchangesInTx)) {
- return;
- }
- if (shouldSkipSearch(transactionsRequest, [])) {
- return;
- }
- if (pi.status === PeerPushCreditStatus.DialogProposed) {
- // We don't report proposed push credit transactions, user needs
- // to scan URI again and confirm to see it.
- return;
- }
-
- const ctx = new PeerPushCreditTransactionContext(
- wex,
- pi.peerPushCreditId,
- );
- const txDetails = await ctx.lookupFullTransaction(tx);
- if (txDetails) {
- transactions.push(txDetails);
- }
- });
-
- await iterRecordsForPeerPullCredit(tx, filter, async (pi) => {
- const currency = Amounts.currencyOf(pi.amount);
- const exchangesInTx = [pi.exchangeBaseUrl];
- if (shouldSkipCurrency(transactionsRequest, currency, exchangesInTx)) {
- return;
- }
- if (shouldSkipSearch(transactionsRequest, [])) {
- return;
- }
-
- const ctx = new PeerPullCreditTransactionContext(wex, pi.pursePub);
- const txDetails = await ctx.lookupFullTransaction(tx);
- if (txDetails) {
- transactions.push(txDetails);
- }
- });
-
- await iterRecordsForRefund(tx, filter, async (refundGroup) => {
- const currency = Amounts.currencyOf(refundGroup.amountRaw);
-
- const exchangesInTx: string[] = [];
- const p = await tx.purchases.get(refundGroup.proposalId);
- if (!p || !p.payInfo || !p.payInfo.payCoinSelection) {
- //refund with no payment
- return;
- }
-
- // FIXME: This is very slow, should become obsolete with materialized transactions.
- for (const cp of p.payInfo.payCoinSelection.coinPubs) {
- const c = await tx.coins.get(cp);
- if (c?.exchangeBaseUrl) {
- exchangesInTx.push(c.exchangeBaseUrl);
- }
- }
-
- if (shouldSkipCurrency(transactionsRequest, currency, exchangesInTx)) {
- return;
- }
-
- const ctx = new RefundTransactionContext(wex, refundGroup.refundGroupId);
- const txDetails = await ctx.lookupFullTransaction(tx);
- if (txDetails) {
- transactions.push(txDetails);
- }
- });
-
- await iterRecordsForRefresh(tx, filter, async (rg) => {
- const exchangesInTx = rg.infoPerExchange
- ? Object.keys(rg.infoPerExchange)
- : [];
- if (shouldSkipCurrency(transactionsRequest, rg.currency, exchangesInTx)) {
- return;
- }
- let required = false;
- const opId = TaskIdentifiers.forRefresh(rg);
- if (transactionsRequest?.includeRefreshes) {
- required = true;
- } else if (rg.operationStatus !== RefreshOperationStatus.Finished) {
- const ort = await tx.operationRetries.get(opId);
- if (ort) {
- required = true;
- }
- }
- if (required) {
- const ctx = new RefreshTransactionContext(wex, rg.refreshGroupId);
- const txDetails = await ctx.lookupFullTransaction(tx);
- if (txDetails) {
- transactions.push(txDetails);
- }
- }
- });
-
- await iterRecordsForWithdrawal(tx, filter, async (wsr) => {
- if (
- wsr.rawWithdrawalAmount === undefined ||
- wsr.exchangeBaseUrl == undefined
- ) {
- // skip prepared withdrawals which has not been confirmed
- return;
- }
- const exchangesInTx = [wsr.exchangeBaseUrl];
+ const allMetaTransactions =
+ await tx.transactionsMeta.indexes.byStatus.getAll(keyRange);
+ for (const metaTx of allMetaTransactions) {
if (
shouldSkipCurrency(
transactionsRequest,
- Amounts.currencyOf(wsr.rawWithdrawalAmount),
- exchangesInTx,
+ metaTx.currency,
+ metaTx.exchanges,
)
) {
- return;
- }
-
- if (shouldSkipSearch(transactionsRequest, [])) {
- return;
- }
-
- switch (wsr.wgInfo.withdrawalType) {
- case WithdrawalRecordType.PeerPullCredit:
- // Will be reported by the corresponding p2p transaction.
- // FIXME: If this is an orphan withdrawal, still report it as a withdrawal!
- // FIXME: Still report if requested with verbose option?
- return;
- case WithdrawalRecordType.PeerPushCredit:
- // Will be reported by the corresponding p2p transaction.
- // FIXME: If this is an orphan withdrawal, still report it as a withdrawal!
- // FIXME: Still report if requested with verbose option?
- return;
- case WithdrawalRecordType.BankIntegrated:
- case WithdrawalRecordType.BankManual: {
- const ctx = new WithdrawTransactionContext(
- wex,
- wsr.withdrawalGroupId,
- );
- const dbTxn = await ctx.lookupFullTransaction(tx);
- if (!dbTxn) {
- return;
- }
- transactions.push(dbTxn);
- return;
- }
- case WithdrawalRecordType.Recoup:
- // FIXME: Do we also report a transaction here?
- return;
- }
- });
-
- await iterRecordsForDenomLoss(tx, filter, async (rec) => {
- const amount = Amounts.parseOrThrow(rec.amount);
- const exchangesInTx = [rec.exchangeBaseUrl];
- if (
- shouldSkipCurrency(transactionsRequest, amount.currency, exchangesInTx)
- ) {
- return;
- }
- const ctx = new DenomLossTransactionContext(wex, rec.denomLossEventId);
- const txDetails = await ctx.lookupFullTransaction(tx);
- if (txDetails) {
- transactions.push(txDetails);
- }
- });
-
- await iterRecordsForDeposit(tx, filter, async (dg) => {
- const amount = Amounts.parseOrThrow(dg.amount);
- const exchangesInTx = dg.infoPerExchange
- ? Object.keys(dg.infoPerExchange)
- : [];
- if (
- shouldSkipCurrency(transactionsRequest, amount.currency, exchangesInTx)
- ) {
- return;
- }
-
- const ctx = new DepositTransactionContext(wex, dg.depositGroupId);
- const txDetails = await ctx.lookupFullTransaction(tx);
- if (txDetails) {
- transactions.push(txDetails);
- }
- });
-
- await iterRecordsForPurchase(tx, filter, async (purchase) => {
- const download = purchase.download;
- if (!download) {
- return;
- }
- if (!purchase.payInfo) {
- return;
- }
-
- const exchangesInTx: string[] = [];
- for (const cp of purchase.payInfo.payCoinSelection?.coinPubs ?? []) {
- const c = await tx.coins.get(cp);
- if (c?.exchangeBaseUrl) {
- exchangesInTx.push(c.exchangeBaseUrl);
- }
+ continue;
}
+ const parsedTx = parseTransactionIdentifier(metaTx.transactionId);
if (
- shouldSkipCurrency(
- transactionsRequest,
- download.currency,
- exchangesInTx,
- )
+ parsedTx?.tag === TransactionType.Refresh &&
+ !transactionsRequest?.includeRefreshes
) {
- return;
- }
- const contractTermsRecord = await tx.contractTerms.get(
- download.contractTermsHash,
- );
- if (!contractTermsRecord) {
- return;
- }
- if (
- shouldSkipSearch(transactionsRequest, [
- contractTermsRecord?.contractTermsRaw?.summary || "",
- ])
- ) {
- return;
+ continue;
}
- const ctx = new PayMerchantTransactionContext(wex, purchase.proposalId);
+ const ctx = await getContextForTransaction(wex, metaTx.transactionId);
const txDetails = await ctx.lookupFullTransaction(tx);
- if (txDetails) {
- transactions.push(txDetails);
+ if (!txDetails) {
+ continue;
}
- });
+ transactions.push(txDetails);
+ }
});
// One-off checks, because of a bug where the wallet previously
@@ -542,6 +266,73 @@ export async function getTransactions(
return { transactions: [...txPending, ...txNotPending] };
}
+/**
+ * Re-create materialized transactions from scratch.
+ *
+ * Used for migrations.
+ */
+export async function rematerializeTransactions(
+ wex: WalletExecutionContext,
+ tx: WalletDbAllStoresReadWriteTransaction,
+): Promise<void> {
+ logger.info("re-materializing transactions");
+
+ const allTxMeta = await tx.transactionsMeta.getAll();
+ for (const txMeta of allTxMeta) {
+ await tx.transactionsMeta.delete(txMeta.transactionId);
+ }
+
+ await tx.peerPushDebit.iter().forEachAsync(async (x) => {
+ const ctx = new PeerPushDebitTransactionContext(wex, x.pursePub);
+ await ctx.updateTransactionMeta(tx);
+ });
+
+ await tx.peerPushCredit.iter().forEachAsync(async (x) => {
+ const ctx = new PeerPushCreditTransactionContext(wex, x.peerPushCreditId);
+ await ctx.updateTransactionMeta(tx);
+ });
+
+ await tx.peerPullCredit.iter().forEachAsync(async (x) => {
+ const ctx = new PeerPullCreditTransactionContext(wex, x.pursePub);
+ await ctx.updateTransactionMeta(tx);
+ });
+
+ await tx.peerPullDebit.iter().forEachAsync(async (x) => {
+ const ctx = new PeerPullDebitTransactionContext(wex, x.peerPullDebitId);
+ await ctx.updateTransactionMeta(tx);
+ });
+
+ await tx.refundGroups.iter().forEachAsync(async (x) => {
+ const ctx = new RefundTransactionContext(wex, x.refundGroupId);
+ await ctx.updateTransactionMeta(tx);
+ });
+
+ await tx.refreshGroups.iter().forEachAsync(async (x) => {
+ const ctx = new RefreshTransactionContext(wex, x.refreshGroupId);
+ await ctx.updateTransactionMeta(tx);
+ });
+
+ await tx.withdrawalGroups.iter().forEachAsync(async (x) => {
+ const ctx = new WithdrawTransactionContext(wex, x.withdrawalGroupId);
+ await ctx.updateTransactionMeta(tx);
+ });
+
+ await tx.denomLossEvents.iter().forEachAsync(async (x) => {
+ const ctx = new DenomLossTransactionContext(wex, x.denomLossEventId);
+ await ctx.updateTransactionMeta(tx);
+ });
+
+ await tx.depositGroups.iter().forEachAsync(async (x) => {
+ const ctx = new DepositTransactionContext(wex, x.depositGroupId);
+ await ctx.updateTransactionMeta(tx);
+ });
+
+ await tx.purchases.iter().forEachAsync(async (x) => {
+ const ctx = new PayMerchantTransactionContext(wex, x.proposalId);
+ await ctx.updateTransactionMeta(tx);
+ });
+}
+
export type ParsedTransactionIdentifier =
| { tag: TransactionType.Deposit; depositGroupId: string }
| { tag: TransactionType.Payment; proposalId: string }
@@ -873,187 +664,3 @@ export function notifyTransition(
});
}
}
-
-/**
- * Iterate refresh records based on a filter.
- */
-async function iterRecordsForRefresh(
- tx: WalletDbReadOnlyTransaction<["refreshGroups"]>,
- filter: TransactionRecordFilter,
- f: (r: RefreshGroupRecord) => Promise<void>,
-): Promise<void> {
- let refreshGroups: RefreshGroupRecord[];
- if (filter.onlyState === "nonfinal") {
- const keyRange = GlobalIDB.KeyRange.bound(
- RefreshOperationStatus.Pending,
- RefreshOperationStatus.Suspended,
- );
- refreshGroups = await tx.refreshGroups.indexes.byStatus.getAll(keyRange);
- } else {
- refreshGroups = await tx.refreshGroups.indexes.byStatus.getAll();
- }
-
- for (const r of refreshGroups) {
- await f(r);
- }
-}
-
-async function iterRecordsForWithdrawal(
- tx: WalletDbReadOnlyTransaction<["withdrawalGroups"]>,
- filter: TransactionRecordFilter,
- f: (r: WithdrawalGroupRecord) => Promise<void>,
-): Promise<void> {
- let withdrawalGroupRecords: WithdrawalGroupRecord[];
- if (filter.onlyState === "nonfinal") {
- const keyRange = GlobalIDB.KeyRange.bound(
- OPERATION_STATUS_NONFINAL_FIRST,
- OPERATION_STATUS_NONFINAL_LAST,
- );
- withdrawalGroupRecords =
- await tx.withdrawalGroups.indexes.byStatus.getAll(keyRange);
- } else {
- withdrawalGroupRecords =
- await tx.withdrawalGroups.indexes.byStatus.getAll();
- }
- for (const wgr of withdrawalGroupRecords) {
- await f(wgr);
- }
-}
-
-async function iterRecordsForDeposit(
- tx: WalletDbReadOnlyTransaction<["depositGroups"]>,
- filter: TransactionRecordFilter,
- f: (r: DepositGroupRecord) => Promise<void>,
-): Promise<void> {
- let dgs: DepositGroupRecord[];
- if (filter.onlyState === "nonfinal") {
- const keyRange = GlobalIDB.KeyRange.bound(
- OPERATION_STATUS_NONFINAL_FIRST,
- OPERATION_STATUS_NONFINAL_LAST,
- );
- dgs = await tx.depositGroups.indexes.byStatus.getAll(keyRange);
- } else {
- dgs = await tx.depositGroups.indexes.byStatus.getAll();
- }
-
- for (const dg of dgs) {
- await f(dg);
- }
-}
-
-async function iterRecordsForDenomLoss(
- tx: WalletDbReadOnlyTransaction<["denomLossEvents"]>,
- filter: TransactionRecordFilter,
- f: (r: DenomLossEventRecord) => Promise<void>,
-): Promise<void> {
- let dgs: DenomLossEventRecord[];
- if (filter.onlyState === "nonfinal") {
- const keyRange = GlobalIDB.KeyRange.bound(
- OPERATION_STATUS_NONFINAL_FIRST,
- OPERATION_STATUS_NONFINAL_LAST,
- );
- dgs = await tx.denomLossEvents.indexes.byStatus.getAll(keyRange);
- } else {
- dgs = await tx.denomLossEvents.indexes.byStatus.getAll();
- }
-
- for (const dg of dgs) {
- await f(dg);
- }
-}
-
-async function iterRecordsForRefund(
- tx: WalletDbReadOnlyTransaction<["refundGroups"]>,
- filter: TransactionRecordFilter,
- f: (r: RefundGroupRecord) => Promise<void>,
-): Promise<void> {
- if (filter.onlyState === "nonfinal") {
- const keyRange = GlobalIDB.KeyRange.bound(
- OPERATION_STATUS_NONFINAL_FIRST,
- OPERATION_STATUS_NONFINAL_LAST,
- );
- await tx.refundGroups.indexes.byStatus.iter(keyRange).forEachAsync(f);
- } else {
- await tx.refundGroups.iter().forEachAsync(f);
- }
-}
-
-async function iterRecordsForPurchase(
- tx: WalletDbReadOnlyTransaction<["purchases"]>,
- filter: TransactionRecordFilter,
- f: (r: PurchaseRecord) => Promise<void>,
-): Promise<void> {
- if (filter.onlyState === "nonfinal") {
- const keyRange = GlobalIDB.KeyRange.bound(
- OPERATION_STATUS_NONFINAL_FIRST,
- OPERATION_STATUS_NONFINAL_LAST,
- );
- await tx.purchases.indexes.byStatus.iter(keyRange).forEachAsync(f);
- } else {
- await tx.purchases.indexes.byStatus.iter().forEachAsync(f);
- }
-}
-
-async function iterRecordsForPeerPullCredit(
- tx: WalletDbReadOnlyTransaction<["peerPullCredit"]>,
- filter: TransactionRecordFilter,
- f: (r: PeerPullCreditRecord) => Promise<void>,
-): Promise<void> {
- if (filter.onlyState === "nonfinal") {
- const keyRange = GlobalIDB.KeyRange.bound(
- OPERATION_STATUS_NONFINAL_FIRST,
- OPERATION_STATUS_NONFINAL_LAST,
- );
- await tx.peerPullCredit.indexes.byStatus.iter(keyRange).forEachAsync(f);
- } else {
- await tx.peerPullCredit.indexes.byStatus.iter().forEachAsync(f);
- }
-}
-
-async function iterRecordsForPeerPullDebit(
- tx: WalletDbReadOnlyTransaction<["peerPullDebit"]>,
- filter: TransactionRecordFilter,
- f: (r: PeerPullPaymentIncomingRecord) => Promise<void>,
-): Promise<void> {
- if (filter.onlyState === "nonfinal") {
- const keyRange = GlobalIDB.KeyRange.bound(
- OPERATION_STATUS_NONFINAL_FIRST,
- OPERATION_STATUS_NONFINAL_LAST,
- );
- await tx.peerPullDebit.indexes.byStatus.iter(keyRange).forEachAsync(f);
- } else {
- await tx.peerPullDebit.indexes.byStatus.iter().forEachAsync(f);
- }
-}
-
-async function iterRecordsForPeerPushDebit(
- tx: WalletDbReadOnlyTransaction<["peerPushDebit"]>,
- filter: TransactionRecordFilter,
- f: (r: PeerPushDebitRecord) => Promise<void>,
-): Promise<void> {
- if (filter.onlyState === "nonfinal") {
- const keyRange = GlobalIDB.KeyRange.bound(
- OPERATION_STATUS_NONFINAL_FIRST,
- OPERATION_STATUS_NONFINAL_LAST,
- );
- await tx.peerPushDebit.indexes.byStatus.iter(keyRange).forEachAsync(f);
- } else {
- await tx.peerPushDebit.indexes.byStatus.iter().forEachAsync(f);
- }
-}
-
-async function iterRecordsForPeerPushCredit(
- tx: WalletDbReadOnlyTransaction<["peerPushCredit"]>,
- filter: TransactionRecordFilter,
- f: (r: PeerPushPaymentIncomingRecord) => Promise<void>,
-): Promise<void> {
- if (filter.onlyState === "nonfinal") {
- const keyRange = GlobalIDB.KeyRange.bound(
- OPERATION_STATUS_NONFINAL_FIRST,
- OPERATION_STATUS_NONFINAL_LAST,
- );
- await tx.peerPushCredit.indexes.byStatus.iter(keyRange).forEachAsync(f);
- } else {
- await tx.peerPushCredit.indexes.byStatus.iter().forEachAsync(f);
- }
-}
diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
@@ -336,6 +336,7 @@ import {
getTransactionById,
getTransactions,
parseTransactionIdentifier,
+ rematerializeTransactions,
restartAll as restartAllRunningTasks,
resumeTransaction,
retryAll,
@@ -427,6 +428,41 @@ async function fillDefaults(wex: WalletExecutionContext): Promise<void> {
}
}
+/**
+ * Incremented each time we want to re-materialize transactions.
+ */
+const MATERIALIZED_TRANSACTIONS_VERSION = 1;
+
+async function migrateMaterializedTransactions(
+ wex: WalletExecutionContext,
+): Promise<void> {
+ await wex.db.runAllStoresReadWriteTx({}, async (tx) => {
+ const ver = await tx.config.get("materializedTransactionsVersion");
+ if (ver) {
+ if (ver.key !== ConfigRecordKey.MaterializedTransactionsVersion) {
+ logger.error("invalid configuration (materializedTransactionsVersion)");
+ return;
+ }
+ if (ver.value == MATERIALIZED_TRANSACTIONS_VERSION) {
+ return;
+ }
+ if (ver.value > MATERIALIZED_TRANSACTIONS_VERSION) {
+ logger.error(
+ "database is newer than code (materializedTransactionsVersion)",
+ );
+ return;
+ }
+ }
+
+ await rematerializeTransactions(wex, tx);
+
+ await tx.config.put({
+ key: ConfigRecordKey.MaterializedTransactionsVersion,
+ value: MATERIALIZED_TRANSACTIONS_VERSION,
+ });
+ });
+}
+
export async function getDenomInfo(
wex: WalletExecutionContext,
tx: WalletDbReadOnlyTransaction<["denominations"]>,
@@ -707,6 +743,9 @@ async function recoverStoredBackup(
});
logger.info(`backup found, now importing`);
await importDb(wex.db.idbHandle(), bd);
+ await wex.db.runAllStoresReadWriteTx({}, async (tx) => {
+ await rematerializeTransactions(wex, tx);
+ });
logger.info(`import done`);
}
@@ -796,6 +835,9 @@ async function handleSetWalletRunConfig(
logger.trace("filling defaults");
await fillDefaults(wex);
}
+
+ await migrateMaterializedTransactions(wex);
+
const resp: InitResponse = {
versionInfo: await handleGetVersion(wex),
};
@@ -1894,7 +1936,11 @@ const handlers: { [T in WalletApiOperation]: HandlerWithValidator<T> } = {
[WalletApiOperation.ImportDb]: {
codec: codecForImportDbRequest(),
handler: async (wex, req) => {
+ // FIXME: This should atomically re-materialize transactions!
await importDb(wex.db.idbHandle(), req.dump);
+ await wex.db.runAllStoresReadWriteTx({}, async (tx) => {
+ await rematerializeTransactions(wex, tx);
+ });
return {};
},
},
