//! The native sink's C ABI ([extensions/native-sink.md] §9): what a native host's player
//! integration calls, on any platform, to verify what it fetched, sign what stream-audit has a
//! player sign, and end a session nobody can see. Built with features `native_sink` and `ffi`; it
//! uses the envelope surface's buffers and errors ([`crate::ffi`]).
//!
//! A session is an opaque handle. Every call takes its inputs as bytes, NUL-terminated UTF-8
//! strings and JSON, and writes JSON or bytes to a [`WaiBuffer`] the caller frees with
//! `wai_buffer_free`. A JSON input names its members exactly as §9 lists them; an unknown
//! member is refused, so a host learns at once that it is talking to an older library. The
//! handle is safe to call from any thread: each call holds the session's lock for its duration.
//!
//! Status codes are [`crate::ffi`]'s, plus:
//! - `WAI_ERR_SINK_INPUT` (8): an input that does not read (a configuration, an observation, an
//!   energy entry, an event name); `wai_last_error` starts with the reason
//!   (`native_config_invalid`, `native_input_invalid`);
//! - `WAI_ERR_SINK_ENDED` (9): a call after the session ended (`native_session_ended`).
//!
//! A verdict that refuses a playlist or an object is **not** an error: the call returns
//! `WAI_OK` and the verdict says `"present": false` with the reason. The host then withholds the
//! bytes from its decoder.
//!
//! [extensions/native-sink.md]: ../../extensions/native-sink.md

use std::ffi::{c_char, c_int, CStr};
use std::panic::{self, AssertUnwindSafe};
use std::sync::Mutex;

use ed25519_dalek::SigningKey;
use serde_json::{json, Value};

use crate::ffi::{clear_buf, input_slice, set_err, write_buf, WaiBuffer, WAI_ERR_INVALID_ARG, WAI_ERR_PANIC, WAI_OK};
use crate::native_sink::{self as ns, Config, Observation, ObjectMeta, Reason, Sink};

/// An input that does not read: a configuration, an observation, an energy entry, an event
/// (native-sink §9). `wai_last_error` starts with the reason.
pub const WAI_ERR_SINK_INPUT: c_int = 8;
/// A call on a session that has ended (native-sink §9).
pub const WAI_ERR_SINK_ENDED: c_int = 9;

/// The ABI's version: a host checks it before it opens a session (native-sink §9).
pub const WAI_SINK_ABI_VERSION: u32 = 1;

/// One native sink session (native-sink §3). Opaque to C.
pub struct WaiSink {
    inner: Mutex<Sink>,
}

fn err_of(r: Reason) -> (c_int, String) {
    let code = match r {
        Reason::SessionEnded => WAI_ERR_SINK_ENDED,
        _ => WAI_ERR_SINK_INPUT,
    };
    (code, r.code().to_string())
}

unsafe fn cstr<'a>(p: *const c_char, what: &str) -> Result<&'a str, (c_int, String)> {
    if p.is_null() {
        return Err((WAI_ERR_INVALID_ARG, format!("{what} is null")));
    }
    unsafe { CStr::from_ptr(p) }.to_str().map_err(|_| (WAI_ERR_INVALID_ARG, format!("{what} is not UTF-8")))
}

unsafe fn json_arg(p: *const c_char, what: &str) -> Result<Value, (c_int, String)> {
    let s = unsafe { cstr(p, what) }?;
    serde_json::from_str(s).map_err(|e| (WAI_ERR_SINK_INPUT, format!("native_input_invalid: {what}: {e}")))
}

unsafe fn bytes_arg<'a>(p: *const u8, len: usize, what: &str) -> Result<&'a [u8], (c_int, String)> {
    unsafe { input_slice(p, len) }.ok_or((WAI_ERR_INVALID_ARG, format!("{what} null with len>0")))
}

