//! kcp event-log resources (ADR 0005):
//! `kcp-events://watch/{workspace}/{resource}{?namespace,labelSelector,fieldSelector,kcp}`.
//!
//! One watcher per URI, shared by all subscribers. It seeds known objects
//! with one list, then streams `kubectl get -w --output-watch-events -o json`
//! (pushed by the API server; no polling), dedupes, keeps a 256-event ring
//! with `seq`, and wakes subscribers. When the stream ends it backs off,
//! re-lists to reconcile changes missed in the gap, and watches again.

use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use serde_json::{json, Value};
use tokio::io::AsyncReadExt;
use tokio::sync::broadcast;
use tokio::task::JoinHandle;

use rmcp::model::{
    ListResourceTemplatesResult, ReadResourceResponse, ReadResourceResult, ResourceContents,
    ResourceTemplate, ServerCapabilities, ServerConfig, SubscriptionFilter,
};
use rmcp::service::SubscriptionContext;
use rmcp::ErrorData;

use super::service::KcpServiceHandle;
use super::tools::append_namespace;
use crate::resources::EVENT_LOG_MIME;

pub const SCHEME: &str = "kcp-events://watch/";
pub const TEMPLATE: &str =
    "kcp-events://watch/{workspace}/{resource}{?namespace,labelSelector,fieldSelector,kcp}";
pub const DESCRIPTION: &str = "Kubernetes watch events for one resource in a workspace (event-log resource; pushed, no polling). {resource} is a plural name from get_api_resources (e.g. computevms). namespace=* watches every namespace. Events: {\"seq\", \"type\": ADDED|MODIFIED|DELETED|ERROR, \"object\", \"old\" (MODIFIED: previous object)}.\nExample: await watch_start(name=\"cm\", uri=\"kcp-events://watch/root:app/configmaps?namespace=*\")";

pub const API_SCHEME: &str = "kcp-api://";
pub const API_TEMPLATE: &str = "kcp-api://{workspace}/resources{?kcp}";
pub const API_DESCRIPTION: &str = "The workspace's API resources (same JSON as get_api_resources): names to use in kcp-events URIs and tools. Read-only; not watchable.\nExample: kinds = await read_resource(uri=\"kcp-api://root/resources\")";

/// `kcp-api://{workspace}/resources{?kcp}` → (workspace, kcp).
pub fn parse_api_uri(uri: &str) -> Result<(String, Option<String>), String> {
    let rest = uri
        .strip_prefix(API_SCHEME)
        .ok_or_else(|| format!("not a kcp api URI (expected {API_TEMPLATE}): {uri}"))?;
    let (path, query) = rest.split_once('?').unwrap_or((rest, ""));
    let Some(workspace) = path.strip_suffix("/resources").filter(|w| !w.is_empty()) else {
        return Err(format!("expected {API_TEMPLATE}, got {uri}"));
    };
    let mut kcp = None;
    for pair in query.split('&').filter(|pair| !pair.is_empty()) {
        match pair.split_once('=') {
            Some(("kcp", value)) => kcp = Some(decode(value)),
            _ => return Err(format!("unknown kcp api parameter {pair:?}")),
        }
    }
    Ok((decode(workspace), kcp))
}

const RING: usize = 256;
const MAX_BACKOFF: Duration = Duration::from_secs(30);
const HEALTHY_STREAM: Duration = Duration::from_secs(30);
const MAX_FAILURES: u32 = 5;
/// Longest a first read waits for the watcher's seed list.
const SEED_WAIT: Duration = Duration::from_secs(30);

/// A parsed watch URI.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct WatchTarget {
    pub workspace: String,
    pub resource: String,
    pub namespace: Option<String>,
    pub label_selector: Option<String>,
    pub field_selector: Option<String>,
    pub kcp: Option<String>,
}

