Dstar.Page behaviour (dstar v0.3.0)

Copy Markdown View Source

One module per page: render, event handlers, stream callbacks, and colocated function components.

defmodule MyAppWeb.CounterPage do
  use Dstar.Page

  def mount(conn, _params), do: assign(conn, count: 0)

  def render(assigns) do
    ~H"""
    <div data-signals:count={@count}>
      <button data-on:click={event("increment")}>+1</button>
    </div>
    """
  end

  def handle_event(conn, "increment", signals) do
    patch_signals(conn, %{count: (signals["count"] || 0) + 1})
  end
end

Route it with Dstar.Router.dstar/2:

import Dstar.Router
dstar "/counter", MyAppWeb.CounterPage

All requests are driven by Dstar.Page.Plug — pages contain no control flow, only callbacks. Conn in, conn out.

Authorization

dstar/2 registers three independent routes. Session-wide authentication belongs in the surrounding router pipeline — it already covers GET, event POST, stream POST, and dstar_components/2.

mount/2 runs only on the GET. Putting a tenant or ownership check there does not protect a direct POST /path/_event/:event or POST /path. Those skip mount/2 and would otherwise start SSE (status 200, text/event-stream) before handle_event/3 or handle_connect/2 can return a normal 401/403.

Use the optional authorize/2 callback for page-local checks. It runs after signals are read and before Dstar.SSE.start/1 / start_stream/2. Halt or send a response to reject; return the conn to continue. Assigns and private data you set are visible to later callbacks.

Route and component-module allowlists select code. They do not authorize the current user to touch a particular record. Interpolating an id into the event name ("change_title:" <> id) is fine; the handler or authorize/2 must still prove the user may touch id.

Stream deduplication

Define stream_key/1 to opt a Page stream into Dstar.Utility.StreamRegistry. The coordinator atomically claims {stream_key, tabId} before SSE and handle_connect/2. A valid keyed claim failure returns a normal halted 503 and neither callback nor loop runs; missing/invalid tabId intentionally starts unkeyed.

Takeovers carry an ownership generation. The loop ignores stale generations left in a reused keep-alive process and synchronously releases the exact current generation before handle_disconnect/1, closing the stale release-versus-kill race.

Options

  • :idle_check — ms between connection liveness checks in the stream loop (default 30_000).
  • :max_signal_bytes — maximum raw signal JSON size for event and stream requests (default 1_000_000). Configure Plug.Parsers to the same or a smaller limit when it reads bodies before the page plug.

Summary

Callbacks

Pre-SSE gate for event and stream POSTs. Optional.

Stream open: subscribe to topics, assign loop state. Optional — defining it enables POST /path streaming.

Stream close: release what handle_connect/2 acquired — unsubscribe, untrack presence, drop caches. Optional.

Handles a Datastar event POST. SSE is already started. Optional.

Handles one message from the library-owned receive loop. Optional.

GET: load data and assign what render/1 needs. Optional.

The full-page HEEx template. Required.

If defined, the stream atomically claims {result, tabId} through Dstar.start_stream/2 before handle_connect/2. Optional.

Callbacks

authorize(t, arg2)

(optional)
@callback authorize(
  Plug.Conn.t(),
  {:event, event :: String.t()} | {:stream, params :: map()}
) ::
  Plug.Conn.t()

Pre-SSE gate for event and stream POSTs. Optional.

Invoked after query params and signals are safely read, and before Dstar.SSE.start/1 or Dstar.start_stream/2. The second argument is {:event, event} or {:stream, params}.

Same skip rule as mount/2: if the returned conn is halted or has already staged/sent a response (state != :unset), the plug returns it as a normal HTTP response and does not start SSE, call handle_event/3 / handle_connect/2, or register a stream_key/1.

def authorize(conn, {:event, "export"}) do
  if admin?(conn) do
    conn
  else
    conn
    |> Plug.Conn.send_resp(403, "Forbidden")
    |> Plug.Conn.halt()
  end
end

def authorize(conn, {:event, _event}), do: conn

def authorize(conn, {:stream, _params}) do
  if conn.assigns[:current_user] do
    conn
  else
    conn
    |> Plug.Conn.send_resp(401, "Unauthorized")
    |> Plug.Conn.halt()
  end
end

Session-wide authentication still belongs in the router pipeline. This callback is for page/resource checks and preconditions. Signals are already parsed and cached in body_params; read them with Dstar.Signals.read/1 if a check needs the payload. Invalid input was already returned as a normal 400/413 response.

Does not run on GET — that is mount/2.

handle_connect(t, params)

(optional)
@callback handle_connect(Plug.Conn.t(), params :: map()) :: Plug.Conn.t()

Stream open: subscribe to topics, assign loop state. Optional — defining it enables POST /path streaming.

handle_disconnect(t)

(optional)
@callback handle_disconnect(Plug.Conn.t()) :: any()

Stream close: release what handle_connect/2 acquired — unsubscribe, untrack presence, drop caches. Optional.

Runs on every exit from the receive loop: a {:halt, conn} from handle_info/2, a dead client, or a takeover by a newer stream for the same stream_key/1. The library synchronously releases the exact claim generation before invoking this callback; this callback is for app-owned state.

The client is usually gone by this point, so treat the conn as write-only-on-a-best-effort basis. A crash here is logged and swallowed — teardown continues either way. The return value is ignored.

Best-effort, not guaranteed

This runs when the loop ends, which requires the process to survive long enough to notice. A takeover that kills the process outright skips it: only Thousand Island (so Bandit HTTP/1.1) connection processes trap exits, whereas Bandit's HTTP/2 stream processes and Cowboy request processes do not, and the registry's escalation kill is untrappable by design.

Anything cleaned up by process monitors — Phoenix.PubSub subscriptions, Phoenix.Presence, the registry entry itself — is released regardless. Reserve this callback for state that is not monitor-backed (external caches, rows in another system), and make it idempotent: on a takeover it can interleave with, or run after, the replacing stream's handle_connect/2, so cleanup keyed on the tab rather than on this stream can otherwise clobber what the new stream just wrote.

handle_event(t, event, signals)

(optional)
@callback handle_event(Plug.Conn.t(), event :: String.t(), signals :: map()) ::
  Plug.Conn.t()

Handles a Datastar event POST. SSE is already started. Optional.

handle_info(msg, t)

(optional)
@callback handle_info(msg :: term(), Plug.Conn.t()) ::
  Plug.Conn.t() | {:halt, Plug.Conn.t()}

Handles one message from the library-owned receive loop. Optional.

mount(t, params)

(optional)
@callback mount(Plug.Conn.t(), params :: map()) :: Plug.Conn.t()

GET: load data and assign what render/1 needs. Optional.

render(assigns)

@callback render(assigns :: map()) :: Phoenix.LiveView.Rendered.t()

The full-page HEEx template. Required.

stream_key(t)

(optional)
@callback stream_key(Plug.Conn.t()) :: term()

If defined, the stream atomically claims {result, tabId} through Dstar.start_stream/2 before handle_connect/2. Optional.

A valid keyed claim failure returns a halted non-SSE 503 and skips the connect callback/loop. Missing or invalid tabId is the intentional unkeyed fallback.