1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
use crate::models::{
ColumnAssignment, CurveMapping, DataStream, StandardCurveUpsert, StreamDescriptor,
StreamFetchRequest, StreamReadings, StreamStatusEvents,
};
pub type BackendError = Box<dyn std::error::Error + Send + Sync>;
/// Source-specific extraction logic. Implement this to build a sync service;
/// registration, cursors, batching, retries and reporting are the driver's job.
#[async_trait::async_trait]
pub trait SourceBackend: Send + Sync + 'static {
/// The source_system string streams register under (ie. "vaisala", "cnet").
fn source_system(&self) -> &str;
/// Run discovery on every cycle instead of once at startup plus full syncs.
/// Return true when discovery is cheap and new streams appear without
/// operator action (ie. rows added to a portal database).
/// Whether this backend replicates a mutable source by re-reading its full content each
/// cycle. When true, the driver fetches with no cursor (`since: None`) on every pass, so the
/// backend can attach completeness windows and the store converges on the source
/// (corrections applied, removals withdrawn). Append sources keep the incremental cursor.
fn reconciled(&self) -> bool {
false
}
fn rediscover_every_cycle(&self) -> bool {
false
}
/// Enumerate the streams this source provides.
async fn discover_streams(&self) -> Result<Vec<StreamDescriptor>, BackendError>;
/// Standard curves to register with the API before stream registration.
/// Default: none.
async fn discover_standard_curves(&self) -> Result<Vec<StandardCurveUpsert>, BackendError> {
Ok(Vec::new())
}
/// Receives the API-side identities the discovered curves resolved to, so
/// the backend can stamp readings with curve UUIDs. Default: ignored.
async fn apply_curve_mappings(&self, _mappings: &[CurveMapping]) -> Result<(), BackendError> {
Ok(())
}
/// Receives the authoritative replicate column-to-index mapping for one
/// family stream, as the API pinned it. The driver calls this whenever it
/// learns a mapping (register response, or the copy persisted on stream
/// metadata), before it asks for readings; the backend must assign each
/// value's `replicate_index` from the mapping rather than from column
/// position. Default: ignored, for sources without replicate families.
async fn apply_replicate_assignments(
&self,
_source_key: &str,
_assignments: &[ColumnAssignment],
) -> Result<(), BackendError> {
Ok(())
}
/// Fetch new readings. Receives all requests in one call so the backend
/// can batch upstream queries; each stream carries its own cursor.
async fn fetch_readings(
&self,
requests: &[StreamFetchRequest],
) -> Result<Vec<StreamReadings>, BackendError>;
/// Device or status telemetry. Default: none.
async fn fetch_status_events(
&self,
_streams: &[DataStream],
) -> Result<Vec<StreamStatusEvents>, BackendError> {
Ok(Vec::new())
}
/// Custom command handling, forwarded from the runner.
async fn handle_command(
&self,
command: &str,
_payload: Option<serde_json::Value>,
) -> Result<serde_json::Value, BackendError> {
Err(format!("Unknown command: {command}").into())
}
}