impl WatchTarget {
    pub fn parse(uri: &str) -> Result<Self, String> {
        let rest = uri
            .strip_prefix(SCHEME)
            .ok_or_else(|| format!("not a kcp watch URI (expected {TEMPLATE}): {uri}"))?;
        let (path, query) = rest.split_once('?').unwrap_or((rest, ""));
        let mut segments = path.split('/').filter(|segment| !segment.is_empty());
        let (Some(workspace), Some(resource), None) =
            (segments.next(), segments.next(), segments.next())
        else {
            return Err(format!("expected {TEMPLATE}, got {uri}"));
        };
        let mut target = Self {
            workspace: decode(workspace),
            resource: decode(resource),
            namespace: None,
            label_selector: None,
            field_selector: None,
            kcp: None,
        };
        for pair in query.split('&').filter(|pair| !pair.is_empty()) {
            let (key, value) = pair.split_once('=').unwrap_or((pair, ""));
            let value = Some(decode(value));
            match key {
                "namespace" => target.namespace = value,
                "labelSelector" => target.label_selector = value,
                "fieldSelector" => target.field_selector = value,
                "kcp" => target.kcp = value,
                other => return Err(format!("unknown kcp watch parameter {other:?}")),
            }
        }
        Ok(target)
    }

    /// `kubectl get` arguments shared by the seed list and the watch.
    fn selection(&self) -> Vec<String> {
        let mut args = vec!["get".to_string(), self.resource.clone()];
        append_namespace(&mut args, self.namespace.as_deref(), false);
        if let Some(selector) = &self.label_selector {
            args.push("-l".to_string());
            args.push(selector.clone());
        }
        if let Some(selector) = &self.field_selector {
            args.push("--field-selector".to_string());
            args.push(selector.clone());
        }
        args
    }
}

/// Minimal percent-decoding for URI components.
fn decode(text: &str) -> String {
    let bytes = text.as_bytes();
    let mut out = Vec::with_capacity(bytes.len());
    let mut index = 0;
    while index < bytes.len() {
        if bytes[index] == b'%' && index + 2 < bytes.len() {
            if let Ok(byte) = u8::from_str_radix(&text[index + 1..index + 3], 16) {
                out.push(byte);
                index += 3;
                continue;
            }
        }
        out.push(if bytes[index] == b'+' {
            b' '
        } else {
            bytes[index]
        });
        index += 1;
    }
    String::from_utf8_lossy(&out).into_owned()
}

/// Incremental parser for concatenated JSON values (kubectl watch output).
#[derive(Default)]
pub struct JsonStream {
    buffer: Vec<u8>,
}

impl JsonStream {
    pub fn feed(&mut self, bytes: &[u8]) -> Vec<Value> {
        self.buffer.extend_from_slice(bytes);
        let mut values = Vec::new();
        let mut consumed = 0;
        loop {
            let mut stream =
                serde_json::Deserializer::from_slice(&self.buffer[consumed..]).into_iter::<Value>();
            match stream.next() {
                Some(Ok(value)) => {
                    consumed += stream.byte_offset();
                    values.push(value);
                }
                Some(Err(error)) if error.is_eof() => break,
                Some(Err(_)) => {
                    // Not JSON (e.g. a warning line): skip to the next object.
                    match self.buffer[consumed + 1..]
                        .iter()
                        .position(|byte| *byte == b'{')
                    {
                        Some(offset) => consumed += offset + 1,
                        None => {
                            consumed = self.buffer.len();
                            break;
                        }
                    }
                }
                None => break,
            }
        }
        self.buffer.drain(..consumed);
        values
    }
}

/// Known objects by uid, for dedupe and gap reconciliation.
#[derive(Default)]
pub struct KnownObjects {
    objects: HashMap<String, (String, Value)>,
}

