reddb_server/runtime/
impl_events.rs

1//! Event operator execution.
2
3use std::collections::HashSet;
4
5use super::*;
6
7impl RedDBRuntime {
8    pub fn execute_events_backfill(
9        &self,
10        raw_query: &str,
11        query: &EventsBackfillQuery,
12    ) -> RedDBResult<RuntimeQueryResult> {
13        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
14
15        let contract = self
16            .db()
17            .collection_contract_arc(&query.collection)
18            .ok_or_else(|| RedDBError::NotFound(query.collection.clone()))?;
19        if contract.declared_model == crate::catalog::CollectionModel::Queue {
20            return Err(RedDBError::Query(
21                "queues cannot be event backfill sources".to_string(),
22            ));
23        }
24
25        let subscription = contract
26            .subscriptions
27            .iter()
28            .find(|sub| sub.enabled && sub.target_queue == query.target_queue)
29            .cloned()
30            .ok_or_else(|| {
31                RedDBError::Query(format!(
32                    "no enabled event subscription from '{}' to '{}'",
33                    query.collection, query.target_queue
34                ))
35            })?;
36
37        let command_filter = match query.where_filter.as_deref() {
38            Some(sql) => Some(parse_backfill_filter(sql)?),
39            None => None,
40        };
41        let subscription_filter = subscription
42            .where_filter
43            .as_deref()
44            .and_then(parse_subscription_filter);
45        let tenant_column = self.tenant_column(&query.collection);
46        let rls_filter = if tenant_column.is_none()
47            && crate::runtime::impl_core::rls_is_enabled(self, &query.collection)
48        {
49            match crate::runtime::impl_core::rls_policy_filter(
50                self,
51                &query.collection,
52                reddb_rql::ast::PolicyAction::Select,
53            ) {
54                Some(filter) => Some(filter),
55                None => return Ok(backfill_result(raw_query, 0, 0, 0, &query.target_queue)),
56            }
57        } else {
58            None
59        };
60
61        let store = self.inner.db.store();
62        let manager = store
63            .get_collection(&query.collection)
64            .ok_or_else(|| RedDBError::NotFound(query.collection.clone()))?;
65        let snap_ctx = crate::runtime::impl_core::capture_current_snapshot()
66            .expect("event backfill executes inside a statement snapshot");
67        let mut entities = manager.scan(Some(&snap_ctx), |_| true);
68        entities.sort_by_key(|entity| entity.id.raw());
69
70        let effective_queue = crate::runtime::mutation::effective_queue_name(&subscription);
71        let mut existing_event_ids = queue_event_ids(store.as_ref(), &effective_queue)?;
72        let subscription_id = subscription_identity(&subscription);
73        let mut matched = 0u64;
74        let mut enqueued = 0u64;
75        let mut skipped = 0u64;
76
77        for entity in entities {
78            if !row_matches_filter(self, &query.collection, &entity, command_filter.as_ref()) {
79                continue;
80            }
81            if !row_matches_filter(
82                self,
83                &query.collection,
84                &entity,
85                subscription_filter.as_ref(),
86            ) {
87                continue;
88            }
89            if !row_matches_tenant_scope(&query.collection, &entity, tenant_column.as_deref()) {
90                continue;
91            }
92            if !row_matches_filter(self, &query.collection, &entity, rls_filter.as_ref()) {
93                continue;
94            }
95
96            if query.limit.is_some_and(|limit| matched >= limit) {
97                break;
98            }
99            matched += 1;
100
101            let after = crate::runtime::mutation::entity_row_json(&entity);
102            let (event_id, payload) = crate::runtime::mutation::backfill_event_payload(
103                &query.collection,
104                entity.id.raw(),
105                &after,
106                &subscription_id,
107                subscription.redact_fields.as_slice(),
108            )?;
109            if existing_event_ids.contains(&event_id) {
110                skipped += 1;
111                continue;
112            }
113
114            self.enqueue_event_payload(&effective_queue, Value::Json(payload))?;
115            existing_event_ids.insert(event_id);
116            enqueued += 1;
117        }
118
119        self.invalidate_result_cache_for_table(&effective_queue);
120        Ok(backfill_result(
121            raw_query,
122            matched,
123            enqueued,
124            skipped,
125            &effective_queue,
126        ))
127    }
128}
129
130fn parse_backfill_filter(sql: &str) -> RedDBResult<Filter> {
131    reddb_rql::Parser::new(sql)
132        .and_then(|mut parser| parser.parse_filter())
133        .map_err(|err| RedDBError::Query(format!("invalid EVENTS BACKFILL WHERE predicate: {err}")))
134}
135
136fn parse_subscription_filter(sql: &str) -> Option<Filter> {
137    reddb_rql::Parser::new(sql)
138        .ok()
139        .and_then(|mut parser| parser.parse_filter().ok())
140}
141
142fn row_matches_filter(
143    runtime: &RedDBRuntime,
144    collection: &str,
145    entity: &UnifiedEntity,
146    filter: Option<&Filter>,
147) -> bool {
148    filter.is_none_or(|filter| {
149        crate::runtime::query_exec::evaluate_entity_filter_with_db(
150            Some(&runtime.inner.db),
151            entity,
152            filter,
153            collection,
154            collection,
155        )
156    })
157}
158
159fn row_matches_tenant_scope(
160    collection: &str,
161    entity: &UnifiedEntity,
162    tenant_column: Option<&str>,
163) -> bool {
164    let Some(tenant_column) = tenant_column else {
165        return true;
166    };
167    let Some(tenant) = crate::runtime::impl_core::current_tenant() else {
168        return true;
169    };
170    let row = crate::runtime::mutation::entity_row_json(entity);
171    json_path_string(&row, tenant_column).is_some_and(|value| value == tenant)
172        || json_path_string(&row, &format!("{collection}.{tenant_column}"))
173            .is_some_and(|value| value == tenant)
174}
175
176fn json_path_string<'a>(value: &'a crate::json::Value, path: &str) -> Option<&'a str> {
177    let mut current = value;
178    for part in path.split('.') {
179        current = current.get(part)?;
180    }
181    current.as_str()
182}
183
184fn subscription_identity(subscription: &crate::catalog::SubscriptionDescriptor) -> String {
185    if subscription.name.is_empty() {
186        format!("{}->{}", subscription.source, subscription.target_queue)
187    } else {
188        subscription.name.clone()
189    }
190}
191
192fn queue_event_ids(store: &UnifiedStore, queue: &str) -> RedDBResult<HashSet<String>> {
193    let Some(manager) = store.get_collection(queue) else {
194        return Ok(HashSet::new());
195    };
196    let mut ids = HashSet::new();
197    // Deliberately visibility-free: the backfill dedup set must include
198    // consumed/ACKed messages (committed xmax, pre-vacuum). Filtering by
199    // snapshot would drop them from the set and re-enqueue events that
200    // were already delivered (issue #2137 review).
201    for entity in manager.query_all(|entity| matches!(entity.kind, EntityKind::QueueMessage { .. }))
202    {
203        let EntityData::QueueMessage(message) = entity.data else {
204            continue;
205        };
206        let Value::Json(bytes) = message.payload else {
207            continue;
208        };
209        let Ok(json) = crate::json::from_slice::<crate::json::Value>(&bytes) else {
210            continue;
211        };
212        if let Some(event_id) = json.get("event_id").and_then(|value| value.as_str()) {
213            ids.insert(event_id.to_string());
214        }
215    }
216    Ok(ids)
217}
218
219fn backfill_result(
220    raw_query: &str,
221    matched: u64,
222    enqueued: u64,
223    skipped: u64,
224    queue: &str,
225) -> RuntimeQueryResult {
226    let mut result = UnifiedResult::with_columns(vec![
227        "matched".into(),
228        "enqueued".into(),
229        "skipped".into(),
230        "queue".into(),
231    ]);
232    let mut record = UnifiedRecord::new();
233    record.set("matched", Value::UnsignedInteger(matched));
234    record.set("enqueued", Value::UnsignedInteger(enqueued));
235    record.set("skipped", Value::UnsignedInteger(skipped));
236    record.set("queue", Value::text(queue.to_string()));
237    result.push(record);
238
239    RuntimeQueryResult {
240        query: raw_query.to_string(),
241        mode: QueryMode::Sql,
242        statement: "events_backfill",
243        engine: "runtime-events",
244        result,
245        affected_rows: enqueued,
246        statement_type: "insert",
247        bookmark: None,
248        notice: None,
249    }
250}
reddb_server/runtime/impl_events.rs

reddb_server/runtime/
impl_events.rs
