khadhroony-solana-project/crates/ksp-job-backfill-lib/src/conversion.rs

// file: crates/ksp-job-backfill-lib/src/conversion.rs
// version: 4

use sha2::Digest; // rust-rules: trait-import

/// KSP-owned source-independent RAW transaction format identifier produced by this Backfill vertical.
pub const RAW_TRANSACTION_FORMAT_ID: &str = "ksp.solana.raw_transaction";
/// Initial KSP-owned RAW transaction format version produced by this Backfill vertical.
pub const RAW_TRANSACTION_FORMAT_VERSION: u32 = 1;

const RAW_TRANSACTION_METHOD_CODE: &str = "getTransaction";
const RAW_TRANSACTION_OBSERVATION_CONTRACT_VERSION: u32 = 1;
const RAW_TRANSACTION_PROTOCOL_CODE: &str = "solana.http.json_rpc";

/// Complete in-memory RAW transaction acquisition ready for the later Store persistence tranche.
#[derive(Debug)]
pub struct BackfillRawAcquisition {
    inner: Box<BackfillRawAcquisitionInner>,
}

#[derive(Debug)]
struct BackfillRawAcquisitionInner {
    transaction: ksp_store_lib::RawTransaction,
    observation: ksp_store_lib::RawTransactionObservation,
}

impl crate::BackfillRawAcquisition {
    /// Returns the canonical RAW transaction produced from the typed Transport response.
    #[must_use]
    pub const fn transaction(&self) -> &ksp_store_lib::RawTransaction {
        return &self.inner.transaction;
    }

    /// Returns the acquisition observation whose provenance records the actual successful endpoint.
    #[must_use]
    pub const fn observation(&self) -> &ksp_store_lib::RawTransactionObservation {
        return &self.inner.observation;
    }

    /// Consumes the in-memory acquisition into the canonical transaction and its observation.
    #[must_use]
    pub fn into_parts(self) -> (ksp_store_lib::RawTransaction, ksp_store_lib::RawTransactionObservation) {
        let inner = *self.inner;
        return (inner.transaction, inner.observation);
    }
}

/// Result of hydrating one deterministic Backfill candidate through observed `getTransaction`.
#[derive(Debug)]
pub enum BackfillHydrationOutcome {
    /// The RPC returned one complete transaction and conversion produced canonical RAW plus provenance.
    Available(crate::BackfillRawAcquisition),
    /// The RPC returned JSON `null`; only the canonical transaction identity exists and no provenance is fabricated.
    Missing(ksp_store_lib::RawTransactionReference),
}

impl crate::BackfillHydrationOutcome {
    /// Returns the network-scoped transaction identity represented by this hydration outcome.
    #[must_use]
    pub fn reference(&self) -> &ksp_store_lib::RawTransactionReference {
        return match self {
            Self::Available(acquisition) => acquisition.transaction().reference(),
            Self::Missing(reference) => reference,
        };
    }

    /// Returns whether `getTransaction` returned JSON `null`.
    #[must_use]
    pub const fn is_missing(&self) -> bool {
        return matches!(self, Self::Missing(_));
    }
}