fn identity(object: &Value) -> Option<(String, String)> {
    let metadata = object.get("metadata")?;
    let uid = metadata
        .get("uid")
        .and_then(Value::as_str)
        .map(str::to_string)
        .or_else(|| {
            let name = metadata.get("name")?.as_str()?;
            let namespace = metadata
                .get("namespace")
                .and_then(Value::as_str)
                .unwrap_or("");
            Some(format!("{namespace}/{name}"))
        })?;
    let version = metadata
        .get("resourceVersion")
        .and_then(Value::as_str)
        .unwrap_or("")
        .to_string();
    Some((uid, version))
}

/// Drops `metadata.managedFields`, which is large and rarely useful.
pub fn slim(mut object: Value) -> Value {
    if let Some(metadata) = object.get_mut("metadata").and_then(Value::as_object_mut) {
        metadata.remove("managedFields");
    }
    object
}

impl KnownObjects {
    pub fn seed(&mut self, items: &[Value]) {
        self.objects = items
            .iter()
            .filter_map(|item| {
                identity(item).map(|(uid, version)| (uid, (version, slim(item.clone()))))
            })
            .collect();
    }

    /// A watch event to deliver, or `None` for duplicates and bookmarks.
    /// `MODIFIED` carries the previously known object as `old`.
    pub fn apply(&mut self, event: &Value) -> Option<Change> {
        let kind = event.get("type").and_then(Value::as_str)?.to_string();
        let object = slim(event.get("object")?.clone());
        match kind.as_str() {
            "ADDED" | "MODIFIED" => {
                let (uid, version) = identity(&object)?;
                if self
                    .objects
                    .get(&uid)
                    .is_some_and(|(known, _)| *known == version)
                {
                    return None;
                }
                let old = self
                    .objects
                    .insert(uid, (version, object.clone()))
                    .map(|(_, old)| old)
                    .filter(|_| kind == "MODIFIED");
                Some(Change { kind, object, old })
            }
            "DELETED" => {
                let (uid, _) = identity(&object)?;
                self.objects.remove(&uid);
                Some(Change::new(kind, object))
            }
            "ERROR" => Some(Change::new(kind, object)),
            _ => None,
        }
    }

    /// Synthetic events for what changed while no stream was open.
    pub fn reconcile(&mut self, items: &[Value]) -> Vec<Change> {
        let mut events = Vec::new();
        let mut current = HashMap::new();
        for item in items {
            let Some((uid, version)) = identity(item) else {
                continue;
            };
            let object = slim(item.clone());
            match self.objects.get(&uid) {
                None => events.push(Change::new("ADDED".to_string(), object.clone())),
                Some((known, old)) if *known != version => events.push(Change {
                    kind: "MODIFIED".to_string(),
                    object: object.clone(),
                    old: Some(old.clone()),
                }),
                Some(_) => {}
            }
            current.insert(uid, (version, object));
        }
        for (uid, (_, object)) in &self.objects {
            if !current.contains_key(uid) {
                events.push(Change::new("DELETED".to_string(), object.clone()));
            }
        }
        self.objects = current;
        events
    }
}

/// One event for the ring.
#[derive(Clone, Debug, PartialEq)]
pub struct Change {
    pub kind: String,
    pub object: Value,
    /// Previous object (`MODIFIED` only).
    pub old: Option<Value>,
}

impl Change {
    fn new(kind: String, object: Value) -> Self {
        Self {
            kind,
            object,
            old: None,
        }
    }
}

/// Ring of recent events with monotonically increasing `seq`.
struct Ring {
    seq: u64,
    events: VecDeque<Value>,
}

impl Ring {
    fn push(&mut self, change: Change) {
        self.seq += 1;
        let mut event = json!({"seq": self.seq, "type": change.kind, "object": change.object});
        if let Some(old) = change.old {
            event["old"] = old;
        }
        self.events.push_back(event);
        while self.events.len() > RING {
            self.events.pop_front();
        }
    }
}

