add purses-get to build

contrib/gana

@@ -1 +1 @@
-Subproject commit c2580e60259ba3aea2e69ea9da43482008b90d7c
+Subproject commit bffe32411e8ded537c5615ea054b43b3f7334bcd

src/exchange/Makefile.am

@@ -106,6 +106,7 @@ taler_exchange_httpd_SOURCES = \
   taler-exchange-httpd_mhd.c taler-exchange-httpd_mhd.h \
   taler-exchange-httpd_purses_create.c taler-exchange-httpd_purses_create.h \
   taler-exchange-httpd_purses_deposit.c taler-exchange-httpd_purses_deposit.h \
+  taler-exchange-httpd_purses_get.c taler-exchange-httpd_purses_get.h \
   taler-exchange-httpd_purses_merge.c taler-exchange-httpd_purses_merge.h \
   taler-exchange-httpd_recoup.c taler-exchange-httpd_recoup.h \
   taler-exchange-httpd_recoup-refresh.c taler-exchange-httpd_recoup-refresh.h \

src/exchange/taler-exchange-httpd_purses_get.c

@@ -15,7 +15,7 @@
 */
 /**
  * @file taler-exchange-httpd_purses_get.c
- * @brief Handle GET /purses/$PID requests
+ * @brief Handle GET /purses/$PID/$TARGET requests
  * @author Christian Grothoff
  */
 #include "platform.h"
@@ -23,67 +23,350 @@
 #include <jansson.h>
 #include <microhttpd.h>
 #include "taler_mhd_lib.h"
-#include "taler-exchange-httpd_purses.h"
+#include "taler_dbevents.h"
+#include "taler-exchange-httpd_keys.h"
+#include "taler-exchange-httpd_purses_get.h"
 #include "taler-exchange-httpd_mhd.h"
 #include "taler-exchange-httpd_responses.h"
 
-// FIXME: add long-polling support!
+
+/**
+ * Information about an ongoing /purses GET operation.
+ */
+struct GetContext
+{
+  /**
+   * Kept in a DLL.
+   */
+  struct GetContext *next;
+
+  /**
+   * Kept in a DLL.
+   */
+  struct GetContext *prev;
+
+  /**
+   * Connection we are handling.
+   */
+  struct MHD_Connection *connection;
+
+  /**
+   * Subscription for the database event we are
+   * waiting for.
+   */
+  struct GNUNET_DB_EventHandler *eh;
+
+  /**
+   * Public key of our purse.
+   */
+  struct TALER_PurseContractPublicKeyP purse_pub;
+
+  /**
+   * When does this purse expire?
+   */
+  struct GNUNET_TIME_Timestamp purse_expiration;
+
+  /**
+   * When was this purse merged?
+   */
+  struct GNUNET_TIME_Timestamp merge_timestamp;
+
+  /**
+   * When was the full amount deposited into this purse?
+   */
+  struct GNUNET_TIME_Timestamp deposit_timestamp;
+
+  /**
+   * How much is the purse (supposed) to be worth?
+   */
+  struct TALER_Amount amount;
+
+  /**
+   * How much was deposited into the purse so far?
+   */
+  struct TALER_Amount deposited;
+
+  /**
+   * Hash over the contract of the purse.
+   */
+  struct TALER_PrivateContractHashP h_contract;
+
+  /**
+   * When will this request time out?
+   */
+  struct GNUNET_TIME_Absolute timeout;
+
+  /**
+   * true to wait for merge, false to wait for deposit.
+   */
+  bool wait_for_merge;
+
+  /**
+   * True if we are still suspended.
+   */
+  bool suspended;
+};
+
+
+/**
+ * Head of DLL of suspended GET requests.
+ */
+static struct GetContext *gc_head;
+
+/**
+ * Tail of DLL of suspended GET requests.
+ */
+static struct GetContext *gc_tail;
+
+
+void
+TEH_purses_get_cleanup ()
+{
+  struct GetContext *gc;
+
+  while (NULL != (gc = gc_head))
+  {
+    GNUNET_CONTAINER_DLL_remove (gc_head,
+                                 gc_tail,
+                                 gc);
+    if (gc->suspended)
+    {
+      gc->suspended = false;
+      MHD_resume_connection (gc->connection);
+    }
+  }
+}
+
+
+/**
+ * Function called once a connection is done to
+ * clean up the `struct GetContext` state.
+ *
+ * @param rc context to clean up for
+ */
+static void
+gc_cleanup (struct TEH_RequestContext *rc)
+{
+  struct GetContext *gc = rc->rh_ctx;
+
+  GNUNET_assert (! gc->suspended);
+  if (NULL != gc->eh)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                "Cancelling DB event listening\n");
+    TEH_plugin->event_listen_cancel (TEH_plugin->cls,
+                                     gc->eh);
+    gc->eh = NULL;
+  }
+  GNUNET_free (gc);
+}
+
+
+/**
+ * Function called on events received from Postgres.
+ * Wakes up long pollers.
+ *
+ * @param cls the `struct TEH_RequestContext *`
+ * @param extra additional event data provided
+ * @param extra_size number of bytes in @a extra
+ */
+static void
+db_event_cb (void *cls,
+             const void *extra,
+             size_t extra_size)
+{
+  struct TEH_RequestContext *rc = cls;
+  struct GetContext *gc = rc->rh_ctx;
+  struct GNUNET_AsyncScopeSave old_scope;
+
+  (void) extra;
+  (void) extra_size;
+  if (NULL == gc)
+    return; /* event triggered while main transaction
+               was still running */
+  if (! gc->suspended)
+    return; /* might get multiple wake-up events */
+  gc->suspended = false;
+  GNUNET_async_scope_enter (&rc->async_scope_id,
+                            &old_scope);
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Resuming from long-polling on purse\n");
+  TEH_check_invariants ();
+  GNUNET_CONTAINER_DLL_remove (gc_head,
+                               gc_tail,
+                               gc);
+  MHD_resume_connection (gc->connection);
+  TALER_MHD_daemon_trigger ();
+  TEH_check_invariants ();
+  GNUNET_async_scope_restore (&old_scope);
+}
+
 
 MHD_RESULT
