ExWapp.Session.Worker (ExWapp v0.1.2)

Copy Markdown View Source

Coordinates the lifecycle of a WhatsApp Web connection.

When started with runtime.ownership.vext_managed: true, the worker behaves as a transport engine only: it still emits raw lifecycle facts, but leaves reconnect and product-level stop policy to the external owner.

This GenServer is designed for long-running production use with:

  • Automatic reconnection with exponential backoff
  • Connection health monitoring
  • Graceful degradation on errors
  • Telemetry events for observability
  • Crash recovery via supervision

What lives here, and what does not

This module owns the process: the state struct, the GenServer callbacks and the decision of what each incoming message means. The work each decision triggers lives in a module named after the concern, all of them called from this process — a module is not a process, so the transport is still linked to the session, monitors are registered against it and timers land in its mailbox.

Supervision

For production use, run Session under ExWapp.Session.Supervisor:

{:ok, sup} = ExWapp.Session.Supervisor.start_link(
  store: {ExWapp.Store.Ets, path: "/var/lib/myapp/wa.etf"}
)
session = ExWapp.Session.Supervisor.session(sup)

Telemetry Events

The session emits the following telemetry events:

  • [:ex_wapp, :session, :init] - Session process initialized
  • [:ex_wapp, :session, :connect] - Connection attempt
  • [:ex_wapp, :session, :connected] - Successfully connected
  • [:ex_wapp, :session, :disconnected] - Connection lost
  • [:ex_wapp, :session, :reconnect] - Reconnection attempt
  • [:ex_wapp, :session, :error] - Error occurred
  • [:ex_wapp, :session, :qr] - QR lifecycle event (:code, :success, :error)
  • [:ex_wapp, :session, :message_received] - Message persisted/forwarded
  • [:ex_wapp, :session, :message_sent] - Message sent to transport
  • [:ex_wapp, :session, :terminate] - Session terminate callback

State Machine

idle -> connecting -> handshaking -> syncing -> connected
  ^                                                  |
  |                                                  v
  +<---------------- disconnected <------------------+

Summary

Types

Human readable events emitted during the pairing flow.

Session status

t()

Functions

Returns a specification to start this module under a supervisor.

Gracefully disconnects from WhatsApp servers.

Get chat info (metadata) for a JID.

Get a specific contact by JID.

Returns current send health counters/guard state.

Returns current policy status (safety, quota, rate limiter snapshots).

Returns the resolved runtime configuration for this session.

Returns the session_id for this worker (set at start_link via :session_id option).

List all contacts.

Builds a lazy stream of incoming messages.

Pause outbound sends for this session.

Builds a lazy stream of QR pairing events.

Requests phone-number pairing while the Noise transport is awaiting authentication.

Resume outbound sends for this session.

Send an app state mutation (server-synced).

Starts the session process.

Returns the current session state for inspection.

Returns session statistics.

Stops the session worker immediately.

Subscribe to incoming messages.

Subscribes to QR code events.

Fetch and apply the contact app-state collection (critical_unblock_low).

Unsubscribe from messages.

Unsubscribes from QR code events.

Types

qr_event()

@type qr_event() ::
  {:code, String.t()}
  | {:pairing_code, String.t()}
  | :success
  | {:error, term()}

Human readable events emitted during the pairing flow.

status()

@type status() ::
  :idle | :connecting | :handshaking | :syncing | :connected | :disconnected

Session status

t()

