arminius_lib_db_read_receipts/

lib.rs

// SPDX-FileCopyrightText: 2025 BWI GmbH
//
// SPDX-License-Identifier: Apache-2.0

//! Queries related to the `read_receipts` table.

use std::{collections::BTreeMap, time::SystemTime};

use arminius_lib_db::{Database, DatabaseError, ResultExt as _};
use arminius_lib_model::{
	read_receipts::{DbReceiptType, SyncReceiptsMap},
	types::StreamId,
};
use ruma::{
	EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, UserId,
	api::client::filter::RoomEventFilter,
	events::receipt::{Receipt, ReceiptThread},
};

/// A read receipt stored in the database.
#[derive(Debug)]
pub struct ReadReceiptRow {
	/// The unique ID of the receipt. This is used in order to delete older
	/// receipts.
	pub nid: i32,
	/// The event the receipt acknowledges up to.
	pub event_id: OwnedEventId,
	/// The type of receipt (public or private).
	pub receipt_type: DbReceiptType,
	/// The thread the receipt applies to.
	pub thread: ReceiptThread,
	/// Ordering of the event. See the `22_event_ordering.sql` migration.
	pub event_ordering: Vec<i64>,
	/// Timestamp at which the user
	pub ts: chrono::DateTime<chrono::Utc>,
}

/// Read receipts fetched for syncing.
#[derive(Debug)]
pub struct SyncRoomReceipts {
	/// The maximum read receipt stream ID found in the fetched receipts.
	pub max_stream_id: StreamId,
	/// The actual read receipts map.
	pub receipts: SyncReceiptsMap,
}

