linkrs_server/

startup.rs

//! Server startup functions
//!
//! Orchestrates storage, sync, transaction manager, and graph service initialization.

use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

#[cfg(feature = "embedding")]
use llm_embedding::{EmbeddingConfig, EmbeddingService, OpenAICompatibleProvider};
#[cfg(feature = "vector")]
use log::warn;
use log::{error, info};
#[cfg(feature = "vector-qdrant")]
use vector_client::VectorManager;

use crate::config::Config;
use crate::storage::{
    GraphStorage, MetricsStorage, PersistenceConfig, PropertyGraphConfig, ResourceConfig,
    StorageCommitOps, StorageTtlPurgeOps, SyncWrapper,
};
use crate::{GraphService, HttpServer};
use linkrs_core::error::DBResult;
use linkrs_core::types::set_bcrypt_cost;
#[cfg(feature = "vector")]
use linkrs_sync::backend::VectorBackend;
use linkrs_transaction::{TransactionConfig, TransactionManager, TransactionManagerConfig};

/// Start the service using the user configuration directory.
pub async fn start_service() -> DBResult<()> {
    let (config, config_path) = match Config::load_user_config() {
        Ok(config) => {
            let path = Config::user_config_path().ok().filter(|p| p.exists());
            (config, path)
        }
        Err(e) => {
            error!("Failed to load user config, using default config: {}", e);
            (Config::default(), None)
        }
    };
    start_service_with_config_path(config, config_path).await
}

/// Start the service using the configuration object.
pub async fn start_service_with_config(config: Config) -> DBResult<()> {
    start_service_with_config_path(config, None).await
}