/// Run `f` on the session under its lock, catching a panic; the status, and on error the
/// message in `wai_last_error`.
fn with_sink<T>(s: *const WaiSink, f: impl FnOnce(&mut Sink) -> Result<T, (c_int, String)>) -> Result<T, c_int> {
    if s.is_null() {
        set_err("sink is null");
        return Err(WAI_ERR_INVALID_ARG);
    }
    let sink = unsafe { &*s };
    let r = panic::catch_unwind(AssertUnwindSafe(|| {
        let mut g = sink.inner.lock().unwrap_or_else(|p| p.into_inner());
        f(&mut g)
    }));
    match r {
        Ok(Ok(v)) => Ok(v),
        Ok(Err((code, msg))) => {
            set_err(msg);
            Err(code)
        }
        Err(_) => {
            set_err("internal panic in the native sink");
            Err(WAI_ERR_PANIC)
        }
    }
}

/// Write `v` to `out`, or clear `out` and return the status.
unsafe fn finish(out: *mut WaiBuffer, r: Result<Vec<u8>, c_int>) -> c_int {
    match r {
        Ok(v) => {
            if unsafe { write_buf(out, v) } {
                WAI_OK
            } else {
                set_err("output buffer pointer is null");
                WAI_ERR_INVALID_ARG
            }
        }
        Err(code) => {
            unsafe { clear_buf(out) };
            code
        }
    }
}

fn json_bytes(v: &Value) -> Vec<u8> {
    serde_json::to_vec(v).expect("JSON serializes")
}

/// The ABI version this library implements, [`WAI_SINK_ABI_VERSION`].
#[unsafe(no_mangle)]
pub extern "C" fn wai_sink_abi_version() -> u32 {
    WAI_SINK_ABI_VERSION
}

/// The microseconds since 1970-01-01T00:00:00Z on this host's clock: the time a host passes to
/// the calls below when it has no clock of its own to hand.
#[unsafe(no_mangle)]
pub extern "C" fn wai_sink_now_us() -> u64 {
    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_micros() as u64)
}

/// Open a session (native-sink §3) at `start_us` under `config_json` (§3.1):
/// `{"channel", "owner", "presentation", "window_us", "ad_classes", "content_addressed",
/// "require", "unseen_grace_ms", "idle_ms"}`, of which `channel` and `owner` are required.
///
/// With `seed` null, the session key is fresh from the operating system's random generator, as
/// stream-audit §3.2 requires. A 32-byte `seed` derives it instead: for conformance vectors and
/// tests only, never for a viewer's session. Writes the handle to `out`; free it with
/// [`wai_sink_free`].
///
/// # Safety
/// `config_json` NUL-terminated; `seed` null or valid for `seed_len` bytes; `out` writable.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_open(config_json: *const c_char, seed: *const u8, seed_len: usize, start_us: u64, out: *mut *mut WaiSink) -> c_int {
    if out.is_null() {
        set_err("out is null");
        return WAI_ERR_INVALID_ARG;
    }
    unsafe { *out = std::ptr::null_mut() };
    let r = panic::catch_unwind(|| -> Result<Sink, (c_int, String)> {
        let v = unsafe { json_arg(config_json, "config_json") }?;
        let cfg = Config::from_json(&v).map_err(err_of)?;
        let key = if seed.is_null() {
            let mut b = [0u8; 32];
            getrandom::fill(&mut b).map_err(|e| (WAI_ERR_INVALID_ARG, format!("the platform's random generator failed: {e}")))?;
            SigningKey::from_bytes(&b)
        } else {
            let s = unsafe { bytes_arg(seed, seed_len, "seed") }?;
            let s: [u8; 32] = s.try_into().map_err(|_| (WAI_ERR_INVALID_ARG, "a seed is 32 bytes".to_string()))?;
            SigningKey::from_bytes(&s)
        };
        Ok(Sink::open(cfg, key, start_us))
    });
    match r {
        Ok(Ok(s)) => {
            unsafe { *out = Box::into_raw(Box::new(WaiSink { inner: Mutex::new(s) })) };
            WAI_OK
        }
        Ok(Err((code, msg))) => {
            set_err(msg);
            code
        }
        Err(_) => {
            set_err("internal panic in wai_sink_open");
            WAI_ERR_PANIC
        }
    }
}

