polyc-query 2026.8.3

//! `wallet_link_lifecycle` typed-table decoder (`#2123`) — the durable
//! link/renew/revoke provenance sibling of [`crate::decode::refusals`]: rows
//! come from [`polyc_facts::verified_wallet_link_lifecycle_events`], the SAME
//! signature-verified fold `crates/facts/src/wallet_link_lifecycle.rs` owns,
//! not a second, independently-written decode of `wallet_link_lifecycle`
//! payloads. See [`crate::decode::refusals`]'s own module docs for the full
//! fold-coupled-decode rationale this table follows verbatim — this module
//! states only what's different for lifecycle events.
//!
//! # Uniform keys (`#1311`)
//!
//! Same discipline as every other typed table: `partition`, `position`, and
//! `turn_id` are derived via [`polyc_proto::kinds::parse`]. Unlike a settled
//! receipt or a payment refusal, a `wallet_link_lifecycle` event is NEVER
//! tagged with a turn id at all — the ceremony that mints it is redeemed from
//! a browser POST outside any turn (see
//! [`polyc_proto::kinds::WALLET_LINK_LIFECYCLE`]'s own doc) — so `turn_id` is
//! always `NULL` for every row this table ever decodes. `event_time` is
//! deferred to #1327, unmodified from every earlier table's rationale.
//!
//! # Column selection
//!
//! Every [`polyc_crypto::approval::VerifiedWalletLinkLifecycle`] field is
//! kept: `transition` (a stable, machine-readable tag — never validated
//! against a known-tag set at decode time, see "Decode totality" below),
//! `subject`, `wallet_address`, `currency`, `chain_id`, `limit_base_units`/
//! `limit_human`/`period_secs`/`expiry_unix` (decimal strings, empty where
//! the transition carries none — `renewed` leaves `period_secs`/
//! `expiry_unix` empty, `revoked` leaves every spend-policy field empty),
//! `recipients`, `conversation_id`, `timestamp`, and `signer_public_key`.
//! Nothing here is key material: no `key_ref`, `access_key_id`,
//! `credential_id`, `webauthn_credential_id`, `pending_secret_ref`,
//! `challenge`, or `key_id` ever reaches the signed payload in the first
//! place — the recorder (`crates/control-plane/src/wallet_link.rs`) simply
//! never names one of these fields when building the signed payload, a
//! property pinned by that module's own sentinel-value tests (one per
//! `Pending*` source struct, each exhaustively destructured with no `..`),
//! not enforced by a compiler-checked exhaustive match — so there is no
//! redaction decision to make on those columns here; only
//! `signer_public_key` needs one, below.
//!
//! # Redaction: `signer_public_key` is Fleet-only
//!
//! The identical raw-table/redacted-view treatment
//! [`crate::decode::refusals`]'s module docs describe for its own
//! `signer_public_key` column: [`crate::engine`] registers this table's full
//! decode as `wallet_link_lifecycle_raw` for every scope, builds
//! `wallet_link_lifecycle` as a `CREATE VIEW` whose column list is
//! scope-dependent (every column for
//! [`Fleet`](crate::session::QueryScope::Fleet), every column except
//! `signer_public_key` for every other scope), then deregisters
//! `wallet_link_lifecycle_raw` for every non-Fleet scope. Every other
//! column stays visible at every scope: they are facts about the caller's
//! OWN wallet-link ceremony, never another participant's identity or key
//! material.
//!
//! # Decode totality (INV-W5b analogue)
//!
//! [`decode_wallet_link_lifecycle_events`] never panics and never drops a
//! structurally valid, verified lifecycle event, regardless of
//! `transition`'s value: `transition` is carried through verbatim as
//! whatever string the signed payload names, with no validation against a
//! known tag set at decode time. A future transition variant this build has
//! no tag for still decodes to a normal row — see this module's tests for
//! the `KNOWN_TRANSITIONS` set the write side mints today (test-only
//! lockstep, no shared dependency — mirrors
//! [`crate::decode::refusals`]'s own `KNOWN_REASONS`).
//!
//! # Committed-turn filter: structurally inapplicable
//!
//! Unlike [`crate::engine::REFUSALS_TABLE`] (whose exemption is a DELIBERATE
//! choice not to apply an otherwise-meaningful filter) or `SUMMARY_TABLE`
//! (whose `turn_id` is a synthetic tag the filter would be meaningless
//! against), this table's exemption is a THIRD, distinct rationale: every
//! `wallet_link_lifecycle` event is appended bare (no turn tag at all,
//! `turn_id` always `NULL` per the "Uniform keys" section above), so a
//! committed-turn `WHERE EXISTS` semijoin against `events.turn_id` would drop
//! every row of this table unconditionally — there is no turn for any row to
//! ever join against. See
//! [`crate::engine::WALLET_LINK_LIFECYCLE_TABLE`]'s own doc for where this is
//! stated on the write side of the view.

