Defines a reducer module for managing state slices.

Usage

defmodule MyApp.Reducers.UserReducer do
  use Phoenix.SessionProcess, :reducer
  alias Phoenix.SessionProcess.Action

  @name :user
  @action_prefix "user"

  def init_state do
    %{users: [], loading: false, query: nil}
  end

  @throttle {"fetch-list", "3000ms"}
  def handle_action(%Action{type: "fetch-list"}, state) do
    # Throttled: Only executes once per 3 seconds
    %{state | loading: true}
  end

  @debounce {"update-query", "500ms"}
  def handle_action(%Action{type: "update-query", payload: query}, state) do
    # Debounced: Waits 500ms after last call
    %{state | query: query}
  end

  def handle_async(%Action{type: "load", payload: %{page: page}}, dispatch, _state) do
    # Async action with dispatch callback
    # dispatch signature: dispatch(type, payload \ nil, meta \ [])
    # Must return cancellation function
    task = Task.async(fn ->
      data = fetch_data(page)
      dispatch.("load_success", data)
    end)

    fn -> Task.shutdown(task, :brutal_kill) end
  end
end

Callbacks

• init_state/0 - Define initial state for this reducer's slice (optional, defaults to %{})
• handle_action/2 - Handle synchronous actions (receives Action struct), return updated state
• handle_async/3 - Handle async actions with dispatch callback, return cancellation function (() -> any())
  • dispatch signature: dispatch(type, payload \ nil, meta \ []) where type is binary, meta is keyword list
  • Default implementation returns fn -> nil end

Module Attributes

• @throttle {action_pattern, duration} - Throttle action (execute immediately, then block)
• @debounce {action_pattern, duration} - Debounce action (delay execution, reset timer)

Duration format: "500ms", "1s", "5m", "1h"

State Initialization

Each reducer defines its initial state via init_state/0. When using combined_reducers/0, the SessionProcess automatically calls each reducer's init_state/0 to build the complete initial state with each reducer's slice.

@spec call(binary(), any(), :infinity | non_neg_integer()) :: any()

Makes a synchronous call to a session process.

Sends a synchronous request to the session process and waits for a response. The request is handled by the session process's handle_call/3 callback.

Parameters

• session_id - Unique binary identifier for the session
• request - The request message to send
• timeout - Maximum time to wait for response in milliseconds (default: 15,000)

Returns

• Response from the session process's handle_call/3 callback
• {:error, {:session_not_found, id}} - Session does not exist
• {:error, {:timeout, timeout}} - Request timed out

Return Value Conventions

The return value is whatever the session process's handle_call/3 callback replies with. The built-in handlers injected by use Phoenix.SessionProcess, :process wrap their replies in {:ok, result} tuples:

• :get_state — returns {:ok, state}
• {:select_state, selector} — returns {:ok, selected_value}
• {:subscribe_with_selector, selector, event, pid} — returns {:ok, subscription_id}
• {:unsubscribe, subscription_id} — returns {:ok, :unsubscribed}

Custom handle_call/3 implementations in your session process module should follow this {:ok, result} convention for consistency, especially if the results will be consumed through higher-level helpers like get_state/2 or select_state/2.

Examples

{:ok, _pid} = Phoenix.SessionProcess.start_session("user_123")
{:ok, state} = Phoenix.SessionProcess.call("user_123", :get_state)
{:ok, :pong} = Phoenix.SessionProcess.call("user_123", :ping, 5_000)

@spec cast(binary(), any()) :: :ok | {:error, term()}

Sends an asynchronous message to a session process.

Sends a fire-and-forget message to the session process. The message is handled by the session process's handle_cast/2 callback. Does not wait for a response.

Parameters

• session_id - Unique binary identifier for the session
• request - The message to send

Returns

• :ok - Message sent successfully
• {:error, {:session_not_found, id}} - Session does not exist

Examples

{:ok, _pid} = Phoenix.SessionProcess.start_session("user_123")
:ok = Phoenix.SessionProcess.cast("user_123", {:put, :user_id, 123})
:ok = Phoenix.SessionProcess.cast("user_123", {:delete, :old_key})

@spec child_spec(any()) :: Supervisor.child_spec()

Returns a child specification for starting the SessionProcess supervision tree.

This is automatically called when adding {Phoenix.SessionProcess, []} to a supervision tree. The function delegates to Phoenix.SessionProcess.Supervisor.child_spec/1.

Parameters

• init_arg - Initialization argument passed to start_link/1

Returns

• Child specification map with :id, :start, and other supervisor options

Examples

The child spec is automatically used in supervision trees:

children = [
  {Phoenix.SessionProcess, []}  # child_spec/1 is called automatically
]

@spec dispatch(binary(), String.t(), term(), keyword()) :: :ok | {:error, term()}

