0.5.2

kb_lib/src/db.rs

@@ -19,20 +19,26 @@ pub use crate::db::dtos::KbDbMetadataDto;
 pub use crate::db::dtos::KbDbRuntimeEventDto;
 pub use crate::db::dtos::KbKnownHttpEndpointDto;
 pub use crate::db::dtos::KbKnownWsEndpointDto;
+pub use crate::db::dtos::KbObservedTokenDto;
 pub use crate::db::entities::KbDbMetadataEntity;
 pub use crate::db::entities::KbDbRuntimeEventEntity;
 pub use crate::db::entities::KbKnownHttpEndpointEntity;
 pub use crate::db::entities::KbKnownWsEndpointEntity;
+pub use crate::db::entities::KbObservedTokenEntity;
 pub use crate::db::queries::get_db_metadata;
 pub use crate::db::queries::get_known_http_endpoint;
 pub use crate::db::queries::get_known_ws_endpoint;
+pub use crate::db::queries::get_observed_token_by_mint;
 pub use crate::db::queries::insert_db_runtime_event;
 pub use crate::db::queries::list_db_metadata;
 pub use crate::db::queries::list_known_http_endpoints;
 pub use crate::db::queries::list_known_ws_endpoints;
+pub use crate::db::queries::list_observed_tokens;
 pub use crate::db::queries::list_recent_db_runtime_events;
 pub use crate::db::queries::upsert_db_metadata;
 pub use crate::db::queries::upsert_known_http_endpoint;
 pub use crate::db::queries::upsert_known_ws_endpoint;
+pub use crate::db::queries::upsert_observed_token;
 pub use crate::db::types::KbDatabaseBackend;
 pub use crate::db::types::KbDbRuntimeEventLevel;
+pub use crate::db::types::KbObservedTokenStatus;

kb_lib/src/db/dtos.rs

@@ -6,8 +6,10 @@ mod db_metadata;
 mod db_runtime_event;
 mod known_http_endpoint;
 mod known_ws_endpoint;
+mod observed_token;
 
 pub use crate::db::dtos::db_metadata::KbDbMetadataDto;
 pub use crate::db::dtos::db_runtime_event::KbDbRuntimeEventDto;
 pub use crate::db::dtos::known_http_endpoint::KbKnownHttpEndpointDto;
 pub use crate::db::dtos::known_ws_endpoint::KbKnownWsEndpointDto;
+pub use crate::db::dtos::observed_token::KbObservedTokenDto;

kb_lib/src/db/dtos/observed_token.rs

@@ -0,0 +1,129 @@
+// file: kb_lib/src/db/dtos/observed_token.rs
+
+//! Observed token DTO.
+
+/// Application-facing observed token DTO.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct KbObservedTokenDto {
+    /// Optional numeric primary key.
+    pub id: std::option::Option<i64>,
+    /// Token mint address.
+    pub mint: std::string::String,
+    /// Optional token symbol.
+    pub symbol: std::option::Option<std::string::String>,
+    /// Optional token display name.
+    pub name: std::option::Option<std::string::String>,
+    /// Optional decimals value.
+    pub decimals: std::option::Option<u8>,
+    /// Token program id.
+    pub token_program: std::string::String,
+    /// Local status.
+    pub status: crate::KbObservedTokenStatus,
+    /// First seen timestamp.
+    pub first_seen_at: chrono::DateTime<chrono::Utc>,
+    /// Last seen timestamp.
+    pub last_seen_at: chrono::DateTime<chrono::Utc>,
+    /// Last update timestamp.
+    pub updated_at: chrono::DateTime<chrono::Utc>,
+}
+
+impl KbObservedTokenDto {
+    /// Creates a new observed token DTO with current timestamps.
+    pub fn new(
+        mint: std::string::String,
+        symbol: std::option::Option<std::string::String>,
+        name: std::option::Option<std::string::String>,
+        decimals: std::option::Option<u8>,
+        token_program: std::string::String,
+        status: crate::KbObservedTokenStatus,
+    ) -> Self {
+        let now = chrono::Utc::now();
+        Self {
+            id: None,
+            mint,
+            symbol,
+            name,
+            decimals,
+            token_program,
+            status,
+            first_seen_at: now,
+            last_seen_at: now,
+            updated_at: now,
+        }
+    }
+}
+
+impl TryFrom<crate::KbObservedTokenEntity> for KbObservedTokenDto {
+    type Error = crate::KbError;
+
+    fn try_from(
+        entity: crate::KbObservedTokenEntity,
+    ) -> Result<Self, Self::Error> {
+        let status_result = crate::KbObservedTokenStatus::from_i16(entity.status);
+        let status = match status_result {
+            Ok(status) => status,
+            Err(error) => return Err(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 observed token 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 observed token last_seen_at '{}': {}",
+                    entity.last_seen_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 observed token updated_at '{}': {}",
+                    entity.updated_at,
+                    error
+                )));
+            },
+        };
+        let decimals = match entity.decimals {
+            Some(decimals) => {
+                let decimals_u8_result = u8::try_from(decimals);
+                match decimals_u8_result {
+                    Ok(decimals_u8) => Some(decimals_u8),
+                    Err(error) => {
+                        return Err(crate::KbError::Db(format!(
+                            "cannot convert observed token decimals '{}' to u8: {}",
+                            decimals,
+                            error
+                        )));
+                    },
+                }
+            },
+            None => None,
+        };
+        Ok(Self {
+            id: Some(entity.id),
+            mint: entity.mint,
+            symbol: entity.symbol,
+            name: entity.name,
+            decimals,
+            token_program: entity.token_program,
+            status,
+            first_seen_at,
+            last_seen_at,
+            updated_at,
+        })
+    }
+}

