jet/noisebell
jet/noisebell
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: remove webhooks, prepping for simple flow

Browse source
This commit is contained in:
Jet Pham 2025-06-06 18:59:16 -07:00
parent 9bc01997d7
commit e761320bc7
No known key found for this signature in database
7 changed files with 936 additions and 557 deletions
Download patch file Download diff file Expand all files Collapse all files

54
src/discord.rs Normal file
View file

@ -0,0 +1,54 @@
use std::env;
use std::time::Instant;
use anyhow::Result;
use serenity::prelude::*;
use serenity::model::channel::ChannelId;
use tracing::{info, error};
pub struct DiscordClient {
client: Client,
channel_id: ChannelId,
}
impl DiscordClient {
pub async fn new() -> Result<Self> {
let token = env::var("DISCORD_TOKEN").expect("Expected DISCORD_TOKEN in environment");
// Validate token format
if let Err(e) = serenity::utils::token::validate(&token) {
return Err(anyhow::anyhow!("Invalid Discord token format: {}", e));
}
let channel_id = env::var("DISCORD_CHANNEL_ID")
.expect("Expected DISCORD_CHANNEL_ID in environment")
.parse::<u64>()?;
let intents = GatewayIntents::GUILD_MESSAGES;
let client = Client::builder(&token, intents)
.await
.expect("Error creating Discord client");
Ok(Self {
client,
channel_id: ChannelId(channel_id),
})
}
pub async fn send_circuit_event(&self, event: &crate::gpio::CircuitEvent) -> Result<()> {
let start = Instant::now();
info!("Sending Discord message for circuit event: {:?}", event);
let message = format!("Circuit state changed: {:?}", event);
if let Err(why) = self.channel_id.say(&self.client.http, message).await {
error!("Error sending Discord message: {:?}", why);
return Err(anyhow::anyhow!("Failed to send Discord message: {}", why));
}
let duration = start.elapsed();
info!("Discord message sent successfully in {:?}", duration);
Ok(())
}
}

64
src/main.rs
View file

