refactor: merge repo-actor crate into cli crate
All checks were successful
Rust / build (push) Successful in 2m29s
ci/woodpecker/push/cron-docker-builder Pipeline was successful
ci/woodpecker/push/tag-created Pipeline was successful
ci/woodpecker/push/push-next Pipeline was successful

This commit is contained in:
Paul Campbell 2024-07-28 08:58:32 +01:00
parent 12ecc308d5
commit c1981d862c
59 changed files with 836 additions and 880 deletions

View file

@ -15,7 +15,6 @@ categories = { workspace = true }
git-next-core = { workspace = true }
git-next-server-actor = { workspace = true }
git-next-forge = { workspace = true }
git-next-repo-actor = { workspace = true }
# CLI parsing
clap = { workspace = true }
@ -28,6 +27,18 @@ console-subscriber = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
# git
async-trait = { workspace = true }
# Conventional Commit check
git-conventional = { workspace = true }
# TOML parsing
toml = { workspace = true }
# base64 decoding
base64 = { workspace = true }
# Actors
actix = { workspace = true }
actix-rt = { workspace = true }
@ -54,6 +65,9 @@ inotify = { workspace = true }
# Testing
assert2 = { workspace = true }
test-log = { workspace = true }
rand = { workspace = true }
pretty_assertions = { workspace = true }
mockall = { workspace = true }
[lints.clippy]
nursery = { level = "warn", priority = -1 }

View file

