lily/server

The Server is the authoritative half of a Lily app. It holds the real state and routes every message a client sends to the right store, either that client’s own per-connection session store, or a named topic store shared between clients. Clients apply their own changes optimistically, the server has the final say and streams the corrected state back, so the two stay in sync (see transport for the frames that carry it).

It compiles for both targets, but not using the Erlang one means that you aren’t leveraging BEAM’s model and a full-stack JS model might be better suited.

You build a server with new, handing it the same three values your client gets, the initial model, the serialiser, and the shared Wiring, then start it and register your topics with topic.new:

import gleam/result

import lily/server
import lily/topic

pub fn main() {
  let assert Ok(server) =
    server.new(
      initial: shared.initial_model(),
      serialiser: shared.serialiser(),
      wiring: shared.wiring(),
    )
    |> server.start

  let assert Ok(_) =
    topic.new(server, id: "chat")
    |> result.map(topic.with_store)
}

Handing those same three values to client.start is what makes both ends agree on the model, the update logic, and the wire encoding, so define them once in your shared package and pass the same values to each side. See store for how the Wiring splits the model into a session slice and topic slices.

The server never imports mist or wisp, it’s completely transport-agnostic (the same way Lustre leaves the transport to you), so you wire it into whatever WebSocket or HTTP handler you’re running with three calls. Create a stable id for each connection with generate_client_id, register it with connect (handing over the send callback the server uses to push frames back to that one client), feed every inbound frame to incoming, and call disconnect when the socket closes:

let client_id = server.generate_client_id()
server.connect(server, client_id:, send: process.send(outgoing_subject, _))

// for every frame the socket receives:
server.incoming(server, client_id:, bytes:)

// when it closes:
server.disconnect(server, client_id:)

From there most apps just need the lifecycle hooks. on_connect and on_disconnect fire with the client id for presence tracking or audit logging, on_message runs after each session message is applied (giving you the decoded message, the whole model after the update, and the client id, the natural home for side effects like writing to a database), and on_topic_message fires for each client-incoming topic message:

server.on_connect(server, fn(client_id) {
  logging.info("client connected: " <> client_id)
})

server.on_message(server, fn(message, _model, _client_id) {
  case message {
    Session(SaveDocument(doc)) -> db.write(doc)
    _ -> Nil
  }
})

When the server itself needs to change a client’s state, rather than react to something the client sent, reach for dispatch_to. It applies a message to one client’s session store and pushes the update straight back to them, ideal for seeding a session right after connect or for delivering the result of some async work once it lands (it’s safe to call from any process, so spawn the slow thing and dispatch when it’s ready):

server.on_connect(server, fn(client_id) {
  server.dispatch_to(server, client_id:, message: WelcomeUser(client_id))
})

dispatch_to_all does the same to every connected client at once, for a server-wide change that also mutates session state. If you only want to send something out without touching session state, broadcast through a topic instead.

Shut a server down with stop, which asks every topic to stop first so subscribers’ slices reset cleanly before the underlying actor terminates.

Types

Server

opaque </>

Handle to a running Lily server. Wraps platform-specific internals (OTP actor on Erlang, Reference cell on JavaScript). Also carries serialiser and initial_model for zero-copy access by topic.new.

pub opaque type Server(model, message)

Values

connect

</>

pub fn connect(
  server: Server(model, message),
  client_id client_id: String,
  send send: fn(BitArray) -> Nil,
) -> Nil

server.connect(server, client_id: id, send: process.send(outgoing_subject, _))

disconnect

</>

pub fn disconnect(
  server: Server(model, message),
  client_id client_id: String,
) -> Nil

Unregister a client connection from the server and all subscribed topics.

dispatch_to

</>

pub fn dispatch_to(
  server: Server(model, message),
  client_id client_id: String,
  message message: message,
) -> Nil

Apply message to the session store of one connected client and send a SessionUpdate frame back to that client. Use for server-initiated per-client updates, e.g. pushing a fresh slice after authentication, server timers that affect one user, async DB results.

No-op if no client with client_id is currently connected. Safe to call from any process, including spawned tasks, so it composes with async work patterns, spawn a process to do the slow thing, then call dispatch_to when the result is ready.

