//! Boot-time disk→relay event reconcile ("event sync"). //! //! Reconciles the on-disk JSON stores (`personas.json`, `teams.json`, //! `managed-agents.json`) into signed retention events queued for relay //! publish. Runs after identity resolution (event signing needs the owner //! keys), unlike the pre-identity migrations in [`crate::migration`]. use std::path::Path; /// Reconcile personas, teams, and managed agents into signed retention /// events. All readers consume the already-synced /// `personas.json`/`teams.json`/`managed-agents.json` that /// `sync_team_personas` wrote in [`crate::migration::run_boot_migrations`] /// (see its `# Ordering` guard). Event signing needs the resolved owner keys, /// so this runs after identity resolution, not in the boot migrations. pub fn run_event_sync( app: &tauri::AppHandle, owner_keys: &nostr::Keys, db_path: &Path, ) -> Result<(), String> { // Persona and agent legs stay best-effort: they log and swallow, and their // failure does not undo the boot team-membership repair. The team leg is // fatal — it establishes the superseding local head (a monotonic // `created_at`) that lets `retain_inbound_event`'s equal/older guard reject // a stale relay roster. If it fails, the caller must not let the frontend // expose the community and start inbound replay against an un-superseded // disk state. migrate_personas_to_events(app, owner_keys, db_path); migrate_teams_to_events(app, owner_keys, db_path)?; crate::managed_agents::reconcile::reconcile_agents_to_events(app, owner_keys, db_path); Ok(()) } /// Run the scoped event reconcile to completion on the blocking pool. /// /// Callers that must not let downstream work observe a not-yet-retained disk /// state (e.g. `apply_workspace` before the frontend can start inbound history /// replay) await this so the repaired local heads are durably retained — with a /// superseding `monotonic_created_at` — before an old relay head can race in. /// The owner keys are moved in so the task never touches the `AppState::keys` /// mutex; the reconcile itself is synchronous JSON/SQLite/signing work, so it /// runs on the blocking pool rather than an async worker. /// /// Returns `Err` if the task fails to join or the fatal team leg errors, so the /// caller can withhold community exposure until the superseding head is durable. pub async fn run_event_sync_blocking( app: tauri::AppHandle, owner_keys: nostr::Keys, db_path: std::path::PathBuf, ) -> Result<(), String> { tauri::async_runtime::spawn_blocking(move || run_event_sync(&app, &owner_keys, &db_path)) .await .map_err(|e| format!("event-sync: spawn_blocking failed: {e}"))? } /// Reconcile `personas.json` into the persona-event retention store. /// /// Must run AFTER `fold_personas_into_agent_store` and /// `detach_directory_backed_teams` (depends on field renames and store /// unification being complete) and AFTER the persisted identity is resolved /// (it signs every retained event with the owner's keys). /// /// Per-record reconcile: for each non-builtin persona it compares the freshly /// serialized event content against the retained row at the same coordinate /// and re-retains (marking `pending_sync = 1`) only when the row is absent or /// its content differs. An unchanged persona is left untouched, so a launch /// after a no-op edit does not churn `pending_sync`; a persona added or edited /// on disk between launches is picked up and republished. There is no /// whole-store sentinel — comparing per coordinate is what lets newly added /// personas reach the relay. /// /// Strategy: write to local SQLite retention first (durable copy), mark as /// `pending_sync = 1` for later relay publish. Migration succeeds on local /// write, not relay acknowledgment. Every retained row is a real signed /// event — there is no placeholder path. pub fn migrate_personas_to_events(app: &tauri::AppHandle, keys: &nostr::Keys, db_path: &Path) { use crate::managed_agents::managed_agents_base_dir; let Ok(base_dir) = managed_agents_base_dir(app) else { return; }; match migrate_personas_in_dir_at(&base_dir, keys, db_path) { Ok(0) => {} Ok(migrated) => { eprintln!( "buzz-desktop: persona-event-migration: {migrated} personas migrated to retention" ); } Err(e) => { eprintln!("buzz-desktop: persona-event-migration: {e}"); } } } /// Core reconcile logic, decoupled from the Tauri `AppHandle` for testing. /// /// Returns the number of personas (re)written to the retention store. Returns /// `Ok(0)` when every non-builtin persona already has a matching retained row /// (or there are none to reconcile). #[cfg(test)] fn migrate_personas_in_dir(base_dir: &Path, keys: &nostr::Keys) -> Result { migrate_personas_in_dir_at(base_dir, keys, &base_dir.join("retention.db")) } fn migrate_personas_in_dir_at( base_dir: &Path, keys: &nostr::Keys, db_path: &Path, ) -> Result { use crate::managed_agents::{ persona_events::{build_persona_event, monotonic_created_at, persona_d_tag}, retention::{get_retained_event, open_retention_db, retain_event, RetainedEvent}, AgentDefinition, }; use buzz_core_pkg::kind::KIND_PERSONA; use nostr::JsonUtil; let pubkey = keys.public_key().to_hex(); // Post-fold (Phase 1A.2): definitions live as key-less records in the // unified agent store, presented in the legacy shape. Pre-fold boots // (run_event_sync runs after run_boot_migrations, so the fold has // already happened) never reach this path with personas.json present — // but read it as a fallback for one release in case the fold errored. let records: Vec = { let personas_path = base_dir.join("personas.json"); if personas_path.exists() { let content = std::fs::read_to_string(&personas_path) .map_err(|e| format!("failed to read personas.json: {e}"))?; serde_json::from_str(&content) .map_err(|e| format!("failed to parse personas.json: {e}"))? } else { let agents_path = base_dir.join("managed-agents.json"); if !agents_path.exists() { return Ok(0); } let content = std::fs::read_to_string(&agents_path) .map_err(|e| format!("failed to read managed-agents.json: {e}"))?; let all: Vec = serde_json::from_str(&content) .map_err(|e| format!("failed to parse managed-agents.json: {e}"))?; all.iter() .filter(|record| record.pubkey.is_empty()) .filter_map(|record| record.to_definition_view()) .collect() } }; if records.is_empty() { return Ok(0); } // Open (or create) the retention database. let conn = open_retention_db(db_path).map_err(|e| format!("failed to open retention db: {e}"))?; let mut migrated = 0u32; for record in &records { // Skip built-in personas — they're always available from code. if record.is_builtin { continue; } let d_tag = persona_d_tag(record); // Fetch the retained head first so the rebuilt event can supersede it: // build at the default `now` and a future-dated head (clock skew, or an // interactive same-second `max(now, head+1)` bump) would make // `retain_event`'s `created_at >= ...` guard SILENTLY skip the UPDATE // while `migrated` over-reports. Mirror the interactive sites' monotonic // bump (F1) so a changed body always lands. let existing = get_retained_event(&conn, KIND_PERSONA, &pubkey, &d_tag)?; let mut scoped_record = record.clone(); scoped_record.shared = existing .as_ref() .and_then(|row| nostr::Event::from_json(&row.raw_event).ok()) .is_some_and(|event| buzz_core_pkg::kind::event_is_shared(&event)); let event = build_persona_event(&scoped_record) .map_err(|e| format!("failed to build event for '{}': {e}", record.display_name))? .custom_created_at(monotonic_created_at( existing.as_ref().map(|row| row.created_at), )) .sign_with_keys(keys) .map_err(|e| format!("failed to sign event for '{}': {e}", record.display_name))?; // Per-coordinate reconcile: skip when an identical body is already // retained, so an unchanged persona doesn't reset `pending_sync`. // Content is timestamp-independent, so the monotonic bump above never // forces a spurious republish. let event_content = event.content.to_string(); if existing .as_ref() .is_some_and(|row| row.content == event_content) { continue; } let retained = RetainedEvent { kind: KIND_PERSONA, pubkey: pubkey.clone(), d_tag, content: event_content, // Safety: nostr timestamps are seconds and stay below i64::MAX // until year 2262. created_at: event.created_at.as_secs() as i64, raw_event: event.as_json(), pending_sync: true, }; // The monotonic bump guarantees `created_at > head`, so the upsert's // `>=` guard always lands the UPDATE — `migrated` counts only real, // retained republishes. retain_event(&conn, &retained) .map_err(|e| format!("failed to retain '{}': {e}", record.display_name))?; migrated += 1; } Ok(migrated) } /// Reconcile `teams.json` into kind:30176 team events in the retention store. /// /// Mirrors [`migrate_personas_to_events`] for teams: it picks up team metadata /// edits (name/description/persona_ids) made on disk between launches and /// queues them for relay publish. Managed agents (kind:30177) are deliberately /// NOT reconciled here — they have no pack/dir source and are backfilled from /// `managed-agents.json` elsewhere. /// /// Must run after the persisted identity is resolved (it signs each event with /// the owner's keys). pub fn migrate_teams_to_events( app: &tauri::AppHandle, keys: &nostr::Keys, db_path: &Path, ) -> Result<(), String> { use crate::managed_agents::managed_agents_base_dir; let base_dir = managed_agents_base_dir(app) .map_err(|e| format!("team-event-migration: base dir unavailable: {e}"))?; match migrate_teams_in_dir_at(&base_dir, keys, db_path) { Ok(0) => Ok(()), Ok(migrated) => { eprintln!("buzz-desktop: team-event-migration: {migrated} teams migrated to retention"); Ok(()) } Err(e) => Err(format!("team-event-migration: {e}")), } } /// Core team reconcile logic, decoupled from the Tauri `AppHandle` for testing. /// /// Returns the number of teams (re)written to the retention store. The /// per-coordinate content compare matches [`migrate_personas_in_dir`]: an /// unchanged team is skipped so a launch does not churn `pending_sync`. #[cfg(test)] fn migrate_teams_in_dir(base_dir: &Path, keys: &nostr::Keys) -> Result { migrate_teams_in_dir_at(base_dir, keys, &base_dir.join("retention.db")) } fn migrate_teams_in_dir_at( base_dir: &Path, keys: &nostr::Keys, db_path: &Path, ) -> Result { use crate::managed_agents::{ persona_events::monotonic_created_at, retention::{get_retained_event, open_retention_db, retain_event, RetainedEvent}, team_events::build_team_event, TeamRecord, }; use buzz_core_pkg::kind::KIND_TEAM; use nostr::JsonUtil; let pubkey = keys.public_key().to_hex(); let teams_path = base_dir.join("teams.json"); if !teams_path.exists() { return Ok(0); } let content = std::fs::read_to_string(&teams_path) .map_err(|e| format!("failed to read teams.json: {e}"))?; let records: Vec = serde_json::from_str(&content).map_err(|e| format!("failed to parse teams.json: {e}"))?; if records.is_empty() { return Ok(0); } let conn = open_retention_db(db_path).map_err(|e| format!("failed to open retention db: {e}"))?; let mut migrated = 0u32; for record in &records { // Skip built-in teams — they're always available from code. if record.is_builtin { continue; } // Team d-tag is the team id (team_events.rs: no slug fallback). let d_tag = record.id.clone(); // Fetch the head first so the monotonic bump can supersede a // future-dated head — see migrate_personas_in_dir (F1/F8). let existing = get_retained_event(&conn, KIND_TEAM, &pubkey, &d_tag)?; let event = build_team_event(record) .map_err(|e| format!("failed to build event for team '{}': {e}", record.name))? .custom_created_at(monotonic_created_at( existing.as_ref().map(|row| row.created_at), )) .sign_with_keys(keys) .map_err(|e| format!("failed to sign event for team '{}': {e}", record.name))?; let event_content = event.content.to_string(); if existing .as_ref() .is_some_and(|row| row.content == event_content) { continue; } let retained = RetainedEvent { kind: KIND_TEAM, pubkey: pubkey.clone(), d_tag, content: event_content, created_at: event.created_at.as_secs() as i64, raw_event: event.as_json(), pending_sync: true, }; // Monotonic bump guarantees the upsert UPDATE lands — `migrated` counts // only real republishes. retain_event(&conn, &retained) .map_err(|e| format!("failed to retain team '{}': {e}", record.name))?; migrated += 1; } Ok(migrated) } #[cfg(test)] #[path = "event_sync_tests.rs"] mod tests; #[cfg(test)] #[path = "event_sync_team_events_tests.rs"] mod team_events_tests;