majjo/stoatchat
1
0
Fork 0
mirror of https://github.com/stoatchat/stoatchat.git synced 2026-08-31 00:10:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: broadcast events

Browse Source 
Signed-off-by: Zomatree <me@zomatree.live>
This commit is contained in:
Zomatree
2026-08-03 04:13:56 +01:00
parent 7802e9561b
commit a41b291504
13 changed files with 70 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
crates/core/database/src/amqp/amqp.rs
View File

@@ -413,4 +413,36 @@ impl AMQP {
Ok(())
}
pub async fn publish_event_broadcast(
&self,
channels: Vec<String>,
event: &EventV1,
) -> Result<(), AMQPError> {
let mut headers = FieldTable::default();
headers.insert(
"c".into(),
AMQPValue::FieldArray(
channels
.into_iter()
.map(|c| AMQPValue::LongString(c.into()))
.collect::<Vec<_>>()
.into(),
),
);
let config = config().await;
self.publish_event
.basic_publish(
config.rabbit.default_exchange.clone().into(),
config.rabbit.queues.events.into(),
BasicPublishOptions::default(),
&serde_json::to_vec(event).unwrap(),
BasicProperties::default().with_headers(headers),
)
.await?;
Ok(())
}
}

23
crates/core/database/src/events/client.rs
View File

@@ -387,7 +387,6 @@ pub enum EventV1 {
impl EventV1 {
/// Publish helper wrapper
pub async fn p(self, channel: String) {
//redis_kiss::p(channel.clone(), &self).await;
#[cfg(debug_assertions)]
info!("Publishing event to {channel}: {self:?}");
@@ -400,19 +399,17 @@ impl EventV1 {
};
}
/// Publish user event
pub async fn p_user(self, id: String, db: &Database) {
self.clone().p(id.clone()).await;
}
pub async fn p_broadcast(self, channels: Vec<String>) {
#[cfg(debug_assertions)]
info!("Broadcasting event to channels: {channels:?}: {self:?}");
/// Publish private event
pub async fn private(self, id: String) {
self.p(format!("{id}")).await;
}
/// Publish server member event
pub async fn server(self, id: String) {
self.p(format!("{id}")).await;
if let Err(e) = get_amqp().publish_event_broadcast(channels, &self).await {
if cfg!(debug_assertions) {
panic!("{e:?}");
} else {
log::error!("{e:?}");
};
};
}
/// Publish internal global event

2
crates/core/database/src/models/accounts/model.rs
View File

@@ -619,7 +619,7 @@ impl Account {
user_id: self.id.clone(),
exclude_session_id,
}
.private(self.id.clone())
.p(self.id.clone())
.await;
Ok(())

2
crates/core/database/src/models/bots/model.rs
View File

@@ -146,7 +146,7 @@ impl Bot {
db.update_bot(&self.id, &partial, remove).await?;
if partial.token.is_some() {
EventV1::Logout.private(self.id.clone()).await;
EventV1::Logout.p(self.id.clone()).await;
}
self.apply_options(partial);

19
crates/core/database/src/models/channels/model.rs
View File

@@ -295,10 +295,9 @@ impl Channel {
db.insert_channel(&channel).await?;
let event = EventV1::ChannelCreate(channel.clone().into());
for recipient in recipients {
event.clone().private(recipient).await;
}
EventV1::ChannelCreate(channel.clone().into())
.p_broadcast(recipients)
.await;
Ok(channel)
}
@@ -328,9 +327,9 @@ impl Channel {
db.insert_channel(&channel).await?;
if let Channel::DirectMessage { .. } = &channel {
let event = EventV1::ChannelCreate(channel.clone().into());
event.clone().private(user_a.id.clone()).await;
event.private(user_b.id.clone()).await;
EventV1::ChannelCreate(channel.clone().into())
.p_broadcast(vec![user_a.id.clone(), user_b.id.clone()])
.await;
};
Ok(channel)
@@ -392,7 +391,7 @@ impl Channel {
.ok();
EventV1::ChannelCreate(self.clone().into())
.private(user.id.to_string())
.p(user.id.to_string())
.await;
Ok(())
@@ -556,7 +555,7 @@ impl Channel {
slowmode.take();
}
_ => {}
}
},
}
}
@@ -774,7 +773,7 @@ impl Channel {
user: user.to_string(),
message_id: message.to_string(),
}
.private(user.to_string())
.p(user.to_string())
.await;
crate::util::acker::ack_channel(user, self.id(), message, amqp).await

2
crates/core/database/src/models/server_members/model.rs
View File

@@ -185,7 +185,7 @@ impl Member {
emojis: emojis.into_iter().map(|emoji| emoji.into()).collect(),
voice_states,
}
.private(user.id.clone())
.p(user.id.clone())
.await;
if let Some(id) = server

2
crates/core/database/src/models/sessions/model.rs
View File

@@ -60,7 +60,7 @@ impl Session {
user_id: self.user_id.clone(),
session_id: self.id,
}
.private(self.user_id)
.p(self.user_id)
.await;
Ok(())

2
crates/core/database/src/models/user_settings/model.rs
View File

@@ -20,7 +20,7 @@ impl UserSettingsImpl for UserSettings {
id: user.to_string(),
update: self,
}
.private(user.to_string())
.p(user.to_string())
.await;
Ok(())

6
crates/core/database/src/models/users/model.rs
View File

@@ -514,14 +514,14 @@ impl User {
id: target.id.clone(),
user: self.clone().into(db, Some(&*target)).await,
}
.private(target.id.clone())
.p(target.id.clone())
.await;
EventV1::UserRelationship {
id: self.id.clone(),
user: target.clone().into(db, Some(&*self)).await,
}
.private(self.id.clone())
.p(self.id.clone())
.await;
Ok(())
@@ -680,7 +680,7 @@ impl User {
clear: remove.into_iter().map(|v| v.into()).collect(),
event_id: Some(Ulid::new().to_string()),
}
.p_user(self.id.clone(), db)
.p(self.id.clone())
.await;
Ok(())

2
crates/core/database/src/util/acker.rs
View File

@@ -66,7 +66,7 @@ pub async fn ack_server(user: &User, server: &Server, db: &Database, amqp: &AMQP
user: user.id.clone(),
message_id: channel_last_msg,
}
.private(user.id.clone())
.p(user.id.clone())
.await;
}
}