struct Watcher {
    ring: Mutex<Ring>,
    updated: broadcast::Sender<()>,
    subscribers: AtomicUsize,
    task: Mutex<Option<JoinHandle<()>>>,
    /// Outcome of the first list: `None` while pending.
    seeded: tokio::sync::watch::Sender<Option<Result<(), String>>>,
}

impl Watcher {
    fn push(&self, change: Change) {
        self.ring.lock().expect("ring lock").push(change);
        let _ = self.updated.send(());
    }

    fn body(&self) -> String {
        let ring = self.ring.lock().expect("ring lock");
        json!({"events": ring.events}).to_string()
    }
}

/// All watchers of one server process.
#[derive(Default)]
pub struct WatchHub {
    watchers: Mutex<HashMap<String, Arc<Watcher>>>,
}

impl WatchHub {
    /// The event-log body for `uri`: `{"events": [...]}` (empty when no one
    /// watches it). A watched URI's first read waits for the watcher's seed
    /// list (a signal, not a poll), so a bad resource fails the read and the
    /// client's `watch_start` instead of surfacing later as an `ERROR` event.
    pub async fn read(&self, uri: &str) -> Result<(String, &'static str), String> {
        WatchTarget::parse(uri)?;
        let watcher = self.watchers.lock().expect("hub lock").get(uri).cloned();
        let Some(watcher) = watcher else {
            return Ok((json!({"events": []}).to_string(), EVENT_LOG_MIME));
        };
        let mut seeded = watcher.seeded.subscribe();
        let outcome = tokio::time::timeout(SEED_WAIT, seeded.wait_for(Option::is_some))
            .await
            .ok()
            .and_then(Result::ok)
            .and_then(|state| state.clone());
        if let Some(Err(error)) = outcome {
            return Err(error);
        }
        Ok((watcher.body(), EVENT_LOG_MIME))
    }

    /// Joins (or starts) the watcher for `uri`; returns its update channel.
    pub fn acquire(
        &self,
        service: &KcpServiceHandle,
        uri: &str,
    ) -> Result<broadcast::Receiver<()>, String> {
        let target = WatchTarget::parse(uri)?;
        let mut watchers = self.watchers.lock().expect("hub lock");
        let watcher = watchers
            .entry(uri.to_string())
            .or_insert_with(|| {
                let watcher = Arc::new(Watcher {
                    ring: Mutex::new(Ring {
                        seq: 0,
                        events: VecDeque::new(),
                    }),
                    updated: broadcast::channel(64).0,
                    subscribers: AtomicUsize::new(0),
                    task: Mutex::new(None),
                    seeded: tokio::sync::watch::channel(None).0,
                });
                let task = tokio::spawn(run_watcher(service.clone(), target, watcher.clone()));
                *watcher.task.lock().expect("task lock") = Some(task);
                watcher
            })
            .clone();
        watcher.subscribers.fetch_add(1, Ordering::SeqCst);
        Ok(watcher.updated.subscribe())
    }

    /// Leaves `uri`; the last subscriber stops the watcher.
    pub fn release(&self, uri: &str) {
        let mut watchers = self.watchers.lock().expect("hub lock");
        let last = watchers
            .get(uri)
            .is_some_and(|watcher| watcher.subscribers.fetch_sub(1, Ordering::SeqCst) == 1);
        if last {
            if let Some(watcher) = watchers.remove(uri) {
                if let Some(task) = watcher.task.lock().expect("task lock").take() {
                    task.abort();
                }
            }
        }
    }
}

/// Server capabilities: tools plus subscribable resources.
pub fn server_info() -> ServerConfig {
    ServerConfig::new(
        ServerCapabilities::builder()
            .enable_tools()
            .enable_resources()
            .enable_resources_subscribe()
            .build(),
    )
}

pub fn resource_templates() -> ListResourceTemplatesResult {
    ListResourceTemplatesResult::with_all_items(vec![
        ResourceTemplate::new(TEMPLATE, "kcp-events")
            .with_description(DESCRIPTION)
            .with_mime_type(EVENT_LOG_MIME),
        ResourceTemplate::new(API_TEMPLATE, "kcp-api")
            .with_description(API_DESCRIPTION)
            .with_mime_type("application/json"),
    ])
}

