Sasedev/khadhroony-bobobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.7.35

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-05-14 07:52:36 +02:00
parent cfa1ff2289
commit 348ae7f153
29 changed files with 1956 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

302
kb_lib/src/db/queries/fee_event.rs Normal file
View File

@@ -0,0 +1,302 @@
// file: kb_lib/src/db/queries/fee_event.rs
//! Queries for `k_sol_fee_events`.
/// Returns one fee event by decoded event id.
pub async fn query_fee_events_get_by_decoded_event_id(
database: &crate::Database,
decoded_event_id: i64,
) -> Result<std::option::Option<crate::FeeEventDto>, crate::Error> {
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let query_result = sqlx::query_as::<sqlx::Sqlite, crate::FeeEventEntity>(
r#"
SELECT
id,
transaction_id,
decoded_event_id,
dex_id,
pool_id,
pair_id,
signature,
slot,
protocol_name,
program_id,
event_kind,
pool_account,
actor_wallet,
fee_token_mint,
fee_amount_raw,
payload_json,
executed_at,
created_at
FROM k_sol_fee_events
WHERE decoded_event_id = ?
LIMIT 1
"#,
)
.bind(decoded_event_id)
.fetch_optional(pool)
.await;
let entity_option = match query_result {
Ok(entity_option) => entity_option,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot fetch k_sol_fee_events by decoded_event_id '{}' on sqlite: {}",
decoded_event_id, error
)));
},
};
match entity_option {
Some(entity) => {
let dto_result = crate::FeeEventDto::try_from(entity);
match dto_result {
Ok(dto) => return Ok(Some(dto)),
Err(error) => return Err(error),
}
},
None => return Ok(None),
}
},
}
}
/// Inserts or updates one normalized fee event row.
pub async fn query_fee_events_upsert(
database: &crate::Database,
dto: &crate::FeeEventDto,
) -> Result<i64, crate::Error> {
let slot_i64 = match dto.slot {
Some(slot) => {
let slot_result = i64::try_from(slot);
match slot_result {
Ok(slot) => Some(slot),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert fee event slot '{}' to i64: {}",
slot, error
)));
},
}
},
None => None,
};
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let existing_id = match dto.decoded_event_id {
Some(decoded_event_id) => {
let existing_result = sqlx::query_scalar::<sqlx::Sqlite, i64>(
r#"
SELECT id
FROM k_sol_fee_events
WHERE decoded_event_id = ?
LIMIT 1
"#,
)
.bind(decoded_event_id)
.fetch_optional(pool)
.await;
match existing_result {
Ok(existing_id) => existing_id,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot fetch k_sol_fee_events id for decoded_event_id '{}' on sqlite: {}",
decoded_event_id, error
)));
},
}
},
None => None,
};
if let Some(id) = existing_id {
let update_result = sqlx::query(
r#"
UPDATE k_sol_fee_events
SET
transaction_id = ?,
dex_id = ?,
pool_id = ?,
pair_id = ?,
signature = ?,
slot = ?,
protocol_name = ?,
program_id = ?,
event_kind = ?,
pool_account = ?,
actor_wallet = ?,
fee_token_mint = ?,
fee_amount_raw = ?,
payload_json = ?,
executed_at = ?
WHERE id = ?
"#,
)
.bind(dto.transaction_id)
.bind(dto.dex_id)
.bind(dto.pool_id)
.bind(dto.pair_id)
.bind(dto.signature.clone())
.bind(slot_i64)
.bind(dto.protocol_name.clone())
.bind(dto.program_id.clone())
.bind(dto.event_kind.clone())
.bind(dto.pool_account.clone())
.bind(dto.actor_wallet.clone())
.bind(dto.fee_token_mint.clone())
.bind(dto.fee_amount_raw.clone())
.bind(dto.payload_json.clone())
.bind(dto.executed_at.to_rfc3339())
.bind(id)
.execute(pool)
.await;
if let Err(error) = update_result {
return Err(crate::Error::Db(format!(
"cannot update k_sol_fee_events id '{}' on sqlite: {}",
id, error
)));
}
return Ok(id);
}
let insert_result = sqlx::query(
r#"
INSERT INTO k_sol_fee_events (
transaction_id,
decoded_event_id,
dex_id,
pool_id,
pair_id,
signature,
slot,
protocol_name,
program_id,
event_kind,
pool_account,
actor_wallet,
fee_token_mint,
fee_amount_raw,
payload_json,
executed_at,
created_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"#,
)
.bind(dto.transaction_id)
.bind(dto.decoded_event_id)
.bind(dto.dex_id)
.bind(dto.pool_id)
.bind(dto.pair_id)
.bind(dto.signature.clone())
.bind(slot_i64)
.bind(dto.protocol_name.clone())
.bind(dto.program_id.clone())
.bind(dto.event_kind.clone())
.bind(dto.pool_account.clone())
.bind(dto.actor_wallet.clone())
.bind(dto.fee_token_mint.clone())
.bind(dto.fee_amount_raw.clone())
.bind(dto.payload_json.clone())
.bind(dto.executed_at.to_rfc3339())
.bind(dto.created_at.to_rfc3339())
.execute(pool)
.await;
if let Err(error) = insert_result {
return Err(crate::Error::Db(format!(
"cannot insert k_sol_fee_events on sqlite: {}",
error
)));
}
let id_result = sqlx::query_scalar::<sqlx::Sqlite, i64>(
r#"
SELECT id
FROM k_sol_fee_events
WHERE transaction_id = ?
AND protocol_name = ?
AND event_kind = ?
AND signature = ?
ORDER BY id DESC
LIMIT 1
"#,
)
.bind(dto.transaction_id)
.bind(dto.protocol_name.clone())
.bind(dto.event_kind.clone())
.bind(dto.signature.clone())
.fetch_one(pool)
.await;
match id_result {
Ok(id) => return Ok(id),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot fetch inserted k_sol_fee_events id for signature '{}' on sqlite: {}",
dto.signature, error
)));
},
}
},
}
}
/// Lists recent fee events ordered from newest to oldest.
pub async fn query_fee_events_list_recent(
database: &crate::Database,
limit: u32,
) -> Result<std::vec::Vec<crate::FeeEventDto>, crate::Error> {
if limit == 0 {
return Ok(std::vec::Vec::new());
}
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let query_result = sqlx::query_as::<sqlx::Sqlite, crate::FeeEventEntity>(
r#"
SELECT
id,
transaction_id,
decoded_event_id,
dex_id,
pool_id,
pair_id,
signature,
slot,
protocol_name,
program_id,
event_kind,
pool_account,
actor_wallet,
fee_token_mint,
fee_amount_raw,
payload_json,
executed_at,
created_at
FROM k_sol_fee_events
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::Error::Db(format!(
"cannot list k_sol_fee_events on sqlite: {}",
error
)));
},
};
let mut dtos = std::vec::Vec::with_capacity(entities.len());
for entity in entities {
let dto_result = crate::FeeEventDto::try_from(entity);
match dto_result {
Ok(dto) => dtos.push(dto),
Err(error) => return Err(error),
}
}
return Ok(dtos);
},
}
}

