+74 -133

api/src/api/xrpc_dynamic.rs

reviewed

··· 4 4 }; 5 5 use axum::{ 6 6 extract::{Path, Query, State}, 7 7 - http::{HeaderMap, StatusCode}, 7 7 + http::HeaderMap, 8 8 response::Json, 9 9 }; 10 10 use chrono::Utc; ··· 14 14 use crate::auth::{ 15 15 extract_bearer_token, get_atproto_auth_for_user_cached, verify_oauth_token_cached, 16 16 }; 17 17 + use crate::errors::AppError; 17 18 use crate::models::{ 18 19 IndexedRecord, Record, SliceRecordsOutput, SliceRecordsParams, SortField, WhereCondition, 19 20 }; 20 21 use std::collections::HashMap; 21 22 22 22 - // Helper function to convert StatusCode errors to JSON error responses 23 23 - fn status_to_error_response(status: StatusCode) -> (StatusCode, Json<serde_json::Value>) { 24 24 - let message = match status { 25 25 - StatusCode::UNAUTHORIZED => "Authentication required", 26 26 - StatusCode::FORBIDDEN => "Access forbidden", 27 27 - StatusCode::NOT_FOUND => "Resource not found", 28 28 - StatusCode::BAD_REQUEST => "Invalid request", 29 29 - StatusCode::INTERNAL_SERVER_ERROR => "Internal server error", 30 30 - _ => "Request failed", 31 31 - }; 32 32 - 33 33 - ( 34 34 - status, 35 35 - Json(serde_json::json!({ 36 36 - "error": status.as_str(), 37 37 - "message": message 38 38 - })), 39 39 - ) 40 40 - } 41 23 42 24 #[derive(Deserialize)] 43 25 pub struct GetRecordParams { ··· 50 32 Path(method): Path<String>, 51 33 State(state): State<AppState>, 52 34 Query(params): Query<serde_json::Value>, 53 53 - ) -> Result<Json<serde_json::Value>, StatusCode> { 35 35 + ) -> Result<Json<serde_json::Value>, AppError> { 54 36 // Parse the XRPC method (e.g., "social.grain.gallery.getRecords") 55 37 if method.ends_with(".getRecords") { 56 38 let collection = method.trim_end_matches(".getRecords").to_string(); ··· 62 44 let collection = method.trim_end_matches(".getRecord").to_string(); 63 45 dynamic_get_record_impl(collection, state, params).await 64 46 } else { 65 65 - Err(StatusCode::NOT_FOUND) 47 47 + Err(AppError::NotFound("Unknown XRPC method".to_string())) 66 48 } 67 49 } 68 50 ··· 72 54 State(state): State<AppState>, 73 55 headers: HeaderMap, 74 56 Json(body): Json<serde_json::Value>, 75 75 - ) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> { 57 57 + ) -> Result<Json<serde_json::Value>, AppError> { 76 58 // Handle dynamic collection methods (e.g., social.grain.gallery.createRecord) 77 59 if method.ends_with(".getRecords") { 78 60 let collection = method.trim_end_matches(".getRecords").to_string(); ··· 90 72 let collection = method.trim_end_matches(".deleteRecord").to_string(); 91 73 dynamic_collection_delete_impl(state, headers, body, collection).await 92 74 } else { 93 93 - Err(status_to_error_response(StatusCode::NOT_FOUND)) 75 75 + Err(AppError::NotFound("Method not found".to_string())) 94 76 } 95 77 } 96 78 ··· 99 81 collection: String, 100 82 state: AppState, 101 83 params: serde_json::Value, 102 102 - ) -> Result<Json<serde_json::Value>, StatusCode> { 84 84 + ) -> Result<Json<serde_json::Value>, AppError> { 103 85 // Parse parameters into SliceRecordsParams format 104 86 let slice = params 105 87 .get("slice") 106 88 .and_then(|v| v.as_str()) 107 107 - .ok_or(StatusCode::BAD_REQUEST)? 89 89 + .ok_or(AppError::BadRequest("Missing slice parameter".to_string()))? 108 90 .to_string(); 109 91 110 92 let limit = params.get("limit").and_then(|v| { ··· 186 168 ); 187 169 } 188 170 189 189 - let where_clause = if where_conditions.is_empty() { 190 190 - None 191 191 - } else { 192 192 - Some(crate::models::WhereClause { 193 193 - conditions: where_conditions, 194 194 - or_conditions: None, 195 195 - }) 196 196 - }; 171 171 + // Add collection filter to where conditions 172 172 + where_conditions.insert( 173 173 + "collection".to_string(), 174 174 + WhereCondition { 175 175 + eq: Some(serde_json::Value::String(collection.clone())), 176 176 + in_values: None, 177 177 + contains: None, 178 178 + }, 179 179 + ); 180 180 + 181 181 + let where_clause = Some(crate::models::WhereClause { 182 182 + conditions: where_conditions, 183 183 + or_conditions: None, 184 184 + }); 197 185 198 186 let records_params = SliceRecordsParams { 199 187 slice, ··· 208 196 .database 209 197 .get_slice_collections_list(&records_params.slice) 210 198 .await 211 211 - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; 199 199 + .map_err(|_| AppError::Internal("Database error".to_string()))?; 212 200 213 201 // Special handling: network.slices.lexicon is always allowed as it defines the schema 214 202 if collection != "network.slices.lexicon" && !slice_collections.contains(&collection) { 215 215 - return Err(StatusCode::NOT_FOUND); 203 203 + return Err(AppError::NotFound("Collection not found".to_string())); 216 204 } 217 205 218 206 // Use the unified database method ··· 227 215 ) 228 216 .await 229 217 { 230 230 - Ok((mut records, cursor)) => { 231 231 - // Filter records to only include the specific collection 232 232 - records.retain(|record| record.collection == collection); 218 218 + Ok((records, cursor)) => { 219 219 + // No need to filter - collection filter is in the SQL query now 233 220 234 221 let indexed_records: Vec<IndexedRecord> = records 235 222 .into_iter() ··· 249 236 }; 250 237 251 238 Ok(Json( 252 252 - serde_json::to_value(output).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?, 239 239 + serde_json::to_value(output).map_err(|_| AppError::Internal("Serialization error".to_string()))?, 253 240 )) 254 241 } 255 255 - Err(_) => Err(StatusCode::INTERNAL_SERVER_ERROR), 242 242 + Err(_) => Err(AppError::Internal("Database error".to_string())), 256 243 } 257 244 } 258 245 ··· 261 248 collection: String, 262 249 state: AppState, 263 250 params: serde_json::Value, 264 264 - ) -> Result<Json<serde_json::Value>, StatusCode> { 251 251 + ) -> Result<Json<serde_json::Value>, AppError> { 265 252 let get_params: GetRecordParams = 266 266 - serde_json::from_value(params).map_err(|_| StatusCode::BAD_REQUEST)?; 253 253 + serde_json::from_value(params).map_err(|_| AppError::BadRequest("Invalid parameters".to_string()))?; 267 254 268 255 // First verify the collection belongs to this slice 269 256 let slice_collections = state 270 257 .database 271 258 .get_slice_collections_list(&get_params.slice) 272 259 .await 273 273 - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; 260 260 + .map_err(|_| AppError::Internal("Database error".to_string()))?; 274 261 275 262 // Special handling: network.slices.lexicon is always allowed as it defines the schema 276 263 if collection != "network.slices.lexicon" && !slice_collections.contains(&collection) { 277 277 - return Err(StatusCode::NOT_FOUND); 264 264 + return Err(AppError::NotFound("Collection not found".to_string())); 278 265 } 279 266 280 267 // Use direct database query by URI for efficiency 281 268 match state.database.get_record(&get_params.uri).await { 282 269 Ok(Some(record)) => { 283 270 let json_value = 284 284 - serde_json::to_value(record).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; 271 271 + serde_json::to_value(record).map_err(|_| AppError::Internal("Serialization error".to_string()))?; 285 272 Ok(Json(json_value)) 286 273 } 287 287 - Ok(None) => Err(StatusCode::NOT_FOUND), 288 288 - Err(_e) => Err(StatusCode::INTERNAL_SERVER_ERROR), 274 274 + Ok(None) => Err(AppError::NotFound("Record not found".to_string())), 275 275 + Err(_e) => Err(AppError::Internal("Database error".to_string())), 289 276 } 290 277 } 291 278 ··· 294 281 collection: String, 295 282 state: AppState, 296 283 body: serde_json::Value, 297 297 - ) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> { 284 284 + ) -> Result<Json<serde_json::Value>, AppError> { 298 285 // Parse the JSON body into SliceRecordsParams 299 299 - let mut records_params: SliceRecordsParams = serde_json::from_value(body).map_err(|_| { 300 300 - ( 301 301 - StatusCode::BAD_REQUEST, 302 302 - Json(serde_json::json!({"error": "Invalid request body"})), 303 303 - ) 304 304 - })?; 286 286 + let mut records_params: SliceRecordsParams = serde_json::from_value(body).map_err(|_| AppError::BadRequest("Invalid request body".to_string()))?; 305 287 306 288 // First verify the collection belongs to this slice 307 289 let slice_collections = state 308 290 .database 309 291 .get_slice_collections_list(&records_params.slice) 310 292 .await 311 311 - .map_err(|_| { 312 312 - ( 313 313 - StatusCode::INTERNAL_SERVER_ERROR, 314 314 - Json(serde_json::json!({"error": "Database error"})), 315 315 - ) 316 316 - })?; 293 293 + .map_err(|_| AppError::Internal("Database error".to_string()))?; 317 294 318 295 // Special handling: network.slices.lexicon is always allowed as it defines the schema 319 296 if collection != "network.slices.lexicon" && !slice_collections.contains(&collection) { 320 320 - return Err(( 321 321 - StatusCode::NOT_FOUND, 322 322 - Json(serde_json::json!({"error": "Collection not found"})), 323 323 - )); 297 297 + return Err(AppError::NotFound("Collection not found".to_string())); 324 298 } 325 299 326 300 // Add collection filter to where conditions ··· 373 347 cursor, 374 348 }; 375 349 376 376 - Ok(Json(serde_json::to_value(output).map_err(|_| { 377 377 - ( 378 378 - StatusCode::INTERNAL_SERVER_ERROR, 379 379 - Json(serde_json::json!({"error": "Serialization error"})), 380 380 - ) 381 381 - })?)) 350 350 + Ok(Json(serde_json::to_value(output).map_err(|e| AppError::Internal(format!("Serialization error: {}", e)))?)) 382 351 } 383 383 - Err(_) => Err(( 384 384 - StatusCode::INTERNAL_SERVER_ERROR, 385 385 - Json(serde_json::json!({"error": "Database error"})), 386 386 - )), 352 352 + Err(e) => Err(AppError::Internal(format!("Database error: {}", e))), 387 353 } 388 354 } 389 355 ··· 392 358 collection: String, 393 359 state: AppState, 394 360 params: serde_json::Value, 395 395 - ) -> Result<Json<serde_json::Value>, StatusCode> { 361 361 + ) -> Result<Json<serde_json::Value>, AppError> { 396 362 // Convert query parameters to SliceRecordsParams 397 363 let mut records_params: SliceRecordsParams = 398 398 - serde_json::from_value(params).map_err(|_| StatusCode::BAD_REQUEST)?; 364 364 + serde_json::from_value(params).map_err(|_| AppError::BadRequest("Invalid parameters".to_string()))?; 399 365 400 366 // First verify the collection belongs to this slice 401 367 let slice_collections = state 402 368 .database 403 369 .get_slice_collections_list(&records_params.slice) 404 370 .await 405 405 - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; 371 371 + .map_err(|_| AppError::Internal("Database error".to_string()))?; 406 372 407 373 // Special handling: network.slices.lexicon is always allowed as it defines the schema 408 374 if collection != "network.slices.lexicon" && !slice_collections.contains(&collection) { 409 409 - return Err(StatusCode::NOT_FOUND); 375 375 + return Err(AppError::NotFound("Collection not found".to_string())); 410 376 } 411 377 412 378 // Add collection filter to where conditions ··· 452 418 collection: String, 453 419 state: AppState, 454 420 body: serde_json::Value, 455 455 - ) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> { 421 421 + ) -> Result<Json<serde_json::Value>, AppError> { 456 422 // Parse the JSON body into SliceRecordsParams 457 457 - let mut records_params: SliceRecordsParams = serde_json::from_value(body).map_err(|_| { 458 458 - ( 459 459 - StatusCode::BAD_REQUEST, 460 460 - Json(serde_json::json!({"error": "Invalid request body"})), 461 461 - ) 462 462 - })?; 423 423 + let mut records_params: SliceRecordsParams = serde_json::from_value(body).map_err(|_| AppError::BadRequest("Invalid request body".to_string()))?; 463 424 464 425 // First verify the collection belongs to this slice 465 426 let slice_collections = state 466 427 .database 467 428 .get_slice_collections_list(&records_params.slice) 468 429 .await 469 469 - .map_err(|_| { 470 470 - ( 471 471 - StatusCode::INTERNAL_SERVER_ERROR, 472 472 - Json(serde_json::json!({"error": "Database error"})), 473 473 - ) 474 474 - })?; 430 430 + .map_err(|_| AppError::Internal("Database error".to_string()))?; 475 431 476 432 // Special handling: network.slices.lexicon is always allowed as it defines the schema 477 433 if collection != "network.slices.lexicon" && !slice_collections.contains(&collection) { 478 478 - return Err(( 479 479 - StatusCode::NOT_FOUND, 480 480 - Json(serde_json::json!({"error": "Collection not found"})), 481 481 - )); 434 434 + return Err(AppError::NotFound("Collection not found".to_string())); 482 435 } 483 436 484 437 // Add collection filter to where conditions ··· 525 478 headers: HeaderMap, 526 479 body: serde_json::Value, 527 480 collection: String, 528 528 - ) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> { 481 481 + ) -> Result<Json<serde_json::Value>, AppError> { 529 482 // Extract and verify OAuth token 530 530 - let token = extract_bearer_token(&headers).map_err(status_to_error_response)?; 483 483 + let token = extract_bearer_token(&headers).map_err(|_| AppError::AuthRequired("Missing bearer token".to_string()))?; 531 484 let user_info = verify_oauth_token_cached( 532 485 &token, 533 486 &state.config.auth_base_url, 534 487 Some(state.auth_cache.clone()), 535 488 ) 536 489 .await 537 537 - .map_err(status_to_error_response)?; 490 490 + .map_err(|_| AppError::AuthRequired("Invalid token".to_string()))?; 538 491 539 492 // Get AT Protocol DPoP auth and PDS URL (with caching) 540 493 let (dpop_auth, pds_url) = get_atproto_auth_for_user_cached( ··· 543 496 Some(state.auth_cache.clone()), 544 497 ) 545 498 .await 546 546 - .map_err(status_to_error_response)?; 499 499 + .map_err(|_| AppError::AuthRequired("Invalid token".to_string()))?; 547 500 548 501 // Extract the repo DID from user info 549 502 let repo = user_info.did.unwrap_or(user_info.sub); ··· 555 508 let slice_uri = body 556 509 .get("slice") 557 510 .and_then(|v| v.as_str()) 558 558 - .ok_or_else(|| status_to_error_response(StatusCode::BAD_REQUEST))? 511 511 + .ok_or_else(|| AppError::BadRequest("Missing parameter".to_string()))? 559 512 .to_string(); 560 513 561 514 let record_key = body ··· 566 519 567 520 let record_data = body 568 521 .get("record") 569 569 - .ok_or_else(|| status_to_error_response(StatusCode::BAD_REQUEST))? 522 522 + .ok_or_else(|| AppError::BadRequest("Missing parameter".to_string()))? 570 523 .clone(); 571 524 572 525 // Validate the record against its lexicon ··· 588 541 if let Err(e) = 589 542 slices_lexicon::validate_record(lexicons, &collection, record_data.clone()) 590 543 { 591 591 - return Err(( 592 592 - StatusCode::BAD_REQUEST, 593 593 - Json(serde_json::json!({ 594 594 - "error": "ValidationError", 595 595 - "message": format!("{}", e) 596 596 - })), 597 597 - )); 544 544 + return Err(AppError::BadRequest(format!("ValidationError: {}", e))); 598 545 } 599 546 } 600 547 _ => { ··· 620 567 create_request, 621 568 ) 622 569 .await 623 623 - .map_err(|_e| status_to_error_response(StatusCode::INTERNAL_SERVER_ERROR))?; 570 570 + .map_err(|_e| AppError::Internal("AT Protocol request failed".to_string()))?; 624 571 625 572 // Extract URI and CID from the response enum 626 573 let (uri, cid) = match result { 627 574 CreateRecordResponse::StrongRef { uri, cid, .. } => (uri, cid), 628 575 CreateRecordResponse::Error(_e) => { 629 629 - return Err(status_to_error_response(StatusCode::INTERNAL_SERVER_ERROR)); 576 576 + return Err(AppError::Internal("AT Protocol response error".to_string())); 630 577 } 631 578 }; 632 579 ··· 656 603 headers: HeaderMap, 657 604 body: serde_json::Value, 658 605 collection: String, 659 659 - ) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> { 606 606 + ) -> Result<Json<serde_json::Value>, AppError> { 660 607 // Extract and verify OAuth token 661 661 - let token = extract_bearer_token(&headers).map_err(status_to_error_response)?; 608 608 + let token = extract_bearer_token(&headers).map_err(|_| AppError::AuthRequired("Missing bearer token".to_string()))?; 662 609 let user_info = verify_oauth_token_cached( 663 610 &token, 664 611 &state.config.auth_base_url, 665 612 Some(state.auth_cache.clone()), 666 613 ) 667 614 .await 668 668 - .map_err(status_to_error_response)?; 615 615 + .map_err(|_| AppError::AuthRequired("Invalid token".to_string()))?; 669 616 670 617 // Get AT Protocol DPoP auth and PDS URL (with caching) 671 618 let (dpop_auth, pds_url) = get_atproto_auth_for_user_cached( ··· 674 621 Some(state.auth_cache.clone()), 675 622 ) 676 623 .await 677 677 - .map_err(status_to_error_response)?; 624 624 + .map_err(|_| AppError::AuthRequired("Invalid token".to_string()))?; 678 625 679 626 // Extract slice URI, rkey, and record value from structured body 680 627 let slice_uri = body 681 628 .get("slice") 682 629 .and_then(|v| v.as_str()) 683 683 - .ok_or_else(|| status_to_error_response(StatusCode::BAD_REQUEST))? 630 630 + .ok_or_else(|| AppError::BadRequest("Missing parameter".to_string()))? 684 631 .to_string(); 685 632 686 633 let rkey = body 687 634 .get("rkey") 688 635 .and_then(|v| v.as_str()) 689 689 - .ok_or_else(|| status_to_error_response(StatusCode::BAD_REQUEST))? 636 636 + .ok_or_else(|| AppError::BadRequest("Missing parameter".to_string()))? 690 637 .to_string(); 691 638 692 639 let record_data = body 693 640 .get("record") 694 694 - .ok_or_else(|| status_to_error_response(StatusCode::BAD_REQUEST))? 641 641 + .ok_or_else(|| AppError::BadRequest("Missing parameter".to_string()))? 695 642 .clone(); 696 643 697 644 // Extract repo from user info ··· 716 663 if let Err(e) = 717 664 slices_lexicon::validate_record(lexicons, &collection, record_data.clone()) 718 665 { 719 719 - return Err(( 720 720 - StatusCode::BAD_REQUEST, 721 721 - Json(serde_json::json!({ 722 722 - "error": "ValidationError", 723 723 - "message": format!("{}", e) 724 724 - })), 725 725 - )); 666 666 + return Err(AppError::BadRequest(format!("ValidationError: {}", e))); 726 667 } 727 668 } 728 669 _ => { ··· 751 692 put_request, 752 693 ) 753 694 .await 754 754 - .map_err(|_| status_to_error_response(StatusCode::INTERNAL_SERVER_ERROR))?; 695 695 + .map_err(|_| AppError::Internal("AT Protocol request failed".to_string()))?; 755 696 756 697 // Extract URI and CID from the response enum 757 698 let (uri, cid) = match result { 758 699 PutRecordResponse::StrongRef { uri, cid, .. } => (uri, cid), 759 700 PutRecordResponse::Error(_) => { 760 760 - return Err(status_to_error_response(StatusCode::INTERNAL_SERVER_ERROR)); 701 701 + return Err(AppError::Internal("AT Protocol response error".to_string())); 761 702 } 762 703 }; 763 704 ··· 787 728 headers: HeaderMap, 788 729 body: serde_json::Value, 789 730 collection: String, 790 790 - ) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> { 731 731 + ) -> Result<Json<serde_json::Value>, AppError> { 791 732 // Extract and verify OAuth token 792 792 - let token = extract_bearer_token(&headers).map_err(status_to_error_response)?; 733 733 + let token = extract_bearer_token(&headers).map_err(|_| AppError::AuthRequired("Missing bearer token".to_string()))?; 793 734 let user_info = verify_oauth_token_cached( 794 735 &token, 795 736 &state.config.auth_base_url, 796 737 Some(state.auth_cache.clone()), 797 738 ) 798 739 .await 799 799 - .map_err(status_to_error_response)?; 740 740 + .map_err(|_| AppError::AuthRequired("Invalid token".to_string()))?; 800 741 801 742 // Get AT Protocol DPoP auth and PDS URL (with caching) 802 743 let (dpop_auth, pds_url) = get_atproto_auth_for_user_cached( ··· 805 746 Some(state.auth_cache.clone()), 806 747 ) 807 748 .await 808 808 - .map_err(status_to_error_response)?; 749 749 + .map_err(|_| AppError::AuthRequired("Invalid token".to_string()))?; 809 750 810 751 // Extract repo and rkey from body 811 752 let repo = user_info.did.unwrap_or(user_info.sub); 812 753 let rkey = body["rkey"] 813 754 .as_str() 814 814 - .ok_or_else(|| status_to_error_response(StatusCode::BAD_REQUEST))? 755 755 + .ok_or_else(|| AppError::BadRequest("Missing parameter".to_string()))? 815 756 .to_string(); 816 757 817 758 // Create HTTP client ··· 833 774 delete_request, 834 775 ) 835 776 .await 836 836 - .map_err(|_| status_to_error_response(StatusCode::INTERNAL_SERVER_ERROR))?; 777 777 + .map_err(|_| AppError::Internal("AT Protocol request failed".to_string()))?; 837 778 838 779 // Also delete from local database (from all slices) 839 780 let uri = format!("at://{}/{}/{}", repo, collection, rkey);