pub async fn read_resource(
    hub: &WatchHub,
    service: &KcpServiceHandle,
    uri: &str,
) -> Result<ReadResourceResponse, ErrorData> {
    let (text, mime) = if uri.starts_with(API_SCHEME) {
        read_api(service, uri)
            .await
            .map(|text| (text, "application/json"))
    } else {
        hub.read(uri).await
    }
    .map_err(|error| ErrorData::resource_not_found(error, None))?;
    Ok(ReadResourceResponse::Complete(ReadResourceResult::new(
        vec![ResourceContents::TextResourceContents {
            uri: uri.to_string(),
            mime_type: Some(mime.to_string()),
            text,
            meta: None,
        }],
    )))
}

async fn read_api(service: &KcpServiceHandle, uri: &str) -> Result<String, String> {
    let (workspace, kcp) = parse_api_uri(uri)?;
    let kcp = service.resolve_kcp(kcp.as_deref()).await?;
    let workspace = service.workspace_or_default(&kcp, Some(&workspace));
    super::tools::api_resources(service, &kcp, &workspace)
        .await
        .map(|value| value.to_string())
}

/// Accepts every requested `kcp-events://` URI that parses.
pub fn accepted_subscriptions(requested: &SubscriptionFilter) -> Option<SubscriptionFilter> {
    let uris: Vec<String> = requested
        .resource_subscriptions
        .iter()
        .flatten()
        .filter(|uri| WatchTarget::parse(uri).is_ok())
        .cloned()
        .collect();
    Some(
        SubscriptionFilter::builder()
            .resource_subscriptions(uris)
            .build(),
    )
}

/// Serves one `subscriptions/listen` stream until the client cancels.
pub async fn listen(
    hub: &WatchHub,
    service: &KcpServiceHandle,
    context: SubscriptionContext,
) -> Result<(), ErrorData> {
    let uris = context
        .accepted()
        .resource_subscriptions
        .clone()
        .unwrap_or_default();
    let (updated_tx, mut updated) = tokio::sync::mpsc::channel::<String>(64);
    let mut forwarders = Vec::new();
    let mut acquired = Vec::new();
    for uri in &uris {
        let mut receiver = match hub.acquire(service, uri) {
            Ok(receiver) => receiver,
            Err(_) => continue,
        };
        acquired.push(uri.clone());
        let tx = updated_tx.clone();
        let uri = uri.clone();
        forwarders.push(tokio::spawn(async move {
            loop {
                match receiver.recv().await {
                    Ok(()) | Err(broadcast::error::RecvError::Lagged(_)) => {
                        if tx.send(uri.clone()).await.is_err() {
                            return;
                        }
                    }
                    Err(broadcast::error::RecvError::Closed) => return,
                }
            }
        }));
    }
    drop(updated_tx);

    loop {
        tokio::select! {
            _ = context.cancelled() => break,
            uri = updated.recv() => match uri {
                Some(uri) => {
                    if context.sink().notify_resource_updated(uri).await.is_err() {
                        break;
                    }
                }
                None => break,
            },
        }
    }
    for forwarder in forwarders {
        forwarder.abort();
    }
    for uri in acquired {
        hub.release(&uri);
    }
    Ok(())
}

async fn list(service: &KcpServiceHandle, target: &WatchTarget) -> Result<Vec<Value>, String> {
    let kcp = service.resolve_kcp(target.kcp.as_deref()).await?;
    let mut args = target.selection();
    args.push("-o".to_string());
    args.push("json".to_string());
    let text = match service.read(&kcp, &target.workspace, &args).await {
        Ok(text) => text,
        Err(error) => {
            return Err(
                super::tools::with_resource_hint(service, &kcp, &target.workspace, error).await,
            )
        }
    };
    let value: Value =
        serde_json::from_str(&text).map_err(|error| format!("list is not JSON: {error}"))?;
    Ok(value
        .get("items")
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default())
}

