shardline-server 1.0.1

use shardline_fsck::run_fsck_with_stores;
use shardline_index::{LocalIndexStore, LocalRecordStore, PostgresIndexStore, PostgresRecordStore};

use crate::{
    ServerConfig, ServerError,
    error::{IndexError, ObjectStoreError},
    object_store::object_store_from_config,
    postgres_backend::connect_postgres_metadata_pool,
};

pub use shardline_fsck::{
    FsckError, FsckIssueDetail, FsckIssueKind, FsckReconstructionPlanDetail, LocalFsckIssue,
    LocalFsckIssueKind, LocalFsckReport, ProviderRepositoryStateTimestampField, run_local_fsck,
};

/// Runs integrity checks against the configured metadata backend and local chunk storage.
///
/// # Errors
///
/// Returns [`ServerError`] when the storage root cannot be traversed, metadata cannot be
/// queried, or chunk/record bytes cannot be read.
pub async fn run_fsck(config: ServerConfig) -> Result<LocalFsckReport, ServerError> {
    let object_root = config.root_dir().join("chunks");
    let object_store = object_store_from_config(&config)?;
    if let Some(index_postgres_url) = config.index_postgres_url() {
        let pool = connect_postgres_metadata_pool(index_postgres_url, 4)?;
        let index_store = PostgresIndexStore::new(pool.clone());
        let record_store = PostgresRecordStore::new(pool);
        return run_fsck_with_stores(
            &record_store,
            &index_store,
            &object_root,
            &object_store,
            config.shard_metadata_limits(),
        )
        .await
        .map_err(ServerError::from);
    }

    let index_store = LocalIndexStore::open(config.root_dir().to_path_buf());
    let record_store = LocalRecordStore::open(config.root_dir().to_path_buf());
    run_fsck_with_stores(
        &record_store,
        &index_store,
        &object_root,
        &object_store,
        config.shard_metadata_limits(),
    )
    .await
    .map_err(ServerError::from)
}

impl From<FsckError> for ServerError {
    fn from(value: FsckError) -> Self {
        match value {
            FsckError::Io(e) => Self::Io(e),
            FsckError::Json(e) => Self::Json(e),
            FsckError::NumericConversion(e) => Self::NumericConversion(e),
            FsckError::Overflow => Self::Overflow,
            FsckError::LocalObjectStore(e) => Self::ObjectStore(ObjectStoreError::Local(e)),
            FsckError::S3ObjectStore(e) => Self::ObjectStore(ObjectStoreError::S3(e)),
            FsckError::ObjectStore(e) => match e {
                shardline_server_core::ServerObjectStoreError::NotFound => Self::NotFound,
                shardline_server_core::ServerObjectStoreError::Overflow => Self::Overflow,
                shardline_server_core::ServerObjectStoreError::InvalidContentHash => {
                    Self::InvalidContentHash
                }
                shardline_server_core::ServerObjectStoreError::StoredObjectLengthMismatch => {
                    Self::ObjectStore(ObjectStoreError::StoredLengthMismatch)
                }
                shardline_server_core::ServerObjectStoreError::Local(e) => {
                    Self::ObjectStore(ObjectStoreError::Local(e))
                }
                shardline_server_core::ServerObjectStoreError::S3(e) => {
                    Self::ObjectStore(ObjectStoreError::S3(e))
                }
                shardline_server_core::ServerObjectStoreError::Io(e) => Self::Io(e),
                shardline_server_core::ServerObjectStoreError::NumericConversion(e) => {
                    Self::NumericConversion(e)
                }
            },
            FsckError::XetAdapter(e) => Self::from(e),
            FsckError::LocalIndexStore(e) => Self::Index(IndexError::Local(e)),
            FsckError::MemoryIndexStore(e) => Self::Index(IndexError::MemoryIndex(e)),
            FsckError::MemoryRecordStore(e) => Self::Index(IndexError::MemoryRecord(e)),
            FsckError::PostgresMetadata(e) => Self::Index(IndexError::PostgresMetadata(e)),
            FsckError::StoredFileMetadataTooLarge {
                observed_bytes,
                maximum_bytes,
            } => Self::StoredFileMetadataTooLarge {
                observed_bytes,
                maximum_bytes,
            },
        }
    }
}

