From f8a23091737083e1cb463c5ea6424723ab0e1a53 Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Wed, 29 Apr 2026 15:54:03 +0200 Subject: [PATCH] 0.7.15 --- CHANGELOG.md | 1 + Cargo.toml | 2 +- ROADMAP.md | 13 +- kb_lib/src/db.rs | 11 + kb_lib/src/db/dtos.rs | 4 + kb_lib/src/db/dtos/wallet.rs | 69 +++ kb_lib/src/db/dtos/wallet_participation.rs | 134 +++++ kb_lib/src/db/entities.rs | 4 + kb_lib/src/db/entities/wallet.rs | 18 + .../src/db/entities/wallet_participation.rs | 32 ++ kb_lib/src/db/queries.rs | 9 + kb_lib/src/db/queries/wallet.rs | 141 +++++ kb_lib/src/db/queries/wallet_participation.rs | 224 ++++++++ kb_lib/src/db/schema.rs | 110 ++++ kb_lib/src/lib.rs | 14 + kb_lib/src/tx_resolution.rs | 17 +- kb_lib/src/wallet_observation.rs | 488 ++++++++++++++++++ 17 files changed, 1281 insertions(+), 10 deletions(-) create mode 100644 kb_lib/src/db/dtos/wallet.rs create mode 100644 kb_lib/src/db/dtos/wallet_participation.rs create mode 100644 kb_lib/src/db/entities/wallet.rs create mode 100644 kb_lib/src/db/entities/wallet_participation.rs create mode 100644 kb_lib/src/db/queries/wallet.rs create mode 100644 kb_lib/src/db/queries/wallet_participation.rs create mode 100644 kb_lib/src/wallet_observation.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 4217bd9..0472253 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,3 +45,4 @@ 0.7.12 - Ajout du premier support DexLab Swap/Pool avec décodage create_pool/swap, persistance des événements décodés et détection métier automatique pool/pair/listing 0.7.13 - Extension de la couche launch origins avec Bags et Moonit, ajout d’un enregistrement programmatique des mappings Bags, et détection automatique Moonit via suffixe de token mint 0.7.14 - Ajout d’une couche consolidée de traçabilité fondatrice multi-DEX avec enregistrement des pool origins, rattachement au decoded event, au pool/pair/listing et à l’éventuelle launch attribution +0.7.15 - Ajout d’une couche wallets observés et participations observées, avec extraction des rôles depuis les payloads décodés et rattachement transaction / decoded event / pool / pair diff --git a/Cargo.toml b/Cargo.toml index 581d4dc..b855704 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ members = [ ] [workspace.package] -version = "0.7.14" +version = "0.7.15" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-bobobot" diff --git a/ROADMAP.md b/ROADMAP.md index 12a7033..c2126c7 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -596,14 +596,13 @@ Réalisé : - préparation de la future couche analytique sur une base multi-DEX plus cohérente. ### 6.047. Version `0.7.15` — Wallets, holdings et participants observés -Objectif : préparer le suivi des acteurs on-chain autour des pools et tokens détectés. +Réalisé : -À faire : - -- préparer le rattachement des signatures, instructions et événements à des wallets observés, -- introduire la notion de holdings utiles au suivi des tokens, -- préparer l’identification des créateurs, mint authorities, wallets d’activité et contreparties, -- éviter de limiter l’analyse future au seul niveau token/pool sans vision des participants. +- ajout d’une première couche `wallets` pour les adresses observées dans le pipeline, +- ajout d’une première couche `wallet participations` pour rattacher une adresse à une transaction, un decoded event, un pool et un pair, +- extraction des rôles observés depuis les payloads décodés (`creator`, `payer`, `owner`, `user`), +- branchement automatique depuis le pipeline de résolution transactionnelle, +- report des holdings à l’étape suivante afin de conserver une séparation nette entre acteurs observés et balances observées. ### 6.048. Version `0.7.16` — Séries de prix, volumes et agrégats DEX Objectif : préparer la couche analytique fine à partir des événements métier normalisés. diff --git a/kb_lib/src/db.rs b/kb_lib/src/db.rs index bad8f32..74fc40e 100644 --- a/kb_lib/src/db.rs +++ b/kb_lib/src/db.rs @@ -40,6 +40,8 @@ pub use dtos::KbSwapDto; pub use dtos::KbTokenBurnEventDto; pub use dtos::KbTokenDto; pub use dtos::KbTokenMintEventDto; +pub use dtos::KbWalletDto; +pub use dtos::KbWalletParticipationDto; pub use entities::KbAnalysisSignalEntity; pub use entities::KbChainInstructionEntity; pub use entities::KbChainSlotEntity; @@ -65,6 +67,8 @@ pub use entities::KbSwapEntity; pub use entities::KbTokenBurnEventEntity; pub use entities::KbTokenEntity; pub use entities::KbTokenMintEventEntity; +pub use entities::KbWalletEntity; +pub use entities::KbWalletParticipationEntity; pub use queries::delete_chain_instructions_by_transaction_id; pub use queries::get_chain_slot; pub use queries::get_chain_transaction_by_signature; @@ -82,6 +86,8 @@ pub use queries::get_pool_by_address; pub use queries::get_pool_listing_by_pool_id; pub use queries::get_pool_origin_by_pool_id; pub use queries::get_token_by_mint; +pub use queries::get_wallet_by_address; +pub use queries::get_wallet_participation_by_unique_key; pub use queries::insert_analysis_signal; pub use queries::insert_chain_instruction; pub use queries::insert_db_runtime_event; @@ -110,6 +116,9 @@ pub use queries::list_recent_onchain_observations; pub use queries::list_recent_swaps; pub use queries::list_recent_token_burn_events; pub use queries::list_recent_token_mint_events; +pub use queries::list_wallet_participations_by_pool_id; +pub use queries::list_wallet_participations_by_wallet_id; +pub use queries::list_wallets; pub use queries::upsert_chain_slot; pub use queries::upsert_chain_transaction; pub use queries::upsert_db_metadata; @@ -131,6 +140,8 @@ pub use queries::upsert_swap; pub use queries::upsert_token; pub use queries::upsert_token_burn_event; pub use queries::upsert_token_mint_event; +pub use queries::upsert_wallet; +pub use queries::upsert_wallet_participation; pub use types::KbAnalysisSignalSeverity; pub use types::KbDatabaseBackend; pub use types::KbDbRuntimeEventLevel; diff --git a/kb_lib/src/db/dtos.rs b/kb_lib/src/db/dtos.rs index dde9c58..dbfbb7f 100644 --- a/kb_lib/src/db/dtos.rs +++ b/kb_lib/src/db/dtos.rs @@ -27,6 +27,8 @@ mod swap; mod token; mod token_burn_event; mod token_mint_event; +mod wallet; +mod wallet_participation; pub use analysis_signal::KbAnalysisSignalDto; pub use chain_instruction::KbChainInstructionDto; @@ -53,3 +55,5 @@ pub use swap::KbSwapDto; pub use token::KbTokenDto; pub use token_burn_event::KbTokenBurnEventDto; pub use token_mint_event::KbTokenMintEventDto; +pub use wallet::KbWalletDto; +pub use wallet_participation::KbWalletParticipationDto; diff --git a/kb_lib/src/db/dtos/wallet.rs b/kb_lib/src/db/dtos/wallet.rs new file mode 100644 index 0000000..e61d93b --- /dev/null +++ b/kb_lib/src/db/dtos/wallet.rs @@ -0,0 +1,69 @@ +// file: kb_lib/src/db/dtos/wallet.rs + +//! Wallet DTO. + +/// Application-facing wallet DTO. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct KbWalletDto { + /// Optional numeric primary key. + pub id: std::option::Option, + /// Stable wallet address. + pub address: std::string::String, + /// Optional user-defined label. + pub label: std::option::Option, + /// Creation timestamp. + pub first_seen_at: chrono::DateTime, + /// Update timestamp. + pub last_seen_at: chrono::DateTime, +} + +impl KbWalletDto { + /// Creates a new wallet DTO. + pub fn new( + address: std::string::String, + label: std::option::Option, + ) -> Self { + let now = chrono::Utc::now(); + Self { + id: None, + address, + label, + first_seen_at: now, + last_seen_at: now, + } + } +} + +impl TryFrom for KbWalletDto { + type Error = crate::KbError; + + fn try_from(entity: crate::KbWalletEntity) -> Result { + let first_seen_at_result = chrono::DateTime::parse_from_rfc3339(&entity.first_seen_at); + let first_seen_at = match first_seen_at_result { + Ok(first_seen_at) => first_seen_at.with_timezone(&chrono::Utc), + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot parse wallet first_seen_at '{}': {}", + entity.first_seen_at, error + ))); + } + }; + let last_seen_at_result = chrono::DateTime::parse_from_rfc3339(&entity.last_seen_at); + let last_seen_at = match last_seen_at_result { + Ok(last_seen_at) => last_seen_at.with_timezone(&chrono::Utc), + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot parse wallet last_seen_at '{}': {}", + entity.last_seen_at, error + ))); + } + }; + Ok(Self { + id: Some(entity.id), + address: entity.address, + label: entity.label, + first_seen_at, + last_seen_at, + }) + } +} diff --git a/kb_lib/src/db/dtos/wallet_participation.rs b/kb_lib/src/db/dtos/wallet_participation.rs new file mode 100644 index 0000000..bbdf81f --- /dev/null +++ b/kb_lib/src/db/dtos/wallet_participation.rs @@ -0,0 +1,134 @@ +// file: kb_lib/src/db/dtos/wallet_participation.rs + +//! Wallet-participation DTO. + +/// Application-facing wallet-participation DTO. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct KbWalletParticipationDto { + /// Optional numeric primary key. + pub id: std::option::Option, + /// Related wallet id. + pub wallet_id: i64, + /// Related transaction id. + pub transaction_id: i64, + /// Optional related decoded event id. + pub decoded_event_id: std::option::Option, + /// Optional related pool id. + pub pool_id: std::option::Option, + /// Optional related pair id. + pub pair_id: std::option::Option, + /// Stable participation role. + pub role: std::string::String, + /// Stable unique key used for idempotent upserts. + pub unique_key: std::string::String, + /// Observation source kind. + pub source_kind: crate::KbObservationSourceKind, + /// Optional logical source endpoint name. + pub source_endpoint_name: std::option::Option, + /// Creation timestamp. + pub created_at: chrono::DateTime, + /// Update timestamp. + pub updated_at: chrono::DateTime, +} + +impl KbWalletParticipationDto { + /// Creates a new wallet-participation DTO. + pub fn new( + wallet_id: i64, + transaction_id: i64, + decoded_event_id: std::option::Option, + pool_id: std::option::Option, + pair_id: std::option::Option, + role: std::string::String, + source_kind: crate::KbObservationSourceKind, + source_endpoint_name: std::option::Option, + ) -> Self { + let now = chrono::Utc::now(); + let unique_key = kb_build_wallet_participation_unique_key( + wallet_id, + transaction_id, + decoded_event_id, + pool_id, + pair_id, + role.as_str(), + ); + + Self { + id: None, + wallet_id, + transaction_id, + decoded_event_id, + pool_id, + pair_id, + role, + unique_key, + source_kind, + source_endpoint_name, + created_at: now, + updated_at: now, + } + } +} + +impl TryFrom for KbWalletParticipationDto { + type Error = crate::KbError; + + fn try_from(entity: crate::KbWalletParticipationEntity) -> Result { + let source_kind_result = crate::KbObservationSourceKind::from_i16(entity.source_kind); + let source_kind = match source_kind_result { + Ok(source_kind) => source_kind, + Err(error) => return Err(error), + }; + let created_at_result = chrono::DateTime::parse_from_rfc3339(&entity.created_at); + let created_at = match created_at_result { + Ok(created_at) => created_at.with_timezone(&chrono::Utc), + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot parse wallet_participation created_at '{}': {}", + entity.created_at, error + ))); + } + }; + let updated_at_result = chrono::DateTime::parse_from_rfc3339(&entity.updated_at); + let updated_at = match updated_at_result { + Ok(updated_at) => updated_at.with_timezone(&chrono::Utc), + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot parse wallet_participation updated_at '{}': {}", + entity.updated_at, error + ))); + } + }; + Ok(Self { + id: Some(entity.id), + wallet_id: entity.wallet_id, + transaction_id: entity.transaction_id, + decoded_event_id: entity.decoded_event_id, + pool_id: entity.pool_id, + pair_id: entity.pair_id, + role: entity.role, + unique_key: entity.unique_key, + source_kind, + source_endpoint_name: entity.source_endpoint_name, + created_at, + updated_at, + }) + } +} + +fn kb_build_wallet_participation_unique_key( + wallet_id: i64, + transaction_id: i64, + decoded_event_id: std::option::Option, + pool_id: std::option::Option, + pair_id: std::option::Option, + role: &str, +) -> std::string::String { + let decoded_event_id_value = decoded_event_id.unwrap_or_default(); + let pool_id_value = pool_id.unwrap_or_default(); + let pair_id_value = pair_id.unwrap_or_default(); + format!( + "{}:{}:{}:{}:{}:{}", + wallet_id, transaction_id, decoded_event_id_value, pool_id_value, pair_id_value, role + ) +} diff --git a/kb_lib/src/db/entities.rs b/kb_lib/src/db/entities.rs index 14ddc53..4f13d96 100644 --- a/kb_lib/src/db/entities.rs +++ b/kb_lib/src/db/entities.rs @@ -29,6 +29,8 @@ mod swap; mod token; mod token_burn_event; mod token_mint_event; +mod wallet; +mod wallet_participation; pub use analysis_signal::KbAnalysisSignalEntity; pub use chain_instruction::KbChainInstructionEntity; @@ -55,3 +57,5 @@ pub use swap::KbSwapEntity; pub use token::KbTokenEntity; pub use token_burn_event::KbTokenBurnEventEntity; pub use token_mint_event::KbTokenMintEventEntity; +pub use wallet::KbWalletEntity; +pub use wallet_participation::KbWalletParticipationEntity; diff --git a/kb_lib/src/db/entities/wallet.rs b/kb_lib/src/db/entities/wallet.rs new file mode 100644 index 0000000..3750faf --- /dev/null +++ b/kb_lib/src/db/entities/wallet.rs @@ -0,0 +1,18 @@ +// file: kb_lib/src/db/entities/wallet.rs + +//! Wallet entity. + +/// Persisted wallet row. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)] +pub struct KbWalletEntity { + /// Numeric primary key. + pub id: i64, + /// Stable wallet address. + pub address: std::string::String, + /// Optional user-defined label. + pub label: std::option::Option, + /// Creation timestamp encoded as RFC3339 UTC text. + pub first_seen_at: std::string::String, + /// Update timestamp encoded as RFC3339 UTC text. + pub last_seen_at: std::string::String, +} diff --git a/kb_lib/src/db/entities/wallet_participation.rs b/kb_lib/src/db/entities/wallet_participation.rs new file mode 100644 index 0000000..aaf47e1 --- /dev/null +++ b/kb_lib/src/db/entities/wallet_participation.rs @@ -0,0 +1,32 @@ +// file: kb_lib/src/db/entities/wallet_participation.rs + +//! Wallet-participation entity. + +/// Persisted wallet-participation row. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)] +pub struct KbWalletParticipationEntity { + /// Numeric primary key. + pub id: i64, + /// Related wallet id. + pub wallet_id: i64, + /// Related transaction id. + pub transaction_id: i64, + /// Optional related decoded event id. + pub decoded_event_id: std::option::Option, + /// Optional related pool id. + pub pool_id: std::option::Option, + /// Optional related pair id. + pub pair_id: std::option::Option, + /// Stable participation role. + pub role: std::string::String, + /// Stable unique key used for idempotent upserts. + pub unique_key: std::string::String, + /// Observation source kind. + pub source_kind: i16, + /// Optional logical source endpoint name. + pub source_endpoint_name: std::option::Option, + /// Creation timestamp encoded as RFC3339 UTC text. + pub created_at: std::string::String, + /// Update timestamp encoded as RFC3339 UTC text. + pub updated_at: std::string::String, +} diff --git a/kb_lib/src/db/queries.rs b/kb_lib/src/db/queries.rs index aec7ca0..81b55b4 100644 --- a/kb_lib/src/db/queries.rs +++ b/kb_lib/src/db/queries.rs @@ -31,6 +31,8 @@ mod swap; mod token; mod token_burn_event; mod token_mint_event; +mod wallet; +mod wallet_participation; pub use analysis_signal::insert_analysis_signal; pub use analysis_signal::list_recent_analysis_signals; @@ -98,3 +100,10 @@ pub use token_burn_event::list_recent_token_burn_events; pub use token_burn_event::upsert_token_burn_event; pub use token_mint_event::list_recent_token_mint_events; pub use token_mint_event::upsert_token_mint_event; +pub use wallet::get_wallet_by_address; +pub use wallet::list_wallets; +pub use wallet::upsert_wallet; +pub use wallet_participation::get_wallet_participation_by_unique_key; +pub use wallet_participation::list_wallet_participations_by_pool_id; +pub use wallet_participation::list_wallet_participations_by_wallet_id; +pub use wallet_participation::upsert_wallet_participation; diff --git a/kb_lib/src/db/queries/wallet.rs b/kb_lib/src/db/queries/wallet.rs new file mode 100644 index 0000000..32c6411 --- /dev/null +++ b/kb_lib/src/db/queries/wallet.rs @@ -0,0 +1,141 @@ +// file: kb_lib/src/db/queries/wallet.rs + +//! Queries for `kb_wallets`. + +/// Inserts or updates one wallet row and returns its stable internal id. +pub async fn upsert_wallet( + database: &crate::KbDatabase, + dto: &crate::KbWalletDto, +) -> Result { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query( + r#" +INSERT INTO kb_wallets ( + address, + label, + first_seen_at, + last_seen_at +) +VALUES (?, ?, ?, ?) +ON CONFLICT(address) DO UPDATE SET + label = COALESCE(excluded.label, kb_wallets.label), + last_seen_at = excluded.last_seen_at + "#, + ) + .bind(dto.address.clone()) + .bind(dto.label.clone()) + .bind(dto.first_seen_at.to_rfc3339()) + .bind(dto.last_seen_at.to_rfc3339()) + .execute(pool) + .await; + if let Err(error) = query_result { + return Err(crate::KbError::Db(format!( + "cannot upsert kb_wallets on sqlite: {}", + error + ))); + } + let id_result = sqlx::query_scalar::( + r#" +SELECT id +FROM kb_wallets +WHERE address = ? +LIMIT 1 + "#, + ) + .bind(dto.address.clone()) + .fetch_one(pool) + .await; + match id_result { + Ok(id) => Ok(id), + Err(error) => Err(crate::KbError::Db(format!( + "cannot fetch kb_wallets id for address '{}' on sqlite: {}", + dto.address, error + ))), + } + } + } +} + +/// Returns one wallet identified by its address, if it exists. +pub async fn get_wallet_by_address( + database: &crate::KbDatabase, + address: &str, +) -> Result, crate::KbError> { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query_as::( + r#" +SELECT + id, + address, + label, + first_seen_at, + last_seen_at +FROM kb_wallets +WHERE address = ? +LIMIT 1 + "#, + ) + .bind(address) + .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_wallets '{}' on sqlite: {}", + address, error + ))); + } + }; + match entity_option { + Some(entity) => crate::KbWalletDto::try_from(entity).map(Some), + None => Ok(None), + } + } + } +} + +/// Lists all persisted wallets ordered by address. +pub async fn list_wallets( + database: &crate::KbDatabase, +) -> Result, crate::KbError> { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query_as::( + r#" +SELECT + id, + address, + label, + first_seen_at, + last_seen_at +FROM kb_wallets +ORDER BY address ASC + "#, + ) + .fetch_all(pool) + .await; + let entities = match query_result { + Ok(entities) => entities, + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot list kb_wallets on sqlite: {}", + error + ))); + } + }; + let mut dtos = std::vec::Vec::new(); + for entity in entities { + let dto_result = crate::KbWalletDto::try_from(entity); + let dto = match dto_result { + Ok(dto) => dto, + Err(error) => return Err(error), + }; + dtos.push(dto); + } + Ok(dtos) + } + } +} diff --git a/kb_lib/src/db/queries/wallet_participation.rs b/kb_lib/src/db/queries/wallet_participation.rs new file mode 100644 index 0000000..72396c8 --- /dev/null +++ b/kb_lib/src/db/queries/wallet_participation.rs @@ -0,0 +1,224 @@ +// file: kb_lib/src/db/queries/wallet_participation.rs + +//! Queries for `kb_wallet_participations`. + +/// Inserts or updates one wallet-participation row and returns its stable internal id. +pub async fn upsert_wallet_participation( + database: &crate::KbDatabase, + dto: &crate::KbWalletParticipationDto, +) -> Result { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query( + r#" +INSERT INTO kb_wallet_participations ( + wallet_id, + transaction_id, + decoded_event_id, + pool_id, + pair_id, + role, + unique_key, + source_kind, + source_endpoint_name, + created_at, + updated_at +) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(unique_key) DO UPDATE SET + updated_at = excluded.updated_at + "#, + ) + .bind(dto.wallet_id) + .bind(dto.transaction_id) + .bind(dto.decoded_event_id) + .bind(dto.pool_id) + .bind(dto.pair_id) + .bind(dto.role.clone()) + .bind(dto.unique_key.clone()) + .bind(dto.source_kind.to_i16()) + .bind(dto.source_endpoint_name.clone()) + .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_wallet_participations on sqlite: {}", + error + ))); + } + let id_result = sqlx::query_scalar::( + r#" +SELECT id +FROM kb_wallet_participations +WHERE unique_key = ? +LIMIT 1 + "#, + ) + .bind(dto.unique_key.clone()) + .fetch_one(pool) + .await; + match id_result { + Ok(id) => Ok(id), + Err(error) => Err(crate::KbError::Db(format!( + "cannot fetch kb_wallet_participations id for unique_key '{}' on sqlite: {}", + dto.unique_key, error + ))), + } + } + } +} + +/// Returns one wallet-participation identified by its unique key, if it exists. +pub async fn get_wallet_participation_by_unique_key( + database: &crate::KbDatabase, + unique_key: &str, +) -> Result, crate::KbError> { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query_as::( + r#" +SELECT + id, + wallet_id, + transaction_id, + decoded_event_id, + pool_id, + pair_id, + role, + unique_key, + source_kind, + source_endpoint_name, + created_at, + updated_at +FROM kb_wallet_participations +WHERE unique_key = ? +LIMIT 1 + "#, + ) + .bind(unique_key) + .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_wallet_participations by unique_key '{}' on sqlite: {}", + unique_key, error + ))); + } + }; + match entity_option { + Some(entity) => crate::KbWalletParticipationDto::try_from(entity).map(Some), + None => Ok(None), + } + } + } +} + +/// Lists wallet-participation rows for one wallet id. +pub async fn list_wallet_participations_by_wallet_id( + database: &crate::KbDatabase, + wallet_id: i64, +) -> Result, crate::KbError> { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query_as::( + r#" +SELECT + id, + wallet_id, + transaction_id, + decoded_event_id, + pool_id, + pair_id, + role, + unique_key, + source_kind, + source_endpoint_name, + created_at, + updated_at +FROM kb_wallet_participations +WHERE wallet_id = ? +ORDER BY created_at ASC, id ASC + "#, + ) + .bind(wallet_id) + .fetch_all(pool) + .await; + let entities = match query_result { + Ok(entities) => entities, + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot list kb_wallet_participations by wallet_id '{}' on sqlite: {}", + wallet_id, error + ))); + } + }; + let mut dtos = std::vec::Vec::new(); + for entity in entities { + let dto_result = crate::KbWalletParticipationDto::try_from(entity); + let dto = match dto_result { + Ok(dto) => dto, + Err(error) => return Err(error), + }; + dtos.push(dto); + } + Ok(dtos) + } + } +} + +/// Lists wallet-participation rows for one pool id. +pub async fn list_wallet_participations_by_pool_id( + database: &crate::KbDatabase, + pool_id: i64, +) -> Result, crate::KbError> { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query_as::( + r#" +SELECT + id, + wallet_id, + transaction_id, + decoded_event_id, + pool_id, + pair_id, + role, + unique_key, + source_kind, + source_endpoint_name, + created_at, + updated_at +FROM kb_wallet_participations +WHERE pool_id = ? +ORDER BY created_at ASC, id ASC + "#, + ) + .bind(pool_id) + .fetch_all(pool) + .await; + let entities = match query_result { + Ok(entities) => entities, + Err(error) => { + return Err(crate::KbError::Db(format!( + "cannot list kb_wallet_participations by pool_id '{}' on sqlite: {}", + pool_id, error + ))); + } + }; + let mut dtos = std::vec::Vec::new(); + for entity in entities { + let dto_result = crate::KbWalletParticipationDto::try_from(entity); + let dto = match dto_result { + Ok(dto) => dto, + Err(error) => return Err(error), + }; + dtos.push(dto); + } + Ok(dtos) + } + } +} diff --git a/kb_lib/src/db/schema.rs b/kb_lib/src/db/schema.rs index 369f396..09d62db 100644 --- a/kb_lib/src/db/schema.rs +++ b/kb_lib/src/db/schema.rs @@ -274,6 +274,26 @@ pub(crate) async fn ensure_schema(database: &crate::KbDatabase) -> Result<(), cr if let Err(error) = result { return Err(error); } + let result = create_kb_wallets_table(pool).await; + if let Err(error) = result { + return Err(error); + } + let result = create_kb_wallet_participations_table(pool).await; + if let Err(error) = result { + return Err(error); + } + let result = create_kb_idx_wallet_participations_wallet_id(pool).await; + if let Err(error) = result { + return Err(error); + } + let result = create_kb_idx_wallet_participations_pool_id(pool).await; + if let Err(error) = result { + return Err(error); + } + let result = create_kb_idx_wallet_participations_transaction_id(pool).await; + if let Err(error) = result { + return Err(error); + } Ok(()) } } @@ -1498,3 +1518,93 @@ ON kb_pool_origins(founding_transaction_id) ) .await } + +async fn create_kb_wallets_table(pool: &sqlx::SqlitePool) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_wallets_table", + r#" +CREATE TABLE IF NOT EXISTS kb_wallets ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + address TEXT NOT NULL UNIQUE, + label TEXT NULL, + first_seen_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL +) + "#, + ) + .await +} + +async fn create_kb_wallet_participations_table( + pool: &sqlx::SqlitePool, +) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_wallet_participations_table", + r#" +CREATE TABLE IF NOT EXISTS kb_wallet_participations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + wallet_id INTEGER NOT NULL, + transaction_id INTEGER NOT NULL, + decoded_event_id INTEGER NULL, + pool_id INTEGER NULL, + pair_id INTEGER NULL, + role TEXT NOT NULL, + unique_key TEXT NOT NULL UNIQUE, + source_kind INTEGER NOT NULL, + source_endpoint_name TEXT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY(wallet_id) REFERENCES kb_wallets(id) ON DELETE CASCADE, + FOREIGN KEY(transaction_id) REFERENCES kb_chain_transactions(id) ON DELETE CASCADE, + FOREIGN KEY(decoded_event_id) REFERENCES kb_dex_decoded_events(id) ON DELETE SET NULL, + FOREIGN KEY(pool_id) REFERENCES kb_pools(id) ON DELETE SET NULL, + FOREIGN KEY(pair_id) REFERENCES kb_pairs(id) ON DELETE SET NULL +) + "#, + ) + .await +} + +async fn create_kb_idx_wallet_participations_wallet_id( + pool: &sqlx::SqlitePool, +) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_idx_wallet_participations_wallet_id", + r#" +CREATE INDEX IF NOT EXISTS kb_idx_wallet_participations_wallet_id +ON kb_wallet_participations(wallet_id) + "#, + ) + .await +} + +async fn create_kb_idx_wallet_participations_pool_id( + pool: &sqlx::SqlitePool, +) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_idx_wallet_participations_pool_id", + r#" +CREATE INDEX IF NOT EXISTS kb_idx_wallet_participations_pool_id +ON kb_wallet_participations(pool_id) + "#, + ) + .await +} + +async fn create_kb_idx_wallet_participations_transaction_id( + pool: &sqlx::SqlitePool, +) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_idx_wallet_participations_transaction_id", + r#" +CREATE INDEX IF NOT EXISTS kb_idx_wallet_participations_transaction_id +ON kb_wallet_participations(transaction_id) + "#, + ) + .await +} diff --git a/kb_lib/src/lib.rs b/kb_lib/src/lib.rs index 199e1b0..ee7b69a 100644 --- a/kb_lib/src/lib.rs +++ b/kb_lib/src/lib.rs @@ -26,6 +26,7 @@ mod tracing; mod tx_model; mod tx_resolution; mod types; +mod wallet_observation; mod ws_client; mod ws_manager; @@ -101,6 +102,10 @@ pub use db::KbTokenDto; pub use db::KbTokenEntity; pub use db::KbTokenMintEventDto; pub use db::KbTokenMintEventEntity; +pub use db::KbWalletDto; +pub use db::KbWalletEntity; +pub use db::KbWalletParticipationDto; +pub use db::KbWalletParticipationEntity; pub use db::delete_chain_instructions_by_transaction_id; pub use db::get_chain_slot; pub use db::get_chain_transaction_by_signature; @@ -118,6 +123,8 @@ pub use db::get_pool_by_address; pub use db::get_pool_listing_by_pool_id; pub use db::get_pool_origin_by_pool_id; pub use db::get_token_by_mint; +pub use db::get_wallet_by_address; +pub use db::get_wallet_participation_by_unique_key; pub use db::insert_analysis_signal; pub use db::insert_chain_instruction; pub use db::insert_db_runtime_event; @@ -146,6 +153,9 @@ pub use db::list_recent_onchain_observations; pub use db::list_recent_swaps; pub use db::list_recent_token_burn_events; pub use db::list_recent_token_mint_events; +pub use db::list_wallet_participations_by_pool_id; +pub use db::list_wallet_participations_by_wallet_id; +pub use db::list_wallets; pub use db::upsert_chain_slot; pub use db::upsert_chain_transaction; pub use db::upsert_db_metadata; @@ -167,6 +177,8 @@ pub use db::upsert_swap; pub use db::upsert_token; pub use db::upsert_token_burn_event; pub use db::upsert_token_mint_event; +pub use db::upsert_wallet; +pub use db::upsert_wallet_participation; pub use detect::KbDetectionObservationInput; pub use detect::KbDetectionPersistenceService; pub use detect::KbDetectionPoolCandidateInput; @@ -264,6 +276,8 @@ pub use tx_resolution::KbWsTransactionResolutionEnvelope; pub use tx_resolution::KbWsTransactionResolutionRelay; pub use tx_resolution::KbWsTransactionResolutionRelayStats; pub use types::KbConnectionState; +pub use wallet_observation::KbWalletObservationResult; +pub use wallet_observation::KbWalletObservationService; pub use ws_client::WsClient; pub use ws_client::WsEvent; pub use ws_client::WsOutgoingMessage; diff --git a/kb_lib/src/tx_resolution.rs b/kb_lib/src/tx_resolution.rs index f46ca66..106d236 100644 --- a/kb_lib/src/tx_resolution.rs +++ b/kb_lib/src/tx_resolution.rs @@ -98,12 +98,13 @@ pub struct KbWsTransactionResolutionRelayStats { pub struct KbTransactionResolutionService { http_pool: std::sync::Arc, persistence: crate::KbDetectionPersistenceService, + http_role: std::string::String, transaction_model: crate::KbTransactionModelService, dex_decode_service: crate::KbDexDecodeService, dex_detect_service: crate::KbDexDetectService, launch_origin_service: crate::KbLaunchOriginService, pool_origin_service: crate::KbPoolOriginService, - http_role: std::string::String, + wallet_observation_service: crate::KbWalletObservationService, resolved_signatures: std::sync::Arc>>, } @@ -121,15 +122,17 @@ impl KbTransactionResolutionService { let dex_detect_service = crate::KbDexDetectService::new(database.clone()); let launch_origin_service = crate::KbLaunchOriginService::new(database.clone()); let pool_origin_service = crate::KbPoolOriginService::new(database.clone()); + let wallet_observation_service = crate::KbWalletObservationService::new(database.clone()); Self { http_pool, persistence, + http_role, transaction_model, dex_decode_service, dex_detect_service, launch_origin_service, pool_origin_service, - http_role, + wallet_observation_service, resolved_signatures: std::sync::Arc::new(tokio::sync::Mutex::new( std::collections::HashSet::new(), )), @@ -333,6 +336,15 @@ impl KbTransactionResolutionService { Err(error) => return Err(error), }; let pool_origin_count = pool_origins.len(); + let wallet_observations_result = self + .wallet_observation_service + .record_transaction_by_signature(request.signature.as_str()) + .await; + let wallet_observations = match wallet_observations_result { + Ok(wallet_observations) => wallet_observations, + Err(error) => return Err(error), + }; + let wallet_participation_count = wallet_observations.len(); let detection_results = match detection_results_result { Ok(detection_results) => detection_results, Err(error) => return Err(error), @@ -349,6 +361,7 @@ impl KbTransactionResolutionService { "detectedObjectCount": detected_object_count, "launchAttributionCount": launch_attribution_count, "poolOriginCount": pool_origin_count, + "walletParticipationCount": wallet_participation_count, "triggerPayload": request.trigger_payload.clone(), "transaction": transaction_value }); diff --git a/kb_lib/src/wallet_observation.rs b/kb_lib/src/wallet_observation.rs new file mode 100644 index 0000000..76fcf10 --- /dev/null +++ b/kb_lib/src/wallet_observation.rs @@ -0,0 +1,488 @@ +// file: kb_lib/src/wallet_observation.rs + +//! Wallet-observation service. + +/// One wallet-observation result. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct KbWalletObservationResult { + /// Related wallet id. + pub wallet_id: i64, + /// Wallet address. + pub wallet_address: std::string::String, + /// Persisted wallet-participation id. + pub wallet_participation_id: i64, + /// Stable observed role. + pub role: std::string::String, + /// Optional related pool id. + pub pool_id: std::option::Option, + /// Optional related pair id. + pub pair_id: std::option::Option, + /// Whether the participation row was newly created. + pub created_participation: bool, +} + +/// Wallet-observation service. +#[derive(Debug, Clone)] +pub struct KbWalletObservationService { + database: std::sync::Arc, + persistence: crate::KbDetectionPersistenceService, +} + +impl KbWalletObservationService { + /// Creates a new wallet-observation service. + pub fn new(database: std::sync::Arc) -> Self { + let persistence = crate::KbDetectionPersistenceService::new(database.clone()); + Self { + database, + persistence, + } + } + + /// Records observed wallets and participations for one resolved transaction signature. + pub async fn record_transaction_by_signature( + &self, + signature: &str, + ) -> Result, crate::KbError> { + let transaction_result = + crate::get_chain_transaction_by_signature(self.database.as_ref(), signature).await; + let transaction_option = match transaction_result { + Ok(transaction_option) => transaction_option, + Err(error) => return Err(error), + }; + let transaction = match transaction_option { + Some(transaction) => transaction, + None => { + return Err(crate::KbError::InvalidState(format!( + "cannot record wallet observations for unknown transaction '{}'", + signature + ))); + } + }; + let transaction_id = match transaction.id { + Some(transaction_id) => transaction_id, + None => { + return Err(crate::KbError::InvalidState(format!( + "transaction '{}' has no internal id", + signature + ))); + } + }; + let decoded_events_result = crate::list_dex_decoded_events_by_transaction_id( + self.database.as_ref(), + transaction_id, + ) + .await; + let decoded_events = match decoded_events_result { + Ok(decoded_events) => decoded_events, + Err(error) => return Err(error), + }; + let mut results = std::vec::Vec::new(); + for decoded_event in &decoded_events { + let decoded_event_id = match decoded_event.id { + Some(decoded_event_id) => decoded_event_id, + None => { + return Err(crate::KbError::InvalidState( + "decoded event has no internal id".to_string(), + )); + } + }; + let pool_id = match decoded_event.pool_account.clone() { + Some(pool_address) => { + let pool_result = + crate::get_pool_by_address(self.database.as_ref(), pool_address.as_str()) + .await; + let pool_option = match pool_result { + Ok(pool_option) => pool_option, + Err(error) => return Err(error), + }; + match pool_option { + Some(pool) => pool.id, + None => None, + } + } + None => None, + }; + let pair_id = match pool_id { + Some(pool_id) => { + let pair_result = + crate::get_pair_by_pool_id(self.database.as_ref(), pool_id).await; + let pair_option = match pair_result { + Ok(pair_option) => pair_option, + Err(error) => return Err(error), + }; + match pair_option { + Some(pair) => pair.id, + None => None, + } + } + None => None, + }; + let payload_result = + serde_json::from_str::(decoded_event.payload_json.as_str()); + let payload = match payload_result { + Ok(payload) => payload, + Err(error) => { + return Err(crate::KbError::Json(format!( + "cannot parse decoded_event payload_json '{}': {}", + decoded_event.payload_json, error + ))); + } + }; + let observed_roles = kb_collect_wallet_roles(&payload); + for (role, wallet_address) in observed_roles { + let wallet_result = + crate::get_wallet_by_address(self.database.as_ref(), wallet_address.as_str()) + .await; + let wallet_option = match wallet_result { + Ok(wallet_option) => wallet_option, + Err(error) => return Err(error), + }; + let wallet_id = match wallet_option { + Some(existing) => match existing.id { + Some(wallet_id) => { + let dto = crate::KbWalletDto::new( + wallet_address.clone(), + existing.label.clone(), + ); + let upsert_result = + crate::upsert_wallet(self.database.as_ref(), &dto).await; + match upsert_result { + Ok(_) => wallet_id, + Err(error) => return Err(error), + } + } + None => { + return Err(crate::KbError::InvalidState( + "wallet has no internal id".to_string(), + )); + } + }, + None => { + let dto = crate::KbWalletDto::new(wallet_address.clone(), None); + let upsert_result = + crate::upsert_wallet(self.database.as_ref(), &dto).await; + match upsert_result { + Ok(wallet_id) => wallet_id, + Err(error) => return Err(error), + } + } + }; + let participation_dto = crate::KbWalletParticipationDto::new( + wallet_id, + transaction_id, + Some(decoded_event_id), + pool_id, + pair_id, + role.clone(), + crate::KbObservationSourceKind::HttpRpc, + transaction.source_endpoint_name.clone(), + ); + let existing_participation_result = crate::get_wallet_participation_by_unique_key( + self.database.as_ref(), + participation_dto.unique_key.as_str(), + ) + .await; + let existing_participation_option = match existing_participation_result { + Ok(existing_participation_option) => existing_participation_option, + Err(error) => return Err(error), + }; + let created_participation = existing_participation_option.is_none(); + let participation_id_result = + crate::upsert_wallet_participation(self.database.as_ref(), &participation_dto) + .await; + let wallet_participation_id = match participation_id_result { + Ok(wallet_participation_id) => wallet_participation_id, + Err(error) => return Err(error), + }; + if created_participation { + let payload = serde_json::json!({ + "walletAddress": wallet_address, + "role": role, + "protocolName": decoded_event.protocol_name, + "eventKind": decoded_event.event_kind, + "poolId": pool_id, + "pairId": pair_id, + "transactionSignature": transaction.signature + }); + let observation_result = self + .persistence + .record_observation(&crate::KbDetectionObservationInput::new( + "wallet.participation".to_string(), + crate::KbObservationSourceKind::HttpRpc, + transaction.source_endpoint_name.clone(), + transaction.signature.clone(), + transaction.slot, + payload.clone(), + )) + .await; + let observation_id = match observation_result { + Ok(observation_id) => observation_id, + Err(error) => return Err(error), + }; + let signal_result = self + .persistence + .record_signal(&crate::KbDetectionSignalInput::new( + "signal.wallet.participation.observed".to_string(), + crate::KbAnalysisSignalSeverity::Low, + transaction.signature.clone(), + Some(observation_id), + None, + payload, + )) + .await; + if let Err(error) = signal_result { + return Err(error); + } + } + results.push(crate::KbWalletObservationResult { + wallet_id, + wallet_address, + wallet_participation_id, + role, + pool_id, + pair_id, + created_participation, + }); + } + } + Ok(results) + } +} + +fn kb_collect_wallet_roles( + payload: &serde_json::Value, +) -> std::vec::Vec<(std::string::String, std::string::String)> { + let role_mappings = [ + ("creator", vec!["creator", "poolCreator"]), + ("payer", vec!["payer", "funder"]), + ("owner", vec!["owner"]), + ("user", vec!["user"]), + ]; + let mut seen = std::collections::HashSet::::new(); + let mut results = std::vec::Vec::new(); + for (role, keys) in role_mappings { + let addresses = kb_extract_strings_for_candidate_keys(payload, &keys); + for address in addresses { + let dedupe_key = format!("{}:{}", role, address); + if seen.contains(&dedupe_key) { + continue; + } + seen.insert(dedupe_key); + results.push((role.to_string(), address)); + } + } + results +} + +fn kb_extract_strings_for_candidate_keys( + value: &serde_json::Value, + candidate_keys: &[&str], +) -> std::vec::Vec { + let mut values = std::vec::Vec::new(); + kb_extract_strings_for_candidate_keys_inner(value, candidate_keys, &mut values); + values +} + +fn kb_extract_strings_for_candidate_keys_inner( + value: &serde_json::Value, + candidate_keys: &[&str], + values: &mut std::vec::Vec, +) { + if let Some(object) = value.as_object() { + for candidate_key in candidate_keys { + let direct_option = object.get(*candidate_key); + if let Some(direct) = direct_option { + let text_option = direct.as_str(); + if let Some(text) = text_option { + values.push(text.to_string()); + } + } + } + for nested_value in object.values() { + kb_extract_strings_for_candidate_keys_inner(nested_value, candidate_keys, values); + } + return; + } + if let Some(array) = value.as_array() { + for nested_value in array { + kb_extract_strings_for_candidate_keys_inner(nested_value, candidate_keys, values); + } + } +} + +#[cfg(test)] +mod tests { + async fn make_database() -> std::sync::Arc { + let tempdir_result = tempfile::tempdir(); + let tempdir = match tempdir_result { + Ok(tempdir) => tempdir, + Err(error) => panic!("tempdir must succeed: {}", error), + }; + let database_path = tempdir.path().join("wallet_observation.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_result = crate::KbDatabase::connect_and_initialize(&config).await; + let database = match database_result { + Ok(database) => database, + Err(error) => panic!("database init must succeed: {}", error), + }; + std::sync::Arc::new(database) + } + + async fn seed_fluxbeam_transaction( + database: std::sync::Arc, + signature: &str, + ) { + let transaction_model = crate::KbTransactionModelService::new(database.clone()); + let dex_decode = crate::KbDexDecodeService::new(database.clone()); + let dex_detect = crate::KbDexDetectService::new(database); + let resolved_transaction = serde_json::json!({ + "slot": 930001, + "blockTime": 1779300001, + "version": 0, + "transaction": { + "message": { + "instructions": [ + { + "programId": crate::KB_FLUXBEAM_PROGRAM_ID, + "program": "fluxbeam", + "stackHeight": 1, + "accounts": [ + "WalletObsPool111", + "WalletObsLpMint111", + "WalletObsTokenA111", + "So11111111111111111111111111111111111111112", + "WalletObserved111" + ], + "parsed": { + "info": { + "instruction": "create_pool", + "pool": "WalletObsPool111", + "lpMint": "WalletObsLpMint111", + "tokenA": "WalletObsTokenA111", + "tokenB": "So11111111111111111111111111111111111111112", + "payer": "WalletObserved111" + } + }, + "data": "opaque" + } + ] + } + }, + "meta": { + "err": null, + "logMessages": [ + "Program log: Instruction: CreatePool" + ] + } + }); + let project_result = transaction_model + .persist_resolved_transaction( + signature, + Some("helius_primary_http".to_string()), + &resolved_transaction, + ) + .await; + if let Err(error) = project_result { + panic!("projection must succeed: {}", error); + } + let decode_result = dex_decode.decode_transaction_by_signature(signature).await; + if let Err(error) = decode_result { + panic!("dex decode must succeed: {}", error); + } + let detect_result = dex_detect.detect_transaction_by_signature(signature).await; + if let Err(error) = detect_result { + panic!("dex detect must succeed: {}", error); + } + } + + #[tokio::test] + async fn record_transaction_by_signature_creates_wallet_and_participations() { + let database = make_database().await; + seed_fluxbeam_transaction(database.clone(), "sig-wallet-observation-1").await; + let service = crate::KbWalletObservationService::new(database.clone()); + let record_result = service + .record_transaction_by_signature("sig-wallet-observation-1") + .await; + let results = match record_result { + Ok(results) => results, + Err(error) => panic!("wallet observation must succeed: {}", error), + }; + assert_eq!(results.len(), 2); + let wallets_result = crate::list_wallets(database.as_ref()).await; + let wallets = match wallets_result { + Ok(wallets) => wallets, + Err(error) => panic!("wallet list must succeed: {}", error), + }; + assert_eq!(wallets.len(), 1); + let wallet = &wallets[0]; + let wallet_id = match wallet.id { + Some(wallet_id) => wallet_id, + None => panic!("wallet must have an id"), + }; + let participations_result = + crate::list_wallet_participations_by_wallet_id(database.as_ref(), wallet_id).await; + let participations = match participations_result { + Ok(participations) => participations, + Err(error) => panic!("participation list must succeed: {}", error), + }; + assert_eq!(participations.len(), 2); + let mut roles = std::vec::Vec::new(); + for participation in &participations { + roles.push(participation.role.clone()); + } + roles.sort(); + assert_eq!(roles, vec!["creator".to_string(), "payer".to_string()]); + } + + #[tokio::test] + async fn record_transaction_by_signature_is_idempotent() { + let database = make_database().await; + seed_fluxbeam_transaction(database.clone(), "sig-wallet-observation-2").await; + let service = crate::KbWalletObservationService::new(database.clone()); + let first_result = service + .record_transaction_by_signature("sig-wallet-observation-2") + .await; + let first_results = match first_result { + Ok(first_results) => first_results, + Err(error) => panic!("first wallet observation must succeed: {}", error), + }; + assert_eq!(first_results.len(), 2); + assert!(first_results[0].created_participation); + assert!(first_results[1].created_participation); + let second_result = service + .record_transaction_by_signature("sig-wallet-observation-2") + .await; + let second_results = match second_result { + Ok(second_results) => second_results, + Err(error) => panic!("second wallet observation must succeed: {}", error), + }; + assert_eq!(second_results.len(), 2); + assert!(!second_results[0].created_participation); + assert!(!second_results[1].created_participation); + let wallets_result = crate::list_wallets(database.as_ref()).await; + let wallets = match wallets_result { + Ok(wallets) => wallets, + Err(error) => panic!("wallet list must succeed: {}", error), + }; + assert_eq!(wallets.len(), 1); + let wallet_id = wallets[0].id.unwrap_or_default(); + let participations_result = + crate::list_wallet_participations_by_wallet_id(database.as_ref(), wallet_id).await; + let participations = match participations_result { + Ok(participations) => participations, + Err(error) => panic!("participation list must succeed: {}", error), + }; + assert_eq!(participations.len(), 2); + } +}