name: rust-ci

on: push





jobs:
  rust-ci-run:
    name: Run rust-clippy analyzing and tests
    runs-on: ubuntu-latest
    permissions:
      contents: read

    steps:

      # SETUP
      - uses: actions/checkout@v6


      - uses: dtolnay/rust-toolchain@stable

      - uses: Swatinem/rust-cache@v2

      # TESTS
      - name: Run tests
        run: cargo test --all-targets --all-features --release

      # CLIPPY
      - name: Run rust-clippy
        run: cargo clippy --all-targets --all-features -- -D warnings

name = "bytes"
version = "1.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3"

[[package]]
name = "cc"
version = "1.2.52"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cd4932aefd12402b36c60956a4fe0035421f544799057659ff86f923657aada3"
dependencies = [
 "find-msvc-tools",
 "jobserver",
 "libc",
 "shlex",
]

name = "fastrand"
version = "2.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be"

[[package]]
name = "find-msvc-tools"
version = "0.1.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f449e6c6c08c865631d4890cfacf252b3d396c9bcc83adb6623cdb02a8336c41"

[[package]]
name = "flate2"
version = "1.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bfe33edd8e85a12a67454e37f8c75e730830d83e313556ab9ebf9ee7fbeb3bfb"
dependencies = [

dependencies = [
 "icu_normalizer",
 "icu_properties",
]

[[package]]
name = "indexmap"
version = "2.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7714e70437a7dc3ac8eb7e6f8df75fd8eb422675fc7678aff7364301092b1017"
dependencies = [
 "equivalent",
 "hashbrown 0.16.1",
]

[[package]]
name = "ipnet"

checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
dependencies = [
 "spin",
]

[[package]]
name = "libc"
version = "0.2.180"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bcc35a38544a891a5f7c865aca548a982ccb3b8650a5b06d0fd33a10283c56fc"

[[package]]
name = "libm"
version = "0.2.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f9fbbcab51052fe104eb5e5d351cf728d30a5be1fe14d9be8a3b097481fb97de"

 "futures",
 "futures-util",
 "lazy_static",
 "regex",
 "reqwest",
 "rss",
 "sedregex",
 "serde",
 "smol",
 "sqlx",
 "stacked_errors",
 "tgbot",
 "toml",
 "url",
]

[[package]]
name = "rustc-hash"
version = "2.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"

 "futures-sink",
 "pin-project-lite",
 "tokio",
]

[[package]]
name = "toml"
version = "0.9.11+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f3afc9a848309fe1aaffaed6e1546a7a14de1f935dc9d89d32afd9a44bab7c46"
dependencies = [
 "indexmap",
 "serde_core",
 "serde_spanned",
 "toml_datetime",
 "toml_parser",
 "toml_writer",
 "winnow",
]

[[package]]
name = "toml_datetime"
version = "0.7.5+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"

version = "1.0.6+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a3198b4b0a8e11f09dd03e133c0280504d0801269e9afa46362ffde1cbeebf44"
dependencies = [
 "winnow",
]

[[package]]
name = "toml_writer"
version = "1.0.6+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ab16f14aed21ee8bfd8ec22513f7287cd4a91aa92e44edfe2c17ddd004e92607"

[[package]]
name = "tower"
version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d039ad9159c98b70ecfd540b2573b97f7f52c3e8d9f8ad57a24b916a536975f9"
dependencies = [
 "futures-core",

 "quote",
 "syn",
 "synstructure",
]

[[package]]
name = "zerocopy"
version = "0.8.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "668f5168d10b9ee831de31933dc111a459c97ec93225beb307aed970d1372dfd"
dependencies = [
 "zerocopy-derive",
]

[[package]]
name = "zerocopy-derive"
version = "0.8.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2c7962b26b0a8685668b671ee4b54d007a67d4eaf05fda79ac0ecf41e32270f1"
dependencies = [
 "proc-macro2",
 "quote",
 "syn",
]

[[package]]

[package]
name = "rsstg"
version = "0.5.3"
authors = [ "arcade@b1t.name" ]
edition = "2024"
license = "0BSD"
repository = "http://fs.b1t.name/rsstg"

[dependencies]
async-compat = "0.2.5"
atom_syndication = { version = "0.12.4", features = [ "with-serde" ] }
chrono = "0.4.38"
config = { version = "0.15", default-features = false, features = [ "toml" ] }
tgbot = "0.41"
futures = "0.3.30"
futures-util = "0.3.30"
lazy_static = "1.5.0"
regex = "1.10.6"
reqwest = { version = "0.13.1", features = [ "brotli", "socks", "deflate" ]}
rss = "2.0.9"
sedregex = "0.2.5"
serde = "1.0.228"
smol = "2.0.2"
stacked_errors = "0.7.1"
sqlx = { version = "0.8", features = [ "postgres", "runtime-tokio-rustls", "chrono", "macros" ], default-features = false }
toml = "0.9.11+spec-1.1.0"
url = "2.5.8"

