Factor out last 6 functions (shit job)

9 files changed, 584 insertions, 373 deletions

diff --git a/src/backenddb/Makefile.am b/src/backenddb/Makefile.am
index daa14832..258ddb75 100644
--- a/src/backenddb/Makefile.am
+++ b/src/backenddb/Makefile.am
@@ -163,6 +163,9 @@ libtaler_plugin_merchantdb_postgres_la_SOURCES = \
   pg_update_webhook.h pg_update_webhook.c \
   pg_lookup_webhook_by_event.h pg_lookup_webhook_by_event.c \
   pg_delete_pending_webhook.h pg_delete_pending_webhook.c \
+  pg_insert_pending_webhook.h pg_insert_pending_webhook.c \
+  pg_update_pending_webhook.h pg_update_pending_webhook.c \
+  pg_lookup_pending_webhooks.h pg_lookup_pending_webhooks.c \
   plugin_merchantdb_postgres.c \
   pg_helper.h pg_helper.c
 libtaler_plugin_merchantdb_postgres_la_LIBADD = \
diff --git a/src/backenddb/pg_insert_pending_webhook.c b/src/backenddb/pg_insert_pending_webhook.c
new file mode 100644
index 00000000..abb3234f
--- /dev/null
+++ b/src/backenddb/pg_insert_pending_webhook.c
@@ -0,0 +1,70 @@
+/*
+   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 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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License along with
+   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file backenddb/pg_insert_pending_webhook.c
+ * @brief Implementation of the insert_pending_webhook function for Postgres
+ * @author Iván Ávalos
+ */
+#include "platform.h"
+#include <taler/taler_error_codes.h>
+#include <taler/taler_dbevents.h>
+#include <taler/taler_pq_lib.h>
+#include "pg_insert_pending_webhook.h"
+#include "pg_helper.h"
+
+enum GNUNET_DB_QueryStatus
+TMH_PG_insert_pending_webhook (void *cls,
+                               const char *instance_id,
+                               uint64_t webhook_serial,
+                               const char *url,
+                               const char *http_method,
+                               const char *header,
+                               const char *body)
+{
+  struct PostgresClosure *pg = cls;
+  struct GNUNET_PQ_QueryParam params[] = {
+    GNUNET_PQ_query_param_string (instance_id),
+    GNUNET_PQ_query_param_uint64 (&webhook_serial),
+    GNUNET_PQ_query_param_string (url),
+    GNUNET_PQ_query_param_string (http_method),
+    NULL == header
+    ? GNUNET_PQ_query_param_null ()
+    : GNUNET_PQ_query_param_string (header),
+    NULL == body
+    ? GNUNET_PQ_query_param_null ()
+    : GNUNET_PQ_query_param_string (body),
+    GNUNET_PQ_query_param_end
+  };
+  check_connection (pg);
+  PREPARE (pg,
+           "insert_pending_webhook",
+           "INSERT INTO merchant_pending_webhooks"
+           "(merchant_serial"
+           ",webhook_serial"
+           ",url"
+           ",http_method"
+           ",header"
+           ",body"
+           ")"
+           " SELECT mi.merchant_serial,"
+           " $2, $3, $4, $5, $6"
+           " FROM merchant_instances mi"
+           " WHERE mi.merchant_id=$1");
+
+  return GNUNET_PQ_eval_prepared_non_select (pg->conn,
+                                             "insert_pending_webhook",
+                                             params);
+}
diff --git a/src/backenddb/pg_insert_pending_webhook.h b/src/backenddb/pg_insert_pending_webhook.h
new file mode 100644
index 00000000..0107fd2c
--- /dev/null
+++ b/src/backenddb/pg_insert_pending_webhook.h
@@ -0,0 +1,49 @@
+/*
+   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 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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License along with
+   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file backenddb/pg_insert_pending_webhook.h
+ * @brief implementation of the insert_pending_webhook function for Postgres
+ * @author Iván Ávalos
+ */
+#ifndef PG_INSERT_PENDING_WEBHOOK_H
+#define PG_INSERT_PENDING_WEBHOOK_H
+
+#include <taler/taler_util.h>
+#include <taler/taler_json_lib.h>
+#include "taler_merchantdb_plugin.h"
+
+/**
+ * Insert webhook in the pending webhook.
+ *
+ * @param cls closure
+ * @param instance_id instance to insert webhook for
+ * @param webhook_serial webhook to insert in the pending webhook
+ * @param url to make the request to
+ * @param http_method for the webhook
+ * @param header of the webhook
+ * @param body of the webhook
+ * @return database result code
+ */
+enum GNUNET_DB_QueryStatus
+TMH_PG_insert_pending_webhook (void *cls,
+                               const char *instance_id,
+                               uint64_t webhook_serial,
+                               const char *url,
+                               const char *http_method,
+                               const char *header,
+                               const char *body);
+
+#endif
diff --git a/src/backenddb/pg_insert_reserve.c b/src/backenddb/pg_insert_reserve.c
index 67e2652a..173d9304 100644
--- a/src/backenddb/pg_insert_reserve.c
+++ b/src/backenddb/pg_insert_reserve.c
@@ -115,6 +115,22 @@ RETRY:
       GNUNET_PQ_query_param_end
     };
 
