Author SHA1 Message Date
SalastilandClaude Sonnet 5.5 9382bc04ea Merge development into master: 1.3.3
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-23 16:26:18 -04:00
SalastilandClaude Sonnet 5.5 5983d71baf 1.3.3
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-23 16:26:14 -04:00
SalastilandClaude Sonnet 5.5 b1862bb1f2 Discord (#338): the green new-member badge closes a server's box for the member's first week
A member who joined a server less than seven days ago wears Discord's green badge, and (presumed, from a
flagging in a server where the account had spoken from the official apps) what is said from another client in
that time is treated as spam. Every channel of such a server is read-only with a banner saying so and when it
ends; it opens by itself at the week's end. The owner is exempt; what the server's own rules ask comes first.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-23 12:27:22 -04:00
SalastilandClaude Sonnet 5.5 64a13fa9fc Telegram links: t.me and tg:// links open the chat (joined if need be) and the post (#26)
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-23 11:56:23 -04:00
SalastilandClaude Sonnet 5.5 c9c98dec3a Telegram folders: the refusal for an empty folder is worded as the library words it (seen on the real account)
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-23 11:27:06 -04:00
SalastilandClaude Sonnet 5.5 430e89fdd3 Telegram: stickers, bot commands in groups, topic unread counts, QR sign-in, chat folders (#26)
- stickers: the account's recent, favourite and installed packs are listed (a few packs at a time, kept
  15 minutes) and one is sent as its document, in a topic too; pictures are the preview Telegram ships
  with each, fetched when drawn
- bot commands in groups: the commands of every bot in a group, read with the group (basic chat or
  supergroup), shared by its topics; a command two bots have is offered once per bot as cmd@bot
- topic unread counts: a topic's own unread and mention counts seed the window's count (Buffer
  unreadSeed/mentionsSeed, set_buffer_unread); reading a topic reads up to the newest message Telegram
  says it holds, and a topic read on another device clears here
- QR sign-in: exportLoginToken shown as a tg://login QR, renewed as Telegram says (updateLoginToken, with
  a 15 s ask-again as a fallback), DC migration through importLoginToken, then the cloud password if
  asked; addTelegramAccountQr and a telegramLoginQr event
- folders: the account's dialog filters are read after the conversations (what each holds by its rules is
  worked out from what is known of every chat), told to the window as telegramFolders, kept current from
  updateDialogFilter*, and edited from the window: create (with its first chat: Telegram keeps no empty
  folder), rename, delete, reorder, and file a chat into/out of the folder it is drawn under, taking it out of
  a rule-held folder as an exception

