git-next/crates/repo-actor/src/lib.rs

317 lines
11 KiB
Rust
Raw Normal View History

mod branch;
mod load;
pub mod status;
pub mod webhook;
2024-04-09 10:44:01 +01:00
2024-05-14 07:59:31 +01:00
#[cfg(test)]
mod tests;
use std::time::Duration;
2024-04-09 10:44:01 +01:00
use actix::prelude::*;
use git::validation::positions::{validate_positions, Positions};
2024-05-23 16:50:36 +01:00
use crate as repo_actor;
use git_next_config as config;
use git_next_forge as forge;
use git_next_git as git;
2024-05-23 16:50:36 +01:00
2024-04-09 10:44:01 +01:00
use kxio::network::Network;
use tracing::{debug, info, warn, Instrument};
2024-04-09 10:44:01 +01:00
#[derive(Debug, derive_more::Display)]
#[display("{}:{}:{}", generation, repo_details.forge.forge_name(), repo_details.repo_alias)]
2024-04-09 10:44:01 +01:00
pub struct RepoActor {
generation: git::Generation,
message_token: MessageToken,
repo_details: git::RepoDetails,
webhook: config::server::Webhook,
2024-05-23 16:50:36 +01:00
webhook_id: Option<webhook::WebhookId>, // INFO: if [None] then no webhook is configured
webhook_auth: Option<webhook::WebhookAuth>, // INFO: if [None] then no webhook is configured
last_main_commit: Option<git::Commit>,
last_next_commit: Option<git::Commit>,
last_dev_commit: Option<git::Commit>,
2024-05-23 16:50:36 +01:00
repository: Option<git::OpenRepository>,
2024-04-09 10:44:01 +01:00
net: Network,
forge: forge::Forge,
2024-04-09 10:44:01 +01:00
}
impl RepoActor {
pub fn new(
details: git::RepoDetails,
webhook: config::server::Webhook,
generation: git::Generation,
net: Network,
repo: git::Repository,
) -> Self {
let forge = match details.forge.forge_type() {
#[cfg(feature = "forgejo")]
config::ForgeType::ForgeJo => {
2024-05-23 16:50:36 +01:00
forge::Forge::new_forgejo(details.clone(), net.clone(), repo)
}
config::ForgeType::MockForge => forge::Forge::new_mock(),
};
debug!(?forge, "new");
2024-04-09 10:44:01 +01:00
Self {
generation,
message_token: MessageToken::new(),
repo_details: details,
webhook,
webhook_id: None,
webhook_auth: None,
last_main_commit: None,
last_next_commit: None,
last_dev_commit: None,
repository: None,
2024-04-09 10:44:01 +01:00
net,
forge,
2024-04-09 10:44:01 +01:00
}
}
}
impl Actor for RepoActor {
type Context = Context<Self>;
#[tracing::instrument(name = "RepoActor::stopping", skip_all, fields(repo = %self.repo_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.repo_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,
}
}
2024-04-09 10:44:01 +01:00
}
#[derive(Message)]
#[rtype(result = "()")]
2024-04-23 07:09:30 +01:00
pub struct CloneRepo;
impl Handler<CloneRepo> for RepoActor {
2024-04-09 10:44:01 +01:00
type Result = ();
#[tracing::instrument(name = "RepoActor::CloneRepo", skip_all, fields(repo = %self.repo_details, gitdir = %self.repo_details.gitdir))]
2024-04-23 07:09:30 +01:00
fn handle(&mut self, _msg: CloneRepo, ctx: &mut Self::Context) -> Self::Result {
let gitdir = self.repo_details.gitdir.clone();
2024-04-23 07:09:30 +01:00
match self.forge.repo_clone(gitdir) {
Ok(repository) => {
self.repository.replace(repository);
if self.repo_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}"),
2024-04-23 07:09:30 +01:00
}
}
}
#[derive(Message)]
#[rtype(result = "()")]
pub struct LoadConfigFromRepo;
impl Handler<LoadConfigFromRepo> for RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::LoadConfigFromRepo", skip_all, fields(repo = %self.repo_details))]
2024-04-23 07:09:30 +01:00
fn handle(&mut self, _msg: LoadConfigFromRepo, ctx: &mut Self::Context) -> Self::Result {
let details = self.repo_details.clone();
2024-04-09 10:44:01 +01:00
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);
2024-04-09 10:44:01 +01:00
}
}
#[derive(Message)]
#[rtype(result = "()")]
2024-05-23 16:50:36 +01:00
struct LoadedConfig(git_next_config::RepoConfig);
2024-04-09 10:44:01 +01:00
impl Handler<LoadedConfig> for RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::LoadedConfig", skip_all, fields(repo = %self.repo_details, branches = %msg.0))]
fn handle(&mut self, msg: LoadedConfig, ctx: &mut Self::Context) -> Self::Result {
let repo_config = msg.0;
self.repo_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<ValidateRepo> for RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::ValidateRepo", skip_all, fields(repo = %self.repo_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.repo_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.repo_details.repo_config.clone(),
) {
let repo_details = self.repo_details.clone();
let addr = ctx.address();
let message_token = self.message_token;
async move {
match validate_positions(&repository, &repo_details, repo_config) {
Ok(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);
}
2024-04-09 10:44:01 +01:00
}
}
#[derive(Debug, derive_more::Constructor, Message)]
#[rtype(result = "()")]
pub struct StartMonitoring {
main: git::Commit,
next: git::Commit,
dev: git::Commit,
dev_commit_history: Vec<git::Commit>,
}
impl Handler<StartMonitoring> for RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::StartMonitoring", skip_all,
fields(token = %self.message_token, repo = %self.repo_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.repo_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,
self.repo_details.clone(),
repo_config,
repository,
addr,
self.message_token,
)
.in_current_span()
.into_actor(self)
.wait(ctx);
}
}
}
}
#[derive(Message)]
#[rtype(result = "()")]
2024-05-23 16:50:36 +01:00
pub struct WebhookRegistered(webhook::WebhookId, webhook::WebhookAuth);
impl Handler<WebhookRegistered> for RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::WebhookRegistered", skip_all, fields(repo = %self.repo_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<AdvanceMainTo> for RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::AdvanceMainTo", skip_all, fields(repo = %self.repo_details, commit = %msg.0))]
fn handle(&mut self, msg: AdvanceMainTo, ctx: &mut Self::Context) -> Self::Result {
let Some(repo_config) = self.repo_details.repo_config.clone() else {
warn!("No config loaded");
return;
};
let Some(repository) = self.repository.clone() else {
warn!("No repository opened");
return;
};
let repo_details = self.repo_details.clone();
let addr = ctx.address();
let message_token = self.message_token;
async move {
branch::advance_main(msg.0, &repo_details, &repo_config, &repository).await;
match repo_config.source() {
2024-05-23 16:50:36 +01:00
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)
}
}