use crate::domain::{key::TrustRecordKey, *};
use crate::storage::repository::*;
use anyhow::anyhow;
use base64::Engine as _;
use base64::engine::general_purpose::STANDARD as base64;
use serde_json::Value;
use std::{
collections::HashMap,
path::{Path, PathBuf},
str::FromStr,
sync::Arc,
time::{Duration, SystemTime},
};
use tokio_util::sync::CancellationToken;
use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use tokio::time::sleep;
use tracing::{error, info};
#[derive(Clone)]
pub struct FileStorage {
file_path: PathBuf,
update_interval: Duration,
records: Arc<RwLock<HashMap<TrustRecordKey, TrustRecord>>>,
last_modified: Arc<RwLock<Option<SystemTime>>>,
shutdown: CancellationToken,
}
impl FileStorage {
pub async fn try_new<P: Into<PathBuf>>(
file_path: P,
update_interval_sec: u64,
) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
let file_path = file_path.into();
let update_interval = Duration::from_secs(update_interval_sec);
let records = Arc::new(RwLock::new(HashMap::new()));
let last_modified = Arc::new(RwLock::new(None));
let (initial_records, modified) = Self::load_if_modified(&file_path, None)
.await?
.ok_or_else(|| {
anyhow!("unable to load trust records from {}", file_path.display())
.into_boxed_dyn_error()
})?;
{
let mut guard = records.write().await;
*guard = initial_records;
}
{
let mut guard = last_modified.write().await;
*guard = Some(modified);
}
let storage = Self {
file_path: file_path.clone(),
update_interval,
records: Arc::clone(&records),
last_modified: Arc::clone(&last_modified),
shutdown: CancellationToken::new(),
};
storage.spawn_sync_task();
Ok(storage)
}
fn spawn_sync_task(&self) {
let file_path = self.file_path.clone();
let update_interval = self.update_interval;
let records = Arc::clone(&self.records);
let last_modified = Arc::clone(&self.last_modified);
let shutdown = self.shutdown.clone();
tokio::spawn(async move {
loop {
tokio::select! {
_ = shutdown.cancelled() => {
info!(path = %file_path.display(), "CSV sync task shutting down");
break;
}
_ = sleep(update_interval) => {
info!(path = %file_path.display(), "Syncing trust records from file");
let previous = { *last_modified.read().await };
match Self::load_if_modified(&file_path, previous).await {
Ok(Some((new_records, modified))) => {
{
let mut guard = records.write().await;
*guard = new_records;
}
{
let mut guard = last_modified.write().await;
*guard = Some(modified);
}
}
Ok(None) => {}
Err(err) => {
error!(
error = %err,
path = %file_path.display(),
"Failed to sync trust records from file"
);
}
}
}
}
}
});
}
async fn load_if_modified(
path: &Path,
last_seen: Option<SystemTime>,
) -> Result<
Option<(HashMap<TrustRecordKey, TrustRecord>, SystemTime)>,
Box<dyn std::error::Error + Send + Sync>,
> {
let metadata = tokio::fs::metadata(path)
.await
.map_err(|e| format!("file_path: {path:?}, error: {e}"))?;
let modified = metadata.modified()?;
if let Some(previous) = last_seen
&& modified <= previous
{
info!(
path = %path.display(),
"No changes detected in trust records file"
);
return Ok(None);
}
info!(
path = %path.display(),
"Changes detected in trust records file, reloading"
);
let contents = tokio::fs::read_to_string(path).await?.trim().to_string();
let records = Self::parse_csv(&contents)?;
Ok(Some((records, modified)))
}
fn parse_csv(
contents: &str,
) -> Result<HashMap<TrustRecordKey, TrustRecord>, Box<dyn std::error::Error + Send + Sync>>
{
let mut reader = csv::ReaderBuilder::new()
.has_headers(true)
.trim(csv::Trim::All)
.from_reader(contents.as_bytes());
let mut records = HashMap::new();
for result in reader.deserialize::<TrustRecordCsvRow>() {
let row = result?;
let record = row.into_record()?;
let key = TrustRecordKey::from_record(&record);
records.insert(key, record);
}
Ok(records)
}
fn matches_query(record: &TrustRecord, query: &TrustRecordQuery) -> bool {
record.entity_id() == &query.entity_id
&& record.authority_id() == &query.authority_id
&& record.action() == &query.action
&& record.resource() == &query.resource
}
async fn write_to_file(&self) -> Result<(), RepositoryError> {
let records_clone = {
let records = self.records.read().await;
records.values().cloned().collect::<Vec<_>>()
};
let mut csv_records = Vec::new();
for record in records_clone.iter() {
csv_records.push(TrustRecordCsvRow::from_record(record));
}
let mut wtr = csv::Writer::from_writer(vec![]);
for row in csv_records {
wtr.serialize(&row)
.map_err(|e| RepositoryError::SerializationFailed(e.to_string()))?;
}
let csv_data = wtr
.into_inner()
.map_err(|e| RepositoryError::SerializationFailed(e.to_string()))?;
tokio::fs::write(&self.file_path, csv_data)
.await
.map_err(|e| RepositoryError::QueryFailed(format!("Failed to write CSV file: {e}")))?;
let metadata = tokio::fs::metadata(&self.file_path).await.map_err(|e| {
RepositoryError::QueryFailed(format!("Failed to get file metadata: {e}"))
})?;
let modified = metadata.modified().map_err(|e| {
RepositoryError::QueryFailed(format!("Failed to get modified time: {e}"))
})?;
let mut guard = self.last_modified.write().await;
*guard = Some(modified);
Ok(())
}
}
#[async_trait::async_trait]
impl TrustRecordRepository for FileStorage {
async fn find_by_query(
&self,
query: TrustRecordQuery,
) -> Result<Option<TrustRecord>, RepositoryError> {
let records = Arc::clone(&self.records);
let guard = records.read().await;
let result = guard
.values()
.find(|&record| FileStorage::matches_query(record, &query))
.cloned();
Ok(result)
}
}
#[async_trait::async_trait]
impl TrustRecordAdminRepository for FileStorage {
async fn create(&self, record: TrustRecord) -> Result<(), RepositoryError> {
let key = TrustRecordKey::from_record(&record);
{
let mut records = self.records.write().await;
if records.contains_key(&key) {
return Err(RepositoryError::RecordAlreadyExists(format!(
"Record already exists: {record}"
)));
}
records.insert(key, record);
}
self.write_to_file().await
}
async fn update(&self, record: TrustRecord) -> Result<(), RepositoryError> {
let key = TrustRecordKey::from_record(&record);
{
let mut records = self.records.write().await;
if !records.contains_key(&key) {
return Err(RepositoryError::RecordNotFound(format!(
"Record not found: {record}"
)));
}
records.insert(key, record);
}
self.write_to_file().await
}
async fn delete(&self, query: TrustRecordQuery) -> Result<(), RepositoryError> {
let key = TrustRecordKey::from_query(&query);
{
let mut records = self.records.write().await;
if records.remove(&key).is_none() {
return Err(RepositoryError::RecordNotFound(format!(
"Record not found: {key}",
)));
}
}
self.write_to_file().await
}
async fn list(&self) -> Result<TrustRecordList, RepositoryError> {
let records = self.records.read().await;
let records_vec: Vec<TrustRecord> = records.values().cloned().collect();
Ok(TrustRecordList::new(records_vec))
}
async fn read(&self, query: TrustRecordQuery) -> Result<TrustRecord, RepositoryError> {
let records = self.records.read().await;
let result = records
.values()
.find(|&record| FileStorage::matches_query(record, &query))
.cloned();
result.ok_or_else(|| {
RepositoryError::RecordNotFound(format!(
"Record not found: {}|{}|{}|{}",
query.entity_id, query.authority_id, query.action, query.resource
))
})
}
}
impl Drop for FileStorage {
fn drop(&mut self) {
self.shutdown.cancel();
}
}
#[derive(Debug, Deserialize, Serialize)]
struct TrustRecordCsvRow {
entity_id: String,
authority_id: String,
action: String,
resource: String,
recognized: bool,
authorized: bool,
context: Option<String>,
record_type: String,
}
impl TrustRecordCsvRow {
fn parse_context(ctx: Option<String>) -> Option<Value> {
let record_context: Option<Value> = if let Some(c) = ctx {
base64
.decode(&c)
.ok()
.and_then(|db| String::from_utf8(db).ok())
.and_then(|s| serde_json::from_str(&s).ok())
} else {
None
};
record_context
}
fn from_record(record: &TrustRecord) -> Self {
let context = if record.context().as_value().is_object()
|| record.context().as_value().is_array()
{
let json_str = serde_json::to_string(record.context().as_value()).unwrap_or_default();
let encoded = base64.encode(json_str.as_bytes());
Some(encoded)
} else {
None
};
Self {
entity_id: record.entity_id().to_string(),
authority_id: record.authority_id().to_string(),
action: record.action().to_string(),
resource: record.resource().to_string(),
recognized: record.is_recognized(),
authorized: record.is_authorized(),
context,
record_type: record.record_type().to_string(),
}
}
fn into_record(self) -> Result<TrustRecord, Box<dyn std::error::Error + Send + Sync>> {
let ctx = TrustRecordCsvRow::parse_context(self.context);
let mut builder = TrustRecordBuilder::new()
.entity_id(EntityId::new(self.entity_id))
.authority_id(AuthorityId::new(self.authority_id))
.action(Action::new(self.action))
.resource(Resource::new(self.resource))
.recognized(self.recognized)
.authorized(self.authorized)
.record_type(RecordType::from_str(&self.record_type)?);
if let Some(c) = ctx {
builder = builder.context(Context::new(c));
}
builder
.build()
.map_err(|err| anyhow!("invalid trust record: {err}").into())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Write;
use tempfile::NamedTempFile;
use tokio::time::{Duration, sleep};
fn csv_header() -> String {
String::from(
"entity_id,authority_id,action,resource,recognized,authorized,context,record_type\n",
)
}
fn sample_csv(records: &[(&str, &str, &str, &str, &str)]) -> String {
let mut csv = String::new();
for (entity, authority, action, resource, record_type) in records {
csv.push_str(&format!(
"{entity},{authority},{action},{resource},true,true,e30=,{record_type}\n"
));
}
csv
}
#[tokio::test]
async fn fails_when_initial_load_fails() {
let result = FileStorage::try_new("/does/not/exist.csv", 1).await;
assert!(result.is_err());
}
#[tokio::test]
async fn finds_records_from_initial_load() {
let mut file = NamedTempFile::new().unwrap();
write!(file, "{}", csv_header()).unwrap();
write!(
file,
"{}",
sample_csv(&[("e1", "a1", "ac1", "r1", "authorization")])
)
.unwrap();
let storage = FileStorage::try_new(file.path(), 1).await.unwrap();
let query = TrustRecordQuery::new(
EntityId::new("e1"),
AuthorityId::new("a1"),
Action::new("ac1"),
Resource::new("r1"),
);
let result = storage.find_by_query(query).await.unwrap();
assert!(result.is_some());
let record = result.unwrap();
assert_eq!(*record.record_type(), RecordType::Authorization);
}
#[tokio::test]
async fn reloads_when_file_changes() {
let mut file = NamedTempFile::new().unwrap();
write!(file, "{}", csv_header()).unwrap();
write!(
file,
"{}",
sample_csv(&[("e1", "a1", "ac1", "r1", "recognition")])
)
.unwrap();
file.flush().unwrap();
let storage = FileStorage::try_new(file.path(), 1).await.unwrap();
sleep(Duration::from_secs(1)).await;
write!(
file.as_file_mut(),
"{}",
sample_csv(&[("e2", "a2", "ac2", "r2", "recognition")])
)
.unwrap();
file.flush().unwrap();
sleep(Duration::from_secs(2)).await;
let query = TrustRecordQuery::new(
EntityId::new("e2"),
AuthorityId::new("a2"),
Action::new("ac2"),
Resource::new("r2"),
);
let result = storage.find_by_query(query).await.unwrap();
assert!(result.is_some());
assert_eq!(result.clone().unwrap().entity_id().as_str(), "e2");
assert_eq!(*result.unwrap().record_type(), RecordType::Recognition);
}
}