@ -521,7 +521,6 @@ stateDiagram-v2
cli --> core
cli --> forge
cli --> repo_actor
forge --> core
forge --> forge_forgejo
@ -530,9 +529,6 @@ stateDiagram-v2
forge_forgejo --> core
forge_github --> core
repo_actor --> core
repo_actor --> forge
```
## License

View file

@ -1,6 +1,7 @@
//
mod file_watcher;
mod init;
mod repo;
mod server;
mod webhook;

View file

@ -1,19 +1,25 @@
//
use crate::messages::MessageToken;
use crate::repo::messages::MessageToken;
use git_next_core::{
git::{self, repository::open::OpenRepositoryLike},
git::{
commit::Message,
push::{reset, Force},
repository::open::OpenRepositoryLike,
Commit, RepoDetails,
},
RepoConfig,
};
use derive_more::Display;
use tracing::{info, warn};
use tracing::{info, instrument, warn};
// advance next to the next commit towards the head of the dev branch
#[tracing::instrument(fields(next), skip_all)]
#[instrument(fields(next), skip_all)]
pub fn advance_next(
next: &git::Commit,
dev_commit_history: &[git::Commit],
repo_details: git::RepoDetails,
next: &Commit,
dev_commit_history: &[Commit],
repo_details: RepoDetails,
repo_config: RepoConfig,
open_repository: &dyn OpenRepositoryLike,
message_token: MessageToken,
@ -22,23 +28,23 @@ pub fn advance_next(
find_next_commit_on_dev(next, dev_commit_history).ok_or_else(|| Error::NextAtDev)?;
validate_commit_message(commit.message())?;
info!("Advancing next to commit '{}'", commit);
git::push::reset(
reset(
open_repository,
&repo_details,
&repo_config.branches().next(),
&commit.into(),
&git::push::Force::No,
&Force::No,
)?;
Ok(message_token)
}
#[tracing::instrument]
fn validate_commit_message(message: &git::commit::Message) -> Result<()> {
#[instrument]
fn validate_commit_message(message: &Message) -> Result<()> {
let message = &message.to_string();
if message.to_ascii_lowercase().starts_with("wip") {
return Err(Error::IsWorkInProgress);
}
match git_conventional::Commit::parse(message) {
match ::git_conventional::Commit::parse(message) {
Ok(commit) => {
info!(?commit, "Pass");
Ok(())
@ -52,11 +58,8 @@ fn validate_commit_message(message: &git::commit::Message) -> Result<()> {
}
}
pub fn find_next_commit_on_dev(
next: &git::Commit,
dev_commit_history: &[git::Commit],
) -> Option<git::Commit> {
let mut next_commit: Option<&git::Commit> = None;
pub fn find_next_commit_on_dev(next: &Commit, dev_commit_history: &[Commit]) -> Option<Commit> {
let mut next_commit: Option<&Commit> = None;
for commit in dev_commit_history.iter() {
if commit == next {
break;
@ -67,29 +70,30 @@ pub fn find_next_commit_on_dev(
}
// advance main branch to the commit 'next'
#[tracing::instrument(fields(next), skip_all)]
#[instrument(fields(next), skip_all)]
pub fn advance_main(
next: git::Commit,
repo_details: &git::RepoDetails,
next: Commit,
repo_details: &RepoDetails,
repo_config: &RepoConfig,
open_repository: &dyn OpenRepositoryLike,
) -> Result<()> {
info!("Advancing main to next");
git::push::reset(
reset(
open_repository,
repo_details,
&repo_config.branches().main(),
&next.into(),
&git::push::Force::No,
&Force::No,
)?;
Ok(())
}
pub type Result<T> = core::result::Result<T, Error>;
pub type Result<T> = std::result::Result<T, Error>;
#[derive(Debug, thiserror::Error, Display)]
pub enum Error {
#[display("push: {}", 0)]
Push(#[from] git::push::Error),
Push(#[from] crate::git::push::Error),
#[display("no commits to advance next to")]
NextAtDev,

View file

@ -1,18 +1,23 @@
//
use crate as actor;
use actix::prelude::*;
use git_next_core::RepoConfigSource;
impl Handler<actor::messages::AdvanceMain> for actor::RepoActor {
use tracing::warn;
use crate::repo::{
branch::advance_main,
do_send,
messages::{AdvanceMain, LoadConfigFromRepo, ValidateRepo},
RepoActor,
};
impl Handler<AdvanceMain> for RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::AdvanceMainTo", skip_all, fields(repo = %self.repo_details, commit = ?msg))]
fn handle(
&mut self,
msg: actor::messages::AdvanceMain,
ctx: &mut Self::Context,
) -> Self::Result {
fn handle(&mut self, msg: AdvanceMain, ctx: &mut Self::Context) -> Self::Result {
let Some(repo_config) = self.repo_details.repo_config.clone() else {
tracing::warn!("No config loaded");
warn!("No config loaded");
return;
};
let Some(open_repository) = &self.open_repository else {
@ -22,25 +27,21 @@ impl Handler<actor::messages::AdvanceMain> for actor::RepoActor {
let addr = ctx.address();
let message_token = self.message_token;
match actor::branch::advance_main(
match advance_main(
msg.unwrap(),
&repo_details,
&repo_config,
&**open_repository,
) {
Err(err) => {
tracing::warn!("advance main: {err}");
warn!("advance main: {err}");
}
Ok(_) => match repo_config.source() {
RepoConfigSource::Repo => {
actor::do_send(addr, actor::messages::LoadConfigFromRepo, self.log.as_ref());
do_send(addr, LoadConfigFromRepo, self.log.as_ref());
}
RepoConfigSource::Server => {
actor::do_send(
addr,
actor::messages::ValidateRepo::new(message_token),
self.log.as_ref(),
);
do_send(addr, ValidateRepo::new(message_token), self.log.as_ref());
}
},
}

View file

@ -1,15 +1,19 @@
//
use crate as actor;
use actix::prelude::*;
impl Handler<actor::messages::AdvanceNext> for actor::RepoActor {
use tracing::warn;
use crate::repo::{
branch::advance_next,
do_send,
messages::{AdvanceNext, ValidateRepo},
RepoActor,
};
impl Handler<AdvanceNext> for RepoActor {
type Result = ();
fn handle(
&mut self,
msg: actor::messages::AdvanceNext,
ctx: &mut Self::Context,
) -> Self::Result {
fn handle(&mut self, msg: AdvanceNext, ctx: &mut Self::Context) -> Self::Result {
let Some(repo_config) = &self.repo_details.repo_config else {
return;
};
@ -21,7 +25,7 @@ impl Handler<actor::messages::AdvanceNext> for actor::RepoActor {
let repo_config = repo_config.clone();
let addr = ctx.address();
match actor::branch::advance_next(
match advance_next(
&next_commit,
&dev_commit_history,
repo_details,
@ -32,13 +36,9 @@ impl Handler<actor::messages::AdvanceNext> for actor::RepoActor {
Ok(message_token) => {
// pause to allow any CI checks to be started
std::thread::sleep(self.sleep_duration);
actor::do_send(
addr,
actor::messages::ValidateRepo::new(message_token),
self.log.as_ref(),
);
do_send(addr, ValidateRepo::new(message_token), self.log.as_ref());
}
Err(err) => tracing::warn!("advance next: {err}"),
Err(err) => warn!("advance next: {err}"),
}
}
}

View file

@ -0,0 +1,32 @@
//
use actix::prelude::*;
use tracing::{debug, Instrument as _};
use crate::repo::{
do_send,
messages::{CheckCIStatus, ReceiveCIStatus},
RepoActor,
};
impl Handler<CheckCIStatus> for RepoActor {
type Result = ();
fn handle(&mut self, msg: CheckCIStatus, ctx: &mut Self::Context) -> Self::Result {
crate::repo::logger(self.log.as_ref(), "start: CheckCIStatus");
let addr = ctx.address();
let forge = self.forge.duplicate();
let next = msg.unwrap();
let log = self.log.clone();
// get the status - pass, fail, pending (all others map to fail, e.g. error)
async move {
let status = forge.commit_status(&next).await;
debug!("got status: {status:?}");
do_send(addr, ReceiveCIStatus::new((next, status)), log.as_ref());
}
.in_current_span()
.into_actor(self)
.wait(ctx);
}
}

View file

@ -0,0 +1,37 @@
//
use actix::prelude::*;
use git_next_core::git;
use tracing::{debug, instrument, warn};
use crate::repo::{
do_send, logger,
messages::{CloneRepo, LoadConfigFromRepo, RegisterWebhook},
RepoActor,
};
impl Handler<CloneRepo> for RepoActor {
type Result = ();
#[instrument(name = "RepoActor::CloneRepo", skip_all, fields(repo = %self.repo_details))]
fn handle(&mut self, _msg: CloneRepo, ctx: &mut Self::Context) -> Self::Result {
logger(self.log.as_ref(), "Handler: CloneRepo: start");
debug!("Handler: CloneRepo: start");
match git::repository::open(&*self.repository_factory, &self.repo_details) {
Ok(repository) => {
logger(self.log.as_ref(), "open okay");
debug!("open okay");
self.open_repository.replace(repository);
if self.repo_details.repo_config.is_none() {
do_send(ctx.address(), LoadConfigFromRepo, self.log.as_ref());
} else {
do_send(ctx.address(), RegisterWebhook::new(), self.log.as_ref());
}
}
Err(err) => {
logger(self.log.as_ref(), "open failed");
warn!("Could not open repo: {err:?}")
}
}
debug!("Handler: CloneRepo: finish");
}
}

View file

@ -3,17 +3,19 @@ use actix::prelude::*;
use git_next_core::git::UserNotification;
use tracing::Instrument as _;
use tracing::{debug, instrument, Instrument as _};
impl Handler<crate::messages::LoadConfigFromRepo> for crate::RepoActor {
use crate::repo::{
do_send, load,
messages::{LoadConfigFromRepo, ReceiveRepoConfig},
notify_user, RepoActor,
};
impl Handler<LoadConfigFromRepo> for RepoActor {
type Result = ();
#[tracing::instrument(name = "Repocrate::LoadConfigFromRepo", skip_all, fields(repo = %self.repo_details))]
fn handle(
&mut self,
_msg: crate::messages::LoadConfigFromRepo,
ctx: &mut Self::Context,
) -> Self::Result {
tracing::debug!("Handler: LoadConfigFromRepo: start");
#[instrument(name = "Repocrate::repo::LoadConfigFromRepo", skip_all, fields(repo = %self.repo_details))]
fn handle(&mut self, _msg: LoadConfigFromRepo, ctx: &mut Self::Context) -> Self::Result {
debug!("Handler: LoadConfigFromRepo: start");
let Some(open_repository) = &self.open_repository else {
return;
};
@ -25,13 +27,9 @@ impl Handler<crate::messages::LoadConfigFromRepo> for crate::RepoActor {
let notify_user_recipient = self.notify_user_recipient.clone();
let log = self.log.clone();
async move {
match crate::load::config_from_repository(repo_details, &*open_repository).await {
Ok(repo_config) => crate::do_send(
addr,
crate::messages::ReceiveRepoConfig::new(repo_config),
log.as_ref(),
),
Err(err) => crate::notify_user(
match load::config_from_repository(repo_details, &*open_repository).await {
Ok(repo_config) => do_send(addr, ReceiveRepoConfig::new(repo_config), log.as_ref()),
Err(err) => notify_user(
notify_user_recipient.as_ref(),
UserNotification::RepoConfigLoadFailure {
forge_alias,
@ -45,6 +43,6 @@ impl Handler<crate::messages::LoadConfigFromRepo> for crate::RepoActor {
.in_current_span()
.into_actor(self)
.wait(ctx);
tracing::debug!("Handler: LoadConfigFromRepo: finish");
debug!("Handler: LoadConfigFromRepo: finish");
}
}

View file

@ -0,0 +1,55 @@
//
use actix::prelude::*;
use git_next_core::git::{forge::commit::Status, UserNotification};
use tracing::debug;
use crate::repo::{
delay_send, do_send, logger,
messages::{AdvanceMain, ReceiveCIStatus, ValidateRepo},
notify_user, RepoActor,
};
impl Handler<ReceiveCIStatus> for RepoActor {
type Result = ();
fn handle(&mut self, msg: ReceiveCIStatus, ctx: &mut Self::Context) -> Self::Result {
let log = self.log.clone();
logger(log.as_ref(), "start: ReceiveCIStatus");
let addr = ctx.address();
let (next, status) = msg.unwrap();
let forge_alias = self.repo_details.forge.forge_alias().clone();
let repo_alias = self.repo_details.repo_alias.clone();
let message_token = self.message_token;
let sleep_duration = self.sleep_duration;
debug!(?status, "");
match status {
Status::Pass => {
do_send(addr, AdvanceMain::new(next), self.log.as_ref());
}
Status::Pending => {
std::thread::sleep(sleep_duration);
do_send(addr, ValidateRepo::new(message_token), self.log.as_ref());
}
Status::Fail => {
tracing::warn!("Checks have failed");
notify_user(
self.notify_user_recipient.as_ref(),
UserNotification::CICheckFailed {
forge_alias,
repo_alias,
commit: next,
},
log.as_ref(),
);
delay_send(
addr,
sleep_duration,
ValidateRepo::new(message_token),
self.log.as_ref(),
);
}
}
}
}

View file

@ -0,0 +1,20 @@
//
use actix::prelude::*;
use tracing::instrument;
use crate::repo::{
do_send,
messages::{ReceiveRepoConfig, RegisterWebhook},
RepoActor,
};
impl Handler<ReceiveRepoConfig> for RepoActor {
type Result = ();
#[instrument(name = "RepoActor::ReceiveRepoConfig", skip_all, fields(repo = %self.repo_details, branches = ?msg))]
fn handle(&mut self, msg: ReceiveRepoConfig, ctx: &mut Self::Context) -> Self::Result {
let repo_config = msg.unwrap();
self.repo_details.repo_config.replace(repo_config);
do_send(ctx.address(), RegisterWebhook::new(), self.log.as_ref());
}
}

View file

@ -1,8 +1,14 @@
//
use actix::prelude::*;
use tracing::Instrument as _;
use crate::{messages::RegisterWebhook, RepoActor};
use tracing::{debug, Instrument as _};
use crate::repo::{
do_send,
messages::{RegisterWebhook, WebhookRegistered},
notify_user, RepoActor,
};
use git_next_core::git::UserNotification;
impl Handler<RegisterWebhook> for RepoActor {
@ -17,19 +23,19 @@ impl Handler<RegisterWebhook> for RepoActor {
let addr = ctx.address();
let notify_user_recipient = self.notify_user_recipient.clone();
let log = self.log.clone();
tracing::debug!("registering webhook");
debug!("registering webhook");
async move {
match forge.register_webhook(&webhook_url).await {
Ok(registered_webhook) => {
tracing::debug!(?registered_webhook, "");
crate::do_send(
debug!(?registered_webhook, "");
do_send(
addr,
crate::messages::WebhookRegistered::from(registered_webhook),
WebhookRegistered::from(registered_webhook),
log.as_ref(),
);
}
Err(err) => {
crate::notify_user(
notify_user(
notify_user_recipient.as_ref(),
UserNotification::WebhookRegistration {
forge_alias,

View file

@ -1,9 +1,9 @@
//
use actix::prelude::*;
use tracing::Instrument as _;
use crate as actor;
use actor::{messages::UnRegisterWebhook, RepoActor};
use tracing::{debug, warn, Instrument as _};
use crate::repo::{messages::UnRegisterWebhook, RepoActor};
impl Handler<UnRegisterWebhook> for RepoActor {
type Result = ();
@ -11,17 +11,17 @@ impl Handler<UnRegisterWebhook> for RepoActor {
fn handle(&mut self, _msg: UnRegisterWebhook, ctx: &mut Self::Context) -> Self::Result {
if let Some(webhook_id) = self.webhook_id.take() {
let forge = self.forge.duplicate();
tracing::debug!("unregistering webhook");
debug!("unregistering webhook");
async move {
match forge.unregister_webhook(&webhook_id).await {
Ok(_) => tracing::debug!("unregistered webhook"),
Err(err) => tracing::warn!(?err, "unregistering webhook"),
Ok(_) => debug!("unregistered webhook"),
Err(err) => warn!(?err, "unregistering webhook"),
}
}
.in_current_span()
.into_actor(self)
.wait(ctx);
tracing::debug!("unregistering webhook done");
debug!("unregistering webhook done");
}
}
}

View file

@ -0,0 +1,122 @@
//
use actix::prelude::*;
use derive_more::Deref as _;
use tracing::{debug, instrument, Instrument as _};
use crate::repo::{
do_send, logger,
messages::{AdvanceNext, CheckCIStatus, MessageToken, ValidateRepo},
notify_user, RepoActor,
};
use git_next_core::git::validation::positions::{validate_positions, Error, Positions};
impl Handler<ValidateRepo> for RepoActor {
type Result = ();
#[instrument(name = "RepoActor::ValidateRepo", skip_all, fields(repo = %self.repo_details, token = %msg.deref()))]
fn handle(&mut self, msg: ValidateRepo, ctx: &mut Self::Context) -> Self::Result {
logger(self.log.as_ref(), "start: ValidateRepo");
// Message Token - make sure we are only triggered for the latest/current token
match self.token_status(msg.unwrap()) {
TokenStatus::Current => {} // do nothing
TokenStatus::Expired => {
logger(
self.log.as_ref(),
format!("discarded: old message token: {}", self.message_token),
);
return; // message is expired
}
TokenStatus::New(message_token) => {
self.message_token = message_token;
logger(
self.log.as_ref(),
format!("new message token: {}", self.message_token),
);
}
}
logger(
self.log.as_ref(),
format!("accepted token: {}", self.message_token),
);
// Repository positions
let Some(ref open_repository) = self.open_repository else {
logger(self.log.as_ref(), "no open repository");
return;
};
logger(self.log.as_ref(), "have open repository");
let Some(repo_config) = self.repo_details.repo_config.clone() else {
logger(self.log.as_ref(), "no repo config");
return;
};
logger(self.log.as_ref(), "have repo config");
match validate_positions(&**open_repository, &self.repo_details, repo_config) {
Ok(Positions {
main,
next,
dev,
dev_commit_history,
}) => {
debug!(%main, %next, %dev, "positions");
if next != main {
do_send(ctx.address(), CheckCIStatus::new(next), self.log.as_ref());
} else if next != dev {
do_send(
ctx.address(),
AdvanceNext::new((next, dev_commit_history)),
self.log.as_ref(),
)
} else {
// do nothing
}
}
Err(Error::Retryable(message)) => {
logger(self.log.as_ref(), message);
let addr = ctx.address();
let message_token = self.message_token;
let sleep_duration = self.sleep_duration;
let log = self.log.clone();
async move {
debug!("sleeping before retrying...");
logger(log.as_ref(), "before sleep");
actix_rt::time::sleep(sleep_duration).await;
logger(log.as_ref(), "after sleep");
do_send(addr, ValidateRepo::new(message_token), log.as_ref());
}
.in_current_span()
.into_actor(self)
.wait(ctx);
}
Err(Error::UserIntervention(user_notification)) => notify_user(
self.notify_user_recipient.as_ref(),
user_notification,
self.log.as_ref(),
),
Err(Error::NonRetryable(message)) => {
logger(self.log.as_ref(), message);
}
}
}
}
enum TokenStatus {
Current,
Expired,
New(MessageToken),
}
impl RepoActor {
fn token_status(&self, new: MessageToken) -> TokenStatus {
let current = &self.message_token;
if &new > current {
return TokenStatus::New(new);
}
if current > &new {
return TokenStatus::Expired;
}
TokenStatus::Current
}
}

View file

@ -1,22 +1,27 @@
//
use actix::prelude::*;
use crate::{messages::WebhookNotification, RepoActorLog};
use tracing::{info, instrument, warn};
use crate::repo::{
do_send, logger,
messages::{ValidateRepo, WebhookNotification},
RepoActor, RepoActorLog,
};
use git_next_core::{
git::{self, Commit, ForgeLike},
git::{Commit, ForgeLike},
webhook::{push::Branch, Push},
BranchName, WebhookAuth,
};
use tracing::{info, warn};
impl Handler<WebhookNotification> for crate::RepoActor {
impl Handler<WebhookNotification> for RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::WebhookMessage", skip_all, fields(token = %self.message_token, repo = %self.repo_details))]
#[instrument(name = "RepoActor::WebhookMessage", skip_all, fields(token = %self.message_token, repo = %self.repo_details))]
fn handle(&mut self, msg: WebhookNotification, ctx: &mut Self::Context) -> Self::Result {
let Some(config) = &self.repo_details.repo_config else {
crate::logger(self.log.as_ref(), "server has no repo config");
logger(self.log.as_ref(), "server has no repo config");
warn!("No repo config");
return;
};
@ -33,13 +38,13 @@ impl Handler<WebhookNotification> for crate::RepoActor {
let body = msg.body();
match self.forge.parse_webhook_body(body) {
Err(err) => {
crate::logger(self.log.as_ref(), "message parse error - not a push");
logger(self.log.as_ref(), "message parse error - not a push");
warn!(?err, "Not a 'push'");
return;
}
Ok(push) => match push.branch(config.branches()) {
None => {
crate::logger(self.log.as_ref(), "unknown branch");
logger(self.log.as_ref(), "unknown branch");
warn!(
?push,
"Unrecognised branch, we should be filtering to only the ones we want"
@ -89,9 +94,9 @@ impl Handler<WebhookNotification> for crate::RepoActor {
token = %message_token,
"New commit"
);
crate::do_send(
do_send(
ctx.address(),
crate::messages::ValidateRepo::new(message_token),
ValidateRepo::new(message_token),
self.log.as_ref(),
);
}
@ -104,13 +109,13 @@ fn validate_notification(
log: Option<&RepoActorLog>,
) -> Result<(), ()> {
let Some(expected_authorization) = webhook_auth else {
crate::logger(log, "server has no auth token");
logger(log, "server has no auth token");
warn!("Don't know what authorization to expect");
return Err(());
};
if !forge.is_message_authorised(msg, expected_authorization) {
crate::logger(log, "message authorisation is invalid");
logger(log, "message authorisation is invalid");
warn!(
"Invalid authorization - expected {}",
expected_authorization
@ -118,7 +123,7 @@ fn validate_notification(
return Err(());
}
if forge.should_ignore_message(msg) {
crate::logger(log, "forge sent ignorable message");
logger(log, "forge sent ignorable message");
return Err(());
}
Ok(())
@ -130,10 +135,10 @@ fn handle_push(
last_commit: &mut Option<Commit>,
log: Option<&RepoActorLog>,
) -> Result<(), ()> {
crate::logger(log, "message is for dev branch");
let commit = git::Commit::from(push);
logger(log, "message is for dev branch");
let commit = Commit::from(push);
if last_commit.as_ref() == Some(&commit) {
crate::logger(log, format!("not a new commit on {branch}"));
logger(log, format!("not a new commit on {branch}"));
info!(
%branch ,
%commit,

View file

@ -0,0 +1,23 @@
//
use actix::prelude::*;
use tracing::instrument;
use crate::repo::{
do_send,
messages::{ValidateRepo, WebhookRegistered},
RepoActor,
};
impl Handler<WebhookRegistered> for RepoActor {
type Result = ();
#[instrument(name = "RepoActor::WebhookRegistered", skip_all, fields(repo = %self.repo_details, webhook_id = %msg.webhook_id()))]
fn handle(&mut self, msg: WebhookRegistered, ctx: &mut Self::Context) -> Self::Result {
self.webhook_id.replace(msg.webhook_id().clone());
self.webhook_auth.replace(msg.webhook_auth().clone());
do_send(
ctx.address(),
ValidateRepo::new(self.message_token),
self.log.as_ref(),
);
}
}

View file

@ -1,18 +1,18 @@
//
use git_next_core::{
git::{self, repository::open::OpenRepositoryLike},
server, BranchName, RepoConfig,
git::{repository::open::OpenRepositoryLike, RepoDetails},
BranchName, RepoConfig,
};
use std::path::PathBuf;
use derive_more::Display;
use tracing::info;
use tracing::{info, instrument};
/// Loads the [RepoConfig] from the `.git-next.toml` file in the repository
#[tracing::instrument(skip_all, fields(branch = %repo_details.branch))]
#[instrument(skip_all, fields(branch = %repo_details.branch))]
pub async fn config_from_repository(
repo_details: git::RepoDetails,
repo_details: RepoDetails,
open_repository: &dyn OpenRepositoryLike,
) -> Result<RepoConfig> {
info!("Loading .git-next.toml from repo");
@ -34,20 +34,21 @@ fn required_branch(branch_name: &BranchName, branches: &[BranchName]) -> Result<
Ok(())
}
pub type Result<T> = core::result::Result<T, Error>;
pub type Result<T> = std::result::Result<T, Error>;
#[derive(Debug, thiserror::Error, Display)]
pub enum Error {
#[display("file")]
File(#[from] git::file::Error),
File(#[from] crate::git::file::Error),
#[display("config")]
Config(#[from] server::Error),
Config(#[from] git_next_core::server::Error),
#[display("toml")]
Toml(#[from] toml::de::Error),
#[display("push")]
Push(#[from] git::push::Error),
Push(#[from] crate::git::push::Error),
#[display("branch not found: {}", 0)]
BranchNotFound(BranchName),

View file

@ -2,8 +2,8 @@
use derive_more::Display;
use git_next_core::{
git::{self, UserNotification},
message, newtype, webhook, RegisteredWebhook, RepoConfig, WebhookAuth, WebhookId,
git::{forge::commit::Status, Commit, UserNotification},
message, newtype, ForgeNotification, RegisteredWebhook, RepoConfig, WebhookAuth, WebhookId,
};
message!(LoadConfigFromRepo: "Request to load the `git-next.toml` from the git repo.");
@ -52,15 +52,15 @@ impl MessageToken {
}
message!(RegisterWebhook: "Requests that a Webhook be registered with the forge.");
message!(CheckCIStatus: git::Commit: r#"Requests that the CI status for the commit be checked.
message!(CheckCIStatus: Commit: r#"Requests that the CI status for the commit be checked.
Once the CI Status has been received it will be sent via a [ReceiveCIStatus] message.
Contains the commit from the tip of the `next` branch."#); // next commit
message!(ReceiveCIStatus: (git::Commit, git::forge::commit::Status): r#"Notification of the status of the CI checks for the commit.
message!(ReceiveCIStatus: (Commit, Status): r#"Notification of the status of the CI checks for the commit.
Contains a tuple of the commit that was checked (the tip of the `next` branch) and the status."#); // commit and it's status
message!(AdvanceNext: (git::Commit, Vec<git::Commit>): "Request to advance the `next` branch on to the next commit on the `dev branch."); // next commit and the dev commit history
message!(AdvanceMain: git::Commit: "Request to advance the `main` branch on to same commit as the `next` branch."); // next commit
message!(WebhookNotification: webhook::forge_notification::ForgeNotification: "Notification of a webhook message from the forge.");
message!(AdvanceNext: (Commit, Vec<Commit>): "Request to advance the `next` branch on to the next commit on the `dev branch."); // next commit and the dev commit history
message!(AdvanceMain: Commit: "Request to advance the `main` branch on to same commit as the `next` branch."); // next commit
message!(WebhookNotification: ForgeNotification: "Notification of a webhook message from the forge.");
message!(NotifyUser: UserNotification: "Request to send the message payload to the notification webhook");

