1use crate::bus::{BusManager, BusMessage};
2use crate::cache::EntityCache;
3use crate::mutation_batch::{MutationBatch, SlotContext};
4use crate::view::{ViewIndex, ViewSpec};
5use crate::websocket::frame::{Frame, Mode};
6use bytes::Bytes;
7use hyperstack_interpreter::CanonicalLog;
8use serde_json::Value;
9use smallvec::SmallVec;
10use std::sync::Arc;
11use tokio::sync::mpsc;
12use tracing::{debug, error, instrument};
13
14#[cfg(feature = "otel")]
15use crate::metrics::Metrics;
16
17pub struct Projector {
18 view_index: Arc<ViewIndex>,
19 bus_manager: BusManager,
20 entity_cache: EntityCache,
21 mutations_rx: mpsc::Receiver<MutationBatch>,
22 #[cfg(feature = "otel")]
23 metrics: Option<Arc<Metrics>>,
24}
25
26impl Projector {
27 #[cfg(feature = "otel")]
28 pub fn new(
29 view_index: Arc<ViewIndex>,
30 bus_manager: BusManager,
31 entity_cache: EntityCache,
32 mutations_rx: mpsc::Receiver<MutationBatch>,
33 metrics: Option<Arc<Metrics>>,
34 ) -> Self {
35 Self {
36 view_index,
37 bus_manager,
38 entity_cache,
39 mutations_rx,
40 metrics,
41 }
42 }
43
44 #[cfg(not(feature = "otel"))]
45 pub fn new(
46 view_index: Arc<ViewIndex>,
47 bus_manager: BusManager,
48 entity_cache: EntityCache,
49 mutations_rx: mpsc::Receiver<MutationBatch>,
50 ) -> Self {
51 Self {
52 view_index,
53 bus_manager,
54 entity_cache,
55 mutations_rx,
56 }
57 }
58
59 pub async fn run(mut self) {
60 debug!("Projector started");
61
62 let mut json_buffer = Vec::with_capacity(4096);
63
64 while let Some(batch) = self.mutations_rx.recv().await {
65 let _span_guard = batch.span.enter();
66
67 let mut log = CanonicalLog::new();
68 log.set("phase", "projector");
69
70 let batch_size = batch.len();
71 let slot_context = batch.slot_context;
72 let mut frames_published = 0u32;
73 let mut errors = 0u32;
74
75 for mutation in batch.mutations.into_iter() {
76 #[cfg(feature = "otel")]
77 let export = mutation.export.clone();
78
79 match self
80 .process_mutation(mutation, slot_context, &mut json_buffer)
81 .await
82 {
83 Ok(count) => frames_published += count,
84 Err(e) => {
85 error!("Failed to process mutation: {}", e);
86 errors += 1;
87 }
88 }
89
90 #[cfg(feature = "otel")]
91 if let Some(ref metrics) = self.metrics {
92 metrics.record_mutation_processed(&export);
93 }
94 }
95
96 log.set("batch_size", batch_size)
97 .set("frames_published", frames_published)
98 .set("errors", errors);
99
100 #[cfg(feature = "otel")]
101 if let Some(ref metrics) = self.metrics {
102 metrics.record_projector_latency(log.duration_ms());
103 }
104
105 log.emit();
106 }
107
108 debug!("Projector stopped");
109 }
110
111 #[instrument(
112 name = "projector.mutation",
113 skip(self, mutation, slot_context, json_buffer),
114 fields(export = %mutation.export)
115 )]
116 async fn process_mutation(
117 &self,
118 mutation: hyperstack_interpreter::Mutation,
119 slot_context: Option<SlotContext>,
120 json_buffer: &mut Vec<u8>,
121 ) -> anyhow::Result<u32> {
122 let specs = self.view_index.by_export(&mutation.export);
123
124 if specs.is_empty() {
125 return Ok(0);
126 }
127
128 let key = Self::extract_key(&mutation.key);
129 let hyperstack_interpreter::Mutation {
130 mut patch, append, ..
131 } = mutation;
132
133 if let Some(ctx) = slot_context {
135 if let Value::Object(ref mut map) = patch {
136 map.insert("_seq".to_string(), Value::String(ctx.to_seq_string()));
137 }
138 }
139
140 let matching_specs: SmallVec<[&ViewSpec; 4]> = specs
141 .iter()
142 .filter(|spec| spec.filters.matches(&key))
143 .collect();
144
145 let match_count = matching_specs.len();
146 if match_count == 0 {
147 return Ok(0);
148 }
149
150 let mut frames_published = 0u32;
151
152 for (i, spec) in matching_specs.into_iter().enumerate() {
153 let is_last = i == match_count - 1;
154 let patch_data = if is_last {
155 std::mem::take(&mut patch)
156 } else {
157 patch.clone()
158 };
159
160 let projected = spec.projection.apply(patch_data);
161
162 let frame = Frame {
163 mode: spec.mode,
164 export: spec.id.clone(),
165 op: "patch",
166 key: key.clone(),
167 data: projected,
168 append: append.clone(),
169 };
170
171 json_buffer.clear();
172 serde_json::to_writer(&mut *json_buffer, &frame)?;
173 let payload = Arc::new(Bytes::copy_from_slice(json_buffer));
174
175 self.entity_cache
176 .upsert_with_append(&spec.id, &key, frame.data.clone(), &frame.append)
177 .await;
178
179 if spec.mode == Mode::List {
180 self.update_derived_view_caches(&spec.id, &key).await;
181 }
182
183 let message = Arc::new(BusMessage {
184 key: key.clone(),
185 entity: spec.id.clone(),
186 payload,
187 });
188
189 self.publish_frame(spec, message).await;
190 frames_published += 1;
191
192 #[cfg(feature = "otel")]
193 if let Some(ref metrics) = self.metrics {
194 let mode_str = match spec.mode {
195 Mode::List => "list",
196 Mode::State => "state",
197 Mode::Append => "append",
198 };
199 metrics.record_frame_published(mode_str, &spec.export);
200 }
201 }
202
203 Ok(frames_published)
204 }
205
206 fn extract_key(key: &serde_json::Value) -> String {
207 key.as_str()
208 .map(|s| s.to_string())
209 .or_else(|| key.as_u64().map(|n| n.to_string()))
210 .or_else(|| key.as_i64().map(|n| n.to_string()))
211 .or_else(|| {
212 key.as_array().and_then(|arr| {
213 let bytes: Vec<u8> = arr
214 .iter()
215 .filter_map(|v| v.as_u64().map(|n| n as u8))
216 .collect();
217 if bytes.len() == arr.len() {
218 Some(hex::encode(&bytes))
219 } else {
220 None
221 }
222 })
223 })
224 .unwrap_or_else(|| key.to_string())
225 }
226
227 async fn update_derived_view_caches(&self, source_view_id: &str, entity_key: &str) {
228 let derived_views = self.view_index.get_derived_views_for_source(source_view_id);
229 if derived_views.is_empty() {
230 return;
231 }
232
233 let entity_data = match self.entity_cache.get(source_view_id, entity_key).await {
234 Some(data) => data,
235 None => return,
236 };
237
238 let sorted_caches = self.view_index.sorted_caches();
239 let mut caches = sorted_caches.write().await;
240
241 for derived_spec in derived_views {
242 if let Some(cache) = caches.get_mut(&derived_spec.id) {
243 cache.upsert(entity_key.to_string(), entity_data.clone());
244 debug!(
245 "Updated sorted cache for derived view {} with key {}",
246 derived_spec.id, entity_key
247 );
248 }
249 }
250 }
251
252 #[instrument(
253 name = "projector.publish",
254 skip(self, spec, message),
255 fields(view_id = %spec.id, mode = ?spec.mode)
256 )]
257 async fn publish_frame(&self, spec: &ViewSpec, message: Arc<BusMessage>) {
258 match spec.mode {
259 Mode::State => {
260 self.bus_manager
261 .publish_state(&spec.id, &message.key, message.payload.clone())
262 .await;
263 }
264 Mode::List | Mode::Append => {
265 self.bus_manager.publish_list(&spec.id, message).await;
266 }
267 }
268 }
269}