Dispatch an action to a session process asynchronously (fire-and-forget).

The action will be processed through all registered reducers and subscribers will be notified if their selected state changed. This function returns immediately without waiting for the action to be processed.

Parameters

• session_id - Session identifier (binary string)
• action_type - Action type identifier (binary string, required)
• payload - Action payload (any term, defaults to nil)
• meta - Action metadata (keyword list, defaults to []):
  • :reducers - List of reducer names (atoms) to target explicitly. When specified:
    • Bypasses normal prefix routing entirely
    • Only calls the specified reducers
    • Passes full action type WITHOUT prefix stripping
    • Logs warning if any reducers don't exist
  • :reducer_prefix - Prefix to filter reducers by (when :reducers not specified)
  • :async - Route to handle_async/3 (true) or handle_action/2 (false, default)

Action Type Stripping Behavior

Normal dispatch (without meta.reducers):

• Reducers with @action_prefix "user" receive action type with prefix stripped
• Example: dispatch(id, "user.reload") → reducer sees "reload"

Explicit targeting (with meta.reducers):

• Action type is passed unchanged to specified reducers
• Example: dispatch(id, "user.reload", nil, reducers: [:user]) → reducer sees "user.reload"

Returns

• :ok - Action dispatched successfully
• {:error, {:session_not_found, session_id}} - If session doesn't exist

Examples

# Simple action with prefix routing (type stripped)
:ok = SessionProcess.dispatch(session_id, "user.reload")
# → UserReducer receives "reload" (prefix stripped)

# Action with payload
:ok = SessionProcess.dispatch(session_id, "user.update", %{name: "Alice"})

# Force specific reducers (type NOT stripped)
:ok = SessionProcess.dispatch(session_id, "user.reload", nil, reducers: [:user, :cart])
# → ONLY :user and :cart reducers called
# → Both receive "user.reload" (prefix NOT stripped)

# Warning logged for missing reducers
:ok = SessionProcess.dispatch(session_id, "reload", nil, reducers: [:nonexistent])
# → Logs warning: "Missing reducers: [:nonexistent]"

# Use prefix filter
:ok = SessionProcess.dispatch(session_id, "fetch-data", nil, reducer_prefix: "user")

# Async action (routed to handle_async/3)
:ok = SessionProcess.dispatch(session_id, "fetch-data", %{page: 1}, async: true)

@spec dispatch_async(binary(), String.t(), term(), keyword()) ::
  :ok | {:error, term()}

Dispatch an action asynchronously (convenience alias).

This is a convenience function that automatically adds async: true to the meta. It's equivalent to calling dispatch(session_id, type, payload, [meta | async: true]).

When async: true is in the meta, the action will trigger handle_async/3 callbacks in reducers that define them. The handle_async callback receives a dispatch function and must return a cancellation callback for internal lifecycle management.

Parameters

• session_id - Session identifier
• action_type - Action type (binary string)
• payload - Action payload (any term)
• meta - Action metadata (keyword list, async: true will be added automatically)

Returns

• :ok - Action dispatched successfully
• {:error, {:session_not_found, session_id}} - If session doesn't exist

Examples

# Equivalent to: dispatch(id, "user.reload", nil, async: true)
:ok = SessionProcess.dispatch_async(session_id, "user.reload")

# With payload - equivalent to: dispatch(id, "fetch_data", data, async: true)
:ok = SessionProcess.dispatch_async(session_id, "fetch_data", %{page: 1})

# With additional meta - equivalent to: dispatch(id, "reload", nil, [priority: :high, async: true])
:ok = SessionProcess.dispatch_async(session_id, "reload", nil, priority: :high)

@spec find_session(binary()) :: {:ok, pid()} | {:error, :not_found}

Finds a session by its ID and returns its PID if it exists.

This is different from started?/1 in that it returns the actual PID of the session process, which can be used for direct process operations.

Examples

iex> {:ok, pid} = Phoenix.SessionProcess.find_session("existing_session_id")
iex> is_pid(pid)
true

iex> {:error, :not_found} = Phoenix.SessionProcess.find_session("nonexistent_session_id")
iex> {:error, :not_found}

Parameters

• session_id - The session ID to look up

Returns

• {:ok, pid()} - The PID of the session process if found
• {:error, :not_found} - If the session doesn't exist

@spec get_state(binary(), function() | nil) :: any()

Get the current state, optionally applying a selector function locally.

This function retrieves the full state from the session process and optionally applies a selector function on the client side.

Parameters

• session_id - Session identifier
• selector - Optional selector function (state) -> selected_data

Returns

• State or selected data
• {:error, reason} - If session not found

Examples

# Get full state
state = SessionProcess.get_state(session_id)

# Get with inline selector (applied locally)
user = SessionProcess.get_state(session_id, fn s -> s.user end)

