/*
This file is part of GNU Taler
(C) 2022-2023 Taler Systems S.A.
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
*/
import {
Amounts,
CancellationToken,
CheckPeerPushDebitRequest,
CheckPeerPushDebitResponse,
CoinRefreshRequest,
ContractTermsUtil,
HttpStatusCode,
InitiatePeerPushDebitRequest,
InitiatePeerPushDebitResponse,
Logger,
NotificationType,
RefreshReason,
TalerError,
TalerErrorCode,
TalerPreciseTimestamp,
TalerProtocolTimestamp,
TalerProtocolViolationError,
TransactionAction,
TransactionIdStr,
TransactionMajorState,
TransactionMinorState,
TransactionState,
TransactionType,
encodeCrock,
getRandomBytes,
j2s,
} from "@gnu-taler/taler-util";
import {
HttpResponse,
readSuccessResponseJsonOrThrow,
readTalerErrorResponse,
} from "@gnu-taler/taler-util/http";
import {
PendingTaskType,
TaskId,
TaskRunResult,
TaskRunResultType,
TransactionContext,
constructTaskIdentifier,
spendCoins,
} from "./common.js";
import { EncryptContractRequest } from "./crypto/cryptoTypes.js";
import {
PeerPushDebitRecord,
PeerPushDebitStatus,
RefreshOperationStatus,
timestampPreciseToDb,
timestampProtocolFromDb,
timestampProtocolToDb,
} from "./db.js";
import {
codecForExchangePurseStatus,
getTotalPeerPaymentCost,
queryCoinInfosForSelection,
} from "./pay-peer-common.js";
import { createRefreshGroup } from "./refresh.js";
import {
constructTransactionIdentifier,
notifyTransition,
} from "./transactions.js";
import { assertUnreachable } from "./util/assertUnreachable.js";
import { PeerCoinRepair, selectPeerCoins } from "./util/coinSelection.js";
import { checkLogicInvariant } from "./util/invariants.js";
import { InternalWalletState } from "./wallet.js";
const logger = new Logger("pay-peer-push-debit.ts");
export class PeerPushDebitTransactionContext implements TransactionContext {
readonly transactionId: TransactionIdStr;
readonly retryTag: TaskId;
constructor(
public ws: InternalWalletState,
public pursePub: string,
) {
this.transactionId = constructTransactionIdentifier({
tag: TransactionType.PeerPushDebit,
pursePub,
});
this.retryTag = constructTaskIdentifier({
tag: PendingTaskType.PeerPushDebit,
pursePub,
});
}
async deleteTransaction(): Promise {
const { ws, pursePub, transactionId } = this;
await ws.db.runReadWriteTx(["peerPushDebit", "tombstones"], async (tx) => {
const debit = await tx.peerPushDebit.get(pursePub);
if (debit) {
await tx.peerPushDebit.delete(pursePub);
await tx.tombstones.put({ id: transactionId });
}
});
}
async suspendTransaction(): Promise {
const { ws, pursePub, transactionId, retryTag } = this;
const transitionInfo = await ws.db.runReadWriteTx(
["peerPushDebit"],
async (tx) => {
const pushDebitRec = await tx.peerPushDebit.get(pursePub);
if (!pushDebitRec) {
logger.warn(`peer push debit ${pursePub} not found`);
return;
}
let newStatus: PeerPushDebitStatus | undefined = undefined;
switch (pushDebitRec.status) {
case PeerPushDebitStatus.PendingCreatePurse:
newStatus = PeerPushDebitStatus.SuspendedCreatePurse;
break;
case PeerPushDebitStatus.AbortingRefreshDeleted:
newStatus = PeerPushDebitStatus.SuspendedAbortingRefreshDeleted;
break;
case PeerPushDebitStatus.AbortingRefreshExpired:
newStatus = PeerPushDebitStatus.SuspendedAbortingRefreshExpired;
break;
case PeerPushDebitStatus.AbortingDeletePurse:
newStatus = PeerPushDebitStatus.SuspendedAbortingDeletePurse;
break;
case PeerPushDebitStatus.PendingReady:
newStatus = PeerPushDebitStatus.SuspendedReady;
break;
case PeerPushDebitStatus.SuspendedAbortingDeletePurse:
case PeerPushDebitStatus.SuspendedAbortingRefreshDeleted:
case PeerPushDebitStatus.SuspendedAbortingRefreshExpired:
case PeerPushDebitStatus.SuspendedReady:
case PeerPushDebitStatus.SuspendedCreatePurse:
case PeerPushDebitStatus.Done:
case PeerPushDebitStatus.Aborted:
case PeerPushDebitStatus.Failed:
case PeerPushDebitStatus.Expired:
// Do nothing
break;
default:
assertUnreachable(pushDebitRec.status);
}
if (newStatus != null) {
const oldTxState = computePeerPushDebitTransactionState(pushDebitRec);
pushDebitRec.status = newStatus;
const newTxState = computePeerPushDebitTransactionState(pushDebitRec);
await tx.peerPushDebit.put(pushDebitRec);
return {
oldTxState,
newTxState,
};
}
return undefined;
},
);
ws.taskScheduler.stopShepherdTask(retryTag);
notifyTransition(ws, transactionId, transitionInfo);
}
async abortTransaction(): Promise {
const { ws, pursePub, transactionId, retryTag } = this;
const transitionInfo = await ws.db.runReadWriteTx(
["peerPushDebit"],
async (tx) => {
const pushDebitRec = await tx.peerPushDebit.get(pursePub);
if (!pushDebitRec) {
logger.warn(`peer push debit ${pursePub} not found`);
return;
}
let newStatus: PeerPushDebitStatus | undefined = undefined;
switch (pushDebitRec.status) {
case PeerPushDebitStatus.PendingReady:
case PeerPushDebitStatus.SuspendedReady:
newStatus = PeerPushDebitStatus.AbortingDeletePurse;
break;
case PeerPushDebitStatus.SuspendedCreatePurse:
case PeerPushDebitStatus.PendingCreatePurse:
// Network request might already be in-flight!
newStatus = PeerPushDebitStatus.AbortingDeletePurse;
break;
case PeerPushDebitStatus.SuspendedAbortingRefreshDeleted:
case PeerPushDebitStatus.SuspendedAbortingDeletePurse:
case PeerPushDebitStatus.SuspendedAbortingRefreshExpired:
case PeerPushDebitStatus.AbortingRefreshDeleted:
case PeerPushDebitStatus.AbortingRefreshExpired:
case PeerPushDebitStatus.Done:
case PeerPushDebitStatus.AbortingDeletePurse:
case PeerPushDebitStatus.Aborted:
case PeerPushDebitStatus.Expired:
case PeerPushDebitStatus.Failed:
// Do nothing
break;
default:
assertUnreachable(pushDebitRec.status);
}
if (newStatus != null) {
const oldTxState = computePeerPushDebitTransactionState(pushDebitRec);
pushDebitRec.status = newStatus;
const newTxState = computePeerPushDebitTransactionState(pushDebitRec);
await tx.peerPushDebit.put(pushDebitRec);
return {
oldTxState,
newTxState,
};
}
return undefined;
},
);
ws.taskScheduler.stopShepherdTask(retryTag);
notifyTransition(ws, transactionId, transitionInfo);
ws.taskScheduler.startShepherdTask(retryTag);
}
async resumeTransaction(): Promise {
const { ws, pursePub, transactionId, retryTag } = this;
const transitionInfo = await ws.db.runReadWriteTx(
["peerPushDebit"],
async (tx) => {
const pushDebitRec = await tx.peerPushDebit.get(pursePub);
if (!pushDebitRec) {
logger.warn(`peer push debit ${pursePub} not found`);
return;
}
let newStatus: PeerPushDebitStatus | undefined = undefined;
switch (pushDebitRec.status) {
case PeerPushDebitStatus.SuspendedAbortingDeletePurse:
newStatus = PeerPushDebitStatus.AbortingDeletePurse;
break;
case PeerPushDebitStatus.SuspendedAbortingRefreshDeleted:
newStatus = PeerPushDebitStatus.AbortingRefreshDeleted;
break;
case PeerPushDebitStatus.SuspendedAbortingRefreshExpired:
newStatus = PeerPushDebitStatus.AbortingRefreshExpired;
break;
case PeerPushDebitStatus.SuspendedReady:
newStatus = PeerPushDebitStatus.PendingReady;
break;
case PeerPushDebitStatus.SuspendedCreatePurse:
newStatus = PeerPushDebitStatus.PendingCreatePurse;
break;
case PeerPushDebitStatus.PendingCreatePurse:
case PeerPushDebitStatus.AbortingRefreshDeleted:
case PeerPushDebitStatus.AbortingRefreshExpired:
case PeerPushDebitStatus.AbortingDeletePurse:
case PeerPushDebitStatus.PendingReady:
case PeerPushDebitStatus.Done:
case PeerPushDebitStatus.Aborted:
case PeerPushDebitStatus.Failed:
case PeerPushDebitStatus.Expired:
// Do nothing
break;
default:
assertUnreachable(pushDebitRec.status);
}
if (newStatus != null) {
const oldTxState = computePeerPushDebitTransactionState(pushDebitRec);
pushDebitRec.status = newStatus;
const newTxState = computePeerPushDebitTransactionState(pushDebitRec);
await tx.peerPushDebit.put(pushDebitRec);
return {
oldTxState,
newTxState,
};
}
return undefined;
},
);
ws.taskScheduler.startShepherdTask(retryTag);
notifyTransition(ws, transactionId, transitionInfo);
}
async failTransaction(): Promise {
const { ws, pursePub, transactionId, retryTag } = this;
const transitionInfo = await ws.db.runReadWriteTx(
["peerPushDebit"],
async (tx) => {
const pushDebitRec = await tx.peerPushDebit.get(pursePub);
if (!pushDebitRec) {
logger.warn(`peer push debit ${pursePub} not found`);
return;
}
let newStatus: PeerPushDebitStatus | undefined = undefined;
switch (pushDebitRec.status) {
case PeerPushDebitStatus.AbortingRefreshDeleted:
case PeerPushDebitStatus.SuspendedAbortingRefreshDeleted:
// FIXME: What to do about the refresh group?
newStatus = PeerPushDebitStatus.Failed;
break;
case PeerPushDebitStatus.AbortingDeletePurse:
case PeerPushDebitStatus.SuspendedAbortingDeletePurse:
case PeerPushDebitStatus.AbortingRefreshExpired:
case PeerPushDebitStatus.SuspendedAbortingRefreshExpired:
case PeerPushDebitStatus.PendingReady:
case PeerPushDebitStatus.SuspendedReady:
case PeerPushDebitStatus.SuspendedCreatePurse:
case PeerPushDebitStatus.PendingCreatePurse:
newStatus = PeerPushDebitStatus.Failed;
break;
case PeerPushDebitStatus.Done:
case PeerPushDebitStatus.Aborted:
case PeerPushDebitStatus.Failed:
case PeerPushDebitStatus.Expired:
// Do nothing
break;
default:
assertUnreachable(pushDebitRec.status);
}
if (newStatus != null) {
const oldTxState = computePeerPushDebitTransactionState(pushDebitRec);
pushDebitRec.status = newStatus;
const newTxState = computePeerPushDebitTransactionState(pushDebitRec);
await tx.peerPushDebit.put(pushDebitRec);
return {
oldTxState,
newTxState,
};
}
return undefined;
},
);
ws.taskScheduler.stopShepherdTask(retryTag);
notifyTransition(ws, transactionId, transitionInfo);
ws.taskScheduler.startShepherdTask(retryTag);
}
}
export async function checkPeerPushDebit(
ws: InternalWalletState,
req: CheckPeerPushDebitRequest,
): Promise {
const instructedAmount = Amounts.parseOrThrow(req.amount);
logger.trace(
`checking peer push debit for ${Amounts.stringify(instructedAmount)}`,
);
const coinSelRes = await selectPeerCoins(ws, { instructedAmount });
if (coinSelRes.type === "failure") {
throw TalerError.fromDetail(
TalerErrorCode.WALLET_PEER_PUSH_PAYMENT_INSUFFICIENT_BALANCE,
{
insufficientBalanceDetails: coinSelRes.insufficientBalanceDetails,
},
);
}
logger.trace(`selected peer coins (len=${coinSelRes.result.coins.length})`);
const totalAmount = await getTotalPeerPaymentCost(
ws,
coinSelRes.result.coins,
);
logger.trace("computed total peer payment cost");
return {
exchangeBaseUrl: coinSelRes.result.exchangeBaseUrl,
amountEffective: Amounts.stringify(totalAmount),
amountRaw: req.amount,
maxExpirationDate: coinSelRes.result.maxExpirationDate,
};
}
async function handlePurseCreationConflict(
ws: InternalWalletState,
peerPushInitiation: PeerPushDebitRecord,
resp: HttpResponse,
): Promise {
const pursePub = peerPushInitiation.pursePub;
const errResp = await readTalerErrorResponse(resp);
const ctx = new PeerPushDebitTransactionContext(ws, pursePub);
if (errResp.code !== TalerErrorCode.EXCHANGE_GENERIC_INSUFFICIENT_FUNDS) {
await ctx.failTransaction();
return TaskRunResult.finished();
}
// FIXME: Properly parse!
const brokenCoinPub = (errResp as any).coin_pub;
logger.trace(`excluded broken coin pub=${brokenCoinPub}`);
if (!brokenCoinPub) {
// FIXME: Details!
throw new TalerProtocolViolationError();
}
const instructedAmount = Amounts.parseOrThrow(peerPushInitiation.amount);
const sel = peerPushInitiation.coinSel;
const repair: PeerCoinRepair = {
coinPubs: [],
contribs: [],
exchangeBaseUrl: peerPushInitiation.exchangeBaseUrl,
};
for (let i = 0; i < sel.coinPubs.length; i++) {
if (sel.coinPubs[i] != brokenCoinPub) {
repair.coinPubs.push(sel.coinPubs[i]);
repair.contribs.push(Amounts.parseOrThrow(sel.contributions[i]));
}
}
const coinSelRes = await selectPeerCoins(ws, { instructedAmount, repair });
if (coinSelRes.type == "failure") {
// FIXME: Details!
throw Error(
"insufficient balance to re-select coins to repair double spending",
);
}
await ws.db.runReadWriteTx(["peerPushDebit"], async (tx) => {
const myPpi = await tx.peerPushDebit.get(peerPushInitiation.pursePub);
if (!myPpi) {
return;
}
switch (myPpi.status) {
case PeerPushDebitStatus.PendingCreatePurse:
case PeerPushDebitStatus.SuspendedCreatePurse: {
const sel = coinSelRes.result;
myPpi.coinSel = {
coinPubs: sel.coins.map((x) => x.coinPub),
contributions: sel.coins.map((x) => x.contribution),
};
break;
}
default:
return;
}
await tx.peerPushDebit.put(myPpi);
});
return TaskRunResult.progress();
}
async function processPeerPushDebitCreateReserve(
ws: InternalWalletState,
peerPushInitiation: PeerPushDebitRecord,
): Promise {
const pursePub = peerPushInitiation.pursePub;
const purseExpiration = peerPushInitiation.purseExpiration;
const hContractTerms = peerPushInitiation.contractTermsHash;
const ctx = new PeerPushDebitTransactionContext(ws, pursePub);
const transactionId = ctx.transactionId;
logger.trace(`processing ${transactionId} pending(create-reserve)`);
const contractTermsRecord = await ws.db.runReadOnlyTx(
["contractTerms"],
async (tx) => {
return tx.contractTerms.get(hContractTerms);
},
);
if (!contractTermsRecord) {
throw Error(
`db invariant failed, contract terms for ${transactionId} missing`,
);
}
const purseSigResp = await ws.cryptoApi.signPurseCreation({
hContractTerms,
mergePub: peerPushInitiation.mergePub,
minAge: 0,
purseAmount: peerPushInitiation.amount,
purseExpiration: timestampProtocolFromDb(purseExpiration),
pursePriv: peerPushInitiation.pursePriv,
});
const coins = await queryCoinInfosForSelection(
ws,
peerPushInitiation.coinSel,
);
const depositSigsResp = await ws.cryptoApi.signPurseDeposits({
exchangeBaseUrl: peerPushInitiation.exchangeBaseUrl,
pursePub: peerPushInitiation.pursePub,
coins,
});
const encryptContractRequest: EncryptContractRequest = {
contractTerms: contractTermsRecord.contractTermsRaw,
mergePriv: peerPushInitiation.mergePriv,
pursePriv: peerPushInitiation.pursePriv,
pursePub: peerPushInitiation.pursePub,
contractPriv: peerPushInitiation.contractPriv,
contractPub: peerPushInitiation.contractPub,
nonce: peerPushInitiation.contractEncNonce,
};
logger.trace(`encrypt contract request: ${j2s(encryptContractRequest)}`);
const econtractResp = await ws.cryptoApi.encryptContractForMerge(
encryptContractRequest,
);
const createPurseUrl = new URL(
`purses/${peerPushInitiation.pursePub}/create`,
peerPushInitiation.exchangeBaseUrl,
);
const reqBody = {
amount: peerPushInitiation.amount,
merge_pub: peerPushInitiation.mergePub,
purse_sig: purseSigResp.sig,
h_contract_terms: hContractTerms,
purse_expiration: timestampProtocolFromDb(purseExpiration),
deposits: depositSigsResp.deposits,
min_age: 0,
econtract: econtractResp.econtract,
};
logger.trace(`request body: ${j2s(reqBody)}`);
const httpResp = await ws.http.fetch(createPurseUrl.href, {
method: "POST",
body: reqBody,
});
{
const resp = await httpResp.json();
logger.info(`resp: ${j2s(resp)}`);
}
switch (httpResp.status) {
case HttpStatusCode.Ok:
break;
case HttpStatusCode.Forbidden: {
// FIXME: Store this error!
await ctx.failTransaction();
return TaskRunResult.finished();
}
case HttpStatusCode.Conflict: {
// Handle double-spending
return handlePurseCreationConflict(ws, peerPushInitiation, httpResp);
}
default: {
const errResp = await readTalerErrorResponse(httpResp);
return {
type: TaskRunResultType.Error,
errorDetail: errResp,
};
}
}
if (httpResp.status !== HttpStatusCode.Ok) {
// FIXME: do proper error reporting
throw Error("got error response from exchange");
}
await transitionPeerPushDebitTransaction(ws, pursePub, {
stFrom: PeerPushDebitStatus.PendingCreatePurse,
stTo: PeerPushDebitStatus.PendingReady,
});
return TaskRunResult.backoff();
}
async function processPeerPushDebitAbortingDeletePurse(
ws: InternalWalletState,
peerPushInitiation: PeerPushDebitRecord,
): Promise {
const { pursePub, pursePriv } = peerPushInitiation;
const transactionId = constructTransactionIdentifier({
tag: TransactionType.PeerPushDebit,
pursePub,
});
const sigResp = await ws.cryptoApi.signDeletePurse({
pursePriv,
});
const purseUrl = new URL(
`purses/${pursePub}`,
peerPushInitiation.exchangeBaseUrl,
);
const resp = await ws.http.fetch(purseUrl.href, {
method: "DELETE",
headers: {
"taler-purse-signature": sigResp.sig,
},
});
logger.info(`deleted purse with response status ${resp.status}`);
const transitionInfo = await ws.db.runReadWriteTx(
[
"peerPushDebit",
"refreshGroups",
"denominations",
"coinAvailability",
"coins",
],
async (tx) => {
const ppiRec = await tx.peerPushDebit.get(pursePub);
if (!ppiRec) {
return undefined;
}
if (ppiRec.status !== PeerPushDebitStatus.AbortingDeletePurse) {
return undefined;
}
const currency = Amounts.currencyOf(ppiRec.amount);
const oldTxState = computePeerPushDebitTransactionState(ppiRec);
const coinPubs: CoinRefreshRequest[] = [];
for (let i = 0; i < ppiRec.coinSel.coinPubs.length; i++) {
coinPubs.push({
amount: ppiRec.coinSel.contributions[i],
coinPub: ppiRec.coinSel.coinPubs[i],
});
}
const refresh = await createRefreshGroup(
ws,
tx,
currency,
coinPubs,
RefreshReason.AbortPeerPushDebit,
transactionId,
);
ppiRec.status = PeerPushDebitStatus.AbortingRefreshDeleted;
ppiRec.abortRefreshGroupId = refresh.refreshGroupId;
await tx.peerPushDebit.put(ppiRec);
const newTxState = computePeerPushDebitTransactionState(ppiRec);
return {
oldTxState,
newTxState,
};
},
);
notifyTransition(ws, transactionId, transitionInfo);
return TaskRunResult.backoff();
}
interface SimpleTransition {
stFrom: PeerPushDebitStatus;
stTo: PeerPushDebitStatus;
}
async function transitionPeerPushDebitTransaction(
ws: InternalWalletState,
pursePub: string,
transitionSpec: SimpleTransition,
): Promise {
const transactionId = constructTransactionIdentifier({
tag: TransactionType.PeerPushDebit,
pursePub,
});
const transitionInfo = await ws.db.runReadWriteTx(
["peerPushDebit"],
async (tx) => {
const ppiRec = await tx.peerPushDebit.get(pursePub);
if (!ppiRec) {
return undefined;
}
if (ppiRec.status !== transitionSpec.stFrom) {
return undefined;
}
const oldTxState = computePeerPushDebitTransactionState(ppiRec);
ppiRec.status = transitionSpec.stTo;
await tx.peerPushDebit.put(ppiRec);
const newTxState = computePeerPushDebitTransactionState(ppiRec);
return {
oldTxState,
newTxState,
};
},
);
notifyTransition(ws, transactionId, transitionInfo);
}
async function processPeerPushDebitAbortingRefreshDeleted(
ws: InternalWalletState,
peerPushInitiation: PeerPushDebitRecord,
): Promise {
const pursePub = peerPushInitiation.pursePub;
const abortRefreshGroupId = peerPushInitiation.abortRefreshGroupId;
checkLogicInvariant(!!abortRefreshGroupId);
const transactionId = constructTransactionIdentifier({
tag: TransactionType.PeerPushDebit,
pursePub: peerPushInitiation.pursePub,
});
const transitionInfo = await ws.db.runReadWriteTx(
["refreshGroups", "peerPushDebit"],
async (tx) => {
const refreshGroup = await tx.refreshGroups.get(abortRefreshGroupId);
let newOpState: PeerPushDebitStatus | undefined;
if (!refreshGroup) {
// Maybe it got manually deleted? Means that we should
// just go into failed.
logger.warn("no aborting refresh group found for deposit group");
newOpState = PeerPushDebitStatus.Failed;
} else {
if (refreshGroup.operationStatus === RefreshOperationStatus.Finished) {
newOpState = PeerPushDebitStatus.Aborted;
} else if (
refreshGroup.operationStatus === RefreshOperationStatus.Failed
) {
newOpState = PeerPushDebitStatus.Failed;
}
}
if (newOpState) {
const newDg = await tx.peerPushDebit.get(pursePub);
if (!newDg) {
return;
}
const oldTxState = computePeerPushDebitTransactionState(newDg);
newDg.status = newOpState;
const newTxState = computePeerPushDebitTransactionState(newDg);
await tx.peerPushDebit.put(newDg);
return { oldTxState, newTxState };
}
return undefined;
},
);
notifyTransition(ws, transactionId, transitionInfo);
// FIXME: Shouldn't this be finished in some cases?!
return TaskRunResult.backoff();
}
async function processPeerPushDebitAbortingRefreshExpired(
ws: InternalWalletState,
peerPushInitiation: PeerPushDebitRecord,
): Promise {
const pursePub = peerPushInitiation.pursePub;
const abortRefreshGroupId = peerPushInitiation.abortRefreshGroupId;
checkLogicInvariant(!!abortRefreshGroupId);
const transactionId = constructTransactionIdentifier({
tag: TransactionType.PeerPushDebit,
pursePub: peerPushInitiation.pursePub,
});
const transitionInfo = await ws.db.runReadWriteTx(
["peerPushDebit", "refreshGroups"],
async (tx) => {
const refreshGroup = await tx.refreshGroups.get(abortRefreshGroupId);
let newOpState: PeerPushDebitStatus | undefined;
if (!refreshGroup) {
// Maybe it got manually deleted? Means that we should
// just go into failed.
logger.warn("no aborting refresh group found for deposit group");
newOpState = PeerPushDebitStatus.Failed;
} else {
if (refreshGroup.operationStatus === RefreshOperationStatus.Finished) {
newOpState = PeerPushDebitStatus.Expired;
} else if (
refreshGroup.operationStatus === RefreshOperationStatus.Failed
) {
newOpState = PeerPushDebitStatus.Failed;
}
}
if (newOpState) {
const newDg = await tx.peerPushDebit.get(pursePub);
if (!newDg) {
return;
}
const oldTxState = computePeerPushDebitTransactionState(newDg);
newDg.status = newOpState;
const newTxState = computePeerPushDebitTransactionState(newDg);
await tx.peerPushDebit.put(newDg);
return { oldTxState, newTxState };
}
return undefined;
},
);
notifyTransition(ws, transactionId, transitionInfo);
// FIXME: Shouldn't this be finished in some cases?!
return TaskRunResult.backoff();
}
/**
* Process the "pending(ready)" state of a peer-push-debit transaction.
*/
async function processPeerPushDebitReady(
ws: InternalWalletState,
peerPushInitiation: PeerPushDebitRecord,
cancellationToken: CancellationToken,
): Promise {
logger.trace("processing peer-push-debit pending(ready)");
const pursePub = peerPushInitiation.pursePub;
const transactionId = constructTaskIdentifier({
tag: PendingTaskType.PeerPushDebit,
pursePub,
});
const mergeUrl = new URL(
`purses/${pursePub}/merge`,
peerPushInitiation.exchangeBaseUrl,
);
mergeUrl.searchParams.set("timeout_ms", "30000");
logger.info(`long-polling on purse status at ${mergeUrl.href}`);
const resp = await ws.http.fetch(mergeUrl.href, {
// timeout: getReserveRequestTimeout(withdrawalGroup),
cancellationToken,
});
if (resp.status === HttpStatusCode.Ok) {
const purseStatus = await readSuccessResponseJsonOrThrow(
resp,
codecForExchangePurseStatus(),
);
const mergeTimestamp = purseStatus.merge_timestamp;
logger.info(`got purse status ${j2s(purseStatus)}`);
if (!mergeTimestamp || TalerProtocolTimestamp.isNever(mergeTimestamp)) {
return TaskRunResult.backoff();
} else {
await transitionPeerPushDebitTransaction(
ws,
peerPushInitiation.pursePub,
{
stFrom: PeerPushDebitStatus.PendingReady,
stTo: PeerPushDebitStatus.Done,
},
);
return TaskRunResult.progress();
}
} else if (resp.status === HttpStatusCode.Gone) {
logger.info(`purse ${pursePub} is gone, aborting peer-push-debit`);
const transitionInfo = await ws.db.runReadWriteTx(
[
"peerPushDebit",
"refreshGroups",
"denominations",
"coinAvailability",
"coins",
],
async (tx) => {
const ppiRec = await tx.peerPushDebit.get(pursePub);
if (!ppiRec) {
return undefined;
}
if (ppiRec.status !== PeerPushDebitStatus.PendingReady) {
return undefined;
}
const currency = Amounts.currencyOf(ppiRec.amount);
const oldTxState = computePeerPushDebitTransactionState(ppiRec);
const coinPubs: CoinRefreshRequest[] = [];
for (let i = 0; i < ppiRec.coinSel.coinPubs.length; i++) {
coinPubs.push({
amount: ppiRec.coinSel.contributions[i],
coinPub: ppiRec.coinSel.coinPubs[i],
});
}
const refresh = await createRefreshGroup(
ws,
tx,
currency,
coinPubs,
RefreshReason.AbortPeerPushDebit,
transactionId,
);
ppiRec.status = PeerPushDebitStatus.AbortingRefreshExpired;
ppiRec.abortRefreshGroupId = refresh.refreshGroupId;
await tx.peerPushDebit.put(ppiRec);
const newTxState = computePeerPushDebitTransactionState(ppiRec);
return {
oldTxState,
newTxState,
};
},
);
notifyTransition(ws, transactionId, transitionInfo);
return TaskRunResult.backoff();
} else {
logger.warn(`unexpected HTTP status for purse: ${resp.status}`);
return TaskRunResult.longpollReturnedPending();
}
}
export async function processPeerPushDebit(
ws: InternalWalletState,
pursePub: string,
cancellationToken: CancellationToken,
): Promise {
const peerPushInitiation = await ws.db.runReadOnlyTx(
["peerPushDebit"],
async (tx) => {
return tx.peerPushDebit.get(pursePub);
},
);
if (!peerPushInitiation) {
throw Error("peer push payment not found");
}
switch (peerPushInitiation.status) {
case PeerPushDebitStatus.PendingCreatePurse:
return processPeerPushDebitCreateReserve(ws, peerPushInitiation);
case PeerPushDebitStatus.PendingReady:
return processPeerPushDebitReady(
ws,
peerPushInitiation,
cancellationToken,
);
case PeerPushDebitStatus.AbortingDeletePurse:
return processPeerPushDebitAbortingDeletePurse(ws, peerPushInitiation);
case PeerPushDebitStatus.AbortingRefreshDeleted:
return processPeerPushDebitAbortingRefreshDeleted(ws, peerPushInitiation);
case PeerPushDebitStatus.AbortingRefreshExpired:
return processPeerPushDebitAbortingRefreshExpired(ws, peerPushInitiation);
default: {
const txState = computePeerPushDebitTransactionState(peerPushInitiation);
logger.warn(
`not processing peer-push-debit transaction in state ${j2s(txState)}`,
);
}
}
return TaskRunResult.finished();
}
/**
* Initiate sending a peer-to-peer push payment.
*/
export async function initiatePeerPushDebit(
ws: InternalWalletState,
req: InitiatePeerPushDebitRequest,
): Promise {
const instructedAmount = Amounts.parseOrThrow(
req.partialContractTerms.amount,
);
const purseExpiration = req.partialContractTerms.purse_expiration;
const contractTerms = req.partialContractTerms;
const pursePair = await ws.cryptoApi.createEddsaKeypair({});
const mergePair = await ws.cryptoApi.createEddsaKeypair({});
const hContractTerms = ContractTermsUtil.hashContractTerms(contractTerms);
const contractKeyPair = await ws.cryptoApi.createEddsaKeypair({});
const coinSelRes = await selectPeerCoins(ws, { instructedAmount });
if (coinSelRes.type !== "success") {
throw TalerError.fromDetail(
TalerErrorCode.WALLET_PEER_PUSH_PAYMENT_INSUFFICIENT_BALANCE,
{
insufficientBalanceDetails: coinSelRes.insufficientBalanceDetails,
},
);
}
const sel = coinSelRes.result;
logger.info(`selected p2p coins (push):`);
logger.trace(`${j2s(coinSelRes)}`);
const totalAmount = await getTotalPeerPaymentCost(
ws,
coinSelRes.result.coins,
);
logger.info(`computed total peer payment cost`);
const pursePub = pursePair.pub;
const ctx = new PeerPushDebitTransactionContext(ws, pursePub);
const transactionId = ctx.transactionId;
const contractEncNonce = encodeCrock(getRandomBytes(24));
const transitionInfo = await ws.db.runReadWriteTx(
[
"exchanges",
"contractTerms",
"coins",
"coinAvailability",
"denominations",
"refreshGroups",
"peerPushDebit",
],
async (tx) => {
// FIXME: Instead of directly doing a spendCoin here,
// we might want to mark the coins as used and spend them
// after we've been able to create the purse.
await spendCoins(ws, tx, {
allocationId: constructTransactionIdentifier({
tag: TransactionType.PeerPushDebit,
pursePub: pursePair.pub,
}),
coinPubs: sel.coins.map((x) => x.coinPub),
contributions: sel.coins.map((x) =>
Amounts.parseOrThrow(x.contribution),
),
refreshReason: RefreshReason.PayPeerPush,
});
const ppi: PeerPushDebitRecord = {
amount: Amounts.stringify(instructedAmount),
contractPriv: contractKeyPair.priv,
contractPub: contractKeyPair.pub,
contractTermsHash: hContractTerms,
exchangeBaseUrl: sel.exchangeBaseUrl,
mergePriv: mergePair.priv,
mergePub: mergePair.pub,
purseExpiration: timestampProtocolToDb(purseExpiration),
pursePriv: pursePair.priv,
pursePub: pursePair.pub,
timestampCreated: timestampPreciseToDb(TalerPreciseTimestamp.now()),
status: PeerPushDebitStatus.PendingCreatePurse,
contractEncNonce,
coinSel: {
coinPubs: sel.coins.map((x) => x.coinPub),
contributions: sel.coins.map((x) => x.contribution),
},
totalCost: Amounts.stringify(totalAmount),
};
await tx.peerPushDebit.add(ppi);
await tx.contractTerms.put({
h: hContractTerms,
contractTermsRaw: contractTerms,
});
const newTxState = computePeerPushDebitTransactionState(ppi);
return {
oldTxState: { major: TransactionMajorState.None },
newTxState,
};
},
);
notifyTransition(ws, transactionId, transitionInfo);
ws.notify({
type: NotificationType.BalanceChange,
hintTransactionId: transactionId,
});
ws.taskScheduler.startShepherdTask(ctx.retryTag);
return {
contractPriv: contractKeyPair.priv,
mergePriv: mergePair.priv,
pursePub: pursePair.pub,
exchangeBaseUrl: coinSelRes.result.exchangeBaseUrl,
transactionId: constructTransactionIdentifier({
tag: TransactionType.PeerPushDebit,
pursePub: pursePair.pub,
}),
};
}
export function computePeerPushDebitTransactionActions(
ppiRecord: PeerPushDebitRecord,
): TransactionAction[] {
switch (ppiRecord.status) {
case PeerPushDebitStatus.PendingCreatePurse:
return [TransactionAction.Abort, TransactionAction.Suspend];
case PeerPushDebitStatus.PendingReady:
return [TransactionAction.Abort, TransactionAction.Suspend];
case PeerPushDebitStatus.Aborted:
return [TransactionAction.Delete];
case PeerPushDebitStatus.AbortingDeletePurse:
return [TransactionAction.Suspend, TransactionAction.Fail];
case PeerPushDebitStatus.AbortingRefreshDeleted:
return [TransactionAction.Suspend, TransactionAction.Fail];
case PeerPushDebitStatus.AbortingRefreshExpired:
return [TransactionAction.Suspend, TransactionAction.Fail];
case PeerPushDebitStatus.SuspendedAbortingRefreshExpired:
return [TransactionAction.Resume, TransactionAction.Fail];
case PeerPushDebitStatus.SuspendedAbortingDeletePurse:
return [TransactionAction.Resume, TransactionAction.Fail];
case PeerPushDebitStatus.SuspendedAbortingRefreshDeleted:
return [TransactionAction.Resume, TransactionAction.Fail];
case PeerPushDebitStatus.SuspendedCreatePurse:
return [TransactionAction.Resume, TransactionAction.Abort];
case PeerPushDebitStatus.SuspendedReady:
return [TransactionAction.Resume, TransactionAction.Abort];
case PeerPushDebitStatus.Done:
return [TransactionAction.Delete];
case PeerPushDebitStatus.Expired:
return [TransactionAction.Delete];
case PeerPushDebitStatus.Failed:
return [TransactionAction.Delete];
}
}
export function computePeerPushDebitTransactionState(
ppiRecord: PeerPushDebitRecord,
): TransactionState {
switch (ppiRecord.status) {
case PeerPushDebitStatus.PendingCreatePurse:
return {
major: TransactionMajorState.Pending,
minor: TransactionMinorState.CreatePurse,
};
case PeerPushDebitStatus.PendingReady:
return {
major: TransactionMajorState.Pending,
minor: TransactionMinorState.Ready,
};
case PeerPushDebitStatus.Aborted:
return {
major: TransactionMajorState.Aborted,
};
case PeerPushDebitStatus.AbortingDeletePurse:
return {
major: TransactionMajorState.Aborting,
minor: TransactionMinorState.DeletePurse,
};
case PeerPushDebitStatus.AbortingRefreshDeleted:
return {
major: TransactionMajorState.Aborting,
minor: TransactionMinorState.Refresh,
};
case PeerPushDebitStatus.AbortingRefreshExpired:
return {
major: TransactionMajorState.Aborting,
minor: TransactionMinorState.RefreshExpired,
};
case PeerPushDebitStatus.SuspendedAbortingDeletePurse:
return {
major: TransactionMajorState.SuspendedAborting,
minor: TransactionMinorState.DeletePurse,
};
case PeerPushDebitStatus.SuspendedAbortingRefreshExpired:
return {
major: TransactionMajorState.SuspendedAborting,
minor: TransactionMinorState.RefreshExpired,
};
case PeerPushDebitStatus.SuspendedAbortingRefreshDeleted:
return {
major: TransactionMajorState.SuspendedAborting,
minor: TransactionMinorState.Refresh,
};
case PeerPushDebitStatus.SuspendedCreatePurse:
return {
major: TransactionMajorState.Suspended,
minor: TransactionMinorState.CreatePurse,
};
case PeerPushDebitStatus.SuspendedReady:
return {
major: TransactionMajorState.Suspended,
minor: TransactionMinorState.Ready,
};
case PeerPushDebitStatus.Done:
return {
major: TransactionMajorState.Done,
};
case PeerPushDebitStatus.Failed:
return {
major: TransactionMajorState.Failed,
};
case PeerPushDebitStatus.Expired:
return {
major: TransactionMajorState.Expired,
};
}
}