Compare commits
2
Commits
v1.3.1
...
development
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e9ce99fb42 | ||
|
|
7943767c47 |
Generated
+1
-1
@@ -4091,7 +4091,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "nobilis"
|
||||
version = "1.3.1"
|
||||
version = "1.3.2"
|
||||
dependencies = [
|
||||
"aes-gcm",
|
||||
"anyhow",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "nobilis"
|
||||
version = "1.3.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"
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -151,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);
|
||||
|
||||
@@ -207,6 +207,18 @@ 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" });
|
||||
@@ -382,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<'_> {
|
||||
@@ -403,6 +417,7 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
"d": {
|
||||
"token": config.token,
|
||||
"properties": identify_properties(),
|
||||
"capabilities": CAPABILITIES,
|
||||
"compress": false,
|
||||
"large_threshold": 50,
|
||||
// Carried in IDENTIFY as well as pushed live, so a status set
|
||||
@@ -614,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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -632,7 +650,7 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
let listed: Option<Vec<Value>> = if replaying {
|
||||
d["moho_dms"].as_array().cloned()
|
||||
} else {
|
||||
match http_client_for(&config.token).get(format!("{API_BASE}/users/@me/channels")).header("Authorization", &config.token).send().await {
|
||||
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) {
|
||||
@@ -674,7 +692,8 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
spawn_dm_catch_up(state.clone(), config.clone(), channel_map.clone(), newest_in);
|
||||
}
|
||||
|
||||
spawn_backfill(state.clone(), config.token.clone(), config.user_id.clone(), config.display_name.clone(), new_dm_buffers);
|
||||
// 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()) {
|
||||
@@ -846,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",
|
||||
@@ -1270,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.
|
||||
|
||||
@@ -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() {
|
||||
@@ -707,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
|
||||
}
|
||||
|
||||
@@ -785,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.
|
||||
@@ -956,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();
|
||||
@@ -980,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();
|
||||
@@ -1014,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() {
|
||||
@@ -1258,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 {
|
||||
@@ -250,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() {
|
||||
@@ -278,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() {
|
||||
@@ -311,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() {
|
||||
@@ -331,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() {
|
||||
@@ -356,6 +350,15 @@ 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")));
|
||||
|
||||
+118
-2
@@ -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")
|
||||
@@ -285,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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -42,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);
|
||||
}
|
||||
@@ -168,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 };
|
||||
|
||||
@@ -297,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() {
|
||||
@@ -381,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
|
||||
|
||||
@@ -262,6 +262,7 @@ pub async fn send_message(state: &AppState, buffer_id: &str, token: &str, body:
|
||||
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);
|
||||
@@ -403,6 +404,7 @@ pub async fn send_attachments(
|
||||
// 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);
|
||||
@@ -868,7 +870,12 @@ 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(());
|
||||
@@ -906,6 +913,22 @@ fn ack_due(channel_id: &str, msg_id: &str) -> bool {
|
||||
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() {
|
||||
@@ -1022,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");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -148,6 +148,29 @@ pub(super) async fn note_sent(resp: reqwest::Response) {
|
||||
}
|
||||
}
|
||||
|
||||
/// 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 {
|
||||
@@ -584,6 +607,20 @@ mod tests {
|
||||
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();
|
||||
|
||||
@@ -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() {
|
||||
|
||||
+7
-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
|
||||
|
||||
@@ -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,19 @@ 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>> {
|
||||
@@ -1884,6 +1897,18 @@ mod tests {
|
||||
.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();
|
||||
|
||||
Reference in New Issue
Block a user