Add ownership snapshot sync

src/cli/ownership.rs

@@ -18,7 +18,7 @@ use crate::output::table::format_idr;
 use crate::ownership::types::{
     ChangeType, FlowSignal, HolderRow, KseiHolding, OwnershipRelease, OwnershipSource,
 };
-use crate::ownership::{db, entities, graph, parser, remote, search};
+use crate::ownership::{db, entities, graph, parser, remote, search, snapshot};
 
 #[derive(Debug, Args)]
 pub struct OwnershipCmd {
@@ -32,6 +32,8 @@ pub enum OwnershipCommand {
     Discover(DiscoverArgs),
     /// Import ownership data from KSEI PDF or Bing API.
     Import(ImportArgs),
+    /// Install or refresh a maintained ownership SQLite snapshot.
+    Sync(SyncArgs),
     /// Show all holders for a ticker (KSEI + Bing combined).
     Ticker(TickerArgs),
     /// Show all holdings for an entity across tickers.
@@ -80,6 +82,16 @@ pub struct ImportArgs {
     pub force: bool,
 }
 
+#[derive(Debug, Args)]
+pub struct SyncArgs {
+    /// Snapshot manifest location (URL or local path).
+    #[arg(long)]
+    pub manifest: Option<String>,
+    /// Replace the local DB even when already current or newer.
+    #[arg(long)]
+    pub force: bool,
+}
+
 #[derive(Debug, Args)]
 pub struct TickerArgs {
     pub symbol: String,
@@ -173,6 +185,7 @@ pub fn handle(cmd: &OwnershipCommand, config: &IdxConfig) -> Result<(), IdxError
     match cmd {
         OwnershipCommand::Discover(args) => handle_discover(args, config),
         OwnershipCommand::Import(args) => handle_import(args, config),
+        OwnershipCommand::Sync(args) => handle_sync(args, config),
         OwnershipCommand::Ticker(args) => handle_ticker(args, config),
         OwnershipCommand::Entity(args) => handle_entity(args, config),
         OwnershipCommand::Search(args) => handle_search(args, config),
@@ -186,6 +199,51 @@ pub fn handle(cmd: &OwnershipCommand, config: &IdxConfig) -> Result<(), IdxError
     }
 }
 
+fn handle_sync(args: &SyncArgs, config: &IdxConfig) -> Result<(), IdxError> {
+    let manifest_source = snapshot::resolve_manifest_source(args.manifest.as_deref())?;
+    let db_path = db::db_path(config)?;
+    let result = snapshot::sync_snapshot(&manifest_source, &db_path, args.force)?;
+
+    if matches!(config.output, OutputFormat::Json) {
+        return json::print_json(&result);
+    }
+
+    match result.action {
+        snapshot::OwnershipSyncAction::Installed => {
+            println!(
+                "Installed ownership snapshot {} ({} release(s), {} tickers) into {}.",
+                result.latest_as_of_date.format("%Y-%m-%d"),
+                result.release_count,
+                result.ticker_count,
+                result.db_path
+            );
+        }
+        snapshot::OwnershipSyncAction::Updated => {
+            println!(
+                "Updated ownership snapshot to {} ({} release(s), {} tickers) in {}.",
+                result.latest_as_of_date.format("%Y-%m-%d"),
+                result.release_count,
+                result.ticker_count,
+                result.db_path
+            );
+        }
+        snapshot::OwnershipSyncAction::Refreshed => {
+            println!(
+                "Refreshed ownership snapshot {} in {}.",
+                result.latest_as_of_date.format("%Y-%m-%d"),
+                result.db_path
+            );
+        }
+        snapshot::OwnershipSyncAction::NoChange
+        | snapshot::OwnershipSyncAction::SkippedNewer
+        | snapshot::OwnershipSyncAction::SkippedDiverged => {
+            println!("{}", result.reason);
+        }
+    }
+
+    Ok(())
+}
+
 fn handle_discover(args: &DiscoverArgs, config: &IdxConfig) -> Result<(), IdxError> {
     if args.limit == 0 {
         return Err(IdxError::ParseError(
@@ -855,6 +913,7 @@ fn resolve_pdf_input(args: &ImportArgs) -> Result<Option<ResolvedPdfInput>, IdxE
                 path.display()
             )));
         }
+
         return Ok(Some(ResolvedPdfInput {
             pdf_path: path.clone(),
             source_url: None,

src/config.rs

@@ -258,6 +258,7 @@ const KNOWN_CONFIG_KEYS: &[&str] = &[
     "cache.quote_ttl",
     "cache.fundamental_ttl",
     "ownership.db_path",
+    "ownership.snapshot_manifest",
 ];
 
 /// Validates a config key and value before writing
@@ -302,6 +303,10 @@ fn normalize_config_value(key: &str, value: &str) -> Result<toml::Value, IdxErro
             // Ownership DB path is free-form and may be absolute or relative.
             Ok(toml::Value::String(value.to_string()))
         }
+        "ownership.snapshot_manifest" => {
+            // Snapshot manifest can be an absolute/relative path or a URL.
+            Ok(toml::Value::String(value.to_string()))
+        }
         "general.color" => {
             if !value.eq_ignore_ascii_case("true") && !value.eq_ignore_ascii_case("false") {
                 return Err(IdxError::InvalidInput(format!(
@@ -556,6 +561,24 @@ mod tests {
         assert!(super::validate_config_key_value("ownership.db_path", "data/ownership.db").is_ok());
     }
 
+    #[test]
+    fn validate_accepts_ownership_snapshot_manifest() {
+        assert!(
+            super::validate_config_key_value(
+                "ownership.snapshot_manifest",
+                "https://example.com/latest.json"
+            )
+            .is_ok()
+        );
+        assert!(
+            super::validate_config_key_value(
+                "ownership.snapshot_manifest",
+                "snapshots/latest.json"
+            )
+            .is_ok()
+        );
+    }
+
     #[test]
     fn validate_accepts_valid_color() {
         assert!(super::validate_config_key_value("general.color", "true").is_ok());

src/ownership/db.rs

@@ -128,8 +128,8 @@ pub fn ensure_schema(conn: &Connection) -> Result<(), IdxError> {
 }
 
 /// Open ownership database connection and run idempotent schema migration.
-pub fn open_db(_config: &IdxConfig) -> Result<Connection, IdxError> {
-    let db_path = resolve_db_path()?;
+pub fn open_db(config: &IdxConfig) -> Result<Connection, IdxError> {
+    let db_path = db_path(config)?;
 
     if let Some(parent) = db_path.parent() {
         fs::create_dir_all(parent).map_err(|e| IdxError::DatabaseError(e.to_string()))?;
@@ -1088,7 +1088,7 @@ pub fn compute_concentration(percentages_bps: &[i64]) -> ConcentrationMetrics {
     }
 }
 
-fn resolve_db_path() -> Result<PathBuf, IdxError> {
+pub fn db_path(_config: &IdxConfig) -> Result<PathBuf, IdxError> {
     if let Some(custom_path) = get_config_value("ownership.db_path")? {
         let trimmed = custom_path.trim();
         if !trimmed.is_empty() {

src/ownership/mod.rs

@@ -4,4 +4,5 @@ pub mod graph;
 pub mod parser;
 pub mod remote;
 pub mod search;
+pub mod snapshot;
 pub mod types;

src/ownership/parser.rs

@@ -37,11 +37,8 @@ const HOLDER_REGISTER_SCHEMA_MARKERS: &[&str] = &[
 ];
 const ANNOUNCEMENT_WRAPPER_SCHEMA_MARKERS: &[&str] =
     &["TEXT=\"PENGUMUMAN\"", "PT BURSA EFEK INDONESIA (BEI)"];
-const ABOVE_FIVE_SCHEMA_MARKERS: &[&str] = &[
-    "TEXT=\"INVS\"",
-    "REKENING TAMPUNGAN KSEI",
-    "CLOSED MEMBER-",
-];
+const ABOVE_FIVE_SCHEMA_MARKERS: &[&str] =
+    &["TEXT=\"INVS\"", "REKENING TAMPUNGAN KSEI", "CLOSED MEMBER-"];
 const INVESTOR_TYPE_SCHEMA_MARKERS: &[&str] = &[
     "TEXT=\"STOCK_CODE\"",
     "TEXT=\"NUMBER_OF_SHARES\"",

src/ownership/snapshot.rs

@@ -0,0 +1,762 @@
+use std::fs;
+use std::path::{Path, PathBuf};
+
+use chrono::NaiveDate;
+use rusqlite::{Connection, OpenFlags};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+
+use crate::config::get_config_value;
+use crate::error::IdxError;
+use crate::ownership::db;
+use crate::ownership::types::OwnershipRelease;
+
+pub const SNAPSHOT_MANIFEST_CONFIG_KEY: &str = "ownership.snapshot_manifest";
+pub const SNAPSHOT_MANIFEST_ENV: &str = "IDX_OWNERSHIP_SNAPSHOT_MANIFEST";
+pub const SNAPSHOT_MANIFEST_SCHEMA_VERSION: u32 = 1;
+
+const USER_AGENT: &str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36";
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct OwnershipSnapshotManifest {
+    pub schema_version: u32,
+    pub generated_at: String,
+    pub snapshot: OwnershipSnapshotArtifact,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct OwnershipSnapshotArtifact {
+    pub kind: String,
+    pub compression: String,
+    pub version: String,
+    pub download_url: String,
+    pub sqlite_sha256: String,
+    pub size_bytes: u64,
+    pub release_count: usize,
+    pub latest_as_of_date: NaiveDate,
+    pub latest_release_sha256: String,
+    pub latest_row_count: usize,
+    pub ticker_count: usize,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum OwnershipSyncAction {
+    Installed,
+    Updated,
+    Refreshed,
+    NoChange,
+    SkippedNewer,
+    SkippedDiverged,
+}
+
+#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
+pub struct OwnershipSyncResult {
+    pub action: OwnershipSyncAction,
+    pub manifest: String,
+    pub db_path: String,
+    pub snapshot_version: String,
+    pub latest_as_of_date: NaiveDate,
+    pub release_count: usize,
+    pub ticker_count: usize,
+    pub reason: String,
+}
+
+#[derive(Debug, Clone)]
+struct LocalSnapshotState {
+    latest_release: Option<OwnershipRelease>,
+    release_count: usize,
+}
+
+#[derive(Debug, Clone)]
+struct SyncDecision {
+    action: OwnershipSyncAction,
+    should_download: bool,
+    reason: String,
+}
+
+pub fn resolve_manifest_source(explicit: Option<&str>) -> Result<String, IdxError> {
+    if let Some(value) = explicit {
+        let trimmed = value.trim();
+        if !trimmed.is_empty() {
+            return Ok(trimmed.to_string());
+        }
+    }
+
+    if let Ok(value) = std::env::var(SNAPSHOT_MANIFEST_ENV) {
+        let trimmed = value.trim();
+        if !trimmed.is_empty() {
+            return Ok(trimmed.to_string());
+        }
+    }
+
+    if let Some(value) = get_config_value(SNAPSHOT_MANIFEST_CONFIG_KEY)? {
+        let trimmed = value.trim();
+        if !trimmed.is_empty() {
+            return Ok(trimmed.to_string());
+        }
+    }
+
+    Err(IdxError::InvalidInput(format!(
+        "ownership sync needs a snapshot manifest; pass `--manifest` or set `{SNAPSHOT_MANIFEST_CONFIG_KEY}` / `{SNAPSHOT_MANIFEST_ENV}`"
+    )))
+}
+
+pub fn fetch_manifest(source: &str) -> Result<OwnershipSnapshotManifest, IdxError> {
+    let raw = read_text(source, "ownership snapshot manifest")?;
+    parse_manifest(&raw)
+}
+
+pub fn parse_manifest(raw: &str) -> Result<OwnershipSnapshotManifest, IdxError> {
+    let manifest: OwnershipSnapshotManifest = serde_json::from_str(raw).map_err(|e| {
+        IdxError::ParseError(format!(
+            "failed to parse ownership snapshot manifest JSON: {e}"
+        ))
+    })?;
+    validate_manifest(&manifest)?;
+    Ok(manifest)
+}
+
+pub fn sync_snapshot(
+    manifest_source: &str,
+    db_path: &Path,
+    force: bool,
+) -> Result<OwnershipSyncResult, IdxError> {
+    let manifest = fetch_manifest(manifest_source)?;
+    let local_state = inspect_local_db(db_path)?;
+    let decision = build_sync_decision(local_state.as_ref(), &manifest.snapshot, force);
+
+    if !decision.should_download {
+        return Ok(OwnershipSyncResult {
+            action: decision.action,
+            manifest: manifest_source.to_string(),
+            db_path: db_path.display().to_string(),
+            snapshot_version: manifest.snapshot.version.clone(),
+            latest_as_of_date: manifest.snapshot.latest_as_of_date,
+            release_count: manifest.snapshot.release_count,
+            ticker_count: manifest.snapshot.ticker_count,
+            reason: decision.reason,
+        });
+    }
+
+    let bytes = read_bytes(&manifest.snapshot.download_url, "ownership snapshot SQLite")?;
+    validate_downloaded_bytes(&bytes, &manifest.snapshot)?;
+
+    let temp_path = build_temp_path(db_path);
+    if let Some(parent) = temp_path.parent()
+        && !parent.as_os_str().is_empty()
+    {
+        fs::create_dir_all(parent).map_err(|e| {
+            IdxError::Io(format!(
+                "failed to create snapshot temp directory {}: {e}",
+                parent.display()
+            ))
+        })?;
+    }
+    fs::write(&temp_path, &bytes).map_err(|e| {
+        IdxError::Io(format!(
+            "failed to write downloaded snapshot {}: {e}",
+            temp_path.display()
+        ))
+    })?;
+
+    let install_result = (|| -> Result<(), IdxError> {
+        validate_snapshot_db(&temp_path, &manifest.snapshot)?;
+        install_snapshot_file(&temp_path, db_path)
+    })();
+
+    if install_result.is_err() {
+        let _ = fs::remove_file(&temp_path);
+    }
+    install_result?;
+
+    Ok(OwnershipSyncResult {
+        action: decision.action,
+        manifest: manifest_source.to_string(),
+        db_path: db_path.display().to_string(),
+        snapshot_version: manifest.snapshot.version.clone(),
+        latest_as_of_date: manifest.snapshot.latest_as_of_date,
+        release_count: manifest.snapshot.release_count,
+        ticker_count: manifest.snapshot.ticker_count,
+        reason: decision.reason,
+    })
+}
+
+fn validate_manifest(manifest: &OwnershipSnapshotManifest) -> Result<(), IdxError> {
+    if manifest.schema_version != SNAPSHOT_MANIFEST_SCHEMA_VERSION {
+        return Err(IdxError::ParseError(format!(
+            "unsupported ownership snapshot manifest schema_version {}; expected {}",
+            manifest.schema_version, SNAPSHOT_MANIFEST_SCHEMA_VERSION
+        )));
+    }
+
+    if manifest.generated_at.trim().is_empty() {
+        return Err(IdxError::ParseError(
+            "ownership snapshot manifest is missing generated_at".to_string(),
+        ));
+    }
+
+    let snapshot = &manifest.snapshot;
+    if snapshot.kind.trim() != "sqlite" {
+        return Err(IdxError::Unsupported(format!(
+            "ownership snapshot kind `{}` is not supported; expected `sqlite`",
+            snapshot.kind
+        )));
+    }
+    if snapshot.compression.trim() != "none" {
+        return Err(IdxError::Unsupported(format!(
+            "ownership snapshot compression `{}` is not supported; expected `none`",
+            snapshot.compression
+        )));
+    }
+    if snapshot.version.trim().is_empty() {
+        return Err(IdxError::ParseError(
+            "ownership snapshot manifest is missing snapshot.version".to_string(),
+        ));
+    }
+    if snapshot.download_url.trim().is_empty() {
+        return Err(IdxError::ParseError(
+            "ownership snapshot manifest is missing snapshot.download_url".to_string(),
+        ));
+    }
+    if snapshot.size_bytes == 0 {
+        return Err(IdxError::ParseError(
+            "ownership snapshot manifest reported size_bytes=0".to_string(),
+        ));
+    }
+    if snapshot.release_count == 0 {
+        return Err(IdxError::ParseError(
+            "ownership snapshot manifest reported release_count=0".to_string(),
+        ));
+    }
+    if snapshot.latest_row_count == 0 {
+        return Err(IdxError::ParseError(
+            "ownership snapshot manifest reported latest_row_count=0".to_string(),
+        ));
+    }
+    if snapshot.ticker_count == 0 {
+        return Err(IdxError::ParseError(
+            "ownership snapshot manifest reported ticker_count=0".to_string(),
+        ));
+    }
+
+    validate_sha256_hex(&snapshot.sqlite_sha256, "snapshot.sqlite_sha256")?;
+    validate_sha256_hex(
+        &snapshot.latest_release_sha256,
+        "snapshot.latest_release_sha256",
+    )?;
+
+    Ok(())
+}
+
+fn validate_sha256_hex(value: &str, field_name: &str) -> Result<(), IdxError> {
+    let trimmed = value.trim();
+    if trimmed.len() != 64 || !trimmed.chars().all(|ch| ch.is_ascii_hexdigit()) {
+        return Err(IdxError::ParseError(format!(
+            "ownership snapshot manifest field `{field_name}` must be a 64-character hex sha256"
+        )));
+    }
+    Ok(())
+}
+
+fn inspect_local_db(db_path: &Path) -> Result<Option<LocalSnapshotState>, IdxError> {
+    if !db_path.exists() {
+        return Ok(None);
+    }
+
+    let conn = Connection::open(db_path).map_err(|e| {
+        IdxError::DatabaseError(format!("failed to open {}: {e}", db_path.display()))
+    })?;
+    db::ensure_schema(&conn)?;
+    let releases = db::query_releases(&conn)?;
+
+    Ok(Some(LocalSnapshotState {
+        latest_release: releases.first().cloned(),
+        release_count: releases.len(),
+    }))
+}
+
+fn build_sync_decision(
+    local: Option<&LocalSnapshotState>,
+    snapshot: &OwnershipSnapshotArtifact,
+    force: bool,
+) -> SyncDecision {
+    if force {
+        return SyncDecision {
+            action: if local.is_some() {
+                OwnershipSyncAction::Refreshed
+            } else {
+                OwnershipSyncAction::Installed
+            },
+            should_download: true,
+            reason: "force refresh requested".to_string(),
+        };
+    }
+
+    let Some(local) = local else {
+        return SyncDecision {
+            action: OwnershipSyncAction::Installed,
+            should_download: true,
+            reason: "local ownership database does not exist yet".to_string(),
+        };
+    };
+
+    let Some(latest) = &local.latest_release else {
+        return SyncDecision {
+            action: OwnershipSyncAction::Updated,
+            should_download: true,
+            reason: "local ownership database exists but has no imported releases".to_string(),
+        };
+    };
+
+    if latest.as_of_date > snapshot.latest_as_of_date {
+        return SyncDecision {
+            action: OwnershipSyncAction::SkippedNewer,
+            should_download: false,
+            reason: format!(
+                "local ownership database is newer than the published snapshot (local {} > snapshot {})",
+                latest.as_of_date.format("%Y-%m-%d"),
+                snapshot.latest_as_of_date.format("%Y-%m-%d")
+            ),
+        };
+    }
+
+    if latest.as_of_date < snapshot.latest_as_of_date {
+        return SyncDecision {
+            action: OwnershipSyncAction::Updated,
+            should_download: true,
+            reason: format!(
+                "published snapshot advances local ownership data from {} to {}",
+                latest.as_of_date.format("%Y-%m-%d"),
+                snapshot.latest_as_of_date.format("%Y-%m-%d")
+            ),
+        };
+    }
+
+    if latest.sha256 == snapshot.latest_release_sha256 {
+        if latest.row_count != snapshot.latest_row_count {
+            return SyncDecision {
+                action: OwnershipSyncAction::Updated,
+                should_download: true,
+                reason: "local ownership database has the same latest release date but different row_count metadata".to_string(),
+            };
+        }
+
+        if local.release_count < snapshot.release_count {
+            return SyncDecision {
+                action: OwnershipSyncAction::Updated,
+                should_download: true,
+                reason: format!(
+                    "local ownership database is missing historical releases (local {} < snapshot {})",
+                    local.release_count, snapshot.release_count
+                ),
+            };
+        }
+
+        if local.release_count == snapshot.release_count {
+            return SyncDecision {
+                action: OwnershipSyncAction::NoChange,
+                should_download: false,
+                reason: format!(
+                    "ownership snapshot already current at {} with {} release(s)",
+                    snapshot.latest_as_of_date.format("%Y-%m-%d"),
+                    snapshot.release_count
+                ),
+            };
+        }
+    }
+
+    SyncDecision {
+        action: OwnershipSyncAction::SkippedDiverged,
+        should_download: false,
+        reason: format!(
+            "local ownership database differs from the published snapshot for {}; use --force to replace it",
+            snapshot.latest_as_of_date.format("%Y-%m-%d")
+        ),
+    }
+}
+
+fn validate_downloaded_bytes(
+    bytes: &[u8],
+    snapshot: &OwnershipSnapshotArtifact,
+) -> Result<(), IdxError> {
+    let size_bytes = u64::try_from(bytes.len()).map_err(|e| {
+        IdxError::ParseError(format!("snapshot download too large to validate: {e}"))
+    })?;
+    if size_bytes != snapshot.size_bytes {
+        return Err(IdxError::ParseError(format!(
+            "ownership snapshot size mismatch: manifest expected {} bytes, downloaded {} bytes",
+            snapshot.size_bytes, size_bytes
+        )));
+    }
+
+    let actual_sha256 = sha256_hex(bytes);
+    if actual_sha256 != snapshot.sqlite_sha256 {
+        return Err(IdxError::ParseError(format!(
+            "ownership snapshot checksum mismatch: manifest expected {}, downloaded {}",
+            snapshot.sqlite_sha256, actual_sha256
+        )));
+    }
+
+    Ok(())
+}
+
+fn validate_snapshot_db(path: &Path, snapshot: &OwnershipSnapshotArtifact) -> Result<(), IdxError> {
+    let conn =
+        Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY).map_err(|e| {
+            IdxError::DatabaseError(format!(
+                "failed to open downloaded snapshot {}: {e}",
+                path.display()
+            ))
+        })?;
+
+    let quick_check: String = conn
+        .query_row("PRAGMA quick_check(1)", [], |row| row.get(0))
+        .map_err(|e| IdxError::DatabaseError(format!("snapshot quick_check failed: {e}")))?;
+    if quick_check.trim() != "ok" {
+        return Err(IdxError::DatabaseError(format!(
+            "snapshot quick_check failed: {quick_check}"
+        )));
+    }
+
+    let releases = db::query_releases(&conn)?;
+    if releases.len() != snapshot.release_count {
+        return Err(IdxError::ParseError(format!(
+            "downloaded snapshot release_count mismatch: manifest expected {}, sqlite has {}",
+            snapshot.release_count,
+            releases.len()
+        )));
+    }
+
+    let latest = releases.first().ok_or_else(|| {
+        IdxError::ParseError("downloaded snapshot sqlite had no ownership releases".to_string())
+    })?;
+
+    if latest.as_of_date != snapshot.latest_as_of_date {
+        return Err(IdxError::ParseError(format!(
+            "downloaded snapshot latest_as_of_date mismatch: manifest expected {}, sqlite has {}",
+            snapshot.latest_as_of_date.format("%Y-%m-%d"),
+            latest.as_of_date.format("%Y-%m-%d")
+        )));
+    }
+    if latest.sha256 != snapshot.latest_release_sha256 {
+        return Err(IdxError::ParseError(format!(
+            "downloaded snapshot latest_release_sha256 mismatch: manifest expected {}, sqlite has {}",
+            snapshot.latest_release_sha256, latest.sha256
+        )));
+    }
+    if latest.row_count != snapshot.latest_row_count {
+        return Err(IdxError::ParseError(format!(
+            "downloaded snapshot latest_row_count mismatch: manifest expected {}, sqlite has {}",
+            snapshot.latest_row_count, latest.row_count
+        )));
+    }
+
+    let ticker_count: i64 = conn
+        .query_row("SELECT COUNT(*) FROM tickers", [], |row| row.get(0))
+        .map_err(|e| IdxError::DatabaseError(format!("failed counting snapshot tickers: {e}")))?;
+    let ticker_count = usize::try_from(ticker_count)
+        .map_err(|e| IdxError::DatabaseError(format!("invalid snapshot ticker_count: {e}")))?;
+    if ticker_count != snapshot.ticker_count {
+        return Err(IdxError::ParseError(format!(
+            "downloaded snapshot ticker_count mismatch: manifest expected {}, sqlite has {}",
+            snapshot.ticker_count, ticker_count
+        )));
+    }
+
+    Ok(())
+}
+
+fn build_temp_path(db_path: &Path) -> PathBuf {
+    let file_name = db_path
+        .file_name()
+        .and_then(|name| name.to_str())
+        .unwrap_or("ownership.db");
+    let temp_name = format!(".{file_name}.sync-{}.tmp", fastrand::u64(..));
+    match db_path.parent() {
+        Some(parent) if !parent.as_os_str().is_empty() => parent.join(temp_name),
+        _ => PathBuf::from(temp_name),
+    }
+}
+
+fn install_snapshot_file(temp_path: &Path, db_path: &Path) -> Result<(), IdxError> {
+    if let Some(parent) = db_path.parent()
+        && !parent.as_os_str().is_empty()
+    {
+        fs::create_dir_all(parent).map_err(|e| {
+            IdxError::Io(format!(
+                "failed to create ownership database directory {}: {e}",
+                parent.display()
+            ))
+        })?;
+    }
+
+    let backup_path = build_backup_path(db_path);
+    if backup_path.exists() {
+        fs::remove_file(&backup_path).map_err(|e| {
+            IdxError::Io(format!(
+                "failed to remove stale backup {}: {e}",
+                backup_path.display()
+            ))
+        })?;
+    }
+
+    let had_existing = db_path.exists();
+    remove_sqlite_sidecars(db_path)?;
+
+    if had_existing {
+        fs::rename(db_path, &backup_path).map_err(|e| {
+            IdxError::Io(format!(
+                "failed to move existing ownership database {} to backup {}: {e}",
+                db_path.display(),
+                backup_path.display()
+            ))
+        })?;
+    }
+
+    if let Err(err) = fs::rename(temp_path, db_path) {
+        if had_existing {
+            let _ = fs::rename(&backup_path, db_path);
+        }
+        return Err(IdxError::Io(format!(
+            "failed to install ownership snapshot into {}: {err}",
+            db_path.display()
+        )));
+    }
+
+    if had_existing && backup_path.exists() {
+        fs::remove_file(&backup_path).map_err(|e| {
+            IdxError::Io(format!(
+                "failed to remove snapshot backup {}: {e}",
+                backup_path.display()
+            ))
+        })?;
+    }
+
+    remove_sqlite_sidecars(db_path)?;
+    Ok(())
+}
+
+fn build_backup_path(db_path: &Path) -> PathBuf {
+    let file_name = db_path
+        .file_name()
+        .and_then(|name| name.to_str())
+        .unwrap_or("ownership.db");
+    let backup_name = format!(".{file_name}.sync-backup");
+    match db_path.parent() {
+        Some(parent) if !parent.as_os_str().is_empty() => parent.join(backup_name),
+        _ => PathBuf::from(backup_name),
+    }
+}
+
+fn remove_sqlite_sidecars(db_path: &Path) -> Result<(), IdxError> {
+    for suffix in ["-wal", "-shm"] {
+        let file_name = format!(
+            "{}{suffix}",
+            db_path
+                .file_name()
+                .and_then(|name| name.to_str())
+                .unwrap_or("ownership.db")
+        );
+        let sidecar = match db_path.parent() {
+            Some(parent) if !parent.as_os_str().is_empty() => parent.join(&file_name),
+            _ => PathBuf::from(&file_name),
+        };
+
+        if sidecar.exists() {
+            fs::remove_file(&sidecar).map_err(|e| {
+                IdxError::Io(format!(
+                    "failed to remove SQLite sidecar {}: {e}",
+                    sidecar.display()
+                ))
+            })?;
+        }
+    }
+
+    Ok(())
+}
+
+fn sha256_hex(bytes: &[u8]) -> String {
+    let mut hasher = Sha256::new();
+    hasher.update(bytes);
+    let digest = hasher.finalize();
+    digest.iter().map(|byte| format!("{byte:02x}")).collect()
+}
+
+fn read_text(source: &str, context: &str) -> Result<String, IdxError> {
+    if is_http_source(source) {
+        let response = ureq::get(source)
+            .header("User-Agent", USER_AGENT)
+            .header("Accept", "application/json,text/plain;q=0.9,*/*;q=0.8")
+            .call()
+            .map_err(|e| IdxError::Http(format!("failed to fetch {context}: {e}")))?;
+        let mut body = response.into_body();
+        return body
+            .read_to_string()
+            .map_err(|e| IdxError::Http(format!("failed reading {context} body: {e}")));
+    }
+
+    let path = local_source_path(source);
+    fs::read_to_string(&path)
+        .map_err(|e| IdxError::Io(format!("failed to read {context} {}: {e}", path.display())))
+}
+
+fn read_bytes(source: &str, context: &str) -> Result<Vec<u8>, IdxError> {
+    if is_http_source(source) {
+        let response = ureq::get(source)
+            .header("User-Agent", USER_AGENT)
+            .header(
+                "Accept",
+                "application/octet-stream,application/x-sqlite3,*/*;q=0.8",
+            )
+            .call()
+            .map_err(|e| IdxError::Http(format!("failed to fetch {context}: {e}")))?;
+        let mut body = response.into_body();
+        return body
+            .read_to_vec()
+            .map_err(|e| IdxError::Http(format!("failed reading {context} body: {e}")));
+    }
+
+    let path = local_source_path(source);
+    fs::read(&path)
+        .map_err(|e| IdxError::Io(format!("failed to read {context} {}: {e}", path.display())))
+}
+
+fn is_http_source(source: &str) -> bool {
+    let normalized = source.trim().to_ascii_lowercase();
+    normalized.starts_with("http://") || normalized.starts_with("https://")
+}
+
+fn local_source_path(source: &str) -> PathBuf {
+    if let Some(path) = source.trim().strip_prefix("file://") {
+        return PathBuf::from(path);
+    }
+    PathBuf::from(source.trim())
+}
+
+#[cfg(test)]
+mod tests {
+    use chrono::NaiveDate;
+    use rusqlite::Connection;
+
+    use super::{
+        OwnershipSnapshotArtifact, OwnershipSnapshotManifest, OwnershipSyncAction,
+        SNAPSHOT_MANIFEST_SCHEMA_VERSION, build_sync_decision, parse_manifest,
+    };
+    use crate::ownership::db::{ensure_schema, insert_release};
+    use crate::ownership::types::OwnershipRelease;
+
+    #[test]
+    fn parse_manifest_rejects_unknown_schema_version() {
+        let raw = r#"{
+          "schema_version": 99,
+          "generated_at": "2026-03-31T12:00:00Z",
+          "snapshot": {
+            "kind": "sqlite",
+            "compression": "none",
+            "version": "2026-02-27",
+            "download_url": "/tmp/ownership.sqlite",
+            "sqlite_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+            "size_bytes": 123,
+            "release_count": 2,
+            "latest_as_of_date": "2026-02-27",
+            "latest_release_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
+            "latest_row_count": 100,
+            "ticker_count": 5
+          }
+        }"#;
+
+        let err = parse_manifest(raw).expect_err("schema_version should fail");
+        assert!(
+            err.to_string()
+                .contains("unsupported ownership snapshot manifest schema_version")
+        );
+    }
+
+    #[test]
+    fn build_sync_decision_updates_same_latest_release_when_history_is_incomplete() {
+        let snapshot = OwnershipSnapshotArtifact {
+            kind: "sqlite".to_string(),
+            compression: "none".to_string(),
+            version: "2026-02-27".to_string(),
+            download_url: "/tmp/ownership.sqlite".to_string(),
+            sqlite_sha256: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+                .to_string(),
+            size_bytes: 123,
+            release_count: 2,
+            latest_as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(),
+            latest_release_sha256:
+                "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".to_string(),
+            latest_row_count: 7261,
+            ticker_count: 955,
+        };
+        let local = super::LocalSnapshotState {
+            latest_release: Some(OwnershipRelease {
+                id: 1,
+                source_url: None,
+                sha256: snapshot.latest_release_sha256.clone(),
+                as_of_date: snapshot.latest_as_of_date,
+                row_count: snapshot.latest_row_count,
+                imported_at: 0,
+            }),
+            release_count: 1,
+        };
+
+        let decision = build_sync_decision(Some(&local), &snapshot, false);
+        assert_eq!(decision.action, OwnershipSyncAction::Updated);
+        assert!(decision.should_download);
+        assert!(decision.reason.contains("missing historical releases"));
+    }
+
+    #[test]
+    fn manifest_round_trip_parses_valid_payload() {
+        let manifest = OwnershipSnapshotManifest {
+            schema_version: SNAPSHOT_MANIFEST_SCHEMA_VERSION,
+            generated_at: "2026-03-31T12:00:00Z".to_string(),
+            snapshot: OwnershipSnapshotArtifact {
+                kind: "sqlite".to_string(),
+                compression: "none".to_string(),
+                version: "2026-02-27".to_string(),
+                download_url: "/tmp/ownership.sqlite".to_string(),
+                sqlite_sha256: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+                    .to_string(),
+                size_bytes: 123,
+                release_count: 1,
+                latest_as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(),
+                latest_release_sha256:
+                    "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".to_string(),
+                latest_row_count: 7261,
+                ticker_count: 955,
+            },
+        };
+
+        let raw = serde_json::to_string(&manifest).unwrap();
+        let parsed = parse_manifest(&raw).expect("valid manifest");
+        assert_eq!(parsed, manifest);
+    }
+
+    #[test]
+    fn ensure_schema_can_store_release_metadata_needed_for_snapshots() {
+        let conn = Connection::open_in_memory().unwrap();
+        ensure_schema(&conn).unwrap();
+        insert_release(
+            &conn,
+            &OwnershipRelease {
+                id: 0,
+                source_url: Some("https://example.com/ownership.sqlite".to_string()),
+                sha256: "abc".to_string(),
+                as_of_date: NaiveDate::from_ymd_opt(2026, 2, 27).unwrap(),
+                row_count: 1,
+                imported_at: 1,
+            },
+        )
+        .unwrap();
+
+        let release_count: i64 = conn
+            .query_row("SELECT COUNT(*) FROM ownership_releases", [], |row| {
+                row.get(0)
+            })
+            .unwrap();
+        assert_eq!(release_count, 1);
+    }
+}