use std::{
collections::{HashMap, HashSet},
sync::{Arc, Mutex},
};
use async_trait::async_trait;
use super::{
cascade_invalidator::CascadeInvalidator,
collections::CollectionTagging,
fact_table_version::{FactTableCacheConfig, FactTableVersionProvider},
feed::{CacheEvictor, Eviction, FeedOutbox},
result::QueryResultCache,
tviews_feed::TviewsListener,
};
use crate::{
backend::{
BatchRowGate, DatabaseAdapter, DatabaseType, MutationRowGate, PoolMetrics, WhereClause,
WriteRequest, Writer, quote_postgres_identifier,
types::{JsonbValue, OrderByClause, ReadRouting},
},
cache::config::RlsEnforcement,
error::{FraiseQLError, Result},
schema::{CompiledSchema, SourceKind, SourceProbe, sql_source_probes},
};
#[cfg(test)]
mod composed_tests;
mod mutation;
mod query;
#[cfg(test)]
mod tests;
pub use query::view_name_to_entity_type;
#[derive(Debug, Clone, Copy)]
struct RelationRls {
is_view: bool,
rls_enabled: bool,
has_policy: bool,
security_invoker: bool,
}
impl RelationRls {
const fn is_protected(self) -> bool {
if self.is_view {
self.security_invoker
} else {
self.rls_enabled && self.has_policy
}
}
const fn explain(self) -> &'static str {
if self.is_view {
"view is not `security_invoker`, so it runs with its owner's privileges and \
bypasses the caller's RLS policies"
} else if !self.rls_enabled {
"row level security is not enabled on this table"
} else {
"row level security is enabled but no policy is defined, so it isolates nothing"
}
}
}
pub struct CachedDatabaseAdapter<A: DatabaseAdapter> {
pub(super) adapter: A,
pub(super) cache: Arc<QueryResultCache>,
pub(super) schema_version: String,
pub(super) view_ttl_overrides: HashMap<String, u64>,
pub(super) cacheable_views: HashSet<String>,
pub(super) view_secondary_views: HashMap<String, Vec<String>>,
pub(super) opt_in_mode: bool,
pub(super) has_rls: bool,
pub(super) fact_table_config: FactTableCacheConfig,
pub(super) version_provider: Arc<FactTableVersionProvider>,
pub(super) cascade_invalidator: Option<Arc<Mutex<CascadeInvalidator>>>,
pub(super) feed: Option<FeedOutbox>,
pub(super) collections: CollectionTagging,
pub(super) by_id_sources: Arc<HashMap<String, String>>,
}
impl<A: DatabaseAdapter> CachedDatabaseAdapter<A> {
#[must_use]
pub fn new(adapter: A, cache: QueryResultCache, schema_version: String) -> Self {
Self {
adapter,
cache: Arc::new(cache),
schema_version,
view_ttl_overrides: HashMap::new(),
cacheable_views: HashSet::new(),
view_secondary_views: HashMap::new(),
opt_in_mode: false,
has_rls: false,
fact_table_config: FactTableCacheConfig::default(),
version_provider: Arc::new(FactTableVersionProvider::default()),
cascade_invalidator: None,
feed: None,
collections: CollectionTagging::default(),
by_id_sources: Arc::default(),
}
}
#[must_use]
pub fn with_view_ttl_overrides(mut self, overrides: HashMap<String, u64>) -> Self {
self.cacheable_views = overrides.keys().cloned().collect();
self.view_ttl_overrides = overrides;
self.opt_in_mode = true;
self
}
#[must_use]
pub const fn with_rls(mut self, has_rls: bool) -> Self {
self.has_rls = has_rls;
self
}
#[must_use]
pub fn with_cascade_invalidator(mut self, invalidator: CascadeInvalidator) -> Self {
self.cascade_invalidator = Some(Arc::new(Mutex::new(invalidator)));
self
}
#[must_use]
pub fn with_by_id_sources(mut self, sources: HashMap<String, String>) -> Self {
self.by_id_sources = Arc::new(sources);
self
}
#[must_use]
pub fn with_feed(mut self, outbox: FeedOutbox) -> Self {
self.feed = Some(outbox);
self
}
#[must_use]
pub fn evictor(&self) -> CacheEvictor {
CacheEvictor {
cache: Arc::clone(&self.cache),
cascade: self.cascade_invalidator.clone(),
}
}
#[must_use]
pub fn collection_tagging(&self) -> CollectionTagging {
self.collections.clone()
}
#[must_use]
pub fn tviews_listener(&self, catalog: super::tviews_feed::FeedCatalog) -> TviewsListener {
TviewsListener::new(Arc::clone(&self.cache), self.collections.clone(), catalog)
}
fn announce(&self, eviction: impl FnOnce() -> Eviction) {
if let Some(feed) = &self.feed {
if self.cache.is_enabled() {
feed.announce(eviction());
}
}
}
#[must_use]
pub fn with_cache_metadata_from_schema(mut self, schema: &CompiledSchema) -> Self {
self.read_schema_cache_metadata(schema);
self.opt_in_mode = true;
self
}
fn read_schema_cache_metadata(&mut self, schema: &CompiledSchema) {
for query in &schema.queries {
let Some(view) = &query.sql_source else {
continue;
};
if let Some(ttl) = query.cache_ttl_seconds {
self.cacheable_views.insert(view.clone());
self.view_ttl_overrides.insert(view.clone(), ttl);
}
let secondary: Vec<String> = crate::cache::key::extract_accessed_views(query)
.into_iter()
.filter(|v| v != view)
.collect();
if !secondary.is_empty() {
let entry = self.view_secondary_views.entry(view.clone()).or_default();
for v in secondary {
if !entry.contains(&v) {
entry.push(v);
}
}
}
}
}
pub(super) fn accessed_views_for(&self, view: &str) -> Vec<String> {
let mut views = Vec::with_capacity(1);
views.push(view.to_string());
if let Some(secondary) = self.view_secondary_views.get(view) {
views.extend(secondary.iter().cloned());
}
views
}
#[must_use]
pub fn rebuilt_for_schema(self, schema: &CompiledSchema) -> Self {
let _ = self.cache.clear();
let mut rebuilt = Self {
adapter: self.adapter,
cache: self.cache,
schema_version: schema.content_hash(),
view_ttl_overrides: HashMap::new(),
cacheable_views: HashSet::new(),
view_secondary_views: HashMap::new(),
opt_in_mode: true,
has_rls: self.has_rls,
fact_table_config: self.fact_table_config,
version_provider: self.version_provider,
cascade_invalidator: self.cascade_invalidator,
feed: self.feed,
collections: self.collections,
by_id_sources: self.by_id_sources,
};
rebuilt.read_schema_cache_metadata(schema);
rebuilt
}
#[must_use]
pub fn with_fact_table_config(
adapter: A,
cache: QueryResultCache,
schema_version: String,
fact_table_config: FactTableCacheConfig,
) -> Self {
Self {
adapter,
cache: Arc::new(cache),
schema_version,
view_ttl_overrides: HashMap::new(),
cacheable_views: HashSet::new(),
view_secondary_views: HashMap::new(),
opt_in_mode: false,
has_rls: false,
fact_table_config,
version_provider: Arc::new(FactTableVersionProvider::default()),
cascade_invalidator: None,
feed: None,
collections: CollectionTagging::default(),
by_id_sources: Arc::default(),
}
}
#[must_use]
pub const fn inner(&self) -> &A {
&self.adapter
}
#[must_use]
pub fn cache(&self) -> &QueryResultCache {
&self.cache
}
#[must_use]
pub fn schema_version(&self) -> &str {
&self.schema_version
}
#[must_use]
pub const fn fact_table_config(&self) -> &FactTableCacheConfig {
&self.fact_table_config
}
#[must_use]
pub fn version_provider(&self) -> &FactTableVersionProvider {
&self.version_provider
}
pub async fn validate_rls_active(&self, schema: &CompiledSchema) -> Result<()> {
let mut unprotected: Vec<String> = Vec::new();
for probe in sql_source_probes(schema) {
if probe.kind != SourceKind::Relation {
continue;
}
let display = probe.display_name();
match self.relation_rls_status(&probe).await? {
None => unprotected.push(format!("{display} — relation does not exist")),
Some(status) if status.is_protected() => {},
Some(status) => unprotected.push(format!("{display} — {}", status.explain())),
}
}
if unprotected.is_empty() {
return Ok(());
}
Err(FraiseQLError::Configuration {
message: format!(
"Row-Level Security is declared for this multi-tenant schema but is not in \
force on {} of its source relation(s), so tenants are not isolated and \
cached responses can cross tenant boundaries:\n - {}\n\
Enable RLS and declare a policy on each source table \
(`ALTER TABLE … ENABLE ROW LEVEL SECURITY` + `CREATE POLICY …`), define \
views `WITH (security_invoker = true)` so they honour the caller's \
policies, or remove `[security.rls] enabled = true` if this deployment \
does not rely on database RLS.",
unprotected.len(),
unprotected.join("\n - ")
),
})
}
async fn relation_rls_status(&self, probe: &SourceProbe) -> Result<Option<RelationRls>> {
let ident = match &probe.schema {
Some(s) => format!(
"{}.{}",
quote_postgres_identifier(s),
quote_postgres_identifier(&probe.name)
),
None => quote_postgres_identifier(&probe.name),
};
let literal = ident.replace('\'', "''");
let sql = format!(
"SELECT c.relkind::text AS kind, \
c.relrowsecurity AS rls_enabled, \
(SELECT count(*) FROM pg_policy p WHERE p.polrelid = c.oid)::text \
AS policy_count, \
COALESCE((SELECT bool_or(lower(o.option_value) IN ('true','on','1')) \
FROM pg_options_to_table(c.reloptions) o \
WHERE o.option_name = 'security_invoker'), false) \
AS security_invoker \
FROM pg_class c WHERE c.oid = to_regclass('{literal}')"
);
let rows = self.adapter.execute_raw_query(&sql).await?;
let Some(row) = rows.first() else {
return Ok(None);
};
let cell_bool = |k: &str| row.get(k).and_then(serde_json::Value::as_bool).unwrap_or(false);
Ok(Some(RelationRls {
is_view: row.get("kind").and_then(serde_json::Value::as_str) == Some("v"),
rls_enabled: cell_bool("rls_enabled"),
has_policy: row
.get("policy_count")
.and_then(serde_json::Value::as_str)
.is_some_and(|n| n != "0"),
security_invoker: cell_bool("security_invoker"),
}))
}
pub async fn enforce_rls(
&self,
schema: &CompiledSchema,
enforcement: RlsEnforcement,
) -> Result<()> {
if enforcement == RlsEnforcement::Off {
return Ok(());
}
match self.validate_rls_active(schema).await {
Ok(()) => Ok(()),
Err(e) => match enforcement {
RlsEnforcement::Error => Err(e),
RlsEnforcement::Warn => {
tracing::warn!(
"RLS check failed (rls_enforcement = \"warn\"): {}. \
Cross-tenant cache leakage is possible.",
e
);
Ok(())
},
RlsEnforcement::Off => Ok(()), },
}
}
}
impl<A: DatabaseAdapter + Clone> Clone for CachedDatabaseAdapter<A> {
fn clone(&self) -> Self {
Self {
adapter: self.adapter.clone(),
cache: Arc::clone(&self.cache),
schema_version: self.schema_version.clone(),
view_ttl_overrides: self.view_ttl_overrides.clone(),
cacheable_views: self.cacheable_views.clone(),
view_secondary_views: self.view_secondary_views.clone(),
opt_in_mode: self.opt_in_mode,
has_rls: self.has_rls,
fact_table_config: self.fact_table_config.clone(),
version_provider: Arc::clone(&self.version_provider),
cascade_invalidator: self.cascade_invalidator.clone(),
feed: self.feed.clone(),
collections: self.collections.clone(),
by_id_sources: Arc::clone(&self.by_id_sources),
}
}
}
#[async_trait]
impl<A: DatabaseAdapter> DatabaseAdapter for CachedDatabaseAdapter<A> {
async fn execute_with_projection(
&self,
view: &str,
projection: Option<&crate::schema::SqlProjectionHint>,
where_clause: Option<&WhereClause>,
limit: Option<u32>,
offset: Option<u32>,
order_by: Option<&[OrderByClause]>,
) -> Result<Vec<JsonbValue>> {
let request = crate::backend::ProjectionRequest {
view,
projection,
where_clause,
order_by,
limit,
offset,
matched_up_to: None,
};
self.execute_with_projection_impl(&request, &[]).await.map(Arc::unwrap_or_clone)
}
async fn execute_where_query(
&self,
view: &str,
where_clause: Option<&WhereClause>,
limit: Option<u32>,
offset: Option<u32>,
order_by: Option<&[OrderByClause]>,
) -> Result<Vec<JsonbValue>> {
self.execute_where_query_impl(view, where_clause, limit, offset, order_by, &[])
.await
.map(Arc::unwrap_or_clone)
}
async fn execute_with_projection_arc(
&self,
request: &crate::backend::ProjectionRequest<'_>,
) -> Result<Arc<Vec<JsonbValue>>> {
self.execute_with_projection_impl(request, &[]).await
}
async fn execute_where_query_arc(
&self,
view: &str,
where_clause: Option<&WhereClause>,
limit: Option<u32>,
offset: Option<u32>,
order_by: Option<&[OrderByClause]>,
) -> Result<Arc<Vec<JsonbValue>>> {
self.execute_where_query_impl(view, where_clause, limit, offset, order_by, &[])
.await
}
fn database_type(&self) -> DatabaseType {
self.adapter.database_type()
}
fn applies_session_variables(&self) -> bool {
self.adapter.applies_session_variables()
}
fn serves_reads_from_standbys(&self) -> bool {
self.adapter.serves_reads_from_standbys()
}
async fn health_check(&self) -> Result<()> {
self.adapter.health_check().await
}
fn pool_metrics(&self) -> PoolMetrics {
self.adapter.pool_metrics()
}
async fn execute_raw_query(
&self,
sql: &str,
) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
self.execute_aggregation_query(sql).await
}
async fn execute_parameterized_aggregate(
&self,
sql: &str,
params: &[serde_json::Value],
) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
self.adapter.execute_parameterized_aggregate(sql, params).await
}
async fn execute_parameterized_aggregate_with_session(
&self,
sql: &str,
params: &[serde_json::Value],
session_vars: &[(&str, &str)],
routing: ReadRouting,
) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
self.adapter
.execute_parameterized_aggregate_with_session(sql, params, session_vars, routing)
.await
}
fn supports_composed_reads(&self) -> bool {
self.adapter.supports_composed_reads()
}
async fn describe_constraint(
&self,
violation: &fraiseql_error::ConstraintViolation,
) -> Result<fraiseql_db::ConstraintDescription> {
self.adapter.describe_constraint(violation).await
}
async fn tview_profiles(&self, sources: &[String]) -> Result<Vec<fraiseql_db::TviewProfile>> {
self.adapter.tview_profiles(sources).await
}
async fn schema_storage_bytes(&self, schema: &str) -> Result<Option<u64>> {
self.adapter.schema_storage_bytes(schema).await
}
async fn count_where_query(
&self,
view: &str,
where_clause: Option<&WhereClause>,
session_vars: &[(&str, &str)],
routing: ReadRouting,
) -> Result<u64> {
self.adapter.count_where_query(view, where_clause, session_vars, routing).await
}
async fn execute_where_query_arc_with_session(
&self,
view: &str,
where_clause: Option<&WhereClause>,
limit: Option<u32>,
offset: Option<u32>,
order_by: Option<&[OrderByClause]>,
session_vars: &[(&str, &str)],
routing: ReadRouting,
) -> Result<Arc<Vec<JsonbValue>>> {
if routing.allows_cached_result() {
return self
.execute_where_query_impl(view, where_clause, limit, offset, order_by, session_vars)
.await;
}
self.adapter
.execute_where_query_arc_with_session(
view,
where_clause,
limit,
offset,
order_by,
session_vars,
routing,
)
.await
}
async fn execute_with_projection_arc_with_session(
&self,
request: &crate::backend::ProjectionRequest<'_>,
session_vars: &[(&str, &str)],
routing: ReadRouting,
) -> Result<Arc<Vec<JsonbValue>>> {
if routing.allows_cached_result() {
return self.execute_with_projection_impl(request, session_vars).await;
}
self.adapter
.execute_with_projection_arc_with_session(request, session_vars, routing)
.await
}
async fn stream_with_projection(
&self,
request: &crate::backend::ProjectionRequest<'_>,
session_vars: &[(&str, &str)],
routing: ReadRouting,
) -> Result<fraiseql_db::JsonbRowStream> {
self.adapter.stream_with_projection(request, session_vars, routing).await
}
async fn execute_composed_with_session(
&self,
read: &fraiseql_db::ComposedLevel,
session_vars: &[(&str, &str)],
routing: ReadRouting,
) -> Result<Arc<Vec<JsonbValue>>> {
if routing.allows_cached_result() {
return self.execute_composed_impl(read, session_vars, routing).await;
}
self.adapter.execute_composed_with_session(read, session_vars, routing).await
}
async fn stream_row_query(
&self,
view_name: &str,
columns: &[fraiseql_db::types::ColumnSpec],
where_sql: Option<&str>,
order_by: Option<&str>,
limit: Option<u32>,
offset: Option<u32>,
) -> Result<fraiseql_db::ColumnRowStream> {
self.adapter
.stream_row_query(view_name, columns, where_sql, order_by, limit, offset)
.await
}
async fn execute_row_query_with_session(
&self,
view_name: &str,
columns: &[fraiseql_db::types::ColumnSpec],
where_sql: Option<&str>,
order_by: Option<&str>,
limit: Option<u32>,
offset: Option<u32>,
session_vars: &[(&str, &str)],
) -> Result<Vec<Vec<fraiseql_db::types::ColumnValue>>> {
self.adapter
.execute_row_query_with_session(
view_name,
columns,
where_sql,
order_by,
limit,
offset,
session_vars,
)
.await
}
async fn stream_row_query_with_session(
&self,
view_name: &str,
columns: &[fraiseql_db::types::ColumnSpec],
where_sql: Option<&str>,
order_by: Option<&str>,
limit: Option<u32>,
offset: Option<u32>,
session_vars: &[(&str, &str)],
) -> Result<fraiseql_db::ColumnRowStream> {
self.adapter
.stream_row_query_with_session(
view_name,
columns,
where_sql,
order_by,
limit,
offset,
session_vars,
)
.await
}
async fn execute_admin_sql(
&self,
request: &fraiseql_db::AdminSqlRequest,
) -> Result<fraiseql_db::AdminSqlOutcome> {
self.adapter.execute_admin_sql(request).await
}
async fn invalidate_views(&self, views: &[fraiseql_db::ViewName]) -> Result<u64> {
let evicted = CachedDatabaseAdapter::invalidate_views(self, views)?;
self.announce(|| Eviction::Views(views.iter().map(|v| v.as_str().to_owned()).collect()));
Ok(evicted)
}
fn reports_changed_rows(&self) -> bool {
self.cache.is_enabled()
&& (!self.by_id_sources.is_empty() || self.collections.current().is_some())
}
async fn invalidate_written(
&self,
views: &[fraiseql_db::ViewName],
changed: &fraiseql_db::ChangedRows,
) -> Result<u64> {
let evicted = super::invalidation_api::invalidate_written(
&self.cache,
self.cascade_invalidator.as_ref(),
views,
changed,
)?;
self.announce(|| Eviction::Views(views.iter().map(|v| v.as_str().to_owned()).collect()));
Ok(evicted)
}
async fn invalidate_by_entity(&self, entity_type: &str, entity_id: &str) -> Result<u64> {
let evicted = CachedDatabaseAdapter::invalidate_by_entity(self, entity_type, entity_id)?;
self.announce(|| Eviction::Entity {
entity_type: entity_type.to_owned(),
entity_id: entity_id.to_owned(),
});
Ok(evicted)
}
fn result_cache_stats(&self) -> Option<fraiseql_db::ResultCacheStats> {
if !self.cache.is_enabled() {
return None;
}
self.cache.run_pending_tasks();
let metrics = self.cache.metrics().ok()?;
let config = self.cache.config();
Some(fraiseql_db::ResultCacheStats {
entries: metrics.size,
hits: metrics.hits,
misses: metrics.misses,
invalidations: metrics.invalidations,
memory_bytes: metrics.memory_bytes,
ttl_seconds: config.ttl_seconds,
max_entries: config.max_entries,
})
}
async fn clear_result_cache(&self) -> Result<Option<usize>> {
if !self.cache.is_enabled() {
return Ok(None);
}
self.cache.run_pending_tasks();
let before = self.cache.metrics().map(|m| m.size).unwrap_or_default();
self.cache.clear()?;
self.announce(|| Eviction::Everything);
Ok(Some(before))
}
async fn query_stats(&self, limit: u32) -> Result<Vec<fraiseql_db::QueryStatEntry>> {
self.adapter.query_stats(limit).await
}
async fn query_stats_by_id(&self, id: &str) -> Result<Option<fraiseql_db::QueryStatEntry>> {
self.adapter.query_stats_by_id(id).await
}
async fn reset_query_stats(&self) -> Result<()> {
self.adapter.reset_query_stats().await
}
fn on_schema_reload(&self) {
let _ = self.cache.clear();
}
}
#[async_trait]
impl<A: Writer> Writer for CachedDatabaseAdapter<A> {
async fn execute_write(
&self,
request: &WriteRequest<'_>,
gate: MutationRowGate<'_>,
) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
self.adapter.execute_write(request, gate).await
}
async fn execute_writes(
&self,
requests: &[WriteRequest<'_>],
gate: BatchRowGate<'_>,
) -> Result<Vec<Vec<std::collections::HashMap<String, serde_json::Value>>>> {
self.adapter.execute_writes(requests, gate).await
}
async fn bump_fact_table_versions(&self, tables: &[String]) -> Result<()> {
self.bump_fact_table_versions_impl(tables).await
}
}