lee/wanderer
1
0
Fork 0
mirror of https://github.com/wanderer-industries/wanderer synced 2025-12-12 18:56:01 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

fix: sse cleanup

Browse Source
This commit is contained in:
guarzo
2025-07-01 02:32:04 -04:00
parent 4d75b256c4
commit 00cbc77f1d
8 changed files with 173 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
config/runtime.exs
View File

@@ -393,15 +393,15 @@ config :wanderer_app, :license_manager,
# SSE Configuration # SSE Configuration
config :wanderer_app, :sse, config :wanderer_app, :sse,
max_connections_per_map: String.to_integer(System.get_env("SSE_MAX_CONNECTIONS_PER_MAP", "50")), enabled: System.get_env("WANDERER_SSE_ENABLED", "true") == "true",
max_connections_total: config_dir |> get_int_from_path_or_env("WANDERER_SSE_MAX_CONNECTIONS", 1000),
max_connections_per_map: config_dir |> get_int_from_path_or_env("SSE_MAX_CONNECTIONS_PER_MAP", 50),
max_connections_per_api_key: max_connections_per_api_key:
String.to_integer(System.get_env("SSE_MAX_CONNECTIONS_PER_API_KEY", "10")), config_dir |> get_int_from_path_or_env("SSE_MAX_CONNECTIONS_PER_API_KEY", 10),
keepalive_interval: String.to_integer(System.get_env("SSE_KEEPALIVE_INTERVAL", "30000")), keepalive_interval: config_dir |> get_int_from_path_or_env("SSE_KEEPALIVE_INTERVAL", 30000),
connection_timeout: String.to_integer(System.get_env("SSE_CONNECTION_TIMEOUT", "300000")) connection_timeout: config_dir |> get_int_from_path_or_env("SSE_CONNECTION_TIMEOUT", 300000)
# External Events Configuration # External Events Configuration
config :wanderer_app, :external_events, config :wanderer_app, :external_events,
sse_enabled: System.get_env("WANDERER_SSE_ENABLED", "true") == "true",
webhooks_enabled: System.get_env("WANDERER_WEBHOOKS_ENABLED", "true") == "true", webhooks_enabled: System.get_env("WANDERER_WEBHOOKS_ENABLED", "true") == "true",
sse_max_connections: String.to_integer(System.get_env("WANDERER_SSE_MAX_CONNECTIONS", "1000")), webhook_timeout_ms: config_dir |> get_int_from_path_or_env("WANDERER_WEBHOOK_TIMEOUT_MS", 15000)
webhook_timeout_ms: String.to_integer(System.get_env("WANDERER_WEBHOOK_TIMEOUT_MS", "15000"))

49
lib/wanderer_app/external_events/map_event_relay.ex
View File

@@ -96,21 +96,27 @@ defmodule WandererApp.ExternalEvents.MapEventRelay do
@impl true @impl true
def handle_call({:get_events_since_ulid, map_id, since_ulid, limit}, _from, state) do def handle_call({:get_events_since_ulid, map_id, since_ulid, limit}, _from, state) do
# Get all events for this map and filter by ULID # Get all events for this map and filter by ULID
try do case validate_ulid(since_ulid) do
# Events are stored as {event_id, map_id, json_data} :ok ->
# Filter by map_id and event_id (ULID) > since_ulid try do
events = # Events are stored as {event_id, map_id, json_data}
:ets.select(state.ets_table, [ # Filter by map_id and event_id (ULID) > since_ulid
{{:"$1", :"$2", :"$3"}, events =
[{:andalso, {:>, :"$1", since_ulid}, {:==, :"$2", map_id}}], :ets.select(state.ets_table, [
[:"$3"]} {{:"$1", :"$2", :"$3"},
]) [{:andalso, {:>, :"$1", since_ulid}, {:==, :"$2", map_id}}],
|> Enum.take(limit) [:"$3"]}
])
|> Enum.take(limit)
{:reply, {:ok, events}, state} {:reply, {:ok, events}, state}
catch rescue
_, reason -> error in [ArgumentError] ->
{:reply, {:error, reason}, state} {:reply, {:error, {:ets_error, error}}, state}
end
{:error, :invalid_ulid} ->
{:reply, {:error, :invalid_ulid}, state}
end end
end end
@@ -189,6 +195,21 @@ defmodule WandererApp.ExternalEvents.MapEventRelay do
end end
end end
defp validate_ulid(ulid) when is_binary(ulid) do
# ULID format validation: 26 characters, [0-9A-Z] excluding I, L, O, U
case byte_size(ulid) do
26 ->
if ulid =~ ~r/^[0123456789ABCDEFGHJKMNPQRSTVWXYZ]{26}$/ do
:ok
else
{:error, :invalid_ulid}
end
_ ->
{:error, :invalid_ulid}
end
end
defp validate_ulid(_), do: {:error, :invalid_ulid}
defp cleanup_old_events(ets_table) do defp cleanup_old_events(ets_table) do
cutoff_time = DateTime.add(DateTime.utc_now(), -@event_retention_minutes, :minute) cutoff_time = DateTime.add(DateTime.utc_now(), -@event_retention_minutes, :minute)
cutoff_ulid = datetime_to_ulid(cutoff_time) cutoff_ulid = datetime_to_ulid(cutoff_time)

