On room creation, A list of initial state events may be provided that
should be submitted to the room after creation. These events
should be treated as any other state event and submitted to
the same checks.

With this change, state events submitted on room creation will be
submitted through the same helper as those through the usual endpoint.

The submission helper is moved to the timeline service to make it
available everywhere. This will be useful for implementing MSC4140
(issue #903).

@@ -0,0 +43,4 @@
+
+	// We use the room state lock to get the required synchronization guarantees
+	// on transaction IDs
+	let state_lock = services.rooms.state.mutex.lock(room_id.as_str()).await;

@@ -0,0 +110,4 @@
+	let Ruma {
+		body: delayed_events::update_delayed_event::unstable_v2::Request { delay_id, action, .. },
+		..
+	} = body;

@@ -0,0 +128,4 @@
+	let Ruma {
+		body: delayed_events::get_delayed_event::unstable::Request { delay_id, .. },
+		..
+	} = body;

@@ -0,0 +209,4 @@
+		}
+	}
+
+	async fn finalize_delayed_event(

@@ -0,0 +80,4 @@
+				queue.queue.push((Reverse(time), delay_id));
+			},
+			// Loop regularly to check if the service needs to stop even when there are no in-flight delayed events
+			() = sleep(Duration::from_secs(2)) => (),

@@ -23,3 +26,4 @@
 use crate::rooms::state::RoomMutexGuard;
 
 impl super::Service {
+	#[allow(clippy::too_many_arguments)]

@@ -0,0 +28,4 @@
+	} = body;
+
+	let sender_user = identity
+		.sender_user()

@@ -0,0 +111,4 @@
+) -> Result<delayed_events::get_all_delayed_events::unstable::Response> {
+	let sender_user = body
+		.identity
+		.sender_user()

@@ -254,3 +252,1 @@
-			self.depth > 0,
-			"serializing string at the top-level; you can skip serialization instead"
-		);
+	fn serialize_str(self, v: &str) -> Result<Self::Ok> { self.serialize_bytes(v.as_bytes()) }

@@ -0,0 +188,4 @@
+			// There are no synchronization primitives ensuring that the finalized event has
+			// made it to the database before we look for it. A best-effort answer is
+			// acceptable here, so we simply wait a bit and look again.
+			sleep(Duration::from_millis(200)).await;

@@ -0,0 +378,4 @@
+		self.submission_queue_sender.send((submission_time, delay_id.clone()))
+			.map_err(|_err|{
+				err!(Request(Unknown(
+					debug_error!(%event.user_id, "Server was unable to process delayed even request (queue full).")

@@ -0,0 +28,4 @@
+	let mut queue = service
+		.submission_queue
+		.try_lock()
+		.map_err(|_| err!(Err("Attempted to launch multiple instances of the worker")))?;

@@ -0,0 +67,4 @@
+		let next_receive = receiver.recv_async();
+		let next_interrupt = interrupt.recv_async();
+
+		// RescvFuture is cancellation-safe

@@ -70,0 +69,4 @@
+		if *event_type == MessageLikeEventType::CallInvite
+			&& self.services.directory.is_public_room(room_id).await
+		{
+			return Err!(Request(Forbidden("Room call invites are not allowed in public rooms")));

@@ -0,0 +14,4 @@
+) -> Result<delayed_events::send_delayed_event::unstable::Response> {
+	let Ruma {
+		body:
+			delayed_events::send_delayed_event::unstable::Request {