pub mod aggregations;
pub mod operations;
pub use aggregations::CollectionAggregations;
pub use operations::CollectionOperations;
use crate::proxy::{ProxyError, ProxyResult, ScalarValue};
use serde::{Deserialize, Serialize};
#[derive(Clone)]
pub struct CollectionProxy {
pub relationship: String,
pub attribute: String,
pub unique: bool,
pub deduplicate: bool,
pub loading_strategy: Option<crate::proxy::LoadingStrategy>,
pub factory: Option<std::sync::Arc<dyn super::reflection::ReflectableFactory>>,
caching: bool,
cache_ttl: Option<u64>,
streaming: bool,
triggers: bool,
trigger_events: Vec<String>,
stored_procedure: Option<String>,
procedure_params: Vec<(String, String)>,
database: Option<String>,
fallback_database: Option<String>,
async_loading: bool,
concurrent_access: bool,
is_view: bool,
batch_size: Option<usize>,
chunk_size: Option<usize>,
cascade: bool,
version_tracking: bool,
memory_limit: Option<usize>,
}
impl CollectionProxy {
pub fn new(relationship: &str, attribute: &str) -> Self {
Self {
relationship: relationship.to_string(),
attribute: attribute.to_string(),
unique: false,
deduplicate: false,
loading_strategy: None,
factory: None,
caching: false,
cache_ttl: None,
streaming: false,
triggers: false,
trigger_events: Vec::new(),
stored_procedure: None,
procedure_params: Vec::new(),
database: None,
fallback_database: None,
async_loading: false,
concurrent_access: false,
is_view: false,
batch_size: None,
chunk_size: None,
cascade: false,
version_tracking: false,
memory_limit: None,
}
}
pub fn unique(relationship: &str, attribute: &str) -> Self {
Self {
relationship: relationship.to_string(),
attribute: attribute.to_string(),
unique: true,
deduplicate: false,
loading_strategy: None,
factory: None,
caching: false,
cache_ttl: None,
streaming: false,
triggers: false,
trigger_events: Vec::new(),
stored_procedure: None,
procedure_params: Vec::new(),
database: None,
fallback_database: None,
async_loading: false,
concurrent_access: false,
is_view: false,
batch_size: None,
chunk_size: None,
cascade: false,
version_tracking: false,
memory_limit: None,
}
}
pub fn with_factory(
relationship: &str,
attribute: &str,
factory: std::sync::Arc<dyn super::reflection::ReflectableFactory>,
) -> Self {
Self {
relationship: relationship.to_string(),
attribute: attribute.to_string(),
unique: false,
deduplicate: false,
loading_strategy: None,
factory: Some(factory),
caching: false,
cache_ttl: None,
streaming: false,
triggers: false,
trigger_events: Vec::new(),
stored_procedure: None,
procedure_params: Vec::new(),
database: None,
fallback_database: None,
async_loading: false,
concurrent_access: false,
is_view: false,
batch_size: None,
chunk_size: None,
cascade: false,
version_tracking: false,
memory_limit: None,
}
}
pub fn set_factory(
&mut self,
factory: std::sync::Arc<dyn super::reflection::ReflectableFactory>,
) {
self.factory = Some(factory);
}
pub async fn get_values<T>(&self, source: &T) -> ProxyResult<Vec<ScalarValue>>
where
T: super::reflection::Reflectable,
{
let relationship = source
.get_relationship(&self.relationship)
.ok_or_else(|| ProxyError::RelationshipNotFound(self.relationship.clone()))?;
let collection = super::reflection::downcast_relationship::<
Vec<Box<dyn super::reflection::Reflectable>>,
>(relationship)?;
let mut values = Vec::new();
for item in collection.iter() {
let value = item
.get_attribute(&self.attribute)
.ok_or_else(|| ProxyError::AttributeNotFound(self.attribute.clone()))?;
values.push(value);
}
if self.unique {
values.sort_by(|a, b| format!("{:?}", a).cmp(&format!("{:?}", b)));
values.dedup_by(|a, b| format!("{:?}", a) == format!("{:?}", b));
}
Ok(values)
}
pub async fn set_values<T>(&self, source: &mut T, values: Vec<ScalarValue>) -> ProxyResult<()>
where
T: super::reflection::Reflectable,
{
let factory = self
.factory
.as_ref()
.ok_or(ProxyError::FactoryNotConfigured)?;
let relationship = source
.get_relationship_mut(&self.relationship)
.ok_or_else(|| ProxyError::RelationshipNotFound(self.relationship.clone()))?;
let collection = relationship
.downcast_mut::<Vec<Box<dyn super::reflection::Reflectable>>>()
.ok_or_else(|| ProxyError::TypeMismatch {
expected: "Vec<Box<dyn Reflectable>>".to_string(),
actual: "unknown".to_string(),
})?;
collection.clear();
for value in values {
let new_object = factory.create_from_scalar(&self.attribute, value)?;
collection.push(new_object);
}
Ok(())
}
pub async fn append<T>(&self, source: &mut T, value: ScalarValue) -> ProxyResult<()>
where
T: super::reflection::Reflectable,
{
let factory = self
.factory
.as_ref()
.ok_or(ProxyError::FactoryNotConfigured)?;
let relationship = source
.get_relationship_mut(&self.relationship)
.ok_or_else(|| ProxyError::RelationshipNotFound(self.relationship.clone()))?;
let collection = relationship
.downcast_mut::<Vec<Box<dyn super::reflection::Reflectable>>>()
.ok_or_else(|| ProxyError::TypeMismatch {
expected: "Vec<Box<dyn Reflectable>>".to_string(),
actual: "unknown".to_string(),
})?;
let new_object = factory.create_from_scalar(&self.attribute, value)?;
collection.push(new_object);
Ok(())
}
pub async fn remove<T>(&self, source: &mut T, value: ScalarValue) -> ProxyResult<()>
where
T: super::reflection::Reflectable,
{
let relationship = source
.get_relationship_mut(&self.relationship)
.ok_or_else(|| ProxyError::RelationshipNotFound(self.relationship.clone()))?;
let collection = relationship
.downcast_mut::<Vec<Box<dyn super::reflection::Reflectable>>>()
.ok_or_else(|| ProxyError::TypeMismatch {
expected: "Vec<Box<dyn Reflectable>>".to_string(),
actual: "unknown".to_string(),
})?;
collection.retain(|item| {
item.get_attribute(&self.attribute)
.map(|v| v != value)
.unwrap_or(true)
});
Ok(())
}
pub async fn contains<T>(&self, source: &T, value: ScalarValue) -> ProxyResult<bool>
where
T: super::reflection::Reflectable,
{
let values = self.get_values(source).await?;
Ok(values.contains(&value))
}
pub async fn count<T>(&self, source: &T) -> ProxyResult<usize>
where
T: super::reflection::Reflectable,
{
let relationship = source
.get_relationship(&self.relationship)
.ok_or_else(|| ProxyError::RelationshipNotFound(self.relationship.clone()))?;
let collection = super::reflection::downcast_relationship::<
Vec<Box<dyn super::reflection::Reflectable>>,
>(relationship)?;
Ok(collection.len())
}
pub async fn filter<T>(
&self,
source: &T,
condition: crate::proxy::query::FilterCondition,
) -> ProxyResult<Vec<ScalarValue>>
where
T: super::reflection::Reflectable,
{
let values = self.get_values(source).await?;
let filtered: Vec<ScalarValue> = values
.into_iter()
.filter(|v| condition.matches(v))
.collect();
Ok(filtered)
}
pub async fn filter_by<T, F>(&self, source: &T, predicate: F) -> ProxyResult<Vec<ScalarValue>>
where
T: super::reflection::Reflectable,
F: Fn(&ScalarValue) -> bool,
{
let values = self.get_values(source).await?;
let filtered: Vec<ScalarValue> = values.into_iter().filter(|v| predicate(v)).collect();
Ok(filtered)
}
pub fn relationship(&self) -> &str {
&self.relationship
}
pub fn attribute(&self) -> &str {
&self.attribute
}
pub fn is_unique(&self) -> bool {
self.unique
}
pub fn deduplicates(&self) -> bool {
self.deduplicate
}
pub fn with_unique(mut self, unique: bool) -> Self {
self.unique = unique;
self
}
pub fn with_deduplication(mut self, deduplicate: bool) -> Self {
self.deduplicate = deduplicate;
self
}
pub fn with_loading_strategy(mut self, strategy: crate::proxy::LoadingStrategy) -> Self {
self.loading_strategy = Some(strategy);
self
}
pub async fn filter_with_any<T>(
&self,
source: &T,
field: &str,
value: &str,
) -> ProxyResult<Vec<ScalarValue>>
where
T: super::reflection::Reflectable,
{
let rel = source
.get_relationship(&self.relationship)
.ok_or_else(|| ProxyError::RelationshipNotFound(self.relationship.clone()))?;
let collection = super::reflection::downcast_relationship::<
Vec<Box<dyn super::reflection::Reflectable>>,
>(rel)?;
let mut filtered_values = Vec::new();
for item in collection.iter() {
if let Some(attr_value) = item.get_attribute(field) {
let matches = match &attr_value {
ScalarValue::String(s) => s.contains(value),
_ => format!("{:?}", attr_value).contains(value),
};
if matches {
if let Some(value) = item.get_attribute(&self.attribute) {
filtered_values.push(value);
}
}
}
}
Ok(filtered_values)
}
pub async fn filter_with_has<T>(
&self,
source: &T,
relationship: &str,
value: &str,
) -> ProxyResult<Vec<ScalarValue>>
where
T: super::reflection::Reflectable,
{
let rel = source
.get_relationship(&self.relationship)
.ok_or_else(|| ProxyError::RelationshipNotFound(self.relationship.clone()))?;
let collection = super::reflection::downcast_relationship::<
Vec<Box<dyn super::reflection::Reflectable>>,
>(rel)?;
let mut filtered_values = Vec::new();
for item in collection.iter() {
if let Some(item_rel) = item.get_relationship(relationship) {
if let Ok(rel_collection) = super::reflection::downcast_relationship::<
Vec<Box<dyn super::reflection::Reflectable>>,
>(item_rel)
{
let has_value = rel_collection.iter().any(|rel_item| {
if let Some(ScalarValue::String(s)) =
rel_item.get_attribute(&self.attribute).as_ref()
{
s == value
} else {
false
}
});
if has_value {
if let Some(value) = item.get_attribute(&self.attribute) {
filtered_values.push(value);
}
}
}
}
}
Ok(filtered_values)
}
pub fn with_cascade(mut self, cascade: bool) -> Self {
self.cascade = cascade;
self
}
pub async fn merge<T>(
&self,
source1: &T,
other: &Self,
source2: &T,
) -> ProxyResult<Vec<ScalarValue>>
where
T: super::reflection::Reflectable,
{
let mut values1 = self.get_values(source1).await?;
let values2 = other.get_values(source2).await?;
values1.extend(values2);
if self.unique || other.unique {
values1.sort_by(|a, b| format!("{:?}", a).cmp(&format!("{:?}", b)));
values1.dedup_by(|a, b| format!("{:?}", a) == format!("{:?}", b));
}
Ok(values1)
}
pub fn with_batch_size(mut self, batch_size: usize) -> Self {
self.batch_size = Some(batch_size);
self
}
pub fn with_async_loading(mut self, async_loading: bool) -> Self {
self.async_loading = async_loading;
self
}
pub fn with_streaming(mut self, streaming: bool) -> Self {
self.streaming = streaming;
self
}
pub fn with_caching(mut self, caching: bool) -> Self {
self.caching = caching;
self
}
pub fn with_database(mut self, database: &str) -> Self {
self.database = Some(database.to_string());
self
}
pub fn with_memory_limit(mut self, memory_limit: usize) -> Self {
self.memory_limit = Some(memory_limit);
self
}
pub fn with_stored_procedure(mut self, procedure: &str) -> Self {
self.stored_procedure = Some(procedure.to_string());
self
}
pub fn with_triggers(mut self, triggers: bool) -> Self {
self.triggers = triggers;
self
}
pub fn with_version_tracking(mut self, version_tracking: bool) -> Self {
self.version_tracking = version_tracking;
self
}
pub async fn bulk_insert<T>(&self, source: &mut T, items: Vec<ScalarValue>) -> ProxyResult<()>
where
T: super::reflection::Reflectable,
{
let factory = self
.factory
.as_ref()
.ok_or(ProxyError::FactoryNotConfigured)?;
let relationship = source
.get_relationship_mut(&self.relationship)
.ok_or_else(|| ProxyError::RelationshipNotFound(self.relationship.clone()))?;
let collection = relationship
.downcast_mut::<Vec<Box<dyn super::reflection::Reflectable>>>()
.ok_or_else(|| ProxyError::TypeMismatch {
expected: "Vec<Box<dyn Reflectable>>".to_string(),
actual: "unknown".to_string(),
})?;
let batch_size = self.batch_size.unwrap_or(items.len());
for chunk in items.chunks(batch_size) {
for value in chunk {
let new_object = factory.create_from_scalar(&self.attribute, value.clone())?;
collection.push(new_object);
}
}
Ok(())
}
pub async fn clear_on<T>(&self, source: &mut T) -> ProxyResult<()>
where
T: super::reflection::Reflectable,
{
let relationship = source
.get_relationship_mut(&self.relationship)
.ok_or_else(|| ProxyError::RelationshipNotFound(self.relationship.clone()))?;
let collection = relationship
.downcast_mut::<Vec<Box<dyn super::reflection::Reflectable>>>()
.ok_or_else(|| ProxyError::TypeMismatch {
expected: "Vec<Box<dyn Reflectable>>".to_string(),
actual: "unknown".to_string(),
})?;
collection.clear();
Ok(())
}
pub fn is_view(&self) -> bool {
self.is_view
}
pub fn with_view(mut self, is_view: bool) -> Self {
self.is_view = is_view;
self
}
pub async fn update_with_version<T>(
&self,
source: &mut T,
value: ScalarValue,
expected_version: i64,
) -> ProxyResult<()>
where
T: super::reflection::Reflectable,
{
if !self.version_tracking {
return Err(ProxyError::VersionTrackingNotEnabled);
}
let relationship = source
.get_relationship_mut(&self.relationship)
.ok_or_else(|| ProxyError::RelationshipNotFound(self.relationship.clone()))?;
let collection = relationship
.downcast_mut::<Vec<Box<dyn super::reflection::Reflectable>>>()
.ok_or_else(|| ProxyError::TypeMismatch {
expected: "Vec<Box<dyn Reflectable>>".to_string(),
actual: "unknown".to_string(),
})?;
for item in collection.iter_mut() {
if let Some(current_version) = item.get_attribute("version")
&& let ScalarValue::Integer(ver) = current_version
{
if ver == expected_version {
item.set_attribute(&self.attribute, value.clone())?;
item.set_attribute("version", ScalarValue::Integer(ver + 1))?;
} else {
return Err(ProxyError::VersionMismatch {
expected: expected_version,
actual: ver,
});
}
}
}
Ok(())
}
pub fn with_cache_ttl(mut self, ttl: u64) -> Self {
self.cache_ttl = Some(ttl);
self
}
pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
self.chunk_size = Some(chunk_size);
self
}
pub fn with_concurrent_access(mut self, concurrent: bool) -> Self {
self.concurrent_access = concurrent;
self
}
pub fn with_fallback_database(mut self, database: &str) -> Self {
self.fallback_database = Some(database.to_string());
self
}
pub fn with_procedure_params(mut self, params: &[(&str, &str)]) -> Self {
self.procedure_params = params
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
self
}
pub fn with_trigger_events(mut self, events: &[&str]) -> Self {
self.trigger_events = events.iter().map(|e| e.to_string()).collect();
self
}
pub fn is_cached(&self) -> bool {
self.caching
}
pub fn cache_ttl(&self) -> Option<u64> {
self.cache_ttl
}
pub fn has_triggers(&self) -> bool {
self.triggers
}
pub fn trigger_events(&self) -> &[String] {
&self.trigger_events
}
pub fn stored_procedure(&self) -> Option<&str> {
self.stored_procedure.as_deref()
}
pub fn procedure_params(&self) -> &[(String, String)] {
&self.procedure_params
}
pub fn database(&self) -> Option<&str> {
self.database.as_deref()
}
pub fn fallback_database(&self) -> Option<&str> {
self.fallback_database.as_deref()
}
pub fn is_async_loading(&self) -> bool {
self.async_loading
}
pub fn supports_concurrent_access(&self) -> bool {
self.concurrent_access
}
pub fn batch_size(&self) -> Option<usize> {
self.batch_size
}
pub fn chunk_size(&self) -> Option<usize> {
self.chunk_size
}
pub fn is_cascade(&self) -> bool {
self.cascade
}
pub fn is_version_tracking(&self) -> bool {
self.version_tracking
}
pub fn memory_limit(&self) -> Option<usize> {
self.memory_limit
}
}
impl std::fmt::Debug for CollectionProxy {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CollectionProxy")
.field("relationship", &self.relationship)
.field("attribute", &self.attribute)
.field("unique", &self.unique)
.field("deduplicate", &self.deduplicate)
.field("loading_strategy", &self.loading_strategy)
.field("factory", &self.factory.as_ref().map(|_| "Some(<factory>)"))
.finish()
}
}
impl Serialize for CollectionProxy {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
use serde::ser::SerializeStruct;
let mut state = serializer.serialize_struct("CollectionProxy", 8)?;
state.serialize_field("relationship", &self.relationship)?;
state.serialize_field("attribute", &self.attribute)?;
state.serialize_field("unique", &self.unique)?;
state.serialize_field("batch_size", &self.batch_size)?;
state.serialize_field("chunk_size", &self.chunk_size)?;
state.serialize_field("cascade", &self.cascade)?;
state.serialize_field("version_tracking", &self.version_tracking)?;
state.serialize_field("memory_limit", &self.memory_limit)?;
state.end()
}
}
impl<'de> Deserialize<'de> for CollectionProxy {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
struct CollectionProxyVisitor;
impl<'de> serde::de::Visitor<'de> for CollectionProxyVisitor {
type Value = CollectionProxy;
fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
formatter.write_str("struct CollectionProxy")
}
fn visit_map<V>(self, mut map: V) -> Result<CollectionProxy, V::Error>
where
V: serde::de::MapAccess<'de>,
{
let mut relationship = None;
let mut attribute = None;
let mut unique = None;
let mut caching = None;
let mut cache_ttl = None;
let mut batch_size = None;
let mut chunk_size = None;
let mut cascade = None;
let mut version_tracking = None;
let mut memory_limit = None;
while let Some(key) = map.next_key::<String>()? {
match key.as_str() {
"relationship" => {
if relationship.is_some() {
return Err(serde::de::Error::duplicate_field("relationship"));
}
relationship = Some(map.next_value()?);
}
"attribute" => {
if attribute.is_some() {
return Err(serde::de::Error::duplicate_field("attribute"));
}
attribute = Some(map.next_value()?);
}
"unique" => {
if unique.is_some() {
return Err(serde::de::Error::duplicate_field("unique"));
}
unique = Some(map.next_value()?);
}
"caching" => {
if caching.is_some() {
return Err(serde::de::Error::duplicate_field("caching"));
}
caching = Some(map.next_value()?);
}
"cache_ttl" => {
if cache_ttl.is_some() {
return Err(serde::de::Error::duplicate_field("cache_ttl"));
}
cache_ttl = Some(map.next_value()?);
}
"batch_size" => {
if batch_size.is_some() {
return Err(serde::de::Error::duplicate_field("batch_size"));
}
batch_size = Some(map.next_value()?);
}
"chunk_size" => {
if chunk_size.is_some() {
return Err(serde::de::Error::duplicate_field("chunk_size"));
}
chunk_size = Some(map.next_value()?);
}
"cascade" => {
if cascade.is_some() {
return Err(serde::de::Error::duplicate_field("cascade"));
}
cascade = Some(map.next_value()?);
}
"version_tracking" => {
if version_tracking.is_some() {
return Err(serde::de::Error::duplicate_field("version_tracking"));
}
version_tracking = Some(map.next_value()?);
}
"memory_limit" => {
if memory_limit.is_some() {
return Err(serde::de::Error::duplicate_field("memory_limit"));
}
memory_limit = Some(map.next_value()?);
}
_ => {
let _ = map.next_value::<serde::de::IgnoredAny>()?;
}
}
}
let relationship =
relationship.ok_or_else(|| serde::de::Error::missing_field("relationship"))?;
let attribute =
attribute.ok_or_else(|| serde::de::Error::missing_field("attribute"))?;
let unique = unique.ok_or_else(|| serde::de::Error::missing_field("unique"))?;
Ok(CollectionProxy {
relationship,
attribute,
unique,
deduplicate: false, loading_strategy: None, factory: None, caching: caching.unwrap_or(false),
cache_ttl,
streaming: false,
triggers: false,
trigger_events: Vec::new(),
stored_procedure: None,
procedure_params: Vec::new(),
database: None,
fallback_database: None,
async_loading: false,
concurrent_access: false,
is_view: false,
batch_size,
chunk_size,
cascade: cascade.unwrap_or(false),
version_tracking: version_tracking.unwrap_or(false),
memory_limit,
})
}
}
const FIELDS: &[&str] = &[
"relationship",
"attribute",
"unique",
"batch_size",
"chunk_size",
"cascade",
"version_tracking",
"memory_limit",
];
deserializer.deserialize_struct("CollectionProxy", FIELDS, CollectionProxyVisitor)
}
}