zzstoatzz.io
Client side atproto account migrator in your web browser, along with services for backups and adversarial migrations.
8.4 kB
216 lines
1use crate::jobs::account_backup::AccountBackupJobContext;
2use crate::jobs::{account_backup, pds_backup, push_job_json};
3use apalis::prelude::{Data, Error};
4use jacquard::{client::BasicClient, identity::PublicResolver, prelude::IdentityResolver, url};
5use jacquard_api::com_atproto::sync::get_repo_status::GetRepoStatus;
6use jacquard_common::{types::did::Did, types::tid::Tid, xrpc::XrpcClient};
7use serde::{Deserialize, Serialize};
8use sqlx::{Pool, Postgres};
9use std::sync::Arc;
10
11pub const JOB_NAMESPACE: &str = "apalis::ScheduledBackUpStart";
12
13#[derive(Debug, Deserialize, Serialize, Clone)]
14pub struct ScheduledBackUpStartJobContext;
15
16/// Finds the new PDS and saves it in the database if the user has changed PDSs since signing up
17async fn find_new_pds_host(pool: &Pool<Postgres>, did: String) -> Option<String> {
18 let resolver = PublicResolver::default();
19 let did_string = did.clone();
20 let did = match Did::new(did.as_str()) {
21 Ok(did) => did,
22 Err(err) => {
23 log::error!("Failed to parse did {did} : {err}");
24 return None;
25 }
26 };
27
28 let did_doc = match resolver.resolve_did_doc_owned(&did).await {
29 Ok(did_doc) => did_doc,
30 Err(err) => {
31 log::error!("Failed to resolve did doc for {did} : {err}");
32 return None;
33 }
34 };
35
36 match did_doc.pds_endpoint() {
37 None => None,
38 Some(url) => match url.host_str() {
39 None => {
40 log::error!("Failed to parse pds_endpoint for {did} : {url}");
41 return None;
42 }
43 Some(host) => {
44 //I may of copied and pasted this to get it going <.< who needs dry in all this snow
45 let result = sqlx::query(
46 "UPDATE accounts SET pds_host = $2 WHERE did = $1 AND pds_host <> $2",
47 )
48 .bind(did_string)
49 .bind(host)
50 .execute(pool)
51 .await;
52 if let Err(e) = result {
53 log::error!("Failed to update pds_host for {did} : {e}");
54 }
55 // Ok(result.rows_affected())
56 Some(host.to_string())
57 }
58 },
59 }
60}
61
62/// This scheduled job finds:
63/// - accounts that have not been backed up in the last 6 hours and have pds_sign_up = false,
64/// and enqueues AccountBackup jobs for them;
65/// - pds_hosts that are active and have not started a backup in the last 6 hours (tracked via
66/// pds_hosts.last_backup_start), and enqueues PdsBackup jobs for each.
67pub async fn scheduled_back_up_start_job(
68 _job: ScheduledBackUpStartJobContext,
69 pool: Data<Pool<Postgres>>,
70) -> Result<(), Error> {
71 log::info!("Starting a backup for the whole instance");
72 // Record the start of a whole-network backup run
73 sqlx::query(r#"INSERT INTO network_backup_runs DEFAULT VALUES"#)
74 .execute(&*pool)
75 .await
76 .map_err(|e| Error::Failed(Arc::new(Box::new(e))))?;
77 // 1) Query accounts needing backup
78 // Condition: pds_sign_up = false AND (last_backup is NULL OR older than 6h)
79 // We include did and pds_host to build AccountBackupJobContext
80 let accounts: Vec<(String, String)> = sqlx::query_as(
81 r#"
82 SELECT did, pds_host
83 FROM accounts
84 WHERE pds_sign_up = FALSE
85 AND (last_backup IS NULL OR last_backup < NOW() - INTERVAL '6 HOURS')
86 "#,
87 )
88 .fetch_all(&*pool)
89 .await
90 .map_err(|e| Error::Failed(Arc::new(Box::new(e))))?;
91
92 //TODO maybe check here?
93 let agent = BasicClient::unauthenticated();
94 for (did, pds_host) in accounts {
95 //This is a holder we can change the PDS if we need to
96 let mut pds_host_to_call = pds_host.clone();
97
98 //Need to do matches here cause I don't want to fail the whole loop
99 let pds_url = match url::Url::parse(&format!("https://{}", pds_host)) {
100 Ok(url) => url,
101 Err(e) => {
102 log::warn!("Failed to parse pds_host: {pds_host} \n {e}");
103 continue;
104 }
105 };
106
107 //Sets the agent to the new PDS URL
108 agent.set_base_uri(pds_url).await;
109
110 let request = GetRepoStatus {
111 //May just leave this ?
112 did: did.clone().parse().map_err(|_| {
113 Error::Failed(Arc::new(Box::new(std::io::Error::new(
114 std::io::ErrorKind::Other,
115 "Failed to parse did",
116 ))))
117 })?,
118 };
119
120 let rev: Option<Tid> = match agent.send(request).await {
121 Ok(response) => match response.parse() {
122 Ok(output) => {
123 log::info!("{output:?}");
124 if output.active {
125 output.rev
126 } else {
127 match output.status {
128 None => {
129 log::warn!(
130 "{did} has no status and not active. Not sure if this happens"
131 );
132 continue;
133 }
134 Some(status) => {
135 if status == "deactivated" {
136 log::info!(
137 "{did} has been deactivated on this PDS. Let's see if they moved."
138 );
139 match find_new_pds_host(&pool, did.clone()).await {
140 None => continue,
141 Some(new_host) => {
142 log::info!(
143 "{did} has moved to {new_host}. Database updated."
144 );
145 pds_host_to_call = new_host;
146 }
147 }
148 //As long as it's successful, we can just send along None for the sub job to pick up the new rev on the new PDS
149 None
150 } else {
151 //TODO will most likely set accounts to deactivated as i collect more data here
152 log::warn!("{did} has the repo status {status}.");
153 continue;
154 }
155 }
156 }
157 }
158 }
159 Err(err) => {
160 log::warn!("Failed to parse GetRepoStatusResponse for: {did} \n {err}");
161 continue;
162 }
163 },
164 Err(err) => {
165 log::warn!("Failed to send GetRepoStatusRequest for: {did} \n {err}");
166 continue;
167 }
168 };
169
170 let rev_string: Option<String> = rev.map(|r| r.to_string());
171
172 let ctx = AccountBackupJobContext {
173 did,
174 pds_host: pds_host_to_call,
175 rev: rev_string,
176 };
177
178 push_job_json(&pool, account_backup::JOB_NAMESPACE, &ctx).await?;
179 }
180
181 // 3) Query pds_hosts needing backup start
182 // Condition: active = TRUE AND (last_backup_start is NULL OR older than 24h)
183 let pds_hosts: Vec<String> = sqlx::query_scalar(
184 r#"
185 SELECT pds_host
186 FROM pds_hosts
187 WHERE active = TRUE
188 AND (last_backup_start IS NULL OR last_backup_start < NOW() - INTERVAL '6 HOURS')
189 "#,
190 )
191 .fetch_all(&*pool)
192 .await
193 .map_err(|e| Error::Failed(Arc::new(Box::new(e))))?;
194
195 // To avoid racing / double-enqueue, mark last_backup_start immediately for those we are about to enqueue
196 if !pds_hosts.is_empty() {
197 sqlx::query(
198 r#"
199 UPDATE pds_hosts
200 SET last_backup_start = NOW()
201 WHERE pds_host = ANY($1)
202 "#,
203 )
204 .bind(&pds_hosts)
205 .execute(&*pool)
206 .await
207 .map_err(|e| Error::Failed(Arc::new(Box::new(e))))?;
208 }
209
210 for host in pds_hosts {
211 let ctx = crate::jobs::pds_backup::PdsBackupJobContext { pds_host: host };
212 push_job_json(&pool, pds_backup::JOB_NAMESPACE, &ctx).await?;
213 }
214
215 Ok(())
216}