1use std::collections::{BTreeMap, VecDeque};
5use std::fmt;
6
7use serde::{Deserialize, Serialize};
8
9use crate::device::{
10 ConfigError, ControlCommand, Device, DeviceEvent, DevicePlugin, Job, LogEntry, LogLevel,
11 PeripheralSignal,
12};
13use crate::fault::{FaultId, FaultOptions};
14use crate::ids::DeviceId;
15use crate::log::{EventKind, EventLog, EventRecord, DEFAULT_LOG_CAPACITY};
16use crate::nfc::NfcEvent;
17use crate::time::{Clock, Millis};
18
19pub const DEFAULT_JOB_CAPACITY: usize = 1_000;
21pub const MAX_PERIPHERAL_DEPTH: u8 = 8;
23pub const DEFAULT_BUS_CAPACITY: usize = 10_000;
25
26#[derive(Clone, Copy, Debug, PartialEq, Eq)]
28pub struct RegistryConfig {
29 pub log_capacity: usize,
31 pub job_capacity: usize,
33 pub bus_capacity: usize,
35}
36
37impl Default for RegistryConfig {
38 fn default() -> Self {
39 Self {
40 log_capacity: DEFAULT_LOG_CAPACITY,
41 job_capacity: DEFAULT_JOB_CAPACITY,
42 bus_capacity: DEFAULT_BUS_CAPACITY,
43 }
44 }
45}
46
47#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
52#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
53#[serde(tag = "type")]
54pub enum BusEvent {
55 #[serde(rename = "device:added", rename_all = "camelCase")]
57 Added {
58 device_id: DeviceId,
60 kind: String,
62 at: Millis,
64 },
65 #[serde(rename = "device:removed", rename_all = "camelCase")]
67 Removed {
68 device_id: DeviceId,
70 at: Millis,
72 },
73 #[serde(rename = "device:state", rename_all = "camelCase")]
75 State {
76 device_id: DeviceId,
78 state: serde_json::Value,
80 at: Millis,
82 },
83 #[serde(rename = "device:output", rename_all = "camelCase")]
85 Output {
86 device_id: DeviceId,
88 job: Job,
90 },
91}
92
93#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
95#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
96pub struct ActiveFault {
97 pub id: FaultId,
99 #[cfg_attr(
101 feature = "ts",
102 ts(as = "std::collections::HashMap<String, serde_json::Value>")
103 )]
104 pub options: FaultOptions,
105}
106
107#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
109#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
110#[serde(rename_all = "camelCase")]
111pub struct DeviceSummary {
112 pub id: DeviceId,
114 pub kind: String,
116 pub state: serde_json::Value,
118 pub faults: Vec<ActiveFault>,
120 pub supported_faults: Vec<FaultId>,
122 pub attached_to: Option<DeviceId>,
124}
125
126#[derive(Clone, Debug, PartialEq, Eq)]
128pub enum RegistryError {
129 DuplicateKind(String),
131 UnknownKind(String),
133 DuplicateId(DeviceId),
135 NotFound(DeviceId),
137 InvalidConfig(ConfigError),
139 UnsupportedFault {
141 device: DeviceId,
143 fault: FaultId,
145 },
146}
147
148impl RegistryError {
149 pub fn code(&self) -> &'static str {
156 match self {
157 RegistryError::DuplicateKind(_) => "duplicate_kind",
158 RegistryError::UnknownKind(_) => "unknown_kind",
159 RegistryError::DuplicateId(_) => "duplicate_id",
160 RegistryError::NotFound(_) => "not_found",
161 RegistryError::InvalidConfig(_) => "invalid_config",
162 RegistryError::UnsupportedFault { .. } => "unsupported_fault",
163 }
164 }
165}
166
167impl fmt::Display for RegistryError {
168 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
169 match self {
170 RegistryError::DuplicateKind(kind) => {
171 write!(f, "a plugin for kind `{kind}` is already registered")
172 }
173 RegistryError::UnknownKind(kind) => write!(f, "no plugin for device kind `{kind}`"),
174 RegistryError::DuplicateId(id) => write!(f, "device `{id}` already exists"),
175 RegistryError::NotFound(id) => write!(f, "device `{id}` not found"),
176 RegistryError::InvalidConfig(err) => write!(f, "invalid device config: {err}"),
177 RegistryError::UnsupportedFault { device, fault } => {
178 write!(f, "device `{device}` does not support fault `{fault}`")
179 }
180 }
181 }
182}
183
184impl std::error::Error for RegistryError {
185 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
186 match self {
187 RegistryError::InvalidConfig(err) => Some(err),
188 _ => None,
189 }
190 }
191}
192
193struct Slot {
194 kind: &'static str,
195 device: Box<dyn Device>,
196 faults: BTreeMap<FaultId, FaultOptions>,
197 log: EventLog,
198 jobs: VecDeque<Job>,
200}
201
202pub struct Registry {
215 clock: Box<dyn Clock>,
216 config: RegistryConfig,
217 plugins: BTreeMap<&'static str, Box<dyn DevicePlugin>>,
218 devices: BTreeMap<DeviceId, Slot>,
219 bus: VecDeque<BusEvent>,
220 last_now: Millis,
221 next_seq: u64,
222}
223
224impl Registry {
225 pub fn new(clock: impl Clock + 'static) -> Self {
227 Self::with_config(clock, RegistryConfig::default())
228 }
229
230 pub fn with_config(clock: impl Clock + 'static, config: RegistryConfig) -> Self {
232 Self {
233 clock: Box::new(clock),
234 config,
235 plugins: BTreeMap::new(),
236 devices: BTreeMap::new(),
237 bus: VecDeque::new(),
238 last_now: Millis::ZERO,
239 next_seq: 1,
240 }
241 }
242
243 pub fn config(&self) -> RegistryConfig {
245 self.config
246 }
247
248 pub fn now(&mut self) -> Millis {
250 let now = self.clock.now().max(self.last_now);
251 self.last_now = now;
252 now
253 }
254
255 pub fn register(&mut self, plugin: impl DevicePlugin) -> Result<(), RegistryError> {
259 self.register_boxed(Box::new(plugin))
260 }
261
262 pub fn register_boxed(&mut self, plugin: Box<dyn DevicePlugin>) -> Result<(), RegistryError> {
264 let kind = plugin.kind();
265 if self.plugins.contains_key(kind) {
266 return Err(RegistryError::DuplicateKind(kind.to_owned()));
267 }
268 self.plugins.insert(kind, plugin);
269 Ok(())
270 }
271
272 pub fn plugin(&self, kind: &str) -> Option<&dyn DevicePlugin> {
274 self.plugins.get(kind).map(|p| p.as_ref())
275 }
276
277 pub fn plugin_kinds(&self) -> impl Iterator<Item = &'static str> + '_ {
279 self.plugins.keys().copied()
280 }
281
282 pub fn create(
287 &mut self,
288 kind: &str,
289 id: Option<DeviceId>,
290 config: serde_json::Value,
291 ) -> Result<DeviceId, RegistryError> {
292 let plugin = self
293 .plugins
294 .get(kind)
295 .ok_or_else(|| RegistryError::UnknownKind(kind.to_owned()))?;
296 let kind = plugin.kind();
297 let id = match id {
298 Some(id) if self.devices.contains_key(&id) => {
299 return Err(RegistryError::DuplicateId(id))
300 }
301 Some(id) => id,
302 None => self.generate_id(kind),
303 };
304 let device = plugin
305 .create(id.clone(), config)
306 .map_err(RegistryError::InvalidConfig)?;
307 let slot = Slot {
308 kind,
309 device,
310 faults: BTreeMap::new(),
311 log: EventLog::new(self.config.log_capacity),
312 jobs: VecDeque::new(),
313 };
314 self.devices.insert(id.clone(), slot);
315 let at = self.now();
316 self.record(
317 &id,
318 at,
319 EventKind::Created {
320 kind: kind.to_owned(),
321 },
322 );
323 self.publish(BusEvent::Added {
324 device_id: id.clone(),
325 kind: kind.to_owned(),
326 at,
327 });
328 Ok(id)
329 }
330
331 fn generate_id(&self, kind: &str) -> DeviceId {
332 (1u64..)
333 .filter_map(|n| format!("{kind}-{n}").parse::<DeviceId>().ok())
334 .find(|id| !self.devices.contains_key(id))
335 .unwrap_or_else(|| self.fallback_id())
337 }
338
339 fn fallback_id(&self) -> DeviceId {
340 let mut n = self.devices.len() as u64 + 1;
341 loop {
342 if let Ok(id) = format!("device-{n}").parse::<DeviceId>() {
343 if !self.devices.contains_key(&id) {
344 return id;
345 }
346 }
347 n += 1;
348 }
349 }
350
351 pub fn remove(&mut self, id: &DeviceId) -> Result<(), RegistryError> {
353 self.devices
354 .remove(id)
355 .ok_or_else(|| RegistryError::NotFound(id.clone()))?;
356 let at = self.now();
357 self.publish(BusEvent::Removed {
358 device_id: id.clone(),
359 at,
360 });
361 Ok(())
362 }
363
364 pub fn contains(&self, id: &DeviceId) -> bool {
366 self.devices.contains_key(id)
367 }
368
369 pub fn get(&self, id: &DeviceId) -> Option<&dyn Device> {
371 self.devices.get(id).map(|slot| slot.device.as_ref())
372 }
373
374 pub fn summary(&self, id: &DeviceId) -> Option<DeviceSummary> {
376 self.devices.get(id).map(|slot| summarize(id, slot))
377 }
378
379 pub fn list(&self) -> Vec<DeviceSummary> {
381 self.devices
382 .iter()
383 .map(|(id, slot)| summarize(id, slot))
384 .collect()
385 }
386
387 pub fn host_bytes(
391 &mut self,
392 id: &DeviceId,
393 data: &[u8],
394 ) -> Result<Vec<DeviceEvent>, RegistryError> {
395 let now = self.now();
396 self.slot(id)?;
397 self.record(
398 id,
399 now,
400 EventKind::BytesIn {
401 data: bytes::Bytes::copy_from_slice(data),
402 },
403 );
404 let events = self.slot_mut(id)?.device.on_host_bytes(data, now);
405 self.absorb(id, now, &events);
406 Ok(events)
407 }
408
409 pub fn host_signals(
412 &mut self,
413 id: &DeviceId,
414 signals: crate::bridge::SerialSignals,
415 ) -> Result<(), RegistryError> {
416 let now = self.now();
417 self.slot(id)?;
418 self.record(id, now, EventKind::Signals { signals });
419 Ok(())
420 }
421
422 pub fn control(
424 &mut self,
425 id: &DeviceId,
426 command: ControlCommand,
427 ) -> Result<Vec<DeviceEvent>, RegistryError> {
428 let now = self.now();
429 self.slot(id)?;
430 self.record(
431 id,
432 now,
433 EventKind::Control {
434 command: command.clone(),
435 },
436 );
437 let events = self.slot_mut(id)?.device.on_control(command, now);
438 self.absorb(id, now, &events);
439 Ok(events)
440 }
441
442 pub fn log(&mut self, id: &DeviceId, entry: LogEntry) -> Result<(), RegistryError> {
445 let now = self.now();
446 self.slot(id)?;
447 let kind = EventKind::Log {
448 level: entry.level,
449 message: entry.message,
450 };
451 self.record(id, now, kind);
452 Ok(())
453 }
454
455 pub fn set_fault(
462 &mut self,
463 id: &DeviceId,
464 fault: FaultId,
465 active: bool,
466 options: FaultOptions,
467 ) -> Result<Vec<DeviceEvent>, RegistryError> {
468 if !self.slot(id)?.device.faults().contains(&fault) {
469 return Err(RegistryError::UnsupportedFault {
470 device: id.clone(),
471 fault,
472 });
473 }
474 let now = self.now();
475 let slot = self.slot_mut(id)?;
476 if active {
477 slot.faults.insert(fault.clone(), options.clone());
478 } else {
479 slot.faults.remove(&fault);
480 }
481 let events = slot
482 .device
483 .set_fault(fault.clone(), active, options.clone(), now);
484 self.record(
485 id,
486 now,
487 EventKind::Fault {
488 fault_id: fault,
489 active,
490 options,
491 },
492 );
493 self.absorb_with_state(id, now, events)
494 }
495
496 pub fn reset(&mut self, id: &DeviceId) -> Result<Vec<DeviceEvent>, RegistryError> {
499 let now = self.now();
500 let slot = self.slot_mut(id)?;
501 slot.faults.clear();
502 slot.jobs.clear();
503 let events = slot.device.reset(now);
504 self.record(id, now, EventKind::Reset);
505 self.absorb_with_state(id, now, events)
506 }
507
508 fn absorb_with_state(
511 &mut self,
512 id: &DeviceId,
513 now: Millis,
514 events: Vec<DeviceEvent>,
515 ) -> Result<Vec<DeviceEvent>, RegistryError> {
516 if events.iter().any(is_state_change) {
517 self.absorb(id, now, &events);
518 } else {
519 let mut absorbed = events.clone();
520 absorbed.push(DeviceEvent::StateChanged);
521 self.absorb(id, now, &absorbed);
522 }
523 Ok(events)
524 }
525
526 pub fn tick(&mut self) -> Vec<(DeviceId, DeviceEvent)> {
528 let now = self.now();
529 let due: Vec<DeviceId> = self
530 .devices
531 .iter()
532 .filter(|(_, slot)| slot.device.next_deadline().is_some_and(|d| d <= now))
533 .map(|(id, _)| id.clone())
534 .collect();
535 let mut out = Vec::new();
536 for id in due {
537 let Some(slot) = self.devices.get_mut(&id) else {
538 continue;
539 };
540 let events = slot.device.on_tick(now);
541 self.absorb(&id, now, &events);
542 out.extend(events.into_iter().map(|e| (id.clone(), e)));
543 }
544 out
545 }
546
547 pub fn next_deadline(&self) -> Option<Millis> {
549 self.devices
550 .values()
551 .filter_map(|slot| slot.device.next_deadline())
552 .min()
553 }
554
555 pub fn events(
559 &self,
560 id: &DeviceId,
561 since: Option<u64>,
562 ) -> Result<Vec<&EventRecord>, RegistryError> {
563 let slot = self.slot(id)?;
564 Ok(slot.log.since(since.unwrap_or(0)).collect())
565 }
566
567 pub fn jobs(&self, id: &DeviceId) -> Result<impl Iterator<Item = &Job> + '_, RegistryError> {
569 Ok(self.slot(id)?.jobs.iter().rev())
570 }
571
572 pub fn last_job(&self, id: &DeviceId) -> Result<Option<&Job>, RegistryError> {
574 Ok(self.slot(id)?.jobs.back())
575 }
576
577 pub fn drain_bus(&mut self) -> Vec<BusEvent> {
579 self.bus.drain(..).collect()
580 }
581
582 fn slot(&self, id: &DeviceId) -> Result<&Slot, RegistryError> {
585 self.devices
586 .get(id)
587 .ok_or_else(|| RegistryError::NotFound(id.clone()))
588 }
589
590 fn slot_mut(&mut self, id: &DeviceId) -> Result<&mut Slot, RegistryError> {
591 self.devices
592 .get_mut(id)
593 .ok_or_else(|| RegistryError::NotFound(id.clone()))
594 }
595
596 fn record(&mut self, id: &DeviceId, at: Millis, kind: EventKind) {
597 let seq = self.next_seq;
598 if let Some(slot) = self.devices.get_mut(id) {
599 self.next_seq += 1;
600 slot.log.push(EventRecord {
601 seq,
602 at,
603 device_id: id.clone(),
604 kind,
605 });
606 }
607 }
608
609 fn publish(&mut self, event: BusEvent) {
610 if self.bus.len() >= self.config.bus_capacity.max(1) {
611 self.bus.pop_front();
612 }
613 self.bus.push_back(event);
614 }
615
616 fn absorb(&mut self, id: &DeviceId, now: Millis, events: &[DeviceEvent]) {
619 self.absorb_at_depth(id, now, events, 0);
620 }
621
622 fn absorb_at_depth(&mut self, id: &DeviceId, now: Millis, events: &[DeviceEvent], depth: u8) {
623 let mut state_changed = false;
624 let mut signals = Vec::new();
625 let mut spent = Vec::new();
626 for event in events {
627 match event {
628 DeviceEvent::ToHost(data) => {
629 self.record(id, now, EventKind::BytesOut { data: data.clone() });
630 }
631 DeviceEvent::StateChanged => state_changed = true,
632 DeviceEvent::JobCompleted(job) => {
633 self.record(id, now, EventKind::JobCompleted { job_id: job.id });
634 let capacity = self.config.job_capacity.max(1);
635 if let Some(slot) = self.devices.get_mut(id) {
636 if slot.jobs.len() >= capacity {
637 slot.jobs.pop_front();
638 }
639 slot.jobs.push_back(job.clone());
640 }
641 self.publish(BusEvent::Output {
642 device_id: id.clone(),
643 job: job.clone(),
644 });
645 }
646 DeviceEvent::Connected => {
647 self.record(id, now, EventKind::Connected);
648 state_changed = true;
649 }
650 DeviceEvent::Disconnected => {
651 self.record(id, now, EventKind::Disconnected);
652 state_changed = true;
653 }
654 DeviceEvent::Log(entry) => {
655 let kind = EventKind::Log {
656 level: entry.level,
657 message: entry.message.clone(),
658 };
659 self.record(id, now, kind);
660 }
661 DeviceEvent::Peripheral(signal) => signals.push(signal.clone()),
662 DeviceEvent::Wedge(scan) => {
663 self.record(id, now, EventKind::Wedge { scan: scan.clone() });
664 }
665 DeviceEvent::Nfc(event) => {
666 if matches!(event, NfcEvent::ReadError { .. }) {
667 spent.push(FaultId::READ_ERROR);
669 }
670 self.record(
671 id,
672 now,
673 EventKind::Nfc {
674 event: event.clone(),
675 },
676 );
677 }
678 }
679 }
680 for fault in spent {
681 state_changed |= self.spend_fault(id, now, fault);
682 }
683 if state_changed {
684 if let Some(state) = self.devices.get(id).map(|slot| slot.device.state()) {
685 self.record(
686 id,
687 now,
688 EventKind::StateChanged {
689 state: state.clone(),
690 },
691 );
692 self.publish(BusEvent::State {
693 device_id: id.clone(),
694 state,
695 at: now,
696 });
697 }
698 }
699 for signal in signals {
700 self.deliver_peripheral(id, now, signal, depth);
701 }
702 }
703
704 fn spend_fault(&mut self, id: &DeviceId, now: Millis, fault: FaultId) -> bool {
707 let Some(slot) = self.devices.get_mut(id) else {
708 return false;
709 };
710 if slot.faults.remove(&fault).is_none() {
711 return false;
712 }
713 let _ = slot
715 .device
716 .set_fault(fault.clone(), false, FaultOptions::new(), now);
717 self.record(
718 id,
719 now,
720 EventKind::Fault {
721 fault_id: fault,
722 active: false,
723 options: FaultOptions::new(),
724 },
725 );
726 true
727 }
728
729 fn deliver_peripheral(
733 &mut self,
734 from: &DeviceId,
735 now: Millis,
736 signal: PeripheralSignal,
737 depth: u8,
738 ) {
739 if depth >= MAX_PERIPHERAL_DEPTH {
740 let message = format!(
741 "peripheral signal dropped after {MAX_PERIPHERAL_DEPTH} hops (devices signalling in a loop?)"
742 );
743 self.record(
744 from,
745 now,
746 EventKind::Log {
747 level: LogLevel::Warn,
748 message,
749 },
750 );
751 return;
752 }
753 let host = self
754 .devices
755 .get(from)
756 .and_then(|slot| slot.device.attached_to());
757 let mut targets: Vec<DeviceId> = host.into_iter().collect();
758 targets.extend(
759 self.devices
760 .iter()
761 .filter(|(id, slot)| {
762 *id != from && slot.device.attached_to().as_ref() == Some(from)
763 })
764 .map(|(id, _)| id.clone()),
765 );
766 targets.retain(|t| t != from);
767 targets.dedup();
768 for to in targets {
769 let Some(slot) = self.devices.get_mut(&to) else {
770 continue;
772 };
773 let events = slot.device.on_peripheral(from, signal.clone(), now);
774 let kind = EventKind::Peripheral {
775 from: from.clone(),
776 to: to.clone(),
777 signal: signal.clone(),
778 };
779 self.record(from, now, kind.clone());
780 self.record(&to, now, kind);
781 self.absorb_at_depth(&to, now, &events, depth + 1);
782 }
783 }
784}
785
786fn is_state_change(event: &DeviceEvent) -> bool {
787 matches!(
788 event,
789 DeviceEvent::StateChanged | DeviceEvent::Connected | DeviceEvent::Disconnected
790 )
791}
792
793fn summarize(id: &DeviceId, slot: &Slot) -> DeviceSummary {
794 DeviceSummary {
795 id: id.clone(),
796 kind: slot.kind.to_owned(),
797 state: slot.device.state(),
798 faults: slot
799 .faults
800 .iter()
801 .map(|(id, options)| ActiveFault {
802 id: id.clone(),
803 options: options.clone(),
804 })
805 .collect(),
806 supported_faults: slot.device.faults(),
807 attached_to: slot.device.attached_to(),
808 }
809}
810
811impl fmt::Debug for Registry {
812 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
813 f.debug_struct("Registry")
814 .field("config", &self.config)
815 .field("plugins", &self.plugins.keys().collect::<Vec<_>>())
816 .field("devices", &self.devices.keys().collect::<Vec<_>>())
817 .field("bus_len", &self.bus.len())
818 .finish_non_exhaustive()
819 }
820}