Line data Source code
1 : /*
2 : This file is part of TALER
3 : Copyright (C) 2022--2026 Taler Systems SA
4 :
5 : TALER is free software; you can redistribute it and/or modify it under the
6 : terms of the GNU 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 General Public License for more details.
12 :
13 : You should have received a copy of the GNU General Public License along with
14 : TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
15 : */
16 : /**
17 : * @file exchangedb/do_import_credits.c
18 : * @brief Implementation of the do_import_credits function for Postgres
19 : * @author Christian Grothoff
20 : * @author Joseph Xu
21 : */
22 : #include "taler/taler_pq_lib.h"
23 : #include "exchange-database/do_import_credits.h"
24 : #include "helper.h"
25 :
26 :
27 : /**
28 : * Compute the notification channel for a reserve that was credited.
29 : *
30 : * @param reserve_pub reserve to notify on
31 : * @return string to pass to postgres for the notification
32 : */
33 : static char *
34 78 : compute_notify_on_reserve (const struct TALER_ReservePublicKeyP *reserve_pub)
35 : {
36 78 : struct TALER_EXCHANGEDB_ReserveEventP rep = {
37 78 : .header.size = htons (sizeof (rep)),
38 78 : .header.type = htons (TALER_DBEVENT_EXCHANGE_RESERVE_INCOMING),
39 : .reserve_pub = *reserve_pub
40 : };
41 :
42 78 : return GNUNET_PQ_get_event_notify_channel (&rep.header);
43 : }
44 :
45 :
46 : /**
47 : * Compute the notification channel for an account that authenticated itself.
48 : *
49 : * @param h_payto normalized hash of the account that was authenticated
50 : * @return string to pass to postgres for the notification
51 : */
52 : static char *
53 22 : compute_notify_on_kycauth (const struct TALER_NormalizedPaytoHashP *h_payto)
54 : {
55 22 : struct TALER_EXCHANGEDB_KycCompletedEventP rep = {
56 22 : .header.size = htons (sizeof (rep)),
57 22 : .header.type = htons (TALER_DBEVENT_EXCHANGE_KYC_COMPLETED),
58 : .h_payto = *h_payto
59 : };
60 :
61 22 : return GNUNET_PQ_get_event_notify_channel (&rep.header);
62 : }
63 :
64 :
65 : /**
66 : * Closure for #helper_cb().
67 : */
68 : struct Context
69 : {
70 : /**
71 : * Array with entries set to 'true' for transfers we had already imported.
72 : */
73 : bool *transaction_duplicates;
74 :
75 : /**
76 : * Length of the @e transaction_duplicates array.
77 : */
78 : unsigned int length;
79 :
80 : /**
81 : * Set to #GNUNET_SYSERR on failures.
82 : */
83 : enum GNUNET_GenericReturnValue status;
84 : };
85 :
86 :
87 : /**
88 : * Function called with the rows the stored procedure returned, one per
89 : * reserve transfer, in input order.
90 : *
91 : * @param cls closure of type `struct Context *`
92 : * @param result the postgres result
93 : * @param num_results the number of results in @a result
94 : */
95 : static void
96 141 : helper_cb (void *cls,
97 : PGresult *result,
98 : unsigned int num_results)
99 : {
100 141 : struct Context *ctx = cls;
101 :
102 141 : if (num_results > ctx->length)
103 : {
104 : /* The procedure returns one row per reserve transfer we passed in; a
105 : different count means the two have drifted apart, and writing past the
106 : caller's array is not the way to find that out. */
107 0 : GNUNET_break (0);
108 0 : ctx->status = GNUNET_SYSERR;
109 0 : return;
110 : }
111 219 : for (unsigned int i = 0; i<num_results; i++)
112 : {
113 78 : struct GNUNET_PQ_ResultSpec rs[] = {
114 78 : GNUNET_PQ_result_spec_bool ("out_duplicate",
115 78 : &ctx->transaction_duplicates[i]),
116 : GNUNET_PQ_result_spec_end
117 : };
118 :
119 78 : if (GNUNET_OK !=
120 78 : GNUNET_PQ_extract_result (result,
121 : rs,
122 : i))
123 : {
124 0 : GNUNET_break (0);
125 0 : ctx->status = GNUNET_SYSERR;
126 0 : return;
127 : }
128 : }
129 : }
130 :
131 :
132 : enum GNUNET_DB_QueryStatus
133 141 : TALER_EXCHANGEDB_do_import_credits (
134 : struct TALER_EXCHANGEDB_PostgresContext *pg,
135 : const struct TALER_EXCHANGEDB_CreditBatch *batch,
136 : enum GNUNET_DB_QueryStatus *reserve_results)
137 141 : {
138 141 : unsigned int rlen = batch->reserves_length;
139 141 : unsigned int klen = batch->kycauths_length;
140 141 : unsigned int wlen = batch->wads_length;
141 141 : unsigned int dups = 0;
142 :
143 : /* reserve transfers */
144 141 : struct TALER_ReservePublicKeyP reserve_pubs[GNUNET_NZL (rlen)];
145 141 : uint64_t wire_references[GNUNET_NZL (rlen)];
146 141 : struct TALER_Amount balances[GNUNET_NZL (rlen)];
147 141 : struct GNUNET_TIME_Timestamp execution_times[GNUNET_NZL (rlen)];
148 141 : struct TALER_FullPaytoHashP h_full_paytos[GNUNET_NZL (rlen)];
149 141 : struct TALER_NormalizedPaytoHashP h_normalized_paytos[GNUNET_NZL (rlen)];
150 141 : const char *payto_uris[GNUNET_NZL (rlen)];
151 141 : char *notify_s[GNUNET_NZL (rlen)];
152 141 : bool transaction_duplicates[GNUNET_NZL (rlen)];
153 :
154 : /* KYC authentication transfers */
155 141 : union TALER_AccountPublicKeyP ka_account_pubs[GNUNET_NZL (klen)];
156 141 : uint64_t ka_wire_references[GNUNET_NZL (klen)];
157 141 : struct TALER_Amount ka_balances[GNUNET_NZL (klen)];
158 141 : struct GNUNET_TIME_Timestamp ka_execution_times[GNUNET_NZL (klen)];
159 141 : struct TALER_FullPaytoHashP ka_h_full_paytos[GNUNET_NZL (klen)];
160 141 : struct TALER_NormalizedPaytoHashP ka_h_normalized_paytos[GNUNET_NZL (klen)];
161 141 : const char *ka_payto_uris[GNUNET_NZL (klen)];
162 141 : char *ka_notify_s[GNUNET_NZL (klen)];
163 :
164 : /* WAD transfers */
165 141 : struct TALER_WadIdentifierP wad_ids[GNUNET_NZL (wlen)];
166 141 : const char *wad_origin_exchange_urls[GNUNET_NZL (wlen)];
167 141 : struct TALER_Amount wad_balances[GNUNET_NZL (wlen)];
168 141 : struct GNUNET_TIME_Timestamp wad_execution_times[GNUNET_NZL (wlen)];
169 :
170 : struct GNUNET_TIME_Timestamp reserve_expiration
171 141 : = GNUNET_TIME_relative_to_timestamp (pg->idle_reserve_expiration_time);
172 : struct GNUNET_TIME_Timestamp gc
173 141 : = GNUNET_TIME_relative_to_timestamp (pg->legal_reserve_expiration_time);
174 : struct GNUNET_TIME_Absolute lease_until
175 141 : = GNUNET_TIME_relative_to_absolute (batch->lease);
176 : enum GNUNET_DB_QueryStatus qs;
177 :
178 219 : for (unsigned int i = 0; i<rlen; i++)
179 : {
180 78 : const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &batch->reserves[i];
181 :
182 78 : TALER_full_payto_hash (reserve->sender_account_details,
183 : &h_full_paytos[i]);
184 78 : TALER_full_payto_normalize_and_hash (reserve->sender_account_details,
185 : &h_normalized_paytos[i]);
186 78 : notify_s[i] = compute_notify_on_reserve (reserve->reserve_pub);
187 78 : reserve_pubs[i] = *reserve->reserve_pub;
188 78 : balances[i] = *reserve->balance;
189 78 : execution_times[i] = reserve->execution_time;
190 78 : payto_uris[i] = reserve->sender_account_details.full_payto;
191 78 : wire_references[i] = reserve->wire_reference;
192 : }
193 163 : for (unsigned int i = 0; i<klen; i++)
194 : {
195 22 : const struct TALER_EXCHANGEDB_KycauthInInfo *ka = &batch->kycauths[i];
196 :
197 22 : TALER_full_payto_hash (ka->sender_account_details,
198 : &ka_h_full_paytos[i]);
199 22 : TALER_full_payto_normalize_and_hash (ka->sender_account_details,
200 : &ka_h_normalized_paytos[i]);
201 22 : ka_notify_s[i] = compute_notify_on_kycauth (&ka_h_normalized_paytos[i]);
202 22 : ka_account_pubs[i] = *ka->account_pub;
203 22 : ka_balances[i] = *ka->balance;
204 22 : ka_execution_times[i] = ka->execution_time;
205 22 : ka_payto_uris[i] = ka->sender_account_details.full_payto;
206 22 : ka_wire_references[i] = ka->wire_reference;
207 : }
208 141 : for (unsigned int i = 0; i<wlen; i++)
209 : {
210 0 : const struct TALER_EXCHANGEDB_WadInInfo *wad = &batch->wads[i];
211 :
212 0 : wad_ids[i] = *wad->wad_id;
213 0 : wad_origin_exchange_urls[i] = wad->origin_exchange_url;
214 0 : wad_balances[i] = *wad->balance;
215 0 : wad_execution_times[i] = wad->execution_time;
216 : }
217 :
218 141 : PREPARE (pg,
219 : "do_import_credits",
220 : "SELECT"
221 : " out_duplicate"
222 : " FROM exchange_do_import_credits"
223 : " ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17"
224 : " ,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28);");
225 : {
226 141 : struct GNUNET_PQ_QueryParam params[] = {
227 141 : GNUNET_PQ_query_param_string (batch->exchange_account_name),
228 141 : GNUNET_PQ_query_param_timestamp (&gc),
229 141 : GNUNET_PQ_query_param_timestamp (&reserve_expiration),
230 : /* reserve transfers */
231 141 : GNUNET_PQ_query_param_array_auto_from_type (rlen,
232 : reserve_pubs,
233 : pg->conn),
234 141 : GNUNET_PQ_query_param_array_uint64 (rlen,
235 : wire_references,
236 : pg->conn),
237 141 : TALER_PQ_query_param_array_amount (rlen,
238 : balances,
239 : pg->conn),
240 141 : GNUNET_PQ_query_param_array_timestamp (rlen,
241 : execution_times,
242 : pg->conn),
243 141 : GNUNET_PQ_query_param_array_auto_from_type (rlen,
244 : h_full_paytos,
245 : pg->conn),
246 141 : GNUNET_PQ_query_param_array_auto_from_type (rlen,
247 : h_normalized_paytos,
248 : pg->conn),
249 141 : GNUNET_PQ_query_param_array_ptrs_string (rlen,
250 : payto_uris,
251 : pg->conn),
252 141 : GNUNET_PQ_query_param_array_ptrs_string (rlen,
253 : (const char **) notify_s,
254 : pg->conn),
255 : /* KYC authentication transfers */
256 141 : GNUNET_PQ_query_param_array_auto_from_type (klen,
257 : ka_account_pubs,
258 : pg->conn),
259 141 : GNUNET_PQ_query_param_array_uint64 (klen,
260 : ka_wire_references,
261 : pg->conn),
262 141 : TALER_PQ_query_param_array_amount (klen,
263 : ka_balances,
264 : pg->conn),
265 141 : GNUNET_PQ_query_param_array_timestamp (klen,
266 : ka_execution_times,
267 : pg->conn),
268 141 : GNUNET_PQ_query_param_array_auto_from_type (klen,
269 : ka_h_full_paytos,
270 : pg->conn),
271 141 : GNUNET_PQ_query_param_array_auto_from_type (klen,
272 : ka_h_normalized_paytos,
273 : pg->conn),
274 141 : GNUNET_PQ_query_param_array_ptrs_string (klen,
275 : ka_payto_uris,
276 : pg->conn),
277 141 : GNUNET_PQ_query_param_array_ptrs_string (klen,
278 : (const char **) ka_notify_s,
279 : pg->conn),
280 : /* WAD transfers */
281 141 : GNUNET_PQ_query_param_array_auto_from_type (wlen,
282 : wad_ids,
283 : pg->conn),
284 141 : GNUNET_PQ_query_param_array_ptrs_string (wlen,
285 : wad_origin_exchange_urls,
286 : pg->conn),
287 141 : TALER_PQ_query_param_array_amount (wlen,
288 : wad_balances,
289 : pg->conn),
290 141 : GNUNET_PQ_query_param_array_timestamp (wlen,
291 : wad_execution_times,
292 : pg->conn),
293 : /* shard bookkeeping */
294 141 : GNUNET_PQ_query_param_string (batch->job_name),
295 141 : GNUNET_PQ_query_param_uint64 (&batch->shard_start),
296 141 : GNUNET_PQ_query_param_uint64 (&batch->shard_end),
297 141 : GNUNET_PQ_query_param_uint64 (&batch->progress_row),
298 141 : GNUNET_PQ_query_param_absolute_time (&lease_until),
299 : GNUNET_PQ_query_param_end
300 : };
301 141 : struct Context ctx = {
302 : .transaction_duplicates = transaction_duplicates,
303 : .length = rlen,
304 : .status = GNUNET_OK
305 : };
306 :
307 141 : qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
308 : "do_import_credits",
309 : params,
310 : &helper_cb,
311 : &ctx);
312 141 : GNUNET_PQ_cleanup_query_params_closures (params);
313 141 : if ( (qs >= 0) &&
314 141 : (GNUNET_OK != ctx.status) )
315 0 : qs = GNUNET_DB_STATUS_HARD_ERROR;
316 141 : if (qs < 0)
317 0 : GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
318 : "Failed to import credit batch (%d)\n",
319 : qs);
320 : }
321 :
322 219 : for (unsigned int i = 0; i<rlen; i++)
323 78 : GNUNET_free (notify_s[i]);
324 163 : for (unsigned int i = 0; i<klen; i++)
325 22 : GNUNET_free (ka_notify_s[i]);
326 141 : if (qs < 0)
327 0 : return qs;
328 219 : for (unsigned int i = 0; i<rlen; i++)
329 : {
330 78 : if (transaction_duplicates[i])
331 2 : dups++;
332 78 : reserve_results[i] = transaction_duplicates[i]
333 : ? GNUNET_DB_STATUS_SUCCESS_NO_RESULTS
334 78 : : GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
335 : }
336 141 : GNUNET_PQ_event_do_poll (pg->conn);
337 141 : if (0 != dups)
338 2 : GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
339 : "%u/%u duplicates among incoming transactions. Try increasing WIREWATCH_IDLE_SLEEP_INTERVAL in the [exchange] configuration section (if this happens a lot).\n",
340 : dups,
341 : rlen);
342 141 : return qs;
343 : }
|
