Skip to main content

arete_interpreter/
compiler.rs

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    /// Abort the handler with empty mutations when the key register is null
15    /// and the event is an account-state update (not IxState / CpiEvent).
16    /// Placed immediately after key resolution so that downstream opcodes
17    /// (index updates, field mappings, resolvers, emit) never execute with
18    /// a garbage key. The null-key event is then eligible for queueing by
19    /// process_event's miss-handling logic.
20    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    /// Copy from source to dest only if dest is currently null
38    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    /// Sum a numeric value to a field (accumulator)
136    SetFieldSum {
137        object: Register,
138        path: String,
139        value: Register,
140    },
141    /// Increment a counter field by 1
142    SetFieldIncrement {
143        object: Register,
144        path: String,
145    },
146    /// Set field to minimum value
147    SetFieldMin {
148        object: Register,
149        path: String,
150        value: Register,
151    },
152    /// Set field only if a specific instruction type was seen in the same transaction.
153    /// If not seen yet, defers the operation for later completion.
154    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    /// Set field unless stopped by a specific instruction.
166    /// Stop is tracked by a per-entity stop flag.
167    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    /// Add value to unique set and update count
177    /// Maintains internal Set, field stores count
178    AddToUniqueSet {
179        state_id: u32,
180        set_name: String,
181        value: Register,
182        count_object: Register,
183        count_path: String,
184    },
185    /// Conditionally set a field based on a comparison
186    ConditionalSetField {
187        object: Register,
188        path: String,
189        value: Register,
190        condition_field: FieldPath,
191        condition_op: ComparisonOp,
192        condition_value: Value,
193    },
194    /// Conditionally increment a field based on a comparison
195    ConditionalIncrement {
196        object: Register,
197        path: String,
198        condition_field: FieldPath,
199        condition_op: ComparisonOp,
200        condition_value: Value,
201    },
202    /// Evaluate computed fields (calls external hook if provided)
203    /// computed_paths: List of paths that will be computed (for dirty tracking)
204    EvaluateComputedFields {
205        state: Register,
206        computed_paths: Vec<String>,
207    },
208    /// Queue a resolver for asynchronous enrichment
209    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    /// Update PDA reverse lookup table
224    /// Maps a PDA address to its primary key for reverse lookups
225    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    /// Optional callback for evaluating computed fields
241    /// Parameters: state, context_slot (Option<u64>), context_timestamp (i64)
242    #[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        // Store the evaluator callback if provided
388        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        // DEBUG: Collect all handler info before processing
492        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        // DEBUG: Log handler information (optional - can be removed later)
510        // Uncomment to debug handler processing:
511        // if self.entity_name == "PumpfunToken" {
512        //     eprintln!("🔍 Compiling {} handlers for {}", self.spec.handlers.len(), self.entity_name);
513        //     for info in &debug_info {
514        //         eprintln!("{}", info);
515        //     }
516        // }
517
518        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                // Merge strategy: Take ALL operations from BOTH handlers
538                // Keep setup from first, combine all mappings, keep one teardown
539
540                // Split existing handler into: setup, mappings, teardown
541                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; // 0=setup, 1=mappings, 2=teardown
545
546                for opcode in existing_opcodes.iter() {
547                    match opcode {
548                        OpCode::ReadOrInitState { .. } => {
549                            existing_setup.push(opcode.clone());
550                            section = 1; // Next opcodes are mappings
551                        }
552                        OpCode::UpdateState { .. } => {
553                            existing_teardown.push(opcode.clone());
554                            section = 2; // Next opcodes are teardown
555                        }
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                // Extract mappings from new handler (skip setup and teardown)
566                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; // Start capturing mappings
573                        }
574                        OpCode::UpdateState { .. } | OpCode::EmitMutation { .. } => {
575                            section = 2; // Stop capturing
576                        }
577                        _ if section == 1 => {
578                            new_mappings.push(opcode.clone());
579                        }
580                        _ => {} // Skip setup and teardown from new handler
581                    }
582                }
583
584                // Rebuild: setup + existing_mappings + new_mappings + teardown
585                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        // Process instruction_hooks to add SetField/IncrementField operations
598        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                // First, try to load __resolved_primary_key from resolver
610                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                // Copy to key_reg (unconditionally, may be null)
617                ops.push(OpCode::CopyRegister {
618                    source: resolved_key_reg,
619                    dest: key_reg,
620                });
621
622                // If hook has lookup_by, use it to load primary key from instruction accounts
623                if let Some(lookup_path) = &hook.lookup_by {
624                    // Load the primary key from the instruction's lookup_by field (e.g., accounts.signer)
625                    ops.push(OpCode::LoadEventField {
626                        path: lookup_path.clone(),
627                        dest: temp_reg,
628                        default: None,
629                    });
630
631                    // Apply HexEncode transformation (accounts are byte arrays)
632                    ops.push(OpCode::Transform {
633                        source: temp_reg,
634                        dest: temp_reg,
635                        transformation: Transformation::HexEncode,
636                    });
637
638                    // Use this as fallback if __resolved_primary_key was null
639                    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            // Generate opcodes for each action in the hook
662            let hook_opcodes = self.compile_instruction_hook_actions(&hook.actions);
663
664            // Insert hook opcodes before EvaluateComputedFields (if present) or UpdateState
665            // Hook actions (like whale_trade_count increment) must run before computed fields
666            // are evaluated, since computed fields may depend on the modified state
667            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                // Insert hook opcodes before EvaluateComputedFields or UpdateState
678                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        // Guard: if key resolved to null on an account-state event, abort
708        // early with empty mutations so process_event can queue the update
709        // for later reprocessing.  Without this, downstream opcodes would
710        // create a phantom entity keyed by null and produce non-empty
711        // mutations that prevent queueing.
712        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        // Index updates must come AFTER ReadOrInitState so the state table exists.
732        // ReadOrInitState lazily creates the state table via entry().or_insert_with(),
733        // but index opcodes (UpdateLookupIndex, UpdateTemporalIndex, UpdatePdaReverseLookup)
734        // use get_mut() which fails if the table doesn't exist yet.
735        // This ordering also means stale/duplicate updates (caught by ReadOrInitState's
736        // recency check) correctly skip index updates too.
737        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        // Evaluate computed fields after all mappings but before updating state
750        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                // Count doesn't need the value, just increment
978                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                // UniqueCount requires maintaining an internal set
992                // The field stores the count, but we track unique values in a hidden set
993                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                // Apply transform if specified in the source
1021                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                // AsCapture loads the whole source, applies field-level transforms, and wraps in CaptureWrapper
1110                let capture_data_reg = 22; // Temp register for capture data before wrapping
1111                let mut ops = vec![OpCode::LoadEventField {
1112                    path: FieldPath::new(&[]),
1113                    dest: capture_data_reg,
1114                    default: Some(serde_json::json!({})),
1115                }];
1116
1117                // Apply transforms to specific fields in the loaded object
1118                // IMPORTANT: Use registers that don't conflict with key_reg (20)
1119                // Using 24 and 25 to avoid conflicts with key loading (uses 18, 19, 20, 23)
1120                let field_reg = 24;
1121                let transformed_reg = 25;
1122
1123                for (field_name, transform) in field_transforms {
1124                    // Load the field from the capture_data_reg (not from event!)
1125                    // Use GetField opcode to read from a register instead of LoadEventField
1126                    ops.push(OpCode::GetField {
1127                        object: capture_data_reg,
1128                        path: field_name.clone(),
1129                        dest: field_reg,
1130                    });
1131
1132                    // Transform it
1133                    ops.push(OpCode::Transform {
1134                        source: field_reg,
1135                        dest: transformed_reg,
1136                        transformation: transform.clone(),
1137                    });
1138
1139                    // Set it back into the capture data object
1140                    ops.push(OpCode::SetField {
1141                        object: capture_data_reg,
1142                        path: field_name.clone(),
1143                        value: transformed_reg,
1144                    });
1145                }
1146
1147                // Wrap the capture data in CaptureWrapper with metadata
1148                ops.push(OpCode::CreateCapture {
1149                    dest,
1150                    capture_value: capture_data_reg,
1151                });
1152
1153                ops
1154            }
1155            MappingSource::FromContext { field } => {
1156                // Load from instruction context (timestamp, slot, signature)
1157                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        // Resolvers provide either a final key or an intermediate lookup key,
1181        // depending on the handler strategy.
1182        let resolved_key_reg = 19; // Use a temp register
1183        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        // Now do the normal key resolution
1190        match resolution {
1191            KeyResolutionStrategy::Embedded { primary_field } => {
1192                // Enhanced key resolution: check for auto-inheritance when primary_field is empty
1193                let effective_primary_field = if primary_field.segments.is_empty() {
1194                    // Try to auto-detect primary field from account schema
1195                    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                // Skip fallback key loading if effective primary_field is still empty
1205                // This happens for account types that rely solely on __resolved_primary_key
1206                // (e.g., accounts with #[resolve_key_for] resolvers)
1207                if !effective_primary_field.segments.is_empty() {
1208                    let temp_reg = 18;
1209                    let transform_reg = 23; // Register for transformed key
1210
1211                    ops.push(OpCode::LoadEventField {
1212                        path: effective_primary_field.clone(),
1213                        dest: temp_reg,
1214                        default: None,
1215                    });
1216
1217                    // Check if there's a transformation for the primary key field
1218                    // First try the current mappings, then inherited transformations
1219                    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                        // Apply transformation to the loaded key
1225                        ops.push(OpCode::Transform {
1226                            source: temp_reg,
1227                            dest: transform_reg,
1228                            transformation: transform,
1229                        });
1230                        // A concrete embedded key belongs to this entity and
1231                        // must win over a resolver result injected for another
1232                        // handler consuming the same event.
1233                        ops.push(OpCode::CopyRegister {
1234                            source: transform_reg,
1235                            dest: key_reg,
1236                        });
1237                    } else {
1238                        // No transformation, use raw value.
1239                        ops.push(OpCode::CopyRegister {
1240                            source: temp_reg,
1241                            dest: key_reg,
1242                        });
1243                    }
1244                    ops.push(OpCode::CopyRegisterIfNull {
1245                        source: resolved_key_reg,
1246                        dest: key_reg,
1247                    });
1248                } else {
1249                    // Resolver-only embedded handlers have no local key field.
1250                    ops.push(OpCode::CopyRegister {
1251                        source: resolved_key_reg,
1252                        dest: key_reg,
1253                    });
1254                }
1255            }
1256            KeyResolutionStrategy::Lookup { primary_field } => {
1257                let lookup_reg = 15;
1258                let result_reg = 17;
1259
1260                // Prefer resolver-provided key as lookup input.
1261                // When __resolved_primary_key is set (e.g. round_address from
1262                // PDA reverse lookup), use it directly — this gives a one-hop
1263                // lookup (round_address → round_id) instead of a two-hop chain
1264                // (Var address → PDA → round_address → round_id).
1265                ops.push(OpCode::CopyRegister {
1266                    source: resolved_key_reg,
1267                    dest: lookup_reg,
1268                });
1269
1270                let temp_reg = 18;
1271                ops.push(OpCode::LoadEventField {
1272                    path: primary_field.clone(),
1273                    dest: temp_reg,
1274                    default: None,
1275                });
1276                ops.push(OpCode::CopyRegisterIfNull {
1277                    source: temp_reg,
1278                    dest: lookup_reg,
1279                });
1280
1281                let index_name = self.find_lookup_index_for_lookup_field(primary_field, mappings);
1282                let effective_index_name =
1283                    index_name.unwrap_or_else(|| "default_pda_lookup".to_string());
1284
1285                ops.push(OpCode::LookupIndex {
1286                    state_id: self.state_id,
1287                    index_name: effective_index_name,
1288                    lookup_value: lookup_reg,
1289                    dest: result_reg,
1290                });
1291                ops.push(OpCode::CopyRegister {
1292                    source: result_reg,
1293                    dest: key_reg,
1294                });
1295
1296                // Most lookup-based account handlers expect resolver output to be an
1297                // intermediate lookup value (for example, PDA -> round_address -> round_id),
1298                // so a null LookupIndex result must leave the key null for queueing.
1299                // Some stacks register the PDA directly to the entity primary key
1300                // (for example, bonding_curve -> mint). In that case the resolver output
1301                // is already the final key, so preserve it only when the instruction hook's
1302                // seed field matches one of the entity primary key fields.
1303                if self.resolver_outputs_primary_key_directly(primary_field) {
1304                    ops.push(OpCode::CopyRegisterIfNull {
1305                        source: resolved_key_reg,
1306                        dest: key_reg,
1307                    });
1308                }
1309            }
1310            KeyResolutionStrategy::Computed {
1311                primary_field,
1312                compute_partition: _,
1313            } => {
1314                // Copy resolver result to key_reg (may be null)
1315                ops.push(OpCode::CopyRegister {
1316                    source: resolved_key_reg,
1317                    dest: key_reg,
1318                });
1319                let temp_reg = 18;
1320                ops.push(OpCode::LoadEventField {
1321                    path: primary_field.clone(),
1322                    dest: temp_reg,
1323                    default: None,
1324                });
1325                ops.push(OpCode::CopyRegisterIfNull {
1326                    source: temp_reg,
1327                    dest: key_reg,
1328                });
1329            }
1330            KeyResolutionStrategy::TemporalLookup {
1331                lookup_field,
1332                timestamp_field,
1333                index_name,
1334            } => {
1335                // Copy resolver result to key_reg (may be null)
1336                ops.push(OpCode::CopyRegister {
1337                    source: resolved_key_reg,
1338                    dest: key_reg,
1339                });
1340                let lookup_reg = 15;
1341                let timestamp_reg = 16;
1342                let result_reg = 17;
1343
1344                ops.push(OpCode::LoadEventField {
1345                    path: lookup_field.clone(),
1346                    dest: lookup_reg,
1347                    default: None,
1348                });
1349
1350                ops.push(OpCode::LoadEventField {
1351                    path: timestamp_field.clone(),
1352                    dest: timestamp_reg,
1353                    default: None,
1354                });
1355
1356                ops.push(OpCode::LookupTemporalIndex {
1357                    state_id: self.state_id,
1358                    index_name: index_name.clone(),
1359                    lookup_value: lookup_reg,
1360                    timestamp: timestamp_reg,
1361                    dest: result_reg,
1362                });
1363
1364                ops.push(OpCode::CopyRegisterIfNull {
1365                    source: result_reg,
1366                    dest: key_reg,
1367                });
1368            }
1369        }
1370
1371        ops
1372    }
1373
1374    fn find_primary_key_transformation(
1375        &self,
1376        mappings: &[TypedFieldMapping<S>],
1377    ) -> Option<Transformation> {
1378        // Find the first primary key in the identity spec
1379        let primary_key = self.spec.identity.primary_keys.first()?;
1380        let primary_field_name = self.extract_primary_field_name(primary_key)?;
1381
1382        // Look for a mapping that targets this primary key
1383        for mapping in mappings {
1384            // Check if this mapping targets the primary key field
1385            if mapping.target_path == *primary_key
1386                || mapping.target_path.ends_with(&format!(".{}", primary_key))
1387            {
1388                // Check mapping-level transform first
1389                if let Some(transform) = &mapping.transform {
1390                    return Some(transform.clone());
1391                }
1392
1393                // Then check source-level transform
1394                if let MappingSource::FromSource {
1395                    transform: Some(transform),
1396                    ..
1397                } = &mapping.source
1398                {
1399                    return Some(transform.clone());
1400                }
1401            }
1402        }
1403
1404        // If no explicit primary key mapping found, check AsCapture field transforms
1405        for mapping in mappings {
1406            if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1407                if let Some(transform) = field_transforms.get(&primary_field_name) {
1408                    return Some(transform.clone());
1409                }
1410            }
1411        }
1412
1413        None
1414    }
1415
1416    /// Look for primary key mappings in other handlers of the same entity
1417    /// This enables cross-handler inheritance of key transformations
1418    pub fn find_inherited_primary_key_transformation(&self) -> Option<Transformation> {
1419        let primary_key = self.spec.identity.primary_keys.first()?;
1420
1421        // Extract the field name from the primary key path (e.g., "id.authority" -> "authority")
1422        let primary_field_name = self.extract_primary_field_name(primary_key)?;
1423
1424        // Search through all handlers in the spec for primary key mappings
1425        for handler in &self.spec.handlers {
1426            for mapping in &handler.mappings {
1427                // Look for mappings targeting the primary key
1428                if mapping.target_path == *primary_key
1429                    || mapping.target_path.ends_with(&format!(".{}", primary_key))
1430                {
1431                    // Check if this mapping comes from a field matching the primary key name
1432                    if let MappingSource::FromSource {
1433                        path, transform, ..
1434                    } = &mapping.source
1435                    {
1436                        if path.segments.last() == Some(&primary_field_name) {
1437                            // Return mapping-level transform first, then source-level transform
1438                            return mapping.transform.clone().or_else(|| transform.clone());
1439                        }
1440                    }
1441                }
1442
1443                // Also check AsCapture field transforms for the primary field
1444                if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1445                    if let Some(transform) = field_transforms.get(&primary_field_name) {
1446                        return Some(transform.clone());
1447                    }
1448                }
1449            }
1450        }
1451
1452        None
1453    }
1454
1455    /// Extract the field name from a primary key path (e.g., "id.authority" -> "authority")
1456    fn extract_primary_field_name(&self, primary_key: &str) -> Option<String> {
1457        // Split by '.' and take the last segment
1458        primary_key.split('.').next_back().map(|s| s.to_string())
1459    }
1460
1461    /// Auto-detect primary field from account schema when no explicit mapping exists
1462    /// This looks for account types that have an 'authority' field and tries to use it
1463    pub fn auto_detect_primary_field(
1464        &self,
1465        current_mappings: &[TypedFieldMapping<S>],
1466    ) -> Option<FieldPath> {
1467        let primary_key = self.spec.identity.primary_keys.first()?;
1468
1469        // Extract the field name from the primary key (e.g., "id.authority" -> "authority")
1470        let primary_field_name = self.extract_primary_field_name(primary_key)?;
1471
1472        // Check if current handler can access the primary field
1473        if self.current_account_has_primary_field(&primary_field_name, current_mappings) {
1474            return Some(FieldPath::new(&[&primary_field_name]));
1475        }
1476
1477        None
1478    }
1479
1480    /// Check if the current account type has the primary field
1481    /// This is determined by looking at the mappings to see what fields are available
1482    fn current_account_has_primary_field(
1483        &self,
1484        field_name: &str,
1485        mappings: &[TypedFieldMapping<S>],
1486    ) -> bool {
1487        // Look through the mappings to see if any reference the primary field
1488        for mapping in mappings {
1489            if let MappingSource::FromSource { path, .. } = &mapping.source {
1490                // Check if this mapping sources from the primary field
1491                if path.segments.last() == Some(&field_name.to_string()) {
1492                    return true;
1493                }
1494            }
1495        }
1496
1497        false
1498    }
1499
1500    /// Check if handler has access to a specific field in its source account
1501    #[allow(dead_code)]
1502    fn handler_has_field(&self, field_name: &str, mappings: &[TypedFieldMapping<S>]) -> bool {
1503        for mapping in mappings {
1504            if let MappingSource::FromSource { path, .. } = &mapping.source {
1505                if path.segments.last() == Some(&field_name.to_string()) {
1506                    return true;
1507                }
1508            }
1509        }
1510        false
1511    }
1512
1513    /// Check if field exists by looking at mappings (IDL-agnostic approach)
1514    /// This avoids hardcoding account schemas and uses actual mapping evidence
1515    #[allow(dead_code)]
1516    fn field_exists_in_mappings(
1517        &self,
1518        field_name: &str,
1519        mappings: &[TypedFieldMapping<S>],
1520    ) -> bool {
1521        // Look through current mappings to see if the field is referenced
1522        for mapping in mappings {
1523            if let MappingSource::FromSource { path, .. } = &mapping.source {
1524                if path.segments.last() == Some(&field_name.to_string()) {
1525                    return true;
1526                }
1527            }
1528            // Also check AsCapture field transforms
1529            if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1530                if field_transforms.contains_key(field_name) {
1531                    return true;
1532                }
1533            }
1534        }
1535        false
1536    }
1537
1538    fn find_lookup_index_for_field(&self, field_path: &FieldPath) -> Option<String> {
1539        if field_path.segments.is_empty() {
1540            return None;
1541        }
1542
1543        let lookup_field_name = field_path.segments.last().unwrap();
1544
1545        for lookup_index in &self.spec.identity.lookup_indexes {
1546            let index_field_name = lookup_index
1547                .field_name
1548                .split('.')
1549                .next_back()
1550                .unwrap_or(&lookup_index.field_name);
1551            let matches_directly = index_field_name == lookup_field_name;
1552            // An index field named `foo_address` is treated as an alias for the
1553            // bare field `foo` (for example, `mint_address` resolves handlers
1554            // keyed on `mint`). This is intentionally one-way: bare `foo` does
1555            // not imply a `foo_address` lookup index. The macro crate mirrors
1556            // this convention via `lookup_index_leafs` in
1557            // `arete-macros/src/validation/mod.rs`.
1558            let matches_address_alias = index_field_name
1559                .strip_suffix("_address")
1560                .map(|base| base == lookup_field_name)
1561                .unwrap_or(false);
1562
1563            if matches_directly || matches_address_alias {
1564                return Some(format!("{}_lookup_index", index_field_name));
1565            }
1566        }
1567
1568        None
1569    }
1570
1571    /// Find lookup index for a Lookup key resolution by checking if there's a mapping
1572    /// from the primary_field to a lookup index field.
1573    fn find_lookup_index_for_lookup_field(
1574        &self,
1575        primary_field: &FieldPath,
1576        mappings: &[TypedFieldMapping<S>],
1577    ) -> Option<String> {
1578        // Build the primary field path string
1579        let primary_path = primary_field.segments.join(".");
1580
1581        // Check if there's a mapping from this primary field to a lookup index field
1582        for mapping in mappings {
1583            // Check if the mapping source path matches the primary field
1584            if let MappingSource::FromSource { path, .. } = &mapping.source {
1585                let source_path = path.segments.join(".");
1586                if source_path == primary_path {
1587                    // Check if the target is a lookup index field
1588                    for lookup_index in &self.spec.identity.lookup_indexes {
1589                        if mapping.target_path == lookup_index.field_name {
1590                            let index_field_name = lookup_index
1591                                .field_name
1592                                .split('.')
1593                                .next_back()
1594                                .unwrap_or(&lookup_index.field_name);
1595                            return Some(format!("{}_lookup_index", index_field_name));
1596                        }
1597                    }
1598                }
1599            }
1600        }
1601
1602        // Fall back to direct field name matching
1603        self.find_lookup_index_for_field(primary_field)
1604    }
1605
1606    /// Find the source path for a lookup index field by looking at mappings.
1607    /// For example, if target_path is "id.round_address" and the mapping is
1608    /// `id.round_address <- __account_address`, this returns ["__account_address"].
1609    fn find_source_path_for_lookup_index(
1610        &self,
1611        mappings: &[TypedFieldMapping<S>],
1612        lookup_field_name: &str,
1613    ) -> Option<Vec<String>> {
1614        for mapping in mappings {
1615            if mapping.target_path == lookup_field_name {
1616                if let MappingSource::FromSource { path, .. } = &mapping.source {
1617                    return Some(path.segments.clone());
1618                }
1619            }
1620        }
1621        None
1622    }
1623
1624    fn compile_temporal_index_update(
1625        &self,
1626        resolution: &KeyResolutionStrategy,
1627        key_reg: Register,
1628        mappings: &[TypedFieldMapping<S>],
1629    ) -> Vec<OpCode> {
1630        let mut ops = Vec::new();
1631
1632        for lookup_index in &self.spec.identity.lookup_indexes {
1633            let lookup_reg = 17;
1634            let source_field = lookup_index
1635                .field_name
1636                .split('.')
1637                .next_back()
1638                .unwrap_or(&lookup_index.field_name);
1639
1640            match resolution {
1641                KeyResolutionStrategy::Embedded { primary_field: _ } => {
1642                    // For Embedded handlers, find the mapping that targets this lookup index field
1643                    // and use its source path to load the lookup value
1644                    let source_path_opt =
1645                        self.find_source_path_for_lookup_index(mappings, &lookup_index.field_name);
1646
1647                    let load_path = if let Some(ref path) = source_path_opt {
1648                        FieldPath::new(&path.iter().map(|s| s.as_str()).collect::<Vec<_>>())
1649                    } else {
1650                        // Fallback to source_field if no mapping found
1651                        FieldPath::new(&[source_field])
1652                    };
1653
1654                    ops.push(OpCode::LoadEventField {
1655                        path: load_path,
1656                        dest: lookup_reg,
1657                        default: None,
1658                    });
1659
1660                    if let Some(temporal_field_name) = &lookup_index.temporal_field {
1661                        let timestamp_reg = 18;
1662
1663                        ops.push(OpCode::LoadEventField {
1664                            path: FieldPath::new(&[temporal_field_name]),
1665                            dest: timestamp_reg,
1666                            default: None,
1667                        });
1668
1669                        let index_name = format!("{}_temporal_index", source_field);
1670                        ops.push(OpCode::UpdateTemporalIndex {
1671                            state_id: self.state_id,
1672                            index_name,
1673                            lookup_value: lookup_reg,
1674                            primary_key: key_reg,
1675                            timestamp: timestamp_reg,
1676                        });
1677
1678                        let simple_index_name = format!("{}_lookup_index", source_field);
1679                        ops.push(OpCode::UpdateLookupIndex {
1680                            state_id: self.state_id,
1681                            index_name: simple_index_name,
1682                            lookup_value: lookup_reg,
1683                            primary_key: key_reg,
1684                        });
1685                    } else {
1686                        let index_name = format!("{}_lookup_index", source_field);
1687                        ops.push(OpCode::UpdateLookupIndex {
1688                            state_id: self.state_id,
1689                            index_name,
1690                            lookup_value: lookup_reg,
1691                            primary_key: key_reg,
1692                        });
1693                    }
1694
1695                    // Also update PDA reverse lookup table if there's a resolver configured for this entity
1696                    // This allows instruction handlers to look up the primary key from PDA addresses
1697                    // Only do this when the source path is different (e.g., __account_address -> id.round_address)
1698                    if source_path_opt.is_some() {
1699                        ops.push(OpCode::UpdatePdaReverseLookup {
1700                            state_id: self.state_id,
1701                            lookup_name: "default_pda_lookup".to_string(),
1702                            pda_address: lookup_reg,
1703                            primary_key: key_reg,
1704                        });
1705                    }
1706                }
1707                KeyResolutionStrategy::Lookup { primary_field } => {
1708                    // For Lookup handlers, check if there's a mapping that targets this lookup index field
1709                    // If so, the lookup value is the same as the primary_field used for key resolution
1710                    let has_mapping_to_lookup_field = mappings
1711                        .iter()
1712                        .any(|m| m.target_path == lookup_index.field_name);
1713
1714                    if has_mapping_to_lookup_field {
1715                        // Load the lookup value from the event using the primary_field path
1716                        // (this is the same value used for key resolution)
1717                        let path_segments: Vec<&str> =
1718                            primary_field.segments.iter().map(|s| s.as_str()).collect();
1719                        ops.push(OpCode::LoadEventField {
1720                            path: FieldPath::new(&path_segments),
1721                            dest: lookup_reg,
1722                            default: None,
1723                        });
1724
1725                        let index_name = format!("{}_lookup_index", source_field);
1726                        ops.push(OpCode::UpdateLookupIndex {
1727                            state_id: self.state_id,
1728                            index_name,
1729                            lookup_value: lookup_reg,
1730                            primary_key: key_reg,
1731                        });
1732                    }
1733                }
1734                KeyResolutionStrategy::Computed { .. }
1735                | KeyResolutionStrategy::TemporalLookup { .. } => {
1736                    // Computed and TemporalLookup handlers don't populate lookup indexes
1737                }
1738            }
1739        }
1740
1741        ops
1742    }
1743
1744    fn get_event_type(&self, source: &SourceSpec) -> String {
1745        match source {
1746            SourceSpec::Source { type_name, .. } => type_name.clone(),
1747        }
1748    }
1749
1750    fn compile_instruction_hook_actions(&self, actions: &[HookAction]) -> Vec<OpCode> {
1751        let mut ops = Vec::new();
1752        let state_reg = 2;
1753
1754        for action in actions {
1755            match action {
1756                HookAction::SetField {
1757                    target_field,
1758                    source,
1759                    condition,
1760                } => {
1761                    // Check if there's a condition - evaluation handled in VM
1762                    let _ = condition;
1763
1764                    let temp_reg = 11; // Use register 11 for hook values
1765
1766                    // Load the source value
1767                    let load_ops = self.compile_mapping_source(source, temp_reg);
1768                    ops.extend(load_ops);
1769
1770                    // Apply transformation if specified in source
1771                    if let MappingSource::FromSource {
1772                        transform: Some(transform_type),
1773                        ..
1774                    } = source
1775                    {
1776                        ops.push(OpCode::Transform {
1777                            source: temp_reg,
1778                            dest: temp_reg,
1779                            transformation: transform_type.clone(),
1780                        });
1781                    }
1782
1783                    // Conditionally set the field based on parsed condition
1784                    if let Some(cond_expr) = condition {
1785                        if let Some(parsed) = &cond_expr.parsed {
1786                            // Generate condition check opcodes
1787                            let cond_check_ops = self.compile_condition_check(
1788                                parsed,
1789                                temp_reg,
1790                                state_reg,
1791                                target_field,
1792                            );
1793                            ops.extend(cond_check_ops);
1794                        } else {
1795                            // No parsed condition, set unconditionally
1796                            ops.push(OpCode::SetField {
1797                                object: state_reg,
1798                                path: target_field.clone(),
1799                                value: temp_reg,
1800                            });
1801                        }
1802                    } else {
1803                        // No condition, set unconditionally
1804                        ops.push(OpCode::SetField {
1805                            object: state_reg,
1806                            path: target_field.clone(),
1807                            value: temp_reg,
1808                        });
1809                    }
1810                }
1811                HookAction::IncrementField {
1812                    target_field,
1813                    increment_by,
1814                    condition,
1815                } => {
1816                    if let Some(cond_expr) = condition {
1817                        if let Some(parsed) = &cond_expr.parsed {
1818                            // For increment with condition, we need to:
1819                            // 1. Load the condition field
1820                            // 2. Check the condition
1821                            // 3. Conditionally increment
1822                            let cond_check_ops = self.compile_conditional_increment(
1823                                parsed,
1824                                state_reg,
1825                                target_field,
1826                                *increment_by,
1827                            );
1828                            ops.extend(cond_check_ops);
1829                        } else {
1830                            // No parsed condition, increment unconditionally
1831                            ops.push(OpCode::SetFieldIncrement {
1832                                object: state_reg,
1833                                path: target_field.clone(),
1834                            });
1835                        }
1836                    } else {
1837                        // No condition, increment unconditionally
1838                        ops.push(OpCode::SetFieldIncrement {
1839                            object: state_reg,
1840                            path: target_field.clone(),
1841                        });
1842                    }
1843                }
1844                HookAction::RegisterPdaMapping { .. } => {
1845                    if let HookAction::RegisterPdaMapping {
1846                        pda_field,
1847                        seed_field,
1848                        lookup_name,
1849                    } = action
1850                    {
1851                        let pda_reg = 11;
1852                        let seed_reg = 12;
1853
1854                        ops.push(OpCode::LoadEventField {
1855                            path: pda_field.clone(),
1856                            dest: pda_reg,
1857                            default: None,
1858                        });
1859                        ops.push(OpCode::LoadEventField {
1860                            path: seed_field.clone(),
1861                            dest: seed_reg,
1862                            default: None,
1863                        });
1864                        ops.push(OpCode::UpdatePdaReverseLookup {
1865                            state_id: self.state_id,
1866                            lookup_name: lookup_name.clone(),
1867                            pda_address: pda_reg,
1868                            primary_key: seed_reg,
1869                        });
1870                    }
1871                }
1872            }
1873        }
1874
1875        ops
1876    }
1877
1878    fn compile_condition_check(
1879        &self,
1880        condition: &ParsedCondition,
1881        value_reg: Register,
1882        state_reg: Register,
1883        target_field: &str,
1884    ) -> Vec<OpCode> {
1885        match condition {
1886            ParsedCondition::Comparison {
1887                field,
1888                op,
1889                value: cond_value,
1890            } => {
1891                // Generate ConditionalSetField opcode
1892                vec![OpCode::ConditionalSetField {
1893                    object: state_reg,
1894                    path: target_field.to_string(),
1895                    value: value_reg,
1896                    condition_field: field.clone(),
1897                    condition_op: op.clone(),
1898                    condition_value: cond_value.clone(),
1899                }]
1900            }
1901            ParsedCondition::Logical { .. } => {
1902                // Logical conditions not yet supported, fall back to unconditional
1903                tracing::warn!("Logical conditions not yet supported in instruction hooks");
1904                vec![OpCode::SetField {
1905                    object: state_reg,
1906                    path: target_field.to_string(),
1907                    value: value_reg,
1908                }]
1909            }
1910        }
1911    }
1912
1913    fn compile_conditional_increment(
1914        &self,
1915        condition: &ParsedCondition,
1916        state_reg: Register,
1917        target_field: &str,
1918        _increment_by: i64,
1919    ) -> Vec<OpCode> {
1920        match condition {
1921            ParsedCondition::Comparison {
1922                field,
1923                op,
1924                value: cond_value,
1925            } => {
1926                vec![OpCode::ConditionalIncrement {
1927                    object: state_reg,
1928                    path: target_field.to_string(),
1929                    condition_field: field.clone(),
1930                    condition_op: op.clone(),
1931                    condition_value: cond_value.clone(),
1932                }]
1933            }
1934            ParsedCondition::Logical { .. } => {
1935                tracing::warn!("Logical conditions not yet supported in instruction hooks");
1936                vec![OpCode::SetFieldIncrement {
1937                    object: state_reg,
1938                    path: target_field.to_string(),
1939                }]
1940            }
1941        }
1942    }
1943}
1944
1945#[cfg(test)]
1946mod tests {
1947    use super::MultiEntityBytecode;
1948    use crate::ast::{
1949        FieldPath, HookAction, IdentitySpec, InstructionHook, KeyResolutionStrategy,
1950        LookupIndexSpec, MappingSource, PopulationStrategy, SerializableFieldMapping,
1951        SerializableHandlerSpec, SerializableStreamSpec, SourceSpec, TypedStreamSpec,
1952    };
1953    use crate::vm::VmContext;
1954    use serde_json::{json, Value};
1955    use std::collections::BTreeMap;
1956
1957    fn mapping(
1958        target_path: &str,
1959        source_path: &[&str],
1960        population: PopulationStrategy,
1961    ) -> SerializableFieldMapping {
1962        SerializableFieldMapping {
1963            target_path: target_path.to_string(),
1964            source: MappingSource::FromSource {
1965                path: FieldPath::new(source_path),
1966                default: None,
1967                transform: None,
1968            },
1969            transform: None,
1970            population,
1971            condition: None,
1972            when: None,
1973            stop: None,
1974            emit: true,
1975        }
1976    }
1977
1978    fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
1979        TypedStreamSpec::from_serializable(SerializableStreamSpec {
1980            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
1981            state_name: "PumpfunToken".to_string(),
1982            program_id: None,
1983            idl: None,
1984            identity: IdentitySpec {
1985                primary_keys: vec!["id.mint".to_string()],
1986                lookup_indexes: vec![LookupIndexSpec {
1987                    field_name: "id.bonding_curve".to_string(),
1988                    temporal_field: None,
1989                }],
1990            },
1991            handlers: vec![SerializableHandlerSpec {
1992                source: SourceSpec::Source {
1993                    program_id: None,
1994                    discriminator: None,
1995                    type_name: "pump::BondingCurveState".to_string(),
1996                    serialization: None,
1997                    is_account: true,
1998                },
1999                key_resolution: KeyResolutionStrategy::Lookup {
2000                    primary_field: FieldPath::new(&["__account_address"]),
2001                },
2002                mappings: vec![
2003                    mapping(
2004                        "id.bonding_curve",
2005                        &["__account_address"],
2006                        PopulationStrategy::SetOnce,
2007                    ),
2008                    mapping(
2009                        "reserves.virtual_token_reserves",
2010                        &["virtual_token_reserves"],
2011                        PopulationStrategy::LastWrite,
2012                    ),
2013                ],
2014                conditions: vec![],
2015                emit: true,
2016            }],
2017            sections: vec![],
2018            field_mappings: BTreeMap::new(),
2019            resolver_hooks: vec![],
2020            instruction_hooks: vec![InstructionHook {
2021                instruction_type: "pump::BuyIxState".to_string(),
2022                actions: vec![HookAction::RegisterPdaMapping {
2023                    pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2024                    seed_field: FieldPath::new(&["accounts", "mint"]),
2025                    lookup_name: "default_pda_lookup".to_string(),
2026                }],
2027                lookup_by: None,
2028            }],
2029            resolver_specs: vec![],
2030            computed_fields: vec![],
2031            computed_field_specs: vec![],
2032            content_hash: None,
2033            views: vec![],
2034        })
2035    }
2036
2037    fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
2038        TypedStreamSpec::from_serializable(SerializableStreamSpec {
2039            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2040            state_name: "OreRound".to_string(),
2041            program_id: None,
2042            idl: None,
2043            identity: IdentitySpec {
2044                primary_keys: vec!["id.round_id".to_string()],
2045                lookup_indexes: vec![LookupIndexSpec {
2046                    field_name: "id.round_address".to_string(),
2047                    temporal_field: None,
2048                }],
2049            },
2050            handlers: vec![SerializableHandlerSpec {
2051                source: SourceSpec::Source {
2052                    program_id: None,
2053                    discriminator: None,
2054                    type_name: "entropy::VarState".to_string(),
2055                    serialization: None,
2056                    is_account: true,
2057                },
2058                key_resolution: KeyResolutionStrategy::Lookup {
2059                    primary_field: FieldPath::new(&["__account_address"]),
2060                },
2061                mappings: vec![mapping(
2062                    "state.expires_at",
2063                    &["end_at"],
2064                    PopulationStrategy::LastWrite,
2065                )],
2066                conditions: vec![],
2067                emit: true,
2068            }],
2069            sections: vec![],
2070            field_mappings: BTreeMap::new(),
2071            resolver_hooks: vec![],
2072            instruction_hooks: vec![InstructionHook {
2073                instruction_type: "ore::DeployIxState".to_string(),
2074                actions: vec![HookAction::RegisterPdaMapping {
2075                    pda_field: FieldPath::new(&["accounts", "entropyVar"]),
2076                    seed_field: FieldPath::new(&["accounts", "round"]),
2077                    lookup_name: "default_pda_lookup".to_string(),
2078                }],
2079                lookup_by: None,
2080            }],
2081            resolver_specs: vec![],
2082            computed_fields: vec![],
2083            computed_field_specs: vec![],
2084            content_hash: None,
2085            views: vec![],
2086        })
2087    }
2088
2089    fn embedded_account_key_spec() -> TypedStreamSpec<Value> {
2090        TypedStreamSpec::from_serializable(SerializableStreamSpec {
2091            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2092            state_name: "OreTreasury".to_string(),
2093            program_id: None,
2094            idl: None,
2095            identity: IdentitySpec {
2096                primary_keys: vec!["id.address".to_string()],
2097                lookup_indexes: vec![],
2098            },
2099            handlers: vec![SerializableHandlerSpec {
2100                source: SourceSpec::Source {
2101                    program_id: None,
2102                    discriminator: None,
2103                    type_name: "ore::TreasuryState".to_string(),
2104                    serialization: None,
2105                    is_account: true,
2106                },
2107                key_resolution: KeyResolutionStrategy::Embedded {
2108                    primary_field: FieldPath::new(&["__account_address"]),
2109                },
2110                mappings: vec![mapping(
2111                    "id.address",
2112                    &["__account_address"],
2113                    PopulationStrategy::SetOnce,
2114                )],
2115                conditions: vec![],
2116                emit: true,
2117            }],
2118            sections: vec![],
2119            field_mappings: BTreeMap::new(),
2120            resolver_hooks: vec![],
2121            instruction_hooks: vec![],
2122            resolver_specs: vec![],
2123            computed_fields: vec![],
2124            computed_field_specs: vec![],
2125            content_hash: None,
2126            views: vec![],
2127        })
2128    }
2129
2130    #[test]
2131    fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
2132        let bytecode = MultiEntityBytecode::from_single(
2133            "PumpfunToken".to_string(),
2134            direct_pda_to_primary_key_spec(),
2135            0,
2136        );
2137        let mut vm = VmContext::new();
2138
2139        let mutations = vm
2140            .process_event(
2141                &bytecode,
2142                json!({
2143                    "__account_address": "bonding_curve_1",
2144                    "__resolved_primary_key": "mint_1",
2145                    "virtual_token_reserves": 42,
2146                }),
2147                "pump::BondingCurveState",
2148                None,
2149                None,
2150            )
2151            .unwrap();
2152
2153        assert_eq!(mutations.len(), 1);
2154        assert_eq!(mutations[0].key, json!("mint_1"));
2155        assert_eq!(
2156            mutations[0].patch["id"]["bonding_curve"],
2157            json!("bonding_curve_1")
2158        );
2159        assert_eq!(
2160            mutations[0].patch["reserves"]["virtual_token_reserves"],
2161            json!(42)
2162        );
2163    }
2164
2165    #[test]
2166    fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
2167        let bytecode = MultiEntityBytecode::from_single(
2168            "OreRound".to_string(),
2169            pda_to_intermediate_lookup_spec(),
2170            0,
2171        );
2172        let mut vm = VmContext::new();
2173
2174        let mutations = vm
2175            .process_event(
2176                &bytecode,
2177                json!({
2178                    "__account_address": "entropy_var_1",
2179                    "__resolved_primary_key": "round_address_1",
2180                    "end_at": 123,
2181                }),
2182                "entropy::VarState",
2183                None,
2184                None,
2185            )
2186            .unwrap();
2187
2188        assert!(mutations.is_empty());
2189    }
2190
2191    #[test]
2192    fn embedded_account_key_wins_over_an_unrelated_resolved_key() {
2193        let bytecode = MultiEntityBytecode::from_single(
2194            "OreTreasury".to_string(),
2195            embedded_account_key_spec(),
2196            0,
2197        );
2198        let mut vm = VmContext::new();
2199
2200        let mutations = vm
2201            .process_event(
2202                &bytecode,
2203                json!({
2204                    "__account_address": "treasury_pda",
2205                    "__resolved_primary_key": "round_address",
2206                }),
2207                "ore::TreasuryState",
2208                None,
2209                None,
2210            )
2211            .unwrap();
2212
2213        assert_eq!(mutations.len(), 1);
2214        assert_eq!(mutations[0].key, json!("treasury_pda"));
2215        assert_eq!(mutations[0].patch["id"]["address"], json!("treasury_pda"));
2216    }
2217}