Initial commit

.gitignore

@@ -0,0 +1,5 @@
+target/
+config.yaml
+db.sqlite3
+
+.vscode/

Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "calendar-bot"
+version = "1.0.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+anyhow = "1.0.65"
+async-mutex = "1.4.0"
+blocking = "1.2.0"
+bytes = "1.2.1"
+chrono = "0.4.22"
+chrono-tz = "0.6.3"
+diesel = { version = "2.0.2", features = ["chrono", "sqlite"] }
+diesel_migrations = "2.0.0"
+ical = { version = "0.7.0", features = ["ical"], default-features = false }
+log = "0.4.17"
+pretty_env_logger = "0.4.0"
+regex = "1.6.0"
+reqwest = "0.11.12"
+rrule = "0.10.0"
+serde = { version = "1.0.145", features = ["derive"] }
+serde_yaml = "0.9.13"
+teloxide = { version = "0.11.0", features = ["macros", "throttle"] }
+thiserror = "1.0.37"
+tokio = { version = "1.21.2", features = ["macros"] }

diesel.toml

@@ -0,0 +1,5 @@
+# For documentation on how to configure this file,
+# see diesel.rs/guides/configuring-diesel-cli
+
+[print_schema]
+file = "src/schema.rs"

migrations/2022-10-18-144346_create_base_table/down.sql

@@ -0,0 +1 @@
+DROP TABLE chat;

migrations/2022-10-18-144346_create_base_table/up.sql

@@ -0,0 +1,13 @@
+CREATE TABLE chat (
+   id INTEGER PRIMARY KEY NOT NULL,
+   telegram_id BIGINT UNIQUE NOT NULL,
+   calendar TEXT NOT NULL,
+   next_appointment_start BIGINT,
+   next_appointment_end BIGINT,
+   last_reminder BIGINT,
+   pinned_message_id INTEGER
+);
+
+CREATE INDEX chat_telegram_id ON chat (
+   telegram_id
+);

rustfmt.toml

@@ -0,0 +1 @@
+hard_tabs = true

src/appointment.rs

