0.7.19

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;

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;

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<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<i64>,
+    /// Optional last related pool id.
+    pub last_pool_id: std::option::Option<i64>,
+    /// Optional last related pair id.
+    pub last_pair_id: std::option::Option<i64>,
+    /// Optional last observed role.
+    pub last_role: std::option::Option<std::string::String>,
+    /// Optional last observed raw balance-like value.
+    pub balance_raw: std::option::Option<std::string::String>,
+    /// Optional last observed slot.
+    pub last_slot_observed: std::option::Option<i64>,
+    /// Observation source kind.
+    pub source_kind: crate::KbObservationSourceKind,
+    /// Optional logical source endpoint name.
+    pub source_endpoint_name: std::option::Option<std::string::String>,
+    /// Creation timestamp.
+    pub created_at: chrono::DateTime<chrono::Utc>,
+    /// Update timestamp.
+    pub updated_at: chrono::DateTime<chrono::Utc>,
+}
+
+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<i64>,
+        last_pool_id: std::option::Option<i64>,
+        last_pair_id: std::option::Option<i64>,
+        last_role: std::option::Option<std::string::String>,
+        balance_raw: std::option::Option<std::string::String>,
+        last_slot_observed: std::option::Option<i64>,
+        source_kind: crate::KbObservationSourceKind,
+        source_endpoint_name: std::option::Option<std::string::String>,
+    ) -> 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<crate::KbWalletHoldingEntity> for KbWalletHoldingDto {
+    type Error = crate::KbError;
+
+    fn try_from(entity: crate::KbWalletHoldingEntity) -> Result<Self, Self::Error> {
+        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,
+        })
+    }
+}

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;

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<i64>,
+    /// Optional last related pool id.
+    pub last_pool_id: std::option::Option<i64>,
+    /// Optional last related pair id.
+    pub last_pair_id: std::option::Option<i64>,
+    /// Optional last observed role.
+    pub last_role: std::option::Option<std::string::String>,
+    /// Optional last observed raw balance-like value.
+    pub balance_raw: std::option::Option<std::string::String>,
+    /// Optional last observed slot.
+    pub last_slot_observed: std::option::Option<i64>,
+    /// Observation source kind.
+    pub source_kind: i16,
+    /// Optional logical source endpoint name.
+    pub source_endpoint_name: std::option::Option<std::string::String>,
+    /// 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,
+}

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;

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<i64, crate::KbError> {
+    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::<sqlx::Sqlite, i64>(
+                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<std::option::Option<crate::KbWalletHoldingDto>, crate::KbError> {
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let query_result = sqlx::query_as::<sqlx::Sqlite, crate::KbWalletHoldingEntity>(
+                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<std::vec::Vec<crate::KbWalletHoldingDto>, crate::KbError> {
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let query_result = sqlx::query_as::<sqlx::Sqlite, crate::KbWalletHoldingEntity>(
+                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)
+        }
+    }
+}

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
+}

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;

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<tokio::sync::Mutex<std::collections::HashSet<std::string::String>>>,
 }
@@ -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

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<crate::KbDatabase>,
+    persistence: crate::KbDetectionPersistenceService,
+}
+
+impl KbWalletHoldingObservationService {
+    /// Creates a new wallet-holding observation service.
+    pub fn new(database: std::sync::Arc<crate::KbDatabase>) -> 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<std::vec::Vec<crate::KbWalletHoldingObservationResult>, 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<i64>)>::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::<serde_json::Value>(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::<std::string::String>::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<std::string::String> {
+    let mut seen = std::collections::HashSet::<std::string::String>::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<std::string::String> {
+    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<std::string::String>,
+) {
+    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<u64>) -> std::option::Option<i64> {
+    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<crate::KbDatabase> {
+        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<crate::KbDatabase>,
+        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);
+    }
+}