Move postman and web inside src

This commit is contained in:
jude
2024-03-24 20:23:16 +00:00
parent 075fde71df
commit 4a80d42f86
152 changed files with 833 additions and 779 deletions

40
src/postman/metrics.rs Normal file
View File

@@ -0,0 +1,40 @@
use axum::{routing::get, Router};
use lazy_static;
use log::warn;
use prometheus::{register_int_counter, IntCounter, Registry};
lazy_static! {
static ref REGISTRY: Registry = Registry::new();
pub static ref REMINDERS_SENT: IntCounter =
register_int_counter!("reminders_sent", "Number of reminders sent").unwrap();
pub static ref REMINDERS_FAILED: IntCounter =
register_int_counter!("reminders_failed", "Number of reminders failed").unwrap();
}
pub fn init_metrics() {
REGISTRY.register(Box::new(REMINDERS_SENT.clone())).unwrap();
REGISTRY.register(Box::new(REMINDERS_FAILED.clone())).unwrap();
}
pub async fn serve() {
let app = Router::new().route("/metrics", get(metrics));
let metrics_port = std::env("PROMETHEUS_PORT").unwrap();
let listener =
tokio::net::TcpListener::bind(format!("localhost:{}", metrics_port)).await.unwrap();
axum::serve(listener, app).await.unwrap();
}
async fn metrics() -> String {
let encoder = prometheus::TextEncoder::new();
let res_custom = encoder.encode_to_string(&REGISTRY.gather());
match res_custom {
Ok(s) => s,
Err(e) => {
warn!("Error encoding metrics: {:?}", e);
String::new()
}
}
}

50
src/postman/mod.rs Normal file
View File

@@ -0,0 +1,50 @@
mod sender;
use std::env;
use log::{info, warn};
use poise::serenity_prelude::client::Context;
use sqlx::{Executor, MySql};
use tokio::{
sync::broadcast::Receiver,
time::{sleep_until, Duration, Instant},
};
type Database = MySql;
pub async fn initialize(
mut kill: Receiver<()>,
ctx: Context,
pool: impl Executor<'_, Database = Database> + Copy,
) -> Result<(), &'static str> {
tokio::select! {
output = _initialize(ctx, pool) => Ok(output),
_ = kill.recv() => {
warn!("Received terminate signal. Goodbye");
Err("Received terminate signal. Goodbye")
}
}
}
async fn _initialize(ctx: Context, pool: impl Executor<'_, Database = Database> + Copy) {
let remind_interval = env::var("REMIND_INTERVAL")
.map(|inner| inner.parse::<u64>().ok())
.ok()
.flatten()
.unwrap_or(10);
loop {
let sleep_to = Instant::now() + Duration::from_secs(remind_interval);
let reminders = sender::Reminder::fetch_reminders(pool).await;
if reminders.len() > 0 {
info!("Preparing to send {} reminders.", reminders.len());
for reminder in reminders {
reminder.send(pool, ctx.clone()).await;
}
}
sleep_until(sleep_to).await;
}
}

703
src/postman/sender.rs Normal file
View File

