1use anyhow::{anyhow, Context, Result};
2use arrow::array::{Array, BinaryArray};
3use arrow::{array::RecordBatch, datatypes::*};
4use std::sync::Arc;
5mod deserialize;
6pub use deserialize::{deserialize_data, DynType, DynValue, ParamInput};
7mod arrow_converter;
8use arrow_converter::{to_arrow, to_arrow_dtype};
9
10#[derive(Debug, Clone)]
11pub struct InstructionSignature {
12 pub discriminator: Vec<u8>,
13 pub params: Vec<ParamInput>,
14 pub accounts_names: Vec<String>,
15}
16
17#[cfg(feature = "pyo3")]
18impl<'py> pyo3::FromPyObject<'py> for InstructionSignature {
19 fn extract_bound(ob: &pyo3::Bound<'py, pyo3::PyAny>) -> pyo3::PyResult<Self> {
20 use pyo3::types::PyAnyMethods;
21 use pyo3::types::PyTypeMethods;
22
23 let discriminator_ob = ob.getattr("discriminator")?;
24
25 let discriminator_ob_type: String = discriminator_ob.get_type().name()?.to_string();
26 let discriminator = match discriminator_ob_type.as_str() {
27 "str" => {
28 let s: &str = discriminator_ob.extract()?;
29 hex_to_bytes(s).context("failed to decode hex")?
30 }
31 "bytes" => discriminator_ob.extract()?,
32 _ => return Err(anyhow!("unknown type: {}", discriminator_ob_type).into()),
33 };
34
35 let params = ob.getattr("params")?.extract::<Vec<ParamInput>>()?;
36 let accounts_names = ob.getattr("accounts_names")?.extract::<Vec<String>>()?;
37
38 Ok(InstructionSignature {
39 discriminator,
40 params,
41 accounts_names,
42 })
43 }
44}
45
46fn hex_to_bytes(hex_string: &str) -> Result<Vec<u8>> {
47 let hex_string = hex_string.strip_prefix("0x").unwrap_or(hex_string);
48 let hex_string = if hex_string.len() % 2 == 1 {
49 format!("0{}", hex_string)
50 } else {
51 hex_string.to_string()
52 };
53 let out = (0..hex_string.len())
54 .step_by(2)
55 .map(|i| {
56 u8::from_str_radix(&hex_string[i..i + 2], 16)
57 .context("failed to parse hexstring to bytes")
58 })
59 .collect::<Result<Vec<_>, _>>()?;
60
61 Ok(out)
62}
63
64pub fn svm_decode_instructions(
65 signature: InstructionSignature,
66 batch: &RecordBatch,
67 allow_decode_fail: bool,
68) -> Result<RecordBatch> {
69 let data_col = batch.column_by_name("data").unwrap();
70 let data_array = data_col.as_any().downcast_ref::<BinaryArray>().unwrap();
71
72 let account_arrays: Vec<&BinaryArray> = (0..10)
73 .map(|i| {
74 let col_name = format!("a{}", i);
75 let col = batch.column_by_name(&col_name).unwrap();
76 col.as_any().downcast_ref::<BinaryArray>().unwrap()
77 })
78 .collect();
79
80 decode_instructions(signature, &account_arrays, data_array, allow_decode_fail)
81}
82
83pub fn decode_instructions(
84 signature: InstructionSignature,
85 accounts: &[&BinaryArray],
86 data: &BinaryArray,
87 allow_decode_fail: bool,
88) -> Result<RecordBatch> {
89 let num_params = signature.params.len();
90
91 let mut decoded_params_vec: Vec<Vec<Option<DynValue>>> =
92 (0..num_params).map(|_| Vec::new()).collect();
93
94 for row_idx in 0..data.len() {
95 if data.is_null(row_idx) {
96 if allow_decode_fail {
97 log::debug!("Instruction data is null");
98 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
99 continue;
100 } else {
101 return Err(anyhow::anyhow!("Instruction data is null"));
102 }
103 }
104
105 let instruction_data = data.value(row_idx).to_vec();
106 let data_result = match_discriminators(&instruction_data, &signature.discriminator);
107 let data = match data_result {
108 Ok(data) => data,
109 Err(e) if allow_decode_fail => {
110 log::debug!("Error matching discriminators: {:?}", e);
111 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
112 continue;
113 }
114 Err(e) => {
115 return Err(anyhow::anyhow!("Error matching discriminators: {:?}", e));
116 }
117 };
118
119 let decoded_ix_result = deserialize_data(&data, &signature.params);
120 let decoded_ix = match decoded_ix_result {
121 Ok(ix) => ix,
122 Err(e) if allow_decode_fail => {
123 log::debug!("Error deserializing instruction: {:?}", e);
124 decoded_params_vec.iter_mut().for_each(|v| v.push(None));
125 continue;
126 }
127 Err(e) => {
128 return Err(anyhow::anyhow!("Error deserializing instruction: {:?}", e));
129 }
130 };
131
132 for (i, value) in decoded_ix.into_iter().enumerate() {
133 decoded_params_vec[i].push(Some(value));
134 }
135 }
136
137 let data_arrays: Vec<Arc<dyn Array>> = decoded_params_vec
138 .iter()
139 .enumerate()
140 .map(|(i, v)| to_arrow(&signature.params[i].param_type, v.clone()).unwrap())
141 .collect::<Vec<_>>();
142
143 let data_fields = signature
144 .params
145 .iter()
146 .map(|p| Field::new(p.name.clone(), to_arrow_dtype(&p.param_type).unwrap(), true))
147 .collect::<Vec<_>>();
148
149 let acc_names_len = signature.accounts_names.len();
150
151 let mut accounts: Vec<Arc<dyn Array>> = accounts
152 .iter()
153 .map(|arr| {
154 let owned_array = arr.slice(0, arr.len());
155 owned_array as Arc<dyn Array>
156 })
157 .collect();
158
159 let mut acc_fields = Vec::new();
160 if acc_names_len < 10 {
161 let _ = accounts.split_off(acc_names_len);
162 for i in 0..acc_names_len {
163 let field = Field::new(signature.accounts_names[i].clone(), DataType::Binary, true);
164 acc_fields.push(field);
165 }
166 } else {
167 for i in 0..10 {
168 let field = Field::new(signature.accounts_names[i].clone(), DataType::Binary, true);
169 acc_fields.push(field);
170 }
171 }
172
173 let decoded_instructions_array = data_arrays.into_iter().chain(accounts).collect::<Vec<_>>();
174 let decoded_instructions_fields = data_fields
175 .into_iter()
176 .chain(acc_fields.clone())
177 .collect::<Vec<_>>();
178
179 let schema = Arc::new(Schema::new(decoded_instructions_fields));
180 let batch = RecordBatch::try_new(schema, decoded_instructions_array)
181 .context("Failed to create record batch from data arrays")
182 .unwrap();
183
184 Ok(batch)
185}
186
187pub fn match_discriminators(instr_data: &[u8], discriminator: &[u8]) -> Result<Vec<u8>> {
188 let discriminator_len = discriminator.len();
189 if instr_data.len() < discriminator_len {
190 return Err(anyhow::anyhow!(
191 "Instruction data is too short to contain discriminator. Expected at least {} bytes, got {} bytes",
192 discriminator_len,
193 instr_data.len()
194 ));
195 }
196 let disc = &instr_data[..discriminator_len].to_vec();
197 let ix_data = &instr_data[discriminator_len..];
198 if !disc.eq(discriminator) {
199 return Err(anyhow::anyhow!(
200 "Instruction data discriminator doesn't match signature discriminator"
201 ));
202 }
203 Ok(ix_data.to_vec())
204}
205
206pub fn instruction_signature_to_arrow_schema(signature: &InstructionSignature) -> Result<Schema> {
207 let mut fields = Vec::new();
208
209 for param in &signature.params {
210 let field = Field::new(
211 param.name.clone(),
212 to_arrow_dtype(¶m.param_type).unwrap(),
213 true,
214 );
215 fields.push(field);
216 }
217
218 for account in &signature.accounts_names {
219 let field = Field::new(account.clone(), DataType::Binary, true);
220 fields.push(field);
221 }
222
223 Ok(Schema::new(fields))
224}
225
226#[cfg(test)]
227mod tests {
228 use super::*;
229 use crate::deserialize::{DynType, ParamInput};
230 use std::fs::File;
231
232 #[test]
233 #[ignore]
234 fn read_parquet_with_real_data() {
235 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
236
237 let builder = ParquetRecordBatchReaderBuilder::try_new(
238 File::open("instruction_exemple.parquet").unwrap(),
239 )
240 .unwrap();
241 let mut reader = builder.build().unwrap();
242 let instructions = reader.next().unwrap().unwrap();
243 let ix_signature = InstructionSignature {
244 discriminator: vec![229, 23, 203, 151, 122, 227, 173, 42],
286 params: vec![
287 ParamInput {
288 name: "RoutePlan".to_string(),
289 param_type: DynType::Array(Box::new(DynType::Struct(vec![
290 (
291 "Swap".to_string(),
292 DynType::Enum(vec![
293 ("Saber".to_string(), None),
294 ("SaberAddDecimalsDeposit".to_string(), None),
295 ("SaberAddDecimalsWithdraw".to_string(), None),
296 ("TokenSwap".to_string(), None),
297 ("Sencha".to_string(), None),
298 ("Step".to_string(), None),
299 ("Cropper".to_string(), None),
300 ("Raydium".to_string(), None),
301 (
302 "Crema".to_string(),
303 Some(DynType::Struct(vec![(
304 "a_to_b".to_string(),
305 DynType::Bool,
306 )])),
307 ),
308 ("Lifinity".to_string(), None),
309 ("Mercurial".to_string(), None),
310 ("Cykura".to_string(), None),
311 (
312 "Serum".to_string(),
313 Some(DynType::Struct(vec![(
314 "side".to_string(),
315 DynType::Enum(vec![
316 ("Bid".to_string(), None),
317 ("Ask".to_string(), None),
318 ]),
319 )])),
320 ),
321 ("MarinadeDeposit".to_string(), None),
322 ("MarinadeUnstake".to_string(), None),
323 (
324 "Aldrin".to_string(),
325 Some(DynType::Struct(vec![(
326 "side".to_string(),
327 DynType::Enum(vec![
328 ("Bid".to_string(), None),
329 ("Ask".to_string(), None),
330 ]),
331 )])),
332 ),
333 (
334 "AldrinV2".to_string(),
335 Some(DynType::Struct(vec![(
336 "side".to_string(),
337 DynType::Enum(vec![
338 ("Bid".to_string(), None),
339 ("Ask".to_string(), None),
340 ]),
341 )])),
342 ),
343 (
344 "Whirlpool".to_string(),
345 Some(DynType::Struct(vec![(
346 "a_to_b".to_string(),
347 DynType::Bool,
348 )])),
349 ),
350 (
351 "Invariant".to_string(),
352 Some(DynType::Struct(vec![(
353 "x_to_y".to_string(),
354 DynType::Bool,
355 )])),
356 ),
357 ("Meteora".to_string(), None),
358 ("GooseFX".to_string(), None),
359 (
360 "DeltaFi".to_string(),
361 Some(DynType::Struct(vec![(
362 "stable".to_string(),
363 DynType::Bool,
364 )])),
365 ),
366 ("Balansol".to_string(), None),
367 (
368 "MarcoPolo".to_string(),
369 Some(DynType::Struct(vec![(
370 "x_to_y".to_string(),
371 DynType::Bool,
372 )])),
373 ),
374 (
375 "Dradex".to_string(),
376 Some(DynType::Struct(vec![(
377 "side".to_string(),
378 DynType::Enum(vec![
379 ("Bid".to_string(), None),
380 ("Ask".to_string(), None),
381 ]),
382 )])),
383 ),
384 ("LifinityV2".to_string(), None),
385 ("RaydiumClmm".to_string(), None),
386 (
387 "Openbook".to_string(),
388 Some(DynType::Struct(vec![(
389 "side".to_string(),
390 DynType::Enum(vec![
391 ("Bid".to_string(), None),
392 ("Ask".to_string(), None),
393 ]),
394 )])),
395 ),
396 (
397 "Phoenix".to_string(),
398 Some(DynType::Struct(vec![(
399 "side".to_string(),
400 DynType::Enum(vec![
401 ("Bid".to_string(), None),
402 ("Ask".to_string(), None),
403 ]),
404 )])),
405 ),
406 (
407 "Symmetry".to_string(),
408 Some(DynType::Struct(vec![
409 ("from_token_id".to_string(), DynType::U64),
410 ("to_token_id".to_string(), DynType::U64),
411 ])),
412 ),
413 ("TokenSwapV2".to_string(), None),
414 ("HeliumTreasuryManagementRedeemV0".to_string(), None),
415 ("StakeDexStakeWrappedSol".to_string(), None),
416 (
417 "StakeDexSwapViaStake".to_string(),
418 Some(DynType::Struct(vec![(
419 "bridge_stake_seed".to_string(),
420 DynType::U32,
421 )])),
422 ),
423 ("GooseFXV2".to_string(), None),
424 ("Perps".to_string(), None),
425 ("PerpsAddLiquidity".to_string(), None),
426 ("PerpsRemoveLiquidity".to_string(), None),
427 ("MeteoraDlmm".to_string(), None),
428 (
429 "OpenBookV2".to_string(),
430 Some(DynType::Struct(vec![(
431 "side".to_string(),
432 DynType::Enum(vec![
433 ("Bid".to_string(), None),
434 ("Ask".to_string(), None),
435 ]),
436 )])),
437 ),
438 ("RaydiumClmmV2".to_string(), None),
439 (
440 "StakeDexPrefundWithdrawStakeAndDepositStake".to_string(),
441 Some(DynType::Struct(vec![(
442 "bridge_stake_seed".to_string(),
443 DynType::U32,
444 )])),
445 ),
446 (
447 "Clone".to_string(),
448 Some(DynType::Struct(vec![
449 ("pool_index".to_string(), DynType::U8),
450 ("quantity_is_input".to_string(), DynType::Bool),
451 ("quantity_is_collateral".to_string(), DynType::Bool),
452 ])),
453 ),
454 (
455 "SanctumS".to_string(),
456 Some(DynType::Struct(vec![
457 ("src_lst_value_calc_accs".to_string(), DynType::U8),
458 ("dst_lst_value_calc_accs".to_string(), DynType::U8),
459 ("src_lst_index".to_string(), DynType::U32),
460 ("dst_lst_index".to_string(), DynType::U32),
461 ])),
462 ),
463 (
464 "SanctumSAddLiquidity".to_string(),
465 Some(DynType::Struct(vec![
466 ("lst_value_calc_accs".to_string(), DynType::U8),
467 ("lst_index".to_string(), DynType::U32),
468 ])),
469 ),
470 (
471 "SanctumSRemoveLiquidity".to_string(),
472 Some(DynType::Struct(vec![
473 ("lst_value_calc_accs".to_string(), DynType::U8),
474 ("lst_index".to_string(), DynType::U32),
475 ])),
476 ),
477 ("RaydiumCP".to_string(), None),
478 (
479 "WhirlpoolSwapV2".to_string(),
480 Some(DynType::Struct(vec![
481 ("a_to_b".to_string(), DynType::Bool),
482 (
483 "remaining_accounts_info".to_string(),
484 DynType::Struct(vec![(
485 "slices".to_string(),
486 DynType::Array(Box::new(DynType::Struct(vec![(
487 "remaining_accounts_slice".to_string(),
488 DynType::Struct(vec![
489 ("accounts_type".to_string(), DynType::U8),
490 ("length".to_string(), DynType::U8),
491 ]),
492 )]))),
493 )]),
494 ),
495 ])),
496 ),
497 ("OneIntro".to_string(), None),
498 ("PumpdotfunWrappedBuy".to_string(), None),
499 ("PumpdotfunWrappedSell".to_string(), None),
500 ("PerpsV2".to_string(), None),
501 ("PerpsV2AddLiquidity".to_string(), None),
502 ("PerpsV2RemoveLiquidity".to_string(), None),
503 ("MoonshotWrappedBuy".to_string(), None),
504 ("MoonshotWrappedSell".to_string(), None),
505 ("StabbleStableSwap".to_string(), None),
506 ("StabbleWeightedSwap".to_string(), None),
507 (
508 "Obric".to_string(),
509 Some(DynType::Struct(vec![(
510 "x_to_y".to_string(),
511 DynType::Bool,
512 )])),
513 ),
514 ("FoxBuyFromEstimatedCost".to_string(), None),
515 (
516 "FoxClaimPartial".to_string(),
517 Some(DynType::Struct(vec![(
518 "is_y".to_string(),
519 DynType::Bool,
520 )])),
521 ),
522 (
523 "SolFi".to_string(),
524 Some(DynType::Struct(vec![(
525 "is_quote_to_base".to_string(),
526 DynType::Bool,
527 )])),
528 ),
529 ("SolayerDelegateNoInit".to_string(), None),
530 ("SolayerUndelegateNoInit".to_string(), None),
531 (
532 "TokenMill".to_string(),
533 Some(DynType::Struct(vec![(
534 "side".to_string(),
535 DynType::Enum(vec![
536 ("Bid".to_string(), None),
537 ("Ask".to_string(), None),
538 ]),
539 )])),
540 ),
541 ("DaosFunBuy".to_string(), None),
542 ("DaosFunSell".to_string(), None),
543 ("ZeroFi".to_string(), None),
544 ("StakeDexWithdrawWrappedSol".to_string(), None),
545 ("VirtualsBuy".to_string(), None),
546 ("VirtualsSell".to_string(), None),
547 (
548 "Peren".to_string(),
549 Some(DynType::Struct(vec![
550 ("in_index".to_string(), DynType::U8),
551 ("out_index".to_string(), DynType::U8),
552 ])),
553 ),
554 ("PumpdotfunAmmBuy".to_string(), None),
555 ("PumpdotfunAmmSell".to_string(), None),
556 ("Gamma".to_string(), None),
557 ]),
558 ),
559 ("Percent".to_string(), DynType::U8),
560 ("InputIndex".to_string(), DynType::U8),
561 ("OutputIndex".to_string(), DynType::U8),
562 ]))),
563 },
564 ParamInput {
565 name: "InAmount".to_string(),
566 param_type: DynType::U64,
567 },
568 ParamInput {
569 name: "QuotedOutAmount".to_string(),
570 param_type: DynType::U64,
571 },
572 ParamInput {
573 name: "SlippageBps".to_string(),
574 param_type: DynType::U16,
575 },
576 ParamInput {
577 name: "PlatformFeeBps".to_string(),
578 param_type: DynType::U8,
579 },
580 ],
581 accounts_names: vec![
582 "TokenProgram".to_string(),
583 "UserTransferAuthority".to_string(),
584 "UserSourceTokenAccount".to_string(),
585 "UserDestinationTokenAccount".to_string(),
586 "DestinationTokenAccount".to_string(),
587 "PlatformFeeAccount".to_string(),
588 "EventAuthority".to_string(),
589 "Program".to_string(),
590 ],
591 };
592
593 let result = svm_decode_instructions(ix_signature, &instructions, true)
594 .context("decode failed")
595 .unwrap();
596
597 let mut file = File::create("decoded_instructions.parquet").unwrap();
599 let mut writer =
600 parquet::arrow::ArrowWriter::try_new(&mut file, result.schema(), None).unwrap();
601 writer.write(&result).unwrap();
602 writer.close().unwrap();
603 }
604
605 #[test]
606 fn test_instruction_signature_to_arrow_schema() {
608 let signature = InstructionSignature {
610 discriminator: vec![],
611 params: vec![
612 ParamInput {
613 name: "amount".to_string(),
614 param_type: DynType::U64,
615 },
616 ParamInput {
617 name: "is_valid".to_string(),
618 param_type: DynType::Bool,
619 },
620 ParamInput {
621 name: "amm".to_string(),
622 param_type: DynType::FixedArray(Box::new(DynType::U8), 32),
623 },
624 ],
625 accounts_names: vec!["source".to_string(), "destination".to_string()],
626 };
627
628 let schema = instruction_signature_to_arrow_schema(&signature).unwrap();
630
631 assert_eq!(schema.fields().len(), 5); let amount_field = schema.field_with_name("amount").unwrap();
636 assert_eq!(amount_field.name(), "amount");
637 assert!(amount_field.is_nullable());
638
639 let is_valid_field = schema.field_with_name("is_valid").unwrap();
640 assert_eq!(is_valid_field.name(), "is_valid");
641 assert!(is_valid_field.is_nullable());
642
643 let amm_field = schema.field_with_name("amm").unwrap();
644 assert_eq!(amm_field.name(), "amm");
645 assert!(amm_field.is_nullable());
646
647 let source_field = schema.field_with_name("source").unwrap();
649 assert_eq!(source_field.name(), "source");
650 assert_eq!(source_field.data_type(), &DataType::Binary);
651 assert!(source_field.is_nullable());
652
653 let dest_field = schema.field_with_name("destination").unwrap();
654 assert_eq!(dest_field.name(), "destination");
655 assert_eq!(dest_field.data_type(), &DataType::Binary);
656 assert!(dest_field.is_nullable());
657 }
658}