do_import_credits.c (13261B)
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 compute_notify_on_reserve (const struct TALER_ReservePublicKeyP *reserve_pub) 35 { 36 struct TALER_EXCHANGEDB_ReserveEventP rep = { 37 .header.size = htons (sizeof (rep)), 38 .header.type = htons (TALER_DBEVENT_EXCHANGE_RESERVE_INCOMING), 39 .reserve_pub = *reserve_pub 40 }; 41 42 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 compute_notify_on_kycauth (const struct TALER_NormalizedPaytoHashP *h_payto) 54 { 55 struct TALER_EXCHANGEDB_KycCompletedEventP rep = { 56 .header.size = htons (sizeof (rep)), 57 .header.type = htons (TALER_DBEVENT_EXCHANGE_KYC_COMPLETED), 58 .h_payto = *h_payto 59 }; 60 61 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 helper_cb (void *cls, 97 PGresult *result, 98 unsigned int num_results) 99 { 100 struct Context *ctx = cls; 101 102 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 GNUNET_break (0); 108 ctx->status = GNUNET_SYSERR; 109 return; 110 } 111 for (unsigned int i = 0; i<num_results; i++) 112 { 113 struct GNUNET_PQ_ResultSpec rs[] = { 114 GNUNET_PQ_result_spec_bool ("out_duplicate", 115 &ctx->transaction_duplicates[i]), 116 GNUNET_PQ_result_spec_end 117 }; 118 119 if (GNUNET_OK != 120 GNUNET_PQ_extract_result (result, 121 rs, 122 i)) 123 { 124 GNUNET_break (0); 125 ctx->status = GNUNET_SYSERR; 126 return; 127 } 128 } 129 } 130 131 132 enum GNUNET_DB_QueryStatus 133 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 { 138 unsigned int rlen = batch->reserves_length; 139 unsigned int klen = batch->kycauths_length; 140 unsigned int wlen = batch->wads_length; 141 unsigned int dups = 0; 142 143 /* reserve transfers */ 144 struct TALER_ReservePublicKeyP reserve_pubs[GNUNET_NZL (rlen)]; 145 uint64_t wire_references[GNUNET_NZL (rlen)]; 146 struct TALER_Amount balances[GNUNET_NZL (rlen)]; 147 struct GNUNET_TIME_Timestamp execution_times[GNUNET_NZL (rlen)]; 148 struct TALER_FullPaytoHashP h_full_paytos[GNUNET_NZL (rlen)]; 149 struct TALER_NormalizedPaytoHashP h_normalized_paytos[GNUNET_NZL (rlen)]; 150 const char *payto_uris[GNUNET_NZL (rlen)]; 151 char *notify_s[GNUNET_NZL (rlen)]; 152 bool transaction_duplicates[GNUNET_NZL (rlen)]; 153 154 /* KYC authentication transfers */ 155 union TALER_AccountPublicKeyP ka_account_pubs[GNUNET_NZL (klen)]; 156 uint64_t ka_wire_references[GNUNET_NZL (klen)]; 157 struct TALER_Amount ka_balances[GNUNET_NZL (klen)]; 158 struct GNUNET_TIME_Timestamp ka_execution_times[GNUNET_NZL (klen)]; 159 struct TALER_FullPaytoHashP ka_h_full_paytos[GNUNET_NZL (klen)]; 160 struct TALER_NormalizedPaytoHashP ka_h_normalized_paytos[GNUNET_NZL (klen)]; 161 const char *ka_payto_uris[GNUNET_NZL (klen)]; 162 char *ka_notify_s[GNUNET_NZL (klen)]; 163 164 /* WAD transfers */ 165 struct TALER_WadIdentifierP wad_ids[GNUNET_NZL (wlen)]; 166 const char *wad_origin_exchange_urls[GNUNET_NZL (wlen)]; 167 struct TALER_Amount wad_balances[GNUNET_NZL (wlen)]; 168 struct GNUNET_TIME_Timestamp wad_execution_times[GNUNET_NZL (wlen)]; 169 170 struct GNUNET_TIME_Timestamp reserve_expiration 171 = GNUNET_TIME_relative_to_timestamp (pg->idle_reserve_expiration_time); 172 struct GNUNET_TIME_Timestamp gc 173 = GNUNET_TIME_relative_to_timestamp (pg->legal_reserve_expiration_time); 174 struct GNUNET_TIME_Absolute lease_until 175 = GNUNET_TIME_relative_to_absolute (batch->lease); 176 enum GNUNET_DB_QueryStatus qs; 177 178 for (unsigned int i = 0; i<rlen; i++) 179 { 180 const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &batch->reserves[i]; 181 182 TALER_full_payto_hash (reserve->sender_account_details, 183 &h_full_paytos[i]); 184 TALER_full_payto_normalize_and_hash (reserve->sender_account_details, 185 &h_normalized_paytos[i]); 186 notify_s[i] = compute_notify_on_reserve (reserve->reserve_pub); 187 reserve_pubs[i] = *reserve->reserve_pub; 188 balances[i] = *reserve->balance; 189 execution_times[i] = reserve->execution_time; 190 payto_uris[i] = reserve->sender_account_details.full_payto; 191 wire_references[i] = reserve->wire_reference; 192 } 193 for (unsigned int i = 0; i<klen; i++) 194 { 195 const struct TALER_EXCHANGEDB_KycauthInInfo *ka = &batch->kycauths[i]; 196 197 TALER_full_payto_hash (ka->sender_account_details, 198 &ka_h_full_paytos[i]); 199 TALER_full_payto_normalize_and_hash (ka->sender_account_details, 200 &ka_h_normalized_paytos[i]); 201 ka_notify_s[i] = compute_notify_on_kycauth (&ka_h_normalized_paytos[i]); 202 ka_account_pubs[i] = *ka->account_pub; 203 ka_balances[i] = *ka->balance; 204 ka_execution_times[i] = ka->execution_time; 205 ka_payto_uris[i] = ka->sender_account_details.full_payto; 206 ka_wire_references[i] = ka->wire_reference; 207 } 208 for (unsigned int i = 0; i<wlen; i++) 209 { 210 const struct TALER_EXCHANGEDB_WadInInfo *wad = &batch->wads[i]; 211 212 wad_ids[i] = *wad->wad_id; 213 wad_origin_exchange_urls[i] = wad->origin_exchange_url; 214 wad_balances[i] = *wad->balance; 215 wad_execution_times[i] = wad->execution_time; 216 } 217 218 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 struct GNUNET_PQ_QueryParam params[] = { 227 GNUNET_PQ_query_param_string (batch->exchange_account_name), 228 GNUNET_PQ_query_param_timestamp (&gc), 229 GNUNET_PQ_query_param_timestamp (&reserve_expiration), 230 /* reserve transfers */ 231 GNUNET_PQ_query_param_array_auto_from_type (rlen, 232 reserve_pubs, 233 pg->conn), 234 GNUNET_PQ_query_param_array_uint64 (rlen, 235 wire_references, 236 pg->conn), 237 TALER_PQ_query_param_array_amount (rlen, 238 balances, 239 pg->conn), 240 GNUNET_PQ_query_param_array_timestamp (rlen, 241 execution_times, 242 pg->conn), 243 GNUNET_PQ_query_param_array_auto_from_type (rlen, 244 h_full_paytos, 245 pg->conn), 246 GNUNET_PQ_query_param_array_auto_from_type (rlen, 247 h_normalized_paytos, 248 pg->conn), 249 GNUNET_PQ_query_param_array_ptrs_string (rlen, 250 payto_uris, 251 pg->conn), 252 GNUNET_PQ_query_param_array_ptrs_string (rlen, 253 (const char **) notify_s, 254 pg->conn), 255 /* KYC authentication transfers */ 256 GNUNET_PQ_query_param_array_auto_from_type (klen, 257 ka_account_pubs, 258 pg->conn), 259 GNUNET_PQ_query_param_array_uint64 (klen, 260 ka_wire_references, 261 pg->conn), 262 TALER_PQ_query_param_array_amount (klen, 263 ka_balances, 264 pg->conn), 265 GNUNET_PQ_query_param_array_timestamp (klen, 266 ka_execution_times, 267 pg->conn), 268 GNUNET_PQ_query_param_array_auto_from_type (klen, 269 ka_h_full_paytos, 270 pg->conn), 271 GNUNET_PQ_query_param_array_auto_from_type (klen, 272 ka_h_normalized_paytos, 273 pg->conn), 274 GNUNET_PQ_query_param_array_ptrs_string (klen, 275 ka_payto_uris, 276 pg->conn), 277 GNUNET_PQ_query_param_array_ptrs_string (klen, 278 (const char **) ka_notify_s, 279 pg->conn), 280 /* WAD transfers */ 281 GNUNET_PQ_query_param_array_auto_from_type (wlen, 282 wad_ids, 283 pg->conn), 284 GNUNET_PQ_query_param_array_ptrs_string (wlen, 285 wad_origin_exchange_urls, 286 pg->conn), 287 TALER_PQ_query_param_array_amount (wlen, 288 wad_balances, 289 pg->conn), 290 GNUNET_PQ_query_param_array_timestamp (wlen, 291 wad_execution_times, 292 pg->conn), 293 /* shard bookkeeping */ 294 GNUNET_PQ_query_param_string (batch->job_name), 295 GNUNET_PQ_query_param_uint64 (&batch->shard_start), 296 GNUNET_PQ_query_param_uint64 (&batch->shard_end), 297 GNUNET_PQ_query_param_uint64 (&batch->progress_row), 298 GNUNET_PQ_query_param_absolute_time (&lease_until), 299 GNUNET_PQ_query_param_end 300 }; 301 struct Context ctx = { 302 .transaction_duplicates = transaction_duplicates, 303 .length = rlen, 304 .status = GNUNET_OK 305 }; 306 307 qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn, 308 "do_import_credits", 309 params, 310 &helper_cb, 311 &ctx); 312 GNUNET_PQ_cleanup_query_params_closures (params); 313 if ( (qs >= 0) && 314 (GNUNET_OK != ctx.status) ) 315 qs = GNUNET_DB_STATUS_HARD_ERROR; 316 if (qs < 0) 317 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 318 "Failed to import credit batch (%d)\n", 319 qs); 320 } 321 322 for (unsigned int i = 0; i<rlen; i++) 323 GNUNET_free (notify_s[i]); 324 for (unsigned int i = 0; i<klen; i++) 325 GNUNET_free (ka_notify_s[i]); 326 if (qs < 0) 327 return qs; 328 for (unsigned int i = 0; i<rlen; i++) 329 { 330 if (transaction_duplicates[i]) 331 dups++; 332 reserve_results[i] = transaction_duplicates[i] 333 ? GNUNET_DB_STATUS_SUCCESS_NO_RESULTS 334 : GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 335 } 336 GNUNET_PQ_event_do_poll (pg->conn); 337 if (0 != dups) 338 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 return qs; 343 }