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 crate::{Mutation, WholeEntity};
2240 use serde_json::{json, Value};
2241 use std::collections::BTreeMap;
2242
2243 fn mapping(
2244 target_path: &str,
2245 source_path: &[&str],
2246 population: PopulationStrategy,
2247 ) -> SerializableFieldMapping {
2248 SerializableFieldMapping {
2249 target_path: target_path.to_string(),
2250 source: MappingSource::FromSource {
2251 path: FieldPath::new(source_path),
2252 default: None,
2253 transform: None,
2254 },
2255 transform: None,
2256 population,
2257 condition: None,
2258 when: None,
2259 stop: None,
2260 emit: true,
2261 }
2262 }
2263
2264 fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
2265 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2266 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2267 state_name: "PumpfunToken".to_string(),
2268 program_id: None,
2269 idl: None,
2270 identity: IdentitySpec {
2271 primary_keys: vec!["id.mint".to_string()],
2272 lookup_indexes: vec![LookupIndexSpec {
2273 field_name: "id.bonding_curve".to_string(),
2274 temporal_field: None,
2275 }],
2276 },
2277 handlers: vec![SerializableHandlerSpec {
2278 source: SourceSpec::Source {
2279 program_id: None,
2280 discriminator: None,
2281 type_name: "pump::BondingCurveState".to_string(),
2282 serialization: None,
2283 is_account: true,
2284 },
2285 key_resolution: KeyResolutionStrategy::Lookup {
2286 primary_field: FieldPath::new(&["__account_address"]),
2287 },
2288 mappings: vec![
2289 mapping(
2290 "id.bonding_curve",
2291 &["__account_address"],
2292 PopulationStrategy::SetOnce,
2293 ),
2294 mapping(
2295 "reserves.virtual_token_reserves",
2296 &["virtual_token_reserves"],
2297 PopulationStrategy::LastWrite,
2298 ),
2299 ],
2300 conditions: vec![],
2301 emit: true,
2302 }],
2303 sections: vec![],
2304 field_mappings: BTreeMap::new(),
2305 resolver_hooks: vec![],
2306 instruction_hooks: vec![InstructionHook {
2307 instruction_type: "pump::BuyIxState".to_string(),
2308 actions: vec![HookAction::RegisterPdaMapping {
2309 pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2310 seed_field: FieldPath::new(&["accounts", "mint"]),
2311 lookup_name: "default_pda_lookup".to_string(),
2312 }],
2313 lookup_by: None,
2314 }],
2315 resolver_specs: vec![],
2316 computed_fields: vec![],
2317 computed_field_specs: vec![],
2318 content_hash: None,
2319 views: vec![],
2320 })
2321 }
2322
2323 fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
2324 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2325 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2326 state_name: "OreRound".to_string(),
2327 program_id: None,
2328 idl: None,
2329 identity: IdentitySpec {
2330 primary_keys: vec!["id.round_id".to_string()],
2331 lookup_indexes: vec![LookupIndexSpec {
2332 field_name: "id.round_address".to_string(),
2333 temporal_field: None,
2334 }],
2335 },
2336 handlers: vec![SerializableHandlerSpec {
2337 source: SourceSpec::Source {
2338 program_id: None,
2339 discriminator: None,
2340 type_name: "entropy::VarState".to_string(),
2341 serialization: None,
2342 is_account: true,
2343 },
2344 key_resolution: KeyResolutionStrategy::Lookup {
2345 primary_field: FieldPath::new(&["__account_address"]),
2346 },
2347 mappings: vec![mapping(
2348 "state.expires_at",
2349 &["end_at"],
2350 PopulationStrategy::LastWrite,
2351 )],
2352 conditions: vec![],
2353 emit: true,
2354 }],
2355 sections: vec![],
2356 field_mappings: BTreeMap::new(),
2357 resolver_hooks: vec![],
2358 instruction_hooks: vec![InstructionHook {
2359 instruction_type: "ore::DeployIxState".to_string(),
2360 actions: vec![HookAction::RegisterPdaMapping {
2361 pda_field: FieldPath::new(&["accounts", "entropyVar"]),
2362 seed_field: FieldPath::new(&["accounts", "round"]),
2363 lookup_name: "default_pda_lookup".to_string(),
2364 }],
2365 lookup_by: None,
2366 }],
2367 resolver_specs: vec![],
2368 computed_fields: vec![],
2369 computed_field_specs: vec![],
2370 content_hash: None,
2371 views: vec![],
2372 })
2373 }
2374
2375 fn embedded_account_key_spec() -> TypedStreamSpec<Value> {
2376 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2377 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2378 state_name: "OreTreasury".to_string(),
2379 program_id: None,
2380 idl: None,
2381 identity: IdentitySpec {
2382 primary_keys: vec!["id.address".to_string()],
2383 lookup_indexes: vec![],
2384 },
2385 handlers: vec![SerializableHandlerSpec {
2386 source: SourceSpec::Source {
2387 program_id: None,
2388 discriminator: None,
2389 type_name: "ore::TreasuryState".to_string(),
2390 serialization: None,
2391 is_account: true,
2392 },
2393 key_resolution: KeyResolutionStrategy::Embedded {
2394 primary_field: FieldPath::new(&["__account_address"]),
2395 },
2396 mappings: vec![mapping(
2397 "id.address",
2398 &["__account_address"],
2399 PopulationStrategy::SetOnce,
2400 )],
2401 conditions: vec![],
2402 emit: true,
2403 }],
2404 sections: vec![],
2405 field_mappings: BTreeMap::new(),
2406 resolver_hooks: vec![],
2407 instruction_hooks: vec![],
2408 resolver_specs: vec![],
2409 computed_fields: vec![],
2410 computed_field_specs: vec![],
2411 content_hash: None,
2412 views: vec![],
2413 })
2414 }
2415
2416 fn whole_entity_pool_spec() -> TypedStreamSpec<Value> {
2420 let mut hidden = mapping("state.secret", &["secret"], PopulationStrategy::LastWrite);
2421 hidden.emit = false;
2422 let handler = |type_name: &str, mappings, emit| SerializableHandlerSpec {
2423 source: SourceSpec::Source {
2424 program_id: None,
2425 discriminator: None,
2426 type_name: type_name.to_string(),
2427 serialization: None,
2428 is_account: true,
2429 },
2430 key_resolution: KeyResolutionStrategy::Embedded {
2431 primary_field: FieldPath::new(&["__account_address"]),
2432 },
2433 mappings,
2434 conditions: vec![],
2435 emit,
2436 };
2437 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2438 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2439 state_name: "Pool".to_string(),
2440 program_id: None,
2441 idl: None,
2442 identity: IdentitySpec {
2443 primary_keys: vec!["id.address".to_string()],
2444 lookup_indexes: vec![],
2445 },
2446 handlers: vec![
2447 handler(
2448 "amm::PoolState",
2449 vec![
2450 mapping(
2451 "id.address",
2452 &["__account_address"],
2453 PopulationStrategy::SetOnce,
2454 ),
2455 mapping("state.price", &["price"], PopulationStrategy::LastWrite),
2456 mapping("state.fills", &["fill"], PopulationStrategy::Append),
2457 hidden,
2458 ],
2459 true,
2460 ),
2461 handler(
2462 "amm::PoolConfigState",
2463 vec![mapping(
2464 "config.fee",
2465 &["fee"],
2466 PopulationStrategy::LastWrite,
2467 )],
2468 false,
2469 ),
2470 ],
2471 sections: vec![],
2472 field_mappings: BTreeMap::new(),
2473 resolver_hooks: vec![],
2474 instruction_hooks: vec![],
2475 resolver_specs: vec![],
2476 computed_fields: vec![],
2477 computed_field_specs: vec![],
2478 content_hash: None,
2479 views: vec![],
2480 })
2481 }
2482
2483 fn pool_bytecode() -> MultiEntityBytecode {
2484 MultiEntityBytecode::from_single("Pool".to_string(), whole_entity_pool_spec(), 0)
2485 }
2486
2487 fn pool_event(price: u64, fill: u64) -> Value {
2488 pool_event_for("pool_1", price, fill)
2489 }
2490
2491 fn pool_event_for(address: &str, price: u64, fill: u64) -> Value {
2492 json!({"__account_address": address, "price": price, "fill": fill, "secret": "s"})
2493 }
2494
2495 fn linked_vm(
2497 state_config: crate::vm::StateTableConfig,
2498 ) -> (VmContext, crate::vm::WholeEntityRequests) {
2499 let requests = crate::vm::WholeEntityRequests::new(16);
2500 let mut vm = VmContext::new_with_config(state_config);
2501 vm.set_whole_entity_requests(requests.clone());
2502 (vm, requests)
2503 }
2504
2505 fn process(vm: &mut VmContext, bytecode: &MultiEntityBytecode, event: Value) -> Vec<Mutation> {
2506 vm.process_event(bytecode, event, "amm::PoolState", None, None)
2507 .unwrap()
2508 }
2509
2510 #[test]
2511 fn source_deletion_cancels_resync_and_recreates_a_fresh_vm_row() {
2512 let bytecode = pool_bytecode();
2513 let (mut vm, requests) = linked_vm(Default::default());
2514 let first = process(&mut vm, &bytecode, pool_event(1, 10));
2515 let key = &first[0].key;
2516 let target = crate::vm::ResolverTarget {
2517 state_id: 0,
2518 entity_name: "Pool".into(),
2519 primary_key: key.clone(),
2520 extracts: vec![],
2521 };
2522 vm.enqueue_resolver_request(
2523 "ignored".into(),
2524 crate::ast::ResolverType::Token,
2525 json!("mint"),
2526 target.clone(),
2527 );
2528 let old_request = vm.take_resolver_requests().pop().unwrap();
2529 vm.get_state_table_mut(0).unwrap().deferred_when_ops.insert(
2530 ("sig".into(), "instruction".into()),
2531 vec![crate::vm::DeferredWhenOperation {
2532 entity_name: "Pool".into(),
2533 primary_key: key.clone(),
2534 field_path: "old".into(),
2535 field_value: json!(true),
2536 when_instruction: "instruction".into(),
2537 signature: "sig".into(),
2538 slot: 1,
2539 deferred_at: 1,
2540 emit: true,
2541 }],
2542 );
2543 requests.request("Pool", key);
2544 let deletion = vm.delete_entity(&bytecode, "Pool", key).unwrap();
2545 assert!(deletion.is_delete());
2546 assert!(
2547 serde_json::from_value::<Mutation>(serde_json::to_value(&deletion).unwrap())
2548 .unwrap()
2549 .is_delete()
2550 );
2551 assert!(!requests.is_requested("Pool", key));
2552 assert!(vm.get_entity_state(0, key).is_none());
2553 assert!(vm.snapshot_state_table(0).is_empty());
2554 assert!(vm
2555 .get_state_table_mut(0)
2556 .unwrap()
2557 .deferred_when_ops
2558 .is_empty());
2559 assert!(vm.delete_entity(&bytecode, "Unknown", key).is_none());
2560 let mut next = process(&mut vm, &bytecode, pool_event(9, 90));
2561 vm.enqueue_resolver_request(
2562 "ignored".into(),
2563 crate::ast::ResolverType::Token,
2564 json!("mint"),
2565 target,
2566 );
2567 let new_request = vm.take_resolver_requests().pop().unwrap();
2568 assert_ne!(old_request.cache_key, new_request.cache_key);
2569 assert!(vm
2570 .apply_resolver_result(&bytecode, &old_request.cache_key, json!({"stale":true}))
2571 .unwrap()
2572 .is_empty());
2573 assert_eq!(next[0].take_whole_entity_mark(), Some(WholeEntity::Created));
2574 assert_eq!(next[0].patch["state"]["fills"], json!([90]));
2575 }
2576
2577 #[test]
2578 fn deletion_preserves_unrelated_resolver_sharing_and_isolates_recreation() {
2579 let bytecode = pool_bytecode();
2580 let (mut vm, _) = linked_vm(Default::default());
2581 let key = process(&mut vm, &bytecode, pool_event(1, 10))[0]
2582 .key
2583 .clone();
2584 let late_joiner = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20))[0]
2585 .key
2586 .clone();
2587 let target = |primary_key| crate::vm::ResolverTarget {
2588 state_id: 0,
2589 entity_name: "Pool".into(),
2590 primary_key,
2591 extracts: vec![crate::ast::ResolverExtractSpec {
2592 target_path: "resolved".into(),
2593 source_path: Some("value".into()),
2594 transform: None,
2595 }],
2596 };
2597 let enqueue = |vm: &mut VmContext, key| {
2598 vm.enqueue_resolver_request(
2599 "ignored".into(),
2600 crate::ast::ResolverType::Token,
2601 json!("shared-mint"),
2602 target(key),
2603 )
2604 };
2605 enqueue(&mut vm, key.clone());
2606 enqueue(&mut vm, json!("unrelated"));
2607 let old = vm.take_resolver_requests();
2608 assert_eq!(old.len(), 1);
2609 vm.delete_entity(&bytecode, "Pool", &key).unwrap();
2610 enqueue(&mut vm, json!("unrelated"));
2611 enqueue(&mut vm, json!("another-unrelated"));
2612 assert!(
2613 vm.take_resolver_requests().is_empty(),
2614 "unaffected targets reuse the old pending call"
2615 );
2616 enqueue(&mut vm, key.clone());
2617 let fresh = vm.take_resolver_requests();
2618 assert_eq!(fresh.len(), 1);
2619 assert_ne!(fresh[0].cache_key, old[0].cache_key);
2620 vm.delete_entity(&bytecode, "Pool", &late_joiner).unwrap();
2623 process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
2624 enqueue(&mut vm, late_joiner.clone());
2625 let late_joiner_fresh = vm.take_resolver_requests();
2626 assert_eq!(late_joiner_fresh.len(), 1);
2627 assert_ne!(late_joiner_fresh[0].cache_key, old[0].cache_key);
2628 enqueue(&mut vm, json!("unrelated"));
2629 enqueue(&mut vm, json!("another-unrelated"));
2630 enqueue(&mut vm, key.clone());
2631 assert!(
2632 vm.take_resolver_requests().is_empty(),
2633 "unaffected targets reuse pending calls"
2634 );
2635 let old_mutations = vm
2636 .apply_resolver_result(&bytecode, &old[0].cache_key, json!({"value":"old"}))
2637 .unwrap();
2638 assert_eq!(old_mutations.len(), 2);
2639 assert!(old_mutations.iter().all(|mutation| mutation.key != key));
2640 assert!(vm.get_entity_state(0, &key).is_none());
2641 let fresh_mutations = vm
2642 .apply_resolver_result(&bytecode, &fresh[0].cache_key, json!({"value":"fresh"}))
2643 .unwrap();
2644 assert_eq!(fresh_mutations.len(), 1);
2645 assert_eq!(fresh_mutations[0].key, key);
2646 let late_joiner_mutations = vm
2647 .apply_resolver_result(
2648 &bytecode,
2649 &late_joiner_fresh[0].cache_key,
2650 json!({"value":"fresh-late-joiner"}),
2651 )
2652 .unwrap();
2653 assert_eq!(late_joiner_mutations.len(), 1);
2654 assert_eq!(late_joiner_mutations[0].key, late_joiner);
2655 }
2656
2657 #[test]
2658 fn a_requested_entity_is_emitted_whole_once() {
2659 let bytecode = pool_bytecode();
2660 let (mut vm, requests) = linked_vm(Default::default());
2661 let mut update = |price, fill| process(&mut vm, &bytecode, pool_event(price, fill));
2662
2663 let first = update(1, 10);
2664 assert_eq!(first[0].whole_entity_mark(), Some(WholeEntity::Created));
2665 let key = first[0].key.clone();
2666 let second = update(2, 20);
2667 assert_eq!(second.len(), 1);
2668 assert!(!second[0].is_whole_entity());
2669 assert_eq!(
2670 second[0].patch,
2671 json!({"state": {"price": 2, "fills": [20]}})
2672 );
2673 assert_eq!(second[0].append, vec!["state.fills".to_string()]);
2674
2675 requests.request("Pool", &key);
2677 assert!(requests.is_requested("Pool", &key));
2678 let mut third = update(3, 30);
2679 assert_eq!(third.len(), 2);
2680 assert!(
2681 !third[0].is_whole_entity(),
2682 "the change itself is untouched"
2683 );
2684 assert_eq!(
2685 third[0].patch,
2686 json!({"state": {"price": 3, "fills": [30]}})
2687 );
2688 assert_eq!(third[1].key, key);
2689 assert_eq!(third[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
2690 assert_eq!(
2691 third[1].patch,
2692 json!({
2693 "id": {"address": "pool_1"},
2694 "state": {"price": 3, "fills": [10, 20, 30]},
2695 }),
2696 "the whole entity, without its non-emitted field"
2697 );
2698 assert!(
2699 third[1].append.is_empty(),
2700 "its arrays are whole, not appended to"
2701 );
2702 assert!(requests.is_empty(), "a request is answered once");
2703
2704 let fourth = update(4, 40);
2705 assert_eq!(fourth.len(), 1);
2706 assert!(!fourth[0].is_whole_entity());
2707 assert_eq!(
2708 fourth[0].patch,
2709 json!({"state": {"price": 4, "fills": [40]}})
2710 );
2711 }
2712
2713 #[test]
2716 fn a_request_is_answered_by_the_next_call_whatever_it_changes() {
2717 let bytecode = pool_bytecode();
2718 let (mut vm, requests) = linked_vm(Default::default());
2719 process(&mut vm, &bytecode, pool_event_for("pool_1", 1, 10));
2720 process(&mut vm, &bytecode, pool_event_for("pool_2", 1, 10));
2721
2722 requests.request("Pool", &json!("pool_1"));
2723 let mut next = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20));
2724 assert_eq!(next.len(), 2);
2725 assert_eq!(next[0].key, json!("pool_2"));
2726 assert!(!next[0].is_whole_entity());
2727 assert_eq!(next[1].key, json!("pool_1"));
2728 assert_eq!(next[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
2729 assert_eq!(
2730 next[1].patch,
2731 json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
2732 );
2733 assert!(requests.is_empty());
2734
2735 requests.request("Pool", &json!("pool_2"));
2737 let answered = vm
2738 .apply_resolver_result(&bytecode, "no-such-request", json!({}))
2739 .unwrap();
2740 assert_eq!(answered.len(), 1);
2741 assert_eq!(answered[0].key, json!("pool_2"));
2742 assert_eq!(answered[0].whole_entity_mark(), Some(WholeEntity::Resent));
2743
2744 let quiet = process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
2745 assert_eq!(quiet.len(), 1, "a request is answered once");
2746 }
2747
2748 #[test]
2751 fn a_request_for_an_entity_the_vm_does_not_hold_is_dropped() {
2752 let bytecode = pool_bytecode();
2753 let (mut vm, requests) = linked_vm(Default::default());
2754 requests.request("Pool", &json!("other_pool"));
2755 let mutations = process(&mut vm, &bytecode, pool_event(1, 10));
2756 assert_eq!(mutations.len(), 1);
2757 assert_eq!(mutations[0].whole_entity_mark(), Some(WholeEntity::Created));
2758 assert!(requests.is_empty(), "dropped, not kept for later");
2759
2760 let created = process(&mut vm, &bytecode, pool_event_for("other_pool", 1, 10));
2761 assert_eq!(created.len(), 1);
2762 assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
2763 }
2764
2765 #[test]
2766 fn whole_entity_requests_are_bounded() {
2767 let requests = crate::vm::WholeEntityRequests::new(2);
2768 for key in ["a", "b", "c"] {
2769 requests.request("Pool", &json!(key));
2770 }
2771 assert_eq!(requests.len(), 2);
2772 assert!(
2773 !requests.is_requested("Pool", &json!("a")),
2774 "the oldest is forgotten"
2775 );
2776 assert!(requests.is_requested("Pool", &json!("c")));
2777 }
2778
2779 #[test]
2782 fn an_entitys_first_mutation_is_marked_as_its_creation() {
2783 let bytecode = pool_bytecode();
2784 let (mut vm, _requests) = linked_vm(Default::default());
2785
2786 let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
2787 assert_eq!(first.len(), 1, "the creation needs no second mutation");
2788 assert_eq!(
2789 first[0].take_whole_entity_mark(),
2790 Some(WholeEntity::Created)
2791 );
2792 assert_eq!(
2793 first[0].patch,
2794 json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}}),
2795 "the whole row, without its non-emitted field"
2796 );
2797 assert_eq!(first[0].append, vec!["state.fills".to_string()]);
2798
2799 let second = process(&mut vm, &bytecode, pool_event(2, 20));
2800 assert_eq!(second[0].whole_entity_mark(), None);
2801 assert_eq!(
2802 second[0].patch,
2803 json!({"state": {"price": 2, "fills": [20]}})
2804 );
2805 }
2806
2807 #[test]
2810 fn a_vm_without_whole_entity_requests_marks_no_creation() {
2811 let bytecode = pool_bytecode();
2812 let mut vm = VmContext::new();
2813 let first = process(&mut vm, &bytecode, pool_event(1, 10));
2814 assert!(!first[0].is_whole_entity());
2815 assert!(first[0].patch.get(crate::WHOLE_ENTITY_MARKER).is_none());
2816 }
2817
2818 #[test]
2822 fn a_row_started_without_a_mutation_is_created_by_its_first_one() {
2823 let bytecode = pool_bytecode();
2824 let (mut vm, _requests) = linked_vm(Default::default());
2825 let config = vm
2826 .process_event(
2827 &bytecode,
2828 json!({"__account_address": "pool_1", "fee": 30}),
2829 "amm::PoolConfigState",
2830 None,
2831 None,
2832 )
2833 .unwrap();
2834 assert!(config.is_empty(), "the config handler emits nothing");
2835
2836 let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
2837 assert_eq!(
2838 first[0].take_whole_entity_mark(),
2839 Some(WholeEntity::Created)
2840 );
2841 assert_eq!(
2842 first[0].patch,
2843 json!({
2844 "id": {"address": "pool_1"},
2845 "config": {"fee": 30},
2846 "state": {"price": 1, "fills": [10]},
2847 })
2848 );
2849 let second = process(&mut vm, &bytecode, pool_event(2, 20));
2850 assert!(!second[0].is_whole_entity());
2851 }
2852
2853 #[test]
2856 fn an_entity_the_vm_dropped_is_created_again() {
2857 let bytecode = pool_bytecode();
2858 let (mut vm, _requests) = linked_vm(crate::vm::StateTableConfig {
2859 max_entries: 2,
2860 ..Default::default()
2861 });
2862 for pool in ["a", "b", "c"] {
2863 let created = process(&mut vm, &bytecode, pool_event_for(pool, 1, 1));
2864 assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
2865 }
2866 assert!(vm.get_entity_state(0, &json!("a")).is_none(), "dropped");
2867
2868 let mut again = process(&mut vm, &bytecode, pool_event_for("a", 2, 2));
2869 assert_eq!(
2870 again[0].take_whole_entity_mark(),
2871 Some(WholeEntity::Created)
2872 );
2873 assert_eq!(
2874 again[0].patch,
2875 json!({"id": {"address": "a"}, "state": {"price": 2, "fills": [2]}})
2876 );
2877 let changed = process(&mut vm, &bytecode, pool_event_for("c", 2, 2));
2878 assert!(!changed[0].is_whole_entity(), "c was never dropped");
2879 }
2880
2881 #[test]
2886 fn a_restored_row_never_emitted_is_sent_whole_on_request() {
2887 let bytecode = pool_bytecode();
2888 let (mut vm, _requests) = linked_vm(Default::default());
2889 let config = vm
2890 .process_event(
2891 &bytecode,
2892 json!({"__account_address": "pool_1", "fee": 30}),
2893 "amm::PoolConfigState",
2894 None,
2895 None,
2896 )
2897 .unwrap();
2898 assert!(config.is_empty());
2899 let snapshot = vm.dump();
2900
2901 let (mut restored, requests) = linked_vm(Default::default());
2902 restored.hydrate(snapshot);
2903 let first = process(&mut restored, &bytecode, pool_event(1, 10));
2904 assert!(!first[0].is_whole_entity(), "counted as emitted already");
2905 assert_eq!(
2906 first[0].patch,
2907 json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
2908 );
2909
2910 requests.request("Pool", &first[0].key);
2911 let next = process(&mut restored, &bytecode, pool_event_for("pool_2", 1, 1));
2912 let resent = next
2913 .iter()
2914 .find(|mutation| mutation.key == json!("pool_1"))
2915 .expect("the requested entity follows the next call");
2916 assert_eq!(resent.whole_entity_mark(), Some(WholeEntity::Resent));
2917 let mut resent = resent.clone();
2918 resent.take_whole_entity_mark();
2919 assert_eq!(
2920 resent.patch,
2921 json!({
2922 "id": {"address": "pool_1"},
2923 "config": {"fee": 30},
2924 "state": {"price": 1, "fills": [10]},
2925 })
2926 );
2927 }
2928
2929 #[test]
2933 fn a_resolver_result_for_an_entity_not_held_creates_it() {
2934 let bytecode = pool_bytecode();
2935 let (mut vm, _requests) = linked_vm(Default::default());
2936 process(&mut vm, &bytecode, pool_event(1, 10));
2937 for key in ["pool_1", "pool_2"] {
2938 vm.enqueue_resolver_request(
2939 String::new(),
2940 crate::ast::ResolverType::Token,
2941 json!(key),
2942 crate::vm::ResolverTarget {
2943 state_id: 0,
2944 entity_name: "Pool".to_string(),
2945 primary_key: json!(key),
2946 extracts: vec![crate::ast::ResolverExtractSpec {
2947 target_path: "meta.name".to_string(),
2948 source_path: Some("name".to_string()),
2949 transform: None,
2950 }],
2951 },
2952 );
2953 }
2954 let mut applied = Vec::new();
2955 for request in vm.take_resolver_requests() {
2956 applied.extend(
2957 vm.apply_resolver_result(&bytecode, &request.cache_key, json!({"name": "n"}))
2958 .unwrap(),
2959 );
2960 }
2961 applied.sort_by_key(|mutation| mutation.key.to_string());
2962
2963 assert_eq!(applied[0].key, json!("pool_1"));
2964 assert!(!applied[0].is_whole_entity());
2965 assert_eq!(applied[0].patch, json!({"meta": {"name": "n"}}));
2966 assert_eq!(applied[1].key, json!("pool_2"));
2967 assert_eq!(
2968 applied[1].take_whole_entity_mark(),
2969 Some(WholeEntity::Created)
2970 );
2971 assert_eq!(applied[1].patch, json!({"meta": {"name": "n"}}));
2972 }
2973
2974 #[test]
2977 fn a_restored_entity_is_not_created_again() {
2978 let bytecode = pool_bytecode();
2979 let (mut vm, _requests) = linked_vm(Default::default());
2980 process(&mut vm, &bytecode, pool_event(1, 10));
2981 let snapshot = vm.dump();
2982
2983 let (mut restored, _requests) = linked_vm(Default::default());
2984 restored.hydrate(snapshot);
2985 let next = process(&mut restored, &bytecode, pool_event(2, 20));
2986 assert!(!next[0].is_whole_entity());
2987 assert_eq!(next[0].patch, json!({"state": {"price": 2, "fills": [20]}}));
2988 }
2989
2990 #[test]
2991 fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
2992 let bytecode = MultiEntityBytecode::from_single(
2993 "PumpfunToken".to_string(),
2994 direct_pda_to_primary_key_spec(),
2995 0,
2996 );
2997 let mut vm = VmContext::new();
2998
2999 let mutations = vm
3000 .process_event(
3001 &bytecode,
3002 json!({
3003 "__account_address": "bonding_curve_1",
3004 "__resolved_primary_key": "mint_1",
3005 "virtual_token_reserves": 42,
3006 }),
3007 "pump::BondingCurveState",
3008 None,
3009 None,
3010 )
3011 .unwrap();
3012
3013 assert_eq!(mutations.len(), 1);
3014 assert_eq!(mutations[0].key, json!("mint_1"));
3015 assert_eq!(
3016 mutations[0].patch["id"]["bonding_curve"],
3017 json!("bonding_curve_1")
3018 );
3019 assert_eq!(
3020 mutations[0].patch["reserves"]["virtual_token_reserves"],
3021 json!(42)
3022 );
3023 }
3024
3025 #[test]
3026 fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
3027 let bytecode = MultiEntityBytecode::from_single(
3028 "OreRound".to_string(),
3029 pda_to_intermediate_lookup_spec(),
3030 0,
3031 );
3032 let mut vm = VmContext::new();
3033
3034 let mutations = vm
3035 .process_event(
3036 &bytecode,
3037 json!({
3038 "__account_address": "entropy_var_1",
3039 "__resolved_primary_key": "round_address_1",
3040 "end_at": 123,
3041 }),
3042 "entropy::VarState",
3043 None,
3044 None,
3045 )
3046 .unwrap();
3047
3048 assert!(mutations.is_empty());
3049 }
3050
3051 #[test]
3052 fn embedded_account_key_wins_over_an_unrelated_resolved_key() {
3053 let bytecode = MultiEntityBytecode::from_single(
3054 "OreTreasury".to_string(),
3055 embedded_account_key_spec(),
3056 0,
3057 );
3058 let mut vm = VmContext::new();
3059
3060 let mutations = vm
3061 .process_event(
3062 &bytecode,
3063 json!({
3064 "__account_address": "treasury_pda",
3065 "__resolved_primary_key": "round_address",
3066 }),
3067 "ore::TreasuryState",
3068 None,
3069 None,
3070 )
3071 .unwrap();
3072
3073 assert_eq!(mutations.len(), 1);
3074 assert_eq!(mutations[0].key, json!("treasury_pda"));
3075 assert_eq!(mutations[0].patch["id"]["address"], json!("treasury_pda"));
3076 }
3077
3078 fn instruction_handler(
3079 type_name: &str,
3080 key_resolution: KeyResolutionStrategy,
3081 mappings: Vec<SerializableFieldMapping>,
3082 ) -> SerializableHandlerSpec {
3083 SerializableHandlerSpec {
3084 source: SourceSpec::Source {
3085 program_id: None,
3086 discriminator: None,
3087 type_name: type_name.to_string(),
3088 serialization: None,
3089 is_account: false,
3090 },
3091 key_resolution,
3092 mappings,
3093 conditions: vec![],
3094 emit: true,
3095 }
3096 }
3097
3098 fn split_position_spec() -> TypedStreamSpec<Value> {
3103 let embedded = |field: &str| KeyResolutionStrategy::Embedded {
3104 primary_field: FieldPath::new(&["accounts", field]),
3105 };
3106 TypedStreamSpec::from_serializable(SerializableStreamSpec {
3107 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3108 state_name: "Position".to_string(),
3109 program_id: None,
3110 idl: None,
3111 identity: IdentitySpec {
3112 primary_keys: vec!["id.position_address".to_string()],
3113 lookup_indexes: vec![],
3114 },
3115 handlers: vec![
3116 instruction_handler(
3117 "amm::SplitPositionIxState",
3118 embedded("first_position"),
3119 vec![
3120 mapping(
3121 "id.position_address",
3122 &["accounts", "first_position"],
3123 PopulationStrategy::SetOnce,
3124 ),
3125 mapping(
3126 "activity.split_source_count",
3127 &["data"],
3128 PopulationStrategy::Count,
3129 ),
3130 ],
3131 ),
3132 instruction_handler(
3133 "amm::SplitPositionIxState",
3134 embedded("second_position"),
3135 vec![
3136 mapping(
3137 "id.position_address",
3138 &["accounts", "second_position"],
3139 PopulationStrategy::SetOnce,
3140 ),
3141 mapping(
3142 "activity.split_child_count",
3143 &["data"],
3144 PopulationStrategy::Count,
3145 ),
3146 ],
3147 ),
3148 ],
3149 sections: vec![],
3150 field_mappings: BTreeMap::new(),
3151 resolver_hooks: vec![],
3152 instruction_hooks: vec![InstructionHook {
3153 instruction_type: "amm::SplitPositionIxState".to_string(),
3154 actions: vec![HookAction::SetField {
3155 target_field: "activity.last_split_source_slot".to_string(),
3156 source: MappingSource::FromSource {
3157 path: FieldPath::new(&["data", "slot"]),
3158 default: None,
3159 transform: None,
3160 },
3161 condition: None,
3162 }],
3163 lookup_by: Some(FieldPath::new(&["accounts", "first_position"])),
3164 }],
3165 resolver_specs: vec![],
3166 computed_fields: vec![],
3167 computed_field_specs: vec![],
3168 content_hash: None,
3169 views: vec![],
3170 })
3171 }
3172
3173 fn split_position_event(first: &str, second: &str) -> Value {
3174 json!({
3175 "accounts": { "first_position": first, "second_position": second },
3176 "data": { "slot": 77 },
3177 })
3178 }
3179
3180 #[test]
3181 fn one_instruction_routes_each_key_to_its_own_entity() {
3182 let bytecode =
3183 MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
3184 let mut vm = VmContext::new();
3185
3186 for _ in 0..2 {
3187 vm.process_event(
3188 &bytecode,
3189 split_position_event("source_pos", "child_pos"),
3190 "amm::SplitPositionIxState",
3191 None,
3192 None,
3193 )
3194 .unwrap();
3195 }
3196 vm.process_event(
3197 &bytecode,
3198 split_position_event("child_pos", "grandchild_pos"),
3199 "amm::SplitPositionIxState",
3200 None,
3201 None,
3202 )
3203 .unwrap();
3204
3205 let state = |key: &str| vm.get_entity_state(0, &json!(key)).unwrap();
3206
3207 let source = state("source_pos");
3208 assert_eq!(source["id"]["position_address"], json!("source_pos"));
3209 assert_eq!(source["activity"]["split_source_count"], json!(2));
3210 assert_eq!(source["activity"].get("split_child_count"), None);
3211 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3212
3213 let child = state("child_pos");
3215 assert_eq!(child["id"]["position_address"], json!("child_pos"));
3216 assert_eq!(child["activity"]["split_child_count"], json!(2));
3217 assert_eq!(child["activity"]["split_source_count"], json!(1));
3218 assert_eq!(child["activity"]["last_split_source_slot"], json!(77));
3219
3220 let grandchild = state("grandchild_pos");
3221 assert_eq!(grandchild["activity"]["split_child_count"], json!(1));
3222 assert_eq!(grandchild["activity"].get("split_source_count"), None);
3223 assert_eq!(grandchild["activity"].get("last_split_source_slot"), None);
3224 }
3225
3226 #[test]
3227 fn one_instruction_emits_one_mutation_per_key() {
3228 let bytecode =
3229 MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
3230 let mut vm = VmContext::new();
3231
3232 let mutations = vm
3233 .process_event(
3234 &bytecode,
3235 split_position_event("source_pos", "child_pos"),
3236 "amm::SplitPositionIxState",
3237 None,
3238 None,
3239 )
3240 .unwrap();
3241
3242 assert_eq!(mutations.len(), 2);
3243 assert_eq!(mutations[0].key, json!("source_pos"));
3244 assert_eq!(
3245 mutations[0].patch["activity"]["split_source_count"],
3246 json!(1)
3247 );
3248 assert_eq!(
3249 mutations[0].patch["activity"]["last_split_source_slot"],
3250 json!(77)
3251 );
3252 assert_eq!(
3253 mutations[0].patch["activity"].get("split_child_count"),
3254 None
3255 );
3256 assert_eq!(mutations[1].key, json!("child_pos"));
3257 assert_eq!(
3258 mutations[1].patch["activity"]["split_child_count"],
3259 json!(1)
3260 );
3261 assert_eq!(
3262 mutations[1].patch["activity"].get("split_source_count"),
3263 None
3264 );
3265 }
3266
3267 #[test]
3268 fn handlers_with_identical_keys_still_merge_into_one_segment() {
3269 let mut spec = split_position_spec();
3270 spec.instruction_hooks.clear();
3271 spec.handlers[1].key_resolution = KeyResolutionStrategy::Embedded {
3274 primary_field: FieldPath::new(&["accounts", "first_position"]),
3275 };
3276 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3277 let handler = &bytecode.entities["Position"].handlers["amm::SplitPositionIxState"];
3278
3279 assert!(!handler
3280 .iter()
3281 .any(|op| matches!(op, super::OpCode::SegmentBoundary)));
3282 assert_eq!(
3283 handler
3284 .iter()
3285 .filter(|op| matches!(op, super::OpCode::ReadOrInitState { .. }))
3286 .count(),
3287 1
3288 );
3289 }
3290
3291 #[test]
3292 fn hook_without_a_matching_key_gets_its_own_segment() {
3293 let mut spec = split_position_spec();
3294 spec.handlers.truncate(1);
3295 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
3296 primary_field: FieldPath::new(&["accounts", "second_position"]),
3297 };
3298 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
3299 "id.position_address",
3300 &["accounts", "second_position"],
3301 PopulationStrategy::SetOnce,
3302 ));
3303 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3304 let mut vm = VmContext::new();
3305
3306 vm.process_event(
3307 &bytecode,
3308 split_position_event("source_pos", "child_pos"),
3309 "amm::SplitPositionIxState",
3310 None,
3311 None,
3312 )
3313 .unwrap();
3314
3315 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3317 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3318 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
3319 assert_eq!(child["activity"].get("last_split_source_slot"), None);
3320 }
3321
3322 fn add_owner_index(spec: &mut TypedStreamSpec<Value>) {
3324 spec.identity.lookup_indexes.push(LookupIndexSpec {
3325 field_name: "id.owner".to_string(),
3326 temporal_field: None,
3327 });
3328 spec.handlers
3329 .push(crate::ast::TypedHandlerSpec::from_serializable(
3330 SerializableHandlerSpec {
3331 source: SourceSpec::Source {
3332 program_id: None,
3333 discriminator: None,
3334 type_name: "amm::PositionState".to_string(),
3335 serialization: None,
3336 is_account: true,
3337 },
3338 key_resolution: KeyResolutionStrategy::Embedded {
3339 primary_field: FieldPath::new(&["__account_address"]),
3340 },
3341 mappings: vec![
3342 mapping(
3343 "id.position_address",
3344 &["__account_address"],
3345 PopulationStrategy::SetOnce,
3346 ),
3347 mapping("id.owner", &["owner"], PopulationStrategy::SetOnce),
3348 ],
3349 conditions: vec![],
3350 emit: true,
3351 },
3352 ));
3353 }
3354
3355 fn index_owner(
3356 vm: &mut VmContext,
3357 bytecode: &MultiEntityBytecode,
3358 position: &str,
3359 owner: &str,
3360 ) {
3361 vm.process_event(
3362 bytecode,
3363 json!({ "__account_address": position, "owner": owner }),
3364 "amm::PositionState",
3365 None,
3366 None,
3367 )
3368 .unwrap();
3369 }
3370
3371 #[test]
3372 fn unmatched_hook_next_to_other_keys_resolves_lookup_index_fields() {
3373 let mut spec = split_position_spec();
3374 spec.handlers.truncate(1);
3375 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
3376 primary_field: FieldPath::new(&["accounts", "second_position"]),
3377 };
3378 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
3379 "id.position_address",
3380 &["accounts", "second_position"],
3381 PopulationStrategy::SetOnce,
3382 ));
3383 add_owner_index(&mut spec);
3384 spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
3386
3387 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3388 let mut vm = VmContext::new();
3389 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
3390 vm.process_event(
3391 &bytecode,
3392 json!({
3393 "accounts": {
3394 "first_position": "source_pos",
3395 "second_position": "child_pos",
3396 "owner": "owner_1",
3397 },
3398 "data": { "slot": 77 },
3399 }),
3400 "amm::SplitPositionIxState",
3401 None,
3402 None,
3403 )
3404 .unwrap();
3405
3406 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3407 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3408 assert!(vm.get_entity_state(0, &json!("owner_1")).is_none());
3409 }
3410
3411 #[test]
3412 fn unmatched_hook_replays_after_its_lookup_index_is_populated() {
3413 let mut spec = split_position_spec();
3414 spec.handlers.truncate(1);
3415 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
3416 primary_field: FieldPath::new(&["accounts", "second_position"]),
3417 };
3418 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
3419 "id.position_address",
3420 &["accounts", "second_position"],
3421 PopulationStrategy::SetOnce,
3422 ));
3423 add_owner_index(&mut spec);
3424 spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
3425
3426 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3427 let mut vm = VmContext::new();
3428 let mutations = vm
3429 .process_event(
3430 &bytecode,
3431 json!({
3432 "accounts": { "owner": "owner_1", "second_position": "child_pos" },
3433 "data": { "slot": 77 },
3434 }),
3435 "amm::SplitPositionIxState",
3436 None,
3437 None,
3438 )
3439 .unwrap();
3440 assert_eq!(mutations.len(), 1);
3441 assert_eq!(mutations[0].key, json!("child_pos"));
3442
3443 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
3444
3445 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3446 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3447 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
3448 assert_eq!(child["activity"].get("last_split_source_slot"), None);
3449 }
3450
3451 #[test]
3452 fn a_segment_that_misses_its_key_is_replayed_alone() {
3453 let mut spec = split_position_spec();
3454 spec.instruction_hooks.clear();
3455 spec.handlers.swap(0, 1);
3458 spec.handlers[1].key_resolution = KeyResolutionStrategy::Lookup {
3459 primary_field: FieldPath::new(&["accounts", "source_pda"]),
3460 };
3461 spec.handlers[1].mappings.remove(0);
3462 add_owner_index(&mut spec);
3463 spec.instruction_hooks.push(InstructionHook {
3464 instruction_type: "amm::RegisterIxState".to_string(),
3465 actions: vec![HookAction::RegisterPdaMapping {
3466 pda_field: FieldPath::new(&["accounts", "source_pda"]),
3467 seed_field: FieldPath::new(&["accounts", "owner"]),
3468 lookup_name: "default_pda_lookup".to_string(),
3469 }],
3470 lookup_by: None,
3471 });
3472 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3473 let mut vm = VmContext::new();
3474 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
3475
3476 let mutations = vm
3477 .process_event(
3478 &bytecode,
3479 json!({
3480 "accounts": { "source_pda": "pda_1", "second_position": "child_pos" },
3481 "data": {},
3482 }),
3483 "amm::SplitPositionIxState",
3484 None,
3485 None,
3486 )
3487 .unwrap();
3488 assert_eq!(mutations.len(), 1);
3490 assert_eq!(mutations[0].key, json!("child_pos"));
3491
3492 let mutations = vm
3493 .process_event(
3494 &bytecode,
3495 json!({ "accounts": { "source_pda": "pda_1", "owner": "owner_1" } }),
3496 "amm::RegisterIxState",
3497 None,
3498 None,
3499 )
3500 .unwrap();
3501 assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
3503 assert_eq!(mutations[0].key, json!("source_pos"));
3504 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3505 assert_eq!(source["activity"]["split_source_count"], json!(1));
3506 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
3507 assert_eq!(child["activity"]["split_child_count"], json!(1));
3508 }
3509
3510 #[test]
3515 fn a_hook_whose_lookup_field_is_absent_skips_only_its_segment() {
3516 let spec: TypedStreamSpec<Value> =
3517 TypedStreamSpec::from_serializable(SerializableStreamSpec {
3518 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3519 state_name: "Game".to_string(),
3520 program_id: None,
3521 idl: None,
3522 identity: IdentitySpec {
3523 primary_keys: vec!["id.game_id".to_string()],
3524 lookup_indexes: vec![],
3525 },
3526 handlers: vec![instruction_handler(
3527 "game::StartGameIxState",
3528 KeyResolutionStrategy::Embedded {
3529 primary_field: FieldPath::new(&["data", "game_id"]),
3530 },
3531 vec![
3532 SerializableFieldMapping {
3533 source: MappingSource::FromSource {
3534 path: FieldPath::new(&["data", "game_id"]),
3535 default: None,
3536 transform: Some(crate::ast::Transformation::HexEncode),
3537 },
3538 ..mapping("id.game_id", &[], PopulationStrategy::SetOnce)
3539 },
3540 mapping("stats.fee", &["data", "fee"], PopulationStrategy::SetOnce),
3541 ],
3542 )],
3543 sections: vec![],
3544 field_mappings: BTreeMap::new(),
3545 resolver_hooks: vec![],
3546 instruction_hooks: vec![InstructionHook {
3547 instruction_type: "game::StartGameIxState".to_string(),
3548 actions: vec![HookAction::SetField {
3549 target_field: "stats.started_slot".to_string(),
3550 source: MappingSource::FromSource {
3551 path: FieldPath::new(&["data", "slot"]),
3552 default: None,
3553 transform: None,
3554 },
3555 condition: None,
3556 }],
3557 lookup_by: Some(FieldPath::new(&["accounts", "game_id"])),
3559 }],
3560 resolver_specs: vec![],
3561 computed_fields: vec![],
3562 computed_field_specs: vec![],
3563 content_hash: None,
3564 views: vec![],
3565 });
3566 let bytecode = MultiEntityBytecode::from_single("Game".to_string(), spec, 0);
3567 let mut vm = VmContext::new();
3568
3569 let mutations = vm
3570 .process_event(
3571 &bytecode,
3572 json!({
3573 "accounts": { "game_account": "game_pda" },
3574 "data": { "game_id": [1, 2, 3, 4], "fee": 5, "slot": 9 },
3575 }),
3576 "game::StartGameIxState",
3577 None,
3578 None,
3579 )
3580 .unwrap();
3581
3582 assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
3583 assert_eq!(mutations[0].key, json!("01020304"));
3584 let game = vm.get_entity_state(0, &json!("01020304")).unwrap();
3585 assert_eq!(game["id"]["game_id"], json!("01020304"));
3586 assert_eq!(game["stats"]["fee"], json!(5));
3587 assert_eq!(game["stats"].get("started_slot"), None);
3588 assert_eq!(vm.get_entity_state(0, &json!(null)), None);
3590 }
3591
3592 mod fingerprint {
3593 use super::super::{EntityBytecode, MultiEntityBytecode, OpCode};
3594 use crate::ast::FieldPath;
3595 use serde_json::json;
3596 use std::collections::{HashMap, HashSet};
3597
3598 fn sample_handler() -> Vec<OpCode> {
3599 vec![
3600 OpCode::LoadEventField {
3601 path: FieldPath::new(&["id", "mint"]),
3602 dest: 0,
3603 default: Some(json!({"b": 2, "a": 1})),
3604 },
3605 OpCode::CreateObject { dest: 1 },
3606 OpCode::SetField {
3607 object: 1,
3608 path: "state.volume".to_string(),
3609 value: 0,
3610 },
3611 OpCode::EmitMutation {
3612 entity_name: "PumpfunToken".to_string(),
3613 key: 0,
3614 state: 1,
3615 },
3616 ]
3617 }
3618
3619 fn bytecode(handler_insert_order: &[&str]) -> MultiEntityBytecode {
3620 let mut handlers = HashMap::new();
3621 for event_type in handler_insert_order {
3622 handlers.insert(event_type.to_string(), sample_handler());
3623 }
3624 let mut entities = HashMap::new();
3625 entities.insert(
3626 "PumpfunToken".to_string(),
3627 EntityBytecode {
3628 state_id: 0,
3629 handlers,
3630 entity_name: "PumpfunToken".to_string(),
3631 when_events: HashSet::from(["pump::BuyIxState".to_string()]),
3632 non_emitted_fields: HashSet::new(),
3633 computed_paths: vec!["state.market_cap".to_string()],
3634 computed_fields_evaluator: None,
3635 },
3636 );
3637 let mut event_routing = HashMap::new();
3638 event_routing.insert(
3639 "pump::TokenState".to_string(),
3640 vec!["PumpfunToken".to_string()],
3641 );
3642 MultiEntityBytecode {
3643 entities,
3644 event_routing,
3645 when_events: HashSet::from(["pump::BuyIxState".to_string()]),
3646 proto_router: crate::proto_router::ProtoRouter::new(),
3647 }
3648 }
3649
3650 #[test]
3651 fn stable_across_recomputation_and_insertion_order() {
3652 let a = bytecode(&["pump::TokenState", "pump::BuyIxState"]);
3653 let b = bytecode(&["pump::BuyIxState", "pump::TokenState"]);
3654 assert_eq!(a.fingerprint(), a.fingerprint());
3655 assert_eq!(a.fingerprint(), b.fingerprint());
3656 assert_eq!(a.fingerprint().len(), 64);
3657 }
3658
3659 #[test]
3660 fn changes_when_an_opcode_operand_changes() {
3661 let base = bytecode(&["pump::TokenState"]);
3662 let mut modified = bytecode(&["pump::TokenState"]);
3663 let entity = modified.entities.get_mut("PumpfunToken").unwrap();
3664 match &mut entity.handlers.get_mut("pump::TokenState").unwrap()[2] {
3665 OpCode::SetField { path, .. } => *path = "state.volume_usd".to_string(),
3666 other => panic!("unexpected opcode: {other:?}"),
3667 }
3668 assert_ne!(base.fingerprint(), modified.fingerprint());
3669 }
3670
3671 #[test]
3672 fn changes_when_routing_changes() {
3673 let base = bytecode(&["pump::TokenState"]);
3674 let mut modified = bytecode(&["pump::TokenState"]);
3675 modified.event_routing.insert(
3676 "pump::SellIxState".to_string(),
3677 vec!["PumpfunToken".to_string()],
3678 );
3679 assert_ne!(base.fingerprint(), modified.fingerprint());
3680 }
3681
3682 #[test]
3689 fn golden_hash_pins_the_canonical_encoding() {
3690 let expected = "acf7ea8bf39ca830a9e6c2b2d2fb471da68e57e24bbb4c256964a42a498f75f2";
3691 assert_eq!(bytecode(&["pump::TokenState"]).fingerprint(), expected);
3692 }
3693 }
3694}