diff --git a/pallets/ddc-customers/src/lib.rs b/pallets/ddc-customers/src/lib.rs
index 375ae4699..e24c79116 100644
--- a/pallets/ddc-customers/src/lib.rs
+++ b/pallets/ddc-customers/src/lib.rs
@@ -197,7 +197,7 @@ pub mod pallet {
 		/// Bucket with specific id created
 		BucketCreated { cluster_id: ClusterId, bucket_id: BucketId },
 		/// Bucket with specific id updated
-		BucketUpdated { bucket_id: BucketId },
+		BucketUpdated { cluster_id: ClusterId, bucket_id: BucketId },
 		/// Bucket nodes usage with specific id updated
 		BucketTotalNodesUsageUpdated {
 			cluster_id: ClusterId,
@@ -532,8 +532,9 @@ pub mod pallet {
 			ensure!(bucket.owner_id == owner, Error::<T>::NotBucketOwner);
 
 			bucket.is_public = bucket_params.is_public;
+			let cluster_id = bucket.cluster_id;
 			<Buckets<T>>::insert(bucket_id, bucket);
-			Self::deposit_event(Event::<T>::BucketUpdated { bucket_id });
+			Self::deposit_event(Event::<T>::BucketUpdated { cluster_id, bucket_id });
 
 			Ok(())
 		}
diff --git a/pallets/ddc-customers/src/tests.rs b/pallets/ddc-customers/src/tests.rs
index 1a5fa9208..264e790b9 100644
--- a/pallets/ddc-customers/src/tests.rs
+++ b/pallets/ddc-customers/src/tests.rs
@@ -465,7 +465,7 @@ fn set_bucket_params_works() {
 
 		// Checking that event was emitted
 		assert_eq!(System::events().len(), 2);
-		System::assert_last_event(Event::BucketUpdated { bucket_id }.into());
+		System::assert_last_event(Event::BucketUpdated { cluster_id, bucket_id }.into());
 	})
 }
 
diff --git a/pallets/ddc-payouts/src/benchmarking.rs b/pallets/ddc-payouts/src/benchmarking.rs
index 8138b43a2..cdfd93f2a 100644
--- a/pallets/ddc-payouts/src/benchmarking.rs
+++ b/pallets/ddc-payouts/src/benchmarking.rs
@@ -398,7 +398,7 @@ benchmarks! {
 			(provider, bucket_id, node_usage)
 		}).collect();
 