165
crates/cli/src/repo/mod.rs Normal file
View file

@ -0,0 +1,165 @@
//
use actix::prelude::*;
use derive_more::Deref;
use kxio::network::Network;
use messages::NotifyUser;
use std::time::Duration;
use tracing::{info, warn, Instrument};
use git_next_core::{
git::{
self,
repository::{factory::RepositoryFactory, open::OpenRepositoryLike},
UserNotification,
},
server, WebhookAuth, WebhookId,
};
mod branch;
pub mod handlers;
mod load;
pub mod messages;
mod notifications;
#[cfg(test)]
mod tests;
#[derive(Clone, Debug, Default)]
pub struct RepoActorLog(std::sync::Arc<std::sync::RwLock<Vec<String>>>);
impl Deref for RepoActorLog {
type Target = std::sync::Arc<std::sync::RwLock<Vec<String>>>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
/// An actor that represents a Git Repository.
///
/// When this actor is started it is sent the [CloneRepo] message.
#[derive(Debug, derive_more::Display, derive_with::With)]
#[display("{}:{}:{}", generation, repo_details.forge.forge_alias(), repo_details.repo_alias)]
pub struct RepoActor {
sleep_duration: std::time::Duration,
generation: git::Generation,
message_token: messages::MessageToken,
repo_details: git::RepoDetails,
webhook: server::InboundWebhook,
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_factory: Box<dyn RepositoryFactory>,
open_repository: Option<Box<dyn OpenRepositoryLike>>,
net: Network,
forge: Box<dyn git::ForgeLike>,
log: Option<RepoActorLog>,
notify_user_recipient: Option<Recipient<NotifyUser>>,
}
impl RepoActor {
#[allow(clippy::too_many_arguments)]
pub fn new(
repo_details: git::RepoDetails,
forge: Box<dyn git::ForgeLike>,
webhook: server::InboundWebhook,
generation: git::Generation,
net: Network,
repository_factory: Box<dyn RepositoryFactory>,
sleep_duration: std::time::Duration,
notify_user_recipient: Option<Recipient<NotifyUser>>,
) -> Self {
let message_token = messages::MessageToken::default();
Self {
generation,
message_token,
repo_details,
webhook,
webhook_id: None,
webhook_auth: None,
last_main_commit: None,
last_next_commit: None,
last_dev_commit: None,
repository_factory,
open_repository: None,
forge,
net,
sleep_duration,
log: None,
notify_user_recipient,
}
}
}
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 {
tracing::debug!("stopping");
info!("Checking webhook");
match self.webhook_id.take() {
Some(webhook_id) => {
tracing::warn!("stopping - unregistering webhook");
info!(%webhook_id, "Unregistring webhook");
let forge = self.forge.duplicate();
async move {
if let Err(err) = forge.unregister_webhook(&webhook_id).await {
warn!("unregistering webhook: {err}");
}
}
.in_current_span()
.into_actor(self)
.wait(ctx);
Running::Continue
}
None => Running::Stop,
}
}
}
pub fn delay_send<M>(addr: Addr<RepoActor>, delay: Duration, msg: M, log: Option<&RepoActorLog>)
where
M: actix::Message + Send + 'static + std::fmt::Debug,
RepoActor: actix::Handler<M>,
<M as actix::Message>::Result: Send,
{
let log_message = format!("send-after-delay: {:?}", msg);
tracing::debug!(log_message);
logger(log, log_message);
std::thread::sleep(delay);
do_send(addr, msg, log)
}
pub fn do_send<M>(_addr: Addr<RepoActor>, msg: M, log: Option<&RepoActorLog>)
where
M: actix::Message + Send + 'static + std::fmt::Debug,
RepoActor: actix::Handler<M>,
<M as actix::Message>::Result: Send,
{
let log_message = format!("send: {:?}", msg);
tracing::debug!(log_message);
logger(log, log_message);
#[cfg(not(test))]
_addr.do_send(msg)
}
pub fn logger(log: Option<&RepoActorLog>, message: impl Into<String>) {
if let Some(log) = log {
let message: String = message.into();
tracing::debug!(message);
let _ = log.write().map(|mut l| l.push(message));
}
}
pub fn notify_user(
recipient: Option<&Recipient<NotifyUser>>,
user_notification: UserNotification,
log: Option<&RepoActorLog>,
) {
let msg = NotifyUser::from(user_notification);
let log_message = format!("send: {:?}", msg);
tracing::debug!(log_message);
logger(log, log_message);
if let Some(recipient) = &recipient {
recipient.do_send(msg);
}
}

