Line data Source code
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 61 : 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 61 : uint64_t deposit_shard = TALER_EXCHANGEDB_compute_shard (merchant_pub);
37 61 : 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 61 : now = GNUNET_TIME_absolute_round_down (GNUNET_TIME_absolute_get (),
51 : pg->aggregator_shift);
52 61 : 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 61 : struct GNUNET_PQ_QueryParam params[] = {
163 61 : GNUNET_PQ_query_param_absolute_time (&now),
164 61 : GNUNET_PQ_query_param_auto_from_type (merchant_pub),
165 61 : GNUNET_PQ_query_param_auto_from_type (h_payto),
166 61 : GNUNET_PQ_query_param_auto_from_type (wtid),
167 61 : GNUNET_PQ_query_param_uint64 (&deposit_shard),
168 : GNUNET_PQ_query_param_end
169 : };
170 61 : struct GNUNET_PQ_ResultSpec rs[] = {
171 61 : GNUNET_PQ_result_spec_uint64 ("sum_deposit_value",
172 : &sum_deposit_value),
173 61 : GNUNET_PQ_result_spec_uint64 ("sum_deposit_fraction",
174 : &sum_deposit_frac),
175 61 : GNUNET_PQ_result_spec_uint64 ("sum_refund_value",
176 : &sum_refund_value),
177 61 : GNUNET_PQ_result_spec_uint64 ("sum_refund_fraction",
178 : &sum_refund_frac),
179 61 : GNUNET_PQ_result_spec_uint64 ("sum_fee_value",
180 : &sum_fee_value),
181 61 : GNUNET_PQ_result_spec_uint64 ("sum_fee_fraction",
182 : &sum_fee_frac),
183 : GNUNET_PQ_result_spec_end
184 : };
185 :
186 61 : qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
187 : "do_aggregate",
188 : params,
189 : rs);
190 : }
191 61 : if (qs < 0)
192 : {
193 0 : GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs);
194 0 : return qs;
195 : }
196 61 : if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs)
197 : {
198 0 : GNUNET_assert (GNUNET_OK ==
199 : TALER_amount_set_zero (pg->currency,
200 : total));
201 0 : return qs;
202 : }
203 61 : GNUNET_assert (GNUNET_OK ==
204 : TALER_amount_set_zero (pg->currency,
205 : &sum_deposit));
206 61 : GNUNET_assert (GNUNET_OK ==
207 : TALER_amount_set_zero (pg->currency,
208 : &sum_refund));
209 61 : GNUNET_assert (GNUNET_OK ==
210 : TALER_amount_set_zero (pg->currency,
211 : &sum_fee));
212 61 : sum_deposit.value = sum_deposit_frac / TALER_AMOUNT_FRAC_BASE
213 61 : + sum_deposit_value;
214 61 : sum_deposit.fraction = sum_deposit_frac % TALER_AMOUNT_FRAC_BASE;
215 61 : sum_refund.value = sum_refund_frac / TALER_AMOUNT_FRAC_BASE
216 61 : + sum_refund_value;
217 61 : sum_refund.fraction = sum_refund_frac % TALER_AMOUNT_FRAC_BASE;
218 61 : sum_fee.value = sum_fee_frac / TALER_AMOUNT_FRAC_BASE
219 61 : + sum_fee_value;
220 61 : 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 61 : if (0 >
228 61 : TALER_amount_subtract (&delta,
229 : &sum_deposit,
230 : &sum_refund))
231 : {
232 0 : GNUNET_break (0);
233 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
234 : "Refunds (%s) exceed deposits in aggregation\n",
235 : TALER_amount2s (&sum_refund));
236 0 : return GNUNET_DB_STATUS_HARD_ERROR;
237 : }
238 61 : if (0 >
239 61 : TALER_amount_subtract (total,
240 : &delta,
241 : &sum_fee))
242 : {
243 0 : GNUNET_break (0);
244 0 : 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 0 : return GNUNET_DB_STATUS_HARD_ERROR;
248 : }
249 61 : return qs;
250 : }
|
