0.7.15

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  

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"

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.

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;

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;

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<i64>,
+    /// Stable wallet address.
+    pub address: std::string::String,
+    /// Optional user-defined label.
+    pub label: std::option::Option<std::string::String>,
+    /// Creation timestamp.
+    pub first_seen_at: chrono::DateTime<chrono::Utc>,
+    /// Update timestamp.
+    pub last_seen_at: chrono::DateTime<chrono::Utc>,
+}
+
+impl KbWalletDto {
+    /// Creates a new wallet DTO.
+    pub fn new(
+        address: std::string::String,
+        label: std::option::Option<std::string::String>,
+    ) -> Self {
+        let now = chrono::Utc::now();
+        Self {
+            id: None,
+            address,
+            label,
+            first_seen_at: now,
+            last_seen_at: now,
+        }
+    }
+}
+
+impl TryFrom<crate::KbWalletEntity> for KbWalletDto {
+    type Error = crate::KbError;
+
+    fn try_from(entity: crate::KbWalletEntity) -> Result<Self, Self::Error> {
+        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,
+        })
+    }
+}

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<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<i64>,
+    /// Optional related pool id.
+    pub pool_id: std::option::Option<i64>,
+    /// Optional related pair id.
+    pub pair_id: std::option::Option<i64>,
+    /// 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<std::string::String>,
+    /// Creation timestamp.
+    pub created_at: chrono::DateTime<chrono::Utc>,
+    /// Update timestamp.
+    pub updated_at: chrono::DateTime<chrono::Utc>,
+}
+
+impl KbWalletParticipationDto {
+    /// Creates a new wallet-participation DTO.
+    pub fn new(
+        wallet_id: i64,
+        transaction_id: i64,
+        decoded_event_id: std::option::Option<i64>,
+        pool_id: std::option::Option<i64>,
+        pair_id: std::option::Option<i64>,
+        role: std::string::String,
+        source_kind: crate::KbObservationSourceKind,
+        source_endpoint_name: std::option::Option<std::string::String>,
+    ) -> 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<crate::KbWalletParticipationEntity> for KbWalletParticipationDto {
+    type Error = crate::KbError;
+
+    fn try_from(entity: crate::KbWalletParticipationEntity) -> 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_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<i64>,
+    pool_id: std::option::Option<i64>,
+    pair_id: std::option::Option<i64>,
+    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
+    )
+}

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;

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<std::string::String>,
+    /// 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,
+}

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<i64>,
+    /// Optional related pool id.
+    pub pool_id: std::option::Option<i64>,
+    /// Optional related pair id.
+    pub pair_id: std::option::Option<i64>,
+    /// 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<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

@@ -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;

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

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

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

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;

kb_lib/src/tx_resolution.rs

@@ -98,12 +98,13 @@ pub struct KbWsTransactionResolutionRelayStats {
 pub struct KbTransactionResolutionService {
     http_pool: std::sync::Arc<crate::HttpEndpointPool>,
     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<tokio::sync::Mutex<std::collections::HashSet<std::string::String>>>,
 }
@@ -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
         });

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<i64>,
+    /// Optional related pair id.
+    pub pair_id: std::option::Option<i64>,
+    /// Whether the participation row was newly created.
+    pub created_participation: bool,
+}
+
+/// Wallet-observation service.
+#[derive(Debug, Clone)]
+pub struct KbWalletObservationService {
+    database: std::sync::Arc<crate::KbDatabase>,
+    persistence: crate::KbDetectionPersistenceService,
+}
+
+impl KbWalletObservationService {
+    /// Creates a new wallet-observation service.
+    pub fn new(database: std::sync::Arc<crate::KbDatabase>) -> 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<std::vec::Vec<crate::KbWalletObservationResult>, 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::<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 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::<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_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);
+        }
+    }
+}
+
+#[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_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<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);
+        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);
+    }
+}