begin billing -> end billing (without batches) (#378)

## Description
<!-- Describe what change this PR is implementing -->

## Types of Changes
Please select the branch type you are merging and fill in the relevant
template.
<!--- Check the following box with an x if the following applies: -->
- [ ] Hotfix
- [ ] Release
- [ ] Fix or Feature

## Fix or Feature
<!--- Check the following box with an x if the following applies: -->

### Types of Changes
<!--- What types of changes does your code introduce? -->
- [ ] Tech Debt (Code improvements)
- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Dependency upgrade (A change in substrate or any 3rd party crate
version)

### Migrations and Hooks
<!--- Check the following box with an x if the following applies: -->
- [ ] This change requires a runtime migration.
- [ ] Modifies `on_initialize`
- [ ] Modifies `on_finalize`

### Checklist for Fix or Feature
<!--- All boxes need to be checked. Follow this checklist before
requiring PR review -->
- [ ] Change has been tested locally.
- [ ] Change adds / updates tests if applicable.
- [ ] Changelog doc updated.
- [ ] `spec_version` has been incremented.
- [ ] `network-relayer`'s
[events](https://github.com/Cerebellum-Network/network-relayer/blob/dev-cere/shared/substrate/events.go)
have been updated according to the blockchain events if applicable.
- [ ] All CI checks have been passed successfully

## Checklist for Hotfix
<!--- All boxes need to be checked. Follow this checklist before
requiring PR review -->
- [ ] Changelog has been updated.
- [ ] Crate version has been updated.
- [ ] `spec_version` has been incremented.
- [ ] Transaction version has been updated if required.
- [ ] Pull Request to `dev` has been created.
- [ ] Pull Request to `staging` has been created.
- [ ] `network-relayer`'s
[events](https://github.com/Cerebellum-Network/network-relayer/blob/dev-cere/shared/substrate/events.go)
have been updated according to the blockchain events if applicable.
- [ ] All CI checks have been passed successfully

## Checklist for Release
<!--- All boxes need to be checked. Follow this checklist before
requiring PR review -->
- [ ] Change has been deployed to Devnet.
- [ ] Change has been tested in Devnet.
- [ ] Change has been deployed to Qanet.
- [ ] Change has been tested in Qanet.
- [ ] Change has been deployed to Testnet.
- [ ] Change has been tested in Testnet.
- [ ] Changelog has been updated.
- [ ] Crate version has been updated.
- [ ] Spec version has been updated.
- [ ] Transaction version has been updated if required.
- [ ] All CI checks have been passed successfully

pallets/ddc-payouts/src/lib.rs

@@ -36,7 +36,7 @@ use ddc_primitives::{
 		pallet::PalletVisitor as PalletVisitorType,
 		payout::PayoutVisitor,
 	},
-	BatchIndex, BucketId, ClusterId, CustomerUsage, DdcEra, NodeUsage, PayoutState,
+	BatchIndex, BucketId, ClusterId, CustomerUsage, DdcEra, NodeUsage, PayoutError, PayoutState,
 	MAX_PAYOUT_BATCH_COUNT, MAX_PAYOUT_BATCH_SIZE, MILLICENTS,
 };
 use frame_election_provider_support::SortedListProvider;
@@ -570,7 +570,7 @@ pub mod pallet {
 				billing_report.state == PayoutState::ChargingCustomers,
 				Error::<T>::NotExpectedState
 			);
-			validate_batches::<T>(
+			Self::validate_batches(
 				&billing_report.charging_processed_batches,
 				&billing_report.charging_max_batch_index,
 			)?;
@@ -841,7 +841,7 @@ pub mod pallet {
 				Error::<T>::NotExpectedState
 			);
 
-			validate_batches::<T>(
+			Self::validate_batches(
 				&billing_report.rewarding_processed_batches,
 				&billing_report.rewarding_max_batch_index,
 			)?;
@@ -1050,22 +1050,6 @@ pub mod pallet {
 		Ok(total)
 	}
 
-	fn validate_batches<T: Config>(
-		batches: &BoundedBTreeSet<BatchIndex, MaxBatchesCount>,
-		max_batch_index: &BatchIndex,
-	) -> DispatchResult {
-		// Check if the Vec contains all integers between 1 and rewarding_max_batch_index
-		ensure!(!batches.is_empty(), Error::<T>::BatchesMissed);
-
-		ensure!((*max_batch_index + 1) as usize == batches.len(), Error::<T>::BatchesMissed);
-
-		for index in 0..*max_batch_index + 1 {
-			ensure!(batches.contains(&index), Error::<T>::BatchesMissed);
-		}
-
-		Ok(())
-	}
-
 	#[pallet::genesis_config]
 	pub struct GenesisConfig<T: Config> {
 		pub feeder_account: Option<T::AccountId>,
@@ -1223,6 +1207,66 @@ pub mod pallet {
 				None => PayoutState::NotInitialized, // Return NotInitialized if entry doesn't exist
 			}
 		}
+
+		fn all_customer_batches_processed(cluster_id: &ClusterId, era_id: DdcEra) -> bool {
+			let billing_report = match ActiveBillingReports::<T>::try_get(cluster_id, era_id) {
+				Ok(report) => report,
+				Err(_) => return false, /* Return false if there's any error (e.g.,
+				                         * BillingReportDoesNotExist) */
+			};
+
+			Self::validate_batches(
+				&billing_report.charging_processed_batches,
+				&billing_report.charging_max_batch_index,
+			)
+			.is_ok()
+		}
+
+		fn all_provider_batches_processed(cluster_id: &ClusterId, era_id: DdcEra) -> bool {
+			let billing_report = match ActiveBillingReports::<T>::try_get(cluster_id, era_id) {
+				Ok(report) => report,
+				Err(_) => return false, /* Return false if there's any error (e.g.,
+				                         * BillingReportDoesNotExist) */
+			};
+
+			Self::validate_batches(
+				&billing_report.rewarding_processed_batches,
+				&billing_report.rewarding_max_batch_index,
+			)
+			.is_ok()
+		}
+
+		fn get_next_customer_batch_for_payment(
+			cluster_id: &ClusterId,
+			era_id: DdcEra,
+		) -> Result<Option<BatchIndex>, PayoutError> {
+			let billing_report = ActiveBillingReports::<T>::try_get(cluster_id, era_id)
+				.map_err(|_| PayoutError::BillingReportDoesNotExist)?;
+
+			for batch_index in 0..=billing_report.charging_max_batch_index {
+				if !billing_report.charging_processed_batches.contains(&batch_index) {
+					return Ok(Some(batch_index));
+				}
+			}
+
+			Ok(None)
+		}
+
+		fn get_next_provider_batch_for_payment(
+			cluster_id: &ClusterId,
+			era_id: DdcEra,
+		) -> Result<Option<BatchIndex>, PayoutError> {
+			let billing_report = ActiveBillingReports::<T>::try_get(cluster_id, era_id)
+				.map_err(|_| PayoutError::BillingReportDoesNotExist)?;
+
+			for batch_index in 0..=billing_report.rewarding_max_batch_index {
+				if !billing_report.rewarding_processed_batches.contains(&batch_index) {
+					return Ok(Some(batch_index));
+				}
+			}
+
+			Ok(None)
+		}
 	}
 
 	impl<T: Config> Pallet<T> {
@@ -1241,5 +1285,21 @@ pub mod pallet {
 			// be fulfilled with trailing zeros.
 			T::PalletId::get().into_sub_account_truncating(hash)
 		}
+
+		pub(crate) fn validate_batches(
+			batches: &BoundedBTreeSet<BatchIndex, MaxBatchesCount>,
+			max_batch_index: &BatchIndex,
+		) -> DispatchResult {
+			// Check if the Vec contains all integers between 1 and rewarding_max_batch_index
+			ensure!(!batches.is_empty(), Error::<T>::BatchesMissed);
+
+			ensure!((*max_batch_index + 1) as usize == batches.len(), Error::<T>::BatchesMissed);
+
+			for index in 0..*max_batch_index + 1 {
+				ensure!(batches.contains(&index), Error::<T>::BatchesMissed);
+			}
+
+			Ok(())
+		}
 	}
 }