origin_sync/
source.rs

1use crate::SyncTarget;
2use async_trait::async_trait;
3use origin_domain::{Result, SyncId, SyncState, ThrottleReason};
4use std::fmt::Debug;
5use time::Duration;
6use tokio_util::sync::CancellationToken;
7
8/// A service-imposed limit on how soon the next run may happen.
9///
10/// The connector reports it; the engine owns scheduling. It arrives in two shapes —
11/// a *quota* the service reported in the body, and a *minimum poll interval* — but is
12/// handled identically: the next run may not start before `delay` has elapsed.
13#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14pub struct SyncThrottle {
15    pub delay: Duration,
16    pub reason: ThrottleReason,
17}
18
19impl SyncThrottle {
20    /// A quota or cost reported in the body, not a header (G6).
21    pub fn quota(delay: Duration) -> Self {
22        Self {
23            delay,
24            reason: ThrottleReason::Quota,
25        }
26    }
27
28    /// A minimum poll interval the service named (G7).
29    pub fn server_interval(delay: Duration) -> Self {
30        Self {
31            delay,
32            reason: ThrottleReason::ServerInterval,
33        }
34    }
35}
36
37/// What one successful sync produced.
38#[derive(Debug, Clone, Default, PartialEq, Eq)]
39pub struct SyncReport {
40    /// How many records changed. Used for the "3 new notifications" kind of message.
41    pub changed: u64,
42    /// Validator to send on the next request, if the service issued one.
43    pub etag: Option<String>,
44    pub last_modified: Option<String>,
45    /// A throttle the service reported alongside the data. `None` keeps the policy
46    /// cadence; `Some` pushes the next run back to at least `delay` from now.
47    pub throttle: Option<SyncThrottle>,
48}
49
50impl SyncReport {
51    pub fn changed(changed: u64) -> Self {
52        Self {
53            changed,
54            ..Self::default()
55        }
56    }
57
58    pub fn with_etag(mut self, etag: impl Into<String>) -> Self {
59        self.etag = Some(etag.into());
60        self
61    }
62
63    pub fn with_last_modified(mut self, last_modified: impl Into<String>) -> Self {
64        self.last_modified = Some(last_modified.into());
65        self
66    }
67
68    pub fn with_throttle(mut self, throttle: SyncThrottle) -> Self {
69        self.throttle = Some(throttle);
70        self
71    }
72}
73
74/// The outcome a source reports.
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub enum SyncResult {
77    Updated(SyncReport),
78    /// The service confirmed nothing changed — a validator hit. Counts as success.
79    NotModified,
80}
81
82/// Everything a source is given for one run.
83#[derive(Debug)]
84pub struct SyncContext {
85    pub sync_id: SyncId,
86    pub target: SyncTarget,
87    /// State from the previous run: validators, failure streak, timestamps.
88    pub state: SyncState,
89    cancel: CancellationToken,
90}
91
92impl SyncContext {
93    pub(crate) fn new(
94        sync_id: SyncId,
95        target: SyncTarget,
96        state: SyncState,
97        cancel: CancellationToken,
98    ) -> Self {
99        Self {
100            sync_id,
101            target,
102            state,
103            cancel,
104        }
105    }
106
107    /// Validator from the last successful run, to be sent as `If-None-Match`.
108    pub fn etag(&self) -> Option<&str> {
109        self.state.etag.as_deref()
110    }
111
112    /// Validator from the last successful run, to be sent as `If-Modified-Since`.
113    pub fn last_modified(&self) -> Option<&str> {
114        self.state.last_modified.as_deref()
115    }
116
117    /// Check between pages of a paginated fetch.
118    pub fn is_cancelled(&self) -> bool {
119        self.cancel.is_cancelled()
120    }
121
122    pub async fn cancelled(&self) {
123        self.cancel.cancelled().await;
124    }
125}
126
127/// How a particular kind of data is fetched.
128///
129/// Implementations answer *how*, never *when*. They do not sleep, do not retry and do
130/// not decide whether the machine is online — the engine owns all of that.
131#[async_trait]
132pub trait SyncSource: Debug + Send + Sync + 'static {
133    async fn sync(&self, context: &SyncContext) -> Result<SyncResult>;
134}
origin_sync/source.rs

origin_sync/
source.rs