See Also

• select_state/2 - Apply selector on server side (more efficient for large states)

@spec list_session() :: [{binary(), pid()}]

Returns all active sessions as a list of {session_id, pid} tuples.

Examples

iex> is_list(Phoenix.SessionProcess.list_session())
true

# Returns list of {session_id, pid} tuples, or empty list if no sessions exist

@spec list_sessions_by_module(module()) :: [binary()]

Returns all session IDs for sessions managed by a specific module.

Examples

iex> sessions = Phoenix.SessionProcess.list_sessions_by_module(MyApp.SessionProcess)
iex> is_list(sessions)
true
iex> Enum.all?(sessions, &is_binary/1)
true

Parameters

• module - The session process module to filter by

Returns

• [binary()] - List of session IDs managed by the specified module

@spec select_state(binary(), function()) :: any()

Select state using a selector function applied on the server side.

This function sends the selector to the session process where it is applied, returning only the selected data. This is more efficient than get_state/2 when dealing with large state objects, as it avoids transferring the entire state over the process boundary.

Parameters

• session_id - Session identifier
• selector - Selector function (state) -> selected_data (must be a function reference or anonymous function)

Returns

• Selected data from state
• {:error, reason} - If session not found

Examples

# Select specific field (server-side selection)
user = SessionProcess.select_state(session_id, fn s -> s.user end)

# Select nested data
cart_count = SessionProcess.select_state(session_id, fn s -> length(s.cart) end)

# Select computed value
total = SessionProcess.select_state(session_id, fn s ->
  Enum.reduce(s.cart, 0, fn item, acc -> acc + item.price end)
end)

Performance Note

Use select_state/2 instead of get_state/2 when:

• State is large and you only need a small portion
• You want to compute derived values on the server side
• You want to minimize data transfer between processes

@spec session_info() :: %{count: integer(), modules: [module()]}

Returns session statistics including total count and modules in use.

Examples

iex> info = Phoenix.SessionProcess.session_info()
iex> is_map(info)
true
iex> Map.has_key?(info, :count)
true
iex> Map.has_key?(info, :modules)
true

Returns

• %{count: integer(), modules: list(module())} - A map containing the total number of active sessions and a list of unique session process modules.

@spec session_stats() :: %{
  total_sessions: integer(),
  memory_usage: integer(),
  avg_memory_per_session: integer()
}

Returns detailed session statistics including process count and memory usage.

Useful for monitoring and debugging session process performance.

Examples

iex> stats = Phoenix.SessionProcess.session_stats()
iex> is_map(stats)
true
iex> Map.has_key?(stats, :total_sessions)
true
iex> Map.has_key?(stats, :memory_usage)
true
iex> Map.has_key?(stats, :avg_memory_per_session)
true

Returns

• %{total_sessions: integer(), memory_usage: integer(), avg_memory_per_session: integer()} - Session statistics map
  • total_sessions - Total number of active session processes
  • memory_usage - Total memory usage in bytes for all session processes
  • avg_memory_per_session - Average memory usage per session in bytes

@spec start_link(any()) :: Supervisor.on_start()

Starts the SessionProcess supervision tree.

This function delegates to Phoenix.SessionProcess.Supervisor.start_link/1 and should be used in your application's supervision tree for cleaner syntax.

Parameters

• init_arg - Initialization argument (typically an empty list [])

Returns

• {:ok, pid} - Supervisor started successfully
• {:error, reason} - If supervisor failed to start

Examples

In your application supervision tree:

def start(_type, _args) do
  children = [
    # Cleaner syntax using Phoenix.SessionProcess
    {Phoenix.SessionProcess, []}

    # Or explicitly use the Supervisor module
    {Phoenix.SessionProcess.Supervisor, []}
  ]

  Supervisor.start_link(children, strategy: :one_for_one)
end

@spec start_session(binary()) :: {:ok, pid()} | {:error, term()}

Starts a session process using the default configured module.

The session process is registered in the Registry and scheduled for automatic cleanup based on the configured TTL.

Parameters

• session_id - Unique binary identifier for the session

Returns

• {:ok, pid} - Session process started successfully
• {:error, {:already_started, pid}} - Session already exists
• {:error, {:invalid_session_id, id}} - Invalid session ID format
• {:error, {:session_limit_reached, max}} - Maximum sessions exceeded

Examples

{:ok, pid} = Phoenix.SessionProcess.start_session("user_123")
{:error, {:already_started, pid}} = Phoenix.SessionProcess.start_session("user_123")

@spec start_session(binary(), keyword()) :: {:ok, pid()} | {:error, term()}

Starts a session process with options.

This allows you to customize the module and initialization arguments.

Parameters

