gunnar-server 1.1.0

//! **gunnar-server registers ITS OWN plugin into vm1's monetize** (owner
//! 2026-09-25) — the metered build only.
//!
//! The metered gunnar medium carries gunnar's own signed monetize plugin
//! ([`embedded`], staged at build time: `build.rs::self_plugin`), built and
//! signed from the same commit as this binary — its `gunnar_build_ref` section
//! must equal [`gunnar_common::BUILD_REF`], or it is not offered at all
//! ([`OTHER_BUILD`]). The open build carries none and does none of this.
//!
//! **When**: once this box is the elected PRIMARY (the latest feed says so) and
//! vm1 has offered where its plugin door is (`offer.<vm1>.plugin_door`,
//! `<vm1 utility>:8475`), the box asks the door what it holds for `gunnar`
//! (`GET /register/gunnar`). If vm1 already holds this module, from this box,
//! with this box's ticket key, NOTHING is sent. Otherwise it sends the module,
//! its signature and the ticket key (`POST /register/gunnar`), over mutual TLS
//! with its OWN box certificate (the one the installer's CA issued it). So:
//! the first time it is elected — ADDED; every later boot at the same version —
//! nothing; a new version (a new module) — REPLACED; a twin that takes over —
//! its own registration (another box, another ticket key).
//!
//! **What comes back**: vm1's answer carries vm1's monetize verifying key,
//! which is applied here live ([`crate::feed::monetize_arrived`]) — the same
//! key the installer relays as `offer.<vm1>.monetize_pubkey`, so it can no
//! longer race it. vm1 writes the block itself (endpoint from this box's
//! utility address, grants from its template, prices the operator's).

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

use nordisk_pairing::bootstrap::Vm as _;
use nordisk_pairing::join::JoinReply;
use nordisk_pairing::Role;

/// The plugin this build carries (empty when none was staged).
pub mod embedded {
    include!(concat!(env!("OUT_DIR"), "/self_plugin.rs"));
}

/// gunnar's product id.
pub const PRODUCT: &str = "gunnar";
/// The custom section the plugin's build reference is stamped into.
pub const BUILD_REF_SECTION: &str = "gunnar_build_ref";
/// The refusal: the carried plugin is another build.
pub const OTHER_BUILD: &str = "plugin-other-build";
/// The refusal: this build carries no plugin.
pub const NO_PLUGIN: &str = "plugin-not-carried";
/// How often the loop looks again (a feed that makes this box primary, the
/// door's address arriving, a new ticket key).
const POLL: Duration = Duration::from_secs(30);

/// A line that does not change is said again every this many attempts (5 min at [`POLL`]).
pub const REPEAT_EVERY: u32 = 10;

/// **The same line again, with its count** — pure: `Some` every [`REPEAT_EVERY`]
/// attempts of an unchanged line, so a retry loop that keeps failing is never
/// silent (o-ops mock 2026-09-29: 16 failed attempts, one line).
pub fn said_again(same: u32, line: &str) -> Option<String> {
    (same > 0 && same % REPEAT_EVERY == 0).then(|| format!("{line} (still, {same} attempts)"))
}

/// **The carried plugin, if it is this build's own** — pure over its inputs.
pub fn own_plugin<'a>(module: &'a [u8], signature: &'a [u8], build_ref: &str) -> Result<plugin_door_wire::Plugin<'a>, String> {
    if module.is_empty() || signature.is_empty() {
        return Err(format!("REFUSED [{NO_PLUGIN}] this metered build carries no signed plugin (nothing was staged into .plugin-out when it was built) — vm1's monetize will not know gunnar until the operator inserts it"));
    }
    let stamp = plugin_door_wire::custom_section(module, BUILD_REF_SECTION).map(|s| String::from_utf8_lossy(s).to_string());
    match stamp {
        Some(s) if s == build_ref => Ok(plugin_door_wire::Plugin { module, signature }),
        Some(s) => Err(format!("REFUSED [{OTHER_BUILD}] the carried plugin is build `{s}`, this server is `{build_ref}` — a plugin is registered only by the server of its own commit")),
        None => Err(format!("REFUSED [{OTHER_BUILD}] the carried plugin has no `{BUILD_REF_SECTION}` section — it is not gunnar's own build")),
    }
}

