Do not refresh stuff that is not event-sourced anymore

api/src/domain/projectors/mod.rs

@@ -1,3 +1,4 @@
 pub mod event_store;
 pub mod new_indexer;
 pub mod projections;
+pub mod projections_for_refresh;

api/src/domain/projectors/projections_for_refresh.rs

@@ -0,0 +1,189 @@
+use std::{convert::TryFrom, sync::Arc};
+
+use anyhow::Result;
+use async_trait::async_trait;
+use derive_new::new;
+use domain::{
+	BudgetEvent, Event, EventListener, PaymentEvent, PaymentWorkItem, ProjectEvent,
+	SubscriberCallbackError,
+};
+use infrastructure::dbclient::{ImmutableRepository, Repository};
+use olog::IntoField;
+use rust_decimal::Decimal;
+use tracing::instrument;
+
+use crate::models::*;
+
+#[allow(clippy::too_many_arguments)]
+#[derive(new, Clone)]
+pub struct Projector {
+	project_budgets_repository: Arc<dyn ImmutableRepository<ProjectsBudget>>,
+	budget_repository: Arc<dyn Repository<Budget>>,
+	payment_request_repository: Arc<dyn Repository<PaymentRequest>>,
+	payment_repository: Arc<dyn Repository<Payment>>,
+	work_item_repository: Arc<dyn WorkItemRepository>,
+	projects_rewarded_users_repository: Arc<dyn ProjectsRewardedUserRepository>,
+	// TODO: replace the repositories below by API call to indexer in another projector
+	github_repo_index_repository: Arc<dyn GithubRepoIndexRepository>,
+	github_user_index_repository: Arc<dyn GithubUserIndexRepository>,
+}
+
+#[async_trait]
+impl EventListener<Event> for Projector {
+	#[instrument(name = "project_projection", skip(self))]
+	async fn on_event(&self, event: Event) -> Result<(), SubscriberCallbackError> {
+		match event {
+			Event::Budget(event) => match event {
+				BudgetEvent::Created { id, currency } => {
+					self.budget_repository.upsert(Budget {
+						id,
+						initial_amount: Decimal::ZERO,
+						remaining_amount: Decimal::ZERO,
+						currency: currency.try_into()?,
+					})?;
+				},
+				BudgetEvent::Allocated { id, amount, .. } => {
+					let mut budget = self.budget_repository.find_by_id(id)?;
+					budget.remaining_amount += amount;
+					budget.initial_amount += amount;
+					self.budget_repository.update(budget)?;
+				},
+				BudgetEvent::Spent { id, amount } => {
+					let mut budget = self.budget_repository.find_by_id(id)?;
+					budget.remaining_amount -= amount;
+					self.budget_repository.update(budget)?;
+				},
+			},
+			Event::Payment(event) => match event {
+				PaymentEvent::Requested {
+					id: payment_id,
+					project_id,
+					requestor_id,
+					recipient_id,
+					amount,
+					reason,
+					duration_worked,
+					requested_at,
+				} => {
+					self.payment_request_repository
+						.upsert(PaymentRequest {
+							id: payment_id,
+							project_id,
+							requestor_id,
+							recipient_id,
+							amount: *amount.amount(),
+							currency: amount.currency().try_into()?,
+							requested_at,
+							invoice_received_at: None,
+							hours_worked: duration_worked
+								.and_then(|duration_worked| {
+									i32::try_from(duration_worked.num_hours()).ok()
+								})
+								.unwrap_or(0),
+						})
+						.map_err(|e| {
+							olog::error!(error = e.to_field(), "payment_request_repository.upsert");
+							e
+						})?;
+
+					reason.work_items.into_iter().try_for_each(
+						|work_item| -> Result<(), SubscriberCallbackError> {
+							let repo_id = match work_item {
+								PaymentWorkItem::Issue { repo_id, .. }
+								| PaymentWorkItem::CodeReview { repo_id, .. }
+								| PaymentWorkItem::PullRequest { repo_id, .. } => repo_id,
+							};
+
+							self.work_item_repository
+								.try_insert(
+									(project_id, payment_id, recipient_id, work_item).into(),
+								)
+								.map_err(|e| {
+									olog::error!(
+										error = e.to_field(),
+										"error work_item_repository.try_insert"
+									);
+									e
+								})?;
+
+							self.github_repo_index_repository.start_indexing(repo_id).map_err(
+								|e| {
+									olog::error!(
+										error = e.to_field(),
+										"github_repo_index_repository.start_indexing"
+									);
+									e
+								},
+							)?;
+							Ok(())
+						},
+					)?;
+
+					self.github_user_index_repository
+						.try_insert(GithubUserIndex {
+							user_id: recipient_id,
+							..Default::default()
+						})
+						.map_err(|e| {
+							olog::error!(
+								error = e.to_field(),
+								"github_user_index_repository.try_insert"
+							);
+							e
+						})?;
+
+					self.projects_rewarded_users_repository
+						.increase_user_reward_count_for_project(&project_id, &recipient_id)
+						.map_err(|e| {
+							olog::error!(
+								error = e.to_field(),
+								"increase_user_reward_count_for_project"
+							);
+							e
+						})?;
+				},
+				PaymentEvent::Cancelled { id: payment_id } => {
+					let payment_request = self.payment_request_repository.find_by_id(payment_id)?;
+					self.payment_request_repository.delete(payment_id)?;
+					self.work_item_repository.delete_by_payment_id(payment_id)?;
+
+					self.projects_rewarded_users_repository
+						.decrease_user_reward_count_for_project(
+							&payment_request.project_id,
+							&payment_request.recipient_id,
+						)?;
+				},
+				PaymentEvent::Processed {
+					id: payment_id,
+					receipt_id,
+					amount,
+					receipt,
+					processed_at,
+				} => {
+					self.payment_repository.upsert(Payment {
+						id: receipt_id,
+						amount: *amount.amount(),
+						currency_code: amount.currency().to_string(),
+						receipt: serde_json::to_value(receipt)
+							.map_err(|e| SubscriberCallbackError::Discard(e.into()))?,
+						request_id: payment_id,
+						processed_at,
+					})?;
+				},
+				_ => (),
+			},
+			Event::Project(event) => match event {
+				ProjectEvent::BudgetLinked { id, budget_id, .. } => {
+					self.project_budgets_repository.try_insert(ProjectsBudget {
+						project_id: id,
+						budget_id,
+					})?;
+				},
+				_ => (),
+			},
+			_ => (),
+		}
+
+		Ok(())
+	}
+}