+400 -43

api/src/database/cursor.rs

reviewed

··· 1 1 //! Cursor-based pagination utilities. 2 2 //! 3 3 - //! Cursors encode the position in a result set as base64(sort_value::indexed_at::cid) 3 3 + //! Cursors encode the position in a result set as base64(field1|field2|...|cid) 4 4 //! to enable stable pagination even when new records are inserted. 5 5 + //! 6 6 + //! The cursor format: 7 7 + //! - All sort field values are included in the cursor 8 8 + //! - Values are separated by pipe (|) characters 9 9 + //! - CID is always the last element as the ultimate tiebreaker 5 10 6 11 use super::types::SortField; 7 12 use crate::models::Record; 8 13 use base64::{Engine as _, engine::general_purpose}; 9 14 10 10 - /// Generates a base64-encoded cursor from sort value, timestamp, and CID. 15 15 + /// Generates a cursor from a record based on the sort configuration. 11 16 /// 12 12 - /// The cursor format is: `base64(sort_value::indexed_at::cid)` 17 17 + /// Extracts all sort field values from the record and encodes them along with the CID. 18 18 + /// Format: `base64(field1_value|field2_value|...|cid)` 13 19 /// 14 20 /// # Arguments 15 15 - /// * `sort_value` - The value of the primary sort field 16 16 - /// * `indexed_at` - The timestamp when the record was indexed 17 17 - /// * `cid` - The content identifier (CID) of the record 21 21 + /// * `record` - The record to generate a cursor for 22 22 + /// * `sort_by` - Optional array defining sort fields 18 23 /// 19 24 /// # Returns 20 25 /// Base64-encoded cursor string 21 21 - pub fn generate_cursor( 22 22 - sort_value: &str, 23 23 - indexed_at: chrono::DateTime<chrono::Utc>, 24 24 - cid: &str, 25 25 - ) -> String { 26 26 - let cursor_content = format!("{}::{}::{}", sort_value, indexed_at.to_rfc3339(), cid); 26 26 + pub fn generate_cursor_from_record(record: &Record, sort_by: Option<&Vec<SortField>>) -> String { 27 27 + let mut cursor_parts = Vec::new(); 28 28 + 29 29 + // Extract values for all sort fields 30 30 + if let Some(sort_fields) = sort_by { 31 31 + for sort_field in sort_fields { 32 32 + let field_value = extract_field_value(record, &sort_field.field); 33 33 + cursor_parts.push(field_value); 34 34 + } 35 35 + } 36 36 + 37 37 + // Always add CID as the final tiebreaker 38 38 + cursor_parts.push(record.cid.clone()); 39 39 + 40 40 + // Join with pipe and encode 41 41 + let cursor_content = cursor_parts.join("|"); 27 42 general_purpose::URL_SAFE_NO_PAD.encode(cursor_content) 28 43 } 29 44 30 30 - /// Extracts the primary sort field name from a sort array. 45 45 + /// Extracts a field value from a record. 46 46 + /// 47 47 + /// Handles both table columns and JSON fields with nested paths. 48 48 + fn extract_field_value(record: &Record, field: &str) -> String { 49 49 + match field { 50 50 + "indexed_at" => record.indexed_at.to_rfc3339(), 51 51 + "uri" => record.uri.clone(), 52 52 + "cid" => record.cid.clone(), 53 53 + "did" => record.did.clone(), 54 54 + "collection" => record.collection.clone(), 55 55 + _ => { 56 56 + // Handle nested JSON paths 57 57 + let field_path: Vec<&str> = field.split('.').collect(); 58 58 + let mut value = &record.json; 59 59 + 60 60 + for key in &field_path { 61 61 + value = match value.get(key) { 62 62 + Some(v) => v, 63 63 + None => return "NULL".to_string(), 64 64 + }; 65 65 + } 66 66 + 67 67 + match value { 68 68 + serde_json::Value::String(s) => s.clone(), 69 69 + serde_json::Value::Number(n) => n.to_string(), 70 70 + serde_json::Value::Bool(b) => b.to_string(), 71 71 + serde_json::Value::Null => "NULL".to_string(), 72 72 + _ => "NULL".to_string(), 73 73 + } 74 74 + } 75 75 + } 76 76 + } 77 77 + 78 78 + /// Decoded cursor components for pagination. 79 79 + #[derive(Debug, Clone)] 80 80 + pub struct DecodedCursor { 81 81 + /// Field values in the order they appear in sortBy 82 82 + pub field_values: Vec<String>, 83 83 + /// CID (always the last element) 84 84 + pub cid: String, 85 85 + } 86 86 + 87 87 + /// Decodes a base64-encoded cursor back into its components. 31 88 /// 32 32 - /// Returns "indexed_at" if no sort fields are provided. 89 89 + /// The cursor format is: `base64(field1|field2|...|cid)` 33 90 /// 34 91 /// # Arguments 35 35 - /// * `sort_by` - Optional array of sort fields 92 92 + /// * `cursor` - Base64-encoded cursor string 93 93 + /// * `sort_by` - Optional array of sort fields to validate cursor format 36 94 /// 37 95 /// # Returns 38 38 - /// The name of the primary sort field 39 39 - pub fn get_primary_sort_field(sort_by: Option<&Vec<SortField>>) -> String { 40 40 - match sort_by { 41 41 - Some(sort_fields) if !sort_fields.is_empty() => sort_fields[0].field.clone(), 42 42 - _ => "indexed_at".to_string(), 96 96 + /// Result containing DecodedCursor or error if decoding fails 97 97 + pub fn decode_cursor(cursor: &str, sort_by: Option<&Vec<SortField>>) -> Result<DecodedCursor, String> { 98 98 + let decoded_bytes = general_purpose::URL_SAFE_NO_PAD.decode(cursor) 99 99 + .map_err(|e| format!("Failed to decode base64: {}", e))?; 100 100 + let decoded_str = String::from_utf8(decoded_bytes) 101 101 + .map_err(|e| format!("Invalid UTF-8 in cursor: {}", e))?; 102 102 + 103 103 + let parts: Vec<&str> = decoded_str.split('|').collect(); 104 104 + 105 105 + // Validate cursor format matches sortBy fields 106 106 + let expected_parts = if let Some(fields) = sort_by { 107 107 + fields.len() + 1 // sort fields + CID 108 108 + } else { 109 109 + 1 // just CID if no sortBy 110 110 + }; 111 111 + 112 112 + if parts.len() != expected_parts { 113 113 + return Err(format!( 114 114 + "Invalid cursor format: expected {} parts, got {}", 115 115 + expected_parts, 116 116 + parts.len() 117 117 + )); 43 118 } 119 119 + 120 120 + let cid = parts[parts.len() - 1].to_string(); 121 121 + let field_values: Vec<String> = parts[..parts.len() - 1] 122 122 + .iter() 123 123 + .map(|s| s.to_string()) 124 124 + .collect(); 125 125 + 126 126 + Ok(DecodedCursor { 127 127 + field_values, 128 128 + cid, 129 129 + }) 44 130 } 45 131 46 46 - /// Generates a cursor from a record based on the sort configuration. 132 132 + /// Builds cursor-based WHERE conditions for proper multi-field pagination. 47 133 /// 48 48 - /// Extracts the sort value from the record (either from a table column 49 49 - /// or from the JSON field), then creates a cursor encoding that value 50 50 - /// along with indexed_at and cid. 134 134 + /// Creates progressive equality checks for stable multi-field sorting. 135 135 + /// For each field, we OR together: 136 136 + /// 1. field1 > cursor_value1 137 137 + /// 2. field1 = cursor_value1 AND field2 > cursor_value2 138 138 + /// 3. field1 = cursor_value1 AND field2 = cursor_value2 AND field3 > cursor_value3 139 139 + /// ... and so on 140 140 + /// 141 141 + /// Finally: all fields equal AND cid > cursor_cid 51 142 /// 52 143 /// # Arguments 53 53 - /// * `record` - The record to generate a cursor for 54 54 - /// * `sort_by` - Optional array defining sort fields 144 144 + /// * `decoded_cursor` - The decoded cursor components 145 145 + /// * `sort_by` - Optional array of sort fields 146 146 + /// * `param_count` - Mutable counter for parameter numbering 147 147 + /// * `field_types` - Optional array indicating if each field is a datetime 55 148 /// 56 149 /// # Returns 57 57 - /// Base64-encoded cursor string 58 58 - pub fn generate_cursor_from_record(record: &Record, sort_by: Option<&Vec<SortField>>) -> String { 59 59 - let primary_sort_field = get_primary_sort_field(sort_by); 150 150 + /// Tuple of (where_condition_sql, bind_values) 151 151 + pub fn build_cursor_where_condition( 152 152 + decoded_cursor: &DecodedCursor, 153 153 + sort_by: Option<&Vec<SortField>>, 154 154 + param_count: &mut usize, 155 155 + field_types: Option<&[bool]>, 156 156 + ) -> (String, Vec<String>) { 157 157 + let mut bind_values = Vec::new(); 158 158 + let mut clauses = Vec::new(); 159 159 + 160 160 + let sort_fields = match sort_by { 161 161 + Some(fields) if !fields.is_empty() => fields, 162 162 + _ => { 163 163 + // No sort fields, shouldn't happen but handle gracefully 164 164 + return ("1=1".to_string(), vec![]); 165 165 + } 166 166 + }; 167 167 + 168 168 + // Build progressive equality checks for each level 169 169 + for i in 0..sort_fields.len() { 170 170 + let mut clause_parts = Vec::new(); 171 171 + 172 172 + // Add equality checks for all previous fields 173 173 + for (j, sort_field) in sort_fields.iter().enumerate().take(i) { 174 174 + let field = &sort_field.field; 175 175 + let cursor_value = &decoded_cursor.field_values[j]; 176 176 + let is_datetime = field_types.and_then(|types| types.get(j).copied()).unwrap_or(false); 177 177 + 178 178 + let field_ref = build_field_reference(field, is_datetime); 179 179 + let param_cast = if is_datetime { "::timestamp" } else { "" }; 180 180 + 181 181 + clause_parts.push(format!("{} = ${}{}", field_ref, param_count, param_cast)); 182 182 + *param_count += 1; 183 183 + bind_values.push(cursor_value.clone()); 184 184 + } 185 185 + 186 186 + // Add comparison for current field 187 187 + let field = &sort_fields[i].field; 188 188 + let cursor_value = &decoded_cursor.field_values[i]; 189 189 + let direction = &sort_fields[i].direction; 190 190 + let is_datetime = field_types.and_then(|types| types.get(i).copied()).unwrap_or(false); 191 191 + 192 192 + let comparison_op = if direction.to_lowercase() == "desc" { "<" } else { ">" }; 193 193 + let field_ref = build_field_reference(field, is_datetime); 194 194 + let param_cast = if is_datetime { "::timestamp" } else { "" }; 195 195 + 196 196 + clause_parts.push(format!("{} {} ${}{}", field_ref, comparison_op, param_count, param_cast)); 197 197 + *param_count += 1; 198 198 + bind_values.push(cursor_value.clone()); 199 199 + 200 200 + // Combine with AND 201 201 + clauses.push(format!("({})", clause_parts.join(" AND "))); 202 202 + } 203 203 + 204 204 + // Add final clause: all fields equal AND cid comparison 205 205 + let mut final_clause_parts = Vec::new(); 206 206 + for (j, field) in sort_fields.iter().enumerate() { 207 207 + let cursor_value = &decoded_cursor.field_values[j]; 208 208 + let is_datetime = field_types.and_then(|types| types.get(j).copied()).unwrap_or(false); 209 209 + 210 210 + let field_ref = build_field_reference(&field.field, is_datetime); 211 211 + let param_cast = if is_datetime { "::timestamp" } else { "" }; 212 212 + 213 213 + final_clause_parts.push(format!("{} = ${}{}", field_ref, param_count, param_cast)); 214 214 + *param_count += 1; 215 215 + bind_values.push(cursor_value.clone()); 216 216 + } 217 217 + 218 218 + // CID comparison uses the direction of the last sort field 219 219 + let last_direction = &sort_fields[sort_fields.len() - 1].direction; 220 220 + let cid_comparison_op = if last_direction.to_lowercase() == "desc" { "<" } else { ">" }; 221 221 + 222 222 + final_clause_parts.push(format!("cid {} ${}", cid_comparison_op, param_count)); 223 223 + *param_count += 1; 224 224 + bind_values.push(decoded_cursor.cid.clone()); 225 225 + 226 226 + clauses.push(format!("({})", final_clause_parts.join(" AND "))); 60 227 61 61 - let sort_value = match primary_sort_field.as_str() { 62 62 - "indexed_at" => record.indexed_at.to_rfc3339(), 63 63 - field => record 64 64 - .json 65 65 - .get(field) 66 66 - .and_then(|v| match v { 67 67 - serde_json::Value::String(s) if !s.is_empty() => Some(s.clone()), 68 68 - serde_json::Value::Number(n) => Some(n.to_string()), 69 69 - serde_json::Value::Bool(b) => Some(b.to_string()), 70 70 - serde_json::Value::Null => None, 71 71 - _ => None, 72 72 - }) 73 73 - .unwrap_or_else(|| "NULL".to_string()), 228 228 + // Combine all clauses with OR 229 229 + let where_condition = format!("({})", clauses.join(" OR ")); 230 230 + 231 231 + (where_condition, bind_values) 232 232 + } 233 233 + 234 234 + /// Builds a field reference for SQL queries. 235 235 + /// 236 236 + /// Handles table columns, JSON fields, and nested paths with optional timestamp casting. 237 237 + pub(super) fn build_field_reference(field: &str, is_datetime: bool) -> String { 238 238 + // Table columns don't need JSON extraction 239 239 + if matches!(field, "uri" | "cid" | "did" | "collection" | "indexed_at") { 240 240 + return field.to_string(); 241 241 + } 242 242 + 243 243 + // Build JSON path for nested or simple fields 244 244 + let json_path = if field.contains('.') { 245 245 + let parts: Vec<&str> = field.split('.').collect(); 246 246 + let mut path = String::from("json"); 247 247 + for (i, part) in parts.iter().enumerate() { 248 248 + if i == parts.len() - 1 { 249 249 + path.push_str(&format!("->>'{}'", part)); 250 250 + } else { 251 251 + path.push_str(&format!("->'{}'", part)); 252 252 + } 253 253 + } 254 254 + path 255 255 + } else { 256 256 + format!("json->>'{}'", field) 74 257 }; 75 258 76 76 - generate_cursor(&sort_value, record.indexed_at, &record.cid) 259 259 + // Add timestamp cast if needed 260 260 + if is_datetime { 261 261 + format!("({})::timestamp", json_path) 262 262 + } else { 263 263 + json_path 264 264 + } 77 265 } 266 266 + 267 267 + #[cfg(test)] 268 268 + mod tests { 269 269 + use super::*; 270 270 + use chrono::Utc; 271 271 + 272 272 + fn create_test_record() -> Record { 273 273 + Record { 274 274 + uri: "at://did:plc:test/app.bsky.feed.post/123".to_string(), 275 275 + cid: "bafytest123".to_string(), 276 276 + did: "did:plc:test".to_string(), 277 277 + collection: "app.bsky.feed.post".to_string(), 278 278 + json: serde_json::json!({ 279 279 + "text": "Hello world", 280 280 + "createdAt": "2025-01-15T12:00:00Z", 281 281 + "nested": { 282 282 + "field": "value" 283 283 + } 284 284 + }), 285 285 + indexed_at: Utc::now(), 286 286 + slice_uri: Some("at://did:plc:slice/network.slices.slice/abc".to_string()), 287 287 + } 288 288 + } 289 289 + 290 290 + #[test] 291 291 + fn test_generate_cursor_no_sort() { 292 292 + let record = create_test_record(); 293 293 + let cursor = generate_cursor_from_record(&record, None); 294 294 + 295 295 + let decoded = general_purpose::URL_SAFE_NO_PAD.decode(&cursor).unwrap(); 296 296 + let decoded_str = String::from_utf8(decoded).unwrap(); 297 297 + 298 298 + assert_eq!(decoded_str, "bafytest123"); 299 299 + } 300 300 + 301 301 + #[test] 302 302 + fn test_generate_cursor_with_sort() { 303 303 + let record = create_test_record(); 304 304 + let sort_by = vec![ 305 305 + SortField { 306 306 + field: "text".to_string(), 307 307 + direction: "desc".to_string(), 308 308 + }, 309 309 + ]; 310 310 + 311 311 + let cursor = generate_cursor_from_record(&record, Some(&sort_by)); 312 312 + let decoded = general_purpose::URL_SAFE_NO_PAD.decode(&cursor).unwrap(); 313 313 + let decoded_str = String::from_utf8(decoded).unwrap(); 314 314 + 315 315 + assert_eq!(decoded_str, "Hello world|bafytest123"); 316 316 + } 317 317 + 318 318 + #[test] 319 319 + fn test_decode_cursor_single_field() { 320 320 + let sort_by = vec![ 321 321 + SortField { 322 322 + field: "createdAt".to_string(), 323 323 + direction: "desc".to_string(), 324 324 + }, 325 325 + ]; 326 326 + 327 327 + let cursor_content = "2025-01-15T12:00:00Z|bafytest123"; 328 328 + let cursor = general_purpose::URL_SAFE_NO_PAD.encode(cursor_content); 329 329 + 330 330 + let decoded = decode_cursor(&cursor, Some(&sort_by)).unwrap(); 331 331 + 332 332 + assert_eq!(decoded.field_values, vec!["2025-01-15T12:00:00Z"]); 333 333 + assert_eq!(decoded.cid, "bafytest123"); 334 334 + } 335 335 + 336 336 + #[test] 337 337 + fn test_decode_cursor_multiple_fields() { 338 338 + let sort_by = vec![ 339 339 + SortField { 340 340 + field: "text".to_string(), 341 341 + direction: "desc".to_string(), 342 342 + }, 343 343 + SortField { 344 344 + field: "createdAt".to_string(), 345 345 + direction: "desc".to_string(), 346 346 + }, 347 347 + ]; 348 348 + 349 349 + let cursor_content = "Hello world|2025-01-15T12:00:00Z|bafytest123"; 350 350 + let cursor = general_purpose::URL_SAFE_NO_PAD.encode(cursor_content); 351 351 + 352 352 + let decoded = decode_cursor(&cursor, Some(&sort_by)).unwrap(); 353 353 + 354 354 + assert_eq!(decoded.field_values, vec!["Hello world", "2025-01-15T12:00:00Z"]); 355 355 + assert_eq!(decoded.cid, "bafytest123"); 356 356 + } 357 357 + 358 358 + #[test] 359 359 + fn test_decode_cursor_invalid_format() { 360 360 + let sort_by = vec![ 361 361 + SortField { 362 362 + field: "text".to_string(), 363 363 + direction: "desc".to_string(), 364 364 + }, 365 365 + ]; 366 366 + 367 367 + let cursor_content = "bafytest123"; 368 368 + let cursor = general_purpose::URL_SAFE_NO_PAD.encode(cursor_content); 369 369 + 370 370 + let result = decode_cursor(&cursor, Some(&sort_by)); 371 371 + assert!(result.is_err()); 372 372 + } 373 373 + 374 374 + #[test] 375 375 + fn test_build_field_reference_table_column() { 376 376 + assert_eq!(build_field_reference("uri", false), "uri"); 377 377 + assert_eq!(build_field_reference("cid", false), "cid"); 378 378 + assert_eq!(build_field_reference("indexed_at", false), "indexed_at"); 379 379 + } 380 380 + 381 381 + #[test] 382 382 + fn test_build_field_reference_json_field() { 383 383 + assert_eq!(build_field_reference("text", false), "json->>'text'"); 384 384 + assert_eq!( 385 385 + build_field_reference("text", true), 386 386 + "(json->>'text')::timestamp" 387 387 + ); 388 388 + } 389 389 + 390 390 + #[test] 391 391 + fn test_build_field_reference_nested_json() { 392 392 + assert_eq!( 393 393 + build_field_reference("nested.field", false), 394 394 + "json->'nested'->>'field'" 395 395 + ); 396 396 + assert_eq!( 397 397 + build_field_reference("nested.field", true), 398 398 + "(json->'nested'->>'field')::timestamp" 399 399 + ); 400 400 + } 401 401 + 402 402 + #[test] 403 403 + fn test_extract_field_value_table_columns() { 404 404 + let record = create_test_record(); 405 405 + 406 406 + assert_eq!(extract_field_value(&record, "uri"), record.uri); 407 407 + assert_eq!(extract_field_value(&record, "cid"), record.cid); 408 408 + assert_eq!(extract_field_value(&record, "did"), record.did); 409 409 + assert_eq!(extract_field_value(&record, "collection"), record.collection); 410 410 + } 411 411 + 412 412 + #[test] 413 413 + fn test_extract_field_value_json() { 414 414 + let record = create_test_record(); 415 415 + 416 416 + assert_eq!(extract_field_value(&record, "text"), "Hello world"); 417 417 + assert_eq!(extract_field_value(&record, "createdAt"), "2025-01-15T12:00:00Z"); 418 418 + } 419 419 + 420 420 + #[test] 421 421 + fn test_extract_field_value_nested_json() { 422 422 + let record = create_test_record(); 423 423 + 424 424 + assert_eq!(extract_field_value(&record, "nested.field"), "value"); 425 425 + } 426 426 + 427 427 + #[test] 428 428 + fn test_extract_field_value_missing() { 429 429 + let record = create_test_record(); 430 430 + 431 431 + assert_eq!(extract_field_value(&record, "nonexistent"), "NULL"); 432 432 + assert_eq!(extract_field_value(&record, "nested.nonexistent"), "NULL"); 433 433 + } 434 434 + }

