haematite 0.6.1

use std::cell::RefCell;
use std::rc::Rc;
use std::sync::Mutex;

use super::codec::{encode_nodes, encode_record};
use super::core::{StoreState, ensure_open};
use super::opfs::BrowserBackend;
use super::source::tree_failure;
use super::{
    BrowserBranchVersion, BrowserCommitReceipt, BrowserDurabilityGrade, BrowserLocalError,
    BrowserSourceError,
};
use crate::branch::handle::RefBinding;
use crate::branch::registry::advance_in;
use crate::branch::{BranchRefRecord, current_timestamp};
use crate::tree::Hash;
use crate::tree::mutate::batch_mutate_owned_with_source;
use crate::wal::{Mutation, WalBuffer};

struct PreparedCommit {
    backend: Rc<BrowserBackend>,
    generation: String,
    prior: BranchRefRecord,
    replacement: BranchRefRecord,
    materialised: Vec<(usize, Hash, Hash)>,
    node_bytes: Vec<u8>,
}

enum CommitPreparation {
    Noop {
        backend: Rc<BrowserBackend>,
        record: BranchRefRecord,
        heads: Vec<(usize, Hash)>,
    },
    Advance(PreparedCommit),
}

pub(super) async fn commit_durable(
    shared: Rc<RefCell<StoreState>>,
) -> Result<BrowserCommitReceipt, BrowserLocalError> {
    let prepared = prepare_commit(&shared)?;
    let CommitPreparation::Advance(prepared) = prepared else {
        let CommitPreparation::Noop {
            backend,
            record,
            heads,
        } = prepared
        else {
            unreachable!()
        };
        backend
            .cas_check(&record.name, record.created, record.seq)
            .await?;
        return Ok(receipt(&record, heads, false));
    };
    prepared.backend.write_nodes(&prepared.node_bytes).await?;
    let record_bytes = encode_record(&prepared.replacement, &prepared.generation);
    prepared
        .backend
        .cas_replace_install(
            &prepared.prior.name,
            prepared.prior.created,
            prepared.prior.seq,
            &record_bytes,
        )
        .await?;
    let heads = publish_visibility(&shared, &prepared.replacement, &prepared.materialised)?;
    // Commit-event emission is the named absent leg-4 seam. There is no
    // publisher, placeholder, callback, timer, or synthetic event here.
    Ok(receipt(&prepared.replacement, heads, true))
}

fn prepare_commit(
    shared: &Rc<RefCell<StoreState>>,
) -> Result<CommitPreparation, BrowserLocalError> {
    let mut state = shared
        .try_borrow_mut()
        .map_err(|_| BrowserLocalError::WorkerClosed)?;
    ensure_open(&state)?;
    let branch = state.work.clone().ok_or(BrowserLocalError::WorkerClosed)?;
    let states =
        branch
            .lock_states_ascending()
            .map_err(|error| BrowserLocalError::NodeFailure {
                operation: "commit locks",
                source: BrowserSourceError::Branch { error },
            })?;
    let prior = state.records[0].clone();
    if states.iter().all(|(_, shard)| shard.buffer.is_empty()) {
        let heads = states
            .iter()
            .map(|(id, shard)| (*id, shard.current_root))
            .collect();
        return Ok(CommitPreparation::Noop {
            backend: Rc::clone(&state.backend),
            record: prior,
            heads,
        });
    }
    let mut materialised = Vec::with_capacity(states.len());
    for (shard_id, shard) in &states {
        let root = batch_mutate_owned_with_source(
            &mut state.nodes,
            shard.current_root,
            buffered_batch(&shard.buffer),
        )
        .map_err(|error| tree_failure("materialize nodes", error))?;
        materialised.push((*shard_id, shard.current_root, root));
    }
    let mut replacement = prior.clone();
    replacement.seq = prior.seq.saturating_add(1);
    replacement.timestamp = current_timestamp();
    replacement.parents = prior.shards.iter().map(|shard| shard.head).collect();
    for (shard_id, _, root) in &materialised {
        if let Some(shard) = replacement
            .shards
            .iter_mut()
            .find(|shard| shard.shard_id == *shard_id)
        {
            shard.head = *root;
        }
    }
    Ok(CommitPreparation::Advance(PreparedCommit {
        backend: Rc::clone(&state.backend),
        generation: state.generation.clone(),
        prior,
        replacement,
        materialised,
        node_bytes: encode_nodes(state.nodes.serialized()),
    }))
}

fn publish_visibility(
    shared: &Rc<RefCell<StoreState>>,
    replacement: &BranchRefRecord,
    materialised: &[(usize, Hash, Hash)],
) -> Result<Vec<(usize, Hash)>, BrowserLocalError> {
    let mut state = shared
        .try_borrow_mut()
        .map_err(|_| BrowserLocalError::WorkerClosed)?;
    let branch = state.work.clone().ok_or(BrowserLocalError::WorkerClosed)?;
    let mut states =
        branch
            .lock_states_ascending()
            .map_err(|error| BrowserLocalError::NodeFailure {
                operation: "publish visibility",
                source: BrowserSourceError::Branch { error },
            })?;
    let mut binding = branch.binding.as_deref().map(lock_binding);
    let guard = branch
        .registry_guard()
        .ok_or(BrowserLocalError::WorkerClosed)?;
    let mut counts = state.registry.lock_counts();
    for (shard_id, old, new) in materialised {
        if old != new {
            advance_in(&mut counts, *old, *new);
            guard.replace(*old, *new);
        }
        let target_state = states
            .iter_mut()
            .find(|(id, _)| id == shard_id)
            .ok_or(BrowserLocalError::WorkerClosed)?;
        target_state.1.current_root = *new;
        target_state.1.buffer = WalBuffer::new();
    }
    if let Some(binding) = binding.as_deref_mut() {
        binding.seq = replacement.seq;
    }
    let heads = states
        .iter()
        .map(|(id, shard)| (*id, shard.current_root))
        .collect();
    drop(counts);
    drop(binding);
    state.records[0] = replacement.clone();
    Ok(heads)
}

fn receipt(
    record: &BranchRefRecord,
    heads: Vec<(usize, Hash)>,
    advanced: bool,
) -> BrowserCommitReceipt {
    BrowserCommitReceipt {
        grade: BrowserDurabilityGrade::BrowserFlushed,
        version: BrowserBranchVersion {
            name: record.name.clone(),
            created: record.created,
            seq: record.seq,
        },
        heads,
        advanced,
    }
}

fn buffered_batch(buffer: &WalBuffer) -> Vec<(Vec<u8>, Option<Vec<u8>>)> {
    buffer
        .iter()
        .map(|mutation| match mutation {
            Mutation::Put { key, value } => (key.clone(), Some(value.clone())),
            Mutation::Delete { key } => (key.clone(), None),
        })
        .collect()
}

fn lock_binding(binding: &Mutex<RefBinding>) -> std::sync::MutexGuard<'_, RefBinding> {
    match binding.lock() {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    }
}