///|
/// Message delivered by `Client::subscribe`.
///
/// Example:
/// ```moonbit nocheck
/// msg.channel
/// msg.payload.text()
/// ```
pub struct Message {
  /// Channel that produced the message.
  channel : String
  /// Raw message payload.
  payload : &@io.Data
}

///|
/// Message delivered by `Client::psubscribe`.
///
/// Example:
/// ```moonbit nocheck
/// msg.pattern
/// msg.channel
/// msg.payload.text()
/// ```
pub struct PatternMessage {
  /// Pattern that matched the message.
  pattern : String
  /// Concrete channel that produced the message.
  channel : String
  /// Raw message payload.
  payload : &@io.Data
}

///|
priv enum PubSubKind {
  Channel
  Pattern
}

///|
priv enum PubSubEvent {
  ChannelMessage(String, Bytes)
  PatternMessage(String, String, Bytes)
}

///|
fn pubsub_command(
  kind~ : PubSubKind,
  subscribe~ : Bool,
  name~ : String,
) -> ReadOnlyArray[Bytes] {
  let op = match (kind, subscribe) {
    (Channel, true) => b"SUBSCRIBE"
    (Channel, false) => b"UNSUBSCRIBE"
    (Pattern, true) => b"PSUBSCRIBE"
    (Pattern, false) => b"PUNSUBSCRIBE"
  }
  [op, @encoding/utf8.encode(name)]
}

///|
/// Publishes `payload` to `channel` and returns the number of subscribers that
/// received the message.
///
/// Example:
/// ```moonbit nocheck
/// let receivers = client.publish("events", "hello")
/// ```
pub async fn Client::publish(
  self : Client,
  channel : String,
  payload : &@io.Data,
) -> Int {
  let args : Array[Bytes] = [
    b"PUBLISH",
    @encoding/utf8.encode(channel),
    payload.binary(),
  ]
  self.execute(Command(args, value => value.as_int()))
}

///|
/// Subscribes to one channel and invokes `callback` for each received message.
///
/// This is a long-running operation. Run it in its own async task and cancel
/// that task to stop receiving messages.
///
/// Example:
/// ```moonbit nocheck
/// group.spawn(() => {
///   client.subscribe("events", msg => {
///     println(msg.payload.text())
///   })
/// }, allow_failure=true)
/// ```
pub async fn Client::subscribe(
  self : Client,
  channel : String,
  callback : async (Message) -> Unit,
) -> Unit {
  self.subscribe_resp2_channel_loop(channel, callback)
}

///|
/// Pattern-subscribes and invokes `callback` for each matching message.
///
/// This is a long-running operation. Run it in its own async task and cancel
/// that task to stop receiving messages.
///
/// Example:
/// ```moonbit nocheck
/// group.spawn(() => {
///   client.psubscribe("events:*", msg => {
///     println(msg.channel)
///   })
/// }, allow_failure=true)
/// ```
pub async fn Client::psubscribe(
  self : Client,
  pattern : String,
  callback : async (PatternMessage) -> Unit,
) -> Unit {
  self.subscribe_resp2_pattern_loop(pattern, callback)
}

///|
fn parse_pubsub_event(
  values : ReadOnlyArray[@resp.Value],
) -> PubSubEvent? raise {
  guard values.length() > 0 else {
    raise UnexpectedResponse("empty pubsub message")
  }
  let kind = values[0].raw().as_string()
  match kind {
    "message" => {
      guard values.length() == 3 else {
        raise UnexpectedResponse("expected message")
      }
      Some(
        ChannelMessage(values[1].raw().as_string(), values[2].as_pubsub_data()),
      )
    }
    "pmessage" => {
      guard values.length() == 4 else {
        raise UnexpectedResponse("expected pattern message")
      }
      Some(
        PatternMessage(
          values[1].raw().as_string(),
          values[2].raw().as_string(),
          values[3].as_pubsub_data(),
        ),
      )
    }
    "subscribe" | "unsubscribe" | "psubscribe" | "punsubscribe" => None
    _ => None
  }
}

///|
fn @resp.Value::as_pubsub_data(self : @resp.Value) -> Bytes raise {
  match self.raw().desc() {
    String(value, hint=_) => value
    _ => raise UnexpectedResponse("expected pubsub payload")
  }
}