// file: kb_lib/src/db/queries/dex.rs //! Queries for `kb_dexes`. /// Inserts or updates one normalized DEX row by code. pub async fn upsert_dex( database: &crate::KbDatabase, dto: &crate::KbDexDto, ) -> Result { match database.connection() { crate::KbDatabaseConnection::Sqlite(pool) => { let query_result = sqlx::query( r#" INSERT INTO kb_dexes ( code, name, program_id, router_program_id, is_enabled, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(code) DO UPDATE SET name = excluded.name, program_id = excluded.program_id, router_program_id = excluded.router_program_id, is_enabled = excluded.is_enabled, updated_at = excluded.updated_at "#, ) .bind(dto.code.clone()) .bind(dto.name.clone()) .bind(dto.program_id.clone()) .bind(dto.router_program_id.clone()) .bind(if dto.is_enabled { 1_i64 } else { 0_i64 }) .bind(dto.created_at.to_rfc3339()) .bind(dto.updated_at.to_rfc3339()) .execute(pool) .await; if let Err(error) = query_result { return Err(crate::KbError::Db(format!( "cannot upsert kb_dexes on sqlite: {}", error ))); } let id_result = sqlx::query_scalar::( r#" SELECT id FROM kb_dexes WHERE code = ? LIMIT 1 "#, ) .bind(dto.code.clone()) .fetch_one(pool) .await; match id_result { Ok(id) => Ok(id), Err(error) => Err(crate::KbError::Db(format!( "cannot fetch kb_dexes id for code '{}' on sqlite: {}", dto.code, error ))), } } } } /// Reads one normalized DEX row by code. pub async fn get_dex_by_code( database: &crate::KbDatabase, code: &str, ) -> Result, crate::KbError> { match database.connection() { crate::KbDatabaseConnection::Sqlite(pool) => { let query_result = sqlx::query_as::( r#" SELECT id, code, name, program_id, router_program_id, is_enabled, created_at, updated_at FROM kb_dexes WHERE code = ? LIMIT 1 "#, ) .bind(code) .fetch_optional(pool) .await; let entity_option = match query_result { Ok(entity_option) => entity_option, Err(error) => { return Err(crate::KbError::Db(format!( "cannot read kb_dexes '{}' on sqlite: {}", code, error ))); } }; match entity_option { Some(entity) => { let dto_result = crate::KbDexDto::try_from(entity); match dto_result { Ok(dto) => Ok(Some(dto)), Err(error) => Err(error), } } None => Ok(None), } } } } /// Lists normalized DEX rows. pub async fn list_dexes( database: &crate::KbDatabase, ) -> Result, crate::KbError> { match database.connection() { crate::KbDatabaseConnection::Sqlite(pool) => { let query_result = sqlx::query_as::( r#" SELECT id, code, name, program_id, router_program_id, is_enabled, created_at, updated_at FROM kb_dexes ORDER BY code ASC "#, ) .fetch_all(pool) .await; let entities = match query_result { Ok(entities) => entities, Err(error) => { return Err(crate::KbError::Db(format!( "cannot list kb_dexes on sqlite: {}", error ))); } }; let mut dtos = std::vec::Vec::new(); for entity in entities { let dto_result = crate::KbDexDto::try_from(entity); let dto = match dto_result { Ok(dto) => dto, Err(error) => return Err(error), }; dtos.push(dto); } Ok(dtos) } } } #[cfg(test)] mod tests { #[tokio::test] async fn dex_roundtrip_works() { let tempdir = tempfile::tempdir().expect("tempdir must succeed"); let database_path = tempdir.path().join("dex_roundtrip.sqlite3"); let config = crate::KbDatabaseConfig { enabled: true, backend: crate::KbDatabaseBackend::Sqlite, sqlite: crate::KbSqliteDatabaseConfig { path: database_path.to_string_lossy().to_string(), create_if_missing: true, busy_timeout_ms: 5000, max_connections: 1, auto_initialize_schema: true, use_wal: true, }, }; let database = crate::KbDatabase::connect_and_initialize(&config) .await .expect("database init must succeed"); let dex_id = crate::upsert_dex( &database, &crate::KbDexDto::new( "raydium".to_string(), "Raydium".to_string(), None, None, true, ), ) .await .expect("dex upsert must succeed"); assert!(dex_id > 0); let dex = crate::get_dex_by_code(&database, "raydium") .await .expect("get dex must succeed"); assert!(dex.is_some()); assert_eq!(dex.expect("dex must exist").name, "Raydium"); let dexes = crate::list_dexes(&database) .await .expect("list dexes must succeed"); assert_eq!(dexes.len(), 1); } }