Pure Elixir client for Convex — the backend application platform.
Connects to Convex deployments over WebSocket using the native sync protocol. Supports queries, mutations, actions, real-time subscriptions, authentication, and automatic reconnection with exponential backoff.
Open follow-up work and conscious tradeoffs are tracked in TODO.md.
- Pure Elixir — no NIFs, no Rust dependencies
- Real-time subscriptions — live query results pushed to your processes
- Consistent snapshots —
watch_alldelivers atomic multi-query views - Connection state notifications — read or watch
:connecting/:connected - Automatic reconnection — exponential backoff with jitter, full state replay on reconnect
- Auth support — static tokens, callback-based token fetchers, admin auth
- Clean architecture — pure state machine core (
Convex.Sync.Core), effects at the edges
def deps do
[
{:convex, git: "https://github.com/celados/convex.git"}
]
end# Start a client
{:ok, client} = Convex.start_link(deployment_url: "https://your-app.convex.cloud")
# One-shot query
{:ok, messages} = Convex.query(client, "messages:list", %{"channel" => "general"})
# Mutation
{:ok, _} = Convex.mutation(client, "messages:send", %{"body" => "hello from elixir"})
# Action
{:ok, result} = Convex.action(client, "ai:generate", %{"prompt" => "hello"})Subscribe to a query and receive live updates as messages:
{:ok, sub} = Convex.subscribe(client, "messages:list", %{"channel" => "general"})
# Your process receives:
# {:convex, ref, {:ok, [%{"body" => "hello"}, ...]}}
# {:convex, ref, {:error, "something went wrong"}}
receive do
{:convex, ref, {:ok, messages}} ->
IO.inspect(messages, label: "live update")
end
# Stop receiving updates
Convex.unsubscribe(client, sub)Get a continuous stream of consistent snapshots across all active subscriptions:
{:ok, watch} = Convex.watch_all(client)
# Your process receives:
# {:convex_watch, ref, %{subscriber_id => result, ...}}
receive do
{:convex_watch, ref, snapshot} ->
IO.inspect(snapshot, label: "consistent snapshot")
end
Convex.unwatch_all(client, watch)Read the current connection state synchronously:
Convex.connection_state(client)
# => :connecting | :connectedOr watch for future transitions:
{:ok, watch} = Convex.watch_connection_state(client)
receive do
{:convex_connection_state, ref, state} ->
IO.inspect({ref, state}, label: "connection state")
end
Convex.unwatch_connection_state(client, watch)watch_connection_state/1 only emits changes after subscription. If you need the
current value immediately, call connection_state/1.
# Static token
Convex.set_auth(client, "your-jwt-token")
# Clear auth
Convex.set_auth(client, nil)
# Or provide a static token when starting
{:ok, client} = Convex.start_link(
deployment_url: "https://your-app.convex.cloud",
auth: Convex.Auth.Token.user("your-jwt-token")
)
# Token fetcher (called initially and on every reconnect)
{:ok, client} = Convex.start_link(
deployment_url: "https://your-app.convex.cloud",
auth_fetcher: fn force_refresh: force_refresh ->
# Return {:ok, Convex.Auth.Token.t()} or {:error, reason}
fetch_token(force_refresh)
end
)The fetcher runs outside the client process. While authentication is being
resolved, the transport may be open but queries, mutations, actions, and
subscription changes remain queued. They are sent only after authentication has
been resolved, so they never race an Authenticate message.
On reconnect, a successful fetch is sent before replaying outstanding requests. If fetching fails, that connection continues anonymously and the fetcher is tried again on the next reconnect. Replacing an authenticated static token with a fetcher first sends an explicit anonymous authentication message, so a fetch failure cannot leave requests running under the old credential.
Admin authentication uses a deployment deploy key. Treat it as a highly privileged credential: the client redacts it in Inspect output, but it must necessarily be present in protocol traffic sent to your Convex deployment.
identity =
Convex.Auth.Identity.new(
issuer: "https://your-auth-provider.example.com",
subject: "user-123",
custom_claims: %{"isAdmin" => true}
)
Convex.set_admin_auth(client, deploy_key)
Convex.set_admin_auth(client, deploy_key, identity)Identity.new/1 accepts atom-keyed keyword lists or maps. When
token_identifier is not supplied, issuer and subject are required and are
normalized to the upstream issuer <> "|" <> subject identifier. Custom claims
are JSON values and are flattened into the top level of the wire identity, just
like the official Rust client.
The fetcher API can also return an admin token:
auth_fetcher: fn force_refresh: force_refresh ->
if force_refresh do
{:ok, Convex.Auth.Token.admin(deploy_key, identity)}
else
{:ok, Convex.Auth.Token.user(jwt)}
end
endConvex has its own type system. The mapping between Elixir and Convex:
| Elixir | Convex | Notes |
|---|---|---|
nil |
Null | |
true / false |
Boolean | |
"hello" |
String | |
3.14 |
Float64 | |
42 |
Float64 | Native integers encode as Float64 (matches JS number) |
Convex.Int64.new(42) |
Int64 | Explicit wrapper for BigInt fields |
Convex.Bytes.new(<<1,2,3>>) |
Bytes | |
[1.0, "two"] |
Array | |
%{"key" => value} |
Object | String keys only |
Convex.start_link(
deployment_url: "https://your-app.convex.cloud",
# Auth (optional)
auth: Convex.Auth.Token.user("static-jwt"),
# Timeouts
inactivity_check_ms: 5_000, # how often to check server liveness
inactivity_timeout_ms: 30_000, # max silence before reconnect
# Backoff
backoff: [base_ms: 100, max_ms: 15_000],
# Transport (for testing)
transport_module: Convex.Transport.WebSocket # default
)Convex.Client (GenServer) — process orchestration, thin
├── Convex.Sync.Core — pure state machine, no I/O
│ ├── Convex.Sync.QueryToken — canonical query identity
│ └── Convex.Sync.RequestManager — request lifecycle tracking
├── Convex.Protocol.Codec — JSON wire format encode/decode
├── Convex.Value — Convex value type codec
└── Convex.Transport.WebSocket — WebSocket connection (behaviour)
The core state machine is a pure module — (state, input) → {state, effects}. All side effects (WebSocket I/O, process messaging, timers) are handled at the edges by Convex.Client.
Early stage. The sync protocol, value codec, and state machine are implemented and tested against the convex-rs reference implementation. Not yet published on Hex.
MIT