1use crate::ast::*;
2use serde::Serialize;
3use serde_json::Value;
4use std::collections::{HashMap, HashSet};
5use tracing;
6
7pub type Register = usize;
8
9fn stop_field_path(target_path: &str) -> String {
10 format!("__stop:{}", target_path)
11}
12
13struct HandlerSegment {
15 key_identity: Option<Value>,
19 route_field: Option<FieldPath>,
22 ops: Vec<OpCode>,
23}
24
25fn route_field(resolution: &KeyResolutionStrategy) -> Option<FieldPath> {
27 let field = match resolution {
28 KeyResolutionStrategy::Embedded { primary_field }
29 | KeyResolutionStrategy::Lookup { primary_field }
30 | KeyResolutionStrategy::Computed { primary_field, .. } => primary_field,
31 KeyResolutionStrategy::TemporalLookup { lookup_field, .. } => lookup_field,
32 };
33 (!field.segments.is_empty()).then(|| field.clone())
34}
35
36fn merge_handler_opcodes(existing: &mut Vec<OpCode>, new: &[OpCode]) {
40 let mut existing_setup = Vec::new();
42 let mut existing_mappings = Vec::new();
43 let mut existing_teardown = Vec::new();
44 let mut section = 0; for opcode in existing.iter() {
47 match opcode {
48 OpCode::ReadOrInitState { .. } => {
49 existing_setup.push(opcode.clone());
50 section = 1; }
52 OpCode::UpdateState { .. } => {
53 existing_teardown.push(opcode.clone());
54 section = 2; }
56 OpCode::EmitMutation { .. } => {
57 existing_teardown.push(opcode.clone());
58 }
59 _ if section == 0 => existing_setup.push(opcode.clone()),
60 _ if section == 1 => existing_mappings.push(opcode.clone()),
61 _ => existing_teardown.push(opcode.clone()),
62 }
63 }
64
65 let mut new_mappings = Vec::new();
67 section = 0;
68
69 for opcode in new.iter() {
70 match opcode {
71 OpCode::ReadOrInitState { .. } => {
72 section = 1; }
74 OpCode::UpdateState { .. } | OpCode::EmitMutation { .. } => {
75 section = 2; }
77 _ if section == 1 => {
78 new_mappings.push(opcode.clone());
79 }
80 _ => {} }
82 }
83
84 let mut merged = Vec::new();
86 merged.extend(existing_setup);
87 merged.extend(existing_mappings);
88 merged.extend(new_mappings);
89 merged.extend(existing_teardown);
90
91 *existing = merged;
92}
93
94#[derive(Debug, Clone, Serialize)]
95pub enum OpCode {
96 AbortIfNullKey {
103 key: Register,
104 is_account_event: bool,
105 },
106 LoadEventField {
107 path: FieldPath,
108 dest: Register,
109 default: Option<Value>,
110 },
111 LoadConstant {
112 value: Value,
113 dest: Register,
114 },
115 CopyRegister {
116 source: Register,
117 dest: Register,
118 },
119 CopyRegisterIfNull {
121 source: Register,
122 dest: Register,
123 },
124 GetEventType {
125 dest: Register,
126 },
127 CreateObject {
128 dest: Register,
129 },
130 SetField {
131 object: Register,
132 path: String,
133 value: Register,
134 },
135 SetFields {
136 object: Register,
137 fields: Vec<(String, Register)>,
138 },
139 GetField {
140 object: Register,
141 path: String,
142 dest: Register,
143 },
144 ReadOrInitState {
145 state_id: u32,
146 key: Register,
147 default: Value,
148 dest: Register,
149 },
150 UpdateState {
151 state_id: u32,
152 key: Register,
153 value: Register,
154 },
155 AppendToArray {
156 object: Register,
157 path: String,
158 value: Register,
159 },
160 GetCurrentTimestamp {
161 dest: Register,
162 },
163 CreateEvent {
164 dest: Register,
165 event_value: Register,
166 },
167 CreateCapture {
168 dest: Register,
169 capture_value: Register,
170 },
171 Transform {
172 source: Register,
173 dest: Register,
174 transformation: Transformation,
175 },
176 EmitMutation {
177 entity_name: String,
178 key: Register,
179 state: Register,
180 },
181 SetFieldIfNull {
182 object: Register,
183 path: String,
184 value: Register,
185 },
186 SetFieldMax {
187 object: Register,
188 path: String,
189 value: Register,
190 },
191 UpdateTemporalIndex {
192 state_id: u32,
193 index_name: String,
194 lookup_value: Register,
195 primary_key: Register,
196 timestamp: Register,
197 },
198 LookupTemporalIndex {
199 state_id: u32,
200 index_name: String,
201 lookup_value: Register,
202 timestamp: Register,
203 dest: Register,
204 },
205 UpdateLookupIndex {
206 state_id: u32,
207 index_name: String,
208 lookup_value: Register,
209 primary_key: Register,
210 },
211 LookupIndex {
212 state_id: u32,
213 index_name: String,
214 lookup_value: Register,
215 dest: Register,
216 },
217 SetFieldSum {
219 object: Register,
220 path: String,
221 value: Register,
222 },
223 SetFieldIncrement {
225 object: Register,
226 path: String,
227 },
228 SetFieldMin {
230 object: Register,
231 path: String,
232 value: Register,
233 },
234 SetFieldWhen {
237 object: Register,
238 path: String,
239 value: Register,
240 when_instruction: String,
241 entity_name: String,
242 key_reg: Register,
243 condition_field: Option<FieldPath>,
244 condition_op: Option<ComparisonOp>,
245 condition_value: Option<Value>,
246 },
247 SetFieldUnlessStopped {
250 object: Register,
251 path: String,
252 value: Register,
253 stop_field: String,
254 stop_instruction: String,
255 entity_name: String,
256 key_reg: Register,
257 },
258 AddToUniqueSet {
261 state_id: u32,
262 set_name: String,
263 value: Register,
264 count_object: Register,
265 count_path: String,
266 },
267 ConditionalSetField {
269 object: Register,
270 path: String,
271 value: Register,
272 condition_field: FieldPath,
273 condition_op: ComparisonOp,
274 condition_value: Value,
275 },
276 ConditionalIncrement {
278 object: Register,
279 path: String,
280 condition_field: FieldPath,
281 condition_op: ComparisonOp,
282 condition_value: Value,
283 },
284 EvaluateComputedFields {
287 state: Register,
288 computed_paths: Vec<String>,
289 },
290 QueueResolver {
292 state_id: u32,
293 entity_name: String,
294 resolver: ResolverType,
295 input_path: Option<String>,
296 input_value: Option<Value>,
297 url_template: Option<Vec<UrlTemplatePart>>,
298 strategy: ResolveStrategy,
299 extracts: Vec<ResolverExtractSpec>,
300 condition: Option<ResolverCondition>,
301 schedule_at: Option<String>,
302 state: Register,
303 key: Register,
304 },
305 UpdatePdaReverseLookup {
308 state_id: u32,
309 lookup_name: String,
310 pda_address: Register,
311 primary_key: Register,
312 },
313 SegmentBoundary,
322}
323
324pub struct EntityBytecode {
325 pub state_id: u32,
326 pub handlers: HashMap<String, Vec<OpCode>>,
327 pub entity_name: String,
328 pub when_events: HashSet<String>,
329 pub non_emitted_fields: HashSet<String>,
330 pub computed_paths: Vec<String>,
331 #[allow(clippy::type_complexity)]
334 pub computed_fields_evaluator: Option<
335 Box<
336 dyn Fn(
337 &mut Value,
338 Option<u64>,
339 i64,
340 )
341 -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
342 + Send
343 + Sync,
344 >,
345 >,
346}
347
348impl std::fmt::Debug for EntityBytecode {
349 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
350 f.debug_struct("EntityBytecode")
351 .field("state_id", &self.state_id)
352 .field("handlers", &self.handlers)
353 .field("entity_name", &self.entity_name)
354 .field("when_events", &self.when_events)
355 .field("non_emitted_fields", &self.non_emitted_fields)
356 .field("computed_paths", &self.computed_paths)
357 .field(
358 "computed_fields_evaluator",
359 &self.computed_fields_evaluator.is_some(),
360 )
361 .finish()
362 }
363}
364
365#[derive(Debug)]
366pub struct MultiEntityBytecode {
367 pub entities: HashMap<String, EntityBytecode>,
368 pub event_routing: HashMap<String, Vec<String>>,
369 pub when_events: HashSet<String>,
370 pub proto_router: crate::proto_router::ProtoRouter,
371}
372
373impl MultiEntityBytecode {
374 pub fn from_single<S>(entity_name: String, spec: TypedStreamSpec<S>, state_id: u32) -> Self {
375 let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
376 let entity_bytecode = compiler.compile_entity();
377
378 let mut entities = HashMap::new();
379 let mut event_routing = HashMap::new();
380 let mut when_events = HashSet::new();
381
382 for event_type in entity_bytecode.handlers.keys() {
383 event_routing
384 .entry(event_type.clone())
385 .or_insert_with(Vec::new)
386 .push(entity_name.clone());
387 }
388
389 when_events.extend(entity_bytecode.when_events.iter().cloned());
390
391 entities.insert(entity_name, entity_bytecode);
392
393 MultiEntityBytecode {
394 entities,
395 event_routing,
396 when_events,
397 proto_router: crate::proto_router::ProtoRouter::new(),
398 }
399 }
400
401 pub fn from_entities(entities_vec: Vec<(String, Box<dyn std::any::Any>, u32)>) -> Self {
402 let entities = HashMap::new();
403 let event_routing = HashMap::new();
404 let when_events = HashSet::new();
405
406 if let Some((_entity_name, _spec_any, _state_id)) = entities_vec.into_iter().next() {
407 panic!("from_entities requires type information - use builder pattern instead");
408 }
409
410 MultiEntityBytecode {
411 entities,
412 event_routing,
413 when_events,
414 proto_router: crate::proto_router::ProtoRouter::new(),
415 }
416 }
417
418 #[allow(clippy::new_ret_no_self)]
419 pub fn new() -> MultiEntityBytecodeBuilder {
420 MultiEntityBytecodeBuilder {
421 entities: HashMap::new(),
422 event_routing: HashMap::new(),
423 when_events: HashSet::new(),
424 proto_router: crate::proto_router::ProtoRouter::new(),
425 }
426 }
427
428 pub fn fingerprint(&self) -> String {
445 use sha2::{Digest, Sha256};
446
447 fn update_sorted_set(hasher: &mut Sha256, set: &HashSet<String>) {
448 let mut items: Vec<&String> = set.iter().collect();
449 items.sort();
450 for item in items {
451 hasher.update(item.as_bytes());
452 hasher.update([0u8]);
453 }
454 hasher.update([1u8]);
455 }
456
457 fn update_canonical_json(hasher: &mut Sha256, value: &Value) {
464 match value {
465 Value::Object(map) => {
466 let mut keys: Vec<&String> = map.keys().collect();
467 keys.sort();
468 hasher.update(*b"{");
469 for key in keys {
470 hasher.update(key.as_bytes());
471 hasher.update([0u8]);
472 update_canonical_json(hasher, &map[key]);
473 }
474 hasher.update(*b"}");
475 }
476 Value::Array(items) => {
477 hasher.update(*b"[");
478 for item in items {
479 update_canonical_json(hasher, item);
480 hasher.update([0u8]);
481 }
482 hasher.update(*b"]");
483 }
484 scalar => {
487 hasher.update(scalar.to_string().as_bytes());
488 hasher.update([0u8]);
489 }
490 }
491 }
492
493 let mut hasher = Sha256::new();
494
495 let mut entity_names: Vec<&String> = self.entities.keys().collect();
496 entity_names.sort();
497 for entity_name in entity_names {
498 let entity = &self.entities[entity_name];
499 hasher.update(entity_name.as_bytes());
500 hasher.update([0u8]);
501 hasher.update(entity.state_id.to_le_bytes());
502
503 let mut event_types: Vec<&String> = entity.handlers.keys().collect();
504 event_types.sort();
505 for event_type in event_types {
506 hasher.update(event_type.as_bytes());
507 hasher.update([0u8]);
508 let ops = serde_json::to_value(&entity.handlers[event_type]).expect(
509 "opcode stream must serialize for fingerprinting; \
510 a non-serializable OpCode payload breaks snapshot invalidation",
511 );
512 update_canonical_json(&mut hasher, &ops);
513 hasher.update([0u8]);
514 }
515 hasher.update([1u8]);
516
517 update_sorted_set(&mut hasher, &entity.when_events);
518 update_sorted_set(&mut hasher, &entity.non_emitted_fields);
519 for path in &entity.computed_paths {
520 hasher.update(path.as_bytes());
521 hasher.update([0u8]);
522 }
523 hasher.update([entity.computed_fields_evaluator.is_some() as u8]);
524 }
525
526 let mut routes: Vec<(&String, &Vec<String>)> = self.event_routing.iter().collect();
527 routes.sort_by_key(|(event_type, _)| *event_type);
528 for (event_type, entity_names) in routes {
529 hasher.update(event_type.as_bytes());
530 hasher.update([0u8]);
531 let mut sorted_names = entity_names.clone();
532 sorted_names.sort();
533 for name in sorted_names {
534 hasher.update(name.as_bytes());
535 hasher.update([0u8]);
536 }
537 hasher.update([1u8]);
538 }
539
540 update_sorted_set(&mut hasher, &self.when_events);
541
542 hex::encode(hasher.finalize())
543 }
544}
545
546pub struct MultiEntityBytecodeBuilder {
547 entities: HashMap<String, EntityBytecode>,
548 event_routing: HashMap<String, Vec<String>>,
549 when_events: HashSet<String>,
550 proto_router: crate::proto_router::ProtoRouter,
551}
552
553impl MultiEntityBytecodeBuilder {
554 pub fn add_entity<S>(
555 self,
556 entity_name: String,
557 spec: TypedStreamSpec<S>,
558 state_id: u32,
559 ) -> Self {
560 self.add_entity_with_evaluator(
561 entity_name,
562 spec,
563 state_id,
564 None::<
565 fn(
566 &mut Value,
567 Option<u64>,
568 i64,
569 )
570 -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>,
571 >,
572 )
573 }
574
575 pub fn add_entity_with_evaluator<S, F>(
576 mut self,
577 entity_name: String,
578 spec: TypedStreamSpec<S>,
579 state_id: u32,
580 evaluator: Option<F>,
581 ) -> Self
582 where
583 F: Fn(
584 &mut Value,
585 Option<u64>,
586 i64,
587 ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
588 + Send
589 + Sync
590 + 'static,
591 {
592 let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
593 let mut entity_bytecode = compiler.compile_entity();
594
595 if let Some(eval) = evaluator {
597 entity_bytecode.computed_fields_evaluator = Some(Box::new(eval));
598 }
599
600 for event_type in entity_bytecode.handlers.keys() {
601 self.event_routing
602 .entry(event_type.clone())
603 .or_default()
604 .push(entity_name.clone());
605 }
606
607 self.when_events
608 .extend(entity_bytecode.when_events.iter().cloned());
609
610 self.entities.insert(entity_name, entity_bytecode);
611 self
612 }
613
614 pub fn build(self) -> MultiEntityBytecode {
615 MultiEntityBytecode {
616 entities: self.entities,
617 event_routing: self.event_routing,
618 when_events: self.when_events,
619 proto_router: self.proto_router,
620 }
621 }
622}
623
624pub struct TypedCompiler<S> {
625 pub spec: TypedStreamSpec<S>,
626 entity_name: String,
627 state_id: u32,
628}
629
630impl<S> TypedCompiler<S> {
631 pub fn new(spec: TypedStreamSpec<S>, entity_name: String) -> Self {
632 TypedCompiler {
633 spec,
634 entity_name,
635 state_id: 0,
636 }
637 }
638
639 pub fn with_state_id(mut self, state_id: u32) -> Self {
640 self.state_id = state_id;
641 self
642 }
643
644 pub fn compile(&self) -> MultiEntityBytecode {
645 let entity_bytecode = self.compile_entity();
646
647 let mut entities = HashMap::new();
648 let mut event_routing = HashMap::new();
649 let mut when_events = HashSet::new();
650
651 for event_type in entity_bytecode.handlers.keys() {
652 event_routing
653 .entry(event_type.clone())
654 .or_insert_with(Vec::new)
655 .push(self.entity_name.clone());
656 }
657
658 when_events.extend(entity_bytecode.when_events.iter().cloned());
659
660 entities.insert(self.entity_name.clone(), entity_bytecode);
661
662 MultiEntityBytecode {
663 entities,
664 event_routing,
665 when_events,
666 proto_router: crate::proto_router::ProtoRouter::new(),
667 }
668 }
669
670 fn resolver_outputs_primary_key_directly(&self, primary_field: &FieldPath) -> bool {
671 if primary_field.segments.as_slice() != ["__account_address"] {
672 return false;
673 }
674
675 let primary_key_leafs: HashSet<&str> = self
676 .spec
677 .identity
678 .primary_keys
679 .iter()
680 .map(|path| path.rsplit('.').next().unwrap_or(path.as_str()))
681 .collect();
682
683 self.spec.instruction_hooks.iter().any(|hook| {
684 hook.actions.iter().any(|action| {
685 matches!(action, HookAction::RegisterPdaMapping { seed_field, .. }
686 if seed_field
687 .segments
688 .last()
689 .is_some_and(|segment| primary_key_leafs.contains(segment.as_str())))
690 })
691 })
692 }
693
694 fn compile_entity(&self) -> EntityBytecode {
695 let mut when_events: HashSet<String> = HashSet::new();
696 let mut emit_by_path: HashMap<String, bool> = HashMap::new();
697
698 let mut debug_info = Vec::new();
700 for (index, handler_spec) in self.spec.handlers.iter().enumerate() {
701 let event_type = self.get_event_type(&handler_spec.source);
702 let program_id = match &handler_spec.source {
703 crate::ast::SourceSpec::Source { program_id, .. } => {
704 program_id.as_ref().map(|s| s.as_str()).unwrap_or("null")
705 }
706 };
707 debug_info.push(format!(
708 " [{}] EventType={}, Mappings={}, ProgramId={}",
709 index,
710 event_type,
711 handler_spec.mappings.len(),
712 program_id
713 ));
714 }
715
716 let mut segments: HashMap<String, Vec<HandlerSegment>> = HashMap::new();
731
732 for handler_spec in &self.spec.handlers {
733 for mapping in &handler_spec.mappings {
734 if let Some(when) = &mapping.when {
735 when_events.insert(when.clone());
736 }
737 let entry = emit_by_path
738 .entry(mapping.target_path.clone())
739 .or_insert(false);
740 *entry |= mapping.emit;
741 if mapping.stop.is_some() {
742 emit_by_path
743 .entry(stop_field_path(&mapping.target_path))
744 .or_insert(false);
745 }
746 }
747 let opcodes = self.compile_handler(handler_spec);
748 let event_type = self.get_event_type(&handler_spec.source);
749 let key_identity = self.key_identity(handler_spec);
750
751 let event_segments = segments.entry(event_type).or_default();
752 if let Some(existing) = event_segments
753 .iter_mut()
754 .find(|segment| segment.key_identity.as_ref() == Some(&key_identity))
755 {
756 merge_handler_opcodes(&mut existing.ops, &opcodes);
757 } else {
758 event_segments.push(HandlerSegment {
759 key_identity: Some(key_identity),
760 route_field: route_field(&handler_spec.key_resolution),
761 ops: opcodes,
762 });
763 }
764 }
765
766 for hook in &self.spec.instruction_hooks {
768 let event_type = hook.instruction_type.clone();
769 let event_segments = segments.entry(event_type).or_default();
770
771 let segment_index = match &hook.lookup_by {
777 None if !event_segments.is_empty() => Some(0),
778 None => None,
779 Some(lookup_by) => event_segments
780 .iter()
781 .position(|segment| segment.route_field.as_ref() == Some(lookup_by)),
782 };
783 let segment_index = match segment_index {
784 Some(index) => index,
785 None => {
786 let resolution = if event_segments.is_empty() {
791 None
792 } else {
793 hook.lookup_by
794 .as_ref()
795 .and_then(|lookup_by| self.hook_key_resolution(lookup_by))
796 };
797 let ops = match resolution {
798 Some(resolution) => self.compile_keyed_hook_segment(&resolution),
799 None => self.compile_hook_segment(hook),
800 };
801 event_segments.push(HandlerSegment {
802 key_identity: None,
803 route_field: hook.lookup_by.clone(),
804 ops,
805 });
806 event_segments.len() - 1
807 }
808 };
809 let handler_opcodes = &mut event_segments[segment_index].ops;
810
811 let hook_opcodes = self.compile_instruction_hook_actions(&hook.actions);
813
814 let insert_pos = handler_opcodes
818 .iter()
819 .position(|op| matches!(op, OpCode::EvaluateComputedFields { .. }))
820 .or_else(|| {
821 handler_opcodes
822 .iter()
823 .position(|op| matches!(op, OpCode::UpdateState { .. }))
824 });
825
826 if let Some(pos) = insert_pos {
827 for (i, opcode) in hook_opcodes.into_iter().enumerate() {
829 handler_opcodes.insert(pos + i, opcode);
830 }
831 }
832 }
833
834 let handlers: HashMap<String, Vec<OpCode>> = segments
835 .into_iter()
836 .map(|(event_type, event_segments)| {
837 let mut ops = Vec::new();
838 for (index, segment) in event_segments.into_iter().enumerate() {
839 if index > 0 {
840 ops.push(OpCode::SegmentBoundary);
841 }
842 ops.extend(segment.ops);
843 }
844 (event_type, ops)
845 })
846 .collect();
847
848 let non_emitted_fields: HashSet<String> = emit_by_path
849 .into_iter()
850 .filter_map(|(path, emit)| if emit { None } else { Some(path) })
851 .collect();
852
853 EntityBytecode {
854 state_id: self.state_id,
855 handlers,
856 entity_name: self.entity_name.clone(),
857 when_events,
858 non_emitted_fields,
859 computed_paths: self.spec.computed_fields.clone(),
860 computed_fields_evaluator: None,
861 }
862 }
863
864 fn key_identity(&self, spec: &TypedHandlerSpec<S>) -> Value {
868 let key_reg = 20;
869 let ops = self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings);
870 serde_json::to_value(&ops).unwrap_or(Value::Null)
871 }
872
873 fn hook_key_resolution(&self, lookup_by: &FieldPath) -> Option<KeyResolutionStrategy> {
877 let leaf = lookup_by.segments.last()?;
878 let is_primary_key_field = self
879 .spec
880 .identity
881 .primary_keys
882 .iter()
883 .any(|pk| pk.rsplit('.').next() == Some(leaf.as_str()));
884 if is_primary_key_field {
885 Some(KeyResolutionStrategy::Embedded {
886 primary_field: lookup_by.clone(),
887 })
888 } else if self.find_lookup_index_for_field(lookup_by).is_some() {
889 Some(KeyResolutionStrategy::Lookup {
890 primary_field: lookup_by.clone(),
891 })
892 } else {
893 None
894 }
895 }
896
897 fn compile_keyed_hook_segment(&self, resolution: &KeyResolutionStrategy) -> Vec<OpCode> {
901 let key_reg = 20;
902 let state_reg = 2;
903 let mut ops = self.compile_key_loading(resolution, key_reg, &[]);
904 ops.push(OpCode::AbortIfNullKey {
905 key: key_reg,
906 is_account_event: false,
907 });
908 ops.push(OpCode::ReadOrInitState {
909 state_id: self.state_id,
910 key: key_reg,
911 default: serde_json::json!({}),
912 dest: state_reg,
913 });
914 ops.push(OpCode::UpdateState {
915 state_id: self.state_id,
916 key: key_reg,
917 value: state_reg,
918 });
919 ops
920 }
921
922 fn compile_hook_segment(&self, hook: &InstructionHook) -> Vec<OpCode> {
926 let key_reg = 20;
927 let state_reg = 2;
928 let resolved_key_reg = 19;
929 let temp_reg = 18;
930
931 let mut ops = Vec::new();
932
933 ops.push(OpCode::LoadEventField {
935 path: FieldPath::new(&["__resolved_primary_key"]),
936 dest: resolved_key_reg,
937 default: Some(serde_json::json!(null)),
938 });
939
940 ops.push(OpCode::CopyRegister {
942 source: resolved_key_reg,
943 dest: key_reg,
944 });
945
946 if let Some(lookup_path) = &hook.lookup_by {
948 ops.push(OpCode::LoadEventField {
950 path: lookup_path.clone(),
951 dest: temp_reg,
952 default: None,
953 });
954
955 ops.push(OpCode::Transform {
957 source: temp_reg,
958 dest: temp_reg,
959 transformation: Transformation::HexEncode,
960 });
961
962 ops.push(OpCode::CopyRegisterIfNull {
964 source: temp_reg,
965 dest: key_reg,
966 });
967 }
968
969 ops.push(OpCode::ReadOrInitState {
970 state_id: self.state_id,
971 key: key_reg,
972 default: serde_json::json!({}),
973 dest: state_reg,
974 });
975
976 ops.push(OpCode::UpdateState {
977 state_id: self.state_id,
978 key: key_reg,
979 value: state_reg,
980 });
981
982 ops
983 }
984
985 fn compile_handler(&self, spec: &TypedHandlerSpec<S>) -> Vec<OpCode> {
986 let mut ops = Vec::new();
987 let state_reg = 2;
988 let key_reg = 20;
989
990 ops.extend(self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings));
991
992 let is_account_event = matches!(
998 spec.source,
999 SourceSpec::Source {
1000 is_account: true,
1001 ..
1002 }
1003 );
1004 ops.push(OpCode::AbortIfNullKey {
1005 key: key_reg,
1006 is_account_event,
1007 });
1008
1009 ops.push(OpCode::ReadOrInitState {
1010 state_id: self.state_id,
1011 key: key_reg,
1012 default: serde_json::json!({}),
1013 dest: state_reg,
1014 });
1015
1016 ops.extend(self.compile_temporal_index_update(
1023 &spec.key_resolution,
1024 key_reg,
1025 &spec.mappings,
1026 ));
1027
1028 for mapping in &spec.mappings {
1029 ops.extend(self.compile_mapping(mapping, state_reg, key_reg));
1030 }
1031
1032 ops.extend(self.compile_resolvers(state_reg, key_reg));
1033
1034 ops.push(OpCode::EvaluateComputedFields {
1036 state: state_reg,
1037 computed_paths: self.spec.computed_fields.clone(),
1038 });
1039
1040 ops.push(OpCode::UpdateState {
1041 state_id: self.state_id,
1042 key: key_reg,
1043 value: state_reg,
1044 });
1045
1046 if spec.emit {
1047 ops.push(OpCode::EmitMutation {
1048 entity_name: self.entity_name.clone(),
1049 key: key_reg,
1050 state: state_reg,
1051 });
1052 }
1053
1054 ops
1055 }
1056
1057 fn compile_resolvers(&self, state_reg: Register, key_reg: Register) -> Vec<OpCode> {
1058 let mut ops = Vec::new();
1059
1060 for resolver_spec in &self.spec.resolver_specs {
1061 let url_template = match &resolver_spec.resolver {
1062 ResolverType::Url(config) => match &config.url_source {
1063 UrlSource::Template(parts) => Some(parts.clone()),
1064 _ => None,
1065 },
1066 _ => None,
1067 };
1068
1069 ops.push(OpCode::QueueResolver {
1070 state_id: self.state_id,
1071 entity_name: self.entity_name.clone(),
1072 resolver: resolver_spec.resolver.clone(),
1073 input_path: resolver_spec.input_path.clone(),
1074 input_value: resolver_spec.input_value.clone(),
1075 url_template,
1076 strategy: resolver_spec.strategy.clone(),
1077 extracts: resolver_spec.extracts.clone(),
1078 condition: resolver_spec.condition.clone(),
1079 schedule_at: resolver_spec.schedule_at.clone(),
1080 state: state_reg,
1081 key: key_reg,
1082 });
1083 }
1084
1085 ops
1086 }
1087
1088 fn compile_mapping(
1089 &self,
1090 mapping: &TypedFieldMapping<S>,
1091 state_reg: Register,
1092 key_reg: Register,
1093 ) -> Vec<OpCode> {
1094 let mut ops = Vec::new();
1095 let temp_reg = 10;
1096
1097 ops.extend(self.compile_mapping_source(&mapping.source, temp_reg));
1098
1099 if let Some(transform) = &mapping.transform {
1100 ops.push(OpCode::Transform {
1101 source: temp_reg,
1102 dest: temp_reg,
1103 transformation: transform.clone(),
1104 });
1105 }
1106
1107 if let Some(stop_instruction) = &mapping.stop {
1108 if mapping.when.is_some() {
1109 tracing::warn!(
1110 "#[map] stop and when both set for {}. Ignoring when.",
1111 mapping.target_path
1112 );
1113 }
1114 if !matches!(mapping.population, PopulationStrategy::LastWrite)
1115 && !matches!(mapping.population, PopulationStrategy::Merge)
1116 {
1117 tracing::warn!(
1118 "#[map] stop ignores population strategy {:?}",
1119 mapping.population
1120 );
1121 }
1122
1123 ops.push(OpCode::SetFieldUnlessStopped {
1124 object: state_reg,
1125 path: mapping.target_path.clone(),
1126 value: temp_reg,
1127 stop_field: stop_field_path(&mapping.target_path),
1128 stop_instruction: stop_instruction.clone(),
1129 entity_name: self.entity_name.clone(),
1130 key_reg,
1131 });
1132 return ops;
1133 }
1134
1135 if let Some(when_instruction) = &mapping.when {
1136 if !matches!(mapping.population, PopulationStrategy::LastWrite)
1137 && !matches!(mapping.population, PopulationStrategy::Merge)
1138 {
1139 tracing::warn!(
1140 "#[map] when ignores population strategy {:?}",
1141 mapping.population
1142 );
1143 }
1144 let (condition_field, condition_op, condition_value) = mapping
1145 .condition
1146 .as_ref()
1147 .and_then(|cond| cond.parsed.as_ref())
1148 .and_then(|parsed| match parsed {
1149 ParsedCondition::Comparison { field, op, value } => {
1150 Some((Some(field.clone()), Some(op.clone()), Some(value.clone())))
1151 }
1152 ParsedCondition::Logical { .. } => {
1153 tracing::warn!("Logical conditions not yet supported for #[map] when");
1154 None
1155 }
1156 })
1157 .unwrap_or((None, None, None));
1158
1159 ops.push(OpCode::SetFieldWhen {
1160 object: state_reg,
1161 path: mapping.target_path.clone(),
1162 value: temp_reg,
1163 when_instruction: when_instruction.clone(),
1164 entity_name: self.entity_name.clone(),
1165 key_reg,
1166 condition_field,
1167 condition_op,
1168 condition_value,
1169 });
1170 return ops;
1171 }
1172
1173 if let Some(condition) = &mapping.condition {
1174 if let Some(parsed) = &condition.parsed {
1175 match parsed {
1176 ParsedCondition::Comparison {
1177 field,
1178 op,
1179 value: cond_value,
1180 } => {
1181 if matches!(mapping.population, PopulationStrategy::LastWrite)
1182 || matches!(mapping.population, PopulationStrategy::Merge)
1183 {
1184 ops.push(OpCode::ConditionalSetField {
1185 object: state_reg,
1186 path: mapping.target_path.clone(),
1187 value: temp_reg,
1188 condition_field: field.clone(),
1189 condition_op: op.clone(),
1190 condition_value: cond_value.clone(),
1191 });
1192 return ops;
1193 }
1194
1195 if matches!(mapping.population, PopulationStrategy::Count) {
1196 ops.push(OpCode::ConditionalIncrement {
1197 object: state_reg,
1198 path: mapping.target_path.clone(),
1199 condition_field: field.clone(),
1200 condition_op: op.clone(),
1201 condition_value: cond_value.clone(),
1202 });
1203 return ops;
1204 }
1205
1206 tracing::warn!(
1207 "Conditional #[map] not supported for population strategy {:?}",
1208 mapping.population
1209 );
1210 }
1211 ParsedCondition::Logical { .. } => {
1212 tracing::warn!("Logical conditions not yet supported for #[map]");
1213 }
1214 }
1215 }
1216 }
1217
1218 match &mapping.population {
1219 PopulationStrategy::Append => {
1220 ops.push(OpCode::AppendToArray {
1221 object: state_reg,
1222 path: mapping.target_path.clone(),
1223 value: temp_reg,
1224 });
1225 }
1226 PopulationStrategy::LastWrite => {
1227 ops.push(OpCode::SetField {
1228 object: state_reg,
1229 path: mapping.target_path.clone(),
1230 value: temp_reg,
1231 });
1232 }
1233 PopulationStrategy::SetOnce => {
1234 ops.push(OpCode::SetFieldIfNull {
1235 object: state_reg,
1236 path: mapping.target_path.clone(),
1237 value: temp_reg,
1238 });
1239 }
1240 PopulationStrategy::Merge => {
1241 ops.push(OpCode::SetField {
1242 object: state_reg,
1243 path: mapping.target_path.clone(),
1244 value: temp_reg,
1245 });
1246 }
1247 PopulationStrategy::Max => {
1248 ops.push(OpCode::SetFieldMax {
1249 object: state_reg,
1250 path: mapping.target_path.clone(),
1251 value: temp_reg,
1252 });
1253 }
1254 PopulationStrategy::Sum => {
1255 ops.push(OpCode::SetFieldSum {
1256 object: state_reg,
1257 path: mapping.target_path.clone(),
1258 value: temp_reg,
1259 });
1260 }
1261 PopulationStrategy::Count => {
1262 ops.push(OpCode::SetFieldIncrement {
1264 object: state_reg,
1265 path: mapping.target_path.clone(),
1266 });
1267 }
1268 PopulationStrategy::Min => {
1269 ops.push(OpCode::SetFieldMin {
1270 object: state_reg,
1271 path: mapping.target_path.clone(),
1272 value: temp_reg,
1273 });
1274 }
1275 PopulationStrategy::UniqueCount => {
1276 let set_name = format!("{}_unique_set", mapping.target_path);
1279 ops.push(OpCode::AddToUniqueSet {
1280 state_id: self.state_id,
1281 set_name,
1282 value: temp_reg,
1283 count_object: state_reg,
1284 count_path: mapping.target_path.clone(),
1285 });
1286 }
1287 }
1288
1289 ops
1290 }
1291
1292 fn compile_mapping_source(&self, source: &MappingSource, dest: Register) -> Vec<OpCode> {
1293 match source {
1294 MappingSource::FromSource {
1295 path,
1296 default,
1297 transform,
1298 } => {
1299 let mut ops = vec![OpCode::LoadEventField {
1300 path: path.clone(),
1301 dest,
1302 default: default.clone(),
1303 }];
1304
1305 if let Some(transform_type) = transform {
1307 ops.push(OpCode::Transform {
1308 source: dest,
1309 dest,
1310 transformation: transform_type.clone(),
1311 });
1312 }
1313
1314 ops
1315 }
1316 MappingSource::Constant(val) => {
1317 vec![OpCode::LoadConstant {
1318 value: val.clone(),
1319 dest,
1320 }]
1321 }
1322 MappingSource::AsEvent { fields } => {
1323 let mut ops = Vec::new();
1324
1325 if fields.is_empty() {
1326 let event_data_reg = dest + 1;
1327 ops.push(OpCode::LoadEventField {
1328 path: FieldPath::new(&[]),
1329 dest: event_data_reg,
1330 default: Some(serde_json::json!({})),
1331 });
1332 ops.push(OpCode::CreateEvent {
1333 dest,
1334 event_value: event_data_reg,
1335 });
1336 } else {
1337 let data_obj_reg = dest + 1;
1338 ops.push(OpCode::CreateObject { dest: data_obj_reg });
1339
1340 let mut field_registers = Vec::new();
1341 let mut current_reg = dest + 2;
1342
1343 for field_source in fields.iter() {
1344 if let MappingSource::FromSource {
1345 path,
1346 default,
1347 transform,
1348 } = &**field_source
1349 {
1350 ops.push(OpCode::LoadEventField {
1351 path: path.clone(),
1352 dest: current_reg,
1353 default: default.clone(),
1354 });
1355
1356 if let Some(transform_type) = transform {
1357 ops.push(OpCode::Transform {
1358 source: current_reg,
1359 dest: current_reg,
1360 transformation: transform_type.clone(),
1361 });
1362 }
1363
1364 if let Some(field_name) = path.segments.last() {
1365 field_registers.push((field_name.clone(), current_reg));
1366 }
1367 current_reg += 1;
1368 }
1369 }
1370
1371 if !field_registers.is_empty() {
1372 ops.push(OpCode::SetFields {
1373 object: data_obj_reg,
1374 fields: field_registers,
1375 });
1376 }
1377
1378 ops.push(OpCode::CreateEvent {
1379 dest,
1380 event_value: data_obj_reg,
1381 });
1382 }
1383
1384 ops
1385 }
1386 MappingSource::WholeSource => {
1387 vec![OpCode::LoadEventField {
1388 path: FieldPath::new(&[]),
1389 dest,
1390 default: Some(serde_json::json!({})),
1391 }]
1392 }
1393 MappingSource::AsCapture { field_transforms } => {
1394 let capture_data_reg = 22; let mut ops = vec![OpCode::LoadEventField {
1397 path: FieldPath::new(&[]),
1398 dest: capture_data_reg,
1399 default: Some(serde_json::json!({})),
1400 }];
1401
1402 let field_reg = 24;
1406 let transformed_reg = 25;
1407
1408 for (field_name, transform) in field_transforms {
1409 ops.push(OpCode::GetField {
1412 object: capture_data_reg,
1413 path: field_name.clone(),
1414 dest: field_reg,
1415 });
1416
1417 ops.push(OpCode::Transform {
1419 source: field_reg,
1420 dest: transformed_reg,
1421 transformation: transform.clone(),
1422 });
1423
1424 ops.push(OpCode::SetField {
1426 object: capture_data_reg,
1427 path: field_name.clone(),
1428 value: transformed_reg,
1429 });
1430 }
1431
1432 ops.push(OpCode::CreateCapture {
1434 dest,
1435 capture_value: capture_data_reg,
1436 });
1437
1438 ops
1439 }
1440 MappingSource::FromContext { field } => {
1441 vec![OpCode::LoadEventField {
1443 path: FieldPath::new(&["__update_context", field.as_str()]),
1444 dest,
1445 default: Some(serde_json::json!(null)),
1446 }]
1447 }
1448 MappingSource::Computed { .. } => {
1449 vec![]
1450 }
1451 MappingSource::FromState { .. } => {
1452 vec![]
1453 }
1454 }
1455 }
1456
1457 pub fn compile_key_loading(
1458 &self,
1459 resolution: &KeyResolutionStrategy,
1460 key_reg: Register,
1461 mappings: &[TypedFieldMapping<S>],
1462 ) -> Vec<OpCode> {
1463 let mut ops = Vec::new();
1464
1465 let resolved_key_reg = 19; ops.push(OpCode::LoadEventField {
1469 path: FieldPath::new(&["__resolved_primary_key"]),
1470 dest: resolved_key_reg,
1471 default: Some(serde_json::json!(null)),
1472 });
1473
1474 match resolution {
1476 KeyResolutionStrategy::Embedded { primary_field } => {
1477 let effective_primary_field = if primary_field.segments.is_empty() {
1479 if let Some(auto_field) = self.auto_detect_primary_field(mappings) {
1481 auto_field
1482 } else {
1483 primary_field.clone()
1484 }
1485 } else {
1486 primary_field.clone()
1487 };
1488
1489 if !effective_primary_field.segments.is_empty() {
1493 let temp_reg = 18;
1494 let transform_reg = 23; ops.push(OpCode::LoadEventField {
1497 path: effective_primary_field.clone(),
1498 dest: temp_reg,
1499 default: None,
1500 });
1501
1502 let primary_key_transform = self
1505 .find_primary_key_transformation(mappings)
1506 .or_else(|| self.find_inherited_primary_key_transformation());
1507
1508 if let Some(transform) = primary_key_transform {
1509 ops.push(OpCode::Transform {
1511 source: temp_reg,
1512 dest: transform_reg,
1513 transformation: transform,
1514 });
1515 ops.push(OpCode::CopyRegister {
1519 source: transform_reg,
1520 dest: key_reg,
1521 });
1522 } else {
1523 ops.push(OpCode::CopyRegister {
1525 source: temp_reg,
1526 dest: key_reg,
1527 });
1528 }
1529 ops.push(OpCode::CopyRegisterIfNull {
1530 source: resolved_key_reg,
1531 dest: key_reg,
1532 });
1533 } else {
1534 ops.push(OpCode::CopyRegister {
1536 source: resolved_key_reg,
1537 dest: key_reg,
1538 });
1539 }
1540 }
1541 KeyResolutionStrategy::Lookup { primary_field } => {
1542 let lookup_reg = 15;
1543 let result_reg = 17;
1544
1545 ops.push(OpCode::CopyRegister {
1551 source: resolved_key_reg,
1552 dest: lookup_reg,
1553 });
1554
1555 let temp_reg = 18;
1556 ops.push(OpCode::LoadEventField {
1557 path: primary_field.clone(),
1558 dest: temp_reg,
1559 default: None,
1560 });
1561 ops.push(OpCode::CopyRegisterIfNull {
1562 source: temp_reg,
1563 dest: lookup_reg,
1564 });
1565
1566 let index_name = self.find_lookup_index_for_lookup_field(primary_field, mappings);
1567 let effective_index_name =
1568 index_name.unwrap_or_else(|| "default_pda_lookup".to_string());
1569
1570 ops.push(OpCode::LookupIndex {
1571 state_id: self.state_id,
1572 index_name: effective_index_name,
1573 lookup_value: lookup_reg,
1574 dest: result_reg,
1575 });
1576 ops.push(OpCode::CopyRegister {
1577 source: result_reg,
1578 dest: key_reg,
1579 });
1580
1581 if self.resolver_outputs_primary_key_directly(primary_field) {
1589 ops.push(OpCode::CopyRegisterIfNull {
1590 source: resolved_key_reg,
1591 dest: key_reg,
1592 });
1593 }
1594 }
1595 KeyResolutionStrategy::Computed {
1596 primary_field,
1597 compute_partition: _,
1598 } => {
1599 ops.push(OpCode::CopyRegister {
1601 source: resolved_key_reg,
1602 dest: key_reg,
1603 });
1604 let temp_reg = 18;
1605 ops.push(OpCode::LoadEventField {
1606 path: primary_field.clone(),
1607 dest: temp_reg,
1608 default: None,
1609 });
1610 ops.push(OpCode::CopyRegisterIfNull {
1611 source: temp_reg,
1612 dest: key_reg,
1613 });
1614 }
1615 KeyResolutionStrategy::TemporalLookup {
1616 lookup_field,
1617 timestamp_field,
1618 index_name,
1619 } => {
1620 ops.push(OpCode::CopyRegister {
1622 source: resolved_key_reg,
1623 dest: key_reg,
1624 });
1625 let lookup_reg = 15;
1626 let timestamp_reg = 16;
1627 let result_reg = 17;
1628
1629 ops.push(OpCode::LoadEventField {
1630 path: lookup_field.clone(),
1631 dest: lookup_reg,
1632 default: None,
1633 });
1634
1635 ops.push(OpCode::LoadEventField {
1636 path: timestamp_field.clone(),
1637 dest: timestamp_reg,
1638 default: None,
1639 });
1640
1641 ops.push(OpCode::LookupTemporalIndex {
1642 state_id: self.state_id,
1643 index_name: index_name.clone(),
1644 lookup_value: lookup_reg,
1645 timestamp: timestamp_reg,
1646 dest: result_reg,
1647 });
1648
1649 ops.push(OpCode::CopyRegisterIfNull {
1650 source: result_reg,
1651 dest: key_reg,
1652 });
1653 }
1654 }
1655
1656 ops
1657 }
1658
1659 fn find_primary_key_transformation(
1660 &self,
1661 mappings: &[TypedFieldMapping<S>],
1662 ) -> Option<Transformation> {
1663 let primary_key = self.spec.identity.primary_keys.first()?;
1665 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1666
1667 for mapping in mappings {
1669 if mapping.target_path == *primary_key
1671 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1672 {
1673 if let Some(transform) = &mapping.transform {
1675 return Some(transform.clone());
1676 }
1677
1678 if let MappingSource::FromSource {
1680 transform: Some(transform),
1681 ..
1682 } = &mapping.source
1683 {
1684 return Some(transform.clone());
1685 }
1686 }
1687 }
1688
1689 for mapping in mappings {
1691 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1692 if let Some(transform) = field_transforms.get(&primary_field_name) {
1693 return Some(transform.clone());
1694 }
1695 }
1696 }
1697
1698 None
1699 }
1700
1701 pub fn find_inherited_primary_key_transformation(&self) -> Option<Transformation> {
1704 let primary_key = self.spec.identity.primary_keys.first()?;
1705
1706 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1708
1709 for handler in &self.spec.handlers {
1711 for mapping in &handler.mappings {
1712 if mapping.target_path == *primary_key
1714 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1715 {
1716 if let MappingSource::FromSource {
1718 path, transform, ..
1719 } = &mapping.source
1720 {
1721 if path.segments.last() == Some(&primary_field_name) {
1722 return mapping.transform.clone().or_else(|| transform.clone());
1724 }
1725 }
1726 }
1727
1728 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1730 if let Some(transform) = field_transforms.get(&primary_field_name) {
1731 return Some(transform.clone());
1732 }
1733 }
1734 }
1735 }
1736
1737 None
1738 }
1739
1740 fn extract_primary_field_name(&self, primary_key: &str) -> Option<String> {
1742 primary_key.split('.').next_back().map(|s| s.to_string())
1744 }
1745
1746 pub fn auto_detect_primary_field(
1749 &self,
1750 current_mappings: &[TypedFieldMapping<S>],
1751 ) -> Option<FieldPath> {
1752 let primary_key = self.spec.identity.primary_keys.first()?;
1753
1754 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1756
1757 if self.current_account_has_primary_field(&primary_field_name, current_mappings) {
1759 return Some(FieldPath::new(&[&primary_field_name]));
1760 }
1761
1762 None
1763 }
1764
1765 fn current_account_has_primary_field(
1768 &self,
1769 field_name: &str,
1770 mappings: &[TypedFieldMapping<S>],
1771 ) -> bool {
1772 for mapping in mappings {
1774 if let MappingSource::FromSource { path, .. } = &mapping.source {
1775 if path.segments.last() == Some(&field_name.to_string()) {
1777 return true;
1778 }
1779 }
1780 }
1781
1782 false
1783 }
1784
1785 #[allow(dead_code)]
1787 fn handler_has_field(&self, field_name: &str, mappings: &[TypedFieldMapping<S>]) -> bool {
1788 for mapping in mappings {
1789 if let MappingSource::FromSource { path, .. } = &mapping.source {
1790 if path.segments.last() == Some(&field_name.to_string()) {
1791 return true;
1792 }
1793 }
1794 }
1795 false
1796 }
1797
1798 #[allow(dead_code)]
1801 fn field_exists_in_mappings(
1802 &self,
1803 field_name: &str,
1804 mappings: &[TypedFieldMapping<S>],
1805 ) -> bool {
1806 for mapping in mappings {
1808 if let MappingSource::FromSource { path, .. } = &mapping.source {
1809 if path.segments.last() == Some(&field_name.to_string()) {
1810 return true;
1811 }
1812 }
1813 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1815 if field_transforms.contains_key(field_name) {
1816 return true;
1817 }
1818 }
1819 }
1820 false
1821 }
1822
1823 fn find_lookup_index_for_field(&self, field_path: &FieldPath) -> Option<String> {
1824 if field_path.segments.is_empty() {
1825 return None;
1826 }
1827
1828 let lookup_field_name = field_path.segments.last().unwrap();
1829
1830 for lookup_index in &self.spec.identity.lookup_indexes {
1831 let index_field_name = lookup_index
1832 .field_name
1833 .split('.')
1834 .next_back()
1835 .unwrap_or(&lookup_index.field_name);
1836 let matches_directly = index_field_name == lookup_field_name;
1837 let matches_address_alias = index_field_name
1844 .strip_suffix("_address")
1845 .map(|base| base == lookup_field_name)
1846 .unwrap_or(false);
1847
1848 if matches_directly || matches_address_alias {
1849 return Some(format!("{}_lookup_index", index_field_name));
1850 }
1851 }
1852
1853 None
1854 }
1855
1856 fn find_lookup_index_for_lookup_field(
1859 &self,
1860 primary_field: &FieldPath,
1861 mappings: &[TypedFieldMapping<S>],
1862 ) -> Option<String> {
1863 let primary_path = primary_field.segments.join(".");
1865
1866 for mapping in mappings {
1868 if let MappingSource::FromSource { path, .. } = &mapping.source {
1870 let source_path = path.segments.join(".");
1871 if source_path == primary_path {
1872 for lookup_index in &self.spec.identity.lookup_indexes {
1874 if mapping.target_path == lookup_index.field_name {
1875 let index_field_name = lookup_index
1876 .field_name
1877 .split('.')
1878 .next_back()
1879 .unwrap_or(&lookup_index.field_name);
1880 return Some(format!("{}_lookup_index", index_field_name));
1881 }
1882 }
1883 }
1884 }
1885 }
1886
1887 self.find_lookup_index_for_field(primary_field)
1889 }
1890
1891 fn find_source_path_for_lookup_index(
1895 &self,
1896 mappings: &[TypedFieldMapping<S>],
1897 lookup_field_name: &str,
1898 ) -> Option<Vec<String>> {
1899 for mapping in mappings {
1900 if mapping.target_path == lookup_field_name {
1901 if let MappingSource::FromSource { path, .. } = &mapping.source {
1902 return Some(path.segments.clone());
1903 }
1904 }
1905 }
1906 None
1907 }
1908
1909 fn compile_temporal_index_update(
1910 &self,
1911 resolution: &KeyResolutionStrategy,
1912 key_reg: Register,
1913 mappings: &[TypedFieldMapping<S>],
1914 ) -> Vec<OpCode> {
1915 let mut ops = Vec::new();
1916
1917 for lookup_index in &self.spec.identity.lookup_indexes {
1918 let lookup_reg = 17;
1919 let source_field = lookup_index
1920 .field_name
1921 .split('.')
1922 .next_back()
1923 .unwrap_or(&lookup_index.field_name);
1924
1925 match resolution {
1926 KeyResolutionStrategy::Embedded { primary_field: _ } => {
1927 let source_path_opt =
1930 self.find_source_path_for_lookup_index(mappings, &lookup_index.field_name);
1931
1932 let load_path = if let Some(ref path) = source_path_opt {
1933 FieldPath::new(&path.iter().map(|s| s.as_str()).collect::<Vec<_>>())
1934 } else {
1935 FieldPath::new(&[source_field])
1937 };
1938
1939 ops.push(OpCode::LoadEventField {
1940 path: load_path,
1941 dest: lookup_reg,
1942 default: None,
1943 });
1944
1945 if let Some(temporal_field_name) = &lookup_index.temporal_field {
1946 let timestamp_reg = 18;
1947
1948 ops.push(OpCode::LoadEventField {
1949 path: FieldPath::new(&[temporal_field_name]),
1950 dest: timestamp_reg,
1951 default: None,
1952 });
1953
1954 let index_name = format!("{}_temporal_index", source_field);
1955 ops.push(OpCode::UpdateTemporalIndex {
1956 state_id: self.state_id,
1957 index_name,
1958 lookup_value: lookup_reg,
1959 primary_key: key_reg,
1960 timestamp: timestamp_reg,
1961 });
1962
1963 let simple_index_name = format!("{}_lookup_index", source_field);
1964 ops.push(OpCode::UpdateLookupIndex {
1965 state_id: self.state_id,
1966 index_name: simple_index_name,
1967 lookup_value: lookup_reg,
1968 primary_key: key_reg,
1969 });
1970 } else {
1971 let index_name = format!("{}_lookup_index", source_field);
1972 ops.push(OpCode::UpdateLookupIndex {
1973 state_id: self.state_id,
1974 index_name,
1975 lookup_value: lookup_reg,
1976 primary_key: key_reg,
1977 });
1978 }
1979
1980 if source_path_opt.is_some() {
1984 ops.push(OpCode::UpdatePdaReverseLookup {
1985 state_id: self.state_id,
1986 lookup_name: "default_pda_lookup".to_string(),
1987 pda_address: lookup_reg,
1988 primary_key: key_reg,
1989 });
1990 }
1991 }
1992 KeyResolutionStrategy::Lookup { primary_field } => {
1993 let has_mapping_to_lookup_field = mappings
1996 .iter()
1997 .any(|m| m.target_path == lookup_index.field_name);
1998
1999 if has_mapping_to_lookup_field {
2000 let path_segments: Vec<&str> =
2003 primary_field.segments.iter().map(|s| s.as_str()).collect();
2004 ops.push(OpCode::LoadEventField {
2005 path: FieldPath::new(&path_segments),
2006 dest: lookup_reg,
2007 default: None,
2008 });
2009
2010 let index_name = format!("{}_lookup_index", source_field);
2011 ops.push(OpCode::UpdateLookupIndex {
2012 state_id: self.state_id,
2013 index_name,
2014 lookup_value: lookup_reg,
2015 primary_key: key_reg,
2016 });
2017 }
2018 }
2019 KeyResolutionStrategy::Computed { .. }
2020 | KeyResolutionStrategy::TemporalLookup { .. } => {
2021 }
2023 }
2024 }
2025
2026 ops
2027 }
2028
2029 fn get_event_type(&self, source: &SourceSpec) -> String {
2030 match source {
2031 SourceSpec::Source { type_name, .. } => type_name.clone(),
2032 }
2033 }
2034
2035 fn compile_instruction_hook_actions(&self, actions: &[HookAction]) -> Vec<OpCode> {
2036 let mut ops = Vec::new();
2037 let state_reg = 2;
2038
2039 for action in actions {
2040 match action {
2041 HookAction::SetField {
2042 target_field,
2043 source,
2044 condition,
2045 } => {
2046 let _ = condition;
2048
2049 let temp_reg = 11; let load_ops = self.compile_mapping_source(source, temp_reg);
2053 ops.extend(load_ops);
2054
2055 if let MappingSource::FromSource {
2057 transform: Some(transform_type),
2058 ..
2059 } = source
2060 {
2061 ops.push(OpCode::Transform {
2062 source: temp_reg,
2063 dest: temp_reg,
2064 transformation: transform_type.clone(),
2065 });
2066 }
2067
2068 if let Some(cond_expr) = condition {
2070 if let Some(parsed) = &cond_expr.parsed {
2071 let cond_check_ops = self.compile_condition_check(
2073 parsed,
2074 temp_reg,
2075 state_reg,
2076 target_field,
2077 );
2078 ops.extend(cond_check_ops);
2079 } else {
2080 ops.push(OpCode::SetField {
2082 object: state_reg,
2083 path: target_field.clone(),
2084 value: temp_reg,
2085 });
2086 }
2087 } else {
2088 ops.push(OpCode::SetField {
2090 object: state_reg,
2091 path: target_field.clone(),
2092 value: temp_reg,
2093 });
2094 }
2095 }
2096 HookAction::IncrementField {
2097 target_field,
2098 increment_by,
2099 condition,
2100 } => {
2101 if let Some(cond_expr) = condition {
2102 if let Some(parsed) = &cond_expr.parsed {
2103 let cond_check_ops = self.compile_conditional_increment(
2108 parsed,
2109 state_reg,
2110 target_field,
2111 *increment_by,
2112 );
2113 ops.extend(cond_check_ops);
2114 } else {
2115 ops.push(OpCode::SetFieldIncrement {
2117 object: state_reg,
2118 path: target_field.clone(),
2119 });
2120 }
2121 } else {
2122 ops.push(OpCode::SetFieldIncrement {
2124 object: state_reg,
2125 path: target_field.clone(),
2126 });
2127 }
2128 }
2129 HookAction::RegisterPdaMapping { .. } => {
2130 if let HookAction::RegisterPdaMapping {
2131 pda_field,
2132 seed_field,
2133 lookup_name,
2134 } = action
2135 {
2136 let pda_reg = 11;
2137 let seed_reg = 12;
2138
2139 ops.push(OpCode::LoadEventField {
2140 path: pda_field.clone(),
2141 dest: pda_reg,
2142 default: None,
2143 });
2144 ops.push(OpCode::LoadEventField {
2145 path: seed_field.clone(),
2146 dest: seed_reg,
2147 default: None,
2148 });
2149 ops.push(OpCode::UpdatePdaReverseLookup {
2150 state_id: self.state_id,
2151 lookup_name: lookup_name.clone(),
2152 pda_address: pda_reg,
2153 primary_key: seed_reg,
2154 });
2155 }
2156 }
2157 }
2158 }
2159
2160 ops
2161 }
2162
2163 fn compile_condition_check(
2164 &self,
2165 condition: &ParsedCondition,
2166 value_reg: Register,
2167 state_reg: Register,
2168 target_field: &str,
2169 ) -> Vec<OpCode> {
2170 match condition {
2171 ParsedCondition::Comparison {
2172 field,
2173 op,
2174 value: cond_value,
2175 } => {
2176 vec![OpCode::ConditionalSetField {
2178 object: state_reg,
2179 path: target_field.to_string(),
2180 value: value_reg,
2181 condition_field: field.clone(),
2182 condition_op: op.clone(),
2183 condition_value: cond_value.clone(),
2184 }]
2185 }
2186 ParsedCondition::Logical { .. } => {
2187 tracing::warn!("Logical conditions not yet supported in instruction hooks");
2189 vec![OpCode::SetField {
2190 object: state_reg,
2191 path: target_field.to_string(),
2192 value: value_reg,
2193 }]
2194 }
2195 }
2196 }
2197
2198 fn compile_conditional_increment(
2199 &self,
2200 condition: &ParsedCondition,
2201 state_reg: Register,
2202 target_field: &str,
2203 _increment_by: i64,
2204 ) -> Vec<OpCode> {
2205 match condition {
2206 ParsedCondition::Comparison {
2207 field,
2208 op,
2209 value: cond_value,
2210 } => {
2211 vec![OpCode::ConditionalIncrement {
2212 object: state_reg,
2213 path: target_field.to_string(),
2214 condition_field: field.clone(),
2215 condition_op: op.clone(),
2216 condition_value: cond_value.clone(),
2217 }]
2218 }
2219 ParsedCondition::Logical { .. } => {
2220 tracing::warn!("Logical conditions not yet supported in instruction hooks");
2221 vec![OpCode::SetFieldIncrement {
2222 object: state_reg,
2223 path: target_field.to_string(),
2224 }]
2225 }
2226 }
2227 }
2228}
2229
2230#[cfg(test)]
2231mod tests {
2232 use super::MultiEntityBytecode;
2233 use crate::ast::{
2234 FieldPath, HookAction, IdentitySpec, InstructionHook, KeyResolutionStrategy,
2235 LookupIndexSpec, MappingSource, PopulationStrategy, SerializableFieldMapping,
2236 SerializableHandlerSpec, SerializableStreamSpec, SourceSpec, TypedStreamSpec,
2237 };
2238 use crate::vm::VmContext;
2239 use serde_json::{json, Value};
2240 use std::collections::BTreeMap;
2241
2242 fn mapping(
2243 target_path: &str,
2244 source_path: &[&str],
2245 population: PopulationStrategy,
2246 ) -> SerializableFieldMapping {
2247 SerializableFieldMapping {
2248 target_path: target_path.to_string(),
2249 source: MappingSource::FromSource {
2250 path: FieldPath::new(source_path),
2251 default: None,
2252 transform: None,
2253 },
2254 transform: None,
2255 population,
2256 condition: None,
2257 when: None,
2258 stop: None,
2259 emit: true,
2260 }
2261 }
2262
2263 fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
2264 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2265 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2266 state_name: "PumpfunToken".to_string(),
2267 program_id: None,
2268 idl: None,
2269 identity: IdentitySpec {
2270 primary_keys: vec!["id.mint".to_string()],
2271 lookup_indexes: vec![LookupIndexSpec {
2272 field_name: "id.bonding_curve".to_string(),
2273 temporal_field: None,
2274 }],
2275 },
2276 handlers: vec![SerializableHandlerSpec {
2277 source: SourceSpec::Source {
2278 program_id: None,
2279 discriminator: None,
2280 type_name: "pump::BondingCurveState".to_string(),
2281 serialization: None,
2282 is_account: true,
2283 },
2284 key_resolution: KeyResolutionStrategy::Lookup {
2285 primary_field: FieldPath::new(&["__account_address"]),
2286 },
2287 mappings: vec![
2288 mapping(
2289 "id.bonding_curve",
2290 &["__account_address"],
2291 PopulationStrategy::SetOnce,
2292 ),
2293 mapping(
2294 "reserves.virtual_token_reserves",
2295 &["virtual_token_reserves"],
2296 PopulationStrategy::LastWrite,
2297 ),
2298 ],
2299 conditions: vec![],
2300 emit: true,
2301 }],
2302 sections: vec![],
2303 field_mappings: BTreeMap::new(),
2304 resolver_hooks: vec![],
2305 instruction_hooks: vec![InstructionHook {
2306 instruction_type: "pump::BuyIxState".to_string(),
2307 actions: vec![HookAction::RegisterPdaMapping {
2308 pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2309 seed_field: FieldPath::new(&["accounts", "mint"]),
2310 lookup_name: "default_pda_lookup".to_string(),
2311 }],
2312 lookup_by: None,
2313 }],
2314 resolver_specs: vec![],
2315 computed_fields: vec![],
2316 computed_field_specs: vec![],
2317 content_hash: None,
2318 views: vec![],
2319 })
2320 }
2321
2322 fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
2323 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2324 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2325 state_name: "OreRound".to_string(),
2326 program_id: None,
2327 idl: None,
2328 identity: IdentitySpec {
2329 primary_keys: vec!["id.round_id".to_string()],
2330 lookup_indexes: vec![LookupIndexSpec {
2331 field_name: "id.round_address".to_string(),
2332 temporal_field: None,
2333 }],
2334 },
2335 handlers: vec![SerializableHandlerSpec {
2336 source: SourceSpec::Source {
2337 program_id: None,
2338 discriminator: None,
2339 type_name: "entropy::VarState".to_string(),
2340 serialization: None,
2341 is_account: true,
2342 },
2343 key_resolution: KeyResolutionStrategy::Lookup {
2344 primary_field: FieldPath::new(&["__account_address"]),
2345 },
2346 mappings: vec![mapping(
2347 "state.expires_at",
2348 &["end_at"],
2349 PopulationStrategy::LastWrite,
2350 )],
2351 conditions: vec![],
2352 emit: true,
2353 }],
2354 sections: vec![],
2355 field_mappings: BTreeMap::new(),
2356 resolver_hooks: vec![],
2357 instruction_hooks: vec![InstructionHook {
2358 instruction_type: "ore::DeployIxState".to_string(),
2359 actions: vec![HookAction::RegisterPdaMapping {
2360 pda_field: FieldPath::new(&["accounts", "entropyVar"]),
2361 seed_field: FieldPath::new(&["accounts", "round"]),
2362 lookup_name: "default_pda_lookup".to_string(),
2363 }],
2364 lookup_by: None,
2365 }],
2366 resolver_specs: vec![],
2367 computed_fields: vec![],
2368 computed_field_specs: vec![],
2369 content_hash: None,
2370 views: vec![],
2371 })
2372 }
2373
2374 fn embedded_account_key_spec() -> TypedStreamSpec<Value> {
2375 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2376 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2377 state_name: "OreTreasury".to_string(),
2378 program_id: None,
2379 idl: None,
2380 identity: IdentitySpec {
2381 primary_keys: vec!["id.address".to_string()],
2382 lookup_indexes: vec![],
2383 },
2384 handlers: vec![SerializableHandlerSpec {
2385 source: SourceSpec::Source {
2386 program_id: None,
2387 discriminator: None,
2388 type_name: "ore::TreasuryState".to_string(),
2389 serialization: None,
2390 is_account: true,
2391 },
2392 key_resolution: KeyResolutionStrategy::Embedded {
2393 primary_field: FieldPath::new(&["__account_address"]),
2394 },
2395 mappings: vec![mapping(
2396 "id.address",
2397 &["__account_address"],
2398 PopulationStrategy::SetOnce,
2399 )],
2400 conditions: vec![],
2401 emit: true,
2402 }],
2403 sections: vec![],
2404 field_mappings: BTreeMap::new(),
2405 resolver_hooks: vec![],
2406 instruction_hooks: vec![],
2407 resolver_specs: vec![],
2408 computed_fields: vec![],
2409 computed_field_specs: vec![],
2410 content_hash: None,
2411 views: vec![],
2412 })
2413 }
2414
2415 #[test]
2416 fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
2417 let bytecode = MultiEntityBytecode::from_single(
2418 "PumpfunToken".to_string(),
2419 direct_pda_to_primary_key_spec(),
2420 0,
2421 );
2422 let mut vm = VmContext::new();
2423
2424 let mutations = vm
2425 .process_event(
2426 &bytecode,
2427 json!({
2428 "__account_address": "bonding_curve_1",
2429 "__resolved_primary_key": "mint_1",
2430 "virtual_token_reserves": 42,
2431 }),
2432 "pump::BondingCurveState",
2433 None,
2434 None,
2435 )
2436 .unwrap();
2437
2438 assert_eq!(mutations.len(), 1);
2439 assert_eq!(mutations[0].key, json!("mint_1"));
2440 assert_eq!(
2441 mutations[0].patch["id"]["bonding_curve"],
2442 json!("bonding_curve_1")
2443 );
2444 assert_eq!(
2445 mutations[0].patch["reserves"]["virtual_token_reserves"],
2446 json!(42)
2447 );
2448 }
2449
2450 #[test]
2451 fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
2452 let bytecode = MultiEntityBytecode::from_single(
2453 "OreRound".to_string(),
2454 pda_to_intermediate_lookup_spec(),
2455 0,
2456 );
2457 let mut vm = VmContext::new();
2458
2459 let mutations = vm
2460 .process_event(
2461 &bytecode,
2462 json!({
2463 "__account_address": "entropy_var_1",
2464 "__resolved_primary_key": "round_address_1",
2465 "end_at": 123,
2466 }),
2467 "entropy::VarState",
2468 None,
2469 None,
2470 )
2471 .unwrap();
2472
2473 assert!(mutations.is_empty());
2474 }
2475
2476 #[test]
2477 fn embedded_account_key_wins_over_an_unrelated_resolved_key() {
2478 let bytecode = MultiEntityBytecode::from_single(
2479 "OreTreasury".to_string(),
2480 embedded_account_key_spec(),
2481 0,
2482 );
2483 let mut vm = VmContext::new();
2484
2485 let mutations = vm
2486 .process_event(
2487 &bytecode,
2488 json!({
2489 "__account_address": "treasury_pda",
2490 "__resolved_primary_key": "round_address",
2491 }),
2492 "ore::TreasuryState",
2493 None,
2494 None,
2495 )
2496 .unwrap();
2497
2498 assert_eq!(mutations.len(), 1);
2499 assert_eq!(mutations[0].key, json!("treasury_pda"));
2500 assert_eq!(mutations[0].patch["id"]["address"], json!("treasury_pda"));
2501 }
2502
2503 fn instruction_handler(
2504 type_name: &str,
2505 key_resolution: KeyResolutionStrategy,
2506 mappings: Vec<SerializableFieldMapping>,
2507 ) -> SerializableHandlerSpec {
2508 SerializableHandlerSpec {
2509 source: SourceSpec::Source {
2510 program_id: None,
2511 discriminator: None,
2512 type_name: type_name.to_string(),
2513 serialization: None,
2514 is_account: false,
2515 },
2516 key_resolution,
2517 mappings,
2518 conditions: vec![],
2519 emit: true,
2520 }
2521 }
2522
2523 fn split_position_spec() -> TypedStreamSpec<Value> {
2528 let embedded = |field: &str| KeyResolutionStrategy::Embedded {
2529 primary_field: FieldPath::new(&["accounts", field]),
2530 };
2531 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2532 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2533 state_name: "Position".to_string(),
2534 program_id: None,
2535 idl: None,
2536 identity: IdentitySpec {
2537 primary_keys: vec!["id.position_address".to_string()],
2538 lookup_indexes: vec![],
2539 },
2540 handlers: vec![
2541 instruction_handler(
2542 "amm::SplitPositionIxState",
2543 embedded("first_position"),
2544 vec![
2545 mapping(
2546 "id.position_address",
2547 &["accounts", "first_position"],
2548 PopulationStrategy::SetOnce,
2549 ),
2550 mapping(
2551 "activity.split_source_count",
2552 &["data"],
2553 PopulationStrategy::Count,
2554 ),
2555 ],
2556 ),
2557 instruction_handler(
2558 "amm::SplitPositionIxState",
2559 embedded("second_position"),
2560 vec![
2561 mapping(
2562 "id.position_address",
2563 &["accounts", "second_position"],
2564 PopulationStrategy::SetOnce,
2565 ),
2566 mapping(
2567 "activity.split_child_count",
2568 &["data"],
2569 PopulationStrategy::Count,
2570 ),
2571 ],
2572 ),
2573 ],
2574 sections: vec![],
2575 field_mappings: BTreeMap::new(),
2576 resolver_hooks: vec![],
2577 instruction_hooks: vec![InstructionHook {
2578 instruction_type: "amm::SplitPositionIxState".to_string(),
2579 actions: vec![HookAction::SetField {
2580 target_field: "activity.last_split_source_slot".to_string(),
2581 source: MappingSource::FromSource {
2582 path: FieldPath::new(&["data", "slot"]),
2583 default: None,
2584 transform: None,
2585 },
2586 condition: None,
2587 }],
2588 lookup_by: Some(FieldPath::new(&["accounts", "first_position"])),
2589 }],
2590 resolver_specs: vec![],
2591 computed_fields: vec![],
2592 computed_field_specs: vec![],
2593 content_hash: None,
2594 views: vec![],
2595 })
2596 }
2597
2598 fn split_position_event(first: &str, second: &str) -> Value {
2599 json!({
2600 "accounts": { "first_position": first, "second_position": second },
2601 "data": { "slot": 77 },
2602 })
2603 }
2604
2605 #[test]
2606 fn one_instruction_routes_each_key_to_its_own_entity() {
2607 let bytecode =
2608 MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
2609 let mut vm = VmContext::new();
2610
2611 for _ in 0..2 {
2612 vm.process_event(
2613 &bytecode,
2614 split_position_event("source_pos", "child_pos"),
2615 "amm::SplitPositionIxState",
2616 None,
2617 None,
2618 )
2619 .unwrap();
2620 }
2621 vm.process_event(
2622 &bytecode,
2623 split_position_event("child_pos", "grandchild_pos"),
2624 "amm::SplitPositionIxState",
2625 None,
2626 None,
2627 )
2628 .unwrap();
2629
2630 let state = |key: &str| vm.get_entity_state(0, &json!(key)).unwrap();
2631
2632 let source = state("source_pos");
2633 assert_eq!(source["id"]["position_address"], json!("source_pos"));
2634 assert_eq!(source["activity"]["split_source_count"], json!(2));
2635 assert_eq!(source["activity"].get("split_child_count"), None);
2636 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
2637
2638 let child = state("child_pos");
2640 assert_eq!(child["id"]["position_address"], json!("child_pos"));
2641 assert_eq!(child["activity"]["split_child_count"], json!(2));
2642 assert_eq!(child["activity"]["split_source_count"], json!(1));
2643 assert_eq!(child["activity"]["last_split_source_slot"], json!(77));
2644
2645 let grandchild = state("grandchild_pos");
2646 assert_eq!(grandchild["activity"]["split_child_count"], json!(1));
2647 assert_eq!(grandchild["activity"].get("split_source_count"), None);
2648 assert_eq!(grandchild["activity"].get("last_split_source_slot"), None);
2649 }
2650
2651 #[test]
2652 fn one_instruction_emits_one_mutation_per_key() {
2653 let bytecode =
2654 MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
2655 let mut vm = VmContext::new();
2656
2657 let mutations = vm
2658 .process_event(
2659 &bytecode,
2660 split_position_event("source_pos", "child_pos"),
2661 "amm::SplitPositionIxState",
2662 None,
2663 None,
2664 )
2665 .unwrap();
2666
2667 assert_eq!(mutations.len(), 2);
2668 assert_eq!(mutations[0].key, json!("source_pos"));
2669 assert_eq!(
2670 mutations[0].patch["activity"]["split_source_count"],
2671 json!(1)
2672 );
2673 assert_eq!(
2674 mutations[0].patch["activity"]["last_split_source_slot"],
2675 json!(77)
2676 );
2677 assert_eq!(
2678 mutations[0].patch["activity"].get("split_child_count"),
2679 None
2680 );
2681 assert_eq!(mutations[1].key, json!("child_pos"));
2682 assert_eq!(
2683 mutations[1].patch["activity"]["split_child_count"],
2684 json!(1)
2685 );
2686 assert_eq!(
2687 mutations[1].patch["activity"].get("split_source_count"),
2688 None
2689 );
2690 }
2691
2692 #[test]
2693 fn handlers_with_identical_keys_still_merge_into_one_segment() {
2694 let mut spec = split_position_spec();
2695 spec.instruction_hooks.clear();
2696 spec.handlers[1].key_resolution = KeyResolutionStrategy::Embedded {
2699 primary_field: FieldPath::new(&["accounts", "first_position"]),
2700 };
2701 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
2702 let handler = &bytecode.entities["Position"].handlers["amm::SplitPositionIxState"];
2703
2704 assert!(!handler
2705 .iter()
2706 .any(|op| matches!(op, super::OpCode::SegmentBoundary)));
2707 assert_eq!(
2708 handler
2709 .iter()
2710 .filter(|op| matches!(op, super::OpCode::ReadOrInitState { .. }))
2711 .count(),
2712 1
2713 );
2714 }
2715
2716 #[test]
2717 fn hook_without_a_matching_key_gets_its_own_segment() {
2718 let mut spec = split_position_spec();
2719 spec.handlers.truncate(1);
2720 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
2721 primary_field: FieldPath::new(&["accounts", "second_position"]),
2722 };
2723 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
2724 "id.position_address",
2725 &["accounts", "second_position"],
2726 PopulationStrategy::SetOnce,
2727 ));
2728 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
2729 let mut vm = VmContext::new();
2730
2731 vm.process_event(
2732 &bytecode,
2733 split_position_event("source_pos", "child_pos"),
2734 "amm::SplitPositionIxState",
2735 None,
2736 None,
2737 )
2738 .unwrap();
2739
2740 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
2742 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
2743 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
2744 assert_eq!(child["activity"].get("last_split_source_slot"), None);
2745 }
2746
2747 fn add_owner_index(spec: &mut TypedStreamSpec<Value>) {
2749 spec.identity.lookup_indexes.push(LookupIndexSpec {
2750 field_name: "id.owner".to_string(),
2751 temporal_field: None,
2752 });
2753 spec.handlers
2754 .push(crate::ast::TypedHandlerSpec::from_serializable(
2755 SerializableHandlerSpec {
2756 source: SourceSpec::Source {
2757 program_id: None,
2758 discriminator: None,
2759 type_name: "amm::PositionState".to_string(),
2760 serialization: None,
2761 is_account: true,
2762 },
2763 key_resolution: KeyResolutionStrategy::Embedded {
2764 primary_field: FieldPath::new(&["__account_address"]),
2765 },
2766 mappings: vec![
2767 mapping(
2768 "id.position_address",
2769 &["__account_address"],
2770 PopulationStrategy::SetOnce,
2771 ),
2772 mapping("id.owner", &["owner"], PopulationStrategy::SetOnce),
2773 ],
2774 conditions: vec![],
2775 emit: true,
2776 },
2777 ));
2778 }
2779
2780 fn index_owner(
2781 vm: &mut VmContext,
2782 bytecode: &MultiEntityBytecode,
2783 position: &str,
2784 owner: &str,
2785 ) {
2786 vm.process_event(
2787 bytecode,
2788 json!({ "__account_address": position, "owner": owner }),
2789 "amm::PositionState",
2790 None,
2791 None,
2792 )
2793 .unwrap();
2794 }
2795
2796 #[test]
2797 fn unmatched_hook_next_to_other_keys_resolves_lookup_index_fields() {
2798 let mut spec = split_position_spec();
2799 spec.handlers.truncate(1);
2800 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
2801 primary_field: FieldPath::new(&["accounts", "second_position"]),
2802 };
2803 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
2804 "id.position_address",
2805 &["accounts", "second_position"],
2806 PopulationStrategy::SetOnce,
2807 ));
2808 add_owner_index(&mut spec);
2809 spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
2811
2812 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
2813 let mut vm = VmContext::new();
2814 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
2815 vm.process_event(
2816 &bytecode,
2817 json!({
2818 "accounts": {
2819 "first_position": "source_pos",
2820 "second_position": "child_pos",
2821 "owner": "owner_1",
2822 },
2823 "data": { "slot": 77 },
2824 }),
2825 "amm::SplitPositionIxState",
2826 None,
2827 None,
2828 )
2829 .unwrap();
2830
2831 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
2832 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
2833 assert!(vm.get_entity_state(0, &json!("owner_1")).is_none());
2834 }
2835
2836 #[test]
2837 fn unmatched_hook_replays_after_its_lookup_index_is_populated() {
2838 let mut spec = split_position_spec();
2839 spec.handlers.truncate(1);
2840 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
2841 primary_field: FieldPath::new(&["accounts", "second_position"]),
2842 };
2843 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
2844 "id.position_address",
2845 &["accounts", "second_position"],
2846 PopulationStrategy::SetOnce,
2847 ));
2848 add_owner_index(&mut spec);
2849 spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
2850
2851 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
2852 let mut vm = VmContext::new();
2853 let mutations = vm
2854 .process_event(
2855 &bytecode,
2856 json!({
2857 "accounts": { "owner": "owner_1", "second_position": "child_pos" },
2858 "data": { "slot": 77 },
2859 }),
2860 "amm::SplitPositionIxState",
2861 None,
2862 None,
2863 )
2864 .unwrap();
2865 assert_eq!(mutations.len(), 1);
2866 assert_eq!(mutations[0].key, json!("child_pos"));
2867
2868 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
2869
2870 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
2871 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
2872 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
2873 assert_eq!(child["activity"].get("last_split_source_slot"), None);
2874 }
2875
2876 #[test]
2877 fn a_segment_that_misses_its_key_is_replayed_alone() {
2878 let mut spec = split_position_spec();
2879 spec.instruction_hooks.clear();
2880 spec.handlers.swap(0, 1);
2883 spec.handlers[1].key_resolution = KeyResolutionStrategy::Lookup {
2884 primary_field: FieldPath::new(&["accounts", "source_pda"]),
2885 };
2886 spec.handlers[1].mappings.remove(0);
2887 add_owner_index(&mut spec);
2888 spec.instruction_hooks.push(InstructionHook {
2889 instruction_type: "amm::RegisterIxState".to_string(),
2890 actions: vec![HookAction::RegisterPdaMapping {
2891 pda_field: FieldPath::new(&["accounts", "source_pda"]),
2892 seed_field: FieldPath::new(&["accounts", "owner"]),
2893 lookup_name: "default_pda_lookup".to_string(),
2894 }],
2895 lookup_by: None,
2896 });
2897 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
2898 let mut vm = VmContext::new();
2899 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
2900
2901 let mutations = vm
2902 .process_event(
2903 &bytecode,
2904 json!({
2905 "accounts": { "source_pda": "pda_1", "second_position": "child_pos" },
2906 "data": {},
2907 }),
2908 "amm::SplitPositionIxState",
2909 None,
2910 None,
2911 )
2912 .unwrap();
2913 assert_eq!(mutations.len(), 1);
2915 assert_eq!(mutations[0].key, json!("child_pos"));
2916
2917 let mutations = vm
2918 .process_event(
2919 &bytecode,
2920 json!({ "accounts": { "source_pda": "pda_1", "owner": "owner_1" } }),
2921 "amm::RegisterIxState",
2922 None,
2923 None,
2924 )
2925 .unwrap();
2926 assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
2928 assert_eq!(mutations[0].key, json!("source_pos"));
2929 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
2930 assert_eq!(source["activity"]["split_source_count"], json!(1));
2931 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
2932 assert_eq!(child["activity"]["split_child_count"], json!(1));
2933 }
2934
2935 #[test]
2940 fn a_hook_whose_lookup_field_is_absent_skips_only_its_segment() {
2941 let spec: TypedStreamSpec<Value> =
2942 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2943 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2944 state_name: "Game".to_string(),
2945 program_id: None,
2946 idl: None,
2947 identity: IdentitySpec {
2948 primary_keys: vec!["id.game_id".to_string()],
2949 lookup_indexes: vec![],
2950 },
2951 handlers: vec![instruction_handler(
2952 "game::StartGameIxState",
2953 KeyResolutionStrategy::Embedded {
2954 primary_field: FieldPath::new(&["data", "game_id"]),
2955 },
2956 vec![
2957 SerializableFieldMapping {
2958 source: MappingSource::FromSource {
2959 path: FieldPath::new(&["data", "game_id"]),
2960 default: None,
2961 transform: Some(crate::ast::Transformation::HexEncode),
2962 },
2963 ..mapping("id.game_id", &[], PopulationStrategy::SetOnce)
2964 },
2965 mapping("stats.fee", &["data", "fee"], PopulationStrategy::SetOnce),
2966 ],
2967 )],
2968 sections: vec![],
2969 field_mappings: BTreeMap::new(),
2970 resolver_hooks: vec![],
2971 instruction_hooks: vec![InstructionHook {
2972 instruction_type: "game::StartGameIxState".to_string(),
2973 actions: vec![HookAction::SetField {
2974 target_field: "stats.started_slot".to_string(),
2975 source: MappingSource::FromSource {
2976 path: FieldPath::new(&["data", "slot"]),
2977 default: None,
2978 transform: None,
2979 },
2980 condition: None,
2981 }],
2982 lookup_by: Some(FieldPath::new(&["accounts", "game_id"])),
2984 }],
2985 resolver_specs: vec![],
2986 computed_fields: vec![],
2987 computed_field_specs: vec![],
2988 content_hash: None,
2989 views: vec![],
2990 });
2991 let bytecode = MultiEntityBytecode::from_single("Game".to_string(), spec, 0);
2992 let mut vm = VmContext::new();
2993
2994 let mutations = vm
2995 .process_event(
2996 &bytecode,
2997 json!({
2998 "accounts": { "game_account": "game_pda" },
2999 "data": { "game_id": [1, 2, 3, 4], "fee": 5, "slot": 9 },
3000 }),
3001 "game::StartGameIxState",
3002 None,
3003 None,
3004 )
3005 .unwrap();
3006
3007 assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
3008 assert_eq!(mutations[0].key, json!("01020304"));
3009 let game = vm.get_entity_state(0, &json!("01020304")).unwrap();
3010 assert_eq!(game["id"]["game_id"], json!("01020304"));
3011 assert_eq!(game["stats"]["fee"], json!(5));
3012 assert_eq!(game["stats"].get("started_slot"), None);
3013 assert_eq!(vm.get_entity_state(0, &json!(null)), None);
3015 }
3016
3017 mod fingerprint {
3018 use super::super::{EntityBytecode, MultiEntityBytecode, OpCode};
3019 use crate::ast::FieldPath;
3020 use serde_json::json;
3021 use std::collections::{HashMap, HashSet};
3022
3023 fn sample_handler() -> Vec<OpCode> {
3024 vec![
3025 OpCode::LoadEventField {
3026 path: FieldPath::new(&["id", "mint"]),
3027 dest: 0,
3028 default: Some(json!({"b": 2, "a": 1})),
3029 },
3030 OpCode::CreateObject { dest: 1 },
3031 OpCode::SetField {
3032 object: 1,
3033 path: "state.volume".to_string(),
3034 value: 0,
3035 },
3036 OpCode::EmitMutation {
3037 entity_name: "PumpfunToken".to_string(),
3038 key: 0,
3039 state: 1,
3040 },
3041 ]
3042 }
3043
3044 fn bytecode(handler_insert_order: &[&str]) -> MultiEntityBytecode {
3045 let mut handlers = HashMap::new();
3046 for event_type in handler_insert_order {
3047 handlers.insert(event_type.to_string(), sample_handler());
3048 }
3049 let mut entities = HashMap::new();
3050 entities.insert(
3051 "PumpfunToken".to_string(),
3052 EntityBytecode {
3053 state_id: 0,
3054 handlers,
3055 entity_name: "PumpfunToken".to_string(),
3056 when_events: HashSet::from(["pump::BuyIxState".to_string()]),
3057 non_emitted_fields: HashSet::new(),
3058 computed_paths: vec!["state.market_cap".to_string()],
3059 computed_fields_evaluator: None,
3060 },
3061 );
3062 let mut event_routing = HashMap::new();
3063 event_routing.insert(
3064 "pump::TokenState".to_string(),
3065 vec!["PumpfunToken".to_string()],
3066 );
3067 MultiEntityBytecode {
3068 entities,
3069 event_routing,
3070 when_events: HashSet::from(["pump::BuyIxState".to_string()]),
3071 proto_router: crate::proto_router::ProtoRouter::new(),
3072 }
3073 }
3074
3075 #[test]
3076 fn stable_across_recomputation_and_insertion_order() {
3077 let a = bytecode(&["pump::TokenState", "pump::BuyIxState"]);
3078 let b = bytecode(&["pump::BuyIxState", "pump::TokenState"]);
3079 assert_eq!(a.fingerprint(), a.fingerprint());
3080 assert_eq!(a.fingerprint(), b.fingerprint());
3081 assert_eq!(a.fingerprint().len(), 64);
3082 }
3083
3084 #[test]
3085 fn changes_when_an_opcode_operand_changes() {
3086 let base = bytecode(&["pump::TokenState"]);
3087 let mut modified = bytecode(&["pump::TokenState"]);
3088 let entity = modified.entities.get_mut("PumpfunToken").unwrap();
3089 match &mut entity.handlers.get_mut("pump::TokenState").unwrap()[2] {
3090 OpCode::SetField { path, .. } => *path = "state.volume_usd".to_string(),
3091 other => panic!("unexpected opcode: {other:?}"),
3092 }
3093 assert_ne!(base.fingerprint(), modified.fingerprint());
3094 }
3095
3096 #[test]
3097 fn changes_when_routing_changes() {
3098 let base = bytecode(&["pump::TokenState"]);
3099 let mut modified = bytecode(&["pump::TokenState"]);
3100 modified.event_routing.insert(
3101 "pump::SellIxState".to_string(),
3102 vec!["PumpfunToken".to_string()],
3103 );
3104 assert_ne!(base.fingerprint(), modified.fingerprint());
3105 }
3106
3107 #[test]
3114 fn golden_hash_pins_the_canonical_encoding() {
3115 let expected = "acf7ea8bf39ca830a9e6c2b2d2fb471da68e57e24bbb4c256964a42a498f75f2";
3116 assert_eq!(bytecode(&["pump::TokenState"]).fingerprint(), expected);
3117 }
3118 }
3119}