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

/*
  This file is part of TALER
  Copyright (C) 2023 Taler Systems SA

  TALER is free software; you can redistribute it and/or modify it under the
  terms of the GNU Affero General Public License as published by the Free Software
  Foundation; either version 3, or (at your option) any later version.

  TALER is distributed in the hope that it will be useful, but WITHOUT ANY
  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
  A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more details.

  You should have received a copy of the GNU Affero General Public License along with
  TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
*/
/**
 * @file taler-merchant-webhook.c
 * @brief Process that runs webhooks triggered by the merchant backend
 * @author Priscilla HUANG
 */
#include "platform.h"
#include <gnunet/gnunet_util_lib.h>
#include <jansson.h>
#include <pthread.h>
#include "taler_merchantdb_lib.h"
#include "taler_merchantdb_plugin.h"
#include <taler/taler_dbevents.h>


struct WorkResponse
{
  struct WorkResponse *next;
  struct WorkResponse *prev;
  struct GNUNET_CURL_Job *job;
  uint64_t webhook_pending_serial;
  char *body;
  struct curl_slist *job_headers;
};


static struct WorkResponse *w_head;

static struct WorkResponse *w_tail;

static struct GNUNET_DB_EventHandler *event_handler;

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

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

/**
 * Next task to run, if any.
 */
static struct GNUNET_SCHEDULER_Task *task;

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

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

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

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


/**
 * We're being aborted with CTRL-C (or SIGTERM). Shut down.
 *
 * @param cls closure
 */
static void
shutdown_task (void *cls)
{
  struct WorkResponse *w;

  (void) cls;
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Running shutdown\n");
  if (NULL != event_handler)
  {
    db_plugin->event_listen_cancel (event_handler);
    event_handler = NULL;
  }
  if (NULL != task)
  {
    GNUNET_SCHEDULER_cancel (task);
    task = NULL;
  }
  while (NULL != (w = w_head))
  {
    GNUNET_CONTAINER_DLL_remove (w_head,
                                 w_tail,
                                 w);
    GNUNET_CURL_job_cancel (w->job);
    curl_slist_free_all (w->job_headers);
    GNUNET_free (w->body);
    GNUNET_free (w);
  }
  db_plugin->rollback (db_plugin->cls); /* just in case */
  TALER_MERCHANTDB_plugin_unload (db_plugin);
  db_plugin = NULL;
  cfg = NULL;
  if (NULL != ctx)
  {
    GNUNET_CURL_fini (ctx);
    ctx = NULL;
  }
  if (NULL != rc)
  {
    GNUNET_CURL_gnunet_rc_destroy (rc);
    rc = NULL;
  }
}


/**
 * Select webhook to process.
 *
 * @param cls NULL
 */
static void
select_work (void *cls);


/**
 * This function is used by the function `pending_webhooks_cb`. According to the response code,
 * we delete or update the webhook.
 *
 * @param cls closure
 * @param response_code HTTP response code from server, 0 on hard error
 * @param body http body of the response
 * @param body_size number of bytes in @a body
 */