/// Free a session. No record is signed: a host that wants the bundle ends the session first
/// ([`wai_sink_end`]). Safe with null.
///
/// # Safety
/// `s` null or a handle from [`wai_sink_open`], not used after this call.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_free(s: *mut WaiSink) {
    if !s.is_null() {
        drop(unsafe { Box::from_raw(s) });
    }
}

/// The session's identity (§3.2) as JSON: `{"channel", "owner", "player", "session",
/// "start_us", "presentation", "ended"}`.
///
/// # Safety
/// `s` a handle; `out` writable.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_info(s: *const WaiSink, out: *mut WaiBuffer) -> c_int {
    let r = with_sink(s, |k| Ok(json_bytes(&k.info_json())));
    unsafe { finish(out, r) }
}

/// The name of a playlist's receipt file relative to the playlist, `mr/<BLAKE3>.json`
/// (http-delivery §8.6), as UTF-8 bytes: the host fetches it beside the playlist.
///
/// # Safety
/// `playlist` valid for `len` bytes; `out` writable.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_receipt_name(playlist: *const u8, len: usize, out: *mut WaiBuffer) -> c_int {
    let r = match unsafe { input_slice(playlist, len) } {
        Some(b) => Ok(ns::receipt_name(b).into_bytes()),
        None => {
            set_err("playlist null with len>0");
            Err(WAI_ERR_INVALID_ARG)
        }
    };
    unsafe { finish(out, r) }
}

/// A playlist the player fetched from `url` (after redirects), with the playlist receipt the host
/// fetched beside it (`receipt` null when there is none, or none was served), verified and taken
/// into the session (§4.2). Writes the verdict: `{"kind", "listing", "receipt", "receipt_name",
/// "step", "need_parent", "refused", "present", "segments", "ads"}`.
///
/// When `need_parent` is a name, the chain of the playlist's live receipts needs that parent: the
/// host fetches it relative to the playlist, hands it over with [`wai_sink_playlist_parent`], and
/// calls this again with the same playlist.
///
/// # Safety
/// `s` a handle; `url` NUL-terminated; `bytes` valid for `len`; `receipt` null or valid for
/// `receipt_len`; `out` writable.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_playlist(
    s: *const WaiSink,
    url: *const c_char,
    bytes: *const u8,
    len: usize,
    receipt: *const u8,
    receipt_len: usize,
    out: *mut WaiBuffer,
) -> c_int {
    let r = with_sink(s, |k| {
        let url = unsafe { cstr(url, "url") }?;
        let b = unsafe { bytes_arg(bytes, len, "bytes") }?;
        let rc = if receipt.is_null() { None } else { Some(unsafe { bytes_arg(receipt, receipt_len, "receipt") }?) };
        let v = k.playlist(url, b, rc).map_err(err_of)?;
        Ok(json_bytes(&v.to_json(&ns::receipt_name(b))))
    });
    unsafe { finish(out, r) }
}

/// Hand over a parent receipt file a playlist's verdict named in `need_parent`: `name` as the
/// verdict gave it (`mr/<hash>.json`), `bytes` what was served, or null when nothing was
/// (http-delivery §8.6, `playlist_receipt_parent_missing`).
///
/// # Safety
/// `s` a handle; `url`, `name` NUL-terminated; `bytes` null or valid for `len`.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_playlist_parent(s: *const WaiSink, url: *const c_char, name: *const c_char, bytes: *const u8, len: usize) -> c_int {
    match with_sink(s, |k| {
        let url = unsafe { cstr(url, "url") }?;
        let name = unsafe { cstr(name, "name") }?;
        let h = name
            .strip_prefix("mr/")
            .and_then(|n| n.strip_suffix(".json"))
            .and_then(crate::http_binding::unhex32)
            .ok_or((WAI_ERR_SINK_INPUT, "native_input_invalid: a parent is named mr/<64 hex>.json".to_string()))?;
        let b = if bytes.is_null() { None } else { Some(unsafe { bytes_arg(bytes, len, "bytes") }?.to_vec()) };
        if k.ended().is_some() {
            return Err(err_of(Reason::SessionEnded));
        }
        k.playlist_parent(url, h, b);
        Ok(())
    }) {
        Ok(()) => WAI_OK,
        Err(c) => c,
    }
}