[profile.release]
lto = true
codegen-units = 1

Copyright (C) 2020-2026 by Volodymyr Kostyrko <arcade@b1t.name>

Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted.

THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,

use url::Url;

lazy_static! {
	static ref RE_USERNAME: Regex = Regex::new(r"^@([a-zA-Z][a-zA-Z0-9_]+)$").unwrap();
	static ref RE_IV_HASH: Regex = Regex::new(r"^[a-f0-9]{14}$").unwrap();
}

/// Sends an informational message to the message's chat linking to the bot help channel.
pub async fn start (core: &Core, msg: &Message) -> Result<()> {
	core.tg.send("We are open\\. Probably\\. Visit [channel](https://t.me/rsstg_bot_help/3) for details\\.",
		Some(msg.chat.get_id()), Some(MarkdownV2)).await.stack()?;
	Ok(())
}

/// Send the sender's subscription list to the chat.
///
/// Retrieves the message sender's user ID, obtains their subscription list from `core`,
/// and sends the resulting reply into the message chat using MarkdownV2.
pub async fn list (core: &Core, msg: &Message) -> Result<()> {
	let sender = msg.sender.get_user_id()
		.stack_err("Ignoring unreal users.")?;
	let reply = core.list(sender).await.stack()?;
	core.tg.send(reply, Some(msg.chat.get_id()), Some(MarkdownV2)).await.stack()?;
	Ok(())
}

/// Handle channel-management commands that operate on a single numeric source ID.
///
/// This validates that exactly one numeric argument is provided, performs the requested
/// operation (check, clean, enable, delete, disable) against the database or core,
/// and sends the resulting reply to the chat.
///
/// # Parameters
///
/// - `core`: application core containing database and Telegram clients.
/// - `command`: command string (e.g. "/check", "/clean", "/enable", "/delete", "/disable").
/// - `msg`: incoming Telegram message that triggered the command; used to determine sender and chat.
/// - `words`: command arguments; expected to contain exactly one element that parses as a 32-bit integer.
pub async fn command (core: &Core, command: &str, msg: &Message, words: &[String]) -> Result<()> {
	let mut conn = core.db.begin().await.stack()?;
	let sender = msg.sender.get_user_id()
		.stack_err("Ignoring unreal users.")?;
	let reply = if words.len() == 1 {
		match words[0].parse::<i32>() {
			Err(err) => format!("I need a number.\n{}", &err).into(),
			Ok(number) => match command {
				"/check" => core.check(number, false, None).await
					.context("Channel check failed.")?.into(),
				"/clean" => conn.clean(number, sender).await.stack()?,
				"/enable" => conn.enable(number, sender).await.stack()?.into(),
				"/delete" => conn.delete(number, sender).await.stack()?,
				"/disable" => conn.disable(number, sender).await.stack()?.into(),
				_ => bail!("Command {command} {words:?} not handled."),
			},
		}
	} else {
		"This command needs exactly one number.".into()
	};
	core.tg.send(reply, Some(msg.chat.get_id()), None).await.stack()?;
	Ok(())
}

