1use std::collections::{BTreeMap, BTreeSet};
26use std::time::Duration;
27
28use anyhow::{anyhow, Result};
29use serde::Serialize;
30
31use crate::boards::BoardRef;
32use crate::client::{Conditions, Fetched, GitHubClient, GitHubProject, GitHubViewer};
33use crate::mirror::{gist_is_whole, newest_update, Document, Mirror, Scope, SyncState, SINGLETON};
34use crate::targets::Targets;
35
36pub const SWEEP: Duration = Duration::from_secs(120);
42
43pub const SWEEP_UNCONDITIONAL: Duration = Duration::from_secs(300);
49
50pub const SWEEP_TRAFFIC: Duration = Duration::from_secs(6 * 60 * 60);
54
55pub const AFTER_FAILURE: Duration = Duration::from_secs(600);
59
60#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
62pub struct SyncReport {
63 pub changed: Vec<String>,
65 pub unchanged: Vec<String>,
67 #[serde(serialize_with = "failures_as_objects")]
70 pub failed: Vec<(String, String)>,
71 pub no_token: bool,
74 pub problems: Vec<String>,
76}
77
78impl SyncReport {
79 pub fn touched_anything(&self) -> bool {
80 !self.changed.is_empty()
81 }
82}
83
84fn failures_as_objects<S: serde::Serializer>(
85 failed: &[(String, String)],
86 serializer: S,
87) -> Result<S::Ok, S::Error> {
88 use serde::ser::SerializeSeq;
89 #[derive(Serialize)]
90 struct Failure<'a> {
91 scope: &'a str,
92 error: &'a str,
93 }
94 let mut seq = serializer.serialize_seq(Some(failed.len()))?;
95 for (scope, error) in failed {
96 seq.serialize_element(&Failure { scope, error })?;
97 }
98 seq.end()
99}
100
101pub fn wanted(mirror: &Mirror, targets: &Targets) -> Result<Vec<Scope>> {
110 let mut scopes = vec![Scope::Gists, Scope::Notifications, Scope::RateLimit];
111 for (repo, commits) in &targets.repositories {
112 scopes.extend([
113 Scope::Issues(repo.clone()),
114 Scope::Pulls(repo.clone()),
115 Scope::Repository(repo.clone()),
116 Scope::Releases(repo.clone()),
117 Scope::Tags(repo.clone()),
118 Scope::Milestones(repo.clone()),
119 Scope::Labels(repo.clone()),
120 Scope::Traffic(repo.clone()),
121 ]);
122 scopes.extend(commits.iter().map(|commit| Scope::Checks {
123 repo: repo.clone(),
124 commit: commit.clone(),
125 }));
126 }
127 scopes.extend(board_owners(mirror, targets)?.into_iter().map(Scope::Projects));
128 scopes.extend(
132 mirror
133 .states()?
134 .keys()
135 .filter_map(|key| Scope::parse(key))
136 .filter(|scope| match scope {
137 Scope::Comments { repo, .. } => targets.repositories.contains_key(repo),
138 _ => false,
139 }),
140 );
141 Ok(scopes)
142}
143
144pub fn viewer(mirror: &Mirror) -> Result<Option<GitHubViewer>> {
150 mirror
151 .only(&Scope::Viewer)?
152 .map(|payload| {
153 serde_json::from_value(payload).map_err(|e| {
154 anyhow!("the mirrored account cannot be read by this version ({e}); sync again")
155 })
156 })
157 .transpose()
158}
159
160pub fn board_owners(mirror: &Mirror, targets: &Targets) -> Result<BTreeSet<String>> {
163 let mut owners = targets.board_owners();
164 if let Some(viewer) = viewer(mirror)? {
165 owners.insert(viewer.login);
166 owners.extend(viewer.organizations);
167 }
168 Ok(owners)
169}
170
171pub fn resolve_boards(
177 mirror: &Mirror,
178 boards: &BTreeSet<BoardRef>,
179) -> Result<(Vec<GitHubProject>, Vec<String>)> {
180 let mut found = Vec::new();
181 let mut problems = Vec::new();
182 let mut lists: BTreeMap<String, std::result::Result<Vec<GitHubProject>, String>> =
184 BTreeMap::new();
185 for board in boards {
186 if !lists.contains_key(&board.owner) {
187 let list = owner_boards(mirror, &board.owner)?;
188 lists.insert(board.owner.clone(), list);
189 }
190 match &lists[&board.owner] {
191 Err(reason) => problems.push(format!("{board}: {reason}")),
192 Ok(list) => match list.iter().find(|project| project.number == board.number) {
193 Some(project) => found.push(project.clone()),
194 None => problems.push(format!("{} has no board {}", board.owner, board.number)),
195 },
196 }
197 }
198 Ok((found, problems))
199}
200
201pub fn owner_boards(
203 mirror: &Mirror,
204 owner: &str,
205) -> Result<std::result::Result<Vec<GitHubProject>, String>> {
206 let scope = Scope::Projects(owner.to_string());
207 let state = mirror.state(&scope)?;
208 if !state.ever_synced() {
209 return Ok(Err(match &state.last_error {
210 Some(error) => format!("{owner}'s boards could not be read: {error}"),
211 None => format!("{owner}'s boards have not been read yet"),
212 }));
213 }
214 let boards = mirror
215 .payloads(&scope)?
216 .into_iter()
217 .map(|payload| {
218 serde_json::from_value::<GitHubProject>(payload).map_err(|e| {
219 anyhow!("a mirrored board of {owner} cannot be read by this version ({e}); sync again")
220 })
221 })
222 .collect::<Result<Vec<_>>>()?;
223 Ok(Ok(boards))
224}
225
226pub fn board_scopes(mirror: &Mirror, targets: &Targets) -> Result<(Vec<Scope>, Vec<String>)> {
233 let (found, problems) = resolve_boards(mirror, &targets.boards)?;
234 let mut ids: BTreeSet<String> = found.into_iter().map(|board| board.id).collect();
235
236 let mut listed = BTreeSet::new();
237 for owner in board_owners(mirror, targets)? {
238 if let Ok(boards) = owner_boards(mirror, &owner)? {
239 listed.extend(boards.into_iter().map(|board| board.id));
240 }
241 }
242 ids.extend(
243 mirror
244 .states()?
245 .keys()
246 .filter_map(|key| match Scope::parse(key) {
247 Some(Scope::ProjectItems(id)) => Some(id),
248 _ => None,
249 })
250 .filter(|id| listed.contains(id)),
251 );
252 Ok((ids.into_iter().map(Scope::ProjectItems).collect(), problems))
253}
254
255pub async fn sync(
263 mirror: &Mirror,
264 client: Option<&GitHubClient>,
265 targets: &Targets,
266 force: bool,
267) -> SyncReport {
268 let mut report = SyncReport {
269 problems: targets.problems.clone(),
270 ..SyncReport::default()
271 };
272
273 let viewer = vec![Scope::Viewer];
276 if let Some(client) = client {
277 run(mirror, client, &viewer, force, &mut report).await;
278 } else {
279 report.no_token = true;
280 }
281 let first = match wanted(mirror, targets) {
282 Ok(scopes) => scopes,
283 Err(error) => {
284 report.problems.push(format!("the mirror cannot be read: {error:#}"));
285 return report;
286 }
287 };
288 if let Some(client) = client {
290 run(mirror, client, &first, force, &mut report).await;
291 }
292 let cards = match board_scopes(mirror, targets) {
293 Ok((scopes, problems)) => {
294 report.problems.extend(problems);
295 scopes
296 }
297 Err(error) => {
298 report.problems.push(format!("the boards cannot be resolved: {error:#}"));
299 Vec::new()
300 }
301 };
302 if let Some(client) = client {
303 run(mirror, client, &cards, force, &mut report).await;
304 }
305
306 let mut all = viewer;
307 all.extend(first);
308 all.extend(cards);
309 forget_departed(mirror, targets, &all, &mut report);
310 report
311}
312
313pub async fn sync_boards(
322 mirror: &Mirror,
323 client: &GitHubClient,
324 references: &BTreeSet<BoardRef>,
325 force: bool,
326) -> SyncReport {
327 let targets = Targets {
328 boards: references.clone(),
329 ..Targets::default()
330 };
331 let mut report = SyncReport::default();
332 run(mirror, client, &[Scope::Viewer], force, &mut report).await;
333 match board_owners(mirror, &targets) {
334 Ok(owners) => {
335 let lists: Vec<Scope> = owners.into_iter().map(Scope::Projects).collect();
336 run(mirror, client, &lists, force, &mut report).await;
337 }
338 Err(error) => report.problems.push(format!("the mirror cannot be read: {error:#}")),
339 }
340 match board_scopes(mirror, &targets) {
341 Ok((cards, problems)) => {
342 report.problems.extend(problems);
343 run(mirror, client, &cards, force, &mut report).await;
344 }
345 Err(error) => report.problems.push(format!("the boards cannot be resolved: {error:#}")),
346 }
347 report
348}
349
350pub async fn sync_scopes(
355 mirror: &Mirror,
356 client: &GitHubClient,
357 scopes: &[Scope],
358 force: bool,
359) -> SyncReport {
360 let mut report = SyncReport::default();
361 run(mirror, client, scopes, force, &mut report).await;
362 report
363}
364
365async fn run(
366 mirror: &Mirror,
367 client: &GitHubClient,
368 scopes: &[Scope],
369 force: bool,
370 report: &mut SyncReport,
371) {
372 let now = chrono::Utc::now();
373 for scope in scopes {
374 let state = match mirror.state(scope) {
375 Ok(state) => state,
376 Err(error) => {
377 report.failed.push((scope.key(), format!("{error:#}")));
378 continue;
379 }
380 };
381 if !force && !due(scope, &state, now) {
382 continue;
383 }
384 match sync_one(client, mirror, scope, &state).await {
385 Ok(true) => report.changed.push(scope.key()),
386 Ok(false) => report.unchanged.push(scope.key()),
387 Err(error) => {
388 let message = format!("{error:#}");
389 if let Err(write) = mirror.failed(scope, &message) {
390 report
391 .failed
392 .push((scope.key(), format!("{message}; and the failure could not be recorded: {write:#}")));
393 continue;
394 }
395 report.failed.push((scope.key(), message));
396 }
397 }
398 }
399}
400
401async fn sync_one(
408 client: &GitHubClient,
409 mirror: &Mirror,
410 scope: &Scope,
411 held: &SyncState,
412) -> Result<bool> {
413 let conditions = Conditions {
414 etag: held.etag.clone(),
415 since: held.cursor.clone(),
416 };
417
418 match scope {
419 Scope::Issues(repo) => {
420 let (owner, name) = split(repo)?;
421 match client.list_issues_changed(owner, name, &conditions).await? {
422 Fetched::Unchanged => unchanged(mirror, scope),
423 Fetched::Fresh { value, etag } => {
424 let documents = documents_by(&value, |issue| issue.number.to_string())?;
425 incremental(mirror, scope, documents, etag)
426 }
427 }
428 }
429 Scope::Pulls(repo) => {
430 let (owner, name) = split(repo)?;
431 match client.list_pulls_changed(owner, name, &conditions).await? {
432 Fetched::Unchanged => unchanged(mirror, scope),
433 Fetched::Fresh { value, etag } => {
434 let documents = documents_by(&value, |pull| pull.number.to_string())?;
435 incremental(mirror, scope, documents, etag)
436 }
437 }
438 }
439 Scope::Repository(repo) => {
440 let (owner, name) = split(repo)?;
441 match client.repository_changed(owner, name, &conditions).await? {
442 Fetched::Unchanged => unchanged(mirror, scope),
443 Fetched::Fresh { value, etag } => {
444 let changed = mirror.put_only(scope, serde_json::to_value(&value)?)?;
445 mirror.synced(scope, etag.as_deref(), None)?;
446 Ok(changed)
447 }
448 }
449 }
450 Scope::Checks { repo, commit } => {
451 let (owner, name) = split(repo)?;
452 let checks = client.checks_for(owner, name, commit).await?;
453 let document =
454 Document::of(SINGLETON, serde_json::to_value(&checks)?).with_state(checks.state.clone());
455 let changed = mirror.put(scope, &[document])?;
456 mirror.synced(scope, None, None)?;
457 Ok(changed)
458 }
459 Scope::Releases(repo) => {
460 let (owner, name) = split(repo)?;
461 match client.list_releases_changed(owner, name, &conditions).await? {
462 Fetched::Unchanged => unchanged(mirror, scope),
463 Fetched::Fresh { value, etag } => {
464 replace(mirror, scope, documents_by(&value, |release| release.id.to_string())?, etag)
465 }
466 }
467 }
468 Scope::Tags(repo) => {
469 let (owner, name) = split(repo)?;
470 match client.list_tags_changed(owner, name, &conditions).await? {
471 Fetched::Unchanged => unchanged(mirror, scope),
472 Fetched::Fresh { value, etag } => {
473 replace(mirror, scope, documents_by(&value, |tag| tag.name.clone())?, etag)
474 }
475 }
476 }
477 Scope::Milestones(repo) => {
478 let (owner, name) = split(repo)?;
479 match client.list_milestones_changed(owner, name, &conditions).await? {
480 Fetched::Unchanged => unchanged(mirror, scope),
481 Fetched::Fresh { value, etag } => replace(
482 mirror,
483 scope,
484 documents_by(&value, |milestone| milestone.number.to_string())?,
485 etag,
486 ),
487 }
488 }
489 Scope::Labels(repo) => {
490 let (owner, name) = split(repo)?;
491 match client.list_labels_changed(owner, name, &conditions).await? {
492 Fetched::Unchanged => unchanged(mirror, scope),
493 Fetched::Fresh { value, etag } => {
494 replace(mirror, scope, documents_by(&value, |label| label.name.clone())?, etag)
495 }
496 }
497 }
498 Scope::Traffic(repo) => {
499 let (owner, name) = split(repo)?;
500 let traffic = client.traffic(owner, name).await?;
503 let changed = mirror.put_only(scope, serde_json::to_value(&traffic)?)?;
504 mirror.synced(scope, None, None)?;
505 Ok(changed)
506 }
507 Scope::Comments { repo, number } => {
508 let (owner, name) = split(repo)?;
509 match client
510 .list_comments_changed(owner, name, *number, &conditions)
511 .await?
512 {
513 Fetched::Unchanged => unchanged(mirror, scope),
514 Fetched::Fresh { value, etag } => {
515 replace(mirror, scope, documents_by(&value, |comment| comment.id.to_string())?, etag)
516 }
517 }
518 }
519 Scope::Gists => match client.list_gists_changed(&conditions).await? {
520 Fetched::Unchanged => unchanged(mirror, scope),
521 Fetched::Fresh { value, etag } => {
522 let mut documents = Vec::new();
523 for gist in &value {
524 let payload = serde_json::to_value(gist)?;
525 documents.push(if gist_is_whole(gist) {
526 Document::of(gist.id.clone(), payload)
527 } else {
528 Document::summary(gist.id.clone(), payload)
532 });
533 }
534 replace(mirror, scope, documents, etag)
535 }
536 },
537 Scope::Notifications => match client.list_notifications_changed(&conditions).await? {
538 Fetched::Unchanged => unchanged(mirror, scope),
539 Fetched::Fresh { value, etag } => replace(
540 mirror,
541 scope,
542 documents_by(&value, |notification| notification.id.clone())?,
543 etag,
544 ),
545 },
546 Scope::RateLimit => {
547 let limit = client.rate_limit().await?;
548 let changed = mirror.put_only(scope, serde_json::to_value(&limit)?)?;
549 mirror.synced(scope, None, None)?;
550 Ok(changed)
551 }
552 Scope::Viewer => {
553 let viewer = client.viewer().await?;
554 let changed = mirror.put_only(scope, serde_json::to_value(&viewer)?)?;
555 mirror.synced(scope, None, None)?;
556 Ok(changed)
557 }
558 Scope::Projects(login) => {
559 let projects = client.list_projects(login).await?;
560 replace(mirror, scope, documents_by(&projects, |project| project.id.clone())?, None)
561 }
562 Scope::ProjectItems(id) => {
563 let items = client.project_items(id).await?;
564 replace(mirror, scope, documents_by(&items, |item| item.id.clone())?, None)
565 }
566 }
567}
568
569fn unchanged(mirror: &Mirror, scope: &Scope) -> Result<bool> {
571 mirror.synced(scope, None, None)?;
572 Ok(false)
573}
574
575fn incremental(
577 mirror: &Mirror,
578 scope: &Scope,
579 documents: Vec<Document>,
580 etag: Option<String>,
581) -> Result<bool> {
582 let cursor = newest_update(documents.iter());
583 let changed = mirror.put(scope, &documents)?;
584 mirror.synced(scope, etag.as_deref(), cursor.as_deref())?;
585 Ok(changed)
586}
587
588fn replace(
590 mirror: &Mirror,
591 scope: &Scope,
592 documents: Vec<Document>,
593 etag: Option<String>,
594) -> Result<bool> {
595 let changed = mirror.put(scope, &documents)?;
596 mirror.synced(scope, etag.as_deref(), None)?;
597 Ok(changed)
598}
599
600fn documents_by<T: Serialize>(values: &[T], key: impl Fn(&T) -> String) -> Result<Vec<Document>> {
602 values
603 .iter()
604 .map(|value| Ok(Document::of(key(value), serde_json::to_value(value)?)))
605 .collect()
606}
607
608fn split(repo: &str) -> Result<(&str, &str)> {
609 repo.split_once('/')
610 .ok_or_else(|| anyhow!("{repo} is not owner/repo"))
611}
612
613pub fn due(scope: &Scope, state: &SyncState, now: chrono::DateTime<chrono::Utc>) -> bool {
619 let Some(attempted) = state.attempted_at.as_deref().and_then(parse_time) else {
620 return true;
623 };
624 let waited = (now - attempted).to_std().unwrap_or(Duration::ZERO);
625 if state.last_error.is_some() {
626 return waited >= AFTER_FAILURE;
627 }
628 waited >= interval(scope)
629}
630
631pub fn interval(scope: &Scope) -> Duration {
635 match scope {
636 Scope::Traffic(_) => SWEEP_TRAFFIC,
637 Scope::Checks { .. } | Scope::Viewer | Scope::Projects(_) | Scope::ProjectItems(_) => {
638 SWEEP_UNCONDITIONAL
639 }
640 _ => SWEEP,
641 }
642}
643
644fn forget_departed(mirror: &Mirror, targets: &Targets, wanted: &[Scope], report: &mut SyncReport) {
651 if !report.problems.is_empty() {
652 return;
653 }
654 let keep: BTreeSet<String> = wanted.iter().map(Scope::key).collect();
655 let known = match mirror.states() {
656 Ok(known) => known,
657 Err(error) => {
658 report.problems.push(format!("the mirror cannot be read: {error:#}"));
659 return;
660 }
661 };
662 for key in known.keys() {
663 if keep.contains(key) {
664 continue;
665 }
666 let Some(scope) = Scope::parse(key) else {
667 report.problems.push(format!("the mirror holds a scope it cannot read: {key}"));
668 continue;
669 };
670 let forget = match &scope {
671 Scope::Gists | Scope::Notifications | Scope::RateLimit | Scope::Viewer => false,
673 Scope::Checks { .. } | Scope::Projects(_) | Scope::ProjectItems(_) => true,
678 other => other
681 .repository()
682 .is_some_and(|repo| !targets.repositories.contains_key(repo)),
683 };
684 if forget {
685 match mirror.forget(&scope) {
686 Ok(()) => report.changed.push(scope.key()),
687 Err(error) => report
688 .failed
689 .push((scope.key(), format!("cannot forget it: {error:#}"))),
690 }
691 }
692 }
693}
694
695pub const TICK: Duration = Duration::from_secs(30);
699
700pub async fn keep_current(
708 prj: wsb_prj::Prj,
709 mirror: Mirror,
710 mut on_pass: impl FnMut(&SyncReport) + Send,
711) {
712 loop {
713 let targets_of = prj.clone();
714 let read = tokio::task::spawn_blocking(move || {
715 (Targets::of(&targets_of), crate::token::resolve())
716 })
717 .await;
718 let report = match read {
719 Ok((Ok(targets), token)) => {
720 let client = token.ok().map(|token| GitHubClient::new(token.value()));
721 sync(&mirror, client.as_ref(), &targets, false).await
722 }
723 Ok((Err(error), _)) => SyncReport {
724 problems: vec![format!("the projects under {} cannot be read: {error:#}", prj.root().display())],
725 ..SyncReport::default()
726 },
727 Err(error) => SyncReport {
728 problems: vec![format!("reading what to mirror failed: {error}")],
729 ..SyncReport::default()
730 },
731 };
732 on_pass(&report);
733 tokio::time::sleep(TICK).await;
734 }
735}
736
737fn parse_time(value: &str) -> Option<chrono::DateTime<chrono::Utc>> {
738 chrono::DateTime::parse_from_rfc3339(value)
739 .ok()
740 .map(|time| time.with_timezone(&chrono::Utc))
741}
742
743#[cfg(test)]
744#[path = "sync_tests.rs"]
745mod tests;