kb_lib/src/db/entities.rs

@@ -8,8 +8,10 @@ mod db_metadata;
 mod db_runtime_event;
 mod known_http_endpoint;
 mod known_ws_endpoint;
+mod observed_token;
 
 pub use crate::db::entities::db_metadata::KbDbMetadataEntity;
 pub use crate::db::entities::db_runtime_event::KbDbRuntimeEventEntity;
 pub use crate::db::entities::known_http_endpoint::KbKnownHttpEndpointEntity;
 pub use crate::db::entities::known_ws_endpoint::KbKnownWsEndpointEntity;
+pub use crate::db::entities::observed_token::KbObservedTokenEntity;

kb_lib/src/db/entities/observed_token.rs

@@ -0,0 +1,28 @@
+// file: kb_lib/src/db/entities/observed_token.rs
+
+//! Observed token entity.
+
+/// Persisted observed token row.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)]
+pub struct KbObservedTokenEntity {
+    /// Numeric primary key.
+    pub id: i64,
+    /// Token mint address.
+    pub mint: std::string::String,
+    /// Optional token symbol.
+    pub symbol: std::option::Option<std::string::String>,
+    /// Optional token display name.
+    pub name: std::option::Option<std::string::String>,
+    /// Optional decimals value.
+    pub decimals: std::option::Option<i64>,
+    /// Token program id.
+    pub token_program: std::string::String,
+    /// Local status stored as stable integer.
+    pub status: i16,
+    /// First seen timestamp encoded as RFC3339 UTC text.
+    pub first_seen_at: std::string::String,
+    /// Last seen timestamp encoded as RFC3339 UTC text.
+    pub last_seen_at: std::string::String,
+    /// Last update timestamp encoded as RFC3339 UTC text.
+    pub updated_at: std::string::String,
+}

kb_lib/src/db/queries.rs

@@ -6,6 +6,7 @@ mod db_metadata;
 mod db_runtime_event;
 mod known_http_endpoint;
 mod known_ws_endpoint;
+mod observed_token;
 
 pub use crate::db::queries::db_metadata::get_db_metadata;
 pub use crate::db::queries::db_metadata::list_db_metadata;
@@ -18,3 +19,6 @@ pub use crate::db::queries::known_http_endpoint::upsert_known_http_endpoint;
 pub use crate::db::queries::known_ws_endpoint::get_known_ws_endpoint;
 pub use crate::db::queries::known_ws_endpoint::list_known_ws_endpoints;
 pub use crate::db::queries::known_ws_endpoint::upsert_known_ws_endpoint;
