feat: init cqrs executors and queries for billing views

src/billing/adapters/output/db/postgres/bill_view.rs

@@ -3,16 +3,20 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 use std::str::FromStr;
+use std::sync::Arc;
 
 use async_trait::async_trait;
 use cqrs_es::persist::{PersistenceError, ViewContext, ViewRepository};
 use cqrs_es::{EventEnvelope, Query, View};
+use postgres_es::PostgresCqrs;
 use serde::{Deserialize, Serialize};
 use time::OffsetDateTime;
 use uuid::Uuid;
 
 use super::errors::*;
 use super::BillingDBPostgresAdapter;
+use crate::billing::adapters::types::{BillingBillCqrsExec, BillingBillCqrsView};
+use crate::billing::application::services::BillingServicesObj;
 use crate::billing::domain::bill_aggregate::{Bill, BillBuilder};
 use crate::billing::domain::events::BillingEvent;
 use crate::types::currency::{self, Currency, PriceBuilder};
@@ -311,12 +315,24 @@ impl Query<Bill> for BillingDBPostgresAdapter {
     }
 }
 
+pub fn init_cqrs(
+    db: BillingDBPostgresAdapter,
+    services: BillingServicesObj,
+) -> (BillingBillCqrsExec, BillingBillCqrsView) {
+    let queries: Vec<Box<dyn Query<Bill>>> = vec![Box::new(db.clone())];
+
+    let pool = db.pool.clone();
+
+    (
+        Arc::new(postgres_es::postgres_cqrs(pool.clone(), queries, services)),
+        Arc::new(db.clone()),
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
-    use postgres_es::PostgresCqrs;
-
     use crate::{
         billing::{
             application::services::{

src/billing/adapters/output/db/postgres/line_item_view.rs

@@ -1,8 +1,8 @@
 // SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
 //
 // SPDX-License-Identifier: AGPL-3.0-or-later
-
 use std::str::FromStr;
+use std::sync::Arc;
 
 use async_trait::async_trait;
 use cqrs_es::persist::{PersistenceError, ViewContext, ViewRepository};
@@ -13,6 +13,8 @@ use uuid::Uuid;
 
 use super::errors::*;
 use super::BillingDBPostgresAdapter;
+use crate::billing::adapters::types::{BillingLineItemCqrsExec, BillingLineItemCqrsView};
+use crate::billing::application::services::BillingServicesObj;
 use crate::billing::domain::events::BillingEvent;
 use crate::billing::domain::line_item_aggregate::*;
 use crate::types::currency::*;
@@ -363,6 +365,20 @@ impl Query<LineItem> for BillingDBPostgresAdapter {
     }
 }
 
+pub fn init_cqrs(
+    db: BillingDBPostgresAdapter,
+    services: BillingServicesObj,
+) -> (BillingLineItemCqrsExec, BillingLineItemCqrsView) {
+    let queries: Vec<Box<dyn Query<LineItem>>> = vec![Box::new(db.clone())];
+
+    let pool = db.pool.clone();
+
+    (
+        Arc::new(postgres_es::postgres_cqrs(pool.clone(), queries, services)),
+        Arc::new(db.clone()),
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/billing/adapters/output/db/postgres/mod.rs

@@ -8,19 +8,19 @@ use sqlx::postgres::PgPool;
 use crate::db::{migrate::RunMigrations, sqlx_postgres::Postgres};
 
 mod bill_id_exists;
-mod bill_view;
+pub(crate) mod bill_view;
 mod errors;
 mod get_line_items_for_bill_id;
 mod line_item_id_exists;
-mod line_item_view;
+pub(crate) mod line_item_view;
 mod next_token_id;
 mod store_id_exists;
 mod store_name_exists;
-mod store_view;
+pub(crate) mod store_view;
 
 #[derive(Clone)]
 pub struct BillingDBPostgresAdapter {
-    pool: PgPool,
+    pub(crate) pool: PgPool,
 }
 
 impl BillingDBPostgresAdapter {

src/billing/adapters/output/db/postgres/store_view.rs

@@ -1,6 +1,7 @@
 // SPDX-FileCopyrightText: 2024 Aravinth Manivannan <realaravinth@batsense.net>
 //
 // SPDX-License-Identifier: AGPL-3.0-or-later
+use std::sync::Arc;
 
 use async_trait::async_trait;
 use cqrs_es::persist::{PersistenceError, ViewContext, ViewRepository};
@@ -10,6 +11,8 @@ use uuid::Uuid;
 
 use super::errors::*;
 use super::BillingDBPostgresAdapter;
+use crate::billing::adapters::types::{BillingStoreCqrsExec, BillingStoreCqrsView};
+use crate::billing::application::services::BillingServicesObj;
 use crate::billing::domain::events::BillingEvent;
 use crate::billing::domain::store_aggregate::*;
 use crate::utils::parse_aggregate_id::parse_aggregate_id;
@@ -216,6 +219,20 @@ impl Query<Store> for BillingDBPostgresAdapter {
     }
 }
 
+pub fn init_cqrs(
+    db: BillingDBPostgresAdapter,
+    services: BillingServicesObj,
+) -> (BillingStoreCqrsExec, BillingStoreCqrsView) {
+    let queries: Vec<Box<dyn Query<Store>>> = vec![Box::new(db.clone())];
+
+    let pool = db.pool.clone();
+
+    (
+        Arc::new(postgres_es::postgres_cqrs(pool.clone(), queries, services)),
+        Arc::new(db.clone()),
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;