Compare commits
16
Commits
Generated
+2
-1
@@ -4091,7 +4091,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "nobilis"
|
||||
version = "1.2.1"
|
||||
version = "1.3.2"
|
||||
dependencies = [
|
||||
"aes-gcm",
|
||||
"anyhow",
|
||||
@@ -4116,6 +4116,7 @@ dependencies = [
|
||||
"http-body-util",
|
||||
"hyper",
|
||||
"hyper-util",
|
||||
"iana-time-zone",
|
||||
"image",
|
||||
"imagesize",
|
||||
"irc",
|
||||
|
||||
+2
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "nobilis"
|
||||
version = "1.2.1"
|
||||
version = "1.3.2"
|
||||
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.
|
||||
|
||||
@@ -31,7 +31,7 @@ pub async fn list_commands(state: &AppState, account_id: &str, buffer_id: &str,
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/application-commands/search"))
|
||||
.query(¶ms)
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("reading the commands")?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -243,7 +243,7 @@ pub async fn use_component(
|
||||
let resp = http_client_for(&cfg.token)
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/messages/{message_id}"))
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("finding out whose button that is")?;
|
||||
if !resp.status().is_success() {
|
||||
|
||||
@@ -135,7 +135,7 @@ pub async fn list(state: &AppState, account_id: &str, guild_id: &str) -> Result<
|
||||
.get(format!("{API_BASE}/guilds/{guild_id}/scheduled-events"))
|
||||
.query(&[("with_user_count", "true")])
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("reading the events")?;
|
||||
if resp.status().is_success() {
|
||||
@@ -153,7 +153,7 @@ pub async fn list(state: &AppState, account_id: &str, guild_id: &str) -> Result<
|
||||
.get(format!("{API_BASE}/guilds/{guild_id}/scheduled-events/{id}/users"))
|
||||
.query(&[("limit", "100")])
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
{
|
||||
if let Ok(users) = resp.json::<Value>().await {
|
||||
@@ -380,7 +380,7 @@ pub async fn share_targets(state: &AppState, account_id: &str, guild_id: &str) -
|
||||
.get(format!("{API_BASE}/users/{user_id}/profile"))
|
||||
.query(&[("with_mutual_guilds", "true")])
|
||||
.header("Authorization", &token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
{
|
||||
if let Ok(profile) = resp.json::<Value>().await {
|
||||
|
||||
@@ -125,7 +125,7 @@ async fn search_page(token: &str, channel_id: &str, sort_by: &str, archived: boo
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/threads/search"))
|
||||
.query(&query)
|
||||
.header("Authorization", token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("reading the forum")?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -216,7 +216,7 @@ pub async fn list_forum_posts(state: &AppState, account_id: &str, buffer_id: &st
|
||||
.get(format!("{API_BASE}/channels/{id}/messages"))
|
||||
.query(&[("limit", "1"), ("after", "0")])
|
||||
.header("Authorization", token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.ok()?;
|
||||
if !resp.status().is_success() {
|
||||
|
||||
+206
-40
@@ -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,
|
||||
@@ -132,7 +151,7 @@ pub fn spawn(state: AppState, config: DiscordAccountConfig) {
|
||||
state.runtime.insert_task_handle(&account_id, join_handle.abort_handle());
|
||||
}
|
||||
|
||||
pub(super) const RECONNECT_INITIAL_DELAY: Duration = Duration::from_secs(3);
|
||||
pub(super) const RECONNECT_INITIAL_DELAY: Duration = Duration::from_secs(4);
|
||||
|
||||
pub(super) const RECONNECT_MAX_DELAY: Duration = Duration::from_secs(60);
|
||||
|
||||
@@ -173,6 +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,59 @@ pub(super) struct Resume {
|
||||
seq: u64,
|
||||
}
|
||||
|
||||
/// The gateway capabilities this client declares in IDENTIFY, as a user client does (Dissent's `ningen` sends the same
|
||||
/// five of the seven it knows): notes fetched lazily, read states and guild settings in their versioned form, guild
|
||||
/// experiments as populations, and read states for things that are not channels. Each changes only the shape of a
|
||||
/// field READY carries and this client either ignores (`read_state`, notes, experiments) or reads in both shapes
|
||||
/// (`user_guild_settings`, see `mutes::note_settings`).
|
||||
///
|
||||
/// Not declared: DEDUPE_USER_OBJECTS (16) and PRIORITIZED_READY_PAYLOAD (32). They move users out of READY into a
|
||||
/// table by id and split the rest into a second event, and the friends list, the names of conversations and the
|
||||
/// presences here are built from the objects that would no longer be there. They need a real READY to be written
|
||||
/// against, not a guess at one.
|
||||
pub(super) const CAPABILITIES: u32 = (1 << 0) | (1 << 2) | (1 << 3) | (1 << 6) | (1 << 7);
|
||||
|
||||
/// What IDENTIFY says this client is: moho, on Linux, in this machine's language.
|
||||
fn identify_properties() -> Value {
|
||||
let mut properties = json!({ "os": "linux", "browser": "nobilis", "device": "nobilis" });
|
||||
if let Some(locale) = super::install_headers::locale() {
|
||||
properties["system_locale"] = Value::from(locale);
|
||||
}
|
||||
properties
|
||||
}
|
||||
|
||||
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 +297,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());
|
||||
@@ -299,6 +394,8 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
// Exposed so setAccountStatus can push a presence update onto this
|
||||
// connection rather than waiting for a reconnect to carry it.
|
||||
let account_id = config.account_id();
|
||||
// A connection starts with no subscriptions.
|
||||
super::presence::forget_member_subscriptions(&account_id);
|
||||
state.runtime.set_discord_gateway_sender(&account_id, out_tx.clone());
|
||||
struct ClearSenderOnDrop<'a>(&'a AppState, String);
|
||||
impl Drop for ClearSenderOnDrop<'_> {
|
||||
@@ -319,7 +416,8 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
"op": 2,
|
||||
"d": {
|
||||
"token": config.token,
|
||||
"properties": { "os": "linux", "browser": "nobilis", "device": "nobilis" },
|
||||
"properties": identify_properties(),
|
||||
"capabilities": CAPABILITIES,
|
||||
"compress": false,
|
||||
"large_threshold": 50,
|
||||
// Carried in IDENTIFY as well as pushed live, so a status set
|
||||
@@ -347,9 +445,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 +519,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 +588,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 +610,11 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
update_presence_in_rosters(state, user_id, p["status"].as_str().unwrap_or("offline"), Some(activity));
|
||||
}
|
||||
}
|
||||
// Conversations are the ones people notice a gap in,
|
||||
// and there are few enough of them to ask about
|
||||
// directly. Channels wait until opened - a few hundred
|
||||
// requests on every connect would be rate-limited and
|
||||
// mostly wasted.
|
||||
spawn_dm_catch_up(state.clone(), config.clone(), channel_map.clone());
|
||||
|
||||
state.runtime.set_conn_state(state, &account_id, ConnState::Connected, None);
|
||||
tracing::info!("discord[{account_id}]: ready as {username}");
|
||||
// Saved ready is not yet accepted by Discord: RESUMED says when it is.
|
||||
if !replaying {
|
||||
state.runtime.set_conn_state(state, &account_id, ConnState::Connected, None);
|
||||
}
|
||||
tracing::info!("discord[{account_id}]: ready as {username}{}", if replaying { " (from the last run)" } else { "" });
|
||||
|
||||
// For an account in relatively few guilds, READY's own
|
||||
// `guilds` array already carries full guild objects
|
||||
@@ -515,8 +629,11 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
// harmlessly on a guild it's already seen via either
|
||||
// path (channel_map dedup).
|
||||
if let Some(guilds) = d["guilds"].as_array() {
|
||||
for g in guilds {
|
||||
register_guild_channels(state, config, g, channel_map).await;
|
||||
// Our own membership of each, from READY's list of them, so that no guild is asked about.
|
||||
let merged = d["merged_members"].as_array();
|
||||
for (i, g) in guilds.iter().enumerate() {
|
||||
let with_me = guild_with_own_member(g, merged.and_then(|m| m.get(i)), &config.user_id);
|
||||
register_guild_channels(state, config, with_me.as_ref().unwrap_or(g), channel_map).await;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -526,30 +643,62 @@ 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_paced().await {
|
||||
Ok(resp) => {
|
||||
let body = resp.text().await.unwrap_or_default();
|
||||
match serde_json::from_str::<Vec<Value>>(&body) {
|
||||
Ok(channels) => Some(channels),
|
||||
Err(e) => {
|
||||
tracing::warn!("discord[{account_id}]: parsing /users/@me/channels: {e}");
|
||||
None
|
||||
}
|
||||
}
|
||||
Err(e) => tracing::warn!("discord[{account_id}]: parsing /users/@me/channels: {e}"),
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("discord[{account_id}]: fetching /users/@me/channels: {e}");
|
||||
None
|
||||
}
|
||||
}
|
||||
Err(e) => tracing::warn!("discord[{account_id}]: fetching /users/@me/channels: {e}"),
|
||||
};
|
||||
for ch in listed.iter().flatten() {
|
||||
note_newest(&mut newest_in, ch);
|
||||
if let Some(pair) = register_dm_channel(state, &account_id, ch, channel_map, &presences) {
|
||||
new_dm_buffers.push(pair);
|
||||
}
|
||||
}
|
||||
if !replaying {
|
||||
persist.dms(listed.unwrap_or_default());
|
||||
}
|
||||
|
||||
// The direct messages now exist, so the settings
|
||||
// read at the top of READY have somewhere to land.
|
||||
mutes::apply(state, &account_id);
|
||||
|
||||
spawn_backfill(state.clone(), config.token.clone(), config.user_id.clone(), config.display_name.clone(), new_dm_buffers);
|
||||
// Conversations are the ones people notice a gap in,
|
||||
// and there are few enough of them to ask about
|
||||
// directly - and only the ones Discord says have
|
||||
// moved on. Channels wait until opened: a few hundred
|
||||
// requests on every connect would be rate-limited and
|
||||
// mostly wasted.
|
||||
// (A resumed session is replayed what it missed.)
|
||||
if !replaying {
|
||||
spawn_dm_catch_up(state.clone(), config.clone(), channel_map.clone(), newest_in);
|
||||
}
|
||||
|
||||
// Nor is any conversation's history fetched here: it is read when it is opened (see getBacklog).
|
||||
let _ = new_dm_buffers;
|
||||
}
|
||||
"GUILD_CREATE" => {
|
||||
if let (false, Some(id)) = (replaying, d["id"].as_str()) {
|
||||
persist.guild(id, d);
|
||||
}
|
||||
register_guild_channels(state, config, d, channel_map).await;
|
||||
// The channels this guild's settings were about now
|
||||
// exist. READY carried the mutes before any of them
|
||||
@@ -716,6 +865,10 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
// own state rather than needing to guess.
|
||||
"MESSAGE_ACK" => {
|
||||
let Some(channel_id) = d["channel_id"].as_str() else { continue };
|
||||
// Remembered, so what is read here to the same point is not told to Discord a second time.
|
||||
if let Some(message_id) = d["message_id"].as_str() {
|
||||
super::send::acked_elsewhere(channel_id, message_id);
|
||||
}
|
||||
let Some((name, _)) = channel_map.get(channel_id) else { continue };
|
||||
state.events.emit(
|
||||
"bufferRead",
|
||||
@@ -819,6 +972,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 +997,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 +1022,7 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
let real_msg_id = d["id"].as_str().map(|s| s.to_string());
|
||||
let avatar_url = author_avatar_url(author);
|
||||
if author["system"].as_bool() == Some(true) && kind == "dm" {
|
||||
state.runtime.set_buffer_read_only(state, &model::buffer_id(&account_id, &buffer_name), Some(super::people::OFFICIAL_ONLY));
|
||||
super::send_guard::set_permission(state, &account_id, &model::buffer_id(&account_id, &buffer_name), Some(super::people::OFFICIAL_ONLY));
|
||||
}
|
||||
// Kept against their id as well as put on the message.
|
||||
// A direct call's voice states carry no member object,
|
||||
@@ -1133,12 +1293,10 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
mutes::note_settings(state, &account_id, d);
|
||||
}
|
||||
|
||||
"GUILD_MEMBER_ADD" | "GUILD_MEMBER_REMOVE" => {
|
||||
let Some(guild_id) = d["guild_id"].as_str() else { continue };
|
||||
if let Some(buffer_id) = state.runtime.discord_member_list_target(&account_id, guild_id) {
|
||||
request_member_list(state, &buffer_id);
|
||||
}
|
||||
}
|
||||
// Not a reason to ask for the list again: the list is subscribed to, and Discord sends what changes in
|
||||
// it. Asking at every join and leave of a busy server was a request for what was already coming.
|
||||
"GUILD_MEMBER_ADD" | "GUILD_MEMBER_REMOVE" => {}
|
||||
|
||||
|
||||
// A purge. Without this every message in it stayed on
|
||||
// screen, since the per-message deletes are not sent.
|
||||
@@ -1360,6 +1518,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 +1528,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);
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -87,7 +87,7 @@ pub async fn list_gifs(state: &AppState, query: &str) -> Result<Value> {
|
||||
let resp = http_client_for(&config.token)
|
||||
.get(url)
|
||||
.header("Authorization", &config.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("asking Discord for GIFs")?;
|
||||
if !resp.status().is_success() {
|
||||
|
||||
@@ -96,7 +96,7 @@ pub async fn member_verification(state: &AppState, account_id: &str, guild_id: &
|
||||
let resp = http_client_for(&cfg.token)
|
||||
.get(format!("{API_BASE}/guilds/{guild_id}/member-verification"))
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("asking Discord for this server's rules")?;
|
||||
let status = resp.status();
|
||||
@@ -201,7 +201,7 @@ pub(super) async fn resync_guild_once(state: &AppState, config: &DiscordAccountC
|
||||
http_client_for(&config.token)
|
||||
.get(format!("{API_BASE}/{path}"))
|
||||
.header("Authorization", &config.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.ok()?
|
||||
.json::<Value>()
|
||||
@@ -243,6 +243,32 @@ pub(super) async fn resync_guild_once(state: &AppState, config: &DiscordAccountC
|
||||
}
|
||||
}
|
||||
|
||||
/// This account's own member object out of READY's `merged_members`, which lists, for each guild in the order READY
|
||||
/// names them, the account's own membership of it (with a `user_id` where a member elsewhere has a `user`). Handed to
|
||||
/// `register_guild_channels` inside the guild's own payload, where it is looked for first, so that a guild whose
|
||||
/// payload leaves the account out costs no request of its own - one request a guild, at every fresh login, was the
|
||||
/// alternative.
|
||||
pub(super) fn guild_with_own_member(guild: &Value, merged: Option<&Value>, user_id: &str) -> Option<Value> {
|
||||
let has = |list: &Value| list.as_array().is_some_and(|l| l.iter().any(|m| m["user"]["id"].as_str() == Some(user_id)));
|
||||
if has(&guild["members"]) {
|
||||
return None;
|
||||
}
|
||||
let mut me = merged?
|
||||
.as_array()?
|
||||
.iter()
|
||||
.find(|m| m["user_id"].as_str() == Some(user_id) || m["user"]["id"].as_str() == Some(user_id))?
|
||||
.clone();
|
||||
if me["user"]["id"].is_null() {
|
||||
me["user"] = json!({ "id": user_id });
|
||||
}
|
||||
let mut guild = guild.clone();
|
||||
match guild["members"].as_array_mut() {
|
||||
Some(list) => list.push(me),
|
||||
None => guild["members"] = json!([me]),
|
||||
}
|
||||
Some(guild)
|
||||
}
|
||||
|
||||
/// This account's own membership of a guild.
|
||||
///
|
||||
/// From the payload when it is there, and asked for when it is not. The
|
||||
@@ -269,7 +295,7 @@ pub(super) async fn own_member(config: &DiscordAccountConfig, guild: &Value) ->
|
||||
let resp = http_client_for(&config.token)
|
||||
.get(format!("{API_BASE}/users/@me/guilds/{guild_id}/member"))
|
||||
.header("Authorization", &config.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.ok()?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -394,6 +420,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 +579,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
|
||||
@@ -704,7 +733,9 @@ pub(super) async fn register_guild_channels(state: &AppState, config: &DiscordAc
|
||||
},
|
||||
);
|
||||
|
||||
spawn_backfill(state.clone(), config.token.clone(), config.user_id.clone(), config.display_name.clone(), new_channels);
|
||||
// No history is fetched for these: a channel is read when it is opened (see getBacklog), which is the only time
|
||||
// anyone looks at it. Fetching the latest of every channel of every server when it is first seen was a request a
|
||||
// channel, at the moment of a fresh login, for conversations nobody was about to read.
|
||||
visible
|
||||
}
|
||||
|
||||
@@ -782,7 +813,7 @@ pub async fn list_threads(state: &AppState, account_id: &str, buffer_id: &str) -
|
||||
requests.push(format!("{API_BASE}/channels/{channel_id}/threads/archived/public?limit=25"));
|
||||
|
||||
for url in requests {
|
||||
let resp = match http_client_for(&cfg.token).get(&url).header("Authorization", &cfg.token).send().await {
|
||||
let resp = match http_client_for(&cfg.token).get(&url).header("Authorization", &cfg.token).send_paced().await {
|
||||
Ok(resp) if resp.status().is_success() => resp,
|
||||
// A channel with no archive, or one this account may not read the
|
||||
// archive of, is not a failure worth refusing the whole list for.
|
||||
@@ -953,7 +984,7 @@ pub async fn moderate_member(
|
||||
if let Some(reason) = reason.filter(|r| !r.trim().is_empty()) {
|
||||
request = request.header("X-Audit-Log-Reason", reason);
|
||||
}
|
||||
let resp = request.send().await.context(doing)?;
|
||||
let resp = request.send_paced().await.context(doing)?;
|
||||
if !resp.status().is_success() {
|
||||
let status = resp.status();
|
||||
let text = resp.text().await.unwrap_or_default();
|
||||
@@ -977,7 +1008,7 @@ pub async fn set_member_role(
|
||||
let http = http_client_for(&cfg.token);
|
||||
let request = if give { http.put(url).json(&json!({})) } else { http.delete(url) };
|
||||
let doing = if give { "giving them the role" } else { "taking the role away" };
|
||||
let resp = request.header("Authorization", &cfg.token).send().await.context(doing)?;
|
||||
let resp = request.header("Authorization", &cfg.token).send_paced().await.context(doing)?;
|
||||
if !resp.status().is_success() {
|
||||
let status = resp.status();
|
||||
let text = resp.text().await.unwrap_or_default();
|
||||
@@ -1011,7 +1042,7 @@ pub async fn open_thread(state: &AppState, account_id: &str, buffer_id: &str, th
|
||||
let resp = http
|
||||
.get(format!("{API_BASE}/channels/{thread_id}"))
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("opening the thread")?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -1255,3 +1286,30 @@ mod verification_tests {
|
||||
assert_eq!(bare[0]["response"], true);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod merged_member_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn a_guild_that_leaves_the_account_out_is_given_the_membership_ready_listed() {
|
||||
let guild = json!({ "id": "g", "members": [{ "user": { "id": "other" } }] });
|
||||
let merged = json!([{ "user_id": "me", "roles": ["r1"], "pending": true }, { "user_id": "x" }]);
|
||||
let with = guild_with_own_member(&guild, Some(&merged), "me").expect("added");
|
||||
let mine = with["members"].as_array().unwrap().iter().find(|m| m["user"]["id"] == "me").expect("the member is there");
|
||||
assert_eq!(mine["roles"], json!(["r1"]));
|
||||
assert!(member_is_pending(Some(mine)), "what the account's membership says survives");
|
||||
assert_eq!(with["members"].as_array().unwrap().len(), 2, "the others stay");
|
||||
// And it is what `own_member` would find first, without a request.
|
||||
assert!(guild_with_own_member(&with, Some(&merged), "me").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nothing_is_made_up_when_there_is_nothing_to_add() {
|
||||
let guild = json!({ "id": "g" });
|
||||
assert!(guild_with_own_member(&guild, None, "me").is_none());
|
||||
assert!(guild_with_own_member(&guild, Some(&json!([{ "user_id": "someone-else" }])), "me").is_none());
|
||||
let with = guild_with_own_member(&guild, Some(&json!([{ "user_id": "me" }])), "me").unwrap();
|
||||
assert_eq!(with["members"][0]["user"]["id"], "me");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,26 +28,20 @@ impl Drop for HistoryFetch<'_> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Fires off history backfill for a batch of newly-registered buffers as a
|
||||
/// background task, sequentially with a small delay between requests - a
|
||||
/// guild can easily have 30+ channels (confirmed live), and firing that
|
||||
/// many REST requests at once risks Discord's rate limiter; a one-time
|
||||
/// startup cost taking a few extra seconds in the background is a fine
|
||||
/// trade for not tripping it.
|
||||
pub(super) fn spawn_backfill(state: AppState, token: String, user_id: String, display_name: Option<String>, targets: Vec<(String, String)>) {
|
||||
if targets.is_empty() {
|
||||
return;
|
||||
}
|
||||
tokio::spawn(async move {
|
||||
for (buffer_id, channel_id) in targets {
|
||||
backfill_channel_history(&state, &token, &user_id, display_name.as_deref(), &buffer_id, &channel_id).await;
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
}
|
||||
});
|
||||
/// How many messages a conversation is first opened with. Few: the rest is a scroll away (`extend_history`), and a
|
||||
/// client that asks for fifty of every conversation it is shown is asking for what nobody reads.
|
||||
const FIRST_PAGE: &str = "30";
|
||||
|
||||
/// The time a message was sent, from its id: a snowflake's top bits are milliseconds since Discord's epoch. How a
|
||||
/// conversation's last activity is known without fetching anything of it.
|
||||
pub(super) fn snowflake_secs(id: &str) -> Option<i64> {
|
||||
const DISCORD_EPOCH_MS: u64 = 1_420_070_400_000;
|
||||
let id: u64 = id.parse().ok()?;
|
||||
Some(((id >> 22) + DISCORD_EPOCH_MS) as i64 / 1000)
|
||||
}
|
||||
|
||||
/// Backfills up to 50 recent messages for a channel/DM with no scrollback
|
||||
/// yet. Unlike IRC (which has no server-side history at all - nobilis only
|
||||
/// Backfills the most recent messages for a channel/DM with no scrollback
|
||||
/// yet - when it is first opened, and not before. Unlike IRC (which has no server-side history at all - nobilis only
|
||||
/// ever knows what it's personally seen live), Discord actually retains
|
||||
/// and exposes history, so starting every buffer blank here would look
|
||||
/// broken compared to Discord's own client. Seed-once, not sync: a buffer
|
||||
@@ -61,7 +55,7 @@ pub(super) fn spawn_backfill(state: AppState, token: String, user_id: String, di
|
||||
/// "notification" event, neither of which should fire for messages from
|
||||
/// potentially weeks ago just because this is the first time nobilis has
|
||||
/// seen the channel.
|
||||
pub(super) async fn backfill_channel_history(state: &AppState, token: &str, user_id: &str, own_display_name: Option<&str>, buffer_id: &str, channel_id: &str) {
|
||||
pub async fn backfill_channel_history(state: &AppState, token: &str, user_id: &str, own_display_name: Option<&str>, buffer_id: &str, channel_id: &str) {
|
||||
match state.store.has_messages(buffer_id) {
|
||||
Ok(true) => return,
|
||||
Ok(false) => {}
|
||||
@@ -73,9 +67,9 @@ pub(super) async fn backfill_channel_history(state: &AppState, token: &str, user
|
||||
|
||||
let resp = match http_client_for(token)
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
|
||||
.query(&[("limit", "50")])
|
||||
.query(&[("limit", FIRST_PAGE)])
|
||||
.header("Authorization", token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
{
|
||||
Ok(r) => r,
|
||||
@@ -127,7 +121,7 @@ pub async fn refill_history(state: &AppState, buffer_id: &str, limit: u32) -> Re
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
|
||||
.query(&[("limit", limit.clamp(1, 100).to_string().as_str())])
|
||||
.header("Authorization", &config.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("re-reading history")?;
|
||||
if resp.status() == reqwest::StatusCode::TOO_MANY_REQUESTS {
|
||||
@@ -164,12 +158,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 +195,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();
|
||||
@@ -218,7 +244,7 @@ pub async fn catch_up_channel(
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
|
||||
.query(&[("limit", "50"), ("after", after_id.as_str())])
|
||||
.header("Authorization", token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("fetching missed messages")?;
|
||||
if resp.status().is_success() {
|
||||
@@ -246,7 +272,7 @@ pub async fn extend_history(state: &AppState, token: &str, user_id: &str, own_di
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
|
||||
.query(&[("limit", "50"), ("before", before_id.as_str())])
|
||||
.header("Authorization", token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("fetching more history")?;
|
||||
if resp.status().is_success() {
|
||||
@@ -279,7 +305,7 @@ pub async fn load_newer(state: &AppState, account_id: &str, buffer_id: &str, mes
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
|
||||
.query(&[("limit", "50"), ("after", message_id)])
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("reading the rest of the channel")?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -299,7 +325,7 @@ pub async fn load_context(state: &AppState, account_id: &str, buffer_id: &str, m
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
|
||||
.query(&[("limit", "50"), ("around", message_id)])
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("reading that part of the channel")?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -318,3 +344,44 @@ pub async fn load_context(state: &AppState, account_id: &str, buffer_id: &str, m
|
||||
.map(|dt| dt.timestamp())
|
||||
.context("Discord did not return that message")
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod catch_up_tests {
|
||||
use super::*;
|
||||
use serde_json::json;
|
||||
|
||||
/// Discord's own documentation's example: 175928847299117063 was made at 2016-04-30 11:18:25.796 UTC.
|
||||
#[test]
|
||||
fn the_time_of_a_message_is_read_from_its_id() {
|
||||
assert_eq!(snowflake_secs("175928847299117063"), Some(1_462_015_105));
|
||||
assert_eq!(snowflake_secs("not a number"), None);
|
||||
// What `search::snowflake_at` makes of a time reads back as that time.
|
||||
assert_eq!(snowflake_secs(&super::search::snowflake_at(1_700_000_000)), Some(1_700_000_000));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_conversation_is_asked_about_only_when_discord_has_a_newer_message() {
|
||||
assert!(has_newer(Some("200"), Some("100")));
|
||||
assert!(!has_newer(Some("100"), Some("100")));
|
||||
assert!(!has_newer(Some("99"), Some("100")));
|
||||
// Nothing stored, but something said: a first sight is the backfill's, yet asking is harmless.
|
||||
assert!(has_newer(Some("5"), None));
|
||||
// An empty conversation has nothing to miss.
|
||||
assert!(!has_newer(None, Some("100")));
|
||||
// Ids that are not numbers are not compared.
|
||||
assert!(has_newer(Some("abc"), Some("100")));
|
||||
// Snowflakes outgrow 32 bits.
|
||||
assert!(has_newer(Some("1500000000000000001"), Some("1500000000000000000")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn what_a_channel_says_of_its_newest_message_is_noted_as_it_says_it() {
|
||||
let mut into = HashMap::new();
|
||||
note_newest(&mut into, &json!({ "id": "1", "last_message_id": "10" }));
|
||||
note_newest(&mut into, &json!({ "id": "2", "last_message_id": null }));
|
||||
note_newest(&mut into, &json!({ "id": "3" }));
|
||||
assert_eq!(into.get("1"), Some(&Some("10".to_string())));
|
||||
assert_eq!(into.get("2"), Some(&None));
|
||||
assert!(!into.contains_key("3"));
|
||||
}
|
||||
}
|
||||
|
||||
+125
-6
@@ -54,21 +54,98 @@ async fn pace() {
|
||||
}
|
||||
}
|
||||
|
||||
/// Where every read waits its turn: when Discord has said how long to hold off - a 429, or a bucket that ran out - no
|
||||
/// request of any kind goes out before then.
|
||||
static HOLD_UNTIL: std::sync::Mutex<Option<std::time::Instant>> = std::sync::Mutex::new(None);
|
||||
|
||||
/// What is added to a bucket's reset time, as in arikawa's limiter: Discord's own clock and ours do not agree to the
|
||||
/// millisecond, and arriving on the boundary is arriving early.
|
||||
const BUCKET_SLACK: Duration = Duration::from_millis(250);
|
||||
|
||||
/// How many times a read that was rate limited is made again, waiting as long as it was told to.
|
||||
const READ_RETRIES: usize = 2;
|
||||
|
||||
fn hold_for(wait: Duration) {
|
||||
let until = std::time::Instant::now() + wait;
|
||||
let mut held = HOLD_UNTIL.lock().unwrap_or_else(|e| e.into_inner());
|
||||
if held.is_none_or(|at| at < until) {
|
||||
*held = Some(until);
|
||||
}
|
||||
}
|
||||
|
||||
async fn wait_out_hold() {
|
||||
loop {
|
||||
let wait = {
|
||||
let held = HOLD_UNTIL.lock().unwrap_or_else(|e| e.into_inner());
|
||||
held.map(|at| at.saturating_duration_since(std::time::Instant::now())).unwrap_or_default()
|
||||
};
|
||||
if wait.is_zero() {
|
||||
return;
|
||||
}
|
||||
tokio::time::sleep(wait).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Notes what a response says about the bucket it counted against: with none left, nothing goes out until it resets.
|
||||
fn note_bucket(resp: &reqwest::Response) {
|
||||
let header = |name: &str| resp.headers().get(name).and_then(|v| v.to_str().ok());
|
||||
if header("x-ratelimit-remaining") == Some("0") {
|
||||
if let Some(reset) = header("x-ratelimit-reset-after").and_then(|v| v.parse::<f64>().ok()) {
|
||||
hold_for(Duration::from_secs_f64(reset.clamp(0.0, 30.0)) + BUCKET_SLACK);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Sending that behaves as a client of an API with rate limits should: waits out any hold Discord has put on this
|
||||
/// client, notes how much of a bucket is left and waits for it to refill rather than running into the limit, and when
|
||||
/// it is told to slow down anyway waits as long as it was told and asks again - instead of reading a refusal as an
|
||||
/// answer. Reads were sent plain, so a 429 was a silent "nothing found" and the next request followed straight on.
|
||||
pub(super) trait SendPaced {
|
||||
fn send_paced(self) -> impl std::future::Future<Output = reqwest::Result<reqwest::Response>> + Send;
|
||||
}
|
||||
|
||||
impl SendPaced for reqwest::RequestBuilder {
|
||||
async fn send_paced(self) -> reqwest::Result<reqwest::Response> {
|
||||
wait_out_hold().await;
|
||||
let mut pending = Some(self);
|
||||
for attempt in 0..=READ_RETRIES {
|
||||
let Some(current) = pending.take() else { break };
|
||||
// A body that cannot be copied is sent once.
|
||||
let again = current.try_clone();
|
||||
let resp = current.send().await?;
|
||||
note_bucket(&resp);
|
||||
if resp.status() != reqwest::StatusCode::TOO_MANY_REQUESTS || attempt == READ_RETRIES {
|
||||
return Ok(resp);
|
||||
}
|
||||
let Some(again) = again else { return Ok(resp) };
|
||||
let pause = retry_after(&resp);
|
||||
tracing::debug!("discord: rate limited on a read, waiting {}ms", pause.as_millis());
|
||||
hold_for(pause);
|
||||
wait_out_hold().await;
|
||||
pending = Some(again);
|
||||
}
|
||||
unreachable!("the last attempt returns")
|
||||
}
|
||||
}
|
||||
|
||||
/// Like `send_write`, for a request whose body cannot be copied - a file
|
||||
/// streamed from disk - and so is built again for every try.
|
||||
pub(super) async fn send_write_rebuilt<F>(mut build: F) -> Result<reqwest::Response>
|
||||
where
|
||||
F: FnMut() -> Result<reqwest::RequestBuilder>,
|
||||
{
|
||||
wait_out_hold().await;
|
||||
pace().await;
|
||||
for attempt in 0..=RATE_LIMIT_RETRIES {
|
||||
let resp = build()?.send().await.context("talking to Discord")?;
|
||||
note_bucket(&resp);
|
||||
if resp.status() != reqwest::StatusCode::TOO_MANY_REQUESTS || attempt == RATE_LIMIT_RETRIES {
|
||||
return Ok(resp);
|
||||
}
|
||||
let pause = retry_after(&resp);
|
||||
tracing::debug!("discord: rate limited, waiting {}ms", pause.as_millis());
|
||||
tokio::time::sleep(pause).await;
|
||||
hold_for(pause);
|
||||
wait_out_hold().await;
|
||||
}
|
||||
bail!("Discord kept rate-limiting that request")
|
||||
}
|
||||
@@ -85,6 +162,7 @@ where
|
||||
/// history quickly would be a slow client, and what Discord judges an account
|
||||
/// on is what it sends.
|
||||
pub(super) async fn send_write(request: reqwest::RequestBuilder) -> Result<reqwest::Response> {
|
||||
wait_out_hold().await;
|
||||
pace().await;
|
||||
|
||||
let mut pending = Some(request);
|
||||
@@ -94,13 +172,15 @@ pub(super) async fn send_write(request: reqwest::RequestBuilder) -> Result<reqwe
|
||||
// that cannot be cloned is one this cannot retry, and is sent once.
|
||||
let again = current.try_clone();
|
||||
let resp = current.send().await.context("talking to Discord")?;
|
||||
note_bucket(&resp);
|
||||
if resp.status() != reqwest::StatusCode::TOO_MANY_REQUESTS || attempt == RATE_LIMIT_RETRIES {
|
||||
return Ok(resp);
|
||||
}
|
||||
let Some(again) = again else { return Ok(resp) };
|
||||
let pause = retry_after(&resp);
|
||||
tracing::debug!("discord: rate limited, waiting {}ms", pause.as_millis());
|
||||
tokio::time::sleep(pause).await;
|
||||
hold_for(pause);
|
||||
wait_out_hold().await;
|
||||
pending = Some(again);
|
||||
}
|
||||
bail!("Discord kept rate-limiting that request")
|
||||
@@ -125,7 +205,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 +213,17 @@ pub(super) fn http_client_for(key: &str) -> reqwest::Client {
|
||||
/// when the add form's switch says so, or when any Discord account is.
|
||||
pub(super) fn anonymous_client() -> reqwest::Client {
|
||||
let router = crate::net::route::router();
|
||||
discord_client(router.routed(&crate::net::route::pending_key("discord")) || router.wanted(SHARED_KEY))
|
||||
discord_client(router.routed(&crate::net::route::pending_key("discord")) || router.wanted(SHARED_KEY), false)
|
||||
}
|
||||
|
||||
/// Marked while any Discord account is routed.
|
||||
pub(super) const SHARED_KEY: &str = "discord:any-routed";
|
||||
|
||||
fn discord_client(routed: bool) -> reqwest::Client {
|
||||
crate::net::route::router().client_if("discord", routed, |builder| {
|
||||
/// `api` clients say which install, language and timezone they are (see `install_headers`); the others - the CDN,
|
||||
/// the sign-in before there is an account - do not, since nothing there asks.
|
||||
fn discord_client(routed: bool, api: bool) -> reqwest::Client {
|
||||
crate::net::route::router().client_if(if api { "discord-api" } else { "discord" }, routed, |builder| {
|
||||
let builder = if api { builder.default_headers(super::install_headers::headers().clone()) } else { builder };
|
||||
// Pinned to HTTP/1.1. Enabling reqwest's http2 feature (which the
|
||||
// file-upload path needs - see upload::http_client) would otherwise
|
||||
// let every client here negotiate h2 as a side effect, changing the
|
||||
@@ -282,4 +365,40 @@ mod error_tests {
|
||||
let text = discord_error_text(reqwest::StatusCode::BAD_GATEWAY, junk, "adding a friend");
|
||||
assert!(text.contains("502"), "got {text}");
|
||||
}
|
||||
|
||||
/// What a client of an API with rate limits does: told to slow down, it waits as long as it was told and asks
|
||||
/// again; told a bucket is empty, it lets it refill before anything else goes out. Asked of a local listener.
|
||||
#[tokio::test]
|
||||
async fn a_read_waits_out_a_rate_limit_and_a_bucket_that_has_run_dry() {
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let url = format!("http://{}/", listener.local_addr().unwrap());
|
||||
// 1: refused with a one-second wait. 2: answered, saying the bucket is empty for one second. 3: answered.
|
||||
let server = tokio::spawn(async move {
|
||||
let mut seen = Vec::new();
|
||||
for reply in [
|
||||
"HTTP/1.1 429 Too Many Requests\r\nRetry-After: 1\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
|
||||
"HTTP/1.1 200 OK\r\nX-RateLimit-Remaining: 0\r\nX-RateLimit-Reset-After: 1\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
|
||||
"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
|
||||
] {
|
||||
let (mut socket, _) = listener.accept().await.unwrap();
|
||||
let mut buf = vec![0u8; 4096];
|
||||
let _ = socket.read(&mut buf).await.unwrap();
|
||||
seen.push(std::time::Instant::now());
|
||||
socket.write_all(reply.as_bytes()).await.unwrap();
|
||||
}
|
||||
seen
|
||||
});
|
||||
let client = reqwest::Client::new();
|
||||
let started = std::time::Instant::now();
|
||||
let first = client.get(&url).send_paced().await.unwrap();
|
||||
assert_eq!(first.status(), 200, "the refusal was waited out and the read made again");
|
||||
assert!(started.elapsed() >= Duration::from_millis(950), "it waited the second it was told to: {:?}", started.elapsed());
|
||||
let second_started = std::time::Instant::now();
|
||||
let second = client.get(&url).send_paced().await.unwrap();
|
||||
assert_eq!(second.status(), 200);
|
||||
assert!(second_started.elapsed() >= Duration::from_millis(950), "the empty bucket was let refill first: {:?}", second_started.elapsed());
|
||||
let seen = server.await.unwrap();
|
||||
assert_eq!(seen.len(), 3);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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]);
|
||||
}
|
||||
}
|
||||
@@ -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}");
|
||||
}
|
||||
}
|
||||
@@ -249,7 +249,7 @@ pub(super) async fn run_resign(state: &AppState, buffer_id: &str, mut stale: Vec
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
|
||||
.query(&[("limit", "50"), ("around", anchor.as_str())])
|
||||
.header("Authorization", &config.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("fetching a page to re-sign")?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -358,7 +358,7 @@ pub async fn refresh_attachments(state: &AppState, buffer_id: &str, message_id:
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
|
||||
.query(&[("limit", "1"), ("around", message_id)])
|
||||
.header("Authorization", &config.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("re-fetching the message")?;
|
||||
if !resp.status().is_success() {
|
||||
|
||||
@@ -23,7 +23,7 @@ pub async fn list_pinned(state: &AppState, account_id: &str, buffer_id: &str) ->
|
||||
let resp = http_client_for(&cfg.token)
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/pins"))
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("reading the pins")?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -96,7 +96,7 @@ pub async fn set_pinned(state: &AppState, account_id: &str, buffer_id: &str, mes
|
||||
let http = http_client_for(&cfg.token);
|
||||
let request = if pinned { http.put(url) } else { http.delete(url) };
|
||||
let doing = if pinned { "pinning the message" } else { "unpinning the message" };
|
||||
let resp = request.header("Authorization", &cfg.token).send().await.context(doing)?;
|
||||
let resp = request.header("Authorization", &cfg.token).send_paced().await.context(doing)?;
|
||||
if !resp.status().is_success() {
|
||||
let status = resp.status();
|
||||
let text = resp.text().await.unwrap_or_default();
|
||||
@@ -601,7 +601,7 @@ pub async fn name_forward(state: &AppState, account_id: &str, buffer_id: &str, m
|
||||
let resp = http_client_for(&cfg.token)
|
||||
.get(format!("{API_BASE}/channels/{source_channel}/messages?limit=1&around={source_message}"))
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await;
|
||||
// A source this account cannot see is the ordinary case for a forward
|
||||
// out of somebody else's server, and not worth a word to anybody.
|
||||
@@ -1680,7 +1680,7 @@ pub async fn list_reactors(state: &AppState, account_id: &str, buffer_id: &str,
|
||||
let resp = http_client_for(&cfg.token)
|
||||
.get(url)
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("reading who reacted")?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -1743,7 +1743,7 @@ pub async fn reread_embeds(state: &AppState, buffer_id: &str) -> Result<usize> {
|
||||
.get(format!("{API_BASE}/channels/{channel_id}/messages"))
|
||||
.query(&[("limit", "50")])
|
||||
.header("Authorization", &config.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("re-reading the conversation")?;
|
||||
if !resp.status().is_success() {
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -27,7 +27,9 @@ pub(super) fn register_dm_channel(
|
||||
state.runtime.set_discord_channel(state, &buf.id, channel_id);
|
||||
// Discord's own notices come from an account flagged `system`, and nothing
|
||||
// sent back to it goes anywhere.
|
||||
state.runtime.set_buffer_read_only(state, &buf.id, is_system_dm(ch).then_some(OFFICIAL_ONLY));
|
||||
super::send_guard::set_permission(state, account_id, &buf.id, is_system_dm(ch).then_some(OFFICIAL_ONLY));
|
||||
// A conversation with nothing in it yet is not started from here: see send_guard.
|
||||
super::send_guard::note_dm(state, account_id, &buf.id, ch);
|
||||
// Its notices were saved before their cards could be read whole, and saved
|
||||
// history is not fetched again, so they are read once a run, a little after
|
||||
// connecting - one request, for the latest page.
|
||||
@@ -40,6 +42,11 @@ pub(super) fn register_dm_channel(
|
||||
}
|
||||
});
|
||||
}
|
||||
// When it was last spoken in, from the id of its last message - which is all that ordering the list by activity
|
||||
// needs, so nothing is fetched for it.
|
||||
if let Some(at) = ch["last_message_id"].as_str().and_then(super::history::snowflake_secs) {
|
||||
state.runtime.raise_buffer_activity(state, &buf.id, at);
|
||||
}
|
||||
if let Some(avatar) = dm_avatar_url(ch) {
|
||||
state.runtime.set_buffer_avatar(state, &buf.id, &avatar);
|
||||
}
|
||||
@@ -166,7 +173,7 @@ pub async fn profile(state: &AppState, account_id: &str, buffer_id: &str, user_i
|
||||
}
|
||||
}
|
||||
|
||||
let response = http_client_for(&config.token).get(&url).header("Authorization", &config.token).send().await;
|
||||
let response = http_client_for(&config.token).get(&url).header("Authorization", &config.token).send_paced().await;
|
||||
let Ok(body) = response else { return profile };
|
||||
let Ok(body) = body.json::<Value>().await else { return profile };
|
||||
|
||||
@@ -295,7 +302,7 @@ pub async fn adopt_dm_channel(state: &AppState, account_id: &str, channel_id: &s
|
||||
let resp = http_client_for(&config.token)
|
||||
.get(format!("{API_BASE}/channels/{channel_id}"))
|
||||
.header("Authorization", &config.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("asking about that conversation")?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -379,7 +386,7 @@ pub async fn answer_friend_request(state: &AppState, account_id: &str, user_id:
|
||||
let url = format!("{API_BASE}/users/@me/relationships/{user_id}");
|
||||
let request = if accept { http.put(url).json(&json!({})) } else { http.delete(url) };
|
||||
let doing = if accept { "accepting the friend request" } else { "declining the friend request" };
|
||||
let resp = request.header("Authorization", &cfg.token).send().await.context(doing)?;
|
||||
let resp = request.header("Authorization", &cfg.token).send_paced().await.context(doing)?;
|
||||
if !resp.status().is_success() {
|
||||
let status = resp.status();
|
||||
let text = resp.text().await.unwrap_or_default();
|
||||
|
||||
@@ -117,6 +117,46 @@ fn settings_body(status: &str, text: Option<&str>) -> serde_json::Value {
|
||||
/// does when you open a channel, which is also why the roster only exists for
|
||||
/// channels somebody is actually looking at.
|
||||
pub fn request_member_list(state: &AppState, buffer_id: &str) -> bool {
|
||||
subscribe_member_list(state, buffer_id, false)
|
||||
}
|
||||
|
||||
/// The window the list is kept in no longer matches what Discord holds, so it is asked for again - but not more often
|
||||
/// than every few seconds, which is the most a roster that keeps falling out of step deserves.
|
||||
pub(super) fn resync_member_list(state: &AppState, buffer_id: &str) -> bool {
|
||||
subscribe_member_list(state, buffer_id, true)
|
||||
}
|
||||
|
||||
/// What each server has been subscribed to this session: the channel, and when.
|
||||
type Subscribed = HashMap<(String, String), (String, std::time::Instant)>;
|
||||
static SUBSCRIBED: std::sync::Mutex<Option<Subscribed>> = std::sync::Mutex::new(None);
|
||||
|
||||
/// The least time between two subscriptions to one server's list that are not a change of channel.
|
||||
const RESYNC_EVERY: std::time::Duration = std::time::Duration::from_secs(10);
|
||||
|
||||
/// Whether a subscription to this channel's list is to be sent: not if it already was, this session, and not a
|
||||
/// repeat of one sent a moment ago. A connection is subscribed once - as Dissent's is, once a server - and a new one
|
||||
/// starts again (`forget_member_subscriptions`).
|
||||
fn subscription_due(account: &str, guild: &str, channel: &str, resync: bool, now: std::time::Instant) -> bool {
|
||||
let mut slot = SUBSCRIBED.lock().unwrap_or_else(|e| e.into_inner());
|
||||
let map = slot.get_or_insert_with(HashMap::new);
|
||||
let key = (account.to_string(), guild.to_string());
|
||||
match map.get(&key) {
|
||||
Some((was, at)) if was == channel && (!resync || now.duration_since(*at) < RESYNC_EVERY) => false,
|
||||
_ => {
|
||||
map.insert(key, (channel.to_string(), now));
|
||||
true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A new connection holds none of what the last one subscribed to.
|
||||
pub(super) fn forget_member_subscriptions(account: &str) {
|
||||
if let Some(map) = SUBSCRIBED.lock().unwrap_or_else(|e| e.into_inner()).as_mut() {
|
||||
map.retain(|(a, _), _| a != account);
|
||||
}
|
||||
}
|
||||
|
||||
fn subscribe_member_list(state: &AppState, buffer_id: &str, resync: bool) -> bool {
|
||||
let Some(buffer) = state.runtime.get_buffer(buffer_id) else { return false };
|
||||
let account_id = buffer.account_id;
|
||||
let Some(sender) = state.runtime.discord_gateway_sender(&account_id) else { return false };
|
||||
@@ -131,6 +171,11 @@ pub fn request_member_list(state: &AppState, buffer_id: &str) -> bool {
|
||||
// channel, so remember which buffer this was for.
|
||||
state.runtime.set_discord_member_list_target(&account_id, &guild_id, buffer_id);
|
||||
|
||||
// Asked for once. Whoever asked is told it is being sent: the roster arrives, or is already arriving.
|
||||
if !subscription_due(&account_id, &guild_id, &channel_id, resync, std::time::Instant::now()) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// Ranges are 100-member windows; one covers any channel we would show.
|
||||
sender
|
||||
.send(
|
||||
@@ -370,7 +415,7 @@ pub(super) fn update_member_list(state: &AppState, buffer_id: &str, d: &Value) {
|
||||
Applied::Stale => {
|
||||
tracing::debug!("discord: member window for {buffer_id} is out of step; resyncing");
|
||||
state.runtime.set_discord_member_window(buffer_id, Vec::new());
|
||||
request_member_list(state, buffer_id);
|
||||
resync_member_list(state, buffer_id);
|
||||
return;
|
||||
}
|
||||
}
|
||||
@@ -629,3 +674,30 @@ mod status_tests {
|
||||
assert_eq!(settings_body("idle", None)["status"], "idle");
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod subscription_tests {
|
||||
use super::*;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
#[test]
|
||||
fn a_list_is_subscribed_to_once_a_connection_and_again_for_another_channel() {
|
||||
let t = Instant::now();
|
||||
assert!(subscription_due("a1", "g", "c1", false, t));
|
||||
assert!(!subscription_due("a1", "g", "c1", false, t + Duration::from_secs(1)), "the same again is nothing");
|
||||
assert!(!subscription_due("a1", "g", "c1", false, t + Duration::from_secs(600)), "however long after");
|
||||
assert!(subscription_due("a1", "g", "c2", false, t + Duration::from_secs(2)), "another channel is a change");
|
||||
assert!(subscription_due("a1", "g2", "c1", false, t), "another server is its own");
|
||||
forget_member_subscriptions("a1");
|
||||
assert!(subscription_due("a1", "g", "c2", false, t), "a new connection starts again");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_resync_is_not_sent_again_within_a_few_seconds() {
|
||||
let t = Instant::now();
|
||||
assert!(subscription_due("a2", "g", "c1", false, t));
|
||||
assert!(!subscription_due("a2", "g", "c1", true, t + Duration::from_secs(3)));
|
||||
assert!(subscription_due("a2", "g", "c1", true, t + Duration::from_secs(11)));
|
||||
assert!(!subscription_due("a2", "g", "c1", true, t + Duration::from_secs(12)));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -103,7 +103,7 @@ pub async fn search_messages(
|
||||
.get(url)
|
||||
.query(&query)
|
||||
.header("Authorization", &cfg.token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("searching")?;
|
||||
// Discord answers a search of a server it is still indexing with 202 and
|
||||
|
||||
+115
-3
@@ -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,15 @@ 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));
|
||||
super::send_guard::check_text(&content, state.runtime.discord_premium(&account_id), false)?;
|
||||
let mut payload = json!({ "content": content });
|
||||
put_reply(&mut payload, reply_to_id, reply_ping);
|
||||
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 +280,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 +383,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 +398,18 @@ pub async fn send_attachments(
|
||||
total += len;
|
||||
files.push((path.clone(), name, len));
|
||||
}
|
||||
// What Discord would refuse is refused here, before any of it is sent.
|
||||
super::send_guard::ensure_fits(state, buffer_id, &files)?;
|
||||
// A caption is a message like any other, so a name typed in one is a
|
||||
// mention like any other.
|
||||
let account_id = state.runtime.get_buffer(buffer_id).map(|b| b.account_id).unwrap_or_default();
|
||||
let content = resolve_outgoing_mentions(body, mention_candidates(state, &account_id, buffer_id));
|
||||
super::send_guard::check_text(&content, state.runtime.discord_premium(&account_id), true)?;
|
||||
let mut payload = json!({ "content": content });
|
||||
put_reply(&mut payload, reply_to_id, reply_ping);
|
||||
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 +439,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 +472,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 +480,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 +490,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 +513,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 +818,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 +835,24 @@ pub async fn send_typing(state: &AppState, buffer_id: &str, token: &str) -> Resu
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Shortest gap between two typing notices to the same channel. Short of the
|
||||
/// ten seconds an indicator lasts, so it never lapses mid-sentence.
|
||||
const TYPING_EVERY: std::time::Duration = std::time::Duration::from_secs(8);
|
||||
|
||||
/// Whether a typing notice to this channel is due, noting it if so.
|
||||
fn typing_due(channel_id: &str) -> bool {
|
||||
static LAST: std::sync::Mutex<Option<HashMap<String, std::time::Instant>>> = std::sync::Mutex::new(None);
|
||||
let mut last = LAST.lock().unwrap();
|
||||
let last = last.get_or_insert_with(HashMap::new);
|
||||
let now = std::time::Instant::now();
|
||||
if last.get(channel_id).is_some_and(|at| now.duration_since(*at) < TYPING_EVERY) {
|
||||
return false;
|
||||
}
|
||||
last.retain(|_, at| now.duration_since(*at) < TYPING_EVERY);
|
||||
last.insert(channel_id.to_string(), now);
|
||||
true
|
||||
}
|
||||
|
||||
/// Tells Discord this conversation has been read up to its newest message.
|
||||
///
|
||||
/// Unread was purely local before this, so reading a channel here did not
|
||||
@@ -823,7 +870,16 @@ pub async fn ack_read(state: &AppState, buffer_id: &str, token: &str) -> Result<
|
||||
.runtime
|
||||
.get_discord_channel(buffer_id)
|
||||
.ok_or_else(|| anyhow!("no known Discord channel for this buffer"))?;
|
||||
let Some(msg_id) = state.store.newest_msg_id(buffer_id)? else { return Ok(()) };
|
||||
let Some((msg_id, own)) = state.store.newest_msg(buffer_id)? else { return Ok(()) };
|
||||
// Nothing to tell Discord about a conversation whose last word is the account's own: sending a message reads
|
||||
// everything before it, on every device, and Discord's own clients do not ack what they sent.
|
||||
if own {
|
||||
return Ok(());
|
||||
}
|
||||
// Reading a channel again with nothing new in it says nothing new.
|
||||
if !ack_due(&channel_id, &msg_id) {
|
||||
return Ok(());
|
||||
}
|
||||
let resp = send_write(
|
||||
http_client_for(token)
|
||||
.post(format!("{API_BASE}/channels/{channel_id}/messages/{msg_id}/ack"))
|
||||
@@ -831,8 +887,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 +898,44 @@ 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
|
||||
}
|
||||
|
||||
/// Another client of the account read this far: no need to say it again from here.
|
||||
pub(super) fn acked_elsewhere(channel_id: &str, msg_id: &str) {
|
||||
let mut acked = ACKED.lock().unwrap();
|
||||
// Never backwards: an ack from a device that is behind does not make this one's later ack redundant.
|
||||
let newer = match acked.get_or_insert_with(HashMap::new).get(channel_id) {
|
||||
Some(known) => match (known.parse::<u64>(), msg_id.parse::<u64>()) {
|
||||
(Ok(k), Ok(m)) => m > k,
|
||||
_ => true,
|
||||
},
|
||||
None => true,
|
||||
};
|
||||
if newer {
|
||||
acked.get_or_insert_with(HashMap::new).insert(channel_id.to_string(), msg_id.to_string());
|
||||
}
|
||||
}
|
||||
|
||||
/// A refused ack must be tried again, not remembered as done.
|
||||
fn forget_ack(channel_id: &str) {
|
||||
if let Some(acked) = ACKED.lock().unwrap().as_mut() {
|
||||
acked.remove(channel_id);
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn delete_message(state: &AppState, buffer_id: &str, token: &str, msg_id: &str) -> Result<()> {
|
||||
let channel_id = state
|
||||
.runtime
|
||||
@@ -949,3 +1045,19 @@ mod refusal_tests {
|
||||
assert_eq!(describe_seconds(45), "45 seconds");
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod ack_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn an_ack_made_elsewhere_is_not_made_again_and_never_moves_backwards() {
|
||||
// Own ids for this test: the table is shared by the process.
|
||||
let channel = "ack-test-channel";
|
||||
acked_elsewhere(channel, "200");
|
||||
assert!(!ack_due(channel, "200"), "read this far on another device");
|
||||
assert!(ack_due(channel, "300"), "something newer is still to be told");
|
||||
acked_elsewhere(channel, "250");
|
||||
assert!(!ack_due(channel, "300"), "an older ack from elsewhere does not undo what was told");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,695 @@
|
||||
//! 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 characters a message may hold: 4,000 with Nitro, 2,000 without.
|
||||
pub fn message_limit(premium: u8) -> usize {
|
||||
if premium == 2 {
|
||||
4000
|
||||
} else {
|
||||
2000
|
||||
}
|
||||
}
|
||||
|
||||
/// Refuses text Discord would refuse, or nothing at all, before it is sent. A request that is bound to be turned down
|
||||
/// is one more request from an account that is being watched, and says nothing a sentence here does not say better.
|
||||
pub fn check_text(content: &str, premium: u8, may_be_empty: bool) -> Result<()> {
|
||||
let limit = message_limit(premium);
|
||||
let length = content.chars().count();
|
||||
if length > limit {
|
||||
bail!("That message is {length} characters long and Discord takes {limit}{}", if premium == 2 { "" } else { " (4,000 with Nitro)" });
|
||||
}
|
||||
if !may_be_empty && content.trim().is_empty() {
|
||||
bail!("There is nothing to send");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 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 text_discord_would_refuse_is_refused_here() {
|
||||
assert!(check_text("hello", 0, false).is_ok());
|
||||
assert!(check_text(&"a".repeat(2000), 0, false).is_ok());
|
||||
let long = check_text(&"a".repeat(2001), 0, false).unwrap_err().to_string();
|
||||
assert!(long.contains("2001") && long.contains("2000") && long.contains("Nitro"), "{long}");
|
||||
assert!(check_text(&"a".repeat(4000), 2, false).is_ok(), "Nitro takes more");
|
||||
assert!(check_text(&"a".repeat(4001), 2, false).is_err());
|
||||
// Counted in characters, not bytes.
|
||||
assert!(check_text(&"é".repeat(2000), 0, false).is_ok());
|
||||
assert!(check_text(" \n ", 0, false).is_err(), "nothing is nothing");
|
||||
assert!(check_text("", 0, true).is_ok(), "a caption may be empty");
|
||||
}
|
||||
|
||||
#[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)));
|
||||
}
|
||||
}
|
||||
@@ -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"));
|
||||
}
|
||||
}
|
||||
@@ -95,7 +95,7 @@ async fn default_sounds(token: &str) -> Vec<Sound> {
|
||||
let resp = http_client_for(token)
|
||||
.get(format!("{API_BASE}/soundboard-default-sounds"))
|
||||
.header("Authorization", token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.ok()?;
|
||||
resp.status().is_success().then_some(())?;
|
||||
@@ -229,7 +229,7 @@ async fn fetch(token: &str, id: &str) -> Result<Vec<u8>> {
|
||||
}
|
||||
let resp = http_client_for(token)
|
||||
.get(format!("https://cdn.discordapp.com/soundboard-sounds/{id}"))
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("fetching the sound")?;
|
||||
if !resp.status().is_success() {
|
||||
|
||||
@@ -74,7 +74,7 @@ pub async fn lottie_file(token: &str, id: &str) -> Result<String> {
|
||||
if !tokio::fs::try_exists(&path).await.unwrap_or(false) {
|
||||
let resp = http_client_for(token)
|
||||
.get(lottie_url(id))
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.context("fetching the sticker")?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -157,7 +157,7 @@ async fn standard_packs(token: &str) -> Vec<(String, Vec<Sticker>)> {
|
||||
let resp = http_client_for(token)
|
||||
.get(format!("{API_BASE}/sticker-packs"))
|
||||
.header("Authorization", token)
|
||||
.send()
|
||||
.send_paced()
|
||||
.await
|
||||
.ok()?;
|
||||
if !resp.status().is_success() {
|
||||
@@ -241,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(())
|
||||
}
|
||||
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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;
|
||||
|
||||
+40
-1
@@ -473,7 +473,13 @@ pub async fn dispatch(
|
||||
// connect: a few hundred channels would mean a few
|
||||
// hundred requests, most for conversations nobody
|
||||
// is about to read.
|
||||
backend::discord::catch_up_channel(state, &cfg.token, &cfg.user_id, cfg.display_name.as_deref(), buffer_id, &channel_id).await;
|
||||
// A conversation never read here is read for the first time now; one that has been is asked
|
||||
// only about what came after.
|
||||
if state.store.has_messages(buffer_id).unwrap_or(true) {
|
||||
backend::discord::catch_up_channel(state, &cfg.token, &cfg.user_id, cfg.display_name.as_deref(), buffer_id, &channel_id).await;
|
||||
} else {
|
||||
backend::discord::backfill_channel_history(state, &cfg.token, &cfg.user_id, cfg.display_name.as_deref(), buffer_id, &channel_id).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
// Steam keeps every conversation: opening one asks for its
|
||||
@@ -756,6 +762,10 @@ pub async fn dispatch(
|
||||
// name later.
|
||||
state.highlights.forget(id);
|
||||
state.ignores.forget(id);
|
||||
// And the Discord session kept for a restart to resume: a removed account must not leave its
|
||||
// servers and friends in a file.
|
||||
backend::discord::session_store::forget(id);
|
||||
backend::discord::send_guard::forget_account(id);
|
||||
// And everything else this process is holding about it: the
|
||||
// seventy-odd maps keyed by the account or by one of its
|
||||
// buffers. In memory only, so it mattered exactly when
|
||||
@@ -5771,6 +5781,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 {
|
||||
|
||||
@@ -1544,6 +1544,24 @@ impl Runtime {
|
||||
}
|
||||
}
|
||||
|
||||
/// Says a conversation was active at `ts`, if that is later than what is known: for a service that tells when its
|
||||
/// last message was without sending it.
|
||||
pub fn raise_buffer_activity(&self, state: &AppState, buffer_id: &str, ts: i64) {
|
||||
let updated = {
|
||||
let mut buffers = self.buffers.lock().unwrap();
|
||||
match buffers.get_mut(buffer_id) {
|
||||
Some(b) if b.last_activity_ts < ts => {
|
||||
b.last_activity_ts = ts;
|
||||
Some(b.clone())
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
};
|
||||
if let Some(b) = updated {
|
||||
state.events.emit("bufferListChange", serde_json::to_value(&b).unwrap());
|
||||
}
|
||||
}
|
||||
|
||||
pub fn is_server_buffer(&self, buffer_id: &str) -> bool {
|
||||
self.buffers
|
||||
.lock()
|
||||
|
||||
@@ -777,6 +777,44 @@ impl Store {
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
/// The most recent message this buffer holds, and whether it is the account's own.
|
||||
pub fn newest_msg(&self, buffer_id: &str) -> Result<Option<(String, bool)>> {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let row = conn
|
||||
.query_row(
|
||||
"SELECT msg_id, is_own FROM messages WHERE buffer_id = ?1 AND msg_id IS NOT NULL ORDER BY ts DESC LIMIT 1",
|
||||
params![buffer_id],
|
||||
|row| Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? != 0)),
|
||||
)
|
||||
.optional()?;
|
||||
Ok(row)
|
||||
}
|
||||
|
||||
/// The ids of the account's own messages in these buffers from `since` (Unix seconds) on, newest first, at most
|
||||
/// `limit`.
|
||||
pub fn own_message_ids_since(&self, buffer_ids: &[String], since: i64, limit: usize) -> Result<Vec<String>> {
|
||||
let mut found = Vec::new();
|
||||
if buffer_ids.is_empty() {
|
||||
return Ok(found);
|
||||
}
|
||||
let conn = self.conn.lock().unwrap();
|
||||
for chunk in buffer_ids.chunks(200) {
|
||||
let marks = vec!["?"; chunk.len()].join(",");
|
||||
let sql = format!("SELECT msg_id FROM messages WHERE is_own = 1 AND ts >= ? AND msg_id IS NOT NULL AND buffer_id IN ({marks}) ORDER BY ts DESC LIMIT {limit}");
|
||||
let mut args: Vec<&dyn rusqlite::ToSql> = vec![&since];
|
||||
args.extend(chunk.iter().map(|b| b as &dyn rusqlite::ToSql));
|
||||
let mut stmt = conn.prepare(&sql)?;
|
||||
let rows = stmt.query_map(args.as_slice(), |row| row.get::<_, String>(0))?;
|
||||
for id in rows {
|
||||
found.push(id?);
|
||||
}
|
||||
if found.len() >= limit {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(found)
|
||||
}
|
||||
|
||||
/// The most recent message this buffer holds, for asking a service what
|
||||
/// has happened since.
|
||||
pub fn newest_msg_id(&self, buffer_id: &str) -> Result<Option<String>> {
|
||||
@@ -1853,6 +1891,42 @@ mod tests {
|
||||
.expect("appending");
|
||||
}
|
||||
|
||||
/// Like `append_at`, but the account's own.
|
||||
fn append_own_at(s: &Store, buffer: &str, id: &str, ts: i64) {
|
||||
s.append_message(buffer, id, "me", "hi", ts, false, false, "chat", None, &[], true, None, &[], &[], None, None, "channel", None, &[])
|
||||
.expect("appending");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn finds_the_newest_message_and_whether_it_is_the_accounts_own() {
|
||||
let (s, _dir) = store();
|
||||
assert_eq!(s.newest_msg("b").unwrap(), None);
|
||||
append_at(&s, "b", "m1", 100);
|
||||
assert_eq!(s.newest_msg("b").unwrap(), Some(("m1".to_string(), false)));
|
||||
append_own_at(&s, "b", "m2", 200);
|
||||
assert_eq!(s.newest_msg("b").unwrap(), Some(("m2".to_string(), true)));
|
||||
append_at(&s, "b", "m3", 300);
|
||||
assert_eq!(s.newest_msg("b").unwrap(), Some(("m3".to_string(), false)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn finds_what_the_account_has_said_in_a_set_of_rooms_since_a_time() {
|
||||
let (s, _dir) = store();
|
||||
let rooms = |names: &[&str]| names.iter().map(|n| n.to_string()).collect::<Vec<_>>();
|
||||
assert!(s.own_message_ids_since(&[], 0, 5).unwrap().is_empty(), "no rooms is no message");
|
||||
append_at(&s, "a", "m1", 500);
|
||||
assert!(s.own_message_ids_since(&rooms(&["a"]), 0, 5).unwrap().is_empty(), "somebody else's message is not the account's");
|
||||
append_own_at(&s, "b", "m2", 100);
|
||||
append_own_at(&s, "b", "m3", 200);
|
||||
assert_eq!(s.own_message_ids_since(&rooms(&["a", "b"]), 0, 5).unwrap(), vec!["m3", "m2"], "newest first");
|
||||
assert!(s.own_message_ids_since(&rooms(&["a"]), 0, 5).unwrap().is_empty(), "in another room");
|
||||
assert!(s.own_message_ids_since(&rooms(&["b"]), 201, 5).unwrap().is_empty(), "before the account joined");
|
||||
assert_eq!(s.own_message_ids_since(&rooms(&["b"]), 0, 1).unwrap(), vec!["m3"], "no more than asked for");
|
||||
// More rooms than one query holds.
|
||||
let many: Vec<String> = (0..450).map(|i| format!("r{i}")).chain(std::iter::once("b".to_string())).collect();
|
||||
assert_eq!(s.own_message_ids_since(&many, 0, 5).unwrap().len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn finds_the_newest_message_to_mark_a_room_read_up_to() {
|
||||
let (s, _dir) = store();
|
||||
|
||||
Reference in New Issue
Block a user