gemqtt

//// gemqtt provides an MQTT client for Gleam projects running on the BEAM.  It
//// does so by wrapping the [emqtt](https://github.com/emqx/emqtt) library
//// written in Erlang.
////
//// The `gemqtt` package contains functions for configuring and controlling
//// the client process.  See the `publisher` and `subscriber` packages for
//// working with MQTT messages.
////
//// Example client setup:
////
////     let assert Ok(client) =
////       gemqtt.new("test.mosquitto.org")
////       |> gemqtt.set_auth("user", "password")
////       |> gemqtt.start_link
////
////     let assert Ok(_) = gemqtt.connect(client)

import gleam/dict.{type Dict}
import gleam/dynamic.{type Dynamic}
import gleam/erlang/atom.{type Atom}
import gleam/erlang/charlist
import gleam/erlang/process
import gleam/option.{type Option}

/// Errors that can occur when working with MQTT connections.
///
pub type Error {
  AlreadyStarted(process.Pid)
  BadProperty(Option(Atom))
  Noproc

  // Network errors; https://www.erlang.org/doc/man/inet#type-posix
  Closed
  Timeout
  Eaddrinuse
  Eaddrnotavail
  Eafnosupport
  Ealready
  Econnaborted
  Econnrefused
  Econnreset
  Edestaddrreq
  Ehostdown
  Ehostunreach
  Einprogress
  Eisconn
  Emsgsize
  Enetdown
  Enetunreach
  Enopkg
  Enoprotoopt
  Enotconn
  Enotty
  Enotsock
  Eproto
  Eprotonosupport
  Eprototype
  Esocktnosupport
  Etimedout
  Ewouldblock
  Exbadport
  Exbadseq
  Nxdomain

  // TLS errors; https://www.erlang.org/doc/man/ssl#type-reason
  AccessDenied
  BadCertificate
  BadCertificateHashValue
  BadCertificateStatusResponse
  BadRecordMac
  CertificateExpired
  CertificateRevoked
  CertificateUnknown
  CertificateUnobtainable
  CloseNotify
  DecodeError
  DecryptError
  ExportRestriction
  HandshakeFailure
  IllegalParameter
  InappropriateFallback
  InsufficientSecurity
  InternalError
  NoApplicationProtocol
  NoRenegotiation
  ProtocolVersion
  RecordOverflow
  UnexpectedMessage
  UnknownCa
  UnknownPskIdentity
  UnrecognizedName
  UnsupportedCertificate
  UnsupportedExtension
  UserCanceled
}

/// MQTT connection option holder. Use `new` to create it, followed by the set
/// functions in this module to configure it, finally passing it to
/// `start_link` to create a Client.
///
pub opaque type Options {
  Options(
    options: Dict(Atom, Dynamic),
    tls_options: Dict(Atom, Dynamic),
    properties: Properties,
  )
}

/// _emqtt_ properties holder. Properties can control advanced MQTT protocol
/// features; see the documentation on `set_property` in each of the `gemqtt`,
/// `publisher`, and `subscriber` modules.
///
/// There is also information in the [emqtt
/// properties](https://github.com/emqx/emqtt?tab=readme-ov-file#properties)
/// docs for more information.
///
pub type Properties {
  Properties(Dict(Atom, Dynamic))
}

/// MQTT Quality of Service level, controlling how many times a message
/// may be delivered.
///
pub type Qos {
  AtMostOnce
  AtLeastOnce
  ExactlyOnce
}

// TODO Missing options
// TcpOpts
// WsPath
// BridgeMode
// ProtoVer
// Keepalive
// MaxInflight
// RetryInterval
// WillTopic
// WillPayload
// WillRetain
// WillQos
// WillProps
// AckTimeout
// ForcePing

/// Creates a new `Options` holder for the specified MQTT server.  Do not
/// include a port number in the `host` string, instead use `set_port`.
///
pub fn new(host: String) -> Options {
  let host_value =
    host
    |> charlist.from_string
    |> dynamic.from

  let opts =
    dict.new()
    |> dict.insert(atom.create_from_string("host"), host_value)

  Options(
    options: opts,
    tls_options: dict.new(),
    properties: Properties(dict.new()),
  )
}

/// Specifies authentication credentials to use with the MQTT server.
///
pub fn set_auth(opts: Options, username: String, password: String) -> Options {
  opts
  |> set_option(atom.create_from_string("username"), username)
  |> set_option(atom.create_from_string("password"), password)
}

/// If true (the default), the client process will automatically send an
/// acknowledgement (ie PUBACK) when it receives a packet from the server.
///
pub fn set_auto_ack(opts: Options, ack: Bool) -> Options {
  set_option(opts, atom.create_from_string("auto_ack"), ack)
}

/// Controls whether the server should discard any existing session and start a
/// new session, can prevent receiving old retained messages. Defaults to true.
///
pub fn set_clean_start(opts: Options, clean: Bool) -> Options {
  set_option(opts, atom.create_from_string("clean_start"), clean)
}

/// Specifies the client identifier. If not given, the client identifier be
/// assigned by the server in MQTT v5 or be automatically generated by _emqtt_
/// in MQTT v3.1/v3.1.1.
///
pub fn set_client_id(opts: Options, id: String) -> Options {
  set_option(opts, atom.create_from_string("clientid"), id)
}

