lix 0.16.0

//! Repository-scoped synchronization.
//!
//! Lix synchronizes its existing primitives: complete immutable commits,
//! compare-and-swap branch refs, and BLAKE3-addressed binary chunks. Live
//! synchronization has one ordered repository cursor. Current rows and commit
//! topology bootstraps eagerly; historical commit payloads and blobs load on
//! demand. Platform-specific code is limited to tasks, timers, HTTP, and
//! cancellation.

mod blob;
mod bootstrap;
mod commit;
mod contract;
mod http;
mod platform;
mod protocol;
mod repository;
mod runtime;
#[cfg(test)]
mod simulation_tests;
#[cfg(test)]
mod upload_metrics;
mod upload_plan;
mod upload_proof;

use std::sync::Arc;
use std::sync::atomic::{AtomicU8, Ordering};
use std::time::Duration;

use crate::LixError;
use parking_lot::RwLock;

#[cfg(feature = "server-protocol")]
pub(crate) use blob::validate_sync_blob_manifest;
pub(crate) use bootstrap::{
    SyncBootstrapAdmission, inspect_sync_bootstrap_with_adapter, install_sync_bootstrap,
    prepare_sync_bootstrap,
};
pub(crate) use commit::{
    SYNC_CHECKPOINT_SOURCE_SPACE, SYNC_MATERIALIZED_STATE_ALIAS_SPACE,
    stage_delete_materialized_sync_state_alias, stage_delete_sync_checkpoint_source,
    stage_sync_checkpoint_source,
};
pub(crate) use commit::{
    SyncCommit, SyncCommitMemberRef, SyncCommitStateAlias, encode_sync_commit_member,
};
pub(crate) use contract::SyncTransport;
pub(crate) use http::normalize_sync_locator;
pub(crate) use platform::sleep;
pub(crate) use platform::{AuthorityHttp, authority_http};
#[cfg(target_family = "wasm")]
#[doc(hidden)]
pub use platform::{
    BROWSER_TRANSPORT_CONFIG_HEADER, register_browser_sync_transport,
    unregister_browser_sync_transport,
};
pub(crate) use platform::{SyncTransportBounds, SyncTransportFuture};
#[cfg(feature = "server-protocol")]
pub(crate) use protocol::SyncRefUpdate;
pub(crate) use protocol::{
    SyncBlobChunk, SyncBlobManifest, SyncBlobRegistration, SyncBranchHead,
    SyncCheckpointInventoryPage, SyncCommitHeader, SyncEvent, SyncHistoryBoundary,
    SyncHistoryResponse, SyncPushRequest, SyncPushResponse, SyncRepositoryPullResponse,
    SyncSnapshotRow, SyncSnapshotRowPage, encoded_delta_event_len,
};
#[cfg(feature = "server-protocol")]
pub(crate) use repository::admit_sync_authority_storage;
pub(crate) use repository::has_any_sync_replica_state;
pub(crate) use repository::{
    AUTHORITY_STATE_VALUE, SYNC_AUTHORITY_STATE_SPACE, SYNC_REPLICA_STATE_SPACE,
    SYNC_REPOSITORY_EVENT_SPACE, SYNC_SEQUENCE_SPACE, authority_state_key,
    load_pending_sync_export_commit_ids, load_replayable_repository_event_commit_ids,
    replica_state_key, stage_repository_transaction_event, stage_sync_restore_intents,
    validate_repository_transaction_event_transfer,
};
pub(crate) use runtime::{SyncDemand, SyncDemandRetry, SyncRuntime, activate_sync_mode};
pub(crate) use upload_plan::{
    SYNC_UPLOAD_GENERATION_SPACE, stage_invalidate as stage_upload_plan_invalidation,
};
pub(crate) use upload_proof::SYNC_UPLOAD_PROOF_SPACE;

pub(crate) const MAX_SYNC_PULL_RESPONSE_BYTES: usize = 64 * 1024 * 1024;
pub(crate) const MAX_SYNC_HISTORY_PAGE_SIZE: usize = 100;
pub(crate) const MAX_SYNC_BLOB_BATCH_ITEMS: usize = 16;
pub(crate) const MAX_SYNC_REQUEST_ITEMS: usize = 512;
pub(crate) const SYNC_LONG_POLL_TIMEOUT: Duration = Duration::from_secs(30);
// v9 requires canonical checkpoint membership and its global inventory.
// Older writers cannot publish commits without this metadata.
pub(crate) const SYNC_PROTOCOL_VERSION: u32 = 9;
pub(crate) const SYNC_PROTOCOL_VERSION_HEADER: &str = "lix-sync-protocol-version";
pub(crate) const SYNC_PROTOCOL_MISMATCH_CODE: &str = "LIX_SYNC_PROTOCOL_MISMATCH";
pub(crate) const SYNC_REPOSITORY_ID_MISMATCH_CODE: &str = "LIX_SYNC_REPOSITORY_ID_MISMATCH";
pub(crate) const SYNC_IMMUTABLE_OBJECT_MISMATCH_CODE: &str = "LIX_SYNC_IMMUTABLE_OBJECT_MISMATCH";
const MAX_SYNC_REMOTE_ID_BYTES: usize = 4 * 1024;

/// Unforgeable outside the sync module tree. Passing this token makes the
/// durable replica-cache write bypass a compile-time capability, rather than a
/// process-local role flag or a generally callable crate helper.
#[derive(Clone, Copy, Debug)]
pub(crate) struct CertifiedReplicaWriteCapability {
    _private: (),
}