+92 -28

api/src/database/query_builder.rs

reviewed

··· 3 3 //! This module provides helpers for constructing SQL queries dynamically 4 4 //! based on user input while preventing SQL injection attacks. 5 5 6 6 + use super::cursor::build_field_reference; 6 7 use super::types::{SortField, WhereClause, WhereCondition}; 7 8 use crate::models::Record; 8 9 9 9 - /// Builds an ORDER BY clause from an optional array of sort fields. 10 10 - /// 11 11 - /// Handles both table columns (indexed_at, uri, cid, did, collection) 12 12 - /// and JSON fields with nested paths. Always adds indexed_at as a 13 13 - /// tie-breaker if not already present. 10 10 + 11 11 + /// Builds an ORDER BY clause with optional datetime field information. 14 12 /// 15 13 /// # Arguments 16 14 /// * `sort_by` - Optional array of fields to sort by 15 15 + /// * `field_is_datetime` - Optional boolean indicating if primary field is datetime 17 16 /// 18 17 /// # Returns 19 18 /// SQL ORDER BY clause string (without "ORDER BY" prefix) 20 20 - pub fn build_order_by_clause(sort_by: Option<&Vec<SortField>>) -> String { 19 19 + pub fn build_order_by_clause_with_field_info( 20 20 + sort_by: Option<&Vec<SortField>>, 21 21 + field_is_datetime: Option<bool>, 22 22 + ) -> String { 21 23 match sort_by { 22 24 Some(sort_fields) if !sort_fields.is_empty() => { 23 25 let mut order_clauses = Vec::new(); 24 24 - for sort_field in sort_fields { 26 26 + for (index, sort_field) in sort_fields.iter().enumerate() { 25 27 let field = &sort_field.field; 26 28 let direction = match sort_field.direction.to_lowercase().as_str() { 27 29 "desc" => "DESC", 28 30 _ => "ASC", 29 31 }; 30 32 33 33 + let is_primary = index == 0; 34 34 + let is_datetime = is_primary && field_is_datetime == Some(true); 35 35 + 31 36 if field 32 37 .chars() 33 38 .all(|c| c.is_alphanumeric() || c == '_' || c == '.') 34 39 { 35 35 - if field == "indexed_at" 36 36 - || field == "uri" 37 37 - || field == "cid" 38 38 - || field == "did" 39 39 - || field == "collection" 40 40 - { 41 41 - order_clauses.push(format!("{field} {direction}")); 40 40 + let field_ref = build_field_reference(field, is_datetime); 41 41 + 42 42 + if matches!(field.as_str(), "indexed_at" | "uri" | "cid" | "did" | "collection") { 43 43 + order_clauses.push(format!("{field_ref} {direction}")); 42 44 } else { 43 43 - if field.contains('.') { 44 44 - let parts: Vec<&str> = field.split('.').collect(); 45 45 - let mut path = String::from("json"); 46 46 - for (i, part) in parts.iter().enumerate() { 47 47 - if i == parts.len() - 1 { 48 48 - path.push_str(&format!("->>'{}'", part)); 49 49 - } else { 50 50 - path.push_str(&format!("->'{}'", part)); 51 51 - } 52 52 - } 53 53 - order_clauses.push(format!("{path} {direction} NULLS LAST")); 54 54 - } else { 55 55 - order_clauses.push(format!("json->>'{field}' {direction} NULLS LAST")); 56 56 - } 45 45 + order_clauses.push(format!("{field_ref} {direction} NULLS LAST")); 57 46 } 58 47 } 59 48 } ··· 251 240 252 241 query_builder 253 242 } 243 243 + 244 244 + #[cfg(test)] 245 245 + mod tests { 246 246 + use super::*; 247 247 + 248 248 + #[test] 249 249 + fn test_build_order_by_clause_default() { 250 250 + let result = build_order_by_clause_with_field_info(None, None); 251 251 + assert_eq!(result, "indexed_at DESC"); 252 252 + } 253 253 + 254 254 + #[test] 255 255 + fn test_build_order_by_clause_single_field() { 256 256 + let sort_by = vec![SortField { 257 257 + field: "createdAt".to_string(), 258 258 + direction: "desc".to_string(), 259 259 + }]; 260 260 + 261 261 + let result = build_order_by_clause_with_field_info(Some(&sort_by), None); 262 262 + assert_eq!(result, "json->>'createdAt' DESC NULLS LAST, indexed_at DESC"); 263 263 + } 264 264 + 265 265 + #[test] 266 266 + fn test_build_order_by_clause_datetime_field() { 267 267 + let sort_by = vec![SortField { 268 268 + field: "createdAt".to_string(), 269 269 + direction: "desc".to_string(), 270 270 + }]; 271 271 + 272 272 + let result = build_order_by_clause_with_field_info(Some(&sort_by), Some(true)); 273 273 + assert_eq!(result, "(json->>'createdAt')::timestamp DESC NULLS LAST, indexed_at DESC"); 274 274 + } 275 275 + 276 276 + #[test] 277 277 + fn test_build_order_by_clause_table_column() { 278 278 + let sort_by = vec![SortField { 279 279 + field: "indexed_at".to_string(), 280 280 + direction: "asc".to_string(), 281 281 + }]; 282 282 + 283 283 + let result = build_order_by_clause_with_field_info(Some(&sort_by), None); 284 284 + assert_eq!(result, "indexed_at ASC"); 285 285 + } 286 286 + 287 287 + #[test] 288 288 + fn test_build_order_by_clause_nested_json() { 289 289 + let sort_by = vec![SortField { 290 290 + field: "author.name".to_string(), 291 291 + direction: "asc".to_string(), 292 292 + }]; 293 293 + 294 294 + let result = build_order_by_clause_with_field_info(Some(&sort_by), None); 295 295 + assert_eq!(result, "json->'author'->>'name' ASC NULLS LAST, indexed_at DESC"); 296 296 + } 297 297 + 298 298 + #[test] 299 299 + fn test_build_order_by_clause_multiple_fields() { 300 300 + let sort_by = vec![ 301 301 + SortField { 302 302 + field: "text".to_string(), 303 303 + direction: "desc".to_string(), 304 304 + }, 305 305 + SortField { 306 306 + field: "createdAt".to_string(), 307 307 + direction: "asc".to_string(), 308 308 + }, 309 309 + ]; 310 310 + 311 311 + let result = build_order_by_clause_with_field_info(Some(&sort_by), None); 312 312 + assert_eq!( 313 313 + result, 314 314 + "json->>'text' DESC NULLS LAST, json->>'createdAt' ASC NULLS LAST, indexed_at DESC" 315 315 + ); 316 316 + } 317 317 + }