4
lib/wanderer_app/external_events/sse_stream_manager.ex
View File

@@ -70,11 +70,11 @@ defmodule WandererApp.ExternalEvents.SseStreamManager do
@impl true @impl true
def handle_call({:add_client, map_id, api_key, client_pid, event_filter}, _from, state) do def handle_call({:add_client, map_id, api_key, client_pid, event_filter}, _from, state) do
# Check if feature is enabled # Check if feature is enabled
unless Application.get_env(:wanderer_app, :external_events, [])[:sse_enabled] do unless Application.get_env(:wanderer_app, :sse, [])[:enabled] do
{:reply, {:error, :sse_disabled}, state} {:reply, {:error, :sse_disabled}, state}
else else
# Check connection limits # Check connection limits
max_connections = Application.get_env(:wanderer_app, :external_events, [])[:sse_max_connections] || 1000 max_connections = Application.get_env(:wanderer_app, :sse, [])[:max_connections_total] || 1000
case check_connection_limits(state, map_id, api_key, max_connections) do case check_connection_limits(state, map_id, api_key, max_connections) do
:ok -> :ok ->

10
lib/wanderer_app/external_events/webhook_dispatcher.ex
View File

@@ -143,7 +143,7 @@ defmodule WandererApp.ExternalEvents.WebhookDispatcher do
{:ok, subscriptions} {:ok, subscriptions}
rescue rescue
# Catch specific Ash errors # Catch specific Ash errors
error in [Ash.Error.Query.NotFound] -> _error in [Ash.Error.Query.NotFound] ->
{:ok, []} {:ok, []}
error in [Ash.Error.Invalid] -> error in [Ash.Error.Invalid] ->
@@ -365,9 +365,11 @@ defmodule WandererApp.ExternalEvents.WebhookDispatcher do
true <- map.webhooks_enabled do true <- map.webhooks_enabled do
:ok :ok
else else
false -> {:error, :webhooks_disabled} false -> {:error, :webhooks_globally_disabled}
{:error, :not_found} -> {:error, :webhooks_disabled} {:error, :not_found} -> {:error, :map_not_found}
_ -> {:error, :webhooks_disabled} %{webhooks_enabled: false} -> {:error, :webhooks_disabled_for_map}
{:error, reason} -> {:error, reason}
error -> {:error, {:unexpected_error, error}}
end end
end end
end end

150
lib/wanderer_app_web/controllers/api/events_controller.ex
View File