/// Sets the maximum time to wait to connect to the server and the server
/// returns a CONNACK. Defaults to 60s.
///
pub fn set_connect_timeout(opts: Options, seconds timeout: Int) -> Options {
  set_option(opts, atom.create_from_string("connect_timeout"), timeout)
}

/// Sets the Erlang server name for the client process.
///
pub fn set_name(opts: Options, name: String) -> Options {
  set_option(
    opts,
    atom.create_from_string("name"),
    atom.create_from_string(name),
  )
}

/// Specifies the owner, to which the client process will send messages like
/// `{disconnected, ReasonCode, Properties}`.
///
pub fn set_owner(opts: Options, pid: process.Pid) -> Options {
  set_option(opts, atom.create_from_string("owner"), pid)
}

/// Specifies port of the MQTT server to be connected. If not given, the
/// default of 1883 for MQTT or 8883 for MQTT over TLS will be used.
///
pub fn set_port(opts: Options, port: Int) -> Options {
  set_option(opts, atom.create_from_string("port"), port)
}

/// Enables SSL/TLS transport encryption. Defaults to false.
///
pub fn set_tls_enabled(opts: Options, tls: Bool) -> Options {
  set_option(opts, atom.create_from_string("ssl"), tls)
}

/// Primitive function for setting TLS related options. Needs improvement.
///
/// Example usage:
///
///     fn main() {
///       gemqtt.new("mqtt.example.com")
///       |> gemqtt.set_tls_enabled(True)
///       |> gemqtt.set_tls_opt(
///         "verify",
///         dynamic.from(atom.create_from_string("verify_peer")),
///       )
///       |> gemqtt.set_tls_opt(
///         "server_name_indication",
///         dynamic.from(atom.create_from_string("disable")),
///       )
///       |> gemqtt.set_tls_opt("cacerts", cacerts_get())
///     }
///
///     // Returns the OS cacerts.
///     @external(erlang, "public_key", "cacerts_get")
///     fn cacerts_get() -> Dynamic
///
pub fn set_tls_opt(opts: Options, name: String, value: Dynamic) -> Options {
  Options(
    ..opts,
    tls_options: dict.insert(
      opts.tls_options,
      atom.create_from_string(name),
      dynamic.from(value),
    ),
  )
}

/// Sets an MQTT CONNECT packet property.  Please note that properties are
/// validated when `start_link` is called; if _emqtt_ detects errors it will
/// cause your process to fail rather than return an Error.
///
/// Available connect properties:
///
/// - `Session-Expiry-Interval`: Four byte integer
/// - `Receive-Maximum`: Two byte integer
/// - `Maximum-Packet-Size`: Four byte integer
/// - `Topic-Alias-Maximum`: Two byte integer
/// - `Request-Response-Information`: Byte
/// - `Request-Problem-Information`: Byte
/// - `User-Property`: UTF-8 string pair
/// - `Authentication-Method`: UTF-8 encoded string
/// - `Authentication-Data`: Binary data
///
/// Example usage:
///
/// ```
/// gemqtt.new("localhost")
/// |> gemqtt.set_property("Maximum-Packet-Size", 300)
/// |> gemqtt.set_property("User-Property", #("prop-name", "prop-value"))
/// ```
///
pub fn set_property(opts: Options, name: String, value: t) -> Options {
  let Properties(props) = opts.properties
  Options(
    ..opts,
    properties: Properties(dict.insert(
      props,
      atom.create_from_string(name),
      dynamic.from(value),
    )),
  )
}

fn set_option(opts: Options, name: atom.Atom, value: t) -> Options {
  Options(..opts, options: dict.insert(opts.options, name, dynamic.from(value)))
}

/// Client holds the process maintaining the MQTT connection, and is required
/// to publish and subsribe to messages.
///
pub opaque type Client {
  Client(process.Pid)
}

/// Returns the Erlang Pid of the provided Client.
///
pub fn pid_of(client: Client) -> process.Pid {
  let Client(pid) = client
  pid
}

/// Configures a client process and link it to ours.  Does not attempt to
/// connect to the MQTT server.
///
pub fn start_link(opts: Options) -> Result(Client, Error) {
  let Options(
    options: options,
    tls_options: tls_options,
    properties: Properties(properties),
  ) = opts

  let options =
    options
    |> dict.insert(
      atom.create_from_string("properties"),
      dynamic.from(properties),
    )
    |> dict.insert(
      atom.create_from_string("ssl_opts"),
      dynamic.from(dict.to_list(tls_options)),
    )
  start_link_(options)
}

@external(erlang, "emqtt_ffi", "start_link")
fn start_link_(opts: Dict(Atom, Dynamic)) -> Result(Client, Error)

/// Connects to the configured MQTT server.
///
@external(erlang, "emqtt_ffi", "connect")
pub fn connect(client: Client) -> Result(Nil, Error)

/// Sends a DISCONNECT packet to the server, and closes the connection. The
/// Client process will then exit.
///
@external(erlang, "emqtt_ffi", "disconnect")
pub fn disconnect(client: Client) -> Result(Nil, Error)

/// Stops the client process, prefer `disconnect` over this.
///
@external(erlang, "emqtt_ffi", "stop")
pub fn stop(client: Client) -> Result(Nil, Error)

/// Lists this clients active MQTT subscriptions.
///
@external(erlang, "emqtt_ffi", "subscriptions")
pub fn subscriptions(client: Client) -> List(#(String, Dict(Atom, Dynamic)))