bhtune_driver/
opcda.rs

1//! `OpcDaDriver`: the primary [`Driver`] implementation, talking to a DCS/PLC's OPC DA
2//! server through the `opcda-bridge` gateway's gRPC API.
3//!
4//! This module intentionally splits into two halves: a thin async shell (`OpcDaDriver`
5//! itself) that only locks the client and calls into `opcda_bridge`, and small mapping
6//! functions that translate the bridge's typed pages, nodes, capabilities, and search events
7//! into the driver crate's protocol-neutral types. The mapping functions carry the real risk
8//! of a subtle bug and are unit-testable with no I/O; the shell is exercised end-to-end by
9//! mock-gateway smoke tests rather than re-testing the bridge's own gRPC matrix.
10
11use async_trait::async_trait;
12use tokio::sync::Mutex;
13
14use crate::{
15    driver::Driver,
16    error::{DriverError, DriverResult},
17    types::{
18        BrowseBreadcrumb, BrowseNode, BrowseNodeKind, BrowsePage, BrowsePageRequest, BrowseSource,
19        DriverCapabilities, IndexSchedulerDiagnostics, IndexedSearchMatch, IndexedSearchProgress,
20        NamespaceOrganization, Quality, SearchCompleted, SearchEvent, SearchIndexControlAction,
21        SearchIndexRequest, SearchIndexResponse, SearchIndexState, SearchIndexStatus, SearchMatch,
22        SearchMatchMode, SearchProgress, SearchRequest, TagId, TagValue, TagWrite, WriteOutcome,
23    },
24};
25
26/// Default number of children requested for one browse page. The bridge enforces its own
27/// maximum; this value keeps the browser responsive and matches the bridge library default.
28pub const DEFAULT_PAGE_SIZE: u32 = opcda_bridge::DEFAULT_PAGE_SIZE;
29
30/// Default maximum number of search matches requested by the CLI and browser.
31pub const DEFAULT_SEARCH_MAX_RESULTS: u32 = opcda_bridge::DEFAULT_SEARCH_MAX_RESULTS;
32
33/// Default maximum number of matches requested from the persistent namespace index.
34pub const DEFAULT_INDEX_SEARCH_MAX_RESULTS: u32 = opcda_bridge::DEFAULT_INDEX_SEARCH_MAX_RESULTS;
35
36/// A cancellable stream of typed namespace-search events.
37#[derive(Debug)]
38pub struct DriverSearchStream {
39    inner: opcda_bridge::SearchStream,
40}
41
42impl DriverSearchStream {
43    /// Waits for the next event. Dropping the stream cancels the gateway-side search.
44    pub async fn next(&mut self) -> DriverResult<Option<SearchEvent>> {
45        let event = self
46            .inner
47            .message()
48            .await
49            .map_err(|err| map_bridge_error_for(err, "namespace search"))?;
50        event.map(search_event_from_bridge).transpose()
51    }
52}
53
54/// The primary [`Driver`] for v1: reads, writes, and browses OPC DA tags through an
55/// `opcda-bridge` gateway's gRPC API.
56///
57/// Holds its `opcda_bridge::Client` behind a `tokio::sync::Mutex` because the client's own
58/// methods take `&mut self` (it buffers per-call gRPC codec state) while [`Driver`]'s
59/// methods take `&self` (required so a driver can be shared behind `Arc<dyn Driver>`) —
60/// serializing calls through the one connection is a reasonable tradeoff, since a single
61/// tuning session only ever has one read/write/browse in flight at a time, and the
62/// underlying HTTP/2 channel stays cheaply multiplexed regardless of how many logical
63/// callers there are.
64#[derive(Debug)]
65pub struct OpcDaDriver {
66    client: Mutex<opcda_bridge::Client>,
67    server: String,
68}
69
70impl OpcDaDriver {
71    /// Connects to an `opcda-bridge` gateway at `host` (e.g. `"localhost:7600"`) and binds
72    /// to `server` — the OPC DA server's ProgID (e.g. `"Matrikon.OPC.Simulation.1"`) that
73    /// every subsequent `read`/`write`/`browse` call is scoped to.
74    pub async fn connect(host: &str, server: impl Into<String>) -> DriverResult<Self> {
75        let client = opcda_bridge::Client::connect(host)
76            .await
77            .map_err(|err| map_bridge_error_for(err, "connect to OPC DA bridge"))?;
78        Ok(Self {
79            client: Mutex::new(client),
80            server: server.into(),
81        })
82    }
83
84    /// Reports the bridge and OPC server's browse/search capabilities.
85    pub async fn capabilities(&self) -> DriverResult<DriverCapabilities> {
86        let mut client = self.client.lock().await;
87        client
88            .capabilities(self.server.clone())
89            .await
90            .map_err(|err| map_bridge_error_for(err, "capability discovery"))
91            .and_then(capabilities_from_bridge)
92    }
93
94    /// Requests one bounded browse page without following its continuation token.
95    pub async fn browse_page(&self, request: BrowsePageRequest) -> DriverResult<BrowsePage> {
96        let mut client = self.client.lock().await;
97        client
98            .browse_page(opcda_bridge::BrowsePageRequest {
99                server: self.server.clone(),
100                session_id: request.session_id,
101                parent_node_key: request.parent_node_key,
102                page_token: request.page_token,
103                page_size: request.page_size,
104                refresh: request.refresh,
105            })
106            .await
107            .map_err(|err| map_bridge_error_for(err, "paged browse"))
108            .and_then(browse_page_from_bridge)
109    }
110
111    /// Starts a progressive namespace search. Dropping the returned stream cancels the search.
112    pub async fn search_stream(&self, request: SearchRequest) -> DriverResult<DriverSearchStream> {
113        let mut client = self.client.lock().await;
114        let bridge_request = opcda_bridge::SearchRequest {
115            server: self.server.clone(),
116            query: request.query,
117            match_mode: match_search_mode(request.match_mode),
118            session_id: request.session_id,
119            scope_node_key: request.scope_node_key,
120            max_results: request.max_results,
121            include_branches: request.include_branches,
122            refresh: request.refresh,
123        };
124        let inner = client
125            .search_stream(bridge_request)
126            .await
127            .map_err(|err| map_bridge_error_for(err, "namespace search"))?;
128        Ok(DriverSearchStream { inner })
129    }
130
131    /// Collects a complete search stream for callers that do not need progressive delivery.
132    pub async fn search_events(&self, request: SearchRequest) -> DriverResult<Vec<SearchEvent>> {
133        let mut stream = self.search_stream(request).await?;
134        let mut events = Vec::new();
135        while let Some(event) = stream.next().await? {
136            events.push(event);
137        }
138        Ok(events)
139    }
140
141    /// Returns the gateway-owned persistent namespace-index status for this OPC server.
142    pub async fn search_index_status(&self) -> DriverResult<SearchIndexStatus> {
143        let mut client = self.client.lock().await;
144        client
145            .search_index_status(self.server.clone())
146            .await
147            .map_err(|err| map_bridge_error_for(err, "indexed-search status"))
148            .map(search_index_status_from_bridge)
149    }
150
151    /// Enables or disables future automatic refreshes for this OPC server's index.
152    pub async fn set_search_index_auto_refresh(
153        &self,
154        enabled: bool,
155    ) -> DriverResult<SearchIndexStatus> {
156        let mut client = self.client.lock().await;
157        client
158            .set_search_index_auto_refresh(self.server.clone(), enabled)
159            .await
160            .map_err(|err| map_bridge_error_for(err, "indexed-search auto-refresh"))
161            .map(search_index_status_from_bridge)
162    }
163
164    /// Deletes this OPC server's persistent namespace index and enrollment.
165    pub async fn delete_search_index(&self) -> DriverResult<SearchIndexStatus> {
166        let mut client = self.client.lock().await;
167        client
168            .delete_search_index(self.server.clone())
169            .await
170            .map_err(|err| map_bridge_error_for(err, "indexed-search delete"))
171            .map(search_index_status_from_bridge)
172    }
173
174    /// Starts or coalesces a persistent namespace-index refresh for this OPC server.
175    pub async fn refresh_search_index(&self, force: bool) -> DriverResult<SearchIndexStatus> {
176        let mut client = self.client.lock().await;
177        client
178            .refresh_search_index(self.server.clone(), force)
179            .await
180            .map_err(|err| map_bridge_error_for(err, "indexed-search refresh"))
181            .map(search_index_status_from_bridge)
182    }
183
184    /// Pauses, resumes, or cancels a persistent namespace-index build.
185    pub async fn control_search_index(
186        &self,
187        action: SearchIndexControlAction,
188    ) -> DriverResult<SearchIndexStatus> {
189        let mut client = self.client.lock().await;
190        client
191            .control_search_index(self.server.clone(), action.into())
192            .await
193            .map_err(|err| map_bridge_error_for(err, "indexed-search control"))
194            .map(search_index_status_from_bridge)
195    }
196
197    /// Queries the gateway-owned persistent namespace index without falling back to live
198    /// namespace traversal.
199    pub async fn search_index_query(
200        &self,
201        request: SearchIndexRequest,
202    ) -> DriverResult<SearchIndexResponse> {
203        let mut client = self.client.lock().await;
204        client
205            .search_index(opcda_bridge::SearchIndexRequest {
206                server: self.server.clone(),
207                query: request.query,
208                match_mode: match_search_mode(request.match_mode),
209                max_results: request.max_results,
210            })
211            .await
212            .map_err(|err| map_bridge_error_for(err, "indexed search"))
213            .map(search_index_response_from_bridge)
214    }
215}
216
217/// Lists the OPC DA servers registered on the `opcda-bridge` gateway's own host at
218/// `bridge_host` (e.g. `"localhost:7600"`).
219///
220/// A standalone free function rather than a [`Driver`] method or an [`OpcDaDriver`]
221/// associated function: server discovery is a *pre-connection* operation — it needs only a
222/// bridge host, not the OPC DA server ProgID that [`OpcDaDriver::connect`] requires and that
223/// discovery exists to help a caller find in the first place. Connects for the one call and
224/// drops the connection immediately afterward; unlike `OpcDaDriver`, there is no ongoing
225/// session to hold open here.
226///
227/// Note: `opcda_bridge::Client::list_servers` always sends `host: "localhost"` in its
228/// request — i.e. it lists servers registered on *the gateway's own* machine, not on
229/// whatever machine bhtune itself happens to run on. That is exactly right for this
230/// topology (the gateway runs next to the OPC DA server; bhtune runs wherever the
231/// engineer's browser or scheduler is) and is called out here so it is never mistaken for a
232/// bug.
233pub async fn list_opcda_servers(bridge_host: &str) -> DriverResult<Vec<String>> {
234    let mut client = opcda_bridge::Client::connect(bridge_host)
235        .await
236        .map_err(|err| map_bridge_error_for(err, "connect to OPC DA bridge"))?;
237    client
238        .list_servers()
239        .await
240        .map_err(|err| map_bridge_error_for(err, "list OPC DA servers"))
241}
242
243/// Explicitly releases one gateway-side browse session without requiring an OPC server
244/// ProgID. The session ID is the only value the bridge close RPC needs, so this is separate
245/// from [`OpcDaDriver`] for callers such as the CLI's `opc close` command that may no longer
246/// have the server name handy.
247pub async fn close_opcda_browse_session(bridge_host: &str, session_id: &str) -> DriverResult<()> {
248    let mut client = opcda_bridge::Client::connect(bridge_host)
249        .await
250        .map_err(|err| map_bridge_error_for(err, "connect to OPC DA bridge"))?;
251    client
252        .close_browse_session(session_id)
253        .await
254        .map_err(|err| map_bridge_error_for(err, "browse-session close"))
255}
256
257#[async_trait]
258impl Driver for OpcDaDriver {
259    async fn read(&self, tags: &[TagId]) -> DriverResult<Vec<TagValue>> {
260        let mut client = self.client.lock().await;
261        let raw = client
262            .read(self.server.clone(), tags.to_vec())
263            .await
264            .map_err(|err| map_bridge_error_for(err, "read OPC DA tags"))?;
265        Ok(raw.into_iter().map(tag_value_from_raw).collect())
266    }
267
268    async fn write(&self, tag: &TagId, value: TagWrite) -> DriverResult<WriteOutcome> {
269        let mut client = self.client.lock().await;
270        let result = client
271            .write(
272                self.server.clone(),
273                tag.clone(),
274                opc_value_from_write(value),
275            )
276            .await
277            .map_err(|err| map_bridge_error_for(err, "write OPC DA tag"))?;
278        Ok(write_outcome_from_result(result))
279    }
280
281    async fn capabilities(&self) -> DriverResult<DriverCapabilities> {
282        self.capabilities().await
283    }
284
285    async fn browse(&self, request: BrowsePageRequest) -> DriverResult<BrowsePage> {
286        self.browse_page(request).await
287    }
288
289    async fn close_browse_session(&self, session_id: &str) -> DriverResult<()> {
290        let mut client = self.client.lock().await;
291        client
292            .close_browse_session(session_id)
293            .await
294            .map_err(|err| map_bridge_error_for(err, "browse-session close"))
295    }
296
297    async fn search(&self, request: SearchRequest) -> DriverResult<Vec<SearchEvent>> {
298        self.search_events(request).await
299    }
300
301    async fn search_index_status(&self) -> DriverResult<SearchIndexStatus> {
302        self.search_index_status().await
303    }
304
305    async fn refresh_search_index(&self, force: bool) -> DriverResult<SearchIndexStatus> {
306        self.refresh_search_index(force).await
307    }
308
309    async fn control_search_index(
310        &self,
311        action: SearchIndexControlAction,
312    ) -> DriverResult<SearchIndexStatus> {
313        self.control_search_index(action).await
314    }
315
316    async fn set_search_index_auto_refresh(
317        &self,
318        enabled: bool,
319    ) -> DriverResult<SearchIndexStatus> {
320        self.set_search_index_auto_refresh(enabled).await
321    }
322
323    async fn delete_search_index(&self) -> DriverResult<SearchIndexStatus> {
324        self.delete_search_index().await
325    }
326
327    async fn search_index(&self, request: SearchIndexRequest) -> DriverResult<SearchIndexResponse> {
328        self.search_index_query(request).await
329    }
330}
331
332/// Maps `opcda_bridge`'s raw OPC quality string to [`Quality`].
333///
334/// Per `opc-da-client`'s documented contract (the Windows-only library the gateway wraps on
335/// the other side), this is one of `"Good"`, `"Bad"`, `"Uncertain"`, or a synthesized
336/// `"Unknown(0xNNNN)"` for an OPC quality code the library doesn't otherwise recognize. Any
337/// string other than an exact `"Good"`/`"Uncertain"` match — including that
338/// `"Unknown(...)"` case — is treated as [`Quality::Bad`]: an unrecognized quality is
339/// exactly the situation where guessing "trustworthy" would be the wrong default.
340pub fn quality_from_raw(raw: &str) -> Quality {
341    match raw {
342        "Good" => Quality::Good,
343        "Uncertain" => Quality::Uncertain,
344        _ => Quality::Bad,
345    }
346}
347
348/// Maps one `opcda_bridge::TagValue` (a single tag's raw read result) to this crate's
349/// [`TagValue`].
350pub fn tag_value_from_raw(raw: opcda_bridge::TagValue) -> TagValue {
351    TagValue {
352        tag: raw.tag_id,
353        value: raw.value,
354        quality: quality_from_raw(&raw.quality),
355        // The gateway reports each tag's last-change time as a *local*, offset-less
356        // "YYYY-MM-DD HH:MM:SS" string (or a "N/A"/"Invalid" sentinel for tags that have
357        // none), per `opc-da-client`'s documented contract. There is no reliable way to
358        // convert that into a trustworthy `DateTime<Utc>` without knowing the gateway
359        // host's timezone, which isn't part of the bridge protocol and can't safely be
360        // assumed to match wherever `bhtune` itself runs — so this is always `None` rather
361        // than a guess. Purely diagnostic regardless (see `TagValue::timestamp`'s doc
362        // comment): never the tick time the tuning engine itself runs on.
363        timestamp: None,
364    }
365}
366
367/// Maps a [`TagWrite`] to the `opcda_bridge::Value` its `Client::write` expects.
368pub fn opc_value_from_write(write: TagWrite) -> opcda_bridge::Value {
369    match write {
370        TagWrite::Float(f) => opcda_bridge::Value::Float(f64::from(f)),
371        TagWrite::Raw(s) => opcda_bridge::Value::String(s),
372    }
373}
374
375/// Maps an `opcda_bridge::WriteResult` (the RPC-level outcome of one write) to
376/// [`WriteOutcome`].
377pub fn write_outcome_from_result(result: opcda_bridge::WriteResult) -> WriteOutcome {
378    if result.success {
379        WriteOutcome::success()
380    } else {
381        WriteOutcome::failure(
382            result
383                .error
384                .unwrap_or_else(|| "gateway rejected the write".to_string()),
385        )
386    }
387}
388
389/// Maps the bridge's capabilities into the protocol-neutral driver model.
390pub fn capabilities_from_bridge(
391    capabilities: opcda_bridge::Capabilities,
392) -> DriverResult<DriverCapabilities> {
393    Ok(DriverCapabilities {
394        application_version: capabilities.application_version,
395        protocol_version: capabilities.protocol_version,
396        max_page_size: capabilities.max_page_size,
397        supports_browse_sessions: capabilities.supports_browse_sessions,
398        supports_search: capabilities.supports_search,
399        organization: namespace_organization_from_bridge(capabilities.organization),
400        source: browse_source_from_bridge(capabilities.source),
401        supports_indexed_search: capabilities.supports_indexed_search,
402        indexed_search_protocol_version: capabilities.indexed_search_protocol_version,
403        max_indexed_search_results: capabilities.max_indexed_search_results,
404        search_index_state: search_index_state_from_bridge(capabilities.search_index_state),
405    })
406}
407
408pub fn search_index_state_from_bridge(state: opcda_bridge::SearchIndexState) -> SearchIndexState {
409    match state {
410        opcda_bridge::SearchIndexState::Unspecified => SearchIndexState::Unspecified,
411        opcda_bridge::SearchIndexState::NotIndexed => SearchIndexState::NotIndexed,
412        opcda_bridge::SearchIndexState::Partial => SearchIndexState::Partial,
413        opcda_bridge::SearchIndexState::Ready => SearchIndexState::Ready,
414        opcda_bridge::SearchIndexState::Stale => SearchIndexState::Stale,
415        opcda_bridge::SearchIndexState::Refreshing => SearchIndexState::Refreshing,
416        opcda_bridge::SearchIndexState::Promoting => SearchIndexState::Promoting,
417        opcda_bridge::SearchIndexState::Failed => SearchIndexState::Failed,
418        opcda_bridge::SearchIndexState::Deleting => SearchIndexState::Deleting,
419    }
420}
421
422pub fn indexed_search_progress_from_bridge(
423    progress: opcda_bridge::IndexedSearchProgress,
424) -> IndexedSearchProgress {
425    IndexedSearchProgress {
426        branches_visited: progress.branches_visited,
427        entries_seen: progress.entries_seen,
428        unique_items: progress.unique_items,
429        active_time_ms: progress.active_time_ms,
430        paused_time_ms: progress.paused_time_ms,
431        items_per_second: progress.items_per_second,
432        estimated_remaining_ms: progress.estimated_remaining_ms,
433    }
434}
435
436pub fn search_index_status_from_bridge(
437    status: opcda_bridge::SearchIndexStatus,
438) -> SearchIndexStatus {
439    SearchIndexStatus {
440        server: status.server,
441        state: search_index_state_from_bridge(status.state),
442        auto_refresh_enabled: status.auto_refresh_enabled,
443        active_generation: status.active_generation,
444        entry_count: status.entry_count,
445        unique_item_count: status.unique_item_count,
446        started_at: status.started_at,
447        completed_at: status.completed_at,
448        last_error: status.last_error,
449        database_bytes: status.database_bytes,
450        organization: namespace_organization_from_bridge(status.organization),
451        source: browse_source_from_bridge(status.source),
452        progress: status.progress.map(indexed_search_progress_from_bridge),
453        scheduler: IndexSchedulerDiagnostics {
454            next_refresh_at: status.scheduler.next_refresh_at,
455            last_attempt_at: status.scheduler.last_attempt_at,
456            last_success_at: status.scheduler.last_success_at,
457            last_success_duration_ms: status.scheduler.last_success_duration_ms,
458            retry_after: status.scheduler.retry_after,
459            consecutive_failures: status.scheduler.consecutive_failures,
460            circuit_open: status.scheduler.circuit_open,
461        },
462    }
463}
464
465pub fn indexed_search_match_from_bridge(
466    found: opcda_bridge::IndexedSearchMatch,
467) -> IndexedSearchMatch {
468    IndexedSearchMatch {
469        item_id: found.item_id,
470        display_name: found.display_name,
471        kind: match found.kind {
472            opcda_bridge::BrowseNodeKind::Unspecified => BrowseNodeKind::Unspecified,
473            opcda_bridge::BrowseNodeKind::Branch => BrowseNodeKind::Branch,
474            opcda_bridge::BrowseNodeKind::Item => BrowseNodeKind::Item,
475            opcda_bridge::BrowseNodeKind::BranchAndItem => BrowseNodeKind::BranchAndItem,
476        },
477        breadcrumbs: found.breadcrumbs,
478    }
479}
480
481pub fn search_index_response_from_bridge(
482    response: opcda_bridge::SearchIndexResponse,
483) -> SearchIndexResponse {
484    SearchIndexResponse {
485        matches: response
486            .matches
487            .into_iter()
488            .map(indexed_search_match_from_bridge)
489            .collect(),
490        has_more: response.has_more,
491        status: search_index_status_from_bridge(response.status),
492    }
493}
494
495impl From<SearchIndexControlAction> for opcda_bridge::SearchIndexControlAction {
496    fn from(action: SearchIndexControlAction) -> Self {
497        match action {
498            SearchIndexControlAction::Pause => Self::Pause,
499            SearchIndexControlAction::Resume => Self::Resume,
500            SearchIndexControlAction::Cancel => Self::Cancel,
501        }
502    }
503}
504
505pub fn namespace_organization_from_bridge(
506    organization: opcda_bridge::NamespaceOrganization,
507) -> NamespaceOrganization {
508    match organization {
509        opcda_bridge::NamespaceOrganization::Unspecified => NamespaceOrganization::Unspecified,
510        opcda_bridge::NamespaceOrganization::Flat => NamespaceOrganization::Flat,
511        opcda_bridge::NamespaceOrganization::Hierarchical => NamespaceOrganization::Hierarchical,
512    }
513}
514
515pub fn browse_source_from_bridge(source: opcda_bridge::BrowseSource) -> BrowseSource {
516    match source {
517        opcda_bridge::BrowseSource::Unspecified => BrowseSource::Unspecified,
518        opcda_bridge::BrowseSource::Da3 => BrowseSource::Da3,
519        opcda_bridge::BrowseSource::Da2 => BrowseSource::Da2,
520        opcda_bridge::BrowseSource::Flat => BrowseSource::Flat,
521        opcda_bridge::BrowseSource::Derived => BrowseSource::Derived,
522    }
523}
524
525pub fn browse_node_from_bridge(node: opcda_bridge::BrowseNode) -> BrowseNode {
526    BrowseNode {
527        node_key: node.node_key,
528        display_name: node.display_name,
529        kind: match node.kind {
530            opcda_bridge::BrowseNodeKind::Unspecified => BrowseNodeKind::Unspecified,
531            opcda_bridge::BrowseNodeKind::Branch => BrowseNodeKind::Branch,
532            opcda_bridge::BrowseNodeKind::Item => BrowseNodeKind::Item,
533            opcda_bridge::BrowseNodeKind::BranchAndItem => BrowseNodeKind::BranchAndItem,
534        },
535        item_id: node.item_id,
536    }
537}
538
539pub fn browse_page_from_bridge(page: opcda_bridge::BrowsePage) -> DriverResult<BrowsePage> {
540    Ok(BrowsePage {
541        session_id: page.session_id,
542        nodes: page
543            .nodes
544            .into_iter()
545            .map(browse_node_from_bridge)
546            .collect(),
547        next_page_token: page.next_page_token,
548        complete: page.complete,
549        organization: namespace_organization_from_bridge(page.organization),
550        source: browse_source_from_bridge(page.source),
551        warning: page.warning,
552    })
553}
554
555pub fn search_event_from_bridge(event: opcda_bridge::SearchEvent) -> DriverResult<SearchEvent> {
556    match event {
557        opcda_bridge::SearchEvent::Match(found) => Ok(SearchEvent::Match(SearchMatch {
558            node: browse_node_from_bridge(found.node),
559            breadcrumbs: found
560                .breadcrumbs
561                .into_iter()
562                .map(|part| BrowseBreadcrumb {
563                    node_key: part.node_key,
564                    display_name: part.display_name,
565                })
566                .collect(),
567        })),
568        opcda_bridge::SearchEvent::Progress(progress) => {
569            Ok(SearchEvent::Progress(SearchProgress {
570                visited_nodes: progress.visited_nodes,
571                matches: progress.matches,
572                partial: progress.partial,
573            }))
574        }
575        opcda_bridge::SearchEvent::Completed(completed) => {
576            Ok(SearchEvent::Completed(SearchCompleted {
577                complete: completed.complete,
578                cancelled: completed.cancelled,
579                truncated: completed.truncated,
580                warning: completed.warning,
581            }))
582        }
583    }
584}
585
586fn match_search_mode(mode: SearchMatchMode) -> opcda_bridge::SearchMatchMode {
587    match mode {
588        SearchMatchMode::Exact => opcda_bridge::SearchMatchMode::Exact,
589        SearchMatchMode::Prefix => opcda_bridge::SearchMatchMode::Prefix,
590        SearchMatchMode::Contains => opcda_bridge::SearchMatchMode::Contains,
591    }
592}
593
594/// Maps every `opcda_bridge::Error` this driver can encounter — at connect time or during
595/// any RPC — to the matching [`DriverError`] variant: `opcda_bridge::Error::Connect` (the
596/// gRPC channel itself couldn't be established) becomes [`DriverError::Connect`], and
597/// `opcda_bridge::Error::Rpc` (the channel is fine, but the gateway returned a gRPC error
598/// for this specific call) becomes [`DriverError::Operation`], except for an indexed-search
599/// `FailedPrecondition`, which becomes [`DriverError::IndexOperationRejected`] so gateway
600/// configuration and concurrency diagnostics remain actionable. An exhaustive match rather
601/// than a wildcard arm, deliberately: if `opcda_bridge::Error` ever gains a new variant,
602/// this should fail to compile and force a real decision about where it belongs, not
603/// silently fall into one bucket.
604fn map_bridge_error_for(err: opcda_bridge::Error, operation: &'static str) -> DriverError {
605    match &err {
606        opcda_bridge::Error::Connect(_) => DriverError::Connect(Box::new(err)),
607        opcda_bridge::Error::Rpc(status)
608            if is_indexed_search_operation(operation)
609                && matches!(
610                    status.code(),
611                    tonic::Code::InvalidArgument
612                        | tonic::Code::NotFound
613                        | tonic::Code::AlreadyExists
614                        | tonic::Code::FailedPrecondition
615                ) =>
616        {
617            DriverError::IndexOperationRejected {
618                message: status.message().to_string(),
619            }
620        }
621        opcda_bridge::Error::Rpc(status)
622            if (operation == "paged browse" || operation == "browse-session close")
623                && matches!(
624                    status.code(),
625                    tonic::Code::NotFound | tonic::Code::FailedPrecondition
626                ) =>
627        {
628            DriverError::BrowseStateInvalid
629        }
630        opcda_bridge::Error::UnknownIndexServer { .. }
631        | opcda_bridge::Error::IndexNotEnrolled { .. }
632        | opcda_bridge::Error::IndexDeleting { .. } => DriverError::IndexOperationRejected {
633            message: err.to_string(),
634        },
635        opcda_bridge::Error::IncompatibleGateway { .. } => {
636            DriverError::IncompatibleGateway { operation }
637        }
638        opcda_bridge::Error::Rpc(_) | opcda_bridge::Error::Protocol(_) => {
639            DriverError::Operation(Box::new(err))
640        }
641    }
642}
643
644fn is_indexed_search_operation(operation: &str) -> bool {
645    matches!(
646        operation,
647        "indexed-search status"
648            | "indexed-search refresh"
649            | "indexed-search control"
650            | "indexed-search auto-refresh"
651            | "indexed-search delete"
652            | "indexed search"
653    )
654}
655
656#[cfg(test)]
657mod tests {
658    use super::*;
659
660    #[test]
661    fn quality_from_raw_matches_good_and_uncertain_exactly() {
662        assert_eq!(quality_from_raw("Good"), Quality::Good);
663        assert_eq!(quality_from_raw("Uncertain"), Quality::Uncertain);
664    }
665
666    #[test]
667    fn quality_from_raw_treats_bad_as_bad() {
668        assert_eq!(quality_from_raw("Bad"), Quality::Bad);
669    }
670
671    #[test]
672    fn quality_from_raw_treats_unrecognized_codes_as_bad() {
673        // `opc-da-client` synthesizes "Unknown(0xNNNN)" for quality codes it doesn't
674        // otherwise recognize; an unrecognized quality must never be silently trusted.
675        assert_eq!(quality_from_raw("Unknown(0x1234)"), Quality::Bad);
676        assert_eq!(quality_from_raw(""), Quality::Bad);
677    }
678
679    #[test]
680    fn tag_value_from_raw_maps_fields_and_drops_the_unreliable_timestamp() {
681        let raw = opcda_bridge::TagValue {
682            tag_id: "Area1.LIC101.PV".to_string(),
683            value: "42.5".to_string(),
684            quality: "Good".to_string(),
685            timestamp: "2024-01-15 10:23:45".to_string(),
686        };
687        let value = tag_value_from_raw(raw);
688        assert_eq!(value.tag, "Area1.LIC101.PV");
689        assert_eq!(value.value, "42.5");
690        assert_eq!(value.quality, Quality::Good);
691        assert_eq!(value.timestamp, None);
692    }
693
694    #[test]
695    fn tag_value_from_raw_drops_timestamp_even_for_na_sentinel() {
696        let raw = opcda_bridge::TagValue {
697            tag_id: "t".to_string(),
698            value: "0".to_string(),
699            quality: "Bad".to_string(),
700            timestamp: "N/A".to_string(),
701        };
702        assert_eq!(tag_value_from_raw(raw).timestamp, None);
703    }
704
705    #[test]
706    fn opc_value_from_write_maps_float() {
707        assert_eq!(
708            opc_value_from_write(TagWrite::Float(55.5)),
709            opcda_bridge::Value::Float(55.5)
710        );
711    }
712
713    #[test]
714    fn opc_value_from_write_maps_raw_string() {
715        assert_eq!(
716            opc_value_from_write(TagWrite::Raw("AUT".into())),
717            opcda_bridge::Value::String("AUT".to_string())
718        );
719    }
720
721    #[test]
722    fn write_outcome_from_result_maps_success() {
723        let result = opcda_bridge::WriteResult {
724            tag_id: "t".to_string(),
725            success: true,
726            error: None,
727        };
728        let outcome = write_outcome_from_result(result);
729        assert!(outcome.success);
730        assert_eq!(outcome.error_message, None);
731    }
732
733    #[test]
734    fn write_outcome_from_result_maps_failure_with_gateway_message() {
735        let result = opcda_bridge::WriteResult {
736            tag_id: "t".to_string(),
737            success: false,
738            error: Some("access denied".to_string()),
739        };
740        let outcome = write_outcome_from_result(result);
741        assert!(!outcome.success);
742        assert_eq!(outcome.error_message.as_deref(), Some("access denied"));
743    }
744
745    #[test]
746    fn write_outcome_from_result_synthesizes_a_message_when_the_gateway_gave_none() {
747        let result = opcda_bridge::WriteResult {
748            tag_id: "t".to_string(),
749            success: false,
750            error: None,
751        };
752        let outcome = write_outcome_from_result(result);
753        assert!(!outcome.success);
754        assert!(outcome.error_message.is_some());
755    }
756
757    #[test]
758    fn browse_node_from_bridge_preserves_opaque_identity_and_exact_item_id() {
759        let node = browse_node_from_bridge(opcda_bridge::BrowseNode {
760            node_key: "opaque-node".to_string(),
761            display_name: "PV".to_string(),
762            kind: opcda_bridge::BrowseNodeKind::BranchAndItem,
763            item_id: Some("FCS0201!204FI00510.PV".to_string()),
764        });
765        assert_eq!(node.node_key, "opaque-node");
766        assert_eq!(node.display_name, "PV");
767        assert!(node.kind.is_branch());
768        assert!(node.kind.is_item());
769        assert_eq!(node.item_id.as_deref(), Some("FCS0201!204FI00510.PV"));
770    }
771
772    #[test]
773    fn browse_page_from_bridge_maps_nodes_and_continuation_metadata() {
774        let page = browse_page_from_bridge(opcda_bridge::BrowsePage {
775            session_id: "session".to_string(),
776            nodes: vec![opcda_bridge::BrowseNode {
777                node_key: "node".to_string(),
778                display_name: "PV".to_string(),
779                kind: opcda_bridge::BrowseNodeKind::Item,
780                item_id: Some("Area1.LIC101.PV".to_string()),
781            }],
782            next_page_token: Some("next".to_string()),
783            complete: false,
784            organization: opcda_bridge::NamespaceOrganization::Hierarchical,
785            source: opcda_bridge::BrowseSource::Da2,
786            warning: Some("partial".to_string()),
787        })
788        .unwrap();
789        assert_eq!(page.session_id, "session");
790        assert_eq!(page.nodes[0].item_id.as_deref(), Some("Area1.LIC101.PV"));
791        assert_eq!(page.next_page_token.as_deref(), Some("next"));
792        assert!(!page.complete);
793        assert_eq!(page.warning.as_deref(), Some("partial"));
794    }
795
796    #[test]
797    fn protocol_enum_mappers_cover_every_wire_variant() {
798        for (wire, expected) in [
799            (
800                opcda_bridge::NamespaceOrganization::Unspecified,
801                NamespaceOrganization::Unspecified,
802            ),
803            (
804                opcda_bridge::NamespaceOrganization::Flat,
805                NamespaceOrganization::Flat,
806            ),
807            (
808                opcda_bridge::NamespaceOrganization::Hierarchical,
809                NamespaceOrganization::Hierarchical,
810            ),
811        ] {
812            assert_eq!(namespace_organization_from_bridge(wire), expected);
813        }
814        for (wire, expected) in [
815            (
816                opcda_bridge::BrowseSource::Unspecified,
817                BrowseSource::Unspecified,
818            ),
819            (opcda_bridge::BrowseSource::Da3, BrowseSource::Da3),
820            (opcda_bridge::BrowseSource::Da2, BrowseSource::Da2),
821            (opcda_bridge::BrowseSource::Flat, BrowseSource::Flat),
822            (opcda_bridge::BrowseSource::Derived, BrowseSource::Derived),
823        ] {
824            assert_eq!(browse_source_from_bridge(wire), expected);
825        }
826        for (wire, expected) in [
827            (
828                opcda_bridge::SearchIndexState::Unspecified,
829                SearchIndexState::Unspecified,
830            ),
831            (
832                opcda_bridge::SearchIndexState::NotIndexed,
833                SearchIndexState::NotIndexed,
834            ),
835            (
836                opcda_bridge::SearchIndexState::Partial,
837                SearchIndexState::Partial,
838            ),
839            (
840                opcda_bridge::SearchIndexState::Ready,
841                SearchIndexState::Ready,
842            ),
843            (
844                opcda_bridge::SearchIndexState::Stale,
845                SearchIndexState::Stale,
846            ),
847            (
848                opcda_bridge::SearchIndexState::Refreshing,
849                SearchIndexState::Refreshing,
850            ),
851            (
852                opcda_bridge::SearchIndexState::Promoting,
853                SearchIndexState::Promoting,
854            ),
855            (
856                opcda_bridge::SearchIndexState::Failed,
857                SearchIndexState::Failed,
858            ),
859            (
860                opcda_bridge::SearchIndexState::Deleting,
861                SearchIndexState::Deleting,
862            ),
863        ] {
864            assert_eq!(search_index_state_from_bridge(wire), expected);
865        }
866        for (wire, expected) in [
867            (SearchMatchMode::Exact, opcda_bridge::SearchMatchMode::Exact),
868            (
869                SearchMatchMode::Prefix,
870                opcda_bridge::SearchMatchMode::Prefix,
871            ),
872            (
873                SearchMatchMode::Contains,
874                opcda_bridge::SearchMatchMode::Contains,
875            ),
876        ] {
877            assert_eq!(match_search_mode(wire), expected);
878        }
879        for (wire, expected) in [
880            (
881                SearchIndexControlAction::Pause,
882                opcda_bridge::SearchIndexControlAction::Pause,
883            ),
884            (
885                SearchIndexControlAction::Resume,
886                opcda_bridge::SearchIndexControlAction::Resume,
887            ),
888            (
889                SearchIndexControlAction::Cancel,
890                opcda_bridge::SearchIndexControlAction::Cancel,
891            ),
892        ] {
893            assert_eq!(
894                <opcda_bridge::SearchIndexControlAction as From<_>>::from(wire),
895                expected
896            );
897        }
898    }
899
900    #[test]
901    fn indexed_search_mappers_cover_all_node_kinds_and_optional_progress() {
902        let kinds = [
903            opcda_bridge::BrowseNodeKind::Unspecified,
904            opcda_bridge::BrowseNodeKind::Branch,
905            opcda_bridge::BrowseNodeKind::Item,
906            opcda_bridge::BrowseNodeKind::BranchAndItem,
907        ];
908        for kind in kinds {
909            let found = indexed_search_match_from_bridge(opcda_bridge::IndexedSearchMatch {
910                item_id: "item".into(),
911                display_name: "Item".into(),
912                kind,
913                breadcrumbs: vec!["Area".into()],
914            });
915            let expected = match kind {
916                opcda_bridge::BrowseNodeKind::Unspecified => BrowseNodeKind::Unspecified,
917                opcda_bridge::BrowseNodeKind::Branch => BrowseNodeKind::Branch,
918                opcda_bridge::BrowseNodeKind::Item => BrowseNodeKind::Item,
919                opcda_bridge::BrowseNodeKind::BranchAndItem => BrowseNodeKind::BranchAndItem,
920            };
921            assert_eq!(found.kind, expected);
922        }
923        let status = search_index_status_from_bridge(opcda_bridge::SearchIndexStatus {
924            server: "S".into(),
925            state: opcda_bridge::SearchIndexState::Partial,
926            auto_refresh_enabled: true,
927            active_generation: 2,
928            entry_count: 3,
929            unique_item_count: 4,
930            started_at: None,
931            completed_at: None,
932            last_error: Some("warning".into()),
933            database_bytes: 5,
934            organization: opcda_bridge::NamespaceOrganization::Flat,
935            source: opcda_bridge::BrowseSource::Derived,
936            progress: None,
937            effective_limits: None,
938            controller_state: opcda_bridge::IndexControllerState::Unspecified,
939            pause_reason: None,
940            recovery_deadline: None,
941            pause_reason_detail: None,
942            foreground: opcda_bridge::IndexForegroundDiagnostics {
943                active_count: 0,
944                operations: 0,
945                errors: 0,
946                bad_quality: 0,
947                latency_p50_ms: None,
948                latency_p95_ms: None,
949                latency_max_ms: None,
950                last_error: false,
951                last_bad_quality: false,
952            },
953            host: opcda_bridge::IndexHostDiagnostics::default(),
954            storage: opcda_bridge::IndexStorageDiagnostics::default(),
955            scheduler: opcda_bridge::IndexSchedulerDiagnostics::default(),
956            health: opcda_bridge::IndexHealthDiagnostics::default(),
957            promoting: false,
958        });
959        assert_eq!(status.state, SearchIndexState::Partial);
960        assert!(status.progress.is_none());
961    }
962
963    #[test]
964    fn search_event_mapper_handles_match_progress_and_completion() {
965        let events = [
966            opcda_bridge::SearchEvent::Match(opcda_bridge::SearchMatch {
967                node: opcda_bridge::BrowseNode {
968                    node_key: "n".into(),
969                    display_name: "PV".into(),
970                    kind: opcda_bridge::BrowseNodeKind::Item,
971                    item_id: Some("PV".into()),
972                },
973                breadcrumbs: vec![opcda_bridge::BrowseBreadcrumb {
974                    node_key: "root".into(),
975                    display_name: "Root".into(),
976                }],
977            }),
978            opcda_bridge::SearchEvent::Progress(opcda_bridge::SearchProgress {
979                visited_nodes: 2,
980                matches: 1,
981                partial: true,
982            }),
983            opcda_bridge::SearchEvent::Completed(opcda_bridge::SearchCompleted {
984                complete: false,
985                cancelled: true,
986                truncated: true,
987                warning: Some("partial".into()),
988            }),
989        ];
990        assert!(matches!(
991            search_event_from_bridge(events[0].clone()).unwrap(),
992            SearchEvent::Match(_)
993        ));
994        assert!(matches!(
995            search_event_from_bridge(events[1].clone()).unwrap(),
996            SearchEvent::Progress(SearchProgress {
997                visited_nodes: 2,
998                matches: 1,
999                partial: true
1000            })
1001        ));
1002        assert!(matches!(
1003            search_event_from_bridge(events[2].clone()).unwrap(),
1004            SearchEvent::Completed(SearchCompleted {
1005                cancelled: true,
1006                truncated: true,
1007                ..
1008            })
1009        ));
1010    }
1011
1012    #[test]
1013    fn indexed_search_precondition_preserves_gateway_reason() {
1014        let err = map_bridge_error_for(
1015            opcda_bridge::Error::Rpc(tonic::Status::not_found(
1016                "server is not enrolled for namespace indexing",
1017            )),
1018            "indexed-search refresh",
1019        );
1020        assert!(matches!(
1021            err,
1022            DriverError::IndexOperationRejected { message }
1023                if message == "server is not enrolled for namespace indexing"
1024        ));
1025    }
1026
1027    #[test]
1028    fn indexed_search_invalid_argument_is_a_rejected_operation() {
1029        let err = map_bridge_error_for(
1030            opcda_bridge::Error::Rpc(tonic::Status::invalid_argument(
1031                "OPC server is not registered on the gateway",
1032            )),
1033            "indexed-search refresh",
1034        );
1035        assert!(matches!(
1036            err,
1037            DriverError::IndexOperationRejected { message }
1038                if message == "OPC server is not registered on the gateway"
1039        ));
1040    }
1041
1042    #[test]
1043    fn bridge_error_mapping_distinguishes_browse_state_and_gateway_compatibility() {
1044        assert!(matches!(
1045            map_bridge_error_for(
1046                opcda_bridge::Error::Rpc(tonic::Status::not_found("gone")),
1047                "paged browse",
1048            ),
1049            DriverError::BrowseStateInvalid
1050        ));
1051        assert!(matches!(
1052            map_bridge_error_for(
1053                opcda_bridge::Error::Rpc(tonic::Status::failed_precondition("gone")),
1054                "browse-session close",
1055            ),
1056            DriverError::BrowseStateInvalid
1057        ));
1058        assert!(matches!(
1059            map_bridge_error_for(
1060                opcda_bridge::Error::Rpc(tonic::Status::internal("boom")),
1061                "read OPC DA tags",
1062            ),
1063            DriverError::Operation(_)
1064        ));
1065        assert!(matches!(
1066            map_bridge_error_for(
1067                opcda_bridge::Error::IncompatibleGateway {
1068                    operation: "capability discovery",
1069                },
1070                "capability discovery",
1071            ),
1072            DriverError::IncompatibleGateway {
1073                operation: "capability discovery"
1074            }
1075        ));
1076        assert!(matches!(
1077            map_bridge_error_for(
1078                opcda_bridge::Error::Protocol("bad payload".into()),
1079                "read OPC DA tags",
1080            ),
1081            DriverError::Operation(_)
1082        ));
1083    }
1084
1085    #[test]
1086    fn bridge_error_mapping_preserves_typed_index_enrollment_errors() {
1087        for error in [
1088            opcda_bridge::Error::UnknownIndexServer {
1089                server: "Unknown.Server".into(),
1090            },
1091            opcda_bridge::Error::IndexNotEnrolled {
1092                server: "Known.Server".into(),
1093            },
1094        ] {
1095            assert!(matches!(
1096                map_bridge_error_for(error, "indexed-search refresh"),
1097                DriverError::IndexOperationRejected { message }
1098                    if message.contains("Server")
1099            ));
1100        }
1101    }
1102
1103    #[tokio::test]
1104    async fn connect_failure_maps_to_driver_error_connect() {
1105        // Nothing is listening on this port, so `Client::connect` fails at the transport
1106        // level before any RPC is attempted -- exactly the `DriverError::Connect` case.
1107        let err = OpcDaDriver::connect("127.0.0.1:1", "AnyServer")
1108            .await
1109            .unwrap_err();
1110        assert!(matches!(err, DriverError::Connect(_)));
1111    }
1112
1113    #[tokio::test]
1114    async fn list_opcda_servers_connect_failure_maps_to_driver_error_connect() {
1115        let err = list_opcda_servers("127.0.0.1:1").await.unwrap_err();
1116        assert!(matches!(err, DriverError::Connect(_)));
1117    }
1118}
1119
1120/// End-to-end smoke tests against a minimal mock `Bridge` gRPC service. These prove the typed
1121/// page/session/search API is wired together correctly without re-testing the bridge's own RPC
1122/// implementation.
1123#[cfg(test)]
1124mod smoke_tests {
1125    use super::*;
1126    use opcda_bridge_proto::bridge::bridge_server::{Bridge, BridgeServer};
1127    use opcda_bridge_proto::bridge::search_event;
1128    use opcda_bridge_proto::bridge::{
1129        BrowseNode as ProtoBrowseNode, BrowseNodeKind as ProtoNodeKind,
1130        BrowsePage as ProtoBrowsePage, BrowseRequest, BrowseSource as ProtoBrowseSource,
1131        CloseBrowseSessionRequest, ControlSearchIndexRequest, GetCapabilitiesRequest,
1132        GetCapabilitiesResponse, GetGatewayInfoRequest, GetGatewayInfoResponse,
1133        GetSearchIndexStatusRequest, IndexedSearchMatch as ProtoIndexedSearchMatch,
1134        ListServersRequest, ListServersResponse, NamespaceOrganization as ProtoOrganization,
1135        ReadRequest, ReadResponse, RefreshSearchIndexRequest, SearchEvent as ProtoSearchEvent,
1136        SearchIndexResponse as ProtoSearchIndexResponse, SearchIndexState as ProtoSearchIndexState,
1137        SearchIndexStatus as ProtoSearchIndexStatus, SearchProgress as ProtoSearchProgress,
1138        TagValue as ProtoTagValue, WriteRequest, WriteResponse,
1139    };
1140    use std::net::SocketAddr;
1141    use tokio::sync::mpsc;
1142    use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream};
1143    use tonic::transport::Server;
1144    use tonic::{Request, Response, Status};
1145
1146    #[derive(Default)]
1147    struct MockBridgeService {
1148        capabilities_response: GetCapabilitiesResponse,
1149        browse_response: ProtoBrowsePage,
1150        read_response: ReadResponse,
1151        write_response: WriteResponse,
1152        write_error: Option<Status>,
1153        list_servers_response: ListServersResponse,
1154        search_events: Vec<ProtoSearchEvent>,
1155        search_index_status_response: ProtoSearchIndexStatus,
1156        search_index_response: ProtoSearchIndexResponse,
1157        close_error: Option<Status>,
1158        gateway_info_response: GetGatewayInfoResponse,
1159    }
1160
1161    #[tonic::async_trait]
1162    impl Bridge for MockBridgeService {
1163        async fn get_gateway_info(
1164            &self,
1165            _request: Request<GetGatewayInfoRequest>,
1166        ) -> Result<Response<GetGatewayInfoResponse>, Status> {
1167            Ok(Response::new(self.gateway_info_response.clone()))
1168        }
1169
1170        async fn get_capabilities(
1171            &self,
1172            _request: Request<GetCapabilitiesRequest>,
1173        ) -> Result<Response<GetCapabilitiesResponse>, Status> {
1174            Ok(Response::new(self.capabilities_response.clone()))
1175        }
1176
1177        async fn list_servers(
1178            &self,
1179            _request: Request<ListServersRequest>,
1180        ) -> Result<Response<ListServersResponse>, Status> {
1181            Ok(Response::new(self.list_servers_response.clone()))
1182        }
1183
1184        async fn browse(
1185            &self,
1186            _request: Request<BrowseRequest>,
1187        ) -> Result<Response<ProtoBrowsePage>, Status> {
1188            Ok(Response::new(self.browse_response.clone()))
1189        }
1190
1191        async fn close_browse_session(
1192            &self,
1193            _request: Request<CloseBrowseSessionRequest>,
1194        ) -> Result<Response<()>, Status> {
1195            if let Some(status) = self.close_error.clone() {
1196                return Err(status);
1197            }
1198            Ok(Response::new(()))
1199        }
1200
1201        async fn get_search_index_status(
1202            &self,
1203            _request: Request<GetSearchIndexStatusRequest>,
1204        ) -> Result<Response<ProtoSearchIndexStatus>, Status> {
1205            Ok(Response::new(self.search_index_status_response.clone()))
1206        }
1207
1208        async fn refresh_search_index(
1209            &self,
1210            _request: Request<RefreshSearchIndexRequest>,
1211        ) -> Result<Response<ProtoSearchIndexStatus>, Status> {
1212            Ok(Response::new(self.search_index_status_response.clone()))
1213        }
1214
1215        async fn control_search_index(
1216            &self,
1217            _request: Request<ControlSearchIndexRequest>,
1218        ) -> Result<Response<ProtoSearchIndexStatus>, Status> {
1219            Ok(Response::new(self.search_index_status_response.clone()))
1220        }
1221
1222        async fn search_index(
1223            &self,
1224            _request: Request<opcda_bridge_proto::bridge::SearchIndexRequest>,
1225        ) -> Result<Response<ProtoSearchIndexResponse>, Status> {
1226            Ok(Response::new(self.search_index_response.clone()))
1227        }
1228
1229        type SearchStream = ReceiverStream<Result<ProtoSearchEvent, Status>>;
1230
1231        async fn search(
1232            &self,
1233            _request: Request<opcda_bridge_proto::bridge::SearchRequest>,
1234        ) -> Result<Response<Self::SearchStream>, Status> {
1235            let (tx, rx) = mpsc::channel(4);
1236            let events = self.search_events.clone();
1237            tokio::spawn(async move {
1238                for event in events {
1239                    let _ = tx.send(Ok(event)).await;
1240                }
1241            });
1242            Ok(Response::new(ReceiverStream::new(rx)))
1243        }
1244
1245        async fn read(
1246            &self,
1247            _request: Request<ReadRequest>,
1248        ) -> Result<Response<ReadResponse>, Status> {
1249            Ok(Response::new(self.read_response.clone()))
1250        }
1251
1252        async fn write(
1253            &self,
1254            _request: Request<WriteRequest>,
1255        ) -> Result<Response<WriteResponse>, Status> {
1256            if let Some(status) = self.write_error.clone() {
1257                return Err(status);
1258            }
1259            Ok(Response::new(self.write_response.clone()))
1260        }
1261    }
1262
1263    async fn start_mock_server(service: MockBridgeService) -> String {
1264        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
1265        let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
1266        let port = listener.local_addr().unwrap().port();
1267        tokio::spawn(
1268            Server::builder()
1269                .add_service(BridgeServer::new(service))
1270                .serve_with_incoming(TcpListenerStream::new(listener)),
1271        );
1272        format!("127.0.0.1:{port}")
1273    }
1274
1275    fn browse_page() -> ProtoBrowsePage {
1276        ProtoBrowsePage {
1277            session_id: "session".into(),
1278            nodes: vec![
1279                ProtoBrowseNode {
1280                    node_key: "area".into(),
1281                    display_name: "Area1".into(),
1282                    kind: ProtoNodeKind::Branch as i32,
1283                    item_id: None,
1284                },
1285                ProtoBrowseNode {
1286                    node_key: "pv".into(),
1287                    display_name: "PV".into(),
1288                    kind: ProtoNodeKind::BranchAndItem as i32,
1289                    item_id: Some("FCS0201!204FI00510.PV".into()),
1290                },
1291            ],
1292            complete: true,
1293            organization: ProtoOrganization::Hierarchical as i32,
1294            source: ProtoBrowseSource::Da2 as i32,
1295            ..Default::default()
1296        }
1297    }
1298
1299    #[tokio::test]
1300    async fn read_round_trips_through_a_real_gateway_connection() {
1301        let host = start_mock_server(MockBridgeService {
1302            read_response: ReadResponse {
1303                values: vec![ProtoTagValue {
1304                    tag_id: "Area1.LIC101.PV".to_string(),
1305                    value: "42.5".to_string(),
1306                    quality: "Good".to_string(),
1307                    timestamp: "2024-01-15 10:23:45".to_string(),
1308                }],
1309            },
1310            ..Default::default()
1311        })
1312        .await;
1313        let driver = OpcDaDriver::connect(&host, "S1").await.unwrap();
1314        let values = driver.read(&["Area1.LIC101.PV".to_string()]).await.unwrap();
1315        assert_eq!(values[0].quality, Quality::Good);
1316        assert_eq!(values[0].value, "42.5");
1317    }
1318
1319    #[tokio::test]
1320    async fn write_round_trips_a_rejected_write() {
1321        let host = start_mock_server(MockBridgeService {
1322            write_response: WriteResponse {
1323                tag_id: "Area1.LIC101.MV".to_string(),
1324                success: false,
1325                error: Some("tag is read-only".to_string()),
1326            },
1327            ..Default::default()
1328        })
1329        .await;
1330        let driver = OpcDaDriver::connect(&host, "S1").await.unwrap();
1331        let outcome = driver
1332            .write(&"Area1.LIC101.MV".to_string(), TagWrite::Float(55.0))
1333            .await
1334            .unwrap();
1335        assert_eq!(outcome.error_message.as_deref(), Some("tag is read-only"));
1336    }
1337
1338    #[tokio::test]
1339    async fn capabilities_and_browse_preserve_typed_namespace_metadata() {
1340        let host = start_mock_server(MockBridgeService {
1341            capabilities_response: GetCapabilitiesResponse {
1342                application_version: "0.4.0".into(),
1343                protocol_version: "2".into(),
1344                max_page_size: 1000,
1345                supports_browse_sessions: true,
1346                supports_search: true,
1347                organization: ProtoOrganization::Hierarchical as i32,
1348                source: ProtoBrowseSource::Da2 as i32,
1349                supports_indexed_search: true,
1350                indexed_search_protocol_version: "1".into(),
1351                max_indexed_search_results: 50,
1352                search_index_state: ProtoSearchIndexState::Ready as i32,
1353                search_index_promoting: false,
1354            },
1355            browse_response: browse_page(),
1356            ..Default::default()
1357        })
1358        .await;
1359        let driver = OpcDaDriver::connect(&host, "S1").await.unwrap();
1360        let capabilities = driver.capabilities().await.unwrap();
1361        assert_eq!(capabilities.application_version, "0.4.0");
1362        assert!(capabilities.supports_browse_sessions);
1363        assert!(capabilities.supports_indexed_search);
1364        assert_eq!(capabilities.indexed_search_protocol_version, "1");
1365        let page = driver.browse(BrowsePageRequest::root(200)).await.unwrap();
1366        assert_eq!(page.session_id, "session");
1367        assert!(page.nodes[0].kind.is_branch());
1368        assert!(page.nodes[1].kind.is_item());
1369        assert_eq!(
1370            page.nodes[1].item_id.as_deref(),
1371            Some("FCS0201!204FI00510.PV")
1372        );
1373        driver.close_browse_session("session").await.unwrap();
1374    }
1375
1376    #[tokio::test]
1377    async fn indexed_search_round_trips_exact_item_id_and_status() {
1378        let host = start_mock_server(MockBridgeService {
1379            search_index_status_response: ProtoSearchIndexStatus {
1380                server: "S1".into(),
1381                state: ProtoSearchIndexState::Ready as i32,
1382                configured: true,
1383                active_generation: 7,
1384                entry_count: 2,
1385                unique_item_count: 2,
1386                organization: ProtoOrganization::Hierarchical as i32,
1387                source: ProtoBrowseSource::Da2 as i32,
1388                ..Default::default()
1389            },
1390            search_index_response: ProtoSearchIndexResponse {
1391                matches: vec![ProtoIndexedSearchMatch {
1392                    item_id: "FCS0201!204FI00510.PV".into(),
1393                    display_name: "PV".into(),
1394                    kind: ProtoNodeKind::Item as i32,
1395                    breadcrumbs: vec!["FCS0201".into(), "204FI00510".into()],
1396                }],
1397                has_more: false,
1398                status: Some(ProtoSearchIndexStatus {
1399                    server: "S1".into(),
1400                    state: ProtoSearchIndexState::Ready as i32,
1401                    configured: true,
1402                    active_generation: 7,
1403                    entry_count: 2,
1404                    unique_item_count: 2,
1405                    organization: ProtoOrganization::Hierarchical as i32,
1406                    source: ProtoBrowseSource::Da2 as i32,
1407                    ..Default::default()
1408                }),
1409            },
1410            ..Default::default()
1411        })
1412        .await;
1413        let driver = OpcDaDriver::connect(&host, "S1").await.unwrap();
1414        let status = driver.search_index_status().await.unwrap();
1415        assert_eq!(status.state, SearchIndexState::Ready);
1416        assert_eq!(status.active_generation, 7);
1417        driver.set_search_index_auto_refresh(false).await.unwrap();
1418        driver.set_search_index_auto_refresh(true).await.unwrap();
1419        driver.delete_search_index().await.unwrap();
1420        let response = driver
1421            .search_index_query(SearchIndexRequest::new(
1422                "FCS0201!204FI00510",
1423                SearchMatchMode::Prefix,
1424                50,
1425            ))
1426            .await
1427            .unwrap();
1428        assert_eq!(response.matches[0].item_id, "FCS0201!204FI00510.PV");
1429        assert!(!response.has_more);
1430    }
1431
1432    #[tokio::test]
1433    async fn search_stream_preserves_progress_and_completion() {
1434        let host = start_mock_server(MockBridgeService {
1435            search_events: vec![
1436                ProtoSearchEvent {
1437                    event: Some(search_event::Event::Progress(ProtoSearchProgress {
1438                        visited_nodes: 4,
1439                        matches: 0,
1440                        partial: true,
1441                    })),
1442                },
1443                ProtoSearchEvent {
1444                    event: Some(search_event::Event::Completed(
1445                        opcda_bridge_proto::bridge::SearchCompleted {
1446                            complete: true,
1447                            cancelled: false,
1448                            truncated: false,
1449                            warning: None,
1450                        },
1451                    )),
1452                },
1453            ],
1454            ..Default::default()
1455        })
1456        .await;
1457        let driver = OpcDaDriver::connect(&host, "S1").await.unwrap();
1458        let events = driver
1459            .search_events(SearchRequest::new("PV", SearchMatchMode::Contains, 20))
1460            .await
1461            .unwrap();
1462        assert!(matches!(events[0], SearchEvent::Progress(_)));
1463        assert!(matches!(events[1], SearchEvent::Completed(_)));
1464    }
1465
1466    #[tokio::test]
1467    async fn driver_trait_delegates_all_opcda_operations() {
1468        let host = start_mock_server(MockBridgeService {
1469            capabilities_response: GetCapabilitiesResponse {
1470                protocol_version: "2".into(),
1471                ..Default::default()
1472            },
1473            browse_response: ProtoBrowsePage {
1474                complete: true,
1475                ..Default::default()
1476            },
1477            search_index_response: ProtoSearchIndexResponse {
1478                status: Some(ProtoSearchIndexStatus::default()),
1479                ..Default::default()
1480            },
1481            search_events: vec![ProtoSearchEvent {
1482                event: Some(search_event::Event::Completed(
1483                    opcda_bridge_proto::bridge::SearchCompleted {
1484                        complete: true,
1485                        ..Default::default()
1486                    },
1487                )),
1488            }],
1489            ..Default::default()
1490        })
1491        .await;
1492        let driver = OpcDaDriver::connect(&host, "S1").await.unwrap();
1493        assert_eq!(
1494            <OpcDaDriver as Driver>::capabilities(&driver)
1495                .await
1496                .unwrap()
1497                .protocol_version,
1498            "2"
1499        );
1500        <OpcDaDriver as Driver>::browse(&driver, BrowsePageRequest::root(1))
1501            .await
1502            .unwrap();
1503        <OpcDaDriver as Driver>::close_browse_session(&driver, "session")
1504            .await
1505            .unwrap();
1506        let events = <OpcDaDriver as Driver>::search(
1507            &driver,
1508            SearchRequest::new("PV", SearchMatchMode::Contains, 10),
1509        )
1510        .await
1511        .unwrap();
1512        assert_eq!(events.len(), 1);
1513        assert!(
1514            <OpcDaDriver as Driver>::search_index_status(&driver)
1515                .await
1516                .is_ok()
1517        );
1518        assert!(
1519            <OpcDaDriver as Driver>::refresh_search_index(&driver, false)
1520                .await
1521                .is_ok()
1522        );
1523        assert!(
1524            <OpcDaDriver as Driver>::set_search_index_auto_refresh(&driver, false)
1525                .await
1526                .is_ok()
1527        );
1528        assert!(
1529            <OpcDaDriver as Driver>::delete_search_index(&driver)
1530                .await
1531                .is_ok()
1532        );
1533        assert!(
1534            <OpcDaDriver as Driver>::control_search_index(&driver, SearchIndexControlAction::Pause)
1535                .await
1536                .is_ok()
1537        );
1538        assert!(
1539            <OpcDaDriver as Driver>::search_index(
1540                &driver,
1541                SearchIndexRequest::new("PV", SearchMatchMode::Exact, 1)
1542            )
1543            .await
1544            .is_ok()
1545        );
1546    }
1547
1548    #[tokio::test]
1549    async fn mock_gateway_info_returns_the_configured_response() {
1550        let service = MockBridgeService {
1551            gateway_info_response: GetGatewayInfoResponse {
1552                application_version: "test-gateway".into(),
1553                ..Default::default()
1554            },
1555            ..Default::default()
1556        };
1557        let response = service
1558            .get_gateway_info(Request::new(GetGatewayInfoRequest::default()))
1559            .await
1560            .unwrap();
1561        assert_eq!(response.into_inner().application_version, "test-gateway");
1562    }
1563
1564    #[tokio::test]
1565    async fn driver_trait_maps_close_browse_rpc_errors() {
1566        let host = start_mock_server(MockBridgeService {
1567            close_error: Some(Status::internal("close failed")),
1568            ..Default::default()
1569        })
1570        .await;
1571        let driver = OpcDaDriver::connect(&host, "S1").await.unwrap();
1572        assert!(matches!(
1573            <OpcDaDriver as Driver>::close_browse_session(&driver, "session").await,
1574            Err(DriverError::Operation(_))
1575        ));
1576    }
1577
1578    #[tokio::test]
1579    async fn list_opcda_servers_returns_the_gateways_registered_servers() {
1580        let host = start_mock_server(MockBridgeService {
1581            list_servers_response: ListServersResponse {
1582                servers: vec!["Matrikon.OPC.Simulation.1".into()],
1583            },
1584            ..Default::default()
1585        })
1586        .await;
1587        assert_eq!(
1588            list_opcda_servers(&host).await.unwrap(),
1589            vec!["Matrikon.OPC.Simulation.1".to_string()]
1590        );
1591    }
1592
1593    proptest::proptest! {
1594        #[test]
1595        fn arbitrary_protocol_payloads_preserve_safe_fields(
1596            tag in proptest::prelude::any::<String>(),
1597            value in proptest::prelude::any::<String>(),
1598            quality in proptest::prelude::any::<String>(),
1599            timestamp in proptest::prelude::any::<String>(),
1600            write_error in proptest::prelude::prop::option::of(proptest::prelude::any::<String>()),
1601            success in proptest::prelude::any::<bool>(),
1602        ) {
1603            let mapped = tag_value_from_raw(opcda_bridge::TagValue {
1604                tag_id: tag.clone(),
1605                value: value.clone(),
1606                quality,
1607                timestamp,
1608            });
1609            proptest::prop_assert_eq!(mapped.tag, tag.clone());
1610            proptest::prop_assert_eq!(mapped.value, value.clone());
1611            proptest::prop_assert_eq!(mapped.timestamp, None);
1612
1613            let node = browse_node_from_bridge(opcda_bridge::BrowseNode {
1614                node_key: tag.clone(),
1615                display_name: value.clone(),
1616                kind: opcda_bridge::BrowseNodeKind::Item,
1617                item_id: Some(tag.clone()),
1618            });
1619            proptest::prop_assert_eq!(node.node_key, tag);
1620            proptest::prop_assert!(node.kind.is_item());
1621
1622            let outcome = write_outcome_from_result(opcda_bridge::WriteResult {
1623                tag_id: String::new(),
1624                success,
1625                error: write_error.clone(),
1626            });
1627            proptest::prop_assert_eq!(outcome.success, success);
1628            if success {
1629                proptest::prop_assert_eq!(outcome.error_message, None);
1630            } else {
1631                proptest::prop_assert_eq!(
1632                    outcome.error_message,
1633                    Some(write_error.unwrap_or_else(|| "gateway rejected the write".to_string()))
1634                );
1635            }
1636        }
1637
1638        #[test]
1639        fn arbitrary_numeric_writes_keep_the_f32_value(value in -1_000_000.0f32..1_000_000.0f32) {
1640            proptest::prop_assert_eq!(
1641                opc_value_from_write(TagWrite::Float(value)),
1642                opcda_bridge::Value::Float(f64::from(value))
1643            );
1644        }
1645    }
1646}
bhtune_driver/opcda.rs

bhtune_driver/
opcda.rs
