hashtree_cli/

nostr_mirror.rs

use std::collections::{BTreeMap, HashMap, HashSet};
use std::io::Write;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use std::time::Instant;

use anyhow::{Context, Result};
use hashtree_core::{nhash_encode_full, NHashData};
use hashtree_nostr::{
    CrawlConfig, CrawlReport, ListEventsOptions, NostrBridge, NostrEventStore, RelayFetchMode,
    HASHTREE_ROOT_KIND,
};
use nostr::{
    Alphabet, Event, EventBuilder, Filter, Kind, PublicKey, SingleLetterTag, Tag, TagKind,
    Timestamp,
};
use nostr_sdk::{
    pool::RelayLimits, prelude::RelayPoolNotification, Client, ClientOptions, Keys, RelayStatus,
};
use tokio::sync::{oneshot, watch, Mutex as AsyncMutex};
use tracing::{debug, info, warn};

use crate::blossom_push::{collect_cids_for_push, collect_incremental_cids_for_push};
use crate::config::ensure_keys_string;
use crate::fetch::{FetchConfig, Fetcher};
use crate::socialgraph::crawler::SOCIALGRAPH_RELAY_EVENT_MAX_SIZE;
use crate::socialgraph::{self, SocialGraphBackend, SocialGraphStore};
use crate::HashtreeStore;

#[cfg(not(test))]
const MIRROR_STARTUP_DELAY: Duration = Duration::from_secs(8);
#[cfg(test)]
const MIRROR_STARTUP_DELAY: Duration = Duration::from_millis(50);

#[cfg(not(test))]
const MIRROR_CONNECT_SETTLE_DELAY: Duration = Duration::from_secs(1);
#[cfg(test)]
const MIRROR_CONNECT_SETTLE_DELAY: Duration = Duration::from_millis(250);

#[cfg(not(test))]
const MIRROR_AUTHOR_REFRESH_INTERVAL: Duration = Duration::from_secs(30);
#[cfg(test)]
const MIRROR_AUTHOR_REFRESH_INTERVAL: Duration = Duration::from_millis(100);

#[cfg(not(test))]
const MIRROR_RECONNECT_HISTORY_SYNC_COOLDOWN: Duration = Duration::from_secs(30);
#[cfg(test)]
const MIRROR_RECONNECT_HISTORY_SYNC_COOLDOWN: Duration = Duration::from_millis(100);

const KIND_LONG_FORM_CONTENT: u16 = 30_023;
const KIND_PICTURE_FIRST: u16 = 20;
const KIND_COMMENT: u16 = 1_111;
const FULL_ARCHIVE_HISTORY_KINDS: [u16; 9] = [
    1,
    5,
    6,
    7,
    16,
    KIND_PICTURE_FIRST,
    KIND_COMMENT,
    9_735,
    KIND_LONG_FORM_CONTENT,
];
const LEGACY_TEXT_HISTORY_KINDS: [u16; 2] = [1, KIND_LONG_FORM_CONTENT];
const DEFAULT_HISTORY_KINDS: [u16; 13] = [
    0,
    1,
    3,
    5,
    6,
    7,
    16,
    KIND_PICTURE_FIRST,
    KIND_COMMENT,
    9_735,
    10_000,
    30_000,
    KIND_LONG_FORM_CONTENT,
];
const DEFAULT_EVENT_TREE_NAME: &str = "nostr-event-index";
const DEFAULT_PROFILE_SEARCH_TREE_NAME: &str = "profile-search";
const DEFAULT_PROFILES_BY_PUBKEY_TREE_NAME: &str = "profiles-by-pubkey";
const MIRROR_UPLOAD_STATE_DIR: &str = "nostr-mirror";
const MIRROR_UPLOADED_ROOT_SUFFIX: &str = ".uploaded-root";
const NOSTR_INDEX_STATE_DIR: &str = "nostr-index";
const NOSTR_INDEX_LATEST_ROOT_FILE: &str = "latest-root.txt";
const HISTORY_ROOT_REBASE_MAX_ATTEMPTS: usize = 8;
const METADATA_HISTORY_SYNC_PER_AUTHOR_EVENT_LIMIT: usize = 1;
const METADATA_HISTORY_SYNC_AUTHOR_BATCH_SIZE: usize = 64;
const DEFAULT_FULL_TEXT_NOTE_HISTORY_FOLLOW_DISTANCE: u32 = 2;
const DEFAULT_FULL_TEXT_NOTE_HISTORY_MAX_RELAY_PAGES: usize = 0;
const DEFAULT_ARCHIVE_HISTORY_FOLLOW_DISTANCE: u32 = 2;
const DEFAULT_ARCHIVE_HISTORY_MAX_RELAY_PAGES: usize = 0;
const ARCHIVE_HISTORY_PRIORITY_MAX_DISTANCE: u32 = 1;
const ARCHIVE_HISTORY_PRIORITY_SAMPLE_LIMIT: usize = 32;
const FULL_ARCHIVE_FETCH_TIMEOUT_MAX_MULTIPLIER: usize = 4;

#[cfg(not(test))]
const MIRROR_MISSING_PROFILE_BACKFILL_INTERVAL: Duration = Duration::from_secs(300);
#[cfg(test)]
const MIRROR_MISSING_PROFILE_BACKFILL_INTERVAL: Duration = Duration::from_millis(100);

#[cfg(not(test))]
const MIRROR_ROOT_PUBLISH_DEBOUNCE: Duration = Duration::from_secs(5);
#[cfg(test)]
const MIRROR_ROOT_PUBLISH_DEBOUNCE: Duration = Duration::from_millis(20);

#[cfg(not(test))]
const MIRROR_ROOT_PUBLISH_MAX_STALENESS: Duration = Duration::from_secs(30);
#[cfg(test)]
const MIRROR_ROOT_PUBLISH_MAX_STALENESS: Duration = Duration::from_millis(100);

#[cfg(not(test))]
const MIRROR_ROOT_UPLOAD_RETRY_INTERVAL: Duration = Duration::from_secs(60);
#[cfg(test)]
const MIRROR_ROOT_UPLOAD_RETRY_INTERVAL: Duration = Duration::from_millis(100);

#[cfg(not(test))]
const MIRROR_ROOT_PUBLISH_PRIMARY_TIMEOUT: Duration = Duration::from_secs(2);
#[cfg(test)]
const MIRROR_ROOT_PUBLISH_PRIMARY_TIMEOUT: Duration = Duration::from_millis(250);

#[cfg(not(test))]
const MIRROR_ROOT_PUBLISH_RETRY_TIMEOUT: Duration = Duration::from_secs(5);
#[cfg(test)]
const MIRROR_ROOT_PUBLISH_RETRY_TIMEOUT: Duration = Duration::from_secs(2);

#[cfg(not(test))]
const MIRROR_ROOT_UPLOAD_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
#[cfg(test)]
const MIRROR_ROOT_UPLOAD_REQUEST_TIMEOUT: Duration = Duration::from_secs(2);
const MIRROR_ROOT_UPLOAD_CONCURRENCY: usize = 16;

const MISSING_LOCAL_BLOB_PUSH_ERROR: &str = "missing local blob";

fn trim_transient_allocations() {
    #[cfg(all(target_os = "linux", target_env = "gnu"))]
    unsafe {
        // SAFETY: malloc_trim asks glibc to return free heap pages to the OS.
        // It does not touch live allocations and is safe to call after the
        // large, synchronous mirror upload job has dropped its temporary data.
        libc::malloc_trim(0);
    }
}

struct RootPublicationDeferral<'a> {
    count: &'a AtomicUsize,
}

impl<'a> RootPublicationDeferral<'a> {
    fn new(count: &'a AtomicUsize) -> Self {
        count.fetch_add(1, Ordering::AcqRel);
        Self { count }
    }
}

impl Drop for RootPublicationDeferral<'_> {
    fn drop(&mut self) {
        self.count.fetch_sub(1, Ordering::AcqRel);
    }
}

fn decode_hex_pubkey(value: &str) -> Option<[u8; 32]> {
    let bytes = hex::decode(value).ok()?;
    <[u8; 32]>::try_from(bytes.as_slice()).ok()
}

#[derive(Debug, Clone)]
pub struct NostrMirrorConfig {
    pub relays: Vec<String>,
    pub publish_relays: Vec<String>,
    pub blossom_write_servers: Vec<String>,
    pub max_follow_distance: u32,
    pub overmute_threshold: f64,
    pub author_batch_size: usize,
    pub history_sync_author_chunk_size: usize,
    pub history_sync_per_author_event_limit: usize,
    pub missing_profile_backfill_batch_size: usize,
    pub fetch_timeout: Duration,
    pub relay_event_max_size: Option<u32>,
    pub require_negentropy: bool,
    pub kinds: Vec<u16>,
    pub history_sync_on_start: bool,
    pub history_sync_on_reconnect: bool,
    pub full_text_note_history_follow_distance: Option<u32>,
    pub full_text_note_history_max_relay_pages: usize,
    pub archive_history_follow_distance: Option<u32>,
    pub archive_history_max_relay_pages: usize,
    pub published_event_tree_name: Option<String>,
    pub published_profile_search_tree_name: Option<String>,
    pub published_profiles_by_pubkey_tree_name: Option<String>,
}

impl Default for NostrMirrorConfig {
    fn default() -> Self {
        Self {
            relays: Vec::new(),
            publish_relays: Vec::new(),
            blossom_write_servers: Vec::new(),
            max_follow_distance: 2,
            overmute_threshold: 1.0,
            author_batch_size: 256,
            history_sync_author_chunk_size: 5_000,
            history_sync_per_author_event_limit: 256,
            missing_profile_backfill_batch_size: 5_000,
            fetch_timeout: Duration::from_secs(15),
            relay_event_max_size: Some(SOCIALGRAPH_RELAY_EVENT_MAX_SIZE),
            require_negentropy: false,
            kinds: DEFAULT_HISTORY_KINDS.to_vec(),
            history_sync_on_start: true,
            history_sync_on_reconnect: true,
            full_text_note_history_follow_distance: Some(
                DEFAULT_FULL_TEXT_NOTE_HISTORY_FOLLOW_DISTANCE,
            ),
            full_text_note_history_max_relay_pages: DEFAULT_FULL_TEXT_NOTE_HISTORY_MAX_RELAY_PAGES,
            archive_history_follow_distance: Some(DEFAULT_ARCHIVE_HISTORY_FOLLOW_DISTANCE),
            archive_history_max_relay_pages: DEFAULT_ARCHIVE_HISTORY_MAX_RELAY_PAGES,
            published_event_tree_name: Some(DEFAULT_EVENT_TREE_NAME.to_string()),
            published_profile_search_tree_name: Some(DEFAULT_PROFILE_SEARCH_TREE_NAME.to_string()),
            published_profiles_by_pubkey_tree_name: Some(
                DEFAULT_PROFILES_BY_PUBKEY_TREE_NAME.to_string(),
            ),
        }
    }
}

#[derive(Debug, Default)]
struct RootPublishState {
    pending_root: Option<hashtree_core::Cid>,
    last_changed_at: Option<Instant>,
    dirty_since: Option<Instant>,
    last_published_root: Option<hashtree_core::Cid>,
    last_published_at: Option<Instant>,
    last_published_created_at: Option<Timestamp>,
    last_uploaded_root: Option<hashtree_core::Cid>,
    last_uploaded_at: Option<Instant>,
    upload_in_progress_root: Option<hashtree_core::Cid>,
    last_upload_failed_at: Option<Instant>,
    last_upload_error: Option<String>,
    missing_blob_rebuild_required: bool,
}

