Evaluate a mutation against all active shapes. Returns a list of `(session_id, shape_id)` pairs for shapes that match the mutation. The caller then pushes ShapeDelta messages to the matching sessions. For `ShapeType::Document` shapes with a non-empty predicate, the predicate bytes (MessagePack-encoded `MetadataFilter`) are decoded and evaluated against `doc_json`. An empty predicate matches all
(
&self,
tenant_id: u64,
collection: &str,
doc_id: &str,
doc_json: &serde_json::Value,
)
| 120 | /// documents in the collection. A predicate that cannot be decoded is |
| 121 | /// logged as a warning and treated as non-matching (fail-closed). |
| 122 | pub fn evaluate_mutation( |
| 123 | &self, |
| 124 | tenant_id: u64, |
| 125 | collection: &str, |
| 126 | doc_id: &str, |
| 127 | doc_json: &serde_json::Value, |
| 128 | ) -> Vec<(String, ShapeId)> { |
| 129 | let sessions = |
| 130 | crate::control::lock_utils::read_or_recover(self.sessions.read(), "shape_sessions"); |
| 131 | let mut matches = Vec::new(); |
| 132 | |
| 133 | for (session_id, client) in sessions.iter() { |
| 134 | if client.tenant_id != tenant_id { |
| 135 | continue; |
| 136 | } |
| 137 | for (shape_id, shape) in &client.shapes { |
| 138 | if !shape.could_match(collection, doc_id) { |
| 139 | continue; |
| 140 | } |
| 141 | if let ShapeType::Document { predicate, .. } = &shape.shape_type |
| 142 | && !predicate.is_empty() |
| 143 | { |
| 144 | match zerompk::from_msgpack::<MetadataFilter>(predicate) { |
| 145 | Ok(filter) => { |
| 146 | if !matches_metadata_filter(doc_json, &filter) { |
| 147 | continue; |
| 148 | } |
| 149 | } |
| 150 | Err(err) => { |
| 151 | warn!( |
| 152 | shape_id = %shape_id, |
| 153 | error = %err, |
| 154 | "failed to decode shape predicate; treating shape as non-matching" |
| 155 | ); |
| 156 | continue; |
| 157 | } |
| 158 | } |
| 159 | } |
| 160 | matches.push((session_id.clone(), shape_id.clone())); |
| 161 | } |
| 162 | } |
| 163 | |
| 164 | matches |
| 165 | } |
| 166 | |
| 167 | /// Get the session ID for a session's shape state. |
| 168 | pub fn session_info(&self, session_id: &str) -> Option<(u64, usize)> { |