use std::{
	collections::{BTreeMap, HashMap, HashSet},
	fmt::Debug,
	sync::{
		Arc,
		atomic::{AtomicU64, AtomicUsize, Ordering},
	},
	time::{Duration, Instant},
};

use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use conduwuit::{
	debug_info, debug_warn, info, trace,
	utils::{should_continue_backoff, time::exponential_backoff::next_interval},
};
use conduwuit_core::{
	Error, Event, Result, at, debug, err, error,
	matrix::pdu::sticky,
	result::LogErr,
	utils::{
		ReadyExt, calculate_hash,
		future::TryExtExt,
		stream::{BroadbandExt, IterStream, WidebandExt},
	},
	warn,
};
use futures::{
	FutureExt, StreamExt,
	future::{BoxFuture, OptionFuture},
	join, pin_mut,
	stream::FuturesUnordered,
};
use http::StatusCode;
use ruma::{
	CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedRoomId, OwnedServerName, OwnedUserId,
	RoomId, ServerName, UInt,
	api::{
		appservice::event::push_events::v1::EphemeralData,
		federation::transactions::{
			edu::{
				DeviceListUpdateContent, Edu, PresenceContent, PresenceUpdate, ReceiptContent,
				ReceiptData, ReceiptMap,
			},
			send_transaction_message,
		},
	},
	device_id,
	events::{
		AnySyncEphemeralRoomEvent, GlobalAccountDataEventType, push_rules::PushRulesEvent,
		receipt::ReceiptType,
	},
	push,
	serde::Raw,
	uint,
};
use serde_json::value::{RawValue as RawJsonValue, to_raw_value};

use super::{Destination, EduBuf, EduVec, Msg, SendingEvent, Service, data::QueueItem};

#[derive(Debug)]
enum TransactionStatus {
	Running,
	Failed(u32, Instant), // number of times failed, time of last failure
	Retrying(u32),        // number of times failed
}

type SendingError = (Destination, Error);
type SendingResult = Result<Destination, SendingError>;
type SendingFuture<'a> = BoxFuture<'a, SendingResult>;
type SendingFutures<'a> = FuturesUnordered<SendingFuture<'a>>;
type CurTransactionStatus = HashMap<Destination, TransactionStatus>;

const SELECT_PRESENCE_LIMIT: usize = 256;
const SELECT_RECEIPT_LIMIT: usize = 256;
const SELECT_EDU_LIMIT: usize = EDU_LIMIT - 2;
const DEQUEUE_LIMIT: usize = 48;

pub const PDU_LIMIT: usize = 50;
pub const EDU_LIMIT: usize = 100;

impl Service {
	#[tracing::instrument(skip(self), level = "debug")]
	pub(super) async fn sender(self: Arc<Self>, id: usize) -> Result {
		let mut statuses: CurTransactionStatus = CurTransactionStatus::new();
		let mut futures: SendingFutures<'_> = FuturesUnordered::new();

		info!(shard = id, "Running startup netburst");
		self.startup_netburst(id, &mut futures, &mut statuses)
			.boxed()
			.await;
		info!(shard = id, "Finished startup netburst");

		self.work_loop(id, &mut futures, &mut statuses).await;

		if !futures.is_empty() {
			self.finish_responses(&mut futures).boxed().await;
		}

		Ok(())
	}

