Skip to main content

hashtree_cli/storage/
retention.rs

1use anyhow::Result;
2use futures::executor::block_on as sync_block_on;
3use hashtree_core::store::Store;
4use hashtree_core::{to_hex, types::Hash, HashTree, HashTreeConfig, LinkType};
5use serde::de::{self, IgnoredAny, MapAccess, SeqAccess, Visitor};
6use serde::{Deserialize, Serialize};
7use std::collections::HashSet;
8use std::path::{Path, PathBuf};
9use std::time::{SystemTime, UNIX_EPOCH};
10
11use super::{BlobMetadata, HashtreeStore, PRIORITY_FOLLOWED, PRIORITY_OWN};
12
13/// Metadata for a synced tree (for eviction tracking)
14#[derive(Debug, Clone, Serialize)]
15pub struct TreeMeta {
16    /// Pubkey of tree owner
17    pub owner: String,
18    /// Tree name if known (from nostr key like "npub.../name")
19    pub name: Option<String>,
20    /// Unix timestamp when this tree was synced
21    pub synced_at: u64,
22    /// Total size of all blobs in this tree
23    pub total_size: u64,
24    /// Eviction priority: 255=own/pinned, 128=followed, 64=other
25    pub priority: u8,
26}
27
28impl<'de> Deserialize<'de> for TreeMeta {
29    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
30    where
31        D: serde::Deserializer<'de>,
32    {
33        const FIELDS: &[&str] = &[
34            "owner",
35            "name",
36            "synced_at",
37            "last_accessed_at",
38            "total_size",
39            "priority",
40        ];
41
42        struct TreeMetaVisitor;
43
44        impl<'de> Visitor<'de> for TreeMetaVisitor {
45            type Value = TreeMeta;
46
47            fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
48                formatter.write_str("TreeMeta as current or legacy metadata")
49            }
50
51            fn visit_seq<A>(self, mut seq: A) -> std::result::Result<Self::Value, A::Error>
52            where
53                A: SeqAccess<'de>,
54            {
55                let has_accidental_access_field = matches!(seq.size_hint(), Some(6));
56                let owner = seq
57                    .next_element()?
58                    .ok_or_else(|| de::Error::invalid_length(0, &self))?;
59                let name = seq
60                    .next_element()?
61                    .ok_or_else(|| de::Error::invalid_length(1, &self))?;
62                let synced_at = seq
63                    .next_element()?
64                    .ok_or_else(|| de::Error::invalid_length(2, &self))?;
65
66                if has_accidental_access_field {
67                    let _: IgnoredAny = seq
68                        .next_element()?
69                        .ok_or_else(|| de::Error::invalid_length(3, &self))?;
70                }
71
72                let total_size = seq
73                    .next_element()?
74                    .ok_or_else(|| de::Error::invalid_length(3, &self))?;
75                let priority = seq
76                    .next_element()?
77                    .ok_or_else(|| de::Error::invalid_length(4, &self))?;
78
79                Ok(TreeMeta {
80                    owner,
81                    name,
82                    synced_at,
83                    total_size,
84                    priority,
85                })
86            }
87
88            fn visit_map<A>(self, mut map: A) -> std::result::Result<Self::Value, A::Error>
89            where
90                A: MapAccess<'de>,
91            {
92                let mut owner = None;
93                let mut name = None;
94                let mut synced_at = None;
95                let mut total_size = None;
96                let mut priority = None;
97
98                while let Some(key) = map.next_key::<String>()? {
99                    match key.as_str() {
100                        "owner" => owner = Some(map.next_value()?),
101                        "name" => name = Some(map.next_value()?),
102                        "synced_at" => synced_at = Some(map.next_value()?),
103                        "last_accessed_at" => {
104                            let _: IgnoredAny = map.next_value()?;
105                        }
106                        "total_size" => total_size = Some(map.next_value()?),
107                        "priority" => priority = Some(map.next_value()?),
108                        _ => {
109                            let _: IgnoredAny = map.next_value()?;
110                        }
111                    }
112                }
113
114                Ok(TreeMeta {
115                    owner: owner.ok_or_else(|| de::Error::missing_field("owner"))?,
116                    name: name.unwrap_or(None),
117                    synced_at: synced_at.ok_or_else(|| de::Error::missing_field("synced_at"))?,
118                    total_size: total_size.ok_or_else(|| de::Error::missing_field("total_size"))?,
119                    priority: priority.ok_or_else(|| de::Error::missing_field("priority"))?,
120                })
121            }
122        }
123
124        deserializer.deserialize_struct("TreeMeta", FIELDS, TreeMetaVisitor)
125    }
126}
127
128#[derive(Debug)]
129pub struct StorageStats {
130    pub total_dags: usize,
131    pub pinned_dags: usize,
132    pub total_bytes: u64,
133}
134
135/// Storage usage broken down by priority tier
136#[derive(Debug, Clone)]
137pub struct StorageByPriority {
138    /// Own/pinned trees (priority 255)
139    pub own: u64,
140    /// Followed users' trees (priority 128)
141    pub followed: u64,
142    /// Other trees (priority 64)
143    pub other: u64,
144}
145
146#[derive(Debug, Clone)]
147pub struct PinnedItem {
148    pub cid: String,
149    pub name: String,
150    pub is_directory: bool,
151    pub size_bytes: u64,
152}
153
154#[derive(Debug, Clone)]
155pub struct OwnedBlobStats {
156    pub owner: [u8; 32],
157    pub count: usize,
158    pub total_bytes: u64,
159}
160
161fn pinned_item_name(hash: &Hash, meta: Option<&TreeMeta>) -> String {
162    let Some(meta) = meta else {
163        return to_hex(hash);
164    };
165
166    match (meta.owner.as_str(), meta.name.as_deref()) {
167        ("pinned", Some(name)) => name.to_string(),
168        ("", Some(name)) => name.to_string(),
169        (owner, Some(name)) if !owner.is_empty() => format!("{owner}/{name}"),
170        (owner, None) if !owner.is_empty() && owner != "pinned" => owner.to_string(),
171        _ => to_hex(hash),
172    }
173}
174
175fn unix_timestamp_now() -> u64 {
176    SystemTime::now()
177        .duration_since(UNIX_EPOCH)
178        .unwrap_or_default()
179        .as_secs()
180}
181
182impl HashtreeStore {
183    fn socialgraph_root_files(&self) -> [PathBuf; 4] {
184        let socialgraph = self.base_path().join("socialgraph");
185        [
186            socialgraph.join("events-root.msgpack"),
187            socialgraph.join("events-root-ambient.msgpack"),
188            socialgraph.join("profile-search-root.msgpack"),
189            socialgraph.join("profiles-by-pubkey-root.msgpack"),
190        ]
191    }
192
193    fn read_stored_cid(path: &Path) -> Result<Option<Hash>> {
194        #[derive(Deserialize)]
195        struct StoredCid {
196            hash: [u8; 32],
197            #[allow(dead_code)]
198            key: Option<[u8; 32]>,
199        }
200
201        let Ok(bytes) = std::fs::read(path) else {
202            return Ok(None);
203        };
204        let stored: StoredCid = rmp_serde::from_slice(&bytes)
205            .map_err(|e| anyhow::anyhow!("Failed to decode root file {}: {}", path.display(), e))?;
206        Ok(Some(stored.hash))
207    }
208
209    async fn collect_tree_hashes<S: Store>(
210        &self,
211        tree: &HashTree<S>,
212        root: &Hash,
213    ) -> Result<HashSet<Hash>> {
214        let mut hashes = HashSet::new();
215        let mut stack = vec![*root];
216
217        while let Some(hash) = stack.pop() {
218            if !hashes.insert(hash) {
219                continue;
220            }
221
222            let is_tree = tree
223                .is_tree(&hash)
224                .await
225                .map_err(|e| anyhow::anyhow!("Failed to check tree: {}", e))?;
226
227            if !is_tree {
228                continue;
229            }
230
231            if let Some(node) = tree
232                .get_tree_node(&hash)
233                .await
234                .map_err(|e| anyhow::anyhow!("Failed to get tree node: {}", e))?
235            {
236                for link in &node.links {
237                    stack.push(link.hash);
238                }
239            }
240        }
241
242        Ok(hashes)
243    }
244
245    fn protected_hashes(&self) -> Result<HashSet<Hash>> {
246        let mut protected = HashSet::new();
247
248        let rtxn = self.env.read_txn()?;
249        for (key_bytes, _) in self.blob_trees.iter(&rtxn)?.flatten() {
250            if key_bytes.len() >= 32 {
251                let hash: Hash = key_bytes[..32].try_into().unwrap();
252                protected.insert(hash);
253            }
254        }
255        drop(rtxn);
256
257        let tree = HashTree::new(HashTreeConfig::new(self.store_arc()).public());
258        for path in self.socialgraph_root_files() {
259            let Some(root_hash) = Self::read_stored_cid(&path)? else {
260                continue;
261            };
262            protected.extend(sync_block_on(self.collect_tree_hashes(&tree, &root_hash))?);
263        }
264
265        Ok(protected)
266    }
267
268    fn evict_disposable_orphans_to_target(&self, target_bytes: u64) -> Result<u64> {
269        let stats = self
270            .router
271            .writable_stats()
272            .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
273        let mut current_size = stats.total_bytes;
274        if current_size <= target_bytes {
275            return Ok(0);
276        }
277
278        let rtxn = self.env.read_txn()?;
279        let pinned: HashSet<Hash> = self
280            .pins
281            .iter(&rtxn)?
282            .filter_map(|item| item.ok())
283            .filter_map(|(hash_bytes, _)| {
284                if hash_bytes.len() == 32 {
285                    let mut hash = [0u8; 32];
286                    hash.copy_from_slice(hash_bytes);
287                    Some(hash)
288                } else {
289                    None
290                }
291            })
292            .collect();
293        drop(rtxn);
294
295        let protected_hashes = self.protected_hashes()?;
296        let all_hashes = self
297            .router
298            .list_writable()
299            .map_err(|e| anyhow::anyhow!("Failed to list writable hashes: {}", e))?;
300
301        let mut freed = 0u64;
302        for hash in all_hashes {
303            if current_size <= target_bytes {
304                break;
305            }
306
307            if pinned.contains(&hash) || protected_hashes.contains(&hash) {
308                continue;
309            }
310
311            if self.blob_has_owners(&hash)? {
312                continue;
313            }
314
315            let Some(size) = self
316                .router
317                .blob_size_sync(&hash)
318                .map_err(|e| anyhow::anyhow!("Failed to get blob size: {}", e))?
319            else {
320                continue;
321            };
322
323            if self
324                .router
325                .delete_local_only(&hash)
326                .map_err(|e| anyhow::anyhow!("Failed to delete orphaned blob: {}", e))?
327            {
328                freed = freed.saturating_add(size);
329                current_size = current_size.saturating_sub(size);
330                tracing::debug!(
331                    "Deleted disposable orphaned blob {} ({} bytes)",
332                    &to_hex(&hash)[..8],
333                    size
334                );
335            }
336        }
337
338        Ok(freed)
339    }
340
341    pub fn make_room_for_cached_blob(&self, incoming_bytes: u64) -> Result<u64> {
342        if self.max_size_bytes == 0 {
343            return Ok(0);
344        }
345
346        let stats = self
347            .router
348            .writable_stats()
349            .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
350        if stats.total_bytes.saturating_add(incoming_bytes) <= self.max_size_bytes {
351            return Ok(0);
352        }
353
354        let target = if incoming_bytes >= self.max_size_bytes {
355            0
356        } else {
357            (self.max_size_bytes.saturating_mul(9) / 10)
358                .min(self.max_size_bytes.saturating_sub(incoming_bytes))
359        };
360        self.evict_disposable_orphans_to_target(target)
361    }
362
363    pub fn enforce_cached_blob_budget_after_insert(&self, inserted_bytes: u64) -> Result<u64> {
364        if self.max_size_bytes == 0 || inserted_bytes == 0 {
365            return Ok(0);
366        }
367
368        let stats = self
369            .router
370            .writable_stats()
371            .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
372        if stats.total_bytes <= self.max_size_bytes {
373            return Ok(0);
374        }
375
376        let target = if inserted_bytes >= self.max_size_bytes {
377            inserted_bytes
378        } else {
379            (self.max_size_bytes.saturating_mul(9) / 10)
380                .saturating_add(inserted_bytes)
381                .min(self.max_size_bytes)
382        };
383        self.evict_disposable_orphans_to_target(target)
384    }
385
386    pub fn make_room_for_durable_blob(&self, incoming_bytes: u64) -> Result<u64> {
387        if self.max_size_bytes == 0 || incoming_bytes == 0 {
388            return Ok(0);
389        }
390
391        if incoming_bytes > self.max_size_bytes {
392            anyhow::bail!(
393                "storage limit exceeded: incoming blob is {} bytes but limit is {} bytes",
394                incoming_bytes,
395                self.max_size_bytes
396            );
397        }
398
399        let stats = self
400            .router
401            .writable_stats()
402            .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
403        if stats.total_bytes.saturating_add(incoming_bytes) <= self.max_size_bytes {
404            return Ok(0);
405        }
406
407        let target = (self.max_size_bytes.saturating_mul(9) / 10)
408            .min(self.max_size_bytes.saturating_sub(incoming_bytes));
409        let freed = self.evict_with_policy_to_target(stats.total_bytes, target)?;
410
411        let next_stats = self
412            .router
413            .writable_stats()
414            .map_err(|e| anyhow::anyhow!("Failed to get writable stats after eviction: {}", e))?;
415        if next_stats.total_bytes.saturating_add(incoming_bytes) > self.max_size_bytes {
416            anyhow::bail!(
417                "storage limit exceeded: {} bytes used, {} byte incoming blob, {} byte limit",
418                next_stats.total_bytes,
419                incoming_bytes,
420                self.max_size_bytes
421            );
422        }
423
424        Ok(freed)
425    }
426
427    pub fn enforce_durable_blob_budget_after_insert(&self, inserted_bytes: u64) -> Result<u64> {
428        if self.max_size_bytes == 0 || inserted_bytes == 0 {
429            return Ok(0);
430        }
431
432        if inserted_bytes > self.max_size_bytes {
433            anyhow::bail!(
434                "storage limit exceeded: inserted blobs are {} bytes but limit is {} bytes",
435                inserted_bytes,
436                self.max_size_bytes
437            );
438        }
439
440        let stats = self
441            .router
442            .writable_stats()
443            .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
444        if stats.total_bytes <= self.max_size_bytes {
445            return Ok(0);
446        }
447
448        let target = (self.max_size_bytes.saturating_mul(9) / 10)
449            .saturating_add(inserted_bytes)
450            .min(self.max_size_bytes);
451        let freed = self.evict_with_policy_to_target(stats.total_bytes, target)?;
452
453        let next_stats = self
454            .router
455            .writable_stats()
456            .map_err(|e| anyhow::anyhow!("Failed to get writable stats after eviction: {}", e))?;
457        if next_stats.total_bytes > self.max_size_bytes {
458            anyhow::bail!(
459                "storage limit exceeded: {} bytes used after inserting {} bytes, {} byte limit",
460                next_stats.total_bytes,
461                inserted_bytes,
462                self.max_size_bytes
463            );
464        }
465
466        Ok(freed)
467    }
468
469    pub fn relieve_cached_blob_write_pressure(&self, incoming_bytes: u64) -> Result<u64> {
470        let stats = self
471            .router
472            .writable_stats()
473            .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
474        if stats.total_bytes == 0 {
475            return Ok(0);
476        }
477
478        let headroom = incoming_bytes.max(stats.total_bytes / 10).max(1);
479        let target = stats.total_bytes.saturating_sub(headroom);
480        self.evict_disposable_orphans_to_target(target)
481    }
482
483    /// Pin a hash (prevent garbage collection)
484    pub fn pin(&self, hash: &[u8; 32]) -> Result<()> {
485        let mut wtxn = self.env.write_txn()?;
486        self.pins.put(&mut wtxn, hash.as_slice(), &())?;
487        wtxn.commit()?;
488        Ok(())
489    }
490
491    /// Unpin a hash (allow garbage collection)
492    pub fn unpin(&self, hash: &[u8; 32]) -> Result<()> {
493        let mut wtxn = self.env.write_txn()?;
494        self.pins.delete(&mut wtxn, hash.as_slice())?;
495        wtxn.commit()?;
496        Ok(())
497    }
498
499    /// Check if hash is pinned
500    pub fn is_pinned(&self, hash: &[u8; 32]) -> Result<bool> {
501        let rtxn = self.env.read_txn()?;
502        Ok(self.pins.get(&rtxn, hash.as_slice())?.is_some())
503    }
504
505    /// List all pinned hashes (raw bytes)
506    pub fn list_pins_raw(&self) -> Result<Vec<[u8; 32]>> {
507        let rtxn = self.env.read_txn()?;
508        let mut pins = Vec::new();
509
510        for item in self.pins.iter(&rtxn)? {
511            let (hash_bytes, _) = item?;
512            if hash_bytes.len() == 32 {
513                let mut hash = [0u8; 32];
514                hash.copy_from_slice(hash_bytes);
515                pins.push(hash);
516            }
517        }
518
519        Ok(pins)
520    }
521
522    /// List all pinned hashes with names
523    pub fn list_pins_with_names(&self) -> Result<Vec<PinnedItem>> {
524        let rtxn = self.env.read_txn()?;
525        let store = self.store_arc();
526        let tree = HashTree::new(HashTreeConfig::new(store).public());
527        let mut pins = Vec::new();
528
529        for item in self.pins.iter(&rtxn)? {
530            let (hash_bytes, _) = item?;
531            if hash_bytes.len() != 32 {
532                continue;
533            }
534            let mut hash = [0u8; 32];
535            hash.copy_from_slice(hash_bytes);
536
537            // Try to determine if it's a directory
538            let is_directory =
539                sync_block_on(async { tree.is_directory(&hash).await.unwrap_or(false) });
540
541            let meta = self
542                .tree_meta
543                .get(&rtxn, hash.as_slice())?
544                .map(|bytes| {
545                    rmp_serde::from_slice::<TreeMeta>(bytes)
546                        .map_err(|e| anyhow::anyhow!("Failed to deserialize TreeMeta: {}", e))
547                })
548                .transpose()?;
549            let size_bytes = if let Some(meta) = meta.as_ref() {
550                meta.total_size
551            } else {
552                self.router
553                    .blob_size_sync(&hash)
554                    .map_err(|e| anyhow::anyhow!("Failed to get pinned blob size: {}", e))?
555                    .unwrap_or(0)
556            };
557
558            pins.push(PinnedItem {
559                cid: to_hex(&hash),
560                name: pinned_item_name(&hash, meta.as_ref()),
561                is_directory,
562                size_bytes,
563            });
564        }
565
566        Ok(pins)
567    }
568
569    pub fn owned_blob_stats(&self) -> Result<Vec<OwnedBlobStats>> {
570        let rtxn = self.env.read_txn()?;
571        let mut owners = Vec::new();
572
573        for item in self.pubkey_blobs.iter(&rtxn)? {
574            let (owner_bytes, blobs_bytes) = item?;
575            if owner_bytes.len() != 32 {
576                continue;
577            }
578
579            let blobs: Vec<BlobMetadata> = serde_json::from_slice(blobs_bytes)
580                .map_err(|e| anyhow::anyhow!("Failed to deserialize blob metadata: {}", e))?;
581            let mut owner = [0u8; 32];
582            owner.copy_from_slice(owner_bytes);
583            let total_bytes = blobs
584                .iter()
585                .fold(0u64, |total, blob| total.saturating_add(blob.size));
586            owners.push(OwnedBlobStats {
587                owner,
588                count: blobs.len(),
589                total_bytes,
590            });
591        }
592
593        owners.sort_by_key(|stats| stats.owner);
594        Ok(owners)
595    }
596
597    // === Tree indexing for eviction ===
598
599    /// Index a tree after sync - tracks all blobs in the tree for eviction
600    ///
601    /// If `ref_key` is provided (e.g. "npub.../name"), it will replace any existing
602    /// tree with that ref, allowing old versions to be evicted.
603    pub fn index_tree(
604        &self,
605        root_hash: &Hash,
606        owner: &str,
607        name: Option<&str>,
608        priority: u8,
609        ref_key: Option<&str>,
610    ) -> Result<()> {
611        let root_hex = to_hex(root_hash);
612
613        // If ref_key provided, check for and unindex old version
614        if let Some(key) = ref_key {
615            let rtxn = self.env.read_txn()?;
616            if let Some(old_hash_bytes) = self.tree_refs.get(&rtxn, key)? {
617                if old_hash_bytes != root_hash.as_slice() {
618                    let old_hash: Hash = old_hash_bytes
619                        .try_into()
620                        .map_err(|_| anyhow::anyhow!("Invalid hash in tree_refs"))?;
621                    drop(rtxn);
622                    let _ = self.unpin(&old_hash);
623                    // Unindex old tree (will delete orphaned blobs)
624                    let _ = self.unindex_tree(&old_hash);
625                    tracing::debug!("Replaced old tree for ref {}", key);
626                }
627            }
628        }
629
630        let store = self.store_arc();
631        let tree = HashTree::new(HashTreeConfig::new(store).public());
632
633        let (tracked_hashes, total_size) =
634            sync_block_on(async { self.collect_tree_index(&tree, root_hash).await })?;
635
636        let mut wtxn = self.env.write_txn()?;
637
638        // Store blob-tree relationships (64-byte key: blob_hash ++ tree_hash)
639        for tracked_hash in &tracked_hashes {
640            let mut key = [0u8; 64];
641            key[..32].copy_from_slice(tracked_hash);
642            key[32..].copy_from_slice(root_hash);
643            self.blob_trees.put(&mut wtxn, &key[..], &())?;
644        }
645
646        // Store tree metadata
647        let now = unix_timestamp_now();
648        let meta = TreeMeta {
649            owner: owner.to_string(),
650            name: name.map(|s| s.to_string()),
651            synced_at: now,
652            total_size,
653            priority,
654        };
655        let meta_bytes = rmp_serde::to_vec(&meta)
656            .map_err(|e| anyhow::anyhow!("Failed to serialize TreeMeta: {}", e))?;
657        self.tree_meta
658            .put(&mut wtxn, root_hash.as_slice(), &meta_bytes)?;
659
660        // Store ref -> hash mapping if ref_key provided
661        if let Some(key) = ref_key {
662            self.tree_refs.put(&mut wtxn, key, root_hash.as_slice())?;
663        }
664
665        wtxn.commit()?;
666
667        tracing::debug!(
668            "Indexed tree {} ({} blobs, {} bytes, priority {})",
669            &root_hex[..8],
670            tracked_hashes.len(),
671            total_size,
672            priority
673        );
674
675        Ok(())
676    }
677
678    async fn collect_tree_index<S: Store>(
679        &self,
680        tree: &HashTree<S>,
681        root: &Hash,
682    ) -> Result<(HashSet<Hash>, u64)> {
683        let mut hashes = HashSet::new();
684        let mut total_size = 0u64;
685        let mut stack = vec![(*root, true)];
686
687        while let Some((hash, count_bytes)) = stack.pop() {
688            hashes.insert(hash);
689
690            let Some(node) = tree
691                .get_tree_node(&hash)
692                .await
693                .map_err(|e| anyhow::anyhow!("Failed to get tree node: {}", e))?
694            else {
695                if count_bytes {
696                    if let Some(size) = self
697                        .router
698                        .blob_size_sync(&hash)
699                        .map_err(|e| anyhow::anyhow!("Failed to get blob size: {}", e))?
700                    {
701                        total_size = total_size.saturating_add(size);
702                    }
703                }
704                continue;
705            };
706
707            for link in &node.links {
708                hashes.insert(link.hash);
709                match link.link_type {
710                    LinkType::Blob => {
711                        if count_bytes {
712                            total_size = total_size.saturating_add(link.size);
713                        }
714                    }
715                    LinkType::File => {
716                        if count_bytes {
717                            total_size = total_size.saturating_add(link.size);
718                        }
719                        stack.push((link.hash, false));
720                    }
721                    LinkType::Dir | LinkType::Fanout => {
722                        stack.push((link.hash, count_bytes));
723                    }
724                }
725            }
726        }
727
728        Ok((hashes, total_size))
729    }
730
731    /// Unindex a tree - removes blob-tree mappings and deletes orphaned blobs
732    /// Returns the number of bytes freed
733    pub fn unindex_tree(&self, root_hash: &Hash) -> Result<u64> {
734        let root_hex = to_hex(root_hash);
735
736        let store = self.store_arc();
737        let tree = HashTree::new(HashTreeConfig::new(store).public());
738
739        // Walk tree and collect all blob hashes
740        let tracked_hashes = sync_block_on(self.collect_tree_hashes(&tree, root_hash))?;
741
742        let mut wtxn = self.env.write_txn()?;
743        let mut freed = 0u64;
744
745        // For each blob, remove the blob-tree entry and check if orphaned
746        for tracked_hash in &tracked_hashes {
747            // Delete blob-tree entry (64-byte key: blob_hash ++ tree_hash)
748            let mut key = [0u8; 64];
749            key[..32].copy_from_slice(tracked_hash);
750            key[32..].copy_from_slice(root_hash);
751            self.blob_trees.delete(&mut wtxn, &key[..])?;
752
753            // Check if blob is in any other tree (prefix scan on first 32 bytes)
754            let mut has_other_tree = false;
755            for item in self.blob_trees.prefix_iter(&wtxn, &tracked_hash[..])? {
756                if item.is_ok() {
757                    has_other_tree = true;
758                    break;
759                }
760            }
761
762            // If orphaned, delete the blob
763            if !has_other_tree {
764                if let Some(size) = self
765                    .router
766                    .blob_size_sync(tracked_hash)
767                    .map_err(|e| anyhow::anyhow!("Failed to get blob size: {}", e))?
768                {
769                    freed += size;
770                    // Delete locally only - keep S3 as archive
771                    self.router
772                        .delete_local_only(tracked_hash)
773                        .map_err(|e| anyhow::anyhow!("Failed to delete blob: {}", e))?;
774                }
775            }
776        }
777
778        // Delete tree metadata
779        self.tree_meta.delete(&mut wtxn, root_hash.as_slice())?;
780
781        wtxn.commit()?;
782
783        tracing::debug!("Unindexed tree {} ({} bytes freed)", &root_hex[..8], freed);
784
785        Ok(freed)
786    }
787
788    /// Get tree metadata
789    pub fn get_tree_meta(&self, root_hash: &Hash) -> Result<Option<TreeMeta>> {
790        let rtxn = self.env.read_txn()?;
791        if let Some(bytes) = self.tree_meta.get(&rtxn, root_hash.as_slice())? {
792            let meta: TreeMeta = rmp_serde::from_slice(bytes)
793                .map_err(|e| anyhow::anyhow!("Failed to deserialize TreeMeta: {}", e))?;
794            Ok(Some(meta))
795        } else {
796            Ok(None)
797        }
798    }
799
800    pub fn get_tree_ref(&self, key: &str) -> Result<Option<Hash>> {
801        let rtxn = self.env.read_txn()?;
802        let Some(bytes) = self.tree_refs.get(&rtxn, key)? else {
803            return Ok(None);
804        };
805
806        let hash: Hash = bytes
807            .try_into()
808            .map_err(|_| anyhow::anyhow!("Invalid hash in tree_refs"))?;
809        Ok(Some(hash))
810    }
811
812    /// List all indexed trees
813    pub fn list_indexed_trees(&self) -> Result<Vec<(Hash, TreeMeta)>> {
814        let rtxn = self.env.read_txn()?;
815        let mut trees = Vec::new();
816
817        for item in self.tree_meta.iter(&rtxn)? {
818            let (hash_bytes, meta_bytes) = item?;
819            let hash: Hash = hash_bytes
820                .try_into()
821                .map_err(|_| anyhow::anyhow!("Invalid hash in tree_meta"))?;
822            let meta: TreeMeta = rmp_serde::from_slice(meta_bytes)
823                .map_err(|e| anyhow::anyhow!("Failed to deserialize TreeMeta: {}", e))?;
824            trees.push((hash, meta));
825        }
826
827        Ok(trees)
828    }
829
830    /// Get total tracked storage size (sum of all tree_meta.total_size)
831    pub fn tracked_size(&self) -> Result<u64> {
832        let rtxn = self.env.read_txn()?;
833        let mut total = 0u64;
834
835        for item in self.tree_meta.iter(&rtxn)? {
836            let (_, bytes) = item?;
837            let meta: TreeMeta = rmp_serde::from_slice(bytes)
838                .map_err(|e| anyhow::anyhow!("Failed to deserialize TreeMeta: {}", e))?;
839            total += meta.total_size;
840        }
841
842        Ok(total)
843    }
844
845    /// Get evictable trees sorted by (priority ASC, synced_at ASC).
846    ///
847    /// Blob-level access and raw LRU order live in the storage adapter. Indexed
848    /// tree metadata stays cheap and does not try to summarize all descendant
849    /// blob access on every stats or eviction pass.
850    fn get_evictable_trees(&self) -> Result<Vec<(Hash, TreeMeta)>> {
851        let mut trees = self.list_indexed_trees()?;
852
853        // Sort by priority (lower first), then by age.
854        trees.sort_by(|a, b| match a.1.priority.cmp(&b.1.priority) {
855            std::cmp::Ordering::Equal => a.1.synced_at.cmp(&b.1.synced_at),
856            other => other,
857        });
858
859        Ok(trees)
860    }
861
862    /// Run eviction if storage is over quota
863    /// Returns bytes freed
864    ///
865    /// Eviction order:
866    /// 1. Orphaned blobs (not in any indexed tree and not pinned)
867    /// 2. Trees by priority (lowest first) and access age (least recent first)
868    pub fn evict_if_needed(&self) -> Result<u64> {
869        // Get writable-tier storage used. In tiered LMDB mode the legacy env is
870        // cold archive/read-through data and must not drive hot-cache eviction.
871        let stats = self
872            .router
873            .writable_stats()
874            .map_err(|e| anyhow::anyhow!("Failed to get writable stats: {}", e))?;
875        let current = stats.total_bytes;
876
877        if current <= self.max_size_bytes {
878            return Ok(0);
879        }
880
881        // Target 90% of max to avoid constant eviction
882        let target = self.max_size_bytes * 90 / 100;
883        self.evict_with_policy_to_target(current, target)
884    }
885
886    fn evict_with_policy_to_target(&self, current: u64, target: u64) -> Result<u64> {
887        let mut freed = 0u64;
888        let mut current_size = current;
889
890        // Phase 1: Evict orphaned blobs (not in any tree and not pinned)
891        if self.evict_orphans {
892            let orphan_freed = self.evict_disposable_orphans_to_target(target)?;
893            freed += orphan_freed;
894            current_size = current_size.saturating_sub(orphan_freed);
895
896            if orphan_freed > 0 {
897                tracing::info!("Evicted orphaned blobs: {} bytes freed", orphan_freed);
898            }
899        } else {
900            tracing::debug!("Skipping orphan blob eviction; storage.evict_orphans=false");
901        }
902
903        // Check if we're now under target
904        if current_size <= target {
905            if freed > 0 {
906                tracing::info!("Eviction complete: {} bytes freed", freed);
907            }
908            return Ok(freed);
909        }
910
911        // Phase 2: Evict trees by priority (lowest first) and access age (least recent first)
912        // Own trees CAN be evicted (just last), but PINNED trees are never evicted
913        let evictable = self.get_evictable_trees()?;
914
915        for (root_hash, meta) in evictable {
916            if current_size <= target {
917                break;
918            }
919
920            let root_hex = to_hex(&root_hash);
921
922            // Never evict pinned trees
923            if self.is_pinned(&root_hash)? {
924                continue;
925            }
926
927            let tree_freed = self.unindex_tree(&root_hash)?;
928            freed += tree_freed;
929            current_size = current_size.saturating_sub(tree_freed);
930
931            tracing::info!(
932                "Evicted tree {} (owner={}, priority={}, {} bytes)",
933                &root_hex[..8],
934                &meta.owner[..8.min(meta.owner.len())],
935                meta.priority,
936                tree_freed
937            );
938        }
939
940        if freed > 0 {
941            tracing::info!("Eviction complete: {} bytes freed", freed);
942        }
943
944        Ok(freed)
945    }
946
947    /// Get the maximum storage size in bytes
948    pub fn max_size_bytes(&self) -> u64 {
949        self.max_size_bytes
950    }
951
952    /// Get storage usage by priority tier
953    pub fn storage_by_priority(&self) -> Result<StorageByPriority> {
954        let rtxn = self.env.read_txn()?;
955        let mut own = 0u64;
956        let mut followed = 0u64;
957        let mut other = 0u64;
958
959        for item in self.tree_meta.iter(&rtxn)? {
960            let (_, bytes) = item?;
961            let meta: TreeMeta = rmp_serde::from_slice(bytes)
962                .map_err(|e| anyhow::anyhow!("Failed to deserialize TreeMeta: {}", e))?;
963
964            if meta.priority == PRIORITY_OWN {
965                own += meta.total_size;
966            } else if meta.priority >= PRIORITY_FOLLOWED {
967                followed += meta.total_size;
968            } else {
969                other += meta.total_size;
970            }
971        }
972
973        Ok(StorageByPriority {
974            own,
975            followed,
976            other,
977        })
978    }
979
980    /// Get storage statistics
981    pub fn get_storage_stats(&self) -> Result<StorageStats> {
982        let rtxn = self.env.read_txn()?;
983        let total_pins = self.pins.len(&rtxn)? as usize;
984
985        let stats = self
986            .router
987            .stats()
988            .map_err(|e| anyhow::anyhow!("Failed to get stats: {}", e))?;
989
990        Ok(StorageStats {
991            total_dags: stats.count,
992            pinned_dags: total_pins,
993            total_bytes: stats.total_bytes,
994        })
995    }
996}
997
998#[cfg(test)]
999mod tests {
1000    use super::*;
1001    use hashtree_core::Cid;
1002    use hashtree_index::{BTree, BTreeOptions};
1003    use tempfile::TempDir;
1004
1005    use crate::storage::PRIORITY_OTHER;
1006
1007    fn write_root_file(path: &Path, cid: &Cid) {
1008        #[derive(Serialize)]
1009        struct StoredCid {
1010            hash: [u8; 32],
1011            key: Option<[u8; 32]>,
1012        }
1013
1014        std::fs::create_dir_all(path.parent().expect("root file parent")).expect("create dir");
1015        let bytes = rmp_serde::to_vec_named(&StoredCid {
1016            hash: cid.hash,
1017            key: cid.key,
1018        })
1019        .expect("encode cid");
1020        std::fs::write(path, bytes).expect("write root file");
1021    }
1022
1023    fn build_test_tree(store: &HashtreeStore) -> Cid {
1024        let index = BTree::new(store.store_arc(), BTreeOptions { order: Some(8) });
1025        sync_block_on(index.build(vec![
1026            ("alpha".to_string(), "one".to_string()),
1027            ("beta".to_string(), "two".to_string()),
1028            ("gamma".to_string(), "three".to_string()),
1029        ]))
1030        .expect("build btree")
1031        .expect("non-empty root")
1032    }
1033
1034    #[test]
1035    fn orphan_cleanup_keeps_indexed_tree_hashes() {
1036        let temp_dir = TempDir::new().expect("temp dir");
1037        let store = HashtreeStore::with_options(temp_dir.path(), None, 1024).expect("store");
1038        let cid = build_test_tree(&store);
1039
1040        store
1041            .index_tree(
1042                &cid.hash,
1043                "owner",
1044                Some("tree"),
1045                PRIORITY_OTHER,
1046                Some("owner/tree"),
1047            )
1048            .expect("index tree");
1049        let freed = store
1050            .evict_disposable_orphans_to_target(0)
1051            .expect("orphan cleanup");
1052
1053        assert!(freed < 1024);
1054        assert!(store.blob_exists(&cid.hash).expect("root exists"));
1055    }
1056
1057    #[test]
1058    fn list_pins_with_names_uses_indexed_tree_metadata() {
1059        let temp_dir = TempDir::new().expect("temp dir");
1060        let store = HashtreeStore::with_options(temp_dir.path(), None, 1024 * 1024).expect("store");
1061        let cid = build_test_tree(&store);
1062
1063        store.pin(&cid.hash).expect("pin tree");
1064        store
1065            .index_tree(
1066                &cid.hash,
1067                "npub1example",
1068                Some("playlist"),
1069                PRIORITY_OTHER,
1070                Some("npub1example/playlist"),
1071            )
1072            .expect("index tree");
1073
1074        let pins = store.list_pins_with_names().expect("list pins");
1075
1076        assert_eq!(pins.len(), 1);
1077        assert_eq!(pins[0].name, "npub1example/playlist");
1078        assert!(pins[0].size_bytes > 0);
1079    }
1080
1081    #[test]
1082    fn index_tree_records_multilevel_file_size_from_links() {
1083        let temp_dir = TempDir::new().expect("temp dir");
1084        let store = HashtreeStore::with_options(temp_dir.path(), None, 1024 * 1024).expect("store");
1085        let tree = HashTree::new(
1086            HashTreeConfig::new(store.store_arc())
1087                .public()
1088                .with_chunk_size(4)
1089                .with_max_links(2),
1090        );
1091        let data = (0u8..31).collect::<Vec<_>>();
1092        let (cid, size) = sync_block_on(tree.put(&data)).expect("put file");
1093
1094        store
1095            .index_tree(
1096                &cid.hash,
1097                "npub1example",
1098                Some("large-file"),
1099                PRIORITY_OTHER,
1100                Some("npub1example/large-file"),
1101            )
1102            .expect("index tree");
1103
1104        let meta = store
1105            .get_tree_meta(&cid.hash)
1106            .expect("tree meta")
1107            .expect("indexed meta");
1108        assert_eq!(size, data.len() as u64);
1109        assert_eq!(meta.total_size, data.len() as u64);
1110    }
1111
1112    #[test]
1113    fn get_tree_ref_returns_stored_root() {
1114        let temp_dir = TempDir::new().expect("temp dir");
1115        let store = HashtreeStore::with_options(temp_dir.path(), None, 1024 * 1024).expect("store");
1116        let cid = build_test_tree(&store);
1117
1118        store
1119            .index_tree(
1120                &cid.hash,
1121                "npub1example",
1122                Some("playlist"),
1123                PRIORITY_OTHER,
1124                Some("npub1example/playlist"),
1125            )
1126            .expect("index tree");
1127
1128        assert_eq!(
1129            store
1130                .get_tree_ref("npub1example/playlist")
1131                .expect("tree ref lookup"),
1132            Some(cid.hash)
1133        );
1134    }
1135
1136    #[test]
1137    fn tree_meta_deserializes_metadata_without_tree_access_field() {
1138        #[derive(Serialize)]
1139        struct LegacyTreeMeta {
1140            owner: String,
1141            name: Option<String>,
1142            synced_at: u64,
1143            total_size: u64,
1144            priority: u8,
1145        }
1146
1147        let bytes = rmp_serde::to_vec(&LegacyTreeMeta {
1148            owner: "owner".to_string(),
1149            name: Some("tree".to_string()),
1150            synced_at: 123,
1151            total_size: 456,
1152            priority: PRIORITY_OTHER,
1153        })
1154        .expect("serialize legacy metadata");
1155        let meta: TreeMeta = rmp_serde::from_slice(&bytes).expect("deserialize tree metadata");
1156
1157        assert_eq!(meta.owner, "owner");
1158        assert_eq!(meta.name.as_deref(), Some("tree"));
1159        assert_eq!(meta.synced_at, 123);
1160        assert_eq!(meta.total_size, 456);
1161        assert_eq!(meta.priority, PRIORITY_OTHER);
1162    }
1163
1164    #[test]
1165    fn tree_meta_deserializes_accidental_access_field_but_drops_it_on_write() {
1166        #[derive(Serialize)]
1167        struct AccidentalTreeMeta {
1168            owner: String,
1169            name: Option<String>,
1170            synced_at: u64,
1171            last_accessed_at: u64,
1172            total_size: u64,
1173            priority: u8,
1174        }
1175
1176        let bytes = rmp_serde::to_vec(&AccidentalTreeMeta {
1177            owner: "owner".to_string(),
1178            name: Some("tree".to_string()),
1179            synced_at: 123,
1180            last_accessed_at: 999,
1181            total_size: 456,
1182            priority: PRIORITY_OTHER,
1183        })
1184        .expect("serialize accidental metadata");
1185        let meta: TreeMeta = rmp_serde::from_slice(&bytes).expect("deserialize tree metadata");
1186        let encoded = rmp_serde::to_vec(&meta).expect("serialize current metadata");
1187        let reparsed: (String, Option<String>, u64, u64, u8) =
1188            rmp_serde::from_slice(&encoded).expect("parse current metadata shape");
1189
1190        assert_eq!(meta.owner, "owner");
1191        assert_eq!(meta.name.as_deref(), Some("tree"));
1192        assert_eq!(meta.synced_at, 123);
1193        assert_eq!(meta.total_size, 456);
1194        assert_eq!(meta.priority, PRIORITY_OTHER);
1195        assert_eq!(reparsed.0, "owner");
1196        assert_eq!(reparsed.3, 456);
1197        assert_eq!(reparsed.4, PRIORITY_OTHER);
1198    }
1199
1200    #[test]
1201    fn eviction_prefers_oldest_tree_within_priority() {
1202        let temp_dir = TempDir::new().expect("temp dir");
1203        let store = HashtreeStore::with_options(temp_dir.path(), None, 500).expect("store");
1204
1205        let hash1 = hashtree_core::sha256(&[1u8; 200]);
1206        let hash2 = hashtree_core::sha256(&[2u8; 200]);
1207        let hash3 = hashtree_core::sha256(&[3u8; 200]);
1208        store.put_blob(&[1u8; 200]).expect("put blob 1");
1209        store.put_blob(&[2u8; 200]).expect("put blob 2");
1210        store.put_blob(&[3u8; 200]).expect("put blob 3");
1211        store
1212            .index_tree(&hash1, "owner1", Some("tree1"), PRIORITY_OTHER, None)
1213            .expect("index tree 1");
1214        store
1215            .index_tree(&hash2, "owner2", Some("tree2"), PRIORITY_OTHER, None)
1216            .expect("index tree 2");
1217        store
1218            .index_tree(&hash3, "owner3", Some("tree3"), PRIORITY_OTHER, None)
1219            .expect("index tree 3");
1220
1221        let freed = store.evict_if_needed().expect("evict");
1222
1223        assert!(freed > 0);
1224        assert!(
1225            store.get_tree_meta(&hash3).expect("tree meta").is_some(),
1226            "newest tree should survive before older peers at the same priority"
1227        );
1228    }
1229
1230    #[test]
1231    fn orphan_cleanup_keeps_socialgraph_root_hashes() {
1232        let temp_dir = TempDir::new().expect("temp dir");
1233        let store = HashtreeStore::with_options(temp_dir.path(), None, 1024).expect("store");
1234        let cid = build_test_tree(&store);
1235        write_root_file(
1236            &temp_dir.path().join("socialgraph/events-root.msgpack"),
1237            &cid,
1238        );
1239
1240        let freed = store
1241            .evict_disposable_orphans_to_target(0)
1242            .expect("orphan cleanup");
1243
1244        assert!(freed < 1024);
1245        assert!(store.blob_exists(&cid.hash).expect("root exists"));
1246    }
1247}