@@ -7,7 +7,7 @@ defmodule WandererAppWeb.Api.EventsController do
use WandererAppWeb, :controller use WandererAppWeb, :controller
alias WandererApp.ExternalEvents.{SseConnectionTracker, EventFilter, MapEventRelay} alias WandererApp.ExternalEvents.{SseStreamManager, EventFilter, MapEventRelay}
alias WandererApp.Api.Map, as: ApiMap alias WandererApp.Api.Map, as: ApiMap
alias WandererAppWeb.SSE alias WandererAppWeb.SSE
alias Plug.Crypto alias Plug.Crypto
@@ -25,36 +25,17 @@ defmodule WandererAppWeb.Api.EventsController do
Logger.info("SSE stream requested for map #{map_identifier}") Logger.info("SSE stream requested for map #{map_identifier}")
# Check if SSE is enabled # Check if SSE is enabled
unless Application.get_env(:wanderer_app, :external_events, [])[:sse_enabled] do unless Application.get_env(:wanderer_app, :sse, [])[:enabled] do
conn conn
|> put_status(:service_unavailable) |> put_status(:service_unavailable)
|> json(%{error: "Server-Sent Events are disabled on this server"}) |> put_resp_content_type("text/plain")
|> send_resp(503, "Server-Sent Events are disabled on this server")
else else
# Validate API key and get map # Validate API key and get map
case validate_api_key(conn, map_identifier) do case validate_api_key(conn, map_identifier) do
{:ok, map, api_key} -> {:ok, map, api_key} ->
# Check connection limits establish_sse_connection(conn, map.id, api_key, params)
case SseConnectionTracker.check_limits(map.id, api_key) do
:ok ->
establish_sse_connection(conn, map.id, api_key, params)
{:error, :map_limit_exceeded} ->
conn
|> put_status(:too_many_requests)
|> json(%{
error: "Too many connections to this map",
code: "MAP_CONNECTION_LIMIT"
})
{:error, :api_key_limit_exceeded} ->
conn
|> put_status(:too_many_requests)
|> json(%{
error: "Too many connections for this API key",
code: "API_KEY_CONNECTION_LIMIT"
})
end
{:error, status, message} -> {:error, status, message} ->
conn conn
@@ -76,33 +57,56 @@ defmodule WandererAppWeb.Api.EventsController do
conn = SSE.send_headers(conn) conn = SSE.send_headers(conn)
# Track the connection # Track the connection
:ok = SseConnectionTracker.track_connection(map_id, api_key, self()) case SseStreamManager.add_client(map_id, api_key, self(), event_filter) do
{:ok, _} ->
# Send initial connection event # Send initial connection event
conn = SSE.send_event(conn, %{ conn = SSE.send_event(conn, %{
id: Ulid.generate(), id: Ulid.generate(),
event: "connected", event: "connected",
data: %{ data: %{
map_id: map_id, map_id: map_id,
server_time: DateTime.utc_now() |> DateTime.to_iso8601() server_time: DateTime.utc_now() |> DateTime.to_iso8601()
} }
}) })
# Handle backfill if last_event_id is provided # Handle backfill if last_event_id is provided
conn = conn =
case Map.get(params, "last_event_id") do case Map.get(params, "last_event_id") do
nil -> nil ->
conn conn
last_event_id -> last_event_id ->
send_backfill_events(conn, map_id, last_event_id, event_filter) send_backfill_events(conn, map_id, last_event_id, event_filter)
end end
# Subscribe to map events # Subscribe to map events
Phoenix.PubSub.subscribe(WandererApp.PubSub, "external_events:map:#{map_id}") Phoenix.PubSub.subscribe(WandererApp.PubSub, "external_events:map:#{map_id}")
# Start streaming loop # Start streaming loop
stream_events(conn, map_id, api_key, event_filter) stream_events(conn, map_id, api_key, event_filter)
{:error, :map_limit_exceeded} ->
conn
|> put_status(:too_many_requests)
|> json(%{
error: "Too many connections to this map",
code: "MAP_CONNECTION_LIMIT"
})
{:error, :api_key_limit_exceeded} ->
conn
|> put_status(:too_many_requests)
|> json(%{
error: "Too many connections for this API key",
code: "API_KEY_CONNECTION_LIMIT"
})
{:error, reason} ->
Logger.error("Failed to add SSE client: #{inspect(reason)}")
conn
|> put_status(:internal_server_error)
|> send_resp(500, "Internal server error")
end
end end
defp send_backfill_events(conn, map_id, last_event_id, event_filter) do defp send_backfill_events(conn, map_id, last_event_id, event_filter) do
@@ -110,12 +114,17 @@ defmodule WandererAppWeb.Api.EventsController do
{:ok, events} -> {:ok, events} ->
# Filter and send each event # Filter and send each event
Enum.reduce(events, conn, fn event_json, acc_conn -> Enum.reduce(events, conn, fn event_json, acc_conn ->
event = Jason.decode!(event_json) case Jason.decode(event_json) do
{:ok, event} ->
if EventFilter.matches?(event["type"], event_filter) do if EventFilter.matches?(event["type"], event_filter) do
SSE.send_event(acc_conn, event) SSE.send_event(acc_conn, event)
else else
acc_conn acc_conn
end
{:error, reason} ->
Logger.error("Failed to decode event during backfill: #{inspect(reason)}")
acc_conn
end end
end) end)
@@ -129,13 +138,18 @@ defmodule WandererAppWeb.Api.EventsController do
receive do receive do
{:external_event, event_json} -> {:external_event, event_json} ->
# Parse and check if event matches filter # Parse and check if event matches filter
event = Jason.decode!(event_json)
conn = conn =
if EventFilter.matches?(event["type"], event_filter) do case Jason.decode(event_json) do
SSE.send_event(conn, event) {:ok, event} ->
else if EventFilter.matches?(event["type"], event_filter) do
conn SSE.send_event(conn, event)
else
conn
end
{:error, reason} ->
Logger.error("Failed to decode event in stream: #{inspect(reason)}")
conn
end end
# Continue streaming # Continue streaming
@@ -159,11 +173,17 @@ defmodule WandererAppWeb.Api.EventsController do
stream_events(conn, map_id, api_key, event_filter) stream_events(conn, map_id, api_key, event_filter)
end end
rescue rescue
_ -> _error in [Plug.Conn.WrapperError, DBConnection.ConnectionError] ->
# Connection closed, cleanup # Connection closed, cleanup
Logger.info("SSE connection closed for map #{map_id}") Logger.info("SSE connection closed for map #{map_id}")
SseConnectionTracker.remove_connection(map_id, api_key, self()) SseStreamManager.remove_client(map_id, api_key, self())
conn conn
error ->
# Log unexpected errors before cleanup
Logger.error("Unexpected error in SSE stream: #{inspect(error)}")
SseStreamManager.remove_client(map_id, api_key, self())
reraise error, __STACKTRACE__
end end
defp validate_api_key(conn, map_identifier) do defp validate_api_key(conn, map_identifier) do

