//! Target-owned managed bindings. Dependency versions are borrowed identities;
//! the host's weak Block registry never roots an otherwise unreachable graph.
use std::{
    collections::HashSet,
    sync::{Arc, Mutex},
};

use anyhow::{Result, bail};
use endor_core::{NativeContext, ObjectPayload, ObjectRef, Value, Vm, VmError, WeakObjectRef};

use crate::{block, event_dispatch::Subscription, ui_bridge::attach_desktop_type};

/// The output value is borrowed for equality only, including object handles.
/// It never owns a reference or dereferences a possibly reclaimed cached object.
#[derive(Debug, Clone)]
struct Cached {
    reads: Vec<endor_core::ReadDependency>,
    output: Value,
}

#[derive(Debug, Clone)]
pub struct Binding {
    field: String,
    pub compute: Option<ObjectRef>,
    reads: Arc<Mutex<Option<Cached>>>,
}
impl ObjectPayload for Binding {
    fn reference_assignment_only(&self) -> bool {
        true
    }
    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
    fn kind_name(&self) -> &'static str {
        "tessera-binding"
    }
    fn trace(&self, visit: &mut dyn FnMut(ObjectRef)) {
        if let Some(compute) = self.compute {
            visit(compute);
        }
    }
}

pub fn connect(
    cx: &mut NativeContext<'_>,
    target: ObjectRef,
    field: &str,
    compute: ObjectRef,
    desktop: ObjectRef,
) -> Result<ObjectRef, VmError> {
    if field.is_empty() || field.len() > 128 || field == "type" {
        return Err(VmError::InvalidArgument("invalid binding property"));
    }
    let (reference, data) = block::resolve(cx.store(), target)?;
    let mut data = data.clone();
    // Only actual fields can be bound; method names are not writable targets.
    let properties = data
        .properties
        .ok_or(VmError::InvalidArgument("missing property storage"))?;
    cx.store().get_attr(properties, field)?;
    if !cx.store().object(compute)?.payload().is_function() {
        return Err(VmError::InvalidArgument(
            "binding compute must be a Function",
        ));
    }
    if !cx
        .store()
        .call_arity(compute)?
        .ok_or(VmError::InvalidArgument(
            "binding compute requires positional metadata",
        ))?
        .accepts(0)
    {
        return Err(VmError::InvalidArgument(
            "binding compute requires zero unbound parameters",
        ));
    }
    let mut retained = vec![];
    for connection in &data.bindings {
        let mut binding = cx
            .store()
            .payload::<Binding>(*connection, "Binding")?
            .clone();
        if binding.compute.is_some() && binding.field != field {
            retained.push(*connection);
        } else if binding.compute.is_some() {
            binding.compute = None;
            *binding
                .reads
                .lock()
                .map_err(|_| VmError::InvalidArgument("binding cache poisoned"))? = None;
            cx.replace_payload(*connection, binding)?;
        }
    }
    if retained.len() >= 128 {
        return Err(VmError::ExecutionLimit("128 bindings per Block"));
    }
    let connection = cx.allocate(Binding {
        field: field.into(),
        compute: Some(compute),
        reads: Arc::new(Mutex::new(None)),
    })?;
    retained.push(connection);
    data.bindings = retained;
    // Generic text bindings replace the legacy textOf derivation.
    if field == "text" {
        data.text_binding = None;
    }
    cx.replace_payload(reference, data)?;
    let token = cx.allocate(Subscription {
        connection: cx.store().downgrade(connection)?,
    })?;
    attach_desktop_type(cx, token, "Subscription", desktop)?;
    Ok(token)
}

pub fn refresh(
    vm: &mut Vm,
    nodes: &[WeakObjectRef],
    host_desktops: &HashSet<ObjectRef>,
) -> Result<usize> {
    let mut work = 0;
    loop {
        let mut evaluated = false;
        let mut seen = HashSet::new();
        for weak in nodes {
            let Some(target) = vm.upgrade(*weak)? else {
                continue;
            };
            let result: Result<()> = (|| {
                let (reference, data) = block::resolve(vm.object_store(), target)?;
                if !seen.insert(reference) {
                    return Ok(());
                }
                let connections = data.bindings.clone();
                for connection in connections {
                    let binding = vm.payload::<Binding>(connection, "Binding")?.clone();
                    let Some(compute) = binding.compute else {
                        continue;
                    };
                    let cached = binding
                        .reads
                        .lock()
                        .map_err(|_| anyhow::anyhow!("binding cache poisoned"))?
                        .clone();
                    if cached.as_ref().is_some_and(|cached| {
                        cached
                            .reads
                            .iter()
                            .all(|read| vm.object_store().read_dependency_valid(read))
                            && vm
                                .get_attr(target, &binding.field)
                                .is_ok_and(|value| value == cached.output)
                    }) {
                        continue;
                    }
                    work += 1;
                    if work > 10_000 {
                        bail!("property bindings did not stabilize within the execution budget");
                    }
                    evaluated = true;
                    let (value, reads) = vm
                        .call_read_only_precise_with_limits(
                            compute,
                            &[],
                            endor_core::ExecutionLimits {
                                max_frames: 128,
                                max_instructions: 100_000,
                            },
                        )
                        .map_err(|error| crate::endor_host::runtime_error(vm, error))?;
                    let applied: Result<()> = (|| {
                        let reads = reads
                            .dependencies
                            .into_iter()
                            .filter(|read| !host_desktops.contains(&read.object()))
                            .collect();
                        let (_, data) = block::resolve(vm.object_store(), target)?;
                        block::apply_visual_property(&mut data.clone(), &binding.field, &value)?;
                        if vm.get_attr(target, &binding.field)? != value {
                            vm.set_attr(target, &binding.field, value.clone())?;
                        }
                        *binding
                            .reads
                            .lock()
                            .map_err(|_| anyhow::anyhow!("binding cache poisoned"))? =
                            Some(Cached {
                                reads,
                                output: value.clone(),
                            });
                        Ok(())
                    })();
                    vm.release_value(value)?;
                    applied?;
                }
                Ok(())
            })();
            vm.release(target)?;
            result?;
        }
        if !evaluated {
            return Ok(work);
        }
    }
}