use std::collections::HashMap;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, Mutex};
use sha2::Digest as _;
use crate::control::ControlStore;
use crate::doc::{impl_doc, DocRef, ShardedMap};
use crate::error::Result;
use crate::flow::{day_of, iso8601_utc, DAY_MS};
pub const ACCOUNT_AUDIT_REF_PREFIX: &str = "refs/gunnar/account-audit/";
pub const DEFAULT_RETENTION_DAYS: u32 = 365;
pub const GENESIS: &str = crate::audit::GENESIS;
const DOMAIN: &str = "gunnar-account-audit-v1";
pub const MAX_PAGE: usize = 1000;
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Outcome {
Ok,
Refused,
}
impl Outcome {
pub fn as_str(self) -> &'static str {
match self {
Outcome::Ok => "ok",
Outcome::Refused => "refused",
}
}
}
#[derive(Clone, Copy, Debug)]
pub struct AccountEvent<'a> {
pub kind: &'a str,
pub actor: &'a str,
pub actor_role: &'a str,
pub target: &'a str,
pub detail: &'a str,
pub outcome: Outcome,
}
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AccountEntry {
pub seq: u64,
pub at_unix_ms: i64,
pub kind: String,
pub actor: String,
pub actor_role: String,
pub target: String,
pub detail: String,
pub outcome: String,
pub prev: String,
pub hash: String,
}
impl AccountEntry {
pub fn compute_hash(&self, account: &str) -> String {
let mut hasher = sha2::Sha256::new();
let mut field = |bytes: &[u8]| {
hasher.update((bytes.len() as u64).to_le_bytes());
hasher.update(bytes);
};
field(DOMAIN.as_bytes());
field(account.as_bytes());
field(self.prev.as_bytes());
field(&self.seq.to_le_bytes());
field(&self.at_unix_ms.to_le_bytes());
field(self.kind.as_bytes());
field(self.actor.as_bytes());
field(self.actor_role.as_bytes());
field(self.target.as_bytes());
field(self.detail.as_bytes());
field(self.outcome.as_bytes());
hex::encode(hasher.finalize())
}
fn is_kind(&self, kind: &str) -> bool {
self.kind == kind
|| (self.kind.len() > kind.len()
&& self.kind.starts_with(kind)
&& self.kind.as_bytes()[kind.len()] == b'.')
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct AuditQuery {
pub kinds: Vec<String>,
pub actor: String,
pub since_unix_ms: i64,
pub until_unix_ms: i64,
pub text: String,
pub before_seq: Option<u64>,
pub limit: usize,
}
impl AuditQuery {
fn matches(&self, entry: &AccountEntry, text_lower: &str) -> bool {
if !self.kinds.is_empty() && !self.kinds.iter().any(|k| entry.is_kind(k)) {
return false;
}
if !self.actor.is_empty() && entry.actor != self.actor {
return false;
}
if self.since_unix_ms > 0 && entry.at_unix_ms < self.since_unix_ms {
return false;
}
if self.until_unix_ms > 0 && entry.at_unix_ms > self.until_unix_ms {
return false;
}
if !text_lower.is_empty() {
let hit = [&entry.target, &entry.detail, &entry.actor, &entry.kind]
.iter()
.any(|field| field.to_lowercase().contains(text_lower));
if !hit {
return false;
}
}
true
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct AuditPage {
pub entries: Vec<AccountEntry>,
pub next_before_seq: Option<u64>,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ChainReport {
pub holds: bool,
pub entries: u64,
pub tip_hash: String,
pub anchor_seq: Option<u64>,
pub broken_at_seq: Option<u64>,
pub what: String,
}
pub fn verify_entries(
account: &str,
anchor: Option<(u64, &str)>,
entries: &[AccountEntry],
) -> ChainReport {
let (mut want_seq, mut prev) = match anchor {
Some((seq, hash)) => (seq + 1, hash.to_owned()),
None => (0, GENESIS.to_owned()),
};
let mut report = ChainReport {
holds: true,
entries: 0,
tip_hash: prev.clone(),
anchor_seq: anchor.map(|(seq, _)| seq),
broken_at_seq: None,
what: String::new(),
};
for entry in entries {
let what = if entry.seq != want_seq {
Some(format!(
"entry {} is where entry {want_seq} should be: one is missing or out of order",
entry.seq
))
} else if entry.prev != prev {
Some(format!(
"entry {} does not name the entry before it",
entry.seq
))
} else if entry.hash != entry.compute_hash(account) {
Some(format!(
"entry {} was changed after it was written",
entry.seq
))
} else {
None
};
if let Some(what) = what {
report.holds = false;
report.broken_at_seq = Some(entry.seq);
report.what = what;
return report;
}
report.entries += 1;
want_seq = entry.seq + 1;
prev = entry.hash.clone();
report.tip_hash = prev.clone();
}
report
}
#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct Head {
#[serde(default)]
next_seq: u64,
#[serde(default)]
tip: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
anchor_seq: Option<u64>,
#[serde(default, skip_serializing_if = "String::is_empty")]
anchor_hash: String,
#[serde(default)]
days: Vec<i64>,
}
impl Head {
fn prev(&self) -> String {
if !self.tip.is_empty() {
self.tip.clone()
} else if !self.anchor_hash.is_empty() {
self.anchor_hash.clone()
} else {
GENESIS.to_owned()
}
}
}
#[derive(Clone, Default, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct AccountAuditDoc {
#[serde(default)]
head: Head,
#[serde(default)]
days: ShardedMap<Vec<AccountEntry>>,
}
impl_doc!(AccountAuditDoc { days });
fn day_key(day: i64) -> String {
format!("{day:010}")
}
pub struct AccountAuditLog {
store: Arc<dyn ControlStore>,
docs: Mutex<HashMap<String, DocRef<AccountAuditDoc>>>,
retention_days: AtomicU32,
}
impl std::fmt::Debug for AccountAuditLog {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AccountAuditLog")
.field("retention_days", &self.retention_days())
.finish()
}
}
impl AccountAuditLog {
pub fn new(store: Arc<dyn ControlStore>) -> Result<Self> {
Ok(AccountAuditLog {
store,
docs: Mutex::new(HashMap::new()),
retention_days: AtomicU32::new(DEFAULT_RETENTION_DAYS),
})
}
pub fn with_retention_days(self, days: u32) -> Self {
self.retention_days.store(days.max(1), Ordering::Relaxed);
self
}
pub fn retention_days(&self) -> u32 {
self.retention_days.load(Ordering::Relaxed)
}
fn doc(&self, account: &str) -> Result<DocRef<AccountAuditDoc>> {
crate::sandbox::check_component(account, account)?;
let mut docs = self.docs.lock().unwrap_or_else(|p| p.into_inner());
if let Some(doc) = docs.get(account) {
return Ok(doc.clone());
}
let doc = DocRef::named(
Arc::clone(&self.store),
std::borrow::Cow::Owned(format!("{ACCOUNT_AUDIT_REF_PREFIX}{account}")),
)?;
docs.insert(account.to_owned(), doc.clone());
Ok(doc)
}
pub fn append(
&self,
account: &str,
event: AccountEvent<'_>,
at_unix_ms: i64,
) -> Result<AccountEntry> {
let doc = self.doc(account)?;
let cutoff_day = day_of(at_unix_ms) - i64::from(self.retention_days());
doc.update(|doc| {
Self::drop_before(doc, cutoff_day)?;
let mut entry = AccountEntry {
seq: doc.head.next_seq,
at_unix_ms,
kind: event.kind.to_owned(),
actor: event.actor.to_owned(),
actor_role: event.actor_role.to_owned(),
target: event.target.to_owned(),
detail: event.detail.to_owned(),
outcome: event.outcome.as_str().to_owned(),
prev: doc.head.prev(),
hash: String::new(),
};
entry.hash = entry.compute_hash(account);
let day = day_of(at_unix_ms).max(doc.head.days.last().copied().unwrap_or(i64::MIN));
let key = day_key(day);
let mut bucket = doc.days.get(&key)?.unwrap_or_default();
bucket.push(entry.clone());
doc.days.insert(key, bucket)?;
if doc.head.days.last() != Some(&day) {
doc.head.days.push(day);
}
doc.head.next_seq += 1;
doc.head.tip = entry.hash.clone();
Ok(entry)
})
}
fn drop_before(doc: &mut AccountAuditDoc, cutoff_day: i64) -> Result<u64> {
let mut dropped_count = 0;
while let Some(&oldest) = doc.head.days.first() {
if oldest >= cutoff_day {
break;
}
if let Some(dropped) = doc.days.remove(&day_key(oldest))? {
dropped_count += dropped.len() as u64;
if let Some(last) = dropped.last() {
doc.head.anchor_seq = Some(last.seq);
doc.head.anchor_hash = last.hash.clone();
if doc.head.tip == last.hash {
doc.head.tip.clear();
}
}
}
doc.head.days.remove(0);
}
Ok(dropped_count)
}
pub fn prune(&self, account: &str, now_unix_ms: i64) -> Result<u64> {
let doc = self.doc(account)?;
let cutoff_day = day_of(now_unix_ms) - i64::from(self.retention_days());
if doc
.load()?
.head
.days
.first()
.is_none_or(|&oldest| oldest >= cutoff_day)
{
return Ok(0);
}
doc.update(|doc| Self::drop_before(doc, cutoff_day))
}
pub fn query(&self, account: &str, query: &AuditQuery) -> Result<AuditPage> {
let limit = match query.limit {
0 => 50,
n => n.min(MAX_PAGE),
};
let mut page = AuditPage::default();
let mut more = false;
self.walk_newest_first(account, query, &mut |entry| {
if page.entries.len() == limit {
more = true;
return false;
}
page.entries.push(entry);
true
})?;
if more {
page.next_before_seq = page.entries.last().map(|e| e.seq);
}
Ok(page)
}
pub fn all(&self, account: &str, query: &AuditQuery) -> Result<Vec<AccountEntry>> {
let mut out = Vec::new();
self.walk_newest_first(account, query, &mut |entry| {
out.push(entry);
true
})?;
Ok(out)
}
fn walk_newest_first(
&self,
account: &str,
query: &AuditQuery,
take: &mut dyn FnMut(AccountEntry) -> bool,
) -> Result<()> {
let doc = self.doc(account)?.load()?;
let since = query.since_unix_ms;
let text = query.text.to_lowercase();
for &day in doc.head.days.iter().rev() {
if query.until_unix_ms > 0 && day * DAY_MS > query.until_unix_ms {
continue;
}
if (day + 1) * DAY_MS <= since {
break;
}
let Some(bucket) = doc.days.get(&day_key(day))? else {
continue;
};
for entry in bucket.into_iter().rev() {
if query.before_seq.is_some_and(|before| entry.seq >= before) {
continue;
}
if !query.matches(&entry, &text) {
continue;
}
if !take(entry) {
return Ok(());
}
}
}
Ok(())
}
pub fn verify(&self, account: &str) -> Result<ChainReport> {
let doc = self.doc(account)?.load()?;
let mut entries = Vec::new();
for &day in &doc.head.days {
if let Some(bucket) = doc.days.get(&day_key(day))? {
entries.extend(bucket);
}
}
let anchor = doc
.head
.anchor_seq
.map(|seq| (seq, doc.head.anchor_hash.as_str()));
let mut report = verify_entries(account, anchor, &entries);
if report.holds && report.tip_hash != doc.head.prev() {
report.holds = false;
report.what =
"the head names a tip the entries do not end at: entries were cut off".to_owned();
report.broken_at_seq = Some(doc.head.next_seq.saturating_sub(1));
}
Ok(report)
}
pub fn head(&self, account: &str) -> Result<(u64, String)> {
let doc = self.doc(account)?.load()?;
Ok((doc.head.next_seq, doc.head.prev()))
}
}
fn csv_field(out: &mut String, text: &str) {
let defused;
let text = if text.starts_with(['=', '+', '-', '@', '\t', '\r']) {
defused = format!("'{text}");
defused.as_str()
} else {
text
};
if text.contains([',', '"', '\n', '\r']) {
out.push('"');
out.push_str(&text.replace('"', "\"\""));
out.push('"');
} else {
out.push_str(text);
}
}
pub fn to_csv(entries: &[AccountEntry]) -> String {
let mut out = String::from("seq,at_utc,kind,actor,actor_role,target,detail,outcome,hash\n");
for e in entries {
let fields = [
e.seq.to_string(),
iso8601_utc(e.at_unix_ms),
e.kind.clone(),
e.actor.clone(),
e.actor_role.clone(),
e.target.clone(),
e.detail.clone(),
e.outcome.clone(),
e.hash.clone(),
];
for (i, field) in fields.iter().enumerate() {
if i > 0 {
out.push(',');
}
csv_field(&mut out, field);
}
out.push('\n');
}
out
}
pub fn to_json(entries: &[AccountEntry]) -> String {
let rows: Vec<serde_json::Value> = entries
.iter()
.map(|e| {
let mut row = serde_json::to_value(e).unwrap_or(serde_json::Value::Null);
if let Some(map) = row.as_object_mut() {
map.insert(
"at_utc".into(),
serde_json::Value::String(iso8601_utc(e.at_unix_ms)),
);
}
row
})
.collect();
serde_json::to_string_pretty(&rows).unwrap_or_else(|_| "[]".to_owned())
}