allsource-core 0.25.2

High-performance event store core built in Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

//! Explicit wire contract for callers that cannot use tolerant archive reads.
use super::archive_work;
use crate::{
    application::dto::{
        EventDto, QueryEventsRequest, QueryEventsResponse, RetainedEntityIntegrity,
    },
    domain::value_objects::{EntityId, TenantId},
    error::{AllSourceError, Result},
    store::EventStore,
};
use std::sync::Arc;

pub(super) async fn query(
    store: Arc<EventStore>,
    request: QueryEventsRequest,
    offset: usize,
    descending: bool,
    protocol: &str,
) -> Result<QueryEventsResponse> {
    if protocol != "retained-entity-v1"
        || offset != 0
        || descending
        || request.event_type.is_some()
        || request.event_type_prefix.is_some()
        || request.exclude_event_type_prefix.is_some()
        || request.payload_filter.is_some()
        || request.since.is_some()
        || request.until.is_some()
        || request.as_of.is_some()
    {
        return Err(AllSourceError::InvalidInput(
            "Strict retained read requires retained-entity-v1 without filters, offset or descending order"
                .into(),
        ));
    }
    let (Some(tenant), Some(entity), Some(limit @ 1..=1001)) =
        (request.tenant_id, request.entity_id, request.limit)
    else {
        return Err(AllSourceError::InvalidInput(
            "Strict retained read requires tenant_id, entity_id and limit 1..1001".into(),
        ));
    };
    TenantId::new(tenant.clone())?;
    EntityId::new(entity.clone())?;
    let (events, total_count) =
        archive_work::retained_query(store, tenant.clone(), entity.clone(), limit).await?;
    let count = events.len();
    Ok(QueryEventsResponse {
        events: events.into_iter().map(EventDto::from).collect(),
        count,
        total_count,
        has_more: count < total_count,
        entity_version: None,
        archive_integrity: Some(RetainedEntityIntegrity {
            protocol: "retained-entity-v1",
            tenant_id: tenant,
            entity_id: entity,
        }),
    })
}