[CHA-5933] feat: add scheduled messages support

docs/messages/scheduled_messages.md

@@ -0,0 +1,64 @@
+Scheduled messages let a user write a message now and have it sent to a channel later. Server-side, you can schedule, edit, cancel and list scheduled messages on behalf of any user.
+
+Scheduled messages must be enabled for your app. Until the message is sent, only the user who scheduled it can see it; nothing is sent to the channel.
+
+## Scheduling a Message
+
+Send a message with a `scheduled_at` time. It must be at least 1 minute and at most the app's maximum lead time (120 days by default) from now. Slash commands and pending messages cannot be scheduled.
+
+```ruby
+response = channel.schedule_message({ 'text' => 'Good morning!' }, 'user-id', Time.now.utc + 3600)
+
+scheduled_message_id = response['scheduled_message']['id']
+```
+
+The response carries `scheduled_message`. Its `id` is used to edit, cancel or send the message now. The `message` object in the response holds the content as it will be sent; it is not in the channel yet.
+
+## Updating a Scheduled Message
+
+Change the message, the send time, or both. A new message replaces the scheduled one as a whole; the sender stays the user it was scheduled for.
+
+```ruby
+client.update_scheduled_message(scheduled_message_id, message: { 'text' => 'Good afternoon!' })
+
+client.update_scheduled_message(scheduled_message_id, scheduled_at: Time.now.utc + 7200)
+```
+
+## Sending a Scheduled Message Now
+
+```ruby
+client.send_scheduled_message_now(scheduled_message_id)
+```
+
+## Deleting a Scheduled Message
+
+Deleting cancels a scheduled message so it is never sent, or removes a failed one.
+
+```ruby
+client.delete_scheduled_message(scheduled_message_id)
+```
+
+Updating, deleting or sending now fails with a 409 error while the message is being sent.
+
+## Querying Scheduled Messages
+
+List the scheduled messages of a user across channels: pending ones first, then failed ones, each by `scheduled_at`. You can filter by channel and status (`scheduled`, `sending`, `failed`).
+
+```ruby
+response = client.query_scheduled_messages('user-id', filter: { cid: 'messaging:general', status: ['failed'] }, limit: 20)
+
+response['scheduled_messages'].each do |scheduled_message|
+  puts scheduled_message['status']
+end
+
+# next page
+client.query_scheduled_messages('user-id', limit: 20, next: response['next']) if response['next']
+```
+
+## Failed Messages
+
+When a scheduled message cannot be sent, its status becomes `failed` and `failure_reason` says why: `banned`, `no_permission`, `channel_frozen`, `channel_deleted`, `user_deleted`, `moderation_blocked`, `hook_rejected`, `expired`, `internal_error`, `invalid_message` or `pending_not_supported`. A failed message is not retried; delete it, and schedule it again if needed.
+
+## Webhooks
+
+The `scheduled_message.sent` and `scheduled_message.failed` events are sent to webhooks. `scheduled_message.sent` carries the ID of the sent message in `message_id`.

docs/webhooks/webhook_events.md

@@ -50,6 +50,8 @@ When applicable, the following attributes are included to the event user and to
 | reminder.updated             | when a message reminder is updated.                                 |
 | reminder.deleted             | when a message reminder is deleted.                                 |
 | notification.reminder_due    | when a message reminder's due time is reached.                      |
+| scheduled_message.sent       | when a scheduled message is sent; message_id is the sent message.   |
+| scheduled_message.failed     | when a scheduled message fails to send (see failure_reason).        |
 
 ### `message.new`
 

lib/stream-chat/channel.rb

@@ -55,6 +55,23 @@ def send_message(message, user_id, **options)
       @client.post("#{url}/message", data: payload)
     end
 
+    # Schedules a message to be sent to this channel at a later time.
+    #
+    # The message is not sent now: it is stored and sent at scheduled_at on
+    # behalf of user_id. The response carries `scheduled_message`, whose `id`
+    # is used to update, delete or send the scheduled message now.
+    #
+    # @param [StringKeyHash] message The message to send
+    # @param [String] user_id The ID of the user the message is sent as
+    # @param [DateTime, Time, String] scheduled_at When to send the message
+    #   (at least 1 minute and at most the app's maximum lead time from now)
+    # @param [Hash] options Additional SendMessage options, e.g. skip_push
+    # @return [StreamChat::StreamResponse]
+    sig { params(message: StringKeyHash, user_id: String, scheduled_at: T.any(DateTime, Time, String), options: T.untyped).returns(StreamChat::StreamResponse) }
+    def schedule_message(message, user_id, scheduled_at, **options)
+      send_message(message, user_id, **options, scheduled_at: StreamChat.normalize_timestamp(scheduled_at))
+    end
+
     # Sends an event on this channel.
     sig { params(event: StringKeyHash, user_id: String).returns(StreamChat::StreamResponse) }
     def send_event(event, user_id)

lib/stream-chat/client.rb

@@ -1222,6 +1222,58 @@ def query_reminders(user_id, filter_conditions = {}, sort: nil, **options)
       post('reminders/query', data: params)
     end
 
