1use crate::config::{MAX_DIRECTORY_RECORDS, MAX_OBJECTS};
2use crate::error::{Context, Error, ensure};
3use crate::{
4 container::{self, PrivateObject},
5 storage,
6};
7use serde::{Deserialize, Serialize};
8use sha2::{Digest, Sha256};
9use std::{
10 collections::BTreeMap,
11 fs,
12 path::{Path, PathBuf},
13};
14use urma_core::format::Urma;
15
16#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
17#[serde(rename_all = "snake_case")]
18pub enum Family {
19 #[default]
20 Litecoin,
21 Bitcoin,
22 Directory,
23 Ethereum,
24 Storj,
25 GoogleDrive,
26 S3,
27}
28
29#[derive(Debug, Default, Serialize, Deserialize)]
30#[serde(deny_unknown_fields)]
31pub struct Policy {
32 pub preferred: Family,
33}
34
35#[derive(Clone, Debug, Serialize, Deserialize)]
36#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
37pub enum Locator {
38 BlockRange {
39 network: String,
40 start_height: u64,
41 tip_height: u64,
42 tip_hash: String,
43 },
44 Directory {
45 path: PathBuf,
46 },
47}
48
49#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
50#[serde(rename_all = "snake_case")]
51pub enum Evidence {
52 LitecoinCoreTrusted,
53 LocalBytes,
54 CommitmentsAndCoreChain,
55 CommitmentsAndProviderChain,
56}
57
58#[derive(Clone, Debug, Serialize, Deserialize)]
59pub struct Observation {
60 pub backend: Family,
61 pub locator: Locator,
62 pub evidence: Evidence,
63 pub scanned_bytes: usize,
64}
65
66pub trait RecordSource {
67 fn read_records(
68 &self,
69 emit: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
70 ) -> Result<Observation, Error>;
71}
72
73pub struct Recovery {
74 pub objects: BTreeMap<[u8; 32], PrivateObject>,
75 pub rejected_records: u64,
76 pub observation: Observation,
77}
78
79pub fn accept_record(
80 key: &[u8; 32],
81 record: &[u8],
82 objects: &mut BTreeMap<[u8; 32], PrivateObject>,
83) -> Result<u64, Error> {
84 let chunk = match container::open_record(key, record) {
85 Ok(container::RecordMatch::Authenticated(chunk)) => chunk,
86 Ok(container::RecordMatch::Unrelated) => return Ok(0),
87 Err(error) => {
88 tracing::warn!(%error, "invalid private candidate rejected");
89 return Ok(1);
90 }
91 };
92 match objects.entry(chunk.id) {
93 std::collections::btree_map::Entry::Occupied(mut entry) => {
94 match entry.get_mut().insert(chunk) {
95 Ok(()) => {}
96 Err(error) => {
97 tracing::warn!(%error, "authenticated object conflict retained");
98 }
99 }
100 }
101 std::collections::btree_map::Entry::Vacant(entry) => {
102 entry.insert(PrivateObject::new(chunk));
103 }
104 }
105 ensure!(
106 objects.len() <= MAX_OBJECTS,
107 "client object capacity exceeded"
108 );
109 Ok(0)
110}
111
112pub fn recover(source: &dyn RecordSource, key: &[u8; 32]) -> Result<Recovery, Error> {
113 let mut objects = BTreeMap::new();
114 let mut rejected_records = 0u64;
115 let observation = source.read_records(&mut |record| {
116 rejected_records = rejected_records
117 .checked_add(accept_record(key, record, &mut objects)?)
118 .ok_or_else(|| Error::Invalid("rejected record count overflow".into()))?;
119 Ok(())
120 })?;
121 Ok(Recovery {
122 objects,
123 rejected_records,
124 observation,
125 })
126}
127
128pub struct DirectorySource<'a> {
129 pub path: &'a Path,
130}
131
132impl RecordSource for DirectorySource<'_> {
133 fn read_records(
134 &self,
135 emit: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
136 ) -> Result<Observation, Error> {
137 let mut paths = Vec::new();
138 for entry in fs::read_dir(self.path)?.take(MAX_DIRECTORY_RECORDS + 1) {
139 let entry = entry?;
140 ensure!(
141 entry.file_type()?.is_file(),
142 "record directory contains a non-regular entry"
143 );
144 ensure!(
145 entry
146 .path()
147 .extension()
148 .context("record extension missing")?
149 == "urma-record",
150 "unexpected record directory entry"
151 );
152 paths.push(entry.path());
153 }
154 ensure!(
155 paths.len() <= MAX_DIRECTORY_RECORDS,
156 "directory record limit reached"
157 );
158 paths.sort();
159 let mut scanned_bytes = 0;
160 for path in &paths {
161 let record = storage::read_bounded(path, Urma::PRIVATE_RECORD_BYTES)?;
162 scanned_bytes += record.len();
163 emit(&record)?;
164 }
165 Ok(Observation {
166 backend: Family::Directory,
167 locator: Locator::Directory {
168 path: self.path.to_path_buf(),
169 },
170 evidence: Evidence::LocalBytes,
171 scanned_bytes,
172 })
173 }
174}
175
176pub fn store_directory(path: &Path, records: &[Vec<u8>]) -> Result<Observation, Error> {
177 ensure!(
178 !records.is_empty() && records.len() <= MAX_DIRECTORY_RECORDS,
179 "invalid record count"
180 );
181 for record in records {
182 container::inspect_header(record)?;
183 }
184 urma_io::create_private_directory(path)?;
185 let mut stored_bytes = 0;
186 for record in records {
187 let name = format!("{}.urma-record", hex::encode(Sha256::digest(record)));
188 let target = path.join(name);
189 if !target.try_exists()? {
190 storage::write_new(&target, record)?;
191 stored_bytes += record.len();
192 }
193 }
194 Ok(Observation {
195 backend: Family::Directory,
196 locator: Locator::Directory {
197 path: path.to_path_buf(),
198 },
199 evidence: Evidence::LocalBytes,
200 scanned_bytes: stored_bytes,
201 })
202}
203
204pub fn require_implemented(family: Family) -> Result<(), Error> {
205 ensure!(
206 matches!(
207 family,
208 Family::Litecoin | Family::Bitcoin | Family::Directory
209 ),
210 "{family:?} adapter is not implemented; no fallback to another backend"
211 );
212 Ok(())
213}
214
215pub struct ExportReport {
216 pub complete: bool,
217 pub report: serde_json::Value,
218}
219
220pub fn export(recovery: Recovery, output: &Path) -> Result<ExportReport, Error> {
221 ensure!(
222 !output.try_exists()?,
223 "recovery requires a new output directory"
224 );
225 let mut reports = Vec::new();
226 let mut complete = 0usize;
227 for object in recovery.objects.values() {
228 let id = hex::encode(object.id);
229 if object.is_conflicted() {
230 reports.push(
231 serde_json::json!({"id":id,"status":"invalid","reason":"authenticated_conflict"}),
232 );
233 continue;
234 }
235 if !object.is_complete() {
236 reports.push(serde_json::json!({"id":id,"status":"incomplete","received":object.received(),"expected":object.count,"bytes":object.total}));
237 continue;
238 }
239 let bytes = match object.finish() {
240 Ok(bytes) => bytes,
241 Err(error) => {
242 tracing::warn!(%error, %id, "object integrity failed; other objects remain recoverable");
243 reports.push(
244 serde_json::json!({"id":id,"status":"invalid","reason":error.to_string()}),
245 );
246 continue;
247 }
248 };
249 if complete == 0 {
250 urma_io::create_private_directory(output)?;
251 }
252 storage::write_new(&output.join(format!("{id}.bin")), &bytes)?;
253 reports.push(serde_json::json!({"id":id,"status":"complete","bytes":bytes.len(),"content_type":object.content_type.code(),
254 "chunks":object.count,"sha256":hex::encode(Sha256::digest(&bytes))}));
255 complete += 1;
256 }
257 let success = complete > 0 && complete == recovery.objects.len();
258 let status = if success {
259 "complete"
260 } else if recovery.objects.is_empty() {
261 "not_found"
262 } else {
263 "partial"
264 };
265 let mut report = serde_json::json!({"status":status,"objects":reports,"rejected_records":recovery.rejected_records,
266 "scanned_bytes":recovery.observation.scanned_bytes,"observation":recovery.observation});
267 if let Locator::BlockRange {
268 network,
269 start_height,
270 tip_height,
271 tip_hash,
272 } = &recovery.observation.locator
273 {
274 report["network"] = serde_json::json!(network);
275 report["start_height"] = serde_json::json!(start_height);
276 report["tip_height"] = serde_json::json!(tip_height);
277 report["tip_hash"] = serde_json::json!(tip_hash);
278 }
279 Ok(ExportReport {
280 complete: success,
281 report,
282 })
283}