mosaik 0.4.2

use {
	super::{
		super::{
			Functions,
			status::{CandidateInfo, CandidatesMap, HandlerConditions, When},
		},
		Caller,
		builder::CallerConfig,
	},
	crate::{
		discovery::{Catalog, Discovery, rtt::PeerInfo},
		network::LocalNode,
		primitives::{Datum, ShortFmtExt},
	},
	core::marker::PhantomData,
	std::sync::Arc,
	tokio::{sync::watch, task::JoinSet},
	tokio_util::sync::CancellationToken,
};

/// Worker task that maintains the set of eligible handler peers for one
/// caller instance.
///
/// The worker watches the discovery catalog and keeps a ranked candidate
/// snapshot published through a watch channel. Individual calls read the
/// snapshot, rank candidates by RTT, and connect per call — the worker
/// itself never holds connections.
pub(super) struct CallerWorker {
	/// The caller-specific configuration as assembled by
	/// `Network::functions().caller()`.
	config: Arc<CallerConfig>,

	/// A handle to the local node that is used to probe peer RTTs.
	local: LocalNode,

	/// The discovery system handle used to monitor known peers that serve
	/// the desired function.
	discovery: Discovery,

	/// Triggered when the caller handle is dropped.
	cancel: CancellationToken,

	/// The current snapshot of eligible handler peers. This value is
	/// observed by call attempts and availability conditions.
	candidates: watch::Sender<CandidatesMap>,

	/// Sets the online status of the caller when the configured online
	/// conditions are met.
	online: watch::Sender<bool>,

	/// A future that resolves when the caller meets the configured online
	/// conditions.
	online_when: HandlerConditions,

	/// Pending RTT probes for newly discovered handlers that have no RTT
	/// data yet. When a probe completes, the result is fed back to the
	/// discovery system, which triggers a catalog update that re-evaluates
	/// eligibility with RTT data available.
	rtt_probes: JoinSet<()>,
}

impl CallerWorker {
	/// Spawns a new caller worker task and returns the caller handle that
	/// can invoke the function and query availability.
	///
	/// The worker terminates when the returned caller is dropped.
	pub fn spawn<Req: Datum, Res: Datum, E: Datum>(
		config: CallerConfig,
		functions: &Functions,
	) -> Caller<Req, Res, E> {
		let config = Arc::new(config);
		let local = functions.local.clone();
		let cancel = local.termination().child_token();
		let candidates = watch::Sender::new(CandidatesMap::new());
		let online = watch::Sender::new(false);

		let when = When::new(candidates.subscribe(), online.subscribe());
		let online_when = (config.online_when)(when.available());

		online.send_replace(online_when.is_condition_met());

		let metrics_labels = [
			("function", config.function_id.short().to_string()),
			("network", local.network_id().short().to_string()),
		];

		let worker = Self {
			local: local.clone(),
			config: Arc::clone(&config),
			discovery: functions.discovery.clone(),
			cancel: cancel.clone(),
			candidates: candidates.clone(),
			online: online.clone(),
			online_when,
			rtt_probes: JoinSet::new(),
		};

		tokio::spawn(worker.run());

		Caller {
			config,
			local,
			discovery: functions.discovery.clone(),
			candidates: candidates.subscribe(),
			status: When::new(candidates.subscribe(), online.subscribe()),
			metrics_labels,
			_abort: cancel.drop_guard(),
			_marker: PhantomData,
		}
	}

	async fn run(mut self) {
		// Get a watch handle for the discovery catalog and mark it as changed
		// to trigger an initial handler lookup in the current state of the
		// catalog.
		let mut catalog = self.discovery.catalog_watch();
		catalog.mark_changed();

		loop {
			tokio::select! {
				// Triggered when the caller is dropped or the network is shutting
				// down.
				() = self.cancel.cancelled() => {
					break;
				}

				// Triggered when the online conditions for this caller are met.
				() = &mut self.online_when => {
					self.on_online();
				}

				// Triggered when new peers are discovered or existing peers are
				// updated.
				_ = catalog.changed() => {
					// mark the latest catalog snapshot as seen and trigger a scan
					let snapshot = catalog.borrow_and_update().clone();
					self.on_catalog_update(&snapshot);
				}

				// Drive RTT probe tasks (results are fed back to discovery,
				// triggering a catalog update that re-evaluates eligibility with
				// RTT data).
				Some(_) = self.rtt_probes.join_next() => {}
			}
		}
	}

	/// Handles updates to the discovery catalog by recomputing the eligible
	/// candidate set from scratch.
	fn on_catalog_update(&mut self, latest: &Catalog) {
		let mut eligible = CandidatesMap::new();

		let handlers = latest
			.peers()
			.filter(|peer| peer.functions().contains(&self.config.function_id));

		for handler in handlers {
			// If no RTT data exists for this handler, trigger an RTT ping probe
			// before considering it. The probe feeds its result back to the
			// discovery system, which triggers a catalog update that
			// re-evaluates with RTT data available. This avoids calling peers
			// with unknown latency and gives the ranking a complete picture.
			if self.discovery.rtt_tracker().get(handler.id()).is_none() {
				let local = self.local.clone();
				let discovery = self.discovery.clone();
				let addr = handler.address().clone();
				self.rtt_probes.spawn(async move {
					if let Ok((entry, rtt)) = local.ping(addr, None).await
						&& let Some(rtt) = rtt
					{
						discovery.rtt_tracker().record_sample(*entry.id(), rtt);
						discovery.feed(entry);
					}
				});
				continue;
			}

			let info = PeerInfo::from_tracker(handler, self.discovery.rtt_tracker());
			if !(self.config.require)(&info) {
				tracing::debug!(
					function_id = %self.config.function_id.short(),
					handler_id = %handler.id().short(),
					network = %handler.network_id().short(),
					"skipping ineligible handler"
				);
				continue;
			}

			// Validate the handler's tickets against all validators. Tickets
			// are re-validated against the freshest catalog entry before each
			// call attempt, so no expiry timers are needed here.
			if handler.validate_tickets(&self.config.handler_auth).is_err() {
				tracing::debug!(
					function_id = %self.config.function_id.short(),
					handler_id = %handler.id().short(),
					network = %handler.network_id().short(),
					"skipping unauthorized handler"
				);
				continue;
			}

			let rtt = info.rtt();
			eligible.insert(*handler.id(), CandidateInfo {
				entry: handler.clone(),
				rtt,
			});
		}

		self.candidates.send_replace(eligible);

		if !self.online_when.is_condition_met() {
			tracing::trace!(
				function_id = %self.config.function_id.short(),
				handlers = %self.candidates.borrow().len(),
				"caller is offline",
			);
			self.online.send_replace(false);
		}
	}

	/// Triggered when the online conditions for this caller are met.
	fn on_online(&self) {
		tracing::trace!(
			function_id = %self.config.function_id.short(),
			handlers = %self.candidates.borrow().len(),
			"caller is online",
		);

		self.online.send_if_modified(|status| {
			if *status {
				false
			} else {
				*status = true;
				true
			}
		});
	}
}