@ondeinference / onde
Code Commits Issues Pulls Cloud Agent CI/CD
copilot/fix-onde-npm-sdk-v1-1-2-release
copilot/assess-inference-engine-complexity copilot/fix-onde-npm-sdk-v1-1-2-release development feature/diffusion-vision main
rs 213 lines 7.66 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 }
27
28 impl PulseClient {
29 /// Returns true when pulse telemetry is disabled explicitly by the host app.
30 pub fn disabled_by_env() -> bool {
31 matches!(
32 std::env::var("ONDE_DISABLE_PULSE")
33 .ok()
34 .as_deref()
35 .map(str::trim)
36 .map(str::to_ascii_lowercase)
37 .as_deref(),
38 Some("1") | Some("true") | Some("yes") | Some("on")
39 )
40 }
41
42 /// Build a pulse client using the GresIQ credentials embedded in the SDK.
43 ///
44 /// Returns `None` if the SDK was compiled without `GRESIQ_API_KEY` /
45 /// `GRESIQ_API_SECRET` (e.g. a local dev build of onde without `.env`).
46 /// In that case telemetry is silently disabled — no panic, no partial state.
47 ///
48 /// `edge_id` is a stable device identifier (installation UUID).
49 /// Pass an empty string to default to `"onde-unknown"`.
50 pub fn new(environment: Environment, edge_id: String) -> Option<Self> {
51 if Self::disabled_by_env() {
52 return None;
53 }
54
55 let (api_key, api_secret) = match environment {
56 Environment::Dev => (EMBEDDED_API_KEY_DEV?, EMBEDDED_API_SECRET_DEV?),
57 Environment::Production => (
58 EMBEDDED_API_KEY_PRODUCTION?,
59 EMBEDDED_API_SECRET_SECRET_PRODUCTION?,
60 ),
61 };
62
63 let edge_id = if edge_id.is_empty() {
64 "onde-unknown".to_string()
65 } else {
66 edge_id
67 };
68
69 // reqwest 0.12.x requires a live Tokio runtime (with I/O reactor)
70 // to construct a Client. GresiqClient::from_credentials calls
71 // reqwest::Client::new() internally. Guard against panics when
72 // called from a thread/context that lacks a Tokio reactor.
73 if tokio::runtime::Handle::try_current().is_err() {
74 log::warn!(
75 "pulse: no Tokio runtime available — \
76 deferring PulseClient creation"
77 );
78 return None;
79 }
80
81 let credentials = GresiqCredentials {
82 api_key,
83 api_secret,
84 };
85
86 let inner = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
87 GresiqClient::from_credentials(environment, credentials)
88 })) {
89 Ok(client) => client,
90 Err(_) => {
91 log::warn!(
92 "pulse: GresiqClient::from_credentials panicked \
93 (likely missing Tokio reactor) — telemetry disabled"
94 );
95 return None;
96 }
97 };
98
99 Some(PulseClient { inner, edge_id })
100 }
101
102 /// Writes the model-load event to the pulse/model_loaded table and
103 /// **awaits** the result before returning.
104 ///
105 /// This must complete before any `record_inference` call for the same
106 /// edge + model is made, because the API enforces a foreign-key
107 /// constraint: an `inference_event` row is rejected with HTTP 404
108 /// ("Call model_loaded first") if no matching `model_loaded` row exists.
109 ///
110 /// Keeping this synchronous with respect to the model-load path means
111 /// the row is always present by the time the first inference event fires.
112 /// A failed write emits a warn! log line — no retry, no queue, and no
113 /// effect on the caller.
114 pub async fn record_model_loaded(
115 &self,
116 model_id: String,
117 model_name: String,
118 load_duration_ms: u64,
119 ) {
120 let client = self.clone();
121 let send_event = async move {
122 let event = ModelLoadedEvent {
123 edge_id: client.edge_id.clone(),
124 model_id,
125 model_name,
126 load_duration_ms,
127 };
128 if let Err(error) = client.inner.insert("pulse/model_loaded", &event).await {
129 log::warn!("pulse: model_loaded failed: {}", error);
130 }
131 };
132
133 if tokio::runtime::Handle::try_current().is_ok() {
134 send_event.await;
135 return;
136 }
137
138 let join = std::thread::spawn(move || {
139 let runtime = tokio::runtime::Builder::new_current_thread()
140 .enable_all()
141 .build();
142
143 match runtime {
144 Ok(runtime) => {
145 runtime.block_on(send_event);
146 }
147 Err(error) => {
148 log::warn!(
149 "pulse: could not create fallback runtime for model_loaded: {}",
150 error
151 );
152 }
153 }
154 });
155
156 if join.join().is_err() {
157 log::warn!("pulse: fallback model_loaded thread panicked");
158 }
159 }
160
161 /// Same fire-and-forget pattern as record_model_loaded but for inference
162 /// completions. Writes to pulse/inference_event. Logs on failure, no retry.
163 ///
164 /// Swift / UniFFI consumers may call into the SDK from contexts where
165 /// `tokio::spawn` does not have a current runtime handle even though the
166 /// outer API is async. To avoid panicking on Apple platforms, we prefer
167 /// the current Tokio runtime when available and otherwise fall back to a
168 /// short-lived current-thread runtime on a native background thread.
169 pub fn record_inference(
170 &self,
171 model_id: String,
172 request_id: String,
173 duration_ms: u64,
174 status: String,
175 ) {
176 let client = self.clone();
177 let send_event = async move {
178 let event = InferenceEvent {
179 edge_id: client.edge_id.clone(),
180 model_id,
181 request_id,
182 duration_ms,
183 status,
184 };
185 if let Err(error) = client.inner.insert("pulse/inference_event", &event).await {
186 log::warn!("pulse: inference_event failed: {}", error);
187 }
188 };
189
190 if let Ok(handle) = tokio::runtime::Handle::try_current() {
191 handle.spawn(send_event);
192 return;
193 }
194
195 std::thread::spawn(move || {
196 let runtime = tokio::runtime::Builder::new_current_thread()
197 .enable_all()
198 .build();
199
200 match runtime {
201 Ok(runtime) => {
202 runtime.block_on(send_event);
203 }
204 Err(error) => {
205 log::warn!(
206 "pulse: could not create fallback runtime for inference_event: {}",
207 error
208 );
209 }
210 }
211 });
212 }
213 }