mod branch; mod load; pub mod status; pub mod webhook; #[cfg(test)] mod tests; use std::time::Duration; use actix::prelude::*; use crate as repo_actor; use git_next_forge as forge; use git_next_git as git; use kxio::network::Network; use tracing::{debug, info, warn, Instrument}; #[derive(Debug, derive_more::Display)] #[display("{}:{}:{}", generation, details.forge.forge_name(), details.repo_alias)] pub struct RepoActor { generation: git::Generation, message_token: MessageToken, details: git::RepoDetails, webhook: git_next_config::server::Webhook, webhook_id: Option, // INFO: if [None] then no webhook is configured webhook_auth: Option, // INFO: if [None] then no webhook is configured last_main_commit: Option, last_next_commit: Option, last_dev_commit: Option, repository: Option, net: Network, forge: forge::Forge, } impl RepoActor { pub fn new( details: git::RepoDetails, webhook: git_next_config::server::Webhook, generation: git::Generation, net: Network, repo: git::Repository, ) -> Self { let forge = match details.forge.forge_type() { #[cfg(feature = "forgejo")] git_next_config::ForgeType::ForgeJo => { forge::Forge::new_forgejo(details.clone(), net.clone(), repo) } git_next_config::ForgeType::MockForge => forge::Forge::new_mock(), }; debug!(?forge, "new"); Self { generation, message_token: MessageToken::new(), details, webhook, webhook_id: None, webhook_auth: None, last_main_commit: None, last_next_commit: None, last_dev_commit: None, repository: None, net, forge, } } } impl Actor for RepoActor { type Context = Context; #[tracing::instrument(name = "RepoActor::stopping", skip_all, fields(repo = %self.details))] fn stopping(&mut self, ctx: &mut Self::Context) -> Running { info!("Checking webhook"); match self.webhook_id.take() { Some(webhook_id) => { let repo_details = self.details.clone(); let net = self.net.clone(); info!(%webhook_id, "Unregistring webhook"); webhook::unregister(webhook_id, repo_details, net) .in_current_span() .into_actor(self) .wait(ctx); Running::Continue } None => Running::Stop, } } } #[derive(Message)] #[rtype(result = "()")] pub struct CloneRepo; impl Handler for RepoActor { type Result = (); #[tracing::instrument(name = "RepoActor::CloneRepo", skip_all, fields(repo = %self.details, gitdir = %self.details.gitdir))] fn handle(&mut self, _msg: CloneRepo, ctx: &mut Self::Context) -> Self::Result { let gitdir = self.details.gitdir.clone(); match self.forge.repo_clone(gitdir) { Ok(repository) => { self.repository.replace(repository); if self.details.repo_config.is_none() { ctx.address().do_send(LoadConfigFromRepo); } else { ctx.address().do_send(ValidateRepo { message_token: self.message_token, }); } } Err(err) => warn!("Could not Clone repo: {err}"), } } } #[derive(Message)] #[rtype(result = "()")] pub struct LoadConfigFromRepo; impl Handler for RepoActor { type Result = (); #[tracing::instrument(name = "RepoActor::LoadConfigFromRepo", skip_all, fields(repo = %self.details))] fn handle(&mut self, _msg: LoadConfigFromRepo, ctx: &mut Self::Context) -> Self::Result { let details = self.details.clone(); let addr = ctx.address(); let forge = self.forge.clone(); repo_actor::load::load_file(details, addr, forge) .in_current_span() .into_actor(self) .wait(ctx); } } #[derive(Message)] #[rtype(result = "()")] struct LoadedConfig(git_next_config::RepoConfig); impl Handler for RepoActor { type Result = (); #[tracing::instrument(name = "RepoActor::LoadedConfig", skip_all, fields(repo = %self.details, branches = %msg.0))] fn handle(&mut self, msg: LoadedConfig, ctx: &mut Self::Context) -> Self::Result { let repo_config = msg.0; self.details.repo_config.replace(repo_config); ctx.address().do_send(ValidateRepo { message_token: self.message_token, }); } } #[derive(derive_more::Constructor, Message)] #[rtype(result = "()")] pub struct ValidateRepo { message_token: MessageToken, } impl Handler for RepoActor { type Result = (); #[tracing::instrument(name = "RepoActor::ValidateRepo", skip_all, fields(repo = %self.details, token = %msg.message_token))] fn handle(&mut self, msg: ValidateRepo, ctx: &mut Self::Context) -> Self::Result { match msg.message_token { message_token if self.message_token < message_token => { info!(%message_token, "New message token"); self.message_token = msg.message_token; } message_token if self.message_token > message_token => { info!("Dropping message from previous generation"); return; // message is expired } _ => { // do nothing } } if self.webhook_id.is_none() { webhook::register( self.details.clone(), self.webhook.clone(), ctx.address(), self.net.clone(), ) .in_current_span() .into_actor(self) .wait(ctx); } if let (Some(repository), Some(repo_config)) = (self.repository.clone(), self.details.repo_config.clone()) { let forge = self.forge.clone(); let addr = ctx.address(); let message_token = self.message_token; async move { match forge .branches_validate_positions(repository, repo_config) .await { Ok(git::validation::Positions { main, next, dev, dev_commit_history, }) => { addr.do_send(StartMonitoring::new(main, next, dev, dev_commit_history)); } Err(err) => { warn!("{:?}", err); tokio::time::sleep(Duration::from_secs(10)).await; addr.do_send(ValidateRepo::new(message_token)); } } } .in_current_span() .into_actor(self) .wait(ctx); } } } #[derive(Debug, derive_more::Constructor, Message)] #[rtype(result = "()")] pub struct StartMonitoring { main: git::Commit, next: git::Commit, dev: git::Commit, dev_commit_history: Vec, } impl Handler for RepoActor { type Result = (); #[tracing::instrument(name = "RepoActor::StartMonitoring", skip_all, fields(token = %self.message_token, repo = %self.details, main = %msg.main, next= %msg.next, dev = %msg.dev)) ] fn handle(&mut self, msg: StartMonitoring, ctx: &mut Self::Context) -> Self::Result { let Some(repo_config) = self.details.repo_config.clone() else { warn!("No config loaded"); return; }; let next_ahead_of_main = msg.main != msg.next; let dev_ahead_of_next = msg.next != msg.dev; info!(next_ahead_of_main, dev_ahead_of_next, "StartMonitoring"); let addr = ctx.address(); let forge = self.forge.clone(); if next_ahead_of_main { status::check_next(msg.next, addr, forge, self.message_token) .in_current_span() .into_actor(self) .wait(ctx); } else if dev_ahead_of_next { if let Some(repository) = self.repository.clone() { branch::advance_next( msg.next, msg.dev_commit_history, repo_config, forge, repository, addr, self.message_token, ) .in_current_span() .into_actor(self) .wait(ctx); } } } } #[derive(Message)] #[rtype(result = "()")] pub struct WebhookRegistered(webhook::WebhookId, webhook::WebhookAuth); impl Handler for RepoActor { type Result = (); #[tracing::instrument(name = "RepoActor::WebhookRegistered", skip_all, fields(repo = %self.details, webhook_id = %msg.0))] fn handle(&mut self, msg: WebhookRegistered, _ctx: &mut Self::Context) -> Self::Result { self.webhook_id.replace(msg.0); self.webhook_auth.replace(msg.1); } } #[derive(Message)] #[rtype(result = "()")] pub struct AdvanceMainTo(git::Commit); impl Handler for RepoActor { type Result = (); #[tracing::instrument(name = "RepoActor::AdvanceMainTo", skip_all, fields(repo = %self.details, commit = %msg.0))] fn handle(&mut self, msg: AdvanceMainTo, ctx: &mut Self::Context) -> Self::Result { let Some(repo_config) = self.details.repo_config.clone() else { warn!("No config loaded"); return; }; let Some(repository) = self.repository.clone() else { warn!("No repository opened"); return; }; let forge = self.forge.clone(); let addr = ctx.address(); let message_token = self.message_token; async move { branch::advance_main(msg.0, &repo_config, &forge, &repository).await; match repo_config.source() { git_next_config::RepoConfigSource::Repo => addr.do_send(LoadConfigFromRepo), git_next_config::RepoConfigSource::Server => { addr.do_send(ValidateRepo { message_token }) } } } .in_current_span() .into_actor(self) .wait(ctx); } } #[derive(Copy, Clone, Default, Debug, PartialEq, Eq, PartialOrd, Ord, derive_more::Display)] pub struct MessageToken(u32); impl MessageToken { pub fn new() -> Self { Self::default() } pub const fn next(&self) -> Self { Self(self.0 + 1) } }