tuwunel_api_client/rendezvous/msc4388/
put.rs

1use std::result::Result as StdResult;
2
3use axum::{
4	body::Body,
5	extract::{Path, State, rejection::PathRejection},
6};
7use bytes::Bytes;
8use ruma::{
9	OwnedTransactionId, TransactionId,
10	api::client::rendezvous::update_rendezvous_session::unstable::Response,
11};
12use serde::Deserialize;
13use serde_json::{Value as JsonValue, from_slice, from_value};
14use tuwunel_core::{Err, Error as CoreError, Result as CoreResult, err, smallstr::SmallString};
15use tuwunel_service::{
16	Services,
17	rendezvous::{Attempt, Meta, Put},
18};
19
20use super::{Error, Result, ensure_available, ensure_data_size, ensure_txn_id};
21use crate::{ClientIp, RumaResponse, router::read_body};
22
23type Id = SmallString<[u8; 32]>;
24type SequenceToken = SmallString<[u8; 43]>;
25type PathParams = (Id, OwnedTransactionId);
26
27#[derive(Deserialize)]
28struct SendBody {
29	sequence_token: SequenceToken,
30	data: String,
31}
32
33#[tracing::instrument(level = "debug", skip_all)]
34pub(crate) async fn put_msc4388_route(
35	State(services): State<crate::State>,
36	ClientIp(client): ClientIp,
37	path: StdResult<Path<PathParams>, PathRejection>,
38	body: Body,
39) -> Result<RumaResponse<Response>> {
40	ensure_available(&services, client)?;
41
42	let Path((id, txn_id)) = path.map_err(CoreError::from)?;
43
44	ensure_txn_id(&txn_id)?;
45
46	// MSC4388: a recorded transaction is answered before the body is validated,
47	// which the Ruma extractor would parse first.
48	let meta = match services.rendezvous.recorded(&id, &txn_id) {
49		| Some(meta) => meta,
50		| None => send(&services, &id, &txn_id, body).await?,
51	};
52
53	let sequence_token = meta.sequence_token().to_owned();
54
55	Ok(RumaResponse(Response { sequence_token }))
56}
57
58async fn send(services: &Services, id: &str, txn_id: &TransactionId, body: Body) -> Result<Meta> {
59	let SendBody { sequence_token, data } = parse_body(services, body).await?;
60
61	ensure_data_size(services, &data)?;
62
63	let attempt = Attempt { txn_id, sequence_token: &sequence_token };
64
65	match services
66		.rendezvous
67		.put_token(id, attempt, Bytes::from(data))
68	{
69		| Put::NotFound =>
70			Err!(Request(NotFound("Rendezvous session not found"))).map_err(Into::into),
71		| Put::PreconditionFailed(_) => Err(Error::ConcurrentWrite),
72		| Put::Accepted(meta) => Ok(meta),
73	}
74}
75
76async fn parse_body(services: &Services, body: Body) -> CoreResult<SendBody> {
77	let body = read_body(services, body).await?;
78
79	// As the Ruma extractor: syntax outranks a field error, and a duplicate key
80	// keeps its last value.
81	let json: JsonValue = from_slice(&body)
82		.map_err(|_| err!(Request(NotJson("Request body is not valid JSON."))))?;
83
84	from_value(json).map_err(|e| err!(Request(BadJson("Request body is malformed: {e}"))))
85}
tuwunel_api_client/rendezvous/msc4388/put.rs

tuwunel_api_client/rendezvous/msc4388/
put.rs
