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

/*
  This file is part of TALER
  Copyright (C) 2023 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-exchange.c
 * @brief Process that reconciles information about incoming bank transfers with orders by asking the exchange
 * @author Christian Grothoff
 */
#include "platform.h"
#include <gnunet/gnunet_util_lib.h>
#include <jansson.h>
#include <pthread.h>
#include <taler/taler_dbevents.h>
#include "taler_merchant_bank_lib.h"
#include "taler_merchantdb_lib.h"
#include "taler_merchantdb_plugin.h"

/**
 * Timeout for the exchange interaction.  Rather long as we should do
 * long-polling and do not want to wake up too often.
 */
#define EXCHANGE_TIMEOUT GNUNET_TIME_relative_multiply ( \
    GNUNET_TIME_UNIT_MINUTES, \
    30)


/**
 * Information about an exchange.
 */
struct Exchange
{
  /**
   * Kept in a DLL.
   */
  struct Exchange *next;

  /**
   * Kept in a DLL.
   */
  struct Exchange *prev;

  /**
   * Which exchange are we tracking here.
   */
  char *exchange_url;

  /**
   * A connection to this exchange
   */
  struct TALER_EXCHANGE_Handle *conn;

  /**
   * How soon can may we, at the earliest, re-download /keys?
   */
  struct GNUNET_TIME_Absolute first_retry;

  /**
   * How long should we wait between the next retry?
   */
  struct GNUNET_TIME_Relative retry_delay;

  /**
   * Task where we retry fetching /keys from the exchange.
   */
  struct GNUNET_SCHEDULER_Task *retry_task;

  /**
   * False to indicate that there is an ongoing
   * /keys transfer we are waiting for;
   * true to indicate that /keys data is up-to-date.
   */
  bool ready;

};


/**
 * Information about an inquiry job.
 */
struct Inquiry
{
  /**
   * Kept in a DLL.
   */
  struct Inquiry *next;

  /**
   * Kept in a DLL.
   */
  struct Inquiry *prev;

  /**
   * Handle to the exchange that made the transfer.
   */
  struct Exchange *exchange;

  /**
   * When did the transfer happen?
   */
  struct GNUNET_TIME_Absolute execution_time;

  /**
   * Argument for the /wire/transfers request.
   */
  struct TALER_WireTransferIdentifierRawP wtid;

  /**
   * Amount of the wire transfer.
   */
  struct TALER_Amount total;

  /**
   * For which merchant instance is this tracking request?
   */
  char *instance_id;

  /**
   * payto:// URI used for the transfer.
   */
  char *payto_uri;

  /**
   * Row of the wire transfer in our database.
   */
  uint64_t rowid;

  /**
   * Handle for the /wire/transfers request.
   */
  struct TALER_EXCHANGE_TransfersGetHandle *wdh;

  /**
   * Pointer to the detail that we are currently
   * checking in #check_transfer().
   */
  const struct TALER_TrackTransferDetails *current_detail;

  /**
   * Which transaction detail are we currently looking at?
   */
  unsigned int current_offset;

  /**
   * #GNUNET_NO if we did not find a matching coin.
   * #GNUNET_SYSERR if we found a matching coin, but the amounts do not match.
   * #GNUNET_OK if we did find a matching coin.
   */
  enum GNUNET_GenericReturnValue check_transfer_result;

  /**
   * Are we done with the exchange request for this
   * inquiry?
   */
  bool exchange_done;

};


/**
 * Head of known exchanges.
 */
static struct Exchange *e_head;

/**
 * Tail of known exchanges.
 */
static struct Exchange *e_tail;

/**
 * Head of active inquiries.
 */
static struct Inquiry *w_head;

/**
 * Tail of active inquiries.
 */
static struct Inquiry *w_tail;

/**
 * The merchant's configuration.
 */
static const struct GNUNET_CONFIGURATION_Handle *cfg;

/**
 * Our database plugin.
 */
static struct TALER_MERCHANTDB_Plugin *db_plugin;

/**
 * Handle to the context for interacting with the bank.
 */
static struct GNUNET_CURL_Context *ctx;

/**
 * Scheduler context for running the @e ctx.
 */