-TEH_handler_pursess_get (struct TEH_RequestContext *rc,
+TEH_handler_purses_get (struct TEH_RequestContext *rc,
-                         const char *const args[1])
+                        const char *const args[2])
 {
-  struct TALER_PurseContractPublicKeyP purse_pub;
+  struct GetContext *gc = rc->rh_ctx;
-  enum GNUNET_DB_QueryStatus qs;
   MHD_RESULT res;
-  struct GNUNET_TIME_Timestamp merge_timestamp = {0};
 
-  if (GNUNET_OK !=
+  if (NULL == gc)
-      GNUNET_STRINGS_string_to_data (args[0],
-                                     strlen (args[0]),
-                                     &purse_pub,
-                                     sizeof (purse_pub)))
   {
-    GNUNET_break_op (0);
+    gc = GNUNET_new (struct GetContext);
-    return TALER_MHD_reply_with_error (rc->connection,
+    rc->rh_ctx = gc;
-                                       MHD_HTTP_BAD_REQUEST,
+    rc->rh_cleaner = &gc_cleanup;
-                                       TALER_EC_EXCHANGE_PURSES_INVALID_PURSE_PUB,
+    gc->connection = rc->connection;
-                                       args[0]);
+    if (GNUNET_OK !=
-  }
+        GNUNET_STRINGS_string_to_data (args[0],
+                                       strlen (args[0]),
+                                       &gc->purse_pub,
+                                       sizeof (gc->purse_pub)))
+    {
+      GNUNET_break_op (0);
+      return TALER_MHD_reply_with_error (rc->connection,
+                                         MHD_HTTP_BAD_REQUEST,
+                                         TALER_EC_EXCHANGE_GENERIC_PURSE_PUB_MALFORMED,
+                                         args[0]);
+    }
+    if (0 == strcmp (args[1],
+                     "merge"))
+      gc->wait_for_merge = true;
+    else if (0 == strcmp (args[1],
+                          "deposit"))
+      gc->wait_for_merge = false;
+    else
+    {
+      GNUNET_break_op (0);
+      return TALER_MHD_reply_with_error (rc->connection,
+                                         MHD_HTTP_BAD_REQUEST,
+                                         TALER_EC_EXCHANGE_PURSES_INVALID_WAIT_TARGET,
+                                         args[1]);
+    }
+
+    {
+      const char *long_poll_timeout_ms;
+
+      long_poll_timeout_ms
+        = MHD_lookup_connection_value (rc->connection,
+                                       MHD_GET_ARGUMENT_KIND,
+                                       "timeout_ms");
+      if (NULL != long_poll_timeout_ms)
+      {
+        unsigned int timeout_ms;
+        char dummy;
+        struct GNUNET_TIME_Relative timeout;
+
+        if (1 != sscanf (long_poll_timeout_ms,
+                         "%u%c",
+                         &timeout_ms,
+                         &dummy))
+        {
+          GNUNET_break_op (0);
+          return TALER_MHD_reply_with_error (rc->connection,
+                                             MHD_HTTP_BAD_REQUEST,
+                                             TALER_EC_GENERIC_PARAMETER_MALFORMED,
+                                             "timeout_ms (must be non-negative number)");
+        }
+        timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
+                                                 timeout_ms);
+        gc->timeout = GNUNET_TIME_relative_to_absolute (timeout);
+      }
+    }
+
+    if ( (GNUNET_TIME_absolute_is_future (gc->timeout)) &&
+         (NULL == gc->eh) )
+    {
+      struct TALER_PurseEventP rep = {
+        .header.size = htons (sizeof (rep)),
+        .header.type = htons (
+          gc->wait_for_merge
+        ? TALER_DBEVENT_EXCHANGE_PURSE_MERGED
+        : TALER_DBEVENT_EXCHANGE_PURSE_DEPOSITED),
+        .purse_pub = gc->purse_pub
+      };
+
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  "Starting DB event listening\n");
+      gc->eh = TEH_plugin->event_listen (
+        TEH_plugin->cls,
+        GNUNET_TIME_absolute_get_remaining (gc->timeout),
+        &rep.header,
+        &db_event_cb,
+        rc);
+    }
+  } /* end first-time initialization */
+
 #if FIXME