use std::sync::Arc;

use arrow::array::{ArrayRef, BinaryBuilder, StringBuilder, UInt64Builder};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow::error::ArrowError;
use arrow::record_batch::RecordBatch;
use polyc_eventlog::Event;
use polyc_proto::kinds;

/// The stable, machine-readable transition tags this build recognizes — the
/// read-side mirror of the three transitions
/// `crates/control-plane/src/wallet_link.rs`'s recording seam mints via an
/// exhaustive match. A tag NOT in this list is still a valid row (see the
/// module docs' "Decode totality" section) — this is test-only support for
/// pinning that property, not a filter any production decode path consults.
#[cfg(test)]
const KNOWN_TRANSITIONS: [&str; 3] = ["linked", "renewed", "revoked"];

/// Whether `transition` is one of [`KNOWN_TRANSITIONS`] — test-only, see
/// that constant's own doc.
#[cfg(test)]
fn is_known_transition(transition: &str) -> bool {
    KNOWN_TRANSITIONS.contains(&transition)
}

/// One decoded, SIGNATURE-VERIFIED `wallet_link_lifecycle` row.
#[derive(Debug, Clone)]
pub(crate) struct WalletLinkLifecycleRow {
    /// The journal partition this row's event was read from (`conv-{id}`).
    pub partition: String,
    /// The journal's own monotonic append position for this event.
    pub position: u64,
    /// Always `None` — see the module docs' "Uniform keys" section.
    pub turn_id: Option<String>,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::transition`].
    pub transition: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::subject`].
    pub subject: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::wallet_address`].
    pub wallet_address: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::currency`].
    pub currency: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::chain_id`].
    pub chain_id: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::limit_base_units`].
    pub limit_base_units: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::limit_human`].
    pub limit_human: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::period_secs`].
    pub period_secs: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::expiry_unix`].
    pub expiry_unix: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::recipients`].
    pub recipients: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::conversation_id`].
    pub conversation_id: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::timestamp`] —
    /// decimal string of the unix-seconds clock value the ceremony handler
    /// recorded the transition at.
    pub timestamp: String,
    /// [`polyc_crypto::approval::VerifiedWalletLinkLifecycle::signer_public_key`]
    /// — Fleet-only at registration time, see the module docs' "Redaction"
    /// section.
    pub signer_public_key: Vec<u8>,
}

/// The `wallet_link_lifecycle` typed table's full Arrow schema (every
/// column, including `signer_public_key`) — what `wallet_link_lifecycle_raw`
/// registers as, for every scope.
#[must_use]
pub(crate) fn schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new("partition", DataType::Utf8, false),
        Field::new("position", DataType::UInt64, false),
        Field::new("turn_id", DataType::Utf8, true),
        Field::new("transition", DataType::Utf8, false),
        Field::new("subject", DataType::Utf8, false),
        Field::new("wallet_address", DataType::Utf8, false),
        Field::new("currency", DataType::Utf8, false),
        Field::new("chain_id", DataType::Utf8, false),
        Field::new("limit_base_units", DataType::Utf8, false),
        Field::new("limit_human", DataType::Utf8, false),
        Field::new("period_secs", DataType::Utf8, false),
        Field::new("expiry_unix", DataType::Utf8, false),
        Field::new("recipients", DataType::Utf8, false),
        Field::new("conversation_id", DataType::Utf8, false),
        Field::new("timestamp", DataType::Utf8, false),
        Field::new("signer_public_key", DataType::Binary, false),
    ]))
}

