GitHub - rjpruitt16/glixir: Easy Gleam wrapper around elixir otp

GenServer:     [■■■■■■■■□□] 80% - Request/reply types enforced by Gleam, decoder required, but runtime BEAM interop can still fail if types disagree.
Supervisor:    [■■■■■■■■■□] 90% - Phantom-typed with compile-time child spec validation, args/replies bounded by generics.
Registry:      [■■■■■■■■■□] 90% - Phantom-typed with compile-time key/message validation, requires key encoders.
Agent:         [■■■■■■■■■■] 100% - State and API fully generic and type safe!
PubSub:        [■■■■■■■■□□] 80% - JSON-based type safety with user-defined encoders/decoders. Phantom-typed for message_type.
libcluster:    [■■■■■■■□□□] 70% - Node discovery works, cluster membership is runtime dynamic
syn:           [■■■■■■■■□□] 80% - Distributed coordination with type-safe message patterns, runtime node discovery.
Task:          [□□□□□□□□□□] 0% - Not started.

import glixir/libcluster
import gleam/erlang/os
import logging

pub fn main() {
  // Auto-detect environment and start clustering
  let _ = case os.get_env("FLY_APP_NAME") {
    Ok(app_name) -> {
      // Running on Fly.io - automatic DNS-based discovery
      libcluster.start_clustering_fly(app_name)
    }
    Error(_) -> {
      // Local development - use EPMD for simplicity
      libcluster.start_clustering_local("my_app")
    }
  }
  
  // Check cluster status
  io.println("Node: " <> libcluster.current_node_name())
  io.println("Connected to: " <> string.inspect(libcluster.connected_node_names()))
  
  // Your distributed app starts here
  // syn will automatically sync across all discovered nodes!
  start_distributed_app()
}

// Production deployment with custom DNS
pub fn production_clustering() {
  // DNS-based discovery for production
  let assert Ok(_) = libcluster.start_clustering_dns(
    app_name: "my_app",
    query: "my_app.service.consul",  // Consul, Kubernetes DNS, etc
    polling_interval: 5000
  )
  
  // Monitor cluster membership
  case libcluster.is_clustered() {
    True -> {
      let nodes = libcluster.connected_nodes()
      logging.info("Clustered with " <> int.to_string(list.length(nodes)) <> " nodes")
    }
    False -> {
      logging.warning("Running in single-node mode")
    }
  }
}

// Combine with syn for distributed process coordination
pub fn distributed_workers() {
  // Start clustering
  let assert Ok(_) = libcluster.start_clustering_fly("worker_pool")
  
  // syn automatically discovers all nodes via libcluster!
  syn.init_scopes(["workers"])
  
  // This registration is visible across ALL nodes
  syn.register("workers", "worker_1", self())
  
  // Find workers on ANY node in the cluster
  case syn.whereis("workers", "worker_1") {
    Ok(#(pid, node)) -> {
      io.println("Found worker on node: " <> atom.to_string(node))
    }
    Error(_) -> io.println("Worker not found in cluster")
  }
}

import glixir/syn
import gleam/json

pub fn distributed_example() {
  // Initialize scopes at application startup
  syn.init_scopes(["worker_pools", "coordination"])
  
  // Register this process in a distributed pool
  let load_info = #(cpu_usage: 0.3, queue_size: 5)
  let assert Ok(_) = syn.register_worker("image_processors", "worker_1", load_info)
  
  // Find workers across the cluster
  case syn.find_worker("image_processors", "worker_1") {
    Ok(#(pid, #(cpu_usage, queue_size))) -> {
      // Send work to the least loaded worker
      process.send(pid, ProcessImage("photo.jpg"))
    }
    Error(_) -> // Worker not available, try another
  }
  
  // Join coordination group for consensus
  let assert Ok(_) = syn.join_coordination("leader_election")
  
  // Broadcast status for distributed coordination
  let status = json.object([
    #("node", json.string("worker_node_1")),
    #("load", json.float(0.3)),
    #("available", json.bool(True))
  ])
  
  let assert Ok(nodes_notified) = syn.broadcast_status("health_check", status)
  io.println("Status sent to " <> int.to_string(nodes_notified) <> " nodes")
}

