linkrs-api 0.1.0

//! Online migration drain fence shared by embedded and server callers.
//!
//! Bridges `linkrs_transaction::issue_request_drain` into the migration
//! engine's [`linkrs_migration::SchemaWriteFence`] hook so schema switches
//! run inside a bounded write stall while the data backfill stays online.

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

use linkrs_core::event_dispatch::EventSubscriptions;

/// [`linkrs_migration::SchemaWriteFence`] backed by the transaction
/// checkpoint gate.
pub struct CheckpointGateSchemaFence {
    gate: Arc<linkrs_transaction::CheckpointGate>,
    timeout: Duration,
    held: parking_lot::Mutex<Option<linkrs_transaction::MigrationDrainFence>>,
}

impl CheckpointGateSchemaFence {
    pub fn new(gate: Arc<linkrs_transaction::CheckpointGate>, timeout: Duration) -> Self {
        Self {
            gate,
            timeout,
            held: parking_lot::Mutex::new(None),
        }
    }
}

impl linkrs_migration::SchemaWriteFence for CheckpointGateSchemaFence {
    fn hold(&self) -> Result<(), linkrs_migration::MigrationError> {
        let fence = linkrs_transaction::maintenance::issue_request_drain(&self.gate, self.timeout)
            .map_err(|e| linkrs_migration::MigrationError::Lock(e.to_string()))?;
        *self.held.lock() = Some(fence);
        Ok(())
    }

    fn release(&self) {
        if let Some(fence) = self.held.lock().take() {
            fence.complete();
        }
    }
}

/// Single online execution entry shared by embedded, HTTP and gRPC callers.
///
/// Holds a drain fence only across schema-modifying steps (data-only plans
/// skip it entirely), runs the plan, and confirms the switch on success.
/// Concurrency is guarded by the engine's own per-target lock; no separate
/// begin guard is needed here. Metric recording and event-registry setup
/// stay with the callers, which own their respective state sources.
pub fn execute_online_migration<S>(
    storage: &mut S,
    plan: &linkrs_migration::MigrationPlan,
    config: &linkrs_migration::MigrationConfig,
    gate: &Arc<linkrs_transaction::CheckpointGate>,
    event_registry: Option<&Arc<EventSubscriptions<linkrs_migration::MigrationEvent>>>,
) -> Result<linkrs_migration::MigrationReport, linkrs_migration::MigrationError>
where
    S: linkrs_storage::StorageReader
        + linkrs_storage::StorageWriter
        + linkrs_storage::StorageSchemaOps
        + linkrs_storage::AutoCommitGroupOps
        + linkrs_storage::AutoCommitBatchOps
        + ?Sized,
{
    let fence = if plan.has_schema_modifying_steps() {
        Some(CheckpointGateSchemaFence::new(
            Arc::clone(gate),
            Duration::from_millis(config.drain_timeout_ms.max(1)),
        ))
    } else {
        None
    };
    let report = linkrs_migration::execute_migration_plan_with_options(
        storage,
        plan,
        linkrs_migration::ExecuteOptions {
            config: Some(config),
            event_registry,
            schema_fence: fence
                .as_ref()
                .map(|f| f as &dyn linkrs_migration::SchemaWriteFence),
            ..Default::default()
        },
    )?;
    if report.success {
        let (_, audit) =
            linkrs_transaction::maintenance::issue_confirm_switch(plan.version_range.to);
        log::info!("{audit}");
    }
    Ok(report)
}