//! UX1 — Trace + Compat Rule Unblocker: end-to-end trace wiring tests.
//!
//! These tests verify that `flatten_xfa_to_pdf` actually emits trace
//! events into an installed `RecordingSink`. They cover the five hot
//! phases (bind, occur, presence, paginate, suppress) by exercising
//! synthetic XFA documents that drive each emit site.
//!
//! Each test installs a fresh `RecordingSink` and then inspects the
//! collected events for the expected `(Phase, Reason)` anchor. The
//! sink is off by default outside `with_sink`, so production runtime
//! cost is unchanged when no sink is installed.

use std::rc::Rc;
use std::sync::{Arc, Mutex};

use lopdf::{dictionary, Document, Object, Stream};
use pdf_xfa::adobe_compat::{cap_suppression_by_form_dom, AdobeCompatRuleId};
use pdf_xfa::flatten_xfa_to_pdf;
use xfa_layout_engine::trace::{with_global_sink, with_sink, RecordingSink};

/// Serialise tests that install a process-global sink. Cargo runs tests in
/// parallel by default; multiple parallel `with_global_sink` calls would
/// overwrite one another's sink. The mutex is poison-resilient: a panic
/// inside one test does not prevent later tests from acquiring it.
fn global_sink_lock() -> &'static Mutex<()> {
    static LOCK: std::sync::OnceLock<Mutex<()>> = std::sync::OnceLock::new();
    LOCK.get_or_init(|| Mutex::new(()))
}

fn build_xfa_pdf(xdp: &str) -> Vec<u8> {
    let mut doc = Document::with_version("1.4");
    let xdp_bytes = xdp.as_bytes().to_vec();
    let xfa_stream = Stream::new(
        dictionary! { "Length" => Object::Integer(xdp_bytes.len() as i64) },
        xdp_bytes,
    );
    let xfa_id = doc.add_object(Object::Stream(xfa_stream));
    let pages_id = doc.new_object_id();
    let content_id = doc.add_object(Object::Stream(Stream::new(
        dictionary! { "Length" => Object::Integer(0_i64) },
        vec![],
    )));
    let page_id = doc.add_object(Object::Dictionary(dictionary! {
        "Type"     => Object::Name(b"Page".to_vec()),
        "Parent"   => Object::Reference(pages_id),
        "MediaBox" => Object::Array(vec![
            Object::Integer(0), Object::Integer(0),
            Object::Integer(612), Object::Integer(792),
        ]),
        "Contents" => Object::Reference(content_id),
    }));
    doc.objects.insert(
        pages_id,
        Object::Dictionary(dictionary! {
            "Type"  => Object::Name(b"Pages".to_vec()),
            "Kids"  => Object::Array(vec![Object::Reference(page_id)]),
            "Count" => Object::Integer(1),
        }),
    );
    let acroform_id = doc.add_object(Object::Dictionary(dictionary! {
        "XFA" => Object::Reference(xfa_id),
    }));
    let catalog_id = doc.add_object(Object::Dictionary(dictionary! {
        "Type"     => Object::Name(b"Catalog".to_vec()),
        "Pages"    => Object::Reference(pages_id),
        "AcroForm" => Object::Reference(acroform_id),
    }));
    doc.trailer.set("Root", Object::Reference(catalog_id));
    let mut out = Vec::new();
    doc.save_to(&mut out).expect("save lopdf");
    out
}

/// Two-page XFA template with one bound section and one empty section —
/// exercises bind (data binding), paginate (multi-page layout), and the
/// data-empty-page suppress path. Adapted from
/// `crates/pdf-xfa/tests/m3b_phaseD_kappa_page_drop.rs::XDP_VISIBLE_BOUND_DROPS_EMPTY_PAGE`
/// which is a known-good template that flattens cleanly through the full
/// pipeline.
const XDP_BIND_TWO_PAGE: &str = r#"<?xml version="1.0"?>
<xdp:xdp xmlns:xdp="http://ns.adobe.com/xdp/" xmlns:xfa="http://www.xfa.org/schema/xfa-data/1.0/">
  <template xmlns="http://www.xfa.org/schema/xfa-template/3.3/">
    <subform name="form1" layout="paginate" w="8.5in" h="11in">
      <pageSet>
        <pageArea name="Page1">
          <contentArea x="36pt" y="36pt" w="540pt" h="720pt"/>
          <medium stock="default" short="612pt" long="792pt"/>
        </pageArea>
        <pageArea name="Page2">
          <contentArea x="36pt" y="36pt" w="540pt" h="720pt"/>
          <medium stock="default" short="612pt" long="792pt"/>
        </pageArea>
      </pageSet>
      <subform name="section1" layout="tb" w="540pt">
        <field name="CustomerName" w="300pt" h="18pt">
          <ui><textEdit/></ui>
        </field>
      </subform>
      <subform name="section2" layout="tb" w="540pt">
        <breakBefore targetType="pageArea"/>
        <field name="EmptyOptional" w="300pt" h="18pt">
          <ui><textEdit/></ui>
        </field>
      </subform>
    </subform>
  </template>
  <xfa:datasets>
    <xfa:data>
      <form1>
        <CustomerName>Alice Johnson</CustomerName>
      </form1>
    </xfa:data>
  </xfa:datasets>
