1 files changed, 71 insertions, 14 deletions
diff --git a/src/api/k2v/batch.rs b/src/api/k2v/batch.rs
index 82b4f7e3..26d678da 100644
--- a/src/api/k2v/batch.rs
+++ b/src/api/k2v/batch.rs
@@ -5,7 +5,6 @@ use hyper::{Body, Request, Response, StatusCode};
 use serde::{Deserialize, Serialize};
 
 use garage_util::data::*;
-use garage_util::error::Error as GarageError;
 
 use garage_table::{EnumerationOrder, TableSchema};
 
@@ -26,11 +25,7 @@ pub async fn handle_insert_batch(
 
 	let mut items2 = vec![];
 	for it in items {
-		let ct = it
-			.ct
-			.map(|s| CausalContext::parse(&s))
-			.transpose()
-			.ok_or_bad_request("Invalid causality token")?;
+		let ct = it.ct.map(|s| CausalContext::parse_helper(&s)).transpose()?;
 		let v = match it.v {
 			Some(vs) => DvvsValue::Value(
 				BASE64_STANDARD
@@ -68,10 +63,7 @@ pub async fn handle_read_batch(
 		resps.push(resp?);
 	}
 
-	let resp_json = serde_json::to_string_pretty(&resps).map_err(GarageError::from)?;
-	Ok(Response::builder()
-		.status(StatusCode::OK)
-		.body(Body::from(resp_json))?)
+	Ok(json_ok_response(&resps)?)
 }
 
 async fn handle_read_batch_query(
@@ -163,10 +155,7 @@ pub async fn handle_delete_batch(
 		resps.push(resp?);
 	}
 
-	let resp_json = serde_json::to_string_pretty(&resps).map_err(GarageError::from)?;
-	Ok(Response::builder()
-		.status(StatusCode::OK)
-		.body(Body::from(resp_json))?)
+	Ok(json_ok_response(&resps)?)
 }
 
 async fn handle_delete_batch_query(
@@ -260,6 +249,53 @@ async fn handle_delete_batch_query(
 	})
 }
 
+pub(crate) async fn handle_poll_range(
+	garage: Arc<Garage>,
+	bucket_id: Uuid,
+	partition_key: &str,
+	req: Request<Body>,
+) -> Result<Response<Body>, Error> {
+	use garage_model::k2v::sub::PollRange;
+
+	let query = parse_json_body::<PollRangeQuery>(req).await?;
+
+	let timeout_msec = query.timeout.unwrap_or(300).clamp(1, 600) * 1000;
+
+	let resp = garage
+		.k2v
+		.rpc
+		.poll_range(
+			PollRange {
+				partition: K2VItemPartition {
+					bucket_id,
+					partition_key: partition_key.to_string(),
+				},
+				start: query.start,
+				end: query.end,
+				prefix: query.prefix,
+			},
+			query.seen_marker,
+			timeout_msec,
+		)
+		.await?;
+
+	if let Some((items, seen_marker)) = resp {
+		let resp = PollRangeResponse {
+			items: items
+				.into_iter()
+				.map(|(_k, i)| ReadBatchResponseItem::from(i))
+				.collect::<Vec<_>>(),
+			seen_marker,
+		};
+
+		Ok(json_ok_response(&resp)?)
+	} else {
+		Ok(Response::builder()
+			.status(StatusCode::NOT_MODIFIED)
+			.body(Body::empty())?)
+	}
+}
+
 #[derive(Deserialize)]
 struct InsertBatchItem {
 	pk: String,
@@ -364,3 +400,24 @@ struct DeleteBatchResponse {
 	#[serde(rename = "deletedItems")]
 	deleted_items: usize,
 }
+
+#[derive(Deserialize)]
+struct PollRangeQuery {
+	#[serde(default)]
+	prefix: Option<String>,
+	#[serde(default)]
+	start: Option<String>,
+	#[serde(default)]
+	end: Option<String>,
+	#[serde(default)]
+	timeout: Option<u64>,
+	#[serde(default, rename = "seenMarker")]
+	seen_marker: Option<String>,
+}
+
+#[derive(Serialize)]
+struct PollRangeResponse {
+	items: Vec<ReadBatchResponseItem>,
+	#[serde(rename = "seenMarker")]
+	seen_marker: String,
+}