//! Deployment image-availability certification worker.

//!

//! Registrations that enter `Pending` remain there until their content digest

//! is admitted or a deployment deadline receives a certified negative verdict.

//! This worker watches bartender's `AvailabilityPending` log topic (and periodically sweeps the

//! registry so restarts and missed events self-heal), pulls each referenced

//! image through the node's [`ImageResolver`], and proposes a digest-scoped

//! availability certificate. One certificate admits every deployment that

//! references the same immutable content.

use std::sync::Arc;

use std::time::Duration;

use futures::{Stream, StreamExt};

use lyquor_api::{call::CallParams, log::LogClass, store::KVStoreError};

use lyquor_hosting::{ImageResolver, Lyquid};

use lyquor_primitives::{Address, AvailabilityPendingEvent, B256, LyteLog, encode_by_fields};

use lyquor_vm::scheduler::{RunOptions, RunSource};

use tokio_util::sync::CancellationToken;

/// How often the worker re-checks the registry for deployments that are still

/// pending (bounded re-probe of unretrievable images).

const SWEEP_INTERVAL: Duration = Duration::from_secs(15);

/// Run the availability watcher until its task group is cancelled.

pub async fn run_availability_watcher(

    bartender: Lyquid, image_resolver: Arc<ImageResolver>,

    log_stream: impl Stream<Item = Result<LyteLog, KVStoreError>> + Unpin, shutdown: CancellationToken,

) {

    tokio::select! {

        biased;

        () = shutdown.cancelled() => {}

        () = watch(bartender, image_resolver, log_stream) => {}

    }

}

async fn watch(

    bartender: Lyquid, image_resolver: Arc<ImageResolver>,

    mut log_stream: impl Stream<Item = Result<LyteLog, KVStoreError>> + Unpin,

) {

    let mut sweep = tokio::time::interval(SWEEP_INTERVAL);

    sweep.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

    loop {

        tokio::select! {

            _ = sweep.tick() => {

                let pending = match call_bartender::<Vec<AvailabilityPendingEvent>>(

                    &bartender,

                    "get_pending_deployments",

                    encode_by_fields!().into(),

                )

                .await

                {

                    Ok(pending) => pending,

                    Err(e) => {

                        tracing::debug!(error = ?e, "availability sweep failed to list pending deployments");

                        continue;

                    }

                };

                for event in pending {

                    try_certify(&bartender, &image_resolver, &event).await;

                }

            }

            log = log_stream.next() => {

                let Some(log) = log else { break };

                let log = match log {

                    Ok(log) => log,

                    Err(error) => {

                        tracing::error!(class = %LogClass::CoreCorruption, error = ?error, "availability log stream failed; watcher stopped");

                        break;

                    }

                };

                let Some(event) = lyquor_primitives::decode_object::<AvailabilityPendingEvent>(&log.data) else {

                    continue;

                };

                try_certify(&bartender, &image_resolver, &event).await;

            }

        }

    }

}

/// Pull and verify one pending deployment's image, record the local result,

/// then propose the matching positive or negative certificate.

async fn try_certify(bartender: &Lyquid, image_resolver: &ImageResolver, event: &AvailabilityPendingEvent) {

    let AvailabilityPendingEvent {

        id,

        nth,

        image_digest,

        repo_hint,

    } = event;

    if let Some(hint) = repo_hint.as_deref() {

        image_resolver.remember_repository_hint(*image_digest, hint);

    }

    // `ensure_local` stores the pack content-addressed by digest, so success

    // means this node holds bytes matching the registered digest.

    let available = match image_resolver.ensure_local(*image_digest).await {

        Ok(()) => true,

        Err(e) => {

            tracing::debug!(

                lyquid_id = %id,

                deployment = nth,

                image_digest = %image_digest,

                error = ?e,

                "deployment image not retrievable yet"

            );

            false

        }

    };

    match call_bartender::<bool>(

        bartender,

        "note_image_probe",

        encode_by_fields!(image_digest: B256 = *image_digest, available: bool = available).into(),

    )

    .await

    {

        Ok(_) => {}

        Err(e) => {

            tracing::warn!(

                lyquid_id = %id,

                deployment = nth,

                image_digest = %image_digest,

                class = %LogClass::CoreRuntime,

                error = ?e,

                "failed to record deployment image probe"

            );

            return;

        }

    }

    let method = if available {

        "certify_availability"

    } else {

        "certify_unavailability"

    };

    // Every committee node proposes independently. Duplicate positive

    // proposals become no-ops after admission; negative proposals remain

    // deployment-scoped and are checked again at their sequenced position.

    match call_bartender::<bool>(

        bartender,

        method,

        encode_by_fields!(image_digest: B256 = *image_digest).into(),

    )

    .await

    {

        Ok(true) => tracing::info!(

            lyquid_id = %id,

            deployment = nth,

            image_digest = %image_digest,

            available,

            "deployment image availability certified"

        ),

        Ok(false) if available => {

            tracing::debug!(image_digest = %image_digest, available, "deployment image certificate awaiting quorum");

        }

        Ok(false) => tracing::debug!(

            image_digest = %image_digest,

            available,

            "deployment image certificate awaiting deadline or quorum"

        ),

        Err(e) => {

            tracing::warn!(

                lyquid_id = %id,

                deployment = nth,

                image_digest = %image_digest,

                method,

                class = %LogClass::CoreRuntime,

                error = ?e,

                "deployment image certification call failed"

            );

        }

    }

}

async fn call_bartender<T: for<'a> serde::Deserialize<'a> + Send + 'static>(

    bartender: &Lyquid, method: &str, input: lyquor_primitives::Bytes,

) -> Result<T, lyquor_vm::instance::Error> {

    let call = {

        let bar_ln = bartender.latest_number().await;

        bartender

            .call_instance_func_decoded(

                bar_ln,

                CallParams::builder()

                    .caller(Address::ZERO)

                    .method(method.into())

                    .input(input)

                    .build(),

                RunOptions::new(RunSource::InstanceCall),

            )

            .await

    };

    call.await

}