Line data Source code
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 26 : shutdown_task (void *cls)
124 : {
125 : struct WorkResponse *w;
126 :
127 : (void) cls;
128 26 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
129 : "Running shutdown\n");
130 26 : if (NULL != event_handler)
131 : {
132 26 : TALER_MERCHANTDB_event_listen_cancel (event_handler);
133 26 : event_handler = NULL;
134 : }
135 26 : if (NULL != task)
136 : {
137 16 : GNUNET_SCHEDULER_cancel (task);
138 16 : task = NULL;
139 : }
140 26 : while (NULL != (w = w_head))
141 : {
142 0 : GNUNET_CONTAINER_DLL_remove (w_head,
143 : w_tail,
144 : w);
145 0 : w_count--;
146 0 : GNUNET_CURL_job_cancel (w->job);
147 0 : curl_slist_free_all (w->job_headers);
148 0 : GNUNET_free (w->body);
149 0 : GNUNET_free (w);
150 : }
151 26 : if (NULL != pg)
152 : {
153 26 : TALER_MERCHANTDB_disconnect (pg);
154 26 : pg = NULL;
155 : }
156 26 : cfg = NULL;
157 26 : if (NULL != ctx)
158 : {
159 26 : GNUNET_CURL_fini (ctx);
160 26 : ctx = NULL;
161 : }
162 26 : if (NULL != rc)
163 : {
164 26 : GNUNET_CURL_gnunet_rc_destroy (rc);
165 26 : rc = NULL;
166 : }
167 26 : }
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 12 : handle_webhook_response (void *cls,
190 : long response_code,
191 : const void *body,
192 : size_t body_size)
193 : {
194 12 : struct WorkResponse *w = cls;
195 :
196 : (void) body;
197 : (void) body_size;
198 12 : w->job = NULL;
199 12 : GNUNET_CONTAINER_DLL_remove (w_head,
200 : w_tail,
201 : w);
202 12 : w_count--;
203 12 : GNUNET_free (w->body);
204 12 : curl_slist_free_all (w->job_headers);
205 12 : 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 10 : if (NULL != task)
213 0 : GNUNET_SCHEDULER_cancel (task);
214 10 : task = GNUNET_SCHEDULER_add_now (&select_work,
215 : NULL);
216 : }
217 12 : 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 12 : if (2 == response_code / 100) /* any 2xx http status code is OK! */
222 : {
223 : enum GNUNET_DB_QueryStatus qs;
224 :
225 12 : qs = TALER_MERCHANTDB_delete_pending_webhook (pg,
226 : w->webhook_pending_serial);
227 12 : GNUNET_free (w);
228 12 : switch (qs)
229 : {
230 0 : case GNUNET_DB_STATUS_HARD_ERROR:
231 : case GNUNET_DB_STATUS_SOFT_ERROR:
232 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
233 : "Failed to delete webhook, delete returned: %d\n",
234 : qs);
235 0 : global_ret = EXIT_FAILURE;
236 0 : GNUNET_SCHEDULER_shutdown ();
237 0 : return;
238 12 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
239 12 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
240 : "Delete returned: %d\n",
241 : qs);
242 12 : return;
243 0 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
244 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
245 : "Delete returned: %d\n",
246 : qs);
247 0 : return;
248 : }
249 0 : GNUNET_assert (0);
250 : }
251 :
252 : {
253 : struct GNUNET_TIME_Relative next_attempt;
254 : enum GNUNET_DB_QueryStatus qs;
255 0 : switch (response_code)
256 : {
257 0 : case MHD_HTTP_BAD_REQUEST:
258 0 : next_attempt = GNUNET_TIME_UNIT_FOREVER_REL; // never try again
259 0 : break;
260 0 : case MHD_HTTP_INTERNAL_SERVER_ERROR:
261 0 : next_attempt = GNUNET_TIME_UNIT_MINUTES;
262 0 : break;
263 0 : case MHD_HTTP_FORBIDDEN:
264 0 : next_attempt = GNUNET_TIME_UNIT_MINUTES;
265 0 : break;
266 0 : default:
267 0 : next_attempt = GNUNET_TIME_UNIT_HOURS;
268 0 : break;
269 : }
270 0 : qs = TALER_MERCHANTDB_update_pending_webhook (pg,
271 : w->webhook_pending_serial,
272 : GNUNET_TIME_relative_to_absolute (
273 : next_attempt));
274 0 : GNUNET_free (w);
275 0 : switch (qs)
276 : {
277 0 : case GNUNET_DB_STATUS_HARD_ERROR:
278 : case GNUNET_DB_STATUS_SOFT_ERROR:
279 0 : 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 0 : global_ret = EXIT_FAILURE;
285 0 : GNUNET_SCHEDULER_shutdown ();
286 0 : return;
287 0 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
288 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
289 : "Next in %s Rval: %d\n",
290 : GNUNET_TIME_relative2s (next_attempt, true),
291 : qs);
292 0 : return;
293 0 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
294 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
295 : "Next in %s Rval: %d\n",
296 : GNUNET_TIME_relative2s (next_attempt, true),
297 : qs);
298 0 : return;
299 : }
300 0 : 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 12 : 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 12 : struct WorkResponse *w = GNUNET_new (struct WorkResponse);
328 : CURL *eh;
329 12 : struct curl_slist *job_headers = NULL;
330 :
331 : (void) retries;
332 : (void) next_attempt;
333 : (void) cls;
334 12 : GNUNET_CONTAINER_DLL_insert (w_head,
335 : w_tail,
336 : w);
337 12 : w_count++;
338 12 : GNUNET_assert (w_count <= CONCURRENCY_LIMIT);
339 12 : w->webhook_pending_serial = webhook_pending_serial;
340 12 : eh = curl_easy_init ();
341 12 : GNUNET_assert (NULL != eh);
342 12 : GNUNET_assert (CURLE_OK ==
343 : curl_easy_setopt (eh,
344 : CURLOPT_CUSTOMREQUEST,
345 : http_method));
346 12 : GNUNET_assert (CURLE_OK ==
347 : curl_easy_setopt (eh,
348 : CURLOPT_URL,
349 : url));
350 12 : GNUNET_assert (CURLE_OK ==
351 : curl_easy_setopt (eh,
352 : CURLOPT_VERBOSE,
353 : 0L));
354 :
355 : /* conversion body data */
356 12 : if (NULL != body)
357 : {
358 12 : w->body = GNUNET_strdup (body);
359 12 : GNUNET_assert (CURLE_OK ==
360 : curl_easy_setopt (eh,
361 : CURLOPT_POSTFIELDS,
362 : w->body));
363 : }
364 : /* conversion header to job_headers data */
365 12 : if (NULL != header)
366 : {
367 6 : char *header_copy = GNUNET_strdup (header);
368 :
369 6 : for (const char *tok = strtok (header_copy, "\r\n");
370 12 : NULL != tok;
371 6 : tok = strtok (NULL, "\r\n"))
372 : {
373 : // extract all Key: value from 'header_copy'!
374 6 : job_headers = curl_slist_append (job_headers,
375 : tok);
376 : }
377 6 : GNUNET_free (header_copy);
378 6 : GNUNET_assert (CURLE_OK ==
379 : curl_easy_setopt (eh,
380 : CURLOPT_HTTPHEADER,
381 : job_headers));
382 6 : w->job_headers = job_headers;
383 : }
384 12 : GNUNET_assert (CURLE_OK ==
385 : curl_easy_setopt (eh,
386 : CURLOPT_MAXREDIRS,
387 : 5L));
388 12 : GNUNET_assert (CURLE_OK ==
389 : curl_easy_setopt (eh,
390 : CURLOPT_FOLLOWLOCATION,
391 : 1L));
392 12 : GNUNET_assert (CURLE_OK ==
393 : curl_easy_setopt (eh,
394 : CURLOPT_CONNECTTIMEOUT,
395 : WEBHOOK_CONNECT_TIMEOUT_SECONDS));
396 12 : GNUNET_assert (CURLE_OK ==
397 : curl_easy_setopt (eh,
398 : CURLOPT_TIMEOUT,
399 : WEBHOOK_TIMEOUT_SECONDS));
400 :
401 12 : w->job = GNUNET_CURL_job_add_raw (ctx,
402 : eh,
403 : job_headers,
404 : &handle_webhook_response,
405 : w);
406 12 : if (NULL == w->job)
407 : {
408 0 : 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 0 : curl_slist_free_all (w->job_headers);
412 0 : GNUNET_free (w->body);
413 0 : GNUNET_CONTAINER_DLL_remove (w_head,
414 : w_tail,
415 : w);
416 0 : w_count--;
417 0 : GNUNET_free (w);
418 0 : GNUNET_SCHEDULER_shutdown ();
419 0 : 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 7 : 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 7 : if (NULL != w_head)
441 0 : return; /* a batch is in flight; handle_webhook_response() will
442 : re-select once it drains */
443 7 : if (NULL != task)
444 7 : GNUNET_SCHEDULER_cancel (task);
445 7 : 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 0 : 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 0 : task = GNUNET_SCHEDULER_add_at (next_attempt,
480 : &select_work,
481 : NULL);
482 0 : }
483 :
484 :
485 : static void
486 43 : 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 43 : task = NULL;
494 43 : GNUNET_assert (w_count <= CONCURRENCY_LIMIT);
495 43 : limit = CONCURRENCY_LIMIT - w_count;
496 43 : if (0 == limit)
497 : {
498 : /* All slots busy; handle_webhook_response() will select
499 : more work once the batch completed. */
500 0 : GNUNET_break (0);
501 0 : return;
502 : }
503 43 : TALER_MERCHANTDB_preflight (pg);
504 43 : qs = TALER_MERCHANTDB_iterate_pending_webhooks (pg,
505 : limit,
506 : &pending_webhooks_cb,
507 : NULL);
508 43 : switch (qs)
509 : {
510 0 : case GNUNET_DB_STATUS_HARD_ERROR:
511 : case GNUNET_DB_STATUS_SOFT_ERROR:
512 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
513 : "Failed to lookup pending webhooks!\n");
514 0 : global_ret = EXIT_FAILURE;
515 0 : GNUNET_SCHEDULER_shutdown ();
516 0 : return;
517 33 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
518 33 : if (test_mode)
519 : {
520 10 : GNUNET_SCHEDULER_shutdown ();
521 10 : return;
522 : }
523 23 : qs = TALER_MERCHANTDB_iterate_pending_webhooks_next (pg,
524 : &future_webhook_cb,
525 : NULL);
526 23 : switch (qs)
527 : {
528 0 : case GNUNET_DB_STATUS_HARD_ERROR:
529 : case GNUNET_DB_STATUS_SOFT_ERROR:
530 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
531 : "Failed to lookup future webhook!\n");
532 0 : global_ret = EXIT_FAILURE;
533 0 : GNUNET_SCHEDULER_shutdown ();
534 0 : return;
535 0 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
536 0 : return;
537 23 : 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 23 : rel = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5);
542 23 : task = GNUNET_SCHEDULER_add_delayed (rel,
543 : &select_work,
544 : NULL);
545 23 : return;
546 : }
547 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
548 : default:
549 10 : 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 26 : 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 26 : cfg = c;
572 26 : GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
573 : NULL);
574 26 : ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule,
575 : &rc);
576 26 : if (NULL == ctx)
577 : {
578 0 : GNUNET_break (0);
579 0 : GNUNET_SCHEDULER_shutdown ();
580 0 : global_ret = EXIT_FAILURE;
581 0 : return;
582 : }
583 26 : rc = GNUNET_CURL_gnunet_rc_create (ctx);
584 26 : if (NULL ==
585 26 : (pg = TALER_MERCHANTDB_connect (cfg)))
586 : {
587 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
588 : "Failed to initialize DB subsystem. Consider running taler-merchant-dbconfig!\n");
589 0 : GNUNET_SCHEDULER_shutdown ();
590 0 : global_ret = EXIT_FAILURE;
591 0 : return;
592 : }
593 : {
594 26 : struct GNUNET_DB_EventHeaderP es = {
595 26 : .size = htons (sizeof (es)),
596 26 : .type = htons (TALER_DBEVENT_MERCHANT_WEBHOOK_PENDING)
597 : };
598 :
599 52 : event_handler = TALER_MERCHANTDB_event_listen (pg,
600 : &es,
601 26 : GNUNET_TIME_UNIT_FOREVER_REL,
602 : &db_notify,
603 : NULL);
604 : }
605 26 : GNUNET_assert (NULL == task);
606 26 : 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 26 : main (int argc,
619 : char *const *argv)
620 : {
621 26 : struct GNUNET_GETOPT_CommandLineOption options[] = {
622 26 : GNUNET_GETOPT_option_flag ('t',
623 : "test",
624 : "run in test mode and exit when idle",
625 : &test_mode),
626 26 : GNUNET_GETOPT_option_timetravel ('T',
627 : "timetravel"),
628 26 : GNUNET_GETOPT_option_version (VERSION),
629 : GNUNET_GETOPT_OPTION_END
630 : };
631 : enum GNUNET_GenericReturnValue ret;
632 :
633 26 : 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 26 : if (GNUNET_SYSERR == ret)
642 0 : return EXIT_NOTCONFIGURED;
643 26 : if (GNUNET_NO == ret)
644 0 : return EXIT_SUCCESS;
645 26 : return global_ret;
646 : }
647 :
648 :
649 : /* end of taler-merchant-webhook.c */
|