8
kb_lib/src/db/queries/local_pipeline_diagnostics.rs
View File

@@ -50,6 +50,9 @@ SELECT
) AS decoded_unknown_event_count,
(SELECT COUNT(*) FROM k_sol_liquidity_events) AS liquidity_event_count,
(SELECT COUNT(*) FROM k_sol_pool_lifecycle_events) AS pool_lifecycle_event_count,
(SELECT COUNT(*) FROM k_sol_fee_events) AS fee_event_count,
(SELECT COUNT(*) FROM k_sol_reward_events) AS reward_event_count,
(SELECT COUNT(*) FROM k_sol_pool_admin_events) AS pool_admin_event_count,
(
SELECT COUNT(*)
FROM k_sol_dex_decoded_events dde
@@ -361,6 +364,9 @@ SELECT
decoded_unknown_event_count: row.decoded_unknown_event_count,
liquidity_event_count: row.liquidity_event_count,
pool_lifecycle_event_count: row.pool_lifecycle_event_count,
fee_event_count: row.fee_event_count,
reward_event_count: row.reward_event_count,
pool_admin_event_count: row.pool_admin_event_count,
missing_trade_event_count: row.missing_trade_event_count,
decoded_trade_candidate_without_trade_event_count: row
.decoded_trade_candidate_without_trade_event_count,
@@ -772,7 +778,7 @@ ORDER BY
return Ok(summaries);
},
}
}
}
/// Lists local pair trading-readiness summaries.
pub async fn query_local_pair_trading_readiness_diagnostic_list_summaries(

293
kb_lib/src/db/queries/pool_admin_event.rs Normal file
View File

@@ -0,0 +1,293 @@
// file: kb_lib/src/db/queries/pool_admin_event.rs
//! Queries for `k_sol_pool_admin_events`.
/// Returns one pool admin event by decoded event id.
pub async fn query_pool_admin_events_get_by_decoded_event_id(
database: &crate::Database,
decoded_event_id: i64,
) -> Result<std::option::Option<crate::PoolAdminEventDto>, crate::Error> {
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let query_result = sqlx::query_as::<sqlx::Sqlite, crate::PoolAdminEventEntity>(
r#"
SELECT
id,
transaction_id,
decoded_event_id,
dex_id,
pool_id,
pair_id,
signature,
slot,
protocol_name,
program_id,
event_kind,
pool_account,
actor_wallet,
admin_action,
payload_json,
executed_at,
created_at
FROM k_sol_pool_admin_events
WHERE decoded_event_id = ?
LIMIT 1
"#,
)
.bind(decoded_event_id)
.fetch_optional(pool)
.await;
let entity_option = match query_result {
Ok(entity_option) => entity_option,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot fetch k_sol_pool_admin_events by decoded_event_id '{}' on sqlite: {}",
decoded_event_id, error
)));
},
};
match entity_option {
Some(entity) => {
let dto_result = crate::PoolAdminEventDto::try_from(entity);
match dto_result {
Ok(dto) => return Ok(Some(dto)),
Err(error) => return Err(error),
}
},
None => return Ok(None),
}
},
}
}
/// Inserts or updates one normalized pool admin event row.
pub async fn query_pool_admin_events_upsert(
database: &crate::Database,
dto: &crate::PoolAdminEventDto,
) -> Result<i64, crate::Error> {
let slot_i64 = match dto.slot {
Some(slot) => {
let slot_result = i64::try_from(slot);
match slot_result {
Ok(slot) => Some(slot),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert pool admin event slot '{}' to i64: {}",
slot, error
)));
},
}
},
None => None,
};
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let existing_id = match dto.decoded_event_id {
Some(decoded_event_id) => {
let existing_result = sqlx::query_scalar::<sqlx::Sqlite, i64>(
r#"
SELECT id
FROM k_sol_pool_admin_events
WHERE decoded_event_id = ?
LIMIT 1
"#,
)
.bind(decoded_event_id)
.fetch_optional(pool)
.await;
match existing_result {
Ok(existing_id) => existing_id,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot fetch k_sol_pool_admin_events id for decoded_event_id '{}' on sqlite: {}",
decoded_event_id, error
)));
},
}
},
None => None,
};
if let Some(id) = existing_id {
let update_result = sqlx::query(
r#"
UPDATE k_sol_pool_admin_events
SET
transaction_id = ?,
dex_id = ?,
pool_id = ?,
pair_id = ?,
signature = ?,
slot = ?,
protocol_name = ?,
program_id = ?,
event_kind = ?,
pool_account = ?,
actor_wallet = ?,
admin_action = ?,
payload_json = ?,
executed_at = ?
WHERE id = ?
"#,
)
.bind(dto.transaction_id)
.bind(dto.dex_id)
.bind(dto.pool_id)
.bind(dto.pair_id)
.bind(dto.signature.clone())
.bind(slot_i64)
.bind(dto.protocol_name.clone())
.bind(dto.program_id.clone())
.bind(dto.event_kind.clone())
.bind(dto.pool_account.clone())
.bind(dto.actor_wallet.clone())
.bind(dto.admin_action.clone())
.bind(dto.payload_json.clone())
.bind(dto.executed_at.to_rfc3339())
.bind(id)
.execute(pool)
.await;
if let Err(error) = update_result {
return Err(crate::Error::Db(format!(
"cannot update k_sol_pool_admin_events id '{}' on sqlite: {}",
id, error
)));
}
return Ok(id);
}
let insert_result = sqlx::query(
r#"
INSERT INTO k_sol_pool_admin_events (
transaction_id,
decoded_event_id,
dex_id,
pool_id,
pair_id,
signature,
slot,
protocol_name,
program_id,
event_kind,
pool_account,
actor_wallet,
admin_action,
payload_json,
executed_at,
created_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"#,
)
.bind(dto.transaction_id)
.bind(dto.decoded_event_id)
.bind(dto.dex_id)
.bind(dto.pool_id)
.bind(dto.pair_id)
.bind(dto.signature.clone())
.bind(slot_i64)
.bind(dto.protocol_name.clone())
.bind(dto.program_id.clone())
.bind(dto.event_kind.clone())
.bind(dto.pool_account.clone())
.bind(dto.actor_wallet.clone())
.bind(dto.admin_action.clone())
.bind(dto.payload_json.clone())
.bind(dto.executed_at.to_rfc3339())
.bind(dto.created_at.to_rfc3339())
.execute(pool)
.await;
if let Err(error) = insert_result {
return Err(crate::Error::Db(format!(
"cannot insert k_sol_pool_admin_events on sqlite: {}",
error
)));
}
let id_result = sqlx::query_scalar::<sqlx::Sqlite, i64>(
r#"
SELECT id
FROM k_sol_pool_admin_events
WHERE transaction_id = ?
AND protocol_name = ?
AND event_kind = ?
AND signature = ?
ORDER BY id DESC
LIMIT 1
"#,
)
.bind(dto.transaction_id)
.bind(dto.protocol_name.clone())
.bind(dto.event_kind.clone())
.bind(dto.signature.clone())
.fetch_one(pool)
.await;
match id_result {
Ok(id) => return Ok(id),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot fetch inserted k_sol_pool_admin_events id for signature '{}' on sqlite: {}",
dto.signature, error
)));
},
}
},
}
}
/// Lists recent pool admin events ordered from newest to oldest.
pub async fn query_pool_admin_events_list_recent(
database: &crate::Database,
limit: u32,
) -> Result<std::vec::Vec<crate::PoolAdminEventDto>, crate::Error> {
if limit == 0 {
return Ok(std::vec::Vec::new());
}
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let query_result = sqlx::query_as::<sqlx::Sqlite, crate::PoolAdminEventEntity>(
r#"
SELECT
id,
transaction_id,
decoded_event_id,
dex_id,
pool_id,
pair_id,
signature,
slot,
protocol_name,
program_id,
event_kind,
pool_account,
actor_wallet,
admin_action,
payload_json,
executed_at,
created_at
FROM k_sol_pool_admin_events
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::Error::Db(format!(
"cannot list k_sol_pool_admin_events on sqlite: {}",
error
)));
},
};
let mut dtos = std::vec::Vec::with_capacity(entities.len());
for entity in entities {
let dto_result = crate::PoolAdminEventDto::try_from(entity);
match dto_result {
Ok(dto) => dtos.push(dto),
Err(error) => return Err(error),
}
}
return Ok(dtos);
},
}
}