-	}: _(RawOrigin::Signed(dac_account.clone()), cluster_id, era, batch_index, payees)
+	}: _(RawOrigin::Signed(dac_account.clone()), cluster_id, era, batch_index, payees,0, vec![], (0, ActivityHash::default()))
 	verify {
 		assert!(ActiveBillingReports::<T>::contains_key(cluster_id, era));
 		let billing_report = ActiveBillingReports::<T>::get(cluster_id, era).unwrap();
diff --git a/pallets/ddc-payouts/src/lib.rs b/pallets/ddc-payouts/src/lib.rs
index 9ef461565..cd0f20dd4 100644
--- a/pallets/ddc-payouts/src/lib.rs
+++ b/pallets/ddc-payouts/src/lib.rs
@@ -307,6 +307,8 @@ pub mod pallet {
 
 	#[pallet::call]
 	impl<T: Config> Pallet<T> {
+		// todo! remove extrensics from payout pallet and factor the extrensics implementation into
+		// PayoutProcessor trait
 		#[pallet::call_index(0)]
 		#[pallet::weight(T::WeightInfo::begin_billing_report())]
 		pub fn begin_billing_report(
@@ -340,6 +342,8 @@ pub mod pallet {
 			Ok(())
 		}
 
+		// todo! remove extrensics from payout pallet and factor the extrensics implementation into
+		// PayoutProcessor trait
 		#[pallet::call_index(1)]
 		#[pallet::weight(T::WeightInfo::begin_charging_customers())]
 		pub fn begin_charging_customers(
@@ -367,9 +371,12 @@ pub mod pallet {
 			Ok(())
 		}
 
+		// todo! remove extrensics from payout pallet and factor the extrensics implementation into
+		// + pass values by reference PayoutProcessor trait
 		#[pallet::call_index(2)]
 		#[pallet::weight(T::WeightInfo::send_charging_customers_batch(payers.len().saturated_into()))]
-		#[allow(clippy::too_many_arguments)] // todo! need to refactor this
+		// todo! remove clippy::too_many_arguments
+		#[allow(clippy::too_many_arguments)]
 		pub fn send_charging_customers_batch(
 			origin: OriginFor<T>,
 			cluster_id: ClusterId,
@@ -544,6 +551,8 @@ pub mod pallet {
 			Ok(())
 		}
 
+		// todo! remove extrensics from payout pallet and factor the extrensics implementation into
+		// PayoutProcessor trait
 		#[pallet::call_index(3)]
 		#[pallet::weight(T::WeightInfo::end_charging_customers())]
 		pub fn end_charging_customers(
@@ -646,6 +655,8 @@ pub mod pallet {
 			Ok(())
 		}
 
+		// todo! remove extrensics from payout pallet and factor the extrensics implementation into
+		// PayoutProcessor trait
 		#[pallet::call_index(4)]
 		#[pallet::weight(T::WeightInfo::begin_rewarding_providers())]
 		pub fn begin_rewarding_providers(
@@ -678,14 +689,21 @@ pub mod pallet {
 			Ok(())
 		}
 
+		// todo! remove extrensics from payout pallet and factor the extrensics implementation into
+		// + pass values by reference PayoutProcessor trait
 		#[pallet::call_index(5)]
 		#[pallet::weight(T::WeightInfo::send_rewarding_providers_batch(payees.len().saturated_into()))]
+		// todo! remove clippy::too_many_arguments
+		#[allow(clippy::too_many_arguments)]
 		pub fn send_rewarding_providers_batch(
 			origin: OriginFor<T>,
 			cluster_id: ClusterId,
 			era: DdcEra,
 			batch_index: BatchIndex,
 			payees: Vec<(T::AccountId, BucketId, NodeUsage)>,
+			mmr_size: u64,
+			proof: Vec<ActivityHash>,
+			leaf_with_position: (u64, ActivityHash),
 		) -> DispatchResult {
 			let caller = ensure_signed(origin)?;
 			ensure!(T::ValidatorVisitor::is_ocw_validator(caller), Error::<T>::Unauthorised);
@@ -717,7 +735,8 @@ pub mod pallet {
 					era,
 					batch_index,
 					&payees,
-					&[] // todo! pass from newly added input
+					MerkleProof::new(mmr_size, proof),
+					leaf_with_position
 				),
 				Error::<T>::BatchValidationFailed
 			);
@@ -802,6 +821,8 @@ pub mod pallet {
 			Ok(())
 		}
 
+		// todo! remove extrensics from payout pallet and factor the extrensics implementation into
+		// PayoutProcessor trait
 		#[pallet::call_index(6)]
 		#[pallet::weight(T::WeightInfo::end_rewarding_providers())]
 		pub fn end_rewarding_providers(
@@ -853,6 +874,8 @@ pub mod pallet {
 			Ok(())
 		}
 
+		// todo! remove extrensics from payout pallet and factor the extrensics implementation into
+		// PayoutProcessor trait
 		#[pallet::call_index(7)]
 		#[pallet::weight(T::WeightInfo::end_billing_report())]
 		pub fn end_billing_report(
@@ -1081,7 +1104,118 @@ pub mod pallet {
 	}
 
 	impl<T: Config> PayoutVisitor<T> for Pallet<T> {
-		fn get_billing_report_status(cluster_id: ClusterId, era: DdcEra) -> PayoutState {
+		fn begin_billing_report(
+			origin: T::AccountId,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			start_era: i64,
+			end_era: i64,
+		) -> DispatchResult {
+			let origin = frame_system::RawOrigin::Signed(origin).into();
+			Self::begin_billing_report(origin, cluster_id, era_id, start_era, end_era)
+		}
+
+		fn begin_charging_customers(
+			origin: T::AccountId,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			max_batch_index: BatchIndex,
+		) -> DispatchResult {
+			let origin = frame_system::RawOrigin::Signed(origin).into();
+			Self::begin_charging_customers(origin, cluster_id, era_id, max_batch_index)
+		}
+
+		fn send_charging_customers_batch(
+			origin: T::AccountId,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			batch_index: BatchIndex,
+			payers: Vec<(T::AccountId, BucketId, CustomerUsage)>,
+			mmr_size: u64,
+			proof: Vec<ActivityHash>,
+			leaf_with_position: (u64, ActivityHash),
+		) -> DispatchResult {
+			let origin = frame_system::RawOrigin::Signed(origin).into();
+			Self::send_charging_customers_batch(
+				origin,
+				cluster_id,
+				era_id,
+				batch_index,
+				payers,
+				mmr_size,
+				proof,
+				leaf_with_position,
+			)
+		}
+
+		fn end_charging_customers(
+			origin: T::AccountId,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+		) -> DispatchResult {
+			let origin = frame_system::RawOrigin::Signed(origin).into();
+			Self::end_charging_customers(origin, cluster_id, era_id)
+		}
+
+		fn begin_rewarding_providers(
+			origin: T::AccountId,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			max_batch_index: BatchIndex,
+			total_node_usage: NodeUsage,
+		) -> DispatchResult {
+			let origin = frame_system::RawOrigin::Signed(origin).into();
+			Self::begin_rewarding_providers(
+				origin,
+				cluster_id,
+				era_id,
+				max_batch_index,
+				total_node_usage,
+			)
+		}
+
+		fn send_rewarding_providers_batch(
+			origin: T::AccountId,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			batch_index: BatchIndex,
+			payees: Vec<(T::AccountId, BucketId, NodeUsage)>,
+			mmr_size: u64,
+			proof: Vec<ActivityHash>,
+			leaf_with_position: (u64, ActivityHash),
+		) -> DispatchResult {
+			let origin = frame_system::RawOrigin::Signed(origin).into();
+			Self::send_rewarding_providers_batch(
+				origin,
+				cluster_id,
+				era_id,
+				batch_index,
+				payees,
+				mmr_size,
+				proof,
+				leaf_with_position,
+			)
+		}
+
+		fn end_rewarding_providers(
+			origin: T::AccountId,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+		) -> DispatchResult {
+			let origin = frame_system::RawOrigin::Signed(origin).into();
+			Self::end_rewarding_providers(origin, cluster_id, era_id)
+		}
+
+		fn end_billing_report(
+			origin: T::AccountId,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+		) -> DispatchResult {
+			let origin = frame_system::RawOrigin::Signed(origin).into();
+			Self::end_billing_report(origin, cluster_id, era_id)
+		}
+
+		fn get_billing_report_status(cluster_id: &ClusterId, era: DdcEra) -> PayoutState {
 			let billing_report = ActiveBillingReports::<T>::get(cluster_id, era);
 
 			match billing_report {
diff --git a/pallets/ddc-payouts/src/mock.rs b/pallets/ddc-payouts/src/mock.rs
index 84fa047c7..85d6b4953 100644
--- a/pallets/ddc-payouts/src/mock.rs
+++ b/pallets/ddc-payouts/src/mock.rs
@@ -164,7 +164,8 @@ where
 		_era: DdcEra,
 		_batch_index: BatchIndex,
 		_payees: &[(T::AccountId, BucketId, NodeUsage)],
-		_adjacent_hashes: &[ActivityHash],
+		_proof: MerkleProof<ActivityHash, MergeActivityHash>,
+		_leaf_with_position: (u64, ActivityHash),
 	) -> bool {
 		true
 	}
diff --git a/pallets/ddc-payouts/src/tests.rs b/pallets/ddc-payouts/src/tests.rs
index 7dfe16f11..adb43bded 100644
--- a/pallets/ddc-payouts/src/tests.rs
+++ b/pallets/ddc-payouts/src/tests.rs
@@ -3094,6 +3094,9 @@ fn send_rewarding_providers_batch_fails_uninitialised() {
 				era,
 				batch_index,
 				payees.clone(),
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			),
 			Error::<Test>::Unauthorised
 		);
@@ -3105,6 +3108,9 @@ fn send_rewarding_providers_batch_fails_uninitialised() {
 				era,
 				batch_index,
 				payees.clone(),
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			),
 			BadOrigin
 		);
@@ -3116,6 +3122,9 @@ fn send_rewarding_providers_batch_fails_uninitialised() {
 				era,
 				batch_index,
 				payees.clone(),
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			),
 			Error::<Test>::BillingReportDoesNotExist
 		);
@@ -3135,6 +3144,9 @@ fn send_rewarding_providers_batch_fails_uninitialised() {
 				era,
 				batch_index,
 				payees.clone(),
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			),
 			Error::<Test>::NotExpectedState
 		);
@@ -3153,6 +3165,9 @@ fn send_rewarding_providers_batch_fails_uninitialised() {
 				era,
 				batch_index,
 				payees.clone(),
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			),
 			Error::<Test>::NotExpectedState
 		);
@@ -3175,6 +3190,9 @@ fn send_rewarding_providers_batch_fails_uninitialised() {
 				era,
 				batch_index,
 				payees.clone(),
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			),
 			Error::<Test>::NotExpectedState
 		);
@@ -3197,6 +3215,9 @@ fn send_rewarding_providers_batch_fails_uninitialised() {
 				era,
 				batch_index,
 				payees.clone(),
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			),
 			Error::<Test>::NotExpectedState
 		);
@@ -3214,6 +3235,9 @@ fn send_rewarding_providers_batch_fails_uninitialised() {
 				era,
 				batch_index,
 				payees,
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			),
 			Error::<Test>::NotExpectedState
 		);
@@ -3368,6 +3392,9 @@ fn send_rewarding_providers_batch_works() {
 			era,
 			batch_node_index,
 			payees1,
+			0,
+			vec![],
+			(0, ActivityHash::default()),
 		));
 
 		let ratio1_transfer = Perquintill::from_rational(
@@ -3455,6 +3482,9 @@ fn send_rewarding_providers_batch_works() {
 			era,
 			batch_node_index + 1,
 			payees2,
+			0,
+			vec![],
+			(0, ActivityHash::default()),
 		));
 
 		let ratio3_transfer = Perquintill::from_rational(
@@ -3735,6 +3765,9 @@ fn send_rewarding_providers_batch_100_nodes_small_usage_works() {
 				era,
 				batch_node_index,
 				batch.to_vec(),
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			));
 
 			let mut batch_charge = 0;
@@ -4013,6 +4046,9 @@ fn send_rewarding_providers_batch_100_nodes_large_usage_works() {
 				era,
 				batch_node_index,
 				batch.to_vec(),
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			));
 
 			let mut batch_charge = 0;
@@ -4290,6 +4326,9 @@ fn send_rewarding_providers_batch_100_nodes_small_large_usage_works() {
 				era,
 				batch_node_index,
 				batch.to_vec(),
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			));
 
 			let mut batch_charge = 0;
@@ -4526,6 +4565,9 @@ fn send_rewarding_providers_batch_100_nodes_random_usage_works() {
 				era,
 				batch_node_index,
 				batch.to_vec(),
+				0,
+				vec![],
+				(0, ActivityHash::default()),
 			));
 
 			let mut batch_charge = 0;
@@ -4730,6 +4772,9 @@ fn end_rewarding_providers_fails_uninitialised() {
 			era,
 			batch_index,
 			payees,
+			0,
+			vec![],
+			(0, ActivityHash::default()),
 		));
 
 		assert_noop!(
@@ -4830,6 +4875,9 @@ fn end_rewarding_providers_works() {
 			era,
 			batch_index,
 			payees,
+			0,
+			vec![],
+			(0, ActivityHash::default()),
 		));
 
 		assert_ok!(DdcPayouts::end_rewarding_providers(
@@ -4972,6 +5020,9 @@ fn end_billing_report_fails_uninitialised() {
 			era,
 			batch_index,
 			payees.clone(),
+			0,
+			vec![],
+			(0, ActivityHash::default()),
 		));
 
 		assert_noop!(
@@ -4985,6 +5036,9 @@ fn end_billing_report_fails_uninitialised() {
 			era,
 			batch_index + 1,
 			payees,
+			0,
+			vec![],
+			(0, ActivityHash::default()),
 		));
 
 		assert_noop!(
@@ -5067,6 +5121,9 @@ fn end_billing_report_works() {
 			era,
 			batch_index,
 			payees,
+			0,
+			vec![],
+			(0, ActivityHash::default()),
 		));
 
 		assert_ok!(DdcPayouts::end_rewarding_providers(
diff --git a/pallets/ddc-verification/src/lib.rs b/pallets/ddc-verification/src/lib.rs
index acefd64cb..cd8429200 100644
--- a/pallets/ddc-verification/src/lib.rs
+++ b/pallets/ddc-verification/src/lib.rs
@@ -14,7 +14,7 @@ use core::str;
 use ddc_primitives::{
 	traits::{ClusterManager, NodeVisitor, PayoutVisitor, ValidatorVisitor},
 	ActivityHash, BatchIndex, ClusterId, CustomerUsage, DdcEra, NodeParams, NodePubKey, NodeUsage,
-	StorageNodeMode, StorageNodeParams,
+	PayoutState, StorageNodeMode, StorageNodeParams,
 };
 use frame_support::{
 	pallet_prelude::*,
@@ -187,6 +187,21 @@ pub mod pallet {
 			era_id: DdcEra,
 			validator: T::AccountId,
 		},
+		BeginChargingCustomersTransactionError {
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			validator: T::AccountId,
+		},
+		EmptyCustomerActivity {
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			validator: T::AccountId,
+		},
+		BatchIndexConversionFailed {
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			validator: T::AccountId,
+		},
 		NoAvailableSigner {
 			validator: T::AccountId,
 		},
@@ -240,6 +255,18 @@ pub mod pallet {
 			cluster_id: ClusterId,
 			era_id: DdcEra,
 		},
+		BeginChargingCustomersTransactionError {
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+		},
+		EmptyCustomerActivity {
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+		},
+		BatchIndexConversionFailed {
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+		},
 		NoAvailableSigner,
 		NotEnoughDACNodes {
 			num_nodes: u16,
@@ -539,26 +566,44 @@ pub mod pallet {
 				},
 			}
 
-			/* match Self::process_start_payout(&cluster_id) {
+			// todo! factor out as macro as this is repetitive
+			match Self::prepare_begin_billing_report(&cluster_id) {
 				Ok(Some((era_id, start_era, end_era))) => {
 					log::info!(
-						"Start payout processed successfully for cluster_id: {:?}, era_id: {:?}",
+						"process_start_payout processed successfully for cluster_id: {:?}, era_id: {:?}",
 						cluster_id,
 						era_id
 					);
 
-					if let Some((_, res)) = signer.send_signed_transaction(|_acct| {
-						let call = T::PayoutVisitor::get_begin_billing_report_call(
-							cluster_id, era_id, start_era, end_era,
-						);
-						call.into() // Convert to Call<T>
+					if let Some((_, res)) = signer.send_signed_transaction(|_acc| {
+						Call::begin_billing_report { cluster_id, era_id, start_era, end_era }
 					}) {
 						match res {
-							Ok(_) => log::info!("Successfully sent signed transaction"),
-							Err(e) => log::error!("Failed to send signed transaction: {:?}", e),
+							Ok(_) => {
+								// Extrinsic call succeeded
+								log::info!(
+									"Sent begin_billing_report successfully for cluster_id: {:?}, era_id: {:?}",
+									cluster_id,
+									era_id
+								);
+							},
+							Err(e) => {
+								log::error!(
+										"Error to post begin_billing_report for cluster_id: {:?}, era_id: {:?}: {:?}",
+										cluster_id,
+										era_id,
+										e
+									);
+								// Extrinsic call failed
+								errors.push(OCWError::BeginBillingReportTransactionError {
+									cluster_id,
+									era_id,
+								});
+							},
 						}
 					} else {
-						log::error!("No local account available");
+						log::error!("No account available to sign the transaction");
+						errors.push(OCWError::NoAvailableSigner);
 					}
 				},
 				Ok(None) => {
@@ -567,7 +612,58 @@ pub mod pallet {
 				Err(e) => {
 					errors.push(e);
 				},
-			} */
+			}
+
+			// todo! factor out as macro as this is repetitive
+			match Self::prepare_begin_charging_customers(&cluster_id) {
+				Ok(Some((era_id, max_batch_index))) => {
+					log::info!(
+						"process_start_payout processed successfully for cluster_id: {:?}, era_id: {:?}",
+						cluster_id,
+						era_id
+					);
+
+					if let Some((_, res)) = signer.send_signed_transaction(|_acc| {
+						Call::begin_charging_customers { cluster_id, era_id, max_batch_index }
+					}) {
+						match res {
+							Ok(_) => {
+								// Extrinsic call succeeded
+								log::info!(
+									"Sent begin_charging_customers successfully for cluster_id: {:?}, era_id: {:?}",
+									cluster_id,
+									era_id
+								);
+							},
+							Err(e) => {
+								log::error!(
+										"Error to post begin_charging_customers for cluster_id: {:?}, era_id: {:?}: {:?}",
+										cluster_id,
+										era_id,
+										e
+									);
+								// Extrinsic call failed
+								errors.push(OCWError::BeginChargingCustomersTransactionError {
+									cluster_id,
+									era_id,
+								});
+							},
+						}
+					} else {
+						log::error!("No account available to sign the transaction");
+						errors.push(OCWError::NoAvailableSigner);
+					}
+				},
+				Ok(None) => {
+					log::info!(
+						"No era for begin_charging_customers for cluster_id: {:?}",
+						cluster_id
+					);
+				},
+				Err(e) => {
+					errors.push(e);
+				},
+			}
 
 			if !errors.is_empty() {
 				// Send errors as extrinsics
@@ -669,17 +765,48 @@ pub mod pallet {
 		// adjacent_hashes) }
 
 		#[allow(dead_code)]
-		pub(crate) fn process_start_payout(
+		pub(crate) fn prepare_begin_billing_report(
 			cluster_id: &ClusterId,
 		) -> Result<Option<(DdcEra, i64, i64)>, OCWError> {
 			Ok(Self::get_era_for_payout(cluster_id, EraValidationStatus::ReadyForPayout))
+			// todo! get start and end values based on result
 		}
 
-		pub(crate) fn _get_activities_in_consensus<A: Activity, B: Activity>(
-			_cluster_id: &ClusterId,
-			_era_id: DdcEra,
-		) -> Result<(Vec<A>, Vec<B>), Vec<OCWError>> {
-			unimplemented!()
+		pub(crate) fn prepare_begin_charging_customers(
+			cluster_id: &ClusterId,
+		) -> Result<Option<(DdcEra, BatchIndex)>, OCWError> {
+			if let Some((era_id, _start, _end)) =
+				Self::get_era_for_payout(cluster_id, EraValidationStatus::PayoutInProgress)
+			{
+				if T::PayoutVisitor::get_billing_report_status(cluster_id, era_id) ==
+					PayoutState::Initialized
+				{
+					if let Some((_, _, customers_activity_batch_roots, _, _, _)) =
+						Self::fetch_validation_activities::<CustomerActivity, NodeActivity>(
+							cluster_id, era_id,
+						) {
+						if let Some(batch_index) =
+							customers_activity_batch_roots.len().checked_sub(1)
+						{
+							let batch_index: u16 = batch_index.try_into().map_err(|_| {
+								OCWError::BatchIndexConversionFailed {
+									cluster_id: *cluster_id,
+									era_id,
+								}
+							})?;
+							return Ok(Some((era_id, batch_index)));
+						} else {
+							return Err(OCWError::EmptyCustomerActivity {
+								cluster_id: *cluster_id,
+								era_id,
+							});
+						}
+					} /*else {
+						 // todo! no data - what to do?
+					 }*/
+				}
+			}
+			Ok(None)
 		}
 
 		pub(crate) fn derive_key(cluster_id: &ClusterId, era_id: DdcEra) -> Vec<u8> {
@@ -714,7 +841,6 @@ pub mod pallet {
 			sp_io::offchain::local_storage_set(StorageKind::PERSISTENT, &key, &encoded_tuple);
 		}
 
-		#[allow(dead_code)]
 		#[allow(clippy::type_complexity)]
 		pub(crate) fn fetch_validation_activities<A: Decode, B: Decode>(
 			// todo! (4) add tests
@@ -845,7 +971,7 @@ pub mod pallet {
 				.map_err(|_| Error::<T>::FailToVerifyMerkleProof)
 		}
 
-		#[allow(dead_code)]
+		// todo! simplify method by removing start/end from the result
 		pub(crate) fn get_era_for_payout(
 			cluster_id: &ClusterId,
 			status: EraValidationStatus,
@@ -1438,6 +1564,27 @@ pub mod pallet {
 							validator: caller.clone(),
 						});
 					},
+					OCWError::BeginChargingCustomersTransactionError { cluster_id, era_id } => {
+						Self::deposit_event(Event::BeginChargingCustomersTransactionError {
+							cluster_id,
+							era_id,
+							validator: caller.clone(),
+						});
+					},
+					OCWError::EmptyCustomerActivity { cluster_id, era_id } => {
+						Self::deposit_event(Event::EmptyCustomerActivity {
+							cluster_id,
+							era_id,
+							validator: caller.clone(),
+						});
+					},
+					OCWError::BatchIndexConversionFailed { cluster_id, era_id } => {
+						Self::deposit_event(Event::BatchIndexConversionFailed {
+							cluster_id,
+							era_id,
+							validator: caller.clone(),
+						});
+					},
 					OCWError::NoAvailableSigner => {
 						Self::deposit_event(Event::NoAvailableSigner { validator: caller.clone() });
 					},
@@ -1482,20 +1629,173 @@ pub mod pallet {
 
 		#[pallet::call_index(5)]
 		#[pallet::weight(<T as pallet::Config>::WeightInfo::create_billing_reports())] // todo! implement weights
-		pub fn set_cluster_to_validate(
+		pub fn begin_billing_report(
 			origin: OriginFor<T>,
 			cluster_id: ClusterId,
+			era_id: DdcEra,
+			start_era: i64,
+			end_era: i64,
 		) -> DispatchResult {
-			ensure_root(origin)?;
-			ClusterToValidate::<T>::put(cluster_id);
+			let sender = ensure_signed(origin)?;
+			ensure!(Self::is_ocw_validator(sender.clone()), Error::<T>::Unauthorised);
+			T::PayoutVisitor::begin_billing_report(sender, cluster_id, era_id, start_era, end_era)?;
+
+			let mut era_validation = <EraValidations<T>>::get(cluster_id, era_id).unwrap(); // should exist
+			era_validation.status = EraValidationStatus::PayoutInProgress;
+			<EraValidations<T>>::insert(cluster_id, era_id, era_validation);
+
 			Ok(())
 		}
 
 		#[pallet::call_index(6)]
 		#[pallet::weight(<T as pallet::Config>::WeightInfo::create_billing_reports())] // todo! implement weights
+		pub fn begin_charging_customers(
+			origin: OriginFor<T>,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			max_batch_index: BatchIndex,
+		) -> DispatchResult {
+			let sender = ensure_signed(origin)?;
+			ensure!(Self::is_ocw_validator(sender.clone()), Error::<T>::Unauthorised);
+			T::PayoutVisitor::begin_charging_customers(sender, cluster_id, era_id, max_batch_index)
+		}
+
+		#[pallet::call_index(7)]
+		#[pallet::weight(<T as pallet::Config>::WeightInfo::create_billing_reports())] // todo! implement weights
+		// todo! remove clippy::too_many_arguments
+		#[allow(clippy::too_many_arguments)]
+		pub fn send_charging_customers_batch(
+			origin: OriginFor<T>,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			batch_index: BatchIndex,
+			payers: Vec<(T::AccountId, BucketId, CustomerUsage)>,
+			mmr_size: u64,
+			proof: Vec<ActivityHash>,
+			leaf_with_position: (u64, ActivityHash),
+		) -> DispatchResult {
+			let sender = ensure_signed(origin)?;
+			ensure!(Self::is_ocw_validator(sender.clone()), Error::<T>::Unauthorised);
+			T::PayoutVisitor::send_charging_customers_batch(
+				sender,
+				cluster_id,
+				era_id,
+				batch_index,
+				payers,
+				mmr_size,
+				proof,
+				leaf_with_position,
+			)
+		}
+
+		#[pallet::call_index(8)]
+		#[pallet::weight(<T as pallet::Config>::WeightInfo::create_billing_reports())] // todo! implement weights
+		pub fn end_charging_customers(
+			origin: OriginFor<T>,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+		) -> DispatchResult {
+			let sender = ensure_signed(origin)?;
+			ensure!(Self::is_ocw_validator(sender.clone()), Error::<T>::Unauthorised);
+			T::PayoutVisitor::end_charging_customers(sender, cluster_id, era_id)
+		}
+
+		#[pallet::call_index(9)]
+		#[pallet::weight(<T as pallet::Config>::WeightInfo::create_billing_reports())] // todo! implement weights
+		pub fn begin_rewarding_providers(
+			origin: OriginFor<T>,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			max_batch_index: BatchIndex,
+			total_node_usage: NodeUsage,
+		) -> DispatchResult {
+			let sender = ensure_signed(origin)?;
+			ensure!(Self::is_ocw_validator(sender.clone()), Error::<T>::Unauthorised);
+			T::PayoutVisitor::begin_rewarding_providers(
+				sender,
+				cluster_id,
+				era_id,
+				max_batch_index,
+				total_node_usage,
+			)
+		}
+
+		#[pallet::call_index(10)]
+		#[pallet::weight(<T as pallet::Config>::WeightInfo::create_billing_reports())] // todo! implement weights
+		// todo! remove clippy::too_many_arguments
+		#[allow(clippy::too_many_arguments)]
+		pub fn send_rewarding_providers_batch(
+			origin: OriginFor<T>,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+			batch_index: BatchIndex,
+			payees: Vec<(T::AccountId, BucketId, NodeUsage)>,
+			mmr_size: u64,
+			proof: Vec<ActivityHash>,
+			leaf_with_position: (u64, ActivityHash),
+		) -> DispatchResult {
+			let sender = ensure_signed(origin)?;
+			ensure!(Self::is_ocw_validator(sender.clone()), Error::<T>::Unauthorised);
+			T::PayoutVisitor::send_rewarding_providers_batch(
+				sender,
+				cluster_id,
+				era_id,
+				batch_index,
+				payees,
+				mmr_size,
+				proof,
+				leaf_with_position,
+			)
+		}
+
+		#[pallet::call_index(11)]
+		#[pallet::weight(<T as pallet::Config>::WeightInfo::create_billing_reports())] // todo! implement weights
+		pub fn end_rewarding_providers(
+			origin: OriginFor<T>,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+		) -> DispatchResult {
+			let sender = ensure_signed(origin)?;
+			ensure!(Self::is_ocw_validator(sender.clone()), Error::<T>::Unauthorised);
+			T::PayoutVisitor::end_rewarding_providers(sender, cluster_id, era_id)
+		}
+
+		#[pallet::call_index(12)]
+		#[pallet::weight(<T as pallet::Config>::WeightInfo::create_billing_reports())] // todo! implement weights
+		pub fn end_billing_report(
+			origin: OriginFor<T>,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
+		) -> DispatchResult {
+			let sender = ensure_signed(origin)?;
+			ensure!(Self::is_ocw_validator(sender.clone()), Error::<T>::Unauthorised);
+			T::PayoutVisitor::end_billing_report(sender, cluster_id, era_id)?;
+
+			let mut era_validation = <EraValidations<T>>::get(cluster_id, era_id).unwrap(); // should exist
+			era_validation.status = EraValidationStatus::PayoutSuccess;
+			<EraValidations<T>>::insert(cluster_id, era_id, era_validation);
+
+			Ok(())
+		}
+
+		#[pallet::call_index(13)]
+		#[pallet::weight(<T as pallet::Config>::WeightInfo::create_billing_reports())] // todo! implement weights
 		pub fn set_current_validator(origin: OriginFor<T>) -> DispatchResult {
 			let validator: T::AccountId = ensure_signed(origin)?;
 			CurrentValidator::<T>::put(validator);
+
+			Ok(())
+		}
+
+		#[pallet::call_index(14)]
+		#[pallet::weight(<T as pallet::Config>::WeightInfo::create_billing_reports())] // todo! implement weights
+		pub fn set_cluster_to_validate(
+			origin: OriginFor<T>,
+			cluster_id: ClusterId,
+		) -> DispatchResult {
+			ensure_root(origin)?;
+			ClusterToValidate::<T>::put(cluster_id);
+
 			Ok(())
 		}
 	}
@@ -1512,33 +1812,44 @@ pub mod pallet {
 			}
 		}
 
+		// todo! use batch_index and payers as part of the validation
 		fn is_customers_batch_valid(
 			cluster_id: ClusterId,
-			era: DdcEra,
+			era_id: DdcEra,
 			_batch_index: BatchIndex,
 			_payers: &[(T::AccountId, BucketId, CustomerUsage)],
 			proof: MerkleProof<ActivityHash, MergeActivityHash>,
 			leaf_with_position: (u64, ActivityHash),
 		) -> bool {
-			let validation_era = EraValidations::<T>::get(cluster_id, era);
+			let validation_era = EraValidations::<T>::get(cluster_id, era_id);
 
 			match validation_era {
 				Some(valid_era) => {
 					let root = valid_era.payers_merkle_root_hash;
-
 					Self::proof_merkle_leaf(root, proof, vec![leaf_with_position]).unwrap_or(false)
 				},
 				None => false,
 			}
 		}
+
+		// todo! use batch_index and payees as part of the validation
 		fn is_providers_batch_valid(
-			_cluster_id: ClusterId,
-			_era: DdcEra,
+			cluster_id: ClusterId,
+			era_id: DdcEra,
 			_batch_index: BatchIndex,
 			_payees: &[(T::AccountId, BucketId, NodeUsage)],
-			_adjacent_hashes: &[ActivityHash],
+			proof: MerkleProof<ActivityHash, MergeActivityHash>,
+			leaf_with_position: (u64, ActivityHash),
 		) -> bool {
-			true
+			let validation_era = EraValidations::<T>::get(cluster_id, era_id);
+
+			match validation_era {
+				Some(valid_era) => {
+					let root = valid_era.payees_merkle_root_hash;
+					Self::proof_merkle_leaf(root, proof, vec![leaf_with_position]).unwrap_or(false)
+				},
+				None => false,
+			}
 		}
 	}
 
diff --git a/pallets/ddc-verification/src/mock.rs b/pallets/ddc-verification/src/mock.rs
index c6ca52685..01a6a2ff8 100644
--- a/pallets/ddc-verification/src/mock.rs
+++ b/pallets/ddc-verification/src/mock.rs
@@ -1,8 +1,8 @@
 use ddc_primitives::{
 	crypto, sr25519,
 	traits::{ClusterManager, ClusterQuery},
-	ClusterNodeKind, ClusterNodeState, ClusterNodeStatus, ClusterNodesStats, ClusterStatus,
-	PayoutState, StorageNodePubKey, MAX_PAYOUT_BATCH_COUNT, MAX_PAYOUT_BATCH_SIZE,
+	BucketId, ClusterNodeKind, ClusterNodeState, ClusterNodeStatus, ClusterNodesStats,
+	ClusterStatus, PayoutState, StorageNodePubKey, MAX_PAYOUT_BATCH_COUNT, MAX_PAYOUT_BATCH_SIZE,
 };
 use frame_election_provider_support::{
 	bounds::{ElectionBounds, ElectionBoundsBuilder},
@@ -311,9 +311,88 @@ pub fn new_test_ext() -> sp_io::TestExternalities {
 
 pub struct MockPayoutVisitor;
 impl<T: Config> PayoutVisitor<T> for MockPayoutVisitor {
-	fn get_billing_report_status(_cluster_id: ClusterId, _era: DdcEra) -> PayoutState {
+	fn get_billing_report_status(_cluster_id: &ClusterId, _era_id: DdcEra) -> PayoutState {
 		PayoutState::NotInitialized
 	}
+
+	fn begin_billing_report(
+		_origin: T::AccountId,
+		_cluster_id: ClusterId,
+		_era_id: DdcEra,
+		_start_era: i64,
+		_end_era: i64,
+	) -> DispatchResult {
+		Ok(())
+	}
+
+	fn begin_charging_customers(
+		_origin: T::AccountId,
+		_cluster_id: ClusterId,
+		_era_id: DdcEra,
+		_max_batch_index: BatchIndex,
+	) -> DispatchResult {
+		Ok(())
+	}
+
+	fn send_charging_customers_batch(
+		_origin: T::AccountId,
+		_cluster_id: ClusterId,
+		_era_id: DdcEra,
+		_batch_index: BatchIndex,
+		_payers: Vec<(T::AccountId, BucketId, CustomerUsage)>,
+		_mmr_size: u64,
+		_proof: Vec<ActivityHash>,
+		_leaf_with_position: (u64, ActivityHash),
+	) -> DispatchResult {
+		Ok(())
+	}
+
+	fn end_charging_customers(
+		_origin: T::AccountId,
+		_cluster_id: ClusterId,
+		_era_id: DdcEra,
+	) -> DispatchResult {
+		Ok(())
+	}
+
+	fn begin_rewarding_providers(
+		_origin: T::AccountId,
+		_cluster_id: ClusterId,
+		_era_id: DdcEra,
+		_max_batch_index: BatchIndex,
+		_total_node_usage: NodeUsage,
+	) -> DispatchResult {
+		Ok(())
+	}
+
+	fn send_rewarding_providers_batch(
+		_origin: T::AccountId,
+		_cluster_id: ClusterId,
+		_era_id: DdcEra,
+		_batch_index: BatchIndex,
+		_payees: Vec<(T::AccountId, BucketId, NodeUsage)>,
+		_mmr_size: u64,
+		_proof: Vec<ActivityHash>,
+		_leaf_with_position: (u64, ActivityHash),
+	) -> DispatchResult {
+		Ok(())
+	}
+
+	fn end_rewarding_providers(
+		_origin: T::AccountId,
+		_cluster_id: ClusterId,
+		_era_id: DdcEra,
+	) -> DispatchResult {
+		Ok(())
+	}
+
+	fn end_billing_report(
+		_origin: T::AccountId,
+		_cluster_id: ClusterId,
+		_era_id: DdcEra,
+	) -> DispatchResult {
+		Ok(())
+	}
 }
 
 pub struct MockNodeVisitor;
diff --git a/primitives/src/traits/payout.rs b/primitives/src/traits/payout.rs
index 361df6ad8..50763709c 100644
--- a/primitives/src/traits/payout.rs
+++ b/primitives/src/traits/payout.rs
@@ -1,5 +1,87 @@
-use crate::{ClusterId, DdcEra, PayoutState};
+use scale_info::prelude::vec::Vec;
+use sp_runtime::DispatchResult;
+
+use crate::{
+	ActivityHash, BatchIndex, BucketId, ClusterId, CustomerUsage, DdcEra, NodeUsage, PayoutState,
+};
+
+pub trait PayoutProcessor<T: frame_system::Config> {}
 
 pub trait PayoutVisitor<T: frame_system::Config> {
-	fn get_billing_report_status(cluster_id: ClusterId, era: DdcEra) -> PayoutState;
+	// todo! factor out into PayoutProcessor
+	fn begin_billing_report(
+		origin: T::AccountId,
+		cluster_id: ClusterId,
+		era_id: DdcEra,
+		start_era: i64,
+		end_era: i64,
+	) -> DispatchResult;
+
+	// todo! factor out into PayoutProcessor
+	fn begin_charging_customers(
+		origin: T::AccountId,
+		cluster_id: ClusterId,
+		era_id: DdcEra,
+		max_batch_index: BatchIndex,
+	) -> DispatchResult;
+
+	// todo! factor out into PayoutProcessor
+	// todo! remove clippy::too_many_arguments
+	#[allow(clippy::too_many_arguments)]
+	fn send_charging_customers_batch(
+		origin: T::AccountId,
+		cluster_id: ClusterId,
+		era_id: DdcEra,
+		batch_index: BatchIndex,
+		payers: Vec<(T::AccountId, BucketId, CustomerUsage)>,
+		mmr_size: u64,
+		proof: Vec<ActivityHash>,
+		leaf_with_position: (u64, ActivityHash),
+	) -> DispatchResult;
+
+	// todo! factor out into PayoutProcessor
+	fn end_charging_customers(
+		origin: T::AccountId,
+		cluster_id: ClusterId,
+		era_id: DdcEra,
+	) -> DispatchResult;
+
+	// todo! factor out into PayoutProcessor
+	fn begin_rewarding_providers(
+		origin: T::AccountId,
+		cluster_id: ClusterId,
+		era_id: DdcEra,
+		max_batch_index: BatchIndex,
+		total_node_usage: NodeUsage,
+	) -> DispatchResult;
+
+	// todo! factor out into PayoutProcessor
+	// todo! remove clippy::too_many_arguments
+	#[allow(clippy::too_many_arguments)]
+	fn send_rewarding_providers_batch(
+		origin: T::AccountId,
+		cluster_id: ClusterId,
+		era_id: DdcEra,
+		batch_index: BatchIndex,
+		payees: Vec<(T::AccountId, BucketId, NodeUsage)>,
+		mmr_size: u64,
+		proof: Vec<ActivityHash>,
+		leaf_with_position: (u64, ActivityHash),
+	) -> DispatchResult;
+
+	// todo! factor out into PayoutProcessor
+	fn end_rewarding_providers(
+		origin: T::AccountId,
+		cluster_id: ClusterId,
+		era_id: DdcEra,
+	) -> DispatchResult;
+
+	// todo! factor out into PayoutProcessor
+	fn end_billing_report(
+		origin: T::AccountId,
+		cluster_id: ClusterId,
+		era_id: DdcEra,
+	) -> DispatchResult;
+
+	fn get_billing_report_status(cluster_id: &ClusterId, era: DdcEra) -> PayoutState;
 }
diff --git a/primitives/src/traits/validator.rs b/primitives/src/traits/validator.rs
index e61def68e..64522fcc1 100644
--- a/primitives/src/traits/validator.rs
+++ b/primitives/src/traits/validator.rs
@@ -23,6 +23,7 @@ pub trait ValidatorVisitor<T: Config> {
 		era: DdcEra,
 		batch_index: BatchIndex,
 		payees: &[(T::AccountId, BucketId, NodeUsage)],
-		adjacent_hashes: &[ActivityHash],
+		proof: MerkleProof<ActivityHash, MergeActivityHash>,
+		leaf_with_position: (u64, ActivityHash),
 	) -> bool;
 }