/// Start the service using the configuration object, retaining the config
/// file path so config-write endpoints can persist updates.
pub async fn start_service_with_config_path(
    config: Config,
    config_path: Option<PathBuf>,
) -> DBResult<()> {
    info!("Initializing Linkrs service...");

    config.validate().map_err(|error| {
        linkrs_core::DBError::validation(format!("Invalid configuration: {}", error))
    })?;

    let mut config = config;
    match config.resolve_bootstrap_password() {
        Ok(source) => info!(
            "Bootstrap account '{}' password source: {:?}",
            config.server.auth.default_username, source
        ),
        Err(error) => {
            return Err(linkrs_core::DBError::validation(format!(
                "Failed to resolve the bootstrap password: {}",
                error
            )));
        }
    }
    info!(
        "Listeners: http {}:{}, grpc {}:{}",
        config.http_bind_address(),
        config.http_port(),
        config.grpc_bind_address(),
        config.grpc_port()
    );

    // Apply the bcrypt cost factor before any password hashing happens.
    // The cost lives in a global atomic; hot updates via UPDATE CONFIGS
    // re-store it, so later calls take effect for subsequent hashes.
    set_bcrypt_cost(config.server.auth.bcrypt_cost);

    info!(
        "Log system has been initialized: {}/{}",
        config.log_dir(),
        config.log_basename()
    );

    // Create shared StatsManager for all components before wiring storage decorators.
    let slow_query_config = config.to_slow_query_config();
    let m = &config.monitoring;
    let stats_manager = Arc::new(
        linkrs_metrics::StatsManager::with_slow_query_logger(
            m.enabled,
            m.memory_cache_size,
            m.slow_query_threshold_ms * 1000,
            slow_query_config,
        )
        .expect("Failed to create StatsManager with slow query logger"),
    );
    stats_manager.set_timeseries_capacity(m.timeseries_retention_secs);
    stats_manager.set_histogram_max_samples(m.histogram_max_samples);

    let storage_path = PathBuf::from(config.storage_path());
    let mut persistence_config = PersistenceConfig::for_work_dir(&storage_path)
        .with_property_graph_config(property_graph_config_from_config(&config.storage));
    persistence_config.sync_policy = Some(config.storage.sync_mode.to_sync_policy());
    if config.storage.checkpoint_interval_secs > 0 {
        persistence_config.auto_checkpoint_interval =
            Duration::from_secs(config.storage.checkpoint_interval_secs);
    }
    let mut graph_storage =
        GraphStorage::open_with_persistence_config(storage_path, persistence_config)?;
    graph_storage = graph_storage.set_stats_manager(stats_manager.clone());
    let version_manager = graph_storage.version_manager();
    let inner_storage = Arc::new(MetricsStorage::new(graph_storage));
    info!(
        "Storage initialized (persistent mode at {}, metrics enabled)",
        config.storage_path()
    );

    // Build the shared vector backend from configuration (Local or Qdrant).
    #[cfg(feature = "vector")]
    let (vector_backend, local_engine_handle): (
        Option<VectorBackend>,
        Option<Arc<simvec::LocalVectorEngine>>,
    ) = if config.is_vector_enabled() {
        match config.vector_config().engine {
            linkrs_config::VectorEngineKind::Local => {
                let data_dir = config.vector_data_dir();
                match simvec::LocalVectorEngine::open(&data_dir) {
                    Ok(engine) => {
                        if let Some(hnsw) = linkrs_api::vector_config::local_hnsw_config(
                            &config.vector_config().local,
                        ) {
                            engine.set_default_hnsw_config(hnsw);
                        }
                        if let Some(ivf) = linkrs_api::vector_config::local_ivf_config(
                            &config.vector_config().local,
                        ) {
                            engine.set_default_ivf_config(ivf);
                        }
                        if let Some(quant) = linkrs_api::vector_config::local_quantization_config(
                            &config.vector_config().local,
                        ) {
                            engine.set_default_quantization_config(quant);
                        }
                        info!("Local vector engine initialized at {}", data_dir.display());
                        let arc = Arc::new(engine);
                        let handle = Arc::clone(&arc);
                        (Some(VectorBackend::from_local_arc(arc)), Some(handle))
                    }
                    Err(e) => {
                        warn!(
                            "Failed to open local vector engine at {}: {}. Vector search will be disabled.",
                            data_dir.display(),
                            e
                        );
                        (None, None)
                    }
                }
            }
            #[cfg(feature = "vector-qdrant")]
            linkrs_config::VectorEngineKind::Qdrant => {
                match VectorManager::new(config.vector_config().qdrant.clone()).await {
                    Ok(vm) => {
                        info!("VectorManager initialized");
                        (Some(VectorBackend::qdrant(Arc::new(vm))), None)
                    }
                    Err(e) => {
                        warn!(
                            "Failed to create VectorManager: {}. Vector search will be disabled.",
                            e
                        );
                        (None, None)
                    }
                }
            }
        }
    } else {
        (None, None)
    };
    #[cfg(not(feature = "vector"))]
    let _vector_backend = (None::<()>, None::<()>);

    // Forward local vector engine metrics into the shared StatsManager.
    #[cfg(feature = "vector")]
    if let Some(engine) = local_engine_handle {
        crate::vector_metrics::spawn_vector_metrics_sampler(engine, stats_manager.clone());
        info!("vector metrics sampling enabled");
    }

    // Forward remote (Qdrant) vector engine metrics into the shared StatsManager.
    #[cfg(feature = "vector-qdrant")]
    if let Some(manager) = vector_backend.as_ref().and_then(|b| b.as_qdrant_manager()) {
        let conn = &manager.config().connection;
        let http_port = conn.http_port.unwrap_or(6333);
        crate::vector_metrics::spawn_remote_vector_metrics_sampler(
            conn.host.clone(),
            http_port,
            conn.api_key.clone(),
            stats_manager.clone(),
        );
        info!(
            "remote vector metrics sampling enabled ({}:{})",
            conn.host, http_port
        );
    }

    let mut sync_manager = if config.fulltext.enabled || config.is_vector_enabled() {
        use linkrs_sync::SyncManager;

        let sync_manager = if config.fulltext.enabled {
            #[cfg(feature = "fulltext")]
            {
                use linkrs_fulltext::manager::FulltextIndexManager;

                let manager = Arc::new(
                    FulltextIndexManager::new(config.fulltext.clone())
                        .expect("Failed to create FulltextIndexManager"),
                );

                use linkrs_fulltext::SyncConfig;

                let sync_config = SyncConfig {
                    queue_size: config.fulltext.sync.queue_size,
                    commit_interval_ms: config.fulltext.sync.commit_interval_ms,
                    batch_size: config.fulltext.sync.batch_size,
                    failure_policy: config.fulltext.sync.failure_policy,
                };

                let batch_config = linkrs_sync::batch::BatchConfig::from(sync_config.clone());
                let sync_coordinator =
                    Arc::new(linkrs_sync::fulltext_sync::FulltextSyncCoordinator::new(
                        manager.clone(),
                        batch_config,
                    ));

                let sync_manager = SyncManager::with_sync_config(sync_coordinator, sync_config);

                // Attach vector coordinator if a backend is available
                #[cfg(feature = "vector")]
                let sync_manager = if let Some(backend) = &vector_backend {
                    attach_vector_coordinator(sync_manager, backend.clone(), &config).map_err(
                        |error| {
                            linkrs_core::DBError::storage(format!(
                                "Invalid vector embedding configuration: {}",
                                error
                            ))
                        },
                    )?
                } else {
                    sync_manager
                };

                Some(Arc::new(sync_manager))
            }
            #[cfg(not(feature = "fulltext"))]
            {
                let sync_manager = SyncManager::new_without_fulltext();

                // Attach vector coordinator if a backend is available
                #[cfg(feature = "vector")]
                let sync_manager = if let Some(backend) = &vector_backend {
                    attach_vector_coordinator(sync_manager, backend.clone(), &config).map_err(
                        |error| {
                            linkrs_core::DBError::storage(format!(
                                "Invalid vector embedding configuration: {}",
                                error
                            ))
                        },
                    )?
                } else {
                    sync_manager
                };

                Some(Arc::new(sync_manager))
            }
        } else {
            let sync_manager = SyncManager::new_without_fulltext();

            // Attach vector coordinator if a backend is available
            #[cfg(feature = "vector")]
            let sync_manager = if let Some(backend) = &vector_backend {
                attach_vector_coordinator(sync_manager, backend.clone(), &config).map_err(
                    |error| {
                        linkrs_core::DBError::storage(format!(
                            "Invalid vector embedding configuration: {}",
                            error
                        ))
                    },
                )?
            } else {
                sync_manager
            };

            Some(Arc::new(sync_manager))
        };

        if sync_manager.is_some() {
            info!("SyncManager initialized");
        }

        sync_manager
    } else {
        None
    };

    if let Some(manager) = sync_manager.as_mut().and_then(Arc::get_mut) {
        manager.set_stats_manager(stats_manager.clone());
        let outbox_path = PathBuf::from(config.storage_path()).join("outbox/outbox.sqlite");
        if let Err(error) = manager.configure_outbox(outbox_path) {
            return Err(linkrs_core::DBError::storage(format!(
                "Failed to initialize sync outbox: {}",
                error
            )));
        }
        if let Err(error) = manager.retry_outbox_sync() {
            error!("Initial outbox delivery will be retried later: {}", error);
        }
        let recovered = inner_storage
            .recover_outbox_projection(manager)
            .map_err(|error| {
                linkrs_core::DBError::storage(format!(
                    "Failed to recover the SQLite outbox projection: {}",
                    error
                ))
            })?;
        if recovered > 0 {
            info!("Recovered {} outbox intents from committed WAL", recovered);
        }
    }

    if let Some(manager) = sync_manager.as_ref() {
        manager.start().await.map_err(|error| {
            linkrs_core::DBError::storage(format!("Failed to start sync manager: {}", error))
        })?;
        // Fail stranded rebuild generations and collect rebuild scratch
        // state left by a previous process. Idempotent; a failure is logged
        // without blocking startup (recovery is re-runnable on demand).
        match manager.recover_stale_rebuilds().await {
            Ok(0) => {}
            Ok(failed) => info!(
                "Rebuild startup recovery failed {} stranded generation(s)",
                failed
            ),
            Err(error) => log::warn!(
                "Rebuild startup recovery reported an error: {}; continuing startup",
                error
            ),
        }
        // Retention loop (hourly by default, driven by OutboxRetentionConfig).
        #[cfg(feature = "vector")]
        {
            let retention = config.vector.retention.clone();
            if retention.enabled {
                let interval_secs = retention.prune_interval_secs;
                let grace = retention.grace_lsn_distance;
                let max_age = retention.max_applied_age_ms;
                info!(
                    "Outbox retention enabled interval={}s grace={} max_age_ms={}",
                    interval_secs, grace, max_age
                );
                let mgr = manager.clone();
                tokio::spawn(async move {
                    let interval = std::time::Duration::from_secs(interval_secs.max(60));
                    loop {
                        tokio::time::sleep(interval).await;
                        match mgr.run_retention_once(grace, max_age) {
                            Ok((pruned, archived, lsn)) if pruned + archived > 0 => {
                                info!(
                                    "Outbox retention pruned {} archived {} retention_lsn={}",
                                    pruned, archived, lsn
                                );
                            }
                            Ok(_) => {}
                            Err(e) => {
                                log::warn!("Outbox retention cycle failed: {}", e);
                            }
                        }
                    }
                });
            }
        }
    }

    let storage = if let Some(ref sync_manager) = sync_manager {
        let sync_storage =
            SyncWrapper::with_sync_manager((*inner_storage).clone(), sync_manager.clone());
        info!("Sync enabled for fulltext and vector indexes");
        Arc::new(sync_storage)
    } else {
        let sync_storage = SyncWrapper::new((*inner_storage).clone());
        Arc::new(sync_storage)
    };

    // Opt-in background TTL purge. Off by default so deployments keep the
    // explicit `PURGE EXPIRED` contract; a zero interval stays off too.
    if config.storage.ttl_purge_enabled && config.storage.ttl_purge_interval_secs > 0 {
        storage.start_ttl_purge(Duration::from_secs(config.storage.ttl_purge_interval_secs));
        info!(
            "Background TTL purge enabled (interval {}s)",
            config.storage.ttl_purge_interval_secs
        );
    }

    // Create a transaction manager
    let txn_config = TransactionManagerConfig {
        default_timeout: std::time::Duration::from_secs(config.transaction.default_timeout),
        max_concurrent_transactions: config.transaction.max_concurrent_transactions,
        auto_cleanup: true,
        commit_retry_attempts: 3,
        abort_retry_attempts: 3,
        txn_config: TransactionConfig {
            auto_commit: config.transaction.auto_commit,
            ..TransactionConfig::default()
        },
        ..TransactionManagerConfig::default()
    };

    let mut transaction_manager = TransactionManager::with_shared_version_manager(
        txn_config,
        stats_manager.clone(),
        version_manager,
    );
    if let Some(ref sync_manager) = sync_manager {
        transaction_manager = transaction_manager.with_sync_manager(sync_manager.clone());
    }
    transaction_manager = transaction_manager.with_commit_sink(storage.clone());
    let transaction_manager = Arc::new(transaction_manager);
    let _cleanup_task = transaction_manager.start_auto_cleanup_task();
    info!("Transaction manager initialized with StatsManager");

    // Re-drive durable-but-unfinalized commits left by a previous process.
    // Storage WAL replay already ran during `GraphStorage::open...` above,
    // so page-level redo is complete; this completes the transaction-level
    // half (storage finalization + commit-timestamp allocation) from the
    // sidecar file next to the WAL directory. A failure is logged without
    // blocking startup: recovery is re-runnable on demand.
    transaction_manager.set_recovery_sidecar_path(
        PathBuf::from(config.storage_path())
            .join("wal")
            .join("pending.sidecar"),
    );
    match transaction_manager.startup_recovery() {
        Ok(0) => {}
        Ok(recovered) => info!(
            "Transaction startup recovery completed {} commit(s)",
            recovered
        ),
        Err(error) => log::warn!(
            "Transaction startup recovery reported an error: {}; continuing startup",
            error
        ),
    }

    // Create GraphService with shared VectorBackend to avoid duplicate initialization
    #[cfg(feature = "vector")]
    let graph_service = if let Some(backend) = &vector_backend {
        GraphService::with_shared_vector_backend(
            config.clone(),
            storage.clone(),
            transaction_manager.clone(),
            stats_manager.clone(),
            backend.clone(),
        )
        .await
    } else {
        GraphService::new_with_transaction_manager_and_stats(
            config.clone(),
            storage.clone(),
            transaction_manager.clone(),
            stats_manager.clone(),
        )
        .await
    };

    #[cfg(not(feature = "vector"))]
    let graph_service = GraphService::new_with_transaction_manager_and_stats(
        config.clone(),
        storage.clone(),
        transaction_manager.clone(),
        stats_manager.clone(),
    )
    .await;
    info!("Graph service initialized with transaction management");

    // Inject StatsManager into FulltextIndexManager to enable search metrics
    #[cfg(feature = "fulltext")]
    if let Some(sync_api) = graph_service.sync_api() {
        let fulltext_manager = sync_api.sync_manager().fulltext_manager();
        let stats_manager = graph_service.get_stats_manager().clone();
        fulltext_manager.set_stats_manager(stats_manager);
        info!("StatsManager injected into FulltextIndexManager for search metrics");
    }

    // Create HTTP server
    let http_server = Arc::new(HttpServer::new(
        graph_service.clone(),
        Arc::new(parking_lot::RwLock::new((*storage).clone())),
        transaction_manager,
        &config,
        config_path,
    ));
    info!("HTTP server created");

    let http_enabled = config.server.http.enabled;
    let grpc_enabled = {
        #[cfg(feature = "grpc")]
        {
            config.server.grpc.enabled
        }
        #[cfg(not(feature = "grpc"))]
        {
            false
        }
    };
    if !http_enabled && !grpc_enabled {
        return Err(linkrs_core::DBError::validation(
            "at least one of http.enabled and grpc.enabled must be true",
        ));
    }

    if http_enabled {
        info!(
            "Starting HTTP server on {}:{}",
            config.http_bind_address(),
            config.http_port()
        );
    }
    if grpc_enabled {
        info!(
            "Starting gRPC server on {}:{}",
            config.grpc_bind_address(),
            config.grpc_port()
        );
    }

    #[cfg(feature = "grpc")]
    if let Err(e) = super::start_http_and_grpc_servers(http_server, &config).await {
        error!("HTTP/gRPC server error: {}", e);
    }
    #[cfg(not(feature = "grpc"))]
    if let Err(e) = super::start_http_server(http_server, &config).await {
        error!("HTTP server error: {}", e);
    }

    super::shutdown_signal().await;

    info!("Shutting down Linkrs service...");
    graph_service.shutdown();
    Ok(())
}

