diff --git a/server_ex/mix.exs b/server_ex/mix.exs index a024038..a3c107e 100644 --- a/server_ex/mix.exs +++ b/server_ex/mix.exs @@ -24,7 +24,11 @@ defmodule Topical.MixProject do [ {:ex_doc, ">= 0.0.0", only: :dev, runtime: false}, {:jason, "~> 1.4"}, - {:cowboy, "~> 2.9", optional: true} + {:cowboy, "~> 2.9", optional: true}, + {:gun, "~> 2.0", only: :test}, + {:bandit, "~> 1.1", only: :test}, + {:plug, "~> 1.15", only: :test}, + {:websock_adapter, "~> 0.5", only: :test} ] end diff --git a/server_ex/mix.lock b/server_ex/mix.lock index 6483ef4..074ce65 100644 --- a/server_ex/mix.lock +++ b/server_ex/mix.lock @@ -1,12 +1,22 @@ %{ + "bandit": {:hex, :bandit, "1.10.2", "d15ea32eb853b5b42b965b24221eb045462b2ba9aff9a0bda71157c06338cbff", [:mix], [{:hpax, "~> 1.0", [hex: :hpax, repo: "hexpm", optional: false]}, {:plug, "~> 1.18", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:thousand_island, "~> 1.0", [hex: :thousand_island, repo: "hexpm", optional: false]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "27b2a61b647914b1726c2ced3601473be5f7aa6bb468564a688646a689b3ee45"}, "cowboy": {:hex, :cowboy, "2.10.0", "ff9ffeff91dae4ae270dd975642997afe2a1179d94b1887863e43f681a203e26", [:make, :rebar3], [{:cowlib, "2.12.1", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "3afdccb7183cc6f143cb14d3cf51fa00e53db9ec80cdcd525482f5e99bc41d6b"}, "cowlib": {:hex, :cowlib, "2.12.1", "a9fa9a625f1d2025fe6b462cb865881329b5caff8f1854d1cbc9f9533f00e1e1", [:make, :rebar3], [], "hexpm", "163b73f6367a7341b33c794c4e88e7dbfe6498ac42dcd69ef44c5bc5507c8db0"}, "earmark_parser": {:hex, :earmark_parser, "1.4.29", "149d50dcb3a93d9f3d6f3ecf18c918fb5a2d3c001b5d3305c926cddfbd33355b", [:mix], [], "hexpm", "4902af1b3eb139016aed210888748db8070b8125c2342ce3dcae4f38dcc63503"}, "ex_doc": {:hex, :ex_doc, "0.29.1", "b1c652fa5f92ee9cf15c75271168027f92039b3877094290a75abcaac82a9f77", [:mix], [{:earmark_parser, "~> 1.4.19", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "b7745fa6374a36daf484e2a2012274950e084815b936b1319aeebcf7809574f6"}, + "gun": {:hex, :gun, "2.0.1", "160a9a5394800fcba41bc7e6d421295cf9a7894c2252c0678244948e3336ad73", [:make, :rebar3], [{:cowlib, "2.12.1", [hex: :cowlib, repo: "hexpm", optional: false]}], "hexpm", "a10bc8d6096b9502205022334f719cc9a08d9adcfbfc0dbee9ef31b56274a20b"}, + "hpax": {:hex, :hpax, "1.0.3", "ed67ef51ad4df91e75cc6a1494f851850c0bd98ebc0be6e81b026e765ee535aa", [:mix], [], "hexpm", "8eab6e1cfa8d5918c2ce4ba43588e894af35dbd8e91e6e55c817bca5847df34a"}, "jason": {:hex, :jason, "1.4.0", "e855647bc964a44e2f67df589ccf49105ae039d4179db7f6271dfd3843dc27e6", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "79a3791085b2a0f743ca04cec0f7be26443738779d09302e01318f97bdb82121"}, "makeup": {:hex, :makeup, "1.1.0", "6b67c8bc2882a6b6a445859952a602afc1a41c2e08379ca057c0f525366fc3ca", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "0a45ed501f4a8897f580eabf99a2e5234ea3e75a4373c8a52824f6e873be57a6"}, "makeup_elixir": {:hex, :makeup_elixir, "0.16.0", "f8c570a0d33f8039513fbccaf7108c5d750f47d8defd44088371191b76492b0b", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "28b2cbdc13960a46ae9a8858c4bebdec3c9a6d7b4b9e7f4ed1502f8159f338e7"}, "makeup_erlang": {:hex, :makeup_erlang, "0.1.1", "3fcb7f09eb9d98dc4d208f49cc955a34218fc41ff6b84df7c75b3e6e533cc65f", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "174d0809e98a4ef0b3309256cbf97101c6ec01c4ab0b23e926a9e17df2077cbb"}, + "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"}, "nimble_parsec": {:hex, :nimble_parsec, "1.2.3", "244836e6e3f1200c7f30cb56733fd808744eca61fd182f731eac4af635cc6d0b", [:mix], [], "hexpm", "c8d789e39b9131acf7b99291e93dae60ab48ef14a7ee9d58c6964f59efb570b0"}, + "plug": {:hex, :plug, "1.19.1", "09bac17ae7a001a68ae393658aa23c7e38782be5c5c00c80be82901262c394c0", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "560a0017a8f6d5d30146916862aaf9300b7280063651dd7e532b8be168511e62"}, + "plug_crypto": {:hex, :plug_crypto, "2.1.1", "19bda8184399cb24afa10be734f84a16ea0a2bc65054e23a62bb10f06bc89491", [:mix], [], "hexpm", "6470bce6ffe41c8bd497612ffde1a7e4af67f36a15eea5f921af71cf3e11247c"}, "ranch": {:hex, :ranch, "1.8.0", "8c7a100a139fd57f17327b6413e4167ac559fbc04ca7448e9be9057311597a1d", [:make, :rebar3], [], "hexpm", "49fbcfd3682fab1f5d109351b61257676da1a2fdbe295904176d5e521a2ddfe5"}, + "telemetry": {:hex, :telemetry, "1.3.0", "fedebbae410d715cf8e7062c96a1ef32ec22e764197f70cda73d82778d61e7a2", [:rebar3], [], "hexpm", "7015fc8919dbe63764f4b4b87a95b7c0996bd539e0d499be6ec9d7f3875b79e6"}, + "thousand_island": {:hex, :thousand_island, "1.4.3", "2158209580f633be38d43ec4e3ce0a01079592b9657afff9080d5d8ca149a3af", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "6e4ce09b0fd761a58594d02814d40f77daff460c48a7354a15ab353bb998ea0b"}, + "websock": {:hex, :websock, "0.5.3", "2f69a6ebe810328555b6fe5c831a851f485e303a7c8ce6c5f675abeb20ebdadc", [:mix], [], "hexpm", "6105453d7fac22c712ad66fab1d45abdf049868f253cf719b625151460b8b453"}, + "websock_adapter": {:hex, :websock_adapter, "0.5.9", "43dc3ba6d89ef5dec5b1d0a39698436a1e856d000d84bf31a3149862b01a287f", [:mix], [{:bandit, ">= 0.6.0", [hex: :bandit, repo: "hexpm", optional: true]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.6", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "5534d5c9adad3c18a0f58a9371220d75a803bf0b9a3d87e6fe072faaeed76a08"}, } diff --git a/server_ex/test/integration_test.exs b/server_ex/test/integration_test.exs deleted file mode 100644 index d32e07d..0000000 --- a/server_ex/test/integration_test.exs +++ /dev/null @@ -1,412 +0,0 @@ -defmodule Topical.IntegrationTest do - use ExUnit.Case - - setup do - registry_name = :"integration_registry_#{System.unique_integer([:positive])}" - - start_supervised!(%{ - id: registry_name, - start: - {Topical.Registry, :start_link, - [ - [ - name: registry_name, - topics: [ - Topical.Test.CounterTopic, - Topical.Test.AuthorizedTopic, - Topical.Test.CallbackTopic, - Topical.Test.FailingTopic, - Topical.Test.ListTopic, - Topical.Test.MergeTopic, - Topical.Test.LeaderboardTopic, - Topical.Test.DocumentTopic - ] - ] - ]} - }) - - {:ok, registry: registry_name} - end - - describe "subscribe/4" do - test "subscriber receives reset message", %{registry: registry} do - {:ok, ref, _server} = Topical.subscribe(registry, ["counters", "1"], self()) - - assert_receive {:reset, ^ref, %{count: 0}} - end - - test "returns ref that matches messages", %{registry: registry} do - {:ok, ref1, _server1} = Topical.subscribe(registry, ["counters", "1"], self()) - {:ok, ref2, _server2} = Topical.subscribe(registry, ["counters", "2"], self()) - - assert ref1 != ref2 - - assert_receive {:reset, ^ref1, _} - assert_receive {:reset, ^ref2, _} - end - - test "subscriber receives updates after subscribe", %{registry: registry} do - {:ok, ref, _server} = Topical.subscribe(registry, ["counters", "1"], self()) - assert_receive {:reset, ^ref, %{count: 0}} - - Topical.execute(registry, ["counters", "1"], "increment", {}) - - assert_receive {:updates, ^ref, [{:set, [:count], 1}]} - end - end - - describe "unsubscribe/2" do - test "stops receiving updates after unsubscribe", %{registry: registry} do - {:ok, ref, server} = Topical.subscribe(registry, ["counters", "1"], self()) - assert_receive {:reset, ^ref, %{count: 0}} - - Topical.unsubscribe(server, ref) - - Topical.execute(registry, ["counters", "1"], "increment", {}) - - refute_receive {:updates, _, _}, 100 - end - end - - describe "execute/5" do - test "returns result from topic", %{registry: registry} do - {:ok, result} = Topical.execute(registry, ["counters", "1"], "increment", {}) - assert result == 1 - - {:ok, result} = Topical.execute(registry, ["counters", "1"], "increment", {}) - assert result == 2 - end - - test "subscribers receive updates from execute", %{registry: registry} do - {:ok, ref, _server} = Topical.subscribe(registry, ["counters", "1"], self()) - assert_receive {:reset, ^ref, %{count: 0}} - - {:ok, _} = Topical.execute(registry, ["counters", "1"], "set", {42}) - - assert_receive {:updates, ^ref, [{:set, [:count], 42}]} - end - - test "returns error for unknown topic", %{registry: registry} do - assert {:error, :not_found} = - Topical.execute(registry, ["unknown", "topic"], "action", {}) - end - end - - describe "notify/5" do - test "sends notification to topic", %{registry: registry} do - {:ok, ref, _server} = Topical.subscribe(registry, ["counters", "1"], self()) - assert_receive {:reset, ^ref, %{count: 0}} - - :ok = Topical.notify(registry, ["counters", "1"], "increment", {}) - - assert_receive {:updates, ^ref, [{:set, [:count], 1}]} - end - - test "returns :ok without waiting for result", %{registry: registry} do - result = Topical.notify(registry, ["counters", "1"], "set", {100}) - assert result == :ok - end - - test "returns error for unknown topic", %{registry: registry} do - assert {:error, :not_found} = Topical.notify(registry, ["unknown", "topic"], "action", {}) - end - end - - describe "capture/3" do - test "returns current topic value", %{registry: registry} do - Topical.execute(registry, ["counters", "1"], "set", {42}) - - {:ok, value} = Topical.capture(registry, ["counters", "1"]) - assert value == %{count: 42} - end - - test "does not subscribe", %{registry: registry} do - {:ok, _value} = Topical.capture(registry, ["counters", "1"]) - - Topical.execute(registry, ["counters", "1"], "increment", {}) - - refute_receive {:updates, _, _}, 100 - end - - test "returns error for unknown topic", %{registry: registry} do - assert {:error, :not_found} = Topical.capture(registry, ["unknown", "topic"]) - end - end - - describe "multiple subscribers" do - test "all subscribers receive same updates", %{registry: registry} do - {:ok, ref1, _server1} = Topical.subscribe(registry, ["counters", "1"], self()) - {:ok, ref2, _server2} = Topical.subscribe(registry, ["counters", "1"], self()) - - assert_receive {:reset, ^ref1, %{count: 0}} - assert_receive {:reset, ^ref2, %{count: 0}} - - Topical.execute(registry, ["counters", "1"], "increment", {}) - - assert_receive {:updates, ^ref1, [{:set, [:count], 1}]} - assert_receive {:updates, ^ref2, [{:set, [:count], 1}]} - end - - test "unsubscribing one does not affect others", %{registry: registry} do - {:ok, ref1, server} = Topical.subscribe(registry, ["counters", "1"], self()) - {:ok, ref2, _server2} = Topical.subscribe(registry, ["counters", "1"], self()) - - assert_receive {:reset, ^ref1, _} - assert_receive {:reset, ^ref2, _} - - Topical.unsubscribe(server, ref1) - Topical.execute(registry, ["counters", "1"], "increment", {}) - - # ref1 should not receive update - refute_receive {:updates, ^ref1, _}, 100 - # ref2 should still receive update - assert_receive {:updates, ^ref2, [{:set, [:count], 1}]} - end - end - - describe "authorization" do - test "subscribe respects authorization", %{registry: registry} do - assert {:error, :forbidden} = - Topical.subscribe(registry, ["private", "owner1"], self(), %{user_id: "other"}) - - {:ok, _ref, _server} = - Topical.subscribe(registry, ["private", "owner1"], self(), %{user_id: "owner1"}) - end - - test "execute respects authorization", %{registry: registry} do - assert {:error, :forbidden} = - Topical.execute( - registry, - ["private", "owner1"], - "get_data", - {}, - %{user_id: "other"} - ) - - {:ok, _} = - Topical.execute(registry, ["private", "owner1"], "get_data", {}, %{user_id: "owner1"}) - end - - test "notify respects authorization", %{registry: registry} do - assert {:error, :forbidden} = - Topical.notify( - registry, - ["private", "owner1"], - "set_data", - {"test"}, - %{user_id: "other"} - ) - - :ok = - Topical.notify( - registry, - ["private", "owner1"], - "set_data", - {"test"}, - %{user_id: "owner1"} - ) - end - - test "capture respects authorization", %{registry: registry} do - assert {:error, :forbidden} = - Topical.capture(registry, ["private", "owner1"], %{user_id: "other"}) - - {:ok, _value} = Topical.capture(registry, ["private", "owner1"], %{user_id: "owner1"}) - end - end - - describe "callback invocations" do - test "handle_subscribe is called on subscribe", %{registry: registry} do - context = %{user: "test"} - {:ok, ref, _server} = Topical.subscribe(registry, ["callbacks", "1"], self(), context) - - assert_receive {:reset, ^ref, %{callbacks: callbacks}} - assert [{:subscribe, ^context}] = callbacks - end - - test "handle_unsubscribe is called on unsubscribe", %{registry: registry} do - context = %{user: "test"} - {:ok, ref, server} = Topical.subscribe(registry, ["callbacks", "1"], self(), context) - assert_receive {:reset, ^ref, _} - - Topical.unsubscribe(server, ref) - - # Give time for unsubscribe to process - Process.sleep(50) - - {:ok, value} = Topical.capture(registry, ["callbacks", "1"]) - # Note: capture also adds a {:capture, nil} callback, so check for unsubscribe presence - assert Enum.any?(value.callbacks, fn - {:unsubscribe, ^context} -> true - _ -> false - end) - end - - test "handle_capture is called on capture", %{registry: registry} do - context = %{user: "test"} - {:ok, value} = Topical.capture(registry, ["callbacks", "1"], context) - - assert [{:capture, ^context}] = value.callbacks - end - - test "handle_execute is called on execute", %{registry: registry} do - context = %{user: "test"} - {:ok, _} = Topical.execute(registry, ["callbacks", "1"], "action", {"arg"}, context) - - {:ok, value} = Topical.capture(registry, ["callbacks", "1"]) - # Note: capture also adds a {:capture, nil} callback, so check for execute presence - assert Enum.any?(value.callbacks, fn - {:execute, {"arg"}, ^context} -> true - _ -> false - end) - end - - test "handle_notify is called on notify", %{registry: registry} do - context = %{user: "test"} - :ok = Topical.notify(registry, ["callbacks", "1"], "action", {"arg"}, context) - - # Give time for notify to process - Process.sleep(50) - - {:ok, value} = Topical.capture(registry, ["callbacks", "1"]) - # Note: capture also adds a {:capture, nil} callback, so check for notify presence - assert Enum.any?(value.callbacks, fn - {:notify, {"arg"}, ^context} -> true - _ -> false - end) - end - end - - describe "topic timeout" do - test "topic stops after timeout with no subscribers", %{registry: registry} do - # Start a topic and get the server PID - {:ok, sub_ref, server} = Topical.subscribe(registry, ["counters", "timeout-test"], self()) - assert_receive {:reset, _, _} - assert Process.alive?(server) - - # Unsubscribe so there are no subscribers - Topical.unsubscribe(server, sub_ref) - - # We use a reference to monitor the process - ref = Process.monitor(server) - - # Topic should still be running after short delay (timeout is 10 seconds) - refute_receive {:DOWN, ^ref, :process, ^server, _}, 100 - - # Clean up - Process.demonitor(ref, [:flush]) - end - - test "topic does not timeout while subscribed", %{registry: registry} do - {:ok, _sub_ref, server} = Topical.subscribe(registry, ["counters", "sub-test"], self()) - assert_receive {:reset, _, _} - - mon_ref = Process.monitor(server) - - # Should not timeout while subscribed - refute_receive {:DOWN, ^mon_ref, :process, ^server, _}, 200 - - Process.demonitor(mon_ref, [:flush]) - end - end - - describe "list operations" do - test "insert operations broadcast to subscribers", %{registry: registry} do - {:ok, ref, _server} = Topical.subscribe(registry, ["lists", "1"], self()) - assert_receive {:reset, ^ref, %{items: [], next_id: 1}} - - {:ok, 1} = Topical.execute(registry, ["lists", "1"], "add", {"first"}) - - assert_receive {:updates, ^ref, updates} - assert {:insert, [:items], nil, [%{id: 1, value: "first"}]} in updates - end - - test "delete operations broadcast to subscribers", %{registry: registry} do - {:ok, ref, _server} = Topical.subscribe(registry, ["lists", "1"], self()) - assert_receive {:reset, ^ref, _} - - {:ok, _} = Topical.execute(registry, ["lists", "1"], "add", {"first"}) - assert_receive {:updates, ^ref, _} - - {:ok, _} = Topical.execute(registry, ["lists", "1"], "add", {"second"}) - assert_receive {:updates, ^ref, _} - - {:ok, _} = Topical.execute(registry, ["lists", "1"], "remove", {0}) - - assert_receive {:updates, ^ref, [{:delete, [:items], 0, 1}]} - end - end - - describe "merge operations" do - test "merge operations broadcast to subscribers", %{registry: registry} do - {:ok, ref, _server} = Topical.subscribe(registry, ["merge", "1"], self()) - assert_receive {:reset, ^ref, %{data: %{}}} - - {:ok, _} = Topical.execute(registry, ["merge", "1"], "merge", {%{a: 1, b: 2}}) - - assert_receive {:updates, ^ref, [{:merge, [:data], %{a: 1, b: 2}}]} - end - - test "unset operations broadcast to subscribers", %{registry: registry} do - {:ok, ref, _server} = Topical.subscribe(registry, ["merge", "1"], self()) - assert_receive {:reset, ^ref, _} - - {:ok, _} = Topical.execute(registry, ["merge", "1"], "set", {:key, "value"}) - assert_receive {:updates, ^ref, _} - - {:ok, _} = Topical.execute(registry, ["merge", "1"], "unset", {:key}) - - assert_receive {:updates, ^ref, [{:unset, [:data], :key}]} - end - end - - describe "subscriber process death" do - test "subscriber is removed when process dies", %{registry: registry} do - # Spawn a process that subscribes then dies - test_pid = self() - - subscriber = - spawn(fn -> - {:ok, _ref, _server} = Topical.subscribe(registry, ["callbacks", "death-test"], self()) - send(test_pid, :subscribed) - - receive do - :die -> :ok - end - end) - - assert_receive :subscribed - - # Kill the subscriber - send(subscriber, :die) - - # Wait for process to die and unsubscribe to be processed - Process.sleep(100) - - # Check that unsubscribe was called - {:ok, value} = Topical.capture(registry, ["callbacks", "death-test"]) - # Note: capture also adds a {:capture, nil} callback, so check for unsubscribe presence - assert Enum.any?(value.callbacks, fn - {:unsubscribe, _} -> true - _ -> false - end) - end - end - - describe "child_spec/1" do - test "returns valid child spec" do - spec = Topical.child_spec(name: MyApp.Topical, topics: []) - - # Default id is Topical when :server option is not provided - assert spec.id == Topical - assert spec.type == :supervisor - assert {Topical.Registry, :start_link, [_opts]} = spec.start - end - - test "uses :server option as id when provided" do - spec = Topical.child_spec(name: SomeRegistry, server: CustomId, topics: []) - - assert spec.id == CustomId - end - end -end diff --git a/server_ex/test/support/test_plug_router.ex b/server_ex/test/support/test_plug_router.ex new file mode 100644 index 0000000..429e515 --- /dev/null +++ b/server_ex/test/support/test_plug_router.ex @@ -0,0 +1,57 @@ +defmodule Topical.Test.PlugRouter do + use Plug.Router + + plug(:match) + plug(:dispatch) + + get "/socket" do + opts = conn.private[:topical_opts] || [] + registry = Keyword.fetch!(opts, :registry) + init_fn = Keyword.get(opts, :init) + + ws_opts = [registry: registry] ++ if(init_fn, do: [init: init_fn], else: []) + + conn + |> WebSockAdapter.upgrade(Topical.Adapters.Plug.WebSockServer, ws_opts, timeout: 60_000) + |> halt() + end + + get "/topics/*topic" do + opts = conn.private[:topical_opts] || [] + registry = Keyword.fetch!(opts, :registry) + init_fn = Keyword.get(opts, :init) + context = if init_fn, do: init_fn.(conn) |> elem(1), else: nil + + case Topical.capture(registry, topic, context) do + {:ok, value} -> + conn + |> put_resp_content_type("application/json") + |> send_resp(200, Jason.encode!(value)) + + {:error, :not_found} -> + conn + |> put_resp_content_type("application/json") + |> send_resp(404, Jason.encode!(%{"error" => "not_found"})) + + {:error, :unauthorized} -> + conn + |> put_resp_content_type("application/json") + |> send_resp(403, Jason.encode!(%{"error" => "unauthorized"})) + + {:error, error} -> + conn + |> put_resp_content_type("application/json") + |> send_resp(400, Jason.encode!(%{"error" => error})) + end + end + + match _ do + send_resp(conn, 404, "Not found") + end + + def call(conn, opts) do + conn + |> Plug.Conn.put_private(:topical_opts, opts) + |> super(opts) + end +end diff --git a/server_ex/test/support/websocket_test_helper.ex b/server_ex/test/support/websocket_test_helper.ex new file mode 100644 index 0000000..b0986e7 --- /dev/null +++ b/server_ex/test/support/websocket_test_helper.ex @@ -0,0 +1,189 @@ +defmodule Topical.Test.WebSocketTestHelper do + @moduledoc false + + @all_topics [ + Topical.Test.CounterTopic, + Topical.Test.AuthorizedTopic, + Topical.Test.CallbackTopic, + Topical.Test.FailingTopic, + Topical.Test.ListTopic, + Topical.Test.MergeTopic, + Topical.Test.LeaderboardTopic, + Topical.Test.DocumentTopic + ] + + @doc """ + Starts a Topical registry and HTTP server on a random port. + Returns %{port: port, registry: registry_name}. + """ + def start_server(adapter, opts \\ []) do + init_fn = Keyword.get(opts, :init) + registry_name = :"ws_test_registry_#{System.unique_integer([:positive])}" + + ExUnit.Callbacks.start_supervised!(%{ + id: registry_name, + start: {Topical.Registry, :start_link, [[name: registry_name, topics: @all_topics]]} + }) + + port = + case adapter do + :cowboy -> start_cowboy(registry_name, init_fn) + :bandit -> start_bandit(registry_name, init_fn) + end + + %{port: port, registry: registry_name} + end + + defp start_cowboy(registry, init_fn) do + alias Topical.Adapters.Cowboy.WebsocketHandler + + ws_opts = [registry: registry] ++ if(init_fn, do: [init: init_fn], else: []) + + alias Topical.Adapters.Cowboy.RestHandler + rest_opts = [registry: registry] ++ if(init_fn, do: [init: init_fn], else: []) + + dispatch = + :cowboy_router.compile([ + {:_, + [ + {"/socket", WebsocketHandler, ws_opts}, + {"/topics/[...]", RestHandler, rest_opts} + ]} + ]) + + ref = :"cowboy_ws_test_#{System.unique_integer([:positive])}" + trans_opts = %{socket_opts: [port: 0]} + proto_opts = %{env: %{dispatch: dispatch}, connection_type: :supervisor} + + # Ranch returns old-format tuple child specs; convert to map for start_supervised! + {id, start, restart, shutdown, type, modules} = + :ranch.child_spec(ref, :ranch_tcp, trans_opts, :cowboy_clear, proto_opts) + + ExUnit.Callbacks.start_supervised!(%{ + id: id, + start: start, + restart: restart, + shutdown: shutdown, + type: type, + modules: modules + }) + + :ranch.get_port(ref) + end + + defp start_bandit(registry, init_fn) do + plug_opts = [registry: registry] ++ if(init_fn, do: [init: init_fn], else: []) + + bandit = + ExUnit.Callbacks.start_supervised!( + {Bandit, plug: {Topical.Test.PlugRouter, plug_opts}, port: 0, startup_log: false} + ) + + {:ok, {_addr, port}} = ThousandIsland.listener_info(bandit) + port + end + + @doc """ + Opens a Gun WebSocket connection. Returns %{conn: pid, stream_ref: ref}. + """ + def ws_connect(port) do + {:ok, conn} = :gun.open(~c"localhost", port, %{protocols: [:http]}) + {:ok, :http} = :gun.await_up(conn, 5_000) + stream_ref = :gun.ws_upgrade(conn, ~c"/socket") + + receive do + {:gun_upgrade, ^conn, ^stream_ref, ["websocket"], _headers} -> + %{conn: conn, stream_ref: stream_ref} + after + 5_000 -> raise "WebSocket upgrade timed out" + end + end + + @doc """ + JSON-encodes and sends a WebSocket text frame. + """ + def ws_send(%{conn: conn, stream_ref: stream_ref} = _ws, message) do + :gun.ws_send(conn, stream_ref, {:text, Jason.encode!(message)}) + end + + @doc """ + Receives and JSON-decodes the next WebSocket text frame. + """ + def ws_receive(%{conn: conn, stream_ref: stream_ref} = _ws, timeout \\ 5_000) do + receive do + {:gun_ws, ^conn, ^stream_ref, {:text, text}} -> + Jason.decode!(text) + after + timeout -> raise "No WebSocket message received within #{timeout}ms" + end + end + + @doc """ + Receives N WebSocket messages. + """ + def ws_receive_all(ws, count) do + Enum.map(1..count, fn _ -> ws_receive(ws) end) + end + + @doc """ + Asserts no WebSocket message arrives within the timeout. + """ + def ws_refute_receive(%{conn: conn, stream_ref: stream_ref} = _ws, timeout \\ 200) do + receive do + {:gun_ws, ^conn, ^stream_ref, {:text, text}} -> + raise "Expected no WebSocket message, but received: #{text}" + after + timeout -> :ok + end + end + + @doc """ + Closes the WebSocket connection. + """ + def ws_close(%{conn: conn} = _ws) do + :gun.close(conn) + end + + @doc """ + Makes an HTTP GET request to the capture endpoint. Returns {status, body}. + """ + def http_get(port, path) do + {:ok, conn} = :gun.open(~c"localhost", port, %{protocols: [:http]}) + {:ok, :http} = :gun.await_up(conn, 5_000) + stream_ref = :gun.get(conn, String.to_charlist(path)) + {:response, :nofin, status, _headers} = :gun.await(conn, stream_ref, 5_000) + {:ok, body} = :gun.await_body(conn, stream_ref, 5_000) + :gun.close(conn) + {status, Jason.decode!(body)} + end + + # Protocol message builders + + def subscribe_msg(channel_id, topic, params \\ nil) do + if params do + [2, channel_id, topic, params] + else + [2, channel_id, topic] + end + end + + def unsubscribe_msg(channel_id) do + [3, channel_id] + end + + def execute_msg(channel_id, topic, action, args, params \\ nil) do + if params do + [1, channel_id, topic, action, args, params] + else + [1, channel_id, topic, action, args] + end + end + + def notify_msg(topic, action, args, params \\ nil) do + if params do + [0, topic, action, args, params] + else + [0, topic, action, args] + end + end +end diff --git a/server_ex/test/test_helper.exs b/server_ex/test/test_helper.exs index bb0d9b6..18a6b7c 100644 --- a/server_ex/test/test_helper.exs +++ b/server_ex/test/test_helper.exs @@ -2,3 +2,5 @@ ExUnit.start() # Load test support files Code.require_file("support/test_topics.ex", __DIR__) +Code.require_file("support/test_plug_router.ex", __DIR__) +Code.require_file("support/websocket_test_helper.ex", __DIR__) diff --git a/server_ex/test/websocket_test.exs b/server_ex/test/websocket_test.exs new file mode 100644 index 0000000..d3077ec --- /dev/null +++ b/server_ex/test/websocket_test.exs @@ -0,0 +1,400 @@ +for adapter <- [:cowboy, :bandit] do + module_name = + Module.concat(Topical.WebSocketTest, adapter |> Atom.to_string() |> Macro.camelize()) + + defmodule module_name do + use ExUnit.Case, async: false + + import Topical.Test.WebSocketTestHelper + + @adapter adapter + + setup do + %{port: port} = start_server(@adapter) + ws = ws_connect(port) + on_exit(fn -> ws_close(ws) end) + {:ok, ws: ws, port: port} + end + + describe "subscribe & reset" do + test "subscribe receives initial value", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "sub-reset-1"])) + msg = ws_receive(ws) + + # [2, channel_id, value] = topic_reset + assert [2, "ch1", %{"count" => 0}] = msg + end + + test "subscribe to unknown route returns error", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["nonexistent", "route"])) + msg = ws_receive(ws) + + # [0, channel_id, error] = error + assert [0, "ch1", "not_found"] = msg + end + + test "multiple subscribes to different topics get independent resets", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "multi-a"])) + ws_send(ws, subscribe_msg("ch2", ["counters", "multi-b"])) + + msgs = ws_receive_all(ws, 2) + channel_ids = Enum.map(msgs, fn [_, ch_id | _] -> ch_id end) + + assert "ch1" in channel_ids + assert "ch2" in channel_ids + + Enum.each(msgs, fn msg -> + assert [2, _, %{"count" => 0}] = msg + end) + end + end + + describe "execute & result" do + test "execute returns result value", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "exec-1"])) + _reset = ws_receive(ws) + + ws_send(ws, execute_msg("req1", ["counters", "exec-1"], "increment", [])) + msg = ws_receive(ws) + + # [1, channel_id, result] = result + assert [1, "req1", 1] = msg + end + + test "execute broadcasts update to subscriber", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "exec-2"])) + _reset = ws_receive(ws) + + ws_send(ws, execute_msg("req1", ["counters", "exec-2"], "increment", [])) + + # We should get both a result and an update + msgs = ws_receive_all(ws, 2) + types = Enum.map(msgs, fn [type | _] -> type end) + + # result + assert 1 in types + # topic_updates + assert 3 in types + + update_msg = Enum.find(msgs, fn [type | _] -> type == 3 end) + assert [3, "ch1", [[0, ["count"], 1]]] = update_msg + end + + test "execute on unknown topic returns error", %{ws: ws} do + ws_send(ws, execute_msg("req1", ["nonexistent", "topic"], "action", [])) + msg = ws_receive(ws) + + assert [0, "req1", "not_found"] = msg + end + + test "execute with arguments works", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "exec-args"])) + _reset = ws_receive(ws) + + ws_send(ws, execute_msg("req1", ["counters", "exec-args"], "set", [42])) + + msgs = ws_receive_all(ws, 2) + result_msg = Enum.find(msgs, fn [type | _] -> type == 1 end) + assert [1, "req1", 42] = result_msg + end + end + + describe "notify & updates" do + test "notify triggers update to subscriber", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "notify-1"])) + _reset = ws_receive(ws) + + ws_send(ws, notify_msg(["counters", "notify-1"], "increment", [])) + + msg = ws_receive(ws) + assert [3, "ch1", [[0, ["count"], 1]]] = msg + end + + test "notify does not produce a result response", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "notify-2"])) + _reset = ws_receive(ws) + + ws_send(ws, notify_msg(["counters", "notify-2"], "increment", [])) + + msg = ws_receive(ws) + # Should only get update (type 3), not result (type 1) + assert [3, "ch1", _updates] = msg + ws_refute_receive(ws) + end + + test "notify with arguments works", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "notify-args"])) + _reset = ws_receive(ws) + + ws_send(ws, notify_msg(["counters", "notify-args"], "set", [99])) + + msg = ws_receive(ws) + assert [3, "ch1", [[0, ["count"], 99]]] = msg + end + end + + describe "unsubscribe" do + test "unsubscribe stops updates", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "unsub-1"])) + _reset = ws_receive(ws) + + ws_send(ws, unsubscribe_msg("ch1")) + # Give time for unsubscribe to process + Process.sleep(50) + + ws_send(ws, notify_msg(["counters", "unsub-1"], "increment", [])) + ws_refute_receive(ws) + end + + test "unsubscribe unknown channel is a no-op", %{ws: ws} do + ws_send(ws, unsubscribe_msg("nonexistent")) + # Should not crash the connection + ws_send(ws, subscribe_msg("ch1", ["counters", "unsub-noop"])) + msg = ws_receive(ws) + assert [2, "ch1", %{"count" => 0}] = msg + end + + test "can resubscribe after unsubscribe", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "unsub-resub"])) + _reset = ws_receive(ws) + + # Increment before unsubscribe + ws_send(ws, execute_msg("req1", ["counters", "unsub-resub"], "increment", [])) + _msgs = ws_receive_all(ws, 2) + + ws_send(ws, unsubscribe_msg("ch1")) + Process.sleep(50) + + # Resubscribe on new channel + ws_send(ws, subscribe_msg("ch2", ["counters", "unsub-resub"])) + msg = ws_receive(ws) + + # Should get reset with current value (count=1) + assert [2, "ch2", %{"count" => 1}] = msg + end + end + + describe "multiple connections" do + test "two connections both receive updates for shared topic", %{ws: ws, port: port} do + ws2 = ws_connect(port) + + ws_send(ws, subscribe_msg("ch1", ["counters", "multi-conn-1"])) + _reset1 = ws_receive(ws) + + ws_send(ws2, subscribe_msg("ch1", ["counters", "multi-conn-1"])) + _reset2 = ws_receive(ws2) + + ws_send(ws, execute_msg("req1", ["counters", "multi-conn-1"], "increment", [])) + + # ws gets both result and update + msgs1 = ws_receive_all(ws, 2) + assert Enum.any?(msgs1, fn [type | _] -> type == 3 end) + + # ws2 gets update only + msg2 = ws_receive(ws2) + assert [3, "ch1", [[0, ["count"], 1]]] = msg2 + + ws_close(ws2) + end + + test "one unsubscribing doesn't affect the other", %{ws: ws, port: port} do + ws2 = ws_connect(port) + + ws_send(ws, subscribe_msg("ch1", ["counters", "multi-conn-2"])) + _reset1 = ws_receive(ws) + + ws_send(ws2, subscribe_msg("ch1", ["counters", "multi-conn-2"])) + _reset2 = ws_receive(ws2) + + ws_send(ws, unsubscribe_msg("ch1")) + Process.sleep(50) + + ws_send(ws2, execute_msg("req1", ["counters", "multi-conn-2"], "increment", [])) + + # ws should NOT get updates + ws_refute_receive(ws) + + # ws2 gets result + update + msgs = ws_receive_all(ws2, 2) + assert Enum.any?(msgs, fn [type | _] -> type == 1 end) + assert Enum.any?(msgs, fn [type | _] -> type == 3 end) + + ws_close(ws2) + end + end + + describe "authorization" do + test "subscribe without required context returns error", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["private", "user1"])) + msg = ws_receive(ws) + + assert [0, "ch1", "unauthorized"] = msg + end + + test "subscribe with correct context succeeds", %{port: _port} do + %{port: port} = start_server(@adapter, init: fn _req -> {:ok, %{user_id: "user1"}} end) + ws = ws_connect(port) + + ws_send(ws, subscribe_msg("ch1", ["private", "user1"])) + msg = ws_receive(ws) + + assert [2, "ch1", %{"owner" => "user1", "data" => nil}] = msg + ws_close(ws) + end + end + + describe "params" do + test "subscribe with explicit params", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["leaderboards", "chess"], %{"region" => "eu"})) + msg = ws_receive(ws) + + assert [2, "ch1", %{"game_id" => "chess", "region" => "eu", "entries" => []}] = msg + end + + test "subscribe with default params", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["leaderboards", "chess"])) + msg = ws_receive(ws) + + assert [2, "ch1", %{"game_id" => "chess", "region" => "global", "entries" => []}] = msg + end + end + + describe "aliases" do + test "subscribing to same topic twice returns alias response", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "alias-1"])) + _reset = ws_receive(ws) + + ws_send(ws, subscribe_msg("ch2", ["counters", "alias-1"])) + msg = ws_receive(ws) + + # [4, channel_id, existing_channel_id] = topic_alias + assert [4, "ch2", "ch1"] = msg + end + + test "after unsubscribe + resubscribe, no alias (fresh subscription)", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["counters", "alias-2"])) + _reset = ws_receive(ws) + + ws_send(ws, unsubscribe_msg("ch1")) + Process.sleep(50) + + ws_send(ws, subscribe_msg("ch2", ["counters", "alias-2"])) + msg = ws_receive(ws) + + # Should be a reset, not an alias + assert [2, "ch2", %{"count" => 0}] = msg + end + end + + describe "complex updates" do + test "list insert operations", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["lists", "complex-1"])) + _reset = ws_receive(ws) + + ws_send(ws, execute_msg("req1", ["lists", "complex-1"], "add", ["first"])) + msgs = ws_receive_all(ws, 2) + + update_msg = Enum.find(msgs, fn [type | _] -> type == 3 end) + [3, "ch1", updates] = update_msg + + # Should contain an insert operation (opcode 2) + assert Enum.any?(updates, fn [opcode | _] -> opcode == 2 end) + end + + test "merge operations", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["merge", "complex-1"])) + _reset = ws_receive(ws) + + ws_send(ws, execute_msg("req1", ["merge", "complex-1"], "merge", [%{"a" => 1, "b" => 2}])) + msgs = ws_receive_all(ws, 2) + + update_msg = Enum.find(msgs, fn [type | _] -> type == 3 end) + [3, "ch1", updates] = update_msg + + # Should contain a merge operation (opcode 4) + assert Enum.any?(updates, fn [opcode | _] -> opcode == 4 end) + end + + test "unset operations", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["merge", "complex-2"])) + _reset = ws_receive(ws) + + # First set a key + ws_send(ws, execute_msg("req1", ["merge", "complex-2"], "set", ["mykey", "myvalue"])) + _msgs = ws_receive_all(ws, 2) + + # Now unset it + ws_send(ws, execute_msg("req2", ["merge", "complex-2"], "unset", ["mykey"])) + msgs = ws_receive_all(ws, 2) + + update_msg = Enum.find(msgs, fn [type | _] -> type == 3 end) + [3, "ch1", updates] = update_msg + + # Should contain an unset operation (opcode 1) + assert Enum.any?(updates, fn [opcode | _] -> opcode == 1 end) + end + end + + describe "error handling" do + test "topic with init error returns error", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["failing", "init_error"])) + msg = ws_receive(ws) + + assert [0, "ch1", _error] = msg + end + + test "connection survives after error response", %{ws: ws} do + ws_send(ws, subscribe_msg("ch1", ["nonexistent", "route"])) + msg = ws_receive(ws) + assert [0, "ch1", "not_found"] = msg + + # Connection should still work + ws_send(ws, subscribe_msg("ch2", ["counters", "survive-1"])) + msg = ws_receive(ws) + assert [2, "ch2", %{"count" => 0}] = msg + end + end + + describe "capture (HTTP)" do + test "returns current topic value", %{port: port} do + # First set a value via WebSocket + ws = ws_connect(port) + ws_send(ws, subscribe_msg("ch1", ["counters", "capture-1"])) + _reset = ws_receive(ws) + ws_send(ws, execute_msg("req1", ["counters", "capture-1"], "set", [42])) + _msgs = ws_receive_all(ws, 2) + ws_close(ws) + + {status, body} = http_get(port, "/topics/counters/capture-1") + assert status == 200 + assert body == %{"count" => 42} + end + + test "does not subscribe", %{ws: ws, port: port} do + # Capture a topic + {200, _body} = http_get(port, "/topics/counters/capture-nosub") + + # Mutate the topic + ws_send(ws, subscribe_msg("ch1", ["counters", "capture-nosub"])) + _reset = ws_receive(ws) + ws_send(ws, execute_msg("req1", ["counters", "capture-nosub"], "increment", [])) + _msgs = ws_receive_all(ws, 2) + + # Capture again — should see the updated value (proves it's a snapshot, not a subscription) + {200, body} = http_get(port, "/topics/counters/capture-nosub") + assert body == %{"count" => 1} + end + + test "returns 404 for unknown topic", %{port: port} do + {status, body} = http_get(port, "/topics/nonexistent/route") + assert status == 404 + assert body == %{"error" => "not_found"} + end + + test "returns error for unauthorized topic", %{port: port} do + {status, _body} = http_get(port, "/topics/private/user1") + assert status in [400, 403] + end + end + end +end