feat(api): lazy pagination streams (history, members, bans, audit log)

lib/dexcord/api.ex

@@ -446,6 +446,167 @@ defmodule Dexcord.Api do
     end
   end
 
+  # --- Lazy pagination streams -------------------------------------------
+  #
+  # Each returns a lazy `Stream` over `Dexcord.Api.Paginate`: a page is fetched
+  # only when the consumer walks that far, so `Stream.take/2` on a fresh stream
+  # makes exactly one wire hit. A page-fetch `{:error, _}` raises
+  # `Dexcord.Api.Paginate.PageError` mid-stream (streams cannot carry a tagged
+  # tuple), which the caller can `rescue`.
+
+  @history_page_size 100
+  @members_page_size 1000
+  @bans_page_size 1000
+  @audit_log_page_size 100
+
+  @doc """
+  Streams a channel's message history, resolving `messageable` through
+  `Dexcord.Messageable`.
+
+  By default it pages newest→oldest using the `before:` anchor (cursor = the
+  last message's id). Passing `after: id` flips to oldest→newest paging with the
+  `after:` anchor (mirroring discord.py's `oldest_first`, which defaults true iff
+  `after` is given). `limit: n` caps the total number of messages yielded.
+
+  The stream is lazy — pages are fetched on demand — and raises
+  `Dexcord.Api.Paginate.PageError` if a page request fails.
+  """
+  @spec message_history(Dexcord.Messageable.t(), keyword()) :: Enumerable.t()
+  def message_history(messageable, opts \\ []) do
+    {:channel, channel_id} = Dexcord.Messageable.resolve(messageable)
+
+    stream =
+      case Keyword.fetch(opts, :after) do
+        {:ok, after_id} ->
+          Dexcord.Api.Paginate.stream(
+            after_id,
+            @history_page_size,
+            fn cursor ->
+              Dexcord.Api.Messages.get_channel_messages(channel_id,
+                limit: @history_page_size,
+                after: cursor
+              )
+            end,
+            fn last -> last.id end
+          )
+
+        :error ->
+          Dexcord.Api.Paginate.stream(
+            nil,
+            @history_page_size,
+            fn
+              nil ->
+                Dexcord.Api.Messages.get_channel_messages(channel_id, limit: @history_page_size)
+
+              cursor ->
+                Dexcord.Api.Messages.get_channel_messages(channel_id,
+                  limit: @history_page_size,
+                  before: cursor
+                )
+            end,
+            fn last -> last.id end
+          )
+      end
+
+    maybe_take(stream, Keyword.get(opts, :limit))
+  end
+
+  @doc """
+  Streams a guild's members ascending, paging with the `after:` anchor (cursor =
+  the last member's user id), 1000 per page. `limit: n` caps the total.
+  Lazy; raises `Dexcord.Api.Paginate.PageError` on a failed page.
+  """
+  @spec guild_members_stream(Dexcord.Guild.t() | Dexcord.Snowflake.t(), keyword()) ::
+          Enumerable.t()
+  def guild_members_stream(guild_or_id, opts \\ []) do
+    guild_id = resolve_guild_id(guild_or_id)
+    after0 = Keyword.get(opts, :after, 0)
+
+    stream =
+      Dexcord.Api.Paginate.stream(
+        after0,
+        @members_page_size,
+        fn cursor ->
+          Dexcord.Api.Members.list_guild_members(guild_id,
+            limit: @members_page_size,
+            after: cursor
+          )
+        end,
+        fn last -> member_user_id(last) end
+      )
+
+    maybe_take(stream, Keyword.get(opts, :limit))
+  end
+
+  @doc """
+  Streams a guild's bans ascending, paging with the `after:` anchor (cursor =
+  the last ban's user id), 1000 per page. `limit: n` caps the total.
+  Lazy; raises `Dexcord.Api.Paginate.PageError` on a failed page.
+  """
+  @spec guild_bans_stream(Dexcord.Guild.t() | Dexcord.Snowflake.t(), keyword()) :: Enumerable.t()
+  def guild_bans_stream(guild_or_id, opts \\ []) do
+    guild_id = resolve_guild_id(guild_or_id)
+    after0 = Keyword.get(opts, :after, 0)
+
+    stream =
+      Dexcord.Api.Paginate.stream(
+        after0,
+        @bans_page_size,
+        fn cursor ->
+          Dexcord.Api.Members.get_guild_bans(guild_id, limit: @bans_page_size, after: cursor)
+        end,
+        fn last -> last.user.id end
+      )
+
+    maybe_take(stream, Keyword.get(opts, :limit))
+  end
+
+  @doc """
+  Streams a guild's audit-log entries descending, paging with the `before:`
+  anchor (cursor = the last entry's id), 100 per page.
+
+  The `get_guild_audit_log` endpoint returns a `%Dexcord.AuditLog{}` container;
+  this stream yields the `audit_log_entries` out of it (the referenced
+  users/webhooks/etc. on the container are not threaded through). `limit: n`
+  caps the total. Lazy; raises `Dexcord.Api.Paginate.PageError` on a failed page.
+  """
+  @spec audit_log_stream(Dexcord.Guild.t() | Dexcord.Snowflake.t(), keyword()) :: Enumerable.t()
+  def audit_log_stream(guild_or_id, opts \\ []) do
+    guild_id = resolve_guild_id(guild_or_id)
+
+    stream =
+      Dexcord.Api.Paginate.stream(
+        nil,
+        @audit_log_page_size,
+        fn
+          nil ->
+            with {:ok, %Dexcord.AuditLog{} = al} <-
+                   Dexcord.Api.Guilds.get_guild_audit_log(guild_id, limit: @audit_log_page_size),
+                 do: {:ok, al.audit_log_entries}
+
+          cursor ->
+            with {:ok, %Dexcord.AuditLog{} = al} <-
+                   Dexcord.Api.Guilds.get_guild_audit_log(guild_id,
+                     limit: @audit_log_page_size,
+                     before: cursor
+                   ),
+                 do: {:ok, al.audit_log_entries}
+        end,
+        fn last -> last.id end
+      )
+
+    maybe_take(stream, Keyword.get(opts, :limit))
+  end
+
+  defp maybe_take(stream, nil), do: stream
+  defp maybe_take(stream, limit) when is_integer(limit), do: Stream.take(stream, limit)
+
+  defp resolve_guild_id(%{id: id}) when is_integer(id), do: id
+  defp resolve_guild_id(id) when is_integer(id), do: id
+  defp resolve_guild_id(other), do: Dexcord.Snowflake.cast!(other)
+
+  defp member_user_id(%{user_id: user_id, user: user}), do: user_id || (user && user.id)
+
   # --- Generated endpoint facade -----------------------------------------
   #
   # The typed endpoint surface lives in the group modules below (declared with

