repository.rs (10626B)
1 //! Product repository functions with OpenTelemetry instrumentation 2 //! 3 //! This module contains database operations for products, separated from HTTP handlers. 4 //! Each function is instrumented with OpenTelemetry semantic conventions for database spans. 5 6 use sqlx::{PgPool, Postgres, QueryBuilder}; 7 use tracing::instrument; 8 use uuid::Uuid; 9 10 use crate::models::{ProductDetail, ProductQueryParams, ProductWithRating}; 11 12 /// Fetch a single product by UUID with full details 13 /// 14 /// Joins with categories and ratings tables to return complete product information. 15 /// Returns None if the product doesn't exist or has been deleted. 16 #[instrument( 17 name = "SELECT products", 18 skip(pool), 19 fields( 20 db.system.name = "postgresql", 21 db.namespace = "products", 22 db.operation.name = "SELECT", 23 db.collection.name = "products", 24 db.query.text = "SELECT p.*, c.*, AVG(r.rating), COUNT(r.id) FROM products p LEFT JOIN ... WHERE p.uuid = $1", 25 otelmart.product.uuid = %uuid 26 ) 27 )] 28 pub async fn get_product_by_uuid( 29 pool: &PgPool, 30 uuid: Uuid, 31 ) -> Result<Option<ProductDetail>, sqlx::Error> { 32 sqlx::query_as::<_, ProductDetail>( 33 r#" 34 SELECT 35 p.id, 36 p.uuid, 37 p.asin, 38 p.sku, 39 p.gtin, 40 p.product_name, 41 p.brand, 42 p.description, 43 p.url, 44 p.price, 45 p.initial_price, 46 p.discount, 47 p.currency, 48 p.stock_quantity, 49 p.sizes, 50 p.colors, 51 p.image_url, 52 p.available_for_delivery, 53 p.available_for_pickup, 54 p.free_returns, 55 p.is_active, 56 p.created_at, 57 p.updated_at, 58 p.data_timestamp, 59 p.category_id, 60 c.name as category_name, 61 c.uuid as category_uuid, 62 c.slug as category_slug, 63 COALESCE(AVG(r.rating)::FLOAT8, NULL) as average_rating, 64 COUNT(r.id)::BIGINT as rating_count, 65 CAST(p.id AS TEXT) as product_id_str, 66 get_root_category_name(p.category_id) as root_category_name, 67 p.deleted_at 68 FROM products p 69 LEFT JOIN categories c ON p.category_id = c.id 70 LEFT JOIN ratings r ON p.id = r.product_id 71 WHERE p.uuid = $1 AND p.deleted_at IS NULL 72 GROUP BY p.id, p.uuid, p.asin, p.sku, p.gtin, p.product_name, p.brand, p.description, 73 p.url, p.price, p.initial_price, p.discount, p.currency, p.stock_quantity, p.sizes, p.colors, p.image_url, 74 p.available_for_delivery, p.available_for_pickup, p.free_returns, 75 p.is_active, p.created_at, p.updated_at, p.data_timestamp, p.category_id, 76 p.deleted_at, c.name, c.uuid, c.slug 77 "#, 78 ) 79 .bind(uuid) 80 .fetch_optional(pool) 81 .await 82 } 83 84 /// List products with pagination, filtering, and rating aggregation 85 /// 86 /// Uses a CTE to aggregate product data with ratings, then applies 87 /// filters including rating-based filters on the CTE result. 88 /// Returns products and a total count for pagination. 89 #[instrument( 90 name = "SELECT products", 91 skip(pool, params), 92 fields( 93 otel.kind = "client", 94 db.system.name = "postgresql", 95 db.namespace = "products", 96 db.operation.name = "SELECT", 97 db.collection.name = "products", 98 db.query.text = "SELECT * FROM products WHERE ... ORDER BY ... LIMIT ... OFFSET ...", 99 otelmart.page = page, 100 otelmart.page_size = page_size, 101 otelmart.category.id = ?params.category_id, 102 otelmart.brand = ?params.brand 103 ) 104 )] 105 pub async fn list_products( 106 pool: &PgPool, 107 params: &ProductQueryParams, 108 page: i32, 109 page_size: i32, 110 offset: i32, 111 ) -> Result<(Vec<ProductWithRating>, i64), sqlx::Error> { 112 // Check if we have rating filters (determines count query strategy) 113 let has_rating_filters = 114 params.rating_gt.is_some() || params.rating_lt.is_some() || params.rating_eq.is_some(); 115 116 // Get total count - optimized based on whether we have rating filters 117 let total_count = if has_rating_filters { 118 count_with_ratings(pool, params).await? 119 } else { 120 count_simple(pool, params).await? 121 }; 122 123 // Build main query with CTE for product data and ratings 124 let mut main_query = QueryBuilder::<Postgres>::new( 125 r#" 126 WITH product_data AS ( 127 SELECT 128 p.id, 129 p.uuid, 130 p.product_name, 131 p.brand, 132 p.description, 133 p.price, 134 p.initial_price, 135 p.discount, 136 p.stock_quantity, 137 p.image_url, 138 p.category_id, 139 c.name as category_name, 140 p.created_at, 141 p.updated_at, 142 COALESCE(AVG(r.rating)::FLOAT8, NULL) as average_rating, 143 COUNT(r.id)::BIGINT as rating_count 144 FROM products p 145 LEFT JOIN categories c ON p.category_id = c.id 146 LEFT JOIN ratings r ON p.id = r.product_id 147 WHERE p.deleted_at IS NULL AND p.is_active = true AND p.stock_quantity > 0 148 GROUP BY p.id, p.uuid, p.product_name, p.brand, p.description, p.price, 149 p.initial_price, p.discount, p.stock_quantity, p.image_url, p.category_id, c.name, 150 p.created_at, p.updated_at 151 ) 152 SELECT * FROM product_data WHERE 1=1 153 "#, 154 ); 155 156 // Apply filters with safe parameter binding 157 apply_product_filters(&mut main_query, params, "", true); 158 159 // Add ordering and pagination 160 main_query.push(" ORDER BY updated_at DESC LIMIT "); 161 main_query.push_bind(page_size); 162 main_query.push(" OFFSET "); 163 main_query.push_bind(offset); 164 165 // Execute main query 166 let products = main_query.build_query_as().fetch_all(pool).await?; 167 168 Ok((products, total_count)) 169 } 170 171 /// Simple count query without rating joins (faster for non-rating filters) 172 async fn count_simple(pool: &PgPool, params: &ProductQueryParams) -> Result<i64, sqlx::Error> { 173 let mut count_builder = QueryBuilder::<Postgres>::new( 174 "SELECT COUNT(*) FROM products p WHERE p.deleted_at IS NULL AND p.is_active = true AND p.stock_quantity > 0" 175 ); 176 177 // Apply non-rating filters 178 apply_product_filters(&mut count_builder, params, "p.", false); 179 180 count_builder.build_query_scalar().fetch_one(pool).await 181 } 182 183 /// Count query with rating aggregation (for rating-based filters) 184 async fn count_with_ratings( 185 pool: &PgPool, 186 params: &ProductQueryParams, 187 ) -> Result<i64, sqlx::Error> { 188 let mut count_builder = QueryBuilder::<Postgres>::new( 189 r#" 190 SELECT COUNT(DISTINCT p.id) FROM products p 191 LEFT JOIN ratings r ON p.id = r.product_id 192 WHERE p.deleted_at IS NULL AND p.is_active = true AND p.stock_quantity > 0 193 "#, 194 ); 195 196 // Apply non-rating filters to WHERE clause 197 apply_product_filters(&mut count_builder, params, "p.", false); 198 199 // Add GROUP BY for ratings aggregation 200 count_builder.push(" GROUP BY p.id"); 201 202 // Apply rating filters with HAVING clause 203 let mut has_having = false; 204 205 if let Some(rating_gt) = params.rating_gt { 206 count_builder.push(if has_having { " AND " } else { " HAVING " }); 207 has_having = true; 208 count_builder.push("AVG(r.rating) > "); 209 count_builder.push_bind(rating_gt); 210 } 211 212 if let Some(rating_lt) = params.rating_lt { 213 count_builder.push(if has_having { " AND " } else { " HAVING " }); 214 has_having = true; 215 count_builder.push("AVG(r.rating) < "); 216 count_builder.push_bind(rating_lt); 217 } 218 219 if let Some(rating_eq) = params.rating_eq { 220 count_builder.push(if has_having { " AND " } else { " HAVING " }); 221 count_builder.push("AVG(r.rating) = "); 222 count_builder.push_bind(rating_eq); 223 } 224 225 count_builder.build_query_scalar().fetch_one(pool).await 226 } 227 228 /// Apply product filters to a query builder 229 /// 230 /// Centralizes filter logic to avoid duplication across count and main queries. 231 /// 232 /// # Arguments 233 /// * `query` - The QueryBuilder to append filters to 234 /// * `params` - Query parameters containing filter values 235 /// * `column_prefix` - Prefix for column names ("p." for table alias, "" for CTE columns) 236 /// * `include_ratings` - Whether to include rating filters (only on CTE results) 237 fn apply_product_filters<'a>( 238 query: &mut QueryBuilder<'a, Postgres>, 239 params: &'a ProductQueryParams, 240 column_prefix: &str, 241 include_ratings: bool, 242 ) { 243 // Name filter - case-insensitive partial match 244 if let Some(ref name) = params.name { 245 query.push(format!(" AND {}product_name ILIKE ", column_prefix).as_str()); 246 query.push_bind(format!("%{}%", name)); 247 } 248 249 // Category filter - exact match 250 if let Some(category_id) = params.category_id { 251 query.push(format!(" AND {}category_id = ", column_prefix).as_str()); 252 query.push_bind(category_id); 253 } 254 255 // Brand filter - case-insensitive partial match 256 if let Some(ref brand) = params.brand { 257 query.push(format!(" AND {}brand ILIKE ", column_prefix).as_str()); 258 query.push_bind(format!("%{}%", brand)); 259 } 260 261 // Date range filters on updated_at 262 if let Some(start_date) = params.start_date { 263 query.push(format!(" AND {}updated_at >= ", column_prefix).as_str()); 264 query.push_bind(start_date); 265 } 266 267 if let Some(end_date) = params.end_date { 268 query.push(format!(" AND {}updated_at <= ", column_prefix).as_str()); 269 query.push_bind(end_date); 270 } 271 272 // Price range filters 273 if let Some(ref min_price) = params.min_price { 274 query.push(format!(" AND {}price >= ", column_prefix).as_str()); 275 query.push_bind(min_price); 276 } 277 278 if let Some(ref max_price) = params.max_price { 279 query.push(format!(" AND {}price <= ", column_prefix).as_str()); 280 query.push_bind(max_price); 281 } 282 283 // Rating filters (only for main query on CTE columns) 284 if include_ratings { 285 if let Some(rating_gt) = params.rating_gt { 286 query.push(" AND average_rating > "); 287 query.push_bind(rating_gt); 288 } 289 290 if let Some(rating_lt) = params.rating_lt { 291 query.push(" AND average_rating < "); 292 query.push_bind(rating_lt); 293 } 294 295 if let Some(rating_eq) = params.rating_eq { 296 query.push(" AND average_rating = "); 297 query.push_bind(rating_eq); 298 } 299 } 300 }