/// Validate command arguments, check permissions and update or add a channel feed configuration in the database.
///
/// This function parses and validates parameters supplied by a user command (either "/update <id> ..." or "/add ..."),
/// verifies the channel username and feed URL, optionally validates an IV hash and a replacement regexp,
/// ensures both the bot and the command sender are administrators of the target channel, and performs the database update.
///
/// # Parameters
///
/// - `command` — the invoked command, expected to be either `"/update"` (followed by a numeric source id) or `"/add"`.
/// - `msg` — the incoming Telegram message; used to derive the command sender and target chat id for the reply.
/// - `words` — the command arguments: for `"/add"` expected `channel url [iv_hash|'-'] [url_re|'-']`; for `"/update"` the first element must be a numeric `source_id` followed by the same parameters.
pub async fn update (core: &Core, command: &str, msg: &Message, words: &[String]) -> Result<()> {
	let sender = msg.sender.get_user_id()
		.stack_err("Ignoring unreal users.")?;
	let mut source_id: Option<i32> = None;
	let at_least = "Requires at least 3 parameters.";
	let mut i_words = words.iter();
	match command {

					Some(thing)
				}
			}
		},
		None => None,
	};
	let chat_id = ChatUsername::from(channel.as_ref());
	let channel_id = core.tg.client.execute(GetChat::new(chat_id.clone())).await.stack_err("gettting GetChat")?.id;
	let chan_adm = core.tg.client.execute(GetChatAdministrators::new(chat_id)).await
		.context("Sorry, I have no access to that chat.")?;
	let (mut me, mut user) = (false, false);
	for admin in chan_adm {
		let member_id = match admin {
			ChatMember::Creator(member) => member.user.id,
			ChatMember::Administrator(member) => member.user.id,
			ChatMember::Left(_)
			| ChatMember::Kicked(_)
			| ChatMember::Member{..}
			| ChatMember::Restricted(_) => continue,
		};
		if member_id == core.tg.me.id {
			me = true;
		}
		if member_id == sender {
			user = true;
		}
	};
	if ! me   { bail!("I need to be admin on that channel."); };
	if ! user { bail!("You should be admin on that channel."); };
	let mut conn = core.db.begin().await.stack()?;
	let update = conn.update(source_id, channel, channel_id, url, iv_hash, url_re, sender).await.stack()?;
	core.tg.send(update, Some(msg.chat.get_id()), None).await.stack()?;
	Ok(())
}

use crate::{
	command,
	sql::Db,
	tg_bot::Tg,
};

use std::{
	borrow::Cow,
	collections::{
		BTreeMap,
		HashSet,
	},
	sync::Arc,
};

use async_compat::Compat;
use chrono::{
	DateTime,
	Local,
};
use lazy_static::lazy_static;
use regex::Regex;
use reqwest::header::LAST_MODIFIED;




use smol::{
	Timer,
	lock::Mutex,
};
use tgbot::{

	handler::UpdateHandler,
	types::{

		ChatPeerId,
		Command,


		ParseMode,

		Update,
		UpdateType,
		UserPeerId,
	},
};
use stacked_errors::{
	Result,

			set.remove(&self.my_id);
		})
	}
}

#[derive(Clone)]
pub struct Core {


	pub tg: Tg,

	pub db: Db,
	running: Arc<Mutex<HashSet<i32>>>,
	http_client: reqwest::Client,
}

pub struct Post {
	uri: String,
	_title: String,
	_authors: String,
	_summary: String,
}

impl Core {
	/// Create a Core instance from configuration and start its background autofetch loop.
	///
	/// The provided `settings` must include:
	/// - `owner` (integer): default chat id to use as the owner/destination,
	/// - `api_key` (string): Telegram bot API key,
	/// - `api_gateway` (string): Telegram API gateway host,
	/// - `pg` (string): PostgreSQL connection string,
	/// - optional `proxy` (string): proxy URL for the HTTP client.
	///
	/// On success returns an initialized `Core` with Telegram and HTTP clients, database connection,
	/// an empty running set for per-id tokens, and a spawned background task that periodically runs
	/// `autofetch`. If any required setting is missing or initialization fails, an error is returned.
	pub async fn new(settings: config::Config) -> Result<Core> {





		let mut client = reqwest::Client::builder();
		if let Ok(proxy) = settings.get_string("proxy") {
			let proxy = reqwest::Proxy::all(proxy).stack()?;
			client = client.proxy(proxy);
		}


		let core = Core {
			tg: Tg::new(&settings).await.stack()?,


			db: Db::new(&settings.get_string("pg").stack()?)?,
			running: Arc::new(Mutex::new(HashSet::new())),
			http_client: client.build().stack()?,

		};

		let clone = core.clone();
		smol::spawn(Compat::new(async move {
			loop {
				let delay = match &clone.autofetch().await {
					Err(err) => {
						if let Err(err) = clone.tg.send(format!("🛑 {err}"), None, None).await {
							eprintln!("Autofetch error: {err:?}");
						};
						std::time::Duration::from_secs(60)
					},
					Ok(time) => *time,
				};
				Timer::after(delay).await;
			}
		})).detach();
		Ok(core)
	}













	/// Fetches the feed for a source, sends any newly discovered posts to the appropriate chat, and records them in the database.
	///
	/// This acquires a per-source guard to prevent concurrent checks for the same `id`. If a check is already running for
	/// the given `id`, the function returns an error. If `last_scrape` is provided, it is sent as the `If-Modified-Since`
	/// header to the feed request. The function parses RSS or Atom feeds, sends unseen post URLs to either the source's
	/// channel (when `real` is true) or the source owner (when `real` is false), and persists posted entries so they are
	/// not reposted later.

