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 ) -> std::result::Result<(), Box<dyn std::error::Error>>
250 + Send
251 + Sync,
252 >,
253 >,
254}
255
256impl std::fmt::Debug for EntityBytecode {
257 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
258 f.debug_struct("EntityBytecode")
259 .field("state_id", &self.state_id)
260 .field("handlers", &self.handlers)
261 .field("entity_name", &self.entity_name)
262 .field("when_events", &self.when_events)
263 .field("non_emitted_fields", &self.non_emitted_fields)
264 .field("computed_paths", &self.computed_paths)
265 .field(
266 "computed_fields_evaluator",
267 &self.computed_fields_evaluator.is_some(),
268 )
269 .finish()
270 }
271}
272
273#[derive(Debug)]
274pub struct MultiEntityBytecode {
275 pub entities: HashMap<String, EntityBytecode>,
276 pub event_routing: HashMap<String, Vec<String>>,
277 pub when_events: HashSet<String>,
278 pub proto_router: crate::proto_router::ProtoRouter,
279}
280
281impl MultiEntityBytecode {
282 pub fn from_single<S>(entity_name: String, spec: TypedStreamSpec<S>, state_id: u32) -> Self {
283 let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
284 let entity_bytecode = compiler.compile_entity();
285
286 let mut entities = HashMap::new();
287 let mut event_routing = HashMap::new();
288 let mut when_events = HashSet::new();
289
290 for event_type in entity_bytecode.handlers.keys() {
291 event_routing
292 .entry(event_type.clone())
293 .or_insert_with(Vec::new)
294 .push(entity_name.clone());
295 }
296
297 when_events.extend(entity_bytecode.when_events.iter().cloned());
298
299 entities.insert(entity_name, entity_bytecode);
300
301 MultiEntityBytecode {
302 entities,
303 event_routing,
304 when_events,
305 proto_router: crate::proto_router::ProtoRouter::new(),
306 }
307 }
308
309 pub fn from_entities(entities_vec: Vec<(String, Box<dyn std::any::Any>, u32)>) -> Self {
310 let entities = HashMap::new();
311 let event_routing = HashMap::new();
312 let when_events = HashSet::new();
313
314 if let Some((_entity_name, _spec_any, _state_id)) = entities_vec.into_iter().next() {
315 panic!("from_entities requires type information - use builder pattern instead");
316 }
317
318 MultiEntityBytecode {
319 entities,
320 event_routing,
321 when_events,
322 proto_router: crate::proto_router::ProtoRouter::new(),
323 }
324 }
325
326 #[allow(clippy::new_ret_no_self)]
327 pub fn new() -> MultiEntityBytecodeBuilder {
328 MultiEntityBytecodeBuilder {
329 entities: HashMap::new(),
330 event_routing: HashMap::new(),
331 when_events: HashSet::new(),
332 proto_router: crate::proto_router::ProtoRouter::new(),
333 }
334 }
335}
336
337pub struct MultiEntityBytecodeBuilder {
338 entities: HashMap<String, EntityBytecode>,
339 event_routing: HashMap<String, Vec<String>>,
340 when_events: HashSet<String>,
341 proto_router: crate::proto_router::ProtoRouter,
342}
343
344impl MultiEntityBytecodeBuilder {
345 pub fn add_entity<S>(
346 self,
347 entity_name: String,
348 spec: TypedStreamSpec<S>,
349 state_id: u32,
350 ) -> Self {
351 self.add_entity_with_evaluator(
352 entity_name,
353 spec,
354 state_id,
355 None::<
356 fn(
357 &mut Value,
358 Option<u64>,
359 i64,
360 ) -> std::result::Result<(), Box<dyn std::error::Error>>,
361 >,
362 )
363 }
364
365 pub fn add_entity_with_evaluator<S, F>(
366 mut self,
367 entity_name: String,
368 spec: TypedStreamSpec<S>,
369 state_id: u32,
370 evaluator: Option<F>,
371 ) -> Self
372 where
373 F: Fn(&mut Value, Option<u64>, i64) -> std::result::Result<(), Box<dyn std::error::Error>>
374 + Send
375 + Sync
376 + 'static,
377 {
378 let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
379 let mut entity_bytecode = compiler.compile_entity();
380
381 if let Some(eval) = evaluator {
383 entity_bytecode.computed_fields_evaluator = Some(Box::new(eval));
384 }
385
386 for event_type in entity_bytecode.handlers.keys() {
387 self.event_routing
388 .entry(event_type.clone())
389 .or_default()
390 .push(entity_name.clone());
391 }
392
393 self.when_events
394 .extend(entity_bytecode.when_events.iter().cloned());
395
396 self.entities.insert(entity_name, entity_bytecode);
397 self
398 }
399
400 pub fn build(self) -> MultiEntityBytecode {
401 MultiEntityBytecode {
402 entities: self.entities,
403 event_routing: self.event_routing,
404 when_events: self.when_events,
405 proto_router: self.proto_router,
406 }
407 }
408}
409
410pub struct TypedCompiler<S> {
411 pub spec: TypedStreamSpec<S>,
412 entity_name: String,
413 state_id: u32,
414}
415
416impl<S> TypedCompiler<S> {
417 pub fn new(spec: TypedStreamSpec<S>, entity_name: String) -> Self {
418 TypedCompiler {
419 spec,
420 entity_name,
421 state_id: 0,
422 }
423 }
424
425 pub fn with_state_id(mut self, state_id: u32) -> Self {
426 self.state_id = state_id;
427 self
428 }
429
430 pub fn compile(&self) -> MultiEntityBytecode {
431 let entity_bytecode = self.compile_entity();
432
433 let mut entities = HashMap::new();
434 let mut event_routing = HashMap::new();
435 let mut when_events = HashSet::new();
436
437 for event_type in entity_bytecode.handlers.keys() {
438 event_routing
439 .entry(event_type.clone())
440 .or_insert_with(Vec::new)
441 .push(self.entity_name.clone());
442 }
443
444 when_events.extend(entity_bytecode.when_events.iter().cloned());
445
446 entities.insert(self.entity_name.clone(), entity_bytecode);
447
448 MultiEntityBytecode {
449 entities,
450 event_routing,
451 when_events,
452 proto_router: crate::proto_router::ProtoRouter::new(),
453 }
454 }
455
456 fn resolver_outputs_primary_key_directly(&self, primary_field: &FieldPath) -> bool {
457 if primary_field.segments.as_slice() != ["__account_address"] {
458 return false;
459 }
460
461 let primary_key_leafs: HashSet<&str> = self
462 .spec
463 .identity
464 .primary_keys
465 .iter()
466 .map(|path| path.rsplit('.').next().unwrap_or(path.as_str()))
467 .collect();
468
469 self.spec.instruction_hooks.iter().any(|hook| {
470 hook.actions.iter().any(|action| {
471 matches!(action, HookAction::RegisterPdaMapping { seed_field, .. }
472 if seed_field
473 .segments
474 .last()
475 .is_some_and(|segment| primary_key_leafs.contains(segment.as_str())))
476 })
477 })
478 }
479
480 fn compile_entity(&self) -> EntityBytecode {
481 let mut handlers: HashMap<String, Vec<OpCode>> = HashMap::new();
482 let mut when_events: HashSet<String> = HashSet::new();
483 let mut emit_by_path: HashMap<String, bool> = HashMap::new();
484
485 let mut debug_info = Vec::new();
487 for (index, handler_spec) in self.spec.handlers.iter().enumerate() {
488 let event_type = self.get_event_type(&handler_spec.source);
489 let program_id = match &handler_spec.source {
490 crate::ast::SourceSpec::Source { program_id, .. } => {
491 program_id.as_ref().map(|s| s.as_str()).unwrap_or("null")
492 }
493 };
494 debug_info.push(format!(
495 " [{}] EventType={}, Mappings={}, ProgramId={}",
496 index,
497 event_type,
498 handler_spec.mappings.len(),
499 program_id
500 ));
501 }
502
503 for handler_spec in &self.spec.handlers {
513 for mapping in &handler_spec.mappings {
514 if let Some(when) = &mapping.when {
515 when_events.insert(when.clone());
516 }
517 let entry = emit_by_path
518 .entry(mapping.target_path.clone())
519 .or_insert(false);
520 *entry |= mapping.emit;
521 if mapping.stop.is_some() {
522 emit_by_path
523 .entry(stop_field_path(&mapping.target_path))
524 .or_insert(false);
525 }
526 }
527 let opcodes = self.compile_handler(handler_spec);
528 let event_type = self.get_event_type(&handler_spec.source);
529
530 if let Some(existing_opcodes) = handlers.get_mut(&event_type) {
531 let mut existing_setup = Vec::new();
536 let mut existing_mappings = Vec::new();
537 let mut existing_teardown = Vec::new();
538 let mut section = 0; for opcode in existing_opcodes.iter() {
541 match opcode {
542 OpCode::ReadOrInitState { .. } => {
543 existing_setup.push(opcode.clone());
544 section = 1; }
546 OpCode::UpdateState { .. } => {
547 existing_teardown.push(opcode.clone());
548 section = 2; }
550 OpCode::EmitMutation { .. } => {
551 existing_teardown.push(opcode.clone());
552 }
553 _ if section == 0 => existing_setup.push(opcode.clone()),
554 _ if section == 1 => existing_mappings.push(opcode.clone()),
555 _ => existing_teardown.push(opcode.clone()),
556 }
557 }
558
559 let mut new_mappings = Vec::new();
561 section = 0;
562
563 for opcode in opcodes.iter() {
564 match opcode {
565 OpCode::ReadOrInitState { .. } => {
566 section = 1; }
568 OpCode::UpdateState { .. } | OpCode::EmitMutation { .. } => {
569 section = 2; }
571 _ if section == 1 => {
572 new_mappings.push(opcode.clone());
573 }
574 _ => {} }
576 }
577
578 let mut merged = Vec::new();
580 merged.extend(existing_setup);
581 merged.extend(existing_mappings);
582 merged.extend(new_mappings.clone());
583 merged.extend(existing_teardown);
584
585 *existing_opcodes = merged;
586 } else {
587 handlers.insert(event_type, opcodes);
588 }
589 }
590
591 for hook in &self.spec.instruction_hooks {
593 let event_type = hook.instruction_type.clone();
594
595 let handler_opcodes = handlers.entry(event_type.clone()).or_insert_with(|| {
596 let key_reg = 20;
597 let state_reg = 2;
598 let resolved_key_reg = 19;
599 let temp_reg = 18;
600
601 let mut ops = Vec::new();
602
603 ops.push(OpCode::LoadEventField {
605 path: FieldPath::new(&["__resolved_primary_key"]),
606 dest: resolved_key_reg,
607 default: Some(serde_json::json!(null)),
608 });
609
610 ops.push(OpCode::CopyRegister {
612 source: resolved_key_reg,
613 dest: key_reg,
614 });
615
616 if let Some(lookup_path) = &hook.lookup_by {
618 ops.push(OpCode::LoadEventField {
620 path: lookup_path.clone(),
621 dest: temp_reg,
622 default: None,
623 });
624
625 ops.push(OpCode::Transform {
627 source: temp_reg,
628 dest: temp_reg,
629 transformation: Transformation::HexEncode,
630 });
631
632 ops.push(OpCode::CopyRegisterIfNull {
634 source: temp_reg,
635 dest: key_reg,
636 });
637 }
638
639 ops.push(OpCode::ReadOrInitState {
640 state_id: self.state_id,
641 key: key_reg,
642 default: serde_json::json!({}),
643 dest: state_reg,
644 });
645
646 ops.push(OpCode::UpdateState {
647 state_id: self.state_id,
648 key: key_reg,
649 value: state_reg,
650 });
651
652 ops
653 });
654
655 let hook_opcodes = self.compile_instruction_hook_actions(&hook.actions);
657
658 let insert_pos = handler_opcodes
662 .iter()
663 .position(|op| matches!(op, OpCode::EvaluateComputedFields { .. }))
664 .or_else(|| {
665 handler_opcodes
666 .iter()
667 .position(|op| matches!(op, OpCode::UpdateState { .. }))
668 });
669
670 if let Some(pos) = insert_pos {
671 for (i, opcode) in hook_opcodes.into_iter().enumerate() {
673 handler_opcodes.insert(pos + i, opcode);
674 }
675 }
676 }
677
678 let non_emitted_fields: HashSet<String> = emit_by_path
679 .into_iter()
680 .filter_map(|(path, emit)| if emit { None } else { Some(path) })
681 .collect();
682
683 EntityBytecode {
684 state_id: self.state_id,
685 handlers,
686 entity_name: self.entity_name.clone(),
687 when_events,
688 non_emitted_fields,
689 computed_paths: self.spec.computed_fields.clone(),
690 computed_fields_evaluator: None,
691 }
692 }
693
694 fn compile_handler(&self, spec: &TypedHandlerSpec<S>) -> Vec<OpCode> {
695 let mut ops = Vec::new();
696 let state_reg = 2;
697 let key_reg = 20;
698
699 ops.extend(self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings));
700
701 let is_account_event = matches!(
707 spec.source,
708 SourceSpec::Source {
709 is_account: true,
710 ..
711 }
712 );
713 ops.push(OpCode::AbortIfNullKey {
714 key: key_reg,
715 is_account_event,
716 });
717
718 ops.push(OpCode::ReadOrInitState {
719 state_id: self.state_id,
720 key: key_reg,
721 default: serde_json::json!({}),
722 dest: state_reg,
723 });
724
725 ops.extend(self.compile_temporal_index_update(
732 &spec.key_resolution,
733 key_reg,
734 &spec.mappings,
735 ));
736
737 for mapping in &spec.mappings {
738 ops.extend(self.compile_mapping(mapping, state_reg, key_reg));
739 }
740
741 ops.extend(self.compile_resolvers(state_reg, key_reg));
742
743 ops.push(OpCode::EvaluateComputedFields {
745 state: state_reg,
746 computed_paths: self.spec.computed_fields.clone(),
747 });
748
749 ops.push(OpCode::UpdateState {
750 state_id: self.state_id,
751 key: key_reg,
752 value: state_reg,
753 });
754
755 if spec.emit {
756 ops.push(OpCode::EmitMutation {
757 entity_name: self.entity_name.clone(),
758 key: key_reg,
759 state: state_reg,
760 });
761 }
762
763 ops
764 }
765
766 fn compile_resolvers(&self, state_reg: Register, key_reg: Register) -> Vec<OpCode> {
767 let mut ops = Vec::new();
768
769 for resolver_spec in &self.spec.resolver_specs {
770 let url_template = match &resolver_spec.resolver {
771 ResolverType::Url(config) => match &config.url_source {
772 UrlSource::Template(parts) => Some(parts.clone()),
773 _ => None,
774 },
775 _ => None,
776 };
777
778 ops.push(OpCode::QueueResolver {
779 state_id: self.state_id,
780 entity_name: self.entity_name.clone(),
781 resolver: resolver_spec.resolver.clone(),
782 input_path: resolver_spec.input_path.clone(),
783 input_value: resolver_spec.input_value.clone(),
784 url_template,
785 strategy: resolver_spec.strategy.clone(),
786 extracts: resolver_spec.extracts.clone(),
787 condition: resolver_spec.condition.clone(),
788 schedule_at: resolver_spec.schedule_at.clone(),
789 state: state_reg,
790 key: key_reg,
791 });
792 }
793
794 ops
795 }
796
797 fn compile_mapping(
798 &self,
799 mapping: &TypedFieldMapping<S>,
800 state_reg: Register,
801 key_reg: Register,
802 ) -> Vec<OpCode> {
803 let mut ops = Vec::new();
804 let temp_reg = 10;
805
806 ops.extend(self.compile_mapping_source(&mapping.source, temp_reg));
807
808 if let Some(transform) = &mapping.transform {
809 ops.push(OpCode::Transform {
810 source: temp_reg,
811 dest: temp_reg,
812 transformation: transform.clone(),
813 });
814 }
815
816 if let Some(stop_instruction) = &mapping.stop {
817 if mapping.when.is_some() {
818 tracing::warn!(
819 "#[map] stop and when both set for {}. Ignoring when.",
820 mapping.target_path
821 );
822 }
823 if !matches!(mapping.population, PopulationStrategy::LastWrite)
824 && !matches!(mapping.population, PopulationStrategy::Merge)
825 {
826 tracing::warn!(
827 "#[map] stop ignores population strategy {:?}",
828 mapping.population
829 );
830 }
831
832 ops.push(OpCode::SetFieldUnlessStopped {
833 object: state_reg,
834 path: mapping.target_path.clone(),
835 value: temp_reg,
836 stop_field: stop_field_path(&mapping.target_path),
837 stop_instruction: stop_instruction.clone(),
838 entity_name: self.entity_name.clone(),
839 key_reg,
840 });
841 return ops;
842 }
843
844 if let Some(when_instruction) = &mapping.when {
845 if !matches!(mapping.population, PopulationStrategy::LastWrite)
846 && !matches!(mapping.population, PopulationStrategy::Merge)
847 {
848 tracing::warn!(
849 "#[map] when ignores population strategy {:?}",
850 mapping.population
851 );
852 }
853 let (condition_field, condition_op, condition_value) = mapping
854 .condition
855 .as_ref()
856 .and_then(|cond| cond.parsed.as_ref())
857 .and_then(|parsed| match parsed {
858 ParsedCondition::Comparison { field, op, value } => {
859 Some((Some(field.clone()), Some(op.clone()), Some(value.clone())))
860 }
861 ParsedCondition::Logical { .. } => {
862 tracing::warn!("Logical conditions not yet supported for #[map] when");
863 None
864 }
865 })
866 .unwrap_or((None, None, None));
867
868 ops.push(OpCode::SetFieldWhen {
869 object: state_reg,
870 path: mapping.target_path.clone(),
871 value: temp_reg,
872 when_instruction: when_instruction.clone(),
873 entity_name: self.entity_name.clone(),
874 key_reg,
875 condition_field,
876 condition_op,
877 condition_value,
878 });
879 return ops;
880 }
881
882 if let Some(condition) = &mapping.condition {
883 if let Some(parsed) = &condition.parsed {
884 match parsed {
885 ParsedCondition::Comparison {
886 field,
887 op,
888 value: cond_value,
889 } => {
890 if matches!(mapping.population, PopulationStrategy::LastWrite)
891 || matches!(mapping.population, PopulationStrategy::Merge)
892 {
893 ops.push(OpCode::ConditionalSetField {
894 object: state_reg,
895 path: mapping.target_path.clone(),
896 value: temp_reg,
897 condition_field: field.clone(),
898 condition_op: op.clone(),
899 condition_value: cond_value.clone(),
900 });
901 return ops;
902 }
903
904 if matches!(mapping.population, PopulationStrategy::Count) {
905 ops.push(OpCode::ConditionalIncrement {
906 object: state_reg,
907 path: mapping.target_path.clone(),
908 condition_field: field.clone(),
909 condition_op: op.clone(),
910 condition_value: cond_value.clone(),
911 });
912 return ops;
913 }
914
915 tracing::warn!(
916 "Conditional #[map] not supported for population strategy {:?}",
917 mapping.population
918 );
919 }
920 ParsedCondition::Logical { .. } => {
921 tracing::warn!("Logical conditions not yet supported for #[map]");
922 }
923 }
924 }
925 }
926
927 match &mapping.population {
928 PopulationStrategy::Append => {
929 ops.push(OpCode::AppendToArray {
930 object: state_reg,
931 path: mapping.target_path.clone(),
932 value: temp_reg,
933 });
934 }
935 PopulationStrategy::LastWrite => {
936 ops.push(OpCode::SetField {
937 object: state_reg,
938 path: mapping.target_path.clone(),
939 value: temp_reg,
940 });
941 }
942 PopulationStrategy::SetOnce => {
943 ops.push(OpCode::SetFieldIfNull {
944 object: state_reg,
945 path: mapping.target_path.clone(),
946 value: temp_reg,
947 });
948 }
949 PopulationStrategy::Merge => {
950 ops.push(OpCode::SetField {
951 object: state_reg,
952 path: mapping.target_path.clone(),
953 value: temp_reg,
954 });
955 }
956 PopulationStrategy::Max => {
957 ops.push(OpCode::SetFieldMax {
958 object: state_reg,
959 path: mapping.target_path.clone(),
960 value: temp_reg,
961 });
962 }
963 PopulationStrategy::Sum => {
964 ops.push(OpCode::SetFieldSum {
965 object: state_reg,
966 path: mapping.target_path.clone(),
967 value: temp_reg,
968 });
969 }
970 PopulationStrategy::Count => {
971 ops.push(OpCode::SetFieldIncrement {
973 object: state_reg,
974 path: mapping.target_path.clone(),
975 });
976 }
977 PopulationStrategy::Min => {
978 ops.push(OpCode::SetFieldMin {
979 object: state_reg,
980 path: mapping.target_path.clone(),
981 value: temp_reg,
982 });
983 }
984 PopulationStrategy::UniqueCount => {
985 let set_name = format!("{}_unique_set", mapping.target_path);
988 ops.push(OpCode::AddToUniqueSet {
989 state_id: self.state_id,
990 set_name,
991 value: temp_reg,
992 count_object: state_reg,
993 count_path: mapping.target_path.clone(),
994 });
995 }
996 }
997
998 ops
999 }
1000
1001 fn compile_mapping_source(&self, source: &MappingSource, dest: Register) -> Vec<OpCode> {
1002 match source {
1003 MappingSource::FromSource {
1004 path,
1005 default,
1006 transform,
1007 } => {
1008 let mut ops = vec![OpCode::LoadEventField {
1009 path: path.clone(),
1010 dest,
1011 default: default.clone(),
1012 }];
1013
1014 if let Some(transform_type) = transform {
1016 ops.push(OpCode::Transform {
1017 source: dest,
1018 dest,
1019 transformation: transform_type.clone(),
1020 });
1021 }
1022
1023 ops
1024 }
1025 MappingSource::Constant(val) => {
1026 vec![OpCode::LoadConstant {
1027 value: val.clone(),
1028 dest,
1029 }]
1030 }
1031 MappingSource::AsEvent { fields } => {
1032 let mut ops = Vec::new();
1033
1034 if fields.is_empty() {
1035 let event_data_reg = dest + 1;
1036 ops.push(OpCode::LoadEventField {
1037 path: FieldPath::new(&[]),
1038 dest: event_data_reg,
1039 default: Some(serde_json::json!({})),
1040 });
1041 ops.push(OpCode::CreateEvent {
1042 dest,
1043 event_value: event_data_reg,
1044 });
1045 } else {
1046 let data_obj_reg = dest + 1;
1047 ops.push(OpCode::CreateObject { dest: data_obj_reg });
1048
1049 let mut field_registers = Vec::new();
1050 let mut current_reg = dest + 2;
1051
1052 for field_source in fields.iter() {
1053 if let MappingSource::FromSource {
1054 path,
1055 default,
1056 transform,
1057 } = &**field_source
1058 {
1059 ops.push(OpCode::LoadEventField {
1060 path: path.clone(),
1061 dest: current_reg,
1062 default: default.clone(),
1063 });
1064
1065 if let Some(transform_type) = transform {
1066 ops.push(OpCode::Transform {
1067 source: current_reg,
1068 dest: current_reg,
1069 transformation: transform_type.clone(),
1070 });
1071 }
1072
1073 if let Some(field_name) = path.segments.last() {
1074 field_registers.push((field_name.clone(), current_reg));
1075 }
1076 current_reg += 1;
1077 }
1078 }
1079
1080 if !field_registers.is_empty() {
1081 ops.push(OpCode::SetFields {
1082 object: data_obj_reg,
1083 fields: field_registers,
1084 });
1085 }
1086
1087 ops.push(OpCode::CreateEvent {
1088 dest,
1089 event_value: data_obj_reg,
1090 });
1091 }
1092
1093 ops
1094 }
1095 MappingSource::WholeSource => {
1096 vec![OpCode::LoadEventField {
1097 path: FieldPath::new(&[]),
1098 dest,
1099 default: Some(serde_json::json!({})),
1100 }]
1101 }
1102 MappingSource::AsCapture { field_transforms } => {
1103 let capture_data_reg = 22; let mut ops = vec![OpCode::LoadEventField {
1106 path: FieldPath::new(&[]),
1107 dest: capture_data_reg,
1108 default: Some(serde_json::json!({})),
1109 }];
1110
1111 let field_reg = 24;
1115 let transformed_reg = 25;
1116
1117 for (field_name, transform) in field_transforms {
1118 ops.push(OpCode::GetField {
1121 object: capture_data_reg,
1122 path: field_name.clone(),
1123 dest: field_reg,
1124 });
1125
1126 ops.push(OpCode::Transform {
1128 source: field_reg,
1129 dest: transformed_reg,
1130 transformation: transform.clone(),
1131 });
1132
1133 ops.push(OpCode::SetField {
1135 object: capture_data_reg,
1136 path: field_name.clone(),
1137 value: transformed_reg,
1138 });
1139 }
1140
1141 ops.push(OpCode::CreateCapture {
1143 dest,
1144 capture_value: capture_data_reg,
1145 });
1146
1147 ops
1148 }
1149 MappingSource::FromContext { field } => {
1150 vec![OpCode::LoadEventField {
1152 path: FieldPath::new(&["__update_context", field.as_str()]),
1153 dest,
1154 default: Some(serde_json::json!(null)),
1155 }]
1156 }
1157 MappingSource::Computed { .. } => {
1158 vec![]
1159 }
1160 MappingSource::FromState { .. } => {
1161 vec![]
1162 }
1163 }
1164 }
1165
1166 pub fn compile_key_loading(
1167 &self,
1168 resolution: &KeyResolutionStrategy,
1169 key_reg: Register,
1170 mappings: &[TypedFieldMapping<S>],
1171 ) -> Vec<OpCode> {
1172 let mut ops = Vec::new();
1173
1174 let resolved_key_reg = 19; ops.push(OpCode::LoadEventField {
1178 path: FieldPath::new(&["__resolved_primary_key"]),
1179 dest: resolved_key_reg,
1180 default: Some(serde_json::json!(null)),
1181 });
1182
1183 match resolution {
1185 KeyResolutionStrategy::Embedded { primary_field } => {
1186 ops.push(OpCode::CopyRegister {
1188 source: resolved_key_reg,
1189 dest: key_reg,
1190 });
1191
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::CopyRegisterIfNull {
1232 source: transform_reg,
1233 dest: key_reg,
1234 });
1235 } else {
1236 ops.push(OpCode::CopyRegisterIfNull {
1238 source: temp_reg,
1239 dest: key_reg,
1240 });
1241 }
1242 }
1243 }
1246 KeyResolutionStrategy::Lookup { primary_field } => {
1247 let lookup_reg = 15;
1248 let result_reg = 17;
1249
1250 ops.push(OpCode::CopyRegister {
1256 source: resolved_key_reg,
1257 dest: lookup_reg,
1258 });
1259
1260 let temp_reg = 18;
1261 ops.push(OpCode::LoadEventField {
1262 path: primary_field.clone(),
1263 dest: temp_reg,
1264 default: None,
1265 });
1266 ops.push(OpCode::CopyRegisterIfNull {
1267 source: temp_reg,
1268 dest: lookup_reg,
1269 });
1270
1271 let index_name = self.find_lookup_index_for_lookup_field(primary_field, mappings);
1272 let effective_index_name =
1273 index_name.unwrap_or_else(|| "default_pda_lookup".to_string());
1274
1275 ops.push(OpCode::LookupIndex {
1276 state_id: self.state_id,
1277 index_name: effective_index_name,
1278 lookup_value: lookup_reg,
1279 dest: result_reg,
1280 });
1281 ops.push(OpCode::CopyRegister {
1282 source: result_reg,
1283 dest: key_reg,
1284 });
1285
1286 if self.resolver_outputs_primary_key_directly(primary_field) {
1294 ops.push(OpCode::CopyRegisterIfNull {
1295 source: resolved_key_reg,
1296 dest: key_reg,
1297 });
1298 }
1299 }
1300 KeyResolutionStrategy::Computed {
1301 primary_field,
1302 compute_partition: _,
1303 } => {
1304 ops.push(OpCode::CopyRegister {
1306 source: resolved_key_reg,
1307 dest: key_reg,
1308 });
1309 let temp_reg = 18;
1310 ops.push(OpCode::LoadEventField {
1311 path: primary_field.clone(),
1312 dest: temp_reg,
1313 default: None,
1314 });
1315 ops.push(OpCode::CopyRegisterIfNull {
1316 source: temp_reg,
1317 dest: key_reg,
1318 });
1319 }
1320 KeyResolutionStrategy::TemporalLookup {
1321 lookup_field,
1322 timestamp_field,
1323 index_name,
1324 } => {
1325 ops.push(OpCode::CopyRegister {
1327 source: resolved_key_reg,
1328 dest: key_reg,
1329 });
1330 let lookup_reg = 15;
1331 let timestamp_reg = 16;
1332 let result_reg = 17;
1333
1334 ops.push(OpCode::LoadEventField {
1335 path: lookup_field.clone(),
1336 dest: lookup_reg,
1337 default: None,
1338 });
1339
1340 ops.push(OpCode::LoadEventField {
1341 path: timestamp_field.clone(),
1342 dest: timestamp_reg,
1343 default: None,
1344 });
1345
1346 ops.push(OpCode::LookupTemporalIndex {
1347 state_id: self.state_id,
1348 index_name: index_name.clone(),
1349 lookup_value: lookup_reg,
1350 timestamp: timestamp_reg,
1351 dest: result_reg,
1352 });
1353
1354 ops.push(OpCode::CopyRegisterIfNull {
1355 source: result_reg,
1356 dest: key_reg,
1357 });
1358 }
1359 }
1360
1361 ops
1362 }
1363
1364 fn find_primary_key_transformation(
1365 &self,
1366 mappings: &[TypedFieldMapping<S>],
1367 ) -> Option<Transformation> {
1368 let primary_key = self.spec.identity.primary_keys.first()?;
1370 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1371
1372 for mapping in mappings {
1374 if mapping.target_path == *primary_key
1376 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1377 {
1378 if let Some(transform) = &mapping.transform {
1380 return Some(transform.clone());
1381 }
1382
1383 if let MappingSource::FromSource {
1385 transform: Some(transform),
1386 ..
1387 } = &mapping.source
1388 {
1389 return Some(transform.clone());
1390 }
1391 }
1392 }
1393
1394 for mapping in mappings {
1396 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1397 if let Some(transform) = field_transforms.get(&primary_field_name) {
1398 return Some(transform.clone());
1399 }
1400 }
1401 }
1402
1403 None
1404 }
1405
1406 pub fn find_inherited_primary_key_transformation(&self) -> Option<Transformation> {
1409 let primary_key = self.spec.identity.primary_keys.first()?;
1410
1411 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1413
1414 for handler in &self.spec.handlers {
1416 for mapping in &handler.mappings {
1417 if mapping.target_path == *primary_key
1419 || mapping.target_path.ends_with(&format!(".{}", primary_key))
1420 {
1421 if let MappingSource::FromSource {
1423 path, transform, ..
1424 } = &mapping.source
1425 {
1426 if path.segments.last() == Some(&primary_field_name) {
1427 return mapping.transform.clone().or_else(|| transform.clone());
1429 }
1430 }
1431 }
1432
1433 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1435 if let Some(transform) = field_transforms.get(&primary_field_name) {
1436 return Some(transform.clone());
1437 }
1438 }
1439 }
1440 }
1441
1442 None
1443 }
1444
1445 fn extract_primary_field_name(&self, primary_key: &str) -> Option<String> {
1447 primary_key.split('.').next_back().map(|s| s.to_string())
1449 }
1450
1451 pub fn auto_detect_primary_field(
1454 &self,
1455 current_mappings: &[TypedFieldMapping<S>],
1456 ) -> Option<FieldPath> {
1457 let primary_key = self.spec.identity.primary_keys.first()?;
1458
1459 let primary_field_name = self.extract_primary_field_name(primary_key)?;
1461
1462 if self.current_account_has_primary_field(&primary_field_name, current_mappings) {
1464 return Some(FieldPath::new(&[&primary_field_name]));
1465 }
1466
1467 None
1468 }
1469
1470 fn current_account_has_primary_field(
1473 &self,
1474 field_name: &str,
1475 mappings: &[TypedFieldMapping<S>],
1476 ) -> bool {
1477 for mapping in mappings {
1479 if let MappingSource::FromSource { path, .. } = &mapping.source {
1480 if path.segments.last() == Some(&field_name.to_string()) {
1482 return true;
1483 }
1484 }
1485 }
1486
1487 false
1488 }
1489
1490 #[allow(dead_code)]
1492 fn handler_has_field(&self, field_name: &str, mappings: &[TypedFieldMapping<S>]) -> bool {
1493 for mapping in mappings {
1494 if let MappingSource::FromSource { path, .. } = &mapping.source {
1495 if path.segments.last() == Some(&field_name.to_string()) {
1496 return true;
1497 }
1498 }
1499 }
1500 false
1501 }
1502
1503 #[allow(dead_code)]
1506 fn field_exists_in_mappings(
1507 &self,
1508 field_name: &str,
1509 mappings: &[TypedFieldMapping<S>],
1510 ) -> bool {
1511 for mapping in mappings {
1513 if let MappingSource::FromSource { path, .. } = &mapping.source {
1514 if path.segments.last() == Some(&field_name.to_string()) {
1515 return true;
1516 }
1517 }
1518 if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1520 if field_transforms.contains_key(field_name) {
1521 return true;
1522 }
1523 }
1524 }
1525 false
1526 }
1527
1528 fn find_lookup_index_for_field(&self, field_path: &FieldPath) -> Option<String> {
1529 if field_path.segments.is_empty() {
1530 return None;
1531 }
1532
1533 let lookup_field_name = field_path.segments.last().unwrap();
1534
1535 for lookup_index in &self.spec.identity.lookup_indexes {
1536 let index_field_name = lookup_index
1537 .field_name
1538 .split('.')
1539 .next_back()
1540 .unwrap_or(&lookup_index.field_name);
1541 let matches_directly = index_field_name == lookup_field_name;
1542 let matches_address_alias = index_field_name
1549 .strip_suffix("_address")
1550 .map(|base| base == lookup_field_name)
1551 .unwrap_or(false);
1552
1553 if matches_directly || matches_address_alias {
1554 return Some(format!("{}_lookup_index", index_field_name));
1555 }
1556 }
1557
1558 None
1559 }
1560
1561 fn find_lookup_index_for_lookup_field(
1564 &self,
1565 primary_field: &FieldPath,
1566 mappings: &[TypedFieldMapping<S>],
1567 ) -> Option<String> {
1568 let primary_path = primary_field.segments.join(".");
1570
1571 for mapping in mappings {
1573 if let MappingSource::FromSource { path, .. } = &mapping.source {
1575 let source_path = path.segments.join(".");
1576 if source_path == primary_path {
1577 for lookup_index in &self.spec.identity.lookup_indexes {
1579 if mapping.target_path == lookup_index.field_name {
1580 let index_field_name = lookup_index
1581 .field_name
1582 .split('.')
1583 .next_back()
1584 .unwrap_or(&lookup_index.field_name);
1585 return Some(format!("{}_lookup_index", index_field_name));
1586 }
1587 }
1588 }
1589 }
1590 }
1591
1592 self.find_lookup_index_for_field(primary_field)
1594 }
1595
1596 fn find_source_path_for_lookup_index(
1600 &self,
1601 mappings: &[TypedFieldMapping<S>],
1602 lookup_field_name: &str,
1603 ) -> Option<Vec<String>> {
1604 for mapping in mappings {
1605 if mapping.target_path == lookup_field_name {
1606 if let MappingSource::FromSource { path, .. } = &mapping.source {
1607 return Some(path.segments.clone());
1608 }
1609 }
1610 }
1611 None
1612 }
1613
1614 fn compile_temporal_index_update(
1615 &self,
1616 resolution: &KeyResolutionStrategy,
1617 key_reg: Register,
1618 mappings: &[TypedFieldMapping<S>],
1619 ) -> Vec<OpCode> {
1620 let mut ops = Vec::new();
1621
1622 for lookup_index in &self.spec.identity.lookup_indexes {
1623 let lookup_reg = 17;
1624 let source_field = lookup_index
1625 .field_name
1626 .split('.')
1627 .next_back()
1628 .unwrap_or(&lookup_index.field_name);
1629
1630 match resolution {
1631 KeyResolutionStrategy::Embedded { primary_field: _ } => {
1632 let source_path_opt =
1635 self.find_source_path_for_lookup_index(mappings, &lookup_index.field_name);
1636
1637 let load_path = if let Some(ref path) = source_path_opt {
1638 FieldPath::new(&path.iter().map(|s| s.as_str()).collect::<Vec<_>>())
1639 } else {
1640 FieldPath::new(&[source_field])
1642 };
1643
1644 ops.push(OpCode::LoadEventField {
1645 path: load_path,
1646 dest: lookup_reg,
1647 default: None,
1648 });
1649
1650 if let Some(temporal_field_name) = &lookup_index.temporal_field {
1651 let timestamp_reg = 18;
1652
1653 ops.push(OpCode::LoadEventField {
1654 path: FieldPath::new(&[temporal_field_name]),
1655 dest: timestamp_reg,
1656 default: None,
1657 });
1658
1659 let index_name = format!("{}_temporal_index", source_field);
1660 ops.push(OpCode::UpdateTemporalIndex {
1661 state_id: self.state_id,
1662 index_name,
1663 lookup_value: lookup_reg,
1664 primary_key: key_reg,
1665 timestamp: timestamp_reg,
1666 });
1667
1668 let simple_index_name = format!("{}_lookup_index", source_field);
1669 ops.push(OpCode::UpdateLookupIndex {
1670 state_id: self.state_id,
1671 index_name: simple_index_name,
1672 lookup_value: lookup_reg,
1673 primary_key: key_reg,
1674 });
1675 } else {
1676 let index_name = format!("{}_lookup_index", source_field);
1677 ops.push(OpCode::UpdateLookupIndex {
1678 state_id: self.state_id,
1679 index_name,
1680 lookup_value: lookup_reg,
1681 primary_key: key_reg,
1682 });
1683 }
1684
1685 if source_path_opt.is_some() {
1689 ops.push(OpCode::UpdatePdaReverseLookup {
1690 state_id: self.state_id,
1691 lookup_name: "default_pda_lookup".to_string(),
1692 pda_address: lookup_reg,
1693 primary_key: key_reg,
1694 });
1695 }
1696 }
1697 KeyResolutionStrategy::Lookup { primary_field } => {
1698 let has_mapping_to_lookup_field = mappings
1701 .iter()
1702 .any(|m| m.target_path == lookup_index.field_name);
1703
1704 if has_mapping_to_lookup_field {
1705 let path_segments: Vec<&str> =
1708 primary_field.segments.iter().map(|s| s.as_str()).collect();
1709 ops.push(OpCode::LoadEventField {
1710 path: FieldPath::new(&path_segments),
1711 dest: lookup_reg,
1712 default: None,
1713 });
1714
1715 let index_name = format!("{}_lookup_index", source_field);
1716 ops.push(OpCode::UpdateLookupIndex {
1717 state_id: self.state_id,
1718 index_name,
1719 lookup_value: lookup_reg,
1720 primary_key: key_reg,
1721 });
1722 }
1723 }
1724 KeyResolutionStrategy::Computed { .. }
1725 | KeyResolutionStrategy::TemporalLookup { .. } => {
1726 }
1728 }
1729 }
1730
1731 ops
1732 }
1733
1734 fn get_event_type(&self, source: &SourceSpec) -> String {
1735 match source {
1736 SourceSpec::Source { type_name, .. } => type_name.clone(),
1737 }
1738 }
1739
1740 fn compile_instruction_hook_actions(&self, actions: &[HookAction]) -> Vec<OpCode> {
1741 let mut ops = Vec::new();
1742 let state_reg = 2;
1743
1744 for action in actions {
1745 match action {
1746 HookAction::SetField {
1747 target_field,
1748 source,
1749 condition,
1750 } => {
1751 let _ = condition;
1753
1754 let temp_reg = 11; let load_ops = self.compile_mapping_source(source, temp_reg);
1758 ops.extend(load_ops);
1759
1760 if let MappingSource::FromSource {
1762 transform: Some(transform_type),
1763 ..
1764 } = source
1765 {
1766 ops.push(OpCode::Transform {
1767 source: temp_reg,
1768 dest: temp_reg,
1769 transformation: transform_type.clone(),
1770 });
1771 }
1772
1773 if let Some(cond_expr) = condition {
1775 if let Some(parsed) = &cond_expr.parsed {
1776 let cond_check_ops = self.compile_condition_check(
1778 parsed,
1779 temp_reg,
1780 state_reg,
1781 target_field,
1782 );
1783 ops.extend(cond_check_ops);
1784 } else {
1785 ops.push(OpCode::SetField {
1787 object: state_reg,
1788 path: target_field.clone(),
1789 value: temp_reg,
1790 });
1791 }
1792 } else {
1793 ops.push(OpCode::SetField {
1795 object: state_reg,
1796 path: target_field.clone(),
1797 value: temp_reg,
1798 });
1799 }
1800 }
1801 HookAction::IncrementField {
1802 target_field,
1803 increment_by,
1804 condition,
1805 } => {
1806 if let Some(cond_expr) = condition {
1807 if let Some(parsed) = &cond_expr.parsed {
1808 let cond_check_ops = self.compile_conditional_increment(
1813 parsed,
1814 state_reg,
1815 target_field,
1816 *increment_by,
1817 );
1818 ops.extend(cond_check_ops);
1819 } else {
1820 ops.push(OpCode::SetFieldIncrement {
1822 object: state_reg,
1823 path: target_field.clone(),
1824 });
1825 }
1826 } else {
1827 ops.push(OpCode::SetFieldIncrement {
1829 object: state_reg,
1830 path: target_field.clone(),
1831 });
1832 }
1833 }
1834 HookAction::RegisterPdaMapping { .. } => {
1835 if let HookAction::RegisterPdaMapping {
1836 pda_field,
1837 seed_field,
1838 lookup_name,
1839 } = action
1840 {
1841 let pda_reg = 11;
1842 let seed_reg = 12;
1843
1844 ops.push(OpCode::LoadEventField {
1845 path: pda_field.clone(),
1846 dest: pda_reg,
1847 default: None,
1848 });
1849 ops.push(OpCode::LoadEventField {
1850 path: seed_field.clone(),
1851 dest: seed_reg,
1852 default: None,
1853 });
1854 ops.push(OpCode::UpdatePdaReverseLookup {
1855 state_id: self.state_id,
1856 lookup_name: lookup_name.clone(),
1857 pda_address: pda_reg,
1858 primary_key: seed_reg,
1859 });
1860 }
1861 }
1862 }
1863 }
1864
1865 ops
1866 }
1867
1868 fn compile_condition_check(
1869 &self,
1870 condition: &ParsedCondition,
1871 value_reg: Register,
1872 state_reg: Register,
1873 target_field: &str,
1874 ) -> Vec<OpCode> {
1875 match condition {
1876 ParsedCondition::Comparison {
1877 field,
1878 op,
1879 value: cond_value,
1880 } => {
1881 vec![OpCode::ConditionalSetField {
1883 object: state_reg,
1884 path: target_field.to_string(),
1885 value: value_reg,
1886 condition_field: field.clone(),
1887 condition_op: op.clone(),
1888 condition_value: cond_value.clone(),
1889 }]
1890 }
1891 ParsedCondition::Logical { .. } => {
1892 tracing::warn!("Logical conditions not yet supported in instruction hooks");
1894 vec![OpCode::SetField {
1895 object: state_reg,
1896 path: target_field.to_string(),
1897 value: value_reg,
1898 }]
1899 }
1900 }
1901 }
1902
1903 fn compile_conditional_increment(
1904 &self,
1905 condition: &ParsedCondition,
1906 state_reg: Register,
1907 target_field: &str,
1908 _increment_by: i64,
1909 ) -> Vec<OpCode> {
1910 match condition {
1911 ParsedCondition::Comparison {
1912 field,
1913 op,
1914 value: cond_value,
1915 } => {
1916 vec![OpCode::ConditionalIncrement {
1917 object: state_reg,
1918 path: target_field.to_string(),
1919 condition_field: field.clone(),
1920 condition_op: op.clone(),
1921 condition_value: cond_value.clone(),
1922 }]
1923 }
1924 ParsedCondition::Logical { .. } => {
1925 tracing::warn!("Logical conditions not yet supported in instruction hooks");
1926 vec![OpCode::SetFieldIncrement {
1927 object: state_reg,
1928 path: target_field.to_string(),
1929 }]
1930 }
1931 }
1932 }
1933}
1934
1935#[cfg(test)]
1936mod tests {
1937 use super::MultiEntityBytecode;
1938 use crate::ast::{
1939 FieldPath, HookAction, IdentitySpec, InstructionHook, KeyResolutionStrategy,
1940 LookupIndexSpec, MappingSource, PopulationStrategy, SerializableFieldMapping,
1941 SerializableHandlerSpec, SerializableStreamSpec, SourceSpec, TypedStreamSpec,
1942 };
1943 use crate::vm::VmContext;
1944 use serde_json::{json, Value};
1945 use std::collections::BTreeMap;
1946
1947 fn mapping(
1948 target_path: &str,
1949 source_path: &[&str],
1950 population: PopulationStrategy,
1951 ) -> SerializableFieldMapping {
1952 SerializableFieldMapping {
1953 target_path: target_path.to_string(),
1954 source: MappingSource::FromSource {
1955 path: FieldPath::new(source_path),
1956 default: None,
1957 transform: None,
1958 },
1959 transform: None,
1960 population,
1961 condition: None,
1962 when: None,
1963 stop: None,
1964 emit: true,
1965 }
1966 }
1967
1968 fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
1969 TypedStreamSpec::from_serializable(SerializableStreamSpec {
1970 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
1971 state_name: "PumpfunToken".to_string(),
1972 program_id: None,
1973 idl: None,
1974 identity: IdentitySpec {
1975 primary_keys: vec!["id.mint".to_string()],
1976 lookup_indexes: vec![LookupIndexSpec {
1977 field_name: "id.bonding_curve".to_string(),
1978 temporal_field: None,
1979 }],
1980 },
1981 handlers: vec![SerializableHandlerSpec {
1982 source: SourceSpec::Source {
1983 program_id: None,
1984 discriminator: None,
1985 type_name: "pump::BondingCurveState".to_string(),
1986 serialization: None,
1987 is_account: true,
1988 },
1989 key_resolution: KeyResolutionStrategy::Lookup {
1990 primary_field: FieldPath::new(&["__account_address"]),
1991 },
1992 mappings: vec![
1993 mapping(
1994 "id.bonding_curve",
1995 &["__account_address"],
1996 PopulationStrategy::SetOnce,
1997 ),
1998 mapping(
1999 "reserves.virtual_token_reserves",
2000 &["virtual_token_reserves"],
2001 PopulationStrategy::LastWrite,
2002 ),
2003 ],
2004 conditions: vec![],
2005 emit: true,
2006 }],
2007 sections: vec![],
2008 field_mappings: BTreeMap::new(),
2009 resolver_hooks: vec![],
2010 instruction_hooks: vec![InstructionHook {
2011 instruction_type: "pump::BuyIxState".to_string(),
2012 actions: vec![HookAction::RegisterPdaMapping {
2013 pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2014 seed_field: FieldPath::new(&["accounts", "mint"]),
2015 lookup_name: "default_pda_lookup".to_string(),
2016 }],
2017 lookup_by: None,
2018 }],
2019 resolver_specs: vec![],
2020 computed_fields: vec![],
2021 computed_field_specs: vec![],
2022 content_hash: None,
2023 views: vec![],
2024 })
2025 }
2026
2027 fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
2028 TypedStreamSpec::from_serializable(SerializableStreamSpec {
2029 ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2030 state_name: "OreRound".to_string(),
2031 program_id: None,
2032 idl: None,
2033 identity: IdentitySpec {
2034 primary_keys: vec!["id.round_id".to_string()],
2035 lookup_indexes: vec![LookupIndexSpec {
2036 field_name: "id.round_address".to_string(),
2037 temporal_field: None,
2038 }],
2039 },
2040 handlers: vec![SerializableHandlerSpec {
2041 source: SourceSpec::Source {
2042 program_id: None,
2043 discriminator: None,
2044 type_name: "entropy::VarState".to_string(),
2045 serialization: None,
2046 is_account: true,
2047 },
2048 key_resolution: KeyResolutionStrategy::Lookup {
2049 primary_field: FieldPath::new(&["__account_address"]),
2050 },
2051 mappings: vec![mapping(
2052 "state.expires_at",
2053 &["end_at"],
2054 PopulationStrategy::LastWrite,
2055 )],
2056 conditions: vec![],
2057 emit: true,
2058 }],
2059 sections: vec![],
2060 field_mappings: BTreeMap::new(),
2061 resolver_hooks: vec![],
2062 instruction_hooks: vec![InstructionHook {
2063 instruction_type: "ore::DeployIxState".to_string(),
2064 actions: vec![HookAction::RegisterPdaMapping {
2065 pda_field: FieldPath::new(&["accounts", "entropyVar"]),
2066 seed_field: FieldPath::new(&["accounts", "round"]),
2067 lookup_name: "default_pda_lookup".to_string(),
2068 }],
2069 lookup_by: None,
2070 }],
2071 resolver_specs: vec![],
2072 computed_fields: vec![],
2073 computed_field_specs: vec![],
2074 content_hash: None,
2075 views: vec![],
2076 })
2077 }
2078
2079 #[test]
2080 fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
2081 let bytecode = MultiEntityBytecode::from_single(
2082 "PumpfunToken".to_string(),
2083 direct_pda_to_primary_key_spec(),
2084 0,
2085 );
2086 let mut vm = VmContext::new();
2087
2088 let mutations = vm
2089 .process_event(
2090 &bytecode,
2091 json!({
2092 "__account_address": "bonding_curve_1",
2093 "__resolved_primary_key": "mint_1",
2094 "virtual_token_reserves": 42,
2095 }),
2096 "pump::BondingCurveState",
2097 None,
2098 None,
2099 )
2100 .unwrap();
2101
2102 assert_eq!(mutations.len(), 1);
2103 assert_eq!(mutations[0].key, json!("mint_1"));
2104 assert_eq!(
2105 mutations[0].patch["id"]["bonding_curve"],
2106 json!("bonding_curve_1")
2107 );
2108 assert_eq!(
2109 mutations[0].patch["reserves"]["virtual_token_reserves"],
2110 json!(42)
2111 );
2112 }
2113
2114 #[test]
2115 fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
2116 let bytecode = MultiEntityBytecode::from_single(
2117 "OreRound".to_string(),
2118 pda_to_intermediate_lookup_spec(),
2119 0,
2120 );
2121 let mut vm = VmContext::new();
2122
2123 let mutations = vm
2124 .process_event(
2125 &bytecode,
2126 json!({
2127 "__account_address": "entropy_var_1",
2128 "__resolved_primary_key": "round_address_1",
2129 "end_at": 123,
2130 }),
2131 "entropy::VarState",
2132 None,
2133 None,
2134 )
2135 .unwrap();
2136
2137 assert!(mutations.is_empty());
2138 }
2139}