static void
handle_webhook_response (void *cls,
                         long response_code,
                         const void *body,
                         size_t body_size)
{
  struct WorkResponse *w = cls;

  (void) body;
  (void) body_size;
  w->job = NULL;
  GNUNET_CONTAINER_DLL_remove (w_head,
                               w_tail,
                               w);
  GNUNET_free (w->body);
  curl_slist_free_all (w->job_headers);
  if (NULL == w_head)
    task = GNUNET_SCHEDULER_add_now (&select_work,
                                     NULL);
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Webhook %llu returned with status %ld\n",
              (unsigned long long) w->webhook_pending_serial,
              response_code);
  if (2 == response_code / 100) /* any 2xx http status code is OK! */
  {
    enum GNUNET_DB_QueryStatus qs;

    qs = db_plugin->delete_pending_webhook (db_plugin->cls,
                                            w->webhook_pending_serial);
    GNUNET_free (w);
    switch (qs)
    {
    case GNUNET_DB_STATUS_HARD_ERROR:
    case GNUNET_DB_STATUS_SOFT_ERROR:
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  "Failed to delete webhook, delete returned: %d\n",
                  qs);
      global_ret = EXIT_FAILURE;
      GNUNET_SCHEDULER_shutdown ();
      return;
    case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                  "Delete returned: %d\n",
                  qs);
      return;
    case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                  "Delete returned: %d\n",
                  qs);
      return;
    }
    GNUNET_assert (0);
  }

  {
    struct GNUNET_TIME_Relative next_attempt;
    enum GNUNET_DB_QueryStatus qs;
    switch (response_code)
    {
    case MHD_HTTP_BAD_REQUEST:
      next_attempt = GNUNET_TIME_UNIT_FOREVER_REL;   // never try again
      break;
    case MHD_HTTP_INTERNAL_SERVER_ERROR:
      next_attempt = GNUNET_TIME_UNIT_MINUTES;
      break;
    case MHD_HTTP_FORBIDDEN:
      next_attempt = GNUNET_TIME_UNIT_MINUTES;
      break;
    default:
      next_attempt = GNUNET_TIME_UNIT_HOURS;
      break;
    }
    qs = db_plugin->update_pending_webhook (db_plugin->cls,
                                            w->webhook_pending_serial,
                                            GNUNET_TIME_relative_to_absolute (
                                              next_attempt));
    GNUNET_free (w);
    switch (qs)
    {
    case GNUNET_DB_STATUS_HARD_ERROR:
    case GNUNET_DB_STATUS_SOFT_ERROR:
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  "Failed to update pending webhook to next in %s Rval: %d\n",
                  GNUNET_TIME_relative2s (next_attempt,
                                          true),
                  qs);
      global_ret = EXIT_FAILURE;
      GNUNET_SCHEDULER_shutdown ();
      return;
    case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                  "Next in %s Rval: %d\n",
                  GNUNET_TIME_relative2s (next_attempt, true),
                  qs);
      return;
    case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                  "Next in %s Rval: %d\n",
                  GNUNET_TIME_relative2s (next_attempt, true),
                  qs);
      return;
    }
    GNUNET_assert (0);
  }
}


/**
 * Typically called by `select_work`.
 *
 * @param cls a `json_t *` JSON array to build
 * @param webhook_pending_serial reference to the configured webhook template.
 * @param next_attempt is the time we should make the next request to the webhook.
 * @param retries how often have we tried this request to the webhook.
 * @param url to make request to
 * @param http_method use for the webhook
 * @param header of the webhook
 * @param body of the webhook
 */
