Compare commits

..

1 commit

Author SHA1 Message Date
4a0e31081a WIP: dispatch NotifyUser messages to server actor
All checks were successful
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
2024-07-19 19:41:20 +01:00
21 changed files with 127 additions and 218 deletions

View file

@ -10,7 +10,6 @@ mod git_remote;
pub mod push; pub mod push;
mod repo_details; mod repo_details;
pub mod repository; pub mod repository;
mod user_notification;
pub mod validation; pub mod validation;
#[cfg(test)] #[cfg(test)]
@ -25,5 +24,3 @@ pub use git_remote::GitRemote;
pub use repo_details::RepoDetails; pub use repo_details::RepoDetails;
pub use repository::OpenRepository; pub use repository::OpenRepository;
pub use repository::Repository; pub use repository::Repository;
pub use user_notification::UserNotification;

View file

@ -1,15 +0,0 @@
use crate::Commit;
use git_next_config::{ForgeAlias, RepoAlias};
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum UserNotification {
CICheckFailed(ForgeAlias, RepoAlias, Commit),
RepoConfigLoadFailure(ForgeAlias, RepoAlias, String),
WebhookRegistration(ForgeAlias, RepoAlias, String),
DevNotBasedOnMain {
forge_alias: git_next_config::ForgeAlias,
repo_alias: git_next_config::RepoAlias,
dev_branch: git_next_config::BranchName,
main_branch: git_next_config::BranchName,
},
}

View file

@ -1,5 +1,5 @@
// //
use crate::{self as git, UserNotification}; use crate as git;
use git_next_config as config; use git_next_config as config;
pub type Result<T> = core::result::Result<T, Error>; pub type Result<T> = core::result::Result<T, Error>;
@ -40,14 +40,10 @@ pub fn validate_positions(
// Validations: // Validations:
// Dev must be on main branch, else the USER must rebase it // Dev must be on main branch, else the USER must rebase it
if is_not_based_on(&commit_histories.dev, &main) { if is_not_based_on(&commit_histories.dev, &main) {
return Err(Error::UserIntervention( return Err(Error::UserIntervention(format!(
UserNotification::DevNotBasedOnMain { "Branch '{}' not based on '{}'",
forge_alias: repo_details.forge.forge_alias().clone(), dev_branch, main_branch
repo_alias: repo_details.repo_alias.clone(), )));
dev_branch,
main_branch,
},
));
} }
// verify that next is on main or at most one commit on top of main, else reset it back to main // verify that next is on main or at most one commit on top of main, else reset it back to main
if is_not_based_on( if is_not_based_on(
@ -124,8 +120,8 @@ pub enum Error {
#[error("{0} - not retrying")] #[error("{0} - not retrying")]
NonRetryable(String), NonRetryable(String),
#[error("user intervention required")] #[error("{0} - user intervention required")]
UserIntervention(UserNotification), UserIntervention(String),
} }
impl From<git::fetch::Error> for Error { impl From<git::fetch::Error> for Error {
fn from(value: git::fetch::Error) -> Self { fn from(value: git::fetch::Error) -> Self {

View file

@ -30,18 +30,21 @@ impl Handler<actor::messages::AdvanceMain> for actor::RepoActor {
Err(err) => { Err(err) => {
tracing::warn!("advance main: {err}"); tracing::warn!("advance main: {err}");
} }
Ok(_) => match repo_config.source() { Ok(_) => {
let log = self.log.clone();
match repo_config.source() {
git_next_config::RepoConfigSource::Repo => { git_next_config::RepoConfigSource::Repo => {
actor::do_send(addr, actor::messages::LoadConfigFromRepo, self.log.as_ref()); actor::do_send(addr, actor::messages::LoadConfigFromRepo, &log);
} }
git_next_config::RepoConfigSource::Server => { git_next_config::RepoConfigSource::Server => {
actor::do_send( actor::do_send(
addr, addr,
actor::messages::ValidateRepo::new(message_token), actor::messages::ValidateRepo::new(message_token),
self.log.as_ref(), &log,
); );
} }
}, }
}
} }
} }
} }

View file

@ -35,7 +35,7 @@ impl Handler<actor::messages::AdvanceNext> for actor::RepoActor {
actor::do_send( actor::do_send(
addr, addr,
actor::messages::ValidateRepo::new(message_token), actor::messages::ValidateRepo::new(message_token),
self.log.as_ref(), &self.log,
); );
} }
Err(err) => tracing::warn!("advance next: {err}"), Err(err) => tracing::warn!("advance next: {err}"),

