rough structure for taler-merchant-exchange

2 files changed, 432 insertions, 14 deletions

diff --git a/src/backend/taler-merchant-exchange.c b/src/backend/taler-merchant-exchange.c
index e25ab98a..0b6ce56d 100644
--- a/src/backend/taler-merchant-exchange.c
+++ b/src/backend/taler-merchant-exchange.c
@@ -35,8 +35,59 @@
     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 a inquiry job.
+ * Information about an inquiry job.
  */
 struct Inquiry
 {
@@ -50,16 +101,90 @@ struct Inquiry
    */
   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 active inquiryes.
+ * 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 inquiryes.
+ * Tail of active inquiries.
  */
 static struct Inquiry *w_tail;
 
@@ -84,6 +209,11 @@ static struct GNUNET_CURL_Context *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.
  */
@@ -106,13 +236,107 @@ 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 w inquiry job to terminate
+ * @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);
@@ -123,7 +347,7 @@ end_inquiry (struct Inquiry *w)
 /**
  * We're being aborted with CTRL-C (or SIGTERM). Shut down.
  *
- * @param cls closure
+ * @param cls closure (NULL)
  */
 static void
 shutdown_task (void *cls)
@@ -135,9 +359,19 @@ shutdown_task (void *cls)
   {
     struct Inquiry *w = w_head;
 
-    save (w);
     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);
@@ -159,13 +393,166 @@ shutdown_task (void *cls)
 }
 
 
+// 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;
+}
+
+
 /**
- * Run next iteration.
+ * 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
-do_work (void *cls);
+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 */
+}
 
 
 /**
@@ -177,25 +564,49 @@ do_work (void *cls);
 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,
+  struct GNUNET_TIME_Absolute execution_time,
+  const struct TALER_Amount *total)
 {
-  struct Inquiry *w = GNUNET_new (struct Inquiry);
+  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->execution_time = execution_time;
+  w->total = *total;
   GNUNET_CONTAINER_DLL_insert (w_head,
                                w_tail,
                                w);
-  w->job = TALER_EXCHANGE_ (...);
+  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,
                                          &start_inquiry,
                                          NULL);
diff --git a/src/backenddb/merchant-0005.sql b/src/backenddb/merchant-0005.sql
index 1f3b9131..16409b6d 100644
--- a/src/backenddb/merchant-0005.sql
+++ b/src/backenddb/merchant-0005.sql
@@ -24,16 +24,23 @@ SET search_path TO merchant;
 
 ALTER TABLE merchant_instances
   ADD COLUMN user_type INT;
-
 COMMENT ON COLUMN merchant_instances.user_type
   IS 'what type of user is this (individual or business)';
 
 
+ALTER TABLE merchant_transfers
+  ADD COLUMN failed BOOLEAN NOT NULL DEFAULT FALSE,
+  ADD COLUMN validation_status VARCHAR DEFAULT NULL;
+COMMENT ON COLUMN merchant_transfers.failed
+  IS 'set to true on permanent verification failures';
+COMMENT ON COLUMN merchant_instances.validation_status
+  IS 'human-readable description of the state of the validation';
+
+
 ALTER TABLE merchant_accounts
   ADD COLUMN credit_facade_url VARCHAR,
   ADD COLUMN credit_facade_credentials VARCHAR,
   ADD COLUMN last_bank_serial INT8 NOT NULL DEFAULT (0);
-
 COMMENT ON COLUMN merchant_accounts.credit_facade_url
   IS 'Base URL of a facade where the merchant can inquire about incoming bank transactions into this account';
 COMMENT ON COLUMN merchant_accounts.credit_facade_credentials