/*
This file is part of GNU Taler
(C) 2022 GNUnet e.V.
GNU Taler is free software; you can redistribute it and/or modify it under the
terms of the GNU General Public License as published by the Free Software
Foundation; either version 3, or (at your option) any later version.
GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY
WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
A PARTICULAR PURPOSE. See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License along with
GNU Taler; see the file COPYING. If not, see
*/
/**
* Imports.
*/
import {
AbsoluteTime,
AgeRestriction,
AmountJson,
Amounts,
CancellationToken,
CoinRefreshRequest,
CoinStatus,
Duration,
ExchangeEntryState,
ExchangeEntryStatus,
ExchangeListItem,
ExchangeTosStatus,
ExchangeUpdateStatus,
getErrorDetailFromException,
j2s,
Logger,
makeErrorDetail,
NotificationType,
OperationErrorInfo,
RefreshReason,
ScopeInfo,
ScopeType,
TalerError,
TalerErrorCode,
TalerErrorDetail,
TalerPreciseTimestamp,
TombstoneIdStr,
TransactionIdStr,
TransactionType,
WalletNotification,
} from "@gnu-taler/taler-util";
import { CryptoApiStoppedError } from "../crypto/workers/crypto-dispatcher.js";
import {
BackupProviderRecord,
CoinRecord,
DbPreciseTimestamp,
DepositGroupRecord,
ExchangeDetailsRecord,
ExchangeEntryDbRecordStatus,
ExchangeEntryDbUpdateStatus,
ExchangeEntryRecord,
PeerPullCreditRecord,
PeerPullPaymentIncomingRecord,
PeerPushDebitRecord,
PeerPushPaymentIncomingRecord,
PurchaseRecord,
RecoupGroupRecord,
RefreshGroupRecord,
RewardRecord,
timestampPreciseToDb,
WalletStoresV1,
WithdrawalGroupRecord,
} from "../db.js";
import { InternalWalletState } from "../internal-wallet-state.js";
import { PendingTaskType, TaskId } from "../pending-types.js";
import { assertUnreachable } from "../util/assertUnreachable.js";
import { checkDbInvariant, checkLogicInvariant } from "../util/invariants.js";
import { GetReadOnlyAccess, GetReadWriteAccess } from "../util/query.js";
import {
constructTransactionIdentifier,
parseTransactionIdentifier,
} from "./transactions.js";
const logger = new Logger("operations/common.ts");
export interface CoinsSpendInfo {
coinPubs: string[];
contributions: AmountJson[];
refreshReason: RefreshReason;
/**
* Identifier for what the coin has been spent for.
*/
allocationId: TransactionIdStr;
}
export async function makeCoinsVisible(
ws: InternalWalletState,
tx: GetReadWriteAccess<{
coins: typeof WalletStoresV1.coins;
coinAvailability: typeof WalletStoresV1.coinAvailability;
}>,
transactionId: string,
): Promise {
const coins =
await tx.coins.indexes.bySourceTransactionId.getAll(transactionId);
for (const coinRecord of coins) {
if (!coinRecord.visible) {
coinRecord.visible = 1;
await tx.coins.put(coinRecord);
const ageRestriction = coinRecord.maxAge;
const car = await tx.coinAvailability.get([
coinRecord.exchangeBaseUrl,
coinRecord.denomPubHash,
ageRestriction,
]);
if (!car) {
logger.error("missing coin availability record");
continue;
}
const visCount = car.visibleCoinCount ?? 0;
car.visibleCoinCount = visCount + 1;
await tx.coinAvailability.put(car);
}
}
}
export async function makeCoinAvailable(
ws: InternalWalletState,
tx: GetReadWriteAccess<{
coins: typeof WalletStoresV1.coins;
coinAvailability: typeof WalletStoresV1.coinAvailability;
denominations: typeof WalletStoresV1.denominations;
}>,
coinRecord: CoinRecord,
): Promise {
checkLogicInvariant(coinRecord.status === CoinStatus.Fresh);
const existingCoin = await tx.coins.get(coinRecord.coinPub);
if (existingCoin) {
return;
}
const denom = await tx.denominations.get([
coinRecord.exchangeBaseUrl,
coinRecord.denomPubHash,
]);
checkDbInvariant(!!denom);
const ageRestriction = coinRecord.maxAge;
let car = await tx.coinAvailability.get([
coinRecord.exchangeBaseUrl,
coinRecord.denomPubHash,
ageRestriction,
]);
if (!car) {
car = {
maxAge: ageRestriction,
value: denom.value,
currency: denom.currency,
denomPubHash: denom.denomPubHash,
exchangeBaseUrl: denom.exchangeBaseUrl,
freshCoinCount: 0,
visibleCoinCount: 0,
};
}
car.freshCoinCount++;
await tx.coins.put(coinRecord);
await tx.coinAvailability.put(car);
}
export async function spendCoins(
ws: InternalWalletState,
tx: GetReadWriteAccess<{
coins: typeof WalletStoresV1.coins;
coinAvailability: typeof WalletStoresV1.coinAvailability;
refreshGroups: typeof WalletStoresV1.refreshGroups;
denominations: typeof WalletStoresV1.denominations;
}>,
csi: CoinsSpendInfo,
): Promise {
if (csi.coinPubs.length != csi.contributions.length) {
throw Error("assertion failed");
}
if (csi.coinPubs.length === 0) {
return;
}
let refreshCoinPubs: CoinRefreshRequest[] = [];
for (let i = 0; i < csi.coinPubs.length; i++) {
const coin = await tx.coins.get(csi.coinPubs[i]);
if (!coin) {
throw Error("coin allocated for payment doesn't exist anymore");
}
const denom = await ws.getDenomInfo(
ws,
tx,
coin.exchangeBaseUrl,
coin.denomPubHash,
);
checkDbInvariant(!!denom);
const coinAvailability = await tx.coinAvailability.get([
coin.exchangeBaseUrl,
coin.denomPubHash,
coin.maxAge,
]);
checkDbInvariant(!!coinAvailability);
const contrib = csi.contributions[i];
if (coin.status !== CoinStatus.Fresh) {
const alloc = coin.spendAllocation;
if (!alloc) {
continue;
}
if (alloc.id !== csi.allocationId) {
// FIXME: assign error code
logger.info("conflicting coin allocation ID");
logger.info(`old ID: ${alloc.id}, new ID: ${csi.allocationId}`);
throw Error("conflicting coin allocation (id)");
}
if (0 !== Amounts.cmp(alloc.amount, contrib)) {
// FIXME: assign error code
throw Error("conflicting coin allocation (contrib)");
}
continue;
}
coin.status = CoinStatus.Dormant;
coin.spendAllocation = {
id: csi.allocationId,
amount: Amounts.stringify(contrib),
};
const remaining = Amounts.sub(denom.value, contrib);
if (remaining.saturated) {
throw Error("not enough remaining balance on coin for payment");
}
refreshCoinPubs.push({
amount: Amounts.stringify(remaining.amount),
coinPub: coin.coinPub,
});
checkDbInvariant(!!coinAvailability);
if (coinAvailability.freshCoinCount === 0) {
throw Error(
`invalid coin count ${coinAvailability.freshCoinCount} in DB`,
);
}
coinAvailability.freshCoinCount--;
if (coin.visible) {
if (!coinAvailability.visibleCoinCount) {
logger.error("coin availability inconsistent");
} else {
coinAvailability.visibleCoinCount--;
}
}
await tx.coins.put(coin);
await tx.coinAvailability.put(coinAvailability);
}
await ws.refreshOps.createRefreshGroup(
ws,
tx,
Amounts.currencyOf(csi.contributions[0]),
refreshCoinPubs,
csi.refreshReason,
{
originatingTransactionId: csi.allocationId,
},
);
}
/**
* Convert the task ID for a task that processes a transaction int
* the ID for the transaction.
*/
function convertTaskToTransactionId(
taskId: string,
): TransactionIdStr | undefined {
const parsedTaskId = parseTaskIdentifier(taskId);
switch (parsedTaskId.tag) {
case PendingTaskType.PeerPullCredit:
return constructTransactionIdentifier({
tag: TransactionType.PeerPullCredit,
pursePub: parsedTaskId.pursePub,
});
case PendingTaskType.PeerPullDebit:
return constructTransactionIdentifier({
tag: TransactionType.PeerPullDebit,
peerPullDebitId: parsedTaskId.peerPullDebitId,
});
// FIXME: This doesn't distinguish internal-withdrawal.
// Maybe we should have a different task type for that as well?
// Or maybe transaction IDs should be valid task identifiers?
case PendingTaskType.Withdraw:
return constructTransactionIdentifier({
tag: TransactionType.Withdrawal,
withdrawalGroupId: parsedTaskId.withdrawalGroupId,
});
case PendingTaskType.PeerPushCredit:
return constructTransactionIdentifier({
tag: TransactionType.PeerPushCredit,
peerPushCreditId: parsedTaskId.peerPushCreditId,
});
case PendingTaskType.Deposit:
return constructTransactionIdentifier({
tag: TransactionType.Deposit,
depositGroupId: parsedTaskId.depositGroupId,
});
case PendingTaskType.Refresh:
return constructTransactionIdentifier({
tag: TransactionType.Refresh,
refreshGroupId: parsedTaskId.refreshGroupId,
});
case PendingTaskType.RewardPickup:
return constructTransactionIdentifier({
tag: TransactionType.Reward,
walletRewardId: parsedTaskId.walletRewardId,
});
case PendingTaskType.PeerPushDebit:
return constructTransactionIdentifier({
tag: TransactionType.PeerPushDebit,
pursePub: parsedTaskId.pursePub,
});
case PendingTaskType.Purchase:
return constructTransactionIdentifier({
tag: TransactionType.Payment,
proposalId: parsedTaskId.proposalId,
});
default:
return undefined;
}
}
async function makeTransactionRetryNotification(
ws: InternalWalletState,
tx: GetReadOnlyAccess,
pendingTaskId: string,
e: TalerErrorDetail | undefined,
): Promise {
const txId = convertTaskToTransactionId(pendingTaskId);
if (!txId) {
return undefined;
}
const txState = await ws.getTransactionState(ws, tx, txId);
if (!txState) {
return undefined;
}
const notif: WalletNotification = {
type: NotificationType.TransactionStateTransition,
transactionId: txId,
oldTxState: txState,
newTxState: txState,
};
if (e) {
notif.errorInfo = {
code: e.code as number,
hint: e.hint,
};
}
return notif;
}
async function makeExchangeRetryNotification(
ws: InternalWalletState,
tx: GetReadOnlyAccess,
pendingTaskId: string,
e: TalerErrorDetail | undefined,
): Promise {
logger.info("making exchange retry notification");
const parsedTaskId = parseTaskIdentifier(pendingTaskId);
if (parsedTaskId.tag !== PendingTaskType.ExchangeUpdate) {
throw Error("invalid task identifier");
}
const rec = await tx.exchanges.get(parsedTaskId.exchangeBaseUrl);
if (!rec) {
logger.info(`exchange ${parsedTaskId.exchangeBaseUrl} not found`);
return undefined;
}
const notif: WalletNotification = {
type: NotificationType.ExchangeStateTransition,
exchangeBaseUrl: parsedTaskId.exchangeBaseUrl,
oldExchangeState: getExchangeState(rec),
newExchangeState: getExchangeState(rec),
};
if (e) {
notif.errorInfo = {
code: e.code as number,
hint: e.hint,
};
}
return notif;
}
/**
* Generate an appropriate error transition notification
* for applicable tasks.
*
* Namely, transition notifications are generated for:
* - exchange update errors
* - transactions
*/
async function taskToRetryNotification(
ws: InternalWalletState,
tx: GetReadOnlyAccess,
pendingTaskId: string,
e: TalerErrorDetail | undefined,
): Promise {
const parsedTaskId = parseTaskIdentifier(pendingTaskId);
switch (parsedTaskId.tag) {
case PendingTaskType.ExchangeUpdate:
return makeExchangeRetryNotification(ws, tx, pendingTaskId, e);
case PendingTaskType.PeerPullCredit:
case PendingTaskType.PeerPullDebit:
case PendingTaskType.Withdraw:
case PendingTaskType.PeerPushCredit:
case PendingTaskType.Deposit:
case PendingTaskType.Refresh:
case PendingTaskType.RewardPickup:
case PendingTaskType.PeerPushDebit:
case PendingTaskType.Purchase:
return makeTransactionRetryNotification(ws, tx, pendingTaskId, e);
case PendingTaskType.Backup:
case PendingTaskType.ExchangeCheckRefresh:
case PendingTaskType.Recoup:
return undefined;
}
}
async function storePendingTaskError(
ws: InternalWalletState,
pendingTaskId: string,
e: TalerErrorDetail,
): Promise {
logger.info(`storing pending task error for ${pendingTaskId}`);
const maybeNotification = await ws.db.mktxAll().runReadWrite(async (tx) => {
let retryRecord = await tx.operationRetries.get(pendingTaskId);
if (!retryRecord) {
retryRecord = {
id: pendingTaskId,
lastError: e,
retryInfo: DbRetryInfo.reset(),
};
} else {
retryRecord.lastError = e;
retryRecord.retryInfo = DbRetryInfo.increment(retryRecord.retryInfo);
}
await tx.operationRetries.put(retryRecord);
return taskToRetryNotification(ws, tx, pendingTaskId, e);
});
if (maybeNotification) {
ws.notify(maybeNotification);
}
}
export async function resetPendingTaskTimeout(
ws: InternalWalletState,
pendingTaskId: string,
): Promise {
const maybeNotification = await ws.db.mktxAll().runReadWrite(async (tx) => {
let retryRecord = await tx.operationRetries.get(pendingTaskId);
if (retryRecord) {
// Note that we don't reset the lastError, it should still be visible
// while the retry runs.
retryRecord.retryInfo = DbRetryInfo.reset();
await tx.operationRetries.put(retryRecord);
}
return taskToRetryNotification(ws, tx, pendingTaskId, undefined);
});
if (maybeNotification) {
ws.notify(maybeNotification);
}
}
async function storePendingTaskPending(
ws: InternalWalletState,
pendingTaskId: string,
): Promise {
const maybeNotification = await ws.db.mktxAll().runReadWrite(async (tx) => {
let retryRecord = await tx.operationRetries.get(pendingTaskId);
let hadError = false;
if (!retryRecord) {
retryRecord = {
id: pendingTaskId,
retryInfo: DbRetryInfo.reset(),
};
} else {
if (retryRecord.lastError) {
hadError = true;
}
delete retryRecord.lastError;
retryRecord.retryInfo = DbRetryInfo.increment(retryRecord.retryInfo);
}
await tx.operationRetries.put(retryRecord);
if (hadError) {
return taskToRetryNotification(ws, tx, pendingTaskId, undefined);
} else {
return undefined;
}
});
if (maybeNotification) {
ws.notify(maybeNotification);
}
}
async function storePendingTaskFinished(
ws: InternalWalletState,
pendingTaskId: string,
): Promise {
await ws.db
.mktx((x) => [x.operationRetries])
.runReadWrite(async (tx) => {
await tx.operationRetries.delete(pendingTaskId);
});
}
export async function runTaskWithErrorReporting(
ws: InternalWalletState,
opId: TaskId,
f: () => Promise,
): Promise {
let maybeError: TalerErrorDetail | undefined;
try {
const resp = await f();
switch (resp.type) {
case TaskRunResultType.Error:
await storePendingTaskError(ws, opId, resp.errorDetail);
return resp;
case TaskRunResultType.Finished:
await storePendingTaskFinished(ws, opId);
return resp;
case TaskRunResultType.Pending:
await storePendingTaskPending(ws, opId);
return resp;
case TaskRunResultType.Longpoll:
return resp;
}
} catch (e) {
if (e instanceof CryptoApiStoppedError) {
if (ws.stopped) {
logger.warn("crypto API stopped during shutdown, ignoring error");
return {
type: TaskRunResultType.Error,
errorDetail: makeErrorDetail(
TalerErrorCode.WALLET_UNEXPECTED_EXCEPTION,
{},
"Crypto API stopped during shutdown",
),
};
}
}
if (e instanceof TalerError) {
logger.warn("operation processed resulted in error");
logger.warn(`error was: ${j2s(e.errorDetail)}`);
maybeError = e.errorDetail;
await storePendingTaskError(ws, opId, maybeError!);
return {
type: TaskRunResultType.Error,
errorDetail: e.errorDetail,
};
} else if (e instanceof Error) {
// This is a bug, as we expect pending operations to always
// do their own error handling and only throw WALLET_PENDING_OPERATION_FAILED
// or return something.
logger.error(`Uncaught exception: ${e.message}`);
logger.error(`Stack: ${e.stack}`);
maybeError = makeErrorDetail(
TalerErrorCode.WALLET_UNEXPECTED_EXCEPTION,
{
stack: e.stack,
},
`unexpected exception (message: ${e.message})`,
);
await storePendingTaskError(ws, opId, maybeError);
return {
type: TaskRunResultType.Error,
errorDetail: maybeError,
};
} else {
logger.error("Uncaught exception, value is not even an error.");
maybeError = makeErrorDetail(
TalerErrorCode.WALLET_UNEXPECTED_EXCEPTION,
{},
`unexpected exception (not even an error)`,
);
await storePendingTaskError(ws, opId, maybeError);
return {
type: TaskRunResultType.Error,
errorDetail: maybeError,
};
}
}
}
export enum TombstoneTag {
DeleteWithdrawalGroup = "delete-withdrawal-group",
DeleteReserve = "delete-reserve",
DeletePayment = "delete-payment",
DeleteReward = "delete-reward",
DeleteRefreshGroup = "delete-refresh-group",
DeleteDepositGroup = "delete-deposit-group",
DeleteRefund = "delete-refund",
DeletePeerPullDebit = "delete-peer-pull-debit",
DeletePeerPushDebit = "delete-peer-push-debit",
DeletePeerPullCredit = "delete-peer-pull-credit",
DeletePeerPushCredit = "delete-peer-push-credit",
}
export function getExchangeTosStatusFromRecord(
exchange: ExchangeEntryRecord,
): ExchangeTosStatus {
if (!exchange.tosAcceptedEtag) {
return ExchangeTosStatus.Proposed;
}
if (exchange.tosAcceptedEtag == exchange.tosCurrentEtag) {
return ExchangeTosStatus.Accepted;
}
return ExchangeTosStatus.Proposed;
}
export function getExchangeUpdateStatusFromRecord(
r: ExchangeEntryRecord,
): ExchangeUpdateStatus {
switch (r.updateStatus) {
case ExchangeEntryDbUpdateStatus.UnavailableUpdate:
return ExchangeUpdateStatus.UnavailableUpdate;
case ExchangeEntryDbUpdateStatus.Initial:
return ExchangeUpdateStatus.Initial;
case ExchangeEntryDbUpdateStatus.InitialUpdate:
return ExchangeUpdateStatus.InitialUpdate;
case ExchangeEntryDbUpdateStatus.Ready:
return ExchangeUpdateStatus.Ready;
case ExchangeEntryDbUpdateStatus.ReadyUpdate:
return ExchangeUpdateStatus.ReadyUpdate;
case ExchangeEntryDbUpdateStatus.Suspended:
return ExchangeUpdateStatus.Suspended;
}
}
export function getExchangeEntryStatusFromRecord(
r: ExchangeEntryRecord,
): ExchangeEntryStatus {
switch (r.entryStatus) {
case ExchangeEntryDbRecordStatus.Ephemeral:
return ExchangeEntryStatus.Ephemeral;
case ExchangeEntryDbRecordStatus.Preset:
return ExchangeEntryStatus.Preset;
case ExchangeEntryDbRecordStatus.Used:
return ExchangeEntryStatus.Used;
}
}
/**
* Compute the state of an exchange entry from the DB
* record.
*/
export function getExchangeState(r: ExchangeEntryRecord): ExchangeEntryState {
return {
exchangeEntryStatus: getExchangeEntryStatusFromRecord(r),
exchangeUpdateStatus: getExchangeUpdateStatusFromRecord(r),
tosStatus: getExchangeTosStatusFromRecord(r),
};
}
/**
* Mock scope info for an exchange by always returning a regional currency scope.
*/
function mockExchangeScopeInfo(
r: ExchangeEntryRecord,
exchangeDetails: ExchangeDetailsRecord | undefined,
): ScopeInfo | undefined {
const currency = r.presetCurrencyHint ?? exchangeDetails?.currency;
if (currency) {
return {
currency,
type: ScopeType.Exchange,
url: r.baseUrl,
};
}
return undefined;
}
export function makeExchangeListItem(
r: ExchangeEntryRecord,
exchangeDetails: ExchangeDetailsRecord | undefined,
lastError: TalerErrorDetail | undefined,
): ExchangeListItem {
const lastUpdateErrorInfo: OperationErrorInfo | undefined = lastError
? {
error: lastError,
}
: undefined;
return {
exchangeBaseUrl: r.baseUrl,
currency: exchangeDetails?.currency ?? r.presetCurrencyHint,
exchangeUpdateStatus: getExchangeUpdateStatusFromRecord(r),
exchangeEntryStatus: getExchangeEntryStatusFromRecord(r),
tosStatus: getExchangeTosStatusFromRecord(r),
ageRestrictionOptions: exchangeDetails?.ageMask
? AgeRestriction.getAgeGroupsFromMask(exchangeDetails.ageMask)
: [],
paytoUris: exchangeDetails?.wireInfo.accounts.map((x) => x.payto_uri) ?? [],
lastUpdateErrorInfo,
// FIXME: Return real scope info in the future!
scopeInfo: mockExchangeScopeInfo(r, exchangeDetails),
};
}
export interface LongpollResult {
ready: boolean;
}
export function runLongpollAsync(
ws: InternalWalletState,
retryTag: string,
reqFn: (ct: CancellationToken) => Promise,
): void {
const asyncFn = async () => {
if (ws.stopped) {
logger.trace("not long-polling reserve, wallet already stopped");
await storePendingTaskPending(ws, retryTag);
return;
}
const cts = CancellationToken.create();
let res: { ready: boolean } | undefined = undefined;
try {
ws.activeLongpoll[retryTag] = {
cancel: () => {
logger.trace("cancel of reserve longpoll requested");
cts.cancel();
},
};
res = await reqFn(cts.token);
} catch (e) {
const errDetail = getErrorDetailFromException(e);
logger.warn(`got error during long-polling: ${j2s(errDetail)}`);
await storePendingTaskError(ws, retryTag, errDetail);
return;
} finally {
delete ws.activeLongpoll[retryTag];
}
if (!res.ready) {
await storePendingTaskPending(ws, retryTag);
}
ws.workAvailable.trigger();
};
asyncFn();
}
export type ParsedTombstone =
| {
tag: TombstoneTag.DeleteWithdrawalGroup;
withdrawalGroupId: string;
}
| { tag: TombstoneTag.DeleteRefund; refundGroupId: string }
| { tag: TombstoneTag.DeleteReserve; reservePub: string }
| { tag: TombstoneTag.DeleteRefreshGroup; refreshGroupId: string }
| { tag: TombstoneTag.DeleteReward; walletTipId: string }
| { tag: TombstoneTag.DeletePayment; proposalId: string };
export function constructTombstone(p: ParsedTombstone): TombstoneIdStr {
switch (p.tag) {
case TombstoneTag.DeleteWithdrawalGroup:
return `tmb:${p.tag}:${p.withdrawalGroupId}` as TombstoneIdStr;
case TombstoneTag.DeleteRefund:
return `tmb:${p.tag}:${p.refundGroupId}` as TombstoneIdStr;
case TombstoneTag.DeleteReserve:
return `tmb:${p.tag}:${p.reservePub}` as TombstoneIdStr;
case TombstoneTag.DeletePayment:
return `tmb:${p.tag}:${p.proposalId}` as TombstoneIdStr;
case TombstoneTag.DeleteRefreshGroup:
return `tmb:${p.tag}:${p.refreshGroupId}` as TombstoneIdStr;
case TombstoneTag.DeleteReward:
return `tmb:${p.tag}:${p.walletTipId}` as TombstoneIdStr;
default:
assertUnreachable(p);
}
}
/**
* Uniform interface for a particular wallet transaction.
*/
export interface TransactionManager {
get taskId(): TaskId;
get transactionId(): TransactionIdStr;
fail(): Promise;
abort(): Promise;
suspend(): Promise;
resume(): Promise;
process(): Promise;
}
export enum TaskRunResultType {
Finished = "finished",
Pending = "pending",
Error = "error",
Longpoll = "longpoll",
}
export type TaskRunResult =
| TaskRunFinishedResult
| TaskRunErrorResult
| TaskRunLongpollResult
| TaskRunPendingResult;
export namespace TaskRunResult {
export function finished(): TaskRunResult {
return {
type: TaskRunResultType.Finished,
};
}
export function pending(): TaskRunResult {
return {
type: TaskRunResultType.Pending,
};
}
export function longpoll(): TaskRunResult {
return {
type: TaskRunResultType.Longpoll,
};
}
}
export interface TaskRunFinishedResult {
type: TaskRunResultType.Finished;
}
export interface TaskRunPendingResult {
type: TaskRunResultType.Pending;
}
export interface TaskRunErrorResult {
type: TaskRunResultType.Error;
errorDetail: TalerErrorDetail;
}
export interface TaskRunLongpollResult {
type: TaskRunResultType.Longpoll;
}
export interface DbRetryInfo {
firstTry: DbPreciseTimestamp;
nextRetry: DbPreciseTimestamp;
retryCounter: number;
}
export interface RetryPolicy {
readonly backoffDelta: Duration;
readonly backoffBase: number;
readonly maxTimeout: Duration;
}
const defaultRetryPolicy: RetryPolicy = {
backoffBase: 1.5,
backoffDelta: Duration.fromSpec({ seconds: 1 }),
maxTimeout: Duration.fromSpec({ minutes: 2 }),
};
function updateTimeout(
r: DbRetryInfo,
p: RetryPolicy = defaultRetryPolicy,
): void {
const now = AbsoluteTime.now();
if (now.t_ms === "never") {
throw Error("assertion failed");
}
if (p.backoffDelta.d_ms === "forever") {
r.nextRetry = timestampPreciseToDb(
AbsoluteTime.toPreciseTimestamp(AbsoluteTime.never()),
);
return;
}
const nextIncrement =
p.backoffDelta.d_ms * Math.pow(p.backoffBase, r.retryCounter);
const t =
now.t_ms +
(p.maxTimeout.d_ms === "forever"
? nextIncrement
: Math.min(p.maxTimeout.d_ms, nextIncrement));
r.nextRetry = timestampPreciseToDb(TalerPreciseTimestamp.fromMilliseconds(t));
}
export namespace DbRetryInfo {
export function getDuration(
r: DbRetryInfo | undefined,
p: RetryPolicy = defaultRetryPolicy,
): Duration {
if (!r) {
// If we don't have any retry info, run immediately.
return { d_ms: 0 };
}
if (p.backoffDelta.d_ms === "forever") {
return { d_ms: "forever" };
}
const t = p.backoffDelta.d_ms * Math.pow(p.backoffBase, r.retryCounter);
return {
d_ms:
p.maxTimeout.d_ms === "forever" ? t : Math.min(p.maxTimeout.d_ms, t),
};
}
export function reset(p: RetryPolicy = defaultRetryPolicy): DbRetryInfo {
const now = TalerPreciseTimestamp.now();
const info: DbRetryInfo = {
firstTry: timestampPreciseToDb(now),
nextRetry: timestampPreciseToDb(now),
retryCounter: 0,
};
updateTimeout(info, p);
return info;
}
export function increment(
r: DbRetryInfo | undefined,
p: RetryPolicy = defaultRetryPolicy,
): DbRetryInfo {
if (!r) {
return reset(p);
}
const r2 = { ...r };
r2.retryCounter++;
updateTimeout(r2, p);
return r2;
}
}
/**
* Parsed representation of task identifiers.
*/
export type ParsedTaskIdentifier =
| {
tag: PendingTaskType.Withdraw;
withdrawalGroupId: string;
}
| { tag: PendingTaskType.ExchangeUpdate; exchangeBaseUrl: string }
| { tag: PendingTaskType.Backup; backupProviderBaseUrl: string }
| { tag: PendingTaskType.Deposit; depositGroupId: string }
| { tag: PendingTaskType.ExchangeCheckRefresh; exchangeBaseUrl: string }
| { tag: PendingTaskType.PeerPullDebit; peerPullDebitId: string }
| { tag: PendingTaskType.PeerPullCredit; pursePub: string }
| { tag: PendingTaskType.PeerPushCredit; peerPushCreditId: string }
| { tag: PendingTaskType.PeerPushDebit; pursePub: string }
| { tag: PendingTaskType.Purchase; proposalId: string }
| { tag: PendingTaskType.Recoup; recoupGroupId: string }
| { tag: PendingTaskType.RewardPickup; walletRewardId: string }
| { tag: PendingTaskType.Refresh; refreshGroupId: string };
export function parseTaskIdentifier(x: string): ParsedTaskIdentifier {
const task = x.split(":");
if (task.length < 2) {
throw Error("task id should have al least 2 parts separated by ':'");
}
const [type, ...rest] = task;
switch (type) {
case PendingTaskType.Backup:
return { tag: type, backupProviderBaseUrl: decodeURIComponent(rest[0]) };
case PendingTaskType.Deposit:
return { tag: type, depositGroupId: rest[0] };
case PendingTaskType.ExchangeCheckRefresh:
return { tag: type, exchangeBaseUrl: decodeURIComponent(rest[0]) };
case PendingTaskType.ExchangeUpdate:
return { tag: type, exchangeBaseUrl: decodeURIComponent(rest[0]) };
case PendingTaskType.PeerPullCredit:
return { tag: type, pursePub: rest[0] };
case PendingTaskType.PeerPullDebit:
return { tag: type, peerPullDebitId: rest[0] };
case PendingTaskType.PeerPushCredit:
return { tag: type, peerPushCreditId: rest[0] };
case PendingTaskType.PeerPushDebit:
return { tag: type, pursePub: rest[0] };
case PendingTaskType.Purchase:
return { tag: type, proposalId: rest[0] };
case PendingTaskType.Recoup:
return { tag: type, recoupGroupId: rest[0] };
case PendingTaskType.Refresh:
return { tag: type, refreshGroupId: rest[0] };
case PendingTaskType.RewardPickup:
return { tag: type, walletRewardId: rest[0] };
case PendingTaskType.Withdraw:
return { tag: type, withdrawalGroupId: rest[0] };
default:
throw Error("invalid task identifier");
}
}
export function constructTaskIdentifier(p: ParsedTaskIdentifier): TaskId {
switch (p.tag) {
case PendingTaskType.Backup:
return `${p.tag}:${p.backupProviderBaseUrl}` as TaskId;
case PendingTaskType.Deposit:
return `${p.tag}:${p.depositGroupId}` as TaskId;
case PendingTaskType.ExchangeCheckRefresh:
return `${p.tag}:${p.exchangeBaseUrl}` as TaskId;
case PendingTaskType.ExchangeUpdate:
return `${p.tag}:${p.exchangeBaseUrl}` as TaskId;
case PendingTaskType.PeerPullDebit:
return `${p.tag}:${p.peerPullDebitId}` as TaskId;
case PendingTaskType.PeerPushCredit:
return `${p.tag}:${p.peerPushCreditId}` as TaskId;
case PendingTaskType.PeerPullCredit:
return `${p.tag}:${p.pursePub}` as TaskId;
case PendingTaskType.PeerPushDebit:
return `${p.tag}:${p.pursePub}` as TaskId;
case PendingTaskType.Purchase:
return `${p.tag}:${p.proposalId}` as TaskId;
case PendingTaskType.Recoup:
return `${p.tag}:${p.recoupGroupId}` as TaskId;
case PendingTaskType.Refresh:
return `${p.tag}:${p.refreshGroupId}` as TaskId;
case PendingTaskType.RewardPickup:
return `${p.tag}:${p.walletRewardId}` as TaskId;
case PendingTaskType.Withdraw:
return `${p.tag}:${p.withdrawalGroupId}` as TaskId;
default:
assertUnreachable(p);
}
}
export namespace TaskIdentifiers {
export function forWithdrawal(wg: WithdrawalGroupRecord): TaskId {
return `${PendingTaskType.Withdraw}:${wg.withdrawalGroupId}` as TaskId;
}
export function forExchangeUpdate(exch: ExchangeEntryRecord): TaskId {
return `${PendingTaskType.ExchangeUpdate}:${encodeURIComponent(
exch.baseUrl,
)}` as TaskId;
}
export function forExchangeUpdateFromUrl(exchBaseUrl: string): TaskId {
return `${PendingTaskType.ExchangeUpdate}:${encodeURIComponent(
exchBaseUrl,
)}` as TaskId;
}
export function forExchangeCheckRefresh(exch: ExchangeEntryRecord): TaskId {
return `${PendingTaskType.ExchangeCheckRefresh}:${encodeURIComponent(
exch.baseUrl,
)}` as TaskId;
}
export function forTipPickup(tipRecord: RewardRecord): TaskId {
return `${PendingTaskType.RewardPickup}:${tipRecord.walletRewardId}` as TaskId;
}
export function forRefresh(refreshGroupRecord: RefreshGroupRecord): TaskId {
return `${PendingTaskType.Refresh}:${refreshGroupRecord.refreshGroupId}` as TaskId;
}
export function forPay(purchaseRecord: PurchaseRecord): TaskId {
return `${PendingTaskType.Purchase}:${purchaseRecord.proposalId}` as TaskId;
}
export function forRecoup(recoupRecord: RecoupGroupRecord): TaskId {
return `${PendingTaskType.Recoup}:${recoupRecord.recoupGroupId}` as TaskId;
}
export function forDeposit(depositRecord: DepositGroupRecord): TaskId {
return `${PendingTaskType.Deposit}:${depositRecord.depositGroupId}` as TaskId;
}
export function forBackup(backupRecord: BackupProviderRecord): TaskId {
return `${PendingTaskType.Backup}:${encodeURIComponent(
backupRecord.baseUrl,
)}` as TaskId;
}
export function forPeerPushPaymentInitiation(
ppi: PeerPushDebitRecord,
): TaskId {
return `${PendingTaskType.PeerPushDebit}:${ppi.pursePub}` as TaskId;
}
export function forPeerPullPaymentInitiation(
ppi: PeerPullCreditRecord,
): TaskId {
return `${PendingTaskType.PeerPullCredit}:${ppi.pursePub}` as TaskId;
}
export function forPeerPullPaymentDebit(
ppi: PeerPullPaymentIncomingRecord,
): TaskId {
return `${PendingTaskType.PeerPullDebit}:${ppi.peerPullDebitId}` as TaskId;
}
export function forPeerPushCredit(
ppi: PeerPushPaymentIncomingRecord,
): TaskId {
return `${PendingTaskType.PeerPushCredit}:${ppi.peerPushCreditId}` as TaskId;
}
}
/**
* Result of a transaction transition.
*/
export enum TransitionResult {
Transition = 1,
Stay = 2,
}
/**
* Transaction context.
*
* FIXME: Should eventually be implemented by all transactions.
*/
export interface TransactionContext {
abortTransaction(): Promise;
resumeTransaction(): Promise;
failTransaction(): Promise;
}