commit df1e6cc2ad3203c7464ddc170ce3d7ff37b2821f
parent 3ae43d6cd81ce62d112c9352f91c35afe3fd09ac
Author: Florian Dold <dold@taler.net>
Date: Sun, 19 Jul 2026 01:35:01 +0200
db: move shepherd.ts behind the DAL
Diffstat:
3 files changed, 54 insertions(+), 44 deletions(-)
diff --git a/packages/taler-wallet-core/src/dbtx-indexeddb.ts b/packages/taler-wallet-core/src/dbtx-indexeddb.ts
@@ -58,6 +58,7 @@ import {
WalletCoin,
WalletDepositGroup,
WalletRecoupGroup,
+ PurchaseStatus,
WalletReserve,
WalletRefreshGroup,
WalletRefreshSession,
@@ -1509,6 +1510,18 @@ export class IdbWalletTransaction implements WalletDbTransaction {
);
}
+ async getActiveRecoupGroups(): Promise<WalletRecoupGroup[]> {
+ return await this.tx.recoupGroups.indexes.byStatus.getAll(
+ getActiveKeyRange(),
+ );
+ }
+
+ async getPurchasesByStatus(
+ status: PurchaseStatus,
+ ): Promise<WalletPurchase[]> {
+ return await this.tx.purchases.indexes.byStatus.getAll(status);
+ }
+
async getActivePurchases(): Promise<WalletPurchase[]> {
return await this.tx.purchases.indexes.byStatus.getAll(getActiveKeyRange());
}
diff --git a/packages/taler-wallet-core/src/dbtx.ts b/packages/taler-wallet-core/src/dbtx.ts
@@ -58,6 +58,7 @@ import {
WalletCoin,
WalletDepositGroup,
WalletRecoupGroup,
+ PurchaseStatus,
WalletReserve,
WalletRefreshGroup,
WalletRefreshSession,
@@ -778,6 +779,13 @@ export interface WalletDbTransaction {
getActivePeerPullDebits(): Promise<WalletPeerPullDebit[]>;
+ getActiveRecoupGroups(): Promise<WalletRecoupGroup[]>;
+
+ /**
+ * Get all purchases in a specific status.
+ */
+ getPurchasesByStatus(status: PurchaseStatus): Promise<WalletPurchase[]>;
+
getActivePurchases(): Promise<WalletPurchase[]>;
getCoinsByPubs(coinPubs: string[]): Promise<WalletCoin[]>;
diff --git a/packages/taler-wallet-core/src/shepherd.ts b/packages/taler-wallet-core/src/shepherd.ts
@@ -59,9 +59,7 @@ import {
timestampPreciseToDb,
WalletOperationRetry,
} from "./db-common.js";
-import {
- WalletIndexedDbTransaction,
-} from "./db-indexeddb.js";
+import { WalletIndexedDbTransaction } from "./db-indexeddb.js";
import { processValidateDenoms } from "./denominations.js";
import {
computeDepositTransactionStatus,
@@ -114,6 +112,7 @@ import {
computeWithdrawalTransactionStatus,
processWithdrawalGroup,
} from "./withdraw.js";
+import { WalletDbTransaction } from "./dbtx.js";
const logger = new Logger("shepherd.ts");
@@ -429,7 +428,7 @@ export class TaskSchedulerImpl implements TaskScheduler {
await wtx.deleteOperationRetry(taskId);
const notif = await taskToRetryNotification(
this.ws,
- tx,
+ wtx,
taskId,
undefined,
);
@@ -635,7 +634,7 @@ async function storePendingTaskError(
}
await wtx.upsertOperationRetry(retryRecord);
return {
- notification: await taskToRetryNotification(ws, tx, pendingTaskId, e),
+ notification: await taskToRetryNotification(ws, wtx, pendingTaskId, e),
retryRecord,
};
});
@@ -688,7 +687,7 @@ async function storePendingTaskPending(
if (hadError) {
const notif = await taskToRetryNotification(
ws,
- tx,
+ wtx,
pendingTaskId,
undefined,
);
@@ -809,7 +808,7 @@ async function callOperationHandlerForTaskId(
*/
async function taskToRetryNotification(
ws: InternalWalletState,
- tx: WalletIndexedDbTransaction,
+ tx: WalletDbTransaction,
pendingTaskId: string,
e: TalerErrorDetail | undefined,
): Promise<WalletNotification | undefined> {
@@ -840,7 +839,7 @@ async function taskToRetryNotification(
async function getTransactionState(
ws: InternalWalletState,
- tx: WalletIndexedDbTransaction,
+ tx: WalletDbTransaction,
transactionId: string,
): Promise<{ txState: TransactionState; stId: number } | undefined> {
const parsedTxId = parseTransactionIdentifier(transactionId);
@@ -849,7 +848,7 @@ async function getTransactionState(
}
switch (parsedTxId.tag) {
case TransactionType.Deposit: {
- const rec = await tx.depositGroups.get(parsedTxId.depositGroupId);
+ const rec = await tx.getDepositGroup(parsedTxId.depositGroupId);
if (!rec) {
return undefined;
}
@@ -860,7 +859,7 @@ async function getTransactionState(
}
case TransactionType.InternalWithdrawal:
case TransactionType.Withdrawal: {
- const rec = await tx.withdrawalGroups.get(parsedTxId.withdrawalGroupId);
+ const rec = await tx.getWithdrawalGroup(parsedTxId.withdrawalGroupId);
if (!rec) {
return undefined;
}
@@ -870,7 +869,7 @@ async function getTransactionState(
};
}
case TransactionType.Payment: {
- const rec = await tx.purchases.get(parsedTxId.proposalId);
+ const rec = await tx.getPurchase(parsedTxId.proposalId);
if (!rec) {
return;
}
@@ -880,7 +879,7 @@ async function getTransactionState(
};
}
case TransactionType.Refund: {
- const rec = await tx.refundGroups.get(parsedTxId.refundGroupId);
+ const rec = await tx.getRefundGroup(parsedTxId.refundGroupId);
if (!rec) {
return undefined;
}
@@ -890,7 +889,7 @@ async function getTransactionState(
};
}
case TransactionType.PeerPullCredit: {
- const rec = await tx.peerPullCredit.get(parsedTxId.pursePub);
+ const rec = await tx.getPeerPullCredit(parsedTxId.pursePub);
if (!rec) {
return undefined;
}
@@ -900,7 +899,7 @@ async function getTransactionState(
};
}
case TransactionType.PeerPullDebit: {
- const rec = await tx.peerPullDebit.get(parsedTxId.peerPullDebitId);
+ const rec = await tx.getPeerPullDebit(parsedTxId.peerPullDebitId);
if (!rec) {
return undefined;
}
@@ -910,7 +909,7 @@ async function getTransactionState(
};
}
case TransactionType.PeerPushCredit: {
- const rec = await tx.peerPushCredit.get(parsedTxId.peerPushCreditId);
+ const rec = await tx.getPeerPushCredit(parsedTxId.peerPushCreditId);
if (!rec) {
return undefined;
}
@@ -920,7 +919,7 @@ async function getTransactionState(
};
}
case TransactionType.PeerPushDebit: {
- const rec = await tx.peerPushDebit.get(parsedTxId.pursePub);
+ const rec = await tx.getPeerPushDebit(parsedTxId.pursePub);
if (!rec) {
return undefined;
}
@@ -930,7 +929,7 @@ async function getTransactionState(
};
}
case TransactionType.Refresh: {
- const rec = await tx.refreshGroups.get(parsedTxId.refreshGroupId);
+ const rec = await tx.getRefreshGroup(parsedTxId.refreshGroupId);
if (!rec) {
return undefined;
}
@@ -942,7 +941,7 @@ async function getTransactionState(
case TransactionType.Recoup:
throw Error("not yet supported");
case TransactionType.DenomLoss: {
- const rec = await tx.denomLossEvents.get(parsedTxId.denomLossEventId);
+ const rec = await tx.getDenomLossEvent(parsedTxId.denomLossEventId);
if (!rec) {
return undefined;
}
@@ -958,7 +957,7 @@ async function getTransactionState(
async function makeTransactionRetryNotification(
ws: InternalWalletState,
- tx: WalletIndexedDbTransaction,
+ tx: WalletDbTransaction,
pendingTaskId: string,
e: TalerErrorDetail | undefined,
): Promise<WalletNotification | undefined> {
@@ -988,7 +987,7 @@ async function makeTransactionRetryNotification(
async function makeExchangeRetryNotification(
ws: InternalWalletState,
- tx: WalletIndexedDbTransaction,
+ tx: WalletDbTransaction,
pendingTaskId: string,
e: TalerErrorDetail | undefined,
): Promise<WalletNotification | undefined> {
@@ -1001,7 +1000,7 @@ async function makeExchangeRetryNotification(
default:
throw Error("invalid task identifier");
}
- const rec = await tx.exchanges.get(parsedTaskId.exchangeBaseUrl);
+ const rec = await tx.getExchange(parsedTaskId.exchangeBaseUrl);
if (!rec) {
logger.info(`exchange ${parsedTaskId.exchangeBaseUrl} not found`);
@@ -1091,17 +1090,11 @@ export async function getActiveTaskIds(
const res: ActiveTaskIdsResult = {
taskIds: [],
};
- await ws.runStandaloneLegacyWalletDbTx(async (tx) => {
- const active = GlobalIDB.KeyRange.bound(
- OPERATION_STATUS_NONFINAL_FIRST,
- OPERATION_STATUS_NONFINAL_LAST,
- );
-
+ await ws.runStandaloneLegacyWalletDbTx(async (tx, wtx) => {
// Withdrawals
{
- const activeRecs =
- await tx.withdrawalGroups.indexes.byStatus.getAll(active);
+ const activeRecs = await wtx.getActiveWithdrawalGroups();
for (const rec of activeRecs) {
const taskId = constructTaskIdentifier({
tag: PendingTaskType.Withdraw,
@@ -1114,7 +1107,7 @@ export async function getActiveTaskIds(
// Deposits
{
- const activeRecs = await tx.depositGroups.indexes.byStatus.getAll(active);
+ const activeRecs = await wtx.getActiveDepositGroups();
for (const rec of activeRecs) {
const taskId = constructTaskIdentifier({
tag: PendingTaskType.Deposit,
@@ -1127,7 +1120,7 @@ export async function getActiveTaskIds(
// Refreshes
{
- const activeRecs = await tx.refreshGroups.indexes.byStatus.getAll(active);
+ const activeRecs = await wtx.getActiveRefreshGroups();
for (const rec of activeRecs) {
const taskId = constructTaskIdentifier({
tag: PendingTaskType.Refresh,
@@ -1140,7 +1133,7 @@ export async function getActiveTaskIds(
// Purchases
{
- const activeRecs = await tx.purchases.indexes.byStatus.getAll(active);
+ const activeRecs = await wtx.getActivePurchases();
for (const rec of activeRecs) {
const taskId = constructTaskIdentifier({
tag: PendingTaskType.Purchase,
@@ -1153,7 +1146,7 @@ export async function getActiveTaskIds(
// peer-push-debit
{
- const activeRecs = await tx.peerPushDebit.indexes.byStatus.getAll(active);
+ const activeRecs = await wtx.getActivePeerPushDebits();
for (const rec of activeRecs) {
const taskId = constructTaskIdentifier({
tag: PendingTaskType.PeerPushDebit,
@@ -1166,8 +1159,7 @@ export async function getActiveTaskIds(
// peer-push-credit
{
- const activeRecs =
- await tx.peerPushCredit.indexes.byStatus.getAll(active);
+ const activeRecs = await wtx.getActivePeerPushCredits();
for (const rec of activeRecs) {
const taskId = constructTaskIdentifier({
tag: PendingTaskType.PeerPushCredit,
@@ -1180,7 +1172,7 @@ export async function getActiveTaskIds(
// peer-pull-debit
{
- const activeRecs = await tx.peerPullDebit.indexes.byStatus.getAll(active);
+ const activeRecs = await wtx.getActivePeerPullDebits();
for (const rec of activeRecs) {
const taskId = constructTaskIdentifier({
tag: PendingTaskType.PeerPullDebit,
@@ -1193,8 +1185,7 @@ export async function getActiveTaskIds(
// peer-pull-credit
{
- const activeRecs =
- await tx.peerPullCredit.indexes.byStatus.getAll(active);
+ const activeRecs = await wtx.getActivePeerPullCredits();
for (const rec of activeRecs) {
const taskId = constructTaskIdentifier({
tag: PendingTaskType.PeerPullCredit,
@@ -1207,7 +1198,7 @@ export async function getActiveTaskIds(
// recoup
{
- const activeRecs = await tx.recoupGroups.indexes.byStatus.getAll(active);
+ const activeRecs = await wtx.getActiveRecoupGroups();
for (const rec of activeRecs) {
const taskId = constructTaskIdentifier({
tag: PendingTaskType.Recoup,
@@ -1220,7 +1211,7 @@ export async function getActiveTaskIds(
// exchange update and KYC
{
- const exchanges = await tx.exchanges.getAll();
+ const exchanges = await wtx.getExchanges();
for (const rec of exchanges) {
const taskIdUpdate = constructTaskIdentifier({
tag: PendingTaskType.ExchangeUpdate,
@@ -1238,7 +1229,7 @@ export async function getActiveTaskIds(
if (reserveId == null) {
continue;
}
- const reserveRec = await tx.reserves.get(reserveId);
+ const reserveRec = await wtx.getReserve(reserveId);
if (
reserveRec?.status != null &&
reserveRec.status != ReserveRecordStatus.Done
@@ -1268,9 +1259,7 @@ export async function processCleanupExpiredTransactions(
wex: WalletExecutionContext,
): Promise<TaskRunResult> {
await wex.runLegacyWalletDbTx(async (tx) => {
- const expired = await tx.purchases.indexes.byStatus.getAll(
- PurchaseStatus.Expired,
- );
+ const expired = await tx.wtx.getPurchasesByStatus(PurchaseStatus.Expired);
for (const exp of expired) {
if (!exp.timestampExpired) {
continue;
