1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
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;
}
}
|