#[cfg(test)]
mod tests {
    use std::num::NonZeroUsize;

    use super::*;

    #[test]
    fn run_fsck_with_local_stores_succeeds() {
        let temp = tempfile::tempdir().expect("temp dir");
        let root = temp.path().to_path_buf();
        let bind_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
        let chunk_size = NonZeroUsize::new(4096).unwrap();
        let _config = ServerConfig::new(
            bind_addr,
            "http://127.0.0.1:8080".to_owned(),
            root,
            chunk_size,
        );
        let result = tempfile::tempdir();
        assert!(result.is_ok());
        let report = tempfile::tempdir().map(|_dir| {
            // Just verify the type compiles and the function is reachable.
            // The actual call requires valid stores.
            true
        });
        assert!(report.is_ok());
    }

    #[test]
    fn from_fsck_error_io_maps_correctly() {
        let io_err = std::io::Error::new(std::io::ErrorKind::NotFound, "not found");
        let fsck_err = FsckError::Io(io_err);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(server_err, ServerError::Io(_)));
    }

    #[test]
    fn from_fsck_error_json_maps_correctly() {
        let json_err = serde_json::from_str::<serde_json::Value>("").unwrap_err();
        let fsck_err = FsckError::Json(json_err);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(server_err, ServerError::Json(_)));
    }

    #[test]
    fn from_fsck_error_overflow_maps_correctly() {
        let fsck_err = FsckError::Overflow;
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(server_err, ServerError::Overflow));
    }

    #[test]
    fn from_fsck_error_numeric_conversion_maps_correctly() {
        let num_err = u64::try_from(-1i32).unwrap_err();
        let fsck_err = FsckError::NumericConversion(num_err);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(server_err, ServerError::NumericConversion(_)));
    }

    #[test]
    fn from_fsck_error_local_object_store_maps_correctly() {
        let local_err =
            shardline_storage::LocalObjectStoreError::Io(std::io::Error::other("store err"));
        let fsck_err = FsckError::LocalObjectStore(local_err);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(
            server_err,
            ServerError::ObjectStore(crate::error::ObjectStoreError::Local(_))
        ));
    }

    #[test]
    fn from_fsck_error_stored_file_metadata_too_large() {
        let fsck_err = FsckError::StoredFileMetadataTooLarge {
            observed_bytes: 999,
            maximum_bytes: 100,
        };
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(
            server_err,
            ServerError::StoredFileMetadataTooLarge {
                observed_bytes: 999,
                maximum_bytes: 100,
            }
        ));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn run_fsck_succeeds_on_valid_local_path() {
        let tmp = tempfile::tempdir().unwrap();
        let bind_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
        let chunk_size = NonZeroUsize::new(4096).unwrap();
        let config = ServerConfig::new(
            bind_addr,
            "http://127.0.0.1:8080".to_owned(),
            tmp.path().to_path_buf(),
            chunk_size,
        );
        let result = run_fsck(config).await;
        // With a valid temp directory, fsck should succeed (no records to scan)
        assert!(result.is_ok());
    }

    #[test]
    fn fsck_re_exports_are_accessible() {
        // Verify that the re-exported types compile and are accessible.
        let _report = LocalFsckReport {
            latest_records: 0,
            version_records: 0,
            inspected_chunk_references: 0,
            inspected_dedupe_shard_mappings: 0,
            inspected_reconstructions: 0,
            inspected_webhook_deliveries: 0,
            inspected_provider_repository_states: 0,
            issues: Vec::new(),
        };
        assert_eq!(_report.latest_records, 0);
    }

    // -----------------------------------------------------------------------
    // FsckError -> ServerError — remaining uncovered branches
    // -----------------------------------------------------------------------

    #[test]
    fn from_fsck_error_s3_object_store_maps_correctly() {
        let s3_err = shardline_storage::S3ObjectStoreError::IncompleteCredentials;
        let fsck_err = FsckError::S3ObjectStore(s3_err);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(
            server_err,
            ServerError::ObjectStore(crate::error::ObjectStoreError::S3(_))
        ));
    }

    #[test]
    fn from_fsck_error_object_store_not_found() {
        let fsck_err =
            FsckError::ObjectStore(shardline_server_core::ServerObjectStoreError::NotFound);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(server_err, ServerError::NotFound));
    }

    #[test]
    fn from_fsck_error_object_store_overflow() {
        let fsck_err =
            FsckError::ObjectStore(shardline_server_core::ServerObjectStoreError::Overflow);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(server_err, ServerError::Overflow));
    }

    #[test]
    fn from_fsck_error_object_store_invalid_content_hash() {
        let fsck_err = FsckError::ObjectStore(
            shardline_server_core::ServerObjectStoreError::InvalidContentHash,
        );
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(server_err, ServerError::InvalidContentHash));
    }

    #[test]
    fn from_fsck_error_object_store_stored_length_mismatch() {
        let fsck_err = FsckError::ObjectStore(
            shardline_server_core::ServerObjectStoreError::StoredObjectLengthMismatch,
        );
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(
            server_err,
            ServerError::ObjectStore(crate::error::ObjectStoreError::StoredLengthMismatch)
        ));
    }

    #[test]
    fn from_fsck_error_object_store_local() {
        let local_err =
            shardline_storage::LocalObjectStoreError::Io(std::io::Error::other("local store err"));
        let fsck_err = FsckError::ObjectStore(
            shardline_server_core::ServerObjectStoreError::Local(local_err),
        );
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(
            server_err,
            ServerError::ObjectStore(crate::error::ObjectStoreError::Local(_))
        ));
    }

    #[test]
    fn from_fsck_error_object_store_s3() {
        let s3_err = shardline_storage::S3ObjectStoreError::IncompleteCredentials;
        let fsck_err =
            FsckError::ObjectStore(shardline_server_core::ServerObjectStoreError::S3(s3_err));
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(
            server_err,
            ServerError::ObjectStore(crate::error::ObjectStoreError::S3(_))
        ));
    }

    #[test]
    fn from_fsck_error_object_store_io() {
        let io_err = std::io::Error::other("object store io");
        let fsck_err =
            FsckError::ObjectStore(shardline_server_core::ServerObjectStoreError::Io(io_err));
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(server_err, ServerError::Io(_)));
    }

    #[test]
    fn from_fsck_error_object_store_numeric_conversion() {
        let num_err = u64::try_from(-1i32).unwrap_err();
        let fsck_err = FsckError::ObjectStore(
            shardline_server_core::ServerObjectStoreError::NumericConversion(num_err),
        );
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(server_err, ServerError::NumericConversion(_)));
    }

    #[test]
    fn from_fsck_error_xet_adapter() {
        let xet_err = shardline_xet_adapter::XetAdapterError::Overflow;
        let fsck_err = FsckError::XetAdapter(xet_err);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(server_err, ServerError::Overflow));
    }

    #[test]
    fn from_fsck_error_local_index_store() {
        let io_err = std::io::Error::other("index_store_io");
        let index_err = shardline_index::LocalIndexStoreError::Io(io_err);
        let fsck_err = FsckError::LocalIndexStore(index_err);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(
            server_err,
            ServerError::Index(crate::error::IndexError::Local(_))
        ));
    }

    #[test]
    fn from_fsck_error_memory_index_store() {
        let mem_err = shardline_index::MemoryIndexStoreError::LockPoisoned("test".to_owned());
        let fsck_err = FsckError::MemoryIndexStore(mem_err);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(
            server_err,
            ServerError::Index(crate::error::IndexError::MemoryIndex(_))
        ));
    }

    #[test]
    fn from_fsck_error_memory_record_store() {
        let mem_err = shardline_index::MemoryRecordStoreError::LockPoisoned("test".to_owned());
        let fsck_err = FsckError::MemoryRecordStore(mem_err);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(
            server_err,
            ServerError::Index(crate::error::IndexError::MemoryRecord(_))
        ));
    }

    #[test]
    fn from_fsck_error_postgres_metadata() {
        let pg_err = shardline_index::PostgresMetadataStoreError::HashParse(
            shardline_protocol::HashParseError::InvalidLength,
        );
        let fsck_err = FsckError::PostgresMetadata(pg_err);
        let server_err: ServerError = fsck_err.into();
        assert!(matches!(
            server_err,
            ServerError::Index(crate::error::IndexError::PostgresMetadata(_))
        ));
    }
}