1use crate::ast::*;
2use serde_json::Value;
3use std::collections::{HashMap, HashSet};
4use tracing;
5
6pub type Register = usize;
7
8fn stop_field_path(target_path: &str) -> String {
9 format!("__stop:{}", target_path)
10}
11
12#[derive(Debug, Clone)]
13pub enum OpCode {
14 AbortIfNullKey {
21 key: Register,
22 is_account_event: bool,
23 },
24 LoadEventField {
25 path: FieldPath,
26 dest: Register,
27 default: Option<Value>,
28 },
29 LoadConstant {
30 value: Value,
31 dest: Register,
32 },
33 CopyRegister {
34 source: Register,
35 dest: Register,
36 },
37 CopyRegisterIfNull {
39 source: Register,
40 dest: Register,
41 },
42 GetEventType {
43 dest: Register,
44 },
45 CreateObject {
46 dest: Register,
47 },
48 SetField {
49 object: Register,
50 path: String,
51 value: Register,
52 },
53 SetFields {
54 object: Register,
55 fields: Vec<(String, Register)>,
56 },
57 GetField {
58 object: Register,
59 path: String,
60 dest: Register,
61 },
62 ReadOrInitState {
63 state_id: u32,
64 key: Register,
65 default: Value,
66 dest: Register,
67 },
68 UpdateState {
69 state_id: u32,
70 key: Register,
71 value: Register,
72 },
73 AppendToArray {
74 object: Register,
75 path: String,
76 value: Register,
77 },
78 GetCurrentTimestamp {
79 dest: Register,
80 },
81 CreateEvent {
82 dest: Register,
83 event_value: Register,
84 },
85 CreateCapture {
86 dest: Register,
87 capture_value: Register,
88 },
89 Transform {
90 source: Register,
91 dest: Register,
92 transformation: Transformation,
93 },
94 EmitMutation {
95 entity_name: String,
96 key: Register,
97 state: Register,
98 },
99 SetFieldIfNull {
100 object: Register,
101 path: String,
102 value: Register,
103 },
104 SetFieldMax {
105 object: Register,
106 path: String,
107 value: Register,
108 },
109 UpdateTemporalIndex {
110 state_id: u32,
111 index_name: String,
112 lookup_value: Register,
113 primary_key: Register,
114 timestamp: Register,
115 },
116 LookupTemporalIndex {
117 state_id: u32,
118 index_name: String,
119 lookup_value: Register,
120 timestamp: Register,
121 dest: Register,
122 },
123 UpdateLookupIndex {
124 state_id: u32,
125 index_name: String,
126 lookup_value: Register,
127 primary_key: Register,
128 },
129 LookupIndex {
130 state_id: u32,
131 index_name: String,
132 lookup_value: Register,
133 dest: Register,
134 },
135 SetFieldSum {
137 object: Register,
138 path: String,
139 value: Register,
140 },
141 SetFieldIncrement {
143 object: Register,
144 path: String,
145 },
146 SetFieldMin {
148 object: Register,
149 path: String,
150 value: Register,
151 },
152 SetFieldWhen {
155 object: Register,
156 path: String,
157 value: Register,
158 when_instruction: String,
159 entity_name: String,
160 key_reg: Register,
161 condition_field: Option<FieldPath>,
162 condition_op: Option<ComparisonOp>,
163 condition_value: Option<Value>,
164 },
165 SetFieldUnlessStopped {
168 object: Register,
169 path: String,
170 value: Register,
171 stop_field: String,
172 stop_instruction: String,
173 entity_name: String,
174 key_reg: Register,
175 },
176 AddToUniqueSet {
179 state_id: u32,
180 set_name: String,
181 value: Register,
182 count_object: Register,
183 count_path: String,
184 },
185 ConditionalSetField {
187 object: Register,
188 path: String,
189 value: Register,
190 condition_field: FieldPath,
191 condition_op: ComparisonOp,
192 condition_value: Value,
193 },
194 ConditionalIncrement {
196 object: Register,
197 path: String,
198 condition_field: FieldPath,
199 condition_op: ComparisonOp,
200 condition_value: Value,
201 },
202 EvaluateComputedFields {
205 state: Register,
206 computed_paths: Vec<String>,
207 },
208 QueueResolver {
210 state_id: u32,
211 entity_name: String,
212 resolver: ResolverType,
213 input_path: Option<String>,
214 input_value: Option<Value>,
215 url_template: Option<Vec<UrlTemplatePart>>,
216 strategy: ResolveStrategy,
217 extracts: Vec<ResolverExtractSpec>,
218 condition: Option<ResolverCondition>,
219 schedule_at: Option<String>,
220 state: Register,
221 key: Register,
222 },
223 UpdatePdaReverseLookup {
226 state_id: u32,
227 lookup_name: String,
228 pda_address: Register,
229 primary_key: Register,
230 },
231}
232
233pub struct EntityBytecode {
234 pub state_id: u32,
235 pub handlers: HashMap<String, Vec<OpCode>>,
236 pub entity_name: String,
237 pub when_events: HashSet<String>,
238 pub non_emitted_fields: HashSet<String>,
239 pub computed_paths: Vec<String>,
240 #[allow(clippy::type_complexity)]
243 pub computed_fields_evaluator: Option<
244 Box<
245 dyn Fn(
246 &mut Value,
247 Option<u64>,
248 i64,
249 )
250 -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
251 + Send
252 + Sync,
253 >,
254 >,
255}
256
257impl std::fmt::Debug for EntityBytecode {
258 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
259 f.debug_struct("EntityBytecode")
260 .field("state_id", &self.state_id)
261 .field("handlers", &self.handlers)
262 .field("entity_name", &self.entity_name)
263 .field("when_events", &self.when_events)
264 .field("non_emitted_fields", &self.non_emitted_fields)
265 .field("computed_paths", &self.computed_paths)
266 .field(
267 "computed_fields_evaluator",
268 &self.computed_fields_evaluator.is_some(),
269 )
270 .finish()
271 }
272}
273
274#[derive(Debug)]
275pub struct MultiEntityBytecode {
276 pub entities: HashMap<String, EntityBytecode>,
277 pub event_routing: HashMap<String, Vec<String>>,
278 pub when_events: HashSet<String>,
279 pub proto_router: crate::proto_router::ProtoRouter,
280}
281
282impl MultiEntityBytecode {
283 pub fn from_single<S>(entity_name: String, spec: TypedStreamSpec<S>, state_id: u32) -> Self {
284 let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
285 let entity_bytecode = compiler.compile_entity();
286
287 let mut entities = HashMap::new();
288 let mut event_routing = HashMap::new();
289 let mut when_events = HashSet::new();
290
291 for event_type in entity_bytecode.handlers.keys() {
292 event_routing
293 .entry(event_type.clone())
294 .or_insert_with(Vec::new)
295 .push(entity_name.clone());
296 }
297
298 when_events.extend(entity_bytecode.when_events.iter().cloned());
299
300 entities.insert(entity_name, entity_bytecode);
301
302 MultiEntityBytecode {
303 entities,
304 event_routing,
305 when_events,
306 proto_router: crate::proto_router::ProtoRouter::new(),
307 }
308 }
309
310 pub fn from_entities(entities_vec: Vec<(String, Box<dyn std::any::Any>, u32)>) -> Self {
311 let entities = HashMap::new();
312 let event_routing = HashMap::new();
313 let when_events = HashSet::new();
314
315 if let Some((_entity_name, _spec_any, _state_id)) = entities_vec.into_iter().next() {
316 panic!("from_entities requires type information - use builder pattern instead");
317 }
318
319 MultiEntityBytecode {
320 entities,
321 event_routing,
322 when_events,
323 proto_router: crate::proto_router::ProtoRouter::new(),
324 }
325 }
326
327 #[allow(clippy::new_ret_no_self)]
328 pub fn new() -> MultiEntityBytecodeBuilder {
329 MultiEntityBytecodeBuilder {
330 entities: HashMap::new(),
331 event_routing: HashMap::new(),
332 when_events: HashSet::new(),
333 proto_router: crate::proto_router::ProtoRouter::new(),
334 }
335 }
336}
337
338pub struct MultiEntityBytecodeBuilder {
339 entities: HashMap<String, EntityBytecode>,
340 event_routing: HashMap<String, Vec<String>>,
341 when_events: HashSet<String>,
342 proto_router: crate::proto_router::ProtoRouter,
343}
344
345impl MultiEntityBytecodeBuilder {
346 pub fn add_entity<S>(
347 self,
348 entity_name: String,
349 spec: TypedStreamSpec<S>,
350 state_id: u32,
351 ) -> Self {
352 self.add_entity_with_evaluator(
353 entity_name,
354 spec,
355 state_id,
356 None::<
357 fn(
358 &mut Value,
359 Option<u64>,
360 i64,
361 )
362 -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>,
363 >,
364 )
365 }
366
367 pub fn add_entity_with_evaluator<S, F>(
368 mut self,
369 entity_name: String,
370 spec: TypedStreamSpec<S>,
371 state_id: u32,
372 evaluator: Option<F>,
373 ) -> Self
374 where
375 F: Fn(
376 &mut Value,
377 Option<u64>,
378 i64,
379 ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
380 + Send
381 + Sync
382 + 'static,
383 {
384 let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
385 let mut entity_bytecode = compiler.compile_entity();
386
387 if let Some(eval) = evaluator {
389 entity_bytecode.computed_fields_evaluator = Some(Box::new(eval));
390 }
391
392 for event_type in entity_bytecode.handlers.keys() {
393 self.event_routing
394 .entry(event_type.clone())
395 .or_default()
396 .push(entity_name.clone());
397 }
398
399 self.when_events
400 .extend(entity_bytecode.when_events.iter().cloned());
401
402 self.entities.insert(entity_name, entity_bytecode);
403 self
404 }
405
406 pub fn build(self) -> MultiEntityBytecode {
407 MultiEntityBytecode {
408 entities: self.entities,
409 event_routing: self.event_routing,
410 when_events: self.when_events,
411 proto_router: self.proto_router,
412 }
413 }
414}
415
416pub struct TypedCompiler<S> {
417 pub spec: TypedStreamSpec<S>,
418 entity_name: String,
419 state_id: u32,
420}
421
422impl<S> TypedCompiler<S> {
423 pub fn new(spec: TypedStreamSpec<S>, entity_name: String) -> Self {
424 TypedCompiler {
425 spec,
426 entity_name,
427 state_id: 0,
428 }
429 }
430
431 pub fn with_state_id(mut self, state_id: u32) -> Self {
432 self.state_id = state_id;
433 self
434 }
435
436 pub fn compile(&self) -> MultiEntityBytecode {
437 let entity_bytecode = self.compile_entity();
438
439 let mut entities = HashMap::new();
440 let mut event_routing = HashMap::new();
441 let mut when_events = HashSet::new();
442
443 for event_type in entity_bytecode.handlers.keys() {
444 event_routing
445 .entry(event_type.clone())
446 .or_insert_with(Vec::new)
447 .push(self.entity_name.clone());
448 }
449
450 when_events.extend(entity_bytecode.when_events.iter().cloned());
451
452 entities.insert(self.entity_name.clone(), entity_bytecode);
453
454 MultiEntityBytecode {
455 entities,
456 event_routing,
457 when_events,
458 proto_router: crate::proto_router::ProtoRouter::new(),
459 }
460 }
461
462 fn resolver_outputs_primary_key_directly(&self, primary_field: &FieldPath) -> bool {
463 if primary_field.segments.as_slice() != ["__account_address"] {
464 return false;
465 }
466
467 let primary_key_leafs: HashSet<&str> = self
468 .spec
469 .identity
470 .primary_keys
471 .iter()
472 .map(|path| path.rsplit('.').next().unwrap_or(path.as_str()))
473 .collect();
474
475 self.spec.instruction_hooks.iter().any(|hook| {
476 hook.actions.iter().any(|action| {
477 matches!(action, HookAction::RegisterPdaMapping { seed_field, .. }
478 if seed_field
479 .segments
480 .last()
481 .is_some_and(|segment| primary_key_leafs.contains(segment.as_str())))
482 })
483 })
484 }
485
486 fn compile_entity(&self) -> EntityBytecode {
487 let mut handlers: HashMap<String, Vec<OpCode>> = HashMap::new();
488 let mut when_events: HashSet<String> = HashSet::new();
489 let mut emit_by_path: HashMap<String, bool> = HashMap::new();
490
491 let mut debug_info = Vec::new();
493 for (index, handler_spec) in self.spec.handlers.iter().enumerate() {
494 let event_type = self.get_event_type(&handler_spec.source);
495 let program_id = match &handler_spec.source {
496 crate::ast::SourceSpec::Source { program_id, .. } => {
497 program_id.as_ref().map(|s| s.as_str()).unwrap_or("null")
498 }
499 };
500 debug_info.push(format!(
501 " [{}] EventType={}, Mappings={}, ProgramId={}",
502 index,
503 event_type,
504 handler_spec.mappings.len(),
505 program_id
506 ));
507 }
508
509 for handler_spec in &self.spec.handlers {
519 for mapping in &handler_spec.mappings {
520 if let Some(when) = &mapping.when {
521 when_events.insert(when.clone());
522 }
523 let entry = emit_by_path
524 .entry(mapping.target_path.clone())
525 .or_insert(false);
526 *entry |= mapping.emit;
527 if mapping.stop.is_some() {
528 emit_by_path
529 .entry(stop_field_path(&mapping.target_path))
530 .or_insert(false);
531 }
532 }
533 let opcodes = self.compile_handler(handler_spec);
534 let event_type = self.get_event_type(&handler_spec.source);
535
536 if let Some(existing_opcodes) = handlers.get_mut(&event_type) {
537 let mut existing_setup = Vec::new();
542 let mut existing_mappings = Vec::new();
543 let mut existing_teardown = Vec::new();
544 let mut section = 0; for opcode in existing_opcodes.iter() {
547 match opcode {
548 OpCode::ReadOrInitState { .. } => {
549 existing_setup.push(opcode.clone());
550 section = 1; }
552 OpCode::UpdateState { .. } => {
553 existing_teardown.push(opcode.clone());
554 section = 2; }
556 OpCode::EmitMutation { .. } => {
557 existing_teardown.push(opcode.clone());
558 }
559 _ if section == 0 => existing_setup.push(opcode.clone()),
560 _ if section == 1 => existing_mappings.push(opcode.clone()),
561 _ => existing_teardown.push(opcode.clone()),
562 }
563 }
564
565 let mut new_mappings = Vec::new();
567 section = 0;
568
569 for opcode in opcodes.iter() {
570 match opcode {
571 OpCode::ReadOrInitState { .. } => {
572 section = 1; }
574 OpCode::UpdateState { .. } | OpCode::EmitMutation { .. } => {
575 section = 2; }
577 _ if section == 1 => {
578 new_mappings.push(opcode.clone());
579 }
580 _ => {} }
582 }
583
584 let mut merged = Vec::new();
586 merged.extend(existing_setup);
587 merged.extend(existing_mappings);
588 merged.extend(new_mappings.clone());
589 merged.extend(existing_teardown);
590
591 *existing_opcodes = merged;
592 } else {
593 handlers.insert(event_type, opcodes);
594 }
595 }
596
597 for hook in &self.spec.instruction_hooks {
599 let event_type = hook.instruction_type.clone();
600
601 let handler_opcodes = handlers.entry(event_type.clone()).or_insert_with(|| {
602 let key_reg = 20;
603 let state_reg = 2;
604 let resolved_key_reg = 19;
605 let temp_reg = 18;
606
607 let mut ops = Vec::new();
608
609 ops.push(OpCode::LoadEventField {
611 path: FieldPath::new(&["__resolved_primary_key"]),
612 dest: resolved_key_reg,
613 default: Some(serde_json::json!(null)),
614 });
615
616 ops.push(OpCode::CopyRegister {
618 source: resolved_key_reg,
619 dest: key_reg,
620 });
621
622 if let Some(lookup_path) = &hook.lookup_by {
624 ops.push(OpCode::LoadEventField {
626 path: lookup_path.clone(),
627 dest: temp_reg,
628 default: None,
629 });
630
631 ops.push(OpCode::Transform {
633 source: temp_reg,
634 dest: temp_reg,
635 transformation: Transformation::HexEncode,
636 });
637
638 ops.push(OpCode::CopyRegisterIfNull {
640 source: temp_reg,
641 dest: key_reg,
642 });
643 }
644
645 ops.push(OpCode::ReadOrInitState {
646 state_id: self.state_id,
647 key: key_reg,
648 default: serde_json::json!({}),
649 dest: state_reg,
650 });
651
652 ops.push(OpCode::UpdateState {
653 state_id: self.state_id,
654 key: key_reg,
655 value: state_reg,
656 });
657
658 ops
659 });
660
661 let hook_opcodes = self.compile_instruction_hook_actions(&hook.actions);
663
664 let insert_pos = handler_opcodes
668 .iter()
669 .position(|op| matches!(op, OpCode::EvaluateComputedFields { .. }))
670 .or_else(|| {
671 handler_opcodes
672 .iter()
673 .position(|op| matches!(op, OpCode::UpdateState { .. }))
674 });
675
676 if let Some(pos) = insert_pos {
677 for (i, opcode) in hook_opcodes.into_iter().enumerate() {
679 handler_opcodes.insert(pos + i, opcode);
680 }
681 }
682 }
683
684 let non_emitted_fields: HashSet<String> = emit_by_path
685 .into_iter()
686 .filter_map(|(path, emit)| if emit { None } else { Some(path) })
687 .collect();
688
689 EntityBytecode {
690 state_id: self.state_id,
691 handlers,
692 entity_name: self.entity_name.clone(),
693 when_events,
694 non_emitted_fields,
695 computed_paths: self.spec.computed_fields.clone(),
696 computed_fields_evaluator: None,
697 }
698 }
699
700 fn compile_handler(&self, spec: &TypedHandlerSpec<S>) -> Vec<OpCode> {
701 let mut ops = Vec::new();
702 let state_reg = 2;
703 let key_reg = 20;
704
705 ops.extend(self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings));
706
707 let is_account_event = matches!(
713 spec.source,
714 SourceSpec::Source {
715 is_account: true,
716 ..
717 }
718 );
719 ops.push(OpCode::AbortIfNullKey {
720 key: key_reg,
721 is_account_event,
722 });
723
724 ops.push(OpCode::ReadOrInitState {
725 state_id: self.state_id,
726 key: key_reg,
727 default: serde_json::json!({}),
728 dest: state_reg,
729 });
730
731 ops.extend(self.compile_temporal_index_update(
738 &spec.key_resolution,
739 key_reg,
740 &spec.mappings,
741 ));
742
743 for mapping in &spec.mappings {
744 ops.extend(self.compile_mapping(mapping, state_reg, key_reg));
745 }
746
747 ops.extend(self.compile_resolvers(state_reg, key_reg));
748
749 ops.push(OpCode::EvaluateComputedFields {
751 state: state_reg,
752 computed_paths: self.spec.computed_fields.clone(),
753 });
754
755 ops.push(OpCode::UpdateState {
756 state_id: self.state_id,
757 key: key_reg,
758 value: state_reg,
759 });
760
761 if spec.emit {
762 ops.push(OpCode::EmitMutation {
763 entity_name: self.entity_name.clone(),
764 key: key_reg,
765 state: state_reg,
766 });
767 }
768
769 ops
770 }
771
772 fn compile_resolvers(&self, state_reg: Register, key_reg: Register) -> Vec<OpCode> {
773 let mut ops = Vec::new();
774
775 for resolver_spec in &self.spec.resolver_specs {
776 let url_template = match &resolver_spec.resolver {
777 ResolverType::Url(config) => match &config.url_source {
778 UrlSource::Template(parts) => Some(parts.clone()),
779 _ => None,
780 },
781 _ => None,
782 };
783
784 ops.push(OpCode::QueueResolver {
785 state_id: self.state_id,
786 entity_name: self.entity_name.clone(),
787 resolver: resolver_spec.resolver.clone(),
788 input_path: resolver_spec.input_path.clone(),
789 input_value: resolver_spec.input_value.clone(),
790 url_template,
791 strategy: resolver_spec.strategy.clone(),
792 extracts: resolver_spec.extracts.clone(),
793 condition: resolver_spec.condition.clone(),
794 schedule_at: resolver_spec.schedule_at.clone(),
795 state: state_reg,
796 key: key_reg,
797 });
798 }
799
800 ops
801 }
802
803 fn compile_mapping(
804 &self,
805 mapping: &TypedFieldMapping<S>,
806 state_reg: Register,
807 key_reg: Register,
808 ) -> Vec<OpCode> {
809 let mut ops = Vec::new();
810 let temp_reg = 10;
811
812 ops.extend(self.compile_mapping_source(&mapping.source, temp_reg));
813
814 if let Some(transform) = &mapping.transform {
815 ops.push(OpCode::Transform {
816 source: temp_reg,
817 dest: temp_reg,
818 transformation: transform.clone(),
819 });
820 }
821
822 if let Some(stop_instruction) = &mapping.stop {
823 if mapping.when.is_some() {
824 tracing::warn!(
825 "#[map] stop and when both set for {}. Ignoring when.",
826 mapping.target_path
827 );
828 }
829 if !matches!(mapping.population, PopulationStrategy::LastWrite)
830 && !matches!(mapping.population, PopulationStrategy::Merge)
831 {
832 tracing::warn!(
833 "#[map] stop ignores population strategy {:?}",
834 mapping.population
835 );
836 }
837
838 ops.push(OpCode::SetFieldUnlessStopped {
839 object: state_reg,
840 path: mapping.target_path.clone(),
841 value: temp_reg,
842 stop_field: stop_field_path(&mapping.target_path),
843 stop_instruction: stop_instruction.clone(),
844 entity_name: self.entity_name.clone(),
845 key_reg,
846 });
847 return ops;
848 }
849
850 if let Some(when_instruction) = &mapping.when {
851 if !matches!(mapping.population, PopulationStrategy::LastWrite)
852 && !matches!(mapping.population, PopulationStrategy::Merge)
853 {
854 tracing::warn!(
855 "#[map] when ignores population strategy {:?}",
856 mapping.population
857 );
858 }
859 let (condition_field, condition_op, condition_value) = mapping
860 .condition
861 .as_ref()
862 .and_then(|cond| cond.parsed.as_ref())
863 .and_then(|parsed| match parsed {
864 ParsedCondition::Comparison { field, op, value } => {
865 Some((Some(field.clone()), Some(op.clone()), Some(value.clone())))
866 }
867 ParsedCondition::Logical { .. } => {
868 tracing::warn!("Logical conditions not yet supported for #[map] when");
869 None
870 }
871 })
872 .unwrap_or((None, None, None));
873
874 ops.push(OpCode::SetFieldWhen {
875 object: state_reg,
876 path: mapping.target_path.clone(),
877 value: temp_reg,
878 when_instruction: when_instruction.clone(),
879 entity_name: self.entity_name.clone(),
880 key_reg,
881 condition_field,
882 condition_op,
883 condition_value,
884 });
885 return ops;
886 }
887
888 if let Some(condition) = &mapping.condition {
889 if let Some(parsed) = &condition.parsed {
890 match parsed {
891 ParsedCondition::Comparison {
892 field,
893 op,
894 value: cond_value,
895 } => {
896 if matches!(mapping.population, PopulationStrategy::LastWrite)
897 || matches!(mapping.population, PopulationStrategy::Merge)
898 {
899 ops.push(OpCode::ConditionalSetField {
900 object: state_reg,
901 path: mapping.target_path.clone(),
902 value: temp_reg,
903 condition_field: field.clone(),
904 condition_op: op.clone(),
905 condition_value: cond_value.clone(),
906 });
907 return ops;
908 }
909
910 if matches!(mapping.population, PopulationStrategy::Count) {
911 ops.push(OpCode::ConditionalIncrement {
912 object: state_reg,
913 path: mapping.target_path.clone(),
914 condition_field: field.clone(),
915 condition_op: op.clone(),
916 condition_value: cond_value.clone(),
917 });
918 return ops;
919 }
920
921 tracing::warn!(
922 "Conditional #[map] not supported for population strategy {:?}",
923 mapping.population
924 );
925 }
926 ParsedCondition::Logical { .. } => {
927 tracing::warn!("Logical conditions not yet supported for #[map]");
928 }
929 }
930 }
931 }
932
933 match &mapping.population {
934 PopulationStrategy::Append => {
935 ops.push(OpCode::AppendToArray {
936 object: state_reg,
937 path: mapping.target_path.clone(),
938 value: temp_reg,
939 });
940 }
941 PopulationStrategy::LastWrite => {
942 ops.push(OpCode::SetField {
943 object: state_reg,
944 path: mapping.target_path.clone(),
945 value: temp_reg,
946 });
947 }
948 PopulationStrategy::SetOnce => {
949 ops.push(OpCode::SetFieldIfNull {
950 object: state_reg,
951 path: mapping.target_path.clone(),
952 value: temp_reg,
953 });
954 }
955 PopulationStrategy::Merge => {
956 ops.push(OpCode::SetField {
957 object: state_reg,
958 path: mapping.target_path.clone(),
959 value: temp_reg,
960 });
961 }
962 PopulationStrategy::Max => {
963 ops.push(OpCode::SetFieldMax {
964 object: state_reg,
965 path: mapping.target_path.clone(),
966 value: temp_reg,
967 });
968 }
969 PopulationStrategy::Sum => {
970 ops.push(OpCode::SetFieldSum {
971 object: state_reg,
972 path: mapping.target_path.clone(),
973 value: temp_reg,
974 });
975 }
976 PopulationStrategy::Count => {
977 ops.push(OpCode::SetFieldIncrement {
979 object: state_reg,
980 path: mapping.target_path.clone(),
981 });
982 }
983 PopulationStrategy::Min => {
984 ops.push(OpCode::SetFieldMin {
985 object: state_reg,
986 path: mapping.target_path.clone(),
987 value: temp_reg,
988 });
989 }
990 PopulationStrategy::UniqueCount => {
991 let set_name = format!("{}_unique_set", mapping.target_path);
994 ops.push(OpCode::AddToUniqueSet {
995 state_id: self.state_id,
996 set_name,
997 value: temp_reg,
998 count_object: state_reg,
999 count_path: mapping.target_path.clone(),
1000 });
1001 }
1002 }
1003
1004 ops
1005 }
1006
1007 fn compile_mapping_source(&self, source: &MappingSource, dest: Register) -> Vec<OpCode> {
1008 match source {
1009 MappingSource::FromSource {
1010 path,
1011 default,
1012 transform,
1013 } => {
1014 let mut ops = vec![OpCode::LoadEventField {
1015 path: path.clone(),
1016 dest,
1017 default: default.clone(),
1018 }];
1019
1020 if let Some(transform_type) = transform {
1022 ops.push(OpCode::Transform {
1023 source: dest,
1024 dest,
1025 transformation: transform_type.clone(),
1026 });
1027 }
1028
1029 ops
1030 }
1031 MappingSource::Constant(val) => {
1032 vec![OpCode::LoadConstant {
1033 value: val.clone(),
1034 dest,
1035 }]
1036 }
1037 MappingSource::AsEvent { fields } => {
1038 let mut ops = Vec::new();
1039
1040 if fields.is_empty() {
1041 let event_data_reg = dest + 1;
1042 ops.push(OpCode::LoadEventField {
1043 path: FieldPath::new(&[]),
1044 dest: event_data_reg,
1045 default: Some(serde_json::json!({})),
1046 });
1047 ops.push(OpCode::CreateEvent {
1048 dest,
1049 event_value: event_data_reg,
1050 });
1051 } else {
1052 let data_obj_reg = dest + 1;
1053 ops.push(OpCode::CreateObject { dest: data_obj_reg });
1054
1055 let mut field_registers = Vec::new();
1056 let mut current_reg = dest + 2;
1057
1058 for field_source in fields.iter() {
1059 if let MappingSource::FromSource {
1060 path,
1061 default,
1062 transform,
1063 } = &**field_source
1064 {
1065 ops.push(OpCode::LoadEventField {
1066 path: path.clone(),
1067 dest: current_reg,
1068 default: default.clone(),
1069 });
1070
1071 if let Some(transform_type) = transform {
1072 ops.push(OpCode::Transform {
1073 source: current_reg,
1074 dest: current_reg,
1075 transformation: transform_type.clone(),
1076 });
1077 }
1078
1079 if let Some(field_name) = path.segments.last() {
1080 field_registers.push((field_name.clone(), current_reg));
1081 }
1082 current_reg += 1;
1083 }
1084 }
1085
1086 if !field_registers.is_empty() {
1087 ops.push(OpCode::SetFields {
1088 object: data_obj_reg,
1089 fields: field_registers,
1090 });
1091 }
1092
1093 ops.push(OpCode::CreateEvent {
1094 dest,
1095 event_value: data_obj_reg,
1096 });
1097 }
1098
1099 ops
1100 }
1101 MappingSource::WholeSource => {
1102 vec![OpCode::LoadEventField {
1103 path: FieldPath::new(&[]),
1104 dest,
1105 default: Some(serde_json::json!({})),
1106 }]
1107 }
1108 MappingSource::AsCapture { field_transforms } => {
1109 let capture_data_reg = 22; let mut ops = vec![OpCode::LoadEventField {
1112 path: FieldPath::new(&[]),
1113 dest: capture_data_reg,
1114 default: Some(serde_json::json!({})),
1115 }];
1116
1117 let field_reg = 24;
1121 let transformed_reg = 25;
1122
1123 for (field_name, transform) in field_transforms {
1124 ops.push(OpCode::GetField {
1127 object: capture_data_reg,
1128 path: field_name.clone(),
1129 dest: field_reg,
1130 });
1131
1132 ops.push(OpCode::Transform {
1134 source: field_reg,
1135 dest: transformed_reg,
1136 transformation: transform.clone(),
1137 });
1138
1139 ops.push(OpCode::SetField {
1141 object: capture_data_reg,
1142 path: field_name.clone(),
1143 value: transformed_reg,
1144 });
1145 }
1146
1147 ops.push(OpCode::CreateCapture {
1149 dest,
1150 capture_value: capture_data_reg,
1151 });
1152
1153 ops
1154 }
1155 MappingSource::FromContext { field } => {
1156 vec![OpCode::LoadEventField {
1158 path: FieldPath::new(&["__update_context", field.as_str()]),
1159 dest,
1160 default: Some(serde_json::json!(null)),
1161 }]
1162 }
1163 MappingSource::Computed { .. } => {
1164 vec![]
1165 }
1166 MappingSource::FromState { .. } => {
1167 vec![]
1168 }
1169 }
1170 }
1171
1172 pub fn compile_key_loading(
1173 &self,
1174 resolution: &KeyResolutionStrategy,
1175 key_reg: Register,
1176 mappings: &[TypedFieldMapping<S>],
1177 ) -> Vec<OpCode> {
1178 let mut ops = Vec::new();
1179
1180 let resolved_key_reg = 19; ops.push(OpCode::LoadEventField {
1184 path: FieldPath::new(&["__resolved_primary_key"]),
1185 dest: resolved_key_reg,
1186 default: Some(serde_json::json!(null)),
1187 });
1188
1189 match resolution {
1191 KeyResolutionStrategy::Embedded { primary_field } => {
1192 let effective_primary_field = if primary_field.segments.is_empty() {
1194 if let Some(auto_field) = self.auto_detect_primary_field(mappings) {
1196 auto_field
1197 } else {
1198 primary_field.clone()
1199 }
1200 } else {
1201 primary_field.clone()
1202 };
1203
1204 if !effective_primary_field.segments.is_empty() {
1208 let temp_reg = 18;
1209 let transform_reg = 23; ops.push(OpCode::LoadEventField {
1212 path: effective_primary_field.clone(),
1213 dest: temp_reg,
1214 default: None,
1215 });
1216
1217 let primary_key_transform = self
1220 .find_primary_key_transformation(mappings)
1221 .or_else(|| self.find_inherited_primary_key_transformation());
1222
1223 if let Some(transform) = primary_key_transform {
1224 ops.push(OpCode::Transform {
1226 source: temp_reg,
1227 dest: transform_reg,
1228 transformation: transform,
1229 });
1230 ops.push(OpCode::CopyRegister {
1234 source: transform_reg,
1235 dest: key_reg,
1236 });
1237 } else {
1238 ops.push(OpCode::CopyRegister {
1240 source: temp_reg,
1241 dest: key_reg,
1242 });
1243 }
1244 ops.push(OpCode::CopyRegisterIfNull {
1245 source: resolved_key_reg,
1246 dest: key_reg,
1247 });
1248 } else {
1249 ops.push(OpCode::CopyRegister {
1251 source: resolved_key_reg,
1252 dest: key_reg,
1253 });
1254 }
1255 }
1256 KeyResolutionStrategy::Lookup { primary_field } => {
1257 let lookup_reg = 15;
1258 let result_reg = 17;
1259
1260 ops.push(OpCode::CopyRegister {
1266 source: resolved_key_reg,
1267 dest: lookup_reg,
1268 });
1269
1270 let temp_reg = 18;
1271 ops.push(OpCode::LoadEventField {
1272 path: primary_field.clone(),
1273 dest: temp_reg,
1274 default: None,
1275 });
1276 ops.push(OpCode::CopyRegisterIfNull {
1277 source: temp_reg,
1278 dest: lookup_reg,
1279 });
1280
1281 let index_name = self.find_lookup_index_for_lookup_field(primary_field, mappings);
1282 let effective_index_name =
1283 index_name.unwrap_or_else(|| "default_pda_lookup".to_string());
1284
1285 ops.push(OpCode::LookupIndex {
1286 state_id: self.state_id,
1287 index_name: effective_index_name,
1288 lookup_value: lookup_reg,
1289 dest: result_reg,
1290 });
1291 ops.push(OpCode::CopyRegister {
1292 source: result_reg,
1293 dest: key_reg,
1294 });
1295
1296 if self.resolver_outputs_primary_key_directly(primary_field) {
1304 ops.push(OpCode::CopyRegisterIfNull {
1305 source: resolved_key_reg,
1306 dest: key_reg,
1307 });
1308 }
1309 }
1310 KeyResolutionStrategy::Computed {
1311 primary_field,
1312 compute_partition: _,
1313 } => {
1314 ops.push(OpCode::CopyRegister {
1316 source: resolved_key_reg,
1317 dest: key_reg,
1318 });
1319 let temp_reg = 18;
1320 ops.push(OpCode::LoadEventField {
1321 path: primary_field.clone(),
1322 dest: temp_reg,
1323 default: None,
1324 });
1325 ops.push(OpCode::CopyRegisterIfNull {
1326 source: temp_reg,
1327 dest: key_reg,
1328 });
1329 }
1330 KeyResolutionStrategy::TemporalLookup {
1331 lookup_field,
1332 timestamp_field,
1333 index_name,
1334 } => {
1335 ops.push(OpCode::CopyRegister {
1337 source: resolved_key_reg,
1338 dest: key_reg,
1339 });
1340 let lookup_reg = 15;
1341 let timestamp_reg = 16;
1342 let result_reg = 17;
1343
1344 ops.push(OpCode::LoadEventField {
1345 path: lookup_field.clone(),
1346 dest: lookup_reg,
1347 default: None,
1348 });
1349
1350 ops.push(OpCode::LoadEventField {
1351 path: timestamp_field.clone(),
1352 dest: timestamp_reg,
1353 default: None,
1354 });
1355
1356 ops.push(OpCode::LookupTemporalIndex {
1357 state_id: self.state_id,
1358 index_name: index_name.clone(),
1359 lookup_value: lookup_reg,
1360 timestamp: timestamp_reg,
1361 dest: result_reg,
1362 });
1363
1364 ops.push(OpCode::CopyRegisterIfNull {
1365 source: result_reg,
1366 dest: key_reg,
1367 });
1368 }
1369 }
1370
1371 ops
1372 }
1373
1374 fn find_primary_key_transformation(
1375 &self,
1376 mappings: &[TypedFieldMapping<S>],
1377 ) -> Option<Transformation> {
1378 let primary_key = self.spec.identity.primary_keys.first()?;
1380 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1381
1382 for mapping in mappings {
1384 if mapping.target_path == *primary_key
1386 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1387 {
1388 if let Some(transform) = &mapping.transform {
1390 return Some(transform.clone());
1391 }
1392
1393 if let MappingSource::FromSource {
1395 transform: Some(transform),
1396 ..
1397 } = &mapping.source
1398 {
1399 return Some(transform.clone());
1400 }
1401 }
1402 }
1403
1404 for mapping in mappings {
1406 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1407 if let Some(transform) = field_transforms.get(&primary_field_name) {
1408 return Some(transform.clone());
1409 }
1410 }
1411 }
1412
1413 None
1414 }
1415
1416 pub fn find_inherited_primary_key_transformation(&self) -> Option<Transformation> {
1419 let primary_key = self.spec.identity.primary_keys.first()?;
1420
1421 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1423
1424 for handler in &self.spec.handlers {
1426 for mapping in &handler.mappings {
1427 if mapping.target_path == *primary_key
1429 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1430 {
1431 if let MappingSource::FromSource {
1433 path, transform, ..
1434 } = &mapping.source
1435 {
1436 if path.segments.last() == Some(&primary_field_name) {
1437 return mapping.transform.clone().or_else(|| transform.clone());
1439 }
1440 }
1441 }
1442
1443 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1445 if let Some(transform) = field_transforms.get(&primary_field_name) {
1446 return Some(transform.clone());
1447 }
1448 }
1449 }
1450 }
1451
1452 None
1453 }
1454
1455 fn extract_primary_field_name(&self, primary_key: &str) -> Option<String> {
1457 primary_key.split('.').next_back().map(|s| s.to_string())
1459 }
1460
1461 pub fn auto_detect_primary_field(
1464 &self,
1465 current_mappings: &[TypedFieldMapping<S>],
1466 ) -> Option<FieldPath> {
1467 let primary_key = self.spec.identity.primary_keys.first()?;
1468
1469 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1471
1472 if self.current_account_has_primary_field(&primary_field_name, current_mappings) {
1474 return Some(FieldPath::new(&[&primary_field_name]));
1475 }
1476
1477 None
1478 }
1479
1480 fn current_account_has_primary_field(
1483 &self,
1484 field_name: &str,
1485 mappings: &[TypedFieldMapping<S>],
1486 ) -> bool {
1487 for mapping in mappings {
1489 if let MappingSource::FromSource { path, .. } = &mapping.source {
1490 if path.segments.last() == Some(&field_name.to_string()) {
1492 return true;
1493 }
1494 }
1495 }
1496
1497 false
1498 }
1499
1500 #[allow(dead_code)]
1502 fn handler_has_field(&self, field_name: &str, mappings: &[TypedFieldMapping<S>]) -> bool {
1503 for mapping in mappings {
1504 if let MappingSource::FromSource { path, .. } = &mapping.source {
1505 if path.segments.last() == Some(&field_name.to_string()) {
1506 return true;
1507 }
1508 }
1509 }
1510 false
1511 }
1512
1513 #[allow(dead_code)]
1516 fn field_exists_in_mappings(
1517 &self,
1518 field_name: &str,
1519 mappings: &[TypedFieldMapping<S>],
1520 ) -> bool {
1521 for mapping in mappings {
1523 if let MappingSource::FromSource { path, .. } = &mapping.source {
1524 if path.segments.last() == Some(&field_name.to_string()) {
1525 return true;
1526 }
1527 }
1528 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1530 if field_transforms.contains_key(field_name) {
1531 return true;
1532 }
1533 }
1534 }
1535 false
1536 }
1537
1538 fn find_lookup_index_for_field(&self, field_path: &FieldPath) -> Option<String> {
1539 if field_path.segments.is_empty() {
1540 return None;
1541 }
1542
1543 let lookup_field_name = field_path.segments.last().unwrap();
1544
1545 for lookup_index in &self.spec.identity.lookup_indexes {
1546 let index_field_name = lookup_index
1547 .field_name
1548 .split('.')
1549 .next_back()
1550 .unwrap_or(&lookup_index.field_name);
1551 let matches_directly = index_field_name == lookup_field_name;
1552 let matches_address_alias = index_field_name
1559 .strip_suffix("_address")
1560 .map(|base| base == lookup_field_name)
1561 .unwrap_or(false);
1562
1563 if matches_directly || matches_address_alias {
1564 return Some(format!("{}_lookup_index", index_field_name));
1565 }
1566 }
1567
1568 None
1569 }
1570
1571 fn find_lookup_index_for_lookup_field(
1574 &self,
1575 primary_field: &FieldPath,
1576 mappings: &[TypedFieldMapping<S>],
1577 ) -> Option<String> {
1578 let primary_path = primary_field.segments.join(".");
1580
1581 for mapping in mappings {
1583 if let MappingSource::FromSource { path, .. } = &mapping.source {
1585 let source_path = path.segments.join(".");
1586 if source_path == primary_path {
1587 for lookup_index in &self.spec.identity.lookup_indexes {
1589 if mapping.target_path == lookup_index.field_name {
1590 let index_field_name = lookup_index
1591 .field_name
1592 .split('.')
1593 .next_back()
1594 .unwrap_or(&lookup_index.field_name);
1595 return Some(format!("{}_lookup_index", index_field_name));
1596 }
1597 }
1598 }
1599 }
1600 }
1601
1602 self.find_lookup_index_for_field(primary_field)
1604 }
1605
1606 fn find_source_path_for_lookup_index(
1610 &self,
1611 mappings: &[TypedFieldMapping<S>],
1612 lookup_field_name: &str,
1613 ) -> Option<Vec<String>> {
1614 for mapping in mappings {
1615 if mapping.target_path == lookup_field_name {
1616 if let MappingSource::FromSource { path, .. } = &mapping.source {
1617 return Some(path.segments.clone());
1618 }
1619 }
1620 }
1621 None
1622 }
1623
1624 fn compile_temporal_index_update(
1625 &self,
1626 resolution: &KeyResolutionStrategy,
1627 key_reg: Register,
1628 mappings: &[TypedFieldMapping<S>],
1629 ) -> Vec<OpCode> {
1630 let mut ops = Vec::new();
1631
1632 for lookup_index in &self.spec.identity.lookup_indexes {
1633 let lookup_reg = 17;
1634 let source_field = lookup_index
1635 .field_name
1636 .split('.')
1637 .next_back()
1638 .unwrap_or(&lookup_index.field_name);
1639
1640 match resolution {
1641 KeyResolutionStrategy::Embedded { primary_field: _ } => {
1642 let source_path_opt =
1645 self.find_source_path_for_lookup_index(mappings, &lookup_index.field_name);
1646
1647 let load_path = if let Some(ref path) = source_path_opt {
1648 FieldPath::new(&path.iter().map(|s| s.as_str()).collect::<Vec<_>>())
1649 } else {
1650 FieldPath::new(&[source_field])
1652 };
1653
1654 ops.push(OpCode::LoadEventField {
1655 path: load_path,
1656 dest: lookup_reg,
1657 default: None,
1658 });
1659
1660 if let Some(temporal_field_name) = &lookup_index.temporal_field {
1661 let timestamp_reg = 18;
1662
1663 ops.push(OpCode::LoadEventField {
1664 path: FieldPath::new(&[temporal_field_name]),
1665 dest: timestamp_reg,
1666 default: None,
1667 });
1668
1669 let index_name = format!("{}_temporal_index", source_field);
1670 ops.push(OpCode::UpdateTemporalIndex {
1671 state_id: self.state_id,
1672 index_name,
1673 lookup_value: lookup_reg,
1674 primary_key: key_reg,
1675 timestamp: timestamp_reg,
1676 });
1677
1678 let simple_index_name = format!("{}_lookup_index", source_field);
1679 ops.push(OpCode::UpdateLookupIndex {
1680 state_id: self.state_id,
1681 index_name: simple_index_name,
1682 lookup_value: lookup_reg,
1683 primary_key: key_reg,
1684 });
1685 } else {
1686 let index_name = format!("{}_lookup_index", source_field);
1687 ops.push(OpCode::UpdateLookupIndex {
1688 state_id: self.state_id,
1689 index_name,
1690 lookup_value: lookup_reg,
1691 primary_key: key_reg,
1692 });
1693 }
1694
1695 if source_path_opt.is_some() {
1699 ops.push(OpCode::UpdatePdaReverseLookup {
1700 state_id: self.state_id,
1701 lookup_name: "default_pda_lookup".to_string(),
1702 pda_address: lookup_reg,
1703 primary_key: key_reg,
1704 });
1705 }
1706 }
1707 KeyResolutionStrategy::Lookup { primary_field } => {
1708 let has_mapping_to_lookup_field = mappings
1711 .iter()
1712 .any(|m| m.target_path == lookup_index.field_name);
1713
1714 if has_mapping_to_lookup_field {
1715 let path_segments: Vec<&str> =
1718 primary_field.segments.iter().map(|s| s.as_str()).collect();
1719 ops.push(OpCode::LoadEventField {
1720 path: FieldPath::new(&path_segments),
1721 dest: lookup_reg,
1722 default: None,
1723 });
1724
1725 let index_name = format!("{}_lookup_index", source_field);
1726 ops.push(OpCode::UpdateLookupIndex {
1727 state_id: self.state_id,
1728 index_name,
1729 lookup_value: lookup_reg,
1730 primary_key: key_reg,
1731 });
1732 }
1733 }
1734 KeyResolutionStrategy::Computed { .. }
1735 | KeyResolutionStrategy::TemporalLookup { .. } => {
1736 }
1738 }
1739 }
1740
1741 ops
1742 }
1743
1744 fn get_event_type(&self, source: &SourceSpec) -> String {
1745 match source {
1746 SourceSpec::Source { type_name, .. } => type_name.clone(),
1747 }
1748 }
1749
1750 fn compile_instruction_hook_actions(&self, actions: &[HookAction]) -> Vec<OpCode> {
1751 let mut ops = Vec::new();
1752 let state_reg = 2;
1753
1754 for action in actions {
1755 match action {
1756 HookAction::SetField {
1757 target_field,
1758 source,
1759 condition,
1760 } => {
1761 let _ = condition;
1763
1764 let temp_reg = 11; let load_ops = self.compile_mapping_source(source, temp_reg);
1768 ops.extend(load_ops);
1769
1770 if let MappingSource::FromSource {
1772 transform: Some(transform_type),
1773 ..
1774 } = source
1775 {
1776 ops.push(OpCode::Transform {
1777 source: temp_reg,
1778 dest: temp_reg,
1779 transformation: transform_type.clone(),
1780 });
1781 }
1782
1783 if let Some(cond_expr) = condition {
1785 if let Some(parsed) = &cond_expr.parsed {
1786 let cond_check_ops = self.compile_condition_check(
1788 parsed,
1789 temp_reg,
1790 state_reg,
1791 target_field,
1792 );
1793 ops.extend(cond_check_ops);
1794 } else {
1795 ops.push(OpCode::SetField {
1797 object: state_reg,
1798 path: target_field.clone(),
1799 value: temp_reg,
1800 });
1801 }
1802 } else {
1803 ops.push(OpCode::SetField {
1805 object: state_reg,
1806 path: target_field.clone(),
1807 value: temp_reg,
1808 });
1809 }
1810 }
1811 HookAction::IncrementField {
1812 target_field,
1813 increment_by,
1814 condition,
1815 } => {
1816 if let Some(cond_expr) = condition {
1817 if let Some(parsed) = &cond_expr.parsed {
1818 let cond_check_ops = self.compile_conditional_increment(
1823 parsed,
1824 state_reg,
1825 target_field,
1826 *increment_by,
1827 );
1828 ops.extend(cond_check_ops);
1829 } else {
1830 ops.push(OpCode::SetFieldIncrement {
1832 object: state_reg,
1833 path: target_field.clone(),
1834 });
1835 }
1836 } else {
1837 ops.push(OpCode::SetFieldIncrement {
1839 object: state_reg,
1840 path: target_field.clone(),
1841 });
1842 }
1843 }
1844 HookAction::RegisterPdaMapping { .. } => {
1845 if let HookAction::RegisterPdaMapping {
1846 pda_field,
1847 seed_field,
1848 lookup_name,
1849 } = action
1850 {
1851 let pda_reg = 11;
1852 let seed_reg = 12;
1853
1854 ops.push(OpCode::LoadEventField {
1855 path: pda_field.clone(),
1856 dest: pda_reg,
1857 default: None,
1858 });
1859 ops.push(OpCode::LoadEventField {
1860 path: seed_field.clone(),
1861 dest: seed_reg,
1862 default: None,
1863 });
1864 ops.push(OpCode::UpdatePdaReverseLookup {
1865 state_id: self.state_id,
1866 lookup_name: lookup_name.clone(),
1867 pda_address: pda_reg,
1868 primary_key: seed_reg,
1869 });
1870 }
1871 }
1872 }
1873 }
1874
1875 ops
1876 }
1877
1878 fn compile_condition_check(
1879 &self,
1880 condition: &ParsedCondition,
1881 value_reg: Register,
1882 state_reg: Register,
1883 target_field: &str,
1884 ) -> Vec<OpCode> {
1885 match condition {
1886 ParsedCondition::Comparison {
1887 field,
1888 op,
1889 value: cond_value,
1890 } => {
1891 vec![OpCode::ConditionalSetField {
1893 object: state_reg,
1894 path: target_field.to_string(),
1895 value: value_reg,
1896 condition_field: field.clone(),
1897 condition_op: op.clone(),
1898 condition_value: cond_value.clone(),
1899 }]
1900 }
1901 ParsedCondition::Logical { .. } => {
1902 tracing::warn!("Logical conditions not yet supported in instruction hooks");
1904 vec![OpCode::SetField {
1905 object: state_reg,
1906 path: target_field.to_string(),
1907 value: value_reg,
1908 }]
1909 }
1910 }
1911 }
1912
1913 fn compile_conditional_increment(
1914 &self,
1915 condition: &ParsedCondition,
1916 state_reg: Register,
1917 target_field: &str,
1918 _increment_by: i64,
1919 ) -> Vec<OpCode> {
1920 match condition {
1921 ParsedCondition::Comparison {
1922 field,
1923 op,
1924 value: cond_value,
1925 } => {
1926 vec![OpCode::ConditionalIncrement {
1927 object: state_reg,
1928 path: target_field.to_string(),
1929 condition_field: field.clone(),
1930 condition_op: op.clone(),
1931 condition_value: cond_value.clone(),
1932 }]
1933 }
1934 ParsedCondition::Logical { .. } => {
1935 tracing::warn!("Logical conditions not yet supported in instruction hooks");
1936 vec![OpCode::SetFieldIncrement {
1937 object: state_reg,
1938 path: target_field.to_string(),
1939 }]
1940 }
1941 }
1942 }
1943}
1944
1945#[cfg(test)]
1946mod tests {
1947 use super::MultiEntityBytecode;
1948 use crate::ast::{
1949 FieldPath, HookAction, IdentitySpec, InstructionHook, KeyResolutionStrategy,
1950 LookupIndexSpec, MappingSource, PopulationStrategy, SerializableFieldMapping,
1951 SerializableHandlerSpec, SerializableStreamSpec, SourceSpec, TypedStreamSpec,
1952 };
1953 use crate::vm::VmContext;
1954 use serde_json::{json, Value};
1955 use std::collections::BTreeMap;
1956
1957 fn mapping(
1958 target_path: &str,
1959 source_path: &[&str],
1960 population: PopulationStrategy,
1961 ) -> SerializableFieldMapping {
1962 SerializableFieldMapping {
1963 target_path: target_path.to_string(),
1964 source: MappingSource::FromSource {
1965 path: FieldPath::new(source_path),
1966 default: None,
1967 transform: None,
1968 },
1969 transform: None,
1970 population,
1971 condition: None,
1972 when: None,
1973 stop: None,
1974 emit: true,
1975 }
1976 }
1977
1978 fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
1979 TypedStreamSpec::from_serializable(SerializableStreamSpec {
1980 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
1981 state_name: "PumpfunToken".to_string(),
1982 program_id: None,
1983 idl: None,
1984 identity: IdentitySpec {
1985 primary_keys: vec!["id.mint".to_string()],
1986 lookup_indexes: vec![LookupIndexSpec {
1987 field_name: "id.bonding_curve".to_string(),
1988 temporal_field: None,
1989 }],
1990 },
1991 handlers: vec![SerializableHandlerSpec {
1992 source: SourceSpec::Source {
1993 program_id: None,
1994 discriminator: None,
1995 type_name: "pump::BondingCurveState".to_string(),
1996 serialization: None,
1997 is_account: true,
1998 },
1999 key_resolution: KeyResolutionStrategy::Lookup {
2000 primary_field: FieldPath::new(&["__account_address"]),
2001 },
2002 mappings: vec![
2003 mapping(
2004 "id.bonding_curve",
2005 &["__account_address"],
2006 PopulationStrategy::SetOnce,
2007 ),
2008 mapping(
2009 "reserves.virtual_token_reserves",
2010 &["virtual_token_reserves"],
2011 PopulationStrategy::LastWrite,
2012 ),
2013 ],
2014 conditions: vec![],
2015 emit: true,
2016 }],
2017 sections: vec![],
2018 field_mappings: BTreeMap::new(),
2019 resolver_hooks: vec![],
2020 instruction_hooks: vec![InstructionHook {
2021 instruction_type: "pump::BuyIxState".to_string(),
2022 actions: vec![HookAction::RegisterPdaMapping {
2023 pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2024 seed_field: FieldPath::new(&["accounts", "mint"]),
2025 lookup_name: "default_pda_lookup".to_string(),
2026 }],
2027 lookup_by: None,
2028 }],
2029 resolver_specs: vec![],
2030 computed_fields: vec![],
2031 computed_field_specs: vec![],
2032 content_hash: None,
2033 views: vec![],
2034 })
2035 }
2036
2037 fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
2038 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2039 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2040 state_name: "OreRound".to_string(),
2041 program_id: None,
2042 idl: None,
2043 identity: IdentitySpec {
2044 primary_keys: vec!["id.round_id".to_string()],
2045 lookup_indexes: vec![LookupIndexSpec {
2046 field_name: "id.round_address".to_string(),
2047 temporal_field: None,
2048 }],
2049 },
2050 handlers: vec![SerializableHandlerSpec {
2051 source: SourceSpec::Source {
2052 program_id: None,
2053 discriminator: None,
2054 type_name: "entropy::VarState".to_string(),
2055 serialization: None,
2056 is_account: true,
2057 },
2058 key_resolution: KeyResolutionStrategy::Lookup {
2059 primary_field: FieldPath::new(&["__account_address"]),
2060 },
2061 mappings: vec![mapping(
2062 "state.expires_at",
2063 &["end_at"],
2064 PopulationStrategy::LastWrite,
2065 )],
2066 conditions: vec![],
2067 emit: true,
2068 }],
2069 sections: vec![],
2070 field_mappings: BTreeMap::new(),
2071 resolver_hooks: vec![],
2072 instruction_hooks: vec![InstructionHook {
2073 instruction_type: "ore::DeployIxState".to_string(),
2074 actions: vec![HookAction::RegisterPdaMapping {
2075 pda_field: FieldPath::new(&["accounts", "entropyVar"]),
2076 seed_field: FieldPath::new(&["accounts", "round"]),
2077 lookup_name: "default_pda_lookup".to_string(),
2078 }],
2079 lookup_by: None,
2080 }],
2081 resolver_specs: vec![],
2082 computed_fields: vec![],
2083 computed_field_specs: vec![],
2084 content_hash: None,
2085 views: vec![],
2086 })
2087 }
2088
2089 fn embedded_account_key_spec() -> TypedStreamSpec<Value> {
2090 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2091 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2092 state_name: "OreTreasury".to_string(),
2093 program_id: None,
2094 idl: None,
2095 identity: IdentitySpec {
2096 primary_keys: vec!["id.address".to_string()],
2097 lookup_indexes: vec![],
2098 },
2099 handlers: vec![SerializableHandlerSpec {
2100 source: SourceSpec::Source {
2101 program_id: None,
2102 discriminator: None,
2103 type_name: "ore::TreasuryState".to_string(),
2104 serialization: None,
2105 is_account: true,
2106 },
2107 key_resolution: KeyResolutionStrategy::Embedded {
2108 primary_field: FieldPath::new(&["__account_address"]),
2109 },
2110 mappings: vec![mapping(
2111 "id.address",
2112 &["__account_address"],
2113 PopulationStrategy::SetOnce,
2114 )],
2115 conditions: vec![],
2116 emit: true,
2117 }],
2118 sections: vec![],
2119 field_mappings: BTreeMap::new(),
2120 resolver_hooks: vec![],
2121 instruction_hooks: vec![],
2122 resolver_specs: vec![],
2123 computed_fields: vec![],
2124 computed_field_specs: vec![],
2125 content_hash: None,
2126 views: vec![],
2127 })
2128 }
2129
2130 #[test]
2131 fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
2132 let bytecode = MultiEntityBytecode::from_single(
2133 "PumpfunToken".to_string(),
2134 direct_pda_to_primary_key_spec(),
2135 0,
2136 );
2137 let mut vm = VmContext::new();
2138
2139 let mutations = vm
2140 .process_event(
2141 &bytecode,
2142 json!({
2143 "__account_address": "bonding_curve_1",
2144 "__resolved_primary_key": "mint_1",
2145 "virtual_token_reserves": 42,
2146 }),
2147 "pump::BondingCurveState",
2148 None,
2149 None,
2150 )
2151 .unwrap();
2152
2153 assert_eq!(mutations.len(), 1);
2154 assert_eq!(mutations[0].key, json!("mint_1"));
2155 assert_eq!(
2156 mutations[0].patch["id"]["bonding_curve"],
2157 json!("bonding_curve_1")
2158 );
2159 assert_eq!(
2160 mutations[0].patch["reserves"]["virtual_token_reserves"],
2161 json!(42)
2162 );
2163 }
2164
2165 #[test]
2166 fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
2167 let bytecode = MultiEntityBytecode::from_single(
2168 "OreRound".to_string(),
2169 pda_to_intermediate_lookup_spec(),
2170 0,
2171 );
2172 let mut vm = VmContext::new();
2173
2174 let mutations = vm
2175 .process_event(
2176 &bytecode,
2177 json!({
2178 "__account_address": "entropy_var_1",
2179 "__resolved_primary_key": "round_address_1",
2180 "end_at": 123,
2181 }),
2182 "entropy::VarState",
2183 None,
2184 None,
2185 )
2186 .unwrap();
2187
2188 assert!(mutations.is_empty());
2189 }
2190
2191 #[test]
2192 fn embedded_account_key_wins_over_an_unrelated_resolved_key() {
2193 let bytecode = MultiEntityBytecode::from_single(
2194 "OreTreasury".to_string(),
2195 embedded_account_key_spec(),
2196 0,
2197 );
2198 let mut vm = VmContext::new();
2199
2200 let mutations = vm
2201 .process_event(
2202 &bytecode,
2203 json!({
2204 "__account_address": "treasury_pda",
2205 "__resolved_primary_key": "round_address",
2206 }),
2207 "ore::TreasuryState",
2208 None,
2209 None,
2210 )
2211 .unwrap();
2212
2213 assert_eq!(mutations.len(), 1);
2214 assert_eq!(mutations[0].key, json!("treasury_pda"));
2215 assert_eq!(mutations[0].patch["id"]["address"], json!("treasury_pda"));
2216 }
2217}