-  qs = TEH_plugin->select_purse (TEH_plugin->cls,
-                                 &purse_pub,
-                                 &merge_timestamp,
-                                 ...);
-  switch (qs)
   {
-  case GNUNET_DB_STATUS_HARD_ERROR:
+    enum GNUNET_DB_QueryStatus qs;
-    GNUNET_break (0);
+
+    qs = TEH_plugin->select_purse (TEH_plugin->cls,
+                                   &gc->purse_pub,
+                                   &gc->purse_expiration,
+                                   &gc->amount,
+                                   &gc->deposited,
+                                   &gc->h_contract,
+                                   &gc->merge_timestamp,
+                                   &gc->deposit_timestamp);
+    switch (qs)
+    {
+    case GNUNET_DB_STATUS_HARD_ERROR:
+      GNUNET_break (0);
+      return TALER_MHD_reply_with_error (rc->connection,
+                                         MHD_HTTP_INTERNAL_SERVER_ERROR,
+                                         TALER_EC_GENERIC_DB_FETCH_FAILED,
+                                         "select_purse");
+    case GNUNET_DB_STATUS_SOFT_ERROR:
+      GNUNET_break (0);
+      return TALER_MHD_reply_with_error (rc->connection,
+                                         MHD_HTTP_INTERNAL_SERVER_ERROR,
+                                         TALER_EC_GENERIC_DB_FETCH_FAILED,
+                                         "select_purse");
+    case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+      return TALER_MHD_reply_with_error (rc->connection,
+                                         MHD_HTTP_NOT_FOUND,
+                                         TALER_EC_EXCHANGE_PURSE_UNKNOWN,
+                                         NULL);
+    case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
+      break; /* handled below */
+    }
+  }
+  if (GNUNET_TIME_absolute_is_past (gc->purse_expiration))
+  {
     return TALER_MHD_reply_with_error (rc->connection,
-                                       MHD_HTTP_INTERNAL_SERVER_ERROR,
+                                       MHD_HTTP_GONE,
-                                       TALER_EC_GENERIC_DB_FETCH_FAILED,
+                                       TALER_EC_EXCHANGE_PURSE_EXPIRED,
-                                       "select_purses");
+                                       GNUNET_TIME_timestamp2s (
-  case GNUNET_DB_STATUS_SOFT_ERROR:
+                                         gc->purse_expiration));
-    GNUNET_break (0);
-    return TALER_MHD_reply_with_error (rc->connection,
-                                       MHD_HTTP_INTERNAL_SERVER_ERROR,
-                                       TALER_EC_GENERIC_DB_FETCH_FAILED,
-                                       "select_purses");
-  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
-    return TALER_MHD_reply_with_error (rc->connection,
-                                       MHD_HTTP_NOT_FOUND,
-                                       TALER_EC_EXCHANGE_PURSESS_UNKNOWN,
-                                       NULL);
-  case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
-    break; /* handled below */
   }
 #endif