lib/dexcord/api/paginate.ex

@@ -0,0 +1,54 @@
+defmodule Dexcord.Api.Paginate do
+  @moduledoc false
+  # Lazy cursor pagination over Discord's list endpoints. Each page is fetched
+  # only when the stream is consumed that far (so `Stream.take/2` on a fresh
+  # stream makes exactly one wire hit). A page shorter than the page size ends
+  # the stream; an `{:error, _}` from the fetcher raises `PageError` — a stream
+  # cannot carry a tagged-tuple failure mid-flow, so the error surfaces as a
+  # raise the caller can `rescue`.
+
+  defmodule PageError do
+    @moduledoc """
+    Raised when a page fetch inside a pagination stream returns
+    `{:error, _}`. The underlying error is on the `:error` field.
+    """
+    defexception [:error]
+
+    @impl true
+    def message(%{error: e}), do: "pagination request failed: #{inspect(e)}"
+  end
+
+  @doc false
+  # `fetch_page.(cursor)` -> `{:ok, items}` | `{:error, e}`;
+  # `next.(last_item)` -> the cursor for the following page.
+  #
+  # `initial_cursor` seeds the first fetch (may be `nil` for an anchorless first
+  # page). The stream halts on the first page whose length is < `page_size`.
+  @spec stream(term(), pos_integer(), (term() -> {:ok, list()} | {:error, term()}), (term() ->
+                                                                                       term())) ::
+          Enumerable.t()
+  def stream(initial_cursor, page_size, fetch_page, next)
+      when is_integer(page_size) and page_size > 0 and is_function(fetch_page, 1) and
+             is_function(next, 1) do
+    Stream.resource(
+      fn -> {initial_cursor, :go} end,
+      fn
+        {_cursor, :halt} ->
+          {:halt, nil}
+
+        {cursor, :go} ->
+          case fetch_page.(cursor) do
+            {:ok, items} when length(items) < page_size ->
+              {items, {cursor, :halt}}
+
+            {:ok, items} ->
+              {items, {next.(List.last(items)), :go}}
+
+            {:error, error} ->
+              raise PageError, error: error
+          end
+      end,
+      fn _ -> :ok end
+    )
+  end
+end

