observability

1 files changed, 39 insertions, 8 deletions

diff --git a/packages/taler-wallet-core/src/wallet.ts b/packages/taler-wallet-core/src/wallet.ts
index a779fdcec..9185ee48c 100644
--- a/packages/taler-wallet-core/src/wallet.ts
+++ b/packages/taler-wallet-core/src/wallet.ts
@@ -42,7 +42,9 @@ import {
   ListGlobalCurrencyAuditorsResponse,
   ListGlobalCurrencyExchangesResponse,
   Logger,
+  NotificationType,
   ObservabilityContext,
+  ObservabilityEventType,
   OpenedPromise,
   PrepareWithdrawExchangeRequest,
   PrepareWithdrawExchangeResponse,
@@ -199,6 +201,10 @@ import {
   getMaxPeerPushAmount,
 } from "./instructedAmountConversion.js";
 import {
+  ObservableHttpClientLibrary,
+  ObservableTaskScheduler,
+} from "./observable-wrappers.js";
+import {
   confirmPay,
   getContractTermsDetails,
   preparePayForTemplate,
@@ -225,7 +231,7 @@ import {
 } from "./pay-peer-push-debit.js";
 import { DbAccess } from "./query.js";
 import { forceRefresh } from "./refresh.js";
-import { TaskScheduler } from "./shepherd.js";
+import { TaskScheduler, TaskSchedulerImpl } from "./shepherd.js";
 import {
   runIntegrationTest,
   runIntegrationTest2,
@@ -1365,21 +1371,42 @@ async function handleCoreApiRequest(
   id: string,
   payload: unknown,
 ): Promise<CoreApiResponse> {
+  const oc: ObservabilityContext = {
+    observe(evt) {
+      if (ws.config.testing.emitObservabilityEvents) {
+        ws.notify({
+          type: NotificationType.RequestObservabilityEvent,
+          operation,
+          requestId: id,
+          event: evt,
+        });
+      }
+    },
+  };
+
   const wex: WalletExecutionContext = {
     ws,
     cancellationToken: CancellationToken.CONTINUE,
     cryptoApi: ws.cryptoApi,
     db: ws.db,
-    http: ws.http,
-    taskScheduler: ws.taskScheduler,
-    oc: {
-      observe(event) {},
-    },
+    http: new ObservableHttpClientLibrary(ws.http, oc),
+    taskScheduler: new ObservableTaskScheduler(ws.taskScheduler, oc),
+    oc,
   };
 
   try {
     await ws.ensureWalletDbOpen();
-    const result = await dispatchRequestInternal(wex, operation as any, payload);
+    oc.observe({
+      type: ObservabilityEventType.RequestStart,
+    });
+    const result = await dispatchRequestInternal(
+      wex,
+      operation as any,
+      payload,
+    );
+    oc.observe({
+      type: ObservabilityEventType.RequestFinishSuccess,
+    });
     return {
       type: "response",
       operation,
@@ -1391,6 +1418,9 @@ async function handleCoreApiRequest(
     logger.info(
       `finished wallet core request ${operation} with error: ${j2s(err)}`,
     );
+    oc.observe({
+      type: ObservabilityEventType.RequestFinishError,
+    });
     return {
       type: "error",
       operation,
@@ -1460,6 +1490,7 @@ export class Wallet {
       insecureTrustExchange: false,
       denomselAllowLate: false,
       skipDefaults: false,
+      emitObservabilityEvents: false,
     },
   };
 
@@ -1522,7 +1553,7 @@ export class InternalWalletState {
    */
   private resourceLocks: Set<string> = new Set();
 
-  taskScheduler: TaskScheduler = new TaskScheduler(this);
+  taskScheduler: TaskScheduler = new TaskSchedulerImpl(this);
 
   config: Readonly<WalletConfig>;