goat.navy
forked from
tranquil.farm/tranquil-pds
Our Personal Data Server from scratch!
37 kB
1146 lines
1use chrono::{DateTime, TimeZone, Utc};
2use fjall::Keyspace;
3use std::sync::Arc;
4use uuid::Uuid;
5
6use super::MetastoreError;
7use super::encoding::KeyReader;
8use super::keys::{KeyTag, UserHash};
9use super::repo_meta::{RepoMetaValue, RepoStatus, handle_key, repo_meta_key, repo_meta_prefix};
10use super::scan::{count_prefix, point_lookup};
11use super::user_hash::UserHashMap;
12
13use tranquil_types::{CidLink, Did, Handle};
14
15pub struct RepoOps {
16 repo_data: Keyspace,
17 user_hashes: Arc<UserHashMap>,
18}
19
20impl RepoOps {
21 pub fn new(repo_data: Keyspace, user_hashes: Arc<UserHashMap>) -> Self {
22 Self {
23 repo_data,
24 user_hashes,
25 }
26 }
27
28 pub fn create_repo(
29 &self,
30 db: &fjall::Database,
31 user_id: Uuid,
32 did: &Did,
33 handle: &Handle,
34 repo_root_cid: &CidLink,
35 repo_rev: &str,
36 ) -> Result<(), MetastoreError> {
37 let user_hash = UserHash::from_did(did.as_str());
38 let mut batch = db.batch();
39
40 self.user_hashes
41 .stage_insert(&mut batch, user_id, user_hash)?;
42
43 let cid_bytes = cid_link_to_bytes(repo_root_cid)?;
44 let handle_lower = handle.as_str().to_ascii_lowercase();
45
46 let value = RepoMetaValue {
47 repo_root_cid: cid_bytes,
48 repo_rev: repo_rev.to_string(),
49 handle: handle_lower.clone(),
50 status: RepoStatus::Active,
51 deactivated_at_ms: None,
52 takedown_ref: None,
53 did: Some(did.as_str().to_string()),
54 };
55
56 batch.insert(
57 &self.repo_data,
58 repo_meta_key(user_hash).as_slice(),
59 value.serialize(),
60 );
61
62 batch.insert(
63 &self.repo_data,
64 handle_key(&handle_lower).as_slice(),
65 user_hash.raw().to_be_bytes(),
66 );
67
68 match batch.commit() {
69 Ok(()) => Ok(()),
70 Err(e) => {
71 self.user_hashes.rollback_insert(&user_id, &user_hash);
72 Err(MetastoreError::Fjall(e))
73 }
74 }
75 }
76
77 pub fn get_repo_meta(
78 &self,
79 user_id: Uuid,
80 ) -> Result<Option<(UserHash, RepoMetaValue)>, MetastoreError> {
81 let user_hash = match self.user_hashes.get(&user_id) {
82 Some(h) => h,
83 None => return Ok(None),
84 };
85 let key = repo_meta_key(user_hash);
86 Ok(point_lookup(
87 &self.repo_data,
88 key.as_slice(),
89 RepoMetaValue::deserialize,
90 "invalid repo_meta value",
91 )?
92 .map(|v| (user_hash, v)))
93 }
94
95 pub fn write_repo_meta(
96 &self,
97 batch: &mut fjall::OwnedWriteBatch,
98 user_hash: UserHash,
99 value: &RepoMetaValue,
100 ) {
101 batch.insert(
102 &self.repo_data,
103 repo_meta_key(user_hash).as_slice(),
104 value.serialize(),
105 );
106 }
107
108 pub(crate) fn update_repo_root(
109 &self,
110 db: &fjall::Database,
111 user_id: Uuid,
112 repo_root_cid: &CidLink,
113 repo_rev: &str,
114 ) -> Result<(), MetastoreError> {
115 let user_hash = self.resolve_user_hash(user_id)?;
116 let key = repo_meta_key(user_hash);
117
118 let mut value = self.get_meta_value(key.as_slice())?;
119 let cid_bytes = cid_link_to_bytes(repo_root_cid)?;
120 value.repo_root_cid = cid_bytes;
121 value.repo_rev = repo_rev.to_string();
122
123 let mut batch = db.batch();
124 batch.insert(&self.repo_data, key.as_slice(), value.serialize());
125 batch.commit().map_err(MetastoreError::Fjall)
126 }
127
128 pub(crate) fn update_repo_rev(
129 &self,
130 db: &fjall::Database,
131 user_id: Uuid,
132 repo_rev: &str,
133 ) -> Result<(), MetastoreError> {
134 let user_hash = self.resolve_user_hash(user_id)?;
135 let key = repo_meta_key(user_hash);
136
137 let mut value = self.get_meta_value(key.as_slice())?;
138 value.repo_rev = repo_rev.to_string();
139
140 let mut batch = db.batch();
141 batch.insert(&self.repo_data, key.as_slice(), value.serialize());
142 batch.commit().map_err(MetastoreError::Fjall)
143 }
144
145 pub fn update_repo_status(
146 &self,
147 db: &fjall::Database,
148 did: &Did,
149 takedown: Option<bool>,
150 takedown_ref: Option<&str>,
151 deactivated: Option<bool>,
152 ) -> Result<(), MetastoreError> {
153 let user_hash = UserHash::from_did(did.as_str());
154 let key = repo_meta_key(user_hash);
155 let existing = point_lookup(
156 &self.repo_data,
157 key.as_slice(),
158 RepoMetaValue::deserialize,
159 "invalid repo_meta value",
160 )?;
161 let mut value = match existing {
162 Some(v) => v,
163 None => {
164 tracing::warn!(
165 did = did.as_str(),
166 "update_repo_status: repo not found in metastore"
167 );
168 return Ok(());
169 }
170 };
171
172 match value.status {
173 RepoStatus::Suspended | RepoStatus::Deleted => return Ok(()),
174 _ => {}
175 }
176
177 if let Some(taken_down) = takedown {
178 value.takedown_ref = match taken_down {
179 true => Some(takedown_ref.unwrap_or("").to_owned()),
180 false => None,
181 };
182 }
183 if let Some(now_deactivated) = deactivated {
184 value.deactivated_at_ms = match now_deactivated {
185 true => value.deactivated_at_ms.or_else(|| {
186 Some(u64::try_from(chrono::Utc::now().timestamp_millis()).unwrap_or(0))
187 }),
188 false => None,
189 };
190 }
191
192 let is_taken_down = match takedown {
193 Some(v) => v,
194 None => value.takedown_ref.is_some(),
195 };
196 value.status = match (is_taken_down, value.deactivated_at_ms.is_some()) {
197 (true, _) => RepoStatus::Takendown,
198 (false, true) => RepoStatus::Deactivated,
199 (false, false) => RepoStatus::Active,
200 };
201
202 let mut batch = db.batch();
203 batch.insert(&self.repo_data, key.as_slice(), value.serialize());
204 batch.commit().map_err(MetastoreError::Fjall)
205 }
206
207 pub fn update_handle(
208 &self,
209 db: &fjall::Database,
210 user_id: Uuid,
211 new_handle: &Handle,
212 ) -> Result<(), MetastoreError> {
213 let user_hash = self.resolve_user_hash(user_id)?;
214 let key = repo_meta_key(user_hash);
215 let mut value = self.get_meta_value(key.as_slice())?;
216 let new_lower = new_handle.as_str().to_ascii_lowercase();
217
218 let mut batch = db.batch();
219
220 match value.handle.is_empty() {
221 true => {}
222 false => batch.remove(&self.repo_data, handle_key(&value.handle).as_slice()),
223 }
224
225 batch.insert(
226 &self.repo_data,
227 handle_key(&new_lower).as_slice(),
228 user_hash.raw().to_be_bytes(),
229 );
230
231 value.handle = new_lower;
232 batch.insert(&self.repo_data, key.as_slice(), value.serialize());
233
234 batch.commit().map_err(MetastoreError::Fjall)
235 }
236
237 pub fn delete_repo(&self, db: &fjall::Database, user_id: Uuid) -> Result<(), MetastoreError> {
238 let user_hash = self.resolve_user_hash(user_id)?;
239 let key = repo_meta_key(user_hash);
240
241 let meta = self.get_meta_value(key.as_slice())?;
242
243 let mut batch = db.batch();
244 batch.remove(&self.repo_data, key.as_slice());
245
246 match meta.handle.is_empty() {
247 true => {}
248 false => batch.remove(&self.repo_data, handle_key(&meta.handle).as_slice()),
249 }
250
251 self.user_hashes.stage_remove(&mut batch, &user_id);
252
253 match batch.commit() {
254 Ok(()) => Ok(()),
255 Err(e) => {
256 self.user_hashes.rollback_remove(user_id, user_hash);
257 Err(MetastoreError::Fjall(e))
258 }
259 }
260 }
261
262 pub fn get_repo(&self, user_id: Uuid) -> Result<Option<RepoInfo>, MetastoreError> {
263 let user_hash = match self.user_hashes.get(&user_id) {
264 Some(h) => h,
265 None => return Ok(None),
266 };
267 let key = repo_meta_key(user_hash);
268 point_lookup(
269 &self.repo_data,
270 key.as_slice(),
271 RepoMetaValue::deserialize,
272 "invalid repo_meta value",
273 )?
274 .map(|value| {
275 let cid = bytes_to_cid_link(&value.repo_root_cid)?;
276 Ok(RepoInfo {
277 user_id,
278 repo_root_cid: cid,
279 repo_rev: Some(value.repo_rev),
280 })
281 })
282 .transpose()
283 }
284
285 pub fn get_repo_root_for_update(
286 &self,
287 user_id: Uuid,
288 ) -> Result<Option<CidLink>, MetastoreError> {
289 let user_hash = match self.user_hashes.get(&user_id) {
290 Some(h) => h,
291 None => return Ok(None),
292 };
293 let key = repo_meta_key(user_hash);
294 point_lookup(
295 &self.repo_data,
296 key.as_slice(),
297 RepoMetaValue::deserialize,
298 "invalid repo_meta value",
299 )?
300 .map(|v| bytes_to_cid_link(&v.repo_root_cid))
301 .transpose()
302 }
303
304 pub fn get_repo_root_by_did(&self, did: &Did) -> Result<Option<CidLink>, MetastoreError> {
305 let user_hash = UserHash::from_did(did.as_str());
306 let key = repo_meta_key(user_hash);
307 point_lookup(
308 &self.repo_data,
309 key.as_slice(),
310 RepoMetaValue::deserialize,
311 "invalid repo_meta value",
312 )?
313 .map(|v| bytes_to_cid_link(&v.repo_root_cid))
314 .transpose()
315 }
316
317 pub fn get_repo_root_cid_by_user_id(
318 &self,
319 user_id: Uuid,
320 ) -> Result<Option<CidLink>, MetastoreError> {
321 self.get_repo_root_for_update(user_id)
322 }
323
324 pub fn count_repos(&self) -> Result<i64, MetastoreError> {
325 let prefix = repo_meta_prefix();
326 count_prefix(&self.repo_data, prefix.as_slice())
327 }
328
329 pub fn get_repos_without_rev(
330 &self,
331 limit: usize,
332 ) -> Result<Vec<RepoWithoutRevEntry>, MetastoreError> {
333 let prefix = repo_meta_prefix();
334 self.repo_data
335 .prefix(prefix.as_slice())
336 .filter_map(|guard| {
337 let (key_bytes, val_bytes) = match guard.into_inner() {
338 Ok(pair) => pair,
339 Err(e) => return Some(Err(MetastoreError::Fjall(e))),
340 };
341 let value = match RepoMetaValue::deserialize(&val_bytes) {
342 Some(v) => v,
343 None => {
344 return Some(Err(MetastoreError::CorruptData("invalid repo_meta value")));
345 }
346 };
347 match value.repo_rev.is_empty() {
348 true => Some(decode_without_rev_entry(
349 &key_bytes,
350 &value,
351 &self.user_hashes,
352 )),
353 false => None,
354 }
355 })
356 .take(limit)
357 .collect()
358 }
359
360 pub fn get_account_with_repo(
361 &self,
362 did: &Did,
363 ) -> Result<Option<RepoAccountEntry>, MetastoreError> {
364 let user_hash = UserHash::from_did(did.as_str());
365 let key = repo_meta_key(user_hash);
366 point_lookup(
367 &self.repo_data,
368 key.as_slice(),
369 RepoMetaValue::deserialize,
370 "invalid repo_meta value",
371 )?
372 .map(|value| {
373 let user_id =
374 self.user_hashes
375 .get_uuid(&user_hash)
376 .ok_or(MetastoreError::CorruptData(
377 "user_hash has no reverse mapping",
378 ))?;
379 let cid = Some(bytes_to_cid_link(&value.repo_root_cid)?);
380 let deactivated_at = value
381 .deactivated_at_ms
382 .and_then(|ms| i64::try_from(ms).ok())
383 .and_then(|ms| Utc.timestamp_millis_opt(ms).single());
384 Ok(RepoAccountEntry {
385 user_id,
386 did: did.clone(),
387 deactivated_at,
388 takedown_ref: value.takedown_ref,
389 repo_root_cid: cid,
390 })
391 })
392 .transpose()
393 }
394
395 pub fn list_repos_paginated(
396 &self,
397 cursor_user_hash: Option<u64>,
398 limit: usize,
399 ) -> Result<Vec<RepoListEntry>, MetastoreError> {
400 const _: () = assert!(KeyTag::REPO_META.raw() < 0xFF);
401 let upper = KeyTag::REPO_META.exclusive_prefix_bound();
402
403 let start = match cursor_user_hash {
404 Some(cursor) => match cursor.checked_add(1) {
405 Some(next) => repo_meta_key(UserHash::from_raw(next)),
406 None => return Ok(Vec::new()),
407 },
408 None => repo_meta_prefix(),
409 };
410
411 self.repo_data
412 .range(start.as_slice()..upper.as_slice())
413 .take(limit)
414 .map(|guard| {
415 let (k, v) = guard.into_inner().map_err(MetastoreError::Fjall)?;
416 decode_list_entry_from_kv(&k, &v, &self.user_hashes)
417 })
418 .collect()
419 }
420
421 fn resolve_user_hash(&self, user_id: Uuid) -> Result<UserHash, MetastoreError> {
422 self.user_hashes
423 .get(&user_id)
424 .ok_or(MetastoreError::InvalidInput("unknown user_id"))
425 }
426
427 fn get_meta_value(&self, key: &[u8]) -> Result<RepoMetaValue, MetastoreError> {
428 point_lookup(
429 &self.repo_data,
430 key,
431 RepoMetaValue::deserialize,
432 "invalid repo_meta value",
433 )?
434 .ok_or(MetastoreError::CorruptData("repo_meta not found"))
435 }
436
437 pub fn lookup_handle(&self, handle: &Handle) -> Result<Option<Uuid>, MetastoreError> {
438 let handle_lower = handle.as_str().to_ascii_lowercase();
439 let key = handle_key(&handle_lower);
440
441 match self
442 .repo_data
443 .get(key.as_slice())
444 .map_err(MetastoreError::Fjall)?
445 {
446 Some(raw) => {
447 let user_hash = parse_handle_value(&raw)?;
448 self.user_hashes
449 .get_uuid(&user_hash)
450 .ok_or(MetastoreError::CorruptData(
451 "handle maps to unknown user_hash",
452 ))
453 .map(Some)
454 }
455 None => Ok(None),
456 }
457 }
458}
459
460#[derive(Debug, Clone)]
461pub struct RepoInfo {
462 pub user_id: Uuid,
463 pub repo_root_cid: CidLink,
464 pub repo_rev: Option<String>,
465}
466
467#[derive(Debug, Clone)]
468pub struct RepoWithoutRevEntry {
469 pub user_id: Uuid,
470 pub repo_root_cid: CidLink,
471}
472
473#[derive(Debug, Clone)]
474pub struct RepoAccountEntry {
475 pub user_id: Uuid,
476 pub did: Did,
477 pub deactivated_at: Option<DateTime<Utc>>,
478 pub takedown_ref: Option<String>,
479 pub repo_root_cid: Option<CidLink>,
480}
481
482#[derive(Debug, Clone)]
483pub struct RepoListEntry {
484 pub user_id: Uuid,
485 pub user_hash: UserHash,
486 pub did: Option<String>,
487 pub deactivated_at: Option<DateTime<Utc>>,
488 pub takedown_ref: Option<String>,
489 pub repo_root_cid: CidLink,
490 pub repo_rev: Option<String>,
491}
492
493fn decode_list_entry_from_kv(
494 key_bytes: &[u8],
495 val_bytes: &[u8],
496 user_hashes: &UserHashMap,
497) -> Result<RepoListEntry, MetastoreError> {
498 let value = RepoMetaValue::deserialize(val_bytes)
499 .ok_or(MetastoreError::CorruptData("invalid repo_meta value"))?;
500 let user_hash = parse_repo_meta_key_hash(key_bytes)
501 .ok_or(MetastoreError::CorruptData("invalid repo_meta key"))?;
502 let user_id = user_hashes
503 .get_uuid(&user_hash)
504 .ok_or(MetastoreError::CorruptData(
505 "user_hash has no reverse mapping",
506 ))?;
507 let cid = bytes_to_cid_link(&value.repo_root_cid)?;
508 let deactivated_at = value
509 .deactivated_at_ms
510 .and_then(|ms| i64::try_from(ms).ok())
511 .and_then(|ms| Utc.timestamp_millis_opt(ms).single());
512 Ok(RepoListEntry {
513 user_id,
514 user_hash,
515 did: value.did,
516 deactivated_at,
517 takedown_ref: value.takedown_ref,
518 repo_root_cid: cid,
519 repo_rev: match value.repo_rev.is_empty() {
520 true => None,
521 false => Some(value.repo_rev),
522 },
523 })
524}
525
526fn decode_without_rev_entry(
527 key_bytes: &[u8],
528 value: &RepoMetaValue,
529 user_hashes: &UserHashMap,
530) -> Result<RepoWithoutRevEntry, MetastoreError> {
531 let user_hash = parse_repo_meta_key_hash(key_bytes)
532 .ok_or(MetastoreError::CorruptData("invalid repo_meta key"))?;
533 let user_id = user_hashes
534 .get_uuid(&user_hash)
535 .ok_or(MetastoreError::CorruptData(
536 "user_hash has no reverse mapping",
537 ))?;
538 let cid = bytes_to_cid_link(&value.repo_root_cid)?;
539 Ok(RepoWithoutRevEntry {
540 user_id,
541 repo_root_cid: cid,
542 })
543}
544
545pub(crate) fn cid_link_to_bytes(cid_link: &CidLink) -> Result<Vec<u8>, MetastoreError> {
546 let cid = cid_link.to_cid().ok_or(MetastoreError::InvalidInput(
547 "CidLink does not contain a valid CID",
548 ))?;
549 Ok(cid.to_bytes())
550}
551
552pub(crate) fn bytes_to_cid_link(bytes: &[u8]) -> Result<CidLink, MetastoreError> {
553 let cid = cid::Cid::read_bytes(std::io::Cursor::new(bytes))
554 .map_err(|_| MetastoreError::CorruptData("invalid CID bytes in repo_meta"))?;
555 Ok(CidLink::from_cid(&cid))
556}
557
558fn parse_handle_value(raw: &[u8]) -> Result<UserHash, MetastoreError> {
559 let bytes: [u8; 8] = raw
560 .try_into()
561 .map_err(|_| MetastoreError::CorruptData("handle value not 8 bytes"))?;
562 Ok(UserHash::from_raw(u64::from_be_bytes(bytes)))
563}
564
565fn parse_repo_meta_key_hash(key_bytes: &[u8]) -> Option<UserHash> {
566 let mut reader = KeyReader::new(key_bytes);
567 let _tag = reader.tag()?;
568 let hash = reader.u64()?;
569 Some(UserHash::from_raw(hash))
570}
571
572#[cfg(test)]
573mod tests {
574 use super::*;
575 use crate::metastore::{Metastore, MetastoreConfig};
576
577 fn test_config() -> MetastoreConfig {
578 MetastoreConfig {
579 cache_size_bytes: 64 * 1024 * 1024,
580 }
581 }
582
583 fn test_cid_link(seed: u8) -> CidLink {
584 let digest: [u8; 32] = std::array::from_fn(|i| seed.wrapping_add(i as u8));
585 let mh = multihash::Multihash::<64>::wrap(0x12, &digest).unwrap();
586 let c = cid::Cid::new_v1(0x71, mh);
587 CidLink::from_cid(&c)
588 }
589
590 fn open_fresh() -> (tempfile::TempDir, Metastore) {
591 let dir = tempfile::TempDir::new().unwrap();
592 let ms = Metastore::open(dir.path(), test_config()).unwrap();
593 (dir, ms)
594 }
595
596 fn test_did(name: &str) -> Did {
597 Did::from(format!("did:plc:{name}"))
598 }
599
600 fn test_handle(name: &str) -> Handle {
601 Handle::from(format!("{name}.test.invalid"))
602 }
603
604 #[test]
605 fn create_and_get_repo() {
606 let (_dir, ms) = open_fresh();
607 let ops = ms.repo_ops();
608 let user_id = uuid::Uuid::new_v4();
609 let did = test_did("olaren");
610 let handle = test_handle("olaren");
611 let cid = test_cid_link(1);
612
613 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev1")
614 .unwrap();
615
616 let repo = ops.get_repo(user_id).unwrap().unwrap();
617 assert_eq!(repo.user_id, user_id);
618 assert_eq!(repo.repo_root_cid, cid);
619 assert_eq!(repo.repo_rev.as_deref(), Some("rev1"));
620 }
621
622 #[test]
623 fn get_repo_returns_none_for_unknown() {
624 let (_dir, ms) = open_fresh();
625 let ops = ms.repo_ops();
626 assert!(ops.get_repo(uuid::Uuid::new_v4()).unwrap().is_none());
627 }
628
629 #[test]
630 fn update_repo_root() {
631 let (_dir, ms) = open_fresh();
632 let ops = ms.repo_ops();
633 let user_id = uuid::Uuid::new_v4();
634 let did = test_did("teq");
635 let handle = test_handle("teq");
636 let cid1 = test_cid_link(1);
637 let cid2 = test_cid_link(2);
638
639 ops.create_repo(ms.database(), user_id, &did, &handle, &cid1, "rev1")
640 .unwrap();
641 ops.update_repo_root(ms.database(), user_id, &cid2, "rev2")
642 .unwrap();
643
644 let repo = ops.get_repo(user_id).unwrap().unwrap();
645 assert_eq!(repo.repo_root_cid, cid2);
646 assert_eq!(repo.repo_rev.as_deref(), Some("rev2"));
647 }
648
649 #[test]
650 fn update_repo_rev() {
651 let (_dir, ms) = open_fresh();
652 let ops = ms.repo_ops();
653 let user_id = uuid::Uuid::new_v4();
654 let did = test_did("nel");
655 let handle = test_handle("nel");
656 let cid = test_cid_link(3);
657
658 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev1")
659 .unwrap();
660 ops.update_repo_rev(ms.database(), user_id, "rev_updated")
661 .unwrap();
662
663 let repo = ops.get_repo(user_id).unwrap().unwrap();
664 assert_eq!(repo.repo_root_cid, cid);
665 assert_eq!(repo.repo_rev.as_deref(), Some("rev_updated"));
666 }
667
668 #[test]
669 fn delete_repo_removes_meta_and_handle() {
670 let (_dir, ms) = open_fresh();
671 let ops = ms.repo_ops();
672 let user_id = uuid::Uuid::new_v4();
673 let did = test_did("lyna");
674 let handle = test_handle("lyna");
675 let cid = test_cid_link(4);
676
677 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev1")
678 .unwrap();
679 assert!(ops.get_repo(user_id).unwrap().is_some());
680 assert!(ops.lookup_handle(&handle).unwrap().is_some());
681
682 ops.delete_repo(ms.database(), user_id).unwrap();
683 assert!(ops.get_repo(user_id).unwrap().is_none());
684 assert!(ops.lookup_handle(&handle).unwrap().is_none());
685 }
686
687 #[test]
688 fn delete_repo_clears_user_hash_mapping() {
689 let (_dir, ms) = open_fresh();
690 let ops = ms.repo_ops();
691 let user_id = uuid::Uuid::new_v4();
692 let did = test_did("mapped");
693 let handle = test_handle("mapped");
694 let cid = test_cid_link(4);
695
696 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev1")
697 .unwrap();
698 assert!(ms.user_hashes().get(&user_id).is_some());
699
700 ops.delete_repo(ms.database(), user_id).unwrap();
701 assert!(ms.user_hashes().get(&user_id).is_none());
702 }
703
704 #[test]
705 fn delete_repo_allows_recreate_with_same_did() {
706 let (_dir, ms) = open_fresh();
707 let ops = ms.repo_ops();
708 let did = test_did("recreate");
709 let handle_a = test_handle("recreate_a");
710 let handle_b = test_handle("recreate_b");
711 let uid_a = uuid::Uuid::new_v4();
712 let uid_b = uuid::Uuid::new_v4();
713 let cid = test_cid_link(70);
714
715 ops.create_repo(ms.database(), uid_a, &did, &handle_a, &cid, "r1")
716 .unwrap();
717 ops.delete_repo(ms.database(), uid_a).unwrap();
718
719 ops.create_repo(ms.database(), uid_b, &did, &handle_b, &cid, "r2")
720 .unwrap();
721
722 let repo = ops.get_repo(uid_b).unwrap().unwrap();
723 assert_eq!(repo.user_id, uid_b);
724 assert_eq!(repo.repo_rev.as_deref(), Some("r2"));
725 assert!(ops.get_repo(uid_a).unwrap().is_none());
726 }
727
728 #[test]
729 fn delete_nonexistent_user_returns_error() {
730 let (_dir, ms) = open_fresh();
731 let ops = ms.repo_ops();
732 let result = ops.delete_repo(ms.database(), uuid::Uuid::new_v4());
733 assert!(matches!(result, Err(MetastoreError::InvalidInput(_))));
734 }
735
736 #[test]
737 fn handle_lookup_case_insensitive() {
738 let (_dir, ms) = open_fresh();
739 let ops = ms.repo_ops();
740 let user_id = uuid::Uuid::new_v4();
741 let did = test_did("bailey");
742 let handle = test_handle("bailey");
743 let cid = test_cid_link(5);
744
745 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev1")
746 .unwrap();
747
748 let upper_handle = Handle::from("BAILEY.TEST.INVALID".to_string());
749 let found = ops.lookup_handle(&upper_handle).unwrap();
750 assert_eq!(found, Some(user_id));
751 }
752
753 #[test]
754 fn get_repo_root_by_did() {
755 let (_dir, ms) = open_fresh();
756 let ops = ms.repo_ops();
757 let user_id = uuid::Uuid::new_v4();
758 let did = test_did("olaren");
759 let handle = test_handle("olaren");
760 let cid = test_cid_link(6);
761
762 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev1")
763 .unwrap();
764
765 let root = ops.get_repo_root_by_did(&did).unwrap().unwrap();
766 assert_eq!(root, cid);
767
768 let unknown = test_did("nonexistent");
769 assert!(ops.get_repo_root_by_did(&unknown).unwrap().is_none());
770 }
771
772 #[test]
773 fn get_repo_root_for_update() {
774 let (_dir, ms) = open_fresh();
775 let ops = ms.repo_ops();
776 let user_id = uuid::Uuid::new_v4();
777 let did = test_did("teq");
778 let handle = test_handle("teq");
779 let cid = test_cid_link(7);
780
781 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev1")
782 .unwrap();
783
784 let root = ops.get_repo_root_for_update(user_id).unwrap().unwrap();
785 assert_eq!(root, cid);
786 }
787
788 #[test]
789 fn count_repos() {
790 let (_dir, ms) = open_fresh();
791 let ops = ms.repo_ops();
792
793 assert_eq!(ops.count_repos().unwrap(), 0);
794
795 (0..5u8).for_each(|i| {
796 let user_id = uuid::Uuid::new_v4();
797 let did = test_did(&format!("user{i}"));
798 let handle = test_handle(&format!("user{i}"));
799 let cid = test_cid_link(i);
800 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev1")
801 .unwrap();
802 });
803
804 assert_eq!(ops.count_repos().unwrap(), 5);
805 }
806
807 #[test]
808 fn get_account_with_repo() {
809 let (_dir, ms) = open_fresh();
810 let ops = ms.repo_ops();
811 let user_id = uuid::Uuid::new_v4();
812 let did = test_did("nel");
813 let handle = test_handle("nel");
814 let cid = test_cid_link(8);
815
816 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev1")
817 .unwrap();
818
819 let account = ops.get_account_with_repo(&did).unwrap().unwrap();
820 assert_eq!(account.user_id, user_id);
821 assert_eq!(account.did, did);
822 assert_eq!(account.repo_root_cid, Some(cid));
823 assert!(account.deactivated_at.is_none());
824 assert!(account.takedown_ref.is_none());
825 }
826
827 #[test]
828 fn list_repos_paginated_all() {
829 let (_dir, ms) = open_fresh();
830 let ops = ms.repo_ops();
831
832 (0..5u8).for_each(|i| {
833 let user_id = uuid::Uuid::new_v4();
834 let did = test_did(&format!("page{i}"));
835 let handle = test_handle(&format!("page{i}"));
836 let cid = test_cid_link(10 + i);
837 ops.create_repo(
838 ms.database(),
839 user_id,
840 &did,
841 &handle,
842 &cid,
843 &format!("rev{i}"),
844 )
845 .unwrap();
846 });
847
848 let all = ops.list_repos_paginated(None, 100).unwrap();
849 assert_eq!(all.len(), 5);
850
851 all.iter()
852 .zip(all.iter().skip(1))
853 .for_each(|(a, b)| assert!(a.user_hash.raw() < b.user_hash.raw()));
854 }
855
856 #[test]
857 fn list_repos_paginated_with_cursor() {
858 let (_dir, ms) = open_fresh();
859 let ops = ms.repo_ops();
860
861 (0..10u8).for_each(|i| {
862 let user_id = uuid::Uuid::new_v4();
863 let did = test_did(&format!("cursor{i}"));
864 let handle = test_handle(&format!("cursor{i}"));
865 let cid = test_cid_link(20 + i);
866 ops.create_repo(
867 ms.database(),
868 user_id,
869 &did,
870 &handle,
871 &cid,
872 &format!("rev{i}"),
873 )
874 .unwrap();
875 });
876
877 let page1 = ops.list_repos_paginated(None, 3).unwrap();
878 assert_eq!(page1.len(), 3);
879
880 let cursor = page1.last().unwrap().user_hash.raw();
881 let page2 = ops.list_repos_paginated(Some(cursor), 3).unwrap();
882 assert_eq!(page2.len(), 3);
883
884 assert!(page2.first().unwrap().user_hash.raw() > cursor);
885
886 let page2_cursor = page2.last().unwrap().user_hash.raw();
887 let page3 = ops.list_repos_paginated(Some(page2_cursor), 100).unwrap();
888 assert_eq!(page3.len(), 4);
889
890 let total = page1.len() + page2.len() + page3.len();
891 assert_eq!(total, 10);
892 }
893
894 #[test]
895 fn data_survives_reopen() {
896 let dir = tempfile::TempDir::new().unwrap();
897 let user_id = uuid::Uuid::new_v4();
898 let did = test_did("persist");
899 let handle = test_handle("persist");
900 let cid = test_cid_link(99);
901
902 {
903 let ms = Metastore::open(dir.path(), test_config()).unwrap();
904 let ops = ms.repo_ops();
905 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev_persist")
906 .unwrap();
907 ms.persist().unwrap();
908 }
909
910 {
911 let ms = Metastore::open(dir.path(), test_config()).unwrap();
912 let ops = ms.repo_ops();
913 let repo = ops.get_repo(user_id).unwrap().unwrap();
914 assert_eq!(repo.repo_root_cid, cid);
915 assert_eq!(repo.repo_rev.as_deref(), Some("rev_persist"));
916
917 let found = ops.lookup_handle(&handle).unwrap();
918 assert_eq!(found, Some(user_id));
919 }
920 }
921
922 #[test]
923 fn get_repos_without_rev() {
924 let (_dir, ms) = open_fresh();
925 let ops = ms.repo_ops();
926
927 let uid_with = uuid::Uuid::new_v4();
928 let did_with = test_did("with_rev");
929 let handle_with = test_handle("with_rev");
930 ops.create_repo(
931 ms.database(),
932 uid_with,
933 &did_with,
934 &handle_with,
935 &test_cid_link(40),
936 "some_rev",
937 )
938 .unwrap();
939
940 let uid_without = uuid::Uuid::new_v4();
941 let did_without = test_did("without_rev");
942 let handle_without = test_handle("without_rev");
943 ops.create_repo(
944 ms.database(),
945 uid_without,
946 &did_without,
947 &handle_without,
948 &test_cid_link(41),
949 "",
950 )
951 .unwrap();
952
953 let result = ops.get_repos_without_rev(100).unwrap();
954 assert_eq!(result.len(), 1);
955 assert_eq!(result[0].user_id, uid_without);
956 }
957
958 #[test]
959 fn get_repo_root_cid_by_user_id() {
960 let (_dir, ms) = open_fresh();
961 let ops = ms.repo_ops();
962 let user_id = uuid::Uuid::new_v4();
963 let did = test_did("root_cid");
964 let handle = test_handle("root_cid");
965 let cid = test_cid_link(50);
966
967 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev1")
968 .unwrap();
969
970 let root = ops.get_repo_root_cid_by_user_id(user_id).unwrap().unwrap();
971 assert_eq!(root, cid);
972
973 assert!(
974 ops.get_repo_root_cid_by_user_id(uuid::Uuid::new_v4())
975 .unwrap()
976 .is_none()
977 );
978 }
979
980 #[test]
981 fn delete_only_removes_target_handle() {
982 let (_dir, ms) = open_fresh();
983 let ops = ms.repo_ops();
984
985 let uid_a = uuid::Uuid::new_v4();
986 let did_a = test_did("keep_a");
987 let handle_a = test_handle("keep_a");
988 ops.create_repo(
989 ms.database(),
990 uid_a,
991 &did_a,
992 &handle_a,
993 &test_cid_link(60),
994 "r",
995 )
996 .unwrap();
997
998 let uid_b = uuid::Uuid::new_v4();
999 let did_b = test_did("delete_b");
1000 let handle_b = test_handle("delete_b");
1001 ops.create_repo(
1002 ms.database(),
1003 uid_b,
1004 &did_b,
1005 &handle_b,
1006 &test_cid_link(61),
1007 "r",
1008 )
1009 .unwrap();
1010
1011 ops.delete_repo(ms.database(), uid_b).unwrap();
1012
1013 assert!(ops.lookup_handle(&handle_a).unwrap().is_some());
1014 assert!(ops.lookup_handle(&handle_b).unwrap().is_none());
1015 assert!(ops.get_repo(uid_a).unwrap().is_some());
1016 }
1017
1018 #[test]
1019 fn create_repo_rejects_hash_collision() {
1020 let (_dir, ms) = open_fresh();
1021 let ops = ms.repo_ops();
1022
1023 let uid_a = uuid::Uuid::new_v4();
1024 let did_a = test_did("collision_a");
1025 let handle_a = test_handle("collision_a");
1026 let cid = test_cid_link(80);
1027
1028 ops.create_repo(ms.database(), uid_a, &did_a, &handle_a, &cid, "r1")
1029 .unwrap();
1030
1031 let uid_b = uuid::Uuid::new_v4();
1032 let handle_b = test_handle("collision_b");
1033
1034 let result = ops.create_repo(ms.database(), uid_b, &did_a, &handle_b, &cid, "r2");
1035
1036 match result {
1037 Ok(()) => {
1038 let repo = ops.get_repo(uid_a).unwrap().unwrap();
1039 assert_eq!(repo.repo_root_cid, cid);
1040 }
1041 Err(MetastoreError::UserHashCollision { .. }) => {
1042 let repo = ops.get_repo(uid_a).unwrap().unwrap();
1043 assert_eq!(repo.repo_root_cid, cid);
1044 assert_eq!(repo.repo_rev.as_deref(), Some("r1"));
1045 }
1046 Err(e) => panic!("unexpected error: {e}"),
1047 }
1048 }
1049
1050 #[test]
1051 fn get_repo_meta_returns_raw_value() {
1052 let (_dir, ms) = open_fresh();
1053 let ops = ms.repo_ops();
1054 let user_id = uuid::Uuid::new_v4();
1055 let did = test_did("meta_raw");
1056 let handle = test_handle("meta_raw");
1057 let cid = test_cid_link(81);
1058
1059 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "rev_meta")
1060 .unwrap();
1061
1062 let (user_hash, value) = ops.get_repo_meta(user_id).unwrap().unwrap();
1063 assert_eq!(user_hash, UserHash::from_did(did.as_str()));
1064 assert_eq!(value.repo_rev, "rev_meta");
1065 assert_eq!(value.handle, "meta_raw.test.invalid");
1066 assert_eq!(value.status, RepoStatus::Active);
1067 }
1068
1069 #[test]
1070 fn get_repo_meta_returns_none_for_unknown() {
1071 let (_dir, ms) = open_fresh();
1072 let ops = ms.repo_ops();
1073 assert!(ops.get_repo_meta(uuid::Uuid::new_v4()).unwrap().is_none());
1074 }
1075
1076 #[test]
1077 fn write_repo_meta_via_batch() {
1078 let (_dir, ms) = open_fresh();
1079 let ops = ms.repo_ops();
1080 let user_id = uuid::Uuid::new_v4();
1081 let did = test_did("batch_write");
1082 let handle = test_handle("batch_write");
1083 let cid1 = test_cid_link(82);
1084 let cid2 = test_cid_link(83);
1085
1086 ops.create_repo(ms.database(), user_id, &did, &handle, &cid1, "rev1")
1087 .unwrap();
1088
1089 let (user_hash, mut value) = ops.get_repo_meta(user_id).unwrap().unwrap();
1090 value.repo_root_cid = cid_link_to_bytes(&cid2).unwrap();
1091 value.repo_rev = "rev2".to_string();
1092
1093 let mut batch = ms.database().batch();
1094 ops.write_repo_meta(&mut batch, user_hash, &value);
1095 batch.commit().unwrap();
1096
1097 let repo = ops.get_repo(user_id).unwrap().unwrap();
1098 assert_eq!(repo.repo_root_cid, cid2);
1099 assert_eq!(repo.repo_rev.as_deref(), Some("rev2"));
1100 }
1101
1102 #[test]
1103 fn update_handle_swaps_lookup() {
1104 let (_dir, ms) = open_fresh();
1105 let ops = ms.repo_ops();
1106 let user_id = uuid::Uuid::new_v4();
1107 let did = test_did("handle_swap");
1108 let old_handle = test_handle("old_name");
1109 let new_handle = test_handle("new_name");
1110 let cid = test_cid_link(84);
1111
1112 ops.create_repo(ms.database(), user_id, &did, &old_handle, &cid, "r1")
1113 .unwrap();
1114 assert!(ops.lookup_handle(&old_handle).unwrap().is_some());
1115
1116 ops.update_handle(ms.database(), user_id, &new_handle)
1117 .unwrap();
1118
1119 assert!(ops.lookup_handle(&old_handle).unwrap().is_none());
1120 assert_eq!(ops.lookup_handle(&new_handle).unwrap(), Some(user_id));
1121
1122 let (_, meta) = ops.get_repo_meta(user_id).unwrap().unwrap();
1123 assert_eq!(meta.handle, "new_name.test.invalid");
1124 }
1125
1126 #[test]
1127 fn update_handle_case_insensitive() {
1128 let (_dir, ms) = open_fresh();
1129 let ops = ms.repo_ops();
1130 let user_id = uuid::Uuid::new_v4();
1131 let did = test_did("handle_case");
1132 let handle = test_handle("original");
1133 let cid = test_cid_link(85);
1134
1135 ops.create_repo(ms.database(), user_id, &did, &handle, &cid, "r1")
1136 .unwrap();
1137
1138 let mixed_case = Handle::from("UPPER.TEST.INVALID".to_string());
1139 ops.update_handle(ms.database(), user_id, &mixed_case)
1140 .unwrap();
1141
1142 let lower_lookup = Handle::from("upper.test.invalid".to_string());
1143 assert_eq!(ops.lookup_handle(&lower_lookup).unwrap(), Some(user_id));
1144 assert!(ops.lookup_handle(&handle).unwrap().is_none());
1145 }
1146}