coven_database/gate/
audience.rs

1//! Audience resolution and atomic partitioning of one captured host changeset.
2
3use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque};
4
5use rusqlite::ffi;
6use rusqlite::Connection;
7
8use super::ffi::{collect_deletes, for_each_change, ChangeRow, Changegroup};
9use super::model::{
10    fk_column_ref, foreign_keys, rows_referencing, truthy, Gates, SharedRows, TableGate,
11};
12use super::outbound::{
13    deleted_or_live_parent, fk_parent_row, full_state_diff, gate_store_outbound,
14    pre_write_full_state_diff, query_column_present, query_column_text, row_id_for_column_value,
15    DeletedAudiences, DeletedParent, FkParentRow, FullStateDirection, UnresolvedAudience,
16};
17use super::{
18    all_row_ids, query_mapped_rows, query_row_optional, CircleControlFailure, GateError,
19    UnsharedForeignKeyParent,
20};
21use crate::quote_ident;
22use coven_protocol::circle::{
23    row_routing_id, Audience, CircleControlCoord, CircleId, RowRoutingKey,
24};
25use coven_protocol::circle_activation::CircleCurrentState;
26
27mod inbound;
28mod partitioning;
29mod routing;
30mod snapshot_pruning;
31
32pub use inbound::store_audience_transitions;
33pub(crate) use inbound::{
34    align_inbound_scoped_root_audiences, filter_inbound_circle_changeset,
35    filter_inbound_store_rows, normalize_inbound_store_changeset,
36};
37pub(crate) use partitioning::{
38    audience_moves, partition_outbound, validate_accepted_foreign_key_closure,
39    validate_scoped_foreign_key_audiences,
40};
41pub(crate) use routing::{active_circle_control, capture_routing_changes, live_row_audience};
42pub(crate) use snapshot_pruning::{
43    prune_ineligible_scoped_rows, prune_private_routes_without_rows, retain_snapshot_audience_rows,
44    validate_snapshot_routing_state,
45};
46
47pub fn is_routing_table(table: &str) -> bool {
48    matches!(table, "_coven_audience" | "_coven_row_routes")
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct AudiencePartition {
53    pub audience: Audience,
54    pub control: Option<CirclePartitionControl>,
55    pub changeset: Vec<u8>,
56}
57
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub struct AudienceMove {
60    pub source: Audience,
61    pub destination: Audience,
62    pub rows: BTreeSet<(String, String)>,
63    /// The moved row's `_updated_at` after the change that moved it — the version
64    /// at which its whole component now lives in `destination`, and the stamp the
65    /// routing transitions for that component carry.
66    pub stamp: String,
67}
68
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub(crate) struct PartitionedAudienceWrite {
71    pub partitions: Vec<AudiencePartition>,
72    pub moves: Vec<AudienceMove>,
73}
74
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub struct CirclePartitionControl {
77    coordinate: CircleControlCoord,
78    stored_json: String,
79}
80
81#[derive(Debug, thiserror::Error)]
82pub enum CirclePartitionControlError {
83    #[error("parse Circle partition control: {0}")]
84    Json(#[from] serde_json::Error),
85    #[error("invalid Circle partition control: {0}")]
86    Control(#[from] coven_protocol::circle_control::CircleControlCoordError),
87}
88
89impl CirclePartitionControl {
90    pub fn from_stored_json(stored_json: String) -> Result<Self, CirclePartitionControlError> {
91        let coordinate: CircleControlCoord = serde_json::from_str(&stored_json)?;
92        coordinate.validate()?;
93        Ok(Self {
94            coordinate,
95            stored_json,
96        })
97    }
98
99    pub fn coordinate(&self) -> &CircleControlCoord {
100        &self.coordinate
101    }
102
103    pub fn stored_json(&self) -> &str {
104        &self.stored_json
105    }
106}
107
108pub struct RoutingChanges {
109    store_mirror: Vec<u8>,
110    private_routes: BTreeMap<Audience, Vec<u8>>,
111    deleted_rows: BTreeMap<(String, String), Audience>,
112}
113
114#[derive(Default)]
115pub struct StoreAudienceTransitions {
116    by_routing_id: HashMap<String, (Audience, String)>,
117}
118
119#[derive(Debug)]
120pub(crate) struct InboundStoreChangesets {
121    pub mirror: Vec<u8>,
122    pub rows: Vec<u8>,
123}
124
125impl RoutingChanges {
126    pub fn empty() -> Self {
127        Self {
128            store_mirror: Vec::new(),
129            private_routes: BTreeMap::new(),
130            deleted_rows: BTreeMap::new(),
131        }
132    }
133}
134
135struct PartitionGroup {
136    control: Option<CirclePartitionControl>,
137    group: Changegroup,
138}
139
140#[cfg(test)]
141mod tests;
coven_database/gate/audience.rs

coven_database/gate/
audience.rs
