partially revert e507c31306

Signed-off-by: Jason Volk <jason@zemos.net>

src/api/client/context.rs

@@ -82,7 +82,7 @@ pub(crate) async fn get_context_route(
 	let events_before: Vec<_> = services
 		.rooms
 		.timeline
-		.pdus_rev(Some(sender_user), room_id, Some(base_token.saturating_sub(1)))
+		.pdus_rev(Some(sender_user), room_id, Some(base_token))
 		.await?
 		.ready_filter_map(|item| event_filter(item, filter))
 		.filter_map(|item| ignored_filter(&services, item, sender_user))
@@ -94,7 +94,7 @@ pub(crate) async fn get_context_route(
 	let events_after: Vec<_> = services
 		.rooms
 		.timeline
-		.pdus(Some(sender_user), room_id, Some(base_token.saturating_add(1)))
+		.pdus(Some(sender_user), room_id, Some(base_token))
 		.await?
 		.ready_filter_map(|item| event_filter(item, filter))
 		.filter_map(|item| ignored_filter(&services, item, sender_user))
@@ -169,14 +169,12 @@ pub(crate) async fn get_context_route(
 		start: events_before
 			.last()
 			.map(at!(0))
-			.map(|count| count.saturating_sub(1))
 			.as_ref()
 			.map(ToString::to_string),
 
 		end: events_after
 			.last()
 			.map(at!(0))
-			.map(|count| count.saturating_add(1))
 			.as_ref()
 			.map(ToString::to_string),
 

src/api/client/message.rs

@@ -138,10 +138,7 @@ pub(crate) async fn get_message_events_route(
 
 	let start_token = events.first().map(at!(0)).unwrap_or(from);
 
-	let next_token = events
+	let next_token = events.last().map(at!(0));
-		.last()
-		.map(at!(0))
-		.map(|count| count.saturating_inc(body.dir));
 
 	if !cfg!(feature = "element_hacks") {
 		if let Some(next_token) = next_token {

src/api/client/relations.rs

@@ -150,7 +150,6 @@ async fn paginate_relations_with_filter(
 		Direction::Backward => events.first(),
 	}
 	.map(at!(0))
-	.map(|count| count.saturating_inc(dir))
 	.as_ref()
 	.map(ToString::to_string);
 

src/api/client/threads.rs

@@ -46,7 +46,6 @@ pub(crate) async fn get_threads_route(
 			.last()
 			.filter(|_| threads.len() >= limit)
 			.map(at!(0))
-			.map(|count| count.saturating_sub(1))
 			.as_ref()
 			.map(ToString::to_string),
 

src/api/server/backfill.rs

@@ -55,7 +55,7 @@ pub(crate) async fn get_backfill_route(
 		pdus: services
 			.rooms
 			.timeline
-			.pdus_rev(None, &body.room_id, Some(from))
+			.pdus_rev(None, &body.room_id, Some(from.saturating_add(1)))
 			.await?
 			.take(limit)
 			.filter_map(|(_, pdu)| async move {

src/service/rooms/pdu_metadata/data.rs

@@ -57,7 +57,7 @@ impl Data {
 	) -> impl Stream<Item = PdusIterItem> + Send + '_ {
 		let mut current = ArrayVec::<u8, 16>::new();
 		current.extend(target.to_be_bytes());
-		current.extend(from.into_unsigned().to_be_bytes());
+		current.extend(from.saturating_inc(dir).into_unsigned().to_be_bytes());
 		let current = current.as_slice();
 		match dir {
 			Direction::Forward => self.tofrom_relation.raw_keys_from(current).boxed(),

src/service/rooms/threads/mod.rs

@@ -132,7 +132,7 @@ impl Service {
 
 		let current: RawPduId = PduId {
 			shortroomid,
-			shorteventid,
+			shorteventid: shorteventid.saturating_sub(1),
 		}
 		.into();
 

src/service/rooms/timeline/data.rs

@@ -13,7 +13,7 @@ use conduit::{
 };
 use database::{Database, Deserialized, Json, KeyVal, Map};
 use futures::{Stream, StreamExt};
-use ruma::{CanonicalJsonObject, EventId, OwnedRoomId, OwnedUserId, RoomId, UserId};
+use ruma::{api::Direction, CanonicalJsonObject, EventId, OwnedRoomId, OwnedUserId, RoomId, UserId};
 use tokio::sync::Mutex;
 
 use super::{PduId, RawPduId};
@@ -205,7 +205,9 @@ impl Data {
 	pub(super) async fn pdus_rev<'a>(
 		&'a self, user_id: Option<&'a UserId>, room_id: &'a RoomId, until: PduCount,
 	) -> Result<impl Stream<Item = PdusIterItem> + Send + 'a> {
-		let current = self.count_to_id(room_id, until).await?;
+		let current = self
+			.count_to_id(room_id, until, Direction::Backward)
+			.await?;
 		let prefix = current.shortroomid();
 		let stream = self
 			.pduid_pdu
@@ -220,7 +222,7 @@ impl Data {
 	pub(super) async fn pdus<'a>(
 		&'a self, user_id: Option<&'a UserId>, room_id: &'a RoomId, from: PduCount,
 	) -> Result<impl Stream<Item = PdusIterItem> + Send + 'a> {
-		let current = self.count_to_id(room_id, from).await?;
+		let current = self.count_to_id(room_id, from, Direction::Forward).await?;
 		let prefix = current.shortroomid();
 		let stream = self
 			.pduid_pdu
@@ -267,7 +269,7 @@ impl Data {
 		}
 	}
 
-	async fn count_to_id(&self, room_id: &RoomId, shorteventid: PduCount) -> Result<RawPduId> {
+	async fn count_to_id(&self, room_id: &RoomId, shorteventid: PduCount, dir: Direction) -> Result<RawPduId> {
 		let shortroomid: ShortRoomId = self
 			.services
 			.short
@@ -278,7 +280,7 @@ impl Data {
 		// +1 so we don't send the base event
 		let pdu_id = PduId {
 			shortroomid,
-			shorteventid,
+			shorteventid: shorteventid.saturating_inc(dir),
 		};
 
 		Ok(pdu_id.into())