View file

@ -1,12 +1,14 @@
//
use derive_more::Deref as _;
use crate::messages::NotifyUser;
use crate::repo::messages::NotifyUser;
use git_next_core::git::UserNotification;
use serde_json::json;
impl NotifyUser {
pub fn as_json(self, timestamp: time::OffsetDateTime) -> serde_json::Value {
pub fn as_json(&self, timestamp: time::OffsetDateTime) -> serde_json::Value {
let timestamp = timestamp.unix_timestamp().to_string();
match self.deref() {
UserNotification::CICheckFailed {

View file

@ -1,3 +1,5 @@
use crate::git;
//
use super::*;
@ -10,12 +12,11 @@ fn push_is_error_should_error() {
expect::fetch_ok(&mut open_repository);
expect::push(&mut open_repository, Err(git::push::Error::Lock));
let_assert!(
Err(err) =
crate::branch::advance_main(commit, &repo_details, &repo_config, &open_repository)
Err(err) = branch::advance_main(commit, &repo_details, &repo_config, &open_repository)
);
assert!(matches!(
err,
crate::branch::Error::Push(git::push::Error::Lock)
branch::Error::Push(crate::git::push::Error::Lock)
));
}
@ -27,7 +28,5 @@ fn push_is_ok_should_ok() {
let (mut open_repository, repo_details) = given::an_open_repository(&fs);
expect::fetch_ok(&mut open_repository);
expect::push_ok(&mut open_repository);
assert!(
crate::branch::advance_main(commit, &repo_details, &repo_config, &open_repository).is_ok()
);
assert!(branch::advance_main(commit, &repo_details, &repo_config, &open_repository).is_ok());
}

View file

@ -14,7 +14,7 @@ mod when_at_dev {
// no on_push defined - so any call to push will cause an error
let message_token = given::a_message_token();
let_assert!(
Err(err) = crate::branch::advance_next(
Err(err) = branch::advance_next(
&next,
dev_commit_history,
repo_details,
@ -24,7 +24,7 @@ mod when_at_dev {
)
);
tracing::debug!("Got: {err}");
assert!(matches!(err, crate::branch::Error::NextAtDev));
assert!(matches!(err, branch::Error::NextAtDev));
Ok(())
}
}
@ -32,6 +32,7 @@ mod when_at_dev {
mod can_advance {
// dev has at least one commit ahead of next
use super::*;
mod to_wip_commit {
// commit on dev is either invalid message or a WIP
use super::*;
@ -47,7 +48,7 @@ mod can_advance {
// no on_push defined - so any call to push will cause an error
let message_token = given::a_message_token();
let_assert!(
Err(err) = crate::branch::advance_next(
Err(err) = branch::advance_next(
&next,
dev_commit_history,
repo_details,
@ -57,7 +58,7 @@ mod can_advance {
)
);
tracing::debug!("Got: {err}");
assert!(matches!(err, crate::branch::Error::IsWorkInProgress));
assert!(matches!(err, branch::Error::IsWorkInProgress));
Ok(())
}
}
@ -77,7 +78,7 @@ mod can_advance {
// no on_push defined - so any call to push will cause an error
let message_token = given::a_message_token();
let_assert!(
Err(err) = crate::branch::advance_next(
Err(err) = branch::advance_next(
&next,
dev_commit_history,
repo_details,
@ -89,7 +90,7 @@ mod can_advance {
tracing::debug!("Got: {err}");
assert!(matches!(
err,
crate::branch::Error::InvalidCommitMessage{reason}
branch::Error::InvalidCommitMessage{reason}
if reason == "Missing type in the commit summary, expected `type: description`"
));
Ok(())
@ -116,7 +117,7 @@ mod can_advance {
expect::push(&mut open_repository, Err(git::push::Error::Lock));
let message_token = given::a_message_token();
let_assert!(
Err(err) = crate::branch::advance_next(
Err(err) = branch::advance_next(
&next,
dev_commit_history,
repo_details,
@ -126,10 +127,7 @@ mod can_advance {
)
);
tracing::debug!("Got: {err:?}");
assert!(matches!(
err,
crate::branch::Error::Push(git::push::Error::Lock)
));
assert!(matches!(err, branch::Error::Push(git::push::Error::Lock)));
Ok(())
}
}
@ -150,7 +148,7 @@ mod can_advance {
expect::push_ok(&mut open_repository);
let message_token = given::a_message_token();
let_assert!(
Ok(mt) = crate::branch::advance_next(
Ok(mt) = branch::advance_next(
&next,
dev_commit_history,
repo_details,

View file

@ -4,6 +4,9 @@ use super::*;
mod advance_main;
mod advance_next;
use crate::git;
use crate::repo::branch;
#[actix_rt::test]
async fn test_find_next_commit_on_dev() {
let next = given::a_commit();
@ -15,7 +18,7 @@ async fn test_find_next_commit_on_dev() {
given::a_commit(), // parent of next
];
let next_commit = crate::branch::find_next_commit_on_dev(&next, &dev_commit_history);
let next_commit = branch::find_next_commit_on_dev(&next, &dev_commit_history);
assert_eq!(next_commit, Some(expected), "Found the wrong commit");
}

View file

@ -1,3 +1,5 @@
use git_next_core::git::fetch;
//
use super::*;
@ -5,7 +7,7 @@ pub fn fetch_ok(open_repository: &mut MockOpenRepositoryLike) {
expect::fetch(open_repository, Ok(()));
}
pub fn fetch(open_repository: &mut MockOpenRepositoryLike, result: Result<(), git::fetch::Error>) {
pub fn fetch(open_repository: &mut MockOpenRepositoryLike, result: Result<(), fetch::Error>) {
open_repository
.expect_fetch()
.times(1)
@ -16,7 +18,10 @@ pub fn push_ok(open_repository: &mut MockOpenRepositoryLike) {
expect::push(open_repository, Ok(()))
}
pub fn push(open_repository: &mut MockOpenRepositoryLike, result: Result<(), git::push::Error>) {
pub fn push(
open_repository: &mut MockOpenRepositoryLike,
result: Result<(), crate::git::push::Error>,
) {
open_repository
.expect_push()
.times(1)
@ -36,7 +41,7 @@ pub fn open_repository(
pub fn main_commit_log(
validation_repo: &mut MockOpenRepositoryLike,
main_branch: BranchName,
) -> git::Commit {
) -> Commit {
let main_commit = given::a_commit();
let main_branch_log = vec![main_commit.clone()];
validation_repo

View file

@ -109,36 +109,36 @@ pub fn a_repo_config() -> RepoConfig {
RepoConfig::new(given::repo_branches(), RepoConfigSource::Repo)
}
pub fn a_named_commit(name: impl Into<String>) -> git::Commit {
git::Commit::new(a_named_commit_sha(name), a_commit_message())
pub fn a_named_commit(name: impl Into<String>) -> Commit {
Commit::new(a_named_commit_sha(name), a_commit_message())
}
pub fn a_commit() -> git::Commit {
git::Commit::new(a_commit_sha(), a_commit_message())
pub fn a_commit() -> Commit {
Commit::new(a_commit_sha(), a_commit_message())
}
pub fn a_commit_with_message(message: impl Into<git::commit::Message>) -> git::Commit {
git::Commit::new(a_commit_sha(), message.into())
pub fn a_commit_with_message(message: impl Into<crate::git::commit::Message>) -> Commit {
Commit::new(a_commit_sha(), message.into())
}
pub fn a_commit_message() -> git::commit::Message {
git::commit::Message::new(a_name())
pub fn a_commit_message() -> crate::git::commit::Message {
crate::git::commit::Message::new(a_name())
}
pub fn a_named_commit_sha(name: impl Into<String>) -> git::commit::Sha {
git::commit::Sha::new(format!("{}-{}", name.into(), a_name()))
pub fn a_named_commit_sha(name: impl Into<String>) -> Sha {
Sha::new(format!("{}-{}", name.into(), a_name()))
}
pub fn a_commit_sha() -> git::commit::Sha {
git::commit::Sha::new(a_name())
pub fn a_commit_sha() -> Sha {
Sha::new(a_name())
}
pub fn a_filesystem() -> kxio::fs::FileSystem {
kxio::fs::temp().unwrap_or_else(|e| panic!("{}", e))
}
pub fn repo_details(fs: &kxio::fs::FileSystem) -> git::RepoDetails {
let generation = git::Generation::default();
pub fn repo_details(fs: &kxio::fs::FileSystem) -> RepoDetails {
let generation = Generation::default();
let repo_alias = a_repo_alias();
let server_repo_config = a_server_repo_config();
let forge_alias = a_forge_alias();
@ -154,12 +154,7 @@ pub fn repo_details(fs: &kxio::fs::FileSystem) -> git::RepoDetails {
)
}
pub fn an_open_repository(
fs: &kxio::fs::FileSystem,
) -> (
git::repository::open::MockOpenRepositoryLike,
git::RepoDetails,
) {
pub fn an_open_repository(fs: &kxio::fs::FileSystem) -> (MockOpenRepositoryLike, RepoDetails) {
let open_repository = MockOpenRepositoryLike::new();
let gitdir = given::a_git_dir(fs);
let hostname = given::a_hostname();
@ -182,11 +177,11 @@ pub fn a_forge() -> Box<MockForgeLike> {
}
pub fn a_repo_actor(
repo_details: git::RepoDetails,
repo_details: RepoDetails,
repository_factory: Box<dyn RepositoryFactory>,
forge: Box<dyn git::ForgeLike>,
forge: Box<dyn ForgeLike>,
net: kxio::network::Network,
) -> (crate::RepoActor, RepoActorLog) {
) -> (RepoActor, RepoActorLog) {
let forge_alias = repo_details.forge.forge_alias();
let repo_alias = &repo_details.repo_alias;
let webhook_url = given::a_webhook_url(forge_alias, repo_alias);
@ -195,7 +190,7 @@ pub fn a_repo_actor(
let log = RepoActorLog::default();
let actors_log = log.clone();
(
crate::RepoActor::new(
RepoActor::new(
repo_details,
forge,
webhook,
@ -218,8 +213,8 @@ pub fn a_registered_webhook() -> RegisteredWebhook {
RegisteredWebhook::new(given::a_webhook_id(), given::a_webhook_auth())
}
pub fn a_push() -> webhook::Push {
webhook::Push::new(
pub fn a_push() -> Push {
Push::new(
given::a_branch_name("push"),
given::a_name(),
given::a_name(),

View file

@ -32,7 +32,7 @@ async fn when_repo_config_should_fetch_then_push_then_revalidate() -> TestResult
repo_details,
given::a_forge(),
);
addr.send(crate::messages::AdvanceMain::new(next_commit.clone()))
addr.send(crate::repo::messages::AdvanceMain::new(next_commit.clone()))
.await?;
System::current().stop();
@ -77,7 +77,7 @@ async fn when_server_config_should_fetch_then_push_then_revalidate() -> TestResu
repo_details,
given::a_forge(),
);
addr.send(crate::messages::AdvanceMain::new(next_commit.clone()))
addr.send(crate::repo::messages::AdvanceMain::new(next_commit.clone()))
.await?;
System::current().stop();

View file

@ -31,7 +31,7 @@ async fn should_fetch_then_push_then_revalidate() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::AdvanceNext::new((
addr.send(crate::repo::messages::AdvanceNext::new((
next_commit.clone(),
dev_commit_log,
)))

View file

@ -20,8 +20,10 @@ async fn should_passthrough_to_receive_ci_status() -> TestResult {
repo_details,
Box::new(forge),
);
addr.send(crate::messages::CheckCIStatus::new(next_commit.clone()))
.await?;
addr.send(crate::repo::messages::CheckCIStatus::new(
next_commit.clone(),
))
.await?;
System::current().stop();
//then

View file

@ -39,7 +39,7 @@ async fn when_read_file_ok_should_send_config_loaded() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::LoadConfigFromRepo::new())
addr.send(crate::repo::messages::LoadConfigFromRepo::new())
.await?;
System::current().stop();
@ -70,7 +70,7 @@ async fn when_read_file_err_should_notify_user() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::LoadConfigFromRepo::new())
addr.send(crate::repo::messages::LoadConfigFromRepo::new())
.await?;
System::current().stop();

View file

@ -15,7 +15,7 @@ async fn should_store_repo_config_in_actor() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ReceiveRepoConfig::new(
addr.send(crate::repo::messages::ReceiveRepoConfig::new(
new_repo_config.clone(),
))
.await?;
@ -46,7 +46,7 @@ async fn should_register_webhook() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ReceiveRepoConfig::new(
addr.send(crate::repo::messages::ReceiveRepoConfig::new(
new_repo_config.clone(),
))
.await?;

View file

@ -14,7 +14,7 @@ async fn when_pass_should_advance_main_to_next() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ReceiveCIStatus::new((
addr.send(crate::repo::messages::ReceiveCIStatus::new((
next_commit.clone(),
git::forge::commit::Status::Pass,
)))
@ -44,7 +44,7 @@ async fn when_pending_should_recheck_ci_status() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ReceiveCIStatus::new((
addr.send(crate::repo::messages::ReceiveCIStatus::new((
next_commit.clone(),
git::forge::commit::Status::Pending,
)))
@ -74,12 +74,12 @@ async fn when_fail_should_recheck_after_delay() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ReceiveCIStatus::new((
addr.send(crate::repo::messages::ReceiveCIStatus::new((
next_commit.clone(),
git::forge::commit::Status::Fail,
)))
.await?;
tokio::time::sleep(std::time::Duration::from_millis(2)).await;
actix_rt::time::sleep(std::time::Duration::from_millis(2)).await;
System::current().stop();
//then
@ -100,7 +100,7 @@ async fn when_fail_should_notify_user() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ReceiveCIStatus::new((
addr.send(crate::repo::messages::ReceiveCIStatus::new((
next_commit.clone(),
git::forge::commit::Status::Fail,
)))

View file

@ -22,7 +22,8 @@ async fn when_registered_ok_should_send_webhook_registered() -> TestResult {
repo_details,
Box::new(forge),
);
addr.send(crate::messages::RegisterWebhook::new()).await?;
addr.send(crate::repo::messages::RegisterWebhook::new())
.await?;
System::current().stop();
//then
@ -57,7 +58,8 @@ async fn when_registered_error_should_send_notify_user() -> TestResult {
repo_details,
Box::new(forge),
);
addr.send(crate::messages::RegisterWebhook::new()).await?;
addr.send(crate::repo::messages::RegisterWebhook::new())
.await?;
System::current().stop();
//then

View file

@ -38,8 +38,10 @@ async fn repo_with_next_not_an_ancestor_of_dev_should_be_reset() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ValidateRepo::new(MessageToken::default()))
.await?;
addr.send(crate::repo::messages::ValidateRepo::new(
MessageToken::default(),
))
.await?;
System::current().stop();
//then
@ -84,8 +86,10 @@ async fn repo_with_next_not_on_or_near_main_should_be_reset() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ValidateRepo::new(MessageToken::default()))
.await?;
addr.send(crate::repo::messages::ValidateRepo::new(
MessageToken::default(),
))
.await?;
System::current().stop();
//then
@ -130,8 +134,10 @@ async fn repo_with_next_not_based_on_main_should_be_reset() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ValidateRepo::new(MessageToken::default()))
.await?;
addr.send(crate::repo::messages::ValidateRepo::new(
MessageToken::default(),
))
.await?;
System::current().stop();
//then
@ -175,8 +181,10 @@ async fn repo_with_next_ahead_of_main_should_check_ci_status() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ValidateRepo::new(MessageToken::default()))
.await?;
addr.send(crate::repo::messages::ValidateRepo::new(
MessageToken::default(),
))
.await?;
System::current().stop();
//then
@ -221,8 +229,10 @@ async fn repo_with_dev_and_next_on_main_should_do_nothing() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ValidateRepo::new(MessageToken::default()))
.await?;
addr.send(crate::repo::messages::ValidateRepo::new(
MessageToken::default(),
))
.await?;
System::current().stop();
//then
@ -268,8 +278,10 @@ async fn repo_with_dev_ahead_of_next_should_advance_next() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ValidateRepo::new(MessageToken::default()))
.await?;
addr.send(crate::repo::messages::ValidateRepo::new(
MessageToken::default(),
))
.await?;
System::current().stop();
//then
@ -317,8 +329,10 @@ async fn repo_with_dev_not_ahead_of_main_should_notify_user() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ValidateRepo::new(MessageToken::default()))
.await?;
addr.send(crate::repo::messages::ValidateRepo::new(
MessageToken::default(),
))
.await?;
System::current().stop();
//then
@ -341,8 +355,10 @@ async fn should_accept_message_with_current_token() -> TestResult {
);
let actor = actor.with_message_token(MessageToken::new(2_u32));
let addr = actor.start();
addr.send(crate::messages::ValidateRepo::new(MessageToken::new(2_u32)))
.await?;
addr.send(crate::repo::messages::ValidateRepo::new(MessageToken::new(
2_u32,
)))
.await?;
System::current().stop();
//then
@ -365,8 +381,10 @@ async fn should_accept_message_with_new_token() -> TestResult {
);
let actor = actor.with_message_token(MessageToken::new(2_u32));
let addr = actor.start();
addr.send(crate::messages::ValidateRepo::new(MessageToken::new(3_u32)))
.await?;
addr.send(crate::repo::messages::ValidateRepo::new(MessageToken::new(
3_u32,
)))
.await?;
System::current().stop();
//then
@ -389,8 +407,10 @@ async fn should_reject_message_with_expired_token() -> TestResult {
);
let actor = actor.with_message_token(MessageToken::new(4_u32));
let addr = actor.start();
addr.send(crate::messages::ValidateRepo::new(MessageToken::new(3_u32)))
.await?;
addr.send(crate::repo::messages::ValidateRepo::new(MessageToken::new(
3_u32,
)))
.await?;
System::current().stop();
//then
@ -415,9 +435,11 @@ async fn should_send_validate_repo_when_retryable_error() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ValidateRepo::new(MessageToken::default()))
.await?;
tokio::time::sleep(std::time::Duration::from_millis(2)).await;
addr.send(crate::repo::messages::ValidateRepo::new(
MessageToken::default(),
))
.await?;
actix_rt::time::sleep(std::time::Duration::from_millis(2)).await;
System::current().stop();
//then
@ -462,9 +484,11 @@ async fn should_send_notify_user_when_non_retryable_error() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ValidateRepo::new(MessageToken::default()))
.await?;
tokio::time::sleep(std::time::Duration::from_millis(1)).await;
addr.send(crate::repo::messages::ValidateRepo::new(
MessageToken::default(),
))
.await?;
actix_rt::time::sleep(std::time::Duration::from_millis(1)).await;
System::current().stop();
//then

View file

@ -23,7 +23,7 @@ async fn when_no_expected_auth_token_drop_notification() -> TestResult {
//when
actor
.start()
.send(crate::messages::WebhookNotification::new(
.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;
@ -57,7 +57,7 @@ async fn when_no_repo_config_drop_notification() -> TestResult {
//when
actor
.start()
.send(crate::messages::WebhookNotification::new(
.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;
@ -95,7 +95,7 @@ async fn when_message_auth_is_invalid_drop_notification() -> TestResult {
//when
actor
.start()
.send(crate::messages::WebhookNotification::new(
.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;
@ -137,7 +137,7 @@ async fn when_message_is_ignorable_drop_notification() -> TestResult {
//when
actor
.start()
.send(crate::messages::WebhookNotification::new(
.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;
@ -179,7 +179,7 @@ async fn when_message_is_not_a_push_drop_notification() -> TestResult {
//when
actor
.start()
.send(crate::messages::WebhookNotification::new(
.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;
@ -227,7 +227,7 @@ async fn when_message_is_push_on_unknown_branch_drop_notification() -> TestResul
//when
actor
.start()
.send(crate::messages::WebhookNotification::new(
.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;
@ -276,7 +276,7 @@ async fn when_message_is_push_already_seen_commit_to_main() -> TestResult {
//when
actor
.start()
.send(crate::messages::WebhookNotification::new(
.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;
@ -325,7 +325,7 @@ async fn when_message_is_push_already_seen_commit_to_next() -> TestResult {
//when
actor
.start()
.send(crate::messages::WebhookNotification::new(
.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;
@ -374,7 +374,7 @@ async fn when_message_is_push_already_seen_commit_to_dev() -> TestResult {
//when
actor
.start()
.send(crate::messages::WebhookNotification::new(
.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;
@ -421,7 +421,7 @@ async fn when_message_is_push_new_commit_to_main_should_stash_and_validate_repo(
//when
let addr = actor.start();
addr.send(crate::messages::WebhookNotification::new(
addr.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;
@ -469,7 +469,7 @@ async fn when_message_is_push_new_commit_to_next_should_stash_and_validate_repo(
//when
let addr = actor.start();
addr.send(crate::messages::WebhookNotification::new(
addr.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;
@ -517,7 +517,7 @@ async fn when_message_is_push_new_commit_to_dev_should_stash_and_validate_repo()
//when
let addr = actor.start();
addr.send(crate::messages::WebhookNotification::new(
addr.send(crate::repo::messages::WebhookNotification::new(
forge_notification,
))
.await?;

View file

@ -15,7 +15,7 @@ async fn should_store_webhook_details() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::WebhookRegistered::new((
addr.send(crate::repo::messages::WebhookRegistered::new((
webhook_id.clone(),
webhook_auth.clone(),
)))
@ -43,7 +43,7 @@ async fn should_send_validate_repo_message() -> TestResult {
repo_details,
given::a_forge(),
);
addr.send(crate::messages::WebhookRegistered::new((
addr.send(crate::repo::messages::WebhookRegistered::new((
webhook_id.clone(),
webhook_auth.clone(),
)))

View file

@ -1,28 +1,28 @@
//
use super::*;
#[tokio::test]
use crate::git::file;
use crate::repo::load;
#[actix::test]
async fn when_file_not_found_should_error() -> TestResult {
//given
let fs = given::a_filesystem();
let (mut open_repository, repo_details) = given::an_open_repository(&fs);
open_repository
.expect_read_file()
.returning(|_, _| Err(git::file::Error::FileNotFound));
.returning(|_, _| Err(file::Error::FileNotFound));
//when
let_assert!(
Err(err) = crate::load::config_from_repository(repo_details, &open_repository).await
);
let_assert!(Err(err) = load::config_from_repository(repo_details, &open_repository).await);
//then
tracing::debug!("Got: {err:?}");
debug!("Got: {err:?}");
assert!(matches!(
err,
crate::load::Error::File(git::file::Error::FileNotFound)
load::Error::File(crate::git::file::Error::FileNotFound)
));
Ok(())
}
#[tokio::test]
#[actix::test]
async fn when_file_format_invalid_should_error() -> TestResult {
//given
let fs = given::a_filesystem();
@ -32,16 +32,14 @@ async fn when_file_format_invalid_should_error() -> TestResult {
.expect_read_file()
.return_once(move |_, _| Ok(contents));
//when
let_assert!(
Err(err) = crate::load::config_from_repository(repo_details, &open_repository).await
);
let_assert!(Err(err) = load::config_from_repository(repo_details, &open_repository).await);
//then
tracing::debug!("Got: {err:?}");
assert!(matches!(err, crate::load::Error::Toml(_)));
debug!("Got: {err:?}");
assert!(matches!(err, load::Error::Toml(_)));
Ok(())
}
#[tokio::test]
#[actix::test]
async fn when_main_branch_is_missing_should_error() -> TestResult {
//given
let fs = given::a_filesystem();
@ -66,16 +64,14 @@ async fn when_main_branch_is_missing_should_error() -> TestResult {
.expect_remote_branches()
.return_once(move || Ok(branches));
//when
let_assert!(
Err(err) = crate::load::config_from_repository(repo_details, &open_repository).await
);
let_assert!(Err(err) = load::config_from_repository(repo_details, &open_repository).await);
//then
tracing::debug!("Got: {err:?}");
assert!(matches!(err, crate::load::Error::BranchNotFound(branch) if branch == main));
debug!("Got: {err:?}");
assert!(matches!(err, load::Error::BranchNotFound(branch) if branch == main));
Ok(())
}
#[tokio::test]
#[actix::test]
async fn when_next_branch_is_missing_should_error() -> TestResult {
//given
let fs = given::a_filesystem();
@ -100,16 +96,14 @@ async fn when_next_branch_is_missing_should_error() -> TestResult {
.expect_remote_branches()
.return_once(move || Ok(branches));
//when
let_assert!(
Err(err) = crate::load::config_from_repository(repo_details, &open_repository).await
);
let_assert!(Err(err) = load::config_from_repository(repo_details, &open_repository).await);
//then
tracing::debug!("Got: {err:?}");
assert!(matches!(err, crate::load::Error::BranchNotFound(branch) if branch == next));
debug!("Got: {err:?}");
assert!(matches!(err, load::Error::BranchNotFound(branch) if branch == next));
Ok(())
}
#[tokio::test]
#[actix::test]
async fn when_dev_branch_is_missing_should_error() -> TestResult {
//given
let fs = given::a_filesystem();
@ -134,16 +128,14 @@ async fn when_dev_branch_is_missing_should_error() -> TestResult {
.expect_remote_branches()
.return_once(move || Ok(branches));
//when
let_assert!(
Err(err) = crate::load::config_from_repository(repo_details, &open_repository).await
);
let_assert!(Err(err) = load::config_from_repository(repo_details, &open_repository).await);
//then
tracing::debug!("Got: {err:?}");
assert!(matches!(err, crate::load::Error::BranchNotFound(branch) if branch == dev));
debug!("Got: {err:?}");
assert!(matches!(err, load::Error::BranchNotFound(branch) if branch == dev));
Ok(())
}
#[tokio::test]
#[actix::test]
async fn when_valid_file_should_return_repo_config() -> TestResult {
//given
let fs = given::a_filesystem();
@ -169,11 +161,9 @@ async fn when_valid_file_should_return_repo_config() -> TestResult {
.expect_remote_branches()
.return_once(move || Ok(branches));
//when
let_assert!(
Ok(result) = crate::load::config_from_repository(repo_details, &open_repository).await
);
let_assert!(Ok(result) = load::config_from_repository(repo_details, &open_repository).await);
//then
tracing::debug!("Got: {result:?}");
debug!("Got: {result:?}");
assert_eq!(result, repo_config);
Ok(())
}

View file

@ -2,22 +2,27 @@
use actix::prelude::*;
use crate::{
messages::{CloneRepo, MessageToken},
RepoActor, RepoActorLog,
git,
repo::{
messages::{CloneRepo, MessageToken},
RepoActor, RepoActorLog,
},
};
use git_next_core::{
git::{
self,
commit::Sha,
forge::commit::Status,
repository::{
factory::{MockRepositoryFactory, RepositoryFactory},
factory::{mock, MockRepositoryFactory, RepositoryFactory},
open::{MockOpenRepositoryLike, OpenRepositoryLike},
Direction,
},
Generation, MockForgeLike, RepoDetails,
Commit, ForgeLike, Generation, MockForgeLike, RepoDetails,
},
message,
server::{InboundWebhook, WebhookUrl},
webhook::{self, forge_notification::Body},
webhook::{forge_notification::Body, Push},
BranchName, ForgeAlias, ForgeConfig, ForgeNotification, ForgeType, GitDir, Hostname,
RegisteredWebhook, RemoteUrl, RepoAlias, RepoBranches, RepoConfig, RepoConfigSource,
ServerRepoConfig, StoragePathType, WebhookAuth, WebhookId,
@ -25,6 +30,7 @@ use git_next_core::{
use assert2::let_assert;
use mockall::predicate::eq;
use tracing::{debug, error};
use std::{
collections::{BTreeMap, HashMap},
@ -43,7 +49,7 @@ mod when;
impl RepoActorLog {
pub fn no_message_contains(&self, needle: impl AsRef<str> + std::fmt::Display) -> TestResult {
if self.find_in_messages(needle.as_ref())? {
tracing::error!(?self, "");
error!(?self, "");
panic!("found unexpected message: {needle}");
}
Ok(())
@ -54,7 +60,7 @@ impl RepoActorLog {
needle: impl AsRef<str> + std::fmt::Display,
) -> TestResult {
if !self.find_in_messages(needle.as_ref())? {
tracing::error!(?self, "");
error!(?self, "");
panic!("expected message not found: {needle}");
}
Ok(())
@ -84,32 +90,22 @@ impl Handler<ExamineActor> for RepoActor {
}
#[derive(Debug, MessageResponse)]
pub struct RepoActorView {
pub sleep_duration: std::time::Duration,
pub generation: git::Generation,
pub message_token: MessageToken,
pub repo_details: git::RepoDetails,
pub webhook: InboundWebhook,
pub repo_details: RepoDetails,
pub webhook_id: Option<WebhookId>, // INFO: if [None] then no webhook is configured
pub webhook_auth: Option<WebhookAuth>, // INFO: if [None] then no webhook is configured
pub last_main_commit: Option<git::Commit>,
pub last_next_commit: Option<git::Commit>,
pub last_dev_commit: Option<git::Commit>,
pub log: Option<RepoActorLog>,
pub last_main_commit: Option<Commit>,
pub last_next_commit: Option<Commit>,
pub last_dev_commit: Option<Commit>,
}
impl From<&RepoActor> for RepoActorView {
fn from(repo_actor: &RepoActor) -> Self {
Self {
sleep_duration: repo_actor.sleep_duration,
generation: repo_actor.generation,
message_token: repo_actor.message_token,
repo_details: repo_actor.repo_details.clone(),
webhook: repo_actor.webhook.clone(),
webhook_id: repo_actor.webhook_id.clone(),
webhook_auth: repo_actor.webhook_auth.clone(),
last_main_commit: repo_actor.last_main_commit.clone(),
last_next_commit: repo_actor.last_next_commit.clone(),
last_dev_commit: repo_actor.last_dev_commit.clone(),
log: repo_actor.log.clone(),
}
}
}

View file

@ -2,9 +2,9 @@
use super::*;
pub fn start_actor(
repository_factory: git::repository::factory::MockRepositoryFactory,
repo_details: git::RepoDetails,
forge: Box<dyn git::ForgeLike>,
repository_factory: MockRepositoryFactory,
repo_details: RepoDetails,
forge: Box<dyn ForgeLike>,
) -> (actix::Addr<RepoActor>, RepoActorLog) {
let (actor, log) = given::a_repo_actor(
repo_details,
@ -17,25 +17,16 @@ pub fn start_actor(
pub fn start_actor_with_open_repository(
open_repository: Box<dyn OpenRepositoryLike>,
repo_details: git::RepoDetails,
forge: Box<dyn git::ForgeLike>,
repo_details: RepoDetails,
forge: Box<dyn ForgeLike>,
) -> (actix::Addr<RepoActor>, RepoActorLog) {
let (actor, log) = given::a_repo_actor(
repo_details,
git::repository::factory::mock(),
forge,
given::a_network().into(),
);
let (actor, log) = given::a_repo_actor(repo_details, mock(), forge, given::a_network().into());
let actor = actor.with_open_repository(Some(open_repository));
(actor.start(), log)
}
pub fn commit_status(
forge: &mut git::MockForgeLike,
commit: git::Commit,
status: git::forge::commit::Status,
) {
let mut commit_status_forge = git::MockForgeLike::new();
pub fn commit_status(forge: &mut MockForgeLike, commit: Commit, status: Status) {
let mut commit_status_forge = MockForgeLike::new();
commit_status_forge
.expect_commit_status()
.with(mockall::predicate::eq(commit))

View file

@ -5,9 +5,9 @@ use secrecy::ExposeSecret;
use standardwebhooks::Webhook;
use tracing::Instrument;
use crate::server::actor::ServerActor;
use crate::{repo::messages::NotifyUser, server::actor::ServerActor};
use git_next_core::server::{self, Notification, NotificationType};
use git_next_repo_actor::messages::NotifyUser;
impl Handler<NotifyUser> for ServerActor {
type Result = ();

View file

@ -1,6 +1,8 @@
//
use actix::prelude::*;
use tracing::info;
use crate::{
server::actor::{
messages::{ReceiveValidServerConfig, ValidServerConfig},
@ -28,7 +30,7 @@ impl Handler<ReceiveValidServerConfig> for ServerActor {
}
self.generation.inc();
// Webhook Server
tracing::info!("Starting Webhook Server...");
info!("Starting Webhook Server...");
let webhook_router = WebhookRouter::default().start();
let inbound_webhook = server_config.inbound_webhook();
// Forge Actors

View file

@ -1,8 +1,10 @@
//-
use actix::prelude::*;
use tracing::debug;
use crate::{
repo::messages::UnRegisterWebhook,
server::actor::{messages::Shutdown, ServerActor},
webhook::messages::ShutdownWebhook,
};
@ -14,15 +16,15 @@ impl Handler<Shutdown> for ServerActor {
self.repo_actors
.iter()
.for_each(|((forge_alias, repo_alias), addr)| {
tracing::debug!(%forge_alias, %repo_alias, "removing webhook");
addr.do_send(git_next_repo_actor::messages::UnRegisterWebhook::new());
tracing::debug!(%forge_alias, %repo_alias, "removed webhook");
debug!(%forge_alias, %repo_alias, "removing webhook");
addr.do_send(UnRegisterWebhook::new());
debug!(%forge_alias, %repo_alias, "removed webhook");
});
tracing::debug!("server shutdown");
debug!("server shutdown");
if let Some(webhook) = self.webhook_actor_addr.take() {
tracing::debug!("shuting down webhook");
debug!("shutting down webhook");
webhook.do_send(ShutdownWebhook);
tracing::debug!("webhook shutdown");
debug!("webhook shutdown");
}
}
}

View file

@ -1,5 +1,7 @@
//
use actix::prelude::*;
use messages::ReceiveServerConfig;
use tracing::error;
#[cfg(test)]
mod tests;
@ -7,16 +9,19 @@ mod tests;
mod handlers;
pub mod messages;
use crate::webhook::WebhookActor;
use crate::{
repo::{
messages::{CloneRepo, NotifyUser},
RepoActor,
},
webhook::WebhookActor,
};
use git_next_core::{
git::{repository::factory::RepositoryFactory, Generation, RepoDetails},
server::{self, InboundWebhook, ServerConfig, ServerStorage},
ForgeAlias, ForgeConfig, GitDir, RepoAlias, ServerRepoConfig, StoragePathType,
};
use git_next_repo_actor::{
messages::{CloneRepo, NotifyUser},
RepoActor,
};
use kxio::{fs::FileSystem, network::Network};
@ -26,10 +31,6 @@ use std::{
sync::{Arc, RwLock},
};
use tracing::{error, info};
use messages::ReceiveServerConfig;
#[derive(Debug, derive_more::Display, derive_more::From)]
pub enum Error {
#[display("Failed to create data directories")]
@ -98,7 +99,7 @@ impl ServerActor {
return Err(Error::ForgeDirIsNotDirectory { path });
}
} else {
info!(%forge_name, ?path, "creating storage");
tracing::info!(%forge_name, ?path, "creating storage");
self.fs.dir_create_all(&path)?;
}
}
@ -118,7 +119,7 @@ impl ServerActor {
tracing::info_span!("create_forge_repos", name = %forge_name, config = %forge_config);
let _guard = span.enter();
info!("Creating Forge");
tracing::info!("Creating Forge");
let mut repos = vec![];
let creator = self.create_actor(forge_name, forge_config.clone(), server_storage, webhook);
for (repo_alias, server_repo_config) in forge_config.repos() {
@ -127,7 +128,7 @@ impl ServerActor {
server_repo_config,
notify_user_recipient.clone(),
));
info!(
tracing::info!(
alias = %forge_repo.1,
"Created Repo"
);
@ -155,7 +156,7 @@ impl ServerActor {
move |(repo_alias, server_repo_config, notify_user_recipient)| {
let span = tracing::info_span!("create_actor", alias = %repo_alias, config = %server_repo_config);
let _guard = span.enter();
info!("Creating Repo");
tracing::info!("Creating Repo");
let gitdir = server_repo_config.gitdir().map_or_else(
|| {
GitDir::new(
@ -179,7 +180,7 @@ impl ServerActor {
gitdir,
);
let forge = git_next_forge::Forge::create(repo_details.clone(), net.clone());
info!("Starting Repo Actor");
tracing::info!("Starting Repo Actor");
let actor = RepoActor::new(
repo_details,
forge,
@ -203,7 +204,7 @@ impl ServerActor {
let _guard = span.enter();
let addr = actor.start();
addr.do_send(CloneRepo);
info!("Started");
tracing::info!("Started");
(repo_alias, addr)
}

View file

@ -6,7 +6,7 @@ pub mod messages;
pub mod router;
mod server;
use git_next_repo_actor::messages::WebhookNotification;
use crate::repo::messages::WebhookNotification;
use std::net::SocketAddr;

View file

@ -2,11 +2,13 @@
use actix::prelude::*;
use derive_more::Constructor;
use tracing::{debug, info, warn};
use std::collections::BTreeMap;
use tracing::info;
use crate::repo::messages::WebhookNotification;
use git_next_core::{ForgeAlias, RepoAlias};
use git_next_repo_actor::messages::WebhookNotification;
pub struct WebhookRouter {
span: tracing::Span,
@ -37,13 +39,13 @@ impl Handler<WebhookNotification> for WebhookRouter {
let _gaurd = self.span.enter();
let forge_alias = msg.forge_alias();
let repo_alias = msg.repo_alias();
tracing::debug!(forge = %forge_alias, repo = %repo_alias, "Router...");
debug!(forge = %forge_alias, repo = %repo_alias, "Router...");
let Some(forge_repos) = self.recipients.get(forge_alias) else {
tracing::warn!(forge = %forge_alias, "No forge repos found");
warn!(forge = %forge_alias, "No forge repos found");
return;
};
let Some(recipient) = forge_repos.get(repo_alias) else {
tracing::debug!(forge = %forge_alias, repo = %repo_alias, "No recipient found");
debug!(forge = %forge_alias, repo = %repo_alias, "No recipient found");
return;
};
recipient.do_send(msg);

View file

@ -5,8 +5,9 @@ use std::{collections::BTreeMap, net::SocketAddr};
use tracing::{info, warn};
use crate::repo::messages::WebhookNotification;
use git_next_core::{webhook, ForgeAlias, ForgeNotification, RepoAlias};
use git_next_repo_actor::messages::WebhookNotification;
pub async fn start(
socket_addr: SocketAddr,

View file

@ -4,4 +4,4 @@ version = { workspace = true }
edition = { workspace = true }
license = { workspace = true }
repository = { workspace = true }
description = "Config file watcher for git-next, the trunk-based development manager"
description = "[deprecated crate] Config file watcher for git-next, the trunk-based development manager"

View file

@ -4,65 +4,4 @@ version = { workspace = true }
edition = { workspace = true }
license = { workspace = true }
repository = { workspace = true }
description = "Repository support for git-next, the trunk-based development manager"
[features]
default = ["forgejo", "github"]
forgejo = []
github = []
[dependencies]
git-next-core = { workspace = true }
git-next-forge = { workspace = true }
# logging
tracing = { workspace = true }
# base64 decoding
base64 = { workspace = true }
# git
async-trait = { workspace = true }
# fs/network
kxio = { workspace = true }
# TOML parsing
serde = { workspace = true }
serde_json = { workspace = true }
toml = { workspace = true }
# Secrets and Password
secrecy = { workspace = true }
# Conventional Commit check
git-conventional = { workspace = true }
# Webhooks
bytes = { workspace = true }
ulid = { workspace = true }
time = { workspace = true }
# boilerplate
derive_more = { workspace = true }
derive-with = { workspace = true }
thiserror = { workspace = true }
# Actors
actix = { workspace = true }
actix-rt = { workspace = true }
tokio = { workspace = true }
[dev-dependencies]
# Testing
assert2 = { workspace = true }
rand = { workspace = true }
pretty_assertions = { workspace = true }
mockall = { workspace = true }
test-log = { workspace = true }
[lints.clippy]
nursery = { level = "warn", priority = -1 }
# pedantic = "warn"
unwrap_used = "warn"
expect_used = "warn"
description = "[deprecated crate] Repository support for git-next, the trunk-based development manager"

View file

@ -7,3 +7,5 @@ development workflows where each commit must pass CI before being included in
the main branch.
See [git-next](https://crates.io/crates/git-next) for more information.
N.B. this crate has been merged into [git-next](https://crates.io/git-next).

View file

@ -1,34 +0,0 @@
//
use crate as actor;
use actix::prelude::*;
use tracing::Instrument as _;
impl Handler<actor::messages::CheckCIStatus> for actor::RepoActor {
type Result = ();
fn handle(
&mut self,
msg: actor::messages::CheckCIStatus,
ctx: &mut Self::Context,
) -> Self::Result {
actor::logger(self.log.as_ref(), "start: CheckCIStatus");
let addr = ctx.address();
let forge = self.forge.duplicate();
let next = msg.unwrap();
let log = self.log.clone();
// get the status - pass, fail, pending (all others map to fail, e.g. error)
async move {
let status = forge.commit_status(&next).await;
tracing::debug!("got status: {status:?}");
actor::do_send(
addr,
actor::messages::ReceiveCIStatus::new((next, status)),
log.as_ref(),
);
}
.in_current_span()
.into_actor(self)
.wait(ctx);
}
}

View file

@ -1,43 +0,0 @@
//
use actix::prelude::*;
use git_next_core::git;
impl Handler<crate::messages::CloneRepo> for crate::RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::CloneRepo", skip_all, fields(repo = %self.repo_details /*, gitdir = %self.repo_details.gitdir */))]
fn handle(
&mut self,
_msg: crate::messages::CloneRepo,
ctx: &mut Self::Context,
) -> Self::Result {
crate::logger(self.log.as_ref(), "Handler: CloneRepo: start");
tracing::debug!("Handler: CloneRepo: start");
match git::repository::open(&*self.repository_factory, &self.repo_details) {
Ok(repository) => {
crate::logger(self.log.as_ref(), "open okay");
tracing::debug!("open okay");
self.open_repository.replace(repository);
if self.repo_details.repo_config.is_none() {
crate::do_send(
ctx.address(),
crate::messages::LoadConfigFromRepo,
self.log.as_ref(),
);
} else {
crate::do_send(
ctx.address(),
crate::messages::RegisterWebhook::new(),
self.log.as_ref(),
);
}
}
Err(err) => {
crate::logger(self.log.as_ref(), "open failed");
tracing::debug!("err: {err:?}");
tracing::warn!("Could not open repo: {err}")
}
}
tracing::debug!("Handler: CloneRepo: finish");
}
}

View file

@ -1,60 +0,0 @@
//
use actix::prelude::*;
use git_next_core::git::{self, UserNotification};
impl Handler<crate::messages::ReceiveCIStatus> for crate::RepoActor {
type Result = ();
fn handle(
&mut self,
msg: crate::messages::ReceiveCIStatus,
ctx: &mut Self::Context,
) -> Self::Result {
let log = self.log.clone();
crate::logger(log.as_ref(), "start: ReceiveCIStatus");
let addr = ctx.address();
let (next, status) = msg.unwrap();
let forge_alias = self.repo_details.forge.forge_alias().clone();
let repo_alias = self.repo_details.repo_alias.clone();
let message_token = self.message_token;
let sleep_duration = self.sleep_duration;
tracing::debug!(?status, "");
match status {
git::forge::commit::Status::Pass => {
crate::do_send(
addr,
crate::messages::AdvanceMain::new(next),
self.log.as_ref(),
);
}
git::forge::commit::Status::Pending => {
std::thread::sleep(sleep_duration);
crate::do_send(
addr,
crate::messages::ValidateRepo::new(message_token),
self.log.as_ref(),
);
}
git::forge::commit::Status::Fail => {
tracing::warn!("Checks have failed");
crate::notify_user(
self.notify_user_recipient.as_ref(),
UserNotification::CICheckFailed {
forge_alias,
repo_alias,
commit: next,
},
log.as_ref(),
);
crate::delay_send(
addr,
sleep_duration,
crate::messages::ValidateRepo::new(message_token),
self.log.as_ref(),
);
}
}
}
}

View file

@ -1,23 +0,0 @@
//
use actix::prelude::*;
use crate as actor;
impl Handler<actor::messages::ReceiveRepoConfig> for actor::RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::ReceiveRepoConfig", skip_all, fields(repo = %self.repo_details, branches = ?msg))]
fn handle(
&mut self,
msg: actor::messages::ReceiveRepoConfig,
ctx: &mut Self::Context,
) -> Self::Result {
let repo_config = msg.unwrap();
self.repo_details.repo_config.replace(repo_config);
actor::do_send(
ctx.address(),
actor::messages::RegisterWebhook::new(),
self.log.as_ref(),
);
}
}

View file

@ -1,134 +0,0 @@
//
use actix::prelude::*;
use derive_more::Deref as _;
use tracing::Instrument as _;
use crate::messages::MessageToken;
use git_next_core::git;
impl Handler<crate::messages::ValidateRepo> for crate::RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::ValidateRepo", skip_all, fields(repo = %self.repo_details, token = %msg.deref()))]
fn handle(
&mut self,
msg: crate::messages::ValidateRepo,
ctx: &mut Self::Context,
) -> Self::Result {
crate::logger(self.log.as_ref(), "start: ValidateRepo");
// Message Token - make sure we are only triggered for the latest/current token
match self.token_status(msg.unwrap()) {
TokenStatus::Current => {} // do nothing
TokenStatus::Expired => {
crate::logger(
self.log.as_ref(),
format!("discarded: old message token: {}", self.message_token),
);
return; // message is expired
}
TokenStatus::New(message_token) => {
self.message_token = message_token;
crate::logger(
self.log.as_ref(),
format!("new message token: {}", self.message_token),
);
}
}
crate::logger(
self.log.as_ref(),
format!("accepted token: {}", self.message_token),
);
// Repository positions
let Some(ref open_repository) = self.open_repository else {
crate::logger(self.log.as_ref(), "no open repository");
return;
};
crate::logger(self.log.as_ref(), "have open repository");
let Some(repo_config) = self.repo_details.repo_config.clone() else {
crate::logger(self.log.as_ref(), "no repo config");
return;
};
crate::logger(self.log.as_ref(), "have repo config");
match git::validation::positions::validate_positions(
&**open_repository,
&self.repo_details,
repo_config,
) {
Ok(git::validation::positions::Positions {
main,
next,
dev,
dev_commit_history,
}) => {
tracing::debug!(%main, %next, %dev, "positions");
if next != main {
crate::do_send(
ctx.address(),
crate::messages::CheckCIStatus::new(next),
self.log.as_ref(),
);
} else if next != dev {
crate::do_send(
ctx.address(),
crate::messages::AdvanceNext::new((next, dev_commit_history)),
self.log.as_ref(),
)
} else {
// do nothing
}
}
Err(git::validation::positions::Error::Retryable(message)) => {
crate::logger(self.log.as_ref(), message);
let addr = ctx.address();
let message_token = self.message_token;
let sleep_duration = self.sleep_duration;
let log = self.log.clone();
async move {
tracing::debug!("sleeping before retrying...");
crate::logger(log.as_ref(), "before sleep");
tokio::time::sleep(sleep_duration).await;
crate::logger(log.as_ref(), "after sleep");
crate::do_send(
addr,
crate::messages::ValidateRepo::new(message_token),
log.as_ref(),
);
}
.in_current_span()
.into_actor(self)
.wait(ctx);
}
Err(git::validation::positions::Error::UserIntervention(user_notification)) => {
crate::notify_user(
self.notify_user_recipient.as_ref(),
user_notification,
self.log.as_ref(),
)
}
Err(git::validation::positions::Error::NonRetryable(message)) => {
crate::logger(self.log.as_ref(), message);
}
}
}
}
enum TokenStatus {
Current,
Expired,
New(MessageToken),
}
impl crate::RepoActor {
fn token_status(&self, new: MessageToken) -> TokenStatus {
let current = &self.message_token;
if &new > current {
return TokenStatus::New(new);
}
if current > &new {
return TokenStatus::Expired;
}
TokenStatus::Current
}
}

View file

@ -1,22 +0,0 @@
//
use actix::prelude::*;
use crate as actor;
impl Handler<actor::messages::WebhookRegistered> for actor::RepoActor {
type Result = ();
#[tracing::instrument(name = "RepoActor::WebhookRegistered", skip_all, fields(repo = %self.repo_details, webhook_id = %msg.webhook_id()))]
fn handle(
&mut self,
msg: actor::messages::WebhookRegistered,
ctx: &mut Self::Context,
) -> Self::Result {
self.webhook_id.replace(msg.webhook_id().clone());
self.webhook_auth.replace(msg.webhook_auth().clone());
actor::do_send(
ctx.address(),
actor::messages::ValidateRepo::new(self.message_token),
self.log.as_ref(),
);
}
}

View file

@ -1,165 +1 @@
//
use actix::prelude::*;
use derive_more::Deref;
use kxio::network::Network;
use messages::NotifyUser;
use std::time::Duration;
use tracing::{info, warn, Instrument};
use git_next_core::{
git::{
self,
repository::{factory::RepositoryFactory, open::OpenRepositoryLike},
UserNotification,
},
server, WebhookAuth, WebhookId,
};
mod branch;
pub mod handlers;
mod load;
pub mod messages;
mod notifications;
#[cfg(test)]
mod tests;
#[derive(Clone, Debug, Default)]
pub struct RepoActorLog(std::sync::Arc<std::sync::RwLock<Vec<String>>>);
impl Deref for RepoActorLog {
type Target = std::sync::Arc<std::sync::RwLock<Vec<String>>>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
/// An actor that represents a Git Repository.
///
/// When this actor is started it is sent the [CloneRepo] message.
#[derive(Debug, derive_more::Display, derive_with::With)]
#[display("{}:{}:{}", generation, repo_details.forge.forge_alias(), repo_details.repo_alias)]
pub struct RepoActor {
sleep_duration: std::time::Duration,
generation: git::Generation,
message_token: messages::MessageToken,
repo_details: git::RepoDetails,
webhook: server::InboundWebhook,
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_factory: Box<dyn RepositoryFactory>,
open_repository: Option<Box<dyn OpenRepositoryLike>>,
net: Network,
forge: Box<dyn git::ForgeLike>,
log: Option<RepoActorLog>,
notify_user_recipient: Option<Recipient<NotifyUser>>,
}
impl RepoActor {
#[allow(clippy::too_many_arguments)]
pub fn new(
repo_details: git::RepoDetails,
forge: Box<dyn git::ForgeLike>,
webhook: server::InboundWebhook,
generation: git::Generation,
net: Network,
repository_factory: Box<dyn RepositoryFactory>,
sleep_duration: std::time::Duration,
notify_user_recipient: Option<Recipient<NotifyUser>>,
) -> Self {
let message_token = messages::MessageToken::default();
Self {
generation,
message_token,
repo_details,
webhook,
webhook_id: None,
webhook_auth: None,
last_main_commit: None,
last_next_commit: None,
last_dev_commit: None,
repository_factory,
open_repository: None,
forge,
net,
sleep_duration,
log: None,
notify_user_recipient,
}
}
}
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 {
tracing::debug!("stopping");
info!("Checking webhook");
match self.webhook_id.take() {
Some(webhook_id) => {
tracing::warn!("stopping - unregistering webhook");
info!(%webhook_id, "Unregistring webhook");
let forge = self.forge.duplicate();
async move {
if let Err(err) = forge.unregister_webhook(&webhook_id).await {
warn!("unregistering webhook: {err}");
}
}
.in_current_span()
.into_actor(self)
.wait(ctx);
Running::Continue
}
None => Running::Stop,
}
}
}
pub fn delay_send<M>(addr: Addr<RepoActor>, delay: Duration, msg: M, log: Option<&RepoActorLog>)
where
M: actix::Message + Send + 'static + std::fmt::Debug,
RepoActor: actix::Handler<M>,
<M as actix::Message>::Result: Send,
{
let log_message = format!("send-after-delay: {:?}", msg);
tracing::debug!(log_message);
logger(log, log_message);
std::thread::sleep(delay);
do_send(addr, msg, log)
}
pub fn do_send<M>(_addr: Addr<RepoActor>, msg: M, log: Option<&RepoActorLog>)
where
M: actix::Message + Send + 'static + std::fmt::Debug,
RepoActor: actix::Handler<M>,
<M as actix::Message>::Result: Send,
{
let log_message = format!("send: {:?}", msg);
tracing::debug!(log_message);
logger(log, log_message);
#[cfg(not(test))]
_addr.do_send(msg)
}
pub fn logger(log: Option<&RepoActorLog>, message: impl Into<String>) {
if let Some(log) = log {
let message: String = message.into();
tracing::debug!(message);
let _ = log.write().map(|mut l| l.push(message));
}
}
pub fn notify_user(
recipient: Option<&Recipient<NotifyUser>>,
user_notification: UserNotification,
log: Option<&RepoActorLog>,
) {
let msg = NotifyUser::from(user_notification);
let log_message = format!("send: {:?}", msg);
tracing::debug!(log_message);
logger(log, log_message);
if let Some(recipient) = &recipient {
recipient.do_send(msg);
}
}
// moved to /crates/cli/src/repo