	///
	/// `Posted: N` where `N` is the number of posts processed and sent.
	pub async fn check (&self, id: i32, real: bool, last_scrape: Option<DateTime<Local>>) -> Result<String> {
		let mut posted: i32 = 0;
		let mut conn = self.db.begin().await.stack()?;

		let _token = Token::new(&self.running, id).await.stack()?;
		let source = conn.get_source(id, self.tg.owner).await.stack()?;
		conn.set_scrape(id).await.stack()?;
		let destination = ChatPeerId::from(match real {
			true => source.channel_id,
			false => source.owner,
		});
		let mut this_fetch: Option<DateTime<chrono::FixedOffset>> = None;
		let mut posts: BTreeMap<DateTime<chrono::FixedOffset>, Post> = BTreeMap::new();

		let mut builder = self.http_client.get(&source.url);
		if let Some(last_scrape) = last_scrape {
			builder = builder.header(LAST_MODIFIED, last_scrape.to_rfc2822());
		};
		let response = builder.send().await.stack()?;
		#[cfg(debug_assertions)]
		{
			use reqwest::header::{
				CACHE_CONTROL,
				EXPIRES,
			};
			let headers = response.headers();
			let expires = headers.get(EXPIRES);
			let cache = headers.get(CACHE_CONTROL);
			if expires.is_some() || cache.is_some() {
				println!("{} {} {:?} {:?} {:?}", Local::now().to_rfc2822(), &source.url, last_scrape, expires, cache);
			}
		}

									} else {
										&dates[0]
									}
								},
								None => bail!("Feed item misses posting date."),
							}),
						}.stack()?;
						posts.insert(date, Post{
							uri: link.to_string(),
							_title: item.title().unwrap_or("").to_string(),
							_authors: item.author().unwrap_or("").to_string(),
							_summary: item.content().unwrap_or("").to_string(),





						});
					}
				};
			},
			Err(err) => match err {
				rss::Error::InvalidStartTag => {
					match atom_syndication::Feed::read_from(&content[..]) {
						Ok(feed) => {
							for item in feed.entries() {
								let date = item.published()
									.stack_err("Feed item missing publishing date.")?;
								let uri = {
									let links = item.links();
									if links.is_empty() {
										bail!("Feed item missing post links.");
									} else {
										links[0].href().to_string()
									}
								};

								let _authors = item.authors().iter().map(|x| format!("{} <{:?}>", x.name(), x.email())).collect::<Vec<String>>().join(", ");
								let _summary = if let Some(sum) = item.summary() { sum.value.clone() } else { String::new() };
								posts.insert(*date, Post{
									uri,
									_title: item.title().to_string(),
									_authors,
									_summary,
								});
							};
						},
						Err(err) => {
							bail!("Unsupported or mangled content:\n{:?}\n{err}\n{status:#?}\n", &source.url)
						},
					}

				Some(ref x) => sedregex::ReplaceCommand::new(x).stack()?.execute(&post.uri),
				None => post.uri.clone().into(),
			};
			if ! conn.exists(&post_url, id).await.stack()? {
				if this_fetch.is_none() || *date > this_fetch.unwrap() {
					this_fetch = Some(*date);
				};
				self.tg.send( match &source.iv_hash {
					Some(hash) => format!("<a href=\"https://t.me/iv?url={post_url}&rhash={hash}\"> </a>{post_url}"),
					None => format!("{post_url}"),
				}, Some(destination), Some(ParseMode::Html)).await.stack()?;
				conn.add_post(id, date, &post_url).await.stack()?;
				posted += 1;
			};
		};
		posts.clear();
		Ok(format!("Posted: {posted}"))
	}

	/// Determine the delay until the next scheduled fetch and spawn background checks for any overdue sources.
	///
	/// This scans the database queue, spawns background tasks to run checks for sources whose `next_fetch`
	/// is in the past (each task uses a Core clone with the appropriate owner), and computes the shortest
	/// duration until the next `next_fetch`.
	async fn autofetch(&self) -> Result<std::time::Duration> {
		let mut delay = chrono::Duration::minutes(1);
		let now = chrono::Local::now();
		let queue = {
			let mut conn = self.db.begin().await.stack()?;
			conn.get_queue().await.stack()?
		};
		for row in queue {
			if let Some(next_fetch) = row.next_fetch {
				if next_fetch < now {
					if let (Some(owner), Some(source_id), last_scrape) = (row.owner, row.source_id, row.last_scrape) {
						let clone = Core {
							tg: self.tg.with_owner(owner),
							..self.clone()
						};
						let source = {
							let mut conn = self.db.begin().await.stack()?;
							match conn.get_one(owner, source_id).await {
								Ok(Some(source)) => source.to_string(),
								Ok(None) => "Source not found in database?".to_string(),
								Err(err) => format!("Failed to fetch source data:\n{err}"),
							}
						};
						smol::spawn(Compat::new(async move {
							if let Err(err) = clone.check(source_id, true, Some(last_scrape)).await
								&& let Err(err) = clone.tg.send(&format!("🛑 {source}\n{}", encode(&err.to_string())), None, Some(ParseMode::MarkdownV2)).await
							{
								eprintln!("Check error: {err}");


							};
						})).detach();
					}
				} else if next_fetch - now < delay {
					delay = next_fetch - now;
				}
			}

			reply.push(row.to_string());
		};
		Ok(reply.join("\n\n"))
	}
}