@@ -0,0 +1,195 @@
+use std::collections::HashMap;
+
+use anyhow::{Error, Result};
+use bytes::Buf;
+use chrono::{DateTime, Local, LocalResult, NaiveDateTime, TimeZone};
+use ical::{parser::ical::component::IcalEvent, IcalParser};
+use log::warn;
+use reqwest::IntoUrl;
+use rrule::{RRule, RRuleSet, RRuleSetIter, Tz, Unvalidated};
+
+pub async fn fetch_next_appointment<U: IntoUrl>(url: U) -> Result<Option<Appointment<Tz>>> {
+	let response = reqwest::get(url).await?.bytes().await?;
+
+	let calendar = IcalParser::new(response.reader())
+		.next()
+		.ok_or_else(|| Error::msg("ical file does not contain any calendars"))??;
+	let events = calendar
+		.events
+		.into_iter()
+		.filter_map(|event| Event::parse(event).ok());
+
+	let now = Local::now();
+	let now = now.with_timezone(&now.timezone().into());
+
+	let mut series = HashMap::new();
+	let mut ends = HashMap::new();
+	for event in events {
+		ends.insert(event.uid.clone(), event.end);
+		if let Some(rrule) = event.rrule {
+			let mut rrule_set = RRuleSet::new(now)
+				.rrule(rrule.parse::<RRule<Unvalidated>>()?.validate(event.start)?);
+			for exdate in event.exdates {
+				rrule_set = rrule_set.exdate(exdate);
+			}
+			series.insert(event.uid, rrule_set);
+		} else if let Some(recurrence_id) = event.recurrence_id {
+			let uid = event.uid;
+			let (uid, rrule_set) = series.remove_entry(&uid).unwrap();
+			let rrule_set = rrule_set.exdate(recurrence_id).rdate(event.start);
+			series.insert(uid, rrule_set);
+		} else {
+			series.insert(event.uid, RRuleSet::new(now).rdate(event.start));
+		}
+	}
+
+	for (uid, start) in series.iter_appointments() {
+		if start < now {
+			continue;
+		}
+
+		let end = *ends.get(uid).unwrap();
+		// Move the end time to the day of the current series element
+		let end = start.date().and_time(end.naive_local().time()).unwrap();
+
+		return Ok(Some(Appointment { start, end }));
+	}
+
+	Ok(None)
+}
+
+struct AppointmentsIterator<'a> {
+	inner: Vec<(&'a String, DateTime<Tz>, RRuleSetIter<'a>)>,
+}
+
+impl<'a> Iterator for AppointmentsIterator<'a> {
+	type Item = (&'a String, DateTime<Tz>);
+
+	fn next(&mut self) -> Option<Self::Item> {
+		if self.inner.len() == 0 {
+			return None;
+		}
+		self.inner.sort_unstable_by(|a, b| b.1.cmp(&a.1));
+		let mut next_appointment_data = self.inner.pop().unwrap();
+		let uid = next_appointment_data.0;
+		let next_appointment = next_appointment_data.1;
+		if let Some(next_appointment) = next_appointment_data.2.next() {
+			next_appointment_data.1 = next_appointment;
+			self.inner.push(next_appointment_data);
+		}
+		Some((uid, next_appointment))
+	}
+}
+
+trait IterAppointments {
+	fn iter_appointments(&self) -> AppointmentsIterator<'_>;
+}
+
+impl IterAppointments for HashMap<String, RRuleSet> {
+	fn iter_appointments(&self) -> AppointmentsIterator {
+		let mut inner = vec![];
+		for (uid, rrule_set) in self {
+			let mut rrule_set_iterator = rrule_set.into_iter();
+			let next_appointment = rrule_set_iterator.next();
+			if let Some(next_appointment) = next_appointment {
+				inner.push((uid, next_appointment, rrule_set_iterator))
+			}
+		}
+
+		AppointmentsIterator { inner }
+	}
+}
+
+#[derive(Debug, Clone)]
+pub struct Appointment<Tz: TimeZone> {
+	pub start: DateTime<Tz>,
+	pub end: DateTime<Tz>,
+}
+
+impl<Tz: TimeZone> Appointment<Tz> {
+	pub fn with_timezone<Tz2: TimeZone>(&self, tz: &Tz2) -> Appointment<Tz2> {
+		Appointment {
+			start: self.start.with_timezone(tz),
+			end: self.end.with_timezone(tz),
+		}
+	}
+}
+
+#[derive(Debug)]
+struct Event {
+	uid: String,
+	start: DateTime<Tz>,
+	end: DateTime<Tz>,
+	rrule: Option<String>,
+	recurrence_id: Option<DateTime<Tz>>,
+	exdates: Vec<DateTime<Tz>>,
+}
+
+impl Event {
+	fn parse(ical_event: IcalEvent) -> Result<Self> {
+		let mut uid = None;
+		let mut start = None;
+		let mut end = None;
+		let mut rrule = None;
+		let mut recurrence_id = None;
+		let mut exdates = vec![];
+		for property in ical_event.properties {
+			if let Some(value) = property.value {
+				match property.name.as_str() {
+					"UID" => uid = uid.or(Some(value)),
+					"RRULE" => rrule = rrule.or(Some(value)),
+					"DTSTART" => start = start.or(parse_date(property.params, &value)),
+					"DTEND" => end = end.or(parse_date(property.params, &value)),
+					"RECURRENCE-ID" => {
+						recurrence_id = recurrence_id.or(parse_date(property.params, &value))
+					}
+					"EXDATE" => parse_date(property.params, &value)
+						.into_iter()
+						.for_each(|date| exdates.push(date)),
+					_ => {}
+				}
+			}
+		}
+		let uid = uid.ok_or_else(|| Error::msg("Event has no uid"))?;
+		let start = start.ok_or_else(|| Error::msg("Event has no dtstart"))?;
+		let end = end.ok_or_else(|| Error::msg("Event has no dtend"))?;
+		Ok(Event {
+			uid,
+			start,
+			end,
+			rrule,
+			recurrence_id,
+			exdates,
+		})
+	}
+}
+
+fn parse_date(params: Option<Vec<(String, Vec<String>)>>, value: &str) -> Option<DateTime<Tz>> {
+	let params = params?;
+	let datetime = NaiveDateTime::parse_from_str(value, "%Y%m%dT%H%M%S").ok()?;
+	// Find TZID parameter and extract its singular value
+	let tz = params
+		.into_iter()
+		.filter(|(name, _)| name == "TZID")
+		.map(|(_, values)| values)
+		.filter(|values| values.len() == 1)
+		.next()
+		.unwrap()
+		.into_iter()
+		.next()
+		.unwrap();
+	let tz: chrono_tz::Tz = tz.parse().ok()?;
+	let datetime = match tz.from_local_datetime(&datetime) {
+		LocalResult::Single(datetime) => Some(datetime),
+		LocalResult::None => None,
+		LocalResult::Ambiguous(_, _) => {
+			warn!(
+				"Ignoring ambiguous datetime '{}' from timezone '{}'",
+				datetime,
+				tz.name()
+			);
+			None
+		}
+	}?;
+	Some(datetime.with_timezone(&datetime.timezone().into()))
+}

src/bot.rs

@@ -0,0 +1,182 @@
+use anyhow::{Error, Result};
+use chrono::Utc;
+use chrono_tz::Europe;
+use diesel::Connection;
+use diesel::ExpressionMethods;
+use diesel::QueryDsl;
+use diesel::RunQueryDsl;
+use teloxide::adaptors::Throttle;
+use teloxide::dispatching::dialogue;
+use teloxide::dispatching::dialogue::InMemStorage;
+use teloxide::dispatching::{UpdateFilterExt, UpdateHandler};
+use teloxide::dptree::{case, deps};
+use teloxide::prelude::Dispatcher;
+use teloxide::requests::Requester;
+use teloxide::types::ChatId;
+use teloxide::types::ChatMemberKind;
+use teloxide::types::ChatMemberUpdated;
+use teloxide::types::MessageId;
+use teloxide::types::{Message, Update};
+use teloxide::utils::command::BotCommands;
+use teloxide::Bot;
+
+use crate::appointment::fetch_next_appointment;
+use crate::db::ChatInfo;
+use crate::db::DbChat;
+use crate::{schema, Database};
+
+#[derive(BotCommands, Clone)]
+#[command(rename_rule = "lowercase")]
+pub enum Command {
+	#[command()]
+	SetCalendar,
+}
+
+pub async fn spawn(bot: Throttle<Bot>, db: Database) {
+	Dispatcher::builder(bot, build_handler_chain())
+		.dependencies(deps![db, InMemStorage::<()>::new()])
+		.build()
+		.dispatch()
+		.await;
+}
+
+fn build_handler_chain() -> UpdateHandler<Error> {
+	let command_handler = teloxide::filter_command::<Command, _>()
+		.branch(case![Command::SetCalendar].endpoint(set_calendar));
+
+	let my_chat_member_handler = Update::filter_my_chat_member().endpoint(handle_my_chat_member);
+
+	let message_handler = Update::filter_message().branch(command_handler);
+
+	dialogue::enter::<Update, InMemStorage<()>, (), _>()
+		.branch(my_chat_member_handler)
+		.branch(message_handler)
+}
+
+async fn handle_my_chat_member(msg: ChatMemberUpdated, db: Database) -> Result<(), Error> {
+	match msg.new_chat_member.kind {
+		ChatMemberKind::Left | ChatMemberKind::Banned(_) => {
+			db.lock().await.transaction::<_, Error, _>(|db| {
+				use schema::chat::dsl::*;
+				diesel::delete(chat.filter(telegram_id.eq(msg.chat.id.0))).execute(db)?;
+				Ok(())
+			})?;
+		}
+		_ => {}
+	}
+	Ok(())
+}
+
+async fn set_calendar(bot: Throttle<Bot>, msg: Message, db: Database) -> Result<(), Error> {
+	let url = msg.text().map(|url| url.splitn(2, " ").nth(1)).flatten();
+	if url.is_none() {
+		return Ok(());
+	}
+	let url = url.unwrap().trim();
+	if !url.starts_with("http") {
+		return Ok(());
+	}
+
+	db.lock().await.transaction::<_, Error, _>(|db| {
+		use schema::chat::dsl::*;
+		diesel::delete(chat.filter(telegram_id.eq(msg.chat.id.0))).execute(db)?;
+		diesel::insert_into(chat)
+			.values((telegram_id.eq(msg.chat.id.0), calendar.eq(url)))
+			.execute(db)?;
+		Ok(())
+	})?;
+
+	let mut chat_info = ChatInfo::<Utc> {
+		id: msg.chat.id.0,
+		calendar: url.to_owned(),
+		next_appointment: None,
+		last_reminder: None,
+		pinned_message_id: None,
+	};
+
+	fetch_and_announce_appointment(&bot, &mut chat_info, &db).await?;
+
+	Ok(())
+}
+
+pub async fn fetch_and_announce_appointment(
+	bot: &Throttle<Bot>,
+	chat_info: &mut ChatInfo<Utc>,
+	db: &Database,
+) -> Result<()> {
+	let appointment = match fetch_next_appointment(&chat_info.calendar).await? {
+		Some(appointment) => appointment,
+		None => return Ok(()),
+	};
+	let appointment = appointment.with_timezone(&Utc);
+
+	let entry = db.lock().await.transaction(|db| {
+		use schema::chat::dsl::*;
+		chat.filter(telegram_id.eq(chat_info.id))
+			.first::<DbChat>(db)
+	})?;
+
+	let entry = ChatInfo::from(entry);
+
+	let new_appointment = entry
+		.next_appointment
+		.as_ref()
+		.map(|db_appointment| db_appointment.start != appointment.start)
+		.unwrap_or(true);
+
+	if !new_appointment {
+		return Ok(());
+	}
+
+	let now = Utc::now();
+
+	let old_appointment_running = entry
+		.next_appointment
+		.as_ref()
+		.map(|appointment| appointment.start <= now && appointment.end >= now)
+		.unwrap_or(false);
+
+	if old_appointment_running {
+		return Ok(());
+	}
+
+	let date_str = appointment
+		.start
+		.with_timezone(&Europe::Berlin)
+		.format("%d.%m.%Y %H:%M");
+
+	let announcement = bot
+		.send_message(
+			ChatId(chat_info.id),
+			format!("Nächster Termin: {}", date_str),
+		)
+		.await?;
+
+	db.lock().await.transaction(|db| {
+		use schema::chat::dsl::*;
+		diesel::update(chat.filter(telegram_id.eq(chat_info.id)))
+			.set((
+				next_appointment_start.eq(appointment.start.timestamp()),
+				next_appointment_end.eq(appointment.end.timestamp()),
+				last_reminder.eq(now.timestamp()),
+				pinned_message_id.eq(announcement.id.0),
+			))
+			.execute(db)
+	})?;
+
+	chat_info.next_appointment = Some(appointment);
+	chat_info.last_reminder = Some(now);
+	chat_info.pinned_message_id = Some(announcement.id.0);
+
+	if let Some(pinned_message_id) = entry.pinned_message_id {
+		let mut unpin_message = bot.unpin_chat_message(ChatId(chat_info.id));
+		unpin_message.message_id = Some(MessageId(pinned_message_id));
+		unpin_message.await?;
+	}
+
+	let mut pin_message = bot.pin_chat_message(announcement.chat.id, announcement.id);
+	pin_message.disable_notification = Some(false);
+	pin_message.await?;
+
+	Ok(())
+}

src/db.rs

@@ -0,0 +1,53 @@
+use chrono::{DateTime, NaiveDateTime, TimeZone, Utc};
+use diesel::Queryable;
+
+use crate::appointment::Appointment;
+
+#[derive(Queryable)]
+pub struct DbChat {
+	_id: i32,
+	telegram_id: i64,
+	calendar: String,
+	next_appointment_start: Option<i64>,
+	next_appointment_end: Option<i64>,
+	last_reminder: Option<i64>,
+	pinned_message: Option<i32>,
+}
+
+pub struct ChatInfo<Tz: TimeZone> {
+	pub id: i64,
+	pub calendar: String,
+	pub next_appointment: Option<Appointment<Tz>>,
+	pub last_reminder: Option<DateTime<Tz>>,
+	pub pinned_message_id: Option<i32>,
+}
+
+impl From<DbChat> for ChatInfo<Utc> {
+	fn from(db_chat: DbChat) -> Self {
+		let next_appointment = db_chat
+			.next_appointment_start
+			// Join appointments into single option
+			.and_then(|start| Some([start, db_chat.next_appointment_end?]))
+			// Convert timestamps to datetimes
+			.map(|timestamps| {
+				timestamps
+					.map(|timestamp| NaiveDateTime::from_timestamp(timestamp, 0))
+					.map(|date_time| DateTime::<Utc>::from_utc(date_time, Utc))
+			})
+			// Join datetimes into Appointment
+			.map(|[start, end]| Appointment { start, end });
+
+		let last_reminder = db_chat
+			.last_reminder
+			.map(|timestamp| NaiveDateTime::from_timestamp(timestamp, 0))
+			.map(|date_time| DateTime::<Utc>::from_utc(date_time, Utc));
+
+		ChatInfo {
+			id: db_chat.telegram_id,
+			calendar: db_chat.calendar,
+			next_appointment: next_appointment,
+			last_reminder,
+			pinned_message_id: db_chat.pinned_message,
+		}
+	}
+}

src/error.rs

@@ -0,0 +1,9 @@
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum ConfigLoadError {
+	#[error("Failed to open config file: {0}")]
+	OpenFailed(#[source] std::io::Error),
+	#[error("Failed to read config file: {0}")]
+	ReadError(#[source] serde_yaml::Error),
+}

src/main.rs

@@ -0,0 +1,152 @@
+mod appointment;
+mod bot;
+mod db;
+mod error;
+mod schema;
+
+use std::{env, fs::File, io::BufReader, sync::Arc, time::Duration};
+
+use anyhow::Result;
+use async_mutex::Mutex;
+use bot::fetch_and_announce_appointment;
+use chrono::{NaiveTime, Utc};
+use chrono_tz::Europe;
+use db::ChatInfo;
+use diesel::{Connection, RunQueryDsl, SqliteConnection};
+use diesel::{ExpressionMethods, QueryDsl};
+use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
+use error::ConfigLoadError;
+use log::*;
+use serde::{de::Error, Deserialize, Deserializer};
+use teloxide::adaptors::Throttle;
+use teloxide::prelude::RequesterExt;
+use teloxide::requests::Requester;
+use teloxide::types::ChatId;
+use teloxide::{adaptors::throttle::Limits, Bot};
+use tokio::time::interval;
+
+use crate::db::DbChat;
+
+#[derive(Deserialize, Debug)]
+pub struct Config {
+	token: String,
+	data_path: String,
+	poll_interval: u64,
+	#[serde(deserialize_with = "deserialize_time")]
+	reminder_time: NaiveTime,
+}
+
+fn deserialize_time<'de, D: Deserializer<'de>>(deserializer: D) -> Result<NaiveTime, D::Error> {
+	let s: String = Deserialize::deserialize(deserializer)?;
+	NaiveTime::parse_from_str(&s, "%H:%M").map_err(D::Error::custom)
+}
+
+impl Config {
+	fn load() -> Result<Self, ConfigLoadError> {
+		let env_var_name = "CALENDAR_BOT_CONFIG_FILE";
+		let default_filename = "./config.yaml";
+		let path = env::var(env_var_name).unwrap_or_else(|_| {
+			warn!(
+				"Cannot read env var '{}', assuming '{}'",
+				env_var_name, default_filename
+			);
+			default_filename.to_owned()
+		});
+		info!("Reading configuration from {}", path);
+		let file = File::open(path).map_err(ConfigLoadError::OpenFailed)?;
+		let reader = BufReader::new(file);
+		serde_yaml::from_reader(reader).map_err(ConfigLoadError::ReadError)
+	}
+}
+
+const MIGRATIONS: EmbeddedMigrations = embed_migrations!();
+
+pub type Database = Arc<Mutex<SqliteConnection>>;
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() {
+	pretty_env_logger::init();
+	let config = Config::load().unwrap();
+	info!("Connecting to database {}", config.data_path);
+	let mut db = SqliteConnection::establish(&config.data_path).unwrap();
+	db.run_pending_migrations(MIGRATIONS).unwrap();
+	let db = Arc::new(Mutex::new(db));
+
+	let bot = Bot::new(config.token).throttle(Limits::default());
+
+	{
+		let db = db.clone();
+		let bot = bot.clone();
+		tokio::task::spawn(async move {
+			let mut interval = interval(Duration::from_secs(config.poll_interval));
+			interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+			loop {
+				interval.tick().await;
+				// TODO Log the error and continue instead
+				check_task(&bot, config.reminder_time, &db).await.unwrap();
+			}
+		});
+	}
+
+	tokio::time::sleep(Duration::from_secs(10)).await;
+
+	bot::spawn(bot, db).await;
+}
+
+async fn check_task(bot: &Throttle<Bot>, reminder_time: NaiveTime, db: &Database) -> Result<()> {
+	let chats = db.lock().await.transaction(|db| {
+		use schema::chat::dsl::*;
+		chat.load::<DbChat>(db)
+	})?;
+
+	let now = Utc::now().with_timezone(&Europe::Berlin);
+	let today = now.date_naive();
+
+	for chat in chats {
+		let mut chat_info = ChatInfo::from(chat);
+		fetch_and_announce_appointment(bot, &mut chat_info, db).await?;
+
+		let appointment = match chat_info.next_appointment {
+			Some(appointment) => appointment,
+			None => continue,
+		};
+		let appointment = appointment.with_timezone(&Europe::Berlin);
+
+		if appointment.start < now {
+			continue;
+		}
+		if appointment.start.date_naive() != today {
+			continue;
+		}
+
+		let reminder_date_time = now.date().and_time(reminder_time).unwrap();
+
+		if now < reminder_date_time {
+			continue;
+		}
+
+		if chat_info.last_reminder.is_some()
+			&& chat_info.last_reminder.unwrap() >= reminder_date_time
+		{
+			continue;
+		}
+
+		bot.send_message(
+			ChatId(chat_info.id),
+			format!(
+				"Heute um {} Uhr geht's wetier",
+				appointment.start.format("%H:%M")
+			),
+		)
+		.await?;
+
+		db.lock().await.transaction(|db| {
+			use schema::chat::dsl::*;
+			diesel::update(chat.filter(telegram_id.eq(chat_info.id)))
+				.set((last_reminder.eq(now.timestamp()),))
+				.execute(db)
+		})?;
+	}
+
+	Ok(())
+}

src/schema.rs

@@ -0,0 +1,13 @@
+// @generated automatically by Diesel CLI.
+
+diesel::table! {
+    chat (id) {
+        id -> Integer,
+        telegram_id -> BigInt,
+        calendar -> Text,
+        next_appointment_start -> Nullable<BigInt>,
+        next_appointment_end -> Nullable<BigInt>,
+        last_reminder -> Nullable<BigInt>,
+        pinned_message_id -> Nullable<Integer>,
+    }
+}