commit df14901ddb7fa7c7bcd1c5526addf2b478e406c8 parent b95d4ca3e584dabd627a63fcff847d2111fae890 Author: Florian Dold <dold@taler.net> Date: Wed, 19 Aug 2026 21:41:37 +0200 wallet-core: add targeted database indexes and batch queries Diffstat:
13 files changed, 554 insertions(+), 80 deletions(-)
diff --git a/packages/taler-wallet-core/src/coinSelection.test.ts b/packages/taler-wallet-core/src/coinSelection.test.ts @@ -605,6 +605,7 @@ test("deposit available max includes pending-only refresh outputs", async () => exchangeMasterPub: masterPublicKey, maxAge: 0, freshCoinCount: 0, + hasFreshCoins: 0, visibleCoinCount: 0, pendingRefreshOutputCount: 3, } as WalletCoinAvailability; diff --git a/packages/taler-wallet-core/src/common.ts b/packages/taler-wallet-core/src/common.ts @@ -153,6 +153,7 @@ export async function makeCoinAvailable( exchangeBaseUrl: denom.exchangeBaseUrl, exchangeMasterPub: denom.exchangeMasterPub, freshCoinCount: 0, + hasFreshCoins: 0, visibleCoinCount: 0, }; } diff --git a/packages/taler-wallet-core/src/db-common.ts b/packages/taler-wallet-core/src/db-common.ts @@ -177,6 +177,12 @@ export interface WalletTransactionMeta { currency: string; } +/** Stable database cursor for transaction metadata pagination. */ +export interface WalletTransactionMetaCursor { + timestamp: DbPreciseTimestamp; + transactionId: string; +} + /** * Retry state of a task. * @@ -1647,6 +1653,12 @@ export interface WalletCoinAvailability { freshCoinCount: number; /** + * Numeric boolean derived from freshCoinCount for compound database indexes. + * IndexedDB booleans are not valid keys, hence the 0/1 representation. + */ + hasFreshCoins: 0 | 1; + + /** * Number of fresh coins that are available * and visible, i.e. the source transaction is in * a final state. diff --git a/packages/taler-wallet-core/src/db-converter.ts b/packages/taler-wallet-core/src/db-converter.ts @@ -40,6 +40,7 @@ import { import { WalletCoin, + WalletCoinAvailability, WalletPlanchet, WalletPurchase, WalletReserve, @@ -184,6 +185,12 @@ function normalizeCoin(rec: WalletCoin): WalletCoin { : stripped; } +function normalizeCoinAvailability( + rec: WalletCoinAvailability, +): WalletCoinAvailability { + return { ...rec, hasFreshCoins: rec.freshCoinCount > 0 ? 1 : 0 }; +} + /** * The copy plan, in groups. * @@ -397,7 +404,8 @@ const COPY_PLAN: CopyStep[][] = [ step( "coinAvailability", (tx) => tx.getCoinAvailabilities(), - (tx, r) => tx.upsertCoinAvailability(r), + (tx, r) => tx.upsertCoinAvailability(normalizeCoinAvailability(r)), + normalizeCoinAvailability, ), step( "refundGroups", diff --git a/packages/taler-wallet-core/src/db-indexeddb.ts b/packages/taler-wallet-core/src/db-indexeddb.ts @@ -256,7 +256,7 @@ export const CURRENT_DB_CONFIG_KEY = "currentMainDbName"; * backwards-compatible way or object stores and indices * are added. */ -export const WALLET_DB_MINOR_VERSION = 31; +export const WALLET_DB_MINOR_VERSION = 32; // FIXME: Should these be numeric codes? export type KycUserType = "individual" | "business"; @@ -638,6 +638,11 @@ export const WalletIndexedDbStoresV1 = { byTimestamp: describeIndex("byTimestamp", "timestamp", { versionAdded: 13, }), + byTimestampAndId: describeIndex( + "byTimestampAndId", + ["timestamp", "transactionId"], + { versionAdded: 32 }, + ), byStatus: describeIndex("byStatus", "status", { versionAdded: 13, }), @@ -701,6 +706,11 @@ export const WalletIndexedDbStoresV1 = { byExchangeBaseUrl: describeIndex("byExchangeBaseUrl", "exchangeBaseUrl", { versionAdded: 31, }), + byExchangeFreshAndAge: describeIndex( + "byExchangeFreshAndAge", + ["exchangeBaseUrl", "hasFreshCoins", "maxAge"], + { versionAdded: 32 }, + ), }, ), // The pre-re-key store. Keeps its map key equal to its store name: the @@ -1658,7 +1668,11 @@ async function fixup20260807CoinAvailabilityByMasterPub( if (existing) { continue; } - await tx.coinAvailabilityV2.put({ ...av, exchangeMasterPub: masterPub }); + await tx.coinAvailabilityV2.put({ + ...av, + exchangeMasterPub: masterPub, + hasFreshCoins: av.freshCoinCount > 0 ? 1 : 0, + }); } } } @@ -2375,6 +2389,22 @@ function onTalerDbUpgradeNeeded( newVersion, upgradeTransaction, ); + if (oldVersion < 32) { + const store = upgradeTransaction.objectStore("coinAvailabilityV2"); + const req = store.openCursor(); + req.onsuccess = () => { + const cursor = req.result; + if (!cursor) { + return; + } + const value = cursor.value as WalletCoinAvailability; + cursor.update({ + ...value, + hasFreshCoins: value.freshCoinCount > 0 ? 1 : 0, + }); + cursor.continue(); + }; + } } function onMetaDbUpgradeNeeded( diff --git a/packages/taler-wallet-core/src/db-sqlite-migrations.test.ts b/packages/taler-wallet-core/src/db-sqlite-migrations.test.ts @@ -141,6 +141,54 @@ test("exchange source migration upgrades an existing native database", async () } }); +test("wallet query migration backfills availability and creates indexes", async () => { + const { path, cleanup } = withTempDb(); + try { + let db = await openRaw(path); + await initSqliteWalletDb( + db, + schemaMigrations.filter((x) => x.version < 9), + ); + await ( + await db.prepare( + `INSERT INTO coin_availability ( + exchange_base_url, denom_pub_hash, max_age, currency, value, + exchange_master_pub, fresh_coin_count, visible_coin_count + ) VALUES ($url, $dph, $age, 'TESTKUDOS', 'TESTKUDOS:1', $mpk, $fresh, 0)`, + ) + ).run({ + url: "https://migration.example/", + dph: new Uint8Array([1]), + age: 0, + mpk: new Uint8Array([2]), + fresh: 3, + }); + await db.close(); + + db = await openRaw(path); + await initSqliteWalletDb(db); + const rows = await queryAll( + db, + "SELECT has_fresh_coins FROM coin_availability", + ); + assert.strictEqual(Number(rows[0].has_fresh_coins), 1); + const indexes = await queryAll(db, "PRAGMA index_list(coin_availability)"); + assert.ok( + indexes.some((x) => x.name === "coin_availability_by_exchange_fresh_age"), + ); + const txIndexes = await queryAll( + db, + "PRAGMA index_list(transactions_meta)", + ); + assert.ok( + txIndexes.some((x) => x.name === "transactions_meta_by_timestamp_id"), + ); + await db.close(); + } finally { + cleanup(); + } +}); + test("a migration already recorded is not applied twice", async () => { const { path, cleanup } = withTempDb(); try { diff --git a/packages/taler-wallet-core/src/db-sqlite-schema.ts b/packages/taler-wallet-core/src/db-sqlite-schema.ts @@ -85,7 +85,7 @@ * * Bump this when adding a migration to {@link schemaMigrations}. */ -export const SQLITE_SCHEMA_VERSION = 8; +export const SQLITE_SCHEMA_VERSION = 9; /** * Tables of the IndexedDB emulation, children before parents. @@ -1237,10 +1237,8 @@ CREATE TABLE IF NOT EXISTS coin_availability ( CHECK (pending_refresh_output_count >= 0), PRIMARY KEY (exchange_master_pub, denom_pub_hash, max_age) ); --- Column order matches the IndexedDB byExchangeAgeAvailability index, because --- getCoinAvailabilityByExchangeAndAgeRange depends on the *tuple* ordering --- (see the row-value comparison in dbtx-sqlite.ts), not on the columns --- individually. +-- Retained for compatibility with existing databases. Migration 9 adds the +-- correctness-preserving exchange/has-fresh/age index used by current code. CREATE INDEX IF NOT EXISTS coin_availability_by_exchange_age_fresh ON coin_availability (exchange_base_url, max_age, fresh_coin_count); @@ -1329,6 +1327,18 @@ export const schemaMigrations: SchemaMigration[] = [ name: "exchange-entry-source", statements: ["ALTER TABLE exchanges ADD COLUMN source TEXT"], }, + { + version: 9, + name: "wallet-query-indexes", + statements: [ + "ALTER TABLE coin_availability ADD COLUMN has_fresh_coins INTEGER NOT NULL DEFAULT 0 CHECK (has_fresh_coins IN (0, 1))", + "UPDATE coin_availability SET has_fresh_coins = CASE WHEN fresh_coin_count > 0 THEN 1 ELSE 0 END", + "CREATE INDEX coin_availability_by_exchange_fresh_age ON coin_availability (exchange_base_url, has_fresh_coins, max_age)", + "CREATE INDEX transactions_meta_by_timestamp_id ON transactions_meta (timestamp, transaction_id)", + "CREATE INDEX coins_by_exchange_base_url ON coins (exchange_base_url)", + "CREATE INDEX coins_by_master_pub_denom_age_status_pub ON coins (exchange_master_pub, denom_pub_hash, max_age, status, coin_pub)", + ], + }, ]; /** Native tables that contain wallet records (not schema bookkeeping). */ diff --git a/packages/taler-wallet-core/src/dbtx-bench.ts b/packages/taler-wallet-core/src/dbtx-bench.ts @@ -178,6 +178,7 @@ async function populate( currency: "TESTKUDOS", value: "TESTKUDOS:1" as AmountString, freshCoinCount: 10, + hasFreshCoins: 1, visibleCoinCount: 10, }; await tx.upsertCoinAvailability(avail); diff --git a/packages/taler-wallet-core/src/dbtx-conformance-cases.ts b/packages/taler-wallet-core/src/dbtx-conformance-cases.ts @@ -246,6 +246,7 @@ function makeAvail( currency: "TESTKUDOS", value: amt("TESTKUDOS:1"), freshCoinCount: 1, + hasFreshCoins: 1, visibleCoinCount: 1, }; return rec; @@ -1212,6 +1213,27 @@ export const conformanceCases: ConformanceCase[] = [ }, { + name: "reserve: batch lookup preserves requested order and skips missing", + async run(t, runner) { + await runner.runReadWriteTx(async (tx) => { + await tx.upsertReserve(makeReserve("rpub-b1")); + await tx.upsertReserve(makeReserve("rpub-b2")); + }); + const got = await runner.runReadWriteTx((tx) => + tx.getReservesByPubs([ + ck("rpub-b2"), + ck("rpub-missing"), + ck("rpub-b1"), + ]), + ); + t.deepEqual( + got.map((x) => x.reservePub), + [ck("rpub-b2"), ck("rpub-b1")], + ); + }, + }, + + { name: "operation retry: upsert, get, delete", async run(t, runner) { const rec: WalletOperationRetry = { @@ -2054,15 +2076,8 @@ export const conformanceCases: ConformanceCase[] = [ }, { - name: "coin availability: age range bounds a tuple, not each column", + name: "coin availability: age range excludes every zero-fresh row", async run(t, runner) { - // This pins a semantic that is easy to get wrong when translating the - // IndexedDB compound key range to SQL. The range runs from - // (ageLower, 1) to (ageUpper, MAX) *lexicographically*, so the - // "at least one fresh coin" floor applies only at max_age = ageLower: - // a row with a higher max_age and zero fresh coins is still inside it. - // A naive `max_age BETWEEN ? AND ? AND fresh_coin_count >= 1` would - // drop that row. await runner.runReadWriteTx(async (tx) => { const atLowerNoFresh = makeAvail("https://er/", "d-lo", 0); atLowerNoFresh.freshCoinCount = 0; @@ -2080,8 +2095,8 @@ export const conformanceCases: ConformanceCase[] = [ const hashes = got.map((a) => a.denomPubHash).sort(); t.deepEqual( hashes, - [ckh("d-hi"), ckh("d-lf")].sort(), - "excludes only the zero-fresh row at the lower bound", + [ckh("d-lf")], + "excludes zero-fresh rows throughout the age range", ); }, }, @@ -3055,6 +3070,54 @@ export const conformanceCases: ConformanceCase[] = [ }, { + name: "transaction meta: compound cursor pages through timestamp ties", + async run(t, runner) { + await runner.runReadWriteTx(async (tx) => { + for (const id of ["txn:page:c", "txn:page:a", "txn:page:b"]) { + await tx.upsertTransactionMeta({ + transactionId: id, + timestamp: tsPrecise(710), + status: WithdrawalGroupStatus.Done, + currency: "TESTKUDOS", + exchanges: [], + }); + } + }); + const first = await runner.runReadWriteTx((tx) => + tx.listTransactionMetaPage({ direction: "forward", limit: 2 }), + ); + const last = first[first.length - 1]; + const second = await runner.runReadWriteTx((tx) => + tx.listTransactionMetaPage({ + direction: "forward", + limit: 2, + cursor: { + timestamp: last.timestamp, + transactionId: last.transactionId, + }, + }), + ); + const ids = [...first, ...second] + .map((x) => x.transactionId) + .filter((x) => x.startsWith("txn:page:")); + t.deepEqual(ids, ["txn:page:a", "txn:page:b", "txn:page:c"]); + const backwards = await runner.runReadWriteTx((tx) => + tx.listTransactionMetaPage({ + direction: "backward", + limit: 3, + cursor: { timestamp: tsPrecise(710), transactionId: "\uffff" }, + }), + ); + t.deepEqual( + backwards + .map((x) => x.transactionId) + .filter((x) => x.startsWith("txn:page:")), + ["txn:page:c", "txn:page:b", "txn:page:a"], + ); + }, + }, + + { name: "transaction meta: ordered by timestamp, and limited", async run(t, runner) { await runner.runReadWriteTx(async (tx) => { @@ -3171,6 +3234,29 @@ export const conformanceCases: ConformanceCase[] = [ }, { + name: "purchase: batch lookup preserves IDs and exchange hydration", + async run(t, runner) { + const a = makePurchase("prop-batch-a"); + a.exchanges = ["https://batch-a/"]; + const b = makePurchase("prop-batch-b"); + b.exchanges = ["https://batch-b/", "https://batch-c/"]; + await runner.runReadWriteTx(async (tx) => { + await tx.upsertPurchase(a); + await tx.upsertPurchase(b); + }); + const got = await runner.runReadWriteTx((tx) => + tx.getPurchasesByIds(["prop-batch-b", "prop-missing", "prop-batch-a"]), + ); + t.deepEqual( + got.map((x) => x.proposalId), + ["prop-batch-b", "prop-batch-a"], + ); + t.deepEqual(got[0].exchanges, b.exchanges); + t.deepEqual(got[1].exchanges, a.exchanges); + }, + }, + + { name: "purchase: the exchange list has exactly one stored copy", async run(t, runner) { // The native schema keeps this in a junction table rather than a JSON @@ -3691,6 +3777,30 @@ export const conformanceCases: ConformanceCase[] = [ }, { + name: "token: lookup by family hash", + async run(t, runner) { + await runner.runReadWriteTx(async (tx) => { + const a = makeToken("tk-family-a"); + a.tokenFamilyHash = ckh("family-shared"); + const b = makeToken("tk-family-b"); + b.tokenFamilyHash = ckh("family-shared"); + const c = makeToken("tk-family-c"); + c.tokenFamilyHash = ckh("family-other"); + await tx.upsertToken(a); + await tx.upsertToken(b); + await tx.upsertToken(c); + }); + const got = await runner.runReadWriteTx((tx) => + tx.getTokensByFamilyHash(ckh("family-shared")), + ); + t.deepEqual( + got.map((x) => x.tokenUsePub).sort(), + [ck("tk-family-a"), ck("tk-family-b")].sort(), + ); + }, + }, + + { name: "slate: addressed by the full (purchase, choice, output, repeat)", async run(t, runner) { await runner.runReadWriteTx(async (tx) => { diff --git a/packages/taler-wallet-core/src/dbtx-indexeddb.ts b/packages/taler-wallet-core/src/dbtx-indexeddb.ts @@ -48,6 +48,7 @@ import { WalletSlate, WalletDenomination, WalletTransactionMeta, + WalletTransactionMetaCursor, DbPreciseTimestamp, DbProtocolTimestamp, WalletOperationRetry, @@ -408,6 +409,32 @@ export class IdbWalletTransaction implements WalletDbTransaction { ); } + async listTransactionMetaPage(req: { + cursor?: WalletTransactionMetaCursor; + direction: "forward" | "backward"; + limit: number; + }): Promise<WalletTransactionMeta[]> { + const index = this.tx.transactionsMeta.indexes.byTimestampAndId; + const key = req.cursor + ? [req.cursor.timestamp, req.cursor.transactionId] + : undefined; + if (req.direction === "forward") { + const range = key ? GlobalIDB.KeyRange.lowerBound(key, true) : undefined; + return await index.getAll(range, req.limit); + } + const range = key ? GlobalIDB.KeyRange.upperBound(key, true) : undefined; + const cursor = index.iterPrev(range); + const records: WalletTransactionMeta[] = []; + while (records.length < req.limit) { + const next = await cursor.next(); + if (!next.hasValue) { + break; + } + records.push(next.value); + } + return records; + } + async listTransactionMetaByStatus(req: { onlyActive: boolean; }): Promise<WalletTransactionMeta[]> { @@ -583,10 +610,10 @@ export class IdbWalletTransaction implements WalletDbTransaction { const tx = this.tx; // Lower bound of 1 on freshCoinCount: only denominations that actually // have a fresh coin available. - return await tx.coinAvailabilityV2.indexes.byExchangeAgeAvailability.getAll( + return await tx.coinAvailabilityV2.indexes.byExchangeFreshAndAge.getAll( GlobalIDB.KeyRange.bound( - [exchangeBaseUrl, ageLower, 1], - [exchangeBaseUrl, ageUpper, Number.MAX_SAFE_INTEGER], + [exchangeBaseUrl, 1, ageLower], + [exchangeBaseUrl, 1, ageUpper], ), ); } @@ -776,11 +803,13 @@ export class IdbWalletTransaction implements WalletDbTransaction { // Cascade to the denominations of that family. There is no accessor for // "denominations by family" on its own, so this walks the index whose // first component is the family serial. - const all = - await tx.denominationsV2.indexes.byDenominationFamilySerialAndStampExpireWithdraw.getAll(); - const doomed = all.filter( - (d) => d.denominationFamilySerial === denominationFamilySerial, - ); + const doomed = + await tx.denominationsV2.indexes.byDenominationFamilySerialAndStampExpireWithdraw.getAll( + GlobalIDB.KeyRange.bound( + [denominationFamilySerial, Number.MIN_SAFE_INTEGER], + [denominationFamilySerial, Number.MAX_SAFE_INTEGER], + ), + ); for (const d of doomed) { await tx.denominationsV2.delete([d.exchangeMasterPub, d.denomPubHash]); } @@ -986,6 +1015,13 @@ export class IdbWalletTransaction implements WalletDbTransaction { ]); } + async getPurchasesByIds(proposalIds: string[]): Promise<WalletPurchase[]> { + const purchases = await Promise.all( + proposalIds.map((proposalId) => this.tx.purchases.get(proposalId)), + ); + return purchases.filter((x): x is WalletPurchase => x !== undefined); + } + async getPurchasesByUrlAndOrderId( merchantBaseUrl: string, orderId: string, @@ -1209,10 +1245,9 @@ export class IdbWalletTransaction implements WalletDbTransaction { } async countDonationPlanchetsByProposal(proposalId: string): Promise<number> { - const tx = this.tx; - const keys = - await tx.donationPlanchets.indexes.byProposalId.getAllKeys(proposalId); - return keys.length; + return await this.tx.donationPlanchets.indexes.byProposalId.count( + proposalId, + ); } async getWithdrawalGroup( @@ -1290,10 +1325,7 @@ export class IdbWalletTransaction implements WalletDbTransaction { } async countPlanchetsByGroup(withdrawalGroupId: string): Promise<number> { - const tx = this.tx; - const keys = - await tx.planchets.indexes.byGroup.getAllKeys(withdrawalGroupId); - return keys.length; + return await this.tx.planchets.indexes.byGroup.count(withdrawalGroupId); } async deletePlanchetsByGroup(withdrawalGroupId: string): Promise<void> { @@ -1398,6 +1430,15 @@ export class IdbWalletTransaction implements WalletDbTransaction { return await tx.reserves.indexes.byReservePub.get(reservePub); } + async getReservesByPubs(reservePubs: string[]): Promise<WalletReserve[]> { + const reserves = await Promise.all( + reservePubs.map((reservePub) => + this.tx.reserves.indexes.byReservePub.get(reservePub), + ), + ); + return reserves.filter((x): x is WalletReserve => x !== undefined); + } + async listAllReserves(): Promise<WalletReserve[]> { return await this.tx.reserves.getAll(); } @@ -1459,7 +1500,10 @@ export class IdbWalletTransaction implements WalletDbTransaction { async upsertCoinAvailability(rec: WalletCoinAvailability): Promise<void> { const tx = this.tx; - await tx.coinAvailabilityV2.put(rec); + await tx.coinAvailabilityV2.put({ + ...rec, + hasFreshCoins: rec.freshCoinCount > 0 ? 1 : 0, + }); } async getCoinHistory( @@ -1507,6 +1551,12 @@ export class IdbWalletTransaction implements WalletDbTransaction { ); } + async getTokensByFamilyHash(tokenFamilyHash: string): Promise<WalletToken[]> { + return await this.tx.tokens.indexes.byTokenFamilyHash.getAll( + tokenFamilyHash, + ); + } + async getPeerPullCredit( pursePub: string, ): Promise<WalletPeerPullCredit | undefined> { @@ -1923,14 +1973,10 @@ export class IdbWalletTransaction implements WalletDbTransaction { } async getCoinsByPubs(coinPubs: string[]): Promise<WalletCoin[]> { - const coins: WalletCoin[] = []; - for (const pub of coinPubs) { - const coin = await this.tx.coins.get(pub); - if (coin) { - coins.push(coin); - } - } - return coins; + const coins = await Promise.all( + coinPubs.map((pub) => this.tx.coins.get(pub)), + ); + return coins.filter((coin): coin is WalletCoin => coin !== undefined); } async getActiveDepositGroups(): Promise<WalletDepositGroup[]> { diff --git a/packages/taler-wallet-core/src/dbtx-sqlite.ts b/packages/taler-wallet-core/src/dbtx-sqlite.ts @@ -96,6 +96,7 @@ import { ExchangeMigrationReason, WalletDenomination, WalletOperationRetry, + WalletTransactionMetaCursor, WalletRefundGroup, WalletRefundItem, WalletReserve, @@ -421,6 +422,52 @@ function dbToOptJson<T>(v: Sqlite3Value | undefined): T | undefined { return v == null ? undefined : (JSON.parse(v as string) as T); } +/** Native table enumerated for each backend-conversion store. */ +const SQLITE_MIGRATION_TABLES: Record<WalletDbMigrationStore, string> = { + config: "config", + currencyInfo: "currency_info", + contacts: "contacts", + mailboxMessages: "mailbox_messages", + mailboxConfigurations: "mailbox_configurations", + contractTerms: "contract_terms", + tombstones: "tombstones", + operationRetries: "operation_retries", + bankAccounts: "bank_accounts", + globalCurrencyExchanges: "global_currency_exchanges", + globalCurrencyAuditors: "global_currency_auditors", + exchangeBaseUrlFixups: "exchange_base_url_fixups", + exchangeBaseUrlMigrationLog: "exchange_base_url_migration_log", + reserves: "reserves", + exchanges: "exchanges", + exchangeDetails: "exchange_details", + exchangeSignKeys: "exchange_sign_keys", + denominationFamilies: "denomination_families", + denominations: "denominations", + withdrawalGroups: "withdrawal_groups", + purchases: "purchases", + refreshGroups: "refresh_groups", + coins: "coins", + planchets: "planchets", + refreshSessions: "refresh_sessions", + coinHistory: "coin_history", + coinAvailability: "coin_availability", + refundGroups: "refund_groups", + tokens: "tokens", + slates: "slates", + depositGroups: "deposit_groups", + recoupGroups: "recoup_groups", + denomLossEvents: "denom_loss_events", + peerPushDebit: "peer_push_debit", + peerPushCredit: "peer_push_credit", + peerPullDebit: "peer_pull_debit", + peerPullCredit: "peer_pull_credit", + donationSummaries: "donation_summaries", + donationPlanchets: "donation_planchets", + donationReceipts: "donation_receipts", + transactionsMeta: "transactions_meta", + refundItems: "refund_items", +}; + /** * One sqlite transaction. * @@ -447,9 +494,15 @@ export class SqliteWalletTransaction implements WalletDbTransaction { */ private stats: SqliteAccessStats; - /** Window applied to the root SELECT of one migration enumeration call. */ + /** Keyset page applied to the root SELECT of one migration enumeration. */ private migrationPage: - | { offset: number; limit: number; consumed: boolean } + | { + table: string; + afterRowId: number; + limit: number; + consumed: boolean; + nextRowId?: number; + } | undefined; constructor( @@ -489,30 +542,37 @@ export class SqliteWalletTransaction implements WalletDbTransaction { sql: string, params: Record<string, any> = {}, ): Promise<ResultRow[]> { + let isMigrationRoot = false; if (this.migrationPage && !this.migrationPage.consumed) { + isMigrationRoot = true; this.migrationPage.consumed = true; sql = - `SELECT * FROM (${sql}) AS migration_page` + - " LIMIT $migration_limit OFFSET $migration_offset"; + `SELECT rowid AS __migration_rowid, * FROM ${this.migrationPage.table}` + + " WHERE rowid > $migration_after" + + " ORDER BY rowid LIMIT $migration_limit"; params = { - ...params, + migration_after: this.migrationPage.afterRowId, migration_limit: this.migrationPage.limit, - migration_offset: this.migrationPage.offset, }; } const rows = await (await this.prep(sql)).getAll(params); + if (isMigrationRoot && this.migrationPage && rows.length > 0) { + this.migrationPage.nextRowId = num( + rows[rows.length - 1].__migration_rowid, + ); + } this.stats.rowsRead += rows.length; return rows; } async scanMigrationRecords<T>( - _store: WalletDbMigrationStore, + store: WalletDbMigrationStore, read: (tx: WalletDbTransaction) => Promise<T[]>, cursor: unknown | undefined, limit: number, ): Promise<WalletDbMigrationPage<T>> { - const offset = cursor === undefined ? 0 : Number(cursor); - if (!Number.isSafeInteger(offset) || offset < 0) { + const afterRowId = cursor === undefined ? 0 : Number(cursor); + if (!Number.isSafeInteger(afterRowId) || afterRowId < 0) { throw Error("invalid sqlite migration cursor"); } if (!Number.isSafeInteger(limit) || limit <= 0) { @@ -521,7 +581,12 @@ export class SqliteWalletTransaction implements WalletDbTransaction { if (this.migrationPage) { throw Error("nested migration scan is not supported"); } - this.migrationPage = { offset, limit, consumed: false }; + this.migrationPage = { + table: SQLITE_MIGRATION_TABLES[store], + afterRowId, + limit, + consumed: false, + }; try { const records = await read(this); if (!this.migrationPage.consumed) { @@ -529,7 +594,9 @@ export class SqliteWalletTransaction implements WalletDbTransaction { } return { records, - ...(records.length > 0 ? { nextCursor: offset + records.length } : {}), + ...(records.length === limit && this.migrationPage.nextRowId != null + ? { nextCursor: this.migrationPage.nextRowId } + : {}), }; } finally { this.migrationPage = undefined; @@ -746,6 +813,35 @@ export class SqliteWalletTransaction implements WalletDbTransaction { return row ? this.rowToReserve(row) : undefined; } + async getReservesByPubs(reservePubs: string[]): Promise<WalletReserve[]> { + if (reservePubs.length === 0) { + return []; + } + const params: Record<string, Uint8Array> = {}; + const blobs = reservePubs.map((pub) => crockToDb(pub)); + const placeholders = blobs.map((blob, i) => { + params[`p${i}`] = blob; + return `$p${i}`; + }); + const rows = await this.all( + `SELECT * FROM reserves WHERE reserve_pub IN (${placeholders.join(", ")})`, + params, + ); + const byPub = new Map( + rows.map((r) => { + const reservePub = r.reserve_pub; + if (!(reservePub instanceof Uint8Array)) { + throw Error("reserves.reserve_pub must be a BLOB column"); + } + return [blobKey(reservePub), r] as const; + }), + ); + return blobs.flatMap((blob) => { + const row = byPub.get(blobKey(blob)); + return row ? [this.rowToReserve(row)] : []; + }); + } + async listAllReserves(): Promise<WalletReserve[]> { const rows = await this.all("SELECT * FROM reserves"); return rows.map((r) => this.rowToReserve(r)); @@ -1418,6 +1514,7 @@ export class SqliteWalletTransaction implements WalletDbTransaction { currency: str(row.currency), value: dbAmount(row.value), freshCoinCount: num(row.fresh_coin_count), + hasFreshCoins: num(row.has_fresh_coins) === 1 ? 1 : 0, visibleCoinCount: num(row.visible_coin_count), exchangeMasterPub: dbToCrock(row.exchange_master_pub), ...(row.pending_refresh_output_count != null @@ -1446,14 +1543,18 @@ export class SqliteWalletTransaction implements WalletDbTransaction { await this.run( `INSERT INTO coin_availability ( exchange_base_url, denom_pub_hash, max_age, currency, value, - exchange_master_pub, fresh_coin_count, visible_coin_count, + exchange_master_pub, fresh_coin_count, has_fresh_coins, + visible_coin_count, pending_refresh_output_count - ) VALUES ($url, $dph, $age, $cur, $val, $emp, $fresh, $vis, $pend) + ) VALUES ( + $url, $dph, $age, $cur, $val, $emp, $fresh, $hasFresh, $vis, $pend + ) ON CONFLICT(exchange_master_pub, denom_pub_hash, max_age) DO UPDATE SET currency = excluded.currency, value = excluded.value, exchange_master_pub = excluded.exchange_master_pub, fresh_coin_count = excluded.fresh_coin_count, + has_fresh_coins = excluded.has_fresh_coins, visible_coin_count = excluded.visible_coin_count, pending_refresh_output_count = excluded.pending_refresh_output_count`, @@ -1465,6 +1566,7 @@ export class SqliteWalletTransaction implements WalletDbTransaction { val: rec.value, emp: optCrockToDb(rec.exchangeMasterPub), fresh: rec.freshCoinCount, + hasFresh: rec.freshCoinCount > 0 ? 1 : 0, vis: rec.visibleCoinCount, pend: rec.pendingRefreshOutputCount ?? null, }, @@ -1491,22 +1593,15 @@ export class SqliteWalletTransaction implements WalletDbTransaction { ageLower: number, ageUpper: number, ): Promise<WalletCoinAvailability[]> { - // Row-value comparison, not `max_age BETWEEN ... AND fresh_coin_count >= 1`. - // The IndexedDB version bounds a compound key, which compares tuples - // lexicographically, so the freshCoinCount >= 1 floor applies only at - // max_age = ageLower; a row with a higher max_age and zero fresh coins is - // inside the range. The two formulations return different rows, so this - // reproduces the tuple semantics exactly. const rows = await this.all( "SELECT * FROM coin_availability" + " WHERE exchange_base_url = $url" + - " AND (max_age, fresh_coin_count) >= ($lower, 1)" + - " AND (max_age, fresh_coin_count) <= ($upper, $maxint)", + " AND has_fresh_coins = 1" + + " AND max_age BETWEEN $lower AND $upper", { url: exchangeBaseUrl, lower: ageLower, upper: ageUpper, - maxint: Number.MAX_SAFE_INTEGER, }, ); return rows.map((r) => this.rowToCoinAvailability(r)); @@ -2798,6 +2893,31 @@ export class SqliteWalletTransaction implements WalletDbTransaction { return rows.map((r) => this.rowToTransactionMeta(r)); } + async listTransactionMetaPage(req: { + cursor?: WalletTransactionMetaCursor; + direction: "forward" | "backward"; + limit: number; + }): Promise<WalletTransactionMeta[]> { + const backwards = req.direction === "backward"; + const comparison = backwards ? "<" : ">"; + const ordering = backwards ? " DESC" : ""; + const where = req.cursor + ? ` WHERE (timestamp, transaction_id) ${comparison} ($ts, $id)` + : ""; + const rows = await this.all( + `SELECT * FROM transactions_meta${where}` + + ` ORDER BY timestamp${ordering}, transaction_id${ordering}` + + " LIMIT $limit", + { + ...(req.cursor + ? { ts: req.cursor.timestamp, id: req.cursor.transactionId } + : {}), + limit: req.limit, + }, + ); + return rows.map((r) => this.rowToTransactionMeta(r)); + } + async listTransactionMetaByStatus(req: { onlyActive: boolean; }): Promise<WalletTransactionMeta[]> { @@ -3835,12 +3955,34 @@ export class SqliteWalletTransaction implements WalletDbTransaction { } private async hydratePurchases(rows: ResultRow[]): Promise<WalletPurchase[]> { - const out: WalletPurchase[] = []; - for (const row of rows) { - const exchanges = await this.loadPurchaseExchanges(str(row.proposal_id)); - out.push(this.rowToPurchase(row, exchanges)); + if (rows.length === 0) { + return []; } - return out; + const exchangesByProposal = new Map<string, string[]>(); + const proposalIds = rows.map((r) => str(r.proposal_id)); + for (let offset = 0; offset < proposalIds.length; offset += 400) { + const chunk = proposalIds.slice(offset, offset + 400); + const params: Record<string, string> = {}; + const placeholders = chunk.map((id, i) => { + params[`id${i}`] = id; + return `$id${i}`; + }); + const exchangeRows = await this.all( + "SELECT proposal_id, exchange_base_url FROM purchase_exchanges" + + ` WHERE proposal_id IN (${placeholders.join(", ")})` + + " ORDER BY proposal_id, idx", + params, + ); + for (const exchangeRow of exchangeRows) { + const proposalId = str(exchangeRow.proposal_id); + const exchanges = exchangesByProposal.get(proposalId) ?? []; + exchanges.push(str(exchangeRow.exchange_base_url)); + exchangesByProposal.set(proposalId, exchanges); + } + } + return rows.map((row) => + this.rowToPurchase(row, exchangesByProposal.get(str(row.proposal_id))), + ); } async getPurchase(proposalId: string): Promise<WalletPurchase | undefined> { @@ -3969,19 +4111,31 @@ export class SqliteWalletTransaction implements WalletDbTransaction { turi: rec.talerUri ?? null, }, ); - // Replace the exchange rows wholesale: an update that drops an exchange - // must not leave the old row behind. + const oldExchanges = await this.loadPurchaseExchanges(rec.proposalId); + const newExchanges = rec.exchanges; + const oldList = oldExchanges ?? []; + const newList = newExchanges ?? []; + if ( + oldList.length === newList.length && + oldList.every((url, i) => url === newList[i]) + ) { + return; + } await this.run("DELETE FROM purchase_exchanges WHERE proposal_id = $id", { id: rec.proposalId, }); - if (rec.exchanges) { - for (let i = 0; i < rec.exchanges.length; i++) { - await this.run( - "INSERT INTO purchase_exchanges (proposal_id, idx," + - " exchange_base_url) VALUES ($id, $idx, $url)", - { id: rec.proposalId, idx: i, url: rec.exchanges[i] }, - ); - } + if (newExchanges?.length) { + const params: Record<string, string | number> = { id: rec.proposalId }; + const values = newExchanges.map((url, i) => { + params[`idx${i}`] = i; + params[`url${i}`] = url; + return `($id, $idx${i}, $url${i})`; + }); + await this.run( + "INSERT INTO purchase_exchanges (proposal_id, idx, exchange_base_url)" + + ` VALUES ${values.join(", ")}`, + params, + ); } } @@ -4000,6 +4154,33 @@ export class SqliteWalletTransaction implements WalletDbTransaction { ); } + async getPurchasesByIds(proposalIds: string[]): Promise<WalletPurchase[]> { + if (proposalIds.length === 0) { + return []; + } + const rows: ResultRow[] = []; + for (let offset = 0; offset < proposalIds.length; offset += 400) { + const chunk = proposalIds.slice(offset, offset + 400); + const params: Record<string, string> = {}; + const placeholders = chunk.map((id, i) => { + params[`id${i}`] = id; + return `$id${i}`; + }); + rows.push( + ...(await this.all( + `SELECT * FROM purchases WHERE proposal_id IN (${placeholders.join(", ")})`, + params, + )), + ); + } + const purchases = await this.hydratePurchases(rows); + const byId = new Map(purchases.map((p) => [p.proposalId, p])); + return proposalIds.flatMap((id) => { + const purchase = byId.get(id); + return purchase ? [purchase] : []; + }); + } + async getPurchasesByStatus( status: PurchaseStatus, ): Promise<WalletPurchase[]> { @@ -4798,6 +4979,14 @@ export class SqliteWalletTransaction implements WalletDbTransaction { return rows.map((r) => this.rowToToken(r)); } + async getTokensByFamilyHash(tokenFamilyHash: string): Promise<WalletToken[]> { + const rows = await this.all( + "SELECT * FROM tokens WHERE token_family_hash = $h", + { h: crockToDb(tokenFamilyHash) }, + ); + return rows.map((r) => this.rowToToken(r)); + } + // ----------------------------------------------------------- slates private rowToSlate(row: ResultRow): WalletSlate { diff --git a/packages/taler-wallet-core/src/dbtx.ts b/packages/taler-wallet-core/src/dbtx.ts @@ -49,6 +49,7 @@ import { WalletSlate, WalletDenomination, WalletTransactionMeta, + WalletTransactionMetaCursor, DbPreciseTimestamp, DbProtocolTimestamp, WalletOperationRetry, @@ -368,6 +369,13 @@ export interface WalletDbTransaction { limit?: number; }): Promise<WalletTransactionMeta[]>; + /** List a bounded page in stable (timestamp, transactionId) order. */ + listTransactionMetaPage(req: { + cursor?: WalletTransactionMetaCursor; + direction: "forward" | "backward"; + limit: number; + }): Promise<WalletTransactionMeta[]>; + /** * List all transaction metadata, optionally restricted to transactions in a * non-final ("active") state. @@ -719,6 +727,9 @@ export interface WalletDbTransaction { orderId: string, ): Promise<WalletPurchase | undefined>; + /** Get purchases for the requested proposal IDs, skipping missing IDs. */ + getPurchasesByIds(proposalIds: string[]): Promise<WalletPurchase[]>; + /** Get every purchase for a merchant order, including repurchases. */ getPurchasesByUrlAndOrderId( merchantBaseUrl: string, @@ -949,6 +960,9 @@ export interface WalletDbTransaction { reservePub: string, ): Promise<WalletReserve | undefined>; + /** Get reserves for the requested public keys, skipping missing keys. */ + getReservesByPubs(reservePubs: string[]): Promise<WalletReserve[]>; + /** * Create or update a reserve, returning its row id. * @@ -1024,6 +1038,9 @@ export interface WalletDbTransaction { */ getTokensByIssuePubHash(tokenIssuePubHash: string): Promise<WalletToken[]>; + /** Get all tokens belonging to a token family. */ + getTokensByFamilyHash(tokenFamilyHash: string): Promise<WalletToken[]>; + /** * Get an incoming peer pull payment (credit) record by purse public key. */ diff --git a/packages/taler-wallet-core/src/refresh.ts b/packages/taler-wallet-core/src/refresh.ts @@ -438,6 +438,7 @@ async function getCoinAvailabilityForDenom( exchangeBaseUrl: denom.exchangeBaseUrl, exchangeMasterPub: denom.exchangeMasterPub, freshCoinCount: 0, + hasFreshCoins: 0, visibleCoinCount: 0, }; }