Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 28x 28x 28x 28x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 68x 1x 1x 1x 67x 67x 67x 67x 67x 67x 67x 67x 67x 67x 67x 67x 67x 67x 67x 67x 67x 3x 3x 3x 3x 3x 3x 64x 64x 64x 64x 64x 67x 68x 68x 68x 68x 68x 68x 28x | /**
* Payer DFSP — the entry participant.
*
* It sets the business intent once, drives discovery → quote → transfer, and
* branches on the quote a real payer would branch on.
*
* Being the flow's **entry point**, this participant is where both identities
* originate when a request carries none: `traceFrom` mints the causal trace and
* `flowFrom` mints the execution ULID (PRD R9). Every later participant reads
* the same two values back off the request and forwards them unchanged, so one
* execution stays one flow id and one trace however many services it crosses.
* The flow **kind** is deliberately not minted here: it is the deployment's
* stable process name, bound once per participant by `flows.ts`.
*
* A route computes a `{status, body}` result — the same shape `hop` returns —
* and the handler maps it onto the reply. It cannot simply *return* that object:
* fastify answers a returned object with HTTP 200 and the wrapper as the body,
* so a downstream hop would observe a success and unwrap the envelope instead of
* the payload. That is why every route here sets the status explicitly.
*/
import {bindLeg} from '../src/context.ts';
import {decide} from '../src/decide.ts';
import {hop, type Participant} from './participant.ts';
export interface PayerOptions {
hubUrl: string;
/**
* The name the receiving hub runs under (PRD R22). A leg declares **who** the
* caller expects to answer, and that is a deployment fact rather than a literal:
* the same payer calls `hub` in the single-scheme topology and `hubA` in the
* inter-scheme one, so a declaration baked into this file would misreport every
* call it labelled. Required for the same reason the flow kind is: a declared
* receiver that had to be guessed would be a fabricated identity.
*/
hubName: string;
/** FX rate above which a real payer would refuse the quote. */
rateLimit?: number;
}
export function installPayer(participant: Participant, options: PayerOptions): void {
const {logger, app} = participant;
const rateLimit = options.rateLimit ?? 1.25;
app.post('/transfer', async (request, reply) => {
const traceId = participant.traceFrom(request);
// The execution id read off the inbound request, not a fresh one: one
// execution is one ULID however many participants it crosses, so minting
// here would put two flow ids into the same execution's records (D1,
// ruled 2026-09-13). `flowFrom` mints only when the request carries none,
// which is the entry point's job and nobody else's. Being the entry point,
// the payer is also the one participant with no inbound leg: it *declares*
// the legs it calls instead of adopting one.
const flowId = participant.flowFrom(request);
const leg = participant.legFrom(request);
const body = (request.body ?? {}) as {amount?: number; currency?: string; target?: string};
const amount = body.amount ?? 100;
const currency = body.currency ?? 'USD';
const result = await participant.run(traceId, flowId, leg, () =>
participant.phase('discovery', async () => {
// The leg covers this call's own records too — the request it sends
// and the answer it reads back — not just the hop in the middle. A
// reader following `leg=payer.discovery.parties` therefore lands on
// the call, not on one arbitrary line of it (PRD R22).
await bindLeg({id: 'payer.discovery.parties', to: options.hubName}, async () => {
logger.info('looking up payee', {
req: {operation: 'POST', target: '/parties/msisdn'},
amount,
currency,
});
const lookup = await hop(participant, options.hubUrl, '/parties', {
target: body.target ?? 'msisdn-1',
});
logger.info('payee found', {
res: {status: lookup.status},
payeeCurrency: 'EUR',
});
});
return participant.phase('quote', async () => {
const quote = await bindLeg(
{id: 'payer.quote.rates', to: options.hubName},
async () => {
logger.info('requesting fx quote', {from: currency, to: 'EUR'});
return hop(participant, options.hubUrl, '/quotes', {
amount,
from: currency,
to: 'EUR',
});
},
);
const rate = (quote.body as {rate?: number} | undefined)?.rate ?? 1;
// The provider's status is part of the decision's input, not a
// separate branch: a decline is exactly the case where no rate
// arrived, and folding it in keeps this the one place the payer's
// rationale is recorded. Without it the `?? 1` fallback above
// would read a decline as a rate of 1 and the transfer would
// settle on a quote nobody offered.
const accepted = decide(
'quote-acceptable',
{rate, rateLimit, status: quote.status},
[
{
name: 'reject',
when: values =>
(values.rate as number) > (values.rateLimit as number) ||
(values.status as number) >= 400,
run: () => false,
},
{name: 'accept', when: () => true, run: () => true},
],
);
if (!accepted) {
logger.warn('quote refused', {rate, reason: 'rate above limit'});
return {status: 409, body: {reason: 'rate above limit'}};
}
logger.info('quote accepted', {res: {status: quote.status}, rate});
return participant.phase('transfer', async () => {
const started = Date.now();
const settled = await bindLeg(
{id: 'payer.transfer.submit', to: options.hubName},
async () => {
logger.info('submitting transfer', {
req: {operation: 'POST', target: '/transfers'},
amount,
});
return hop(participant, options.hubUrl, '/transfers', {
amount,
currency,
});
},
);
if (settled.status >= 400) {
logger.error('transfer rejected', {
err: {message: `hub returned ${settled.status}`},
res: {status: settled.status, elapsedMs: Date.now() - started},
});
return {status: 502, body: {reason: 'transfer rejected'}};
}
logger.info('transfer complete', {
res: {status: settled.status, elapsedMs: Date.now() - started},
});
return {status: 200, body: {status: 'settled'}};
});
});
}),
);
reply.status(result.status);
return result.body;
});
}
|