// Advanced: Custom coordination patterns
pub fn consensus_example() {
  // Register with metadata for consensus algorithms
  let machine_status = #(queue_lengths: #(0, 5, 2), capacity: 100)
  let assert Ok(_) = syn.register("machines", "machine_1", machine_status)
  
  // Publish for distributed decision making
  let queue_status = json.object([
    #("machine_id", json.string("machine_1")),
    #("free_queue", json.int(0)),
    #("paid_queue", json.int(5)),
    #("capacity", json.int(100))
  ])
  
  let assert Ok(_) = syn.publish_json("coordination", "load_balancing", queue_status, fn(j) { j })
}

## Installation

```sh
gleam add glixir

# lib/glixir_supervisor.ex and lib/glixir_registry.ex
# Copy the helper modules from the docs

import glixir
import gleam/json
import gleam/dynamic/decode
import gleam/erlang/atom

// Define your message types
pub type UserEvent {
  PageView(user_id: String, page: String)
  Purchase(user_id: String, amount: Float)
}

// Create encoder/decoder for type safety
pub fn encode_user_event(event: UserEvent) -> String {
  case event {
    PageView(user_id, page) -> 
      json.object([
        #("type", json.string("page_view")),
        #("user_id", json.string(user_id)),
        #("page", json.string(page))
      ]) |> json.to_string
    
    Purchase(user_id, amount) ->
      json.object([
        #("type", json.string("purchase")),
        #("user_id", json.string(user_id)),
        #("amount", json.float(amount))
      ]) |> json.to_string
  }
}

pub fn decode_user_event(json_string: String) -> Result(UserEvent, String) {
  // Your custom decoder logic here
  json.decode(json_string, dynamic.decode3(
    PageView,
    dynamic.field("user_id", decode.string),
    dynamic.field("page", decode.string),
    dynamic.field("type", decode.string)
  ))
}

// Handler function for PubSub messages
pub fn handle_user_events(json_message: String) -> Nil {
  case decode_user_event(json_message) {
    Ok(PageView(user_id, page)) -> {
      io.println("User " <> user_id <> " viewed " <> page)
    }
    Ok(Purchase(user_id, amount)) -> {
      io.println("User " <> user_id <> " purchased $" <> float.to_string(amount))
    }
    Error(_) -> {
      io.println("Failed to decode user event")
    }
  }
}

pub fn pubsub_example() {
  // Start a phantom-typed PubSub system
  let assert Ok(_pubsub: glixir.PubSub(UserEvent)) = 
    glixir.pubsub_start(atom.create("user_events"))
  
  // Subscribe with your handler function
  let assert Ok(_) = glixir.pubsub_subscribe(
    atom.create("user_events"),
    "user:metrics",
    "my_app",  // Your module name
    "handle_user_events"  // Your handler function
  )
  
  // Broadcast type-safe events from anywhere
  let page_event = PageView("user_123", "/dashboard")
  let assert Ok(_) = glixir.pubsub_broadcast(
    atom.create("user_events"),
    "user:metrics", 
    page_event,
    encode_user_event  // Your encoder
  )
  
  let purchase_event = Purchase("user_123", 99.99)
  let assert Ok(_) = glixir.pubsub_broadcast(
    atom.create("user_events"),
    "user:metrics",
    purchase_event,
    encode_user_event
  )
  
  // Cleanup when done
  let assert Ok(_) = glixir.pubsub_unsubscribe(
    atom.create("user_events"), 
    "user:metrics"
  )
}

// Perfect for metric actors that need their own identity
pub fn handle_metric_update(actor_id: String, json_message: String) -> Nil {
  case decode_metric_message(json_message) {
    Ok(metric) -> {
      // Update this specific actor's metrics
      io.println("Actor " <> actor_id <> " received metric: " <> metric.name)
      // ... update actor state
    }
    Error(_) -> {
      io.println("Invalid metric for actor: " <> actor_id)
    }
  }
}

// Single handler can process multiple message types!
pub fn handle_all_events(json_message: String) -> Nil {
  case decode_user_event(json_message) {
    Ok(user_event) -> handle_user_event(user_event)
    Error(_) -> 
      case decode_system_event(json_message) {
        Ok(system_event) -> handle_system_event(system_event)
        Error(_) -> io.println("Unknown message type: " <> json_message)
      }
  }
}

import glixir
import gleam/dynamic/decode
import gleam/erlang/atom

pub fn main() {
  // Start an agent with initial state (Agent(Int))
  let assert Ok(counter) = glixir.start_agent(fn() { 42 })

  // Get state with a decoder (type safe!)
  let assert Ok(value) = glixir.get_agent(counter, fn(x) { x }, decode.int)
  io.debug(value)  // 42

  // Update state
  let assert Ok(_) = glixir.update_agent(counter, fn(n) { n + 10 })

  // Get and update in one operation
  let assert Ok(old_value) = glixir.get_and_update_agent(
    counter,
    fn(n) { #(n, n * 2) },
    decode.int
  )
  io.debug(old_value)  // 52

  // Stop the agent with a reason (now explicit)
  let assert Ok(_) = glixir.stop_agent(counter, atom.create("normal"))
}

import glixir
import gleam/erlang/process
import gleam/erlang/atom
import gleam/dynamic
import gleam/io

pub type UserMessage {
  RecordMetric(name: String, value: Float)
  GetStats
}

pub fn actor_discovery_example() {
  // Start phantom-typed registry for actor lookup
  let assert Ok(_registry: glixir.Registry(atom.Atom, UserMessage)) = 
    glixir.start_registry(atom.create("user_actors"))
  
  // Start a type-safe dynamic supervisor
  let assert Ok(supervisor) = glixir.start_dynamic_supervisor_named(
    atom.create("user_supervisor")
  )
  
  // String encoder for user ID args
  fn user_id_encode(user_id: String) -> List(dynamic.Dynamic) {
    [dynamic.string(user_id)]
  }
  
  // Simple decoder for replies
  fn simple_decode(_d: dynamic.Dynamic) -> Result(String, String) {
    Ok("started")
  }
  
  // Create a type-safe child spec
  let user_spec = glixir.child_spec(
    id: "user_123",
    module: "MyApp.UserActor", 
    function: "start_link",
    args: "user_123",  // Typed as String!
    restart: glixir.permanent,
    shutdown_timeout: 5000,
    child_type: glixir.worker,
    encode: user_id_encode,
  )
  
  // Start the child with compile-time type safety
  case glixir.start_dynamic_child(supervisor, user_spec, user_id_encode, simple_decode) {
    glixir.ChildStarted(_user_pid, _reply) -> {
      // Actor registers itself in the registry with typed key
      let user_subject = process.new_subject()
      let assert Ok(_) = glixir.register_subject(
        atom.create("user_actors"), 
        atom.create("user_123"),  // Typed Atom key
        user_subject,
        glixir.atom_key_encoder   // Required encoder
      )
      
      // Later... find and message the actor from anywhere!
      case glixir.lookup_subject(
        atom.create("user_actors"), 
        atom.create("user_123"),
        glixir.atom_key_encoder
      ) {
        Ok(subject) -> {
          process.send(subject, RecordMetric("page_views", 1.0))
          io.println("Metric sent to user actor! 📊")
        }
        Error(_) -> io.println("User actor not found")
      }
    }
    glixir.StartChildError(error) -> {
      io.println("Failed to start user actor: " <> error)
    }
  }
}

import glixir
import gleam/erlang/atom
import gleam/dynamic
import gleam/dynamic/decode

// Example: Counter server (request type is atom, reply type is Int)
pub fn main() {
  let assert Ok(server) = glixir.start_genserver("MyApp.Counter", dynamic.int(0))
  let assert Ok(count) = glixir.call_genserver(server, atom.create("get_count"), decode.int)
  io.debug(count)  // 0

  let assert Ok(_) = glixir.cast_genserver(server, atom.create("increment"))

  let assert Ok(count) = glixir.call_genserver(server, atom.create("get_count"), decode.int)
  io.debug(count)  // 1
}