use tracing::info;
use crate::storage::traits::StorageError;
use super::SyncEngine;
impl SyncEngine {
pub async fn register_schema(&self, schema_name: &str, prefix: &str) -> Result<(), StorageError> {
let table_name = format!("{}_items", schema_name);
if let Some(ref sql_store) = self.sql_store {
sql_store.ensure_table(&table_name).await?;
}
self.schema_registry.register(prefix, &table_name);
info!(
schema = %schema_name,
prefix = %prefix,
table = %table_name,
"Schema registered"
);
Ok(())
}
pub async fn register_schema_prefixes(&self, schema_name: &str, prefixes: &[&str]) -> Result<(), StorageError> {
for prefix in prefixes {
self.register_schema(schema_name, prefix).await?;
}
Ok(())
}
pub fn unregister_prefix(&self, prefix: &str) -> bool {
self.schema_registry.unregister(prefix)
}
#[must_use]
pub fn table_for_key(&self, key: &str) -> &'static str {
self.schema_registry.table_for_key(key)
}
#[must_use]
pub fn registered_tables(&self) -> Vec<String> {
self.schema_registry.tables()
}
#[must_use]
pub fn prefixes_for_table(&self, table_name: &str) -> Vec<String> {
self.schema_registry.prefixes_for_table(table_name)
}
#[must_use]
pub fn is_schema_bypass(&self) -> bool {
self.schema_registry.is_bypass()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::SyncEngineConfig;
use tokio::sync::watch;
fn test_config() -> SyncEngineConfig {
SyncEngineConfig {
redis_url: None,
sql_url: None,
wal_path: None,
l1_max_bytes: 1024 * 1024,
..Default::default()
}
}
#[test]
fn test_table_for_key_empty_registry() {
let config = test_config();
let (_tx, rx) = watch::channel(config.clone());
let engine = SyncEngine::new(config, rx);
assert_eq!(engine.table_for_key("view:users:alice"), "sync_items");
assert_eq!(engine.table_for_key("crdt:users:bob"), "sync_items");
}
#[test]
fn test_table_for_key_with_registrations() {
let config = test_config();
let (_tx, rx) = watch::channel(config.clone());
let engine = SyncEngine::new(config, rx);
engine.schema_registry.register("view:users:", "users_items");
engine.schema_registry.register("crdt:users:", "users_items");
assert_eq!(engine.table_for_key("view:users:alice"), "users_items");
assert_eq!(engine.table_for_key("crdt:users:bob"), "users_items");
assert_eq!(engine.table_for_key("unknown:key"), "sync_items");
}
#[test]
fn test_registered_tables() {
let config = test_config();
let (_tx, rx) = watch::channel(config.clone());
let engine = SyncEngine::new(config, rx);
engine.schema_registry.register("view:users:", "users_items");
engine.schema_registry.register("view:orders:", "orders_items");
let tables = engine.registered_tables();
assert!(tables.contains(&"users_items".to_string()));
assert!(tables.contains(&"orders_items".to_string()));
}
#[test]
fn test_unregister_prefix() {
let config = test_config();
let (_tx, rx) = watch::channel(config.clone());
let engine = SyncEngine::new(config, rx);
engine.schema_registry.register("view:users:", "users_items");
assert_eq!(engine.table_for_key("view:users:alice"), "users_items");
assert!(engine.unregister_prefix("view:users:"));
assert_eq!(engine.table_for_key("view:users:alice"), "sync_items");
}
}