//! 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 lyquor_api::call::CallParams;

use lyquor_hosting::{ImageResolver, Lyquid};

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

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

use tokio::sync::mpsc;

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: mpsc::Receiver<LyteLog>,

    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: mpsc::Receiver<LyteLog>) {

    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!("Availability sweep failed to list pending deployments: {e}");

                        continue;

                    }

                };

                for event in pending {

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

                }

            }

            log = log_stream.recv() => {

                let Some(log) = log else { 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!("Availability: image {image_digest} for {id}[{nth}] not retrievable yet: {e}");

            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!("Availability: failed to record image probe for {image_digest} at {id}[{nth}]: {e}");

            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) if available => tracing::info!("Availability: certified image {image_digest} for {id}[{nth}]"),

        Ok(true) => tracing::info!("Availability: certified unavailable image {image_digest} for {id}[{nth}]"),

        Ok(false) if available => {

            tracing::debug!("Availability: positive certificate for {image_digest} not formed (no quorum yet)");

        }

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

            "Availability: negative certificate for {image_digest} not submitted or not formed (deadline or quorum pending)"

        ),

        Err(e) => {

            tracing::warn!("Availability: {method} call for {image_digest} from {id}[{nth}] failed: {e}");

        }

    }

}

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

}