harness,wallet-cli: notification-based testing with RPC wallet

packages/taler-harness/src/harness/harness.ts

@@ -21,8 +21,6 @@
  * @author Florian Dold <dold@taler.net>
  */
 
-const logger = new Logger("harness.ts");
-
 /**
  * Imports
  */
@@ -43,6 +41,7 @@ import {
   parsePaytoUri,
   stringToBytes,
   TalerProtocolDuration,
+  WalletNotification,
 } from "@gnu-taler/taler-util";
 import {
   BankAccessApi,
@@ -57,9 +56,9 @@ import {
 import { deepStrictEqual } from "assert";
 import axiosImp, { AxiosError } from "axios";
 import { ChildProcess, spawn } from "child_process";
-import * as child_process from "child_process";
 import * as fs from "fs";
 import * as http from "http";
+import * as net from "node:net";
 import * as path from "path";
 import * as readline from "readline";
 import { URL } from "url";
@@ -76,6 +75,15 @@ import {
   TipCreateRequest,
   TippingReserveStatus,
 } from "./merchantApiTypes.js";
+import {
+  createRemoteWallet,
+  getClientFromRemoteWallet,
+  makeNotificationWaiter,
+  RemoteWallet,
+  WalletNotificationWaiter,
+} from "@gnu-taler/taler-wallet-core/remote";
+
+const logger = new Logger("harness.ts");
 
 const axios = axiosImp.default;
 
@@ -1831,7 +1839,7 @@ export async function runTestWithState(
 
   const handleSignal = (s: string) => {
     logger.warn(
-      `**** received fatal process event, terminating test ${testName}`,
+      `**** received fatal process event (${s}), terminating test ${testName}`,
     );
     gc.shutdownSync();
     process.exit(1);
@@ -1885,6 +1893,107 @@ export interface WalletCliOpts {
   cryptoWorkerType?: "sync" | "node-worker-thread";
 }
 
+function tryUnixConnect(socketPath: string): Promise<void> {
+  return new Promise((resolve, reject) => {
+    const client = net.createConnection(socketPath);
+    client.on("error", (e) => {
+      reject(e);
+    });
+    client.on("connect", () => {
+      client.end();
+      resolve();
+    });
+  });
+}
+
+export class WalletService {
+  walletProc: ProcessWrapper | undefined;
+
+  constructor(private globalState: GlobalTestState, private name: string) {}
+
+  get socketPath() {
+    const unixPath = path.join(this.globalState.testDir, `${this.name}.sock`);
+    return unixPath;
+  }
+
+  async start(): Promise<void> {
+    const dbPath = path.join(
+      this.globalState.testDir,
+      `walletdb-${this.name}.json`,
+    );
+    const unixPath = this.socketPath;
+    this.globalState.spawnService(
+      "taler-wallet-cli",
+      [
+        "--wallet-db",
+        dbPath,
+        "advanced",
+        "serve",
+        "--unix-path",
+        unixPath,
+      ],
+      `wallet-${this.name}`,
+    );
+  }
+
+  async pingUntilAvailable(): Promise<void> {
+    while (1) {
+      try {
+        await tryUnixConnect(this.socketPath);
+      } catch (e) {
+        logger.info(`connection attempt failed: ${e}`);
+        await delayMs(200);
+        continue;
+      }
+      logger.info("connection to wallet-core succeeded");
+      break;
+    }
+  }
+}
+
+export interface WalletClientArgs {
+  unixPath: string;
+  onNotification?(n: WalletNotification): void;
+}
+
+export class WalletClient {
+  remoteWallet: RemoteWallet | undefined = undefined;
+  waiter: WalletNotificationWaiter = makeNotificationWaiter();
+
+  constructor(private args: WalletClientArgs) {}
+
+  async connect(): Promise<void> {
+    const waiter = this.waiter;
+    const walletClient = this;
+    const w = await createRemoteWallet({
+      socketFilename: this.args.unixPath,
+      notificationHandler(n) {
+        if (walletClient.args.onNotification) {
+          walletClient.args.onNotification(n);
+        }
+        waiter.notify(n);
+        console.log("got notification from wallet-core in WalletClient");
+      },
+    });
+    this.remoteWallet = w;
+
+    this.waiter.waitForNotificationCond;
+  }
+
+  get client() {
+    if (!this.remoteWallet) {
+      throw Error("wallet not connected");
+    }
+    return getClientFromRemoteWallet(this.remoteWallet);
+  }
+
+  waitForNotificationCond(
+    cond: (n: WalletNotification) => boolean,
+  ): Promise<void> {
+    return this.waiter.waitForNotificationCond(cond);
+  }
+}
+
 export class WalletCli {
   private currentTimetravel: Duration | undefined;
   private _client: WalletCoreApiClient;

packages/taler-harness/src/harness/helpers.ts

@@ -180,6 +180,114 @@ export async function createSimpleTestkudosEnvironment(
   };
 }
 
+/**
+ * Run a test case with a simple TESTKUDOS Taler environment, consisting
+ * of one exchange, one bank and one merchant.
+ *
+ * V2 uses a daemonized wallet instead of the CLI wallet.
+ */
+export async function createSimpleTestkudosEnvironmentV2(
+  t: GlobalTestState,
+  coinConfig: CoinConfig[] = defaultCoinConfig.map((x) => x("TESTKUDOS")),
+  opts: EnvOptions = {},
+): Promise<SimpleTestEnvironment> {
+  const db = await setupDb(t);
+
+  const bank = await BankService.create(t, {
+    allowRegistrations: true,
+    currency: "TESTKUDOS",
+    database: db.connStr,
+    httpPort: 8082,
+  });
+
+  const exchange = ExchangeService.create(t, {
+    name: "testexchange-1",
+    currency: "TESTKUDOS",
+    httpPort: 8081,
+    database: db.connStr,
+  });
+
+  const merchant = await MerchantService.create(t, {
+    name: "testmerchant-1",
+    currency: "TESTKUDOS",
+    httpPort: 8083,
+    database: db.connStr,
+  });
+
+  const exchangeBankAccount = await bank.createExchangeAccount(
+    "myexchange",
+    "x",
+  );
+  await exchange.addBankAccount("1", exchangeBankAccount);
+
+  bank.setSuggestedExchange(exchange, exchangeBankAccount.accountPaytoUri);
+
+  await bank.start();
+
+  await bank.pingUntilAvailable();
+
+  const ageMaskSpec = opts.ageMaskSpec;
+
+  if (ageMaskSpec) {
+    exchange.enableAgeRestrictions(ageMaskSpec);
+    // Enable age restriction for all coins.
+    exchange.addCoinConfigList(
+      coinConfig.map((x) => ({
+        ...x,
+        name: `${x.name}-age`,
+        ageRestricted: true,
+      })),
+    );
+    // For mixed age restrictions, we also offer coins without age restrictions
+    if (opts.mixedAgeRestriction) {
+      exchange.addCoinConfigList(
+        coinConfig.map((x) => ({ ...x, ageRestricted: false })),
+      );
+    }
+  } else {
+    exchange.addCoinConfigList(coinConfig);
+  }
+
+  await exchange.start();
+  await exchange.pingUntilAvailable();
+
+  merchant.addExchange(exchange);
+
+  await merchant.start();
+  await merchant.pingUntilAvailable();
+
+  await merchant.addInstance({
+    id: "default",
+    name: "Default Instance",
+    paytoUris: [getPayto("merchant-default")],
+    defaultWireTransferDelay: Duration.toTalerProtocolDuration(
+      Duration.fromSpec({ minutes: 1 }),
+    ),
+  });
+
+  await merchant.addInstance({
+    id: "minst1",
+    name: "minst1",
+    paytoUris: [getPayto("minst1")],
+    defaultWireTransferDelay: Duration.toTalerProtocolDuration(
+      Duration.fromSpec({ minutes: 1 }),
+    ),
+  });
+
+  console.log("setup done!");
+
+  const wallet = new WalletCli(t);
+
+  return {
+    commonDb: db,
+    exchange,
+    merchant,
+    wallet,
+    bank,
+    exchangeBankAccount,
+  };
+}
+
 export interface FaultyMerchantTestEnvironment {
   commonDb: DbInfo;
   bank: BankService;

packages/taler-harness/src/integrationtests/test-wallet-notifications.ts

@@ -0,0 +1,163 @@
+/*
+ This file is part of GNU Taler
+ (C) 2020 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 <http://www.gnu.org/licenses/>
+ */
+
+/**
+ * Imports.
+ */
+import {
+  Amounts,
+  Duration,
+  NotificationType,
+  PreparePayResultType,
+} from "@gnu-taler/taler-util";
+import {
+  BankAccessApi,
+  BankApi,
+  WalletApiOperation,
+} from "@gnu-taler/taler-wallet-core";
+import { CoinConfig, defaultCoinConfig } from "../harness/denomStructures.js";
+import {
+  ExchangeService,
+  FakebankService,
+  getRandomIban,
+  GlobalTestState,
+  MerchantService,
+  setupDb,
+  WalletClient,
+  WalletService,
+} from "../harness/harness.js";
+
+/**
+ * Test for wallet-core notifications.
+ */
+export async function runWalletNotificationsTest(t: GlobalTestState) {
+  // Set up test environment
+
+  const db = await setupDb(t);
+
+  const bank = await FakebankService.create(t, {
+    allowRegistrations: true,
+    currency: "TESTKUDOS",
+    database: db.connStr,
+    httpPort: 8082,
+  });
+
+  const exchange = ExchangeService.create(t, {
+    name: "testexchange-1",
+    currency: "TESTKUDOS",
+    httpPort: 8081,
+    database: db.connStr,
+  });
+
+  const merchant = await MerchantService.create(t, {
+    name: "testmerchant-1",
+    currency: "TESTKUDOS",
+    httpPort: 8083,
+    database: db.connStr,
+  });
+
+  const exchangeBankAccount = await bank.createExchangeAccount(
+    "myexchange",
+    "x",
+  );
+  exchange.addBankAccount("1", exchangeBankAccount);
+
+  bank.setSuggestedExchange(exchange, exchangeBankAccount.accountPaytoUri);
+
+  await bank.start();
+
+  await bank.pingUntilAvailable();
+
+  const coinConfig: CoinConfig[] = defaultCoinConfig.map((x) => x("TESTKUDOS"));
+  exchange.addCoinConfigList(coinConfig);
+
+  await exchange.start();
+  await exchange.pingUntilAvailable();
+
+  merchant.addExchange(exchange);
+
+  await merchant.start();
+  await merchant.pingUntilAvailable();
+
+  // Fakebank uses x-taler-bank, but merchant is configured to only accept sepa!
+  const label = "mymerchant";
+  await merchant.addInstance({
+    id: "default",
+    name: "Default Instance",
+    paytoUris: [
+      `payto://iban/SANDBOXX/${getRandomIban(label)}?receiver-name=${label}`,
+    ],
+    defaultWireTransferDelay: Duration.toTalerProtocolDuration(
+      Duration.fromSpec({ minutes: 1 }),
+    ),
+  });
+
+  console.log("setup done!");
+
+  const walletService = new WalletService(t, "wallet");
+  await walletService.start();
+  await walletService.pingUntilAvailable();
+
+  const walletClient = new WalletClient({
+    unixPath: walletService.socketPath,
+    onNotification(n) {
+      console.log("got notification", n);
+    },
+  });
+  await walletClient.connect();
+  await walletClient.client.call(WalletApiOperation.InitWallet, {
+    skipDefaults: true,
+  });
+
+  const user = await BankApi.createRandomBankUser(bank);
+  const wop = await BankAccessApi.createWithdrawalOperation(
+    bank,
+    user,
+    "TESTKUDOS:20",
+  );
+
+  // Hand it to the wallet
+
+  await walletClient.client.call(
+    WalletApiOperation.GetWithdrawalDetailsForUri,
+    {
+      talerWithdrawUri: wop.taler_withdraw_uri,
+    },
+  );
+
+  // Withdraw (AKA select)
+
+  const withdrawalFinishedReceivedPromise =
+    walletClient.waitForNotificationCond((x) => {
+      return x.type === NotificationType.WithdrawGroupFinished;
+    });
+
+  await walletClient.client.call(
+    WalletApiOperation.AcceptBankIntegratedWithdrawal,
+    {
+      exchangeBaseUrl: exchange.baseUrl,
+      talerWithdrawUri: wop.taler_withdraw_uri,
+    },
+  );
+
+  // Confirm it
+
+  await BankApi.confirmWithdrawalOperation(bank, user, wop);
+
+  await withdrawalFinishedReceivedPromise;
+}
+
+runWalletNotificationsTest.suites = ["wallet"];

packages/taler-harness/src/integrationtests/testrunner.ts

@@ -92,13 +92,14 @@ import { runWithdrawalBankIntegratedTest } from "./test-withdrawal-bank-integrat
 import { runWithdrawalFakebankTest } from "./test-withdrawal-fakebank.js";
 import { runTestWithdrawalManualTest } from "./test-withdrawal-manual.js";
 import { runAgeRestrictionsPeerTest } from "./test-age-restrictions-peer.js";
-import { runWalletBalanceTest } from "./test-wallet-balance.js";
+import { runWalletNotificationsTest } from "./test-wallet-notifications.js";
 import { runAgeRestrictionsMixedMerchantTest } from "./test-age-restrictions-mixed-merchant.js";
 import { runWalletCryptoWorkerTest } from "./test-wallet-cryptoworker.js";
 import { runWithdrawalHighTest } from "./test-withdrawal-high.js";
 import { runKycTest } from "./test-kyc.js";
 import { runPaymentAbortTest } from "./test-payment-abort.js";
 import { runWithdrawalFeesTest } from "./test-withdrawal-fees.js";
+import { runWalletBalanceTest } from "./test-wallet-balance.js";
 
 /**
  * Test runner.
@@ -166,6 +167,7 @@ const allTests: TestMainFunction[] = [
   runPaymentTransientTest,
   runPaymentZeroTest,
   runPayPaidTest,
+  runWalletBalanceTest,
   runPaywallFlowTest,
   runPeerToPeerPullTest,
   runPeerToPeerPushTest,
@@ -180,7 +182,7 @@ const allTests: TestMainFunction[] = [
   runTippingTest,
   runWalletBackupBasicTest,
   runWalletBackupDoublespendTest,
-  runWalletBalanceTest,
+  runWalletNotificationsTest,
   runWalletCryptoWorkerTest,
   runWalletDblessTest,
   runWallettestingTest,

packages/taler-util/src/twrpc-impl.node.ts

@@ -54,6 +54,9 @@ export async function connectRpc<T>(args: RpcConnectArgs<T>): Promise<T> {
   let sockFilename = args.socketFilename;
   return new Promise((resolve, reject) => {
     const client = net.createConnection(sockFilename);
+    client.on("error", (e) => {
+      reject(e);
+    });
     client.on("connect", () => {
       let parsingBody: string | undefined = undefined;
       let bodyChunks: string[] = [];
@@ -102,7 +105,8 @@ export async function connectRpc<T>(args: RpcConnectArgs<T>): Promise<T> {
               try {
                 reqJson = JSON.parse(req);
               } catch (e) {
-                logger.warn("JSON request was invalid");
+                logger.warn("JSON message from server was invalid");
+                logger.info(`message was: ${req}`);
               }
               if (reqJson !== undefined) {
                 logger.info(`request: ${req}`);
@@ -112,6 +116,7 @@ export async function connectRpc<T>(args: RpcConnectArgs<T>): Promise<T> {
                 client.end();
               }
               bodyChunks = [];
+              parsingBody = undefined;
             } else {
               bodyChunks.push(lineStr);
             }
@@ -187,7 +192,7 @@ export async function runRpcServer(args: RpcServerArgs): Promise<void> {
             try {
               reqJson = JSON.parse(req);
             } catch (e) {
-              logger.warn("JSON request was invalid");
+              logger.warn("JSON request from client was invalid");
             }
             if (reqJson !== undefined) {
               logger.info(`request: ${req}`);
@@ -197,6 +202,7 @@ export async function runRpcServer(args: RpcServerArgs): Promise<void> {
               sock.end();
             }
             bodyChunks = [];
+            parsingBody = undefined;
           } else {
             bodyChunks.push(lineStr);
           }
@@ -217,6 +223,6 @@ export async function runRpcServer(args: RpcServerArgs): Promise<void> {
         handlers.onDisconnect();
       });
     });
-    server.listen("wallet-core.sock");
+    server.listen(args.socketFilename);
   });
 }

packages/taler-util/src/twrpc.ts

@@ -14,6 +14,8 @@
  GNU Taler; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
  */
 
+import { CoreApiResponse } from "./wallet-types.js";
+
 /**
  * Implementation for the wallet-core IPC protocol.
  *

packages/taler-wallet-cli/src/index.ts

@@ -60,13 +60,15 @@ import {
   WalletCoreApiClient,
   walletCoreDebugFlags,
 } from "@gnu-taler/taler-wallet-core";
+
+import {
+  createRemoteWallet,
+  getClientFromRemoteWallet,
+  makeNotificationWaiter,
+} from "@gnu-taler/taler-wallet-core/remote";
 import fs from "fs";
 import os from "os";
-import {
+import { JsonMessage, runRpcServer } from "@gnu-taler/taler-util/twrpc";
-  connectRpc,
-  JsonMessage,
-  runRpcServer,
-} from "@gnu-taler/taler-util/twrpc";
 
 // This module also serves as the entry point for the crypto
 // thread worker, and thus must expose these two handlers.
@@ -280,162 +282,33 @@ async function createLocalWallet(
   }
 }
 
-export interface RemoteWallet {
-  /**
-   * Low-level interface for making API requests to wallet-core.
-   */
-  makeCoreApiRequest(
-    operation: string,
-    payload: unknown,
-  ): Promise<CoreApiResponse>;
-
-  /**
-   * Close the connection to the remote wallet.
-   */
-  close(): void;
-}
-
-async function createRemoteWallet(
-  notificationHandler?: (n: WalletNotification) => void,
-): Promise<RemoteWallet> {
-  let nextRequestId = 1;
-  let requestMap: Map<
-    string,
-    {
-      promiseCapability: OpenedPromise<CoreApiResponse>;
-    }
-  > = new Map();
-
-  const ctx = await connectRpc<RemoteWallet>({
-    socketFilename: "wallet-core.sock",
-    onEstablished(connection) {
-      const ctx: RemoteWallet = {
-        makeCoreApiRequest(operation, payload) {
-          const id = `req-${nextRequestId}`;
-          const req: CoreApiRequestEnvelope = {
-            operation,
-            id,
-            args: payload,
-          };
-          const promiseCap = openPromise<CoreApiResponse>();
-          requestMap.set(id, {
-            promiseCapability: promiseCap,
-          });
-          connection.sendMessage(req as unknown as JsonMessage);
-          return promiseCap.promise;
-        },
-        close() {
-          connection.close();
-        },
-      };
-      return {
-        result: ctx,
-        onDisconnect() {
-          logger.info("remote wallet disconnected");
-        },
-        onMessage(m) {
-          // FIXME: use a codec for parsing the response envelope!
-
-          logger.info(`got message from remote wallet: ${j2s(m)}`);
-          if (typeof m !== "object" || m == null) {
-            logger.warn("message from wallet not understood (wrong type)");
-            return;
-          }
-          const type = (m as any).type;
-          if (type === "response" || type === "error") {
-            const id = (m as any).id;
-            if (typeof id !== "string") {
-              logger.warn(
-                "message from wallet not understood (no id in response)",
-              );
-              return;
-            }
-            const h = requestMap.get(id);
-            if (!h) {
-              logger.warn(`no handler registered for response id ${id}`);
-              return;
-            }
-            h.promiseCapability.resolve(m as any);
-          } else if (type === "notification") {
-            logger.info("got notification");
-            if (notificationHandler) {
-              notificationHandler((m as any).payload);
-            }
-          } else {
-            logger.warn("message from wallet not understood");
-          }
-        },
-      };
-    },
-  });
-  return ctx;
-}
-
-/**
- * Get a high-level API client from a remove wallet.
- */
-function getClientFromRemoteWallet(w: RemoteWallet): WalletCoreApiClient {
-  const client: WalletCoreApiClient = {
-    async call(op, payload): Promise<any> {
-      const res = await w.makeCoreApiRequest(op, payload);
-      switch (res.type) {
-        case "error":
-          throw TalerError.fromUncheckedDetail(res.error);
-        case "response":
-          return res.result;
-      }
-    },
-  };
-  return client;
-}
-
 async function withWallet<T>(
   walletCliArgs: WalletCliArgsType,
   f: (ctx: WalletContext) => Promise<T>,
 ): Promise<T> {
-  // Bookkeeping for waiting on notification conditions
+  const waiter = makeNotificationWaiter();
-  let nextCondIndex = 1;
-  const condMap: Map<
-    number,
-    {
-      condition: (n: WalletNotification) => boolean;
-      promiseCapability: OpenedPromise<void>;
-    }
-  > = new Map();
-  function onNotification(n: WalletNotification) {
-    condMap.forEach((cond, condKey) => {
-      if (cond.condition(n)) {
-        cond.promiseCapability.resolve();
-      }
-    });
-  }
-  function waitForNotificationCond(cond: (n: WalletNotification) => boolean) {
-    const promCap = openPromise<void>();
-    condMap.set(nextCondIndex++, {
-      condition: cond,
-      promiseCapability: promCap,
-    });
-    return promCap.promise;
-  }
 
   if (walletCliArgs.wallet.walletConnection) {
     logger.info("creating remote wallet");
-    const w = await createRemoteWallet(onNotification);
+    const w = await createRemoteWallet({
+      notificationHandler: waiter.notify,
+      socketFilename: walletCliArgs.wallet.walletConnection,
+    });
     const ctx: WalletContext = {
       makeCoreApiRequest(operation, payload) {
         return w.makeCoreApiRequest(operation, payload);
       },
       client: getClientFromRemoteWallet(w),
-      waitForNotificationCond,
+      waitForNotificationCond: waiter.waitForNotificationCond,
     };
     const res = await f(ctx);
     w.close();
     return res;
   } else {
-    const w = await createLocalWallet(walletCliArgs, onNotification);
+    const w = await createLocalWallet(walletCliArgs, waiter.notify);
     const ctx: WalletContext = {
       client: w.client,
-      waitForNotificationCond,
+      waitForNotificationCond: waiter.waitForNotificationCond,
       makeCoreApiRequest(operation, payload) {
         return w.handleCoreApiRequest(operation, "my-req", payload);
       },
@@ -1053,7 +926,11 @@ advancedCli
   .subcommand("serve", "serve", {
     help: "Serve the wallet API via a unix domain socket.",
   })
+  .requiredOption("unixPath", ["--unix-path"], clk.STRING, {
+    default: "wallet-core.sock",
+  })
   .action(async (args) => {
+    logger.info(`serving at ${args.serve.unixPath}`);
     const w = await createLocalWallet(args);
     w.runTaskLoop()
       .then((res) => {
@@ -1070,7 +947,7 @@ advancedCli
       });
     });
     await runRpcServer({
-      socketFilename: "wallet-core.sock",
+      socketFilename: args.serve.unixPath,
       onConnect(client) {
         logger.info("connected");
         const clientId = nextClientId++;

packages/taler-wallet-core/package.json

@@ -36,6 +36,9 @@
       "browser": "./lib/index.browser.js",
       "node": "./lib/index.node.js",
       "default": "./lib/index.js"
+    },
+    "./remote": {
+      "node": "./lib/remote.js"
     }
   },
   "devDependencies": {

packages/taler-wallet-core/src/remote.ts

@@ -0,0 +1,187 @@
+/*
+ This file is part of GNU Taler
+ (C) 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 <http://www.gnu.org/licenses/>
+ */
+
+import {
+  CoreApiRequestEnvelope,
+  CoreApiResponse,
+  j2s,
+  Logger,
+  WalletNotification,
+} from "@gnu-taler/taler-util";
+import { connectRpc, JsonMessage } from "@gnu-taler/taler-util/twrpc";
+import { TalerError } from "./errors.js";
+import { OpenedPromise, openPromise } from "./index.js";
+import { WalletCoreApiClient } from "./wallet-api-types.js";
+
+const logger = new Logger("remote.ts");
+
+export interface RemoteWallet {
+  /**
+   * Low-level interface for making API requests to wallet-core.
+   */
+  makeCoreApiRequest(
+    operation: string,
+    payload: unknown,
+  ): Promise<CoreApiResponse>;
+
+  /**
+   * Close the connection to the remote wallet.
+   */
+  close(): void;
+}
+
+export interface RemoteWalletConnectArgs {
+  socketFilename: string;
+  notificationHandler?: (n: WalletNotification) => void;
+}
+
+export async function createRemoteWallet(
+  args: RemoteWalletConnectArgs,
+): Promise<RemoteWallet> {
+  let nextRequestId = 1;
+  let requestMap: Map<
+    string,
+    {
+      promiseCapability: OpenedPromise<CoreApiResponse>;
+    }
+  > = new Map();
+
+  const ctx = await connectRpc<RemoteWallet>({
+    socketFilename: args.socketFilename,
+    onEstablished(connection) {
+      const ctx: RemoteWallet = {
+        makeCoreApiRequest(operation, payload) {
+          const id = `req-${nextRequestId}`;
+          const req: CoreApiRequestEnvelope = {
+            operation,
+            id,
+            args: payload,
+          };
+          const promiseCap = openPromise<CoreApiResponse>();
+          requestMap.set(id, {
+            promiseCapability: promiseCap,
+          });
+          connection.sendMessage(req as unknown as JsonMessage);
+          return promiseCap.promise;
+        },
+        close() {
+          connection.close();
+        },
+      };
+      return {
+        result: ctx,
+        onDisconnect() {
+          logger.info("remote wallet disconnected");
+        },
+        onMessage(m) {
+          // FIXME: use a codec for parsing the response envelope!
+
+          logger.info(`got message from remote wallet: ${j2s(m)}`);
+          if (typeof m !== "object" || m == null) {
+            logger.warn("message from wallet not understood (wrong type)");
+            return;
+          }
+          const type = (m as any).type;
+          if (type === "response" || type === "error") {
+            const id = (m as any).id;
+            if (typeof id !== "string") {
+              logger.warn(
+                "message from wallet not understood (no id in response)",
+              );
+              return;
+            }
+            const h = requestMap.get(id);
+            if (!h) {
+              logger.warn(`no handler registered for response id ${id}`);
+              return;
+            }
+            h.promiseCapability.resolve(m as any);
+          } else if (type === "notification") {
+            logger.info("got notification");
+            if (args.notificationHandler) {
+              args.notificationHandler((m as any).payload);
+            }
+          } else {
+            logger.warn("message from wallet not understood");
+          }
+        },
+      };
+    },
+  });
+  return ctx;
+}
+
+/**
+ * Get a high-level API client from a remove wallet.
+ */
+export function getClientFromRemoteWallet(
+  w: RemoteWallet,
+): WalletCoreApiClient {
+  const client: WalletCoreApiClient = {
+    async call(op, payload): Promise<any> {
+      const res = await w.makeCoreApiRequest(op, payload);
+      switch (res.type) {
+        case "error":
+          throw TalerError.fromUncheckedDetail(res.error);
+        case "response":
+          return res.result;
+      }
+    },
+  };
+  return client;
+}
+
+export interface WalletNotificationWaiter {
+  notify(wn: WalletNotification): void;
+  waitForNotificationCond(
+    cond: (n: WalletNotification) => boolean,
+  ): Promise<void>;
+}
+
+/**
+ * Helper that allows creating a promise that resolves when the
+ * wallet
+ */
+export function makeNotificationWaiter(): WalletNotificationWaiter {
+  // Bookkeeping for waiting on notification conditions
+  let nextCondIndex = 1;
+  const condMap: Map<
+    number,
+    {
+      condition: (n: WalletNotification) => boolean;
+      promiseCapability: OpenedPromise<void>;
+    }
+  > = new Map();
+  function onNotification(n: WalletNotification) {
+    condMap.forEach((cond, condKey) => {
+      if (cond.condition(n)) {
+        cond.promiseCapability.resolve();
+      }
+    });
+  }
+  function waitForNotificationCond(cond: (n: WalletNotification) => boolean) {
+    const promCap = openPromise<void>();
+    condMap.set(nextCondIndex++, {
+      condition: cond,
+      promiseCapability: promCap,
+    });
+    return promCap.promise;
+  }
+  return {
+    waitForNotificationCond,
+    notify: onNotification,
+  };
+}