This commit is contained in:
2026-04-29 10:37:47 +02:00
parent 3313f8245e
commit 7fbeb8826b
8 changed files with 1362 additions and 13 deletions

623
kb_lib/src/dex/fluxbeam.rs Normal file
View File

@@ -0,0 +1,623 @@
// file: kb_lib/src/dex/fluxbeam.rs
//! FluxBeam transaction decoder.
/// FluxBeam program id.
pub const KB_FLUXBEAM_PROGRAM_ID: &str = "FLUXubRmkEi2q6K3Y9kBPg9248ggaZVsoSFhtJHSrm1X";
/// Decoded FluxBeam create-pool event.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct KbFluxbeamCreatePoolDecoded {
/// Parent transaction id.
pub transaction_id: i64,
/// Parent instruction id.
pub instruction_id: i64,
/// Transaction signature.
pub signature: std::string::String,
/// Program id.
pub program_id: std::string::String,
/// Optional pool account.
pub pool_account: std::option::Option<std::string::String>,
/// Optional LP mint.
pub lp_mint: std::option::Option<std::string::String>,
/// Optional token A mint.
pub token_a_mint: std::option::Option<std::string::String>,
/// Optional token B mint.
pub token_b_mint: std::option::Option<std::string::String>,
/// Optional creator / payer.
pub creator: std::option::Option<std::string::String>,
/// Decoded payload.
pub payload_json: serde_json::Value,
}
/// Decoded FluxBeam swap event.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct KbFluxbeamSwapDecoded {
/// Parent transaction id.
pub transaction_id: i64,
/// Parent instruction id.
pub instruction_id: i64,
/// Transaction signature.
pub signature: std::string::String,
/// Program id.
pub program_id: std::string::String,
/// Trade side relative to normalized base.
pub trade_side: crate::KbSwapTradeSide,
/// Optional pool account.
pub pool_account: std::option::Option<std::string::String>,
/// Optional token A mint.
pub token_a_mint: std::option::Option<std::string::String>,
/// Optional token B mint.
pub token_b_mint: std::option::Option<std::string::String>,
/// Decoded payload.
pub payload_json: serde_json::Value,
}
/// Decoded FluxBeam event.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum KbFluxbeamDecodedEvent {
/// Pool creation.
CreatePool(KbFluxbeamCreatePoolDecoded),
/// Swap.
Swap(KbFluxbeamSwapDecoded),
}
/// FluxBeam decoder.
#[derive(Debug, Clone, Default)]
pub struct KbFluxbeamDecoder;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum KbFluxbeamInstructionKind {
CreatePool,
Swap,
Unknown,
}
impl KbFluxbeamDecoder {
/// Creates a new decoder.
pub fn new() -> Self {
Self
}
/// Decodes one projected transaction into zero or more FluxBeam events.
pub fn decode_transaction(
&self,
transaction: &crate::KbChainTransactionDto,
instructions: &[crate::KbChainInstructionDto],
) -> Result<std::vec::Vec<crate::KbFluxbeamDecodedEvent>, crate::KbError> {
let transaction_id_option = transaction.id;
let transaction_id = match transaction_id_option {
Some(transaction_id) => transaction_id,
None => {
return Err(crate::KbError::InvalidState(format!(
"chain transaction '{}' has no internal id",
transaction.signature
)));
}
};
let transaction_json_result =
serde_json::from_str::<serde_json::Value>(transaction.transaction_json.as_str());
let transaction_json = match transaction_json_result {
Ok(transaction_json) => transaction_json,
Err(error) => {
return Err(crate::KbError::Json(format!(
"cannot parse transaction_json for signature '{}': {}",
transaction.signature, error
)));
}
};
let log_messages = kb_extract_log_messages(&transaction_json);
let mut decoded_events = std::vec::Vec::new();
for instruction in instructions {
if instruction.parent_instruction_id.is_some() {
continue;
}
let program_id_option = &instruction.program_id;
let program_id = match program_id_option {
Some(program_id) => program_id,
None => continue,
};
if program_id.as_str() != crate::KB_FLUXBEAM_PROGRAM_ID {
continue;
}
let instruction_id_option = instruction.id;
let instruction_id = match instruction_id_option {
Some(instruction_id) => instruction_id,
None => continue,
};
let accounts_result = kb_parse_accounts_json(instruction.accounts_json.as_str());
let accounts = match accounts_result {
Ok(accounts) => accounts,
Err(error) => return Err(error),
};
let parsed_json_result =
kb_parse_optional_parsed_json(instruction.parsed_json.as_ref());
let parsed_json = match parsed_json_result {
Ok(parsed_json) => parsed_json,
Err(error) => return Err(error),
};
let instruction_kind =
kb_classify_instruction_kind(parsed_json.as_ref(), &log_messages);
let pool_account = kb_extract_string_by_candidate_keys(
parsed_json.as_ref(),
&[
"pool",
"poolAddress",
"poolAccount",
"amm",
"ammPool",
"poolState",
],
)
.or_else(|| kb_extract_account(&accounts, 0));
let lp_mint = kb_extract_string_by_candidate_keys(
parsed_json.as_ref(),
&["lpMint", "lp_mint", "lpTokenMint"],
)
.or_else(|| kb_extract_account(&accounts, 1));
let token_a_mint = kb_extract_string_by_candidate_keys(
parsed_json.as_ref(),
&[
"tokenA",
"tokenAMint",
"mintA",
"baseMint",
"token0Mint",
"mint0",
],
)
.or_else(|| kb_extract_account(&accounts, 2));
let token_b_mint = kb_extract_string_by_candidate_keys(
parsed_json.as_ref(),
&[
"tokenB",
"tokenBMint",
"mintB",
"quoteMint",
"token1Mint",
"mint1",
],
)
.or_else(|| kb_extract_account(&accounts, 3));
let creator = kb_extract_string_by_candidate_keys(
parsed_json.as_ref(),
&["payer", "creator", "user", "owner"],
)
.or_else(|| kb_extract_account(&accounts, 4));
if instruction_kind == KbFluxbeamInstructionKind::CreatePool {
let payload_json = serde_json::json!({
"decoder": "fluxbeam",
"eventKind": "create_pool",
"signature": transaction.signature,
"instructionId": instruction_id,
"instructionIndex": instruction.instruction_index,
"accounts": accounts,
"parsed": parsed_json,
"logMessages": log_messages,
"poolAccount": pool_account,
"lpMint": lp_mint,
"tokenAMint": token_a_mint,
"tokenBMint": token_b_mint,
"creator": creator
});
decoded_events.push(crate::KbFluxbeamDecodedEvent::CreatePool(
crate::KbFluxbeamCreatePoolDecoded {
transaction_id,
instruction_id,
signature: transaction.signature.clone(),
program_id: program_id.clone(),
pool_account,
lp_mint,
token_a_mint,
token_b_mint,
creator,
payload_json,
},
));
continue;
}
if instruction_kind == KbFluxbeamInstructionKind::Swap {
let trade_side = kb_infer_trade_side(&log_messages);
let payload_json = serde_json::json!({
"decoder": "fluxbeam",
"eventKind": "swap",
"signature": transaction.signature,
"instructionId": instruction_id,
"instructionIndex": instruction.instruction_index,
"accounts": accounts,
"parsed": parsed_json,
"logMessages": log_messages,
"poolAccount": pool_account,
"tokenAMint": token_a_mint,
"tokenBMint": token_b_mint,
"tradeSide": format!("{:?}", trade_side)
});
decoded_events.push(crate::KbFluxbeamDecodedEvent::Swap(
crate::KbFluxbeamSwapDecoded {
transaction_id,
instruction_id,
signature: transaction.signature.clone(),
program_id: program_id.clone(),
trade_side,
pool_account,
token_a_mint,
token_b_mint,
payload_json,
},
));
}
}
Ok(decoded_events)
}
}
fn kb_classify_instruction_kind(
parsed_json: std::option::Option<&serde_json::Value>,
log_messages: &[std::string::String],
) -> KbFluxbeamInstructionKind {
let parsed_instruction_name = kb_extract_string_by_candidate_keys(
parsed_json,
&["instruction", "instructionName", "type", "name"],
);
if let Some(parsed_instruction_name) = parsed_instruction_name {
let normalized = kb_normalize_text(parsed_instruction_name.as_str());
if normalized.contains("createpool") || normalized.contains("initializepool") {
return KbFluxbeamInstructionKind::CreatePool;
}
if normalized == "swap" {
return KbFluxbeamInstructionKind::Swap;
}
}
if kb_log_messages_contain_keyword(log_messages, "create_pool")
|| kb_log_messages_contain_keyword(log_messages, "createpool")
|| kb_log_messages_contain_keyword(log_messages, "initialize_pool")
|| kb_log_messages_contain_keyword(log_messages, "initializepool")
{
return KbFluxbeamInstructionKind::CreatePool;
}
if kb_log_messages_contain_keyword(log_messages, "swap") {
return KbFluxbeamInstructionKind::Swap;
}
KbFluxbeamInstructionKind::Unknown
}
fn kb_extract_log_messages(
transaction_json: &serde_json::Value,
) -> std::vec::Vec<std::string::String> {
let mut messages = std::vec::Vec::new();
let meta_option = transaction_json.get("meta");
let meta = match meta_option {
Some(meta) => meta,
None => return messages,
};
let logs_option = meta.get("logMessages");
let logs = match logs_option {
Some(logs) => logs,
None => return messages,
};
let logs_array = match logs.as_array() {
Some(logs_array) => logs_array,
None => return messages,
};
for value in logs_array {
let text_option = value.as_str();
if let Some(text) = text_option {
messages.push(text.to_string());
}
}
messages
}
fn kb_log_messages_contain_keyword(log_messages: &[std::string::String], keyword: &str) -> bool {
let keyword_normalized = kb_normalize_text(keyword);
for log_message in log_messages {
let log_normalized = kb_normalize_text(log_message.as_str());
if log_normalized.contains(keyword_normalized.as_str()) {
return true;
}
}
false
}
fn kb_normalize_text(value: &str) -> std::string::String {
let mut normalized = std::string::String::new();
for character in value.chars() {
if character.is_ascii_alphanumeric() {
normalized.push(character.to_ascii_lowercase());
}
}
normalized
}
fn kb_parse_accounts_json(
accounts_json: &str,
) -> Result<std::vec::Vec<std::string::String>, crate::KbError> {
let values_result = serde_json::from_str::<std::vec::Vec<serde_json::Value>>(accounts_json);
let values = match values_result {
Ok(values) => values,
Err(error) => {
return Err(crate::KbError::Json(format!(
"cannot parse instruction accounts_json '{}': {}",
accounts_json, error
)));
}
};
let mut accounts = std::vec::Vec::new();
for value in values {
let text_option = value.as_str();
if let Some(text) = text_option {
accounts.push(text.to_string());
}
}
Ok(accounts)
}
fn kb_parse_optional_parsed_json(
parsed_json: std::option::Option<&std::string::String>,
) -> Result<std::option::Option<serde_json::Value>, crate::KbError> {
let parsed_json = match parsed_json {
Some(parsed_json) => parsed_json,
None => return Ok(None),
};
let value_result = serde_json::from_str::<serde_json::Value>(parsed_json.as_str());
match value_result {
Ok(value) => Ok(Some(value)),
Err(error) => Err(crate::KbError::Json(format!(
"cannot parse instruction parsed_json '{}': {}",
parsed_json, error
))),
}
}
fn kb_extract_string_by_candidate_keys(
value: std::option::Option<&serde_json::Value>,
candidate_keys: &[&str],
) -> std::option::Option<std::string::String> {
let value = match value {
Some(value) => value,
None => return None,
};
kb_extract_string_by_candidate_keys_inner(value, candidate_keys)
}
fn kb_extract_string_by_candidate_keys_inner(
value: &serde_json::Value,
candidate_keys: &[&str],
) -> std::option::Option<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 direct_text_option = direct.as_str();
if let Some(direct_text) = direct_text_option {
return Some(direct_text.to_string());
}
}
}
for nested_value in object.values() {
let nested_result =
kb_extract_string_by_candidate_keys_inner(nested_value, candidate_keys);
if nested_result.is_some() {
return nested_result;
}
}
return None;
}
if let Some(array) = value.as_array() {
for nested_value in array {
let nested_result =
kb_extract_string_by_candidate_keys_inner(nested_value, candidate_keys);
if nested_result.is_some() {
return nested_result;
}
}
}
None
}
fn kb_extract_account(
accounts: &[std::string::String],
index: usize,
) -> std::option::Option<std::string::String> {
if index >= accounts.len() {
return None;
}
Some(accounts[index].clone())
}
fn kb_infer_trade_side(log_messages: &[std::string::String]) -> crate::KbSwapTradeSide {
if kb_log_messages_contain_keyword(log_messages, "buy") {
return crate::KbSwapTradeSide::BuyBase;
}
if kb_log_messages_contain_keyword(log_messages, "sell") {
return crate::KbSwapTradeSide::SellBase;
}
crate::KbSwapTradeSide::Unknown
}
#[cfg(test)]
mod tests {
fn make_create_transaction() -> crate::KbChainTransactionDto {
let mut dto = crate::KbChainTransactionDto::new(
"sig-fluxbeam-create-1".to_string(),
Some(892001),
Some(1779700001),
Some("helius_primary_http".to_string()),
Some("0".to_string()),
None,
None,
serde_json::json!({
"slot": 892001,
"meta": {
"logMessages": [
"Program log: Instruction: CreatePool"
]
},
"transaction": {
"message": {
"instructions": []
}
}
})
.to_string(),
);
dto.id = Some(701);
dto
}
fn make_create_instruction() -> crate::KbChainInstructionDto {
let mut dto = crate::KbChainInstructionDto::new(
701,
None,
0,
None,
Some(crate::KB_FLUXBEAM_PROGRAM_ID.to_string()),
Some("fluxbeam".to_string()),
Some(1),
serde_json::json!([
"FluxPool111",
"FluxLpMint111",
"FluxTokenA111",
"So11111111111111111111111111111111111111112",
"FluxCreator111"
])
.to_string(),
None,
None,
Some(
serde_json::json!({
"info": {
"instruction": "create_pool",
"pool": "FluxPool111",
"lpMint": "FluxLpMint111",
"tokenA": "FluxTokenA111",
"tokenB": "So11111111111111111111111111111111111111112",
"payer": "FluxCreator111"
}
})
.to_string(),
),
);
dto.id = Some(702);
dto
}
fn make_swap_transaction() -> crate::KbChainTransactionDto {
let mut dto = crate::KbChainTransactionDto::new(
"sig-fluxbeam-swap-1".to_string(),
Some(892002),
Some(1779700002),
Some("helius_primary_http".to_string()),
Some("0".to_string()),
None,
None,
serde_json::json!({
"slot": 892002,
"meta": {
"logMessages": [
"Program log: Instruction: Swap"
]
},
"transaction": {
"message": {
"instructions": []
}
}
})
.to_string(),
);
dto.id = Some(703);
dto
}
fn make_swap_instruction() -> crate::KbChainInstructionDto {
let mut dto = crate::KbChainInstructionDto::new(
703,
None,
0,
None,
Some(crate::KB_FLUXBEAM_PROGRAM_ID.to_string()),
Some("fluxbeam".to_string()),
Some(1),
serde_json::json!([
"FluxSwapPool111",
"FluxSwapLpMint111",
"FluxSwapTokenA111",
"So11111111111111111111111111111111111111112"
])
.to_string(),
None,
None,
Some(
serde_json::json!({
"info": {
"instruction": "swap",
"pool": "FluxSwapPool111",
"tokenA": "FluxSwapTokenA111",
"tokenB": "So11111111111111111111111111111111111111112"
}
})
.to_string(),
),
);
dto.id = Some(704);
dto
}
#[test]
fn fluxbeam_create_pool_is_detected() {
let decoder = crate::KbFluxbeamDecoder::new();
let transaction = make_create_transaction();
let instructions = vec![make_create_instruction()];
let decoded_result = decoder.decode_transaction(&transaction, &instructions);
let decoded = match decoded_result {
Ok(decoded) => decoded,
Err(error) => panic!("decode must succeed: {}", error),
};
assert_eq!(decoded.len(), 1);
match &decoded[0] {
crate::KbFluxbeamDecodedEvent::CreatePool(event) => {
assert_eq!(event.transaction_id, 701);
assert_eq!(event.instruction_id, 702);
assert_eq!(event.pool_account, Some("FluxPool111".to_string()));
assert_eq!(event.lp_mint, Some("FluxLpMint111".to_string()));
assert_eq!(event.token_a_mint, Some("FluxTokenA111".to_string()));
assert_eq!(
event.token_b_mint,
Some("So11111111111111111111111111111111111111112".to_string())
);
}
crate::KbFluxbeamDecodedEvent::Swap(_) => {
panic!("unexpected swap event")
}
}
}
#[test]
fn fluxbeam_swap_is_detected() {
let decoder = crate::KbFluxbeamDecoder::new();
let transaction = make_swap_transaction();
let instructions = vec![make_swap_instruction()];
let decoded_result = decoder.decode_transaction(&transaction, &instructions);
let decoded = match decoded_result {
Ok(decoded) => decoded,
Err(error) => panic!("decode must succeed: {}", error),
};
assert_eq!(decoded.len(), 1);
match &decoded[0] {
crate::KbFluxbeamDecodedEvent::Swap(event) => {
assert_eq!(event.transaction_id, 703);
assert_eq!(event.instruction_id, 704);
assert_eq!(event.pool_account, Some("FluxSwapPool111".to_string()));
assert_eq!(event.token_a_mint, Some("FluxSwapTokenA111".to_string()));
assert_eq!(
event.token_b_mint,
Some("So11111111111111111111111111111111111111112".to_string())
);
}
crate::KbFluxbeamDecodedEvent::CreatePool(_) => {
panic!("unexpected create event")
}
}
}
}