1 files changed, 172 insertions, 0 deletions

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<serenity::Http>,
+    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<Schedule, cron::error::Error> {
+    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;
+    }
+}