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
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
|
use std::sync::Arc;
use chrono::{DateTime, Duration, Local, TimeZone, 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::music::{MusicState, SourceKind, TrackRequest};
/// Shared resources the cron scheduler needs.
#[derive(Clone)]
pub struct CronContext {
pub pool: DbPool,
pub http: Arc<serenity::Http>,
pub copyparty: CopypartyClient,
pub music: Arc<MusicState>,
}
/// 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)
}
/// 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 = Local::now();
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;
}
};
// Find the next scheduled fire strictly after last_run.
// - If last_run is None (freshly created or never run), start from the
// last fire at/after the previous tick boundary so we still catch the
// fire scheduled for "now" on the first scan without doubling up.
// - If last_fire <= now, the job is due; we record last_run = last_fire
// (not `now`) so a long execution delay can't skip a legitimately
// pending later fire.
// - Evaluate the cron expression in local time so "37 21 * * *" means
// 21:37 in the server's timezone, not UTC.
let reference: DateTime<Local> = job.last_run.map_or_else(
|| {
// Step back a tick so the first scan after startup catches the fire
// scheduled for the minute we wake up on.
now - chrono::Duration::seconds(TICK_SECONDS as i64)
},
|utc_last_run| Local.from_utc_datetime(&utc_last_run.naive_utc()),
);
let mut due_fire: Option<DateTime<Local>> = None;
for next_fire in schedule.after(&reference) {
if next_fire > now {
break;
}
due_fire = Some(next_fire);
break;
}
let Some(fire_time) = due_fire else {
continue;
};
info!("Executing cron job {} (fire {})", job.id, fire_time);
match job.r#type.as_str() {
"audio" => execute_audio_job(ctx, &job).await,
"wiadomość" => execute_message_job(ctx, &job).await,
_ => {}
}
// Record last_run so we never re-fire the same schedule instant on the
// next tick. Kept outside the match so a failed/empty send still marks
// the fire as consumed and doesn't retry on every tick.
// Convert local fire_time to UTC for consistent DB storage.
let fire_time_utc = fire_time.with_timezone(&Utc);
if let Err(e) = database::update_cron_last_run(&ctx.pool, job.id, fire_time_utc).await {
error!("Failed to update last_run for cron job {}: {e}", job.id);
}
}
}
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_url) = 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);
// Build the full URL with password
let url = if audio_url.starts_with("http://") || audio_url.starts_with("https://") {
// Already a full URL (stored without password), append it now
if let Some(pw) = &ctx.copyparty.password {
format!("{audio_url}?pw={pw}")
} else {
audio_url.clone()
}
} else {
// Legacy relative path — build full URL with password
ctx.copyparty.build_url(audio_url)
};
info!("Cron audio job {} playing: {url}", job.id);
let channel = match channel_id.to_channel(&ctx.http).await {
Ok(serenity::Channel::Guild(ch)) => ch,
other => {
error!(
"Cron audio job {}: channel {} is not a guild voice channel ({:?})",
job.id,
job.channel_id,
other.map(|_| ())
);
return;
}
};
let request = TrackRequest {
url: url.clone(),
display_name: audio_url
.rsplit('/')
.next()
.unwrap_or(audio_url)
.to_string(),
kind: SourceKind::Stream,
compressed: false,
reupload: None,
requester: None,
};
if let Err(e) = ctx
.music
.play_now(channel.guild_id.get(), channel_id, None, request)
.await
{
error!("Cron audio job {} playback failed: {e}", job.id);
return;
}
info!(
"Cron audio job {} playing '{}' (url: {url}) in guild {}",
job.id, audio_url, channel.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 = Local::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;
}
}
|