299
kb_lib/src/db/queries/reward_event.rs Normal file
View File

@@ -0,0 +1,299 @@
// file: kb_lib/src/db/queries/reward_event.rs
//! Queries for `k_sol_reward_events`.
/// Returns one reward event by decoded event id.
pub async fn query_reward_events_get_by_decoded_event_id(
database: &crate::Database,
decoded_event_id: i64,
) -> Result<std::option::Option<crate::RewardEventDto>, crate::Error> {
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let query_result = sqlx::query_as::<sqlx::Sqlite, crate::RewardEventEntity>(
r#"
SELECT
id,
transaction_id,
decoded_event_id,
dex_id,
pool_id,
pair_id,
signature,
slot,
protocol_name,
program_id,
event_kind,
pool_account,
actor_wallet,
reward_token_mint,
reward_amount_raw,
payload_json,
executed_at,
created_at
FROM k_sol_reward_events
WHERE decoded_event_id = ?
LIMIT 1
"#,
)
.bind(decoded_event_id)
.fetch_optional(pool)
.await;
let entity_option = match query_result {
Ok(entity_option) => entity_option,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot fetch k_sol_reward_events by decoded_event_id '{}' on sqlite: {}",
decoded_event_id, error
)));
},
};
match entity_option {
Some(entity) => {
let dto_result = crate::RewardEventDto::try_from(entity);
match dto_result {
Ok(dto) => return Ok(Some(dto)),
Err(error) => return Err(error),
}
},
None => return Ok(None),
}
},
}
}
/// Inserts or updates one normalized reward event row.
pub async fn query_reward_events_upsert(
database: &crate::Database,
dto: &crate::RewardEventDto,
) -> Result<i64, crate::Error> {
let slot_i64 = match dto.slot {
Some(slot) => {
let slot_result = i64::try_from(slot);
match slot_result {
Ok(slot) => Some(slot),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot convert reward event slot '{}' to i64: {}",
slot, error
)));
},
}
},
None => None,
};
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let existing_id = match dto.decoded_event_id {
Some(decoded_event_id) => {
let existing_result = sqlx::query_scalar::<sqlx::Sqlite, i64>(
r#"
SELECT id
FROM k_sol_reward_events
WHERE decoded_event_id = ?
LIMIT 1
"#,
)
.bind(decoded_event_id)
.fetch_optional(pool)
.await;
match existing_result {
Ok(existing_id) => existing_id,
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot fetch k_sol_reward_events id for decoded_event_id '{}' on sqlite: {}",
decoded_event_id, error
)));
},
}
},
None => None,
};
if let Some(id) = existing_id {
let update_result = sqlx::query(
r#"
UPDATE k_sol_reward_events
SET
transaction_id = ?,
dex_id = ?,
pool_id = ?,
pair_id = ?,
signature = ?,
slot = ?,
protocol_name = ?,
program_id = ?,
event_kind = ?,
pool_account = ?,
actor_wallet = ?,
reward_token_mint = ?,
reward_amount_raw = ?,
payload_json = ?,
executed_at = ?
WHERE id = ?
"#,
)
.bind(dto.transaction_id)
.bind(dto.dex_id)
.bind(dto.pool_id)
.bind(dto.pair_id)
.bind(dto.signature.clone())
.bind(slot_i64)
.bind(dto.protocol_name.clone())
.bind(dto.program_id.clone())
.bind(dto.event_kind.clone())
.bind(dto.pool_account.clone())
.bind(dto.actor_wallet.clone())
.bind(dto.reward_token_mint.clone())
.bind(dto.reward_amount_raw.clone())
.bind(dto.payload_json.clone())
.bind(dto.executed_at.to_rfc3339())
.bind(id)
.execute(pool)
.await;
if let Err(error) = update_result {
return Err(crate::Error::Db(format!(
"cannot update k_sol_reward_events id '{}' on sqlite: {}",
id, error
)));
}
return Ok(id);
}
let insert_result = sqlx::query(
r#"
INSERT INTO k_sol_reward_events (
transaction_id,
decoded_event_id,
dex_id,
pool_id,
pair_id,
signature,
slot,
protocol_name,
program_id,
event_kind,
pool_account,
actor_wallet,
reward_token_mint,
reward_amount_raw,
payload_json,
executed_at,
created_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"#,
)
.bind(dto.transaction_id)
.bind(dto.decoded_event_id)
.bind(dto.dex_id)
.bind(dto.pool_id)
.bind(dto.pair_id)
.bind(dto.signature.clone())
.bind(slot_i64)
.bind(dto.protocol_name.clone())
.bind(dto.program_id.clone())
.bind(dto.event_kind.clone())
.bind(dto.pool_account.clone())
.bind(dto.actor_wallet.clone())
.bind(dto.reward_token_mint.clone())
.bind(dto.reward_amount_raw.clone())
.bind(dto.payload_json.clone())
.bind(dto.executed_at.to_rfc3339())
.bind(dto.created_at.to_rfc3339())
.execute(pool)
.await;
if let Err(error) = insert_result {
return Err(crate::Error::Db(format!(
"cannot insert k_sol_reward_events on sqlite: {}",
error
)));
}
let id_result = sqlx::query_scalar::<sqlx::Sqlite, i64>(
r#"
SELECT id
FROM k_sol_reward_events
WHERE transaction_id = ?
AND protocol_name = ?
AND event_kind = ?
AND signature = ?
ORDER BY id DESC
LIMIT 1
"#,
)
.bind(dto.transaction_id)
.bind(dto.protocol_name.clone())
.bind(dto.event_kind.clone())
.bind(dto.signature.clone())
.fetch_one(pool)
.await;
match id_result {
Ok(id) => return Ok(id),
Err(error) => {
return Err(crate::Error::Db(format!(
"cannot fetch inserted k_sol_reward_events id for signature '{}' on sqlite: {}",
dto.signature, error
)));
},
}
},
}
}
/// Lists recent reward events ordered from newest to oldest.
pub async fn query_reward_events_list_recent(
database: &crate::Database,
limit: u32,
) -> Result<std::vec::Vec<crate::RewardEventDto>, crate::Error> {
if limit == 0 {
return Ok(std::vec::Vec::new());
}
match database.connection() {
crate::DatabaseConnection::Sqlite(pool) => {
let query_result = sqlx::query_as::<sqlx::Sqlite, crate::RewardEventEntity>(
r#"
SELECT
id,
transaction_id,
decoded_event_id,
dex_id,
pool_id,
pair_id,
signature,
slot,
protocol_name,
program_id,
event_kind,
pool_account,
actor_wallet,
reward_token_mint,
reward_amount_raw,
payload_json,
executed_at,
created_at
FROM k_sol_reward_events
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::Error::Db(format!(
"cannot list k_sol_reward_events on sqlite: {}",
error
)));
},
};
let mut dtos = std::vec::Vec::with_capacity(entities.len());
for entity in entities {
let dto_result = crate::RewardEventDto::try_from(entity);
match dto_result {
Ok(dto) => dtos.push(dto),
Err(error) => return Err(error),
}
}
return Ok(dtos);
},
}
}