/// A response body the player received, with its content codings removed (§4.3). `meta_json`:
/// `{"url", "t_us", "media", "status", "cmsd_static", "cmsd_dynamic"}`, of which `url` and `t_us`
/// are required. Writes the verdict: `{"sha256", "cid", "listing", "bound", "media", "key",
/// "acknowledged", "refused", "present"}`. With `"present": false` the host does not hand the
/// bytes to its decoder.
///
/// # Safety
/// `s` a handle; `meta_json` NUL-terminated; `bytes` valid for `len`; `out` writable.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_object(s: *const WaiSink, meta_json: *const c_char, bytes: *const u8, len: usize, out: *mut WaiBuffer) -> c_int {
    let r = with_sink(s, |k| {
        let m = ObjectMeta::from_json(&unsafe { json_arg(meta_json, "meta_json") }?).map_err(err_of)?;
        let b = unsafe { bytes_arg(bytes, len, "bytes") }?;
        let v = k.object(&m, b).map_err(err_of)?;
        Ok(json_bytes(&v.to_json()))
    });
    unsafe { finish(out, r) }
}

/// A log event (§5.2): `start`, `present`, `stall-start`, `stall-end`, `end` or `abandon`, at
/// `t_us`.
///
/// # Safety
/// `s` a handle; `name` NUL-terminated.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_event(s: *const WaiSink, name: *const c_char, t_us: u64) -> c_int {
    match with_sink(s, |k| {
        let n = unsafe { cstr(name, "name") }?;
        k.event(n, t_us).map_err(err_of)
    }) {
        Ok(()) => WAI_OK,
        Err(c) => c,
    }
}

/// An observation of playback (§5.1): `{"t_us", "playing", "visible_bp", "audible",
/// "program_ms", "seeked", "interaction"}`, at least every 200 ms while the player presents.
/// Writes `{"end": null}`, or `{"end": "hidden"}` / `{"end": "idle"}` when the session's
/// termination rule ends it (§6.1): the host then stops its player and ends the session so.
///
/// # Safety
/// `s` a handle; `obs_json` NUL-terminated; `out` writable.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_observe(s: *const WaiSink, obs_json: *const c_char, out: *mut WaiBuffer) -> c_int {
    let r = with_sink(s, |k| {
        let o = Observation::from_json(&unsafe { json_arg(obs_json, "obs_json") }?).map_err(err_of)?;
        let e = k.observe(&o).map_err(err_of)?;
        Ok(json_bytes(&json!({"end": e.map(crate::stream_audit::ended::name)})))
    });
    unsafe { finish(out, r) }
}

/// A role's energy over the session (§7): `{"role", "joules_micro", "energy_provenance",
/// "energy_uncertainty", "energy_counter_filtering", "energy_calibration_ref"}`, the class written
/// as jwp-receipts §2.4.5 writes it. `joules_micro` 0 with no class is the unmetered marker;
/// `{"role", "absent": true}` removes the role's entry. A session opens with the marker for the
/// sink decode (3) and the display (4).
///
/// # Safety
/// `s` a handle; `entry_json` NUL-terminated.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_energy(s: *const WaiSink, entry_json: *const c_char) -> c_int {
    match with_sink(s, |k| {
        let v = unsafe { json_arg(entry_json, "entry_json") }?;
        let o = v.as_object().ok_or(err_of(Reason::InputInvalid))?;
        let role = o.get("role").and_then(Value::as_u64).filter(|r| (1..=4).contains(r)).ok_or(err_of(Reason::InputInvalid))? as u8;
        if o.get("absent") == Some(&Value::Bool(true)) {
            if o.len() != 2 {
                return Err(err_of(Reason::InputInvalid));
            }
            return k.energy_absent(role).map_err(err_of);
        }
        const MEMBERS: &[&str] = &["role", "joules_micro", "energy_provenance", "energy_uncertainty", "energy_counter_filtering", "energy_calibration_ref"];
        if o.keys().any(|m| !MEMBERS.contains(&m.as_str())) {
            return Err(err_of(Reason::InputInvalid));
        }
        let joules = o.get("joules_micro").and_then(Value::as_u64).ok_or(err_of(Reason::InputInvalid))?;
        let class = crate::energy_class::label_from_json(o).map_err(|e| (WAI_ERR_SINK_INPUT, format!("native_input_invalid: {e:?}")))?;
        k.energy(role, joules, class).map_err(err_of)
    }) {
        Ok(()) => WAI_OK,
        Err(c) => c,
    }
}