fn certified_replica_write_capability() -> CertifiedReplicaWriteCapability {
    CertifiedReplicaWriteCapability { _private: () }
}

pub(crate) fn sync_server_protocol_mismatch(server_version: Option<u32>) -> LixError {
    let server = server_version
        .map(|version| version.to_string())
        .unwrap_or_else(|| "missing".to_owned());
    LixError::new(
        SYNC_PROTOCOL_MISMATCH_CODE,
        format!(
            "incompatible sync protocol: client version {SYNC_PROTOCOL_VERSION}, server version {server}; upgrade the client and server to compatible Lix versions"
        ),
    )
    .with_details(serde_json::json!({
        "clientSyncProtocolVersion": SYNC_PROTOCOL_VERSION,
        "serverSyncProtocolVersion": server_version,
    }))
}

pub(crate) fn sync_server_protocol_missing_field(field: &str) -> LixError {
    LixError::new(
        SYNC_PROTOCOL_MISMATCH_CODE,
        format!("incompatible sync protocol: server handshake omitted required {field}"),
    )
    .with_details(serde_json::json!({ "missingField": field }))
}

pub(crate) fn sync_repository_id_mismatch(local: &str, authority: &str) -> LixError {
    LixError::new(
        SYNC_REPOSITORY_ID_MISMATCH_CODE,
        "sync authority lixId does not match the local repository",
    )
    .with_details(serde_json::json!({
        "localLixId": local,
        "authorityLixId": authority,
    }))
}

#[cfg(feature = "server-protocol")]
pub(crate) fn sync_client_protocol_mismatch(client_version: Option<u32>) -> LixError {
    let client = client_version
        .map(|version| version.to_string())
        .unwrap_or_else(|| "invalid".to_owned());
    LixError::new(
        SYNC_PROTOCOL_MISMATCH_CODE,
        format!(
            "incompatible sync protocol: client version {client}, server version {SYNC_PROTOCOL_VERSION}; upgrade the client and server to compatible Lix versions"
        ),
    )
    .with_details(serde_json::json!({
        "clientSyncProtocolVersion": client_version,
        "serverSyncProtocolVersion": SYNC_PROTOCOL_VERSION,
    }))
}

pub(crate) fn validate_blake3_id(value: &str, context: &str) -> Result<(), LixError> {
    if value.len() == 64
        && value
            .as_bytes()
            .iter()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(byte))
    {
        return Ok(());
    }
    Err(LixError::new(
        LixError::CODE_INVALID_PARAM,
        format!("{context} must be 64 lowercase hexadecimal characters"),
    ))
}

pub(crate) fn validate_sync_remote_id(remote_id: &str) -> Result<(), LixError> {
    if remote_id.is_empty() || remote_id.len() > MAX_SYNC_REMOTE_ID_BYTES {
        return Err(LixError::new(
            LixError::CODE_INVALID_PARAM,
            format!("sync remoteId must contain 1 to {MAX_SYNC_REMOTE_ID_BYTES} bytes"),
        ));
    }
    Ok(())
}

pub(crate) fn validate_sync_branch_id(branch_id: &str) -> Result<(), LixError> {
    if crate::storage_codec::id_string::uuid_bytes_from_canonical(branch_id).is_none() {
        return Err(LixError::new(
            LixError::CODE_INVALID_PARAM,
            "sync branchId must be a canonical UUID",
        ));
    }
    Ok(())
}

/// Process-wide role shared by every session on one repository engine.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[repr(u8)]
pub(crate) enum SyncRole {
    #[default]
    Disabled,
    Authority,
    Replica,
}

/// The complete process-local sync coordination state.
///
/// There are no query scopes, hydration registries, branch bindings, or file
/// projection caches. SQL always reads the local hot state. This object only
/// identifies the role and wakes long-polls after local commits. The single
/// runtime worker serializes remote repository events.
#[derive(Clone, Debug)]
pub(crate) struct SyncModeState {
    role: Arc<AtomicU8>,
    replica_remote_id: Arc<RwLock<Option<Arc<str>>>>,
    change_watch: tokio::sync::watch::Sender<u64>,
}

impl Default for SyncModeState {
    fn default() -> Self {
        Self {
            role: Arc::new(AtomicU8::new(SyncRole::Disabled as u8)),
            replica_remote_id: Arc::new(RwLock::new(None)),
            change_watch: tokio::sync::watch::channel(0).0,
        }
    }
}

impl SyncModeState {
    pub(crate) fn role(&self) -> SyncRole {
        match self.role.load(Ordering::Acquire) {
            value if value == SyncRole::Disabled as u8 => SyncRole::Disabled,
            value if value == SyncRole::Authority as u8 => SyncRole::Authority,
            value if value == SyncRole::Replica as u8 => SyncRole::Replica,
            _ => unreachable!("sync role stores only enum discriminants"),
        }
    }

    pub(crate) fn set_role(&self, role: SyncRole) {
        self.role.store(role as u8, Ordering::Release);
    }

    pub(crate) fn replica_remote_id(&self) -> Option<Arc<str>> {
        self.replica_remote_id.read().clone()
    }

    pub(crate) fn set_replica_remote_id(&self, remote_id: impl Into<Arc<str>>) {
        *self.replica_remote_id.write() = Some(remote_id.into());
    }

    pub(crate) fn change_watcher(&self) -> tokio::sync::watch::Receiver<u64> {
        self.change_watch.subscribe()
    }

    pub(crate) fn notify_sync_change(&self) {
        self.change_watch
            .send_modify(|version| *version = version.wrapping_add(1));
    }
}