+    # Updates a scheduled message. Schedule a message with Channel#schedule_message.
+    #
+    # At least one of message or scheduled_at is required. The message, when
+    # given, replaces the scheduled message as a whole; the sender stays the
+    # user the message was scheduled for.
+    #
+    # @param id [String] The ID of the scheduled message
+    # @param message [StringKeyHash, nil] The new message
+    # @param scheduled_at [DateTime, Time, String, nil] The new send time
+    # @return [StreamChat::StreamResponse] API response with scheduled_message
+    sig { params(id: String, message: T.nilable(StringKeyHash), scheduled_at: T.nilable(T.any(DateTime, Time, String))).returns(StreamChat::StreamResponse) }
+    def update_scheduled_message(id, message: nil, scheduled_at: nil)
+      data = {}
+      data[:message] = message if message
+      data[:scheduled_at] = StreamChat.normalize_timestamp(scheduled_at) if scheduled_at
+      patch("scheduled_messages/#{id}", data: data)
+    end
+
+    # Deletes a scheduled message: cancels it so it is never sent, or removes a failed one.
+    #
+    # @param id [String] The ID of the scheduled message
+    # @return [StreamChat::StreamResponse] API response with scheduled_message
+    sig { params(id: String).returns(StreamChat::StreamResponse) }
+    def delete_scheduled_message(id)
+      delete("scheduled_messages/#{id}")
+    end
+
+    # Sends a scheduled message right away.
+    #
+    # @param id [String] The ID of the scheduled message
+    # @return [StreamChat::StreamResponse] API response with scheduled_message
+    sig { params(id: String).returns(StreamChat::StreamResponse) }
+    def send_scheduled_message_now(id)
+      post("scheduled_messages/#{id}/send_now")
+    end
+
+    # Queries the scheduled messages of a user across channels: pending ones
+    # first, then failed ones, each by scheduled_at.
+    #
+    # @param user_id [String] The ID of the user whose scheduled messages to query
+    # @param filter [StringKeyHash, nil] Optional filter, e.g. { cid: 'messaging:general', status: ['failed'] }
+    # @param sort [Array<Hash>, nil] Optional sort; only [{ field: 'scheduled_at', direction: 1 }] is supported
+    # @param options [Hash] Additional query options: limit (1 to 100, default 20) and next
+    # @return [StreamChat::StreamResponse] API response with scheduled_messages and next
+    sig { params(user_id: String, filter: T.nilable(StringKeyHash), sort: T.nilable(T::Array[StringKeyHash]), options: T.untyped).returns(StreamChat::StreamResponse) }
+    def query_scheduled_messages(user_id, filter: nil, sort: nil, **options)
+      data = options.merge({ user_id: user_id })
+      data[:filter] = filter if filter
+      data[:sort] = sort if sort
+      post('scheduled_messages/query', data: data)
+    end
+
     # Send the mark delivered event for this user, only works if the `delivery_receipts` setting is enabled
     #
     # @param [StringKeyHash, nil] data The delivery confirmation data

spec/client_spec.rb

@@ -1408,6 +1408,68 @@ def cleanup_fellowship_leftovers!(client)
     end
   end
 
+  describe 'scheduled messages' do
+    # Scheduled messages are behind a per-app flag. On an app without it the
+    # API rejects scheduled_at (scheduled_messages_disabled), so skip. A
+    # response without scheduled_message means the backend ignored
+    # scheduled_at and posted the message right away, so fail.
+    def schedule_or_skip(text)
+      response = @channel.schedule_message({ 'text' => text }, @frodo, Time.now.utc + 3600)
+      expect(response).to include('scheduled_message'),
+                          "scheduled_at was ignored and the message was sent immediately: #{response.inspect}"
+      response
+    rescue StreamChat::StreamAPIException => e
+      raise unless e.message.include?('scheduled messages are not enabled')
+
+      skip "scheduled messages not enabled for this Stream app: #{e.message}"
+    end
+
+    before(:each) do
+      @scheduled = schedule_or_skip('scheduled hello')['scheduled_message']
+    end
+
+    after(:each) do
+      @client.delete_scheduled_message(@scheduled['id']) if @scheduled
+    rescue StreamChat::StreamAPIException
+      # already deleted or sent by the test
+    end
+
+    it 'schedules a message' do
+      expect(@scheduled['cid']).to eq(@channel.cid)
+      expect(@scheduled['status']).to eq('scheduled')
+      expect(@scheduled['user']['id']).to eq(@frodo)
+      expect(@scheduled['message']['text']).to eq('scheduled hello')
+    end
+
+    it 'queries scheduled messages' do
+      response = @client.query_scheduled_messages(@frodo, filter: { cid: @channel.cid }, limit: 10)
+
+      expect(response['scheduled_messages'].map { |sm| sm['id'] }).to include(@scheduled['id'])
+    end
+
+    it 'updates a scheduled message' do
+      scheduled_at = Time.now.utc + 7200
+      response = @client.update_scheduled_message(@scheduled['id'], message: { 'text' => 'edited' }, scheduled_at: scheduled_at)
+
+      expect(response['scheduled_message']['message']['text']).to eq('edited')
+      expect(Time.parse(response['scheduled_message']['scheduled_at'])).to be_within(1).of(scheduled_at)
+    end
+
+    it 'sends a scheduled message now' do
+      response = @client.send_scheduled_message_now(@scheduled['id'])
+
+      expect(response['scheduled_message']['id']).to eq(@scheduled['id'])
+    end
+
+    it 'deletes a scheduled message' do
+      response = @client.delete_scheduled_message(@scheduled['id'])
+      expect(response['scheduled_message']['id']).to eq(@scheduled['id'])
+
+      remaining = @client.query_scheduled_messages(@frodo, filter: { cid: @channel.cid })
+      expect(remaining['scheduled_messages'].map { |sm| sm['id'] }).not_to include(@scheduled['id'])
+    end
+  end
+
   describe 'live locations' do
     before(:all) do
       @location_test_user = { id: SecureRandom.uuid }

