goat.navy
forked from
tranquil.farm/tranquil-pds
Our Personal Data Server from scratch!
14 kB
564 lines
1use async_trait::async_trait;
2use chrono::{DateTime, Utc};
3use serde::{Deserialize, Serialize};
4use tranquil_types::{AtUri, CidLink, Did, Handle, Nsid, Rkey, Tid};
5use uuid::Uuid;
6
7use crate::DbError;
8use crate::backlink::Backlink;
9use crate::sequence::SequenceNumber;
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, sqlx::Type)]
12#[sqlx(type_name = "text", rename_all = "snake_case")]
13#[serde(rename_all = "snake_case")]
14pub enum RepoEventType {
15 Commit,
16 Identity,
17 Account,
18 Sync,
19}
20
21impl RepoEventType {
22 pub fn as_str(&self) -> &'static str {
23 match self {
24 Self::Commit => "commit",
25 Self::Identity => "identity",
26 Self::Account => "account",
27 Self::Sync => "sync",
28 }
29 }
30}
31
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, sqlx::Type)]
33#[sqlx(type_name = "text", rename_all = "lowercase")]
34#[serde(rename_all = "lowercase")]
35pub enum AccountStatus {
36 Active,
37 Takendown,
38 Suspended,
39 Deactivated,
40 Deleted,
41}
42
43impl AccountStatus {
44 pub fn as_str(&self) -> &'static str {
45 match self {
46 Self::Active => "active",
47 Self::Takendown => "takendown",
48 Self::Suspended => "suspended",
49 Self::Deactivated => "deactivated",
50 Self::Deleted => "deleted",
51 }
52 }
53
54 pub fn for_firehose(&self) -> Option<Self> {
55 match self {
56 Self::Active => None,
57 other => Some(*other),
58 }
59 }
60
61 pub fn parse(s: &str) -> Option<Self> {
62 match s.to_lowercase().as_str() {
63 "active" => Some(Self::Active),
64 "takendown" => Some(Self::Takendown),
65 "suspended" => Some(Self::Suspended),
66 "deactivated" => Some(Self::Deactivated),
67 "deleted" => Some(Self::Deleted),
68 _ => None,
69 }
70 }
71
72 pub fn is_active(&self) -> bool {
73 matches!(self, Self::Active)
74 }
75
76 pub fn is_takendown(&self) -> bool {
77 matches!(self, Self::Takendown)
78 }
79
80 pub fn is_deactivated(&self) -> bool {
81 matches!(self, Self::Deactivated)
82 }
83
84 pub fn is_suspended(&self) -> bool {
85 matches!(self, Self::Suspended)
86 }
87
88 pub fn is_deleted(&self) -> bool {
89 matches!(self, Self::Deleted)
90 }
91
92 pub fn allows_read(&self) -> bool {
93 matches!(self, Self::Active | Self::Deactivated)
94 }
95
96 pub fn allows_write(&self) -> bool {
97 matches!(self, Self::Active)
98 }
99
100 pub fn from_db_fields(
101 takedown_ref: Option<&str>,
102 deactivated_at: Option<DateTime<Utc>>,
103 ) -> Self {
104 if takedown_ref.is_some() {
105 Self::Takendown
106 } else if deactivated_at.is_some() {
107 Self::Deactivated
108 } else {
109 Self::Active
110 }
111 }
112}
113
114impl std::fmt::Display for AccountStatus {
115 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
116 f.write_str(self.as_str())
117 }
118}
119
120#[derive(Debug, Clone, Serialize, Deserialize)]
121pub struct RepoAccountInfo {
122 pub user_id: Uuid,
123 pub did: Did,
124 pub deactivated_at: Option<DateTime<Utc>>,
125 pub takedown_ref: Option<String>,
126 pub repo_root_cid: Option<CidLink>,
127}
128
129#[derive(Debug, Clone, Serialize, Deserialize)]
130pub struct RepoInfo {
131 pub user_id: Uuid,
132 pub repo_root_cid: CidLink,
133 pub repo_rev: Option<Tid>,
134}
135
136#[derive(Debug, Clone, Serialize, Deserialize)]
137pub struct RecordInfo {
138 pub rkey: Rkey,
139 pub record_cid: CidLink,
140}
141
142#[derive(Debug, Clone, Serialize, Deserialize)]
143pub struct FullRecordInfo {
144 pub collection: Nsid,
145 pub rkey: Rkey,
146 pub record_cid: CidLink,
147}
148
149#[derive(Debug, Clone, Serialize, Deserialize)]
150pub struct RecordWithTakedown {
151 pub id: Uuid,
152 pub takedown_ref: Option<String>,
153}
154
155#[derive(Debug, Clone, Serialize, Deserialize)]
156pub struct RepoWithoutRev {
157 pub user_id: Uuid,
158 pub repo_root_cid: CidLink,
159}
160
161#[derive(Debug, Clone)]
162pub struct UserWithoutBlocks {
163 pub user_id: Uuid,
164 pub repo_root_cid: CidLink,
165 pub repo_rev: Option<Tid>,
166}
167
168#[derive(Debug, Clone)]
169pub struct UserNeedingRecordBlobsBackfill {
170 pub user_id: Uuid,
171 pub did: Did,
172}
173
174#[derive(Debug, Clone, Serialize, Deserialize)]
175pub struct RepoSeqEvent {
176 pub seq: SequenceNumber,
177}
178
179#[derive(Debug, Clone, Copy, PartialEq, Eq)]
180pub enum PruneCount {
181 Rows(u64),
182 Segments(u64),
183}
184
185impl PruneCount {
186 pub fn is_zero(&self) -> bool {
187 match self {
188 Self::Rows(n) | Self::Segments(n) => *n == 0,
189 }
190 }
191
192 pub fn count(&self) -> u64 {
193 match self {
194 Self::Rows(n) | Self::Segments(n) => *n,
195 }
196 }
197
198 pub fn unit(&self) -> &'static str {
199 match self {
200 Self::Rows(_) => "rows",
201 Self::Segments(_) => "segments",
202 }
203 }
204}
205
206impl std::fmt::Display for PruneCount {
207 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
208 write!(f, "{} {}", self.count(), self.unit())
209 }
210}
211
212#[derive(Debug, Clone, Serialize, Deserialize)]
213pub struct EventBlockInline {
214 pub cid_bytes: Vec<u8>,
215 pub data: Vec<u8>,
216}
217
218#[derive(Debug, Clone, Serialize, Deserialize)]
219pub enum EventBlocks {
220 Inline(Vec<EventBlockInline>),
221 LegacyCids(Vec<CidLink>),
222}
223
224#[derive(Debug, Clone, Serialize, Deserialize)]
225pub struct SequencedEvent {
226 pub seq: SequenceNumber,
227 pub did: Did,
228 pub created_at: DateTime<Utc>,
229 pub event_type: RepoEventType,
230 pub commit_cid: Option<CidLink>,
231 pub prev_cid: Option<CidLink>,
232 pub prev_data_cid: Option<CidLink>,
233 pub ops: Option<serde_json::Value>,
234 pub blobs: Option<Vec<CidLink>>,
235 pub blocks: Option<EventBlocks>,
236 pub handle: Option<Handle>,
237 pub active: Option<bool>,
238 pub status: Option<AccountStatus>,
239 pub rev: Option<Tid>,
240}
241
242#[derive(Debug, Clone)]
243pub struct CommitEventData {
244 pub did: Did,
245 pub event_type: RepoEventType,
246 pub commit_cid: Option<CidLink>,
247 pub prev_cid: Option<CidLink>,
248 pub ops: Option<serde_json::Value>,
249 pub blobs: Option<Vec<CidLink>>,
250 pub blocks: Option<Vec<EventBlockInline>>,
251 pub prev_data_cid: Option<CidLink>,
252 pub rev: Option<Tid>,
253}
254
255#[derive(Debug, Clone, Serialize, Deserialize)]
256pub struct RepoListItem {
257 pub did: Did,
258 pub deactivated_at: Option<DateTime<Utc>>,
259 pub takedown_ref: Option<String>,
260 pub repo_root_cid: CidLink,
261 pub repo_rev: Option<Tid>,
262}
263
264#[derive(Debug, Clone)]
265pub struct ImportBlock {
266 pub cid_bytes: Vec<u8>,
267 pub data: Vec<u8>,
268}
269
270#[derive(Debug, Clone)]
271pub struct ImportRecord {
272 pub collection: Nsid,
273 pub rkey: Rkey,
274 pub record_cid: CidLink,
275}
276
277#[derive(Debug, Clone, PartialEq, Eq)]
278pub enum ImportRepoError {
279 RepoNotFound,
280 ConcurrentModification,
281 Database(String),
282}
283
284#[derive(Debug, Clone)]
285pub struct RecordUpsert {
286 pub collection: Nsid,
287 pub rkey: Rkey,
288 pub cid: CidLink,
289}
290
291#[derive(Debug, Clone)]
292pub struct RecordDelete {
293 pub collection: Nsid,
294 pub rkey: Rkey,
295}
296
297#[derive(Debug, Clone)]
298pub struct ApplyCommitInput {
299 pub user_id: Uuid,
300 pub did: Did,
301 pub expected_root_cid: Option<CidLink>,
302 pub new_root_cid: CidLink,
303 pub new_rev: Tid,
304 pub new_block_cids: Vec<Vec<u8>>,
305 pub obsolete_block_cids: Vec<Vec<u8>>,
306 pub record_upserts: Vec<RecordUpsert>,
307 pub record_deletes: Vec<RecordDelete>,
308 pub backlinks_to_add: Vec<Backlink>,
309 pub backlinks_to_remove: Vec<AtUri>,
310 pub commit_event: CommitEventData,
311}
312
313#[derive(Debug, Clone)]
314pub struct ApplyCommitResult {
315 pub is_account_active: bool,
316}
317
318#[derive(Debug, Clone, PartialEq, Eq)]
319pub enum ApplyCommitError {
320 RepoNotFound,
321 ConcurrentModification,
322 Database(String),
323}
324
325#[async_trait]
326pub trait RepoRepository: Send + Sync {
327 async fn create_repo(
328 &self,
329 user_id: Uuid,
330 did: &Did,
331 handle: &Handle,
332 repo_root_cid: &CidLink,
333 repo_rev: &Tid,
334 ) -> Result<(), DbError>;
335
336 async fn update_repo_root(
337 &self,
338 user_id: Uuid,
339 repo_root_cid: &CidLink,
340 repo_rev: &Tid,
341 ) -> Result<(), DbError>;
342
343 async fn update_repo_rev(&self, user_id: Uuid, repo_rev: &Tid) -> Result<(), DbError>;
344
345 async fn update_repo_status(
346 &self,
347 did: &Did,
348 takedown: Option<bool>,
349 takedown_ref: Option<&str>,
350 deactivated: Option<bool>,
351 ) -> Result<(), DbError>;
352
353 async fn delete_repo(&self, user_id: Uuid) -> Result<(), DbError>;
354
355 async fn get_repo_root_for_update(&self, user_id: Uuid) -> Result<Option<CidLink>, DbError>;
356
357 async fn get_repo(&self, user_id: Uuid) -> Result<Option<RepoInfo>, DbError>;
358
359 async fn get_repo_root_by_did(&self, did: &Did) -> Result<Option<CidLink>, DbError>;
360
361 async fn count_repos(&self) -> Result<i64, DbError>;
362
363 async fn get_repos_without_rev(&self) -> Result<Vec<RepoWithoutRev>, DbError>;
364
365 async fn upsert_records(
366 &self,
367 repo_id: Uuid,
368 collections: &[Nsid],
369 rkeys: &[Rkey],
370 record_cids: &[CidLink],
371 repo_rev: &Tid,
372 ) -> Result<(), DbError>;
373
374 async fn delete_records(
375 &self,
376 repo_id: Uuid,
377 collections: &[Nsid],
378 rkeys: &[Rkey],
379 ) -> Result<(), DbError>;
380
381 async fn delete_all_records(&self, repo_id: Uuid) -> Result<(), DbError>;
382
383 async fn get_record_cid(
384 &self,
385 repo_id: Uuid,
386 collection: &Nsid,
387 rkey: &Rkey,
388 ) -> Result<Option<CidLink>, DbError>;
389
390 #[allow(clippy::too_many_arguments)]
391 async fn list_records(
392 &self,
393 repo_id: Uuid,
394 collection: &Nsid,
395 cursor: Option<&Rkey>,
396 limit: i64,
397 reverse: bool,
398 rkey_start: Option<&Rkey>,
399 rkey_end: Option<&Rkey>,
400 ) -> Result<Vec<RecordInfo>, DbError>;
401
402 async fn get_all_records(&self, repo_id: Uuid) -> Result<Vec<FullRecordInfo>, DbError>;
403
404 async fn list_collections(&self, repo_id: Uuid) -> Result<Vec<Nsid>, DbError>;
405
406 async fn count_records(&self, repo_id: Uuid) -> Result<i64, DbError>;
407
408 async fn count_all_records(&self) -> Result<i64, DbError>;
409
410 async fn get_record_by_cid(&self, cid: &CidLink)
411 -> Result<Option<RecordWithTakedown>, DbError>;
412
413 async fn referenced_record_cids(
414 &self,
415 repo_id: Uuid,
416 cids: &[CidLink],
417 excluded_keys: &[(&Nsid, &Rkey)],
418 ) -> Result<Vec<CidLink>, DbError>;
419
420 async fn set_record_takedown(
421 &self,
422 cid: &CidLink,
423 takedown_ref: Option<&str>,
424 ) -> Result<(), DbError>;
425
426 async fn insert_user_blocks(
427 &self,
428 user_id: Uuid,
429 block_cids: &[Vec<u8>],
430 repo_rev: &Tid,
431 ) -> Result<(), DbError>;
432
433 async fn delete_user_blocks(
434 &self,
435 user_id: Uuid,
436 block_cids: &[Vec<u8>],
437 ) -> Result<(), DbError>;
438
439 async fn get_user_block_cids_since_rev(
440 &self,
441 user_id: Uuid,
442 since_rev: Option<&Tid>,
443 ) -> Result<Vec<Vec<u8>>, DbError>;
444
445 async fn count_user_blocks(&self, user_id: Uuid) -> Result<i64, DbError>;
446
447 async fn insert_commit_event(&self, data: &CommitEventData) -> Result<(), DbError>;
448
449 async fn insert_identity_event(
450 &self,
451 did: &Did,
452 handle: Option<&Handle>,
453 ) -> Result<(), DbError>;
454
455 async fn insert_account_event(&self, did: &Did, status: AccountStatus) -> Result<(), DbError>;
456
457 async fn insert_sync_event(
458 &self,
459 did: &Did,
460 commit_cid: &CidLink,
461 rev: Option<&Tid>,
462 commit_bytes: &[u8],
463 ) -> Result<(), DbError>;
464
465 async fn insert_genesis_commit_event(
466 &self,
467 did: &Did,
468 commit_cid: &CidLink,
469 mst_root_cid: &CidLink,
470 rev: &Tid,
471 commit_bytes: &[u8],
472 mst_root_bytes: &[u8],
473 ) -> Result<(), DbError>;
474
475 async fn purge_did_events_keeping_latest(&self, did: &Did) -> Result<(), DbError>;
476
477 async fn assign_pending_sequences(&self) -> Result<u64, DbError> {
478 Ok(0)
479 }
480
481 async fn flush_pending_sequences(&self) -> Result<(), DbError> {
482 Ok(())
483 }
484
485 async fn prune_events_older_than(&self, cutoff: DateTime<Utc>) -> Result<PruneCount, DbError>;
486
487 async fn get_max_seq(&self) -> Result<SequenceNumber, DbError>;
488
489 async fn get_min_seq_since(
490 &self,
491 since: DateTime<Utc>,
492 ) -> Result<Option<SequenceNumber>, DbError>;
493
494 async fn get_account_with_repo(&self, did: &Did) -> Result<Option<RepoAccountInfo>, DbError>;
495
496 async fn get_events_since_seq(
497 &self,
498 since_seq: SequenceNumber,
499 limit: Option<i64>,
500 ) -> Result<Vec<SequencedEvent>, DbError>;
501
502 async fn get_events_in_seq_range(
503 &self,
504 start_seq: SequenceNumber,
505 end_seq: SequenceNumber,
506 ) -> Result<Vec<SequencedEvent>, DbError>;
507
508 async fn get_event_by_seq(
509 &self,
510 seq: SequenceNumber,
511 ) -> Result<Option<SequencedEvent>, DbError>;
512
513 async fn get_events_since_cursor(
514 &self,
515 cursor: SequenceNumber,
516 limit: i64,
517 ) -> Result<Vec<SequencedEvent>, DbError>;
518
519 async fn list_repos_paginated(
520 &self,
521 cursor_did: Option<&Did>,
522 limit: i64,
523 ) -> Result<Vec<RepoListItem>, DbError>;
524
525 async fn get_repo_root_cid_by_user_id(&self, user_id: Uuid)
526 -> Result<Option<CidLink>, DbError>;
527
528 async fn import_repo_data(
529 &self,
530 user_id: Uuid,
531 blocks: &[ImportBlock],
532 records: &[ImportRecord],
533 expected_root_cid: Option<&CidLink>,
534 ) -> Result<(), ImportRepoError>;
535
536 async fn apply_commit(
537 &self,
538 input: ApplyCommitInput,
539 ) -> Result<ApplyCommitResult, ApplyCommitError>;
540
541 async fn get_users_without_blocks(&self) -> Result<Vec<UserWithoutBlocks>, DbError>;
542
543 async fn get_users_needing_record_blobs_backfill(
544 &self,
545 limit: i64,
546 ) -> Result<Vec<UserNeedingRecordBlobsBackfill>, DbError>;
547
548 async fn insert_record_blobs(
549 &self,
550 repo_id: Uuid,
551 record_uris: &[AtUri],
552 blob_cids: &[CidLink],
553 ) -> Result<(), DbError>;
554}
555
556#[async_trait]
557pub trait RepoEventNotifier: Send + Sync {
558 async fn subscribe(&self) -> Result<Box<dyn RepoEventReceiver>, DbError>;
559}
560
561#[async_trait]
562pub trait RepoEventReceiver: Send {
563 async fn recv(&mut self) -> Option<()>;
564}