harn_vm/event_log/
mod.rs

1use std::cell::RefCell;
2use std::collections::BTreeMap;
3use std::fmt;
4use std::path::{Path, PathBuf};
5use std::str::FromStr;
6use std::sync::Arc;
7
8use bytes::Bytes;
9use futures::stream::BoxStream;
10use serde::{Deserialize, Serialize};
11
12use crate::runtime_limits::RuntimeLimits;
13
14mod file;
15mod memory;
16mod sqlite;
17mod util;
18
19#[cfg(test)]
20pub(crate) use util::pin_test_occurred_at_ms;
21
22#[cfg(test)]
23mod tests;
24
25pub use file::FileEventLog;
26pub use memory::MemoryEventLog;
27pub use sqlite::SqliteEventLog;
28
29pub type EventId = u64;
30
31/// Topic the LLM observability writer appends `provider_call_response`
32/// (and sibling transcript) events to. Read by `harn portal` and
33/// `harn usage`; kept here so every reader/writer shares one literal.
34pub const HARN_LLM_TRANSCRIPT_TOPIC: &str = "agent.transcript.llm";
35
36pub const HARN_EVENT_LOG_BACKEND_ENV: &str = "HARN_EVENT_LOG_BACKEND";
37pub const HARN_EVENT_LOG_DIR_ENV: &str = "HARN_EVENT_LOG_DIR";
38pub const HARN_EVENT_LOG_SQLITE_PATH_ENV: &str = "HARN_EVENT_LOG_SQLITE_PATH";
39pub const HARN_EVENT_LOG_QUEUE_DEPTH_ENV: &str = "HARN_EVENT_LOG_QUEUE_DEPTH";
40
41#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
42pub struct Topic(String);
43
44impl Topic {
45    pub fn new(value: impl Into<String>) -> Result<Self, LogError> {
46        let value = value.into();
47        if value.is_empty() {
48            return Err(LogError::InvalidTopic("topic cannot be empty".to_string()));
49        }
50        if !value
51            .chars()
52            .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '.' | '_' | '-'))
53        {
54            return Err(LogError::InvalidTopic(format!(
55                "topic '{value}' contains unsupported characters"
56            )));
57        }
58        Ok(Self(value))
59    }
60
61    pub fn as_str(&self) -> &str {
62        &self.0
63    }
64}
65
66impl fmt::Display for Topic {
67    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
68        self.0.fmt(f)
69    }
70}
71
72impl FromStr for Topic {
73    type Err = LogError;
74
75    fn from_str(s: &str) -> Result<Self, Self::Err> {
76        Self::new(s)
77    }
78}
79
80#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
81pub struct ConsumerId(String);
82
83impl ConsumerId {
84    pub fn new(value: impl Into<String>) -> Result<Self, LogError> {
85        let value = value.into();
86        if value.trim().is_empty() {
87            return Err(LogError::InvalidConsumer(
88                "consumer id cannot be empty".to_string(),
89            ));
90        }
91        Ok(Self(value))
92    }
93
94    pub fn as_str(&self) -> &str {
95        &self.0
96    }
97}
98
99impl fmt::Display for ConsumerId {
100    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
101        self.0.fmt(f)
102    }
103}
104
105#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
106#[serde(rename_all = "snake_case")]
107pub enum EventLogBackendKind {
108    Memory,
109    File,
110    Sqlite,
111    Postgres,
112}
113
114impl fmt::Display for EventLogBackendKind {
115    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
116        match self {
117            Self::Memory => write!(f, "memory"),
118            Self::File => write!(f, "file"),
119            Self::Sqlite => write!(f, "sqlite"),
120            Self::Postgres => write!(f, "postgres"),
121        }
122    }
123}
124
125impl FromStr for EventLogBackendKind {
126    type Err = LogError;
127
128    fn from_str(value: &str) -> Result<Self, Self::Err> {
129        match value.trim().to_ascii_lowercase().as_str() {
130            "memory" => Ok(Self::Memory),
131            "file" => Ok(Self::File),
132            "sqlite" => Ok(Self::Sqlite),
133            "postgres" => Ok(Self::Postgres),
134            other => Err(LogError::Config(format!(
135                "unsupported event log backend '{other}'"
136            ))),
137        }
138    }
139}
140
141#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
142pub struct LogEvent {
143    pub kind: String,
144    pub payload: serde_json::Value,
145    #[serde(default)]
146    pub headers: BTreeMap<String, String>,
147    pub occurred_at_ms: i64,
148}
149
150impl LogEvent {
151    pub fn new(kind: impl Into<String>, payload: serde_json::Value) -> Self {
152        Self {
153            kind: kind.into(),
154            payload,
155            headers: BTreeMap::new(),
156            occurred_at_ms: util::now_ms(),
157        }
158    }
159
160    pub fn with_headers(mut self, headers: BTreeMap<String, String>) -> Self {
161        self.headers = headers;
162        self
163    }
164
165    /// Apply the unified redaction policy to this event's headers and
166    /// payload. Backends are intentionally left unaware of redaction —
167    /// emitters that need scrubbed events call this before append, and
168    /// readers that materialize events for display can apply the policy
169    /// again as defense in depth.
170    pub fn redact_in_place(&mut self, policy: &crate::redact::RedactionPolicy) {
171        self.headers = policy.redact_headers(&self.headers);
172        policy.redact_json_in_place(&mut self.payload);
173    }
174}
175
176/// Serialized event payload form for large read paths.
177///
178/// `payload` contains the original JSON bytes for backends that can expose
179/// them directly. Callers that only need to forward or hash the payload can
180/// avoid materializing a `serde_json::Value`; callers that need structured
181/// access can opt in with `payload_json`.
182#[derive(Clone, Debug, PartialEq, Eq)]
183pub struct LogEventBytes {
184    pub kind: String,
185    pub payload: Bytes,
186    pub headers: BTreeMap<String, String>,
187    pub occurred_at_ms: i64,
188}
189
190impl LogEventBytes {
191    pub fn payload_json(&self) -> Result<serde_json::Value, LogError> {
192        serde_json::from_slice(&self.payload)
193            .map_err(|error| LogError::Serde(format!("event log payload parse error: {error}")))
194    }
195
196    pub fn into_log_event(self) -> Result<LogEvent, LogError> {
197        Ok(LogEvent {
198            kind: self.kind,
199            payload: serde_json::from_slice(&self.payload).map_err(|error| {
200                LogError::Serde(format!("event log payload parse error: {error}"))
201            })?,
202            headers: self.headers,
203            occurred_at_ms: self.occurred_at_ms,
204        })
205    }
206}
207
208impl TryFrom<LogEvent> for LogEventBytes {
209    type Error = LogError;
210
211    fn try_from(event: LogEvent) -> Result<Self, Self::Error> {
212        let payload = serde_json::to_vec(&event.payload)
213            .map_err(|error| LogError::Serde(format!("event log payload encode error: {error}")))?;
214        Ok(Self {
215            kind: event.kind,
216            payload: Bytes::from(payload),
217            headers: event.headers,
218            occurred_at_ms: event.occurred_at_ms,
219        })
220    }
221}
222
223#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
224pub struct CompactReport {
225    pub removed: usize,
226    pub remaining: usize,
227    pub latest: Option<EventId>,
228    pub checkpointed: bool,
229}
230
231#[derive(Clone, Debug, PartialEq, Eq)]
232pub struct AppendOutcome {
233    pub event_id: EventId,
234    pub event: LogEvent,
235    pub inserted: bool,
236}
237
238#[derive(Clone, Debug, PartialEq, Eq)]
239pub struct EventLogDescription {
240    pub backend: EventLogBackendKind,
241    pub location: Option<PathBuf>,
242    pub size_bytes: Option<u64>,
243    pub queue_depth: usize,
244}
245
246#[derive(Debug)]
247pub enum LogError {
248    Config(String),
249    InvalidTopic(String),
250    InvalidConsumer(String),
251    Io(String),
252    Serde(String),
253    Sqlite(String),
254    ConsumerLagged(EventId),
255}
256
257impl fmt::Display for LogError {
258    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
259        match self {
260            Self::Config(message)
261            | Self::InvalidTopic(message)
262            | Self::InvalidConsumer(message)
263            | Self::Io(message)
264            | Self::Serde(message)
265            | Self::Sqlite(message) => message.fmt(f),
266            Self::ConsumerLagged(last_id) => {
267                write!(f, "subscriber lagged behind after event {last_id}")
268            }
269        }
270    }
271}
272
273impl std::error::Error for LogError {}
274
275#[allow(async_fn_in_trait)]
276pub trait EventLog: Send + Sync {
277    fn describe(&self) -> EventLogDescription;
278
279    async fn append(&self, topic: &Topic, event: LogEvent) -> Result<EventId, LogError>;
280
281    async fn flush(&self) -> Result<(), LogError>;
282
283    /// Read events strictly after `from`. `None` starts from the
284    /// beginning of the topic.
285    async fn read_range(
286        &self,
287        topic: &Topic,
288        from: Option<EventId>,
289        limit: usize,
290    ) -> Result<Vec<(EventId, LogEvent)>, LogError>;
291
292    async fn read_range_bytes(
293        &self,
294        topic: &Topic,
295        from: Option<EventId>,
296        limit: usize,
297    ) -> Result<Vec<(EventId, LogEventBytes)>, LogError> {
298        let events = self.read_range(topic, from, limit).await?;
299        events
300            .into_iter()
301            .map(|(event_id, event)| Ok((event_id, event.try_into()?)))
302            .collect()
303    }
304
305    /// `async fn` keeps the ergonomic generic surface; the boxed stream
306    /// preserves dyn-dispatch for callers that store `Arc<dyn EventLog>`.
307    async fn subscribe(
308        self: Arc<Self>,
309        topic: &Topic,
310        from: Option<EventId>,
311    ) -> Result<BoxStream<'static, Result<(EventId, LogEvent), LogError>>, LogError>;
312
313    async fn ack(
314        &self,
315        topic: &Topic,
316        consumer: &ConsumerId,
317        up_to: EventId,
318    ) -> Result<(), LogError>;
319
320    async fn consumer_cursor(
321        &self,
322        topic: &Topic,
323        consumer: &ConsumerId,
324    ) -> Result<Option<EventId>, LogError>;
325
326    async fn latest(&self, topic: &Topic) -> Result<Option<EventId>, LogError>;
327
328    async fn compact(&self, topic: &Topic, before: EventId) -> Result<CompactReport, LogError>;
329}
330
331fn backend_from_env() -> Result<EventLogBackendKind, LogError> {
332    Ok(std::env::var(HARN_EVENT_LOG_BACKEND_ENV)
333        .ok()
334        .map(|value| value.parse())
335        .transpose()?
336        .unwrap_or(EventLogBackendKind::Sqlite))
337}
338
339fn queue_depth_from_env() -> usize {
340    std::env::var(HARN_EVENT_LOG_QUEUE_DEPTH_ENV)
341        .ok()
342        .and_then(|value| value.parse::<usize>().ok())
343        .unwrap_or(RuntimeLimits::DEFAULT.default_event_log_queue_depth)
344        .max(1)
345}
346
347#[derive(Clone, Debug)]
348pub struct EventLogConfig {
349    pub backend: EventLogBackendKind,
350    pub file_dir: PathBuf,
351    pub sqlite_path: PathBuf,
352    pub queue_depth: usize,
353}
354
355impl EventLogConfig {
356    pub fn for_base_dir(base_dir: &Path) -> Result<Self, LogError> {
357        let file_dir = match std::env::var(HARN_EVENT_LOG_DIR_ENV) {
358            Ok(value) if !value.trim().is_empty() => util::resolve_path(base_dir, &value),
359            _ => crate::runtime_paths::event_log_dir(base_dir),
360        };
361        let sqlite_path = match std::env::var(HARN_EVENT_LOG_SQLITE_PATH_ENV) {
362            Ok(value) if !value.trim().is_empty() => util::resolve_path(base_dir, &value),
363            _ => crate::runtime_paths::event_log_sqlite_path(base_dir),
364        };
365
366        Ok(Self {
367            backend: backend_from_env()?,
368            file_dir,
369            sqlite_path,
370            queue_depth: queue_depth_from_env(),
371        })
372    }
373
374    /// Config for a state root the caller resolved itself.
375    ///
376    /// Backend and queue depth still come from the environment — those are
377    /// operator tuning knobs that apply to whichever log the process opens.
378    /// The *location* does not: it is taken from `state_root` verbatim, so
379    /// neither `HARN_STATE_DIR` nor the `HARN_EVENT_LOG_*` path overrides can
380    /// redirect a caller that already knows where its state lives. Use this
381    /// whenever the state root is a parameter rather than ambient
382    /// configuration — see [`install_default_at_state_root`].
383    pub fn for_state_root(state_root: &Path) -> Result<Self, LogError> {
384        Ok(Self {
385            backend: backend_from_env()?,
386            file_dir: crate::runtime_paths::event_log_dir_at_state_root(state_root),
387            sqlite_path: crate::runtime_paths::event_log_sqlite_path_at_state_root(state_root),
388            queue_depth: queue_depth_from_env(),
389        })
390    }
391
392    pub fn location(&self) -> Option<PathBuf> {
393        match self.backend {
394            EventLogBackendKind::Memory => None,
395            EventLogBackendKind::File => Some(self.file_dir.clone()),
396            EventLogBackendKind::Sqlite => Some(self.sqlite_path.clone()),
397            EventLogBackendKind::Postgres => None,
398        }
399    }
400}
401
402thread_local! {
403    static ACTIVE_EVENT_LOG: RefCell<Option<Arc<AnyEventLog>>> = const { RefCell::new(None) };
404    static PENDING_DEFAULT_EVENT_LOG: RefCell<Option<EventLogConfig>> = const { RefCell::new(None) };
405}
406
407pub fn install_default_for_base_dir(base_dir: &Path) -> Result<Arc<AnyEventLog>, LogError> {
408    install_active(&EventLogConfig::for_base_dir(base_dir)?)
409}
410
411/// Open `config` and make it this thread's active log, clearing any pending
412/// lazy default so it cannot later replace what the caller just installed.
413fn install_active(config: &EventLogConfig) -> Result<Arc<AnyEventLog>, LogError> {
414    let log = open_event_log(config)?;
415    ACTIVE_EVENT_LOG.with(|slot| {
416        *slot.borrow_mut() = Some(log.clone());
417    });
418    PENDING_DEFAULT_EVENT_LOG.with(|slot| {
419        *slot.borrow_mut() = None;
420    });
421    Ok(log)
422}
423
424/// Install the active event log at a state root the caller resolved itself.
425///
426/// The `_for_base_dir` variants derive their location through
427/// [`crate::runtime_paths::state_root`], so an absolute `HARN_STATE_DIR` in the
428/// environment silently replaces the `base_dir` they were handed. Callers that
429/// were *told* where their state lives — an orchestrator with a configured
430/// state dir, an embedder running isolated VMs side by side — must not be
431/// redirected that way, and must not have to publish their path to the whole
432/// process to be obeyed.
433pub fn install_default_at_state_root(state_root: &Path) -> Result<Arc<AnyEventLog>, LogError> {
434    install_active(&EventLogConfig::for_state_root(state_root)?)
435}
436
437pub fn install_lazy_default_for_base_dir(base_dir: &Path) -> Result<(), LogError> {
438    let config = EventLogConfig::for_base_dir(base_dir)?;
439    let has_active = ACTIVE_EVENT_LOG.with(|slot| slot.borrow().is_some());
440    if !has_active {
441        PENDING_DEFAULT_EVENT_LOG.with(|slot| {
442            *slot.borrow_mut() = Some(config);
443        });
444    }
445    Ok(())
446}
447
448pub fn install_memory_for_current_thread(queue_depth: usize) -> Arc<AnyEventLog> {
449    let log = Arc::new(AnyEventLog::Memory(MemoryEventLog::new(queue_depth.max(1))));
450    ACTIVE_EVENT_LOG.with(|slot| {
451        *slot.borrow_mut() = Some(log.clone());
452    });
453    PENDING_DEFAULT_EVENT_LOG.with(|slot| {
454        *slot.borrow_mut() = None;
455    });
456    log
457}
458
459pub fn install_active_event_log(log: Arc<AnyEventLog>) -> Arc<AnyEventLog> {
460    ACTIVE_EVENT_LOG.with(|slot| {
461        *slot.borrow_mut() = Some(log.clone());
462    });
463    PENDING_DEFAULT_EVENT_LOG.with(|slot| {
464        *slot.borrow_mut() = None;
465    });
466    log
467}
468
469pub fn active_event_log() -> Option<Arc<AnyEventLog>> {
470    if let Some(log) = ACTIVE_EVENT_LOG.with(|slot| slot.borrow().clone()) {
471        return Some(log);
472    }
473
474    let config = PENDING_DEFAULT_EVENT_LOG.with(|slot| slot.borrow_mut().take())?;
475    match open_event_log(&config) {
476        Ok(log) => Some(install_active_event_log(log)),
477        Err(error) => {
478            crate::events::log_warn("event_log.init", &error.to_string());
479            None
480        }
481    }
482}
483
484pub fn reset_active_event_log() {
485    ACTIVE_EVENT_LOG.with(|slot| {
486        *slot.borrow_mut() = None;
487    });
488    PENDING_DEFAULT_EVENT_LOG.with(|slot| {
489        *slot.borrow_mut() = None;
490    });
491}
492
493pub fn describe_for_base_dir(base_dir: &Path) -> Result<EventLogDescription, LogError> {
494    let config = EventLogConfig::for_base_dir(base_dir)?;
495    let description = match config.backend {
496        EventLogBackendKind::Memory => EventLogDescription {
497            backend: EventLogBackendKind::Memory,
498            location: None,
499            size_bytes: None,
500            queue_depth: config.queue_depth,
501        },
502        EventLogBackendKind::File => EventLogDescription {
503            backend: EventLogBackendKind::File,
504            size_bytes: Some(util::dir_size_bytes(&config.file_dir)),
505            location: Some(config.file_dir),
506            queue_depth: config.queue_depth,
507        },
508        EventLogBackendKind::Sqlite => EventLogDescription {
509            backend: EventLogBackendKind::Sqlite,
510            size_bytes: Some(util::sqlite_size_bytes(&config.sqlite_path)),
511            location: Some(config.sqlite_path),
512            queue_depth: config.queue_depth,
513        },
514        EventLogBackendKind::Postgres => EventLogDescription {
515            backend: EventLogBackendKind::Postgres,
516            location: None,
517            size_bytes: None,
518            queue_depth: config.queue_depth,
519        },
520    };
521    Ok(description)
522}
523
524pub fn open_event_log(config: &EventLogConfig) -> Result<Arc<AnyEventLog>, LogError> {
525    match config.backend {
526        EventLogBackendKind::Memory => Ok(Arc::new(AnyEventLog::Memory(MemoryEventLog::new(
527            config.queue_depth,
528        )))),
529        EventLogBackendKind::File => Ok(Arc::new(AnyEventLog::File(FileEventLog::open(
530            config.file_dir.clone(),
531            config.queue_depth,
532        )?))),
533        EventLogBackendKind::Sqlite => Ok(Arc::new(AnyEventLog::Sqlite(SqliteEventLog::open(
534            config.sqlite_path.clone(),
535            config.queue_depth,
536        )?))),
537        EventLogBackendKind::Postgres => Err(LogError::Config(
538            "postgres event logs are host-provided; the built-in event log factory supports memory, file, and sqlite"
539                .to_string(),
540        )),
541    }
542}
543
544pub enum AnyEventLog {
545    Memory(MemoryEventLog),
546    File(FileEventLog),
547    Sqlite(SqliteEventLog),
548}
549
550impl AnyEventLog {
551    pub async fn topics(&self) -> Result<Vec<Topic>, LogError> {
552        match self {
553            Self::Memory(log) => log.topics().await,
554            Self::File(log) => log.topics(),
555            Self::Sqlite(log) => log.topics(),
556        }
557    }
558
559    pub async fn append_idempotent_by_header(
560        &self,
561        topic: &Topic,
562        header: &str,
563        value: &str,
564        event: LogEvent,
565    ) -> Result<AppendOutcome, LogError> {
566        if header.trim().is_empty() {
567            return Err(LogError::Config(
568                "idempotent append header cannot be empty".to_string(),
569            ));
570        }
571        match self {
572            Self::Memory(log) => {
573                log.append_idempotent_by_header(topic, header, value, event)
574                    .await
575            }
576            Self::File(log) => log.append_idempotent_by_header(topic, header, value, event),
577            Self::Sqlite(log) => log.append_idempotent_by_header(topic, header, value, event),
578        }
579    }
580
581    /// Read the event previously appended under `(header, value)`, the read
582    /// counterpart of [`Self::append_idempotent_by_header`]. On the SQLite
583    /// backend this is an indexed JOIN; the memory/file dev backends scan.
584    pub async fn read_idempotent_by_header(
585        &self,
586        topic: &Topic,
587        header: &str,
588        value: &str,
589    ) -> Result<Option<(EventId, LogEvent)>, LogError> {
590        if header.trim().is_empty() {
591            return Err(LogError::Config(
592                "idempotent read header cannot be empty".to_string(),
593            ));
594        }
595        match self {
596            Self::Memory(log) => log.read_idempotent_by_header(topic, header, value).await,
597            Self::File(log) => log.read_idempotent_by_header(topic, header, value),
598            Self::Sqlite(log) => log.read_idempotent_by_header(topic, header, value),
599        }
600    }
601}
602
603impl EventLog for AnyEventLog {
604    fn describe(&self) -> EventLogDescription {
605        match self {
606            Self::Memory(log) => log.describe(),
607            Self::File(log) => log.describe(),
608            Self::Sqlite(log) => log.describe(),
609        }
610    }
611
612    async fn append(&self, topic: &Topic, event: LogEvent) -> Result<EventId, LogError> {
613        match self {
614            Self::Memory(log) => log.append(topic, event).await,
615            Self::File(log) => log.append(topic, event).await,
616            Self::Sqlite(log) => log.append(topic, event).await,
617        }
618    }
619
620    async fn flush(&self) -> Result<(), LogError> {
621        match self {
622            Self::Memory(log) => log.flush().await,
623            Self::File(log) => log.flush().await,
624            Self::Sqlite(log) => log.flush().await,
625        }
626    }
627
628    async fn read_range(
629        &self,
630        topic: &Topic,
631        from: Option<EventId>,
632        limit: usize,
633    ) -> Result<Vec<(EventId, LogEvent)>, LogError> {
634        match self {
635            Self::Memory(log) => log.read_range(topic, from, limit).await,
636            Self::File(log) => log.read_range(topic, from, limit).await,
637            Self::Sqlite(log) => log.read_range(topic, from, limit).await,
638        }
639    }
640
641    async fn read_range_bytes(
642        &self,
643        topic: &Topic,
644        from: Option<EventId>,
645        limit: usize,
646    ) -> Result<Vec<(EventId, LogEventBytes)>, LogError> {
647        match self {
648            Self::Memory(log) => log.read_range_bytes(topic, from, limit).await,
649            Self::File(log) => log.read_range_bytes(topic, from, limit).await,
650            Self::Sqlite(log) => log.read_range_bytes(topic, from, limit).await,
651        }
652    }
653
654    async fn subscribe(
655        self: Arc<Self>,
656        topic: &Topic,
657        from: Option<EventId>,
658    ) -> Result<BoxStream<'static, Result<(EventId, LogEvent), LogError>>, LogError> {
659        let (rx, queue_depth) = match self.as_ref() {
660            Self::Memory(log) => (
661                log.broadcasts.subscribe(topic, log.queue_depth),
662                log.queue_depth,
663            ),
664            Self::File(log) => (
665                log.broadcasts.subscribe(topic, log.queue_depth),
666                log.queue_depth,
667            ),
668            Self::Sqlite(log) => (
669                log.broadcasts.subscribe(topic, log.queue_depth),
670                log.queue_depth,
671            ),
672        };
673        let history = self.read_range(topic, from, usize::MAX).await?;
674        Ok(util::stream_from_broadcast(history, from, rx, queue_depth))
675    }
676
677    async fn ack(
678        &self,
679        topic: &Topic,
680        consumer: &ConsumerId,
681        up_to: EventId,
682    ) -> Result<(), LogError> {
683        match self {
684            Self::Memory(log) => log.ack(topic, consumer, up_to).await,
685            Self::File(log) => log.ack(topic, consumer, up_to).await,
686            Self::Sqlite(log) => log.ack(topic, consumer, up_to).await,
687        }
688    }
689
690    async fn consumer_cursor(
691        &self,
692        topic: &Topic,
693        consumer: &ConsumerId,
694    ) -> Result<Option<EventId>, LogError> {
695        match self {
696            Self::Memory(log) => log.consumer_cursor(topic, consumer).await,
697            Self::File(log) => log.consumer_cursor(topic, consumer).await,
698            Self::Sqlite(log) => log.consumer_cursor(topic, consumer).await,
699        }
700    }
701
702    async fn latest(&self, topic: &Topic) -> Result<Option<EventId>, LogError> {
703        match self {
704            Self::Memory(log) => log.latest(topic).await,
705            Self::File(log) => log.latest(topic).await,
706            Self::Sqlite(log) => log.latest(topic).await,
707        }
708    }
709
710    async fn compact(&self, topic: &Topic, before: EventId) -> Result<CompactReport, LogError> {
711        match self {
712            Self::Memory(log) => log.compact(topic, before).await,
713            Self::File(log) => log.compact(topic, before).await,
714            Self::Sqlite(log) => log.compact(topic, before).await,
715        }
716    }
717}
718
719pub fn sanitize_topic_component(value: &str) -> String {
720    value
721        .chars()
722        .map(|ch| {
723            if ch.is_ascii_alphanumeric() || matches!(ch, '.' | '_' | '-') {
724                ch
725            } else {
726                '_'
727            }
728        })
729        .collect()
730}
harn_vm/event_log/mod.rs

harn_vm/event_log/
mod.rs
