path: root/src/backend/taler-merchant-depositcheck.c

/*
  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 */