/// Resolve the effective embedding config: the backend-independent top
/// level wins, `[vector.qdrant.embedding]` overrides it when both are set.
#[cfg(feature = "embedding")]
fn resolve_effective_embedding(_config: &Config) -> Option<EmbeddingConfig> {
    let top = _config.vector_config().embedding.clone();
    #[cfg(feature = "vector-qdrant")]
    {
        if let Some(qdrant) = _config.vector_config().qdrant.embedding.clone() {
            return Some(qdrant);
        }
    }
    top
}

/// Build the shared document-side service plus the optional query-side
/// service carrying the query preprocessor override. Backend independent:
/// local and remote engines share the same wiring.
#[cfg(feature = "embedding")]
fn build_embedding_services(
    _config: &Config,
    auto_embed: bool,
) -> Result<
    (
        Option<Arc<linkrs_sync::vector_sync::EmbeddingService>>,
        Option<Arc<linkrs_sync::vector_sync::EmbeddingService>>,
    ),
    String,
> {
    let Some(doc_config) = resolve_effective_embedding(_config) else {
        if auto_embed {
            return Err(
                "vector.auto_embed_text requires [vector.embedding] with explicit dimension"
                    .to_string(),
            );
        }
        return Ok((None, None));
    };
    let build = |config: EmbeddingConfig| {
        OpenAICompatibleProvider::new(config)
            .map(|provider| {
                Arc::new(EmbeddingService::new(
                    provider,
                    linkrs_sync::vector_sync::EMBEDDING_BATCH_SIZE,
                ))
            })
            .map_err(|err| format!("Failed to create embedding service: {}", err))
    };
    let doc_service = match build(doc_config.clone()) {
        Ok(service) => service,
        Err(err) => {
            if auto_embed {
                return Err(err);
            }
            warn!("Failed to create embedding service: {}", err);
            return Ok((None, None));
        }
    };
    let query_service = match _config.vector_config().embedding_query_preprocessor.clone() {
        Some(preprocessor) => {
            let mut query_config = doc_config;
            query_config.preprocessor = preprocessor;
            match build(query_config) {
                Ok(service) => Some(service),
                Err(err) => {
                    if auto_embed {
                        return Err(err);
                    }
                    warn!("Failed to create query embedding service: {}", err);
                    None
                }
            }
        }
        None => None,
    };
    Ok((Some(doc_service), query_service))
}

