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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
use crate::models::{
ColumnAssignment, CurveMapping, DataStream, NoteMapping, NoteUpsert, SensorUpsert,
SourceInventory, 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;
/// 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
}
/// 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).
fn rediscover_every_cycle(&self) -> bool {
false
}
/// Enumerate the streams this source provides.
async fn discover_streams(&self) -> Result<Vec<StreamDescriptor>, BackendError>;
/// Instruments from the source's own register, for instruments that have no stream of their
/// own to be minted from. Registered before curves and streams. Default: none.
async fn discover_instruments(&self) -> Result<Vec<SensorUpsert>, BackendError> {
Ok(Vec::new())
}
/// Source-authored site notes. Registered after streams, so a station's
/// site exists by the time its notes arrive. Default: none.
async fn discover_notes(&self) -> Result<Vec<NoteUpsert>, BackendError> {
Ok(Vec::new())
}
/// Receives the API-side outcome for each registered note. Default: ignored.
async fn apply_note_mappings(&self, _mappings: &[NoteMapping]) -> Result<(), BackendError> {
Ok(())
}
/// 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())
}
/// Everything the source offers, taken or declined, for the source audit.
///
/// The default answers with what discovery accepted, which is honest for a backend that
/// declines nothing. A backend that filters its source (a portal connector reading a wide
/// table, say) overrides this and reports the rest with a reason, because a declined column is
/// exactly what no window, receipt or reconciliation pass can see.
async fn source_inventory(&self) -> Result<SourceInventory, BackendError> {
Ok(SourceInventory::of_discovered(
&self.discover_streams().await?,
))
}
}