View file

@ -11,7 +11,7 @@ impl Handler<actor::messages::CheckCIStatus> for actor::RepoActor {
msg: actor::messages::CheckCIStatus, msg: actor::messages::CheckCIStatus,
ctx: &mut Self::Context, ctx: &mut Self::Context,
) -> Self::Result { ) -> Self::Result {
actor::logger(self.log.as_ref(), "start: CheckCIStatus"); actor::logger(&self.log, "start: CheckCIStatus");
let addr = ctx.address(); let addr = ctx.address();
let forge = self.forge.duplicate(); let forge = self.forge.duplicate();
let next = msg.unwrap(); let next = msg.unwrap();
@ -24,7 +24,7 @@ impl Handler<actor::messages::CheckCIStatus> for actor::RepoActor {
actor::do_send( actor::do_send(
addr, addr,
actor::messages::ReceiveCIStatus::new((next, status)), actor::messages::ReceiveCIStatus::new((next, status)),
log.as_ref(), &log,
); );
} }
.in_current_span() .in_current_span()

View file

@ -12,29 +12,29 @@ impl Handler<actor::messages::CloneRepo> for actor::RepoActor {
_msg: actor::messages::CloneRepo, _msg: actor::messages::CloneRepo,
ctx: &mut Self::Context, ctx: &mut Self::Context,
) -> Self::Result { ) -> Self::Result {
actor::logger(self.log.as_ref(), "Handler: CloneRepo: start"); actor::logger(&self.log, "Handler: CloneRepo: start");
tracing::debug!("Handler: CloneRepo: start"); tracing::debug!("Handler: CloneRepo: start");
match git::repository::open(&*self.repository_factory, &self.repo_details) { match git::repository::open(&*self.repository_factory, &self.repo_details) {
Ok(repository) => { Ok(repository) => {
actor::logger(self.log.as_ref(), "open okay"); actor::logger(&self.log, "open okay");
tracing::debug!("open okay"); tracing::debug!("open okay");
self.open_repository.replace(repository); self.open_repository.replace(repository);
if self.repo_details.repo_config.is_none() { if self.repo_details.repo_config.is_none() {
actor::do_send( actor::do_send(
ctx.address(), ctx.address(),
actor::messages::LoadConfigFromRepo, actor::messages::LoadConfigFromRepo,
self.log.as_ref(), &self.log,
); );
} else { } else {
actor::do_send( actor::do_send(
ctx.address(), ctx.address(),
actor::messages::RegisterWebhook::new(), actor::messages::RegisterWebhook::new(),
self.log.as_ref(), &self.log,
); );
} }
} }
Err(err) => { Err(err) => {
actor::logger(self.log.as_ref(), "open failed"); actor::logger(&self.log, "open failed");
tracing::debug!("err: {err:?}"); tracing::debug!("err: {err:?}");
tracing::warn!("Could not open repo: {err}") tracing::warn!("Could not open repo: {err}")
} }

View file

@ -1,6 +1,5 @@
// //
use actix::prelude::*; use actix::prelude::*;
use git_next_git::UserNotification;
use tracing::Instrument as _; use tracing::Instrument as _;
use crate::{self as actor}; use crate::{self as actor};
@ -19,27 +18,18 @@ impl Handler<actor::messages::LoadConfigFromRepo> for actor::RepoActor {
}; };
let open_repository = open_repository.duplicate(); let open_repository = open_repository.duplicate();
let repo_details = self.repo_details.clone(); let repo_details = self.repo_details.clone();
let forge_alias = repo_details.forge.forge_alias().clone();
let repo_alias = repo_details.repo_alias.clone();
let addr = ctx.address(); let addr = ctx.address();
let notify_user_recipient = self.notify_user_recipient.clone();
let log = self.log.clone(); let log = self.log.clone();
async move { async move {
match actor::load::config_from_repository(repo_details, &*open_repository).await { match actor::load::config_from_repository(repo_details, &*open_repository).await {
Ok(repo_config) => actor::do_send( Ok(repo_config) => {
addr, actor::logger(&log, "send: LoadedConfig");
actor::messages::ReceiveRepoConfig::new(repo_config), addr.do_send(actor::messages::ReceiveRepoConfig::new(repo_config))
log.as_ref(), }
), Err(err) => {
Err(err) => actor::notify_user( tracing::warn!(?err, "Failed to load config");
notify_user_recipient.as_ref(), // self.notify_user(UserNotification::RepoConfigLoadFailure(err))
UserNotification::RepoConfigLoadFailure( }
forge_alias,
repo_alias,
err.to_string(),
),
log.as_ref(),
),
} }
} }
.in_current_span() .in_current_span()

View file

@ -1,7 +1,6 @@
// //
use crate::{self as actor}; use crate::{self as actor, UserNotification};
use actix::prelude::*; use actix::prelude::*;
use git::UserNotification;
use git_next_git as git; use git_next_git as git;
impl Handler<actor::messages::ReceiveCIStatus> for actor::RepoActor { impl Handler<actor::messages::ReceiveCIStatus> for actor::RepoActor {
@ -13,43 +12,33 @@ impl Handler<actor::messages::ReceiveCIStatus> for actor::RepoActor {
ctx: &mut Self::Context, ctx: &mut Self::Context,
) -> Self::Result { ) -> Self::Result {
let log = self.log.clone(); let log = self.log.clone();
actor::logger(log.as_ref(), "start: ReceiveCIStatus"); actor::logger(&log, "start: ReceiveCIStatus");
let addr = ctx.address(); let addr = ctx.address();
let (next, status) = msg.unwrap(); 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 message_token = self.message_token;
let sleep_duration = self.sleep_duration; let sleep_duration = self.sleep_duration;
tracing::debug!(?status, ""); tracing::debug!(?status, "");
match status { match status {
git::forge::commit::Status::Pass => { git::forge::commit::Status::Pass => {
actor::do_send( actor::do_send(addr, actor::messages::AdvanceMain::new(next), &self.log);
addr,
actor::messages::AdvanceMain::new(next),
self.log.as_ref(),
);
} }
git::forge::commit::Status::Pending => { git::forge::commit::Status::Pending => {
std::thread::sleep(sleep_duration); std::thread::sleep(sleep_duration);
actor::do_send( actor::do_send(
addr, addr,
actor::messages::ValidateRepo::new(message_token), actor::messages::ValidateRepo::new(message_token),
self.log.as_ref(), &self.log,
); );
} }
git::forge::commit::Status::Fail => { git::forge::commit::Status::Fail => {
tracing::warn!("Checks have failed"); tracing::warn!("Checks have failed");
actor::notify_user( self.notify_user(UserNotification::CICheckFailed(next));
self.notify_user_recipient.as_ref(),
UserNotification::CICheckFailed(forge_alias, repo_alias, next),
log.as_ref(),
);
actor::delay_send( actor::delay_send(
addr, addr,
sleep_duration, sleep_duration,
actor::messages::ValidateRepo::new(message_token), actor::messages::ValidateRepo::new(message_token),
self.log.as_ref(), &self.log,
); );
} }
} }

View file

@ -17,7 +17,7 @@ impl Handler<actor::messages::ReceiveRepoConfig> for actor::RepoActor {
actor::do_send( actor::do_send(
ctx.address(), ctx.address(),
actor::messages::RegisterWebhook::new(), actor::messages::RegisterWebhook::new(),
self.log.as_ref(), &self.log,
); );
} }
} }

View file

@ -2,21 +2,19 @@
use actix::prelude::*; use actix::prelude::*;
use tracing::Instrument as _; use tracing::Instrument as _;
use crate::{self as actor}; use crate as actor;
use actor::{messages::RegisterWebhook, RepoActor}; use actor::{messages::RegisterWebhook, RepoActor};
use git_next_git::UserNotification;
impl Handler<RegisterWebhook> for RepoActor { impl Handler<RegisterWebhook> for RepoActor {
type Result = (); type Result = ();
fn handle(&mut self, _msg: RegisterWebhook, ctx: &mut Self::Context) -> Self::Result { fn handle(&mut self, _msg: RegisterWebhook, ctx: &mut Self::Context) -> Self::Result {
if self.webhook_id.is_none() { if self.webhook_id.is_none() {
let forge_alias = self.repo_details.forge.forge_alias().clone(); let forge_alias = self.repo_details.forge.forge_alias();
let repo_alias = self.repo_details.repo_alias.clone(); let repo_alias = &self.repo_details.repo_alias;
let webhook_url = self.webhook.url(&forge_alias, &repo_alias); let webhook_url = self.webhook.url(forge_alias, repo_alias);
let forge = self.forge.duplicate(); let forge = self.forge.duplicate();
let addr = ctx.address(); let addr = ctx.address();
let notify_user_recipient = self.notify_user_recipient.clone();
let log = self.log.clone(); let log = self.log.clone();
tracing::debug!("registering webhook"); tracing::debug!("registering webhook");
async move { async move {
@ -26,25 +24,16 @@ impl Handler<RegisterWebhook> for RepoActor {
actor::do_send( actor::do_send(
addr, addr,
actor::messages::WebhookRegistered::from(registered_webhook), actor::messages::WebhookRegistered::from(registered_webhook),
log.as_ref(), &log,
);
}
Err(err) => {
actor::notify_user(
notify_user_recipient.as_ref(),
UserNotification::WebhookRegistration(
forge_alias,
repo_alias,
err.to_string(),
),
log.as_ref(),
); );
} }
Err(err) => tracing::warn!(?err, "registering webhook"),
} }
} }
.in_current_span() .in_current_span()
.into_actor(self) .into_actor(self)
.wait(ctx); .wait(ctx);
tracing::debug!("registering webhook done");
} }
} }
} }

