1use std::collections::{HashMap, HashSet};
10use std::hash::BuildHasher;
11use std::sync::Arc;
12use std::time::Duration;
13
14use tokio::sync::oneshot;
15
16use forge_primitives::slack::{
17 SlackBookmark, SlackConversation, SlackFile, SlackFollowedThread, SlackMessage, SlackPin,
18 SlackSearchMatch, SlackSubscription, SlackSubscriptionTarget, SlackThreadOwner, SlackUser,
19 SlackWatchMode,
20};
21use serde::Deserialize;
22use serde::de::DeserializeOwned;
23
24pub trait SlackHost: Send + Sync {
28 fn client(&self, workspace: &str, timeout: Duration) -> Result<Arc<dyn SlackApi>, String>;
34
35 fn user_id(&self, workspace: &str) -> Option<String>;
38
39 fn user_name(&self, workspace: &str, user: &str) -> Option<String>;
44
45 fn set_user_name(&self, workspace: &str, user: &str, name: &str);
47
48 fn subscriptions(&self, workspace: &str) -> Vec<SlackSubscription>;
50
51 fn watermark(&self, workspace: &str, conversation: &str) -> Result<Option<String>, String>;
56
57 fn set_watermark(&self, workspace: &str, conversation: &str, ts: &str);
60
61 fn followed_threads(&self, workspace: &str, conversation: &str) -> Vec<SlackFollowedThread>;
67
68 fn thread_watermark(
73 &self,
74 workspace: &str,
75 conversation: &str,
76 parent_ts: &str,
77 ) -> Result<Option<String>, String>;
78
79 fn set_thread_watermark(&self, workspace: &str, conversation: &str, parent_ts: &str, ts: &str);
83
84 fn mark_thread_participating(&self, workspace: &str, conversation: &str, parent_ts: &str);
87
88 fn thread_full_read_at(
92 &self,
93 workspace: &str,
94 conversation: &str,
95 parent_ts: &str,
96 ) -> Option<std::time::Instant>;
97
98 fn mark_thread_full_read(
100 &self,
101 workspace: &str,
102 conversation: &str,
103 parent_ts: &str,
104 at: std::time::Instant,
105 );
106
107 fn thread_first_walk(&self, workspace: &str, conversation: &str, parent_ts: &str) -> bool;
114
115 fn clear_thread_first_walk(&self, workspace: &str, conversation: &str, parent_ts: &str);
117
118 fn thread_edit_watermark(
126 &self,
127 workspace: &str,
128 conversation: &str,
129 parent_ts: &str,
130 ) -> Option<String>;
131
132 fn set_thread_edit_watermark(
134 &self,
135 workspace: &str,
136 conversation: &str,
137 parent_ts: &str,
138 ts: &str,
139 );
140
141 fn first_author_failure(&self, workspace: &str, user: &str) -> bool;
147
148 fn first_draft_user_failure(&self, workspace: &str, user: &str) -> bool;
152
153 fn follow_thread(
158 &self,
159 workspace: &str,
160 conversation: &str,
161 parent_ts: &str,
162 owner: SlackThreadOwner,
163 since: &str,
164 );
165
166 fn set_connected(&self, workspace: &str, connected: bool);
168
169 fn name_conversation(&self, workspace: &str, conversation: &str, name: &str);
174
175 fn deliver(&self, subscription: &SlackSubscription, messages: &[SlackMessage]) -> bool;
181
182 fn auto_subscribe(&self, workspace: &str, message: &SlackMessage) -> bool;
187
188 fn own_post(&self, workspace: &str, conversation: &str, ts: &str) -> bool;
192
193 fn post_in_flight(&self, workspace: &str, conversation: &str) -> Option<Duration>;
200}
201
202#[async_trait::async_trait]
211pub trait SlackApi: Send + Sync {
212 async fn auth_test(&self) -> Result<AuthTest, SlackError>;
214 async fn list_conversations(&self) -> Result<Vec<SlackConversation>, SlackError>;
215 async fn history(
216 &self,
217 channel: &str,
218 oldest: Option<&str>,
219 newest: Option<&str>,
220 limit: u32,
221 cursor: Option<&str>,
222 ) -> Result<MessagePage, SlackError>;
223 async fn replies(
224 &self,
225 channel: &str,
226 ts: &str,
227 oldest: Option<&str>,
228 limit: u32,
229 cursor: Option<&str>,
230 ) -> Result<MessagePage, SlackError>;
231 async fn post_message(
236 &self,
237 channel: &str,
238 text: &str,
239 thread_ts: Option<&str>,
240 ) -> Result<PostedMessage, SlackError>;
241 async fn update_message(&self, channel: &str, ts: &str, text: &str) -> Result<(), SlackError>;
243 async fn delete_message(&self, channel: &str, ts: &str) -> Result<(), SlackError>;
245 async fn set_reaction(
247 &self,
248 channel: &str,
249 ts: &str,
250 name: &str,
251 add: bool,
252 ) -> Result<(), SlackError>;
253 async fn get_bytes(&self, url: &str) -> Result<Vec<u8>, SlackError>;
255 async fn post_bytes(&self, url: &str, body: Vec<u8>) -> Result<(), SlackError>;
257 async fn file_info(&self, id: &str) -> Result<SlackFile, SlackError>;
259 async fn get_upload_url(
262 &self,
263 name: &str,
264 length: usize,
265 ) -> Result<(String, String), SlackError>;
266 async fn complete_upload(
270 &self,
271 file_id: &str,
272 channel: &str,
273 thread_ts: Option<&str>,
274 ) -> Result<Vec<SlackFile>, SlackError>;
275 async fn search_messages(
278 &self,
279 query: &str,
280 count: u32,
281 cursor: Option<&str>,
282 ) -> Result<SearchPage, SlackError>;
283 async fn user_info(&self, user: &str) -> Result<SlackUser, SlackError>;
285 async fn list_users(&self) -> Result<Vec<SlackUser>, SlackError>;
288 async fn pins(&self, channel: &str) -> Result<Vec<SlackPin>, SlackError>;
290 async fn bookmarks(&self, channel: &str) -> Result<Vec<SlackBookmark>, SlackError>;
292}
293
294#[async_trait::async_trait]
295impl SlackApi for SlackClient {
296 async fn auth_test(&self) -> Result<AuthTest, SlackError> {
297 SlackClient::auth_test(self).await
298 }
299
300 async fn list_conversations(&self) -> Result<Vec<SlackConversation>, SlackError> {
301 SlackClient::list_conversations(self).await
302 }
303
304 async fn history(
305 &self,
306 channel: &str,
307 oldest: Option<&str>,
308 newest: Option<&str>,
309 limit: u32,
310 cursor: Option<&str>,
311 ) -> Result<MessagePage, SlackError> {
312 SlackClient::history(self, channel, oldest, newest, limit, cursor).await
313 }
314
315 async fn replies(
316 &self,
317 channel: &str,
318 ts: &str,
319 oldest: Option<&str>,
320 limit: u32,
321 cursor: Option<&str>,
322 ) -> Result<MessagePage, SlackError> {
323 SlackClient::replies(self, channel, ts, oldest, limit, cursor).await
324 }
325
326 async fn post_message(
327 &self,
328 channel: &str,
329 text: &str,
330 thread_ts: Option<&str>,
331 ) -> Result<PostedMessage, SlackError> {
332 SlackClient::post_message(self, channel, text, thread_ts).await
333 }
334
335 async fn update_message(&self, channel: &str, ts: &str, text: &str) -> Result<(), SlackError> {
336 SlackClient::update_message(self, channel, ts, text).await
337 }
338
339 async fn delete_message(&self, channel: &str, ts: &str) -> Result<(), SlackError> {
340 SlackClient::delete_message(self, channel, ts).await
341 }
342
343 async fn set_reaction(
344 &self,
345 channel: &str,
346 ts: &str,
347 name: &str,
348 add: bool,
349 ) -> Result<(), SlackError> {
350 SlackClient::set_reaction(self, channel, ts, name, add).await
351 }
352
353 async fn get_bytes(&self, url: &str) -> Result<Vec<u8>, SlackError> {
354 SlackClient::get_bytes(self, url).await
355 }
356
357 async fn post_bytes(&self, url: &str, body: Vec<u8>) -> Result<(), SlackError> {
358 SlackClient::post_bytes(self, url, body).await
359 }
360
361 async fn file_info(&self, id: &str) -> Result<SlackFile, SlackError> {
362 SlackClient::file_info(self, id).await
363 }
364
365 async fn get_upload_url(
366 &self,
367 name: &str,
368 length: usize,
369 ) -> Result<(String, String), SlackError> {
370 SlackClient::get_upload_url(self, name, length).await
371 }
372
373 async fn complete_upload(
374 &self,
375 file_id: &str,
376 channel: &str,
377 thread_ts: Option<&str>,
378 ) -> Result<Vec<SlackFile>, SlackError> {
379 SlackClient::complete_upload(self, file_id, channel, thread_ts).await
380 }
381
382 async fn search_messages(
383 &self,
384 query: &str,
385 count: u32,
386 cursor: Option<&str>,
387 ) -> Result<SearchPage, SlackError> {
388 SlackClient::search_messages(self, query, count, cursor).await
389 }
390
391 async fn user_info(&self, user: &str) -> Result<SlackUser, SlackError> {
392 SlackClient::user_info(self, user).await
393 }
394
395 async fn list_users(&self) -> Result<Vec<SlackUser>, SlackError> {
396 SlackClient::list_users(self).await
397 }
398
399 async fn pins(&self, channel: &str) -> Result<Vec<SlackPin>, SlackError> {
400 SlackClient::pins(self, channel).await
401 }
402
403 async fn bookmarks(&self, channel: &str) -> Result<Vec<SlackBookmark>, SlackError> {
404 SlackClient::bookmarks(self, channel).await
405 }
406}
407
408pub const POST_LIMIT: usize = 4000;
412
413const POST_PART_BUDGET: usize = 3800;
416
417pub fn split_for_post(text: &str) -> Vec<String> {
420 if text.chars().count() <= POST_LIMIT {
421 return vec![text.to_owned()];
422 }
423 let chars: Vec<char> = text.chars().collect();
424 let total = chars.len().div_ceil(POST_PART_BUDGET);
425 chars
426 .chunks(POST_PART_BUDGET)
427 .enumerate()
428 .map(|(index, chunk)| {
429 format!("[{}/{}] {}", index + 1, total, chunk.iter().collect::<String>())
430 })
431 .collect()
432}
433
434pub fn mentions(text: &str, user_id: &str) -> bool {
436 text.contains(&format!("<@{user_id}>"))
437}
438
439pub fn mention_ids(text: &str) -> Vec<String> {
444 let mut ids: Vec<String> = Vec::new();
445 let mut rest = text;
446 while let Some(start) = rest.find("<@") {
447 let after = &rest[start + 2..];
448 let Some(end) = after.find('>') else { break };
449 let id = after[..end].split('|').next().unwrap_or_default();
450 if is_user_id(id) && !ids.iter().any(|seen| seen == id) {
451 ids.push(id.to_owned());
452 }
453 rest = &after[end + 1..];
454 }
455 ids
456}
457
458fn is_user_id(id: &str) -> bool {
460 id.chars().next().is_some_and(|first| first.is_ascii_alphabetic())
461 && id.chars().all(|ch| ch.is_ascii_alphanumeric())
462}
463
464pub fn flatten_mrkdwn<S: BuildHasher>(
472 text: &str,
473 names: &HashMap<String, String, S>,
474 limit: usize,
475) -> String {
476 let chars: Vec<char> = text.chars().collect();
477 let mut out = String::with_capacity(text.len().min(limit.saturating_mul(2)));
478 let mut non_space = 0usize;
479 let mut index = 0;
480 while index < chars.len() && non_space <= limit {
481 match chars[index] {
482 '<' => {
485 let token = chars[index + 1..]
486 .iter()
487 .position(|&ch| ch == '>')
488 .map(|offset| index + 1 + offset)
489 .filter(|end| is_markup_token(&chars[index + 1..*end]));
490 if let Some(end) = token {
491 let inner: String = chars[index + 1..end].iter().collect();
492 let flat = flatten_token(&inner, names);
493 non_space += flat.chars().filter(|ch| !ch.is_whitespace()).count();
494 out.push_str(&flat);
495 index = end + 1;
496 } else {
497 out.push('<');
498 non_space += 1;
499 index += 1;
500 }
501 }
502 '*' if opens_bold(&chars, index) => {
505 if let Some(close) = first_bold_closer(&chars, index) {
506 let inner: String = chars[index + 1..close].iter().collect();
507 let flat = flatten_mrkdwn(&inner, names, limit);
508 non_space += flat.chars().filter(|ch| !ch.is_whitespace()).count();
509 out.push_str(&flat);
510 index = close + 1;
511 } else {
512 out.push('*');
513 non_space += 1;
514 index += 1;
515 }
516 }
517 ch => {
518 if !ch.is_whitespace() {
519 non_space += 1;
520 }
521 out.push(ch);
522 index += 1;
523 }
524 }
525 }
526 out
527}
528
529fn opens_bold(chars: &[char], at: usize) -> bool {
534 let before = at.checked_sub(1).and_then(|p| chars.get(p));
535 let after = chars.get(at + 1).copied();
536 before.is_none_or(|p| !p.is_alphanumeric() && *p != '*')
537 && after.is_some_and(|n| !n.is_whitespace() && n != '*')
538}
539
540fn first_bold_closer(chars: &[char], open: usize) -> Option<usize> {
545 ((open + 1)..chars.len()).find(|&j| {
546 chars[j] == '*'
547 && chars.get(j - 1).is_some_and(|p| !p.is_whitespace())
548 && chars.get(j + 1).is_none_or(|n| !n.is_alphanumeric())
549 && !opens_bold(chars, j)
550 })
551}
552
553fn is_markup_token(token: &[char]) -> bool {
556 match token.first().copied() {
557 Some('@' | '#' | '!') => true,
558 Some('h') => {
559 token.starts_with(&['h', 't', 't', 'p', 's', ':', '/', '/'])
560 || token.starts_with(&['h', 't', 't', 'p', ':', '/', '/'])
561 }
562 Some('m') => token.starts_with(&['m', 'a', 'i', 'l', 't', 'o', ':']),
563 _ => false,
564 }
565}
566
567fn flatten_token<S: BuildHasher>(inner: &str, names: &HashMap<String, String, S>) -> String {
571 if let Some(rest) = inner.strip_prefix('@') {
572 let id = rest.split('|').next().unwrap_or_default();
573 if is_user_id(id) {
574 return names.get(id).map_or_else(|| id.to_owned(), |name| format!("@{name}"));
575 }
576 return format!("<{inner}>");
577 }
578 if inner.starts_with('#') || inner.starts_with('!') {
579 return format!("<{inner}>");
580 }
581 match inner.split_once('|') {
582 Some((_, label)) if !label.is_empty() => label.to_owned(),
583 Some((address, _)) => address.to_owned(),
584 None => inner.to_owned(),
585 }
586}
587
588pub fn one_line_excerpt(text: &str, limit: usize) -> String {
592 let flat = text.split_whitespace().collect::<Vec<_>>().join(" ");
593 if flat.chars().count() <= limit {
594 return flat;
595 }
596 let prefix: String = flat.chars().take(limit).collect();
597 let kept = prefix.rfind(' ').map_or(prefix.as_str(), |at| &prefix[..at]);
598 format!("{}...", kept.trim_end())
599}
600
601pub fn shared_messages(file: &SlackFile) -> Vec<(String, String)> {
606 file.shares
607 .public
608 .iter()
609 .chain(file.shares.private.iter())
610 .flat_map(|(conversation, shares)| {
611 shares.iter().map(move |share| (conversation.clone(), share.ts.clone()))
612 })
613 .collect()
614}
615
616fn matches(
621 subscription: &SlackSubscription,
622 conversation: &SlackConversation,
623 text: &str,
624 ts: &str,
625 user_id: &str,
626 own_post: &dyn Fn(&str) -> bool,
627) -> bool {
628 if own_post(ts) {
629 return false;
630 }
631 match &subscription.target {
632 SlackSubscriptionTarget::DirectMessages => conversation.is_im || conversation.is_mpim,
633 SlackSubscriptionTarget::Conversation { id, mode, .. } => {
634 id == &conversation.id
635 && match mode {
636 SlackWatchMode::All => true,
637 SlackWatchMode::MentionsOnly => mentions(text, user_id),
638 }
639 }
640 SlackSubscriptionTarget::Mentions => false,
642 }
643}
644
645pub fn wants(
647 subscriptions: &[SlackSubscription],
648 conversation: &SlackConversation,
649 text: &str,
650 ts: &str,
651 user_id: &str,
652 own_post: &dyn Fn(&str) -> bool,
653) -> bool {
654 subscriptions.iter().any(|s| matches(s, conversation, text, ts, user_id, own_post))
655}
656
657fn targets(subscriptions: &[SlackSubscription], conversation: &SlackConversation) -> bool {
662 subscriptions.iter().any(|subscription| covers(subscription, conversation))
663}
664
665fn covers(subscription: &SlackSubscription, conversation: &SlackConversation) -> bool {
668 match &subscription.target {
669 SlackSubscriptionTarget::DirectMessages => conversation.is_im || conversation.is_mpim,
670 SlackSubscriptionTarget::Conversation { id, .. } => id == &conversation.id,
671 SlackSubscriptionTarget::Mentions => false,
673 }
674}
675
676fn owns_threads(subscription: &SlackSubscription, conversation: &SlackConversation) -> bool {
683 covers(subscription, conversation)
684 && !matches!(
685 &subscription.target,
686 SlackSubscriptionTarget::Conversation { mode: SlackWatchMode::MentionsOnly, .. }
687 )
688}
689
690fn followed_parent_of(message: &SlackHistoryMessage) -> Option<String> {
695 message.thread_ts.clone().or_else(|| is_thread_parent(message).then(|| message.ts.clone()))
696}
697
698fn is_thread_parent(message: &SlackHistoryMessage) -> bool {
704 message.reply_count > 0 && message.thread_ts.as_deref() == Some(message.ts.as_str())
705}
706
707fn participating(replies: &[SlackHistoryMessage], user_id: &str) -> bool {
715 replies
716 .iter()
717 .any(|reply| reply.user.as_deref() == Some(user_id) || mentions(&reply.text, user_id))
718}
719
720struct OwnIdentity<'a> {
722 id: &'a str,
724 name: Option<String>,
727 asked: bool,
728}
729
730impl OwnIdentity<'_> {
731 async fn resolve_name(
735 &mut self,
736 api: &dyn SlackApi,
737 workspace: &str,
738 ) -> Result<Option<&str>, SlackError> {
739 if !self.asked {
740 match api.auth_test().await {
741 Ok(auth) => self.name = Some(auth.user).filter(|name| !name.is_empty()),
742 Err(error @ SlackError::RateLimited { .. }) => return Err(error),
743 Err(error) => {
744 tracing::warn!(
745 target: "forge_connectors::slack",
746 event_name = "slack_auth_test_failed",
747 workspace,
748 %error,
749 "auth.test failed; the author falls back to the name on the message, the app's on an app-authored post",
750 );
751 }
752 }
753 self.asked = true;
754 }
755 Ok(self.name.as_deref())
756 }
757}
758
759async fn author_of(
771 host: &dyn SlackHost,
772 api: &dyn SlackApi,
773 workspace: &str,
774 message: &SlackHistoryMessage,
775 names: &mut HashMap<String, Option<String>>,
776 own: &mut OwnIdentity<'_>,
777) -> Result<Option<String>, SlackError> {
778 let Some(user) = message.user.as_deref() else { return Ok(None) };
779 if let Some(known) = names.get(user) {
780 return Ok(known.clone());
781 }
782 let name = resolve_author(host, api, workspace, message, own).await?;
783 names.insert(user.to_owned(), name.clone());
784 Ok(name)
785}
786
787async fn own_name(
795 host: &dyn SlackHost,
796 api: &dyn SlackApi,
797 workspace: &str,
798 own: &mut OwnIdentity<'_>,
799) -> Result<Option<String>, SlackError> {
800 let id = own.id;
801 if let Some(name) = own.resolve_name(api, workspace).await? {
802 host.set_user_name(workspace, id, name);
803 return Ok(Some(name.to_owned()));
804 }
805 Ok(host.user_name(workspace, own.id).filter(|name| !name.is_empty()))
806}
807
808async fn resolve_author(
811 host: &dyn SlackHost,
812 api: &dyn SlackApi,
813 workspace: &str,
814 message: &SlackHistoryMessage,
815 own: &mut OwnIdentity<'_>,
816) -> Result<Option<String>, SlackError> {
817 let Some(user) = message.user.as_deref() else { return Ok(None) };
818 if user == own.id
821 && let Some(name) = own_name(host, api, workspace, own).await?
822 {
823 return Ok(Some(name));
824 }
825 if let Some(known) = host.user_name(workspace, user) {
826 return Ok((!known.is_empty()).then_some(known));
828 }
829 let inline = message.bot_name.as_deref().filter(|name| !name.is_empty()).map(str::to_owned);
832 let name = match inline {
833 Some(name) => Some(name),
834 None => match api.user_info(user).await {
835 Ok(found) => Some(found.name).filter(|name| !name.is_empty()),
836 Err(error @ SlackError::RateLimited { .. }) => return Err(error),
839 Err(SlackError::Api { error, .. }) if error == "user_not_found" => {
842 tracing::debug!(
843 target: "forge_connectors::slack",
844 workspace,
845 user,
846 error = %error,
847 "slack knows no user by this id; the message carries no name",
848 );
849 None
850 }
851 Err(error) => {
858 if host.first_author_failure(workspace, user) {
859 tracing::warn!(
860 target: "forge_connectors::slack",
861 workspace,
862 user,
863 %error,
864 "resolving a slack author failed; the message carries no name this time",
865 );
866 } else {
867 tracing::debug!(
868 target: "forge_connectors::slack",
869 workspace,
870 user,
871 %error,
872 "resolving a slack author failed again",
873 );
874 }
875 return Ok(None);
876 }
877 },
878 };
879 host.set_user_name(workspace, user, name.as_deref().unwrap_or_default());
880 Ok(name)
881}
882
883pub async fn user_display_name(
888 host: &dyn SlackHost,
889 api: &dyn SlackApi,
890 workspace: &str,
891 user: &str,
892) -> Option<String> {
893 if let Some(known) = host.user_name(workspace, user) {
894 return (!known.is_empty()).then_some(known);
895 }
896 match api.user_info(user).await {
897 Ok(found) => {
898 let name = Some(found.name).filter(|name| !name.is_empty());
899 host.set_user_name(workspace, user, name.as_deref().unwrap_or_default());
900 name
901 }
902 Err(SlackError::Api { error, .. }) if error == "user_not_found" => {
905 host.set_user_name(workspace, user, "");
906 None
907 }
908 Err(error) => {
909 if host.first_draft_user_failure(workspace, user) {
910 tracing::warn!(
911 target: "forge_connectors::slack",
912 event_name = "slack_draft_user_unreadable",
913 workspace,
914 user,
915 %error,
916 "resolving a slack user for a held draft failed; the card draws the id",
917 );
918 } else {
919 tracing::debug!(
920 target: "forge_connectors::slack",
921 event_name = "slack_draft_user_unreadable",
922 workspace,
923 user,
924 %error,
925 "resolving a slack user for a held draft failed again",
926 );
927 }
928 None
929 }
930 }
931}
932
933const API_ROOT: &str = "https://slack.com/api";
934
935pub(crate) const RETRY_FALLBACK: Duration = Duration::from_secs(5);
937pub(crate) const RETRY_FLOOR: Duration = Duration::from_secs(1);
940pub(crate) const RETRY_CAP: Duration = Duration::from_secs(60);
942
943pub(crate) fn retry_delay(retry_after_secs: Option<u64>) -> Duration {
945 match retry_after_secs {
946 None => RETRY_FALLBACK,
947 Some(secs) => Duration::from_secs(secs).clamp(RETRY_FLOOR, RETRY_CAP),
948 }
949}
950
951fn retry_after_of(response: &reqwest::Response) -> Option<u64> {
953 response
954 .headers()
955 .get(reqwest::header::RETRY_AFTER)
956 .and_then(|value| value.to_str().ok())
957 .and_then(|value| value.parse::<u64>().ok())
958}
959
960const BODY_PREFIX_LEN: usize = 200;
962
963fn body_prefix(body: &str) -> String {
964 body.chars().take(BODY_PREFIX_LEN).collect()
965}
966
967fn status_failure(method: &str, status: reqwest::StatusCode, body: &str) -> SlackError {
971 SlackError::Transport {
972 method: method.to_owned(),
973 detail: format!("HTTP {status}: {}", body_prefix(body)),
974 }
975}
976
977fn encode_form(params: &[(&str, String)]) -> String {
979 let mut serializer = form_urlencoded::Serializer::new(String::new());
980 for (key, value) in params {
981 serializer.append_pair(key, value);
982 }
983 serializer.finish()
984}
985
986pub struct SlackClient {
988 http: reqwest::Client,
989 token: String,
990}
991
992impl std::fmt::Debug for SlackClient {
993 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
994 f.debug_struct("SlackClient").finish_non_exhaustive()
995 }
996}
997
998#[derive(Debug, Clone)]
999pub enum SlackError {
1000 Api { method: String, error: String, needed: Option<String> },
1002 Transport { method: String, detail: String },
1004 RateLimited { method: String, retry_after: Duration },
1006 Truncated { method: String, pages: usize },
1011}
1012
1013impl std::fmt::Display for SlackError {
1014 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1015 match self {
1016 Self::Api { method, error, needed } => {
1017 write!(f, "slack {method} failed: {error}")?;
1018 if let Some(needed) = needed {
1019 write!(f, " (needs {needed})")?;
1020 }
1021 Ok(())
1022 }
1023 Self::Transport { method, detail } => write!(f, "slack {method} transport: {detail}"),
1024 Self::RateLimited { method, retry_after } => {
1025 write!(f, "slack {method} rate limited, retry in {}s", retry_after.as_secs())
1026 }
1027 Self::Truncated { method, pages } => {
1028 write!(
1029 f,
1030 "slack {method} exceeded {pages} pages; the walk is longer than forge reads"
1031 )
1032 }
1033 }
1034 }
1035}
1036
1037impl std::error::Error for SlackError {}
1038
1039#[derive(Deserialize)]
1041struct Envelope<T> {
1042 ok: bool,
1043 #[serde(default)]
1044 error: Option<String>,
1045 #[serde(default)]
1046 needed: Option<String>,
1047 #[serde(flatten)]
1048 value: T,
1049}
1050
1051pub(crate) fn decode_envelope<T: DeserializeOwned>(
1053 method: &str,
1054 body: &str,
1055) -> Result<T, SlackError> {
1056 let envelope: Envelope<T> = serde_json::from_str(body).map_err(|detail| {
1057 SlackError::Transport { method: method.to_owned(), detail: detail.to_string() }
1058 })?;
1059 if !envelope.ok {
1060 return Err(SlackError::Api {
1061 method: method.to_owned(),
1062 error: envelope.error.unwrap_or_else(|| "unknown".to_owned()),
1063 needed: envelope.needed,
1064 });
1065 }
1066 Ok(envelope.value)
1067}
1068
1069const SWEEP_LIMIT: u32 = 200;
1071
1072const MAX_HISTORY_PAGES: usize = 200;
1075
1076const MAX_REPLY_PAGES: usize = 200;
1079
1080const FULL_THREAD_READ_INTERVAL: Duration = Duration::from_secs(600);
1085
1086const MAX_FULL_THREAD_READS_PER_SWEEP: usize = 2;
1090
1091pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
1096
1097#[derive(Debug, Default)]
1099pub(crate) struct SweepOutcome {
1100 pub delivered: usize,
1103 pub rate_limited: Option<Duration>,
1106 pub conversation_gone: bool,
1110}
1111
1112fn just_below(ts: &str) -> String {
1121 let (seconds, micros) = ts.split_once('.').unwrap_or((ts, "0"));
1122 let seconds: u64 = seconds.parse().unwrap_or(0);
1123 let micros: u64 = micros.parse().unwrap_or(0);
1124 if micros == 0 {
1128 format!("{}.{:06}", seconds.saturating_sub(1), 999_999)
1129 } else {
1130 format!("{seconds}.{:06}", micros - 1)
1131 }
1132}
1133
1134fn is_newer(ts: &str, watermark: Option<&str>) -> bool {
1138 match watermark {
1139 None => true,
1140 Some(watermark) => ts > watermark,
1141 }
1142}
1143
1144fn is_news(message: &SlackHistoryMessage, cursor: Option<&str>) -> bool {
1149 is_newer(&message.ts, cursor)
1150 || message.edited.as_deref().is_some_and(|edited| is_newer(edited, cursor))
1151}
1152
1153fn newest_stamp(message: &SlackHistoryMessage) -> &str {
1158 match message.edited.as_deref() {
1159 Some(edited) if is_newer(edited, Some(message.ts.as_str())) => edited,
1160 _ => message.ts.as_str(),
1161 }
1162}
1163
1164#[cfg(test)]
1165mod is_newer_tests {
1166 use super::is_newer;
1167
1168 #[test]
1174 fn string_order_survives_trailing_zeros_a_numeric_compare_does_not() {
1175 assert!(
1176 is_newer("1700000000.000100", Some("1700000000.0001")),
1177 "string order, not numeric - these are equal as f64",
1178 );
1179 assert!(!is_newer("1700000000.0001", Some("1700000000.000100")));
1180 }
1181}
1182
1183async fn fetch_history(
1188 api: &dyn SlackApi,
1189 workspace: &str,
1190 channel: &str,
1191 oldest: Option<&str>,
1192) -> Result<Vec<SlackHistoryMessage>, SlackError> {
1193 let mut out = Vec::new();
1194 let mut cursor: Option<String> = None;
1195 let mut pages = 0;
1196 loop {
1197 let page = api.history(channel, oldest, None, SWEEP_LIMIT, cursor.as_deref()).await?;
1198 out.extend(page.messages);
1199 pages += 1;
1200 match page.next_cursor {
1201 Some(next) if pages < MAX_HISTORY_PAGES => cursor = Some(next),
1202 Some(_) => {
1203 tracing::warn!(
1204 target: "forge_connectors::slack",
1205 event_name = "slack_history_page_cap",
1206 workspace,
1207 channel,
1208 pages,
1209 "conversations.history exceeded the page cap; the channel is longer than \
1210 forge reads, so the walk is not treated as complete",
1211 );
1212 return Err(SlackError::Truncated {
1213 method: "conversations.history".to_owned(),
1214 pages,
1215 });
1216 }
1217 None => return Ok(out),
1218 }
1219 }
1220}
1221
1222async fn fetch_recent(
1230 api: &dyn SlackApi,
1231 channel: &str,
1232) -> Result<Vec<SlackHistoryMessage>, SlackError> {
1233 Ok(api.history(channel, None, None, SWEEP_LIMIT, None).await?.messages)
1234}
1235
1236async fn fetch_replies(
1240 api: &dyn SlackApi,
1241 workspace: &str,
1242 channel: &str,
1243 ts: &str,
1244 oldest: Option<&str>,
1245) -> Result<Vec<SlackHistoryMessage>, SlackError> {
1246 let mut out = Vec::new();
1247 let mut cursor: Option<String> = None;
1248 let mut pages = 0;
1249 loop {
1250 let page = api.replies(channel, ts, oldest, SWEEP_LIMIT, cursor.as_deref()).await?;
1251 out.extend(page.messages);
1252 pages += 1;
1253 match page.next_cursor {
1254 Some(next) if pages < MAX_REPLY_PAGES => cursor = Some(next),
1255 Some(_) => {
1256 tracing::warn!(
1257 target: "forge_connectors::slack",
1258 event_name = "slack_thread_page_cap",
1259 workspace,
1260 channel,
1261 parent = %ts,
1262 pages,
1263 "conversations.replies exceeded the page cap; the thread is longer than \
1264 forge reads, so the walk is not treated as complete",
1265 );
1266 return Err(SlackError::Truncated {
1267 method: "conversations.replies".to_owned(),
1268 pages,
1269 });
1270 }
1271 None => return Ok(out),
1272 }
1273 }
1274}
1275
1276pub async fn read_thread(
1282 api: &dyn SlackApi,
1283 channel: &str,
1284 ts: &str,
1285 limit: u32,
1286) -> Result<Vec<SlackHistoryMessage>, SlackError> {
1287 if limit == 0 {
1288 return Ok(Vec::new());
1289 }
1290 let mut out: Vec<SlackHistoryMessage> = Vec::new();
1291 let mut seen: HashSet<String> = HashSet::new();
1292 let mut cursor: Option<String> = None;
1293 let mut pages = 0;
1294 loop {
1295 let page = api.replies(channel, ts, None, limit, cursor.as_deref()).await?;
1296 for message in page.messages {
1297 if seen.insert(message.ts.clone()) {
1298 out.push(message);
1299 if out.len() >= limit as usize {
1300 return Ok(out);
1301 }
1302 }
1303 }
1304 pages += 1;
1305 match page.next_cursor {
1306 Some(next) if pages < MAX_REPLY_PAGES => cursor = Some(next),
1307 Some(_) => {
1308 tracing::warn!(
1309 target: "forge_connectors::slack",
1310 event_name = "slack_thread_read_page_cap",
1311 channel,
1312 parent = %ts,
1313 pages,
1314 "conversations.replies exceeded the page cap; the read is not complete",
1315 );
1316 return Err(SlackError::Truncated {
1317 method: "conversations.replies".to_owned(),
1318 pages,
1319 });
1320 }
1321 None => return Ok(out),
1322 }
1323 }
1324}
1325
1326pub async fn thread_parent(
1330 api: &dyn SlackApi,
1331 channel: &str,
1332 ts: &str,
1333) -> Result<Option<SlackHistoryMessage>, SlackError> {
1334 let page = api.replies(channel, ts, None, 1, None).await?;
1335 Ok(page.messages.into_iter().next())
1336}
1337
1338#[derive(Clone, Copy)]
1343pub(crate) enum SweepScope {
1344 Conversations,
1346 DirectMessages,
1349}
1350
1351fn in_scope(
1355 subscriptions: &[SlackSubscription],
1356 conversation: &SlackConversation,
1357 scope: SweepScope,
1358) -> bool {
1359 let named = subscriptions.iter().any(|subscription| {
1360 matches!(
1361 &subscription.target,
1362 SlackSubscriptionTarget::Conversation { id, .. } if id == &conversation.id
1363 )
1364 });
1365 match scope {
1366 SweepScope::Conversations => named,
1367 SweepScope::DirectMessages => {
1368 (conversation.is_im || conversation.is_mpim)
1369 && !named
1370 && subscriptions
1371 .iter()
1372 .any(|s| matches!(s.target, SlackSubscriptionTarget::DirectMessages))
1373 }
1374 }
1375}
1376
1377#[cfg(test)]
1381pub(crate) async fn sweep(
1382 host: &dyn SlackHost,
1383 api: &dyn SlackApi,
1384 workspace: &str,
1385) -> Result<SweepOutcome, SlackError> {
1386 let conversations = sweep_scope(host, api, workspace, SweepScope::Conversations).await?;
1387 let dms = sweep_scope(host, api, workspace, SweepScope::DirectMessages).await?;
1388 Ok(SweepOutcome {
1389 delivered: conversations.delivered + dms.delivered,
1390 rate_limited: conversations.rate_limited.or(dms.rate_limited),
1391 conversation_gone: conversations.conversation_gone || dms.conversation_gone,
1392 })
1393}
1394
1395pub(crate) async fn sweep_scope(
1399 host: &dyn SlackHost,
1400 api: &dyn SlackApi,
1401 workspace: &str,
1402 scope: SweepScope,
1403) -> Result<SweepOutcome, SlackError> {
1404 let subscriptions = host.subscriptions(workspace);
1405 if subscriptions.is_empty() {
1406 return Ok(SweepOutcome { delivered: 0, ..Default::default() });
1407 }
1408 let user_id = host.user_id(workspace).unwrap_or_default();
1409 if user_id.is_empty() {
1410 tracing::warn!(
1416 target: "forge_connectors::slack",
1417 workspace,
1418 "slack user id unresolved; the sweep is skipped this tick",
1419 );
1420 return Ok(SweepOutcome { delivered: 0, ..Default::default() });
1421 }
1422
1423 let mut conversations = match api.list_conversations().await {
1424 Ok(conversations) => conversations,
1425 Err(SlackError::RateLimited { retry_after, .. }) => {
1426 return Ok(SweepOutcome {
1427 delivered: 0,
1428 rate_limited: Some(retry_after),
1429 ..Default::default()
1430 });
1431 }
1432 Err(error) => return Err(error),
1433 };
1434 for subscription in &subscriptions {
1439 if let SlackSubscriptionTarget::Conversation { id, .. } = &subscription.target
1440 && !conversations.iter().any(|conversation| &conversation.id == id)
1441 {
1442 conversations.push(SlackConversation {
1443 id: id.clone(),
1444 name: None,
1445 is_channel: false,
1446 is_private: false,
1447 is_im: false,
1448 is_mpim: false,
1449 is_archived: false,
1450 user: None,
1451 purpose: None,
1452 topic: None,
1453 });
1454 }
1455 }
1456 for subscription in &subscriptions {
1459 if let SlackSubscriptionTarget::Conversation { id, name: None, .. } = &subscription.target
1460 && let Some(name) = conversations
1461 .iter()
1462 .find(|conversation| &conversation.id == id)
1463 .and_then(|conversation| conversation.name.as_deref())
1464 && !name.is_empty()
1465 && name != id
1466 {
1467 host.name_conversation(workspace, id, name);
1468 }
1469 }
1470
1471 let mut delivered = 0;
1472 let mut rate_limited: Option<Duration> = None;
1475 let mut conversation_gone = false;
1476 let mut own = OwnIdentity { id: &user_id, name: None, asked: false };
1477 for conversation in &conversations {
1478 if !targets(&subscriptions, conversation) || !in_scope(&subscriptions, conversation, scope)
1479 {
1480 continue;
1481 }
1482 let in_flight = host.post_in_flight(workspace, &conversation.id).or_else(|| {
1489 conversation.name.as_deref().and_then(|name| host.post_in_flight(workspace, name))
1490 });
1491 let in_flight = in_flight.or_else(|| {
1492 conversation.user.as_deref().and_then(|user| host.post_in_flight(workspace, user))
1493 });
1494 if let Some(age) = in_flight {
1495 tracing::debug!(
1496 target: "forge_connectors::slack",
1497 event_name = "slack_post_in_flight",
1498 workspace,
1499 conversation = %conversation.id,
1500 age_ms = age.as_millis(),
1501 "a post is in flight; skipping this conversation this tick",
1502 );
1503 continue;
1504 }
1505 let watermark = match host.watermark(workspace, &conversation.id) {
1506 Ok(watermark) => watermark,
1507 Err(error) => {
1508 tracing::warn!(
1512 target: "forge_connectors::slack",
1513 workspace,
1514 conversation = %conversation.id,
1515 %error,
1516 "reading the slack watermark failed; skipping the conversation this tick",
1517 );
1518 continue;
1519 }
1520 };
1521
1522 let window = match fetch_recent(api, &conversation.id).await {
1523 Ok(window) => window,
1524 Err(SlackError::RateLimited { retry_after, .. }) => {
1525 return Ok(SweepOutcome {
1526 delivered,
1527 rate_limited: Some(retry_after),
1528 ..Default::default()
1529 });
1530 }
1531 Err(error) => {
1535 let gone = matches!(&error, SlackError::Api { error: api_error, .. }
1536 if api_error == "channel_not_found" || api_error == "invalid_channel_id");
1537 if gone && (conversation.is_im || conversation.is_mpim) {
1541 tracing::info!(
1542 target: "forge_connectors::slack",
1543 workspace,
1544 conversation = %conversation.id,
1545 %error,
1546 "slack withholds this DM's history from user tokens; skipping it every tick",
1547 );
1548 continue;
1549 }
1550 if gone {
1554 conversation_gone = true;
1555 }
1556 tracing::warn!(
1557 target: "forge_connectors::slack",
1558 workspace,
1559 conversation = %conversation.id,
1560 %error,
1561 "reading a conversation's history failed; skipping it this tick",
1562 );
1563 continue;
1564 }
1565 };
1566
1567 if watermark.is_none() {
1577 if let Some(newest) = window.iter().map(|message| message.ts.as_str()).max() {
1578 tracing::info!(
1579 target: "forge_connectors::slack",
1580 workspace,
1581 conversation = %conversation.id,
1582 cursor = %newest,
1583 "no cursor for this conversation; baselining at the newest message seen",
1584 );
1585 host.set_watermark(workspace, &conversation.id, newest);
1586 }
1587 continue;
1588 }
1589
1590 let reaches_back =
1597 window.last().is_some_and(|oldest| !is_newer(&oldest.ts, watermark.as_deref()));
1598 let history = if reaches_back {
1599 window.clone()
1600 } else {
1601 match fetch_history(api, workspace, &conversation.id, watermark.as_deref()).await {
1602 Ok(history) => history,
1603 Err(SlackError::RateLimited { retry_after, .. }) => {
1604 return Ok(SweepOutcome {
1605 delivered,
1606 rate_limited: Some(retry_after),
1607 ..Default::default()
1608 });
1609 }
1610 Err(error) => {
1611 tracing::warn!(
1612 target: "forge_connectors::slack",
1613 workspace,
1614 conversation = %conversation.id,
1615 %error,
1616 "reading a conversation's history failed; skipping it this tick",
1617 );
1618 continue;
1619 }
1620 }
1621 };
1622 let latest_replies: HashMap<&str, &str> = window
1625 .iter()
1626 .filter_map(|message| Some((message.ts.as_str(), message.latest_reply.as_deref()?)))
1627 .collect();
1628
1629 for message in window.iter().filter(|message| is_thread_parent(message)) {
1636 let anchor = match message.latest_reply.as_deref() {
1637 Some(latest) => just_below(latest),
1638 None => message.ts.clone(),
1639 };
1640 let since = match watermark.as_deref() {
1645 Some(cursor) if is_newer(cursor, Some(anchor.as_str())) => cursor.to_owned(),
1646 _ => anchor,
1647 };
1648 for subscription in
1649 subscriptions.iter().filter(|subscription| owns_threads(subscription, conversation))
1650 {
1651 host.follow_thread(
1652 workspace,
1653 &conversation.id,
1654 &message.ts,
1655 SlackThreadOwner {
1656 project: subscription.project.clone(),
1657 team_role: subscription.team_role.clone(),
1658 },
1659 &since,
1660 );
1661 }
1662 }
1663
1664 let mut seen: HashSet<String> = HashSet::new();
1665 let mut batch: Vec<SlackHistoryMessage> = history
1666 .into_iter()
1667 .filter(|message| {
1668 is_news(message, watermark.as_deref()) && seen.insert(message.ts.clone())
1669 })
1670 .collect();
1671
1672 let parents: Vec<String> = batch
1675 .iter()
1676 .filter(|message| is_thread_parent(message))
1677 .map(|m| m.ts.clone())
1678 .collect();
1679 for parent in parents {
1680 let replies = match fetch_replies(api, workspace, &conversation.id, &parent, None).await
1681 {
1682 Ok(replies) => replies,
1683 Err(SlackError::RateLimited { retry_after, .. }) => {
1684 return Ok(SweepOutcome {
1685 delivered,
1686 rate_limited: Some(retry_after),
1687 ..Default::default()
1688 });
1689 }
1690 Err(error) => {
1693 tracing::warn!(
1694 target: "forge_connectors::slack",
1695 workspace,
1696 conversation = %conversation.id,
1697 parent = %parent,
1698 %error,
1699 "reading a thread's replies failed; skipping them this tick",
1700 );
1701 continue;
1702 }
1703 };
1704 if !participating(&replies, &user_id) {
1707 continue;
1708 }
1709 for reply in replies {
1710 if is_news(&reply, watermark.as_deref()) && seen.insert(reply.ts.clone()) {
1711 batch.push(reply);
1712 }
1713 }
1714 }
1715
1716 let mut names: HashMap<String, Option<String>> = HashMap::new();
1719 for message in &batch {
1720 match author_of(host, api, workspace, message, &mut names, &mut own).await {
1721 Ok(_) => {}
1722 Err(SlackError::RateLimited { retry_after, .. }) => {
1723 return Ok(SweepOutcome {
1724 delivered,
1725 rate_limited: Some(retry_after),
1726 ..Default::default()
1727 });
1728 }
1729 Err(error) => return Err(error),
1730 }
1731 }
1732
1733 let label =
1734 conversation.name.clone().or_else(|| conversation.user.clone()).unwrap_or_default();
1735 let own_posts: HashSet<&str> = batch
1738 .iter()
1739 .map(|message| message.ts.as_str())
1740 .filter(|ts| host.own_post(workspace, &conversation.id, ts))
1741 .collect();
1742 let own_post = |ts: &str| own_posts.contains(ts);
1743 let mut all_handed = true;
1749 let mut handed_edits: HashMap<String, String> = HashMap::new();
1752 for subscription in &subscriptions {
1753 let wanted: Vec<&SlackHistoryMessage> = batch
1754 .iter()
1755 .filter(|message| {
1756 matches(
1757 subscription,
1758 conversation,
1759 &message.text,
1760 &message.ts,
1761 &user_id,
1762 &own_post,
1763 )
1764 })
1765 .collect();
1766 if wanted.is_empty() {
1767 continue;
1768 }
1769 let messages: Vec<SlackMessage> = wanted
1770 .iter()
1771 .map(|message| SlackMessage {
1772 workspace: workspace.to_owned(),
1773 conversation: conversation.id.clone(),
1774 conversation_label: label.clone(),
1775 ts: message.ts.clone(),
1776 thread_ts: message.thread_ts.clone(),
1777 user: message.user.clone(),
1778 author: message
1779 .user
1780 .as_deref()
1781 .and_then(|user| names.get(user).cloned().flatten()),
1782 text: message.text.clone(),
1783 parent_user_id: message.parent_user_id.clone(),
1784 latest_reply: message.latest_reply.clone(),
1785 edited: message.edited.clone(),
1786 files: message.files.clone(),
1787 })
1788 .collect();
1789 if !host.deliver(subscription, &messages) {
1790 all_handed = false;
1791 continue;
1792 }
1793 delivered += 1;
1794 for message in wanted {
1795 if let Some(parent_ts) = followed_parent_of(message)
1796 && message.ts != parent_ts
1797 && let Some(edited) = message.edited.as_deref()
1798 {
1799 let stamp = handed_edits.entry(parent_ts.clone()).or_default();
1800 if edited > stamp.as_str() {
1801 edited.clone_into(stamp);
1802 }
1803 }
1804 if let Some(parent_ts) = followed_parent_of(message) {
1805 host.follow_thread(
1806 workspace,
1807 &conversation.id,
1808 &parent_ts,
1809 SlackThreadOwner {
1810 project: subscription.project.clone(),
1811 team_role: subscription.team_role.clone(),
1812 },
1813 &message.ts,
1814 );
1815 }
1816 }
1817 }
1818
1819 if all_handed {
1824 for (parent_ts, edited) in handed_edits {
1825 let current = host.thread_edit_watermark(workspace, &conversation.id, &parent_ts);
1826 if current.as_deref().is_none_or(|mark| is_newer(&edited, Some(mark))) {
1827 host.set_thread_edit_watermark(
1828 workspace,
1829 &conversation.id,
1830 &parent_ts,
1831 &edited,
1832 );
1833 }
1834 }
1835 if let Some(newest) = batch.iter().map(newest_stamp).max() {
1836 host.set_watermark(workspace, &conversation.id, newest);
1837 }
1838 }
1839
1840 let mut full_reads = 0usize;
1844 for thread in host.followed_threads(workspace, &conversation.id) {
1845 let cursor = match host.thread_watermark(workspace, &conversation.id, &thread.parent_ts)
1846 {
1847 Ok(cursor) => cursor,
1848 Err(error) => {
1849 tracing::warn!(
1850 target: "forge_connectors::slack",
1851 workspace,
1852 conversation = %conversation.id,
1853 parent = %thread.parent_ts,
1854 %error,
1855 "reading a thread cursor failed; skipping the thread this tick",
1856 );
1857 continue;
1858 }
1859 };
1860 let first_walk = host.thread_first_walk(workspace, &conversation.id, &thread.parent_ts);
1865 let now = std::time::Instant::now();
1866 if first_walk
1871 && host
1872 .thread_full_read_at(workspace, &conversation.id, &thread.parent_ts)
1873 .is_some_and(|at| at + FULL_THREAD_READ_INTERVAL > now)
1874 {
1875 continue;
1876 }
1877 if !thread.participating
1878 && !first_walk
1879 && let Some(latest) = latest_replies.get(thread.parent_ts.as_str())
1880 && cursor.as_deref().is_some_and(|cursor| *latest <= cursor)
1881 {
1882 continue;
1883 }
1884 let full_due = thread.participating
1893 && !first_walk
1894 && full_reads < MAX_FULL_THREAD_READS_PER_SWEEP
1895 && host
1896 .thread_full_read_at(workspace, &conversation.id, &thread.parent_ts)
1897 .is_none_or(|at| at + FULL_THREAD_READ_INTERVAL <= now);
1898 if full_due {
1899 full_reads += 1;
1900 }
1901 if full_due || first_walk {
1902 host.mark_thread_full_read(workspace, &conversation.id, &thread.parent_ts, now);
1909 }
1910 let read_from: Option<String> = if full_due || first_walk || !thread.participating {
1911 None
1912 } else {
1913 cursor.as_deref().map(just_below)
1914 };
1915 let replies = match fetch_replies(
1916 api,
1917 workspace,
1918 &conversation.id,
1919 &thread.parent_ts,
1920 read_from.as_deref(),
1921 )
1922 .await
1923 {
1924 Ok(replies) => replies,
1925 Err(SlackError::RateLimited { retry_after, .. }) => {
1929 rate_limited =
1930 Some(rate_limited.map_or(retry_after, |held| held.max(retry_after)));
1931 continue;
1932 }
1933 Err(error) => {
1938 tracing::warn!(
1939 target: "forge_connectors::slack",
1940 event_name = "slack_thread_walk_failed",
1941 workspace,
1942 conversation = %conversation.id,
1943 parent = %thread.parent_ts,
1944 %error,
1945 "walking a followed thread failed; skipping it this tick",
1946 );
1947 continue;
1948 }
1949 };
1950 if first_walk {
1951 host.clear_thread_first_walk(workspace, &conversation.id, &thread.parent_ts);
1952 }
1953 let participated = thread.participating || participating(&replies, &user_id);
1954 if participated && !thread.participating {
1955 host.mark_thread_participating(workspace, &conversation.id, &thread.parent_ts);
1956 }
1957 let edit_mark =
1964 host.thread_edit_watermark(workspace, &conversation.id, &thread.parent_ts);
1965 let news = |reply: &SlackHistoryMessage| {
1966 if first_walk {
1967 is_newer(&reply.ts, cursor.as_deref())
1968 } else if full_due {
1969 is_newer(&reply.ts, cursor.as_deref())
1970 || reply.edited.as_deref().is_some_and(|edited| {
1971 edit_mark.as_deref().is_none_or(|mark| is_newer(edited, Some(mark)))
1972 })
1973 } else {
1974 is_news(reply, cursor.as_deref())
1975 }
1976 };
1977 let newest_seen: Option<String> = replies
1983 .iter()
1984 .filter(|reply| reply.ts != thread.parent_ts && news(reply))
1985 .map(|reply| newest_stamp(reply).to_owned())
1986 .max();
1987 let next_cursor = match (cursor.clone(), newest_seen) {
1991 (Some(current), Some(newest)) if newest > current => Some(newest),
1992 (None, Some(newest)) => Some(newest),
1993 _ => None,
1994 };
1995 if !participated {
1996 if let Some(newest) = next_cursor {
1997 host.set_thread_watermark(
1998 workspace,
1999 &conversation.id,
2000 &thread.parent_ts,
2001 &newest,
2002 );
2003 }
2004 continue;
2005 }
2006 let mut deliverable: Vec<(&SlackHistoryMessage, Option<String>)> = Vec::new();
2010 let mut thread_throttled = false;
2011 for reply in &replies {
2012 if reply.ts == thread.parent_ts || !news(reply) || !seen.insert(reply.ts.clone()) {
2014 continue;
2015 }
2016 if host.own_post(workspace, &conversation.id, &reply.ts) {
2022 continue;
2023 }
2024 match author_of(host, api, workspace, reply, &mut names, &mut own).await {
2028 Ok(author) => deliverable.push((reply, author)),
2029 Err(SlackError::RateLimited { retry_after, .. }) => {
2030 rate_limited =
2031 Some(rate_limited.map_or(retry_after, |held| held.max(retry_after)));
2032 thread_throttled = true;
2033 break;
2034 }
2035 Err(error) => return Err(error),
2036 }
2037 }
2038 if thread_throttled {
2039 continue;
2040 }
2041 let mut walk_failed = false;
2042 for owner in &thread.owners {
2043 let messages: Vec<SlackMessage> = deliverable
2044 .iter()
2045 .map(|(reply, author)| SlackMessage {
2046 workspace: workspace.to_owned(),
2047 conversation: conversation.id.clone(),
2048 conversation_label: label.clone(),
2049 ts: reply.ts.clone(),
2050 thread_ts: Some(thread.parent_ts.clone()),
2051 user: reply.user.clone(),
2052 author: author.clone(),
2053 text: reply.text.clone(),
2054 parent_user_id: reply.parent_user_id.clone(),
2055 latest_reply: reply.latest_reply.clone(),
2056 edited: reply.edited.clone(),
2057 files: reply.files.clone(),
2058 })
2059 .collect();
2060 if messages.is_empty() {
2061 break;
2062 }
2063 let Some(subscription) = subscriptions.iter().find(|subscription| {
2066 subscription.project == owner.project
2067 && subscription.team_role == owner.team_role
2068 }) else {
2069 walk_failed = true;
2070 continue;
2071 };
2072 if host.deliver(subscription, &messages) {
2073 delivered += 1;
2074 } else {
2075 walk_failed = true;
2081 }
2082 }
2083 if !walk_failed {
2084 let delivered_edit =
2092 deliverable.iter().filter_map(|(reply, _)| reply.edited.as_deref()).max();
2093 if let Some(edited) = delivered_edit
2094 && edit_mark.as_deref().is_none_or(|mark| is_newer(edited, Some(mark)))
2095 {
2096 host.set_thread_edit_watermark(
2097 workspace,
2098 &conversation.id,
2099 &thread.parent_ts,
2100 edited,
2101 );
2102 }
2103 if let Some(newest) = next_cursor {
2104 host.set_thread_watermark(
2105 workspace,
2106 &conversation.id,
2107 &thread.parent_ts,
2108 &newest,
2109 );
2110 }
2111 }
2112 }
2113 }
2114
2115 Ok(SweepOutcome { delivered, rate_limited, conversation_gone })
2116}
2117
2118pub const MENTION_CURSOR: &str = "__mentions__";
2122
2123const MENTION_SEARCH_COUNT: u32 = 20;
2126
2127const MAX_SEARCH_PAGES: usize = 200;
2130
2131pub(crate) async fn sweep_mentions(
2135 host: &dyn SlackHost,
2136 api: &dyn SlackApi,
2137 workspace: &str,
2138) -> Result<SweepOutcome, SlackError> {
2139 let subscriptions = host.subscriptions(workspace);
2140 let mention_subscriptions: Vec<&SlackSubscription> = subscriptions
2145 .iter()
2146 .filter(|s| matches!(s.target, SlackSubscriptionTarget::Mentions))
2147 .collect();
2148 if mention_subscriptions.is_empty() {
2149 return Ok(SweepOutcome { delivered: 0, ..Default::default() });
2150 }
2151 let Some(user_id) = host.user_id(workspace).filter(|id| !id.is_empty()) else {
2155 tracing::warn!(
2156 target: "forge_connectors::slack",
2157 workspace,
2158 "slack user id unresolved; the mention sweep is skipped this tick",
2159 );
2160 return Ok(SweepOutcome { delivered: 0, ..Default::default() });
2161 };
2162
2163 let mut own = OwnIdentity { id: &user_id, name: None, asked: false };
2164
2165 let watermark = match host.watermark(workspace, MENTION_CURSOR) {
2166 Ok(watermark) => watermark,
2167 Err(error) => {
2168 tracing::warn!(
2169 target: "forge_connectors::slack",
2170 workspace,
2171 %error,
2172 "reading the mention cursor failed; skipping the mention sweep this tick",
2173 );
2174 return Ok(SweepOutcome { delivered: 0, ..Default::default() });
2175 }
2176 };
2177 let query = format!("<@{user_id}>");
2178
2179 let mut delivered = 0;
2184 let mut newest: Option<String> = None;
2185 let mut cursor: Option<String> = None;
2186 let mut pages = 0;
2187 loop {
2188 let page = match api.search_messages(&query, MENTION_SEARCH_COUNT, cursor.as_deref()).await
2189 {
2190 Ok(page) => page,
2191 Err(SlackError::RateLimited { retry_after, .. }) => {
2192 return Ok(SweepOutcome {
2196 delivered,
2197 rate_limited: Some(retry_after),
2198 ..Default::default()
2199 });
2200 }
2201 Err(error) => return Err(error),
2202 };
2203 pages += 1;
2204
2205 for hit in &page.matches {
2206 if !is_newer(&hit.ts, watermark.as_deref()) {
2208 continue;
2209 }
2210 let in_flight = host.post_in_flight(workspace, &hit.conversation_id).or_else(|| {
2220 hit.conversation_name
2221 .as_deref()
2222 .and_then(|name| host.post_in_flight(workspace, name))
2223 });
2224 if let Some(age) = in_flight {
2225 tracing::debug!(
2226 target: "forge_connectors::slack",
2227 event_name = "slack_post_in_flight",
2228 workspace,
2229 conversation = %hit.conversation_id,
2230 age_ms = age.as_millis(),
2231 "a post is in flight in this conversation; holding the mention cursor",
2232 );
2233 return Ok(SweepOutcome { delivered, ..Default::default() });
2234 }
2235 if host.own_post(workspace, &hit.conversation_id, &hit.ts) {
2240 continue;
2241 }
2242 let author = if hit.user.as_deref() == Some(own.id) {
2246 match own_name(host, api, workspace, &mut own).await {
2247 Ok(name) => {
2248 name.or_else(|| hit.username.clone().filter(|name| !name.is_empty()))
2249 }
2250 Err(SlackError::RateLimited { retry_after, .. }) => {
2251 return Ok(SweepOutcome {
2252 delivered,
2253 rate_limited: Some(retry_after),
2254 ..Default::default()
2255 });
2256 }
2257 Err(error) => return Err(error),
2258 }
2259 } else {
2260 hit.username.clone().filter(|name| !name.is_empty())
2261 };
2262 let message = SlackMessage {
2263 workspace: workspace.to_owned(),
2264 conversation: hit.conversation_id.clone(),
2265 conversation_label: hit
2266 .conversation_name
2267 .clone()
2268 .unwrap_or_else(|| hit.conversation_id.clone()),
2269 ts: hit.ts.clone(),
2270 thread_ts: hit.thread_ts.clone(),
2271 user: hit.user.clone(),
2272 author,
2273 text: hit.text.clone(),
2274 parent_user_id: None,
2277 latest_reply: None,
2278 edited: None,
2279 files: hit.files.clone(),
2280 };
2281 let mut all_handed = true;
2282 for subscription in &mention_subscriptions {
2283 if !host.deliver(subscription, std::slice::from_ref(&message)) {
2284 all_handed = false;
2285 }
2286 }
2287 if !all_handed {
2288 return Ok(SweepOutcome { delivered, ..Default::default() });
2293 }
2294 host.auto_subscribe(workspace, &message);
2297 for subscription in &mention_subscriptions {
2301 if let Some(parent_ts) = hit.thread_ts.clone() {
2302 host.follow_thread(
2303 workspace,
2304 &hit.conversation_id,
2305 &parent_ts,
2306 SlackThreadOwner {
2307 project: subscription.project.clone(),
2308 team_role: subscription.team_role.clone(),
2309 },
2310 &hit.ts,
2311 );
2312 }
2313 }
2314 delivered += 1;
2315 if newest.as_deref().is_none_or(|current| hit.ts.as_str() > current) {
2316 newest = Some(hit.ts.clone());
2317 }
2318 }
2319
2320 match page.next_cursor {
2321 Some(next) if pages < MAX_SEARCH_PAGES => cursor = Some(next),
2322 Some(_) => {
2323 tracing::warn!(
2324 target: "forge_connectors::slack",
2325 workspace,
2326 pages,
2327 "search.messages kept handing back a cursor; stopping the mention walk",
2328 );
2329 break;
2330 }
2331 None => break,
2332 }
2333 }
2334
2335 if let Some(newest) = newest {
2336 host.set_watermark(workspace, MENTION_CURSOR, &newest);
2337 }
2338
2339 Ok(SweepOutcome { delivered, ..Default::default() })
2340}
2341
2342pub(crate) fn next_interval(configured: Duration, rate_limited: Option<Duration>) -> Duration {
2345 rate_limited.map_or(configured, |delay| configured.max(delay.min(RETRY_CAP)))
2346}
2347
2348#[derive(Default)]
2351struct ThrottleEpisode {
2352 active: bool,
2353}
2354
2355impl ThrottleEpisode {
2356 fn note_pass(
2357 &mut self,
2358 workspace: &str,
2359 sweep: &'static str,
2360 configured: Duration,
2361 rate_limited: Option<Duration>,
2362 ) {
2363 match (rate_limited, self.active) {
2364 (Some(delay), false) => {
2365 self.active = true;
2366 tracing::warn!(
2367 target: "forge_connectors::slack::throttle",
2368 workspace = %workspace,
2369 sweep = %sweep,
2370 event_name = "slack_sweep_throttled",
2371 retry_after_secs = delay.as_secs(),
2372 configured_secs = configured.as_secs(),
2373 "slack throttled the sweep",
2374 );
2375 }
2376 (Some(delay), true) => {
2377 tracing::debug!(
2378 target: "forge_connectors::slack::throttle",
2379 workspace = %workspace,
2380 sweep = %sweep,
2381 event_name = "slack_sweep_throttled",
2382 retry_after_secs = delay.as_secs(),
2383 configured_secs = configured.as_secs(),
2384 "slack sweep still throttled",
2385 );
2386 }
2387 (None, true) => {
2388 self.active = false;
2389 tracing::info!(
2390 target: "forge_connectors::slack::throttle",
2391 workspace = %workspace,
2392 sweep = %sweep,
2393 event_name = "slack_sweep_recovered",
2394 "slack sweep recovered from throttling",
2395 );
2396 }
2397 (None, false) => {}
2398 }
2399 }
2400}
2401
2402#[cfg(feature = "testing")]
2404pub mod testing {
2405 use super::ThrottleEpisode;
2406 use std::time::Duration;
2407
2408 pub fn emit_a_repeated_throttle_pass() {
2411 let mut episode = ThrottleEpisode::default();
2412 let throttled = Some(Duration::from_secs(7));
2413 let configured = Duration::from_secs(30);
2414 episode.note_pass("acme", "conversation", configured, throttled);
2415 episode.note_pass("acme", "conversation", configured, throttled);
2416 }
2417}
2418
2419pub async fn run_workspace_pump(
2422 host: Arc<dyn SlackHost>,
2423 workspace: String,
2424 poll_seconds: u64,
2425 dm_poll_seconds: u64,
2426 mut shutdown: oneshot::Receiver<()>,
2427) {
2428 let interval = Duration::from_secs(poll_seconds);
2429 let dm_interval = Duration::from_secs(dm_poll_seconds);
2430 let Ok(client) = host.client(&workspace, REQUEST_TIMEOUT) else {
2431 tracing::warn!(
2432 target: "forge_connectors::slack",
2433 workspace = %workspace,
2434 "no client for this workspace; its pump stays down",
2435 );
2436 return;
2437 };
2438
2439 let (dm_shutdown_tx, dm_shutdown_rx) = oneshot::channel();
2443 let dm_pump = tokio::spawn(run_dm_pump(
2444 Arc::clone(&host),
2445 Arc::clone(&client),
2446 workspace.clone(),
2447 dm_interval,
2448 dm_shutdown_rx,
2449 ));
2450
2451 let mut throttle = ThrottleEpisode::default();
2452 let mut mention_throttle = ThrottleEpisode::default();
2453 let mut wait = interval;
2454 loop {
2455 tokio::select! {
2456 _ = &mut shutdown => break,
2457 () = tokio::time::sleep(wait) => {}
2458 }
2459 wait = interval;
2460 match sweep_scope(host.as_ref(), client.as_ref(), &workspace, SweepScope::Conversations)
2461 .await
2462 {
2463 Ok(outcome) => {
2464 host.set_connected(&workspace, !outcome.conversation_gone);
2468 tracing::debug!(
2469 target: "forge_connectors::slack",
2470 workspace = %workspace,
2471 delivered = outcome.delivered,
2472 "slack sweep finished",
2473 );
2474 throttle.note_pass(&workspace, "conversation", interval, outcome.rate_limited);
2477 wait = next_interval(interval, outcome.rate_limited);
2478 }
2479 Err(error) => {
2480 host.set_connected(&workspace, false);
2481 tracing::warn!(
2482 target: "forge_connectors::slack",
2483 workspace = %workspace,
2484 %error,
2485 "slack sweep failed",
2486 );
2487 }
2488 }
2489 match sweep_mentions(host.as_ref(), client.as_ref(), &workspace).await {
2493 Ok(outcome) => {
2494 mention_throttle.note_pass(&workspace, "mentions", interval, outcome.rate_limited);
2497 if let Some(delay) = outcome.rate_limited {
2498 wait = wait.max(next_interval(interval, Some(delay)));
2501 }
2502 }
2503 Err(error) => {
2504 if host
2510 .subscriptions(&workspace)
2511 .iter()
2512 .any(|sub| matches!(sub.target, SlackSubscriptionTarget::Mentions))
2513 {
2514 host.set_connected(&workspace, false);
2515 }
2516 tracing::warn!(
2517 target: "forge_connectors::slack",
2518 workspace = %workspace,
2519 %error,
2520 "slack mention sweep failed",
2521 );
2522 }
2523 }
2524 }
2525 drop(dm_shutdown_tx);
2528 let _ = dm_pump.await;
2529 host.set_connected(&workspace, false);
2530}
2531
2532async fn run_dm_pump(
2536 host: Arc<dyn SlackHost>,
2537 client: Arc<dyn SlackApi>,
2538 workspace: String,
2539 interval: Duration,
2540 mut shutdown: oneshot::Receiver<()>,
2541) {
2542 let mut throttle = ThrottleEpisode::default();
2543 let mut wait = Duration::ZERO;
2544 loop {
2545 tokio::select! {
2546 _ = &mut shutdown => break,
2547 () = tokio::time::sleep(wait) => {}
2548 }
2549 match sweep_scope(host.as_ref(), client.as_ref(), &workspace, SweepScope::DirectMessages)
2550 .await
2551 {
2552 Ok(outcome) => {
2553 tracing::debug!(
2554 target: "forge_connectors::slack",
2555 workspace = %workspace,
2556 delivered = outcome.delivered,
2557 "slack DM sweep finished",
2558 );
2559 throttle.note_pass(&workspace, "dm", interval, outcome.rate_limited);
2560 wait = next_interval(interval, outcome.rate_limited);
2561 }
2562 Err(error) => {
2563 host.set_connected(&workspace, false);
2567 tracing::warn!(
2568 target: "forge_connectors::slack",
2569 workspace = %workspace,
2570 %error,
2571 "slack DM sweep failed",
2572 );
2573 wait = interval;
2574 }
2575 }
2576 }
2577}
2578
2579#[derive(Debug, Clone, Deserialize)]
2583pub struct PostedMessage {
2584 #[serde(default)]
2585 pub channel: Option<String>,
2586 pub ts: String,
2587}
2588
2589#[derive(Debug, Clone, Deserialize)]
2591pub struct AuthTest {
2592 pub team: String,
2593 pub user: String,
2594 pub team_id: String,
2595 pub user_id: String,
2596 pub url: String,
2597}
2598
2599const CONVERSATION_TYPES: &str = "public_channel,private_channel,im,mpim";
2602
2603const MAX_CONVERSATION_PAGES: usize = 200;
2607
2608#[derive(Debug, Deserialize)]
2610struct RawConversationsPage {
2611 channels: Vec<SlackConversation>,
2612 #[serde(default)]
2613 response_metadata: Metadata,
2614}
2615
2616#[derive(Debug, Default, Deserialize)]
2617struct Metadata {
2618 #[serde(default)]
2619 next_cursor: String,
2620}
2621
2622#[derive(Debug)]
2625struct ConversationsPage {
2626 conversations: Vec<SlackConversation>,
2627 next_cursor: Option<String>,
2628}
2629
2630fn decode_conversations_page(body: &str) -> Result<ConversationsPage, SlackError> {
2632 let raw: RawConversationsPage = decode_envelope("users.conversations", body)?;
2633 Ok(page_from_raw(raw))
2634}
2635
2636fn page_from_raw(raw: RawConversationsPage) -> ConversationsPage {
2637 let cursor = (!raw.response_metadata.next_cursor.is_empty())
2638 .then_some(raw.response_metadata.next_cursor);
2639 ConversationsPage { conversations: raw.channels, next_cursor: cursor }
2640}
2641
2642#[derive(Debug, Clone, PartialEq, Eq)]
2644pub struct SlackHistoryMessage {
2645 pub ts: String,
2646 pub user: Option<String>,
2647 pub text: String,
2648 pub thread_ts: Option<String>,
2651 pub reply_count: u32,
2653 pub bot_name: Option<String>,
2656 pub parent_user_id: Option<String>,
2660 pub latest_reply: Option<String>,
2662 pub edited: Option<String>,
2672 pub files: Vec<SlackFile>,
2674}
2675
2676#[derive(Debug)]
2679pub struct MessagePage {
2680 pub messages: Vec<SlackHistoryMessage>,
2681 pub next_cursor: Option<String>,
2682}
2683
2684#[derive(Debug, Deserialize)]
2685struct RawMessagePage {
2686 #[serde(default)]
2687 messages: Vec<RawWireMessage>,
2688 #[serde(default)]
2689 response_metadata: Metadata,
2690}
2691
2692#[derive(Debug, Deserialize)]
2693struct RawWireMessage {
2694 ts: String,
2695 #[serde(default)]
2696 user: Option<String>,
2697 #[serde(default)]
2698 text: String,
2699 #[serde(default)]
2700 thread_ts: Option<String>,
2701 #[serde(default)]
2702 reply_count: u32,
2703 #[serde(default)]
2704 bot_profile: Option<BotProfile>,
2705 #[serde(default)]
2706 parent_user_id: Option<String>,
2707 #[serde(default)]
2708 latest_reply: Option<String>,
2709 #[serde(default)]
2710 edited: Option<RawEdited>,
2711 #[serde(default)]
2712 files: Vec<SlackFile>,
2713}
2714
2715#[derive(Debug, Deserialize)]
2718struct RawEdited {
2719 #[serde(default)]
2720 ts: Option<String>,
2721}
2722
2723#[derive(Debug, Deserialize)]
2726struct BotProfile {
2727 #[serde(default)]
2728 name: Option<String>,
2729}
2730
2731fn decode_message_page(method: &str, body: &str) -> Result<MessagePage, SlackError> {
2733 let raw: RawMessagePage = decode_envelope(method, body)?;
2734 let cursor = (!raw.response_metadata.next_cursor.is_empty())
2735 .then_some(raw.response_metadata.next_cursor);
2736 let messages = raw
2737 .messages
2738 .into_iter()
2739 .map(|message| SlackHistoryMessage {
2740 ts: message.ts,
2741 user: message.user,
2742 text: message.text,
2743 thread_ts: message.thread_ts,
2744 reply_count: message.reply_count,
2745 bot_name: message.bot_profile.and_then(|profile| profile.name),
2746 parent_user_id: message.parent_user_id,
2747 latest_reply: message.latest_reply,
2748 edited: message.edited.and_then(|edited| edited.ts),
2749 files: message.files,
2750 })
2751 .collect();
2752 Ok(MessagePage { messages, next_cursor: cursor })
2753}
2754
2755fn history_params(
2758 channel: &str,
2759 oldest: Option<&str>,
2760 newest: Option<&str>,
2761 limit: u32,
2762 cursor: Option<&str>,
2763) -> Vec<(&'static str, String)> {
2764 let mut params = vec![("channel", channel.to_owned()), ("limit", limit.to_string())];
2765 if let Some(oldest) = oldest {
2766 params.push(("oldest", oldest.to_owned()));
2767 }
2768 if let Some(newest) = newest {
2769 params.push(("latest", newest.to_owned()));
2770 }
2771 if let Some(cursor) = cursor {
2772 params.push(("cursor", cursor.to_owned()));
2773 }
2774 params
2775}
2776
2777const USER_PAGE_LIMIT: u32 = 200;
2779
2780const MAX_USER_PAGES: usize = 100;
2784
2785#[derive(Debug)]
2788struct UsersPage {
2789 members: Vec<SlackUser>,
2790 next_cursor: Option<String>,
2791}
2792
2793fn decode_users_page(body: &str) -> Result<UsersPage, SlackError> {
2795 #[derive(Deserialize)]
2796 struct RawUsersPage {
2797 #[serde(default)]
2798 members: Vec<SlackUser>,
2799 #[serde(default)]
2800 response_metadata: Metadata,
2801 }
2802 let raw: RawUsersPage = decode_envelope("users.list", body)?;
2803 let cursor = (!raw.response_metadata.next_cursor.is_empty())
2804 .then_some(raw.response_metadata.next_cursor);
2805 Ok(UsersPage { members: raw.members, next_cursor: cursor })
2806}
2807
2808#[async_trait::async_trait]
2812trait UsersPages: Send + Sync {
2813 async fn users_page(&self, cursor: Option<&str>) -> Result<UsersPage, SlackError>;
2814}
2815
2816#[async_trait::async_trait]
2817impl UsersPages for SlackClient {
2818 async fn users_page(&self, cursor: Option<&str>) -> Result<UsersPage, SlackError> {
2819 let mut params = vec![("limit", USER_PAGE_LIMIT.to_string())];
2820 if let Some(cursor) = cursor {
2821 params.push(("cursor", cursor.to_owned()));
2822 }
2823 let body = self.call_text("users.list", ¶ms).await?;
2824 decode_users_page(&body)
2825 }
2826}
2827
2828async fn walk_users(pages: &dyn UsersPages) -> Result<Vec<SlackUser>, SlackError> {
2832 let mut out = Vec::new();
2833 let mut cursor: Option<String> = None;
2834 for _ in 0..MAX_USER_PAGES {
2835 let page = pages.users_page(cursor.as_deref()).await?;
2836 out.extend(page.members);
2837 match page.next_cursor {
2838 Some(next) => cursor = Some(next),
2839 None => return Ok(out),
2840 }
2841 }
2842 tracing::warn!(
2843 target: "forge_connectors::slack",
2844 pages = MAX_USER_PAGES,
2845 users = out.len(),
2846 "users.list kept handing back a cursor; stopping the walk",
2847 );
2848 Ok(out)
2849}
2850
2851const SEARCH_COUNT_CAP: u32 = 100;
2853
2854fn search_params(query: &str, count: u32) -> Vec<(&'static str, String)> {
2858 vec![
2859 ("query", query.to_owned()),
2860 ("sort", "timestamp".to_owned()),
2861 ("count", count.min(SEARCH_COUNT_CAP).to_string()),
2862 ]
2863}
2864
2865#[derive(Debug, Default)]
2867pub struct SearchPage {
2868 pub matches: Vec<SlackSearchMatch>,
2869 pub next_cursor: Option<String>,
2870}
2871
2872fn decode_search(body: &str) -> Result<SearchPage, SlackError> {
2874 #[derive(Default, Deserialize)]
2875 struct SearchMessages {
2876 #[serde(default)]
2877 matches: Vec<RawMatch>,
2878 #[serde(default)]
2879 next_cursor: String,
2880 }
2881 #[derive(Default, Deserialize)]
2882 struct SearchPayload {
2883 #[serde(default)]
2884 messages: SearchMessages,
2885 }
2886 #[derive(Deserialize)]
2887 struct RawMatch {
2888 ts: String,
2889 #[serde(default)]
2890 text: String,
2891 channel: RawChannel,
2892 #[serde(default)]
2893 username: Option<String>,
2894 #[serde(default)]
2897 user: Option<String>,
2898 #[serde(default)]
2899 thread_ts: Option<String>,
2900 #[serde(default)]
2901 files: Vec<SlackFile>,
2902 }
2903 #[derive(Deserialize)]
2904 struct RawChannel {
2905 id: String,
2906 #[serde(default)]
2907 name: Option<String>,
2908 }
2909
2910 let payload: SearchPayload = decode_envelope("search.messages", body)?;
2911 let next_cursor =
2912 (!payload.messages.next_cursor.is_empty()).then_some(payload.messages.next_cursor);
2913 let matches = payload
2914 .messages
2915 .matches
2916 .into_iter()
2917 .map(|hit| SlackSearchMatch {
2918 ts: hit.ts,
2919 text: hit.text,
2920 conversation_id: hit.channel.id,
2921 conversation_name: hit.channel.name,
2922 username: hit.username,
2923 user: hit.user,
2924 thread_ts: hit.thread_ts,
2925 files: hit.files,
2926 })
2927 .collect();
2928 Ok(SearchPage { matches, next_cursor })
2929}
2930
2931fn decode_user(body: &str) -> Result<SlackUser, SlackError> {
2933 #[derive(Deserialize)]
2934 struct Wrapper {
2935 user: SlackUser,
2936 }
2937 let wrapper: Wrapper = decode_envelope("users.info", body)?;
2938 Ok(wrapper.user)
2939}
2940
2941fn decode_file(body: &str) -> Result<SlackFile, SlackError> {
2943 #[derive(Deserialize)]
2944 struct Wrapper {
2945 file: SlackFile,
2946 }
2947 let wrapper: Wrapper = decode_envelope("files.info", body)?;
2948 Ok(wrapper.file)
2949}
2950
2951fn decode_pins(body: &str) -> Result<Vec<SlackPin>, SlackError> {
2953 #[derive(Deserialize)]
2954 struct Wrapper {
2955 #[serde(default)]
2956 items: Vec<SlackPin>,
2957 }
2958 let wrapper: Wrapper = decode_envelope("pins.list", body)?;
2959 Ok(wrapper.items)
2960}
2961
2962fn decode_bookmarks(body: &str) -> Result<Vec<SlackBookmark>, SlackError> {
2964 #[derive(Deserialize)]
2965 struct Wrapper {
2966 #[serde(default)]
2967 bookmarks: Vec<SlackBookmark>,
2968 }
2969 let wrapper: Wrapper = decode_envelope("bookmarks.list", body)?;
2970 Ok(wrapper.bookmarks)
2971}
2972
2973impl SlackClient {
2974 pub fn new(http: reqwest::Client, token: String) -> Self {
2975 Self { http, token }
2976 }
2977
2978 pub async fn call<T: DeserializeOwned>(
2981 &self,
2982 method: &str,
2983 params: &[(&str, String)],
2984 ) -> Result<T, SlackError> {
2985 let body = self.call_text(method, params).await?;
2986 decode_envelope(method, &body)
2987 }
2988
2989 async fn call_text(
2992 &self,
2993 method: &str,
2994 params: &[(&str, String)],
2995 ) -> Result<String, SlackError> {
2996 let url = format!("{API_ROOT}/{method}");
2997 let response = self
2998 .http
2999 .post(&url)
3000 .bearer_auth(&self.token)
3001 .header(reqwest::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
3002 .body(encode_form(params))
3003 .timeout(REQUEST_TIMEOUT)
3006 .send()
3007 .await
3008 .map_err(|err| SlackError::Transport {
3009 method: method.to_owned(),
3010 detail: err.to_string(),
3011 })?;
3012
3013 let status = response.status();
3014 if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
3015 return Err(SlackError::RateLimited {
3016 method: method.to_owned(),
3017 retry_after: retry_delay(retry_after_of(&response)),
3018 });
3019 }
3020
3021 let body = response.text().await.map_err(|err| SlackError::Transport {
3022 method: method.to_owned(),
3023 detail: err.to_string(),
3024 })?;
3025 if !status.is_success() {
3026 return Err(status_failure(method, status, &body));
3027 }
3028 Ok(body)
3029 }
3030
3031 pub async fn auth_test(&self) -> Result<AuthTest, SlackError> {
3033 self.call("auth.test", &[]).await
3034 }
3035
3036 pub async fn history(
3040 &self,
3041 channel: &str,
3042 oldest: Option<&str>,
3043 newest: Option<&str>,
3044 limit: u32,
3045 cursor: Option<&str>,
3046 ) -> Result<MessagePage, SlackError> {
3047 let params = history_params(channel, oldest, newest, limit, cursor);
3048 let body = self.call_text("conversations.history", ¶ms).await?;
3049 decode_message_page("conversations.history", &body)
3050 }
3051
3052 pub async fn replies(
3055 &self,
3056 channel: &str,
3057 ts: &str,
3058 oldest: Option<&str>,
3059 limit: u32,
3060 cursor: Option<&str>,
3061 ) -> Result<MessagePage, SlackError> {
3062 let mut params = vec![
3063 ("channel", channel.to_owned()),
3064 ("ts", ts.to_owned()),
3065 ("limit", limit.to_string()),
3066 ];
3067 if let Some(oldest) = oldest {
3068 params.push(("oldest", oldest.to_owned()));
3069 }
3070 if let Some(cursor) = cursor {
3071 params.push(("cursor", cursor.to_owned()));
3072 }
3073 let body = self.call_text("conversations.replies", ¶ms).await?;
3074 decode_message_page("conversations.replies", &body)
3075 }
3076
3077 pub async fn post_message(
3080 &self,
3081 channel: &str,
3082 text: &str,
3083 thread_ts: Option<&str>,
3084 ) -> Result<PostedMessage, SlackError> {
3085 let mut params = vec![("channel", channel.to_owned()), ("text", text.to_owned())];
3086 if let Some(thread_ts) = thread_ts {
3087 params.push(("thread_ts", thread_ts.to_owned()));
3088 }
3089 self.call("chat.postMessage", ¶ms).await
3090 }
3091
3092 pub async fn update_message(
3096 &self,
3097 channel: &str,
3098 ts: &str,
3099 text: &str,
3100 ) -> Result<(), SlackError> {
3101 let params =
3102 vec![("channel", channel.to_owned()), ("ts", ts.to_owned()), ("text", text.to_owned())];
3103 let _: serde_json::Value = self.call("chat.update", ¶ms).await?;
3104 Ok(())
3105 }
3106
3107 pub async fn delete_message(&self, channel: &str, ts: &str) -> Result<(), SlackError> {
3109 let params = vec![("channel", channel.to_owned()), ("ts", ts.to_owned())];
3110 let _: serde_json::Value = self.call("chat.delete", ¶ms).await?;
3111 Ok(())
3112 }
3113
3114 pub async fn set_reaction(
3117 &self,
3118 channel: &str,
3119 ts: &str,
3120 name: &str,
3121 add: bool,
3122 ) -> Result<(), SlackError> {
3123 let method = if add { "reactions.add" } else { "reactions.remove" };
3124 let params = vec![
3125 ("channel", channel.to_owned()),
3126 ("timestamp", ts.to_owned()),
3127 ("name", name.to_owned()),
3128 ];
3129 let _: serde_json::Value = self.call(method, ¶ms).await?;
3130 Ok(())
3131 }
3132
3133 pub async fn get_bytes(&self, url: &str) -> Result<Vec<u8>, SlackError> {
3136 let response = self.http.get(url).bearer_auth(&self.token).send().await.map_err(|err| {
3137 SlackError::Transport { method: "download".to_owned(), detail: err.to_string() }
3138 })?;
3139 if response.status() == reqwest::StatusCode::TOO_MANY_REQUESTS {
3140 return Err(SlackError::RateLimited {
3141 method: "download".to_owned(),
3142 retry_after: retry_delay(retry_after_of(&response)),
3143 });
3144 }
3145 let status = response.status();
3146 let bytes = response.bytes().await.map_err(|err| SlackError::Transport {
3147 method: "download".to_owned(),
3148 detail: err.to_string(),
3149 })?;
3150 if !status.is_success() {
3151 return Err(SlackError::Transport {
3154 method: "download".to_owned(),
3155 detail: format!("HTTP {status}"),
3156 });
3157 }
3158 Ok(bytes.to_vec())
3159 }
3160
3161 pub async fn post_bytes(&self, url: &str, body: Vec<u8>) -> Result<(), SlackError> {
3164 let response = self.http.post(url).body(body).send().await.map_err(|err| {
3165 SlackError::Transport { method: "upload".to_owned(), detail: err.to_string() }
3166 })?;
3167 let status = response.status();
3168 if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
3169 return Err(SlackError::RateLimited {
3170 method: "upload".to_owned(),
3171 retry_after: retry_delay(retry_after_of(&response)),
3172 });
3173 }
3174 if !status.is_success() {
3175 return Err(SlackError::Transport {
3176 method: "upload".to_owned(),
3177 detail: format!("HTTP {status}"),
3178 });
3179 }
3180 Ok(())
3181 }
3182
3183 pub async fn file_info(&self, id: &str) -> Result<SlackFile, SlackError> {
3185 let params = vec![("file", id.to_owned())];
3186 let body = self.call_text("files.info", ¶ms).await?;
3187 decode_file(&body)
3188 }
3189
3190 pub async fn get_upload_url(
3193 &self,
3194 name: &str,
3195 length: usize,
3196 ) -> Result<(String, String), SlackError> {
3197 #[derive(Deserialize)]
3198 struct UploadTarget {
3199 upload_url: String,
3200 file_id: String,
3201 }
3202 let params = vec![("filename", name.to_owned()), ("length", length.to_string())];
3203 let target: UploadTarget = self.call("files.getUploadURLExternal", ¶ms).await?;
3204 Ok((target.upload_url, target.file_id))
3205 }
3206
3207 pub async fn complete_upload(
3210 &self,
3211 file_id: &str,
3212 channel: &str,
3213 thread_ts: Option<&str>,
3214 ) -> Result<Vec<SlackFile>, SlackError> {
3215 #[derive(Deserialize)]
3216 struct Uploaded {
3217 #[serde(default)]
3218 files: Vec<SlackFile>,
3219 }
3220 let files = serde_json::json!([{ "id": file_id }]).to_string();
3221 let mut params = vec![("files", files), ("channel_id", channel.to_owned())];
3222 if let Some(thread_ts) = thread_ts {
3223 params.push(("thread_ts", thread_ts.to_owned()));
3224 }
3225 let uploaded: Uploaded = self.call("files.completeUploadExternal", ¶ms).await?;
3226 Ok(uploaded.files)
3227 }
3228
3229 pub async fn search_messages(
3232 &self,
3233 query: &str,
3234 count: u32,
3235 cursor: Option<&str>,
3236 ) -> Result<SearchPage, SlackError> {
3237 let mut params = search_params(query, count);
3238 if let Some(cursor) = cursor {
3239 params.push(("cursor", cursor.to_owned()));
3240 }
3241 let body = self.call_text("search.messages", ¶ms).await?;
3242 decode_search(&body)
3243 }
3244
3245 pub async fn user_info(&self, user: &str) -> Result<SlackUser, SlackError> {
3247 let params = vec![("user", user.to_owned())];
3248 let body = self.call_text("users.info", ¶ms).await?;
3249 decode_user(&body)
3250 }
3251
3252 pub async fn list_users(&self) -> Result<Vec<SlackUser>, SlackError> {
3256 walk_users(self).await
3257 }
3258
3259 pub async fn pins(&self, channel: &str) -> Result<Vec<SlackPin>, SlackError> {
3261 let params = vec![("channel", channel.to_owned())];
3262 let body = self.call_text("pins.list", ¶ms).await?;
3263 decode_pins(&body)
3264 }
3265
3266 pub async fn bookmarks(&self, channel: &str) -> Result<Vec<SlackBookmark>, SlackError> {
3269 let params = vec![("channel_id", channel.to_owned())];
3270 let body = self.call_text("bookmarks.list", ¶ms).await?;
3271 decode_bookmarks(&body)
3272 }
3273
3274 pub async fn list_conversations(&self) -> Result<Vec<SlackConversation>, SlackError> {
3276 let mut out = Vec::new();
3277 let mut cursor: Option<String> = None;
3278 let mut pages = 0;
3279 loop {
3280 let mut params =
3281 vec![("types", CONVERSATION_TYPES.to_owned()), ("limit", "200".to_owned())];
3282 if let Some(cursor) = &cursor {
3283 params.push(("cursor", cursor.clone()));
3284 }
3285 let body = self.call_text("users.conversations", ¶ms).await?;
3286 let page = decode_conversations_page(&body)?;
3287 out.extend(page.conversations);
3288 pages += 1;
3289 match page.next_cursor {
3290 Some(_) if pages >= MAX_CONVERSATION_PAGES => {
3291 tracing::warn!(
3292 target: "forge_connectors::slack",
3293 pages,
3294 conversations = out.len(),
3295 "users.conversations kept handing back a cursor; stopping the walk",
3296 );
3297 return Ok(out);
3298 }
3299 Some(next) => cursor = Some(next),
3300 None => return Ok(out),
3301 }
3302 }
3303 }
3304}
3305
3306#[cfg(test)]
3307mod tests {
3308 use super::*;
3309 use forge_primitives::slack::{SlackFileShares, SlackThreadRecord};
3310 use std::collections::HashMap;
3311
3312 #[test]
3316 fn retry_delay_honours_the_header_and_floors_absurd_values() {
3317 assert_eq!(retry_delay(Some(7)), Duration::from_secs(7), "an honest header is honoured");
3318 assert_eq!(retry_delay(None), Duration::from_secs(5), "a missing header falls back");
3319 assert_eq!(
3321 retry_delay(Some(100_000)),
3322 Duration::from_secs(60),
3323 "an absurd header is capped"
3324 );
3325 assert_eq!(retry_delay(Some(0)), Duration::from_secs(1), "a zero header still waits");
3328 }
3329
3330 #[test]
3331 fn a_gateway_failure_reports_the_status_and_a_bounded_body_prefix() {
3332 let mut body = "bad gateway from the edge".to_owned();
3333 body.push_str(&"x".repeat(1000));
3334 let err = status_failure("users.conversations", reqwest::StatusCode::BAD_GATEWAY, &body);
3335 match err {
3336 SlackError::Transport { method, detail } => {
3337 assert!(detail.contains("502"), "the status is named, got: {detail}");
3338 assert!(detail.contains("bad gateway"), "the body prefix is carried: {detail}");
3339 assert!(detail.len() < 300, "the body is bounded, got {} chars", detail.len());
3340 assert_eq!(method, "users.conversations");
3341 }
3342 other => panic!("wrong variant: {other:?}"),
3343 }
3344 }
3345
3346 #[test]
3347 fn an_ok_false_envelope_is_an_api_error_not_a_decode_error() {
3348 let body = r#"{"ok":false,"error":"missing_scope","needed":"search:read.public"}"#;
3349 let err = decode_envelope::<serde_json::Value>("conversations.history", body)
3350 .expect_err("ok:false must be an error");
3351 match err {
3352 SlackError::Api { method, error, needed } => {
3353 assert_eq!(method, "conversations.history");
3354 assert_eq!(error, "missing_scope");
3355 assert_eq!(needed.as_deref(), Some("search:read.public"));
3356 }
3357 other => panic!("wrong variant: {other:?}"),
3358 }
3359 }
3360
3361 #[test]
3366 fn auth_test_yields_the_tokens_own_user_id() {
3367 let body = r#"{"ok":true,"team":"acme","user":"ved","team_id":"T1","user_id":"U123","url":"https://x.slack.com/"}"#;
3368 let auth: AuthTest = decode_envelope("auth.test", body).expect("decodes");
3369 assert_eq!(auth.user_id, "U123", "the id is what the filters compare");
3370 }
3371
3372 #[test]
3373 fn an_ok_true_envelope_decodes_its_payload() {
3374 let body = r#"{"ok":true,"team":"Trust Machines","user":"ved","team_id":"T1","user_id":"U1","url":"https://x.slack.com/"}"#;
3375 let got: AuthTest = decode_envelope("auth.test", body).expect("decodes");
3376 assert_eq!(got.team_id, "T1", "the flattened payload carries team_id");
3377 assert_eq!(got.user, "ved", "the flattened payload carries user");
3378 }
3379
3380 #[test]
3381 fn list_conversations_decodes_mixed_types_and_keeps_the_cursor() {
3382 let body = r#"{
3383 "ok": true,
3384 "channels": [
3385 {"id":"C1","name":"general","is_channel":true},
3386 {"id":"D1","is_im":true,"user":"U9"},
3387 {"id":"G1","name":"secret","is_private":true}
3388 ],
3389 "response_metadata": {"next_cursor": "abc123"}
3390 }"#;
3391 let page = decode_conversations_page(body).expect("decodes");
3392 assert_eq!(page.conversations.len(), 3, "every channel in the page is kept");
3393 assert_eq!(page.conversations[1].user.as_deref(), Some("U9"), "a DM keeps its partner");
3394 assert_eq!(page.next_cursor.as_deref(), Some("abc123"), "a non-empty cursor pages on");
3395
3396 let empty = decode_conversations_page(
3397 r#"{"ok":true,"channels":[],"response_metadata":{"next_cursor":""}}"#,
3398 )
3399 .expect("an empty cursor means the last page");
3400 assert_eq!(empty.next_cursor, None, "an empty cursor ends the walk");
3401 }
3402
3403 #[test]
3404 fn purpose_and_topic_decode_and_a_dm_omits_them() {
3405 let page = decode_conversations_page(
3406 r#"{"ok":true,"channels":[{"id":"C1","name":"general","is_channel":true,
3407 "purpose":{"value":"Deploy chatter","creator":"U1","last_set":1700000000},
3408 "topic":{"value":"release coordination"}}]}"#,
3409 )
3410 .expect("decodes");
3411 let channel = &page.conversations[0];
3412 assert_eq!(
3413 channel.purpose.as_ref().expect("a channel carries its purpose").value,
3414 "Deploy chatter",
3415 );
3416 assert_eq!(
3417 channel.topic.as_ref().expect("a channel carries its topic").value,
3418 "release coordination",
3419 "only the value is read; the metadata stays behind",
3420 );
3421
3422 let dm = decode_conversations_page(r#"{"ok":true,"channels":[{"id":"D1","is_im":true}]}"#)
3423 .expect("decodes");
3424 assert!(
3425 dm.conversations[0].purpose.is_none() && dm.conversations[0].topic.is_none(),
3426 "a DM carries neither object",
3427 );
3428 }
3429
3430 #[test]
3431 fn pins_list_decodes_the_pinned_rows() {
3432 let body = r#"{"ok":true,"items":[
3433 {"type":"message","created":1700000000,"created_by":"U1",
3434 "message":{"type":"message","ts":"1700000000.000100","user":"U2",
3435 "text":"the pinned text"}}]}"#;
3436 let pins = decode_pins(body).expect("decodes");
3437 assert_eq!(pins.len(), 1, "one row per pinned message");
3438 let message = pins[0].message.as_ref().expect("a pin carries its message");
3439 assert_eq!(message.ts, "1700000000.000100", "the ts survives verbatim");
3440 assert_eq!(message.text, "the pinned text");
3441 assert_eq!(message.user.as_deref(), Some("U2"));
3442 assert_eq!(pins[0].created_by.as_deref(), Some("U1"), "who pinned it comes along");
3443 }
3444
3445 #[test]
3446 fn bookmarks_list_decodes_the_saved_links() {
3447 let body = r#"{"ok":true,"bookmarks":[
3448 {"id":"Bk1","type":"link","channel_id":"C1","title":"Runbook",
3449 "link":"https://example.com","date_created":1700000000}]}"#;
3450 let bookmarks = decode_bookmarks(body).expect("decodes");
3451 assert_eq!(bookmarks.len(), 1);
3452 assert_eq!(bookmarks[0].id, "Bk1");
3453 assert_eq!(bookmarks[0].title.as_deref(), Some("Runbook"));
3454 assert_eq!(bookmarks[0].link.as_deref(), Some("https://example.com"));
3455 }
3456
3457 #[test]
3460 fn encode_form_escapes_what_a_message_body_carries() {
3461 let encoded = encode_form(&[("text", "a&b=c+d\ncafé".to_owned())]);
3462 assert_eq!(encoded, "text=a%26b%3Dc%2Bd%0Acaf%C3%A9");
3463 assert_eq!(
3464 encode_form(&[("text", "two words".to_owned())]),
3465 "text=two+words",
3466 "a space is a form `+`, not %20"
3467 );
3468 }
3469
3470 #[test]
3471 fn encode_form_keeps_an_empty_value_and_emits_nothing_for_no_params() {
3472 assert_eq!(encode_form(&[]), "", "no params is an empty body");
3473 assert_eq!(
3474 encode_form(&[("cursor", String::new())]),
3475 "cursor=",
3476 "an empty value still emits its name"
3477 );
3478 }
3479
3480 fn sub_for(target: SlackSubscriptionTarget) -> SlackSubscription {
3481 SlackSubscription {
3482 id: uuid::Uuid::new_v4(),
3483 workspace: "acme".to_owned(),
3484 project: "forge".to_owned(),
3485 team_role: None,
3486 target,
3487 created_at: std::time::SystemTime::UNIX_EPOCH,
3488 }
3489 }
3490
3491 fn conversation_channel(id: &str, name: &str) -> SlackConversation {
3492 SlackConversation {
3493 id: id.to_owned(),
3494 name: Some(name.to_owned()),
3495 is_channel: true,
3496 is_private: false,
3497 is_im: false,
3498 is_mpim: false,
3499 is_archived: false,
3500 user: None,
3501 purpose: None,
3502 topic: None,
3503 }
3504 }
3505
3506 fn conversation_im(id: &str, user: &str) -> SlackConversation {
3507 SlackConversation {
3508 id: id.to_owned(),
3509 name: None,
3510 is_channel: false,
3511 is_private: false,
3512 is_im: true,
3513 is_mpim: false,
3514 is_archived: false,
3515 user: Some(user.to_owned()),
3516 purpose: None,
3517 topic: None,
3518 }
3519 }
3520
3521 #[test]
3522 fn a_mention_is_the_literal_token_and_nothing_looser() {
3523 assert!(mentions("hey <@U123> look", "U123"));
3524 assert!(!mentions("hey U123 look", "U123"), "a bare id is not a mention");
3525 assert!(!mentions("hey <@U1234> look", "U123"), "a longer id must not match");
3526 assert!(mentions("<@U123>", "U123"), "a mention with no surrounding text still counts");
3527 }
3528
3529 fn not_ours(_ts: &str) -> bool {
3531 false
3532 }
3533
3534 fn ours(_ts: &str) -> bool {
3536 true
3537 }
3538
3539 #[test]
3540 fn a_dm_subscription_wants_every_dm() {
3541 let subs = vec![sub_for(SlackSubscriptionTarget::DirectMessages)];
3542 let dm = conversation_im("D1", "U9");
3543 assert!(wants(&subs, &dm, "anything", "100.1", "U1", ¬_ours));
3544 }
3545
3546 #[test]
3547 fn a_mentions_only_channel_filters_on_the_token() {
3548 let subs = vec![sub_for(SlackSubscriptionTarget::Conversation {
3549 id: "C1".to_owned(),
3550 name: None,
3551 mode: SlackWatchMode::MentionsOnly,
3552 })];
3553 let channel = conversation_channel("C1", "general");
3554 assert!(!wants(&subs, &channel, "no mention here", "100.1", "U1", ¬_ours));
3555 assert!(wants(&subs, &channel, "ping <@U1>", "100.1", "U1", ¬_ours));
3556 }
3557
3558 #[test]
3562 fn a_self_authored_message_with_an_unledgered_ts_is_delivered() {
3563 let subs = vec![sub_for(SlackSubscriptionTarget::Conversation {
3564 id: "C1".to_owned(),
3565 name: None,
3566 mode: SlackWatchMode::All,
3567 })];
3568 let channel = conversation_channel("C1", "general");
3569 assert!(
3570 wants(&subs, &channel, "what I typed", "1700000000.000100", "U1", ¬_ours),
3571 "a thread Ved talks in must not read one-sided",
3572 );
3573 }
3574
3575 #[test]
3578 fn a_ledgered_ts_is_refused_however_it_is_authored() {
3579 let subs = vec![sub_for(SlackSubscriptionTarget::Conversation {
3580 id: "C1".to_owned(),
3581 name: None,
3582 mode: SlackWatchMode::All,
3583 })];
3584 let channel = conversation_channel("C1", "general");
3585 assert!(
3586 !wants(&subs, &channel, "the agent's own words", "1700000000.000100", "U1", &ours),
3587 "a post forge made comes back to it unless the ledger refuses it",
3588 );
3589 }
3590
3591 #[test]
3595 fn a_mention_target_is_not_a_conversation_target() {
3596 let subs = vec![sub_for(SlackSubscriptionTarget::Mentions)];
3597 let channel = conversation_channel("C1", "general");
3598 assert!(!wants(&subs, &channel, "no mention here", "100.1", "U1", ¬_ours));
3599 assert!(!wants(&subs, &channel, "a mention <@U1>", "100.1", "U1", ¬_ours));
3600 }
3601
3602 #[test]
3603 fn an_unsubscribed_conversation_is_never_wanted() {
3604 let subs = vec![sub_for(SlackSubscriptionTarget::Conversation {
3605 id: "C1".to_owned(),
3606 name: None,
3607 mode: SlackWatchMode::All,
3608 })];
3609 assert!(!wants(
3610 &subs,
3611 &conversation_channel("C2", "other"),
3612 "hi",
3613 "100.1",
3614 "U1",
3615 ¬_ours
3616 ));
3617 }
3618
3619 #[test]
3620 fn a_history_page_decodes_its_messages_and_the_cursor() {
3621 let body = r#"{
3622 "ok": true,
3623 "messages": [
3624 {"ts":"100.000001","user":"U9","text":"hello","thread_ts":"100.000001","reply_count":3},
3625 {"ts":"200.000002","user":"U8","text":"plain"}
3626 ],
3627 "response_metadata": {"next_cursor": "cur1"}
3628 }"#;
3629 let page = decode_message_page("conversations.history", body).expect("decodes");
3630 assert_eq!(page.messages.len(), 2, "every message in the page is kept");
3631 assert_eq!(page.messages[0].ts, "100.000001", "the ts survives as the string Slack sent");
3632 assert_eq!(page.messages[0].reply_count, 3, "the reply count drives thread following");
3633 assert_eq!(page.messages[0].thread_ts.as_deref(), Some("100.000001"));
3634 assert_eq!(page.messages[1].thread_ts, None, "a top-level message has no thread");
3635 assert_eq!(page.next_cursor.as_deref(), Some("cur1"));
3636
3637 let last = decode_message_page(
3638 "conversations.history",
3639 r#"{"ok":true,"messages":[],"response_metadata":{"next_cursor":""}}"#,
3640 )
3641 .expect("an empty cursor means the last page");
3642 assert_eq!(last.next_cursor, None);
3643 }
3644
3645 #[test]
3651 fn an_edited_message_decodes_the_edit_ts() {
3652 let body = r#"{
3653 "ok": true,
3654 "messages": [
3655 {"ts":"100.000001","user":"U9","text":"the correction",
3656 "edited":{"user":"U9","ts":"1790186900.123456"}},
3657 {"ts":"90.000001","user":"U9","text":"untouched"}
3658 ]
3659 }"#;
3660 let page = decode_message_page("conversations.history", body).expect("decodes");
3661 assert_eq!(
3662 page.messages[0].edited.as_deref(),
3663 Some("1790186900.123456"),
3664 "the edit's own ts is what says the text changed after sending",
3665 );
3666 assert_eq!(page.messages[1].edited, None, "an untouched message carries no edit");
3667 }
3668
3669 #[test]
3673 fn search_decodes_the_user_id_beside_the_username() {
3674 let body = r#"{
3675 "ok": true,
3676 "messages": {
3677 "matches": [
3678 {"ts":"200.1","text":"ping <@U1>","channel":{"id":"C1","name":"general"},
3679 "username":"Forge","user":"U1"}
3680 ],
3681 "next_cursor": ""
3682 }
3683 }"#;
3684 let page = decode_search(body).expect("decodes");
3685 let hit = &page.matches[0];
3686 assert_eq!(hit.user.as_deref(), Some("U1"), "the id is what the own-id rule reads");
3687 assert_eq!(hit.username.as_deref(), Some("Forge"), "and the app's handle rides along");
3688 }
3689
3690 #[test]
3693 fn history_params_carry_each_bound_only_when_it_is_asked_for() {
3694 let bounded = history_params("C1", Some("100.0"), Some("200.0"), 50, Some("cur"));
3695 assert_eq!(
3696 bounded,
3697 vec![
3698 ("channel", "C1".to_owned()),
3699 ("limit", "50".to_owned()),
3700 ("oldest", "100.0".to_owned()),
3701 ("latest", "200.0".to_owned()),
3702 ("cursor", "cur".to_owned()),
3703 ],
3704 "the newest bound travels as `latest`",
3705 );
3706
3707 let bare = history_params("C1", None, None, 50, None);
3708 assert!(
3709 !bare.iter().any(|(key, _)| *key == "oldest" || *key == "latest"),
3710 "an absent bound is omitted entirely: {bare:?}",
3711 );
3712 }
3713
3714 #[test]
3717 fn the_ids_a_draft_tags_are_its_user_mentions_once_each() {
3718 assert_eq!(
3719 mention_ids("thanks <@U2>, <@U3|carol> and <@U2> again"),
3720 vec!["U2", "U3"],
3721 "a labelled mention is its id, and a repeated tag is one person",
3722 );
3723 assert!(
3724 mention_ids("<#C1> <https://example.com|the doc> <@> <@123>").is_empty(),
3725 "a channel, a link, an empty mention and a non-user token tag nobody",
3726 );
3727 }
3728
3729 #[test]
3733 fn a_quoted_excerpt_flattens_its_slack_markup() {
3734 let names = HashMap::from([("U2".to_owned(), "bob".to_owned())]);
3735 assert_eq!(
3736 flatten_mrkdwn("<@U2> and <@U2|carol> and <@U9>", &names, 80),
3737 "@bob and @bob and U9",
3738 "a named mention draws its name; an unnamed one keeps its id",
3739 );
3740 assert_eq!(
3741 flatten_mrkdwn(
3742 "<https://example.com/docs|the docs> and <https://example.com>",
3743 &names,
3744 80,
3745 ),
3746 "the docs and https://example.com",
3747 "a titled link draws its label, and a bare link its address",
3748 );
3749 assert_eq!(
3750 flatten_mrkdwn("<https://example.com|>", &names, 80),
3751 "https://example.com",
3752 "an empty label draws the address",
3753 );
3754 assert_eq!(
3755 flatten_mrkdwn("*<https://example.com|the docs>* and *green*", &names, 80),
3756 "the docs and green",
3757 "a link inside bold flattens too, and bold loses its stars",
3758 );
3759 assert_eq!(
3760 flatten_mrkdwn("src/*.rs and tests/*.rs", &names, 80),
3761 "src/*.rs and tests/*.rs",
3762 "a glob keeps its stars rather than reading as bold",
3763 );
3764 assert_eq!(
3765 flatten_mrkdwn("2 * 3 * 4 and **bold**", &names, 80),
3766 "2 * 3 * 4 and **bold**",
3767 "a spaced star and a doubled marker are literals, not bold",
3768 );
3769 assert_eq!(
3770 flatten_mrkdwn("a < b, and <#C1|general> with <!here>", &names, 80),
3771 "a < b, and <#C1|general> with <!here>",
3772 "a channel reference, a broadcast and a literal bracket keep their own text",
3773 );
3774 assert_eq!(
3775 flatten_mrkdwn("<!here|*x*>", &names, 80),
3776 "<!here|*x*>",
3777 "a labelled broadcast is left whole, markup and all",
3778 );
3779 assert_eq!(
3780 flatten_mrkdwn("<@|broken> and <@123>", &names, 80),
3781 "<@|broken> and <@123>",
3782 "an unrecognised mention keeps its brackets",
3783 );
3784 }
3785
3786 #[test]
3789 fn an_excerpt_is_bounded_after_flattening() {
3790 let names = HashMap::from([("U2".to_owned(), "bob".to_owned())]);
3791 let parent = "*<https://example.com|the staging deploy>* is <@U2> and every check is \
3792 green while the p99 is still high and the canary holds at one percent";
3793 let flat = flatten_mrkdwn(parent, &names, 80);
3794 let excerpt = one_line_excerpt(&flat, 80);
3795 assert!(excerpt.ends_with("..."), "the line is cut: {excerpt:?}");
3796 assert!(excerpt.chars().count() <= 83, "the bound holds after flattening: {excerpt:?}");
3797 assert!(
3798 !excerpt.contains('<') && !excerpt.contains('*'),
3799 "no markup survives the flattening: {excerpt:?}",
3800 );
3801 let kept = excerpt.strip_suffix("...").expect("the mark is there");
3802 assert!(
3803 flat.starts_with(kept) && !kept.ends_with(' '),
3804 "the cut is the flattened opening at a word boundary: {excerpt:?}",
3805 );
3806
3807 let multibyte = format!("*{}*", "ééé ".repeat(30).trim_end());
3810 let flat = flatten_mrkdwn(&multibyte, &names, 80);
3811 let excerpt = one_line_excerpt(&flat, 80);
3812 assert!(excerpt.ends_with("..."), "the glyph run is cut: {excerpt:?}");
3813 assert!(
3814 excerpt.chars().count() <= 83,
3815 "the char bound holds on multibyte text: {excerpt:?}"
3816 );
3817 let kept = excerpt.strip_suffix("...").expect("the mark is there");
3818 assert!(
3819 flat.starts_with(kept) && !kept.ends_with(' '),
3820 "the cut lands on a word boundary in multibyte text: {excerpt:?}",
3821 );
3822 assert!(
3823 kept.chars().count() >= 76,
3824 "the cut keeps the character budget rather than a byte count: {excerpt:?}",
3825 );
3826 assert!(!excerpt.contains('*'), "the bold markers are gone: {excerpt:?}");
3827 }
3828
3829 #[test]
3832 fn the_early_stop_counts_characters_not_bytes() {
3833 let names: HashMap<String, String> = HashMap::new();
3834 let bold = format!("*{}* and more words", "é".repeat(50));
3835 assert!(
3836 flatten_mrkdwn(&bold, &names, 80).ends_with(" and more words"),
3837 "a multibyte bold span that fits the budget does not end the walk",
3838 );
3839 let link = format!("<https://example.com|{}> and more words", "é".repeat(50));
3840 assert!(
3841 flatten_mrkdwn(&link, &names, 80).ends_with(" and more words"),
3842 "a multibyte link label that fits the budget does not end the walk",
3843 );
3844 }
3845
3846 #[test]
3849 fn an_excerpt_is_one_line_cut_at_a_word_boundary() {
3850 assert_eq!(one_line_excerpt("the deploy\n is green", 80), "the deploy is green");
3851 let long = "the canary is holding at 1% and every check is green but the p99 is still high";
3852 let cut = one_line_excerpt(long, 40);
3853 assert!(cut.ends_with("..."), "the cut is marked: {cut:?}");
3854 assert!(cut.chars().count() <= 43, "the cut keeps the bound: {cut:?}");
3855 let kept = cut.strip_suffix("...").expect("the mark is there");
3856 assert!(!kept.ends_with(' '), "no trailing space before the mark: {cut:?}");
3857 assert!(long.starts_with(kept), "the excerpt is the parent's opening words: {cut:?}");
3858 assert_eq!(
3859 one_line_excerpt("abcdefghij", 4),
3860 "abcd...",
3861 "a word with no boundary is cut at the bound",
3862 );
3863 assert_eq!(
3864 one_line_excerpt("one two three four five six", 20),
3865 "one two three four...",
3866 "the cut lands on the word boundary inside the bound",
3867 );
3868 }
3869
3870 #[tokio::test]
3873 async fn a_thread_parent_is_the_first_message_of_the_replies_read() {
3874 let fake = Arc::new(FakeHost::default());
3875 FakeHost::set_api_self(&fake);
3876 fake.seed_replies(
3877 "C1",
3878 "100.0",
3879 vec![
3880 history_message("100.0", "U2", "the parent"),
3881 history_message("200.0", "U3", "a reply"),
3882 ],
3883 );
3884 let api = Arc::clone(&fake) as Arc<dyn SlackApi>;
3885
3886 let parent = thread_parent(api.as_ref(), "C1", "100.0")
3887 .await
3888 .expect("the read lands")
3889 .expect("the thread has a parent");
3890 assert_eq!(parent.ts, "100.0", "the parent is the message the thread starts at");
3891 assert_eq!(parent.text, "the parent");
3892 assert_eq!(parent.user.as_deref(), Some("U2"), "the author id rides the parent");
3893 assert_eq!(
3894 *fake.replies_limits.lock().expect("lock"),
3895 vec![1],
3896 "the card needs the parent alone, so the read asks for one message",
3897 );
3898 }
3899
3900 #[tokio::test]
3903 async fn a_thread_parent_read_that_fails_is_not_a_guess() {
3904 let fake = Arc::new(FakeHost::default());
3905 FakeHost::set_api_self(&fake);
3906 fake.rate_limit_replies_at(0);
3907 let api = Arc::clone(&fake) as Arc<dyn SlackApi>;
3908 assert!(
3909 thread_parent(api.as_ref(), "C1", "100.0").await.is_err(),
3910 "the caller decides what a failed read draws; the connector does not guess",
3911 );
3912 }
3913
3914 #[tokio::test]
3916 async fn a_cached_user_name_answers_without_a_lookup() {
3917 let fake = Arc::new(FakeHost::default());
3918 FakeHost::set_api_self(&fake);
3919 fake.set_user_name("acme", "U2", "bob");
3920 let api = Arc::clone(&fake) as Arc<dyn SlackApi>;
3921
3922 let name = user_display_name(fake.as_ref(), api.as_ref(), "acme", "U2").await;
3923 assert_eq!(name.as_deref(), Some("bob"));
3924 assert_eq!(*fake.user_info_calls.lock().expect("lock"), 0, "the cache is the point");
3925 assert_eq!(
3926 *fake.cache_workspaces.lock().expect("lock"),
3927 vec!["acme".to_owned(), "acme".to_owned()],
3928 "the seeding write and the read both name the draft's workspace",
3929 );
3930 }
3931
3932 #[tokio::test]
3934 async fn a_user_lookup_names_the_person_and_is_remembered() {
3935 let fake = Arc::new(FakeHost::default());
3936 FakeHost::set_api_self(&fake);
3937 fake.seed_user_name("U2", "bob");
3938 let api = Arc::clone(&fake) as Arc<dyn SlackApi>;
3939
3940 assert_eq!(
3941 user_display_name(fake.as_ref(), api.as_ref(), "acme", "U2").await.as_deref(),
3942 Some("bob"),
3943 );
3944 assert_eq!(
3945 user_display_name(fake.as_ref(), api.as_ref(), "acme", "U2").await.as_deref(),
3946 Some("bob"),
3947 );
3948 assert_eq!(*fake.user_info_calls.lock().expect("lock"), 1, "a resolved name is cached");
3949 }
3950
3951 #[tokio::test]
3955 async fn a_failed_user_lookup_is_none_and_remembers_nothing() {
3956 let fake = Arc::new(FakeHost::default());
3957 FakeHost::set_api_self(&fake);
3958 fake.fail_user_info_transiently();
3959 let api = Arc::clone(&fake) as Arc<dyn SlackApi>;
3960
3961 assert_eq!(user_display_name(fake.as_ref(), api.as_ref(), "acme", "U2").await, None);
3962 assert!(
3963 fake.resolved_names().is_empty(),
3964 "a failure is not a name to remember: {:?}",
3965 fake.resolved_names(),
3966 );
3967 }
3968
3969 #[tokio::test]
3972 async fn a_settled_miss_is_remembered_as_no_name() {
3973 let fake = Arc::new(FakeHost::default());
3974 FakeHost::set_api_self(&fake);
3975 let api = Arc::clone(&fake) as Arc<dyn SlackApi>;
3976
3977 assert_eq!(user_display_name(fake.as_ref(), api.as_ref(), "acme", "U404").await, None);
3978 assert_eq!(user_display_name(fake.as_ref(), api.as_ref(), "acme", "U404").await, None);
3979 assert_eq!(*fake.user_info_calls.lock().expect("lock"), 1, "a settled miss costs one call");
3980 assert_eq!(
3981 fake.resolved_names().get("U404").map(String::as_str),
3982 Some(""),
3983 "the empty name is the record that the miss is settled",
3984 );
3985 }
3986
3987 #[test]
3990 fn a_replies_page_keeps_the_repeated_parent_for_the_caller_to_dedupe() {
3991 let body = r#"{
3992 "ok": true,
3993 "messages": [
3994 {"ts":"100.0","user":"U9","text":"root"},
3995 {"ts":"200.1","user":"U8","text":"reply one"},
3996 {"ts":"100.0","user":"U9","text":"root"},
3997 {"ts":"300.2","user":"U8","text":"reply two"}
3998 ]
3999 }"#;
4000 let page = decode_message_page("conversations.replies", body).expect("decodes");
4001 assert_eq!(page.messages.len(), 4, "the decoder does not dedupe");
4002 assert_eq!(
4003 page.messages.iter().filter(|m| m.ts == "100.0").count(),
4004 2,
4005 "the parent arrives twice and the caller is the one that folds it",
4006 );
4007 }
4008
4009 #[test]
4015 fn a_parent_carries_its_latest_reply_and_a_reply_does_not() {
4016 let page = decode_message_page(
4017 "conversations.history",
4018 include_str!("../fixtures/slack_parent.json"),
4019 )
4020 .expect("the captured parent decodes");
4021 let parent = &page.messages[0];
4022 assert_eq!(
4023 parent.latest_reply.as_deref(),
4024 Some("1790186555.957249"),
4025 "a parent names its newest reply",
4026 );
4027 assert_eq!(parent.parent_user_id, None, "a parent carries no parent_user_id");
4028 assert_eq!(
4029 parent.thread_ts.as_deref(),
4030 Some(parent.ts.as_str()),
4031 "a real parent's thread_ts is its own ts, never None",
4032 );
4033 assert_eq!(parent.reply_count, 2, "and it reports how many replies it has");
4034
4035 let replies = decode_message_page(
4036 "conversations.replies",
4037 include_str!("../fixtures/slack_reply.json"),
4038 )
4039 .expect("the captured thread decodes");
4040 let reply = &replies.messages[1];
4041 assert!(reply.parent_user_id.is_some(), "parent_user_id is what marks a reply");
4042 assert_eq!(reply.latest_reply, None, "a reply names no newest reply");
4043 assert_ne!(
4044 reply.thread_ts.as_deref(),
4045 Some(reply.ts.as_str()),
4046 "a reply's thread_ts is the parent's, not its own",
4047 );
4048 }
4049
4050 #[derive(Default)]
4054 struct FakeHost {
4055 subscriptions: std::sync::Mutex<Vec<SlackSubscription>>,
4056 user_id_slot: std::sync::Mutex<Option<String>>,
4057 api_self: std::sync::Mutex<Option<Arc<dyn SlackApi>>>,
4061 rate_limited: Option<Duration>,
4062 conversations: std::sync::Mutex<Vec<SlackConversation>>,
4063 history: std::sync::Mutex<HashMap<String, Vec<Vec<SlackHistoryMessage>>>>,
4064 history_errors: std::sync::Mutex<HashMap<String, SlackError>>,
4066 history_rate_limited_calls: std::sync::Mutex<HashMap<String, usize>>,
4070 search_rate_limited_calls: std::sync::Mutex<usize>,
4073 user_names: std::sync::Mutex<HashMap<String, String>>,
4076 user_info_calls: std::sync::Mutex<usize>,
4077 auth_test_calls: std::sync::Mutex<usize>,
4080 auth_test_transient: std::sync::Mutex<bool>,
4083 auth_test_rate_limited: std::sync::Mutex<Option<Duration>>,
4085 user_info_rate_limited: std::sync::Mutex<Option<Duration>>,
4087 user_info_transient: std::sync::Mutex<bool>,
4090 replies_calls: std::sync::Mutex<usize>,
4093 replies_rate_limit_at: std::sync::Mutex<Option<usize>>,
4095 replies_limits: std::sync::Mutex<Vec<u32>>,
4097 cache_workspaces: std::sync::Mutex<Vec<String>>,
4100 history_calls: std::sync::Mutex<usize>,
4103 resolved_names: std::sync::Mutex<HashMap<String, String>>,
4105 reported_failures: std::sync::Mutex<HashSet<String>>,
4107 replies: std::sync::Mutex<HashMap<String, Vec<Vec<SlackHistoryMessage>>>>,
4108 watermarks: std::sync::Mutex<HashMap<String, String>>,
4109 connected: std::sync::Mutex<Option<bool>>,
4110 delivered: std::sync::Mutex<Vec<SlackMessage>>,
4111 delivered_owners: std::sync::Mutex<Vec<SlackThreadOwner>>,
4114 batches: std::sync::Mutex<Vec<Vec<SlackMessage>>>,
4117 search: std::sync::Mutex<Vec<Vec<SlackSearchMatch>>>,
4118 search_calls: std::sync::Mutex<usize>,
4119 auto_subscribed: std::sync::Mutex<Vec<String>>,
4120 deliver_failures: std::sync::Mutex<std::collections::HashSet<usize>>,
4121 deliver_attempts: std::sync::Mutex<usize>,
4122 threads: std::sync::Mutex<HashMap<String, SlackThreadRecord>>,
4124 thread_full_reads: std::sync::Mutex<HashMap<String, std::time::Instant>>,
4127 thread_first_walks: std::sync::Mutex<HashSet<String>>,
4128 thread_edit_watermarks: std::sync::Mutex<HashMap<String, String>>,
4129 replies_oldest: std::sync::Mutex<Vec<Option<String>>>,
4132 own_posts: std::sync::Mutex<HashSet<String>>,
4134 inflight: std::sync::Mutex<HashSet<String>>,
4137 }
4138
4139 fn conversation_for(id: &str) -> SlackConversation {
4140 let is_im = id.starts_with('D');
4141 SlackConversation {
4142 id: id.to_owned(),
4143 name: (!is_im).then(|| id.to_lowercase()),
4144 is_channel: !is_im,
4145 is_private: false,
4146 is_im,
4147 is_mpim: false,
4148 is_archived: false,
4149 user: is_im.then(|| "U9".to_owned()),
4150 purpose: None,
4151 topic: None,
4152 }
4153 }
4154
4155 impl FakeHost {
4156 fn with_subscriptions(subscriptions: Vec<SlackSubscription>) -> Self {
4157 Self {
4158 subscriptions: std::sync::Mutex::new(subscriptions),
4159 user_id_slot: std::sync::Mutex::new(Some("U1".to_owned())),
4160 ..Self::default()
4161 }
4162 }
4163
4164 fn set_api_self(me: &Arc<Self>) {
4167 *me.api_self.lock().expect("lock") = Some(Arc::clone(me) as Arc<dyn SlackApi>);
4168 }
4169
4170 fn seed_history(&self, channel: &str, messages: Vec<SlackHistoryMessage>) {
4171 self.seed_history_pages(channel, vec![messages]);
4172 }
4173
4174 fn seed_history_unlisted(&self, channel: &str, messages: Vec<SlackHistoryMessage>) {
4178 self.history.lock().expect("lock").insert(channel.to_owned(), vec![messages]);
4179 }
4180
4181 fn seed_history_pages(&self, channel: &str, pages: Vec<Vec<SlackHistoryMessage>>) {
4184 {
4185 let mut conversations = self.conversations.lock().expect("lock");
4186 if !conversations.iter().any(|c| c.id == channel) {
4187 conversations.push(conversation_for(channel));
4188 }
4189 }
4190 self.history.lock().expect("lock").insert(channel.to_owned(), pages);
4191 }
4192
4193 fn seed_replies(&self, channel: &str, parent: &str, messages: Vec<SlackHistoryMessage>) {
4194 self.seed_replies_pages(channel, parent, vec![messages]);
4195 }
4196
4197 fn seed_replies_pages(
4201 &self,
4202 channel: &str,
4203 parent: &str,
4204 pages: Vec<Vec<SlackHistoryMessage>>,
4205 ) {
4206 self.replies.lock().expect("lock").insert(format!("{channel}/{parent}"), pages);
4207 }
4208
4209 fn fail_history_with(&self, channel: &str, error: SlackError) {
4211 self.history_errors.lock().expect("lock").insert(channel.to_owned(), error);
4212 }
4213
4214 fn set_user_id(&self, id: Option<String>) {
4217 *self.user_id_slot.lock().expect("lock") = id;
4218 }
4219
4220 fn fail_with_rate_limit(&mut self, retry_after: Duration) {
4221 self.rate_limited = Some(retry_after);
4222 }
4223
4224 fn rate_limit_history_calls(&self, channel: &str, calls: usize) {
4227 self.history_rate_limited_calls.lock().expect("lock").insert(channel.to_owned(), calls);
4228 }
4229
4230 fn rate_limit_search_calls(&self, calls: usize) {
4232 *self.search_rate_limited_calls.lock().expect("lock") = calls;
4233 }
4234
4235 fn rate_limit_replies_at(&self, call: usize) {
4237 *self.replies_rate_limit_at.lock().expect("lock") = Some(call);
4238 }
4239
4240 fn delivered(&self) -> Vec<SlackMessage> {
4241 self.delivered.lock().expect("lock").clone()
4242 }
4243
4244 fn batches(&self) -> Vec<Vec<SlackMessage>> {
4245 self.batches.lock().expect("lock").clone()
4246 }
4247
4248 fn seed_user_name(&self, user: &str, name: &str) {
4249 self.user_names.lock().expect("lock").insert(user.to_owned(), name.to_owned());
4250 }
4251
4252 fn record_own_post(&self, workspace: &str, conversation: &str, ts: &str) {
4254 self.own_posts.lock().expect("lock").insert(format!("{workspace}/{conversation}/{ts}"));
4255 }
4256
4257 fn set_post_in_flight(&self, workspace: &str, address: &str, in_flight: bool) {
4259 let key = format!("{workspace}/{address}");
4260 let mut marks = self.inflight.lock().expect("lock");
4261 if in_flight {
4262 marks.insert(key);
4263 } else {
4264 marks.remove(&key);
4265 }
4266 }
4267
4268 fn rate_limit_user_info(&self, retry_after: Duration) {
4269 *self.user_info_rate_limited.lock().expect("lock") = Some(retry_after);
4270 }
4271
4272 fn fail_user_info_transiently(&self) {
4273 *self.user_info_transient.lock().expect("lock") = true;
4274 }
4275
4276 fn resolved_names(&self) -> HashMap<String, String> {
4277 self.resolved_names.lock().expect("lock").clone()
4278 }
4279
4280 fn replies_calls(&self) -> usize {
4281 *self.replies_calls.lock().expect("lock")
4282 }
4283
4284 fn replies_oldest(&self) -> Vec<Option<String>> {
4287 self.replies_oldest.lock().expect("lock").clone()
4288 }
4289
4290 fn age_thread_full_read(
4293 &self,
4294 workspace: &str,
4295 conversation: &str,
4296 parent: &str,
4297 age: Duration,
4298 ) {
4299 let at =
4300 std::time::Instant::now().checked_sub(age).expect("the test clock is past the age");
4301 self.thread_full_reads
4302 .lock()
4303 .expect("lock")
4304 .insert(format!("{workspace}/{conversation}/{parent}"), at);
4305 }
4306
4307 fn restart(&self) {
4310 self.thread_full_reads.lock().expect("lock").clear();
4311 self.thread_first_walks.lock().expect("lock").clear();
4312 self.thread_edit_watermarks.lock().expect("lock").clear();
4313 }
4314
4315 fn history_calls(&self) -> usize {
4316 *self.history_calls.lock().expect("lock")
4317 }
4318
4319 fn user_info_calls(&self) -> usize {
4320 *self.user_info_calls.lock().expect("lock")
4321 }
4322
4323 fn auth_test_calls(&self) -> usize {
4324 *self.auth_test_calls.lock().expect("lock")
4325 }
4326
4327 fn fail_auth_test_transiently(&self) {
4328 *self.auth_test_transient.lock().expect("lock") = true;
4329 }
4330
4331 fn rate_limit_auth_test(&self, retry_after: Duration) {
4332 *self.auth_test_rate_limited.lock().expect("lock") = Some(retry_after);
4333 }
4334
4335 fn connected(&self) -> Option<bool> {
4336 *self.connected.lock().expect("lock")
4337 }
4338
4339 fn seed_search(&self, matches: Vec<SlackSearchMatch>) {
4340 self.seed_search_pages(vec![matches]);
4341 }
4342
4343 fn seed_search_pages(&self, pages: Vec<Vec<SlackSearchMatch>>) {
4346 *self.search.lock().expect("lock") = pages;
4347 }
4348
4349 fn count_for(&self, method: &str) -> usize {
4352 match method {
4353 "search.messages" => *self.search_calls.lock().expect("lock"),
4354 other => panic!("the fake does not count {other}"),
4355 }
4356 }
4357
4358 fn auto_subscribed(&self) -> Vec<String> {
4359 self.auto_subscribed.lock().expect("lock").clone()
4360 }
4361
4362 fn fail_delivery_at(&self, index: usize) {
4365 self.deliver_failures.lock().expect("lock").insert(index);
4366 }
4367
4368 fn delivered_owners(&self) -> Vec<SlackThreadOwner> {
4369 self.delivered_owners.lock().expect("lock").clone()
4370 }
4371
4372 fn thread_cursor(
4373 &self,
4374 workspace: &str,
4375 conversation: &str,
4376 parent: &str,
4377 ) -> Option<String> {
4378 self.threads
4379 .lock()
4380 .expect("lock")
4381 .get(&format!("{workspace}/{conversation}/{parent}"))
4382 .map(|record| record.cursor.clone())
4383 }
4384 }
4385
4386 impl SlackHost for FakeHost {
4387 fn client(
4388 &self,
4389 _workspace: &str,
4390 _timeout: Duration,
4391 ) -> Result<Arc<dyn SlackApi>, String> {
4392 let exposed = self.api_self.lock().expect("lock").clone();
4393 exposed.ok_or("the fake exposes no API; call set_api_self first".to_owned())
4394 }
4395
4396 fn user_id(&self, _workspace: &str) -> Option<String> {
4397 self.user_id_slot.lock().expect("lock").clone()
4398 }
4399
4400 fn user_name(&self, workspace: &str, user: &str) -> Option<String> {
4401 self.cache_workspaces.lock().expect("lock").push(workspace.to_owned());
4402 self.resolved_names.lock().expect("lock").get(user).cloned()
4403 }
4404
4405 fn set_user_name(&self, workspace: &str, user: &str, name: &str) {
4406 self.cache_workspaces.lock().expect("lock").push(workspace.to_owned());
4407 self.resolved_names.lock().expect("lock").insert(user.to_owned(), name.to_owned());
4408 }
4409
4410 fn first_author_failure(&self, _workspace: &str, user: &str) -> bool {
4411 self.reported_failures.lock().expect("lock").insert(user.to_owned())
4412 }
4413
4414 fn first_draft_user_failure(&self, _workspace: &str, user: &str) -> bool {
4415 self.reported_failures.lock().expect("lock").insert(user.to_owned())
4416 }
4417
4418 fn subscriptions(&self, _workspace: &str) -> Vec<SlackSubscription> {
4419 self.subscriptions.lock().expect("lock").clone()
4420 }
4421
4422 fn watermark(&self, workspace: &str, conversation: &str) -> Result<Option<String>, String> {
4423 Ok(self
4424 .watermarks
4425 .lock()
4426 .expect("lock")
4427 .get(&format!("{workspace}/{conversation}"))
4428 .cloned())
4429 }
4430
4431 fn set_watermark(&self, workspace: &str, conversation: &str, ts: &str) {
4432 self.watermarks
4433 .lock()
4434 .expect("lock")
4435 .insert(format!("{workspace}/{conversation}"), ts.to_owned());
4436 }
4437
4438 fn set_connected(&self, _workspace: &str, connected: bool) {
4439 *self.connected.lock().expect("lock") = Some(connected);
4440 }
4441
4442 fn name_conversation(&self, workspace: &str, conversation: &str, name: &str) {
4446 for sub in self.subscriptions.lock().expect("lock").iter_mut() {
4447 if sub.workspace != workspace {
4448 continue;
4449 }
4450 if let SlackSubscriptionTarget::Conversation { id, name: stored, .. } =
4451 &mut sub.target
4452 && id.as_str() == conversation
4453 {
4454 *stored = Some(name.to_owned());
4455 }
4456 }
4457 }
4458
4459 fn deliver(&self, subscription: &SlackSubscription, messages: &[SlackMessage]) -> bool {
4460 let attempt = {
4463 let mut attempts = self.deliver_attempts.lock().expect("lock");
4464 *attempts += 1;
4465 *attempts - 1
4466 };
4467 if self.deliver_failures.lock().expect("lock").contains(&attempt) {
4468 return false;
4469 }
4470 let owner = SlackThreadOwner {
4471 project: subscription.project.clone(),
4472 team_role: subscription.team_role.clone(),
4473 };
4474 self.batches.lock().expect("lock").push(messages.to_vec());
4475 for message in messages {
4476 self.delivered.lock().expect("lock").push(message.clone());
4477 self.delivered_owners.lock().expect("lock").push(owner.clone());
4478 }
4479 true
4480 }
4481
4482 fn followed_threads(
4483 &self,
4484 workspace: &str,
4485 conversation: &str,
4486 ) -> Vec<SlackFollowedThread> {
4487 let prefix = format!("{workspace}/{conversation}/");
4488 self.threads
4489 .lock()
4490 .expect("lock")
4491 .iter()
4492 .filter(|(key, _)| key.starts_with(&prefix))
4493 .map(|(key, record)| SlackFollowedThread {
4494 parent_ts: key
4495 .rsplit('/')
4496 .next()
4497 .expect("the parent is the last part")
4498 .to_owned(),
4499 owners: record.owners.clone(),
4500 participating: record.participating,
4501 })
4502 .collect()
4503 }
4504
4505 fn thread_watermark(
4506 &self,
4507 workspace: &str,
4508 conversation: &str,
4509 parent_ts: &str,
4510 ) -> Result<Option<String>, String> {
4511 Ok(self
4512 .threads
4513 .lock()
4514 .expect("lock")
4515 .get(&format!("{workspace}/{conversation}/{parent_ts}"))
4516 .map(|record| record.cursor.clone()))
4517 }
4518
4519 fn set_thread_watermark(
4520 &self,
4521 workspace: &str,
4522 conversation: &str,
4523 parent_ts: &str,
4524 ts: &str,
4525 ) {
4526 let mut threads = self.threads.lock().expect("lock");
4527 let record = threads
4528 .entry(format!("{workspace}/{conversation}/{parent_ts}"))
4529 .or_insert_with(|| SlackThreadRecord {
4530 cursor: String::new(),
4531 edit_watermark: None,
4532 owners: Vec::new(),
4533 participating: false,
4534 });
4535 record.cursor = ts.to_owned();
4536 }
4537
4538 fn mark_thread_participating(&self, workspace: &str, conversation: &str, parent_ts: &str) {
4539 let mut threads = self.threads.lock().expect("lock");
4540 if let Some(record) =
4541 threads.get_mut(&format!("{workspace}/{conversation}/{parent_ts}"))
4542 {
4543 record.participating = true;
4544 }
4545 }
4546
4547 fn thread_full_read_at(
4548 &self,
4549 workspace: &str,
4550 conversation: &str,
4551 parent_ts: &str,
4552 ) -> Option<std::time::Instant> {
4553 self.thread_full_reads
4554 .lock()
4555 .expect("lock")
4556 .get(&format!("{workspace}/{conversation}/{parent_ts}"))
4557 .copied()
4558 }
4559
4560 fn mark_thread_full_read(
4561 &self,
4562 workspace: &str,
4563 conversation: &str,
4564 parent_ts: &str,
4565 at: std::time::Instant,
4566 ) {
4567 self.thread_full_reads
4568 .lock()
4569 .expect("lock")
4570 .insert(format!("{workspace}/{conversation}/{parent_ts}"), at);
4571 }
4572
4573 fn thread_first_walk(&self, workspace: &str, conversation: &str, parent_ts: &str) -> bool {
4574 self.thread_first_walks
4575 .lock()
4576 .expect("lock")
4577 .contains(&format!("{workspace}/{conversation}/{parent_ts}"))
4578 }
4579
4580 fn clear_thread_first_walk(&self, workspace: &str, conversation: &str, parent_ts: &str) {
4581 self.thread_first_walks
4582 .lock()
4583 .expect("lock")
4584 .remove(&format!("{workspace}/{conversation}/{parent_ts}"));
4585 }
4586
4587 fn thread_edit_watermark(
4588 &self,
4589 workspace: &str,
4590 conversation: &str,
4591 parent_ts: &str,
4592 ) -> Option<String> {
4593 let key = format!("{workspace}/{conversation}/{parent_ts}");
4594 if let Some(mark) = self.thread_edit_watermarks.lock().expect("lock").get(&key).cloned()
4595 {
4596 return Some(mark);
4597 }
4598 self.threads.lock().expect("lock").get(&key).and_then(|record| {
4603 record.edit_watermark.clone().or_else(|| Some(record.cursor.clone()))
4604 })
4605 }
4606
4607 fn set_thread_edit_watermark(
4608 &self,
4609 workspace: &str,
4610 conversation: &str,
4611 parent_ts: &str,
4612 ts: &str,
4613 ) {
4614 let key = format!("{workspace}/{conversation}/{parent_ts}");
4615 self.thread_edit_watermarks.lock().expect("lock").insert(key.clone(), ts.to_owned());
4616 if let Some(record) = self.threads.lock().expect("lock").get_mut(&key) {
4617 record.edit_watermark = Some(ts.to_owned());
4618 }
4619 }
4620
4621 fn follow_thread(
4622 &self,
4623 workspace: &str,
4624 conversation: &str,
4625 parent_ts: &str,
4626 owner: SlackThreadOwner,
4627 since: &str,
4628 ) {
4629 let key = format!("{workspace}/{conversation}/{parent_ts}");
4630 let created = {
4631 let mut threads = self.threads.lock().expect("lock");
4632 let entry = threads.entry(key.clone());
4633 let created = matches!(&entry, std::collections::hash_map::Entry::Vacant(_));
4634 let record = entry.or_insert_with(|| {
4635 SlackThreadRecord {
4639 cursor: since.to_owned(),
4640 edit_watermark: Some(since.to_owned()),
4641 owners: Vec::new(),
4642 participating: false,
4643 }
4644 });
4645 if !record.owners.contains(&owner) {
4646 record.owners.push(owner);
4647 }
4648 created
4649 };
4650 if created {
4651 self.thread_first_walks.lock().expect("lock").insert(key.clone());
4652 self.thread_edit_watermarks.lock().expect("lock").insert(key, since.to_owned());
4653 }
4654 }
4655
4656 fn auto_subscribe(&self, workspace: &str, message: &SlackMessage) -> bool {
4657 let already =
4658 self.subscriptions.lock().expect("lock").iter().any(|sub| match &sub.target {
4659 SlackSubscriptionTarget::Conversation { id, .. } => id == &message.conversation,
4660 SlackSubscriptionTarget::DirectMessages | SlackSubscriptionTarget::Mentions => {
4661 false
4662 }
4663 });
4664 if already {
4665 return false;
4666 }
4667 self.watermarks
4670 .lock()
4671 .expect("lock")
4672 .insert(format!("{}/{}", workspace, message.conversation), message.ts.clone());
4673 self.subscriptions.lock().expect("lock").push(SlackSubscription {
4674 id: uuid::Uuid::new_v4(),
4675 workspace: workspace.to_owned(),
4676 project: "forge".to_owned(),
4677 team_role: None,
4678 target: SlackSubscriptionTarget::Conversation {
4679 id: message.conversation.clone(),
4680 name: None,
4681 mode: SlackWatchMode::All,
4682 },
4683 created_at: std::time::SystemTime::UNIX_EPOCH,
4684 });
4685 self.auto_subscribed.lock().expect("lock").push(message.conversation.clone());
4686 true
4687 }
4688
4689 fn own_post(&self, workspace: &str, conversation: &str, ts: &str) -> bool {
4690 self.own_posts
4691 .lock()
4692 .expect("lock")
4693 .contains(&format!("{workspace}/{conversation}/{ts}"))
4694 }
4695
4696 fn post_in_flight(&self, workspace: &str, conversation: &str) -> Option<Duration> {
4697 self.inflight
4698 .lock()
4699 .expect("lock")
4700 .contains(&format!("{workspace}/{conversation}"))
4701 .then(Duration::default)
4702 }
4703 }
4704
4705 #[async_trait::async_trait]
4706 impl SlackApi for FakeHost {
4707 async fn auth_test(&self) -> Result<AuthTest, SlackError> {
4708 *self.auth_test_calls.lock().expect("lock") += 1;
4709 if let Some(retry_after) = *self.auth_test_rate_limited.lock().expect("lock") {
4710 return Err(SlackError::RateLimited {
4711 method: "auth.test".to_owned(),
4712 retry_after,
4713 });
4714 }
4715 if *self.auth_test_transient.lock().expect("lock") {
4716 return Err(SlackError::Transport {
4717 method: "auth.test".to_owned(),
4718 detail: "the socket is down".to_owned(),
4719 });
4720 }
4721 Ok(AuthTest {
4722 team: "Test".to_owned(),
4723 user: "tester".to_owned(),
4724 team_id: "T1".to_owned(),
4725 user_id: self.user_id_slot.lock().expect("lock").clone().unwrap_or_default(),
4726 url: "https://test.slack.com/".to_owned(),
4727 })
4728 }
4729
4730 async fn list_conversations(&self) -> Result<Vec<SlackConversation>, SlackError> {
4731 if let Some(retry_after) = self.rate_limited {
4732 return Err(SlackError::RateLimited {
4733 method: "users.conversations".to_owned(),
4734 retry_after,
4735 });
4736 }
4737 Ok(self.conversations.lock().expect("lock").clone())
4738 }
4739
4740 async fn history(
4741 &self,
4742 channel: &str,
4743 oldest: Option<&str>,
4744 _newest: Option<&str>,
4745 _limit: u32,
4746 cursor: Option<&str>,
4747 ) -> Result<MessagePage, SlackError> {
4748 *self.history_calls.lock().expect("lock") += 1;
4749 {
4750 let mut calls = self.history_rate_limited_calls.lock().expect("lock");
4751 if let Some(left) = calls.get_mut(channel)
4752 && *left > 0
4753 {
4754 *left -= 1;
4755 return Err(SlackError::RateLimited {
4756 method: "conversations.history".to_owned(),
4757 retry_after: Duration::from_secs(1),
4758 });
4759 }
4760 }
4761 if let Some(retry_after) = self.rate_limited {
4762 return Err(SlackError::RateLimited {
4763 method: "conversations.history".to_owned(),
4764 retry_after,
4765 });
4766 }
4767 if let Some(error) = self.history_errors.lock().expect("lock").get(channel) {
4768 return Err(error.clone());
4769 }
4770 let pages = self.history.lock().expect("lock");
4771 let Some(seeded) = pages.get(channel) else {
4772 return Ok(MessagePage { messages: Vec::new(), next_cursor: None });
4773 };
4774 let index = cursor.and_then(|cursor| cursor.parse::<usize>().ok()).unwrap_or(0);
4775 let messages = seeded.get(index).cloned().unwrap_or_default();
4780 let messages = match oldest {
4781 Some(oldest) => {
4782 messages.into_iter().filter(|message| message.ts.as_str() > oldest).collect()
4783 }
4784 None => messages,
4785 };
4786 Ok(MessagePage {
4787 messages,
4788 next_cursor: (index + 1 < seeded.len()).then(|| (index + 1).to_string()),
4789 })
4790 }
4791
4792 async fn replies(
4793 &self,
4794 channel: &str,
4795 ts: &str,
4796 oldest: Option<&str>,
4797 limit: u32,
4798 cursor: Option<&str>,
4799 ) -> Result<MessagePage, SlackError> {
4800 self.replies_limits.lock().expect("lock").push(limit);
4801 let call = {
4802 let mut calls = self.replies_calls.lock().expect("lock");
4803 let call = *calls;
4804 *calls += 1;
4805 call
4806 };
4807 if *self.replies_rate_limit_at.lock().expect("lock") == Some(call) {
4808 return Err(SlackError::RateLimited {
4809 method: "conversations.replies".to_owned(),
4810 retry_after: Duration::from_secs(5),
4811 });
4812 }
4813 self.replies_oldest.lock().expect("lock").push(oldest.map(str::to_owned));
4814 let pages = self.replies.lock().expect("lock");
4815 let Some(seeded) = pages.get(&format!("{channel}/{ts}")) else {
4816 return Ok(MessagePage { messages: Vec::new(), next_cursor: None });
4817 };
4818 let index = cursor.and_then(|cursor| cursor.parse::<usize>().ok()).unwrap_or(0);
4819 let messages = seeded.get(index).cloned().unwrap_or_default();
4823 let messages = match oldest {
4824 Some(oldest) => {
4825 messages.into_iter().filter(|message| message.ts.as_str() > oldest).collect()
4826 }
4827 None => messages,
4828 };
4829 Ok(MessagePage {
4830 messages,
4831 next_cursor: (index + 1 < seeded.len()).then(|| (index + 1).to_string()),
4832 })
4833 }
4834
4835 async fn post_message(
4836 &self,
4837 _channel: &str,
4838 _text: &str,
4839 _thread_ts: Option<&str>,
4840 ) -> Result<PostedMessage, SlackError> {
4841 Ok(PostedMessage { channel: None, ts: "1.0".to_owned() })
4842 }
4843
4844 async fn update_message(
4845 &self,
4846 _channel: &str,
4847 _ts: &str,
4848 _text: &str,
4849 ) -> Result<(), SlackError> {
4850 Ok(())
4851 }
4852
4853 async fn delete_message(&self, _channel: &str, _ts: &str) -> Result<(), SlackError> {
4854 Ok(())
4855 }
4856
4857 async fn set_reaction(
4858 &self,
4859 _channel: &str,
4860 _ts: &str,
4861 _name: &str,
4862 _add: bool,
4863 ) -> Result<(), SlackError> {
4864 Ok(())
4865 }
4866
4867 async fn get_bytes(&self, _url: &str) -> Result<Vec<u8>, SlackError> {
4868 Ok(Vec::new())
4869 }
4870
4871 async fn post_bytes(&self, _url: &str, _body: Vec<u8>) -> Result<(), SlackError> {
4872 Ok(())
4873 }
4874
4875 async fn file_info(&self, _id: &str) -> Result<SlackFile, SlackError> {
4876 Ok(SlackFile {
4877 id: String::new(),
4878 name: String::new(),
4879 url_private: String::new(),
4880 shares: SlackFileShares::default(),
4881 })
4882 }
4883
4884 async fn get_upload_url(
4885 &self,
4886 _name: &str,
4887 _length: usize,
4888 ) -> Result<(String, String), SlackError> {
4889 Ok((String::new(), String::new()))
4890 }
4891
4892 async fn complete_upload(
4893 &self,
4894 _file_id: &str,
4895 _channel: &str,
4896 _thread_ts: Option<&str>,
4897 ) -> Result<Vec<SlackFile>, SlackError> {
4898 Ok(Vec::new())
4899 }
4900
4901 async fn search_messages(
4902 &self,
4903 query: &str,
4904 _count: u32,
4905 cursor: Option<&str>,
4906 ) -> Result<SearchPage, SlackError> {
4907 *self.search_calls.lock().expect("lock") += 1;
4908 {
4909 let mut calls = self.search_rate_limited_calls.lock().expect("lock");
4910 if *calls > 0 {
4911 *calls -= 1;
4912 return Err(SlackError::RateLimited {
4913 method: "search.messages".to_owned(),
4914 retry_after: Duration::from_secs(1),
4915 });
4916 }
4917 }
4918 let seeded = self.search.lock().expect("lock");
4919 let index = cursor.and_then(|cursor| cursor.parse::<usize>().ok()).unwrap_or(0);
4920 let _ = query;
4921 let page = seeded.get(index).cloned().unwrap_or_default();
4922 Ok(SearchPage {
4923 matches: page,
4924 next_cursor: (index + 1 < seeded.len()).then(|| (index + 1).to_string()),
4925 })
4926 }
4927
4928 async fn user_info(&self, user: &str) -> Result<SlackUser, SlackError> {
4929 *self.user_info_calls.lock().expect("lock") += 1;
4930 if let Some(retry_after) = *self.user_info_rate_limited.lock().expect("lock") {
4931 return Err(SlackError::RateLimited {
4932 method: "users.info".to_owned(),
4933 retry_after,
4934 });
4935 }
4936 if *self.user_info_transient.lock().expect("lock") {
4937 return Err(SlackError::Transport {
4938 method: "users.info".to_owned(),
4939 detail: "connection reset".to_owned(),
4940 });
4941 }
4942 let name = self.user_names.lock().expect("lock").get(user).cloned();
4943 let Some(name) = name else {
4944 return Err(SlackError::Api {
4945 method: "users.info".to_owned(),
4946 error: "user_not_found".to_owned(),
4947 needed: None,
4948 });
4949 };
4950 Ok(SlackUser {
4951 id: user.to_owned(),
4952 name,
4953 real_name: None,
4954 tz: None,
4955 deleted: false,
4956 profile: None,
4957 })
4958 }
4959
4960 async fn list_users(&self) -> Result<Vec<SlackUser>, SlackError> {
4961 Ok(Vec::new())
4962 }
4963
4964 async fn pins(&self, _channel: &str) -> Result<Vec<SlackPin>, SlackError> {
4965 Ok(Vec::new())
4966 }
4967
4968 async fn bookmarks(&self, _channel: &str) -> Result<Vec<SlackBookmark>, SlackError> {
4969 Ok(Vec::new())
4970 }
4971 }
4972
4973 #[test]
4974 fn a_search_always_asks_for_timestamp_order() {
4975 let params = search_params("granite", 20);
4978 assert_eq!(
4979 params.iter().find(|(key, _)| *key == "sort").map(|(_, value)| value.as_str()),
4980 Some("timestamp"),
4981 );
4982 }
4983
4984 #[test]
4988 fn a_search_hit_carries_its_files() {
4989 let body = r#"{"ok":true,"messages":{"matches":[
4990 {"ts":"100.1","text":"","channel":{"id":"C1"},"user":"U9",
4991 "files":[{"id":"F1","name":"notes.txt","url_private":"https://files/x"}]}]}}"#;
4992 let page = decode_search(body).expect("decodes");
4993 assert_eq!(page.matches.len(), 1);
4994 assert_eq!(page.matches[0].files.len(), 1, "the file rides on the hit");
4995 assert_eq!(page.matches[0].files[0].id, "F1");
4996 }
4997
4998 #[test]
4999 fn a_search_decodes_matches_and_their_channel_names() {
5000 let body = r#"{"ok":true,"messages":{"total":2,"matches":[
5001 {"ts":"100.1","text":"hello","channel":{"id":"C1","name":"general"},"username":"ved"},
5002 {"ts":"200.2","text":"world","channel":{"id":"D1","is_im":true},"username":"other"}]}}"#;
5003 let page = decode_search(body).expect("decodes");
5004 let matches = &page.matches;
5005 assert_eq!(matches.len(), 2);
5006 assert_eq!(matches[0].conversation_id, "C1");
5007 assert_eq!(matches[0].conversation_name.as_deref(), Some("general"));
5008 assert_eq!(matches[0].username.as_deref(), Some("ved"));
5009 assert_eq!(matches[1].conversation_name, None, "a DM has no name");
5010 assert_eq!(page.next_cursor, None);
5011 }
5012
5013 #[test]
5014 fn a_user_lookup_decodes_the_display_name() {
5015 let body = r#"{"ok":true,"user":{"id":"U1","name":"ved","real_name":"Vedhavyas S","tz":"Asia/Kolkata"}}"#;
5016 let user = decode_user(body).expect("decodes");
5017 assert_eq!(user.name, "ved");
5018 assert_eq!(user.tz.as_deref(), Some("Asia/Kolkata"));
5019 }
5020
5021 #[test]
5025 fn a_users_page_decodes_a_members_display_name_and_deleted_flag() {
5026 let body = r#"{
5027 "ok": true,
5028 "members": [
5029 {"id":"U1","name":"ved","real_name":"Vedhavyas S",
5030 "profile":{"display_name":"ved"}},
5031 {"id":"U2","name":"gina","deleted":true,
5032 "profile":{"display_name":"Gina Abrams"}}
5033 ],
5034 "response_metadata": {"next_cursor": "next"}
5035 }"#;
5036 let page = decode_users_page(body).expect("decodes");
5037 assert_eq!(page.members.len(), 2, "both rows survive the decode");
5038 assert_eq!(
5039 page.members[0].profile.as_ref().and_then(|p| p.display_name.as_deref()),
5040 Some("ved"),
5041 "the display name is what people type and the handle is not",
5042 );
5043 assert!(!page.members[0].deleted);
5044 assert!(page.members[1].deleted, "a deactivated member is flagged, not dropped");
5045 assert_eq!(page.next_cursor.as_deref(), Some("next"), "the walk can continue");
5046 }
5047
5048 #[derive(Default)]
5051 struct EndlessUsers {
5052 pages: std::sync::atomic::AtomicUsize,
5053 }
5054
5055 #[async_trait::async_trait]
5056 impl UsersPages for EndlessUsers {
5057 async fn users_page(&self, _cursor: Option<&str>) -> Result<UsersPage, SlackError> {
5058 self.pages.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
5059 Ok(UsersPage {
5060 members: vec![SlackUser {
5061 id: "U1".to_owned(),
5062 name: "ved".to_owned(),
5063 real_name: None,
5064 tz: None,
5065 deleted: false,
5066 profile: None,
5067 }],
5068 next_cursor: Some("more".to_owned()),
5069 })
5070 }
5071 }
5072
5073 #[tokio::test]
5076 async fn a_user_walk_stops_at_its_page_cap_and_keeps_what_it_read() {
5077 let source = EndlessUsers::default();
5078
5079 let users = walk_users(&source).await.expect("the cap ends the walk without failing");
5080
5081 assert_eq!(users.len(), MAX_USER_PAGES, "every page read is kept");
5082 assert_eq!(
5083 source.pages.load(std::sync::atomic::Ordering::SeqCst),
5084 MAX_USER_PAGES,
5085 "the walk stopped at the cap instead of following the cursor forever",
5086 );
5087 }
5088
5089 #[test]
5090 fn a_file_lookup_decodes_its_private_url() {
5091 let body = r#"{"ok":true,"file":{"id":"F1","name":"notes.txt","url_private":"https://files.slack.com/x"}}"#;
5092 let file = decode_file(body).expect("decodes");
5093 assert_eq!(file.name, "notes.txt");
5094 assert_eq!(file.url_private, "https://files.slack.com/x");
5095 }
5096
5097 #[test]
5101 fn a_files_shares_name_the_messages_it_was_posted_as() {
5102 let body = r#"{"ok":true,"file":{"id":"F1","name":"notes.txt",
5103 "shares":{
5104 "public":{"C1":[{"ts":"1700000000.000100"}]},
5105 "private":{"D2":[{"ts":"1700000000.000200"}]}}}}"#;
5106 let file = decode_file(body).expect("decodes");
5107 assert_eq!(
5108 shared_messages(&file),
5109 vec![
5110 ("C1".to_owned(), "1700000000.000100".to_owned()),
5111 ("D2".to_owned(), "1700000000.000200".to_owned()),
5112 ],
5113 "both visibility groups name their own conversation's message",
5114 );
5115
5116 let unshared =
5117 decode_file(r#"{"ok":true,"file":{"id":"F2","shares":null}}"#).expect("decodes");
5118 assert!(
5119 shared_messages(&unshared).is_empty(),
5120 "an explicit null shares is no shares, not a decode failure",
5121 );
5122 }
5123
5124 #[test]
5125 fn a_draft_at_the_limit_is_one_post_and_a_longer_one_is_split() {
5126 let exact = "a".repeat(POST_LIMIT);
5127 assert_eq!(split_for_post(&exact).len(), 1, "a draft that fits posts as one message");
5128
5129 let over = "a".repeat(POST_LIMIT + 1);
5130 let parts = split_for_post(&over);
5131 assert!(parts.len() > 1, "one character over must be split, not truncated: {parts:?}");
5132 assert!(
5133 parts.iter().all(|part| part.chars().count() <= POST_LIMIT),
5134 "no part may reach the point Slack truncates at",
5135 );
5136 assert_eq!(
5137 parts.iter().map(|part| part.matches('a').count()).sum::<usize>(),
5138 POST_LIMIT + 1,
5139 "splitting must carry every character",
5140 );
5141 }
5142
5143 #[derive(Default, Debug)]
5146 struct SeenRequest {
5147 authorization: Option<String>,
5148 body: Vec<u8>,
5149 }
5150
5151 fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option<usize> {
5152 haystack.windows(needle.len()).position(|window| window == needle)
5153 }
5154
5155 fn content_length(head: &str) -> Option<usize> {
5159 head.lines()
5160 .filter_map(|line| line.split_once(':'))
5161 .find(|(name, _)| name.trim().eq_ignore_ascii_case("content-length"))
5162 .and_then(|(_, value)| value.trim().parse().ok())
5163 }
5164
5165 fn header_value(head: &str, wanted: &str) -> Option<String> {
5166 head.lines()
5167 .filter_map(|line| line.split_once(':'))
5168 .find(|(name, _)| name.trim().eq_ignore_ascii_case(wanted))
5169 .map(|(_, value)| value.trim().to_owned())
5170 }
5171
5172 fn spawn_one_shot_server(
5177 status: &'static str,
5178 response_body: &'static str,
5179 ) -> (String, std::thread::JoinHandle<SeenRequest>) {
5180 use std::io::{Read, Write};
5181
5182 let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind a port");
5183 let addr = listener.local_addr().expect("local addr");
5184 let handle = std::thread::spawn(move || {
5185 let (mut stream, _) = listener.accept().expect("accept one connection");
5186 let mut raw = Vec::new();
5187 let mut chunk = [0u8; 1024];
5188 loop {
5189 let read = stream.read(&mut chunk).expect("read the request");
5190 if read == 0 {
5191 break;
5192 }
5193 raw.extend_from_slice(&chunk[..read]);
5194 if let Some(end) = find_subslice(&raw, b"\r\n\r\n") {
5195 let head = String::from_utf8_lossy(&raw[..end]).to_string();
5196 let declared = content_length(&head).unwrap_or(0);
5197 if raw.len() >= end + 4 + declared {
5198 break;
5199 }
5200 }
5201 }
5202
5203 let end = find_subslice(&raw, b"\r\n\r\n").expect("a complete request");
5204 let head = String::from_utf8_lossy(&raw[..end]).to_string();
5205 let authorization = header_value(&head, "authorization");
5206 let body = raw[end + 4..].to_vec();
5207
5208 let response = format!(
5209 "HTTP/1.1 {status}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{response_body}",
5210 response_body.len()
5211 );
5212 let _ = stream.write_all(response.as_bytes());
5213 let _ = stream.flush();
5214 SeenRequest { authorization, body }
5215 });
5216 (format!("http://{addr}/file"), handle)
5217 }
5218
5219 #[tokio::test]
5220 async fn get_bytes_sends_the_bearer_header() {
5221 let (url, seen) = spawn_one_shot_server("200 OK", "body-bytes");
5222 let client = SlackClient::new(reqwest::Client::new(), "xoxp-test".to_owned());
5223
5224 let got = client.get_bytes(&url).await.expect("fetch");
5225 assert_eq!(got, b"body-bytes");
5226 assert_eq!(
5227 seen.join().expect("server thread").authorization.as_deref(),
5228 Some("Bearer xoxp-test"),
5229 "a private file URL is a 403 without this header",
5230 );
5231 }
5232
5233 #[tokio::test]
5234 async fn a_non_success_download_is_an_error_not_an_empty_body() {
5235 let (url, seen) = spawn_one_shot_server("403 Forbidden", "");
5236 let client = SlackClient::new(reqwest::Client::new(), "xoxp-test".to_owned());
5237
5238 assert!(
5239 client.get_bytes(&url).await.is_err(),
5240 "a 403 must not be mistaken for a zero-byte file",
5241 );
5242 seen.join().expect("server thread");
5243 }
5244
5245 #[tokio::test]
5246 async fn post_bytes_sends_the_body_unchanged_and_no_bearer() {
5247 let (url, seen) = spawn_one_shot_server("200 OK", "OK - 5");
5248 let client = SlackClient::new(reqwest::Client::new(), "xoxp-test".to_owned());
5249
5250 client.post_bytes(&url, b"hello".to_vec()).await.expect("upload");
5251 let seen = seen.join().expect("server thread");
5252 assert_eq!(seen.body, b"hello");
5253 assert_eq!(
5254 seen.authorization, None,
5255 "the upload URL is pre-signed and must not carry the bearer",
5256 );
5257 }
5258
5259 fn history_message(ts: &str, user: &str, text: &str) -> SlackHistoryMessage {
5260 SlackHistoryMessage {
5261 ts: ts.to_owned(),
5262 user: Some(user.to_owned()),
5263 text: text.to_owned(),
5264 thread_ts: None,
5265 reply_count: 0,
5266 bot_name: None,
5267 parent_user_id: None,
5268 latest_reply: None,
5269 edited: None,
5270 files: Vec::new(),
5271 }
5272 }
5273
5274 fn parent_with_replies(ts: &str, replies: u32) -> SlackHistoryMessage {
5281 SlackHistoryMessage {
5282 thread_ts: Some(ts.to_owned()),
5283 reply_count: replies,
5284 ..history_message(ts, "U9", "trigger")
5285 }
5286 }
5287
5288 fn parent_with_latest_reply(ts: &str, replies: u32, latest: &str) -> SlackHistoryMessage {
5292 SlackHistoryMessage {
5293 latest_reply: Some(latest.to_owned()),
5294 ..parent_with_replies(ts, replies)
5295 }
5296 }
5297
5298 fn reply(ts: &str, parent: &str, user: &str, text: &str) -> SlackHistoryMessage {
5301 SlackHistoryMessage {
5302 thread_ts: Some(parent.to_owned()),
5303 parent_user_id: Some("U9".to_owned()),
5304 ..history_message(ts, user, text)
5305 }
5306 }
5307
5308 #[test]
5311 fn a_parent_with_replies_is_selected_for_its_thread() {
5312 assert!(
5313 is_thread_parent(&parent_with_replies("1.0", 2)),
5314 "a real parent must be selected, or its replies are never fetched",
5315 );
5316 assert!(
5317 !is_thread_parent(&history_message("1.0", "U9", "plain")),
5318 "a message with no replies starts no thread",
5319 );
5320 assert!(
5321 !is_thread_parent(&reply("1.1", "1.0", "U8", "a reply")),
5322 "a reply is not a parent, however it is spelled",
5323 );
5324 }
5325
5326 #[tokio::test]
5331 async fn a_first_sight_dm_baselines_from_one_read() {
5332 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
5333 host.seed_history_pages(
5334 "D1",
5335 vec![
5336 vec![history_message("200.0", "U9", "newest")],
5337 vec![history_message("100.0", "U9", "older")],
5338 ],
5339 );
5340
5341 let outcome = sweep(&host, &host, "acme").await.expect("sweep");
5342
5343 assert_eq!(outcome.delivered, 0, "a first sight delivers nothing");
5344 assert_eq!(host.history_calls(), 1, "one read, not the whole walk");
5345 assert_eq!(
5346 host.watermark("acme", "D1").expect("watermark"),
5347 Some("200.0".to_owned()),
5348 "and the cursor starts at the newest message seen",
5349 );
5350 }
5351
5352 #[tokio::test]
5357 async fn the_dm_class_is_read_on_its_own_clock() {
5358 let host = Arc::new(FakeHost::with_subscriptions(vec![sub_dm("acme")]));
5359 FakeHost::set_api_self(&host);
5360 host.set_watermark("acme", "D1", "050.0");
5361 host.seed_history(
5362 "D1",
5363 vec![history_message("200.0", "U9", "a dm"), history_message("050.0", "U9", "old")],
5366 );
5367
5368 let (tx, rx) = oneshot::channel();
5369 let pump = tokio::spawn(run_workspace_pump(host.clone(), "acme".to_owned(), 1, 100, rx));
5370 for _ in 0..60 {
5371 tokio::time::sleep(Duration::from_millis(50)).await;
5372 if !host.delivered().is_empty() {
5373 break;
5374 }
5375 }
5376 assert_eq!(
5377 host.delivered().first().map(|message| message.ts.as_str()),
5378 Some("200.0"),
5379 "the first DM pass runs without waiting for its interval",
5380 );
5381 let after_first = host.history_calls();
5382 assert_eq!(after_first, 1, "one pass, one read");
5383
5384 tokio::time::sleep(Duration::from_millis(2_500)).await;
5387 assert_eq!(host.history_calls(), after_first, "the DM pass does not ride the fast clock");
5388
5389 tx.send(()).expect("signal shutdown");
5390 let _ = pump.await;
5391 }
5392
5393 #[test]
5398 fn just_below_borrows_a_second_at_a_zero_fraction() {
5399 assert!(
5400 "1790186555.000000" > "1790186555.0",
5401 "the trap this exists for: the padded form sorts after the short one",
5402 );
5403 for ts in ["1790186555.000000", "1790186555.0", "1790186555.957249"] {
5404 let below = just_below(ts);
5405 assert!(below.as_str() < ts, "`{below}` must sort below `{ts}`");
5406 }
5407 assert_eq!(just_below("1790186555.000000"), "1790186554.999999");
5408 assert_eq!(just_below("1790186555.957249"), "1790186555.957248");
5409 }
5410
5411 #[tokio::test]
5416 async fn each_clock_reads_only_its_own_conversations() {
5417 let host = FakeHost::with_subscriptions(vec![
5418 sub_dm("acme"),
5419 sub_channel("acme", "C1", SlackWatchMode::All),
5420 ]);
5421 host.set_watermark("acme", "D1", "050.0");
5422 host.seed_history("D1", vec![history_message("200.0", "U9", "a dm")]);
5423 host.set_watermark("acme", "C1", "050.0");
5424 host.seed_history("C1", vec![history_message("200.0", "U9", "a channel")]);
5425
5426 let outcome = sweep_scope(&host, &host, "acme", SweepScope::Conversations)
5427 .await
5428 .expect("the conversation pass");
5429 assert_eq!(outcome.delivered, 1, "the named channel only");
5430 assert_eq!(host.delivered()[0].conversation, "C1", "and the DM is not on this clock");
5431
5432 let outcome = sweep_scope(&host, &host, "acme", SweepScope::DirectMessages)
5433 .await
5434 .expect("the DM pass");
5435 assert_eq!(outcome.delivered, 1, "the DM only");
5436 assert_eq!(host.delivered()[1].conversation, "D1", "and the channel is not on that one");
5437 }
5438
5439 #[tokio::test]
5443 async fn a_named_dm_is_read_on_the_conversation_clock_only() {
5444 let host = FakeHost::with_subscriptions(vec![
5445 sub_dm("acme"),
5446 sub_channel("acme", "D1", SlackWatchMode::All),
5447 ]);
5448 host.set_watermark("acme", "D1", "050.0");
5449 host.seed_history("D1", vec![history_message("200.0", "U9", "a dm")]);
5450
5451 let dms = sweep_scope(&host, &host, "acme", SweepScope::DirectMessages)
5452 .await
5453 .expect("the DM pass");
5454 assert_eq!(dms.delivered, 0, "the named DM is not the DM clock's to read");
5455 assert!(host.delivered().is_empty(), "so it was not read at all: {:?}", host.delivered());
5456
5457 sweep_scope(&host, &host, "acme", SweepScope::Conversations)
5458 .await
5459 .expect("the conversation pass");
5460 assert!(
5461 host.delivered().iter().any(|message| message.conversation == "D1"),
5462 "it is the conversation clock's: {:?}",
5463 host.delivered(),
5464 );
5465 }
5466
5467 #[tokio::test]
5471 async fn a_sweep_with_nothing_new_delivers_nothing() {
5472 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
5473 host.set_watermark("acme", "D1", "100.0");
5474 host.seed_history("D1", vec![history_message("100.0", "U9", "already delivered")]);
5475
5476 let outcome = sweep(&host, &host, "acme").await.expect("sweep");
5477 assert_eq!(outcome.delivered, 0, "a quiet tick must not deliver an empty bundle");
5478 assert!(host.batches().is_empty(), "and must not call the seam at all");
5479 }
5480
5481 #[tokio::test]
5485 async fn a_sweep_delivers_what_the_user_typed_and_refuses_what_forge_posted() {
5486 let host =
5487 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
5488 host.set_watermark("acme", "C1", "100.0");
5489 host.seed_history(
5490 "C1",
5491 vec![
5492 history_message("200.2", "U1", "the agent's own post"),
5493 history_message("200.1", "U1", "what Ved typed"),
5494 ],
5495 );
5496 host.record_own_post("acme", "C1", "200.2");
5497
5498 sweep(&host, &host, "acme").await.expect("sweep");
5499
5500 let texts: Vec<_> = host.delivered().into_iter().map(|m| m.text).collect();
5501 assert_eq!(
5502 texts,
5503 vec!["what Ved typed".to_owned()],
5504 "the ledger refuses the agent's post by ts; the user's own words arrive",
5505 );
5506 }
5507
5508 #[tokio::test]
5513 async fn a_conversation_with_a_post_in_flight_is_skipped_without_moving_its_cursor() {
5514 let host =
5515 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
5516 host.set_watermark("acme", "C1", "100.0");
5517 host.seed_history("C1", vec![history_message("200.0", "U1", "what Ved typed")]);
5518 host.set_post_in_flight("acme", "C1", true);
5519
5520 let outcome = sweep(&host, &host, "acme").await.expect("sweep");
5521
5522 assert_eq!(outcome.delivered, 0, "the tick is skipped while a post is in flight");
5523 assert!(host.delivered().is_empty(), "nothing is read from that conversation");
5524 assert_eq!(
5525 host.watermark("acme", "C1").expect("watermark"),
5526 Some("100.0".to_owned()),
5527 "and the cursor holds, so the next tick reads the window whole",
5528 );
5529
5530 host.set_post_in_flight("acme", "C1", false);
5531 let outcome = sweep(&host, &host, "acme").await.expect("the next tick");
5532 assert_eq!(
5533 outcome.delivered, 1,
5534 "the held message is delivered once the post has finished"
5535 );
5536 assert_eq!(
5537 host.delivered().first().map(|message| message.ts.as_str()),
5538 Some("200.0"),
5539 "and it is the message the skip held, not a lost window",
5540 );
5541 }
5542
5543 #[tokio::test]
5547 async fn an_alias_addressed_mark_still_skips_the_conversation() {
5548 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
5549 host.set_watermark("acme", "D1", "050.0");
5550 host.seed_history("D1", vec![history_message("200.0", "U9", "a dm")]);
5551 host.set_post_in_flight("acme", "U9", true);
5552
5553 let outcome = sweep(&host, &host, "acme").await.expect("sweep");
5554
5555 assert_eq!(
5556 outcome.delivered, 0,
5557 "the DM's partner address marks the conversation in flight"
5558 );
5559 assert!(host.delivered().is_empty(), "nothing is read from that DM");
5560 }
5561
5562 #[tokio::test]
5568 async fn a_mention_in_a_conversation_with_a_post_in_flight_holds_the_cursor() {
5569 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
5570 host.seed_search(vec![
5571 search_match("300.2", "C1", "newer <@U1>"),
5572 search_match("200.1", "C2", "skipped <@U1>"),
5573 ]);
5574 host.set_post_in_flight("acme", "C2", true);
5575
5576 let first = sweep_mentions(&host, &host, "acme").await.expect("first sweep");
5577
5578 assert_eq!(first.delivered, 1, "the newer hit in another conversation is delivered");
5579 assert_eq!(
5580 host.watermark("acme", MENTION_CURSOR).expect("watermark"),
5581 None,
5582 "and the cursor holds below the skipped hit rather than passing it",
5583 );
5584
5585 host.set_post_in_flight("acme", "C2", false);
5586 sweep_mentions(&host, &host, "acme").await.expect("second sweep");
5587
5588 let delivered = host.delivered();
5589 let conversations: Vec<&str> =
5590 delivered.iter().map(|message| message.conversation.as_str()).collect();
5591 assert!(
5592 conversations.contains(&"C2"),
5593 "the skipped mention is delivered once the post finished: {conversations:?}",
5594 );
5595 }
5596
5597 #[tokio::test]
5601 async fn two_active_conversations_deliver_two_bundles() {
5602 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
5603 host.set_watermark("acme", "D1", "100.0");
5604 host.set_watermark("acme", "D2", "100.0");
5605 host.seed_history(
5606 "D1",
5607 vec![
5609 history_message("300.1", "U9", "d1 newest"),
5610 history_message("200.1", "U9", "d1 older"),
5611 ],
5612 );
5613 host.seed_history("D2", vec![history_message("200.2", "U8", "d2 only")]);
5614
5615 let outcome = sweep(&host, &host, "acme").await.expect("sweep");
5616 assert_eq!(outcome.delivered, 2, "one delivery per conversation, never one merged one");
5617 let batches = host.batches();
5618 assert_eq!(batches.len(), 2, "two blocks, never one per message");
5619 assert_ne!(batches[0][0].conversation, batches[1][0].conversation);
5620 assert_eq!(
5621 batches.iter().map(Vec::len).sum::<usize>(),
5622 3,
5623 "every message rides its own conversation's block",
5624 );
5625 }
5626
5627 fn sub_dm(workspace: &str) -> SlackSubscription {
5628 let mut sub = sub_for(SlackSubscriptionTarget::DirectMessages);
5629 sub.workspace = workspace.to_owned();
5630 sub
5631 }
5632
5633 fn sub_mentions(workspace: &str) -> SlackSubscription {
5634 let mut sub = sub_for(SlackSubscriptionTarget::Mentions);
5635 sub.workspace = workspace.to_owned();
5636 sub
5637 }
5638
5639 fn sub_conversation_owned_by(
5640 workspace: &str,
5641 project: &str,
5642 team_role: Option<&str>,
5643 id: &str,
5644 ) -> SlackSubscription {
5645 let mut sub = sub_for(SlackSubscriptionTarget::Conversation {
5646 id: id.to_owned(),
5647 name: None,
5648 mode: SlackWatchMode::All,
5649 });
5650 sub.workspace = workspace.to_owned();
5651 sub.project = project.to_owned();
5652 sub.team_role = team_role.map(str::to_owned);
5653 sub
5654 }
5655
5656 fn thread_owner(project: &str, team_role: Option<&str>) -> SlackThreadOwner {
5657 SlackThreadOwner { project: project.to_owned(), team_role: team_role.map(str::to_owned) }
5658 }
5659
5660 fn search_match(ts: &str, conversation: &str, text: &str) -> SlackSearchMatch {
5661 SlackSearchMatch {
5662 ts: ts.to_owned(),
5663 text: text.to_owned(),
5664 conversation_id: conversation.to_owned(),
5665 conversation_name: None,
5666 username: Some("U9".to_owned()),
5667 user: Some("U9".to_owned()),
5668 thread_ts: None,
5669 files: Vec::new(),
5670 }
5671 }
5672
5673 #[tokio::test]
5677 async fn a_mention_sweep_makes_exactly_one_search_call_for_the_workspace() {
5678 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
5679 host.seed_search(vec![search_match("200.1", "C1", "ping <@U1>")]);
5680
5681 sweep_mentions(&host, &host, "acme").await.expect("sweep");
5682 assert_eq!(host.count_for("search.messages"), 1);
5683 }
5684
5685 #[tokio::test]
5686 async fn only_mention_matches_newer_than_the_cursor_are_delivered() {
5687 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
5688 host.set_watermark("acme", MENTION_CURSOR, "150.0");
5689 host.seed_search(vec![
5690 search_match("200.1", "C1", "newer <@U1>"),
5691 search_match("100.1", "C1", "older <@U1>"),
5692 ]);
5693
5694 let outcome = sweep_mentions(&host, &host, "acme").await.expect("sweep");
5695 assert_eq!(outcome.delivered, 1);
5696 assert_eq!(host.delivered()[0].text, "newer <@U1>");
5697 }
5698
5699 #[tokio::test]
5700 async fn the_mention_cursor_advances_to_the_newest_ts() {
5701 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
5702 host.seed_search(vec![
5703 search_match("200.1", "C1", "a <@U1>"),
5704 search_match("300.2", "C2", "b <@U1>"),
5705 ]);
5706
5707 sweep_mentions(&host, &host, "acme").await.expect("sweep");
5708 assert_eq!(
5709 host.watermark("acme", MENTION_CURSOR).expect("watermark"),
5710 Some("300.2".to_owned())
5711 );
5712 }
5713
5714 #[tokio::test]
5718 async fn a_mention_sweep_with_no_targets_does_not_search_at_all() {
5719 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
5720
5721 sweep_mentions(&host, &host, "acme").await.expect("sweep");
5722 assert_eq!(
5723 host.count_for("search.messages"),
5724 0,
5725 "an unsubscribed workspace must not spend a call",
5726 );
5727 }
5728
5729 #[tokio::test]
5733 async fn a_freshly_auto_subscribed_conversation_starts_from_the_mention() {
5734 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
5735 host.seed_search(vec![search_match("200.1", "C1", "ping <@U1>")]);
5736 sweep_mentions(&host, &host, "acme").await.expect("sweep");
5737
5738 host.seed_history(
5740 "C1",
5741 vec![
5742 history_message("250.1", "U9", "after the mention"),
5743 history_message("150.1", "U9", "before the mention"),
5744 ],
5745 );
5746 sweep(&host, &host, "acme").await.expect("sweep");
5747
5748 let texts: Vec<_> = host.delivered().into_iter().map(|m| m.text).collect();
5749 assert_eq!(
5750 texts,
5751 vec!["ping <@U1>".to_owned(), "after the mention".to_owned()],
5752 "the subscription starts at the mention, not at the channel's history",
5753 );
5754 }
5755
5756 #[tokio::test]
5760 async fn a_mention_backlog_wider_than_one_page_is_delivered_without_loss() {
5761 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
5762 let all: Vec<SlackSearchMatch> = (0..25)
5763 .rev()
5764 .map(|n| search_match(&format!("1{n:02}.1"), "C1", "ping <@U1>"))
5765 .collect();
5766 let newest = all[0].ts.clone();
5767 let oldest = all[24].ts.clone();
5768 let (page_one, page_two) = all.split_at(20);
5769 host.seed_search_pages(vec![page_one.to_vec(), page_two.to_vec()]);
5770
5771 let outcome = sweep_mentions(&host, &host, "acme").await.expect("sweep");
5772 assert_eq!(outcome.delivered, 25, "every mention in the backlog arrives");
5773 assert_eq!(host.watermark("acme", MENTION_CURSOR).expect("watermark"), Some(newest));
5774 assert_ne!(
5775 host.watermark("acme", MENTION_CURSOR).expect("watermark"),
5776 Some(oldest),
5777 "nothing is skipped"
5778 );
5779 }
5780
5781 #[tokio::test]
5787 async fn the_mention_sweep_refuses_its_own_ledgered_post_and_delivers_the_users_own() {
5788 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
5789 let mut agent_post = search_match("300.2", "C1", "mine <@U1>");
5790 agent_post.user = Some("U1".to_owned());
5791 let mut user_post = search_match("250.1", "C1", "what I wrote <@U1>");
5792 user_post.user = Some("U1".to_owned());
5793 host.seed_search(vec![search_match("200.1", "C1", "theirs <@U1>"), agent_post, user_post]);
5794 host.record_own_post("acme", "C1", "300.2");
5795
5796 let outcome = sweep_mentions(&host, &host, "acme").await.expect("sweep");
5797 assert_eq!(outcome.delivered, 2, "the agent's post is refused; every other hit arrives");
5798 let texts: Vec<_> = host.delivered().into_iter().map(|m| m.text).collect();
5799 assert!(!texts.iter().any(|t| t == "mine <@U1>"), "no self-echo: {texts:?}");
5800 assert!(
5801 texts.iter().any(|t| t == "what I wrote <@U1>"),
5802 "the user's own mention is delivered: {texts:?}",
5803 );
5804 }
5805
5806 #[tokio::test]
5811 async fn a_failed_delivery_holds_the_cursor_at_its_previous_value() {
5812 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
5813 host.set_watermark("acme", "D1", "050.0");
5814 host.seed_history(
5815 "D1",
5816 vec![
5818 history_message("300.1", "U9", "newest"),
5819 history_message("200.1", "U9", "in the same batch"),
5820 history_message("100.1", "U9", "oldest"),
5821 ],
5822 );
5823 host.fail_delivery_at(0);
5824
5825 let outcome = sweep(&host, &host, "acme").await.expect("sweep");
5826 assert_eq!(outcome.delivered, 0, "a block that did not land is not delivered");
5827 assert!(host.delivered().is_empty(), "nothing of a failed block reached the session");
5828 assert_eq!(
5829 host.watermark("acme", "D1"),
5830 Ok(Some("050.0".to_owned())),
5831 "the cursor holds below the failure instead of jumping past it",
5832 );
5833 }
5834
5835 #[tokio::test]
5836 async fn a_mention_creates_a_conversation_subscription() {
5837 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
5838 host.seed_search(vec![search_match("200.1", "C1", "ping <@U1>")]);
5839
5840 sweep_mentions(&host, &host, "acme").await.expect("sweep");
5841 assert_eq!(host.auto_subscribed(), vec!["C1".to_owned()]);
5842 }
5843
5844 #[tokio::test]
5845 async fn an_already_watched_conversation_is_not_subscribed_twice() {
5846 let host = FakeHost::with_subscriptions(vec![
5847 sub_mentions("acme"),
5848 sub_channel("acme", "C1", SlackWatchMode::All),
5849 ]);
5850 host.seed_search(vec![search_match("200.1", "C1", "ping <@U1>")]);
5851
5852 sweep_mentions(&host, &host, "acme").await.expect("sweep");
5853 assert!(host.auto_subscribed().is_empty(), "C1 is already watched");
5854 }
5855
5856 fn sub_channel(workspace: &str, id: &str, mode: SlackWatchMode) -> SlackSubscription {
5857 let mut sub =
5858 sub_for(SlackSubscriptionTarget::Conversation { id: id.to_owned(), name: None, mode });
5859 sub.workspace = workspace.to_owned();
5860 sub
5861 }
5862
5863 #[tokio::test]
5864 async fn a_sweep_delivers_only_what_follows_the_watermark() {
5865 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
5866 host.seed_history(
5867 "D1",
5868 vec![history_message("200.1", "U9", "newer"), history_message("100.1", "U9", "older")],
5869 );
5870 host.set_watermark("acme", "D1", "150.0");
5871
5872 let outcome = sweep(&host, &host, "acme").await.expect("sweep");
5873 assert_eq!(outcome.delivered, 1);
5874 assert_eq!(host.delivered()[0].text, "newer");
5875 }
5876
5877 #[tokio::test]
5878 async fn a_sweep_advances_the_watermark_to_the_newest_seen() {
5879 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
5880 host.set_watermark("acme", "D1", "050.0");
5881 host.seed_history(
5882 "D1",
5883 vec![history_message("300.2", "U9", "newest"), history_message("200.1", "U9", "older")],
5885 );
5886
5887 sweep(&host, &host, "acme").await.expect("sweep");
5888 assert_eq!(host.watermark("acme", "D1").expect("watermark"), Some("300.2".to_owned()));
5889 }
5890
5891 #[tokio::test]
5897 async fn an_edit_to_a_delivered_message_arrives_and_the_cursor_clears_it() {
5898 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
5899 host.set_watermark("acme", "D1", "100.0");
5900 let mut corrected = history_message("090.0", "U9", "deploy is RED");
5901 corrected.edited = Some("150.0".to_owned());
5902 host.seed_history("D1", vec![corrected]);
5903
5904 let outcome = sweep(&host, &host, "acme").await.expect("the edit's sweep");
5905 assert_eq!(outcome.delivered, 1, "the correction is delivered");
5906 assert_eq!(host.delivered()[0].text, "deploy is RED", "with its new text");
5907 assert_eq!(
5908 host.delivered()[0].edited.as_deref(),
5909 Some("150.0"),
5910 "and it carries the edit ts the delivery memory keys on",
5911 );
5912 assert_eq!(
5913 host.watermark("acme", "D1"),
5914 Ok(Some("150.0".to_owned())),
5915 "the cursor clears the edit, so the next sweep does not re-offer it",
5916 );
5917
5918 let again = sweep(&host, &host, "acme").await.expect("a quiet sweep");
5919 assert_eq!(again.delivered, 0, "an edit is delivered once, not every tick");
5920 }
5921
5922 #[tokio::test]
5926 async fn a_sweep_reads_every_history_page_before_advancing() {
5927 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
5928 host.set_watermark("acme", "D1", "050.0");
5929 host.seed_history_pages(
5930 "D1",
5931 vec![
5933 vec![history_message("300.3", "U9", "three")],
5934 vec![history_message("200.2", "U9", "two")],
5935 vec![history_message("100.1", "U9", "one")],
5936 ],
5937 );
5938
5939 sweep(&host, &host, "acme").await.expect("sweep");
5940 let texts: Vec<_> = host.delivered().into_iter().map(|m| m.text).collect();
5941 assert_eq!(texts, vec!["three".to_owned(), "two".to_owned(), "one".to_owned()]);
5942 assert_eq!(
5943 host.watermark("acme", "D1").expect("watermark"),
5944 Some("300.3".to_owned()),
5945 "the cursor clears every page that was read",
5946 );
5947 }
5948
5949 #[tokio::test]
5950 async fn a_rate_limit_is_reported_rather_than_retried_inline() {
5951 let mut host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
5952 host.fail_with_rate_limit(Duration::from_secs(7));
5953
5954 let outcome = sweep(&host, &host, "acme").await.expect("a 429 is not an error");
5955 assert_eq!(outcome.rate_limited, Some(Duration::from_secs(7)));
5956 }
5957
5958 #[tokio::test]
5959 async fn a_message_in_a_thread_pulls_its_replies_and_dedupes_the_parent() {
5960 let host =
5963 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
5964 host.set_watermark("acme", "C1", "050.0");
5965 let root =
5966 SlackHistoryMessage { text: "root".to_owned(), ..parent_with_replies("100.0", 2) };
5967 host.seed_history("C1", vec![root.clone()]);
5968 host.seed_replies(
5969 "C1",
5970 "100.0",
5971 vec![
5972 root.clone(),
5973 history_message("150.0", "U1", "the user's own reply"),
5974 reply("200.1", "100.0", "U8", "reply one"),
5975 root,
5976 reply("300.2", "100.0", "U8", "reply two"),
5977 ],
5978 );
5979
5980 sweep(&host, &host, "acme").await.expect("sweep");
5981 let texts: Vec<_> = host.delivered().into_iter().map(|m| m.text).collect();
5982 assert_eq!(
5983 texts,
5984 vec![
5985 "root".to_owned(),
5986 "the user's own reply".to_owned(),
5987 "reply one".to_owned(),
5988 "reply two".to_owned(),
5989 ],
5990 "the parent must not repeat per page, and the user's own reply rides in",
5991 );
5992 }
5993
5994 #[tokio::test]
5999 async fn a_corrected_reply_rides_in_on_its_parents_redelivery() {
6000 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6001 "acme",
6002 "forge",
6003 Some("tester"),
6004 "C1",
6005 )]);
6006 host.set_watermark("acme", "C1", "300.0");
6009 let mut corrected_parent = parent_with_replies("100.0", 2);
6010 corrected_parent.edited = Some("340.0".to_owned());
6011 let mut corrected = reply("150.0", "100.0", "U8", "the answer, corrected");
6012 corrected.edited = Some("350.0".to_owned());
6013 host.seed_history("C1", vec![corrected_parent.clone()]);
6014 host.seed_replies(
6015 "C1",
6016 "100.0",
6017 vec![
6018 corrected_parent,
6019 history_message("105.0", "U1", "the user's own reply"),
6020 corrected,
6021 ],
6022 );
6023 host.follow_thread("acme", "C1", "100.0", thread_owner("forge", Some("tester")), "150.0");
6024
6025 sweep(&host, &host, "acme").await.expect("the corrections' sweep");
6026
6027 let delivered = host.delivered();
6028 assert!(
6029 delivered.iter().any(|message| message.ts == "100.0" && message.text == "trigger"),
6030 "the corrected parent is delivered: {delivered:?}",
6031 );
6032 assert!(
6033 delivered
6034 .iter()
6035 .any(|message| message.ts == "150.0" && message.text == "the answer, corrected"),
6036 "and its thread's corrected reply rides the fresh read: {delivered:?}",
6037 );
6038
6039 host.restart();
6044 let calls_before = host.replies_calls();
6045 sweep(&host, &host, "acme").await.expect("the first sweep after a restart");
6046 assert!(host.replies_calls() > calls_before, "the thread was walked after the restart");
6047 assert_eq!(
6048 host.replies_oldest().last(),
6049 Some(&None),
6050 "the walk reads the thread whole: {:?}",
6051 host.replies_oldest(),
6052 );
6053 assert_eq!(
6054 host.delivered()
6055 .iter()
6056 .filter(|message| message.text == "the answer, corrected")
6057 .count(),
6058 1,
6059 "the correction is not handed over twice: {:?}",
6060 host.delivered(),
6061 );
6062 }
6063
6064 #[tokio::test]
6071 async fn a_corrected_reply_in_a_participating_thread_arrives_without_its_parent() {
6072 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6073 "acme",
6074 "forge",
6075 Some("tester"),
6076 "C1",
6077 )]);
6078 let parent = parent_with_latest_reply("100.0", 2, "150.0");
6079 host.set_watermark("acme", "C1", "050.0");
6080 host.seed_history("C1", vec![parent.clone()]);
6081 host.seed_replies(
6082 "C1",
6083 "100.0",
6084 vec![
6085 parent.clone(),
6086 history_message("105.0", "U1", "the user's own reply"),
6087 reply("150.0", "100.0", "U8", "the answer"),
6088 ],
6089 );
6090 sweep(&host, &host, "acme").await.expect("trigger sweep");
6091 assert!(
6092 host.delivered().iter().any(|message| message.ts == "150.0"),
6093 "the trigger sweep delivers the reply: {:?}",
6094 host.delivered(),
6095 );
6096
6097 let mut corrected = reply("150.0", "100.0", "U8", "the answer, corrected");
6100 corrected.edited = Some("990.0".to_owned());
6101 host.seed_history("C1", vec![parent_with_latest_reply("100.0", 2, "150.0")]);
6102 host.seed_replies(
6103 "C1",
6104 "100.0",
6105 vec![parent, history_message("105.0", "U1", "the user's own reply"), corrected],
6106 );
6107
6108 sweep(&host, &host, "acme").await.expect("the correction's sweep");
6109
6110 let delivered = host.delivered();
6111 assert!(
6112 delivered
6113 .iter()
6114 .any(|message| message.ts == "150.0" && message.text == "the answer, corrected"),
6115 "the corrected reply arrives though its parent never moved: {delivered:?}",
6116 );
6117 assert_eq!(
6118 host.thread_cursor("acme", "C1", "100.0").as_deref(),
6119 Some("990.0"),
6120 "and the thread cursor clears the edit, so the next tick does not re-offer it",
6121 );
6122
6123 host.age_thread_full_read("acme", "C1", "100.0", Duration::from_secs(601));
6127 sweep(&host, &host, "acme").await.expect("the whole read");
6128 let corrected = host
6129 .delivered()
6130 .iter()
6131 .filter(|message| message.ts == "150.0" && message.text == "the answer, corrected")
6132 .count();
6133 assert_eq!(
6134 corrected,
6135 1,
6136 "the correction arrives once, not once per path: {:?}",
6137 host.delivered()
6138 );
6139 }
6140
6141 #[tokio::test]
6147 async fn a_recreated_thread_does_not_replay_an_edit_its_seed_already_passed() {
6148 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6149 "acme",
6150 "forge",
6151 Some("tester"),
6152 "C1",
6153 )]);
6154 host.set_watermark("acme", "C1", "300.0");
6157 let parent = parent_with_latest_reply("100.0", 2, "200.0");
6158 let mut corrected = reply("150.0", "100.0", "U8", "the answer, corrected");
6159 corrected.edited = Some("990.0".to_owned());
6160 host.seed_history("C1", vec![]);
6161 host.seed_replies(
6162 "C1",
6163 "100.0",
6164 vec![parent, history_message("105.0", "U1", "his own reply"), corrected],
6165 );
6166 host.follow_thread("acme", "C1", "100.0", thread_owner("forge", Some("tester")), "200.0");
6168
6169 sweep(&host, &host, "acme").await.expect("the first walk");
6170 assert!(
6171 !host.delivered().iter().any(|message| message.ts == "150.0"),
6172 "the first walk filters by ts: {:?}",
6173 host.delivered(),
6174 );
6175
6176 host.age_thread_full_read("acme", "C1", "100.0", Duration::from_secs(601));
6178 sweep(&host, &host, "acme").await.expect("the periodic whole read");
6179 assert!(
6180 host.delivered()
6181 .iter()
6182 .any(|message| message.ts == "150.0" && message.text == "the answer, corrected"),
6183 "the whole read finds the correction: {:?}",
6184 host.delivered(),
6185 );
6186 }
6187
6188 #[tokio::test]
6195 async fn a_restart_does_not_re_deliver_an_edit_the_thread_row_already_judged() {
6196 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6197 "acme",
6198 "forge",
6199 Some("tester"),
6200 "C1",
6201 )]);
6202 let parent = parent_with_latest_reply("100.0", 2, "150.0");
6205 host.set_watermark("acme", "C1", "050.0");
6206 host.seed_history("C1", vec![parent.clone()]);
6207 host.seed_replies(
6208 "C1",
6209 "100.0",
6210 vec![
6211 parent.clone(),
6212 history_message("105.0", "U1", "the user's own reply"),
6213 reply("150.0", "100.0", "U8", "the answer"),
6214 ],
6215 );
6216 sweep(&host, &host, "acme").await.expect("the trigger sweep");
6217 assert!(
6218 host.delivered().iter().any(|message| message.ts == "150.0"),
6219 "the trigger sweep delivers the reply: {:?}",
6220 host.delivered(),
6221 );
6222
6223 let mut corrected = reply("150.0", "100.0", "U8", "the answer, corrected");
6225 corrected.edited = Some("990.0".to_owned());
6226 host.seed_history("C1", vec![parent_with_latest_reply("100.0", 2, "150.0")]);
6229 host.seed_replies(
6230 "C1",
6231 "100.0",
6232 vec![parent, history_message("105.0", "U1", "the user's own reply"), corrected],
6233 );
6234 sweep(&host, &host, "acme").await.expect("the correction's sweep");
6235 assert_eq!(
6236 host.delivered()
6237 .iter()
6238 .filter(|message| message.text == "the answer, corrected")
6239 .count(),
6240 1,
6241 "the correction is handed over once: {:?}",
6242 host.delivered(),
6243 );
6244
6245 host.restart();
6247 let calls_before = host.replies_calls();
6248
6249 sweep(&host, &host, "acme").await.expect("the first sweep after a restart");
6250 assert!(
6251 host.replies_calls() > calls_before,
6252 "the thread was walked after the restart: {calls_before} calls before",
6253 );
6254 assert_eq!(
6255 host.replies_oldest().last(),
6256 Some(&None),
6257 "and the first post-restart walk reads the thread whole: {:?}",
6258 host.replies_oldest(),
6259 );
6260 assert_eq!(
6261 host.delivered()
6262 .iter()
6263 .filter(|message| message.text == "the answer, corrected")
6264 .count(),
6265 1,
6266 "the correction is not handed over again after the restart: {:?}",
6267 host.delivered(),
6268 );
6269 }
6270
6271 #[tokio::test]
6274 async fn a_channel_past_the_page_cap_refuses_instead_of_returning_partial_history() {
6275 let host = FakeHost::default();
6276 let pages: Vec<Vec<SlackHistoryMessage>> = (0..=MAX_HISTORY_PAGES)
6277 .map(|n| vec![history_message(&format!("{}.0", 200 + n), "U8", "a message")])
6278 .collect();
6279 host.seed_history_pages("C1", pages);
6280
6281 let err = fetch_history(&host, "acme", "C1", None)
6282 .await
6283 .expect_err("the walk must not read as complete");
6284
6285 assert!(
6286 matches!(&err, SlackError::Truncated { pages, .. } if *pages == MAX_HISTORY_PAGES),
6287 "the refusal names the cap it hit: {err:?}",
6288 );
6289 }
6290
6291 #[tokio::test]
6294 async fn a_thread_read_past_the_page_cap_refuses() {
6295 let host = FakeHost::default();
6296 let pages: Vec<Vec<SlackHistoryMessage>> = (0..=MAX_REPLY_PAGES)
6297 .map(|n| vec![reply(&format!("{}.0", 200 + n), "100.0", "U8", "a reply")])
6298 .collect();
6299 host.seed_replies_pages("C1", "100.0", pages);
6300
6301 let err = read_thread(&host, "C1", "100.0", 50_000)
6302 .await
6303 .expect_err("the read must not stop at the cap and call it done");
6304
6305 assert!(
6306 matches!(&err, SlackError::Truncated { pages, .. } if *pages == MAX_REPLY_PAGES),
6307 "{err:?}",
6308 );
6309 }
6310
6311 #[tokio::test]
6316 async fn a_thread_past_the_page_cap_refuses_instead_of_returning_the_oldest_pages() {
6317 let host = FakeHost::default();
6318 let pages: Vec<Vec<SlackHistoryMessage>> = (0..=MAX_REPLY_PAGES)
6319 .map(|n| vec![reply(&format!("{}.0", 200 + n), "100.0", "U8", "a reply")])
6320 .collect();
6321 host.seed_replies_pages("C1", "100.0", pages);
6322
6323 let err = fetch_replies(&host, "acme", "C1", "100.0", None)
6324 .await
6325 .expect_err("the walk must not read as complete");
6326
6327 assert!(
6328 matches!(&err, SlackError::Truncated { pages, .. } if *pages == MAX_REPLY_PAGES),
6329 "the refusal names the cap it hit: {err:?}",
6330 );
6331 }
6332
6333 #[tokio::test]
6337 async fn a_throttled_thread_does_not_starve_the_threads_after_it() {
6338 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6339 "acme",
6340 "forge",
6341 Some("tester"),
6342 "C1",
6343 )]);
6344 host.set_watermark("acme", "C1", "300.0");
6346 host.seed_history("C1", vec![]);
6347 for (parent, reply_ts) in [("100.0", "150.0"), ("200.0", "250.0")] {
6348 host.seed_replies("C1", parent, vec![reply(reply_ts, parent, "U8", "the answer")]);
6349 host.follow_thread(
6350 "acme",
6351 "C1",
6352 parent,
6353 thread_owner("forge", Some("tester")),
6354 "050.0",
6355 );
6356 host.mark_thread_participating("acme", "C1", parent);
6357 }
6358 host.rate_limit_replies_at(0);
6359
6360 let outcome = sweep(&host, &host, "acme").await.expect("a 429 is not an error");
6361
6362 assert_eq!(
6363 outcome.rate_limited,
6364 Some(Duration::from_secs(5)),
6365 "the throttle still reaches the pump",
6366 );
6367 assert_eq!(host.replies_calls(), 2, "the throttled thread did not end the walk");
6368 assert_eq!(
6369 host.delivered().len(),
6370 1,
6371 "the thread after the throttled one still delivered: {:?}",
6372 host.delivered(),
6373 );
6374 }
6375
6376 #[tokio::test]
6380 async fn an_edit_below_a_newer_reply_still_arrives_on_the_whole_read() {
6381 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6382 "acme",
6383 "forge",
6384 Some("tester"),
6385 "C1",
6386 )]);
6387 host.set_watermark("acme", "C1", "300.0");
6388 host.seed_history("C1", vec![]);
6389 let parent = reply("100.0", "100.0", "U9", "the parent");
6390 host.seed_replies(
6391 "C1",
6392 "100.0",
6393 vec![
6394 parent.clone(),
6395 reply("150.0", "100.0", "U8", "the answer"),
6396 reply("300.0", "100.0", "U8", "the second answer"),
6397 ],
6398 );
6399 host.follow_thread("acme", "C1", "100.0", thread_owner("forge", Some("tester")), "050.0");
6400 host.mark_thread_participating("acme", "C1", "100.0");
6401 sweep(&host, &host, "acme").await.expect("the first walk");
6402 assert_eq!(
6403 host.thread_cursor("acme", "C1", "100.0"),
6404 Some("300.0".to_owned()),
6405 "the first walk delivered both replies",
6406 );
6407
6408 let mut corrected = reply("150.0", "100.0", "U8", "the answer, corrected");
6411 corrected.edited = Some("400.0".to_owned());
6412 host.seed_replies(
6413 "C1",
6414 "100.0",
6415 vec![
6416 parent,
6417 corrected,
6418 reply("300.0", "100.0", "U8", "the second answer"),
6419 reply("500.0", "100.0", "U8", "a newer answer"),
6420 ],
6421 );
6422 sweep(&host, &host, "acme").await.expect("the cheap tick");
6423 assert!(
6424 host.delivered().iter().any(|message| message.ts == "500.0"),
6425 "the newer reply lands first: {:?}",
6426 host.delivered(),
6427 );
6428 assert_eq!(
6429 host.thread_cursor("acme", "C1", "100.0"),
6430 Some("500.0".to_owned()),
6431 "and moves the cursor past the edit at 400",
6432 );
6433
6434 host.age_thread_full_read("acme", "C1", "100.0", Duration::from_secs(601));
6435 sweep(&host, &host, "acme").await.expect("the whole read");
6436
6437 assert!(
6438 host.delivered()
6439 .iter()
6440 .any(|message| message.ts == "150.0" && message.text == "the answer, corrected"),
6441 "the correction arrives even though its stamp is behind the cursor: {:?}",
6442 host.delivered(),
6443 );
6444 assert_eq!(
6445 host.thread_cursor("acme", "C1", "100.0"),
6446 Some("500.0".to_owned()),
6447 "and the cursor is never moved back",
6448 );
6449 }
6450
6451 #[tokio::test]
6455 async fn a_failed_delivery_holds_the_edit_watermark() {
6456 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6457 "acme",
6458 "forge",
6459 Some("tester"),
6460 "C1",
6461 )]);
6462 host.set_watermark("acme", "C1", "300.0");
6463 host.seed_history("C1", vec![]);
6464 host.seed_replies(
6465 "C1",
6466 "100.0",
6467 vec![
6468 reply("150.0", "100.0", "U8", "the answer"),
6469 reply("300.0", "100.0", "U8", "the second answer"),
6470 ],
6471 );
6472 host.follow_thread("acme", "C1", "100.0", thread_owner("forge", Some("tester")), "050.0");
6473 host.mark_thread_participating("acme", "C1", "100.0");
6474 sweep(&host, &host, "acme").await.expect("the first walk");
6475
6476 let mut corrected = reply("150.0", "100.0", "U8", "the answer, corrected");
6477 corrected.edited = Some("400.0".to_owned());
6478 host.seed_replies(
6479 "C1",
6480 "100.0",
6481 vec![corrected, reply("300.0", "100.0", "U8", "the second answer")],
6482 );
6483 host.age_thread_full_read("acme", "C1", "100.0", Duration::from_secs(601));
6484 host.fail_delivery_at(1);
6487 sweep(&host, &host, "acme").await.expect("the whole read");
6488 assert!(
6489 !host.delivered().iter().any(|message| message.text.contains("corrected")),
6490 "the failed block did not land: {:?}",
6491 host.delivered(),
6492 );
6493 assert_eq!(
6494 host.thread_cursor("acme", "C1", "100.0"),
6495 Some("300.0".to_owned()),
6496 "the cursor holds below the failure",
6497 );
6498
6499 host.age_thread_full_read("acme", "C1", "100.0", Duration::from_secs(601));
6500 sweep(&host, &host, "acme").await.expect("the retry");
6501 assert!(
6502 host.delivered().iter().any(|message| message.text == "the answer, corrected"),
6503 "the retry still finds the correction: {:?}",
6504 host.delivered(),
6505 );
6506
6507 host.seed_replies(
6511 "C1",
6512 "100.0",
6513 vec![
6514 reply("150.0", "100.0", "U8", "the answer, corrected"),
6515 reply("300.0", "100.0", "U8", "the second answer"),
6516 reply("500.0", "100.0", "U8", "a newer answer"),
6517 ],
6518 );
6519 sweep(&host, &host, "acme").await.expect("the newer reply's cheap tick");
6520 assert_eq!(
6521 host.thread_cursor("acme", "C1", "100.0"),
6522 Some("500.0".to_owned()),
6523 "the cursor moved past the correction",
6524 );
6525
6526 host.restart();
6529 let mut later = reply("150.0", "100.0", "U8", "the answer, corrected again");
6530 later.edited = Some("450.0".to_owned());
6531 host.seed_replies(
6532 "C1",
6533 "100.0",
6534 vec![
6535 later,
6536 reply("300.0", "100.0", "U8", "the second answer"),
6537 reply("500.0", "100.0", "U8", "a newer answer"),
6538 ],
6539 );
6540 sweep(&host, &host, "acme").await.expect("the first sweep after a restart");
6541 assert_eq!(
6542 host.replies_oldest().last(),
6543 Some(&None),
6544 "the walk reads the thread whole: {:?}",
6545 host.replies_oldest(),
6546 );
6547 assert!(
6548 host.delivered().iter().any(|message| message.text == "the answer, corrected again"),
6549 "an edit below the cursor but above the stamp still arrives: {:?}",
6550 host.delivered(),
6551 );
6552 }
6553
6554 #[test]
6558 fn the_fake_host_seeds_an_unstamped_thread_row_from_its_cursor() {
6559 let host = FakeHost::default();
6560 host.set_thread_watermark("acme", "C1", "100.0", "500.0");
6561
6562 assert_eq!(
6563 host.thread_edit_watermark("acme", "C1", "100.0").as_deref(),
6564 Some("500.0"),
6565 "a row with no stamp reads its cursor, as the real host does",
6566 );
6567 }
6568
6569 #[tokio::test]
6572 async fn the_whole_read_runs_on_its_own_clock_not_every_tick() {
6573 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6574 "acme",
6575 "forge",
6576 Some("tester"),
6577 "C1",
6578 )]);
6579 host.set_watermark("acme", "C1", "300.0");
6580 host.seed_history("C1", vec![]);
6581 host.seed_replies("C1", "100.0", vec![reply("150.0", "100.0", "U8", "the answer")]);
6582 host.follow_thread("acme", "C1", "100.0", thread_owner("forge", Some("tester")), "050.0");
6583 host.mark_thread_participating("acme", "C1", "100.0");
6584
6585 sweep(&host, &host, "acme").await.expect("the first walk");
6586 assert!(
6587 host.replies_oldest().last().expect("a replies call").is_none(),
6588 "the first walk reads whole: {:?}",
6589 host.replies_oldest(),
6590 );
6591
6592 sweep(&host, &host, "acme").await.expect("the cheap tick");
6593 assert!(
6594 host.replies_oldest().last().expect("a replies call").is_some(),
6595 "the tick after a whole read is the cheap one: {:?}",
6596 host.replies_oldest(),
6597 );
6598
6599 host.age_thread_full_read("acme", "C1", "100.0", Duration::from_secs(601));
6600 sweep(&host, &host, "acme").await.expect("the aged tick");
6601 assert!(
6602 host.replies_oldest().last().expect("a replies call").is_none(),
6603 "an aged stamp makes the read whole again: {:?}",
6604 host.replies_oldest(),
6605 );
6606 }
6607
6608 #[tokio::test]
6611 async fn the_sweep_reads_at_most_two_threads_whole() {
6612 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6613 "acme",
6614 "forge",
6615 Some("tester"),
6616 "C1",
6617 )]);
6618 host.set_watermark("acme", "C1", "300.0");
6619 host.seed_history("C1", vec![]);
6620 for (parent, reply_ts) in [("100.0", "150.0"), ("200.0", "250.0"), ("400.0", "450.0")] {
6621 host.seed_replies("C1", parent, vec![reply(reply_ts, parent, "U8", "the answer")]);
6622 host.follow_thread(
6623 "acme",
6624 "C1",
6625 parent,
6626 thread_owner("forge", Some("tester")),
6627 "050.0",
6628 );
6629 host.mark_thread_participating("acme", "C1", parent);
6630 }
6631 sweep(&host, &host, "acme").await.expect("the first walks");
6632 for parent in ["100.0", "200.0", "400.0"] {
6633 host.age_thread_full_read("acme", "C1", parent, Duration::from_secs(601));
6634 }
6635
6636 let before = host.replies_calls();
6637 sweep(&host, &host, "acme").await.expect("the budgeted sweep");
6638 let calls = host.replies_oldest();
6639 let tail = &calls[before..];
6640 let whole = tail.iter().filter(|oldest| oldest.is_none()).count();
6641 assert_eq!(
6642 whole, MAX_FULL_THREAD_READS_PER_SWEEP,
6643 "the budget caps whole reads per sweep: {tail:?}",
6644 );
6645
6646 let before = host.replies_calls();
6647 sweep(&host, &host, "acme").await.expect("the next sweep");
6648 let calls = host.replies_oldest();
6649 let whole = calls[before..].iter().filter(|oldest| oldest.is_none()).count();
6650 assert_eq!(whole, 1, "the thread the budget skipped is read whole next tick");
6651 }
6652
6653 #[tokio::test]
6656 async fn a_throttled_first_walk_keeps_the_replay_guard() {
6657 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6658 "acme",
6659 "forge",
6660 Some("tester"),
6661 "C1",
6662 )]);
6663 host.set_watermark("acme", "C1", "300.0");
6664 host.seed_history("C1", vec![]);
6665 let parent = parent_with_latest_reply("100.0", 2, "200.0");
6666 let mut corrected = reply("150.0", "100.0", "U8", "the answer, corrected");
6667 corrected.edited = Some("990.0".to_owned());
6668 host.seed_replies("C1", "100.0", vec![parent, corrected]);
6669 host.follow_thread("acme", "C1", "100.0", thread_owner("forge", Some("tester")), "200.0");
6670 host.mark_thread_participating("acme", "C1", "100.0");
6671 host.rate_limit_replies_at(0);
6672
6673 let first = sweep(&host, &host, "acme").await.expect("a 429 is not an error");
6674 assert_eq!(first.rate_limited, Some(Duration::from_secs(5)));
6675 assert!(
6676 host.thread_first_walks.lock().expect("lock").contains("acme/C1/100.0"),
6677 "the first-walk mark survives the 429",
6678 );
6679
6680 host.age_thread_full_read("acme", "C1", "100.0", Duration::from_secs(601));
6681 sweep(&host, &host, "acme").await.expect("the retry");
6682 assert!(
6683 !host.delivered().iter().any(|message| message.ts == "150.0"),
6684 "the retried first walk still filters by ts: {:?}",
6685 host.delivered(),
6686 );
6687 }
6688
6689 #[tokio::test]
6692 async fn a_throttled_author_lookup_inside_a_thread_does_not_end_the_sweep() {
6693 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6694 "acme",
6695 "forge",
6696 Some("tester"),
6697 "C1",
6698 )]);
6699 host.set_watermark("acme", "C1", "300.0");
6700 host.seed_history("C1", vec![]);
6701 host.set_user_name("acme", "U9", "colleague");
6702 host.seed_replies("C1", "100.0", vec![reply("150.0", "100.0", "U8", "the answer")]);
6703 host.seed_replies("C1", "200.0", vec![reply("250.0", "200.0", "U9", "the other answer")]);
6704 for parent in ["100.0", "200.0"] {
6705 host.follow_thread(
6706 "acme",
6707 "C1",
6708 parent,
6709 thread_owner("forge", Some("tester")),
6710 "050.0",
6711 );
6712 host.mark_thread_participating("acme", "C1", parent);
6713 }
6714 host.rate_limit_user_info(Duration::from_secs(7));
6715
6716 let outcome = sweep(&host, &host, "acme").await.expect("a 429 is not an error");
6717
6718 assert_eq!(
6719 outcome.rate_limited,
6720 Some(Duration::from_secs(7)),
6721 "the throttle reaches the pump",
6722 );
6723 assert_eq!(host.replies_calls(), 2, "the walk moved on to the second thread");
6724 assert_eq!(
6725 host.delivered().len(),
6726 1,
6727 "the thread whose author is known still delivered: {:?}",
6728 host.delivered(),
6729 );
6730 }
6731
6732 #[tokio::test]
6735 async fn a_thread_past_the_page_cap_holds_its_cursor() {
6736 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6737 "acme",
6738 "forge",
6739 Some("tester"),
6740 "C1",
6741 )]);
6742 host.set_watermark("acme", "C1", "300.0");
6743 host.seed_history("C1", vec![]);
6744 let pages: Vec<Vec<SlackHistoryMessage>> = (0..=MAX_REPLY_PAGES)
6745 .map(|n| vec![reply(&format!("{}.0", 200 + n), "100.0", "U8", "a reply")])
6746 .collect();
6747 host.seed_replies_pages("C1", "100.0", pages);
6748 host.follow_thread("acme", "C1", "100.0", thread_owner("forge", Some("tester")), "050.0");
6749 host.mark_thread_participating("acme", "C1", "100.0");
6750
6751 sweep(&host, &host, "acme").await.expect("a truncated walk is not a sweep error");
6752
6753 assert_eq!(
6754 host.thread_cursor("acme", "C1", "100.0").as_deref(),
6755 Some("050.0"),
6756 "a partial read cannot advance the cursor",
6757 );
6758 assert!(host.delivered().is_empty(), "nothing from a partial read is delivered");
6759 assert!(
6760 host.thread_first_walks.lock().expect("lock").contains("acme/C1/100.0"),
6761 "and the first walk is not spent on a refusal",
6762 );
6763 assert!(
6764 host.thread_full_read_at("acme", "C1", "100.0").is_some(),
6765 "the failed attempt is stamped, so the next tick does not walk 200 pages again",
6766 );
6767
6768 let calls = host.replies_calls();
6771 sweep(&host, &host, "acme").await.expect("the tick after the refusal");
6772 assert_eq!(
6773 host.replies_calls(),
6774 calls,
6775 "a refused first walk backs off instead of re-walking 200 pages",
6776 );
6777 }
6778
6779 #[tokio::test]
6782 async fn a_throttled_own_name_in_the_mention_sweep_backs_off() {
6783 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
6784 let mut hit = search_match("200.1", "C1", "note to self <@U1>");
6785 hit.user = Some("U1".to_owned());
6786 hit.username = Some("Forge".to_owned());
6787 host.seed_search(vec![hit]);
6788 host.rate_limit_auth_test(Duration::from_secs(9));
6789
6790 let outcome = sweep_mentions(&host, &host, "acme").await.expect("a 429 is not an error");
6791
6792 assert_eq!(
6793 outcome.rate_limited,
6794 Some(Duration::from_secs(9)),
6795 "the throttle reaches the pump",
6796 );
6797 assert!(host.delivered().is_empty(), "nothing is delivered without the name");
6798 }
6799
6800 #[tokio::test]
6806 async fn a_reply_arriving_on_the_sweep_after_the_trigger_delivers_to_the_owning_session() {
6807 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6808 "acme",
6809 "forge",
6810 Some("tester"),
6811 "C1",
6812 )]);
6813 let parent = parent_with_replies("100.0", 2);
6814 host.set_watermark("acme", "C1", "050.0");
6815 host.seed_history("C1", vec![parent.clone()]);
6816 seed_user_participation(&host, &parent);
6817
6818 sweep(&host, &host, "acme").await.expect("trigger sweep");
6819 assert_eq!(host.delivered()[0].ts, "100.0", "the trigger itself is delivered");
6820 assert_eq!(
6821 host.thread_cursor("acme", "C1", "100.0"),
6822 Some("105.0".to_owned()),
6823 "and the walk has read his own reply, so the cursor is past it",
6824 );
6825
6826 let mut arriving = history_message("150.0", "U8", "the reply");
6831 arriving.edited = Some("160.0".to_owned());
6832 host.seed_replies(
6833 "C1",
6834 "100.0",
6835 vec![parent, history_message("105.0", "U1", "the user's own reply"), arriving],
6836 );
6837 sweep(&host, &host, "acme").await.expect("follow-up sweep");
6838
6839 let delivered = host.delivered();
6840 let replies: Vec<&SlackMessage> =
6841 delivered.iter().filter(|message| message.ts == "150.0").collect();
6842 assert_eq!(replies.len(), 1, "the reply is delivered once");
6843 assert_eq!(replies[0].thread_ts.as_deref(), Some("100.0"), "as a reply of its thread");
6844 assert_eq!(
6845 replies[0].edited.as_deref(),
6846 Some("160.0"),
6847 "and the walk's delivery carries the edit ts",
6848 );
6849 assert_eq!(
6850 host.delivered_owners().last().map(|owner| owner.team_role.clone()),
6851 Some(Some("tester".to_owned())),
6852 "the reply reaches the OWNING session",
6853 );
6854 assert_eq!(
6855 delivered.iter().filter(|message| message.ts == "100.0").count(),
6856 1,
6857 "the repeated parent is deduped",
6858 );
6859 assert_eq!(
6860 host.thread_cursor("acme", "C1", "100.0"),
6861 Some("160.0".to_owned()),
6862 "the thread cursor clears the delivered reply's edit too, so the next tick does not \
6863 re-offer it",
6864 );
6865 }
6866
6867 #[tokio::test]
6872 async fn every_owner_gets_a_thread_reply_and_a_failure_holds_the_cursor() {
6873 let host = FakeHost::with_subscriptions(vec![
6874 sub_conversation_owned_by("acme", "forge", Some("a"), "C1"),
6875 sub_conversation_owned_by("acme", "forge", Some("b"), "C1"),
6876 ]);
6877 let parent = parent_with_replies("100.0", 1);
6878 host.set_watermark("acme", "C1", "050.0");
6879 host.seed_history("C1", vec![parent.clone()]);
6880 seed_user_participation(&host, &parent);
6881 sweep(&host, &host, "acme").await.expect("trigger sweep");
6882 assert_eq!(
6883 host.followed_threads("acme", "C1").first().map(|thread| thread.owners.len()),
6884 Some(2),
6885 "both sessions own the thread after both were delivered the trigger",
6886 );
6887
6888 host.seed_replies(
6889 "C1",
6890 "100.0",
6891 vec![
6892 parent,
6893 history_message("105.0", "U1", "the user's own reply"),
6894 history_message("150.0", "U8", "the reply"),
6895 ],
6896 );
6897 host.fail_delivery_at(2);
6901 sweep(&host, &host, "acme").await.expect("follow-up sweep");
6902
6903 assert_eq!(
6904 host.delivered().iter().filter(|message| message.ts == "150.0").count(),
6905 1,
6906 "the owner whose delivery succeeded still received the reply",
6907 );
6908 assert_eq!(
6909 host.thread_cursor("acme", "C1", "100.0"),
6910 Some("105.0".to_owned()),
6911 "the cursor stays below a reply that did not reach every owner",
6912 );
6913 }
6914
6915 #[tokio::test]
6920 async fn a_mentions_only_subscription_owns_no_thread() {
6921 let host =
6922 FakeHost::with_subscriptions(vec![sub_for(SlackSubscriptionTarget::Conversation {
6923 id: "C1".to_owned(),
6924 name: None,
6925 mode: SlackWatchMode::MentionsOnly,
6926 })]);
6927 let parent = parent_with_replies("100.0", 2);
6928 host.set_watermark("acme", "C1", "050.0");
6929 host.seed_history("C1", vec![parent.clone()]);
6930 host.seed_replies(
6931 "C1",
6932 "100.0",
6933 vec![parent.clone(), history_message("105.0", "U1", "the user's own reply")],
6934 );
6935 sweep(&host, &host, "acme").await.expect("trigger sweep");
6936
6937 assert!(
6938 host.followed_threads("acme", "C1").is_empty(),
6939 "the mentions subscription owns no thread",
6940 );
6941
6942 host.seed_history("C1", vec![]);
6943 host.seed_replies(
6944 "C1",
6945 "100.0",
6946 vec![
6947 parent,
6948 history_message("105.0", "U1", "the user's own reply"),
6949 reply("150.0", "100.0", "U8", "an answer"),
6950 ],
6951 );
6952 sweep(&host, &host, "acme").await.expect("follow-up sweep");
6953
6954 assert!(
6955 !host.delivered().iter().any(|message| message.ts == "150.0"),
6956 "and a reply in it reaches nobody who asked for mentions only: {:?}",
6957 host.delivered(),
6958 );
6959 }
6960
6961 #[tokio::test]
6965 async fn a_thread_walks_replies_arrive_as_one_block_per_owner() {
6966 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
6967 "acme",
6968 "forge",
6969 Some("tester"),
6970 "C1",
6971 )]);
6972 let parent = parent_with_replies("100.0", 3);
6973 host.set_watermark("acme", "C1", "050.0");
6974 host.seed_history("C1", vec![parent.clone()]);
6975 seed_user_participation(&host, &parent);
6976 sweep(&host, &host, "acme").await.expect("trigger sweep");
6977 let blocks_before = host.batches().len();
6978
6979 host.seed_replies(
6980 "C1",
6981 "100.0",
6982 vec![
6983 parent,
6984 history_message("105.0", "U1", "the user's own reply"),
6985 reply("150.0", "100.0", "U8", "one"),
6986 reply("160.0", "100.0", "U8", "two"),
6987 reply("170.0", "100.0", "U8", "three"),
6988 ],
6989 );
6990 sweep(&host, &host, "acme").await.expect("follow-up sweep");
6991
6992 let batches = host.batches();
6993 assert_eq!(batches.len(), blocks_before + 1, "one block, never one per reply");
6994 assert_eq!(batches.last().map(Vec::len), Some(3), "holding every reply the walk had");
6995 }
6996
6997 #[tokio::test]
7001 async fn a_thread_the_user_is_only_mentioned_in_delivers() {
7002 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7003 "acme",
7004 "forge",
7005 Some("tester"),
7006 "C1",
7007 )]);
7008 let parent = parent_with_latest_reply("100.0", 2, "150.0");
7009 host.set_watermark("acme", "C1", "050.0");
7010 host.seed_history("C1", vec![parent.clone()]);
7011 host.seed_replies(
7012 "C1",
7013 "100.0",
7014 vec![
7015 parent,
7016 history_message("105.0", "U9", "hey <@U1>, look at this"),
7017 reply("150.0", "100.0", "U8", "the answer"),
7018 ],
7019 );
7020
7021 sweep(&host, &host, "acme").await.expect("sweep");
7022
7023 assert!(
7024 host.delivered().iter().any(|message| message.ts == "150.0"),
7025 "a thread he is named in delivers what follows: {:?}",
7026 host.delivered(),
7027 );
7028 }
7029
7030 #[tokio::test]
7036 async fn a_thread_first_followed_does_not_start_below_the_conversation_cursor() {
7037 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7038 "acme",
7039 "forge",
7040 Some("tester"),
7041 "C1",
7042 )]);
7043 let parent = parent_with_latest_reply("100.0", 2, "150.0");
7044 host.set_watermark("acme", "C1", "200.0");
7045 host.seed_history("C1", vec![parent.clone()]);
7046 host.seed_replies(
7047 "C1",
7048 "100.0",
7049 vec![
7050 parent.clone(),
7051 history_message("120.0", "U1", "the user's own reply"),
7052 reply("150.0", "100.0", "U8", "posted before the subscription"),
7053 ],
7054 );
7055
7056 sweep(&host, &host, "acme").await.expect("sweep");
7057
7058 assert!(
7059 !host.delivered().iter().any(|message| message.ts == "150.0"),
7060 "a reply older than the conversation cursor is not news: {:?}",
7061 host.delivered(),
7062 );
7063 assert!(
7064 host.thread_cursor("acme", "C1", "100.0")
7065 .as_deref()
7066 .is_some_and(|cursor| cursor == "200.0"),
7067 "the first follow never starts below the conversation cursor: {:?}",
7068 host.thread_cursor("acme", "C1", "100.0"),
7069 );
7070
7071 let moved = parent_with_latest_reply("100.0", 3, "250.0");
7074 host.seed_history("C1", vec![moved.clone()]);
7075 host.seed_replies(
7076 "C1",
7077 "100.0",
7078 vec![
7079 moved,
7080 history_message("120.0", "U1", "the user's own reply"),
7081 reply("150.0", "100.0", "U8", "posted before the subscription"),
7082 reply("250.0", "100.0", "U8", "posted after the subscription"),
7083 ],
7084 );
7085
7086 sweep(&host, &host, "acme").await.expect("moved sweep");
7087
7088 let delivered = host.delivered();
7089 assert!(
7090 delivered.iter().any(|message| message.ts == "250.0"),
7091 "the first follow still delivers a thread that moves: {delivered:?}",
7092 );
7093 assert!(
7094 !delivered.iter().any(|message| message.ts == "150.0"),
7095 "the reply behind the cursor stays out: {delivered:?}",
7096 );
7097 }
7098
7099 #[tokio::test]
7103 async fn a_thread_first_followed_above_the_cursor_seeds_below_its_newest_reply() {
7104 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7105 "acme",
7106 "forge",
7107 Some("tester"),
7108 "C1",
7109 )]);
7110 let parent = parent_with_latest_reply("100.0", 3, "250.0");
7111 host.set_watermark("acme", "C1", "150.0");
7112 host.seed_history("C1", vec![parent.clone()]);
7113 host.seed_replies(
7114 "C1",
7115 "100.0",
7116 vec![
7117 parent,
7118 history_message("200.0", "U1", "the user's own reply"),
7119 reply("230.0", "100.0", "U8", "older than the newest reply"),
7120 reply("250.0", "100.0", "U8", "the reply that brought the thread in"),
7121 ],
7122 );
7123
7124 sweep(&host, &host, "acme").await.expect("sweep");
7125
7126 let delivered_messages = host.delivered();
7127 let delivered: Vec<&str> =
7128 delivered_messages.iter().map(|message| message.ts.as_str()).collect();
7129 assert!(delivered.contains(&"250.0"), "the newest reply is delivered: {delivered:?}");
7130 assert!(
7131 !delivered.contains(&"230.0"),
7132 "a reply below the newest reply is not news on a first follow: {delivered:?}",
7133 );
7134 }
7135
7136 #[tokio::test]
7140 async fn an_existing_thread_keeps_its_cursor_when_the_conversation_moves() {
7141 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7142 "acme",
7143 "forge",
7144 Some("tester"),
7145 "C1",
7146 )]);
7147 let parent = parent_with_latest_reply("100.0", 2, "250.0");
7148 host.set_watermark("acme", "C1", "300.0");
7149 host.seed_history("C1", vec![parent.clone()]);
7150 host.seed_replies(
7151 "C1",
7152 "100.0",
7153 vec![
7154 parent,
7155 history_message("160.0", "U1", "the user's own reply"),
7156 reply("170.0", "100.0", "U8", "between the thread and conversation cursors"),
7157 reply("250.0", "100.0", "U8", "the newest reply"),
7158 ],
7159 );
7160 host.follow_thread("acme", "C1", "100.0", thread_owner("forge", Some("tester")), "150.0");
7161 host.mark_thread_participating("acme", "C1", "100.0");
7162
7163 sweep(&host, &host, "acme").await.expect("sweep");
7164
7165 let delivered = host.delivered();
7166 assert!(
7167 delivered.iter().any(|message| message.ts == "170.0"),
7168 "the row's own cursor governs, not the conversation's: {delivered:?}",
7169 );
7170 assert!(
7171 delivered.iter().any(|message| message.ts == "250.0"),
7172 "and the thread keeps delivering: {delivered:?}",
7173 );
7174 }
7175
7176 #[tokio::test]
7179 async fn a_thread_the_user_is_absent_from_delivers_nothing() {
7180 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7181 "acme",
7182 "forge",
7183 Some("tester"),
7184 "C1",
7185 )]);
7186 let parent = parent_with_replies("100.0", 2);
7187 host.set_watermark("acme", "C1", "050.0");
7188 host.seed_history("C1", vec![parent.clone()]);
7189 sweep(&host, &host, "acme").await.expect("trigger sweep");
7190
7191 host.seed_history("C1", vec![]);
7192 host.seed_replies(
7193 "C1",
7194 "100.0",
7195 vec![
7196 parent,
7197 history_message("105.0", "U9", "one"),
7198 reply("150.0", "100.0", "U8", "two"),
7199 ],
7200 );
7201 sweep(&host, &host, "acme").await.expect("follow-up sweep");
7202
7203 assert!(
7204 !host.delivered().iter().any(|message| message.ts == "150.0"),
7205 "a thread the user is not in stays silent: {:?}",
7206 host.delivered(),
7207 );
7208 }
7209
7210 #[tokio::test]
7214 async fn a_user_reply_behind_the_thread_cursor_still_marks_the_thread() {
7215 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7216 "acme",
7217 "forge",
7218 Some("tester"),
7219 "C1",
7220 )]);
7221 let parent = parent_with_replies("100.0", 2);
7222 host.set_watermark("acme", "C1", "050.0");
7223 host.seed_history("C1", vec![parent.clone()]);
7224 sweep(&host, &host, "acme").await.expect("trigger sweep");
7225
7226 host.set_thread_watermark("acme", "C1", "100.0", "140.0");
7228 host.seed_history("C1", vec![]);
7229 host.seed_replies(
7230 "C1",
7231 "100.0",
7232 vec![
7233 parent,
7234 history_message("105.0", "U1", "the user's own reply"),
7235 reply("150.0", "100.0", "U8", "what was said next"),
7236 ],
7237 );
7238 sweep(&host, &host, "acme").await.expect("follow-up sweep");
7239
7240 assert!(
7241 host.delivered().iter().any(|message| message.ts == "150.0"),
7242 "his reply behind the cursor still puts him in the thread: {:?}",
7243 host.delivered(),
7244 );
7245 }
7246
7247 #[tokio::test]
7256 async fn an_answer_to_a_root_that_had_no_replies_yet_is_delivered() {
7257 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7258 "acme",
7259 "forge",
7260 Some("tester"),
7261 "C1",
7262 )]);
7263 host.set_watermark("acme", "C1", "050.0");
7264 let root = history_message("100.0", "U1", "deploy is green");
7265 host.seed_history("C1", vec![root]);
7266 sweep(&host, &host, "acme").await.expect("the root's sweep");
7267 assert_eq!(
7268 host.delivered().first().map(|message| message.ts.as_str()),
7269 Some("100.0"),
7270 "his own message is delivered: it is his side of the thread",
7271 );
7272 assert!(
7273 host.followed_threads("acme", "C1").is_empty(),
7274 "a message with no replies anchors nothing",
7275 );
7276
7277 let parent = parent_with_latest_reply("100.0", 2, "150.0");
7279 host.seed_history("C1", vec![parent.clone()]);
7280 host.seed_replies(
7281 "C1",
7282 "100.0",
7283 vec![
7284 parent,
7285 history_message("105.0", "U1", "my own reply"),
7286 reply("150.0", "100.0", "U8", "the answer"),
7287 ],
7288 );
7289 sweep(&host, &host, "acme").await.expect("the answer's sweep");
7290
7291 assert!(
7292 host.delivered().iter().any(|message| message.ts == "150.0"),
7293 "the answer to his own reply is delivered: {:?}",
7294 host.delivered(),
7295 );
7296
7297 let calls = host.replies_calls();
7302 sweep(&host, &host, "acme").await.expect("a quiet sweep");
7303 assert_eq!(host.replies_calls(), calls + 1, "a participating thread is read every tick");
7304
7305 let parent = parent_with_latest_reply("100.0", 3, "170.0");
7309 host.seed_history("C1", vec![parent.clone()]);
7310 host.seed_replies(
7311 "C1",
7312 "100.0",
7313 vec![parent, reply("170.0", "100.0", "U8", "the next one")],
7314 );
7315 sweep(&host, &host, "acme").await.expect("another sweep");
7316
7317 assert!(
7318 host.delivered().iter().any(|message| message.ts == "170.0"),
7319 "a thread the user is already known to be in keeps delivering: {:?}",
7320 host.delivered(),
7321 );
7322 }
7323
7324 #[tokio::test]
7328 async fn a_colleagues_reply_under_the_users_own_message_is_delivered() {
7329 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7330 "acme",
7331 "forge",
7332 Some("tester"),
7333 "C1",
7334 )]);
7335 let mut mine = parent_with_replies("100.0", 1);
7336 mine.user = Some("U1".to_owned());
7337 host.set_watermark("acme", "C1", "050.0");
7338 host.seed_history("C1", vec![mine.clone()]);
7339 sweep(&host, &host, "acme").await.expect("trigger sweep");
7340 assert_eq!(
7341 host.delivered().first().map(|message| message.ts.as_str()),
7342 Some("100.0"),
7343 "his own message is delivered",
7344 );
7345
7346 host.seed_history("C1", vec![]);
7347 host.seed_replies("C1", "100.0", vec![mine, reply("150.0", "100.0", "U8", "the answer")]);
7348 sweep(&host, &host, "acme").await.expect("follow-up sweep");
7349
7350 assert!(
7351 host.delivered().iter().any(|message| message.ts == "150.0"),
7352 "the answer to his own message arrives: {:?}",
7353 host.delivered(),
7354 );
7355 }
7356
7357 #[tokio::test]
7360 async fn a_settled_absent_author_is_looked_up_once() {
7361 let host =
7362 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
7363 host.set_watermark("acme", "C1", "050.0");
7364 host.seed_history("C1", vec![history_message("200.0", "U9", "first")]);
7365 sweep(&host, &host, "acme").await.expect("first sweep");
7366 assert_eq!(host.user_info_calls(), 1, "one author, one lookup");
7367
7368 host.seed_history("C1", vec![history_message("400.0", "U9", "a later one")]);
7369 sweep(&host, &host, "acme").await.expect("second sweep");
7370 assert_eq!(
7371 host.user_info_calls(),
7372 1,
7373 "and the settled answer is kept, or every message from him costs a call",
7374 );
7375 }
7376
7377 #[tokio::test]
7380 async fn a_transient_author_lookup_failure_is_retried_not_kept() {
7381 let host =
7382 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
7383 host.set_watermark("acme", "C1", "050.0");
7384 host.seed_history("C1", vec![history_message("200.0", "U9", "first")]);
7385 host.fail_user_info_transiently();
7386
7387 sweep(&host, &host, "acme").await.expect("a failed lookup is not a failed sweep");
7388 assert_eq!(host.user_info_calls(), 1, "the author is asked about");
7389 assert!(
7390 host.resolved_names().is_empty(),
7391 "and nothing about him is remembered: {:?}",
7392 host.resolved_names(),
7393 );
7394
7395 host.seed_history("C1", vec![history_message("400.0", "U9", "a later one")]);
7396 sweep(&host, &host, "acme").await.expect("the next sweep");
7397 assert_eq!(host.user_info_calls(), 2, "so the next sweep asks again");
7398 }
7399
7400 #[tokio::test]
7404 async fn two_messages_from_a_failing_author_cost_one_lookup() {
7405 let host =
7406 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
7407 host.set_watermark("acme", "C1", "050.0");
7408 host.seed_history(
7409 "C1",
7410 vec![history_message("300.0", "U9", "second"), history_message("200.0", "U9", "first")],
7411 );
7412 host.fail_user_info_transiently();
7413
7414 sweep(&host, &host, "acme").await.expect("sweep");
7415 assert_eq!(host.user_info_calls(), 1, "one author, one lookup, two messages");
7416 }
7417
7418 #[tokio::test]
7421 async fn a_rate_limited_author_lookup_is_reported_to_the_pump() {
7422 let host =
7423 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
7424 host.set_watermark("acme", "C1", "050.0");
7425 host.seed_history("C1", vec![history_message("200.0", "U9", "hello")]);
7426 host.rate_limit_user_info(Duration::from_secs(9));
7427
7428 let outcome = sweep(&host, &host, "acme").await.expect("a 429 is not an error");
7429 assert_eq!(
7430 outcome.rate_limited,
7431 Some(Duration::from_secs(9)),
7432 "a throttled lookup arms the pump's backoff instead of being swallowed",
7433 );
7434 }
7435
7436 #[tokio::test]
7439 async fn a_rate_limited_auth_test_is_reported_to_the_pump() {
7440 let host =
7441 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
7442 host.set_watermark("acme", "C1", "050.0");
7443 let mut message = history_message("200.0", "U1", "hello");
7444 message.bot_name = Some("Forge".to_owned());
7445 host.seed_history("C1", vec![message]);
7446 host.rate_limit_auth_test(Duration::from_secs(7));
7447
7448 let outcome = sweep(&host, &host, "acme").await.expect("a 429 is not an error");
7449 assert_eq!(
7450 outcome.rate_limited,
7451 Some(Duration::from_secs(7)),
7452 "a throttled auth.test arms the pump's backoff instead of being swallowed",
7453 );
7454 }
7455
7456 #[tokio::test]
7461 async fn a_thread_the_user_is_not_in_is_read_once_then_left_alone() {
7462 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7463 "acme",
7464 "forge",
7465 Some("tester"),
7466 "C1",
7467 )]);
7468 let parent = parent_with_replies("100.0", 2);
7469 host.set_watermark("acme", "C1", "050.0");
7470 host.seed_history("C1", vec![parent.clone()]);
7471 sweep(&host, &host, "acme").await.expect("trigger sweep");
7472 assert_eq!(host.thread_cursor("acme", "C1", "100.0"), Some("100.0".to_owned()));
7473
7474 let parent = parent_with_latest_reply("100.0", 2, "150.0");
7475 host.seed_history("C1", vec![parent.clone()]);
7476 host.seed_replies(
7477 "C1",
7478 "100.0",
7479 vec![
7480 parent,
7481 history_message("105.0", "U9", "one"),
7482 reply("150.0", "100.0", "U8", "someone else"),
7483 ],
7484 );
7485 sweep(&host, &host, "acme").await.expect("follow-up sweep");
7486 assert_eq!(
7487 host.thread_cursor("acme", "C1", "100.0"),
7488 Some("150.0".to_owned()),
7489 "the cursor advances over what was read, delivered or not",
7490 );
7491
7492 let calls = host.replies_calls();
7493 sweep(&host, &host, "acme").await.expect("a quiet sweep");
7494 assert_eq!(host.replies_calls(), calls, "and the thread is not read again until it moves");
7495 }
7496
7497 #[tokio::test]
7502 async fn a_thread_walks_replies_carry_a_resolved_author() {
7503 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7504 "acme",
7505 "forge",
7506 Some("tester"),
7507 "C1",
7508 )]);
7509 let parent = parent_with_replies("100.0", 2);
7510 host.set_watermark("acme", "C1", "050.0");
7511 host.seed_user_name("U8", "colleague");
7512 host.seed_history("C1", vec![parent.clone()]);
7513 seed_user_participation(&host, &parent);
7514 sweep(&host, &host, "acme").await.expect("trigger sweep");
7515
7516 host.seed_history("C1", vec![]);
7517 host.seed_replies(
7518 "C1",
7519 "100.0",
7520 vec![
7521 parent,
7522 history_message("105.0", "U1", "the user's own reply"),
7523 reply("150.0", "100.0", "U8", "what was said next"),
7524 ],
7525 );
7526 sweep(&host, &host, "acme").await.expect("follow-up sweep");
7527
7528 assert_eq!(
7529 host.delivered()
7530 .iter()
7531 .find(|message| message.ts == "150.0")
7532 .and_then(|message| message.author.as_deref()),
7533 Some("colleague"),
7534 "the reply's author is a name, not the id the block would drop",
7535 );
7536 }
7537
7538 #[tokio::test]
7543 async fn a_thread_the_user_replies_in_after_the_parent_left_the_page_still_delivers() {
7544 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7545 "acme",
7546 "forge",
7547 Some("tester"),
7548 "C1",
7549 )]);
7550 let parent = parent_with_replies("100.0", 1);
7551 host.set_watermark("acme", "C1", "050.0");
7552 host.seed_history("C1", vec![parent.clone()]);
7553 host.seed_replies("C1", "100.0", vec![parent.clone()]);
7554 sweep(&host, &host, "acme").await.expect("trigger sweep");
7555 assert_eq!(host.delivered().len(), 1, "the parent itself is delivered");
7556
7557 host.seed_history("C1", vec![]);
7560 host.seed_replies(
7561 "C1",
7562 "100.0",
7563 vec![
7564 parent,
7565 reply("150.0", "100.0", "U1", "my own reply"),
7566 reply("160.0", "100.0", "U8", "the answer"),
7567 ],
7568 );
7569 sweep(&host, &host, "acme").await.expect("follow-up sweep");
7570
7571 assert!(
7572 host.delivered().iter().any(|message| message.ts == "160.0"),
7573 "the answer to the user's own reply is delivered: {:?}",
7574 host.delivered(),
7575 );
7576 }
7577
7578 fn seed_user_participation(host: &FakeHost, parent: &SlackHistoryMessage) {
7583 host.seed_replies(
7584 "C1",
7585 &parent.ts,
7586 vec![parent.clone(), history_message("105.0", "U1", "the user's own reply")],
7587 );
7588 }
7589
7590 #[tokio::test]
7593 async fn a_mention_delivers_the_username_slack_sent() {
7594 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
7595 let mut hit = search_match("200.1", "C1", "ping <@U1>");
7596 hit.username = Some("architect".to_owned());
7597 host.seed_search(vec![hit]);
7598
7599 sweep_mentions(&host, &host, "acme").await.expect("sweep");
7600 assert_eq!(
7601 host.delivered()[0].author.as_deref(),
7602 Some("architect"),
7603 "the handle Slack sent is what the block shows",
7604 );
7605 }
7606
7607 #[tokio::test]
7613 async fn the_users_own_mention_takes_his_name_over_the_apps() {
7614 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
7615 let mut first = search_match("300.2", "C1", "note to self <@U1>");
7616 first.user = Some("U1".to_owned());
7617 first.username = Some("Forge".to_owned());
7618 let mut second = search_match("200.1", "C2", "another <@U1>");
7619 second.user = Some("U1".to_owned());
7620 second.username = Some("Forge".to_owned());
7621 host.seed_search(vec![first, second]);
7622
7623 sweep_mentions(&host, &host, "acme").await.expect("sweep");
7624
7625 let delivered = host.delivered();
7626 assert_eq!(delivered.len(), 2, "both of his own mentions arrive");
7627 assert!(
7628 delivered.iter().all(|message| message.author.as_deref() == Some("tester")),
7629 "his own mentions carry his handle, not the app's name: {delivered:?}",
7630 );
7631 assert_eq!(host.auth_test_calls(), 1, "one identity for the sweep, not one per hit");
7632 }
7633
7634 #[tokio::test]
7637 async fn a_conversation_message_resolves_its_author_once_per_user() {
7638 let host =
7639 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
7640 host.set_watermark("acme", "C1", "050.0");
7641 host.seed_user_name("U9", "ved");
7642 host.seed_history(
7643 "C1",
7644 vec![history_message("300.0", "U9", "second"), history_message("200.0", "U9", "first")],
7645 );
7646
7647 sweep(&host, &host, "acme").await.expect("sweep");
7648
7649 let delivered = host.delivered();
7650 assert_eq!(delivered.len(), 2, "both messages are delivered");
7651 assert!(
7652 delivered.iter().all(|message| message.author.as_deref() == Some("ved")),
7653 "each carries the resolved handle: {delivered:?}",
7654 );
7655 assert_eq!(host.user_info_calls(), 1, "the second message costs no lookup");
7656 }
7657
7658 #[tokio::test]
7660 async fn a_bot_message_takes_its_name_from_the_payload() {
7661 let host =
7662 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
7663 host.set_watermark("acme", "C1", "050.0");
7664 let mut message = history_message("300.0", "U0ATEK2EAGP", "transfer");
7665 message.bot_name = Some("architect2".to_owned());
7666 host.seed_history("C1", vec![message]);
7667
7668 sweep(&host, &host, "acme").await.expect("sweep");
7669
7670 assert_eq!(
7671 host.delivered()[0].author.as_deref(),
7672 Some("architect2"),
7673 "a bot names itself, and the block shows that",
7674 );
7675 assert_eq!(host.user_info_calls(), 0, "a bot's name is free");
7676 }
7677
7678 #[tokio::test]
7682 async fn the_users_own_message_takes_his_name_over_the_apps() {
7683 let host = FakeHost::with_subscriptions(vec![
7684 sub_channel("acme", "C1", SlackWatchMode::All),
7685 sub_channel("acme", "C2", SlackWatchMode::All),
7686 ]);
7687 host.set_watermark("acme", "C1", "050.0");
7688 host.set_watermark("acme", "C2", "050.0");
7689 let mut first = history_message("200.0", "U1", "once I do e2e testing, we should be good");
7690 first.bot_name = Some("Forge".to_owned());
7691 let mut second = history_message("300.0", "U1", "then production");
7692 second.bot_name = Some("Forge".to_owned());
7693 host.seed_history("C1", vec![first]);
7694 host.seed_history("C2", vec![second]);
7695
7696 sweep(&host, &host, "acme").await.expect("sweep");
7697
7698 let delivered = host.delivered();
7699 assert_eq!(delivered.len(), 2, "both conversations are delivered");
7700 assert!(
7701 delivered.iter().all(|message| message.author.as_deref() == Some("tester")),
7702 "his own messages carry his handle, not the app's name: {delivered:?}",
7703 );
7704 assert_eq!(
7705 host.auth_test_calls(),
7706 1,
7707 "one identity for the sweep, not one per message or per conversation",
7708 );
7709 assert_eq!(host.user_info_calls(), 0, "his own handle costs no users.info call");
7710 }
7711
7712 #[tokio::test]
7715 async fn a_later_failed_auth_test_keeps_the_name_already_learned() {
7716 let host =
7717 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
7718 host.set_watermark("acme", "C1", "050.0");
7719 let mut first = history_message("200.0", "U1", "first");
7720 first.bot_name = Some("Forge".to_owned());
7721 host.seed_history("C1", vec![first]);
7722 sweep(&host, &host, "acme").await.expect("first sweep");
7723 assert_eq!(
7724 host.delivered().last().and_then(|message| message.author.as_deref()),
7725 Some("tester"),
7726 "the first sweep learns his handle",
7727 );
7728
7729 host.fail_auth_test_transiently();
7730 let mut second = history_message("300.0", "U1", "second");
7731 second.bot_name = Some("Forge".to_owned());
7732 host.seed_history("C1", vec![second]);
7733 sweep(&host, &host, "acme").await.expect("second sweep");
7734 assert_eq!(
7735 host.delivered().last().and_then(|message| message.author.as_deref()),
7736 Some("tester"),
7737 "a failed auth.test does not bring the app's name back; the host remembers his",
7738 );
7739 }
7740
7741 #[tokio::test]
7744 async fn an_unresolvable_author_keeps_no_name() {
7745 let host =
7746 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
7747 host.set_watermark("acme", "C1", "050.0");
7748 host.seed_history("C1", vec![history_message("300.0", "U9", "hello")]);
7749
7750 sweep(&host, &host, "acme").await.expect("sweep");
7751
7752 assert_eq!(host.delivered()[0].author, None, "no name is invented for a failed lookup");
7753 assert_eq!(
7754 host.delivered()[0].user.as_deref(),
7755 Some("U9"),
7756 "and the id is still there for the block",
7757 );
7758 }
7759
7760 #[tokio::test]
7763 async fn a_plain_message_follows_no_thread() {
7764 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7765 "acme", "forge", None, "C1",
7766 )]);
7767 host.set_watermark("acme", "C1", "050.0");
7768 host.seed_history("C1", vec![history_message("100.0", "U9", "plain")]);
7769
7770 sweep(&host, &host, "acme").await.expect("sweep");
7771 assert_eq!(host.delivered().len(), 1, "the message itself is delivered");
7772 assert!(
7773 host.followed_threads("acme", "C1").is_empty(),
7774 "a message with no thread and no replies anchors nothing",
7775 );
7776 }
7777
7778 #[tokio::test]
7782 async fn a_mention_in_a_thread_anchors_that_thread() {
7783 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
7784 host.set_watermark("acme", MENTION_CURSOR, "100.0");
7785 let mut hit = search_match("150.0", "C1", "ping <@U1>");
7786 hit.thread_ts = Some("99.0".to_owned());
7787 host.seed_search(vec![hit]);
7788
7789 sweep_mentions(&host, &host, "acme").await.expect("mention sweep");
7790 let threads = host.followed_threads("acme", "C1");
7791 assert_eq!(threads.len(), 1, "the thread of the mention is tracked");
7792 assert_eq!(threads[0].parent_ts, "99.0");
7793 assert_eq!(threads[0].owners, vec![thread_owner("forge", None)]);
7794 assert_eq!(
7795 host.thread_cursor("acme", "C1", "99.0"),
7796 Some("150.0".to_owned()),
7797 "the cursor starts at the mention's ts, never the weeks-old parent",
7798 );
7799 }
7800
7801 #[tokio::test]
7805 async fn read_thread_returns_the_parent_once_across_pages() {
7806 let host = FakeHost::with_subscriptions(Vec::new());
7807 let parent = parent_with_replies("100.0", 2);
7808 host.seed_replies_pages(
7809 "C1",
7810 "100.0",
7811 vec![
7812 vec![parent.clone(), history_message("110.0", "U1", "first reply")],
7813 vec![parent, history_message("120.0", "U1", "second reply")],
7814 ],
7815 );
7816
7817 let thread = read_thread(&host, "C1", "100.0", 200).await.expect("the thread is read");
7818 let ts: Vec<&str> = thread.iter().map(|message| message.ts.as_str()).collect();
7819 assert_eq!(
7820 ts,
7821 ["100.0", "110.0", "120.0"],
7822 "the parent appears once and every reply follows it in order",
7823 );
7824 }
7825
7826 #[tokio::test]
7829 async fn read_thread_stops_at_the_limit() {
7830 let host = FakeHost::with_subscriptions(Vec::new());
7831 let parent = parent_with_replies("100.0", 2);
7832 host.seed_replies(
7833 "C1",
7834 "100.0",
7835 vec![
7836 parent,
7837 history_message("110.0", "U1", "first reply"),
7838 history_message("120.0", "U1", "second reply"),
7839 ],
7840 );
7841
7842 let thread = read_thread(&host, "C1", "100.0", 2).await.expect("the thread is read");
7843
7844 assert_eq!(thread.len(), 2, "the limit bounds the read");
7845 assert_eq!(thread[1].ts, "110.0", "and keeps the oldest reply that fit");
7846 }
7847
7848 #[tokio::test]
7853 async fn a_thread_walk_pages_to_the_end_deduping_the_parent() {
7854 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7855 "acme",
7856 "forge",
7857 Some("tester"),
7858 "C1",
7859 )]);
7860 let parent = parent_with_replies("100.0", 3);
7861 host.set_watermark("acme", "C1", "050.0");
7862 host.seed_history("C1", vec![parent.clone()]);
7863 seed_user_participation(&host, &parent);
7864 sweep(&host, &host, "acme").await.expect("trigger sweep");
7865
7866 host.seed_replies_pages(
7867 "C1",
7868 "100.0",
7869 vec![
7870 vec![
7871 parent.clone(),
7872 history_message("105.0", "U1", "the user's own reply"),
7873 history_message("300.0", "U8", "reply three"),
7874 ],
7875 vec![
7876 parent,
7877 history_message("200.0", "U8", "reply two"),
7878 history_message("150.0", "U8", "reply one"),
7879 ],
7880 ],
7881 );
7882 sweep(&host, &host, "acme").await.expect("follow-up sweep");
7883
7884 let delivered = host.delivered();
7885 let seen_ts = |ts: &str| delivered.iter().filter(|message| message.ts == ts).count();
7886 assert_eq!(seen_ts("100.0"), 1, "the parent is delivered once, never per page");
7887 assert_eq!(seen_ts("300.0"), 1, "the newest page's reply arrives");
7888 assert_eq!(seen_ts("200.0"), 1, "the older page's replies arrive too");
7889 assert_eq!(seen_ts("150.0"), 1, "nothing is lost at the page boundary");
7890 assert_eq!(
7891 host.thread_cursor("acme", "C1", "100.0"),
7892 Some("300.0".to_owned()),
7893 "the cursor lands on the newest reply",
7894 );
7895 }
7896
7897 #[tokio::test]
7902 async fn the_thread_walk_refuses_the_agents_ledgered_reply_not_the_users_own() {
7903 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7904 "acme",
7905 "forge",
7906 Some("tester"),
7907 "C1",
7908 )]);
7909 let parent = parent_with_replies("100.0", 2);
7910 host.set_watermark("acme", "C1", "050.0");
7911 host.seed_history("C1", vec![parent.clone()]);
7912 seed_user_participation(&host, &parent);
7913 sweep(&host, &host, "acme").await.expect("trigger sweep");
7914 host.record_own_post("acme", "C1", "150.0");
7915
7916 host.seed_replies(
7917 "C1",
7918 "100.0",
7919 vec![
7920 parent,
7921 history_message("150.0", "U1", "the agent's own reply"),
7922 history_message("140.0", "U8", "someone else's reply"),
7923 history_message("160.0", "U1", "what Ved typed"),
7924 ],
7925 );
7926 sweep(&host, &host, "acme").await.expect("follow-up sweep");
7927
7928 let delivered = host.delivered();
7929 assert!(
7930 !delivered.iter().any(|message| message.text == "the agent's own reply"),
7931 "the agent's own reply must not echo back: {delivered:?}",
7932 );
7933 assert!(
7934 delivered.iter().any(|message| message.text == "what Ved typed"),
7935 "the user's own reply is delivered: {delivered:?}",
7936 );
7937 assert!(
7938 delivered.iter().any(|message| message.text == "someone else's reply"),
7939 "and everyone else's replies still arrive",
7940 );
7941 }
7942
7943 #[tokio::test]
7947 async fn a_thread_walk_failure_holds_the_cursor_at_its_previous_value() {
7948 let host = FakeHost::with_subscriptions(vec![sub_conversation_owned_by(
7949 "acme",
7950 "forge",
7951 Some("tester"),
7952 "C1",
7953 )]);
7954 let parent = parent_with_replies("100.0", 2);
7955 host.set_watermark("acme", "C1", "050.0");
7956 host.seed_history("C1", vec![parent.clone()]);
7957 seed_user_participation(&host, &parent);
7958 sweep(&host, &host, "acme").await.expect("trigger sweep");
7959 assert_eq!(host.thread_cursor("acme", "C1", "100.0"), Some("105.0".to_owned()));
7960
7961 host.seed_replies(
7962 "C1",
7963 "100.0",
7964 vec![
7965 parent,
7966 history_message("105.0", "U1", "the user's own reply"),
7967 history_message("300.0", "U8", "newer reply"),
7968 history_message("200.0", "U8", "older reply fails"),
7969 ],
7970 );
7971 host.fail_delivery_at(1);
7973 sweep(&host, &host, "acme").await.expect("follow-up sweep");
7974
7975 assert_eq!(
7976 host.thread_cursor("acme", "C1", "100.0"),
7977 Some("105.0".to_owned()),
7978 "the walk's cursor holds below the failed block instead of jumping past it",
7979 );
7980 }
7981
7982 #[tokio::test]
7985 async fn a_mention_reaches_every_mentions_subscription() {
7986 let host = FakeHost::with_subscriptions(vec![
7987 sub_conversation_owned_by("acme", "forge", Some("a"), "C1"),
7988 {
7989 let mut sub = sub_mentions("acme");
7990 sub.team_role = Some("a".to_owned());
7991 sub
7992 },
7993 {
7994 let mut sub = sub_mentions("acme");
7995 sub.team_role = Some("b".to_owned());
7996 sub
7997 },
7998 ]);
7999 host.set_watermark("acme", MENTION_CURSOR, "100.0");
8000 host.seed_search(vec![search_match("150.0", "C9", "ping <@U1>")]);
8001
8002 sweep_mentions(&host, &host, "acme").await.expect("sweep");
8003 let owners: Vec<Option<String>> =
8004 host.delivered_owners().iter().map(|owner| owner.team_role.clone()).collect();
8005 assert!(
8006 owners.iter().any(|role| role.as_deref() == Some("a")),
8007 "the first owner receives the mention: {owners:?}",
8008 );
8009 assert!(
8010 owners.iter().any(|role| role.as_deref() == Some("b")),
8011 "the second mention subscription is not starved: {owners:?}",
8012 );
8013 }
8014
8015 #[tokio::test]
8018 async fn an_unresolved_user_id_skips_the_conversation_sweep() {
8019 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
8020 host.set_user_id(None);
8021 host.seed_history("D1", vec![history_message("300.0", "U9", "pending")]);
8022
8023 let outcome = sweep(&host, &host, "acme").await.expect("sweep");
8024 assert_eq!(outcome.delivered, 0, "nothing is delivered half-blind");
8025 assert!(host.delivered().is_empty(), "the pending window is not swept without the id");
8026 assert_eq!(
8027 host.watermark("acme", "D1").expect("watermark"),
8028 None,
8029 "and the cursor is untouched, so the next sweep picks the window up",
8030 );
8031 }
8032
8033 #[tokio::test]
8038 async fn a_deleted_conversation_flips_the_glyph_down_through_the_pump() {
8039 let host = Arc::new(FakeHost::with_subscriptions(vec![
8043 sub_channel("acme", "C1", SlackWatchMode::All),
8044 sub_dm("acme"),
8045 ]));
8046 FakeHost::set_api_self(&host);
8047 host.set_watermark("acme", "C1", "050.0");
8048 host.set_watermark("acme", "D1", "050.0");
8049 host.seed_history("D1", vec![history_message("200.0", "U9", "a dm")]);
8050 host.fail_history_with(
8051 "C1",
8052 SlackError::Api {
8053 method: "conversations.history".to_owned(),
8054 error: "channel_not_found".to_owned(),
8055 needed: None,
8056 },
8057 );
8058
8059 let (tx, rx) = oneshot::channel();
8060 let pump = tokio::spawn(run_workspace_pump(host.clone(), "acme".to_owned(), 1, 1, rx));
8061 for _ in 0..60 {
8064 tokio::time::sleep(Duration::from_millis(50)).await;
8065 if host.connected().is_some() {
8066 break;
8067 }
8068 }
8069
8070 assert_eq!(
8071 host.connected(),
8072 Some(false),
8073 "a dead conversation must not read as a connected row after a wired tick",
8074 );
8075
8076 tx.send(()).expect("signal shutdown");
8077 tokio::time::timeout(Duration::from_secs(2), pump)
8078 .await
8079 .expect("the pump must not outlive its shutdown")
8080 .expect("no panic");
8081 }
8082
8083 #[tokio::test]
8089 async fn a_slackbot_dm_channel_not_found_leaves_the_glyph_connected() {
8090 let host = Arc::new(FakeHost::with_subscriptions(vec![sub_dm("acme")]));
8091 FakeHost::set_api_self(&host);
8092 host.seed_history("D1", vec![]);
8094 host.set_watermark("acme", "D1", "050.0");
8095 host.fail_history_with(
8096 "D1",
8097 SlackError::Api {
8098 method: "conversations.history".to_owned(),
8099 error: "channel_not_found".to_owned(),
8100 needed: None,
8101 },
8102 );
8103 host.seed_history("D2", vec![history_message("200.0", "U9", "hello from a dm")]);
8106 host.set_watermark("acme", "D2", "100.0");
8107
8108 let (tx, rx) = oneshot::channel();
8109 let pump = tokio::spawn(run_workspace_pump(host.clone(), "acme".to_owned(), 1, 1, rx));
8110 for _ in 0..60 {
8111 tokio::time::sleep(Duration::from_millis(50)).await;
8112 if host.connected().is_some() {
8113 break;
8114 }
8115 }
8116
8117 assert!(
8118 !host.delivered().is_empty(),
8119 "the tick ran and delivered the readable DM: {:?}",
8120 host.delivered(),
8121 );
8122 assert_eq!(
8123 host.connected(),
8124 Some(true),
8125 "the unreadable DM is a known quirk, not a dead connector",
8126 );
8127
8128 tx.send(()).expect("signal shutdown");
8129 tokio::time::timeout(Duration::from_secs(2), pump)
8130 .await
8131 .expect("the pump must not outlive its shutdown")
8132 .expect("no panic");
8133 }
8134
8135 #[derive(Clone, Default)]
8136 struct LogCapture(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
8137
8138 impl std::io::Write for LogCapture {
8139 fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
8140 self.0.lock().expect("lock").extend_from_slice(buf);
8141 Ok(buf.len())
8142 }
8143 fn flush(&mut self) -> std::io::Result<()> {
8144 Ok(())
8145 }
8146 }
8147
8148 impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for LogCapture {
8149 type Writer = Self;
8150 fn make_writer(&'a self) -> Self::Writer {
8151 self.clone()
8152 }
8153 }
8154
8155 #[tokio::test]
8159 async fn the_unreadable_dm_skip_logs_its_reason() {
8160 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
8161 host.seed_history("D1", vec![]);
8162 host.set_watermark("acme", "D1", "050.0");
8163 host.fail_history_with(
8164 "D1",
8165 SlackError::Api {
8166 method: "conversations.history".to_owned(),
8167 error: "channel_not_found".to_owned(),
8168 needed: None,
8169 },
8170 );
8171
8172 let capture = LogCapture::default();
8173 let subscriber = tracing_subscriber::fmt()
8174 .with_writer(capture.clone())
8175 .with_max_level(tracing::Level::INFO)
8176 .finish();
8177 let _guard = tracing::subscriber::set_default(subscriber);
8178
8179 let outcome = sweep(&host, &host, "acme").await.expect("sweep");
8180 assert!(!outcome.conversation_gone, "the quirk does not flip the glyph");
8181
8182 let log = String::from_utf8_lossy(&capture.0.lock().expect("lock")).into_owned();
8183 assert!(log.contains("D1"), "the skip names the conversation: {log}");
8184 assert!(log.contains("withholds this DM's history"), "the skip states its reason: {log}");
8185 }
8186
8187 fn throttle_lines(log: &str, message: &str, sweep: &str) -> usize {
8190 let field = format!("sweep={sweep}");
8191 log.lines().filter(|line| line.contains(message) && line.contains(&field)).count()
8192 }
8193
8194 #[tokio::test]
8198 async fn a_throttled_sweep_warns_once_per_episode_and_says_when_it_recovers() {
8199 let host = Arc::new(FakeHost::with_subscriptions(vec![
8200 sub_channel("acme", "C1", SlackWatchMode::All),
8201 sub_dm("acme"),
8202 sub_mentions("acme"),
8203 ]));
8204 FakeHost::set_api_self(&host);
8205 host.set_watermark("acme", "C1", "050.0");
8209 host.seed_history(
8210 "C1",
8211 vec![history_message("200.0", "U9", "news"), history_message("050.0", "U9", "old")],
8212 );
8213 host.set_watermark("acme", "D1", "050.0");
8214 host.seed_history(
8215 "D1",
8216 vec![history_message("200.0", "U9", "a dm"), history_message("050.0", "U9", "old")],
8217 );
8218 host.rate_limit_history_calls("C1", 2);
8219 host.rate_limit_history_calls("D1", 2);
8220 host.rate_limit_search_calls(2);
8221
8222 let capture = LogCapture::default();
8223 let subscriber = tracing_subscriber::fmt()
8224 .with_writer(capture.clone())
8225 .with_max_level(tracing::Level::DEBUG)
8226 .with_ansi(false)
8227 .finish();
8228 let _guard = tracing::subscriber::set_default(subscriber);
8229
8230 let (tx, rx) = oneshot::channel();
8231 let pump = tokio::spawn(run_workspace_pump(host.clone(), "acme".to_owned(), 1, 1, rx));
8232 for _ in 0..160 {
8233 tokio::time::sleep(Duration::from_millis(50)).await;
8234 let seen = String::from_utf8_lossy(&capture.0.lock().expect("lock")).into_owned();
8235 if throttle_lines(&seen, "slack sweep recovered from throttling", "conversation") == 1
8236 && throttle_lines(&seen, "slack sweep recovered from throttling", "dm") == 1
8237 && throttle_lines(&seen, "slack sweep recovered from throttling", "mentions") == 1
8238 {
8239 break;
8240 }
8241 }
8242 tx.send(()).expect("signal shutdown");
8243 tokio::time::timeout(Duration::from_secs(2), pump)
8244 .await
8245 .expect("the pump must not outlive its shutdown")
8246 .expect("no panic");
8247
8248 let log = String::from_utf8_lossy(&capture.0.lock().expect("lock")).into_owned();
8249 for sweep in ["conversation", "dm", "mentions"] {
8250 assert_eq!(
8251 throttle_lines(&log, "slack throttled the sweep", sweep),
8252 1,
8253 "{sweep}: one warning per throttled episode: {log}",
8254 );
8255 assert!(
8256 throttle_lines(&log, "slack sweep still throttled", sweep) >= 1,
8257 "{sweep}: the passes inside the episode are still reported, at debug: {log}",
8258 );
8259 assert_eq!(
8260 throttle_lines(&log, "slack sweep recovered from throttling", sweep),
8261 1,
8262 "{sweep}: one recovery line per episode: {log}",
8263 );
8264 }
8265 }
8266
8267 #[test]
8270 fn the_per_pass_throttle_detail_stays_below_info() {
8271 let capture = LogCapture::default();
8272 let subscriber = tracing_subscriber::fmt()
8273 .with_writer(capture.clone())
8274 .with_max_level(tracing::Level::INFO)
8275 .with_ansi(false)
8276 .finish();
8277 let _guard = tracing::subscriber::set_default(subscriber);
8278
8279 let mut episode = ThrottleEpisode::default();
8280 let throttled = Some(Duration::from_secs(7));
8281 let configured = Duration::from_secs(30);
8282 episode.note_pass("acme", "conversation", configured, throttled);
8283 episode.note_pass("acme", "conversation", configured, throttled);
8284 episode.note_pass("acme", "conversation", configured, None);
8285
8286 let log = String::from_utf8_lossy(&capture.0.lock().expect("lock")).into_owned();
8287 assert_eq!(
8288 throttle_lines(&log, "slack throttled the sweep", "conversation"),
8289 1,
8290 "the episode announces itself once: {log}",
8291 );
8292 assert_eq!(
8293 throttle_lines(&log, "slack sweep still throttled", "conversation"),
8294 0,
8295 "the repeated passes are debug: {log}",
8296 );
8297 assert_eq!(
8298 throttle_lines(&log, "slack sweep recovered from throttling", "conversation"),
8299 1,
8300 "and the end of the episode is announced: {log}",
8301 );
8302 }
8303
8304 #[tokio::test]
8309 async fn an_auto_subscribed_channel_the_user_never_joined_is_swept() {
8310 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
8311 host.set_watermark("acme", MENTION_CURSOR, "100.0");
8312 host.seed_search(vec![search_match("150.0", "C9", "ping <@U1>")]);
8313 sweep_mentions(&host, &host, "acme").await.expect("mention sweep");
8314 assert_eq!(host.auto_subscribed(), vec!["C9".to_owned()], "the mention subscribes C9");
8315
8316 host.seed_history_unlisted("C9", vec![history_message("200.0", "U8", "the follow-up")]);
8319 sweep(&host, &host, "acme").await.expect("conversation sweep");
8320
8321 let delivered = host.delivered();
8322 assert!(
8323 delivered.iter().any(|message| message.text == "the follow-up"),
8324 "the non-member channel's messages still arrive: {delivered:?}",
8325 );
8326 }
8327
8328 #[tokio::test]
8332 async fn a_dm_class_subscription_baselines_on_first_sight_instead_of_replaying() {
8333 let host = FakeHost::with_subscriptions(vec![sub_dm("acme")]);
8334 host.seed_history(
8337 "D1",
8338 vec![
8340 history_message("300.1", "U9", "old one"),
8341 history_message("200.1", "U9", "older two"),
8342 ],
8343 );
8344
8345 sweep(&host, &host, "acme").await.expect("first sweep");
8346 assert!(
8347 host.delivered().is_empty(),
8348 "the DM's existing history is baselined, not replayed: {:?}",
8349 host.delivered(),
8350 );
8351 assert_eq!(
8352 host.watermark("acme", "D1").expect("watermark"),
8353 Some("300.1".to_owned()),
8354 "the cursor starts at the newest message seen",
8355 );
8356
8357 host.seed_history(
8358 "D1",
8359 vec![
8360 history_message("400.1", "U9", "new one"),
8361 history_message("300.1", "U9", "old one"),
8362 ],
8363 );
8364 sweep(&host, &host, "acme").await.expect("second sweep");
8365 let texts: Vec<_> = host.delivered().into_iter().map(|m| m.text).collect();
8366 assert_eq!(texts, vec!["new one".to_owned()], "only what follows the baseline arrives");
8367 }
8368
8369 #[tokio::test]
8374 async fn a_sweep_backfills_a_missing_conversation_name() {
8375 let host =
8376 FakeHost::with_subscriptions(vec![sub_channel("acme", "C1", SlackWatchMode::All)]);
8377 host.conversations.lock().expect("lock").push(conversation_channel("C1", "ved-test"));
8378
8379 sweep(&host, &host, "acme").await.expect("sweep");
8380
8381 let stored = host
8382 .subscriptions("acme")
8383 .into_iter()
8384 .find_map(|sub| match sub.target {
8385 SlackSubscriptionTarget::Conversation { id, name, .. } if id == "C1" => Some(name),
8386 _ => None,
8387 })
8388 .expect("the C1 record");
8389 assert_eq!(
8390 stored.as_deref(),
8391 Some("ved-test"),
8392 "the sweep stamps the name the directory reports",
8393 );
8394 }
8395
8396 #[tokio::test]
8400 async fn a_sweep_rejects_a_name_that_is_empty_or_the_id() {
8401 let host = FakeHost::with_subscriptions(vec![
8402 sub_channel("acme", "C1", SlackWatchMode::All),
8403 sub_channel("acme", "C2", SlackWatchMode::All),
8404 ]);
8405 host.conversations
8406 .lock()
8407 .expect("lock")
8408 .extend([conversation_channel("C1", ""), conversation_channel("C2", "C2")]);
8409
8410 sweep(&host, &host, "acme").await.expect("sweep");
8411
8412 let names: Vec<Option<String>> = host
8413 .subscriptions("acme")
8414 .into_iter()
8415 .filter_map(|sub| match sub.target {
8416 SlackSubscriptionTarget::Conversation { id, name, .. }
8417 if id == "C1" || id == "C2" =>
8418 {
8419 Some(name)
8420 }
8421 _ => None,
8422 })
8423 .collect();
8424 assert_eq!(names, vec![None, None], "neither an empty name nor the id is stamped");
8425 }
8426
8427 #[tokio::test]
8431 async fn a_sweep_leaves_an_already_named_record_alone() {
8432 let mut sub = sub_channel("acme", "C1", SlackWatchMode::All);
8433 sub.target = SlackSubscriptionTarget::Conversation {
8434 id: "C1".to_owned(),
8435 name: Some("stale-name".to_owned()),
8436 mode: SlackWatchMode::All,
8437 };
8438 let host = FakeHost::with_subscriptions(vec![sub]);
8439 host.conversations.lock().expect("lock").push(conversation_channel("C1", "renamed"));
8440
8441 sweep(&host, &host, "acme").await.expect("sweep");
8442
8443 let stored = host
8444 .subscriptions("acme")
8445 .into_iter()
8446 .find_map(|sub| match sub.target {
8447 SlackSubscriptionTarget::Conversation { id, name, .. } if id == "C1" => Some(name),
8448 _ => None,
8449 })
8450 .expect("the C1 record");
8451 assert_eq!(
8452 stored.as_deref(),
8453 Some("stale-name"),
8454 "a record that already has a name is never rewritten",
8455 );
8456 }
8457
8458 #[tokio::test]
8462 async fn a_failed_mention_delivery_holds_the_cursor_and_the_follow() {
8463 let host = FakeHost::with_subscriptions(vec![sub_mentions("acme")]);
8464 host.set_watermark("acme", MENTION_CURSOR, "100.0");
8465 host.seed_search(vec![search_match("150.0", "C9", "ping <@U1>")]);
8466 host.fail_delivery_at(0);
8467
8468 let outcome = sweep_mentions(&host, &host, "acme").await.expect("sweep");
8469 assert_eq!(outcome.delivered, 0, "the failed mention is not counted");
8470 assert_eq!(
8471 host.watermark("acme", MENTION_CURSOR).expect("watermark"),
8472 Some("100.0".to_owned()),
8473 "the cursor holds at its previous value",
8474 );
8475 assert!(
8476 host.auto_subscribed().is_empty(),
8477 "a mention that did not reach a session subscribes nothing",
8478 );
8479 }
8480
8481 #[tokio::test]
8485 async fn a_conversation_fanout_failure_holds_the_cursor_for_every_owner() {
8486 let host = FakeHost::with_subscriptions(vec![
8487 sub_conversation_owned_by("acme", "forge", Some("a"), "C1"),
8488 sub_conversation_owned_by("acme", "forge", Some("b"), "C1"),
8489 ]);
8490 host.set_watermark("acme", "C1", "050.0");
8491 host.seed_history("C1", vec![history_message("300.0", "U9", "the message")]);
8492 host.fail_delivery_at(1);
8494
8495 sweep(&host, &host, "acme").await.expect("sweep");
8496 let delivered = host.delivered();
8497 assert_eq!(
8498 delivered.len(),
8499 1,
8500 "the owner whose delivery succeeded still received it: {delivered:?}",
8501 );
8502 assert_eq!(
8503 host.watermark("acme", "C1").expect("watermark"),
8504 Some("050.0".to_owned()),
8505 "the cursor holds below a message that did not reach every owner",
8506 );
8507 }
8508
8509 #[tokio::test]
8510 async fn the_pump_stops_when_the_shutdown_fires() {
8511 let host = Arc::new(FakeHost::with_subscriptions(vec![sub_dm("acme")]));
8512 FakeHost::set_api_self(&host);
8513 let (tx, rx) = oneshot::channel();
8514 let pump =
8515 tokio::spawn(run_workspace_pump(host.clone(), "acme".to_owned(), 3600, 3600, rx));
8516 tokio::time::sleep(Duration::from_millis(50)).await;
8517 tx.send(()).expect("signal shutdown");
8518 tokio::time::timeout(Duration::from_secs(2), pump)
8519 .await
8520 .expect("the pump must not outlive its shutdown")
8521 .expect("no panic");
8522 assert_eq!(host.connected(), Some(false), "the pump marks itself disconnected on exit");
8523 }
8524
8525 #[test]
8526 fn next_interval_never_sweeps_faster_than_the_configured_cadence() {
8527 assert_eq!(
8528 next_interval(Duration::from_secs(30), Some(Duration::from_secs(7))),
8529 Duration::from_secs(30),
8530 "a short Retry-After must not shorten the cadence",
8531 );
8532 assert_eq!(next_interval(Duration::from_secs(30), None), Duration::from_secs(30));
8533 assert_eq!(
8534 next_interval(Duration::from_secs(30), Some(Duration::from_secs(600))),
8535 Duration::from_secs(60),
8536 "a hostile Retry-After is capped",
8537 );
8538 assert_eq!(
8539 next_interval(Duration::from_secs(90), Some(Duration::from_secs(600))),
8540 Duration::from_secs(90),
8541 "the cap bounds the Retry-After, not the configured cadence",
8542 );
8543 }
8544
8545 #[test]
8546 fn a_token_never_reaches_the_debug_output() {
8547 let client = SlackClient::new(reqwest::Client::new(), "xoxp-supersecret".to_owned());
8548 let rendered = format!("{client:?}");
8549 assert!(!rendered.contains("supersecret"), "token leaked: {rendered}");
8550 }
8551}