[CHA-5933] feat: add scheduled messages

docs/messages/scheduled_messages.md

@@ -0,0 +1,89 @@
+Scheduled messages let a user write a message now and have it sent to a channel later. Until it is sent, a scheduled message is visible only to the user who scheduled it and to server-side callers. Server-side, you can schedule, update, cancel, send and list the scheduled messages of any user.
+
+Scheduled messages must be enabled for your app. Requests return an error with the `scheduled_messages_disabled` reason otherwise.
+
+## Limits
+
+- The send time must be at least 1 minute and at most the app's maximum lead time (120 days by default) from now
+- A user can have at most 100 scheduled messages, and at most 10 in one channel (app defaults)
+- Slash commands cannot be scheduled
+
+## Scheduling a Message
+
+Pass `scheduled_at` to `send_message`. It accepts a `datetime` (a naive one is taken to be in UTC) or an ISO 8601 string. The message is stored instead of sent, and the response carries the scheduled message.
+
+```python
+from datetime import datetime, timedelta, timezone
+
+response = channel.send_message(
+    {"text": "See you tomorrow"},
+    "user-id",
+    scheduled_at=datetime.now(timezone.utc) + timedelta(days=1),
+)
+
+scheduled = response["scheduled_message"]
+print(scheduled["id"], scheduled["status"], scheduled["scheduled_at"])
+```
+
+`response["message"]` holds the content as it will be sent; it is not in the channel yet, and its ID is the scheduled message ID. The ID of the sent message is only known when it is sent (see the `scheduled_message.sent` event below).
+
+## Updating a Scheduled Message
+
+Change the message, the send time, or both. A new message replaces the scheduled one as a whole.
+
+```python
+client.update_scheduled_message(
+    scheduled["id"],
+    message={"text": "See you the day after"},
+    scheduled_at=datetime.now(timezone.utc) + timedelta(days=2),
+)
+
+# Only move the send time
+client.update_scheduled_message(
+    scheduled["id"], scheduled_at=datetime.now(timezone.utc) + timedelta(days=2)
+)
+```
+
+## Sending a Scheduled Message Now
+
+```python
+client.send_scheduled_message_now(scheduled["id"])
+```
+
+## Deleting a Scheduled Message
+
+Cancels a scheduled message so it is never sent, or removes one that failed to send.
+
+```python
+client.delete_scheduled_message(scheduled["id"])
+```
+
+Update, send now and delete return an error while the message is being sent. A message that failed to send can only be deleted.
+
+## Querying Scheduled Messages
+
+Lists the pending (`scheduled`, `sending`) and `failed` messages of one user across channels: pending ones first, then failed ones, each by send time.
+
+```python
+response = client.query_scheduled_messages(
+    "user-id",
+    filter={"cid": "messaging:general", "status": ["failed"]},
+    options={"limit": 20},
+)
+
+# Next page
+if response.get("next"):
+    response = client.query_scheduled_messages(
+        "user-id", options={"next": response["next"]}
+    )
+```
+
+## Events
+
+| Event | Webhook | Description |
+| --- | --- | --- |
+| `scheduled_message.created` | No | A message was scheduled |
+| `scheduled_message.updated` | No | A scheduled message was updated or sent now |
+| `scheduled_message.deleted` | No | A scheduled message was canceled |
+| `scheduled_message.sent` | Yes | A scheduled message was sent. `message_id` holds the ID of the sent message |
+| `scheduled_message.failed` | Yes | A scheduled message failed to send. `scheduled_message.failure_reason` says why |

stream_chat/async_chat/channel.py

@@ -4,13 +4,16 @@
 
 from stream_chat.base.channel import ChannelInterface, add_user_id
 from stream_chat.base.exceptions import StreamChannelException
+from stream_chat.types.scheduled_messages import format_scheduled_at
 from stream_chat.types.stream_response import StreamResponse
 
 
 class Channel(ChannelInterface):
     async def send_message(
         self, message: Dict, user_id: str, **options: Any
     ) -> StreamResponse:
+        if options.get("scheduled_at") is not None:
+            options["scheduled_at"] = format_scheduled_at(options["scheduled_at"])
         payload = {"message": add_user_id(message, user_id), **options}
         return await self.client.post(f"{self.url}/message", data=payload)
 

stream_chat/async_chat/client.py

@@ -28,6 +28,11 @@
 from stream_chat.types.campaign import CampaignData, QueryCampaignsOptions
 from stream_chat.types.channel_batch import ChannelsBatchOptions
 from stream_chat.types.draft import QueryDraftsFilter, QueryDraftsOptions
+from stream_chat.types.scheduled_messages import (
+    QueryScheduledMessagesFilter,
+    QueryScheduledMessagesOptions,
+    format_scheduled_at,
+)
 from stream_chat.types.segment import (
     QuerySegmentsOptions,
     QuerySegmentTargetsOptions,
@@ -992,6 +997,47 @@ async def query_reminders(
         params["user_id"] = user_id
         return await self.post("reminders/query", data=params)
 
+    async def update_scheduled_message(
+        self,
+        scheduled_message_id: str,
+        message: Optional[Dict] = None,
+        scheduled_at: Optional[Union[str, datetime.datetime]] = None,
+    ) -> StreamResponse:
+        if message is None and scheduled_at is None:
+            raise ValueError("message or scheduled_at is required")
+        data: Dict[str, Any] = {}
+        if message is not None:
+            data["message"] = message
+        if scheduled_at is not None:
+            data["scheduled_at"] = format_scheduled_at(scheduled_at)
+        return await self.patch(f"scheduled_messages/{scheduled_message_id}", data=data)
+
+    async def delete_scheduled_message(
+        self, scheduled_message_id: str
+    ) -> StreamResponse:
+        return await self.delete(f"scheduled_messages/{scheduled_message_id}")
+
+    async def send_scheduled_message_now(
+        self, scheduled_message_id: str
+    ) -> StreamResponse:
+        return await self.post(f"scheduled_messages/{scheduled_message_id}/send_now")
+
+    async def query_scheduled_messages(
+        self,
+        user_id: str,
+        filter: Optional[QueryScheduledMessagesFilter] = None,
+        sort: Optional[List[SortParam]] = None,
+        options: Optional[QueryScheduledMessagesOptions] = None,
+    ) -> StreamResponse:
+        data: Dict[str, Any] = {"user_id": user_id}
+        if filter is not None:
+            data["filter"] = cast(dict, filter)
+        if sort is not None:
+            data["sort"] = sort
+        if options is not None:
+            data.update(cast(dict, options))
+        return await self.post("scheduled_messages/query", data=data)
+
     async def get_user_locations(self, user_id: str, **options: Any) -> StreamResponse:
         params = {"user_id": user_id, **options}
         return await self.get("users/live_locations", params=params)

stream_chat/base/channel.py

@@ -40,6 +40,10 @@ def send_message(
 
         :param message: the Message object
         :param user_id: the ID of the user that created the message
+        :param options: additional request fields, e.g. skip_push. Pass
+            scheduled_at (a datetime, naive taken as UTC, or an ISO 8601 string)
+            to store the message and send it at that time instead of now; the
+            response then carries scheduled_message
         :return: the Server Response
         """
         pass

stream_chat/base/client.py

@@ -10,6 +10,10 @@
 from stream_chat.types.base import SortParam
 from stream_chat.types.campaign import CampaignData, QueryCampaignsOptions
 from stream_chat.types.draft import QueryDraftsFilter, QueryDraftsOptions
+from stream_chat.types.scheduled_messages import (
+    QueryScheduledMessagesFilter,
+    QueryScheduledMessagesOptions,
+)
 from stream_chat.types.segment import (
     QuerySegmentsOptions,
     QuerySegmentTargetsOptions,
@@ -1589,6 +1593,69 @@ def query_reminders(
         """
         pass
 
+    @abc.abstractmethod
+    def update_scheduled_message(
+        self,
+        scheduled_message_id: str,
+        message: Optional[Dict] = None,
+        scheduled_at: Optional[Union[str, datetime.datetime]] = None,
+    ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        """
+        Replaces the message and/or the send time of a scheduled message.
+        At least one of message and scheduled_at must be set.
+
+        :param scheduled_message_id: The ID of the scheduled message
+        :param message: Replaces the scheduled message as a whole. The sender stays
+            the user who scheduled it
+        :param scheduled_at: The new send time. A naive datetime is taken to be in UTC
+        :return: API response with the scheduled message
+        """
+        pass
+
+    @abc.abstractmethod
+    def delete_scheduled_message(
+        self, scheduled_message_id: str
+    ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        """
+        Cancels a scheduled message so it is never sent, or removes a failed one.
+
+        :param scheduled_message_id: The ID of the scheduled message
+        :return: API response with the scheduled message
+        """
+        pass
+
+    @abc.abstractmethod
+    def send_scheduled_message_now(
+        self, scheduled_message_id: str
+    ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        """
+        Moves the send time of a scheduled message to now and queues the send.
+
+        :param scheduled_message_id: The ID of the scheduled message
+        :return: API response with the scheduled message
+        """
+        pass
+
+    @abc.abstractmethod
+    def query_scheduled_messages(
+        self,
+        user_id: str,
+        filter: Optional[QueryScheduledMessagesFilter] = None,
+        sort: Optional[List[SortParam]] = None,
+        options: Optional[QueryScheduledMessagesOptions] = None,
+    ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        """
+        Lists the scheduled and failed messages of one user across channels:
+        pending ones first, then failed ones, each by scheduled_at.
+
+        :param user_id: The ID of the user whose scheduled messages to list
+        :param filter: Narrows the result by channel (cid) and status
+        :param sort: Only scheduled_at ascending is supported
+        :param options: Pagination: limit (1 to 100, default 20) and next
+        :return: API response with scheduled_messages and next
+        """
+        pass
+
     @abc.abstractmethod
     def get_user_locations(
         self, user_id: str, **options: Any

stream_chat/channel.py

@@ -4,13 +4,16 @@
 
 from stream_chat.base.channel import ChannelInterface, add_user_id
 from stream_chat.base.exceptions import StreamChannelException
+from stream_chat.types.scheduled_messages import format_scheduled_at
 from stream_chat.types.stream_response import StreamResponse
 
 
 class Channel(ChannelInterface):
     def send_message(
         self, message: Dict, user_id: str, **options: Any
     ) -> StreamResponse:
+        if options.get("scheduled_at") is not None:
+            options["scheduled_at"] = format_scheduled_at(options["scheduled_at"])
         payload = {"message": add_user_id(message, user_id), **options}
         return self.client.post(f"{self.url}/message", data=payload)
 

stream_chat/client.py

@@ -25,6 +25,11 @@
 from stream_chat.types.campaign import CampaignData, QueryCampaignsOptions
 from stream_chat.types.channel_batch import ChannelsBatchOptions
 from stream_chat.types.draft import QueryDraftsFilter, QueryDraftsOptions
+from stream_chat.types.scheduled_messages import (
+    QueryScheduledMessagesFilter,
+    QueryScheduledMessagesOptions,
+    format_scheduled_at,
+)
 from stream_chat.types.segment import (
     QuerySegmentsOptions,
     QuerySegmentTargetsOptions,
@@ -954,6 +959,43 @@ def query_reminders(
         params["user_id"] = user_id
         return self.post("reminders/query", data=params)
 
+    def update_scheduled_message(
+        self,
+        scheduled_message_id: str,
+        message: Optional[Dict] = None,
+        scheduled_at: Optional[Union[str, datetime.datetime]] = None,
+    ) -> StreamResponse:
+        if message is None and scheduled_at is None:
+            raise ValueError("message or scheduled_at is required")
+        data: Dict[str, Any] = {}
+        if message is not None:
+            data["message"] = message
+        if scheduled_at is not None:
+            data["scheduled_at"] = format_scheduled_at(scheduled_at)
+        return self.patch(f"scheduled_messages/{scheduled_message_id}", data=data)
+
+    def delete_scheduled_message(self, scheduled_message_id: str) -> StreamResponse:
+        return self.delete(f"scheduled_messages/{scheduled_message_id}")
+
+    def send_scheduled_message_now(self, scheduled_message_id: str) -> StreamResponse:
+        return self.post(f"scheduled_messages/{scheduled_message_id}/send_now")
+
+    def query_scheduled_messages(
+        self,
+        user_id: str,
+        filter: Optional[QueryScheduledMessagesFilter] = None,
+        sort: Optional[List[SortParam]] = None,
+        options: Optional[QueryScheduledMessagesOptions] = None,
+    ) -> StreamResponse:
+        data: Dict[str, Any] = {"user_id": user_id}
+        if filter is not None:
+            data["filter"] = cast(dict, filter)
+        if sort is not None:
+            data["sort"] = sort
+        if options is not None:
+            data.update(cast(dict, options))
+        return self.post("scheduled_messages/query", data=data)
+
     def get_user_locations(self, user_id: str, **options: Any) -> StreamResponse:
         params = {"user_id": user_id, **options}
         return self.get("users/live_locations", params=params)

stream_chat/tests/async_chat/test_scheduled_messages.py

@@ -0,0 +1,60 @@
+from datetime import datetime, timedelta, timezone
+from typing import Dict
+
+import pytest
+
+from stream_chat.async_chat import StreamChatAsync
+from stream_chat.base.exceptions import StreamAPIException
+
+
+async def schedule_message(channel, user_id: str) -> Dict:
+    """Schedules a message, skipping the test when the app cannot schedule.
+
+    Scheduled messages are enabled per app by Stream, not through the app
+    settings API, so the CI app cannot be baselined into having them.
+    """
+    try:
+        response = await channel.send_message(
+            {"text": "scheduled hello"},
+            user_id,
+            scheduled_at=datetime.now(timezone.utc) + timedelta(hours=1),
+        )
+    except StreamAPIException as e:
+        if "scheduled_messages_disabled" in e.response_text:
+            pytest.skip("scheduled messages are not enabled for this app")
+        raise
+    if "scheduled_message" not in response:
+        pytest.skip("the API does not support scheduled messages yet")
+    return response["scheduled_message"]
+
+
+class TestScheduledMessages:
+    @pytest.mark.asyncio
+    async def test_scheduled_message_lifecycle(
+        self, client: StreamChatAsync, channel, random_user
+    ):
+        scheduled = await schedule_message(channel, random_user["id"])
+        assert scheduled["cid"] == channel.cid
+        assert scheduled["status"] == "scheduled"
+
+        response = await client.query_scheduled_messages(
+            random_user["id"], filter={"cid": channel.cid}
+        )
+        assert [m["id"] for m in response["scheduled_messages"]] == [scheduled["id"]]
+
+        response = await client.update_scheduled_message(
+            scheduled["id"], message={"text": "edited"}
+        )
+        assert response["scheduled_message"]["message"]["text"] == "edited"
+
+        response = await client.delete_scheduled_message(scheduled["id"])
+        assert response["scheduled_message"]["id"] == scheduled["id"]
+
+    @pytest.mark.asyncio
+    async def test_send_scheduled_message_now(
+        self, client: StreamChatAsync, channel, random_user
+    ):
+        scheduled = await schedule_message(channel, random_user["id"])
+
+        response = await client.send_scheduled_message_now(scheduled["id"])
+        assert response["scheduled_message"]["id"] == scheduled["id"]