+pub use crate::db::queries::observed_token::get_observed_token_by_mint;
+pub use crate::db::queries::observed_token::list_observed_tokens;
+pub use crate::db::queries::observed_token::upsert_observed_token;

kb_lib/src/db/queries/observed_token.rs

@@ -0,0 +1,230 @@
+// file: kb_lib/src/db/queries/observed_token.rs
+
+//! Queries for `kb_observed_tokens`.
+
+/// Inserts or updates one observed token by mint.
+pub async fn upsert_observed_token(
+    database: &crate::KbDatabase,
+    dto: &crate::KbObservedTokenDto,
+) -> Result<i64, crate::KbError> {
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let decimals_i64 = dto.decimals.map(i64::from);
+            let insert_result = sqlx::query(
+                r#"
+INSERT INTO kb_observed_tokens (
+    mint,
+    symbol,
+    name,
+    decimals,
+    token_program,
+    status,
+    first_seen_at,
+    last_seen_at,
+    updated_at
+)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+ON CONFLICT(mint) DO UPDATE SET
+    symbol = excluded.symbol,
+    name = excluded.name,
+    decimals = excluded.decimals,
+    token_program = excluded.token_program,
+    status = excluded.status,
+    last_seen_at = excluded.last_seen_at,
+    updated_at = excluded.updated_at
+                "#,
+            )
+            .bind(dto.mint.clone())
+            .bind(dto.symbol.clone())
+            .bind(dto.name.clone())
+            .bind(decimals_i64)
+            .bind(dto.token_program.clone())
+            .bind(dto.status.to_i16())
+            .bind(dto.first_seen_at.to_rfc3339())
+            .bind(dto.last_seen_at.to_rfc3339())
+            .bind(dto.updated_at.to_rfc3339())
+            .execute(pool)
+            .await;
+            if let Err(error) = insert_result {
+                return Err(crate::KbError::Db(format!(
+                    "cannot upsert kb_observed_tokens on sqlite: {}",
+                    error
+                )));
+            }
+            let select_result = sqlx::query_scalar::<sqlx::Sqlite, i64>(
+                r#"
+SELECT id
+FROM kb_observed_tokens
+WHERE mint = ?
+LIMIT 1
+                "#,
+            )
+            .bind(dto.mint.clone())
+            .fetch_one(pool)
+            .await;
+            match select_result {
+                Ok(id) => Ok(id),
+                Err(error) => Err(crate::KbError::Db(format!(
+                    "cannot fetch kb_observed_tokens id for mint '{}' on sqlite: {}",
+                    dto.mint, error
+                ))),
+            }
+        }
+    }
+}
+
+/// Reads one observed token by mint.
+pub async fn get_observed_token_by_mint(
+    database: &crate::KbDatabase,
+    mint: &str,
+) -> Result<std::option::Option<crate::KbObservedTokenDto>, crate::KbError> {
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let query_result = sqlx::query_as::<sqlx::Sqlite, crate::KbObservedTokenEntity>(
+                r#"
+SELECT
+    id,
+    mint,
+    symbol,
+    name,
+    decimals,
+    token_program,
+    status,
+    first_seen_at,
+    last_seen_at,
+    updated_at
+FROM kb_observed_tokens
+WHERE mint = ?
+LIMIT 1
+                    "#,
+            )
+            .bind(mint)
+            .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 observed token '{}' on sqlite: {}",
+                        mint, error
+                    )));
+                }
+            };
+            match entity_option {
+                Some(entity) => {
+                    let dto_result = crate::KbObservedTokenDto::try_from(entity);
+                    match dto_result {
+                        Ok(dto) => Ok(Some(dto)),
+                        Err(error) => Err(error),
+                    }
+                }
+                None => Ok(None),
+            }
+        }
+    }
+}
+
+/// Lists observed tokens ordered by newest first.
+pub async fn list_observed_tokens(
+    database: &crate::KbDatabase,
+    limit: u32,
+) -> Result<std::vec::Vec<crate::KbObservedTokenDto>, crate::KbError> {
+    if limit == 0 {
+        return Ok(std::vec::Vec::new());
+    }
+    match database.connection() {
+        crate::KbDatabaseConnection::Sqlite(pool) => {
+            let query_result = sqlx::query_as::<sqlx::Sqlite, crate::KbObservedTokenEntity>(
+                r#"
+SELECT
+    id,
+    mint,
+    symbol,
+    name,
+    decimals,
+    token_program,
+    status,
+    first_seen_at,
+    last_seen_at,
+    updated_at
+FROM kb_observed_tokens
+ORDER BY id DESC
+LIMIT ?
+                    "#,
+            )
+            .bind(i64::from(limit))
+            .fetch_all(pool)
+            .await;
+            let entities = match query_result {
+                Ok(entities) => entities,
+                Err(error) => {
+                    return Err(crate::KbError::Db(format!(
+                        "cannot list observed tokens on sqlite: {}",
+                        error
+                    )));
+                }
+            };
+            let mut dtos = std::vec::Vec::new();
+            for entity in entities {
+                let dto_result = crate::KbObservedTokenDto::try_from(entity);
+                let dto = match dto_result {
+                    Ok(dto) => dto,
+                    Err(error) => return Err(error),
+                };
+                dtos.push(dto);
+            }
+            Ok(dtos)
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[tokio::test]
+    async fn observed_token_roundtrip_works() {
+        let tempdir = tempfile::tempdir().expect("tempdir must succeed");
+        let database_path = tempdir.path().join("observed_token.sqlite3");
+        let config = crate::KbDatabaseConfig {
+            enabled: true,
+            backend: crate::KbDatabaseBackend::Sqlite,
+            sqlite: crate::KbSqliteDatabaseConfig {
+                path: database_path.to_string_lossy().to_string(),
+                create_if_missing: true,
+                busy_timeout_ms: 5000,
+                max_connections: 1,
+                auto_initialize_schema: true,
+                use_wal: true,
+            },
+        };
+        let database = crate::KbDatabase::connect_and_initialize(&config)
+            .await
+            .expect("database init must succeed");
+        let dto = crate::KbObservedTokenDto::new(
+            "So11111111111111111111111111111111111111112".to_string(),
+            Some("WSOL".to_string()),
+            Some("Wrapped SOL".to_string()),
+            Some(9),
+            crate::SPL_TOKEN_PROGRAM_ID.to_string(),
+            crate::KbObservedTokenStatus::Active,
+        );
+        let inserted_id = crate::upsert_observed_token(&database, &dto)
+            .await
+            .expect("upsert must succeed");
+        assert!(inserted_id > 0);
+        let fetched = crate::get_observed_token_by_mint(
+            &database,
+            "So11111111111111111111111111111111111111112",
+        )
+        .await
+        .expect("fetch must succeed");
+        assert!(fetched.is_some());
+        let fetched = fetched.expect("token must exist");
+        assert_eq!(fetched.symbol.as_deref(), Some("WSOL"));
+        assert_eq!(fetched.decimals, Some(9));
+        assert_eq!(fetched.status, crate::KbObservedTokenStatus::Active);
+        let listed = crate::list_observed_tokens(&database, 10)
+            .await
+            .expect("list must succeed");
+        assert_eq!(listed.len(), 1);
+    }
+}

kb_lib/src/db/schema.rs

@@ -3,7 +3,9 @@
 //! Database schema initialization.
 
 /// Ensures that the database schema exists.
-pub(crate) async fn ensure_schema(database: &crate::KbDatabase) -> Result<(), crate::KbError> {
+pub(crate) async fn ensure_schema(
+    database: &crate::KbDatabase,
+) -> Result<(), crate::KbError> {
     match database.connection() {
         crate::KbDatabaseConnection::Sqlite(pool) => {
             let metadata_table_result = sqlx::query(
@@ -99,6 +101,58 @@ ON kb_db_runtime_events (created_at)
                     error
                 )));
             }
+            let observed_tokens_result = sqlx::query(
+                r#"
+CREATE TABLE IF NOT EXISTS kb_observed_tokens (
+    id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+    mint TEXT NOT NULL UNIQUE,
+    symbol TEXT NULL,
+    name TEXT NULL,
+    decimals INTEGER NULL,
+    token_program TEXT NOT NULL,
+    status INTEGER NOT NULL,
+    first_seen_at TEXT NOT NULL,
+    last_seen_at TEXT NOT NULL,
+    updated_at TEXT NOT NULL
+)
+                "#,
+            )
+            .execute(pool)
+            .await;
+            if let Err(error) = observed_tokens_result {
+                return Err(crate::KbError::Db(format!(
+                    "cannot create table kb_observed_tokens on sqlite: {}",
+                    error
+                )));
+            }
+            let observed_tokens_mint_index_result = sqlx::query(
+                r#"
+CREATE UNIQUE INDEX IF NOT EXISTS kb_idx_observed_tokens_mint
+ON kb_observed_tokens (mint)
+                "#,
+            )
+            .execute(pool)
+            .await;
+            if let Err(error) = observed_tokens_mint_index_result {
+                return Err(crate::KbError::Db(format!(
+                    "cannot create index kb_idx_observed_tokens_mint on sqlite: {}",
+                    error
+                )));
+            }
+            let observed_tokens_status_index_result = sqlx::query(
+                r#"
+CREATE INDEX IF NOT EXISTS kb_idx_observed_tokens_status
+ON kb_observed_tokens (status)
+                "#,
+            )
+            .execute(pool)
+            .await;
+            if let Err(error) = observed_tokens_status_index_result {
+                return Err(crate::KbError::Db(format!(
+                    "cannot create index kb_idx_observed_tokens_status on sqlite: {}",
+                    error
+                )));
+            }
             let schema_version = crate::KbDbMetadataDto::new(
                 "schema_version".to_string(),
                 env!("CARGO_PKG_VERSION").to_string(),
@@ -108,6 +162,6 @@ ON kb_db_runtime_events (created_at)
                 return Err(error);
             }
             Ok(())
-        }
+        },
     }
 }