@type t() :: %ExWapp.Session.Worker{
  conn: pid() | nil,
  conn_ref: reference() | nil,
  connected_at: DateTime.t() | nil,
  digest_iq_id: binary() | nil,
  health_timer: reference() | nil,
  inbound_router: pid() | nil,
  initial_sync: ExWapp.Session.InitialSync.t(),
  last_disconnect_reason: term() | nil,
  message_subscribers: %{required(reference()) => {pid(), reference()}},
  opts: ExWapp.Session.Options.t(),
  outbound_confirmations: map(),
  outbound_queue: pid() | nil,
  outgoing_retry_payloads: map(),
  pairing: ExWapp.Pairing.state(),
  passive_iq_id: binary() | nil,
  pending_iqs: term(),
  pending_media_retries: map(),
  pending_prekey_upload: list(),
  post_auth_phase: atom(),
  prekey_count_query_id: binary() | nil,
  prekey_upload_iq_id: binary() | nil,
  qr_subscribers: %{required(reference()) => {pid(), reference()}},
  quota: map(),
  rate_limiter: map(),
  reconnect_attempts: non_neg_integer(),
  retry_counters: map(),
  runtime_cfg: map(),
  safety: map(),
  send_coordinator: ExWapp.Session.SendCoordinator.t(),
  send_health: map(),
  send_worker_sup: pid() | nil,
  server_time_offset_ms: integer(),
  session_id: term() | nil,
  session_invalid: term(),
  signal_sessions: term(),
  stats: map(),
  status: status(),
  store: ExWapp.Store.t() | nil,
  store_owned?: boolean(),
  transport_mod: module(),
  transport_ready?: boolean()
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

disconnect(server)

@spec disconnect(GenServer.server()) :: :ok

Gracefully disconnects from WhatsApp servers.

Stops reconnection attempts and closes the connection cleanly.

get_chat_info(server, jid)

@spec get_chat_info(GenServer.server(), String.t()) ::
  {:ok, ExWapp.Chat.chat()} | {:error, :not_found}

Get chat info (metadata) for a JID.

get_contact(server, jid)

@spec get_contact(GenServer.server(), String.t()) :: ExWapp.Contact.t() | nil

Get a specific contact by JID.

get_health_status(server)

@spec get_health_status(GenServer.server()) :: map()

Returns current send health counters/guard state.

get_policy_status(server)

@spec get_policy_status(GenServer.server()) :: map()

Returns current policy status (safety, quota, rate limiter snapshots).

get_runtime_config(server)

@spec get_runtime_config(GenServer.server()) :: map()

Returns the resolved runtime configuration for this session.

get_session_id(server)

@spec get_session_id(GenServer.server()) :: term() | nil

Returns the session_id for this worker (set at start_link via :session_id option).

Returns nil when no explicit session_id was provided.

list_contacts(server)

@spec list_contacts(GenServer.server()) :: [ExWapp.Contact.t()]

List all contacts.

message_stream(server)

@spec message_stream(GenServer.server()) :: Enumerable.t()

Builds a lazy stream of incoming messages.

Each element is a tuple of {jid, message}.

pause_sends(server)

@spec pause_sends(GenServer.server()) :: :ok

Pause outbound sends for this session.

qr_stream(server)

@spec qr_stream(GenServer.server()) :: Enumerable.t()

Builds a lazy stream of QR pairing events.

The stream yields events until pairing succeeds, fails, or is cancelled.

request_pairing_code(server, phone, opts \\ [])

@spec request_pairing_code(GenServer.server(), String.t(), keyword()) ::
  {:ok, String.t()} | {:error, term()}

Requests phone-number pairing while the Noise transport is awaiting authentication.

resume_sends(server)

@spec resume_sends(GenServer.server()) :: :ok

Resume outbound sends for this session.

send_app_state(server, patch_info)

@spec send_app_state(GenServer.server(), map()) ::
  {:ok, map(), map()} | {:error, term()}

Send an app state mutation (server-synced).

start_link(opts \\ [])

@spec start_link(keyword() | ExWapp.Session.Options.t()) :: GenServer.on_start()

Starts the session process.

Options

  • :name - Register the process under a given name
  • :store - Storage adapter (see ExWapp.Store)
  • :supervisor - Parent supervisor PID (set automatically by Supervisor)
  • :auto_connect - Connect automatically on start (default: false)
  • :reconnect - Enable automatic reconnection (default: true)
  • :runtime - Runtime hardening settings override (deep-merged)

state(server)

@spec state(GenServer.server()) :: status()

Returns the current session state for inspection.

stats(server)

@spec stats(GenServer.server()) :: map()

Returns session statistics.

stop_worker(server)

@spec stop_worker(GenServer.server()) :: :ok

Stops the session worker immediately.

subscribe_messages(server)

@spec subscribe_messages(GenServer.server()) :: {:ok, reference()}

Subscribe to incoming messages.

Returns a reference that will be included in message notifications. Messages arrive as {:ex_wapp_message, ref, jid, message}.

subscribe_qr(server)

@spec subscribe_qr(GenServer.server()) :: {:ok, reference()}

Subscribes to QR code events.

sync_contacts(server)

@spec sync_contacts(GenServer.server()) :: :ok | {:error, term()}

Fetch and apply the contact app-state collection (critical_unblock_low).

unsubscribe_messages(server, ref)

@spec unsubscribe_messages(GenServer.server(), reference()) :: :ok

Unsubscribe from messages.

unsubscribe_qr(server, ref)

@spec unsubscribe_qr(GenServer.server(), reference()) :: :ok

Unsubscribes from QR code events.