distributed 4.2.0

CQRS/ES framework for Rust using Plain Old Rust Structs — append-only events, replay, snapshots, outbox, service bus, and pluggable infrastructure
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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

use super::*;

pub(super) fn model_has_visible_causal_owner(surface: &Surface, model: &str) -> bool {
    surface
        .projectors
        .iter()
        .filter(|projector| projector.models.iter().any(|candidate| candidate == model))
        .take(2)
        .count()
        == 1
}

fn selected_query_models(surface: &Surface) -> BTreeSet<String> {
    let mut selected = surface
        .query_fields
        .iter()
        .map(|root| root.model_name.clone())
        .collect::<BTreeSet<_>>();
    let mut pending = selected.iter().cloned().collect::<Vec<_>>();
    while let Some(model_name) = pending.pop() {
        let Some(model) = surface.models.get(&model_name) else {
            continue;
        };
        for relationship in &model.relationships {
            if selected.insert(relationship.target_model.clone()) {
                pending.push(relationship.target_model.clone());
            }
        }
    }
    selected
}

fn selected_query_dependencies(surface: &Surface) -> BTreeSet<String> {
    let mut dependencies = surface
        .query_fields
        .iter()
        .flat_map(|root| root.dependencies.iter().cloned())
        .collect::<BTreeSet<_>>();
    for model_name in selected_query_models(surface) {
        let Some(model) = surface.models.get(&model_name) else {
            continue;
        };
        dependencies.insert(model.table_name.clone());
        for relationship in &model.relationships {
            dependencies.extend(relationship.dependencies.iter().cloned());
            if let Some(aggregate) = &relationship.aggregate {
                dependencies.extend(aggregate.dependencies.iter().cloned());
            }
        }
    }
    dependencies
}

pub(super) fn query_footprint_has_record_evidence(surface: &Surface) -> bool {
    let models = selected_query_models(surface);
    !models.is_empty()
        && models
            .iter()
            .all(|model| model_has_visible_causal_owner(surface, model))
}

fn projector_partition_matches_authorization(
    surface: &Surface,
    projector: &crate::graphql::surface::SurfaceProjectionOwner,
) -> bool {
    projector.models.iter().all(|model| {
        surface
            .models
            .get(model)
            .is_some_and(|model| matches!(model.row_policy, SurfaceRowPolicy::Unrestricted))
    })
}

pub(super) fn query_footprint_supports_live_resume(surface: &Surface) -> bool {
    if surface.subscription_fields.is_empty() {
        return false;
    }
    let dependencies = selected_query_dependencies(surface);
    if dependencies.is_empty() {
        return false;
    }
    dependencies.iter().all(|dependency| {
        let mut owners = surface.projectors.iter().filter(|projector| {
            projector
                .dependencies
                .iter()
                .any(|candidate| candidate == dependency)
        });
        let Some(owner) = owners.next() else {
            return false;
        };
        owners.next().is_none()
            && owner.partition.preserves_source_sequence()
            && owner.change_epoch.is_some()
            && projector_partition_matches_authorization(surface, owner)
    })
}