do_aggregate.c (12134B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2022, 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 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_aggregate.c 18 * @brief Implementation of the do_aggregate function for Postgres 19 * @author Christian Grothoff 20 */ 21 #include "taler/taler_error_codes.h" 22 #include "taler/taler_pq_lib.h" 23 #include "exchange-database/compute_shard.h" 24 #include "exchange-database/do_aggregate.h" 25 #include "helper.h" 26 27 28 enum GNUNET_DB_QueryStatus 29 TALER_EXCHANGEDB_do_aggregate ( 30 struct TALER_EXCHANGEDB_PostgresContext *pg, 31 const struct TALER_FullPaytoHashP *h_payto, 32 const struct TALER_MerchantPublicKeyP *merchant_pub, 33 const struct TALER_WireTransferIdentifierRawP *wtid, 34 struct TALER_Amount *total) 35 { 36 uint64_t deposit_shard = TALER_EXCHANGEDB_compute_shard (merchant_pub); 37 struct GNUNET_TIME_Absolute now = {0}; 38 uint64_t sum_deposit_value; 39 uint64_t sum_deposit_frac; 40 uint64_t sum_refund_value; 41 uint64_t sum_refund_frac; 42 uint64_t sum_fee_value; 43 uint64_t sum_fee_frac; 44 enum GNUNET_DB_QueryStatus qs; 45 struct TALER_Amount sum_deposit; 46 struct TALER_Amount sum_refund; 47 struct TALER_Amount sum_fee; 48 struct TALER_Amount delta; 49 50 now = GNUNET_TIME_absolute_round_down (GNUNET_TIME_absolute_get (), 51 pg->aggregator_shift); 52 PREPARE (pg, 53 "do_aggregate", 54 "WITH bdep AS (" /* restrict to our merchant and account and mark as done */ 55 " UPDATE batch_deposits" 56 " SET done=TRUE" 57 " WHERE NOT (done OR policy_blocked)" /* only actually executable deposits */ 58 " AND refund_deadline<$1" 59 " AND shard=$5" /* only for efficiency, merchant_pub is what we really filter by */ 60 " AND merchant_pub=$2" /* filter by target merchant */ 61 " AND wire_target_h_payto=$3" /* merchant could have a 2nd bank account */ 62 " RETURNING" 63 " batch_deposit_serial_id)" 64 " ,cdep AS (" 65 " SELECT" 66 " coin_deposit_serial_id" 67 " ,batch_deposit_serial_id" 68 " ,coin_pub" 69 " ,amount_with_fee AS amount" 70 " FROM coin_deposits" 71 " WHERE batch_deposit_serial_id IN (SELECT batch_deposit_serial_id FROM bdep))" 72 " ,ref AS (" /* find applicable refunds -- NOTE: may do a full join on the master, maybe find a left-join way to integrate with query above to push it to the shards? */ 73 " SELECT" 74 " amount_with_fee AS refund" 75 " ,coin_pub" 76 " ,batch_deposit_serial_id" /* theoretically, coin could be in multiple refunded transactions */ 77 " FROM refunds" 78 " WHERE coin_pub IN (SELECT coin_pub FROM cdep)" 79 " AND batch_deposit_serial_id IN (SELECT batch_deposit_serial_id FROM bdep))" 80 " ,ref_by_coin AS (" /* total up refunds by coin */ 81 " SELECT" 82 " SUM((ref.refund).val) AS sum_refund_val" 83 " ,SUM((ref.refund).frac) AS sum_refund_frac" 84 " ,coin_pub" 85 " ,batch_deposit_serial_id" /* theoretically, coin could be in multiple refunded transactions */ 86 " FROM ref" 87 " GROUP BY coin_pub, batch_deposit_serial_id)" 88 " ,norm_ref_by_coin AS (" /* normalize */ 89 " SELECT" 90 " sum_refund_val + sum_refund_frac / 100000000 AS norm_refund_val" 91 " ,sum_refund_frac % 100000000 AS norm_refund_frac" 92 " ,coin_pub" 93 " ,batch_deposit_serial_id" /* theoretically, coin could be in multiple refunded transactions */ 94 " FROM ref_by_coin)" 95 " ,fully_refunded_coins AS (" /* find applicable refunds -- NOTE: may do a full join on the master, maybe find a left-join way to integrate with query above to push it to the shards? */ 96 " SELECT" 97 " cdep.coin_pub" 98 " FROM norm_ref_by_coin norm" 99 " JOIN cdep" 100 " ON (norm.coin_pub = cdep.coin_pub" 101 " AND norm.batch_deposit_serial_id = cdep.batch_deposit_serial_id" 102 " AND norm.norm_refund_val = (cdep.amount).val" 103 " AND norm.norm_refund_frac = (cdep.amount).frac))" 104 " ,remainders AS (" /* what is left of each deposit after refunds */ 105 " SELECT" 106 " cdep.coin_pub" 107 " ,cdep.batch_deposit_serial_id" 108 " ,CAST( (cdep.amount).val" 109 " - COALESCE(norm.norm_refund_val,0)" 110 " - CASE WHEN (cdep.amount).frac" 111 " < COALESCE(norm.norm_refund_frac,0)" 112 " THEN 1 ELSE 0 END AS INT8) AS rem_val" 113 " ,CAST( (cdep.amount).frac" 114 " - COALESCE(norm.norm_refund_frac,0)" 115 " + CASE WHEN (cdep.amount).frac" 116 " < COALESCE(norm.norm_refund_frac,0)" 117 " THEN 100000000 ELSE 0 END AS INT8) AS rem_frac" 118 " FROM cdep" 119 " LEFT JOIN norm_ref_by_coin norm" 120 " ON (norm.coin_pub = cdep.coin_pub" 121 " AND norm.batch_deposit_serial_id = cdep.batch_deposit_serial_id))" 122 " ,fees AS (" /* find deposit fees for not fully refunded deposits */ 123 /* The fee is capped at what is left of the deposit after refunds: 124 nothing bounds a partial refund to (deposit - deposit fee), and 125 charging the full fee on top of a larger refund would make the 126 amount to be wired out negative. */ 127 " SELECT" 128 " CASE WHEN ( ((denom.fee_deposit).val,(denom.fee_deposit).frac)" 129 " <= (rem.rem_val,rem.rem_frac) )" 130 " THEN (denom.fee_deposit).val" 131 " ELSE rem.rem_val END AS fee_val" 132 " ,CASE WHEN ( ((denom.fee_deposit).val,(denom.fee_deposit).frac)" 133 " <= (rem.rem_val,rem.rem_frac) )" 134 " THEN (denom.fee_deposit).frac" 135 " ELSE rem.rem_frac END AS fee_frac" 136 " ,rem.batch_deposit_serial_id" /* ensures we get the fee for each coin, not once per denomination */ 137 " FROM remainders rem" 138 " JOIN known_coins kc" /* NOTE: may do a full join on the master, maybe find a left-join way to integrate with query above to push it to the shards? */ 139 " ON (kc.coin_pub = rem.coin_pub)" 140 " JOIN denominations denom" 141 " USING (denominations_serial)" 142 " WHERE rem.coin_pub NOT IN (SELECT coin_pub FROM fully_refunded_coins))" 143 " ,dummy AS (" /* add deposits to aggregation_tracking */ 144 " INSERT INTO aggregation_tracking" 145 " (batch_deposit_serial_id" 146 " ,wtid_raw)" 147 " SELECT batch_deposit_serial_id,$4" 148 " FROM bdep)" 149 "SELECT" /* calculate totals (deposits, refunds and fees) */ 150 " CAST(COALESCE(SUM((cdep.amount).val),0) AS INT8) AS sum_deposit_value" 151 /* cast needed, otherwise we get NUMBER */ 152 " ,COALESCE(SUM((cdep.amount).frac),0) AS sum_deposit_fraction" /* SUM over INT returns INT8 */ 153 " ,CAST(COALESCE(SUM((ref.refund).val),0) AS INT8) AS sum_refund_value" 154 " ,COALESCE(SUM((ref.refund).frac),0) AS sum_refund_fraction" 155 " ,CAST(COALESCE(SUM(fees.fee_val),0) AS INT8) AS sum_fee_value" 156 " ,CAST(COALESCE(SUM(fees.fee_frac),0) AS INT8) AS sum_fee_fraction" 157 " FROM cdep " 158 " FULL OUTER JOIN ref ON (FALSE)" /* We just want all sums */ 159 " FULL OUTER JOIN fees ON (FALSE);"); 160 161 { 162 struct GNUNET_PQ_QueryParam params[] = { 163 GNUNET_PQ_query_param_absolute_time (&now), 164 GNUNET_PQ_query_param_auto_from_type (merchant_pub), 165 GNUNET_PQ_query_param_auto_from_type (h_payto), 166 GNUNET_PQ_query_param_auto_from_type (wtid), 167 GNUNET_PQ_query_param_uint64 (&deposit_shard), 168 GNUNET_PQ_query_param_end 169 }; 170 struct GNUNET_PQ_ResultSpec rs[] = { 171 GNUNET_PQ_result_spec_uint64 ("sum_deposit_value", 172 &sum_deposit_value), 173 GNUNET_PQ_result_spec_uint64 ("sum_deposit_fraction", 174 &sum_deposit_frac), 175 GNUNET_PQ_result_spec_uint64 ("sum_refund_value", 176 &sum_refund_value), 177 GNUNET_PQ_result_spec_uint64 ("sum_refund_fraction", 178 &sum_refund_frac), 179 GNUNET_PQ_result_spec_uint64 ("sum_fee_value", 180 &sum_fee_value), 181 GNUNET_PQ_result_spec_uint64 ("sum_fee_fraction", 182 &sum_fee_frac), 183 GNUNET_PQ_result_spec_end 184 }; 185 186 qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn, 187 "do_aggregate", 188 params, 189 rs); 190 } 191 if (qs < 0) 192 { 193 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 194 return qs; 195 } 196 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 197 { 198 GNUNET_assert (GNUNET_OK == 199 TALER_amount_set_zero (pg->currency, 200 total)); 201 return qs; 202 } 203 GNUNET_assert (GNUNET_OK == 204 TALER_amount_set_zero (pg->currency, 205 &sum_deposit)); 206 GNUNET_assert (GNUNET_OK == 207 TALER_amount_set_zero (pg->currency, 208 &sum_refund)); 209 GNUNET_assert (GNUNET_OK == 210 TALER_amount_set_zero (pg->currency, 211 &sum_fee)); 212 sum_deposit.value = sum_deposit_frac / TALER_AMOUNT_FRAC_BASE 213 + sum_deposit_value; 214 sum_deposit.fraction = sum_deposit_frac % TALER_AMOUNT_FRAC_BASE; 215 sum_refund.value = sum_refund_frac / TALER_AMOUNT_FRAC_BASE 216 + sum_refund_value; 217 sum_refund.fraction = sum_refund_frac % TALER_AMOUNT_FRAC_BASE; 218 sum_fee.value = sum_fee_frac / TALER_AMOUNT_FRAC_BASE 219 + sum_fee_value; 220 sum_fee.fraction = sum_fee_frac % TALER_AMOUNT_FRAC_BASE; \ 221 /* With the fee capped at the un-refunded remainder above, neither 222 subtraction can go negative. Should the invariant ever be violated 223 again, refuse the aggregation (the caller rolls the transaction back and 224 reports the failure) rather than abort() the daemon: an abort here stops 225 *all* payouts for the shard and, since the transaction is rolled back, 226 the very same batch is picked up and aborts again on restart. */ 227 if (0 > 228 TALER_amount_subtract (&delta, 229 &sum_deposit, 230 &sum_refund)) 231 { 232 GNUNET_break (0); 233 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 234 "Refunds (%s) exceed deposits in aggregation\n", 235 TALER_amount2s (&sum_refund)); 236 return GNUNET_DB_STATUS_HARD_ERROR; 237 } 238 if (0 > 239 TALER_amount_subtract (total, 240 &delta, 241 &sum_fee)) 242 { 243 GNUNET_break (0); 244 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 245 "Deposit fees (%s) exceed what is left of the deposits after refunds in aggregation\n", 246 TALER_amount2s (&sum_fee)); 247 return GNUNET_DB_STATUS_HARD_ERROR; 248 } 249 return qs; 250 }