+    PREPARE(pg,
+            "insert_reserve_key",
+            "INSERT INTO merchant_reward_reserve_keys"
+            "(reserve_serial"
+            ",reserve_priv"
+            ",exchange_url"
+            ",master_pub"
+            ")"
+            "SELECT reserve_serial, $3, $4, $5"
+            " FROM merchant_reward_reserves"
+            " WHERE reserve_pub=$2"
+            "   AND merchant_serial="
+            "     (SELECT merchant_serial"
+            "        FROM merchant_instances"
+            "       WHERE merchant_id=$1)");
+
     qs = GNUNET_PQ_eval_prepared_non_select (pg->conn,
                                              "insert_reserve_key",
                                              params);
diff --git a/src/backenddb/pg_lookup_pending_webhooks.c b/src/backenddb/pg_lookup_pending_webhooks.c
new file mode 100644
index 00000000..d1d3eda9
--- /dev/null
+++ b/src/backenddb/pg_lookup_pending_webhooks.c
@@ -0,0 +1,261 @@
+/*
+   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 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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License along with
+   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file backenddb/pg_lookup_pending_webhooks.c
+ * @brief Implementation of the lookup_pending_webhooks function for Postgres
+ * @author Iván Ávalos
+ */
+#include "platform.h"
+#include <taler/taler_error_codes.h>
+#include <taler/taler_dbevents.h>
+#include <taler/taler_pq_lib.h>
+#include "pg_lookup_pending_webhooks.h"
+#include "pg_helper.h"
+
+/**
+ * Context used for lookup_pending_webhooks_cb().
+ */
+struct LookupPendingWebhookContext
+{
+  /**
+   * Function to call with the results.
+   */
+  TALER_MERCHANTDB_PendingWebhooksCallback cb;
+
+  /**
+   * Closure for @a cb.
+   */
+  void *cb_cls;
+
+  /**
+   * Did database result extraction fail?
+   */
+  bool extract_failed;
+};
+
+
+/**
+ * Function to be called with the results of a SELECT statement
+ * that has returned @a num_results results about webhook.
+ *
+ * @param[in,out] cls of type `struct LookupPendingWebhookContext *`
+ * @param result the postgres result
+ * @param num_results the number of results in @a result
+ */
+static void
+lookup_pending_webhooks_cb (void *cls,
+                            PGresult *result,
+                            unsigned int num_results)
+{
+  struct LookupPendingWebhookContext *pwlc = cls;
+
+  for (unsigned int i = 0; i < num_results; i++)
+  {
+    uint64_t webhook_pending_serial;
+    struct GNUNET_TIME_Absolute next_attempt;
+    uint32_t retries;
+    char *url;
+    char *http_method;
+    char *header = NULL;
+    char *body = NULL;
+    struct GNUNET_PQ_ResultSpec rs[] = {
+      GNUNET_PQ_result_spec_uint64 ("webhook_pending_serial",
+                                    &webhook_pending_serial),
+      GNUNET_PQ_result_spec_absolute_time ("next_attempt",
+                                           &next_attempt),
+      GNUNET_PQ_result_spec_uint32 ("retries",
+                                    &retries),
+      GNUNET_PQ_result_spec_string ("url",
+                                    &url),
+      GNUNET_PQ_result_spec_string ("http_method",
+                                    &http_method),
+      GNUNET_PQ_result_spec_allow_null (
+        GNUNET_PQ_result_spec_string ("header",
+                                      &header),
+        NULL),
+      GNUNET_PQ_result_spec_allow_null (
+        GNUNET_PQ_result_spec_string ("body",
+                                      &body),
+        NULL),
+      GNUNET_PQ_result_spec_end
+    };
+
+    if (GNUNET_OK !=
+        GNUNET_PQ_extract_result (result,
+                                  rs,
+                                  i))
+    {
+      GNUNET_break (0);
+      pwlc->extract_failed = true;
+      return;
+    }
+    pwlc->cb (pwlc->cb_cls,
+              webhook_pending_serial,
+              next_attempt,
+              retries,
+              url,
+              http_method,
+              header,
+              body);
+    GNUNET_PQ_cleanup_result (rs);
+  }
+}
+
+
+enum GNUNET_DB_QueryStatus
+TMH_PG_lookup_pending_webhooks (void *cls,
+                                TALER_MERCHANTDB_PendingWebhooksCallback cb,
+                                void *cb_cls)
+{
+  struct PostgresClosure *pg = cls;
+  struct LookupPendingWebhookContext pwlc = {
+    .cb = cb,
+    .cb_cls = cb_cls,
+    .extract_failed = false,
+  };
+  struct GNUNET_TIME_Absolute now = GNUNET_TIME_absolute_get ();
+  struct GNUNET_PQ_QueryParam params_null[] = {
+    GNUNET_PQ_query_param_absolute_time (&now),
+    GNUNET_PQ_query_param_end
+  };
+
+  enum GNUNET_DB_QueryStatus qs;
+
+  check_connection (pg);
+  PREPARE (pg,
+           "lookup_pending_webhooks",
+           "SELECT"
+           " webhook_pending_serial"
+           ",next_attempt"
+           ",retries"
+           ",url"
+           ",http_method"
+           ",header"
+           ",body"
+           " FROM merchant_pending_webhooks"
+           " WHERE next_attempt <= $1"
+           "   ORDER BY next_attempt ASC"
+           );
+
+  qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
+                                             "lookup_pending_webhooks",
+                                             params_null,
+                                             &lookup_pending_webhooks_cb,
+                                             &pwlc);
+
+  if (pwlc.extract_failed)
+    return GNUNET_DB_STATUS_HARD_ERROR;
+  return qs;
+}
+
+
+enum GNUNET_DB_QueryStatus
+TMH_PG_lookup_future_webhook (void *cls,
+                              TALER_MERCHANTDB_PendingWebhooksCallback cb,
+                              void *cb_cls)
+{
+  struct PostgresClosure *pg = cls;
+  struct LookupPendingWebhookContext pwlc = {
+    .cb = cb,
+    .cb_cls = cb_cls,
+    .extract_failed = false,
+  };
+  struct GNUNET_PQ_QueryParam params_null[] = {
+    GNUNET_PQ_query_param_end
+  };
+
+  enum GNUNET_DB_QueryStatus qs;
+
+  check_connection (pg);
+  PREPARE (pg,
+           "lookup_future_webhook",
+           "SELECT"
+           " webhook_pending_serial"
+           ",next_attempt"
+           ",retries"
+           ",url"
+           ",http_method"
+           ",header"
+           ",body"
+           " FROM merchant_pending_webhooks"
+           " ORDER BY next_attempt ASC LIMIT 1"
+           );
+
+  qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
+                                             "lookup_future_webhook",
+                                             params_null,
+                                             &lookup_pending_webhooks_cb,
+                                             &pwlc);
+
+  if (pwlc.extract_failed)
+    return GNUNET_DB_STATUS_HARD_ERROR;
+  return qs;
+}
+
+
+enum GNUNET_DB_QueryStatus
+TMH_PG_lookup_all_webhooks (void *cls,
+                            const char *instance_id,
+                            uint64_t min_row,
+                            uint32_t max_results,
+                            TALER_MERCHANTDB_PendingWebhooksCallback cb,
+                            void *cb_cls)
+{
+  struct PostgresClosure *pg = cls;
+  struct LookupPendingWebhookContext pwlc = {
+    .cb = cb,
+    .cb_cls = cb_cls,
+    .extract_failed = false,
+  };
+  uint64_t max_results64 = max_results;
+  struct GNUNET_PQ_QueryParam params[] = {
+    GNUNET_PQ_query_param_string (instance_id),
+    GNUNET_PQ_query_param_uint64 (&min_row),
+    GNUNET_PQ_query_param_uint64 (&max_results64),
+    GNUNET_PQ_query_param_end
+  };
+
+  enum GNUNET_DB_QueryStatus qs;
+
+  check_connection (pg);
+  PREPARE (pg,
+           "lookup_all_webhooks",
+           " SELECT"
+           " webhook_pending_serial"
+           ",next_attempt"
+           ",retries"
+           ",url"
+           ",http_method"
+           ",header"
+           ",body"
+           " FROM merchant_pending_webhooks"
+           " JOIN merchant_instances"
+           "   USING (merchant_serial)"
+           " WHERE merchant_instances.merchant_id=$1"
+           " AND webhook_pending_serial > $2"
+           "  ORDER BY webhook_pending_serial"
+           "   ASC LIMIT $3");
+
+  qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
+                                             "lookup_all_webhooks",
+                                             params,
+                                             &lookup_pending_webhooks_cb,
+                                             &pwlc);
+
+  if (pwlc.extract_failed)
+    return GNUNET_DB_STATUS_HARD_ERROR;
+  return qs;
+}
diff --git a/src/backenddb/pg_lookup_pending_webhooks.h b/src/backenddb/pg_lookup_pending_webhooks.h
new file mode 100644
index 00000000..d7322f56
--- /dev/null
+++ b/src/backenddb/pg_lookup_pending_webhooks.h
@@ -0,0 +1,78 @@
+/*
+   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 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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License along with
+   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file backenddb/pg_lookup_pending_webhooks.h
+ * @brief implementation of the lookup_pending_webhooks function for Postgres
+ * @author Iván Ávalos
+ */
+#ifndef PG_LOOKUP_PENDING_WEBHOOKS_H
+#define PG_LOOKUP_PENDING_WEBHOOKS_H
+
+#include <taler/taler_util.h>
+#include <taler/taler_json_lib.h>
+#include "taler_merchantdb_plugin.h"
+
+/**
+ * Lookup the webhook that need to be send in priority.
+ * send.
+ *
+ * @param cls closure
+ * @param cb pending webhook callback
+ * @param cb_cls callback closure
+ */
+// WHERE next_attempt <= now ORDER BY next_attempt ASC
+enum GNUNET_DB_QueryStatus
+TMH_PG_lookup_pending_webhooks (void *cls,
+                                TALER_MERCHANTDB_PendingWebhooksCallback cb,
+                                void *cb_cls);
+
+
+/**
+ * Lookup future webhook in the pending webhook that need to be send.
+ * With that we can know how long the system can 'sleep'.
+ *
+ * @param cls closure
+ * @param cb pending webhook callback
+ * @param cb_cls callback closure
+ */
+// ORDER BY next_attempt ASC LIMIT 1
+enum GNUNET_DB_QueryStatus
+TMH_PG_lookup_future_webhook (void *cls,
+                              TALER_MERCHANTDB_PendingWebhooksCallback cb,
+                              void *cb_cls);
+
+
+/**
+   * Lookup all the webhooks in the pending webhook.
+   * Use by the administrator
+   *
+   * @param cls closure
+   * @param instance_id to lookup webhooks for this instance particularly
+   * @param min_row to see the list of the pending webhook that it is started with this minimum row.
+   * @param max_results to see the list of the pending webhook that it is end with this max results.
+   * @param cb pending webhook callback
+   * @param cb_cls callback closure
+   */
+// WHERE webhook_pending_serial > min_row ORDER BY webhook_pending_serial ASC LIMIT max_results
+enum GNUNET_DB_QueryStatus
+TMH_PG_lookup_all_webhooks (void *cls,
+                            const char *instance_id,
+                            uint64_t min_row,
+                            uint32_t max_results,
+                            TALER_MERCHANTDB_PendingWebhooksCallback cb,
+                            void *cb_cls);
+
+#endif
diff --git a/src/backenddb/pg_update_pending_webhook.c b/src/backenddb/pg_update_pending_webhook.c
new file mode 100644
index 00000000..23ef5f04
--- /dev/null
+++ b/src/backenddb/pg_update_pending_webhook.c
@@ -0,0 +1,51 @@
+/*
+   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 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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License along with
+   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file backenddb/pg_update_pending_webhook.c
+ * @brief Implementation of the update_pending_webhook function for Postgres
+ * @author Iván Ávalos
+ */
+#include "platform.h"
+#include <taler/taler_error_codes.h>
+#include <taler/taler_dbevents.h>
+#include <taler/taler_pq_lib.h>
+#include "pg_update_pending_webhook.h"
+#include "pg_helper.h"
+
+enum GNUNET_DB_QueryStatus
+TMH_PG_update_pending_webhook (void *cls,
+                               uint64_t webhook_pending_serial,
+                               struct GNUNET_TIME_Absolute next_attempt)
+{
+  struct PostgresClosure *pg = cls;
+  struct GNUNET_PQ_QueryParam params[] = {
+    GNUNET_PQ_query_param_uint64 (&webhook_pending_serial),
+    GNUNET_PQ_query_param_absolute_time (&next_attempt),
+    GNUNET_PQ_query_param_end
+  };
+
+  check_connection (pg);
+  PREPARE (pg,
+           "update_pending_webhook",
+           "UPDATE merchant_pending_webhooks SET"
+           " retries=retries+1"
+           ",next_attempt=$2"
+           " WHERE webhook_pending_serial=$1");
+
+  return GNUNET_PQ_eval_prepared_non_select (pg->conn,
+                                             "update_pending_webhook",
+                                             params);
+}
diff --git a/src/backenddb/pg_update_pending_webhook.h b/src/backenddb/pg_update_pending_webhook.h
new file mode 100644
index 00000000..2ccf519e
--- /dev/null
+++ b/src/backenddb/pg_update_pending_webhook.h
@@ -0,0 +1,41 @@
+/*
+   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 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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License along with
+   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file backenddb/pg_update_pending_webhook.h
+ * @brief implementation of the update_pending_webhook function for Postgres
+ * @author Iván Ávalos
+ */
+#ifndef PG_UPDATE_PENDING_WEBHOOK_H
+#define PG_UPDATE_PENDING_WEBHOOK_H
+
+#include <taler/taler_util.h>
+#include <taler/taler_json_lib.h>
+#include "taler_merchantdb_plugin.h"
+
+/**
+ * Update the pending webhook. It is use if the webhook can't be send.
+ *
+ * @param cls closure
+ * @param webhook_pending_serial pending_webhook that need to be update
+ * @param next_attempt when to try the webhook next
+ * @return database result code
+ */
+enum GNUNET_DB_QueryStatus
+TMH_PG_update_pending_webhook (void *cls,
+                               uint64_t webhook_pending_serial,
+                               struct GNUNET_TIME_Absolute next_attempt);
+
+#endif
diff --git a/src/backenddb/plugin_merchantdb_postgres.c b/src/backenddb/plugin_merchantdb_postgres.c
index 35f2ecb9..1f1f54c6 100644
--- a/src/backenddb/plugin_merchantdb_postgres.c
+++ b/src/backenddb/plugin_merchantdb_postgres.c
@@ -128,6 +128,11 @@
 #include "pg_update_webhook.h"
 #include "pg_lookup_webhook_by_event.h"
 #include "pg_delete_pending_webhook.h"
