refactor(batcher): add user state & other code quality refactors

Makefile

@@ -271,7 +271,7 @@ batcher_send_risc0_burst:
 		--proof ../../scripts/test_files/risc_zero/fibonacci_proof_generator/risc_zero_fibonacci.proof \
         --vm_program ../../scripts/test_files/risc_zero/fibonacci_proof_generator/fibonacci_id.bin \
         --public_input ../../scripts/test_files/risc_zero/fibonacci_proof_generator/risc_zero_fibonacci.pub \
-        --repetitions $(BURST_SIZE) \
+        --repetitions 500 \
 		--proof_generator_addr 0x66f9664f97F2b50F62D13eA064982f936dE76657 \
 		--rpc_url $(RPC_URL) \
 		--payment_service_addr $(BATCHER_PAYMENTS_CONTRACT_ADDRESS)
@@ -361,7 +361,8 @@ batcher_send_halo2_ipa_task_burst_5: batcher/target/release/aligned
 		--proof ../../scripts/test_files/halo2_ipa/proof.bin \
 		--public_input ../../scripts/test_files/halo2_ipa/pub_input.bin \
 		--vk ../../scripts/test_files/halo2_ipa/params.bin \
-		--repetitions 5 \
+		--repetitions 1000 \
+		--private_key 0xdbda1821b80551c9d65939329250298aa3472ba22feea921c0cf5d620ea67b97 \
 		--rpc_url $(RPC_URL) \
 		--payment_service_addr $(BATCHER_PAYMENTS_CONTRACT_ADDRESS)
 

batcher/aligned-batcher/src/connection.rs

@@ -0,0 +1,68 @@
+use std::sync::Arc;
+
+use aligned_sdk::{
+    communication::serialization::cbor_serialize,
+    core::types::{BatchInclusionData, ResponseMessage, VerificationCommitmentBatch},
+};
+use futures_util::{stream::SplitSink, SinkExt};
+use lambdaworks_crypto::merkle_tree::merkle::MerkleTree;
+use log::{error, info};
+use serde::Serialize;
+use tokio::{net::TcpStream, sync::RwLock};
+use tokio_tungstenite::{
+    tungstenite::{Error, Message},
+    WebSocketStream,
+};
+
+use crate::types::{batch_queue::BatchQueueEntry, errors::BatcherError};
+
+pub(crate) type WsMessageSink = Arc<RwLock<SplitSink<WebSocketStream<TcpStream>, Message>>>;
+
+pub(crate) async fn send_batch_inclusion_data_responses(
+    finalized_batch: Vec<BatchQueueEntry>,
+    batch_merkle_tree: &MerkleTree<VerificationCommitmentBatch>,
+) -> Result<(), BatcherError> {
+    for (vd_batch_idx, entry) in finalized_batch.iter().enumerate() {
+        let batch_inclusion_data = BatchInclusionData::new(vd_batch_idx, batch_merkle_tree);
+        let response = ResponseMessage::BatchInclusionData(batch_inclusion_data);
+
+        let serialized_response = cbor_serialize(&response)
+            .map_err(|e| BatcherError::SerializationError(e.to_string()))?;
+
+        let Some(ws_sink) = entry.messaging_sink.as_ref() else {
+            return Err(BatcherError::WsSinkEmpty);
+        };
+
+        let sending_result = ws_sink
+            .write()
+            .await
+            .send(Message::binary(serialized_response))
+            .await;
+
+        match sending_result {
+            Err(Error::AlreadyClosed) => (),
+            Err(e) => error!("Error while sending batch inclusion data response: {}", e),
+            Ok(_) => (),
+        }
+
+        info!("Response sent");
+    }
+
+    Ok(())
+}
+
+pub(crate) async fn send_message<T: Serialize>(ws_conn_sink: WsMessageSink, message: T) {
+    match cbor_serialize(&message) {
+        Ok(serialized_response) => {
+            if let Err(err) = ws_conn_sink
+                .write()
+                .await
+                .send(Message::binary(serialized_response))
+                .await
+            {
+                error!("Error while sending message: {}", err)
+            }
+        }
+        Err(e) => error!("Error while serializing message: {}", e),
+    }
+}