</xdp:xdp>"#;

#[test]
fn bind_emitter_fires_in_production_flatten() {
    // `flatten_xfa_to_pdf` runs its inner pipeline on a worker thread, so
    // we must install a Send + Sync sink via `with_global_sink` rather
    // than the thread-local-only `with_sink`. The Arc<RecordingSink> we
    // keep here lets us read events back after the call returns.
    let _guard = global_sink_lock().lock().unwrap_or_else(|e| e.into_inner());
    let pdf = build_xfa_pdf(XDP_BIND_TWO_PAGE);
    let sink: Arc<RecordingSink> = Arc::new(RecordingSink::new());
    let _flat = with_global_sink(sink.clone() as _, || flatten_xfa_to_pdf(&pdf))
        .expect("flatten should succeed");

    let events = sink.events();
    let bind_events: Vec<_> = events.iter().filter(|e| e.phase.tag() == "bind").collect();
    assert!(
        !bind_events.is_empty(),
        "expected at least one bind trace event from production flatten; got {} events total",
        events.len()
    );
    // The event must describe the merger outcome with a stable input field.
    let bind = bind_events[0];
    assert!(
        bind.decision
            .as_deref()
            .unwrap_or("")
            .contains("any_data_bound="),
        "bind event decision must report data-bound state: {:?}",
        bind.decision
    );
}

#[test]
fn paginate_emitter_fires_in_production_flatten() {
    let _guard = global_sink_lock().lock().unwrap_or_else(|e| e.into_inner());
    let pdf = build_xfa_pdf(XDP_BIND_TWO_PAGE);
    let sink: Arc<RecordingSink> = Arc::new(RecordingSink::new());
    let _flat = with_global_sink(sink.clone() as _, || flatten_xfa_to_pdf(&pdf))
        .expect("flatten should succeed");

    let events = sink.events();
    let paginate_events: Vec<_> = events
        .iter()
        .filter(|e| e.phase.tag() == "paginate")
        .collect();
    assert!(
        !paginate_events.is_empty(),
        "expected at least one paginate event from production flatten"
    );
    // The paginate event must carry available/needed_h numerics (we encode
    // page-count and form-DOM-target into those slots in flatten.rs).
    let pg = paginate_events[0];
    let input = pg.input.as_deref().unwrap_or("");
    assert!(
        input.contains("available_h="),
        "paginate event input must include available_h: {input}"
    );
}

#[test]
fn no_sink_means_no_overhead_and_flatten_still_runs() {
    // Outside `with_sink`, flatten must produce the same output. This test
    // simply checks the no-sink path does not panic and produces a PDF.
    let pdf = build_xfa_pdf(XDP_BIND_TWO_PAGE);
    let out = flatten_xfa_to_pdf(&pdf).expect("flatten without sink should succeed");
    assert!(out.starts_with(b"%PDF-"), "output must be a PDF");
}

#[test]
fn cap_rule_decision_round_trip() {
    // Pure-function check that the migrated rule is the canonical entry
    // point for the form-DOM cap. This anchors the test against the
    // executable rule rather than re-implementing the inline check.
    let d1 = cap_suppression_by_form_dom(7, Some(17));
    assert_eq!(d1.max_suppress, 0);
    assert_eq!(
        d1.rule,
        AdobeCompatRuleId::CapSuppressionByFormDomPageAreaCount
    );

    let d2 = cap_suppression_by_form_dom(7, None);
    assert_eq!(d2.max_suppress, usize::MAX);
}

#[test]
fn cap_rule_silent_when_no_form_dom() {
    // 7dbbe9d9 counterexample shape: no form DOM, the cap rule must
    // remain inactive and not emit a trace event. Uses the cheap
    // thread-local sink (no thread crossings here) and the global-sink
    // lock to keep this test sequential with the cross-thread ones.
    let _guard = global_sink_lock().lock().unwrap_or_else(|e| e.into_inner());
    let sink: Rc<RecordingSink> = Rc::new(RecordingSink::new());
    with_sink(sink.clone(), || {
        let _ = cap_suppression_by_form_dom(2, None);
    });
    let cap_events: Vec<_> = sink
        .events()
        .into_iter()
        .filter(|e| e.reason.tag() == "suppress_capped_by_form_dom")
        .collect();
    assert!(
        cap_events.is_empty(),
        "cap rule must be silent when no form DOM is present"
    );
}