/// Streams one watch until it ends; returns why.
async fn stream(
    service: &KcpServiceHandle,
    target: &WatchTarget,
    known: &mut KnownObjects,
    watcher: &Watcher,
) -> String {
    let kcp = match service.resolve_kcp(target.kcp.as_deref()).await {
        Ok(kcp) => kcp,
        Err(error) => return error,
    };
    let mut args = target.selection();
    args.extend(
        ["-w", "--output-watch-events", "-o", "json"]
            .into_iter()
            .map(str::to_string),
    );
    let mut child = match service.read_stream(&kcp, &target.workspace, &args).await {
        Ok(child) => child,
        Err(error) => return error,
    };
    let Some(mut stdout) = child.stdout.take() else {
        return "kubectl watch has no stdout".to_string();
    };
    let mut parser = JsonStream::default();
    let mut chunk = vec![0u8; 64 * 1024];
    loop {
        match stdout.read(&mut chunk).await {
            Ok(0) => break,
            Ok(read) => {
                for event in parser.feed(&chunk[..read]) {
                    if let Some(change) = known.apply(&event) {
                        watcher.push(change);
                    }
                }
            }
            Err(error) => return format!("watch stream read failed: {error}"),
        }
    }
    let mut stderr = String::new();
    if let Some(mut pipe) = child.stderr.take() {
        let _ = pipe.read_to_string(&mut stderr).await;
    }
    let status = child.wait().await;
    format!(
        "watch stream ended ({}){}",
        status
            .map(|status| status.to_string())
            .unwrap_or_else(|error| error.to_string()),
        if stderr.trim().is_empty() {
            String::new()
        } else {
            format!(": {}", stderr.trim())
        }
    )
}

