use chrono::Utc;
use serde_json::{json, Value};
use uuid::Uuid;
use crate::actor::Actor;
use crate::error::{Error, Result};
use crate::owner_ref::{OwnerKind, OwnerRef};
use crate::runtime::Valence;
use crate::schema::SchemaRegistry;
use crate::RecordId;
pub const SKIP_OWNERSHIP_TABLES: &[&str] =
&["valence_data_ownership", "valence_ownership_transfer"];
pub fn skip_ownership_for_table(table: &str) -> bool {
SKIP_OWNERSHIP_TABLES.contains(&table)
}
const OWNERSHIP_ROW_UUID_NS: Uuid = Uuid::from_u128(0xe7b3c1d0_f2a4_5b8e_9c6d_a1b2e3f40516);
pub fn ownership_row_id(valence_model: &str, record_id: &str) -> String {
let name = format!("{valence_model}\n{record_id}");
Uuid::new_v5(&OWNERSHIP_ROW_UUID_NS, name.as_bytes()).to_string()
}
pub fn system_valence(v: &Valence) -> Valence {
v.with_actor(Actor::System {
operation: "valence_data_ownership".to_string(),
})
}
pub fn normalize_pending_deletion_query_value(raw: &str) -> String {
let s = raw.trim();
if s.len() >= 2 {
let bytes = s.as_bytes();
if (bytes[0] == b'\'' && bytes[s.len() - 1] == b'\'')
|| (bytes[0] == b'"' && bytes[s.len() - 1] == b'"')
{
return s[1..s.len() - 1].to_string();
}
}
s.to_string()
}
pub fn ownership_colocate_enabled() -> bool {
static ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ENABLED.get_or_init(|| {
!matches!(
std::env::var("VALENCE_OWNERSHIP_COLOCATE").as_deref(),
Ok("0") | Ok("false") | Ok("FALSE")
)
})
}
pub fn ownership_unified_fetch_enabled() -> bool {
static ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ENABLED.get_or_init(|| {
!matches!(
std::env::var("VALENCE_OWNERSHIP_UNIFIED_FETCH").as_deref(),
Ok("0") | Ok("false") | Ok("FALSE")
)
})
}
pub fn ownership_get_join_enabled() -> bool {
static ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ENABLED.get_or_init(|| {
!matches!(
std::env::var("VALENCE_OWNERSHIP_GET_JOIN").as_deref(),
Ok("0") | Ok("false") | Ok("FALSE")
)
})
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum OwnershipGateStatus {
NotFetched,
Absent,
Status(String),
}
impl OwnershipGateStatus {
pub fn is_pending_deletion(&self) -> bool {
matches!(self, Self::Status(s) if s == "pending_deletion")
}
#[must_use]
pub(crate) fn from_optional_status(raw: Option<Value>) -> Self {
match raw.and_then(|v| v.as_str().map(str::to_string)) {
Some(s) => Self::Status(s),
None => Self::Absent,
}
}
}
#[derive(Clone, Debug)]
pub struct RecordOwnershipBundle {
pub row: Option<Value>,
pub ownership_status: OwnershipGateStatus,
}
#[derive(Debug, Clone, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
pub struct OwnerSchemaRowCount {
pub valence_model: String,
pub active_rows: u64,
pub pending_deletion_rows: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
pub struct OwnerDataSummary {
pub owned_rows: u64,
pub tables_with_data: u64,
pub pending_deletion_rows: u64,
pub rows_by_schema: Vec<OwnerSchemaRowCount>,
}
pub const OWNER_SUMMARY_CONCURRENCY: usize = 8;
pub fn schema_skipped_for_owner_summary(table: &str, schema: &crate::Schema) -> bool {
skip_ownership_for_table(table) || schema.ownership.as_ref().is_some_and(|o| o.system_owned)
}
#[allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "backend count values may be represented as floating-point JSON"
)]
pub fn parse_count_from_row(row: &Value) -> u64 {
row.as_u64()
.or_else(|| row.as_i64().map(|n| n as u64))
.or_else(|| row.as_f64().map(|f| f as u64))
.or_else(|| {
row.get("n")
.or_else(|| row.get("count"))
.and_then(|v| v.as_u64().or_else(|| v.as_f64().map(|f| f as u64)))
})
.unwrap_or(0)
}
fn push_unique_owner_id(values: &mut Vec<String>, id: impl Into<String>) {
let id = id.into();
if !values.iter().any(|v| v == &id) {
values.push(id);
}
}
pub fn owner_id_query_values(owner_id: &str, owner_type: &str) -> Vec<String> {
let mut values = vec![owner_id.to_string()];
if owner_type != OwnerKind::User.as_str() {
return values;
}
if let Some(bare) = owner_id.strip_prefix("user:") {
if !bare.is_empty() {
push_unique_owner_id(&mut values, bare);
}
} else if !owner_id.contains(':') {
push_unique_owner_id(&mut values, format!("user:{owner_id}"));
}
values
}
pub fn parse_owner_kind(s: &str) -> OwnerKind {
match s {
"user" => OwnerKind::User,
"account" => OwnerKind::Account,
"application" => OwnerKind::Application,
"service" => OwnerKind::Service,
_ => OwnerKind::System,
}
}
pub fn owner_ref_from_ownership_json(v: &Value) -> Option<OwnerRef> {
let owner_id = v.get("owner_id")?.as_str()?.to_string();
let t = v.get("owner_type")?.as_str()?;
Some(OwnerRef {
owner_id,
owner_kind: parse_owner_kind(t),
})
}
pub fn normalize_record_id_for_ownership(entity_id: &str) -> String {
let s = entity_id.trim();
let Some((head, rest)) = s.split_once(':') else {
return s.to_string();
};
if head.is_empty() || rest.is_empty() {
return s.to_string();
}
if SchemaRegistry::global().get_schema(head).is_some() {
return rest.to_string();
}
if !rest.contains(':') {
return rest.to_string();
}
s.to_string()
}
pub async fn append_transfer_history_row(
valence_model: &str,
record_id: &str,
oid: &str,
from_owner_id: &str,
from_owner_type: &str,
new_owner: &OwnerRef,
reason: Option<String>,
v: &Valence,
) -> Result<()> {
let transferred_by = match v.actor() {
Actor::User { user_id } => user_id.clone(),
Actor::ServiceUser { service_name } => format!("service:{service_name}"),
Actor::System { operation } => format!("system:{operation}"),
Actor::Anonymous => "anonymous".to_string(),
};
let tid = Uuid::new_v4().to_string();
let rid = RecordId::new("valence_data_ownership", oid);
let transfer = json!({
"id": tid,
"ownership_id": rid,
"from_owner_id": from_owner_id,
"from_owner_type": from_owner_type,
"to_owner_id": new_owner.owner_id,
"to_owner_type": new_owner.owner_kind.as_str(),
"transferred_at": Utc::now(),
"transferred_by": transferred_by,
"reason": reason,
});
let sys = system_valence(v);
let tback = sys.backend_for_table("valence_ownership_transfer")?;
tback
.create_record("valence_ownership_transfer", transfer)
.await
.map_err(|e| Error::database(e.to_string()))?;
crate::instrumentation::record_ownership_transfer(
valence_model,
record_id,
from_owner_id,
from_owner_type,
&new_owner.owner_id,
new_owner.owner_kind.as_str(),
&transferred_by,
);
Ok(())
}