+
+  // FIXME: compare amount to deposited amount;
+  // if below, set 'deposit_timestamp' to zero!
+
+  if (GNUNET_TIME_absolute_is_future (gc->timeout) &&
+      ( ((gc->wait_for_merge) &&
+         GNUNET_TIME_absolute_is_zero (gc->merge_timestamp.abs_time)) ||
+        ((! gc->wait_for_merge) &&
+         GNUNET_TIME_absolute_is_zero (gc->deposit_timestamp.abs_time)) ))
+  {
+    gc->suspended = true;
+    GNUNET_CONTAINER_DLL_insert (gc_head,
+                                 gc_tail,
+                                 gc);
+    MHD_suspend_connection (gc->connection);
+    return MHD_YES;
+  }
+
+  // FIXME: add exchange signature!?
+  // FIXME: return amount?
   res = TALER_MHD_REPLY_JSON_PACK (
     rc->connection,
     MHD_HTTP_OK,
     GNUNET_JSON_pack_timestamp ("merge_timestamp",
-                                &merge_timestamp));
+                                gc->merge_timestamp),
-  GNUNET_free (epurses);
+    GNUNET_JSON_pack_timestamp ("deposit_timestamp",
+                                gc->deposit_timestamp)
+    );
   return res;
 }
 

src/exchange/taler-exchange-httpd_purses_get.h

@@ -0,0 +1,51 @@
+/*
+  This file is part of TALER
+  Copyright (C) 2022 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-exchange-httpd_purses_get.h
+ * @brief Handle /purses/$PURSE_PUB/$TARGET GET requests
+ * @author Christian Grothoff
+ */
+#ifndef TALER_EXCHANGE_HTTPD_PURSES_GET_H
+#define TALER_EXCHANGE_HTTPD_PURSES_GET_H
+
+#include <microhttpd.h>
+#include "taler-exchange-httpd.h"
+
+
+/**
+ * Shutdown purses-get subsystem.  Resumes all
+ * suspended long-polling clients and cleans up
+ * data structures.
+ */
+void
+TEH_purses_get_cleanup (void);
+
+
+/**
+ * Handle a GET "/purses/$PID/$TARGET" request.  Parses the
+ * given "purse_pub" in @a args (which should contain the
+ * EdDSA public key of a purse) and then respond with the
+ * status of the purse.
+ *
+ * @param rc request context
+ * @param args array of additional options (length: 2, the purse_pub and a target)
+ * @return MHD result code
+ */
+MHD_RESULT
+TEH_handler_purses_get (struct TEH_RequestContext *rc,
+                        const char *const args[2]);
+
+#endif

src/include/taler_exchange_service.h

@@ -4127,8 +4127,8 @@ struct TALER_EXCHANGE_PurseGetHandle;
  *
  * @param exchange exchange handle
  * @param purse_priv private key of the purse
- * @param merge_timeout how long to wait for a merge to happen
+ * @param timeout how long to wait for a change to happen
- * @param deposit_timeout how long to wait for a deposit to happen
+ * @param wait_for_merge true to wait for a merge event, otherwise wait for a deposit event
  * @param cb function to call with the exchange's result
  * @param cb_cls closure for @a cb
  * @return the request handle; NULL upon error
