lark_channel/message/
sender.rs

1//! High-level outbound message sender.
2//!
3//! `MessageSender` wraps the low-level `lark_openapi::OpenApiClient` with
4//! Channel-oriented behavior such as automatic idempotency keys and conservative
5//! retry handling.
6
7use std::future::Future;
8use std::sync::atomic::{AtomicU64, Ordering};
9use std::time::{SystemTime, UNIX_EPOCH};
10
11use crate::card::{Card, CardId};
12use crate::lark_openapi::{
13    MessageCreateOptions, MessageReplyOptions, OpenApiClient, OpenApiTransport,
14};
15use crate::{Error, MessageContent, MessageId, PostContent, Recipient, Result};
16
17const MAX_OPENAPI_UUID_CHARS: usize = 50;
18static NEXT_IDEMPOTENCY_SEQUENCE: AtomicU64 = AtomicU64::new(1);
19
20/// High-level sender for outbound text messages and replies.
21///
22/// `MessageSender` generates an idempotency key once per logical send/reply by
23/// default, or accepts a caller-provided `uuid`, and reuses that value across
24/// transport-failure retries. API and OpenAPI HTTP status errors are returned
25/// directly.
26#[derive(Debug, Clone)]
27pub struct MessageSender<T> {
28    client: OpenApiClient<T>,
29    options: MessageSenderOptions,
30}
31
32impl<T> MessageSender<T>
33where
34    T: OpenApiTransport,
35{
36    /// Creates a sender with default options.
37    pub fn new(client: OpenApiClient<T>) -> Self {
38        Self {
39            client,
40            options: MessageSenderOptions::default(),
41        }
42    }
43
44    /// Creates a sender with caller-provided options.
45    pub fn with_options(client: OpenApiClient<T>, options: MessageSenderOptions) -> Self {
46        Self { client, options }
47    }
48
49    /// Returns the wrapped low-level OpenAPI client.
50    pub fn client(&self) -> &OpenApiClient<T> {
51        &self.client
52    }
53
54    /// Returns the sender options.
55    pub fn options(&self) -> &MessageSenderOptions {
56        &self.options
57    }
58
59    /// Starts building a message send operation with caller-provided content.
60    pub fn message(&self, recipient: Recipient, content: MessageContent) -> MessageBuilder<'_, T> {
61        MessageBuilder {
62            sender: self,
63            recipient,
64            content,
65            uuid: None,
66            max_attempts: None,
67        }
68    }
69
70    /// Starts building a plain text message send operation.
71    pub fn text_message(
72        &self,
73        recipient: Recipient,
74        text: impl Into<String>,
75    ) -> MessageBuilder<'_, T> {
76        self.message(recipient, MessageContent::Text { text: text.into() })
77    }
78
79    /// Starts building a rich-text `post` message send operation.
80    pub fn post_message(&self, recipient: Recipient, post: PostContent) -> MessageBuilder<'_, T> {
81        self.message(recipient, MessageContent::Post { post })
82    }
83
84    /// Starts building a native Markdown `post` message send operation.
85    pub fn markdown_message(
86        &self,
87        recipient: Recipient,
88        markdown: impl Into<String>,
89    ) -> MessageBuilder<'_, T> {
90        self.post_message(recipient, PostContent::markdown(markdown))
91    }
92
93    /// Starts building an inline CardKit message send operation.
94    pub fn card_message(&self, recipient: Recipient, card: Card) -> MessageBuilder<'_, T> {
95        self.message(
96            recipient,
97            MessageContent::Card {
98                card: card.into_value(),
99            },
100        )
101    }
102
103    /// Starts building a message that references a pre-created CardKit entity.
104    ///
105    /// A CardKit entity can only be sent once.
106    pub fn card_reference_message(
107        &self,
108        recipient: Recipient,
109        card_id: CardId,
110    ) -> MessageBuilder<'_, T> {
111        self.message(recipient, MessageContent::CardReference { card_id })
112    }
113
114    /// Starts building a message reply operation with caller-provided content.
115    pub fn reply(
116        &self,
117        parent_message_id: MessageId,
118        content: MessageContent,
119    ) -> MessageReplyBuilder<'_, T> {
120        MessageReplyBuilder {
121            sender: self,
122            parent_message_id,
123            content,
124            uuid: None,
125            reply_in_thread: None,
126            max_attempts: None,
127        }
128    }
129
130    /// Starts building a plain text reply operation.
131    pub fn text_reply(
132        &self,
133        parent_message_id: MessageId,
134        text: impl Into<String>,
135    ) -> MessageReplyBuilder<'_, T> {
136        self.reply(
137            parent_message_id,
138            MessageContent::Text { text: text.into() },
139        )
140    }
141
142    /// Starts building a rich-text `post` reply operation.
143    pub fn post_reply(
144        &self,
145        parent_message_id: MessageId,
146        post: PostContent,
147    ) -> MessageReplyBuilder<'_, T> {
148        self.reply(parent_message_id, MessageContent::Post { post })
149    }
150
151    /// Starts building a native Markdown `post` reply operation.
152    pub fn markdown_reply(
153        &self,
154        parent_message_id: MessageId,
155        markdown: impl Into<String>,
156    ) -> MessageReplyBuilder<'_, T> {
157        self.post_reply(parent_message_id, PostContent::markdown(markdown))
158    }
159
160    /// Starts building an inline CardKit reply operation.
161    pub fn card_reply(
162        &self,
163        parent_message_id: MessageId,
164        card: Card,
165    ) -> MessageReplyBuilder<'_, T> {
166        self.reply(
167            parent_message_id,
168            MessageContent::Card {
169                card: card.into_value(),
170            },
171        )
172    }
173
174    /// Starts building a reply that references a pre-created CardKit entity.
175    ///
176    /// A CardKit entity can only be sent once.
177    pub fn card_reference_reply(
178        &self,
179        parent_message_id: MessageId,
180        card_id: CardId,
181    ) -> MessageReplyBuilder<'_, T> {
182        self.reply(parent_message_id, MessageContent::CardReference { card_id })
183    }
184
185    pub(super) async fn retry_transport_errors<R, F, Fut>(
186        &self,
187        max_attempts: usize,
188        mut operation: F,
189    ) -> Result<R>
190    where
191        F: FnMut() -> Fut,
192        Fut: Future<Output = Result<R>>,
193    {
194        let mut attempts = 0;
195
196        loop {
197            attempts += 1;
198            match operation().await {
199                Ok(value) => return Ok(value),
200                Err(error) if attempts < max_attempts && is_retryable(&error) => {}
201                Err(error) => return Err(error),
202            }
203        }
204    }
205
206    pub(super) fn max_attempts(&self, max_attempts: Option<usize>) -> usize {
207        max_attempts
208            .unwrap_or_else(|| self.options.max_attempts())
209            .max(1)
210    }
211}
212
213/// Builder for a single message send operation.
214#[derive(Debug)]
215pub struct MessageBuilder<'a, T> {
216    sender: &'a MessageSender<T>,
217    recipient: Recipient,
218    content: MessageContent,
219    uuid: Option<String>,
220    max_attempts: Option<usize>,
221}
222
223impl<T> MessageBuilder<'_, T>
224where
225    T: OpenApiTransport,
226{
227    /// Sets the caller-provided OpenAPI `uuid`.
228    ///
229    /// Use this when an upstream system already has a stable request, task, or
230    /// event identifier that should be reused across process restarts or queue
231    /// replays.
232    pub fn uuid(mut self, uuid: impl Into<String>) -> Self {
233        self.uuid = Some(uuid.into());
234        self
235    }
236
237    /// Sets the maximum number of attempts for this send operation.
238    ///
239    /// Values below `1` are clamped to `1`.
240    pub fn max_attempts(mut self, max_attempts: usize) -> Self {
241        self.max_attempts = Some(max_attempts.max(1));
242        self
243    }
244
245    /// Sends the message.
246    pub async fn send(self) -> Result<MessageId> {
247        let uuid = resolve_uuid(self.uuid)?;
248        let max_attempts = self.sender.max_attempts(self.max_attempts);
249        let recipient = self.recipient;
250        let content = self.content;
251
252        self.sender
253            .retry_transport_errors(max_attempts, || {
254                self.sender.client.create_message_with_options(
255                    recipient.clone(),
256                    content.clone(),
257                    MessageCreateOptions::with_uuid(uuid.clone()),
258                )
259            })
260            .await
261    }
262}
263
264/// Builder for a single message reply operation.
265#[derive(Debug)]
266pub struct MessageReplyBuilder<'a, T> {
267    sender: &'a MessageSender<T>,
268    parent_message_id: MessageId,
269    content: MessageContent,
270    uuid: Option<String>,
271    reply_in_thread: Option<bool>,
272    max_attempts: Option<usize>,
273}
274
275impl<T> MessageReplyBuilder<'_, T>
276where
277    T: OpenApiTransport,
278{
279    /// Sets the caller-provided OpenAPI `uuid`.
280    ///
281    /// Use this when an upstream system already has a stable request, task, or
282    /// event identifier that should be reused across process restarts or queue
283    /// replays.
284    pub fn uuid(mut self, uuid: impl Into<String>) -> Self {
285        self.uuid = Some(uuid.into());
286        self
287    }
288
289    /// Requests thread placement when the target conversation supports it.
290    pub fn reply_in_thread(mut self, reply_in_thread: bool) -> Self {
291        self.reply_in_thread = Some(reply_in_thread);
292        self
293    }
294
295    /// Sets the maximum number of attempts for this reply operation.
296    ///
297    /// Values below `1` are clamped to `1`.
298    pub fn max_attempts(mut self, max_attempts: usize) -> Self {
299        self.max_attempts = Some(max_attempts.max(1));
300        self
301    }
302
303    /// Sends the reply.
304    pub async fn send(self) -> Result<MessageId> {
305        let uuid = resolve_uuid(self.uuid)?;
306        let max_attempts = self.sender.max_attempts(self.max_attempts);
307        let parent_message_id = self.parent_message_id;
308        let content = self.content;
309        let reply_in_thread = self.reply_in_thread;
310
311        self.sender
312            .retry_transport_errors(max_attempts, || {
313                let mut reply_options = MessageReplyOptions::with_uuid(uuid.clone());
314                if let Some(reply_in_thread) = reply_in_thread {
315                    reply_options = reply_options.reply_in_thread(reply_in_thread);
316                }
317                self.sender.client.reply_message_with_options(
318                    parent_message_id.clone(),
319                    content.clone(),
320                    reply_options,
321                )
322            })
323            .await
324    }
325}
326
327/// Options for `MessageSender`.
328#[derive(Debug, Clone, PartialEq, Eq)]
329pub struct MessageSenderOptions {
330    max_attempts: usize,
331}
332
333impl Default for MessageSenderOptions {
334    fn default() -> Self {
335        Self { max_attempts: 3 }
336    }
337}
338
339impl MessageSenderOptions {
340    /// Creates options with the default retry attempt count.
341    pub fn new() -> Self {
342        Self::default()
343    }
344
345    /// Creates options with a custom maximum attempt count.
346    ///
347    /// Values below `1` are clamped to `1`.
348    pub fn with_max_attempts(max_attempts: usize) -> Self {
349        let mut options = Self::new();
350        options.set_max_attempts(max_attempts);
351        options
352    }
353
354    /// Sets the maximum number of attempts for one logical send or reply.
355    ///
356    /// Values below `1` are clamped to `1`.
357    pub fn set_max_attempts(&mut self, max_attempts: usize) -> &mut Self {
358        self.max_attempts = max_attempts.max(1);
359        self
360    }
361
362    /// Returns the configured maximum attempt count.
363    pub fn max_attempts(&self) -> usize {
364        self.max_attempts.max(1)
365    }
366}
367
368fn is_retryable(error: &Error) -> bool {
369    matches!(error, Error::Transport(_))
370}
371
372pub(super) fn resolve_uuid(uuid: Option<String>) -> Result<String> {
373    match uuid {
374        Some(uuid) => validate_uuid(uuid),
375        None => Ok(generate_idempotency_key()),
376    }
377}
378
379fn validate_uuid(uuid: String) -> Result<String> {
380    if uuid.is_empty() {
381        return Err(Error::Validation(
382            "message uuid must not be empty".to_owned(),
383        ));
384    }
385    if uuid.chars().count() > MAX_OPENAPI_UUID_CHARS {
386        return Err(Error::Validation(format!(
387            "message uuid must be at most {MAX_OPENAPI_UUID_CHARS} characters"
388        )));
389    }
390    Ok(uuid)
391}
392
393pub(super) fn generate_idempotency_key() -> String {
394    let pid = std::process::id();
395    let sequence = NEXT_IDEMPOTENCY_SEQUENCE.fetch_add(1, Ordering::Relaxed);
396    let nanos = SystemTime::now()
397        .duration_since(UNIX_EPOCH)
398        .map(|duration| duration.as_nanos())
399        .unwrap_or_default();
400
401    format!("lc-{pid:x}-{nanos:x}-{sequence:x}")
402}
403
404#[cfg(test)]
405mod tests {
406    use std::collections::VecDeque;
407    use std::future::Future;
408    use std::sync::{Arc, Mutex, MutexGuard};
409    use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
410
411    use serde_json::{Value, json};
412    use url::Url;
413
414    use super::*;
415    use crate::ChannelConfig;
416    use crate::lark_openapi::{BoxFuture, HttpRequest, HttpResponse};
417
418    #[test]
419    fn send_text_generates_uuid_and_reuses_it_across_transport_retry() {
420        let transport = FakeTransport::new(vec![
421            FakeResponse::http(
422                200,
423                json!({
424                    "code": 0,
425                    "msg": "ok",
426                    "tenant_access_token": "tenant-token-1",
427                    "expire": 7200
428                }),
429            ),
430            FakeResponse::transport_error("temporary network error"),
431            FakeResponse::http(
432                200,
433                json!({
434                    "code": 0,
435                    "msg": "ok",
436                    "data": {
437                        "message_id": "om_123"
438                    }
439                }),
440            ),
441        ]);
442        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
443        let sender = MessageSender::new(client);
444
445        let message_id = block_on(
446            sender
447                .text_message(Recipient::Chat("oc_123".to_owned()), "hello")
448                .send(),
449        )
450        .expect("sent message after retry");
451
452        assert_eq!(message_id, MessageId("om_123".to_owned()));
453
454        let calls = transport.calls();
455        assert_eq!(calls.len(), 3);
456        assert_eq!(
457            calls[1].url.as_str(),
458            "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id"
459        );
460        assert_eq!(calls[1].body["uuid"], calls[2].body["uuid"]);
461        assert!(
462            calls[1].body["uuid"]
463                .as_str()
464                .is_some_and(|uuid| { uuid.starts_with("lc-") && uuid.len() <= 50 })
465        );
466    }
467
468    #[test]
469    fn text_message_reuses_caller_uuid_across_transport_retry() {
470        let transport = FakeTransport::new(vec![
471            FakeResponse::http(
472                200,
473                json!({
474                    "code": 0,
475                    "msg": "ok",
476                    "tenant_access_token": "tenant-token-1",
477                    "expire": 7200
478                }),
479            ),
480            FakeResponse::transport_error("temporary network error"),
481            FakeResponse::http(
482                200,
483                json!({
484                    "code": 0,
485                    "msg": "ok",
486                    "data": {
487                        "message_id": "om_123"
488                    }
489                }),
490            ),
491        ]);
492        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
493        let sender = MessageSender::new(client);
494
495        let message_id = block_on(
496            sender
497                .text_message(Recipient::Chat("oc_123".to_owned()), "hello")
498                .uuid("upstream-task-123")
499                .send(),
500        )
501        .expect("sent message after retry");
502
503        assert_eq!(message_id, MessageId("om_123".to_owned()));
504
505        let calls = transport.calls();
506        assert_eq!(calls.len(), 3);
507        assert_eq!(calls[1].body["uuid"], "upstream-task-123");
508        assert_eq!(calls[2].body["uuid"], "upstream-task-123");
509    }
510
511    #[test]
512    fn message_forwards_custom_content() {
513        let transport = FakeTransport::new(vec![
514            FakeResponse::http(
515                200,
516                json!({
517                    "code": 0,
518                    "msg": "ok",
519                    "tenant_access_token": "tenant-token-1",
520                    "expire": 7200
521                }),
522            ),
523            FakeResponse::http(
524                200,
525                json!({
526                    "code": 0,
527                    "msg": "ok",
528                    "data": {
529                        "message_id": "om_custom"
530                    }
531                }),
532            ),
533        ]);
534        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
535        let sender = MessageSender::new(client);
536
537        let message_id = block_on(
538            sender
539                .message(
540                    Recipient::Chat("oc_123".to_owned()),
541                    MessageContent::Custom {
542                        msg_type: "custom".to_owned(),
543                        content: json!({ "body": "hello" }),
544                    },
545                )
546                .uuid("custom-uuid")
547                .send(),
548        )
549        .expect("sent custom message");
550
551        assert_eq!(message_id, MessageId("om_custom".to_owned()));
552
553        let calls = transport.calls();
554        assert_eq!(calls.len(), 2);
555        assert_eq!(calls[1].body["msg_type"], "custom");
556        assert_eq!(calls[1].body["content"], r#"{"body":"hello"}"#);
557        assert_eq!(calls[1].body["uuid"], "custom-uuid");
558    }
559
560    #[test]
561    fn markdown_message_uses_post_content_and_managed_uuid() {
562        let transport = FakeTransport::new(vec![
563            FakeResponse::http(
564                200,
565                json!({
566                    "code": 0,
567                    "msg": "ok",
568                    "tenant_access_token": "tenant-token-1",
569                    "expire": 7200
570                }),
571            ),
572            FakeResponse::http(
573                200,
574                json!({
575                    "code": 0,
576                    "msg": "ok",
577                    "data": {
578                        "message_id": "om_markdown"
579                    }
580                }),
581            ),
582        ]);
583        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
584        let sender = MessageSender::new(client);
585
586        let message_id = block_on(
587            sender
588                .markdown_message(
589                    Recipient::Chat("oc_123".to_owned()),
590                    "**hello** [docs](https://open.feishu.cn)",
591                )
592                .send(),
593        )
594        .expect("sent markdown message");
595
596        assert_eq!(message_id, MessageId("om_markdown".to_owned()));
597        let body = &transport.calls()[1].body;
598        assert_eq!(body["msg_type"], "post");
599        assert_eq!(
600            body["content"],
601            "{\"zh_cn\":{\"content\":[[{\"tag\":\"md\",\"text\":\"**hello** [docs](https://open.feishu.cn)\"}]]}}"
602        );
603        assert!(
604            body["uuid"]
605                .as_str()
606                .is_some_and(|uuid| uuid.starts_with("lc-"))
607        );
608    }
609
610    #[test]
611    fn markdown_reply_uses_post_content_and_managed_uuid() {
612        let transport = FakeTransport::new(vec![
613            FakeResponse::http(
614                200,
615                json!({
616                    "code": 0,
617                    "msg": "ok",
618                    "tenant_access_token": "tenant-token-1",
619                    "expire": 7200
620                }),
621            ),
622            FakeResponse::http(
623                200,
624                json!({
625                    "code": 0,
626                    "msg": "ok",
627                    "data": {
628                        "message_id": "om_markdown_reply"
629                    }
630                }),
631            ),
632        ]);
633        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
634        let sender = MessageSender::new(client);
635
636        let message_id = block_on(
637            sender
638                .markdown_reply(MessageId("om_parent".to_owned()), "**reply**")
639                .send(),
640        )
641        .expect("replied with markdown");
642
643        assert_eq!(message_id, MessageId("om_markdown_reply".to_owned()));
644        let calls = transport.calls();
645        assert_eq!(
646            calls[1].url.as_str(),
647            "https://open.feishu.cn/open-apis/im/v1/messages/om_parent/reply"
648        );
649        assert_eq!(calls[1].body["msg_type"], "post");
650        assert_eq!(
651            calls[1].body["content"],
652            "{\"zh_cn\":{\"content\":[[{\"tag\":\"md\",\"text\":\"**reply**\"}]]}}"
653        );
654        assert!(
655            calls[1].body["uuid"]
656                .as_str()
657                .is_some_and(|uuid| uuid.starts_with("lc-"))
658        );
659    }
660
661    #[test]
662    fn card_message_uses_validated_interactive_content() {
663        let transport = FakeTransport::new(vec![
664            FakeResponse::http(
665                200,
666                json!({
667                    "code": 0,
668                    "msg": "ok",
669                    "tenant_access_token": "tenant-token-1",
670                    "expire": 7200
671                }),
672            ),
673            FakeResponse::http(
674                200,
675                json!({
676                    "code": 0,
677                    "msg": "ok",
678                    "data": { "message_id": "om_card" }
679                }),
680            ),
681        ]);
682        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
683        let sender = MessageSender::new(client);
684        let card = Card::builder().markdown("**hello**").build().expect("card");
685
686        let message_id = block_on(
687            sender
688                .card_message(Recipient::Chat("oc_123".to_owned()), card)
689                .send(),
690        )
691        .expect("sent card");
692
693        assert_eq!(message_id, MessageId("om_card".to_owned()));
694        let body = &transport.calls()[1].body;
695        assert_eq!(body["msg_type"], "interactive");
696        let content: Value =
697            serde_json::from_str(body["content"].as_str().expect("content string"))
698                .expect("card json");
699        assert_eq!(content["schema"], "2.0");
700        assert_eq!(content["config"]["update_multi"], true);
701    }
702
703    #[test]
704    fn card_reference_reply_uses_card_entity_content_and_thread_option() {
705        let transport = FakeTransport::new(vec![
706            FakeResponse::http(
707                200,
708                json!({
709                    "code": 0,
710                    "msg": "ok",
711                    "tenant_access_token": "tenant-token-1",
712                    "expire": 7200
713                }),
714            ),
715            FakeResponse::http(
716                200,
717                json!({
718                    "code": 0,
719                    "msg": "ok",
720                    "data": { "message_id": "om_card_reply" }
721                }),
722            ),
723        ]);
724        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
725        let sender = MessageSender::new(client);
726
727        let message_id = block_on(
728            sender
729                .card_reference_reply(
730                    MessageId("om_parent".to_owned()),
731                    CardId::new("7355372766134157313").expect("card id"),
732                )
733                .reply_in_thread(true)
734                .send(),
735        )
736        .expect("replied with card reference");
737
738        assert_eq!(message_id, MessageId("om_card_reply".to_owned()));
739        let body = &transport.calls()[1].body;
740        assert_eq!(body["msg_type"], "interactive");
741        assert_eq!(body["reply_in_thread"], true);
742        let content: Value =
743            serde_json::from_str(body["content"].as_str().expect("content string"))
744                .expect("card reference json");
745        assert_eq!(content["type"], "card");
746        assert_eq!(content["data"]["card_id"], "7355372766134157313");
747    }
748
749    #[test]
750    fn text_message_rejects_empty_uuid() {
751        let transport = FakeTransport::new(vec![]);
752        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
753        let sender = MessageSender::new(client);
754
755        let error = block_on(
756            sender
757                .text_message(Recipient::Chat("oc_123".to_owned()), "hello")
758                .uuid("")
759                .send(),
760        )
761        .expect_err("validation error");
762
763        assert!(matches!(
764            error,
765            Error::Validation(message) if message == "message uuid must not be empty"
766        ));
767        assert!(transport.calls().is_empty());
768    }
769
770    #[test]
771    fn text_message_rejects_uuid_longer_than_fifty_characters() {
772        let transport = FakeTransport::new(vec![]);
773        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
774        let sender = MessageSender::new(client);
775
776        let error = block_on(
777            sender
778                .text_message(Recipient::Chat("oc_123".to_owned()), "hello")
779                .uuid("x".repeat(51))
780                .send(),
781        )
782        .expect_err("validation error");
783
784        assert!(matches!(
785            error,
786            Error::Validation(message) if message == "message uuid must be at most 50 characters"
787        ));
788        assert!(transport.calls().is_empty());
789    }
790
791    #[test]
792    fn send_text_does_not_retry_api_errors() {
793        let transport = FakeTransport::new(vec![
794            FakeResponse::http(
795                200,
796                json!({
797                    "code": 0,
798                    "msg": "ok",
799                    "tenant_access_token": "tenant-token-1",
800                    "expire": 7200
801                }),
802            ),
803            FakeResponse::http(
804                200,
805                json!({
806                    "code": 99991672,
807                    "msg": "missing permission"
808                }),
809            ),
810        ]);
811        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
812        let sender = MessageSender::new(client);
813
814        let error = block_on(
815            sender
816                .text_message(Recipient::Chat("oc_123".to_owned()), "hello")
817                .send(),
818        )
819        .expect_err("api error");
820
821        assert!(matches!(
822            error,
823            Error::Api {
824                code: 99991672,
825                message
826            } if message == "missing permission"
827        ));
828        assert_eq!(transport.calls().len(), 2);
829    }
830
831    #[test]
832    fn send_text_respects_max_attempts() {
833        let transport = FakeTransport::new(vec![
834            FakeResponse::http(
835                200,
836                json!({
837                    "code": 0,
838                    "msg": "ok",
839                    "tenant_access_token": "tenant-token-1",
840                    "expire": 7200
841                }),
842            ),
843            FakeResponse::transport_error("temporary network error"),
844            FakeResponse::http(
845                200,
846                json!({
847                    "code": 0,
848                    "msg": "ok",
849                    "data": {
850                        "message_id": "om_123"
851                    }
852                }),
853            ),
854        ]);
855        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
856        let sender =
857            MessageSender::with_options(client, MessageSenderOptions::with_max_attempts(1));
858
859        let error = block_on(
860            sender
861                .text_message(Recipient::Chat("oc_123".to_owned()), "hello")
862                .send(),
863        )
864        .expect_err("transport error");
865
866        assert!(matches!(
867            error,
868            Error::Transport(message) if message == "temporary network error"
869        ));
870        assert_eq!(transport.calls().len(), 2);
871    }
872
873    #[test]
874    fn send_text_does_not_retry_http_status_errors() {
875        let transport = FakeTransport::new(vec![
876            FakeResponse::http(
877                200,
878                json!({
879                    "code": 0,
880                    "msg": "ok",
881                    "tenant_access_token": "tenant-token-1",
882                    "expire": 7200
883                }),
884            ),
885            FakeResponse::http(
886                500,
887                json!({
888                    "code": 0,
889                    "msg": "ok"
890                }),
891            ),
892        ]);
893        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
894        let sender = MessageSender::new(client);
895
896        let error = block_on(
897            sender
898                .text_message(Recipient::Chat("oc_123".to_owned()), "hello")
899                .send(),
900        )
901        .expect_err("http status error");
902
903        assert!(matches!(error, Error::HttpStatus { status: 500 }));
904        assert_eq!(transport.calls().len(), 2);
905    }
906
907    #[test]
908    fn reply_text_reuses_uuid_across_transport_retry() {
909        let transport = FakeTransport::new(vec![
910            FakeResponse::http(
911                200,
912                json!({
913                    "code": 0,
914                    "msg": "ok",
915                    "tenant_access_token": "tenant-token-1",
916                    "expire": 7200
917                }),
918            ),
919            FakeResponse::transport_error("temporary network error"),
920            FakeResponse::http(
921                200,
922                json!({
923                    "code": 0,
924                    "msg": "ok",
925                    "data": {
926                        "message_id": "om_reply"
927                    }
928                }),
929            ),
930        ]);
931        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
932        let sender = MessageSender::new(client);
933
934        let message_id = block_on(
935            sender
936                .text_reply(MessageId("om_parent".to_owned()), "reply")
937                .send(),
938        )
939        .expect("replied after retry");
940
941        assert_eq!(message_id, MessageId("om_reply".to_owned()));
942
943        let calls = transport.calls();
944        assert_eq!(calls.len(), 3);
945        assert_eq!(
946            calls[1].url.as_str(),
947            "https://open.feishu.cn/open-apis/im/v1/messages/om_parent/reply"
948        );
949        assert_eq!(calls[1].body["uuid"], calls[2].body["uuid"]);
950    }
951
952    #[test]
953    fn text_reply_reuses_caller_uuid_across_transport_retry() {
954        let transport = FakeTransport::new(vec![
955            FakeResponse::http(
956                200,
957                json!({
958                    "code": 0,
959                    "msg": "ok",
960                    "tenant_access_token": "tenant-token-1",
961                    "expire": 7200
962                }),
963            ),
964            FakeResponse::transport_error("temporary network error"),
965            FakeResponse::http(
966                200,
967                json!({
968                    "code": 0,
969                    "msg": "ok",
970                    "data": {
971                        "message_id": "om_reply"
972                    }
973                }),
974            ),
975        ]);
976        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
977        let sender = MessageSender::new(client);
978
979        let message_id = block_on(
980            sender
981                .text_reply(MessageId("om_parent".to_owned()), "reply")
982                .uuid("upstream-reply-123")
983                .send(),
984        )
985        .expect("replied after retry");
986
987        assert_eq!(message_id, MessageId("om_reply".to_owned()));
988
989        let calls = transport.calls();
990        assert_eq!(calls.len(), 3);
991        assert_eq!(calls[1].body["uuid"], "upstream-reply-123");
992        assert_eq!(calls[2].body["uuid"], "upstream-reply-123");
993    }
994
995    #[test]
996    fn text_reply_forwards_thread_flag() {
997        let transport = FakeTransport::new(vec![
998            FakeResponse::http(
999                200,
1000                json!({
1001                    "code": 0,
1002                    "msg": "ok",
1003                    "tenant_access_token": "tenant-token-1",
1004                    "expire": 7200
1005                }),
1006            ),
1007            FakeResponse::http(
1008                200,
1009                json!({
1010                    "code": 0,
1011                    "msg": "ok",
1012                    "data": {
1013                        "message_id": "om_reply"
1014                    }
1015                }),
1016            ),
1017        ]);
1018        let client = OpenApiClient::new(ChannelConfig::new("cli_a", "secret"), transport.clone());
1019        let sender = MessageSender::new(client);
1020
1021        let message_id = block_on(
1022            sender
1023                .text_reply(MessageId("om_parent".to_owned()), "reply")
1024                .uuid("reply-uuid")
1025                .reply_in_thread(true)
1026                .send(),
1027        )
1028        .expect("replied");
1029
1030        assert_eq!(message_id, MessageId("om_reply".to_owned()));
1031
1032        let calls = transport.calls();
1033        assert_eq!(calls.len(), 2);
1034        assert_eq!(calls[1].body["uuid"], "reply-uuid");
1035        assert_eq!(calls[1].body["reply_in_thread"], true);
1036    }
1037
1038    #[test]
1039    fn max_attempts_is_clamped_to_one() {
1040        let options = MessageSenderOptions::with_max_attempts(0);
1041
1042        assert_eq!(options.max_attempts(), 1);
1043    }
1044
1045    #[derive(Clone, Debug)]
1046    struct FakeTransport {
1047        state: Arc<Mutex<FakeState>>,
1048    }
1049
1050    impl FakeTransport {
1051        fn new(responses: Vec<FakeResponse>) -> Self {
1052            Self {
1053                state: Arc::new(Mutex::new(FakeState {
1054                    responses: responses.into(),
1055                    calls: Vec::new(),
1056                })),
1057            }
1058        }
1059
1060        fn calls(&self) -> Vec<FakeCall> {
1061            self.state().calls.clone()
1062        }
1063
1064        fn state(&self) -> MutexGuard<'_, FakeState> {
1065            self.state.lock().expect("fake transport state poisoned")
1066        }
1067    }
1068
1069    impl OpenApiTransport for FakeTransport {
1070        fn send_json(&self, request: HttpRequest) -> BoxFuture<'static, Result<HttpResponse>> {
1071            let response = {
1072                let mut state = self.state();
1073                state.calls.push(FakeCall {
1074                    url: request.url,
1075                    body: request.body,
1076                });
1077                state.responses.pop_front().expect("fake response")
1078            };
1079
1080            Box::pin(async move { response.into_result() })
1081        }
1082    }
1083
1084    #[derive(Debug)]
1085    struct FakeState {
1086        responses: VecDeque<FakeResponse>,
1087        calls: Vec<FakeCall>,
1088    }
1089
1090    #[derive(Clone, Debug)]
1091    struct FakeCall {
1092        url: Url,
1093        body: Value,
1094    }
1095
1096    #[derive(Debug)]
1097    enum FakeResponse {
1098        Http(HttpResponse),
1099        TransportError(String),
1100    }
1101
1102    impl FakeResponse {
1103        fn http(status: u16, body: Value) -> Self {
1104            Self::Http(HttpResponse::json(status, body))
1105        }
1106
1107        fn transport_error(message: impl Into<String>) -> Self {
1108            Self::TransportError(message.into())
1109        }
1110
1111        fn into_result(self) -> Result<HttpResponse> {
1112            match self {
1113                Self::Http(response) => Ok(response),
1114                Self::TransportError(message) => Err(Error::Transport(message)),
1115            }
1116        }
1117    }
1118
1119    impl From<HttpResponse> for FakeResponse {
1120        fn from(response: HttpResponse) -> Self {
1121            Self::Http(response)
1122        }
1123    }
1124
1125    fn block_on<F>(future: F) -> F::Output
1126    where
1127        F: Future,
1128    {
1129        let waker = noop_waker();
1130        let mut context = Context::from_waker(&waker);
1131        let mut future = Box::pin(future);
1132
1133        match future.as_mut().poll(&mut context) {
1134            Poll::Ready(output) => output,
1135            Poll::Pending => panic!("test future unexpectedly pending"),
1136        }
1137    }
1138
1139    fn noop_waker() -> Waker {
1140        unsafe { Waker::from_raw(noop_raw_waker()) }
1141    }
1142
1143    fn noop_raw_waker() -> RawWaker {
1144        fn clone(_: *const ()) -> RawWaker {
1145            noop_raw_waker()
1146        }
1147
1148        fn wake(_: *const ()) {}
1149        fn wake_by_ref(_: *const ()) {}
1150        fn drop(_: *const ()) {}
1151
1152        RawWaker::new(
1153            std::ptr::null(),
1154            &RawWakerVTable::new(clone, wake, wake_by_ref, drop),
1155        )
1156    }
1157}
lark_channel/message/sender.rs

lark_channel/message/
sender.rs
