roar

import whisper

/// A distributed pub/sub router that synchronizes messages across a cluster.
///
/// Roar combines local message delivery (via Whisper) with automatic distribution
/// to all nodes in the same scope. Messages published on one node are automatically
/// forwarded to subscribers on all other nodes sharing the same scope identifier.
///
/// ## Example
///
/// ```gleam
/// let router = roar.new(capacity: 100, scope: "my_app")
/// let sub = roar.subscribe(router, "events")
///
/// roar.publish(router, "events", "Hello from node 1!")
/// // Subscribers on ALL nodes in "my_app" scope receive this message
/// ```
pub opaque type Roar(a) {
  Roar(local: whisper.Whisper(a), scope: String)
}

/// Create a new distributed pub/sub router.
///
/// - `capacity`: Maximum number of messages buffered per subscription
/// - `scope`: Unique identifier for this router's cluster scope. Only routers
///   with matching scopes will exchange messages across nodes.
///
/// ## Example
///
/// ```gleam
/// let router = roar.new(capacity: 50, scope: "chat_system")
/// ```
pub fn new(capacity capacity: Int, scope scope: String) -> Roar(a) {
  Roar(local: whisper.new(capacity), scope: scope)
}

/// Register a callback function to be invoked when messages are published to a topic.
///
/// The callback is triggered for messages published locally and from remote nodes
/// in the same scope. Returns a cancellation function to stop listening.
///
/// ## Example
///
/// ```gleam
/// let cancel = roar.on(router, "notifications", fn(msg) {
///   io.println("Received: " <> msg)
/// })
///
/// // Later, stop listening
/// cancel()
/// ```
pub fn on(roar: Roar(a), topic: String, listener: fn(a) -> Nil) -> fn() -> Nil {
  let local_cancel = whisper.on(roar.local, topic, listener)
  let distributed_cancel = platform_on_distributed(roar.scope, topic, listener)

  fn() {
    local_cancel()
    distributed_cancel()
  }
}

@external(erlang, "roar_ffi", "on_distributed")
fn platform_on_distributed(
  scope: String,
  topic: String,
  listener: fn(a) -> Nil,
) -> fn() -> Nil

/// Subscribe to a topic and receive messages through a buffered subscription.
///
/// Messages published locally or from remote nodes in the same scope are queued
/// in the subscription buffer. Use the returned subscription's `receive` function
/// to consume messages, and `cancel` to unsubscribe.
///
/// ## Example
///
/// ```gleam
/// let sub = roar.subscribe(router, "events")
///
/// case sub.receive() {
///   Ok(message) -> io.println("Got: " <> message)
///   Error(Nil) -> io.println("No messages available")
/// }
///
/// sub.cancel()
/// ```
pub fn subscribe(roar: Roar(a), topic: String) -> whisper.Subscription(a) {
  let sub = whisper.subscribe(roar.local, topic)
  platform_subscribe_distributed(roar.scope, topic, roar.local)
  sub
}

@external(erlang, "roar_ffi", "subscribe_distributed")
fn platform_subscribe_distributed(
  scope: String,
  topic: String,
  local_whisper: whisper.Whisper(a),
) -> Nil

/// Publish a message to all subscribers of a topic across the cluster.
///
/// The message is delivered to:
/// - All local subscribers on this node
/// - All subscribers on remote nodes in the same scope
///
/// ## Example
///
/// ```gleam
/// roar.publish(router, "alerts", "System maintenance in 5 minutes")
/// // All subscribers across all nodes receive this message
/// ```
pub fn publish(roar: Roar(a), topic: String, message: a) -> Nil {
  whisper.publish(roar.local, topic, message)
  platform_publish_distributed(roar.scope, topic, message)
}

@external(erlang, "roar_ffi", "publish_distributed")
fn platform_publish_distributed(scope: String, topic: String, message: a) -> Nil