44
55use crate :: {
66 asset_registry:: AssetRegistry ,
7- state:: { AssetsStorageConfig , State } ,
7+ state:: { AssetsStorageConfig , State , StoreTransactions } ,
88} ;
99use acropolis_common:: {
1010 messages:: { CardanoMessage , Message , StateQuery , StateQueryResponse } ,
@@ -26,12 +26,14 @@ const DEFAULT_ASSET_DELTAS_SUBSCRIBE_TOPIC: (&str, &str) =
2626 ( "asset-deltas-subscribe-topic" , "cardano.asset.deltas" ) ;
2727const DEFAULT_UTXO_DELTAS_SUBSCRIBE_TOPIC : ( & str , & str ) =
2828 ( "utxo-deltas-subscribe-topic" , "cardano.utxo.deltas" ) ;
29+ const DEFAULT_ADDRESS_DELTAS_SUBSCRIBE_TOPIC : ( & str , & str ) =
30+ ( "address-deltas-subscribe-topic" , "cardano.address.delta" ) ;
2931
3032// Configuration defaults
3133const DEFAULT_STORE_ASSETS : ( & str , bool ) = ( "store-assets" , false ) ;
3234const DEFAULT_STORE_INFO : ( & str , bool ) = ( "store-info" , false ) ;
3335const DEFAULT_STORE_HISTORY : ( & str , bool ) = ( "store-history" , false ) ;
34- const DEFAULT_STORE_TRANSACTIONS : ( & str , bool ) = ( "store-transactions" , false ) ;
36+ const DEFAULT_STORE_TRANSACTIONS : ( & str , & str ) = ( "store-transactions" , "none" ) ;
3537const DEFAULT_STORE_ADDRESSES : ( & str , bool ) = ( "store-addresses" , false ) ;
3638const DEFAULT_INDEX_BY_POLICY : ( & str , bool ) = ( "index-by-policy" , false ) ;
3739
@@ -48,13 +50,18 @@ impl AssetsState {
4850 history : Arc < Mutex < StateHistory < State > > > ,
4951 mut asset_deltas_subscription : Box < dyn Subscription < Message > > ,
5052 mut utxo_deltas_subscription : Option < Box < dyn Subscription < Message > > > ,
53+ mut address_deltas_subscription : Option < Box < dyn Subscription < Message > > > ,
5154 storage_config : AssetsStorageConfig ,
5255 registry : Arc < Mutex < AssetRegistry > > ,
5356 ) -> Result < ( ) > {
5457 if let Some ( sub) = utxo_deltas_subscription. as_mut ( ) {
5558 let _ = sub. read ( ) . await ?;
5659 info ! ( "Consumed initial message from utxo_deltas_subscription" ) ;
5760 }
61+ if let Some ( sub) = address_deltas_subscription. as_mut ( ) {
62+ let _ = sub. read ( ) . await ?;
63+ info ! ( "Consumed initial message from address_deltas_subscription" ) ;
64+ }
5865 // Main loop of synchronised messages
5966 loop {
6067 // Get current state snapshot
@@ -115,20 +122,58 @@ impl AssetsState {
115122 CardanoMessage :: UTXODeltas ( utxo_deltas_msg) ,
116123 ) ) => {
117124 Self :: check_sync ( & current_block, block_info, "utxo" ) ;
118- let mut reg = registry. lock ( ) . await ;
119- state =
120- match state. handle_cip68_metadata ( & utxo_deltas_msg. deltas , & mut * reg) {
125+
126+ if storage_config. store_info {
127+ let reg = registry. lock ( ) . await ;
128+ state =
129+ match state. handle_cip68_metadata ( & utxo_deltas_msg. deltas , & * reg) {
130+ Ok ( new_state) => new_state,
131+ Err ( e) => {
132+ error ! ( "CIP-68 metadata handling error: {e:#}" ) ;
133+ state
134+ }
135+ } ;
136+ }
137+
138+ if storage_config. store_transactions . is_enabled ( ) {
139+ let reg = registry. lock ( ) . await ;
140+ state = match state. handle_transactions ( & utxo_deltas_msg. deltas , & * reg)
141+ {
121142 Ok ( new_state) => new_state,
122143 Err ( e) => {
123- error ! ( "CIP-68 metadata handling error: {e:#}" ) ;
144+ error ! ( "Transactions handling error: {e:#}" ) ;
124145 state
125146 }
126147 } ;
148+ }
127149 }
128150 other => error ! ( "Unexpected message on utxo-deltas subscription: {other:?}" ) ,
129151 }
130152 }
131153
154+ if let Some ( sub) = address_deltas_subscription. as_mut ( ) {
155+ let ( _, address_msg) = sub. read ( ) . await ?;
156+ match address_msg. as_ref ( ) {
157+ Message :: Cardano ( (
158+ ref block_info,
159+ CardanoMessage :: AddressDeltas ( address_deltas_msg) ,
160+ ) ) => {
161+ Self :: check_sync ( & current_block, block_info, "address" ) ;
162+
163+ let reg = registry. lock ( ) . await ;
164+ state = match state. handle_address_deltas ( & address_deltas_msg. deltas , & * reg)
165+ {
166+ Ok ( new_state) => new_state,
167+ Err ( e) => {
168+ error ! ( "Address deltas handling error: {e:#}" ) ;
169+ state
170+ }
171+ } ;
172+ }
173+ other => error ! ( "Unexpected message on address-deltas subscription: {other:?}" ) ,
174+ }
175+ }
176+
132177 // Commit state
133178 {
134179 let mut h = history. lock ( ) . await ;
@@ -165,12 +210,27 @@ impl AssetsState {
165210 config. get_string ( key. 0 ) . unwrap_or_else ( |_| key. 1 . to_string ( ) )
166211 }
167212
213+ fn get_transactions_flag ( config : & Config , key : ( & str , & str ) ) -> StoreTransactions {
214+ let val = get_string_flag ( config, key) ;
215+ match val. as_str ( ) {
216+ "none" => StoreTransactions :: None ,
217+ "all" => StoreTransactions :: All ,
218+ s => {
219+ if let Ok ( n) = s. parse :: < u64 > ( ) {
220+ StoreTransactions :: Last ( n)
221+ } else {
222+ StoreTransactions :: None
223+ }
224+ }
225+ }
226+ }
227+
168228 // Get configuration flags and topis
169229 let storage_config = AssetsStorageConfig {
170230 store_assets : get_bool_flag ( & config, DEFAULT_STORE_ASSETS ) ,
171231 store_info : get_bool_flag ( & config, DEFAULT_STORE_INFO ) ,
172232 store_history : get_bool_flag ( & config, DEFAULT_STORE_HISTORY ) ,
173- store_transactions : get_bool_flag ( & config, DEFAULT_STORE_TRANSACTIONS ) ,
233+ store_transactions : get_transactions_flag ( & config, DEFAULT_STORE_TRANSACTIONS ) ,
174234 store_addresses : get_bool_flag ( & config, DEFAULT_STORE_ADDRESSES ) ,
175235 index_by_policy : get_bool_flag ( & config, DEFAULT_INDEX_BY_POLICY ) ,
176236 } ;
@@ -180,13 +240,22 @@ impl AssetsState {
180240 info ! ( "Creating subscriber on '{asset_deltas_subscribe_topic}'" ) ;
181241
182242 let utxo_deltas_subscribe_topic: Option < String > =
183- if storage_config. store_info || storage_config. store_transactions {
243+ if storage_config. store_info || storage_config. store_transactions . is_enabled ( ) {
184244 let topic = get_string_flag ( & config, DEFAULT_UTXO_DELTAS_SUBSCRIBE_TOPIC ) ;
185245 info ! ( "Creating subscriber on '{topic}'" ) ;
186246 Some ( topic)
187247 } else {
188248 None
189249 } ;
250+
251+ let address_deltas_subscribe_topic: Option < String > = if storage_config. store_addresses {
252+ let topic = get_string_flag ( & config, DEFAULT_ADDRESS_DELTAS_SUBSCRIBE_TOPIC ) ;
253+ info ! ( "Creating subscriber on '{topic}'" ) ;
254+ Some ( topic)
255+ } else {
256+ None
257+ } ;
258+
190259 let assets_query_topic = get_string_flag ( & config, DEFAULT_ASSETS_QUERY_TOPIC ) ;
191260 info ! ( "Creating asset query handler on '{assets_query_topic}'" ) ;
192261
@@ -215,7 +284,10 @@ impl AssetsState {
215284 ) ) ) ;
216285 } ;
217286
218- let state = history. lock ( ) . await . get_current_state ( ) ;
287+ let state = {
288+ let h = history. lock ( ) . await ;
289+ h. get_current_state ( )
290+ } ;
219291
220292 let response = match query {
221293 AssetsStateQuery :: GetAssetsList => {
@@ -295,7 +367,7 @@ impl AssetsState {
295367 Err ( e) => AssetsStateQueryResponse :: Error ( e. to_string ( ) ) ,
296368 } ,
297369 None => {
298- if state. config . store_transactions {
370+ if state. config . store_transactions . is_enabled ( ) {
299371 AssetsStateQueryResponse :: NotFound
300372 } else {
301373 AssetsStateQueryResponse :: Error (
@@ -354,13 +426,19 @@ impl AssetsState {
354426 } else {
355427 None
356428 } ;
429+ let address_deltas_sub = if let Some ( topic) = & address_deltas_subscribe_topic {
430+ Some ( context. subscribe ( topic) . await ?)
431+ } else {
432+ None
433+ } ;
357434
358435 // Start run task
359436 context. run ( async move {
360437 Self :: run (
361438 history_run,
362439 asset_deltas_sub,
363440 utxo_deltas_sub,
441+ address_deltas_sub,
364442 storage_config,
365443 registry_run,
366444 )
0 commit comments