diff --git a/src/rpc.ts b/src/rpc.ts index acf919e..7038494 100644 --- a/src/rpc.ts +++ b/src/rpc.ts @@ -172,14 +172,6 @@ type Cip56TokenSdk = { expirationDate?: Date }) => Promise<[unknown, unknown[]]> } - utxos: { - list: (params: { - partyId: string - includeLocked: boolean - limit: number - continueUntilCompletion: boolean - }) => Promise - } } } @@ -218,13 +210,31 @@ type TokenInstrumentId = { id?: string } +type TokenHoldingView = { + amount?: string + instrumentId?: TokenInstrumentId + lock?: unknown +} + type TokenHolding = { contractId: string - interfaceViewValue?: { - amount?: string - instrumentId?: TokenInstrumentId - lock?: unknown - } + interfaceViewValue?: TokenHoldingView + activeContract?: unknown + fetchedAtOffset?: number +} + +type ActiveContractsPage = { + activeContracts?: { + contractEntry?: { + JsActiveContract?: { + createdEvent?: { + contractId: string + interfaceViews?: { viewValue?: TokenHoldingView }[] + } + } + } + }[] + nextPageToken?: string | null } type TokenHoldingSummary = { @@ -258,6 +268,28 @@ type AmuletPreapprovalStatus = { const TRANSFER_PREAPPROVAL_PROPOSAL_TEMPLATE_ID = '#splice-wallet:Splice.Wallet.TransferPreapproval:TransferPreapprovalProposal' +const HOLDING_INTERFACE_ID = '#splice-api-token-holding-v1:Splice.Api.Token.HoldingV1:Holding' + +const holdingEventFormat = (partyId: string) => ({ + filtersByParty: { + [partyId]: { + cumulative: [ + { + identifierFilter: { + InterfaceFilter: { + value: { + interfaceId: HOLDING_INTERFACE_ID, + includeInterfaceView: true, + includeCreatedEventBlob: false, + }, + }, + }, + }, + ], + }, + }, + verbose: false, +}) const TRANSFER_PREAPPROVAL_PROPOSAL_MAX_ATTEMPTS = 31 const TRANSFER_PREAPPROVAL_PROPOSAL_RETRY_MS = 1_000 const AMULET_TAP_AMOUNT = '100' @@ -490,7 +522,9 @@ const summarizeHoldingUtxos = ( const groups = new Map() for (const holding of filterHoldingsByInstrument(holdings, requestedInstrument)) { const key = instrumentKey(holding.interfaceViewValue?.instrumentId ?? requestedInstrument) - groups.set(key, [...(groups.get(key) ?? []), holding]) + const group = groups.get(key) + if (group === undefined) groups.set(key, [holding]) + else group.push(holding) } return [...groups.entries()] .map(([key, tokenHoldings]) => { @@ -745,27 +779,54 @@ export const createRpc = (config: WalletServiceConfig, deps: RpcDependencies = { return await sdk.token.transfer.pending(partyId) } - const cip56ListHoldings = async (params: unknown): Promise => { - const p = objectParam>(params, 'cip56.listHoldings') - const partyId = requiredStringParam(p, 'partyId') - const sdk = await getTokenSdk() - return await sdk.token.utxos.list({ - partyId, - includeLocked: true, - limit: 100, - continueUntilCompletion: true, - }) + // Keeps the holding UTXO read behind one helper so Scan fallback cannot diverge. + // Pages the ACS snapshot rather than calling `sdk.token.utxos.list`: that either replays + // the whole update stream (`continueUntilCompletion`, thousands of requests per read) or + // returns one response bounded by the participant's `http-list-max-elements-limit`. + const listHoldingUtxos = async (partyId: string): Promise => { + // A page token is only valid against the offset and event format of the request that + // issued it, and a first page that lets the participant pick its own offset hands back + // a token the next page is refused for, so the offset is read and pinned up front. + const { offset } = (await ledgerApi({ + resource: '/v2/state/ledger-end', + requestMethod: 'get', + })) as { offset: number } + const eventFormat = holdingEventFormat(partyId) + const holdings: TokenHolding[] = [] + let pageToken: string | undefined + do { + const page = (await ledgerApi({ + resource: '/v2/state/active-contracts-page', + requestMethod: 'post', + body: { eventFormat, activeAtOffset: offset, pageToken }, + })) as ActiveContractsPage + for (const entry of page.activeContracts ?? []) { + const activeContract = entry.contractEntry?.JsActiveContract + const createdEvent = activeContract?.createdEvent + if (createdEvent === undefined) continue + // The request asks for one interface view per contract. A contract without it means + // the participant could not render the holding, and dropping it would understate a + // balance, so the read fails instead. + const interfaceViewValue = createdEvent.interfaceViews?.[0]?.viewValue + if (interfaceViewValue === undefined) { + throw new Error(`Holding ${createdEvent.contractId} has no ${HOLDING_INTERFACE_ID} view`) + } + holdings.push({ + contractId: createdEvent.contractId, + activeContract, + interfaceViewValue, + fetchedAtOffset: offset, + }) + } + // Documented as optional and possibly empty, so an empty token ends the read too. + pageToken = page.nextPageToken || undefined + } while (pageToken !== undefined) + return holdings } - // Keeps the generic SDK UTXO list behind one helper so Scan fallback cannot diverge. - const listHoldingUtxos = async (partyId: string): Promise => { - const sdk = await getTokenSdk() - return (await sdk.token.utxos.list({ - partyId, - includeLocked: true, - limit: 100, - continueUntilCompletion: true, - })) as TokenHolding[] + const cip56ListHoldings = async (params: unknown): Promise => { + const p = objectParam>(params, 'cip56.listHoldings') + return await listHoldingUtxos(requiredStringParam(p, 'partyId')) } // Uses Scan's Amulet aggregate endpoint for fast CC balances. diff --git a/test/rpc.test.ts b/test/rpc.test.ts index b1a2273..64c05db 100644 --- a/test/rpc.test.ts +++ b/test/rpc.test.ts @@ -28,6 +28,64 @@ const baseConfig = () => ({ }, }) +const jsonResponse = (value: unknown) => + new Response(JSON.stringify(value), { + status: 200, + headers: { 'content-type': 'application/json' }, + }) + +const ACS_OFFSET = 42 + +const holdingView = (suffix: string, amount: string, lock: unknown = null) => ({ + contractId: `holding-cid-${suffix}`, + viewValue: { + owner: 'receiver::party', + amount, + instrumentId: { admin: 'admin::party', id: 'Amulet' }, + lock, + }, +}) + +type HoldingFixture = { contractId: string; viewValue: Record } + +const activeContract = ({ contractId, viewValue }: HoldingFixture) => ({ + createdEvent: { contractId, interfaceViews: [{ viewValue }] }, +}) + +// Serves what the ACS holdings read consumes: a ledger end, then one +// `active-contracts-page` response per page, the last one without a token. Anything else +// (Scan) falls to `onOther`, so a test can fail Scan and still serve the ledger. +const ledgerAcsFetch = ( + pages: HoldingFixture[][], + onOther: () => Response = () => new Response('scan unavailable', { status: 503 }), +) => { + const pageBodies: { pageToken?: string; activeAtOffset?: number }[] = [] + const remaining = [...pages] + const fetch = async (url: URL | string, init?: RequestInit): Promise => { + const href = String(url) + if (href.endsWith('/v2/state/ledger-end')) return jsonResponse({ offset: ACS_OFFSET }) + if (!href.endsWith('/v2/state/active-contracts-page')) return onOther() + pageBodies.push(JSON.parse(String(init?.body))) + const page = remaining.shift() ?? [] + return jsonResponse({ + activeContracts: page.map((fixture) => ({ + contractEntry: { JsActiveContract: activeContract(fixture) }, + })), + activeAtOffset: ACS_OFFSET, + nextPageToken: remaining.length === 0 ? null : `page-${pages.length - remaining.length}`, + }) + } + return { fetch, pageBodies } +} + +const expectedHoldings = (fixtures: HoldingFixture[]) => + fixtures.map((fixture) => ({ + contractId: fixture.contractId, + activeContract: activeContract(fixture), + interfaceViewValue: fixture.viewValue, + fetchedAtOffset: ACS_OFFSET, + })) + describe('rpc dispatcher', () => { it('returns -32600 when jsonrpc is not "2.0"', async () => { const rpc = createRpc(baseConfig()) @@ -820,36 +878,15 @@ describe('CIP-56 token helpers', () => { assert.equal(cancelCalled, false) }) - it('lists token holding UTXOs through the SDK token namespace without reshaping contracts', async () => { - // Scenario: the wallet needs the active CIP-56 holdings for a party, but the - // Node-only wallet SDK must stay behind wallet-service. The RPC returns the - // SDK holding contracts unchanged so the browser boundary remains thin. - const holdingContracts = [ - { - contractId: 'holding-cid-1', - interfaceViewValue: { - owner: 'receiver::party', - amount: '666.0000000000', - instrumentId: { admin: 'admin::party', id: 'Amulet' }, - }, - }, - ] - const seen: { params?: unknown; tokenConfig?: unknown } = {} - const rpc = createRpc(baseConfig(), { - sdkFactory: async (options) => { - seen.tokenConfig = (options as { token?: unknown }).token - return { - token: { - utxos: { - list: async (params: unknown) => { - seen.params = params - return holdingContracts - }, - }, - }, - } - }, - }) + it('lists token holding UTXOs from every page of the ACS snapshot', async () => { + // Scenario: a party can hold more Holding contracts than the participant returns in + // one response, so a single-page read would under-report the balance. Every page must + // be followed, against the offset pinned by the first request. + const ledger = ledgerAcsFetch([ + [holdingView('1', '4.0000000000'), holdingView('2', '3.0000000000')], + [holdingView('3', '2.0000000000')], + ]) + const rpc = createRpc(baseConfig(), { fetch: ledger.fetch }) const res = (await rpc.handle({ jsonrpc: '2.0', @@ -859,18 +896,18 @@ describe('CIP-56 token helpers', () => { })) as JsonRpcResponse assert.ok('result' in res) - assert.deepEqual(res.result, holdingContracts) - assert.deepEqual(seen.params, { - partyId: 'receiver::party', - includeLocked: true, - limit: 100, - continueUntilCompletion: true, - }) - assert.deepEqual(seen.tokenConfig, { - validatorUrl: 'http://localhost:2000/api/validator', - auth: { method: 'static', token: 'backend.jwt' }, - registries: ['http://localhost:2000/api/validator/v0/scan-proxy'], - }) + assert.deepEqual( + (res.result as { contractId: string }[]).map((holding) => holding.contractId), + ['holding-cid-1', 'holding-cid-2', 'holding-cid-3'], + ) + assert.deepEqual( + ledger.pageBodies.map((body) => body.pageToken), + [undefined, 'page-1'], + ) + assert.deepEqual( + ledger.pageBodies.map((body) => body.activeAtOffset), + [ACS_OFFSET, ACS_OFFSET], + ) }) it('lists Amulet holding summaries through Scan without listing UTXOs', async () => { @@ -947,41 +984,12 @@ describe('CIP-56 token helpers', () => { it('falls back to UTXO summaries when Scan cannot summarize Amulet', async () => { // Scenario: local Scan may be unavailable or lagging. The summary RPC must still - // return a correct balance by falling back to the existing SDK UTXO path. - const holdingContracts = [ - { - contractId: 'holding-cid-1', - interfaceViewValue: { - owner: 'receiver::party', - amount: '4.0000000000', - instrumentId: { admin: 'admin::party', id: 'Amulet' }, - lock: null, - }, - }, - { - contractId: 'holding-cid-2', - interfaceViewValue: { - owner: 'receiver::party', - amount: '3.0000000000', - instrumentId: { admin: 'admin::party', id: 'Amulet' }, - lock: { holders: ['validator::party'] }, - }, - }, + // return a correct balance by falling back to the ACS holdings read. + const holdings = [ + holdingView('1', '4.0000000000'), + holdingView('2', '3.0000000000', { holders: ['validator::party'] }), ] - const seen: { params?: unknown } = {} - const rpc = createRpc(baseConfig(), { - fetch: async () => new Response('scan unavailable', { status: 503 }), - sdkFactory: async () => ({ - token: { - utxos: { - list: async (params: unknown) => { - seen.params = params - return holdingContracts - }, - }, - }, - }), - }) + const rpc = createRpc(baseConfig(), { fetch: ledgerAcsFetch([holdings]).fetch }) const res = (await rpc.handle({ jsonrpc: '2.0', @@ -1000,49 +1008,29 @@ describe('CIP-56 token helpers', () => { utxoCount: 2, lockedCount: 1, unlockedCount: 1, - holdings: holdingContracts, + holdings: expectedHoldings(holdings), source: 'utxos', }, ]) - assert.deepEqual(seen.params, { - partyId: 'receiver::party', - includeLocked: true, - limit: 100, - continueUntilCompletion: true, - }) }) it('summarizes non-Amulet tokens from UTXOs without calling Scan', async () => { // Scenario: Scan aggregates only CC/Amulet. Other CIP-56 tokens must use the // generic UTXO list and filter by the requested instrument. let scanCalled = false + const mockToken = { + contractId: 'holding-cid-1', + viewValue: { amount: '5', instrumentId: { admin: 'issuer::party', id: 'MockToken' } }, + } + const amulet = { + contractId: 'holding-cid-2', + viewValue: { amount: '99', instrumentId: { admin: 'admin::party', id: 'Amulet' } }, + } const rpc = createRpc(baseConfig(), { - fetch: async () => { + fetch: ledgerAcsFetch([[mockToken, amulet]], () => { scanCalled = true return new Response('{}') - }, - sdkFactory: async () => ({ - token: { - utxos: { - list: async () => [ - { - contractId: 'holding-cid-1', - interfaceViewValue: { - amount: '5', - instrumentId: { admin: 'issuer::party', id: 'MockToken' }, - }, - }, - { - contractId: 'holding-cid-2', - interfaceViewValue: { - amount: '99', - instrumentId: { admin: 'admin::party', id: 'Amulet' }, - }, - }, - ], - }, - }, - }), + }).fetch, }) const res = (await rpc.handle({ @@ -1057,13 +1045,6 @@ describe('CIP-56 token helpers', () => { assert.ok('result' in res) assert.equal(scanCalled, false) - const expectedHolding = { - contractId: 'holding-cid-1', - interfaceViewValue: { - amount: '5', - instrumentId: { admin: 'issuer::party', id: 'MockToken' }, - }, - } assert.deepEqual(res.result, [ { key: 'issuer::party:MockToken', @@ -1073,7 +1054,7 @@ describe('CIP-56 token helpers', () => { utxoCount: 1, lockedCount: 0, unlockedCount: 1, - holdings: [expectedHolding], + holdings: expectedHoldings([mockToken]), source: 'utxos', }, ]) @@ -1233,25 +1214,25 @@ describe('canton credentials', () => { // would leave Amulet and token-standard calls presenting an expired credential. const tokens = ['first.jwt', 'first.jwt', 'second.jwt'] const built: string[] = [] - const listHoldings = { + const listPending = { jsonrpc: '2.0' as const, id: 1, - method: 'cip56.listHoldings', + method: 'cip56.listPendingTransfers', params: { partyId: 'alice::fp' }, } const rpc = createRpc(oauthConfig(), { tokenProvider: { getToken: async () => tokens.shift() ?? 'second.jwt' }, sdkFactory: async (options) => { built.push((options as { auth: { token: string } }).auth.token) - return { token: { utxos: { list: async () => [] } } } + return { token: { transfer: { pending: async () => [] } } } }, }) - await rpc.handle(listHoldings) - await rpc.handle(listHoldings) + await rpc.handle(listPending) + await rpc.handle(listPending) assert.deepEqual(built, ['first.jwt']) - await rpc.handle(listHoldings) + await rpc.handle(listPending) assert.deepEqual(built, ['first.jwt', 'second.jwt']) })