/// **Where to register, if this box should** — pure: this box is the primary
/// the latest feed names, and vm1 offered its door. `Err` says why not (yet).
pub fn target(reply: Option<&JoinReply>) -> Result<(SocketAddr, String), String> {
    let reply = reply.ok_or("not fed yet")?;
    if reply.role != Some(Role::Primary) {
        return Err(format!("this box is {:?}, not the primary: only the primary registers gunnar's plugin", reply.role));
    }
    let at = nordisk_pairing::feedbus::offered(&reply.feed, plugin_door_wire::OFFER_PLUGIN_DOOR).ok_or("vm1 has not offered its plugin door (`plugin_door`) — an open front, or not relayed yet")?;
    let door: SocketAddr = at.parse().map_err(|e| format!("vm1 offered plugin_door `{at}`, which is not host:port: {e}"))?;
    Ok((door, reply.me.name.clone()))
}

/// **The last attempt, for `Admin.Topology`** (2026-10-04): where vm1's
/// plugin door is, whether the last attempt took, and its own line — the
/// evidence that the monetize this box registers with is there.
static LAST: std::sync::Mutex<Option<gunnar_grpc::topology::Register>> = std::sync::Mutex::new(None);

/// The last attempt, or `None` when this box has not tried (not the primary,
/// vm1 offered no door, no plugin carried).
pub fn last() -> Option<gunnar_grpc::topology::Register> {
    LAST.lock().ok().and_then(|l| l.clone())
}

fn remember(door: SocketAddr, ok: bool, line: &str) {
    if let Ok(mut last) = LAST.lock() {
        *last = Some(gunnar_grpc::topology::Register { door: door.to_string(), ok, line: line.to_owned(), at_unix_ms: estate_clock::now_unix_ms() });
    }
}

/// What one attempt concluded — the key that makes the next attempt a no-op.
#[derive(Debug, Clone, PartialEq, Eq)]
struct Registered {
    door: SocketAddr,
    name: String,
    digest: String,
    ticket: Option<String>,
}

/// **One attempt**: ask vm1, send only if it holds something else. Answers
/// the line to say. Applies vm1's monetize key when the answer carries it.
pub async fn once(door: SocketAddr, me: &plugin_door_wire::Me<'_>, plugin: plugin_door_wire::Plugin<'_>) -> Result<String, String> {
    let done = plugin_door_wire::register_if_changed(door, me, PRODUCT, plugin).await?;
    if let Some(pk) = done.monetize_pubkey() {
        let _ = crate::feed::monetize_arrived(Some(pk));
    }
    match done {
        plugin_door_wire::Done::Unchanged(_) => Ok(format!("vm1 ({door}) already holds this build's plugin from `{}` — nothing sent", me.name)),
        plugin_door_wire::Done::Sent { answer, .. } if answer.taken() => Ok(format!("{} — {}", answer.outcome, answer.line)),
        plugin_door_wire::Done::Sent { status, answer } => Err(format!("vm1 answered {status}: {}", answer.line)),
    }
}

