lightweight
com.atproto.sync.listReposByCollection
sketch out index queries + fix status types
+96
-18
+53
-1
src/storage/index.rs
+53
-1
src/storage/index.rs
···
5
5
6
6
use crate::error::Result;
7
7
use crate::storage::{DbRef, keys};
8
+
use crate::sync::resync::RepoSnapshot;
8
9
9
10
/// Insert a `(collection, did)` pair into both the rbc and cbr indexes.
10
11
pub fn insert(db: &DbRef, did: Did<'_>, collection: Nsid<'_>) -> Result<()> {
···
24
25
Ok(())
25
26
}
26
27
27
-
/// Remove all index entries for `did` (account deletion / full resync).
28
+
/// Remove all index entries for `did` (account deletion).
28
29
///
29
30
/// Reads all collections from the cbr index, then deletes both sides in a batch.
30
31
pub fn remove_all(db: &DbRef, did: Did<'_>) -> Result<()> {
···
54
55
/// Iterate over DIDs in the rbc index for `collection`, starting after `cursor`.
55
56
///
56
57
/// Returns at most `limit` DIDs.
58
+
///
59
+
/// TODO: we can fjall range to the collection's next-after-max (might even be
60
+
/// exposed now?) or maybe use prefix + seek for the start?
57
61
pub fn scan_rbc(
58
62
db: &DbRef,
59
63
collection: Nsid<'_>,
···
90
94
}
91
95
Ok(dids)
92
96
}
97
+
98
+
/// Iterate over DIDs in the rbc index for `collection`, starting after `cursor`.
99
+
///
100
+
/// Returns at most `limit` DIDs.
101
+
///
102
+
/// TODO: we can fjall range to the collection's next-after-max (might even be
103
+
/// exposed now?) or maybe use prefix + seek for the start?
104
+
pub fn scan_cbr(
105
+
db: &DbRef,
106
+
did: Did<'_>,
107
+
cursor: Option<Nsid<'_>>,
108
+
limit: usize,
109
+
) -> Result<Vec<Nsid<'static>>> {
110
+
let prefix = keys::cbr_prefix(did.clone());
111
+
let prefix_len = prefix.len();
112
+
113
+
let start_key: Vec<u8> = match cursor {
114
+
None => prefix.clone(),
115
+
Some(ref col) => keys::cbr(did.clone(), col.clone()),
116
+
};
117
+
118
+
let mut cols = Vec::with_capacity(limit);
119
+
for guard in db.cbr.range(start_key..) {
120
+
let (k, _v) = guard.into_inner()?;
121
+
if !k.starts_with(&prefix) {
122
+
break;
123
+
}
124
+
if cols.len() >= limit {
125
+
break;
126
+
}
127
+
// Skip the cursor key itself.
128
+
if cursor
129
+
.clone()
130
+
.is_some_and(|c| k.as_ref() == keys::cbr(did.clone(), c).as_slice())
131
+
{
132
+
continue;
133
+
}
134
+
if let Some(col) = keys::cbr_parse_collection(&k, prefix_len) {
135
+
cols.push(col);
136
+
}
137
+
}
138
+
Ok(cols)
139
+
}
140
+
141
+
pub fn sync_repo(db: &DbRef, did: Did<'_>, snapshot: RepoSnapshot) -> Result<()> {
142
+
// TODO
143
+
Ok(())
144
+
}
+43
-17
src/storage/repo.rs
+43
-17
src/storage/repo.rs
···
5
5
use crate::error::Result;
6
6
use crate::storage::{DbRef, keys};
7
7
8
-
/// High-level lifecycle state of a repo.
8
+
/// tap's "RepoState" type
9
9
#[derive(Debug, Clone, PartialEq, Eq)]
10
10
pub enum RepoState {
11
-
/// Repo is being indexed for the first time.
12
-
Indexing,
13
-
/// Repo is fully indexed and up to date.
11
+
Pending,
12
+
Desynchronized,
13
+
Resyncing,
14
14
Active,
15
-
/// Repo has been tombstoned / deactivated.
16
-
Tombstoned,
15
+
Takendown,
16
+
Suspended,
17
+
Deactivated,
18
+
Error,
17
19
}
18
20
19
-
/// Processing status used to drive backfill / resync decisions.
21
+
impl RepoState {
22
+
pub fn as_str(&self) -> &str {
23
+
match self {
24
+
RepoState::Pending => "pending",
25
+
RepoState::Desynchronized => "desynchronized",
26
+
RepoState::Resyncing => "resyncing",
27
+
RepoState::Active => "active",
28
+
RepoState::Takendown => "takendown",
29
+
RepoState::Suspended => "suspended",
30
+
RepoState::Deactivated => "deactivated",
31
+
RepoState::Error => "error",
32
+
}
33
+
}
34
+
}
35
+
36
+
/// tap's "AccountStatus" type
20
37
#[derive(Debug, Clone, PartialEq, Eq)]
21
-
pub enum RepoStatus {
22
-
/// No record yet.
23
-
Unknown,
24
-
/// Queued for backfill probing.
25
-
Pending,
26
-
/// Backfill complete.
27
-
Complete,
28
-
/// An error occurred; `error` field has details.
29
-
Error,
38
+
pub enum AccountStatus {
39
+
Active,
40
+
Takendown,
41
+
Suspended,
42
+
Deactivated,
43
+
Deleted,
44
+
}
45
+
46
+
impl AccountStatus {
47
+
pub fn as_str(&self) -> &str {
48
+
match self {
49
+
AccountStatus::Active => "active",
50
+
AccountStatus::Takendown => "takendown",
51
+
AccountStatus::Suspended => "suspended",
52
+
AccountStatus::Deactivated => "deactivated",
53
+
AccountStatus::Deleted => "deleted",
54
+
}
55
+
}
30
56
}
31
57
32
58
/// Stored record for a repository.
33
59
#[derive(Debug, Clone)]
34
60
pub struct RepoRecord {
35
61
pub state: RepoState,
36
-
pub status: RepoStatus,
62
+
pub status: AccountStatus,
37
63
pub error: Option<String>,
38
64
}
39
65