skeleton logic for merchant service to detect KYC requirements automatically in the background

1 files changed, 689 insertions, 0 deletions

diff --git a/src/backend/taler-merchant-depositcheck.c b/src/backend/taler-merchant-depositcheck.c
new file mode 100644
index 00000000..6982d891
--- /dev/null
+++ b/src/backend/taler-merchant-depositcheck.c
@@ -0,0 +1,689 @@
+/*
+  This file is part of TALER
+  Copyright (C) 2024 Taler Systems SA
+
+  TALER is free software; you can redistribute it and/or modify it under the
+  terms of the GNU Affero General Public License as published by the Free Software
+  Foundation; either version 3, or (at your option) any later version.
+
+  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 Affero General Public License for more details.
+
+  You should have received a copy of the GNU Affero General Public License along with
+  TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+*/
+/**
+ * @file taler-merchant-depositcheck.c
+ * @brief Process that inquires with the exchange for deposits that should have been wired
+ * @author Christian Grothoff
+ */
+#include "platform.h"
+#include <gnunet/gnunet_util_lib.h>
+#include <jansson.h>
+#include <pthread.h>
+#include "taler_merchantdb_lib.h"
+#include "taler_merchantdb_plugin.h"
+#include <taler/taler_dbevents.h>
+
+
+/**
+ * Information we keep per exchange.
+ */
+struct ExchangeState
+{
+
+  /**
+   * Kept in a DLL.
+   */
+  struct ExchangeState *next;
+
+  /**
+   * Kept in a DLL.
+   */
+  struct ExchangeState *prev;
+
+  /**
+   * Which exchange is this state for?
+   */
+  const char *base_url;
+
+  /**
+   * Key material of the exchange.
+   */
+  struct TALER_EXCHANGE_Keys *keys;
+
+  /**
+   * Handle for active /keys request.
+   */
+  struct TALER_EXCHANGE_GetKeysHandle *gkh;
+};
+
+
+/**
+ * Information we keep per exchange interaction.
+ */
+struct ExchangeInteraction
+{
+  /**
+   * Kept in a DLL.
+   */
+  struct ExchangeInteraction *next;
+
+  /**
+   * Kept in a DLL.
+   */
+  struct ExchangeInteraction *prev;
+
+  /**
+   * Exchange we are interacting with.
+   */
+  struct ExchangeState *es;
+
+  /**
+   * Wire deadline for the deposit.
+   */
+  struct GNUNET_TIME_Absolute wire_deadline;
+
+  /**
+   * Target account hash of the deposit.
+   */
+  struct TALER_MerchantWireHashP h_wire;
+
+  /**
+   * Deposited amount.
+   */
+  struct TALER_Amount amount_with_fee;
+
+  /**
+   * Deposit fee paid.
+   */
+  struct TALER_Amount deposit_fee;
+
+  /**
+   * Public key of the deposited coin.
+   */
+  struct TALER_CoinSpendPublicKeyP coin_pub;
+
+  /**
+   * Hash over the @e contract_terms.
+   */
+  struct TALER_PrivateContractHashP h_contract_terms;
+
+  /**
+   * Merchant instance's private key.
+   */
+  struct TALER_MerchantPrivateKeyP merchant_priv;
+
+};
+
+
+/**
+ * Head of list of exchanges we interact with.
+ */
+static struct ExchangeState *e_head;
+
+/**
+ * Tail of list of exchanges we interact with.
+ */
+static struct ExchangeState *e_tail;
+
+/**
+ * Head of list of active exchange interactions.
+ */
+static struct ExchangeInteraction *w_head;
+
+/**
+ * Tail of list of active exchange interactions.
+ */
+static struct ExchangeInteraction *w_tail;
+
+/**
+ * Notification handler from database on new work.
+ */
+static struct GNUNET_DB_EventHandler *eh;
+
+/**
+ * The merchant's configuration.
+ */
+static const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+/**
+ * Our database plugin.
+ */
+static struct TALER_MERCHANTDB_Plugin *db_plugin;
+
+/**
+ * Next wire deadline that @e task is scheduled for.
+ */
+static struct GNUNET_TIME_Absolute next_deadline;
+
+/**
+ * Next task to run, if any.
+ */
+static struct GNUNET_SCHEDULER_Task *task;
+
+/**
+ * Handle to the context for interacting with the exchange.
+ */
+static struct GNUNET_CURL_Context *ctx;
+
+/**
+ * Scheduler context for running the @e ctx.
+ */
+static struct GNUNET_CURL_RescheduleContext *rc;
+
+/**
+ * Value to return from main(). 0 on success, non-zero on errors.
+ */
+static int global_ret;
+
+/**
+ * #GNUNET_YES if we are in test mode and should exit when idle.
+ */
+static int test_mode;
+
+
+/**
+ * We're being aborted with CTRL-C (or SIGTERM). Shut down.
+ *
+ * @param cls closure
+ */
+static void
+shutdown_task (void *cls)
+{
+  struct ExchangeState *es;
+  struct ExchangeInteraction *w;
+
+  (void) cls;
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Running shutdown\n");
+  if (NULL != eh)
+  {
+    db_plugin->event_listen_cancel (eh);
+    eh = NULL;
+  }
+  if (NULL != task)
+  {
+    GNUNET_SCHEDULER_cancel (task);
+    task = NULL;
+  }
+  while (NULL != (w = w_head))
+  {
+    GNUNET_CONTAINER_DLL_remove (w_head,
+                                 w_tail,
+                                 w);
+    GNUNET_free (w);
+  }
+  while (NULL != (es = e_head))
+  {
+    GNUNET_CONTAINER_DLL_remove (e_head,
+                                 e_tail,
+                                 es);
+    GNUNET_free (es->base_url);
+    GNUNET_free (es);
+  }
+  db_plugin->rollback (db_plugin->cls); /* just in case */
+  TALER_MERCHANTDB_plugin_unload (db_plugin);
+  db_plugin = NULL;
+  cfg = NULL;
+  if (NULL != ctx)
+  {
+    GNUNET_CURL_fini (ctx);
+    ctx = NULL;
+  }
+  if (NULL != rc)
+  {
+    GNUNET_CURL_gnunet_rc_destroy (rc);
+    rc = NULL;
+  }
+}
+
+
+/**
+ * Task to get more deposits to work on from the database.
+ *
+ * @param cls NULL
+ */
+static void
+select_work (void *cls);
+
+
+/**
+ * Make sure to run the select_work() task at
+ * the @a next_deadline.
+ *
+ * @param next_deadline deadline when work becomes ready
+ */
+static void
+run_at (struct GNUNET_TIME_Absolute next_deadline)
+{
+  if (GNUNET_TIME_absolute_cmp (deadline,
+                                <,
+                                next_deadline))
+  {
+    if (NULL != task)
+      GNUNET_SCHEDULER_cancel (task);
+    next_deadline = deadline;
+    task = GNUNET_SCHEDULER_add_at (deadline,
+                                    &select_work,
+                                    NULL);
+  }
+}
+
+
+/**
+ * Function called with detailed wire transfer data.
+ *
+ * @param cls closure with a `struct TransferQuery *`
+ * @param dr HTTP response data
+ */
+static void
+deposit_get_cb (void *cls,
+                const struct TALER_EXCHANGE_GetDepositResponse *dr)
+{
+  struct ExchangeState *es = cls;
+
+  switch (dr->hr.http_status)
+  {
+  case MHD_HTTP_OK:
+    {
+      enum GNUNET_DB_QueryStatus qs;
+
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  "Exchange returned wire transfer over %s for deposited coin %s\n",
+                  TALER_amount2s (&dr->details.ok.coin_contribution),
+                  TALER_B2S (&tq->coin_pub));
+      qs = TMH_db->insert_deposit_to_transfer (TMH_db->cls,
+                                               tq->deposit_serial,
+                                               &dr->details.ok);
+      if (qs < 0)
+      {
+        GNUNET_break (0);
+        GNUNET_SCHEDULER_shutdown ();
+        return;
+      }
+      break;
+    }
+  case MHD_HTTP_ACCEPTED:
+    {
+      /* got a 'preliminary' reply from the exchange,
+         remember our target UUID */
+      enum GNUNET_DB_QueryStatus qs;
+      struct GNUNET_TIME_Timestamp now;
+
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  "Exchange returned KYC requirement (%d/%d) for deposited coin %s\n",
+                  dr->details.accepted.kyc_ok,
+                  dr->details.accepted.aml_decision,
+                  TALER_B2S (&tq->coin_pub));
+      now = GNUNET_TIME_timestamp_get ();
+      qs = TMH_db->account_kyc_set_status (
+        TMH_db->cls,
+        gorc->hc->instance->settings.id,
+        &tq->h_wire,
+        tq->exchange_url,
+        dr->details.accepted.requirement_row,
+        NULL,
+        NULL,
+        now,
+        dr->details.accepted.kyc_ok,
+        dr->details.accepted.aml_decision);
+      if (qs < 0)
+      {
+        GNUNET_break (0);
+        GNUNET_SCHEDULER_shutdown ();
+        return;
+      }
+      break;
+    }
+  default:
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  "Exchange returned tracking failure for deposited coin %s\n",
+                  TALER_B2S (&tq->coin_pub));
+      /* FIXME: how to handle? */
+      return;
+    }
+  } /* end switch */
+
+  GNUNET_CONTAINER_DLL_remove (w_head,
+                               w_tail,
+                               w);
+  GNUNET_free (w);
+  // FIXME: not the best condition to go for more work,
+  // one down exchange would halt us entirely!
+  if (NULL == w_head)
+    task = GNUNET_SCHEDULER_add_now (&select_work,
+                                     NULL);
+}
+
+
+/**
+ * Initiate request with the exchange about deposits.
+ *
+ * @param[in,out] exchange interaction handle
+ */
+static void
+inquire_at_exchange (struct ExchangeInteraction *w)
+{
+  struct ExchangeState *es = w->es;
+
+  GNUNET_assert (NULL == w->dgh);
+  w->dgh = TALER_EXCHANGE_deposits_get (
+    ctx,
+    es->exchange_url,
+    es->keys,
+    &w->merchant_priv,
+    &w->h_wire,
+    &w->h_contract_terms,
+    &w->coin_pub,
+    GNUNET_TIME_UNIT_ZERO,
+    &deposit_get_cb,
+    w);
+}
+
+
+/**
+ * Function called with information about who is auditing
+ * a particular exchange and what keys the exchange is using.
+ * The ownership over the @a keys object is passed to
+ * the callee, thus it is given explicitly and not
+ * (only) via @a kr.
+ *
+ * @param cls closure
+ * @param kr response from /keys
+ * @param[in] keys keys object passed to callback with
+ *  reference counter of 1. Must be freed by callee
+ *  using #TALER_EXCHANGE_keys_decref(). NULL on failure.
+ */
+static void
+keys_cb (
+  void *cls,
+  const struct TALER_EXCHANGE_KeysResponse *kr,
+  struct TALER_EXCHANGE_Keys *keys)
+{
+  struct ExchangeState *es = cls;
+
+  es->gkh = NULL;
+  if (NULL == keys)
+    return;
+  if (NULL != es->keys)
+    TALER_EXCHANGE_keys_decref (keys);
+  es->keys = TALER_EXCHANGE_keys_incref (keys);
+  /* Trigger all deposits blocked on fetching /keys */
+  for (struct ExchangeInteraction *w = w_head;
+       NULL != w;
+       w = w->next)
+  {
+    if (w->es != es)
+      continue;
+    if (NULL != w->dgh)
+      continue;
+    inquire_at_exchange (w);
+  }
+}
+
+
+/**
+ * Download /keys from an exchange.
+ *
+ * @param[in,out] es exchange state
+ */
+static void
+fetch_keys (struct ExchangeState *es)
+{
+  GNUNET_assert (NULL == es->gkh);
+  es->gkh = TALER_EXCHANGE_get_keys (ctx,
+                                     es->base_url,
+                                     es->keys,
+                                     &keys_cb,
+                                     es);
+}
+
+
+/**
+ * Typically called by `select_work`.
+ *
+ * @param cls NULL
+ * @param deposit_serial identifies the deposit operation
+ * @param exchange_url URL of the exchange that issued @a coin_pub
+ * @param amount_with_fee amount the exchange will deposit for this coin
+ * @param deposit_fee fee the exchange will charge for this coin
+ * @param h_wire hash of the merchant's wire account into which the deposit was made
+ * @param coin_pub public key of the deposited coin
+ */
+static void
+pending_deposits_cb (
+  void *cls,
+  uint64_t deposit_serial,
+  struct GNUNET_TIME_Absolute wire_deadline, /* missing in DB! Funky migration needed! */
+  const char *exchange_url,
+  const struct TALER_PrivateContractHashP *h_contract_terms,
+  const struct TALER_MerchantPrivateKeyP *merchant_priv,
+  const struct TALER_MerchantWireHashP *h_wire,
+  const struct TALER_Amount *amount_with_fee,
+  const struct TALER_Amount *deposit_fee,
+  const struct TALER_CoinSpendPublicKeyP *coin_pub)
+{
+  struct ExchangeInteraction *w = GNUNET_new (struct ExchangeInteraction);
+  struct ExchangeState *es = NULL;
+
+  (void) cls;
+  if (GNUNET_TIME_absolute_is_future (wire_deadline))
+  {
+    run_at (wire_deadline);
+    return;
+  }
+  w->wire_deadline = wire_deadline;
+  w->h_contract_terms = *h_contract_terms;
+  w->merchant_priv = *merchant_priv;
+  w->h_wire = *h_wire;
+  w->amount_with_fee = *amount_with_fee;
+  w->deposit_fee = *deposit_fee;
+  w->coin_pub = *coin_pub;
+  GNUNET_CONTAINER_DLL_insert (w_head,
+                               w_tail,
+                               w);
+  for (es = e_head; NULL != es; es = es->next)
+    if (0 == strcmp (exchange_url,
+                     es->base_url))
+      break;
+  if (NULL == es)
+  {
+    es = GNUNET_new (struct ExchangeState);
+    es->base_url = GNUNET_strdup (exchange_url);
+    GNUNET_CONTAINER_DLL_insert (e_head,
+                                 e_tail,
+                                 es);
+  }
+  w->es = es;
+  if ( (NULL == es->keys) ||
+       (GNUNET_TIME_absolute_is_past (es->keys->key_data_expiration)) )
+  {
+    fetch_keys (es);
+    return;
+  }
+  inquire_at_exchange (w);
+}
+
+
+/**
+ * Function called on events received from Postgres.
+ *
+ * @param cls closure, NULL
+ * @param extra additional event data provided, timestamp with wire deadline
+ * @param extra_size number of bytes in @a extra
+ */
+static void
+db_notify (void *cls,
+           const void *extra,
+           size_t extra_size)
+{
+  struct GNUNET_TIME_Absolute deadline;
+  struct GNUNET_TIME_AbsoluteNBO nbo_deadline;
+
+  (void) cls;
+  if (sizeof (nbo_deadline) != extra_size)
+  {
+    GNUNET_break (0);
+    return;
+  }
+  memcpy (&nbo_deadline,
+          extra,
+          extra_size);
+  deadline = GNUNET_TIME_absolute_ntoh (nbo_deadline);
+  run_at (deadline);
+}
+
+
+static void
+select_work (void *cls)
+{
+  bool retry = false;
+
+  (void) cls;
+  task = NULL;
+  while (1)
+  {
+    struct GNUNET_TIME_Absolute now;
+    enum GNUNET_DB_QueryStatus qs;
+
+    now = GNUNET_TIME_absolute_get ();
+    db_plugin->preflight (db_plugin->cls);
+    qs = db_plugin->lookup_pending_deposits (db_plugin->cls,
+                                             now,
+                                             retry,
+                                             &pending_deposits_cb,
+                                             NULL);
+    switch (qs)
+    {
+    case GNUNET_DB_STATUS_HARD_ERROR:
+    case GNUNET_DB_STATUS_SOFT_ERROR:
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Transaction failed!\n");
+      global_ret = EXIT_FAILURE;
+      GNUNET_SCHEDULER_shutdown ();
+      return;
+    case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+      if (test_mode)
+      {
+        GNUNET_SCHEDULER_shutdown ();
+        return;
+      }
+      retry = true;
+      continue;
+    case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
+    default:
+      return; /* wait for completion, then select more work. */
+    }
+  }
+}
+
+
+/**
+ * First task.
+ *
+ * @param cls closure, NULL
+ * @param args remaining command-line arguments
+ * @param cfgfile name of the configuration file used (for saving, can be NULL!)
+ * @param c configuration
+ */
+static void
+run (void *cls,
+     char *const *args,
+     const char *cfgfile,
+     const struct GNUNET_CONFIGURATION_Handle *c)
+{
+  (void) args;
+  (void) cfgfile;
+
+  cfg = c;
+  GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
+                                 NULL);
+  ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule,
+                          &rc);
+  rc = GNUNET_CURL_gnunet_rc_create (ctx);
+  if (NULL == ctx)
+  {
+    GNUNET_break (0);
+    GNUNET_SCHEDULER_shutdown ();
+    return;
+  }
+  if (NULL ==
+      (db_plugin = TALER_MERCHANTDB_plugin_load (cfg)))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "Failed to initialize DB subsystem\n");
+    GNUNET_SCHEDULER_shutdown ();
+    return;
+  }
+  if (GNUNET_OK !=
+      db_plugin->connect (db_plugin->cls))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "Failed to connect to database\n");
+    GNUNET_SCHEDULER_shutdown ();
+    return;
+  }
+  {
+    struct GNUNET_DB_EventHeaderP es = {
+      .size = htons (sizeof (es)),
+      .type = htons (TALER_DBEVENT_MERCHANT_NEW_WIRE_DEADLINE)
+    };
+
+    eh = db_plugin->event_listen (db_plugin->cls,
+                                  &es,
+                                  GNUNET_TIME_UNIT_FOREVER_REL,
+                                  &db_notify,
+                                  NULL);
+  }
+  GNUNET_assert (NULL == task);
+  task = GNUNET_SCHEDULER_add_now (&select_work,
+                                   NULL);
+}
+
+
+/**
+ * The main function of the taler-merchant-depositcheck
+ *
+ * @param argc number of arguments from the command line
+ * @param argv command line arguments
+ * @return 0 ok, 1 on error
+ */
+int
+main (int argc,
+      char *const *argv)
+{
+  struct GNUNET_GETOPT_CommandLineOption options[] = {
+    GNUNET_GETOPT_option_flag ('t',
+                               "test",
+                               "run in test mode and exit when idle",
+                               &test_mode),
+    GNUNET_GETOPT_option_version (VERSION "-" VCS_VERSION),
+    GNUNET_GETOPT_OPTION_END
+  };
+  enum GNUNET_GenericReturnValue ret;
+
+  if (GNUNET_OK !=
+      GNUNET_STRINGS_get_utf8_args (argc, argv,
+                                    &argc, &argv))
+    return EXIT_INVALIDARGUMENT;
+  TALER_OS_init ();
+  ret = GNUNET_PROGRAM_run (
+    argc, argv,
+    "taler-merchant-depositcheck",
+    gettext_noop (
+      "background process that checks with the exchange on deposits that are past the wire deadline"),
+    options,
+    &run, NULL);
+  GNUNET_free_nz ((void *) argv);
+  if (GNUNET_SYSERR == ret)
+    return EXIT_INVALIDARGUMENT;
+  if (GNUNET_NO == ret)
+    return EXIT_SUCCESS;
+  return global_ret;
+}
+
+
+/* end of taler-merchant-depositcheck.c */