1use super::{
4 build_in_clause, map_db_error, params_refs, parse_datetime_opt_row, parse_datetime_row,
5 parse_uuid_row, with_immediate_transaction,
6};
7use chrono::Utc;
8use r2d2::Pool;
9use r2d2_sqlite::SqliteConnectionManager;
10use rusqlite::OptionalExtension;
11use stateset_core::{
12 AgentFeedback, AgentFeedbackFilter, AgentFeedbackResponse, AgentReputationRepository,
13 CommerceError, CreateAgentFeedback, CreateAgentFeedbackResponse, FeedbackSummary, Result,
14};
15use uuid::Uuid;
16
17#[derive(Debug)]
19pub struct SqliteAgentReputationRepository {
20 pool: Pool<SqliteConnectionManager>,
21}
22
23impl SqliteAgentReputationRepository {
24 #[must_use]
25 pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
26 Self { pool }
27 }
28
29 fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
30 self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
31 }
32
33 fn row_to_feedback(row: &rusqlite::Row<'_>) -> rusqlite::Result<AgentFeedback> {
34 let value_i64: i64 = row.get("value")?;
35 Ok(AgentFeedback {
36 id: parse_uuid_row(&row.get::<_, String>("id")?, "agent_feedback", "id")?,
37 agent_registry: row.get("agent_registry")?,
38 agent_id: row.get("agent_id")?,
39 client_address: row.get("client_address")?,
40 feedback_index: row.get::<_, i64>("feedback_index")? as u64,
41 value: i128::from(value_i64),
42 value_decimals: row.get::<_, i64>("value_decimals")? as u8,
43 tag1: row.get("tag1")?,
44 tag2: row.get("tag2")?,
45 endpoint: row.get("endpoint")?,
46 feedback_uri: row.get("feedback_uri")?,
47 feedback_hash: row.get("feedback_hash")?,
48 is_revoked: row.get::<_, i32>("is_revoked")? == 1,
49 created_at: parse_datetime_row(
50 &row.get::<_, String>("created_at")?,
51 "agent_feedback",
52 "created_at",
53 )?,
54 revoked_at: parse_datetime_opt_row(
55 row.get::<_, Option<String>>("revoked_at")?,
56 "agent_feedback",
57 "revoked_at",
58 )?,
59 })
60 }
61
62 fn row_to_response(row: &rusqlite::Row<'_>) -> rusqlite::Result<AgentFeedbackResponse> {
63 Ok(AgentFeedbackResponse {
64 id: parse_uuid_row(&row.get::<_, String>("id")?, "agent_feedback_response", "id")?,
65 agent_registry: row.get("agent_registry")?,
66 agent_id: row.get("agent_id")?,
67 client_address: row.get("client_address")?,
68 feedback_index: row.get::<_, i64>("feedback_index")? as u64,
69 responder_address: row.get("responder_address")?,
70 response_uri: row.get("response_uri")?,
71 response_hash: row.get("response_hash")?,
72 created_at: parse_datetime_row(
73 &row.get::<_, String>("created_at")?,
74 "agent_feedback_response",
75 "created_at",
76 )?,
77 })
78 }
79
80 fn value_to_i64(value: i128) -> Result<i64> {
81 if value > i128::from(i64::MAX) || value < i128::from(i64::MIN) {
82 return Err(CommerceError::ValidationError(
83 "feedback value exceeds i64 range".to_string(),
84 ));
85 }
86 Ok(value as i64)
87 }
88
89 fn scale_value(value: i128, from_decimals: u8, to_decimals: u8) -> Result<i128> {
90 if from_decimals == to_decimals {
91 return Ok(value);
92 }
93 let diff = if to_decimals > from_decimals {
94 u32::from(to_decimals - from_decimals)
95 } else {
96 u32::from(from_decimals - to_decimals)
97 };
98 let factor = 10_i128.checked_pow(diff).ok_or_else(|| {
99 CommerceError::ValidationError("decimal scaling overflow".to_string())
100 })?;
101
102 if to_decimals > from_decimals {
103 value.checked_mul(factor).ok_or_else(|| {
104 CommerceError::ValidationError("decimal scaling overflow".to_string())
105 })
106 } else {
107 Ok(value / factor)
108 }
109 }
110}
111
112impl AgentReputationRepository for SqliteAgentReputationRepository {
113 fn give_feedback(&self, input: CreateAgentFeedback) -> Result<AgentFeedback> {
114 if input.value_decimals > 18 {
115 return Err(CommerceError::ValidationError(
116 "value_decimals must be between 0 and 18".to_string(),
117 ));
118 }
119
120 let value_i64 = Self::value_to_i64(input.value)?;
121 let now = Utc::now();
122 let id = Uuid::new_v4();
123
124 let agent_registry = input.agent_registry.clone();
125 let agent_id = input.agent_id.clone();
126 let client_address = input.client_address.clone();
127
128 let next_index = with_immediate_transaction(&self.pool, |tx| {
129 let next_index: i64 = tx.query_row(
130 "SELECT COALESCE(MAX(feedback_index), 0) + 1
131 FROM agent_feedback
132 WHERE agent_registry = ? AND agent_id = ? AND client_address = ?",
133 rusqlite::params![agent_registry, agent_id, client_address],
134 |row| row.get(0),
135 )?;
136
137 tx.execute(
138 "INSERT INTO agent_feedback (
139 id, agent_registry, agent_id, client_address, feedback_index,
140 value, value_decimals, tag1, tag2, endpoint, feedback_uri, feedback_hash,
141 is_revoked, created_at
142 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?)",
143 rusqlite::params![
144 id.to_string(),
145 input.agent_registry,
146 input.agent_id,
147 input.client_address,
148 next_index,
149 value_i64,
150 i64::from(input.value_decimals),
151 input.tag1,
152 input.tag2,
153 input.endpoint,
154 input.feedback_uri,
155 input.feedback_hash,
156 now.to_rfc3339(),
157 ],
158 )?;
159
160 Ok(next_index as u64)
161 })?;
162
163 self.read_feedback(&agent_registry, &agent_id, &client_address, next_index)?
164 .ok_or(CommerceError::NotFound)
165 }
166
167 fn revoke_feedback(
168 &self,
169 agent_registry: &str,
170 agent_id: &str,
171 client_address: &str,
172 feedback_index: u64,
173 ) -> Result<AgentFeedback> {
174 let conn = self.conn()?;
175 let rows = conn
176 .execute(
177 "UPDATE agent_feedback SET is_revoked = 1, revoked_at = ?
178 WHERE agent_registry = ? AND agent_id = ? AND client_address = ? AND feedback_index = ?",
179 rusqlite::params![
180 Utc::now().to_rfc3339(),
181 agent_registry,
182 agent_id,
183 client_address,
184 feedback_index as i64,
185 ],
186 )
187 .map_err(map_db_error)?;
188
189 if rows == 0 {
190 return Err(CommerceError::NotFound);
191 }
192
193 self.read_feedback(agent_registry, agent_id, client_address, feedback_index)?
194 .ok_or(CommerceError::NotFound)
195 }
196
197 fn read_feedback(
198 &self,
199 agent_registry: &str,
200 agent_id: &str,
201 client_address: &str,
202 feedback_index: u64,
203 ) -> Result<Option<AgentFeedback>> {
204 let conn = self.conn()?;
205 let mut stmt = conn
206 .prepare(
207 "SELECT * FROM agent_feedback
208 WHERE agent_registry = ? AND agent_id = ? AND client_address = ? AND feedback_index = ?",
209 )
210 .map_err(map_db_error)?;
211
212 stmt.query_row(
213 rusqlite::params![agent_registry, agent_id, client_address, feedback_index as i64],
214 Self::row_to_feedback,
215 )
216 .optional()
217 .map_err(map_db_error)
218 }
219
220 fn read_all_feedback(&self, filter: AgentFeedbackFilter) -> Result<Vec<AgentFeedback>> {
221 let conn = self.conn()?;
222 let mut conditions = vec!["1=1".to_string()];
223 let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![];
224
225 if let Some(ref registry) = filter.agent_registry {
226 conditions.push("agent_registry = ?".to_string());
227 params.push(Box::new(registry.clone()));
228 }
229 if let Some(ref agent_id) = filter.agent_id {
230 conditions.push("agent_id = ?".to_string());
231 params.push(Box::new(agent_id.clone()));
232 }
233 if let Some(ref clients) = filter.client_addresses {
234 if !clients.is_empty() {
235 let placeholders = build_in_clause(clients.len());
236 conditions.push(format!("client_address IN ({placeholders})"));
237 for client in clients {
238 params.push(Box::new(client.clone()));
239 }
240 }
241 }
242 if let Some(ref tag1) = filter.tag1 {
243 conditions.push("tag1 = ?".to_string());
244 params.push(Box::new(tag1.clone()));
245 }
246 if let Some(ref tag2) = filter.tag2 {
247 conditions.push("tag2 = ?".to_string());
248 params.push(Box::new(tag2.clone()));
249 }
250
251 let include_revoked = filter.include_revoked.unwrap_or(false);
252 if !include_revoked {
253 conditions.push("is_revoked = 0".to_string());
254 }
255
256 let limit = filter.limit.unwrap_or(100).min(1000);
257 let offset = filter.offset.unwrap_or(0);
258
259 let sql = format!(
260 "SELECT * FROM agent_feedback WHERE {} ORDER BY created_at DESC LIMIT ? OFFSET ?",
261 conditions.join(" AND ")
262 );
263 params.push(Box::new(i64::from(limit)));
264 params.push(Box::new(i64::from(offset)));
265
266 let param_refs = params_refs(¶ms);
267 let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
268 let rows = stmt
269 .query_map(rusqlite::params_from_iter(param_refs), Self::row_to_feedback)
270 .map_err(map_db_error)?;
271
272 let mut results = Vec::new();
273 for row in rows {
274 results.push(row.map_err(map_db_error)?);
275 }
276 Ok(results)
277 }
278
279 fn get_summary(
280 &self,
281 agent_registry: &str,
282 agent_id: &str,
283 client_addresses: Vec<String>,
284 tag1: Option<String>,
285 tag2: Option<String>,
286 ) -> Result<FeedbackSummary> {
287 if client_addresses.is_empty() {
288 return Err(CommerceError::ValidationError(
289 "client_addresses must be provided".to_string(),
290 ));
291 }
292
293 let conn = self.conn()?;
294 let mut conditions = vec!["agent_registry = ?".to_string(), "agent_id = ?".to_string()];
295 let mut params: Vec<Box<dyn rusqlite::ToSql>> =
296 vec![Box::new(agent_registry.to_string()), Box::new(agent_id.to_string())];
297
298 let placeholders = build_in_clause(client_addresses.len());
299 conditions.push(format!("client_address IN ({placeholders})"));
300 for client in client_addresses {
301 params.push(Box::new(client));
302 }
303
304 if let Some(tag1_val) = tag1 {
305 conditions.push("tag1 = ?".to_string());
306 params.push(Box::new(tag1_val));
307 }
308 if let Some(tag2_val) = tag2 {
309 conditions.push("tag2 = ?".to_string());
310 params.push(Box::new(tag2_val));
311 }
312
313 conditions.push("is_revoked = 0".to_string());
314
315 let sql = format!(
316 "SELECT value, value_decimals FROM agent_feedback WHERE {}",
317 conditions.join(" AND ")
318 );
319
320 let param_refs = params_refs(¶ms);
321 let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
322 let mut rows = stmt.query(rusqlite::params_from_iter(param_refs)).map_err(map_db_error)?;
323
324 let mut values: Vec<(i128, u8)> = Vec::new();
325 while let Some(row) = rows.next().map_err(map_db_error)? {
326 let value: i64 = row.get(0).map_err(map_db_error)?;
327 let decimals: i64 = row.get(1).map_err(map_db_error)?;
328 values.push((i128::from(value), decimals as u8));
329 }
330
331 if values.is_empty() {
332 return Ok(FeedbackSummary { count: 0, summary_value: 0, summary_value_decimals: 0 });
333 }
334
335 let max_decimals = values.iter().map(|(_, d)| *d).max().unwrap_or(0);
336
337 let mut sum: i128 = 0;
338 for (value, decimals) in &values {
339 let scaled = Self::scale_value(*value, *decimals, max_decimals)?;
340 sum = sum.checked_add(scaled).ok_or_else(|| {
341 CommerceError::ValidationError("feedback summary overflow".to_string())
342 })?;
343 }
344
345 Ok(FeedbackSummary {
346 count: values.len() as u64,
347 summary_value: sum,
348 summary_value_decimals: max_decimals,
349 })
350 }
351
352 fn append_response(&self, input: CreateAgentFeedbackResponse) -> Result<AgentFeedbackResponse> {
353 let conn = self.conn()?;
354 let id = Uuid::new_v4();
355 let now = Utc::now();
356
357 let feedback_exists: Option<i64> = conn
358 .query_row(
359 "SELECT 1 FROM agent_feedback
360 WHERE agent_registry = ? AND agent_id = ? AND client_address = ? AND feedback_index = ?",
361 rusqlite::params![
362 input.agent_registry,
363 input.agent_id,
364 input.client_address,
365 input.feedback_index as i64,
366 ],
367 |row| row.get(0),
368 )
369 .optional()
370 .map_err(map_db_error)?;
371
372 if feedback_exists.is_none() {
373 return Err(CommerceError::NotFound);
374 }
375
376 conn.execute(
377 "INSERT INTO agent_feedback_responses (
378 id, agent_registry, agent_id, client_address, feedback_index,
379 responder_address, response_uri, response_hash, created_at
380 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
381 rusqlite::params![
382 id.to_string(),
383 input.agent_registry,
384 input.agent_id,
385 input.client_address,
386 input.feedback_index as i64,
387 input.responder_address,
388 input.response_uri,
389 input.response_hash,
390 now.to_rfc3339(),
391 ],
392 )
393 .map_err(map_db_error)?;
394
395 let mut stmt = conn
396 .prepare("SELECT * FROM agent_feedback_responses WHERE id = ?")
397 .map_err(map_db_error)?;
398
399 stmt.query_row([id.to_string()], Self::row_to_response).map_err(map_db_error)
400 }
401
402 fn get_response_count(
403 &self,
404 agent_registry: &str,
405 agent_id: &str,
406 client_address: &str,
407 feedback_index: u64,
408 responders: Option<Vec<String>>,
409 ) -> Result<u64> {
410 let conn = self.conn()?;
411 let mut conditions = vec![
412 "agent_registry = ?".to_string(),
413 "agent_id = ?".to_string(),
414 "client_address = ?".to_string(),
415 "feedback_index = ?".to_string(),
416 ];
417 let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![
418 Box::new(agent_registry.to_string()),
419 Box::new(agent_id.to_string()),
420 Box::new(client_address.to_string()),
421 Box::new(feedback_index as i64),
422 ];
423
424 if let Some(responders) = responders {
425 if !responders.is_empty() {
426 let placeholders = build_in_clause(responders.len());
427 conditions.push(format!("responder_address IN ({placeholders})"));
428 for responder in responders {
429 params.push(Box::new(responder));
430 }
431 }
432 }
433
434 let sql = format!(
435 "SELECT COUNT(*) FROM agent_feedback_responses WHERE {}",
436 conditions.join(" AND ")
437 );
438
439 let param_refs = params_refs(¶ms);
440 let count: i64 = conn
441 .query_row(&sql, rusqlite::params_from_iter(param_refs), |row| row.get(0))
442 .map_err(map_db_error)?;
443
444 Ok(count as u64)
445 }
446
447 fn get_clients(&self, agent_registry: &str, agent_id: &str) -> Result<Vec<String>> {
448 let conn = self.conn()?;
449 let mut stmt = conn
450 .prepare(
451 "SELECT DISTINCT client_address FROM agent_feedback
452 WHERE agent_registry = ? AND agent_id = ?",
453 )
454 .map_err(map_db_error)?;
455
456 let rows = stmt
457 .query_map([agent_registry, agent_id], |row| row.get::<_, String>(0))
458 .map_err(map_db_error)?;
459
460 let mut results = Vec::new();
461 for row in rows {
462 results.push(row.map_err(map_db_error)?);
463 }
464 Ok(results)
465 }
466
467 fn get_last_index(
468 &self,
469 agent_registry: &str,
470 agent_id: &str,
471 client_address: &str,
472 ) -> Result<u64> {
473 let conn = self.conn()?;
474 let index: Option<i64> = conn
475 .query_row(
476 "SELECT MAX(feedback_index) FROM agent_feedback
477 WHERE agent_registry = ? AND agent_id = ? AND client_address = ?",
478 rusqlite::params![agent_registry, agent_id, client_address],
479 |row| row.get(0),
480 )
481 .optional()
482 .map_err(map_db_error)?;
483
484 Ok(index.unwrap_or(0) as u64)
485 }
486}