taler-merchant-webhook.c (19033B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2023 Taler Systems SA 4 5 TALER is free software; you can redistribute it and/or modify it under the 6 terms of the GNU Affero General Public License as published by the Free Software 7 Foundation; either version 3, or (at your option) any later version. 8 9 TALER is distributed in the hope that it will be useful, but WITHOUT ANY 10 WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 11 A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. 12 13 You should have received a copy of the GNU Affero General Public License along with 14 TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> 15 */ 16 /** 17 * @file src/backend/taler-merchant-webhook.c 18 * @brief Process that runs webhooks triggered by the merchant backend 19 * @author Priscilla HUANG 20 */ 21 #include "platform.h" 22 #include "microhttpd.h" 23 #include <gnunet/gnunet_util_lib.h> 24 #include <jansson.h> 25 #include <pthread.h> 26 #include "taler/taler_merchant_util.h" 27 #include "merchantdb_lib.h" 28 #include "merchantdb_lib.h" 29 #include <taler/taler_dbevents.h> 30 #include "merchant-database/delete_pending_webhook.h" 31 #include "merchant-database/iterate_pending_webhooks.h" 32 #include "merchant-database/update_pending_webhook.h" 33 #include "merchant-database/event_listen.h" 34 #include "merchant-database/preflight.h" 35 36 37 /** 38 * Maximum number of webhooks we execute concurrently. 39 */ 40 #define CONCURRENCY_LIMIT 32 41 42 /** 43 * How long (in seconds) may a single webhook request take before we give 44 * up on it? Without a limit a single unresponsive webhook target would 45 * occupy one of the #CONCURRENCY_LIMIT slots forever and (as we only 46 * SELECT() again once *all* requests of a batch completed) stall webhook 47 * processing entirely. 48 */ 49 #define WEBHOOK_TIMEOUT_SECONDS 60L 50 51 /** 52 * How long (in seconds) may establishing the TCP/TLS connection take? 53 */ 54 #define WEBHOOK_CONNECT_TIMEOUT_SECONDS 15L 55 56 57 struct WorkResponse 58 { 59 struct WorkResponse *next; 60 struct WorkResponse *prev; 61 struct GNUNET_CURL_Job *job; 62 uint64_t webhook_pending_serial; 63 char *body; 64 struct curl_slist *job_headers; 65 }; 66 67 68 static struct WorkResponse *w_head; 69 70 static struct WorkResponse *w_tail; 71 72 /** 73 * Number of entries in the @e w_head DLL, that is the number 74 * of webhooks currently in flight. Never exceeds 75 * #CONCURRENCY_LIMIT. 76 */ 77 static uint64_t w_count; 78 79 static struct GNUNET_DB_EventHandler *event_handler; 80 81 /** 82 * The merchant's configuration. 83 */ 84 static const struct GNUNET_CONFIGURATION_Handle *cfg; 85 86 /** 87 * Our database connection. 88 */ 89 static struct TALER_MERCHANTDB_PostgresContext *pg; 90 91 /** 92 * Next task to run, if any. 93 */ 94 static struct GNUNET_SCHEDULER_Task *task; 95 96 /** 97 * Handle to the context for interacting with the bank / wire gateway. 98 */ 99 static struct GNUNET_CURL_Context *ctx; 100 101 /** 102 * Scheduler context for running the @e ctx. 103 */ 104 static struct GNUNET_CURL_RescheduleContext *rc; 105 106 /** 107 * Value to return from main(). 0 on success, non-zero on errors. 108 */ 109 static int global_ret; 110 111 /** 112 * #GNUNET_YES if we are in test mode and should exit when idle. 113 */ 114 static int test_mode; 115 116 117 /** 118 * We're being aborted with CTRL-C (or SIGTERM). Shut down. 119 * 120 * @param cls closure 121 */ 122 static void 123 shutdown_task (void *cls) 124 { 125 struct WorkResponse *w; 126 127 (void) cls; 128 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 129 "Running shutdown\n"); 130 if (NULL != event_handler) 131 { 132 TALER_MERCHANTDB_event_listen_cancel (event_handler); 133 event_handler = NULL; 134 } 135 if (NULL != task) 136 { 137 GNUNET_SCHEDULER_cancel (task); 138 task = NULL; 139 } 140 while (NULL != (w = w_head)) 141 { 142 GNUNET_CONTAINER_DLL_remove (w_head, 143 w_tail, 144 w); 145 w_count--; 146 GNUNET_CURL_job_cancel (w->job); 147 curl_slist_free_all (w->job_headers); 148 GNUNET_free (w->body); 149 GNUNET_free (w); 150 } 151 if (NULL != pg) 152 { 153 TALER_MERCHANTDB_disconnect (pg); 154 pg = NULL; 155 } 156 cfg = NULL; 157 if (NULL != ctx) 158 { 159 GNUNET_CURL_fini (ctx); 160 ctx = NULL; 161 } 162 if (NULL != rc) 163 { 164 GNUNET_CURL_gnunet_rc_destroy (rc); 165 rc = NULL; 166 } 167 } 168 169 170 /** 171 * Select webhook to process. 172 * 173 * @param cls NULL 174 */ 175 static void 176 select_work (void *cls); 177 178 179 /** 180 * This function is used by the function `pending_webhooks_cb`. According to the response code, 181 * we delete or update the webhook. 182 * 183 * @param cls closure 184 * @param response_code HTTP response code from server, 0 on hard error 185 * @param body http body of the response 186 * @param body_size number of bytes in @a body 187 */ 188 static void 189 handle_webhook_response (void *cls, 190 long response_code, 191 const void *body, 192 size_t body_size) 193 { 194 struct WorkResponse *w = cls; 195 196 (void) body; 197 (void) body_size; 198 w->job = NULL; 199 GNUNET_CONTAINER_DLL_remove (w_head, 200 w_tail, 201 w); 202 w_count--; 203 GNUNET_free (w->body); 204 curl_slist_free_all (w->job_headers); 205 if (0 == w_count) 206 { 207 /* We only SELECT() again after having finished all requests of 208 the current batch: the rows we are working on are only updated 209 (or deleted) once their request completed, so selecting earlier 210 would simply return the very same webhooks again and run them 211 a second time. */ 212 if (NULL != task) 213 GNUNET_SCHEDULER_cancel (task); 214 task = GNUNET_SCHEDULER_add_now (&select_work, 215 NULL); 216 } 217 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 218 "Webhook %llu returned with status %ld\n", 219 (unsigned long long) w->webhook_pending_serial, 220 response_code); 221 if (2 == response_code / 100) /* any 2xx http status code is OK! */ 222 { 223 enum GNUNET_DB_QueryStatus qs; 224 225 qs = TALER_MERCHANTDB_delete_pending_webhook (pg, 226 w->webhook_pending_serial); 227 GNUNET_free (w); 228 switch (qs) 229 { 230 case GNUNET_DB_STATUS_HARD_ERROR: 231 case GNUNET_DB_STATUS_SOFT_ERROR: 232 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 233 "Failed to delete webhook, delete returned: %d\n", 234 qs); 235 global_ret = EXIT_FAILURE; 236 GNUNET_SCHEDULER_shutdown (); 237 return; 238 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 239 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 240 "Delete returned: %d\n", 241 qs); 242 return; 243 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 244 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 245 "Delete returned: %d\n", 246 qs); 247 return; 248 } 249 GNUNET_assert (0); 250 } 251 252 { 253 struct GNUNET_TIME_Relative next_attempt; 254 enum GNUNET_DB_QueryStatus qs; 255 switch (response_code) 256 { 257 case MHD_HTTP_BAD_REQUEST: 258 next_attempt = GNUNET_TIME_UNIT_FOREVER_REL; // never try again 259 break; 260 case MHD_HTTP_INTERNAL_SERVER_ERROR: 261 next_attempt = GNUNET_TIME_UNIT_MINUTES; 262 break; 263 case MHD_HTTP_FORBIDDEN: 264 next_attempt = GNUNET_TIME_UNIT_MINUTES; 265 break; 266 default: 267 next_attempt = GNUNET_TIME_UNIT_HOURS; 268 break; 269 } 270 qs = TALER_MERCHANTDB_update_pending_webhook (pg, 271 w->webhook_pending_serial, 272 GNUNET_TIME_relative_to_absolute ( 273 next_attempt)); 274 GNUNET_free (w); 275 switch (qs) 276 { 277 case GNUNET_DB_STATUS_HARD_ERROR: 278 case GNUNET_DB_STATUS_SOFT_ERROR: 279 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 280 "Failed to update pending webhook to next in %s Rval: %d\n", 281 GNUNET_TIME_relative2s (next_attempt, 282 true), 283 qs); 284 global_ret = EXIT_FAILURE; 285 GNUNET_SCHEDULER_shutdown (); 286 return; 287 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 288 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 289 "Next in %s Rval: %d\n", 290 GNUNET_TIME_relative2s (next_attempt, true), 291 qs); 292 return; 293 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 294 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 295 "Next in %s Rval: %d\n", 296 GNUNET_TIME_relative2s (next_attempt, true), 297 qs); 298 return; 299 } 300 GNUNET_assert (0); 301 } 302 } 303 304 305 /** 306 * Typically called by `select_work`. 307 * 308 * @param cls a `json_t *` JSON array to build 309 * @param webhook_pending_serial reference to the configured webhook template. 310 * @param next_attempt is the time we should make the next request to the webhook. 311 * @param retries how often have we tried this request to the webhook. 312 * @param url to make request to 313 * @param http_method use for the webhook 314 * @param header of the webhook 315 * @param body of the webhook 316 */ 317 static void 318 pending_webhooks_cb (void *cls, 319 uint64_t webhook_pending_serial, 320 struct GNUNET_TIME_Absolute next_attempt, 321 uint32_t retries, 322 const char *url, 323 const char *http_method, 324 const char *header, 325 const char *body) 326 { 327 struct WorkResponse *w = GNUNET_new (struct WorkResponse); 328 CURL *eh; 329 struct curl_slist *job_headers = NULL; 330 331 (void) retries; 332 (void) next_attempt; 333 (void) cls; 334 GNUNET_CONTAINER_DLL_insert (w_head, 335 w_tail, 336 w); 337 w_count++; 338 GNUNET_assert (w_count <= CONCURRENCY_LIMIT); 339 w->webhook_pending_serial = webhook_pending_serial; 340 eh = curl_easy_init (); 341 GNUNET_assert (NULL != eh); 342 GNUNET_assert (CURLE_OK == 343 curl_easy_setopt (eh, 344 CURLOPT_CUSTOMREQUEST, 345 http_method)); 346 GNUNET_assert (CURLE_OK == 347 curl_easy_setopt (eh, 348 CURLOPT_URL, 349 url)); 350 GNUNET_assert (CURLE_OK == 351 curl_easy_setopt (eh, 352 CURLOPT_VERBOSE, 353 0L)); 354 355 /* conversion body data */ 356 if (NULL != body) 357 { 358 w->body = GNUNET_strdup (body); 359 GNUNET_assert (CURLE_OK == 360 curl_easy_setopt (eh, 361 CURLOPT_POSTFIELDS, 362 w->body)); 363 } 364 /* conversion header to job_headers data */ 365 if (NULL != header) 366 { 367 char *header_copy = GNUNET_strdup (header); 368 369 for (const char *tok = strtok (header_copy, "\r\n"); 370 NULL != tok; 371 tok = strtok (NULL, "\r\n")) 372 { 373 // extract all Key: value from 'header_copy'! 374 job_headers = curl_slist_append (job_headers, 375 tok); 376 } 377 GNUNET_free (header_copy); 378 GNUNET_assert (CURLE_OK == 379 curl_easy_setopt (eh, 380 CURLOPT_HTTPHEADER, 381 job_headers)); 382 w->job_headers = job_headers; 383 } 384 GNUNET_assert (CURLE_OK == 385 curl_easy_setopt (eh, 386 CURLOPT_MAXREDIRS, 387 5L)); 388 GNUNET_assert (CURLE_OK == 389 curl_easy_setopt (eh, 390 CURLOPT_FOLLOWLOCATION, 391 1L)); 392 GNUNET_assert (CURLE_OK == 393 curl_easy_setopt (eh, 394 CURLOPT_CONNECTTIMEOUT, 395 WEBHOOK_CONNECT_TIMEOUT_SECONDS)); 396 GNUNET_assert (CURLE_OK == 397 curl_easy_setopt (eh, 398 CURLOPT_TIMEOUT, 399 WEBHOOK_TIMEOUT_SECONDS)); 400 401 w->job = GNUNET_CURL_job_add_raw (ctx, 402 eh, 403 job_headers, 404 &handle_webhook_response, 405 w); 406 if (NULL == w->job) 407 { 408 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 409 "Failed to start the curl job for pending webhook #%llu\n", 410 (unsigned long long) webhook_pending_serial); 411 curl_slist_free_all (w->job_headers); 412 GNUNET_free (w->body); 413 GNUNET_CONTAINER_DLL_remove (w_head, 414 w_tail, 415 w); 416 w_count--; 417 GNUNET_free (w); 418 GNUNET_SCHEDULER_shutdown (); 419 return; 420 } 421 } 422 423 424 /** 425 * Function called on events received from Postgres. 426 * 427 * @param cls closure, NULL 428 * @param extra additional event data provided 429 * @param extra_size number of bytes in @a extra 430 */ 431 static void 432 db_notify (void *cls, 433 const void *extra, 434 size_t extra_size) 435 { 436 (void) cls; 437 (void) extra; 438 (void) extra_size; 439 440 if (NULL != w_head) 441 return; /* a batch is in flight; handle_webhook_response() will 442 re-select once it drains */ 443 if (NULL != task) 444 GNUNET_SCHEDULER_cancel (task); 445 task = GNUNET_SCHEDULER_add_now (&select_work, 446 NULL); 447 } 448 449 450 /** 451 * Typically called by `select_work`. 452 * 453 * @param cls a `json_t *` JSON array to build 454 * @param webhook_pending_serial reference to the configured webhook template. 455 * @param next_attempt is the time we should make the next request to the webhook. 456 * @param retries how often have we tried this request to the webhook. 457 * @param url to make request to 458 * @param http_method use for the webhook 459 * @param header of the webhook 460 * @param body of the webhook 461 */ 462 static void 463 future_webhook_cb (void *cls, 464 uint64_t webhook_pending_serial, 465 struct GNUNET_TIME_Absolute next_attempt, 466 uint32_t retries, 467 const char *url, 468 const char *http_method, 469 const char *header, 470 const char *body) 471 { 472 (void) webhook_pending_serial; 473 (void) retries; 474 (void) url; 475 (void) http_method; 476 (void) header; 477 (void) body; 478 479 task = GNUNET_SCHEDULER_add_at (next_attempt, 480 &select_work, 481 NULL); 482 } 483 484 485 static void 486 select_work (void *cls) 487 { 488 enum GNUNET_DB_QueryStatus qs; 489 struct GNUNET_TIME_Relative rel; 490 uint64_t limit; 491 492 (void) cls; 493 task = NULL; 494 GNUNET_assert (w_count <= CONCURRENCY_LIMIT); 495 limit = CONCURRENCY_LIMIT - w_count; 496 if (0 == limit) 497 { 498 /* All slots busy; handle_webhook_response() will select 499 more work once the batch completed. */ 500 GNUNET_break (0); 501 return; 502 } 503 TALER_MERCHANTDB_preflight (pg); 504 qs = TALER_MERCHANTDB_iterate_pending_webhooks (pg, 505 limit, 506 &pending_webhooks_cb, 507 NULL); 508 switch (qs) 509 { 510 case GNUNET_DB_STATUS_HARD_ERROR: 511 case GNUNET_DB_STATUS_SOFT_ERROR: 512 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 513 "Failed to lookup pending webhooks!\n"); 514 global_ret = EXIT_FAILURE; 515 GNUNET_SCHEDULER_shutdown (); 516 return; 517 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 518 if (test_mode) 519 { 520 GNUNET_SCHEDULER_shutdown (); 521 return; 522 } 523 qs = TALER_MERCHANTDB_iterate_pending_webhooks_next (pg, 524 &future_webhook_cb, 525 NULL); 526 switch (qs) 527 { 528 case GNUNET_DB_STATUS_HARD_ERROR: 529 case GNUNET_DB_STATUS_SOFT_ERROR: 530 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 531 "Failed to lookup future webhook!\n"); 532 global_ret = EXIT_FAILURE; 533 GNUNET_SCHEDULER_shutdown (); 534 return; 535 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 536 return; 537 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 538 /* wait 5 min */ 539 /* Note: this should not even be necessary if all webhooks 540 use the events properly... */ 541 rel = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5); 542 task = GNUNET_SCHEDULER_add_delayed (rel, 543 &select_work, 544 NULL); 545 return; 546 } 547 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 548 default: 549 return; // wait for completion, then select more work. 550 } 551 } 552 553 554 /** 555 * First task. 556 * 557 * @param cls closure, NULL 558 * @param args remaining command-line arguments 559 * @param cfgfile name of the configuration file used (for saving, can be NULL!) 560 * @param c configuration 561 */ 562 static void 563 run (void *cls, 564 char *const *args, 565 const char *cfgfile, 566 const struct GNUNET_CONFIGURATION_Handle *c) 567 { 568 (void) args; 569 (void) cfgfile; 570 571 cfg = c; 572 GNUNET_SCHEDULER_add_shutdown (&shutdown_task, 573 NULL); 574 ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule, 575 &rc); 576 if (NULL == ctx) 577 { 578 GNUNET_break (0); 579 GNUNET_SCHEDULER_shutdown (); 580 global_ret = EXIT_FAILURE; 581 return; 582 } 583 rc = GNUNET_CURL_gnunet_rc_create (ctx); 584 if (NULL == 585 (pg = TALER_MERCHANTDB_connect (cfg))) 586 { 587 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 588 "Failed to initialize DB subsystem. Consider running taler-merchant-dbconfig!\n"); 589 GNUNET_SCHEDULER_shutdown (); 590 global_ret = EXIT_FAILURE; 591 return; 592 } 593 { 594 struct GNUNET_DB_EventHeaderP es = { 595 .size = htons (sizeof (es)), 596 .type = htons (TALER_DBEVENT_MERCHANT_WEBHOOK_PENDING) 597 }; 598 599 event_handler = TALER_MERCHANTDB_event_listen (pg, 600 &es, 601 GNUNET_TIME_UNIT_FOREVER_REL, 602 &db_notify, 603 NULL); 604 } 605 GNUNET_assert (NULL == task); 606 task = GNUNET_SCHEDULER_add_now (&select_work, 607 NULL); 608 } 609 610 611 /** 612 * The main function of the taler-merchant-webhook 613 * @param argc number of arguments from the command line 614 * @param argv command line arguments 615 * @return 0 ok, 1 on error 616 */ 617 int 618 main (int argc, 619 char *const *argv) 620 { 621 struct GNUNET_GETOPT_CommandLineOption options[] = { 622 GNUNET_GETOPT_option_flag ('t', 623 "test", 624 "run in test mode and exit when idle", 625 &test_mode), 626 GNUNET_GETOPT_option_timetravel ('T', 627 "timetravel"), 628 GNUNET_GETOPT_option_version (VERSION), 629 GNUNET_GETOPT_OPTION_END 630 }; 631 enum GNUNET_GenericReturnValue ret; 632 633 ret = GNUNET_PROGRAM_run ( 634 TALER_MERCHANT_project_data (), 635 argc, argv, 636 "taler-merchant-webhook", 637 gettext_noop ( 638 "background process that executes webhooks"), 639 options, 640 &run, NULL); 641 if (GNUNET_SYSERR == ret) 642 return EXIT_INVALIDARGUMENT; 643 if (GNUNET_NO == ret) 644 return EXIT_SUCCESS; 645 return global_ret; 646 } 647 648 649 /* end of taler-merchant-webhook.c */