diff --git a/CHANGELOG.md b/CHANGELOG.md index 3bcef4a..f3b8664 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,3 +49,4 @@ 0.7.16 - Ajout d’une première couche trade events et pair metrics avec normalisation des swaps, agrégation par paire et branchement automatique dans le pipeline de résolution transactionnelle 0.7.17 - Ajout d’une première couche WS hybride avec collecte de cibles `programSubscribe` / `accountSubscribe` et persistance technique dédupliquée des notifications `logs / program / account` 0.7.18 - Ajout d’un premier backfill historique ciblé par token mint, basé sur `getSignaturesForAddress` + `getTransaction`, avec réutilisation du pipeline interne pour reconstruire transactions, pools, swaps, origins, wallets et métriques +0.7.19 - Ajout d’une première couche holdings observés avec agrégation par couple wallet/token et branchement automatique dans le pipeline de résolution transactionnelle diff --git a/Cargo.toml b/Cargo.toml index 0b4d920..0435d4e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ members = [ ] [workspace.package] -version = "0.7.18" +version = "0.7.19" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-bobobot" diff --git a/ROADMAP.md b/ROADMAP.md index 16f2bdc..bb62b9b 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -635,15 +635,13 @@ Réalisé : - conservation d’un périmètre ciblé sur des tokens encore actifs au lieu d’un scan exhaustif de la blockchain. ### 6.051. Version `0.7.19` — Holdings observés -Objectif : compléter la couche acteurs observés avec une première vision des balances et comptes token observés, sans viser encore un portefeuille temps réel exhaustif. +Réalisé : -À faire : - -- ajouter une table de holdings observés rattachés à `wallets` et `tokens`, -- distinguer snapshots observés et états recalculés, -- permettre l’enregistrement de balances observées depuis les payloads décodés ou les lectures RPC ciblées, -- rattacher un holding observé à une transaction, un slot et une source, -- préparer l’évolution future vers des reconstructions de portefeuille plus complètes. +- ajout d’une première table `wallet holdings` pour agréger les couples `wallet/token` observés, +- rattachement des holdings observés à la dernière transaction, au dernier decoded event, au dernier pool et au dernier pair connus, +- conservation d’un champ `balance_raw` optionnel sans prétendre encore reconstruire un portefeuille complet, +- alimentation automatique à partir des événements DEX déjà décodés et des wallets déjà observés, +- branchement automatique dans le pipeline de résolution transactionnelle. ### 6.052. Version `0.7.20` — Candles / OHLCV Objectif : compléter la première couche d’agrégats DEX avec des séries temporelles exploitables par paire. diff --git a/kb_lib/src/db.rs b/kb_lib/src/db.rs index 4db195e..b6c5a46 100644 --- a/kb_lib/src/db.rs +++ b/kb_lib/src/db.rs @@ -43,6 +43,7 @@ pub use dtos::KbTokenDto; pub use dtos::KbTokenMintEventDto; pub use dtos::KbTradeEventDto; pub use dtos::KbWalletDto; +pub use dtos::KbWalletHoldingDto; pub use dtos::KbWalletParticipationDto; pub use entities::KbAnalysisSignalEntity; pub use entities::KbChainInstructionEntity; @@ -72,6 +73,7 @@ pub use entities::KbTokenEntity; pub use entities::KbTokenMintEventEntity; pub use entities::KbTradeEventEntity; pub use entities::KbWalletEntity; +pub use entities::KbWalletHoldingEntity; pub use entities::KbWalletParticipationEntity; pub use queries::delete_chain_instructions_by_transaction_id; pub use queries::get_chain_slot; @@ -93,6 +95,7 @@ pub use queries::get_pool_origin_by_pool_id; pub use queries::get_token_by_mint; pub use queries::get_trade_event_by_decoded_event_id; pub use queries::get_wallet_by_address; +pub use queries::get_wallet_holding_by_wallet_and_token; pub use queries::get_wallet_participation_by_unique_key; pub use queries::insert_analysis_signal; pub use queries::insert_chain_instruction; @@ -124,6 +127,7 @@ 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_trade_events_by_pair_id; +pub use queries::list_wallet_holdings_by_wallet_id; pub use queries::list_wallet_participations_by_pool_id; pub use queries::list_wallet_participations_by_wallet_id; pub use queries::list_wallets; @@ -151,6 +155,7 @@ pub use queries::upsert_token_burn_event; pub use queries::upsert_token_mint_event; pub use queries::upsert_trade_event; pub use queries::upsert_wallet; +pub use queries::upsert_wallet_holding; pub use queries::upsert_wallet_participation; pub use types::KbAnalysisSignalSeverity; pub use types::KbDatabaseBackend; diff --git a/kb_lib/src/db/dtos.rs b/kb_lib/src/db/dtos.rs index 8dcaaf2..78c0748 100644 --- a/kb_lib/src/db/dtos.rs +++ b/kb_lib/src/db/dtos.rs @@ -30,6 +30,7 @@ mod token_burn_event; mod token_mint_event; mod trade_event; mod wallet; +mod wallet_holding; mod wallet_participation; pub use analysis_signal::KbAnalysisSignalDto; @@ -60,4 +61,5 @@ pub use token_burn_event::KbTokenBurnEventDto; pub use token_mint_event::KbTokenMintEventDto; pub use trade_event::KbTradeEventDto; pub use wallet::KbWalletDto; +pub use wallet_holding::KbWalletHoldingDto; pub use wallet_participation::KbWalletParticipationDto; diff --git a/kb_lib/src/db/dtos/wallet_holding.rs b/kb_lib/src/db/dtos/wallet_holding.rs new file mode 100644 index 0000000..8a4e913 --- /dev/null +++ b/kb_lib/src/db/dtos/wallet_holding.rs @@ -0,0 +1,124 @@ +// file: kb_lib/src/db/dtos/wallet_holding.rs + +//! Wallet-holding DTO. + +/// Application-facing wallet-holding DTO. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct KbWalletHoldingDto { + /// Optional numeric primary key. + pub id: std::option::Option, + /// Related wallet id. + pub wallet_id: i64, + /// Related token id. + pub token_id: i64, + /// First transaction id that observed this wallet/token relationship. + pub first_transaction_id: i64, + /// Last transaction id that observed this wallet/token relationship. + pub last_transaction_id: i64, + /// Optional last decoded-event id. + pub last_decoded_event_id: std::option::Option, + /// Optional last related pool id. + pub last_pool_id: std::option::Option, + /// Optional last related pair id. + pub last_pair_id: std::option::Option, + /// Optional last observed role. + pub last_role: std::option::Option, + /// Optional last observed raw balance-like value. + pub balance_raw: std::option::Option, + /// Optional last observed slot. + pub last_slot_observed: std::option::Option, + /// 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 KbWalletHoldingDto { + /// Creates a new wallet-holding DTO. + pub fn new( + wallet_id: i64, + token_id: i64, + first_transaction_id: i64, + last_transaction_id: i64, + last_decoded_event_id: std::option::Option, + last_pool_id: std::option::Option, + last_pair_id: std::option::Option, + last_role: std::option::Option, + balance_raw: std::option::Option, + last_slot_observed: std::option::Option, + source_kind: crate::KbObservationSourceKind, + source_endpoint_name: std::option::Option, + ) -> Self { + let now = chrono::Utc::now(); + Self { + id: None, + wallet_id, + token_id, + first_transaction_id, + last_transaction_id, + last_decoded_event_id, + last_pool_id, + last_pair_id, + last_role, + balance_raw, + last_slot_observed, + source_kind, + source_endpoint_name, + created_at: now, + updated_at: now, + } + } +} + +impl TryFrom for KbWalletHoldingDto { + type Error = crate::KbError; + + fn try_from(entity: crate::KbWalletHoldingEntity) -> 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_holding 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_holding updated_at '{}': {}", + entity.updated_at, error + ))); + } + }; + Ok(Self { + id: Some(entity.id), + wallet_id: entity.wallet_id, + token_id: entity.token_id, + first_transaction_id: entity.first_transaction_id, + last_transaction_id: entity.last_transaction_id, + last_decoded_event_id: entity.last_decoded_event_id, + last_pool_id: entity.last_pool_id, + last_pair_id: entity.last_pair_id, + last_role: entity.last_role, + balance_raw: entity.balance_raw, + last_slot_observed: entity.last_slot_observed, + source_kind, + source_endpoint_name: entity.source_endpoint_name, + created_at, + updated_at, + }) + } +} diff --git a/kb_lib/src/db/entities.rs b/kb_lib/src/db/entities.rs index c1557f9..d5895cb 100644 --- a/kb_lib/src/db/entities.rs +++ b/kb_lib/src/db/entities.rs @@ -32,6 +32,7 @@ mod token_burn_event; mod token_mint_event; mod trade_event; mod wallet; +mod wallet_holding; mod wallet_participation; pub use analysis_signal::KbAnalysisSignalEntity; @@ -62,4 +63,5 @@ pub use token_burn_event::KbTokenBurnEventEntity; pub use token_mint_event::KbTokenMintEventEntity; pub use trade_event::KbTradeEventEntity; pub use wallet::KbWalletEntity; +pub use wallet_holding::KbWalletHoldingEntity; pub use wallet_participation::KbWalletParticipationEntity; diff --git a/kb_lib/src/db/entities/wallet_holding.rs b/kb_lib/src/db/entities/wallet_holding.rs new file mode 100644 index 0000000..9e70eb3 --- /dev/null +++ b/kb_lib/src/db/entities/wallet_holding.rs @@ -0,0 +1,38 @@ +// file: kb_lib/src/db/entities/wallet_holding.rs + +//! Wallet-holding entity. + +/// Persisted wallet-holding row. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)] +pub struct KbWalletHoldingEntity { + /// Numeric primary key. + pub id: i64, + /// Related wallet id. + pub wallet_id: i64, + /// Related token id. + pub token_id: i64, + /// First transaction id that observed this wallet/token relationship. + pub first_transaction_id: i64, + /// Last transaction id that observed this wallet/token relationship. + pub last_transaction_id: i64, + /// Optional last decoded-event id. + pub last_decoded_event_id: std::option::Option, + /// Optional last related pool id. + pub last_pool_id: std::option::Option, + /// Optional last related pair id. + pub last_pair_id: std::option::Option, + /// Optional last observed role. + pub last_role: std::option::Option, + /// Optional last observed raw balance-like value. + pub balance_raw: std::option::Option, + /// Optional last observed slot. + pub last_slot_observed: std::option::Option, + /// 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 77b041c..0cfdb72 100644 --- a/kb_lib/src/db/queries.rs +++ b/kb_lib/src/db/queries.rs @@ -34,6 +34,7 @@ mod token_burn_event; mod token_mint_event; mod trade_event; mod wallet; +mod wallet_holding; mod wallet_participation; pub use analysis_signal::insert_analysis_signal; @@ -111,6 +112,9 @@ pub use trade_event::upsert_trade_event; pub use wallet::get_wallet_by_address; pub use wallet::list_wallets; pub use wallet::upsert_wallet; +pub use wallet_holding::get_wallet_holding_by_wallet_and_token; +pub use wallet_holding::list_wallet_holdings_by_wallet_id; +pub use wallet_holding::upsert_wallet_holding; 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; diff --git a/kb_lib/src/db/queries/wallet_holding.rs b/kb_lib/src/db/queries/wallet_holding.rs new file mode 100644 index 0000000..cc07e57 --- /dev/null +++ b/kb_lib/src/db/queries/wallet_holding.rs @@ -0,0 +1,195 @@ +// file: kb_lib/src/db/queries/wallet_holding.rs + +//! Queries for `kb_wallet_holdings`. + +/// Inserts or updates one wallet-holding row and returns its stable internal id. +pub async fn upsert_wallet_holding( + database: &crate::KbDatabase, + dto: &crate::KbWalletHoldingDto, +) -> Result { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query( + r#" +INSERT INTO kb_wallet_holdings ( + wallet_id, + token_id, + first_transaction_id, + last_transaction_id, + last_decoded_event_id, + last_pool_id, + last_pair_id, + last_role, + balance_raw, + last_slot_observed, + source_kind, + source_endpoint_name, + created_at, + updated_at +) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(wallet_id, token_id) DO UPDATE SET + last_transaction_id = excluded.last_transaction_id, + last_decoded_event_id = excluded.last_decoded_event_id, + last_pool_id = excluded.last_pool_id, + last_pair_id = excluded.last_pair_id, + last_role = excluded.last_role, + balance_raw = COALESCE(excluded.balance_raw, kb_wallet_holdings.balance_raw), + last_slot_observed = excluded.last_slot_observed, + source_kind = excluded.source_kind, + source_endpoint_name = excluded.source_endpoint_name, + updated_at = excluded.updated_at + "#, + ) + .bind(dto.wallet_id) + .bind(dto.token_id) + .bind(dto.first_transaction_id) + .bind(dto.last_transaction_id) + .bind(dto.last_decoded_event_id) + .bind(dto.last_pool_id) + .bind(dto.last_pair_id) + .bind(dto.last_role.clone()) + .bind(dto.balance_raw.clone()) + .bind(dto.last_slot_observed) + .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_holdings on sqlite: {}", + error + ))); + } + let id_result = sqlx::query_scalar::( + r#" +SELECT id +FROM kb_wallet_holdings +WHERE wallet_id = ? AND token_id = ? +LIMIT 1 + "#, + ) + .bind(dto.wallet_id) + .bind(dto.token_id) + .fetch_one(pool) + .await; + match id_result { + Ok(id) => Ok(id), + Err(error) => Err(crate::KbError::Db(format!( + "cannot fetch kb_wallet_holdings id for wallet_id '{}' token_id '{}' on sqlite: {}", + dto.wallet_id, dto.token_id, error + ))), + } + } + } +} + +/// Returns one wallet-holding row identified by `(wallet_id, token_id)`, if it exists. +pub async fn get_wallet_holding_by_wallet_and_token( + database: &crate::KbDatabase, + wallet_id: i64, + token_id: i64, +) -> Result, crate::KbError> { + match database.connection() { + crate::KbDatabaseConnection::Sqlite(pool) => { + let query_result = sqlx::query_as::( + r#" +SELECT + id, + wallet_id, + token_id, + first_transaction_id, + last_transaction_id, + last_decoded_event_id, + last_pool_id, + last_pair_id, + last_role, + balance_raw, + last_slot_observed, + source_kind, + source_endpoint_name, + created_at, + updated_at +FROM kb_wallet_holdings +WHERE wallet_id = ? AND token_id = ? +LIMIT 1 + "#, + ) + .bind(wallet_id) + .bind(token_id) + .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_holdings by wallet_id '{}' token_id '{}' on sqlite: {}", + wallet_id, token_id, error + ))); + } + }; + match entity_option { + Some(entity) => crate::KbWalletHoldingDto::try_from(entity).map(Some), + None => Ok(None), + } + } + } +} + +/// Lists wallet-holding rows for one wallet id. +pub async fn list_wallet_holdings_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, + token_id, + first_transaction_id, + last_transaction_id, + last_decoded_event_id, + last_pool_id, + last_pair_id, + last_role, + balance_raw, + last_slot_observed, + source_kind, + source_endpoint_name, + created_at, + updated_at +FROM kb_wallet_holdings +WHERE wallet_id = ? +ORDER BY token_id 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_holdings by wallet_id '{}' on sqlite: {}", + wallet_id, error + ))); + } + }; + let mut dtos = std::vec::Vec::new(); + for entity in entities { + let dto_result = crate::KbWalletHoldingDto::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 57f8609..f716bcd 100644 --- a/kb_lib/src/db/schema.rs +++ b/kb_lib/src/db/schema.rs @@ -310,6 +310,18 @@ pub(crate) async fn ensure_schema(database: &crate::KbDatabase) -> Result<(), cr if let Err(error) = result { return Err(error); } + let result = create_kb_wallet_holdings_table(pool).await; + if let Err(error) = result { + return Err(error); + } + let result = create_kb_idx_wallet_holdings_wallet_id(pool).await; + if let Err(error) = result { + return Err(error); + } + let result = create_kb_idx_wallet_holdings_token_id(pool).await; + if let Err(error) = result { + return Err(error); + } Ok(()) } } @@ -1713,3 +1725,66 @@ CREATE TABLE IF NOT EXISTS kb_pair_metrics ( ) .await } + +async fn create_kb_wallet_holdings_table(pool: &sqlx::SqlitePool) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_wallet_holdings_table", + r#" +CREATE TABLE IF NOT EXISTS kb_wallet_holdings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + wallet_id INTEGER NOT NULL, + token_id INTEGER NOT NULL, + first_transaction_id INTEGER NOT NULL, + last_transaction_id INTEGER NOT NULL, + last_decoded_event_id INTEGER NULL, + last_pool_id INTEGER NULL, + last_pair_id INTEGER NULL, + last_role TEXT NULL, + balance_raw TEXT NULL, + last_slot_observed INTEGER NULL, + source_kind INTEGER NOT NULL, + source_endpoint_name TEXT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(wallet_id, token_id), + FOREIGN KEY(wallet_id) REFERENCES kb_wallets(id) ON DELETE CASCADE, + FOREIGN KEY(token_id) REFERENCES kb_tokens(id) ON DELETE CASCADE, + FOREIGN KEY(first_transaction_id) REFERENCES kb_chain_transactions(id) ON DELETE CASCADE, + FOREIGN KEY(last_transaction_id) REFERENCES kb_chain_transactions(id) ON DELETE CASCADE, + FOREIGN KEY(last_decoded_event_id) REFERENCES kb_dex_decoded_events(id) ON DELETE SET NULL, + FOREIGN KEY(last_pool_id) REFERENCES kb_pools(id) ON DELETE SET NULL, + FOREIGN KEY(last_pair_id) REFERENCES kb_pairs(id) ON DELETE SET NULL +) + "#, + ) + .await +} + +async fn create_kb_idx_wallet_holdings_wallet_id( + pool: &sqlx::SqlitePool, +) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_idx_wallet_holdings_wallet_id", + r#" +CREATE INDEX IF NOT EXISTS kb_idx_wallet_holdings_wallet_id +ON kb_wallet_holdings(wallet_id) + "#, + ) + .await +} + +async fn create_kb_idx_wallet_holdings_token_id( + pool: &sqlx::SqlitePool, +) -> Result<(), crate::KbError> { + execute_sqlite_schema_statement( + pool, + "create_kb_idx_wallet_holdings_token_id", + r#" +CREATE INDEX IF NOT EXISTS kb_idx_wallet_holdings_token_id +ON kb_wallet_holdings(token_id) + "#, + ) + .await +} diff --git a/kb_lib/src/lib.rs b/kb_lib/src/lib.rs index 82045f8..04c5806 100644 --- a/kb_lib/src/lib.rs +++ b/kb_lib/src/lib.rs @@ -28,6 +28,7 @@ mod trade_aggregation; mod tx_model; mod tx_resolution; mod types; +mod wallet_holding_observation; mod wallet_observation; mod ws_client; mod ws_hybrid_observation; @@ -113,6 +114,8 @@ pub use db::KbTradeEventDto; pub use db::KbTradeEventEntity; pub use db::KbWalletDto; pub use db::KbWalletEntity; +pub use db::KbWalletHoldingDto; +pub use db::KbWalletHoldingEntity; pub use db::KbWalletParticipationDto; pub use db::KbWalletParticipationEntity; pub use db::delete_chain_instructions_by_transaction_id; @@ -135,6 +138,7 @@ pub use db::get_pool_origin_by_pool_id; pub use db::get_token_by_mint; pub use db::get_trade_event_by_decoded_event_id; pub use db::get_wallet_by_address; +pub use db::get_wallet_holding_by_wallet_and_token; pub use db::get_wallet_participation_by_unique_key; pub use db::insert_analysis_signal; pub use db::insert_chain_instruction; @@ -166,6 +170,7 @@ 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_trade_events_by_pair_id; +pub use db::list_wallet_holdings_by_wallet_id; pub use db::list_wallet_participations_by_pool_id; pub use db::list_wallet_participations_by_wallet_id; pub use db::list_wallets; @@ -193,6 +198,7 @@ pub use db::upsert_token_burn_event; pub use db::upsert_token_mint_event; pub use db::upsert_trade_event; pub use db::upsert_wallet; +pub use db::upsert_wallet_holding; pub use db::upsert_wallet_participation; pub use detect::KbDetectionObservationInput; pub use detect::KbDetectionPersistenceService; @@ -295,6 +301,8 @@ pub use tx_resolution::KbWsTransactionResolutionEnvelope; pub use tx_resolution::KbWsTransactionResolutionRelay; pub use tx_resolution::KbWsTransactionResolutionRelayStats; pub use types::KbConnectionState; +pub use wallet_holding_observation::KbWalletHoldingObservationResult; +pub use wallet_holding_observation::KbWalletHoldingObservationService; pub use wallet_observation::KbWalletObservationResult; pub use wallet_observation::KbWalletObservationService; pub use ws_client::WsClient; diff --git a/kb_lib/src/tx_resolution.rs b/kb_lib/src/tx_resolution.rs index ec57805..06b89c2 100644 --- a/kb_lib/src/tx_resolution.rs +++ b/kb_lib/src/tx_resolution.rs @@ -106,6 +106,7 @@ pub struct KbTransactionResolutionService { pool_origin_service: crate::KbPoolOriginService, wallet_observation_service: crate::KbWalletObservationService, trade_aggregation_service: crate::KbTradeAggregationService, + wallet_holding_observation_service: crate::KbWalletHoldingObservationService, resolved_signatures: std::sync::Arc>>, } @@ -125,6 +126,8 @@ impl KbTransactionResolutionService { let pool_origin_service = crate::KbPoolOriginService::new(database.clone()); let wallet_observation_service = crate::KbWalletObservationService::new(database.clone()); let trade_aggregation_service = crate::KbTradeAggregationService::new(database.clone()); + let wallet_holding_observation_service = + crate::KbWalletHoldingObservationService::new(database.clone()); Self { http_pool, persistence, @@ -136,6 +139,7 @@ impl KbTransactionResolutionService { pool_origin_service, wallet_observation_service, trade_aggregation_service, + wallet_holding_observation_service, resolved_signatures: std::sync::Arc::new(tokio::sync::Mutex::new( std::collections::HashSet::new(), )), @@ -353,6 +357,15 @@ impl KbTransactionResolutionService { Err(error) => return Err(error), }; let wallet_participation_count = wallet_observations.len(); + let wallet_holding_observations_result = self + .wallet_holding_observation_service + .record_transaction_by_signature(request.signature.as_str()) + .await; + let wallet_holding_observations = match wallet_holding_observations_result { + Ok(wallet_holding_observations) => wallet_holding_observations, + Err(error) => return Err(error), + }; + let wallet_holding_count = wallet_holding_observations.len(); let trade_aggregations_result = self .trade_aggregation_service .record_transaction_by_signature(request.signature.as_str()) @@ -374,6 +387,7 @@ impl KbTransactionResolutionService { "launchAttributionCount": launch_attribution_count, "poolOriginCount": pool_origin_count, "walletParticipationCount": wallet_participation_count, + "walletHoldingCount": wallet_holding_count, "tradeEventCount": trade_event_count, "triggerPayload": request.trigger_payload.clone(), "transaction": transaction_value diff --git a/kb_lib/src/wallet_holding_observation.rs b/kb_lib/src/wallet_holding_observation.rs new file mode 100644 index 0000000..875ee6d --- /dev/null +++ b/kb_lib/src/wallet_holding_observation.rs @@ -0,0 +1,543 @@ +// file: kb_lib/src/wallet_holding_observation.rs + +//! Wallet-holding observation service. + +/// One wallet-holding observation result. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct KbWalletHoldingObservationResult { + /// Persisted wallet-holding id. + pub wallet_holding_id: i64, + /// Related wallet id. + pub wallet_id: i64, + /// Related token id. + pub token_id: i64, + /// Whether the holding row was newly created. + pub created_holding: bool, +} + +/// Wallet-holding observation service. +#[derive(Debug, Clone)] +pub struct KbWalletHoldingObservationService { + database: std::sync::Arc, + persistence: crate::KbDetectionPersistenceService, +} + +impl KbWalletHoldingObservationService { + /// Creates a new wallet-holding observation service. + pub fn new(database: std::sync::Arc) -> Self { + let persistence = crate::KbDetectionPersistenceService::new(database.clone()); + Self { + database, + persistence, + } + } + + /// Records observed wallet holdings 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 holdings 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(); + let mut seen_pairs = + std::collections::HashSet::<(i64, i64, std::option::Option)>::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 wallet_roles = kb_collect_wallet_roles(&payload); + let token_mints = kb_collect_token_mints(decoded_event); + for (role, wallet_address) in wallet_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 = match wallet_option { + Some(wallet) => wallet, + None => continue, + }; + let wallet_id = match wallet.id { + Some(wallet_id) => wallet_id, + None => { + return Err(crate::KbError::InvalidState( + "wallet has no internal id".to_string(), + )); + } + }; + for token_mint in &token_mints { + let token_result = + crate::get_token_by_mint(self.database.as_ref(), token_mint.as_str()).await; + let token_option = match token_result { + Ok(token_option) => token_option, + Err(error) => return Err(error), + }; + let token = match token_option { + Some(token) => token, + None => continue, + }; + let token_id = match token.id { + Some(token_id) => token_id, + None => { + return Err(crate::KbError::InvalidState( + "token has no internal id".to_string(), + )); + } + }; + let dedupe_key = (wallet_id, token_id, Some(decoded_event_id)); + if seen_pairs.contains(&dedupe_key) { + continue; + } + seen_pairs.insert(dedupe_key); + let existing_result = crate::get_wallet_holding_by_wallet_and_token( + self.database.as_ref(), + wallet_id, + token_id, + ) + .await; + let existing_option = match existing_result { + Ok(existing_option) => existing_option, + Err(error) => return Err(error), + }; + let created_holding = existing_option.is_none(); + let first_transaction_id = match existing_option.clone() { + Some(existing) => existing.first_transaction_id, + None => transaction_id, + }; + let slot_i64 = kb_convert_slot_to_i64(transaction.slot); + let dto = crate::KbWalletHoldingDto::new( + wallet_id, + token_id, + first_transaction_id, + transaction_id, + Some(decoded_event_id), + pool_id, + pair_id, + Some(role.clone()), + None, + slot_i64, + crate::KbObservationSourceKind::Dex, + transaction.source_endpoint_name.clone(), + ); + let upsert_result = + crate::upsert_wallet_holding(self.database.as_ref(), &dto).await; + let wallet_holding_id = match upsert_result { + Ok(wallet_holding_id) => wallet_holding_id, + Err(error) => return Err(error), + }; + if created_holding { + let payload = serde_json::json!({ + "walletAddress": wallet_address, + "tokenMint": token_mint, + "role": role, + "transactionSignature": transaction.signature, + "poolId": pool_id, + "pairId": pair_id + }); + let observation_result = self + .persistence + .record_observation(&crate::KbDetectionObservationInput::new( + "wallet.holding".to_string(), + crate::KbObservationSourceKind::Dex, + 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.holding.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::KbWalletHoldingObservationResult { + wallet_holding_id, + wallet_id, + token_id, + created_holding, + }); + } + } + } + Ok(results) + } +} + +fn kb_collect_wallet_roles( + payload: &serde_json::Value, +) -> std::vec::Vec<(std::string::String, std::string::String)> { + let role_mappings = [ + ("payer", vec!["payer", "funder"]), + ("creator", vec!["creator", "poolCreator"]), + ("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_collect_token_mints( + decoded_event: &crate::KbDexDecodedEventDto, +) -> std::vec::Vec { + let mut seen = std::collections::HashSet::::new(); + let mut values = std::vec::Vec::new(); + if let Some(token_a_mint) = decoded_event.token_a_mint.clone() { + if !seen.contains(token_a_mint.as_str()) { + seen.insert(token_a_mint.clone()); + values.push(token_a_mint); + } + } + if let Some(token_b_mint) = decoded_event.token_b_mint.clone() { + if !seen.contains(token_b_mint.as_str()) { + seen.insert(token_b_mint.clone()); + values.push(token_b_mint); + } + } + + values +} + +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); + } + } +} + +fn kb_convert_slot_to_i64(slot: std::option::Option) -> std::option::Option { + match slot { + Some(slot) => match i64::try_from(slot) { + Ok(slot) => Some(slot), + Err(_) => None, + }, + None => None, + } +} + +#[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_holding_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_create_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.clone()); + let wallet_observation = crate::KbWalletObservationService::new(database.clone()); + let resolved_transaction = serde_json::json!({ + "slot": 950001, + "blockTime": 1779600001, + "version": 0, + "transaction": { + "message": { + "instructions": [ + { + "programId": crate::KB_FLUXBEAM_PROGRAM_ID, + "program": "fluxbeam", + "stackHeight": 1, + "accounts": [ + "HoldingPool111", + "HoldingLpMint111", + "HoldingTokenA111", + "So11111111111111111111111111111111111111112", + "HoldingWallet111" + ], + "parsed": { + "info": { + "instruction": "create_pool", + "pool": "HoldingPool111", + "lpMint": "HoldingLpMint111", + "tokenA": "HoldingTokenA111", + "tokenB": "So11111111111111111111111111111111111111112", + "payer": "HoldingWallet111" + } + }, + "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); + } + let wallet_result = wallet_observation + .record_transaction_by_signature(signature) + .await; + if let Err(error) = wallet_result { + panic!("wallet observation must succeed: {}", error); + } + } + + #[tokio::test] + async fn record_transaction_by_signature_creates_wallet_holdings() { + let database = make_database().await; + seed_fluxbeam_create_transaction(database.clone(), "sig-wallet-holding-1").await; + let service = crate::KbWalletHoldingObservationService::new(database.clone()); + let record_result = service + .record_transaction_by_signature("sig-wallet-holding-1") + .await; + let results = match record_result { + Ok(results) => results, + Err(error) => panic!("wallet holding observation must succeed: {}", error), + }; + assert_eq!(results.len(), 2); + let wallet_result = + crate::get_wallet_by_address(database.as_ref(), "HoldingWallet111").await; + let wallet_option = match wallet_result { + Ok(wallet_option) => wallet_option, + Err(error) => panic!("wallet fetch must succeed: {}", error), + }; + let wallet = match wallet_option { + Some(wallet) => wallet, + None => panic!("wallet must exist"), + }; + let wallet_id = match wallet.id { + Some(wallet_id) => wallet_id, + None => panic!("wallet must have an id"), + }; + let holdings_result = + crate::list_wallet_holdings_by_wallet_id(database.as_ref(), wallet_id).await; + let holdings = match holdings_result { + Ok(holdings) => holdings, + Err(error) => panic!("wallet holding list must succeed: {}", error), + }; + assert_eq!(holdings.len(), 2); + let mut token_ids = std::vec::Vec::new(); + for holding in &holdings { + token_ids.push(holding.token_id); + assert_eq!(holding.last_role, Some("payer".to_string())); + } + token_ids.sort(); + assert_eq!(token_ids.len(), 2); + } + + #[tokio::test] + async fn record_transaction_by_signature_is_idempotent() { + let database = make_database().await; + seed_fluxbeam_create_transaction(database.clone(), "sig-wallet-holding-2").await; + let service = crate::KbWalletHoldingObservationService::new(database.clone()); + let first_result = service + .record_transaction_by_signature("sig-wallet-holding-2") + .await; + let first_results = match first_result { + Ok(first_results) => first_results, + Err(error) => panic!("first wallet holding observation must succeed: {}", error), + }; + assert_eq!(first_results.len(), 2); + assert!(first_results[0].created_holding); + assert!(first_results[1].created_holding); + let second_result = service + .record_transaction_by_signature("sig-wallet-holding-2") + .await; + let second_results = match second_result { + Ok(second_results) => second_results, + Err(error) => panic!("second wallet holding observation must succeed: {}", error), + }; + assert_eq!(second_results.len(), 2); + assert!(!second_results[0].created_holding); + assert!(!second_results[1].created_holding); + let wallet_result = + crate::get_wallet_by_address(database.as_ref(), "HoldingWallet111").await; + let wallet_option = match wallet_result { + Ok(wallet_option) => wallet_option, + Err(error) => panic!("wallet fetch must succeed: {}", error), + }; + let wallet = match wallet_option { + Some(wallet) => wallet, + None => panic!("wallet must exist"), + }; + let wallet_id = wallet.id.unwrap_or_default(); + let holdings_result = + crate::list_wallet_holdings_by_wallet_id(database.as_ref(), wallet_id).await; + let holdings = match holdings_result { + Ok(holdings) => holdings, + Err(error) => panic!("wallet holding list must succeed: {}", error), + }; + assert_eq!(holdings.len(), 2); + } +}