+#include "pg_insert_pending_webhook.h"
+#include "pg_update_pending_webhook.h"
+#include "pg_lookup_pending_webhooks.h"
+// ^^^^^ + lookup_future_webhook
+// ^^^^^ + lookup_all_webhooks
 #include "pg_set_transfer_status_to_confirmed.h"
 #include "pg_insert_exchange_keys.h"
 #include "pg_select_exchange_keys.h"
@@ -290,295 +295,6 @@ check_connection (struct PostgresClosure *pg)
   GNUNET_PQ_reconnect_if_down (pg->conn);
 }
 
-
-/**
- * Insert webhook in the pending webhook.
- *
- * @param cls closure
- * @param instance_id instance to insert webhook for
- * @param webhook_serial webhook to insert in the pending webhook
- * @param url to make the request to
- * @param http_method for the webhook
- * @param header of the webhook
- * @param body of the webhook
- * @return database result code
- */
-static enum GNUNET_DB_QueryStatus
-postgres_insert_pending_webhook (void *cls,
-                                 const char *instance_id,
-                                 uint64_t webhook_serial,
-                                 const char *url,
-                                 const char *http_method,
-                                 const char *header,
-                                 const char *body)
-{
-  struct PostgresClosure *pg = cls;
-  struct GNUNET_PQ_QueryParam params[] = {
-    GNUNET_PQ_query_param_string (instance_id),
-    GNUNET_PQ_query_param_uint64 (&webhook_serial),
-    GNUNET_PQ_query_param_string (url),
-    GNUNET_PQ_query_param_string (http_method),
-    NULL == header
-    ? GNUNET_PQ_query_param_null ()
-    : GNUNET_PQ_query_param_string (header),
-    NULL == body
-    ? GNUNET_PQ_query_param_null ()
-    : GNUNET_PQ_query_param_string (body),
-    GNUNET_PQ_query_param_end
-  };
-  check_connection (pg);
-  return GNUNET_PQ_eval_prepared_non_select (pg->conn,
-                                             "insert_pending_webhook",
-                                             params);
-}
-
-
-/**
- * Context used for postgres_lookup_future_webhook().
- */
-struct LookupPendingWebhookContext
-{
-  /**
-   * Function to call with the results.
-   */
-  TALER_MERCHANTDB_PendingWebhooksCallback cb;
-
-  /**
-   * Closure for @a cb.
-   */
-  void *cb_cls;
-
-  /**
-   * Did database result extraction fail?
-   */
-  bool extract_failed;
-};
-
-
-/**
- * Function to be called with the results of a SELECT statement
- * that has returned @a num_results results about webhook.
- *
- * @param[in,out] cls of type `struct LookupPendingWebhookContext *`
- * @param result the postgres result
- * @param num_results the number of results in @a result
- */
-static void
-lookup_pending_webhooks_cb (void *cls,
-                            PGresult *result,
-                            unsigned int num_results)
-{
-  struct LookupPendingWebhookContext *pwlc = cls;
-
-  for (unsigned int i = 0; i < num_results; i++)
-  {
-    uint64_t webhook_pending_serial;
-    struct GNUNET_TIME_Absolute next_attempt;
-    uint32_t retries;
-    char *url;
-    char *http_method;
-    char *header = NULL;
-    char *body = NULL;
-    struct GNUNET_PQ_ResultSpec rs[] = {
-      GNUNET_PQ_result_spec_uint64 ("webhook_pending_serial",
-                                    &webhook_pending_serial),
-      GNUNET_PQ_result_spec_absolute_time ("next_attempt",
-                                           &next_attempt),
-      GNUNET_PQ_result_spec_uint32 ("retries",
-                                    &retries),
-      GNUNET_PQ_result_spec_string ("url",
-                                    &url),
-      GNUNET_PQ_result_spec_string ("http_method",
-                                    &http_method),
-      GNUNET_PQ_result_spec_allow_null (
-        GNUNET_PQ_result_spec_string ("header",
-                                      &header),
-        NULL),
-      GNUNET_PQ_result_spec_allow_null (
-        GNUNET_PQ_result_spec_string ("body",
-                                      &body),
-        NULL),
-      GNUNET_PQ_result_spec_end
-    };
-
-    if (GNUNET_OK !=
-        GNUNET_PQ_extract_result (result,
-                                  rs,
-                                  i))
-    {
-      GNUNET_break (0);
-      pwlc->extract_failed = true;
-      return;
-    }
-    pwlc->cb (pwlc->cb_cls,
-              webhook_pending_serial,
-              next_attempt,
-              retries,
-              url,
-              http_method,
-              header,
-              body);
-    GNUNET_PQ_cleanup_result (rs);
-  }
-}
-
-
-/**
- * Lookup the webhook that need to be send in priority.
- * send.
- *
- * @param cls closure
- * @param cb pending webhook callback
- * @param cb_cls callback closure
- */
-// WHERE next_attempt <= now ORDER BY next_attempt ASC
-static enum GNUNET_DB_QueryStatus
-postgres_lookup_pending_webhooks (void *cls,
-                                  TALER_MERCHANTDB_PendingWebhooksCallback cb,
-                                  void *cb_cls)
-{
-  struct PostgresClosure *pg = cls;
-  struct LookupPendingWebhookContext pwlc = {
-    .cb = cb,
-    .cb_cls = cb_cls,
-    .extract_failed = false,
-  };
-  struct GNUNET_TIME_Absolute now = GNUNET_TIME_absolute_get ();
-  struct GNUNET_PQ_QueryParam params_null[] = {
-    GNUNET_PQ_query_param_absolute_time (&now),
-    GNUNET_PQ_query_param_end
-  };
-
-  enum GNUNET_DB_QueryStatus qs;
-
-  check_connection (pg);
-  qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
-                                             "lookup_pending_webhooks",
-                                             params_null,
-                                             &lookup_pending_webhooks_cb,
-                                             &pwlc);
-
-  if (pwlc.extract_failed)
-    return GNUNET_DB_STATUS_HARD_ERROR;
-  return qs;
-}
-
-
-/**
- * Lookup future webhook in the pending webhook that need to be send.
- * With that we can know how long the system can 'sleep'.
- *
- * @param cls closure
- * @param cb pending webhook callback
- * @param cb_cls callback closure
- */
-// ORDER BY next_attempt ASC LIMIT 1
-static enum GNUNET_DB_QueryStatus
-postgres_lookup_future_webhook (void *cls,
-                                TALER_MERCHANTDB_PendingWebhooksCallback cb,
-                                void *cb_cls)
-{
-  struct PostgresClosure *pg = cls;
-  struct LookupPendingWebhookContext pwlc = {
-    .cb = cb,
-    .cb_cls = cb_cls,
-    .extract_failed = false,
-  };
-  struct GNUNET_PQ_QueryParam params_null[] = {
-    GNUNET_PQ_query_param_end
-  };
-
-  enum GNUNET_DB_QueryStatus qs;
-
-  check_connection (pg);
-  qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
-                                             "lookup_future_webhook",
-                                             params_null,
-                                             &lookup_pending_webhooks_cb,
-                                             &pwlc);
-
-  if (pwlc.extract_failed)
-    return GNUNET_DB_STATUS_HARD_ERROR;
-  return qs;
-}
-
-
-/**
-   * Lookup all the webhooks in the pending webhook.
-   * Use by the administrator
-   *
-   * @param cls closure
-   * @param instance_id to lookup webhooks for this instance particularly
-   * @param min_row to see the list of the pending webhook that it is started with this minimum row.
-   * @param max_results to see the list of the pending webhook that it is end with this max results.
-   * @param cb pending webhook callback
-   * @param cb_cls callback closure
-   */
-// WHERE webhook_pending_serial > min_row ORDER BY webhook_pending_serial ASC LIMIT max_results
-static enum GNUNET_DB_QueryStatus
-postgres_lookup_all_webhooks (void *cls,
-                              const char *instance_id,
-                              uint64_t min_row,
-                              uint32_t max_results,
-                              TALER_MERCHANTDB_PendingWebhooksCallback cb,
-                              void *cb_cls)
-{
-  struct PostgresClosure *pg = cls;
-  struct LookupPendingWebhookContext pwlc = {
-    .cb = cb,
-    .cb_cls = cb_cls,
-    .extract_failed = false,
-  };
-  uint64_t max_results64 = max_results;
-  struct GNUNET_PQ_QueryParam params[] = {
-    GNUNET_PQ_query_param_string (instance_id),
-    GNUNET_PQ_query_param_uint64 (&min_row),
-    GNUNET_PQ_query_param_uint64 (&max_results64),
-    GNUNET_PQ_query_param_end
-  };
-
-  enum GNUNET_DB_QueryStatus qs;
-
-  check_connection (pg);
-  qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
-                                             "lookup_all_webhooks",
-                                             params,
-                                             &lookup_pending_webhooks_cb,
-                                             &pwlc);
-
-  if (pwlc.extract_failed)
-    return GNUNET_DB_STATUS_HARD_ERROR;
-  return qs;
-}
-
-
-/**
- * Update the pending webhook. It is use if the webhook can't be send.
- *
- * @param cls closure
- * @param webhook_pending_serial pending_webhook that need to be update
- * @param next_attempt when to try the webhook next
- * @return database result code
- */
-static enum GNUNET_DB_QueryStatus
-postgres_update_pending_webhook (void *cls,
-                                 uint64_t webhook_pending_serial,
-                                 struct GNUNET_TIME_Absolute next_attempt)
-{
-  struct PostgresClosure *pg = cls;
-  struct GNUNET_PQ_QueryParam params[] = {
-    GNUNET_PQ_query_param_uint64 (&webhook_pending_serial),
-    GNUNET_PQ_query_param_absolute_time (&next_attempt),
-    GNUNET_PQ_query_param_end
-  };
-
-  check_connection (pg);
-  return GNUNET_PQ_eval_prepared_non_select (pg->conn,
-                                             "update_pending_webhook",
-                                             params);
-}
-
-
 /**
  * Establish connection to the database.
  *
@@ -592,85 +308,6 @@ postgres_connect (void *cls)
   struct GNUNET_PQ_PreparedStatement ps[] = {
     GNUNET_PQ_make_prepare ("end_transaction",
                             "COMMIT"),
-    /* For postgres_insert_reserve() */
-    GNUNET_PQ_make_prepare ("insert_reserve_key",
-                            "INSERT INTO merchant_reward_reserve_keys"
-                            "(reserve_serial"
-                            ",reserve_priv"
-                            ",exchange_url"
-                            ",master_pub"
-                            ")"
-                            "SELECT reserve_serial, $3, $4, $5"
-                            " FROM merchant_reward_reserves"
-                            " WHERE reserve_pub=$2"
-                            "   AND merchant_serial="
-                            "     (SELECT merchant_serial"
-                            "        FROM merchant_instances"
-                            "       WHERE merchant_id=$1)"),
-    /* for postgres_insert_pending_webhook() */
-    GNUNET_PQ_make_prepare ("insert_pending_webhook",
-                            "INSERT INTO merchant_pending_webhooks"
-                            "(merchant_serial"
-                            ",webhook_serial"
-                            ",url"
-                            ",http_method"
-                            ",header"
-                            ",body"
-                            ")"
-                            " SELECT mi.merchant_serial,"
-                            " $2, $3, $4, $5, $6"
-                            " FROM merchant_instances mi"
-                            " WHERE mi.merchant_id=$1"),
-    /* for postgres_update_pending_webhook() */
-    GNUNET_PQ_make_prepare ("update_pending_webhook",
-                            "UPDATE merchant_pending_webhooks SET"
-                            " retries=retries+1"
-                            ",next_attempt=$2"
-                            " WHERE webhook_pending_serial=$1"),
-    /* for postgres_lookup_pending_webhooks() */
-    GNUNET_PQ_make_prepare ("lookup_pending_webhooks",
-                            "SELECT"
-                            " webhook_pending_serial"
-                            ",next_attempt"
-                            ",retries"
-                            ",url"
-                            ",http_method"
-                            ",header"
-                            ",body"
-                            " FROM merchant_pending_webhooks"
-                            " WHERE next_attempt <= $1"
-                            "   ORDER BY next_attempt ASC"
-                            ),
-    /* for postgres_lookup_future_webhook() */
-    GNUNET_PQ_make_prepare ("lookup_future_webhook",
-                            "SELECT"
-                            " webhook_pending_serial"
-                            ",next_attempt"
-                            ",retries"
-                            ",url"
-                            ",http_method"
-                            ",header"
-                            ",body"
-                            " FROM merchant_pending_webhooks"
-                            " ORDER BY next_attempt ASC LIMIT 1"
-                            ),
-    /* for postgres_lookup_all_webhooks() */
-    GNUNET_PQ_make_prepare ("lookup_all_webhooks",
-                            " SELECT"
-                            " webhook_pending_serial"
-                            ",next_attempt"
-                            ",retries"
-                            ",url"
-                            ",http_method"
-                            ",header"
-                            ",body"
-                            " FROM merchant_pending_webhooks"
-                            " JOIN merchant_instances"
-                            "   USING (merchant_serial)"
-                            " WHERE merchant_instances.merchant_id=$1"
-                            " AND webhook_pending_serial > $2"
-                            "  ORDER BY webhook_pending_serial"
-                            "   ASC LIMIT $3"),
     GNUNET_PQ_PREPARED_STATEMENT_END
   };
   struct GNUNET_PQ_ExecuteStatement es[] = {
@@ -946,13 +583,18 @@ libtaler_plugin_merchantdb_postgres_init (void *cls)
     = &TMH_PG_update_webhook;
   plugin->lookup_webhook_by_event
     = &TMH_PG_lookup_webhook_by_event;
-  plugin->lookup_all_webhooks = &postgres_lookup_all_webhooks;
-  plugin->lookup_future_webhook = &postgres_lookup_future_webhook;
-  plugin->lookup_pending_webhooks = &postgres_lookup_pending_webhooks;
+  plugin->lookup_all_webhooks
+    = &TMH_PG_lookup_all_webhooks;
+  plugin->lookup_future_webhook
+    = &TMH_PG_lookup_future_webhook;
+  plugin->lookup_pending_webhooks
+    = &TMH_PG_lookup_pending_webhooks;
   plugin->delete_pending_webhook
     = &TMH_PG_delete_pending_webhook;
-  plugin->insert_pending_webhook = &postgres_insert_pending_webhook;
-  plugin->update_pending_webhook = &postgres_update_pending_webhook;
+  plugin->insert_pending_webhook
+    = &TMH_PG_insert_pending_webhook;
+  plugin->update_pending_webhook
+    = &TMH_PG_update_pending_webhook;
   plugin->delete_exchange_accounts
     = &TMH_PG_delete_exchange_accounts;
   plugin->select_accounts_by_exchange