View file

@ -14,14 +14,14 @@ impl Handler<actor::messages::ValidateRepo> for actor::RepoActor {
msg: actor::messages::ValidateRepo, msg: actor::messages::ValidateRepo,
ctx: &mut Self::Context, ctx: &mut Self::Context,
) -> Self::Result { ) -> Self::Result {
actor::logger(self.log.as_ref(), "start: ValidateRepo"); actor::logger(&self.log, "start: ValidateRepo");
// Message Token - make sure we are only triggered for the latest/current token // Message Token - make sure we are only triggered for the latest/current token
match self.token_status(msg.unwrap()) { match self.token_status(msg.unwrap()) {
TokenStatus::Current => {} // do nothing TokenStatus::Current => {} // do nothing
TokenStatus::Expired => { TokenStatus::Expired => {
actor::logger( actor::logger(
self.log.as_ref(), &self.log,
format!("discarded: old message token: {}", self.message_token), format!("discarded: old message token: {}", self.message_token),
); );
return; // message is expired return; // message is expired
@ -29,27 +29,24 @@ impl Handler<actor::messages::ValidateRepo> for actor::RepoActor {
TokenStatus::New(message_token) => { TokenStatus::New(message_token) => {
self.message_token = message_token; self.message_token = message_token;
actor::logger( actor::logger(
self.log.as_ref(), &self.log,
format!("new message token: {}", self.message_token), format!("new message token: {}", self.message_token),
); );
} }
} }
actor::logger( actor::logger(&self.log, format!("accepted token: {}", self.message_token));
self.log.as_ref(),
format!("accepted token: {}", self.message_token),
);
// Repository positions // Repository positions
let Some(ref open_repository) = self.open_repository else { let Some(ref open_repository) = self.open_repository else {
actor::logger(self.log.as_ref(), "no open repository"); actor::logger(&self.log, "no open repository");
return; return;
}; };
actor::logger(self.log.as_ref(), "have open repository"); actor::logger(&self.log, "have open repository");
let Some(repo_config) = self.repo_details.repo_config.clone() else { let Some(repo_config) = self.repo_details.repo_config.clone() else {
actor::logger(self.log.as_ref(), "no repo config"); actor::logger(&self.log, "no repo config");
return; return;
}; };
actor::logger(self.log.as_ref(), "have repo config"); actor::logger(&self.log, "have repo config");
match git::validation::positions::validate_positions( match git::validation::positions::validate_positions(
&**open_repository, &**open_repository,
@ -67,50 +64,46 @@ impl Handler<actor::messages::ValidateRepo> for actor::RepoActor {
actor::do_send( actor::do_send(
ctx.address(), ctx.address(),
actor::messages::CheckCIStatus::new(next), actor::messages::CheckCIStatus::new(next),
self.log.as_ref(), &self.log,
); );
} else if next != dev { } else if next != dev {
actor::do_send( actor::do_send(
ctx.address(), ctx.address(),
actor::messages::AdvanceNext::new((next, dev_commit_history)), actor::messages::AdvanceNext::new((next, dev_commit_history)),
self.log.as_ref(), &self.log,
) )
} else { } else {
// do nothing // do nothing
} }
} }
Err(git::validation::positions::Error::Retryable(message)) => { Err(git::validation::positions::Error::Retryable(message)) => {
actor::logger(self.log.as_ref(), message); actor::logger(&self.log, message);
let addr = ctx.address(); let addr = ctx.address();
let message_token = self.message_token; let message_token = self.message_token;
let sleep_duration = self.sleep_duration; let sleep_duration = self.sleep_duration;
let log = self.log.clone(); let log = self.log.clone();
async move { async move {
tracing::debug!("sleeping before retrying..."); tracing::debug!("sleeping before retrying...");
actor::logger(log.as_ref(), "before sleep"); actor::logger(&log, "before sleep");
tokio::time::sleep(sleep_duration).await; tokio::time::sleep(sleep_duration).await;
actor::logger(log.as_ref(), "after sleep"); actor::logger(&log, "after sleep");
actor::do_send( actor::do_send(
addr, addr,
actor::messages::ValidateRepo::new(message_token), actor::messages::ValidateRepo::new(message_token),
log.as_ref(), &log,
); );
} }
.in_current_span() .in_current_span()
.into_actor(self) .into_actor(self)
.wait(ctx); .wait(ctx);
} }
Err(git::validation::positions::Error::UserIntervention(user_notification)) => { Err(git::validation::positions::Error::NonRetryable(message))
actor::notify_user( | Err(git::validation::positions::Error::UserIntervention(message)) => {
self.notify_user_recipient.as_ref(), actor::logger(&self.log, message);
user_notification,
self.log.as_ref(),
)
}
Err(git::validation::positions::Error::NonRetryable(message)) => {
actor::logger(self.log.as_ref(), message);
} }
} }
tracing::debug!("Handler: ValidateRepo: finish");
} }
} }