static struct GNUNET_CURL_RescheduleContext *rc;

/**
 * Main task for #find_work().
 */
static struct GNUNET_SCHEDULER_Task *task;

/**
 * Event handler to learn that there are new transfers
 * to check.
 */
static struct GNUNET_DB_EventHandler *eh;

/**
 * Value to return from main(). 0 on success, non-zero on errors.
 */
static int global_ret;

/**
 * How many transactions should we fetch at most per batch?
 */
static unsigned int batch_size = 32;

/**
 * #GNUNET_YES if we are in test mode and should exit when idle.
 */
static int test_mode;


/**
 * Initiate download from exchange.
 *
 * @param cls a `struct Inquiry *`
 */
static void
exchange_request (void *cls);


/**
 * Function called with information about who is auditing
 * a particular exchange and what keys the exchange is using.
 *
 * @param cls closure with a `struct Exchange *`
 * @param hr HTTP response data
 * @param keys information about the various keys used
 *        by the exchange, NULL if /keys failed
 * @param compat protocol compatibility information
 */
static void
cert_cb (
  void *cls,
  const struct TALER_EXCHANGE_HttpResponse *hr,
  const struct TALER_EXCHANGE_Keys *keys,
  enum TALER_EXCHANGE_VersionCompatibility compat)
{
  struct Exchange *e = cls;

  switch (hr->http_status)
  {
  case MHD_HTTP_OK:
    e->ready = true;
    for (struct Inquiry w = w_head;
         NULL != w;
         w = w->next)
    {
      if (w->exchange_done)
        continue;
      if (w->exchange != e)
        continue;
      if ( (NULL == w->task) &&
           (NULL == w->wdh) )
      {
        w->task = GNUNET_SCHEDULER_add_now (&exchange_request,
                                            w);
      }
    }
    // FIXME: schedule retry?
    break;
  default:
    // FIXME: schedule retry!
    break;
  }
}


/**
 * Lookup our internal data structure for the given
 * @a exchange_url or create one if we do not yet have
 * one.
 *
 * @param exchange_url base URL of the exchange
 * @return our state for this exchange
 */
static struct Exchange *
find_exchange (const char *exchange_url)
{
  struct Exchange *e;

  for (e = e_head; NULL != e; e = e->next)
    if (0 == strcmp (exchange_url,
                     e->exchange_url))
      return e;
  e = GNUNET_new (struct Exchange);
  e->exchange_url = GNUNET_strdup (exchange_url);
  GNUNET_CONTAINER_DLL_insert (e_head,
                               e_tail,
                               e);
  e->conn = TALER_EXCHANGE_connect (ctx,
                                    exchange_url,
                                    &cert_cb,
                                    e,
                                    TALER_EXCHANGE_OPTION_END);
  return e;
}


/**
 * Free resources of @a w.
 *
 * @param[in] w inquiry job to terminate
 */
static void
end_inquiry (struct Inquiry *w)
{
  if (NULL != w->wdh)
  {
    TALER_EXCHANGE_transfers_get_cancel (w->wdh);
    w->wdh = NULL;
  }
  GNUNET_free (w->instance_id);
  GNUNET_free (w->payto_uri);
  GNUNET_CONTAINER_DLL_remove (w_head,
                               w_tail,
                               w);
  GNUNET_free (w);
}


/**
 * We're being aborted with CTRL-C (or SIGTERM). Shut down.
 *
 * @param cls closure (NULL)
 */
static void
shutdown_task (void *cls)
{
  (void) cls;
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Running shutdown\n");
  while (NULL != w_head)
  {
    struct Inquiry *w = w_head;

    end_inquiry (w);
  }
  while (NULL != e_head)
  {
    struct Exchange *e = e_head;

    GNUNET_free (e->exchange_url);
    TALER_EXCHANGE_disconnect (e->conn);
    GNUNET_CONTAINER_DLL_remove (e_head,
                                 e_tail,
                                 e);
    GNUNET_free (e);
  }
  if (NULL != eh)
  {
    db_plugin->event_listen_cancel (eh);
    eh = NULL;
  }
  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;
  }
}


