Gateway

Cargo.toml

@@ -28,23 +28,35 @@ log4rs = { version = "1.3.0", features = [
 num-bigint = "0.4.5"
 num-traits = "0.2.19"
 openssl = "0.10.64"
-poem = { version = "3.0.1", features = ["websocket"] }
+poem = "3.0.1"
 utoipa = { version = "5.0.0-alpha.0", features = [] }
 rand = "0.8.5"
 regex = "1.10.4"
-reqwest = "0.12.4"
+reqwest = { version = "0.12.5", default-features = false, features = [
+    "http2",
+    "macos-system-configuration",
+    "charset",
+    "rustls-tls-webpki-roots",
+] }
 serde = { version = "1.0.203", features = ["derive"] }
 serde_json = { version = "1.0.117", features = ["raw_value"] }
 sqlx = { version = "0.7.4", features = [
     "json",
     "chrono",
     "ipnetwork",
-    "runtime-tokio-native-tls",
+    "runtime-tokio-rustls",
     "any",
 ] }
 thiserror = "1.0.61"
 tokio = { version = "1.38.0", features = ["full"] }
-sentry = "0.33.0"
+sentry = { version = "0.34.0", default-features = false, features = [
+    "backtrace",
+    "contexts",
+    "debug-images",
+    "panic",
+    "reqwest",
+    "rustls",
+] }
 clap = { version = "4.5.4", features = ["derive"] }
 
 chorus = { git = "http://github.com/polyphony-chat/chorus", rev = "d591616", features = [
@@ -54,3 +66,8 @@ serde_path_to_error = "0.1.16"
 percent-encoding = "2.3.1"
 hex = "0.4.3"
 itertools = "0.13.0"
+tokio-tungstenite = { version = "0.23.1", features = [
+    "rustls-tls-webpki-roots",
+] }
+pubserve = { version = "1.1.0", features = ["async", "send"] }
+parking_lot = { version = "0.12.3", features = ["deadlock_detection"] }

src/api/mod.rs

@@ -1,12 +1,13 @@
 use poem::{
-    EndpointExt,
-    IntoResponse,
     listener::TcpListener,
-    middleware::{NormalizePath, TrailingSlash}, Route, Server, web::Json,
+    middleware::{NormalizePath, TrailingSlash},
+    web::Json,
+    EndpointExt, IntoResponse, Route, Server,
 };
 use serde_json::json;
 use sqlx::MySqlPool;
 
+use crate::SharedEventPublisherMap;
 use crate::{
     api::{
         middleware::{
@@ -21,8 +22,9 @@ use crate::{
 mod middleware;
 mod routes;
 
-pub async fn start_api(db: MySqlPool) -> Result<(), Error> {
+pub async fn start_api(db: MySqlPool, publisher_map: SharedEventPublisherMap) -> Result<(), Error> {
     log::info!(target: "symfonia::api::cfg", "Loading configuration");
+
     let config = Config::init(&db).await?;
 
     if config.sentry.enabled {
@@ -69,6 +71,7 @@ pub async fn start_api(db: MySqlPool) -> Result<(), Error> {
         .nest("/api/v9", routes)
         .data(db)
         .data(config)
+        .data(publisher_map)
         .with(NormalizePath::new(TrailingSlash::Trim))
         .catch_all_error(custom_error);
 

src/api/routes/guilds/id/channels.rs

@@ -1,10 +1,10 @@
 use chorus::types::{
-    ChannelModifySchema, ChannelType, jwt::Claims, ModifyChannelPositionsSchema, Snowflake,
+    jwt::Claims, ChannelModifySchema, ChannelType, ModifyChannelPositionsSchema, Snowflake,
 };
 use poem::{
     handler,
-    IntoResponse,
-    Response, web::{Data, Json, Path},
+    web::{Data, Json, Path},
+    IntoResponse, Response,
 };
 use reqwest::StatusCode;
 use sqlx::MySqlPool;
@@ -139,41 +139,47 @@ mod tests {
                     position: Some(0),
                     ..Default::default()
                 },
+                ..Default::default()
             },
             Channel {
                 inner: chorus::types::Channel {
                     id: Snowflake::default(),
                     position: Some(1),
                     ..Default::default()
                 },
+                ..Default::default()
             },
             Channel {
                 inner: chorus::types::Channel {
                     id: Snowflake::default(),
                     position: Some(2),
                     ..Default::default()
                 },
+                ..Default::default()
             },
             Channel {
                 inner: chorus::types::Channel {
                     id: Snowflake::default(),
                     position: Some(3),
                     ..Default::default()
                 },
+                ..Default::default()
             },
             Channel {
                 inner: chorus::types::Channel {
                     id: Snowflake::default(),
                     position: Some(4),
                     ..Default::default()
                 },
+                ..Default::default()
             },
             Channel {
                 inner: chorus::types::Channel {
                     id: Snowflake::default(),
                     position: Some(5),
                     ..Default::default()
                 },
+                ..Default::default()
             },
         ];
 

src/database/entities/application.rs

@@ -1,12 +1,15 @@
+use super::*;
+
 use std::ops::{Deref, DerefMut};
+use std::sync::Arc;
 
-use bitflags::Flags;
 use chorus::types::{ApplicationFlags, Snowflake};
+use parking_lot::RwLock;
 use serde::{Deserialize, Serialize};
-use sqlx::{FromRow, MySqlPool};
+use sqlx::MySqlPool;
 
 use crate::{
-    database::entities::{Config, user::User},
+    database::entities::{user::User, Config},
     errors::Error,
 };
 
@@ -17,6 +20,9 @@ pub struct Application {
     pub owner_id: Snowflake,
     pub bot_user_id: Option<Snowflake>,
     pub team_id: Option<Snowflake>,
+    #[sqlx(skip)]
+    #[serde(skip)]
+    pub publisher: SharedEventPublisher,
 }
 
 impl Deref for Application {
@@ -62,6 +68,7 @@ impl Application {
             owner_id: owner_id.to_owned(),
             bot_user_id,
             team_id: None,
+            publisher: Arc::new(RwLock::new(pubserve::Publisher::new())),
         };
 
         let _res = sqlx::query("INSERT INTO applications (id, name, summary, hook, bot_public, verify_key, owner_id, flags, integration_public, discoverability_state, discovery_eligibility_flags) VALUES (?, ?, ?, true, true, ?, ?, ?, true, 1, 2240)")

src/database/entities/channel.rs

@@ -1,25 +1,29 @@
-use std::ops::{Deref, DerefMut};
-
+use super::*;
 use chorus::types::{
-    ChannelMessagesAnchor, ChannelModifySchema, ChannelType, CreateChannelInviteSchema, InviteType,
-    MessageSendSchema, PermissionOverwrite, Snowflake,
+    ChannelDelete, ChannelMessagesAnchor, ChannelModifySchema, ChannelType, ChannelUpdate,
+    CreateChannelInviteSchema, InviteType, MessageSendSchema, PermissionOverwrite, Snowflake,
 };
 use itertools::Itertools;
+use pubserve::Publisher;
 use serde::{Deserialize, Serialize};
-use sqlx::{MySqlPool, types::Json};
+use sqlx::{types::Json, MySqlPool};
+use std::ops::{Deref, DerefMut};
 
 use crate::{
     database::entities::{
-        GuildMember, invite::Invite, message::Message, read_state::ReadState, recipient::Recipient,
+        invite::Invite, message::Message, read_state::ReadState, recipient::Recipient, GuildMember,
         User, Webhook,
     },
     errors::{ChannelError, Error, GuildError, UserError},
 };
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, sqlx::FromRow)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, sqlx::FromRow, Default)]
 pub struct Channel {
     #[sqlx(flatten)]
     pub(crate) inner: chorus::types::Channel,
+    #[sqlx(skip)]
+    #[serde(skip)]
+    pub publisher: SharedEventPublisher,
 }
 
 impl Deref for Channel {
@@ -87,6 +91,7 @@ impl Channel {
                 guild_id,
                 ..Default::default()
             },
+            ..Default::default()
         };
 
         sqlx::query("INSERT INTO channels (id, type, name, nsfw, guild_id, parent_id, flags, permission_overwrites, default_thread_rate_limit_per_user, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NOW())")

src/database/entities/guild.rs

@@ -1,11 +1,10 @@
-use std::{
-    ops::{Deref, DerefMut},
-    sync::{Arc, RwLock},
-};
+use super::*;
+
+use std::ops::{Deref, DerefMut};
 
 use chorus::types::{
-    ChannelType, NSFWLevel, PermissionFlags, PremiumTier,
-    PublicUser, Snowflake, SystemChannelFlags, types::guild_configuration::GuildFeaturesList, WelcomeScreenObject,
+    types::guild_configuration::GuildFeaturesList, ChannelType, NSFWLevel, PermissionFlags,
+    PremiumTier, PublicUser, Snowflake, SystemChannelFlags, WelcomeScreenObject,
 };
 use serde::{Deserialize, Serialize};
 use sqlx::{FromRow, MySqlPool, QueryBuilder, Row};
@@ -31,6 +30,9 @@ pub struct Guild {
     pub parent: Option<String>,
     pub template_id: Option<Snowflake>,
     pub nsfw: bool,
+    #[sqlx(skip)]
+    #[serde(skip)]
+    pub publisher: SharedEventPublisher,
 }
 
 impl Deref for Guild {

src/database/entities/mod.rs

@@ -16,6 +16,8 @@ pub use user_settings::*;
 pub use voice_state::*;
 pub use webhook::*;
 
+use crate::SharedEventPublisher;
+
 mod application;
 mod attachment;
 mod audit_log;

src/database/entities/role.rs

@@ -1,3 +1,5 @@
+use super::*;
+
 use std::ops::{Deref, DerefMut};
 
 use chorus::types::{PermissionFlags, Snowflake};
@@ -11,6 +13,9 @@ pub struct Role {
     #[sqlx(flatten)]
     inner: chorus::types::RoleObject,
     pub guild_id: Snowflake,
+    #[sqlx(skip)]
+    #[serde(skip)]
+    pub publisher: SharedEventPublisher,
 }
 
 impl Deref for Role {

src/database/entities/user.rs

@@ -1,3 +1,5 @@
+use super::*;
+
 use std::{
     default::Default,
     ops::{Deref, DerefMut},
@@ -28,6 +30,9 @@ pub struct User {
     #[sqlx(skip)]
     pub settings: UserSettings,
     pub extended_settings: sqlx::types::Json<Value>,
+    #[sqlx(skip)]
+    #[serde(skip)]
+    pub publisher: SharedEventPublisher,
 }
 
 impl Deref for User {

src/gateway/mod.rs

@@ -0,0 +1,20 @@
+mod types;
+
+use log::info;
+use sqlx::MySqlPool;
+pub use types::*;
+
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+use crate::errors::Error;
+use crate::SharedEventPublisherMap;
+
+pub async fn start_gateway(
+    db: MySqlPool,
+    publisher_map: SharedEventPublisherMap,
+) -> Result<(), Error> {
+    info!(target: "symfonia::gateway", "Starting gateway server");
+    // `publishers` will live for the lifetime of the gateway server, in the main gateway thread
+    Ok(())
+}