Merge pull request #26 from dimacurrentai/cursor

Vibe-coded a few more endpoints that use the DB.

Author: dkorolev

step07_redb/code/README.md

@@ -0,0 +1,41 @@
+# Trying Cursor
+
+## Add JSON-parsing Endpoints
+
+Prompt:
+
+```
+Add an endpoint, `/sums`.
+
+This endpoint should accept POST-ed JSONs of the following format: `{"id": "foobar", "a": 2, "b": 3, "c": 5}`.
+
+Do not use standard Axum ways to parse the JSON. You should accept any `Content-Type`. Parse the JSON into the respective type. Do not analyze individual fields of the JSON object! Use the standard tool such as `serde`.
+
+If the JSON schema does not match, the endpoint should return an error.
+
+If the JSON schema does matech but `c` is not `a+b`, the endpoint should return an error too.
+
+Otherwise the endpoint should return OK. Nothing else is required for now.
+```
+
+## Keep `id`-s Unique
+
+Prompt:
+
+```
+Make sure to store the `.id`-s of the POST-ed payloads to `/sums` in the database.
+
+Process requests with new `.id`-s same way as before, just store those IDs.
+
+When a request comes with a duplicate `.id`, disregard it with the respective message.
+
+Also, the `.id`-s should be non-empty strings.
+```
+
+## Track and Return Recent Requests
+
+```
+Now, let's make sure some one thousand (configurable with a constant) most recent requests to `/sums` are journaled.
+
+Add a GET endpoint to return the most recent N, for a provided N, up to this configured constant number of them that are stored.
+```

step07_redb/code/src/main.rs

@@ -1,8 +1,12 @@
-use axum::{extract::State, routing::get, serve, Router};
-use hyper::header::HeaderMap;
+use axum::{
+  extract::{Query, State},
+  routing::{get, post},
+  serve, Router,
+};
+use hyper::{body::Bytes, header::HeaderMap, StatusCode};
 use redb::{Database, ReadableTable, TableDefinition};
 use serde::{Deserialize, Serialize};
-use std::{error::Error, fs, net::SocketAddr, path::Path};
+use std::{error::Error, fs, net::SocketAddr, path::Path, time::SystemTime};
 use tokio::signal::unix::{signal, SignalKind};
 use tokio::{net::TcpListener, sync::mpsc};
 
@@ -15,6 +19,8 @@ use crate::{
   lib::http,
 };
 
+const DEFAULT_REQUEST_SIZE: u64 = 10;
+const MAX_REQUEST_SIZE: u64 = 100;
 #[repr(transparent)]
 struct CounterType(bool);
 
