krabka_sdk_conformance/
mock_adapter.rs

1//! In-process mock adapter used to validate the harness and vectors.
2
3use std::collections::HashMap;
4
5use base64::{Engine as _, engine::general_purpose::STANDARD};
6use serde_json::json;
7use tokio::io::{AsyncBufRead, AsyncBufReadExt as _, AsyncWrite, AsyncWriteExt as _};
8
9use crate::protocol::{
10    AdapterError, CONTRACT_MAJOR, CONTRACT_MINOR_QUEUE_RPC, CONTRACT_MINOR_V1_0, Command,
11    ErrorKind, Filter, FilterOp, Header, Message, QueueAckEntry, QueueAckType, QueueRenewEntry,
12    Response,
13};
14
15#[derive(Debug, Clone)]
16struct StoredMessage {
17    topic: String,
18    partition: i32,
19    offset: i64,
20    value_b64: String,
21    headers: Vec<Header>,
22    queue_states: HashMap<String, QueueDeliveryState>,
23}
24
25#[derive(Debug, Clone)]
26struct QueueSession {
27    topic: String,
28    group: String,
29    max_messages: u32,
30}
31
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33enum QueueMessageState {
34    Available,
35    Acquired,
36    Accepted,
37    Rejected,
38}
39
40#[derive(Debug, Clone)]
41struct QueueDeliveryState {
42    state: QueueMessageState,
43    session_id: Option<String>,
44    delivery_count: u32,
45}
46
47#[derive(Debug, Clone)]
48struct Subscription {
49    topics: Vec<String>,
50    filter: Option<Filter>,
51    next_index: usize,
52}
53
54/// Fault injection mode for harness negative tests.
55#[derive(Debug, Clone, Copy, PartialEq, Eq)]
56pub enum FaultMode {
57    /// Normal mock behavior.
58    None,
59    /// Return an incorrect response for publish commands.
60    WrongPublish,
61    /// Return an incorrect response for configure commands.
62    WrongConfigure,
63}
64
65/// Mock adapter with deterministic in-memory publish/subscribe behavior.
66#[derive(Debug, Clone)]
67pub struct MockAdapter {
68    endpoint: String,
69    bearer: Option<String>,
70    messages: Vec<StoredMessage>,
71    subscription: Option<Subscription>,
72    fault_mode: FaultMode,
73    contract_minor: u16,
74    next_session_id: u64,
75    queue_sessions: HashMap<String, QueueSession>,
76}
77
78impl Default for MockAdapter {
79    fn default() -> Self {
80        Self {
81            endpoint: "mock://gateway".into(),
82            bearer: None,
83            messages: vec![],
84            subscription: None,
85            fault_mode: FaultMode::None,
86            contract_minor: CONTRACT_MINOR_V1_0,
87            next_session_id: 1,
88            queue_sessions: HashMap::new(),
89        }
90    }
91}
92
93impl MockAdapter {
94    /// Create a mock adapter.
95    #[must_use]
96    pub fn new() -> Self {
97        Self::default()
98    }
99
100    /// Create a mock adapter with a fault mode.
101    #[must_use]
102    pub fn with_fault_mode(fault_mode: FaultMode) -> Self {
103        Self {
104            fault_mode,
105            ..Self::default()
106        }
107    }
108
109    /// Create a mock adapter with an explicit contract minor.
110    #[must_use]
111    pub fn with_contract_minor(contract_minor: u16, fault_mode: FaultMode) -> Self {
112        Self {
113            fault_mode,
114            contract_minor,
115            ..Self::default()
116        }
117    }
118
119    /// Handle one protocol command.
120    #[must_use]
121    pub fn handle(&mut self, command: Command) -> Response {
122        match command {
123            Command::Hello => Response::Hello {
124                contract_major: CONTRACT_MAJOR,
125                contract_minor: self.contract_minor,
126                language: "mock".into(),
127            },
128            Command::Configure { endpoint, bearer } => {
129                self.endpoint = endpoint;
130                self.bearer = bearer;
131                let bearer_configured = self.bearer.is_some();
132                if self.fault_mode == FaultMode::WrongConfigure {
133                    Response::Ok(json!({ "bearer_configured": !bearer_configured }))
134                } else {
135                    Response::Ok(json!({ "bearer_configured": bearer_configured }))
136                }
137            }
138            Command::Publish {
139                topic,
140                value_b64,
141                headers,
142            } => self.publish(topic, value_b64, headers),
143            Command::PublishEvent { topic, event } => {
144                if event.id.is_empty() {
145                    return Response::Error(AdapterError::with_message(
146                        ErrorKind::InvalidArgument,
147                        "CloudEvent id is required",
148                    ));
149                }
150                let mut headers = vec![
151                    header("ce_id", event.id),
152                    header("ce_source", event.source),
153                    header("ce_type", event.type_),
154                    header("ce_specversion", event.specversion),
155                ];
156                if let Some(content_type) = event.datacontenttype {
157                    headers.push(header("content-type", content_type));
158                }
159                self.publish(topic, event.data_b64, headers)
160            }
161            Command::Subscribe {
162                topics,
163                group: _,
164                filter,
165            } => {
166                if let Some(error) = filter.as_ref().and_then(validate_filter) {
167                    return Response::Error(error);
168                }
169                self.subscription = Some(Subscription {
170                    topics,
171                    filter,
172                    next_index: 0,
173                });
174                Response::Ok(json!({}))
175            }
176            Command::NextMessage { timeout_ms: _ } => self.next_message(),
177            Command::QueueAcquire {
178                topic,
179                group,
180                max,
181                lock_duration_ms,
182                session_id,
183            } => self.queue_acquire(&topic, &group, max, lock_duration_ms, &session_id),
184            Command::QueueAck { .. } => Self::unimplemented_queues(),
185            Command::QueueAcknowledge {
186                session_id,
187                entries,
188            } => self.queue_acknowledge(&session_id, &entries),
189            Command::QueueRenew {
190                session_id,
191                entries,
192            } => self.queue_renew(&session_id, &entries),
193            Command::DbConnect { .. } => Response::Error(AdapterError::unimplemented(
194                "database",
195                "chapter-f-control-plane",
196            )),
197            Command::AuthSignIn { .. } => Response::Error(AdapterError::with_message(
198                ErrorKind::Unauthenticated,
199                "identity APIs are not part of contract v1",
200            )),
201            Command::BlobPut { .. } | Command::BlobGet { .. } => {
202                Response::Error(AdapterError::unimplemented("blob", "chapter-b-blob-api"))
203            }
204        }
205    }
206
207    fn publish(&mut self, topic: String, value_b64: String, headers: Vec<Header>) -> Response {
208        if self.fault_mode == FaultMode::WrongPublish {
209            return Response::Ok(json!({ "partition": 9, "offset": 999, "deduplicated": true }));
210        }
211        if self.endpoint.starts_with("unreachable://") {
212            return Response::Error(AdapterError::with_message(
213                ErrorKind::Transport,
214                "endpoint unreachable",
215            ));
216        }
217        if topic.is_empty() {
218            return Response::Error(AdapterError::with_message(
219                ErrorKind::InvalidArgument,
220                "topic is required",
221            ));
222        }
223        if topic == "__missing_topic" {
224            return Response::Error(AdapterError::with_message(
225                ErrorKind::NotFound,
226                "topic not found",
227            ));
228        }
229        let offset = self
230            .messages
231            .iter()
232            .filter(|message| message.topic == topic)
233            .count()
234            .try_into()
235            .expect("message count fits i64");
236        self.messages.push(StoredMessage {
237            topic,
238            partition: 0,
239            offset,
240            value_b64,
241            headers,
242            queue_states: HashMap::new(),
243        });
244        Response::Ok(json!({ "partition": 0, "offset": offset, "deduplicated": false }))
245    }
246
247    fn next_message(&mut self) -> Response {
248        let Some(subscription) = self.subscription.as_mut() else {
249            return Response::Error(AdapterError::with_message(
250                ErrorKind::InvalidArgument,
251                "subscribe before next_message",
252            ));
253        };
254        while let Some(message) = self.messages.get(subscription.next_index) {
255            subscription.next_index += 1;
256            if !subscription
257                .topics
258                .iter()
259                .any(|topic| topic == &message.topic)
260            {
261                continue;
262            }
263            if !filter_matches(subscription.filter.as_ref(), message) {
264                continue;
265            }
266            return Response::Message(Message {
267                topic: message.topic.clone(),
268                partition: message.partition,
269                offset: message.offset,
270                value_b64: message.value_b64.clone(),
271                headers: message.headers.clone(),
272            });
273        }
274        Response::Error(AdapterError::with_message(
275            ErrorKind::NotFound,
276            "no message available",
277        ))
278    }
279
280    fn queue_acquire(
281        &mut self,
282        topic: &str,
283        group: &str,
284        max: u32,
285        lock_duration_ms: u64,
286        requested_session_id: &str,
287    ) -> Response {
288        if self.contract_minor < CONTRACT_MINOR_QUEUE_RPC {
289            return Self::unimplemented_queues();
290        }
291        if group.is_empty() {
292            return Response::Error(AdapterError::with_message(
293                ErrorKind::InvalidArgument,
294                "queue group is required",
295            ));
296        }
297        if topic.is_empty() {
298            return Response::Error(AdapterError::with_message(
299                ErrorKind::InvalidArgument,
300                "queue topic is required",
301            ));
302        }
303        if lock_duration_ms != 30_000 {
304            return Response::Error(AdapterError::with_message(
305                ErrorKind::InvalidArgument,
306                "queue lock_duration_ms must be 30000; per-acquire lock durations are not supported",
307            ));
308        }
309
310        let effective_max = max.clamp(1, 500);
311        let session_id = if requested_session_id.is_empty() {
312            let session_id = self.next_queue_session_id();
313            self.queue_sessions.insert(
314                session_id.clone(),
315                QueueSession {
316                    topic: topic.to_string(),
317                    group: group.to_string(),
318                    max_messages: effective_max,
319                },
320            );
321            session_id
322        } else {
323            let Some(session) = self.queue_sessions.get(requested_session_id) else {
324                return Response::Error(AdapterError::with_message(
325                    ErrorKind::InvalidArgument,
326                    "queue session expired; re-acquire",
327                ));
328            };
329            if session.topic != topic || session.group != group {
330                return Response::Error(AdapterError::with_message(
331                    ErrorKind::InvalidArgument,
332                    "group_id and topics are fixed when a queue session is created",
333                ));
334            }
335            if max != 0 && effective_max != session.max_messages {
336                return Response::Error(AdapterError::with_message(
337                    ErrorKind::InvalidArgument,
338                    "max_messages is fixed when a queue session is created",
339                ));
340            }
341            requested_session_id.to_string()
342        };
343        let max = usize::try_from(effective_max).expect("u32 fits usize");
344        let messages =
345            self.messages
346                .iter_mut()
347                .filter(|message| {
348                    message.topic == topic
349                        && message
350                            .queue_states
351                            .get(group)
352                            .is_none_or(|state| state.state == QueueMessageState::Available)
353                })
354                .take(max)
355                .map(|message| {
356                    let message_id = queue_message_id(message);
357                    let state = message.queue_states.entry(group.to_string()).or_insert(
358                        QueueDeliveryState {
359                            state: QueueMessageState::Available,
360                            session_id: None,
361                            delivery_count: 0,
362                        },
363                    );
364                    state.state = QueueMessageState::Acquired;
365                    state.session_id = Some(session_id.clone());
366                    state.delivery_count = state.delivery_count.saturating_add(1);
367                    json!({
368                        "message_id": message_id,
369                        "topic": message.topic,
370                        "partition": message.partition,
371                        "offset": message.offset,
372                        "value_b64": message.value_b64,
373                        "headers": message.headers,
374                        "delivery_count": state.delivery_count,
375                    })
376                })
377                .collect::<Vec<_>>();
378
379        Response::Ok(json!({ "session_id": session_id, "messages": messages }))
380    }
381
382    fn queue_acknowledge(&mut self, session_id: &str, entries: &[QueueAckEntry]) -> Response {
383        if self.contract_minor < CONTRACT_MINOR_QUEUE_RPC {
384            return Self::unimplemented_queues();
385        }
386        if session_id.is_empty() {
387            return Response::Error(AdapterError::with_message(
388                ErrorKind::InvalidArgument,
389                "queue session_id is required",
390            ));
391        }
392        if !self.queue_sessions.contains_key(session_id) {
393            return Response::Error(AdapterError::with_message(
394                ErrorKind::InvalidArgument,
395                "queue session expired; re-acquire",
396            ));
397        }
398
399        let results = entries
400            .iter()
401            .map(|entry| self.queue_ack_result(session_id, entry))
402            .collect::<Vec<_>>();
403        Response::Ok(json!({ "results": results }))
404    }
405
406    fn queue_renew(&mut self, session_id: &str, entries: &[QueueRenewEntry]) -> Response {
407        if self.contract_minor < CONTRACT_MINOR_QUEUE_RPC {
408            return Self::unimplemented_queues();
409        }
410        if session_id.is_empty() {
411            return Response::Error(AdapterError::with_message(
412                ErrorKind::InvalidArgument,
413                "queue session_id is required",
414            ));
415        }
416        if !self.queue_sessions.contains_key(session_id) {
417            return Response::Error(AdapterError::with_message(
418                ErrorKind::InvalidArgument,
419                "queue session expired; re-acquire",
420            ));
421        }
422
423        let results = entries
424            .iter()
425            .map(|entry| self.queue_renew_result(session_id, entry))
426            .collect::<Vec<_>>();
427        Response::Ok(json!({ "results": results }))
428    }
429
430    fn next_queue_session_id(&mut self) -> String {
431        let session_id = format!("queue-session-{}", self.next_session_id);
432        self.next_session_id = self.next_session_id.saturating_add(1);
433        session_id
434    }
435
436    fn queue_ack_result(&mut self, session_id: &str, entry: &QueueAckEntry) -> serde_json::Value {
437        let group = self
438            .queue_sessions
439            .get(session_id)
440            .expect("queue session was validated")
441            .group
442            .clone();
443        let Some(state) = self.queue_state_mut(&group, session_id, &entry.message_id) else {
444            return json!({
445                "message_id": entry.message_id,
446                "error": { "kind": "invalid_argument", "message": "queue message is not acquired" },
447            });
448        };
449        state.state = match entry.ack_type {
450            QueueAckType::Accept => QueueMessageState::Accepted,
451            QueueAckType::Release => QueueMessageState::Available,
452            QueueAckType::Reject => QueueMessageState::Rejected,
453        };
454        state.session_id = None;
455        json!({ "message_id": entry.message_id, "error": null })
456    }
457
458    fn queue_renew_result(
459        &mut self,
460        session_id: &str,
461        entry: &QueueRenewEntry,
462    ) -> serde_json::Value {
463        let group = self
464            .queue_sessions
465            .get(session_id)
466            .expect("queue session was validated")
467            .group
468            .clone();
469        if self
470            .queue_state_mut(&group, session_id, &entry.message_id)
471            .is_some()
472        {
473            return json!({ "message_id": entry.message_id, "error": null });
474        }
475
476        json!({
477            "message_id": entry.message_id,
478            "error": { "kind": "invalid_argument", "message": "queue message is not acquired" },
479        })
480    }
481
482    fn queue_state_mut(
483        &mut self,
484        group: &str,
485        session_id: &str,
486        message_id: &str,
487    ) -> Option<&mut QueueDeliveryState> {
488        self.messages
489            .iter_mut()
490            .find(|message| queue_message_id(message) == message_id)
491            .and_then(|message| message.queue_states.get_mut(group))
492            .filter(|state| {
493                state.state == QueueMessageState::Acquired
494                    && state.session_id.as_deref() == Some(session_id)
495            })
496    }
497
498    fn unimplemented_queues() -> Response {
499        Response::Error(AdapterError::unimplemented(
500            "queues",
501            "gateway-sharegroup-rpc",
502        ))
503    }
504}
505
506fn validate_filter(filter: &Filter) -> Option<AdapterError> {
507    if filter.op != FilterOp::Equals {
508        return Some(AdapterError::with_message(
509            ErrorKind::InvalidArgument,
510            "only equals filters are supported",
511        ));
512    }
513    let Some(field) = filter.path.strip_prefix("$.") else {
514        return Some(AdapterError::with_message(
515            ErrorKind::InvalidArgument,
516            "filter path must start with $.",
517        ));
518    };
519    if field.is_empty()
520        || !field.split('.').all(|segment| {
521            let mut characters = segment.chars();
522            characters.next().is_some_and(|first| {
523                (first == '_' || first.is_ascii_alphabetic())
524                    && characters
525                        .all(|character| character == '_' || character.is_ascii_alphanumeric())
526            })
527        })
528    {
529        return Some(AdapterError::with_message(
530            ErrorKind::InvalidArgument,
531            "filter path must contain identifier segments",
532        ));
533    }
534    if filter.value.is_array() || filter.value.is_object() {
535        return Some(AdapterError::with_message(
536            ErrorKind::InvalidArgument,
537            "filter value must be a string, number, boolean, or null",
538        ));
539    }
540    None
541}
542
543fn queue_message_id(message: &StoredMessage) -> String {
544    format!("{}:{}:{}", message.topic, message.partition, message.offset)
545}
546
547fn header(name: &str, value: String) -> Header {
548    Header {
549        name: name.into(),
550        value_b64: Some(STANDARD.encode(value)),
551    }
552}
553
554fn filter_matches(filter: Option<&Filter>, message: &StoredMessage) -> bool {
555    let Some(filter) = filter else {
556        return true;
557    };
558    if filter.op != FilterOp::Equals {
559        return false;
560    }
561    let Ok(value_bytes) = STANDARD.decode(&message.value_b64) else {
562        return false;
563    };
564    let Ok(json) = serde_json::from_slice::<serde_json::Value>(&value_bytes) else {
565        return false;
566    };
567    let Some(field) = filter.path.strip_prefix("$.") else {
568        return false;
569    };
570    json.get(field) == Some(&filter.value)
571}
572
573/// Run a mock adapter over JSON lines.
574///
575/// # Errors
576///
577/// Returns an error when JSON-lines input/output fails or a command cannot be
578/// decoded.
579pub async fn run_stdio<R, W>(
580    mut input: R,
581    mut output: W,
582    fault_mode: FaultMode,
583    contract_minor: u16,
584) -> Result<(), MockAdapterError>
585where
586    R: AsyncBufRead + Unpin,
587    W: AsyncWrite + Unpin,
588{
589    let mut adapter = MockAdapter::with_contract_minor(contract_minor, fault_mode);
590    let mut line = String::new();
591    loop {
592        line.clear();
593        if input.read_line(&mut line).await? == 0 {
594            break;
595        }
596        let command = serde_json::from_str::<Command>(line.trim_end())?;
597        let response = adapter.handle(command);
598        let encoded = serde_json::to_string(&response)?;
599        output.write_all(encoded.as_bytes()).await?;
600        output.write_all(b"\n").await?;
601        output.flush().await?;
602    }
603    Ok(())
604}
605
606/// Errors returned by the mock adapter runner.
607#[derive(Debug, thiserror::Error)]
608#[non_exhaustive]
609pub enum MockAdapterError {
610    /// I/O failed.
611    #[error("mock adapter io: {0}")]
612    Io(#[from] std::io::Error),
613    /// JSON failed to decode or encode.
614    #[error("mock adapter json: {0}")]
615    Json(#[from] serde_json::Error),
616}
617
618#[cfg(test)]
619mod tests {
620    use super::*;
621
622    #[test]
623    fn mock_adapter_answers_hello_and_publish() {
624        let mut adapter = MockAdapter::new();
625        let hello = adapter.handle(Command::Hello);
626        assert_eq!(
627            hello,
628            Response::Hello {
629                contract_major: CONTRACT_MAJOR,
630                contract_minor: CONTRACT_MINOR_V1_0,
631                language: "mock".into()
632            }
633        );
634
635        let published = adapter.handle(Command::Publish {
636            topic: "t".into(),
637            value_b64: "aGk=".into(),
638            headers: vec![],
639        });
640        assert2::assert!(let Response::Ok(value) = published);
641        assert_eq!(value["offset"], 0);
642    }
643}
krabka_sdk_conformance/mock_adapter.rs

krabka_sdk_conformance/
mock_adapter.rs
