diff --git a/core/crates/solstone-core-entity/src/fixture_tests.rs b/core/crates/solstone-core-entity/src/fixture_tests.rs index eacae0232..b4a2149be 100644 --- a/core/crates/solstone-core-entity/src/fixture_tests.rs +++ b/core/crates/solstone-core-entity/src/fixture_tests.rs @@ -52,6 +52,7 @@ const JOURNAL_IO_ALLOWED: &[&str] = &[ "restore_snapshot", "JournalSnapshot", "SnapshotDirectory", + "SnapshotFile", "SnapshotError", "AppendError", "atomic_replace", diff --git a/core/crates/solstone-core-entity/src/merge_tests.rs b/core/crates/solstone-core-entity/src/merge_tests.rs index eafbda673..c2e0fc266 100644 --- a/core/crates/solstone-core-entity/src/merge_tests.rs +++ b/core/crates/solstone-core-entity/src/merge_tests.rs @@ -226,6 +226,67 @@ fn committed_merge_payload_records_facet_inverse_entries() { fs::remove_dir_all(journal).unwrap(); } +#[test] +fn committed_merge_payload_records_identity_support() { + let journal = voiceprint_journal(); + let source = json!({ + "id": "source", + "name": "Source", + "aka": ["New Alias", "Existing Alias"], + "emails": ["new@example.test", "existing@example.test"], + "title": "Engineer" + }); + let target = json!({ + "id": "target", + "name": "Target", + "aka": ["Existing Alias"], + "emails": ["existing@example.test"] + }); + save_entity_identity(&journal, "source", &source, None).unwrap(); + save_entity_identity(&journal, "target", &target, None).unwrap(); + + let report = + commit_entity_merge(&journal, "source", "target", EntityMergeOptions::default()).unwrap(); + let payload = load_entity_merge_payload(&journal, "target", &report.merge_id).unwrap(); + let identity = &payload["manifest"]["identity"]; + assert!( + identity["aka_support"] + .as_array() + .unwrap() + .contains(&json!({"key":"new alias","target_preexisting":false})) + ); + assert!( + identity["aka_support"] + .as_array() + .unwrap() + .contains(&json!({"key":"existing alias","target_preexisting":true})) + ); + assert!( + identity["email_support"] + .as_array() + .unwrap() + .contains(&json!({"key":"new@example.test","target_preexisting":false})) + ); + assert!( + identity["email_support"] + .as_array() + .unwrap() + .contains(&json!({"key":"existing@example.test","target_preexisting":true})) + ); + assert!( + identity["scalar_support"] + .as_array() + .unwrap() + .contains(&json!({ + "key":"title", + "target_prevalue":null, + "target_prevalue_missing":true, + "source_value":"Engineer" + })) + ); + fs::remove_dir_all(journal).unwrap(); +} + #[test] fn email_dedup_keeps_first_seen_order() { assert_eq!( diff --git a/core/crates/solstone-core-entity/src/store/merge.rs b/core/crates/solstone-core-entity/src/store/merge.rs index 72c030c77..962018685 100644 --- a/core/crates/solstone-core-entity/src/store/merge.rs +++ b/core/crates/solstone-core-entity/src/store/merge.rs @@ -17,6 +17,7 @@ use solstone_core_journal_io::PathOrDay; use solstone_core_journal_io::SnapshotError; use solstone_core_journal_io::append_jsonl; use solstone_core_journal_io::atomic_replace; +use solstone_core_journal_io::capture_snapshot; use solstone_core_journal_io::contained_path; use solstone_core_journal_io::day_dirs; use solstone_core_journal_io::iter_segments; @@ -36,7 +37,7 @@ use crate::{ use super::merge_payload::{ MergePayloadError, list_entity_merge_payload_ids, move_entity_merge_payload, - record_entity_merge_payload, + record_entity_merge_payload, snapshot_payload, }; use super::merge_rollback::MergeRollback; use super::voiceprints::{VoiceprintArchive, read_voiceprints_npz, write_voiceprints_npz}; @@ -233,7 +234,7 @@ pub(crate) fn commit_entity_merge_with_injector( aliases_added: plan.aliases_added, emails_added: plan.emails_added, }; - let mut payload = payload_for_merge(&merge_id, source_id, target_id, &plan); + let mut payload = payload_for_merge(journal, &merge_id, source_id, target_id, &plan)?; let mut touched_facets = Vec::new(); let mut stats = MergeStats::default(); for phase in PHASES { @@ -924,6 +925,9 @@ struct MergePlan { target_after: Value, aliases_added: usize, emails_added: usize, + aka_support: Vec, + email_support: Vec, + scalar_support: Vec, source_display_name: String, target_display_name: String, principal_transferred: bool, @@ -980,11 +984,14 @@ fn plan_merge( })?; let aliases_before = values(&target, "aka"); let mut alias_values = aliases_before.clone(); - alias_values.extend(values(&source, "aka")); + let mut source_aliases = values(&source, "aka"); + alias_values.extend(source_aliases.iter().cloned()); if options.keep_source_as_aka && let Some(name) = source.get("name").and_then(Value::as_str) { - alias_values.push(name.to_owned()); + let name = name.to_owned(); + alias_values.push(name.clone()); + source_aliases.push(name); } let aliases = dedupe_akas(&alias_values); object.insert( @@ -992,11 +999,13 @@ fn plan_merge( Value::Array(aliases.iter().cloned().map(Value::String).collect()), ); let emails_before = values(&target, "emails"); - let emails = dedupe_emails(&emails_before, &values(&source, "emails")); + let source_emails = values(&source, "emails"); + let emails = dedupe_emails(&emails_before, &source_emails); object.insert( "emails".to_owned(), Value::Array(emails.iter().cloned().map(Value::String).collect()), ); + let mut scalar_support = Vec::new(); for (field, value) in source.as_object().expect("identity object") { if ![ "id", @@ -1010,10 +1019,18 @@ fn plan_merge( "is_principal", ] .contains(&field.as_str()) - && is_blank(object.get(field)) && !is_blank(Some(value)) { - object.insert(field.clone(), value.clone()); + let target_prevalue = target.get(field).cloned().unwrap_or(Value::Null); + scalar_support.push(json!({ + "key": field, + "target_prevalue": target_prevalue, + "target_prevalue_missing": target.get(field).is_none(), + "source_value": value, + })); + if is_blank(object.get(field)) { + object.insert(field.clone(), value.clone()); + } } } if source.get("is_principal").and_then(Value::as_bool) == Some(true) { @@ -1036,12 +1053,31 @@ fn plan_merge( target_after: after, aliases_added: aliases.len().saturating_sub(aliases_before.len()), emails_added: emails.len().saturating_sub(emails_before.len()), + aka_support: support_for_values(&source_aliases, &aliases_before), + email_support: support_for_values(&source_emails, &emails_before), + scalar_support, source_display_name, target_display_name, principal_transferred, }) } +fn support_for_values(source_values: &[String], target_values: &[String]) -> Vec { + let target_keys = target_values + .iter() + .map(|value| value.to_lowercase()) + .collect::>(); + let mut seen = HashSet::new(); + source_values + .iter() + .filter_map(|value| { + let key = value.to_lowercase(); + seen.insert(key.clone()) + .then(|| json!({"key": key, "target_preexisting": target_keys.contains(&key)})) + }) + .collect() +} + pub(crate) fn dedupe_akas(values: &[String]) -> Vec { let mut seen = HashSet::new(); let mut output = Vec::new(); @@ -1130,6 +1166,39 @@ fn check_aka_cross_references( ))) } } -fn payload_for_merge(merge_id: &str, source_id: &str, target_id: &str, plan: &MergePlan) -> Value { - json!({"schema_version":1,"merge_id":merge_id,"source_id":source_id,"target_id":target_id,"commit_seq":null,"source_state":{"identity":plan.source_before,"snapshots":[{"rel":format!("entities/{source_id}"),"files":[]}]},"result_counts":{},"manifest":{"identity":{"target_before":plan.target_before,"aka_support":[],"email_support":[],"scalar_support":[]},"voiceprints":{"support":[]},"facets":{"entries":[]},"segments":{"entries":[]},"activities":{"entries":[]},"observation_relations":{"entries":[]},"rebased_merge_ids":[]}}) +fn payload_for_merge( + journal: &Path, + merge_id: &str, + source_id: &str, + target_id: &str, + plan: &MergePlan, +) -> Result { + let mut snapshots = vec![source_snapshot_payload( + journal, + &format!("entities/{source_id}"), + )?]; + let facets = contained_path(journal, "facets") + .map_err(|error| EntityMergeError::Refused(error.to_string()))?; + for entry in + list_dir_entries(&facets).map_err(|error| EntityMergeError::Refused(error.to_string()))? + { + if entry.kind != DirEntryKind::Directory { + continue; + } + let facet = entry.name.to_string_lossy(); + let relative = format!("facets/{facet}/entities/{source_id}"); + let path = contained_path(journal, &relative) + .map_err(|error| EntityMergeError::Refused(error.to_string()))?; + if path_lexists(&path).map_err(|error| EntityMergeError::Refused(error.to_string()))? { + snapshots.push(source_snapshot_payload(journal, &relative)?); + } + } + Ok( + json!({"schema_version":1,"merge_id":merge_id,"source_id":source_id,"target_id":target_id,"commit_seq":null,"source_state":{"identity":plan.source_before,"snapshots":snapshots},"result_counts":{},"manifest":{"identity":{"target_before":plan.target_before,"aka_support":plan.aka_support,"email_support":plan.email_support,"scalar_support":plan.scalar_support},"voiceprints":{"support":[]},"facets":{"entries":[]},"segments":{"entries":[]},"activities":{"entries":[]},"observation_relations":{"entries":[]},"rebased_merge_ids":[]}}), + ) +} + +fn source_snapshot_payload(journal: &Path, relative: &str) -> Result { + let snapshot = capture_snapshot(journal, relative)?; + Ok(json!({"rel":relative,"files":[],"snapshot":snapshot_payload(&snapshot)})) } diff --git a/core/crates/solstone-core-entity/src/store/merge_payload.rs b/core/crates/solstone-core-entity/src/store/merge_payload.rs index 4caf7628f..2803b2176 100644 --- a/core/crates/solstone-core-entity/src/store/merge_payload.rs +++ b/core/crates/solstone-core-entity/src/store/merge_payload.rs @@ -12,7 +12,9 @@ use solstone_core_journal_io::JournalSnapshot; use solstone_core_journal_io::JsonWriteOptions; use solstone_core_journal_io::MalformedPolicy; use solstone_core_journal_io::PathError; +use solstone_core_journal_io::SnapshotDirectory; use solstone_core_journal_io::SnapshotError; +use solstone_core_journal_io::SnapshotFile; use solstone_core_journal_io::contained_path; use solstone_core_journal_io::list_dir_entries; use solstone_core_journal_io::path_lexists; @@ -316,6 +318,75 @@ pub(crate) fn validate_merge_payload( )?; Ok(()) } + +pub(crate) fn snapshot_payload(snapshot: &JournalSnapshot) -> Value { + match snapshot { + JournalSnapshot::Missing { path } => serde_json::json!({"kind":"missing","path":path}), + JournalSnapshot::File(file) => { + serde_json::json!({"kind":"file","path":file.path,"bytes":file.bytes,"mode":file.mode}) + } + JournalSnapshot::Directory(directory) => serde_json::json!({ + "kind":"directory", + "path":directory.path, + "entries":directory.entries.iter().map(snapshot_payload).collect::>(), + }), + } +} + +pub(crate) fn snapshot_from_payload(value: &Value) -> Result { + let object = value + .as_object() + .ok_or_else(|| invalid("merge payload snapshot image is not an object"))?; + let kind = optional_string(object, "kind", "merge payload snapshot image missing kind")?; + let path = optional_string(object, "path", "merge payload snapshot image missing path")?; + match kind { + "missing" => Ok(JournalSnapshot::Missing { + path: path.to_owned(), + }), + "file" => { + let bytes = required_array( + object, + "bytes", + "merge payload snapshot file missing bytes", + "merge payload snapshot file bytes is not a list", + )? + .iter() + .map(|value| { + value + .as_u64() + .and_then(|value| u8::try_from(value).ok()) + .ok_or_else(|| invalid("merge payload snapshot file bytes are invalid")) + }) + .collect::, _>>()?; + let mode = object + .get("mode") + .and_then(Value::as_u64) + .and_then(|value| u32::try_from(value).ok()) + .ok_or_else(|| invalid("merge payload snapshot file missing mode"))?; + Ok(JournalSnapshot::File(SnapshotFile { + path: path.to_owned(), + bytes, + mode, + })) + } + "directory" => { + let entries = required_array( + object, + "entries", + "merge payload snapshot directory missing entries", + "merge payload snapshot directory entries is not a list", + )? + .iter() + .map(snapshot_from_payload) + .collect::, _>>()?; + Ok(JournalSnapshot::Directory(SnapshotDirectory { + path: path.to_owned(), + entries, + })) + } + _ => Err(invalid("merge payload snapshot image has unknown kind")), + } +} fn required_string<'a>( object: &'a serde_json::Map, key: &str,