	#[tracing::instrument(
		name = "work",
		level = "trace"
		skip_all,
		fields(
			futures = %futures.len(),
			statuses = %statuses.len(),
		),
	)]
	async fn work_loop<'a>(
		&'a self,
		id: usize,
		futures: &mut SendingFutures<'a>,
		statuses: &mut CurTransactionStatus,
	) {
		let receiver = self
			.channels
			.get(id)
			.map(|(_, receiver)| receiver.clone())
			.expect("Missing channel for sender worker");

		while !receiver.is_closed() {
			tokio::select! {
				Some(response) = futures.next() => {
					self.handle_response(response, futures, statuses).await;
				},
				request = receiver.recv_async() => match request {
					Ok(request) => self.handle_request(request, futures, statuses).await,
					Err(_) => return,
				},
			}
		}
	}

	/// Handles a response, calling the appropriate sub-handler depending on if
	/// the request was successful or not.
	#[tracing::instrument(name = "response", level = "debug", skip_all)]
	async fn handle_response<'a>(
		&'a self,
		response: SendingResult,
		futures: &mut SendingFutures<'a>,
		statuses: &mut CurTransactionStatus,
	) {
		match response {
			| Ok(dest) => self.handle_response_ok(&dest, futures, statuses).await,
			| Err((dest, e)) => {
				if let Destination::Federation(dest) = &dest {
					if self.services.federation.should_mark_stale(&e)
						|| e.status_code() == StatusCode::NOT_FOUND
					{
						debug!("{dest} is now unhealthy & stale due to a connect error: {e:?}");
						self.services.federation.mark_destination_stale(dest);
						self.services.federation.hit_unhealthy(dest.clone());
					} else if e.status_code().is_server_error() {
						debug!("{dest} is now unhealthy due to server error response: {e:?}");
						self.services.federation.hit_unhealthy(dest.clone());
					}
				}
				Self::handle_response_err(dest, statuses, &e);
			},
		}
	}

	/// Handles a response error, incrementing the backoff factor.
	fn handle_response_err(dest: Destination, statuses: &mut CurTransactionStatus, er: &Error) {
		debug!(dest = ?dest, "{er:?}");
		let dest2 = dest.clone();
		statuses.entry(dest).and_modify(|e| {
			*e = match e {
				| TransactionStatus::Running => TransactionStatus::Failed(1, Instant::now()),
				| &mut TransactionStatus::Retrying(ref n) =>
					TransactionStatus::Failed(n.saturating_add(1), Instant::now()),
				| TransactionStatus::Failed(..) => {
					panic!(
						"{}",
						format!(
							"Request to {dest2:?} that was not even running ({e:?}) failed: \
							 {er:?}"
						)
					)
				},
			}
		});
	}

	/// Handles a successful response, queueing up more sends if necessary.
	#[allow(clippy::needless_pass_by_ref_mut)]
	async fn handle_response_ok<'a>(
		&'a self,
		dest: &Destination,
		futures: &mut SendingFutures<'a>,
		statuses: &mut CurTransactionStatus,
	) {
		let _cork = self.db.db.cork();
		self.db.delete_all_active_requests_for(dest).await;
		if let Destination::Federation(server_name) = dest {
			self.services.federation.mark_healthy(server_name);
		}

		// Find events that have been added since starting the last request
		let new_events = self
			.db
			.queued_requests(dest)
			.take(DEQUEUE_LIMIT)
			.collect::<Vec<_>>()
			.await;

		// Insert any pdus we found
		if !new_events.is_empty() {
			self.db.mark_as_active(new_events.iter());

			let new_events_vec = new_events.into_iter().map(at!(1)).collect();
			futures.push(self.send_events(dest.clone(), new_events_vec));
		} else {
			statuses.remove(dest);
		}
	}

	/// Handles sending a new request by queuing up any new events. If there are
	/// no new events to send the destination, it is removed from the status map
	/// instead.
	#[tracing::instrument(name = "request", level = "debug", skip_all)]
	#[allow(clippy::needless_pass_by_ref_mut)]
	async fn handle_request<'a>(
		&'a self,
		msg: Msg,
		futures: &mut SendingFutures<'a>,
		statuses: &mut CurTransactionStatus,
	) {
		let iv = vec![(msg.queue_id, msg.event)];
		match self.select_events(&msg.dest, iv, statuses).await {
			| Ok(Some(events)) => {
				if events.is_empty() {
					// Nothing more to send to this remote
					statuses.remove(&msg.dest);
				} else {
					futures.push(self.send_events(msg.dest, events));
				}
			},
			| Ok(None) => {
				trace!(
					?msg.dest,
					"Ignoring request to send to destination (nothing to send/already busy)"
				);
			},
			| Err(e) => error!(error=?e, ?msg.dest, "Failed to select events"),
		}
	}

	/// Cleans up dangling futures by waiting for them to complete, aborting
	/// early if the shutdown timeout is reached.
	#[tracing::instrument(
		name = "finish",
		level = "info",
		skip_all,
		fields(futures = %futures.len()),
	)]
	async fn finish_responses<'a>(&'a self, futures: &mut SendingFutures<'a>) {
		use tokio::{
			select,
			time::{Instant, sleep_until},
		};

		let timeout = self.server.config.sender_shutdown_timeout;
		let timeout = Duration::from_secs(timeout);
		let now = Instant::now();
		let deadline = now.checked_add(timeout).unwrap_or(now);
		loop {
			info!("Waiting for {} requests to complete...", futures.len());
			select! {
				() = sleep_until(deadline) => return,
				response = futures.next() => match response {
					Some(Ok(dest)) => self.db.delete_all_active_requests_for(&dest).await,
					Some(_) => continue,
					None => return,
				},
			}
		}
	}

	/// Performs startup netburst, which sends any previously queued up but not
	/// yet sent requests immediately.
	#[tracing::instrument(
		name = "netburst",
		level = "debug",
		skip_all,
		fields(futures = %futures.len()),
	)]
	#[allow(clippy::needless_pass_by_ref_mut)]
	async fn startup_netburst<'a>(
		&'a self,
		id: usize,
		futures: &mut SendingFutures<'a>,
		statuses: &mut CurTransactionStatus,
	) {
		let keep =
			usize::try_from(self.server.config.startup_netburst_keep).unwrap_or(usize::MAX);
		let mut txns = HashMap::<Destination, Vec<SendingEvent>>::new();
		let mut active = self.db.active_requests().boxed();

		while let Some((key, event, dest)) = active.next().await {
			if self.shard_id(&dest) != id {
				continue;
			}

			let entry = txns.entry(dest.clone()).or_default();
			let over_budget =
				self.server.config.startup_netburst_keep >= 0 && entry.len() >= keep;

			if over_budget && !self.survives_netburst_trim(&event).await {
				warn!(
					startup_netburst_keep = self.server.config.startup_netburst_keep,
					queue_size = entry.len(),
					"Dropping unsent event to {dest:?}: {:?}",
					String::from_utf8_lossy(&key)
				);
				self.db.delete_active_request(&key);
			} else {
				entry.push(event);
			}
		}

		for (dest, events) in txns {
			if self.server.config.startup_netburst && !events.is_empty() {
				statuses.insert(dest.clone(), TransactionStatus::Running);
				futures.push(self.send_events(dest.clone(), events));
			}
		}
	}

	/// Whether this event skips the netburst budget, which MSC4354 asks of
	/// still-sticky events. Only netburst delivers this queue, so with it
	/// disabled there is nothing to keep the event for.
	async fn survives_netburst_trim(&self, event: &SendingEvent) -> bool {
		if !self.server.config.allow_sticky_events || !self.server.config.startup_netburst {
			return false;
		}

		let SendingEvent::Pdu(pdu_id) = event else {
			return false;
		};

		let Ok(pdu) = self.services.timeline.get_pdu_from_id(pdu_id).await else {
			return false;
		};

		let now = MilliSecondsSinceUnixEpoch::now().get().into();

		pdu.sticky
			.as_deref()
			.is_some_and(|sticky| sticky::is_sticky(pdu.origin_server_ts, sticky, now))
	}

	/// Selects any new events to send to the given destination.
	#[tracing::instrument(
		name = "select",
		level = "debug",
		skip_all,
		fields(
			?dest,
			new_events = %new_events.len(),
		)
	)]
	async fn select_events(
		&self,
		dest: &Destination,
		new_events: Vec<QueueItem>, // Events we want to send: event and full key
		statuses: &mut CurTransactionStatus,
	) -> Result<Option<Vec<SendingEvent>>> {
		let (allow, retry) = self.should_attempt_send(dest, statuses)?;

		// Nothing can be done for this remote, bail out.
		if !allow {
			return Ok(None);
		}

		let _cork = self.db.db.cork();
		let mut events = Vec::new();

		// Must retry any previous transaction for this remote.
		if retry {
			self.db
				.active_requests_for(dest)
				.ready_for_each(|(_, e)| events.push(e))
				.await;

			return Ok(Some(events));
		}

		// Compose the next transaction
		let _cork = self.db.db.cork();
		if !new_events.is_empty() {
			self.db.mark_as_active(new_events.iter());
			for (_, e) in new_events {
				events.push(e);
			}
		}

		// Add EDU's into the transaction
		if let Destination::Federation(server_name) = dest {
			if let Ok((select_edus, last_count)) = self.select_edus(server_name).await {
				debug_assert!(select_edus.len() <= EDU_LIMIT, "exceeded edus limit");
				let select_edus = select_edus.into_iter().map(SendingEvent::Edu);

				events.extend(select_edus);
				self.db.set_latest_educount(server_name, last_count);
			}
		}

		Ok(Some(events))
	}

	/// Determines if the destination should be retried, honouring exponential
	/// backoff.
	fn should_attempt_send(
		&self,
		dest: &Destination,
		statuses: &mut CurTransactionStatus,
	) -> Result<(bool, bool)> {
		let (mut allow, mut retry) = (true, false);
		if let Destination::Federation(server_name) = dest {
			let entry = statuses.entry(dest.clone()).and_modify(|e| match e {
				| TransactionStatus::Running | TransactionStatus::Retrying(_) => {
					allow = false; // already running
					trace!("Transaction is already running for {dest:?}");
				},
				| TransactionStatus::Failed(tries, _) => {
					*e = TransactionStatus::Retrying(*tries);
					retry = true;
					trace!("Previous transaction failed for {dest:?} ({e:?}), will retry");
				},
			});
			if let Some(retry_after) = self.services.federation.retry_after(server_name) {
				allow = allow && retry_after.is_zero();
			}
			if allow {
				trace!(current_status=?entry, "Inserting running status for {dest:?}");
				entry.or_insert(TransactionStatus::Running);
			}
			return Ok((allow, retry));
		}
		statuses
			.entry(dest.clone())
			.and_modify(|e| match e {
				| TransactionStatus::Failed(tries, time) => {
					let min = Duration::from_secs(self.server.config.sender_retry_backoff_base);
					let max = Duration::from_secs(self.server.config.sender_retry_backoff_limit);
					if should_continue_backoff(min, max, time.elapsed(), *tries)
						&& !matches!(dest, Destination::Appservice(_))
					{
						let retry_after = next_interval(min, max, *tries);
						debug_warn!("Not retrying destination for another {retry_after:?}");
						allow = false;
					} else {
						retry = true;
						debug_info!("Will retry destination ({} tries)", *tries);
						*e = TransactionStatus::Retrying(*tries);
					}
				},
				| TransactionStatus::Running | TransactionStatus::Retrying(_) => {
					allow = false; // already running
				},
			})
			.or_insert(TransactionStatus::Running);

		Ok((allow, retry))
	}

	/// Selects a number of EDUs to send to the target.
	#[tracing::instrument(name = "edus", level = "debug", skip_all)]
	async fn select_edus(&self, server_name: &ServerName) -> Result<(EduVec, u64)> {
		// selection window
		let since = self.db.get_latest_educount(server_name).await;
		let since_upper = self.services.globals.current_count()?;
		let batch = (since, since_upper);
		debug_assert!(batch.0 <= batch.1, "since range must not be negative");

		let events_len = AtomicUsize::default();
		let max_edu_count = AtomicU64::new(since);

		let device_changes =
			self.select_edus_device_changes(server_name, batch, &max_edu_count, &events_len);

		let receipts: OptionFuture<_> = self
			.server
			.config
			.allow_outgoing_read_receipts
			.then(|| self.select_edus_receipts(server_name, batch, &max_edu_count))
			.into();

		let presence: OptionFuture<_> = self
			.server
			.config
			.allow_outgoing_presence
			.then(|| self.select_edus_presence(server_name, batch, &max_edu_count))
			.into();

		let (device_changes, receipts, presence) = join!(device_changes, receipts, presence);

		let mut events = device_changes;
		events.extend(presence.into_iter().flatten());
		events.extend(receipts.into_iter().flatten());

		// TODO(nex): some EDUs like typing need flattening

		Ok((events, max_edu_count.load(Ordering::Acquire)))
	}

	/// Look for device changes
	#[tracing::instrument(
		name = "device_changes",
		level = "trace",
		skip(self, server_name, max_edu_count)
	)]
	async fn select_edus_device_changes(
		&self,
		server_name: &ServerName,
		since: (u64, u64),
		max_edu_count: &AtomicU64,
		events_len: &AtomicUsize,
	) -> EduVec {
		let mut events = EduVec::new();
		let server_rooms = self.services.state_cache.server_rooms(server_name);

		pin_mut!(server_rooms);
		let mut device_list_changes = HashSet::<OwnedUserId>::new();
		while let Some(room_id) = server_rooms.next().await {
			let keys_changed = self
				.services
				.users
				.room_keys_changed(&room_id, Some(since.0), None)
				.ready_filter(|(user_id, _)| self.services.globals.user_is_local(user_id));

			pin_mut!(keys_changed);
			while let Some((user_id, count)) = keys_changed.next().await {
				if count > since.1 {
					break;
				}

				max_edu_count.fetch_max(count, Ordering::Relaxed);
				if !device_list_changes.insert(user_id.clone()) {
					continue;
				}

				// Empty prev id forces synapse to resync; because synapse resyncs,
				// we can just insert placeholder data
				let edu = Edu::DeviceListUpdate(DeviceListUpdateContent::new(
					user_id,
					device_id!("placeholder").to_owned(),
					uint!(1),
				));

				let mut buf = EduBuf::new();
				serde_json::to_writer(&mut buf, &edu)
					.expect("failed to serialize device list update to JSON");

				events.push(buf);
				if events_len.fetch_add(1, Ordering::Relaxed) >= SELECT_EDU_LIMIT - 1 {
					return events;
				}
			}
		}

		events
	}

	/// Look for read receipts in this room
	#[tracing::instrument(
		name = "receipts",
		level = "trace",
		skip(self, server_name, max_edu_count)
	)]
	async fn select_edus_receipts(
		&self,
		server_name: &ServerName,
		since: (u64, u64),
		max_edu_count: &AtomicU64,
	) -> Option<EduBuf> {
		let mut num = 0;
		let receipts: BTreeMap<OwnedRoomId, ReceiptMap> = self
			.services
			.state_cache
			.server_rooms(server_name)
			.broad_filter_map(|room_id| async move {
				let receipt_map = self
					.select_edus_receipts_room(&room_id, since, max_edu_count, &mut num)
					.await;

				receipt_map
					.read
					.is_empty()
					.eq(&false)
					.then_some((room_id, receipt_map))
			})
			.collect()
			.await;

		if receipts.is_empty() {
			return None;
		}

		let receipt_content = Edu::Receipt(ReceiptContent::new(receipts));

		let mut buf = EduBuf::new();
		serde_json::to_writer(&mut buf, &receipt_content)
			.expect("Failed to serialize Receipt EDU to JSON vec");

		Some(buf)
	}

	/// Look for read receipts in this room
	#[tracing::instrument(name = "receipts", level = "trace", skip(self, since, max_edu_count))]
	async fn select_edus_receipts_room(
		&self,
		room_id: &RoomId,
		since: (u64, u64),
		max_edu_count: &AtomicU64,
		num: &mut usize,
	) -> ReceiptMap {
		let receipts = self
			.services
			.read_receipt
			.readreceipts_since(room_id, Some(since.0));

		pin_mut!(receipts);
		let mut read = BTreeMap::<OwnedUserId, ReceiptData>::new();
		while let Some((user_id, count, read_receipt)) = receipts.next().await {
			if count > since.1 {
				break;
			}

			max_edu_count.fetch_max(count, Ordering::Relaxed);
			if !self.services.globals.user_is_local(&user_id) {
				continue;
			}

			let Ok(event) = serde_json::from_str(read_receipt.json().get()) else {
				error!(%user_id, %count, ?read_receipt, "Invalid edu event in read_receipts.");
				continue;
			};

			let AnySyncEphemeralRoomEvent::Receipt(r) = event else {
				error!(%user_id, %count, ?event, "Invalid event type in read_receipts");
				continue;
			};

			let (event_id, mut receipt) = r
				.content
				.0
				.into_iter()
				.next()
				.expect("we only use one event per read receipt");

			let receipt = receipt
				.remove(&ReceiptType::Read)
				.expect("our read receipts always set this")
				.remove(&user_id)
				.expect("our read receipts always have the user here");

			let receipt_data = ReceiptData::new(receipt, vec![event_id.clone()]);

			if read.insert(user_id, receipt_data).is_none() {
				*num = num.saturating_add(1);
				if *num >= SELECT_RECEIPT_LIMIT {
					break;
				}
			}
		}

		ReceiptMap::new(read)
	}

	/// Look for presence
	#[tracing::instrument(
		name = "presence",
		level = "trace",
		skip(self, server_name, max_edu_count)
	)]
	async fn select_edus_presence(
		&self,
		server_name: &ServerName,
		since: (u64, u64),
		max_edu_count: &AtomicU64,
	) -> Option<EduBuf> {
		let presence_since = self.services.presence.presence_since(since.0);

		pin_mut!(presence_since);
		let mut presence_updates = HashMap::<OwnedUserId, PresenceUpdate>::new();
		while let Some((user_id, count, presence_bytes)) = presence_since.next().await {
			if count > since.1 {
				break;
			}

			max_edu_count.fetch_max(count, Ordering::Relaxed);
			if !self.services.globals.user_is_local(user_id) {
				continue;
			}

			if !self
				.services
				.state_cache
				.server_sees_user(server_name, user_id)
				.await
			{
				continue;
			}

			let Ok(presence_event) = self
				.services
				.presence
				.from_json_bytes_to_event(presence_bytes, user_id)
				.await
				.log_err()
			else {
				continue;
			};

			let mut update = PresenceUpdate::new(
				user_id.to_owned(),
				presence_event.content.presence,
				presence_event
					.content
					.last_active_ago
					.unwrap_or_else(|| uint!(0)),
			);
			update.currently_active = presence_event.content.currently_active.unwrap_or_default();
			update.status_msg = presence_event.content.status_msg;

			presence_updates.insert(user_id.into(), update);
			if presence_updates.len() >= SELECT_PRESENCE_LIMIT {
				break;
			}
		}

		if presence_updates.is_empty() {
			return None;
		}

		let presence_content =
			Edu::Presence(PresenceContent::new(presence_updates.into_values().collect()));

		let mut buf = EduBuf::new();
		serde_json::to_writer(&mut buf, &presence_content)
			.expect("failed to serialize Presence EDU to JSON");

		Some(buf)
	}

	/// Actually sends fetched events to the destination.
	/// If the events vec is empty, a stub response is returned, and a debug
	/// warning is logged.
	fn send_events(&self, dest: Destination, events: Vec<SendingEvent>) -> SendingFuture<'_> {
		if events.is_empty() {
			debug_warn!(?dest, "Attempted to send empty transaction!");
			return std::future::ready(Ok(dest)).boxed();
		}
		match dest {
			| Destination::Federation(server) =>
				self.send_events_dest_federation(server, events).boxed(),
			| Destination::Appservice(id) => self.send_events_dest_appservice(id, events).boxed(),
			| Destination::Push(user_id, pushkey) =>
				self.send_events_dest_push(user_id, pushkey, events).boxed(),
		}
	}

	#[tracing::instrument(
		name = "appservice",
		level = "debug",
		skip(self, events),
		fields(
			events = %events.len(),
		),
	)]
	async fn send_events_dest_appservice(
		&self,
		id: String,
		events: Vec<SendingEvent>,
	) -> SendingResult {
		let Some(appservice) = self.services.appservice.get_registration(&id).await else {
			return Err((
				Destination::Appservice(id.clone()),
				err!(Database(warn!(?id, "Missing appservice registration"))),
			));
		};

		let mut pdu_jsons = Vec::with_capacity(
			events
				.iter()
				.filter(|event| matches!(event, SendingEvent::Pdu(_)))
				.count(),
		);
		let mut edu_jsons: Vec<Raw<EphemeralData>> = Vec::with_capacity(
			events
				.iter()
				.filter(|event| matches!(event, SendingEvent::Edu(_)))
				.count(),
		);
		for event in &events {
			match event {
				| SendingEvent::Pdu(pdu_id) => {
					if let Ok(pdu) = self.services.timeline.get_pdu_from_id(pdu_id).await {
						pdu_jsons.push(pdu.to_format());
					}
				},
				| SendingEvent::Edu(edu) =>
					if appservice.receive_ephemeral {
						if let Ok(edu) = serde_json::from_slice(edu) {
							edu_jsons.push(Raw::from_json(edu));
						}
					},
				| SendingEvent::Flush => {}, // flush only; no new content
			}
		}

		let txn_hash = calculate_hash(events.iter().filter_map(|e| match e {
			| SendingEvent::Edu(b) => Some(&**b),
			| SendingEvent::Pdu(b) => Some(b.as_ref()),
			| SendingEvent::Flush => None,
		}));

		let txn_id = &*URL_SAFE_NO_PAD.encode(txn_hash);

		let mut request =
			ruma::api::appservice::event::push_events::v1::Request::new(txn_id.into(), pdu_jsons);
		request.ephemeral = edu_jsons;
		request.to_device = Vec::new(); // TODO

		match self.send_appservice_request(appservice, request).await {
			| Ok(_) => Ok(Destination::Appservice(id)),
			| Err(e) => Err((Destination::Appservice(id), e)),
		}
	}

	#[tracing::instrument(
		name = "push",
		level = "info",
		skip(self, events),
		fields(
			events = %events.len(),
		),
	)]
	async fn send_events_dest_push(
		&self,
		user_id: OwnedUserId,
		pushkey: String,
		events: Vec<SendingEvent>,
	) -> SendingResult {
		let Ok(pusher) = self.services.pusher.get_pusher(&user_id, &pushkey).await else {
			return Err((
				Destination::Push(user_id.clone(), pushkey.clone()),
				err!(Database(error!(%user_id, ?pushkey, "Missing pusher"))),
			));
		};

		let mut pdus = Vec::with_capacity(
			events
				.iter()
				.filter(|event| matches!(event, SendingEvent::Pdu(_)))
				.count(),
		);
		for event in &events {
			match event {
				| SendingEvent::Pdu(pdu_id) => {
					if let Ok(pdu) = self.services.timeline.get_pdu_from_id(pdu_id).await {
						pdus.push(pdu);
					}
				},
				| SendingEvent::Edu(_) | SendingEvent::Flush => {
					// Push gateways don't need EDUs (?) and flush only;
					// no new content
				},
			}
		}

		for pdu in pdus {
			// Redacted events are not notification targets (we don't send push for them)
			if pdu.is_redacted() {
				continue;
			}

			let rules_for_user = self
				.services
				.account_data
				.get_global(&user_id, GlobalAccountDataEventType::PushRules)
				.await
				.map_or_else(
					|_| push::Ruleset::server_default(&user_id),
					|ev: PushRulesEvent| ev.content.global,
				);

			let unread: UInt = self
				.services
				.user
				.notification_count(&user_id, &pdu.room_id_or_hash())
				.await
				.try_into()
				.expect("notification count can't go that high");

			let _response = self
				.services
				.pusher
				.send_push_notice(&user_id, unread, &pusher, rules_for_user, &pdu)
				.await
				.map_err(|e| (Destination::Push(user_id.clone(), pushkey.clone()), e));
		}

		Ok(Destination::Push(user_id, pushkey))
	}

	async fn send_events_dest_federation(
		&self,
		server: OwnedServerName,
		events: Vec<SendingEvent>,
	) -> SendingResult {
		let pdus: Vec<_> = events
			.iter()
			.filter_map(|pdu| match pdu {
				| SendingEvent::Pdu(pdu) => Some(pdu),
				| _ => None,
			})
			.stream()
			.wide_filter_map(|pdu_id| self.services.timeline.get_pdu_json_from_id(pdu_id).ok())
			.wide_then(|pdu| self.convert_to_outgoing_federation_event(pdu))
			.collect()
			.await;

		let edus: Vec<Raw<Edu>> = events
			.iter()
			.filter_map(|edu| match edu {
				| SendingEvent::Edu(edu) => Some(edu.as_ref()),
				| _ => None,
			})
			.map(serde_json::from_slice)
			.filter_map(Result::ok)
			.collect();

		if pdus.is_empty() && edus.is_empty() {
			return Ok(Destination::Federation(server));
		}

		let preimage = pdus
			.iter()
			.map(|raw| raw.get().as_bytes())
			.chain(edus.iter().map(|raw| raw.json().get().as_bytes()));

		let txn_hash = calculate_hash(preimage);
		let txn_id = &*URL_SAFE_NO_PAD.encode(txn_hash);
		let mut request = send_transaction_message::v1::Request::new(
			txn_id.into(),
			self.server.name.clone(),
			MilliSecondsSinceUnixEpoch::now(),
		);
		request.pdus = pdus;
		request.edus = edus;
		let pdu_count = request.pdus.len();
		let edu_count = request.edus.len();

		debug_info!(
			%txn_id,
			pdus=pdu_count,
			edus=edu_count,
			%server,
			"Sending transaction to remote"
		);
		let start = Instant::now();
		let result = self
			.services
			.federation
			.execute_signed(&self.services.client.sender, &server, request)
			.await;
		debug_info!(
			%txn_id,
			pdus=pdu_count,
			edus=edu_count,
			elapsed=?start.elapsed(),
			%server,
			"Finished sending transaction"
		);

		for (event_id, result) in result.iter().flat_map(|resp| resp.pdus.iter()) {
			if let Err(e) = result {
				info!(
					%txn_id,
					%server,
					%event_id,
					remote_error=?e,
					"remote server encountered an error while processing an event"
				);
			}
		}

		match result {
			| Err(error) => Err((Destination::Federation(server), error)),
			| Ok(_) => Ok(Destination::Federation(server)),
		}
	}

	/// Converts and sanitises an outgoing PDU object for federation
	/// transmission.
	pub async fn convert_to_outgoing_federation_event(
		&self,
		mut pdu_json: CanonicalJsonObject,
	) -> Box<RawJsonValue> {
		if let Some(unsigned) = pdu_json
			.get_mut("unsigned")
			.and_then(|val| val.as_object_mut())
		{
			// TODO: remove all unsigned data over federation.
			unsigned.remove("transaction_id");
		}
		// We don't support any room versions that have the event_id in the PDU JSON,
		// so we can safely remove it here.
		pdu_json.remove("event_id");

		to_raw_value(&pdu_json).expect("CanonicalJson is valid serde_json::Value")
	}
}