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