eidetica/database/mod.rs
1//! Database module provides functionality for managing collections of related entries.
2//!
3//! A `Database` represents a hierarchical structure of entries, like a traditional database
4//! or a branch in a version control system. Each database has a root entry and maintains
5//! the history and relationships between entries. Database holds a weak reference to its
6//! parent Instance, accessing storage and coordination services through that handle.
7
8use std::{future::Future, sync::Arc};
9
10use rand::{Rng, RngCore, distributions::Alphanumeric};
11use serde_json;
12
13#[cfg(all(unix, feature = "service"))]
14use crate::instance::backend::RemoteBackend;
15#[cfg(all(unix, feature = "service"))]
16use crate::service::client::RemoteConnection;
17use crate::{
18 Error, Instance, Result, Snapshot, Transaction, WeakInstance,
19 auth::{
20 crypto::{PrivateKey, PublicKey},
21 errors::AuthError,
22 settings::AuthSettings,
23 types::{AuthKey, Permission, SigKey},
24 validation::AuthValidator,
25 },
26 backend::VerificationStatus,
27 constants::{ROOT, SETTINGS},
28 crdt::{CRDT, Doc},
29 entry::{Entry, ID},
30 instance::{WriteCallback, WriteEvent, WriteSource, backend::Backend, errors::InstanceError},
31 store::{SettingsStore, Store},
32};
33
34#[cfg(test)]
35mod tests;
36
37tokio::task_local! {
38 /// Set while a `verify()`/validation pass is on the call stack.
39 ///
40 /// Verification reads the database (delegation resolution opens trees,
41 /// reads settings → tips), and the access-time auto-verify hook in
42 /// [`Database::snapshot`] would otherwise re-enter verification
43 /// unboundedly. While this is set, the hook is suppressed and reads
44 /// return raw (still `Failed`-filtered) tips.
45 static IN_VERIFY: bool;
46}
47
48fn auto_verify_suppressed() -> bool {
49 IN_VERIFY.try_with(|v| *v).unwrap_or(false)
50}
51
52/// Outcome of reconstructing the `_settings` state an entry pins.
53///
54/// An entry records, in its signed metadata, the `_settings` tips its
55/// signature must be validated against. We can only verify it if this node
56/// holds that full pinned `_settings` ancestor set.
57enum PinnedSettings {
58 /// The pinned `_settings` set is fully present; here is its auth config.
59 Complete(AuthSettings),
60 /// This node does not yet hold the full pinned `_settings` set, so the
61 /// entry cannot be verified yet (it stays `Unverified`).
62 Incomplete,
63}
64
65/// Summary of a [`Database::verify`] pass.
66#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
67pub struct VerifyReport {
68 /// Entries promoted `Unverified` → `Verified` this pass.
69 pub verified: usize,
70 /// Entries marked `Unverified` → `Failed` (definitively bad) this pass.
71 pub failed: usize,
72 /// Entries left `Unverified` (pinned `_settings` not yet held locally).
73 pub still_unverified: usize,
74}
75
76/// A signing key bound to its identity in a database's auth settings.
77///
78/// Pairs the cryptographic signing key with information about how to look up
79/// permissions in the database's auth configuration. The identity determines
80/// which entry in `_settings.auth` this key maps to.
81#[derive(Clone, Debug)]
82pub struct DatabaseKey {
83 signing_key: Box<PrivateKey>,
84 identity: SigKey,
85}
86
87impl DatabaseKey {
88 /// Identity = pubkey derived from signing key. Most common case.
89 pub fn new(signing_key: PrivateKey) -> Self {
90 let pubkey = signing_key.public_key();
91 Self {
92 signing_key: Box::new(signing_key),
93 identity: SigKey::from_pubkey(&pubkey),
94 }
95 }
96
97 /// Identity = explicit SigKey (name, global, delegation, etc.)
98 pub fn with_identity(signing_key: PrivateKey, identity: SigKey) -> Self {
99 Self {
100 signing_key: Box::new(signing_key),
101 identity,
102 }
103 }
104
105 /// Identity = global permission with actual pubkey embedded for verification.
106 pub fn global(signing_key: PrivateKey) -> Self {
107 let pubkey = signing_key.public_key();
108 Self {
109 signing_key: Box::new(signing_key),
110 identity: SigKey::global(&pubkey),
111 }
112 }
113
114 /// Identity = key name lookup.
115 pub fn with_name(signing_key: PrivateKey, name: impl Into<String>) -> Self {
116 Self {
117 signing_key: Box::new(signing_key),
118 identity: SigKey::from_name(name),
119 }
120 }
121
122 /// Get the signing key.
123 pub fn signing_key(&self) -> &PrivateKey {
124 &self.signing_key
125 }
126
127 /// Get the public key.
128 pub fn public_key(&self) -> PublicKey {
129 self.signing_key.public_key()
130 }
131
132 /// Get the identity used for auth settings lookup.
133 pub fn identity(&self) -> &SigKey {
134 &self.identity
135 }
136
137 /// Consume self and return the parts.
138 pub fn into_parts(self) -> (PrivateKey, SigKey) {
139 (*self.signing_key, self.identity)
140 }
141}
142
143impl From<PrivateKey> for DatabaseKey {
144 /// Convert a `PrivateKey` into a `DatabaseKey` with pubkey-derived identity.
145 ///
146 /// This is equivalent to [`DatabaseKey::new`] and covers the most common case
147 /// where the key's identity in auth settings is its own public key.
148 fn from(signing_key: PrivateKey) -> Self {
149 Self::new(signing_key)
150 }
151}
152
153/// Represents a collection of related entries, like a traditional database or a branch in a version control system.
154///
155/// Each `Database` is identified by the ID of its root `Entry` and manages the history of data
156/// associated with that root. It interacts with the underlying storage through the Instance handle.
157#[derive(Clone, Debug)]
158pub struct Database {
159 root: ID,
160 instance: WeakInstance,
161 /// Storage seam `Transaction`/`Store` reads flow through. On a local
162 /// instance this is a clone of the instance's own [`Backend`] (forwarding
163 /// to the backing engine); on a connected instance a per-handle
164 /// [`RemoteBackend`] bound to this database's acting identity. Derived from
165 /// `instance`/construction only — carrying it across
166 /// `with_key`/`allow_unverified` (`..self`) rebuilds is correct.
167 ops: Arc<dyn Backend>,
168 /// Signing key bound to its auth identity for this database
169 key: Option<DatabaseKey>,
170 /// When `false` (default), reads expose only the maximal all-`Verified`
171 /// prefix of the DAG (the "Verified frontier"). When `true`, reads also
172 /// include `Unverified` entries. `Failed` entries are dropped regardless.
173 /// Set via [`Database::allow_unverified`].
174 allow_unverified: bool,
175}
176
177impl Database {
178 /// Creates a new `Database` instance with a user-provided signing key.
179 ///
180 /// This constructor creates a new database using a signing key that's already in memory
181 /// (e.g., from UserKeyManager), without requiring the key to be stored in the backend.
182 /// This is the preferred method for creating databases in a User context where keys
183 /// are managed separately from the backend.
184 ///
185 /// The created database will use a `DatabaseKey` for all subsequent operations,
186 /// meaning transactions will use the provided key directly rather than looking it up
187 /// from backend storage.
188 ///
189 /// # Auth Bootstrapping
190 ///
191 /// Auth is always bootstrapped with the signing key as `Admin(0)`. Passing auth
192 /// configuration in `initial_settings` is an error — additional keys must be added
193 /// via follow-up transactions after creation.
194 ///
195 /// # Arguments
196 /// * `instance` - Instance handle for storage and coordination
197 /// * `signing_key` - The signing key to use for the initial commit and subsequent operations.
198 /// This key should already be decrypted and ready to use. The public key is derived
199 /// automatically and used as the key identifier in auth settings.
200 /// * `initial_settings` - `Doc` CRDT containing the initial settings for the database.
201 /// Use `Doc::new()` for an empty settings document.
202 ///
203 /// # Returns
204 /// A `Result` containing the new `Database` instance configured with a `DatabaseKey`.
205 ///
206 /// # Example
207 /// ```rust,no_run
208 /// # use eidetica::*;
209 /// # use eidetica::backend::database::InMemory;
210 /// # use eidetica::auth::crypto::generate_keypair;
211 /// # use eidetica::crdt::Doc;
212 /// # #[tokio::main]
213 /// # async fn main() -> Result<()> {
214 /// let instance = Instance::open_backend(Box::new(InMemory::new())).await?;
215 /// let (signing_key, _public_key) = generate_keypair();
216 ///
217 /// let mut settings = Doc::new();
218 /// settings.set("name", "my_database");
219 ///
220 /// // Create database with user-managed key (no backend storage needed)
221 /// let database = Database::create(&instance, signing_key, settings).await?;
222 ///
223 /// // All transactions automatically use the provided key
224 /// let tx = database.new_transaction().await?;
225 /// # Ok(())
226 /// # }
227 /// ```
228 pub async fn create(
229 instance: &Instance,
230 signing_key: PrivateKey,
231 initial_settings: Doc,
232 ) -> Result<Self> {
233 Self::create_with_init(instance, signing_key, initial_settings, async |_| Ok(())).await
234 }
235
236 /// Creates a new database with an initialization callback that runs inside
237 /// the genesis transaction.
238 ///
239 /// This is the underlying constructor used by [`Self::create`] and by
240 /// `User::new_database()` (the builder API). The callback receives the
241 /// genesis transaction after `_settings` and `_root` have been staged but
242 /// before commit, allowing additional subtrees — stores, initial doc data,
243 /// records — to be written into the same entry that establishes the
244 /// database root.
245 ///
246 /// All writes performed in the callback become part of the single genesis
247 /// entry: one signed entry, one backend write, atomic. If the callback
248 /// returns an error, the transaction is dropped without committing and no
249 /// database is created.
250 ///
251 /// # Arguments
252 /// * `instance` - Instance handle for storage and coordination
253 /// * `signing_key` - The private key for this database (becomes Admin(0))
254 /// * `initial_settings` - Initial settings document (must not contain auth)
255 /// * `init` - Callback run against the genesis transaction before commit
256 pub async fn create_with_init<F>(
257 instance: &Instance,
258 signing_key: PrivateKey,
259 initial_settings: Doc,
260 init: F,
261 ) -> Result<Self>
262 where
263 F: AsyncFnOnce(&Transaction) -> Result<()>,
264 {
265 let mut initial_settings = initial_settings;
266 let pubkey = signing_key.public_key();
267
268 // Reject preconfigured auth — Database::create owns auth bootstrapping entirely.
269 if initial_settings.get("auth").is_some() {
270 return Err(Error::Auth(Box::new(AuthError::InvalidAuthConfiguration {
271 reason: "initial_settings must not contain auth configuration; \
272 Database::create bootstraps auth with the signing key as Admin(0)"
273 .to_string(),
274 })));
275 }
276
277 // Bootstrap auth with the signing key as Admin(0)
278 let mut auth_settings = AuthSettings::new();
279 auth_settings.add_key(&pubkey, AuthKey::active(None, Permission::Admin(0)))?;
280 initial_settings.set("auth", auth_settings.as_doc().clone());
281
282 // Create the initial root entry using a temporary Database and Transaction.
283 // This placeholder ID should not exist in the backend, so snapshot will be empty.
284 let bootstrap_placeholder_id = format!(
285 "bootstrap_root_{}",
286 rand::thread_rng()
287 .sample_iter(&Alphanumeric)
288 .take(10)
289 .map(char::from)
290 .collect::<String>()
291 );
292
293 // Create temporary database for bootstrap with DatabaseKey.
294 // This allows the bootstrap transaction to use the provided key directly.
295 let temp_database_for_bootstrap = Database {
296 root: ID::from_bytes(bootstrap_placeholder_id.as_bytes()),
297 instance: instance.downgrade(),
298 ops: instance.backend().clone(),
299 key: Some(DatabaseKey::new(signing_key.clone())),
300 allow_unverified: false,
301 };
302
303 // Create the transaction - it will use the provided key automatically
304 let txn = temp_database_for_bootstrap.new_transaction().await?;
305
306 // IMPORTANT: For the root entry, we need to set the database root to empty/default
307 // so that is_root() returns true and all_roots() can find it
308 txn.set_entry_root(ID::default())?;
309
310 // Populate the SETTINGS and ROOT subtrees for the very first entry
311 txn.update_subtree(SETTINGS, serde_json::to_vec(&initial_settings)?)
312 .await?;
313 txn.update_subtree(ROOT, serde_json::to_vec("")?).await?; // Standard practice for root entry's _root
314
315 // Add entropy to the entry metadata to ensure unique database IDs even with identical settings
316 txn.set_metadata_entropy(rand::thread_rng().next_u64())?;
317
318 // Lock system subtrees (`_settings`, `_root`, `_index`) for the
319 // duration of the init callback. The callback gets a `&Transaction`
320 // that rejects `_*` opens via `get_store` / `Store::open`, so callers
321 // can't accidentally clobber subtrees that `create_with_init` itself
322 // manages. Internal paths (Registry, Store::register's `_index`
323 // writes) bypass `get_store` and remain functional. The guard releases
324 // on drop, so the lock lifts on both early-return and panic-unwind.
325 {
326 let _lock = txn.lock_system_subtrees();
327 init(&txn).await?;
328 }
329
330 // Commit the initial entry
331 let new_root_id = txn.commit().await?;
332
333 // Construct the returned Database, wiring its `ops` to match the
334 // instance's flavour:
335 //
336 // - **Connected (remote) instance** — bind a `RemoteBackend` to the
337 // *new database's* identity (the signing-key's pubkey, self-signed
338 // as `Admin(0)` by the genesis). The connection's login pubkey is
339 // the *caller's* (e.g. the registering admin), which is **not** a
340 // member of the new tree's auth. If we cloned the instance's
341 // session backend here, every read `Transaction::commit` performs
342 // on this database would carry the connection's login identity, and
343 // the server's per-tree gate would deny it. A per-database identity
344 // makes all reads use the tree's own member key — but the server's
345 // gate also requires that key to be in the connection's *session
346 // keyset*, so we `register_session_key(signing_key)` first to do the
347 // proof-of-possession handshake that adds it.
348 // - **Local instance** — clone the instance's backend, unchanged.
349 #[cfg(all(unix, feature = "service"))]
350 if let Some(conn) = instance.remote_connection() {
351 let pubkey_for_identity = signing_key.public_key();
352 conn.register_session_key(&signing_key).await?;
353 return Ok(Self {
354 root: new_root_id.clone(),
355 instance: instance.downgrade(),
356 ops: Arc::new(RemoteBackend::new(
357 conn,
358 Some(SigKey::from_pubkey(&pubkey_for_identity)),
359 )),
360 key: Some(DatabaseKey::new(signing_key)),
361 allow_unverified: false,
362 });
363 }
364
365 Ok(Self {
366 root: new_root_id,
367 instance: instance.downgrade(),
368 ops: instance.backend().clone(),
369 key: Some(DatabaseKey::new(signing_key)),
370 allow_unverified: false,
371 })
372 }
373
374 /// Opens an existing database by its root ID.
375 ///
376 /// Verifies the root entry exists in the backend, then returns a handle
377 /// for read-only access. To perform authenticated writes, chain
378 /// `.with_key(key)` after opening.
379 ///
380 /// # Arguments
381 /// * `instance` - Instance handle for storage and coordination
382 /// * `root_id` - The root entry ID of the database to open
383 ///
384 /// # Errors
385 /// Returns an error if the root entry does not exist in the backend.
386 ///
387 /// # Example
388 /// ```rust,no_run
389 /// # use eidetica::*;
390 /// # use eidetica::backend::database::InMemory;
391 /// # use eidetica::auth::crypto::generate_keypair;
392 /// # #[tokio::main]
393 /// # async fn main() -> Result<()> {
394 /// # let instance = Instance::open_backend(Box::new(InMemory::new())).await?;
395 /// # let (signing_key, _verifying_key) = generate_keypair();
396 /// # let root_id = ID::from_bytes(b"existing_database_root_id");
397 /// // Open database for reading
398 /// let db = Database::open(&instance, &root_id).await?;
399 ///
400 /// // Open database with a signing key for writes
401 /// let db = Database::open(&instance, &root_id).await?.with_key(signing_key);
402 /// let tx = db.new_transaction().await?;
403 /// # Ok(())
404 /// # }
405 /// ```
406 pub async fn open(instance: &Instance, root_id: &ID) -> Result<Self> {
407 // Verify the root entry exists. Surfaces "database doesn't exist"
408 // at open time instead of at first read/write.
409 instance.backend().get(root_id).await?;
410
411 Ok(Self {
412 root: root_id.clone(),
413 instance: instance.downgrade(),
414 ops: instance.backend().clone(),
415 key: None,
416 allow_unverified: false,
417 })
418 }
419
420 /// Open a database for remote access through a service connection.
421 ///
422 /// Constructs a [`Database`] handle whose backing
423 /// [`Backend`](crate::instance::backend::Backend) is a
424 /// [`RemoteBackend`](crate::instance::backend::RemoteBackend) bound to
425 /// `identity`, so every [`Transaction`]/[`Store`] read and write travels
426 /// over the connection as a `DatabaseOp` under that identity. The
427 /// `identity` must match the database's auth settings for the caller's
428 /// key. `Instance::connect` must be used to create the instance.
429 #[cfg(all(unix, feature = "service"))]
430 pub async fn open_remote(
431 instance: &Instance,
432 conn: RemoteConnection,
433 root_id: &ID,
434 identity: SigKey,
435 ) -> Result<Self> {
436 instance.backend().get(root_id).await?;
437 Ok(Self {
438 root: root_id.clone(),
439 instance: instance.downgrade(),
440 ops: Arc::new(RemoteBackend::new(conn, Some(identity))),
441 key: None,
442 allow_unverified: false,
443 })
444 }
445
446 /// Attach a signing key to this database handle.
447 ///
448 /// The key is stored for use by future transactions. No validation is
449 /// performed; invalid keys will cause errors at commit time or when
450 /// calling [`current_permission`](Self::current_permission).
451 ///
452 /// Calling `with_key` again replaces any previously-attached key — the
453 /// most recent call wins.
454 ///
455 /// To discover which `SigKey` identity to use for a given public key,
456 /// use [`Database::find_sigkeys`].
457 pub fn with_key(self, key: impl Into<DatabaseKey>) -> Self {
458 Self {
459 key: Some(key.into()),
460 ..self
461 }
462 }
463
464 /// Include `Unverified` entries in this handle's reads.
465 ///
466 /// By default a `Database` exposes only the **Verified frontier**: the
467 /// maximal prefix of the DAG (an ancestor-closed set, starting at the
468 /// root) in which every entry is `Verified`. Tips that are still
469 /// `Unverified` — and everything reachable only through them — are hidden,
470 /// so a default read never reflects state this node could not authenticate.
471 ///
472 /// Calling `allow_unverified` opts this handle into the looser view that
473 /// also includes `Unverified` entries (everything except `Failed`, which
474 /// is always dropped). Use it when you explicitly want to observe
475 /// not-yet-verified data — e.g. freshly synced entries whose pinned
476 /// `_settings` this node does not hold yet.
477 ///
478 /// This is a per-handle setting and composes with [`with_key`](Self::with_key):
479 ///
480 /// ```rust,no_run
481 /// # use eidetica::*;
482 /// # use eidetica::auth::crypto::generate_keypair;
483 /// # async fn example(instance: Instance, root_id: ID) -> Result<()> {
484 /// # let (signing_key, _) = generate_keypair();
485 /// let db = Database::open(&instance, &root_id)
486 /// .await?
487 /// .with_key(signing_key)
488 /// .allow_unverified();
489 /// # Ok(())
490 /// # }
491 /// ```
492 ///
493 /// # Note on CRDT coherence
494 ///
495 /// The Verified frontier is a *prefix* cut, not a per-value filter. An
496 /// interior `Unverified` entry hides all of its descendants from the
497 /// default view even if those descendants are themselves `Verified`,
498 /// because exposing them without their unverifiable ancestor would yield
499 /// an incoherent CRDT state. Run [`verify`](Self::verify) to promote the
500 /// blocking entry, or `allow_unverified` to read past it.
501 pub fn allow_unverified(self) -> Self {
502 Self {
503 allow_unverified: true,
504 ..self
505 }
506 }
507
508 /// Validate a `DatabaseKey` against this database's auth settings.
509 ///
510 /// Checks that:
511 /// 1. The signing key derives to the public key claimed by the identity
512 /// 2. The identity exists in the database's auth settings
513 ///
514 /// Returns the effective permission for the validated key. Callers wanting
515 /// to fail fast on an invalid key should call
516 /// [`current_permission`](Self::current_permission), which wraps this.
517 async fn validate_key(&self, key: &DatabaseKey) -> Result<Permission> {
518 let settings_store = self.get_settings().await?;
519 let auth_settings = settings_store.auth_snapshot().await?;
520 let actual_pubkey = key.public_key();
521 let instance = match key.identity() {
522 // Delegation resolution needs an Instance for cross-tree lookups;
523 // direct identities resolve from `auth_settings` alone.
524 SigKey::Delegation { .. } => Some(self.instance()?),
525 _ => None,
526 };
527 crate::auth::validation::permissions::resolve_identity_permission(
528 &actual_pubkey,
529 key.identity(),
530 &auth_settings,
531 instance.as_ref(),
532 )
533 .await
534 }
535
536 /// Find all SigKeys that a public key can use to access a database.
537 ///
538 /// This static helper method loads a database's authentication settings and returns
539 /// all possible SigKeys that can be used with the given public key. This is useful for
540 /// discovering authentication options before opening a database.
541 ///
542 /// Returns all matching SigKeys including:
543 /// - Specific key names where the pubkey matches
544 /// - Global permission if available
545 /// - Single-hop delegation paths (pubkey found in a directly delegated tree)
546 ///
547 /// The results are **sorted by permission level, highest first**, making it easy to
548 /// select the most privileged access available.
549 ///
550 /// # Arguments
551 /// * `instance` - Instance handle for storage and coordination
552 /// * `root_id` - Root entry ID of the database to check
553 /// * `pubkey` - Public key string (e.g., "Ed25519:abc123...") to look up
554 ///
555 /// # Returns
556 /// A vector of (SigKey, Permission) tuples, sorted by permission (highest first).
557 /// Returns empty vector if no valid access methods are found.
558 ///
559 /// # Errors
560 /// Returns an error if:
561 /// - Database cannot be loaded
562 /// - Auth settings cannot be parsed
563 ///
564 /// # Example
565 /// ```rust,no_run
566 /// # use eidetica::*;
567 /// # use eidetica::database::DatabaseKey;
568 /// # use eidetica::backend::database::InMemory;
569 /// # use eidetica::auth::crypto::generate_keypair;
570 /// # use eidetica::auth::types::SigKey;
571 /// # #[tokio::main]
572 /// # async fn main() -> Result<()> {
573 /// # let instance = Instance::open_backend(Box::new(InMemory::new())).await?;
574 /// # let (signing_key, pubkey) = generate_keypair();
575 /// # let root_id = ID::from_bytes(b"database_root_id");
576 /// // Find all SigKeys this pubkey can use (sorted highest permission first)
577 /// let sigkeys = Database::find_sigkeys(&instance, &root_id, &pubkey).await?;
578 ///
579 /// // Use the first available SigKey (highest permission)
580 /// if let Some((sigkey, _permission)) = sigkeys.first() {
581 /// let key = DatabaseKey::with_identity(signing_key, sigkey.clone());
582 /// let database = Database::open(&instance, &root_id).await?.with_key(key);
583 /// }
584 /// # Ok(())
585 /// # }
586 /// ```
587 pub async fn find_sigkeys(
588 instance: &Instance,
589 root_id: &ID,
590 pubkey: &PublicKey,
591 ) -> Result<Vec<(SigKey, Permission)>> {
592 use crate::auth::{
593 types::DelegationStep,
594 validation::{AuthValidator, permissions::select_effective_permission},
595 };
596
597 // Create temporary database to load settings (no key source needed for reading)
598 let temp_db = Self::open(instance, root_id).await?;
599
600 // Load auth settings
601 let settings_store = temp_db.get_settings().await?;
602 let auth_settings = settings_store.auth_snapshot().await?;
603
604 // Find direct SigKeys for this pubkey
605 let mut results = auth_settings.find_all_sigkeys_for_pubkey(pubkey);
606
607 // Scan single-hop delegation paths. Resolution — permission-bounds
608 // clamping, key status, the tip floor — is delegated to the validator,
609 // the single delegation walker, so this stays pure discovery: find which
610 // delegated trees list the pubkey, then resolve each through the shared
611 // path at the delegated tree's *current* tips.
612 // FIXME: deep nested delegations can't use this
613 if let Ok(delegated_trees) = auth_settings.get_all_delegated_trees() {
614 let mut validator = AuthValidator::new();
615 for delegated_root_id in delegated_trees.keys() {
616 // Load the delegated tree's auth to see which of the pubkey's
617 // hints it lists (enumeration only — the validator re-resolves).
618 let delegated_auth = match Self::open(instance, delegated_root_id).await {
619 Ok(db) => match db.get_settings().await {
620 Ok(s) => match s.auth_snapshot().await {
621 Ok(a) => a,
622 Err(_) => continue,
623 },
624 Err(_) => continue,
625 },
626 Err(_) => continue,
627 };
628 let delegated_sigkeys = delegated_auth.find_all_sigkeys_for_pubkey(pubkey);
629 if delegated_sigkeys.is_empty() {
630 continue;
631 }
632
633 // Current tips = the "live authority" question (caller-supplied),
634 // distinct from a signature's claimed tips. They are at or above
635 // the committed floor, so the validator's floor check passes.
636 let tips = match instance.backend().snapshot(delegated_root_id).await {
637 Ok(snap) => snap.tips().to_vec(),
638 Err(_) => continue,
639 };
640
641 for (delegated_sk, _) in delegated_sigkeys {
642 let delegation_sigkey = SigKey::Delegation {
643 path: vec![DelegationStep {
644 tree: delegated_root_id.clone(),
645 tips: tips.clone(),
646 }],
647 hint: delegated_sk.hint().clone(),
648 };
649 // Resolve through the single delegation walker; bounds,
650 // status, and the tip floor all live there, not here.
651 let Ok(resolved) = validator
652 .resolve_sig_key(&delegation_sigkey, &auth_settings, Some(instance))
653 .await
654 else {
655 continue;
656 };
657 if let Some(perm) = select_effective_permission(&resolved, pubkey) {
658 results.push((delegation_sigkey, perm));
659 }
660 }
661 }
662 }
663
664 // Sort by permission, highest first
665 results.sort_by_key(|b| std::cmp::Reverse(b.1));
666 Ok(results)
667 }
668
669 /// Whether `pubkey` may access the database at `root_id` with at least
670 /// `permission`.
671 ///
672 /// This is the **pubkey-only** access decision — the caller holds a key but
673 /// has not presented a signing identity (bootstrap deciding whether a key
674 /// already qualifies, for example). It resolves the same authority
675 /// [`find_sigkeys`](Self::find_sigkeys) does — direct grants, the global
676 /// `*` grant, and delegated authority — and never counts revoked keys.
677 ///
678 /// Delegated authority is discovered **one hop deep**: only databases this
679 /// one delegates to directly are searched, so a key reachable through a
680 /// chain of delegations answers `false` here. The limit belongs to
681 /// discovery, not to delegation — when an identity *is* presented (signing
682 /// an entry, a service operation), authorization goes through the resolver
683 /// behind [`validate_key`](Self::validate_key), which walks a delegation
684 /// path of any length because the signer names the path rather than making
685 /// the resolver search for it.
686 pub async fn can_access(
687 instance: &Instance,
688 root_id: &ID,
689 pubkey: &PublicKey,
690 permission: &Permission,
691 ) -> Result<bool> {
692 let sigkeys = Self::find_sigkeys(instance, root_id, pubkey).await?;
693 Ok(sigkeys
694 .first()
695 .is_some_and(|(_, granted)| *granted >= *permission))
696 }
697
698 /// Get the auth identity for this database's configured key.
699 pub fn auth_identity(&self) -> Option<&SigKey> {
700 self.key.as_ref().map(|k| &k.identity)
701 }
702
703 /// Register a callback to be invoked when entries are written to this database.
704 ///
705 /// The callback fires for **both** local writes (transaction commits) and remote
706 /// writes (sync). Branch on [`WriteEvent::source`](crate::WriteEvent::source) inside
707 /// the closure if you only care about one.
708 ///
709 /// Returns a [`WriteCallback`] handle. **Drop it to unregister.** Call
710 /// [`WriteCallback::detach`] to leave the callback registered for the life
711 /// of the [`Instance`] without holding the handle.
712 ///
713 /// **Important:** Callbacks are registered at the Instance level and fire for all
714 /// writes to the database tree (identified by root ID), regardless of which
715 /// `Database` handle performed the write or registered the callback.
716 ///
717 /// # Callback contract
718 ///
719 /// - **Local writes**: fires once per transaction commit. The
720 /// [`WriteEvent`] carries cursor brackets only — call
721 /// [`Database::ids_added`] with `event.previous_tips()` and
722 /// `event.post_tips()` to enumerate the single new entry.
723 /// - **Remote writes**: fires once per sync batch (not per entry).
724 /// `ids_added(prev, post)` yields the full set of new IDs in topo
725 /// order.
726 /// - All entries that fall inside the cursor advance are fully
727 /// persisted before the callback fires.
728 /// - Errors are logged but do not prevent other callbacks from running.
729 /// - The `db` argument is a **read-only** [`Database`] handle (no
730 /// [`DatabaseKey`] configured): you can read settings, entries, and
731 /// metadata, but cannot commit transactions through it. To write from
732 /// inside a callback, resolve through `db.instance()?` and open the
733 /// database with the appropriate key.
734 /// - **Reentrance**: writes are serialized per-tree via an async lock
735 /// that is held while callbacks run. A callback must not commit a
736 /// transaction on the same tree it was invoked for — that would
737 /// deadlock. Spawn a task or write to a different tree instead.
738 ///
739 /// # Callback timing
740 ///
741 /// On a **local** [`Instance`], callbacks complete before
742 /// [`Transaction::commit`](crate::Transaction::commit)`.await` returns —
743 /// the firing path is inline with the write and the per-tree lock is
744 /// held the whole way.
745 ///
746 /// On a **connected** [`Instance`] (one built via
747 /// [`Instance::connect`](crate::Instance::connect)), callbacks fire
748 /// when a `Notification::DatabaseWrite` arrives back from the daemon —
749 /// typically microseconds after `commit().await` returns, but
750 /// asynchronous to it. This is by design: the daemon is the **sole
751 /// orderer** of writes on a connected setup, so every subscriber —
752 /// including the committing client — observes callbacks in the same
753 /// daemon-canonical order. Two clients submitting concurrently see
754 /// each other's writes in the same sequence; a single client's
755 /// callbacks never race a remote-source event into a different
756 /// observable order than the daemon has it. The trade-off is a small
757 /// asynchrony at the commit boundary — if you need a synchronous
758 /// "callback fired before commit returned" guarantee, use a local
759 /// [`Instance`].
760 ///
761 /// Ordering is preserved end-to-end, per tree. On the daemon the
762 /// per-tree write lock is held across the callback fan-out, and each
763 /// subscription's notification is emitted by a *synchronous* channel
764 /// send inside that locked section (see the `SubscribeWrites` handler
765 /// in `service::server`), so two clients writing the same tree
766 /// concurrently cannot interleave their frames. On the client,
767 /// notifications are routed by `root_id` to a per-tree worker that
768 /// `await`s each user callback to completion before pulling the next —
769 /// so two callbacks for the *same* tree never overlap and never
770 /// reorder. Callbacks for *different* trees run on independent workers
771 /// and progress concurrently: a slow callback on one tree does not
772 /// stall another. Spawn from inside the callback if you need fanout
773 /// within a single tree.
774 ///
775 /// # Settled-state trigger — but not a settled-state bracket
776 ///
777 /// Callbacks fire **only for entries that have passed local
778 /// verification** — i.e. entries the system considers `Verified`.
779 /// Direct `put_entry(Verified, _)` (local commits, trusted
780 /// in-process writes) fires immediately. Off-node entries that
781 /// arrive `Unverified` (sync ingest, wire-submitted entries) are
782 /// stored silently; the subsequent verification pass is where
783 /// promotion to `Verified` happens, and a Verified fire follows
784 /// from there once that pass runs.
785 ///
786 /// Both ingest paths run that pass inline: `put_remote_entries`
787 /// (sync) and the service `SubmitSignedEntry` handler each call
788 /// `verify()` immediately after storing the batch, so a promoted
789 /// entry fires its `Verified` event without waiting for a reader to
790 /// trigger the access-time auto-verify hook. A listen-only
791 /// subscriber therefore sees sync-arrived content promptly.
792 ///
793 /// What fires is settled; what the event *brackets* is not. The
794 /// event's cursors are raw DAG frontiers, so an `Unverified` or
795 /// `Failed` entry that happens to be a tip lands inside the bracket
796 /// and [`Self::ids_added`] will enumerate it. Filter on
797 /// `Backend::get_verification_status` before acting on IDs derived
798 /// from a bracket if your callback ingests entry contents.
799 ///
800 /// On a connected instance the first `on_write` registration for a
801 /// given tree lazily sends a `SubscribeWrites` op to the daemon;
802 /// further registrations on the same tree reuse that subscription.
803 /// Dropping the last callback for a tree marks the subscription idle
804 /// rather than unsubscribing immediately, so a quick re-registration
805 /// costs no round-trip; a sweep sends `UnsubscribeWrites` once the
806 /// grace window elapses. Disconnecting the client unsubscribes
807 /// everything.
808 ///
809 /// # Example
810 /// ```rust,no_run
811 /// # use eidetica::*;
812 /// # use eidetica::crdt::Doc;
813 /// # use eidetica::backend::database::InMemory;
814 /// # use eidetica::auth::crypto::PrivateKey;
815 /// # #[tokio::main]
816 /// # async fn main() -> Result<()> {
817 /// let instance = Instance::open_backend(Box::new(InMemory::new())).await?;
818 /// # let signing_key = PrivateKey::generate();
819 /// # let database = Database::create(&instance, signing_key, Doc::new()).await?;
820 ///
821 /// let cb = database.on_write(|event, db| {
822 /// let source = event.source();
823 /// let prev = event.previous_tips().clone();
824 /// let post = event.post_tips().clone();
825 /// let db = db.clone();
826 /// async move {
827 /// let new_ids = db.ids_added(&prev, &post).await?;
828 /// println!("{} entries written to {} ({source:?})", new_ids.len(), db.root_id());
829 /// Ok(())
830 /// }
831 /// }).await?;
832 ///
833 /// // Drop `cb` to unregister, or:
834 /// cb.detach(); // keep registered for the life of the Instance
835 /// # Ok(())
836 /// # }
837 /// ```
838 ///
839 /// If a callback needs the [`Instance`], call [`Database::instance`] on
840 /// the `db` argument.
841 pub async fn on_write<F, Fut>(&self, callback: F) -> Result<WriteCallback>
842 where
843 F: for<'a> Fn(&'a WriteEvent, &'a Database) -> Fut + Send + std::marker::Sync + 'static,
844 Fut: std::future::Future<Output = Result<()>> + Send + 'static,
845 {
846 // Convenience: read current tips, then register at those tips.
847 //
848 // The "current tips" here come from this method's own
849 // `get_tips` call, **not** from any read the caller may have
850 // done before calling `on_write`. If the caller built initial
851 // state from a separate read at some `T0`, a write may have
852 // landed between that read and this `get_tips`, putting the
853 // cursor at `T1 ≥ T0`. The first callback fire's
854 // `previous_tips` will then be `T1`, not `T0`, and the caller
855 // observes `previous_tips != their_initial_tips` (a small
856 // race-window mismatch — typically one lock-acquisition apart
857 // on a local instance).
858 //
859 // Use [`Self::on_write_at_tips`] to close that window: pass
860 // exactly the tips your initial-state read used, and the first
861 // fire's `previous_tips` will match.
862 // Propagate rather than defaulting: `Snapshot::EMPTY` is the
863 // documented "I have no initial state; replay from the beginning"
864 // cursor, so masking a transient read error here would silently turn
865 // "subscribe from now" into a full-history replay on the first fire.
866 let tips = self.snapshot().await?;
867 self.on_write_at_tips(tips, callback).await
868 }
869
870 /// Register a callback with an explicit initial cursor.
871 ///
872 /// Primitive form of [`Self::on_write`]. The `tips` you pass become
873 /// this callback's initial cursor, and the first event the
874 /// callback receives will have `previous_tips = tips`. Subsequent
875 /// events' `previous_tips` are the previous event's post-write
876 /// tips — each callback owns its own continuous timeline.
877 ///
878 /// Use this when you need a hard guarantee that the cursor matches
879 /// some other tip set you've already used (typically the tips
880 /// returned by an initial-state read you did before subscribing):
881 ///
882 /// ```rust,no_run
883 /// # use eidetica::*;
884 /// # async fn doit(db: Database) -> Result<()> {
885 /// let tips = db.snapshot().await?;
886 /// // ... read initial state at `tips` ...
887 /// let _cb = db.on_write_at_tips(tips, |_event, _db| async move { Ok(()) }).await?;
888 /// // First callback fire's previous_tips will exactly equal the
889 /// // `tips` we read at — no race-window mismatch with subsequent
890 /// // commits.
891 /// # Ok(())
892 /// # }
893 /// ```
894 pub async fn on_write_at_tips<F, Fut>(
895 &self,
896 tips: Snapshot,
897 callback: F,
898 ) -> Result<WriteCallback>
899 where
900 F: for<'a> Fn(&'a WriteEvent, &'a Database) -> Fut + Send + std::marker::Sync + 'static,
901 Fut: std::future::Future<Output = Result<()>> + Send + 'static,
902 {
903 let instance = self.instance()?;
904 let tree_id = self.root_id().clone();
905 // Local registry uses `tips` as the per-callback cursor; the
906 // wire path (if any) also needs it so the daemon-side
907 // subscription cursor is pinned at the same value. Clone once
908 // here; a `Snapshot` is a tiny tip-set.
909 let id = instance.register_write_callback(tree_id.clone(), tips.clone(), callback);
910 let cb = WriteCallback::new_per_database(instance.downgrade(), tree_id.clone(), id);
911
912 // On a connected (daemon-backed) instance, ensure the daemon is
913 // pushing notifications for this tree before returning — otherwise
914 // an immediately-following commit could race the subscribe and
915 // lose its notification, since the daemon doesn't replay missed
916 // events. `subscribe_writes` is itself concurrency-safe and
917 // idempotent: it short-circuits when the tree is already
918 // subscribed and serialises racing registrations through a
919 // per-tree `Notify`, so every caller observes the subscription
920 // as live before returning. Database handles built with a key
921 // carry an explicit identity; keyless handles fall back to
922 // `SigKey::default()`, which the daemon resolves to the
923 // connection's login pubkey.
924 //
925 // The `tips` we hand to `subscribe_writes` become the daemon's
926 // subscription cursor, so the first notification's
927 // `previous_tips` exactly equals the tips this caller passed
928 // in — no race-window mismatch between the daemon's view and
929 // the client's local cursor.
930 #[cfg(all(unix, feature = "service"))]
931 if let Some(conn) = instance.remote_connection() {
932 let identity = self.auth_identity().cloned().unwrap_or_default();
933 conn.subscribe_writes(tree_id, identity, tips).await?;
934 } else {
935 let _ = tips;
936 }
937
938 Ok(cb)
939 }
940
941 /// Get the ID of the root entry
942 pub fn root_id(&self) -> &ID {
943 &self.root
944 }
945
946 /// Upgrade the weak instance reference to a strong reference.
947 ///
948 /// `Database` holds a [`WeakInstance`](crate::WeakInstance), so this can
949 /// fail if the owning [`Instance`] has already been dropped.
950 pub fn instance(&self) -> Result<Instance> {
951 self.instance
952 .upgrade()
953 .ok_or_else(|| Error::Instance(Box::new(InstanceError::InstanceDropped)))
954 }
955
956 /// Get a clone of the backend seam.
957 pub fn backend(&self) -> Result<Arc<dyn Backend>> {
958 Ok(self.instance()?.backend().clone())
959 }
960
961 /// The storage seam this handle's `Transaction`/`Store` reads/writes flow
962 /// through (a [`LocalBackend`](crate::instance::backend::LocalBackend) clone
963 /// of the instance's backend, or a per-handle
964 /// [`RemoteBackend`](crate::instance::backend::RemoteBackend)).
965 pub(crate) fn ops(&self) -> &dyn Backend {
966 self.ops.as_ref()
967 }
968
969 /// Retrieve the root entry from the backend
970 pub async fn get_root(&self) -> Result<Entry> {
971 let instance = self.instance()?;
972 instance.get(&self.root).await
973 }
974
975 /// Get a read-only settings store for the database.
976 ///
977 /// Returns a SettingsStore that provides access to the database's settings.
978 /// Since this creates an internal transaction that is never committed, any
979 /// modifications made through the returned store will not persist.
980 ///
981 /// For making persistent changes to settings, create a transaction and use
982 /// `Transaction::get_settings()` instead.
983 ///
984 /// # Returns
985 /// A `Result` containing the `SettingsStore` for settings or an error.
986 ///
987 /// # Example
988 /// ```rust,no_run
989 /// # use eidetica::Database;
990 /// # async fn example(database: Database) -> eidetica::Result<()> {
991 /// // Read-only access
992 /// let settings = database.get_settings().await?;
993 /// let name = settings.get_name().await?;
994 ///
995 /// // For modifications, use a transaction:
996 /// let txn = database.new_transaction().await?;
997 /// let settings = txn.get_settings()?;
998 /// settings.set_name("new_name").await?;
999 /// txn.commit().await?;
1000 /// # Ok(())
1001 /// # }
1002 /// ```
1003 pub async fn get_settings(&self) -> Result<SettingsStore> {
1004 let txn = self.new_transaction().await?;
1005 txn.get_settings()
1006 }
1007
1008 /// Get the name of the database from its settings store
1009 pub async fn get_name(&self) -> Result<String> {
1010 let settings = self.get_settings().await?;
1011 settings.get_name().await
1012 }
1013
1014 /// Create a new atomic transaction on this database
1015 ///
1016 /// This creates a new atomic transaction containing a new Entry.
1017 /// The atomic transaction will be initialized with the current state of the database.
1018 /// If a default authentication key is set, the transaction will use it for signing.
1019 ///
1020 /// # Returns
1021 /// A `Result<Transaction>` containing the new atomic transaction
1022 pub async fn new_transaction(&self) -> Result<Transaction> {
1023 let snapshot = self.snapshot().await?;
1024 self.new_transaction_at(&snapshot).await
1025 }
1026
1027 /// Create a new atomic transaction on this database anchored at a specific snapshot.
1028 ///
1029 /// The transaction's parents are taken from the provided snapshot's tips instead of
1030 /// the database's current state. This allows creating complex DAG structures
1031 /// like diamond patterns for testing and advanced use cases.
1032 ///
1033 /// # Arguments
1034 /// * `snapshot` - The snapshot to anchor the transaction at
1035 ///
1036 /// # Returns
1037 /// A `Result<Transaction>` containing the new atomic transaction
1038 pub async fn new_transaction_at(&self, snapshot: &Snapshot) -> Result<Transaction> {
1039 let mut txn = Transaction::new_at(self, snapshot).await?;
1040
1041 // Set provided signing key from DatabaseKey
1042 if let Some(key) = &self.key {
1043 txn.set_provided_key(*key.signing_key.clone(), key.identity.clone());
1044 }
1045
1046 Ok(txn)
1047 }
1048
1049 /// Gather everything a client needs to build and sign a transaction
1050 /// locally for the given stores, with parents drawn from `scope`'s
1051 /// projection.
1052 ///
1053 /// This is **single-sourced**: both the server's `BeginTransaction`
1054 /// handler and the Phase-3 remote seam call it, so
1055 /// `Transaction::commit`'s build-sign path has one source of truth for
1056 /// context gathering.
1057 ///
1058 /// `scope=AllowUnverified` opens against the raw DAG (only `Failed`
1059 /// dropped); the default `Verified` scope uses the Verified frontier.
1060 /// The returned [`TransactionContext`] carries everything needed for
1061 /// one round-trip transaction build: main parents + heights, per-store
1062 /// subtree parents + heights, settings tips, and the merged `_settings`
1063 /// CRDT state this entry is authored against.
1064 #[cfg(all(unix, feature = "service"))]
1065 pub async fn transaction_context(
1066 &self,
1067 stores: &[String],
1068 scope: crate::service::protocol::ReadScope,
1069 ) -> Result<crate::service::protocol::TransactionContext> {
1070 use crate::service::protocol::{ReadScope, TransactionContext};
1071
1072 // -- scope-sensitive main tips --------------------------------
1073 let db_for_tips = Database {
1074 allow_unverified: matches!(scope, ReadScope::AllowUnverified),
1075 ..self.clone()
1076 };
1077 let main_snapshot = db_for_tips.snapshot().await?;
1078 let main_tips = main_snapshot.tips();
1079
1080 // -- main parents: (tip, height) ------------------------------
1081 let mut main_parents = Vec::with_capacity(main_tips.len());
1082 for tip in main_tips {
1083 let entry = self.ops().get(tip).await?;
1084 main_parents.push((tip.clone(), entry.height()));
1085 }
1086
1087 // -- per-store subtree parents: (tip, subtree_height) ---------
1088 let mut subtree_parents = std::collections::BTreeMap::new();
1089 for store in stores {
1090 let child_snap = self
1091 .ops()
1092 .store_snapshot_at(self.root_id(), store, &main_snapshot)
1093 .await?;
1094 let mut pairs = Vec::with_capacity(child_snap.len());
1095 for tip in child_snap.tips() {
1096 let entry = self.ops().get(tip).await?;
1097 let height = entry.subtree_height(store).unwrap_or(0);
1098 pairs.push((tip.clone(), height));
1099 }
1100 subtree_parents.insert(store.clone(), pairs);
1101 }
1102
1103 // -- settings tips (pinned in entry metadata) -----------------
1104 let settings_tips = self
1105 .ops()
1106 .store_snapshot_at(self.root_id(), SETTINGS, &main_snapshot)
1107 .await?
1108 .into_tips();
1109
1110 // -- merged _settings state as serde_json::Value --------------
1111 let txn = Transaction::new_at(self, &main_snapshot).await?;
1112 let settings_doc: Doc = txn.get_full_state(SETTINGS).await?;
1113 let settings_value = serde_json::to_value(&settings_doc)?;
1114
1115 Ok(TransactionContext {
1116 main_parents,
1117 subtree_parents,
1118 settings_tips,
1119 settings_value,
1120 })
1121 }
1122
1123 /// Server-materialized merged state of an **unencrypted** store, as a
1124 /// `serde_json::Value` against the database's Verified frontier.
1125 ///
1126 /// Creates an ephemeral transaction, deserializes every entry's
1127 /// store data as [`Doc`], and merges them via Doc's LWW merge —
1128 /// the same merge `Store<T>` would perform client-side. All current
1129 /// store types (DocStore, Table, Settings) serialize their data as
1130 /// JSON, so `Doc`-typed deserialization works universally.
1131 ///
1132 /// # Encrypted stores
1133 ///
1134 /// Encrypted stores cannot be materialized this way (the ephemeral
1135 /// transaction has no encryptor, so `serde_json::from_slice::<Doc>`
1136 /// would fail on ciphertext). The caller must use
1137 /// [`get_store_entries`](Self::get_store_entries) for encrypted
1138 /// stores and decrypt+merge client-side.
1139 pub async fn get_store_state(&self, store: &str) -> Result<serde_json::Value> {
1140 let txn = self.new_transaction().await?;
1141 let state: Doc = txn.get_full_state(store).await?;
1142 Ok(serde_json::to_value(&state)?)
1143 }
1144
1145 /// Ordered (by subtree height), verifiable, opaque store entries
1146 /// reachable from `tips` within `scope`.
1147 ///
1148 /// This is the **universal** primitive — works for encrypted and
1149 /// unencrypted stores alike because it returns raw [`Entry`] records
1150 /// with opaque [`RawData`](crate::entry::RawData); no deserialization
1151 /// or merge runs server-side. The per-subtree-height ordering
1152 /// (ascending, then by ID for tiebreaking) is exactly the canonical
1153 /// CRDT replay order produced by
1154 /// [`sort_entries_by_subtree_height`](crate::backend::database::in_memory::cache::sort_entries_by_subtree_height).
1155 ///
1156 /// When `scope` is [`ReadScope::Verified`] and `tips` are the
1157 /// Verified-frontier tips from [`snapshot`](Self::snapshot),
1158 /// every returned entry is guaranteed `Verified` (the frontier is
1159 /// ancestor-closed). For [`ReadScope::AllowUnverified`], entries
1160 /// reachable from unverified tips are included.
1161 #[cfg(all(unix, feature = "service"))]
1162 pub async fn get_store_entries(
1163 &self,
1164 store: &str,
1165 tips: &[ID],
1166 _scope: crate::service::protocol::ReadScope,
1167 ) -> Result<Vec<Entry>> {
1168 let snapshot = Snapshot::from(tips.to_vec());
1169 self.ops().store_at(self.root_id(), store, &snapshot).await
1170 }
1171
1172 /// Execute a closure within a transaction and commit the result.
1173 ///
1174 /// This is a convenience wrapper for the common pattern of creating a transaction,
1175 /// performing store operations, and committing. The transaction is committed after
1176 /// the closure returns `Ok`. If the closure returns `Err`, the transaction is
1177 /// dropped without committing.
1178 ///
1179 /// For read-only access, use [`get_store_viewer`](Self::get_store_viewer) instead.
1180 ///
1181 /// # Arguments
1182 /// * `f` - A closure that receives the [`Transaction`] and performs store operations.
1183 /// The closure should return `Ok(R)` on success.
1184 ///
1185 /// # Returns
1186 /// On success, returns the value produced by the closure after committing.
1187 /// The commit ID is not returned; use [`new_transaction`](Self::new_transaction)
1188 /// directly if you need it.
1189 ///
1190 /// # Errors
1191 /// Returns an error if transaction creation, the closure, or commit fails.
1192 /// If the closure fails, the transaction is not committed.
1193 ///
1194 /// # Example
1195 /// ```rust,no_run
1196 /// # use eidetica::*;
1197 /// # use eidetica::store::Table;
1198 /// # use serde::{Serialize, Deserialize};
1199 /// # #[derive(Clone, Serialize, Deserialize)]
1200 /// # struct Todo { title: String }
1201 /// # async fn example(db: Database) -> Result<()> {
1202 /// // Insert a record and get its generated key
1203 /// let key = db.with_transaction(|txn| async move {
1204 /// let store = txn.get_store::<Table<Todo>>("todos").await?;
1205 /// store.insert(Todo { title: "Buy milk".into() }).await
1206 /// }).await?;
1207 ///
1208 /// // Multiple operations in one atomic transaction
1209 /// db.with_transaction(|txn| async move {
1210 /// let store = txn.get_store::<Table<Todo>>("todos").await?;
1211 /// store.insert(Todo { title: "First".into() }).await?;
1212 /// store.insert(Todo { title: "Second".into() }).await?;
1213 /// Ok(())
1214 /// }).await?;
1215 /// # Ok(())
1216 /// # }
1217 /// ```
1218 pub async fn with_transaction<F, Fut, R>(&self, f: F) -> Result<R>
1219 where
1220 F: FnOnce(Transaction) -> Fut + Send,
1221 Fut: Future<Output = Result<R>> + Send,
1222 {
1223 let txn = self.new_transaction().await?;
1224 let commit_handle = txn.clone();
1225 let result = f(txn).await?;
1226 commit_handle.commit().await?;
1227 Ok(result)
1228 }
1229
1230 /// Insert an entry into the database without modifying or validating it.
1231 /// Primarily for testing / full control over raw entry storage.
1232 ///
1233 /// The entry is stored `Unverified`: this path runs no validation, so it
1234 /// cannot honestly claim the entry is verified, and the storage API no
1235 /// longer accepts a caller-asserted status. Only the local validation
1236 /// pass promotes entries to `Verified`.
1237 pub async fn insert_raw(&self, entry: Entry) -> Result<ID> {
1238 let instance = self.instance()?;
1239 let id = entry.id();
1240
1241 instance.put(entry).await?;
1242
1243 Ok(id)
1244 }
1245
1246 /// Get a Store type that will handle accesses to the Store
1247 /// This will return a Store initialized to point at the current state of the database.
1248 ///
1249 /// The returned store should NOT be used to modify the database, as it intentionally does not
1250 /// expose the Transaction. Since the Transaction is never committed, it does not have any
1251 /// effect on the database.
1252 pub async fn get_store_viewer<T>(&self, name: impl Into<String>) -> Result<T>
1253 where
1254 T: Store,
1255 {
1256 let txn = self.new_transaction().await?;
1257 T::load(&txn, name.into()).await
1258 }
1259
1260 /// Get the current tips (leaf entries) of the main database branch.
1261 ///
1262 /// Tips represent the latest entries in the database's main history, forming the heads of the DAG.
1263 ///
1264 /// If any raw tip is `Unverified`, an opportunistic [`Self::verify`] pass
1265 /// runs first (entries arrive `Unverified` from sync; this promotes the
1266 /// ones whose pinned `_settings` are now held).
1267 ///
1268 /// The returned tips then depend on the handle's view:
1269 ///
1270 /// - **default** — the **Verified frontier**: the tips of the maximal
1271 /// ancestor-closed, all-`Verified` prefix of the DAG. A still-`Unverified`
1272 /// tip is replaced by its nearest `Verified` ancestors; anything reachable
1273 /// only through an `Unverified` entry is excluded.
1274 /// - **[`allow_unverified`](Self::allow_unverified)** — the raw tips with
1275 /// only `Failed` entries dropped (`Unverified` tips are kept).
1276 ///
1277 /// `Failed` entries are dropped in both cases. While a [`verify`](Self::verify)
1278 /// pass is on the stack the frontier is bypassed (its own reads must see the
1279 /// raw DAG to reconstruct pinned `_settings`); a remote backend returns its
1280 /// raw tips unchanged (the server owns verification).
1281 ///
1282 /// # Returns
1283 /// A `Result` containing the [`Snapshot`] of tip entries or an error.
1284 pub async fn snapshot(&self) -> Result<Snapshot> {
1285 let instance = self.instance()?;
1286
1287 // On a remote instance the server owns verification: `snapshot`
1288 // already returns the server-side Verified frontier (or empty for a
1289 // not-yet-propagated tree, e.g. `Database::create`'s bootstrap
1290 // placeholder root — `EntryNotFound` is mapped to empty to match
1291 // `Backend::snapshot`'s contract). Return it directly: the local
1292 // verification machinery below (status probe, auto-verify,
1293 // `verified_frontier`) is local-only and would fail on a remote
1294 // backend anyway (e.g. `verified_frontier`'s `backend.get_tree(...)`).
1295 //
1296 // Delegate to `self.ops()` rather than calling the connection
1297 // directly: when this handle was built via `Database::create` or
1298 // `Database::open_remote` its `ops` is a `RemoteBackend` carrying the
1299 // *per-database* identity (the new tree's own member key, or the
1300 // caller's chosen identity), which the server's per-tree gate accepts.
1301 // Routing through `conn.session_identity()` here would instead use the
1302 // connection's (caller's) session pubkey, which is not a member of a
1303 // freshly-created tree and gets denied. A handle from `Database::open`
1304 // on a connected instance instead clones the instance's session
1305 // backend, keeping the session-identity semantics for that path.
1306 #[cfg(all(unix, feature = "service"))]
1307 if instance.remote_connection().is_some() {
1308 return match self.ops().snapshot(&self.root).await {
1309 Ok(snap) => Ok(snap),
1310 Err(e) if e.is_not_found() => Ok(Snapshot::EMPTY),
1311 Err(e) => Err(e),
1312 };
1313 }
1314
1315 // Local path: verification-status probing needs the concrete engine.
1316 let backend = instance.require_local_engine()?;
1317 let tips = self.ops().snapshot(&self.root).await?.into_tips();
1318
1319 // Verification status ops are local-only. On a remote backend the
1320 // server owns verification (and stores everything Unverified until
1321 // it verifies); the client returns verified tips unchanged.
1322 if let Some(first) = tips.first()
1323 && backend.get_verification_status(first).await.is_err()
1324 {
1325 return Ok(Snapshot::new(tips));
1326 }
1327
1328 // Access-time opportunistic verification: if any tip is still
1329 // Unverified, attempt to resolve it now. Best-effort — a failure or a
1330 // still-incomplete pin must not block the read. Suppressed while a
1331 // verify pass is already on the stack (its own reads land here).
1332 let tips = if auto_verify_suppressed() {
1333 tips
1334 } else {
1335 let mut any_unverified = false;
1336 for t in &tips {
1337 if backend
1338 .get_verification_status(t)
1339 .await
1340 .unwrap_or(VerificationStatus::Unverified)
1341 == VerificationStatus::Unverified
1342 {
1343 any_unverified = true;
1344 break;
1345 }
1346 }
1347 if any_unverified {
1348 // Boxed: this call closes a snapshot → verify →
1349 // validate_entry → delegation → get_settings → snapshot
1350 // async cycle; the box gives it a finite future size.
1351 let _ = Box::pin(self.verify()).await;
1352 self.ops().snapshot(&self.root).await?.into_tips()
1353 } else {
1354 tips
1355 }
1356 };
1357
1358 // Default view: cut to the Verified frontier. Suppressed while a
1359 // verify pass is on the stack — its reads must see the raw DAG to
1360 // reconstruct pinned `_settings` (the frontier filter itself depends
1361 // on verification status, which is exactly what verify is computing).
1362 if !self.allow_unverified && !auto_verify_suppressed() {
1363 return self.verified_frontier().await.map(Snapshot::new);
1364 }
1365
1366 // `allow_unverified` view: keep Unverified tips, drop only Failed.
1367 let mut visible = Vec::with_capacity(tips.len());
1368 for t in tips {
1369 if backend.get_verification_status(&t).await? != VerificationStatus::Failed {
1370 visible.push(t);
1371 }
1372 }
1373 Ok(Snapshot::new(visible))
1374 }
1375
1376 /// Compute the tips of the maximal all-`Verified` prefix of the DAG.
1377 ///
1378 /// An entry is in the prefix iff it is `Verified` **and** every one of its
1379 /// parents is in the prefix (the prefix is ancestor-closed). The frontier
1380 /// is the set of prefix entries that are not the parent of any other
1381 /// prefix entry — i.e. the tips of the verified subgraph.
1382 ///
1383 /// Returns an empty vector if the root itself is not `Verified` (nothing
1384 /// is observable in the default view until verification reaches the root).
1385 async fn verified_frontier(&self) -> Result<Vec<ID>> {
1386 let instance = self.instance()?;
1387 let backend = instance.require_local_engine()?;
1388
1389 // Topologically sorted (height then ID): every parent precedes its
1390 // children, so a single forward pass can decide prefix membership.
1391 let entries = backend.get_tree(self.root_id()).await?;
1392
1393 let mut in_prefix: std::collections::HashSet<ID> = std::collections::HashSet::new();
1394 let mut covered: std::collections::HashSet<ID> = std::collections::HashSet::new();
1395
1396 for e in &entries {
1397 let id = e.id();
1398 if backend.get_verification_status(&id).await? != VerificationStatus::Verified {
1399 continue;
1400 }
1401 let parents = e.parents().unwrap_or_default();
1402 if parents.iter().all(|p| in_prefix.contains(p)) {
1403 in_prefix.insert(id);
1404 // Every parent now has a verified child, so it is interior to
1405 // the prefix and cannot itself be a frontier tip.
1406 for p in parents {
1407 covered.insert(p);
1408 }
1409 }
1410 }
1411
1412 let frontier: Vec<ID> = entries
1413 .into_iter()
1414 .map(|e| e.id())
1415 .filter(|id| in_prefix.contains(id) && !covered.contains(id))
1416 .collect();
1417 Ok(frontier)
1418 }
1419
1420 /// Get the full `Entry` objects for the current tips of the main database branch.
1421 ///
1422 /// # Returns
1423 /// A `Result` containing a vector of the tip `Entry` objects or an error.
1424 pub async fn get_tip_entries(&self) -> Result<Vec<Entry>> {
1425 let instance = self.instance()?;
1426 let snapshot = self.snapshot().await?;
1427 let mut entries = Vec::new();
1428 for id in snapshot.tips() {
1429 entries.push(instance.get(id).await?);
1430 }
1431 Ok(entries)
1432 }
1433
1434 /// Get a single entry by ID from this database.
1435 ///
1436 /// This is the primary method for retrieving entries after commit operations.
1437 /// It provides safe, high-level access to entry data without exposing backend details.
1438 ///
1439 /// The method verifies that the entry belongs to this database by checking its root ID.
1440 /// If the entry exists but belongs to a different database, an error is returned.
1441 ///
1442 /// # Arguments
1443 /// * `entry_id` - The ID of the entry to retrieve (accepts anything that converts to ID/String)
1444 ///
1445 /// # Returns
1446 /// A `Result` containing the `Entry` or an error if not found or not part of this database
1447 ///
1448 /// # Example
1449 /// ```rust,no_run
1450 /// # use eidetica::*;
1451 /// # use eidetica::Instance;
1452 /// # use eidetica::backend::database::InMemory;
1453 /// # use eidetica::crdt::Doc;
1454 /// # #[tokio::main]
1455 /// # async fn main() -> Result<()> {
1456 /// # let (_instance, mut user) = Instance::create_backend(
1457 /// # Box::new(InMemory::new()),
1458 /// # NewUser::passwordless("test"),
1459 /// # ).await?;
1460 /// # let key_id = user.add_private_key(None).await?;
1461 /// # let tree = user.create_database(Doc::new(), &key_id).await?;
1462 /// # let txn = tree.new_transaction().await?;
1463 /// let entry_id = txn.commit().await?;
1464 /// let entry = tree.get_entry(&entry_id).await?; // Using &ID
1465 /// let entry = tree.get_entry(entry_id.clone()).await?; // Using ID
1466 /// println!("Entry signature: {:?}", entry.sig);
1467 /// # Ok(())
1468 /// # }
1469 /// ```
1470 pub async fn get_entry<I: Into<ID>>(&self, entry_id: I) -> Result<Entry> {
1471 let id = entry_id.into();
1472 // Route through `self.ops()` so handles built via `Database::create`
1473 // or `Database::open_remote` read with the per-DB identity from
1474 // their `RemoteBackend`. Going through `instance.get(id)` would
1475 // use the connection's login pubkey, which on a remote instance is
1476 // denied by the per-tree gate when the login key isn't a member of
1477 // this tree (e.g. user-tree key created via `User::add_private_key`
1478 // and used to author a database that doesn't grant the root key).
1479 let entry = self.ops().get(&id).await?;
1480
1481 // Check if the entry belongs to this database
1482 if !entry.in_tree(&self.root) {
1483 return Err(InstanceError::EntryNotInDatabase {
1484 entry_id: id,
1485 database_id: self.root.clone(),
1486 }
1487 .into());
1488 }
1489
1490 Ok(entry)
1491 }
1492
1493 /// Get multiple entries by ID efficiently.
1494 ///
1495 /// This method retrieves multiple entries more efficiently than multiple `get_entry()` calls
1496 /// by minimizing conversion overhead and pre-allocating the result vector.
1497 ///
1498 /// The method verifies that all entries belong to this database by checking their root IDs.
1499 /// If any entry exists but belongs to a different database, an error is returned.
1500 ///
1501 /// # Parameters
1502 /// * `entry_ids` - An iterable of entry IDs to retrieve
1503 ///
1504 /// # Returns
1505 /// A `Result` containing a vector of `Entry` objects or an error if any entry is not found or not part of this database
1506 ///
1507 /// # Example
1508 /// ```rust,no_run
1509 /// # use eidetica::*;
1510 /// # use eidetica::Instance;
1511 /// # use eidetica::backend::database::InMemory;
1512 /// # use eidetica::crdt::Doc;
1513 /// # #[tokio::main]
1514 /// # async fn main() -> Result<()> {
1515 /// # let (_instance, mut user) = Instance::create_backend(
1516 /// # Box::new(InMemory::new()),
1517 /// # NewUser::passwordless("test"),
1518 /// # ).await?;
1519 /// # let key_id = user.add_private_key(None).await?;
1520 /// # let tree = user.create_database(Doc::new(), &key_id).await?;
1521 /// let entry_ids = vec![ID::from_bytes("id1"), ID::from_bytes("id2")];
1522 /// let entries = tree.get_entries(entry_ids).await?;
1523 /// # Ok(())
1524 /// # }
1525 /// ```
1526 pub async fn get_entries<I, T>(&self, entry_ids: I) -> Result<Vec<Entry>>
1527 where
1528 I: IntoIterator<Item = T>,
1529 T: std::borrow::Borrow<ID>,
1530 {
1531 let ids: Vec<ID> = entry_ids.into_iter().map(|t| t.borrow().clone()).collect();
1532 let instance = self.instance()?;
1533 let mut entries = Vec::with_capacity(ids.len());
1534
1535 for id in ids {
1536 let entry = instance.get(&id).await?;
1537
1538 // Check if the entry belongs to this database
1539 if !entry.in_tree(&self.root) {
1540 return Err(InstanceError::EntryNotInDatabase {
1541 entry_id: id,
1542 database_id: self.root.clone(),
1543 }
1544 .into());
1545 }
1546
1547 entries.push(entry);
1548 }
1549
1550 Ok(entries)
1551 }
1552
1553 // === AUTHENTICATION HELPERS ===
1554
1555 /// Verify an entry's signature and authentication against the database's configuration that was valid at the time of entry creation.
1556 ///
1557 /// This method validates that:
1558 /// 1. The entry belongs to this database
1559 /// 2. The entry is properly signed with a key that was authorized in the database's authentication settings at the time the entry was created
1560 /// 3. The signature is cryptographically valid
1561 ///
1562 /// The method uses the entry's metadata to determine which authentication settings were active when the entry was signed,
1563 /// ensuring that entries remain valid even if keys are later revoked or settings change.
1564 ///
1565 /// # Arguments
1566 /// * `entry_id` - The ID of the entry to verify (accepts anything that converts to ID/String)
1567 ///
1568 /// # Returns
1569 /// A `Result` containing `true` if the entry is valid and properly authenticated, `false` if authentication fails
1570 ///
1571 /// # Errors
1572 /// Returns an error if:
1573 /// - The entry is not found
1574 /// - The entry does not belong to this database
1575 /// - The entry's metadata cannot be parsed
1576 /// - The historical authentication settings cannot be retrieved
1577 pub async fn verify_entry_signature<I: Into<ID>>(&self, entry_id: I) -> Result<bool> {
1578 let entry = self.get_entry(entry_id).await?;
1579
1580 // Validate against the `_settings` the entry pins, not current
1581 // settings — so a later key revocation cannot retroactively
1582 // invalidate (or validate) historical entries.
1583 match self.get_historical_settings_for_entry(&entry).await? {
1584 // We do not hold the pinned `_settings` set, so we cannot make a
1585 // verification decision: report not-verified rather than guess.
1586 PinnedSettings::Incomplete => Ok(false),
1587 PinnedSettings::Complete(auth_settings) => {
1588 let instance = self.instance()?;
1589 let mut validator = AuthValidator::new();
1590 validator
1591 .validate_entry(&entry, &auth_settings, Some(&instance))
1592 .await
1593 }
1594 }
1595 }
1596
1597 /// Get the permission level for this database's configured signing key.
1598 ///
1599 /// Returns the effective permission for the key that was configured when opening
1600 /// or creating this database. This uses the already-resolved identity stored in
1601 /// the database's `DatabaseKey`.
1602 ///
1603 /// # Returns
1604 /// The effective Permission for the configured signing key.
1605 ///
1606 /// # Errors
1607 /// Returns an error if:
1608 /// - No signing key is configured (database opened without authentication)
1609 /// - The database settings cannot be retrieved
1610 /// - The key is no longer valid in the current auth settings
1611 ///
1612 /// # Example
1613 /// ```rust,no_run
1614 /// # use eidetica::*;
1615 /// # use eidetica::crdt::Doc;
1616 /// # use eidetica::backend::database::InMemory;
1617 /// # use eidetica::auth::crypto::generate_keypair;
1618 /// # #[tokio::main]
1619 /// # async fn main() -> Result<()> {
1620 /// # let instance = Instance::open_backend(Box::new(InMemory::new())).await?;
1621 /// # let (signing_key, _public_key) = generate_keypair();
1622 /// # let database = Database::create(&instance, signing_key, Doc::new()).await?;
1623 /// // Check if the current key has Admin permission
1624 /// let permission = database.current_permission().await?;
1625 /// if permission.can_admin() {
1626 /// println!("Current key has Admin permission!");
1627 /// }
1628 /// # Ok(())
1629 /// # }
1630 /// ```
1631 pub async fn current_permission(&self) -> Result<Permission> {
1632 let key = self
1633 .key
1634 .as_ref()
1635 .ok_or(AuthError::InvalidAuthConfiguration {
1636 reason: "No signing key configured for this database".to_string(),
1637 })?;
1638 self.validate_key(key).await
1639 }
1640
1641 /// Reconstruct the `_settings` auth config an entry's signature is pinned
1642 /// to, from the `settings_snapshot` recorded in its signed metadata.
1643 ///
1644 /// Validation must run against the settings the entry pinned — not the
1645 /// current settings — so granting authority later cannot retroactively
1646 /// invalidate an entry that pinned less, and (once revocation lands)
1647 /// removals are handled on a separate, current-settings path.
1648 ///
1649 /// Returns [`PinnedSettings::Incomplete`] when this node does not hold the
1650 /// full pinned `_settings` ancestor set; the caller must then leave the
1651 /// entry `Unverified` rather than guess against whatever it does hold.
1652 async fn get_historical_settings_for_entry(&self, entry: &Entry) -> Result<PinnedSettings> {
1653 let instance = self.instance()?;
1654 let backend = instance.backend();
1655
1656 // The pin: `_settings` snapshot recorded in the entry's signed metadata.
1657 let settings_tips: Vec<ID> = match entry.metadata() {
1658 Some(raw) => match serde_json::from_slice::<crate::transaction::EntryMetadata>(raw) {
1659 Ok(md) => md.settings_snapshot.into_tips(),
1660 // Unparsable metadata ⇒ we cannot establish the pin.
1661 Err(_) => return Ok(PinnedSettings::Incomplete),
1662 },
1663 None => Vec::new(),
1664 };
1665
1666 // Resolve the effective `_settings` tips to validate against.
1667 let effective_tips: Vec<ID> = if settings_tips.is_empty() {
1668 if entry.in_subtree(SETTINGS) {
1669 // Genesis / bootstrap: no prior `_settings` exists, so the
1670 // entry is self-authorising — validate against the auth it
1671 // itself establishes (TOFU), mirroring how the transaction
1672 // validates initial database creation. Seeding the
1673 // reconstruction with the entry itself folds in its own
1674 // `_settings` contribution.
1675 vec![entry.id()]
1676 } else {
1677 // No auth context at all (no settings ever configured) —
1678 // mirrors the transaction path's "auth never configured" case.
1679 return Ok(PinnedSettings::Complete(AuthSettings::new()));
1680 }
1681 } else {
1682 settings_tips
1683 };
1684
1685 // Completeness: every pinned tip and its full `_settings` ancestor
1686 // closure must be present locally. `store_at` silently
1687 // skips absent ancestors, so an explicit walk is required — a missing
1688 // ancestor would otherwise yield a wrong (partial) auth config.
1689 let mut stack: Vec<ID> = effective_tips.clone();
1690 let mut seen: std::collections::HashSet<ID> = std::collections::HashSet::new();
1691 while let Some(id) = stack.pop() {
1692 if !seen.insert(id.clone()) {
1693 continue;
1694 }
1695 let Ok(e) = backend.get(&id).await else {
1696 return Ok(PinnedSettings::Incomplete);
1697 };
1698 // Walk both the `_settings` subtree DAG and the main parents that
1699 // carry it, so the closure can't be short-circuited.
1700 for p in e.subtree_parents(SETTINGS).unwrap_or_default() {
1701 stack.push(p);
1702 }
1703 for p in e.parents().unwrap_or_default() {
1704 stack.push(p);
1705 }
1706 }
1707
1708 // Reconstruct the merged `_settings` Doc as of the pinned tips.
1709 // Entries come back root-first; `_settings` is a system subtree and
1710 // is never encrypted, so deserialize directly.
1711 let effective_snapshot = Snapshot::from(effective_tips.clone());
1712 let entries = backend
1713 .store_at(self.root_id(), SETTINGS, &effective_snapshot)
1714 .await?;
1715 let mut settings_doc = Doc::default();
1716 for e in &entries {
1717 if let Ok(data) = e.data(SETTINGS) {
1718 let part: Doc = serde_json::from_slice(data)?;
1719 settings_doc = settings_doc.merge(&part)?;
1720 }
1721 }
1722
1723 let auth_settings = match settings_doc.get("auth") {
1724 Some(crate::crdt::doc::Value::Doc(auth_doc)) => auth_doc.clone().into(),
1725 _ => AuthSettings::new(),
1726 };
1727 Ok(PinnedSettings::Complete(auth_settings))
1728 }
1729
1730 /// Return the IDs of entries reachable from `post_tips` but not from
1731 /// `previous_tips`, in topological order (parents before children).
1732 ///
1733 /// This is the canonical way to enumerate the entries added between two
1734 /// cursors of this database — typically the cursors supplied by a
1735 /// [`WriteEvent`](crate::instance::WriteEvent)'s
1736 /// [`previous_tips()`](crate::instance::WriteEvent::previous_tips) and
1737 /// [`post_tips()`](crate::instance::WriteEvent::post_tips). Callbacks that
1738 /// only care that *something* changed can ignore this; callbacks that
1739 /// need to enumerate or fetch entry contents call this to expand the
1740 /// cursor advance into a concrete set of IDs.
1741 ///
1742 /// The walk is bounded by the cursor diff — cost is proportional to the
1743 /// number of entries *added* between the two cursors, not to the full
1744 /// DAG.
1745 ///
1746 /// # Behavior
1747 ///
1748 /// - If `previous_tips == post_tips`, returns an empty vector.
1749 /// - If `previous_tips` is empty, returns every ID reachable from
1750 /// `post_tips` (i.e. the full ancestor closure of those tips).
1751 /// - If `post_tips` references an entry that does not exist locally,
1752 /// returns an `EntryNotFound` error.
1753 /// - Verification status is **not** filtered, and event-driven callers
1754 /// are not exempt. `WriteEvent` is *triggered* only by Verified
1755 /// writes, but its cursors are raw DAG frontiers, so an `Unverified`
1756 /// or `Failed` entry sitting as a tip is inside the bracket and is
1757 /// enumerated here. Every caller that fetches or ingests these IDs
1758 /// should filter via `Backend::get_verification_status`.
1759 ///
1760 /// # Errors
1761 ///
1762 /// - `EntryNotFound` if any entry reachable from `post_tips` *above* the
1763 /// `previous_tips` frontier is missing locally. Entries at or below the
1764 /// frontier are never fetched, so a partial history there is tolerated —
1765 /// the conservative direction (we may over-report rather than
1766 /// under-report).
1767 pub async fn ids_added(
1768 &self,
1769 previous_tips: &Snapshot,
1770 post_tips: &Snapshot,
1771 ) -> Result<Vec<ID>> {
1772 use std::collections::{HashMap, HashSet, VecDeque};
1773
1774 // Set-equality on the canonical tip-sets: order- and
1775 // duplication-insensitive, so a cursor that hasn't advanced
1776 // short-circuits regardless of how its tips were ordered.
1777 if previous_tips == post_tips {
1778 return Ok(Vec::new());
1779 }
1780
1781 // `previous_tips` is the stop-frontier: everything reachable from it has
1782 // already been observed. Walk *backward from post_tips* and halt at the
1783 // first entry in that frontier, so cost is bounded by the cursor diff —
1784 // the entries added between the two cursors — not the full DAG. (Same
1785 // shape as `sync::utils::collect_ancestors_to_send`.)
1786 //
1787 // The callback cursor model always advances `previous_tips` as a
1788 // complete frontier (a cut), so halting at its members is exact. A
1789 // partial or stale `previous_tips` can only over-report — never
1790 // under-report, since any genuinely-new entry is reached before the walk
1791 // meets the frontier — which is the conservative direction.
1792 let boundary: HashSet<ID> = previous_tips.tips().iter().cloned().collect();
1793
1794 // Routes through `self.ops()` so handles built via
1795 // `Database::create` / `Database::open_remote` walk over the wire with
1796 // the per-DB identity. On a local instance this is a clone of the local
1797 // backend; on a connected instance each `get(id)` is a permission-checked
1798 // round-trip to the daemon, which is the security gate the cursor-only
1799 // push model relies on.
1800 let ops = self.ops();
1801
1802 // Guard against a **backward** bracket — `previous_tips` ahead of
1803 // `post_tips`. The boundary is then made of *descendants* of the
1804 // walk's starting point, so it is never reached and the walk descends
1805 // to the root, reporting the whole history as "added". On a connected
1806 // instance every step is a permission-checked round-trip, so the cost
1807 // is O(history) fetches rather than a merely-wrong answer.
1808 //
1809 // Detected via entry height, which increases monotonically from parent
1810 // to child: every ancestor of a `post_tips` entry has height at most
1811 // `max(post heights)`, so if every boundary tip sits strictly above
1812 // that, none of them can ever be reached by the walk. Bounded by the
1813 // tip counts (both small), and works identically on a local and a
1814 // connected instance — unlike the backend-level reachability
1815 // primitive, which a remote handle cannot call.
1816 //
1817 // Nothing was *added* moving backward, so the answer is empty.
1818 //
1819 // Best-effort: if any tip cannot be resolved we skip the guard and
1820 // fall through to the normal walk rather than failing the caller. This
1821 // only detects the strictly-backward case; forked/incomparable cursors
1822 // still over-report, which is the documented conservative direction.
1823 if !boundary.is_empty() {
1824 let mut max_post: Option<u64> = None;
1825 let mut min_prev: Option<u64> = None;
1826 let mut resolved = true;
1827 for id in post_tips.tips() {
1828 match ops.get(id).await {
1829 Ok(e) => max_post = Some(max_post.map_or(e.height(), |h| h.max(e.height()))),
1830 Err(_) => {
1831 resolved = false;
1832 break;
1833 }
1834 }
1835 }
1836 if resolved {
1837 for id in &boundary {
1838 match ops.get(id).await {
1839 Ok(e) => {
1840 min_prev = Some(min_prev.map_or(e.height(), |h| h.min(e.height())))
1841 }
1842 Err(_) => {
1843 resolved = false;
1844 break;
1845 }
1846 }
1847 }
1848 }
1849 if resolved
1850 && let (Some(post_h), Some(prev_h)) = (max_post, min_prev)
1851 && prev_h > post_h
1852 {
1853 return Ok(Vec::new());
1854 }
1855 }
1856
1857 // Walk parents from `post_tips`, collecting every ID until the boundary.
1858 // A non-boundary entry that's missing locally is a hard error — the
1859 // cursor references unknown state; boundary entries are never fetched.
1860 let mut added: HashMap<ID, Entry> = HashMap::new();
1861 let mut visited: HashSet<ID> = HashSet::new();
1862 let mut queue: VecDeque<ID> = post_tips.tips().iter().cloned().collect();
1863 while let Some(id) = queue.pop_front() {
1864 if boundary.contains(&id) || !visited.insert(id.clone()) {
1865 continue;
1866 }
1867 let entry = ops.get(&id).await?;
1868 for p in entry.parents().unwrap_or_default() {
1869 queue.push_back(p);
1870 }
1871 added.insert(id, entry);
1872 }
1873
1874 // 3. Topo sort the added set (Kahn's, scoped). Parents outside
1875 // the set are pre-observed boundary entries and don't count
1876 // toward in-degree.
1877 let mut in_degree: HashMap<ID, usize> = HashMap::with_capacity(added.len());
1878 let mut children: HashMap<ID, Vec<ID>> = HashMap::new();
1879 for (id, entry) in &added {
1880 let mut d = 0usize;
1881 for p in entry.parents().unwrap_or_default() {
1882 if added.contains_key(&p) {
1883 d += 1;
1884 children.entry(p).or_default().push(id.clone());
1885 }
1886 }
1887 in_degree.insert(id.clone(), d);
1888 }
1889 let mut topo_queue: VecDeque<ID> = in_degree
1890 .iter()
1891 .filter(|&(_, &d)| d == 0)
1892 .map(|(id, _)| id.clone())
1893 .collect();
1894 let mut order: Vec<ID> = Vec::with_capacity(added.len());
1895 while let Some(id) = topo_queue.pop_front() {
1896 if let Some(kids) = children.get(&id) {
1897 for kid in kids {
1898 let d = in_degree.get_mut(kid).expect("in_degree entry exists");
1899 *d -= 1;
1900 if *d == 0 {
1901 topo_queue.push_back(kid.clone());
1902 }
1903 }
1904 }
1905 order.push(id);
1906 }
1907
1908 Ok(order)
1909 }
1910
1911 /// Attempt to verify every `Unverified` entry in this database.
1912 ///
1913 /// For each `Unverified` entry, reconstruct the `_settings` it pins
1914 /// (see [`Self::get_historical_settings_for_entry`]) and validate its
1915 /// signature + permissions against that:
1916 ///
1917 /// - an ancestor is `Failed` → this entry is `Failed` too (quarantine
1918 /// propagates down the branch);
1919 /// - an ancestor is still `Unverified`, or not held locally yet (partial
1920 /// sync) → left `Unverified` (retried once the ancestor verifies / the
1921 /// missing entry arrives);
1922 /// - pinned `_settings` not fully held locally → left `Unverified`
1923 /// (a later pass retries once the set syncs in);
1924 /// - signature + permissions valid → promoted to `Verified`;
1925 /// - definitively invalid → marked `Failed` (dropped from reads).
1926 ///
1927 /// Verification is **prefix-closed**: an entry is `Verified` only if its
1928 /// entire ancestor history is `Verified`. It is therefore impossible for a
1929 /// tip to be `Verified` while one of its ancestors is not, which is what
1930 /// makes the Verified set ancestor-closed (see [`Self::allow_unverified`]).
1931 ///
1932 /// Already-`Verified` entries are never demoted here; that is a separate,
1933 /// not-yet-built path. Local-only — verification is a per-node decision
1934 /// and is never delegated to a peer.
1935 pub async fn verify(&self) -> Result<VerifyReport> {
1936 use std::collections::{HashMap, HashSet, VecDeque};
1937
1938 let instance = self.instance()?;
1939
1940 // Hold the per-tree write lock for the whole pass + fire so the
1941 // promoted-batch event serialises against concurrent
1942 // `put_entry` fires on this tree. Callers must not hold this
1943 // lock when calling `verify`. (The two main callers,
1944 // `put_remote_entries` and the service `SubmitSignedEntry`
1945 // handler, release their own lock before calling verify.)
1946 let lock = instance.tree_lock(self.root_id());
1947 let _guard = lock.lock().await;
1948
1949 // Suppress the access-time auto-verify hook for the whole pass:
1950 // validation reads the database (delegation → settings → tips) and
1951 // must not recurse back into verification.
1952 let (report, any_promoted, fire_tips) = IN_VERIFY
1953 .scope(true, async move {
1954 let backend = instance.require_local_engine()?;
1955
1956 // 1. Outer boundary of the Unverified region: raw DAG
1957 // tips. (Failed/Verified tips both terminate the walk
1958 // below; no pre-filter needed.)
1959 let raw_tips = backend.snapshot(self.root_id()).await?;
1960
1961 // 2. Walk parents from those tips, collecting the
1962 // Unverified region. `Verified` and `Failed` entries
1963 // are the inner boundary — by prefix-closure their
1964 // ancestors are already settled. Demotions cascade
1965 // through `Instance::demote_to_unverified`, so a
1966 // `Verified` entry hiding an `Unverified` descendant
1967 // cannot occur and we never need to descend past
1968 // a Verified entry.
1969 let mut unverified: HashMap<ID, Entry> = HashMap::new();
1970 let mut visited: HashSet<ID> = HashSet::new();
1971 let mut queue: VecDeque<ID> = raw_tips.tips().iter().cloned().collect();
1972 while let Some(id) = queue.pop_front() {
1973 if !visited.insert(id.clone()) {
1974 continue;
1975 }
1976 let status = match backend.get_verification_status(&id).await {
1977 Ok(s) => s,
1978 // Tree-internal reference we don't hold yet
1979 // (partial sync): skip; descendants stay
1980 // Unverified and a later pass picks them up
1981 // once the parent arrives.
1982 Err(e) if e.is_not_found() => continue,
1983 Err(e) => return Err(e),
1984 };
1985 match status {
1986 VerificationStatus::Verified | VerificationStatus::Failed => continue,
1987 VerificationStatus::Unverified => {
1988 let entry = backend.get(&id).await?;
1989 for p in entry.parents().unwrap_or_default() {
1990 queue.push_back(p);
1991 }
1992 unverified.insert(id, entry);
1993 }
1994 }
1995 }
1996
1997 // 3. Topo-sort the Unverified subgraph (Kahn's,
1998 // scoped). Parents outside the set are pre-settled
1999 // boundary entries and don't contribute to
2000 // in-degree.
2001 let mut in_degree: HashMap<ID, usize> = HashMap::with_capacity(unverified.len());
2002 let mut children: HashMap<ID, Vec<ID>> = HashMap::new();
2003 for (id, entry) in &unverified {
2004 let mut d = 0usize;
2005 for p in entry.parents().unwrap_or_default() {
2006 if unverified.contains_key(&p) {
2007 d += 1;
2008 children.entry(p).or_default().push(id.clone());
2009 }
2010 }
2011 in_degree.insert(id.clone(), d);
2012 }
2013 let mut topo_queue: VecDeque<ID> = in_degree
2014 .iter()
2015 .filter(|&(_, &d)| d == 0)
2016 .map(|(id, _)| id.clone())
2017 .collect();
2018 let mut order: Vec<ID> = Vec::with_capacity(unverified.len());
2019 while let Some(id) = topo_queue.pop_front() {
2020 order.push(id.clone());
2021 if let Some(kids) = children.get(&id) {
2022 for kid in kids {
2023 let d = in_degree.get_mut(kid).expect("in_degree entry exists");
2024 *d -= 1;
2025 if *d == 0 {
2026 topo_queue.push_back(kid.clone());
2027 }
2028 }
2029 }
2030 }
2031
2032 // 4. Process parents-before-children. Same per-entry
2033 // logic as the legacy `get_tree`-walk verify, just
2034 // bounded to the Unverified region.
2035 let mut report = VerifyReport::default();
2036 let mut any_promoted = false;
2037 for id in &order {
2038 let entry = unverified.get(id).expect("topo id is in set");
2039 let parents = entry.parents().unwrap_or_default();
2040 let mut compromised = false;
2041 let mut blocked = false;
2042 for p in &parents {
2043 match backend.get_verification_status(p).await {
2044 Ok(VerificationStatus::Verified) => {}
2045 Ok(VerificationStatus::Failed) => compromised = true,
2046 Ok(VerificationStatus::Unverified) => blocked = true,
2047 Err(e) if e.is_not_found() => blocked = true,
2048 Err(e) => return Err(e),
2049 }
2050 }
2051 if compromised {
2052 backend
2053 .update_verification_status(id, VerificationStatus::Failed)
2054 .await?;
2055 report.failed += 1;
2056 continue;
2057 }
2058 if blocked {
2059 report.still_unverified += 1;
2060 continue;
2061 }
2062
2063 match self.get_historical_settings_for_entry(entry).await? {
2064 PinnedSettings::Incomplete => report.still_unverified += 1,
2065 PinnedSettings::Complete(auth_settings) => {
2066 let mut validator = AuthValidator::new();
2067 let valid = validator
2068 .validate_entry(entry, &auth_settings, Some(&instance))
2069 .await
2070 .unwrap_or(false);
2071 if valid {
2072 backend
2073 .update_verification_status(id, VerificationStatus::Verified)
2074 .await?;
2075 report.verified += 1;
2076 any_promoted = true;
2077 } else {
2078 backend
2079 .update_verification_status(id, VerificationStatus::Failed)
2080 .await?;
2081 report.failed += 1;
2082 }
2083 }
2084 }
2085 }
2086
2087 Ok::<_, crate::Error>((report, any_promoted, raw_tips))
2088 })
2089 .await?;
2090
2091 // Fire one batched `Verified` event for the promotion, if any.
2092 // Outside the IN_VERIFY scope so the callbacks' own reads aren't
2093 // suppressed. Still under the per-tree write lock — the lock is
2094 // dropped when this method returns.
2095 //
2096 // Pre- and post-pass tips are *the same* raw backend tips: verify
2097 // only mutates verification statuses, never the DAG structure.
2098 // For per-callback cursor advancement that means the cursor moves
2099 // to `pre_tips` (which includes the just-promoted entries), which
2100 // is the correct frontier — subscribers see the same
2101 // `previous_tips` they'd see for any subsequent fire until
2102 // something else writes to the tree.
2103 let joins = if any_promoted {
2104 let instance = self.instance()?;
2105 Some(
2106 instance
2107 .spawn_write_callbacks(
2108 self.root_id(),
2109 &fire_tips,
2110 &fire_tips,
2111 WriteSource::Remote,
2112 )
2113 .await,
2114 )
2115 } else {
2116 None
2117 };
2118
2119 // Release the per-tree lock before awaiting the user callbacks. The
2120 // cursor advances were already committed under the lock by
2121 // `spawn_write_callbacks` (which is what preserves event ordering);
2122 // the closures must run lock-free, because a callback that reads
2123 // tips can trip the access-time auto-verify hook → `verify()` →
2124 // `tree_lock` and would deadlock against a still-held `_guard`.
2125 drop(_guard);
2126
2127 if let Some(mut joins) = joins {
2128 while joins.join_next().await.is_some() {}
2129 }
2130
2131 Ok(report)
2132 }
2133
2134 // === DATABASE QUERIES ===
2135
2136 /// Get all entries in this database.
2137 ///
2138 /// ⚠️ **Warning**: This method loads all entries into memory. Use with caution on large databases.
2139 /// Consider using `snapshot()` or `get_tip_entries()` for more efficient access patterns.
2140 ///
2141 /// # Returns
2142 /// A `Result` containing a vector of all `Entry` objects in the database
2143 pub async fn get_all_entries(&self) -> Result<Vec<Entry>> {
2144 let instance = self.instance()?;
2145 instance.require_local_engine()?.get_tree(&self.root).await
2146 }
2147}