/// Build the optional post-recall rerank service. Absent config disables
/// rerank; invalid config warns and disables. Startup never fails for rerank.
#[cfg(feature = "rerank")]
fn build_rerank_service(
    _config: &Config,
) -> Option<(
    Arc<llm_rerank::CohereRerankProvider>,
    linkrs_config::VectorRerankConfig,
)> {
    let rerank = _config.vector_config().rerank.clone()?;
    if let Err(err) = rerank.validate() {
        warn!(
            "Invalid vector rerank configuration, rerank disabled: {}",
            err
        );
        return None;
    }
    match llm_rerank::CohereRerankProvider::new(rerank.endpoint.clone()) {
        Ok(provider) => {
            info!("Vector rerank enabled (model {})", rerank.endpoint.model);
            Some((Arc::new(provider), rerank))
        }
        Err(err) => {
            warn!("Failed to create rerank service, rerank disabled: {}", err);
            None
        }
    }
}

/// Attach a vector sync coordinator backed by `backend` to a SyncManager.
#[cfg(feature = "vector")]
fn attach_vector_coordinator(
    sync_manager: linkrs_sync::SyncManager,
    backend: VectorBackend,
    _config: &Config,
) -> Result<linkrs_sync::SyncManager, String> {
    let handle = tokio::runtime::Handle::current();
    let auto_embed = _config.vector_config().auto_embed_text;
    #[cfg(feature = "embedding")]
    let (embedding_service, query_service) = build_embedding_services(_config, auto_embed)
        .map_err(|error| format!("Invalid vector embedding configuration: {}", error))?;
    #[cfg(feature = "embedding")]
    let vector_coordinator = {
        let coordinator = linkrs_sync::vector_sync::VectorSyncCoordinator::new(
            backend.clone(),
            embedding_service,
            handle,
        );
        let coordinator = match query_service {
            Some(query) => coordinator.with_query_embedding_service(query),
            None => coordinator,
        };
        #[cfg(feature = "rerank")]
        let coordinator = match build_rerank_service(_config) {
            Some((service, rerank)) => coordinator.with_rerank_service(service, rerank),
            None => coordinator,
        };
        Arc::new(coordinator)
    };
    #[cfg(not(feature = "embedding"))]
    let vector_coordinator = {
        let coordinator = linkrs_sync::vector_sync::VectorSyncCoordinator::new_without_embedding(
            backend.clone(),
            handle,
        );
        #[cfg(feature = "rerank")]
        let coordinator = match build_rerank_service(_config) {
            Some((service, rerank)) => coordinator.with_rerank_service(service, rerank),
            None => coordinator,
        };
        Arc::new(coordinator)
    };
    #[cfg(not(feature = "embedding"))]
    if auto_embed {
        return Err(
            "vector.auto_embed_text requires the embedding build with a configured service"
                .to_string(),
        );
    }
    {
        let granularity = _config.vector_config().collection.granularity;
        vector_coordinator.set_granularity(granularity);
        info!("Vector collection granularity: {:?}", granularity);
    }
    info!("Vector index sync enabled");
    let mut manager = sync_manager.with_vector_coordinator(vector_coordinator);
    manager.set_auto_embed_text(auto_embed);
    if auto_embed {
        info!("Vector write-time auto-embedding enabled");
    }
    // Backend-aware delivery policy: batch / lease / retries / concurrency
    // derived from `VectorConfig` so `manager.rs` does not branch on `is_local`.
    let policy = linkrs_sync::backend::BackendDeliveryPolicy::from_config(_config.vector_config());
    info!(
        "Vector delivery policy: {} batch={} lease_ms={} retries={} concurrency={}",
        policy.policy_name,
        policy.batch_size,
        policy.lease_ms,
        policy.max_retries,
        policy.max_concurrency
    );
    manager.configure_backend_policy(policy);
    Ok(manager)
}