/// Decode already-framed [`WalletLinkLifecycleRow`]s into the
/// `wallet_link_lifecycle_raw` table's Arrow `RecordBatch`, in [`schema`]
/// order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_wallet_link_lifecycle_batch(
    rows: &[WalletLinkLifecycleRow],
) -> Result<RecordBatch, ArrowError> {
    let mut partition_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut position_b = UInt64Builder::with_capacity(rows.len());
    let mut turn_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut transition_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut subject_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut wallet_address_b = StringBuilder::with_capacity(rows.len(), rows.len() * 42);
    let mut currency_b = StringBuilder::with_capacity(rows.len(), rows.len() * 42);
    let mut chain_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut limit_base_units_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut limit_human_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut period_secs_b = StringBuilder::with_capacity(rows.len(), rows.len() * 8);
    let mut expiry_unix_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut recipients_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut conversation_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut timestamp_b = StringBuilder::with_capacity(rows.len(), rows.len() * 12);
    let mut signer_public_key_b = BinaryBuilder::with_capacity(rows.len(), rows.len() * 32);

    for row in rows {
        partition_b.append_value(&row.partition);
        position_b.append_value(row.position);
        match &row.turn_id {
            Some(id) => turn_id_b.append_value(id),
            None => turn_id_b.append_null(),
        }
        transition_b.append_value(&row.transition);
        subject_b.append_value(&row.subject);
        wallet_address_b.append_value(&row.wallet_address);
        currency_b.append_value(&row.currency);
        chain_id_b.append_value(&row.chain_id);
        limit_base_units_b.append_value(&row.limit_base_units);
        limit_human_b.append_value(&row.limit_human);
        period_secs_b.append_value(&row.period_secs);
        expiry_unix_b.append_value(&row.expiry_unix);
        recipients_b.append_value(&row.recipients);
        conversation_id_b.append_value(&row.conversation_id);
        timestamp_b.append_value(&row.timestamp);
        signer_public_key_b.append_value(&row.signer_public_key);
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(partition_b.finish()),
        Arc::new(position_b.finish()),
        Arc::new(turn_id_b.finish()),
        Arc::new(transition_b.finish()),
        Arc::new(subject_b.finish()),
        Arc::new(wallet_address_b.finish()),
        Arc::new(currency_b.finish()),
        Arc::new(chain_id_b.finish()),
        Arc::new(limit_base_units_b.finish()),
        Arc::new(limit_human_b.finish()),
        Arc::new(period_secs_b.finish()),
        Arc::new(expiry_unix_b.finish()),
        Arc::new(recipients_b.finish()),
        Arc::new(conversation_id_b.finish()),
        Arc::new(timestamp_b.finish()),
        Arc::new(signer_public_key_b.finish()),
    ];
    RecordBatch::try_new(schema(), columns)
}

