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 ops.push(OpCode::CopyRegister {
1194 source: resolved_key_reg,
1195 dest: key_reg,
1196 });
1197
1198 let effective_primary_field = if primary_field.segments.is_empty() {
1200 if let Some(auto_field) = self.auto_detect_primary_field(mappings) {
1202 auto_field
1203 } else {
1204 primary_field.clone()
1205 }
1206 } else {
1207 primary_field.clone()
1208 };
1209
1210 if !effective_primary_field.segments.is_empty() {
1214 let temp_reg = 18;
1215 let transform_reg = 23; ops.push(OpCode::LoadEventField {
1218 path: effective_primary_field.clone(),
1219 dest: temp_reg,
1220 default: None,
1221 });
1222
1223 let primary_key_transform = self
1226 .find_primary_key_transformation(mappings)
1227 .or_else(|| self.find_inherited_primary_key_transformation());
1228
1229 if let Some(transform) = primary_key_transform {
1230 ops.push(OpCode::Transform {
1232 source: temp_reg,
1233 dest: transform_reg,
1234 transformation: transform,
1235 });
1236 ops.push(OpCode::CopyRegisterIfNull {
1238 source: transform_reg,
1239 dest: key_reg,
1240 });
1241 } else {
1242 ops.push(OpCode::CopyRegisterIfNull {
1244 source: temp_reg,
1245 dest: key_reg,
1246 });
1247 }
1248 }
1249 }
1252 KeyResolutionStrategy::Lookup { primary_field } => {
1253 let lookup_reg = 15;
1254 let result_reg = 17;
1255
1256 ops.push(OpCode::CopyRegister {
1262 source: resolved_key_reg,
1263 dest: lookup_reg,
1264 });
1265
1266 let temp_reg = 18;
1267 ops.push(OpCode::LoadEventField {
1268 path: primary_field.clone(),
1269 dest: temp_reg,
1270 default: None,
1271 });
1272 ops.push(OpCode::CopyRegisterIfNull {
1273 source: temp_reg,
1274 dest: lookup_reg,
1275 });
1276
1277 let index_name = self.find_lookup_index_for_lookup_field(primary_field, mappings);
1278 let effective_index_name =
1279 index_name.unwrap_or_else(|| "default_pda_lookup".to_string());
1280
1281 ops.push(OpCode::LookupIndex {
1282 state_id: self.state_id,
1283 index_name: effective_index_name,
1284 lookup_value: lookup_reg,
1285 dest: result_reg,
1286 });
1287 ops.push(OpCode::CopyRegister {
1288 source: result_reg,
1289 dest: key_reg,
1290 });
1291
1292 if self.resolver_outputs_primary_key_directly(primary_field) {
1300 ops.push(OpCode::CopyRegisterIfNull {
1301 source: resolved_key_reg,
1302 dest: key_reg,
1303 });
1304 }
1305 }
1306 KeyResolutionStrategy::Computed {
1307 primary_field,
1308 compute_partition: _,
1309 } => {
1310 ops.push(OpCode::CopyRegister {
1312 source: resolved_key_reg,
1313 dest: key_reg,
1314 });
1315 let temp_reg = 18;
1316 ops.push(OpCode::LoadEventField {
1317 path: primary_field.clone(),
1318 dest: temp_reg,
1319 default: None,
1320 });
1321 ops.push(OpCode::CopyRegisterIfNull {
1322 source: temp_reg,
1323 dest: key_reg,
1324 });
1325 }
1326 KeyResolutionStrategy::TemporalLookup {
1327 lookup_field,
1328 timestamp_field,
1329 index_name,
1330 } => {
1331 ops.push(OpCode::CopyRegister {
1333 source: resolved_key_reg,
1334 dest: key_reg,
1335 });
1336 let lookup_reg = 15;
1337 let timestamp_reg = 16;
1338 let result_reg = 17;
1339
1340 ops.push(OpCode::LoadEventField {
1341 path: lookup_field.clone(),
1342 dest: lookup_reg,
1343 default: None,
1344 });
1345
1346 ops.push(OpCode::LoadEventField {
1347 path: timestamp_field.clone(),
1348 dest: timestamp_reg,
1349 default: None,
1350 });
1351
1352 ops.push(OpCode::LookupTemporalIndex {
1353 state_id: self.state_id,
1354 index_name: index_name.clone(),
1355 lookup_value: lookup_reg,
1356 timestamp: timestamp_reg,
1357 dest: result_reg,
1358 });
1359
1360 ops.push(OpCode::CopyRegisterIfNull {
1361 source: result_reg,
1362 dest: key_reg,
1363 });
1364 }
1365 }
1366
1367 ops
1368 }
1369
1370 fn find_primary_key_transformation(
1371 &self,
1372 mappings: &[TypedFieldMapping<S>],
1373 ) -> Option<Transformation> {
1374 let primary_key = self.spec.identity.primary_keys.first()?;
1376 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1377
1378 for mapping in mappings {
1380 if mapping.target_path == *primary_key
1382 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1383 {
1384 if let Some(transform) = &mapping.transform {
1386 return Some(transform.clone());
1387 }
1388
1389 if let MappingSource::FromSource {
1391 transform: Some(transform),
1392 ..
1393 } = &mapping.source
1394 {
1395 return Some(transform.clone());
1396 }
1397 }
1398 }
1399
1400 for mapping in mappings {
1402 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1403 if let Some(transform) = field_transforms.get(&primary_field_name) {
1404 return Some(transform.clone());
1405 }
1406 }
1407 }
1408
1409 None
1410 }
1411
1412 pub fn find_inherited_primary_key_transformation(&self) -> Option<Transformation> {
1415 let primary_key = self.spec.identity.primary_keys.first()?;
1416
1417 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1419
1420 for handler in &self.spec.handlers {
1422 for mapping in &handler.mappings {
1423 if mapping.target_path == *primary_key
1425 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1426 {
1427 if let MappingSource::FromSource {
1429 path, transform, ..
1430 } = &mapping.source
1431 {
1432 if path.segments.last() == Some(&primary_field_name) {
1433 return mapping.transform.clone().or_else(|| transform.clone());
1435 }
1436 }
1437 }
1438
1439 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1441 if let Some(transform) = field_transforms.get(&primary_field_name) {
1442 return Some(transform.clone());
1443 }
1444 }
1445 }
1446 }
1447
1448 None
1449 }
1450
1451 fn extract_primary_field_name(&self, primary_key: &str) -> Option<String> {
1453 primary_key.split('.').next_back().map(|s| s.to_string())
1455 }
1456
1457 pub fn auto_detect_primary_field(
1460 &self,
1461 current_mappings: &[TypedFieldMapping<S>],
1462 ) -> Option<FieldPath> {
1463 let primary_key = self.spec.identity.primary_keys.first()?;
1464
1465 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1467
1468 if self.current_account_has_primary_field(&primary_field_name, current_mappings) {
1470 return Some(FieldPath::new(&[&primary_field_name]));
1471 }
1472
1473 None
1474 }
1475
1476 fn current_account_has_primary_field(
1479 &self,
1480 field_name: &str,
1481 mappings: &[TypedFieldMapping<S>],
1482 ) -> bool {
1483 for mapping in mappings {
1485 if let MappingSource::FromSource { path, .. } = &mapping.source {
1486 if path.segments.last() == Some(&field_name.to_string()) {
1488 return true;
1489 }
1490 }
1491 }
1492
1493 false
1494 }
1495
1496 #[allow(dead_code)]
1498 fn handler_has_field(&self, field_name: &str, mappings: &[TypedFieldMapping<S>]) -> bool {
1499 for mapping in mappings {
1500 if let MappingSource::FromSource { path, .. } = &mapping.source {
1501 if path.segments.last() == Some(&field_name.to_string()) {
1502 return true;
1503 }
1504 }
1505 }
1506 false
1507 }
1508
1509 #[allow(dead_code)]
1512 fn field_exists_in_mappings(
1513 &self,
1514 field_name: &str,
1515 mappings: &[TypedFieldMapping<S>],
1516 ) -> bool {
1517 for mapping in mappings {
1519 if let MappingSource::FromSource { path, .. } = &mapping.source {
1520 if path.segments.last() == Some(&field_name.to_string()) {
1521 return true;
1522 }
1523 }
1524 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1526 if field_transforms.contains_key(field_name) {
1527 return true;
1528 }
1529 }
1530 }
1531 false
1532 }
1533
1534 fn find_lookup_index_for_field(&self, field_path: &FieldPath) -> Option<String> {
1535 if field_path.segments.is_empty() {
1536 return None;
1537 }
1538
1539 let lookup_field_name = field_path.segments.last().unwrap();
1540
1541 for lookup_index in &self.spec.identity.lookup_indexes {
1542 let index_field_name = lookup_index
1543 .field_name
1544 .split('.')
1545 .next_back()
1546 .unwrap_or(&lookup_index.field_name);
1547 let matches_directly = index_field_name == lookup_field_name;
1548 let matches_address_alias = index_field_name
1555 .strip_suffix("_address")
1556 .map(|base| base == lookup_field_name)
1557 .unwrap_or(false);
1558
1559 if matches_directly || matches_address_alias {
1560 return Some(format!("{}_lookup_index", index_field_name));
1561 }
1562 }
1563
1564 None
1565 }
1566
1567 fn find_lookup_index_for_lookup_field(
1570 &self,
1571 primary_field: &FieldPath,
1572 mappings: &[TypedFieldMapping<S>],
1573 ) -> Option<String> {
1574 let primary_path = primary_field.segments.join(".");
1576
1577 for mapping in mappings {
1579 if let MappingSource::FromSource { path, .. } = &mapping.source {
1581 let source_path = path.segments.join(".");
1582 if source_path == primary_path {
1583 for lookup_index in &self.spec.identity.lookup_indexes {
1585 if mapping.target_path == lookup_index.field_name {
1586 let index_field_name = lookup_index
1587 .field_name
1588 .split('.')
1589 .next_back()
1590 .unwrap_or(&lookup_index.field_name);
1591 return Some(format!("{}_lookup_index", index_field_name));
1592 }
1593 }
1594 }
1595 }
1596 }
1597
1598 self.find_lookup_index_for_field(primary_field)
1600 }
1601
1602 fn find_source_path_for_lookup_index(
1606 &self,
1607 mappings: &[TypedFieldMapping<S>],
1608 lookup_field_name: &str,
1609 ) -> Option<Vec<String>> {
1610 for mapping in mappings {
1611 if mapping.target_path == lookup_field_name {
1612 if let MappingSource::FromSource { path, .. } = &mapping.source {
1613 return Some(path.segments.clone());
1614 }
1615 }
1616 }
1617 None
1618 }
1619
1620 fn compile_temporal_index_update(
1621 &self,
1622 resolution: &KeyResolutionStrategy,
1623 key_reg: Register,
1624 mappings: &[TypedFieldMapping<S>],
1625 ) -> Vec<OpCode> {
1626 let mut ops = Vec::new();
1627
1628 for lookup_index in &self.spec.identity.lookup_indexes {
1629 let lookup_reg = 17;
1630 let source_field = lookup_index
1631 .field_name
1632 .split('.')
1633 .next_back()
1634 .unwrap_or(&lookup_index.field_name);
1635
1636 match resolution {
1637 KeyResolutionStrategy::Embedded { primary_field: _ } => {
1638 let source_path_opt =
1641 self.find_source_path_for_lookup_index(mappings, &lookup_index.field_name);
1642
1643 let load_path = if let Some(ref path) = source_path_opt {
1644 FieldPath::new(&path.iter().map(|s| s.as_str()).collect::<Vec<_>>())
1645 } else {
1646 FieldPath::new(&[source_field])
1648 };
1649
1650 ops.push(OpCode::LoadEventField {
1651 path: load_path,
1652 dest: lookup_reg,
1653 default: None,
1654 });
1655
1656 if let Some(temporal_field_name) = &lookup_index.temporal_field {
1657 let timestamp_reg = 18;
1658
1659 ops.push(OpCode::LoadEventField {
1660 path: FieldPath::new(&[temporal_field_name]),
1661 dest: timestamp_reg,
1662 default: None,
1663 });
1664
1665 let index_name = format!("{}_temporal_index", source_field);
1666 ops.push(OpCode::UpdateTemporalIndex {
1667 state_id: self.state_id,
1668 index_name,
1669 lookup_value: lookup_reg,
1670 primary_key: key_reg,
1671 timestamp: timestamp_reg,
1672 });
1673
1674 let simple_index_name = format!("{}_lookup_index", source_field);
1675 ops.push(OpCode::UpdateLookupIndex {
1676 state_id: self.state_id,
1677 index_name: simple_index_name,
1678 lookup_value: lookup_reg,
1679 primary_key: key_reg,
1680 });
1681 } else {
1682 let index_name = format!("{}_lookup_index", source_field);
1683 ops.push(OpCode::UpdateLookupIndex {
1684 state_id: self.state_id,
1685 index_name,
1686 lookup_value: lookup_reg,
1687 primary_key: key_reg,
1688 });
1689 }
1690
1691 if source_path_opt.is_some() {
1695 ops.push(OpCode::UpdatePdaReverseLookup {
1696 state_id: self.state_id,
1697 lookup_name: "default_pda_lookup".to_string(),
1698 pda_address: lookup_reg,
1699 primary_key: key_reg,
1700 });
1701 }
1702 }
1703 KeyResolutionStrategy::Lookup { primary_field } => {
1704 let has_mapping_to_lookup_field = mappings
1707 .iter()
1708 .any(|m| m.target_path == lookup_index.field_name);
1709
1710 if has_mapping_to_lookup_field {
1711 let path_segments: Vec<&str> =
1714 primary_field.segments.iter().map(|s| s.as_str()).collect();
1715 ops.push(OpCode::LoadEventField {
1716 path: FieldPath::new(&path_segments),
1717 dest: lookup_reg,
1718 default: None,
1719 });
1720
1721 let index_name = format!("{}_lookup_index", source_field);
1722 ops.push(OpCode::UpdateLookupIndex {
1723 state_id: self.state_id,
1724 index_name,
1725 lookup_value: lookup_reg,
1726 primary_key: key_reg,
1727 });
1728 }
1729 }
1730 KeyResolutionStrategy::Computed { .. }
1731 | KeyResolutionStrategy::TemporalLookup { .. } => {
1732 }
1734 }
1735 }
1736
1737 ops
1738 }
1739
1740 fn get_event_type(&self, source: &SourceSpec) -> String {
1741 match source {
1742 SourceSpec::Source { type_name, .. } => type_name.clone(),
1743 }
1744 }
1745
1746 fn compile_instruction_hook_actions(&self, actions: &[HookAction]) -> Vec<OpCode> {
1747 let mut ops = Vec::new();
1748 let state_reg = 2;
1749
1750 for action in actions {
1751 match action {
1752 HookAction::SetField {
1753 target_field,
1754 source,
1755 condition,
1756 } => {
1757 let _ = condition;
1759
1760 let temp_reg = 11; let load_ops = self.compile_mapping_source(source, temp_reg);
1764 ops.extend(load_ops);
1765
1766 if let MappingSource::FromSource {
1768 transform: Some(transform_type),
1769 ..
1770 } = source
1771 {
1772 ops.push(OpCode::Transform {
1773 source: temp_reg,
1774 dest: temp_reg,
1775 transformation: transform_type.clone(),
1776 });
1777 }
1778
1779 if let Some(cond_expr) = condition {
1781 if let Some(parsed) = &cond_expr.parsed {
1782 let cond_check_ops = self.compile_condition_check(
1784 parsed,
1785 temp_reg,
1786 state_reg,
1787 target_field,
1788 );
1789 ops.extend(cond_check_ops);
1790 } else {
1791 ops.push(OpCode::SetField {
1793 object: state_reg,
1794 path: target_field.clone(),
1795 value: temp_reg,
1796 });
1797 }
1798 } else {
1799 ops.push(OpCode::SetField {
1801 object: state_reg,
1802 path: target_field.clone(),
1803 value: temp_reg,
1804 });
1805 }
1806 }
1807 HookAction::IncrementField {
1808 target_field,
1809 increment_by,
1810 condition,
1811 } => {
1812 if let Some(cond_expr) = condition {
1813 if let Some(parsed) = &cond_expr.parsed {
1814 let cond_check_ops = self.compile_conditional_increment(
1819 parsed,
1820 state_reg,
1821 target_field,
1822 *increment_by,
1823 );
1824 ops.extend(cond_check_ops);
1825 } else {
1826 ops.push(OpCode::SetFieldIncrement {
1828 object: state_reg,
1829 path: target_field.clone(),
1830 });
1831 }
1832 } else {
1833 ops.push(OpCode::SetFieldIncrement {
1835 object: state_reg,
1836 path: target_field.clone(),
1837 });
1838 }
1839 }
1840 HookAction::RegisterPdaMapping { .. } => {
1841 if let HookAction::RegisterPdaMapping {
1842 pda_field,
1843 seed_field,
1844 lookup_name,
1845 } = action
1846 {
1847 let pda_reg = 11;
1848 let seed_reg = 12;
1849
1850 ops.push(OpCode::LoadEventField {
1851 path: pda_field.clone(),
1852 dest: pda_reg,
1853 default: None,
1854 });
1855 ops.push(OpCode::LoadEventField {
1856 path: seed_field.clone(),
1857 dest: seed_reg,
1858 default: None,
1859 });
1860 ops.push(OpCode::UpdatePdaReverseLookup {
1861 state_id: self.state_id,
1862 lookup_name: lookup_name.clone(),
1863 pda_address: pda_reg,
1864 primary_key: seed_reg,
1865 });
1866 }
1867 }
1868 }
1869 }
1870
1871 ops
1872 }
1873
1874 fn compile_condition_check(
1875 &self,
1876 condition: &ParsedCondition,
1877 value_reg: Register,
1878 state_reg: Register,
1879 target_field: &str,
1880 ) -> Vec<OpCode> {
1881 match condition {
1882 ParsedCondition::Comparison {
1883 field,
1884 op,
1885 value: cond_value,
1886 } => {
1887 vec![OpCode::ConditionalSetField {
1889 object: state_reg,
1890 path: target_field.to_string(),
1891 value: value_reg,
1892 condition_field: field.clone(),
1893 condition_op: op.clone(),
1894 condition_value: cond_value.clone(),
1895 }]
1896 }
1897 ParsedCondition::Logical { .. } => {
1898 tracing::warn!("Logical conditions not yet supported in instruction hooks");
1900 vec![OpCode::SetField {
1901 object: state_reg,
1902 path: target_field.to_string(),
1903 value: value_reg,
1904 }]
1905 }
1906 }
1907 }
1908
1909 fn compile_conditional_increment(
1910 &self,
1911 condition: &ParsedCondition,
1912 state_reg: Register,
1913 target_field: &str,
1914 _increment_by: i64,
1915 ) -> Vec<OpCode> {
1916 match condition {
1917 ParsedCondition::Comparison {
1918 field,
1919 op,
1920 value: cond_value,
1921 } => {
1922 vec![OpCode::ConditionalIncrement {
1923 object: state_reg,
1924 path: target_field.to_string(),
1925 condition_field: field.clone(),
1926 condition_op: op.clone(),
1927 condition_value: cond_value.clone(),
1928 }]
1929 }
1930 ParsedCondition::Logical { .. } => {
1931 tracing::warn!("Logical conditions not yet supported in instruction hooks");
1932 vec![OpCode::SetFieldIncrement {
1933 object: state_reg,
1934 path: target_field.to_string(),
1935 }]
1936 }
1937 }
1938 }
1939}
1940
1941#[cfg(test)]
1942mod tests {
1943 use super::MultiEntityBytecode;
1944 use crate::ast::{
1945 FieldPath, HookAction, IdentitySpec, InstructionHook, KeyResolutionStrategy,
1946 LookupIndexSpec, MappingSource, PopulationStrategy, SerializableFieldMapping,
1947 SerializableHandlerSpec, SerializableStreamSpec, SourceSpec, TypedStreamSpec,
1948 };
1949 use crate::vm::VmContext;
1950 use serde_json::{json, Value};
1951 use std::collections::BTreeMap;
1952
1953 fn mapping(
1954 target_path: &str,
1955 source_path: &[&str],
1956 population: PopulationStrategy,
1957 ) -> SerializableFieldMapping {
1958 SerializableFieldMapping {
1959 target_path: target_path.to_string(),
1960 source: MappingSource::FromSource {
1961 path: FieldPath::new(source_path),
1962 default: None,
1963 transform: None,
1964 },
1965 transform: None,
1966 population,
1967 condition: None,
1968 when: None,
1969 stop: None,
1970 emit: true,
1971 }
1972 }
1973
1974 fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
1975 TypedStreamSpec::from_serializable(SerializableStreamSpec {
1976 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
1977 state_name: "PumpfunToken".to_string(),
1978 program_id: None,
1979 idl: None,
1980 identity: IdentitySpec {
1981 primary_keys: vec!["id.mint".to_string()],
1982 lookup_indexes: vec![LookupIndexSpec {
1983 field_name: "id.bonding_curve".to_string(),
1984 temporal_field: None,
1985 }],
1986 },
1987 handlers: vec![SerializableHandlerSpec {
1988 source: SourceSpec::Source {
1989 program_id: None,
1990 discriminator: None,
1991 type_name: "pump::BondingCurveState".to_string(),
1992 serialization: None,
1993 is_account: true,
1994 },
1995 key_resolution: KeyResolutionStrategy::Lookup {
1996 primary_field: FieldPath::new(&["__account_address"]),
1997 },
1998 mappings: vec![
1999 mapping(
2000 "id.bonding_curve",
2001 &["__account_address"],
2002 PopulationStrategy::SetOnce,
2003 ),
2004 mapping(
2005 "reserves.virtual_token_reserves",
2006 &["virtual_token_reserves"],
2007 PopulationStrategy::LastWrite,
2008 ),
2009 ],
2010 conditions: vec![],
2011 emit: true,
2012 }],
2013 sections: vec![],
2014 field_mappings: BTreeMap::new(),
2015 resolver_hooks: vec![],
2016 instruction_hooks: vec![InstructionHook {
2017 instruction_type: "pump::BuyIxState".to_string(),
2018 actions: vec![HookAction::RegisterPdaMapping {
2019 pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2020 seed_field: FieldPath::new(&["accounts", "mint"]),
2021 lookup_name: "default_pda_lookup".to_string(),
2022 }],
2023 lookup_by: None,
2024 }],
2025 resolver_specs: vec![],
2026 computed_fields: vec![],
2027 computed_field_specs: vec![],
2028 content_hash: None,
2029 views: vec![],
2030 })
2031 }
2032
2033 fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
2034 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2035 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2036 state_name: "OreRound".to_string(),
2037 program_id: None,
2038 idl: None,
2039 identity: IdentitySpec {
2040 primary_keys: vec!["id.round_id".to_string()],
2041 lookup_indexes: vec![LookupIndexSpec {
2042 field_name: "id.round_address".to_string(),
2043 temporal_field: None,
2044 }],
2045 },
2046 handlers: vec![SerializableHandlerSpec {
2047 source: SourceSpec::Source {
2048 program_id: None,
2049 discriminator: None,
2050 type_name: "entropy::VarState".to_string(),
2051 serialization: None,
2052 is_account: true,
2053 },
2054 key_resolution: KeyResolutionStrategy::Lookup {
2055 primary_field: FieldPath::new(&["__account_address"]),
2056 },
2057 mappings: vec![mapping(
2058 "state.expires_at",
2059 &["end_at"],
2060 PopulationStrategy::LastWrite,
2061 )],
2062 conditions: vec![],
2063 emit: true,
2064 }],
2065 sections: vec![],
2066 field_mappings: BTreeMap::new(),
2067 resolver_hooks: vec![],
2068 instruction_hooks: vec![InstructionHook {
2069 instruction_type: "ore::DeployIxState".to_string(),
2070 actions: vec![HookAction::RegisterPdaMapping {
2071 pda_field: FieldPath::new(&["accounts", "entropyVar"]),
2072 seed_field: FieldPath::new(&["accounts", "round"]),
2073 lookup_name: "default_pda_lookup".to_string(),
2074 }],
2075 lookup_by: None,
2076 }],
2077 resolver_specs: vec![],
2078 computed_fields: vec![],
2079 computed_field_specs: vec![],
2080 content_hash: None,
2081 views: vec![],
2082 })
2083 }
2084
2085 #[test]
2086 fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
2087 let bytecode = MultiEntityBytecode::from_single(
2088 "PumpfunToken".to_string(),
2089 direct_pda_to_primary_key_spec(),
2090 0,
2091 );
2092 let mut vm = VmContext::new();
2093
2094 let mutations = vm
2095 .process_event(
2096 &bytecode,
2097 json!({
2098 "__account_address": "bonding_curve_1",
2099 "__resolved_primary_key": "mint_1",
2100 "virtual_token_reserves": 42,
2101 }),
2102 "pump::BondingCurveState",
2103 None,
2104 None,
2105 )
2106 .unwrap();
2107
2108 assert_eq!(mutations.len(), 1);
2109 assert_eq!(mutations[0].key, json!("mint_1"));
2110 assert_eq!(
2111 mutations[0].patch["id"]["bonding_curve"],
2112 json!("bonding_curve_1")
2113 );
2114 assert_eq!(
2115 mutations[0].patch["reserves"]["virtual_token_reserves"],
2116 json!(42)
2117 );
2118 }
2119
2120 #[test]
2121 fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
2122 let bytecode = MultiEntityBytecode::from_single(
2123 "OreRound".to_string(),
2124 pda_to_intermediate_lookup_spec(),
2125 0,
2126 );
2127 let mut vm = VmContext::new();
2128
2129 let mutations = vm
2130 .process_event(
2131 &bytecode,
2132 json!({
2133 "__account_address": "entropy_var_1",
2134 "__resolved_primary_key": "round_address_1",
2135 "end_at": 123,
2136 }),
2137 "entropy::VarState",
2138 None,
2139 None,
2140 )
2141 .unwrap();
2142
2143 assert!(mutations.is_empty());
2144 }
2145}