Menu
Public repository / master
AkurAI-Notes / crates/notes-api/src/handlers/crud.rs
Latest change 28fbbec68cbd9555cfd06328a0616d17d6e001f6 - AKURAI-NOTES-10 AKURAI-NOTES-11 AKURAI-NOTES-12: consolidate notes cleanup by Ólafur Búi Ólafsson
use crate::auth::session;
use crate::embed::{maybe_reembed, EmbedStore};
use crate::handlers::RouteHandler;
use crate::router_setup::{
bad_request, internal_error, json_response, not_found, unauthorized, upper_bound, NotesState,
};
use crate::telemetry::{EmitOutcome, NotesTelemetry};
use akurai_http::{Request, Response};
use akurai_json::Value;
use std::sync::{Arc, Mutex};
fn parse_body(req: &Request) -> Result<Value, String> {
if req.body.is_empty() {
return Err("empty request body".into());
}
let text = std::str::from_utf8(&req.body).map_err(|e| format!("body not UTF-8: {e}"))?;
akurai_json::parse(text).map_err(|e| format!("invalid JSON: {e}"))
}
fn extract_id(path: &str) -> Option<u64> {
let segments: Vec<&str> = path.split('/').collect();
segments.last().and_then(|s| s.parse::<u64>().ok())
}
fn build_key(coll_name: &str, id: u64) -> Vec<u8> {
let mut key = format!("{}:", coll_name).into_bytes();
key.extend_from_slice(&id.to_be_bytes());
key
}
/// Allocate the next id for a collection, reconciling the persisted
/// `<collection>:_counter` against the actual max existing id.
///
/// A counter-only `counter + 1` is unsafe: after a DB restore/import the records
/// keep their original ids but the counter can be left stale below the real max,
/// so it would hand out an id that already exists and the create would overwrite
/// that record. Reconciling against the real max id (and writing the counter
/// forward) makes allocation self-healing. Caller must hold the db lock.
pub fn allocate_next_id(db: &mut akurai_storage::BTree, collection: &str) -> Result<u64, String> {
let counter_key = format!("{}:_counter", collection);
let prefix = format!("{}:", collection);
let start = prefix.as_bytes().to_vec();
let end = upper_bound(&start);
let from_counter = match db
.get(counter_key.as_bytes())
.map_err(|e| format!("counter read: {e}"))?
{
Some(bytes) => match String::from_utf8_lossy(&bytes).trim().parse::<u64>() {
Ok(id) if id < u64::MAX => id + 1,
Ok(_) => return Err("ID space exhausted".into()),
Err(_) => 1,
},
None => 1,
};
// Never reuse an id that already exists. Record keys are `<collection>:` +
// id.to_be_bytes() (8 bytes); the only other in-range key is the 8-byte
// `_counter` suffix, skipped explicitly.
let mut max_existing = 0u64;
if let Ok(entries) = db.range(&start, &end) {
let plen = prefix.len();
for (key, _val) in entries {
let suffix = &key[plen..];
if suffix.len() == 8 && suffix != b"_counter" {
max_existing = max_existing.max(u64::from_be_bytes(suffix.try_into().unwrap()));
}
}
}
let next_id = from_counter.max(max_existing.saturating_add(1));
db.insert(counter_key.as_bytes(), next_id.to_string().as_bytes())
.map_err(|e| format!("counter write: {e}"))?;
db.commit().map_err(|e| format!("commit: {e}"))?;
Ok(next_id)
}
struct OwnerIds {
sub: String,
email: String,
}
fn get_owner_ids(state: &NotesState, req: &Request) -> Option<OwnerIds> {
let sid = session::extract_session_id(req)?;
let mut db = state.db.lock().ok()?;
let user = session::get_user(&mut db, &sid)?;
Some(OwnerIds {
sub: user.sub,
email: user.email,
})
}
fn owner_matches(record: &Value, owner: &OwnerIds) -> bool {
match record.get("ownerSub").and_then(|v| v.as_str()) {
Some(s) => s == owner.sub || s == owner.email,
None => false,
}
}
/// Count notes owned by the given sub in the collection.
fn count_notes(db: &mut akurai_storage::BTree, coll: &str, owner: &OwnerIds) -> i64 {
let prefix = format!("{coll}:").into_bytes();
let end = upper_bound(&prefix);
let mut count = 0i64;
if let Ok(entries) = db.range(&prefix, &end) {
for (_key_bytes, val_bytes) in entries {
if let Ok(json) = akurai_json::parse(std::str::from_utf8(&val_bytes).unwrap_or("")) {
if owner_matches(&json, owner) {
count += 1;
}
}
}
}
count
}
/// Emit note-related telemetry facts after a mutation.
fn emit_note_facts(telemetry: Option<&NotesTelemetry>, now_secs: i64, total: i64) {
let Some(t) = telemetry else { return };
match t.emit_note_count(now_secs, total) {
EmitOutcome::Rejected | EmitOutcome::Closed => {
eprintln!("[telemetry] note_count emit failed: {total} notes");
}
_ => {}
}
if total == 1 && t.emit_first_note(now_secs) == EmitOutcome::Rejected {
eprintln!("[telemetry] first_note emit failed");
}
}
pub fn list_route(state: Arc<Mutex<NotesState>>, coll: &'static str) -> RouteHandler {
Box::new(move |req: &Request| {
let state_lock = match state.lock() {
Ok(s) => s,
Err(_) => return internal_error("lock poisoned"),
};
let owner = match get_owner_ids(&state_lock, req) {
Some(o) => o,
None => {
return unauthorized();
}
};
let mut db = match state_lock.db.lock() {
Ok(db) => db,
Err(_) => return internal_error("db lock poisoned"),
};
let prefix = format!("{}:", coll).into_bytes();
let end = upper_bound(&prefix);
let mut records = Vec::new();
if let Ok(entries) = db.range(&prefix, &end) {
for (_key_bytes, val_bytes) in entries {
if let Ok(json) = akurai_json::parse(std::str::from_utf8(&val_bytes).unwrap_or(""))
{
if owner_matches(&json, &owner) {
records.push(json);
}
}
}
}
json_response(Value::Array(records))
})
}
pub fn get_route(state: Arc<Mutex<NotesState>>, coll: &'static str) -> RouteHandler {
Box::new(move |req: &Request| {
let id = match extract_id(&req.path) {
Some(id) => id,
None => return bad_request("missing id"),
};
let state_lock = match state.lock() {
Ok(s) => s,
Err(_) => return internal_error("lock poisoned"),
};
let owner = match get_owner_ids(&state_lock, req) {
Some(o) => o,
None => {
return unauthorized();
}
};
let mut db = match state_lock.db.lock() {
Ok(db) => db,
Err(_) => return internal_error("db lock poisoned"),
};
let key = build_key(coll, id);
match db.get(&key) {
Ok(Some(val)) => match akurai_json::parse(std::str::from_utf8(&val).unwrap_or("")) {
Ok(json) => {
if !owner_matches(&json, &owner) {
return not_found(&format!("{coll}/{id}"));
}
json_response(json)
}
Err(_) => internal_error("corrupted data"),
},
Ok(None) => not_found(&format!("{coll}/{id}")),
Err(e) => internal_error(&format!("storage: {e}")),
}
})
}
pub fn create_route(state: Arc<Mutex<NotesState>>, coll: &'static str) -> RouteHandler {
Box::new(move |req: &Request| {
let body = match parse_body(req) {
Ok(v) => v,
Err(e) => return bad_request(&e),
};
let state_lock = match state.lock() {
Ok(s) => s,
Err(_) => return internal_error("lock poisoned"),
};
let owner = match get_owner_ids(&state_lock, req) {
Some(o) => o,
None => {
return unauthorized();
}
};
let mut db = match state_lock.db.lock() {
Ok(db) => db,
Err(_) => return internal_error("db lock poisoned"),
};
let now = NotesState::now();
let next_id = match allocate_next_id(&mut db, coll) {
Ok(id) => id,
Err(e) => return internal_error(&e),
};
let mut obj = match body {
Value::Object(fields) => fields,
_ => return bad_request("expected JSON object"),
};
// Force ownerSub to the session user's sub
let mut found_owner = false;
for pair in obj.iter_mut() {
if pair.0 == "ownerSub" {
pair.1 = Value::Str(owner.sub.clone());
found_owner = true;
break;
}
}
if !found_owner {
obj.push(("ownerSub".into(), Value::Str(owner.sub.clone())));
}
obj.push(("id".into(), Value::Int(next_id as i64)));
obj.push(("createdAt".into(), Value::Int(now)));
obj.push(("updatedAt".into(), Value::Int(now)));
let record = Value::Object(obj);
let key = build_key(coll, next_id);
let json_str = record.to_json();
match db.insert(&key, json_str.as_bytes()) {
Ok(_) => {
if let Err(e) = db.commit() {
return internal_error(&format!("commit: {e}"));
}
let body_str = record
.get("body")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
if !body_str.is_empty() {
maybe_reembed(
&mut db,
&state_lock.embed_url,
&state_lock.embed_model,
&state_lock.embed_api_key,
next_id as i64,
&body_str,
);
}
// Emit telemetry after successful creation (best-effort, non-blocking).
if coll == "notes" {
let total = count_notes(&mut db, coll, &owner);
emit_note_facts(state_lock.telemetry.as_ref(), now, total);
}
let body = record.to_json();
Response::new(201)
.with_header("Content-Type", "application/json")
.with_header("Access-Control-Allow-Origin", "*")
.with_body("application/json", body.into_bytes())
}
Err(e) => internal_error(&format!("storage: {e}")),
}
})
}
pub fn update_route(state: Arc<Mutex<NotesState>>, coll: &'static str) -> RouteHandler {
Box::new(move |req: &Request| {
let id = match extract_id(&req.path) {
Some(id) => id,
None => return bad_request("missing id"),
};
let body = match parse_body(req) {
Ok(v) => v,
Err(e) => return bad_request(&e),
};
let state_lock = match state.lock() {
Ok(s) => s,
Err(_) => return internal_error("lock poisoned"),
};
let owner = match get_owner_ids(&state_lock, req) {
Some(o) => o,
None => {
return unauthorized();
}
};
let mut db = match state_lock.db.lock() {
Ok(db) => db,
Err(_) => return internal_error("db lock poisoned"),
};
let key = build_key(coll, id);
let existing = match db.get(&key) {
Ok(Some(val)) => match akurai_json::parse(std::str::from_utf8(&val).unwrap_or("")) {
Ok(v) => v,
Err(_) => return internal_error("corrupted data"),
},
Ok(None) => return not_found(&format!("{coll}/{id}")),
Err(e) => return internal_error(&format!("storage: {e}")),
};
let old_body = existing
.get("body")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
// Owner check — match by sub or email (before destructuring)
if !owner_matches(&existing, &owner) {
return not_found(&format!("{coll}/{id}"));
}
let mut existing_fields = match existing {
Value::Object(f) => f,
_ => return internal_error("corrupted record"),
};
if let Value::Object(update_fields) = body {
for (k, v) in update_fields {
if k != "id" && k != "createdAt" && k != "ownerSub" {
let mut found = false;
for pair in existing_fields.iter_mut() {
if pair.0 == k {
pair.1 = v.clone();
found = true;
break;
}
}
if !found {
existing_fields.push((k.clone(), v.clone()));
}
}
}
}
let now = NotesState::now();
let mut found_updated = false;
for pair in existing_fields.iter_mut() {
if pair.0 == "updatedAt" {
pair.1 = Value::Int(now);
found_updated = true;
break;
}
}
if !found_updated {
existing_fields.push(("updatedAt".into(), Value::Int(now)));
}
let record = Value::Object(existing_fields);
let json_str = record.to_json();
match db.insert(&key, json_str.as_bytes()) {
Ok(_) => {
if let Err(e) = db.commit() {
return internal_error(&format!("commit: {e}"));
}
let new_body = record
.get("body")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
if old_body != new_body {
if new_body.is_empty() {
let _ = EmbedStore { db: &mut db }.remove(id as i64);
} else {
maybe_reembed(
&mut db,
&state_lock.embed_url,
&state_lock.embed_model,
&state_lock.embed_api_key,
id as i64,
&new_body,
);
}
}
json_response(record)
}
Err(e) => internal_error(&format!("storage: {e}")),
}
})
}
pub fn delete_route(state: Arc<Mutex<NotesState>>, coll: &'static str) -> RouteHandler {
Box::new(move |req: &Request| {
let id = match extract_id(&req.path) {
Some(id) => id,
None => return bad_request("missing id"),
};
let state_lock = match state.lock() {
Ok(s) => s,
Err(_) => return internal_error("lock poisoned"),
};
let owner = match get_owner_ids(&state_lock, req) {
Some(o) => o,
None => {
return unauthorized();
}
};
let mut db = match state_lock.db.lock() {
Ok(db) => db,
Err(_) => return internal_error("db lock poisoned"),
};
let key = build_key(coll, id);
// Owner check before delete
match db.get(&key) {
Ok(Some(val)) => {
if let Ok(json) = akurai_json::parse(std::str::from_utf8(&val).unwrap_or("")) {
if !owner_matches(&json, &owner) {
return not_found(&format!("{coll}/{id}"));
}
}
}
Ok(None) => return not_found(&format!("{coll}/{id}")),
Err(e) => return internal_error(&format!("storage: {e}")),
}
match db.delete(&key) {
Ok(true) => {
if let Err(e) = db.commit() {
return internal_error(&format!("commit: {e}"));
}
let mut store = EmbedStore { db: &mut db };
let _ = store.remove(id as i64);
json_response(Value::Object(vec![
("deleted".into(), Value::Bool(true)),
("id".into(), Value::Int(id as i64)),
]))
}
Ok(false) => not_found(&format!("{coll}/{id}")),
Err(e) => internal_error(&format!("storage: {e}")),
}
})
}
#[cfg(test)]
mod alloc_tests {
use super::*;
use akurai_storage::BTree;
fn temp_db() -> BTree {
let mut p = std::env::temp_dir();
let n = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
p.push(format!("notes_alloc_test_{}_{}", std::process::id(), n));
BTree::open(&p).expect("open temp db")
}
fn put_record(db: &mut BTree, coll: &str, id: u64) {
let mut key = format!("{coll}:").into_bytes();
key.extend_from_slice(&id.to_be_bytes());
db.insert(&key, format!("{{\"id\":{id}}}").as_bytes())
.unwrap();
db.commit().unwrap();
}
#[test]
fn stale_counter_never_collides_with_existing_ids() {
let mut db = temp_db();
put_record(&mut db, "notes", 5);
put_record(&mut db, "notes", 34);
db.insert(b"notes:_counter", b"17").unwrap();
db.commit().unwrap();
assert_eq!(allocate_next_id(&mut db, "notes").unwrap(), 35);
let c = db.get(b"notes:_counter").unwrap().unwrap();
assert_eq!(String::from_utf8_lossy(&c), "35");
assert_eq!(allocate_next_id(&mut db, "notes").unwrap(), 36);
}
#[test]
fn empty_collection_starts_at_one() {
let mut db = temp_db();
assert_eq!(allocate_next_id(&mut db, "notes").unwrap(), 1);
assert_eq!(allocate_next_id(&mut db, "notes").unwrap(), 2);
}
#[test]
fn healthy_counter_ahead_of_data_is_respected() {
let mut db = temp_db();
put_record(&mut db, "notes", 3);
db.insert(b"notes:_counter", b"10").unwrap();
db.commit().unwrap();
assert_eq!(allocate_next_id(&mut db, "notes").unwrap(), 11);
}
#[test]
fn counter_key_is_excluded_from_scan() {
let mut db = temp_db();
put_record(&mut db, "notes", 2);
db.insert(b"notes:_counter", b"1").unwrap();
db.commit().unwrap();
assert_eq!(allocate_next_id(&mut db, "notes").unwrap(), 3);
}
}