/// Filter `partition`'s framed `events` to `wallet_link_lifecycle` rows, fold
/// each payload through the SHARED verified fold
/// (`polyc_facts::verified_wallet_link_lifecycle_events`), and pair a
/// successfully-verified event with that row's uniform key columns.
///
/// Mirrors [`crate::decode::refusals::decode_refusals_events`]'s own
/// one-event-at-a-time loop for the identical reason: the shared fold walks
/// a slice of `Event` with no position of its own, so pairing happens one
/// event at a time rather than losing the association across the whole
/// partition slice.
///
/// A payload that fails to verify — malformed JSON, a signature that
/// doesn't check out, or a signer outside `trusted_signers` — is silently
/// skipped: there is no row for an event that doesn't verify, forged or
/// otherwise.
#[must_use]
pub(crate) fn decode_wallet_link_lifecycle_events(
    partition: &str,
    events: &[(u64, Event)],
    trusted_signers: &[Vec<u8>],
) -> Vec<WalletLinkLifecycleRow> {
    events
        .iter()
        .filter_map(|(position, event)| {
            let (base, turn_id) = kinds::parse(&event.kind);
            if base != kinds::WALLET_LINK_LIFECYCLE {
                return None;
            }
            let verified = polyc_facts::verified_wallet_link_lifecycle_events(
                std::slice::from_ref(event),
                trusted_signers,
            )
            .next()?;
            Some(WalletLinkLifecycleRow {
                partition: partition.to_string(),
                position: *position,
                turn_id: turn_id.map(|id| id.to_string()),
                transition: verified.transition,
                subject: verified.subject,
                wallet_address: verified.wallet_address,
                currency: verified.currency,
                chain_id: verified.chain_id,
                limit_base_units: verified.limit_base_units,
                limit_human: verified.limit_human,
                period_secs: verified.period_secs,
                expiry_unix: verified.expiry_unix,
                recipients: verified.recipients,
                conversation_id: verified.conversation_id,
                timestamp: verified.timestamp,
                signer_public_key: verified.signer_public_key,
            })
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use arrow::array::Array as _;
    use polyc_crypto::approval::{
        ApprovalSigner, WalletLinkLifecyclePayload, wallet_link_lifecycle_payload,
    };

    use super::*;

    fn signed_lifecycle_event(
        signer: &ApprovalSigner,
        transition: &str,
        conversation_id: &str,
    ) -> Vec<u8> {
        let (payload, sig, pk) = wallet_link_lifecycle_payload(
            &WalletLinkLifecyclePayload {
                kind: kinds::WALLET_LINK_LIFECYCLE,
                transition,
                subject: "persona-1",
                wallet_address: "0x1111111111111111111111111111111111111111",
                currency: "0x2222222222222222222222222222222222222222",
                chain_id: "42431",
                limit_base_units: "5000000",
                limit_human: "5",
                period_secs: "86400",
                expiry_unix: "1780086400",
                recipients: "",
                conversation_id,
                timestamp: "1780000000",
            },
            signer,
        );
        let _ = (sig, pk);
        payload
    }

    #[test]
    fn schema_shape() {
        let schema = schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            names,
            vec![
                "partition",
                "position",
                "turn_id",
                "transition",
                "subject",
                "wallet_address",
                "currency",
                "chain_id",
                "limit_base_units",
                "limit_human",
                "period_secs",
                "expiry_unix",
                "recipients",
                "conversation_id",
                "timestamp",
                "signer_public_key",
            ]
        );
    }

    #[test]
    fn decode_wallet_link_lifecycle_batch_round_trips() {
        let rows = vec![WalletLinkLifecycleRow {
            partition: "conv-a".to_string(),
            position: 5,
            turn_id: None,
            transition: "linked".to_string(),
            subject: "persona-1".to_string(),
            wallet_address: "0x1111111111111111111111111111111111111111".to_string(),
            currency: "0x2222222222222222222222222222222222222222".to_string(),
            chain_id: "42431".to_string(),
            limit_base_units: "5000000".to_string(),
            limit_human: "5".to_string(),
            period_secs: "86400".to_string(),
            expiry_unix: "1780086400".to_string(),
            recipients: String::new(),
            conversation_id: "conv-a".to_string(),
            timestamp: "1780000000".to_string(),
            signer_public_key: vec![1, 2, 3, 4],
        }];
        let batch = decode_wallet_link_lifecycle_batch(&rows).expect("batch build");
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), schema());

        let transition = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(transition.value(0), "linked");
    }

    #[test]
    fn decode_wallet_link_lifecycle_events_verifies_and_decodes() {
        let signer = ApprovalSigner::from_seed(1);
        let bytes = signed_lifecycle_event(&signer, "linked", "conv-real");
        let events = vec![
            (1, Event::new(kinds::TURN_START, Vec::new())),
            (2, Event::new(kinds::WALLET_LINK_LIFECYCLE, bytes)),
        ];
        let trusted_signers = vec![signer.public_key_bytes()];

        let decoded = decode_wallet_link_lifecycle_events("conv-real", &events, &trusted_signers);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].partition, "conv-real");
        assert_eq!(decoded[0].position, 2);
        assert_eq!(decoded[0].turn_id, None);
        assert_eq!(decoded[0].transition, "linked");
        assert_eq!(decoded[0].conversation_id, "conv-real");
        assert_eq!(decoded[0].signer_public_key, signer.public_key_bytes());
    }

    /// A bare kind never carries a turn id — even if a stray `:` suffix
    /// somehow appeared, this event kind's own recorder never mints one
    /// (INV analogue: no synthetic turn UUID).
    #[test]
    fn decode_wallet_link_lifecycle_events_bare_kind_has_no_turn_id() {
        let signer = ApprovalSigner::from_seed(5);
        let bytes = signed_lifecycle_event(&signer, "revoked", "conv-bare");
        let events = vec![(7, Event::new(kinds::WALLET_LINK_LIFECYCLE, bytes))];
        let trusted_signers = vec![signer.public_key_bytes()];

        let decoded = decode_wallet_link_lifecycle_events("conv-bare", &events, &trusted_signers);
        assert_eq!(decoded.len(), 1);
        assert_eq!(decoded[0].turn_id, None);
    }

    /// INV-W5b analogue: an unrecognized transition tag (a variant a future
    /// binary added, which this build has no [`KNOWN_TRANSITIONS`] entry
    /// for) still decodes to a normal row — never a panic, never a dropped
    /// event.
    #[test]
    fn decode_wallet_link_lifecycle_events_keeps_an_unknown_transition_tag() {
        let signer = ApprovalSigner::from_seed(2);
        let bytes = signed_lifecycle_event(&signer, "some_future_transition", "conv-unknown");
        let events = vec![(1, Event::new(kinds::WALLET_LINK_LIFECYCLE, bytes))];
        let trusted_signers = vec![signer.public_key_bytes()];

        let decoded =
            decode_wallet_link_lifecycle_events("conv-unknown", &events, &trusted_signers);
        assert_eq!(
            decoded.len(),
            1,
            "an unknown transition tag must not be dropped"
        );
        assert_eq!(decoded[0].transition, "some_future_transition");
        assert!(!is_known_transition(&decoded[0].transition));
    }

    #[test]
    fn decode_wallet_link_lifecycle_events_drops_an_event_from_an_untrusted_signer() {
        let trusted = ApprovalSigner::from_seed(3);
        let untrusted = ApprovalSigner::from_seed(4);
        let forged = signed_lifecycle_event(&untrusted, "linked", "conv-forged");
        let events = vec![(1, Event::new(kinds::WALLET_LINK_LIFECYCLE, forged))];
        let trusted_signers = vec![trusted.public_key_bytes()];

        let decoded = decode_wallet_link_lifecycle_events("conv-forged", &events, &trusted_signers);
        assert_eq!(
            decoded.len(),
            0,
            "an event signed by a key outside trusted_signers must never appear as a row"
        );
    }

    #[test]
    fn decode_wallet_link_lifecycle_events_drops_a_malformed_payload() {
        let events = vec![(
            1,
            Event::new(kinds::WALLET_LINK_LIFECYCLE, vec![0xFF, 0xFE, 0xFD]),
        )];
        let decoded = decode_wallet_link_lifecycle_events("conv-corrupt", &events, &[]);
        assert_eq!(decoded.len(), 0);
    }

    #[test]
    fn unrelated_kind_is_not_decoded_as_a_wallet_link_lifecycle_event() {
        let signer = ApprovalSigner::from_seed(6);
        let bytes = signed_lifecycle_event(&signer, "linked", "conv-unrelated");
        let events = vec![(1, Event::new(kinds::USAGE, bytes))];
        let trusted_signers = vec![signer.public_key_bytes()];

        let decoded =
            decode_wallet_link_lifecycle_events("conv-unrelated", &events, &trusted_signers);
        assert_eq!(
            decoded.len(),
            0,
            "a usage-kind event must never decode as a wallet-link lifecycle event"
        );
    }

    /// [`KNOWN_TRANSITIONS`] must stay in lockstep with the three transitions
    /// the write side (`crates/control-plane/src/wallet_link.rs`) actually
    /// mints — a test-only mirror, not a shared dependency (see the module
    /// docs' "Decode totality" section for why).
    #[test]
    fn known_transitions_cover_every_recording_seam_tag() {
        for tag in ["linked", "renewed", "revoked"] {
            assert!(
                is_known_transition(tag),
                "{tag} must be a known transition tag"
            );
        }
        assert!(!is_known_transition("not_a_real_tag"));
    }
}