307 lines
10 KiB
Rust
307 lines
10 KiB
Rust
mod branch;
|
|
pub mod config;
|
|
mod load;
|
|
pub mod status;
|
|
pub mod webhook;
|
|
|
|
#[cfg(test)]
|
|
mod tests;
|
|
|
|
use std::time::Duration;
|
|
|
|
use actix::prelude::*;
|
|
use git::OpenRepository;
|
|
use git_next_config::{server::Webhook, ForgeType, RepoConfig, RepoConfigSource};
|
|
use git_next_git::{self as git, Generation, RepoDetails};
|
|
use git_next_gitforge::{self as gitforge, validation};
|
|
use kxio::network::Network;
|
|
use tracing::{debug, info, warn, Instrument};
|
|
|
|
// use crate::{actors::repo::webhook::WebhookAuth, config::Webhook, gitforge};
|
|
|
|
use crate::webhook::WebhookAuth;
|
|
|
|
use self::webhook::WebhookId;
|
|
|
|
#[derive(Debug, derive_more::Display)]
|
|
#[display("{}:{}:{}", generation, details.forge.forge_name(), details.repo_alias)]
|
|
pub struct RepoActor {
|
|
generation: Generation,
|
|
message_token: gitforge::MessageToken,
|
|
details: RepoDetails,
|
|
webhook: Webhook,
|
|
webhook_id: Option<WebhookId>, // INFO: if [None] then no webhook is configured
|
|
webhook_auth: Option<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>,
|
|
repository: Option<OpenRepository>,
|
|
net: Network,
|
|
forge: gitforge::Forge,
|
|
}
|
|
impl RepoActor {
|
|
pub fn new(
|
|
details: RepoDetails,
|
|
webhook: Webhook,
|
|
generation: Generation,
|
|
net: Network,
|
|
repo: git::Repository,
|
|
) -> Self {
|
|
let forge = match details.forge.forge_type() {
|
|
#[cfg(feature = "forgejo")]
|
|
ForgeType::ForgeJo => gitforge::Forge::new_forgejo(details.clone(), net.clone(), repo),
|
|
ForgeType::MockForge => gitforge::Forge::new_mock(),
|
|
};
|
|
debug!(?forge, "new");
|
|
Self {
|
|
generation,
|
|
message_token: gitforge::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<Self>;
|
|
#[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<CloneRepo> 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<LoadConfigFromRepo> 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();
|
|
config::load(details, addr, forge)
|
|
.in_current_span()
|
|
.into_actor(self)
|
|
.wait(ctx);
|
|
}
|
|
}
|
|
|
|
#[derive(Message)]
|
|
#[rtype(result = "()")]
|
|
struct LoadedConfig(RepoConfig);
|
|
impl Handler<LoadedConfig> 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: gitforge::MessageToken,
|
|
}
|
|
impl Handler<ValidateRepo> 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(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<git::Commit>,
|
|
}
|
|
impl Handler<StartMonitoring> 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(WebhookId, WebhookAuth);
|
|
impl Handler<WebhookRegistered> 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<AdvanceMainTo> 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() {
|
|
RepoConfigSource::Repo => addr.do_send(LoadConfigFromRepo),
|
|
RepoConfigSource::Server => addr.do_send(ValidateRepo { message_token }),
|
|
}
|
|
}
|
|
.in_current_span()
|
|
.into_actor(self)
|
|
.wait(ctx);
|
|
}
|
|
}
|