spec/scheduled_messages_spec.rb

@@ -0,0 +1,102 @@
+# typed: false
+# frozen_string_literal: true
+
+require 'json'
+require 'stream-chat'
+
+# Request-shape tests for the scheduled message methods. They run against a
+# stubbed HTTP adapter, so they need no Stream credentials; the round trip
+# against a live app is covered in client_spec.rb.
+describe 'scheduled messages requests' do
+  let(:stubs) { Faraday::Adapter::Test::Stubs.new }
+  let(:request_bodies) { [] }
+  let(:client) do
+    client = StreamChat::Client.new('key', 'secret')
+    client.set_http_client(Faraday.new(url: 'https://chat.stream-io-api.com') { |f| f.adapter :test, stubs })
+    client
+  end
+  let(:response_body) { { scheduled_message: { id: 'sm-1' } }.to_json }
+
+  def stub_request(method, path)
+    stubs.public_send(method, path) do |env|
+      request_bodies << env.body
+      [200, { 'Content-Type' => 'application/json' }, response_body]
+    end
+  end
+
+  def sent_body
+    JSON.parse(request_bodies.last)
+  end
+
+  after { stubs.verify_stubbed_calls }
+
+  it 'schedule_message sends the message with scheduled_at' do
+    stub_request(:post, '/channels/messaging/general/message')
+    scheduled_at = Time.utc(2030, 1, 2, 3, 4, 5)
+
+    response = client.channel('messaging', channel_id: 'general')
+                     .schedule_message({ text: 'later' }, 'jane', scheduled_at, skip_push: true)
+
+    expect(response['scheduled_message']['id']).to eq('sm-1')
+    expect(sent_body).to eq(
+      'message' => { 'text' => 'later', 'user' => { 'id' => 'jane' } },
+      'scheduled_at' => '2030-01-02T03:04:05Z',
+      'skip_push' => true
+    )
+  end
+
+  it 'update_scheduled_message patches message and scheduled_at' do
+    stub_request(:patch, '/scheduled_messages/sm-1')
+
+    client.update_scheduled_message('sm-1', message: { text: 'edited' }, scheduled_at: '2030-01-02T03:04:05Z')
+
+    expect(sent_body).to eq(
+      'message' => { 'text' => 'edited' },
+      'scheduled_at' => '2030-01-02T03:04:05Z'
+    )
+  end
+
+  it 'update_scheduled_message sends only the fields given' do
+    stub_request(:patch, '/scheduled_messages/sm-1')
+
+    client.update_scheduled_message('sm-1', scheduled_at: DateTime.new(2030, 1, 2, 3, 4, 5))
+
+    expect(sent_body).to eq('scheduled_at' => '2030-01-02T03:04:05+00:00')
+  end
+
+  it 'delete_scheduled_message deletes by id' do
+    stub_request(:delete, '/scheduled_messages/sm-1')
+
+    response = client.delete_scheduled_message('sm-1')
+
+    expect(response['scheduled_message']['id']).to eq('sm-1')
+  end
+
+  it 'send_scheduled_message_now posts to send_now' do
+    stub_request(:post, '/scheduled_messages/sm-1/send_now')
+
+    response = client.send_scheduled_message_now('sm-1')
+
+    expect(response['scheduled_message']['id']).to eq('sm-1')
+  end
+
+  it 'query_scheduled_messages sends user_id, filter, sort and pagination' do
+    stub_request(:post, '/scheduled_messages/query')
+
+    client.query_scheduled_messages(
+      'jane',
+      filter: { cid: 'messaging:general', status: %w[scheduled failed] },
+      sort: [{ field: 'scheduled_at', direction: 1 }],
+      limit: 10,
+      next: 'cursor'
+    )
+
+    expect(sent_body).to eq(
+      'user_id' => 'jane',
+      'filter' => { 'cid' => 'messaging:general', 'status' => %w[scheduled failed] },
+      'sort' => [{ 'field' => 'scheduled_at', 'direction' => 1 }],
+      'limit' => 10,
+      'next' => 'cursor'
+    )
+  end
+end