16
crates/daemons/voice-ingress/src/api.rs
View File

@@ -159,16 +159,14 @@ pub async fn ingress(
.collect()
};
for recipient in call_recipients {
EventV1::VoiceCallUpdate {
initiator_id: user.id.clone(),
channel_id: channel_id.clone(),
started_at: Some(joined_at),
ended: false,
}
.private(recipient)
.await
EventV1::VoiceCallUpdate {
initiator_id: user.id.clone(),
channel_id: channel_id.clone(),
started_at: Some(joined_at),
ended: false,
}
.p_broadcast(call_recipients)
.await
}
if let Err(e) = amqp

4
crates/delta/src/routes/channels/message_send.rs
View File

@@ -111,7 +111,7 @@ pub async fn message_send(
retry_after: ttl as u64,
}],
}
.private(user.id.clone())
.p(user.id.clone())
.await;
return Err(create_error!(InSlowmode {
retry_after: ttl as u64
@@ -125,7 +125,7 @@ pub async fn message_send(
retry_after: *channel_slowmode,
}],
}
.private(user.id.clone())
.p(user.id.clone())
.await;
}
}

2
crates/delta/src/routes/servers/member_edit.rs
View File

@@ -286,7 +286,7 @@ pub async fn edit(
to: new_voice_channel.id().to_string(),
token,
}
.private(target_user.id.clone())
.p(target_user.id.clone())
.await;
};
} else if can_publish.is_some()