fn property_graph_config_from_config(
    storage: &crate::config::StorageConfig,
) -> PropertyGraphConfig {
    let mut config = PropertyGraphConfig::default();
    let cache_memory = match usize::try_from(storage.max_memory_bytes) {
        Ok(value) => value.min(config.cache_memory),
        Err(_) => config.cache_memory,
    };
    config.cache_memory = cache_memory;
    config.resources = ResourceConfig {
        max_memory_bytes: storage.max_memory_bytes,
        index_memory_bytes: storage.index_memory_bytes,
        memory_soft_ratio: storage.memory_soft_ratio,
        memory_hard_ratio: storage.memory_hard_ratio,
        max_active_snapshots: storage.max_active_snapshots,
        max_snapshot_age: Duration::from_secs(storage.max_snapshot_age_secs),
        max_tombstones: storage.max_tombstones,
        max_tombstone_bytes: storage.max_tombstone_bytes,
        index_gc_batch: storage.index_gc_batch,
        operation_timeout: Duration::from_secs(storage.operation_timeout_secs),
        dirty_flush_operations: storage.dirty_flush_operations,
        dirty_flush_bytes: storage.dirty_flush_bytes,
        cache_ttl: (storage.cache_ttl_secs > 0)
            .then(|| Duration::from_secs(storage.cache_ttl_secs)),
        cache_tti: (storage.cache_tti_secs > 0)
            .then(|| Duration::from_secs(storage.cache_tti_secs)),
        index_pool_capacity_bytes: storage.index_pool_capacity_bytes,
        index_eviction_enabled: storage.index_eviction_enabled,
        index_eviction_high_ratio: storage.index_eviction_high_ratio,
        index_eviction_low_ratio: storage.index_eviction_low_ratio,
        ..config.resources
    };
    config
}

/// Execute a single query directly (for CLI / quick testing).
pub async fn execute_query(query_str: &str) -> DBResult<()> {
    info!("Executing query: {}", query_str);

    let config = crate::config::Config::default();
    let inner_storage = Arc::new(GraphStorage::new()?);

    let sync_storage = SyncWrapper::new((*inner_storage).clone());
    let storage = Arc::new(sync_storage);

    let graph_service =
        GraphService::<SyncWrapper<GraphStorage>>::new_for_test(config, storage).await;

    let session = match graph_service
        .get_session_manager()
        .create_session("anonymous".into(), "127.0.0.1".to_string())
        .await
    {
        Ok(session) => session,
        Err(e) => {
            error!("Failed to create session: {}", e);
            return Err(linkrs_core::error::DBError::from(
                crate::session::SessionError::manager_error(format!(
                    "Failed to create session: {}",
                    e
                )),
            ));
        }
    };

    let session_id = session.id();

    match graph_service.execute(session_id, query_str).await {
        Ok(result) => {
            info!("Query executed successfully: {:?}", result);
        }
        Err(e) => {
            error!("Query execution error: {}", e);
        }
    }

    Ok(())
}