From cd8ed8bc5793985df2c1692072a932ceee517b95 Mon Sep 17 00:00:00 2001 From: salmonik Date: Wed, 8 Jul 2026 02:13:35 +0200 Subject: OPEN THE RUST GATES Signed-off-by: salmonik --- src/tasks/mod.rs | 172 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 172 insertions(+) create mode 100644 src/tasks/mod.rs (limited to 'src/tasks/mod.rs') diff --git a/src/tasks/mod.rs b/src/tasks/mod.rs new file mode 100644 index 0000000..f3f8811 --- /dev/null +++ b/src/tasks/mod.rs @@ -0,0 +1,172 @@ +use std::sync::Arc; + +use chrono::{Duration, Timelike, Utc}; +use cron::Schedule; +use poise::serenity_prelude as serenity; +use std::str::FromStr; +use tokio::time::sleep; +use tracing::{error, info}; + +use crate::copyparty::CopypartyClient; +use crate::database::{self, DbPool}; +use crate::voice::VoiceManager; + +/// Shared resources the cron scheduler needs. +#[derive(Clone)] +pub struct CronContext { + pub pool: DbPool, + pub http: Arc, + pub voice_manager: VoiceManager, + pub copyparty: CopypartyClient, +} + +/// Normalize user-friendly cron expressions to the 7-field format the `cron` crate expects. +/// Supports: 5-field (min hour dom month dow) and 6-field (sec min hour dom month dow). +fn parse_cron(raw: &str) -> Result { + let fields: Vec<&str> = raw.split_whitespace().collect(); + let normalized = match fields.len() { + 7 => raw.to_string(), + 6 => format!("{} *", raw), // sec min hour dom month dow → +year + 5 => format!("0 {} *", raw), // min hour dom month dow → +sec +year + _ => raw.to_string(), // let the crate reject it + }; + Schedule::from_str(&normalized) +} + +/// Check window: look for cron fires in the last WINDOW_SECONDS. +/// 61s gives 1s overlap with 60s tick to avoid missing exact-boundary fires. +const WINDOW_SECONDS: i64 = 61; +/// Tick interval for the scheduler loop. +const TICK_SECONDS: u64 = 60; + +pub async fn check_and_execute_jobs(ctx: &CronContext) { + let jobs = match database::list_active_cron_jobs(&ctx.pool).await { + Ok(jobs) => jobs, + Err(e) => { + error!("Failed to load cron jobs: {e}"); + return; + } + }; + + let now = Utc::now(); + let window_start = now - Duration::seconds(WINDOW_SECONDS); + + for job in jobs { + let schedule = match parse_cron(&job.cron_expression) { + Ok(s) => s, + Err(e) => { + error!("Invalid cron expression '{}': {e}", job.cron_expression); + continue; + } + }; + + // Check if any fire time fell within the last WINDOW_SECONDS. + let mut has_fired = false; + for next_fire in schedule.after(&window_start) { + if next_fire > now { + break; + } + has_fired = true; + break; // one fire per window is enough + } + + if !has_fired { + continue; + } + + info!("Executing cron job {}", job.id); + match job.r#type.as_str() { + "audio" => execute_audio_job(ctx, &job).await, + "wiadomość" => execute_message_job(ctx, &job).await, + _ => {} + } + } +} + +async fn execute_message_job(ctx: &CronContext, job: &crate::database::CronJob) { + let Some(ref msg_text) = job.message else { + return; + }; + let channel_id = serenity::ChannelId::new(job.channel_id as u64); + if let Err(e) = channel_id.say(&ctx.http, msg_text).await { + error!("Cron job {} failed to send message: {e}", job.id); + } else { + info!("Cron job {} sent message to {}", job.id, job.channel_id); + } +} + +async fn execute_audio_job(ctx: &CronContext, job: &crate::database::CronJob) { + let Some(ref audio_path) = job.audio_filename else { + error!("Cron audio job {} has no audio_filename", job.id); + return; + }; + + let channel_id = serenity::ChannelId::new(job.channel_id as u64); + + // Resolve guild + build URL concurrently. + let url = ctx.copyparty.build_url(audio_path); + info!("Cron audio job {} streaming via ffmpeg: {url}", job.id); + let guild_channel_fut = channel_id.to_channel(&ctx.http); + let input_fut = crate::voice::ffmpeg_stream_input(&url); + + let (channel_result, input_result) = tokio::join!(guild_channel_fut, input_fut); + + let guild_id = match channel_result { + Ok(serenity::Channel::Guild(ch)) => ch.guild_id, + other => { + error!( + "Cron audio job {}: channel {} is not a guild voice channel ({:?})", + job.id, + job.channel_id, + other.map(|_| ()) + ); + return; + } + }; + + let input = match input_result { + Ok(i) => i, + Err(e) => { + error!( + "Cron audio job {} ffmpeg stream failed ({url}): {e}", + job.id + ); + return; + } + }; + + // Connect to voice and play. + let call = match ctx.voice_manager.connect(guild_id, channel_id).await { + Ok(c) => c, + Err(e) => { + error!("Cron audio job {} voice connect failed: {e}", job.id); + return; + } + }; + + { + let mut call_lock = call.lock().await; + call_lock.play_input(input); + } + + info!( + "Cron audio job {} playing '{}' (url: {url}) in guild {}", + job.id, audio_path, guild_id + ); +} + +pub async fn cron_scheduler_loop(ctx: CronContext) { + // Align first tick to the start of the next minute, then run every TICK_SECONDS. + let now = Utc::now(); + let next_minute = (now + Duration::minutes(1)) + .with_second(0) + .and_then(|t| t.with_nanosecond(0)) + .unwrap_or(now); + let initial_delay = (next_minute - now).to_std().unwrap_or_default(); + sleep(initial_delay).await; + + loop { + check_and_execute_jobs(&ctx).await; + sleep(std::time::Duration::from_secs(TICK_SECONDS)).await; + } +} -- cgit v1.2.3