server.on_connect(server, fn(client_id) {
  server.dispatch_to(server, client_id:, message: WelcomeUser(client_id))
})

dispatch_to_all

</>

pub fn dispatch_to_all(
  server: Server(model, message),
  message message: message,
) -> Nil

Apply message to every connected client’s session store and send each of them a SessionUpdate frame. Use for server-wide announcements that also mutate session state, such as forcing a feature-flag refresh. For fire-and-forget broadcasts without session mutation, use a topic instead.

server.dispatch_to_all(server, message: SystemBannerUpdated(banner))

generate_client_id

</>

pub fn generate_client_id() -> String

Generate a cryptographically random 32-character hex client identifier. Pair with connect so every connection carries a stable, server-issued id.

let client_id = server.generate_client_id()
server.connect(server, client_id:, send:)

incoming

</>

pub fn incoming(
  server: Server(model, message),
  client_id client_id: String,
  bytes bytes: BitArray,
) -> Nil

Process an incoming frame from a client. Decodes the frame and routes it: SessionMessage to the session store; TopicMessage, Subscribe, and Unsubscribe to the topic actor; Resync to a per-target snapshot fan-out.

new

</>

pub fn new(
  initial initial: model,
  serialiser serialiser: transport.Serialiser(model, message),
  wiring wiring: store.Wiring(model, message),
) -> @internal Builder(model, message)

Start building a server. Provide the shared initial model (used as the zero-state for per-connection session stores and for topic snapshot construction) and the serialiser.

server.new(
  initial: shared.initial_model(),
  serialiser: shared.serialiser(),
  wiring: shared.wiring(),
)

on_connect

</>

pub fn on_connect(
  server: Server(model, message),
  hook: fn(String) -> Nil,
) -> Nil

Register a hook that runs after a client successfully connects, receiving the client id assigned by the server. Use for presence tracking, audit logging, or seeding the client’s session via dispatch_to.

server.on_connect(server, fn(client_id) {
  logging.info("client connected: " <> client_id)
})

on_disconnect

</>

pub fn on_disconnect(
  server: Server(model, message),
  hook: fn(String) -> Nil,
) -> Nil

Register a hook that runs after a client disconnects, receiving the client id that just left. Use for presence cleanup or audit logging.

server.on_disconnect(server, fn(client_id) {
  logging.info("client disconnected: " <> client_id)
})

on_message

</>

pub fn on_message(
  server: Server(model, message),
  hook: fn(message, model, String) -> Nil,
) -> Nil

Register a hook that runs after each session message is applied. Receives the decoded message, the full outer model after the message has been applied to this client’s session store, and the originating client id.

server.on_message(server, fn(message, model, client_id) {
  case message {
    Session(SaveDocument(doc)) -> db.write(doc)
    _ -> Nil
  }
})

on_topic_message

</>

pub fn on_topic_message(
  server: Server(model, message),
  hook: fn(message, String, String) -> Nil,
) -> Nil

Register a hook that runs for each client-incoming topic message. Receives the decoded message, the topic id, and the originating client id. Fires before the topic actor processes the message, regardless of whether the topic is stateful, ephemeral, or unknown. Does not fire for server-initiated topic.dispatch / topic.broadcast calls.

server.on_topic_message(server, fn(message, topic_id, _client_id) {
  logging.auto_log(logging.Info, #(topic_id, message))
})

start

</>

pub fn start(
  builder: @internal Builder(model, message),
) -> Result(Server(model, message), Nil)

Start the configured server.

Topics are added then afterwards via topic.new(server, ...).

let assert Ok(server) =
  server.new(
    initial: shared.initial_model(),
    serialiser: shared.serialiser(),
    wiring: shared.wiring(),
  )
  |> server.start

stop

</>

pub fn stop(server: Server(model, message)) -> Nil

Stop a running server. Every registered topic actor is asked to stop first; each subscriber receives a final Acknowledge(Topic(id), seq) so client slices reset cleanly. The underlying server actor then terminates (Erlang) or its Reference state cell is cleared (JavaScript). Connected session clients receive no extra frame.