impl UpdateHandler for Core {
	/// Dispatches an incoming Telegram update to a matching command handler and reports handler errors to the originating chat.
	///
	/// This method inspects the update; if it contains a message that can be parsed as a bot command,
	/// it executes the corresponding command handler. If the handler returns an error, the error text
	/// is sent back to the message's chat using MarkdownV2 formatting. Unknown commands produce an error
	/// which is also reported to the chat.
	async fn handle (&self, update: Update) {
		if let UpdateType::Message(msg) = update.update_type 
			&& let Ok(cmd) = Command::try_from(msg)
		{
			let msg = cmd.get_message();
			let words = cmd.get_args();
			let command = cmd.get_name();
			let res = match command {
				"/check" | "/clean" | "/enable" | "/delete" | "/disable" => command::command(self, command, msg, words).await,
				"/start" => command::start(self, msg).await,
				"/list" => command::list(self, msg).await,
				"/add" | "/update" => command::update(self, command, msg, words).await,
				any => Err(anyhow!("Unknown command: {any}")),
			};
			if let Err(err) = res 
				&& let Err(err2) = self.tg.send(format!("\\#error\n```\n{err}\n```"),
					Some(msg.chat.get_id()),
					Some(ParseMode::MarkdownV2)
				).await
			{
				dbg!(err2);

			}


		} // TODO: debug log for skipped updates?;
	}
}

//! This is telegram bot to fetch RSS/ATOM feeds and post results on public
//! channels

#![warn(missing_docs)]

mod command;
mod core;
mod sql;
mod tg_bot;

use async_compat::Compat;
use stacked_errors::{
	Result,
	StackableErr,
};
use tgbot::handler::LongPoll;

fn main () -> Result<()> {
	smol::block_on(Compat::new(async {
		async_main().await.unwrap();
	}));

	Ok(())
}

/// Initialises configuration and the bot core, then runs the Telegram long-poll loop.
///
/// This function loads configuration (with a default API gateway), constructs the application
/// core, and starts the long-polling loop that handles incoming Telegram updates.
async fn async_main () -> Result<()> {
	let settings = config::Config::builder()
		.set_default("api_gateway", "https://api.telegram.org").stack()?
		.add_source(config::File::with_name("rsstg"))
		.build()
		.stack()?;

	let core = core::Core::new(settings).await.stack()?;

	LongPoll::new(core.tg.client.clone(), core).run().await;

	Ok(())
}

			.execute(&mut *self.0).await.stack()?.rows_affected() {
			1 => { Ok("Source enabled.") },
			0 => { Ok("Source not found.") },
			_ => { bail!("Database error.") },
		}
	}

	/// Checks whether a post with the given URL exists for the specified source.
	///
	/// # Parameters
	/// - `post_url`: The URL of the post to check.
	/// - `id`: The source identifier (converted to `i64`).
	///
	/// # Returns
	/// `true` if a post with the URL exists for the source, `false` otherwise.
	pub async fn exists <I> (&mut self, post_url: &str, id: I) -> Result<bool>
	where I: Into<i64> {
		let row = sqlx::query("select exists(select true from rsstg_post where url = $1 and source_id = $2) as exists;")
			.bind(post_url)
			.bind(id.into())
			.fetch_one(&mut *self.0).await.stack()?;
		row.try_get("exists")


			.stack_err("Database error: can't check whether post exists.")

	}

	/// Get all pending events for (now + 1 minute)
	pub async fn get_queue (&mut self) -> Result<Vec<Queue>> {
		let block: Vec<Queue> = sqlx::query_as("select source_id, next_fetch, owner, last_scrape from rsstg_order natural left join rsstg_source where next_fetch < now() + interval '1 minute';")
			.fetch_all(&mut *self.0).await.stack()?;
		Ok(block)