polyc-query 2026.8.3

//! `persona_credentials` (#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::PersonaCredential;

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

/// One decoded `persona_credentials` row: `persona_id` (supplied by the
/// caller, the same convention [`PersonaWalletRow::new`] follows) plus
/// every [`PersonaCredential`] field EXCEPT the three key-material fields —
/// see the parent module docs' "Bulk-export invariant (QRY-8)" section.
#[derive(Debug, Clone)]
pub(crate) struct PersonaCredentialRow {
    /// The persona this credential is bound to — this table's join key back
    /// to `personas`.
    pub persona_id: String,
    /// Relying-party id the enrollment ceremony ran under.
    pub rp_id: String,
    /// Origin the enrollment ceremony ran under.
    pub origin: String,
    /// Enrollment time (ms since epoch).
    pub created_at_ms: u64,
    /// Whether the credential has been revoked.
    pub revoked: bool,
    /// Revocation time (ms since epoch); `0` while active.
    pub revoked_at_ms: u64,
}

impl PersonaCredentialRow {
    /// Pair `persona_id` with `credential`'s reusable, non-key-material
    /// fields.
    #[must_use]
    pub(crate) fn new(persona_id: &str, credential: &PersonaCredential) -> Self {
        Self {
            persona_id: persona_id.to_string(),
            rp_id: credential.rp_id.clone(),
            origin: credential.origin.clone(),
            created_at_ms: credential.created_at_ms,
            revoked: credential.revoked,
            revoked_at_ms: credential.revoked_at_ms,
        }
    }
}

/// The `persona_credentials` table's Arrow schema — every column non-null;
/// see [`PersonaCredentialRow`]'s own field docs. Deliberately absent:
/// `credential_id`, `p256_public_key_sec1`, `signing_public_key` — see the
/// parent module docs' "Bulk-export invariant (QRY-8)" section.
#[must_use]
pub(crate) fn persona_credentials_schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new("persona_id", DataType::Utf8, false),
        Field::new("rp_id", DataType::Utf8, false),
        Field::new("origin", DataType::Utf8, false),
        Field::new("created_at_ms", DataType::UInt64, false),
        Field::new("revoked", DataType::Boolean, false),
        Field::new("revoked_at_ms", DataType::UInt64, false),
    ]))
}

/// Decode [`PersonaCredentialRow`]s into the `persona_credentials` table's
/// Arrow `RecordBatch`, in [`persona_credentials_schema`] column order.
///
/// # Errors
///
/// Returns [`ArrowError`] if Arrow batch construction fails.
pub(crate) fn decode_persona_credentials_batch(
    rows: &[PersonaCredentialRow],
) -> Result<RecordBatch, ArrowError> {
    let mut persona_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 36);
    let mut rp_id_b = StringBuilder::with_capacity(rows.len(), rows.len() * 16);
    let mut origin_b = StringBuilder::with_capacity(rows.len(), rows.len() * 24);
    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);
        rp_id_b.append_value(&row.rp_id);
        origin_b.append_value(&row.origin);
        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(rp_id_b.finish()),
        Arc::new(origin_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_credentials_schema(), columns)
}

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

    use super::*;

    /// A known [`PersonaCredential`] round-trips through
    /// [`PersonaCredentialRow`] and [`decode_persona_credentials_batch`],
    /// with every key-material field dropped.
    #[test]
    fn persona_credential_round_trips_to_its_row_without_key_material() {
        let credential = PersonaCredential {
            persona_id: "persona-1".to_string(),
            credential_id: vec![1, 2, 3],
            p256_public_key_sec1: vec![4, 5, 6],
            signing_public_key: vec![7, 8, 9],
            rp_id: "polychrome.example".to_string(),
            origin: "https://polychrome.example".to_string(),
            created_at_ms: 1_000,
            revoked: false,
            revoked_at_ms: 0,
            ..Default::default()
        };
        let row = PersonaCredentialRow::new("persona-1", &credential);
        let batch = decode_persona_credentials_batch(&[row]).expect("batch build");

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

        let rp_id = batch
            .column(1)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap()
            .value(0);
        assert_eq!(rp_id, "polychrome.example");
    }

    #[test]
    fn persona_credentials_schema_shape() {
        let schema = persona_credentials_schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            names,
            vec![
                "persona_id",
                "rp_id",
                "origin",
                "created_at_ms",
                "revoked",
                "revoked_at_ms",
            ]
        );
        for field in schema.fields() {
            assert!(!field.is_nullable(), "{} must be non-null", field.name());
        }
        for banned in [
            "credential_id",
            "p256_public_key_sec1",
            "signing_public_key",
        ] {
            assert!(
                !names.contains(&banned),
                "persona_credentials must never expose {banned}"
            );
        }
    }
}