kb_lib/src/db/types.rs

@@ -3,7 +3,9 @@
 //! Database shared types.
 
 mod database_backend;
+mod observed_token_status;
 mod runtime_event_level;
 
 pub use crate::db::types::database_backend::KbDatabaseBackend;
+pub use crate::db::types::observed_token_status::KbObservedTokenStatus;
 pub use crate::db::types::runtime_event_level::KbDbRuntimeEventLevel;

kb_lib/src/db/types/observed_token_status.rs

@@ -0,0 +1,42 @@
+// file: kb_lib/src/db/types/observed_token_status.rs
+
+//! Observed token status.
+
+/// Local status of one observed token.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub enum KbObservedTokenStatus {
+    /// Newly discovered token.
+    New,
+    /// Token currently tracked.
+    Active,
+    /// Token ignored by local filtering rules.
+    Ignored,
+    /// Token marked as suspicious or blocked.
+    Blocked,
+}
+
+impl KbObservedTokenStatus {
+    /// Converts the status to its stable integer representation.
+    pub fn to_i16(self) -> i16 {
+        match self {
+            Self::New => 0,
+            Self::Active => 1,
+            Self::Ignored => 2,
+            Self::Blocked => 3,
+        }
+    }
+
+    /// Restores a status from its stable integer representation.
+    pub fn from_i16(value: i16) -> Result<Self, crate::KbError> {
+        match value {
+            0 => Ok(Self::New),
+            1 => Ok(Self::Active),
+            2 => Ok(Self::Ignored),
+            3 => Ok(Self::Blocked),
+            _ => Err(crate::KbError::Db(format!(
+                "invalid KbObservedTokenStatus value: {}",
+                value
+            ))),
+        }
+    }
+}

kb_lib/src/lib.rs

@@ -78,6 +78,12 @@ pub use crate::db::KbKnownHttpEndpointDto;
 pub use crate::db::KbKnownHttpEndpointEntity;
 pub use crate::db::KbKnownWsEndpointDto;
 pub use crate::db::KbKnownWsEndpointEntity;
+pub use crate::db::KbObservedTokenDto;
+pub use crate::db::KbObservedTokenEntity;
+pub use crate::db::KbObservedTokenStatus;
+pub use crate::db::get_observed_token_by_mint;
+pub use crate::db::list_observed_tokens;
+pub use crate::db::upsert_observed_token;
 pub use crate::db::get_known_http_endpoint;
 pub use crate::db::get_known_ws_endpoint;
 pub use crate::db::insert_db_runtime_event;