pub use vta_sdk::contexts::ContextRecord;
use chrono::Utc;
use vta_sdk::context_path::parent_path;
use vta_sdk::context_policy::ContextPolicy;
use vti_common::error::AppError;
use vti_common::store::KeyspaceHandle;
fn ctx_key(id: &str) -> String {
format!("ctx:{id}")
}
pub async fn get_context(ks: &KeyspaceHandle, id: &str) -> Result<Option<ContextRecord>, AppError> {
ks.get(ctx_key(id)).await
}
pub async fn effective_context_policy(
ks: &KeyspaceHandle,
context_id: &str,
) -> Result<ContextPolicy, AppError> {
let mut ids: Vec<String> = Vec::new();
let mut cur: Option<String> = Some(context_id.to_string());
while let Some(id) = cur {
cur = parent_path(&id).map(str::to_string);
ids.push(id);
}
ids.reverse();
let mut policies: Vec<ContextPolicy> = Vec::new();
for id in &ids {
if let Some(rec) = get_context(ks, id).await?
&& let Some(policy) = rec.context_policy
{
policies.push(policy);
}
}
Ok(ContextPolicy::resolve(policies.iter()))
}
pub async fn enforce_daily_quota(
ks: &KeyspaceHandle,
context_id: &str,
op_class: &str,
limit: u64,
) -> Result<(), AppError> {
let day = Utc::now().format("%Y-%m-%d");
let key = format!("quota:{context_id}:{op_class}:{day}");
let slot = vti_common::store::counter::allocate_u32(ks, &key).await?;
if u64::from(slot) >= limit {
return Err(AppError::Forbidden(format!(
"daily quota exceeded for {op_class} in context {context_id} ({limit}/day)"
)));
}
Ok(())
}
pub async fn store_context(ks: &KeyspaceHandle, record: &ContextRecord) -> Result<(), AppError> {
ks.insert(ctx_key(&record.id), record).await
}
pub async fn store_new_context(
ks: &KeyspaceHandle,
record: &ContextRecord,
) -> Result<bool, AppError> {
ks.insert_if_absent(ctx_key(&record.id), record).await
}
pub async fn delete_context(ks: &KeyspaceHandle, id: &str) -> Result<(), AppError> {
ks.remove(ctx_key(id)).await
}
pub async fn list_contexts(ks: &KeyspaceHandle) -> Result<Vec<ContextRecord>, AppError> {
let raw = ks.prefix_iter_raw("ctx:").await?;
let mut records = Vec::with_capacity(raw.len());
let mut skipped = 0usize;
for (key, value) in raw {
match serde_json::from_slice::<ContextRecord>(&value) {
Ok(record) => records.push(record),
Err(e) => {
skipped += 1;
tracing::warn!(
key = %String::from_utf8_lossy(&key),
error = %e,
"skipping undeserializable context row in list_contexts"
);
}
}
}
if skipped > 0 {
tracing::warn!(skipped, "list_contexts skipped corrupt rows");
}
Ok(records)
}
pub async fn allocate_context_index(
ks: &KeyspaceHandle,
base_prefix: &str,
counter_key: &str,
) -> Result<(u32, String), AppError> {
let current = vti_common::store::counter::allocate_u32(ks, counter_key).await?;
let base_path = format!("{base_prefix}/{current}'");
Ok((current, base_path))
}
pub fn validate_slug(id: &str) -> Result<(), AppError> {
if id.is_empty() {
return Err(AppError::Validation("context id cannot be empty".into()));
}
if id.len() > 64 {
return Err(AppError::Validation(
"context id must be 64 characters or fewer".into(),
));
}
if !id
.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
{
return Err(AppError::Validation(
"context id must contain only lowercase alphanumeric characters and hyphens".into(),
));
}
if id.starts_with('-') || id.ends_with('-') {
return Err(AppError::Validation(
"context id must not start or end with a hyphen".into(),
));
}
Ok(())
}
pub async fn create_context(
contexts_ks: &KeyspaceHandle,
id: &str,
name: &str,
) -> Result<ContextRecord, Box<dyn std::error::Error>> {
validate_slug(id).map_err(|e| format!("{e}"))?;
let (index, base_path) = allocate_context_index(contexts_ks, CONTEXT_KEY_BASE, "ctx_counter")
.await
.map_err(|e| format!("{e}"))?;
let now = Utc::now();
let record = ContextRecord {
id: id.to_string(),
name: name.to_string(),
did: None,
description: None,
parent: None,
base_path,
index,
created_at: now,
updated_at: now,
context_policy: None,
};
if !store_new_context(contexts_ks, &record)
.await
.map_err(|e| format!("{e}"))?
{
return Err(format!("context already exists: {id}").into());
}
Ok(record)
}
pub const CONTEXT_KEY_BASE: &str = "m/26'/2'";
#[cfg(test)]
mod tests {
use super::*;
use vti_common::config::StoreConfig;
use vti_common::store::Store;
fn temp_ks() -> (KeyspaceHandle, tempfile::TempDir) {
let dir = tempfile::tempdir().expect("tempdir");
let store = Store::open(&StoreConfig {
data_dir: dir.path().to_path_buf(),
})
.expect("open store");
(
store.keyspace(vta_keyspaces::CONTEXTS).expect("keyspace"),
dir,
)
}
#[tokio::test]
async fn daily_quota_allows_up_to_limit_then_forbids() {
let (ks, _dir) = temp_ks();
enforce_daily_quota(&ks, "sales", "sign", 2).await.unwrap();
enforce_daily_quota(&ks, "sales", "sign", 2).await.unwrap();
let err = enforce_daily_quota(&ks, "sales", "sign", 2).await;
assert!(matches!(err, Err(AppError::Forbidden(_))), "{err:?}");
enforce_daily_quota(&ks, "sales", "vault/release", 1)
.await
.unwrap();
let err2 = enforce_daily_quota(&ks, "sales", "vault/release", 1).await;
assert!(matches!(err2, Err(AppError::Forbidden(_))), "{err2:?}");
enforce_daily_quota(&ks, "eng", "sign", 2).await.unwrap();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn allocate_context_index_is_collision_free_under_concurrency() {
let (ks, _dir) = temp_ks();
let n = 64usize;
let mut handles = Vec::with_capacity(n);
for _ in 0..n {
let ks = ks.clone();
handles.push(tokio::spawn(async move {
allocate_context_index(&ks, CONTEXT_KEY_BASE, "ctx_counter")
.await
.expect("alloc")
}));
}
let mut paths = std::collections::HashSet::with_capacity(n);
for h in handles {
let (_, base_path) = h.await.expect("join");
assert!(
paths.insert(base_path.clone()),
"duplicate context base path {base_path}"
);
}
assert_eq!(paths.len(), n);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_same_id_creates_admit_exactly_one() {
let (ks, _dir) = temp_ks();
let mut handles = Vec::new();
for _ in 0..16 {
let ks = ks.clone();
handles.push(tokio::spawn(async move {
create_context(&ks, "contested", "Contested").await.ok()
}));
}
let mut winners = Vec::new();
for h in handles {
if let Some(rec) = h.await.expect("join") {
winners.push(rec);
}
}
assert_eq!(winners.len(), 1, "exactly one same-id create may win");
let stored = get_context(&ks, "contested")
.await
.expect("get")
.expect("record exists");
assert_eq!(
stored.base_path, winners[0].base_path,
"stored record must be the winner's — no overwrite by losers"
);
}
#[test]
fn validate_slug_accepts_canonical_shapes() {
for ok in ["a", "vta", "trust-registry", "ctx-1", "a1-b2-c3"] {
validate_slug(ok).unwrap_or_else(|e| panic!("{ok:?} should be valid: {e}"));
}
validate_slug(&"a".repeat(64)).expect("64 chars is the limit, not over it");
}
#[test]
fn validate_slug_rejects_non_canonical_shapes() {
for bad in [
"MyApp", "ctx.v2", "my_app", "-lead", "trail-", "", "a/b", "a b",
] {
assert!(
validate_slug(bad).is_err(),
"{bad:?} must be rejected as a context id"
);
}
assert!(validate_slug(&"a".repeat(65)).is_err(), "65 chars is over");
}
#[tokio::test]
async fn low_level_create_context_enforces_the_slug_rule() {
let (ks, _dir) = temp_ks();
let err = create_context(&ks, "MyApp", "My App")
.await
.expect_err("non-slug id must be refused");
assert!(
err.to_string().contains("lowercase"),
"error should name the rule, got: {err}"
);
assert!(
get_context(&ks, "MyApp").await.expect("get").is_none(),
"a rejected id must not have been persisted"
);
create_context(&ks, "myapp", "My App")
.await
.expect("the slug form is accepted");
}
}