// FIXME: where is this used? Where do we check the totals!?
/**
 * Check that the given @a wire_fee is what the @a exchange_pub should charge
 * at the @a execution_time.  If the fee is correct (according to our
 * database), return #GNUNET_OK.  If we do not have the fee structure in our
 * DB, we just accept it and return #GNUNET_NO; if we have proof that the fee
 * is bogus, we respond with the proof to the client and return
 * #GNUNET_SYSERR.
 *
 * @param ptc context of the transfer to respond to
 * @param execution_time time of the wire transfer
 * @param wire_fee fee claimed by the exchange
 * @return #GNUNET_SYSERR if we returned hard proof of
 *   missbehavior from the exchange to the client
 */
static enum GNUNET_GenericReturnValue
check_wire_fee (struct PostTransfersContext *ptc,
                struct GNUNET_TIME_Timestamp execution_time,
                const struct TALER_Amount *wire_fee)
{
  struct TALER_WireFeeSet fees;
  struct TALER_MasterSignatureP master_sig;
  struct GNUNET_TIME_Timestamp start_date;
  struct GNUNET_TIME_Timestamp end_date;
  enum GNUNET_DB_QueryStatus qs;
  char *wire_method;

  wire_method = TALER_payto_get_method (ptc->payto_uri);
  qs = TMH_db->lookup_wire_fee (TMH_db->cls,
                                &ptc->master_pub,
                                wire_method,
                                execution_time,
                                &fees,
                                &start_date,
                                &end_date,
                                &master_sig);
  switch (qs)
  {
  case GNUNET_DB_STATUS_HARD_ERROR:
    GNUNET_break (0);
    ptc->response_code = MHD_HTTP_INTERNAL_SERVER_ERROR;
    ptc->response = TALER_MHD_make_error (TALER_EC_GENERIC_DB_FETCH_FAILED,
                                          "lookup_wire_fee");
    return GNUNET_SYSERR;
  case GNUNET_DB_STATUS_SOFT_ERROR:
    ptc->soft_retry = true;
    return GNUNET_NO;
  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Failed to find wire fee for `%s' and method `%s' at %s in DB, accepting blindly that the fee is %s\n",
                TALER_B2S (&ptc->master_pub),
                wire_method,
                GNUNET_TIME_timestamp2s (execution_time),
                TALER_amount2s (wire_fee));
    GNUNET_free (wire_method);
    return GNUNET_NO;
  case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
    break;
  }
  if (0 <= TALER_amount_cmp (&fees.wire,
                             wire_fee))
  {
    GNUNET_free (wire_method);
    return GNUNET_OK;   /* expected_fee >= wire_fee */
  }
  /* Wire fee check failed, export proof to auditor! */
  // FIXME: report error!
  GNUNET_free (wire_method);
  return GNUNET_SYSERR;
}


/**
 * Function called with detailed wire transfer data, including all
 * of the coin transactions that were combined into the wire transfer.
 *
 * @param cls closure a `struct Inquiry *`
 * @param hr HTTP response details
 * @param td transfer data
 */
static void
wire_transfer_cb (void *cls,
                  const struct TALER_EXCHANGE_HttpResponse *hr,
                  const struct TALER_EXCHANGE_TransferData *td)
{
  struct Inquiry *w = cls;
  enum GNUNET_DB_QueryStatus qs;

  w->wdh = NULL;
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Got response code %u from exchange for GET /transfers/$WTID\n",
              hr->http_status);
  switch (hr->http_status)
  {
  case MHD_HTTP_OK:
    break;
  case MHD_HTTP_NOT_FOUND:
    // FIXME: record permanent failure in DB!
    return;
  default:
    // FIXME: record transient failure in DB!
    return;
  }
  TMH_db->preflight (TMH_db->cls);
  /* Ok, exchange answer is acceptable, store it */
  qs = TMH_db->insert_transfer_details (TMH_db->cls,
                                        w->instance_id,
                                        w->exchange_url,
                                        w->payto_uri,
                                        &w->wtid,
                                        td);
  if (0 > qs)
  {
    /* Always report on DB error as well to enable diagnostics */
    GNUNET_break (GNUNET_DB_STATUS_HARD_ERROR == qs);
    // FIXME: shutdown?
    return;
  }
  if (0 == qs)
  {
    GNUNET_break (0);
    return;
  }
  // FIXME: check wire fee somewhere!??!
  if (0 !=
      TALER_amount_cmp (&td->total_amount,
                        &w->total))
  {
    // FIXME: record inconsistency in DB!
    return;
  }
  // FIXME: record success in DB!
}


