1use crate::ast::*;
2use serde::Serialize;
3use serde_json::Value;
4use std::collections::{HashMap, HashSet};
5use tracing;
6
7pub type Register = usize;
8
9const GUARD_REG: Register = 30;
12
13fn stop_field_path(target_path: &str) -> String {
14 format!("__stop:{}", target_path)
15}
16
17struct HandlerSegment {
19 key_identity: Option<Value>,
23 route_field: Option<FieldPath>,
26 ops: Vec<OpCode>,
27}
28
29fn route_field(resolution: &KeyResolutionStrategy) -> Option<FieldPath> {
31 let field = match resolution {
32 KeyResolutionStrategy::Embedded { primary_field }
33 | KeyResolutionStrategy::Lookup { primary_field, .. }
34 | KeyResolutionStrategy::Computed { primary_field, .. } => primary_field,
35 KeyResolutionStrategy::TemporalLookup { lookup_field, .. } => lookup_field,
36 };
37 (!field.segments.is_empty()).then(|| field.clone())
38}
39
40fn merge_handler_opcodes(existing: &mut Vec<OpCode>, new: &[OpCode]) {
44 let mut existing_setup = Vec::new();
46 let mut existing_mappings = Vec::new();
47 let mut existing_teardown = Vec::new();
48 let mut section = 0; for opcode in existing.iter() {
51 match opcode {
52 OpCode::ReadOrInitState { .. } => {
53 existing_setup.push(opcode.clone());
54 section = 1; }
56 OpCode::UpdateState { .. } => {
57 existing_teardown.push(opcode.clone());
58 section = 2; }
60 OpCode::EmitMutation { .. } => {
61 existing_teardown.push(opcode.clone());
62 }
63 _ if section == 0 => existing_setup.push(opcode.clone()),
64 _ if section == 1 => existing_mappings.push(opcode.clone()),
65 _ => existing_teardown.push(opcode.clone()),
66 }
67 }
68
69 let mut new_mappings = Vec::new();
71 section = 0;
72
73 for opcode in new.iter() {
74 match opcode {
75 OpCode::ReadOrInitState { .. } => {
76 section = 1; }
78 OpCode::UpdateState { .. } | OpCode::EmitMutation { .. } => {
79 section = 2; }
81 _ if section == 1 => {
82 new_mappings.push(opcode.clone());
83 }
84 _ => {} }
86 }
87
88 let mut merged = Vec::new();
90 merged.extend(existing_setup);
91 merged.extend(existing_mappings);
92 merged.extend(new_mappings);
93 merged.extend(existing_teardown);
94
95 *existing = merged;
96}
97
98fn is_false(value: &bool) -> bool {
99 !*value
100}
101
102fn index_guard<S>(mappings: &[TypedFieldMapping<S>], field: &str) -> Option<ComputedExpr> {
106 let mut guards = mappings
107 .iter()
108 .filter(|mapping| mapping.target_path == field)
109 .map(
110 |mapping| match mapping.condition.as_ref()?.parsed.as_ref()? {
111 ParsedCondition::Guard { expr } => Some(expr.clone()),
112 _ => None,
113 },
114 )
115 .collect::<Option<Vec<_>>>()?
116 .into_iter();
117 let first = guards.next()?;
118 Some(guards.fold(first, |left, right| ComputedExpr::Binary {
119 op: BinaryOp::Or,
120 left: Box::new(left),
121 right: Box::new(right),
122 }))
123}
124
125fn declared_index(resolution: &KeyResolutionStrategy) -> Option<&str> {
127 match resolution {
128 KeyResolutionStrategy::Lookup {
129 index_name: Some(index_name),
130 ..
131 } => Some(index_name),
132 _ => None,
133 }
134}
135
136#[derive(Debug, Clone, Serialize)]
137pub enum OpCode {
138 AbortIfNullKey {
145 key: Register,
146 is_account_event: bool,
147 },
148 EvaluateEventExpr {
152 expr: ComputedExpr,
153 transform: Option<Transformation>,
154 dest: Register,
155 },
156 LoadEventField {
157 path: FieldPath,
158 dest: Register,
159 default: Option<Value>,
160 },
161 LoadConstant {
162 value: Value,
163 dest: Register,
164 },
165 CopyRegister {
166 source: Register,
167 dest: Register,
168 },
169 CopyRegisterIfNull {
171 source: Register,
172 dest: Register,
173 },
174 GetEventType {
175 dest: Register,
176 },
177 CreateObject {
178 dest: Register,
179 },
180 SetField {
181 object: Register,
182 path: String,
183 value: Register,
184 },
185 SetFields {
186 object: Register,
187 fields: Vec<(String, Register)>,
188 },
189 GetField {
190 object: Register,
191 path: String,
192 dest: Register,
193 },
194 ReadOrInitState {
195 state_id: u32,
196 key: Register,
197 default: Value,
198 dest: Register,
199 #[serde(skip_serializing_if = "Option::is_none")]
205 dedup_route: Option<String>,
206 },
207 UpdateState {
208 state_id: u32,
209 key: Register,
210 value: Register,
211 },
212 AppendToArray {
213 object: Register,
214 path: String,
215 value: Register,
216 },
217 GetCurrentTimestamp {
221 dest: Register,
222 },
223 CreateEvent {
226 dest: Register,
227 event_value: Register,
228 },
229 CreateCapture {
232 dest: Register,
233 capture_value: Register,
234 },
235 Transform {
236 source: Register,
237 dest: Register,
238 transformation: Transformation,
239 },
240 EmitMutation {
241 entity_name: String,
242 key: Register,
243 state: Register,
244 },
245 SetFieldIfNull {
246 object: Register,
247 path: String,
248 value: Register,
249 },
250 SetFieldMax {
251 object: Register,
252 path: String,
253 value: Register,
254 },
255 UpdateTemporalIndex {
256 state_id: u32,
257 index_name: String,
258 lookup_value: Register,
259 primary_key: Register,
260 timestamp: Register,
261 },
262 LookupTemporalIndex {
263 state_id: u32,
264 index_name: String,
265 lookup_value: Register,
266 timestamp: Register,
267 dest: Register,
268 },
269 UpdateLookupIndex {
270 state_id: u32,
271 index_name: String,
272 lookup_value: Register,
273 primary_key: Register,
274 },
275 LookupIndex {
276 state_id: u32,
277 index_name: String,
278 lookup_value: Register,
279 dest: Register,
280 #[serde(skip_serializing_if = "is_false")]
283 strict: bool,
284 },
285 SetFieldSum {
287 object: Register,
288 path: String,
289 value: Register,
290 },
291 SetFieldIncrement {
293 object: Register,
294 path: String,
295 },
296 SetFieldMin {
298 object: Register,
299 path: String,
300 value: Register,
301 },
302 SetFieldWhen {
305 object: Register,
306 path: String,
307 value: Register,
308 when_instruction: String,
309 entity_name: String,
310 key_reg: Register,
311 condition_field: Option<FieldPath>,
312 condition_op: Option<ComparisonOp>,
313 condition_value: Option<Value>,
314 },
315 SetFieldUnlessStopped {
318 object: Register,
319 path: String,
320 value: Register,
321 stop_field: String,
322 stop_instruction: String,
323 entity_name: String,
324 key_reg: Register,
325 },
326 AddToUniqueSet {
329 state_id: u32,
330 set_name: String,
331 value: Register,
332 count_object: Register,
333 count_path: String,
334 },
335 ConditionalSetField {
337 object: Register,
338 path: String,
339 value: Register,
340 condition_field: FieldPath,
341 condition_op: ComparisonOp,
342 condition_value: Value,
343 },
344 ConditionalIncrement {
346 object: Register,
347 path: String,
348 condition_field: FieldPath,
349 condition_op: ComparisonOp,
350 condition_value: Value,
351 },
352 EvaluateComputedFields {
355 state: Register,
356 computed_paths: Vec<String>,
357 },
358 QueueResolver {
360 state_id: u32,
361 entity_name: String,
362 resolver: ResolverType,
363 input_path: Option<String>,
364 input_value: Option<Value>,
365 url_template: Option<Vec<UrlTemplatePart>>,
366 strategy: ResolveStrategy,
367 extracts: Vec<ResolverExtractSpec>,
368 condition: Option<ResolverCondition>,
369 schedule_at: Option<String>,
370 state: Register,
371 key: Register,
372 },
373 UpdatePdaReverseLookup {
376 state_id: u32,
377 lookup_name: String,
378 pda_address: Register,
379 primary_key: Register,
380 },
381 SkipUnless {
387 predicate: Register,
388 count: usize,
389 },
390 CaptureSibling {
394 state_id: u32,
395 paths: Vec<FieldPath>,
396 },
397 RegisterPdaFromSibling {
403 state_id: u32,
404 lookup_name: String,
405 sibling: SiblingSelector,
406 pda_field: FieldPath,
407 seed_field: FieldPath,
408 pda_address: Register,
409 primary_key: Register,
410 },
411 SegmentBoundary,
420}
421
422pub struct EntityBytecode {
423 pub state_id: u32,
424 pub handlers: HashMap<String, Vec<OpCode>>,
425 pub entity_name: String,
426 pub when_events: HashSet<String>,
427 pub non_emitted_fields: HashSet<String>,
428 pub computed_paths: Vec<String>,
429 #[allow(clippy::type_complexity)]
432 pub computed_fields_evaluator: Option<
433 Box<
434 dyn Fn(
435 &mut Value,
436 Option<u64>,
437 i64,
438 )
439 -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
440 + Send
441 + Sync,
442 >,
443 >,
444}
445
446impl std::fmt::Debug for EntityBytecode {
447 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
448 f.debug_struct("EntityBytecode")
449 .field("state_id", &self.state_id)
450 .field("handlers", &self.handlers)
451 .field("entity_name", &self.entity_name)
452 .field("when_events", &self.when_events)
453 .field("non_emitted_fields", &self.non_emitted_fields)
454 .field("computed_paths", &self.computed_paths)
455 .field(
456 "computed_fields_evaluator",
457 &self.computed_fields_evaluator.is_some(),
458 )
459 .finish()
460 }
461}
462
463#[derive(Debug)]
464pub struct MultiEntityBytecode {
465 pub entities: HashMap<String, EntityBytecode>,
466 pub event_routing: HashMap<String, Vec<String>>,
467 pub when_events: HashSet<String>,
468 pub proto_router: crate::proto_router::ProtoRouter,
469}
470
471impl MultiEntityBytecode {
472 pub fn from_single<S>(entity_name: String, spec: TypedStreamSpec<S>, state_id: u32) -> Self {
473 let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
474 let entity_bytecode = compiler.compile_entity();
475
476 let mut entities = HashMap::new();
477 let mut event_routing = HashMap::new();
478 let mut when_events = HashSet::new();
479
480 for event_type in entity_bytecode.handlers.keys() {
481 event_routing
482 .entry(event_type.clone())
483 .or_insert_with(Vec::new)
484 .push(entity_name.clone());
485 }
486
487 when_events.extend(entity_bytecode.when_events.iter().cloned());
488
489 entities.insert(entity_name, entity_bytecode);
490
491 MultiEntityBytecode {
492 entities,
493 event_routing,
494 when_events,
495 proto_router: crate::proto_router::ProtoRouter::new(),
496 }
497 }
498
499 pub fn from_entities(entities_vec: Vec<(String, Box<dyn std::any::Any>, u32)>) -> Self {
500 let entities = HashMap::new();
501 let event_routing = HashMap::new();
502 let when_events = HashSet::new();
503
504 if let Some((_entity_name, _spec_any, _state_id)) = entities_vec.into_iter().next() {
505 panic!("from_entities requires type information - use builder pattern instead");
506 }
507
508 MultiEntityBytecode {
509 entities,
510 event_routing,
511 when_events,
512 proto_router: crate::proto_router::ProtoRouter::new(),
513 }
514 }
515
516 #[allow(clippy::new_ret_no_self)]
517 pub fn new() -> MultiEntityBytecodeBuilder {
518 MultiEntityBytecodeBuilder {
519 entities: HashMap::new(),
520 event_routing: HashMap::new(),
521 when_events: HashSet::new(),
522 proto_router: crate::proto_router::ProtoRouter::new(),
523 }
524 }
525
526 pub fn fingerprint(&self) -> String {
543 use sha2::{Digest, Sha256};
544
545 fn update_sorted_set(hasher: &mut Sha256, set: &HashSet<String>) {
546 let mut items: Vec<&String> = set.iter().collect();
547 items.sort();
548 for item in items {
549 hasher.update(item.as_bytes());
550 hasher.update([0u8]);
551 }
552 hasher.update([1u8]);
553 }
554
555 fn update_canonical_json(hasher: &mut Sha256, value: &Value) {
562 match value {
563 Value::Object(map) => {
564 let mut keys: Vec<&String> = map.keys().collect();
565 keys.sort();
566 hasher.update(*b"{");
567 for key in keys {
568 hasher.update(key.as_bytes());
569 hasher.update([0u8]);
570 update_canonical_json(hasher, &map[key]);
571 }
572 hasher.update(*b"}");
573 }
574 Value::Array(items) => {
575 hasher.update(*b"[");
576 for item in items {
577 update_canonical_json(hasher, item);
578 hasher.update([0u8]);
579 }
580 hasher.update(*b"]");
581 }
582 scalar => {
585 hasher.update(scalar.to_string().as_bytes());
586 hasher.update([0u8]);
587 }
588 }
589 }
590
591 let mut hasher = Sha256::new();
592
593 let mut entity_names: Vec<&String> = self.entities.keys().collect();
594 entity_names.sort();
595 for entity_name in entity_names {
596 let entity = &self.entities[entity_name];
597 hasher.update(entity_name.as_bytes());
598 hasher.update([0u8]);
599 hasher.update(entity.state_id.to_le_bytes());
600
601 let mut event_types: Vec<&String> = entity.handlers.keys().collect();
602 event_types.sort();
603 for event_type in event_types {
604 hasher.update(event_type.as_bytes());
605 hasher.update([0u8]);
606 let ops = serde_json::to_value(&entity.handlers[event_type]).expect(
607 "opcode stream must serialize for fingerprinting; \
608 a non-serializable OpCode payload breaks snapshot invalidation",
609 );
610 update_canonical_json(&mut hasher, &ops);
611 hasher.update([0u8]);
612 }
613 hasher.update([1u8]);
614
615 update_sorted_set(&mut hasher, &entity.when_events);
616 update_sorted_set(&mut hasher, &entity.non_emitted_fields);
617 for path in &entity.computed_paths {
618 hasher.update(path.as_bytes());
619 hasher.update([0u8]);
620 }
621 hasher.update([entity.computed_fields_evaluator.is_some() as u8]);
622 }
623
624 let mut routes: Vec<(&String, &Vec<String>)> = self.event_routing.iter().collect();
625 routes.sort_by_key(|(event_type, _)| *event_type);
626 for (event_type, entity_names) in routes {
627 hasher.update(event_type.as_bytes());
628 hasher.update([0u8]);
629 let mut sorted_names = entity_names.clone();
630 sorted_names.sort();
631 for name in sorted_names {
632 hasher.update(name.as_bytes());
633 hasher.update([0u8]);
634 }
635 hasher.update([1u8]);
636 }
637
638 update_sorted_set(&mut hasher, &self.when_events);
639
640 hex::encode(hasher.finalize())
641 }
642}
643
644pub struct MultiEntityBytecodeBuilder {
645 entities: HashMap<String, EntityBytecode>,
646 event_routing: HashMap<String, Vec<String>>,
647 when_events: HashSet<String>,
648 proto_router: crate::proto_router::ProtoRouter,
649}
650
651impl MultiEntityBytecodeBuilder {
652 pub fn add_entity<S>(
653 self,
654 entity_name: String,
655 spec: TypedStreamSpec<S>,
656 state_id: u32,
657 ) -> Self {
658 self.add_entity_with_evaluator(
659 entity_name,
660 spec,
661 state_id,
662 None::<
663 fn(
664 &mut Value,
665 Option<u64>,
666 i64,
667 )
668 -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>,
669 >,
670 )
671 }
672
673 pub fn add_entity_with_evaluator<S, F>(
674 mut self,
675 entity_name: String,
676 spec: TypedStreamSpec<S>,
677 state_id: u32,
678 evaluator: Option<F>,
679 ) -> Self
680 where
681 F: Fn(
682 &mut Value,
683 Option<u64>,
684 i64,
685 ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
686 + Send
687 + Sync
688 + 'static,
689 {
690 let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
691 let mut entity_bytecode = compiler.compile_entity();
692
693 if let Some(eval) = evaluator {
695 entity_bytecode.computed_fields_evaluator = Some(Box::new(eval));
696 }
697
698 for event_type in entity_bytecode.handlers.keys() {
699 self.event_routing
700 .entry(event_type.clone())
701 .or_default()
702 .push(entity_name.clone());
703 }
704
705 self.when_events
706 .extend(entity_bytecode.when_events.iter().cloned());
707
708 self.entities.insert(entity_name, entity_bytecode);
709 self
710 }
711
712 pub fn build(self) -> MultiEntityBytecode {
713 MultiEntityBytecode {
714 entities: self.entities,
715 event_routing: self.event_routing,
716 when_events: self.when_events,
717 proto_router: self.proto_router,
718 }
719 }
720}
721
722pub struct TypedCompiler<S> {
723 pub spec: TypedStreamSpec<S>,
724 entity_name: String,
725 state_id: u32,
726}
727
728impl<S> TypedCompiler<S> {
729 pub fn new(spec: TypedStreamSpec<S>, entity_name: String) -> Self {
730 TypedCompiler {
731 spec,
732 entity_name,
733 state_id: 0,
734 }
735 }
736
737 pub fn with_state_id(mut self, state_id: u32) -> Self {
738 self.state_id = state_id;
739 self
740 }
741
742 pub fn compile(&self) -> MultiEntityBytecode {
743 let entity_bytecode = self.compile_entity();
744
745 let mut entities = HashMap::new();
746 let mut event_routing = HashMap::new();
747 let mut when_events = HashSet::new();
748
749 for event_type in entity_bytecode.handlers.keys() {
750 event_routing
751 .entry(event_type.clone())
752 .or_insert_with(Vec::new)
753 .push(self.entity_name.clone());
754 }
755
756 when_events.extend(entity_bytecode.when_events.iter().cloned());
757
758 entities.insert(self.entity_name.clone(), entity_bytecode);
759
760 MultiEntityBytecode {
761 entities,
762 event_routing,
763 when_events,
764 proto_router: crate::proto_router::ProtoRouter::new(),
765 }
766 }
767
768 fn resolver_outputs_primary_key_directly(&self, primary_field: &FieldPath) -> bool {
769 if primary_field.segments.as_slice() != ["__account_address"] {
770 return false;
771 }
772
773 let primary_key_leafs: HashSet<&str> = self
774 .spec
775 .identity
776 .primary_keys
777 .iter()
778 .map(|path| path.rsplit('.').next().unwrap_or(path.as_str()))
779 .collect();
780
781 self.spec.instruction_hooks.iter().any(|hook| {
782 hook.actions.iter().any(|action| {
783 matches!(action, HookAction::RegisterPdaMapping { seed_field, .. }
784 if seed_field
785 .segments
786 .last()
787 .is_some_and(|segment| primary_key_leafs.contains(segment.as_str())))
788 })
789 })
790 }
791
792 fn compile_entity(&self) -> EntityBytecode {
793 let mut when_events: HashSet<String> = HashSet::new();
794 let mut emit_by_path: HashMap<String, bool> = HashMap::new();
795
796 let mut debug_info = Vec::new();
798 for (index, handler_spec) in self.spec.handlers.iter().enumerate() {
799 let event_type = self.get_event_type(&handler_spec.source);
800 let program_id = match &handler_spec.source {
801 crate::ast::SourceSpec::Source { program_id, .. } => {
802 program_id.as_ref().map(|s| s.as_str()).unwrap_or("null")
803 }
804 };
805 debug_info.push(format!(
806 " [{}] EventType={}, Mappings={}, ProgramId={}",
807 index,
808 event_type,
809 handler_spec.mappings.len(),
810 program_id
811 ));
812 }
813
814 let mut segments: HashMap<String, Vec<HandlerSegment>> = HashMap::new();
829
830 for handler_spec in &self.spec.handlers {
831 for mapping in &handler_spec.mappings {
832 if let Some(when) = &mapping.when {
833 when_events.insert(when.clone());
834 }
835 let entry = emit_by_path
836 .entry(mapping.target_path.clone())
837 .or_insert(false);
838 *entry |= mapping.emit;
839 if mapping.stop.is_some() {
840 emit_by_path
841 .entry(stop_field_path(&mapping.target_path))
842 .or_insert(false);
843 }
844 }
845 let opcodes = self.compile_handler(handler_spec);
846 let event_type = self.get_event_type(&handler_spec.source);
847 let key_identity = self.key_identity(handler_spec);
848
849 let event_segments = segments.entry(event_type).or_default();
850 if let Some(existing) = event_segments
851 .iter_mut()
852 .find(|segment| segment.key_identity.as_ref() == Some(&key_identity))
853 {
854 merge_handler_opcodes(&mut existing.ops, &opcodes);
855 } else {
856 event_segments.push(HandlerSegment {
857 key_identity: Some(key_identity),
858 route_field: route_field(&handler_spec.key_resolution),
859 ops: opcodes,
860 });
861 }
862 }
863
864 for hook in &self.spec.instruction_hooks {
866 let event_type = hook.instruction_type.clone();
867 let event_segments = segments.entry(event_type).or_default();
868
869 let segment_index = match &hook.lookup_by {
875 None if !event_segments.is_empty() => Some(0),
876 None => None,
877 Some(lookup_by) => event_segments
878 .iter()
879 .position(|segment| segment.route_field.as_ref() == Some(lookup_by)),
880 };
881 let segment_index = match segment_index {
882 Some(index) => index,
883 None => {
884 let resolution = if event_segments.is_empty() {
889 None
890 } else {
891 hook.lookup_by
892 .as_ref()
893 .and_then(|lookup_by| self.hook_key_resolution(lookup_by))
894 };
895 let ops = match resolution {
896 Some(resolution) => self.compile_keyed_hook_segment(&resolution),
897 None => self.compile_hook_segment(hook),
898 };
899 event_segments.push(HandlerSegment {
900 key_identity: None,
901 route_field: hook.lookup_by.clone(),
902 ops,
903 });
904 event_segments.len() - 1
905 }
906 };
907 let handler_opcodes = &mut event_segments[segment_index].ops;
908
909 let hook_opcodes = self.compile_instruction_hook_actions(&hook.actions);
911
912 let insert_pos = handler_opcodes
916 .iter()
917 .position(|op| matches!(op, OpCode::EvaluateComputedFields { .. }))
918 .or_else(|| {
919 handler_opcodes
920 .iter()
921 .position(|op| matches!(op, OpCode::UpdateState { .. }))
922 });
923
924 if let Some(pos) = insert_pos {
925 for (i, opcode) in hook_opcodes.into_iter().enumerate() {
927 handler_opcodes.insert(pos + i, opcode);
928 }
929 }
930 }
931
932 for (event_type, paths) in self.sibling_capture_paths() {
937 let event_segments = segments.entry(event_type).or_default();
938 event_segments.insert(
939 0,
940 HandlerSegment {
941 key_identity: None,
942 route_field: None,
943 ops: vec![OpCode::CaptureSibling {
944 state_id: self.state_id,
945 paths,
946 }],
947 },
948 );
949 }
950
951 let handlers: HashMap<String, Vec<OpCode>> = segments
952 .into_iter()
953 .map(|(event_type, event_segments)| {
954 let mut ops = Vec::new();
955 for (index, segment) in event_segments.into_iter().enumerate() {
956 if index > 0 {
957 ops.push(OpCode::SegmentBoundary);
958 }
959 ops.extend(segment.ops);
960 }
961 (event_type, ops)
962 })
963 .collect();
964
965 let non_emitted_fields: HashSet<String> = emit_by_path
966 .into_iter()
967 .filter_map(|(path, emit)| if emit { None } else { Some(path) })
968 .collect();
969
970 EntityBytecode {
971 state_id: self.state_id,
972 handlers,
973 entity_name: self.entity_name.clone(),
974 when_events,
975 non_emitted_fields,
976 computed_paths: self.spec.computed_fields.clone(),
977 computed_fields_evaluator: None,
978 }
979 }
980
981 fn sibling_capture_paths(&self) -> Vec<(String, Vec<FieldPath>)> {
984 let mut captures: std::collections::BTreeMap<String, Vec<FieldPath>> =
985 std::collections::BTreeMap::new();
986 for hook in &self.spec.instruction_hooks {
987 for action in &hook.actions {
988 let HookAction::RegisterPdaMapping {
989 pda_field,
990 seed_field,
991 sibling: Some(sibling),
992 ..
993 } = action
994 else {
995 continue;
996 };
997 let paths = captures
998 .entry(sibling.instruction_type.clone())
999 .or_default();
1000 for (from_sibling, path) in [
1001 (sibling.pda_from_sibling, pda_field),
1002 (sibling.seed_from_sibling, seed_field),
1003 ] {
1004 if from_sibling && !paths.iter().any(|known| known.segments == path.segments) {
1005 paths.push(path.clone());
1006 }
1007 }
1008 }
1009 }
1010 captures.into_iter().collect()
1011 }
1012
1013 fn key_identity(&self, spec: &TypedHandlerSpec<S>) -> Value {
1017 let key_reg = 20;
1018 let ops = self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings);
1019 serde_json::to_value(&ops).unwrap_or(Value::Null)
1020 }
1021
1022 fn hook_key_resolution(&self, lookup_by: &FieldPath) -> Option<KeyResolutionStrategy> {
1026 let leaf = lookup_by.segments.last()?;
1027 let is_primary_key_field = self
1028 .spec
1029 .identity
1030 .primary_keys
1031 .iter()
1032 .any(|pk| pk.rsplit('.').next() == Some(leaf.as_str()));
1033 if is_primary_key_field {
1034 Some(KeyResolutionStrategy::Embedded {
1035 primary_field: lookup_by.clone(),
1036 })
1037 } else if self.find_lookup_index_for_field(lookup_by).is_some() {
1038 Some(KeyResolutionStrategy::Lookup {
1039 primary_field: lookup_by.clone(),
1040 index_name: None,
1041 })
1042 } else {
1043 None
1044 }
1045 }
1046
1047 fn compile_keyed_hook_segment(&self, resolution: &KeyResolutionStrategy) -> Vec<OpCode> {
1051 let key_reg = 20;
1052 let state_reg = 2;
1053 let mut ops = self.compile_key_loading(resolution, key_reg, &[]);
1054 ops.push(OpCode::AbortIfNullKey {
1055 key: key_reg,
1056 is_account_event: false,
1057 });
1058 ops.push(OpCode::ReadOrInitState {
1059 state_id: self.state_id,
1060 key: key_reg,
1061 default: serde_json::json!({}),
1062 dest: state_reg,
1063 dedup_route: None,
1064 });
1065 ops.push(OpCode::UpdateState {
1066 state_id: self.state_id,
1067 key: key_reg,
1068 value: state_reg,
1069 });
1070 ops
1071 }
1072
1073 fn compile_hook_segment(&self, hook: &InstructionHook) -> Vec<OpCode> {
1077 let key_reg = 20;
1078 let state_reg = 2;
1079 let resolved_key_reg = 19;
1080 let temp_reg = 18;
1081
1082 let mut ops = Vec::new();
1083
1084 ops.push(OpCode::LoadEventField {
1086 path: FieldPath::new(&["__resolved_primary_key"]),
1087 dest: resolved_key_reg,
1088 default: Some(serde_json::json!(null)),
1089 });
1090
1091 ops.push(OpCode::CopyRegister {
1093 source: resolved_key_reg,
1094 dest: key_reg,
1095 });
1096
1097 if let Some(lookup_path) = &hook.lookup_by {
1099 ops.push(OpCode::LoadEventField {
1101 path: lookup_path.clone(),
1102 dest: temp_reg,
1103 default: None,
1104 });
1105
1106 ops.push(OpCode::Transform {
1108 source: temp_reg,
1109 dest: temp_reg,
1110 transformation: Transformation::HexEncode,
1111 });
1112
1113 ops.push(OpCode::CopyRegisterIfNull {
1115 source: temp_reg,
1116 dest: key_reg,
1117 });
1118 }
1119
1120 ops.push(OpCode::ReadOrInitState {
1121 state_id: self.state_id,
1122 key: key_reg,
1123 default: serde_json::json!({}),
1124 dest: state_reg,
1125 dedup_route: None,
1126 });
1127
1128 ops.push(OpCode::UpdateState {
1129 state_id: self.state_id,
1130 key: key_reg,
1131 value: state_reg,
1132 });
1133
1134 ops
1135 }
1136
1137 fn compile_handler(&self, spec: &TypedHandlerSpec<S>) -> Vec<OpCode> {
1138 let mut ops = Vec::new();
1139 let state_reg = 2;
1140 let key_reg = 20;
1141
1142 ops.extend(self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings));
1143
1144 let is_account_event = matches!(
1150 spec.source,
1151 SourceSpec::Source {
1152 is_account: true,
1153 ..
1154 }
1155 );
1156 ops.push(OpCode::AbortIfNullKey {
1157 key: key_reg,
1158 is_account_event,
1159 });
1160
1161 ops.push(OpCode::ReadOrInitState {
1162 state_id: self.state_id,
1163 key: key_reg,
1164 default: serde_json::json!({}),
1165 dest: state_reg,
1166 dedup_route: declared_index(&spec.key_resolution)
1167 .map(|index| format!("{index}:{:?}", spec.key_resolution)),
1168 });
1169
1170 ops.extend(self.compile_temporal_index_update(
1177 &spec.key_resolution,
1178 key_reg,
1179 &spec.mappings,
1180 ));
1181
1182 for mapping in &spec.mappings {
1183 ops.extend(self.compile_mapping(mapping, state_reg, key_reg));
1184 }
1185
1186 ops.extend(self.compile_resolvers(state_reg, key_reg));
1187
1188 ops.push(OpCode::EvaluateComputedFields {
1190 state: state_reg,
1191 computed_paths: self.spec.computed_fields.clone(),
1192 });
1193
1194 ops.push(OpCode::UpdateState {
1195 state_id: self.state_id,
1196 key: key_reg,
1197 value: state_reg,
1198 });
1199
1200 if spec.emit {
1201 ops.push(OpCode::EmitMutation {
1202 entity_name: self.entity_name.clone(),
1203 key: key_reg,
1204 state: state_reg,
1205 });
1206 }
1207
1208 ops
1209 }
1210
1211 fn compile_resolvers(&self, state_reg: Register, key_reg: Register) -> Vec<OpCode> {
1212 let mut ops = Vec::new();
1213
1214 for resolver_spec in &self.spec.resolver_specs {
1215 let url_template = match &resolver_spec.resolver {
1216 ResolverType::Url(config) => match &config.url_source {
1217 UrlSource::Template(parts) => Some(parts.clone()),
1218 _ => None,
1219 },
1220 _ => None,
1221 };
1222
1223 ops.push(OpCode::QueueResolver {
1224 state_id: self.state_id,
1225 entity_name: self.entity_name.clone(),
1226 resolver: resolver_spec.resolver.clone(),
1227 input_path: resolver_spec.input_path.clone(),
1228 input_value: resolver_spec.input_value.clone(),
1229 url_template,
1230 strategy: resolver_spec.strategy.clone(),
1231 extracts: resolver_spec.extracts.clone(),
1232 condition: resolver_spec.condition.clone(),
1233 schedule_at: resolver_spec.schedule_at.clone(),
1234 state: state_reg,
1235 key: key_reg,
1236 });
1237 }
1238
1239 ops
1240 }
1241
1242 fn compile_mapping(
1243 &self,
1244 mapping: &TypedFieldMapping<S>,
1245 state_reg: Register,
1246 key_reg: Register,
1247 ) -> Vec<OpCode> {
1248 let guard = mapping
1249 .condition
1250 .as_ref()
1251 .and_then(|condition| condition.parsed.as_ref())
1252 .and_then(|parsed| match parsed {
1253 ParsedCondition::Guard { expr } => Some(expr),
1254 _ => None,
1255 });
1256 let Some(guard) = guard else {
1257 return self.compile_mapping_ops(
1258 mapping,
1259 mapping.condition.as_ref(),
1260 state_reg,
1261 key_reg,
1262 );
1263 };
1264 let inner = self.compile_mapping_ops(mapping, None, state_reg, key_reg);
1267 let mut ops = Vec::with_capacity(inner.len() + 2);
1268 ops.push(OpCode::EvaluateEventExpr {
1269 expr: guard.clone(),
1270 transform: None,
1271 dest: GUARD_REG,
1272 });
1273 ops.push(OpCode::SkipUnless {
1274 predicate: GUARD_REG,
1275 count: inner.len(),
1276 });
1277 ops.extend(inner);
1278 ops
1279 }
1280
1281 fn compile_mapping_ops(
1282 &self,
1283 mapping: &TypedFieldMapping<S>,
1284 condition: Option<&ConditionExpr>,
1285 state_reg: Register,
1286 key_reg: Register,
1287 ) -> Vec<OpCode> {
1288 let mut ops = Vec::new();
1289 let temp_reg = 10;
1290
1291 ops.extend(self.compile_mapping_source(&mapping.source, temp_reg));
1292
1293 if let Some(transform) = &mapping.transform {
1294 ops.push(OpCode::Transform {
1295 source: temp_reg,
1296 dest: temp_reg,
1297 transformation: transform.clone(),
1298 });
1299 }
1300
1301 if let Some(stop_instruction) = &mapping.stop {
1302 if mapping.when.is_some() {
1303 tracing::warn!(
1304 "#[map] stop and when both set for {}. Ignoring when.",
1305 mapping.target_path
1306 );
1307 }
1308 if !matches!(mapping.population, PopulationStrategy::LastWrite)
1309 && !matches!(mapping.population, PopulationStrategy::Merge)
1310 {
1311 tracing::warn!(
1312 "#[map] stop ignores population strategy {:?}",
1313 mapping.population
1314 );
1315 }
1316
1317 ops.push(OpCode::SetFieldUnlessStopped {
1318 object: state_reg,
1319 path: mapping.target_path.clone(),
1320 value: temp_reg,
1321 stop_field: stop_field_path(&mapping.target_path),
1322 stop_instruction: stop_instruction.clone(),
1323 entity_name: self.entity_name.clone(),
1324 key_reg,
1325 });
1326 return ops;
1327 }
1328
1329 if let Some(when_instruction) = &mapping.when {
1330 if !matches!(mapping.population, PopulationStrategy::LastWrite)
1331 && !matches!(mapping.population, PopulationStrategy::Merge)
1332 {
1333 tracing::warn!(
1334 "#[map] when ignores population strategy {:?}",
1335 mapping.population
1336 );
1337 }
1338 let (condition_field, condition_op, condition_value) = condition
1339 .and_then(|cond| cond.parsed.as_ref())
1340 .and_then(|parsed| match parsed {
1341 ParsedCondition::Comparison { field, op, value } => {
1342 Some((Some(field.clone()), Some(op.clone()), Some(value.clone())))
1343 }
1344 ParsedCondition::Logical { .. } => {
1345 tracing::warn!("Logical conditions not yet supported for #[map] when");
1346 None
1347 }
1348 ParsedCondition::Guard { .. } => None,
1350 })
1351 .unwrap_or((None, None, None));
1352
1353 ops.push(OpCode::SetFieldWhen {
1354 object: state_reg,
1355 path: mapping.target_path.clone(),
1356 value: temp_reg,
1357 when_instruction: when_instruction.clone(),
1358 entity_name: self.entity_name.clone(),
1359 key_reg,
1360 condition_field,
1361 condition_op,
1362 condition_value,
1363 });
1364 return ops;
1365 }
1366
1367 if let Some(condition) = condition {
1368 if let Some(parsed) = &condition.parsed {
1369 match parsed {
1370 ParsedCondition::Comparison {
1371 field,
1372 op,
1373 value: cond_value,
1374 } => {
1375 if matches!(mapping.population, PopulationStrategy::LastWrite)
1376 || matches!(mapping.population, PopulationStrategy::Merge)
1377 {
1378 ops.push(OpCode::ConditionalSetField {
1379 object: state_reg,
1380 path: mapping.target_path.clone(),
1381 value: temp_reg,
1382 condition_field: field.clone(),
1383 condition_op: op.clone(),
1384 condition_value: cond_value.clone(),
1385 });
1386 return ops;
1387 }
1388
1389 if matches!(mapping.population, PopulationStrategy::Count) {
1390 ops.push(OpCode::ConditionalIncrement {
1391 object: state_reg,
1392 path: mapping.target_path.clone(),
1393 condition_field: field.clone(),
1394 condition_op: op.clone(),
1395 condition_value: cond_value.clone(),
1396 });
1397 return ops;
1398 }
1399
1400 tracing::warn!(
1401 "Conditional #[map] not supported for population strategy {:?}",
1402 mapping.population
1403 );
1404 }
1405 ParsedCondition::Logical { .. } => {
1406 tracing::warn!("Logical conditions not yet supported for #[map]");
1407 }
1408 ParsedCondition::Guard { .. } => {}
1410 }
1411 }
1412 }
1413
1414 match &mapping.population {
1415 PopulationStrategy::Append => {
1416 ops.push(OpCode::AppendToArray {
1417 object: state_reg,
1418 path: mapping.target_path.clone(),
1419 value: temp_reg,
1420 });
1421 }
1422 PopulationStrategy::LastWrite => {
1423 ops.push(OpCode::SetField {
1424 object: state_reg,
1425 path: mapping.target_path.clone(),
1426 value: temp_reg,
1427 });
1428 }
1429 PopulationStrategy::SetOnce => {
1430 ops.push(OpCode::SetFieldIfNull {
1431 object: state_reg,
1432 path: mapping.target_path.clone(),
1433 value: temp_reg,
1434 });
1435 }
1436 PopulationStrategy::Merge => {
1437 ops.push(OpCode::SetField {
1438 object: state_reg,
1439 path: mapping.target_path.clone(),
1440 value: temp_reg,
1441 });
1442 }
1443 PopulationStrategy::Max => {
1444 ops.push(OpCode::SetFieldMax {
1445 object: state_reg,
1446 path: mapping.target_path.clone(),
1447 value: temp_reg,
1448 });
1449 }
1450 PopulationStrategy::Sum => {
1451 ops.push(OpCode::SetFieldSum {
1452 object: state_reg,
1453 path: mapping.target_path.clone(),
1454 value: temp_reg,
1455 });
1456 }
1457 PopulationStrategy::Count => {
1458 ops.push(OpCode::SetFieldIncrement {
1460 object: state_reg,
1461 path: mapping.target_path.clone(),
1462 });
1463 }
1464 PopulationStrategy::Min => {
1465 ops.push(OpCode::SetFieldMin {
1466 object: state_reg,
1467 path: mapping.target_path.clone(),
1468 value: temp_reg,
1469 });
1470 }
1471 PopulationStrategy::UniqueCount => {
1472 let set_name = format!("{}_unique_set", mapping.target_path);
1475 ops.push(OpCode::AddToUniqueSet {
1476 state_id: self.state_id,
1477 set_name,
1478 value: temp_reg,
1479 count_object: state_reg,
1480 count_path: mapping.target_path.clone(),
1481 });
1482 }
1483 }
1484
1485 ops
1486 }
1487
1488 fn compile_mapping_source(&self, source: &MappingSource, dest: Register) -> Vec<OpCode> {
1489 match source {
1490 MappingSource::FromSource {
1491 path,
1492 default,
1493 transform,
1494 } => {
1495 let mut ops = vec![OpCode::LoadEventField {
1496 path: path.clone(),
1497 dest,
1498 default: default.clone(),
1499 }];
1500
1501 if let Some(transform_type) = transform {
1503 ops.push(OpCode::Transform {
1504 source: dest,
1505 dest,
1506 transformation: transform_type.clone(),
1507 });
1508 }
1509
1510 ops
1511 }
1512 MappingSource::Constant(val) => {
1513 vec![OpCode::LoadConstant {
1514 value: val.clone(),
1515 dest,
1516 }]
1517 }
1518 MappingSource::AsEvent { fields } => {
1519 let mut ops = Vec::new();
1520
1521 if fields.is_empty() {
1522 let event_data_reg = dest + 1;
1523 ops.push(OpCode::LoadEventField {
1524 path: FieldPath::new(&[]),
1525 dest: event_data_reg,
1526 default: Some(serde_json::json!({})),
1527 });
1528 ops.push(OpCode::CreateEvent {
1529 dest,
1530 event_value: event_data_reg,
1531 });
1532 } else {
1533 let data_obj_reg = dest + 1;
1534 ops.push(OpCode::CreateObject { dest: data_obj_reg });
1535
1536 let mut field_registers = Vec::new();
1537 let mut current_reg = dest + 2;
1538
1539 for field_source in fields.iter() {
1540 if let MappingSource::FromSource {
1541 path,
1542 default,
1543 transform,
1544 } = &**field_source
1545 {
1546 ops.push(OpCode::LoadEventField {
1547 path: path.clone(),
1548 dest: current_reg,
1549 default: default.clone(),
1550 });
1551
1552 if let Some(transform_type) = transform {
1553 ops.push(OpCode::Transform {
1554 source: current_reg,
1555 dest: current_reg,
1556 transformation: transform_type.clone(),
1557 });
1558 }
1559
1560 if let Some(field_name) = path.segments.last() {
1561 field_registers.push((field_name.clone(), current_reg));
1562 }
1563 current_reg += 1;
1564 }
1565 }
1566
1567 if !field_registers.is_empty() {
1568 ops.push(OpCode::SetFields {
1569 object: data_obj_reg,
1570 fields: field_registers,
1571 });
1572 }
1573
1574 ops.push(OpCode::CreateEvent {
1575 dest,
1576 event_value: data_obj_reg,
1577 });
1578 }
1579
1580 ops
1581 }
1582 MappingSource::WholeSource => {
1583 vec![OpCode::LoadEventField {
1584 path: FieldPath::new(&[]),
1585 dest,
1586 default: Some(serde_json::json!({})),
1587 }]
1588 }
1589 MappingSource::AsCapture { field_transforms } => {
1590 let capture_data_reg = 22; let mut ops = vec![OpCode::LoadEventField {
1593 path: FieldPath::new(&[]),
1594 dest: capture_data_reg,
1595 default: Some(serde_json::json!({})),
1596 }];
1597
1598 let field_reg = 24;
1602 let transformed_reg = 25;
1603
1604 for (field_name, transform) in field_transforms {
1605 ops.push(OpCode::GetField {
1608 object: capture_data_reg,
1609 path: field_name.clone(),
1610 dest: field_reg,
1611 });
1612
1613 ops.push(OpCode::Transform {
1615 source: field_reg,
1616 dest: transformed_reg,
1617 transformation: transform.clone(),
1618 });
1619
1620 ops.push(OpCode::SetField {
1622 object: capture_data_reg,
1623 path: field_name.clone(),
1624 value: transformed_reg,
1625 });
1626 }
1627
1628 ops.push(OpCode::CreateCapture {
1630 dest,
1631 capture_value: capture_data_reg,
1632 });
1633
1634 ops
1635 }
1636 MappingSource::FromContext { field } => {
1637 vec![OpCode::LoadEventField {
1639 path: FieldPath::new(&["__update_context", field.as_str()]),
1640 dest,
1641 default: Some(serde_json::json!(null)),
1642 }]
1643 }
1644 MappingSource::FromExpr { expr } => {
1645 vec![OpCode::EvaluateEventExpr {
1646 expr: expr.clone(),
1647 transform: None,
1648 dest,
1649 }]
1650 }
1651 MappingSource::Computed { .. } => {
1652 vec![]
1653 }
1654 MappingSource::FromState { .. } => {
1655 vec![]
1656 }
1657 }
1658 }
1659
1660 pub fn compile_key_loading(
1661 &self,
1662 resolution: &KeyResolutionStrategy,
1663 key_reg: Register,
1664 mappings: &[TypedFieldMapping<S>],
1665 ) -> Vec<OpCode> {
1666 let mut ops = Vec::new();
1667
1668 let resolved_key_reg = 19; ops.push(OpCode::LoadEventField {
1672 path: FieldPath::new(&["__resolved_primary_key"]),
1673 dest: resolved_key_reg,
1674 default: Some(serde_json::json!(null)),
1675 });
1676
1677 match resolution {
1679 KeyResolutionStrategy::Embedded { primary_field }
1680 if primary_field.segments.is_empty()
1681 && self.find_primary_key_expr(mappings).is_some() =>
1682 {
1683 let expr = self
1686 .find_primary_key_expr(mappings)
1687 .expect("checked by the guard");
1688 if let Some(guard) = self.find_primary_key_guard(mappings) {
1691 ops.push(OpCode::EvaluateEventExpr {
1692 expr: guard,
1693 transform: None,
1694 dest: GUARD_REG,
1695 });
1696 ops.push(OpCode::SkipUnless {
1697 predicate: GUARD_REG,
1698 count: 1,
1699 });
1700 }
1701 ops.push(OpCode::EvaluateEventExpr {
1705 expr,
1706 transform: self.find_primary_key_transformation(mappings),
1707 dest: key_reg,
1708 });
1709 ops.push(OpCode::CopyRegisterIfNull {
1710 source: resolved_key_reg,
1711 dest: key_reg,
1712 });
1713 }
1714 KeyResolutionStrategy::Embedded { primary_field } => {
1715 let effective_primary_field = if primary_field.segments.is_empty() {
1717 if let Some(auto_field) = self.auto_detect_primary_field(mappings) {
1719 auto_field
1720 } else {
1721 primary_field.clone()
1722 }
1723 } else {
1724 primary_field.clone()
1725 };
1726
1727 if !effective_primary_field.segments.is_empty() {
1731 let temp_reg = 18;
1732 let transform_reg = 23; ops.push(OpCode::LoadEventField {
1735 path: effective_primary_field.clone(),
1736 dest: temp_reg,
1737 default: None,
1738 });
1739
1740 let primary_key_transform = self
1743 .find_primary_key_transformation(mappings)
1744 .or_else(|| self.find_inherited_primary_key_transformation());
1745
1746 if let Some(transform) = primary_key_transform {
1747 ops.push(OpCode::Transform {
1749 source: temp_reg,
1750 dest: transform_reg,
1751 transformation: transform,
1752 });
1753 ops.push(OpCode::CopyRegister {
1757 source: transform_reg,
1758 dest: key_reg,
1759 });
1760 } else {
1761 ops.push(OpCode::CopyRegister {
1763 source: temp_reg,
1764 dest: key_reg,
1765 });
1766 }
1767 ops.push(OpCode::CopyRegisterIfNull {
1768 source: resolved_key_reg,
1769 dest: key_reg,
1770 });
1771 } else {
1772 ops.push(OpCode::CopyRegister {
1774 source: resolved_key_reg,
1775 dest: key_reg,
1776 });
1777 }
1778 }
1779 KeyResolutionStrategy::Lookup {
1780 primary_field,
1781 index_name: declared,
1782 } => {
1783 let lookup_reg = 15;
1784 let result_reg = 17;
1785
1786 ops.push(OpCode::CopyRegister {
1792 source: resolved_key_reg,
1793 dest: lookup_reg,
1794 });
1795
1796 let temp_reg = 18;
1797 ops.push(OpCode::LoadEventField {
1798 path: primary_field.clone(),
1799 dest: temp_reg,
1800 default: None,
1801 });
1802 ops.push(OpCode::CopyRegisterIfNull {
1803 source: temp_reg,
1804 dest: lookup_reg,
1805 });
1806
1807 let effective_index_name = match declared {
1811 Some(declared) => declared.clone(),
1812 None => self
1813 .find_lookup_index_for_lookup_field(primary_field, mappings)
1814 .unwrap_or_else(|| "default_pda_lookup".to_string()),
1815 };
1816
1817 ops.push(OpCode::LookupIndex {
1818 state_id: self.state_id,
1819 index_name: effective_index_name,
1820 lookup_value: lookup_reg,
1821 dest: result_reg,
1822 strict: declared.is_some(),
1823 });
1824 ops.push(OpCode::CopyRegister {
1825 source: result_reg,
1826 dest: key_reg,
1827 });
1828
1829 if self.resolver_outputs_primary_key_directly(primary_field) {
1837 ops.push(OpCode::CopyRegisterIfNull {
1838 source: resolved_key_reg,
1839 dest: key_reg,
1840 });
1841 }
1842 }
1843 KeyResolutionStrategy::Computed {
1844 primary_field,
1845 compute_partition: _,
1846 } => {
1847 ops.push(OpCode::CopyRegister {
1849 source: resolved_key_reg,
1850 dest: key_reg,
1851 });
1852 let temp_reg = 18;
1853 ops.push(OpCode::LoadEventField {
1854 path: primary_field.clone(),
1855 dest: temp_reg,
1856 default: None,
1857 });
1858 ops.push(OpCode::CopyRegisterIfNull {
1859 source: temp_reg,
1860 dest: key_reg,
1861 });
1862 }
1863 KeyResolutionStrategy::TemporalLookup {
1864 lookup_field,
1865 timestamp_field,
1866 index_name,
1867 } => {
1868 ops.push(OpCode::CopyRegister {
1870 source: resolved_key_reg,
1871 dest: key_reg,
1872 });
1873 let lookup_reg = 15;
1874 let timestamp_reg = 16;
1875 let result_reg = 17;
1876
1877 ops.push(OpCode::LoadEventField {
1878 path: lookup_field.clone(),
1879 dest: lookup_reg,
1880 default: None,
1881 });
1882
1883 ops.push(OpCode::LoadEventField {
1884 path: timestamp_field.clone(),
1885 dest: timestamp_reg,
1886 default: None,
1887 });
1888
1889 ops.push(OpCode::LookupTemporalIndex {
1890 state_id: self.state_id,
1891 index_name: index_name.clone(),
1892 lookup_value: lookup_reg,
1893 timestamp: timestamp_reg,
1894 dest: result_reg,
1895 });
1896
1897 ops.push(OpCode::CopyRegisterIfNull {
1898 source: result_reg,
1899 dest: key_reg,
1900 });
1901 }
1902 }
1903
1904 ops
1905 }
1906
1907 fn find_primary_key_expr(&self, mappings: &[TypedFieldMapping<S>]) -> Option<ComputedExpr> {
1910 let primary_key = self.spec.identity.primary_keys.first()?;
1911 mappings.iter().find_map(|mapping| match &mapping.source {
1912 MappingSource::FromExpr { expr } if mapping.target_path == *primary_key => {
1913 Some(expr.clone())
1914 }
1915 _ => None,
1916 })
1917 }
1918
1919 fn find_primary_key_guard(&self, mappings: &[TypedFieldMapping<S>]) -> Option<ComputedExpr> {
1922 let primary_key = self.spec.identity.primary_keys.first()?;
1923 mappings.iter().find_map(|mapping| {
1924 if mapping.target_path != *primary_key
1925 || !matches!(mapping.source, MappingSource::FromExpr { .. })
1926 {
1927 return None;
1928 }
1929 match mapping.condition.as_ref()?.parsed.as_ref()? {
1930 ParsedCondition::Guard { expr } => Some(expr.clone()),
1931 _ => None,
1932 }
1933 })
1934 }
1935
1936 fn find_primary_key_transformation(
1937 &self,
1938 mappings: &[TypedFieldMapping<S>],
1939 ) -> Option<Transformation> {
1940 let primary_key = self.spec.identity.primary_keys.first()?;
1942 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1943
1944 for mapping in mappings {
1946 if mapping.target_path == *primary_key
1948 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1949 {
1950 if let Some(transform) = &mapping.transform {
1952 return Some(transform.clone());
1953 }
1954
1955 if let MappingSource::FromSource {
1957 transform: Some(transform),
1958 ..
1959 } = &mapping.source
1960 {
1961 return Some(transform.clone());
1962 }
1963 }
1964 }
1965
1966 for mapping in mappings {
1968 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1969 if let Some(transform) = field_transforms.get(&primary_field_name) {
1970 return Some(transform.clone());
1971 }
1972 }
1973 }
1974
1975 None
1976 }
1977
1978 pub fn find_inherited_primary_key_transformation(&self) -> Option<Transformation> {
1981 let primary_key = self.spec.identity.primary_keys.first()?;
1982
1983 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1985
1986 for handler in &self.spec.handlers {
1988 for mapping in &handler.mappings {
1989 if mapping.target_path == *primary_key
1991 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1992 {
1993 if let MappingSource::FromSource {
1995 path, transform, ..
1996 } = &mapping.source
1997 {
1998 if path.segments.last() == Some(&primary_field_name) {
1999 return mapping.transform.clone().or_else(|| transform.clone());
2001 }
2002 }
2003 }
2004
2005 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
2007 if let Some(transform) = field_transforms.get(&primary_field_name) {
2008 return Some(transform.clone());
2009 }
2010 }
2011 }
2012 }
2013
2014 None
2015 }
2016
2017 fn extract_primary_field_name(&self, primary_key: &str) -> Option<String> {
2019 primary_key.split('.').next_back().map(|s| s.to_string())
2021 }
2022
2023 pub fn auto_detect_primary_field(
2026 &self,
2027 current_mappings: &[TypedFieldMapping<S>],
2028 ) -> Option<FieldPath> {
2029 let primary_key = self.spec.identity.primary_keys.first()?;
2030
2031 let primary_field_name = self.extract_primary_field_name(primary_key)?;
2033
2034 if self.current_account_has_primary_field(&primary_field_name, current_mappings) {
2036 return Some(FieldPath::new(&[&primary_field_name]));
2037 }
2038
2039 None
2040 }
2041
2042 fn current_account_has_primary_field(
2045 &self,
2046 field_name: &str,
2047 mappings: &[TypedFieldMapping<S>],
2048 ) -> bool {
2049 for mapping in mappings {
2051 if let MappingSource::FromSource { path, .. } = &mapping.source {
2052 if path.segments.last() == Some(&field_name.to_string()) {
2054 return true;
2055 }
2056 }
2057 }
2058
2059 false
2060 }
2061
2062 #[allow(dead_code)]
2064 fn handler_has_field(&self, field_name: &str, mappings: &[TypedFieldMapping<S>]) -> bool {
2065 for mapping in mappings {
2066 if let MappingSource::FromSource { path, .. } = &mapping.source {
2067 if path.segments.last() == Some(&field_name.to_string()) {
2068 return true;
2069 }
2070 }
2071 }
2072 false
2073 }
2074
2075 #[allow(dead_code)]
2078 fn field_exists_in_mappings(
2079 &self,
2080 field_name: &str,
2081 mappings: &[TypedFieldMapping<S>],
2082 ) -> bool {
2083 for mapping in mappings {
2085 if let MappingSource::FromSource { path, .. } = &mapping.source {
2086 if path.segments.last() == Some(&field_name.to_string()) {
2087 return true;
2088 }
2089 }
2090 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
2092 if field_transforms.contains_key(field_name) {
2093 return true;
2094 }
2095 }
2096 }
2097 false
2098 }
2099
2100 fn find_lookup_index_for_field(&self, field_path: &FieldPath) -> Option<String> {
2101 if field_path.segments.is_empty() {
2102 return None;
2103 }
2104
2105 let lookup_field_name = field_path.segments.last().unwrap();
2106
2107 for lookup_index in &self.spec.identity.lookup_indexes {
2108 let index_field_name = lookup_index
2109 .field_name
2110 .split('.')
2111 .next_back()
2112 .unwrap_or(&lookup_index.field_name);
2113 let matches_directly = index_field_name == lookup_field_name;
2114 let matches_address_alias = index_field_name
2121 .strip_suffix("_address")
2122 .map(|base| base == lookup_field_name)
2123 .unwrap_or(false);
2124
2125 if matches_directly || matches_address_alias {
2126 return Some(format!("{}_lookup_index", index_field_name));
2127 }
2128 }
2129
2130 None
2131 }
2132
2133 fn declares_index(&self, index_name: &str) -> bool {
2136 self.spec
2137 .handlers
2138 .iter()
2139 .any(|handler| declared_index(&handler.key_resolution) == Some(index_name))
2140 }
2141
2142 fn find_lookup_index_for_lookup_field(
2145 &self,
2146 primary_field: &FieldPath,
2147 mappings: &[TypedFieldMapping<S>],
2148 ) -> Option<String> {
2149 let primary_path = primary_field.segments.join(".");
2151
2152 for mapping in mappings {
2154 if let MappingSource::FromSource { path, .. } = &mapping.source {
2156 let source_path = path.segments.join(".");
2157 if source_path == primary_path {
2158 for lookup_index in &self.spec.identity.lookup_indexes {
2160 if mapping.target_path == lookup_index.field_name {
2161 let index_field_name = lookup_index
2162 .field_name
2163 .split('.')
2164 .next_back()
2165 .unwrap_or(&lookup_index.field_name);
2166 return Some(format!("{}_lookup_index", index_field_name));
2167 }
2168 }
2169 }
2170 }
2171 }
2172
2173 self.find_lookup_index_for_field(primary_field)
2175 }
2176
2177 fn find_source_path_for_lookup_index(
2181 &self,
2182 mappings: &[TypedFieldMapping<S>],
2183 lookup_field_name: &str,
2184 ) -> Option<Vec<String>> {
2185 for mapping in mappings {
2186 if mapping.target_path == lookup_field_name {
2187 if let MappingSource::FromSource { path, .. } = &mapping.source {
2188 return Some(path.segments.clone());
2189 }
2190 }
2191 }
2192 None
2193 }
2194
2195 fn compile_temporal_index_update(
2196 &self,
2197 resolution: &KeyResolutionStrategy,
2198 key_reg: Register,
2199 mappings: &[TypedFieldMapping<S>],
2200 ) -> Vec<OpCode> {
2201 let mut ops = Vec::new();
2202
2203 for lookup_index in &self.spec.identity.lookup_indexes {
2204 let lookup_reg = 17;
2205 let source_field = lookup_index
2206 .field_name
2207 .split('.')
2208 .next_back()
2209 .unwrap_or(&lookup_index.field_name);
2210
2211 let start = ops.len();
2212 match resolution {
2213 KeyResolutionStrategy::Embedded { primary_field: _ } => {
2214 let source_path_opt =
2217 self.find_source_path_for_lookup_index(mappings, &lookup_index.field_name);
2218
2219 let load_path = if let Some(ref path) = source_path_opt {
2220 FieldPath::new(&path.iter().map(|s| s.as_str()).collect::<Vec<_>>())
2221 } else {
2222 FieldPath::new(&[source_field])
2224 };
2225
2226 if let Some(mapping) = mappings
2227 .iter()
2228 .find(|m| m.target_path == lookup_index.field_name)
2229 {
2230 let value_ops = self.compile_mapping_source(&mapping.source, lookup_reg);
2231 if value_ops.is_empty() {
2232 continue;
2233 }
2234 ops.extend(value_ops);
2235 if let Some(transform) = &mapping.transform {
2236 ops.push(OpCode::Transform {
2237 source: lookup_reg,
2238 dest: lookup_reg,
2239 transformation: transform.clone(),
2240 });
2241 }
2242 } else {
2243 ops.push(OpCode::LoadEventField {
2244 path: load_path,
2245 dest: lookup_reg,
2246 default: None,
2247 });
2248 }
2249
2250 if let Some(temporal_field_name) = &lookup_index.temporal_field {
2251 let timestamp_reg = 18;
2252
2253 ops.push(OpCode::LoadEventField {
2254 path: FieldPath::new(&[temporal_field_name]),
2255 dest: timestamp_reg,
2256 default: None,
2257 });
2258
2259 let index_name = format!("{}_temporal_index", source_field);
2260 ops.push(OpCode::UpdateTemporalIndex {
2261 state_id: self.state_id,
2262 index_name,
2263 lookup_value: lookup_reg,
2264 primary_key: key_reg,
2265 timestamp: timestamp_reg,
2266 });
2267
2268 let simple_index_name = format!("{}_lookup_index", source_field);
2269 ops.push(OpCode::UpdateLookupIndex {
2270 state_id: self.state_id,
2271 index_name: simple_index_name,
2272 lookup_value: lookup_reg,
2273 primary_key: key_reg,
2274 });
2275 } else {
2276 let index_name = format!("{}_lookup_index", source_field);
2277 ops.push(OpCode::UpdateLookupIndex {
2278 state_id: self.state_id,
2279 index_name,
2280 lookup_value: lookup_reg,
2281 primary_key: key_reg,
2282 });
2283 }
2284
2285 if source_path_opt.is_some() {
2289 ops.push(OpCode::UpdatePdaReverseLookup {
2290 state_id: self.state_id,
2291 lookup_name: "default_pda_lookup".to_string(),
2292 pda_address: lookup_reg,
2293 primary_key: key_reg,
2294 });
2295 }
2296 }
2297 KeyResolutionStrategy::Lookup { .. } => {
2298 if let Some(mapping) = mappings
2299 .iter()
2300 .find(|m| m.target_path == lookup_index.field_name)
2301 {
2302 let value_ops = self.compile_mapping_source(&mapping.source, lookup_reg);
2303 if value_ops.is_empty() {
2304 continue;
2305 }
2306 ops.extend(value_ops);
2307 if let Some(transform) = &mapping.transform {
2308 ops.push(OpCode::Transform {
2309 source: lookup_reg,
2310 dest: lookup_reg,
2311 transformation: transform.clone(),
2312 });
2313 }
2314 ops.push(OpCode::UpdateLookupIndex {
2315 state_id: self.state_id,
2316 index_name: format!("{}_lookup_index", source_field),
2317 lookup_value: lookup_reg,
2318 primary_key: key_reg,
2319 });
2320 }
2321 }
2322 KeyResolutionStrategy::Computed { .. }
2323 | KeyResolutionStrategy::TemporalLookup { .. } => {
2324 }
2326 }
2327 let index_name = format!("{}_lookup_index", source_field);
2330 if self.declares_index(&index_name) {
2331 if let Some(guard) = index_guard(mappings, &lookup_index.field_name) {
2332 let count = ops.len() - start;
2333 ops.splice(
2334 start..start,
2335 [
2336 OpCode::EvaluateEventExpr {
2337 expr: guard,
2338 transform: None,
2339 dest: GUARD_REG,
2340 },
2341 OpCode::SkipUnless {
2342 predicate: GUARD_REG,
2343 count,
2344 },
2345 ],
2346 );
2347 }
2348 }
2349 }
2350
2351 ops
2352 }
2353
2354 fn get_event_type(&self, source: &SourceSpec) -> String {
2355 match source {
2356 SourceSpec::Source { type_name, .. } => type_name.clone(),
2357 }
2358 }
2359
2360 fn compile_instruction_hook_actions(&self, actions: &[HookAction]) -> Vec<OpCode> {
2361 let mut ops = Vec::new();
2362 let state_reg = 2;
2363
2364 for action in actions {
2365 match action {
2366 HookAction::SetField {
2367 target_field,
2368 source,
2369 condition,
2370 } => {
2371 let _ = condition;
2373
2374 let temp_reg = 11; let load_ops = self.compile_mapping_source(source, temp_reg);
2378 ops.extend(load_ops);
2379
2380 if let MappingSource::FromSource {
2382 transform: Some(transform_type),
2383 ..
2384 } = source
2385 {
2386 ops.push(OpCode::Transform {
2387 source: temp_reg,
2388 dest: temp_reg,
2389 transformation: transform_type.clone(),
2390 });
2391 }
2392
2393 if let Some(cond_expr) = condition {
2395 if let Some(parsed) = &cond_expr.parsed {
2396 let cond_check_ops = self.compile_condition_check(
2398 parsed,
2399 temp_reg,
2400 state_reg,
2401 target_field,
2402 );
2403 ops.extend(cond_check_ops);
2404 } else {
2405 ops.push(OpCode::SetField {
2407 object: state_reg,
2408 path: target_field.clone(),
2409 value: temp_reg,
2410 });
2411 }
2412 } else {
2413 ops.push(OpCode::SetField {
2415 object: state_reg,
2416 path: target_field.clone(),
2417 value: temp_reg,
2418 });
2419 }
2420 }
2421 HookAction::IncrementField {
2422 target_field,
2423 increment_by,
2424 condition,
2425 } => {
2426 if let Some(cond_expr) = condition {
2427 if let Some(parsed) = &cond_expr.parsed {
2428 let cond_check_ops = self.compile_conditional_increment(
2433 parsed,
2434 state_reg,
2435 target_field,
2436 *increment_by,
2437 );
2438 ops.extend(cond_check_ops);
2439 } else {
2440 ops.push(OpCode::SetFieldIncrement {
2442 object: state_reg,
2443 path: target_field.clone(),
2444 });
2445 }
2446 } else {
2447 ops.push(OpCode::SetFieldIncrement {
2449 object: state_reg,
2450 path: target_field.clone(),
2451 });
2452 }
2453 }
2454 HookAction::RegisterPdaMapping {
2455 pda_field,
2456 seed_field,
2457 lookup_name,
2458 sibling,
2459 seed_expr,
2460 guard,
2461 } => {
2462 let pda_reg = 11;
2463 let seed_reg = 12;
2464 let mut registration = Vec::new();
2465 let pda_from_sibling = sibling
2466 .as_ref()
2467 .is_some_and(|sibling| sibling.pda_from_sibling);
2468 let seed_from_sibling = sibling
2469 .as_ref()
2470 .is_some_and(|sibling| sibling.seed_from_sibling);
2471
2472 registration.push(if pda_from_sibling {
2475 OpCode::LoadConstant {
2476 value: Value::Null,
2477 dest: pda_reg,
2478 }
2479 } else {
2480 OpCode::LoadEventField {
2481 path: pda_field.clone(),
2482 dest: pda_reg,
2483 default: None,
2484 }
2485 });
2486 registration.push(match (seed_from_sibling, seed_expr) {
2487 (true, _) => OpCode::LoadConstant {
2488 value: Value::Null,
2489 dest: seed_reg,
2490 },
2491 (false, Some(expr)) => OpCode::EvaluateEventExpr {
2492 expr: expr.clone(),
2493 transform: None,
2494 dest: seed_reg,
2495 },
2496 (false, None) => OpCode::LoadEventField {
2497 path: seed_field.clone(),
2498 dest: seed_reg,
2499 default: None,
2500 },
2501 });
2502 registration.push(match sibling {
2503 None => OpCode::UpdatePdaReverseLookup {
2504 state_id: self.state_id,
2505 lookup_name: lookup_name.clone(),
2506 pda_address: pda_reg,
2507 primary_key: seed_reg,
2508 },
2509 Some(sibling) => OpCode::RegisterPdaFromSibling {
2510 state_id: self.state_id,
2511 lookup_name: lookup_name.clone(),
2512 sibling: sibling.clone(),
2513 pda_field: pda_field.clone(),
2514 seed_field: seed_field.clone(),
2515 pda_address: pda_reg,
2516 primary_key: seed_reg,
2517 },
2518 });
2519
2520 if let Some(guard) = guard {
2521 ops.push(OpCode::EvaluateEventExpr {
2522 expr: guard.clone(),
2523 transform: None,
2524 dest: GUARD_REG,
2525 });
2526 ops.push(OpCode::SkipUnless {
2527 predicate: GUARD_REG,
2528 count: registration.len(),
2529 });
2530 }
2531 ops.extend(registration);
2532 }
2533 }
2534 }
2535
2536 ops
2537 }
2538
2539 fn compile_condition_check(
2540 &self,
2541 condition: &ParsedCondition,
2542 value_reg: Register,
2543 state_reg: Register,
2544 target_field: &str,
2545 ) -> Vec<OpCode> {
2546 match condition {
2547 ParsedCondition::Comparison {
2548 field,
2549 op,
2550 value: cond_value,
2551 } => {
2552 vec![OpCode::ConditionalSetField {
2554 object: state_reg,
2555 path: target_field.to_string(),
2556 value: value_reg,
2557 condition_field: field.clone(),
2558 condition_op: op.clone(),
2559 condition_value: cond_value.clone(),
2560 }]
2561 }
2562 ParsedCondition::Logical { .. } => {
2563 tracing::warn!("Logical conditions not yet supported in instruction hooks");
2565 vec![OpCode::SetField {
2566 object: state_reg,
2567 path: target_field.to_string(),
2568 value: value_reg,
2569 }]
2570 }
2571 ParsedCondition::Guard { expr } => vec![
2572 OpCode::EvaluateEventExpr {
2573 expr: expr.clone(),
2574 transform: None,
2575 dest: GUARD_REG,
2576 },
2577 OpCode::SkipUnless {
2578 predicate: GUARD_REG,
2579 count: 1,
2580 },
2581 OpCode::SetField {
2582 object: state_reg,
2583 path: target_field.to_string(),
2584 value: value_reg,
2585 },
2586 ],
2587 }
2588 }
2589
2590 fn compile_conditional_increment(
2591 &self,
2592 condition: &ParsedCondition,
2593 state_reg: Register,
2594 target_field: &str,
2595 _increment_by: i64,
2596 ) -> Vec<OpCode> {
2597 match condition {
2598 ParsedCondition::Comparison {
2599 field,
2600 op,
2601 value: cond_value,
2602 } => {
2603 vec![OpCode::ConditionalIncrement {
2604 object: state_reg,
2605 path: target_field.to_string(),
2606 condition_field: field.clone(),
2607 condition_op: op.clone(),
2608 condition_value: cond_value.clone(),
2609 }]
2610 }
2611 ParsedCondition::Logical { .. } => {
2612 tracing::warn!("Logical conditions not yet supported in instruction hooks");
2613 vec![OpCode::SetFieldIncrement {
2614 object: state_reg,
2615 path: target_field.to_string(),
2616 }]
2617 }
2618 ParsedCondition::Guard { expr } => vec![
2619 OpCode::EvaluateEventExpr {
2620 expr: expr.clone(),
2621 transform: None,
2622 dest: GUARD_REG,
2623 },
2624 OpCode::SkipUnless {
2625 predicate: GUARD_REG,
2626 count: 1,
2627 },
2628 OpCode::SetFieldIncrement {
2629 object: state_reg,
2630 path: target_field.to_string(),
2631 },
2632 ],
2633 }
2634 }
2635}
2636
2637#[cfg(test)]
2638mod tests {
2639 use super::MultiEntityBytecode;
2640 use crate::ast::{
2641 FieldPath, HookAction, IdentitySpec, InstructionHook, KeyResolutionStrategy,
2642 LookupIndexSpec, MappingSource, PopulationStrategy, SerializableFieldMapping,
2643 SerializableHandlerSpec, SerializableStreamSpec, SourceSpec, TypedStreamSpec,
2644 };
2645 use crate::vm::VmContext;
2646 use crate::{Mutation, WholeEntity};
2647 use serde_json::{json, Value};
2648 use std::collections::BTreeMap;
2649
2650 fn mapping(
2651 target_path: &str,
2652 source_path: &[&str],
2653 population: PopulationStrategy,
2654 ) -> SerializableFieldMapping {
2655 SerializableFieldMapping {
2656 target_path: target_path.to_string(),
2657 source: MappingSource::FromSource {
2658 path: FieldPath::new(source_path),
2659 default: None,
2660 transform: None,
2661 },
2662 transform: None,
2663 population,
2664 condition: None,
2665 when: None,
2666 stop: None,
2667 emit: true,
2668 }
2669 }
2670
2671 fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
2672 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2673 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2674 state_name: "PumpfunToken".to_string(),
2675 program_id: None,
2676 idl: None,
2677 identity: IdentitySpec {
2678 primary_keys: vec!["id.mint".to_string()],
2679 lookup_indexes: vec![LookupIndexSpec {
2680 field_name: "id.bonding_curve".to_string(),
2681 temporal_field: None,
2682 }],
2683 },
2684 handlers: vec![SerializableHandlerSpec {
2685 source: SourceSpec::Source {
2686 program_id: None,
2687 discriminator: None,
2688 type_name: "pump::BondingCurveState".to_string(),
2689 serialization: None,
2690 is_account: true,
2691 },
2692 key_resolution: KeyResolutionStrategy::Lookup {
2693 primary_field: FieldPath::new(&["__account_address"]),
2694 index_name: None,
2695 },
2696 mappings: vec![
2697 mapping(
2698 "id.bonding_curve",
2699 &["__account_address"],
2700 PopulationStrategy::SetOnce,
2701 ),
2702 mapping(
2703 "reserves.virtual_token_reserves",
2704 &["virtual_token_reserves"],
2705 PopulationStrategy::LastWrite,
2706 ),
2707 ],
2708 conditions: vec![],
2709 emit: true,
2710 }],
2711 sections: vec![],
2712 field_mappings: BTreeMap::new(),
2713 resolver_hooks: vec![],
2714 instruction_hooks: vec![InstructionHook {
2715 instruction_type: "pump::BuyIxState".to_string(),
2716 actions: vec![HookAction::RegisterPdaMapping {
2717 pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2718 seed_field: FieldPath::new(&["accounts", "mint"]),
2719 lookup_name: "default_pda_lookup".to_string(),
2720 sibling: None,
2721 seed_expr: None,
2722 guard: None,
2723 }],
2724 lookup_by: None,
2725 }],
2726 resolver_specs: vec![],
2727 computed_fields: vec![],
2728 computed_field_specs: vec![],
2729 content_hash: None,
2730 views: vec![],
2731 scope_field: None,
2732 })
2733 }
2734
2735 fn timed_bonding_curve_spec() -> TypedStreamSpec<Value> {
2742 let with_source = |target: &str, source: MappingSource| SerializableFieldMapping {
2743 source,
2744 ..mapping(target, &[], PopulationStrategy::LastWrite)
2745 };
2746 let timestamped = || {
2747 vec![
2748 with_source(
2749 "events.reserves",
2750 MappingSource::AsEvent {
2751 fields: vec![Box::new(MappingSource::FromSource {
2752 path: FieldPath::new(&["virtual_token_reserves"]),
2753 default: None,
2754 transform: None,
2755 })],
2756 },
2757 ),
2758 with_source(
2759 "snapshots.curve",
2760 MappingSource::AsCapture {
2761 field_transforms: BTreeMap::new(),
2762 },
2763 ),
2764 with_source(
2765 "state.seen_at",
2766 MappingSource::FromContext {
2767 field: "timestamp".to_string(),
2768 },
2769 ),
2770 ]
2771 };
2772 let source = |type_name: &str, is_account: bool| SourceSpec::Source {
2773 program_id: None,
2774 discriminator: None,
2775 type_name: type_name.to_string(),
2776 serialization: None,
2777 is_account,
2778 };
2779 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2780 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2781 state_name: "PumpfunToken".to_string(),
2782 program_id: None,
2783 idl: None,
2784 identity: IdentitySpec {
2785 primary_keys: vec!["id.mint".to_string()],
2786 lookup_indexes: vec![
2787 LookupIndexSpec {
2788 field_name: "id.bonding_curve".to_string(),
2789 temporal_field: None,
2790 },
2791 LookupIndexSpec {
2792 field_name: "id.creator".to_string(),
2793 temporal_field: None,
2794 },
2795 LookupIndexSpec {
2796 field_name: "id.round".to_string(),
2797 temporal_field: None,
2798 },
2799 ],
2800 },
2801 handlers: vec![
2802 SerializableHandlerSpec {
2803 source: source("pump::BondingCurveState", true),
2804 key_resolution: KeyResolutionStrategy::Lookup {
2805 primary_field: FieldPath::new(&["__account_address"]),
2806 index_name: None,
2807 },
2808 mappings: [
2809 vec![mapping(
2810 "id.bonding_curve",
2811 &["__account_address"],
2812 PopulationStrategy::SetOnce,
2813 )],
2814 timestamped(),
2815 ]
2816 .concat(),
2817 conditions: vec![],
2818 emit: true,
2819 },
2820 SerializableHandlerSpec {
2821 source: source("pump::RoundState", true),
2822 key_resolution: KeyResolutionStrategy::Lookup {
2823 primary_field: FieldPath::new(&["round"]),
2824 index_name: None,
2825 },
2826 mappings: timestamped(),
2827 conditions: vec![],
2828 emit: true,
2829 },
2830 SerializableHandlerSpec {
2831 source: source("pump::OpenRoundIxState", false),
2832 key_resolution: KeyResolutionStrategy::Embedded {
2833 primary_field: FieldPath::new(&["accounts", "mint"]),
2834 },
2835 mappings: vec![
2836 mapping(
2837 "id.mint",
2838 &["accounts", "mint"],
2839 PopulationStrategy::SetOnce,
2840 ),
2841 mapping(
2842 "id.round",
2843 &["accounts", "round"],
2844 PopulationStrategy::SetOnce,
2845 ),
2846 ],
2847 conditions: vec![],
2848 emit: true,
2849 },
2850 SerializableHandlerSpec {
2851 source: source("pump::LinkIxState", false),
2852 key_resolution: KeyResolutionStrategy::Embedded {
2853 primary_field: FieldPath::new(&["accounts", "mint"]),
2854 },
2855 mappings: vec![
2856 mapping(
2857 "id.mint",
2858 &["accounts", "mint"],
2859 PopulationStrategy::SetOnce,
2860 ),
2861 mapping(
2862 "id.bonding_curve",
2863 &["accounts", "bonding_curve"],
2864 PopulationStrategy::SetOnce,
2865 ),
2866 ],
2867 conditions: vec![],
2868 emit: true,
2869 },
2870 SerializableHandlerSpec {
2871 source: source("pump::CreateIxState", false),
2872 key_resolution: KeyResolutionStrategy::Embedded {
2873 primary_field: FieldPath::new(&["accounts", "mint"]),
2874 },
2875 mappings: vec![
2876 mapping(
2877 "id.mint",
2878 &["accounts", "mint"],
2879 PopulationStrategy::SetOnce,
2880 ),
2881 mapping(
2882 "id.creator",
2883 &["accounts", "user"],
2884 PopulationStrategy::SetOnce,
2885 ),
2886 ],
2887 conditions: vec![],
2888 emit: true,
2889 },
2890 ],
2891 sections: vec![],
2892 field_mappings: BTreeMap::new(),
2893 resolver_hooks: vec![],
2894 instruction_hooks: vec![InstructionHook {
2895 instruction_type: "pump::BuyIxState".to_string(),
2896 actions: vec![HookAction::RegisterPdaMapping {
2897 pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2898 seed_field: FieldPath::new(&["accounts", "user"]),
2899 lookup_name: "default_pda_lookup".to_string(),
2900 sibling: None,
2901 seed_expr: None,
2902 guard: None,
2903 }],
2904 lookup_by: None,
2905 }],
2906 resolver_specs: vec![],
2907 computed_fields: vec![],
2908 computed_field_specs: vec![],
2909 content_hash: None,
2910 views: vec![],
2911 scope_field: None,
2912 })
2913 }
2914
2915 #[test]
2923 fn a_queued_account_update_replays_with_its_own_timestamp() {
2924 use crate::vm::UpdateContext;
2925
2926 let accounts = json!({
2927 "bonding_curve": "bonding_curve_1",
2928 "mint": "mint_1",
2929 "user": "user_1",
2930 "round": 7,
2931 });
2932 let curve = (
2933 "pump::BondingCurveState",
2934 json!({ "__account_address": "bonding_curve_1", "virtual_token_reserves": 42 }),
2935 );
2936 let round = (
2937 "pump::RoundState",
2938 json!({ "__account_address": "round_1", "round": 7, "virtual_token_reserves": 42 }),
2939 );
2940 let instruction = |vm: &mut VmContext, bytecode: &MultiEntityBytecode, name: &str| {
2941 let mut context =
2942 UpdateContext::new_instruction(20, "instruction-signature".to_string(), 0);
2943 context.timestamp = Some(2_000);
2944 vm.process_event(
2945 bytecode,
2946 json!({ "accounts": accounts }),
2947 name,
2948 Some(&context),
2949 None,
2950 )
2951 .unwrap();
2952 };
2953 let scenarios: [(&[&str], _, &[&str]); 4] = [
2956 (&[], &curve, &["pump::LinkIxState"]),
2957 (&[], &curve, &["pump::CreateIxState", "pump::BuyIxState"]),
2958 (&["pump::BuyIxState"], &curve, &["pump::CreateIxState"]),
2959 (&[], &round, &["pump::OpenRoundIxState"]),
2960 ];
2961 for (before, (account_type, account), after) in scenarios {
2962 let bytecode = MultiEntityBytecode::from_single(
2963 "PumpfunToken".to_string(),
2964 timed_bonding_curve_spec(),
2965 0,
2966 );
2967 let mut vm = VmContext::new();
2968 for name in before {
2969 instruction(&mut vm, &bytecode, name);
2970 }
2971 let mut account_context =
2972 UpdateContext::new_account(10, "curve-signature".to_string(), 1);
2973 account_context.timestamp = Some(1_000);
2974
2975 let queued = vm
2976 .process_event(
2977 &bytecode,
2978 account.clone(),
2979 account_type,
2980 Some(&account_context),
2981 None,
2982 )
2983 .unwrap();
2984 assert!(queued.is_empty(), "{after:?}: {queued:?}");
2985
2986 std::thread::sleep(std::time::Duration::from_millis(1_100));
2989 for name in after {
2990 instruction(&mut vm, &bytecode, name);
2991 }
2992
2993 let token = vm
2994 .get_entity_state(0, &json!("mint_1"))
2995 .unwrap_or_else(|| panic!("{after:?}: no entity"));
2996 for path in [
2997 "/events/reserves/timestamp",
2998 "/snapshots/curve/timestamp",
2999 "/state/seen_at",
3000 ] {
3001 assert_eq!(
3002 token.pointer(path),
3003 Some(&json!(1_000)),
3004 "{before:?} {after:?} {path}: {token:#}"
3005 );
3006 }
3007 assert_eq!(
3008 token.pointer("/events/reserves/data/virtual_token_reserves"),
3009 Some(&json!(42))
3010 );
3011 assert_eq!(token.pointer("/events/reserves/slot"), Some(&json!(10)));
3012 assert_eq!(
3013 token.pointer("/snapshots/curve/signature"),
3014 Some(&json!("curve-signature"))
3015 );
3016 }
3017 }
3018
3019 fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
3020 TypedStreamSpec::from_serializable(SerializableStreamSpec {
3021 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3022 state_name: "OreRound".to_string(),
3023 program_id: None,
3024 idl: None,
3025 identity: IdentitySpec {
3026 primary_keys: vec!["id.round_id".to_string()],
3027 lookup_indexes: vec![LookupIndexSpec {
3028 field_name: "id.round_address".to_string(),
3029 temporal_field: None,
3030 }],
3031 },
3032 handlers: vec![SerializableHandlerSpec {
3033 source: SourceSpec::Source {
3034 program_id: None,
3035 discriminator: None,
3036 type_name: "entropy::VarState".to_string(),
3037 serialization: None,
3038 is_account: true,
3039 },
3040 key_resolution: KeyResolutionStrategy::Lookup {
3041 primary_field: FieldPath::new(&["__account_address"]),
3042 index_name: None,
3043 },
3044 mappings: vec![mapping(
3045 "state.expires_at",
3046 &["end_at"],
3047 PopulationStrategy::LastWrite,
3048 )],
3049 conditions: vec![],
3050 emit: true,
3051 }],
3052 sections: vec![],
3053 field_mappings: BTreeMap::new(),
3054 resolver_hooks: vec![],
3055 instruction_hooks: vec![InstructionHook {
3056 instruction_type: "ore::DeployIxState".to_string(),
3057 actions: vec![HookAction::RegisterPdaMapping {
3058 pda_field: FieldPath::new(&["accounts", "entropyVar"]),
3059 seed_field: FieldPath::new(&["accounts", "round"]),
3060 lookup_name: "default_pda_lookup".to_string(),
3061 sibling: None,
3062 seed_expr: None,
3063 guard: None,
3064 }],
3065 lookup_by: None,
3066 }],
3067 resolver_specs: vec![],
3068 computed_fields: vec![],
3069 computed_field_specs: vec![],
3070 content_hash: None,
3071 views: vec![],
3072 scope_field: None,
3073 })
3074 }
3075
3076 fn embedded_account_key_spec() -> TypedStreamSpec<Value> {
3077 TypedStreamSpec::from_serializable(SerializableStreamSpec {
3078 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3079 state_name: "OreTreasury".to_string(),
3080 program_id: None,
3081 idl: None,
3082 identity: IdentitySpec {
3083 primary_keys: vec!["id.address".to_string()],
3084 lookup_indexes: vec![],
3085 },
3086 handlers: vec![SerializableHandlerSpec {
3087 source: SourceSpec::Source {
3088 program_id: None,
3089 discriminator: None,
3090 type_name: "ore::TreasuryState".to_string(),
3091 serialization: None,
3092 is_account: true,
3093 },
3094 key_resolution: KeyResolutionStrategy::Embedded {
3095 primary_field: FieldPath::new(&["__account_address"]),
3096 },
3097 mappings: vec![mapping(
3098 "id.address",
3099 &["__account_address"],
3100 PopulationStrategy::SetOnce,
3101 )],
3102 conditions: vec![],
3103 emit: true,
3104 }],
3105 sections: vec![],
3106 field_mappings: BTreeMap::new(),
3107 resolver_hooks: vec![],
3108 instruction_hooks: vec![],
3109 resolver_specs: vec![],
3110 computed_fields: vec![],
3111 computed_field_specs: vec![],
3112 content_hash: None,
3113 views: vec![],
3114 scope_field: None,
3115 })
3116 }
3117
3118 fn whole_entity_pool_spec() -> TypedStreamSpec<Value> {
3122 let mut hidden = mapping("state.secret", &["secret"], PopulationStrategy::LastWrite);
3123 hidden.emit = false;
3124 let handler = |type_name: &str, mappings, emit| SerializableHandlerSpec {
3125 source: SourceSpec::Source {
3126 program_id: None,
3127 discriminator: None,
3128 type_name: type_name.to_string(),
3129 serialization: None,
3130 is_account: true,
3131 },
3132 key_resolution: KeyResolutionStrategy::Embedded {
3133 primary_field: FieldPath::new(&["__account_address"]),
3134 },
3135 mappings,
3136 conditions: vec![],
3137 emit,
3138 };
3139 TypedStreamSpec::from_serializable(SerializableStreamSpec {
3140 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3141 state_name: "Pool".to_string(),
3142 program_id: None,
3143 idl: None,
3144 identity: IdentitySpec {
3145 primary_keys: vec!["id.address".to_string()],
3146 lookup_indexes: vec![],
3147 },
3148 handlers: vec![
3149 handler(
3150 "amm::PoolState",
3151 vec![
3152 mapping(
3153 "id.address",
3154 &["__account_address"],
3155 PopulationStrategy::SetOnce,
3156 ),
3157 mapping("state.price", &["price"], PopulationStrategy::LastWrite),
3158 mapping("state.fills", &["fill"], PopulationStrategy::Append),
3159 hidden,
3160 ],
3161 true,
3162 ),
3163 handler(
3164 "amm::PoolConfigState",
3165 vec![mapping(
3166 "config.fee",
3167 &["fee"],
3168 PopulationStrategy::LastWrite,
3169 )],
3170 false,
3171 ),
3172 ],
3173 sections: vec![],
3174 field_mappings: BTreeMap::new(),
3175 resolver_hooks: vec![],
3176 instruction_hooks: vec![],
3177 resolver_specs: vec![],
3178 computed_fields: vec![],
3179 computed_field_specs: vec![],
3180 content_hash: None,
3181 views: vec![],
3182 scope_field: None,
3183 })
3184 }
3185
3186 fn pool_bytecode() -> MultiEntityBytecode {
3187 MultiEntityBytecode::from_single("Pool".to_string(), whole_entity_pool_spec(), 0)
3188 }
3189
3190 #[test]
3194 fn compiled_event_capture_and_context_mappings_carry_the_update_timestamp() {
3195 let with_source = |target: &str, source: MappingSource| SerializableFieldMapping {
3196 source,
3197 ..mapping(target, &[], PopulationStrategy::LastWrite)
3198 };
3199 let spec: TypedStreamSpec<Value> =
3200 TypedStreamSpec::from_serializable(SerializableStreamSpec {
3201 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3202 state_name: "Pool".to_string(),
3203 program_id: None,
3204 idl: None,
3205 identity: IdentitySpec {
3206 primary_keys: vec!["id.address".to_string()],
3207 lookup_indexes: vec![],
3208 },
3209 handlers: vec![SerializableHandlerSpec {
3210 source: SourceSpec::Source {
3211 program_id: None,
3212 discriminator: None,
3213 type_name: "amm::PoolState".to_string(),
3214 serialization: None,
3215 is_account: true,
3216 },
3217 key_resolution: KeyResolutionStrategy::Embedded {
3218 primary_field: FieldPath::new(&["__account_address"]),
3219 },
3220 mappings: vec![
3221 mapping(
3222 "id.address",
3223 &["__account_address"],
3224 PopulationStrategy::SetOnce,
3225 ),
3226 with_source(
3227 "events.priced",
3228 MappingSource::AsEvent {
3229 fields: vec![Box::new(MappingSource::FromSource {
3230 path: FieldPath::new(&["price"]),
3231 default: None,
3232 transform: None,
3233 })],
3234 },
3235 ),
3236 with_source(
3237 "snapshots.pool",
3238 MappingSource::AsCapture {
3239 field_transforms: BTreeMap::new(),
3240 },
3241 ),
3242 with_source(
3243 "state.updated_at",
3244 MappingSource::FromContext {
3245 field: "timestamp".to_string(),
3246 },
3247 ),
3248 ],
3249 conditions: vec![],
3250 emit: true,
3251 }],
3252 sections: vec![],
3253 field_mappings: BTreeMap::new(),
3254 resolver_hooks: vec![],
3255 instruction_hooks: vec![],
3256 resolver_specs: vec![],
3257 computed_fields: vec![],
3258 computed_field_specs: vec![],
3259 content_hash: None,
3260 views: vec![],
3261 scope_field: None,
3262 });
3263 let bytecode = MultiEntityBytecode::from_single("Pool".to_string(), spec, 0);
3264 let context = crate::vm::UpdateContext::with_timestamp(7, "sig-7".to_string(), 1_234_567);
3265
3266 let mut vm = VmContext::new();
3267 vm.process_event(
3268 &bytecode,
3269 pool_event(100, 1),
3270 "amm::PoolState",
3271 Some(&context),
3272 None,
3273 )
3274 .unwrap();
3275
3276 let pool = vm.get_entity_state(0, &json!("pool_1")).unwrap();
3277 for path in [
3278 "/events/priced/timestamp",
3279 "/snapshots/pool/timestamp",
3280 "/state/updated_at",
3281 ] {
3282 assert_eq!(
3283 pool.pointer(path),
3284 Some(&json!(1_234_567)),
3285 "{path}: {pool:#}"
3286 );
3287 }
3288 assert_eq!(pool.pointer("/events/priced/data/price"), Some(&json!(100)));
3289 assert_eq!(pool.pointer("/snapshots/pool/slot"), Some(&json!(7)));
3290 }
3291
3292 fn pool_event(price: u64, fill: u64) -> Value {
3293 pool_event_for("pool_1", price, fill)
3294 }
3295
3296 fn pool_event_for(address: &str, price: u64, fill: u64) -> Value {
3297 json!({"__account_address": address, "price": price, "fill": fill, "secret": "s"})
3298 }
3299
3300 fn linked_vm(
3302 state_config: crate::vm::StateTableConfig,
3303 ) -> (VmContext, crate::vm::WholeEntityRequests) {
3304 let requests = crate::vm::WholeEntityRequests::new(16);
3305 let mut vm = VmContext::new_with_config(state_config);
3306 vm.set_whole_entity_requests(requests.clone());
3307 (vm, requests)
3308 }
3309
3310 fn process(vm: &mut VmContext, bytecode: &MultiEntityBytecode, event: Value) -> Vec<Mutation> {
3311 vm.process_event(bytecode, event, "amm::PoolState", None, None)
3312 .unwrap()
3313 }
3314
3315 #[test]
3316 fn source_deletion_cancels_resync_and_recreates_a_fresh_vm_row() {
3317 let bytecode = pool_bytecode();
3318 let (mut vm, requests) = linked_vm(Default::default());
3319 let first = process(&mut vm, &bytecode, pool_event(1, 10));
3320 let key = &first[0].key;
3321 let target = crate::vm::ResolverTarget {
3322 state_id: 0,
3323 entity_name: "Pool".into(),
3324 primary_key: key.clone(),
3325 extracts: vec![],
3326 };
3327 vm.enqueue_resolver_request(
3328 "ignored".into(),
3329 crate::ast::ResolverType::Token,
3330 json!("mint"),
3331 target.clone(),
3332 );
3333 let old_request = vm.take_resolver_requests().pop().unwrap();
3334 vm.get_state_table_mut(0).unwrap().deferred_when_ops.insert(
3335 ("sig".into(), "instruction".into()),
3336 vec![crate::vm::DeferredWhenOperation {
3337 entity_name: "Pool".into(),
3338 primary_key: key.clone(),
3339 field_path: "old".into(),
3340 field_value: json!(true),
3341 when_instruction: "instruction".into(),
3342 signature: "sig".into(),
3343 slot: 1,
3344 deferred_at: 1,
3345 emit: true,
3346 }],
3347 );
3348 requests.request("Pool", key);
3349 let deletion = vm.delete_entity(&bytecode, "Pool", key).unwrap();
3350 assert!(deletion.is_delete());
3351 assert!(
3352 serde_json::from_value::<Mutation>(serde_json::to_value(&deletion).unwrap())
3353 .unwrap()
3354 .is_delete()
3355 );
3356 assert!(!requests.is_requested("Pool", key));
3357 assert!(vm.get_entity_state(0, key).is_none());
3358 assert!(vm.snapshot_state_table(0).is_empty());
3359 assert!(vm
3360 .get_state_table_mut(0)
3361 .unwrap()
3362 .deferred_when_ops
3363 .is_empty());
3364 assert!(vm.delete_entity(&bytecode, "Unknown", key).is_none());
3365 let mut next = process(&mut vm, &bytecode, pool_event(9, 90));
3366 vm.enqueue_resolver_request(
3367 "ignored".into(),
3368 crate::ast::ResolverType::Token,
3369 json!("mint"),
3370 target,
3371 );
3372 let new_request = vm.take_resolver_requests().pop().unwrap();
3373 assert_ne!(old_request.cache_key, new_request.cache_key);
3374 assert!(vm
3375 .apply_resolver_result(&bytecode, &old_request.cache_key, json!({"stale":true}))
3376 .unwrap()
3377 .is_empty());
3378 assert_eq!(next[0].take_whole_entity_mark(), Some(WholeEntity::Created));
3379 assert_eq!(next[0].patch["state"]["fills"], json!([90]));
3380 }
3381
3382 #[test]
3383 fn deletion_preserves_unrelated_resolver_sharing_and_isolates_recreation() {
3384 let bytecode = pool_bytecode();
3385 let (mut vm, _) = linked_vm(Default::default());
3386 let key = process(&mut vm, &bytecode, pool_event(1, 10))[0]
3387 .key
3388 .clone();
3389 let late_joiner = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20))[0]
3390 .key
3391 .clone();
3392 let target = |primary_key| crate::vm::ResolverTarget {
3393 state_id: 0,
3394 entity_name: "Pool".into(),
3395 primary_key,
3396 extracts: vec![crate::ast::ResolverExtractSpec {
3397 target_path: "resolved".into(),
3398 source_path: Some("value".into()),
3399 transform: None,
3400 }],
3401 };
3402 let enqueue = |vm: &mut VmContext, key| {
3403 vm.enqueue_resolver_request(
3404 "ignored".into(),
3405 crate::ast::ResolverType::Token,
3406 json!("shared-mint"),
3407 target(key),
3408 )
3409 };
3410 enqueue(&mut vm, key.clone());
3411 enqueue(&mut vm, json!("unrelated"));
3412 let old = vm.take_resolver_requests();
3413 assert_eq!(old.len(), 1);
3414 vm.delete_entity(&bytecode, "Pool", &key).unwrap();
3415 enqueue(&mut vm, json!("unrelated"));
3416 enqueue(&mut vm, json!("another-unrelated"));
3417 assert!(
3418 vm.take_resolver_requests().is_empty(),
3419 "unaffected targets reuse the old pending call"
3420 );
3421 enqueue(&mut vm, key.clone());
3422 let fresh = vm.take_resolver_requests();
3423 assert_eq!(fresh.len(), 1);
3424 assert_ne!(fresh[0].cache_key, old[0].cache_key);
3425 vm.delete_entity(&bytecode, "Pool", &late_joiner).unwrap();
3428 process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
3429 enqueue(&mut vm, late_joiner.clone());
3430 let late_joiner_fresh = vm.take_resolver_requests();
3431 assert_eq!(late_joiner_fresh.len(), 1);
3432 assert_ne!(late_joiner_fresh[0].cache_key, old[0].cache_key);
3433 enqueue(&mut vm, json!("unrelated"));
3434 enqueue(&mut vm, json!("another-unrelated"));
3435 enqueue(&mut vm, key.clone());
3436 assert!(
3437 vm.take_resolver_requests().is_empty(),
3438 "unaffected targets reuse pending calls"
3439 );
3440 let old_mutations = vm
3441 .apply_resolver_result(&bytecode, &old[0].cache_key, json!({"value":"old"}))
3442 .unwrap();
3443 assert_eq!(old_mutations.len(), 2);
3444 assert!(old_mutations.iter().all(|mutation| mutation.key != key));
3445 assert!(vm.get_entity_state(0, &key).is_none());
3446 let fresh_mutations = vm
3447 .apply_resolver_result(&bytecode, &fresh[0].cache_key, json!({"value":"fresh"}))
3448 .unwrap();
3449 assert_eq!(fresh_mutations.len(), 1);
3450 assert_eq!(fresh_mutations[0].key, key);
3451 let late_joiner_mutations = vm
3452 .apply_resolver_result(
3453 &bytecode,
3454 &late_joiner_fresh[0].cache_key,
3455 json!({"value":"fresh-late-joiner"}),
3456 )
3457 .unwrap();
3458 assert_eq!(late_joiner_mutations.len(), 1);
3459 assert_eq!(late_joiner_mutations[0].key, late_joiner);
3460 }
3461
3462 #[test]
3463 fn a_requested_entity_is_emitted_whole_once() {
3464 let bytecode = pool_bytecode();
3465 let (mut vm, requests) = linked_vm(Default::default());
3466 let mut update = |price, fill| process(&mut vm, &bytecode, pool_event(price, fill));
3467
3468 let first = update(1, 10);
3469 assert_eq!(first[0].whole_entity_mark(), Some(WholeEntity::Created));
3470 let key = first[0].key.clone();
3471 let second = update(2, 20);
3472 assert_eq!(second.len(), 1);
3473 assert!(!second[0].is_whole_entity());
3474 assert_eq!(
3475 second[0].patch,
3476 json!({"state": {"price": 2, "fills": [20]}})
3477 );
3478 assert_eq!(second[0].append, vec!["state.fills".to_string()]);
3479
3480 requests.request("Pool", &key);
3482 assert!(requests.is_requested("Pool", &key));
3483 let mut third = update(3, 30);
3484 assert_eq!(third.len(), 2);
3485 assert!(
3486 !third[0].is_whole_entity(),
3487 "the change itself is untouched"
3488 );
3489 assert_eq!(
3490 third[0].patch,
3491 json!({"state": {"price": 3, "fills": [30]}})
3492 );
3493 assert_eq!(third[1].key, key);
3494 assert_eq!(third[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
3495 assert_eq!(
3496 third[1].patch,
3497 json!({
3498 "id": {"address": "pool_1"},
3499 "state": {"price": 3, "fills": [10, 20, 30]},
3500 }),
3501 "the whole entity, without its non-emitted field"
3502 );
3503 assert!(
3504 third[1].append.is_empty(),
3505 "its arrays are whole, not appended to"
3506 );
3507 assert!(requests.is_empty(), "a request is answered once");
3508
3509 let fourth = update(4, 40);
3510 assert_eq!(fourth.len(), 1);
3511 assert!(!fourth[0].is_whole_entity());
3512 assert_eq!(
3513 fourth[0].patch,
3514 json!({"state": {"price": 4, "fills": [40]}})
3515 );
3516 }
3517
3518 #[test]
3521 fn a_request_is_answered_by_the_next_call_whatever_it_changes() {
3522 let bytecode = pool_bytecode();
3523 let (mut vm, requests) = linked_vm(Default::default());
3524 process(&mut vm, &bytecode, pool_event_for("pool_1", 1, 10));
3525 process(&mut vm, &bytecode, pool_event_for("pool_2", 1, 10));
3526
3527 requests.request("Pool", &json!("pool_1"));
3528 let mut next = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20));
3529 assert_eq!(next.len(), 2);
3530 assert_eq!(next[0].key, json!("pool_2"));
3531 assert!(!next[0].is_whole_entity());
3532 assert_eq!(next[1].key, json!("pool_1"));
3533 assert_eq!(next[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
3534 assert_eq!(
3535 next[1].patch,
3536 json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
3537 );
3538 assert!(requests.is_empty());
3539
3540 requests.request("Pool", &json!("pool_2"));
3542 let answered = vm
3543 .apply_resolver_result(&bytecode, "no-such-request", json!({}))
3544 .unwrap();
3545 assert_eq!(answered.len(), 1);
3546 assert_eq!(answered[0].key, json!("pool_2"));
3547 assert_eq!(answered[0].whole_entity_mark(), Some(WholeEntity::Resent));
3548
3549 let quiet = process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
3550 assert_eq!(quiet.len(), 1, "a request is answered once");
3551 }
3552
3553 #[test]
3556 fn a_request_for_an_entity_the_vm_does_not_hold_is_dropped() {
3557 let bytecode = pool_bytecode();
3558 let (mut vm, requests) = linked_vm(Default::default());
3559 requests.request("Pool", &json!("other_pool"));
3560 let mutations = process(&mut vm, &bytecode, pool_event(1, 10));
3561 assert_eq!(mutations.len(), 1);
3562 assert_eq!(mutations[0].whole_entity_mark(), Some(WholeEntity::Created));
3563 assert!(requests.is_empty(), "dropped, not kept for later");
3564
3565 let created = process(&mut vm, &bytecode, pool_event_for("other_pool", 1, 10));
3566 assert_eq!(created.len(), 1);
3567 assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
3568 }
3569
3570 #[test]
3571 fn whole_entity_requests_are_bounded() {
3572 let requests = crate::vm::WholeEntityRequests::new(2);
3573 for key in ["a", "b", "c"] {
3574 requests.request("Pool", &json!(key));
3575 }
3576 assert_eq!(requests.len(), 2);
3577 assert!(
3578 !requests.is_requested("Pool", &json!("a")),
3579 "the oldest is forgotten"
3580 );
3581 assert!(requests.is_requested("Pool", &json!("c")));
3582 }
3583
3584 #[test]
3587 fn an_entitys_first_mutation_is_marked_as_its_creation() {
3588 let bytecode = pool_bytecode();
3589 let (mut vm, _requests) = linked_vm(Default::default());
3590
3591 let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
3592 assert_eq!(first.len(), 1, "the creation needs no second mutation");
3593 assert_eq!(
3594 first[0].take_whole_entity_mark(),
3595 Some(WholeEntity::Created)
3596 );
3597 assert_eq!(
3598 first[0].patch,
3599 json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}}),
3600 "the whole row, without its non-emitted field"
3601 );
3602 assert_eq!(first[0].append, vec!["state.fills".to_string()]);
3603
3604 let second = process(&mut vm, &bytecode, pool_event(2, 20));
3605 assert_eq!(second[0].whole_entity_mark(), None);
3606 assert_eq!(
3607 second[0].patch,
3608 json!({"state": {"price": 2, "fills": [20]}})
3609 );
3610 }
3611
3612 #[test]
3615 fn a_vm_without_whole_entity_requests_marks_no_creation() {
3616 let bytecode = pool_bytecode();
3617 let mut vm = VmContext::new();
3618 let first = process(&mut vm, &bytecode, pool_event(1, 10));
3619 assert!(!first[0].is_whole_entity());
3620 assert!(first[0].patch.get(crate::WHOLE_ENTITY_MARKER).is_none());
3621 }
3622
3623 #[test]
3627 fn a_row_started_without_a_mutation_is_created_by_its_first_one() {
3628 let bytecode = pool_bytecode();
3629 let (mut vm, _requests) = linked_vm(Default::default());
3630 let config = vm
3631 .process_event(
3632 &bytecode,
3633 json!({"__account_address": "pool_1", "fee": 30}),
3634 "amm::PoolConfigState",
3635 None,
3636 None,
3637 )
3638 .unwrap();
3639 assert!(config.is_empty(), "the config handler emits nothing");
3640
3641 let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
3642 assert_eq!(
3643 first[0].take_whole_entity_mark(),
3644 Some(WholeEntity::Created)
3645 );
3646 assert_eq!(
3647 first[0].patch,
3648 json!({
3649 "id": {"address": "pool_1"},
3650 "config": {"fee": 30},
3651 "state": {"price": 1, "fills": [10]},
3652 })
3653 );
3654 let second = process(&mut vm, &bytecode, pool_event(2, 20));
3655 assert!(!second[0].is_whole_entity());
3656 }
3657
3658 #[test]
3661 fn an_entity_the_vm_dropped_is_created_again() {
3662 let bytecode = pool_bytecode();
3663 let (mut vm, _requests) = linked_vm(crate::vm::StateTableConfig {
3664 max_entries: 2,
3665 ..Default::default()
3666 });
3667 for pool in ["a", "b", "c"] {
3668 let created = process(&mut vm, &bytecode, pool_event_for(pool, 1, 1));
3669 assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
3670 }
3671 assert!(vm.get_entity_state(0, &json!("a")).is_none(), "dropped");
3672
3673 let mut again = process(&mut vm, &bytecode, pool_event_for("a", 2, 2));
3674 assert_eq!(
3675 again[0].take_whole_entity_mark(),
3676 Some(WholeEntity::Created)
3677 );
3678 assert_eq!(
3679 again[0].patch,
3680 json!({"id": {"address": "a"}, "state": {"price": 2, "fills": [2]}})
3681 );
3682 let changed = process(&mut vm, &bytecode, pool_event_for("c", 2, 2));
3683 assert!(!changed[0].is_whole_entity(), "c was never dropped");
3684 }
3685
3686 #[test]
3691 fn a_restored_row_never_emitted_is_sent_whole_on_request() {
3692 let bytecode = pool_bytecode();
3693 let (mut vm, _requests) = linked_vm(Default::default());
3694 let config = vm
3695 .process_event(
3696 &bytecode,
3697 json!({"__account_address": "pool_1", "fee": 30}),
3698 "amm::PoolConfigState",
3699 None,
3700 None,
3701 )
3702 .unwrap();
3703 assert!(config.is_empty());
3704 let snapshot = vm.dump();
3705
3706 let (mut restored, requests) = linked_vm(Default::default());
3707 restored.hydrate(snapshot);
3708 let first = process(&mut restored, &bytecode, pool_event(1, 10));
3709 assert!(!first[0].is_whole_entity(), "counted as emitted already");
3710 assert_eq!(
3711 first[0].patch,
3712 json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
3713 );
3714
3715 requests.request("Pool", &first[0].key);
3716 let next = process(&mut restored, &bytecode, pool_event_for("pool_2", 1, 1));
3717 let resent = next
3718 .iter()
3719 .find(|mutation| mutation.key == json!("pool_1"))
3720 .expect("the requested entity follows the next call");
3721 assert_eq!(resent.whole_entity_mark(), Some(WholeEntity::Resent));
3722 let mut resent = resent.clone();
3723 resent.take_whole_entity_mark();
3724 assert_eq!(
3725 resent.patch,
3726 json!({
3727 "id": {"address": "pool_1"},
3728 "config": {"fee": 30},
3729 "state": {"price": 1, "fills": [10]},
3730 })
3731 );
3732 }
3733
3734 #[test]
3738 fn a_resolver_result_for_an_entity_not_held_creates_it() {
3739 let bytecode = pool_bytecode();
3740 let (mut vm, _requests) = linked_vm(Default::default());
3741 process(&mut vm, &bytecode, pool_event(1, 10));
3742 for key in ["pool_1", "pool_2"] {
3743 vm.enqueue_resolver_request(
3744 String::new(),
3745 crate::ast::ResolverType::Token,
3746 json!(key),
3747 crate::vm::ResolverTarget {
3748 state_id: 0,
3749 entity_name: "Pool".to_string(),
3750 primary_key: json!(key),
3751 extracts: vec![crate::ast::ResolverExtractSpec {
3752 target_path: "meta.name".to_string(),
3753 source_path: Some("name".to_string()),
3754 transform: None,
3755 }],
3756 },
3757 );
3758 }
3759 let mut applied = Vec::new();
3760 for request in vm.take_resolver_requests() {
3761 applied.extend(
3762 vm.apply_resolver_result(&bytecode, &request.cache_key, json!({"name": "n"}))
3763 .unwrap(),
3764 );
3765 }
3766 applied.sort_by_key(|mutation| mutation.key.to_string());
3767
3768 assert_eq!(applied[0].key, json!("pool_1"));
3769 assert!(!applied[0].is_whole_entity());
3770 assert_eq!(applied[0].patch, json!({"meta": {"name": "n"}}));
3771 assert_eq!(applied[1].key, json!("pool_2"));
3772 assert_eq!(
3773 applied[1].take_whole_entity_mark(),
3774 Some(WholeEntity::Created)
3775 );
3776 assert_eq!(applied[1].patch, json!({"meta": {"name": "n"}}));
3777 }
3778
3779 #[test]
3782 fn a_restored_entity_is_not_created_again() {
3783 let bytecode = pool_bytecode();
3784 let (mut vm, _requests) = linked_vm(Default::default());
3785 process(&mut vm, &bytecode, pool_event(1, 10));
3786 let snapshot = vm.dump();
3787
3788 let (mut restored, _requests) = linked_vm(Default::default());
3789 restored.hydrate(snapshot);
3790 let next = process(&mut restored, &bytecode, pool_event(2, 20));
3791 assert!(!next[0].is_whole_entity());
3792 assert_eq!(next[0].patch, json!({"state": {"price": 2, "fills": [20]}}));
3793 }
3794
3795 #[test]
3796 fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
3797 let bytecode = MultiEntityBytecode::from_single(
3798 "PumpfunToken".to_string(),
3799 direct_pda_to_primary_key_spec(),
3800 0,
3801 );
3802 let mut vm = VmContext::new();
3803
3804 let mutations = vm
3805 .process_event(
3806 &bytecode,
3807 json!({
3808 "__account_address": "bonding_curve_1",
3809 "__resolved_primary_key": "mint_1",
3810 "virtual_token_reserves": 42,
3811 }),
3812 "pump::BondingCurveState",
3813 None,
3814 None,
3815 )
3816 .unwrap();
3817
3818 assert_eq!(mutations.len(), 1);
3819 assert_eq!(mutations[0].key, json!("mint_1"));
3820 assert_eq!(
3821 mutations[0].patch["id"]["bonding_curve"],
3822 json!("bonding_curve_1")
3823 );
3824 assert_eq!(
3825 mutations[0].patch["reserves"]["virtual_token_reserves"],
3826 json!(42)
3827 );
3828 }
3829
3830 #[test]
3831 fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
3832 let bytecode = MultiEntityBytecode::from_single(
3833 "OreRound".to_string(),
3834 pda_to_intermediate_lookup_spec(),
3835 0,
3836 );
3837 let mut vm = VmContext::new();
3838
3839 let mutations = vm
3840 .process_event(
3841 &bytecode,
3842 json!({
3843 "__account_address": "entropy_var_1",
3844 "__resolved_primary_key": "round_address_1",
3845 "end_at": 123,
3846 }),
3847 "entropy::VarState",
3848 None,
3849 None,
3850 )
3851 .unwrap();
3852
3853 assert!(mutations.is_empty());
3854 }
3855
3856 #[test]
3857 fn embedded_account_key_wins_over_an_unrelated_resolved_key() {
3858 let bytecode = MultiEntityBytecode::from_single(
3859 "OreTreasury".to_string(),
3860 embedded_account_key_spec(),
3861 0,
3862 );
3863 let mut vm = VmContext::new();
3864
3865 let mutations = vm
3866 .process_event(
3867 &bytecode,
3868 json!({
3869 "__account_address": "treasury_pda",
3870 "__resolved_primary_key": "round_address",
3871 }),
3872 "ore::TreasuryState",
3873 None,
3874 None,
3875 )
3876 .unwrap();
3877
3878 assert_eq!(mutations.len(), 1);
3879 assert_eq!(mutations[0].key, json!("treasury_pda"));
3880 assert_eq!(mutations[0].patch["id"]["address"], json!("treasury_pda"));
3881 }
3882
3883 fn instruction_handler(
3884 type_name: &str,
3885 key_resolution: KeyResolutionStrategy,
3886 mappings: Vec<SerializableFieldMapping>,
3887 ) -> SerializableHandlerSpec {
3888 SerializableHandlerSpec {
3889 source: SourceSpec::Source {
3890 program_id: None,
3891 discriminator: None,
3892 type_name: type_name.to_string(),
3893 serialization: None,
3894 is_account: false,
3895 },
3896 key_resolution,
3897 mappings,
3898 conditions: vec![],
3899 emit: true,
3900 }
3901 }
3902
3903 fn split_position_spec() -> TypedStreamSpec<Value> {
3908 let embedded = |field: &str| KeyResolutionStrategy::Embedded {
3909 primary_field: FieldPath::new(&["accounts", field]),
3910 };
3911 TypedStreamSpec::from_serializable(SerializableStreamSpec {
3912 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3913 state_name: "Position".to_string(),
3914 program_id: None,
3915 idl: None,
3916 identity: IdentitySpec {
3917 primary_keys: vec!["id.position_address".to_string()],
3918 lookup_indexes: vec![],
3919 },
3920 handlers: vec![
3921 instruction_handler(
3922 "amm::SplitPositionIxState",
3923 embedded("first_position"),
3924 vec![
3925 mapping(
3926 "id.position_address",
3927 &["accounts", "first_position"],
3928 PopulationStrategy::SetOnce,
3929 ),
3930 mapping(
3931 "activity.split_source_count",
3932 &["data"],
3933 PopulationStrategy::Count,
3934 ),
3935 ],
3936 ),
3937 instruction_handler(
3938 "amm::SplitPositionIxState",
3939 embedded("second_position"),
3940 vec![
3941 mapping(
3942 "id.position_address",
3943 &["accounts", "second_position"],
3944 PopulationStrategy::SetOnce,
3945 ),
3946 mapping(
3947 "activity.split_child_count",
3948 &["data"],
3949 PopulationStrategy::Count,
3950 ),
3951 ],
3952 ),
3953 ],
3954 sections: vec![],
3955 field_mappings: BTreeMap::new(),
3956 resolver_hooks: vec![],
3957 instruction_hooks: vec![InstructionHook {
3958 instruction_type: "amm::SplitPositionIxState".to_string(),
3959 actions: vec![HookAction::SetField {
3960 target_field: "activity.last_split_source_slot".to_string(),
3961 source: MappingSource::FromSource {
3962 path: FieldPath::new(&["data", "slot"]),
3963 default: None,
3964 transform: None,
3965 },
3966 condition: None,
3967 }],
3968 lookup_by: Some(FieldPath::new(&["accounts", "first_position"])),
3969 }],
3970 resolver_specs: vec![],
3971 computed_fields: vec![],
3972 computed_field_specs: vec![],
3973 content_hash: None,
3974 views: vec![],
3975 scope_field: None,
3976 })
3977 }
3978
3979 fn split_position_event(first: &str, second: &str) -> Value {
3980 json!({
3981 "accounts": { "first_position": first, "second_position": second },
3982 "data": { "slot": 77 },
3983 })
3984 }
3985
3986 #[test]
3987 fn one_instruction_routes_each_key_to_its_own_entity() {
3988 let bytecode =
3989 MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
3990 let mut vm = VmContext::new();
3991
3992 for _ in 0..2 {
3993 vm.process_event(
3994 &bytecode,
3995 split_position_event("source_pos", "child_pos"),
3996 "amm::SplitPositionIxState",
3997 None,
3998 None,
3999 )
4000 .unwrap();
4001 }
4002 vm.process_event(
4003 &bytecode,
4004 split_position_event("child_pos", "grandchild_pos"),
4005 "amm::SplitPositionIxState",
4006 None,
4007 None,
4008 )
4009 .unwrap();
4010
4011 let state = |key: &str| vm.get_entity_state(0, &json!(key)).unwrap();
4012
4013 let source = state("source_pos");
4014 assert_eq!(source["id"]["position_address"], json!("source_pos"));
4015 assert_eq!(source["activity"]["split_source_count"], json!(2));
4016 assert_eq!(source["activity"].get("split_child_count"), None);
4017 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
4018
4019 let child = state("child_pos");
4021 assert_eq!(child["id"]["position_address"], json!("child_pos"));
4022 assert_eq!(child["activity"]["split_child_count"], json!(2));
4023 assert_eq!(child["activity"]["split_source_count"], json!(1));
4024 assert_eq!(child["activity"]["last_split_source_slot"], json!(77));
4025
4026 let grandchild = state("grandchild_pos");
4027 assert_eq!(grandchild["activity"]["split_child_count"], json!(1));
4028 assert_eq!(grandchild["activity"].get("split_source_count"), None);
4029 assert_eq!(grandchild["activity"].get("last_split_source_slot"), None);
4030 }
4031
4032 #[test]
4033 fn one_instruction_emits_one_mutation_per_key() {
4034 let bytecode =
4035 MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
4036 let mut vm = VmContext::new();
4037
4038 let mutations = vm
4039 .process_event(
4040 &bytecode,
4041 split_position_event("source_pos", "child_pos"),
4042 "amm::SplitPositionIxState",
4043 None,
4044 None,
4045 )
4046 .unwrap();
4047
4048 assert_eq!(mutations.len(), 2);
4049 assert_eq!(mutations[0].key, json!("source_pos"));
4050 assert_eq!(
4051 mutations[0].patch["activity"]["split_source_count"],
4052 json!(1)
4053 );
4054 assert_eq!(
4055 mutations[0].patch["activity"]["last_split_source_slot"],
4056 json!(77)
4057 );
4058 assert_eq!(
4059 mutations[0].patch["activity"].get("split_child_count"),
4060 None
4061 );
4062 assert_eq!(mutations[1].key, json!("child_pos"));
4063 assert_eq!(
4064 mutations[1].patch["activity"]["split_child_count"],
4065 json!(1)
4066 );
4067 assert_eq!(
4068 mutations[1].patch["activity"].get("split_source_count"),
4069 None
4070 );
4071 }
4072
4073 #[test]
4074 fn handlers_with_identical_keys_still_merge_into_one_segment() {
4075 let mut spec = split_position_spec();
4076 spec.instruction_hooks.clear();
4077 spec.handlers[1].key_resolution = KeyResolutionStrategy::Embedded {
4080 primary_field: FieldPath::new(&["accounts", "first_position"]),
4081 };
4082 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
4083 let handler = &bytecode.entities["Position"].handlers["amm::SplitPositionIxState"];
4084
4085 assert!(!handler
4086 .iter()
4087 .any(|op| matches!(op, super::OpCode::SegmentBoundary)));
4088 assert_eq!(
4089 handler
4090 .iter()
4091 .filter(|op| matches!(op, super::OpCode::ReadOrInitState { .. }))
4092 .count(),
4093 1
4094 );
4095 }
4096
4097 #[test]
4098 fn hook_without_a_matching_key_gets_its_own_segment() {
4099 let mut spec = split_position_spec();
4100 spec.handlers.truncate(1);
4101 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
4102 primary_field: FieldPath::new(&["accounts", "second_position"]),
4103 };
4104 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
4105 "id.position_address",
4106 &["accounts", "second_position"],
4107 PopulationStrategy::SetOnce,
4108 ));
4109 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
4110 let mut vm = VmContext::new();
4111
4112 vm.process_event(
4113 &bytecode,
4114 split_position_event("source_pos", "child_pos"),
4115 "amm::SplitPositionIxState",
4116 None,
4117 None,
4118 )
4119 .unwrap();
4120
4121 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
4123 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
4124 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
4125 assert_eq!(child["activity"].get("last_split_source_slot"), None);
4126 }
4127
4128 fn add_owner_index(spec: &mut TypedStreamSpec<Value>) {
4130 spec.identity.lookup_indexes.push(LookupIndexSpec {
4131 field_name: "id.owner".to_string(),
4132 temporal_field: None,
4133 });
4134 spec.handlers
4135 .push(crate::ast::TypedHandlerSpec::from_serializable(
4136 SerializableHandlerSpec {
4137 source: SourceSpec::Source {
4138 program_id: None,
4139 discriminator: None,
4140 type_name: "amm::PositionState".to_string(),
4141 serialization: None,
4142 is_account: true,
4143 },
4144 key_resolution: KeyResolutionStrategy::Embedded {
4145 primary_field: FieldPath::new(&["__account_address"]),
4146 },
4147 mappings: vec![
4148 mapping(
4149 "id.position_address",
4150 &["__account_address"],
4151 PopulationStrategy::SetOnce,
4152 ),
4153 mapping("id.owner", &["owner"], PopulationStrategy::SetOnce),
4154 ],
4155 conditions: vec![],
4156 emit: true,
4157 },
4158 ));
4159 }
4160
4161 fn index_owner(
4162 vm: &mut VmContext,
4163 bytecode: &MultiEntityBytecode,
4164 position: &str,
4165 owner: &str,
4166 ) {
4167 vm.process_event(
4168 bytecode,
4169 json!({ "__account_address": position, "owner": owner }),
4170 "amm::PositionState",
4171 None,
4172 None,
4173 )
4174 .unwrap();
4175 }
4176
4177 #[test]
4178 fn unmatched_hook_next_to_other_keys_resolves_lookup_index_fields() {
4179 let mut spec = split_position_spec();
4180 spec.handlers.truncate(1);
4181 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
4182 primary_field: FieldPath::new(&["accounts", "second_position"]),
4183 };
4184 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
4185 "id.position_address",
4186 &["accounts", "second_position"],
4187 PopulationStrategy::SetOnce,
4188 ));
4189 add_owner_index(&mut spec);
4190 spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
4192
4193 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
4194 let mut vm = VmContext::new();
4195 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
4196 vm.process_event(
4197 &bytecode,
4198 json!({
4199 "accounts": {
4200 "first_position": "source_pos",
4201 "second_position": "child_pos",
4202 "owner": "owner_1",
4203 },
4204 "data": { "slot": 77 },
4205 }),
4206 "amm::SplitPositionIxState",
4207 None,
4208 None,
4209 )
4210 .unwrap();
4211
4212 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
4213 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
4214 assert!(vm.get_entity_state(0, &json!("owner_1")).is_none());
4215 }
4216
4217 #[test]
4218 fn unmatched_hook_replays_after_its_lookup_index_is_populated() {
4219 let mut spec = split_position_spec();
4220 spec.handlers.truncate(1);
4221 spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
4222 primary_field: FieldPath::new(&["accounts", "second_position"]),
4223 };
4224 spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
4225 "id.position_address",
4226 &["accounts", "second_position"],
4227 PopulationStrategy::SetOnce,
4228 ));
4229 add_owner_index(&mut spec);
4230 spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
4231
4232 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
4233 let mut vm = VmContext::new();
4234 let mutations = vm
4235 .process_event(
4236 &bytecode,
4237 json!({
4238 "accounts": { "owner": "owner_1", "second_position": "child_pos" },
4239 "data": { "slot": 77 },
4240 }),
4241 "amm::SplitPositionIxState",
4242 None,
4243 None,
4244 )
4245 .unwrap();
4246 assert_eq!(mutations.len(), 1);
4247 assert_eq!(mutations[0].key, json!("child_pos"));
4248
4249 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
4250
4251 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
4252 assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
4253 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
4254 assert_eq!(child["activity"].get("last_split_source_slot"), None);
4255 }
4256
4257 #[test]
4258 fn a_segment_that_misses_its_key_is_replayed_alone() {
4259 let mut spec = split_position_spec();
4260 spec.instruction_hooks.clear();
4261 spec.handlers.swap(0, 1);
4264 spec.handlers[1].key_resolution = KeyResolutionStrategy::Lookup {
4265 primary_field: FieldPath::new(&["accounts", "source_pda"]),
4266 index_name: None,
4267 };
4268 spec.handlers[1].mappings.remove(0);
4269 add_owner_index(&mut spec);
4270 spec.instruction_hooks.push(InstructionHook {
4271 instruction_type: "amm::RegisterIxState".to_string(),
4272 actions: vec![HookAction::RegisterPdaMapping {
4273 pda_field: FieldPath::new(&["accounts", "source_pda"]),
4274 seed_field: FieldPath::new(&["accounts", "owner"]),
4275 lookup_name: "default_pda_lookup".to_string(),
4276 sibling: None,
4277 seed_expr: None,
4278 guard: None,
4279 }],
4280 lookup_by: None,
4281 });
4282 let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
4283 let mut vm = VmContext::new();
4284 index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
4285
4286 let mutations = vm
4287 .process_event(
4288 &bytecode,
4289 json!({
4290 "accounts": { "source_pda": "pda_1", "second_position": "child_pos" },
4291 "data": {},
4292 }),
4293 "amm::SplitPositionIxState",
4294 None,
4295 None,
4296 )
4297 .unwrap();
4298 assert_eq!(mutations.len(), 1);
4300 assert_eq!(mutations[0].key, json!("child_pos"));
4301
4302 let mutations = vm
4303 .process_event(
4304 &bytecode,
4305 json!({ "accounts": { "source_pda": "pda_1", "owner": "owner_1" } }),
4306 "amm::RegisterIxState",
4307 None,
4308 None,
4309 )
4310 .unwrap();
4311 assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
4313 assert_eq!(mutations[0].key, json!("source_pos"));
4314 let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
4315 assert_eq!(source["activity"]["split_source_count"], json!(1));
4316 let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
4317 assert_eq!(child["activity"]["split_child_count"], json!(1));
4318 }
4319
4320 #[test]
4325 fn a_hook_whose_lookup_field_is_absent_skips_only_its_segment() {
4326 let spec: TypedStreamSpec<Value> =
4327 TypedStreamSpec::from_serializable(SerializableStreamSpec {
4328 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
4329 state_name: "Game".to_string(),
4330 program_id: None,
4331 idl: None,
4332 identity: IdentitySpec {
4333 primary_keys: vec!["id.game_id".to_string()],
4334 lookup_indexes: vec![],
4335 },
4336 handlers: vec![instruction_handler(
4337 "game::StartGameIxState",
4338 KeyResolutionStrategy::Embedded {
4339 primary_field: FieldPath::new(&["data", "game_id"]),
4340 },
4341 vec![
4342 SerializableFieldMapping {
4343 source: MappingSource::FromSource {
4344 path: FieldPath::new(&["data", "game_id"]),
4345 default: None,
4346 transform: Some(crate::ast::Transformation::HexEncode),
4347 },
4348 ..mapping("id.game_id", &[], PopulationStrategy::SetOnce)
4349 },
4350 mapping("stats.fee", &["data", "fee"], PopulationStrategy::SetOnce),
4351 ],
4352 )],
4353 sections: vec![],
4354 field_mappings: BTreeMap::new(),
4355 resolver_hooks: vec![],
4356 instruction_hooks: vec![InstructionHook {
4357 instruction_type: "game::StartGameIxState".to_string(),
4358 actions: vec![HookAction::SetField {
4359 target_field: "stats.started_slot".to_string(),
4360 source: MappingSource::FromSource {
4361 path: FieldPath::new(&["data", "slot"]),
4362 default: None,
4363 transform: None,
4364 },
4365 condition: None,
4366 }],
4367 lookup_by: Some(FieldPath::new(&["accounts", "game_id"])),
4369 }],
4370 resolver_specs: vec![],
4371 computed_fields: vec![],
4372 computed_field_specs: vec![],
4373 content_hash: None,
4374 views: vec![],
4375 scope_field: None,
4376 });
4377 let bytecode = MultiEntityBytecode::from_single("Game".to_string(), spec, 0);
4378 let mut vm = VmContext::new();
4379
4380 let mutations = vm
4381 .process_event(
4382 &bytecode,
4383 json!({
4384 "accounts": { "game_account": "game_pda" },
4385 "data": { "game_id": [1, 2, 3, 4], "fee": 5, "slot": 9 },
4386 }),
4387 "game::StartGameIxState",
4388 None,
4389 None,
4390 )
4391 .unwrap();
4392
4393 assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
4394 assert_eq!(mutations[0].key, json!("01020304"));
4395 let game = vm.get_entity_state(0, &json!("01020304")).unwrap();
4396 assert_eq!(game["id"]["game_id"], json!("01020304"));
4397 assert_eq!(game["stats"]["fee"], json!(5));
4398 assert_eq!(game["stats"].get("started_slot"), None);
4399 assert_eq!(vm.get_entity_state(0, &json!(null)), None);
4401 }
4402
4403 mod fingerprint {
4404 use super::super::{EntityBytecode, MultiEntityBytecode, OpCode};
4405 use crate::ast::FieldPath;
4406 use serde_json::json;
4407 use std::collections::{HashMap, HashSet};
4408
4409 fn sample_handler() -> Vec<OpCode> {
4410 vec![
4411 OpCode::LoadEventField {
4412 path: FieldPath::new(&["id", "mint"]),
4413 dest: 0,
4414 default: Some(json!({"b": 2, "a": 1})),
4415 },
4416 OpCode::CreateObject { dest: 1 },
4417 OpCode::SetField {
4418 object: 1,
4419 path: "state.volume".to_string(),
4420 value: 0,
4421 },
4422 OpCode::EmitMutation {
4423 entity_name: "PumpfunToken".to_string(),
4424 key: 0,
4425 state: 1,
4426 },
4427 ]
4428 }
4429
4430 fn bytecode(handler_insert_order: &[&str]) -> MultiEntityBytecode {
4431 let mut handlers = HashMap::new();
4432 for event_type in handler_insert_order {
4433 handlers.insert(event_type.to_string(), sample_handler());
4434 }
4435 let mut entities = HashMap::new();
4436 entities.insert(
4437 "PumpfunToken".to_string(),
4438 EntityBytecode {
4439 state_id: 0,
4440 handlers,
4441 entity_name: "PumpfunToken".to_string(),
4442 when_events: HashSet::from(["pump::BuyIxState".to_string()]),
4443 non_emitted_fields: HashSet::new(),
4444 computed_paths: vec!["state.market_cap".to_string()],
4445 computed_fields_evaluator: None,
4446 },
4447 );
4448 let mut event_routing = HashMap::new();
4449 event_routing.insert(
4450 "pump::TokenState".to_string(),
4451 vec!["PumpfunToken".to_string()],
4452 );
4453 MultiEntityBytecode {
4454 entities,
4455 event_routing,
4456 when_events: HashSet::from(["pump::BuyIxState".to_string()]),
4457 proto_router: crate::proto_router::ProtoRouter::new(),
4458 }
4459 }
4460
4461 #[test]
4462 fn stable_across_recomputation_and_insertion_order() {
4463 let a = bytecode(&["pump::TokenState", "pump::BuyIxState"]);
4464 let b = bytecode(&["pump::BuyIxState", "pump::TokenState"]);
4465 assert_eq!(a.fingerprint(), a.fingerprint());
4466 assert_eq!(a.fingerprint(), b.fingerprint());
4467 assert_eq!(a.fingerprint().len(), 64);
4468 }
4469
4470 #[test]
4471 fn changes_when_an_opcode_operand_changes() {
4472 let base = bytecode(&["pump::TokenState"]);
4473 let mut modified = bytecode(&["pump::TokenState"]);
4474 let entity = modified.entities.get_mut("PumpfunToken").unwrap();
4475 match &mut entity.handlers.get_mut("pump::TokenState").unwrap()[2] {
4476 OpCode::SetField { path, .. } => *path = "state.volume_usd".to_string(),
4477 other => panic!("unexpected opcode: {other:?}"),
4478 }
4479 assert_ne!(base.fingerprint(), modified.fingerprint());
4480 }
4481
4482 #[test]
4483 fn changes_when_routing_changes() {
4484 let base = bytecode(&["pump::TokenState"]);
4485 let mut modified = bytecode(&["pump::TokenState"]);
4486 modified.event_routing.insert(
4487 "pump::SellIxState".to_string(),
4488 vec!["PumpfunToken".to_string()],
4489 );
4490 assert_ne!(base.fingerprint(), modified.fingerprint());
4491 }
4492
4493 #[test]
4500 fn golden_hash_pins_the_canonical_encoding() {
4501 let expected = "acf7ea8bf39ca830a9e6c2b2d2fb471da68e57e24bbb4c256964a42a498f75f2";
4502 assert_eq!(bytecode(&["pump::TokenState"]).fingerprint(), expected);
4503 }
4504 }
4505}