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