use std::{
collections::{HashMap, HashSet},
sync::Mutex,
};
use anyhow::{anyhow, bail, Context, Result};
use futures::StreamExt;
use iroh_docs::{
actor::{OpenOpts, SyncHandle},
api::protocol::{AddrInfoOptions, ShareMode},
engine::{DefaultAuthorStorage, Engine},
protocol::Docs,
store::{Query, Store},
Author, AuthorId, Capability, ContentStatus, DocTicket, NamespaceId, SignedEntry,
};
use js_sys::{Array, Function, Promise, Reflect, Uint8Array};
use serde::{Deserialize, Serialize};
use tokio::io::{AsyncReadExt, AsyncSeekExt};
use wasm_bindgen::{JsCast, JsValue};
use wasm_bindgen_futures::JsFuture;
use crate::wasm_indexeddb_blob_store::IndexedDbBlobStore;
const EVENT_CAPACITY: usize = 256;
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PersistedIdentity {
default_author: Vec<u8>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PersistedAuthor {
author_id: String,
author: Vec<u8>,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PersistedCapability {
namespace_id: String,
capability: Vec<u8>,
capability_kind: String,
generation: u64,
share_revision: u64,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PersistedEntry {
namespace_id: String,
signed_entry: Vec<u8>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct ReplicaSnapshot {
authors: Vec<PersistedAuthor>,
namespaces: Vec<PersistedCapability>,
entries: Vec<PersistedEntry>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct PersistentBlobManifest {
pub(crate) content_hash: String,
pub(crate) size: u64,
pub(crate) chunk_count: u32,
pub(crate) chunk_bytes: u32,
pub(crate) bao_ready: bool,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct PersistentBlobResumeState {
pub(crate) expected_size: u64,
pub(crate) chunk_bytes: u32,
pub(crate) received_chunks: Vec<u32>,
pub(crate) complete: bool,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct ReplicaEntryRecord {
record_id: String,
namespace_id: String,
author_id: String,
key_hex: String,
timestamp: u64,
content_hash: String,
content_length: u64,
#[serde(with = "serde_bytes")]
signed_entry: Vec<u8>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct ReplicaOutbox {
operation_id: String,
#[serde(with = "serde_bytes")]
payload: Vec<u8>,
created_at: u64,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct ReplicaMutation {
entry: ReplicaEntryRecord,
#[serde(skip_serializing_if = "Option::is_none")]
outbox: Option<ReplicaOutbox>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct ReplicaCapabilityRecord {
namespace_id: String,
#[serde(with = "serde_bytes")]
capability: Vec<u8>,
capability_kind: &'static str,
generation: u64,
share_revision: u64,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WasmNamespaceDescriptor {
namespace_id: String,
capability_kind: &'static str,
#[serde(with = "serde_bytes")]
capability: Vec<u8>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WasmDocEntry {
namespace_id: String,
author_id: String,
#[serde(with = "serde_bytes")]
key: Vec<u8>,
timestamp: u64,
content_hash: String,
content_length: u64,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WasmMutationReceipt {
content_hash: String,
operation_id: String,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WasmDeleteReceipt {
removed: u32,
operation_id: String,
}
#[derive(Clone)]
pub(crate) struct JsReplicaStore {
inner: JsValue,
}
impl std::fmt::Debug for JsReplicaStore {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("JsReplicaStore")
.finish_non_exhaustive()
}
}
impl JsReplicaStore {
pub(crate) fn new(inner: JsValue) -> Result<Self> {
if inner.is_null() || inner.is_undefined() {
bail!("iroh-docs IndexedDB adapter is required");
}
Ok(Self { inner })
}
async fn call(&self, method: &str, args: &[JsValue]) -> Result<JsValue> {
let function = Reflect::get(&self.inner, &JsValue::from_str(method))
.map_err(js_error)?
.dyn_into::<Function>()
.map_err(|_| anyhow!("IndexedDB adapter method {method} is missing"))?;
let call_args = Array::new();
for arg in args {
call_args.push(arg);
}
let result = function.apply(&self.inner, &call_args).map_err(js_error)?;
JsFuture::from(Promise::resolve(&result))
.await
.map_err(js_error)
.with_context(|| format!("IndexedDB adapter method {method} failed"))
}
async fn identity(&self) -> Result<PersistedIdentity> {
let value = self.call("loadIdentity", &[]).await?;
if value.is_null() || value.is_undefined() {
bail!("persistent iroh-docs identity is not initialized");
}
serde_wasm_bindgen::from_value(value).context("decoding persistent iroh-docs identity")
}
async fn snapshot(&self) -> Result<ReplicaSnapshot> {
let value = self.call("exportSnapshot", &[]).await?;
serde_wasm_bindgen::from_value(value).context("decoding persistent iroh-docs snapshot")
}
async fn import_author(&self, author: &Author) -> Result<()> {
let bytes = Uint8Array::from(author.to_bytes().as_slice());
self.call(
"importAuthor",
&[JsValue::from_str(&author.id().to_string()), bytes.into()],
)
.await?;
Ok(())
}
async fn import_namespace(
&self,
capability: &Capability,
generation: u64,
share_revision: u64,
) -> Result<()> {
let (kind, bytes) = capability.raw();
let value = serde_wasm_bindgen::to_value(&ReplicaCapabilityRecord {
namespace_id: capability.id().to_string(),
capability: bytes.to_vec(),
capability_kind: if kind == 1 { "write" } else { "read" },
generation,
share_revision,
})
.context("encoding iroh-docs capability")?;
self.call("importNamespace", &[value]).await?;
Ok(())
}
async fn delete_namespace(&self, namespace: NamespaceId) -> Result<()> {
self.call(
"deleteNamespace",
&[JsValue::from_str(&namespace.to_string())],
)
.await?;
Ok(())
}
async fn persist_entry(&self, entry: &SignedEntry, durable_outbox: bool) -> Result<String> {
let signed_entry = postcard::to_stdvec(entry).context("encoding signed iroh-docs entry")?;
let record_id = hex::encode(blake3::hash(&signed_entry).as_bytes());
let mutation = ReplicaMutation {
entry: ReplicaEntryRecord {
record_id: record_id.clone(),
namespace_id: entry.entry().namespace().to_string(),
author_id: entry.author_bytes().to_string(),
key_hex: hex::encode(entry.key()),
timestamp: entry.timestamp(),
content_hash: entry.content_hash().to_string(),
content_length: entry.content_len(),
signed_entry: signed_entry.clone(),
},
outbox: durable_outbox.then(|| ReplicaOutbox {
operation_id: record_id.clone(),
payload: signed_entry,
created_at: (js_sys::Date::now().max(0.0)) as u64,
}),
};
let value =
serde_wasm_bindgen::to_value(&mutation).context("encoding iroh-docs mutation")?;
self.call("applyEntryTransaction", &[value]).await?;
Ok(record_id)
}
pub(crate) async fn flush(&self) -> Result<()> {
self.call("flush", &[]).await?;
Ok(())
}
async fn acknowledge_outbox(&self, operation_id: &str) -> Result<()> {
self.call("acknowledgeOutbox", &[JsValue::from_str(operation_id)])
.await?;
Ok(())
}
pub(crate) async fn blob_manifests(&self) -> Result<Vec<PersistentBlobManifest>> {
let value = self.call("listBlobManifests", &[]).await?;
serde_wasm_bindgen::from_value(value).context("decoding persistent blob manifests")
}
pub(crate) async fn blob_resume_state(
&self,
content_hash: &str,
) -> Result<Option<PersistentBlobResumeState>> {
let value = self
.call("getBlobResumeState", &[JsValue::from_str(content_hash)])
.await?;
if value.is_null() || value.is_undefined() {
return Ok(None);
}
serde_wasm_bindgen::from_value(value)
.context("decoding persistent blob resume state")
.map(Some)
}
pub(crate) async fn read_blob_range(
&self,
content_hash: &str,
offset: u64,
length: usize,
) -> Result<Option<Vec<u8>>> {
ensure_js_safe_integer(offset, "persistent blob range offset")?;
let value = self
.call(
"readBlobRange",
&[
JsValue::from_str(content_hash),
JsValue::from_f64(offset as f64),
JsValue::from_f64(length as f64),
],
)
.await?;
if value.is_null() || value.is_undefined() {
return Ok(None);
}
Ok(Some(Uint8Array::new(&value).to_vec()))
}
pub(crate) async fn write_blob_range(
&self,
content_hash: &str,
size: u64,
chunk_bytes: usize,
offset: u64,
bytes: &[u8],
) -> Result<()> {
ensure_js_safe_integer(size, "persistent blob size")?;
ensure_js_safe_integer(offset, "persistent blob range offset")?;
self.call(
"writeBlobRange",
&[
JsValue::from_str(content_hash),
JsValue::from_f64(size as f64),
JsValue::from_f64(chunk_bytes as f64),
JsValue::from_f64(offset as f64),
Uint8Array::from(bytes).into(),
],
)
.await?;
Ok(())
}
pub(crate) async fn finalize_blob_if_complete(&self, content_hash: &str) -> Result<bool> {
self.call("finalizeBlobIfComplete", &[JsValue::from_str(content_hash)])
.await?
.as_bool()
.ok_or_else(|| anyhow!("IndexedDB adapter returned an invalid blob completion result"))
}
pub(crate) async fn read_blob_outboard_node(
&self,
content_hash: &str,
node_id: u64,
) -> Result<Option<[u8; 64]>> {
let value = self
.call(
"readBlobOutboardNode",
&[
JsValue::from_str(content_hash),
JsValue::from_str(&node_id.to_string()),
],
)
.await?;
if value.is_null() || value.is_undefined() {
return Ok(None);
}
let bytes = Uint8Array::new(&value).to_vec();
Ok(Some(bytes.try_into().map_err(|_| {
anyhow!("persistent Bao node must contain exactly 64 bytes")
})?))
}
pub(crate) async fn write_blob_outboard_node(
&self,
content_hash: &str,
node_id: u64,
pair: &[u8; 64],
) -> Result<()> {
self.call(
"writeBlobOutboardNode",
&[
JsValue::from_str(content_hash),
JsValue::from_str(&node_id.to_string()),
Uint8Array::from(pair.as_slice()).into(),
],
)
.await?;
Ok(())
}
pub(crate) async fn mark_blob_bao_ready(&self, content_hash: &str) -> Result<()> {
self.call("markBlobBaoReady", &[JsValue::from_str(content_hash)])
.await?;
Ok(())
}
pub(crate) async fn write_blob_staging_chunk(
&self,
session_id: &str,
chunk_index: u32,
bytes: &[u8],
) -> Result<()> {
self.call(
"writeBlobStagingChunk",
&[
JsValue::from_str(session_id),
JsValue::from_f64(f64::from(chunk_index)),
Uint8Array::from(bytes).into(),
],
)
.await?;
Ok(())
}
pub(crate) async fn finalize_blob_staging(
&self,
session_id: &str,
content_hash: &str,
size: u64,
chunk_bytes: usize,
) -> Result<()> {
ensure_js_safe_integer(size, "persistent staged blob size")?;
self.call(
"finalizeBlobStaging",
&[
JsValue::from_str(session_id),
JsValue::from_str(content_hash),
JsValue::from_f64(size as f64),
JsValue::from_f64(chunk_bytes as f64),
],
)
.await?;
Ok(())
}
pub(crate) async fn abort_blob_staging(&self, session_id: &str) -> Result<()> {
self.call("abortBlobStaging", &[JsValue::from_str(session_id)])
.await?;
Ok(())
}
pub(crate) async fn delete_blob(&self, content_hash: &str, force: bool) -> Result<bool> {
self.call(
"deleteBlob",
&[JsValue::from_str(content_hash), JsValue::from_bool(force)],
)
.await?
.as_bool()
.ok_or_else(|| anyhow!("IndexedDB adapter returned an invalid blob deletion result"))
}
pub(crate) async fn begin_blob_import(
&self,
content_hash: &str,
size: u64,
chunk_bytes: usize,
) -> Result<()> {
ensure_js_safe_integer(size, "persistent blob size")?;
self.call(
"beginBlobImport",
&[
JsValue::from_str(content_hash),
JsValue::from_f64(size as f64),
JsValue::from_f64(chunk_bytes as f64),
],
)
.await?;
Ok(())
}
pub(crate) async fn write_blob_chunk(
&self,
content_hash: &str,
size: u64,
chunk_bytes: usize,
chunk_index: u32,
bytes: &[u8],
) -> Result<()> {
ensure_js_safe_integer(size, "persistent blob size")?;
self.call(
"writeBlobChunk",
&[
JsValue::from_str(content_hash),
JsValue::from_f64(size as f64),
JsValue::from_f64(chunk_bytes as f64),
JsValue::from_f64(f64::from(chunk_index)),
Uint8Array::from(bytes).into(),
],
)
.await?;
Ok(())
}
pub(crate) async fn finalize_blob_import(
&self,
content_hash: &str,
size: u64,
chunk_bytes: usize,
) -> Result<()> {
ensure_js_safe_integer(size, "persistent blob size")?;
self.call(
"finalizeBlobImport",
&[
JsValue::from_str(content_hash),
JsValue::from_f64(size as f64),
JsValue::from_f64(chunk_bytes as f64),
],
)
.await?;
Ok(())
}
}
pub(crate) struct WasmPersistentDocsActor {
sync: SyncHandle,
default_author: AuthorId,
store: JsReplicaStore,
capabilities: HashMap<NamespaceId, RuntimeCapabilityFence>,
docs: Docs,
blobs: iroh_blobs::BlobsProtocol,
downloader: iroh_blobs::api::downloader::Downloader,
blob_retry_after_ms: Mutex<HashMap<iroh_blobs::Hash, u64>>,
gossip: iroh_gossip::net::Gossip,
}
#[derive(Debug, Clone, Copy)]
struct RuntimeCapabilityFence {
writable: bool,
generation: u64,
share_revision: u64,
}
impl std::fmt::Debug for WasmPersistentDocsActor {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("WasmPersistentDocsActor")
.field("namespaces", &self.capabilities.len())
.finish_non_exhaustive()
}
}
impl WasmPersistentDocsActor {
pub(crate) async fn hydrate(store: JsReplicaStore, endpoint: iroh::Endpoint) -> Result<Self> {
let identity = store.identity().await?;
if identity.default_author.len() != 32 {
bail!("persistent default iroh-docs author must be exactly 32 bytes");
}
let snapshot = store.snapshot().await?;
let mut redb = Store::memory();
let mut default_author_bytes = [0u8; 32];
default_author_bytes.copy_from_slice(&identity.default_author);
let persistent_default_author = Author::from_bytes(&default_author_bytes);
let persistent_default_author_id = persistent_default_author.id();
redb.import_author(persistent_default_author)?;
for persisted in snapshot.authors {
let bytes: [u8; 32] = persisted
.author
.try_into()
.map_err(|_| anyhow!("persisted author {} is not 32 bytes", persisted.author_id))?;
let author = Author::from_bytes(&bytes);
if author.id().to_string() != persisted.author_id {
bail!("persisted author id does not match its secret");
}
redb.import_author(author)?;
}
let mut capabilities = Vec::new();
for persisted in snapshot.namespaces {
let bytes: [u8; 32] = persisted.capability.clone().try_into().map_err(|_| {
anyhow!(
"persisted namespace {} capability is not 32 bytes",
persisted.namespace_id
)
})?;
let kind = match persisted.capability_kind.as_str() {
"write" => 1,
"read" => 2,
other => bail!("unsupported persisted capability kind {other}"),
};
let capability = Capability::from_raw(kind, &bytes)?;
if capability.id().to_string() != persisted.namespace_id {
bail!("persisted namespace id does not match its capability");
}
redb.import_namespace(capability.clone())?;
capabilities.push((capability, persisted.generation, persisted.share_revision));
}
let blobs_store = IndexedDbBlobStore::open(store.clone()).await?;
let blobs_api = (*blobs_store).clone();
let blobs = iroh_blobs::BlobsProtocol::new(&blobs_api, None);
let gossip = iroh_gossip::net::Gossip::builder().spawn(endpoint.clone());
let downloader = blobs_api.downloader(&endpoint);
let engine = Engine::spawn(
endpoint,
gossip.clone(),
redb,
blobs_api,
downloader.clone(),
DefaultAuthorStorage::Mem,
None,
)
.await?;
let generated_default_author = engine.default_author.get();
engine
.default_author
.set(persistent_default_author_id, &engine.sync)
.await?;
if generated_default_author != persistent_default_author_id {
engine.sync.delete_author(generated_default_author).await?;
}
let sync = engine.sync.clone();
let docs = Docs::new(engine);
let mut runtime_capabilities = HashMap::new();
for (capability, generation, share_revision) in &capabilities {
sync.open(capability.id(), OpenOpts::default().sync())
.await?;
runtime_capabilities.insert(
capability.id(),
RuntimeCapabilityFence {
writable: capability_kind(capability) == "write",
generation: *generation,
share_revision: *share_revision,
},
);
}
for persisted in snapshot.entries {
let namespace: NamespaceId = persisted
.namespace_id
.parse()
.context("parsing persisted iroh-docs namespace")?;
if !runtime_capabilities.contains_key(&namespace) {
bail!("persisted entry references an unknown namespace");
}
let entry: SignedEntry = postcard::from_bytes(&persisted.signed_entry)
.context("decoding persisted signed iroh-docs entry")?;
sync.insert_remote(namespace, entry, [0u8; 32], ContentStatus::Complete)
.await
.or_else(ignore_newer_entry)?;
}
let actor = Self {
sync,
default_author: persistent_default_author_id,
store,
capabilities: runtime_capabilities,
docs,
blobs,
downloader,
blob_retry_after_ms: Mutex::new(HashMap::new()),
gossip,
};
for namespace in actor.capabilities.keys().copied() {
actor.subscribe_persistence(namespace).await?;
}
Ok(actor)
}
pub(crate) fn docs_protocol(&self) -> Docs {
self.docs.clone()
}
pub(crate) fn blobs_protocol(&self) -> iroh_blobs::BlobsProtocol {
self.blobs.clone()
}
pub(crate) fn gossip_protocol(&self) -> iroh_gossip::net::Gossip {
self.gossip.clone()
}
async fn subscribe_persistence(&self, namespace: NamespaceId) -> Result<()> {
let (sender, receiver) = async_channel::bounded(EVENT_CAPACITY);
self.sync.subscribe(namespace, sender).await?;
let store = self.store.clone();
n0_future::task::spawn(async move {
while let Ok(event) = receiver.recv().await {
let (entry, local) = match event {
iroh_docs::Event::LocalInsert { entry, .. } => (entry, true),
iroh_docs::Event::RemoteInsert { entry, .. } => (entry, false),
};
if let Err(error) = store.persist_entry(&entry, local).await {
web_sys::console::error_1(&JsValue::from_str(&format!(
"[OpenRTC][iroh-docs][persistence] {error:#}"
)));
break;
}
}
});
Ok(())
}
pub(crate) async fn import_author(&self, bytes: Vec<u8>) -> Result<String> {
let bytes: [u8; 32] = bytes
.try_into()
.map_err(|_| anyhow!("iroh-docs author must be exactly 32 bytes"))?;
let author = Author::from_bytes(&bytes);
let id = self.sync.import_author(author.clone()).await?;
self.store.import_author(&author).await?;
Ok(id.to_string())
}
pub(crate) async fn import_namespace(
&mut self,
kind: &str,
bytes: Vec<u8>,
generation: u64,
share_revision: u64,
) -> Result<String> {
let bytes: [u8; 32] = bytes
.try_into()
.map_err(|_| anyhow!("iroh-docs capability must be exactly 32 bytes"))?;
let kind = match kind {
"write" => 1,
"read" => 2,
other => bail!("unsupported iroh-docs capability kind {other}"),
};
let capability = Capability::from_raw(kind, &bytes)?;
self.require_current_or_new_capability(
capability.id(),
kind == 1,
generation,
share_revision,
)?;
let namespace = self.sync.import_namespace(capability.clone()).await?;
let is_new = !self.capabilities.contains_key(&namespace);
if is_new {
self.sync
.open(namespace, OpenOpts::default().sync())
.await?;
self.subscribe_persistence(namespace).await?;
}
self.store
.import_namespace(&capability, generation, share_revision)
.await?;
self.capabilities.insert(
namespace,
RuntimeCapabilityFence {
writable: kind == 1,
generation,
share_revision,
},
);
Ok(namespace.to_string())
}
pub(crate) async fn create_namespace(
&mut self,
generation: u64,
share_revision: u64,
) -> Result<WasmNamespaceDescriptor> {
let doc = self.docs.api().create().await?;
let namespace = doc.id();
let secret = self.sync.export_secret_key(namespace).await?;
let capability = Capability::Write(secret);
if !self.capabilities.contains_key(&namespace) {
self.subscribe_persistence(namespace).await?;
}
self.store
.import_namespace(&capability, generation, share_revision)
.await?;
self.capabilities.insert(
namespace,
RuntimeCapabilityFence {
writable: true,
generation,
share_revision,
},
);
Ok(namespace_descriptor(&capability))
}
pub(crate) async fn import_ticket(
&mut self,
ticket: &str,
generation: u64,
share_revision: u64,
) -> Result<String> {
let ticket: DocTicket = ticket.parse().context("parsing iroh-docs ticket")?;
let capability = ticket.capability.clone();
let namespace = capability.id();
let writable = capability_kind(&capability) == "write";
self.require_current_or_new_capability(namespace, writable, generation, share_revision)?;
self.docs.api().import(ticket).await?;
if !self.capabilities.contains_key(&namespace) {
self.subscribe_persistence(namespace).await?;
}
self.store
.import_namespace(&capability, generation, share_revision)
.await?;
self.capabilities.insert(
namespace,
RuntimeCapabilityFence {
writable,
generation,
share_revision,
},
);
Ok(namespace.to_string())
}
pub(crate) async fn share(&self, namespace: &str, writable: bool) -> Result<String> {
let namespace = parse_namespace(namespace)?;
let fence = require_namespace(&self.capabilities, namespace)?;
if writable && !fence.writable {
bail!("iroh-docs namespace is read-only at the current capability generation");
}
let doc = self
.docs
.api()
.open(namespace)
.await?
.ok_or_else(|| anyhow!("iroh-docs namespace is unavailable"))?;
let ticket = doc
.share(
if writable {
ShareMode::Write
} else {
ShareMode::Read
},
AddrInfoOptions::RelayAndAddresses,
)
.await?;
Ok(ticket.to_string())
}
pub(crate) async fn remove_namespace(
&mut self,
namespace: &str,
generation: u64,
share_revision: u64,
) -> Result<()> {
let namespace = parse_namespace(namespace)?;
let current = require_namespace(&self.capabilities, namespace)?;
if (generation, share_revision) < (current.generation, current.share_revision) {
bail!("stale iroh-docs capability removal");
}
self.store.delete_namespace(namespace).await?;
let drop_result = self.docs.api().drop_doc(namespace).await;
self.capabilities.remove(&namespace);
drop_result
}
pub(crate) async fn set_bytes(
&self,
namespace: &str,
key: Vec<u8>,
value: Vec<u8>,
) -> Result<WasmMutationReceipt> {
let namespace = parse_namespace(namespace)?;
require_writable_namespace(&self.capabilities, namespace)?;
let doc = self
.docs
.api()
.open(namespace)
.await?
.ok_or_else(|| anyhow!("iroh-docs namespace is unavailable"))?;
let hash = doc
.set_bytes(self.default_author, key.clone(), value)
.await?;
let entry = self
.sync
.get_exact(namespace, self.default_author, key.into(), true)
.await?
.ok_or_else(|| anyhow!("local iroh-docs entry was not committed"))?;
let operation_id = self.store.persist_entry(&entry, true).await?;
Ok(WasmMutationReceipt {
content_hash: hash.to_string(),
operation_id,
})
}
pub(crate) async fn set_hash(
&self,
namespace: &str,
key: Vec<u8>,
hash: &str,
size: u64,
) -> Result<String> {
let namespace = parse_namespace(namespace)?;
require_writable_namespace(&self.capabilities, namespace)?;
let hash: iroh_blobs::Hash = hash.parse().context("parsing iroh-blob hash")?;
match self.blobs.store().blobs().status(hash).await? {
iroh_blobs::api::blobs::BlobStatus::Complete { size: actual } if actual == size => {}
iroh_blobs::api::blobs::BlobStatus::Complete { size: actual } => {
bail!("iroh-blob size mismatch: expected {size}, found {actual}");
}
_ => bail!("iroh-blob is not complete in the active browser store"),
}
let doc = self
.docs
.api()
.open(namespace)
.await?
.ok_or_else(|| anyhow!("iroh-docs namespace is unavailable"))?;
doc.set_hash(self.default_author, key.clone(), hash, size)
.await?;
let entry = self
.sync
.get_exact(namespace, self.default_author, key.into(), true)
.await?
.ok_or_else(|| anyhow!("local iroh-docs entry was not committed"))?;
self.store.persist_entry(&entry, true).await
}
pub(crate) async fn delete_prefix(
&self,
namespace: &str,
prefix: Vec<u8>,
) -> Result<WasmDeleteReceipt> {
let namespace = parse_namespace(namespace)?;
require_writable_namespace(&self.capabilities, namespace)?;
let doc = self
.docs
.api()
.open(namespace)
.await?
.ok_or_else(|| anyhow!("iroh-docs namespace is unavailable"))?;
let removed = doc.del(self.default_author, prefix.clone()).await?;
let tombstone = self
.sync
.get_exact(namespace, self.default_author, prefix.into(), true)
.await?
.ok_or_else(|| anyhow!("iroh-docs tombstone was not committed"))?;
let operation_id = self.store.persist_entry(&tombstone, true).await?;
Ok(WasmDeleteReceipt {
removed: u32::try_from(removed).context("deleted iroh-docs entry count exceeds u32")?,
operation_id,
})
}
pub(crate) async fn query(
&self,
namespace: &str,
key_prefix: Vec<u8>,
) -> Result<Vec<WasmDocEntry>> {
let namespace = parse_namespace(namespace)?;
require_namespace(&self.capabilities, namespace)?;
let doc = self
.docs
.api()
.open(namespace)
.await?
.ok_or_else(|| anyhow!("iroh-docs namespace is unavailable"))?;
let entries = doc
.get_many(Query::single_latest_per_key().key_prefix(key_prefix))
.await?;
futures::pin_mut!(entries);
let mut output = Vec::new();
while let Some(entry) = entries.next().await {
let entry = entry?;
output.push(WasmDocEntry {
namespace_id: entry.namespace().to_string(),
author_id: entry.author().to_string(),
key: entry.key().to_vec(),
timestamp: entry.timestamp(),
content_hash: entry.content_hash().to_string(),
content_length: entry.content_len(),
});
}
Ok(output)
}
pub(crate) async fn hydrate_blob(&self, hash: &str) -> Result<()> {
let hash: iroh_blobs::Hash = hash.parse().context("parsing iroh-blob hash")?;
let mut status = self.blobs.store().blobs().status(hash).await?;
if !matches!(status, iroh_blobs::api::blobs::BlobStatus::Complete { .. }) {
let now_ms = js_sys::Date::now().max(0.0) as u64;
let retry_allowed = {
let mut retry_after = self
.blob_retry_after_ms
.lock()
.expect("browser blob retry mutex poisoned");
if retry_after
.get(&hash)
.is_some_and(|retry_at| *retry_at > now_ms)
{
false
} else {
retry_after.insert(hash, now_ms.saturating_add(5_000));
true
}
};
if retry_allowed {
let mut providers = HashSet::new();
for namespace in self.capabilities.keys().copied() {
if let Some(peers) = self.sync.get_sync_peers(namespace).await? {
for peer in peers {
providers.insert(iroh::PublicKey::from_bytes(&peer)?);
}
}
}
if !providers.is_empty() {
let _ = self
.downloader
.download(
iroh_blobs::HashAndFormat::raw(hash),
providers.into_iter().collect::<Vec<_>>(),
)
.await;
status = self.blobs.store().blobs().status(hash).await?;
}
}
}
let size = match status {
iroh_blobs::api::blobs::BlobStatus::Complete { size } => {
self.blob_retry_after_ms
.lock()
.expect("browser blob retry mutex poisoned")
.remove(&hash);
size
}
_ => bail!("persistent browser blob is unavailable"),
};
if size == 0 {
return Ok(());
}
let mut reader = self.blobs.store().blobs().reader(hash);
let mut byte = [0u8; 1];
reader
.read_exact(&mut byte)
.await
.context("reading persistent browser blob prefix")?;
reader
.seek(std::io::SeekFrom::Start(size - 1))
.await
.context("seeking persistent browser blob suffix")?;
reader
.read_exact(&mut byte)
.await
.context("reading persistent browser blob suffix")?;
Ok(())
}
pub(crate) async fn acknowledge_outbox(&self, operation_id: &str) -> Result<()> {
if operation_id.trim().is_empty() {
bail!("iroh-docs outbox operation id is required");
}
self.store.acknowledge_outbox(operation_id).await
}
fn require_current_or_new_capability(
&self,
namespace: NamespaceId,
writable: bool,
generation: u64,
share_revision: u64,
) -> Result<()> {
let Some(current) = self.capabilities.get(&namespace) else {
return Ok(());
};
if (generation, share_revision) < (current.generation, current.share_revision) {
bail!("stale iroh-docs capability update");
}
if (generation, share_revision) == (current.generation, current.share_revision)
&& writable != current.writable
{
bail!("conflicting iroh-docs capability at the same generation and revision");
}
Ok(())
}
pub(crate) async fn flush(&self) -> Result<()> {
self.sync.flush_store().await?;
self.store.flush().await
}
pub(crate) async fn shutdown(&self) -> Result<()> {
self.flush().await?;
self.sync.shutdown().await?;
Ok(())
}
}
fn namespace_descriptor(capability: &Capability) -> WasmNamespaceDescriptor {
let (_, bytes) = capability.raw();
WasmNamespaceDescriptor {
namespace_id: capability.id().to_string(),
capability_kind: capability_kind(capability),
capability: bytes.to_vec(),
}
}
fn capability_kind(capability: &Capability) -> &'static str {
if capability.raw().0 == 1 {
"write"
} else {
"read"
}
}
fn parse_namespace(namespace: &str) -> Result<NamespaceId> {
namespace.parse().context("parsing iroh-docs namespace")
}
fn require_namespace(
capabilities: &HashMap<NamespaceId, RuntimeCapabilityFence>,
namespace: NamespaceId,
) -> Result<RuntimeCapabilityFence> {
capabilities
.get(&namespace)
.copied()
.ok_or_else(|| anyhow!("iroh-docs namespace is not imported by this runtime"))
}
fn require_writable_namespace(
capabilities: &HashMap<NamespaceId, RuntimeCapabilityFence>,
namespace: NamespaceId,
) -> Result<()> {
if !require_namespace(capabilities, namespace)?.writable {
bail!("iroh-docs namespace is read-only at the current capability generation");
}
Ok(())
}
fn ensure_js_safe_integer(value: u64, field: &str) -> Result<()> {
if value > 9_007_199_254_740_991 {
bail!("{field} exceeds JavaScript's safe integer range");
}
Ok(())
}
fn js_error(value: JsValue) -> anyhow::Error {
if let Some(message) = value.as_string() {
return anyhow!(message);
}
if let Ok(message) = Reflect::get(&value, &JsValue::from_str("message")) {
if let Some(message) = message.as_string() {
return anyhow!(message);
}
}
anyhow!("JavaScript IndexedDB adapter rejected an operation")
}
fn ignore_newer_entry(error: anyhow::Error) -> Result<()> {
if error.to_string().contains("newer entry exists") {
Ok(())
} else {
Err(error)
}
}