15
lib/wanderer_app_web/controllers/map_api_controller.ex
View File

@@ -875,12 +875,18 @@ defmodule WandererAppWeb.MapAPIController do
} }
def toggle_webhooks(conn, %{"map_id" => map_identifier, "enabled" => enabled}) do def toggle_webhooks(conn, %{"map_id" => map_identifier, "enabled" => enabled}) do
with :ok <- check_global_webhooks_enabled(), with {:ok, enabled_boolean} <- validate_boolean_param(enabled, "enabled"),
:ok <- check_global_webhooks_enabled(),
{:ok, map} <- resolve_map_identifier(map_identifier), {:ok, map} <- resolve_map_identifier(map_identifier),
:ok <- check_map_owner(conn, map), :ok <- check_map_owner(conn, map),
{:ok, updated_map} <- WandererApp.Api.Map.toggle_webhooks(map, %{webhooks_enabled: enabled}) do {:ok, updated_map} <- WandererApp.Api.Map.toggle_webhooks(map, %{webhooks_enabled: enabled_boolean}) do
json(conn, %{webhooks_enabled: updated_map.webhooks_enabled}) json(conn, %{webhooks_enabled: updated_map.webhooks_enabled})
else else
{:error, :invalid_boolean} ->
conn
|> put_status(:bad_request)
|> json(%{error: "The 'enabled' parameter must be a boolean value"})
{:error, :webhooks_disabled} -> {:error, :webhooks_disabled} ->
conn conn
|> put_status(:service_unavailable) |> put_status(:service_unavailable)
@@ -905,6 +911,11 @@ defmodule WandererAppWeb.MapAPIController do
# Helper functions for webhook toggle # Helper functions for webhook toggle
defp validate_boolean_param(value, _param_name) when is_boolean(value), do: {:ok, value}
defp validate_boolean_param("true", _param_name), do: {:ok, true}
defp validate_boolean_param("false", _param_name), do: {:ok, false}
defp validate_boolean_param(_, _param_name), do: {:error, :invalid_boolean}
defp check_global_webhooks_enabled do defp check_global_webhooks_enabled do
if Application.get_env(:wanderer_app, :external_events)[:webhooks_enabled] do if Application.get_env(:wanderer_app, :external_events)[:webhooks_enabled] do
:ok :ok

