853 lines
43 KiB
Rust
853 lines
43 KiB
Rust
// file: kb_lib/src/db/queries/dex_event_coverage_entry.rs
|
|
|
|
//! Queries for `k_sol_dex_event_coverage_entries`.
|
|
|
|
/// Inserts or updates one DEX event coverage entry.
|
|
pub async fn query_dex_event_coverage_entries_upsert(
|
|
database: &crate::Database,
|
|
dto: &crate::DexEventCoverageEntryDto,
|
|
) -> Result<i64, crate::Error> {
|
|
match database.connection() {
|
|
crate::DatabaseConnection::Sqlite(pool) => {
|
|
let query_result = sqlx::query(
|
|
r#"
|
|
INSERT INTO k_sol_dex_event_coverage_entries (
|
|
coverage_key,
|
|
decoder_code,
|
|
program_id,
|
|
program_family,
|
|
surface_kind,
|
|
source_repo,
|
|
source_path,
|
|
entry_kind,
|
|
entry_name,
|
|
discriminator_hex,
|
|
discriminator_len,
|
|
event_family,
|
|
expected_db_target,
|
|
proof_status,
|
|
local_event_kind,
|
|
observed_count,
|
|
materialized_count,
|
|
trade_count,
|
|
first_signature,
|
|
last_signature,
|
|
notes,
|
|
created_at,
|
|
updated_at
|
|
)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(coverage_key) DO UPDATE SET
|
|
decoder_code = excluded.decoder_code,
|
|
program_id = excluded.program_id,
|
|
program_family = excluded.program_family,
|
|
surface_kind = excluded.surface_kind,
|
|
source_repo = excluded.source_repo,
|
|
source_path = excluded.source_path,
|
|
entry_kind = excluded.entry_kind,
|
|
entry_name = excluded.entry_name,
|
|
discriminator_hex = excluded.discriminator_hex,
|
|
discriminator_len = excluded.discriminator_len,
|
|
event_family = excluded.event_family,
|
|
expected_db_target = excluded.expected_db_target,
|
|
proof_status = excluded.proof_status,
|
|
local_event_kind = excluded.local_event_kind,
|
|
observed_count = excluded.observed_count,
|
|
materialized_count = excluded.materialized_count,
|
|
trade_count = excluded.trade_count,
|
|
first_signature = excluded.first_signature,
|
|
last_signature = excluded.last_signature,
|
|
notes = excluded.notes,
|
|
updated_at = excluded.updated_at
|
|
"#,
|
|
)
|
|
.bind(dto.coverage_key.clone())
|
|
.bind(dto.decoder_code.clone())
|
|
.bind(dto.program_id.clone())
|
|
.bind(dto.program_family.clone())
|
|
.bind(dto.surface_kind.clone())
|
|
.bind(dto.source_repo.clone())
|
|
.bind(dto.source_path.clone())
|
|
.bind(dto.entry_kind.clone())
|
|
.bind(dto.entry_name.clone())
|
|
.bind(dto.discriminator_hex.clone())
|
|
.bind(dto.discriminator_len)
|
|
.bind(dto.event_family.clone())
|
|
.bind(dto.expected_db_target.clone())
|
|
.bind(dto.proof_status.clone())
|
|
.bind(dto.local_event_kind.clone())
|
|
.bind(dto.observed_count)
|
|
.bind(dto.materialized_count)
|
|
.bind(dto.trade_count)
|
|
.bind(dto.first_signature.clone())
|
|
.bind(dto.last_signature.clone())
|
|
.bind(dto.notes.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::Error::Db(format!(
|
|
"cannot upsert k_sol_dex_event_coverage_entries on sqlite: {}",
|
|
error
|
|
)));
|
|
}
|
|
let id_result = sqlx::query_scalar::<sqlx::Sqlite, i64>(
|
|
r#"
|
|
SELECT id
|
|
FROM k_sol_dex_event_coverage_entries
|
|
WHERE coverage_key = ?
|
|
LIMIT 1
|
|
"#,
|
|
)
|
|
.bind(dto.coverage_key.clone())
|
|
.fetch_one(pool)
|
|
.await;
|
|
match id_result {
|
|
Ok(id) => return Ok(id),
|
|
Err(error) => {
|
|
return Err(crate::Error::Db(format!(
|
|
"cannot fetch k_sol_dex_event_coverage_entries id for coverage_key '{}' on sqlite: {}",
|
|
dto.coverage_key, error
|
|
)));
|
|
},
|
|
}
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Deletes all event coverage rows for one decoder.
|
|
pub async fn query_dex_event_coverage_entries_delete_by_decoder(
|
|
database: &crate::Database,
|
|
decoder_code: &str,
|
|
) -> Result<u64, crate::Error> {
|
|
match database.connection() {
|
|
crate::DatabaseConnection::Sqlite(pool) => {
|
|
let query_result = sqlx::query(
|
|
r#"
|
|
DELETE FROM k_sol_dex_event_coverage_entries
|
|
WHERE decoder_code = ?
|
|
"#,
|
|
)
|
|
.bind(decoder_code.to_string())
|
|
.execute(pool)
|
|
.await;
|
|
match query_result {
|
|
Ok(query_result) => return Ok(query_result.rows_affected()),
|
|
Err(error) => {
|
|
return Err(crate::Error::Db(format!(
|
|
"cannot delete k_sol_dex_event_coverage_entries for decoder_code '{}' on sqlite: {}",
|
|
decoder_code, error
|
|
)));
|
|
},
|
|
}
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Lists event coverage rows for one decoder.
|
|
pub async fn query_dex_event_coverage_entries_list_by_decoder(
|
|
database: &crate::Database,
|
|
decoder_code: &str,
|
|
) -> Result<std::vec::Vec<crate::DexEventCoverageEntryDto>, crate::Error> {
|
|
match database.connection() {
|
|
crate::DatabaseConnection::Sqlite(pool) => {
|
|
let query_result = sqlx::query_as::<sqlx::Sqlite, crate::DexEventCoverageEntryEntity>(
|
|
r#"
|
|
SELECT
|
|
id,
|
|
coverage_key,
|
|
decoder_code,
|
|
program_id,
|
|
program_family,
|
|
surface_kind,
|
|
source_repo,
|
|
source_path,
|
|
entry_kind,
|
|
entry_name,
|
|
discriminator_hex,
|
|
discriminator_len,
|
|
event_family,
|
|
expected_db_target,
|
|
proof_status,
|
|
local_event_kind,
|
|
observed_count,
|
|
materialized_count,
|
|
trade_count,
|
|
first_signature,
|
|
last_signature,
|
|
notes,
|
|
created_at,
|
|
updated_at
|
|
FROM k_sol_dex_event_coverage_entries
|
|
WHERE decoder_code = ?
|
|
ORDER BY entry_kind ASC, entry_name ASC, discriminator_hex ASC, id ASC
|
|
"#,
|
|
)
|
|
.bind(decoder_code.to_string())
|
|
.fetch_all(pool)
|
|
.await;
|
|
let entities = match query_result {
|
|
Ok(entities) => entities,
|
|
Err(error) => {
|
|
return Err(crate::Error::Db(format!(
|
|
"cannot list k_sol_dex_event_coverage_entries for decoder_code '{}' on sqlite: {}",
|
|
decoder_code, error
|
|
)));
|
|
},
|
|
};
|
|
return dex_event_coverage_entry_entities_to_dtos(entities);
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Lists event coverage summaries grouped by decoder.
|
|
pub async fn query_dex_event_coverage_entries_list_summary_by_decoder(
|
|
database: &crate::Database,
|
|
) -> Result<std::vec::Vec<crate::DexEventCoverageSummaryDto>, crate::Error> {
|
|
match database.connection() {
|
|
crate::DatabaseConnection::Sqlite(pool) => {
|
|
let query_result =
|
|
sqlx::query_as::<sqlx::Sqlite, crate::DexEventCoverageSummaryEntity>(
|
|
r#"
|
|
SELECT
|
|
decoder_code,
|
|
COUNT(*) AS listed_entry_count,
|
|
SUM(CASE WHEN local_event_kind IS NOT NULL AND local_event_kind <> '' THEN 1 ELSE 0 END) AS decoded_entry_count,
|
|
SUM(CASE WHEN observed_count > 0 THEN 1 ELSE 0 END) AS observed_entry_count,
|
|
SUM(CASE WHEN materialized_count > 0 THEN 1 ELSE 0 END) AS materialized_entry_count,
|
|
COALESCE(SUM(observed_count), 0) AS total_observed_count,
|
|
COALESCE(SUM(materialized_count), 0) AS total_materialized_count,
|
|
COALESCE(SUM(trade_count), 0) AS trade_count,
|
|
SUM(CASE WHEN expected_db_target = 'k_sol_dex_decoded_events_only' THEN 1 ELSE 0 END) AS audit_only_entry_count,
|
|
SUM(CASE WHEN expected_db_target IS NULL OR expected_db_target = '' THEN 1 ELSE 0 END) AS missing_db_target_entry_count,
|
|
SUM(CASE WHEN proof_status = 'upstream_git_unverified' THEN 1 ELSE 0 END) AS upstream_git_unverified_entry_count,
|
|
SUM(CASE WHEN proof_status = 'upstream_git_mapped_unverified' THEN 1 ELSE 0 END) AS upstream_git_mapped_unverified_entry_count,
|
|
SUM(CASE WHEN proof_status = 'upstream_git_local_corpus_observed' THEN 1 ELSE 0 END) AS upstream_git_local_corpus_observed_entry_count,
|
|
SUM(CASE WHEN proof_status = 'upstream_git_local_corpus_materialized' THEN 1 ELSE 0 END) AS upstream_git_local_corpus_materialized_entry_count
|
|
FROM k_sol_dex_event_coverage_entries
|
|
GROUP BY decoder_code
|
|
ORDER BY decoder_code ASC
|
|
"#,
|
|
)
|
|
.fetch_all(pool)
|
|
.await;
|
|
let entities = match query_result {
|
|
Ok(entities) => entities,
|
|
Err(error) => {
|
|
return Err(crate::Error::Db(format!(
|
|
"cannot list k_sol_dex_event_coverage_entries summaries on sqlite: {}",
|
|
error
|
|
)));
|
|
},
|
|
};
|
|
let mut dtos = std::vec::Vec::new();
|
|
for entity in entities {
|
|
let dto_result = crate::DexEventCoverageSummaryDto::try_from(entity);
|
|
let dto = match dto_result {
|
|
Ok(dto) => dto,
|
|
Err(error) => return Err(error),
|
|
};
|
|
dtos.push(dto);
|
|
}
|
|
return Ok(dtos);
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Refreshes local observed/materialized counts for every coverage row.
|
|
pub async fn query_dex_event_coverage_entries_refresh_local_counts(
|
|
database: &crate::Database,
|
|
) -> Result<u64, crate::Error> {
|
|
return query_dex_event_coverage_entries_refresh_local_counts_internal(database, None).await;
|
|
}
|
|
|
|
/// Refreshes local observed/materialized counts for one decoder coverage subset.
|
|
pub async fn query_dex_event_coverage_entries_refresh_local_counts_by_decoder(
|
|
database: &crate::Database,
|
|
decoder_code: &str,
|
|
) -> Result<u64, crate::Error> {
|
|
return query_dex_event_coverage_entries_refresh_local_counts_internal(
|
|
database,
|
|
Some(decoder_code),
|
|
)
|
|
.await;
|
|
}
|
|
|
|
async fn query_dex_event_coverage_entries_refresh_local_counts_internal(
|
|
database: &crate::Database,
|
|
decoder_code: std::option::Option<&str>,
|
|
) -> Result<u64, crate::Error> {
|
|
match database.connection() {
|
|
crate::DatabaseConnection::Sqlite(pool) => {
|
|
let now = chrono::Utc::now().to_rfc3339();
|
|
let decoder_code_filter: std::option::Option<String> = match decoder_code {
|
|
Some(value) => Some(value.to_string()),
|
|
None => None,
|
|
};
|
|
let update_result = sqlx::query(
|
|
r#"
|
|
UPDATE k_sol_dex_event_coverage_entries
|
|
SET
|
|
observed_count = (
|
|
SELECT COUNT(*)
|
|
FROM k_sol_dex_decoded_events de
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
),
|
|
trade_count = CASE
|
|
WHEN expected_db_target = 'k_sol_trade_events' THEN (
|
|
SELECT COUNT(te.id)
|
|
FROM k_sol_dex_decoded_events de
|
|
JOIN k_sol_trade_events te ON te.decoded_event_id = de.id
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
ELSE 0
|
|
END,
|
|
|
|
materialized_count = CASE
|
|
WHEN expected_db_target = 'k_sol_trade_events' THEN (
|
|
SELECT COUNT(te.id)
|
|
FROM k_sol_dex_decoded_events de
|
|
JOIN k_sol_trade_events te ON te.decoded_event_id = de.id
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
WHEN expected_db_target = 'k_sol_liquidity_events' THEN (
|
|
SELECT COUNT(le.id)
|
|
FROM k_sol_dex_decoded_events de
|
|
JOIN k_sol_liquidity_events le ON le.decoded_event_id = de.id
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
WHEN expected_db_target = 'k_sol_pool_lifecycle_events' THEN (
|
|
SELECT COUNT(pe.id)
|
|
FROM k_sol_dex_decoded_events de
|
|
JOIN k_sol_pool_lifecycle_events pe ON pe.decoded_event_id = de.id
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
WHEN expected_db_target = 'k_sol_fee_events' THEN (
|
|
SELECT COUNT(fe.id)
|
|
FROM k_sol_dex_decoded_events de
|
|
JOIN k_sol_fee_events fe ON fe.decoded_event_id = de.id
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
WHEN expected_db_target = 'k_sol_reward_events' THEN (
|
|
SELECT COUNT(re.id)
|
|
FROM k_sol_dex_decoded_events de
|
|
JOIN k_sol_reward_events re ON re.decoded_event_id = de.id
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
WHEN expected_db_target = 'k_sol_pool_admin_events' THEN (
|
|
SELECT COUNT(ae.id)
|
|
FROM k_sol_dex_decoded_events de
|
|
JOIN k_sol_pool_admin_events ae ON ae.decoded_event_id = de.id
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
WHEN expected_db_target = 'k_sol_orderbook_events' THEN (
|
|
SELECT COUNT(oe.id)
|
|
FROM k_sol_dex_decoded_events de
|
|
JOIN k_sol_orderbook_events oe ON oe.decoded_event_id = de.id
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
WHEN expected_db_target = 'k_sol_launch_events' THEN (
|
|
SELECT COUNT(le.id)
|
|
FROM k_sol_dex_decoded_events de
|
|
JOIN k_sol_launch_events le ON le.decoded_event_id = de.id
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
ELSE materialized_count
|
|
END,
|
|
first_signature = (
|
|
SELECT tx.signature
|
|
FROM k_sol_dex_decoded_events de
|
|
JOIN k_sol_chain_transactions tx ON tx.id = de.transaction_id
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
ORDER BY COALESCE(tx.slot, 0) ASC, tx.id ASC
|
|
LIMIT 1
|
|
),
|
|
last_signature = (
|
|
SELECT tx.signature
|
|
FROM k_sol_dex_decoded_events de
|
|
JOIN k_sol_chain_transactions tx ON tx.id = de.transaction_id
|
|
WHERE (
|
|
(k_sol_dex_event_coverage_entries.program_id IS NULL OR de.program_id = k_sol_dex_event_coverage_entries.program_id)
|
|
AND (
|
|
(
|
|
k_sol_dex_event_coverage_entries.local_event_kind IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.local_event_kind <> ''
|
|
AND de.event_kind = k_sol_dex_event_coverage_entries.local_event_kind
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.entry_name IS NOT NULL
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamEntryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamInstructionName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.upstreamEventName') = k_sol_dex_event_coverage_entries.entry_name
|
|
OR json_extract(de.payload_json, '$.entryName') = k_sol_dex_event_coverage_entries.entry_name
|
|
)
|
|
)
|
|
OR (
|
|
k_sol_dex_event_coverage_entries.discriminator_hex IS NOT NULL
|
|
AND k_sol_dex_event_coverage_entries.discriminator_hex <> ''
|
|
AND (
|
|
json_extract(de.payload_json, '$.upstreamDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.instructionDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.anchorEventDiscriminator') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
OR json_extract(de.payload_json, '$.discriminatorHex') = k_sol_dex_event_coverage_entries.discriminator_hex
|
|
)
|
|
)
|
|
)
|
|
)
|
|
ORDER BY COALESCE(tx.slot, 0) DESC, tx.id DESC
|
|
LIMIT 1
|
|
),
|
|
updated_at = ?
|
|
WHERE (? IS NULL OR decoder_code = ?)
|
|
"#,
|
|
)
|
|
.bind(now.clone())
|
|
.bind(decoder_code_filter.clone())
|
|
.bind(decoder_code_filter.clone())
|
|
.execute(pool)
|
|
.await;
|
|
let refreshed_entry_count = match update_result {
|
|
Ok(result) => result.rows_affected(),
|
|
Err(error) => {
|
|
return Err(crate::Error::Db(format!(
|
|
"cannot refresh k_sol_dex_event_coverage_entries local counts on sqlite: {}",
|
|
error
|
|
)));
|
|
},
|
|
};
|
|
let proof_result = sqlx::query(
|
|
r#"
|
|
UPDATE k_sol_dex_event_coverage_entries
|
|
SET
|
|
proof_status = CASE
|
|
WHEN materialized_count > 0 OR trade_count > 0 THEN 'upstream_git_local_corpus_materialized'
|
|
WHEN observed_count > 0 THEN 'upstream_git_local_corpus_observed'
|
|
WHEN local_event_kind IS NOT NULL AND local_event_kind <> '' THEN 'upstream_git_mapped_unverified'
|
|
ELSE proof_status
|
|
END,
|
|
updated_at = ?
|
|
WHERE (? IS NULL OR decoder_code = ?)
|
|
"#,
|
|
)
|
|
.bind(now)
|
|
.bind(decoder_code_filter.clone())
|
|
.bind(decoder_code_filter)
|
|
.execute(pool)
|
|
.await;
|
|
if let Err(error) = proof_result {
|
|
return Err(crate::Error::Db(format!(
|
|
"cannot refresh k_sol_dex_event_coverage_entries proof statuses on sqlite: {}",
|
|
error
|
|
)));
|
|
}
|
|
return Ok(refreshed_entry_count);
|
|
},
|
|
}
|
|
}
|
|
|
|
fn dex_event_coverage_entry_entities_to_dtos(
|
|
entities: std::vec::Vec<crate::DexEventCoverageEntryEntity>,
|
|
) -> Result<std::vec::Vec<crate::DexEventCoverageEntryDto>, crate::Error> {
|
|
let mut dtos = std::vec::Vec::new();
|
|
for entity in entities {
|
|
let dto_result = crate::DexEventCoverageEntryDto::try_from(entity);
|
|
let dto = match dto_result {
|
|
Ok(dto) => dto,
|
|
Err(error) => return Err(error),
|
|
};
|
|
dtos.push(dto);
|
|
}
|
|
return Ok(dtos);
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
async fn make_database() -> crate::Database {
|
|
let tempdir = tempfile::tempdir().expect("tempdir must succeed");
|
|
let database_path = tempdir.path().join("dex_event_coverage.sqlite3");
|
|
let config = crate::DatabaseConfig {
|
|
enabled: true,
|
|
backend: crate::DatabaseBackend::Sqlite,
|
|
sqlite: crate::SqliteDatabaseConfig {
|
|
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,
|
|
},
|
|
};
|
|
return crate::Database::connect_and_initialize(&config)
|
|
.await
|
|
.expect("database init must succeed");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dex_event_coverage_entry_roundtrip_and_summary_work() {
|
|
let database = make_database().await;
|
|
let upstream_service = crate::UpstreamRegistryService::new();
|
|
let request = crate::UpstreamRegistrySearchRequestDto {
|
|
decoder_code: Some("raydium_cpmm".to_string()),
|
|
program_id: None,
|
|
program_family: None,
|
|
surface_kind: None,
|
|
entry_kind: Some(crate::ENTRY_KIND_INSTRUCTION.to_string()),
|
|
proof_status: None,
|
|
limit: Some(1),
|
|
};
|
|
let upstream_result = upstream_service.search(&request);
|
|
assert_eq!(upstream_result.entries.len(), 1);
|
|
let upstream_entry = upstream_result.entries.first().expect("entry must exist");
|
|
let mut dto = crate::DexEventCoverageEntryDto::from_upstream_registry_entry(
|
|
upstream_entry,
|
|
Some("swap".to_string()),
|
|
Some(crate::DexEventCoverageEntryDto::DB_TARGET_DECODED_EVENTS_ONLY.to_string()),
|
|
Some("raydium_cpmm.swap".to_string()),
|
|
);
|
|
dto.proof_status = crate::PROOF_STATUS_UPSTREAM_GIT_LOCAL_CORPUS_OBSERVED.to_string();
|
|
dto.observed_count = 3;
|
|
let id = crate::query_dex_event_coverage_entries_upsert(&database, &dto)
|
|
.await
|
|
.expect("coverage upsert must succeed");
|
|
assert!(id > 0);
|
|
let rows =
|
|
crate::query_dex_event_coverage_entries_list_by_decoder(&database, "raydium_cpmm")
|
|
.await
|
|
.expect("coverage list must succeed");
|
|
assert_eq!(rows.len(), 1);
|
|
assert_eq!(rows[0].observed_count, 3);
|
|
let summaries = crate::query_dex_event_coverage_entries_list_summary_by_decoder(&database)
|
|
.await
|
|
.expect("coverage summary must succeed");
|
|
assert_eq!(summaries.len(), 1);
|
|
assert_eq!(summaries[0].decoder_code, "raydium_cpmm");
|
|
assert_eq!(summaries[0].listed_entry_count, 1);
|
|
assert_eq!(summaries[0].decoded_entry_count, 1);
|
|
assert_eq!(summaries[0].observed_entry_count, 1);
|
|
assert_eq!(summaries[0].total_observed_count, 3);
|
|
assert_eq!(summaries[0].audit_only_entry_count, 1);
|
|
}
|
|
}
|