All unit-tested; none of it run against Telegram yet.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-23 11:19:39 -04:00
SalastilandClaude Sonnet 5.5 35573351a3 Discord (#338 H1-H12): no requests after a 401 or three 403s on a route, a captcha closes its place, heartbeats carry the sequence, the server's own rules checked before a send (verification, onboarding, timeout, slowmode), DMs need five of the account's own messages, empty channels locked, history asked once, the official wire shape and two-step uploads, one send at a time a channel
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-22 22:44:11 -04:00
SalastilandClaude Sonnet 5.5 26389385bc Discord (#338): READY with deduplicated users put back and declared, the gateway stream compressed, no conversation-list request when READY names them, acks while the newest line is read at the bottom
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-22 21:55:57 -04:00
SalastilandClaude Sonnet 5.5 cb01043f2c Discord first-word lock: a system message about the account (its own join) is not the account speaking
The join message of a server is authored by the account and counted as its first word, so a server was unlocked the moment it was joined. Only an ordinary message or a reply counts, live and in search.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-22 20:41:05 -04:00
SalastilandClaude Sonnet 5.5 ec34657d92 Merge development into master: 1.3.2
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-21 23:36:05 -04:00
SalastilandClaude Sonnet 5.5 e9ce99fb42 1.3.2
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-21 23:36:03 -04:00
SalastilandClaude Sonnet 5.5 7943767c47 Discord: safeguards after Dissent's (#338): paced reads that obey rate limits, capabilities in IDENTIFY, no history fetched until a conversation is opened, one member-list subscription a connection, own membership from READY, acks only of what is read, text checked before it is sent
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-21 23:23:16 -04:00
SalastilandClaude Sonnet 5.5 99253cbd5b Merge development into master: 1.3.1
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-21 21:03:46 -04:00
SalastilandClaude Sonnet 5.5 04e4e85f97 1.3.1
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-21 21:03:44 -04:00
SalastilandClaude Sonnet 5.5 b9a79e4b9c Discord: no first word from this client (a server the account has not spoken in, a conversation with nothing in it, are locked), a nonce and allowed_mentions on every message, files checked against the upload limit, a pause between a message and a file
The account's own messages since it joined are looked for in what is stored and then in Discord's search (one request a server, kept); the messages this client sent do not count. The lock is refused in the daemon and shown by the window as the conversation's read-only reason.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-21 20:39:40 -04:00
SalastilandClaude Sonnet 5.5 264ee7a1f5 Merge development into master: 1.3.0, Telegram
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-21 19:03:03 -04:00
SalastilandClaude Sonnet 5.5 6345ac4df6 1.3.0
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-21 19:02:54 -04:00
SalastilandClaude Sonnet 5.5 3bf68e747a Discord: send the install id, locale and timezone with API requests
X-Installation-ID (random, made once, kept in the data dir), X-Discord-Locale,
X-Discord-Timezone and Accept-Language on the API client; system_locale in
IDENTIFY. The user agent stays moho/<version> (nobilis); no super properties.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-21 18:49:18 -04:00
SalastilandClaude Sonnet 5.5 e8b25d99bf Discord: resume across restarts, ask only about DMs that moved on, throttle typing, skip repeat acks
A restart within 150 s of the last heartbeat ack resumes the saved session
(READY and GUILD_CREATE kept in discord-sessions/, replayed through the live
handlers) instead of logging in afresh; a refused resume or a 4003/4007/4009
close discards it. DM catch-up compares last_message_id with what is stored
and skips the rest; typing posts are at most one per 8 s per channel; an ack
is not repeated for the same newest message.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-21 18:19:19 -04:00
SalastilandClaude Sonnet 5.5 58bf5ac83a Discord: Connect anyway lets a held login through (still counted)
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-20 23:52:42 -04:00
SalastilandClaude Sonnet 5.5 8a7d73346e Discord: a limit on fresh logins per account (3 an hour, 8 a day, kept across restarts); the connection is held and the window says why
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-20 23:41:56 -04:00
SalastilandClaude Sonnet 5.5 54f3e3f697 Telegram: the GIF picker - the account's saved GIFs and Telegram's GIF search, sent as the files they are
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-20 18:41:45 -04:00
SalastilandClaude Sonnet 5.5 931cd458af Merge development into master: Telegram
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-19 22:47:35 -04:00
SalastilandClaude Sonnet 5.5 5ab8eafceb Merge development into master: 1.2.1, Steam identity
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-15 17:14:18 -04:00
SalastilandClaude Sonnet 5.5 2355fcb6ac Merge development into master: 1.2.0, Twitch
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-14 21:54:56 -04:00
42 changed files with 6157 additions and 188 deletions

No files matched your search

Generated
+2 -1
View File
@@ -4091,7 +4091,7 @@ dependencies = [
[[package]]
name = "nobilis"
version = "1.2.1"
version = "1.3.3"
dependencies = [
"aes-gcm",
"anyhow",
@@ -4116,6 +4116,7 @@ dependencies = [
"http-body-util",
"hyper",
"hyper-util",
"iana-time-zone",
"image",
"imagesize",
"irc",
+2 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nobilis"
version = "1.2.1"
version = "1.3.3"
edition = "2021"
description = "Chat daemon: IRC, Discord, Sneedchat and Matrix behind one unified JSON model over a Unix socket"
license = "GPL-3.0-or-later"
@@ -127,6 +127,7 @@ sonora = "0.2"
# Soundboard sounds are MP3 or Ogg Vorbis, decoded here to be mixed into a call.
symphonia = { version = "0.5", default-features = false, features = ["mp3", "ogg", "vorbis"] }
imagesize = "0.13"
iana-time-zone = "0.1.65"
# Two calls, both in main.rs: see keep_the_heap_honest() there for why a chat
# daemon has anything to say to its allocator.
+2 -2
View File
@@ -31,7 +31,7 @@ pub async fn list_commands(state: &AppState, account_id: &str, buffer_id: &str,
.get(format!("{API_BASE}/channels/{channel_id}/application-commands/search"))
.query(&params)
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await
.context("reading the commands")?;
if !resp.status().is_success() {
@@ -243,7 +243,7 @@ pub async fn use_component(
let resp = http_client_for(&cfg.token)
.get(format!("{API_BASE}/channels/{channel_id}/messages/{message_id}"))
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await
.context("finding out whose button that is")?;
if !resp.status().is_success() {
+3 -3
View File
@@ -135,7 +135,7 @@ pub async fn list(state: &AppState, account_id: &str, guild_id: &str) -> Result<
.get(format!("{API_BASE}/guilds/{guild_id}/scheduled-events"))
.query(&[("with_user_count", "true")])
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await
.context("reading the events")?;
if resp.status().is_success() {
@@ -153,7 +153,7 @@ pub async fn list(state: &AppState, account_id: &str, guild_id: &str) -> Result<
.get(format!("{API_BASE}/guilds/{guild_id}/scheduled-events/{id}/users"))
.query(&[("limit", "100")])
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await
{
if let Ok(users) = resp.json::<Value>().await {
@@ -380,7 +380,7 @@ pub async fn share_targets(state: &AppState, account_id: &str, guild_id: &str) -
.get(format!("{API_BASE}/users/{user_id}/profile"))
.query(&[("with_mutual_guilds", "true")])
.header("Authorization", &token)
.send()
.send_paced()
.await
{
if let Ok(profile) = resp.json::<Value>().await {
+2 -2
View File
@@ -125,7 +125,7 @@ async fn search_page(token: &str, channel_id: &str, sort_by: &str, archived: boo
.get(format!("{API_BASE}/channels/{channel_id}/threads/search"))
.query(&query)
.header("Authorization", token)
.send()
.send_paced()
.await
.context("reading the forum")?;
if !resp.status().is_success() {
@@ -216,7 +216,7 @@ pub async fn list_forum_posts(state: &AppState, account_id: &str, buffer_id: &st
.get(format!("{API_BASE}/channels/{id}/messages"))
.query(&[("limit", "1"), ("after", "0")])
.header("Authorization", token)
.send()
.send_paced()
.await
.ok()?;
if !resp.status().is_success() {
+428 -47
View File
@@ -7,7 +7,134 @@
use super::*;
pub(super) const GATEWAY_URL: &str = "wss://gateway.discord.gg/?v=10&encoding=json";
pub(super) const GATEWAY_URL: &str = "wss://gateway.discord.gg/?v=10&encoding=json&compress=zlib-stream";
/// The same address asking for the stream compressed, as Discord's web client and Dissent do: what a session was last
/// given to resume at (kept on disk) may not say so, and a gateway that is not asked to compress does not.
fn compressed(url: &str) -> String {
if url.contains("compress=zlib-stream") {
url.to_string()
} else if url.contains('?') {
format!("{url}&compress=zlib-stream")
} else {
format!("{url}?compress=zlib-stream")
}
}
/// Discord's `zlib-stream`: every message is a piece of one zlib stream that lasts as long as the connection, each
/// ending in the four bytes of a sync flush (`00 00 FF FF`) - a message may arrive in several frames, and is whole when
/// the last of them ends so. Inflating needs the whole stream so far, not the frame alone.
pub(super) struct Inflater {
inflate: flate2::Decompress,
pending: Vec<u8>,
}
impl Inflater {
const SYNC_FLUSH: [u8; 4] = [0x00, 0x00, 0xff, 0xff];
pub(super) fn new() -> Self {
Inflater { inflate: flate2::Decompress::new(true), pending: Vec::new() }
}
/// Takes a frame; hands back the message it completes, if it does.
pub(super) fn push(&mut self, frame: &[u8]) -> Result<Option<String>> {
self.pending.extend_from_slice(frame);
if !self.pending.ends_with(&Self::SYNC_FLUSH) {
return Ok(None);
}
let input = std::mem::take(&mut self.pending);
let mut out: Vec<u8> = Vec::with_capacity(input.len() * 4);
let start = self.inflate.total_in();
loop {
let consumed = (self.inflate.total_in() - start) as usize;
if out.capacity() == out.len() {
out.reserve(input.len().max(4096));
}
let (before_in, before_out) = (self.inflate.total_in(), out.len());
self.inflate.decompress_vec(&input[consumed..], &mut out, flate2::FlushDecompress::Sync).context("inflating the gateway's stream")?;
let all_in = (self.inflate.total_in() - start) as usize >= input.len();
// Done when everything given has been taken and the output stopped for want of input, not of room: while
// the buffer is full there may be more to come out of what was taken.
if all_in && out.len() < out.capacity() {
break;
}
if self.inflate.total_in() == before_in && out.len() == before_out {
bail!("the gateway's stream stopped making sense");
}
}
Ok(Some(String::from_utf8(out).context("the gateway's message was not text")?))
}
}
#[cfg(test)]
mod heartbeat_tests {
use super::heartbeat_message;
#[test]
fn a_heartbeat_says_how_far_the_connection_has_got() {
assert_eq!(heartbeat_message(0), r#"{"d":null,"op":1}"#);
assert_eq!(heartbeat_message(42), r#"{"d":42,"op":1}"#);
}
}
#[cfg(test)]
mod inflate_tests {
use super::*;
/// A zlib stream the way Discord makes one: one compressor, each message flushed (sync) at its end.
fn frames(messages: &[&str]) -> Vec<Vec<u8>> {
let mut compress = flate2::Compress::new(flate2::Compression::default(), true);
messages
.iter()
.map(|m| {
let mut out = Vec::with_capacity(m.len() + 64);
let start = compress.total_in();
while ((compress.total_in() - start) as usize) < m.len() {
let done = (compress.total_in() - start) as usize;
compress.compress_vec(&m.as_bytes()[done..], &mut out, flate2::FlushCompress::None).unwrap();
if out.capacity() == out.len() {
out.reserve(64);
}
}
loop {
compress.compress_vec(&[], &mut out, flate2::FlushCompress::Sync).unwrap();
if out.ends_with(&Inflater::SYNC_FLUSH) {
break;
}
out.reserve(64);
}
out
})
.collect()
}
#[test]
fn messages_of_one_stream_are_read_in_turn() {
let big = format!(r#"{{"op":0,"t":"READY","d":{{"name":"{}"}}}}"#, "x".repeat(50_000));
let sent = [r#"{"op":10,"d":{"heartbeat_interval":41250}}"#, big.as_str(), r#"{"op":11}"#];
let mut inflater = Inflater::new();
for (frame, want) in frames(&sent).iter().zip(sent) {
assert_eq!(inflater.push(frame).unwrap().as_deref(), Some(want));
}
}
#[test]
fn a_message_in_two_frames_is_whole_when_the_second_arrives() {
let sent = [r#"{"op":0,"t":"MESSAGE_CREATE","d":{"content":"hello hello hello hello"}}"#];
let all = &frames(&sent)[0];
let (first, second) = all.split_at(all.len() / 2);
let mut inflater = Inflater::new();
assert_eq!(inflater.push(first).unwrap(), None, "not yet");
assert_eq!(inflater.push(second).unwrap().as_deref(), Some(sent[0]));
}
#[test]
fn the_address_asks_for_the_stream_compressed_once() {
assert_eq!(compressed("wss://x/?v=10&encoding=json"), "wss://x/?v=10&encoding=json&compress=zlib-stream");
assert_eq!(compressed(GATEWAY_URL), GATEWAY_URL);
assert_eq!(compressed("wss://x"), "wss://x?compress=zlib-stream");
}
}
/// Discord's gateway close code for "the token in your IDENTIFY payload is
/// invalid/revoked" - documented at discord.com/developers/docs/topics/
@@ -59,6 +186,22 @@ impl std::fmt::Display for GatewayRateLimited {
impl std::error::Error for GatewayRateLimited {}
/// Close codes that say the session is gone and cannot be resumed: not authenticated, an invalid sequence number, a
/// session that timed out. The next connection logs in afresh.
const SESSION_LOST_CODES: [u16; 3] = [4003, 4007, 4009];
/// The gateway closed the connection and ended the session with it.
#[derive(Debug)]
pub(super) struct SessionLost(u16);
impl std::fmt::Display for SessionLost {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "the gateway ended the session (close code {})", self.0)
}
}
impl std::error::Error for SessionLost {}
/// Discord's op 7: move to another server, resuming where this left off.
/// Its own type so the retry loop can tell it from a failure.
#[derive(Debug)]
@@ -73,12 +216,25 @@ impl std::fmt::Display for ReconnectRequested {
impl std::error::Error for ReconnectRequested {}
pub(super) async fn next_json<S>(stream: &mut S) -> Result<Value>
where
S: futures::Stream<Item = std::result::Result<WsMessage, tokio_tungstenite::tungstenite::Error>> + Unpin,
{
next_json_inflating(stream, None).await
}
/// `next_json` for a connection that was asked to compress its stream: binary frames are pieces of it.
pub(super) async fn next_json_inflating<S>(stream: &mut S, mut inflater: Option<&mut Inflater>) -> Result<Value>
where
S: futures::Stream<Item = std::result::Result<WsMessage, tokio_tungstenite::tungstenite::Error>> + Unpin,
{
loop {
match stream.next().await {
Some(Ok(WsMessage::Text(text))) => return serde_json::from_str(&text).context("parsing gateway JSON"),
Some(Ok(WsMessage::Binary(frame))) if inflater.is_some() => {
if let Some(text) = inflater.as_mut().expect("checked").push(&frame)? {
return serde_json::from_str(&text).context("parsing gateway JSON");
}
}
Some(Ok(WsMessage::Close(frame))) => {
if frame.as_ref().is_some_and(|f| u16::from(f.code) == CLOSE_CODE_AUTH_FAILED) {
return Err(GatewayAuthFailed.into());
@@ -86,6 +242,9 @@ where
if frame.as_ref().is_some_and(|f| u16::from(f.code) == CLOSE_CODE_RATE_LIMITED) {
return Err(GatewayRateLimited.into());
}
if let Some(code) = frame.as_ref().map(|f| u16::from(f.code)).filter(|c| SESSION_LOST_CODES.contains(c)) {
return Err(SessionLost(code).into());
}
bail!("connection closed: {frame:?}")
}
Some(Ok(_)) => continue,
@@ -132,7 +291,7 @@ pub fn spawn(state: AppState, config: DiscordAccountConfig) {
state.runtime.insert_task_handle(&account_id, join_handle.abort_handle());
}
pub(super) const RECONNECT_INITIAL_DELAY: Duration = Duration::from_secs(3);
pub(super) const RECONNECT_INITIAL_DELAY: Duration = Duration::from_secs(4);
pub(super) const RECONNECT_MAX_DELAY: Duration = Duration::from_secs(60);
@@ -173,6 +332,11 @@ pub(super) struct GatewaySession {
channel_map: HashMap<String, (String, String)>,
/// Each guild's payload as GUILD_CREATE delivered it.
guild_context: HashMap<String, Value>,
/// The session as kept on disk, for a restart to resume.
persist: super::session_store::SessionFile,
/// What a resumed session is not sent again - READY and the GUILD_CREATEs saved from the last run - to be handled
/// as though they had just arrived, before anything is read from the socket.
pending: std::collections::VecDeque<Value>,
}
/// Enough to pick a session back up: what it was called, where to reconnect,
@@ -183,20 +347,100 @@ pub(super) struct Resume {
seq: u64,
}
/// A heartbeat: its data is the number of the last dispatch received, and null before there has been one.
pub(super) fn heartbeat_message(seq: u64) -> String {
json!({ "op": 1, "d": if seq == 0 { Value::Null } else { Value::from(seq) } }).to_string()
}
/// The gateway capabilities this client declares in IDENTIFY, as a user client does (Dissent's `ningen` sends the same
/// six of the seven it knows): notes fetched lazily, read states and guild settings in their versioned form, user
/// objects deduplicated, guild experiments as populations, and read states for things that are not channels. Each
/// changes only the shape of a field READY carries and this client either ignores (`read_state`, notes, experiments),
/// reads in both shapes (`user_guild_settings`, see `mutes::note_settings`) or has put back (the deduplicated users,
/// see `ready_shape`).
///
/// Not declared: PRIORITIZED_READY_PAYLOAD (32). It splits what READY carries in two, the second part (presences, the
/// rest of each guild's members, who is in which voice channel) arriving as READY_SUPPLEMENTAL after the guilds are
/// registered, and every one of those is read at the moment of registering here. It buys the client a faster first
/// event; the identity it adds is the one bit.
pub(super) const CAPABILITIES: u32 = (1 << 0) | (1 << 2) | (1 << 3) | (1 << 4) | (1 << 6) | (1 << 7);
/// What IDENTIFY says this client is: moho, on Linux, in this machine's language.
fn identify_properties() -> Value {
let mut properties = json!({ "os": "linux", "browser": "nobilis", "device": "nobilis" });
if let Some(locale) = super::install_headers::locale() {
properties["system_locale"] = Value::from(locale);
}
properties
}
/// Conversations asked about this run, found or not: one that Discord will not describe is not asked about again for
/// every message in it.
static ASKED_ABOUT: std::sync::Mutex<Option<std::collections::HashSet<String>>> = std::sync::Mutex::new(None);
/// Looks up a direct message that READY did not list - the one request it costs is made when somebody speaks in it,
/// which is the moment it matters - and registers it like any other.
async fn unknown_dm(state: &AppState, config: &DiscordAccountConfig, channel_id: &str, channel_map: &mut HashMap<String, (String, String)>) {
let first = ASKED_ABOUT.lock().unwrap_or_else(|e| e.into_inner()).get_or_insert_with(Default::default).insert(format!("{}:{channel_id}", config.user_id));
if !first {
return;
}
let resp = http_client_for(&config.token).get(format!("{API_BASE}/channels/{channel_id}")).header("Authorization", &config.token).send_paced().await;
let Ok(resp) = resp else { return };
if !resp.status().is_success() {
return;
}
let Ok(channel) = resp.json::<Value>().await else { return };
// 1 is a direct message and 3 a group.
if matches!(channel["type"].as_i64(), Some(1) | Some(3)) {
register_dm_channel(state, &config.account_id(), &channel, channel_map, &HashMap::new());
}
}
pub(super) async fn run_gateway_with_retry(state: &AppState, config: &DiscordAccountConfig, account_id: &str) {
let mut delay = RECONNECT_INITIAL_DELAY;
let mut session = GatewaySession::default();
let mut session = GatewaySession {
persist: super::session_store::SessionFile::new(account_id, &config.user_id, &config.token),
..Default::default()
};
// A restart within a few minutes of the last session picks it up, as Discord's own client does, rather than
// logging in afresh: see session_store.
if let Some((position, replay)) = session.persist.load(&config.token, chrono::Utc::now().timestamp()) {
tracing::info!("discord[{account_id}]: resuming the session from before the restart");
session.resume = Some(Resume { session_id: position.session_id, url: position.url, seq: position.seq });
session.pending = replay.into();
}
// A login Discord has already refused is not tried again: the person signs in afresh, which is a new token.
if token_is_halted(&config.token) {
state.runtime.set_conn_state(state, account_id, ConnState::AuthFailed, Some("Discord has refused this account's login - sign in again from Accounts"));
return;
}
state.runtime.set_conn_state(state, account_id, ConnState::Connecting, None);
loop {
// A connection that will log in afresh (not pick a session back up) is counted, and held when the account has
// made too many lately: see identify_budget.
if session.resume.is_none() {
hold_for_login_budget(state, account_id).await;
}
let result = std::panic::AssertUnwindSafe(run_gateway(state, config, &mut session)).catch_unwind().await;
let requested_reconnect =
matches!(&result, Ok(Err(e)) if e.downcast_ref::<ReconnectRequested>().is_some());
let rate_limited = matches!(&result, Ok(Err(e)) if e.downcast_ref::<GatewayRateLimited>().is_some());
// A session Discord has ended is not resumed, here or after a restart: the next connection is a login.
if matches!(&result, Ok(Err(e)) if e.downcast_ref::<SessionLost>().is_some()) {
session.resume = None;
session.persist.clear();
session.pending.clear();
session.channel_map.clear();
session.guild_context.clear();
}
let detail = match result {
Ok(Ok(())) => "gateway session ended".to_string(),
Ok(Err(e)) => {
if e.downcast_ref::<GatewayAuthFailed>().is_some() {
tracing::warn!("discord[{account_id}]: {e} - giving up, needs a fresh login");
// Every other request with it stops too: see http::Safety.
halt_token(&config.token);
state.runtime.set_conn_state(state, account_id, ConnState::AuthFailed, Some(&e.to_string()));
return;
}
@@ -229,13 +473,42 @@ pub(super) async fn run_gateway_with_retry(state: &AppState, config: &DiscordAcc
}
}
/// Waits until the account may make a fresh login, saying in the window why it is waiting. The person may let it
/// through sooner ("Connect anyway"); that login is still counted.
async fn hold_for_login_budget(state: &AppState, account_id: &str) {
loop {
match super::identify_budget::ask(account_id) {
Ok(()) => return,
Err(wait) => {
let detail = super::identify_budget::explain(wait);
tracing::warn!("discord[{account_id}]: {detail}");
state.runtime.set_conn_state(state, account_id, ConnState::Connecting, None);
state.runtime.report_retry(state, account_id, &detail, Duration::from_secs(wait.max(1) as u64));
let pass = super::identify_budget::begin_hold(account_id);
let let_through = tokio::select! {
_ = tokio::time::sleep(Duration::from_secs(wait.max(1) as u64)) => false,
_ = pass.notified() => true,
};
super::identify_budget::end_hold(account_id);
if let_through {
tracing::warn!("discord[{account_id}]: connecting past the login limit, as asked");
super::identify_budget::record(account_id);
return;
}
}
}
}
}
pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig, session: &mut GatewaySession) -> Result<()> {
// Disjoint borrows, so the resume state can be updated while the maps
// are being passed around mutably.
let GatewaySession { resume, channel_map, guild_context } = session;
let GatewaySession { resume, channel_map, guild_context, persist, pending } = session;
// Discord asks that a resume go to the url it handed out with the
// session rather than to the front door.
let connect_url = resume.as_ref().map(|r| r.url.clone()).unwrap_or_else(|| GATEWAY_URL.to_string());
let connect_url = compressed(&resume.as_ref().map(|r| r.url.clone()).unwrap_or_else(|| GATEWAY_URL.to_string()));
// One stream for the whole connection, and a new one for the next: the compressor at the far end starts afresh.
let mut inflater = Inflater::new();
let (ws, _resp) = tokio::time::timeout(CONNECT_TIMEOUT, crate::net::route::websocket(&config.token, connect_url.as_str()))
.await
.map_err(|_| anyhow!("timed out connecting to Discord's gateway"))?
@@ -259,7 +532,7 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
}
let _forwarder_guard = AbortOnDrop(forwarder);
let hello = next_json(&mut stream).await.context("waiting for gateway hello")?;
let hello = next_json_inflating(&mut stream, Some(&mut inflater)).await.context("waiting for gateway hello")?;
let heartbeat_interval = hello["d"]["heartbeat_interval"].as_u64().unwrap_or(41_250).max(1);
// Discord's own documented client contract: if an ack (op 11) hasn't
@@ -274,6 +547,10 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
// false-positive as a missed ack.
let ack_pending = Arc::new(AtomicBool::new(false));
let stale_notify = Arc::new(Notify::new());
// The number of the last dispatch received, which a heartbeat says: it tells Discord how far this connection has
// got, as it does for every client. (A resume starts from where the saved session had got to.)
let last_seq = Arc::new(std::sync::atomic::AtomicU64::new(resume.as_ref().map(|r| r.seq).unwrap_or(0)));
let hb_seq = last_seq.clone();
let hb_tx = out_tx.clone();
let hb_ack_pending = ack_pending.clone();
let hb_stale_notify = stale_notify.clone();
@@ -286,7 +563,7 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
hb_stale_notify.notify_one();
break;
}
if hb_tx.send(json!({ "op": 1, "d": Value::Null }).to_string()).is_err() {
if hb_tx.send(heartbeat_message(hb_seq.load(Ordering::SeqCst))).is_err() {
break;
}
}
@@ -299,6 +576,8 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
// Exposed so setAccountStatus can push a presence update onto this
// connection rather than waiting for a reconnect to carry it.
let account_id = config.account_id();
// A connection starts with no subscriptions.
super::presence::forget_member_subscriptions(&account_id);
state.runtime.set_discord_gateway_sender(&account_id, out_tx.clone());
struct ClearSenderOnDrop<'a>(&'a AppState, String);
impl Drop for ClearSenderOnDrop<'_> {
@@ -319,7 +598,8 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
"op": 2,
"d": {
"token": config.token,
"properties": { "os": "linux", "browser": "nobilis", "device": "nobilis" },
"properties": identify_properties(),
"capabilities": CAPABILITIES,
"compress": false,
"large_threshold": 50,
// Carried in IDENTIFY as well as pushed live, so a status set
@@ -347,17 +627,30 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
let mut tally_since = std::time::Instant::now();
loop {
let msg = tokio::select! {
result = next_json(&mut stream) => result?,
_ = stale_notify.notified() => bail!("no heartbeat ack received - connection is zombied"),
// What a resume does not send comes first, from the last run; `replaying` says so to the handlers.
let (msg, replaying) = match pending.front().cloned() {
Some(saved) => {
pending.pop_front();
(saved, true)
}
None => {
let msg = tokio::select! {
result = next_json_inflating(&mut stream, Some(&mut inflater)) => result?,
_ = stale_notify.notified() => bail!("no heartbeat ack received - connection is zombied"),
};
(msg, false)
}
};
let op = msg.get("op").and_then(|v| v.as_i64()).unwrap_or(-1);
// Every dispatch is numbered, and a resume says how far it got.
// Tracked before the match so it counts events this client chose
// not to handle - the server's count includes them either way, and
// resuming from a lower number would replay what was already seen.
if let (Some(seq), Some(r)) = (msg.get("s").and_then(|v| v.as_u64()), resume.as_mut()) {
r.seq = seq;
if let Some(seq) = msg.get("s").and_then(|v| v.as_u64()) {
last_seq.store(seq, Ordering::SeqCst);
if let Some(r) = resume.as_mut() {
r.seq = seq;
}
}
match op {
0 => {
@@ -386,6 +679,15 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
tracing::info!("discord[{account_id}]: resumed");
}
"READY" => {
// A new login has missed what the last one did not: what was asked about is asked afresh.
if !replaying {
super::history::forget_asked(&account_id);
}
// Kept as it came (it is what a resume replays), and read with its users put back: see
// ready_shape. A READY that was not dehydrated is read as it is.
let raw = d;
let hydrated = super::ready_shape::rehydrate(raw);
let d = hydrated.as_ref().unwrap_or(raw);
let username = d["user"]["global_name"]
.as_str()
.filter(|s| !s.is_empty())
@@ -393,6 +695,8 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
.unwrap_or("me")
.to_string();
state.runtime.set_own_identity(&account_id, &username);
// What the account can say of itself to a server's rules: a verified email or phone, how old.
super::send_guard::note_account(&account_id, &d["user"]);
// Nitro, which is what lets this account's emoji be
// used away from the guild they belong to. Read here
// because READY is the only place the payload carries
@@ -411,14 +715,19 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
// Also what an interaction has to name - see
// Runtime::set_discord_gateway_session.
state.runtime.set_discord_gateway_session(&account_id, session_id);
*resume = Some(Resume {
let next = Resume {
session_id: session_id.to_string(),
url: d["resume_gateway_url"]
.as_str()
.map(|u| format!("{u}/?v=10&encoding=json"))
.map(|u| format!("{u}/?v=10&encoding=json&compress=zlib-stream"))
.unwrap_or_else(|| GATEWAY_URL.to_string()),
seq: msg.get("s").and_then(|v| v.as_u64()).unwrap_or(0),
});
};
// A new session replaces what was kept; the saved one handled again is that same one.
if !replaying {
persist.ready(&next.session_id, &next.url, next.seq, raw.clone());
}
*resume = Some(next);
}
if let Some(hash) = d["user"]["avatar"].as_str() {
let url = format!("https://cdn.discordapp.com/avatars/{}/{hash}.png", config.user_id);
@@ -475,8 +784,13 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
}
let mut new_dm_buffers: Vec<(String, String)> = Vec::new();
// What Discord says each conversation's newest message
// is, so catching up asks only about the ones that
// have something newer than is stored.
let mut newest_in: HashMap<String, Option<String>> = HashMap::new();
if let Some(dms) = d["private_channels"].as_array() {
for ch in dms {
note_newest(&mut newest_in, ch);
if let Some(pair) = register_dm_channel(state, &account_id, ch, channel_map, &presences) {
new_dm_buffers.push(pair);
}
@@ -492,15 +806,11 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
update_presence_in_rosters(state, user_id, p["status"].as_str().unwrap_or("offline"), Some(activity));
}
}
// Conversations are the ones people notice a gap in,
// and there are few enough of them to ask about
// directly. Channels wait until opened - a few hundred
// requests on every connect would be rate-limited and
// mostly wasted.
spawn_dm_catch_up(state.clone(), config.clone(), channel_map.clone());
state.runtime.set_conn_state(state, &account_id, ConnState::Connected, None);
tracing::info!("discord[{account_id}]: ready as {username}");
// Saved ready is not yet accepted by Discord: RESUMED says when it is.
if !replaying {
state.runtime.set_conn_state(state, &account_id, ConnState::Connected, None);
}
tracing::info!("discord[{account_id}]: ready as {username}{}", if replaying { " (from the last run)" } else { "" });
// For an account in relatively few guilds, READY's own
// `guilds` array already carries full guild objects
@@ -515,8 +825,11 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
// harmlessly on a guild it's already seen via either
// path (channel_map dedup).
if let Some(guilds) = d["guilds"].as_array() {
for g in guilds {
register_guild_channels(state, config, g, channel_map).await;
// Our own membership of each, from READY's list of them, so that no guild is asked about.
let merged = d["merged_members"].as_array();
for (i, g) in guilds.iter().enumerate() {
let with_me = guild_with_own_member(g, merged.and_then(|m| m.get(i)), &config.user_id);
register_guild_channels(state, config, with_me.as_ref().unwrap_or(g), channel_map).await;
}
}
@@ -526,30 +839,67 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
// traffic for channel ids that never appeared there) -
// fetch the full list over REST as a follow-up rather
// than trusting the gateway snapshot alone.
match http_client_for(&config.token).get(format!("{API_BASE}/users/@me/channels")).header("Authorization", &config.token).send().await {
Ok(resp) => {
let body = resp.text().await.unwrap_or_default();
match serde_json::from_str::<Vec<Value>>(&body) {
Ok(channels) => {
for ch in &channels {
if let Some(pair) = register_dm_channel(state, &account_id, ch, channel_map, &presences) {
new_dm_buffers.push(pair);
}
//
// A session picked up again after a restart already
// has that list, kept with the rest of it - asking
// again is a request for nothing.
let listed: Option<Vec<Value>> = if replaying {
d["moho_dms"].as_array().cloned()
} else if d["private_channels"].as_array().is_some_and(|l| !l.is_empty()) {
// READY names the account's conversations, as it does for Discord's own clients, and
// any it leaves out is found when somebody speaks in it (see `unknown_dm`): no request
// for a list that was just sent.
None
} else {
match http_client_for(&config.token).get(format!("{API_BASE}/users/@me/channels")).header("Authorization", &config.token).send_paced().await {
Ok(resp) => {
let body = resp.text().await.unwrap_or_default();
match serde_json::from_str::<Vec<Value>>(&body) {
Ok(channels) => Some(channels),
Err(e) => {
tracing::warn!("discord[{account_id}]: parsing /users/@me/channels: {e}");
None
}
}
Err(e) => tracing::warn!("discord[{account_id}]: parsing /users/@me/channels: {e}"),
}
Err(e) => {
tracing::warn!("discord[{account_id}]: fetching /users/@me/channels: {e}");
None
}
}
Err(e) => tracing::warn!("discord[{account_id}]: fetching /users/@me/channels: {e}"),
};
for ch in listed.iter().flatten() {
note_newest(&mut newest_in, ch);
if let Some(pair) = register_dm_channel(state, &account_id, ch, channel_map, &presences) {
new_dm_buffers.push(pair);
}
}
if !replaying {
persist.dms(listed.unwrap_or_default());
}
// The direct messages now exist, so the settings
// read at the top of READY have somewhere to land.
mutes::apply(state, &account_id);
spawn_backfill(state.clone(), config.token.clone(), config.user_id.clone(), config.display_name.clone(), new_dm_buffers);
// Conversations are the ones people notice a gap in,
// and there are few enough of them to ask about
// directly - and only the ones Discord says have
// moved on. Channels wait until opened: a few hundred
// requests on every connect would be rate-limited and
// mostly wasted.
// (A resumed session is replayed what it missed.)
if !replaying {
spawn_dm_catch_up(state.clone(), config.clone(), channel_map.clone(), newest_in);
}
// Nor is any conversation's history fetched here: it is read when it is opened (see getBacklog).
let _ = new_dm_buffers;
}
"GUILD_CREATE" => {
if let (false, Some(id)) = (replaying, d["id"].as_str()) {
persist.guild(id, d);
}
register_guild_channels(state, config, d, channel_map).await;
// The channels this guild's settings were about now
// exist. READY carried the mutes before any of them
@@ -640,6 +990,10 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
if let Some(topic) = d.get("topic") {
state.runtime.set_buffer_topic(state, &crate::model::buffer_id(&account_id, &old_name), topic.as_str().unwrap_or_default());
}
// And so is a channel's slowmode.
if let Some(seconds) = d["rate_limit_per_user"].as_u64() {
super::send_guard::set_slowmode(&crate::model::buffer_id(&account_id, &old_name), seconds);
}
continue;
}
state.runtime.remove_buffer(state, &crate::model::buffer_id(&account_id, &old_name));
@@ -716,6 +1070,10 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
// own state rather than needing to guess.
"MESSAGE_ACK" => {
let Some(channel_id) = d["channel_id"].as_str() else { continue };
// Remembered, so what is read here to the same point is not told to Discord a second time.
if let Some(message_id) = d["message_id"].as_str() {
super::send::acked_elsewhere(channel_id, message_id);
}
let Some((name, _)) = channel_map.get(channel_id) else { continue };
state.events.emit(
"bufferRead",
@@ -801,6 +1159,8 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
if d["user"]["id"].as_str() != Some(config.user_id.as_str()) {
continue;
}
// What the server's rules ask of the account: pending, flags, roles, a timeout.
super::send_guard::note_member(state, &account_id, guild_id, Some(d));
if d["pending"].as_bool().unwrap_or(false) {
continue;
}
@@ -819,6 +1179,12 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
}
"GUILD_DELETE" => {
if let Some(id) = d["id"].as_str() {
persist.guild_gone(id);
if d["unavailable"].as_bool() != Some(true) {
super::send_guard::forget_guild(&account_id, id);
}
}
let Some(guild_id) = d["id"].as_str() else { continue };
if d["unavailable"].as_bool() == Some(true) {
continue;
@@ -834,10 +1200,15 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
}
"MESSAGE_CREATE" => {
let channel_id = d["channel_id"].as_str().unwrap_or_default();
// A conversation READY did not name: asked about, once, when it first speaks.
if !channel_map.contains_key(channel_id) && d["guild_id"].is_null() {
unknown_dm(state, config, channel_id, channel_map).await;
}
let Some((buffer_name, kind)) = channel_map.get(channel_id).cloned() else {
tracing::info!("discord[{account_id}]: a message in {channel_id} was dropped - that channel is not one this connection knows ({} are)", channel_map.len());
continue;
};
super::send_guard::heard(state, config, d, &model::buffer_id(&account_id, &buffer_name), kind == "dm");
let author = &d["author"];
let from = author["global_name"]
.as_str()
@@ -862,7 +1233,7 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
let real_msg_id = d["id"].as_str().map(|s| s.to_string());
let avatar_url = author_avatar_url(author);
if author["system"].as_bool() == Some(true) && kind == "dm" {
state.runtime.set_buffer_read_only(state, &model::buffer_id(&account_id, &buffer_name), Some(super::people::OFFICIAL_ONLY));
super::send_guard::set_permission(state, &account_id, &model::buffer_id(&account_id, &buffer_name), Some(super::people::OFFICIAL_ONLY));
}
// Kept against their id as well as put on the message.
// A direct call's voice states carry no member object,
@@ -1133,12 +1504,10 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
mutes::note_settings(state, &account_id, d);
}
"GUILD_MEMBER_ADD" | "GUILD_MEMBER_REMOVE" => {
let Some(guild_id) = d["guild_id"].as_str() else { continue };
if let Some(buffer_id) = state.runtime.discord_member_list_target(&account_id, guild_id) {
request_member_list(state, &buffer_id);
}
}
// Not a reason to ask for the list again: the list is subscribed to, and Discord sends what changes in
// it. Asking at every join and leave of a busy server was a request for what was already coming.
"GUILD_MEMBER_ADD" | "GUILD_MEMBER_REMOVE" => {}
// A purge. Without this every message in it stayed on
// screen, since the per-message deletes are not sent.
@@ -1353,6 +1722,10 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
}
}
}
// Discord asking for a heartbeat now: answered at once, as its own clients answer.
1 => {
let _ = out_tx.send(heartbeat_message(last_seq.load(Ordering::SeqCst)));
}
7 => return Err(ReconnectRequested.into()),
// The resume was refused, or the session is gone. Everything
// carried across goes with it: the next connection gets a fresh
@@ -1360,6 +1733,8 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
// holding buffers for channels this account may no longer be in.
9 => {
*resume = None;
persist.clear();
pending.clear();
channel_map.clear();
guild_context.clear();
bail!("session invalidated by gateway")
@@ -1368,7 +1743,13 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
// sending the *next* one, so a real ack landing between ticks
// is exactly what keeps this connection from ever being
// declared zombied in the first place.
11 => ack_pending.store(false, Ordering::SeqCst),
11 => {
ack_pending.store(false, Ordering::SeqCst);
// Alive as of now: what a restart soon enough will resume.
if let Some(r) = resume.as_ref() {
persist.acked(r.seq);
}
}
_ => {}
}
}
+1 -1
View File
@@ -87,7 +87,7 @@ pub async fn list_gifs(state: &AppState, query: &str) -> Result<Value> {
let resp = http_client_for(&config.token)
.get(url)
.header("Authorization", &config.token)
.send()
.send_paced()
.await
.context("asking Discord for GIFs")?;
if !resp.status().is_success() {
+77 -11
View File
@@ -96,7 +96,7 @@ pub async fn member_verification(state: &AppState, account_id: &str, guild_id: &
let resp = http_client_for(&cfg.token)
.get(format!("{API_BASE}/guilds/{guild_id}/member-verification"))
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await
.context("asking Discord for this server's rules")?;
let status = resp.status();
@@ -201,7 +201,7 @@ pub(super) async fn resync_guild_once(state: &AppState, config: &DiscordAccountC
http_client_for(&config.token)
.get(format!("{API_BASE}/{path}"))
.header("Authorization", &config.token)
.send()
.send_paced()
.await
.ok()?
.json::<Value>()
@@ -243,6 +243,32 @@ pub(super) async fn resync_guild_once(state: &AppState, config: &DiscordAccountC
}
}
/// This account's own member object out of READY's `merged_members`, which lists, for each guild in the order READY
/// names them, the account's own membership of it (with a `user_id` where a member elsewhere has a `user`). Handed to
/// `register_guild_channels` inside the guild's own payload, where it is looked for first, so that a guild whose
/// payload leaves the account out costs no request of its own - one request a guild, at every fresh login, was the
/// alternative.
pub(super) fn guild_with_own_member(guild: &Value, merged: Option<&Value>, user_id: &str) -> Option<Value> {
let has = |list: &Value| list.as_array().is_some_and(|l| l.iter().any(|m| m["user"]["id"].as_str() == Some(user_id)));
if has(&guild["members"]) {
return None;
}
let mut me = merged?
.as_array()?
.iter()
.find(|m| m["user_id"].as_str() == Some(user_id) || m["user"]["id"].as_str() == Some(user_id))?
.clone();
if me["user"]["id"].is_null() {
me["user"] = json!({ "id": user_id });
}
let mut guild = guild.clone();
match guild["members"].as_array_mut() {
Some(list) => list.push(me),
None => guild["members"] = json!([me]),
}
Some(guild)
}
/// This account's own membership of a guild.
///
/// From the payload when it is there, and asked for when it is not. The
@@ -269,7 +295,7 @@ pub(super) async fn own_member(config: &DiscordAccountConfig, guild: &Value) ->
let resp = http_client_for(&config.token)
.get(format!("{API_BASE}/users/@me/guilds/{guild_id}/member"))
.header("Authorization", &config.token)
.send()
.send_paced()
.await
.ok()?;
if !resp.status().is_success() {
@@ -365,6 +391,7 @@ pub(super) fn can_view_channel(is_owner: bool, guild_id: &str, roles: &[Value],
/// written by almost nobody - rules, announcements - and showing a message box
/// that every send is refused from is a worse answer than saying so.
#[allow(clippy::too_many_arguments)]
#[cfg(test)]
pub(super) fn can_send_in_channel(is_owner: bool, guild_id: &str, roles: &[Value], member_role_ids: &[String], user_id: &str, channel: &Value) -> bool {
channel_permissions(is_owner, guild_id, roles, member_role_ids, user_id, channel) & PERM_SEND_MESSAGES != 0
}
@@ -394,6 +421,11 @@ pub(super) async fn register_guild_channels(state: &AppState, config: &DiscordAc
// the way the roles-only version could.
let me = own_member(config, guild).await;
let own_roles = member_role_ids(me.as_ref());
// What the server asks of its members, and what the account is to it: see rules.
super::send_guard::note_guild(&config.account_id(), guild, &config.user_id);
super::send_guard::note_member(state, &config.account_id(), guild_id, me.as_ref());
// The account may not have spoken here: see send_guard. Begun before any channel is registered, so they open locked.
super::send_guard::open_guild(state, config, guild);
// Kept where every path that registers a guild passes, rather than in the
@@ -549,15 +581,20 @@ pub(super) async fn register_guild_channels(state: &AppState, config: &DiscordAc
// Said on the conversation, so the window can put a banner where the
// message box would be. Re-asked every time the guild is read, since
// permissions change under a channel that is already registered.
let read_only = (!can_send_in_channel(is_owner, guild_id, &roles, &member_role_ids, &config.user_id, ch)).then_some(NO_SEND_PERMISSION);
let permissions = channel_permissions(is_owner, guild_id, &roles, &member_role_ids, &config.user_id, ch);
let read_only = (permissions & PERM_SEND_MESSAGES == 0).then_some(NO_SEND_PERMISSION);
if channel_map.contains_key(channel_id) {
state.runtime.set_buffer_read_only(state, &crate::model::buffer_id(&account_id, &name), read_only);
let existing = crate::model::buffer_id(&account_id, &name);
super::send_guard::note_channel(state, &existing, ch, permissions, is_owner);
super::send_guard::set_permission(state, &account_id, &existing, read_only);
continue;
}
let buf = state.runtime.ensure_buffer(state, &account_id, &name, "channel");
state.runtime.set_buffer_read_only(state, &buf.id, read_only);
// The guild is known before the reason is worked out: the lock is the guild's.
state.runtime.set_discord_channel(state, &buf.id, channel_id);
state.runtime.set_discord_guild(&buf.id, guild_id);
super::send_guard::note_channel(state, &buf.id, ch, permissions, is_owner);
super::send_guard::set_permission(state, &account_id, &buf.id, read_only);
state.runtime.set_buffer_group(state, &buf.id, &group_id);
// Where Discord itself puts this channel. Uncategorised channels sit
@@ -704,7 +741,9 @@ pub(super) async fn register_guild_channels(state: &AppState, config: &DiscordAc
},
);
spawn_backfill(state.clone(), config.token.clone(), config.user_id.clone(), config.display_name.clone(), new_channels);
// No history is fetched for these: a channel is read when it is opened (see getBacklog), which is the only time
// anyone looks at it. Fetching the latest of every channel of every server when it is first seen was a request a
// channel, at the moment of a fresh login, for conversations nobody was about to read.
visible
}
@@ -782,7 +821,7 @@ pub async fn list_threads(state: &AppState, account_id: &str, buffer_id: &str) -
requests.push(format!("{API_BASE}/channels/{channel_id}/threads/archived/public?limit=25"));
for url in requests {
let resp = match http_client_for(&cfg.token).get(&url).header("Authorization", &cfg.token).send().await {
let resp = match http_client_for(&cfg.token).get(&url).header("Authorization", &cfg.token).send_paced().await {
Ok(resp) if resp.status().is_success() => resp,
// A channel with no archive, or one this account may not read the
// archive of, is not a failure worth refusing the whole list for.
@@ -953,7 +992,7 @@ pub async fn moderate_member(
if let Some(reason) = reason.filter(|r| !r.trim().is_empty()) {
request = request.header("X-Audit-Log-Reason", reason);
}
let resp = request.send().await.context(doing)?;
let resp = request.send_paced().await.context(doing)?;
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
@@ -977,7 +1016,7 @@ pub async fn set_member_role(
let http = http_client_for(&cfg.token);
let request = if give { http.put(url).json(&json!({})) } else { http.delete(url) };
let doing = if give { "giving them the role" } else { "taking the role away" };
let resp = request.header("Authorization", &cfg.token).send().await.context(doing)?;
let resp = request.header("Authorization", &cfg.token).send_paced().await.context(doing)?;
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
@@ -1011,7 +1050,7 @@ pub async fn open_thread(state: &AppState, account_id: &str, buffer_id: &str, th
let resp = http
.get(format!("{API_BASE}/channels/{thread_id}"))
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await
.context("opening the thread")?;
if !resp.status().is_success() {
@@ -1255,3 +1294,30 @@ mod verification_tests {
assert_eq!(bare[0]["response"], true);
}
}
#[cfg(test)]
mod merged_member_tests {
use super::*;
#[test]
fn a_guild_that_leaves_the_account_out_is_given_the_membership_ready_listed() {
let guild = json!({ "id": "g", "members": [{ "user": { "id": "other" } }] });
let merged = json!([{ "user_id": "me", "roles": ["r1"], "pending": true }, { "user_id": "x" }]);
let with = guild_with_own_member(&guild, Some(&merged), "me").expect("added");
let mine = with["members"].as_array().unwrap().iter().find(|m| m["user"]["id"] == "me").expect("the member is there");
assert_eq!(mine["roles"], json!(["r1"]));
assert!(member_is_pending(Some(mine)), "what the account's membership says survives");
assert_eq!(with["members"].as_array().unwrap().len(), 2, "the others stay");
// And it is what `own_member` would find first, without a request.
assert!(guild_with_own_member(&with, Some(&merged), "me").is_none());
}
#[test]
fn nothing_is_made_up_when_there_is_nothing_to_add() {
let guild = json!({ "id": "g" });
assert!(guild_with_own_member(&guild, None, "me").is_none());
assert!(guild_with_own_member(&guild, Some(&json!([{ "user_id": "someone-else" }])), "me").is_none());
let with = guild_with_own_member(&guild, Some(&json!([{ "user_id": "me" }])), "me").unwrap();
assert_eq!(with["members"][0]["user"]["id"], "me");
}
}
+217 -28
View File
@@ -28,26 +28,20 @@ impl Drop for HistoryFetch<'_> {
}
}
/// Fires off history backfill for a batch of newly-registered buffers as a
/// background task, sequentially with a small delay between requests - a
/// guild can easily have 30+ channels (confirmed live), and firing that
/// many REST requests at once risks Discord's rate limiter; a one-time
/// startup cost taking a few extra seconds in the background is a fine
/// trade for not tripping it.
pub(super) fn spawn_backfill(state: AppState, token: String, user_id: String, display_name: Option<String>, targets: Vec<(String, String)>) {
if targets.is_empty() {
return;
}
tokio::spawn(async move {
for (buffer_id, channel_id) in targets {
backfill_channel_history(&state, &token, &user_id, display_name.as_deref(), &buffer_id, &channel_id).await;
tokio::time::sleep(Duration::from_millis(250)).await;
}
});
/// How many messages a conversation is first opened with. Few: the rest is a scroll away (`extend_history`), and a
/// client that asks for fifty of every conversation it is shown is asking for what nobody reads.
const FIRST_PAGE: &str = "30";
/// The time a message was sent, from its id: a snowflake's top bits are milliseconds since Discord's epoch. How a
/// conversation's last activity is known without fetching anything of it.
pub(super) fn snowflake_secs(id: &str) -> Option<i64> {
const DISCORD_EPOCH_MS: u64 = 1_420_070_400_000;
let id: u64 = id.parse().ok()?;
Some(((id >> 22) + DISCORD_EPOCH_MS) as i64 / 1000)
}
/// Backfills up to 50 recent messages for a channel/DM with no scrollback
/// yet. Unlike IRC (which has no server-side history at all - nobilis only
/// Backfills the most recent messages for a channel/DM with no scrollback
/// yet - when it is first opened, and not before. Unlike IRC (which has no server-side history at all - nobilis only
/// ever knows what it's personally seen live), Discord actually retains
/// and exposes history, so starting every buffer blank here would look
/// broken compared to Discord's own client. Seed-once, not sync: a buffer
@@ -61,7 +55,7 @@ pub(super) fn spawn_backfill(state: AppState, token: String, user_id: String, di
/// "notification" event, neither of which should fire for messages from
/// potentially weeks ago just because this is the first time nobilis has
/// seen the channel.
pub(super) async fn backfill_channel_history(state: &AppState, token: &str, user_id: &str, own_display_name: Option<&str>, buffer_id: &str, channel_id: &str) {
pub async fn backfill_channel_history(state: &AppState, token: &str, user_id: &str, own_display_name: Option<&str>, buffer_id: &str, channel_id: &str) {
match state.store.has_messages(buffer_id) {
Ok(true) => return,
Ok(false) => {}
@@ -73,9 +67,9 @@ pub(super) async fn backfill_channel_history(state: &AppState, token: &str, user
let resp = match http_client_for(token)
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
.query(&[("limit", "50")])
.query(&[("limit", FIRST_PAGE)])
.header("Authorization", token)
.send()
.send_paced()
.await
{
Ok(r) => r,
@@ -127,7 +121,7 @@ pub async fn refill_history(state: &AppState, buffer_id: &str, limit: u32) -> Re
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
.query(&[("limit", limit.clamp(1, 100).to_string().as_str())])
.header("Authorization", &config.token)
.send()
.send_paced()
.await
.context("re-reading history")?;
if resp.status() == reqwest::StatusCode::TOO_MANY_REQUESTS {
@@ -164,12 +158,99 @@ pub async fn refill_history(state: &AppState, buffer_id: &str, limit: u32) -> Re
Ok(missing.len())
}
/// Asks every open conversation what it missed, one at a time.
/// What has been asked of Discord about each conversation's history this connection, so it is not asked again for what
/// it has already said. A conversation is caught up once in a while, not at every opening of it (its messages arrive
/// live); older history is asked for until its start is reached and never after; a failed ask waits before the next.
#[derive(Default)]
pub struct Asked {
caught_up: HashMap<String, std::time::Instant>,
exhausted: std::collections::HashSet<String>,
failed: HashMap<String, std::time::Instant>,
}
/// How long after a catch-up another is not made for the same conversation.
const CATCH_UP_EVERY: Duration = Duration::from_secs(10 * 60);
/// How long after a failed request for older history the next one waits.
const FAILED_WAIT: Duration = Duration::from_secs(60);
impl Asked {
pub fn catch_up_due(&mut self, buffer: &str, now: std::time::Instant) -> bool {
if self.caught_up.get(buffer).is_some_and(|at| now.duration_since(*at) < CATCH_UP_EVERY) {
return false;
}
self.caught_up.insert(buffer.to_string(), now);
true
}
/// The catch-up did not come to anything: the next opening asks again.
pub fn catch_up_failed(&mut self, buffer: &str) {
self.caught_up.remove(buffer);
}
pub fn older_due(&mut self, buffer: &str, now: std::time::Instant) -> bool {
!self.exhausted.contains(buffer) && !self.failed.get(buffer).is_some_and(|at| now.duration_since(*at) < FAILED_WAIT)
}
/// Asked for `limit` and got `got`: fewer than asked for is the start of the conversation.
pub fn older_answered(&mut self, buffer: &str, got: usize, limit: usize) {
self.failed.remove(buffer);
if got < limit {
self.exhausted.insert(buffer.to_string());
}
}
pub fn older_failed(&mut self, buffer: &str, now: std::time::Instant) {
self.failed.insert(buffer.to_string(), now);
}
/// A new login has missed what the last one did not: everything of the account's is asked about afresh.
pub fn forget_account(&mut self, account: &str) {
let prefix = format!("{account}|");
self.caught_up.retain(|b, _| !b.starts_with(&prefix));
self.failed.retain(|b, _| !b.starts_with(&prefix));
}
}
static ASKED: std::sync::Mutex<Option<Asked>> = std::sync::Mutex::new(None);
fn asked<R>(f: impl FnOnce(&mut Asked) -> R) -> R {
f(ASKED.lock().unwrap_or_else(|e| e.into_inner()).get_or_insert_with(Asked::default))
}
pub(super) fn forget_asked(account: &str) {
asked(|a| a.forget_account(account));
}
/// Notes what a channel object says its newest message is: `None` for a
/// conversation with no messages, nothing at all when it does not say.
pub(super) fn note_newest(into: &mut HashMap<String, Option<String>>, channel: &Value) {
let (Some(id), Some(last)) = (channel["id"].as_str(), channel.get("last_message_id")) else { return };
into.insert(id.to_string(), last.as_str().map(String::from));
}
/// Whether Discord's newest message is newer than the newest stored. Ids are
/// snowflakes, which grow with time; one that will not parse is taken as
/// newer, since asking is the safe answer.
fn has_newer(theirs: Option<&str>, ours: Option<&str>) -> bool {
let Some(theirs) = theirs else { return false };
let Some(ours) = ours else { return true };
match (theirs.parse::<u64>(), ours.parse::<u64>()) {
(Ok(t), Ok(o)) => t > o,
_ => true,
}
}
/// Asks the open conversations that moved on what they missed, one at a time.
///
/// Spaced out deliberately: this runs right after connecting, alongside
/// everything else a fresh connection does, and a burst of requests is the
/// quickest way to be rate-limited by a service that has just let us in.
pub(super) fn spawn_dm_catch_up(state: AppState, config: DiscordAccountConfig, channels: HashMap<String, (String, String)>) {
pub(super) fn spawn_dm_catch_up(
state: AppState,
config: DiscordAccountConfig,
channels: HashMap<String, (String, String)>,
newest_in: HashMap<String, Option<String>>,
) {
tokio::spawn(async move {
let account_id = config.account_id();
let dms: Vec<(String, String)> = channels
@@ -177,6 +258,14 @@ pub(super) fn spawn_dm_catch_up(state: AppState, config: DiscordAccountConfig, c
.filter(|(_, (_, kind))| kind == "dm")
.filter_map(|(channel_id, (name, _))| {
let buffer = state.runtime.list_buffers().into_iter().find(|b| b.account_id == account_id && &b.name == name)?;
// Discord's word for the newest message, when it gave one:
// nothing newer than what is stored means nothing missed.
if let Some(theirs) = newest_in.get(channel_id) {
let ours = state.store.newest_msg_id(&buffer.id).ok().flatten();
if !has_newer(theirs.as_deref(), ours.as_deref()) {
return None;
}
}
Some((buffer.id, channel_id.clone()))
})
.collect();
@@ -206,8 +295,12 @@ pub async fn catch_up_channel(
buffer_id: &str,
channel_id: &str,
) {
if !asked(|a| a.catch_up_due(buffer_id, std::time::Instant::now())) {
return;
}
let Some(_fetch) = HistoryFetch::begin(state, buffer_id) else {
tracing::info!("discord: catching up {buffer_id} skipped - a fetch for it is already under way");
asked(|a| a.catch_up_failed(buffer_id));
return;
};
let result: Result<()> = async {
@@ -218,7 +311,7 @@ pub async fn catch_up_channel(
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
.query(&[("limit", "50"), ("after", after_id.as_str())])
.header("Authorization", token)
.send()
.send_paced()
.await
.context("fetching missed messages")?;
if resp.status().is_success() {
@@ -229,16 +322,22 @@ pub async fn catch_up_channel(
store_history_messages(state, buffer_id, &messages, user_id, own_display_name);
} else {
tracing::info!("discord: catching up {buffer_id}: Discord answered {}", resp.status());
asked(|a| a.catch_up_failed(buffer_id));
}
Ok(())
}
.await;
if let Err(e) = result {
tracing::warn!("discord: catching up {buffer_id}: {e}");
asked(|a| a.catch_up_failed(buffer_id));
}
}
pub async fn extend_history(state: &AppState, token: &str, user_id: &str, own_display_name: Option<&str>, buffer_id: &str, channel_id: &str) {
// The start of a conversation is asked for once: after it is reached, nothing is older.
if !asked(|a| a.older_due(buffer_id, std::time::Instant::now())) {
return;
}
let Some(_fetch) = HistoryFetch::begin(state, buffer_id) else { return };
let result: Result<()> = async {
let Some(before_id) = state.store.oldest_msg_id(buffer_id)? else { return Ok(()) };
@@ -246,18 +345,22 @@ pub async fn extend_history(state: &AppState, token: &str, user_id: &str, own_di
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
.query(&[("limit", "50"), ("before", before_id.as_str())])
.header("Authorization", token)
.send()
.send_paced()
.await
.context("fetching more history")?;
if resp.status().is_success() {
let messages: Vec<Value> = resp.json().await.context("parsing more history")?;
asked(|a| a.older_answered(buffer_id, messages.len(), 50));
store_history_messages(state, buffer_id, &messages, user_id, own_display_name);
} else {
asked(|a| a.older_failed(buffer_id, std::time::Instant::now()));
}
Ok(())
}
.await;
if let Err(e) = result {
tracing::warn!("discord: extending history for {buffer_id}: {e}");
asked(|a| a.older_failed(buffer_id, std::time::Instant::now()));
}
}
@@ -279,7 +382,7 @@ pub async fn load_newer(state: &AppState, account_id: &str, buffer_id: &str, mes
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
.query(&[("limit", "50"), ("after", message_id)])
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await
.context("reading the rest of the channel")?;
if !resp.status().is_success() {
@@ -299,7 +402,7 @@ pub async fn load_context(state: &AppState, account_id: &str, buffer_id: &str, m
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
.query(&[("limit", "50"), ("around", message_id)])
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await
.context("reading that part of the channel")?;
if !resp.status().is_success() {
@@ -318,3 +421,89 @@ pub async fn load_context(state: &AppState, account_id: &str, buffer_id: &str, m
.map(|dt| dt.timestamp())
.context("Discord did not return that message")
}
#[cfg(test)]
mod catch_up_tests {
use super::*;
use serde_json::json;
/// Discord's own documentation's example: 175928847299117063 was made at 2016-04-30 11:18:25.796 UTC.
#[test]
fn the_time_of_a_message_is_read_from_its_id() {
assert_eq!(snowflake_secs("175928847299117063"), Some(1_462_015_105));
assert_eq!(snowflake_secs("not a number"), None);
// What `search::snowflake_at` makes of a time reads back as that time.
assert_eq!(snowflake_secs(&super::search::snowflake_at(1_700_000_000)), Some(1_700_000_000));
}
#[test]
fn a_conversation_is_asked_about_only_when_discord_has_a_newer_message() {
assert!(has_newer(Some("200"), Some("100")));
assert!(!has_newer(Some("100"), Some("100")));
assert!(!has_newer(Some("99"), Some("100")));
// Nothing stored, but something said: a first sight is the backfill's, yet asking is harmless.
assert!(has_newer(Some("5"), None));
// An empty conversation has nothing to miss.
assert!(!has_newer(None, Some("100")));
// Ids that are not numbers are not compared.
assert!(has_newer(Some("abc"), Some("100")));
// Snowflakes outgrow 32 bits.
assert!(has_newer(Some("1500000000000000001"), Some("1500000000000000000")));
}
#[test]
fn what_a_channel_says_of_its_newest_message_is_noted_as_it_says_it() {
let mut into = HashMap::new();
note_newest(&mut into, &json!({ "id": "1", "last_message_id": "10" }));
note_newest(&mut into, &json!({ "id": "2", "last_message_id": null }));
note_newest(&mut into, &json!({ "id": "3" }));
assert_eq!(into.get("1"), Some(&Some("10".to_string())));
assert_eq!(into.get("2"), Some(&None));
assert!(!into.contains_key("3"));
}
}
#[cfg(test)]
mod asked_tests {
use super::*;
use std::time::Instant;
#[test]
fn a_conversation_is_caught_up_once_in_a_while_and_again_after_a_failure() {
let mut a = Asked::default();
let t = Instant::now();
assert!(a.catch_up_due("b", t));
assert!(!a.catch_up_due("b", t + Duration::from_secs(60)), "opened again a minute later");
assert!(a.catch_up_due("other", t));
assert!(a.catch_up_due("b", t + Duration::from_secs(601)), "ten minutes on");
a.catch_up_failed("b");
assert!(a.catch_up_due("b", t + Duration::from_secs(602)), "a failed one is asked again");
}
#[test]
fn older_history_is_asked_for_until_the_start_is_reached_and_not_after() {
let mut a = Asked::default();
let t = Instant::now();
assert!(a.older_due("b", t));
a.older_answered("b", 50, 50);
assert!(a.older_due("b", t), "a full page means more");
a.older_answered("b", 12, 50);
assert!(!a.older_due("b", t + Duration::from_secs(9999)), "the start was reached");
assert!(a.older_due("c", t));
a.older_failed("c", t);
assert!(!a.older_due("c", t + Duration::from_secs(30)));
assert!(a.older_due("c", t + Duration::from_secs(61)));
}
#[test]
fn a_new_login_asks_about_its_own_conversations_afresh() {
let mut a = Asked::default();
let t = Instant::now();
a.catch_up_due("discord:1|chat", t);
a.catch_up_due("discord:2|chat", t);
a.forget_account("discord:1");
assert!(a.catch_up_due("discord:1|chat", t));
assert!(!a.catch_up_due("discord:2|chat", t));
}
}
+332 -8
View File
@@ -54,21 +54,249 @@ async fn pace() {
}
}
/// Where every read waits its turn: when Discord has said how long to hold off - a 429, or a bucket that ran out - no
/// request of any kind goes out before then.
static HOLD_UNTIL: std::sync::Mutex<Option<std::time::Instant>> = std::sync::Mutex::new(None);
/// What is added to a bucket's reset time, as in arikawa's limiter: Discord's own clock and ours do not agree to the
/// millisecond, and arriving on the boundary is arriving early.
const BUCKET_SLACK: Duration = Duration::from_millis(250);
/// How many times a read that was rate limited is made again, waiting as long as it was told to.
const READ_RETRIES: usize = 2;
fn hold_for(wait: Duration) {
let until = std::time::Instant::now() + wait;
let mut held = HOLD_UNTIL.lock().unwrap_or_else(|e| e.into_inner());
if held.is_none_or(|at| at < until) {
*held = Some(until);
}
}
async fn wait_out_hold() {
loop {
let wait = {
let held = HOLD_UNTIL.lock().unwrap_or_else(|e| e.into_inner());
held.map(|at| at.saturating_duration_since(std::time::Instant::now())).unwrap_or_default()
};
if wait.is_zero() {
return;
}
tokio::time::sleep(wait).await;
}
}
/// Notes what a response says about the bucket it counted against: with none left, nothing goes out until it resets.
fn note_bucket(resp: &reqwest::Response) {
let header = |name: &str| resp.headers().get(name).and_then(|v| v.to_str().ok());
if header("x-ratelimit-remaining") == Some("0") {
if let Some(reset) = header("x-ratelimit-reset-after").and_then(|v| v.parse::<f64>().ok()) {
hold_for(Duration::from_secs_f64(reset.clamp(0.0, 30.0)) + BUCKET_SLACK);
}
}
}
/// What this client has learned about where it is no longer welcome, and so does not ask again: a token Discord has
/// refused (every request stops until the person signs in again - a revoked login that keeps asking is a client
/// hammering a dead door), and a route that keeps answering 403 (three times in five minutes closes it for five, so a
/// channel that cannot be read is not asked about at every opening).
#[derive(Default)]
pub struct Safety {
halted: std::collections::HashSet<String>,
circuits: HashMap<(String, String, String), Circuit>,
}
struct Circuit {
refusals: u8,
last: std::time::Instant,
open_until: Option<std::time::Instant>,
}
/// Why a request was not made.
#[derive(Debug, PartialEq, Eq)]
pub enum Blocked {
/// The token was refused (401): nothing more is sent with it.
Refused,
/// The route was refused (403) repeatedly: not for this long.
Closed(Duration),
}
impl std::fmt::Display for Blocked {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Blocked::Refused => write!(f, "Discord has refused this account's login, so nothing more is sent for it - sign in again from Accounts"),
Blocked::Closed(wait) => write!(f, "Discord has refused this three times; not asking again for {} minutes", wait.as_secs().div_ceil(60).max(1)),
}
}
}
impl std::error::Error for Blocked {}
const CIRCUIT_REFUSALS: u8 = 3;
const CIRCUIT_WINDOW: Duration = Duration::from_secs(5 * 60);
const CIRCUIT_OPEN_FOR: Duration = Duration::from_secs(5 * 60);
/// Routes remembered: they include ids, so a few refusals on many messages must not be kept all run.
const MAX_CIRCUITS: usize = 512;
impl Safety {
pub fn preflight_at(&mut self, token: &str, method: &str, path: &str, now: std::time::Instant) -> Result<(), Blocked> {
if self.halted.contains(token) {
return Err(Blocked::Refused);
}
self.circuits.retain(|_, c| match c.open_until {
Some(until) => until > now,
None => now.saturating_duration_since(c.last) < CIRCUIT_WINDOW,
});
let key = (token.to_string(), method.to_string(), path.to_string());
match self.circuits.get(&key).and_then(|c| c.open_until) {
Some(until) => Err(Blocked::Closed(until.saturating_duration_since(now))),
None => Ok(()),
}
}
pub fn record_at(&mut self, token: &str, method: &str, path: &str, status: u16, now: std::time::Instant) {
let key = (token.to_string(), method.to_string(), path.to_string());
if status == 401 {
self.halted.insert(token.to_string());
return;
}
if status != 403 {
self.circuits.remove(&key);
return;
}
if !self.circuits.contains_key(&key) && self.circuits.len() >= MAX_CIRCUITS {
if let Some(oldest) = self.circuits.iter().min_by_key(|(_, c)| c.last).map(|(k, _)| k.clone()) {
self.circuits.remove(&oldest);
}
}
let circuit = self.circuits.entry(key).or_insert(Circuit { refusals: 0, last: now, open_until: None });
if now.saturating_duration_since(circuit.last) >= CIRCUIT_WINDOW {
circuit.refusals = 0;
circuit.open_until = None;
}
circuit.refusals = circuit.refusals.saturating_add(1);
circuit.last = now;
if circuit.refusals >= CIRCUIT_REFUSALS {
circuit.open_until = Some(now + CIRCUIT_OPEN_FOR);
}
}
pub fn halt(&mut self, token: &str) {
self.halted.insert(token.to_string());
}
pub fn is_halted(&self, token: &str) -> bool {
self.halted.contains(token)
}
}
static SAFETY: std::sync::Mutex<Option<Safety>> = std::sync::Mutex::new(None);
fn safety<R>(f: impl FnOnce(&mut Safety) -> R) -> R {
let mut slot = SAFETY.lock().unwrap_or_else(|e| e.into_inner());
f(slot.get_or_insert_with(Safety::default))
}
/// Nothing more is sent with this token: Discord has said it is no good.
pub(super) fn halt_token(token: &str) {
safety(|s| s.halt(token));
}
pub(super) fn token_is_halted(token: &str) -> bool {
safety(|s| s.is_halted(token))
}
/// Which token, method and path a request is for, where it can be told without sending it.
fn inspect(request: &reqwest::RequestBuilder) -> Option<(String, String, String)> {
let built = request.try_clone()?.build().ok()?;
let token = built.headers().get("authorization")?.to_str().ok()?.to_string();
Some((token, built.method().as_str().to_string(), built.url().path().to_string()))
}
/// Asked before a request is sent, told after: see `Safety`.
fn preflight(who: &Option<(String, String, String)>) -> Result<()> {
if let Some((token, method, path)) = who {
safety(|s| s.preflight_at(token, method, path, std::time::Instant::now())).map_err(anyhow::Error::from)?;
}
Ok(())
}
fn note_status(who: &Option<(String, String, String)>, status: reqwest::StatusCode) {
if let Some((token, method, path)) = who {
if status == reqwest::StatusCode::UNAUTHORIZED {
tracing::warn!("discord: a request was refused as unauthorised: nothing more is sent with this login");
}
safety(|s| s.record_at(token, method, path, status.as_u16(), std::time::Instant::now()));
}
}
/// Whether a 429's body is a channel's slowmode (code 20016): the wait is the person's, and sending again when it is
/// over is not for this client to do on their behalf.
fn is_slowmode(body: &[u8]) -> bool {
serde_json::from_slice::<Value>(body).ok().and_then(|v| v["code"].as_u64()) == Some(20016)
}
/// A response put back together once its body has been read, so a caller that reads it again finds it whole.
fn rebuilt(status: reqwest::StatusCode, headers: reqwest::header::HeaderMap, body: bytes::Bytes) -> reqwest::Response {
let mut response = tokio_tungstenite::tungstenite::http::Response::new(body);
*response.status_mut() = status;
*response.headers_mut() = headers;
reqwest::Response::from(response)
}
/// Sending that behaves as a client of an API with rate limits should: waits out any hold Discord has put on this
/// client, notes how much of a bucket is left and waits for it to refill rather than running into the limit, and when
/// it is told to slow down anyway waits as long as it was told and asks again - instead of reading a refusal as an
/// answer. Reads were sent plain, so a 429 was a silent "nothing found" and the next request followed straight on.
pub(super) trait SendPaced {
fn send_paced(self) -> impl std::future::Future<Output = Result<reqwest::Response>> + Send;
}
impl SendPaced for reqwest::RequestBuilder {
async fn send_paced(self) -> Result<reqwest::Response> {
let who = inspect(&self);
preflight(&who)?;
wait_out_hold().await;
let mut pending = Some(self);
for attempt in 0..=READ_RETRIES {
let Some(current) = pending.take() else { break };
// A body that cannot be copied is sent once.
let again = current.try_clone();
let resp = current.send().await.context("talking to Discord")?;
note_bucket(&resp);
note_status(&who, resp.status());
if resp.status() != reqwest::StatusCode::TOO_MANY_REQUESTS || attempt == READ_RETRIES {
return Ok(resp);
}
let Some(again) = again else { return Ok(resp) };
let pause = retry_after(&resp);
tracing::debug!("discord: rate limited on a read, waiting {}ms", pause.as_millis());
hold_for(pause);
wait_out_hold().await;
pending = Some(again);
}
unreachable!("the last attempt returns")
}
}
/// Like `send_write`, for a request whose body cannot be copied - a file
/// streamed from disk - and so is built again for every try.
pub(super) async fn send_write_rebuilt<F>(mut build: F) -> Result<reqwest::Response>
where
F: FnMut() -> Result<reqwest::RequestBuilder>,
{
wait_out_hold().await;
pace().await;
for attempt in 0..=RATE_LIMIT_RETRIES {
let resp = build()?.send().await.context("talking to Discord")?;
note_bucket(&resp);
if resp.status() != reqwest::StatusCode::TOO_MANY_REQUESTS || attempt == RATE_LIMIT_RETRIES {
return Ok(resp);
}
let pause = retry_after(&resp);
tracing::debug!("discord: rate limited, waiting {}ms", pause.as_millis());
tokio::time::sleep(pause).await;
hold_for(pause);
wait_out_hold().await;
}
bail!("Discord kept rate-limiting that request")
}
@@ -85,6 +313,9 @@ where
/// history quickly would be a slow client, and what Discord judges an account
/// on is what it sends.
pub(super) async fn send_write(request: reqwest::RequestBuilder) -> Result<reqwest::Response> {
let who = inspect(&request);
preflight(&who)?;
wait_out_hold().await;
pace().await;
let mut pending = Some(request);
@@ -94,13 +325,23 @@ pub(super) async fn send_write(request: reqwest::RequestBuilder) -> Result<reqwe
// that cannot be cloned is one this cannot retry, and is sent once.
let again = current.try_clone();
let resp = current.send().await.context("talking to Discord")?;
note_bucket(&resp);
note_status(&who, resp.status());
if resp.status() != reqwest::StatusCode::TOO_MANY_REQUESTS || attempt == RATE_LIMIT_RETRIES {
return Ok(resp);
}
let Some(again) = again else { return Ok(resp) };
let pause = retry_after(&resp);
// A channel's slowmode is answered with a 429 too. It is the person's wait, and a message they have not seen
// refused is not one to send again for them when it ends.
let (status, headers) = (resp.status(), resp.headers().clone());
let body = resp.bytes().await.context("reading Discord's answer")?;
if is_slowmode(&body) {
return Ok(rebuilt(status, headers, body));
}
let pause = pause_for(headers.get("retry-after").and_then(|v| v.to_str().ok()));
let Some(again) = again else { return Ok(rebuilt(status, headers, body)) };
tracing::debug!("discord: rate limited, waiting {}ms", pause.as_millis());
tokio::time::sleep(pause).await;
hold_for(pause);
wait_out_hold().await;
pending = Some(again);
}
bail!("Discord kept rate-limiting that request")
@@ -125,7 +366,13 @@ fn pause_for(seconds: Option<&str>) -> Duration {
/// The client for requests made under `key` - the account's token, or its
/// id - routed through Tor or the configured proxy when that account is.
pub(super) fn http_client_for(key: &str) -> reqwest::Client {
discord_client(crate::net::route::router().routed(key))
discord_client(crate::net::route::router().routed(key), true)
}
/// The client for a request that is not for Discord's API but is made on an account's behalf - an upload to the storage
/// address Discord hands out - routed as the account is, and carrying none of the headers that belong to the API.
pub(super) fn bare_client_for(key: &str) -> reqwest::Client {
discord_client(crate::net::route::router().routed(key), false)
}
/// The client for requests that belong to no account yet - the sign-in - or
@@ -133,14 +380,17 @@ pub(super) fn http_client_for(key: &str) -> reqwest::Client {
/// when the add form's switch says so, or when any Discord account is.
pub(super) fn anonymous_client() -> reqwest::Client {
let router = crate::net::route::router();
discord_client(router.routed(&crate::net::route::pending_key("discord")) || router.wanted(SHARED_KEY))
discord_client(router.routed(&crate::net::route::pending_key("discord")) || router.wanted(SHARED_KEY), false)
}
/// Marked while any Discord account is routed.
pub(super) const SHARED_KEY: &str = "discord:any-routed";
fn discord_client(routed: bool) -> reqwest::Client {
crate::net::route::router().client_if("discord", routed, |builder| {
/// `api` clients say which install, language and timezone they are (see `install_headers`); the others - the CDN,
/// the sign-in before there is an account - do not, since nothing there asks.
fn discord_client(routed: bool, api: bool) -> reqwest::Client {
crate::net::route::router().client_if(if api { "discord-api" } else { "discord" }, routed, |builder| {
let builder = if api { builder.default_headers(super::install_headers::headers().clone()) } else { builder };
// Pinned to HTTP/1.1. Enabling reqwest's http2 feature (which the
// file-upload path needs - see upload::http_client) would otherwise
// let every client here negotiate h2 as a side effect, changing the
@@ -282,4 +532,78 @@ mod error_tests {
let text = discord_error_text(reqwest::StatusCode::BAD_GATEWAY, junk, "adding a friend");
assert!(text.contains("502"), "got {text}");
}
/// What a client of an API with rate limits does: told to slow down, it waits as long as it was told and asks
/// again; told a bucket is empty, it lets it refill before anything else goes out. Asked of a local listener.
#[tokio::test]
async fn a_read_waits_out_a_rate_limit_and_a_bucket_that_has_run_dry() {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}/", listener.local_addr().unwrap());
// 1: refused with a one-second wait. 2: answered, saying the bucket is empty for one second. 3: answered.
let server = tokio::spawn(async move {
let mut seen = Vec::new();
for reply in [
"HTTP/1.1 429 Too Many Requests\r\nRetry-After: 1\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
"HTTP/1.1 200 OK\r\nX-RateLimit-Remaining: 0\r\nX-RateLimit-Reset-After: 1\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
] {
let (mut socket, _) = listener.accept().await.unwrap();
let mut buf = vec![0u8; 4096];
let _ = socket.read(&mut buf).await.unwrap();
seen.push(std::time::Instant::now());
socket.write_all(reply.as_bytes()).await.unwrap();
}
seen
});
let client = reqwest::Client::new();
let started = std::time::Instant::now();
let first = client.get(&url).send_paced().await.unwrap();
assert_eq!(first.status(), 200, "the refusal was waited out and the read made again");
assert!(started.elapsed() >= Duration::from_millis(950), "it waited the second it was told to: {:?}", started.elapsed());
let second_started = std::time::Instant::now();
let second = client.get(&url).send_paced().await.unwrap();
assert_eq!(second.status(), 200);
assert!(second_started.elapsed() >= Duration::from_millis(950), "the empty bucket was let refill first: {:?}", second_started.elapsed());
let seen = server.await.unwrap();
assert_eq!(seen.len(), 3);
}
#[test]
fn a_refused_login_stops_everything_and_repeated_403s_close_their_route() {
use std::time::Instant;
let mut s = Safety::default();
let t = Instant::now();
let at = |secs: u64| t + Duration::from_secs(secs);
assert!(s.preflight_at("tok", "GET", "/channels/1/messages", t).is_ok());
// Two refusals are not three.
s.record_at("tok", "GET", "/channels/1/messages", 403, at(1));
s.record_at("tok", "GET", "/channels/1/messages", 403, at(2));
assert!(s.preflight_at("tok", "GET", "/channels/1/messages", at(3)).is_ok());
s.record_at("tok", "GET", "/channels/1/messages", 403, at(4));
let closed = s.preflight_at("tok", "GET", "/channels/1/messages", at(5)).unwrap_err();
assert!(matches!(closed, Blocked::Closed(w) if w > Duration::from_secs(290)));
// Other routes, other methods and other accounts are unaffected.
assert!(s.preflight_at("tok", "GET", "/channels/2/messages", at(5)).is_ok());
assert!(s.preflight_at("tok", "POST", "/channels/1/messages", at(5)).is_ok());
assert!(s.preflight_at("other", "GET", "/channels/1/messages", at(5)).is_ok());
// It opens again after the five minutes, and a success clears the count.
assert!(s.preflight_at("tok", "GET", "/channels/1/messages", at(4 + 301)).is_ok());
s.record_at("tok", "GET", "/channels/9/messages", 403, at(10));
s.record_at("tok", "GET", "/channels/9/messages", 200, at(11));
s.record_at("tok", "GET", "/channels/9/messages", 403, at(12));
s.record_at("tok", "GET", "/channels/9/messages", 403, at(13));
assert!(s.preflight_at("tok", "GET", "/channels/9/messages", at(14)).is_ok(), "403, 200, 403, 403 is not three in a row");
// A 401 ends all of it for that token, for good.
s.record_at("tok", "GET", "/users/@me", 401, at(20));
assert_eq!(s.preflight_at("tok", "POST", "/anything", at(21)), Err(Blocked::Refused));
assert!(s.preflight_at("fresh", "POST", "/anything", at(21)).is_ok(), "a new login is a new token");
}
#[test]
fn a_slowmode_answer_is_told_from_any_other_429() {
assert!(is_slowmode(br#"{"message":"The resource is being rate limited.","retry_after":3.5,"code":20016}"#));
assert!(!is_slowmode(br#"{"message":"You are being rate limited.","retry_after":1.2,"global":false}"#));
assert!(!is_slowmode(b"not json"));
}
}
+179
View File
@@ -0,0 +1,179 @@
//! How often an account may log in to Discord afresh.
//!
//! A Discord client keeps one session open for days and, when the connection drops, *resumes* it. A fresh login (an
//! IDENTIFY) is what happens when a client starts or its session is lost, and a client that does it many times a day
//! looks like automation to Discord's spam detection. Restarting moho is exactly that, so the number of fresh logins an
//! account makes is kept - across restarts, in a file - and when it is over the limit the connection is held until it
//! is under again, with the window told why. A resume costs nothing and is never held.
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Mutex;
/// Fresh logins an account may make in any hour.
pub const PER_HOUR: usize = 3;
/// And in any day.
pub const PER_DAY: usize = 8;
const HOUR: i64 = 3600;
const DAY: i64 = 24 * HOUR;
/// Whether one more fresh login may be made at `now` (Unix seconds), given the times of the earlier ones. Answers
/// the number of seconds to wait when not. The list is kept as it should be saved: only what still counts.
pub fn admit(history: &mut Vec<i64>, now: i64) -> Result<(), i64> {
history.retain(|t| now - t < DAY);
history.sort_unstable();
let in_hour: Vec<i64> = history.iter().copied().filter(|t| now - t < HOUR).collect();
let mut wait = 0;
// Over the hour's limit: when the one that makes room ages out.
if in_hour.len() >= PER_HOUR {
wait = wait.max(in_hour[in_hour.len() - PER_HOUR] + HOUR - now);
}
// Over the day's.
if history.len() >= PER_DAY {
wait = wait.max(history[history.len() - PER_DAY] + DAY - now);
}
if wait > 0 {
return Err(wait.max(1));
}
history.push(now);
Ok(())
}
#[derive(Default, Serialize, Deserialize)]
struct Saved {
/// Account id -> when its fresh logins were made.
accounts: HashMap<String, Vec<i64>>,
}
fn file() -> std::path::PathBuf {
crate::default_data_dir().join("discord-logins.json")
}
fn held() -> &'static Mutex<Option<Saved>> {
static H: Mutex<Option<Saved>> = Mutex::new(None);
&H
}
/// Asks to make a fresh login now. `Err` carries how many seconds to wait first; `Ok` means it has been counted.
pub fn ask(account_id: &str) -> Result<(), i64> {
let now = chrono::Utc::now().timestamp();
let mut slot = held().lock().unwrap_or_else(|e| e.into_inner());
let saved = slot.get_or_insert_with(|| std::fs::read_to_string(file()).ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or_default());
let history = saved.accounts.entry(account_id.to_string()).or_default();
let answer = admit(history, now);
if answer.is_ok() {
if let Ok(text) = serde_json::to_string(saved) {
if let Some(dir) = file().parent() {
let _ = std::fs::create_dir_all(dir);
}
let _ = std::fs::write(file(), text);
}
}
answer
}
/// The accounts whose login is being held now, each with the way to let it through.
type Held = Mutex<HashMap<String, std::sync::Arc<tokio::sync::Notify>>>;
fn waiting() -> &'static Held {
static W: std::sync::OnceLock<Held> = std::sync::OnceLock::new();
W.get_or_init(Default::default)
}
/// Marks an account as being held, answering what to wait on besides the clock.
pub fn begin_hold(account_id: &str) -> std::sync::Arc<tokio::sync::Notify> {
waiting().lock().unwrap_or_else(|e| e.into_inner()).entry(account_id.to_string()).or_default().clone()
}
/// The hold is over, by the clock or by the person.
pub fn end_hold(account_id: &str) {
waiting().lock().unwrap_or_else(|e| e.into_inner()).remove(account_id);
}
/// Lets a held login through now. `false` when nothing was being held, so that pressing the button twice, or after the
/// hold ended, cannot leave a pass waiting for some later hold.
pub fn connect_anyway(account_id: &str) -> bool {
match waiting().lock().unwrap_or_else(|e| e.into_inner()).get(account_id) {
Some(note) => {
note.notify_one();
true
}
None => false,
}
}
/// Counts a login that was let through past the limit: it still counts towards the next ones.
pub fn record(account_id: &str) {
let now = chrono::Utc::now().timestamp();
let mut slot = held().lock().unwrap_or_else(|e| e.into_inner());
let saved = slot.get_or_insert_with(|| std::fs::read_to_string(file()).ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or_default());
let history = saved.accounts.entry(account_id.to_string()).or_default();
history.retain(|t| now - t < DAY);
history.push(now);
if let Ok(text) = serde_json::to_string(saved) {
let _ = std::fs::write(file(), text);
}
}
/// What the window says while a login is held.
pub fn explain(wait: i64) -> String {
let at = chrono::Local::now() + chrono::Duration::seconds(wait);
format!(
"Holding the Discord connection: this account has already logged in {PER_HOUR} times in the last hour (or {PER_DAY} in a day), and many fresh logins can get an account flagged as spam. It will connect at {}.",
at.format("%H:%M")
)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_few_logins_an_hour_are_let_through_and_the_next_waits_for_the_oldest_to_age_out() {
let mut history = Vec::new();
let start = 1_000_000;
assert_eq!(admit(&mut history, start), Ok(()));
assert_eq!(admit(&mut history, start + 60), Ok(()));
assert_eq!(admit(&mut history, start + 120), Ok(()));
// The fourth, ten minutes in: held until the first is an hour old.
assert_eq!(admit(&mut history, start + 600), Err(HOUR - 600));
assert_eq!(history.len(), 3, "a held login is not counted");
// An hour after the first, there is room.
assert_eq!(admit(&mut history, start + HOUR), Ok(()));
}
#[test]
fn the_day_has_its_own_limit_however_the_hours_fall() {
let mut history = Vec::new();
let start = 5_000_000;
// Three an hour for three hours is nine; the eighth is held until the first is a day old.
let mut made = 0;
for hour in 0..3 {
for n in 0..3 {
if admit(&mut history, start + hour * HOUR + n).is_ok() {
made += 1;
}
}
}
assert_eq!(made, PER_DAY);
let wait = admit(&mut history, start + 3 * HOUR).unwrap_err();
assert_eq!(wait, DAY - 3 * HOUR);
}
#[test]
fn a_pass_is_only_given_to_a_login_that_is_being_held() {
assert!(!connect_anyway("discord:test-never-held"), "nothing is held, so nothing is let through");
let _pass = begin_hold("discord:test-held");
assert!(connect_anyway("discord:test-held"));
end_hold("discord:test-held");
assert!(!connect_anyway("discord:test-held"), "after the hold ends there is nothing to let through");
}
#[test]
fn old_logins_are_forgotten() {
let mut history = vec![1, 2, 3, 4, 5];
assert_eq!(admit(&mut history, 10 * DAY), Ok(()));
assert_eq!(history, vec![10 * DAY]);
}
}
+136
View File
@@ -0,0 +1,136 @@
//! What a Discord client tells the API about the install it runs on: which install, in what language, in what
//! timezone.
//!
//! Discord's own clients send these with every request (`X-Installation-ID`, `X-Discord-Locale`,
//! `X-Discord-Timezone`), so an API that sees a long-lived account from one install, in one place, with one
//! language, sees what it expects. They say nothing about *what* is connecting - that stays `moho/<version>
//! (nobilis)` - and the install id is moho's own, made here once and kept: it is not a copy of anything Discord
//! issued.
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
use std::sync::OnceLock;
/// The id of this install: random, made the first time it is needed, and kept in the data directory so that every
/// run - and every account on it - is the same install.
pub fn installation_id() -> String {
let file = crate::default_data_dir().join("discord-installation-id");
if let Some(id) = std::fs::read_to_string(&file).ok().map(|t| t.trim().to_string()).filter(|t| well_formed(t)) {
return id;
}
let bytes: [u8; 16] = rand::random();
let id: String = bytes.iter().map(|b| format!("{b:02x}")).collect();
if let Some(dir) = file.parent() {
let _ = std::fs::create_dir_all(dir);
}
let _ = std::fs::write(&file, &id);
id
}
fn well_formed(id: &str) -> bool {
id.len() == 32 && id.bytes().all(|b| b.is_ascii_hexdigit())
}
/// The language this machine is set to, as Discord writes it (`en-US`): from the environment's usual variables, in
/// their usual order of precedence. `None` when it says nothing, or says "C".
pub fn locale() -> Option<String> {
let raw = ["LC_ALL", "LC_MESSAGES", "LANG"].iter().filter_map(|v| std::env::var(v).ok()).find(|v| !v.is_empty())?;
locale_from(&raw)
}
fn locale_from(raw: &str) -> Option<String> {
// `en_US.UTF-8@euro` -> `en-US`
let tag = raw.split(['.', '@']).next()?.replace('_', "-");
let mut parts = tag.split('-');
let language = parts.next().filter(|l| (2..=3).contains(&l.len()) && l.bytes().all(|b| b.is_ascii_alphabetic()))?;
if language.eq_ignore_ascii_case("C") || language.eq_ignore_ascii_case("POSIX") {
return None;
}
let language = language.to_ascii_lowercase();
match parts.next().filter(|r| r.len() == 2 && r.bytes().all(|b| b.is_ascii_alphabetic())) {
Some(region) => Some(format!("{language}-{}", region.to_ascii_uppercase())),
None => Some(language),
}
}
/// The IANA name of this machine's timezone (`America/New_York`), when it can be told.
pub fn timezone() -> Option<String> {
iana_time_zone::get_timezone().ok().filter(|t| !t.is_empty())
}
/// The headers every API request carries.
pub fn headers() -> &'static HeaderMap {
static HEADERS: OnceLock<HeaderMap> = OnceLock::new();
HEADERS.get_or_init(|| {
let mut pairs: Vec<(&'static str, String)> = vec![("x-installation-id", installation_id())];
if let Some(locale) = locale() {
let language = locale.split('-').next().unwrap_or("en").to_string();
pairs.push(("accept-language", format!("{locale},{language};q=0.9")));
pairs.push(("x-discord-locale", locale));
}
if let Some(zone) = timezone() {
pairs.push(("x-discord-timezone", zone));
}
let mut map = HeaderMap::new();
for (name, value) in pairs {
if let Ok(value) = HeaderValue::from_str(&value) {
map.insert(HeaderName::from_static(name), value);
}
}
map
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_locale_is_written_the_way_discord_writes_it() {
assert_eq!(locale_from("en_US.UTF-8").as_deref(), Some("en-US"));
assert_eq!(locale_from("de_DE@euro").as_deref(), Some("de-DE"));
assert_eq!(locale_from("fr").as_deref(), Some("fr"));
assert_eq!(locale_from("pt-br").as_deref(), Some("pt-BR"));
}
#[test]
fn a_machine_with_no_language_set_sends_none() {
assert_eq!(locale_from("C"), None);
assert_eq!(locale_from("POSIX"), None);
assert_eq!(locale_from("C.UTF-8"), None);
assert_eq!(locale_from(""), None);
assert_eq!(locale_from("12345"), None);
}
#[test]
fn an_installation_id_is_recognised_only_when_whole() {
assert!(well_formed("0123456789abcdef0123456789abcdef"));
assert!(!well_formed("0123"));
assert!(!well_formed("zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz"));
}
/// What actually leaves: the API client carries the headers and still says it is moho, and the client for the CDN
/// and sign-in carries none of them. Asked of a local listener, so nothing reaches Discord.
#[tokio::test]
async fn the_api_client_sends_them_and_the_other_does_not() {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
async fn request_seen_by_a_listener(client: reqwest::Client) -> String {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}/", listener.local_addr().unwrap());
let server = tokio::spawn(async move {
let (mut socket, _) = listener.accept().await.unwrap();
let mut buf = vec![0u8; 8192];
let n = socket.read(&mut buf).await.unwrap();
socket.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").await.unwrap();
String::from_utf8_lossy(&buf[..n]).to_lowercase()
});
let _ = client.get(url).send().await;
server.await.unwrap()
}
let api = request_seen_by_a_listener(crate::backend::discord::http::http_client_for("a-test-token")).await;
assert!(api.contains("x-installation-id: "), "{api}");
assert!(api.contains("user-agent: moho/"), "{api}");
assert!(!api.contains("x-super-properties"), "{api}");
let other = request_seen_by_a_listener(crate::backend::discord::http::anonymous_client()).await;
assert!(!other.contains("x-installation-id"), "{other}");
}
}
+2 -2
View File
@@ -249,7 +249,7 @@ pub(super) async fn run_resign(state: &AppState, buffer_id: &str, mut stale: Vec
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
.query(&[("limit", "50"), ("around", anchor.as_str())])
.header("Authorization", &config.token)
.send()
.send_paced()
.await
.context("fetching a page to re-sign")?;
if !resp.status().is_success() {
@@ -358,7 +358,7 @@ pub async fn refresh_attachments(state: &AppState, buffer_id: &str, message_id:
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
.query(&[("limit", "1"), ("around", message_id)])
.header("Authorization", &config.token)
.send()
.send_paced()
.await
.context("re-fetching the message")?;
if !resp.status().is_success() {
+5 -5
View File
@@ -23,7 +23,7 @@ pub async fn list_pinned(state: &AppState, account_id: &str, buffer_id: &str) ->
let resp = http_client_for(&cfg.token)
.get(format!("{API_BASE}/channels/{channel_id}/pins"))
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await
.context("reading the pins")?;
if !resp.status().is_success() {
@@ -96,7 +96,7 @@ pub async fn set_pinned(state: &AppState, account_id: &str, buffer_id: &str, mes
let http = http_client_for(&cfg.token);
let request = if pinned { http.put(url) } else { http.delete(url) };
let doing = if pinned { "pinning the message" } else { "unpinning the message" };
let resp = request.header("Authorization", &cfg.token).send().await.context(doing)?;
let resp = request.header("Authorization", &cfg.token).send_paced().await.context(doing)?;
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
@@ -601,7 +601,7 @@ pub async fn name_forward(state: &AppState, account_id: &str, buffer_id: &str, m
let resp = http_client_for(&cfg.token)
.get(format!("{API_BASE}/channels/{source_channel}/messages?limit=1&around={source_message}"))
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await;
// A source this account cannot see is the ordinary case for a forward
// out of somebody else's server, and not worth a word to anybody.
@@ -1680,7 +1680,7 @@ pub async fn list_reactors(state: &AppState, account_id: &str, buffer_id: &str,
let resp = http_client_for(&cfg.token)
.get(url)
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await
.context("reading who reacted")?;
if !resp.status().is_success() {
@@ -1743,7 +1743,7 @@ pub async fn reread_embeds(state: &AppState, buffer_id: &str) -> Result<usize> {
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
.query(&[("limit", "50")])
.header("Authorization", &config.token)
.send()
.send_paced()
.await
.context("re-reading the conversation")?;
if !resp.status().is_success() {
+6
View File
@@ -49,10 +49,16 @@ pub mod voice;
pub mod calls;
pub mod commands;
pub mod gateway;
pub mod identify_budget;
pub mod dave;
pub mod golive;
pub mod guilds;
pub mod history;
pub mod install_headers;
pub mod ready_shape;
pub mod rules;
pub mod send_guard;
pub mod session_store;
pub mod http;
pub mod login;
pub mod media;
+11 -4
View File
@@ -27,7 +27,9 @@ pub(super) fn register_dm_channel(
state.runtime.set_discord_channel(state, &buf.id, channel_id);
// Discord's own notices come from an account flagged `system`, and nothing
// sent back to it goes anywhere.
state.runtime.set_buffer_read_only(state, &buf.id, is_system_dm(ch).then_some(OFFICIAL_ONLY));
super::send_guard::set_permission(state, account_id, &buf.id, is_system_dm(ch).then_some(OFFICIAL_ONLY));
// A conversation with nothing in it yet is not started from here: see send_guard.
super::send_guard::note_dm(state, account_id, &buf.id, ch);
// Its notices were saved before their cards could be read whole, and saved
// history is not fetched again, so they are read once a run, a little after
// connecting - one request, for the latest page.
@@ -40,6 +42,11 @@ pub(super) fn register_dm_channel(
}
});
}
// When it was last spoken in, from the id of its last message - which is all that ordering the list by activity
// needs, so nothing is fetched for it.
if let Some(at) = ch["last_message_id"].as_str().and_then(super::history::snowflake_secs) {
state.runtime.raise_buffer_activity(state, &buf.id, at);
}
if let Some(avatar) = dm_avatar_url(ch) {
state.runtime.set_buffer_avatar(state, &buf.id, &avatar);
}
@@ -166,7 +173,7 @@ pub async fn profile(state: &AppState, account_id: &str, buffer_id: &str, user_i
}
}
let response = http_client_for(&config.token).get(&url).header("Authorization", &config.token).send().await;
let response = http_client_for(&config.token).get(&url).header("Authorization", &config.token).send_paced().await;
let Ok(body) = response else { return profile };
let Ok(body) = body.json::<Value>().await else { return profile };
@@ -295,7 +302,7 @@ pub async fn adopt_dm_channel(state: &AppState, account_id: &str, channel_id: &s
let resp = http_client_for(&config.token)
.get(format!("{API_BASE}/channels/{channel_id}"))
.header("Authorization", &config.token)
.send()
.send_paced()
.await
.context("asking about that conversation")?;
if !resp.status().is_success() {
@@ -379,7 +386,7 @@ pub async fn answer_friend_request(state: &AppState, account_id: &str, user_id:
let url = format!("{API_BASE}/users/@me/relationships/{user_id}");
let request = if accept { http.put(url).json(&json!({})) } else { http.delete(url) };
let doing = if accept { "accepting the friend request" } else { "declining the friend request" };
let resp = request.header("Authorization", &cfg.token).send().await.context(doing)?;
let resp = request.header("Authorization", &cfg.token).send_paced().await.context(doing)?;
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
+73 -1
View File
@@ -117,6 +117,46 @@ fn settings_body(status: &str, text: Option<&str>) -> serde_json::Value {
/// does when you open a channel, which is also why the roster only exists for
/// channels somebody is actually looking at.
pub fn request_member_list(state: &AppState, buffer_id: &str) -> bool {
subscribe_member_list(state, buffer_id, false)
}
/// The window the list is kept in no longer matches what Discord holds, so it is asked for again - but not more often
/// than every few seconds, which is the most a roster that keeps falling out of step deserves.
pub(super) fn resync_member_list(state: &AppState, buffer_id: &str) -> bool {
subscribe_member_list(state, buffer_id, true)
}
/// What each server has been subscribed to this session: the channel, and when.
type Subscribed = HashMap<(String, String), (String, std::time::Instant)>;
static SUBSCRIBED: std::sync::Mutex<Option<Subscribed>> = std::sync::Mutex::new(None);
/// The least time between two subscriptions to one server's list that are not a change of channel.
const RESYNC_EVERY: std::time::Duration = std::time::Duration::from_secs(10);
/// Whether a subscription to this channel's list is to be sent: not if it already was, this session, and not a
/// repeat of one sent a moment ago. A connection is subscribed once - as Dissent's is, once a server - and a new one
/// starts again (`forget_member_subscriptions`).
fn subscription_due(account: &str, guild: &str, channel: &str, resync: bool, now: std::time::Instant) -> bool {
let mut slot = SUBSCRIBED.lock().unwrap_or_else(|e| e.into_inner());
let map = slot.get_or_insert_with(HashMap::new);
let key = (account.to_string(), guild.to_string());
match map.get(&key) {
Some((was, at)) if was == channel && (!resync || now.duration_since(*at) < RESYNC_EVERY) => false,
_ => {
map.insert(key, (channel.to_string(), now));
true
}
}
}
/// A new connection holds none of what the last one subscribed to.
pub(super) fn forget_member_subscriptions(account: &str) {
if let Some(map) = SUBSCRIBED.lock().unwrap_or_else(|e| e.into_inner()).as_mut() {
map.retain(|(a, _), _| a != account);
}
}
fn subscribe_member_list(state: &AppState, buffer_id: &str, resync: bool) -> bool {
let Some(buffer) = state.runtime.get_buffer(buffer_id) else { return false };
let account_id = buffer.account_id;
let Some(sender) = state.runtime.discord_gateway_sender(&account_id) else { return false };
@@ -131,6 +171,11 @@ pub fn request_member_list(state: &AppState, buffer_id: &str) -> bool {
// channel, so remember which buffer this was for.
state.runtime.set_discord_member_list_target(&account_id, &guild_id, buffer_id);
// Asked for once. Whoever asked is told it is being sent: the roster arrives, or is already arriving.
if !subscription_due(&account_id, &guild_id, &channel_id, resync, std::time::Instant::now()) {
return true;
}
// Ranges are 100-member windows; one covers any channel we would show.
sender
.send(
@@ -370,7 +415,7 @@ pub(super) fn update_member_list(state: &AppState, buffer_id: &str, d: &Value) {
Applied::Stale => {
tracing::debug!("discord: member window for {buffer_id} is out of step; resyncing");
state.runtime.set_discord_member_window(buffer_id, Vec::new());
request_member_list(state, buffer_id);
resync_member_list(state, buffer_id);
return;
}
}
@@ -629,3 +674,30 @@ mod status_tests {
assert_eq!(settings_body("idle", None)["status"], "idle");
}
}
#[cfg(test)]
mod subscription_tests {
use super::*;
use std::time::{Duration, Instant};
#[test]
fn a_list_is_subscribed_to_once_a_connection_and_again_for_another_channel() {
let t = Instant::now();
assert!(subscription_due("a1", "g", "c1", false, t));
assert!(!subscription_due("a1", "g", "c1", false, t + Duration::from_secs(1)), "the same again is nothing");
assert!(!subscription_due("a1", "g", "c1", false, t + Duration::from_secs(600)), "however long after");
assert!(subscription_due("a1", "g", "c2", false, t + Duration::from_secs(2)), "another channel is a change");
assert!(subscription_due("a1", "g2", "c1", false, t), "another server is its own");
forget_member_subscriptions("a1");
assert!(subscription_due("a1", "g", "c2", false, t), "a new connection starts again");
}
#[test]
fn a_resync_is_not_sent_again_within_a_few_seconds() {
let t = Instant::now();
assert!(subscription_due("a2", "g", "c1", false, t));
assert!(!subscription_due("a2", "g", "c1", true, t + Duration::from_secs(3)));
assert!(subscription_due("a2", "g", "c1", true, t + Duration::from_secs(11)));
assert!(!subscription_due("a2", "g", "c1", true, t + Duration::from_secs(12)));
}
}
+152
View File
@@ -0,0 +1,152 @@
//! READY as a client that asked for DEDUPE_USER_OBJECTS receives it, put back as every reader here expects it.
//!
//! With that capability Discord "dehydrates" READY (<https://docs.discord.food/topics/gateway-events#ready>): every
//! user object is moved into one `users` table and left behind as an id - `user_id` in members, presences and
//! relationships, `recipient_ids` in private channels - each guild's `members` is collapsed into READY's
//! `merged_members` (one list a guild, in the order of `guilds`), and `presences`, with each guild's, becomes
//! `merged_presences`. Everything that reads READY - the friends list, the names of conversations, who is where in a
//! voice channel - was written against the objects. So they are put back first, here, once, and nothing else has to
//! know the shape changed. A READY without a `users` table is already whole and comes back unchanged.
use serde_json::{json, Map, Value};
use std::collections::HashMap;
/// READY with its users put back, or `None` when there is nothing to put back.
pub fn rehydrate(ready: &Value) -> Option<Value> {
let users: HashMap<String, Value> = ready["users"]
.as_array()?
.iter()
.filter_map(|u| Some((u["id"].as_str()?.to_string(), u.clone())))
.collect();
let mut out = ready.clone();
// The members the guilds were relieved of, back where they were: in the guild, by its place in the list.
if let Some(merged) = ready["merged_members"].as_array() {
if let Some(guilds) = out["guilds"].as_array_mut() {
for (guild, members) in guilds.iter_mut().zip(merged) {
if guild["members"].is_null() && members.is_array() {
guild["members"] = members.clone();
}
}
}
}
// And the presences: the friends' as READY's own `presences`, each guild's with the guild.
if let Some(merged) = ready["merged_presences"].as_object() {
if out["presences"].is_null() {
out["presences"] = merged.get("friends").cloned().unwrap_or_else(|| json!([]));
}
if let (Some(per_guild), Some(guilds)) = (merged.get("guilds").and_then(Value::as_array), out["guilds"].as_array_mut()) {
for (guild, presences) in guilds.iter_mut().zip(per_guild) {
if guild["presences"].is_null() && presences.is_array() {
guild["presences"] = presences.clone();
}
}
}
}
restore(&mut out, &users);
Some(out)
}
/// Gives every object that names a user by id the user itself, and every conversation that lists its recipients by id
/// the recipients, wherever in the payload they are.
fn restore(value: &mut Value, users: &HashMap<String, Value>) {
match value {
Value::Object(object) => {
fill(object, users);
for child in object.values_mut() {
restore(child, users);
}
}
Value::Array(list) => {
for child in list {
restore(child, users);
}
}
_ => {}
}
}
fn fill(object: &mut Map<String, Value>, users: &HashMap<String, Value>) {
if !object.contains_key("user") {
if let Some(user) = object.get("user_id").and_then(Value::as_str).and_then(|id| users.get(id)) {
object.insert("user".to_string(), user.clone());
}
}
if !object.contains_key("recipients") {
if let Some(ids) = object.get("recipient_ids").and_then(Value::as_array) {
let recipients: Vec<Value> = ids.iter().filter_map(|id| users.get(id.as_str()?).cloned()).collect();
object.insert("recipients".to_string(), Value::Array(recipients));
}
}
}
#[cfg(test)]
mod tests {
use super::*;
/// READY as the documentation describes it for a client with DEDUPE_USER_OBJECTS: users in a table, ids in their
/// place, members and presences merged.
fn dehydrated() -> Value {
json!({
"user": { "id": "1", "username": "me" },
"users": [
{ "id": "1", "username": "me" },
{ "id": "2", "username": "ada", "global_name": "Ada" },
{ "id": "3", "username": "bob" }
],
"relationships": [{ "id": "2", "type": 1, "user_id": "2" }, { "id": "3", "type": 3, "user_id": "3" }],
"private_channels": [{ "id": "20", "type": 1, "recipient_ids": ["2"], "last_message_id": null }, { "id": "21", "type": 3, "recipient_ids": ["2", "3"] }],
"merged_presences": { "friends": [{ "user_id": "2", "status": "online", "activities": [] }], "guilds": [[{ "user_id": "3", "status": "idle" }]] },
"guilds": [{ "id": "g", "voice_states": [{ "user_id": "3", "channel_id": "v" }] }],
"merged_members": [[{ "user_id": "1", "roles": ["r"], "pending": false }, { "user_id": "3", "roles": [] }]]
})
}
#[test]
fn a_dehydrated_ready_is_made_whole_again() {
let whole = rehydrate(&dehydrated()).expect("there was a users table");
// Friends and their requests, by the user object every reader looks for.
assert_eq!(whole["relationships"][0]["user"]["username"], "ada");
assert_eq!(whole["relationships"][1]["user"]["id"], "3");
// Conversations with the people in them.
assert_eq!(whole["private_channels"][0]["recipients"][0]["global_name"], "Ada");
assert_eq!(whole["private_channels"][1]["recipients"].as_array().unwrap().len(), 2);
// Presences where READY used to carry them.
assert_eq!(whole["presences"][0]["user"]["id"], "2");
assert_eq!(whole["presences"][0]["status"], "online");
assert_eq!(whole["guilds"][0]["presences"][0]["user"]["username"], "bob");
// The guild's members, and our own among them as `own_member` looks for it.
let members = whole["guilds"][0]["members"].as_array().unwrap();
assert_eq!(members.len(), 2);
assert_eq!(members[0]["user"]["id"], "1");
assert_eq!(members[0]["roles"][0], "r");
// Voice states, which name their user the same way.
assert_eq!(whole["guilds"][0]["voice_states"][0]["user"]["username"], "bob");
}
#[test]
fn a_ready_that_was_never_dehydrated_is_left_as_it_was() {
let whole = json!({ "user": { "id": "1" }, "relationships": [{ "id": "2", "user": { "id": "2" } }] });
assert!(rehydrate(&whole).is_none());
}
#[test]
fn what_is_already_there_is_not_overwritten_and_the_unknown_is_left_alone() {
let mut ready = dehydrated();
ready["relationships"][0]["user"] = json!({ "id": "2", "username": "already" });
ready["private_channels"][0]["recipient_ids"] = json!(["2", "999"]);
ready["guilds"][0]["members"] = json!([{ "user": { "id": "1" } }]);
let whole = rehydrate(&ready).unwrap();
assert_eq!(whole["relationships"][0]["user"]["username"], "already");
assert_eq!(whole["private_channels"][0]["recipients"].as_array().unwrap().len(), 1, "an id nobody told us about is not made up");
assert_eq!(whole["guilds"][0]["members"].as_array().unwrap().len(), 1, "a guild's own members are not replaced");
}
#[test]
fn the_whole_thing_can_be_done_twice() {
let once = rehydrate(&dehydrated()).unwrap();
let twice = rehydrate(&once).unwrap();
assert_eq!(once, twice);
}
}
+329
View File
@@ -0,0 +1,329 @@
//! What Discord's own rules say a message needs before it is worth sending - checked here, from what login and the
//! guild already told us, so a message that is bound to be turned down is not sent at all. A refused send is one more
//! request from an account that is being watched, and says nothing that a sentence cannot say sooner.
//!
//! The server rules are Discord's documented ones (and Concord's reading of them, `discord/verification.rs`): a
//! server's verification level asks for a verified email (low), an account older than five minutes (medium), a member
//! of ten minutes' standing (high) or a verified phone (very high); membership screening and onboarding must be done;
//! and a member who is timed out may not speak. The owner, anybody with every permission, anybody with a role (for
//! the levels) and anybody the server exempts (`BYPASSES_VERIFICATION`) are not held to the levels.
use std::time::{Duration, Instant};
/// Member flags (the `flags` of a guild member).
const FLAG_COMPLETED_ONBOARDING: u64 = 1 << 1;
const FLAG_BYPASSES_VERIFICATION: u64 = 1 << 2;
/// How long a member of a server wears Discord's green "new member" badge by their name: the first week.
pub const NEW_MEMBER_SECS: i64 = 7 * 24 * 60 * 60;
const ACCOUNT_AGE_SECS: i64 = 5 * 60;
const MEMBER_AGE_SECS: i64 = 10 * 60;
/// What a server asks of everyone, as its payload says.
#[derive(Debug, Clone, Default)]
pub struct GuildFacts {
/// 0 none, 1 low, 2 medium, 3 high, 4 very high.
pub verification: u8,
pub community: bool,
pub owner: bool,
}
/// This account's membership of it.
#[derive(Debug, Clone, Default)]
pub struct MemberFacts {
pub pending: Option<bool>,
pub flags: Option<u64>,
pub has_roles: bool,
/// When it joined, Unix seconds.
pub joined_at: Option<i64>,
/// Until when it may not speak (a timeout), Unix seconds.
pub timed_out_until: Option<i64>,
}
/// And the account itself.
#[derive(Debug, Clone, Default)]
pub struct AccountFacts {
pub email_verified: Option<bool>,
pub phone_verified: Option<bool>,
/// When it was made, Unix seconds (from its id).
pub created_at: i64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Restriction {
Screening,
Onboarding,
EmailRequired,
PhoneRequired,
AccountTooNew(i64),
MemberTooNew(i64),
TimedOut(i64),
/// Wearing the green new-member badge: for as many more seconds.
NewMember(i64),
}
impl Restriction {
/// The sentence a refused send gives.
pub fn sentence(&self) -> String {
match self {
Restriction::Screening => "This server asks you to accept its rules first. Do that from the notice above the conversation, or in Discord's own app.".into(),
Restriction::Onboarding => "This server asks you to finish its onboarding first - do that in Discord's own app.".into(),
Restriction::EmailRequired => "This server needs a verified email on your Discord account before you can talk here.".into(),
Restriction::PhoneRequired => "This server needs a verified phone number on your Discord account before you can talk here.".into(),
Restriction::AccountTooNew(s) => format!("This server asks for an account more than five minutes old: {}.", wait(*s)),
Restriction::MemberTooNew(s) => format!("This server asks you to have been a member for ten minutes before you talk: {}.", wait(*s)),
Restriction::TimedOut(s) => format!("You are timed out in this server: {}.", wait(*s)),
Restriction::NewMember(s) => format!(
"You joined this server recently, so Discord shows the green new-member badge by your name here, for about {} more. Messages from another client while it shows are treated as spam: use Discord's own app until it is gone.",
long_wait(*s)
),
}
}
/// Whether it is the kind that stays until the person does something, and so is worth a lock over the box. The
/// others pass by themselves, and are said when a send is tried. (Screening has a notice of its own.)
pub fn banner(&self) -> Option<&'static str> {
match self {
Restriction::Onboarding => Some("This server asks you to finish its onboarding first - do that in Discord's own app."),
Restriction::EmailRequired => Some("This server needs a verified email on your Discord account before you can talk here."),
Restriction::PhoneRequired => Some("This server needs a verified phone number on your Discord account before you can talk here."),
Restriction::NewMember(_) => Some("You are a new member of this server: Discord shows a green badge by your name for the first week, and treats messages from another client as spam while it shows. Use Discord's own app until it is gone; this opens again then."),
_ => None,
}
}
}
/// A wait counted in days or hours, for one that is long.
fn long_wait(seconds: i64) -> String {
let hours = (seconds.max(0) + 3599) / 3600;
if hours >= 48 {
format!("{} days", (hours + 23) / 24)
} else if hours > 1 {
format!("{hours} hours")
} else {
"an hour".to_string()
}
}
fn wait(seconds: i64) -> String {
if seconds >= 120 {
format!("wait {} more minutes", (seconds + 59) / 60)
} else {
format!("wait {} more seconds", seconds.max(1))
}
}
/// Whether the account may speak in a server now, and if not why. `full` is every permission (the owner, an
/// administrator); `onboarding_enabled` is what the server's onboarding says of itself, where it was asked.
pub fn participation(
guild: &GuildFacts,
member: Option<&MemberFacts>,
account: &AccountFacts,
full: bool,
onboarding_enabled: Option<bool>,
now: i64,
) -> Option<Restriction> {
if guild.owner {
return None;
}
// Not known is not blocked: a check that stops everything for want of data is a lock nobody can open.
let member = member?;
if let Some(until) = member.timed_out_until.filter(|u| *u > now) {
return Some(Restriction::TimedOut(until - now));
}
if member.pending == Some(true) {
return Some(Restriction::Screening);
}
if full {
return None;
}
let flags = member.flags.unwrap_or(0);
if flags & FLAG_BYPASSES_VERIFICATION != 0 {
return None;
}
if onboarding_enabled == Some(true) && flags & FLAG_COMPLETED_ONBOARDING == 0 {
return Some(Restriction::Onboarding);
}
if guild.verification == 0 || member.has_roles || account.phone_verified == Some(true) {
return None;
}
if guild.verification >= 4 {
return (account.phone_verified == Some(false)).then_some(Restriction::PhoneRequired);
}
if account.email_verified == Some(false) {
return Some(Restriction::EmailRequired);
}
if guild.verification >= 2 {
let age = now - account.created_at;
if account.created_at > 0 && age < ACCOUNT_AGE_SECS {
return Some(Restriction::AccountTooNew(ACCOUNT_AGE_SECS - age));
}
}
if guild.verification >= 3 {
if let Some(joined) = member.joined_at {
let age = now - joined;
if age < MEMBER_AGE_SECS {
return Some(Restriction::MemberTooNew(MEMBER_AGE_SECS - age));
}
}
}
None
}
/// Whether the account still wears the green new-member badge: it joined the server in the last week. Discord treats
/// what a person says from another client in that time as spam, so it is a reason not to speak from here, whatever
/// else the server allows. The owner has no such badge; `joined` is when the account joined if the member does not say.
pub fn new_member(guild: &GuildFacts, member: Option<&MemberFacts>, joined: Option<i64>, now: i64) -> Option<Restriction> {
if guild.owner {
return None;
}
let joined = member.and_then(|m| m.joined_at).or(joined)?;
let age = now - joined;
// A join time in the future is a clock that is wrong, and not a reason to lock anything.
(0..NEW_MEMBER_SECS).contains(&age).then_some(Restriction::NewMember(NEW_MEMBER_SECS - age))
}
/// How long until a channel's slowmode lets the account speak again: `seconds` is the channel's setting, `last` the
/// account's last message there, `bypass` the permissions that are not held to it (manage messages, manage channel).
pub fn slowmode_remaining(seconds: u64, last: Option<Instant>, bypass: bool, now: Instant) -> Option<Duration> {
if seconds == 0 || bypass {
return None;
}
let ends = last? + Duration::from_secs(seconds);
(ends > now).then(|| ends - now)
}
/// Own messages (not sent from here) a direct message needs before it counts as a conversation that is begun.
pub const ESTABLISHED_DM_MESSAGES: usize = 5;
#[cfg(test)]
mod tests {
use super::*;
fn guild(level: u8) -> GuildFacts {
GuildFacts { verification: level, community: false, owner: false }
}
fn account() -> AccountFacts {
AccountFacts { email_verified: Some(true), phone_verified: Some(false), created_at: 1_000_000 }
}
const NOW: i64 = 2_000_000;
#[test]
fn an_open_server_asks_nothing_of_a_member_in_good_standing() {
let member = MemberFacts { pending: Some(false), flags: Some(0), ..Default::default() };
assert_eq!(participation(&guild(0), Some(&member), &account(), false, None, NOW), None);
}
#[test]
fn each_level_asks_for_what_it_says() {
let member = MemberFacts { pending: Some(false), flags: Some(0), joined_at: Some(NOW - 60), ..Default::default() };
let mut a = account();
a.email_verified = Some(false);
assert_eq!(participation(&guild(1), Some(&member), &a, false, None, NOW), Some(Restriction::EmailRequired));
a.email_verified = Some(true);
assert_eq!(participation(&guild(1), Some(&member), &a, false, None, NOW), None);
// Medium: the account must be over five minutes old.
a.created_at = NOW - 100;
assert_eq!(participation(&guild(2), Some(&member), &a, false, None, NOW), Some(Restriction::AccountTooNew(200)));
a.created_at = 1_000_000;
assert_eq!(participation(&guild(2), Some(&member), &a, false, None, NOW), None);
// High: and the membership over ten minutes.
assert_eq!(participation(&guild(3), Some(&member), &a, false, None, NOW), Some(Restriction::MemberTooNew(540)));
let old = MemberFacts { joined_at: Some(NOW - 700), ..member.clone() };
assert_eq!(participation(&guild(3), Some(&old), &a, false, None, NOW), None);
// Very high: a phone, and nothing else stands in for it but a role or an exemption.
assert_eq!(participation(&guild(4), Some(&old), &a, false, None, NOW), Some(Restriction::PhoneRequired));
a.phone_verified = Some(true);
assert_eq!(participation(&guild(4), Some(&old), &a, false, None, NOW), None);
}
#[test]
fn the_owner_a_role_and_an_exemption_are_not_held_to_the_levels() {
let mut a = account();
a.phone_verified = Some(false);
let plain = MemberFacts { pending: Some(false), flags: Some(0), joined_at: Some(NOW - 5), ..Default::default() };
assert!(participation(&guild(4), Some(&plain), &a, false, None, NOW).is_some());
let owner = GuildFacts { owner: true, ..guild(4) };
assert_eq!(participation(&owner, Some(&plain), &a, false, None, NOW), None);
let roled = MemberFacts { has_roles: true, ..plain.clone() };
assert_eq!(participation(&guild(4), Some(&roled), &a, false, None, NOW), None);
let exempt = MemberFacts { flags: Some(1 << 2), ..plain.clone() };
assert_eq!(participation(&guild(4), Some(&exempt), &a, false, None, NOW), None);
assert_eq!(participation(&guild(4), Some(&plain), &a, true, None, NOW), None, "every permission");
}
#[test]
fn screening_onboarding_and_a_timeout_come_before_the_levels() {
let a = account();
let pending = MemberFacts { pending: Some(true), ..Default::default() };
assert_eq!(participation(&guild(0), Some(&pending), &a, false, None, NOW), Some(Restriction::Screening));
let member = MemberFacts { pending: Some(false), flags: Some(0), ..Default::default() };
assert_eq!(participation(&guild(0), Some(&member), &a, false, Some(true), NOW), Some(Restriction::Onboarding));
assert_eq!(participation(&guild(0), Some(&member), &a, false, Some(false), NOW), None);
let done = MemberFacts { flags: Some(1 << 1), ..member.clone() };
assert_eq!(participation(&guild(0), Some(&done), &a, false, Some(true), NOW), None);
let timed = MemberFacts { timed_out_until: Some(NOW + 300), ..member };
assert_eq!(participation(&guild(0), Some(&timed), &a, true, None, NOW), Some(Restriction::TimedOut(300)), "even with every permission");
let over = MemberFacts { timed_out_until: Some(NOW - 5), ..timed };
assert_eq!(participation(&guild(0), Some(&over), &a, false, None, NOW), None);
}
#[test]
fn what_is_not_known_is_not_a_block() {
assert_eq!(participation(&guild(4), None, &account(), false, None, NOW), None);
let unknown = AccountFacts { email_verified: None, phone_verified: None, created_at: 0 };
let member = MemberFacts { pending: Some(false), flags: Some(0), joined_at: Some(NOW - 5), ..Default::default() };
assert_eq!(participation(&guild(4), Some(&member), &unknown, false, None, NOW), None);
assert_eq!(participation(&guild(1), Some(&member), &unknown, false, None, NOW), None);
}
#[test]
fn a_member_of_under_a_week_wears_the_badge_until_the_week_is_up() {
let day = 24 * 60 * 60;
let member = |ago: i64| MemberFacts { joined_at: Some(NOW - ago), ..Default::default() };
assert_eq!(new_member(&guild(0), Some(&member(60)), None, NOW), Some(Restriction::NewMember(7 * day - 60)));
assert_eq!(new_member(&guild(0), Some(&member(6 * day)), None, NOW), Some(Restriction::NewMember(day)));
assert_eq!(new_member(&guild(0), Some(&member(7 * day)), None, NOW), None, "gone at a week");
assert_eq!(new_member(&guild(0), Some(&member(90 * day)), None, NOW), None);
}
#[test]
fn the_badge_is_not_worn_by_the_owner_or_by_a_member_nothing_is_known_of() {
let fresh = MemberFacts { joined_at: Some(NOW - 5), ..Default::default() };
assert_eq!(new_member(&GuildFacts { owner: true, ..guild(0) }, Some(&fresh), None, NOW), None);
assert_eq!(new_member(&guild(0), None, None, NOW), None, "not known is not blocked");
assert_eq!(new_member(&guild(0), Some(&MemberFacts::default()), None, NOW), None);
// The server's own join time stands in for a member object that does not carry one.
assert!(new_member(&guild(0), Some(&MemberFacts::default()), Some(NOW - 5), NOW).is_some());
assert_eq!(new_member(&guild(0), Some(&MemberFacts { joined_at: Some(NOW + 500), ..Default::default() }), None, NOW), None, "a join in the future is a wrong clock");
}
#[test]
fn the_badge_says_how_long_in_days_and_is_a_lock_over_the_box() {
let day = 24 * 60 * 60;
assert!(Restriction::NewMember(3 * day).sentence().contains("3 days"));
assert!(Restriction::NewMember(5 * 3600).sentence().contains("5 hours"));
assert!(Restriction::NewMember(600).sentence().contains("an hour"));
assert!(Restriction::NewMember(day).banner().is_some(), "it lasts days, so it is a lock and not a thing said when a send is tried");
}
#[test]
fn slowmode_lets_the_account_speak_when_it_is_over() {
let t = Instant::now();
let s = Duration::from_secs;
assert_eq!(slowmode_remaining(0, Some(t), false, t), None);
assert_eq!(slowmode_remaining(10, None, false, t), None, "nothing said yet");
assert_eq!(slowmode_remaining(10, Some(t), false, t + s(3)), Some(s(7)));
assert_eq!(slowmode_remaining(10, Some(t), false, t + s(10)), None);
assert_eq!(slowmode_remaining(10, Some(t), true, t + s(3)), None, "manage messages is not held to it");
}
#[test]
fn the_sentences_say_how_long_to_wait() {
assert!(Restriction::AccountTooNew(200).sentence().contains("4 more minutes"));
assert!(Restriction::MemberTooNew(30).sentence().contains("30 more seconds"));
assert!(Restriction::TimedOut(1).sentence().contains("1 more seconds") || Restriction::TimedOut(1).sentence().contains("wait"));
assert!(Restriction::Onboarding.banner().is_some() && Restriction::AccountTooNew(5).banner().is_none());
}
}
+1 -1
View File
@@ -103,7 +103,7 @@ pub async fn search_messages(
.get(url)
.query(&query)
.header("Authorization", &cfg.token)
.send()
.send_paced()
.await
.context("searching")?;
// Discord answers a search of a server it is still indexing with 202 and
+433 -44
View File
@@ -35,6 +35,33 @@ pub(super) fn put_reply(payload: &mut Value, reply_to_id: Option<&str>, ping_aut
}
}
/// Says where the message answered is, as Discord's clients do: the channel, and the server where there is one. The id
/// alone is enough for the API, and is not what its own clients send.
pub(super) fn put_reply_place(payload: &mut Value, channel_id: &str, guild_id: Option<&str>) {
if payload["message_reference"].is_object() {
payload["message_reference"]["channel_id"] = Value::from(channel_id);
if let Some(guild) = guild_id {
payload["message_reference"]["guild_id"] = Value::from(guild);
}
}
}
/// Said when Discord refuses a message that creates one: a captcha closes the place for a while (moho does not answer
/// those, and trying again is what escalates), a channel's slowmode says how long is left, anything else is as
/// `refusal_text` has it.
pub(super) fn refused(state: &AppState, buffer_id: &str, status: reqwest::StatusCode, text: &str) -> String {
let parsed: Value = serde_json::from_str(text).unwrap_or(Value::Null);
if parsed.get("captcha_key").is_some() {
super::send_guard::hold_captcha(state, buffer_id);
return "Discord asked for a captcha before this message. moho does not answer Discord's captchas - trying again is what gets an account blocked - so this place is closed to it for a while: say something here from Discord's own app.".to_string();
}
if parsed["code"].as_u64() == Some(20016) {
let seconds = parsed["retry_after"].as_f64().unwrap_or(0.0).ceil() as u64;
return format!("This channel has slowmode on: wait {} more seconds.", seconds.max(1));
}
refusal_text(status, text)
}
/// REST message send - Discord's gateway is receive-only from the client's
/// perspective for user accounts; sending is always a plain HTTP POST.
/// Turns the names somebody typed into the mentions Discord understands.
@@ -220,6 +247,8 @@ pub async fn forward_message(
.runtime
.get_discord_channel(to_buffer)
.ok_or_else(|| anyhow!("there is no Discord conversation to send it to"))?;
super::send_guard::ensure_may_speak(state, to_buffer).await?;
let _turn = super::send_guard::channel_turn(&target_channel).await;
let mut reference = json!({
// 1 is Discord's own number for a forward, as against 0 for a reply.
"type": 1,
@@ -235,15 +264,21 @@ pub async fn forward_message(
http_client_for(token)
.post(format!("{API_BASE}/channels/{target_channel}/messages"))
.header("Authorization", token)
.json(&json!({ "content": "", "message_reference": reference })),
.json(&{
let mut payload = json!({ "content": "", "message_reference": reference });
super::send_guard::stamp(&mut payload);
payload
}),
)
.await
.context("forwarding the message")?;
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
bail!("{}", discord_error_text(status, &text, "forwarding that message"));
bail!("{}", refused(state, to_buffer, status, &text));
}
super::send_guard::record_own_send(to_buffer);
super::send_guard::note_sent(resp).await;
Ok(())
}
@@ -252,10 +287,18 @@ pub async fn send_message(state: &AppState, buffer_id: &str, token: &str, body:
.runtime
.get_discord_channel(buffer_id)
.ok_or_else(|| anyhow!("no known Discord channel for this buffer"))?;
// Nothing is said anywhere the account has not spoken before: see send_guard.
super::send_guard::ensure_may_speak(state, buffer_id).await?;
let account_id = state.runtime.get_buffer(buffer_id).map(|b| b.account_id).unwrap_or_default();
let content = resolve_outgoing_mentions(body, mention_candidates(state, &account_id, buffer_id));
super::send_guard::check_text(&content, state.runtime.discord_premium(&account_id), false)?;
let mut payload = json!({ "content": content });
put_reply(&mut payload, reply_to_id, reply_ping);
put_reply_place(&mut payload, &channel_id, state.runtime.get_discord_guild(buffer_id).as_deref());
super::send_guard::stamp(&mut payload);
// One send at a time to a channel, and each a moment after the last.
let _turn = super::send_guard::channel_turn(&channel_id).await;
super::send_guard::take_turn(&account_id, false).await;
let resp = send_write(
http_client_for(token)
.post(format!("{API_BASE}/channels/{channel_id}/messages"))
@@ -267,8 +310,10 @@ pub async fn send_message(state: &AppState, buffer_id: &str, token: &str, body:
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
bail!("{}", refusal_text(status, &text));
bail!("{}", refused(state, buffer_id, status, &text));
}
super::send_guard::record_own_send(buffer_id);
super::send_guard::note_sent(resp).await;
Ok(())
}
@@ -371,6 +416,8 @@ pub async fn send_attachments(
.runtime
.get_discord_channel(buffer_id)
.ok_or_else(|| anyhow!("no known Discord channel for this buffer"))?;
super::send_guard::ensure_may_speak(state, buffer_id).await?;
let _turn = super::send_guard::channel_turn(&channel_id).await;
if let Some(p) = progress {
p.at(Phase::Preparing, 0, Meter::HOST);
}
@@ -385,44 +432,197 @@ pub async fn send_attachments(
total += len;
files.push((path.clone(), name, len));
}
// What Discord would refuse is refused here, before any of it is sent.
super::send_guard::ensure_fits(state, buffer_id, &files)?;
// A caption is a message like any other, so a name typed in one is a
// mention like any other.
let account_id = state.runtime.get_buffer(buffer_id).map(|b| b.account_id).unwrap_or_default();
let content = resolve_outgoing_mentions(body, mention_candidates(state, &account_id, buffer_id));
super::send_guard::check_text(&content, state.runtime.discord_premium(&account_id), true)?;
let mut payload = json!({ "content": content });
put_reply(&mut payload, reply_to_id, reply_ping);
put_reply_place(&mut payload, &channel_id, state.runtime.get_discord_guild(buffer_id).as_deref());
super::send_guard::stamp(&mut payload);
// A file does not follow straight on from a message.
super::send_guard::take_turn(&account_id, true).await;
if let Some(p) = progress {
p.sent(0, total, Meter::HOST);
}
let client = http_client_for(token);
let url = format!("{API_BASE}/channels/{channel_id}/messages");
let resp = send_write_rebuilt(|| {
// Built afresh for every try: a streamed body is used up by sending
// it, so a rate-limited request has to open its files again.
let meter = std::sync::Arc::new(Meter {
sent: std::sync::atomic::AtomicU64::new(0),
total,
last: std::sync::Mutex::new((std::time::Instant::now(), 0)),
progress: progress.cloned(),
gap: Meter::GAP,
});
let mut form = reqwest::multipart::Form::new().text("payload_json", payload.to_string());
for (i, (path, name, len)) in files.iter().enumerate() {
form = form.part(format!("files[{i}]"), counted_part(path, name, *len, meter.clone())?);
let meter = std::sync::Arc::new(Meter {
sent: std::sync::atomic::AtomicU64::new(0),
total,
last: std::sync::Mutex::new((std::time::Instant::now(), 0)),
progress: progress.cloned(),
gap: Meter::GAP,
});
let uploads: Vec<Upload> = files.iter().map(|(path, name, len)| Upload { path: path.clone(), name: name.clone(), len: *len, voice: None }).collect();
let resp = match upload_in_steps(state, token, &channel_id, &uploads, &mut payload, &meter).await? {
Some(resp) => resp,
// Where the way Discord's own apps send files is not there, the one bots use is.
None => {
let client = http_client_for(token);
let url = format!("{API_BASE}/channels/{channel_id}/messages");
for key in ["attachments", "type", "sticker_ids", "channel_id"] {
if let Some(object) = payload.as_object_mut() {
object.remove(key);
}
}
send_write_rebuilt(|| {
// Built afresh for every try: a streamed body is used up by sending it, so a rate-limited request has
// to open its files again.
let meter = std::sync::Arc::new(Meter {
sent: std::sync::atomic::AtomicU64::new(0),
total,
last: std::sync::Mutex::new((std::time::Instant::now(), 0)),
progress: progress.cloned(),
gap: Meter::GAP,
});
let mut form = reqwest::multipart::Form::new().text("payload_json", payload.to_string());
for (i, (path, name, len)) in files.iter().enumerate() {
form = form.part(format!("files[{i}]"), counted_part(path, name, *len, meter.clone())?);
}
Ok(client.post(&url).header("Authorization", token).multipart(form))
})
.await
.context("uploading Discord attachment")?
}
Ok(client.post(&url).header("Authorization", token).multipart(form))
})
.await
.context("uploading Discord attachment")?;
};
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
bail!("{}", refusal_text(status, &text));
bail!("{}", refused(state, buffer_id, status, &text));
}
super::send_guard::record_own_send(buffer_id);
super::send_guard::note_sent(resp).await;
Ok(())
}
/// A file on its way, and what it is when it is a voice message.
struct Upload {
path: String,
name: String,
len: u64,
/// A voice message's duration and waveform.
voice: Option<(f64, String)>,
}
/// What Discord answered to a request for somewhere to put the files: for each, by its place in the request, where to
/// send its bytes and the name to give the message. None when the answer is not that.
fn parse_slots(body: &Value, count: usize) -> Option<Vec<(String, String)>> {
let list = body["attachments"].as_array()?;
let mut slots: Vec<Option<(String, String)>> = vec![None; count];
for slot in list {
let id = slot["id"].as_u64().or_else(|| slot["id"].as_str().and_then(|s| s.parse().ok()))? as usize;
let url = slot["upload_url"].as_str().filter(|u| u.starts_with("https://"))?;
let name = slot["upload_filename"].as_str().filter(|n| !n.is_empty())?;
*slots.get_mut(id)? = Some((url.to_string(), name.to_string()));
}
slots.into_iter().collect()
}
/// The `attachments` a message names once its files are uploaded.
fn attachments_named(uploads: &[Upload], slots: &[(String, String)]) -> Value {
Value::Array(
uploads
.iter()
.zip(slots)
.enumerate()
.map(|(i, (upload, (_, uploaded)))| {
let mut entry = json!({ "id": i.to_string(), "filename": upload.name, "uploaded_filename": uploaded });
if let Some((duration, waveform)) = &upload.voice {
entry["duration_secs"] = json!(duration);
entry["waveform"] = json!(waveform);
}
entry
})
.collect(),
)
}
/// The body of an upload: the file, read from disk as it goes, counted by `meter`.
fn counted_body(path: &str, meter: std::sync::Arc<Meter>) -> Result<reqwest::Body> {
let file = std::fs::File::open(path).with_context(|| format!("reading {path}"))?;
let stream = tokio_util::io::ReaderStream::with_capacity(tokio::fs::File::from_std(file), 64 * 1024);
let counted = futures::StreamExt::map(stream, move |chunk| {
if let Ok(bytes) = &chunk {
meter.add(bytes.len());
}
chunk
});
Ok(reqwest::Body::wrap_stream(counted))
}
/// Sends files the way Discord's own apps do: ask where to put them (a file too big is refused here, before a byte
/// goes), put each there, then send the message with the files named - not the way bots do, with the bytes inside the
/// message. Returns the answer to the message, or `None` when Discord has no such way to offer (the endpoint is
/// missing, or answers something that is not slots) and the old one is to be used.
async fn upload_in_steps(
state: &AppState,
token: &str,
channel_id: &str,
uploads: &[Upload],
payload: &mut Value,
meter: &std::sync::Arc<Meter>,
) -> Result<Option<reqwest::Response>> {
let _ = state;
let request = json!({
"files": uploads.iter().enumerate().map(|(i, u)| json!({ "filename": u.name, "file_size": u.len, "id": i.to_string() })).collect::<Vec<_>>(),
});
let resp = send_write(
http_client_for(token).post(format!("{API_BASE}/channels/{channel_id}/attachments")).header("Authorization", token).json(&request),
)
.await
.context("asking Discord where to put the files")?;
let status = resp.status();
if matches!(status.as_u16(), 404 | 405 | 501) {
return Ok(None);
}
if status.as_u16() == 413 {
bail!("Discord says a file is too big to send here");
}
if !status.is_success() {
let text = resp.text().await.unwrap_or_default();
bail!("{}", discord_error_text(status, &text, "asking where to put the files"));
}
let body: Value = resp.json().await.context("reading where Discord said to put the files")?;
let Some(slots) = parse_slots(&body, uploads.len()) else { return Ok(None) };
// The bytes go to the address Discord gave, which is another host: none of Discord's headers go with them.
let storage = bare_client_for(token);
for (upload, (url, _)) in uploads.iter().zip(&slots) {
let put = storage
.put(url)
.header(reqwest::header::CONTENT_TYPE, "application/octet-stream")
.header(reqwest::header::CONTENT_LENGTH, upload.len)
.body(counted_body(&upload.path, meter.clone())?)
.send()
.await
.with_context(|| format!("uploading {}", upload.name))?;
if !put.status().is_success() {
bail!("Discord's storage refused {} ({})", upload.name, put.status());
}
}
payload["attachments"] = attachments_named(uploads, &slots);
payload["type"] = json!(0);
payload["sticker_ids"] = json!([]);
payload["channel_id"] = json!(channel_id);
// A message that names its files is not sent as a bare one.
if let Some(object) = payload.as_object_mut() {
object.remove("tts");
if object.get("flags") == Some(&json!(0)) {
object.remove("flags");
}
}
let sent = send_write(
http_client_for(token).post(format!("{API_BASE}/channels/{channel_id}/messages")).header("Authorization", token).json(&*payload),
)
.await
.context("sending the message with the files")?;
Ok(Some(sent))
}
/// Sending a recording as a voice message rather than as a sound file.
///
/// Three things make Discord treat it as one, and all three are required: the
@@ -452,6 +652,7 @@ pub async fn send_voice_message(
.runtime
.get_discord_channel(buffer_id)
.ok_or_else(|| anyhow!("no known Discord channel for this buffer"))?;
super::send_guard::ensure_may_speak(state, buffer_id).await?;
let bytes = tokio::fs::read(file_path)
.await
.with_context(|| format!("reading {}", file_path.display()))?;
@@ -459,7 +660,7 @@ pub async fn send_voice_message(
// than on this, but a file that arrives anywhere else should say what it
// is - and an unnamed part is rejected outright.
let file_name = "voice-message.ogg";
let payload = json!({
let mut payload = json!({
"content": "",
"flags": IS_VOICE_MESSAGE,
"attachments": [{
@@ -469,28 +670,55 @@ pub async fn send_voice_message(
"waveform": waveform,
}],
});
let form = reqwest::multipart::Form::new()
.text("payload_json", payload.to_string())
.part(
"files[0]",
reqwest::multipart::Part::bytes(bytes)
.file_name(file_name)
.mime_str("audio/ogg")
.context("audio/ogg is a valid mime type")?,
);
let resp = send_write(
http_client_for(token)
.post(format!("{API_BASE}/channels/{channel_id}/messages"))
.header("Authorization", token)
.multipart(form),
)
.await
.context("uploading the voice message")?;
super::send_guard::stamp(&mut payload);
let _turn = super::send_guard::channel_turn(&channel_id).await;
let account_id = state.runtime.get_buffer(buffer_id).map(|b| b.account_id).unwrap_or_default();
super::send_guard::take_turn(&account_id, true).await;
// Sent the way Discord's own apps send a recording, with the old way (the bytes inside the message) where that is
// not there.
let meter = std::sync::Arc::new(Meter {
sent: std::sync::atomic::AtomicU64::new(0),
total: bytes.len() as u64,
last: std::sync::Mutex::new((std::time::Instant::now(), 0)),
progress: None,
gap: Meter::GAP,
});
let upload = Upload { path: file_path.to_string_lossy().into_owned(), name: file_name.to_string(), len: bytes.len() as u64, voice: Some((duration_secs, waveform.to_string())) };
let resp = match upload_in_steps(state, token, &channel_id, std::slice::from_ref(&upload), &mut payload, &meter).await? {
Some(resp) => resp,
None => {
for key in ["type", "sticker_ids", "channel_id"] {
if let Some(object) = payload.as_object_mut() {
object.remove(key);
}
}
payload["attachments"] = json!([{ "id": "0", "filename": file_name, "duration_secs": duration_secs, "waveform": waveform }]);
let form = reqwest::multipart::Form::new()
.text("payload_json", payload.to_string())
.part(
"files[0]",
reqwest::multipart::Part::bytes(bytes)
.file_name(file_name)
.mime_str("audio/ogg")
.context("audio/ogg is a valid mime type")?,
);
send_write(
http_client_for(token)
.post(format!("{API_BASE}/channels/{channel_id}/messages"))
.header("Authorization", token)
.multipart(form),
)
.await
.context("uploading the voice message")?
}
};
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
bail!("{}", refusal_text(status, &text));
bail!("{}", refused(state, buffer_id, status, &text));
}
super::send_guard::record_own_send(buffer_id);
super::send_guard::note_sent(resp).await;
Ok(())
}
@@ -795,6 +1023,12 @@ pub async fn send_typing(state: &AppState, buffer_id: &str, token: &str) -> Resu
.runtime
.get_discord_channel(buffer_id)
.ok_or_else(|| anyhow!("no known Discord channel for this buffer"))?;
// The window repeats its notice every few seconds for every service, and
// a Discord indicator stands for ten, so most of those would only be
// noise on the wire: one a channel per `TYPING_EVERY` is plenty.
if !typing_due(&channel_id) {
return Ok(());
}
send_write(
http_client_for(token)
.post(format!("{API_BASE}/channels/{channel_id}/typing"))
@@ -806,6 +1040,24 @@ pub async fn send_typing(state: &AppState, buffer_id: &str, token: &str) -> Resu
Ok(())
}
/// Shortest gap between two typing notices to the same channel. Short of the
/// ten seconds an indicator lasts, so it never lapses mid-sentence.
const TYPING_EVERY: std::time::Duration = std::time::Duration::from_secs(8);
/// Whether a typing notice to this channel is due, noting it if so.
fn typing_due(channel_id: &str) -> bool {
static LAST: std::sync::Mutex<Option<HashMap<String, std::time::Instant>>> = std::sync::Mutex::new(None);
let mut last = LAST.lock().unwrap();
let last = last.get_or_insert_with(HashMap::new);
let now = std::time::Instant::now();
if last.get(channel_id).is_some_and(|at| now.duration_since(*at) < TYPING_EVERY) {
return false;
}
last.retain(|_, at| now.duration_since(*at) < TYPING_EVERY);
last.insert(channel_id.to_string(), now);
true
}
/// Tells Discord this conversation has been read up to its newest message.
///
/// Unread was purely local before this, so reading a channel here did not
@@ -823,16 +1075,27 @@ pub async fn ack_read(state: &AppState, buffer_id: &str, token: &str) -> Result<
.runtime
.get_discord_channel(buffer_id)
.ok_or_else(|| anyhow!("no known Discord channel for this buffer"))?;
let Some(msg_id) = state.store.newest_msg_id(buffer_id)? else { return Ok(()) };
let Some((msg_id, own)) = state.store.newest_msg(buffer_id)? else { return Ok(()) };
// Nothing to tell Discord about a conversation whose last word is the account's own: sending a message reads
// everything before it, on every device, and Discord's own clients do not ack what they sent.
if own {
return Ok(());
}
// Reading a channel again with nothing new in it says nothing new.
if !ack_due(&channel_id, &msg_id) {
return Ok(());
}
let resp = send_write(
http_client_for(token)
.post(format!("{API_BASE}/channels/{channel_id}/messages/{msg_id}/ack"))
.header("Authorization", token)
.json(&json!({ "token": serde_json::Value::Null }))
.json(&json!({ "token": serde_json::Value::Null, "last_viewed": last_viewed_day() }))
)
.await
.inspect_err(|_| forget_ack(&channel_id))
.context("acking Discord read state")?;
if !resp.status().is_success() {
forget_ack(&channel_id);
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
bail!("{}", refusal_text(status, &text));
@@ -840,6 +1103,55 @@ pub async fn ack_read(state: &AppState, buffer_id: &str, token: &str) -> Result<
Ok(())
}
/// The day, counted from Discord's epoch (2015-01-01) and rounded up, that a read mark says the channel was last
/// looked at: Discord's own clients send it with every ack.
pub(super) fn last_viewed_day() -> i64 {
last_viewed_day_at(chrono::Utc::now().timestamp())
}
pub(super) fn last_viewed_day_at(unix_secs: i64) -> i64 {
const DISCORD_EPOCH_SECS: i64 = 1_420_070_400;
(unix_secs - DISCORD_EPOCH_SECS + 86_399) / 86_400
}
/// The newest message each channel was last acked up to, this run.
static ACKED: std::sync::Mutex<Option<HashMap<String, String>>> = std::sync::Mutex::new(None);
/// Whether acking `msg_id` in this channel would say anything new, noting it
/// if so.
fn ack_due(channel_id: &str, msg_id: &str) -> bool {
let mut acked = ACKED.lock().unwrap();
let acked = acked.get_or_insert_with(HashMap::new);
if acked.get(channel_id).is_some_and(|id| id == msg_id) {
return false;
}
acked.insert(channel_id.to_string(), msg_id.to_string());
true
}
/// Another client of the account read this far: no need to say it again from here.
pub(super) fn acked_elsewhere(channel_id: &str, msg_id: &str) {
let mut acked = ACKED.lock().unwrap();
// Never backwards: an ack from a device that is behind does not make this one's later ack redundant.
let newer = match acked.get_or_insert_with(HashMap::new).get(channel_id) {
Some(known) => match (known.parse::<u64>(), msg_id.parse::<u64>()) {
(Ok(k), Ok(m)) => m > k,
_ => true,
},
None => true,
};
if newer {
acked.get_or_insert_with(HashMap::new).insert(channel_id.to_string(), msg_id.to_string());
}
}
/// A refused ack must be tried again, not remembered as done.
fn forget_ack(channel_id: &str) {
if let Some(acked) = ACKED.lock().unwrap().as_mut() {
acked.remove(channel_id);
}
}
pub async fn delete_message(state: &AppState, buffer_id: &str, token: &str, msg_id: &str) -> Result<()> {
let channel_id = state
.runtime
@@ -949,3 +1261,80 @@ mod refusal_tests {
assert_eq!(describe_seconds(45), "45 seconds");
}
}
#[cfg(test)]
mod ack_tests {
use super::*;
#[test]
fn an_ack_made_elsewhere_is_not_made_again_and_never_moves_backwards() {
// Own ids for this test: the table is shared by the process.
let channel = "ack-test-channel";
acked_elsewhere(channel, "200");
assert!(!ack_due(channel, "200"), "read this far on another device");
assert!(ack_due(channel, "300"), "something newer is still to be told");
acked_elsewhere(channel, "250");
assert!(!ack_due(channel, "300"), "an older ack from elsewhere does not undo what was told");
}
}
#[cfg(test)]
mod wire_tests {
use super::*;
#[test]
fn a_reply_says_where_the_message_it_answers_is() {
let mut payload = json!({ "content": "hi" });
put_reply(&mut payload, Some("42"), true);
put_reply_place(&mut payload, "7", Some("9"));
assert_eq!(payload["message_reference"], json!({ "message_id": "42", "channel_id": "7", "guild_id": "9" }));
let mut dm = json!({ "content": "hi" });
put_reply(&mut dm, Some("42"), true);
put_reply_place(&mut dm, "7", None);
assert_eq!(dm["message_reference"], json!({ "message_id": "42", "channel_id": "7" }));
// Not a reply: nothing is added.
let mut plain = json!({ "content": "hi" });
put_reply_place(&mut plain, "7", Some("9"));
assert!(plain.get("message_reference").is_none());
}
#[test]
fn a_read_mark_counts_days_from_discords_epoch_rounded_up() {
assert_eq!(last_viewed_day_at(1_420_070_400), 0);
assert_eq!(last_viewed_day_at(1_420_070_401), 1);
// 2026-10-22 00:00 UTC is 4,191 days after 2015-01-01.
assert_eq!(last_viewed_day_at(1_420_070_400 + 4_191 * 86_400), 4_191);
}
#[test]
fn where_to_put_the_files_is_read_from_the_answer() {
let body = json!({ "attachments": [
{ "id": 1, "upload_url": "https://storage.example/b?sig=2", "upload_filename": "u/b.png" },
{ "id": "0", "upload_url": "https://storage.example/a?sig=1", "upload_filename": "u/a.png" }
] });
let slots = parse_slots(&body, 2).unwrap();
assert_eq!(slots[0], ("https://storage.example/a?sig=1".to_string(), "u/a.png".to_string()));
assert_eq!(slots[1].1, "u/b.png");
// Anything that is not a slot for every file is not an answer to use.
assert!(parse_slots(&body, 3).is_none(), "one missing");
assert!(parse_slots(&json!({ "attachments": [{ "id": 0, "upload_url": "http://insecure/x", "upload_filename": "n" }] }), 1).is_none(), "plain http");
assert!(parse_slots(&json!({ "message": "no" }), 1).is_none());
assert!(parse_slots(&json!({ "attachments": [{ "id": 5, "upload_url": "https://x/y", "upload_filename": "n" }] }), 1).is_none(), "an id out of range");
}
#[test]
fn the_message_names_each_file_and_a_voice_message_says_how_long_it_is() {
let uploads = vec![
Upload { path: "/a".into(), name: "a.png".into(), len: 3, voice: None },
Upload { path: "/v".into(), name: "voice-message.ogg".into(), len: 9, voice: Some((2.5, "AAAA".into())) },
];
let slots = vec![("https://s/1".to_string(), "u/a.png".to_string()), ("https://s/2".to_string(), "u/v.ogg".to_string())];
let named = attachments_named(&uploads, &slots);
assert_eq!(named[0], json!({ "id": "0", "filename": "a.png", "uploaded_filename": "u/a.png" }));
assert_eq!(named[1]["id"], "1");
assert_eq!(named[1]["uploaded_filename"], "u/v.ogg");
assert_eq!(named[1]["duration_secs"], json!(2.5));
assert_eq!(named[1]["waveform"], "AAAA");
}
}
File diff suppressed because it is too large. Load diff
+304
View File
@@ -0,0 +1,304 @@
//! A gateway session kept on disk, so a restart can pick it up instead of logging in afresh.
//!
//! Discord keeps a dropped session for a few minutes and replays what it missed to a client that resumes it; its own
//! client resumes only when the last heartbeat was acknowledged under three minutes ago, and logs in afresh otherwise.
//! A restart of moho is a dropped connection like any other, so it does the same - for which the session has to
//! survive the process: what to resume (`Position`), and what a resume does *not* send again, because Discord sends
//! READY and GUILD_CREATE only to a fresh login (`State`). The state is replayed through the very code that handles
//! them live, so a resumed start ends up with the same friends, servers, mutes and plan as a fresh one.
//!
//! Two files per account, because the position changes every heartbeat and the state hardly ever.
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::path::PathBuf;
/// How old the last acknowledged heartbeat may be for a session to be worth resuming. Discord's client uses three
/// minutes; a little under, since connecting takes time and a refused resume costs a fresh login.
pub const FRESH_FOR: i64 = 150;
/// Past this the state is not kept (and so nothing is resumed): a few huge servers are not worth a file that size
/// rewritten on every change.
const MAX_STATE_BYTES: usize = 24 * 1024 * 1024;
#[derive(Serialize, Deserialize, Clone)]
pub struct Position {
pub user_id: String,
/// Which token this session belongs to, as a hash - the token itself is not written here.
pub token: String,
pub session_id: String,
pub url: String,
pub seq: u64,
/// When a heartbeat was last acknowledged, Unix seconds.
pub acked_at: i64,
}
#[derive(Serialize, Deserialize, Default)]
pub struct State {
pub session_id: String,
/// READY's payload, with the direct messages found over REST added as `moho_dms`.
pub ready: Value,
/// Each server's GUILD_CREATE payload, by id.
pub guilds: HashMap<String, Value>,
}
/// One account's session files, and what is known of them in memory.
#[derive(Default)]
pub struct SessionFile {
/// Where the files go; the daemon's data directory unless a test says otherwise.
root: Option<PathBuf>,
account: String,
user_id: String,
token: String,
position: Option<Position>,
state: Option<State>,
state_dirty: bool,
}
pub fn token_hash(token: &str) -> String {
Sha256::digest(token.as_bytes()).iter().take(12).map(|b| format!("{b:02x}")).collect()
}
fn default_dir() -> PathBuf {
crate::default_data_dir().join("discord-sessions")
}
/// Whether a saved position can still be resumed at `now`, by this token.
pub fn resumable(position: &Position, user_id: &str, token: &str, now: i64) -> bool {
position.user_id == user_id && position.token == token_hash(token) && now - position.acked_at < FRESH_FOR && position.acked_at <= now + 5
}
impl SessionFile {
fn dir(&self) -> PathBuf {
self.root.clone().unwrap_or_else(default_dir)
}
fn path(&self, what: &str) -> PathBuf {
let safe: String = self.account.chars().map(|c| if c.is_ascii_alphanumeric() || c == '-' { c } else { '_' }).collect();
self.dir().join(format!("{safe}.{what}.json"))
}
pub fn new(account: &str, user_id: &str, token: &str) -> Self {
SessionFile { account: account.to_string(), user_id: user_id.to_string(), token: token_hash(token), ..Default::default() }
}
/// The saved session as READY and GUILD_CREATE dispatches in the order they were first received, with the
/// position to resume from; `None` when there is nothing fresh enough to resume.
pub fn load(&mut self, token: &str, now: i64) -> Option<(Position, Vec<Value>)> {
let position: Position = serde_json::from_str(&std::fs::read_to_string(self.path("pos")).ok()?).ok()?;
if !resumable(&position, &self.user_id, token, now) {
return None;
}
let state: State = serde_json::from_str(&std::fs::read_to_string(self.path("state")).ok()?).ok()?;
if state.session_id != position.session_id || state.ready.is_null() {
return None;
}
let mut replay = vec![dispatch("READY", position.seq, state.ready.clone())];
replay.extend(state.guilds.values().map(|g| dispatch("GUILD_CREATE", position.seq, g.clone())));
self.position = Some(position.clone());
self.state = Some(state);
Some((position, replay))
}
/// A fresh READY: a new session, with nothing of the last one worth keeping.
pub fn ready(&mut self, session_id: &str, url: &str, seq: u64, ready: Value) {
self.position = Some(Position {
user_id: self.user_id.clone(),
token: self.token.clone(),
session_id: session_id.to_string(),
url: url.to_string(),
seq,
acked_at: chrono::Utc::now().timestamp(),
});
self.state = Some(State { session_id: session_id.to_string(), ready, guilds: HashMap::new() });
self.state_dirty = true;
}
/// The direct messages listed over REST at READY, kept with it: a resume does not list them again.
pub fn dms(&mut self, list: Vec<Value>) {
if let Some(state) = self.state.as_mut() {
state.ready["moho_dms"] = Value::Array(list);
self.state_dirty = true;
}
}
pub fn guild(&mut self, id: &str, payload: &Value) {
if let Some(state) = self.state.as_mut() {
state.guilds.insert(id.to_string(), payload.clone());
self.state_dirty = true;
}
}
pub fn guild_gone(&mut self, id: &str) {
if let Some(state) = self.state.as_mut() {
if state.guilds.remove(id).is_some() {
self.state_dirty = true;
}
}
}
/// A heartbeat was acknowledged: the session is alive as of now, and has seen events up to `seq`.
pub fn acked(&mut self, seq: u64) {
let Some(position) = self.position.as_mut() else { return };
position.seq = seq;
position.acked_at = chrono::Utc::now().timestamp();
self.write();
}
/// The session is gone (refused, or ended by Discord): nothing is left to resume.
pub fn clear(&mut self) {
self.position = None;
self.state = None;
self.state_dirty = false;
let _ = std::fs::remove_file(self.path("pos"));
let _ = std::fs::remove_file(self.path("state"));
}
fn write(&mut self) {
let (Some(position), Some(state)) = (self.position.as_ref(), self.state.as_ref()) else { return };
let _ = std::fs::create_dir_all(self.dir());
if self.state_dirty {
match serde_json::to_string(state) {
Ok(text) if text.len() <= MAX_STATE_BYTES => {
if write_private(&self.path("state"), &text) {
self.state_dirty = false;
}
}
_ => {
// Too big to keep: this session cannot be resumed after a restart, which is the cost of that
// many servers.
let _ = std::fs::remove_file(self.path("pos"));
self.position = None;
self.state = None;
return;
}
}
}
if let Ok(text) = serde_json::to_string(position) {
write_private(&self.path("pos"), &text);
}
}
}
/// Removes whatever is kept for an account that has been taken away.
pub fn forget(account: &str) {
let mut file = SessionFile { account: account.to_string(), ..Default::default() };
file.clear();
}
fn dispatch(kind: &str, seq: u64, d: Value) -> Value {
serde_json::json!({ "op": 0, "t": kind, "s": seq, "d": d, "moho_replay": true })
}
/// Written to a temporary file and renamed, so a restart in the middle of it never reads half a file; readable by
/// this user alone, as the session is as good as a token for as long as it lives.
fn write_private(path: &std::path::Path, text: &str) -> bool {
let tmp = path.with_extension("tmp");
let ok = (|| -> std::io::Result<()> {
#[cfg(unix)]
{
use std::io::Write;
use std::os::unix::fs::OpenOptionsExt;
let mut f = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(&tmp)?;
f.write_all(text.as_bytes())?;
}
#[cfg(not(unix))]
std::fs::write(&tmp, text)?;
std::fs::rename(&tmp, path)
})();
ok.is_ok()
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
/// A directory of its own, gone when the test is.
struct Scratch(PathBuf);
impl Scratch {
fn new(name: &str) -> Self {
let dir = std::env::temp_dir().join(format!("nobilis-session-store-{name}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
Scratch(dir)
}
fn path(&self) -> &std::path::Path {
&self.0
}
}
impl Drop for Scratch {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.0);
}
}
fn in_dir(dir: &std::path::Path) -> SessionFile {
SessionFile { root: Some(dir.to_path_buf()), ..SessionFile::new("discord:7", "7", "tok") }
}
#[test]
fn a_saved_session_comes_back_as_the_dispatches_it_was_made_of() {
let dir = Scratch::new("comes_back");
let mut first = in_dir(dir.path());
first.ready("s", "wss://x", 3, json!({ "session_id": "s", "guilds": [] }));
first.dms(vec![json!({ "id": "9" })]);
first.guild("g1", &json!({ "id": "g1" }));
first.acked(42);
let (position, replay) = in_dir(dir.path()).load("tok", chrono::Utc::now().timestamp()).expect("fresh enough");
assert_eq!((position.session_id.as_str(), position.seq), ("s", 42));
// READY first, then the servers, each marked as saved and numbered where the session had got to.
assert_eq!(replay[0]["t"], "READY");
assert_eq!(replay[0]["d"]["moho_dms"][0]["id"], "9");
assert_eq!(replay[1]["t"], "GUILD_CREATE");
assert!(replay.iter().all(|m| m["s"] == 42 && m["moho_replay"] == true));
}
#[test]
fn a_session_cleared_or_of_another_token_is_not_resumed() {
let dir = Scratch::new("cleared");
let mut file = in_dir(dir.path());
file.ready("s", "wss://x", 1, json!({}));
file.acked(2);
let now = chrono::Utc::now().timestamp();
assert!(in_dir(dir.path()).load("other", now).is_none());
file.clear();
assert!(in_dir(dir.path()).load("tok", now).is_none());
}
#[test]
fn a_session_that_was_never_acknowledged_is_not_saved() {
let dir = Scratch::new("never_acked");
let mut file = in_dir(dir.path());
file.ready("s", "wss://x", 1, json!({}));
assert!(in_dir(dir.path()).load("tok", chrono::Utc::now().timestamp()).is_none());
}
fn pos(acked_at: i64) -> Position {
Position { user_id: "7".into(), token: token_hash("tok"), session_id: "s".into(), url: "wss://x".into(), seq: 3, acked_at }
}
#[test]
fn a_session_is_resumed_only_while_its_last_ack_is_fresh() {
assert!(resumable(&pos(1000), "7", "tok", 1000 + FRESH_FOR - 1));
assert!(!resumable(&pos(1000), "7", "tok", 1000 + FRESH_FOR));
assert!(!resumable(&pos(1000), "7", "tok", 1000 + 3600));
}
#[test]
fn a_session_belongs_to_one_account_and_one_token() {
assert!(!resumable(&pos(1000), "8", "tok", 1010));
assert!(!resumable(&pos(1000), "7", "other", 1010));
// A clock set back is not a fresh session.
assert!(!resumable(&pos(5000), "7", "tok", 1000));
}
#[test]
fn the_token_itself_is_not_what_is_written() {
assert!(!token_hash("tok").contains("tok"));
assert_eq!(token_hash("tok"), token_hash("tok"));
assert_ne!(token_hash("tok"), token_hash("tok2"));
}
}
+2 -2
View File
@@ -95,7 +95,7 @@ async fn default_sounds(token: &str) -> Vec<Sound> {
let resp = http_client_for(token)
.get(format!("{API_BASE}/soundboard-default-sounds"))
.header("Authorization", token)
.send()
.send_paced()
.await
.ok()?;
resp.status().is_success().then_some(())?;
@@ -229,7 +229,7 @@ async fn fetch(token: &str, id: &str) -> Result<Vec<u8>> {
}
let resp = http_client_for(token)
.get(format!("https://cdn.discordapp.com/soundboard-sounds/{id}"))
.send()
.send_paced()
.await
.context("fetching the sound")?;
if !resp.status().is_success() {
+12 -4
View File
@@ -74,7 +74,7 @@ pub async fn lottie_file(token: &str, id: &str) -> Result<String> {
if !tokio::fs::try_exists(&path).await.unwrap_or(false) {
let resp = http_client_for(token)
.get(lottie_url(id))
.send()
.send_paced()
.await
.context("fetching the sticker")?;
if !resp.status().is_success() {
@@ -157,7 +157,7 @@ async fn standard_packs(token: &str) -> Vec<(String, Vec<Sticker>)> {
let resp = http_client_for(token)
.get(format!("{API_BASE}/sticker-packs"))
.header("Authorization", token)
.send()
.send_paced()
.await
.ok()?;
if !resp.status().is_success() {
@@ -241,19 +241,27 @@ fn guild_list(state: &AppState, account_id: &str, buffer_id: &str) -> Vec<Value>
/// Sends a sticker as a message of its own.
pub async fn send(state: &AppState, buffer_id: &str, token: &str, sticker_id: &str) -> Result<()> {
let channel_id = state.runtime.get_discord_channel(buffer_id).ok_or_else(|| anyhow!("no known Discord channel for this buffer"))?;
super::send_guard::ensure_may_speak(state, buffer_id).await?;
let _turn = super::send_guard::channel_turn(&channel_id).await;
let resp = send_write(
http_client_for(token)
.post(format!("{API_BASE}/channels/{channel_id}/messages"))
.header("Authorization", token)
.json(&json!({ "sticker_ids": [sticker_id] })),
.json(&{
let mut payload = json!({ "sticker_ids": [sticker_id] });
super::send_guard::stamp(&mut payload);
payload
}),
)
.await
.context("sending a Discord sticker")?;
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
bail!("Discord API error {status}: {text}");
bail!("{}", super::send::refused(state, buffer_id, status, &text));
}
super::send_guard::record_own_send(buffer_id);
super::send_guard::note_sent(resp).await;
Ok(())
}
+125 -11
View File
@@ -3,7 +3,13 @@
//! Telegram keeps a bot's list of commands (`/help`, `/weblink`, ...) with the bot, and its own apps show it as a
//! Menu button beside the message box and as suggestions while typing a slash. It is read once from the bot's profile
//! and kept for a while, since it changes when the bot's author changes it and not otherwise.
//!
//! In a group it is the commands of every bot that is in the group, which Telegram lists with the group itself (and
//! scopes: a bot can offer other commands to a group's administrators than to its members). Two bots can both have a
//! `/help`, and a group says which one is meant by writing the bot's name after it, `/help@somebot`: a command that
//! more than one bot has is offered once for each bot, in that form.
use super::peers;
use super::send::{explain, peer_for};
use crate::model::Buffer;
use crate::state::AppState;
@@ -20,6 +26,8 @@ const KEEP: Duration = Duration::from_secs(15 * 60);
pub struct Command {
pub name: String,
pub description: String,
/// Whose it is, where that is not the conversation itself: a bot in a group, as `@name`.
pub source: Option<String>,
}
/// When each conversation's list was read, and what it said.
@@ -38,31 +46,90 @@ fn from_info(info: &tl::enums::BotInfo) -> Vec<Command> {
.flatten()
.map(|c| {
let tl::enums::BotCommand::Command(c) = c;
Command { name: c.command.trim_start_matches('/').to_string(), description: c.description.clone() }
Command { name: c.command.trim_start_matches('/').to_string(), description: c.description.clone(), source: None }
})
.filter(|c| !c.name.is_empty())
.collect()
}
/// The commands of the bot a conversation is with; nothing for a conversation that is not with a bot.
/// The commands of a bot's conversation, or of the bots in a group; nothing for any other conversation.
pub async fn commands(state: &AppState, buffer: &Buffer) -> Result<Vec<Command>, String> {
if buffer.kind != "dm" || buffer.name_icon.as_deref() != Some("smart_toy") {
let with_bot = buffer.kind == "dm" && buffer.name_icon.as_deref() == Some("smart_toy");
// A channel's posts are not commands, and a person is not a bot. A group (or a topic of one) has bots.
let in_group = buffer.kind != "dm" && buffer.category.as_deref() != Some(peers::CHANNELS);
if !with_bot && !in_group {
return Ok(Vec::new());
}
if let Some((when, list)) = kept().lock().unwrap_or_else(|e| e.into_inner()).get(&buffer.id) {
// A group's topics are one group: they share a list.
let key = format!("{}|{}", buffer.account_id, buffer.remote_id.as_deref().map(|r| r.split(':').next().unwrap_or(r)).unwrap_or(&buffer.id));
if let Some((when, list)) = kept().lock().unwrap_or_else(|e| e.into_inner()).get(&key) {
if when.elapsed() < KEEP {
return Ok(list.clone());
}
}
let (handle, peer) = peer_for(state, buffer).await?;
let answer = handle.client.invoke(&tl::functions::users::GetFullUser { id: peer.into() }).await.map_err(|e| explain(&e))?;
let tl::enums::users::UserFull::Full(full) = answer;
let tl::enums::UserFull::Full(user) = full.full_user;
let list = user.bot_info.as_ref().map(from_info).unwrap_or_default();
kept().lock().unwrap_or_else(|e| e.into_inner()).insert(buffer.id.clone(), (Instant::now(), list.clone()));
let list = if with_bot {
let answer = handle.client.invoke(&tl::functions::users::GetFullUser { id: peer.into() }).await.map_err(|e| explain(&e))?;
let tl::enums::users::UserFull::Full(full) = answer;
let tl::enums::UserFull::Full(user) = full.full_user;
user.bot_info.as_ref().map(from_info).unwrap_or_default()
} else {
let (infos, users) = if peer.id.kind() == grammers_session::types::PeerKind::Channel {
let answer = handle.client.invoke(&tl::functions::channels::GetFullChannel { channel: peer.into() }).await.map_err(|e| explain(&e))?;
let tl::enums::messages::ChatFull::Full(full) = answer;
let tl::enums::ChatFull::ChannelFull(group) = full.full_chat else { return Ok(Vec::new()) };
(group.bot_info, full.users)
} else {
let answer = handle.client.invoke(&tl::functions::messages::GetFullChat { chat_id: peer.id.bare_id_unchecked() }).await.map_err(|e| explain(&e))?;
let tl::enums::messages::ChatFull::Full(full) = answer;
let tl::enums::ChatFull::Full(group) = full.full_chat else { return Ok(Vec::new()) };
(group.bot_info.unwrap_or_default(), full.users)
};
merge(&infos, &users)
};
kept().lock().unwrap_or_else(|e| e.into_inner()).insert(key, (Instant::now(), list.clone()));
Ok(list)
}
/// The @name a user is reached by.
fn username_of(user: &tl::enums::User) -> Option<String> {
let tl::enums::User::User(user) = user else { return None };
user.username.clone().or_else(|| user.usernames.as_ref()?.iter().find_map(|u| {
let tl::enums::Username::Username(u) = u;
u.active.then(|| u.username.clone())
}))
}
/// The commands of the bots in a group, each named for its bot when more than one bot has it.
fn merge(infos: &[tl::enums::BotInfo], users: &[tl::enums::User]) -> Vec<Command> {
let mut per_bot: Vec<(String, Vec<Command>)> = Vec::new();
for info in infos {
let tl::enums::BotInfo::Info(raw) = info;
let Some(id) = raw.user_id else { continue };
let Some(name) = users.iter().find(|u| matches!(u, tl::enums::User::User(u) if u.id == id)).and_then(username_of) else { continue };
per_bot.push((name, from_info(info)));
}
label(&per_bot)
}
/// Each bot's commands as the group's menu, a name that two bots share written with the bot's after it.
fn label(per_bot: &[(String, Vec<Command>)]) -> Vec<Command> {
let mut taken: HashMap<&str, usize> = HashMap::new();
for (_, list) in per_bot {
for c in list {
*taken.entry(c.name.as_str()).or_default() += 1;
}
}
let mut out = Vec::new();
for (bot, list) in per_bot {
for c in list {
let name = if taken.get(c.name.as_str()).copied().unwrap_or(0) > 1 { format!("{}@{bot}", c.name) } else { c.name.clone() };
out.push(Command { name, description: c.description.clone(), source: Some(format!("@{bot}")) });
}
}
out
}
#[cfg(test)]
mod tests {
use super::*;
@@ -88,9 +155,56 @@ mod tests {
assert_eq!(
from_info(&info),
vec![
Command { name: "help".into(), description: "Show help for bot commands".into() },
Command { name: "weblink".into(), description: "Get available link to ZLibrary".into() },
Command { name: "help".into(), description: "Show help for bot commands".into(), source: None },
Command { name: "weblink".into(), description: "Get available link to ZLibrary".into(), source: None },
]
);
}
fn cmd(name: &str, description: &str) -> Command {
Command { name: name.into(), description: description.into(), source: None }
}
#[test]
fn a_group_offers_each_bots_commands_and_names_the_bot_only_where_two_share_one() {
let per_bot = vec![
("rose_bot".to_string(), vec![cmd("help", "Rose help"), cmd("warn", "Warn a user")]),
("music_bot".to_string(), vec![cmd("help", "Music help"), cmd("play", "Play a song")]),
];
let got: Vec<(String, Option<String>)> = label(&per_bot).into_iter().map(|c| (c.name, c.source)).collect();
assert_eq!(
got,
vec![
("help@rose_bot".to_string(), Some("@rose_bot".to_string())),
("warn".to_string(), Some("@rose_bot".to_string())),
("help@music_bot".to_string(), Some("@music_bot".to_string())),
("play".to_string(), Some("@music_bot".to_string())),
]
);
}
#[test]
fn a_lone_bots_commands_are_not_named_for_it() {
let got = label(&[("only_bot".to_string(), vec![cmd("start", "Begin")])]);
assert_eq!(got.iter().map(|c| c.name.as_str()).collect::<Vec<_>>(), vec!["start"]);
assert!(label(&[]).is_empty());
}
#[test]
fn a_bot_the_group_does_not_name_cannot_be_written_to_and_is_left_out() {
let info = tl::enums::BotInfo::Info(tl::types::BotInfo {
has_preview_medias: false,
user_id: Some(3),
description: None,
description_photo: None,
description_document: None,
commands: Some(vec![tl::enums::BotCommand::Command(tl::types::BotCommand { ephemeral: false, command: "ghost".into(), description: "Nobody".into() })]),
menu_button: None,
privacy_policy_url: None,
app_settings: None,
verifier_settings: None,
});
let users = vec![tl::enums::User::Empty(tl::types::UserEmpty { id: 3 })];
assert!(merge(&[info], &users).is_empty());
}
}
+4
View File
@@ -164,6 +164,10 @@ async fn stay(state: &AppState, config: &TelegramAccountConfig, account_id: &str
if let Err(e) = super::dialogs::sync(&state, &account_id, &handle).await {
tracing::warn!("telegram[{account_id}]: reading the conversations: {e:#}");
}
// After the conversations: what a folder holds by its rules is worked out from them.
if let Err(e) = super::folders::sync(&state, &account_id, &handle).await {
tracing::warn!("telegram[{account_id}]: reading the folders: {e:#}");
}
}
});
+23
View File
@@ -15,12 +15,35 @@ use grammers_client::tl;
/// How many conversations are read on connect. Telegram pages them a hundred at a time.
const LIMIT: usize = 200;
/// What the account's folders look at in a conversation, noted for working out what each folder holds.
fn note_for_folders(account_id: &str, dialog: &grammers_client::peer::Dialog) {
use super::folders::{Facts, Kind};
use grammers_client::peer::Peer;
let (kind, contact) = match dialog.peer() {
Peer::User(user) if user.is_bot() => (Kind::Bot, false),
Peer::User(user) => (Kind::Person, user.contact()),
Peer::Group(_) | Peer::Community(_) => (Kind::Group, false),
Peer::Channel(_) => (Kind::Broadcast, false),
};
let tl::enums::Dialog::Dialog(d) = &dialog.raw else { return };
let facts = Facts {
kind,
contact,
muted: super::notify::silent(&d.notify_settings, chrono::Utc::now().timestamp()),
unread: d.unread_count > 0 || d.unread_mark,
// Telegram keeps its archive as the folder numbered 1.
archived: d.folder_id == Some(1),
};
super::folders::note(account_id, peers::remote_id(dialog.peer().id()), facts);
}
/// Makes a buffer of each of the account's conversations, newest activity first, and records the last message of each.
pub async fn sync(state: &AppState, account_id: &str, handle: &Handle) -> Result<usize> {
let mut dialogs = handle.client.iter_dialogs();
let mut count = 0;
while count < LIMIT {
let Some(dialog) = dialogs.next().await.map_err(|e| anyhow::anyhow!("{e}"))? else { break };
note_for_folders(account_id, &dialog);
// A group with Topics is a conversation for each topic, found by asking the group.
if super::topics::is_forum(dialog.peer()) {
peers::rail_for(state, account_id, dialog.peer(), count as i64);
+734
View File
@@ -0,0 +1,734 @@
//! Chat folders: the tabs Telegram lets an account sort its chats into, kept in step with moho's categories.
//!
//! A folder is a name and a rule for what is in it: chats named one by one, and kinds of chat (contacts, groups,
//! channels, bots) with some taken back out (muted, read, archived, or named one by one). The window draws each folder
//! as a category on the account's page, and a change on either side is made on the other:
//!
//! - a folder made, renamed, deleted or moved in order on Telegram turns up as the same change to the category
//! (`telegramFolders`, said whenever the account's folders are read: on connect and on every
//! `updateDialogFilter*` Telegram sends);
//! - a category made, renamed, deleted or moved in order here is made on Telegram, and a conversation filed under a
//! category goes into that folder.
//!
//! A chat is in one category here and can be in several folders there. It is drawn under the first folder (in
//! Telegram's order) that holds it, and filing it somewhere else takes it out of that one and puts it in the new one;
//! the folders it is in besides are left as they are. Taking it out of a folder it is in by a rule - a chat in
//! "Groups" because it is a group - is done the way Telegram's own apps do it, by adding it to the folder's
//! exceptions.
//!
//! What Telegram does not allow is also not allowed here, and said in its words: a folder needs something in it, so
//! a category is made on Telegram when its first conversation is filed into it, and the last one cannot be taken
//! out. Groups are not conversations on the account's page (each is an entry on the rail), so they are not shown
//! under a category, but a folder that holds them still does, and an edit made here never touches them.
//!
//! Which chats a rule holds is worked out from what is known of each chat when the conversations are read - whether
//! it is a contact, muted, unread or archived - so a folder defined by "unread" is as current as that read.
use super::send::{explain, peer_for};
use super::{peers, Handle};
use crate::model::Buffer;
use crate::state::AppState;
use anyhow::{anyhow, Result};
use grammers_client::tl;
use grammers_session::types::PeerId;
use serde_json::{json, Value};
use std::collections::{HashMap, HashSet};
use std::sync::Mutex;
/// The first number a folder made by hand may have: 0 is the list of every chat.
const FIRST_ID: i32 = 2;
/// What kind of chat a conversation is, as a folder's rules tell them apart.
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum Kind {
Person,
Bot,
Group,
Broadcast,
}
/// What a folder's rules look at in a chat.
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct Facts {
pub kind: Kind,
pub contact: bool,
pub muted: bool,
pub unread: bool,
pub archived: bool,
}
/// A folder's rules, read from Telegram's description of it.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct Rules {
contacts: bool,
non_contacts: bool,
groups: bool,
broadcasts: bool,
bots: bool,
exclude_muted: bool,
exclude_read: bool,
exclude_archived: bool,
/// Chats named one by one, pinned ones among them, as the ids this client calls them by.
include: Vec<String>,
exclude: Vec<String>,
}
impl Rules {
/// Whether a chat is in by the kinds of chat the folder takes (and not taken back out), leaving the chats named
/// one by one aside.
fn by_kind(&self, facts: &Facts) -> bool {
let wanted = match facts.kind {
Kind::Person => (self.contacts && facts.contact) || (self.non_contacts && !facts.contact),
Kind::Bot => self.bots,
Kind::Group => self.groups,
Kind::Broadcast => self.broadcasts,
};
wanted && !(self.exclude_muted && facts.muted) && !(self.exclude_read && !facts.unread) && !(self.exclude_archived && facts.archived)
}
/// Whether the folder holds a chat.
pub fn holds(&self, remote: &str, facts: Option<&Facts>) -> bool {
if self.include.iter().any(|r| r == remote) {
return true;
}
if self.exclude.iter().any(|r| r == remote) {
return false;
}
facts.is_some_and(|f| self.by_kind(f))
}
}
/// A folder as Telegram holds it, with what is needed to change it.
#[derive(Clone, Debug)]
pub struct Folder {
pub id: i32,
pub title: String,
pub emoticon: Option<String>,
pub rules: Rules,
/// Shared folders (a link anyone can join by) have no rules and no exceptions.
pub shared: bool,
raw: tl::enums::DialogFilter,
}
/// The chat an input peer names, as this client calls it; `None` for none.
pub fn remote_of_input(peer: &tl::enums::InputPeer, own: i64) -> Option<String> {
use tl::enums::InputPeer as P;
let id = match peer {
P::PeerSelf => PeerId::user_unchecked(own),
P::User(u) => PeerId::user_unchecked(u.user_id),
P::UserFromMessage(u) => PeerId::user_unchecked(u.user_id),
P::Chat(c) => PeerId::chat_unchecked(c.chat_id),
P::Channel(c) => PeerId::channel_unchecked(c.channel_id),
P::ChannelFromMessage(c) => PeerId::channel_unchecked(c.channel_id),
P::Empty => return None,
};
Some(peers::remote_id(id))
}
fn text_of(title: &tl::enums::TextWithEntities) -> String {
let tl::enums::TextWithEntities::Entities(t) = title;
t.text.clone()
}
fn remotes(list: &[tl::enums::InputPeer], own: i64) -> Vec<String> {
list.iter().filter_map(|p| remote_of_input(p, own)).collect()
}
/// A folder read from Telegram's description of it; `None` for the entry that stands for every chat.
pub fn folder_of(raw: &tl::enums::DialogFilter, own: i64) -> Option<Folder> {
match raw {
tl::enums::DialogFilter::Default => None,
tl::enums::DialogFilter::Filter(f) => {
let mut include = remotes(&f.include_peers, own);
for pinned in remotes(&f.pinned_peers, own) {
if !include.contains(&pinned) {
include.push(pinned);
}
}
Some(Folder {
id: f.id,
title: text_of(&f.title),
emoticon: f.emoticon.clone(),
rules: Rules {
contacts: f.contacts,
non_contacts: f.non_contacts,
groups: f.groups,
broadcasts: f.broadcasts,
bots: f.bots,
exclude_muted: f.exclude_muted,
exclude_read: f.exclude_read,
exclude_archived: f.exclude_archived,
include,
exclude: remotes(&f.exclude_peers, own),
},
shared: false,
raw: raw.clone(),
})
}
tl::enums::DialogFilter::Chatlist(f) => {
let mut include = remotes(&f.include_peers, own);
for pinned in remotes(&f.pinned_peers, own) {
if !include.contains(&pinned) {
include.push(pinned);
}
}
Some(Folder { id: f.id, title: text_of(&f.title), emoticon: f.emoticon.clone(), rules: Rules { include, ..Default::default() }, shared: true, raw: raw.clone() })
}
}
}
/// The conversations a folder holds, by the ids this client calls them by: every one it names, and every one of the
/// chats whose facts are known that its rules take.
pub fn members(folder: &Folder, facts: &HashMap<String, Facts>) -> Vec<String> {
let mut out: Vec<String> = folder.rules.include.clone();
let mut seen: HashSet<&str> = folder.rules.include.iter().map(String::as_str).collect();
let mut found: Vec<&String> = facts.keys().collect();
found.sort();
for remote in found {
if !seen.contains(remote.as_str()) && folder.rules.holds(remote, facts.get(remote)) {
seen.insert(remote);
out.push(remote.clone());
}
}
out
}
/// The first folder, in the order given, that holds a chat: the one it is drawn under.
pub fn home_of(folders: &[Folder], facts: &HashMap<String, Facts>, remote: &str) -> Option<i32> {
folders.iter().find(|f| f.rules.holds(remote, facts.get(remote))).map(|f| f.id)
}
/// The folder with a chat put in it.
pub fn with_peer(raw: &tl::enums::DialogFilter, peer: &tl::enums::InputPeer, own: i64) -> Option<tl::enums::DialogFilter> {
let remote = remote_of_input(peer, own)?;
let mut edited = raw.clone();
match &mut edited {
tl::enums::DialogFilter::Default => return None,
tl::enums::DialogFilter::Filter(f) => {
f.exclude_peers.retain(|p| remote_of_input(p, own).as_deref() != Some(remote.as_str()));
if !f.include_peers.iter().chain(&f.pinned_peers).any(|p| remote_of_input(p, own).as_deref() == Some(remote.as_str())) {
f.include_peers.push(peer.clone());
}
}
tl::enums::DialogFilter::Chatlist(f) => {
if !f.include_peers.iter().chain(&f.pinned_peers).any(|p| remote_of_input(p, own).as_deref() == Some(remote.as_str())) {
f.include_peers.push(peer.clone());
}
}
}
Some(edited)
}
/// The folder with a chat taken out: no longer named, and - if its rules would still take it - made an exception.
pub fn without_peer(raw: &tl::enums::DialogFilter, peer: &tl::enums::InputPeer, own: i64, facts: Option<&Facts>) -> Option<tl::enums::DialogFilter> {
let remote = remote_of_input(peer, own)?;
let named = |p: &tl::enums::InputPeer| remote_of_input(p, own).as_deref() == Some(remote.as_str());
let mut edited = raw.clone();
match &mut edited {
tl::enums::DialogFilter::Default => return None,
tl::enums::DialogFilter::Filter(f) => {
f.include_peers.retain(|p| !named(p));
f.pinned_peers.retain(|p| !named(p));
let kept = folder_of(&tl::enums::DialogFilter::Filter(f.clone()), own)?;
if facts.is_some_and(|facts| kept.rules.by_kind(facts)) && !f.exclude_peers.iter().any(named) {
f.exclude_peers.push(peer.clone());
}
}
tl::enums::DialogFilter::Chatlist(f) => {
f.include_peers.retain(|p| !named(p));
f.pinned_peers.retain(|p| !named(p));
}
}
Some(edited)
}
/// A folder with another name.
pub fn renamed(raw: &tl::enums::DialogFilter, name: &str) -> Option<tl::enums::DialogFilter> {
let title = tl::enums::TextWithEntities::Entities(tl::types::TextWithEntities { text: name.to_string(), entities: Vec::new() });
let mut edited = raw.clone();
match &mut edited {
tl::enums::DialogFilter::Default => return None,
tl::enums::DialogFilter::Filter(f) => f.title = title,
tl::enums::DialogFilter::Chatlist(f) => f.title = title,
}
Some(edited)
}
/// A folder that holds one chat and nothing else: how a category becomes a folder.
pub fn starting_with(id: i32, name: &str, peer: &tl::enums::InputPeer) -> tl::enums::DialogFilter {
tl::enums::DialogFilter::Filter(tl::types::DialogFilter {
contacts: false,
non_contacts: false,
groups: false,
broadcasts: false,
bots: false,
exclude_muted: false,
exclude_read: false,
exclude_archived: false,
title_noanimate: false,
id,
title: tl::enums::TextWithEntities::Entities(tl::types::TextWithEntities { text: name.to_string(), entities: Vec::new() }),
emoticon: None,
color: None,
pinned_peers: Vec::new(),
include_peers: vec![peer.clone()],
exclude_peers: Vec::new(),
})
}
/// The number for a folder made now: the next one after those in use.
pub fn next_id(folders: &[tl::enums::DialogFilter]) -> i32 {
folders
.iter()
.map(|f| match f {
tl::enums::DialogFilter::Default => 0,
tl::enums::DialogFilter::Filter(f) => f.id,
tl::enums::DialogFilter::Chatlist(f) => f.id,
})
.max()
.map_or(FIRST_ID, |m| (m + 1).max(FIRST_ID))
}
fn id_of(raw: &tl::enums::DialogFilter) -> i32 {
match raw {
tl::enums::DialogFilter::Default => 0,
tl::enums::DialogFilter::Filter(f) => f.id,
tl::enums::DialogFilter::Chatlist(f) => f.id,
}
}
// What is known of each account's chats and folders, kept for the window to ask for and for the edits below.
type FactsByAccount = Mutex<HashMap<String, HashMap<String, Facts>>>;
fn all_facts() -> &'static FactsByAccount {
static F: std::sync::OnceLock<FactsByAccount> = std::sync::OnceLock::new();
F.get_or_init(Default::default)
}
type Held = Mutex<HashMap<String, Vec<Value>>>;
fn held() -> &'static Held {
static H: std::sync::OnceLock<Held> = std::sync::OnceLock::new();
H.get_or_init(Default::default)
}
/// Notes what a folder's rules look at in a chat, as the conversations are read.
pub fn note(account_id: &str, remote: String, facts: Facts) {
all_facts().lock().unwrap_or_else(|e| e.into_inner()).entry(account_id.to_string()).or_default().insert(remote, facts);
}
fn facts_of(account_id: &str) -> HashMap<String, Facts> {
all_facts().lock().unwrap_or_else(|e| e.into_inner()).get(account_id).cloned().unwrap_or_default()
}
/// Forgets an account's folders when it goes.
pub fn forget(account_id: &str) {
all_facts().lock().unwrap_or_else(|e| e.into_inner()).remove(account_id);
held().lock().unwrap_or_else(|e| e.into_inner()).remove(account_id);
}
/// The folders as the window is told of them: in Telegram's order, each with the conversations it holds.
pub fn summary(folders: &[Folder], facts: &HashMap<String, Facts>) -> Vec<Value> {
folders
.iter()
.map(|f| json!({ "id": f.id, "name": f.title, "emoticon": f.emoticon, "members": members(f, facts), "shared": f.shared }))
.collect()
}
/// What the window last heard of an account's folders.
pub fn current(account_id: &str) -> Value {
Value::Array(held().lock().unwrap_or_else(|e| e.into_inner()).get(account_id).cloned().unwrap_or_default())
}
async fn read(handle: &Handle) -> Result<Vec<tl::enums::DialogFilter>> {
let answer = handle.client.invoke(&tl::functions::messages::GetDialogFilters {}).await.map_err(|e| anyhow!("{}", explain(&e)))?;
let tl::enums::messages::DialogFilters::Filters(found) = answer;
Ok(found.filters)
}
/// Reads the account's folders and tells the window, if they are not what it was last told.
pub async fn sync(state: &AppState, account_id: &str, handle: &Handle) -> Result<Vec<Value>> {
let raws = read(handle).await?;
let folders: Vec<Folder> = raws.iter().filter_map(|r| folder_of(r, handle.own.user_id)).collect();
let list = summary(&folders, &facts_of(account_id));
let changed = {
let mut all = held().lock().unwrap_or_else(|e| e.into_inner());
let before = all.insert(account_id.to_string(), list.clone());
before.as_ref() != Some(&list)
};
if changed {
state.events.emit("telegramFolders", json!({ "accountId": account_id, "folders": list }));
}
Ok(list)
}
/// Telegram says the account's folders changed, on this client or another: read them again.
pub fn changed(state: &AppState, account_id: &str, handle: &Handle) {
let (state, account_id, handle) = (state.clone(), account_id.to_string(), handle.clone());
tokio::spawn(async move {
if let Err(e) = sync(&state, &account_id, &handle).await {
tracing::debug!("telegram[{account_id}]: reading the folders: {e:#}");
}
});
}
fn handle_for(state: &AppState, account_id: &str) -> Result<Handle, String> {
state.runtime.telegram_handle(account_id).ok_or_else(|| "that Telegram account is not connected".to_string())
}
async fn save(handle: &Handle, id: i32, filter: Option<tl::enums::DialogFilter>) -> Result<(), String> {
handle.client.invoke(&tl::functions::messages::UpdateDialogFilter { id, filter }).await.map_err(|e| folder_words(&explain(&e)))?;
Ok(())
}
/// Telegram's refusals about folders, said so a person can act on them.
fn folder_words(raw: &str) -> String {
// Telegram's codes arrive as FILTER_INCLUDE_EMPTY or, once worded for a person, as "filter include empty".
let upper = raw.to_uppercase().replace(' ', "_");
for (code, words) in [
("FILTER_INCLUDE_EMPTY", "Telegram does not keep an empty folder: it needs at least one chat in it. Delete the folder instead"),
("FILTER_TITLE_EMPTY", "A folder needs a name"),
("DIALOG_FILTERS_TOO_MUCH", "This account already has as many folders as Telegram allows"),
("FILTER_NOT_SUPPORTED", "Telegram does not allow that change to this folder"),
("CHATLISTS_TOO_MUCH", "Telegram will not share any more folders from this account"),
("PEERS_LIST_EMPTY", "Telegram does not keep an empty folder: it needs at least one chat in it. Delete the folder instead"),
] {
if upper.contains(code) {
return words.to_string();
}
}
raw.to_string()
}
/// A folder by number, in the order Telegram has them.
fn find(folders: &[Folder], id: i32) -> Result<&Folder, String> {
folders.iter().find(|f| f.id == id).ok_or_else(|| "Telegram has no such folder any more".to_string())
}
/// Makes a folder from a conversation: how a category made here comes to exist on Telegram.
pub async fn create(state: &AppState, buffer: &Buffer, name: &str) -> Result<(), String> {
let name = name.trim();
if name.is_empty() {
return Err("A folder needs a name".to_string());
}
let (handle, peer) = peer_for(state, buffer).await?;
let raws = read(&handle).await.map_err(|e| e.to_string())?;
let filter = starting_with(next_id(&raws), name, &peer.into());
save(&handle, id_of(&filter), Some(filter)).await?;
sync(state, &buffer.account_id, &handle).await.map_err(|e| e.to_string())?;
Ok(())
}
/// Gives a folder another name.
pub async fn rename(state: &AppState, account_id: &str, id: i32, name: &str) -> Result<(), String> {
let name = name.trim();
if name.is_empty() {
return Err("A folder needs a name".to_string());
}
let handle = handle_for(state, account_id)?;
let raws = read(&handle).await.map_err(|e| e.to_string())?;
let raw = raws.iter().find(|r| id_of(r) == id && id != 0).ok_or("Telegram has no such folder any more")?;
save(&handle, id, renamed(raw, name)).await?;
sync(state, account_id, &handle).await.map_err(|e| e.to_string())?;
Ok(())
}
/// Deletes a folder. The chats in it stay where they are; only the tab goes.
pub async fn delete(state: &AppState, account_id: &str, id: i32) -> Result<(), String> {
let handle = handle_for(state, account_id)?;
if id == 0 {
return Err("The list of every chat cannot be deleted".to_string());
}
save(&handle, id, None).await?;
sync(state, account_id, &handle).await.map_err(|e| e.to_string())?;
Ok(())
}
/// Puts the account's folders in the order given. Folders not named keep their places after those that are.
pub async fn reorder(state: &AppState, account_id: &str, order: &[i32]) -> Result<(), String> {
let handle = handle_for(state, account_id)?;
let raws = read(&handle).await.map_err(|e| e.to_string())?;
let all: Vec<i32> = raws.iter().map(id_of).collect();
let wanted = ordered(&all, order);
if wanted == all {
return Ok(());
}
handle.client.invoke(&tl::functions::messages::UpdateDialogFiltersOrder { order: wanted }).await.map_err(|e| folder_words(&explain(&e)))?;
sync(state, account_id, &handle).await.map_err(|e| e.to_string())?;
Ok(())
}
/// The folders' numbers in the order asked for. Those named take the places of those named, in the order named; any
/// not named - the entry for every chat, which is not drawn and so cannot have been moved - stay where they were.
pub fn ordered(all: &[i32], asked: &[i32]) -> Vec<i32> {
let mut named: Vec<i32> = Vec::new();
for id in asked {
if all.contains(id) && !named.contains(id) {
named.push(*id);
}
}
let mut fill = named.clone().into_iter();
let mut out = all.to_vec();
for slot in out.iter_mut().filter(|id| named.contains(id)) {
*slot = fill.next().unwrap_or(*slot);
}
out
}
/// Files a conversation under a folder (`to`), or under none: out of the one it is drawn under and into the new one.
pub async fn file(state: &AppState, buffer: &Buffer, to: Option<i32>) -> Result<(), String> {
let (handle, peer) = peer_for(state, buffer).await?;
let own = handle.own.user_id;
let input: tl::enums::InputPeer = peer.into();
let remote = remote_of_input(&input, own).ok_or("that is not a chat Telegram can file")?;
let raws = read(&handle).await.map_err(|e| e.to_string())?;
let folders: Vec<Folder> = raws.iter().filter_map(|r| folder_of(r, own)).collect();
let facts = facts_of(&buffer.account_id);
let from = home_of(&folders, &facts, &remote);
if from == to {
return Ok(());
}
// Put in first: if Telegram refuses to take it out of the old one, it is at least where it was asked to go.
if let Some(id) = to {
let folder = find(&folders, id)?;
let edited = with_peer(&folder.raw, &input, own).ok_or("Telegram has no such folder any more")?;
save(&handle, id, Some(edited)).await?;
}
if let Some(id) = from {
let folder = find(&folders, id)?;
let edited = without_peer(&folder.raw, &input, own, facts.get(&remote)).ok_or("Telegram has no such folder any more")?;
if let Err(e) = save(&handle, id, Some(edited)).await {
let _ = sync(state, &buffer.account_id, &handle).await;
return Err(format!("It is in the new folder, but could not be taken out of \"{}\": {e}", folder.title));
}
}
sync(state, &buffer.account_id, &handle).await.map_err(|e| e.to_string())?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
const OWN: i64 = 1;
fn user(id: i64) -> tl::enums::InputPeer {
tl::enums::InputPeer::User(tl::types::InputPeerUser { user_id: id, access_hash: 7 })
}
fn channel(id: i64) -> tl::enums::InputPeer {
tl::enums::InputPeer::Channel(tl::types::InputPeerChannel { channel_id: id, access_hash: 7 })
}
fn filter(id: i32, name: &str, edit: impl FnOnce(&mut tl::types::DialogFilter)) -> tl::enums::DialogFilter {
let tl::enums::DialogFilter::Filter(mut f) = starting_with(id, name, &user(100)) else { unreachable!() };
f.include_peers.clear();
edit(&mut f);
tl::enums::DialogFilter::Filter(f)
}
/// A channel's id as this client writes it.
fn ch(id: i64) -> String {
peers::remote_id(PeerId::channel_unchecked(id))
}
fn facts(kind: Kind) -> Facts {
Facts { kind, contact: false, muted: false, unread: true, archived: false }
}
#[test]
fn a_chat_is_called_by_the_id_this_client_gives_it() {
assert_eq!(remote_of_input(&user(100), OWN).as_deref(), Some("100"));
assert_eq!(remote_of_input(&channel(555), OWN), Some(ch(555)));
assert!(ch(555).starts_with("-100") && ch(555).ends_with("555"));
assert_eq!(remote_of_input(&tl::enums::InputPeer::Chat(tl::types::InputPeerChat { chat_id: 9 }), OWN).as_deref(), Some("-9"));
assert_eq!(remote_of_input(&tl::enums::InputPeer::PeerSelf, OWN).as_deref(), Some("1"), "saved messages are the account's own chat");
assert_eq!(remote_of_input(&tl::enums::InputPeer::Empty, OWN), None);
}
#[test]
fn a_folder_holds_the_chats_it_names_and_the_kinds_it_takes_less_the_exceptions() {
let raw = filter(2, "Work", |f| {
f.bots = true;
f.groups = true;
f.exclude_muted = true;
f.include_peers = vec![user(100)];
f.pinned_peers = vec![user(101)];
f.exclude_peers = vec![user(200)];
});
let folder = folder_of(&raw, OWN).unwrap();
assert_eq!((folder.id, folder.title.as_str(), folder.shared), (2, "Work", false));
let mut known = HashMap::new();
known.insert("100".to_string(), facts(Kind::Person));
known.insert("200".to_string(), facts(Kind::Bot));
known.insert("201".to_string(), facts(Kind::Bot));
known.insert(ch(300), facts(Kind::Group));
known.insert(ch(301), Facts { muted: true, ..facts(Kind::Group) });
known.insert(ch(302), facts(Kind::Broadcast));
known.insert("102".to_string(), facts(Kind::Person));
// Named ones first (pinned among them, though Telegram keeps them apart), then the kinds, in a steady order.
assert_eq!(members(&folder, &known), vec!["100".to_string(), "101".to_string(), ch(300), "201".to_string()]);
}
#[test]
fn contacts_and_everyone_else_are_told_apart() {
let raw = filter(2, "People", |f| f.contacts = true);
let folder = folder_of(&raw, OWN).unwrap();
let friend = Facts { contact: true, ..facts(Kind::Person) };
assert!(folder.rules.holds("5", Some(&friend)));
assert!(!folder.rules.holds("6", Some(&facts(Kind::Person))));
assert!(!folder.rules.holds("7", None), "a chat nothing is known of is held only if it is named");
let others = folder_of(&filter(3, "Strangers", |f| f.non_contacts = true), OWN).unwrap();
assert!(others.rules.holds("6", Some(&facts(Kind::Person))));
assert!(!others.rules.holds("5", Some(&friend)));
}
#[test]
fn read_and_archived_chats_can_be_left_out_of_a_folder() {
let raw = filter(2, "Unread", |f| {
f.groups = true;
f.exclude_read = true;
f.exclude_archived = true;
});
let folder = folder_of(&raw, OWN).unwrap();
assert!(folder.rules.holds("-1", Some(&facts(Kind::Group))));
assert!(!folder.rules.holds("-1", Some(&Facts { unread: false, ..facts(Kind::Group) })));
assert!(!folder.rules.holds("-1", Some(&Facts { archived: true, ..facts(Kind::Group) })));
}
#[test]
fn the_entry_for_every_chat_is_not_a_folder() {
assert!(folder_of(&tl::enums::DialogFilter::Default, OWN).is_none());
}
#[test]
fn a_chat_is_drawn_under_the_first_folder_that_holds_it() {
let a = folder_of(&filter(2, "Groups", |f| f.groups = true), OWN).unwrap();
let b = folder_of(&filter(3, "Mine", |f| f.include_peers = vec![channel(300)]), OWN).unwrap();
let mut known = HashMap::new();
known.insert(ch(300), facts(Kind::Group));
known.insert("5".to_string(), facts(Kind::Person));
assert_eq!(home_of(&[a.clone(), b.clone()], &known, &ch(300)), Some(2));
assert_eq!(home_of(&[b, a], &known, &ch(300)), Some(3));
assert_eq!(home_of(&[], &known, "5"), None);
}
#[test]
fn putting_a_chat_in_names_it_once_and_lifts_an_exception() {
let raw = filter(2, "Work", |f| f.exclude_peers = vec![user(100)]);
let put = with_peer(&raw, &user(100), OWN).unwrap();
let tl::enums::DialogFilter::Filter(f) = &put else { unreachable!() };
assert_eq!(f.include_peers.len(), 1);
assert!(f.exclude_peers.is_empty());
let again = with_peer(&put, &user(100), OWN).unwrap();
let tl::enums::DialogFilter::Filter(f) = &again else { unreachable!() };
assert_eq!(f.include_peers.len(), 1, "a chat already named is not named twice");
}
#[test]
fn taking_a_chat_out_of_a_named_list_is_taking_it_off_the_list() {
let raw = filter(2, "Work", |f| {
f.include_peers = vec![user(100), user(101)];
f.pinned_peers = vec![user(100)];
});
let out = without_peer(&raw, &user(100), OWN, Some(&facts(Kind::Person))).unwrap();
let tl::enums::DialogFilter::Filter(f) = &out else { unreachable!() };
assert_eq!(remotes(&f.include_peers, OWN), vec!["101"]);
assert!(f.pinned_peers.is_empty());
assert!(f.exclude_peers.is_empty(), "nothing else held it, so there is no exception to make");
}
#[test]
fn taking_a_chat_out_of_a_folder_that_holds_it_by_a_rule_makes_it_an_exception() {
let raw = filter(2, "Groups", |f| f.groups = true);
let out = without_peer(&raw, &channel(300), OWN, Some(&facts(Kind::Group))).unwrap();
let tl::enums::DialogFilter::Filter(f) = &out else { unreachable!() };
assert_eq!(remotes(&f.exclude_peers, OWN), vec![ch(300)]);
let folder = folder_of(&out, OWN).unwrap();
assert!(!folder.rules.holds(&ch(300), Some(&facts(Kind::Group))));
assert!(folder.rules.holds(&ch(301), Some(&facts(Kind::Group))), "the others are still held");
// And not twice.
let again = without_peer(&out, &channel(300), OWN, Some(&facts(Kind::Group))).unwrap();
let tl::enums::DialogFilter::Filter(f) = &again else { unreachable!() };
assert_eq!(f.exclude_peers.len(), 1);
}
#[test]
fn a_shared_folder_has_no_exceptions_to_make() {
let raw = tl::enums::DialogFilter::Chatlist(tl::types::DialogFilterChatlist {
has_my_invites: false,
title_noanimate: false,
id: 4,
title: tl::enums::TextWithEntities::Entities(tl::types::TextWithEntities { text: "Shared".into(), entities: vec![] }),
emoticon: None,
color: None,
pinned_peers: vec![],
include_peers: vec![user(100), user(101)],
});
let folder = folder_of(&raw, OWN).unwrap();
assert!(folder.shared);
let out = without_peer(&raw, &user(100), OWN, Some(&facts(Kind::Person))).unwrap();
assert_eq!(folder_of(&out, OWN).unwrap().rules.include, vec!["101"]);
assert_eq!(folder_of(&renamed(&out, "Team").unwrap(), OWN).unwrap().title, "Team");
}
#[test]
fn renaming_keeps_everything_but_the_name() {
let raw = filter(2, "Work", |f| {
f.bots = true;
f.include_peers = vec![user(100)];
f.emoticon = Some("💼".into());
});
let renamed = renamed(&raw, "Job").unwrap();
let (before, after) = (folder_of(&raw, OWN).unwrap(), folder_of(&renamed, OWN).unwrap());
assert_eq!(after.title, "Job");
assert_eq!((before.rules, before.emoticon, before.id), (after.rules, after.emoticon, after.id));
assert!(super::renamed(&tl::enums::DialogFilter::Default, "x").is_none());
}
#[test]
fn a_new_folder_has_the_next_number_after_those_in_use_and_never_below_two() {
assert_eq!(next_id(&[]), 2);
assert_eq!(next_id(&[tl::enums::DialogFilter::Default]), 2);
assert_eq!(next_id(&[tl::enums::DialogFilter::Default, filter(2, "A", |_| {}), filter(7, "B", |_| {})]), 8);
}
#[test]
fn a_new_folder_starts_with_the_one_chat_it_was_made_for() {
let made = folder_of(&starting_with(3, "Friends", &user(100)), OWN).unwrap();
assert_eq!((made.id, made.title.as_str()), (3, "Friends"));
assert_eq!(made.rules.include, vec!["100"]);
assert!(!made.rules.holds("101", Some(&facts(Kind::Person))), "no kind of chat is taken until asked for");
}
#[test]
fn folders_are_put_in_the_order_asked_for_and_the_entry_for_every_chat_keeps_its_place() {
assert_eq!(ordered(&[0, 2, 3, 4], &[4, 2, 3]), vec![0, 4, 2, 3]);
assert_eq!(ordered(&[2, 3, 0, 4], &[4, 3]), vec![2, 4, 0, 3], "the two named swap places, the rest stay");
assert_eq!(ordered(&[0, 2, 3], &[9, 3, 3, 2]), vec![0, 3, 2], "numbers that are not folders, and repeats, are dropped");
assert_eq!(ordered(&[0, 2, 3], &[]), vec![0, 2, 3]);
}
#[test]
fn the_summary_is_in_telegrams_order_with_each_folders_conversations() {
let a = folder_of(&filter(3, "Second", |f| f.include_peers = vec![user(100)]), OWN).unwrap();
let b = folder_of(&filter(2, "First", |f| f.bots = true), OWN).unwrap();
let mut known = HashMap::new();
known.insert("200".to_string(), facts(Kind::Bot));
let out = summary(&[a, b], &known);
assert_eq!(out[0]["id"], 3);
assert_eq!(out[0]["members"], json!(["100"]));
assert_eq!(out[1]["name"], "First");
assert_eq!(out[1]["members"], json!(["200"]));
}
#[test]
fn telegrams_refusals_about_folders_are_said_in_words() {
assert!(folder_words("rpc error 400: FILTER_INCLUDE_EMPTY").contains("empty folder"));
assert!(folder_words("DIALOG_FILTERS_TOO_MUCH").contains("as many folders"));
// As the library words it after `explain`: seen from the real account.
assert!(folder_words("Telegram refused: filter include empty").contains("empty folder"));
assert_eq!(folder_words("SOMETHING_ELSE"), "SOMETHING_ELSE");
}
}
+197
View File
@@ -0,0 +1,197 @@
//! GIFs: the account's saved ones, and a search through Telegram's own GIF bot.
//!
//! Telegram does not send a GIF as a link: a GIF is a short silent video that lives on Telegram, and sending one is
//! sending that file. The account keeps a list of saved GIFs, and the search is the inline bot `@gif`, which answers
//! with files of the same kind. Both are listed as the picker's cells; the pictures and the videos are named in the
//! cache and fetched when the picker draws them, from the files remembered here (a file is fetched with its access
//! key and a reference that goes stale, so what was listed is what can be fetched).
use super::media::{download_thumb, download_to, extension_of, MEDIA};
use super::messages;
use super::send::{explain, peer_for};
use super::Handle;
use crate::model::Buffer;
use crate::state::AppState;
use anyhow::{anyhow, Context, Result};
use grammers_client::media::Document;
use grammers_client::message::InputMessage;
use grammers_client::tl;
use serde_json::{json, Value};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
/// How many GIFs a list shows.
const LIMIT: usize = 100;
/// The documents that were listed, by (account, id): what the picker names is what can be fetched or sent.
type Known = Mutex<HashMap<(i64, i64), tl::types::Document>>;
fn known() -> &'static Known {
static K: std::sync::OnceLock<Known> = std::sync::OnceLock::new();
K.get_or_init(Default::default)
}
/// A GIF's cached file: `tgif_<account>_<id>_<thumb|full>.<ext>`.
fn name(account: i64, id: i64, thumb: bool, ext: &str) -> String {
format!("tgif_{account}_{id}_{}.{ext}", if thumb { "thumb" } else { "full" })
}
/// What a `tgif_` name says: the account, the document and which of the two files it is.
fn parse_name(file: &str) -> Option<(i64, i64, bool)> {
let rest = file.strip_prefix("tgif_")?;
let stem = rest.rsplit_once('.')?.0;
let mut parts = stem.split('_');
let account = parts.next()?.parse().ok()?;
let id = parts.next()?.parse().ok()?;
let thumb = match parts.next()? {
"thumb" => true,
"full" => false,
_ => return None,
};
parts.next().is_none().then_some((account, id, thumb))
}
fn dimensions(doc: &tl::types::Document) -> (i32, i32) {
doc.attributes
.iter()
.find_map(|a| match a {
tl::enums::DocumentAttribute::Video(v) => Some((v.w, v.h)),
tl::enums::DocumentAttribute::ImageSize(s) => Some((s.w, s.h)),
_ => None,
})
.unwrap_or((0, 0))
}
fn title_of(doc: &tl::types::Document) -> String {
doc.attributes
.iter()
.find_map(|a| match a {
tl::enums::DocumentAttribute::Filename(f) => Some(f.file_name.clone()),
_ => None,
})
.unwrap_or_else(|| "GIF".to_string())
}
pub(super) fn wrapper(doc: &tl::types::Document) -> Document {
Document::from_raw_media(tl::types::MessageMediaDocument {
nopremium: false,
spoiler: false,
video: true,
round: false,
voice: false,
document: Some(tl::enums::Document::Document(doc.clone())),
alt_documents: None,
video_cover: None,
video_timestamp: None,
ttl_seconds: None,
})
}
/// A listed document as the picker's cell: its files named in the cache, fetched when drawn.
fn entry(account: i64, doc: &tl::types::Document) -> Value {
let (w, h) = dimensions(doc);
let full = format!("file://{}", crate::media_cache::dir(MEDIA).join(name(account, doc.id, false, &extension_of(None, Some(&doc.mime_type)))).display());
let thumb = format!("file://{}", crate::media_cache::dir(MEDIA).join(name(account, doc.id, true, "jpg")).display());
json!({ "id": doc.id.to_string(), "title": title_of(doc), "url": "", "src": full, "gifSrc": full, "preview": thumb, "width": w, "height": h })
}
fn remember(account: i64, docs: Vec<tl::types::Document>) -> Vec<Value> {
let mut out = Vec::new();
let mut held = known().lock().unwrap_or_else(|e| e.into_inner());
for doc in docs.into_iter().take(LIMIT) {
out.push(entry(account, &doc));
held.insert((account, doc.id), doc);
}
out
}
fn documents(list: Vec<tl::enums::Document>) -> Vec<tl::types::Document> {
list.into_iter()
.filter_map(|d| match d {
tl::enums::Document::Document(d) => Some(d),
_ => None,
})
.collect()
}
/// What the picker shows: the saved GIFs when nothing is typed, else what Telegram's GIF search finds.
pub async fn list(state: &AppState, buffer: &Buffer, query: &str) -> Result<Value, String> {
let (handle, peer) = peer_for(state, buffer).await?;
let account = handle.own.user_id;
let docs = if query.trim().is_empty() {
let answer = handle.client.invoke(&tl::functions::messages::GetSavedGifs { hash: 0 }).await.map_err(|e| explain(&e))?;
match answer {
tl::enums::messages::SavedGifs::Gifs(g) => documents(g.gifs),
tl::enums::messages::SavedGifs::NotModified => Vec::new(),
}
} else {
search(&handle, peer, query.trim()).await.map_err(|e| e.to_string())?
};
Ok(json!({ "gifs": remember(account, docs), "categories": [] }))
}
async fn search(handle: &Handle, peer: grammers_session::types::PeerRef, query: &str) -> Result<Vec<tl::types::Document>> {
let bot = handle.client.resolve_username("gif").await.map_err(|e| anyhow!("{}", explain(&e)))?.ok_or_else(|| anyhow!("Telegram's GIF search is not available"))?;
let bot_ref = bot.to_ref().await.ok().flatten().ok_or_else(|| anyhow!("Telegram's GIF search is not available"))?;
let answer = handle
.client
.invoke(&tl::functions::messages::GetInlineBotResults { bot: bot_ref.into(), peer: peer.into(), geo_point: None, query: query.to_string(), offset: String::new() })
.await
.map_err(|e| anyhow!("{}", explain(&e)))?;
let tl::enums::messages::BotResults::Results(found) = answer;
Ok(found
.results
.into_iter()
.filter_map(|r| match r {
tl::enums::BotInlineResult::BotInlineMediaResult(m) => m.document,
_ => None,
})
.filter_map(|d| match d {
tl::enums::Document::Document(d) => Some(d),
_ => None,
})
.collect())
}
/// Sends a GIF that was listed, to a conversation.
pub async fn send(state: &AppState, buffer: &Buffer, id: &str) -> Result<(), String> {
let id: i64 = id.parse().map_err(|_| "that is not a GIF".to_string())?;
let (handle, peer) = peer_for(state, buffer).await?;
let doc = known().lock().unwrap_or_else(|e| e.into_inner()).get(&(handle.own.user_id, id)).cloned().ok_or_else(|| "That GIF is no longer listed: open the picker again".to_string())?;
let media = wrapper(&doc).to_raw_input_media();
// In a topic of a group the GIF answers the topic's first message, which is how it is sent there.
let reply = super::topics::topic_of_buffer(buffer);
let sent = handle.client.send_message(peer, InputMessage::new().media(tl::enums::InputMedia::Document(media)).reply_to(reply)).await.map_err(|e| explain(&e))?;
messages::record(state, &buffer.account_id, &handle.own, &sent);
Ok(())
}
/// Fetches one of a listed GIF's files again, for the window asking for it.
pub async fn restore(state: &AppState, file: &str, path: &Path) -> Result<Option<PathBuf>> {
let Some((account, id, thumb)) = parse_name(file) else { return Ok(None) };
let handle = state.runtime.telegram_handle(&format!("telegram:{account}")).context("that Telegram account is not connected")?;
let doc = known().lock().unwrap_or_else(|e| e.into_inner()).get(&(account, id)).cloned().context("that GIF is no longer listed")?;
let wrapped = wrapper(&doc);
if thumb {
download_thumb(&handle.client, &wrapped, path).await?;
} else {
download_to(&handle.client, &wrapped, path, false).await?;
}
Ok(Some(path.to_path_buf()))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_gifs_files_are_named_and_read_back() {
let video = name(1234567890, 5_000_000_000_001, false, "mp4");
assert_eq!(video, "tgif_1234567890_5000000000001_full.mp4");
assert_eq!(parse_name(&video), Some((1234567890, 5_000_000_000_001, false)));
assert_eq!(parse_name(&name(1, 7, true, "jpg")), Some((1, 7, true)));
assert_eq!(parse_name("tg_1_n5_2_full.jpg"), None);
assert_eq!(parse_name("tgif_1_2_other.jpg"), None);
}
}
+271
View File
@@ -0,0 +1,271 @@
//! Telegram links: what a `t.me` address or a `tg://` link points at, and taking the window there.
//!
//! A link names a public chat (`t.me/disclosetv`, `tg://resolve?domain=disclosetv`), a message in one
//! (`t.me/disclosetv/21953`, `...&post=21953`), a message in a chat the account is already in
//! (`t.me/c/1234567/89`, `tg://privatepost?channel=1234567&post=89`) or a private invite (`t.me/+hash`,
//! `tg://join?invite=hash`); in a group with topics the topic comes before the message
//! (`t.me/group/5/21953`). Opening one makes the conversation if the account has none (a chat it is not in is
//! joined, which is what a link to a channel asks for) and answers where it is, and which message to go to.
//!
//! The web page Telegram shows for a post has a "View in Telegram" button that opens `tg://resolve?domain=..&post=..`,
//! which is how a click in a browser comes here once moho is the handler for `tg:` links.
use super::peers;
use super::send::explain;
use super::{dialogs, topics, Handle};
use crate::model::Buffer;
use crate::state::AppState;
use grammers_client::peer::Peer;
use grammers_client::tl;
use grammers_session::types::PeerId;
use grammers_session::Session;
use serde_json::{json, Value};
/// What a link points at.
#[derive(Debug, PartialEq)]
pub enum Link {
/// A chat named by its public @name.
Public { name: String, topic: Option<i32>, post: Option<i32> },
/// A chat named by its number, which only an account that is in it can open.
Private { channel: i64, topic: Option<i32>, post: Option<i32> },
/// A private invite.
Invite(String),
}
fn number(text: &str) -> Option<i32> {
text.parse::<i32>().ok().filter(|n| *n > 0)
}
fn name_ok(name: &str) -> bool {
(5..=32).contains(&name.len()) && name.chars().next().is_some_and(|c| c.is_ascii_alphabetic()) && name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
}
fn invite_ok(hash: &str) -> bool {
!hash.is_empty() && hash.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
}
/// Reads a `t.me` / `telegram.me` / `telegram.dog` address or a `tg://` link; `None` for anything else, including the
/// pages on those hosts that are not a chat (sticker packs, proxies, the web preview).
pub fn parse(input: &str) -> Option<Link> {
let url = url::Url::parse(input.trim()).ok()?;
let query = |key: &str| url.query_pairs().find(|(k, _)| k == key).map(|(_, v)| v.into_owned());
match url.scheme() {
"tg" => match url.host_str()? {
"resolve" => {
let name = query("domain")?;
if name == "telegrampassport" || !name_ok(&name) {
return None;
}
let topic = query("thread").or_else(|| query("topic")).and_then(|t| number(&t));
Some(Link::Public { name, topic, post: query("post").and_then(|p| number(&p)) })
}
"privatepost" => {
let channel: i64 = query("channel")?.parse().ok().filter(|c| *c > 0)?;
Some(Link::Private { channel, topic: query("thread").or_else(|| query("topic")).and_then(|t| number(&t)), post: query("post").and_then(|p| number(&p)) })
}
"join" => Some(Link::Invite(query("invite").filter(|h| invite_ok(h))?)),
_ => None,
},
"http" | "https" => {
if !matches!(url.host_str()?.trim_start_matches("www."), "t.me" | "telegram.me" | "telegram.dog") {
return None;
}
let parts: Vec<&str> = url.path_segments()?.filter(|p| !p.is_empty()).collect();
let first = *parts.first()?;
if let Some(hash) = first.strip_prefix('+') {
return invite_ok(hash).then(|| Link::Invite(hash.to_string()));
}
match first {
"joinchat" => parts.get(1).filter(|h| invite_ok(h)).map(|h| Link::Invite((*h).to_string())),
"c" => {
let channel: i64 = parts.get(1)?.parse().ok().filter(|c| *c > 0)?;
let (topic, post) = match (parts.get(2).and_then(|p| number(p)), parts.get(3).and_then(|p| number(p))) {
(Some(topic), Some(post)) => (Some(topic), Some(post)),
(Some(post), None) => (None, Some(post)),
_ => (None, None),
};
Some(Link::Private { channel, topic, post })
}
// Pages that are about something other than a chat.
"s" | "addstickers" | "addemoji" | "setlanguage" | "proxy" | "socks" | "share" | "login" | "iv" | "addtheme" | "bg" | "boost" | "m" | "addlist" | "invoice" | "giftcode" | "contact" => None,
name if name_ok(name) => {
let (topic, post) = match (parts.get(1).and_then(|p| number(p)), parts.get(2).and_then(|p| number(p))) {
(Some(topic), Some(post)) => (Some(topic), Some(post)),
(Some(post), None) => (None, Some(post)),
_ => (None, None),
};
Some(Link::Public { name: name.to_string(), topic, post })
}
_ => None,
}
}
_ => None,
}
}
/// Where a link led.
pub struct Opened {
pub account_id: String,
pub buffer: Buffer,
pub message_id: Option<String>,
}
/// The first connected Telegram account, when the window did not say which.
fn account_for(state: &AppState, wanted: Option<&str>) -> Result<(String, Handle), String> {
if let Some(id) = wanted {
return state.runtime.telegram_handle(id).map(|h| (id.to_string(), h)).ok_or_else(|| "that Telegram account is not connected".to_string());
}
state
.accounts
.all_telegram()
.into_iter()
.find_map(|a| {
let id = a.account_id();
state.runtime.telegram_handle(&id).map(|h| (id, h))
})
.ok_or_else(|| "Add a Telegram account, and wait for it to connect, to open that link".to_string())
}
/// Whether the chat is one this account has a conversation for.
fn known(state: &AppState, account_id: &str, peer: &Peer) -> bool {
let remote = peers::remote_id(peer.id());
state.runtime.list_buffers().iter().any(|b| b.account_id == account_id && b.remote_id.as_deref().is_some_and(|r| r == remote || r.strip_prefix(remote.as_str()).is_some_and(|rest| rest.starts_with(':'))))
}
fn chats_of(updates: &tl::enums::Updates) -> Vec<tl::enums::Chat> {
match updates {
tl::enums::Updates::Updates(u) => u.chats.clone(),
tl::enums::Updates::Combined(u) => u.chats.clone(),
_ => Vec::new(),
}
}
/// Joins a chat by its invite, and says which chat it was.
async fn join_invite(state: &AppState, account_id: &str, handle: &Handle, hash: &str) -> Result<Peer, String> {
let checked = handle.client.invoke(&tl::functions::messages::CheckChatInvite { hash: hash.to_string() }).await.map_err(|e| invite_words(&e))?;
// Already in it: the invite names a chat the account has.
if let tl::enums::ChatInvite::Already(already) = checked {
return Ok(Peer::from_raw(&handle.client, already.chat));
}
let answer = handle.client.invoke(&tl::functions::messages::ImportChatInvite { hash: hash.to_string() }).await.map_err(|e| invite_words(&e))?;
let tl::enums::messages::ChatInviteJoinResult::Ok(joined) = answer else {
return Err("That chat has to be joined from Telegram's own app first".to_string());
};
let chat = chats_of(&joined.updates).into_iter().next().ok_or("Telegram did not say which chat that was")?;
let peer = Peer::from_raw(&handle.client, chat);
let _ = dialogs::sync(state, account_id, handle).await;
Ok(peer)
}
fn invite_words(e: &grammers_client::InvocationError) -> String {
match e {
grammers_client::InvocationError::Rpc(r) if r.name == "INVITE_HASH_EXPIRED" || r.name == "INVITE_HASH_INVALID" => "That invite link is no longer valid".to_string(),
grammers_client::InvocationError::Rpc(r) if r.name == "INVITE_REQUEST_SENT" => "Your request to join was sent: an administrator has to accept it".to_string(),
other => explain(other),
}
}
/// The conversation a message of a chat is in: the chat's own, or in a group with topics the topic's.
fn buffer_in(state: &AppState, account_id: &str, peer: &Peer, topic: Option<i32>, post: Option<i32>) -> Buffer {
if topics::is_forum(peer) {
let remote = peers::remote_id(peer.id());
return match (topic, post) {
(Some(topic), _) => topics::buffer_for_topic(state, account_id, peer, topic, None),
(None, Some(post)) => topics::holding(state, account_id, &remote, post).unwrap_or_else(|| topics::buffer_for_topic(state, account_id, peer, topics::GENERAL, None)),
(None, None) => topics::buffer_for_topic(state, account_id, peer, topics::GENERAL, None),
};
}
peers::buffer_for(state, account_id, peer)
}
/// Opens a link: finds or makes the conversation it points into, joining the chat if the account is not in it.
pub async fn open(state: &AppState, account: Option<&str>, input: &str) -> Result<Opened, String> {
let link = parse(input).ok_or_else(|| "That is not a Telegram link moho can open".to_string())?;
let (account_id, handle) = account_for(state, account)?;
let (peer, topic, post) = match link {
Link::Invite(hash) => (join_invite(state, &account_id, &handle, &hash).await?, None, None),
Link::Public { name, topic, post } => {
let peer = handle.client.resolve_username(&name).await.map_err(|e| explain(&e))?.ok_or_else(|| format!("Nobody on Telegram is called @{name}"))?;
if !matches!(peer, Peer::User(_)) && !known(state, &account_id, &peer) {
// A link to a channel asks to be in it.
let peer_ref = peer.to_ref().await.ok().flatten().ok_or("Telegram did not say how to reach that chat")?;
handle.client.join_chat(peer_ref).await.map_err(|e| invite_words(&e))?;
let _ = dialogs::sync(state, &account_id, &handle).await;
}
(peer, topic, post)
}
Link::Private { channel, topic, post } => {
let peer_ref = handle.session.peer_ref(PeerId::channel_unchecked(channel)).await.ok().flatten().ok_or("This account is not in that chat, so a link by its number cannot open it")?;
(handle.client.resolve_peer(peer_ref).await.map_err(|e| explain(&e))?, topic, post)
}
};
let buffer = buffer_in(state, &account_id, &peer, topic, post);
Ok(Opened { account_id, buffer, message_id: post.map(|p| p.to_string()) })
}
/// What the window is told.
pub fn answer(opened: &Opened) -> Value {
json!({ "accountId": opened.account_id, "bufferId": opened.buffer.id, "messageId": opened.message_id })
}
#[cfg(test)]
mod tests {
use super::*;
fn public(name: &str, topic: Option<i32>, post: Option<i32>) -> Option<Link> {
Some(Link::Public { name: name.into(), topic, post })
}
#[test]
fn a_web_link_to_a_post_names_the_channel_and_the_post() {
assert_eq!(parse("https://t.me/disclosetv/21953"), public("disclosetv", None, Some(21953)));
assert_eq!(parse("https://t.me/disclosetv/21953?single"), public("disclosetv", None, Some(21953)));
assert_eq!(parse("http://telegram.me/disclosetv"), public("disclosetv", None, None));
assert_eq!(parse("https://www.t.me/disclosetv/5"), public("disclosetv", None, Some(5)));
}
#[test]
fn in_a_group_with_topics_the_topic_comes_before_the_message() {
assert_eq!(parse("https://t.me/somegroup/7/21953"), public("somegroup", Some(7), Some(21953)));
assert_eq!(parse("tg://resolve?domain=somegroup&post=21953&thread=7"), public("somegroup", Some(7), Some(21953)));
}
#[test]
fn the_app_link_a_browsers_view_in_telegram_button_opens_is_read() {
assert_eq!(parse("tg://resolve?domain=disclosetv&post=21953"), public("disclosetv", None, Some(21953)));
assert_eq!(parse("tg://resolve?domain=disclosetv"), public("disclosetv", None, None));
// A bot started with a parameter is the bot's chat.
assert_eq!(parse("tg://resolve?domain=zlibrary_bot&start=abc"), public("zlibrary_bot", None, None));
assert_eq!(parse("tg://resolve?domain=telegrampassport&bot_id=1"), None);
}
#[test]
fn a_chat_named_by_its_number_is_private() {
assert_eq!(parse("https://t.me/c/1234567/89"), Some(Link::Private { channel: 1234567, topic: None, post: Some(89) }));
assert_eq!(parse("https://t.me/c/1234567/5/89"), Some(Link::Private { channel: 1234567, topic: Some(5), post: Some(89) }));
assert_eq!(parse("tg://privatepost?channel=1234567&post=89"), Some(Link::Private { channel: 1234567, topic: None, post: Some(89) }));
assert_eq!(parse("https://t.me/c/notanumber/89"), None);
}
#[test]
fn invites_are_read_in_every_form() {
let invite = |h: &str| Some(Link::Invite(h.to_string()));
assert_eq!(parse("https://t.me/+AbC_d-123"), invite("AbC_d-123"));
assert_eq!(parse("https://t.me/joinchat/AbC_d-123"), invite("AbC_d-123"));
assert_eq!(parse("tg://join?invite=AbC_d-123"), invite("AbC_d-123"));
assert_eq!(parse("tg://join?invite="), None);
}
#[test]
fn pages_that_are_not_a_chat_and_other_hosts_are_left_alone() {
assert_eq!(parse("https://t.me/addstickers/SomePack"), None);
assert_eq!(parse("https://t.me/s/disclosetv"), None);
assert_eq!(parse("https://t.me/proxy?server=x&port=1"), None);
assert_eq!(parse("https://t.me/"), None);
assert_eq!(parse("https://t.me/ab"), None, "too short to be a name");
assert_eq!(parse("https://example.com/disclosetv/1"), None);
assert_eq!(parse("https://t.me.evil.example/disclosetv/1"), None);
assert_eq!(parse("irc://irc.libera.chat/moho"), None);
assert_eq!(parse("not a link"), None);
}
}
+223 -7
View File
@@ -10,9 +10,20 @@
//! - `telegramLoginStatus` - what it is doing, for a line of text
//! - `telegramLoginNeeds` - what Telegram is asking for: `step` is `code` or `password` (with the password's hint),
//! and `error` says why the last answer was refused
//! - `telegramLoginQr` - a picture to scan, for a sign-in by QR code (below)
//! - `telegramLoginResult` - done, with the account, or the reason it failed
//!
//! and the window answers with `submitTelegramCode` and `submitTelegramPassword`.
//!
//! # By QR code
//!
//! The other way in is to scan a code with Telegram on a phone that is already signed in (Settings, Devices, Link
//! Desktop Device): no number is typed and no code is sent. Telegram gives a login token that lasts about half a
//! minute; it is shown as a QR code and asked for again when it runs out, and Telegram says when it was scanned
//! (`updateLoginToken`; the token is asked for again every so often too, in case that update does not reach a
//! connection that has no account yet). The scanned token comes back as the account, or - when the account lives
//! in another datacenter - as a pointer to it, where the token is handed in and the key made there becomes the
//! session's home. An account with a cloud password still asks for it, after the scan.
use super::link::{self, Link};
use super::session::{self, FileSession};
@@ -20,6 +31,8 @@ use crate::accounts::TelegramAccountConfig;
use crate::state::AppState;
use anyhow::{anyhow, bail, Context, Result};
use grammers_client::SignInError;
use grammers_client::tl;
use grammers_session::Session;
use serde_json::json;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
@@ -77,12 +90,29 @@ pub fn normalize_phone(typed: &str) -> Option<String> {
(7..=15).contains(&digits.len()).then(|| format!("+{digits}"))
}
/// Starts a sign-in. Answers at once; the rest is events.
/// How the account is being signed in.
enum Method {
/// A phone number, then the code sent to it.
Phone(String),
/// A QR code scanned from a phone that is already signed in.
Qr,
}
/// Starts a sign-in by phone number. Answers at once; the rest is events.
pub fn start(state: AppState, login_id: String, api_id: i32, api_hash: String, phone: String) {
begin(state, login_id, api_id, api_hash, Method::Phone(phone));
}
/// Starts a sign-in by QR code. Answers at once; the rest is events.
pub fn start_qr(state: AppState, login_id: String, api_id: i32, api_hash: String) {
begin(state, login_id, api_id, api_hash, Method::Qr);
}
fn begin(state: AppState, login_id: String, api_id: i32, api_hash: String, method: Method) {
let (tx, rx) = mpsc::unbounded_channel();
waiting().lock().unwrap().insert(login_id.clone(), tx);
tokio::spawn(async move {
let outcome = flow(&state, &login_id, api_id, &api_hash, &phone, rx).await;
let outcome = flow(&state, &login_id, api_id, &api_hash, method, rx).await;
waiting().lock().unwrap().remove(&login_id);
match outcome {
Ok(Some(saved)) => {
@@ -114,15 +144,21 @@ async fn answer(rx: &mut mpsc::UnboundedReceiver<Answer>) -> Option<Answer> {
tokio::time::timeout(PATIENCE, rx.recv()).await.ok().flatten()
}
async fn flow(state: &AppState, login_id: &str, api_id: i32, api_hash: &str, phone: &str, mut rx: mpsc::UnboundedReceiver<Answer>) -> Result<Option<TelegramAccountConfig>> {
async fn flow(state: &AppState, login_id: &str, api_id: i32, api_hash: &str, method: Method, mut rx: mpsc::UnboundedReceiver<Answer>) -> Result<Option<TelegramAccountConfig>> {
if api_id <= 0 || api_hash.trim().is_empty() {
bail!("an api_id and an api_hash are needed: get them at my.telegram.org/apps");
}
let phone = normalize_phone(phone).ok_or_else(|| anyhow!("that does not look like a phone number; type it with its country code"))?;
let typed_phone = match &method {
Method::Phone(phone) => Some(normalize_phone(phone).ok_or_else(|| anyhow!("that does not look like a phone number; type it with its country code"))?),
Method::Qr => None,
};
status(state, login_id, "Connecting to Telegram");
let session = Arc::new(FileSession::fresh(session::path_for_login(login_id)));
let link = link::open(state, &crate::net::route::pending_key("telegram"), api_id, session.clone()).await?;
let outcome = signed_in(state, login_id, &link, api_hash, &phone, &mut rx).await;
let mut link = link::open(state, &crate::net::route::pending_key("telegram"), api_id, session.clone()).await?;
let outcome = match &typed_phone {
Some(phone) => signed_in(state, login_id, &link, api_hash, phone, &mut rx).await,
None => scanned(state, login_id, &mut link, api_id, api_hash, &mut rx).await,
};
let user = match outcome {
Ok(Some(user)) => user,
other => {
@@ -134,7 +170,8 @@ async fn flow(state: &AppState, login_id: &str, api_id: i32, api_hash: &str, pho
let config = TelegramAccountConfig {
api_id,
api_hash: api_hash.trim().to_string(),
phone,
// What was typed, or - signed in by QR code, where nothing was - what Telegram says the number is.
phone: typed_phone.or_else(|| user.phone().map(|p| format!("+{}", p.trim_start_matches('+')))).unwrap_or_default(),
user_id: user.id().bare_id_unchecked(),
username: user.username().map(str::to_string),
name: user.full_name(),
@@ -176,6 +213,159 @@ async fn signed_in(state: &AppState, login_id: &str, link: &Link, api_hash: &str
}
}
/// How long the QR code is offered before the sign-in gives up: a form left open should not keep a task.
const QR_PATIENCE: Duration = Duration::from_secs(10 * 60);
/// How often the token is asked for again while nothing has been heard of a scan.
const QR_POLL: Duration = Duration::from_secs(15);
/// What asking for the login token came to.
#[derive(Debug)]
enum Exported {
/// A token to show, and when it stops being good (seconds since the epoch, Telegram's clock).
Token { token: Vec<u8>, expires: i32 },
/// Scanned and approved: the account.
Signed(Box<tl::types::auth::Authorization>),
/// Scanned, but the account has a cloud password.
NeedsPassword,
}
/// The link a phone's camera opens in Telegram.
pub fn qr_address(token: &[u8]) -> String {
use base64::Engine;
format!("tg://login?token={}", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(token))
}
fn qr_path(login_id: &str) -> std::path::PathBuf {
crate::media_cache::transient_dir().join(format!("moho-telegram-qr-{}.png", login_id.replace(|c: char| !c.is_ascii_alphanumeric(), "-")))
}
/// Asks Telegram for the login token, and follows it to the account's datacenter if it was scanned.
async fn export_token(client: &grammers_client::Client, session: &FileSession, api_id: i32, api_hash: &str) -> Result<Exported> {
let asked = client.invoke(&tl::functions::auth::ExportLoginToken { api_id, api_hash: api_hash.to_string(), except_ids: Vec::new() }).await;
let answer = match asked {
Ok(answer) => answer,
Err(e) if e.is("SESSION_PASSWORD_NEEDED") => return Ok(Exported::NeedsPassword),
Err(e) => bail!("{}", explain(&e.to_string())),
};
match answer {
tl::enums::auth::LoginToken::Token(t) => Ok(Exported::Token { token: t.token, expires: t.expires }),
tl::enums::auth::LoginToken::Success(s) => signed(s.authorization),
// The account is kept in another datacenter: the token is handed in there, and the key made there is the
// session's from then on.
tl::enums::auth::LoginToken::MigrateTo(m) => {
let imported = client.invoke_in_dc(m.dc_id, &tl::functions::auth::ImportLoginToken { token: m.token }).await;
match imported {
Ok(tl::enums::auth::LoginToken::Success(s)) => {
session.set_home_dc_id(m.dc_id).await.map_err(|e| anyhow!("{e}"))?;
signed(s.authorization)
}
Ok(_) => bail!("Telegram did not finish the sign-in in the account's datacenter"),
Err(e) if e.is("SESSION_PASSWORD_NEEDED") => {
session.set_home_dc_id(m.dc_id).await.map_err(|e| anyhow!("{e}"))?;
Ok(Exported::NeedsPassword)
}
Err(e) => bail!("{}", explain(&e.to_string())),
}
}
}
}
fn signed(authorization: tl::enums::auth::Authorization) -> Result<Exported> {
match authorization {
tl::enums::auth::Authorization::Authorization(a) => Ok(Exported::Signed(Box::new(a))),
tl::enums::auth::Authorization::SignUpRequired(_) => bail!("this number has no Telegram account. Make one with Telegram's own app first"),
}
}
/// What the library does when it signs in by code, which it does not do for a login token: the account's own entry
/// in the session, and where its updates stand.
async fn settle(client: &grammers_client::Client, session: &FileSession, authorization: tl::types::auth::Authorization) -> Result<grammers_client::peer::User> {
use grammers_session::types::{PeerInfo, UpdateState, UpdatesState};
let update_state = client.invoke(&tl::functions::updates::GetState {}).await.ok();
let user = grammers_client::peer::User::from_raw(client, authorization.user);
let auth = user.to_ref().await.map_err(|e| anyhow!("{e}"))?.context("Telegram did not say who signed in")?.auth;
session
.cache_peer(PeerInfo::User { id: user.id().bare_id_unchecked(), auth: Some(auth), bot: Some(user.is_bot()), is_self: Some(true) })
.await
.map_err(|e| anyhow!("{e}"))?;
if let Some(tl::enums::updates::State::State(state)) = update_state {
session
.set_update_state(UpdateState::All(UpdatesState { pts: state.pts, qts: state.qts, date: state.date, seq: state.seq, channels: Vec::new() }))
.await
.map_err(|e| anyhow!("{e}"))?;
}
Ok(user)
}
/// The conversation with Telegram, up to being signed in by QR code.
async fn scanned(state: &AppState, login_id: &str, link: &mut Link, api_id: i32, api_hash: &str, rx: &mut mpsc::UnboundedReceiver<Answer>) -> Result<Option<grammers_client::peer::User>> {
let client = link.client.clone();
let session = link.session.clone();
if client.is_authorized().await.context("asking Telegram whether this session is signed in")? {
bail!("this session is already signed in");
}
// Telegram says a token was scanned with an update; a connection that cannot read it is covered by asking again.
let mut stream = match link.updates.take() {
Some(updates) => client.stream_updates(updates, grammers_mtsender::UpdatesConfiguration { catch_up: false }).await.ok(),
None => None,
};
let began = tokio::time::Instant::now();
let mut shown: Vec<u8> = Vec::new();
status(state, login_id, "Asking Telegram for a code to scan");
loop {
if began.elapsed() > QR_PATIENCE {
bail!("The code was not scanned in time. Start again to get a new one");
}
match export_token(&client, &session, api_id, api_hash).await? {
Exported::Signed(authorization) => return Ok(Some(settle(&client, &session, *authorization).await?)),
Exported::NeedsPassword => return ask_cloud_password(state, login_id, &client, rx).await,
Exported::Token { token, expires } => {
if token != shown {
let path = qr_path(login_id);
super::super::steam::login::write_qr(&qr_address(&token), &path)?;
state.events.emit("telegramLoginQr", json!({ "loginId": login_id, "path": path.to_string_lossy(), "expires": expires }));
shown = token;
}
// Until it is scanned, it runs out, or it is time to ask again.
let now = chrono::Utc::now().timestamp();
let until = Duration::from_secs((i64::from(expires) - now - 2).clamp(1, QR_POLL.as_secs() as i64) as u64);
let heard = async {
match stream.as_mut() {
Some(stream) => loop {
match stream.next().await {
Ok(grammers_client::update::Update::Raw(raw)) if matches!(raw.raw, tl::enums::Update::LoginToken) => return,
Ok(_) => {}
// Not readable here: the poll is what notices a scan.
Err(_) => return std::future::pending().await,
}
},
None => std::future::pending().await,
}
};
tokio::select! {
_ = tokio::time::sleep(until) => {}
_ = heard => {}
given = rx.recv() => {
// The form was closed.
if given.is_none() {
return Ok(None);
}
}
}
}
}
}
}
/// Asks for the cloud password of an account that was scanned in, and signs it in.
async fn ask_cloud_password(state: &AppState, login_id: &str, client: &grammers_client::Client, rx: &mut mpsc::UnboundedReceiver<Answer>) -> Result<Option<grammers_client::peer::User>> {
let password: tl::types::account::Password = client.invoke(&tl::functions::account::GetPassword {}).await.map_err(|e| anyhow!("{}", explain(&e.to_string())))?.into();
let token = grammers_client::client::PasswordToken::new(password);
needs(state, login_id, "password", token.hint(), None);
ask_password(state, login_id, client, token, rx).await
}
async fn ask_password(state: &AppState, login_id: &str, client: &grammers_client::Client, mut token: grammers_client::client::PasswordToken, rx: &mut mpsc::UnboundedReceiver<Answer>) -> Result<Option<grammers_client::peer::User>> {
loop {
let Some(given) = answer(rx).await else { return Ok(None) };
@@ -239,6 +429,32 @@ mod tests {
assert!(submit_code("telegram-t", "12345").is_err());
}
#[test]
fn a_login_token_is_offered_as_telegrams_link_with_the_token_in_base64url() {
// The alphabet is the URL-safe one, and there is no padding: a phone reads the link as it is written.
assert_eq!(qr_address(&[0xfb, 0xff]), "tg://login?token=-_8");
assert_eq!(qr_address(&[1, 2, 3, 4]), "tg://login?token=AQIDBA");
assert!(qr_address(&[0; 32]).starts_with("tg://login?token="));
}
#[test]
fn a_scanned_codes_picture_is_written_where_the_window_can_read_it() {
let path = std::env::temp_dir().join(format!("moho-telegram-qr-test-{}.png", std::process::id()));
super::super::super::steam::login::write_qr(&qr_address(&[7; 32]), &path).unwrap();
let bytes = std::fs::read(&path).unwrap();
std::fs::remove_file(&path).ok();
assert_eq!(&bytes[..4], &[0x89, b'P', b'N', b'G']);
}
#[test]
fn a_sign_in_by_qr_is_waited_on_like_one_by_phone_and_given_up_by_closing_the_form() {
let (tx, mut rx) = mpsc::unbounded_channel();
waiting().lock().unwrap().insert("telegram-qr".into(), tx);
cancel("telegram-qr");
assert!(rx.try_recv().is_err());
assert!(submit_password("telegram-qr", "x").is_err(), "nothing is waiting for a password once the form is closed");
}
#[test]
fn telegrams_refusals_are_said_so_a_person_can_act_on_them() {
assert!(explain("rpc error 400: API_ID_INVALID").contains("my.telegram.org"));
+8
View File
@@ -406,6 +406,14 @@ pub async fn restore_cached(state: &AppState, path: &Path) -> Result<PathBuf> {
fetch_avatar(&handle, &peer, photo, path).await?;
return Ok(path.to_path_buf());
}
// A GIF the picker listed is fetched from what the picker remembered of it.
if name.starts_with("tgif_") {
return super::gifs::restore(state, name, path).await?.context("not a cached Telegram GIF");
}
// So is a sticker's picture.
if name.starts_with("tstk_") {
return super::stickers::restore(state, name, path).await?.context("not a cached Telegram sticker");
}
let (account, remote, message, variant) = parse_media_name(name).context("not a cached Telegram file")?;
// A preview already on its way is waited for, not fetched a second time.
for _ in 0..120 {
+8
View File
@@ -7,6 +7,7 @@
//! - this file - the identity moho gives Telegram, and what every other file shares
//! - `session` - what the library remembers between runs, in a private file of its own per account
//! - `link` - one connection to Telegram: the sender pool, the client and the stream of updates
//! - `links` - what a t.me or tg:// link points at, and opening it
//! - `login` - adding an account: phone, code, cloud password
//! - `connect` - an account's connection, kept up and reconnected
//! - `dialogs` - the account's conversations, read when it connects
@@ -29,6 +30,9 @@
//! - `keyboards` - a bot's inline buttons
//! - `bots` - a bot's command menu
//! - `topics` - a group with Topics as several conversations
//! - `folders` - the account's chat folders, kept in step with moho's categories
//! - `gifs` - saved GIFs and the GIF search
//! - `stickers` - the account's sticker packs, and sending one
//! - `peers` - a conversation as a buffer: its name, its kind, its heading in the list
//! - `messages` - a message as moho's, with its entities as html
//!
@@ -49,10 +53,13 @@ pub mod admin;
pub mod bots;
pub mod connect;
pub mod dialogs;
pub mod folders;
pub mod gifs;
pub mod history;
pub mod instant;
pub mod keyboards;
pub mod link;
pub mod links;
pub mod login;
pub mod markup;
pub mod media;
@@ -67,6 +74,7 @@ pub mod profile;
pub mod reactions;
pub mod search;
pub mod send;
pub mod stickers;
pub mod threads;
pub mod topics;
pub mod updates;
+3 -1
View File
@@ -191,10 +191,12 @@ pub async fn mark_read(state: &AppState, buffer: &Buffer) -> Result<(), String>
let (handle, peer) = peer_for(state, buffer).await?;
// A topic is read on its own: to the newest message of it that is here.
if let Some(topic) = super::topics::topic_of_buffer(buffer) {
let newest = state.store.newest_message_id(&buffer.id).ok().flatten().and_then(|id| id.parse::<i32>().ok()).unwrap_or(0);
// Or to the newest Telegram said it has, which may be a message this client never kept.
let newest = state.store.newest_message_id(&buffer.id).ok().flatten().and_then(|id| id.parse::<i32>().ok()).unwrap_or(0).max(super::topics::top_of(&buffer.id).unwrap_or(0));
if newest > 0 {
handle.client.invoke(&tl::functions::messages::ReadDiscussion { peer: peer.into(), msg_id: topic, read_max_id: newest }).await.map_err(|e| explain(&e))?;
}
state.runtime.set_buffer_unread(state, &buffer.id, 0, 0);
return Ok(());
}
handle.client.mark_as_read(peer).await.map_err(|e| explain(&e))
+260
View File
@@ -0,0 +1,260 @@
//! Stickers: the packs an account has installed, the ones it uses most, and sending one.
//!
//! A sticker is a document that lives on Telegram, so sending one is sending that file, the way a GIF is. The account
//! keeps three lists the picker draws: the stickers it sent lately, the ones it marked as favourites, and its
//! installed packs. A pack is asked for on its own, which is a request each, so they are read a few at a time and
//! kept for a while. Each sticker is drawn from the still picture Telegram sends with it (an animated or video
//! sticker has one too), named in the cache and fetched when the picker draws it, from the document remembered
//! here: a file is fetched with its access key and a reference that goes stale, so what was listed is what can be
//! fetched.
use super::gifs::wrapper;
use super::media::{download_thumb, download_to, MEDIA};
use super::messages;
use super::send::{explain, peer_for};
use super::Handle;
use crate::model::Buffer;
use crate::state::AppState;
use anyhow::{Context, Result};
use futures::StreamExt;
use grammers_client::message::InputMessage;
use grammers_client::tl;
use serde_json::{json, Value};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::time::{Duration, Instant};
/// How many of the most recent stickers are listed.
const RECENT: usize = 40;
/// How many installed packs are read. An account can have hundreds; the picker is for the ones in daily use, which
/// Telegram lists first.
const PACKS: usize = 40;
/// How many packs are asked for at once.
const AT_ONCE: usize = 4;
/// How long a pack's stickers are kept before it is asked for again.
const KEEP: Duration = Duration::from_secs(15 * 60);
/// The documents that were listed, by (account, id): what the picker names is what can be fetched or sent.
type Known = Mutex<HashMap<(i64, i64), tl::types::Document>>;
fn known() -> &'static Known {
static K: std::sync::OnceLock<Known> = std::sync::OnceLock::new();
K.get_or_init(Default::default)
}
/// A pack's stickers, when they were read.
type Packs = Mutex<HashMap<(i64, i64), (Instant, Vec<tl::types::Document>)>>;
fn packs() -> &'static Packs {
static P: std::sync::OnceLock<Packs> = std::sync::OnceLock::new();
P.get_or_init(Default::default)
}
/// A sticker's cached picture: `tstk_<account>_<id>.webp`.
fn name(account: i64, id: i64) -> String {
format!("tstk_{account}_{id}.webp")
}
/// What a `tstk_` name says: the account and the document.
fn parse_name(file: &str) -> Option<(i64, i64)> {
let stem = file.strip_prefix("tstk_")?.strip_suffix(".webp")?;
let (account, id) = stem.split_once('_')?;
Some((account.parse().ok()?, id.parse().ok()?))
}
/// The emoji a sticker stands for, which is also what it is called.
fn emoji_of(doc: &tl::types::Document) -> String {
doc.attributes
.iter()
.find_map(|a| match a {
tl::enums::DocumentAttribute::Sticker(s) => Some(s.alt.clone()),
_ => None,
})
.unwrap_or_default()
}
/// A listed document as the picker's cell, its picture named in the cache and fetched when drawn.
fn entry(account: i64, pack: &str, doc: &tl::types::Document) -> Value {
let emoji = emoji_of(doc);
let url = format!("file://{}", crate::media_cache::dir(MEDIA).join(name(account, doc.id)).display());
json!({ "id": doc.id.to_string(), "name": if emoji.is_empty() { "Sticker".to_string() } else { emoji.clone() }, "pack": pack, "body": emoji, "url": url })
}
fn documents(list: Vec<tl::enums::Document>) -> Vec<tl::types::Document> {
list.into_iter()
.filter_map(|d| match d {
tl::enums::Document::Document(d) => Some(d),
_ => None,
})
.collect()
}
/// Adds a heading's stickers to the list and remembers the documents.
fn push(out: &mut Vec<Value>, account: i64, pack: &str, docs: Vec<tl::types::Document>) {
let mut held = known().lock().unwrap_or_else(|e| e.into_inner());
for doc in docs {
// A sticker can be in the recent list and in its pack; the picker shows it under each heading, since the
// heading is how it is found, but it is remembered and fetched once.
out.push(entry(account, pack, &doc));
held.insert((account, doc.id), doc);
}
}
/// One installed pack's stickers, from what was kept or from Telegram.
async fn pack_documents(handle: Handle, account: i64, id: i64, access_hash: i64) -> Result<Vec<tl::types::Document>> {
let key = (account, id);
if let Some((when, docs)) = packs().lock().unwrap_or_else(|e| e.into_inner()).get(&key) {
if when.elapsed() < KEEP {
return Ok(docs.clone());
}
}
let answer = handle
.client
.invoke(&tl::functions::messages::GetStickerSet { stickerset: tl::enums::InputStickerSet::Id(tl::types::InputStickerSetId { id, access_hash }), hash: 0 })
.await
.map_err(|e| anyhow::anyhow!("{}", explain(&e)))?;
let docs = match answer {
tl::enums::messages::StickerSet::Set(s) => documents(s.documents),
tl::enums::messages::StickerSet::NotModified => Vec::new(),
};
packs().lock().unwrap_or_else(|e| e.into_inner()).insert(key, (Instant::now(), docs.clone()));
Ok(docs)
}
/// What the picker shows: the recent stickers, the favourites, then the installed packs in the account's order.
pub async fn list(state: &AppState, buffer: &Buffer) -> Result<Value, String> {
let (handle, _) = peer_for(state, buffer).await?;
let account = handle.own.user_id;
let mut out = Vec::new();
// The two short lists are a nicety: a failure reading them leaves the packs to be listed.
match handle.client.invoke(&tl::functions::messages::GetRecentStickers { attached: false, hash: 0 }).await {
Ok(tl::enums::messages::RecentStickers::Stickers(r)) => {
let docs: Vec<_> = documents(r.stickers).into_iter().take(RECENT).collect();
push(&mut out, account, "Recent", docs);
}
Ok(_) => {}
Err(e) => tracing::debug!("telegram: recent stickers: {}", explain(&e)),
}
match handle.client.invoke(&tl::functions::messages::GetFavedStickers { hash: 0 }).await {
Ok(tl::enums::messages::FavedStickers::Stickers(f)) => push(&mut out, account, "Favourites", documents(f.stickers)),
Ok(_) => {}
Err(e) => tracing::debug!("telegram: favourite stickers: {}", explain(&e)),
}
let sets = match handle.client.invoke(&tl::functions::messages::GetAllStickers { hash: 0 }).await.map_err(|e| explain(&e))? {
tl::enums::messages::AllStickers::Stickers(s) => s.sets,
tl::enums::messages::AllStickers::NotModified => Vec::new(),
};
let sets: Vec<tl::types::StickerSet> = sets
.into_iter()
.map(|s| {
let tl::enums::StickerSet::Set(s) = s;
s
})
// Masks are for putting on photos and custom emoji are typed into text: neither is sent as a sticker.
.filter(|s| !s.masks && !s.emojis && !s.archived)
.take(PACKS)
.collect();
let fetched: Vec<_> = futures::stream::iter(sets.into_iter().map(|set| {
let handle = handle.clone();
async move { (set.title, pack_documents(handle, account, set.id, set.access_hash).await) }
}))
.buffered(AT_ONCE)
.collect()
.await;
for (title, result) in fetched {
match result {
Ok(docs) => push(&mut out, account, &title, docs),
Err(e) => tracing::debug!("telegram: sticker pack {title}: {e}"),
}
}
Ok(Value::Array(out))
}
/// Sends a sticker that was listed, to a conversation.
pub async fn send(state: &AppState, buffer: &Buffer, id: &str) -> Result<(), String> {
let id: i64 = id.parse().map_err(|_| "that is not a sticker".to_string())?;
let (handle, peer) = peer_for(state, buffer).await?;
let doc = known().lock().unwrap_or_else(|e| e.into_inner()).get(&(handle.own.user_id, id)).cloned().ok_or_else(|| "That sticker is no longer listed: open the picker again".to_string())?;
let media = wrapper(&doc).to_raw_input_media();
// In a topic of a group the sticker answers the topic's first message, which is how it is sent there.
let reply = super::topics::topic_of_buffer(buffer);
let sent = handle.client.send_message(peer, InputMessage::new().media(tl::enums::InputMedia::Document(media)).reply_to(reply)).await.map_err(|e| explain(&e))?;
messages::record(state, &buffer.account_id, &handle.own, &sent);
Ok(())
}
/// Fetches a listed sticker's picture again, for the window asking for it.
pub async fn restore(state: &AppState, file: &str, path: &Path) -> Result<Option<PathBuf>> {
let Some((account, id)) = parse_name(file) else { return Ok(None) };
let handle = state.runtime.telegram_handle(&format!("telegram:{account}")).context("that Telegram account is not connected")?;
let doc = known().lock().unwrap_or_else(|e| e.into_inner()).get(&(account, id)).cloned().context("that sticker is no longer listed")?;
let wrapped = wrapper(&doc);
// A still sticker is its own picture; everything else is drawn from the preview that comes with it.
if wrapped.thumbs().is_empty() && doc.mime_type == "image/webp" {
download_to(&handle.client, &wrapped, path, false).await?;
} else {
download_thumb(&handle.client, &wrapped, path).await?;
}
Ok(Some(path.to_path_buf()))
}
#[cfg(test)]
mod tests {
use super::*;
fn sticker(id: i64, alt: &str) -> tl::types::Document {
tl::types::Document {
id,
access_hash: 9,
file_reference: vec![1],
date: 0,
mime_type: "image/webp".into(),
size: 10,
thumbs: None,
video_thumbs: None,
dc_id: 2,
attributes: vec![tl::enums::DocumentAttribute::Sticker(tl::types::DocumentAttributeSticker {
mask: false,
alt: alt.into(),
stickerset: tl::enums::InputStickerSet::Empty,
mask_coords: None,
})],
}
}
#[test]
fn a_stickers_picture_is_named_and_read_back() {
let file = name(1234567890, 5_000_000_000_001);
assert_eq!(file, "tstk_1234567890_5000000000001.webp");
assert_eq!(parse_name(&file), Some((1234567890, 5_000_000_000_001)));
assert_eq!(parse_name("tgif_1_2_full.mp4"), None);
assert_eq!(parse_name("tstk_1_x.webp"), None);
}
#[test]
fn a_sticker_is_called_by_its_emoji() {
let cell = entry(5, "Cats", &sticker(77, "😺"));
assert_eq!(cell["id"], "77");
assert_eq!(cell["name"], "😺");
assert_eq!(cell["body"], "😺");
assert_eq!(cell["pack"], "Cats");
assert!(cell["url"].as_str().unwrap().ends_with("tstk_5_77.webp"));
// One without an emoji is still a cell that can be told from a missing one.
assert_eq!(entry(5, "Cats", &sticker(78, ""))["name"], "Sticker");
}
#[test]
fn a_sticker_listed_twice_is_remembered_once_and_shown_under_each_heading() {
let mut out = Vec::new();
push(&mut out, 424242, "Recent", vec![sticker(1, "a")]);
push(&mut out, 424242, "Cats", vec![sticker(1, "a"), sticker(2, "b")]);
assert_eq!(out.len(), 3);
assert!(known().lock().unwrap().contains_key(&(424242, 2)));
}
}
+77
View File
@@ -131,6 +131,28 @@ pub fn holding(state: &AppState, account_id: &str, remote: &str, msg_id: i32) ->
topics.iter().find(|b| state.store.get_message(&b.id, &msg_id.to_string()).ok().flatten().is_some()).or(topics.first()).map(|b| (*b).clone())
}
/// The newest message of each topic as Telegram last said, by the topic's conversation: what reading it reads up to,
/// since this client may not have kept the messages that are unread.
fn tops() -> &'static std::sync::Mutex<std::collections::HashMap<String, i32>> {
static T: std::sync::OnceLock<std::sync::Mutex<std::collections::HashMap<String, i32>>> = std::sync::OnceLock::new();
T.get_or_init(Default::default)
}
/// The newest message Telegram has told of in a topic's conversation, if it has.
pub fn top_of(buffer_id: &str) -> Option<i32> {
tops().lock().unwrap_or_else(|e| e.into_inner()).get(buffer_id).copied()
}
/// How many messages in a topic are unread, and how many of those mention this account.
fn counts(topic: &tl::types::ForumTopic) -> (u32, u32) {
(topic.unread_count.max(0) as u32, topic.unread_mentions_count.max(0) as u32)
}
/// Whether a read position reaches everything the topic is known to hold.
pub fn read_all(read_max: i32, newest_here: i32, top: Option<i32>) -> bool {
read_max >= newest_here.max(top.unwrap_or(0))
}
/// Reads the group's topics and makes a conversation of each, named as the group names it. The one conversation the
/// group had before it had topics is removed: its messages are in the topics now.
pub async fn sync(state: &AppState, account_id: &str, handle: &Handle, peer: &Peer) -> Result<usize> {
@@ -156,6 +178,10 @@ pub async fn sync(state: &AppState, account_id: &str, handle: &Handle, peer: &Pe
let now = chrono::Utc::now().timestamp();
let reason = if topic.closed { Some("This topic is closed.") } else { peers::read_only_for(peer, now) };
state.runtime.set_buffer_read_only(state, &buffer.id, reason);
// The number Telegram shows beside the topic: a topic never opened here has no messages to count.
let (unread, mentions) = counts(topic);
state.runtime.set_buffer_unread(state, &buffer.id, unread, mentions);
tops().lock().unwrap_or_else(|e| e.into_inner()).insert(buffer.id.clone(), topic.top_message);
count += 1;
}
Ok(count)
@@ -197,4 +223,55 @@ mod tests {
assert_eq!(display_name("Chasm", " Support "), "Chasm/#Support");
assert_eq!(peers::peer_id_of("-1001234:45"), peers::peer_id_of("-1001234"));
}
#[test]
fn a_topics_unread_numbers_are_what_telegram_counts() {
let mut topic = tl::types::ForumTopic {
my: false,
closed: false,
pinned: false,
short: false,
hidden: false,
title_missing: false,
id: 45,
date: 0,
peer: tl::enums::Peer::Channel(tl::types::PeerChannel { channel_id: 1 }),
title: "Support".into(),
icon_color: 0,
icon_emoji_id: None,
top_message: 90,
read_inbox_max_id: 80,
read_outbox_max_id: 0,
unread_count: 7,
unread_mentions_count: 2,
unread_reactions_count: 0,
unread_poll_votes_count: 0,
from_id: tl::enums::Peer::Channel(tl::types::PeerChannel { channel_id: 1 }),
notify_settings: tl::enums::PeerNotifySettings::Settings(tl::types::PeerNotifySettings {
show_previews: None,
silent: None,
mute_until: None,
ios_sound: None,
android_sound: None,
other_sound: None,
stories_muted: None,
stories_hide_sender: None,
stories_ios_sound: None,
stories_android_sound: None,
stories_other_sound: None,
}),
draft: None,
};
assert_eq!(counts(&topic), (7, 2));
topic.unread_count = -1;
assert_eq!(counts(&topic).0, 0, "a count below nothing is nothing");
}
#[test]
fn a_topic_is_read_when_the_position_reaches_what_is_known_of_it() {
assert!(read_all(90, 85, Some(90)));
assert!(!read_all(80, 85, Some(90)), "the position is short of the newest Telegram has told of");
assert!(!read_all(80, 85, None), "and of the newest kept here");
assert!(read_all(0, 0, None), "a topic holding nothing is read");
}
}
+13
View File
@@ -42,6 +42,16 @@ fn read_up_to(state: &AppState, account_id: &str, remote: &str, max_id: i32) {
}
}
/// A topic was read on another of the account's apps: up to the message it names.
fn topic_read(state: &AppState, account_id: &str, group: &str, topic: i32, max_id: i32) {
let Some(buffer) = buffer_of(state, account_id, &super::topics::remote_of(group, topic)) else { return };
let newest = state.store.newest_message_id(&buffer.id).ok().flatten().and_then(|id| id.parse::<i32>().ok()).unwrap_or(0);
if super::topics::read_all(max_id, newest, super::topics::top_of(&buffer.id)) {
state.runtime.set_buffer_unread(state, &buffer.id, 0, 0);
state.events.emit("bufferRead", serde_json::json!({ "accountId": account_id, "bufferId": buffer.id }));
}
}
/// The pins of a conversation changed: read them again, which says so to every window.
fn pins_changed(state: &AppState, account_id: &str, _handle: &Handle, remote: &str) {
let Some(buffer) = buffer_of(state, account_id, remote) else { return };
@@ -81,6 +91,9 @@ pub fn raw(state: &AppState, account_id: &str, handle: &Handle, update: &tl::enu
U::PinnedMessages(u) => pins_changed(state, account_id, handle, &peers::remote_id(PeerId::from(u.peer.clone()))),
U::PinnedChannelMessages(u) => pins_changed(state, account_id, handle, &peers::remote_id(PeerId::channel_unchecked(u.channel_id))),
U::ReadHistoryInbox(u) => read_up_to(state, account_id, &peers::remote_id(PeerId::from(u.peer.clone())), u.max_id),
// The account's folders were made, changed, deleted or put in another order, on this client or another.
U::DialogFilter(_) | U::DialogFilters | U::DialogFilterOrder(_) => super::folders::changed(state, account_id, handle),
U::ReadChannelDiscussionInbox(u) => topic_read(state, account_id, &peers::remote_id(PeerId::channel_unchecked(u.channel_id)), u.top_msg_id, u.read_max_id),
U::ReadChannelInbox(u) => read_up_to(state, account_id, &peers::remote_id(PeerId::channel_unchecked(u.channel_id)), u.max_id),
_ => {}
}
+14
View File
@@ -395,6 +395,18 @@ pub struct Buffer {
/// posts, each of which is a thread. Opened as a pane of its own.
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub forum: bool,
/// How many messages the service itself counts as unread here, where it keeps a count a client could not work
/// out from the messages it has: a Telegram topic of a group that was never opened here. A starting point for
/// the window's own count, which then follows live messages; absent when there is none.
#[serde(rename = "unreadSeed", default, skip_serializing_if = "is_zero")]
pub unread_seed: u32,
/// How many of those mention this account.
#[serde(rename = "mentionsSeed", default, skip_serializing_if = "is_zero")]
pub mentions_seed: u32,
}
fn is_zero(n: &u32) -> bool {
*n == 0
}
/// The rail entry a buffer belongs to when its protocol has no grouping of
@@ -1048,6 +1060,8 @@ mod tests {
topic: None,
read_only: None,
forum: false,
unread_seed: 0,
mentions_seed: 0,
}
}
+124 -2
View File
@@ -473,7 +473,13 @@ pub async fn dispatch(
// connect: a few hundred channels would mean a few
// hundred requests, most for conversations nobody
// is about to read.
backend::discord::catch_up_channel(state, &cfg.token, &cfg.user_id, cfg.display_name.as_deref(), buffer_id, &channel_id).await;
// A conversation never read here is read for the first time now; one that has been is asked
// only about what came after.
if state.store.has_messages(buffer_id).unwrap_or(true) {
backend::discord::catch_up_channel(state, &cfg.token, &cfg.user_id, cfg.display_name.as_deref(), buffer_id, &channel_id).await;
} else {
backend::discord::backfill_channel_history(state, &cfg.token, &cfg.user_id, cfg.display_name.as_deref(), buffer_id, &channel_id).await;
}
}
}
// Steam keeps every conversation: opening one asks for its
@@ -756,6 +762,10 @@ pub async fn dispatch(
// name later.
state.highlights.forget(id);
state.ignores.forget(id);
// And the Discord session kept for a restart to resume: a removed account must not leave its
// servers and friends in a file.
backend::discord::session_store::forget(id);
backend::discord::send_guard::forget_account(id);
// And everything else this process is holding about it: the
// seventy-odd maps keyed by the account or by one of its
// buffers. In memory only, so it mattered exactly when
@@ -796,6 +806,7 @@ pub async fn dispatch(
match state.accounts.remove(id) {
Ok(true) => {
if let Some(config) = telegram_account {
backend::telegram::folders::forget(&config.account_id());
let state = state.clone();
tokio::spawn(async move { backend::telegram::connect::sign_out(&state, &config).await });
}
@@ -2290,7 +2301,7 @@ pub async fn dispatch(
"name": c.name,
"usage": "",
"description": c.description,
"source": buffer.name,
"source": c.source.clone().unwrap_or_else(|| buffer.name.clone()),
"kind": "bot",
}));
}
@@ -4623,6 +4634,19 @@ pub async fn dispatch(
(Some(serde_json::json!({ "loginId": login_id })), None)
}
// The same sign-in without a number: a QR code to scan with Telegram on a phone that is signed in already.
"addTelegramAccountQr" => {
let (Some(api_id), Some(api_hash)) = (params.get("apiId").and_then(|v| v.as_i64().or_else(|| v.as_str().and_then(|s| s.trim().parse().ok()))), p_str_opt(params, "apiHash")) else {
return (None, Some("addTelegramAccountQr requires \"apiId\" and \"apiHash\"".to_string()));
};
let Ok(api_id) = i32::try_from(api_id) else {
return (None, Some("that api_id is not a number Telegram would have given".to_string()));
};
let login_id = format!("telegram-{}", crate::model::next_message_id());
backend::telegram::login::start_qr(state.clone(), login_id.clone(), api_id, api_hash.trim().to_string());
(Some(serde_json::json!({ "loginId": login_id })), None)
}
// The code Telegram sent, and the cloud password it may ask for next, for a sign-in that is waiting.
"submitTelegramCode" | "submitTelegramPassword" => {
let (Some(login_id), Some(value)) = (p_str_opt(params, "loginId"), p_str_opt(params, "value")) else {
@@ -5771,6 +5795,104 @@ pub async fn dispatch(
// bufferId (not roomId) throughout, matching every other message-
// scoped Matrix method - the buffer->room_id lookup happens inside
// moderation.rs itself, same as sendMessage/editMessage/etc.
// The GIFs a Telegram account can send: its saved ones, or what Telegram's GIF search finds; and sending one.
"listTelegramGifs" | "sendTelegramGif" => {
let Some(buffer) = p_str_opt(params, "bufferId").and_then(|b| state.runtime.get_buffer(b)).filter(|b| b.account_id.starts_with("telegram:")) else {
return (None, Some(format!("{method} requires a Telegram \"bufferId\"")));
};
if method == "listTelegramGifs" {
return match backend::telegram::gifs::list(state, &buffer, p_str(params, "query", "")).await {
Ok(list) => (Some(list), None),
Err(e) => (None, Some(e)),
};
}
match backend::telegram::gifs::send(state, &buffer, p_str(params, "id", "")).await {
Ok(()) => (Some(ok_node()), None),
Err(e) => (None, Some(e)),
}
}
// A t.me or tg:// link followed: the conversation it points into (joined if need be) and the message to go to.
"openTelegramLink" => {
let Some(url) = p_str_opt(params, "url") else {
return (None, Some("openTelegramLink requires \"url\"".to_string()));
};
match backend::telegram::links::open(state, p_str_opt(params, "accountId"), url).await {
Ok(opened) => (Some(backend::telegram::links::answer(&opened)), None),
Err(e) => (None, Some(e)),
}
}
// The stickers a Telegram account can send, and sending one.
"listTelegramStickers" | "sendTelegramSticker" => {
let Some(buffer) = p_str_opt(params, "bufferId").and_then(|b| state.runtime.get_buffer(b)).filter(|b| b.account_id.starts_with("telegram:")) else {
return (None, Some(format!("{method} requires a Telegram \"bufferId\"")));
};
if method == "listTelegramStickers" {
return match backend::telegram::stickers::list(state, &buffer).await {
Ok(list) => (Some(list), None),
Err(e) => (None, Some(e)),
};
}
match backend::telegram::stickers::send(state, &buffer, p_str(params, "id", "")).await {
Ok(()) => (Some(ok_node()), None),
Err(e) => (None, Some(e)),
}
}
// An account's chat folders, which the window shows as categories on the account's page, and the changes it
// makes to them: see backend/telegram/folders.rs.
"telegramFolders" => match p_str_opt(params, "accountId") {
Some(account_id) if account_id.starts_with("telegram:") => (Some(backend::telegram::folders::current(account_id)), None),
_ => (None, Some("telegramFolders requires a Telegram \"accountId\"".to_string())),
},
"createTelegramFolder" | "fileTelegramChat" => {
let Some(buffer) = p_str_opt(params, "bufferId").and_then(|b| state.runtime.get_buffer(b)).filter(|b| b.account_id.starts_with("telegram:")) else {
return (None, Some(format!("{method} requires a Telegram \"bufferId\"")));
};
let done = if method == "createTelegramFolder" {
backend::telegram::folders::create(state, &buffer, p_str(params, "name", "")).await
} else {
let to = params.get("folderId").and_then(|v| v.as_i64()).and_then(|n| i32::try_from(n).ok());
backend::telegram::folders::file(state, &buffer, to).await
};
match done {
Ok(()) => (Some(ok_node()), None),
Err(e) => (None, Some(e)),
}
}
"renameTelegramFolder" | "deleteTelegramFolder" | "reorderTelegramFolders" => {
let Some(account_id) = p_str_opt(params, "accountId").filter(|a| a.starts_with("telegram:")) else {
return (None, Some(format!("{method} requires a Telegram \"accountId\"")));
};
let folder = params.get("folderId").and_then(|v| v.as_i64()).and_then(|n| i32::try_from(n).ok());
let done = match (method, folder) {
("renameTelegramFolder", Some(id)) => backend::telegram::folders::rename(state, account_id, id, p_str(params, "name", "")).await,
("deleteTelegramFolder", Some(id)) => backend::telegram::folders::delete(state, account_id, id).await,
("reorderTelegramFolders", _) => {
let order: Vec<i32> = params.get("order").and_then(|v| v.as_array()).map(|a| a.iter().filter_map(|n| n.as_i64().and_then(|n| i32::try_from(n).ok())).collect()).unwrap_or_default();
backend::telegram::folders::reorder(state, account_id, &order).await
}
_ => Err(format!("{method} requires a \"folderId\"")),
};
match done {
Ok(()) => (Some(ok_node()), None),
Err(e) => (None, Some(e)),
}
}
// Lets a Discord connection that is being held for having logged in too often through now.
"connectDiscordAnyway" => {
let Some(account_id) = p_str_opt(params, "accountId").filter(|a| a.starts_with("discord:")) else {
return (None, Some("connectDiscordAnyway requires a Discord \"accountId\"".to_string()));
};
if backend::discord::identify_budget::connect_anyway(account_id) {
(Some(ok_node()), None)
} else {
(None, Some("That account is not being held".to_string()))
}
}
// Pressing a button under a bot's message; answers what the bot said back, for the window to show.
"pressTelegramButton" => {
let Some(buffer) = p_str_opt(params, "bufferId").and_then(|b| state.runtime.get_buffer(b)).filter(|b| b.account_id.starts_with("telegram:")) else {
+39
View File
@@ -1544,6 +1544,24 @@ impl Runtime {
}
}
/// Says a conversation was active at `ts`, if that is later than what is known: for a service that tells when its
/// last message was without sending it.
pub fn raise_buffer_activity(&self, state: &AppState, buffer_id: &str, ts: i64) {
let updated = {
let mut buffers = self.buffers.lock().unwrap();
match buffers.get_mut(buffer_id) {
Some(b) if b.last_activity_ts < ts => {
b.last_activity_ts = ts;
Some(b.clone())
}
_ => None,
}
};
if let Some(b) = updated {
state.events.emit("bufferListChange", serde_json::to_value(&b).unwrap());
}
}
pub fn is_server_buffer(&self, buffer_id: &str) -> bool {
self.buffers
.lock()
@@ -1711,6 +1729,8 @@ impl Runtime {
topic: None,
read_only: None,
forum: false,
unread_seed: 0,
mentions_seed: 0,
};
buffers.insert(id, buffer.clone());
state.events.emit("bufferListChange", serde_json::to_value(&buffer).unwrap());
@@ -3279,6 +3299,25 @@ impl Runtime {
}
}
/// What the service counts as unread in a conversation, and how many of those mention this account: see
/// `Buffer::unread_seed`. Said to the windows when it changes, which set their own count to it.
pub fn set_buffer_unread(&self, state: &AppState, buffer_id: &str, unread: u32, mentions: u32) {
let updated = {
let mut buffers = self.buffers.lock().unwrap();
match buffers.get_mut(buffer_id) {
Some(b) if (b.unread_seed, b.mentions_seed) != (unread, mentions) => {
b.unread_seed = unread;
b.mentions_seed = mentions;
Some(b.clone())
}
_ => None,
}
};
if let Some(b) = updated {
state.events.emit("bufferListChange", serde_json::to_value(&b).unwrap());
}
}
/// A Matrix room's topic, kept against the room so a buffer made after it
/// was read still gets it.
pub fn set_matrix_room_topic(&self, state: &AppState, account_id: &str, room_id: &str, topic: &str) {
+74
View File
@@ -777,6 +777,44 @@ impl Store {
Ok(id)
}
/// The most recent message this buffer holds, and whether it is the account's own.
pub fn newest_msg(&self, buffer_id: &str) -> Result<Option<(String, bool)>> {
let conn = self.conn.lock().unwrap();
let row = conn
.query_row(
"SELECT msg_id, is_own FROM messages WHERE buffer_id = ?1 AND msg_id IS NOT NULL ORDER BY ts DESC LIMIT 1",
params![buffer_id],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? != 0)),
)
.optional()?;
Ok(row)
}
/// The ids of the account's own messages in these buffers from `since` (Unix seconds) on, newest first, at most
/// `limit`.
pub fn own_message_ids_since(&self, buffer_ids: &[String], since: i64, limit: usize) -> Result<Vec<String>> {
let mut found = Vec::new();
if buffer_ids.is_empty() {
return Ok(found);
}
let conn = self.conn.lock().unwrap();
for chunk in buffer_ids.chunks(200) {
let marks = vec!["?"; chunk.len()].join(",");
let sql = format!("SELECT msg_id FROM messages WHERE is_own = 1 AND ts >= ? AND msg_id IS NOT NULL AND buffer_id IN ({marks}) ORDER BY ts DESC LIMIT {limit}");
let mut args: Vec<&dyn rusqlite::ToSql> = vec![&since];
args.extend(chunk.iter().map(|b| b as &dyn rusqlite::ToSql));
let mut stmt = conn.prepare(&sql)?;
let rows = stmt.query_map(args.as_slice(), |row| row.get::<_, String>(0))?;
for id in rows {
found.push(id?);
}
if found.len() >= limit {
break;
}
}
Ok(found)
}
/// The most recent message this buffer holds, for asking a service what
/// has happened since.
pub fn newest_msg_id(&self, buffer_id: &str) -> Result<Option<String>> {
@@ -1853,6 +1891,42 @@ mod tests {
.expect("appending");
}
/// Like `append_at`, but the account's own.
fn append_own_at(s: &Store, buffer: &str, id: &str, ts: i64) {
s.append_message(buffer, id, "me", "hi", ts, false, false, "chat", None, &[], true, None, &[], &[], None, None, "channel", None, &[])
.expect("appending");
}
#[test]
fn finds_the_newest_message_and_whether_it_is_the_accounts_own() {
let (s, _dir) = store();
assert_eq!(s.newest_msg("b").unwrap(), None);
append_at(&s, "b", "m1", 100);
assert_eq!(s.newest_msg("b").unwrap(), Some(("m1".to_string(), false)));
append_own_at(&s, "b", "m2", 200);
assert_eq!(s.newest_msg("b").unwrap(), Some(("m2".to_string(), true)));
append_at(&s, "b", "m3", 300);
assert_eq!(s.newest_msg("b").unwrap(), Some(("m3".to_string(), false)));
}
#[test]
fn finds_what_the_account_has_said_in_a_set_of_rooms_since_a_time() {
let (s, _dir) = store();
let rooms = |names: &[&str]| names.iter().map(|n| n.to_string()).collect::<Vec<_>>();
assert!(s.own_message_ids_since(&[], 0, 5).unwrap().is_empty(), "no rooms is no message");
append_at(&s, "a", "m1", 500);
assert!(s.own_message_ids_since(&rooms(&["a"]), 0, 5).unwrap().is_empty(), "somebody else's message is not the account's");
append_own_at(&s, "b", "m2", 100);
append_own_at(&s, "b", "m3", 200);
assert_eq!(s.own_message_ids_since(&rooms(&["a", "b"]), 0, 5).unwrap(), vec!["m3", "m2"], "newest first");
assert!(s.own_message_ids_since(&rooms(&["a"]), 0, 5).unwrap().is_empty(), "in another room");
assert!(s.own_message_ids_since(&rooms(&["b"]), 201, 5).unwrap().is_empty(), "before the account joined");
assert_eq!(s.own_message_ids_since(&rooms(&["b"]), 0, 1).unwrap(), vec!["m3"], "no more than asked for");
// More rooms than one query holds.
let many: Vec<String> = (0..450).map(|i| format!("r{i}")).chain(std::iter::once("b".to_string())).collect();
assert_eq!(s.own_message_ids_since(&many, 0, 5).unwrap().len(), 2);
}
#[test]
fn finds_the_newest_message_to_mark_a_room_read_up_to() {
let (s, _dir) = store();