/// Attach another signed record of the session (stream-audit §8.3), by its registered name.
///
/// # Safety
/// `s` a handle; `name` NUL-terminated; `bytes` valid for `len`.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_attach(s: *const WaiSink, name: *const c_char, bytes: *const u8, len: usize) -> c_int {
    match with_sink(s, |k| {
        let n = unsafe { cstr(name, "name") }?;
        let b = unsafe { bytes_arg(bytes, len, "bytes") }?.to_vec();
        k.attach(n, b).map_err(err_of)
    }) {
        Ok(()) => WAI_OK,
        Err(c) => c,
    }
}

/// The `application/cmcd` batch of the log's observations not yet taken (stream-audit §7.2), as
/// UTF-8 bytes, one report per line; empty when there is none. The host posts it to the
/// collector's `v1/cmcd` while the session runs.
///
/// # Safety
/// `s` a handle; `out` writable.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_take_cmcd(s: *const WaiSink, out: *mut WaiBuffer) -> c_int {
    let r = with_sink(s, |k| k.take_cmcd().map(String::into_bytes).map_err(err_of));
    unsafe { finish(out, r) }
}

/// End the session (stream-audit §8.2) at `t_us`, `ended` one of `ended`, `user`, `hidden`,
/// `idle`, `error`: its records are signed, and the bundle (stream-audit §9) is written to `out`
/// as the JSON the host posts to the collector's `v1/sessions`.
///
/// # Safety
/// `s` a handle; `ended` NUL-terminated; `out` writable.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_sink_end(s: *const WaiSink, ended: *const c_char, t_us: u64, out: *mut WaiBuffer) -> c_int {
    let r = with_sink(s, |k| {
        let n = unsafe { cstr(ended, "ended") }?;
        let code = ns::ended_code(n).ok_or(err_of(Reason::InputInvalid))?;
        let b = k.end(code, t_us).map_err(err_of)?;
        Ok(ns::bundle_bytes(&b))
    });
    unsafe { finish(out, r) }
}

/// Verify a bundle as the owner's collector does (stream-audit §10), for the owner `owner` (32
/// bytes) serving the channels named in `channels_json` (a JSON array of names). Writes the
/// outcome JSON of stream-audit §10.3: `{"outcome": "accept", …}` or `{"outcome": "refuse", "reason", "at"}`.
///
/// # Safety
/// `bundle` valid for `len`; `owner` valid for 32 bytes; `channels_json` NUL-terminated; `out`
/// writable.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn wai_audit_verify_bundle(bundle: *const u8, len: usize, owner: *const u8, channels_json: *const c_char, out: *mut WaiBuffer) -> c_int {
    let r = panic::catch_unwind(|| -> Result<Vec<u8>, (c_int, String)> {
        let b = unsafe { bytes_arg(bundle, len, "bundle") }?;
        if owner.is_null() {
            return Err((WAI_ERR_INVALID_ARG, "owner is null".into()));
        }
        let owner: [u8; 32] = unsafe { std::slice::from_raw_parts(owner, 32) }.try_into().expect("32 bytes");
        let ch = unsafe { json_arg(channels_json, "channels_json") }?;
        let channels: Vec<String> = serde_json::from_value(ch).map_err(|e| (WAI_ERR_SINK_INPUT, format!("native_input_invalid: channels_json: {e}")))?;
        Ok(json_bytes(&crate::stream_audit::verify_bundle(b, &owner, &channels).to_json()))
    });
    let r = match r {
        Ok(Ok(v)) => Ok(v),
        Ok(Err((code, msg))) => {
            set_err(msg);
            Err(code)
        }
        Err(_) => {
            set_err("internal panic in wai_audit_verify_bundle");
            Err(WAI_ERR_PANIC)
        }
    };
    unsafe { finish(out, r) }
}
