polyc-query 2026.8.3

//! `persona_wallets` (#1578, Phase D) — see the parent module's own doc for
//! the full column-selection and QRY-8 bulk-export rationale shared across
//! every table in `crate::decode::persona`.

use std::sync::Arc;

use arrow::array::{ArrayRef, BooleanBuilder, StringBuilder, UInt64Builder};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow::error::ArrowError;
use arrow::record_batch::RecordBatch;
use polyc_proto::proto::polychrome::persona::v1::WalletLink;

// Doc-link-only import: `ParticipationRow::new` is referenced from
// `PersonaWalletRow`'s doc comment below but never named in code here.
#[allow(unused_imports)]
use super::core::ParticipationRow;

/// One decoded `persona_wallets` row: `persona_id` (supplied by the caller —
/// [`WalletLink`] already carries its own `persona_id` field, but every
/// other row type in this crate's `persona` module takes it as a separate
/// constructor argument for the same reason [`ParticipationRow::new`] does,
/// so this stays consistent rather than a special case) plus the
/// [`WalletLink`] fields a query surface can answer questions about.
///
/// Absent, and each for its own reason: `key_ref` (a custody handle — see the
/// parent module docs' "Column selection" section), `webauthn_credential_id`
/// (an authenticator identifier this surface has no question for), and
/// `delegated_key_address`. The last is non-secret and could be exposed; it
/// is left out only because nothing queries it yet. Adding it would let an
/// operator enumerate the links that predate it, which is what POLY-159
/// needs — that belongs in a change that adds the column deliberately, not
/// as a side effect of this one.
#[derive(Debug, Clone)]
pub(crate) struct PersonaWalletRow {
    /// The persona this wallet is linked to — this table's join key back to
    /// `personas`.
    pub persona_id: String,
    /// Onchain wallet address the delegated key acts for (0x…). Non-secret.
    pub wallet_address: String,
    /// Settlement-token contract address the delegation is scoped to (0x…).
    pub currency: String,
    /// Unix time (seconds) after which the delegated key expires; `0` = no
    /// recorded expiry.
    pub expiry_unix: u64,
    /// Link creation time (ms since epoch).
    pub created_at_ms: u64,
    /// Whether the link has been revoked.
    pub revoked: bool,
    /// Revocation time (ms since epoch); `0` while active.
    pub revoked_at_ms: u64,
}

impl PersonaWalletRow {
    /// Pair `persona_id` with `wallet`'s queryable fields; see
    /// [`PersonaWalletRow`] for what is left out and why.
    #[must_use]
    pub(crate) fn new(persona_id: &str, wallet: &WalletLink) -> Self {
        Self {
            persona_id: persona_id.to_string(),
            wallet_address: wallet.wallet_address.clone(),
            currency: wallet.currency.clone(),
            expiry_unix: wallet.expiry_unix,
            created_at_ms: wallet.created_at_ms,
            revoked: wallet.revoked,
            revoked_at_ms: wallet.revoked_at_ms,
        }
    }
}

/// The `persona_wallets` table's Arrow schema — every column non-null; see
/// [`PersonaWalletRow`]'s own field docs for the full absent-column list and
/// its reasons. `key_ref` in particular is barred by the parent module docs'
/// "Bulk-export invariant (QRY-8)" section, not merely unqueried.
#[must_use]
pub(crate) fn persona_wallets_schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new("persona_id", DataType::Utf8, false),
        Field::new("wallet_address", DataType::Utf8, false),
        Field::new("currency", DataType::Utf8, false),
        Field::new("expiry_unix", DataType::UInt64, false),
        Field::new("created_at_ms", DataType::UInt64, false),
        Field::new("revoked", DataType::Boolean, false),
        Field::new("revoked_at_ms", DataType::UInt64, false),
    ]))
}