View file

@ -17,30 +17,23 @@ impl Handler<WebhookNotification> for actor::RepoActor {
#[tracing::instrument(name = "RepoActor::WebhookMessage", skip_all, fields(token = %self.message_token, repo = %self.repo_details))] #[tracing::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 { fn handle(&mut self, msg: WebhookNotification, ctx: &mut Self::Context) -> Self::Result {
let Some(config) = &self.repo_details.repo_config else { let Some(config) = &self.repo_details.repo_config else {
actor::logger(self.log.as_ref(), "server has no repo config"); actor::logger(&self.log, "server has no repo config");
warn!("No repo config"); warn!("No repo config");
return; return;
}; };
if validate_notification( if validate_notification(&msg, &self.webhook_auth, &*self.forge, &self.log).is_err() {
&msg,
self.webhook_auth.as_ref(),
&*self.forge,
self.log.as_ref(),
)
.is_err()
{
return; return;
} }
let body = msg.body(); let body = msg.body();
match self.forge.parse_webhook_body(body) { match self.forge.parse_webhook_body(body) {
Err(err) => { Err(err) => {
actor::logger(self.log.as_ref(), "message parse error - not a push"); actor::logger(&self.log, "message parse error - not a push");
warn!(?err, "Not a 'push'"); warn!(?err, "Not a 'push'");
return; return;
} }
Ok(push) => match push.branch(config.branches()) { Ok(push) => match push.branch(config.branches()) {
None => { None => {
actor::logger(self.log.as_ref(), "unknown branch"); actor::logger(&self.log, "unknown branch");
warn!( warn!(
?push, ?push,
"Unrecognised branch, we should be filtering to only the ones we want" "Unrecognised branch, we should be filtering to only the ones we want"
@ -52,7 +45,7 @@ impl Handler<WebhookNotification> for actor::RepoActor {
push, push,
config.branches().main(), config.branches().main(),
&mut self.last_main_commit, &mut self.last_main_commit,
self.log.as_ref(), &self.log,
) )
.is_err() .is_err()
{ {
@ -64,7 +57,7 @@ impl Handler<WebhookNotification> for actor::RepoActor {
push, push,
config.branches().next(), config.branches().next(),
&mut self.last_next_commit, &mut self.last_next_commit,
self.log.as_ref(), &self.log,
) )
.is_err() .is_err()
{ {
@ -76,7 +69,7 @@ impl Handler<WebhookNotification> for actor::RepoActor {
push, push,
config.branches().dev(), config.branches().dev(),
&mut self.last_dev_commit, &mut self.last_dev_commit,
self.log.as_ref(), &self.log,
) )
.is_err() .is_err()
{ {
@ -93,16 +86,16 @@ impl Handler<WebhookNotification> for actor::RepoActor {
actor::do_send( actor::do_send(
ctx.address(), ctx.address(),
actor::messages::ValidateRepo::new(message_token), actor::messages::ValidateRepo::new(message_token),
self.log.as_ref(), &self.log,
); );
} }
} }
fn validate_notification( fn validate_notification(
msg: &WebhookNotification, msg: &WebhookNotification,
webhook_auth: Option<&WebhookAuth>, webhook_auth: &Option<WebhookAuth>,
forge: &dyn ForgeLike, forge: &dyn ForgeLike,
log: Option<&RepoActorLog>, log: &Option<RepoActorLog>,
) -> Result<(), ()> { ) -> Result<(), ()> {
let Some(expected_authorization) = webhook_auth else { let Some(expected_authorization) = webhook_auth else {
actor::logger(log, "server has no auth token"); actor::logger(log, "server has no auth token");
@ -129,7 +122,7 @@ fn handle_push(
push: Push, push: Push,
branch: BranchName, branch: BranchName,
last_commit: &mut Option<Commit>, last_commit: &mut Option<Commit>,
log: Option<&RepoActorLog>, log: &Option<RepoActorLog>,
) -> Result<(), ()> { ) -> Result<(), ()> {
actor::logger(log, "message is for dev branch"); actor::logger(log, "message is for dev branch");
let commit = git::Commit::from(push); let commit = git::Commit::from(push);

View file

@ -16,7 +16,7 @@ impl Handler<actor::messages::WebhookRegistered> for actor::RepoActor {
actor::do_send( actor::do_send(
ctx.address(), ctx.address(),
actor::messages::ValidateRepo::new(self.message_token), actor::messages::ValidateRepo::new(self.message_token),
self.log.as_ref(), &self.log,
); );
} }
} }

View file

@ -2,6 +2,7 @@ mod branch;
pub mod handlers; pub mod handlers;
mod load; mod load;
pub mod messages; pub mod messages;
mod user_notification;
#[cfg(test)] #[cfg(test)]
mod tests; mod tests;
@ -12,8 +13,9 @@ use actix::prelude::*;
use derive_more::Deref; use derive_more::Deref;
use git_next_config as config; use git_next_config as config;
use git_next_git::{self as git, UserNotification}; use git_next_git as git;
use messages::NotifyUser; use messages::NotifyUser;
pub use user_notification::UserNotification;
use kxio::network::Network; use kxio::network::Network;
use tracing::{info, warn, Instrument}; use tracing::{info, warn, Instrument};
@ -83,6 +85,12 @@ impl RepoActor {
notify_user_recipient, notify_user_recipient,
} }
} }
fn notify_user(&mut self, user_notification: UserNotification) {
if let Some(recipient) = &self.notify_user_recipient {
recipient.do_send(NotifyUser::from(user_notification));
}
}
} }
impl Actor for RepoActor { impl Actor for RepoActor {
type Context = Context<Self>; type Context = Context<Self>;
@ -110,8 +118,12 @@ impl Actor for RepoActor {
} }
} }
pub fn delay_send<M>(addr: Addr<RepoActor>, delay: Duration, msg: M, log: Option<&RepoActorLog>) pub fn delay_send<M>(
where addr: Addr<RepoActor>,
delay: Duration,
msg: M,
log: &Option<crate::RepoActorLog>,
) where
M: actix::Message + Send + 'static + std::fmt::Debug, M: actix::Message + Send + 'static + std::fmt::Debug,
RepoActor: actix::Handler<M>, RepoActor: actix::Handler<M>,
<M as actix::Message>::Result: Send, <M as actix::Message>::Result: Send,
@ -123,7 +135,7 @@ where
do_send(addr, msg, log) do_send(addr, msg, log)
} }
pub fn do_send<M>(_addr: Addr<RepoActor>, msg: M, log: Option<&RepoActorLog>) pub fn do_send<M>(_addr: Addr<RepoActor>, msg: M, log: &Option<crate::RepoActorLog>)
where where
M: actix::Message + Send + 'static + std::fmt::Debug, M: actix::Message + Send + 'static + std::fmt::Debug,
RepoActor: actix::Handler<M>, RepoActor: actix::Handler<M>,
@ -136,23 +148,10 @@ where
_addr.do_send(msg) _addr.do_send(msg)
} }
pub fn logger(log: Option<&RepoActorLog>, message: impl Into<String>) { pub fn logger(log: &Option<crate::RepoActorLog>, message: impl Into<String>) {
if let Some(log) = log { if let Some(log) = log {
let message: String = message.into(); let message: String = message.into();
tracing::debug!(message); tracing::debug!(message);
let _ = log.write().map(|mut l| l.push(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

@ -2,11 +2,12 @@
use config::newtype; use config::newtype;
use derive_more::Display; use derive_more::Display;
use git::UserNotification;
use git_next_actor_macros::message; use git_next_actor_macros::message;
use git_next_config as config; use git_next_config as config;
use git_next_git as git; use git_next_git as git;
use crate::UserNotification;
message!(LoadConfigFromRepo: "Request to load the `git-next.toml` from the git repo."); message!(LoadConfigFromRepo: "Request to load the `git-next.toml` from the git repo.");
message!(CloneRepo: "Request to clone (or open) the git repo."); message!(CloneRepo: "Request to clone (or open) the git repo.");
message!(ReceiveRepoConfig: config::RepoConfig: r#"Notification that the `git-next.toml` file has been loaded from the repo and parsed. message!(ReceiveRepoConfig: config::RepoConfig: r#"Notification that the `git-next.toml` file has been loaded from the repo and parsed.

View file

@ -45,12 +45,16 @@ async fn when_read_file_ok_should_send_config_loaded() -> TestResult {
//then //then
tracing::debug!(?log, ""); tracing::debug!(?log, "");
log.require_message_containing("send: ReceiveRepoConfig")?; log.read().map_err(|e| e.to_string()).map(|l| {
log.no_message_contains("send: NotifyUsers")?; assert!(l
.iter()
.any(|message| message.contains("send: LoadedConfig")))
})?;
Ok(()) Ok(())
} }
#[actix::test] #[actix::test]
#[ignore] //TODO: (#95) should notify user
async fn when_read_file_err_should_notify_user() -> TestResult { async fn when_read_file_err_should_notify_user() -> TestResult {
//given //given
let fs = given::a_filesystem(); let fs = given::a_filesystem();
@ -76,7 +80,13 @@ async fn when_read_file_err_should_notify_user() -> TestResult {
//then //then
tracing::debug!(?log, ""); tracing::debug!(?log, "");
log.require_message_containing("send: NotifyUser")?; log.read().map_err(|e| e.to_string()).map(|l| {
log.no_message_contains("send: ReceiveRepoConfig")?; assert!(l.iter().any(|message| message.contains("send: NotifyUser")));
assert!(
!l.iter()
.any(|message| message.contains("send: LoadedConfig")),
"not send LoadedConfig"
);
})?;
Ok(()) Ok(())
} }

View file

@ -88,6 +88,7 @@ async fn when_fail_should_recheck_after_delay() -> TestResult {
} }
#[test_log::test(actix::test)] #[test_log::test(actix::test)]
#[ignore] //TODO: (#95) should notify user
async fn when_fail_should_notify_user() -> TestResult { async fn when_fail_should_notify_user() -> TestResult {
//given //given
let fs = given::a_filesystem(); let fs = given::a_filesystem();

View file

@ -36,6 +36,7 @@ async fn when_registered_ok_should_send_webhook_registered() -> TestResult {
} }
#[actix::test] #[actix::test]
#[ignore] //TODO: (#95) should notify user
async fn when_registered_error_should_send_notify_user() -> TestResult { async fn when_registered_error_should_send_notify_user() -> TestResult {
//given //given
let fs = given::a_filesystem(); let fs = given::a_filesystem();

View file

@ -279,52 +279,7 @@ async fn repo_with_dev_ahead_of_next_should_advance_next() -> TestResult {
Ok(()) Ok(())
} }
#[test_log::test(actix::test)] // TODO: (#95) test: repo with dev not a child of main should notify user
async fn repo_with_dev_not_ahead_of_main_should_notify_user() -> TestResult {
//given
let fs = given::a_filesystem();
let (mut open_repository, repo_details) = given::an_open_repository(&fs);
#[allow(clippy::unwrap_used)]
let repo_config = repo_details.repo_config.clone().unwrap();
// Validate repo branches
expect::fetch_ok(&mut open_repository);
let branches = repo_config.branches();
// commit_log main
let main_commit = expect::main_commit_log(&mut open_repository, branches.main());
// next - on main
let next_commit = main_commit.clone();
let next_branch_log = vec![next_commit.clone(), given::a_commit()];
// dev - not ahead of next
let dev_commit = given::a_named_commit("dev");
let dev_branch_log = vec![dev_commit];
// commit_log next
open_repository
.expect_commit_log()
.times(1)
.with(eq(branches.next()), eq([main_commit.clone()]))
.return_once(move |_, _| Ok(next_branch_log));
// commit_log dev
let dev_commit_log = dev_branch_log.clone();
open_repository
.expect_commit_log()
.times(1)
.with(eq(branches.dev()), eq([main_commit]))
.return_once(move |_, _| Ok(dev_commit_log));
//when
let (addr, log) = when::start_actor_with_open_repository(
Box::new(open_repository),
repo_details,
given::a_forge(),
);
addr.send(crate::messages::ValidateRepo::new(MessageToken::default()))
.await?;
System::current().stop();
//then
log.require_message_containing("send: NotifyUser")?;
Ok(())
}
#[test_log::test(actix::test)] #[test_log::test(actix::test)]
async fn should_accept_message_with_current_token() -> TestResult { async fn should_accept_message_with_current_token() -> TestResult {
@ -427,7 +382,7 @@ async fn should_send_validate_repo_when_retryable_error() -> TestResult {
} }
#[test_log::test(actix::test)] #[test_log::test(actix::test)]
async fn should_send_notify_user_when_non_retryable_error() -> TestResult { async fn should_send_nothing_when_non_retryable_error() -> TestResult {
//given //given
let fs = given::a_filesystem(); let fs = given::a_filesystem();
let (mut open_repository, repo_details) = given::an_open_repository(&fs); let (mut open_repository, repo_details) = given::an_open_repository(&fs);
@ -469,6 +424,6 @@ async fn should_send_notify_user_when_non_retryable_error() -> TestResult {
//then //then
log.require_message_containing("accepted token")?; log.require_message_containing("accepted token")?;
log.require_message_containing("send: NotifyUser")?; log.no_message_contains("send:")?;
Ok(()) Ok(())
} }

View file

@ -0,0 +1,7 @@
use git_next_git::Commit;
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum UserNotification {
CICheckFailed(Commit),
RepoConfigLoadFailure(String),
}