api/src/presentation/http/bootstrap.rs

@@ -9,7 +9,7 @@
 use rocket::{Build, Rocket};
 
 use crate::{
-	domain::projectors::{self, new_indexer, projections},
+	domain::projectors::{self, new_indexer, projections, projections_for_refresh},
 	infrastructure::{simple_storage, web3::ens},
 	presentation::{graphql, http, http::github_client_pat_factory::GithubClientPatFactory},
 	Config,
@@ -46,6 +46,16 @@
 		database.clone(),
 		database.clone(),
 	);
+	let projector_for_refresh = projections_for_refresh::Projector::new(
+		database.clone(),
+		database.clone(),
+		database.clone(),
+		database.clone(),
+		database.clone(),
+		database.clone(),
+		database.clone(),
+		database.clone(),
+	);
 
 	let event_publisher = CompositePublisher::new(vec![
 		Arc::new(EventPublisher::new(
@@ -91,6 +101,7 @@
 		database.clone(),
 		database,
 		projector,
+		projector_for_refresh,
 	);
 	Ok(rocket_build)
 }

api/src/presentation/http/mod.rs

@@ -12,7 +12,7 @@
 
 use crate::{
 	application,
-	domain::{projectors::projections::Projector, DustyBotService, ImageStoreService},
+	domain::{DustyBotService, ImageStoreService},
 	infrastructure::web3::ens,
 	models::*,
 	presentation::{graphql, http::github_client_pat_factory::GithubClientPatFactory},
@@ -58,7 +58,8 @@
 	application_event_store: Arc<dyn EventStore<Application>>,
 	budget_event_store: Arc<dyn EventStore<Budget>>,
 	payment_event_store: Arc<dyn EventStore<Payment>>,
-	projector: Projector,
+	projector: crate::domain::projectors::projections::Projector,
+	projector_for_refresh: crate::domain::projectors::projections_for_refresh::Projector,
 ) -> Rocket<Build> {
 	let update_user_profile_info_usecase = application::user::update_profile_info::Usecase::new(
 		user_profile_info_repository.clone(),
@@ -103,6 +104,7 @@
 		.manage(budget_event_store)
 		.manage(payment_event_store)
 		.manage(projector)
+		.manage(projector_for_refresh)
 		.attach(http::guards::Cors)
 		.mount(
 			"/",

api/src/presentation/http/routes/internal/events/refresh.rs

@@ -16,7 +16,7 @@
 use thiserror::Error;
 use uuid::Uuid;
 
-use crate::domain::projectors::projections;
+use crate::domain::projectors::projections_for_refresh;
 
 #[derive(Debug, Error)]
 enum Error {
@@ -57,11 +57,11 @@
 	budget_event_store: &State<Arc<dyn EventStore<Budget>>>,
 	application_event_store: &State<Arc<dyn EventStore<Application>>>,
 	payment_event_store: &State<Arc<dyn EventStore<Payment>>>,
-	projector: &State<projections::Projector>,
+	projector_for_refresh: &State<projections_for_refresh::Projector>,
 ) -> Result<(), HttpApiProblem> {
 	let mut registry = Registry::new();
 
-	let projector = Arc::new((*projector).clone());
+	let projector = Arc::new((*projector_for_refresh).clone());
 
 	Refresher::new((*project_event_store).clone(), projector.clone()).register(&mut registry)?;
 	Refresher::new((*application_event_store).clone(), projector.clone())