@@ -4137,8 +4137,8 @@ struct TALER_EXCHANGE_PurseGetHandle *
 TALER_EXCHANGE_purse_get (
   struct TALER_EXCHANGE_Handle *exchange,
   const struct TALER_PurseContractPrivateKeyP *purse_priv,
-  struct GNUNET_TIME_Relative merge_timeout,
+  struct GNUNET_TIME_Relative timeout,
-  struct GNUNET_TIME_Relative deposit_timeout,
+  bool wait_for_merge,
   TALER_EXCHANGE_PurseGetCallback cb,
   void *cb_cls);
 

src/include/taler_exchangedb_plugin.h

@@ -121,6 +121,24 @@ struct TALER_ReserveEventP
 };
 
 
+/**
+ * Signature of events signalling a purse changed its status.
+ */
+struct TALER_PurseEventP
+{
+  /**
+   * Of type #TALER_DBEVENT_EXCHANGE_PURSE_MERGED or
+   * #TALER_DBEVENT_EXCHANGE_PURSE_DEPOSITED.
+   */
+  struct GNUNET_DB_EventHeaderP header;
+
+  /**
+   * Public key of the purse the event is about.
+   */
+  struct TALER_PurseContractPublicKeyP purse_pub;
+};
+
+
 /**
  * Signature of events signalling a KYC process was completed.
  */

src/lib/exchange_api_purses_get.c

@@ -94,6 +94,7 @@ handle_purse_get_finished (void *cls,
     break;
   case MHD_HTTP_OK:
     {
+      // FIXME: check exchange signature!
       struct GNUNET_JSON_Specification spec[] = {
         GNUNET_JSON_spec_timestamp ("merge_timestamp",
                                     &dr.details.success.merge_timestamp),
@@ -163,15 +164,15 @@ struct TALER_EXCHANGE_PurseGetHandle *
 TALER_EXCHANGE_purse_get (
   struct TALER_EXCHANGE_Handle *exchange,
   const struct TALER_PurseContractPrivateKeyP *purse_priv,
-  struct GNUNET_TIME_Relative merge_timeout,
+  struct GNUNET_TIME_Relative timeout,
-  struct GNUNET_TIME_Relative deposit_timeout,
+  bool wait_for_merge,
   TALER_EXCHANGE_PurseGetCallback cb,
   void *cb_cls)
 {
   struct TALER_EXCHANGE_PurseGetHandle *pgh;
   CURL *eh;
   struct TALER_PurseContractPublicKeyP purse_pub;
-  char arg_str[sizeof (purse_pub) * 2 + 48];
+  char arg_str[sizeof (purse_pub) * 2 + 64];
 
   if (GNUNET_YES !=
       TEAH_handle_is_ready (exchange))
@@ -188,18 +189,33 @@ TALER_EXCHANGE_purse_get (
   {
     char cpub_str[sizeof (purse_pub) * 2];
     char *end;
+    char timeout_str[32];
 
     end = GNUNET_STRINGS_data_to_string (&purse_pub,
                                          sizeof (purse_pub),
                                          cpub_str,
                                          sizeof (cpub_str));
     *end = '\0';
-    GNUNET_snprintf (arg_str,
+    GNUNET_snprintf (timeout_str,
-                     sizeof (arg_str),
+                     sizeof (timeout_str),
-                     "/purses/%s",
+                     "%llu",
-                     cpub_str);
+                     (unsigned long long)
+                     (timeout.rel_value_us
+                      / GNUNET_TIME_UNIT_MILLISECONDS.rel_value_us));
+    if (GNUNET_TIME_relative_is_zero (timeout))
+      GNUNET_snprintf (arg_str,
+                       sizeof (arg_str),
+                       "/purses/%s/%s",
+                       cpub_str,
+                       wait_for_merge ? "merge" : "deposit");
+    else
+      GNUNET_snprintf (arg_str,
+                       sizeof (arg_str),
+                       "/purses/%s/%s?timeout_ms=%s",
+                       cpub_str,
+                       wait_for_merge ? "merge" : "deposit",
+                       timeout_str);
   }
-
   pgh->url = TEAH_path_to_url (exchange,
                                arg_str);
   if (NULL == pgh->url)