/// Decode [`PersonaWalletRow`]s into the `persona_wallets` table's Arrow
/// `RecordBatch`, in [`persona_wallets_schema`] column order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_persona_wallets_batch(
    rows: &[PersonaWalletRow],
) -> Result<RecordBatch, ArrowError> {
    let mut persona_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    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 expiry_unix_b = UInt64Builder::with_capacity(rows.len());
    let mut created_at_ms_b = UInt64Builder::with_capacity(rows.len());
    let mut revoked_b = BooleanBuilder::with_capacity(rows.len());
    let mut revoked_at_ms_b = UInt64Builder::with_capacity(rows.len());

    for row in rows {
        persona_id_b.append_value(&row.persona_id);
        wallet_address_b.append_value(&row.wallet_address);
        currency_b.append_value(&row.currency);
        expiry_unix_b.append_value(row.expiry_unix);
        created_at_ms_b.append_value(row.created_at_ms);
        revoked_b.append_value(row.revoked);
        revoked_at_ms_b.append_value(row.revoked_at_ms);
    }

    let columns: Vec<ArrayRef> = vec![
        Arc::new(persona_id_b.finish()),
        Arc::new(wallet_address_b.finish()),
        Arc::new(currency_b.finish()),
        Arc::new(expiry_unix_b.finish()),
        Arc::new(created_at_ms_b.finish()),
        Arc::new(revoked_b.finish()),
        Arc::new(revoked_at_ms_b.finish()),
    ];
    RecordBatch::try_new(persona_wallets_schema(), columns)
}

#[cfg(test)]
mod tests {
    use arrow::array::Array as _;

    use super::*;

    /// A known [`WalletLink`] round-trips through [`PersonaWalletRow`] and
    /// [`decode_persona_wallets_batch`] to exactly its own row, `key_ref`
    /// excluded.
    #[test]
    fn persona_wallet_round_trips_to_its_row() {
        let wallet = WalletLink {
            persona_id: "persona-1".to_string(),
            wallet_address: "0xabc".to_string(),
            currency: "0xusdc".to_string(),
            expiry_unix: 1_700_000_000,
            key_ref: "keychain://persona-1".to_string(),
            created_at_ms: 1_000,
            revoked: true,
            revoked_at_ms: 2_000,
            ..Default::default()
        };
        let row = PersonaWalletRow::new("persona-1", &wallet);
        let batch = decode_persona_wallets_batch(&[row]).expect("batch build");

        assert_eq!(batch.num_rows(), 1);
        assert_eq!(batch.schema(), persona_wallets_schema());

        let string_col = |i: usize| {
            batch
                .column(i)
                .as_any()
                .downcast_ref::<arrow::array::StringArray>()
                .unwrap()
                .value(0)
                .to_owned()
        };
        assert_eq!(string_col(0), "persona-1");
        assert_eq!(string_col(1), "0xabc");
        assert_eq!(string_col(2), "0xusdc");
        let expiry_unix = batch
            .column(3)
            .as_any()
            .downcast_ref::<arrow::array::UInt64Array>()
            .unwrap()
            .value(0);
        assert_eq!(expiry_unix, 1_700_000_000);
        let created_at_ms = batch
            .column(4)
            .as_any()
            .downcast_ref::<arrow::array::UInt64Array>()
            .unwrap()
            .value(0);
        assert_eq!(created_at_ms, 1_000);
        let revoked = batch
            .column(5)
            .as_any()
            .downcast_ref::<arrow::array::BooleanArray>()
            .unwrap()
            .value(0);
        assert!(revoked);
        let revoked_at_ms = batch
            .column(6)
            .as_any()
            .downcast_ref::<arrow::array::UInt64Array>()
            .unwrap()
            .value(0);
        assert_eq!(revoked_at_ms, 2_000);
    }

    #[test]
    fn persona_wallets_schema_shape() {
        let schema = persona_wallets_schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            names,
            vec![
                "persona_id",
                "wallet_address",
                "currency",
                "expiry_unix",
                "created_at_ms",
                "revoked",
                "revoked_at_ms",
            ]
        );
        for field in schema.fields() {
            assert!(!field.is_nullable(), "{} must be non-null", field.name());
        }
        assert!(
            !names.contains(&"key_ref"),
            "persona_wallets must never expose key_ref"
        );
    }
}