/// Hydrates one candidate with the typed observed Transport path and converts it to canonical RAW v1.
///
/// The caller supplies the local receipt timestamp because wall-clock ownership remains outside this
/// pure conversion tranche. Transport retains endpoint selection and retry. This function never
/// persists to Store; persistence begins in `pre.007`.
pub async fn hydrate_backfill_candidate(
    transport: &ksp_onchain_transport_lib::HttpTransportPool,
    request: &crate::BackfillRequest,
    candidate: &crate::BackfillCandidate,
    received_at: ksp_store_lib::RawTimestamp,
) -> ksp_core_lib::Result<crate::BackfillHydrationOutcome> {
    let reference = canonical_reference(request, candidate);
    let reference = match reference {
        std::result::Result::Ok(reference) => reference,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let config = ksp_onchain_transport_lib::SolanaGetTransactionConfig::new(
        std::option::Option::Some(request.commitment().transport()),
        std::option::Option::Some(ksp_onchain_transport_lib::SolanaTransactionEncoding::Base64),
        std::option::Option::Some(0),
    );
    let observed = transport.get_transaction_observed(request.role(), candidate.identity().signature().as_str(), std::option::Option::Some(&config)).await;
    let observed = match observed {
        std::result::Result::Ok(observed) => observed,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let provider = observed.provider().as_str().to_owned();
    let endpoint = observed.endpoint_name().to_owned();
    let transaction = observed.into_value();
    let transaction = match transaction {
        std::option::Option::Some(transaction) => transaction,
        std::option::Option::None => return std::result::Result::Ok(crate::BackfillHydrationOutcome::Missing(reference)),
    };
    let fields = CanonicalTransactionFields {
        slot: transaction.slot(),
        block_time: transaction.block_time(),
        transaction: transaction.transaction(),
        meta: transaction.meta(),
        version: transaction.version(),
        transaction_index: transaction.transaction_index(),
    };
    let acquisition = convert_available_fields(request, reference, fields, provider.as_str(), endpoint.as_str(), received_at);
    return match acquisition {
        std::result::Result::Ok(acquisition) => std::result::Result::Ok(crate::BackfillHydrationOutcome::Available(acquisition)),
        std::result::Result::Err(error) => std::result::Result::Err(error),
    };
}

/// Decodes one validated Base58 signature to exactly 64 canonical bytes without a Solana SDK dependency.
pub(crate) fn decode_backfill_signature(signature: &crate::BackfillSignature) -> ksp_core_lib::Result<ksp_store_lib::RawTransactionSignature> {
    let text = signature.as_str().as_bytes();
    let mut decoded = [0_u8; 64];
    let mut leading_zeroes = 0_usize;
    for byte in text {
        if *byte != b'1' {
            break;
        }
        leading_zeroes += 1;
    }
    for byte in text {
        let digit = match base58_digit(*byte) {
            std::option::Option::Some(digit) => digit,
            std::option::Option::None => return std::result::Result::Err(conversion_error("signature")),
        };
        let mut carry = u32::from(digit);
        for output in decoded.iter_mut().rev() {
            let value = (u32::from(*output) * 58) + carry;
            *output = (value & 0xff) as u8;
            carry = value >> 8;
        }
        if carry != 0 {
            return std::result::Result::Err(conversion_error("signature"));
        }
    }
    let significant_len = match decoded.iter().position(|byte| return *byte != 0) {
        std::option::Option::Some(index) => decoded.len() - index,
        std::option::Option::None => 0,
    };
    if leading_zeroes + significant_len != decoded.len() {
        return std::result::Result::Err(conversion_error("signature"));
    }
    return std::result::Result::Ok(ksp_store_lib::RawTransactionSignature::new(decoded));
}

struct CanonicalTransactionFields<'a> {
    slot: u64,
    block_time: std::option::Option<i64>,
    transaction: &'a ksp_onchain_transport_lib::SolanaEncodedTransaction,
    meta: &'a ksp_onchain_transport_lib::SolanaWireField<serde_json::Value>,
    version: &'a ksp_onchain_transport_lib::SolanaWireField<ksp_onchain_transport_lib::SolanaTransactionVersion>,
    transaction_index: &'a ksp_onchain_transport_lib::SolanaWireField<u32>,
}

fn canonical_reference(request: &crate::BackfillRequest, candidate: &crate::BackfillCandidate) -> ksp_core_lib::Result<ksp_store_lib::RawTransactionReference> {
    if candidate.identity().network() != request.network() {
        return std::result::Result::Err(conversion_error("candidate.network"));
    }
    let signature = crate::decode_backfill_signature(candidate.identity().signature());
    let signature = match signature {
        std::result::Result::Ok(signature) => signature,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    return std::result::Result::Ok(ksp_store_lib::RawTransactionReference::new(candidate.identity().network().clone(), signature));
}

fn convert_available_fields(
    request: &crate::BackfillRequest,
    reference: ksp_store_lib::RawTransactionReference,
    fields: CanonicalTransactionFields<'_>,
    provider: &str,
    endpoint: &str,
    received_at: ksp_store_lib::RawTimestamp,
) -> ksp_core_lib::Result<crate::BackfillRawAcquisition> {
    let block_time = convert_block_time(fields.block_time);
    let block_time = match block_time {
        std::result::Result::Ok(block_time) => block_time,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let bytes = canonical_payload_bytes(&fields);
    let bytes = match bytes {
        std::result::Result::Ok(bytes) => bytes,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let hash: [u8; 32] = sha2::Sha256::digest(bytes.as_slice()).into();
    let format_id = ksp_store_lib::RawFormatId::new(crate::RAW_TRANSACTION_FORMAT_ID);
    let format_id = match format_id {
        std::result::Result::Ok(format_id) => format_id,
        std::result::Result::Err(_) => return std::result::Result::Err(conversion_error("payload.format_id")),
    };
    let payload = ksp_store_lib::RawPayload::try_new(
        format_id,
        crate::RAW_TRANSACTION_FORMAT_VERSION,
        bytes.into_boxed_slice(),
        ksp_store_lib::RawContentHash::new(hash),
    );
    let payload = match payload {
        std::result::Result::Ok(payload) => payload,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let provenance = build_provenance(request, provider, endpoint, received_at);
    let provenance = match provenance {
        std::result::Result::Ok(provenance) => provenance,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let observation_key = observation_key(request, &reference, provider, endpoint);
    let transaction = ksp_store_lib::RawTransaction::new(reference.clone(), fields.slot, block_time, payload);
    let observation = ksp_store_lib::RawTransactionObservation::new(observation_key, reference, provenance);
    return std::result::Result::Ok(crate::BackfillRawAcquisition { inner: Box::new(BackfillRawAcquisitionInner { transaction, observation }) });
}

fn convert_block_time(value: std::option::Option<i64>) -> ksp_core_lib::Result<std::option::Option<ksp_store_lib::RawTimestamp>> {
    let seconds = match value {
        std::option::Option::Some(seconds) => seconds,
        std::option::Option::None => return std::result::Result::Ok(std::option::Option::None),
    };
    let seconds = match u64::try_from(seconds) {
        std::result::Result::Ok(seconds) => seconds,
        std::result::Result::Err(_) => return std::result::Result::Err(conversion_error("block_time")),
    };
    let millis = match seconds.checked_mul(1_000) {
        std::option::Option::Some(millis) => millis,
        std::option::Option::None => return std::result::Result::Err(conversion_error("block_time")),
    };
    let timestamp = ksp_store_lib::RawTimestamp::from_unix_millis(millis);
    return match timestamp {
        std::result::Result::Ok(timestamp) => std::result::Result::Ok(std::option::Option::Some(timestamp)),
        std::result::Result::Err(_) => std::result::Result::Err(conversion_error("block_time")),
    };
}

fn canonical_payload_bytes(fields: &CanonicalTransactionFields<'_>) -> ksp_core_lib::Result<std::vec::Vec<u8>> {
    let (transaction_data, transaction_encoding) = match fields.transaction {
        ksp_onchain_transport_lib::SolanaEncodedTransaction::Binary { data, encoding } => {
            if *encoding != ksp_onchain_transport_lib::SolanaTransactionBinaryEncoding::Base64 {
                return std::result::Result::Err(conversion_error("transaction.encoding"));
            }
            (data.as_str(), "base64")
        },
        ksp_onchain_transport_lib::SolanaEncodedTransaction::LegacyBinary(_) | ksp_onchain_transport_lib::SolanaEncodedTransaction::Json(_) => {
            return std::result::Result::Err(conversion_error("transaction.encoding"));
        },
    };
    let mut bytes = std::vec::Vec::new();
    bytes.extend_from_slice(b"{\"transaction\":[");
    if let std::result::Result::Err(error) = append_json_string(&mut bytes, transaction_data) {
        return std::result::Result::Err(error);
    }
    bytes.push(b',');
    if let std::result::Result::Err(error) = append_json_string(&mut bytes, transaction_encoding) {
        return std::result::Result::Err(error);
    }
    bytes.push(b']');
    if let std::result::Result::Err(error) = append_wire_value(&mut bytes, "meta", fields.meta, |output, value| return append_canonical_json(output, value)) {
        return std::result::Result::Err(error);
    }
    let version_result = append_wire_value(&mut bytes, "version", fields.version, |output, value| {
        return match value {
            ksp_onchain_transport_lib::SolanaTransactionVersion::Legacy => append_json_string(output, "legacy"),
            ksp_onchain_transport_lib::SolanaTransactionVersion::Number(number) => {
                output.extend_from_slice(number.to_string().as_bytes());
                std::result::Result::Ok(())
            },
        };
    });
    if let std::result::Result::Err(error) = version_result {
        return std::result::Result::Err(error);
    }
    let transaction_index_result = append_wire_value(&mut bytes, "transactionIndex", fields.transaction_index, |output, value| {
        output.extend_from_slice(value.to_string().as_bytes());
        return std::result::Result::Ok(());
    });
    if let std::result::Result::Err(error) = transaction_index_result {
        return std::result::Result::Err(error);
    }
    bytes.push(b'}');
    return std::result::Result::Ok(bytes);
}

fn append_wire_value<T, F>(
    output: &mut std::vec::Vec<u8>,
    key: &str,
    field: &ksp_onchain_transport_lib::SolanaWireField<T>,
    mut append_value: F,
) -> ksp_core_lib::Result<()>
where
    F: FnMut(&mut std::vec::Vec<u8>, &T) -> ksp_core_lib::Result<()>,
{
    return match field {
        ksp_onchain_transport_lib::SolanaWireField::Omitted => std::result::Result::Ok(()),
        ksp_onchain_transport_lib::SolanaWireField::Null => {
            output.push(b',');
            let key_result = append_json_string(output, key);
            if let std::result::Result::Err(error) = key_result {
                return std::result::Result::Err(error);
            }
            output.extend_from_slice(b":null");
            std::result::Result::Ok(())
        },
        ksp_onchain_transport_lib::SolanaWireField::Value(value) => {
            output.push(b',');
            let key_result = append_json_string(output, key);
            if let std::result::Result::Err(error) = key_result {
                return std::result::Result::Err(error);
            }
            output.push(b':');
            append_value(output, value)
        },
    };
}

fn append_canonical_json(output: &mut std::vec::Vec<u8>, value: &serde_json::Value) -> ksp_core_lib::Result<()> {
    return match value {
        serde_json::Value::Null => {
            output.extend_from_slice(b"null");
            std::result::Result::Ok(())
        },
        serde_json::Value::Bool(value) => {
            if *value {
                output.extend_from_slice(b"true");
            } else {
                output.extend_from_slice(b"false");
            }
            std::result::Result::Ok(())
        },
        serde_json::Value::Number(value) => {
            output.extend_from_slice(value.to_string().as_bytes());
            std::result::Result::Ok(())
        },
        serde_json::Value::String(value) => append_json_string(output, value.as_str()),
        serde_json::Value::Array(values) => {
            output.push(b'[');
            for (index, item) in values.iter().enumerate() {
                if index != 0 {
                    output.push(b',');
                }
                let item_result = append_canonical_json(output, item);
                if let std::result::Result::Err(error) = item_result {
                    return std::result::Result::Err(error);
                }
            }
            output.push(b']');
            std::result::Result::Ok(())
        },
        serde_json::Value::Object(values) => {
            output.push(b'{');
            let mut keys = values.keys().collect::<std::vec::Vec<_>>();
            keys.sort_unstable();
            for (index, key) in keys.iter().enumerate() {
                if index != 0 {
                    output.push(b',');
                }
                let key_result = append_json_string(output, key.as_str());
                if let std::result::Result::Err(error) = key_result {
                    return std::result::Result::Err(error);
                }
                output.push(b':');
                let item = values.get(key.as_str());
                let item = match item {
                    std::option::Option::Some(item) => item,
                    std::option::Option::None => return std::result::Result::Err(conversion_error("payload.meta")),
                };
                let item_result = append_canonical_json(output, item);
                if let std::result::Result::Err(error) = item_result {
                    return std::result::Result::Err(error);
                }
            }
            output.push(b'}');
            std::result::Result::Ok(())
        },
    };
}

fn append_json_string(output: &mut std::vec::Vec<u8>, value: &str) -> ksp_core_lib::Result<()> {
    let encoded = serde_json::to_vec(value);
    return match encoded {
        std::result::Result::Ok(encoded) => {
            output.extend_from_slice(encoded.as_slice());
            std::result::Result::Ok(())
        },
        std::result::Result::Err(_) => std::result::Result::Err(conversion_error("payload.json")),
    };
}

fn build_provenance(
    request: &crate::BackfillRequest,
    provider: &str,
    endpoint: &str,
    received_at: ksp_store_lib::RawTimestamp,
) -> ksp_core_lib::Result<ksp_store_lib::RawAcquisitionProvenance> {
    let provider = match provenance_code(provider, "provenance.provider") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let protocol = match provenance_code(RAW_TRANSACTION_PROTOCOL_CODE, "provenance.protocol") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let method = match provenance_code(RAW_TRANSACTION_METHOD_CODE, "provenance.method") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let endpoint = match provenance_code(endpoint, "provenance.endpoint") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let commitment = match provenance_code(request.commitment().code(), "provenance.commitment") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let capture_session = match provenance_code(request.job_id().as_str(), "provenance.capture_session") {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let provenance = ksp_store_lib::RawAcquisitionProvenance::new(provider, protocol, method, ksp_store_lib::RawAcquisitionOrigin::Backfill, received_at)
        .with_endpoint_id(endpoint)
        .with_commitment(commitment)
        .with_capture_session_id(capture_session);
    return std::result::Result::Ok(provenance);
}

fn provenance_code(value: &str, field: &'static str) -> ksp_core_lib::Result<ksp_store_lib::RawProvenanceCode> {
    let code = ksp_store_lib::RawProvenanceCode::new(value);
    return match code {
        std::result::Result::Ok(code) => std::result::Result::Ok(code),
        std::result::Result::Err(_) => std::result::Result::Err(conversion_error(field)),
    };
}

fn observation_key(
    request: &crate::BackfillRequest,
    reference: &ksp_store_lib::RawTransactionReference,
    provider: &str,
    endpoint: &str,
) -> ksp_store_lib::RawObservationKey {
    let mut hasher = sha2::Sha256::new();
    hasher.update(b"ksp.job.backfill.raw_transaction_observation.v1\0");
    hash_bytes(&mut hasher, request.job_id().as_str().as_bytes());
    hash_bytes(&mut hasher, request.scope_fingerprint().as_bytes());
    hash_bytes(&mut hasher, reference.signature().as_bytes());
    hash_bytes(&mut hasher, provider.as_bytes());
    hash_bytes(&mut hasher, endpoint.as_bytes());
    hash_bytes(&mut hasher, request.commitment().code().as_bytes());
    hasher.update(RAW_TRANSACTION_OBSERVATION_CONTRACT_VERSION.to_be_bytes());
    let bytes: [u8; 32] = hasher.finalize().into();
    return ksp_store_lib::RawObservationKey::new(bytes);
}

fn hash_bytes(hasher: &mut sha2::Sha256, value: &[u8]) {
    hasher.update((value.len() as u64).to_be_bytes());
    hasher.update(value);
}

fn base58_digit(byte: u8) -> std::option::Option<u8> {
    return match byte {
        b'1'..=b'9' => std::option::Option::Some(byte - b'1'),
        b'A'..=b'H' => std::option::Option::Some((byte - b'A') + 9),
        b'J'..=b'N' => std::option::Option::Some((byte - b'J') + 17),
        b'P'..=b'Z' => std::option::Option::Some((byte - b'P') + 22),
        b'a'..=b'k' => std::option::Option::Some((byte - b'a') + 33),
        b'm'..=b'z' => std::option::Option::Some((byte - b'm') + 44),
        _ => std::option::Option::None,
    };
}

fn conversion_error(field: &'static str) -> ksp_core_lib::Error {
    return ksp_core_lib::Error::new(crate::ERROR_CODE_BACKFILL_RAW_CONVERSION_INVALID, "invalid deterministic Backfill RAW conversion")
        .with_context("field", field);
}

#[cfg(test)]
#[path = "../unit_tests/conversion.rs"]
mod tests;