change rowid type to uint64_t everywhere; start with reserve-analysis in auditor

src/auditor/taler-auditor.c

@@ -34,11 +34,267 @@ static int global_ret;
  */
 static struct TALER_EXCHANGEDB_Plugin *edb;
 
+/**
+ * Our session with the #edb.
+ */
+static struct TALER_EXCHANGEDB_Session *esession;
+
 /**
  * Handle to access the auditor's database.
  */
 static struct TALER_AUDITORDB_Plugin *adb;
 
+/**
+ * Last reserve_in serial ID seen.
+ */
+static uint64_t reserve_in_serial_id;
+
+/**
+ * Last reserve_out serial ID seen.
+ */
+static uint64_t reserve_out_serial_id;
+
+
+/**
+ * Summary data we keep per reserve.
+ */
+struct ReserveSummary
+{
+  /**
+   * Public key of the reserve.
+   */
+  struct TALER_ReservePublicKeyP reserve_pub;
+
+  /**
+   * Sum of all incoming transfers.
+   */
+  struct TALER_Amount total_in;
+
+  /**
+   * Sum of all outgoing transfers.
+   */
+  struct TALER_Amount total_out;
+
+};
+
+
+/**
+ * Map from hash of reserve's public key to a `struct ReserveSummary`.
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *reserves;
+
+
+/**
+ * Function called with details about incoming wire transfers.
+ *
+ * @param cls NULL
+ * @param rowid unique serial ID for the refresh session in our DB
+ * @param reserve_pub public key of the reserve (also the WTID)
+ * @param credit amount that was received
+ * @param sender_account_details information about the sender's bank account
+ * @param transfer_details information that uniquely identifies the wire transfer
+ * @param execution_date when did we receive the funds
+ * @return #GNUNET_OK to continue to iterate, #GNUNET_SYSERR to stop
+ */
+static int
+handle_reserve_in (void *cls,
+                   uint64_t rowid,
+                   const struct TALER_ReservePublicKeyP *reserve_pub,
+                   const struct TALER_Amount *credit,
+                   const json_t *sender_account_details,
+                   const json_t *transfer_details,
+                   struct GNUNET_TIME_Absolute execution_date)
+{
+  struct GNUNET_HashCode key;
+  struct ReserveSummary *rs;
+
+  /* TODO: somewhere we need to check that 'reserve_in' data actually
+     matches wire transfers from the bank. Not sure if this should be
+     done within the core auditor logic though... */
+
+  GNUNET_assert (rowid == reserve_in_serial_id + 1); /* should be monotonically increasing */
+  reserve_in_serial_id = GNUNET_MAX (rowid,
+                                     reserve_in_serial_id);
+  GNUNET_CRYPTO_hash (reserve_pub,
+                      sizeof (*reserve_pub),
+                      &key);
+  rs = GNUNET_CONTAINER_multihashmap_get (reserves,
+                                          &key);
+  if (NULL == rs)
+  {
+    rs = GNUNET_new (struct ReserveSummary);
+    rs->reserve_pub = *reserve_pub;
+    rs->total_in = *credit;
+    GNUNET_assert (GNUNET_OK ==
+                   TALER_amount_get_zero (credit->currency,
+                                          &rs->total_out));
+    GNUNET_assert (GNUNET_OK ==
+                   GNUNET_CONTAINER_multihashmap_put (reserves,
+                                                      &key,
+                                                      rs,
+                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+  }
+  else
+  {
+    GNUNET_assert (GNUNET_OK ==
+                   TALER_amount_add (&rs->total_in,
+                                     &rs->total_in,
+                                     credit));
+  }
+  return GNUNET_OK;
+}
+
+
+/**
+ * Function called with details about withdraw operations.
+ *
+ * @param cls closure
+ * @param rowid unique serial ID for the refresh session in our DB
+ * @param h_blind_ev blinded hash of the coin's public key
+ * @param denom_pub public denomination key of the deposited coin
+ * @param denom_sig signature over the deposited coin
+ * @param reserve_pub public key of the reserve
+ * @param reserve_sig signature over the withdraw operation
+ * @param execution_date when did the wallet withdraw the coin
+ * @param amount_with_fee amount that was withdrawn
+ * @return #GNUNET_OK to continue to iterate, #GNUNET_SYSERR to stop
+ */
+static int
+handle_reserve_out (void *cls,
+                    uint64_t rowid,
+                    const struct GNUNET_HashCode *h_blind_ev,
+                    const struct TALER_DenominationPublicKey *denom_pub,
+                    const struct TALER_DenominationSignature *denom_sig,
+                    const struct TALER_ReservePublicKeyP *reserve_pub,
+                    const struct TALER_ReserveSignatureP *reserve_sig,
+                    struct GNUNET_TIME_Absolute execution_date,
+                    const struct TALER_Amount *amount_with_fee)
+{
+  struct GNUNET_HashCode key;
+  struct ReserveSummary *rs;
+
+  /* TODO: check signatures, in particluar the reserve_sig! */
+  GNUNET_assert (rowid == reserve_out_serial_id + 1); /* should be monotonically increasing */
+  reserve_in_serial_id = GNUNET_MAX (rowid,
+                                     reserve_out_serial_id);
+  GNUNET_CRYPTO_hash (reserve_pub,
+                      sizeof (*reserve_pub),
+                      &key);
+  rs = GNUNET_CONTAINER_multihashmap_get (reserves,
+                                          &key);
+  if (NULL == rs)
+  {
+    rs = GNUNET_new (struct ReserveSummary);
+    rs->reserve_pub = *reserve_pub;
+    rs->total_out = *amount_with_fee;
+    GNUNET_assert (GNUNET_OK ==
+                   TALER_amount_get_zero (amount_with_fee->currency,
+                                          &rs->total_in));
+    GNUNET_assert (GNUNET_OK ==
+                   GNUNET_CONTAINER_multihashmap_put (reserves,
+                                                      &key,
+                                                      rs,
+                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+  }
+  else
+  {
+    GNUNET_assert (GNUNET_OK ==
+                   TALER_amount_add (&rs->total_out,
+                                     &rs->total_out,
+                                     amount_with_fee));
+  }
+  return GNUNET_OK;
+}
+
+
+/**
+ * Check that the reserve summary matches what the exchange database
+ * thinks about the reserve, and update our own state of the reserve.
+ *
+ * Remove all reserves that we are happy with from the DB.
+ *
+ * @param cls NULL
+ * @param key hash of the reserve public key
+ * @param value a `struct ReserveSummary`
+ * @return #GNUNET_OK to process more entries
+ */
+static int
+verify_reserve_balance (void *cls,
+                        const struct GNUNET_HashCode *key,
+                        void *value)
+{
+  struct ReserveSummary *rs = value;
+  struct TALER_EXCHANGEDB_Reserve reserve;
+  struct TALER_Amount balance;
+
+  if (GNUNET_OK !=
+      edb->reserve_get (edb->cls,
+                        esession,
+                        &reserve))
+  {
+    GNUNET_break (0);
+    return GNUNET_OK;
+  }
+  /* TODO: check reserve.expiry? */
+  /* FIXME: get previous reserve state from auditor DB */
+
+  /* FIXME: simplified computation as we have no previous reserve state yet */
+  if (GNUNET_SYSERR ==
+      TALER_amount_subtract (&balance,
+                             &rs->total_in,
+                             &rs->total_out))
+  {
+    GNUNET_break (0);
+    return GNUNET_OK;
+  }
+  if (0 != TALER_amount_cmp (&balance,
+                             &reserve.balance))
+  {
+    GNUNET_break (0);
+    return GNUNET_OK;
+  }
+
+  /* FIXME: commit new reserve state from auditor DB */
+
+  GNUNET_assert (GNUNET_YES ==
+                 GNUNET_CONTAINER_multihashmap_remove (reserves,
+                                                       key,
+                                                       rs));
+  GNUNET_free (rs);
+  return GNUNET_OK;
+}
+
+
+/**
+ * Analyze reserves for being well-formed.
+ *
+ * @return #GNUNET_OK on success, #GNUNET_SYSERR on invariant violation
+ */
+static int
+analyze_reserves ()
+{
+  reserves = GNUNET_CONTAINER_multihashmap_create (512,
+                                                   GNUNET_NO);
+  /* FIXME: check return values... */
+  edb->select_reserves_in_above_serial_id (edb->cls,
+                                           esession,
+                                           reserve_in_serial_id,
+                                           &handle_reserve_in,
+                                           NULL);
+  edb->select_reserves_out_above_serial_id (edb->cls,
+                                            esession,
+                                            reserve_out_serial_id,
+                                            &handle_reserve_out,
+                                            NULL);
+  GNUNET_CONTAINER_multihashmap_iterate (reserves,
+                                         &verify_reserve_balance,
+                                         NULL);
+  /* FIXME: any values left in #reserves indicate errors! */
+  GNUNET_CONTAINER_multihashmap_destroy (reserves);
+
+  return GNUNET_OK;
+}
+
 
 /**
  * Main function that will be run.
@@ -71,7 +327,22 @@ run (void *cls,
     TALER_EXCHANGEDB_plugin_unload (edb);
     return;
   }
+  esession = edb->get_session (edb->cls);
+  if (NULL == esession)
+  {
+    fprintf (stderr,
+             "Failed to initialize exchange session.\n");
+    global_ret = 1;
+    TALER_AUDITORDB_plugin_unload (adb);
+    TALER_EXCHANGEDB_plugin_unload (edb);
+    return;
+  }
 
+  /* FIXME: init these from auditordb */
+  reserve_in_serial_id = 0;
+  reserve_out_serial_id = 0;
+
+  analyze_reserves ();
 
   TALER_AUDITORDB_plugin_unload (adb);
   TALER_EXCHANGEDB_plugin_unload (edb);

src/exchange/taler-exchange-aggregator.c

@@ -410,7 +410,7 @@ exchange_serve_process_config ()
  */
 static int
 deposit_cb (void *cls,
-            unsigned long long row_id,
+            uint64_t row_id,
             const struct TALER_MerchantPublicKeyP *merchant_pub,
             const struct TALER_CoinSpendPublicKeyP *coin_pub,
             const struct TALER_Amount *amount_with_fee,
@@ -428,7 +428,7 @@ deposit_cb (void *cls,
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                 "Fatally malformed record at row %llu\n",
-                row_id);
+                (unsigned long long) row_id);
     return GNUNET_SYSERR;
   }
   au->row_id = row_id;
@@ -485,7 +485,7 @@ deposit_cb (void *cls,
  */
 static int
 aggregate_cb (void *cls,
-              unsigned long long row_id,
+              uint64_t row_id,
               const struct TALER_MerchantPublicKeyP *merchant_pub,
               const struct TALER_CoinSpendPublicKeyP *coin_pub,
               const struct TALER_Amount *amount_with_fee,
@@ -509,7 +509,7 @@ aggregate_cb (void *cls,
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                 "Fatally malformed record at %llu\n",
-                row_id);
+                (unsigned long long) row_id);
     return GNUNET_SYSERR;
   }
   /* add to total */
@@ -520,7 +520,7 @@ aggregate_cb (void *cls,
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                 "Overflow or currency incompatibility during aggregation at %llu\n",
-                row_id);
+                (unsigned long long) row_id);
     /* Skip this one, but keep going! */
     return GNUNET_OK;
   }
@@ -950,7 +950,7 @@ wire_confirm_cb (void *cls,
  */
 static void
 wire_prepare_cb (void *cls,
-                 unsigned long long rowid,
+                 uint64_t rowid,
                  const char *wire_method,
                  const char *buf,
                  size_t buf_size)
@@ -958,7 +958,7 @@ wire_prepare_cb (void *cls,
   wpd->row_id = rowid;
   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
               "Starting wire transfer %llu\n",
-              rowid);
+              (unsigned long long) rowid);
   wpd->wp = find_plugin (wire_method);
   wpd->eh = wpd->wp->wire_plugin->execute_wire_transfer (wpd->wp->wire_plugin->cls,
                                                          buf,

src/exchangedb/plugin_exchangedb_postgres.c

@@ -2289,11 +2289,10 @@ postgres_have_deposit (void *cls,
 static int
 postgres_mark_deposit_tiny (void *cls,
                             struct TALER_EXCHANGEDB_Session *session,
-                            unsigned long long rowid)
+                            uint64_t rowid)
 {
-  uint64_t serial_id = rowid;
   struct GNUNET_PQ_QueryParam params[] = {
-    GNUNET_PQ_query_param_uint64 (&serial_id),
+    GNUNET_PQ_query_param_uint64 (&rowid),
     GNUNET_PQ_query_param_end
   };
   PGresult *result;
@@ -2393,11 +2392,10 @@ postgres_test_deposit_done (void *cls,
 static int
 postgres_mark_deposit_done (void *cls,
                             struct TALER_EXCHANGEDB_Session *session,
-                            unsigned long long rowid)
+                            uint64_t rowid)
 {
-  uint64_t serial_id = rowid;
   struct GNUNET_PQ_QueryParam params[] = {
-    GNUNET_PQ_query_param_uint64 (&serial_id),
+    GNUNET_PQ_query_param_uint64 (&rowid),
     GNUNET_PQ_query_param_end
   };
   PGresult *result;
@@ -4171,11 +4169,10 @@ postgres_wire_prepare_data_insert (void *cls,
 static int
 postgres_wire_prepare_data_mark_finished (void *cls,
                                           struct TALER_EXCHANGEDB_Session *session,
-                                          unsigned long long rowid)
+                                          uint64_t rowid)
 {
-  uint64_t prewire_uuid = rowid;
   struct GNUNET_PQ_QueryParam params[] = {
-    GNUNET_PQ_query_param_uint64 (&prewire_uuid),
+    GNUNET_PQ_query_param_uint64 (&rowid),
     GNUNET_PQ_query_param_end
   };
   PGresult *result;

src/exchangedb/test_exchangedb.c

@@ -70,7 +70,7 @@ static struct TALER_EXCHANGEDB_Plugin *plugin;
  */
 static void
 dead_prepare_cb (void *cls,
-                 unsigned long long rowid,
+                 uint64_t rowid,
                  const char *wire_method,
                  const char *buf,
                  size_t buf_size)
@@ -78,6 +78,7 @@ dead_prepare_cb (void *cls,
   GNUNET_assert (0);
 }
 
+
 /**
  * Counter used in auditor-related db functions. Used to count
  * expected rows.
@@ -91,7 +92,7 @@ unsigned int auditor_row_cnt;
  */
 static void
 mark_prepare_cb (void *cls,
-                 unsigned long long rowid,
+                 uint64_t rowid,
                  const char *wire_method,
                  const char *buf,
                  size_t buf_size)
@@ -122,7 +123,7 @@ mark_prepare_cb (void *cls,
  */
 void
 audit_wire_cb (void *cls,
-               unsigned long long rowid,
+               uint64_t rowid,
                const char *wire_method,
                const char *buf,
                size_t buf_size,
@@ -523,7 +524,7 @@ check_transfer_data (void *cls,
  */
 int
 audit_refresh_session_cb (void *cls,
-                          unsigned long long rowid, /* FIXME: decide data type for serial_id! */
+                          uint64_t rowid,
                           const struct TALER_CoinSpendPublicKeyP *coin_pub,
                           const struct TALER_CoinSpendSignatureP *coin_sig,
                           const struct TALER_Amount *amount_with_fee,
@@ -883,7 +884,7 @@ cb_wtid_check (void *cls,
 /**
  * Here #deposit_cb() will store the row ID of the deposit.
  */
-static unsigned long long deposit_rowid;
+static uint64_t deposit_rowid;
 
 
 /**
@@ -908,7 +909,7 @@ static unsigned long long deposit_rowid;
  */
 static int
 deposit_cb (void *cls,
-            unsigned long long rowid,
+            uint64_t rowid,
             const struct TALER_MerchantPublicKeyP *merchant_pub,
             const struct TALER_CoinSpendPublicKeyP *coin_pub,
             const struct TALER_Amount *amount_with_fee,
@@ -968,10 +969,10 @@ deposit_cb (void *cls,
  * @param receiver_wire_account wire details for the merchant, NULL from iterate_matching_deposits()
  * @param done flag set if the deposit was already executed (or not)
  * @return #GNUNET_OK to continue to iterate, #GNUNET_SYSERR to stop
- */ 
+ */
 int
 audit_deposit_cb (void *cls,
-                  unsigned long long rowid,
+                  uint64_t rowid,
                   const struct TALER_MerchantPublicKeyP *merchant_pub,
                   const struct TALER_CoinSpendPublicKeyP *coin_pub,
                   const struct TALER_CoinSpendSignatureP *coin_sig,
@@ -1005,7 +1006,7 @@ audit_deposit_cb (void *cls,
  */
 int
 audit_refund_cb (void *cls,
-                 unsigned long long rowid, /* FIXME: decide data type for serial_id! */
+                 uint64_t rowid,
                  const struct TALER_CoinSpendPublicKeyP *coin_pub,
                  const struct TALER_MerchantPublicKeyP *merchant_pub,
                  const struct TALER_MerchantSignatureP *merchant_sig,
@@ -1033,7 +1034,7 @@ audit_refund_cb (void *cls,
  */
 int
 audit_reserve_in_cb (void *cls,
-                     unsigned long long rowid, /* FIXME: decide data type for serial_id! */
+                     uint64_t rowid,
                      const struct TALER_ReservePublicKeyP *reserve_pub,
                      const struct TALER_Amount *credit,
                      const json_t *sender_account_details,
@@ -1060,7 +1061,7 @@ audit_reserve_in_cb (void *cls,
  */
 int
 audit_reserve_out_cb (void *cls,
-                      unsigned long long rowid, /* FIXME: decide data type for serial_id! */
+                      uint64_t rowid,
                       const struct GNUNET_HashCode *h_blind_ev,
                       const struct TALER_DenominationPublicKey *denom_pub,
                       const struct TALER_DenominationSignature *denom_sig,

src/include/taler_exchangedb_plugin.h

@@ -571,7 +571,7 @@ struct TALER_EXCHANGEDB_Session;
  */
 typedef int
 (*TALER_EXCHANGEDB_DepositIterator)(void *cls,
-                                    unsigned long long rowid,
+                                    uint64_t rowid,
                                     const struct TALER_MerchantPublicKeyP *merchant_pub,
                                     const struct TALER_CoinSpendPublicKeyP *coin_pub,
                                     const struct TALER_Amount *amount_with_fee,
@@ -593,7 +593,7 @@ typedef int
  */
 typedef void
 (*TALER_EXCHANGEDB_WirePreparationIterator) (void *cls,
-                                             unsigned long long rowid,
+                                             uint64_t rowid,
                                              const char *wire_method,
                                              const char *buf,
                                              size_t buf_size);
@@ -621,7 +621,7 @@ typedef void
  */
 typedef int
 (*TALER_EXCHANGEDB_DepositCallback)(void *cls,
-                                    unsigned long long rowid,
+                                    uint64_t rowid,
                                     const struct TALER_MerchantPublicKeyP *merchant_pub,
                                     const struct TALER_CoinSpendPublicKeyP *coin_pub,
                                     const struct TALER_CoinSpendSignatureP *coin_sig,
@@ -656,7 +656,7 @@ typedef int
  */
 typedef int
 (*TALER_EXCHANGEDB_RefreshSessionCallback)(void *cls,
-                                           unsigned long long rowid, /* FIXME: decide data type for serial_id! */
+                                           uint64_t rowid,
                                            const struct TALER_CoinSpendPublicKeyP *coin_pub,
                                            const struct TALER_CoinSpendSignatureP *coin_sig,
                                            const struct TALER_Amount *amount_with_fee,
@@ -681,7 +681,7 @@ typedef int
  */
 typedef int
 (*TALER_EXCHANGEDB_RefundCallback)(void *cls,
-                                   unsigned long long rowid, /* FIXME: decide data type for serial_id! */
+                                   uint64_t rowid,
                                    const struct TALER_CoinSpendPublicKeyP *coin_pub,
                                    const struct TALER_MerchantPublicKeyP *merchant_pub,
                                    const struct TALER_MerchantSignatureP *merchant_sig,
@@ -705,7 +705,7 @@ typedef int
  */
 typedef int
 (*TALER_EXCHANGEDB_ReserveInCallback)(void *cls,
-                                      unsigned long long rowid, /* FIXME: decide data type for serial_id! */
+                                      uint64_t rowid,
                                       const struct TALER_ReservePublicKeyP *reserve_pub,
                                       const struct TALER_Amount *credit,
                                       const json_t *sender_account_details,
@@ -729,7 +729,7 @@ typedef int
  */
 typedef int
 (*TALER_EXCHANGEDB_WithdrawCallback)(void *cls,
-                                     unsigned long long rowid, /* FIXME: decide data type for serial_id! */
+                                     uint64_t rowid,
                                      const struct GNUNET_HashCode *h_blind_ev,
                                      const struct TALER_DenominationPublicKey *denom_pub,
                                      const struct TALER_DenominationSignature *denom_sig,
@@ -814,7 +814,7 @@ typedef void
  */
 typedef void
 (*TALER_EXCHANGEDB_WirePreparationCallback) (void *cls,
-                                             unsigned long long rowid,
+                                             uint64_t rowid,
                                              const char *wire_method,
                                              const char *buf,
                                              size_t buf_size,
@@ -1105,7 +1105,7 @@ struct TALER_EXCHANGEDB_Plugin
   int
   (*mark_deposit_tiny) (void *cls,
                         struct TALER_EXCHANGEDB_Session *session,
-                        unsigned long long rowid);
+                        uint64_t rowid);
 
 
   /**
@@ -1137,7 +1137,7 @@ struct TALER_EXCHANGEDB_Plugin
   int
   (*mark_deposit_done) (void *cls,
                         struct TALER_EXCHANGEDB_Session *session,
-                        unsigned long long rowid);
+                        uint64_t rowid);
 
 
   /**
@@ -1551,7 +1551,7 @@ struct TALER_EXCHANGEDB_Plugin
   int
   (*wire_prepare_data_mark_finished)(void *cls,
                                      struct TALER_EXCHANGEDB_Session *session,
-                                     unsigned long long rowid);
+                                     uint64_t rowid);
 
 
   /**