@@ -25,6 +31,9 @@ impl CounterType {
 
 static COUNTERS: TableDefinition<bool, u64> = TableDefinition::new("counters");
 static STRINGS: TableDefinition<u64, &str> = TableDefinition::new("strings");
+static PROCESSED_IDS: TableDefinition<&str, bool> = TableDefinition::new("processed_ids");
+static SUMS_JOURNAL: TableDefinition<u64, &[u8]> = TableDefinition::new("sums_journal");
+static JOURNAL_META: TableDefinition<&str, u64> = TableDefinition::new("journal_meta");
 
 struct IncCounterRequest {
   counter_type: CounterType,
@@ -66,6 +75,98 @@ impl DbRequestHandler for IncStringRequest {
   }
 }
 
+struct CheckAndStoreIdRequest(String);
+
+enum CheckAndStoreIdResponse {
+  Unique,
+  Duplicate,
+}
+
+impl DbRequestHandler for CheckAndStoreIdRequest {
+  type Response = CheckAndStoreIdResponse;
+
+  fn handle(&self, db: &Database) -> AsyncDbResult<Self::Response> {
+    let write_txn = db.begin_write()?;
+    {
+      let mut table = write_txn.open_table(PROCESSED_IDS)?;
+      if table.get(self.0.as_str())?.is_some() {
+        return Ok(CheckAndStoreIdResponse::Duplicate);
+      }
+      table.insert(self.0.as_str(), true)?;
+    }
+    write_txn.commit()?;
+    Ok(CheckAndStoreIdResponse::Unique)
+  }
+}
+
+#[derive(Deserialize, Debug)]
+struct SumRequest {
+  id: String,
+  a: i32,
+  b: i32,
+  c: i32,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct JournalEntry {
+  id: String,
+  a: i32,
+  b: i32,
+  c: i32,
+  timestamp_us: u64,
+}
+
+struct AddToJournalRequest(JournalEntry);
+
+impl DbRequestHandler for AddToJournalRequest {
+  type Response = ();
+
+  fn handle(&self, db: &Database) -> AsyncDbResult<Self::Response> {
+    let serialized = serde_json::to_vec(&self.0)?;
+
+    let write_txn = db.begin_write()?;
+    {
+      let mut journal_meta = write_txn.open_table(JOURNAL_META)?;
+      let mut journal = write_txn.open_table(SUMS_JOURNAL)?;
+
+      let idx = journal_meta.get("idx")?.map(|v| v.value()).unwrap_or(0);
+      journal.insert(idx, serialized.as_slice())?;
+      journal_meta.insert("idx", idx + 1)?;
+    }
+    write_txn.commit()?;
+    Ok(())
+  }
+}
+
+struct GetJournalRequest(u64);
+
+impl DbRequestHandler for GetJournalRequest {
+  type Response = Vec<JournalEntry>;
+
+  fn handle(&self, db: &Database) -> AsyncDbResult<Self::Response> {
+    let read_txn = db.begin_read()?;
+    let journal_meta = read_txn.open_table(JOURNAL_META)?;
+    let journal = read_txn.open_table(SUMS_JOURNAL)?;
+
+    let idx = journal_meta.get("idx")?.map(|v| v.value()).unwrap_or(0);
+
+    let mut result = Vec::new();
+    let n = std::cmp::min(std::cmp::min(self.0, idx), MAX_REQUEST_SIZE);
+
+    let i0 = idx - n;
+
+    for i in 0..n {
+      if let Some(entry_bytes) = journal.get(i0 + i)? {
+        if let Ok(entry) = serde_json::from_slice::<JournalEntry>(entry_bytes.value()) {
+          result.push(entry);
+        }
+      }
+    }
+
+    Ok(result)
+  }
+}
+
 #[derive(Clone)]
 struct AppState {
   redb: AsyncRedb,
@@ -80,6 +181,7 @@ enum JSONResponse {
   Message { text: String },
   Counters { counter_runs: u64, counter_requests: u64 },
   StringCounter { value: String },
+  Journal { entries: Vec<JournalEntry> },
 }
 
 async fn health_handler() -> &'static str {
@@ -109,6 +211,63 @@ async fn string_handler(State(state): State<AppState>, headers: HeaderMap) -> im
   http::json_or_html(headers, &json_string).await
 }
 
+#[derive(Deserialize)]
+struct JournalQuery {
+  #[serde(default = "default_request_size")]
+  n: u64,
+}
+
+fn default_request_size() -> u64 {
+  DEFAULT_REQUEST_SIZE
+}
+
+async fn journal_handler(
+  State(state): State<AppState>, headers: HeaderMap, Query(query): Query<JournalQuery>,
+) -> impl axum::response::IntoResponse {
+  let entries = state.redb.run(GetJournalRequest(query.n)).await.unwrap_or_default();
+  let response = JSONResponse::Journal { entries };
+  let json_string = serde_json::to_string(&response).unwrap();
+  http::json_or_html(headers, &json_string).await
+}
+
+async fn sums_handler(State(state): State<AppState>, body: Bytes) -> Result<&'static str, (StatusCode, String)> {
+  match serde_json::from_slice::<SumRequest>(&body) {
+    Ok(req) => {
+      if req.id.is_empty() {
+        Err((StatusCode::BAD_REQUEST, "Error: id cannot be empty.\n".to_string()))
+      } else {
+        match state.redb.run(CheckAndStoreIdRequest(req.id.clone())).await {
+          Ok(inner_result) => match inner_result {
+            CheckAndStoreIdResponse::Unique => {
+              let result = req.c == req.a + req.b;
+              if result {
+                let _ = state
+                  .redb
+                  .run(AddToJournalRequest(JournalEntry {
+                    id: req.id.clone(),
+                    a: req.a,
+                    b: req.b,
+                    c: req.c,
+                    timestamp_us: SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_micros() as u64,
+                  }))
+                  .await;
+                Ok("OK\n")
+              } else {
+                Err((StatusCode::BAD_REQUEST, "Error: c must equal a + b.\n".to_string()))
+              }
+            }
+            CheckAndStoreIdResponse::Duplicate => {
+              Err((StatusCode::CONFLICT, "Error: Duplicate request ID.\n".to_string()))
+            }
+          },
+          Err(e) => Err((StatusCode::INTERNAL_SERVER_ERROR, e.to_string())),
+        }
+      }
+    }
+    Err(e) => Err((StatusCode::BAD_REQUEST, format!("Invalid JSON: {}.\n", e))),
+  }
+}
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
   fs::create_dir_all(&Path::new("./.db"))?;
@@ -125,6 +284,8 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     .route("/quit", get(quit_handler))
     .route("/json", get(json_handler))
     .route("/string", get(string_handler))
+    .route("/journal", get(journal_handler))
+    .route("/sums", post(sums_handler))
     .with_state(state);
 
   let addr = SocketAddr::from(([0, 0, 0, 0], 3000));

step07_redb/iteration.sh

@@ -24,3 +24,19 @@ curl -s localhost:3000/string
 echo
 curl -s localhost:3000/string
 echo
+
+curl -s http://localhost:3000/journal | jq .
+echo
+
+curl -s -d '{"id": "test", "a": 2, "b": 2, "c": 5}' http://localhost:3000/sums
+curl -s -d '{"id": "test", "a": 2, "b": 3, "c": 5}' http://localhost:3000/sums
+
+sleep 1
+TS=$(date +%s)
+echo "TS: $TS"
+
+curl -s -d '{"id": "test'$TS'-a", "a": 2, "b": 2, "c": 5}' http://localhost:3000/sums
+curl -s -d '{"id": "test'$TS'-b", "a": 2, "b": 3, "c": 5}' http://localhost:3000/sums
+
+curl -s http://localhost:3000/journal | jq .
+echo