static void
pending_webhooks_cb (void *cls,
                     uint64_t webhook_pending_serial,
                     struct GNUNET_TIME_Absolute next_attempt,
                     uint32_t retries,
                     const char *url,
                     const char *http_method,
                     const char *header,
                     const char *body)
{
  struct WorkResponse *w = GNUNET_new (struct WorkResponse);
  CURL *eh;
  struct curl_slist *job_headers = NULL;

  (void) retries;
  (void) next_attempt;
  (void) cls;
  GNUNET_CONTAINER_DLL_insert (w_head,
                               w_tail,
                               w);
  w->webhook_pending_serial = webhook_pending_serial;
  eh = curl_easy_init ();
  GNUNET_assert (NULL != eh);
  GNUNET_assert (CURLE_OK ==
                 curl_easy_setopt (eh,
                                   CURLOPT_CUSTOMREQUEST,
                                   http_method));
  GNUNET_assert (CURLE_OK ==
                 curl_easy_setopt (eh,
                                   CURLOPT_URL,
                                   url));
  GNUNET_assert (CURLE_OK ==
                 curl_easy_setopt (eh,
                                   CURLOPT_VERBOSE,
                                   0L));

  /* conversion body data */
  if (NULL != body)
  {
    w->body = GNUNET_strdup (body);
    GNUNET_assert (CURLE_OK ==
                   curl_easy_setopt (eh,
                                     CURLOPT_POSTFIELDS,
                                     w->body));
  }
  /* conversion header to job_headers data */
  if (NULL != header)
  {
    char *header_copy = GNUNET_strdup (header);

    for (const char *tok = strtok (header_copy, "\n");
         NULL != tok;
         tok = strtok (NULL, "\n"))
    {
      // extract all Key: value from 'header_copy'!
      job_headers = curl_slist_append (job_headers,
                                       tok);
    }
    GNUNET_free (header_copy);
    GNUNET_assert (CURLE_OK ==
                   curl_easy_setopt (eh,
                                     CURLOPT_HTTPHEADER,
                                     job_headers));
    w->job_headers = job_headers;
  }
  GNUNET_assert (CURLE_OK ==
                 curl_easy_setopt (eh,
                                   CURLOPT_MAXREDIRS,
                                   5));
  GNUNET_assert (CURLE_OK ==
                 curl_easy_setopt (eh,
                                   CURLOPT_FOLLOWLOCATION,
                                   1));

  w->job = GNUNET_CURL_job_add_raw (ctx,
                                    eh,
                                    job_headers,
                                    &handle_webhook_response,
                                    w);
  if (NULL == w->job)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to start the curl job for pending webhook #%llu\n",
                (unsigned long long) webhook_pending_serial);
    curl_slist_free_all (w->job_headers);
    GNUNET_free (w->body);
    GNUNET_CONTAINER_DLL_remove (w_head,
                                 w_tail,
                                 w);
    GNUNET_free (w);
    GNUNET_SCHEDULER_shutdown ();
    return;
  }
}


/**
 * Function called on events received from Postgres.
 *
 * @param cls closure, NULL
 * @param extra additional event data provided
 * @param extra_size number of bytes in @a extra
 */
static void
db_notify (void *cls,
           const void *extra,
           size_t extra_size)
{
  (void) cls;
  (void) extra;
  (void) extra_size;

  GNUNET_assert (NULL != task);
  GNUNET_SCHEDULER_cancel (task);
  task = GNUNET_SCHEDULER_add_now (&select_work,
                                   NULL);
}


/**
 * Typically called by `select_work`.
 *
 * @param cls a `json_t *` JSON array to build
 * @param webhook_pending_serial reference to the configured webhook template.
 * @param next_attempt is the time we should make the next request to the webhook.
 * @param retries how often have we tried this request to the webhook.
 * @param url to make request to
 * @param http_method use for the webhook
 * @param header of the webhook
 * @param body of the webhook
 */
static void
future_webhook_cb (void *cls,
                   uint64_t webhook_pending_serial,
                   struct GNUNET_TIME_Absolute next_attempt,
                   uint32_t retries,
                   const char *url,
                   const char *http_method,
                   const char *header,
                   const char *body)
{
  (void) webhook_pending_serial;
  (void) retries;
  (void) url;
  (void) http_method;
  (void) header;
  (void) body;

  task = GNUNET_SCHEDULER_add_at (next_attempt,
                                  &select_work,
                                  NULL);
}


static void
select_work (void *cls)
{
  enum GNUNET_DB_QueryStatus qs;
  struct GNUNET_TIME_Relative rel;

  (void) cls;
  task = NULL;
  db_plugin->preflight (db_plugin->cls);
  qs = db_plugin->lookup_pending_webhooks (db_plugin->cls,
                                           &pending_webhooks_cb,
                                           NULL);
  switch (qs)
  {
  case GNUNET_DB_STATUS_HARD_ERROR:
  case GNUNET_DB_STATUS_SOFT_ERROR:
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to lookup pending webhooks!\n");
    global_ret = EXIT_FAILURE;
    GNUNET_SCHEDULER_shutdown ();
    return;
  case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
    if (test_mode)
    {
      GNUNET_SCHEDULER_shutdown ();
      return;
    }
    qs = db_plugin->lookup_future_webhook (db_plugin->cls,
                                           &future_webhook_cb,
                                           NULL);
    switch (qs)
    {
    case GNUNET_DB_STATUS_HARD_ERROR:
    case GNUNET_DB_STATUS_SOFT_ERROR:
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  "Failed to lookup future webhook!\n");
      global_ret = EXIT_FAILURE;
      GNUNET_SCHEDULER_shutdown ();
      return;
    case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
      return;
    case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
      /* wait 5 min */
      /* Note: this should not even be necessary if all webhooks
         use the events properly... */
      rel = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5);
      task = GNUNET_SCHEDULER_add_delayed (rel,
                                           &select_work,
                                           NULL);
      return;
    }
  case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
  default:
    return; // wait for completion, then select more work.
  }
}


