Compare commits

...

9 commits

Author SHA1 Message Date
3802ed126f FIXUP server-actor handler notify-users
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-23 19:26:32 +01:00
68cb9d6e41 FIXUP server-actor lib 2024-07-23 19:26:32 +01:00
d6245ed65e WIP: feat: post webhook to user 2024-07-23 19:25:51 +01:00
85ed7aeaa1 WIP: feat: dispatch NotifyUser messages to server actor (2/2) 2024-07-23 19:24:55 +01:00
67fc1b1b12 refactor: use Option<&T> over &Option<T> 2024-07-23 19:24:55 +01:00
f6456433b4 WIP: feat: dispatch NotifyUser messages to server actor (1/2) 2024-07-23 19:24:55 +01:00
357ad0d713 FIXUP server-actor handler receive valid server config 2024-07-23 19:24:53 +01:00
e29a2d29b6 WIP: feat: support sending messages to a notification webhook 2024-07-23 19:23:08 +01:00
f6bbdf7cce FIXUP config 2024-07-23 19:23:08 +01:00
41 changed files with 508 additions and 146 deletions

View file

@ -78,6 +78,8 @@ git-conventional = "0.12"
bytes = "1.6" bytes = "1.6"
ulid = "1.1" ulid = "1.1"
warp = "0.3" warp = "0.3"
time = "0.3"
standardwebhooks = "1.0"
# boilerplate # boilerplate
derive_more = { version = "1.0.0-beta.6", features = [ derive_more = { version = "1.0.0-beta.6", features = [

View file

@ -1 +1,3 @@
crate::newtype!(BranchName: String, derive_more::Display, Default: "The name of a Git branch"); use serde::Serialize;
crate::newtype!(BranchName: String, derive_more::Display, Default, Serialize: "The name of a Git branch");

View file

@ -1,4 +1,6 @@
crate::newtype!(ForgeAlias: String, Hash, PartialOrd, Ord, derive_more::Display, Default: "The name of a Forge to connect to"); use serde::Serialize;
crate::newtype!(ForgeAlias: String, Hash, PartialOrd, Ord, derive_more::Display, Default, Serialize: "The name of a Forge to connect to");
impl From<&ForgeAlias> for std::path::PathBuf { impl From<&ForgeAlias> for std::path::PathBuf {
fn from(value: &ForgeAlias) -> Self { fn from(value: &ForgeAlias) -> Self {
Self::from(&value.0) Self::from(&value.0)

View file

@ -1,7 +1,8 @@
use derive_more::Display; use derive_more::Display;
use serde::Serialize;
use crate::newtype; use crate::newtype;
newtype!(RepoAlias: String, Display, Default, PartialOrd, Ord: r#"The alias of a repo. newtype!(RepoAlias: String, Display, Default, PartialOrd, Ord, Serialize: r#"The alias of a repo.
This is the alias for the repo within `git-next-server.toml`."#); This is the alias for the repo within `git-next-server.toml`."#);

View file

@ -199,7 +199,7 @@ impl Notification {
webhook: None, webhook: None,
} }
} }
pub const fn webhook(webhook: OutboundWebhook) -> Self { pub const fn new_webhook(webhook: OutboundWebhook) -> Self {
Self { Self {
r#type: NotificationType::Webhook, r#type: NotificationType::Webhook,
webhook: Some(webhook), webhook: Some(webhook),
@ -210,6 +210,10 @@ impl Notification {
self.r#type self.r#type
} }
pub const fn webhook(&self) -> Option<&OutboundWebhook> {
self.webhook.as_ref()
}
pub fn webhook_url(&self) -> Option<String> { pub fn webhook_url(&self) -> Option<String> {
self.webhook.clone().map(|x| x.url) self.webhook.clone().map(|x| x.url)
} }
@ -239,3 +243,11 @@ pub struct OutboundWebhook {
url: String, url: String,
secret: String, secret: String,
} }
impl OutboundWebhook {
pub fn url(&self) -> &str {
self.url.as_ref()
}
pub fn secret(&self) -> Secret<String> {
Secret::new(self.secret.clone())
}
}

View file

@ -753,7 +753,7 @@ mod given {
ServerStorage::new(a_name().into()) ServerStorage::new(a_name().into())
} }
pub fn a_notification_config() -> Notification { pub fn a_notification_config() -> Notification {
Notification::webhook(an_outbound_webhook()) Notification::new_webhook(an_outbound_webhook())
} }
pub fn some_forge_configs() -> BTreeMap<String, ForgeConfig> { pub fn some_forge_configs() -> BTreeMap<String, ForgeConfig> {
[(a_name(), a_forge_config())].into() [(a_name(), a_forge_config())].into()

View file

@ -25,11 +25,13 @@ async-trait = { workspace = true }
# fs/network # fs/network
kxio = { workspace = true } kxio = { workspace = true }
# # TOML parsing # TOML parsing
# serde = { workspace = true }
serde_json = { workspace = true } serde_json = { workspace = true }
# toml = { workspace = true } # toml = { workspace = true }
# webhooks - user notification
serde = { workspace = true }
# Secrets and Password # Secrets and Password
secrecy = { workspace = true } secrecy = { workspace = true }

View file

@ -2,6 +2,7 @@ use config::newtype;
use derive_more::Display; use derive_more::Display;
// //
use git_next_config as config; use git_next_config as config;
use serde::Serialize;
#[derive( #[derive(
Clone, Clone,
@ -13,6 +14,7 @@ use git_next_config as config;
Ord, Ord,
derive_more::Constructor, derive_more::Constructor,
derive_more::Display, derive_more::Display,
Serialize,
)] )]
#[display("{}", sha)] #[display("{}", sha)]
pub struct Commit { pub struct Commit {
@ -37,8 +39,8 @@ impl From<config::webhook::Push> for Commit {
} }
} }
newtype!(Sha: String, Display, Hash,PartialOrd, Ord: "The unique SHA for a git commit."); newtype!(Sha: String, Display, Hash,PartialOrd, Ord, Serialize: "The unique SHA for a git commit.");
newtype!(Message: String, Hash, PartialOrd, Ord: "The commit message for a git commit."); newtype!(Message: String, Display, Hash, PartialOrd, Ord, Serialize: "The commit message for a git commit.");
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub struct Histories { pub struct Histories {

View file

@ -10,6 +10,7 @@ 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)]
@ -24,3 +25,4 @@ 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

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

View file

@ -1,5 +1,5 @@
// //
use crate as git; use crate::{self as git, UserNotification};
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,10 +40,14 @@ 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(format!( return Err(Error::UserIntervention(
"Branch '{}' not based on '{}'", UserNotification::DevNotBasedOnMain {
dev_branch, main_branch forge_alias: repo_details.forge.forge_alias().clone(),
))); 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(
@ -120,8 +124,8 @@ pub enum Error {
#[error("{0} - not retrying")] #[error("{0} - not retrying")]
NonRetryable(String), NonRetryable(String),
#[error("{0} - user intervention required")] #[error("user intervention required")]
UserIntervention(String), UserIntervention(UserNotification),
} }
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

@ -43,6 +43,7 @@ git-conventional = { workspace = true }
# Webhooks # Webhooks
bytes = { workspace = true } bytes = { workspace = true }
ulid = { workspace = true } ulid = { workspace = true }
time = { workspace = true }
# boilerplate # boilerplate
derive_more = { workspace = true } derive_more = { workspace = true }

View file

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

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, self.log.as_ref(),
); );
} }
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, "start: CheckCIStatus"); actor::logger(self.log.as_ref(), "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, log.as_ref(),
); );
} }
.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, "Handler: CloneRepo: start"); actor::logger(self.log.as_ref(), "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, "open okay"); actor::logger(self.log.as_ref(), "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, self.log.as_ref(),
); );
} else { } else {
actor::do_send( actor::do_send(
ctx.address(), ctx.address(),
actor::messages::RegisterWebhook::new(), actor::messages::RegisterWebhook::new(),
&self.log, self.log.as_ref(),
); );
} }
} }
Err(err) => { Err(err) => {
actor::logger(&self.log, "open failed"); actor::logger(self.log.as_ref(), "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,8 +1,9 @@
// //
use actix::prelude::*; use actix::prelude::*;
use git_next_git::UserNotification;
use tracing::Instrument as _; use tracing::Instrument as _;
use crate as actor; use crate::{self as actor};
impl Handler<actor::messages::LoadConfigFromRepo> for actor::RepoActor { impl Handler<actor::messages::LoadConfigFromRepo> for actor::RepoActor {
type Result = (); type Result = ();
@ -18,16 +19,27 @@ 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) => { Ok(repo_config) => actor::do_send(
actor::logger(&log, "send: LoadedConfig"); addr,
addr.do_send(actor::messages::ReceiveRepoConfig::new(repo_config)) actor::messages::ReceiveRepoConfig::new(repo_config),
} log.as_ref(),
Err(err) => tracing::warn!(?err, "Failed to load config"), ),
// TODO: (#95) should notify user Err(err) => actor::notify_user(
notify_user_recipient.as_ref(),
UserNotification::RepoConfigLoadFailure {
forge_alias,
repo_alias,
reason: err.to_string(),
},
log.as_ref(),
),
} }
} }
.in_current_span() .in_current_span()

