lily/topic

Topics are basically ways for clients to listen to a certain broadcast channel, similar to pub/sub patterns within other libraries like Phoenix. This is useful for cases where your own model/store need to easily react to other states shared between a few clients (e.g. chat rooms).

Lily provides two main types of topics, an ephemeral topic that simply broadcasts Push frames with no sequence and no replay:

let assert Ok(typing) = topic.new(server, id: "typing")
// Broadcast from anywhere, server or client
topic.broadcast(typing, UserIsTyping(client_id))

And also more stateful topics which have their own store. Pipe through with_store to make a topic stateful, the topic actor reading its update logic from the server’s Wiring and sends TopicUpdate frames to every subscriber:

let assert Ok(chat) = topic.new(server, id: "chat")
let chat =
  chat
  |> topic.with_store
  |> topic.with_on_subscribe(fn(client_id) {
    [Chat(UserJoined(client_id))]
  })

Making it stateful means the server owns the topic’s state authoritatively. It assigns a sequence number to every change, hands a Snapshot to whoever subscribes so late joiners catch up, and fans out a TopicUpdate as the state changes. The update logic isn’t passed here, it’s borrowed from the store.topic(id:) entry in your shared Wiring whose id matches, so a "chat" topic reads its update function and slice from store.topic(id: "chat", ...). That one entry is what links the server actor to the client’s model slice, so the same id shows up in the wiring, in topic.new, and in client.subscribe.

On the client you subscribe after connecting with client.subscribe, and the topic’s slice is hydrated from the snapshot when it arrives:

runtime
|> client.connect(with: connector, serialiser: shared.serialiser())
|> client.subscribe("chat")

After that, you don’t wire up anything to consume updates. When a TopicUpdate from a stateful topic or a Push from an ephemeral one lands, Lily runs the payload through your update function exactly like a message you’d dispatched locally, with the matching slice updates, your client.on_message hook fires, and the affected components re-render.

With the chat topic set up on the backend (the topic.new plus with_store from above), a client (frontend) sends by dispatching, which the server fans out to the other subscribers as a TopicUpdate:

dispatch(Chat(NewChatMessage(body)))

And consuming that update takes no extra code, a component slicing the chat re-renders whether the change was local or a TopicUpdate from elsewhere (another client/frontend):

component.simple(
  slice: fn(model: Model) { model.chat },
  render: fn(chat, _) { chat_view(chat) },
)

For dynamic topics keyed by a parsed identifier (e.g. "room:42"), use topic.kind to register a factory that creates topic actors on first subscribe:

let assert Ok(_) =
  topic.kind(
    server,
    prefix: "room:",
    parse_id: int.parse,
    configure: fn(room_id, topic) {
      topic
      |> topic.with_store
      |> topic.with_can_subscribe(fn(client_id, _topic_id) {
        auth.may_join_room(client_id, room_id) // your own helper
      })
    },
  )

On the wiring side this pairs with store.topic_kind rather than store.topic. It binds the whole "room:" family to one keyed model slice, so a client can be in several rooms at once and each outgoing message, incoming update, and snapshot still lands on the right one. Back the slice with a keyed collection like Dict(String, RoomState), the key is the bit after the prefix ("42" for "room:42").

pub type Ephemeral

pub type Stateful

pub opaque type Topic(model, message, kind)

pub fn broadcast(
  topic: Topic(model, message, kind),
  message: message,
) -> Nil

topic.broadcast(typing_topic, UserIsTyping(client_id))

pub fn broadcast_from(
  topic: Topic(model, message, kind),
  except client_id: String,
  message message: message,
) -> Nil

topic.broadcast_from(
  typing_topic,
  except: client_id,
  message: UserIsTyping(client_id),
)

pub fn dispatch(
  topic: Topic(model, message, Stateful),
  message: message,
) -> Nil

topic.dispatch(chat_topic, Chat(NewChatMessage(body)))

pub fn kind(
  server: server.Server(model, message),
  prefix prefix: String,
  parse_id parse_id: fn(String) -> Result(parsed, Nil),
  configure configure: fn(
    parsed,
    Topic(model, message, Ephemeral),
  ) -> Topic(model, message, kind),
) -> Result(Nil, Nil)

let assert Ok(_) =
  topic.kind(
    server,
    prefix: "room:",
    parse_id: int.parse,
    configure: fn(room_id, topic) {
      topic |> topic.with_store
    },
  )

pub fn new(
  server: server.Server(model, message),
  id id: String,
) -> Result(Topic(model, message, Ephemeral), Nil)

let assert Ok(typing) = topic.new(server, id: "typing")

pub fn stop(topic: Topic(model, message, kind)) -> Nil

topic.stop(chat_topic)

pub fn subscribe(
  topic: Topic(model, message, kind),
  client_id: String,
) -> Nil

topic.subscribe(chat_topic, client_id)

pub fn unsubscribe(
  topic: Topic(model, message, kind),
  client_id: String,
) -> Nil

topic.unsubscribe(chat_topic, client_id)

pub fn with_can_subscribe(
  topic: Topic(model, message, kind),
  predicate: fn(String, String) -> Bool,
) -> Topic(model, message, kind)

topic.with_can_subscribe(chat_topic, fn(client_id, _topic_id) {
  auth.is_authenticated(client_id)
})

pub fn with_on_subscribe(
  topic: Topic(model, message, kind),
  hook: fn(String) -> List(message),
) -> Topic(model, message, kind)

topic.with_on_subscribe(chat_topic, fn(client_id) {
  [Chat(UserJoined(client_id))]
})

pub fn with_on_unsubscribe(
  topic: Topic(model, message, kind),
  hook: fn(String) -> List(message),
) -> Topic(model, message, kind)

topic.with_on_unsubscribe(chat_topic, fn(_client_id) { [] })

pub fn with_store(
  topic: Topic(model, message, Ephemeral),
) -> Topic(model, message, Stateful)

topic.new(server, id: "chat")
|> topic.with_store