/**
 * Initiate download from an exchange for a given inquiry.
 *
 * @param cls a `struct Inquiry *`
 */
static void
exchange_request (void *cls)
{
  struct Inquiry *w = cls;

  GNUNET_assert (w->exchange->ready);
  // FIXME: rate-limit number of parallel requests per
  // exchange!
  w->wdh = TALER_EXCHANGE_transfers_get (w->exchange->conn,
                                         &w->wtid,
                                         &wire_transfer_cb,
                                         w);
  if (NULL == w->wdh)
  {
    GNUNET_break (0);
    /* FIXME: update DB: status: failed on exchange! */
    return;
  }
  /* FIXME: update DB: status: waiting on exchange /transfers */
}


/**
 * Function called with information about a transfer we
 * should ask the exchange about.
 *
 * @param cls closure (NULL)
 * @param rowid row of the transfer in the merchant database
 * @param instance_id instance that received the transfer
 * @param exchange_url base URL of the exchange that initiated the transfer
 * @param payto_uri account of the merchant that received the transfer
 * @param wtid wire transfer subject identifying the aggregation
 * @param total total amount that was wired
 * @param next_attempt when should we next try to interact with the exchange
 */
static void
start_inquiry (
  void *cls,
  uint64_t rowid,
  const char *instance_id,
  const char *exchange_url,
  const char *payto_uri,
  const struct TALER_WireTransferIdentifierRawP *wtid,
  const struct TALER_Amount *total,
  struct GNUNET_TIME_Absolute next_attempt)
{
  struct Inquiry *w;

  (void) cls;
  w = GNUNET_new (struct Inquiry);
  w->payto_uri = GNUNET_strdup (payto_uri);
  w->instance_id = GNUNET_strdup (instance_id);
  w->rowid = rowid;
  w->wtid = *wtid;
  w->total = *total;
  GNUNET_CONTAINER_DLL_insert (w_head,
                               w_tail,
                               w);
  w->exchange = find_exchange (exchange_url);
  if (w->exchange->ready)
    w->task = GNUNET_SCHEDULER_add_now (&exchange_request,
                                        w);
  /* FIXME: update DB: status: waiting on exchange /keys */
}


/**
 * Finds new transfers that require work in the merchant
 * database.
 *
 * @param cls NULL
 */
static void
find_work (void *cls)
{
  enum GNUNET_DB_QueryStatus qs;

  task = NULL;
  // NOTE: SELECT WHERE confirmed AND NOT verified AND NOT failed?;
  // FIXME: use LIMIT clause! => When do we try again if LIMIT applied?
  qs = db_plugin->select_open_transfers (db_plugin->cls,
                                         LIMIT - active,
                                         &start_inquiry,
                                         NULL);
  if (qs < 0)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to obtain open transfers from database\n");
    GNUNET_SCHEDULER_shutdown ();
    return;
  }
}


/**
 * Function called when transfers are added to the merchant database.  We look
 * for more work.
 *
 * @param cls closure (NULL)
 * @param extra additional event data provided
 * @param extra_size number of bytes in @a extra
 */
static void
transfer_added (void *cls,
                const void *extra,
                size_t extra_size)
{
  (void) cls;
  (void) extra;
  (void) extra_size;
  if (NULL != task)
    return;
  task = GNUNET_SCHEDULER_add_now (&find_work,
                                   NULL);
}


/**
 * 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_XXX)
    };

    eh = db_plugin->event_listen (db_plugin->cls,
                                  &es,
                                  GNUNET_TIME_UNIT_FOREVER_REL,
                                  &transfer_added,
                                  NULL);
  }
  task = GNUNET_SCHEDULER_add_now (&find_work,
                                   NULL);
}


/**
 * The main function of taler-merchant-exchange
 *
 * @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-exchange",
    gettext_noop (
      "background process that reconciles bank transfers with orders by asking the exchange"),
    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-exchange.c */