Author SHA1 Message Date
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
19 changed files with 1908 additions and 44 deletions
Generated
+2 -1
View File
@@ -4091,7 +4091,7 @@ dependencies = [
[[package]]
name = "nobilis"
version = "1.2.1"
version = "1.3.1"
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.1"
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.
+175 -30
View File
@@ -59,6 +59,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)]
@@ -86,6 +102,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,
@@ -173,6 +192,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,15 +207,47 @@ pub(super) struct Resume {
seq: u64,
}
/// 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
}
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();
}
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)) => {
@@ -229,10 +285,37 @@ 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());
@@ -319,7 +402,7 @@ 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(),
"compress": false,
"large_threshold": 50,
// Carried in IDENTIFY as well as pushed live, so a status set
@@ -347,9 +430,19 @@ 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(&mut stream) => 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.
@@ -411,14 +504,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"))
.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, d.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 +573,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 +595,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
@@ -526,30 +625,61 @@ 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 {
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) => 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);
// 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);
}
spawn_backfill(state.clone(), config.token.clone(), config.user_id.clone(), config.display_name.clone(), 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
@@ -819,6 +949,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;
@@ -838,6 +974,7 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
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 +999,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,
@@ -1360,6 +1497,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 +1507,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);
}
}
_ => {}
}
}
+5 -2
View File
@@ -394,6 +394,8 @@ 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());
// 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
@@ -551,13 +553,14 @@ pub(super) async fn register_guild_channels(state: &AppState, config: &DiscordAc
// 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);
if channel_map.contains_key(channel_id) {
state.runtime.set_buffer_read_only(state, &crate::model::buffer_id(&account_id, &name), read_only);
super::send_guard::set_permission(state, &account_id, &crate::model::buffer_id(&account_id, &name), 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::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
+66 -2
View File
@@ -164,12 +164,36 @@ 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.
/// 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 +201,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();
@@ -318,3 +350,35 @@ 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;
#[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"));
}
}
+7 -4
View File
@@ -125,7 +125,7 @@ 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 requests that belong to no account yet - the sign-in - or
@@ -133,14 +133,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
+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}");
}
}
+4
View File
@@ -49,10 +49,14 @@ 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 send_guard;
pub mod session_store;
pub mod http;
pub mod login;
pub mod media;
+3 -1
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.
+75 -2
View File
@@ -220,6 +220,7 @@ 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)?;
let mut reference = json!({
// 1 is Discord's own number for a forward, as against 0 for a reply.
"type": 1,
@@ -235,7 +236,11 @@ 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")?;
@@ -244,6 +249,7 @@ pub async fn forward_message(
let text = resp.text().await.unwrap_or_default();
bail!("{}", discord_error_text(status, &text, "forwarding that message"));
}
super::send_guard::note_sent(resp).await;
Ok(())
}
@@ -252,10 +258,14 @@ 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)?;
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));
let mut payload = json!({ "content": content });
put_reply(&mut payload, reply_to_id, reply_ping);
super::send_guard::stamp(&mut payload);
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"))
@@ -269,6 +279,7 @@ pub async fn send_message(state: &AppState, buffer_id: &str, token: &str, body:
let text = resp.text().await.unwrap_or_default();
bail!("{}", refusal_text(status, &text));
}
super::send_guard::note_sent(resp).await;
Ok(())
}
@@ -371,6 +382,7 @@ 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)?;
if let Some(p) = progress {
p.at(Phase::Preparing, 0, Meter::HOST);
}
@@ -385,12 +397,17 @@ 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));
let mut payload = json!({ "content": content });
put_reply(&mut payload, reply_to_id, reply_ping);
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);
}
@@ -420,6 +437,7 @@ pub async fn send_attachments(
let text = resp.text().await.unwrap_or_default();
bail!("{}", refusal_text(status, &text));
}
super::send_guard::note_sent(resp).await;
Ok(())
}
@@ -452,6 +470,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)?;
let bytes = tokio::fs::read(file_path)
.await
.with_context(|| format!("reading {}", file_path.display()))?;
@@ -459,7 +478,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,6 +488,7 @@ pub async fn send_voice_message(
"waveform": waveform,
}],
});
super::send_guard::stamp(&mut payload);
let form = reqwest::multipart::Form::new()
.text("payload_json", payload.to_string())
.part(
@@ -491,6 +511,7 @@ pub async fn send_voice_message(
let text = resp.text().await.unwrap_or_default();
bail!("{}", refusal_text(status, &text));
}
super::send_guard::note_sent(resp).await;
Ok(())
}
@@ -795,6 +816,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 +833,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
@@ -824,6 +869,10 @@ pub async fn ack_read(state: &AppState, buffer_id: &str, token: &str) -> Result<
.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(()) };
// 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"))
@@ -831,8 +880,10 @@ pub async fn ack_read(state: &AppState, buffer_id: &str, token: &str) -> Result<
.json(&json!({ "token": serde_json::Value::Null }))
)
.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 +891,28 @@ pub async fn ack_read(state: &AppState, buffer_id: &str, token: &str) -> Result<
Ok(())
}
/// 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
}
/// 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
+658
View File
@@ -0,0 +1,658 @@
//! What this client will say on Discord, and when.
//!
//! Discord judges an account by what it sends, and it judges the first thing said somewhere hardest. Accounts that
//! start talking in a server, or to a person, from a client that is not Discord's own have been flagged for spam - and
//! logged out of everywhere - on the strength of that one message, however ordinary it was; the same account's later
//! messages, in places it has already spoken, are left alone. So this client does not say the first word:
//!
//! - **A server** where the account has never spoken since it joined is locked, every channel in it. Whether it has
//! spoken is asked of what is stored here and then of Discord's own search (one request a server, the answer kept),
//! and the lock lifts as soon as there is a message of the account's there - including one sent from Discord's own
//! app while this one is running.
//! - **A conversation** with no messages in it at all is locked the same way, and lifts when one arrives.
//!
//! The lock is the daemon's, not the window's: a send to a locked place is refused here, whoever asks. The window shows
//! the reason where the message box would be, through the conversation's `read_only`.
//!
//! The rest is how what is sent goes out: a `nonce` on every message and an explicit `allowed_mentions`, as Discord's
//! clients send; a file checked against what the account may upload before any of it leaves; and a pause between a
//! message and a file, since a file sent straight after a message is what has been flagged too.
use super::*;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::sync::Mutex;
use std::time::{Duration, Instant};
/// Shown where the box would be in a server the account has not spoken in.
pub const FIRST_WORD: &str =
"Send your first message in this server from Discord's own app or website. Discord flags accounts that start talking in a server from another client.";
/// Shown while that is being found out.
pub const CHECKING: &str = "Checking whether you have spoken in this server before…";
/// Shown in a conversation with no messages in it.
pub const EMPTY_DM: &str =
"Start this conversation from Discord's own app or website. Discord flags accounts that send the first message to somebody from another client.";
/// Why a place is closed to this client, and nothing else.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Lock {
None,
Checking,
FirstWord,
EmptyDm,
}
impl Lock {
pub fn reason(self) -> Option<&'static str> {
match self {
Lock::None => None,
Lock::Checking => Some(CHECKING),
Lock::FirstWord => Some(FIRST_WORD),
Lock::EmptyDm => Some(EMPTY_DM),
}
}
}
/// What is known, per process. Plain data and plain methods so it can be tested without a connection.
#[derive(Default)]
pub struct Guard {
/// Account -> the servers it is known to have spoken in.
spoken: HashMap<String, HashSet<String>>,
/// Servers being looked into now.
checking: HashSet<(String, String)>,
/// Servers already looked into this run, whatever came of it: a server is not searched again every time it is read.
checked: HashSet<(String, String)>,
/// When the account joined each server (Unix seconds), where Discord said.
joined: HashMap<(String, String), i64>,
/// Each server's boost tier, which raises what may be uploaded there.
tiers: HashMap<(String, String), u8>,
/// Why a conversation is read-only for a reason that is not ours (no permission, Discord's own notices).
permission: HashMap<String, Option<&'static str>>,
/// Conversations that have no messages in them.
empty_dms: HashSet<String>,
/// When this account last sent a message, and whether it carried a file.
last_sent: HashMap<String, (Instant, bool)>,
/// The messages this client has sent, by id. They are the account's own, but they are not what the lock asks for:
/// it wants a first word from Discord's own client.
sent: HashSet<String>,
sent_order: std::collections::VecDeque<String>,
}
#[derive(Serialize, Deserialize, Default)]
struct Saved {
accounts: HashMap<String, Vec<String>>,
/// Ids of messages this client sent, oldest first.
#[serde(default)]
sent: Vec<String>,
}
/// How many sent messages are remembered: enough to outlast the rest of the history that could be mistaken for a
/// first word, and no more.
const REMEMBER_SENT: usize = 5000;
impl Guard {
pub fn is_spoken(&self, account: &str, guild: &str) -> bool {
self.spoken.get(account).is_some_and(|g| g.contains(guild))
}
/// What closes a place: `guild` is None for a conversation.
pub fn lock_of(&self, account: &str, guild: Option<&str>, buffer: &str) -> Lock {
match guild {
Some(guild) if self.is_spoken(account, guild) => Lock::None,
Some(guild) if self.checking.contains(&(account.to_string(), guild.to_string())) => Lock::Checking,
Some(_) => Lock::FirstWord,
None if self.empty_dms.contains(buffer) => Lock::EmptyDm,
None => Lock::None,
}
}
/// The reason a conversation is read-only: a permission the account lacks first, since that is the truer thing to
/// say, then the lock.
pub fn effective(&self, account: &str, guild: Option<&str>, buffer: &str) -> Option<&'static str> {
self.permission.get(buffer).copied().flatten().or_else(|| self.lock_of(account, guild, buffer).reason())
}
}
impl Guard {
pub fn note_sent(&mut self, id: &str) -> bool {
if !self.sent.insert(id.to_string()) {
return false;
}
self.sent_order.push_back(id.to_string());
while self.sent_order.len() > REMEMBER_SENT {
if let Some(old) = self.sent_order.pop_front() {
self.sent.remove(&old);
}
}
true
}
/// Whether any of these message ids is one this client did not send.
pub fn any_not_sent_here<'a>(&self, ids: impl IntoIterator<Item = &'a str>) -> bool {
ids.into_iter().any(|id| !self.sent.contains(id))
}
}
/// A message this client has just sent, from what Discord answered: remembered as sent here.
pub(super) async fn note_sent(resp: reqwest::Response) {
if let Ok(body) = resp.json::<Value>().await {
if let Some(id) = body["id"].as_str() {
with(|g| {
if g.note_sent(id) {
save(g);
}
});
}
}
}
/// The most a file may be where it is sent: what the account's Nitro allows, or what the server's boosts do, whichever
/// is more.
pub fn upload_limit(premium: u8, guild_tier: u8) -> u64 {
const MIB: u64 = 1024 * 1024;
let account = match premium {
2 => 500 * MIB,
1 | 3 => 50 * MIB,
_ => 10 * MIB,
};
let guild = match guild_tier {
3 => 100 * MIB,
2 => 50 * MIB,
_ => 10 * MIB,
};
account.max(guild)
}
fn file() -> std::path::PathBuf {
crate::default_data_dir().join("discord-spoken.json")
}
static GUARD: Mutex<Option<Guard>> = Mutex::new(None);
/// Runs `f` on the process's state, loading what was kept the first time.
fn with<R>(f: impl FnOnce(&mut Guard) -> R) -> R {
let mut slot = GUARD.lock().unwrap_or_else(|e| e.into_inner());
let guard = slot.get_or_insert_with(|| {
let saved: Saved = std::fs::read_to_string(file()).ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or_default();
Guard {
spoken: saved.accounts.into_iter().map(|(a, g)| (a, g.into_iter().collect())).collect(),
sent: saved.sent.iter().cloned().collect(),
sent_order: saved.sent.into_iter().collect(),
..Default::default()
}
});
f(guard)
}
fn save(guard: &Guard) {
let saved = Saved {
accounts: guard.spoken.iter().map(|(a, g)| (a.clone(), g.iter().cloned().collect())).collect(),
sent: guard.sent_order.iter().cloned().collect(),
};
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);
}
}
/// Says why a conversation is read-only for a reason of its own (or that it is not), and applies what the lock adds.
pub(super) fn set_permission(state: &AppState, account: &str, buffer: &str, reason: Option<&'static str>) {
let guild = state.runtime.get_discord_guild(buffer);
let effective = with(|g| {
g.permission.insert(buffer.to_string(), reason);
g.effective(account, guild.as_deref(), buffer)
});
state.runtime.set_buffer_read_only(state, buffer, effective);
}
/// Puts every conversation in a server back to what it should say now.
fn reapply_guild(state: &AppState, account: &str, guild: &str) {
for buffer in state.runtime.discord_buffers_in_guild(account, guild) {
let effective = with(|g| g.effective(account, Some(guild), &buffer));
state.runtime.set_buffer_read_only(state, &buffer, effective);
}
}
/// A server is being registered: note what is wanted of it, and start finding out whether the account has spoken in
/// it. Called before its channels are registered, so they begin locked rather than open.
pub(super) fn open_guild(state: &AppState, config: &DiscordAccountConfig, guild: &Value) {
let Some(guild_id) = guild["id"].as_str() else { return };
let account = config.account_id();
let key = (account.clone(), guild_id.to_string());
let joined = guild["joined_at"].as_str().and_then(|t| chrono::DateTime::parse_from_rfc3339(t).ok()).map(|t| t.timestamp());
let start = with(|g| {
if let Some(at) = joined {
g.joined.insert(key.clone(), at);
}
g.tiers.insert(key.clone(), guild["premium_tier"].as_u64().unwrap_or(0).min(3) as u8);
if g.is_spoken(&account, guild_id) || g.checking.contains(&key) || g.checked.contains(&key) {
return false;
}
g.checking.insert(key.clone());
true
});
if start {
spawn_check(state.clone(), config.clone(), guild_id.to_string());
}
}
/// An account has said something in a server: it has spoken there.
pub(super) fn spoke_in_guild(state: &AppState, account: &str, guild: &str) {
let changed = with(|g| {
let key = (account.to_string(), guild.to_string());
let was_checking = g.checking.remove(&key);
let new = g.spoken.entry(account.to_string()).or_default().insert(guild.to_string());
if new {
save(g);
}
new || was_checking
});
if changed {
reapply_guild(state, account, guild);
}
}
/// A server is gone (left, or removed from): speaking there again, if it is joined again, is a first word again.
pub(super) fn forget_guild(account: &str, guild: &str) {
with(|g| {
if let Some(set) = g.spoken.get_mut(account) {
if set.remove(guild) {
save(g);
}
}
let key = (account.to_string(), guild.to_string());
g.checking.remove(&key);
g.checked.remove(&key);
g.joined.remove(&key);
g.tiers.remove(&key);
});
}
/// An account is removed: nothing of it is kept.
pub fn forget_account(account: &str) {
with(|g| {
if g.spoken.remove(account).is_some() {
save(g);
}
g.checking.retain(|(a, _)| a != account);
g.checked.retain(|(a, _)| a != account);
g.joined.retain(|(a, _), _| a != account);
g.tiers.retain(|(a, _), _| a != account);
g.last_sent.remove(account);
});
}
/// A conversation as it was registered: empty ones are locked.
pub(super) fn note_dm(state: &AppState, account: &str, buffer: &str, channel: &Value) {
// Only a channel that says it has no messages: one that does not say is not known to be empty.
let empty = channel.get("last_message_id").is_some_and(Value::is_null) && state.store.newest_msg_id(buffer).ok().flatten().is_none();
with(|g| {
if empty {
g.empty_dms.insert(buffer.to_string());
} else {
g.empty_dms.remove(buffer);
}
});
let effective = with(|g| g.effective(account, None, buffer));
state.runtime.set_buffer_read_only(state, buffer, effective);
}
/// A message has arrived in a conversation, so it is not empty any more.
pub(super) fn dm_has_messages(state: &AppState, account: &str, buffer: &str) {
let was_empty = with(|g| g.empty_dms.remove(buffer));
if was_empty {
let effective = with(|g| g.effective(account, None, buffer));
state.runtime.set_buffer_read_only(state, buffer, effective);
}
}
/// A dispatch for a message was received in a place this connection knows: whatever it means for the locks.
pub(super) fn heard(state: &AppState, config: &DiscordAccountConfig, d: &Value, buffer: &str, is_dm: bool) {
let account = config.account_id();
if is_dm {
dm_has_messages(state, &account, buffer);
} else if d["author"]["id"].as_str() == Some(config.user_id.as_str()) {
if let Some(guild) = d["guild_id"].as_str() {
spoke_in_guild(state, &account, guild);
}
}
}
/// Refuses a send to a place the account has not spoken in.
pub(super) fn ensure_may_speak(state: &AppState, buffer: &str) -> Result<()> {
let account = state.runtime.get_buffer(buffer).map(|b| b.account_id).unwrap_or_default();
let guild = state.runtime.get_discord_guild(buffer);
match with(|g| g.lock_of(&account, guild.as_deref(), buffer)).reason() {
Some(reason) => bail!("{reason}"),
None => Ok(()),
}
}
/// Refuses a file Discord would refuse, before any of it is sent. What Discord does with one that is too big is cut
/// the upload off, which looks like nothing a client does; saying so first is the better answer.
pub(super) fn ensure_fits(state: &AppState, buffer: &str, files: &[(String, String, u64)]) -> Result<()> {
let account = state.runtime.get_buffer(buffer).map(|b| b.account_id).unwrap_or_default();
let guild = state.runtime.get_discord_guild(buffer);
let tier = guild.as_deref().and_then(|g| with(|s| s.tiers.get(&(account.clone(), g.to_string())).copied())).unwrap_or(0);
let limit = upload_limit(state.runtime.discord_premium(&account), tier);
for (_, name, len) in files {
if *len > limit {
bail!("{name} is {} and the most you can send here is {}", human(*len), human(limit));
}
}
Ok(())
}
fn human(bytes: u64) -> String {
const MIB: f64 = 1024.0 * 1024.0;
format!("{:.1} MB", bytes as f64 / MIB)
}
/// How long to wait before this account sends another message, given what it sent last: a file straight after a
/// message, or a message straight after a file, is closer together than anybody working a client sends them.
pub fn gap_needed(last: Option<(Duration, bool)>, attachment: bool) -> Duration {
const FILE_AFTER_MESSAGE: Duration = Duration::from_secs(5);
const MESSAGE_AFTER_FILE: Duration = Duration::from_secs(3);
let Some((since, last_was_file)) = last else { return Duration::ZERO };
let needed = match (attachment, last_was_file) {
(true, _) => FILE_AFTER_MESSAGE,
(false, true) => MESSAGE_AFTER_FILE,
(false, false) => Duration::ZERO,
};
needed.saturating_sub(since)
}
/// Waits out the pause, and takes this send's turn.
pub(super) async fn take_turn(account: &str, attachment: bool) {
let wait = with(|g| {
let last = g.last_sent.get(account).map(|(at, file)| (at.elapsed(), *file));
let wait = gap_needed(last, attachment);
g.last_sent.insert(account.to_string(), (Instant::now() + wait, attachment));
wait
});
if !wait.is_zero() {
tokio::time::sleep(wait).await;
}
}
/// A fresh message id, as Discord's clients make one: a snowflake from now, so it sorts with the messages around it
/// and cannot repeat.
pub fn nonce() -> String {
const EPOCH_MS: i64 = 1_420_070_400_000;
let now = chrono::Utc::now().timestamp_millis();
let low: u64 = rand::random::<u64>() & ((1 << 22) - 1);
((((now - EPOCH_MS).max(0) as u64) << 22) | low).to_string()
}
/// What every message carries beyond its words: the nonce, and who may be mentioned, said outright.
pub(super) fn stamp(payload: &mut Value) {
if payload.get("nonce").is_none() {
payload["nonce"] = Value::from(nonce());
}
if payload.get("allowed_mentions").is_none() {
let mut allowed = json!({ "parse": ["users", "roles", "everyone"] });
if payload.get("message_reference").is_some() {
allowed["replied_user"] = Value::Bool(true);
}
payload["allowed_mentions"] = allowed;
}
}
// --- finding out whether the account has spoken in a server ---------------------------------------------------------
/// One server at a time, a little apart: a search is the heaviest thing this asks of Discord.
static TURN: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
const SPACING: Duration = Duration::from_millis(2500);
const INDEXING_TRIES: usize = 6;
/// What Discord's search said of the account's messages in a server.
#[derive(Debug, PartialEq, Eq)]
pub enum Searched {
Spoke,
Never,
/// Still indexing, or asked to slow down: try again after this long.
Wait(Duration),
}
/// Reads a search answer. The count is Discord's own; a 202 or a 429 is not an answer. Messages that this client sent
/// are the account's but are not the first word the lock waits for: it is `Spoke` only if one of the hits was sent
/// from somewhere else, or if there are more results than the hits shown (which cannot all be ours).
pub fn read_search(status: u16, retry_after: Option<f64>, body: &Value, sent_here: &HashSet<String>) -> Result<Searched> {
match status {
200 => {
let total = body["total_results"].as_u64().unwrap_or(0);
let hits: Vec<&str> = body["messages"]
.as_array()
.into_iter()
.flatten()
.filter_map(|group| group.as_array()?.iter().find(|m| m["hit"].as_bool() == Some(true)).or_else(|| group.as_array()?.first()))
.filter_map(|m| m["id"].as_str())
.collect();
let elsewhere = hits.iter().any(|id| !sent_here.contains(*id));
Ok(if total > 0 && (elsewhere || total as usize > hits.len()) { Searched::Spoke } else { Searched::Never })
}
202 | 429 => {
let seconds = retry_after.or_else(|| body["retry_after"].as_f64()).unwrap_or(5.0).clamp(1.0, 60.0);
Ok(Searched::Wait(Duration::from_secs_f64(seconds)))
}
other => bail!("Discord answered {other} to the search"),
}
}
fn spawn_check(state: AppState, config: DiscordAccountConfig, guild_id: String) {
tokio::spawn(async move {
// The server's channels are registered a moment after this is started.
tokio::time::sleep(Duration::from_secs(2)).await;
let account = config.account_id();
let key = (account.clone(), guild_id.clone());
let since = with(|g| g.joined.get(&key).copied());
// What is kept here first: no request for what is already known, and no waiting behind the searches.
let buffers = state.runtime.discord_buffers_in_guild(&account, &guild_id);
let kept = state.store.own_message_ids_since(&buffers, since.unwrap_or(0), 200).unwrap_or_default();
if with(|g| g.any_not_sent_here(kept.iter().map(String::as_str))) {
spoke_in_guild(&state, &account, &guild_id);
return;
}
// Then Discord's own search, one server at a time.
let _turn = TURN.lock().await;
let mut spoke = false;
for attempt in 0..INDEXING_TRIES {
match search_own(&config, &guild_id, since).await {
Ok(Searched::Spoke) => {
spoke = true;
break;
}
Ok(Searched::Never) => break,
Ok(Searched::Wait(wait)) => {
tracing::debug!("discord[{account}]: searching {guild_id}, try {attempt}: waiting {}s", wait.as_secs());
tokio::time::sleep(wait).await;
}
Err(e) => {
// Not known either way. Left locked - which is the cautious answer - and tried again at the next start.
tracing::warn!("discord[{account}]: could not look for your messages in {guild_id}: {e:#}");
break;
}
}
}
if spoke {
spoke_in_guild(&state, &account, &guild_id);
} else {
with(|g| {
g.checking.remove(&key);
g.checked.insert(key.clone());
});
reapply_guild(&state, &account, &guild_id);
tracing::info!("discord[{account}]: no message of yours in {guild_id} since joining: locked until you send one from Discord's own client");
}
tokio::time::sleep(SPACING).await;
});
}
/// Asks Discord whether the account has any message in a server since it joined.
async fn search_own(config: &DiscordAccountConfig, guild_id: &str, since: Option<i64>) -> Result<Searched> {
let mut query = vec![("author_id".to_string(), config.user_id.clone()), ("include_nsfw".to_string(), "true".to_string())];
if let Some(at) = since {
query.push(("min_id".to_string(), super::search::snowflake_at(at)));
}
let resp = http_client_for(&config.token)
.get(format!("{API_BASE}/guilds/{guild_id}/messages/search"))
.query(&query)
.header("Authorization", &config.token)
.send()
.await
.context("searching for your messages")?;
let status = resp.status().as_u16();
let retry = resp.headers().get("retry-after").and_then(|v| v.to_str().ok()).and_then(|v| v.parse::<f64>().ok());
let body: Value = resp.json().await.unwrap_or(Value::Null);
let sent_here = with(|g| g.sent.clone());
read_search(status, retry, &body, &sent_here)
}
#[cfg(test)]
mod tests {
use super::*;
fn guard() -> Guard {
Guard::default()
}
#[test]
fn a_server_never_spoken_in_is_locked_and_one_spoken_in_is_not() {
let mut g = guard();
assert_eq!(g.lock_of("a", Some("g"), "b"), Lock::FirstWord);
g.checking.insert(("a".into(), "g".into()));
assert_eq!(g.lock_of("a", Some("g"), "b"), Lock::Checking);
g.spoken.entry("a".into()).or_default().insert("g".into());
assert_eq!(g.lock_of("a", Some("g"), "b"), Lock::None);
}
#[test]
fn speaking_in_one_account_does_not_open_another() {
let mut g = guard();
g.spoken.entry("a".into()).or_default().insert("g".into());
assert_eq!(g.lock_of("other", Some("g"), "b"), Lock::FirstWord);
assert_eq!(g.lock_of("a", Some("elsewhere"), "b"), Lock::FirstWord);
}
#[test]
fn an_empty_conversation_is_locked_and_one_with_messages_is_not() {
let mut g = guard();
assert_eq!(g.lock_of("a", None, "dm"), Lock::None);
g.empty_dms.insert("dm".into());
assert_eq!(g.lock_of("a", None, "dm"), Lock::EmptyDm);
}
#[test]
fn a_missing_permission_is_said_before_the_lock() {
let mut g = guard();
g.permission.insert("b".into(), Some("You may not send here."));
assert_eq!(g.effective("a", Some("g"), "b"), Some("You may not send here."));
g.permission.insert("b".into(), None);
assert_eq!(g.effective("a", Some("g"), "b"), Some(FIRST_WORD));
g.spoken.entry("a".into()).or_default().insert("g".into());
assert_eq!(g.effective("a", Some("g"), "b"), None);
}
#[test]
fn what_may_be_uploaded_is_the_more_of_the_account_and_the_server() {
const MIB: u64 = 1024 * 1024;
assert_eq!(upload_limit(0, 0), 10 * MIB);
assert_eq!(upload_limit(2, 0), 500 * MIB);
assert_eq!(upload_limit(3, 0), 50 * MIB);
assert_eq!(upload_limit(0, 2), 50 * MIB);
assert_eq!(upload_limit(0, 3), 100 * MIB);
assert_eq!(upload_limit(3, 3), 100 * MIB);
assert_eq!(upload_limit(2, 3), 500 * MIB);
}
#[test]
fn a_file_waits_after_a_message_and_a_message_after_a_file() {
let s = Duration::from_secs;
assert_eq!(gap_needed(None, true), Duration::ZERO);
assert_eq!(gap_needed(Some((s(1), false)), true), s(4));
assert_eq!(gap_needed(Some((s(9), false)), true), Duration::ZERO);
assert_eq!(gap_needed(Some((s(1), true)), false), s(2));
assert_eq!(gap_needed(Some((s(1), true)), true), s(4));
// Messages one after the other are talking, and are not slowed here.
assert_eq!(gap_needed(Some((Duration::ZERO, false)), false), Duration::ZERO);
}
#[test]
fn a_nonce_is_a_number_that_does_not_repeat() {
let a = nonce();
let b = nonce();
assert_ne!(a, b);
assert!(a.parse::<u64>().is_ok());
assert!(a.len() >= 18, "{a}");
}
#[test]
fn every_message_is_stamped_and_what_is_already_said_is_kept() {
let mut plain = json!({ "content": "hi" });
stamp(&mut plain);
assert!(plain["nonce"].as_str().is_some());
assert_eq!(plain["allowed_mentions"]["parse"], json!(["users", "roles", "everyone"]));
assert!(plain["allowed_mentions"].get("replied_user").is_none());
let mut reply = json!({ "content": "hi", "message_reference": { "message_id": "1" } });
stamp(&mut reply);
assert_eq!(reply["allowed_mentions"]["replied_user"], json!(true));
let mut quiet = json!({ "content": "hi", "message_reference": { "message_id": "1" }, "allowed_mentions": { "parse": [], "replied_user": false }, "nonce": "7" });
stamp(&mut quiet);
assert_eq!(quiet["nonce"], json!("7"));
assert_eq!(quiet["allowed_mentions"]["replied_user"], json!(false));
}
fn hit(id: &str) -> Value {
json!([{ "id": id, "hit": true }])
}
#[test]
fn a_search_answer_is_read_as_spoke_never_or_wait() {
let none = HashSet::new();
let one = json!({ "total_results": 1, "messages": [hit("9")] });
assert_eq!(read_search(200, None, &one, &none).unwrap(), Searched::Spoke);
assert_eq!(read_search(200, None, &json!({ "total_results": 0, "messages": [] }), &none).unwrap(), Searched::Never);
assert_eq!(read_search(202, None, &json!({ "retry_after": 7.0 }), &none).unwrap(), Searched::Wait(Duration::from_secs(7)));
assert_eq!(read_search(429, Some(2.0), &Value::Null, &none).unwrap(), Searched::Wait(Duration::from_secs(2)));
// Not an answer: neither "never" nor "spoke".
assert!(read_search(403, None, &Value::Null, &none).is_err());
assert!(read_search(500, None, &Value::Null, &none).is_err());
// A wait is never an hour.
assert_eq!(read_search(202, Some(9999.0), &Value::Null, &none).unwrap(), Searched::Wait(Duration::from_secs(60)));
}
#[test]
fn what_this_client_sent_is_not_the_first_word_the_lock_waits_for() {
let sent: HashSet<String> = ["9".to_string()].into();
let mine = json!({ "total_results": 1, "messages": [hit("9")] });
assert_eq!(read_search(200, None, &mine, &sent).unwrap(), Searched::Never);
let other = json!({ "total_results": 2, "messages": [hit("9"), hit("10")] });
assert_eq!(read_search(200, None, &other, &sent).unwrap(), Searched::Spoke);
// More results than hits shown: they cannot all be the one this client sent.
let many = json!({ "total_results": 40, "messages": [hit("9")] });
assert_eq!(read_search(200, None, &many, &sent).unwrap(), Searched::Spoke);
}
#[test]
fn sent_messages_are_remembered_a_bounded_number() {
let mut g = guard();
assert!(g.note_sent("1"));
assert!(!g.note_sent("1"), "twice is once");
assert!(g.any_not_sent_here(["1", "2"]));
assert!(!g.any_not_sent_here(["1"]));
for i in 0..(REMEMBER_SENT + 10) {
g.note_sent(&format!("m{i}"));
}
assert_eq!(g.sent.len(), REMEMBER_SENT);
assert!(!g.sent.contains("1") && !g.sent.contains("m0") && g.sent.contains(&format!("m{}", REMEMBER_SENT + 9)));
}
}
+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"));
}
}
+7 -1
View File
@@ -241,11 +241,16 @@ 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)?;
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")?;
@@ -254,6 +259,7 @@ pub async fn send(state: &AppState, buffer_id: &str, token: &str, sticker_id: &s
let text = resp.text().await.unwrap_or_default();
bail!("Discord API error {status}: {text}");
}
super::send_guard::note_sent(resp).await;
Ok(())
}
+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())
}
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);
}
}
+4
View File
@@ -406,6 +406,10 @@ 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");
}
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 {
+2
View File
@@ -29,6 +29,7 @@
//! - `keyboards` - a bot's inline buttons
//! - `bots` - a bot's command menu
//! - `topics` - a group with Topics as several conversations
//! - `gifs` - saved GIFs and the GIF search
//! - `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,6 +50,7 @@ pub mod admin;
pub mod bots;
pub mod connect;
pub mod dialogs;
pub mod gifs;
pub mod history;
pub mod instant;
pub mod keyboards;
+33
View File
@@ -756,6 +756,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
@@ -5771,6 +5775,35 @@ 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)),
}
}
// 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 {
+49
View File
@@ -777,6 +777,31 @@ impl Store {
Ok(id)
}
/// 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 +1878,30 @@ 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_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();