• session_id - Unique binary identifier for the session
• opts - Keyword list of options:
  • :module - Module implementing the session process behavior (defaults to configured module)
  • :args - Initialization arguments passed to init/1 (defaults to nil)

Returns

• {:ok, pid} - Session process started successfully
• {:error, {:already_started, pid}} - Session already exists
• {:error, {:invalid_session_id, id}} - Invalid session ID format
• {:error, {:session_limit_reached, max}} - Maximum sessions exceeded

Examples

# Use default module
{:ok, pid} = Phoenix.SessionProcess.start_session("user_123")

# Use custom module
{:ok, pid} = Phoenix.SessionProcess.start_session("user_123", module: MyApp.CustomSessionProcess)

# Use custom module with initialization arguments
{:ok, pid} = Phoenix.SessionProcess.start_session("user_123",
  module: MyApp.SessionProcess,
  args: %{user_id: 123})

# Use default module with initialization arguments
{:ok, pid} = Phoenix.SessionProcess.start_session("user_456", args: [debug: true])

iex> result = Phoenix.SessionProcess.start_session("valid_session", module: Phoenix.SessionProcess.DefaultSessionProcess)
iex> match?({:ok, _pid}, result) or match?({:error, {:already_started, _pid}}, result)
true

iex> result = Phoenix.SessionProcess.start_session("valid_with_args", module: Phoenix.SessionProcess.DefaultSessionProcess, args: %{user_id: 123})
iex> match?({:ok, _pid}, result) or match?({:error, {:already_started, _pid}}, result)
true

@spec started?(binary()) :: boolean()

Checks if a session process is currently running.

Parameters

• session_id - Unique binary identifier for the session

Returns

• true - Session process exists and is running
• false - Session process does not exist

Examples

{:ok, _pid} = Phoenix.SessionProcess.start_session("user_123")
true = Phoenix.SessionProcess.started?("user_123")
false = Phoenix.SessionProcess.started?("nonexistent")

@spec subscribe(binary(), function(), atom(), pid()) ::
  {:ok, reference()} | {:error, term()}

Subscribe to state changes with a selector function.

The selector will be called with the current state immediately, and you'll receive a message with the selected value. Then, on each state change, if the selected value changes, you'll receive another message.

Parameters

• session_id - Session identifier
• selector - Function that extracts data from state: (state) -> selected_data
• event_name - Atom for the message event (default: :state_changed)
• pid - Process to receive notifications (default: self())

Returns

• {:ok, subscription_id} - Unique reference for this subscription
• {:error, reason} - If session not found

Messages

Subscriber receives: {event_name, selected_data}

Examples

# Subscribe to user data
{:ok, sub_id} = SessionProcess.subscribe(
  session_id,
  fn state -> state.user end,
  :user_changed
)

# Immediately receive current user
receive do
  {:user_changed, user} -> IO.puts("Current user: #{inspect(user)}")
end

# Receive updates when user changes
receive do
  {:user_changed, new_user} -> IO.puts("User updated: #{inspect(new_user)}")
end

# Later, unsubscribe
SessionProcess.unsubscribe(session_id, sub_id)

@spec terminate(binary()) :: :ok | {:error, :not_found}

Terminates a session process.

This gracefully shuts down the session process and removes it from the Registry. Emits telemetry events for session stop.

Parameters

• session_id - Unique binary identifier for the session

Returns

• :ok - Session terminated successfully
• {:error, :not_found} - Session does not exist

Examples

{:ok, _pid} = Phoenix.SessionProcess.start_session("user_123")
:ok = Phoenix.SessionProcess.terminate("user_123")
{:error, :not_found} = Phoenix.SessionProcess.terminate("user_123")

@spec touch(binary()) :: :ok | {:error, :not_found | {:cleanup_unavailable, binary()}}

Refreshes the TTL for a session, extending its lifetime.

Call this function when you want to keep a session alive beyond its normal TTL. This is useful for active sessions that should not expire even if they haven't received calls or casts recently.

Parameters

• session_id - Unique binary identifier for the session

Returns

• :ok - Session TTL refreshed successfully
• {:error, :not_found} - Session does not exist

Examples

{:ok, _pid} = Phoenix.SessionProcess.start_session("user_123")

# Keep session alive
:ok = Phoenix.SessionProcess.touch("user_123")

# Session TTL is reset to full duration

@spec unsubscribe(binary(), reference()) :: :ok | {:error, term()}

Unsubscribe from state changes.

Parameters

• session_id - Session identifier
• subscription_id - Reference returned from subscribe/4

Returns

• :ok - Successfully unsubscribed
• {:error, reason} - If session not found

Examples

{:ok, sub_id} = SessionProcess.subscribe(session_id, selector, :event)
# ... later ...
:ok = SessionProcess.unsubscribe(session_id, sub_id)