use std::cell::RefCell;
use std::collections::BTreeMap;
use std::rc::Rc;
use super::codec::{decode_nodes, decode_record, encode_nodes, encode_record};
use super::node::BrowserNodeStore;
use super::opfs::BrowserBackend;
use super::source::{branch_get_failure, node_backend_failure};
use super::{
BrowserCommitDurability, BrowserCommitReceipt, BrowserLocalError, BrowserRequest,
BrowserSourceError,
};
use crate::branch::handle::{DEFAULT_SHARD_ID, RefBinding};
use crate::branch::{BranchHandle, BranchKind, BranchRefRecord, BranchRegistry, BranchShardRef};
use crate::store::NodeStore;
use crate::tree::{LeafNode, Node};
pub struct BrowserLocalStore {
shared: Rc<RefCell<StoreState>>,
}
pub struct BrowserWorkHandle {
shared: Rc<RefCell<StoreState>>,
}
pub(super) struct StoreState {
pub(super) name: String,
pub(super) generation: String,
pub(super) backend: Rc<BrowserBackend>,
pub(super) nodes: BrowserNodeStore,
pub(super) records: Vec<BranchRefRecord>,
pub(super) registry: BranchRegistry,
pub(super) work: Option<BranchHandle>,
pub(super) closed: bool,
}
impl BrowserLocalStore {
pub fn open_or_create(store_name: &str) -> BrowserRequest<Self> {
let store_name = store_name.to_owned();
BrowserRequest::new(async move {
let opened = BrowserBackend::open(&store_name).await?;
let mut records = Vec::with_capacity(opened.records.len());
for bytes in opened.records {
let (generation, record) =
decode_record(&bytes).map_err(|source| BrowserLocalError::RecordFailure {
operation: "list/read at open",
source,
})?;
if generation != opened.generation {
return Err(BrowserLocalError::StoreGenerationLost {
component: "record generation",
});
}
records.push(record);
}
let node_map = opened.nodes.map_or_else(
|| Ok(BTreeMap::default()),
|bytes| {
decode_nodes(&bytes).map_err(|source| BrowserLocalError::NodeFailure {
operation: "decode nodes",
source,
})
},
)?;
if !records.is_empty() && node_map.is_empty() {
return Err(BrowserLocalError::StoreGenerationLost { component: "nodes" });
}
let nodes = BrowserNodeStore::from_serialized(node_map)
.map_err(|error| node_backend_failure("validate nodes", error))?;
Ok(Self {
shared: Rc::new(RefCell::new(StoreState {
name: store_name,
generation: opened.generation,
backend: Rc::new(opened.backend),
nodes,
records,
registry: BranchRegistry::new(),
work: None,
closed: false,
})),
})
})
}
pub fn local_work(&self) -> BrowserRequest<BrowserWorkHandle> {
let shared = Rc::clone(&self.shared);
BrowserRequest::new(async move {
let (backend, generation, create) = {
let mut state = shared
.try_borrow_mut()
.map_err(|_| BrowserLocalError::WorkerClosed)?;
ensure_open(&state)?;
match state.records.as_slice() {
[] => {
let empty = Node::Leaf(LeafNode::new(Vec::new()).map_err(|error| {
BrowserLocalError::NodeFailure {
operation: "create empty root",
source: BrowserSourceError::Node { error },
}
})?);
let root = state
.nodes
.put(&empty)
.map_err(|error| node_backend_failure("create empty root", error))?;
let created = browser_timestamp();
let record = BranchRefRecord {
name: state.name.clone(),
created,
kind: BranchKind::Work,
namespace_lineage: None,
seq: 0,
timestamp: created,
shards: vec![BranchShardRef {
shard_id: DEFAULT_SHARD_ID,
fork_anchor: root,
head: root,
}],
parents: Vec::new(),
};
let node_bytes = encode_nodes(state.nodes.serialized());
(
Rc::clone(&state.backend),
state.generation.clone(),
Some((record, node_bytes)),
)
}
[_] => (Rc::clone(&state.backend), state.generation.clone(), None),
records => {
return Err(BrowserLocalError::NonCanonicalBranchSet {
records: records.len(),
});
}
}
};
if let Some((record, node_bytes)) = create {
backend.write_nodes(&node_bytes).await?;
let record_bytes = encode_record(&record, &generation);
if backend
.create_exclusive(&record.name, &record_bytes)
.await?
.is_some()
{
return Err(BrowserLocalError::NonCanonicalBranchSet { records: 1 });
}
shared
.try_borrow_mut()
.map_err(|_| BrowserLocalError::WorkerClosed)?
.records
.push(record);
}
{
let mut state = shared
.try_borrow_mut()
.map_err(|_| BrowserLocalError::WorkerClosed)?;
validate_local_record(&state)?;
if state.work.is_none() {
state.work = Some(open_handle(&state)?);
}
}
Ok(BrowserWorkHandle { shared })
})
}
pub fn close(self) -> BrowserRequest<()> {
BrowserRequest::new(async move {
let backend = {
let mut state = self
.shared
.try_borrow_mut()
.map_err(|_| BrowserLocalError::WorkerClosed)?;
ensure_open(&state)?;
state.closed = true;
state.work = None;
Rc::clone(&state.backend)
};
backend.close().await
})
}
}
impl BrowserWorkHandle {
pub fn put(&self, key: &[u8], value: &[u8]) -> BrowserRequest<()> {
let shared = Rc::clone(&self.shared);
let key = key.to_vec();
let value = value.to_vec();
BrowserRequest::new(async move {
let state = shared
.try_borrow()
.map_err(|_| BrowserLocalError::WorkerClosed)?;
ensure_open(&state)?;
state
.work
.as_ref()
.ok_or(BrowserLocalError::WorkerClosed)?
.put(DEFAULT_SHARD_ID, key, value)
.map_err(|error| BrowserLocalError::NodeFailure {
operation: "branch put",
source: BrowserSourceError::Branch { error },
})
})
}
pub fn get(&self, key: &[u8]) -> BrowserRequest<Option<Vec<u8>>> {
let shared = Rc::clone(&self.shared);
let key = key.to_vec();
BrowserRequest::new(async move {
let state = shared
.try_borrow()
.map_err(|_| BrowserLocalError::WorkerClosed)?;
ensure_open(&state)?;
state
.work
.as_ref()
.ok_or(BrowserLocalError::WorkerClosed)?
.get_with_source(DEFAULT_SHARD_ID, &state.nodes, &key)
.map_err(|error| branch_get_failure("branch get", error))
})
}
pub fn commit_branch(
&self,
durability: BrowserCommitDurability,
) -> BrowserRequest<BrowserCommitReceipt> {
let shared = Rc::clone(&self.shared);
BrowserRequest::new(async move {
match durability {
BrowserCommitDurability::Durable => super::commit::commit_durable(shared).await,
}
})
}
}
fn validate_local_record(state: &StoreState) -> Result<(), BrowserLocalError> {
let [record] = state.records.as_slice() else {
return Err(BrowserLocalError::NonCanonicalBranchSet {
records: state.records.len(),
});
};
if record.kind != BranchKind::Work {
return Err(BrowserLocalError::WrongBranchKind {
found: record.kind.to_string(),
});
}
if record.name != state.name
|| record.namespace_lineage.is_some()
|| record.shards.len() != 1
|| record.shards[0].shard_id != DEFAULT_SHARD_ID
{
return Err(BrowserLocalError::NonCanonicalBranchSet { records: 1 });
}
Ok(())
}
fn open_handle(state: &StoreState) -> Result<BranchHandle, BrowserLocalError> {
let record = &state.records[0];
let handle = BranchHandle::from_anchor_head_pairs(
record
.shards
.iter()
.map(|shard| (shard.shard_id, shard.fork_anchor, shard.head)),
)
.map_err(|error| BrowserLocalError::RecordFailure {
operation: "open Work record",
source: BrowserSourceError::Branch { error },
})?;
let roots = record
.shards
.iter()
.flat_map(|shard| [shard.fork_anchor, shard.head]);
Ok(handle
.with_registry_guard(state.registry.register_roots(roots))
.with_binding(RefBinding {
name: record.name.clone(),
created: record.created,
seq: record.seq,
}))
}
pub(super) const fn ensure_open(state: &StoreState) -> Result<(), BrowserLocalError> {
if state.closed {
Err(BrowserLocalError::WorkerClosed)
} else {
Ok(())
}
}
pub(super) fn browser_timestamp() -> u64 {
let nanos = js_sys::Date::now() * 1_000_000.0;
if nanos.is_finite() && nanos > 0.0 {
format!("{nanos:.0}").parse().unwrap_or(u64::MAX)
} else {
0
}
}