12
lib/wanderer_app_web/controllers/map_webhooks_api_controller.ex
View File

@@ -341,13 +341,18 @@ defmodule WandererAppWeb.MapWebhooksAPIController do
def create(conn, %{"map_identifier" => map_identifier} = params) do def create(conn, %{"map_identifier" => map_identifier} = params) do
# Check if webhooks are enabled # Check if webhooks are enabled
unless Application.get_env(:wanderer_app, :external_events, [])[:webhooks_enabled] do if not Application.get_env(:wanderer_app, :external_events, [])[:webhooks_enabled] do
conn conn
|> put_status(:service_unavailable) |> put_status(:service_unavailable)
|> json(%{error: "Webhooks are disabled on this server"}) |> json(%{error: "Webhooks are disabled on this server"})
else else
with {:ok, map} <- get_map(conn, map_identifier), do_create_webhook(conn, map_identifier, params)
{:ok, webhook_params} <- validate_create_params(params, map.id) do end
end
defp do_create_webhook(conn, map_identifier, params) do
with {:ok, map} <- get_map(conn, map_identifier),
{:ok, webhook_params} <- validate_create_params(params, map.id) do
case MapWebhookSubscription.create(webhook_params) do case MapWebhookSubscription.create(webhook_params) do
{:ok, webhook} -> {:ok, webhook} ->
@@ -384,7 +389,6 @@ defmodule WandererAppWeb.MapWebhooksAPIController do
|> put_status(:internal_server_error) |> put_status(:internal_server_error)
|> json(%{error: "Internal server error"}) |> json(%{error: "Internal server error"})
end end
end
end end
def update(conn, %{"map_identifier" => map_identifier, "id" => webhook_id} = params) do def update(conn, %{"map_identifier" => map_identifier, "id" => webhook_id} = params) do

29
priv/posts/2025/06-21-webhooks.md
View File

@@ -56,12 +56,20 @@ const apiToken = "your-map-api-token";
// Optional: Filter specific events // Optional: Filter specific events
const eventTypes = ["add_system", "map_kill"].join(","); const eventTypes = ["add_system", "map_kill"].join(",");
const url = `https://wanderer.ltd/api/maps/${mapId}/events/stream?events=${eventTypes}`;
const eventSource = new EventSource(url, { // Note: Native EventSource doesn't support custom headers
headers: { // You have two options:
'Authorization': `Bearer ${apiToken}`
} // Option 1: Include the API token as a query parameter
const url = `https://wanderer.ltd/api/maps/${mapId}/events/stream?events=${eventTypes}&token=${apiToken}`;
const eventSource = new EventSource(url);
// Option 2: Use an EventSource polyfill that supports headers
import { EventSourcePolyfill } from 'event-source-polyfill';
const eventSource = new EventSourcePolyfill(url, {
headers: {
'Authorization': `Bearer ${apiToken}`
}
}); });
// Handle connection opened // Handle connection opened
@@ -107,16 +115,13 @@ const url = `https://wanderer.ltd/api/maps/${mapId}/events/stream`;
``` ```
### Event Backfill ### Event Backfill
SSE supports automatic backfill using the `Last-Event-ID` header: SSE supports automatic backfill when reconnecting:
```javascript ```javascript
// Reconnect with backfill from last received event // Reconnect with backfill from last received event
const eventSource = new EventSource(url, { // Add the last_event_id as a query parameter
headers: { const url = `https://wanderer.ltd/api/maps/${mapId}/events/stream?token=${apiToken}&last_event_id=${lastEventId}`;
'Authorization': `Bearer ${apiToken}`, const eventSource = new EventSource(url);
'Last-Event-ID': lastEventId // Will receive events since this ID
}
});
``` ```
## Webhook Setup ## Webhook Setup