mirror of
https://forgejo.ellis.link/continuwuation/continuwuity.git
synced 2026-11-14 04:37:00 +00:00
perf: Strip unsigned over federation and generate it on the fly locally
This commit is contained in:
14 files changed
+263
-79
No files matched your search
@@ -98,7 +98,17 @@ pub(crate) async fn get_context_route(
|
||||
.pdus_rev(room_id, Some(base_count))
|
||||
.ignore_err()
|
||||
.then(async |mut pdu| {
|
||||
pdu.1.set_unsigned(Some(sender_user));
|
||||
let ctx = services
|
||||
.rooms
|
||||
.timeline
|
||||
.get_unsigned_context(&pdu.1, Some(sender_user))
|
||||
.await;
|
||||
pdu.1.set_unsigned(
|
||||
ctx.user_id,
|
||||
ctx.membership,
|
||||
ctx.prev_content,
|
||||
ctx.redacted_because,
|
||||
);
|
||||
if let Err(e) = services
|
||||
.rooms
|
||||
.pdu_metadata
|
||||
@@ -121,7 +131,17 @@ pub(crate) async fn get_context_route(
|
||||
.pdus(room_id, Some(base_count))
|
||||
.ignore_err()
|
||||
.then(async |mut pdu| {
|
||||
pdu.1.set_unsigned(Some(sender_user));
|
||||
let ctx = services
|
||||
.rooms
|
||||
.timeline
|
||||
.get_unsigned_context(&pdu.1, Some(sender_user))
|
||||
.await;
|
||||
pdu.1.set_unsigned(
|
||||
ctx.user_id,
|
||||
ctx.membership,
|
||||
ctx.prev_content,
|
||||
ctx.redacted_because,
|
||||
);
|
||||
if let Err(e) = services
|
||||
.rooms
|
||||
.pdu_metadata
|
||||
|
||||
@@ -140,7 +140,17 @@ pub(crate) async fn get_message_events_route(
|
||||
.wide_filter_map(|item| visibility_filter(&services, item, sender_user))
|
||||
.take(limit)
|
||||
.then(async |mut pdu| {
|
||||
pdu.1.set_unsigned(Some(sender_user));
|
||||
let ctx = services
|
||||
.rooms
|
||||
.timeline
|
||||
.get_unsigned_context(&pdu.1, Some(sender_user))
|
||||
.await;
|
||||
pdu.1.set_unsigned(
|
||||
ctx.user_id,
|
||||
ctx.membership,
|
||||
ctx.prev_content,
|
||||
ctx.redacted_because,
|
||||
);
|
||||
if let Err(e) = services
|
||||
.rooms
|
||||
.pdu_metadata
|
||||
|
||||
@@ -58,7 +58,12 @@ pub(crate) async fn get_room_event_route(
|
||||
debug_warn!("Failed to add bundled aggregations to event: {e}");
|
||||
}
|
||||
|
||||
event.set_unsigned(Some(body.identity.expect_sender_user()?));
|
||||
let ctx = services
|
||||
.rooms
|
||||
.timeline
|
||||
.get_unsigned_context(&event, Some(sender_user))
|
||||
.await;
|
||||
event.set_unsigned(ctx.user_id, ctx.membership, ctx.prev_content, ctx.redacted_because);
|
||||
|
||||
Ok(get_room_event::v3::Response::new(event.into_format()))
|
||||
}
|
||||
@@ -53,7 +53,17 @@ pub(crate) async fn room_initial_sync_route(
|
||||
.pdus_rev(room_id, None)
|
||||
.try_take(limit)
|
||||
.and_then(async |mut pdu| {
|
||||
pdu.1.set_unsigned(Some(sender_user));
|
||||
let ctx = services
|
||||
.rooms
|
||||
.timeline
|
||||
.get_unsigned_context(&pdu.1, Some(sender_user))
|
||||
.await;
|
||||
pdu.1.set_unsigned(
|
||||
ctx.user_id,
|
||||
ctx.membership,
|
||||
ctx.prev_content,
|
||||
ctx.redacted_because,
|
||||
);
|
||||
if let Err(e) = services
|
||||
.rooms
|
||||
.pdu_metadata
|
||||
|
||||
@@ -7,7 +7,7 @@ use conduwuit::{
|
||||
Event, PduCount, Result, debug_warn, err,
|
||||
matrix::pdu::PduEvent,
|
||||
ref_at, trace,
|
||||
utils::stream::{BroadbandExt, ReadyExt, TryIgnore},
|
||||
utils::stream::{BroadbandExt, ReadyExt, TryIgnore, WidebandExt},
|
||||
};
|
||||
use conduwuit_service::Services;
|
||||
use futures::StreamExt;
|
||||
@@ -83,9 +83,19 @@ async fn load_timeline(
|
||||
.pdus_rev(room_id, ending_count.map(|count| count.saturating_add(1)))
|
||||
.ignore_err()
|
||||
.ready_take_while(move |&(pducount, _)| pducount > starting_count)
|
||||
.map(move |mut pdu| {
|
||||
pdu.1.set_unsigned(Some(sender_user));
|
||||
pdu
|
||||
.wide_filter_map(async |mut pdu| {
|
||||
let ctx = services
|
||||
.rooms
|
||||
.timeline
|
||||
.get_unsigned_context(&pdu.1, Some(sender_user))
|
||||
.await;
|
||||
pdu.1.set_unsigned(
|
||||
ctx.user_id,
|
||||
ctx.membership,
|
||||
ctx.prev_content,
|
||||
ctx.redacted_because,
|
||||
);
|
||||
Some(pdu)
|
||||
})
|
||||
.then(async move |mut pdu| {
|
||||
if let Err(e) = services
|
||||
@@ -108,9 +118,19 @@ async fn load_timeline(
|
||||
.timeline
|
||||
.pdus_rev(room_id, ending_count.map(|count| count.saturating_add(1)))
|
||||
.ignore_err()
|
||||
.map(move |mut pdu| {
|
||||
pdu.1.set_unsigned(Some(sender_user));
|
||||
pdu
|
||||
.wide_filter_map(async |mut pdu| {
|
||||
let ctx = services
|
||||
.rooms
|
||||
.timeline
|
||||
.get_unsigned_context(&pdu.1, Some(sender_user))
|
||||
.await;
|
||||
pdu.1.set_unsigned(
|
||||
ctx.user_id,
|
||||
ctx.membership,
|
||||
ctx.prev_content,
|
||||
ctx.redacted_because,
|
||||
);
|
||||
Some(pdu)
|
||||
})
|
||||
.then(async move |mut pdu| {
|
||||
if let Err(e) = services
|
||||
|
||||
@@ -322,7 +322,12 @@ async fn build_sticky_events(
|
||||
continue;
|
||||
}
|
||||
|
||||
pdu.set_unsigned(Some(syncing_user));
|
||||
let ctx = services
|
||||
.rooms
|
||||
.timeline
|
||||
.get_unsigned_context(&pdu, Some(sync_context.syncing_user))
|
||||
.await;
|
||||
pdu.set_unsigned(ctx.user_id, ctx.membership, ctx.prev_content, ctx.redacted_because);
|
||||
events.push(Event::into_format(pdu));
|
||||
}
|
||||
|
||||
|
||||
@@ -1189,12 +1189,12 @@ async fn collect_sticky_events(
|
||||
|
||||
// MSC4480 lets us ignore the limit for a newly visible room's backlog
|
||||
for (room_id, pdu) in backlog {
|
||||
push_sticky_event(&mut response, sender_user, room_id, pdu);
|
||||
push_sticky_event(services, &mut response, sender_user, room_id, pdu).await;
|
||||
}
|
||||
|
||||
let mut next_batch = None;
|
||||
for (count, room_id, pdu) in stream.into_iter().take(limit) {
|
||||
push_sticky_event(&mut response, sender_user, room_id, pdu);
|
||||
push_sticky_event(services, &mut response, sender_user, room_id, pdu).await;
|
||||
next_batch = Some(count);
|
||||
}
|
||||
|
||||
@@ -1205,13 +1205,19 @@ async fn collect_sticky_events(
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
fn push_sticky_event(
|
||||
async fn push_sticky_event(
|
||||
services: &Services,
|
||||
response: &mut sync_events::v5::response::StickyEvents,
|
||||
sender_user: &UserId,
|
||||
room_id: OwnedRoomId,
|
||||
mut pdu: PduEvent,
|
||||
) {
|
||||
pdu.set_unsigned(Some(sender_user));
|
||||
let ctx = services
|
||||
.rooms
|
||||
.timeline
|
||||
.get_unsigned_context(&pdu, Some(sender_user))
|
||||
.await;
|
||||
pdu.set_unsigned(ctx.user_id, ctx.membership, ctx.prev_content, ctx.redacted_because);
|
||||
response
|
||||
.rooms
|
||||
.entry(room_id)
|
||||
|
||||
@@ -3,7 +3,6 @@ use std::cmp;
|
||||
use axum::extract::State;
|
||||
use conduwuit::{
|
||||
Err, Event, PduCount, Result, info,
|
||||
result::LogErr,
|
||||
utils::{IterStream, ReadyExt, stream::TryTools},
|
||||
};
|
||||
use futures::{FutureExt, StreamExt, TryStreamExt};
|
||||
@@ -80,15 +79,6 @@ pub(crate) async fn get_backfill_route(
|
||||
.await
|
||||
.then_some(pdu))
|
||||
})
|
||||
.and_then(async |mut pdu| {
|
||||
// Strip the transaction ID, as that is private
|
||||
pdu.remove_transaction_id().log_err().ok();
|
||||
// Add age, as this is specified
|
||||
pdu.add_age().log_err().ok();
|
||||
// It's not clear if we should strip or add any more data, leave as
|
||||
// is. In particular: Redaction?
|
||||
Ok(pdu)
|
||||
})
|
||||
.try_filter_map(|pdu| async move {
|
||||
Ok(services
|
||||
.rooms
|
||||
|
||||
Reference in new issue
Block a user