+239 -15

api/src/database/records.rs

reviewed

··· 5 5 //! sorting, and pagination. 6 6 7 7 use super::client::Database; 8 8 - use super::cursor::generate_cursor_from_record; 9 9 - use super::query_builder::{bind_where_parameters, build_order_by_clause, build_where_conditions}; 8 8 + use super::cursor::{build_cursor_where_condition, decode_cursor, generate_cursor_from_record}; 9 9 + use super::query_builder::{bind_where_parameters, build_order_by_clause_with_field_info, build_where_conditions}; 10 10 use super::types::{SortField, WhereClause}; 11 11 use crate::errors::DatabaseError; 12 12 use crate::models::{IndexedRecord, Record}; ··· 15 15 /// Inserts a single record into the database. 16 16 /// 17 17 /// Uses ON CONFLICT to update existing records with matching URI and slice_uri. 18 18 - #[allow(dead_code)] 19 18 pub async fn insert_record(&self, record: &Record) -> Result<(), DatabaseError> { 20 19 sqlx::query!( 21 20 r#"INSERT INTO "record" ("uri", "cid", "did", "collection", "json", "indexed_at", "slice_uri") ··· 211 210 where_clause: Option<&WhereClause>, 212 211 ) -> Result<(Vec<Record>, Option<String>), DatabaseError> { 213 212 let limit = limit.unwrap_or(50).min(100); 214 214 - let order_by = build_order_by_clause(sort_by); 215 213 216 214 let mut where_clauses = Vec::new(); 217 215 let mut param_count = 1; ··· 223 221 .and_then(|v| v.as_str()) 224 222 == Some("network.slices.lexicon"); 225 223 224 224 + // Extract collection name from where clause for lexicon lookup 225 225 + let collection = where_clause 226 226 + .as_ref() 227 227 + .and_then(|wc| wc.conditions.get("collection")) 228 228 + .and_then(|c| c.eq.as_ref()) 229 229 + .and_then(|v| v.as_str()); 230 230 + 231 231 + // Determine which sort fields are datetime fields by checking lexicons 232 232 + let field_types: Option<Vec<bool>> = if let Some(collection_name) = collection { 233 233 + if let Some(sort_fields) = sort_by { 234 234 + // Fetch lexicons to check field types for ALL sort fields 235 235 + match self.get_lexicons_by_slice(slice_uri).await { 236 236 + Ok(lexicons) => { 237 237 + let types: Vec<bool> = sort_fields 238 238 + .iter() 239 239 + .map(|field| is_field_datetime(&lexicons, collection_name, &field.field)) 240 240 + .collect(); 241 241 + Some(types) 242 242 + } 243 243 + Err(_) => None, 244 244 + } 245 245 + } else { 246 246 + None 247 247 + } 248 248 + } else { 249 249 + None 250 250 + }; 251 251 + 252 252 + // Get first field type for ORDER BY (for backward compatibility) 253 253 + let primary_field_is_datetime = field_types.as_ref().and_then(|types| types.first().copied()); 254 254 + 255 255 + // Build ORDER BY clause with datetime field information 256 256 + let order_by = build_order_by_clause_with_field_info(sort_by, primary_field_is_datetime); 257 257 + 226 258 if is_lexicon { 227 259 where_clauses.push(format!("json->>'slice' = ${}", param_count)); 228 260 } else { ··· 230 262 } 231 263 param_count += 1; 232 264 233 233 - if cursor.is_some() { 234 234 - where_clauses.push(format!("indexed_at < ${}", param_count)); 235 235 - param_count += 1; 265 265 + // Build all other WHERE conditions first (including collection filter) 266 266 + // For non-lexicon records, exclude the 'slice' field since we handle it via slice_uri 267 267 + let mut filtered_where_clause = None; 268 268 + let filtered_clause; 269 269 + 270 270 + if is_lexicon { 271 271 + filtered_where_clause = where_clause; 272 272 + } else if let Some(wc) = where_clause { 273 273 + let mut filtered_conditions = std::collections::HashMap::new(); 274 274 + for (field, condition) in &wc.conditions { 275 275 + if field != "slice" { 276 276 + filtered_conditions.insert(field.clone(), condition.clone()); 277 277 + } 278 278 + } 279 279 + 280 280 + filtered_clause = WhereClause { 281 281 + conditions: filtered_conditions, 282 282 + or_conditions: wc.or_conditions.clone(), 283 283 + }; 284 284 + filtered_where_clause = Some(&filtered_clause); 236 285 } 237 286 238 287 let (and_conditions, or_conditions) = 239 239 - build_where_conditions(where_clause, &mut param_count); 288 288 + build_where_conditions(filtered_where_clause, &mut param_count); 240 289 where_clauses.extend(and_conditions); 241 290 291 291 + // Add cursor conditions last to ensure proper parameter order 292 292 + let mut cursor_bind_values = Vec::new(); 293 293 + if let Some(cursor_str) = cursor { 294 294 + match decode_cursor(cursor_str, sort_by) { 295 295 + Ok(decoded_cursor) => { 296 296 + // Use the datetime field information we already computed 297 297 + let field_type_slice = field_types.as_deref(); 298 298 + let (cursor_where, bind_values) = 299 299 + build_cursor_where_condition(&decoded_cursor, sort_by, &mut param_count, field_type_slice); 300 300 + where_clauses.push(cursor_where); 301 301 + cursor_bind_values = bind_values; 302 302 + } 303 303 + Err(e) => { 304 304 + // Log the error but don't fail the request 305 305 + eprintln!("Invalid cursor format: {}", e); 306 306 + } 307 307 + } 308 308 + } 309 309 + 242 310 if !or_conditions.is_empty() { 243 311 let or_clause = format!("({})", or_conditions.join(" OR ")); 244 312 where_clauses.push(or_clause); 245 313 } 246 314 247 315 let where_sql = where_clauses.join(" AND "); 316 316 + 317 317 + // Assign limit parameter AFTER all other parameters 318 318 + let limit_param = param_count; 319 319 + 248 320 let query = format!( 249 321 "SELECT uri, cid, did, collection, json, indexed_at, slice_uri 250 322 FROM record 251 323 WHERE {} 252 324 ORDER BY {} 253 325 LIMIT ${}", 254 254 - where_sql, order_by, param_count 326 326 + where_sql, order_by, limit_param 255 327 ); 256 328 257 329 let mut query_builder = sqlx::query_as::<_, Record>(&query); 258 330 259 331 query_builder = query_builder.bind(slice_uri); 260 332 261 261 - if let Some(cursor_time) = cursor { 262 262 - let cursor_dt = cursor_time 263 263 - .parse::<chrono::DateTime<chrono::Utc>>() 264 264 - .unwrap_or_else(|_| chrono::Utc::now()); 265 265 - query_builder = query_builder.bind(cursor_dt); 333 333 + // Bind WHERE condition parameters (including collection filter) 334 334 + query_builder = bind_where_parameters(query_builder, filtered_where_clause); 335 335 + 336 336 + // Bind cursor values after WHERE conditions 337 337 + for cursor_value in cursor_bind_values { 338 338 + query_builder = query_builder.bind(cursor_value); 266 339 } 267 340 268 268 - query_builder = bind_where_parameters(query_builder, where_clause); 269 341 query_builder = query_builder.bind(limit as i64); 270 342 271 343 let records = query_builder.fetch_all(&self.pool).await?; ··· 466 538 Ok(lexicon_definitions) 467 539 } 468 540 } 541 541 + 542 542 + /// Helper function to check if a field is a datetime field in the lexicon 543 543 + fn is_field_datetime(lexicons: &[serde_json::Value], collection: &str, field: &str) -> bool { 544 544 + for lexicon in lexicons { 545 545 + if let Some(id) = lexicon.get("id").and_then(|v| v.as_str()) 546 546 + && id == collection 547 547 + && let Some(defs) = lexicon.get("defs") 548 548 + && let Some(main) = defs.get("main") 549 549 + && let Some(record) = main.get("record") 550 550 + && let Some(properties) = record.get("properties") 551 551 + && let Some(field_def) = properties.get(field) 552 552 + && let Some(format) = field_def.get("format").and_then(|v| v.as_str()) 553 553 + { 554 554 + return format == "datetime"; 555 555 + } 556 556 + } 557 557 + false 558 558 + } 559 559 + 560 560 + #[cfg(test)] 561 561 + mod tests { 562 562 + use super::*; 563 563 + 564 564 + #[test] 565 565 + fn test_is_field_datetime_found() { 566 566 + let lexicons = vec![serde_json::json!({ 567 567 + "lexicon": 1, 568 568 + "id": "app.bsky.feed.post", 569 569 + "defs": { 570 570 + "main": { 571 571 + "record": { 572 572 + "properties": { 573 573 + "createdAt": { 574 574 + "type": "string", 575 575 + "format": "datetime" 576 576 + }, 577 577 + "text": { 578 578 + "type": "string" 579 579 + } 580 580 + } 581 581 + } 582 582 + } 583 583 + } 584 584 + })]; 585 585 + 586 586 + assert!(is_field_datetime(&lexicons, "app.bsky.feed.post", "createdAt")); 587 587 + } 588 588 + 589 589 + #[test] 590 590 + fn test_is_field_datetime_not_datetime() { 591 591 + let lexicons = vec![serde_json::json!({ 592 592 + "lexicon": 1, 593 593 + "id": "app.bsky.feed.post", 594 594 + "defs": { 595 595 + "main": { 596 596 + "record": { 597 597 + "properties": { 598 598 + "text": { 599 599 + "type": "string" 600 600 + } 601 601 + } 602 602 + } 603 603 + } 604 604 + } 605 605 + })]; 606 606 + 607 607 + assert!(!is_field_datetime(&lexicons, "app.bsky.feed.post", "text")); 608 608 + } 609 609 + 610 610 + #[test] 611 611 + fn test_is_field_datetime_missing_field() { 612 612 + let lexicons = vec![serde_json::json!({ 613 613 + "lexicon": 1, 614 614 + "id": "app.bsky.feed.post", 615 615 + "defs": { 616 616 + "main": { 617 617 + "record": { 618 618 + "properties": { 619 619 + "text": { 620 620 + "type": "string" 621 621 + } 622 622 + } 623 623 + } 624 624 + } 625 625 + } 626 626 + })]; 627 627 + 628 628 + assert!(!is_field_datetime(&lexicons, "app.bsky.feed.post", "nonexistent")); 629 629 + } 630 630 + 631 631 + #[test] 632 632 + fn test_is_field_datetime_wrong_collection() { 633 633 + let lexicons = vec![serde_json::json!({ 634 634 + "lexicon": 1, 635 635 + "id": "app.bsky.feed.post", 636 636 + "defs": { 637 637 + "main": { 638 638 + "record": { 639 639 + "properties": { 640 640 + "createdAt": { 641 641 + "type": "string", 642 642 + "format": "datetime" 643 643 + } 644 644 + } 645 645 + } 646 646 + } 647 647 + } 648 648 + })]; 649 649 + 650 650 + assert!(!is_field_datetime(&lexicons, "app.bsky.actor.profile", "createdAt")); 651 651 + } 652 652 + 653 653 + #[test] 654 654 + fn test_is_field_datetime_multiple_lexicons() { 655 655 + let lexicons = vec![ 656 656 + serde_json::json!({ 657 657 + "lexicon": 1, 658 658 + "id": "app.bsky.feed.post", 659 659 + "defs": { 660 660 + "main": { 661 661 + "record": { 662 662 + "properties": { 663 663 + "text": { 664 664 + "type": "string" 665 665 + } 666 666 + } 667 667 + } 668 668 + } 669 669 + } 670 670 + }), 671 671 + serde_json::json!({ 672 672 + "lexicon": 1, 673 673 + "id": "app.bsky.actor.profile", 674 674 + "defs": { 675 675 + "main": { 676 676 + "record": { 677 677 + "properties": { 678 678 + "createdAt": { 679 679 + "type": "string", 680 680 + "format": "datetime" 681 681 + } 682 682 + } 683 683 + } 684 684 + } 685 685 + } 686 686 + }), 687 687 + ]; 688 688 + 689 689 + assert!(is_field_datetime(&lexicons, "app.bsky.actor.profile", "createdAt")); 690 690 + assert!(!is_field_datetime(&lexicons, "app.bsky.feed.post", "text")); 691 691 + } 692 692 + }

+2 -2

api/src/database/types.rs

reviewed

··· 13 13 /// - `eq`: Exact match (field = value) 14 14 /// - `in_values`: Array membership (field IN (...)) 15 15 /// - `contains`: Pattern matching (field ILIKE '%value%') 16 16 - #[derive(Debug, Serialize, Deserialize)] 16 16 + #[derive(Debug, Clone, Serialize, Deserialize)] 17 17 #[serde(rename_all = "camelCase")] 18 18 pub struct WhereCondition { 19 19 pub eq: Option<Value>, ··· 39 39 /// } 40 40 /// } 41 41 /// ``` 42 42 - #[derive(Debug, Serialize, Deserialize)] 42 42 + #[derive(Debug, Clone, Serialize, Deserialize)] 43 43 #[serde(rename_all = "camelCase")] 44 44 pub struct WhereClause { 45 45 #[serde(flatten)]

+1

frontend/src/features/slices/waitlist/api.ts

reviewed

··· 70 70 slice: { eq: sliceUri }, 71 71 }, 72 72 sortBy: [{ field: "createdAt", direction: "desc" }], 73 73 + limit: 20, 73 74 }); 74 75 75 76 if (!requestsResponse.records || requestsResponse.records.length === 0) {