@ondeinference / onde
Code Commits Issues Pulls Cloud Agent CI/CD
feature/diffusion-vision
copilot/assess-inference-engine-complexity copilot/fix-onde-npm-sdk-v1-1-2-release development feature/diffusion-vision main
rs 332 lines 12.1 KB
Raw
1 use smbcloud_gresiq_sdk::{Environment, GresiqClient, GresiqCredentials};
2
3 use super::events::{InferenceEvent, ModelLoadedEvent};
4
5 /// GresIQ credentials embedded at SDK build time — one pair per environment.
6 /// Consumer apps never set these — they're Onde Inference's own credentials.
7 const EMBEDDED_API_KEY_DEV: Option<&str> = option_env!("GRESIQ_API_KEY_DEV");
8 const EMBEDDED_API_SECRET_DEV: Option<&str> = option_env!("GRESIQ_API_SECRET_DEV");
9 const EMBEDDED_API_KEY_PRODUCTION: Option<&str> = option_env!("GRESIQ_API_KEY_PRODUCTION");
10 const EMBEDDED_API_SECRET_SECRET_PRODUCTION: Option<&str> =
11 option_env!("GRESIQ_API_SECRET_PRODUCTION");
12
13 /// Onde telemetry client. Wraps GresiqClient so pulse events land in the
14 /// right GresIQ-managed tables without consumer apps knowing anything about
15 /// the GresIQ auth layer underneath.
16 ///
17 /// GresIQ credentials are embedded in the SDK at build time.
18 /// Consumer apps only provide an `edge_id` (stable device identifier).
19 ///
20 /// Cheap to clone: the inner GresiqClient holds an Arc-backed reqwest::Client,
21 /// so cloning is a pointer bump, not a new TCP connection.
22 #[derive(Debug, Clone)]
23 pub struct PulseClient {
24 inner: GresiqClient,
25 edge_id: String,
26 onde_app_id: Option<String>,
27 }
28
29 impl PulseClient {
30 /// Returns true when pulse telemetry is disabled explicitly by the host app.
31 pub fn disabled_by_env() -> bool {
32 matches!(
33 std::env::var("ONDE_DISABLE_PULSE")
34 .ok()
35 .as_deref()
36 .map(str::trim)
37 .map(str::to_ascii_lowercase)
38 .as_deref(),
39 Some("1") | Some("true") | Some("yes") | Some("on")
40 )
41 }
42
43 /// ADR-0003 M2 dual-write toggle. When enabled (the default), every pulse
44 /// event is *also* written to the GresIQ document gateway
45 /// (`/gresiq/v1/collections/:collection`) in addition to the semantic
46 /// `pulse/*` routes, which still serve reads until the M5 cut-over. Set
47 /// `ONDE_PULSE_DUAL_WRITE=0` (or `false`/`no`/`off`) to suppress it.
48 fn dual_write_enabled() -> bool {
49 !matches!(
50 std::env::var("ONDE_PULSE_DUAL_WRITE")
51 .ok()
52 .as_deref()
53 .map(str::trim)
54 .map(str::to_ascii_lowercase)
55 .as_deref(),
56 Some("0") | Some("false") | Some("no") | Some("off")
57 )
58 }
59
60 /// Build a pulse client using the GresIQ credentials embedded in the SDK.
61 ///
62 /// Returns `None` if the SDK was compiled without `GRESIQ_API_KEY` /
63 /// `GRESIQ_API_SECRET` (e.g. a local dev build of onde without `.env`).
64 /// In that case telemetry is silently disabled — no panic, no partial state.
65 ///
66 /// `edge_id` is a stable device identifier (installation UUID).
67 /// Pass an empty string to default to `"onde-unknown"`.
68 ///
69 /// `onde_app_id` is the Onde app UUID (from the ondeinference.com
70 /// dashboard) that owns this telemetry. `None` is fine — events then carry
71 /// no app association (open-source/direct Rust consumers).
72 pub fn new(
73 environment: Environment,
74 edge_id: String,
75 onde_app_id: Option<String>,
76 ) -> Option<Self> {
77 if Self::disabled_by_env() {
78 return None;
79 }
80
81 let (api_key, api_secret) = match environment {
82 Environment::Dev => (EMBEDDED_API_KEY_DEV?, EMBEDDED_API_SECRET_DEV?),
83 Environment::Production => (
84 EMBEDDED_API_KEY_PRODUCTION?,
85 EMBEDDED_API_SECRET_SECRET_PRODUCTION?,
86 ),
87 };
88
89 let edge_id = if edge_id.is_empty() {
90 "onde-unknown".to_string()
91 } else {
92 edge_id
93 };
94
95 // reqwest 0.12.x requires a live Tokio runtime (with I/O reactor)
96 // to construct a Client. GresiqClient::from_credentials calls
97 // reqwest::Client::new() internally. Guard against panics when
98 // called from a thread/context that lacks a Tokio reactor.
99 if tokio::runtime::Handle::try_current().is_err() {
100 log::warn!(
101 "pulse: no Tokio runtime available — \
102 deferring PulseClient creation"
103 );
104 return None;
105 }
106
107 let credentials = GresiqCredentials {
108 api_key,
109 api_secret,
110 };
111
112 let inner = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
113 GresiqClient::from_credentials(environment, credentials)
114 })) {
115 Ok(client) => client,
116 Err(_) => {
117 log::warn!(
118 "pulse: GresiqClient::from_credentials panicked \
119 (likely missing Tokio reactor) — telemetry disabled"
120 );
121 return None;
122 }
123 };
124
125 Some(PulseClient {
126 inner,
127 edge_id,
128 onde_app_id,
129 })
130 }
131
132 /// Writes the model-load event to the pulse/model_loaded table and
133 /// **awaits** the result before returning.
134 ///
135 /// This must complete before any `record_inference` call for the same
136 /// edge + model is made, because the API enforces a foreign-key
137 /// constraint: an `inference_event` row is rejected with HTTP 404
138 /// ("Call model_loaded first") if no matching `model_loaded` row exists.
139 ///
140 /// Keeping this synchronous with respect to the model-load path means
141 /// the row is always present by the time the first inference event fires.
142 /// A failed write emits a warn! log line — no retry, no queue, and no
143 /// effect on the caller.
144 pub async fn record_model_loaded(
145 &self,
146 model_id: String,
147 model_name: String,
148 load_duration_ms: u64,
149 ) {
150 let client = self.clone();
151 let send_event = async move {
152 let event = ModelLoadedEvent {
153 edge_id: client.edge_id.clone(),
154 model_id,
155 model_name,
156 load_duration_ms,
157 onde_app_id: client.onde_app_id.clone(),
158 };
159 if let Err(error) = client.inner.insert("pulse/model_loaded", &event).await {
160 log::warn!("pulse: model_loaded failed: {}", error);
161 }
162 client.dual_write_model_loaded(&event).await;
163 };
164
165 if tokio::runtime::Handle::try_current().is_ok() {
166 send_event.await;
167 return;
168 }
169
170 let join = std::thread::spawn(move || {
171 let runtime = tokio::runtime::Builder::new_current_thread()
172 .enable_all()
173 .build();
174
175 match runtime {
176 Ok(runtime) => {
177 runtime.block_on(send_event);
178 }
179 Err(error) => {
180 log::warn!(
181 "pulse: could not create fallback runtime for model_loaded: {}",
182 error
183 );
184 }
185 }
186 });
187
188 if join.join().is_err() {
189 log::warn!("pulse: fallback model_loaded thread panicked");
190 }
191 }
192
193 /// Same fire-and-forget pattern as record_model_loaded but for inference
194 /// completions. Writes to pulse/inference_event. Logs on failure, no retry.
195 ///
196 /// Swift / UniFFI consumers may call into the SDK from contexts where
197 /// `tokio::spawn` does not have a current runtime handle even though the
198 /// outer API is async. To avoid panicking on Apple platforms, we prefer
199 /// the current Tokio runtime when available and otherwise fall back to a
200 /// short-lived current-thread runtime on a native background thread.
201 pub fn record_inference(
202 &self,
203 model_id: String,
204 request_id: String,
205 duration_ms: u64,
206 status: String,
207 ) {
208 let client = self.clone();
209 let send_event = async move {
210 let event = InferenceEvent {
211 edge_id: client.edge_id.clone(),
212 model_id,
213 request_id,
214 duration_ms,
215 status,
216 onde_app_id: client.onde_app_id.clone(),
217 };
218 if let Err(error) = client.inner.insert("pulse/inference_event", &event).await {
219 log::warn!("pulse: inference_event failed: {}", error);
220 }
221 client.dual_write_inference(&event).await;
222 };
223
224 if let Ok(handle) = tokio::runtime::Handle::try_current() {
225 handle.spawn(send_event);
226 return;
227 }
228
229 std::thread::spawn(move || {
230 let runtime = tokio::runtime::Builder::new_current_thread()
231 .enable_all()
232 .build();
233
234 match runtime {
235 Ok(runtime) => {
236 runtime.block_on(send_event);
237 }
238 Err(error) => {
239 log::warn!(
240 "pulse: could not create fallback runtime for inference_event: {}",
241 error
242 );
243 }
244 }
245 });
246 }
247
248 // ── ADR-0003 M2 dual-write to the document gateway ───────────────────────
249 //
250 // These mirror each semantic pulse event into JSONB collections under the
251 // Onde gresiq_app. The relational `pulse/*` routes still own reads until the
252 // M5 cut-over, so these writes are fire-and-forget: a failure is logged and
253 // never affects inference. `onde_app_id` rides as a plain doc field — there
254 // is no FK in the document store.
255
256 /// Mirror a model-load into `pulse_edges`, `pulse_models`, and
257 /// `pulse_deployments`. Each collection is a separate upsert (the
258 /// FK-ordered chain of the semantic route becomes client-orchestrated
259 /// multi-insert; the document store has no relationships to order).
260 async fn dual_write_model_loaded(&self, event: &ModelLoadedEvent) {
261 if !Self::dual_write_enabled() {
262 return;
263 }
264
265 self.write_doc(
266 "pulse_edges",
267 Some(&event.edge_id),
268 &serde_json::json!({ "edge_id": event.edge_id }),
269 )
270 .await;
271
272 // One model assignment per app (the relational table is unique on
273 // app_id); key by onde_app_id when present, else by the model itself.
274 let model_key = event
275 .onde_app_id
276 .clone()
277 .unwrap_or_else(|| event.model_id.clone());
278 self.write_doc(
279 "pulse_models",
280 Some(&model_key),
281 &serde_json::json!({
282 "slug": event.model_id,
283 "model_id": event.model_id,
284 "model_name": event.model_name,
285 "onde_app_id": event.onde_app_id,
286 }),
287 )
288 .await;
289
290 let deployment_key = format!("{}:{}", event.edge_id, event.model_id);
291 self.write_doc(
292 "pulse_deployments",
293 Some(&deployment_key),
294 &serde_json::json!({
295 "edge_id": event.edge_id,
296 "model_id": event.model_id,
297 "load_duration_ms": event.load_duration_ms,
298 "onde_app_id": event.onde_app_id,
299 }),
300 )
301 .await;
302 }
303
304 /// Mirror an inference event into the append-only `pulse_inference_events`
305 /// collection (no doc_key → server-generated UUID).
306 async fn dual_write_inference(&self, event: &InferenceEvent) {
307 if !Self::dual_write_enabled() {
308 return;
309 }
310
311 self.write_doc(
312 "pulse_inference_events",
313 None,
314 &serde_json::json!({
315 "edge_id": event.edge_id,
316 "model_id": event.model_id,
317 "request_id": event.request_id,
318 "duration_ms": event.duration_ms,
319 "status": event.status,
320 "onde_app_id": event.onde_app_id,
321 }),
322 )
323 .await;
324 }
325
326 /// Upsert (or append, when `key` is `None`) one document, logging failures.
327 async fn write_doc(&self, collection: &str, key: Option<&str>, doc: &serde_json::Value) {
328 if let Err(error) = self.inner.upsert_document(collection, key, doc).await {
329 log::warn!("pulse: dual-write to {} failed: {}", collection, error);
330 }
331 }
332 }