/// **Run for the life of the process** (metered, paired boxes): on its own
/// thread; returns at once.
pub fn start(installed: Arc<crate::installer_listener::Installed>, stores: Arc<crate::stores::ServerStores>) {
    let plugin = match own_plugin(embedded::MODULE, embedded::SIGNATURE, gunnar_common::BUILD_REF) {
        Ok(p) => p,
        Err(why) => {
            tracing::warn!(event = "plugin.self_register", "{why}");
            return;
        }
    };
    let spawned = std::thread::Builder::new().name("self-register".into()).spawn(move || {
        let rt = match tokio::runtime::Builder::new_current_thread().enable_all().build() {
            Ok(r) => r,
            Err(e) => {
                tracing::error!(error = %e, "plugin: no runtime — gunnar's plugin is not registered");
                return;
            }
        };
        let digest = plugin_door_wire::digest(plugin.module);
        let mut done: Option<Registered> = None;
        let mut said = String::new();
        let mut same = 0u32;
        let mut say = |line: String, warn: bool| {
            let out = if line != said {
                same = 1;
                Some(line.clone())
            } else {
                same += 1;
                said_again(same, &line)
            };
            if let Some(out) = out {
                if warn {
                    tracing::warn!(event = "plugin.self_register", "plugin: {out}");
                } else {
                    tracing::info!(event = "plugin.self_register", "plugin: {out}");
                }
            }
            said = line;
        };
        loop {
            let reply = installed.latest();
            match target(reply.as_ref()) {
                Err(why) => say(why, false),
                Ok((door, name)) => {
                    let ticket = match stores.actor_ticket_key() {
                        Ok(k) => Some(k.public_key_hex().to_owned()),
                        Err(e) => {
                            say(format!("the ticket key cannot be read yet: {e:#}"), true);
                            None
                        }
                    };
                    let key = Registered { door, name: name.clone(), digest: digest.clone(), ticket: ticket.clone() };
                    if ticket.is_some() && done.as_ref() != Some(&key) {
                        let identity = installed.identity();
                        if !identity.cert_pem.contains("BEGIN CERTIFICATE") || identity.cert_pem == installed.bootstrap_cert_pem() {
                            say("this box holds no certificate of its own yet — waiting for the installer's".into(), false);
                        } else {
                            let me = plugin_door_wire::Me { name: &name, identity: &identity, ca_pem: installed.installer_ca_pem(), ticket_key: ticket.as_deref() };
                            match rt.block_on(once(door, &me, plugin)) {
                                Ok(line) => {
                                    remember(door, true, &line);
                                    say(line, false);
                                    done = Some(key);
                                }
                                Err(e) => {
                                    let line = format!("registering gunnar's plugin at {door}: {e} — trying again");
                                    remember(door, false, &line);
                                    say(line, true)
                                }
                            }
                        }
                    }
                }
            }
            std::thread::sleep(POLL);
        }
    });
    if let Err(e) = spawned {
        tracing::error!(error = %e, "plugin: the self-registration thread could not start");
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_failure_that_does_not_change_is_said_again_with_its_count() {
        // o-ops mock 2026-09-29: main retried every 30 s for 8 minutes and said it ONCE
        // ("no HTTP answer" at 19:59:27, then nothing) — the retries were invisible.
        assert_eq!(said_again(1, "x"), None, "the first time is said by the change itself");
        assert_eq!(said_again(REPEAT_EVERY - 1, "x"), None);
        let again = said_again(REPEAT_EVERY, "no HTTP answer").unwrap();
        assert!(again.contains("no HTTP answer") && again.contains(&REPEAT_EVERY.to_string()), "{again}");
        assert!(said_again(2 * REPEAT_EVERY, "x").is_some());
        assert!(REPEAT_EVERY * POLL.as_secs() as u32 <= 300, "said at least every 5 minutes");
    }
    use nordisk_pairing::join::{Assigned, Peer};
    use std::collections::BTreeMap;

    fn stamped(build_ref: &str) -> Vec<u8> {
        let mut m = b"\0asm\x01\0\0\0".to_vec();
        let name = BUILD_REF_SECTION.as_bytes();
        let body_len = 1 + name.len() + build_ref.len();
        m.push(0);
        m.push(body_len as u8);
        m.push(name.len() as u8);
        m.extend(name);
        m.extend(build_ref.as_bytes());
        m
    }

    #[test]
    fn only_this_builds_own_signed_plugin_is_offered() {
        let m = stamped("v1.4.0-9-gabc1234");
        assert!(own_plugin(&m, b"sig", "v1.4.0-9-gabc1234").is_ok());
        assert!(own_plugin(&m, b"sig", "v1.4.0-10-gdef5678").unwrap_err().contains("[plugin-other-build]"));
        assert!(own_plugin(b"\0asm\x01\0\0\0", b"sig", "v1").unwrap_err().contains("[plugin-other-build]"), "no stamp");
        assert!(own_plugin(&[], &[], "v1").unwrap_err().contains("[plugin-not-carried]"));
        assert!(own_plugin(&m, &[], "v1.4.0-9-gabc1234").unwrap_err().contains("[plugin-not-carried]"), "unsigned is not carried");
    }

    fn reply(role: Option<Role>, door: Option<&str>) -> JoinReply {
        let mut feed = BTreeMap::new();
        if let Some(d) = door {
            feed.insert("offer.vm1.plugin_door".to_string(), d.to_string());
        }
        JoinReply {
            estate: "gunnar.test".into(),
            me: Assigned { name: "vm2".into(), overlay: [10, 99, 0, 2].into() },
            cert_pem: None,
            role,
            primary_overlay: None,
            primary_utility: None,
            peers: Vec::<Peer>::new(),
            feed,
        }
    }

    #[test]
    fn only_the_primary_registers_and_only_where_vm1_offered_its_door() {
        assert_eq!(target(Some(&reply(Some(Role::Primary), Some("10.13.8.1:8475")))).unwrap(), ("10.13.8.1:8475".parse().unwrap(), "vm2".to_string()));
        assert!(target(Some(&reply(Some(Role::Twin), Some("10.13.8.1:8475")))).unwrap_err().contains("not the primary"));
        assert!(target(Some(&reply(None, Some("10.13.8.1:8475")))).is_err(), "not elected yet");
        assert!(target(Some(&reply(Some(Role::Primary), None))).unwrap_err().contains("plugin_door"));
        assert!(target(Some(&reply(Some(Role::Primary), Some("vm1")))).unwrap_err().contains("not host:port"));
        assert!(target(None).is_err());
    }
}