@@ -0,0 +1,703 @@
use std::env;
use chrono::{DateTime, Days, Months, TimeDelta};
use chrono_tz::Tz;
use lazy_static::lazy_static;
use log::{error, info, warn};
use num_integer::Integer;
use poise::serenity_prelude::{
all::{CreateAttachment, CreateEmbedFooter},
builder::{CreateEmbed, CreateEmbedAuthor, CreateMessage, ExecuteWebhook},
http::{CacheHttp, Http, HttpError},
model::{
channel::Channel,
id::{ChannelId, MessageId},
webhook::Webhook,
},
Error, Result,
};
use regex::{Captures, Regex};
use serde::Deserialize;
use sqlx::{
types::{
chrono::{NaiveDateTime, Utc},
Json,
},
Executor,
};
use crate::Database;
lazy_static! {
pub static ref TIMEFROM_REGEX: Regex =
Regex::new(r#"<<timefrom:(?P<time>\d+):(?P<format>.+)?>>"#).unwrap();
pub static ref TIMENOW_REGEX: Regex =
Regex::new(r#"<<timenow:(?P<timezone>(?:\w|/|_)+):(?P<format>.+)?>>"#).unwrap();
pub static ref LOG_TO_DATABASE: bool = env::var("LOG_TO_DATABASE").map_or(true, |v| v == "1");
}
fn fmt_displacement(format: &str, seconds: u64) -> String {
let mut seconds = seconds;
let mut days: u64 = 0;
let mut hours: u64 = 0;
let mut minutes: u64 = 0;
for (rep, time_type, div) in
[("%d", &mut days, 86400), ("%h", &mut hours, 3600), ("%m", &mut minutes, 60)].iter_mut()
{
if format.contains(*rep) {
let (divided, new_seconds) = seconds.div_rem(&div);
**time_type = divided;
seconds = new_seconds;
}
}
format
.replace("%s", &seconds.to_string())
.replace("%m", &minutes.to_string())
.replace("%h", &hours.to_string())
.replace("%d", &days.to_string())
}
pub fn substitute(string: &str) -> String {
let new = TIMEFROM_REGEX.replace(string, |caps: &Captures| {
let final_time = caps.name("time").map(|m| m.as_str().parse::<i64>().ok()).flatten();
let format = caps.name("format").map(|m| m.as_str());
if let (Some(final_time), Some(format)) = (final_time, format) {
match DateTime::from_timestamp(final_time, 0) {
Some(dt) => {
let now = Utc::now();
let difference = {
if now < dt {
dt - Utc::now()
} else {
Utc::now() - dt
}
};
fmt_displacement(format, difference.num_seconds() as u64)
}
None => String::new(),
}
} else {
String::new()
}
});
TIMENOW_REGEX
.replace(&new, |caps: &Captures| {
let timezone = caps.name("timezone").map(|m| m.as_str().parse::<Tz>().ok()).flatten();
let format = caps.name("format").map(|m| m.as_str());
if let (Some(timezone), Some(format)) = (timezone, format) {
let now = Utc::now().with_timezone(&timezone);
now.format(format).to_string()
} else {
String::new()
}
})
.to_string()
}
struct Embed {
title: String,
description: String,
image_url: Option<String>,
thumbnail_url: Option<String>,
footer: String,
footer_url: Option<String>,
author: String,
author_url: Option<String>,
color: u32,
fields: Json<Vec<EmbedField>>,
}
#[derive(Deserialize)]
struct EmbedField {
title: String,
value: String,
inline: bool,
}
impl Embed {
pub async fn from_id(
pool: impl Executor<'_, Database = Database> + Copy,
id: u32,
) -> Option<Self> {
match sqlx::query_as!(
Self,
r#"
SELECT
`embed_title` AS title,
`embed_description` AS description,
`embed_image_url` AS image_url,
`embed_thumbnail_url` AS thumbnail_url,
`embed_footer` AS footer,
`embed_footer_url` AS footer_url,
`embed_author` AS author,
`embed_author_url` AS author_url,
`embed_color` AS color,
IFNULL(`embed_fields`, '[]') AS "fields:_"
FROM reminders
WHERE `id` = ?"#,
id
)
.fetch_one(pool)
.await
{
Ok(mut embed) => {
embed.title = substitute(&embed.title);
embed.description = substitute(&embed.description);
embed.footer = substitute(&embed.footer);
embed.fields.iter_mut().for_each(|field| {
field.title = substitute(&field.title);
field.value = substitute(&field.value);
});
if embed.has_content() {
Some(embed)
} else {
None
}
}
Err(e) => {
warn!("Error loading embed from reminder: {:?}", e);
None
}
}
}
pub fn has_content(&self) -> bool {
if self.title.is_empty()
&& self.description.is_empty()
&& self.image_url.is_none()
&& self.thumbnail_url.is_none()
&& self.footer.is_empty()
&& self.footer_url.is_none()
&& self.author.is_empty()
&& self.author_url.is_none()
&& self.fields.0.is_empty()
{
false
} else {
true
}
}
}
impl Into<CreateEmbed> for Embed {
fn into(self) -> CreateEmbed {
let mut author = CreateEmbedAuthor::new(&self.author);
if let Some(author_icon) = &self.author_url {
author = author.icon_url(author_icon);
}
let mut footer = CreateEmbedFooter::new(&self.footer);
if let Some(footer_icon) = &self.footer_url {
footer = footer.icon_url(footer_icon);
}
let mut embed = CreateEmbed::default()
.title(&self.title)
.description(&self.description)
.color(self.color)
.author(author)
.footer(footer);
for field in &self.fields.0 {
embed = embed.field(&field.title, &field.value, field.inline);
}
if let Some(image_url) = &self.image_url {
embed = embed.image(image_url);
}
if let Some(thumbnail_url) = &self.thumbnail_url {
embed = embed.thumbnail(thumbnail_url);
}
embed
}
}
pub struct Reminder {
id: u32,
channel_id: u64,
thread_id: Option<u64>,
webhook_id: Option<u64>,
webhook_token: Option<String>,
channel_paused: bool,
channel_paused_until: Option<NaiveDateTime>,
enabled: bool,
tts: bool,
pin: bool,
content: String,
attachment: Option<Vec<u8>>,
attachment_name: Option<String>,
utc_time: DateTime<Utc>,
timezone: String,
restartable: bool,
expires: Option<DateTime<Utc>>,
interval_seconds: Option<u32>,
interval_days: Option<u32>,
interval_months: Option<u32>,
avatar: Option<String>,
username: Option<String>,
}
impl Reminder {
pub async fn fetch_reminders(pool: impl Executor<'_, Database = Database> + Copy) -> Vec<Self> {
match sqlx::query_as_unchecked!(
Reminder,
r#"
SELECT
reminders.`id` AS id,
channels.`channel` AS channel_id,
reminders.`thread_id` AS thread_id,
channels.`webhook_id` AS webhook_id,
channels.`webhook_token` AS webhook_token,
channels.`paused` AS 'channel_paused',
channels.`paused_until` AS 'channel_paused_until',
reminders.`enabled` AS 'enabled',
reminders.`tts` AS tts,
reminders.`pin` AS pin,
reminders.`content` AS content,
reminders.`attachment` AS attachment,
reminders.`attachment_name` AS attachment_name,
reminders.`utc_time` AS 'utc_time',
reminders.`timezone` AS timezone,
reminders.`restartable` AS restartable,
reminders.`expires` AS 'expires',
reminders.`interval_seconds` AS 'interval_seconds',
reminders.`interval_days` AS 'interval_days',
reminders.`interval_months` AS 'interval_months',
reminders.`avatar` AS avatar,
reminders.`username` AS username
FROM
reminders
INNER JOIN
channels
ON
reminders.channel_id = channels.id
WHERE
reminders.`status` = 'pending' AND
reminders.`id` IN (
SELECT
MIN(id)
FROM
reminders
WHERE
reminders.`utc_time` <= NOW() AND
`status` = 'pending' AND
(
reminders.`interval_seconds` IS NOT NULL
OR reminders.`interval_months` IS NOT NULL
OR reminders.`interval_days` IS NOT NULL
OR reminders.enabled
)
GROUP BY channel_id
)
"#,
)
.fetch_all(pool)
.await
{
Ok(reminders) => reminders
.into_iter()
.map(|mut rem| {
rem.content = substitute(&rem.content);
rem
})
.collect::<Vec<Self>>(),
Err(e) => {
warn!("Could not fetch reminders: {:?}", e);
vec![]
}
}
}
async fn reset_webhook(&self, pool: impl Executor<'_, Database = Database> + Copy) {
let _ = sqlx::query!(
"
UPDATE channels SET webhook_id = NULL, webhook_token = NULL WHERE channel = ?
",
self.channel_id
)
.execute(pool)
.await;
}
async fn refresh(&self, pool: impl Executor<'_, Database = Database> + Copy) {
if self.interval_seconds.is_some()
|| self.interval_months.is_some()
|| self.interval_days.is_some()
{
// If all intervals are zero then dont care
if self.interval_seconds == Some(0)
&& self.interval_days == Some(0)
&& self.interval_months == Some(0)
{
self.set_sent(pool).await;
}
let now = Utc::now();
let mut updated_reminder_time =
self.utc_time.with_timezone(&self.timezone.parse().unwrap_or(Tz::UTC));
let mut fail_count = 0;
while updated_reminder_time < now && fail_count < 4 {
if let Some(interval) = self.interval_months {
if interval != 0 {
updated_reminder_time = updated_reminder_time
.checked_add_months(Months::new(interval))
.unwrap_or_else(|| {
warn!(
"{}: Could not add {} months to a reminder",
interval, self.id
);
fail_count += 1;
updated_reminder_time
});
}
}
if let Some(interval) = self.interval_days {
if interval != 0 {
updated_reminder_time = updated_reminder_time
.checked_add_days(Days::new(interval as u64))
.unwrap_or_else(|| {
warn!("{}: Could not add {} days to a reminder", self.id, interval);
fail_count += 1;
updated_reminder_time
})
}
}
if let Some(interval) = self.interval_seconds {
updated_reminder_time += TimeDelta::try_seconds(interval as i64)
.unwrap_or_else(|| {
warn!("{}: Could not add {} seconds to a reminder", self.id, interval);
fail_count += 1;
TimeDelta::zero()
});
}
}
if fail_count >= 4 {
self.log_error(
"Failed to update 4 times and so is being deleted",
None::<&'static str>,
)
.await;
self.set_failed(pool, "Failed to update 4 times and so is being deleted").await;
} else if self.expires.map_or(false, |expires| updated_reminder_time > expires) {
self.set_sent(pool).await;
} else {
sqlx::query!(
"UPDATE reminders SET `utc_time` = ? WHERE `id` = ?",
updated_reminder_time.with_timezone(&Utc),
self.id
)
.execute(pool)
.await
.expect(&format!("Could not update time on Reminder {}", self.id));
}
} else {
self.set_sent(pool).await;
}
}
async fn log_error(&self, error: &'static str, debug_info: Option<impl std::fmt::Debug>) {
let message = match debug_info {
Some(info) => format!(
"{}
{:?}",
error, info
),
None => error.to_string(),
};
error!("[Reminder {}] {}", self.id, message);
}
async fn log_success(&self) {}
async fn set_sent(&self, pool: impl Executor<'_, Database = Database> + Copy) {
sqlx::query!("UPDATE reminders SET `status` = 'sent' WHERE `id` = ?", self.id)
.execute(pool)
.await
.expect(&format!("Could not delete Reminder {}", self.id));
}
async fn set_failed(
&self,
pool: impl Executor<'_, Database = Database> + Copy,
message: &'static str,
) {
sqlx::query!(
"UPDATE reminders SET `status` = 'failed', `status_message` = ? WHERE `id` = ?",
message,
self.id
)
.execute(pool)
.await
.expect(&format!("Could not delete Reminder {}", self.id));
}
async fn pin_message<M: Into<MessageId>>(&self, message_id: M, http: impl AsRef<Http>) {
let _ = http.as_ref().pin_message(self.channel_id.into(), message_id.into(), None).await;
}
pub async fn send(
&self,
pool: impl Executor<'_, Database = Database> + Copy,
cache_http: impl CacheHttp,
) {
async fn send_to_channel(
cache_http: impl CacheHttp,
reminder: &Reminder,
embed: Option<CreateEmbed>,
) -> Result<()> {
let channel = if let Some(thread_id) = reminder.thread_id {
ChannelId::new(thread_id).to_channel(&cache_http).await
} else {
ChannelId::new(reminder.channel_id).to_channel(&cache_http).await
};
let mut message = CreateMessage::new().content(&reminder.content).tts(reminder.tts);
if let (Some(attachment), Some(name)) =
(&reminder.attachment, &reminder.attachment_name)
{
message =
message.add_file(CreateAttachment::bytes(attachment as &[u8], name.as_str()));
}
if let Some(embed) = embed {
message = message.embed(embed);
}
match channel {
Ok(Channel::Guild(channel)) => {
match channel.send_message(&cache_http, message).await {
Ok(m) => {
if reminder.pin {
reminder.pin_message(m.id, cache_http.http()).await;
}
Ok(())
}
Err(e) => Err(e),
}
}
Ok(Channel::Private(channel)) => {
match channel.send_message(&cache_http.http(), message).await {
Ok(m) => {
if reminder.pin {
reminder.pin_message(m.id, cache_http.http()).await;
}
Ok(())
}
Err(e) => Err(e),
}
}
Err(e) => Err(e),
_ => Err(Error::Other("Channel not of valid type")),
}
}
async fn send_to_webhook(
cache_http: impl CacheHttp,
reminder: &Reminder,
webhook: Webhook,
embed: Option<CreateEmbed>,
) -> Result<()> {
let mut builder = if let Some(thread_id) = reminder.thread_id {
ExecuteWebhook::new()
.content(&reminder.content)
.tts(reminder.tts)
.in_thread(thread_id)
} else {
ExecuteWebhook::new().content(&reminder.content).tts(reminder.tts)
};
if let Some(username) = &reminder.username {
if !username.is_empty() {
builder = builder.username(username);
}
}
if let Some(avatar) = &reminder.avatar {
builder = builder.avatar_url(avatar);
}
if let (Some(attachment), Some(name)) =
(&reminder.attachment, &reminder.attachment_name)
{
builder =
builder.add_file(CreateAttachment::bytes(attachment as &[u8], name.as_str()));
}
if let Some(embed) = embed {
builder = builder.embeds(vec![embed]);
}
match webhook
.execute(&cache_http.http(), reminder.pin || reminder.restartable, builder)
.await
{
Ok(m) => {
if reminder.pin {
if let Some(message) = m {
reminder.pin_message(message.id, cache_http.http()).await;
}
}
Ok(())
}
Err(e) => Err(e),
}
}
if self.enabled
&& !(self.channel_paused
&& self
.channel_paused_until
.map_or(true, |inner| inner >= Utc::now().naive_local()))
{
let _ = sqlx::query!(
"
UPDATE `channels` SET paused = 0, paused_until = NULL WHERE `channel` = ?
",
self.channel_id
)
.execute(pool)
.await;
let embed = Embed::from_id(pool, self.id).await.map(|e| e.into());
let result = if let (Some(webhook_id), Some(webhook_token)) =
(self.webhook_id, &self.webhook_token)
{
let webhook_res = cache_http
.http()
.get_webhook_with_token(webhook_id.into(), webhook_token)
.await;
if let Ok(webhook) = webhook_res {
send_to_webhook(cache_http, &self, webhook, embed).await
} else {
warn!("Webhook vanished for reminder {}: {:?}", self.id, webhook_res);
self.reset_webhook(pool).await;
send_to_channel(cache_http, &self, embed).await
}
} else {
send_to_channel(cache_http, &self, embed).await
};
if let Err(e) = result {
if let Error::Http(error) = e {
if let HttpError::UnsuccessfulRequest(http_error) = error {
match http_error.error.code {
10003 => {
self.log_error(
"Could not be sent as channel does not exist",
None::<&'static str>,
)
.await;
self.set_failed(
pool,
"Could not be sent as channel does not exist",
)
.await;
}
10004 => {
self.log_error(
"Could not be sent as guild does not exist",
None::<&'static str>,
)
.await;
self.set_failed(pool, "Could not be sent as guild does not exist")
.await;
}
50001 => {
self.log_error(
"Could not be sent as missing access",
None::<&'static str>,
)
.await;
self.set_failed(pool, "Could not be sent as missing access").await;
}
50007 => {
self.log_error(
"Could not be sent as user has DMs disabled",
None::<&'static str>,
)
.await;
self.set_failed(pool, "Could not be sent as user has DMs disabled")
.await;
}
50013 => {
self.log_error(
"Could not be sent as permissions are invalid",
None::<&'static str>,
)
.await;
self.set_failed(
pool,
"Could not be sent as permissions are invalid",
)
.await;
}
_ => {
self.log_error("HTTP error sending reminder", Some(http_error))
.await;
self.refresh(pool).await;
}
}
} else {
self.log_error("(Likely) a parsing error", Some(error)).await;
self.refresh(pool).await;
}
} else {
self.log_error("Non-HTTP error", Some(e)).await;
self.refresh(pool).await;
}
} else {
self.log_success().await;
self.refresh(pool).await;
}
} else {
info!("Reminder {} is paused", self.id);
self.refresh(pool).await;
}
}
}