View file

@ -1,6 +1,7 @@
// //
use crate as actor; use crate::{self as actor};
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 {
@ -12,33 +13,47 @@ 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, "start: ReceiveCIStatus"); actor::logger(log.as_ref(), "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(addr, actor::messages::AdvanceMain::new(next), &self.log); actor::do_send(
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, self.log.as_ref(),
); );
} }
git::forge::commit::Status::Fail => { git::forge::commit::Status::Fail => {
tracing::warn!("Checks have failed"); tracing::warn!("Checks have failed");
// TODO: (#95) test: repo with next ahead of main and failing CI should notify user actor::notify_user(
self.notify_user_recipient.as_ref(),
UserNotification::CICheckFailed {
forge_alias,
repo_alias,
commit: 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, self.log.as_ref(),
); );
} }
} }

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, self.log.as_ref(),
); );
} }
} }

View file

@ -2,19 +2,21 @@
use actix::prelude::*; use actix::prelude::*;
use tracing::Instrument as _; use tracing::Instrument as _;
use crate as actor; use crate::{self 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(); let forge_alias = self.repo_details.forge.forge_alias().clone();
let repo_alias = &self.repo_details.repo_alias; let repo_alias = self.repo_details.repo_alias.clone();
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 {
@ -24,16 +26,25 @@ 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, log.as_ref(),
);
}
Err(err) => {
actor::notify_user(
notify_user_recipient.as_ref(),
UserNotification::WebhookRegistration {
forge_alias,
repo_alias,
reason: 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, "start: ValidateRepo"); actor::logger(self.log.as_ref(), "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, self.log.as_ref(),
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,24 +29,27 @@ 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, self.log.as_ref(),
format!("new message token: {}", self.message_token), format!("new message token: {}", self.message_token),
); );
} }
} }
actor::logger(&self.log, format!("accepted token: {}", self.message_token)); actor::logger(
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, "no open repository"); actor::logger(self.log.as_ref(), "no open repository");
return; return;
}; };
actor::logger(&self.log, "have open repository"); actor::logger(self.log.as_ref(), "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, "no repo config"); actor::logger(self.log.as_ref(), "no repo config");
return; return;
}; };
actor::logger(&self.log, "have repo config"); actor::logger(self.log.as_ref(), "have repo config");
match git::validation::positions::validate_positions( match git::validation::positions::validate_positions(
&**open_repository, &**open_repository,
@ -64,46 +67,50 @@ 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, self.log.as_ref(),
); );
} 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, self.log.as_ref(),
) )
} else { } else {
// do nothing // do nothing
} }
} }
Err(git::validation::positions::Error::Retryable(message)) => { Err(git::validation::positions::Error::Retryable(message)) => {
actor::logger(&self.log, message); actor::logger(self.log.as_ref(), 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, "before sleep"); actor::logger(log.as_ref(), "before sleep");
tokio::time::sleep(sleep_duration).await; tokio::time::sleep(sleep_duration).await;
actor::logger(&log, "after sleep"); actor::logger(log.as_ref(), "after sleep");
actor::do_send( actor::do_send(
addr, addr,
actor::messages::ValidateRepo::new(message_token), actor::messages::ValidateRepo::new(message_token),
&log, log.as_ref(),
); );
} }
.in_current_span() .in_current_span()
.into_actor(self) .into_actor(self)
.wait(ctx); .wait(ctx);
} }
Err(git::validation::positions::Error::NonRetryable(message)) Err(git::validation::positions::Error::UserIntervention(user_notification)) => {
| Err(git::validation::positions::Error::UserIntervention(message)) => { actor::notify_user(
actor::logger(&self.log, message); self.notify_user_recipient.as_ref(),
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,23 +17,30 @@ 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, "server has no repo config"); actor::logger(self.log.as_ref(), "server has no repo config");
warn!("No repo config"); warn!("No repo config");
return; return;
}; };
if validate_notification(&msg, &self.webhook_auth, &*self.forge, &self.log).is_err() { if validate_notification(
&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, "message parse error - not a push"); actor::logger(self.log.as_ref(), "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, "unknown branch"); actor::logger(self.log.as_ref(), "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"
@ -45,7 +52,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, self.log.as_ref(),
) )
.is_err() .is_err()
{ {
@ -57,7 +64,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, self.log.as_ref(),
) )
.is_err() .is_err()
{ {
@ -69,7 +76,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, self.log.as_ref(),
) )
.is_err() .is_err()
{ {
@ -86,16 +93,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, self.log.as_ref(),
); );
} }
} }
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");
@ -122,7 +129,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, self.log.as_ref(),
); );
} }
} }

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 notifications;
#[cfg(test)] #[cfg(test)]
mod tests; mod tests;
@ -12,7 +13,8 @@ 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 as git; use git_next_git::{self as git, UserNotification};
use messages::NotifyUser;
use kxio::network::Network; use kxio::network::Network;
use tracing::{info, warn, Instrument}; use tracing::{info, warn, Instrument};
@ -48,8 +50,10 @@ pub struct RepoActor {
net: Network, net: Network,
forge: Box<dyn git::ForgeLike>, forge: Box<dyn git::ForgeLike>,
log: Option<RepoActorLog>, log: Option<RepoActorLog>,
notify_user_recipient: Option<Recipient<NotifyUser>>,
} }
impl RepoActor { impl RepoActor {
#[allow(clippy::too_many_arguments)]
pub fn new( pub fn new(
repo_details: git::RepoDetails, repo_details: git::RepoDetails,
forge: Box<dyn git::ForgeLike>, forge: Box<dyn git::ForgeLike>,
@ -58,6 +62,7 @@ impl RepoActor {
net: Network, net: Network,
repository_factory: Box<dyn git::repository::RepositoryFactory>, repository_factory: Box<dyn git::repository::RepositoryFactory>,
sleep_duration: std::time::Duration, sleep_duration: std::time::Duration,
notify_user_recipient: Option<Recipient<NotifyUser>>,
) -> Self { ) -> Self {
let message_token = messages::MessageToken::default(); let message_token = messages::MessageToken::default();
Self { Self {
@ -76,6 +81,7 @@ impl RepoActor {
net, net,
sleep_duration, sleep_duration,
log: None, log: None,
notify_user_recipient,
} }
} }
} }
@ -105,12 +111,8 @@ impl Actor for RepoActor {
} }
} }
pub fn delay_send<M>( pub fn delay_send<M>(addr: Addr<RepoActor>, delay: Duration, msg: M, log: Option<&RepoActorLog>)
addr: Addr<RepoActor>, where
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,
@ -122,7 +124,7 @@ pub fn delay_send<M>(
do_send(addr, msg, log) do_send(addr, msg, log)
} }
pub fn do_send<M>(_addr: Addr<RepoActor>, msg: M, log: &Option<crate::RepoActorLog>) pub fn do_send<M>(_addr: Addr<RepoActor>, msg: M, log: Option<&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>,
@ -135,10 +137,23 @@ where
_addr.do_send(msg) _addr.do_send(msg)
} }
pub fn logger(log: &Option<crate::RepoActorLog>, message: impl Into<String>) { pub fn logger(log: Option<&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,6 +2,7 @@
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;
@ -63,3 +64,4 @@ Contains a tuple of the commit that was checked (the tip of the `next` branch) a
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!(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!(AdvanceMain: git::Commit: "Request to advance the `main` branch on to same commit as the `next` branch."); // next commit
message!(WebhookNotification: config::webhook::forge_notification::ForgeNotification: "Notification of a webhook message from the forge."); message!(WebhookNotification: config::webhook::forge_notification::ForgeNotification: "Notification of a webhook message from the forge.");
message!(NotifyUser: UserNotification: "Request to send the message payload to the notification webhook");

View file

@ -0,0 +1,72 @@
use derive_more::Deref as _;
use git_next_git::UserNotification;
use serde_json::json;
use crate::messages::NotifyUser;
impl NotifyUser {
pub fn as_json(self, timestamp: time::OffsetDateTime) -> serde_json::Value {
let timestamp = timestamp.unix_timestamp().to_string();
match self.deref() {
UserNotification::CICheckFailed {
forge_alias,
repo_alias,
commit,
} => json!({
"type": "cicheck.failed",
"timestamp": timestamp,
"data": {
"forge_alias": forge_alias,
"repo_alias": repo_alias,
"commit": {
"sha": commit.sha(),
"message": commit.message()
}
}
}),
UserNotification::RepoConfigLoadFailure {
forge_alias,
repo_alias,
reason,
} => json!({
"type": "config.load.failed",
"timestamp": timestamp,
"data": {
"forge_alias": forge_alias,
"repo_alias": repo_alias,
"reason": reason
}
}),
UserNotification::WebhookRegistration {
forge_alias,
repo_alias,
reason,
} => json!({
"type": "webhook.registration.failed",
"timestamp": timestamp,
"data": {
"forge_alias": forge_alias,
"repo_alias": repo_alias,
"reason": reason
}
}),
UserNotification::DevNotBasedOnMain {
forge_alias,
repo_alias,
dev_branch,
main_branch,
} => json!({
"type": "branch.dev.not-on-main",
"timestamp": timestamp,
"data": {
"forge_alias": forge_alias,
"repo_alias": repo_alias,
"branches": {
"dev": dev_branch,
"main": main_branch
}
}
}),
}
}
}

View file

@ -209,6 +209,7 @@ pub fn a_repo_actor(
net, net,
repository_factory, repository_factory,
std::time::Duration::from_nanos(1), std::time::Duration::from_nanos(1),
None,
) )
.with_log(actors_log), .with_log(actors_log),
log, log,

View file

@ -45,16 +45,12 @@ async fn when_read_file_ok_should_send_config_loaded() -> TestResult {
//then //then
tracing::debug!(?log, ""); tracing::debug!(?log, "");
log.read().map_err(|e| e.to_string()).map(|l| { log.require_message_containing("send: ReceiveRepoConfig")?;
assert!(l log.no_message_contains("send: NotifyUsers")?;
.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();
@ -80,13 +76,7 @@ async fn when_read_file_err_should_notify_user() -> TestResult {
//then //then
tracing::debug!(?log, ""); tracing::debug!(?log, "");
log.read().map_err(|e| e.to_string()).map(|l| { log.require_message_containing("send: NotifyUser")?;
assert!(l.iter().any(|message| message.contains("send: NotifyUser"))); log.no_message_contains("send: ReceiveRepoConfig")?;
assert!(
!l.iter()
.any(|message| message.contains("send: LoadedConfig")),
"not send LoadedConfig"
);
})?;
Ok(()) Ok(())
} }

View file

@ -88,7 +88,6 @@ 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,7 +36,6 @@ 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,7 +279,52 @@ async fn repo_with_dev_ahead_of_next_should_advance_next() -> TestResult {
Ok(()) Ok(())
} }
// TODO: (#95) test: repo with dev not a child of main should notify user #[test_log::test(actix::test)]
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 {
@ -382,7 +427,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_nothing_when_non_retryable_error() -> TestResult { async fn should_send_notify_user_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);
@ -424,6 +469,6 @@ async fn should_send_nothing_when_non_retryable_error() -> TestResult {
//then //then
log.require_message_containing("accepted token")?; log.require_message_containing("accepted token")?;
log.no_message_contains("send:")?; log.require_message_containing("send: NotifyUser")?;
Ok(()) Ok(())
} }

View file

@ -28,6 +28,14 @@ derive-with = { workspace = true }
# Actors # Actors
actix = { workspace = true } actix = { workspace = true }
# Webhooks
serde = { workspace = true }
serde_json = { workspace = true }
ulid = { workspace = true }
time = { workspace = true }
secrecy = { workspace = true }
standardwebhooks = { workspace = true }
[dev-dependencies] [dev-dependencies]
# Testing # Testing
# assert2 = { workspace = true } # assert2 = { workspace = true }

View file

@ -3,9 +3,9 @@ use actix::prelude::*;
use git_next_config::server::ServerConfig; use git_next_config::server::ServerConfig;
use git_next_file_watcher_actor::FileUpdated; use git_next_file_watcher_actor::FileUpdated;
use crate::{messages::ReceiveServerConfig, Server}; use crate::{messages::ReceiveServerConfig, ServerActor};
impl Handler<FileUpdated> for Server { impl Handler<FileUpdated> for ServerActor {
type Result = (); type Result = ();
fn handle(&mut self, _msg: FileUpdated, ctx: &mut Self::Context) -> Self::Result { fn handle(&mut self, _msg: FileUpdated, ctx: &mut Self::Context) -> Self::Result {

View file

@ -1,4 +1,5 @@
mod file_updated; mod file_updated;
mod notify_user;
mod receive_server_config; mod receive_server_config;
mod receive_valid_server_config; mod receive_valid_server_config;
mod shutdown; mod shutdown;

View file

@ -0,0 +1,81 @@
//
use actix::prelude::*;
use git_next_config::server::{Notification, NotificationType};
use git_next_repo_actor::messages::NotifyUser;
use secrecy::ExposeSecret;
use standardwebhooks::Webhook;
use tracing::Instrument;
use crate::ServerActor;
impl Handler<NotifyUser> for ServerActor {
type Result = ();
fn handle(&mut self, msg: NotifyUser, ctx: &mut Self::Context) -> Self::Result {
let Some(server_config) = &self.server_config else {
return;
};
let notification_config = server_config.notification().clone();
let net = self.net.clone();
async move {
match notification_config.r#type() {
NotificationType::None => { /* do nothing */ }
NotificationType::Webhook => send_webhook(msg, notification_config, net).await,
}
}
.in_current_span()
.into_actor(self)
.wait(ctx);
}
}
async fn send_webhook(
msg: NotifyUser,
notification_config: Notification,
net: kxio::network::Network,
) {
let Some(webhook_config) = notification_config.webhook() else {
tracing::warn!("Invalid notification configuration (config) - can't sent notification");
return;
};
let Ok(webhook) = Webhook::new(webhook_config.secret().expose_secret()) else {
tracing::warn!("Invalid notification configuration (signer) - can't sent notification");
return;
};
do_send_webhook(msg, webhook, webhook_config, net).await;
}
async fn do_send_webhook(
msg: NotifyUser,
webhook: Webhook,
webhook_config: &git_next_config::server::OutboundWebhook,
net: kxio::network::Network,
) {
let message_id = format!("msg_{}", ulid::Ulid::new());
let timestamp = time::OffsetDateTime::now_utc();
let payload = msg.as_json(timestamp);
let timestamp = timestamp.unix_timestamp();
let to_sign = format!("{message_id}.{timestamp}.{payload}");
tracing::info!(?to_sign, "");
#[allow(clippy::expect_used)]
let signature = webhook
.sign(&message_id, timestamp, payload.to_string().as_ref())
.expect("signature");
tracing::info!(?signature, "");
let url = webhook_config.url();
use kxio::network::{NetRequest, NetUrl, RequestBody, ResponseType};
let net_url = NetUrl::new(url.to_string());
let request = NetRequest::post(net_url)
.body(RequestBody::Json(payload))
.header("webhook-id", &message_id)
.header("webhook-timestamp", &timestamp.to_string())
.header("webhook-signature", &signature)
.response_type(ResponseType::None)
.build();
net.post_json::<()>(request).await.map_or_else(
|err| {
tracing::warn!(?err, "sending webhook");
},
|_| (),
);
}

View file

@ -2,10 +2,10 @@ use actix::prelude::*;
use crate::{ use crate::{
messages::{ReceiveServerConfig, ReceiveValidServerConfig, ValidServerConfig}, messages::{ReceiveServerConfig, ReceiveValidServerConfig, ValidServerConfig},
Server, ServerActor,
}; };
impl Handler<ReceiveServerConfig> for Server { impl Handler<ReceiveServerConfig> for ServerActor {
type Result = (); type Result = ();
#[allow(clippy::cognitive_complexity)] #[allow(clippy::cognitive_complexity)]

View file

@ -1,32 +1,61 @@
use actix::prelude::*; use actix::prelude::*;
use git_next_config::server::NotificationType;
use git_next_webhook_actor::{AddWebhookRecipient, ShutdownWebhook, WebhookActor, WebhookRouter}; use git_next_webhook_actor::{AddWebhookRecipient, ShutdownWebhook, WebhookActor, WebhookRouter};
use standardwebhooks::Webhook;
use crate::{ use crate::{
messages::{ReceiveValidServerConfig, ValidServerConfig}, messages::{ReceiveValidServerConfig, ValidServerConfig},
Server, ServerActor,
}; };
impl Handler<ReceiveValidServerConfig> for Server { impl Handler<ReceiveValidServerConfig> for ServerActor {
type Result = (); type Result = ();
fn handle(&mut self, msg: ReceiveValidServerConfig, _ctx: &mut Self::Context) -> Self::Result { fn handle(&mut self, msg: ReceiveValidServerConfig, ctx: &mut Self::Context) -> Self::Result {
let ValidServerConfig { let ValidServerConfig {
server_config, server_config,
socket_address, socket_address,
server_storage, server_storage,
} = msg.unwrap(); } = msg.unwrap();
if let Some(webhook) = self.webhook.take() { // shutdown any existing webhook actor
webhook.do_send(ShutdownWebhook); if let Some(webhook_actor_addr) = self.webhook_actor_addr.take() {
webhook_actor_addr.do_send(ShutdownWebhook);
}
match server_config.notification().r#type() {
NotificationType::None => { /* do nothing */ }
NotificationType::Webhook => {
// Create webhook signer
use secrecy::ExposeSecret;
let webhook = server_config
.notification()
.webhook_secret()
.map(|secret| Webhook::new(secret.expose_secret()))
.transpose()
.map_err(|e| {
tracing::error!(
"Invalid notification webhook secret (will not send notifications): {e}"
)
})
.ok()
.flatten();
self.webhook = webhook;
}
} }
self.generation.inc(); self.generation.inc();
// Webhook Server // Webhook Server
tracing::info!("Starting Webhook Server..."); tracing::info!("Starting Webhook Server...");
let webhook_router = WebhookRouter::default().start(); let webhook_router = WebhookRouter::default().start();
let webhook = server_config.inbound_webhook(); let inbound_webhook = server_config.inbound_webhook();
// Forge Actors // Forge Actors
for (forge_alias, forge_config) in server_config.forges() { for (forge_alias, forge_config) in server_config.forges() {
let repo_actors = self let repo_actors = self
.create_forge_repos(forge_config, forge_alias.clone(), &server_storage, webhook) .create_forge_repos(
forge_config,
forge_alias.clone(),
&server_storage,
inbound_webhook,
ctx.address().recipient(),
)
.into_iter() .into_iter()
.map(|a| self.start_actor(a)) .map(|a| self.start_actor(a))
.collect::<Vec<_>>(); .collect::<Vec<_>>();
@ -45,7 +74,9 @@ impl Handler<ReceiveValidServerConfig> for Server {
.insert((forge_alias.clone(), repo_alias), addr); .insert((forge_alias.clone(), repo_alias), addr);
}); });
} }
let webhook = WebhookActor::new(socket_address, webhook_router.recipient()).start(); let webhook_actor_addr =
self.webhook.replace(webhook); WebhookActor::new(socket_address, webhook_router.recipient()).start();
self.webhook_actor_addr.replace(webhook_actor_addr);
self.server_config.replace(server_config);
} }
} }

View file

@ -2,9 +2,9 @@
use actix::prelude::*; use actix::prelude::*;
use git_next_webhook_actor::ShutdownWebhook; use git_next_webhook_actor::ShutdownWebhook;
use crate::{messages::Shutdown, Server}; use crate::{messages::Shutdown, ServerActor};
impl Handler<Shutdown> for Server { impl Handler<Shutdown> for ServerActor {
type Result = (); type Result = ();
fn handle(&mut self, _msg: Shutdown, _ctx: &mut Self::Context) -> Self::Result { fn handle(&mut self, _msg: Shutdown, _ctx: &mut Self::Context) -> Self::Result {
@ -16,7 +16,7 @@ impl Handler<Shutdown> for Server {
tracing::debug!(%forge_alias, %repo_alias, "removed webhook"); tracing::debug!(%forge_alias, %repo_alias, "removed webhook");
}); });
tracing::debug!("server shutdown"); tracing::debug!("server shutdown");
if let Some(webhook) = self.webhook.take() { if let Some(webhook) = self.webhook_actor_addr.take() {
tracing::debug!("shuting down webhook"); tracing::debug!("shuting down webhook");
webhook.do_send(ShutdownWebhook); webhook.do_send(ShutdownWebhook);
tracing::debug!("webhook shutdown"); tracing::debug!("webhook shutdown");

View file

@ -10,6 +10,7 @@ use git_next_config as config;
use git_next_config::server::{InboundWebhook, ServerConfig, ServerStorage}; use git_next_config::server::{InboundWebhook, ServerConfig, ServerStorage};
use git_next_config::{ForgeAlias, ForgeConfig, GitDir, RepoAlias, ServerRepoConfig}; use git_next_config::{ForgeAlias, ForgeConfig, GitDir, RepoAlias, ServerRepoConfig};
use git_next_git::{Generation, RepoDetails}; use git_next_git::{Generation, RepoDetails};
use git_next_repo_actor::messages::NotifyUser;
use git_next_repo_actor::{messages::CloneRepo, RepoActor}; use git_next_repo_actor::{messages::CloneRepo, RepoActor};
use git_next_webhook_actor as webhook; use git_next_webhook_actor as webhook;
use kxio::{fs::FileSystem, network::Network}; use kxio::{fs::FileSystem, network::Network};
@ -43,9 +44,10 @@ type Result<T> = core::result::Result<T, Error>;
#[derive(derive_with::With)] #[derive(derive_with::With)]
#[with(message_log)] #[with(message_log)]
pub struct Server { pub struct ServerActor {
server_config: Option<ServerConfig>,
generation: Generation, generation: Generation,
webhook: Option<Addr<WebhookActor>>, webhook_actor_addr: Option<Addr<WebhookActor>>,
fs: FileSystem, fs: FileSystem,
net: Network, net: Network,
repository_factory: Box<dyn RepositoryFactory>, repository_factory: Box<dyn RepositoryFactory>,
@ -55,11 +57,11 @@ pub struct Server {
// testing // testing
message_log: Option<Arc<RwLock<Vec<String>>>>, message_log: Option<Arc<RwLock<Vec<String>>>>,
} }
impl Actor for Server { impl Actor for ServerActor {
type Context = Context<Self>; type Context = Context<Self>;
} }
impl Server { impl ServerActor {
pub fn new( pub fn new(
fs: FileSystem, fs: FileSystem,
net: Network, net: Network,
@ -68,8 +70,9 @@ impl Server {
) -> Self { ) -> Self {
let generation = Generation::default(); let generation = Generation::default();
Self { Self {
server_config: None,
generation, generation,
webhook: None, webhook_actor_addr: None,
fs, fs,
net, net,
repository_factory: repo, repository_factory: repo,
@ -105,6 +108,7 @@ impl Server {
forge_name: ForgeAlias, forge_name: ForgeAlias,
server_storage: &ServerStorage, server_storage: &ServerStorage,
webhook: &InboundWebhook, webhook: &InboundWebhook,
notify_user_recipient: Recipient<NotifyUser>,
) -> Vec<(ForgeAlias, RepoAlias, RepoActor)> { ) -> Vec<(ForgeAlias, RepoAlias, RepoActor)> {
let span = let span =
tracing::info_span!("create_forge_repos", name = %forge_name, config = %forge_config); tracing::info_span!("create_forge_repos", name = %forge_name, config = %forge_config);
@ -114,7 +118,11 @@ impl Server {
let mut repos = vec![]; let mut repos = vec![];
let creator = self.create_actor(forge_name, forge_config.clone(), server_storage, webhook); let creator = self.create_actor(forge_name, forge_config.clone(), server_storage, webhook);
for (repo_alias, server_repo_config) in forge_config.repos() { for (repo_alias, server_repo_config) in forge_config.repos() {
let forge_repo = creator((repo_alias, server_repo_config)); let forge_repo = creator((
repo_alias,
server_repo_config,
notify_user_recipient.clone(),
));
info!( info!(
alias = %forge_repo.1, alias = %forge_repo.1,
"Created Repo" "Created Repo"
@ -130,14 +138,17 @@ impl Server {
forge_config: ForgeConfig, forge_config: ForgeConfig,
server_storage: &ServerStorage, server_storage: &ServerStorage,
webhook: &InboundWebhook, webhook: &InboundWebhook,
) -> impl Fn((RepoAlias, &ServerRepoConfig)) -> (ForgeAlias, RepoAlias, RepoActor) { ) -> impl Fn(
(RepoAlias, &ServerRepoConfig, Recipient<NotifyUser>),
) -> (ForgeAlias, RepoAlias, RepoActor) {
let server_storage = server_storage.clone(); let server_storage = server_storage.clone();
let webhook = webhook.clone(); let webhook = webhook.clone();
let net = self.net.clone(); let net = self.net.clone();
let repository_factory = self.repository_factory.duplicate(); let repository_factory = self.repository_factory.duplicate();
let generation = self.generation; let generation = self.generation;
let sleep_duration = self.sleep_duration; let sleep_duration = self.sleep_duration;
move |(repo_alias, server_repo_config)| { // let notify_user_recipient = server_addr.recipient();
move |(repo_alias, server_repo_config, notify_user_recipient)| {
let span = tracing::info_span!("create_actor", alias = %repo_alias, config = %server_repo_config); let span = tracing::info_span!("create_actor", alias = %repo_alias, config = %server_repo_config);
let _guard = span.enter(); let _guard = span.enter();
info!("Creating Repo"); info!("Creating Repo");
@ -173,6 +184,7 @@ impl Server {
net.clone(), net.clone(),
repository_factory.duplicate(), repository_factory.duplicate(),
sleep_duration, sleep_duration,
Some(notify_user_recipient),
); );
(forge_name.clone(), repo_alias, actor) (forge_name.clone(), repo_alias, actor)
} }

View file

@ -1,5 +1,5 @@
// //
use crate::{tests::given, ReceiveServerConfig, Server}; use crate::{tests::given, ReceiveServerConfig, ServerActor};
use actix::prelude::*; use actix::prelude::*;
use git_next_config::server::{Http, InboundWebhook, Notification, ServerConfig, ServerStorage}; use git_next_config::server::{Http, InboundWebhook, Notification, ServerConfig, ServerStorage};
use std::{ use std::{
@ -17,7 +17,7 @@ async fn when_webhook_url_has_trailing_slash_should_not_send() {
let duration = std::time::Duration::from_millis(1); let duration = std::time::Duration::from_millis(1);
// sut // sut
let server = Server::new(fs.clone(), net.into(), repo, duration); let server = ServerActor::new(fs.clone(), net.into(), repo, duration);
// collaborators // collaborators
let http = Http::new("0.0.0.0".to_string(), 80); let http = Http::new("0.0.0.0".to_string(), 80);

View file

@ -1,7 +1,7 @@
// //
use actix::prelude::*; use actix::prelude::*;
use git_next_file_watcher_actor::{FileUpdated, FileWatcher}; use git_next_file_watcher_actor::{FileUpdated, FileWatcher};
use git_next_server_actor::Server; use git_next_server_actor::ServerActor;
use kxio::{fs::FileSystem, network::Network}; use kxio::{fs::FileSystem, network::Network};
use std::path::PathBuf; use std::path::PathBuf;
use tracing::{error, info, level_filters::LevelFilter}; use tracing::{error, info, level_filters::LevelFilter};
@ -40,7 +40,7 @@ pub fn start(
info!("Starting Server..."); info!("Starting Server...");
let execution = async move { let execution = async move {
let server = Server::new(fs.clone(), net.clone(), repo, sleep_duration).start(); let server = ServerActor::new(fs.clone(), net.clone(), repo, sleep_duration).start();
server.do_send(FileUpdated); server.do_send(FileUpdated);
info!("Starting File Watcher..."); info!("Starting File Watcher...");