ptr.pet
at protocol indexer with flexible filtering, xrpc queries, and a cursor-backed event stream, built on fjall
at-protocol
atproto
indexer
rust
fjall
[ingest] handle all account statuses
+61
-1
src/buffer/processor.rs
+61
-1
src/buffer/processor.rs
···
103
103
};
104
104
ops::emit_identity_event(&self.state.db, evt);
105
105
}
106
-
SubscribeReposMessage::Account(ref account) => {
106
+
SubscribeReposMessage::Account(account) => {
107
107
debug!("processing buffered account for {did}");
108
108
let evt = AccountEvt {
109
109
did: did.to_smolstr(),
···
111
111
status: account.status.as_ref().map(|s| s.to_smolstr()),
112
112
};
113
113
ops::emit_account_event(&self.state.db, evt);
114
+
115
+
let state = self.state.clone();
116
+
let did = did.clone();
117
+
let account = account.clone(); // Account is 'static in BufferedMessage
118
+
119
+
tokio::task::spawn_blocking(move || -> Result<()> {
120
+
// handle status updates
121
+
if !account.active {
122
+
use jacquard::api::com_atproto::sync::subscribe_repos::AccountStatus;
123
+
if let Some(status) = &account.status {
124
+
match status {
125
+
AccountStatus::Deleted => {
126
+
info!("account {did} deleted, wiping data");
127
+
ops::delete_repo(&state.db, &did)?;
128
+
}
129
+
AccountStatus::Takendown => {
130
+
ops::update_repo_status(
131
+
&state.db,
132
+
&did,
133
+
crate::types::RepoStatus::Takendown,
134
+
)?;
135
+
}
136
+
AccountStatus::Suspended => {
137
+
ops::update_repo_status(
138
+
&state.db,
139
+
&did,
140
+
crate::types::RepoStatus::Suspended,
141
+
)?;
142
+
}
143
+
AccountStatus::Deactivated => {
144
+
ops::update_repo_status(
145
+
&state.db,
146
+
&did,
147
+
crate::types::RepoStatus::Deactivated,
148
+
)?;
149
+
}
150
+
AccountStatus::Throttled | AccountStatus::Desynchronized => {
151
+
let status_str = status.as_str().to_smolstr();
152
+
ops::update_repo_status(
153
+
&state.db,
154
+
&did,
155
+
crate::types::RepoStatus::Error(status_str),
156
+
)?;
157
+
}
158
+
AccountStatus::Other(s) => {
159
+
warn!("unknown account status for {did}: {s}");
160
+
}
161
+
}
162
+
} else {
163
+
warn!("account {did} inactive but no status provided");
164
+
}
165
+
} else {
166
+
// active account, clear any error/suspension states if they exist
167
+
// we set it to Synced because we are receiving live events for it
168
+
ops::update_repo_status(&state.db, &did, crate::types::RepoStatus::Synced)?;
169
+
}
170
+
Ok(())
171
+
})
172
+
.await
173
+
.into_diagnostic()??;
114
174
}
115
175
_ => {
116
176
warn!("unknown message type in buffer for {did}");
+19
src/ops.rs
+19
src/ops.rs
···
87
87
Ok(())
88
88
}
89
89
90
+
pub fn update_repo_status(
91
+
db: &Db,
92
+
did: &jacquard::types::did::Did,
93
+
status: crate::types::RepoStatus,
94
+
) -> Result<()> {
95
+
debug!("updating repo status for {did} to {status:?}");
96
+
let (updated, mut batch) =
97
+
Db::update_repo_state(db.inner.batch(), &db.repos, did, |state, _val| {
98
+
state.status = status.clone();
99
+
state.last_updated_at = chrono::Utc::now().timestamp();
100
+
Ok((true, ()))
101
+
})?;
102
+
103
+
if updated.is_some() {
104
+
batch.commit().into_diagnostic()?;
105
+
}
106
+
Ok(())
107
+
}
108
+
90
109
pub fn apply_commit(db: &Db, commit: &Commit<'_>, live: bool) -> Result<()> {
91
110
let did = &commit.repo;
92
111
debug!("applying commit {} for {did}", &commit.commit);