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>)> {
237 let (found, problems) = resolve_boards(mirror, &targets.boards)?;
238 let mut ids: BTreeSet<String> = found.into_iter().map(|board| board.id).collect();
239 for owner in board_owners(mirror, targets)? {
240 if let Ok(boards) = owner_boards(mirror, &owner)? {
241 ids.extend(boards.into_iter().map(|board| board.id));
242 }
243 }
244 Ok((ids.into_iter().map(Scope::ProjectItems).collect(), problems))
245}
246
247pub async fn sync(
255 mirror: &Mirror,
256 client: Option<&GitHubClient>,
257 targets: &Targets,
258 force: bool,
259) -> SyncReport {
260 let mut report = SyncReport {
261 problems: targets.problems.clone(),
262 ..SyncReport::default()
263 };
264
265 let viewer = vec![Scope::Viewer];
268 if let Some(client) = client {
269 run(mirror, client, &viewer, force, &mut report).await;
270 } else {
271 report.no_token = true;
272 }
273 let first = match wanted(mirror, targets) {
274 Ok(scopes) => scopes,
275 Err(error) => {
276 report.problems.push(format!("the mirror cannot be read: {error:#}"));
277 return report;
278 }
279 };
280 if let Some(client) = client {
282 run(mirror, client, &first, force, &mut report).await;
283 }
284 let cards = match board_scopes(mirror, targets) {
285 Ok((scopes, problems)) => {
286 report.problems.extend(problems);
287 scopes
288 }
289 Err(error) => {
290 report.problems.push(format!("the boards cannot be resolved: {error:#}"));
291 Vec::new()
292 }
293 };
294 if let Some(client) = client {
295 run(mirror, client, &cards, force, &mut report).await;
296 }
297
298 let mut all = viewer;
299 all.extend(first);
300 all.extend(cards);
301 forget_departed(mirror, targets, &all, &mut report);
302 report
303}
304
305pub async fn sync_boards(
313 mirror: &Mirror,
314 client: &GitHubClient,
315 references: &BTreeSet<BoardRef>,
316 force: bool,
317) -> SyncReport {
318 let targets = Targets {
319 boards: references.clone(),
320 ..Targets::default()
321 };
322 let mut report = SyncReport::default();
323 run(mirror, client, &[Scope::Viewer], force, &mut report).await;
324 match board_owners(mirror, &targets) {
325 Ok(owners) => {
326 let lists: Vec<Scope> = owners.into_iter().map(Scope::Projects).collect();
327 run(mirror, client, &lists, force, &mut report).await;
328 }
329 Err(error) => report.problems.push(format!("the mirror cannot be read: {error:#}")),
330 }
331 match board_scopes(mirror, &targets) {
332 Ok((cards, problems)) => {
333 report.problems.extend(problems);
334 run(mirror, client, &cards, force, &mut report).await;
335 }
336 Err(error) => report.problems.push(format!("the boards cannot be resolved: {error:#}")),
337 }
338 report
339}
340
341pub async fn sync_scopes(
346 mirror: &Mirror,
347 client: &GitHubClient,
348 scopes: &[Scope],
349 force: bool,
350) -> SyncReport {
351 let mut report = SyncReport::default();
352 run(mirror, client, scopes, force, &mut report).await;
353 report
354}
355
356async fn run(
357 mirror: &Mirror,
358 client: &GitHubClient,
359 scopes: &[Scope],
360 force: bool,
361 report: &mut SyncReport,
362) {
363 let now = chrono::Utc::now();
364 for scope in scopes {
365 let state = match mirror.state(scope) {
366 Ok(state) => state,
367 Err(error) => {
368 report.failed.push((scope.key(), format!("{error:#}")));
369 continue;
370 }
371 };
372 if !force && !due(scope, &state, now) {
373 continue;
374 }
375 match sync_one(client, mirror, scope, &state).await {
376 Ok(true) => report.changed.push(scope.key()),
377 Ok(false) => report.unchanged.push(scope.key()),
378 Err(error) => {
379 let message = format!("{error:#}");
380 if let Err(write) = mirror.failed(scope, &message) {
381 report
382 .failed
383 .push((scope.key(), format!("{message}; and the failure could not be recorded: {write:#}")));
384 continue;
385 }
386 report.failed.push((scope.key(), message));
387 }
388 }
389 }
390}
391
392async fn sync_one(
399 client: &GitHubClient,
400 mirror: &Mirror,
401 scope: &Scope,
402 held: &SyncState,
403) -> Result<bool> {
404 let conditions = Conditions {
405 etag: held.etag.clone(),
406 since: held.cursor.clone(),
407 };
408
409 match scope {
410 Scope::Issues(repo) => {
411 let (owner, name) = split(repo)?;
412 match client.list_issues_changed(owner, name, &conditions).await? {
413 Fetched::Unchanged => unchanged(mirror, scope),
414 Fetched::Fresh { value, etag } => {
415 let documents = documents_by(&value, |issue| issue.number.to_string())?;
416 incremental(mirror, scope, documents, etag)
417 }
418 }
419 }
420 Scope::Pulls(repo) => {
421 let (owner, name) = split(repo)?;
422 match client.list_pulls_changed(owner, name, &conditions).await? {
423 Fetched::Unchanged => unchanged(mirror, scope),
424 Fetched::Fresh { value, etag } => {
425 let documents = documents_by(&value, |pull| pull.number.to_string())?;
426 incremental(mirror, scope, documents, etag)
427 }
428 }
429 }
430 Scope::Repository(repo) => {
431 let (owner, name) = split(repo)?;
432 match client.repository_changed(owner, name, &conditions).await? {
433 Fetched::Unchanged => unchanged(mirror, scope),
434 Fetched::Fresh { value, etag } => {
435 let changed = mirror.put_only(scope, serde_json::to_value(&value)?)?;
436 mirror.synced(scope, etag.as_deref(), None)?;
437 Ok(changed)
438 }
439 }
440 }
441 Scope::Checks { repo, commit } => {
442 let (owner, name) = split(repo)?;
443 let checks = client.checks_for(owner, name, commit).await?;
444 let document =
445 Document::of(SINGLETON, serde_json::to_value(&checks)?).with_state(checks.state.clone());
446 let changed = mirror.put(scope, &[document])?;
447 mirror.synced(scope, None, None)?;
448 Ok(changed)
449 }
450 Scope::Releases(repo) => {
451 let (owner, name) = split(repo)?;
452 match client.list_releases_changed(owner, name, &conditions).await? {
453 Fetched::Unchanged => unchanged(mirror, scope),
454 Fetched::Fresh { value, etag } => {
455 replace(mirror, scope, documents_by(&value, |release| release.id.to_string())?, etag)
456 }
457 }
458 }
459 Scope::Tags(repo) => {
460 let (owner, name) = split(repo)?;
461 match client.list_tags_changed(owner, name, &conditions).await? {
462 Fetched::Unchanged => unchanged(mirror, scope),
463 Fetched::Fresh { value, etag } => {
464 replace(mirror, scope, documents_by(&value, |tag| tag.name.clone())?, etag)
465 }
466 }
467 }
468 Scope::Milestones(repo) => {
469 let (owner, name) = split(repo)?;
470 match client.list_milestones_changed(owner, name, &conditions).await? {
471 Fetched::Unchanged => unchanged(mirror, scope),
472 Fetched::Fresh { value, etag } => replace(
473 mirror,
474 scope,
475 documents_by(&value, |milestone| milestone.number.to_string())?,
476 etag,
477 ),
478 }
479 }
480 Scope::Labels(repo) => {
481 let (owner, name) = split(repo)?;
482 match client.list_labels_changed(owner, name, &conditions).await? {
483 Fetched::Unchanged => unchanged(mirror, scope),
484 Fetched::Fresh { value, etag } => {
485 replace(mirror, scope, documents_by(&value, |label| label.name.clone())?, etag)
486 }
487 }
488 }
489 Scope::Traffic(repo) => {
490 let (owner, name) = split(repo)?;
491 let traffic = client.traffic(owner, name).await?;
494 let changed = mirror.put_only(scope, serde_json::to_value(&traffic)?)?;
495 mirror.synced(scope, None, None)?;
496 Ok(changed)
497 }
498 Scope::Comments { repo, number } => {
499 let (owner, name) = split(repo)?;
500 match client
501 .list_comments_changed(owner, name, *number, &conditions)
502 .await?
503 {
504 Fetched::Unchanged => unchanged(mirror, scope),
505 Fetched::Fresh { value, etag } => {
506 replace(mirror, scope, documents_by(&value, |comment| comment.id.to_string())?, etag)
507 }
508 }
509 }
510 Scope::Gists => match client.list_gists_changed(&conditions).await? {
511 Fetched::Unchanged => unchanged(mirror, scope),
512 Fetched::Fresh { value, etag } => {
513 let mut documents = Vec::new();
514 for gist in &value {
515 let payload = serde_json::to_value(gist)?;
516 documents.push(if gist_is_whole(gist) {
517 Document::of(gist.id.clone(), payload)
518 } else {
519 Document::summary(gist.id.clone(), payload)
523 });
524 }
525 replace(mirror, scope, documents, etag)
526 }
527 },
528 Scope::Notifications => match client.list_notifications_changed(&conditions).await? {
529 Fetched::Unchanged => unchanged(mirror, scope),
530 Fetched::Fresh { value, etag } => replace(
531 mirror,
532 scope,
533 documents_by(&value, |notification| notification.id.clone())?,
534 etag,
535 ),
536 },
537 Scope::RateLimit => {
538 let limit = client.rate_limit().await?;
539 let changed = mirror.put_only(scope, serde_json::to_value(&limit)?)?;
540 mirror.synced(scope, None, None)?;
541 Ok(changed)
542 }
543 Scope::Viewer => {
544 let viewer = client.viewer().await?;
545 let changed = mirror.put_only(scope, serde_json::to_value(&viewer)?)?;
546 mirror.synced(scope, None, None)?;
547 Ok(changed)
548 }
549 Scope::Projects(login) => {
550 let projects = client.list_projects(login).await?;
551 replace(mirror, scope, documents_by(&projects, |project| project.id.clone())?, None)
552 }
553 Scope::ProjectItems(id) => {
554 let mut items = client.project_items(id).await?;
555 let listed: BTreeSet<String> = items.iter().map(|item| item.id.clone()).collect();
562 let unlisted: Vec<String> = mirror
563 .documents(scope)?
564 .into_iter()
565 .map(|document| document.key)
566 .filter(|key| !listed.contains(key))
567 .collect();
568 if !unlisted.is_empty() {
569 items.extend(client.project_items_on(id, &unlisted).await?);
570 }
571 replace(mirror, scope, documents_by(&items, |item| item.id.clone())?, None)
572 }
573 }
574}
575
576fn unchanged(mirror: &Mirror, scope: &Scope) -> Result<bool> {
578 mirror.synced(scope, None, None)?;
579 Ok(false)
580}
581
582fn incremental(
584 mirror: &Mirror,
585 scope: &Scope,
586 documents: Vec<Document>,
587 etag: Option<String>,
588) -> Result<bool> {
589 let cursor = newest_update(documents.iter());
590 let changed = mirror.put(scope, &documents)?;
591 mirror.synced(scope, etag.as_deref(), cursor.as_deref())?;
592 Ok(changed)
593}
594
595fn replace(
597 mirror: &Mirror,
598 scope: &Scope,
599 documents: Vec<Document>,
600 etag: Option<String>,
601) -> Result<bool> {
602 let changed = mirror.put(scope, &documents)?;
603 mirror.synced(scope, etag.as_deref(), None)?;
604 Ok(changed)
605}
606
607fn documents_by<T: Serialize>(values: &[T], key: impl Fn(&T) -> String) -> Result<Vec<Document>> {
609 values
610 .iter()
611 .map(|value| Ok(Document::of(key(value), serde_json::to_value(value)?)))
612 .collect()
613}
614
615fn split(repo: &str) -> Result<(&str, &str)> {
616 repo.split_once('/')
617 .ok_or_else(|| anyhow!("{repo} is not owner/repo"))
618}
619
620pub fn due(scope: &Scope, state: &SyncState, now: chrono::DateTime<chrono::Utc>) -> bool {
626 let Some(attempted) = state.attempted_at.as_deref().and_then(parse_time) else {
627 return true;
630 };
631 let waited = (now - attempted).to_std().unwrap_or(Duration::ZERO);
632 if state.last_error.is_some() {
633 return waited >= AFTER_FAILURE;
634 }
635 waited >= interval(scope)
636}
637
638pub fn interval(scope: &Scope) -> Duration {
642 match scope {
643 Scope::Traffic(_) => SWEEP_TRAFFIC,
644 Scope::Checks { .. } | Scope::Viewer | Scope::Projects(_) | Scope::ProjectItems(_) => {
645 SWEEP_UNCONDITIONAL
646 }
647 _ => SWEEP,
648 }
649}
650
651fn forget_departed(mirror: &Mirror, targets: &Targets, wanted: &[Scope], report: &mut SyncReport) {
658 if !report.problems.is_empty() {
659 return;
660 }
661 let keep: BTreeSet<String> = wanted.iter().map(Scope::key).collect();
662 let known = match mirror.states() {
663 Ok(known) => known,
664 Err(error) => {
665 report.problems.push(format!("the mirror cannot be read: {error:#}"));
666 return;
667 }
668 };
669 for key in known.keys() {
670 if keep.contains(key) {
671 continue;
672 }
673 let Some(scope) = Scope::parse(key) else {
674 report.problems.push(format!("the mirror holds a scope it cannot read: {key}"));
675 continue;
676 };
677 let forget = match &scope {
678 Scope::Gists | Scope::Notifications | Scope::RateLimit | Scope::Viewer => false,
680 Scope::Checks { .. } | Scope::Projects(_) | Scope::ProjectItems(_) => true,
685 other => other
688 .repository()
689 .is_some_and(|repo| !targets.repositories.contains_key(repo)),
690 };
691 if forget {
692 match mirror.forget(&scope) {
693 Ok(()) => report.changed.push(scope.key()),
694 Err(error) => report
695 .failed
696 .push((scope.key(), format!("cannot forget it: {error:#}"))),
697 }
698 }
699 }
700}
701
702pub const TICK: Duration = Duration::from_secs(30);
706
707pub async fn keep_current(
715 workshop: wsb_prj::Workshop,
716 mirror: Mirror,
717 mut on_pass: impl FnMut(&SyncReport) + Send,
718) {
719 loop {
720 let targets_of = workshop.clone();
721 let read = tokio::task::spawn_blocking(move || {
722 (Targets::of(&targets_of), crate::token::resolve())
723 })
724 .await;
725 let report = match read {
726 Ok((Ok(targets), token)) => {
727 let client = token.ok().map(|token| GitHubClient::new(token.value()));
728 sync(&mirror, client.as_ref(), &targets, false).await
729 }
730 Ok((Err(error), _)) => SyncReport {
731 problems: vec![format!("the projects in {} cannot be read: {error:#}", workshop.root().display())],
732 ..SyncReport::default()
733 },
734 Err(error) => SyncReport {
735 problems: vec![format!("reading what to mirror failed: {error}")],
736 ..SyncReport::default()
737 },
738 };
739 on_pass(&report);
740 tokio::time::sleep(TICK).await;
741 }
742}
743
744fn parse_time(value: &str) -> Option<chrono::DateTime<chrono::Utc>> {
745 chrono::DateTime::parse_from_rfc3339(value)
746 .ok()
747 .map(|time| time.with_timezone(&chrono::Utc))
748}
749
750#[cfg(test)]
751#[path = "sync_tests.rs"]
752mod tests;