/// Trait declaring SQL queries related to the `read_receipts` table.
// Queries are usually used as extension traits so suggest `import as _`.
#[rust_analyzer::prefer_underscore_import]
pub trait ReadReceiptsQueries: Database {
	/// Fetches all the read receipts of an user in a room. Optionally filter in
	/// only private receipts.
	#[tracing::instrument(skip(self))]
	async fn fetch_read_receipts(
		&mut self,
		room_id: &RoomId,
		user_id: &UserId,
		only_private_receipts: bool,
	) -> Result<Vec<ReadReceiptRow>, DatabaseError> {
		sqlx::query!(
			r#"
			SELECT
				read_receipts.nid,
				events.event_id,
				read_receipts.receipt_type AS "receipt_type:DbReceiptType",
				read_receipts.thread,
				event_ordering.ordering,
				read_receipts.ts
			FROM
				read_receipts
				JOIN rooms ON read_receipts.room_nid = rooms.nid
				JOIN events ON read_receipts.event_nid = events.nid
				JOIN event_ordering ON event_ordering.event_nid = events.nid
			WHERE
				rooms.room_id = $1
				AND read_receipts.user_id = $2
				AND (
					CASE WHEN $3::bool IS TRUE THEN
						read_receipts.receipt_type = 'm.read.private'
					ELSE
						TRUE
					END)
			ORDER BY
				event_ordering.ordering DESC
			"#,
			room_id.to_string(),
			user_id.to_string(),
			only_private_receipts
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.map(|row| {
			let event_id =
				EventId::parse(row.event_id).invalid_data("Invalid event ID in database")?;

			let thread = ReceiptThread::try_from(row.thread)
				.invalid_data("Invalid read receipt thread in database")?;

			Ok(ReadReceiptRow {
				nid: row.nid,
				event_id,
				receipt_type: row.receipt_type,
				thread,
				event_ordering: row.ordering,
				ts: row.ts,
			})
		})
		.collect::<Result<Vec<_>, _>>()
	}

	/// Inserts a new read receipt in the database.
	#[expect(
		trivial_casts,
		reason = "The casts in `sqlx` queries are not regular rust casts, but used to infer custom
		types from SQL queries. See
		<https://docs.rs/sqlx/latest/sqlx/macro.query.html#type-overrides-bind-parameters-postgres-only>"
	)]
	#[tracing::instrument(skip(self))]
	async fn insert_read_receipt(
		&mut self,
		room_id: &RoomId,
		user_id: &UserId,
		event_id: &EventId,
		receipt_type: DbReceiptType,
		thread: &ReceiptThread,
		ts: chrono::DateTime<chrono::Utc>,
	) -> Result<(), DatabaseError> {
		sqlx::query!(
			"
			INSERT INTO read_receipts
				(room_nid, user_id, event_nid, receipt_type, thread, ts)
			VALUES (
				(SELECT nid FROM rooms WHERE room_id = $1),
				$2,
				(SELECT nid FROM events WHERE event_id = $3),
				$4,
				$5,
				$6
			)
			",
			room_id.to_string(),
			user_id.to_string(),
			event_id.to_string(),
			receipt_type as DbReceiptType,
			thread.as_str(),
			ts
		)
		.execute(self.connection().await?)
		.await?;

		Ok(())
	}

	/// Deletes all the read receipts given in the `nids` list.
	#[tracing::instrument(skip(self))]
	async fn delete_read_receipts(&mut self, nids: &[i32]) -> Result<(), DatabaseError> {
		if sqlx::query!("DELETE FROM read_receipts WHERE nid = ANY($1::int[])", nids)
			.execute(self.connection().await?)
			.await?
			.rows_affected() as usize
			!= nids.len()
		{
			return Err(DatabaseError::SqlxError(sqlx::Error::RowNotFound));
		}

		Ok(())
	}

	/// Fetches all the new read receipts to sync.
	///
	/// NOTE: This will filter in only read receipts that the given user is
	/// allowed to see.
	#[tracing::instrument(skip(self))]
	async fn fetch_sync_read_receipts(
		&mut self,
		user_id: &UserId,
		from_stream_id: StreamId,
		filter: Option<&RoomEventFilter>,
	) -> Result<SyncRoomReceipts, DatabaseError> {
		let types = filter.and_then(|filter| filter.types.clone());
		let not_types = filter.map(|filter| filter.not_types.clone()).unwrap_or_default();
		let should_have_receipts = !(types
			.is_some_and(|types| !types.contains(&"m.receipt".to_owned()))
			|| not_types.contains(&"m.receipt".to_owned()));

		let rows = sqlx::query!(
			r#"
			SELECT
				rooms.room_id,
				read_receipts.user_id,
				events.event_id,
				read_receipts.stream_id,
				read_receipts.receipt_type AS "receipt_type:DbReceiptType",
				read_receipts.thread,
				read_receipts.ts
			FROM
				read_receipts
				JOIN room_membership_state ($2) ON read_receipts.room_nid = room_membership_state.room_nid
				JOIN rooms ON read_receipts.room_nid = rooms.nid
				JOIN events ON read_receipts.event_nid = events.nid
			WHERE
				-- Filter in rooms the user is joined in
				room_membership_state.membership = 'join'
				-- Filter in receipts that were stored after the given stream ID
				AND (read_receipts.stream_id > $1
					AND read_receipts.stream_id <= max_safe_seq ('read_receipts_streamids'))
				-- Users are allowed to see all of their own read receipts, and only the public
				-- ones of other users
				AND (read_receipts.user_id = $2
					OR read_receipts.receipt_type = 'm.read')
				AND $3
			"#,
			from_stream_id,
			user_id.to_string(),
			should_have_receipts
		)
		.fetch_all(self.connection().await?)
		.await?;

		let mut max_stream_id = StreamId::MIN;
		let mut receipts: SyncReceiptsMap = BTreeMap::new();

		for row in rows {
			max_stream_id = max_stream_id.max(row.stream_id);

			let room_id = RoomId::parse(row.room_id).invalid_data("Invalid room ID in database")?;
			let user_id = UserId::parse(row.user_id).invalid_data("Invalid user ID in database")?;
			let event_id =
				EventId::parse(row.event_id).invalid_data("Invalid event ID in database")?;
			let receipt_type = row.receipt_type.into();
			let receipt_thread = ReceiptThread::try_from(row.thread)
				.invalid_data("Invalid receipt thread in database")?;

			let ts = MilliSecondsSinceUnixEpoch::from_system_time(SystemTime::from(row.ts)).ok_or(
				DatabaseError::InvalidData("Invalid read receipt timestamp in database".to_owned()),
			)?;

			let mut receipt = Receipt::new(ts);
			receipt.thread = receipt_thread;

			receipts
				.entry(room_id)
				.or_default()
				.entry(event_id)
				.or_default()
				.entry(receipt_type)
				.or_default()
				.entry(user_id)
				.or_insert(receipt);
		}

		Ok(SyncRoomReceipts { max_stream_id, receipts })
	}
}

impl<T> ReadReceiptsQueries for T where T: Database {}