test/dexcord/pagination_test.exs

@@ -0,0 +1,177 @@
+defmodule Dexcord.PaginationTest do
+  @moduledoc """
+  Wire tests for the lazy pagination streams on `Dexcord.Api`
+  (api-surface.AC3.7), driven through `Dexcord.FakeRest`.
+
+  The point of these streams is laziness: a page is fetched only when the
+  consumer walks that far, so `Stream.take/2` on a fresh stream makes exactly
+  one wire hit. These tests assert that directly (the second hit is refuted).
+  """
+  use ExUnit.Case, async: false
+
+  alias Dexcord.Api
+  alias Dexcord.Api.Paginate
+  alias Dexcord.Api.Ratelimit
+  alias Dexcord.FakeRest
+
+  @token "test.token.value"
+
+  setup do
+    Dexcord.EnvSandbox.sandbox_env()
+    Dexcord.Config.put(%{token: @token, handler: nil, intents: 0})
+
+    start_supervised!({Finch, name: Dexcord.Finch})
+    start_supervised!(Ratelimit)
+    start_supervised!(Dexcord.Cache)
+    start_supervised!(FakeRest)
+
+    Application.put_env(:dexcord, :api_base_url, FakeRest.base_url())
+    FakeRest.subscribe(self())
+
+    :ok
+  end
+
+  # Build a JSON array body of message objects with the given ids (as strings).
+  defp messages_json(ids),
+    do: "[" <> Enum.map_join(ids, ",", fn id -> ~s({"id":"#{id}"}) end) <> "]"
+
+  # Build a JSON array of member objects (nested user id) with the given ids.
+  defp members_json(ids),
+    do: "[" <> Enum.map_join(ids, ",", fn id -> ~s({"user":{"id":"#{id}"}}) end) <> "]"
+
+  describe "AC3.7: message_history pages lazily across multiple hits" do
+    test "two sequential pages yield every message in order, before= cursor on hit 2" do
+      page1_ids = Enum.to_list(200..101//-1)
+      page2_ids = [100, 99, 98]
+      assert length(page1_ids) == 100
+
+      FakeRest.stub(
+        :get,
+        "/channels/1/messages",
+        FakeRest.resp(200, body: messages_json(page1_ids))
+      )
+
+      FakeRest.stub(
+        :get,
+        "/channels/1/messages",
+        FakeRest.resp(200, body: messages_json(page2_ids))
+      )
+
+      result = Api.message_history(1) |> Enum.to_list()
+
+      assert Enum.map(result, & &1.id) == page1_ids ++ page2_ids
+
+      # Exactly two hits; the second carries before=<last id of page 1>.
+      assert_receive {:rest_hit, %{method: "GET", path: "/channels/1/messages", query_string: q1}}
+      refute q1 =~ "before="
+      assert_receive {:rest_hit, %{path: "/channels/1/messages", query_string: q2}}
+      assert q2 =~ "before=101"
+      refute_receive {:rest_hit, %{path: "/channels/1/messages"}}, 50
+    end
+
+    test "LAZINESS: Stream.take(5) makes exactly one wire hit" do
+      # A single, sticky 100-message page: if the stream were eager it would
+      # loop forever hitting the wire. take(5) must fetch exactly one page.
+      full_page = Enum.to_list(500..401//-1)
+      assert length(full_page) == 100
+
+      FakeRest.stub(
+        :get,
+        "/channels/7/messages",
+        FakeRest.resp(200, body: messages_json(full_page))
+      )
+
+      result = Api.message_history(7) |> Stream.take(5) |> Enum.to_list()
+
+      assert Enum.map(result, & &1.id) == [500, 499, 498, 497, 496]
+      assert_receive {:rest_hit, %{path: "/channels/7/messages"}}
+      refute_receive {:rest_hit, %{path: "/channels/7/messages"}}, 50
+    end
+
+    test "after: flips to ascending paging with after= cursor" do
+      page1_ids = Enum.to_list(101..200)
+      page2_ids = [201, 202, 203]
+      assert length(page1_ids) == 100
+
+      FakeRest.stub(
+        :get,
+        "/channels/2/messages",
+        FakeRest.resp(200, body: messages_json(page1_ids))
+      )
+
+      FakeRest.stub(
+        :get,
+        "/channels/2/messages",
+        FakeRest.resp(200, body: messages_json(page2_ids))
+      )
+
+      result = Api.message_history(2, after: 100) |> Enum.to_list()
+
+      assert Enum.map(result, & &1.id) == page1_ids ++ page2_ids
+
+      assert_receive {:rest_hit, %{path: "/channels/2/messages", query_string: q1}}
+      assert q1 =~ "after=100"
+      assert_receive {:rest_hit, %{path: "/channels/2/messages", query_string: q2}}
+      assert q2 =~ "after=200"
+      refute q2 =~ "before="
+    end
+
+    test "limit: caps the total number of messages via Stream.take" do
+      full_page = Enum.to_list(300..201//-1)
+
+      FakeRest.stub(
+        :get,
+        "/channels/3/messages",
+        FakeRest.resp(200, body: messages_json(full_page))
+      )
+
+      result = Api.message_history(3, limit: 3) |> Enum.to_list()
+
+      assert Enum.map(result, & &1.id) == [300, 299, 298]
+      assert_receive {:rest_hit, %{path: "/channels/3/messages"}}
+      refute_receive {:rest_hit, %{path: "/channels/3/messages"}}, 50
+    end
+  end
+
+  describe "guild_members_stream pages ascending on the last user_id" do
+    test "cursor is the last member's user id" do
+      page1_ids = Enum.to_list(1..1000)
+      page2_ids = [1001, 1002]
+      assert length(page1_ids) == 1000
+
+      FakeRest.stub(:get, "/guilds/9/members", FakeRest.resp(200, body: members_json(page1_ids)))
+      FakeRest.stub(:get, "/guilds/9/members", FakeRest.resp(200, body: members_json(page2_ids)))
+
+      result = Api.guild_members_stream(9) |> Enum.to_list()
+
+      assert Enum.map(result, & &1.user.id) == page1_ids ++ page2_ids
+
+      assert_receive {:rest_hit, %{path: "/guilds/9/members", query_string: q1}}
+      assert q1 =~ "after=0"
+      assert_receive {:rest_hit, %{path: "/guilds/9/members", query_string: q2}}
+      assert q2 =~ "after=1000"
+    end
+  end
+
+  describe "error mid-stream" do
+    test "an {:error, _} page raises Dexcord.Api.Paginate.PageError" do
+      full_page = Enum.to_list(200..101//-1)
+
+      FakeRest.stub(
+        :get,
+        "/channels/5/messages",
+        FakeRest.resp(200, body: messages_json(full_page))
+      )
+
+      FakeRest.stub(
+        :get,
+        "/channels/5/messages",
+        FakeRest.resp(500, body: ~s({"message":"boom"}))
+      )
+
+      assert_raise Paginate.PageError, fn ->
+        Api.message_history(5) |> Enum.to_list()
+      end
+    end
+  end
+end