wallet-core: simplify/unify DB access

1 files changed, 227 insertions, 245 deletions

diff --git a/packages/taler-wallet-core/src/operations/backup/index.ts b/packages/taler-wallet-core/src/operations/backup/index.ts
index e4e4e43f6..948b8eb85 100644
--- a/packages/taler-wallet-core/src/operations/backup/index.ts
+++ b/packages/taler-wallet-core/src/operations/backup/index.ts
@@ -79,7 +79,7 @@ import {
   ConfigRecord,
   ConfigRecordKey,
   WalletBackupConfState,
-  WalletStoresV1,
+  WalletDbReadOnlyTransaction,
   timestampOptionalPreciseFromDb,
   timestampPreciseToDb,
 } from "../../db.js";
@@ -88,7 +88,6 @@ import {
   checkDbInvariant,
   checkLogicInvariant,
 } from "../../util/invariants.js";
-import { GetReadOnlyAccess } from "../../util/query.js";
 import { addAttentionRequest, removeAttentionRequest } from "../attention.js";
 import {
   TaskIdentifiers,
@@ -187,11 +186,12 @@ async function runBackupCycleForProvider(
   ws: InternalWalletState,
   args: BackupForProviderArgs,
 ): Promise<TaskRunResult> {
-  const provider = await ws.db
-    .mktx((x) => [x.backupProviders])
-    .runReadOnly(async (tx) => {
+  const provider = await ws.db.runReadOnlyTx(
+    ["backupProviders"],
+    async (tx) => {
       return tx.backupProviders.get(args.backupProviderBaseUrl);
-    });
+    },
+  );
 
   if (!provider) {
     logger.warn("provider disappeared");
@@ -248,22 +248,20 @@ async function runBackupCycleForProvider(
   logger.trace(`sync response status: ${resp.status}`);
 
   if (resp.status === HttpStatusCode.NotModified) {
-    await ws.db
-      .mktx((x) => [x.backupProviders])
-      .runReadWrite(async (tx) => {
-        const prov = await tx.backupProviders.get(provider.baseUrl);
-        if (!prov) {
-          return;
-        }
-        prov.lastBackupCycleTimestamp = timestampPreciseToDb(
-          TalerPreciseTimestamp.now(),
-        );
-        prov.state = {
-          tag: BackupProviderStateTag.Ready,
-          nextBackupTimestamp: timestampPreciseToDb(getNextBackupTimestamp()),
-        };
-        await tx.backupProviders.put(prov);
-      });
+    await ws.db.runReadWriteTx(["backupProviders"], async (tx) => {
+      const prov = await tx.backupProviders.get(provider.baseUrl);
+      if (!prov) {
+        return;
+      }
+      prov.lastBackupCycleTimestamp = timestampPreciseToDb(
+        TalerPreciseTimestamp.now(),
+      );
+      prov.state = {
+        tag: BackupProviderStateTag.Ready,
+        nextBackupTimestamp: timestampPreciseToDb(getNextBackupTimestamp()),
+      };
+      await tx.backupProviders.put(prov);
+    });
 
     removeAttentionRequest(ws, {
       entityId: provider.baseUrl,
@@ -296,46 +294,42 @@ async function runBackupCycleForProvider(
     if (res === undefined) {
       //claimed
 
-      await ws.db
-        .mktx((x) => [x.backupProviders, x.operationRetries])
-        .runReadWrite(async (tx) => {
-          const prov = await tx.backupProviders.get(provider.baseUrl);
-          if (!prov) {
-            logger.warn("backup provider not found anymore");
-            return;
-          }
-          prov.shouldRetryFreshProposal = true;
-          prov.state = {
-            tag: BackupProviderStateTag.Retrying,
-          };
-          await tx.backupProviders.put(prov);
-        });
-
-      throw Error("not implemented");
-      // return {
-      //   type: TaskRunResultType.Pending,
-      // };
-    }
-    const result = res;
-
-    await ws.db
-      .mktx((x) => [x.backupProviders, x.operationRetries])
-      .runReadWrite(async (tx) => {
+      await ws.db.runReadWriteTx(["backupProviders"], async (tx) => {
         const prov = await tx.backupProviders.get(provider.baseUrl);
         if (!prov) {
           logger.warn("backup provider not found anymore");
           return;
         }
-        const opId = TaskIdentifiers.forBackup(prov);
-        //await scheduleRetryInTx(ws, tx, opId);
-        prov.currentPaymentProposalId = result.proposalId;
-        prov.shouldRetryFreshProposal = false;
+        prov.shouldRetryFreshProposal = true;
         prov.state = {
           tag: BackupProviderStateTag.Retrying,
         };
         await tx.backupProviders.put(prov);
       });
 
+      throw Error("not implemented");
+      // return {
+      //   type: TaskRunResultType.Pending,
+      // };
+    }
+    const result = res;
+
+    await ws.db.runReadWriteTx(["backupProviders"], async (tx) => {
+      const prov = await tx.backupProviders.get(provider.baseUrl);
+      if (!prov) {
+        logger.warn("backup provider not found anymore");
+        return;
+      }
+      // const opId = TaskIdentifiers.forBackup(prov);
+      // await scheduleRetryInTx(ws, tx, opId);
+      prov.currentPaymentProposalId = result.proposalId;
+      prov.shouldRetryFreshProposal = false;
+      prov.state = {
+        tag: BackupProviderStateTag.Retrying,
+      };
+      await tx.backupProviders.put(prov);
+    });
+
     addAttentionRequest(
       ws,
       {
@@ -353,23 +347,21 @@ async function runBackupCycleForProvider(
   }
 
   if (resp.status === HttpStatusCode.NoContent) {
-    await ws.db
-      .mktx((x) => [x.backupProviders])
-      .runReadWrite(async (tx) => {
-        const prov = await tx.backupProviders.get(provider.baseUrl);
-        if (!prov) {
-          return;
-        }
-        prov.lastBackupHash = encodeCrock(currentBackupHash);
-        prov.lastBackupCycleTimestamp = timestampPreciseToDb(
-          TalerPreciseTimestamp.now(),
-        );
-        prov.state = {
-          tag: BackupProviderStateTag.Ready,
-          nextBackupTimestamp: timestampPreciseToDb(getNextBackupTimestamp()),
-        };
-        await tx.backupProviders.put(prov);
-      });
+    await ws.db.runReadWriteTx(["backupProviders"], async (tx) => {
+      const prov = await tx.backupProviders.get(provider.baseUrl);
+      if (!prov) {
+        return;
+      }
+      prov.lastBackupHash = encodeCrock(currentBackupHash);
+      prov.lastBackupCycleTimestamp = timestampPreciseToDb(
+        TalerPreciseTimestamp.now(),
+      );
+      prov.state = {
+        tag: BackupProviderStateTag.Ready,
+        nextBackupTimestamp: timestampPreciseToDb(getNextBackupTimestamp()),
+      };
+      await tx.backupProviders.put(prov);
+    });
 
     removeAttentionRequest(ws, {
       entityId: provider.baseUrl,
@@ -388,24 +380,22 @@ async function runBackupCycleForProvider(
     // const blob = await decryptBackup(backupConfig, backupEnc);
     // FIXME: Re-implement backup import with merging
     // await importBackup(ws, blob, cryptoData);
-    await ws.db
-      .mktx((x) => [x.backupProviders, x.operationRetries])
-      .runReadWrite(async (tx) => {
-        const prov = await tx.backupProviders.get(provider.baseUrl);
-        if (!prov) {
-          logger.warn("backup provider not found anymore");
-          return;
-        }
-        prov.lastBackupHash = encodeCrock(hash(backupEnc));
-        // FIXME: Allocate error code for this situation?
-        // FIXME: Add operation retry record!
-        const opId = TaskIdentifiers.forBackup(prov);
-        //await scheduleRetryInTx(ws, tx, opId);
-        prov.state = {
-          tag: BackupProviderStateTag.Retrying,
-        };
-        await tx.backupProviders.put(prov);
-      });
+    await ws.db.runReadWriteTx(["backupProviders"], async (tx) => {
+      const prov = await tx.backupProviders.get(provider.baseUrl);
+      if (!prov) {
+        logger.warn("backup provider not found anymore");
+        return;
+      }
+      prov.lastBackupHash = encodeCrock(hash(backupEnc));
+      // FIXME: Allocate error code for this situation?
+      // FIXME: Add operation retry record!
+      const opId = TaskIdentifiers.forBackup(prov);
+      //await scheduleRetryInTx(ws, tx, opId);
+      prov.state = {
+        tag: BackupProviderStateTag.Retrying,
+      };
+      await tx.backupProviders.put(prov);
+    });
     logger.info("processed existing backup");
     // Now upload our own, merged backup.
     return await runBackupCycleForProvider(ws, args);
@@ -427,11 +417,12 @@ export async function processBackupForProvider(
   ws: InternalWalletState,
   backupProviderBaseUrl: string,
 ): Promise<TaskRunResult> {
-  const provider = await ws.db
-    .mktx((x) => [x.backupProviders])
-    .runReadOnly(async (tx) => {
+  const provider = await ws.db.runReadOnlyTx(
+    ["backupProviders"],
+    async (tx) => {
       return await tx.backupProviders.get(backupProviderBaseUrl);
-    });
+    },
+  );
   if (!provider) {
     throw Error("unknown backup provider");
   }
@@ -457,11 +448,9 @@ export async function removeBackupProvider(
   ws: InternalWalletState,
   req: RemoveBackupProviderRequest,
 ): Promise<void> {
-  await ws.db
-    .mktx((x) => [x.backupProviders])
-    .runReadWrite(async (tx) => {
-      await tx.backupProviders.delete(req.provider);
-    });
+  await ws.db.runReadWriteTx(["backupProviders"], async (tx) => {
+    await tx.backupProviders.delete(req.provider);
+  });
 }
 
 export interface RunBackupCycleRequest {
@@ -487,9 +476,9 @@ export async function runBackupCycle(
   ws: InternalWalletState,
   req: RunBackupCycleRequest,
 ): Promise<void> {
-  const providers = await ws.db
-    .mktx((x) => [x.backupProviders])
-    .runReadOnly(async (tx) => {
+  const providers = await ws.db.runReadOnlyTx(
+    ["backupProviders"],
+    async (tx) => {
       if (req.providers) {
         const rs = await Promise.all(
           req.providers.map((id) => tx.backupProviders.get(id)),
@@ -497,7 +486,8 @@ export async function runBackupCycle(
         return rs.filter(notEmpty);
       }
       return await tx.backupProviders.iter().toArray();
-    });
+    },
+  );
 
   for (const provider of providers) {
     await runBackupCycleForProvider(ws, {
@@ -587,62 +577,56 @@ export async function addBackupProvider(
   logger.info(`adding backup provider ${j2s(req)}`);
   await provideBackupState(ws);
   const canonUrl = canonicalizeBaseUrl(req.backupProviderBaseUrl);
-  await ws.db
-    .mktx((x) => [x.backupProviders])
-    .runReadWrite(async (tx) => {
-      const oldProv = await tx.backupProviders.get(canonUrl);
-      if (oldProv) {
-        logger.info("old backup provider found");
-        if (req.activate) {
-          oldProv.state = {
-            tag: BackupProviderStateTag.Ready,
-            nextBackupTimestamp: timestampPreciseToDb(
-              TalerPreciseTimestamp.now(),
-            ),
-          };
-          logger.info("setting existing backup provider to active");
-          await tx.backupProviders.put(oldProv);
-        }
-        return;
-      }
-    });
-  const termsUrl = new URL("config", canonUrl);
-  const resp = await ws.http.fetch(termsUrl.href);
-  const terms = await readSuccessResponseJsonOrThrow(
-    resp,
-    codecForSyncTermsOfServiceResponse(),
-  );
-  await ws.db
-    .mktx((x) => [x.backupProviders])
-    .runReadWrite(async (tx) => {
-      let state: BackupProviderState;
-      //FIXME: what is the difference provisional and ready?
+  await ws.db.runReadWriteTx(["backupProviders"], async (tx) => {
+    const oldProv = await tx.backupProviders.get(canonUrl);
+    if (oldProv) {
+      logger.info("old backup provider found");
       if (req.activate) {
-        state = {
+        oldProv.state = {
           tag: BackupProviderStateTag.Ready,
           nextBackupTimestamp: timestampPreciseToDb(
             TalerPreciseTimestamp.now(),
           ),
         };
-      } else {
-        state = {
-          tag: BackupProviderStateTag.Provisional,
-        };
+        logger.info("setting existing backup provider to active");
+        await tx.backupProviders.put(oldProv);
       }
-      await tx.backupProviders.put({
-        state,
-        name: req.name,
-        terms: {
-          annualFee: terms.annual_fee,
-          storageLimitInMegabytes: terms.storage_limit_in_megabytes,
-          supportedProtocolVersion: terms.version,
-        },
-        shouldRetryFreshProposal: false,
-        paymentProposalIds: [],
-        baseUrl: canonUrl,
-        uids: [encodeCrock(getRandomBytes(32))],
-      });
+      return;
+    }
+  });
+  const termsUrl = new URL("config", canonUrl);
+  const resp = await ws.http.fetch(termsUrl.href);
+  const terms = await readSuccessResponseJsonOrThrow(
+    resp,
+    codecForSyncTermsOfServiceResponse(),
+  );
+  await ws.db.runReadWriteTx(["backupProviders"], async (tx) => {
+    let state: BackupProviderState;
+    //FIXME: what is the difference provisional and ready?
+    if (req.activate) {
+      state = {
+        tag: BackupProviderStateTag.Ready,
+        nextBackupTimestamp: timestampPreciseToDb(TalerPreciseTimestamp.now()),
+      };
+    } else {
+      state = {
+        tag: BackupProviderStateTag.Provisional,
+      };
+    }
+    await tx.backupProviders.put({
+      state,
+      name: req.name,
+      terms: {
+        annualFee: terms.annual_fee,
+        storageLimitInMegabytes: terms.storage_limit_in_megabytes,
+        supportedProtocolVersion: terms.version,
+      },
+      shouldRetryFreshProposal: false,
+      paymentProposalIds: [],
+      baseUrl: canonUrl,
+      uids: [encodeCrock(getRandomBytes(32))],
     });
+  });
 
   return await runFirstBackupCycleForProvider(ws, {
     backupProviderBaseUrl: canonUrl,
@@ -827,9 +811,9 @@ export async function getBackupInfo(
   ws: InternalWalletState,
 ): Promise<BackupInfo> {
   const backupConfig = await provideBackupState(ws);
-  const providerRecords = await ws.db
-    .mktx((x) => [x.backupProviders, x.operationRetries])
-    .runReadOnly(async (tx) => {
+  const providerRecords = await ws.db.runReadOnlyTx(
+    ["backupProviders", "operationRetries"],
+    async (tx) => {
       return await tx.backupProviders.iter().mapAsync(async (bp) => {
         const opId = TaskIdentifiers.forBackup(bp);
         const retryRecord = await tx.operationRetries.get(opId);
@@ -838,7 +822,8 @@ export async function getBackupInfo(
           retryRecord,
         };
       });
-    });
+    },
+  );
   const providers: ProviderInfo[] = [];
   for (const x of providerRecords) {
     providers.push({
@@ -872,11 +857,12 @@ export async function getBackupRecovery(
   ws: InternalWalletState,
 ): Promise<BackupRecovery> {
   const bs = await provideBackupState(ws);
-  const providers = await ws.db
-    .mktx((x) => [x.backupProviders])
-    .runReadOnly(async (tx) => {
+  const providers = await ws.db.runReadOnlyTx(
+    ["backupProviders"],
+    async (tx) => {
       return await tx.backupProviders.iter().toArray();
-    });
+    },
+  );
   return {
     providers: providers
       .filter((x) => x.state.tag !== BackupProviderStateTag.Provisional)
@@ -894,50 +880,48 @@ async function backupRecoveryTheirs(
   ws: InternalWalletState,
   br: BackupRecovery,
 ) {
-  await ws.db
-    .mktx((x) => [x.config, x.backupProviders])
-    .runReadWrite(async (tx) => {
-      let backupStateEntry: ConfigRecord | undefined = await tx.config.get(
-        ConfigRecordKey.WalletBackupState,
-      );
-      checkDbInvariant(!!backupStateEntry);
-      checkDbInvariant(
-        backupStateEntry.key === ConfigRecordKey.WalletBackupState,
-      );
-      backupStateEntry.value.lastBackupNonce = undefined;
-      backupStateEntry.value.lastBackupTimestamp = undefined;
-      backupStateEntry.value.lastBackupCheckTimestamp = undefined;
-      backupStateEntry.value.lastBackupPlainHash = undefined;
-      backupStateEntry.value.walletRootPriv = br.walletRootPriv;
-      backupStateEntry.value.walletRootPub = encodeCrock(
-        eddsaGetPublic(decodeCrock(br.walletRootPriv)),
-      );
-      await tx.config.put(backupStateEntry);
-      for (const prov of br.providers) {
-        const existingProv = await tx.backupProviders.get(prov.url);
-        if (!existingProv) {
-          await tx.backupProviders.put({
-            baseUrl: prov.url,
-            name: prov.name,
-            paymentProposalIds: [],
-            shouldRetryFreshProposal: false,
-            state: {
-              tag: BackupProviderStateTag.Ready,
-              nextBackupTimestamp: timestampPreciseToDb(
-                TalerPreciseTimestamp.now(),
-              ),
-            },
-            uids: [encodeCrock(getRandomBytes(32))],
-          });
-        }
-      }
-      const providers = await tx.backupProviders.iter().toArray();
-      for (const prov of providers) {
-        prov.lastBackupCycleTimestamp = undefined;
-        prov.lastBackupHash = undefined;
-        await tx.backupProviders.put(prov);
+  await ws.db.runReadWriteTx(["backupProviders", "config"], async (tx) => {
+    let backupStateEntry: ConfigRecord | undefined = await tx.config.get(
+      ConfigRecordKey.WalletBackupState,
+    );
+    checkDbInvariant(!!backupStateEntry);
+    checkDbInvariant(
+      backupStateEntry.key === ConfigRecordKey.WalletBackupState,
+    );
+    backupStateEntry.value.lastBackupNonce = undefined;
+    backupStateEntry.value.lastBackupTimestamp = undefined;
+    backupStateEntry.value.lastBackupCheckTimestamp = undefined;
+    backupStateEntry.value.lastBackupPlainHash = undefined;
+    backupStateEntry.value.walletRootPriv = br.walletRootPriv;
+    backupStateEntry.value.walletRootPub = encodeCrock(
+      eddsaGetPublic(decodeCrock(br.walletRootPriv)),
+    );
+    await tx.config.put(backupStateEntry);
+    for (const prov of br.providers) {
+      const existingProv = await tx.backupProviders.get(prov.url);
+      if (!existingProv) {
+        await tx.backupProviders.put({
+          baseUrl: prov.url,
+          name: prov.name,
+          paymentProposalIds: [],
+          shouldRetryFreshProposal: false,
+          state: {
+            tag: BackupProviderStateTag.Ready,
+            nextBackupTimestamp: timestampPreciseToDb(
+              TalerPreciseTimestamp.now(),
+            ),
+          },
+          uids: [encodeCrock(getRandomBytes(32))],
+        });
       }
-    });
+    }
+    const providers = await tx.backupProviders.iter().toArray();
+    for (const prov of providers) {
+      prov.lastBackupCycleTimestamp = undefined;
+      prov.lastBackupHash = undefined;
+      await tx.backupProviders.put(prov);
+    }
+  });
 }
 
 async function backupRecoveryOurs(ws: InternalWalletState, br: BackupRecovery) {
@@ -949,11 +933,12 @@ export async function loadBackupRecovery(
   br: RecoveryLoadRequest,
 ): Promise<void> {
   const bs = await provideBackupState(ws);
-  const providers = await ws.db
-    .mktx((x) => [x.backupProviders])
-    .runReadOnly(async (tx) => {
+  const providers = await ws.db.runReadOnlyTx(
+    ["backupProviders"],
+    async (tx) => {
       return await tx.backupProviders.iter().toArray();
-    });
+    },
+  );
   let strategy = br.strategy;
   if (
     br.recovery.walletRootPriv != bs.walletRootPriv &&
@@ -996,11 +981,12 @@ export async function decryptBackup(
 export async function provideBackupState(
   ws: InternalWalletState,
 ): Promise<WalletBackupConfState> {
-  const bs: ConfigRecord | undefined = await ws.db
-    .mktx((stores) => [stores.config])
-    .runReadOnly(async (tx) => {
+  const bs: ConfigRecord | undefined = await ws.db.runReadOnlyTx(
+    ["config"],
+    async (tx) => {
       return await tx.config.get(ConfigRecordKey.WalletBackupState);
-    });
+    },
+  );
   if (bs) {
     checkDbInvariant(bs.key === ConfigRecordKey.WalletBackupState);
     return bs.value;
@@ -1012,34 +998,32 @@ export async function provideBackupState(
   // FIXME: device ID should be configured when wallet is initialized
   // and be based on hostname
   const deviceId = `wallet-core-${encodeCrock(d)}`;
-  return await ws.db
-    .mktx((x) => [x.config])
-    .runReadWrite(async (tx) => {
-      let backupStateEntry: ConfigRecord | undefined = await tx.config.get(
-        ConfigRecordKey.WalletBackupState,
-      );
-      if (!backupStateEntry) {
-        backupStateEntry = {
-          key: ConfigRecordKey.WalletBackupState,
-          value: {
-            deviceId,
-            walletRootPub: k.pub,
-            walletRootPriv: k.priv,
-            lastBackupPlainHash: undefined,
-          },
-        };
-        await tx.config.put(backupStateEntry);
-      }
-      checkDbInvariant(
-        backupStateEntry.key === ConfigRecordKey.WalletBackupState,
-      );
-      return backupStateEntry.value;
-    });
+  return await ws.db.runReadWriteTx(["config"], async (tx) => {
+    let backupStateEntry: ConfigRecord | undefined = await tx.config.get(
+      ConfigRecordKey.WalletBackupState,
+    );
+    if (!backupStateEntry) {
+      backupStateEntry = {
+        key: ConfigRecordKey.WalletBackupState,
+        value: {
+          deviceId,
+          walletRootPub: k.pub,
+          walletRootPriv: k.priv,
+          lastBackupPlainHash: undefined,
+        },
+      };
+      await tx.config.put(backupStateEntry);
+    }
+    checkDbInvariant(
+      backupStateEntry.key === ConfigRecordKey.WalletBackupState,
+    );
+    return backupStateEntry.value;
+  });
 }
 
 export async function getWalletBackupState(
   ws: InternalWalletState,
-  tx: GetReadOnlyAccess<{ config: typeof WalletStoresV1.config }>,
+  tx: WalletDbReadOnlyTransaction<["config"]>,
 ): Promise<WalletBackupConfState> {
   const bs = await tx.config.get(ConfigRecordKey.WalletBackupState);
   checkDbInvariant(!!bs, "wallet backup state should be in DB");
@@ -1052,21 +1036,19 @@ export async function setWalletDeviceId(
   deviceId: string,
 ): Promise<void> {
   await provideBackupState(ws);
-  await ws.db
-    .mktx((x) => [x.config])
-    .runReadWrite(async (tx) => {
-      let backupStateEntry: ConfigRecord | undefined = await tx.config.get(
-        ConfigRecordKey.WalletBackupState,
-      );
-      if (
-        !backupStateEntry ||
-        backupStateEntry.key !== ConfigRecordKey.WalletBackupState
-      ) {
-        return;
-      }
-      backupStateEntry.value.deviceId = deviceId;
-      await tx.config.put(backupStateEntry);
-    });
+  await ws.db.runReadWriteTx(["config"], async (tx) => {
+    let backupStateEntry: ConfigRecord | undefined = await tx.config.get(
+      ConfigRecordKey.WalletBackupState,
+    );
+    if (
+      !backupStateEntry ||
+      backupStateEntry.key !== ConfigRecordKey.WalletBackupState
+    ) {
+      return;
+    }
+    backupStateEntry.value.deviceId = deviceId;
+    await tx.config.put(backupStateEntry);
+  });
 }
 
 export async function getWalletDeviceId(