/**
 * First task.
 *
 * @param cls closure, NULL
 * @param args remaining command-line arguments
 * @param cfgfile name of the configuration file used (for saving, can be NULL!)
 * @param c configuration
 */
static void
run (void *cls,
     char *const *args,
     const char *cfgfile,
     const struct GNUNET_CONFIGURATION_Handle *c)
{
  (void) args;
  (void) cfgfile;

  cfg = c;
  GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
                                 NULL);
  ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule,
                          &rc);
  rc = GNUNET_CURL_gnunet_rc_create (ctx);
  if (NULL == ctx)
  {
    GNUNET_break (0);
    GNUNET_SCHEDULER_shutdown ();
    global_ret = EXIT_FAILURE;
    return;
  }
  if (NULL ==
      (db_plugin = TALER_MERCHANTDB_plugin_load (cfg)))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to initialize DB subsystem\n");
    GNUNET_SCHEDULER_shutdown ();
    global_ret = EXIT_NOTCONFIGURED;
    return;
  }
  if (GNUNET_OK !=
      db_plugin->connect (db_plugin->cls))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Failed to connect to database\n");
    GNUNET_SCHEDULER_shutdown ();
    global_ret = EXIT_FAILURE;
    return;
  }
  {
    struct GNUNET_DB_EventHeaderP es = {
      .size = htons (sizeof (es)),
      .type = htons (TALER_DBEVENT_MERCHANT_WEBHOOK_PENDING)
    };

    event_handler = db_plugin->event_listen (db_plugin->cls,
                                             &es,
                                             GNUNET_TIME_UNIT_FOREVER_REL,
                                             &db_notify,
                                             NULL);
  }
  GNUNET_assert (NULL == task);
  task = GNUNET_SCHEDULER_add_now (&select_work,
                                   NULL);
}


/**
 * The main function of the taler-merchant-webhook
 * @param argc number of arguments from the command line
 * @param argv command line arguments
 * @return 0 ok, 1 on error
 */
int
main (int argc,
      char *const *argv)
{
  struct GNUNET_GETOPT_CommandLineOption options[] = {
    GNUNET_GETOPT_option_flag ('t',
                               "test",
                               "run in test mode and exit when idle",
                               &test_mode),
    GNUNET_GETOPT_option_timetravel ('T',
                                     "timetravel"),
    GNUNET_GETOPT_option_version (VERSION "-" VCS_VERSION),
    GNUNET_GETOPT_OPTION_END
  };
  enum GNUNET_GenericReturnValue ret;

  if (GNUNET_OK !=
      GNUNET_STRINGS_get_utf8_args (argc, argv,
                                    &argc, &argv))
    return EXIT_INVALIDARGUMENT;
  TALER_OS_init ();
  ret = GNUNET_PROGRAM_run (
    argc, argv,
    "taler-merchant-webhook",
    gettext_noop (
      "background process that executes webhooks"),
    options,
    &run, NULL);
  GNUNET_free_nz ((void *) argv);
  if (GNUNET_SYSERR == ret)
    return EXIT_INVALIDARGUMENT;
  if (GNUNET_NO == ret)
    return EXIT_SUCCESS;
  return global_ret;
}


/* end of taler-merchant-webhook.c */