diff --git a/Cargo.lock b/Cargo.lock index 1bb7cc1..6cc3849 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3325,6 +3325,7 @@ dependencies = [ "serde_json", "serde_yaml", "subtle", + "tempfile", "thiserror 2.0.18", "tidaldb", "tokio", diff --git a/tidal-server/Cargo.toml b/tidal-server/Cargo.toml index fc1ad76..7281c4e 100644 --- a/tidal-server/Cargo.toml +++ b/tidal-server/Cargo.toml @@ -23,3 +23,6 @@ tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } tidaldb = { path = "../tidal" } + +[dev-dependencies] +tempfile = "3" diff --git a/tidal-server/src/config.rs b/tidal-server/src/config.rs index 933b821..311ab79 100644 --- a/tidal-server/src/config.rs +++ b/tidal-server/src/config.rs @@ -3,13 +3,17 @@ use std::path::Path; use std::time::Duration; use serde::Deserialize; +use tidaldb::ranking::profile::{ + Boost, CandidateStrategy, DiversitySpec, Exclude, Gate, Penalty, ProfileDecay, RankingProfile, + SignalAgg, Sort, +}; use tidaldb::schema::{DecaySpec, EntityKind, Schema, SchemaBuilder, TextFieldType, Window}; use crate::error::{Result, ServerError}; const DEFAULT_SCHEMA_YAML: &str = include_str!("../config/default-schema.yaml"); -pub fn load_schema(path: Option<&Path>) -> Result { +pub fn load_schema(path: Option<&Path>) -> Result<(Schema, Vec)> { let raw = read_config(path, DEFAULT_SCHEMA_YAML)?; let spec: SchemaSpec = serde_yaml::from_str(&raw) .map_err(|e| ServerError::SchemaConfig(format!("parse schema yaml: {e}")))?; @@ -20,14 +24,17 @@ pub fn load_schema(path: Option<&Path>) -> Result { )); } + // Collect signal names for profile validation. + let signal_names: Vec = spec.signals.iter().map(|s| s.name.clone()).collect(); + let mut builder = SchemaBuilder::new(); - for signal in spec.signals { + for signal in &spec.signals { let mut sig = builder.signal( &signal.name, parse_entity_kind(&signal.entity)?, signal.decay.to_decay_spec()?, ); - if let Some(windows) = signal.windows { + if let Some(ref windows) = signal.windows { let parsed: Result> = windows.iter().map(|w| parse_window(w)).collect(); sig = sig.windows(&parsed?); } @@ -37,13 +44,13 @@ pub fn load_schema(path: Option<&Path>) -> Result { let _ = sig.add(); } - if let Some(text_fields) = spec.text_fields { + if let Some(ref text_fields) = spec.text_fields { for field in text_fields { builder.text_field(&field.name, parse_text_field_type(&field.kind)?); } } - if let Some(embeddings) = spec.embedding_slots { + if let Some(ref embeddings) = spec.embedding_slots { for slot in embeddings { builder.embedding_slot( &slot.name, @@ -53,7 +60,22 @@ pub fn load_schema(path: Option<&Path>) -> Result { } } - builder.build().map_err(ServerError::SchemaBuild) + let schema = builder.build().map_err(ServerError::SchemaBuild)?; + + // Parse profiles and validate signal references. + let profiles = if let Some(profile_specs) = spec.profiles { + let mut parsed = Vec::with_capacity(profile_specs.len()); + for spec in profile_specs { + let profile = spec.to_ranking_profile()?; + validate_profile_signals(&profile, &signal_names)?; + parsed.push(profile); + } + parsed + } else { + Vec::new() + }; + + Ok((schema, profiles)) } fn read_config(path: Option<&Path>, fallback: &str) -> Result { @@ -70,6 +92,8 @@ struct SchemaSpec { text_fields: Option>, #[serde(default)] embedding_slots: Option>, + #[serde(default)] + profiles: Option>, } #[derive(Debug, Deserialize)] @@ -171,3 +195,598 @@ fn parse_text_field_type(input: &str) -> Result { ))), } } + +// ── Profile YAML types ────────────────────────────────────────────────────── + +#[derive(Debug, Deserialize)] +struct ProfileSpec { + name: String, + #[serde(default = "default_version")] + version: u32, + #[serde(default)] + candidate_strategy: Option, + #[serde(default)] + sort: Option, + #[serde(default)] + boosts: Vec, + #[serde(default)] + decay: Option, + #[serde(default)] + gates: Vec, + #[serde(default)] + penalties: Vec, + #[serde(default)] + excludes: Vec, + #[serde(default)] + diversity: Option, + #[serde(default)] + exploration: f64, +} + +fn default_version() -> u32 { + 1 +} + +#[derive(Debug, Deserialize)] +struct BoostSpec { + signal: String, + agg: String, + window: String, + weight: f64, +} + +#[derive(Debug, Deserialize)] +struct GateSpec { + signal: String, + agg: String, + window: String, + min_threshold: f64, +} + +#[derive(Debug, Deserialize)] +struct PenaltySpec { + signal: String, + agg: String, + window: String, + weight: f64, +} + +#[derive(Debug, Deserialize)] +struct ExcludeSpec { + signal: String, + agg: String, + window: String, + above: f64, +} + +#[derive(Debug, Deserialize)] +struct ProfileDecaySpec { + signal: String, + half_life_secs: u64, + weight: f64, +} + +#[derive(Debug, Deserialize)] +struct DiversitySpecConfig { + #[serde(default)] + max_per_creator: Option, + #[serde(default)] + format_mix_max_fraction: Option, +} + +/// Sort can be a simple string (`"trending"`, `"new"`) or a map with parameters. +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum SortSpec { + Simple(String), + Parameterized(std::collections::HashMap), +} + +/// Candidate strategy can be a simple string or a map with parameters. +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum CandidateStrategySpec { + Simple(String), + Parameterized(std::collections::HashMap), +} + +// ── Profile conversion ────────────────────────────────────────────────────── + +impl ProfileSpec { + fn to_ranking_profile(&self) -> Result { + let candidate_strategy = match &self.candidate_strategy { + Some(spec) => parse_candidate_strategy(spec)?, + None => CandidateStrategy::Scan { + sort_field: "created_at".into(), + }, + }; + + let sort = match &self.sort { + Some(spec) => Some(parse_sort(spec)?), + None => None, + }; + + let boosts = self + .boosts + .iter() + .map(|b| { + Ok(Boost { + signal: b.signal.clone(), + agg: parse_agg(&b.agg)?, + window: parse_window(&b.window)?, + weight: b.weight, + }) + }) + .collect::>>()?; + + let gates = self + .gates + .iter() + .map(|g| { + Ok(Gate { + signal: g.signal.clone(), + agg: parse_agg(&g.agg)?, + window: parse_window(&g.window)?, + min_threshold: g.min_threshold, + }) + }) + .collect::>>()?; + + let penalties = self + .penalties + .iter() + .map(|p| { + Ok(Penalty { + signal: p.signal.clone(), + agg: parse_agg(&p.agg)?, + window: parse_window(&p.window)?, + weight: p.weight, + }) + }) + .collect::>>()?; + + let excludes = self + .excludes + .iter() + .map(|e| { + Ok(Exclude { + signal: e.signal.clone(), + agg: parse_agg(&e.agg)?, + window: parse_window(&e.window)?, + above: e.above, + }) + }) + .collect::>>()?; + + let decay = self.decay.as_ref().map(|d| ProfileDecay { + signal: d.signal.clone(), + half_life_secs: d.half_life_secs, + weight: d.weight, + }); + + let diversity = match &self.diversity { + Some(d) => DiversitySpec { + max_per_creator: d.max_per_creator, + format_mix_max_fraction: d.format_mix_max_fraction, + }, + None => DiversitySpec::default(), + }; + + Ok(RankingProfile { + name: self.name.clone(), + version: self.version, + candidate_strategy, + boosts, + decay, + gates, + penalties, + excludes, + diversity, + exploration: self.exploration, + sort, + is_builtin: false, + }) + } +} + +fn parse_agg(input: &str) -> Result { + match input.trim().to_lowercase().as_str() { + "value" => Ok(SignalAgg::Value), + "velocity" => Ok(SignalAgg::Velocity), + "decay_score" => Ok(SignalAgg::DecayScore), + "ratio" => Ok(SignalAgg::Ratio), + "relative_velocity" => Ok(SignalAgg::RelativeVelocity), + other => Err(ServerError::SchemaConfig(format!( + "unknown signal aggregation '{other}'" + ))), + } +} + +fn parse_sort(spec: &SortSpec) -> Result { + match spec { + SortSpec::Simple(s) => match s.trim().to_lowercase().as_str() { + "trending" => Ok(Sort::Trending), + "rising" => Ok(Sort::Rising), + "controversial" => Ok(Sort::Controversial), + "hidden_gems" => Ok(Sort::HiddenGems), + "shuffle" => Ok(Sort::Shuffle), + "new" => Ok(Sort::New), + "most_followed" => Ok(Sort::MostFollowed), + "creator_engagement_rate" => Ok(Sort::CreatorEngagementRate), + "alphabetical_asc" => Ok(Sort::AlphabeticalAsc), + "alphabetical_desc" => Ok(Sort::AlphabeticalDesc), + "shortest" => Ok(Sort::Shortest), + "longest" => Ok(Sort::Longest), + "live_viewer_count" => Ok(Sort::LiveViewerCount), + "date_saved" => Ok(Sort::DateSaved), + other => Err(ServerError::SchemaConfig(format!( + "unknown sort mode '{other}'" + ))), + }, + SortSpec::Parameterized(map) => { + if let Some(val) = map.get("hot") { + let gravity = extract_f64(val, "gravity")?.unwrap_or(1.8); + Ok(Sort::Hot { gravity }) + } else if let Some(val) = map.get("top_window") { + let window = extract_window(val, "window")?; + Ok(Sort::TopWindow { window }) + } else if let Some(val) = map.get("most_viewed") { + let window = extract_window(val, "window")?; + Ok(Sort::MostViewed { window }) + } else if let Some(val) = map.get("most_liked") { + let window = extract_window(val, "window")?; + Ok(Sort::MostLiked { window }) + } else if let Some(val) = map.get("most_commented") { + let window = extract_window(val, "window")?; + Ok(Sort::MostCommented { window }) + } else if let Some(val) = map.get("most_shared") { + let window = extract_window(val, "window")?; + Ok(Sort::MostShared { window }) + } else { + Err(ServerError::SchemaConfig(format!( + "unknown parameterized sort: {map:?}" + ))) + } + } + } +} + +fn parse_candidate_strategy(spec: &CandidateStrategySpec) -> Result { + match spec { + CandidateStrategySpec::Simple(s) => match s.trim().to_lowercase().as_str() { + "scan" => Ok(CandidateStrategy::Scan { + sort_field: "created_at".into(), + }), + "relationship" => Ok(CandidateStrategy::Relationship), + "hybrid" => Ok(CandidateStrategy::Hybrid), + "cohort_trending" => Ok(CandidateStrategy::CohortTrending), + other => Err(ServerError::SchemaConfig(format!( + "unknown candidate strategy '{other}'" + ))), + }, + CandidateStrategySpec::Parameterized(map) => { + if let Some(val) = map.get("ann") { + let slot = extract_string(val, "slot")?; + let limit = extract_u64(val, "limit")?.unwrap_or(100) as usize; + Ok(CandidateStrategy::Ann { slot, limit }) + } else if let Some(val) = map.get("signal_ranked") { + let signal = extract_string(val, "signal")?; + let window_str = extract_string_field(val, "window")?; + let window = parse_window(&window_str)?; + Ok(CandidateStrategy::SignalRanked { signal, window }) + } else if let Some(val) = map.get("scan") { + let sort_field = + extract_string_field(val, "sort_field").unwrap_or_else(|_| "created_at".into()); + Ok(CandidateStrategy::Scan { sort_field }) + } else { + Err(ServerError::SchemaConfig(format!( + "unknown parameterized candidate strategy: {map:?}" + ))) + } + } + } +} + +// ── YAML value extraction helpers ─────────────────────────────────────────── + +fn extract_f64(val: &serde_yaml::Value, field: &str) -> Result> { + match val { + serde_yaml::Value::Mapping(map) => { + if let Some(v) = map.get(serde_yaml::Value::String(field.to_owned())) { + v.as_f64() + .or_else(|| v.as_i64().map(|i| i as f64)) + .map(Some) + .ok_or_else(|| ServerError::SchemaConfig(format!("'{field}' must be a number"))) + } else { + Ok(None) + } + } + _ => Ok(None), + } +} + +fn extract_u64(val: &serde_yaml::Value, field: &str) -> Result> { + match val { + serde_yaml::Value::Mapping(map) => { + if let Some(v) = map.get(serde_yaml::Value::String(field.to_owned())) { + v.as_u64().map(Some).ok_or_else(|| { + ServerError::SchemaConfig(format!("'{field}' must be a positive integer")) + }) + } else { + Ok(None) + } + } + _ => Ok(None), + } +} + +fn extract_string(val: &serde_yaml::Value, field: &str) -> Result { + extract_string_field(val, field) +} + +fn extract_string_field(val: &serde_yaml::Value, field: &str) -> Result { + match val { + serde_yaml::Value::Mapping(map) => { + if let Some(v) = map.get(serde_yaml::Value::String(field.to_owned())) { + v.as_str() + .map(|s| s.to_owned()) + .ok_or_else(|| ServerError::SchemaConfig(format!("'{field}' must be a string"))) + } else { + Err(ServerError::SchemaConfig(format!( + "missing required field '{field}'" + ))) + } + } + _ => Err(ServerError::SchemaConfig(format!( + "expected mapping with field '{field}'" + ))), + } +} + +fn extract_window(val: &serde_yaml::Value, field: &str) -> Result { + let s = extract_string_field(val, field)?; + parse_window(&s) +} + +// ── Signal validation ─────────────────────────────────────────────────────── + +fn validate_profile_signals(profile: &RankingProfile, signal_names: &[String]) -> Result<()> { + let check = |signal: &str, context: &str| -> Result<()> { + if !signal_names.iter().any(|s| s == signal) { + Err(ServerError::SchemaConfig(format!( + "profile '{}': {} references unknown signal '{signal}'", + profile.name, context + ))) + } else { + Ok(()) + } + }; + + for boost in &profile.boosts { + check(&boost.signal, "boost")?; + } + for gate in &profile.gates { + check(&gate.signal, "gate")?; + } + for penalty in &profile.penalties { + check(&penalty.signal, "penalty")?; + } + for exclude in &profile.excludes { + check(&exclude.signal, "exclude")?; + } + if let Some(ref decay) = profile.decay { + check(&decay.signal, "decay")?; + } + Ok(()) +} + +// ── Tests ─────────────────────────────────────────────────────────────────── + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use super::*; + + #[test] + fn parse_schema_with_profiles() { + let yaml = r#" +signals: + - name: chat + entity: item + decay: + exponential: + half_life_seconds: 1209600 + windows: [24h, 7d] + - name: favorite + entity: item + decay: + exponential: + half_life_seconds: 2592000 + windows: [7d, 30d] + +profiles: + - name: for_you + sort: + hot: + gravity: 1.5 + boosts: + - signal: chat + agg: decay_score + window: all_time + weight: 1.5 + - signal: favorite + agg: decay_score + window: all_time + weight: 2.5 + diversity: + max_per_creator: 2 + exploration: 0.1 +"#; + let spec: SchemaSpec = serde_yaml::from_str(yaml).unwrap(); + assert!(spec.profiles.is_some()); + let profiles = spec.profiles.unwrap(); + assert_eq!(profiles.len(), 1); + let profile = profiles[0].to_ranking_profile().unwrap(); + assert_eq!(profile.name, "for_you"); + assert_eq!(profile.boosts.len(), 2); + assert!(!profile.is_builtin); + assert!((profile.exploration - 0.1).abs() < f64::EPSILON); + assert!( + matches!(profile.sort, Some(Sort::Hot { gravity }) if (gravity - 1.5).abs() < f64::EPSILON) + ); + assert_eq!(profile.diversity.max_per_creator, Some(2)); + } + + #[test] + fn parse_simple_sort_modes() { + let cases = vec![ + ("trending", true), + ("new", true), + ("shuffle", true), + ("rising", true), + ("controversial", true), + ("hidden_gems", true), + ("bad_sort", false), + ]; + for (input, should_ok) in cases { + let result = parse_sort(&SortSpec::Simple(input.into())); + assert_eq!(result.is_ok(), should_ok, "sort '{input}'"); + } + } + + #[test] + fn parse_parameterized_sort() { + let mut map = std::collections::HashMap::new(); + let mut inner = serde_yaml::Mapping::new(); + inner.insert( + serde_yaml::Value::String("window".into()), + serde_yaml::Value::String("7d".into()), + ); + map.insert("top_window".into(), serde_yaml::Value::Mapping(inner)); + let result = parse_sort(&SortSpec::Parameterized(map)).unwrap(); + assert!(matches!( + result, + Sort::TopWindow { + window: Window::SevenDays + } + )); + } + + #[test] + fn parse_agg_variants() { + assert!(matches!(parse_agg("value").unwrap(), SignalAgg::Value)); + assert!(matches!( + parse_agg("velocity").unwrap(), + SignalAgg::Velocity + )); + assert!(matches!( + parse_agg("decay_score").unwrap(), + SignalAgg::DecayScore + )); + assert!(matches!(parse_agg("ratio").unwrap(), SignalAgg::Ratio)); + assert!(matches!( + parse_agg("relative_velocity").unwrap(), + SignalAgg::RelativeVelocity + )); + assert!(parse_agg("unknown").is_err()); + } + + #[test] + fn parse_candidate_strategy_variants() { + let scan = parse_candidate_strategy(&CandidateStrategySpec::Simple("scan".into())).unwrap(); + assert!(matches!(scan, CandidateStrategy::Scan { .. })); + + let rel = parse_candidate_strategy(&CandidateStrategySpec::Simple("relationship".into())) + .unwrap(); + assert!(matches!(rel, CandidateStrategy::Relationship)); + } + + #[test] + fn unknown_signal_in_profile_rejected() { + let profile = RankingProfile { + name: "test".into(), + version: 1, + candidate_strategy: CandidateStrategy::Scan { + sort_field: "created_at".into(), + }, + boosts: vec![Boost { + signal: "nonexistent".into(), + agg: SignalAgg::Value, + window: Window::AllTime, + weight: 1.0, + }], + decay: None, + gates: vec![], + penalties: vec![], + excludes: vec![], + diversity: DiversitySpec::default(), + exploration: 0.0, + sort: None, + is_builtin: false, + }; + let signal_names = vec!["chat".into(), "favorite".into()]; + let result = validate_profile_signals(&profile, &signal_names); + assert!(result.is_err()); + let err = result.unwrap_err().to_string(); + assert!(err.contains("nonexistent"), "error: {err}"); + } + + #[test] + fn load_schema_returns_tuple() { + let (schema, profiles) = load_schema(None).unwrap(); + // Default schema has no profiles section. + assert!(profiles.is_empty()); + assert!(schema.signals().next().is_some()); + } + + #[test] + fn schema_with_no_profiles_section() { + let yaml = r#" +signals: + - name: view + entity: item + decay: + exponential: + half_life_seconds: 604800 + windows: [24h, 7d] +"#; + let tmp = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(tmp.path(), yaml).unwrap(); + let (_, profiles) = load_schema(Some(tmp.path())).unwrap(); + assert!(profiles.is_empty()); + } + + #[test] + fn profile_defaults() { + let yaml = r#" +signals: + - name: view + entity: item + decay: + exponential: + half_life_seconds: 604800 + windows: [24h, 7d] + +profiles: + - name: minimal + boosts: + - signal: view + agg: value + window: all_time + weight: 1.0 +"#; + let tmp = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(tmp.path(), yaml).unwrap(); + let (_, profiles) = load_schema(Some(tmp.path())).unwrap(); + assert_eq!(profiles.len(), 1); + let p = &profiles[0]; + assert_eq!(p.version, 1); + assert!(!p.is_builtin); + assert!(p.sort.is_none()); + assert!((p.exploration).abs() < f64::EPSILON); + assert!(matches!( + p.candidate_strategy, + CandidateStrategy::Scan { .. } + )); + } +} diff --git a/tidal-server/src/main.rs b/tidal-server/src/main.rs index f06d8c3..a236674 100644 --- a/tidal-server/src/main.rs +++ b/tidal-server/src/main.rs @@ -76,9 +76,11 @@ fn init_tracing() { } async fn run_standalone(args: StandaloneArgs) -> Result<()> { - let schema = load_schema(args.schema.as_deref())?; + let (schema, profiles) = load_schema(args.schema.as_deref())?; - let mut builder = TidalDb::builder().with_schema(schema.clone()); + let mut builder = TidalDb::builder() + .with_schema(schema.clone()) + .with_profiles(profiles); if let Some(dir) = args.data_dir { builder = builder.with_data_dir(dir); } else { diff --git a/tidal-server/tests/middleware.rs b/tidal-server/tests/middleware.rs index f198917..e1a7a21 100644 --- a/tidal-server/tests/middleware.rs +++ b/tidal-server/tests/middleware.rs @@ -17,10 +17,11 @@ use tower::ServiceExt; fn make_state() -> Arc { // Use the default schema so ranking profiles are available — same setup as // `run_standalone` in main.rs. - let schema = tidal_server::config::load_schema(None).unwrap(); + let (schema, profiles) = tidal_server::config::load_schema(None).unwrap(); let db = TidalDb::builder() .ephemeral() .with_schema(schema) + .with_profiles(profiles) .open() .unwrap(); Arc::new(ServerState::new(db)) diff --git a/tidal/src/db/builder.rs b/tidal/src/db/builder.rs index dfa1732..42f31ce 100644 --- a/tidal/src/db/builder.rs +++ b/tidal/src/db/builder.rs @@ -10,6 +10,7 @@ use super::metrics::MetricsState; use super::paths::Paths; use crate::load::RateLimiterConfig; +use crate::ranking::profile::RankingProfile; use crate::schema::Schema; /// Fluent builder for constructing a [`TidalDb`] instance. @@ -55,6 +56,8 @@ pub struct TidalDbBuilder { /// When set and `NodeRole::Follower`, the receiver is started automatically. /// When set and `NodeRole::Leader`, the WAL shipper is spawned automatically. transport: Option>, + /// Schema-defined profiles that override matching builtins. + profiles: Vec, } impl TidalDbBuilder { @@ -67,6 +70,7 @@ impl TidalDbBuilder { schema: None, rate_limiter_config: None, transport: None, + profiles: Vec::new(), } } @@ -84,6 +88,18 @@ impl TidalDbBuilder { self } + /// Attach schema-defined ranking profiles that override matching builtins. + /// + /// Profiles provided here are registered via [`override_register`] after + /// built-in profiles during [`open`](Self::open). This allows schema YAML + /// to replace built-in profile definitions with deployment-specific signal + /// names and tuning parameters. + #[must_use] + pub fn with_profiles(mut self, profiles: Vec) -> Self { + self.profiles = profiles; + self + } + /// Configure per-agent session rate limiting. /// /// When set, each `(agent_id, session_id)` pair gets a token bucket with @@ -328,7 +344,7 @@ impl TidalDbBuilder { if let Some(schema) = self.schema { // Wire in storage, WAL, signal ledger, and M2 indexes. - let result = TidalDb::open_with_schema(&self.config, schema)?; + let result = TidalDb::open_with_schema(&self.config, schema, self.profiles)?; // ── Fix B: Schema fingerprint persistence ─────────────────── // Persistent mode only. Compute a BLAKE3 hash of the schema's diff --git a/tidal/src/db/builder_tests.rs b/tidal/src/db/builder_tests.rs index e3e1517..c1ed775 100644 --- a/tidal/src/db/builder_tests.rs +++ b/tidal/src/db/builder_tests.rs @@ -28,6 +28,7 @@ fn builder_persistent_requires_data_dir() { schema: None, rate_limiter_config: None, transport: None, + profiles: Vec::new(), }; let result = builder.validate(); assert!(result.is_err()); diff --git a/tidal/src/db/open.rs b/tidal/src/db/open.rs index e9dab8f..31b285d 100644 --- a/tidal/src/db/open.rs +++ b/tidal/src/db/open.rs @@ -9,6 +9,7 @@ use crate::entities::{ CreatorItemsBitmap, HardNegIndex, InteractionLedger, PreferenceVectors, UserStateIndex, }; use crate::ranking::builtins::register_builtins; +use crate::ranking::profile::RankingProfile; use crate::ranking::registry::ProfileRegistry; use crate::schema::{DurabilityError, EntityId, Schema, TidalError, Timestamp}; use crate::signals::{NoopWalWriter, SignalLedger, SignalTypeId}; @@ -61,6 +62,7 @@ impl super::TidalDb { pub(crate) fn open_with_schema( config: &super::Config, schema: Schema, + schema_profiles: Vec, ) -> crate::Result { let last_seq = Arc::new(AtomicU64::new(0)); @@ -70,6 +72,17 @@ impl super::TidalDb { TidalError::internal("open", format!("failed to register builtin profiles: {e}")) })?; + // Register schema-defined profiles, overriding matching builtins. + for profile in schema_profiles { + let name = profile.name.clone(); + profile_registry.override_register(profile).map_err(|e| { + TidalError::internal( + "open", + format!("failed to register schema profile '{name}': {e}"), + ) + })?; + } + // Initialize M2 indexes (empty -- populated as items are written). let category_index = BitmapIndex::new("category"); let format_index = BitmapIndex::new("format"); diff --git a/tidal/src/ranking/registry.rs b/tidal/src/ranking/registry.rs index 49a8a84..b4b36f4 100644 --- a/tidal/src/ranking/registry.rs +++ b/tidal/src/ranking/registry.rs @@ -115,6 +115,46 @@ impl ProfileRegistry { Ok(()) } + /// Override-register a profile, replacing all existing versions for that name. + /// + /// Clears any previously registered versions (including builtins) and inserts + /// the new profile as the sole version. Applies the same validation as + /// [`register`](Self::register) (name format, exploration bounds, gate + /// thresholds) but skips version monotonicity since all prior versions are + /// removed. + /// + /// This is used by schema-defined profiles to cleanly replace built-in + /// defaults without version gymnastics. + /// + /// # Errors + /// + /// Same as [`register`](Self::register) except `VersionConflict` is impossible. + pub fn override_register(&mut self, profile: RankingProfile) -> Result<(), ProfileError> { + if !is_valid_name(&profile.name) { + return Err(ProfileError::InvalidName(profile.name)); + } + + if !(0.0..=0.5).contains(&profile.exploration) { + return Err(ProfileError::ExplorationOutOfRange(profile.exploration)); + } + + for gate in &profile.gates { + if matches!(gate.agg, SignalAgg::DecayScore | SignalAgg::Ratio) { + if !(0.0..=1.0).contains(&gate.min_threshold) { + return Err(ProfileError::GateThresholdOutOfRange(gate.min_threshold)); + } + } else if gate.min_threshold < 0.0 { + return Err(ProfileError::GateThresholdOutOfRange(gate.min_threshold)); + } + } + + // Clear all existing versions for this name, then insert the new one. + let versions = self.profiles.entry(profile.name.clone()).or_default(); + versions.clear(); + versions.insert(profile.version, profile); + Ok(()) + } + /// Get the latest version of a named profile. /// /// # Errors @@ -289,4 +329,68 @@ mod tests { let result = registry.get("nonexistent"); assert!(matches!(result, Err(ProfileError::NotFound(_)))); } + + #[test] + fn override_register_replaces_builtin() { + let mut registry = ProfileRegistry::new(); + let mut builtin = minimal_profile("for_you", 1); + builtin.is_builtin = true; + registry.register(builtin).unwrap(); + + // Override with a schema-defined version. + let mut override_profile = minimal_profile("for_you", 1); + override_profile.exploration = 0.1; + registry.override_register(override_profile).unwrap(); + + let profile = registry.get("for_you").unwrap(); + assert!(!profile.is_builtin); + assert!((profile.exploration - 0.1).abs() < f64::EPSILON); + } + + #[test] + fn override_register_clears_all_versions() { + let mut registry = ProfileRegistry::new(); + registry.register(minimal_profile("trending", 1)).unwrap(); + registry.register(minimal_profile("trending", 2)).unwrap(); + + // Override clears both v1 and v2. + let override_profile = minimal_profile("trending", 1); + registry.override_register(override_profile).unwrap(); + + // Only v1 exists now. + assert!(registry.get_version("trending", 2).is_err()); + assert!(registry.get_version("trending", 1).is_ok()); + } + + #[test] + fn override_register_validates_constraints() { + let mut registry = ProfileRegistry::new(); + + // Invalid name. + let result = registry.override_register(minimal_profile("BAD NAME", 1)); + assert!(matches!(result, Err(ProfileError::InvalidName(_)))); + + // Exploration out of range. + let mut profile = minimal_profile("bad_explore", 1); + profile.exploration = 0.9; + let result = registry.override_register(profile); + assert!(matches!( + result, + Err(ProfileError::ExplorationOutOfRange(_)) + )); + + // Gate threshold out of range. + let mut profile = minimal_profile("bad_gate", 1); + profile.gates.push(Gate { + signal: "view".into(), + agg: SignalAgg::DecayScore, + window: Window::AllTime, + min_threshold: 1.5, + }); + let result = registry.override_register(profile); + assert!(matches!( + result, + Err(ProfileError::GateThresholdOutOfRange(_)) + )); + } }