@ -1,21 +1,14 @@
mod gpio;
mod webhook;
mod discord;
use std::time::Duration;
use std::fs;
use std::sync::Arc;
use std::time::Duration;
use anyhow::Result;
use axum::{
routing::post,
Router,
Json,
extract::State,
};
use tracing::{error, info};
use tracing_appender::rolling::{RollingFileAppender, Rotation};
use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt};
use tracing_subscriber::filter::LevelFilter;
use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt};
#[tokio::main]
async fn main() -> Result<()> {
@ -31,7 +24,7 @@ async fn main() -> Result<()> {
.build("logs")?;
let (non_blocking, _guard) = tracing_appender::non_blocking(file_appender);
// Only show our logs and hide hyper logs
let filter = tracing_subscriber::filter::Targets::new()
.with_target("noisebell", LevelFilter::INFO)
@ -44,37 +37,25 @@ async fn main() -> Result<()> {
.with(fmt::Layer::default().with_writer(non_blocking))
.init();
const DEFAULT_GPIO_PIN: u8 = 17;
const DEFAULT_WEBHOOK_RETRIES: u32 = 3;
const DEFAULT_SERVER_PORT: u16 = 8080;
info!("initializing Discord client");
let discord_client = discord::DiscordClient::new().await?;
info!("initializing webhook notifier");
let webhook_notifier = Arc::new(webhook::WebhookNotifier::new(DEFAULT_WEBHOOK_RETRIES)?);
const DEFAULT_GPIO_PIN: u8 = 17;
info!("initializing gpio monitor");
let mut gpio_monitor = gpio::GpioMonitor::new(DEFAULT_GPIO_PIN, Duration::from_millis(100))?;
let app = Router::new()
.route("/endpoints", post(add_endpoint))
.with_state(webhook_notifier.clone());
let server_addr = format!("127.0.0.1:{}", DEFAULT_SERVER_PORT);
info!("Starting API server on http://{}", server_addr);
let server = tokio::spawn(async move {
let listener = tokio::net::TcpListener::bind(&server_addr).await?;
axum::serve(listener, app.into_make_service())
.await?;
Ok::<_, anyhow::Error>(())
});
let discord_client = std::sync::Arc::new(discord_client);
let discord_client_clone = discord_client.clone();
let callback = move |event: gpio::CircuitEvent| {
info!("Circuit state changed: {:?}", event);
let notifier = webhook_notifier.clone();
let discord_client = discord_client_clone.clone();
tokio::spawn(async move {
notifier.notify_all("circuit_state_change", event).await;
if let Err(e) = discord_client.send_circuit_event(&event).await {
error!("Failed to send Discord message: {}", e);
}
});
};
@ -84,22 +65,5 @@ async fn main() -> Result<()> {
error!("GPIO monitoring error: {}", e);
}
// Wait for the server to complete (it shouldn't unless there's an error)
if let Err(e) = server.await? {
error!("Server error: {}", e);
}
Ok(())
}
async fn add_endpoint(
State(notifier): State<Arc<webhook::WebhookNotifier>>,
Json(endpoint): Json<webhook::Endpoint>,
) -> Result<(), axum::http::StatusCode> {
notifier.add_endpoint(endpoint)
.await
.map_err(|e| {
error!("Failed to add endpoint: {}", e);
axum::http::StatusCode::INTERNAL_SERVER_ERROR
})
}

146
src/webhook.rs
View file

@ -1,146 +0,0 @@
use anyhow::Result;
use reqwest::Client;
use serde::{Deserialize, Serialize};
use std::fs;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{error, info};
use std::time::Duration;
use futures::future::join_all;
use crate::gpio::CircuitEvent;
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct Endpoint {
url: String,
description: String,
}
#[derive(Debug, Deserialize, Serialize)]
struct EndpointsConfig {
endpoints: Vec<Endpoint>,
}
#[derive(Debug, Serialize)]
struct WebhookPayload {
event_type: String,
timestamp: String,
new_state: String,
}
#[derive(Clone)]
pub struct WebhookNotifier {
client: Client,
endpoints: Arc<RwLock<Vec<Endpoint>>>,
max_retries: u32,
}
impl WebhookNotifier {
pub fn new(max_retries: u32) -> Result<Self> {
let endpoints = if let Ok(config) = fs::read_to_string("endpoints.json") {
let endpoints_config: EndpointsConfig = serde_json::from_str(&config)?;
endpoints_config.endpoints
} else {
Vec::new()
};
Ok(Self {
client: Client::new(),
endpoints: Arc::new(RwLock::new(endpoints)),
max_retries,
})
}
pub async fn add_endpoint(&self, endpoint: Endpoint) -> Result<()> {
let mut endpoints = self.endpoints.write().await;
endpoints.retain(|e| e.description != endpoint.description);
endpoints.push(endpoint);
self.save_endpoints().await?;
Ok(())
}
async fn save_endpoints(&self) -> Result<()> {
let endpoints = self.endpoints.read().await;
let config = EndpointsConfig {
endpoints: endpoints.clone(),
};
fs::write("endpoints.json", serde_json::to_string_pretty(&config)?)?;
Ok(())
}
async fn send_webhook(&self, endpoint: &Endpoint, payload: &WebhookPayload) -> Result<()> {
match self.client
.post(&endpoint.url)
.json(payload)
.send()
.await
{
Ok(response) => {
if response.status().is_success() {
info!(
"Successfully sent webhook to {}: {}",
endpoint.description,
response.status()
);
Ok(())
} else {
error!(
"Webhook request to {} failed with status: {}",
endpoint.description,
response.status()
);
Err(anyhow::anyhow!("Failed with status: {}", response.status()))
}
}
Err(e) => {
error!(
"Failed to send webhook to {}: {}",
endpoint.description,
e
);
Err(anyhow::anyhow!("Request failed: {}", e))
}
}
}
async fn send_webhook_with_retries(&self, endpoint: &Endpoint, payload: &WebhookPayload) {
let mut attempts = 0;
while attempts < self.max_retries {
match self.send_webhook(endpoint, payload).await {
Ok(_) => break,
Err(e) => {
attempts += 1;
if attempts < self.max_retries {
tokio::time::sleep(Duration::from_secs(1)).await;
} else {
error!("Failed to send webhook to {} after {} attempts: {}",
endpoint.description, self.max_retries, e);
}
}
}
}
}
pub async fn notify_all(&self, event_type: &str, state: CircuitEvent) {
let payload = WebhookPayload {
event_type: event_type.to_string(),
timestamp: chrono::Utc::now().to_rfc3339(),
new_state: state.to_string(),
};
let endpoints = self.endpoints.read().await;
let webhook_futures: Vec<_> = endpoints.iter()
.map(|endpoint| {
info!("Sending webhook to {}: {}", endpoint.description, serde_json::to_string(&payload).unwrap());
self.send_webhook_with_retries(endpoint, &payload)
})
.collect();
join_all(webhook_futures).await;
}
}