WIP: feat: dispatch NotifyUser messages to server actor (1/2)

This commit is contained in:
Paul Campbell 2024-07-19 19:41:20 +01:00
parent 7d77ef8a84
commit eb3dbe4696
20 changed files with 105 additions and 48 deletions

View file

@ -2,7 +2,7 @@
use actix::prelude::*; use actix::prelude::*;
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 = ();
@ -26,8 +26,10 @@ impl Handler<actor::messages::LoadConfigFromRepo> for actor::RepoActor {
actor::logger(&log, "send: LoadedConfig"); actor::logger(&log, "send: LoadedConfig");
addr.do_send(actor::messages::ReceiveRepoConfig::new(repo_config)) addr.do_send(actor::messages::ReceiveRepoConfig::new(repo_config))
} }
Err(err) => tracing::warn!(?err, "Failed to load config"), Err(err) => {
// TODO: (#95) should notify user tracing::warn!(?err, "Failed to load config");
// self.notify_user(UserNotification::RepoConfigLoadFailure(err))
}
} }
} }
.in_current_span() .in_current_span()

View file

@ -1,5 +1,5 @@
// //
use crate as actor; use crate::{self as actor, UserNotification};
use actix::prelude::*; use actix::prelude::*;
use git_next_git as git; use git_next_git as git;
@ -33,7 +33,7 @@ impl Handler<actor::messages::ReceiveCIStatus> for actor::RepoActor {
} }
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 self.notify_user(UserNotification::CICheckFailed(next));
actor::delay_send( actor::delay_send(
addr, addr,
sleep_duration, sleep_duration,

View file

@ -2,7 +2,7 @@
use actix::prelude::*; use actix::prelude::*;
use tracing::Instrument as _; use tracing::Instrument as _;
use crate as actor; use crate::{self as actor, messages::NotifyUser, UserNotification};
use actor::{messages::RegisterWebhook, RepoActor}; use actor::{messages::RegisterWebhook, RepoActor};
impl Handler<RegisterWebhook> for RepoActor { impl Handler<RegisterWebhook> for RepoActor {
@ -10,11 +10,12 @@ impl Handler<RegisterWebhook> for RepoActor {
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 {
@ -27,13 +28,24 @@ impl Handler<RegisterWebhook> for RepoActor {
&log, &log,
); );
} }
Err(err) => tracing::warn!(?err, "registering webhook"), Err(err) => {
let msg = NotifyUser::from(UserNotification::WebhookRegistration(
forge_alias,
repo_alias,
err.to_string(),
));
let log_message = format!("send: {:?}", msg);
tracing::debug!(log_message);
actor::logger(&log, log_message);
if let Some(notify_user_recipient) = notify_user_recipient {
notify_user_recipient.do_send(msg);
}
}
} }
} }
.in_current_span() .in_current_span()
.into_actor(self) .into_actor(self)
.wait(ctx); .wait(ctx);
tracing::debug!("registering webhook done");
} }
} }
} }

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;
@ -13,6 +14,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 as git;
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};
@ -48,8 +51,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 +63,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 +82,13 @@ impl RepoActor {
net, net,
sleep_duration, sleep_duration,
log: None, log: None,
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));
} }
} }
} }

View file

@ -6,6 +6,8 @@ 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.
@ -63,3 +65,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

@ -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

@ -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

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

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,15 @@
//
use actix::prelude::*;
use git_next_config::server::NotificationType;
use git_next_repo_actor::messages::NotifyUser;
use crate::ServerActor;
impl Handler<NotifyUser> for ServerActor {
type Result = ();
fn handle(&mut self, _msg: NotifyUser, _ctx: &mut Self::Context) -> Self::Result {
// TODO: (#95) should notify user
// send post to notification webhook url
}
}

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

@ -3,13 +3,13 @@ use git_next_webhook_actor::{AddWebhookRecipient, ShutdownWebhook, WebhookActor,
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,
@ -22,11 +22,17 @@ impl Handler<ReceiveValidServerConfig> for Server {
// 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<_>>();

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 {

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,7 +44,8 @@ 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: Option<Addr<WebhookActor>>,
fs: FileSystem, fs: FileSystem,
@ -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,
@ -105,6 +107,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 +117,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 +137,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 +183,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...");

View file

@ -1,2 +1 @@
mod notify_user;
mod shutdown_webhook; mod shutdown_webhook;

View file

@ -1,12 +0,0 @@
//
use actix::prelude::*;
use crate::{messages::NotifyUser, WebhookActor};
impl Handler<NotifyUser> for WebhookActor {
type Result = ();
fn handle(&mut self, _msg: NotifyUser, _ctx: &mut Self::Context) -> Self::Result {
// TODO: (#95) should notify user
}
}

View file

@ -2,5 +2,3 @@
use git_next_actor_macros::message; use git_next_actor_macros::message;
message!(ShutdownWebhook: "Request to shutdown the Webhook actor"); message!(ShutdownWebhook: "Request to shutdown the Webhook actor");
message!(NotifyUser: String: "Request to send the message payload to the notification webhook");