use crate::http::{Endpoints, Error};
use crate::ids::{ChannelId, ChatroomId, UserId};
use crate::{content, events, public_api, realtime, website};
use serde::Serialize;
use serde_json::{Map, Value, json};
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[non_exhaustive]
pub struct DispatchError {
pub kind: &'static str,
pub status: Option<u16>,
pub message: String,
}
impl DispatchError {
fn call(message: impl Into<String>) -> Self {
Self { kind: "call", status: None, message: message.into() }
}
}
impl From<Error> for DispatchError {
fn from(e: Error) -> Self {
let kind = match &e {
Error::Status { .. } => "status",
Error::Json(_) => "json",
Error::Invalid(_) => "invalid",
_ => "shape",
};
Self { kind, status: e.status(), message: e.to_string() }
}
}
type Out = Result<Value, DispatchError>;
struct Args<'a>(&'a Map<String, Value>);
impl Args<'_> {
fn value(&self, key: &str) -> Result<&Value, DispatchError> {
self.0.get(key).filter(|v| !v.is_null()).ok_or_else(|| DispatchError::call(format!("missing `{key}`")))
}
fn str(&self, key: &str) -> Result<&str, DispatchError> {
self.value(key)?.as_str().ok_or_else(|| DispatchError::call(format!("`{key}` must be a string")))
}
fn opt_str(&self, key: &str) -> Option<&str> {
self.0.get(key).and_then(Value::as_str)
}
fn u64(&self, key: &str) -> Result<u64, DispatchError> {
let v = self.value(key)?;
v.as_u64()
.or_else(|| v.as_f64().filter(|f| *f >= 0.0 && f.fract() == 0.0).map(|f| f as u64))
.ok_or_else(|| DispatchError::call(format!("`{key}` must be a non-negative integer")))
}
fn opt_u64(&self, key: &str) -> Result<Option<u64>, DispatchError> {
match self.0.get(key) {
None | Some(Value::Null) => Ok(None),
Some(_) => self.u64(key).map(Some),
}
}
fn u16(&self, key: &str) -> Result<u16, DispatchError> {
u16::try_from(self.u64(key)?).map_err(|_| DispatchError::call(format!("`{key}` is out of range")))
}
fn u32(&self, key: &str) -> Result<u32, DispatchError> {
u32::try_from(self.u64(key)?).map_err(|_| DispatchError::call(format!("`{key}` is out of range")))
}
fn bool(&self, key: &str) -> bool {
self.0.get(key).and_then(Value::as_bool).unwrap_or(false)
}
fn strings(&self, key: &str) -> Result<Vec<String>, DispatchError> {
self.value(key)?
.as_array()
.and_then(|a| a.iter().map(|x| x.as_str().map(str::to_owned)).collect())
.ok_or_else(|| DispatchError::call(format!("`{key}` must be a list of strings")))
}
fn ids(&self, key: &str) -> Result<Vec<u64>, DispatchError> {
self.value(key)?
.as_array()
.and_then(|a| a.iter().map(Value::as_u64).collect())
.ok_or_else(|| DispatchError::call(format!("`{key}` must be a list of integers")))
}
fn endpoints(&self) -> Result<Endpoints, DispatchError> {
let mut base = serde_json::to_value(Endpoints::default()).unwrap_or_default();
if let (Some(b), Some(Value::Object(given))) = (base.as_object_mut(), self.0.get("endpoints")) {
for (k, v) in given {
b.insert(k.clone(), v.clone());
}
}
serde_json::from_value(base).map_err(|e| DispatchError::call(format!("`endpoints`: {e}")))
}
fn status_body(&self) -> Result<(u16, &str), DispatchError> {
Ok((self.u16("status")?, self.str("body")?))
}
}
fn to<T: Serialize>(v: T) -> Out {
serde_json::to_value(v).map_err(|e| DispatchError::call(e.to_string()))
}
fn ok<T: Serialize>(r: Result<T, Error>) -> Out {
to(r?)
}
pub fn call(name: &str, args: &Value) -> Out {
let empty = Map::new();
let a = Args(match args {
Value::Object(map) => map,
Value::Null => &empty,
_ => return Err(DispatchError::call("arguments must be an object")),
});
match name {
"events.parse" => to(events::parse(a.str("name")?, a.value("data").unwrap_or(&Value::Null))),
"content.parse" => to(content::parse(a.str("text")?, a.bool("combine"))),
"content.emote_url" => to(content::emote_url(a.str("id")?)),
"realtime.chat_channels" => {
to(realtime::chat_channels(ChatroomId(a.u64("chatroom_id")?), a.opt_u64("channel_id")?.map(ChannelId)))
}
"realtime.chat_connection_request" => {
to(realtime::chat_connection_request(&a.endpoints()?, ChannelId(a.u64("channel_id")?), a.str("client_id")?))
}
"realtime.connection_request" => to(realtime::connection_request(&a.endpoints()?, a.str("client_id")?)),
"realtime.parse_connection" => {
let (s, b) = a.status_body()?;
ok(realtime::parse_connection(s, b))
}
"realtime.token_request" => to(realtime::token_request(&a.endpoints()?, a.str("client_id")?)),
"realtime.parse_token" => {
let (s, b) = a.status_body()?;
ok(realtime::parse_token(s, b))
}
"public_api.token_request" => {
to(public_api::token_request(&a.endpoints()?, a.str("client_id")?, a.str("client_secret")?))
}
"public_api.parse_token" => {
let (s, b) = a.status_body()?;
ok(public_api::parse_token(s, b))
}
"public_api.livestreams_requests" => {
let ids: Vec<UserId> = a.ids("user_ids")?.into_iter().map(UserId).collect();
to(public_api::livestreams_requests(&a.endpoints()?, a.str("token")?, &ids))
}
"public_api.top_livestreams_request" => {
let sort = match a.opt_str("sort").unwrap_or("viewer_count") {
"viewer_count" => public_api::Sort::ViewerCount,
"started_at" => public_api::Sort::StartedAt,
other => return Err(DispatchError::call(format!("unknown sort `{other}`"))),
};
ok(public_api::top_livestreams_request(
&a.endpoints()?,
a.str("token")?,
a.opt_u64("category_id")?,
a.opt_str("language"),
sort,
a.opt_u64("limit")?.unwrap_or(100) as u32,
))
}
"public_api.parse_livestreams" => {
let (s, b) = a.status_body()?;
ok(public_api::parse_livestreams(s, b))
}
"public_api.channels_by_ids_request" => {
let ids: Vec<UserId> = a.ids("user_ids")?.into_iter().map(UserId).collect();
ok(public_api::channels_by_ids_request(&a.endpoints()?, a.str("token")?, &ids))
}
"public_api.channels_by_slugs_request" => {
let slugs = a.strings("slugs")?;
let refs: Vec<&str> = slugs.iter().map(String::as_str).collect();
ok(public_api::channels_by_slugs_request(&a.endpoints()?, a.str("token")?, &refs))
}
"public_api.parse_channels" => {
let (s, b) = a.status_body()?;
ok(public_api::parse_channels(s, b))
}
"public_api.public_key_request" => to(public_api::public_key_request(&a.endpoints()?)),
"public_api.parse_public_key" => {
let (s, b) = a.status_body()?;
ok(public_api::parse_public_key(s, b))
}
"public_api.subscribe_request" => {
let list = a.value("events")?.as_array().ok_or_else(|| DispatchError::call("`events` must be a list"))?;
let events: Vec<(String, u32)> = list
.iter()
.map(|e| {
let name = e.get("name").and_then(Value::as_str);
let version = e.get("version").and_then(Value::as_u64).unwrap_or(1) as u32;
name.map(|n| (n.to_owned(), version))
})
.collect::<Option<_>>()
.ok_or_else(|| DispatchError::call("`events` must be [{name, version}]"))?;
let refs: Vec<(&str, u32)> = events.iter().map(|(n, v)| (n.as_str(), *v)).collect();
ok(public_api::subscribe_request(
&a.endpoints()?,
a.str("token")?,
UserId(a.u64("broadcaster_user_id")?),
&refs,
))
}
"public_api.subscriptions_request" => to(public_api::subscriptions_request(&a.endpoints()?, a.str("token")?)),
"public_api.parse_subscriptions" => {
let (s, b) = a.status_body()?;
ok(public_api::parse_subscriptions(s, b))
}
"public_api.unsubscribe_requests" => {
let ids = a.strings("ids")?;
let refs: Vec<&str> = ids.iter().map(String::as_str).collect();
to(public_api::unsubscribe_requests(&a.endpoints()?, a.str("token")?, &refs))
}
"website.channel_request" => to(website::channel_request(&a.endpoints()?, a.str("slug")?)),
"website.parse_channel" => {
let (s, b) = a.status_body()?;
ok(website::parse_channel(s, b))
}
"website.history_request" => to(website::history_request(&a.endpoints()?, ChannelId(a.u64("channel_id")?))),
"website.parse_history" => {
let (s, b) = a.status_body()?;
ok(website::parse_history(s, b))
}
"website.channel_user_request" => {
to(website::channel_user_request(&a.endpoints()?, a.str("channel_slug")?, a.str("username")?))
}
"website.parse_channel_user" => {
let (s, b) = a.status_body()?;
ok(website::parse_channel_user(s, b))
}
"website.me_request" => to(website::me_request(&a.endpoints()?, a.str("session")?, a.str("channel_slug")?)),
"website.parse_me" => {
let (s, b) = a.status_body()?;
ok(website::parse_me(s, b))
}
"website.ban_request" => {
to(website::ban_request(&a.endpoints()?, a.str("session")?, a.str("channel_slug")?, a.str("username")?))
}
"website.timeout_request" => ok(website::timeout_request(
&a.endpoints()?,
a.str("session")?,
a.str("channel_slug")?,
a.str("username")?,
a.u32("minutes")?,
)),
"website.unban_request" => {
to(website::unban_request(&a.endpoints()?, a.str("session")?, a.str("channel_slug")?, a.str("username")?))
}
"website.delete_message_request" => to(website::delete_message_request(
&a.endpoints()?,
a.str("session")?,
ChatroomId(a.u64("chatroom_id")?),
a.str("message_id")?,
)),
"website.pin_request" => ok(website::pin_request(
&a.endpoints()?,
a.str("session")?,
a.str("channel_slug")?,
a.value("message")?,
a.u32("minutes")?,
)),
"website.unpin_request" => {
to(website::unpin_request(&a.endpoints()?, a.str("session")?, a.str("channel_slug")?))
}
"website.send_request" => {
let reply = match a.0.get("reply_to") {
None | Some(Value::Null) => None,
Some(r) => {
Some(Args(r.as_object().ok_or_else(|| DispatchError::call("`reply_to` must be an object"))?))
}
};
let target = match &reply {
None => None,
Some(r) => Some(website::ReplyTarget {
message_id: r.str("message_id")?,
content: r.opt_str("content").unwrap_or(""),
sender_id: r.u64("sender_id")?,
sender_username: r.str("sender_username")?,
}),
};
ok(website::send_request(
&a.endpoints()?,
a.str("session")?,
ChatroomId(a.u64("chatroom_id")?),
a.str("content")?,
a.str("message_ref")?,
target.as_ref(),
))
}
"website.parse_action" => {
let (s, b) = a.status_body()?;
ok(website::parse_action(s, b))
}
#[cfg(feature = "webhooks")]
"webhooks.verify" => webhook_verify(&a),
#[cfg(feature = "webhooks")]
"webhooks.parse" => crate::webhooks::parse(a.str("event_type")?, a.str("body")?)
.map_err(|e| DispatchError { kind: "json", status: None, message: e.to_string() })
.and_then(to),
other => Err(DispatchError::call(format!("no function `{other}`"))),
}
}
#[cfg(feature = "webhooks")]
fn webhook_verify(a: &Args<'_>) -> Out {
use crate::webhooks;
let pairs: Vec<(String, String)> = match a.value("headers")? {
Value::Array(list) => {
list.iter().filter_map(|p| Some((p.get(0)?.as_str()?.to_owned(), p.get(1)?.as_str()?.to_owned()))).collect()
}
Value::Object(map) => map.iter().filter_map(|(k, v)| Some((k.clone(), v.as_str()?.to_owned()))).collect(),
_ => return Err(DispatchError::call("`headers` must be [[name, value]] or an object")),
};
let refuse = |e: String| DispatchError { kind: "webhook", status: None, message: e };
let headers = webhooks::Headers::from_pairs(pairs.iter().map(|(k, v)| (k.as_str(), v.as_str())))
.map_err(|e| refuse(e.to_string()))?;
let body = a.str("body")?;
webhooks::verify(a.str("public_key_pem")?, &headers, body.as_bytes()).map_err(|e| refuse(e.to_string()))?;
let event = webhooks::parse(&headers.event_type, body).map_err(|e| refuse(e.to_string()))?;
Ok(json!({ "headers": to(&headers)?, "event": to(event)? }))
}
#[derive(Debug, Clone)]
pub struct SessionHandle(realtime::Session);
impl SessionHandle {
pub fn new(channels: Vec<String>) -> Self {
Self(realtime::Session::new(realtime::SessionOptions::new(channels)))
}
pub fn call(&mut self, method: &str, args: &Value) -> Out {
let empty = Map::new();
let a = Args(args.as_object().unwrap_or(&empty));
let s = &mut self.0;
match method {
"open" => to(s.open(a.str("token")?, a.u64("now")?)),
"receive" => to(s.receive(a.str("text")?, a.u64("now")?)),
"refresh" => to(s.refresh(a.str("token")?, a.u64("now")?)),
"tick" => to(s.tick(a.u64("now")?)),
"closed" => to(s.closed()),
"deadline" => to(s.deadline()),
"is_subscribed" => to(s.is_subscribed(a.str("channel")?)),
"is_connected" => to(s.is_connected()),
other => Err(DispatchError::call(format!("no session method `{other}`"))),
}
}
}
pub const FUNCTIONS: &[&str] = &[
"events.parse",
"content.parse",
"content.emote_url",
"realtime.chat_channels",
"realtime.chat_connection_request",
"realtime.connection_request",
"realtime.parse_connection",
"realtime.token_request",
"realtime.parse_token",
"public_api.token_request",
"public_api.parse_token",
"public_api.livestreams_requests",
"public_api.top_livestreams_request",
"public_api.parse_livestreams",
"public_api.channels_by_ids_request",
"public_api.channels_by_slugs_request",
"public_api.parse_channels",
"public_api.public_key_request",
"public_api.parse_public_key",
"public_api.subscribe_request",
"public_api.subscriptions_request",
"public_api.parse_subscriptions",
"public_api.unsubscribe_requests",
"website.channel_request",
"website.parse_channel",
"website.history_request",
"website.parse_history",
"website.channel_user_request",
"website.parse_channel_user",
"website.me_request",
"website.parse_me",
"website.ban_request",
"website.timeout_request",
"website.unban_request",
"website.delete_message_request",
"website.pin_request",
"website.unpin_request",
"website.send_request",
"website.parse_action",
"webhooks.verify",
"webhooks.parse",
];
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn every_listed_function_is_dispatched() {
for name in FUNCTIONS {
let e = call(name, &json!({})).err();
assert!(
!matches!(&e, Some(DispatchError { message, .. }) if message.starts_with("no function")),
"{name} is listed but not dispatched"
);
}
assert_eq!(call("nope", &Value::Null).unwrap_err().message, "no function `nope`");
}
#[test]
fn errors_keep_kicks_status() {
let e = call("website.parse_channel", &json!({"status": 404, "body": "{}"})).unwrap_err();
assert_eq!((e.kind, e.status), ("status", Some(404)));
let e = call("website.channel_request", &json!({})).unwrap_err();
assert_eq!((e.kind, e.message.as_str()), ("call", "missing `slug`"));
}
#[test]
fn requests_and_partial_endpoints() {
let r =
call("realtime.token_request", &json!({"client_id": "c", "endpoints": {"realtime": "http://x"}})).unwrap();
assert_eq!(r["url"], "http://x/auth/connection");
assert_eq!(r["method"], "POST");
}
#[test]
fn a_session_by_name() {
let mut s = SessionHandle::new(vec!["a".into()]);
let out = s.call("open", &json!({"token": "t", "now": 0})).unwrap();
assert_eq!(out[0]["action"], "send");
assert_eq!(s.call("is_connected", &Value::Null).unwrap(), json!(false));
assert_eq!(s.call("deadline", &Value::Null).unwrap(), json!(45_000));
}
}