11use {
22 borsh:: BorshDeserialize ,
33 clap:: Parser ,
4- core:: panic,
5- observation:: {
6- Body ,
7- SignedBody ,
8- } ,
94 posted_message:: PostedMessageUnreliableData ,
105 secp256k1:: SecretKey ,
6+ signed_body:: SignedBody ,
117 solana_account_decoder:: UiAccountEncoding ,
128 solana_client:: {
139 nonblocking:: pubsub_client:: PubsubClient ,
1612 RpcAccountInfoConfig ,
1713 RpcProgramAccountsConfig ,
1814 } ,
15+ rpc_filter:: {
16+ Memcmp ,
17+ RpcFilterType ,
18+ } ,
1919 } ,
2020 solana_sdk:: pubkey:: Pubkey ,
2121 std:: {
@@ -25,14 +25,16 @@ use {
2525 } ,
2626 tokio:: time:: sleep,
2727 tokio_stream:: StreamExt ,
28+ wormhole_sdk:: {
29+ vaa:: Body ,
30+ Address ,
31+ Chain ,
32+ } ,
2833} ;
2934
3035mod config;
31- mod observation;
3236mod posted_message;
33- mod serde_array;
34-
35- const PYTHNET_CHAIN_ID : u16 = 26 ;
37+ mod signed_body;
3638
3739struct ListenerConfig {
3840 ws_url : String ,
@@ -41,7 +43,8 @@ struct ListenerConfig {
4143 accumulator_address : Pubkey ,
4244}
4345
44- fn find_message_pda ( wormhole_pid : & Pubkey , ring_index : u32 ) -> Pubkey {
46+ fn find_message_pda ( wormhole_pid : & Pubkey , slot : u64 ) -> Pubkey {
47+ let ring_index = ( slot % 10_000 ) as u32 ;
4548 Pubkey :: find_program_address (
4649 & [ b"AccumulatorMessage" , & ring_index. to_be_bytes ( ) ] ,
4750 wormhole_pid,
@@ -55,7 +58,10 @@ async fn run_listener(config: ListenerConfig) -> Result<(), PubsubClientError> {
5558 . program_subscribe (
5659 & config. wormhole_pid ,
5760 Some ( RpcProgramAccountsConfig {
58- filters : None ,
61+ filters : Some ( vec ! [ RpcFilterType :: Memcmp ( Memcmp :: new(
62+ 0 ,
63+ solana_client:: rpc_filter:: MemcmpEncodedBytes :: Bytes ( b"msu" . to_vec( ) ) ,
64+ ) ) ] ) ,
5965 account_config : RpcAccountInfoConfig {
6066 encoding : Some ( UiAccountEncoding :: Base64 ) ,
6167 data_slice : None ,
@@ -71,55 +77,54 @@ async fn run_listener(config: ListenerConfig) -> Result<(), PubsubClientError> {
7177 . await ?;
7278
7379 while let Some ( update) = stream. next ( ) . await {
74- let message_pda =
75- find_message_pda ( & config . wormhole_pid , ( update. context . slot % 10_000 ) as u32 ) ;
76- if message_pda . to_string ( ) != update . value . pubkey {
80+ if find_message_pda ( & config . wormhole_pid , update . context . slot ) . to_string ( )
81+ != update. value . pubkey
82+ {
7783 continue ; // Skip updates that are not for the expected PDA
7884 }
7985
80- let unreliable_data: Option < PostedMessageUnreliableData > = update
81- . value
82- . account
83- . data
84- . decode ( )
85- . and_then ( |data| BorshDeserialize :: deserialize ( & mut data. as_slice ( ) ) . ok ( ) ) ;
86-
87- if let Some ( unreliable_data) = unreliable_data {
88- if PYTHNET_CHAIN_ID != unreliable_data. emitter_chain {
89- continue ;
90- }
91- if config. accumulator_address != Pubkey :: from ( unreliable_data. emitter_address ) {
92- continue ;
93- }
94-
95- let body = Body {
96- timestamp : unreliable_data. submission_time ,
97- nonce : unreliable_data. nonce ,
98- emitter_chain : unreliable_data. emitter_chain ,
99- emitter_address : unreliable_data. emitter_address ,
100- sequence : unreliable_data. sequence ,
101- consistency_level : unreliable_data. consistency_level ,
102- payload : unreliable_data. payload . clone ( ) ,
86+ let unreliable_data: PostedMessageUnreliableData = {
87+ let data = match update. value . account . data . decode ( ) {
88+ Some ( data) => data,
89+ None => {
90+ tracing:: error!( "Failed to decode account data" ) ;
91+ continue ;
92+ }
10393 } ;
10494
105- match body. sign ( config. secret_key . secret_bytes ( ) ) {
106- Ok ( signature) => {
107- let signed_body = SignedBody {
108- version : unreliable_data. vaa_version ,
109- signature,
110- body,
111- } ;
112- println ! ( "Signed Body: {:?}" , signed_body) ;
95+ match BorshDeserialize :: deserialize ( & mut data. as_slice ( ) ) {
96+ Ok ( data) => data,
97+ Err ( e) => {
98+ tracing:: error!( error = ?e, "Invalid unreliable data format" ) ;
99+ continue ;
113100 }
114- Err ( e) => tracing:: error!( error = ?e, "Failed to sign body" ) ,
115101 }
102+ } ;
103+
104+ if Chain :: Pythnet != unreliable_data. emitter_chain . into ( ) {
105+ continue ;
106+ }
107+ if config. accumulator_address != Pubkey :: from ( unreliable_data. emitter_address ) {
108+ continue ;
116109 }
110+
111+ let body = Body {
112+ timestamp : unreliable_data. submission_time ,
113+ nonce : unreliable_data. nonce ,
114+ emitter_chain : unreliable_data. emitter_chain . into ( ) ,
115+ emitter_address : Address ( unreliable_data. emitter_address ) ,
116+ sequence : unreliable_data. sequence ,
117+ consistency_level : unreliable_data. consistency_level ,
118+ payload : unreliable_data. payload . clone ( ) ,
119+ } ;
120+
121+ match SignedBody :: try_new ( body, config. secret_key ) {
122+ Ok ( signed_body) => println ! ( "Signed Body: {:?}" , signed_body) ,
123+ Err ( e) => tracing:: error!( error = ?e, "Failed to sign body" ) ,
124+ } ;
117125 }
118126
119- tokio:: spawn ( async move {
120- // Wait for the stream to finish
121- unsubscribe ( ) . await
122- } ) ;
127+ tokio:: spawn ( async move { unsubscribe ( ) . await } ) ;
123128
124129 Err ( PubsubClientError :: ConnectionClosed (
125130 "Stream ended" . to_string ( ) ,
@@ -137,11 +142,7 @@ fn load_secret_key(path: String) -> SecretKey {
137142 . expect ( "Invalid secret key file" )
138143 . trim ( )
139144 . to_string ( ) ;
140- if let Ok ( secret_key) = SecretKey :: from_str ( & content) {
141- return secret_key;
142- }
143-
144- panic ! ( "Invalid secret key" ) ;
145+ SecretKey :: from_str ( & content) . expect ( "Invalid secret key" )
145146}
146147
147148#[ tokio:: main]
@@ -152,8 +153,8 @@ async fn main() {
152153 . await
153154 . expect ( "Invalid WebSocket URL" ) ;
154155 drop ( client) ; // Drop the client to avoid holding the connection open
155- let accumulator_address =
156- Pubkey :: from_str ( & run_options . accumulator_address ) . expect ( "Invalid accumulator address" ) ;
156+ let accumulator_address = Pubkey :: from_str ( "G9LV2mp9ua1znRAfYwZz5cPiJMAbo1T6mbjdQsDZuMJg" )
157+ . expect ( "Invalid accumulator address" ) ;
157158 let wormhole_pid =
158159 Pubkey :: from_str ( & run_options. wormhole_pid ) . expect ( "Invalid Wormhole program ID" ) ;
159160
0 commit comments