//! xxHash-based row ID generation for stream tables.
//!
//! Row IDs are deterministic 64-bit hashes used to identify rows in
//! incrementally-maintained stream tables.
//!
//! ## Hash functions
//!
//! - `pg_trickle_hash`: Single-value hash using xxh64.
//! - `pg_trickle_hash_multi`: Multi-value composite hash using xxh3 streaming
//!   API (PERF-1). The xxh3 hasher processes each value and separator byte
//!   directly, eliminating the intermediate String allocation used in the
//!   previous xxh64 approach.

use pgrx::prelude::*;
use xxhash_rust::xxh64;

/// PERF-1: Separator byte injected between values in `pg_trickle_hash_multi`.
///
/// Using a single separator byte instead of a char pushes 1 byte through the
/// streaming hasher, avoiding any heap allocation for the separator.
const SEPARATOR: &[u8] = b"\x1E";
/// PERF-1: NULL sentinel used in both single and multi-value hashes.
const NULL_SENTINEL: &[u8] = b"\x00NULL\x00";

/// Compute a 64-bit xxHash row ID from a text representation.
///
/// This function is exposed as a SQL function for use in INSERT statements
/// and delta query generation.
///
/// NULL input is mapped to a deterministic sentinel (`\x00NULL\x00`) —
/// the same encoding used by [`pg_trickle_hash_multi`] — so that rows
/// with NULL-valued group keys receive a non-NULL `__pgt_row_id`.
#[pg_extern(schema = "pgtrickle", immutable, parallel_safe)]
fn pg_trickle_hash(input: Option<&str>) -> i64 {
    // Use a fixed seed for deterministic hashing
    const SEED: u64 = 0x517cc1b727220a95;
    let text = input.unwrap_or("\x00NULL\x00");
    let hash = xxh64::xxh64(text.as_bytes(), SEED);
    hash as i64
}

/// Compute a row ID by hashing multiple text values.
///
/// PERF-1: Uses the xxh3 streaming API (`update` / `digest`) to process each
/// value and separator byte directly into the hasher, eliminating the
/// intermediate `String` allocation required by the previous implementation.
/// For a 10-column composite key, this removes ~10 heap allocations per row
/// on the CDC hot path.
///
/// The hash output is identical to the previous xxh64-based implementation
/// **except** that it now uses xxh3 which produces different numeric values.
/// This change is transparent to users: row IDs are internal implementation
/// details regenerated by `REINITIALIZE` when needed.
#[pg_extern(schema = "pgtrickle", immutable, parallel_safe)]
fn pg_trickle_hash_multi(inputs: Vec<Option<String>>) -> i64 {
    // PERF-1: Use xxh3 streaming to avoid per-row String allocation.
    use xxhash_rust::xxh3::Xxh3;
    const SEED: u64 = 0x517cc1b727220a95;

    let mut hasher = Xxh3::with_seed(SEED);
    for (i, input) in inputs.iter().enumerate() {
        if i > 0 {
            hasher.update(SEPARATOR);
        }
        match input {
            Some(val) => hasher.update(val.as_bytes()),
            None => hasher.update(NULL_SENTINEL),
        }
    }

    hasher.digest() as i64
}

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

    #[test]
    fn test_hash_determinism() {
        let hash1 = xxh64::xxh64(b"hello world", 0x517cc1b727220a95);
        let hash2 = xxh64::xxh64(b"hello world", 0x517cc1b727220a95);
        assert_eq!(hash1, hash2);
    }

    #[test]
    fn test_hash_different_inputs() {
        let hash1 = xxh64::xxh64(b"hello", 0x517cc1b727220a95);
        let hash2 = xxh64::xxh64(b"world", 0x517cc1b727220a95);
        assert_ne!(hash1, hash2);
    }

    // ── PERF-1: Tests for xxh3 streaming multi-hash ───────────────────────

    /// Helper: compute multi-hash the same way pg_trickle_hash_multi does.
    fn multi_hash(inputs: &[Option<&str>]) -> i64 {
        use xxhash_rust::xxh3::Xxh3;
        const SEED: u64 = 0x517cc1b727220a95;
        let mut hasher = Xxh3::with_seed(SEED);
        for (i, input) in inputs.iter().enumerate() {
            if i > 0 {
                hasher.update(SEPARATOR);
            }
            match input {
                Some(val) => hasher.update(val.as_bytes()),
                None => hasher.update(NULL_SENTINEL),
            }
        }
        hasher.digest() as i64
    }

    #[test]
    fn test_null_handling_in_multi_hash() {
        let h1 = multi_hash(&[Some("a"), None, Some("b")]);
        let h2 = multi_hash(&[Some("a"), None, Some("c")]);
        assert_ne!(h1, h2);
    }

    #[test]
    fn test_multi_hash_determinism() {
        let h1 = multi_hash(&[Some("a"), Some("b"), Some("c")]);
        let h2 = multi_hash(&[Some("a"), Some("b"), Some("c")]);
        assert_eq!(h1, h2);
    }

    // ── pg_trickle_hash() tests via raw xxh64 (same logic, avoids pg_extern) ──

    #[test]
    fn test_pg_trickle_hash_empty_string() {
        const SEED: u64 = 0x517cc1b727220a95;
        let hash = xxh64::xxh64(b"", SEED);
        // Should produce a valid non-zero hash (xxHash of empty with non-zero seed)
        assert_ne!(hash, 0);
    }

    #[test]
    fn test_pg_trickle_hash_i64_range() {
        // Verify the cast from u64 to i64 doesn't panic
        const SEED: u64 = 0x517cc1b727220a95;
        let hash = xxh64::xxh64(b"test", SEED);
        let _ = hash as i64; // Should not panic
    }

    // ── pg_trickle_hash_multi() separator logic ────────────────────────────

    #[test]
    fn test_multi_hash_separator_prevents_collision() {
        // "ab" + "c" vs "a" + "bc" — record separator should differentiate
        let h1 = multi_hash(&[Some("ab"), Some("c")]);
        let h2 = multi_hash(&[Some("a"), Some("bc")]);
        assert_ne!(h1, h2);
    }

    #[test]
    fn test_multi_hash_null_vs_string_null() {
        // None encoded as \x00NULL\x00 vs literal string "NULL"
        let h1 = multi_hash(&[None]);
        let h2 = multi_hash(&[Some("NULL")]);
        assert_ne!(
            h1, h2,
            "NULL marker and string 'NULL' should hash differently"
        );
    }

    #[test]
    fn test_multi_hash_no_intermediate_string_alloc() {
        // Verify that hashing many values produces a valid i64
        // (This is a smoke test — the real benefit is zero-alloc, measurable
        // only in benchmarks, but we verify correctness here.)
        let h = multi_hash(&[Some("1"), Some("2"), Some("3"), None, Some("5")]);
        let _ = h; // valid i64, no panic
    }
}
