//! Deployment image-availability certification worker.

//!

//! Deployments registered while an availability committee is active enter the

//! bartender registry as `Pending` and are not hosted anywhere. This worker

//! watches bartender's `AvailabilityPending` log topic (and periodically

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

//! pending deployment's image through the node's [`ImageResolver`] — which

//! content-verifies the bytes against the digest — and, on success, records

//! the verified digest in bartender's node-local instance state and proposes

//! an availability certificate. Once a threshold of committee nodes validates

//! the claim, the certified `attest_available` call flips the deployment

//! `Live` and hosting proceeds through the normal registration path.

use std::sync::Arc;

use std::time::Duration;

use lyquor_api::call::CallParams;

use lyquor_hosting::{ImageResolver, Lyquid};

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

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

use tokio::sync::mpsc;

use tokio_util::sync::CancellationToken;

use tokio_util::task::TaskTracker;

/// 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);

/// Background worker that certifies image availability for pending deployments.

pub struct AvailabilityWatcher {

    shutdown: CancellationToken,

    task_tracker: TaskTracker,

}

impl AvailabilityWatcher {

    /// Start the availability worker.

    pub fn start(bartender: Lyquid, image_resolver: Arc<ImageResolver>, log_stream: mpsc::Receiver<LyteLog>) -> Self {

        let shutdown = CancellationToken::new();

        let task_tracker = TaskTracker::new();

        let token = shutdown.clone();

        task_tracker.spawn(async move {

            tokio::select! {

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

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

            }

        });

        Self { shutdown, task_tracker }

    }

    /// Stop the worker and wait for it to finish.

    pub async fn shutdown(self) {

        self.shutdown.cancel();

        self.task_tracker.close();

        self.task_tracker.wait().await;

    }

}

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, then flag it and propose

/// the availability certificate. Failures are left for the next sweep.

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.

    if let Err(e) = image_resolver.ensure_local(*image_digest).await {

        tracing::debug!("Availability: image {image_digest} for {id}[{nth}] not retrievable yet: {e}");

        return;

    }

    match call_bartender::<bool>(

        bartender,

        "note_image_verified",

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

    )

    .await

    {

        Ok(_) => {}

        Err(e) => {

            tracing::warn!("Availability: failed to record verified image {image_digest} for {id}[{nth}]: {e}");

            return;

        }

    }

    // Every committee node proposes independently; duplicate certificates are

    // deduplicated by the certified call's nonce replay protection, and

    // `certify_availability` no-ops once the deployment is no longer Pending.

    match call_bartender::<bool>(

        bartender,

        "certify_availability",

        encode_by_fields!(id: LyquidID = *id, nth: u32 = *nth).into(),

    )

    .await

    {

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

        Ok(false) => {

            tracing::debug!("Availability: certificate for {id}[{nth}] not formed (no quorum yet or already settled)");

        }

        Err(e) => tracing::warn!("Availability: certification call for {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

}