struct RootUploadTask {
    cancel: watch::Sender<bool>,
    finished: oneshot::Receiver<()>,
    join: std::thread::JoinHandle<()>,
    drain_after_start: bool,
}

struct RootUploadJob {
    store: Arc<HashtreeStore>,
    root: hashtree_core::Cid,
    previous_root: Option<hashtree_core::Cid>,
    servers: Vec<String>,
    data_dir: PathBuf,
    profile_publication: bool,
    log_label: String,
}

struct RootPublishTask {
    finished: oneshot::Receiver<()>,
    join: std::thread::JoinHandle<()>,
}

struct MirrorBackgroundTask {
    label: &'static str,
    finished: oneshot::Receiver<()>,
    join: std::thread::JoinHandle<()>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct HistorySyncPlan {
    relay_fetch_mode: RelayFetchMode,
    author_batch_size: usize,
    per_author_event_limit: usize,
    relay_page_size: usize,
    max_relay_pages: usize,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct ArchiveHistorySettings {
    follow_distance: u32,
    max_relay_pages: usize,
    kinds: Vec<u16>,
}

pub struct BackgroundNostrMirror {
    config: NostrMirrorConfig,
    store: Arc<HashtreeStore>,
    graph_store: Arc<SocialGraphStore>,
    client: Client,
    publish_client: Option<Client>,
    event_publish_state: Arc<Mutex<RootPublishState>>,
    profile_search_publish_state: Arc<Mutex<RootPublishState>>,
    profiles_by_pubkey_publish_state: Arc<Mutex<RootPublishState>>,
    pending_live_events: Mutex<BTreeMap<String, Event>>,
    missing_profile_cursor: Mutex<usize>,
    history_sync_lock: AsyncMutex<()>,
    root_publication_deferrals: AtomicUsize,
    profile_publication_fence_logged: AtomicBool,
    root_upload_tasks: Mutex<Vec<RootUploadTask>>,
    root_publish_task: Mutex<Option<RootPublishTask>>,
    background_tasks: Mutex<Vec<MirrorBackgroundTask>>,
    shutting_down: AtomicBool,
    shutdown_tx: watch::Sender<bool>,
    shutdown_rx: watch::Receiver<bool>,
}

impl BackgroundNostrMirror {
    pub async fn new(
        config: NostrMirrorConfig,
        store: Arc<HashtreeStore>,
        graph_store: Arc<SocialGraphStore>,
        publish_keys: Option<Keys>,
    ) -> Result<Self> {
        let client = if let Some(max_size) = config.relay_event_max_size {
            let mut limits = RelayLimits::default();
            limits.events.max_size = Some(max_size);
            Client::builder()
                .signer(Keys::generate())
                .opts(ClientOptions::new().relay_limits(limits))
                .build()
        } else {
            Client::new(Keys::generate())
        };
        for relay in &config.relays {
            client
                .add_relay(relay)
                .await
                .with_context(|| format!("add mirror relay {relay}"))?;
        }
        client.connect().await;

        let publish_client = if let Some(keys) = publish_keys {
            if config.publish_relays.is_empty() {
                None
            } else {
                let client = Client::new(keys);
                for relay in &config.publish_relays {
                    client
                        .add_relay(relay)
                        .await
                        .with_context(|| format!("add mirror publish relay {relay}"))?;
                }
                client.connect().await;
                Some(client)
            }
        } else {
            None
        };

        let (shutdown_tx, shutdown_rx) = watch::channel(false);
        let event_publish_state = Self::root_publish_state_from_disk(
            store.base_path(),
            config.published_event_tree_name.as_deref(),
            "event root",
        );
        let profile_search_publish_state = Self::root_publish_state_from_disk(
            store.base_path(),
            config.published_profile_search_tree_name.as_deref(),
            "profile-search root",
        );
        let profiles_by_pubkey_publish_state = Self::root_publish_state_from_disk(
            store.base_path(),
            config.published_profiles_by_pubkey_tree_name.as_deref(),
            "profiles-by-pubkey root",
        );
        Ok(Self {
            config,
            store,
            graph_store,
            client,
            publish_client,
            event_publish_state: Arc::new(Mutex::new(event_publish_state)),
            profile_search_publish_state: Arc::new(Mutex::new(profile_search_publish_state)),
            profiles_by_pubkey_publish_state: Arc::new(Mutex::new(
                profiles_by_pubkey_publish_state,
            )),
            pending_live_events: Mutex::new(BTreeMap::new()),
            missing_profile_cursor: Mutex::new(0),
            history_sync_lock: AsyncMutex::new(()),
            root_publication_deferrals: AtomicUsize::new(0),
            profile_publication_fence_logged: AtomicBool::new(false),
            root_upload_tasks: Mutex::new(Vec::new()),
            root_publish_task: Mutex::new(None),
            background_tasks: Mutex::new(Vec::new()),
            shutting_down: AtomicBool::new(false),
            shutdown_tx,
            shutdown_rx,
        })
    }

    fn root_publish_state_from_disk(
        base_path: &std::path::Path,
        tree_name: Option<&str>,
        log_label: &str,
    ) -> RootPublishState {
        let Some(tree_name) = tree_name else {
            return RootPublishState::default();
        };
        let path = Self::uploaded_root_state_path(base_path, tree_name);
        let Some(root) = Self::read_uploaded_root_state(&path, log_label) else {
            return RootPublishState::default();
        };

        RootPublishState {
            last_uploaded_root: Some(root),
            ..RootPublishState::default()
        }
    }

    fn uploaded_root_state_path(base_path: &std::path::Path, tree_name: &str) -> PathBuf {
        let safe_tree_name = tree_name
            .chars()
            .map(|ch| {
                if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
                    ch
                } else {
                    '_'
                }
            })
            .collect::<String>();
        base_path
            .join(MIRROR_UPLOAD_STATE_DIR)
            .join(format!("{safe_tree_name}{MIRROR_UPLOADED_ROOT_SUFFIX}"))
    }

    fn read_uploaded_root_state(
        path: &std::path::Path,
        log_label: &str,
    ) -> Option<hashtree_core::Cid> {
        let raw = match std::fs::read_to_string(path) {
            Ok(raw) => raw,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return None,
            Err(err) => {
                warn!(
                    "Nostr mirror failed to read uploaded {} state {}: {}",
                    log_label,
                    path.display(),
                    err
                );
                return None;
            }
        };
        let trimmed = raw.trim();
        if trimmed.is_empty() {
            return None;
        }
        match hashtree_core::Cid::parse(trimmed) {
            Ok(root) => Some(root),
            Err(err) => {
                warn!(
                    "Nostr mirror ignored invalid uploaded {} state {}: {}",
                    log_label,
                    path.display(),
                    err
                );
                None
            }
        }
    }

    fn write_uploaded_root_state(
        path: &std::path::Path,
        root: &hashtree_core::Cid,
        log_label: &str,
    ) -> Result<()> {
        let parent = path
            .parent()
            .context("uploaded root state path has no parent directory")?;
        let parent_was_present = parent.exists();
        std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
        let file_name = path
            .file_name()
            .and_then(|name| name.to_str())
            .context("uploaded root state path has no UTF-8 file name")?;
        let mut temporary = tempfile::Builder::new()
            .prefix(&format!(".{file_name}."))
            .suffix(".tmp")
            .tempfile_in(parent)
            .with_context(|| format!("create temporary uploaded {log_label} state"))?;
        temporary
            .write_all(format!("{root}\n").as_bytes())
            .with_context(|| format!("write temporary uploaded {log_label} state"))?;
        temporary
            .as_file()
            .sync_all()
            .with_context(|| format!("fsync temporary uploaded {log_label} state"))?;
        temporary
            .persist(path)
            .map_err(|error| error.error)
            .with_context(|| format!("replace uploaded {log_label} state {}", path.display()))?;
        #[cfg(unix)]
        {
            std::fs::File::open(parent)
                .with_context(|| format!("open {} for fsync", parent.display()))?
                .sync_all()
                .with_context(|| format!("fsync {}", parent.display()))?;
            if !parent_was_present {
                if let Some(grandparent) = parent.parent() {
                    std::fs::File::open(grandparent)
                        .with_context(|| format!("open {} for fsync", grandparent.display()))?
                        .sync_all()
                        .with_context(|| format!("fsync {}", grandparent.display()))?;
                }
            }
        }
        Ok(())
    }

    fn write_latest_event_root_state(
        base_path: &std::path::Path,
        root: Option<&hashtree_core::Cid>,
    ) -> Result<()> {
        let path = base_path
            .join(NOSTR_INDEX_STATE_DIR)
            .join(NOSTR_INDEX_LATEST_ROOT_FILE);
        let Some(root) = root else {
            if path.exists() {
                std::fs::remove_file(&path)
                    .with_context(|| format!("remove {}", path.display()))?;
            }
            return Ok(());
        };
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)
                .with_context(|| format!("create {}", parent.display()))?;
        }
        let encoded = nhash_encode_full(&NHashData {
            hash: root.hash,
            decrypt_key: root.key,
        })
        .context("encode mirrored Nostr index root")?;
        let tmp_path = path.with_extension("tmp");
        std::fs::write(&tmp_path, format!("{encoded}\n"))
            .with_context(|| format!("write {}", tmp_path.display()))?;
        #[cfg(windows)]
        if path.exists() {
            std::fs::remove_file(&path)
                .with_context(|| format!("remove stale {}", path.display()))?;
        }
        std::fs::rename(&tmp_path, &path).with_context(|| format!("replace {}", path.display()))?;
        Ok(())
    }

    pub fn shutdown(&self) {
        self.shutting_down.store(true, Ordering::Release);
        // Synchronize with both task-registration critical sections. Once
        // shutdown returns, no publisher can still be between its final
        // shutdown check and registering work that cleanup would miss.
        {
            let _publication = self.root_publish_task.lock().expect("root publish task");
        }
        {
            let _background = self
                .background_tasks
                .lock()
                .expect("mirror background tasks");
        }
        let tasks = self.root_upload_tasks.lock().expect("root upload tasks");
        for task in tasks.iter() {
            let _ = task.cancel.send(true);
        }
        drop(tasks);
        let _ = self.shutdown_tx.send(true);
    }

    async fn finish_root_upload_tasks(&self) {
        let tasks = {
            let mut tasks = self.root_upload_tasks.lock().expect("root upload tasks");
            tasks.drain(..).collect::<Vec<_>>()
        };
        for task in &tasks {
            let _ = task.cancel.send(true);
        }
        for task in tasks {
            let RootUploadTask {
                finished,
                join,
                drain_after_start,
                ..
            } = task;
            let completion = finished.await;
            match join.join() {
                Ok(()) if completion.is_ok() => {}
                Ok(()) => warn!(
                    "Nostr mirror root upload completion channel closed unexpectedly: guarded={drain_after_start}"
                ),
                Err(_) => warn!(
                    "Nostr mirror root upload thread panicked during shutdown: guarded={drain_after_start}"
                ),
            }
        }
    }

    async fn finish_background_tasks(&self) {
        let tasks = {
            let mut tasks = self
                .background_tasks
                .lock()
                .expect("mirror background tasks");
            tasks.drain(..).collect::<Vec<_>>()
        };
        for task in tasks {
            let MirrorBackgroundTask {
                label,
                finished,
                join,
            } = task;
            let completion = finished.await;
            match join.join() {
                Ok(()) if completion.is_ok() => {}
                Ok(()) => warn!("Nostr mirror {label} completion channel closed unexpectedly"),
                Err(_) => warn!("Nostr mirror {label} thread panicked during shutdown"),
            }
        }
    }

    fn sync_publish_roots_from_store(&self) -> Result<()> {
        self.note_public_events_root_change()?;
        self.note_profile_search_root_change()?;
        self.note_profiles_by_pubkey_root_change()?;
        Ok(())
    }

    async fn publish_pending_roots(
        &self,
        force_event: bool,
        force_profile_search: bool,
        force_profiles_by_pubkey: bool,
    ) -> (Result<()>, Result<()>, Result<()>) {
        tokio::join!(
            self.maybe_publish_event_root(force_event),
            self.maybe_publish_profile_search_root(force_profile_search),
            self.maybe_publish_profiles_by_pubkey_root(force_profiles_by_pubkey),
        )
    }

    fn spawn_pending_root_publish(
        self: &Arc<Self>,
        force_event: bool,
        force_profile_search: bool,
        force_profiles_by_pubkey: bool,
        priority: bool,
    ) {
        let mut task = self.root_publish_task.lock().expect("root publish task");
        if self.shutting_down.load(Ordering::Acquire) {
            return;
        }
        if let Some(previous) = task.take() {
            if !previous.join.is_finished() {
                *task = Some(previous);
                return;
            }
            if previous.join.join().is_err() {
                warn!("Nostr mirror root publication thread panicked");
            }
        }
        let mirror = Arc::clone(self);
        let (finished_tx, finished) = oneshot::channel();
        let join = std::thread::spawn(move || {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("build Nostr root publication runtime");
            runtime.block_on(async move {
                let (event_result, profile_search_result, profiles_by_pubkey_result) = if priority {
                    mirror
                        .publish_priority_roots(
                            force_event,
                            force_profile_search,
                            force_profiles_by_pubkey,
                        )
                        .await
                } else {
                    mirror
                        .publish_pending_roots(
                            force_event,
                            force_profile_search,
                            force_profiles_by_pubkey,
                        )
                        .await
                };
                if let Err(error) = event_result {
                    warn!("Nostr mirror event-root publish failed: {error:#}");
                }
                if let Err(error) = profile_search_result {
                    warn!("Nostr mirror profile-search publish failed: {error:#}");
                }
                if let Err(error) = profiles_by_pubkey_result {
                    warn!("Nostr mirror profiles-by-pubkey publish failed: {error:#}");
                }
            });
            let _ = finished_tx.send(());
        });
        *task = Some(RootPublishTask { finished, join });
    }

    async fn finish_pending_root_publish_task(&self) {
        let task = self
            .root_publish_task
            .lock()
            .expect("root publish task")
            .take();
        if let Some(task) = task {
            let RootPublishTask { finished, join } = task;
            let completion = finished.await;
            match join.join() {
                Ok(()) if completion.is_ok() => {}
                Ok(()) => {
                    warn!("Nostr mirror root publication completion channel closed unexpectedly")
                }
                Err(_) => warn!("Nostr mirror root publication thread panicked"),
            }
        }
    }

    async fn publish_priority_roots(
        &self,
        force_event: bool,
        force_profile_search: bool,
        force_profiles_by_pubkey: bool,
    ) -> (Result<()>, Result<()>, Result<()>) {
        let (profile_search_result, profiles_by_pubkey_result) = tokio::join!(
            async {
                if force_profile_search {
                    self.maybe_publish_profile_search_root(true).await
                } else {
                    Ok(())
                }
            },
            async {
                if force_profiles_by_pubkey {
                    self.maybe_publish_profiles_by_pubkey_root(true).await
                } else {
                    Ok(())
                }
            },
        );
        let event_result = if force_event {
            self.maybe_publish_event_root(true).await
        } else {
            Ok(())
        };
        (
            event_result,
            profile_search_result,
            profiles_by_pubkey_result,
        )
    }

    pub async fn run(self: Arc<Self>) -> Result<()> {
        if self.config.relays.is_empty() || self.config.max_follow_distance == 0 {
            return Ok(());
        }

        let result = self.run_active().await;
        self.finish_run(result.is_ok()).await;
        result
    }

    async fn run_active(self: &Arc<Self>) -> Result<()> {
        info!(
            "Nostr mirror starting: relays={} max_follow_distance={} negentropy_only={} kinds={:?} history_sync_author_chunk_size={} history_sync_on_start={} history_sync_on_reconnect={}",
            self.config.relays.len(),
            self.config.max_follow_distance,
            self.config.require_negentropy,
            self.config.kinds,
            self.config.history_sync_author_chunk_size.max(1),
            self.config.history_sync_on_start,
            self.config.history_sync_on_reconnect
        );

        tokio::time::sleep(MIRROR_STARTUP_DELAY).await;
        tokio::time::sleep(MIRROR_CONNECT_SETTLE_DELAY).await;
        let live_since = Timestamp::now();
        self.sync_publish_roots_from_store()?;
        self.spawn_pending_root_publish(true, true, true, true);

        let initial_authors = self.collect_authors()?;
        if initial_authors.is_empty() {
            info!("Nostr mirror: no social-graph authors to mirror yet");
        }

        let mut subscribed_authors = HashSet::new();
        self.subscribe_authors_since(&initial_authors, live_since, &mut subscribed_authors)
            .await?;

        if !initial_authors.is_empty() && self.config.history_sync_on_start {
            self.spawn_startup_history_sync(initial_authors.clone());
        }

        let mut relay_statuses = self.capture_relay_statuses().await;
        let mut last_reconnect_history_sync_at: Option<Instant> = None;
        let mut last_missing_profile_backfill_at: Option<Instant> = None;
        let mut notifications = self.client.notifications();
        let mut shutdown_rx = self.shutdown_rx.clone();
        let mut refresh_interval = tokio::time::interval(MIRROR_AUTHOR_REFRESH_INTERVAL);
        let mut publish_interval = tokio::time::interval(MIRROR_ROOT_PUBLISH_DEBOUNCE);

        loop {
            tokio::select! {
                _ = shutdown_rx.changed() => {
                    if *shutdown_rx.borrow() {
                        break;
                    }
                }
                _ = refresh_interval.tick() => {
                    let authors = self.collect_authors()?;
                    let mut reconnected_relay = None;
                    for (relay_url, status) in self.capture_relay_statuses().await {
                        let previous = relay_statuses.insert(relay_url.clone(), status);
                        if reconnected_relay.is_none()
                            && Self::should_history_sync_on_reconnect(
                                self.config.history_sync_on_reconnect,
                                previous,
                                status,
                            )
                        {
                            reconnected_relay = Some(relay_url);
                        }
                    }
                    if let Some(relay_url) = reconnected_relay {
                        if Self::should_run_reconnect_history_sync(
                            last_reconnect_history_sync_at.as_ref(),
                        ) && !authors.is_empty()
                        {
                            info!(
                                "Nostr mirror relay reconnected; running catch-up history sync: relay={} authors={} negentropy_only={}",
                                relay_url,
                                authors.len(),
                                self.config.require_negentropy
                            );
                            self.spawn_author_history_sync(
                                "relay reconnect catch-up",
                                authors.clone(),
                                false,
                                false,
                            );
                            last_reconnect_history_sync_at = Some(Instant::now());
                        }
                    }
                    let new_authors = authors
                        .iter()
                        .filter(|author| !subscribed_authors.contains(*author))
                        .cloned()
                        .collect::<Vec<_>>();
                    if !new_authors.is_empty() {
                        debug!(
                            "Nostr mirror discovered {} newly reachable author(s)",
                            new_authors.len()
                        );
                        self.subscribe_authors_since(
                            &new_authors,
                            Timestamp::now(),
                            &mut subscribed_authors,
                        )
                        .await?;
                        self.spawn_author_history_sync(
                            "new-author catch-up",
                            new_authors.clone(),
                            true,
                            true,
                        );
                    }
                    if self.should_backfill_missing_profiles(last_missing_profile_backfill_at) {
                        let missing_profile_authors = self.collect_missing_profile_authors(
                            self.config.missing_profile_backfill_batch_size,
                        )?;
                        if !missing_profile_authors.is_empty() {
                            info!(
                                "Nostr mirror missing-profile backfill starting: authors={}",
                                missing_profile_authors.len()
                            );
                            self.spawn_missing_profile_backfill(missing_profile_authors);
                            last_missing_profile_backfill_at = Some(Instant::now());
                        }
                    }
                }
                _ = publish_interval.tick() => {
                    self.sync_publish_roots_from_store()?;
                    if let Err(err) = self.flush_live_events().await {
                        warn!("Nostr mirror live event flush failed: {:#}", err);
                    }
                    self.spawn_pending_root_publish(false, false, false, false);
                }
                notification = notifications.recv() => {
                    match notification {
                        Ok(RelayPoolNotification::Event { event, .. }) => {
                            self.ingest_live_event(&event)?;
                        }
                        Ok(RelayPoolNotification::Shutdown) => break,
                        Ok(_) => {}
                        Err(err) => {
                            warn!("Nostr mirror notification error: {}", err);
                            break;
                        }
                    }
                }
            }
        }

        Ok(())
    }

    async fn finish_run(&self, graceful: bool) {
        // This is intentionally the first cleanup action. It closes both task
        // registration races and cancels workers that have not entered a
        // guarded external publication yet.
        self.shutdown();
        self.finish_pending_root_publish_task().await;
        self.finish_background_tasks().await;

        if graceful {
            if let Err(err) = self.flush_live_events().await {
                warn!(
                    "Nostr mirror live event flush failed during shutdown: {:#}",
                    err
                );
            }
            if let Err(err) = self.sync_publish_roots_from_store() {
                warn!(
                    "Nostr mirror root-state refresh failed during shutdown: {:#}",
                    err
                );
            }
            let (event_result, profile_search_result, profiles_by_pubkey_result) =
                self.publish_pending_roots(true, true, true).await;
            if let Err(err) = event_result {
                warn!(
                    "Nostr mirror event-root publish failed during shutdown: {:#}",
                    err
                );
            }
            if let Err(err) = profile_search_result {
                warn!(
                    "Nostr mirror profile-search publish failed during shutdown: {:#}",
                    err
                );
            }
            if let Err(err) = profiles_by_pubkey_result {
                warn!(
                    "Nostr mirror profiles-by-pubkey publish failed during shutdown: {:#}",
                    err
                );
            }
        }
        self.finish_root_upload_tasks().await;
        let _ = self.client.disconnect().await;
        if let Some(client) = self.publish_client.as_ref() {
            let _ = client.disconnect().await;
        }
    }

    fn spawn_background_task(&self, label: &'static str, task: impl FnOnce() + Send + 'static) {
        let mut tasks = self
            .background_tasks
            .lock()
            .expect("mirror background tasks");
        if self.shutting_down.load(Ordering::Acquire) {
            return;
        }
        tasks.retain(|task| !task.join.is_finished());
        let (finished_tx, finished) = oneshot::channel();
        let join = std::thread::spawn(move || {
            task();
            let _ = finished_tx.send(());
        });
        tasks.push(MirrorBackgroundTask {
            label,
            finished,
            join,
        });
    }

    fn spawn_startup_history_sync(self: &Arc<Self>, initial_authors: Vec<String>) {
        let mirror = Arc::clone(self);
        self.spawn_background_task("startup history sync", move || {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("build nostr mirror startup history sync runtime");
            runtime.block_on(async move {
                let mut shutdown_rx = mirror.shutdown_rx.clone();
                let _guard = tokio::select! {
                    guard = mirror.history_sync_lock.lock() => guard,
                    _ = shutdown_rx.changed() => return,
                };
                if mirror.shutting_down.load(Ordering::Acquire) {
                    return;
                }
                if let Err(err) = mirror.run_startup_history_sync(initial_authors).await {
                    warn!("Nostr mirror startup history sync failed: {:#}", err);
                }
            });
        });
    }

    async fn run_startup_history_sync(&self, initial_authors: Vec<String>) -> Result<()> {
        self.history_sync_authors(initial_authors).await?;
        if self.shutting_down.load(Ordering::Acquire) {
            return Ok(());
        }
        self.history_sync_archive_for_reachable_authors().await?;
        if self.shutting_down.load(Ordering::Acquire) {
            return Ok(());
        }
        if self.should_backfill_missing_profiles(None) {
            let missing_profile_authors = self
                .collect_missing_profile_authors(self.config.missing_profile_backfill_batch_size)?;
            if !missing_profile_authors.is_empty() {
                info!(
                    "Nostr mirror missing-profile backfill starting: authors={}",
                    missing_profile_authors.len()
                );
                self.history_sync_authors_with_kinds(
                    missing_profile_authors,
                    &[Kind::Metadata.as_u16()],
                )
                .await?;
            }
        }
        Ok(())
    }

    fn spawn_author_history_sync(
        self: &Arc<Self>,
        label: &'static str,
        authors: Vec<String>,
        include_archive_history: bool,
        wait_for_existing_sync: bool,
    ) {
        let mirror = Arc::clone(self);
        self.spawn_background_task(label, move || {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("build nostr mirror author history sync runtime");
            runtime.block_on(async move {
                if wait_for_existing_sync {
                    let mut shutdown_rx = mirror.shutdown_rx.clone();
                    let _guard = tokio::select! {
                        guard = mirror.history_sync_lock.lock() => guard,
                        _ = shutdown_rx.changed() => return,
                    };
                    if mirror.shutting_down.load(Ordering::Acquire) {
                        return;
                    }
                    if let Err(err) = mirror
                        .run_author_history_sync(authors, include_archive_history)
                        .await
                    {
                        warn!("Nostr mirror {label} failed: {:#}", err);
                    }
                    return;
                }

                if mirror.shutting_down.load(Ordering::Acquire) {
                    return;
                }
                let Ok(_guard) = mirror.history_sync_lock.try_lock() else {
                    info!("Nostr mirror {label} skipped; another history sync is running");
                    return;
                };
                if let Err(err) = mirror
                    .run_author_history_sync(authors, include_archive_history)
                    .await
                {
                    warn!("Nostr mirror {label} failed: {:#}", err);
                }
            });
        });
    }

    async fn run_author_history_sync(
        &self,
        authors: Vec<String>,
        include_archive_history: bool,
    ) -> Result<()> {
        self.history_sync_authors(authors.clone()).await?;
        if include_archive_history && !self.shutting_down.load(Ordering::Acquire) {
            self.history_sync_archive_for_authors(authors).await?;
        }
        Ok(())
    }

    fn spawn_missing_profile_backfill(self: &Arc<Self>, authors: Vec<String>) {
        let mirror = Arc::clone(self);
        self.spawn_background_task("missing-profile backfill", move || {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("build nostr mirror missing profile runtime");
            runtime.block_on(async move {
                if mirror.shutting_down.load(Ordering::Acquire) {
                    return;
                }
                let Ok(_guard) = mirror.history_sync_lock.try_lock() else {
                    info!(
                        "Nostr mirror missing-profile backfill skipped; another history sync is running"
                    );
                    return;
                };
                if let Err(err) = mirror
                    .history_sync_authors_with_kinds(authors, &[Kind::Metadata.as_u16()])
                    .await
                {
                    warn!("Nostr mirror missing-profile backfill failed: {:#}", err);
                }
            });
        });
    }

    async fn capture_relay_statuses(&self) -> HashMap<String, RelayStatus> {
        let mut statuses = HashMap::new();
        for (relay_url, relay) in self.client.relays().await {
            statuses.insert(relay_url.to_string(), relay.status());
        }
        statuses
    }

    async fn has_connected_publish_relay(&self) -> bool {
        let Some(client) = self.publish_client.as_ref() else {
            return false;
        };
        Self::client_has_connected_relay(client).await
    }

    async fn client_has_connected_relay(client: &Client) -> bool {
        for (_relay_url, relay) in client.relays().await {
            if relay.status() == RelayStatus::Connected {
                return true;
            }
        }
        false
    }

    fn collect_authors(&self) -> Result<Vec<String>> {
        self.collect_authors_with_max_distance(self.config.max_follow_distance)
    }

    fn collect_authors_with_max_distance(&self, max_distance: u32) -> Result<Vec<String>> {
        let mut authors = Vec::new();
        let mut seen = HashSet::new();
        for distance in 0..=max_distance {
            for pubkey in socialgraph::SocialGraphBackend::users_by_follow_distance(
                self.graph_store.as_ref(),
                distance,
            )
            .with_context(|| format!("load social-graph distance {distance}"))?
            {
                if self
                    .graph_store
                    .is_overmuted_user(&pubkey, self.config.overmute_threshold)?
                {
                    continue;
                }
                let hex = hex::encode(pubkey);
                if seen.insert(hex.clone()) {
                    authors.push(hex);
                }
            }
        }
        Ok(authors)
    }

    async fn prioritize_archive_history_authors(
        &self,
        authors: Vec<String>,
    ) -> Result<Vec<String>> {
        let Some(root) = self.graph_store.public_events_root()? else {
            return Ok(authors);
        };

        let event_store = NostrEventStore::new(self.store.store_arc());
        let mut prioritized = Vec::with_capacity(authors.len());
        let mut sampled = 0usize;
        for (index, author) in authors.into_iter().enumerate() {
            let distance = match decode_hex_pubkey(&author) {
                Some(pubkey) => self
                    .graph_store
                    .follow_distance(&pubkey)?
                    .unwrap_or(u32::MAX),
                None => u32::MAX,
            };
            let indexed_text_sample = if distance <= ARCHIVE_HISTORY_PRIORITY_MAX_DISTANCE {
                sampled = sampled.saturating_add(1);
                event_store
                    .list_by_author_and_kind(
                        Some(&root),
                        &author,
                        Kind::TextNote.as_u16() as u32,
                        ListEventsOptions {
                            limit: Some(ARCHIVE_HISTORY_PRIORITY_SAMPLE_LIMIT),
                            ..ListEventsOptions::default()
                        },
                    )
                    .await
                    .with_context(|| {
                        format!("sample indexed text-note history for author {author}")
                    })?
                    .len()
            } else {
                ARCHIVE_HISTORY_PRIORITY_SAMPLE_LIMIT
            };
            prioritized.push((distance, indexed_text_sample, index, author));
        }

        prioritized.sort_by(|left, right| {
            left.0
                .cmp(&right.0)
                .then_with(|| left.1.cmp(&right.1))
                .then_with(|| left.2.cmp(&right.2))
        });
        info!(
            "Nostr mirror configured archive history prioritized authors: authors={} sampled_distance_le_{}={}",
            prioritized.len(),
            ARCHIVE_HISTORY_PRIORITY_MAX_DISTANCE,
            sampled
        );
        Ok(prioritized
            .into_iter()
            .map(|(_, _, _, author)| author)
            .collect())
    }

    fn full_archive_history_kinds_for_config(config: &NostrMirrorConfig) -> Vec<u16> {
        FULL_ARCHIVE_HISTORY_KINDS
            .into_iter()
            .filter(|kind| config.kinds.contains(kind))
            .collect()
    }

    fn legacy_text_history_kinds_for_config(config: &NostrMirrorConfig) -> Vec<u16> {
        LEGACY_TEXT_HISTORY_KINDS
            .into_iter()
            .filter(|kind| config.kinds.contains(kind))
            .collect()
    }

    fn archive_history_settings_for_config(
        config: &NostrMirrorConfig,
    ) -> Option<ArchiveHistorySettings> {
        let (follow_distance, max_relay_pages, kinds) =
            if config.archive_history_max_relay_pages > 0 {
                (
                    config.archive_history_follow_distance?,
                    config.archive_history_max_relay_pages,
                    Self::full_archive_history_kinds_for_config(config),
                )
            } else {
                (
                    config.full_text_note_history_follow_distance?,
                    config.full_text_note_history_max_relay_pages,
                    Self::legacy_text_history_kinds_for_config(config),
                )
            };
        if max_relay_pages == 0 || kinds.is_empty() {
            return None;
        }
        Some(ArchiveHistorySettings {
            follow_distance: follow_distance.min(config.max_follow_distance),
            max_relay_pages,
            kinds,
        })
    }

    fn archive_history_settings(&self) -> Option<ArchiveHistorySettings> {
        Self::archive_history_settings_for_config(&self.config)
    }

    fn history_sync_kinds_for_config(config: &NostrMirrorConfig) -> Vec<u16> {
        config.kinds.clone()
    }

    fn collect_missing_profile_authors(&self, limit: usize) -> Result<Vec<String>> {
        if limit == 0 {
            return Ok(Vec::new());
        }

        let authors = self.collect_authors()?;
        if authors.is_empty() {
            return Ok(Vec::new());
        }

        let mut cursor = self
            .missing_profile_cursor
            .lock()
            .expect("missing profile cursor");
        let mut index = (*cursor).min(authors.len());
        let mut scanned = 0usize;
        let mut missing = Vec::new();

        while scanned < authors.len() && missing.len() < limit {
            let author = &authors[index];
            if self.graph_store.latest_profile_event(author)?.is_none() {
                missing.push(author.clone());
            }
            index += 1;
            if index == authors.len() {
                index = 0;
            }
            scanned += 1;
        }

        *cursor = index;
        Ok(missing)
    }

    fn should_backfill_missing_profiles(&self, last_run: Option<Instant>) -> bool {
        if self.config.missing_profile_backfill_batch_size == 0
            || !self.config.kinds.contains(&Kind::Metadata.as_u16())
        {
            return false;
        }
        match last_run {
            Some(last_run) => last_run.elapsed() >= MIRROR_MISSING_PROFILE_BACKFILL_INTERVAL,
            None => true,
        }
    }

    fn should_history_sync_on_reconnect(
        history_sync_on_reconnect: bool,
        previous: Option<RelayStatus>,
        status: RelayStatus,
    ) -> bool {
        history_sync_on_reconnect
            && status == RelayStatus::Connected
            && matches!(
                previous,
                Some(
                    RelayStatus::Initialized
                        | RelayStatus::Pending
                        | RelayStatus::Connecting
                        | RelayStatus::Disconnected
                        | RelayStatus::Terminated
                )
            )
    }

    fn should_run_reconnect_history_sync(last_run: Option<&Instant>) -> bool {
        match last_run {
            None => true,
            Some(last_run) => last_run.elapsed() >= MIRROR_RECONNECT_HISTORY_SYNC_COOLDOWN,
        }
    }

    fn is_metadata_only_history_sync(kinds: &[u16]) -> bool {
        !kinds.is_empty() && kinds.iter().all(|kind| *kind == Kind::Metadata.as_u16())
    }

    fn history_sync_kinds_affect_profile_or_graph(kinds: &[u16]) -> bool {
        kinds.is_empty()
            || kinds.iter().any(|kind| {
                *kind == Kind::Metadata.as_u16()
                    || *kind == Kind::ContactList.as_u16()
                    || *kind == Kind::MuteList.as_u16()
            })
    }

    fn history_sync_plan_for(
        config: &NostrMirrorConfig,
        _authors: usize,
        kinds: &[u16],
    ) -> HistorySyncPlan {
        let author_batch_size = config.author_batch_size.max(1);
        let per_author_event_limit = config.history_sync_per_author_event_limit.max(1);
        let relay_page_size = 1_000;
        let max_relay_pages = 10;

        if Self::is_metadata_only_history_sync(kinds) {
            return HistorySyncPlan {
                relay_fetch_mode: RelayFetchMode::AuthorBatches,
                author_batch_size: author_batch_size.min(METADATA_HISTORY_SYNC_AUTHOR_BATCH_SIZE),
                per_author_event_limit: METADATA_HISTORY_SYNC_PER_AUTHOR_EVENT_LIMIT,
                relay_page_size,
                max_relay_pages,
            };
        }

        HistorySyncPlan {
            relay_fetch_mode: RelayFetchMode::AuthorBatches,
            author_batch_size,
            per_author_event_limit,
            relay_page_size,
            max_relay_pages,
        }
    }

    fn history_sync_plan(&self, authors: usize, kinds: &[u16]) -> HistorySyncPlan {
        Self::history_sync_plan_for(&self.config, authors, kinds)
    }

    fn full_archive_history_plan(
        mut plan: HistorySyncPlan,
        max_relay_pages: usize,
    ) -> HistorySyncPlan {
        plan.relay_fetch_mode = RelayFetchMode::AuthorBatches;
        plan.max_relay_pages = max_relay_pages.max(1);
        plan.per_author_event_limit = plan
            .per_author_event_limit
            .max(plan.relay_page_size.saturating_mul(plan.max_relay_pages));
        plan
    }

    fn full_archive_fetch_timeout(base: Duration, max_relay_pages: usize) -> Duration {
        let multiplier = max_relay_pages.clamp(1, FULL_ARCHIVE_FETCH_TIMEOUT_MAX_MULTIPLIER) as u32;
        base.checked_mul(multiplier).unwrap_or(Duration::MAX)
    }

    fn history_sync_chunk_size_for_config(
        config: &NostrMirrorConfig,
        _authors: usize,
        _kinds: &[u16],
        full_author_history: bool,
        chunk_size_override: Option<usize>,
    ) -> usize {
        let configured = chunk_size_override
            .unwrap_or(config.history_sync_author_chunk_size)
            .max(1);
        if full_author_history {
            configured.min(config.author_batch_size.max(1))
        } else {
            configured
        }
    }

    async fn history_sync_authors(&self, authors: Vec<String>) -> Result<()> {
        let kinds = Self::history_sync_kinds_for_config(&self.config);
        if kinds.is_empty() {
            info!("Nostr mirror history sync skipped: no enabled history kinds");
            return Ok(());
        }
        self.history_sync_authors_with_kinds(authors, &kinds).await
    }

    async fn history_sync_authors_with_kinds(
        &self,
        authors: Vec<String>,
        kinds: &[u16],
    ) -> Result<()> {
        self.history_sync_authors_with_kinds_and_mode(authors, kinds, false, None)
            .await
    }

    async fn history_sync_archive_for_reachable_authors(&self) -> Result<()> {
        let Some(settings) = self.archive_history_settings() else {
            info!("Nostr mirror configured archive history sync skipped: disabled settings");
            return Ok(());
        };
        let distance = settings.follow_distance;
        info!(
            "Nostr mirror configured archive history author collection starting: max_follow_distance={distance}"
        );
        let authors = self
            .prioritize_archive_history_authors(self.collect_authors_with_max_distance(distance)?)
            .await?;
        self.history_sync_distance_filtered_archive(authors, settings)
            .await
    }

    async fn history_sync_archive_for_authors(&self, authors: Vec<String>) -> Result<()> {
        let Some(settings) = self.archive_history_settings() else {
            info!("Nostr mirror configured archive history sync skipped: disabled settings");
            return Ok(());
        };
        let distance = settings.follow_distance;
        let mut close_authors = Vec::new();
        for author in authors {
            let Some(pubkey) = decode_hex_pubkey(&author) else {
                continue;
            };
            if self
                .graph_store
                .follow_distance(&pubkey)?
                .is_some_and(|actual_distance| actual_distance <= distance)
            {
                close_authors.push(author);
            }
        }
        if close_authors.is_empty() {
            return Ok(());
        }
        let close_authors = self
            .prioritize_archive_history_authors(close_authors)
            .await?;

        self.history_sync_distance_filtered_archive(close_authors, settings)
            .await
    }

    async fn history_sync_distance_filtered_archive(
        &self,
        close_authors: Vec<String>,
        settings: ArchiveHistorySettings,
    ) -> Result<()> {
        if close_authors.is_empty() {
            return Ok(());
        }

        info!(
            "Nostr mirror configured archive history sync starting: authors={} max_follow_distance={} max_relay_pages={}",
            close_authors.len(),
            settings.follow_distance,
            settings.max_relay_pages
        );
        info!(
            "Nostr mirror configured archive per-kind sync starting: kinds={:?} authors={}",
            settings.kinds,
            close_authors.len()
        );
        self.history_sync_authors_with_kinds_and_mode(
            close_authors,
            &settings.kinds,
            true,
            Some(settings.max_relay_pages),
        )
        .await
    }

    async fn history_sync_authors_with_kinds_and_mode(
        &self,
        authors: Vec<String>,
        kinds: &[u16],
        full_author_history: bool,
        max_relay_pages: Option<usize>,
    ) -> Result<()> {
        let update_profile_and_graph = Self::history_sync_kinds_affect_profile_or_graph(kinds);
        let chunk_size = Self::history_sync_chunk_size_for_config(
            &self.config,
            authors.len(),
            kinds,
            full_author_history,
            None,
        );
        self.history_sync_authors_chunked(
            authors,
            |current_root, author_chunk| async move {
                self.history_sync_author_chunk(
                    current_root,
                    author_chunk,
                    kinds,
                    full_author_history,
                    max_relay_pages,
                )
                .await
            },
            update_profile_and_graph,
            Some(chunk_size),
        )
        .await
    }

    async fn history_sync_authors_chunked<F, Fut>(
        &self,
        authors: Vec<String>,
        mut run_chunk: F,
        update_profile_and_graph: bool,
        chunk_size_override: Option<usize>,
    ) -> Result<()>
    where
        F: FnMut(Option<hashtree_core::Cid>, Vec<String>) -> Fut,
        Fut: std::future::Future<Output = Result<CrawlReport>>,
    {
        if authors.is_empty() {
            return Ok(());
        }
        if self.shutting_down.load(Ordering::Acquire) {
            return Ok(());
        }
        let _publication_deferral = RootPublicationDeferral::new(&self.root_publication_deferrals);

        info!(
            "Nostr mirror history sync starting: authors={} relays={} negentropy_only={}",
            authors.len(),
            self.config.relays.len(),
            self.config.require_negentropy
        );

        let mut current_root = self.graph_store.public_events_root_for_write()?;
        let mut last_error = None;
        let mut applied_chunks = 0usize;
        let mut failed_chunks = 0usize;
        let mut authors_since_publish = 0usize;
        let publish_checkpoint_authors = self.config.history_sync_author_chunk_size.max(1);
        let chunk_size = chunk_size_override
            .unwrap_or(self.config.history_sync_author_chunk_size)
            .max(1);
        let total_chunks = authors.len().div_ceil(chunk_size);

        for (chunk_index, author_chunk) in authors.chunks(chunk_size).enumerate() {
            if self.shutting_down.load(Ordering::Acquire) {
                return Ok(());
            }
            let author_chunk = author_chunk.to_vec();
            let author_count = author_chunk.len();
            info!(
                "Nostr mirror history sync chunk starting: chunk={}/{} authors={}",
                chunk_index + 1,
                total_chunks,
                author_count
            );
            let mut shutdown_rx = self.shutdown_rx.clone();
            let chunk_result = tokio::select! {
                result = run_chunk(current_root.clone(), author_chunk.clone()) => result,
                _ = shutdown_rx.changed() => return Ok(()),
            };
            let report = match chunk_result {
                Ok(report) => report,
                Err(err) => {
                    failed_chunks = failed_chunks.saturating_add(1);
                    warn!(
                        "Nostr mirror history sync chunk failed: chunk={}/{} authors={} error={:#}",
                        chunk_index + 1,
                        total_chunks,
                        author_count,
                        err
                    );
                    last_error = Some(err);
                    trim_transient_allocations();
                    continue;
                }
            };

            if self.shutting_down.load(Ordering::Acquire) {
                return Ok(());
            }
            if report.root != current_root {
                current_root = self
                    .apply_history_root_with_options(
                        current_root.clone(),
                        report.root.clone(),
                        update_profile_and_graph,
                        false,
                        Some(&report.applied_events),
                    )
                    .await?;
                info!(
                    "Nostr mirror history sync updated trusted root: chunk={}/{} authors_processed={} events_selected={} events_seen={}",
                    chunk_index + 1,
                    total_chunks,
                    report.authors_processed,
                    report.events_selected,
                    report.events_seen
                );
            } else {
                current_root = self.graph_store.public_events_root_for_write()?;
            }
            applied_chunks = applied_chunks.saturating_add(1);
            authors_since_publish = authors_since_publish.saturating_add(author_count);
            if authors_since_publish >= publish_checkpoint_authors {
                self.publish_history_roots(update_profile_and_graph).await;
                authors_since_publish = 0;
            }
            trim_transient_allocations();
        }

        if applied_chunks == 0 {
            return Err(last_error
                .unwrap_or_else(|| anyhow::anyhow!("mirror history sync made no progress"))
                .context(format!(
                    "mirror history sync failed: applied_chunks=0 failed_chunks={failed_chunks}"
                )));
        }
        if failed_chunks > 0 {
            warn!(
                "Nostr mirror history sync completed with skipped chunks: applied_chunks={} failed_chunks={}",
                applied_chunks, failed_chunks
            );
        }
        // A remainder smaller than the checkpoint cadence still owns the newest
        // durable local root. Force one final publication attempt before the
        // deferral guard lets the periodic publisher resume.
        self.publish_history_roots(update_profile_and_graph).await;
        if failed_chunks > 0 {
            return Err(last_error
                .unwrap_or_else(|| anyhow::anyhow!("mirror history sync skipped chunks"))
                .context(format!(
                    "mirror history sync incomplete: applied_chunks={applied_chunks} failed_chunks={failed_chunks}"
                )));
        }
        Ok(())
    }

    async fn history_sync_author_chunk(
        &self,
        current_root: Option<hashtree_core::Cid>,
        authors: Vec<String>,
        kinds: &[u16],
        full_author_history: bool,
        max_relay_pages: Option<usize>,
    ) -> Result<CrawlReport> {
        let mut last_error = None;
        let mut report = None;
        let mut plan = self.history_sync_plan(authors.len(), kinds);
        if full_author_history {
            plan = Self::full_archive_history_plan(
                plan,
                max_relay_pages.unwrap_or(plan.max_relay_pages),
            );
        }
        let fetch_timeout = if full_author_history {
            Self::full_archive_fetch_timeout(self.config.fetch_timeout, plan.max_relay_pages)
        } else {
            self.config.fetch_timeout
        };
        for attempt in 0..3 {
            let mut last_logged_authors = 0usize;
            let bridge = NostrBridge::new(
                self.store.store_arc(),
                CrawlConfig {
                    relays: self.config.relays.clone(),
                    author_allowlist: Some(authors.clone()),
                    max_live_bytes: None,
                    max_events_seen: None,
                    max_authors: None,
                    max_follow_distance: None,
                    author_batch_size: plan.author_batch_size,
                    per_author_event_limit: plan.per_author_event_limit,
                    per_author_kind_event_limit: Some(plan.per_author_event_limit),
                    per_author_live_bytes: None,
                    fetch_timeout,
                    kinds: Some(kinds.to_vec()),
                    relay_fetch_mode: plan.relay_fetch_mode,
                    require_negentropy: self.config.require_negentropy,
                    relay_event_max_size: self.config.relay_event_max_size,
                    relay_page_size: plan.relay_page_size,
                    max_relay_pages: plan.max_relay_pages,
                    full_author_history,
                },
            );

            let crawl = bridge.crawl_with_progress(
                self.graph_store.as_ref(),
                current_root.as_ref(),
                |progress| {
                    let log_interval = self.config.author_batch_size.saturating_mul(8).max(2_048);
                    let should_log = progress.authors_processed == progress.authors_considered
                        || progress.authors_processed == 0
                        || progress
                            .authors_processed
                            .saturating_sub(last_logged_authors)
                            >= log_interval;
                    if should_log {
                        last_logged_authors = progress.authors_processed;
                        info!(
                            "Nostr mirror history sync progress: authors_processed={}/{} events_selected={} events_seen={}",
                            progress.authors_processed,
                            progress.authors_considered,
                            progress.events_selected,
                            progress.events_seen
                        );
                    }
                },
            );
            // Reconciliation and each fallback relay page are already bounded.
            // A second fixed timeout around the whole batched crawl can only
            // abort valid multi-author, multi-kind work in the middle of a pass.
            let crawl_result: Result<CrawlReport> = crawl.await.map_err(Into::into);

            match crawl_result {
                Ok(next_report) => {
                    report = Some(next_report);
                    break;
                }
                Err(err) => {
                    last_error = Some(err);
                    if attempt < 2 {
                        tokio::time::sleep(Duration::from_millis(500)).await;
                    }
                }
            }
        }
        report
            .ok_or_else(|| last_error.expect("history sync retry captured error"))
            .context("run mirror history sync")
    }

    #[cfg(test)]
    async fn apply_history_root(&self, root: Option<&hashtree_core::Cid>) -> Result<()> {
        let expected_root = self.graph_store.public_events_root_for_write()?;
        self.apply_history_root_with_options(expected_root, root.cloned(), true, true, None)
            .await
            .map(|_| ())
    }

    async fn apply_history_root_with_options(
        &self,
        expected_root: Option<hashtree_core::Cid>,
        root: Option<hashtree_core::Cid>,
        update_profile_and_graph: bool,
        publish_roots: bool,
        applied_events: Option<&[hashtree_nostr::StoredNostrEvent]>,
    ) -> Result<Option<hashtree_core::Cid>> {
        let event_store = NostrEventStore::new(self.store.store_arc());
        let merge_events = match (root.as_ref(), applied_events) {
            (_, Some(events)) => events.to_vec(),
            (Some(root), None) => event_store
                .list_recent(Some(root), ListEventsOptions::default())
                .await
                .context("list complete trusted mirrored events for root application")?,
            (None, None) => Vec::new(),
        };
        let incremental = applied_events.is_some();
        let mut expected_root = expected_root;
        let mut candidate_root = root;
        let mut applied = false;

        for attempt in 0..=HISTORY_ROOT_REBASE_MAX_ATTEMPTS {
            let events = if update_profile_and_graph {
                let stored = if incremental {
                    merge_events.clone()
                } else {
                    match candidate_root.as_ref() {
                        Some(root) => event_store
                            .list_recent(Some(root), ListEventsOptions::default())
                            .await
                            .context("list complete rebased mirrored events")?,
                        None => Vec::new(),
                    }
                };
                stored
                    .into_iter()
                    .map(socialgraph::stored_event_to_nostr_event)
                    .collect::<Result<Vec<_>>>()?
            } else {
                Vec::new()
            };

            match self
                .graph_store
                .apply_public_events_root_and_projections(
                    expected_root.as_ref(),
                    candidate_root.as_ref(),
                    &events,
                    update_profile_and_graph && !incremental,
                )
                .context("compare-and-apply mirrored event root and derived projections")?
            {
                socialgraph::PublicEventsRootApplyOutcome::Applied => {
                    applied = true;
                    break;
                }
                socialgraph::PublicEventsRootApplyOutcome::Conflict { current_root } => {
                    if attempt == HISTORY_ROOT_REBASE_MAX_ATTEMPTS {
                        anyhow::bail!(
                            "mirrored event root changed during all {} bounded rebase attempts",
                            HISTORY_ROOT_REBASE_MAX_ATTEMPTS
                        );
                    }
                    info!(
                        "Nostr mirror history root advanced during candidate application; rebasing without overwriting the newer root: attempt={}/{} events_applied={}",
                        attempt + 1,
                        HISTORY_ROOT_REBASE_MAX_ATTEMPTS,
                        merge_events.len()
                    );
                    expected_root = current_root;
                    if candidate_root.is_none() || merge_events.is_empty() {
                        candidate_root = expected_root.clone();
                        break;
                    }
                    candidate_root = event_store
                        .build(expected_root.as_ref(), merge_events.clone())
                        .await
                        .context("rebuild history candidate from exact current event root")?;
                }
            }
        }

        if !applied {
            return Ok(candidate_root);
        }
        if update_profile_and_graph {
            self.note_profile_search_root_change()?;
            self.note_profiles_by_pubkey_root_change()?;
        }
        self.note_public_events_root_change()?;
        if !publish_roots {
            return Ok(candidate_root);
        }
        self.publish_history_roots(update_profile_and_graph).await;
        Ok(candidate_root)
    }

    async fn publish_history_roots(&self, update_profile_and_graph: bool) {
        if self.shutting_down.load(Ordering::Acquire) {
            return;
        }
        let (event_result, profile_search_result, profiles_by_pubkey_result) = self
            .publish_priority_roots(true, update_profile_and_graph, update_profile_and_graph)
            .await;
        if let Err(err) = event_result {
            warn!(
                "Nostr mirror event-root publish failed after root update: {:#}",
                err
            );
        }
        if let Err(err) = profile_search_result {
            warn!(
                "Nostr mirror profile-search publish failed after root update: {:#}",
                err
            );
        }
        if let Err(err) = profiles_by_pubkey_result {
            warn!(
                "Nostr mirror profiles-by-pubkey publish failed after root update: {:#}",
                err
            );
        }
    }

    async fn subscribe_authors_since(
        &self,
        authors: &[String],
        since: Timestamp,
        subscribed_authors: &mut HashSet<String>,
    ) -> Result<()> {
        let new_authors = authors
            .iter()
            .filter(|author| !subscribed_authors.contains(*author))
            .cloned()
            .collect::<Vec<_>>();
        if new_authors.is_empty() {
            return Ok(());
        }

        for chunk in new_authors.chunks(self.config.author_batch_size.max(1)) {
            let pubkeys = chunk
                .iter()
                .filter_map(|author| PublicKey::from_hex(author).ok())
                .collect::<Vec<_>>();
            if pubkeys.is_empty() {
                continue;
            }

            let filter = Filter::new()
                .authors(pubkeys)
                .kinds(self.config.kinds.iter().copied().map(Kind::from))
                .since(since);

            if let Err(err) = self.client.subscribe(filter, None).await {
                warn!(
                    "Nostr mirror author subscription failed: authors={} error={:#}",
                    chunk.len(),
                    err
                );
                continue;
            }
            subscribed_authors.extend(chunk.iter().cloned());
        }
        Ok(())
    }

    fn ingest_live_event(&self, event: &Event) -> Result<()> {
        self.pending_live_events
            .lock()
            .expect("pending live events")
            .insert(event.id.to_hex(), event.clone());
        Ok(())
    }

    async fn flush_live_events(&self) -> Result<()> {
        let pending = {
            let mut pending = self
                .pending_live_events
                .lock()
                .expect("pending live events");
            if pending.is_empty() {
                return Ok(());
            }
            std::mem::take(&mut *pending)
        };
        let events = pending.into_values().collect::<Vec<_>>();
        let event_count = events.len();
        let previous_event_root = self.graph_store.public_events_root()?;
        let previous_profile_search_root = self.graph_store.profile_search_root()?;
        let previous_profiles_by_pubkey_root = self.graph_store.profiles_by_pubkey_root()?;

        socialgraph::ingest_parsed_events_with_storage_class(
            self.graph_store.as_ref(),
            &events,
            socialgraph::EventStorageClass::Public,
        )
        .context("ingest live mirrored event batch")?;

        let next_event_root = self.graph_store.public_events_root()?;
        let next_profile_search_root = self.graph_store.profile_search_root()?;
        let next_profiles_by_pubkey_root = self.graph_store.profiles_by_pubkey_root()?;
        let event_root_changed = next_event_root != previous_event_root;
        let profile_search_root_changed = next_profile_search_root != previous_profile_search_root;
        let profiles_by_pubkey_root_changed =
            next_profiles_by_pubkey_root != previous_profiles_by_pubkey_root;

        if event_root_changed {
            self.note_public_events_root_change()?;
        }
        if profile_search_root_changed {
            self.note_profile_search_root_change()?;
        }
        if profiles_by_pubkey_root_changed {
            self.note_profiles_by_pubkey_root_change()?;
        }
        info!(
            "Nostr mirror flushed live events: events={} event_root_changed={} profile_search_root_changed={} profiles_by_pubkey_root_changed={}",
            event_count,
            event_root_changed,
            profile_search_root_changed,
            profiles_by_pubkey_root_changed
        );
        Ok(())
    }

    fn note_public_events_root_change(&self) -> Result<()> {
        let root = self.graph_store.public_events_root()?;
        if let Err(err) = Self::write_latest_event_root_state(self.store.base_path(), root.as_ref())
        {
            warn!("Nostr mirror failed to persist queryable event-root state: {err:#}");
        }
        Self::note_root_change(
            self.config.published_event_tree_name.as_deref(),
            &self.event_publish_state,
            root,
        )
    }

    fn note_profile_search_root_change(&self) -> Result<()> {
        let root = self.graph_store.profile_search_root()?;
        Self::note_root_change(
            self.config.published_profile_search_tree_name.as_deref(),
            &self.profile_search_publish_state,
            root,
        )
    }

    fn note_profiles_by_pubkey_root_change(&self) -> Result<()> {
        let root = self.graph_store.profiles_by_pubkey_root()?;
        Self::note_root_change(
            self.config
                .published_profiles_by_pubkey_tree_name
                .as_deref(),
            &self.profiles_by_pubkey_publish_state,
            root,
        )
    }

    fn note_root_change(
        tree_name: Option<&str>,
        publish_state: &Arc<Mutex<RootPublishState>>,
        root: Option<hashtree_core::Cid>,
    ) -> Result<()> {
        let Some(_tree_name) = tree_name else {
            return Ok(());
        };

        let mut state = publish_state.lock().expect("root publish state");
        let now = Instant::now();

        if state.pending_root == root {
            return Ok(());
        }

        state.pending_root = root;
        state.last_upload_failed_at = None;
        state.last_upload_error = None;
        state.last_changed_at = Some(now);
        if state.dirty_since.is_none() {
            state.dirty_since = Some(now);
        }
        Ok(())
    }

    async fn maybe_publish_event_root(&self, force: bool) -> Result<()> {
        if self.take_missing_blob_event_upload_error() {
            return self.rebuild_event_indexes_after_missing_blobs(force).await;
        }
        let result = self
            .maybe_publish_root(
                self.config.published_event_tree_name.as_deref(),
                &self.event_publish_state,
                "event root",
                force,
                false,
                false,
            )
            .await;
        let Err(error) = result else {
            if self.take_missing_blob_event_upload_error() {
                return self.rebuild_event_indexes_after_missing_blobs(force).await;
            }
            return Ok(());
        };
        if !is_missing_local_blob_push_error(&error) {
            return Err(error);
        }

        self.rebuild_event_indexes_after_missing_blobs(force).await
    }

    fn take_missing_blob_event_upload_error(&self) -> bool {
        let mut state = self
            .event_publish_state
            .lock()
            .expect("event root publish state");
        if state.upload_in_progress_root.is_some() {
            return false;
        }
        if !state.missing_blob_rebuild_required {
            return false;
        }
        state.missing_blob_rebuild_required = false;
        state.last_upload_error = None;
        state.last_upload_failed_at = None;
        true
    }

    async fn rebuild_event_indexes_after_missing_blobs(&self, force: bool) -> Result<()> {
        warn!(
            "Nostr mirror event root DAG references missing local blobs; rebuilding event indexes from stored events"
        );
        let (public_count, ambient_count) = self
            .graph_store
            .rebuild_event_indexes_from_stored_events_async()
            .await
            .context("rebuild event indexes after missing event blobs")?;
        info!(
            "Nostr mirror rebuilt event indexes after missing blobs: public={} ambient={}",
            public_count, ambient_count
        );
        self.sync_publish_roots_from_store()?;

        self.maybe_publish_root(
            self.config.published_event_tree_name.as_deref(),
            &self.event_publish_state,
            "event root",
            force,
            false,
            false,
        )
        .await
    }

    async fn maybe_publish_profile_search_root(&self, force: bool) -> Result<()> {
        let Some(publication) = self.acquire_profile_publication_or_defer().await else {
            return Ok(());
        };
        let result = self
            .maybe_publish_root(
                self.config.published_profile_search_tree_name.as_deref(),
                &self.profile_search_publish_state,
                "profile search root",
                force,
                true,
                true,
            )
            .await;
        drop(publication);
        result
    }

    async fn maybe_publish_profiles_by_pubkey_root(&self, force: bool) -> Result<()> {
        let Some(publication) = self.acquire_profile_publication_or_defer().await else {
            return Ok(());
        };
        let result = self
            .maybe_publish_root(
                self.config
                    .published_profiles_by_pubkey_tree_name
                    .as_deref(),
                &self.profiles_by_pubkey_publish_state,
                "profiles-by-pubkey root",
                force,
                true,
                true,
            )
            .await;
        drop(publication);
        result
    }

    async fn acquire_profile_publication_or_defer(
        &self,
    ) -> Option<socialgraph::ProfilePublicationGuard> {
        match socialgraph::acquire_profile_publication_guard(self.store.base_path()).await {
            Ok(publication) => {
                self.profile_publication_fence_logged
                    .store(false, Ordering::Release);
                Some(publication)
            }
            Err(error) => {
                if !self
                    .profile_publication_fence_logged
                    .swap(true, Ordering::AcqRel)
                {
                    warn!(
                        "Nostr mirror deferred all profile-root publication: {:#}",
                        error
                    );
                }
                None
            }
        }
    }

    async fn maybe_publish_root(
        &self,
        tree_name: Option<&str>,
        publish_state: &Arc<Mutex<RootPublishState>>,
        log_label: &str,
        force: bool,
        publish_before_upload_ready_on_force: bool,
        profile_publication: bool,
    ) -> Result<()> {
        if !force && self.root_publication_deferrals.load(Ordering::Acquire) > 0 {
            return Ok(());
        }
        let Some(tree_name) = tree_name else {
            return Ok(());
        };

        let pending_root = {
            let state = publish_state.lock().expect("root publish state");
            let Some(pending_root) = state.pending_root.clone() else {
                return Ok(());
            };

            let now = Instant::now();
            let debounce_ready = state.last_changed_at.is_some_and(|changed_at| {
                now.duration_since(changed_at) >= MIRROR_ROOT_PUBLISH_DEBOUNCE
            });
            let stale_ready = state.dirty_since.is_some_and(|dirty_since| {
                now.duration_since(dirty_since) >= MIRROR_ROOT_PUBLISH_MAX_STALENESS
            });
            if !force && !debounce_ready && !stale_ready {
                return Ok(());
            }

            pending_root
        };

        let upload_started = self.maybe_start_background_root_upload(
            tree_name,
            &pending_root,
            publish_state,
            log_label,
            profile_publication,
        );
        let upload_required = !self.config.blossom_write_servers.is_empty();
        let (upload_ready, publish_root) = {
            let state = publish_state.lock().expect("root publish state");
            let upload_ready =
                !upload_required || state.last_uploaded_root.as_ref() == Some(&pending_root);
            let publish_root = if upload_ready {
                Some(pending_root.clone())
            } else {
                state.last_uploaded_root.clone().filter(|uploaded_root| {
                    state.last_published_root.as_ref() != Some(uploaded_root)
                })
            };
            (upload_ready, publish_root)
        };
        let publish_before_upload_ready =
            force && upload_required && !upload_ready && publish_before_upload_ready_on_force;
        let publish_root = if let Some(publish_root) = publish_root {
            publish_root
        } else if publish_before_upload_ready {
            pending_root.clone()
        } else {
            if upload_started {
                info!(
                    "Nostr mirror uploading {} DAG before publish: tree={} hash={}",
                    log_label,
                    tree_name,
                    hex::encode(pending_root.hash),
                );
            }
            return Ok(());
        };

        let mut successful_relays = Vec::new();
        let mut failed_relays = Vec::new();
        let mut published_now = false;
        let publish_required =
            self.publish_client.is_some() && !self.config.publish_relays.is_empty();
        if publish_required {
            let Some(publish_client) = self.publish_client.as_ref() else {
                unreachable!("publish_required implies publish_client");
            };
            if !self.has_connected_publish_relay().await {
                return Ok(());
            }
            if publish_before_upload_ready {
                info!(
                    "Nostr mirror publishing {} before Blossom upload completes: tree={} hash={}",
                    log_label,
                    tree_name,
                    hex::encode(pending_root.hash),
                );
            } else if !upload_ready {
                info!(
                    "Nostr mirror publishing uploaded {} while newer root is still uploading: tree={} published_hash={} pending_hash={}",
                    log_label,
                    tree_name,
                    hex::encode(publish_root.hash),
                    hex::encode(pending_root.hash),
                );
            }

            let already_published = {
                let state = publish_state.lock().expect("root publish state");
                state.last_published_root.as_ref() == Some(&publish_root)
            };
            if !already_published {
                let publish_relays = self.config.publish_relays.clone();
                let latest_known_created_at = {
                    let state = publish_state.lock().expect("root publish state");
                    state.last_published_created_at
                };
                let publish_created_at =
                    next_replaceable_created_at(Timestamp::now(), latest_known_created_at);
                let event = publish_client
                    .sign_event_builder(Self::build_public_root_event(
                        tree_name,
                        &publish_root,
                        publish_created_at,
                    ))
                    .await
                    .with_context(|| format!("sign {log_label} event"))?;
                let publish_result = self
                    .publish_root_event_to_relays(publish_client, &publish_relays, &event)
                    .await
                    .with_context(|| format!("publish {log_label} event"))?;
                successful_relays = publish_result.0;
                failed_relays = publish_result.1;
                if successful_relays.is_empty() {
                    let failure_summary = if failed_relays.is_empty() {
                        "no publish relays accepted the event".to_string()
                    } else {
                        failed_relays.join("; ")
                    };
                    anyhow::bail!("no publish relays accepted the event ({failure_summary})");
                }

                let mut state = publish_state.lock().expect("root publish state");
                if state.pending_root.as_ref() == Some(&pending_root)
                    || state.last_uploaded_root.as_ref() == Some(&publish_root)
                    || publish_before_upload_ready
                {
                    state.last_published_root = Some(publish_root.clone());
                    state.last_published_at = Some(Instant::now());
                    state.last_published_created_at = Some(event.created_at);
                }
                published_now = true;
            }
        }

        {
            let mut state = publish_state.lock().expect("root publish state");
            if state.pending_root.as_ref() == Some(&pending_root) {
                let upload_satisfied = self.config.blossom_write_servers.is_empty()
                    || state.last_uploaded_root.as_ref() == Some(&pending_root);
                let publish_satisfied =
                    !publish_required || state.last_published_root.as_ref() == Some(&pending_root);
                if upload_satisfied && publish_satisfied {
                    state.dirty_since = None;
                }
            }
        }

        if published_now {
            info!(
                "Nostr mirror published {}: tree={} hash={} relays={:?}",
                log_label,
                tree_name,
                hex::encode(publish_root.hash),
                successful_relays,
            );
        }
        if !failed_relays.is_empty() {
            warn!(
                "Nostr mirror publish had relay failures: tree={} failures={:?}",
                tree_name, failed_relays
            );
        }
        Ok(())
    }

    async fn collect_root_upload_cids(
        store: &HashtreeStore,
        root: hashtree_core::Cid,
        previous_root: Option<hashtree_core::Cid>,
    ) -> Result<Vec<hashtree_core::Cid>> {
        let fetcher = Fetcher::new(FetchConfig::default());
        if let Some(previous_root) = previous_root {
            match collect_incremental_cids_for_push(
                store,
                root.clone(),
                previous_root,
                Some(&fetcher),
            )
            .await
            {
                Ok(cids) => return Ok(cids),
                Err(error) => {
                    warn!(
                        "Nostr mirror Blossom DAG diff failed; falling back to full push: {error:#}"
                    );
                }
            }
        }
        collect_cids_for_push(store, root, Some(&fetcher)).await
    }

    async fn upload_root_blob_to_any_server(
        keys: &Keys,
        servers: &[String],
        data: &[u8],
        cancelled: &watch::Receiver<bool>,
    ) -> Result<bool> {
        let mut last_error = None;
        for server in servers {
            if *cancelled.borrow() {
                return Ok(false);
            }
            // A single-server client makes one bounded PUT attempt. We never
            // select cancellation against that future: after the request has
            // begun, its result is awaited before the publication guard can
            // be released. Cancellation is observed before the next server.
            let client = hashtree_blossom::BlossomClient::new_empty(keys.clone())
                .with_write_servers(vec![server.clone()])
                .with_timeout(MIRROR_ROOT_UPLOAD_REQUEST_TIMEOUT);
            match client.upload(data).await {
                Ok(_) => return Ok(true),
                Err(error) => {
                    if *cancelled.borrow() {
                        return Ok(false);
                    }
                    last_error = Some(format!("{server}: {error}"));
                }
            }
        }
        anyhow::bail!(
            "all configured file servers rejected the blob{}",
            last_error
                .as_deref()
                .map(|error| format!(" (last: {error})"))
                .unwrap_or_default()
        )
    }

    async fn upload_root_cids_until_cancelled(
        store: &HashtreeStore,
        cids: &[hashtree_core::Cid],
        servers: &[String],
        cancelled: &watch::Receiver<bool>,
    ) -> Result<bool> {
        let (nsec, _) = ensure_keys_string()?;
        let keys = Keys::parse(&nsec).context("parse keys for mirror Blossom publication")?;

        let batch_count = cids.len().div_ceil(MIRROR_ROOT_UPLOAD_CONCURRENCY);
        for (batch_index, batch) in cids.chunks(MIRROR_ROOT_UPLOAD_CONCURRENCY).enumerate() {
            if *cancelled.borrow() {
                return Ok(false);
            }
            let mut blobs = Vec::with_capacity(batch.len());
            for cid in batch {
                let data = store
                    .get_blob(&cid.hash)?
                    .ok_or_else(|| anyhow::anyhow!("missing local blob while uploading {}", cid))?;
                blobs.push(data);
            }

            let results =
                futures::future::join_all(blobs.iter().map(|data| {
                    Self::upload_root_blob_to_any_server(&keys, servers, data, cancelled)
                }))
                .await;
            let mut first_error = None;
            let mut fully_uploaded = true;
            for result in results {
                match result {
                    Ok(uploaded) => fully_uploaded &= uploaded,
                    Err(error) if first_error.is_none() => first_error = Some(error),
                    Err(_) => {}
                }
            }
            if let Some(error) = first_error {
                return Err(error);
            }
            if !fully_uploaded {
                return Ok(false);
            }
            if *cancelled.borrow() && batch_index + 1 < batch_count {
                return Ok(false);
            }
        }
        Ok(true)
    }

    async fn run_root_upload(
        job: RootUploadJob,
        cancelled: &mut watch::Receiver<bool>,
    ) -> Result<bool> {
        let RootUploadJob {
            store,
            root,
            previous_root,
            servers,
            data_dir,
            profile_publication,
            log_label,
        } = job;
        // DAG discovery and any missing-local-blob hydration happen before
        // acquiring the external-publication guard and are freely
        // cancellable: they cannot publish a stale profile root.
        let collect = Self::collect_root_upload_cids(store.as_ref(), root, previous_root);
        tokio::pin!(collect);
        let cids = tokio::select! {
            result = &mut collect => result?,
            _ = cancelled.changed() => return Ok(false),
        };
        if *cancelled.borrow() {
            return Ok(false);
        }

        let _profile_publication = if profile_publication {
            let publication = tokio::select! {
                result = socialgraph::acquire_profile_publication_guard(&data_dir) => {
                    result.with_context(|| {
                        format!("acquire detached {log_label} Blossom publication guard")
                    })?
                }
                _ = cancelled.changed() => return Ok(false),
            };
            Some(publication)
        } else {
            None
        };

        Self::upload_root_cids_until_cancelled(store.as_ref(), &cids, &servers, cancelled).await
    }

    fn maybe_start_background_root_upload(
        &self,
        tree_name: &str,
        pending_root: &hashtree_core::Cid,
        publish_state: &Arc<Mutex<RootPublishState>>,
        log_label: &str,
        profile_publication: bool,
    ) -> bool {
        if self.config.blossom_write_servers.is_empty() {
            return false;
        }

        // Hold the registry lock across the final shutdown check and task
        // registration. shutdown() takes the same lock after setting the flag,
        // so cleanup cannot miss a worker that was concurrently being born.
        let mut tasks = self.root_upload_tasks.lock().expect("root upload tasks");
        if self.shutting_down.load(Ordering::Acquire) {
            return false;
        }
        let previous_uploaded_root = {
            let mut state = publish_state.lock().expect("root publish state");
            if state.last_uploaded_root.as_ref() == Some(pending_root)
                || state.upload_in_progress_root.is_some()
            {
                return false;
            }
            if state
                .last_upload_failed_at
                .is_some_and(|failed_at| failed_at.elapsed() < MIRROR_ROOT_UPLOAD_RETRY_INTERVAL)
            {
                return false;
            }
            state.upload_in_progress_root = Some(pending_root.clone());
            state.last_uploaded_root.clone()
        };

        let store = Arc::clone(&self.store);
        let upload_state_path = Self::uploaded_root_state_path(store.base_path(), tree_name);
        let servers = self.config.blossom_write_servers.clone();
        let root = pending_root.clone();
        let publish_state = Arc::clone(publish_state);
        let log_label = log_label.to_string();
        let data_dir = store.base_path().to_path_buf();
        let (cancel, mut cancelled) = watch::channel(false);
        let (finished_tx, finished) = oneshot::channel();
        let join = std::thread::spawn(move || {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("build nostr mirror root upload runtime");
            let result = runtime.block_on(Self::run_root_upload(
                RootUploadJob {
                    store: Arc::clone(&store),
                    root: root.clone(),
                    previous_root: previous_uploaded_root,
                    servers,
                    data_dir,
                    profile_publication,
                    log_label: log_label.clone(),
                },
                &mut cancelled,
            ));
            let mut state = publish_state.lock().expect("root publish state");
            if state.upload_in_progress_root.as_ref() == Some(&root) {
                state.upload_in_progress_root = None;
            }
            match result {
                Ok(true) => {
                    if let Err(err) =
                        Self::write_uploaded_root_state(&upload_state_path, &root, &log_label)
                    {
                        warn!("Nostr mirror failed to persist uploaded root state: {err:#}");
                    }
                    state.last_uploaded_root = Some(root.clone());
                    state.last_uploaded_at = Some(Instant::now());
                    if state.pending_root.as_ref() == Some(&root) {
                        state.last_upload_failed_at = None;
                        state.last_upload_error = None;
                        state.missing_blob_rebuild_required = false;
                    }
                    info!(
                        "Nostr mirror uploaded {} DAG to Blossom: hash={}",
                        log_label,
                        hex::encode(root.hash)
                    );
                }
                Ok(false) => {}
                Err(err) => {
                    if state.pending_root.as_ref() == Some(&root) {
                        state.last_upload_failed_at = Some(Instant::now());
                        state.last_upload_error = Some(format!("{err:#}"));
                    }
                    if is_missing_local_blob_message(&format!("{err:#}")) {
                        state.missing_blob_rebuild_required = true;
                    }
                    warn!(
                        "Nostr mirror {} DAG upload failed: hash={} error={:#}",
                        log_label,
                        hex::encode(root.hash),
                        err
                    );
                }
            }
            drop(state);
            trim_transient_allocations();
            let _ = finished_tx.send(());
        });
        tasks.retain(|task| !task.join.is_finished());
        let task = RootUploadTask {
            cancel,
            finished,
            join,
            drain_after_start: profile_publication,
        };
        tasks.push(task);

        true
    }

    async fn publish_root_event_to_relays(
        &self,
        publish_client: &Client,
        relays: &[String],
        event: &Event,
    ) -> Result<(Vec<String>, Vec<String>)> {
        let primary_send = Self::send_root_event_to_relays(publish_client, relays, event);
        let (mut successful_relays, mut failed_relays) =
            match tokio::time::timeout(MIRROR_ROOT_PUBLISH_PRIMARY_TIMEOUT, primary_send).await {
                Ok(result) => result,
                Err(_) => (
                    Vec::new(),
                    vec![format!(
                        "publish relays: primary publish timed out after {:?}",
                        MIRROR_ROOT_PUBLISH_PRIMARY_TIMEOUT
                    )],
                ),
            };

        if successful_relays.is_empty() {
            let (retry_successes, retry_failures) = self
                .publish_root_event_with_fresh_client(relays, event)
                .await;
            successful_relays.extend(retry_successes);
            failed_relays.extend(retry_failures);
        }

        Ok((successful_relays, failed_relays))
    }

    async fn send_root_event_to_relays(
        publish_client: &Client,
        relays: &[String],
        event: &Event,
    ) -> (Vec<String>, Vec<String>) {
        let mut successful_relays = Vec::new();
        let mut failed_relays = Vec::new();

        match publish_client
            .send_event_to(relays.iter().map(|relay| relay.as_str()), event)
            .await
        {
            Ok(output) => {
                for relay in relays {
                    let relay_url = relay.trim_end_matches('/');
                    if output
                        .success
                        .iter()
                        .any(|url| url.as_str().trim_end_matches('/') == relay_url)
                    {
                        successful_relays.push(relay.clone());
                    }
                }
                failed_relays.extend(
                    output
                        .failed
                        .into_iter()
                        .map(|(url, reason)| format!("{url}: {reason}")),
                );
            }
            Err(err) => {
                failed_relays.push(format!("publish relays: {err}"));
            }
        }

        (successful_relays, failed_relays)
    }

    async fn publish_root_event_with_fresh_client(
        &self,
        relays: &[String],
        event: &Event,
    ) -> (Vec<String>, Vec<String>) {
        let client = Client::new(Keys::generate());
        let mut setup_failures = Vec::new();
        for relay in relays {
            if let Err(err) = client.add_relay(relay).await {
                setup_failures.push(format!("{relay}: add relay failed: {err}"));
            }
        }

        client.connect().await;
        let publish = Self::send_root_event_to_relays(&client, relays, event);
        let retry_timeout = self
            .config
            .fetch_timeout
            .min(MIRROR_ROOT_PUBLISH_RETRY_TIMEOUT);
        let result = tokio::time::timeout(retry_timeout, publish).await;
        let _ = client.disconnect().await;

        match result {
            Ok((successful_relays, mut failed_relays)) => {
                failed_relays.extend(setup_failures);
                (successful_relays, failed_relays)
            }
            Err(_) => {
                setup_failures.push(format!(
                    "fresh publish client timed out after {:?}",
                    retry_timeout
                ));
                (Vec::new(), setup_failures)
            }
        }
    }

    fn build_public_root_event(
        tree_name: &str,
        cid: &hashtree_core::Cid,
        created_at: Timestamp,
    ) -> EventBuilder {
        let mut tags = vec![
            Tag::identifier(tree_name.to_string()),
            Tag::custom(
                TagKind::SingleLetter(SingleLetterTag::lowercase(Alphabet::L)),
                vec!["hashtree"],
            ),
            Tag::custom(TagKind::Custom("hash".into()), vec![hex::encode(cid.hash)]),
        ];
        if let Some(key) = cid.key {
            tags.push(Tag::custom(
                TagKind::Custom("key".into()),
                vec![hex::encode(key)],
            ));
        }

        EventBuilder::new(Kind::Custom(HASHTREE_ROOT_KIND as u16), "")
            .tags(tags)
            .custom_created_at(created_at)
    }
}

fn is_missing_local_blob_push_error(error: &anyhow::Error) -> bool {
    error
        .chain()
        .any(|cause| cause.to_string().contains(MISSING_LOCAL_BLOB_PUSH_ERROR))
}

fn is_missing_local_blob_message(message: &str) -> bool {
    message.contains(MISSING_LOCAL_BLOB_PUSH_ERROR)
}

fn next_replaceable_created_at(now: Timestamp, latest_existing: Option<Timestamp>) -> Timestamp {
    match latest_existing {
        Some(latest) if latest >= now => Timestamp::from_secs(latest.as_secs().saturating_add(1)),
        _ => now,
    }
}

#[cfg(test)]
mod tests;