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