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 EvaluateEventExpr {
110 expr: ComputedExpr,
111 transform: Option<Transformation>,
112 dest: Register,
113 },
114 LoadEventField {
115 path: FieldPath,
116 dest: Register,
117 default: Option<Value>,
118 },
119 LoadConstant {
120 value: Value,
121 dest: Register,
122 },
123 CopyRegister {
124 source: Register,
125 dest: Register,
126 },
127 CopyRegisterIfNull {
129 source: Register,
130 dest: Register,
131 },
132 GetEventType {
133 dest: Register,
134 },
135 CreateObject {
136 dest: Register,
137 },
138 SetField {
139 object: Register,
140 path: String,
141 value: Register,
142 },
143 SetFields {
144 object: Register,
145 fields: Vec<(String, Register)>,
146 },
147 GetField {
148 object: Register,
149 path: String,
150 dest: Register,
151 },
152 ReadOrInitState {
153 state_id: u32,
154 key: Register,
155 default: Value,
156 dest: Register,
157 },
158 UpdateState {
159 state_id: u32,
160 key: Register,
161 value: Register,
162 },
163 AppendToArray {
164 object: Register,
165 path: String,
166 value: Register,
167 },
168 GetCurrentTimestamp {
172 dest: Register,
173 },
174 CreateEvent {
177 dest: Register,
178 event_value: Register,
179 },
180 CreateCapture {
183 dest: Register,
184 capture_value: Register,
185 },
186 Transform {
187 source: Register,
188 dest: Register,
189 transformation: Transformation,
190 },
191 EmitMutation {
192 entity_name: String,
193 key: Register,
194 state: Register,
195 },
196 SetFieldIfNull {
197 object: Register,
198 path: String,
199 value: Register,
200 },
201 SetFieldMax {
202 object: Register,
203 path: String,
204 value: Register,
205 },
206 UpdateTemporalIndex {
207 state_id: u32,
208 index_name: String,
209 lookup_value: Register,
210 primary_key: Register,
211 timestamp: Register,
212 },
213 LookupTemporalIndex {
214 state_id: u32,
215 index_name: String,
216 lookup_value: Register,
217 timestamp: Register,
218 dest: Register,
219 },
220 UpdateLookupIndex {
221 state_id: u32,
222 index_name: String,
223 lookup_value: Register,
224 primary_key: Register,
225 },
226 LookupIndex {
227 state_id: u32,
228 index_name: String,
229 lookup_value: Register,
230 dest: Register,
231 },
232 SetFieldSum {
234 object: Register,
235 path: String,
236 value: Register,
237 },
238 SetFieldIncrement {
240 object: Register,
241 path: String,
242 },
243 SetFieldMin {
245 object: Register,
246 path: String,
247 value: Register,
248 },
249 SetFieldWhen {
252 object: Register,
253 path: String,
254 value: Register,
255 when_instruction: String,
256 entity_name: String,
257 key_reg: Register,
258 condition_field: Option<FieldPath>,
259 condition_op: Option<ComparisonOp>,
260 condition_value: Option<Value>,
261 },
262 SetFieldUnlessStopped {
265 object: Register,
266 path: String,
267 value: Register,
268 stop_field: String,
269 stop_instruction: String,
270 entity_name: String,
271 key_reg: Register,
272 },
273 AddToUniqueSet {
276 state_id: u32,
277 set_name: String,
278 value: Register,
279 count_object: Register,
280 count_path: String,
281 },
282 ConditionalSetField {
284 object: Register,
285 path: String,
286 value: Register,
287 condition_field: FieldPath,
288 condition_op: ComparisonOp,
289 condition_value: Value,
290 },
291 ConditionalIncrement {
293 object: Register,
294 path: String,
295 condition_field: FieldPath,
296 condition_op: ComparisonOp,
297 condition_value: Value,
298 },
299 EvaluateComputedFields {
302 state: Register,
303 computed_paths: Vec<String>,
304 },
305 QueueResolver {
307 state_id: u32,
308 entity_name: String,
309 resolver: ResolverType,
310 input_path: Option<String>,
311 input_value: Option<Value>,
312 url_template: Option<Vec<UrlTemplatePart>>,
313 strategy: ResolveStrategy,
314 extracts: Vec<ResolverExtractSpec>,
315 condition: Option<ResolverCondition>,
316 schedule_at: Option<String>,
317 state: Register,
318 key: Register,
319 },
320 UpdatePdaReverseLookup {
323 state_id: u32,
324 lookup_name: String,
325 pda_address: Register,
326 primary_key: Register,
327 },
328 SegmentBoundary,
337}
338
339pub struct EntityBytecode {
340 pub state_id: u32,
341 pub handlers: HashMap<String, Vec<OpCode>>,
342 pub entity_name: String,
343 pub when_events: HashSet<String>,
344 pub non_emitted_fields: HashSet<String>,
345 pub computed_paths: Vec<String>,
346 #[allow(clippy::type_complexity)]
349 pub computed_fields_evaluator: Option<
350 Box<
351 dyn Fn(
352 &mut Value,
353 Option<u64>,
354 i64,
355 )
356 -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
357 + Send
358 + Sync,
359 >,
360 >,
361}
362
363impl std::fmt::Debug for EntityBytecode {
364 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
365 f.debug_struct("EntityBytecode")
366 .field("state_id", &self.state_id)
367 .field("handlers", &self.handlers)
368 .field("entity_name", &self.entity_name)
369 .field("when_events", &self.when_events)
370 .field("non_emitted_fields", &self.non_emitted_fields)
371 .field("computed_paths", &self.computed_paths)
372 .field(
373 "computed_fields_evaluator",
374 &self.computed_fields_evaluator.is_some(),
375 )
376 .finish()
377 }
378}
379
380#[derive(Debug)]
381pub struct MultiEntityBytecode {
382 pub entities: HashMap<String, EntityBytecode>,
383 pub event_routing: HashMap<String, Vec<String>>,
384 pub when_events: HashSet<String>,
385 pub proto_router: crate::proto_router::ProtoRouter,
386}
387
388impl MultiEntityBytecode {
389 pub fn from_single<S>(entity_name: String, spec: TypedStreamSpec<S>, state_id: u32) -> Self {
390 let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
391 let entity_bytecode = compiler.compile_entity();
392
393 let mut entities = HashMap::new();
394 let mut event_routing = HashMap::new();
395 let mut when_events = HashSet::new();
396
397 for event_type in entity_bytecode.handlers.keys() {
398 event_routing
399 .entry(event_type.clone())
400 .or_insert_with(Vec::new)
401 .push(entity_name.clone());
402 }
403
404 when_events.extend(entity_bytecode.when_events.iter().cloned());
405
406 entities.insert(entity_name, entity_bytecode);
407
408 MultiEntityBytecode {
409 entities,
410 event_routing,
411 when_events,
412 proto_router: crate::proto_router::ProtoRouter::new(),
413 }
414 }
415
416 pub fn from_entities(entities_vec: Vec<(String, Box<dyn std::any::Any>, u32)>) -> Self {
417 let entities = HashMap::new();
418 let event_routing = HashMap::new();
419 let when_events = HashSet::new();
420
421 if let Some((_entity_name, _spec_any, _state_id)) = entities_vec.into_iter().next() {
422 panic!("from_entities requires type information - use builder pattern instead");
423 }
424
425 MultiEntityBytecode {
426 entities,
427 event_routing,
428 when_events,
429 proto_router: crate::proto_router::ProtoRouter::new(),
430 }
431 }
432
433 #[allow(clippy::new_ret_no_self)]
434 pub fn new() -> MultiEntityBytecodeBuilder {
435 MultiEntityBytecodeBuilder {
436 entities: HashMap::new(),
437 event_routing: HashMap::new(),
438 when_events: HashSet::new(),
439 proto_router: crate::proto_router::ProtoRouter::new(),
440 }
441 }
442
443 pub fn fingerprint(&self) -> String {
460 use sha2::{Digest, Sha256};
461
462 fn update_sorted_set(hasher: &mut Sha256, set: &HashSet<String>) {
463 let mut items: Vec<&String> = set.iter().collect();
464 items.sort();
465 for item in items {
466 hasher.update(item.as_bytes());
467 hasher.update([0u8]);
468 }
469 hasher.update([1u8]);
470 }
471
472 fn update_canonical_json(hasher: &mut Sha256, value: &Value) {
479 match value {
480 Value::Object(map) => {
481 let mut keys: Vec<&String> = map.keys().collect();
482 keys.sort();
483 hasher.update(*b"{");
484 for key in keys {
485 hasher.update(key.as_bytes());
486 hasher.update([0u8]);
487 update_canonical_json(hasher, &map[key]);
488 }
489 hasher.update(*b"}");
490 }
491 Value::Array(items) => {
492 hasher.update(*b"[");
493 for item in items {
494 update_canonical_json(hasher, item);
495 hasher.update([0u8]);
496 }
497 hasher.update(*b"]");
498 }
499 scalar => {
502 hasher.update(scalar.to_string().as_bytes());
503 hasher.update([0u8]);
504 }
505 }
506 }
507
508 let mut hasher = Sha256::new();
509
510 let mut entity_names: Vec<&String> = self.entities.keys().collect();
511 entity_names.sort();
512 for entity_name in entity_names {
513 let entity = &self.entities[entity_name];
514 hasher.update(entity_name.as_bytes());
515 hasher.update([0u8]);
516 hasher.update(entity.state_id.to_le_bytes());
517
518 let mut event_types: Vec<&String> = entity.handlers.keys().collect();
519 event_types.sort();
520 for event_type in event_types {
521 hasher.update(event_type.as_bytes());
522 hasher.update([0u8]);
523 let ops = serde_json::to_value(&entity.handlers[event_type]).expect(
524 "opcode stream must serialize for fingerprinting; \
525 a non-serializable OpCode payload breaks snapshot invalidation",
526 );
527 update_canonical_json(&mut hasher, &ops);
528 hasher.update([0u8]);
529 }
530 hasher.update([1u8]);
531
532 update_sorted_set(&mut hasher, &entity.when_events);
533 update_sorted_set(&mut hasher, &entity.non_emitted_fields);
534 for path in &entity.computed_paths {
535 hasher.update(path.as_bytes());
536 hasher.update([0u8]);
537 }
538 hasher.update([entity.computed_fields_evaluator.is_some() as u8]);
539 }
540
541 let mut routes: Vec<(&String, &Vec<String>)> = self.event_routing.iter().collect();
542 routes.sort_by_key(|(event_type, _)| *event_type);
543 for (event_type, entity_names) in routes {
544 hasher.update(event_type.as_bytes());
545 hasher.update([0u8]);
546 let mut sorted_names = entity_names.clone();
547 sorted_names.sort();
548 for name in sorted_names {
549 hasher.update(name.as_bytes());
550 hasher.update([0u8]);
551 }
552 hasher.update([1u8]);
553 }
554
555 update_sorted_set(&mut hasher, &self.when_events);
556
557 hex::encode(hasher.finalize())
558 }
559}
560
561pub struct MultiEntityBytecodeBuilder {
562 entities: HashMap<String, EntityBytecode>,
563 event_routing: HashMap<String, Vec<String>>,
564 when_events: HashSet<String>,
565 proto_router: crate::proto_router::ProtoRouter,
566}
567
568impl MultiEntityBytecodeBuilder {
569 pub fn add_entity<S>(
570 self,
571 entity_name: String,
572 spec: TypedStreamSpec<S>,
573 state_id: u32,
574 ) -> Self {
575 self.add_entity_with_evaluator(
576 entity_name,
577 spec,
578 state_id,
579 None::<
580 fn(
581 &mut Value,
582 Option<u64>,
583 i64,
584 )
585 -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>,
586 >,
587 )
588 }
589
590 pub fn add_entity_with_evaluator<S, F>(
591 mut self,
592 entity_name: String,
593 spec: TypedStreamSpec<S>,
594 state_id: u32,
595 evaluator: Option<F>,
596 ) -> Self
597 where
598 F: Fn(
599 &mut Value,
600 Option<u64>,
601 i64,
602 ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
603 + Send
604 + Sync
605 + 'static,
606 {
607 let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
608 let mut entity_bytecode = compiler.compile_entity();
609
610 if let Some(eval) = evaluator {
612 entity_bytecode.computed_fields_evaluator = Some(Box::new(eval));
613 }
614
615 for event_type in entity_bytecode.handlers.keys() {
616 self.event_routing
617 .entry(event_type.clone())
618 .or_default()
619 .push(entity_name.clone());
620 }
621
622 self.when_events
623 .extend(entity_bytecode.when_events.iter().cloned());
624
625 self.entities.insert(entity_name, entity_bytecode);
626 self
627 }
628
629 pub fn build(self) -> MultiEntityBytecode {
630 MultiEntityBytecode {
631 entities: self.entities,
632 event_routing: self.event_routing,
633 when_events: self.when_events,
634 proto_router: self.proto_router,
635 }
636 }
637}
638
639pub struct TypedCompiler<S> {
640 pub spec: TypedStreamSpec<S>,
641 entity_name: String,
642 state_id: u32,
643}
644
645impl<S> TypedCompiler<S> {
646 pub fn new(spec: TypedStreamSpec<S>, entity_name: String) -> Self {
647 TypedCompiler {
648 spec,
649 entity_name,
650 state_id: 0,
651 }
652 }
653
654 pub fn with_state_id(mut self, state_id: u32) -> Self {
655 self.state_id = state_id;
656 self
657 }
658
659 pub fn compile(&self) -> MultiEntityBytecode {
660 let entity_bytecode = self.compile_entity();
661
662 let mut entities = HashMap::new();
663 let mut event_routing = HashMap::new();
664 let mut when_events = HashSet::new();
665
666 for event_type in entity_bytecode.handlers.keys() {
667 event_routing
668 .entry(event_type.clone())
669 .or_insert_with(Vec::new)
670 .push(self.entity_name.clone());
671 }
672
673 when_events.extend(entity_bytecode.when_events.iter().cloned());
674
675 entities.insert(self.entity_name.clone(), entity_bytecode);
676
677 MultiEntityBytecode {
678 entities,
679 event_routing,
680 when_events,
681 proto_router: crate::proto_router::ProtoRouter::new(),
682 }
683 }
684
685 fn resolver_outputs_primary_key_directly(&self, primary_field: &FieldPath) -> bool {
686 if primary_field.segments.as_slice() != ["__account_address"] {
687 return false;
688 }
689
690 let primary_key_leafs: HashSet<&str> = self
691 .spec
692 .identity
693 .primary_keys
694 .iter()
695 .map(|path| path.rsplit('.').next().unwrap_or(path.as_str()))
696 .collect();
697
698 self.spec.instruction_hooks.iter().any(|hook| {
699 hook.actions.iter().any(|action| {
700 matches!(action, HookAction::RegisterPdaMapping { seed_field, .. }
701 if seed_field
702 .segments
703 .last()
704 .is_some_and(|segment| primary_key_leafs.contains(segment.as_str())))
705 })
706 })
707 }
708
709 fn compile_entity(&self) -> EntityBytecode {
710 let mut when_events: HashSet<String> = HashSet::new();
711 let mut emit_by_path: HashMap<String, bool> = HashMap::new();
712
713 let mut debug_info = Vec::new();
715 for (index, handler_spec) in self.spec.handlers.iter().enumerate() {
716 let event_type = self.get_event_type(&handler_spec.source);
717 let program_id = match &handler_spec.source {
718 crate::ast::SourceSpec::Source { program_id, .. } => {
719 program_id.as_ref().map(|s| s.as_str()).unwrap_or("null")
720 }
721 };
722 debug_info.push(format!(
723 " [{}] EventType={}, Mappings={}, ProgramId={}",
724 index,
725 event_type,
726 handler_spec.mappings.len(),
727 program_id
728 ));
729 }
730
731 let mut segments: HashMap<String, Vec<HandlerSegment>> = HashMap::new();
746
747 for handler_spec in &self.spec.handlers {
748 for mapping in &handler_spec.mappings {
749 if let Some(when) = &mapping.when {
750 when_events.insert(when.clone());
751 }
752 let entry = emit_by_path
753 .entry(mapping.target_path.clone())
754 .or_insert(false);
755 *entry |= mapping.emit;
756 if mapping.stop.is_some() {
757 emit_by_path
758 .entry(stop_field_path(&mapping.target_path))
759 .or_insert(false);
760 }
761 }
762 let opcodes = self.compile_handler(handler_spec);
763 let event_type = self.get_event_type(&handler_spec.source);
764 let key_identity = self.key_identity(handler_spec);
765
766 let event_segments = segments.entry(event_type).or_default();
767 if let Some(existing) = event_segments
768 .iter_mut()
769 .find(|segment| segment.key_identity.as_ref() == Some(&key_identity))
770 {
771 merge_handler_opcodes(&mut existing.ops, &opcodes);
772 } else {
773 event_segments.push(HandlerSegment {
774 key_identity: Some(key_identity),
775 route_field: route_field(&handler_spec.key_resolution),
776 ops: opcodes,
777 });
778 }
779 }
780
781 for hook in &self.spec.instruction_hooks {
783 let event_type = hook.instruction_type.clone();
784 let event_segments = segments.entry(event_type).or_default();
785
786 let segment_index = match &hook.lookup_by {
792 None if !event_segments.is_empty() => Some(0),
793 None => None,
794 Some(lookup_by) => event_segments
795 .iter()
796 .position(|segment| segment.route_field.as_ref() == Some(lookup_by)),
797 };
798 let segment_index = match segment_index {
799 Some(index) => index,
800 None => {
801 let resolution = if event_segments.is_empty() {
806 None
807 } else {
808 hook.lookup_by
809 .as_ref()
810 .and_then(|lookup_by| self.hook_key_resolution(lookup_by))
811 };
812 let ops = match resolution {
813 Some(resolution) => self.compile_keyed_hook_segment(&resolution),
814 None => self.compile_hook_segment(hook),
815 };
816 event_segments.push(HandlerSegment {
817 key_identity: None,
818 route_field: hook.lookup_by.clone(),
819 ops,
820 });
821 event_segments.len() - 1
822 }
823 };
824 let handler_opcodes = &mut event_segments[segment_index].ops;
825
826 let hook_opcodes = self.compile_instruction_hook_actions(&hook.actions);
828
829 let insert_pos = handler_opcodes
833 .iter()
834 .position(|op| matches!(op, OpCode::EvaluateComputedFields { .. }))
835 .or_else(|| {
836 handler_opcodes
837 .iter()
838 .position(|op| matches!(op, OpCode::UpdateState { .. }))
839 });
840
841 if let Some(pos) = insert_pos {
842 for (i, opcode) in hook_opcodes.into_iter().enumerate() {
844 handler_opcodes.insert(pos + i, opcode);
845 }
846 }
847 }
848
849 let handlers: HashMap<String, Vec<OpCode>> = segments
850 .into_iter()
851 .map(|(event_type, event_segments)| {
852 let mut ops = Vec::new();
853 for (index, segment) in event_segments.into_iter().enumerate() {
854 if index > 0 {
855 ops.push(OpCode::SegmentBoundary);
856 }
857 ops.extend(segment.ops);
858 }
859 (event_type, ops)
860 })
861 .collect();
862
863 let non_emitted_fields: HashSet<String> = emit_by_path
864 .into_iter()
865 .filter_map(|(path, emit)| if emit { None } else { Some(path) })
866 .collect();
867
868 EntityBytecode {
869 state_id: self.state_id,
870 handlers,
871 entity_name: self.entity_name.clone(),
872 when_events,
873 non_emitted_fields,
874 computed_paths: self.spec.computed_fields.clone(),
875 computed_fields_evaluator: None,
876 }
877 }
878
879 fn key_identity(&self, spec: &TypedHandlerSpec<S>) -> Value {
883 let key_reg = 20;
884 let ops = self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings);
885 serde_json::to_value(&ops).unwrap_or(Value::Null)
886 }
887
888 fn hook_key_resolution(&self, lookup_by: &FieldPath) -> Option<KeyResolutionStrategy> {
892 let leaf = lookup_by.segments.last()?;
893 let is_primary_key_field = self
894 .spec
895 .identity
896 .primary_keys
897 .iter()
898 .any(|pk| pk.rsplit('.').next() == Some(leaf.as_str()));
899 if is_primary_key_field {
900 Some(KeyResolutionStrategy::Embedded {
901 primary_field: lookup_by.clone(),
902 })
903 } else if self.find_lookup_index_for_field(lookup_by).is_some() {
904 Some(KeyResolutionStrategy::Lookup {
905 primary_field: lookup_by.clone(),
906 })
907 } else {
908 None
909 }
910 }
911
912 fn compile_keyed_hook_segment(&self, resolution: &KeyResolutionStrategy) -> Vec<OpCode> {
916 let key_reg = 20;
917 let state_reg = 2;
918 let mut ops = self.compile_key_loading(resolution, key_reg, &[]);
919 ops.push(OpCode::AbortIfNullKey {
920 key: key_reg,
921 is_account_event: false,
922 });
923 ops.push(OpCode::ReadOrInitState {
924 state_id: self.state_id,
925 key: key_reg,
926 default: serde_json::json!({}),
927 dest: state_reg,
928 });
929 ops.push(OpCode::UpdateState {
930 state_id: self.state_id,
931 key: key_reg,
932 value: state_reg,
933 });
934 ops
935 }
936
937 fn compile_hook_segment(&self, hook: &InstructionHook) -> Vec<OpCode> {
941 let key_reg = 20;
942 let state_reg = 2;
943 let resolved_key_reg = 19;
944 let temp_reg = 18;
945
946 let mut ops = Vec::new();
947
948 ops.push(OpCode::LoadEventField {
950 path: FieldPath::new(&["__resolved_primary_key"]),
951 dest: resolved_key_reg,
952 default: Some(serde_json::json!(null)),
953 });
954
955 ops.push(OpCode::CopyRegister {
957 source: resolved_key_reg,
958 dest: key_reg,
959 });
960
961 if let Some(lookup_path) = &hook.lookup_by {
963 ops.push(OpCode::LoadEventField {
965 path: lookup_path.clone(),
966 dest: temp_reg,
967 default: None,
968 });
969
970 ops.push(OpCode::Transform {
972 source: temp_reg,
973 dest: temp_reg,
974 transformation: Transformation::HexEncode,
975 });
976
977 ops.push(OpCode::CopyRegisterIfNull {
979 source: temp_reg,
980 dest: key_reg,
981 });
982 }
983
984 ops.push(OpCode::ReadOrInitState {
985 state_id: self.state_id,
986 key: key_reg,
987 default: serde_json::json!({}),
988 dest: state_reg,
989 });
990
991 ops.push(OpCode::UpdateState {
992 state_id: self.state_id,
993 key: key_reg,
994 value: state_reg,
995 });
996
997 ops
998 }
999
1000 fn compile_handler(&self, spec: &TypedHandlerSpec<S>) -> Vec<OpCode> {
1001 let mut ops = Vec::new();
1002 let state_reg = 2;
1003 let key_reg = 20;
1004
1005 ops.extend(self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings));
1006
1007 let is_account_event = matches!(
1013 spec.source,
1014 SourceSpec::Source {
1015 is_account: true,
1016 ..
1017 }
1018 );
1019 ops.push(OpCode::AbortIfNullKey {
1020 key: key_reg,
1021 is_account_event,
1022 });
1023
1024 ops.push(OpCode::ReadOrInitState {
1025 state_id: self.state_id,
1026 key: key_reg,
1027 default: serde_json::json!({}),
1028 dest: state_reg,
1029 });
1030
1031 ops.extend(self.compile_temporal_index_update(
1038 &spec.key_resolution,
1039 key_reg,
1040 &spec.mappings,
1041 ));
1042
1043 for mapping in &spec.mappings {
1044 ops.extend(self.compile_mapping(mapping, state_reg, key_reg));
1045 }
1046
1047 ops.extend(self.compile_resolvers(state_reg, key_reg));
1048
1049 ops.push(OpCode::EvaluateComputedFields {
1051 state: state_reg,
1052 computed_paths: self.spec.computed_fields.clone(),
1053 });
1054
1055 ops.push(OpCode::UpdateState {
1056 state_id: self.state_id,
1057 key: key_reg,
1058 value: state_reg,
1059 });
1060
1061 if spec.emit {
1062 ops.push(OpCode::EmitMutation {
1063 entity_name: self.entity_name.clone(),
1064 key: key_reg,
1065 state: state_reg,
1066 });
1067 }
1068
1069 ops
1070 }
1071
1072 fn compile_resolvers(&self, state_reg: Register, key_reg: Register) -> Vec<OpCode> {
1073 let mut ops = Vec::new();
1074
1075 for resolver_spec in &self.spec.resolver_specs {
1076 let url_template = match &resolver_spec.resolver {
1077 ResolverType::Url(config) => match &config.url_source {
1078 UrlSource::Template(parts) => Some(parts.clone()),
1079 _ => None,
1080 },
1081 _ => None,
1082 };
1083
1084 ops.push(OpCode::QueueResolver {
1085 state_id: self.state_id,
1086 entity_name: self.entity_name.clone(),
1087 resolver: resolver_spec.resolver.clone(),
1088 input_path: resolver_spec.input_path.clone(),
1089 input_value: resolver_spec.input_value.clone(),
1090 url_template,
1091 strategy: resolver_spec.strategy.clone(),
1092 extracts: resolver_spec.extracts.clone(),
1093 condition: resolver_spec.condition.clone(),
1094 schedule_at: resolver_spec.schedule_at.clone(),
1095 state: state_reg,
1096 key: key_reg,
1097 });
1098 }
1099
1100 ops
1101 }
1102
1103 fn compile_mapping(
1104 &self,
1105 mapping: &TypedFieldMapping<S>,
1106 state_reg: Register,
1107 key_reg: Register,
1108 ) -> Vec<OpCode> {
1109 let mut ops = Vec::new();
1110 let temp_reg = 10;
1111
1112 ops.extend(self.compile_mapping_source(&mapping.source, temp_reg));
1113
1114 if let Some(transform) = &mapping.transform {
1115 ops.push(OpCode::Transform {
1116 source: temp_reg,
1117 dest: temp_reg,
1118 transformation: transform.clone(),
1119 });
1120 }
1121
1122 if let Some(stop_instruction) = &mapping.stop {
1123 if mapping.when.is_some() {
1124 tracing::warn!(
1125 "#[map] stop and when both set for {}. Ignoring when.",
1126 mapping.target_path
1127 );
1128 }
1129 if !matches!(mapping.population, PopulationStrategy::LastWrite)
1130 && !matches!(mapping.population, PopulationStrategy::Merge)
1131 {
1132 tracing::warn!(
1133 "#[map] stop ignores population strategy {:?}",
1134 mapping.population
1135 );
1136 }
1137
1138 ops.push(OpCode::SetFieldUnlessStopped {
1139 object: state_reg,
1140 path: mapping.target_path.clone(),
1141 value: temp_reg,
1142 stop_field: stop_field_path(&mapping.target_path),
1143 stop_instruction: stop_instruction.clone(),
1144 entity_name: self.entity_name.clone(),
1145 key_reg,
1146 });
1147 return ops;
1148 }
1149
1150 if let Some(when_instruction) = &mapping.when {
1151 if !matches!(mapping.population, PopulationStrategy::LastWrite)
1152 && !matches!(mapping.population, PopulationStrategy::Merge)
1153 {
1154 tracing::warn!(
1155 "#[map] when ignores population strategy {:?}",
1156 mapping.population
1157 );
1158 }
1159 let (condition_field, condition_op, condition_value) = mapping
1160 .condition
1161 .as_ref()
1162 .and_then(|cond| cond.parsed.as_ref())
1163 .and_then(|parsed| match parsed {
1164 ParsedCondition::Comparison { field, op, value } => {
1165 Some((Some(field.clone()), Some(op.clone()), Some(value.clone())))
1166 }
1167 ParsedCondition::Logical { .. } => {
1168 tracing::warn!("Logical conditions not yet supported for #[map] when");
1169 None
1170 }
1171 })
1172 .unwrap_or((None, None, None));
1173
1174 ops.push(OpCode::SetFieldWhen {
1175 object: state_reg,
1176 path: mapping.target_path.clone(),
1177 value: temp_reg,
1178 when_instruction: when_instruction.clone(),
1179 entity_name: self.entity_name.clone(),
1180 key_reg,
1181 condition_field,
1182 condition_op,
1183 condition_value,
1184 });
1185 return ops;
1186 }
1187
1188 if let Some(condition) = &mapping.condition {
1189 if let Some(parsed) = &condition.parsed {
1190 match parsed {
1191 ParsedCondition::Comparison {
1192 field,
1193 op,
1194 value: cond_value,
1195 } => {
1196 if matches!(mapping.population, PopulationStrategy::LastWrite)
1197 || matches!(mapping.population, PopulationStrategy::Merge)
1198 {
1199 ops.push(OpCode::ConditionalSetField {
1200 object: state_reg,
1201 path: mapping.target_path.clone(),
1202 value: temp_reg,
1203 condition_field: field.clone(),
1204 condition_op: op.clone(),
1205 condition_value: cond_value.clone(),
1206 });
1207 return ops;
1208 }
1209
1210 if matches!(mapping.population, PopulationStrategy::Count) {
1211 ops.push(OpCode::ConditionalIncrement {
1212 object: state_reg,
1213 path: mapping.target_path.clone(),
1214 condition_field: field.clone(),
1215 condition_op: op.clone(),
1216 condition_value: cond_value.clone(),
1217 });
1218 return ops;
1219 }
1220
1221 tracing::warn!(
1222 "Conditional #[map] not supported for population strategy {:?}",
1223 mapping.population
1224 );
1225 }
1226 ParsedCondition::Logical { .. } => {
1227 tracing::warn!("Logical conditions not yet supported for #[map]");
1228 }
1229 }
1230 }
1231 }
1232
1233 match &mapping.population {
1234 PopulationStrategy::Append => {
1235 ops.push(OpCode::AppendToArray {
1236 object: state_reg,
1237 path: mapping.target_path.clone(),
1238 value: temp_reg,
1239 });
1240 }
1241 PopulationStrategy::LastWrite => {
1242 ops.push(OpCode::SetField {
1243 object: state_reg,
1244 path: mapping.target_path.clone(),
1245 value: temp_reg,
1246 });
1247 }
1248 PopulationStrategy::SetOnce => {
1249 ops.push(OpCode::SetFieldIfNull {
1250 object: state_reg,
1251 path: mapping.target_path.clone(),
1252 value: temp_reg,
1253 });
1254 }
1255 PopulationStrategy::Merge => {
1256 ops.push(OpCode::SetField {
1257 object: state_reg,
1258 path: mapping.target_path.clone(),
1259 value: temp_reg,
1260 });
1261 }
1262 PopulationStrategy::Max => {
1263 ops.push(OpCode::SetFieldMax {
1264 object: state_reg,
1265 path: mapping.target_path.clone(),
1266 value: temp_reg,
1267 });
1268 }
1269 PopulationStrategy::Sum => {
1270 ops.push(OpCode::SetFieldSum {
1271 object: state_reg,
1272 path: mapping.target_path.clone(),
1273 value: temp_reg,
1274 });
1275 }
1276 PopulationStrategy::Count => {
1277 ops.push(OpCode::SetFieldIncrement {
1279 object: state_reg,
1280 path: mapping.target_path.clone(),
1281 });
1282 }
1283 PopulationStrategy::Min => {
1284 ops.push(OpCode::SetFieldMin {
1285 object: state_reg,
1286 path: mapping.target_path.clone(),
1287 value: temp_reg,
1288 });
1289 }
1290 PopulationStrategy::UniqueCount => {
1291 let set_name = format!("{}_unique_set", mapping.target_path);
1294 ops.push(OpCode::AddToUniqueSet {
1295 state_id: self.state_id,
1296 set_name,
1297 value: temp_reg,
1298 count_object: state_reg,
1299 count_path: mapping.target_path.clone(),
1300 });
1301 }
1302 }
1303
1304 ops
1305 }
1306
1307 fn compile_mapping_source(&self, source: &MappingSource, dest: Register) -> Vec<OpCode> {
1308 match source {
1309 MappingSource::FromSource {
1310 path,
1311 default,
1312 transform,
1313 } => {
1314 let mut ops = vec![OpCode::LoadEventField {
1315 path: path.clone(),
1316 dest,
1317 default: default.clone(),
1318 }];
1319
1320 if let Some(transform_type) = transform {
1322 ops.push(OpCode::Transform {
1323 source: dest,
1324 dest,
1325 transformation: transform_type.clone(),
1326 });
1327 }
1328
1329 ops
1330 }
1331 MappingSource::Constant(val) => {
1332 vec![OpCode::LoadConstant {
1333 value: val.clone(),
1334 dest,
1335 }]
1336 }
1337 MappingSource::AsEvent { fields } => {
1338 let mut ops = Vec::new();
1339
1340 if fields.is_empty() {
1341 let event_data_reg = dest + 1;
1342 ops.push(OpCode::LoadEventField {
1343 path: FieldPath::new(&[]),
1344 dest: event_data_reg,
1345 default: Some(serde_json::json!({})),
1346 });
1347 ops.push(OpCode::CreateEvent {
1348 dest,
1349 event_value: event_data_reg,
1350 });
1351 } else {
1352 let data_obj_reg = dest + 1;
1353 ops.push(OpCode::CreateObject { dest: data_obj_reg });
1354
1355 let mut field_registers = Vec::new();
1356 let mut current_reg = dest + 2;
1357
1358 for field_source in fields.iter() {
1359 if let MappingSource::FromSource {
1360 path,
1361 default,
1362 transform,
1363 } = &**field_source
1364 {
1365 ops.push(OpCode::LoadEventField {
1366 path: path.clone(),
1367 dest: current_reg,
1368 default: default.clone(),
1369 });
1370
1371 if let Some(transform_type) = transform {
1372 ops.push(OpCode::Transform {
1373 source: current_reg,
1374 dest: current_reg,
1375 transformation: transform_type.clone(),
1376 });
1377 }
1378
1379 if let Some(field_name) = path.segments.last() {
1380 field_registers.push((field_name.clone(), current_reg));
1381 }
1382 current_reg += 1;
1383 }
1384 }
1385
1386 if !field_registers.is_empty() {
1387 ops.push(OpCode::SetFields {
1388 object: data_obj_reg,
1389 fields: field_registers,
1390 });
1391 }
1392
1393 ops.push(OpCode::CreateEvent {
1394 dest,
1395 event_value: data_obj_reg,
1396 });
1397 }
1398
1399 ops
1400 }
1401 MappingSource::WholeSource => {
1402 vec![OpCode::LoadEventField {
1403 path: FieldPath::new(&[]),
1404 dest,
1405 default: Some(serde_json::json!({})),
1406 }]
1407 }
1408 MappingSource::AsCapture { field_transforms } => {
1409 let capture_data_reg = 22; let mut ops = vec![OpCode::LoadEventField {
1412 path: FieldPath::new(&[]),
1413 dest: capture_data_reg,
1414 default: Some(serde_json::json!({})),
1415 }];
1416
1417 let field_reg = 24;
1421 let transformed_reg = 25;
1422
1423 for (field_name, transform) in field_transforms {
1424 ops.push(OpCode::GetField {
1427 object: capture_data_reg,
1428 path: field_name.clone(),
1429 dest: field_reg,
1430 });
1431
1432 ops.push(OpCode::Transform {
1434 source: field_reg,
1435 dest: transformed_reg,
1436 transformation: transform.clone(),
1437 });
1438
1439 ops.push(OpCode::SetField {
1441 object: capture_data_reg,
1442 path: field_name.clone(),
1443 value: transformed_reg,
1444 });
1445 }
1446
1447 ops.push(OpCode::CreateCapture {
1449 dest,
1450 capture_value: capture_data_reg,
1451 });
1452
1453 ops
1454 }
1455 MappingSource::FromContext { field } => {
1456 vec![OpCode::LoadEventField {
1458 path: FieldPath::new(&["__update_context", field.as_str()]),
1459 dest,
1460 default: Some(serde_json::json!(null)),
1461 }]
1462 }
1463 MappingSource::FromExpr { expr } => {
1464 vec![OpCode::EvaluateEventExpr {
1465 expr: expr.clone(),
1466 transform: None,
1467 dest,
1468 }]
1469 }
1470 MappingSource::Computed { .. } => {
1471 vec![]
1472 }
1473 MappingSource::FromState { .. } => {
1474 vec![]
1475 }
1476 }
1477 }
1478
1479 pub fn compile_key_loading(
1480 &self,
1481 resolution: &KeyResolutionStrategy,
1482 key_reg: Register,
1483 mappings: &[TypedFieldMapping<S>],
1484 ) -> Vec<OpCode> {
1485 let mut ops = Vec::new();
1486
1487 let resolved_key_reg = 19; ops.push(OpCode::LoadEventField {
1491 path: FieldPath::new(&["__resolved_primary_key"]),
1492 dest: resolved_key_reg,
1493 default: Some(serde_json::json!(null)),
1494 });
1495
1496 match resolution {
1498 KeyResolutionStrategy::Embedded { primary_field }
1499 if primary_field.segments.is_empty()
1500 && self.find_primary_key_expr(mappings).is_some() =>
1501 {
1502 let expr = self
1505 .find_primary_key_expr(mappings)
1506 .expect("checked by the guard");
1507 ops.push(OpCode::EvaluateEventExpr {
1511 expr,
1512 transform: self.find_primary_key_transformation(mappings),
1513 dest: key_reg,
1514 });
1515 ops.push(OpCode::CopyRegisterIfNull {
1516 source: resolved_key_reg,
1517 dest: key_reg,
1518 });
1519 }
1520 KeyResolutionStrategy::Embedded { primary_field } => {
1521 let effective_primary_field = if primary_field.segments.is_empty() {
1523 if let Some(auto_field) = self.auto_detect_primary_field(mappings) {
1525 auto_field
1526 } else {
1527 primary_field.clone()
1528 }
1529 } else {
1530 primary_field.clone()
1531 };
1532
1533 if !effective_primary_field.segments.is_empty() {
1537 let temp_reg = 18;
1538 let transform_reg = 23; ops.push(OpCode::LoadEventField {
1541 path: effective_primary_field.clone(),
1542 dest: temp_reg,
1543 default: None,
1544 });
1545
1546 let primary_key_transform = self
1549 .find_primary_key_transformation(mappings)
1550 .or_else(|| self.find_inherited_primary_key_transformation());
1551
1552 if let Some(transform) = primary_key_transform {
1553 ops.push(OpCode::Transform {
1555 source: temp_reg,
1556 dest: transform_reg,
1557 transformation: transform,
1558 });
1559 ops.push(OpCode::CopyRegister {
1563 source: transform_reg,
1564 dest: key_reg,
1565 });
1566 } else {
1567 ops.push(OpCode::CopyRegister {
1569 source: temp_reg,
1570 dest: key_reg,
1571 });
1572 }
1573 ops.push(OpCode::CopyRegisterIfNull {
1574 source: resolved_key_reg,
1575 dest: key_reg,
1576 });
1577 } else {
1578 ops.push(OpCode::CopyRegister {
1580 source: resolved_key_reg,
1581 dest: key_reg,
1582 });
1583 }
1584 }
1585 KeyResolutionStrategy::Lookup { primary_field } => {
1586 let lookup_reg = 15;
1587 let result_reg = 17;
1588
1589 ops.push(OpCode::CopyRegister {
1595 source: resolved_key_reg,
1596 dest: lookup_reg,
1597 });
1598
1599 let temp_reg = 18;
1600 ops.push(OpCode::LoadEventField {
1601 path: primary_field.clone(),
1602 dest: temp_reg,
1603 default: None,
1604 });
1605 ops.push(OpCode::CopyRegisterIfNull {
1606 source: temp_reg,
1607 dest: lookup_reg,
1608 });
1609
1610 let index_name = self.find_lookup_index_for_lookup_field(primary_field, mappings);
1611 let effective_index_name =
1612 index_name.unwrap_or_else(|| "default_pda_lookup".to_string());
1613
1614 ops.push(OpCode::LookupIndex {
1615 state_id: self.state_id,
1616 index_name: effective_index_name,
1617 lookup_value: lookup_reg,
1618 dest: result_reg,
1619 });
1620 ops.push(OpCode::CopyRegister {
1621 source: result_reg,
1622 dest: key_reg,
1623 });
1624
1625 if self.resolver_outputs_primary_key_directly(primary_field) {
1633 ops.push(OpCode::CopyRegisterIfNull {
1634 source: resolved_key_reg,
1635 dest: key_reg,
1636 });
1637 }
1638 }
1639 KeyResolutionStrategy::Computed {
1640 primary_field,
1641 compute_partition: _,
1642 } => {
1643 ops.push(OpCode::CopyRegister {
1645 source: resolved_key_reg,
1646 dest: key_reg,
1647 });
1648 let temp_reg = 18;
1649 ops.push(OpCode::LoadEventField {
1650 path: primary_field.clone(),
1651 dest: temp_reg,
1652 default: None,
1653 });
1654 ops.push(OpCode::CopyRegisterIfNull {
1655 source: temp_reg,
1656 dest: key_reg,
1657 });
1658 }
1659 KeyResolutionStrategy::TemporalLookup {
1660 lookup_field,
1661 timestamp_field,
1662 index_name,
1663 } => {
1664 ops.push(OpCode::CopyRegister {
1666 source: resolved_key_reg,
1667 dest: key_reg,
1668 });
1669 let lookup_reg = 15;
1670 let timestamp_reg = 16;
1671 let result_reg = 17;
1672
1673 ops.push(OpCode::LoadEventField {
1674 path: lookup_field.clone(),
1675 dest: lookup_reg,
1676 default: None,
1677 });
1678
1679 ops.push(OpCode::LoadEventField {
1680 path: timestamp_field.clone(),
1681 dest: timestamp_reg,
1682 default: None,
1683 });
1684
1685 ops.push(OpCode::LookupTemporalIndex {
1686 state_id: self.state_id,
1687 index_name: index_name.clone(),
1688 lookup_value: lookup_reg,
1689 timestamp: timestamp_reg,
1690 dest: result_reg,
1691 });
1692
1693 ops.push(OpCode::CopyRegisterIfNull {
1694 source: result_reg,
1695 dest: key_reg,
1696 });
1697 }
1698 }
1699
1700 ops
1701 }
1702
1703 fn find_primary_key_expr(&self, mappings: &[TypedFieldMapping<S>]) -> Option<ComputedExpr> {
1706 let primary_key = self.spec.identity.primary_keys.first()?;
1707 mappings.iter().find_map(|mapping| match &mapping.source {
1708 MappingSource::FromExpr { expr } if mapping.target_path == *primary_key => {
1709 Some(expr.clone())
1710 }
1711 _ => None,
1712 })
1713 }
1714
1715 fn find_primary_key_transformation(
1716 &self,
1717 mappings: &[TypedFieldMapping<S>],
1718 ) -> Option<Transformation> {
1719 let primary_key = self.spec.identity.primary_keys.first()?;
1721 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1722
1723 for mapping in mappings {
1725 if mapping.target_path == *primary_key
1727 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1728 {
1729 if let Some(transform) = &mapping.transform {
1731 return Some(transform.clone());
1732 }
1733
1734 if let MappingSource::FromSource {
1736 transform: Some(transform),
1737 ..
1738 } = &mapping.source
1739 {
1740 return Some(transform.clone());
1741 }
1742 }
1743 }
1744
1745 for mapping in mappings {
1747 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1748 if let Some(transform) = field_transforms.get(&primary_field_name) {
1749 return Some(transform.clone());
1750 }
1751 }
1752 }
1753
1754 None
1755 }
1756
1757 pub fn find_inherited_primary_key_transformation(&self) -> Option<Transformation> {
1760 let primary_key = self.spec.identity.primary_keys.first()?;
1761
1762 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1764
1765 for handler in &self.spec.handlers {
1767 for mapping in &handler.mappings {
1768 if mapping.target_path == *primary_key
1770 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1771 {
1772 if let MappingSource::FromSource {
1774 path, transform, ..
1775 } = &mapping.source
1776 {
1777 if path.segments.last() == Some(&primary_field_name) {
1778 return mapping.transform.clone().or_else(|| transform.clone());
1780 }
1781 }
1782 }
1783
1784 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1786 if let Some(transform) = field_transforms.get(&primary_field_name) {
1787 return Some(transform.clone());
1788 }
1789 }
1790 }
1791 }
1792
1793 None
1794 }
1795
1796 fn extract_primary_field_name(&self, primary_key: &str) -> Option<String> {
1798 primary_key.split('.').next_back().map(|s| s.to_string())
1800 }
1801
1802 pub fn auto_detect_primary_field(
1805 &self,
1806 current_mappings: &[TypedFieldMapping<S>],
1807 ) -> Option<FieldPath> {
1808 let primary_key = self.spec.identity.primary_keys.first()?;
1809
1810 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1812
1813 if self.current_account_has_primary_field(&primary_field_name, current_mappings) {
1815 return Some(FieldPath::new(&[&primary_field_name]));
1816 }
1817
1818 None
1819 }
1820
1821 fn current_account_has_primary_field(
1824 &self,
1825 field_name: &str,
1826 mappings: &[TypedFieldMapping<S>],
1827 ) -> bool {
1828 for mapping in mappings {
1830 if let MappingSource::FromSource { path, .. } = &mapping.source {
1831 if path.segments.last() == Some(&field_name.to_string()) {
1833 return true;
1834 }
1835 }
1836 }
1837
1838 false
1839 }
1840
1841 #[allow(dead_code)]
1843 fn handler_has_field(&self, field_name: &str, mappings: &[TypedFieldMapping<S>]) -> bool {
1844 for mapping in mappings {
1845 if let MappingSource::FromSource { path, .. } = &mapping.source {
1846 if path.segments.last() == Some(&field_name.to_string()) {
1847 return true;
1848 }
1849 }
1850 }
1851 false
1852 }
1853
1854 #[allow(dead_code)]
1857 fn field_exists_in_mappings(
1858 &self,
1859 field_name: &str,
1860 mappings: &[TypedFieldMapping<S>],
1861 ) -> bool {
1862 for mapping in mappings {
1864 if let MappingSource::FromSource { path, .. } = &mapping.source {
1865 if path.segments.last() == Some(&field_name.to_string()) {
1866 return true;
1867 }
1868 }
1869 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1871 if field_transforms.contains_key(field_name) {
1872 return true;
1873 }
1874 }
1875 }
1876 false
1877 }
1878
1879 fn find_lookup_index_for_field(&self, field_path: &FieldPath) -> Option<String> {
1880 if field_path.segments.is_empty() {
1881 return None;
1882 }
1883
1884 let lookup_field_name = field_path.segments.last().unwrap();
1885
1886 for lookup_index in &self.spec.identity.lookup_indexes {
1887 let index_field_name = lookup_index
1888 .field_name
1889 .split('.')
1890 .next_back()
1891 .unwrap_or(&lookup_index.field_name);
1892 let matches_directly = index_field_name == lookup_field_name;
1893 let matches_address_alias = index_field_name
1900 .strip_suffix("_address")
1901 .map(|base| base == lookup_field_name)
1902 .unwrap_or(false);
1903
1904 if matches_directly || matches_address_alias {
1905 return Some(format!("{}_lookup_index", index_field_name));
1906 }
1907 }
1908
1909 None
1910 }
1911
1912 fn find_lookup_index_for_lookup_field(
1915 &self,
1916 primary_field: &FieldPath,
1917 mappings: &[TypedFieldMapping<S>],
1918 ) -> Option<String> {
1919 let primary_path = primary_field.segments.join(".");
1921
1922 for mapping in mappings {
1924 if let MappingSource::FromSource { path, .. } = &mapping.source {
1926 let source_path = path.segments.join(".");
1927 if source_path == primary_path {
1928 for lookup_index in &self.spec.identity.lookup_indexes {
1930 if mapping.target_path == lookup_index.field_name {
1931 let index_field_name = lookup_index
1932 .field_name
1933 .split('.')
1934 .next_back()
1935 .unwrap_or(&lookup_index.field_name);
1936 return Some(format!("{}_lookup_index", index_field_name));
1937 }
1938 }
1939 }
1940 }
1941 }
1942
1943 self.find_lookup_index_for_field(primary_field)
1945 }
1946
1947 fn find_source_path_for_lookup_index(
1951 &self,
1952 mappings: &[TypedFieldMapping<S>],
1953 lookup_field_name: &str,
1954 ) -> Option<Vec<String>> {
1955 for mapping in mappings {
1956 if mapping.target_path == lookup_field_name {
1957 if let MappingSource::FromSource { path, .. } = &mapping.source {
1958 return Some(path.segments.clone());
1959 }
1960 }
1961 }
1962 None
1963 }
1964
1965 fn compile_temporal_index_update(
1966 &self,
1967 resolution: &KeyResolutionStrategy,
1968 key_reg: Register,
1969 mappings: &[TypedFieldMapping<S>],
1970 ) -> Vec<OpCode> {
1971 let mut ops = Vec::new();
1972
1973 for lookup_index in &self.spec.identity.lookup_indexes {
1974 let lookup_reg = 17;
1975 let source_field = lookup_index
1976 .field_name
1977 .split('.')
1978 .next_back()
1979 .unwrap_or(&lookup_index.field_name);
1980
1981 match resolution {
1982 KeyResolutionStrategy::Embedded { primary_field: _ } => {
1983 let source_path_opt =
1986 self.find_source_path_for_lookup_index(mappings, &lookup_index.field_name);
1987
1988 let load_path = if let Some(ref path) = source_path_opt {
1989 FieldPath::new(&path.iter().map(|s| s.as_str()).collect::<Vec<_>>())
1990 } else {
1991 FieldPath::new(&[source_field])
1993 };
1994
1995 ops.push(OpCode::LoadEventField {
1996 path: load_path,
1997 dest: lookup_reg,
1998 default: None,
1999 });
2000
2001 if let Some(temporal_field_name) = &lookup_index.temporal_field {
2002 let timestamp_reg = 18;
2003
2004 ops.push(OpCode::LoadEventField {
2005 path: FieldPath::new(&[temporal_field_name]),
2006 dest: timestamp_reg,
2007 default: None,
2008 });
2009
2010 let index_name = format!("{}_temporal_index", source_field);
2011 ops.push(OpCode::UpdateTemporalIndex {
2012 state_id: self.state_id,
2013 index_name,
2014 lookup_value: lookup_reg,
2015 primary_key: key_reg,
2016 timestamp: timestamp_reg,
2017 });
2018
2019 let simple_index_name = format!("{}_lookup_index", source_field);
2020 ops.push(OpCode::UpdateLookupIndex {
2021 state_id: self.state_id,
2022 index_name: simple_index_name,
2023 lookup_value: lookup_reg,
2024 primary_key: key_reg,
2025 });
2026 } else {
2027 let index_name = format!("{}_lookup_index", source_field);
2028 ops.push(OpCode::UpdateLookupIndex {
2029 state_id: self.state_id,
2030 index_name,
2031 lookup_value: lookup_reg,
2032 primary_key: key_reg,
2033 });
2034 }
2035
2036 if source_path_opt.is_some() {
2040 ops.push(OpCode::UpdatePdaReverseLookup {
2041 state_id: self.state_id,
2042 lookup_name: "default_pda_lookup".to_string(),
2043 pda_address: lookup_reg,
2044 primary_key: key_reg,
2045 });
2046 }
2047 }
2048 KeyResolutionStrategy::Lookup { primary_field } => {
2049 let has_mapping_to_lookup_field = mappings
2052 .iter()
2053 .any(|m| m.target_path == lookup_index.field_name);
2054
2055 if has_mapping_to_lookup_field {
2056 let path_segments: Vec<&str> =
2059 primary_field.segments.iter().map(|s| s.as_str()).collect();
2060 ops.push(OpCode::LoadEventField {
2061 path: FieldPath::new(&path_segments),
2062 dest: lookup_reg,
2063 default: None,
2064 });
2065
2066 let index_name = format!("{}_lookup_index", source_field);
2067 ops.push(OpCode::UpdateLookupIndex {
2068 state_id: self.state_id,
2069 index_name,
2070 lookup_value: lookup_reg,
2071 primary_key: key_reg,
2072 });
2073 }
2074 }
2075 KeyResolutionStrategy::Computed { .. }
2076 | KeyResolutionStrategy::TemporalLookup { .. } => {
2077 }
2079 }
2080 }
2081
2082 ops
2083 }
2084
2085 fn get_event_type(&self, source: &SourceSpec) -> String {
2086 match source {
2087 SourceSpec::Source { type_name, .. } => type_name.clone(),
2088 }
2089 }
2090
2091 fn compile_instruction_hook_actions(&self, actions: &[HookAction]) -> Vec<OpCode> {
2092 let mut ops = Vec::new();
2093 let state_reg = 2;
2094
2095 for action in actions {
2096 match action {
2097 HookAction::SetField {
2098 target_field,
2099 source,
2100 condition,
2101 } => {
2102 let _ = condition;
2104
2105 let temp_reg = 11; let load_ops = self.compile_mapping_source(source, temp_reg);
2109 ops.extend(load_ops);
2110
2111 if let MappingSource::FromSource {
2113 transform: Some(transform_type),
2114 ..
2115 } = source
2116 {
2117 ops.push(OpCode::Transform {
2118 source: temp_reg,
2119 dest: temp_reg,
2120 transformation: transform_type.clone(),
2121 });
2122 }
2123
2124 if let Some(cond_expr) = condition {
2126 if let Some(parsed) = &cond_expr.parsed {
2127 let cond_check_ops = self.compile_condition_check(
2129 parsed,
2130 temp_reg,
2131 state_reg,
2132 target_field,
2133 );
2134 ops.extend(cond_check_ops);
2135 } else {
2136 ops.push(OpCode::SetField {
2138 object: state_reg,
2139 path: target_field.clone(),
2140 value: temp_reg,
2141 });
2142 }
2143 } else {
2144 ops.push(OpCode::SetField {
2146 object: state_reg,
2147 path: target_field.clone(),
2148 value: temp_reg,
2149 });
2150 }
2151 }
2152 HookAction::IncrementField {
2153 target_field,
2154 increment_by,
2155 condition,
2156 } => {
2157 if let Some(cond_expr) = condition {
2158 if let Some(parsed) = &cond_expr.parsed {
2159 let cond_check_ops = self.compile_conditional_increment(
2164 parsed,
2165 state_reg,
2166 target_field,
2167 *increment_by,
2168 );
2169 ops.extend(cond_check_ops);
2170 } else {
2171 ops.push(OpCode::SetFieldIncrement {
2173 object: state_reg,
2174 path: target_field.clone(),
2175 });
2176 }
2177 } else {
2178 ops.push(OpCode::SetFieldIncrement {
2180 object: state_reg,
2181 path: target_field.clone(),
2182 });
2183 }
2184 }
2185 HookAction::RegisterPdaMapping { .. } => {
2186 if let HookAction::RegisterPdaMapping {
2187 pda_field,
2188 seed_field,
2189 lookup_name,
2190 } = action
2191 {
2192 let pda_reg = 11;
2193 let seed_reg = 12;
2194
2195 ops.push(OpCode::LoadEventField {
2196 path: pda_field.clone(),
2197 dest: pda_reg,
2198 default: None,
2199 });
2200 ops.push(OpCode::LoadEventField {
2201 path: seed_field.clone(),
2202 dest: seed_reg,
2203 default: None,
2204 });
2205 ops.push(OpCode::UpdatePdaReverseLookup {
2206 state_id: self.state_id,
2207 lookup_name: lookup_name.clone(),
2208 pda_address: pda_reg,
2209 primary_key: seed_reg,
2210 });
2211 }
2212 }
2213 }
2214 }
2215
2216 ops
2217 }
2218
2219 fn compile_condition_check(
2220 &self,
2221 condition: &ParsedCondition,
2222 value_reg: Register,
2223 state_reg: Register,
2224 target_field: &str,
2225 ) -> Vec<OpCode> {
2226 match condition {
2227 ParsedCondition::Comparison {
2228 field,
2229 op,
2230 value: cond_value,
2231 } => {
2232 vec![OpCode::ConditionalSetField {
2234 object: state_reg,
2235 path: target_field.to_string(),
2236 value: value_reg,
2237 condition_field: field.clone(),
2238 condition_op: op.clone(),
2239 condition_value: cond_value.clone(),
2240 }]
2241 }
2242 ParsedCondition::Logical { .. } => {
2243 tracing::warn!("Logical conditions not yet supported in instruction hooks");
2245 vec![OpCode::SetField {
2246 object: state_reg,
2247 path: target_field.to_string(),
2248 value: value_reg,
2249 }]
2250 }
2251 }
2252 }
2253
2254 fn compile_conditional_increment(
2255 &self,
2256 condition: &ParsedCondition,
2257 state_reg: Register,
2258 target_field: &str,
2259 _increment_by: i64,
2260 ) -> Vec<OpCode> {
2261 match condition {
2262 ParsedCondition::Comparison {
2263 field,
2264 op,
2265 value: cond_value,
2266 } => {
2267 vec![OpCode::ConditionalIncrement {
2268 object: state_reg,
2269 path: target_field.to_string(),
2270 condition_field: field.clone(),
2271 condition_op: op.clone(),
2272 condition_value: cond_value.clone(),
2273 }]
2274 }
2275 ParsedCondition::Logical { .. } => {
2276 tracing::warn!("Logical conditions not yet supported in instruction hooks");
2277 vec![OpCode::SetFieldIncrement {
2278 object: state_reg,
2279 path: target_field.to_string(),
2280 }]
2281 }
2282 }
2283 }
2284}
2285
2286#[cfg(test)]
2287mod tests {
2288 use super::MultiEntityBytecode;
2289 use crate::ast::{
2290 FieldPath, HookAction, IdentitySpec, InstructionHook, KeyResolutionStrategy,
2291 LookupIndexSpec, MappingSource, PopulationStrategy, SerializableFieldMapping,
2292 SerializableHandlerSpec, SerializableStreamSpec, SourceSpec, TypedStreamSpec,
2293 };
2294 use crate::vm::VmContext;
2295 use crate::{Mutation, WholeEntity};
2296 use serde_json::{json, Value};
2297 use std::collections::BTreeMap;
2298
2299 fn mapping(
2300 target_path: &str,
2301 source_path: &[&str],
2302 population: PopulationStrategy,
2303 ) -> SerializableFieldMapping {
2304 SerializableFieldMapping {
2305 target_path: target_path.to_string(),
2306 source: MappingSource::FromSource {
2307 path: FieldPath::new(source_path),
2308 default: None,
2309 transform: None,
2310 },
2311 transform: None,
2312 population,
2313 condition: None,
2314 when: None,
2315 stop: None,
2316 emit: true,
2317 }
2318 }
2319
2320 fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
2321 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2322 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2323 state_name: "PumpfunToken".to_string(),
2324 program_id: None,
2325 idl: None,
2326 identity: IdentitySpec {
2327 primary_keys: vec!["id.mint".to_string()],
2328 lookup_indexes: vec![LookupIndexSpec {
2329 field_name: "id.bonding_curve".to_string(),
2330 temporal_field: None,
2331 }],
2332 },
2333 handlers: vec![SerializableHandlerSpec {
2334 source: SourceSpec::Source {
2335 program_id: None,
2336 discriminator: None,
2337 type_name: "pump::BondingCurveState".to_string(),
2338 serialization: None,
2339 is_account: true,
2340 },
2341 key_resolution: KeyResolutionStrategy::Lookup {
2342 primary_field: FieldPath::new(&["__account_address"]),
2343 },
2344 mappings: vec![
2345 mapping(
2346 "id.bonding_curve",
2347 &["__account_address"],
2348 PopulationStrategy::SetOnce,
2349 ),
2350 mapping(
2351 "reserves.virtual_token_reserves",
2352 &["virtual_token_reserves"],
2353 PopulationStrategy::LastWrite,
2354 ),
2355 ],
2356 conditions: vec![],
2357 emit: true,
2358 }],
2359 sections: vec![],
2360 field_mappings: BTreeMap::new(),
2361 resolver_hooks: vec![],
2362 instruction_hooks: vec![InstructionHook {
2363 instruction_type: "pump::BuyIxState".to_string(),
2364 actions: vec![HookAction::RegisterPdaMapping {
2365 pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2366 seed_field: FieldPath::new(&["accounts", "mint"]),
2367 lookup_name: "default_pda_lookup".to_string(),
2368 }],
2369 lookup_by: None,
2370 }],
2371 resolver_specs: vec![],
2372 computed_fields: vec![],
2373 computed_field_specs: vec![],
2374 content_hash: None,
2375 views: vec![],
2376 })
2377 }
2378
2379 fn timed_bonding_curve_spec() -> TypedStreamSpec<Value> {
2386 let with_source = |target: &str, source: MappingSource| SerializableFieldMapping {
2387 source,
2388 ..mapping(target, &[], PopulationStrategy::LastWrite)
2389 };
2390 let timestamped = || {
2391 vec![
2392 with_source(
2393 "events.reserves",
2394 MappingSource::AsEvent {
2395 fields: vec![Box::new(MappingSource::FromSource {
2396 path: FieldPath::new(&["virtual_token_reserves"]),
2397 default: None,
2398 transform: None,
2399 })],
2400 },
2401 ),
2402 with_source(
2403 "snapshots.curve",
2404 MappingSource::AsCapture {
2405 field_transforms: BTreeMap::new(),
2406 },
2407 ),
2408 with_source(
2409 "state.seen_at",
2410 MappingSource::FromContext {
2411 field: "timestamp".to_string(),
2412 },
2413 ),
2414 ]
2415 };
2416 let source = |type_name: &str, is_account: bool| SourceSpec::Source {
2417 program_id: None,
2418 discriminator: None,
2419 type_name: type_name.to_string(),
2420 serialization: None,
2421 is_account,
2422 };
2423 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2424 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2425 state_name: "PumpfunToken".to_string(),
2426 program_id: None,
2427 idl: None,
2428 identity: IdentitySpec {
2429 primary_keys: vec!["id.mint".to_string()],
2430 lookup_indexes: vec![
2431 LookupIndexSpec {
2432 field_name: "id.bonding_curve".to_string(),
2433 temporal_field: None,
2434 },
2435 LookupIndexSpec {
2436 field_name: "id.creator".to_string(),
2437 temporal_field: None,
2438 },
2439 LookupIndexSpec {
2440 field_name: "id.round".to_string(),
2441 temporal_field: None,
2442 },
2443 ],
2444 },
2445 handlers: vec![
2446 SerializableHandlerSpec {
2447 source: source("pump::BondingCurveState", true),
2448 key_resolution: KeyResolutionStrategy::Lookup {
2449 primary_field: FieldPath::new(&["__account_address"]),
2450 },
2451 mappings: [
2452 vec![mapping(
2453 "id.bonding_curve",
2454 &["__account_address"],
2455 PopulationStrategy::SetOnce,
2456 )],
2457 timestamped(),
2458 ]
2459 .concat(),
2460 conditions: vec![],
2461 emit: true,
2462 },
2463 SerializableHandlerSpec {
2464 source: source("pump::RoundState", true),
2465 key_resolution: KeyResolutionStrategy::Lookup {
2466 primary_field: FieldPath::new(&["round"]),
2467 },
2468 mappings: timestamped(),
2469 conditions: vec![],
2470 emit: true,
2471 },
2472 SerializableHandlerSpec {
2473 source: source("pump::OpenRoundIxState", false),
2474 key_resolution: KeyResolutionStrategy::Embedded {
2475 primary_field: FieldPath::new(&["accounts", "mint"]),
2476 },
2477 mappings: vec![
2478 mapping(
2479 "id.mint",
2480 &["accounts", "mint"],
2481 PopulationStrategy::SetOnce,
2482 ),
2483 mapping(
2484 "id.round",
2485 &["accounts", "round"],
2486 PopulationStrategy::SetOnce,
2487 ),
2488 ],
2489 conditions: vec![],
2490 emit: true,
2491 },
2492 SerializableHandlerSpec {
2493 source: source("pump::LinkIxState", false),
2494 key_resolution: KeyResolutionStrategy::Embedded {
2495 primary_field: FieldPath::new(&["accounts", "mint"]),
2496 },
2497 mappings: vec![
2498 mapping(
2499 "id.mint",
2500 &["accounts", "mint"],
2501 PopulationStrategy::SetOnce,
2502 ),
2503 mapping(
2504 "id.bonding_curve",
2505 &["accounts", "bonding_curve"],
2506 PopulationStrategy::SetOnce,
2507 ),
2508 ],
2509 conditions: vec![],
2510 emit: true,
2511 },
2512 SerializableHandlerSpec {
2513 source: source("pump::CreateIxState", false),
2514 key_resolution: KeyResolutionStrategy::Embedded {
2515 primary_field: FieldPath::new(&["accounts", "mint"]),
2516 },
2517 mappings: vec![
2518 mapping(
2519 "id.mint",
2520 &["accounts", "mint"],
2521 PopulationStrategy::SetOnce,
2522 ),
2523 mapping(
2524 "id.creator",
2525 &["accounts", "user"],
2526 PopulationStrategy::SetOnce,
2527 ),
2528 ],
2529 conditions: vec![],
2530 emit: true,
2531 },
2532 ],
2533 sections: vec![],
2534 field_mappings: BTreeMap::new(),
2535 resolver_hooks: vec![],
2536 instruction_hooks: vec![InstructionHook {
2537 instruction_type: "pump::BuyIxState".to_string(),
2538 actions: vec![HookAction::RegisterPdaMapping {
2539 pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2540 seed_field: FieldPath::new(&["accounts", "user"]),
2541 lookup_name: "default_pda_lookup".to_string(),
2542 }],
2543 lookup_by: None,
2544 }],
2545 resolver_specs: vec![],
2546 computed_fields: vec![],
2547 computed_field_specs: vec![],
2548 content_hash: None,
2549 views: vec![],
2550 })
2551 }
2552
2553 #[test]
2561 fn a_queued_account_update_replays_with_its_own_timestamp() {
2562 use crate::vm::UpdateContext;
2563
2564 let accounts = json!({
2565 "bonding_curve": "bonding_curve_1",
2566 "mint": "mint_1",
2567 "user": "user_1",
2568 "round": 7,
2569 });
2570 let curve = (
2571 "pump::BondingCurveState",
2572 json!({ "__account_address": "bonding_curve_1", "virtual_token_reserves": 42 }),
2573 );
2574 let round = (
2575 "pump::RoundState",
2576 json!({ "__account_address": "round_1", "round": 7, "virtual_token_reserves": 42 }),
2577 );
2578 let instruction = |vm: &mut VmContext, bytecode: &MultiEntityBytecode, name: &str| {
2579 let mut context =
2580 UpdateContext::new_instruction(20, "instruction-signature".to_string(), 0);
2581 context.timestamp = Some(2_000);
2582 vm.process_event(
2583 bytecode,
2584 json!({ "accounts": accounts }),
2585 name,
2586 Some(&context),
2587 None,
2588 )
2589 .unwrap();
2590 };
2591 let scenarios: [(&[&str], _, &[&str]); 4] = [
2594 (&[], &curve, &["pump::LinkIxState"]),
2595 (&[], &curve, &["pump::CreateIxState", "pump::BuyIxState"]),
2596 (&["pump::BuyIxState"], &curve, &["pump::CreateIxState"]),
2597 (&[], &round, &["pump::OpenRoundIxState"]),
2598 ];
2599 for (before, (account_type, account), after) in scenarios {
2600 let bytecode = MultiEntityBytecode::from_single(
2601 "PumpfunToken".to_string(),
2602 timed_bonding_curve_spec(),
2603 0,
2604 );
2605 let mut vm = VmContext::new();
2606 for name in before {
2607 instruction(&mut vm, &bytecode, name);
2608 }
2609 let mut account_context =
2610 UpdateContext::new_account(10, "curve-signature".to_string(), 1);
2611 account_context.timestamp = Some(1_000);
2612
2613 let queued = vm
2614 .process_event(
2615 &bytecode,
2616 account.clone(),
2617 account_type,
2618 Some(&account_context),
2619 None,
2620 )
2621 .unwrap();
2622 assert!(queued.is_empty(), "{after:?}: {queued:?}");
2623
2624 std::thread::sleep(std::time::Duration::from_millis(1_100));
2627 for name in after {
2628 instruction(&mut vm, &bytecode, name);
2629 }
2630
2631 let token = vm
2632 .get_entity_state(0, &json!("mint_1"))
2633 .unwrap_or_else(|| panic!("{after:?}: no entity"));
2634 for path in [
2635 "/events/reserves/timestamp",
2636 "/snapshots/curve/timestamp",
2637 "/state/seen_at",
2638 ] {
2639 assert_eq!(
2640 token.pointer(path),
2641 Some(&json!(1_000)),
2642 "{before:?} {after:?} {path}: {token:#}"
2643 );
2644 }
2645 assert_eq!(
2646 token.pointer("/events/reserves/data/virtual_token_reserves"),
2647 Some(&json!(42))
2648 );
2649 assert_eq!(token.pointer("/events/reserves/slot"), Some(&json!(10)));
2650 assert_eq!(
2651 token.pointer("/snapshots/curve/signature"),
2652 Some(&json!("curve-signature"))
2653 );
2654 }
2655 }
2656
2657 fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
2658 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2659 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2660 state_name: "OreRound".to_string(),
2661 program_id: None,
2662 idl: None,
2663 identity: IdentitySpec {
2664 primary_keys: vec!["id.round_id".to_string()],
2665 lookup_indexes: vec![LookupIndexSpec {
2666 field_name: "id.round_address".to_string(),
2667 temporal_field: None,
2668 }],
2669 },
2670 handlers: vec![SerializableHandlerSpec {
2671 source: SourceSpec::Source {
2672 program_id: None,
2673 discriminator: None,
2674 type_name: "entropy::VarState".to_string(),
2675 serialization: None,
2676 is_account: true,
2677 },
2678 key_resolution: KeyResolutionStrategy::Lookup {
2679 primary_field: FieldPath::new(&["__account_address"]),
2680 },
2681 mappings: vec![mapping(
2682 "state.expires_at",
2683 &["end_at"],
2684 PopulationStrategy::LastWrite,
2685 )],
2686 conditions: vec![],
2687 emit: true,
2688 }],
2689 sections: vec![],
2690 field_mappings: BTreeMap::new(),
2691 resolver_hooks: vec![],
2692 instruction_hooks: vec![InstructionHook {
2693 instruction_type: "ore::DeployIxState".to_string(),
2694 actions: vec![HookAction::RegisterPdaMapping {
2695 pda_field: FieldPath::new(&["accounts", "entropyVar"]),
2696 seed_field: FieldPath::new(&["accounts", "round"]),
2697 lookup_name: "default_pda_lookup".to_string(),
2698 }],
2699 lookup_by: None,
2700 }],
2701 resolver_specs: vec![],
2702 computed_fields: vec![],
2703 computed_field_specs: vec![],
2704 content_hash: None,
2705 views: vec![],
2706 })
2707 }
2708
2709 fn embedded_account_key_spec() -> TypedStreamSpec<Value> {
2710 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2711 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2712 state_name: "OreTreasury".to_string(),
2713 program_id: None,
2714 idl: None,
2715 identity: IdentitySpec {
2716 primary_keys: vec!["id.address".to_string()],
2717 lookup_indexes: vec![],
2718 },
2719 handlers: vec![SerializableHandlerSpec {
2720 source: SourceSpec::Source {
2721 program_id: None,
2722 discriminator: None,
2723 type_name: "ore::TreasuryState".to_string(),
2724 serialization: None,
2725 is_account: true,
2726 },
2727 key_resolution: KeyResolutionStrategy::Embedded {
2728 primary_field: FieldPath::new(&["__account_address"]),
2729 },
2730 mappings: vec![mapping(
2731 "id.address",
2732 &["__account_address"],
2733 PopulationStrategy::SetOnce,
2734 )],
2735 conditions: vec![],
2736 emit: true,
2737 }],
2738 sections: vec![],
2739 field_mappings: BTreeMap::new(),
2740 resolver_hooks: vec![],
2741 instruction_hooks: vec![],
2742 resolver_specs: vec![],
2743 computed_fields: vec![],
2744 computed_field_specs: vec![],
2745 content_hash: None,
2746 views: vec![],
2747 })
2748 }
2749
2750 fn whole_entity_pool_spec() -> TypedStreamSpec<Value> {
2754 let mut hidden = mapping("state.secret", &["secret"], PopulationStrategy::LastWrite);
2755 hidden.emit = false;
2756 let handler = |type_name: &str, mappings, emit| SerializableHandlerSpec {
2757 source: SourceSpec::Source {
2758 program_id: None,
2759 discriminator: None,
2760 type_name: type_name.to_string(),
2761 serialization: None,
2762 is_account: true,
2763 },
2764 key_resolution: KeyResolutionStrategy::Embedded {
2765 primary_field: FieldPath::new(&["__account_address"]),
2766 },
2767 mappings,
2768 conditions: vec![],
2769 emit,
2770 };
2771 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2772 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2773 state_name: "Pool".to_string(),
2774 program_id: None,
2775 idl: None,
2776 identity: IdentitySpec {
2777 primary_keys: vec!["id.address".to_string()],
2778 lookup_indexes: vec![],
2779 },
2780 handlers: vec![
2781 handler(
2782 "amm::PoolState",
2783 vec![
2784 mapping(
2785 "id.address",
2786 &["__account_address"],
2787 PopulationStrategy::SetOnce,
2788 ),
2789 mapping("state.price", &["price"], PopulationStrategy::LastWrite),
2790 mapping("state.fills", &["fill"], PopulationStrategy::Append),
2791 hidden,
2792 ],
2793 true,
2794 ),
2795 handler(
2796 "amm::PoolConfigState",
2797 vec![mapping(
2798 "config.fee",
2799 &["fee"],
2800 PopulationStrategy::LastWrite,
2801 )],
2802 false,
2803 ),
2804 ],
2805 sections: vec![],
2806 field_mappings: BTreeMap::new(),
2807 resolver_hooks: vec![],
2808 instruction_hooks: vec![],
2809 resolver_specs: vec![],
2810 computed_fields: vec![],
2811 computed_field_specs: vec![],
2812 content_hash: None,
2813 views: vec![],
2814 })
2815 }
2816
2817 fn pool_bytecode() -> MultiEntityBytecode {
2818 MultiEntityBytecode::from_single("Pool".to_string(), whole_entity_pool_spec(), 0)
2819 }
2820
2821 #[test]
2825 fn compiled_event_capture_and_context_mappings_carry_the_update_timestamp() {
2826 let with_source = |target: &str, source: MappingSource| SerializableFieldMapping {
2827 source,
2828 ..mapping(target, &[], PopulationStrategy::LastWrite)
2829 };
2830 let spec: TypedStreamSpec<Value> =
2831 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2832 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2833 state_name: "Pool".to_string(),
2834 program_id: None,
2835 idl: None,
2836 identity: IdentitySpec {
2837 primary_keys: vec!["id.address".to_string()],
2838 lookup_indexes: vec![],
2839 },
2840 handlers: vec![SerializableHandlerSpec {
2841 source: SourceSpec::Source {
2842 program_id: None,
2843 discriminator: None,
2844 type_name: "amm::PoolState".to_string(),
2845 serialization: None,
2846 is_account: true,
2847 },
2848 key_resolution: KeyResolutionStrategy::Embedded {
2849 primary_field: FieldPath::new(&["__account_address"]),
2850 },
2851 mappings: vec![
2852 mapping(
2853 "id.address",
2854 &["__account_address"],
2855 PopulationStrategy::SetOnce,
2856 ),
2857 with_source(
2858 "events.priced",
2859 MappingSource::AsEvent {
2860 fields: vec![Box::new(MappingSource::FromSource {
2861 path: FieldPath::new(&["price"]),
2862 default: None,
2863 transform: None,
2864 })],
2865 },
2866 ),
2867 with_source(
2868 "snapshots.pool",
2869 MappingSource::AsCapture {
2870 field_transforms: BTreeMap::new(),
2871 },
2872 ),
2873 with_source(
2874 "state.updated_at",
2875 MappingSource::FromContext {
2876 field: "timestamp".to_string(),
2877 },
2878 ),
2879 ],
2880 conditions: vec![],
2881 emit: true,
2882 }],
2883 sections: vec![],
2884 field_mappings: BTreeMap::new(),
2885 resolver_hooks: vec![],
2886 instruction_hooks: vec![],
2887 resolver_specs: vec![],
2888 computed_fields: vec![],
2889 computed_field_specs: vec![],
2890 content_hash: None,
2891 views: vec![],
2892 });
2893 let bytecode = MultiEntityBytecode::from_single("Pool".to_string(), spec, 0);
2894 let context = crate::vm::UpdateContext::with_timestamp(7, "sig-7".to_string(), 1_234_567);
2895
2896 let mut vm = VmContext::new();
2897 vm.process_event(
2898 &bytecode,
2899 pool_event(100, 1),
2900 "amm::PoolState",
2901 Some(&context),
2902 None,
2903 )
2904 .unwrap();
2905
2906 let pool = vm.get_entity_state(0, &json!("pool_1")).unwrap();
2907 for path in [
2908 "/events/priced/timestamp",
2909 "/snapshots/pool/timestamp",
2910 "/state/updated_at",
2911 ] {
2912 assert_eq!(
2913 pool.pointer(path),
2914 Some(&json!(1_234_567)),
2915 "{path}: {pool:#}"
2916 );
2917 }
2918 assert_eq!(pool.pointer("/events/priced/data/price"), Some(&json!(100)));
2919 assert_eq!(pool.pointer("/snapshots/pool/slot"), Some(&json!(7)));
2920 }
2921
2922 fn pool_event(price: u64, fill: u64) -> Value {
2923 pool_event_for("pool_1", price, fill)
2924 }
2925
2926 fn pool_event_for(address: &str, price: u64, fill: u64) -> Value {
2927 json!({"__account_address": address, "price": price, "fill": fill, "secret": "s"})
2928 }
2929
2930 fn linked_vm(
2932 state_config: crate::vm::StateTableConfig,
2933 ) -> (VmContext, crate::vm::WholeEntityRequests) {
2934 let requests = crate::vm::WholeEntityRequests::new(16);
2935 let mut vm = VmContext::new_with_config(state_config);
2936 vm.set_whole_entity_requests(requests.clone());
2937 (vm, requests)
2938 }
2939
2940 fn process(vm: &mut VmContext, bytecode: &MultiEntityBytecode, event: Value) -> Vec<Mutation> {
2941 vm.process_event(bytecode, event, "amm::PoolState", None, None)
2942 .unwrap()
2943 }
2944
2945 #[test]
2946 fn source_deletion_cancels_resync_and_recreates_a_fresh_vm_row() {
2947 let bytecode = pool_bytecode();
2948 let (mut vm, requests) = linked_vm(Default::default());
2949 let first = process(&mut vm, &bytecode, pool_event(1, 10));
2950 let key = &first[0].key;
2951 let target = crate::vm::ResolverTarget {
2952 state_id: 0,
2953 entity_name: "Pool".into(),
2954 primary_key: key.clone(),
2955 extracts: vec![],
2956 };
2957 vm.enqueue_resolver_request(
2958 "ignored".into(),
2959 crate::ast::ResolverType::Token,
2960 json!("mint"),
2961 target.clone(),
2962 );
2963 let old_request = vm.take_resolver_requests().pop().unwrap();
2964 vm.get_state_table_mut(0).unwrap().deferred_when_ops.insert(
2965 ("sig".into(), "instruction".into()),
2966 vec![crate::vm::DeferredWhenOperation {
2967 entity_name: "Pool".into(),
2968 primary_key: key.clone(),
2969 field_path: "old".into(),
2970 field_value: json!(true),
2971 when_instruction: "instruction".into(),
2972 signature: "sig".into(),
2973 slot: 1,
2974 deferred_at: 1,
2975 emit: true,
2976 }],
2977 );
2978 requests.request("Pool", key);
2979 let deletion = vm.delete_entity(&bytecode, "Pool", key).unwrap();
2980 assert!(deletion.is_delete());
2981 assert!(
2982 serde_json::from_value::<Mutation>(serde_json::to_value(&deletion).unwrap())
2983 .unwrap()
2984 .is_delete()
2985 );
2986 assert!(!requests.is_requested("Pool", key));
2987 assert!(vm.get_entity_state(0, key).is_none());
2988 assert!(vm.snapshot_state_table(0).is_empty());
2989 assert!(vm
2990 .get_state_table_mut(0)
2991 .unwrap()
2992 .deferred_when_ops
2993 .is_empty());
2994 assert!(vm.delete_entity(&bytecode, "Unknown", key).is_none());
2995 let mut next = process(&mut vm, &bytecode, pool_event(9, 90));
2996 vm.enqueue_resolver_request(
2997 "ignored".into(),
2998 crate::ast::ResolverType::Token,
2999 json!("mint"),
3000 target,
3001 );
3002 let new_request = vm.take_resolver_requests().pop().unwrap();
3003 assert_ne!(old_request.cache_key, new_request.cache_key);
3004 assert!(vm
3005 .apply_resolver_result(&bytecode, &old_request.cache_key, json!({"stale":true}))
3006 .unwrap()
3007 .is_empty());
3008 assert_eq!(next[0].take_whole_entity_mark(), Some(WholeEntity::Created));
3009 assert_eq!(next[0].patch["state"]["fills"], json!([90]));
3010 }
3011
3012 #[test]
3013 fn deletion_preserves_unrelated_resolver_sharing_and_isolates_recreation() {
3014 let bytecode = pool_bytecode();
3015 let (mut vm, _) = linked_vm(Default::default());
3016 let key = process(&mut vm, &bytecode, pool_event(1, 10))[0]
3017 .key
3018 .clone();
3019 let late_joiner = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20))[0]
3020 .key
3021 .clone();
3022 let target = |primary_key| crate::vm::ResolverTarget {
3023 state_id: 0,
3024 entity_name: "Pool".into(),
3025 primary_key,
3026 extracts: vec![crate::ast::ResolverExtractSpec {
3027 target_path: "resolved".into(),
3028 source_path: Some("value".into()),
3029 transform: None,
3030 }],
3031 };
3032 let enqueue = |vm: &mut VmContext, key| {
3033 vm.enqueue_resolver_request(
3034 "ignored".into(),
3035 crate::ast::ResolverType::Token,
3036 json!("shared-mint"),
3037 target(key),
3038 )
3039 };
3040 enqueue(&mut vm, key.clone());
3041 enqueue(&mut vm, json!("unrelated"));
3042 let old = vm.take_resolver_requests();
3043 assert_eq!(old.len(), 1);
3044 vm.delete_entity(&bytecode, "Pool", &key).unwrap();
3045 enqueue(&mut vm, json!("unrelated"));
3046 enqueue(&mut vm, json!("another-unrelated"));
3047 assert!(
3048 vm.take_resolver_requests().is_empty(),
3049 "unaffected targets reuse the old pending call"
3050 );
3051 enqueue(&mut vm, key.clone());
3052 let fresh = vm.take_resolver_requests();
3053 assert_eq!(fresh.len(), 1);
3054 assert_ne!(fresh[0].cache_key, old[0].cache_key);
3055 vm.delete_entity(&bytecode, "Pool", &late_joiner).unwrap();
3058 process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
3059 enqueue(&mut vm, late_joiner.clone());
3060 let late_joiner_fresh = vm.take_resolver_requests();
3061 assert_eq!(late_joiner_fresh.len(), 1);
3062 assert_ne!(late_joiner_fresh[0].cache_key, old[0].cache_key);
3063 enqueue(&mut vm, json!("unrelated"));
3064 enqueue(&mut vm, json!("another-unrelated"));
3065 enqueue(&mut vm, key.clone());
3066 assert!(
3067 vm.take_resolver_requests().is_empty(),
3068 "unaffected targets reuse pending calls"
3069 );
3070 let old_mutations = vm
3071 .apply_resolver_result(&bytecode, &old[0].cache_key, json!({"value":"old"}))
3072 .unwrap();
3073 assert_eq!(old_mutations.len(), 2);
3074 assert!(old_mutations.iter().all(|mutation| mutation.key != key));
3075 assert!(vm.get_entity_state(0, &key).is_none());
3076 let fresh_mutations = vm
3077 .apply_resolver_result(&bytecode, &fresh[0].cache_key, json!({"value":"fresh"}))
3078 .unwrap();
3079 assert_eq!(fresh_mutations.len(), 1);
3080 assert_eq!(fresh_mutations[0].key, key);
3081 let late_joiner_mutations = vm
3082 .apply_resolver_result(
3083 &bytecode,
3084 &late_joiner_fresh[0].cache_key,
3085 json!({"value":"fresh-late-joiner"}),
3086 )
3087 .unwrap();
3088 assert_eq!(late_joiner_mutations.len(), 1);
3089 assert_eq!(late_joiner_mutations[0].key, late_joiner);
3090 }
3091
3092 #[test]
3093 fn a_requested_entity_is_emitted_whole_once() {
3094 let bytecode = pool_bytecode();
3095 let (mut vm, requests) = linked_vm(Default::default());
3096 let mut update = |price, fill| process(&mut vm, &bytecode, pool_event(price, fill));
3097
3098 let first = update(1, 10);
3099 assert_eq!(first[0].whole_entity_mark(), Some(WholeEntity::Created));
3100 let key = first[0].key.clone();
3101 let second = update(2, 20);
3102 assert_eq!(second.len(), 1);
3103 assert!(!second[0].is_whole_entity());
3104 assert_eq!(
3105 second[0].patch,
3106 json!({"state": {"price": 2, "fills": [20]}})
3107 );
3108 assert_eq!(second[0].append, vec!["state.fills".to_string()]);
3109
3110 requests.request("Pool", &key);
3112 assert!(requests.is_requested("Pool", &key));
3113 let mut third = update(3, 30);
3114 assert_eq!(third.len(), 2);
3115 assert!(
3116 !third[0].is_whole_entity(),
3117 "the change itself is untouched"
3118 );
3119 assert_eq!(
3120 third[0].patch,
3121 json!({"state": {"price": 3, "fills": [30]}})
3122 );
3123 assert_eq!(third[1].key, key);
3124 assert_eq!(third[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
3125 assert_eq!(
3126 third[1].patch,
3127 json!({
3128 "id": {"address": "pool_1"},
3129 "state": {"price": 3, "fills": [10, 20, 30]},
3130 }),
3131 "the whole entity, without its non-emitted field"
3132 );
3133 assert!(
3134 third[1].append.is_empty(),
3135 "its arrays are whole, not appended to"
3136 );
3137 assert!(requests.is_empty(), "a request is answered once");
3138
3139 let fourth = update(4, 40);
3140 assert_eq!(fourth.len(), 1);
3141 assert!(!fourth[0].is_whole_entity());
3142 assert_eq!(
3143 fourth[0].patch,
3144 json!({"state": {"price": 4, "fills": [40]}})
3145 );
3146 }
3147
3148 #[test]
3151 fn a_request_is_answered_by_the_next_call_whatever_it_changes() {
3152 let bytecode = pool_bytecode();
3153 let (mut vm, requests) = linked_vm(Default::default());
3154 process(&mut vm, &bytecode, pool_event_for("pool_1", 1, 10));
3155 process(&mut vm, &bytecode, pool_event_for("pool_2", 1, 10));
3156
3157 requests.request("Pool", &json!("pool_1"));
3158 let mut next = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20));
3159 assert_eq!(next.len(), 2);
3160 assert_eq!(next[0].key, json!("pool_2"));
3161 assert!(!next[0].is_whole_entity());
3162 assert_eq!(next[1].key, json!("pool_1"));
3163 assert_eq!(next[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
3164 assert_eq!(
3165 next[1].patch,
3166 json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
3167 );
3168 assert!(requests.is_empty());
3169
3170 requests.request("Pool", &json!("pool_2"));
3172 let answered = vm
3173 .apply_resolver_result(&bytecode, "no-such-request", json!({}))
3174 .unwrap();
3175 assert_eq!(answered.len(), 1);
3176 assert_eq!(answered[0].key, json!("pool_2"));
3177 assert_eq!(answered[0].whole_entity_mark(), Some(WholeEntity::Resent));
3178
3179 let quiet = process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
3180 assert_eq!(quiet.len(), 1, "a request is answered once");
3181 }
3182
3183 #[test]
3186 fn a_request_for_an_entity_the_vm_does_not_hold_is_dropped() {
3187 let bytecode = pool_bytecode();
3188 let (mut vm, requests) = linked_vm(Default::default());
3189 requests.request("Pool", &json!("other_pool"));
3190 let mutations = process(&mut vm, &bytecode, pool_event(1, 10));
3191 assert_eq!(mutations.len(), 1);
3192 assert_eq!(mutations[0].whole_entity_mark(), Some(WholeEntity::Created));
3193 assert!(requests.is_empty(), "dropped, not kept for later");
3194
3195 let created = process(&mut vm, &bytecode, pool_event_for("other_pool", 1, 10));
3196 assert_eq!(created.len(), 1);
3197 assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
3198 }
3199
3200 #[test]
3201 fn whole_entity_requests_are_bounded() {
3202 let requests = crate::vm::WholeEntityRequests::new(2);
3203 for key in ["a", "b", "c"] {
3204 requests.request("Pool", &json!(key));
3205 }
3206 assert_eq!(requests.len(), 2);
3207 assert!(
3208 !requests.is_requested("Pool", &json!("a")),
3209 "the oldest is forgotten"
3210 );
3211 assert!(requests.is_requested("Pool", &json!("c")));
3212 }
3213
3214 #[test]
3217 fn an_entitys_first_mutation_is_marked_as_its_creation() {
3218 let bytecode = pool_bytecode();
3219 let (mut vm, _requests) = linked_vm(Default::default());
3220
3221 let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
3222 assert_eq!(first.len(), 1, "the creation needs no second mutation");
3223 assert_eq!(
3224 first[0].take_whole_entity_mark(),
3225 Some(WholeEntity::Created)
3226 );
3227 assert_eq!(
3228 first[0].patch,
3229 json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}}),
3230 "the whole row, without its non-emitted field"
3231 );
3232 assert_eq!(first[0].append, vec!["state.fills".to_string()]);
3233
3234 let second = process(&mut vm, &bytecode, pool_event(2, 20));
3235 assert_eq!(second[0].whole_entity_mark(), None);
3236 assert_eq!(
3237 second[0].patch,
3238 json!({"state": {"price": 2, "fills": [20]}})
3239 );
3240 }
3241
3242 #[test]
3245 fn a_vm_without_whole_entity_requests_marks_no_creation() {
3246 let bytecode = pool_bytecode();
3247 let mut vm = VmContext::new();
3248 let first = process(&mut vm, &bytecode, pool_event(1, 10));
3249 assert!(!first[0].is_whole_entity());
3250 assert!(first[0].patch.get(crate::WHOLE_ENTITY_MARKER).is_none());
3251 }
3252
3253 #[test]
3257 fn a_row_started_without_a_mutation_is_created_by_its_first_one() {
3258 let bytecode = pool_bytecode();
3259 let (mut vm, _requests) = linked_vm(Default::default());
3260 let config = vm
3261 .process_event(
3262 &bytecode,
3263 json!({"__account_address": "pool_1", "fee": 30}),
3264 "amm::PoolConfigState",
3265 None,
3266 None,
3267 )
3268 .unwrap();
3269 assert!(config.is_empty(), "the config handler emits nothing");
3270
3271 let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
3272 assert_eq!(
3273 first[0].take_whole_entity_mark(),
3274 Some(WholeEntity::Created)
3275 );
3276 assert_eq!(
3277 first[0].patch,
3278 json!({
3279 "id": {"address": "pool_1"},
3280 "config": {"fee": 30},
3281 "state": {"price": 1, "fills": [10]},
3282 })
3283 );
3284 let second = process(&mut vm, &bytecode, pool_event(2, 20));
3285 assert!(!second[0].is_whole_entity());
3286 }
3287
3288 #[test]
3291 fn an_entity_the_vm_dropped_is_created_again() {
3292 let bytecode = pool_bytecode();
3293 let (mut vm, _requests) = linked_vm(crate::vm::StateTableConfig {
3294 max_entries: 2,
3295 ..Default::default()
3296 });
3297 for pool in ["a", "b", "c"] {
3298 let created = process(&mut vm, &bytecode, pool_event_for(pool, 1, 1));
3299 assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
3300 }
3301 assert!(vm.get_entity_state(0, &json!("a")).is_none(), "dropped");
3302
3303 let mut again = process(&mut vm, &bytecode, pool_event_for("a", 2, 2));
3304 assert_eq!(
3305 again[0].take_whole_entity_mark(),
3306 Some(WholeEntity::Created)
3307 );
3308 assert_eq!(
3309 again[0].patch,
3310 json!({"id": {"address": "a"}, "state": {"price": 2, "fills": [2]}})
3311 );
3312 let changed = process(&mut vm, &bytecode, pool_event_for("c", 2, 2));
3313 assert!(!changed[0].is_whole_entity(), "c was never dropped");
3314 }
3315
3316 #[test]
3321 fn a_restored_row_never_emitted_is_sent_whole_on_request() {
3322 let bytecode = pool_bytecode();
3323 let (mut vm, _requests) = linked_vm(Default::default());
3324 let config = vm
3325 .process_event(
3326 &bytecode,
3327 json!({"__account_address": "pool_1", "fee": 30}),
3328 "amm::PoolConfigState",
3329 None,
3330 None,
3331 )
3332 .unwrap();
3333 assert!(config.is_empty());
3334 let snapshot = vm.dump();
3335
3336 let (mut restored, requests) = linked_vm(Default::default());
3337 restored.hydrate(snapshot);
3338 let first = process(&mut restored, &bytecode, pool_event(1, 10));
3339 assert!(!first[0].is_whole_entity(), "counted as emitted already");
3340 assert_eq!(
3341 first[0].patch,
3342 json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
3343 );
3344
3345 requests.request("Pool", &first[0].key);
3346 let next = process(&mut restored, &bytecode, pool_event_for("pool_2", 1, 1));
3347 let resent = next
3348 .iter()
3349 .find(|mutation| mutation.key == json!("pool_1"))
3350 .expect("the requested entity follows the next call");
3351 assert_eq!(resent.whole_entity_mark(), Some(WholeEntity::Resent));
3352 let mut resent = resent.clone();
3353 resent.take_whole_entity_mark();
3354 assert_eq!(
3355 resent.patch,
3356 json!({
3357 "id": {"address": "pool_1"},
3358 "config": {"fee": 30},
3359 "state": {"price": 1, "fills": [10]},
3360 })
3361 );
3362 }
3363
3364 #[test]
3368 fn a_resolver_result_for_an_entity_not_held_creates_it() {
3369 let bytecode = pool_bytecode();
3370 let (mut vm, _requests) = linked_vm(Default::default());
3371 process(&mut vm, &bytecode, pool_event(1, 10));
3372 for key in ["pool_1", "pool_2"] {
3373 vm.enqueue_resolver_request(
3374 String::new(),
3375 crate::ast::ResolverType::Token,
3376 json!(key),
3377 crate::vm::ResolverTarget {
3378 state_id: 0,
3379 entity_name: "Pool".to_string(),
3380 primary_key: json!(key),
3381 extracts: vec![crate::ast::ResolverExtractSpec {
3382 target_path: "meta.name".to_string(),
3383 source_path: Some("name".to_string()),
3384 transform: None,
3385 }],
3386 },
3387 );
3388 }
3389 let mut applied = Vec::new();
3390 for request in vm.take_resolver_requests() {
3391 applied.extend(
3392 vm.apply_resolver_result(&bytecode, &request.cache_key, json!({"name": "n"}))
3393 .unwrap(),
3394 );
3395 }
3396 applied.sort_by_key(|mutation| mutation.key.to_string());
3397
3398 assert_eq!(applied[0].key, json!("pool_1"));
3399 assert!(!applied[0].is_whole_entity());
3400 assert_eq!(applied[0].patch, json!({"meta": {"name": "n"}}));
3401 assert_eq!(applied[1].key, json!("pool_2"));
3402 assert_eq!(
3403 applied[1].take_whole_entity_mark(),
3404 Some(WholeEntity::Created)
3405 );
3406 assert_eq!(applied[1].patch, json!({"meta": {"name": "n"}}));
3407 }
3408
3409 #[test]
3412 fn a_restored_entity_is_not_created_again() {
3413 let bytecode = pool_bytecode();
3414 let (mut vm, _requests) = linked_vm(Default::default());
3415 process(&mut vm, &bytecode, pool_event(1, 10));
3416 let snapshot = vm.dump();
3417
3418 let (mut restored, _requests) = linked_vm(Default::default());
3419 restored.hydrate(snapshot);
3420 let next = process(&mut restored, &bytecode, pool_event(2, 20));
3421 assert!(!next[0].is_whole_entity());
3422 assert_eq!(next[0].patch, json!({"state": {"price": 2, "fills": [20]}}));
3423 }
3424
3425 #[test]
3426 fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
3427 let bytecode = MultiEntityBytecode::from_single(
3428 "PumpfunToken".to_string(),
3429 direct_pda_to_primary_key_spec(),
3430 0,
3431 );
3432 let mut vm = VmContext::new();
3433
3434 let mutations = vm
3435 .process_event(
3436 &bytecode,
3437 json!({
3438 "__account_address": "bonding_curve_1",
3439 "__resolved_primary_key": "mint_1",
3440 "virtual_token_reserves": 42,
3441 }),
3442 "pump::BondingCurveState",
3443 None,
3444 None,
3445 )
3446 .unwrap();
3447
3448 assert_eq!(mutations.len(), 1);
3449 assert_eq!(mutations[0].key, json!("mint_1"));
3450 assert_eq!(
3451 mutations[0].patch["id"]["bonding_curve"],
3452 json!("bonding_curve_1")
3453 );
3454 assert_eq!(
3455 mutations[0].patch["reserves"]["virtual_token_reserves"],
3456 json!(42)
3457 );
3458 }
3459
3460 #[test]
3461 fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
3462 let bytecode = MultiEntityBytecode::from_single(
3463 "OreRound".to_string(),
3464 pda_to_intermediate_lookup_spec(),
3465 0,
3466 );
3467 let mut vm = VmContext::new();
3468
3469 let mutations = vm
3470 .process_event(
3471 &bytecode,
3472 json!({
3473 "__account_address": "entropy_var_1",
3474 "__resolved_primary_key": "round_address_1",
3475 "end_at": 123,
3476 }),
3477 "entropy::VarState",
3478 None,
3479 None,
3480 )
3481 .unwrap();
3482
3483 assert!(mutations.is_empty());
3484 }
3485
3486 #[test]
3487 fn embedded_account_key_wins_over_an_unrelated_resolved_key() {
3488 let bytecode = MultiEntityBytecode::from_single(
3489 "OreTreasury".to_string(),
3490 embedded_account_key_spec(),
3491 0,
3492 );
3493 let mut vm = VmContext::new();
3494
3495 let mutations = vm
3496 .process_event(
3497 &bytecode,
3498 json!({
3499 "__account_address": "treasury_pda",
3500 "__resolved_primary_key": "round_address",
3501 }),
3502 "ore::TreasuryState",
3503 None,
3504 None,
3505 )
3506 .unwrap();
3507
3508 assert_eq!(mutations.len(), 1);
3509 assert_eq!(mutations[0].key, json!("treasury_pda"));
3510 assert_eq!(mutations[0].patch["id"]["address"], json!("treasury_pda"));
3511 }
3512
3513 fn instruction_handler(
3514 type_name: &str,
3515 key_resolution: KeyResolutionStrategy,
3516 mappings: Vec<SerializableFieldMapping>,
3517 ) -> SerializableHandlerSpec {
3518 SerializableHandlerSpec {
3519 source: SourceSpec::Source {
3520 program_id: None,
3521 discriminator: None,
3522 type_name: type_name.to_string(),
3523 serialization: None,
3524 is_account: false,
3525 },
3526 key_resolution,
3527 mappings,
3528 conditions: vec![],
3529 emit: true,
3530 }
3531 }
3532
3533 fn split_position_spec() -> TypedStreamSpec<Value> {
3538 let embedded = |field: &str| KeyResolutionStrategy::Embedded {
3539 primary_field: FieldPath::new(&["accounts", field]),
3540 };
3541 TypedStreamSpec::from_serializable(SerializableStreamSpec {
3542 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3543 state_name: "Position".to_string(),
3544 program_id: None,
3545 idl: None,
3546 identity: IdentitySpec {
3547 primary_keys: vec!["id.position_address".to_string()],
3548 lookup_indexes: vec![],
3549 },
3550 handlers: vec![
3551 instruction_handler(
3552 "amm::SplitPositionIxState",
3553 embedded("first_position"),
3554 vec![
3555 mapping(
3556 "id.position_address",
3557 &["accounts", "first_position"],
3558 PopulationStrategy::SetOnce,
3559 ),
3560 mapping(
3561 "activity.split_source_count",
3562 &["data"],
3563 PopulationStrategy::Count,
3564 ),
3565 ],
3566 ),
3567 instruction_handler(
3568 "amm::SplitPositionIxState",
3569 embedded("second_position"),
3570 vec![
3571 mapping(
3572 "id.position_address",
3573 &["accounts", "second_position"],
3574 PopulationStrategy::SetOnce,
3575 ),
3576 mapping(
3577 "activity.split_child_count",
3578 &["data"],
3579 PopulationStrategy::Count,
3580 ),
3581 ],
3582 ),
3583 ],
3584 sections: vec![],
3585 field_mappings: BTreeMap::new(),
3586 resolver_hooks: vec![],
3587 instruction_hooks: vec![InstructionHook {
3588 instruction_type: "amm::SplitPositionIxState".to_string(),
3589 actions: vec![HookAction::SetField {
3590 target_field: "activity.last_split_source_slot".to_string(),
3591 source: MappingSource::FromSource {
3592 path: FieldPath::new(&["data", "slot"]),
3593 default: None,
3594 transform: None,
3595 },
3596 condition: None,
3597 }],
3598 lookup_by: Some(FieldPath::new(&["accounts", "first_position"])),
3599 }],
3600 resolver_specs: vec![],
3601 computed_fields: vec![],
3602 computed_field_specs: vec![],
3603 content_hash: None,
3604 views: vec![],
3605 })
3606 }
3607
3608 fn split_position_event(first: &str, second: &str) -> Value {
3609 json!({
3610 "accounts": { "first_position": first, "second_position": second },
3611 "data": { "slot": 77 },
3612 })
3613 }
3614
3615 #[test]
3616 fn one_instruction_routes_each_key_to_its_own_entity() {
3617 let bytecode =
3618 MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
3619 let mut vm = VmContext::new();
3620
3621 for _ in 0..2 {
3622 vm.process_event(
3623 &bytecode,
3624 split_position_event("source_pos", "child_pos"),
3625 "amm::SplitPositionIxState",
3626 None,
3627 None,
3628 )
3629 .unwrap();
3630 }
3631 vm.process_event(
3632 &bytecode,
3633 split_position_event("child_pos", "grandchild_pos"),
3634 "amm::SplitPositionIxState",
3635 None,
3636 None,
3637 )
3638 .unwrap();
3639
3640 let state = |key: &str| vm.get_entity_state(0, &json!(key)).unwrap();
3641
3642 let source = state("source_pos");
3643 assert_eq!(source["id"]["position_address"], json!("source_pos"));
3644 assert_eq!(source["activity"]["split_source_count"], json!(2));
3645 assert_eq!(source["activity"].get("split_child_count"), None);
3646 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3647
3648 let child = state("child_pos");
3650 assert_eq!(child["id"]["position_address"], json!("child_pos"));
3651 assert_eq!(child["activity"]["split_child_count"], json!(2));
3652 assert_eq!(child["activity"]["split_source_count"], json!(1));
3653 assert_eq!(child["activity"]["last_split_source_slot"], json!(77));
3654
3655 let grandchild = state("grandchild_pos");
3656 assert_eq!(grandchild["activity"]["split_child_count"], json!(1));
3657 assert_eq!(grandchild["activity"].get("split_source_count"), None);
3658 assert_eq!(grandchild["activity"].get("last_split_source_slot"), None);
3659 }
3660
3661 #[test]
3662 fn one_instruction_emits_one_mutation_per_key() {
3663 let bytecode =
3664 MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
3665 let mut vm = VmContext::new();
3666
3667 let mutations = vm
3668 .process_event(
3669 &bytecode,
3670 split_position_event("source_pos", "child_pos"),
3671 "amm::SplitPositionIxState",
3672 None,
3673 None,
3674 )
3675 .unwrap();
3676
3677 assert_eq!(mutations.len(), 2);
3678 assert_eq!(mutations[0].key, json!("source_pos"));
3679 assert_eq!(
3680 mutations[0].patch["activity"]["split_source_count"],
3681 json!(1)
3682 );
3683 assert_eq!(
3684 mutations[0].patch["activity"]["last_split_source_slot"],
3685 json!(77)
3686 );
3687 assert_eq!(
3688 mutations[0].patch["activity"].get("split_child_count"),
3689 None
3690 );
3691 assert_eq!(mutations[1].key, json!("child_pos"));
3692 assert_eq!(
3693 mutations[1].patch["activity"]["split_child_count"],
3694 json!(1)
3695 );
3696 assert_eq!(
3697 mutations[1].patch["activity"].get("split_source_count"),
3698 None
3699 );
3700 }
3701
3702 #[test]
3703 fn handlers_with_identical_keys_still_merge_into_one_segment() {
3704 let mut spec = split_position_spec();
3705 spec.instruction_hooks.clear();
3706 spec.handlers[1].key_resolution = KeyResolutionStrategy::Embedded {
3709 primary_field: FieldPath::new(&["accounts", "first_position"]),
3710 };
3711 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3712 let handler = &bytecode.entities["Position"].handlers["amm::SplitPositionIxState"];
3713
3714 assert!(!handler
3715 .iter()
3716 .any(|op| matches!(op, super::OpCode::SegmentBoundary)));
3717 assert_eq!(
3718 handler
3719 .iter()
3720 .filter(|op| matches!(op, super::OpCode::ReadOrInitState { .. }))
3721 .count(),
3722 1
3723 );
3724 }
3725
3726 #[test]
3727 fn hook_without_a_matching_key_gets_its_own_segment() {
3728 let mut spec = split_position_spec();
3729 spec.handlers.truncate(1);
3730 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
3731 primary_field: FieldPath::new(&["accounts", "second_position"]),
3732 };
3733 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
3734 "id.position_address",
3735 &["accounts", "second_position"],
3736 PopulationStrategy::SetOnce,
3737 ));
3738 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3739 let mut vm = VmContext::new();
3740
3741 vm.process_event(
3742 &bytecode,
3743 split_position_event("source_pos", "child_pos"),
3744 "amm::SplitPositionIxState",
3745 None,
3746 None,
3747 )
3748 .unwrap();
3749
3750 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3752 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3753 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
3754 assert_eq!(child["activity"].get("last_split_source_slot"), None);
3755 }
3756
3757 fn add_owner_index(spec: &mut TypedStreamSpec<Value>) {
3759 spec.identity.lookup_indexes.push(LookupIndexSpec {
3760 field_name: "id.owner".to_string(),
3761 temporal_field: None,
3762 });
3763 spec.handlers
3764 .push(crate::ast::TypedHandlerSpec::from_serializable(
3765 SerializableHandlerSpec {
3766 source: SourceSpec::Source {
3767 program_id: None,
3768 discriminator: None,
3769 type_name: "amm::PositionState".to_string(),
3770 serialization: None,
3771 is_account: true,
3772 },
3773 key_resolution: KeyResolutionStrategy::Embedded {
3774 primary_field: FieldPath::new(&["__account_address"]),
3775 },
3776 mappings: vec![
3777 mapping(
3778 "id.position_address",
3779 &["__account_address"],
3780 PopulationStrategy::SetOnce,
3781 ),
3782 mapping("id.owner", &["owner"], PopulationStrategy::SetOnce),
3783 ],
3784 conditions: vec![],
3785 emit: true,
3786 },
3787 ));
3788 }
3789
3790 fn index_owner(
3791 vm: &mut VmContext,
3792 bytecode: &MultiEntityBytecode,
3793 position: &str,
3794 owner: &str,
3795 ) {
3796 vm.process_event(
3797 bytecode,
3798 json!({ "__account_address": position, "owner": owner }),
3799 "amm::PositionState",
3800 None,
3801 None,
3802 )
3803 .unwrap();
3804 }
3805
3806 #[test]
3807 fn unmatched_hook_next_to_other_keys_resolves_lookup_index_fields() {
3808 let mut spec = split_position_spec();
3809 spec.handlers.truncate(1);
3810 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
3811 primary_field: FieldPath::new(&["accounts", "second_position"]),
3812 };
3813 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
3814 "id.position_address",
3815 &["accounts", "second_position"],
3816 PopulationStrategy::SetOnce,
3817 ));
3818 add_owner_index(&mut spec);
3819 spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
3821
3822 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3823 let mut vm = VmContext::new();
3824 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
3825 vm.process_event(
3826 &bytecode,
3827 json!({
3828 "accounts": {
3829 "first_position": "source_pos",
3830 "second_position": "child_pos",
3831 "owner": "owner_1",
3832 },
3833 "data": { "slot": 77 },
3834 }),
3835 "amm::SplitPositionIxState",
3836 None,
3837 None,
3838 )
3839 .unwrap();
3840
3841 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3842 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3843 assert!(vm.get_entity_state(0, &json!("owner_1")).is_none());
3844 }
3845
3846 #[test]
3847 fn unmatched_hook_replays_after_its_lookup_index_is_populated() {
3848 let mut spec = split_position_spec();
3849 spec.handlers.truncate(1);
3850 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
3851 primary_field: FieldPath::new(&["accounts", "second_position"]),
3852 };
3853 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
3854 "id.position_address",
3855 &["accounts", "second_position"],
3856 PopulationStrategy::SetOnce,
3857 ));
3858 add_owner_index(&mut spec);
3859 spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
3860
3861 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3862 let mut vm = VmContext::new();
3863 let mutations = vm
3864 .process_event(
3865 &bytecode,
3866 json!({
3867 "accounts": { "owner": "owner_1", "second_position": "child_pos" },
3868 "data": { "slot": 77 },
3869 }),
3870 "amm::SplitPositionIxState",
3871 None,
3872 None,
3873 )
3874 .unwrap();
3875 assert_eq!(mutations.len(), 1);
3876 assert_eq!(mutations[0].key, json!("child_pos"));
3877
3878 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
3879
3880 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3881 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3882 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
3883 assert_eq!(child["activity"].get("last_split_source_slot"), None);
3884 }
3885
3886 #[test]
3887 fn a_segment_that_misses_its_key_is_replayed_alone() {
3888 let mut spec = split_position_spec();
3889 spec.instruction_hooks.clear();
3890 spec.handlers.swap(0, 1);
3893 spec.handlers[1].key_resolution = KeyResolutionStrategy::Lookup {
3894 primary_field: FieldPath::new(&["accounts", "source_pda"]),
3895 };
3896 spec.handlers[1].mappings.remove(0);
3897 add_owner_index(&mut spec);
3898 spec.instruction_hooks.push(InstructionHook {
3899 instruction_type: "amm::RegisterIxState".to_string(),
3900 actions: vec![HookAction::RegisterPdaMapping {
3901 pda_field: FieldPath::new(&["accounts", "source_pda"]),
3902 seed_field: FieldPath::new(&["accounts", "owner"]),
3903 lookup_name: "default_pda_lookup".to_string(),
3904 }],
3905 lookup_by: None,
3906 });
3907 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3908 let mut vm = VmContext::new();
3909 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
3910
3911 let mutations = vm
3912 .process_event(
3913 &bytecode,
3914 json!({
3915 "accounts": { "source_pda": "pda_1", "second_position": "child_pos" },
3916 "data": {},
3917 }),
3918 "amm::SplitPositionIxState",
3919 None,
3920 None,
3921 )
3922 .unwrap();
3923 assert_eq!(mutations.len(), 1);
3925 assert_eq!(mutations[0].key, json!("child_pos"));
3926
3927 let mutations = vm
3928 .process_event(
3929 &bytecode,
3930 json!({ "accounts": { "source_pda": "pda_1", "owner": "owner_1" } }),
3931 "amm::RegisterIxState",
3932 None,
3933 None,
3934 )
3935 .unwrap();
3936 assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
3938 assert_eq!(mutations[0].key, json!("source_pos"));
3939 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3940 assert_eq!(source["activity"]["split_source_count"], json!(1));
3941 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
3942 assert_eq!(child["activity"]["split_child_count"], json!(1));
3943 }
3944
3945 #[test]
3950 fn a_hook_whose_lookup_field_is_absent_skips_only_its_segment() {
3951 let spec: TypedStreamSpec<Value> =
3952 TypedStreamSpec::from_serializable(SerializableStreamSpec {
3953 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3954 state_name: "Game".to_string(),
3955 program_id: None,
3956 idl: None,
3957 identity: IdentitySpec {
3958 primary_keys: vec!["id.game_id".to_string()],
3959 lookup_indexes: vec![],
3960 },
3961 handlers: vec![instruction_handler(
3962 "game::StartGameIxState",
3963 KeyResolutionStrategy::Embedded {
3964 primary_field: FieldPath::new(&["data", "game_id"]),
3965 },
3966 vec![
3967 SerializableFieldMapping {
3968 source: MappingSource::FromSource {
3969 path: FieldPath::new(&["data", "game_id"]),
3970 default: None,
3971 transform: Some(crate::ast::Transformation::HexEncode),
3972 },
3973 ..mapping("id.game_id", &[], PopulationStrategy::SetOnce)
3974 },
3975 mapping("stats.fee", &["data", "fee"], PopulationStrategy::SetOnce),
3976 ],
3977 )],
3978 sections: vec![],
3979 field_mappings: BTreeMap::new(),
3980 resolver_hooks: vec![],
3981 instruction_hooks: vec![InstructionHook {
3982 instruction_type: "game::StartGameIxState".to_string(),
3983 actions: vec![HookAction::SetField {
3984 target_field: "stats.started_slot".to_string(),
3985 source: MappingSource::FromSource {
3986 path: FieldPath::new(&["data", "slot"]),
3987 default: None,
3988 transform: None,
3989 },
3990 condition: None,
3991 }],
3992 lookup_by: Some(FieldPath::new(&["accounts", "game_id"])),
3994 }],
3995 resolver_specs: vec![],
3996 computed_fields: vec![],
3997 computed_field_specs: vec![],
3998 content_hash: None,
3999 views: vec![],
4000 });
4001 let bytecode = MultiEntityBytecode::from_single("Game".to_string(), spec, 0);
4002 let mut vm = VmContext::new();
4003
4004 let mutations = vm
4005 .process_event(
4006 &bytecode,
4007 json!({
4008 "accounts": { "game_account": "game_pda" },
4009 "data": { "game_id": [1, 2, 3, 4], "fee": 5, "slot": 9 },
4010 }),
4011 "game::StartGameIxState",
4012 None,
4013 None,
4014 )
4015 .unwrap();
4016
4017 assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
4018 assert_eq!(mutations[0].key, json!("01020304"));
4019 let game = vm.get_entity_state(0, &json!("01020304")).unwrap();
4020 assert_eq!(game["id"]["game_id"], json!("01020304"));
4021 assert_eq!(game["stats"]["fee"], json!(5));
4022 assert_eq!(game["stats"].get("started_slot"), None);
4023 assert_eq!(vm.get_entity_state(0, &json!(null)), None);
4025 }
4026
4027 mod fingerprint {
4028 use super::super::{EntityBytecode, MultiEntityBytecode, OpCode};
4029 use crate::ast::FieldPath;
4030 use serde_json::json;
4031 use std::collections::{HashMap, HashSet};
4032
4033 fn sample_handler() -> Vec<OpCode> {
4034 vec![
4035 OpCode::LoadEventField {
4036 path: FieldPath::new(&["id", "mint"]),
4037 dest: 0,
4038 default: Some(json!({"b": 2, "a": 1})),
4039 },
4040 OpCode::CreateObject { dest: 1 },
4041 OpCode::SetField {
4042 object: 1,
4043 path: "state.volume".to_string(),
4044 value: 0,
4045 },
4046 OpCode::EmitMutation {
4047 entity_name: "PumpfunToken".to_string(),
4048 key: 0,
4049 state: 1,
4050 },
4051 ]
4052 }
4053
4054 fn bytecode(handler_insert_order: &[&str]) -> MultiEntityBytecode {
4055 let mut handlers = HashMap::new();
4056 for event_type in handler_insert_order {
4057 handlers.insert(event_type.to_string(), sample_handler());
4058 }
4059 let mut entities = HashMap::new();
4060 entities.insert(
4061 "PumpfunToken".to_string(),
4062 EntityBytecode {
4063 state_id: 0,
4064 handlers,
4065 entity_name: "PumpfunToken".to_string(),
4066 when_events: HashSet::from(["pump::BuyIxState".to_string()]),
4067 non_emitted_fields: HashSet::new(),
4068 computed_paths: vec!["state.market_cap".to_string()],
4069 computed_fields_evaluator: None,
4070 },
4071 );
4072 let mut event_routing = HashMap::new();
4073 event_routing.insert(
4074 "pump::TokenState".to_string(),
4075 vec!["PumpfunToken".to_string()],
4076 );
4077 MultiEntityBytecode {
4078 entities,
4079 event_routing,
4080 when_events: HashSet::from(["pump::BuyIxState".to_string()]),
4081 proto_router: crate::proto_router::ProtoRouter::new(),
4082 }
4083 }
4084
4085 #[test]
4086 fn stable_across_recomputation_and_insertion_order() {
4087 let a = bytecode(&["pump::TokenState", "pump::BuyIxState"]);
4088 let b = bytecode(&["pump::BuyIxState", "pump::TokenState"]);
4089 assert_eq!(a.fingerprint(), a.fingerprint());
4090 assert_eq!(a.fingerprint(), b.fingerprint());
4091 assert_eq!(a.fingerprint().len(), 64);
4092 }
4093
4094 #[test]
4095 fn changes_when_an_opcode_operand_changes() {
4096 let base = bytecode(&["pump::TokenState"]);
4097 let mut modified = bytecode(&["pump::TokenState"]);
4098 let entity = modified.entities.get_mut("PumpfunToken").unwrap();
4099 match &mut entity.handlers.get_mut("pump::TokenState").unwrap()[2] {
4100 OpCode::SetField { path, .. } => *path = "state.volume_usd".to_string(),
4101 other => panic!("unexpected opcode: {other:?}"),
4102 }
4103 assert_ne!(base.fingerprint(), modified.fingerprint());
4104 }
4105
4106 #[test]
4107 fn changes_when_routing_changes() {
4108 let base = bytecode(&["pump::TokenState"]);
4109 let mut modified = bytecode(&["pump::TokenState"]);
4110 modified.event_routing.insert(
4111 "pump::SellIxState".to_string(),
4112 vec!["PumpfunToken".to_string()],
4113 );
4114 assert_ne!(base.fingerprint(), modified.fingerprint());
4115 }
4116
4117 #[test]
4124 fn golden_hash_pins_the_canonical_encoding() {
4125 let expected = "acf7ea8bf39ca830a9e6c2b2d2fb471da68e57e24bbb4c256964a42a498f75f2";
4126 assert_eq!(bytecode(&["pump::TokenState"]).fingerprint(), expected);
4127 }
4128 }
4129}