async fn run_watcher(service: KcpServiceHandle, target: WatchTarget, watcher: Arc<Watcher>) {
    let mut known = KnownObjects::default();
    let mut seeded = false;
    let mut backoff = Duration::from_secs(1);
    let mut failures = 0u32;
    loop {
        match list(&service, &target).await {
            Ok(items) => {
                if seeded {
                    for change in known.reconcile(&items) {
                        watcher.push(change);
                    }
                } else {
                    known.seed(&items);
                    seeded = true;
                }
                watcher.seeded.send_replace(Some(Ok(())));
            }
            Err(error) => {
                if !seeded {
                    watcher.seeded.send_replace(Some(Err(error.clone())));
                }
                failures += 1;
                watcher.push(Change::new("ERROR".to_string(), json!({"message": error})));
                if failures >= MAX_FAILURES {
                    watcher.push(Change::new(
                        "ERROR".to_string(),
                        json!({"message": format!("giving up after {MAX_FAILURES} failures")}),
                    ));
                    return;
                }
                tokio::time::sleep(backoff).await;
                backoff = (backoff * 2).min(MAX_BACKOFF);
                continue;
            }
        }

        let started = Instant::now();
        let reason = stream(&service, &target, &mut known, &watcher).await;
        if started.elapsed() >= HEALTHY_STREAM {
            backoff = Duration::from_secs(1);
            failures = 0;
        } else {
            failures += 1;
            if failures >= MAX_FAILURES {
                watcher.push(Change::new(
                    "ERROR".to_string(),
                    json!({"message": format!("{reason}; giving up after {MAX_FAILURES} failures")}),
                ));
                return;
            }
        }
        tokio::time::sleep(backoff).await;
        backoff = (backoff * 2).min(MAX_BACKOFF);
    }
}

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

    fn object(name: &str, version: &str) -> Value {
        json!({"metadata": {"name": name, "namespace": "default", "uid": format!("uid-{name}"), "resourceVersion": version, "managedFields": [1]}})
    }

    #[test]
    fn parses_watch_uris() {
        let target = WatchTarget::parse(
            "kcp-events://watch/root:app/configmaps?namespace=*&labelSelector=app%3Dweb&kcp=local",
        )
        .unwrap();
        assert_eq!(target.workspace, "root:app");
        assert_eq!(target.resource, "configmaps");
        assert_eq!(target.namespace.as_deref(), Some("*"));
        assert_eq!(target.label_selector.as_deref(), Some("app=web"));
        assert_eq!(target.kcp.as_deref(), Some("local"));
        assert_eq!(
            target.selection(),
            ["get", "configmaps", "--all-namespaces", "-l", "app=web"]
        );
        assert!(WatchTarget::parse("kcp-events://watch/root:app").is_err());
        assert!(WatchTarget::parse("kcp-events://watch/root:app/cm?bogus=1").is_err());
        assert!(WatchTarget::parse("memo://notes").is_err());
    }

    #[test]
    fn json_stream_handles_split_and_concatenated_values() {
        let mut parser = JsonStream::default();
        assert!(parser.feed(b"{\"type\": \"ADD").is_empty());
        let values = parser.feed(b"ED\"}\n{\n  \"type\": \"DELETED\"\n}{\"a\"");
        assert_eq!(
            values,
            vec![json!({"type": "ADDED"}), json!({"type": "DELETED"})]
        );
        assert_eq!(parser.feed(b": 1}"), vec![json!({"a": 1})]);
        assert_eq!(parser.feed(b"Warning: x\n{\"b\":2}"), vec![json!({"b": 2})]);
    }

    #[test]
    fn known_objects_dedupe_and_reconcile() {
        let mut known = KnownObjects::default();
        known.seed(&[object("a", "1"), object("b", "1")]);
        assert!(known
            .apply(&json!({"type": "ADDED", "object": object("a", "1")}))
            .is_none());
        let change = known
            .apply(&json!({"type": "MODIFIED", "object": object("a", "2")}))
            .unwrap();
        assert_eq!(change.kind, "MODIFIED");
        assert!(change.object["metadata"].get("managedFields").is_none());
        let old = change.old.expect("MODIFIED carries the previous object");
        assert_eq!(old["metadata"]["resourceVersion"], "1");
        assert!(old["metadata"].get("managedFields").is_none());
        let added = known
            .apply(&json!({"type": "ADDED", "object": object("n", "1")}))
            .unwrap();
        assert_eq!(added.old, None);
        assert!(known
            .apply(&json!({"type": "BOOKMARK", "object": object("a", "2")}))
            .is_none());

        let events = known.reconcile(&[object("a", "3"), object("c", "1"), object("n", "1")]);
        let mut summary: Vec<String> = events
            .iter()
            .map(|change| {
                format!(
                    "{} {} {}",
                    change.kind,
                    change.object["metadata"]["name"].as_str().unwrap(),
                    change.old.is_some()
                )
            })
            .collect();
        summary.sort();
        assert_eq!(
            summary,
            ["ADDED c false", "DELETED b false", "MODIFIED a true"]
        );
    }

    #[test]
    fn ring_keeps_bounded_seq_ordered_events() {
        let mut ring = Ring {
            seq: 0,
            events: VecDeque::new(),
        };
        for index in 0..300 {
            ring.push(Change::new("ADDED".to_string(), json!({"i": index})));
        }
        assert_eq!(ring.events.len(), RING);
        assert_eq!(ring.events.front().unwrap()["seq"], 45);
        assert_eq!(ring.events.back().unwrap()["seq"], 300);
        assert!(ring.events.back().unwrap().get("old").is_none());
        ring.push(Change {
            kind: "MODIFIED".to_string(),
            object: json!({"v": 2}),
            old: Some(json!({"v": 1})),
        });
        assert_eq!(ring.events.back().unwrap()["old"], json!({"v": 1}));
    }
}