Skip to main content

arete_interpreter/
compiler.rs

1use crate::ast::*;
2use serde::Serialize;
3use serde_json::Value;
4use std::collections::{HashMap, HashSet};
5use tracing;
6
7pub type Register = usize;
8
9fn stop_field_path(target_path: &str) -> String {
10    format!("__stop:{}", target_path)
11}
12
13/// One independently keyed handler program for an event type.
14struct HandlerSegment {
15    /// Canonical encoding of the key-loading opcodes of the handlers merged
16    /// into this segment. `None` for segments created from an instruction hook
17    /// alone.
18    key_identity: Option<Value>,
19    /// The event field this segment routes by, used to attach instruction
20    /// hooks to the segment that shares their `lookup_by`.
21    route_field: Option<FieldPath>,
22    ops: Vec<OpCode>,
23}
24
25/// The event field a key resolution strategy routes by, if any.
26fn route_field(resolution: &KeyResolutionStrategy) -> Option<FieldPath> {
27    let field = match resolution {
28        KeyResolutionStrategy::Embedded { primary_field }
29        | KeyResolutionStrategy::Lookup { primary_field }
30        | KeyResolutionStrategy::Computed { primary_field, .. } => primary_field,
31        KeyResolutionStrategy::TemporalLookup { lookup_field, .. } => lookup_field,
32    };
33    (!field.segments.is_empty()).then(|| field.clone())
34}
35
36/// Merge the mapping opcodes of `new` into `existing`, which resolve the same
37/// key: keep the setup (key loading and state read) of `existing`, append the
38/// mappings of both, and keep one teardown (state write and mutation).
39fn merge_handler_opcodes(existing: &mut Vec<OpCode>, new: &[OpCode]) {
40    // Split existing handler into: setup, mappings, teardown
41    let mut existing_setup = Vec::new();
42    let mut existing_mappings = Vec::new();
43    let mut existing_teardown = Vec::new();
44    let mut section = 0; // 0=setup, 1=mappings, 2=teardown
45
46    for opcode in existing.iter() {
47        match opcode {
48            OpCode::ReadOrInitState { .. } => {
49                existing_setup.push(opcode.clone());
50                section = 1; // Next opcodes are mappings
51            }
52            OpCode::UpdateState { .. } => {
53                existing_teardown.push(opcode.clone());
54                section = 2; // Next opcodes are teardown
55            }
56            OpCode::EmitMutation { .. } => {
57                existing_teardown.push(opcode.clone());
58            }
59            _ if section == 0 => existing_setup.push(opcode.clone()),
60            _ if section == 1 => existing_mappings.push(opcode.clone()),
61            _ => existing_teardown.push(opcode.clone()),
62        }
63    }
64
65    // Extract mappings from new handler (skip setup and teardown)
66    let mut new_mappings = Vec::new();
67    section = 0;
68
69    for opcode in new.iter() {
70        match opcode {
71            OpCode::ReadOrInitState { .. } => {
72                section = 1; // Start capturing mappings
73            }
74            OpCode::UpdateState { .. } | OpCode::EmitMutation { .. } => {
75                section = 2; // Stop capturing
76            }
77            _ if section == 1 => {
78                new_mappings.push(opcode.clone());
79            }
80            _ => {} // Skip setup and teardown from new handler
81        }
82    }
83
84    // Rebuild: setup + existing_mappings + new_mappings + teardown
85    let mut merged = Vec::new();
86    merged.extend(existing_setup);
87    merged.extend(existing_mappings);
88    merged.extend(new_mappings);
89    merged.extend(existing_teardown);
90
91    *existing = merged;
92}
93
94#[derive(Debug, Clone, Serialize)]
95pub enum OpCode {
96    /// Abort the handler with empty mutations when the key register is null
97    /// and the event is an account-state update (not IxState / CpiEvent).
98    /// Placed immediately after key resolution so that downstream opcodes
99    /// (index updates, field mappings, resolvers, emit) never execute with
100    /// a garbage key. The null-key event is then eligible for queueing by
101    /// process_event's miss-handling logic.
102    AbortIfNullKey {
103        key: Register,
104        is_account_event: bool,
105    },
106    LoadEventField {
107        path: FieldPath,
108        dest: Register,
109        default: Option<Value>,
110    },
111    LoadConstant {
112        value: Value,
113        dest: Register,
114    },
115    CopyRegister {
116        source: Register,
117        dest: Register,
118    },
119    /// Copy from source to dest only if dest is currently null
120    CopyRegisterIfNull {
121        source: Register,
122        dest: Register,
123    },
124    GetEventType {
125        dest: Register,
126    },
127    CreateObject {
128        dest: Register,
129    },
130    SetField {
131        object: Register,
132        path: String,
133        value: Register,
134    },
135    SetFields {
136        object: Register,
137        fields: Vec<(String, Register)>,
138    },
139    GetField {
140        object: Register,
141        path: String,
142        dest: Register,
143    },
144    ReadOrInitState {
145        state_id: u32,
146        key: Register,
147        default: Value,
148        dest: Register,
149    },
150    UpdateState {
151        state_id: u32,
152        key: Register,
153        value: Register,
154    },
155    AppendToArray {
156        object: Register,
157        path: String,
158        value: Register,
159    },
160    /// Load the Unix-seconds timestamp of the update being processed (its
161    /// `UpdateContext` timestamp, else the wall clock). It is the update's
162    /// time, not the processing time, so a replay reproduces it.
163    GetCurrentTimestamp {
164        dest: Register,
165    },
166    /// Wrap an event as `{timestamp, data, slot?, signature?, event_index?,
167    /// ix_path?}`, stamped with the update's timestamp.
168    CreateEvent {
169        dest: Register,
170        event_value: Register,
171    },
172    /// Wrap a captured account as `{timestamp, account_address, data, slot?,
173    /// signature?}`, stamped with the update's timestamp.
174    CreateCapture {
175        dest: Register,
176        capture_value: Register,
177    },
178    Transform {
179        source: Register,
180        dest: Register,
181        transformation: Transformation,
182    },
183    EmitMutation {
184        entity_name: String,
185        key: Register,
186        state: Register,
187    },
188    SetFieldIfNull {
189        object: Register,
190        path: String,
191        value: Register,
192    },
193    SetFieldMax {
194        object: Register,
195        path: String,
196        value: Register,
197    },
198    UpdateTemporalIndex {
199        state_id: u32,
200        index_name: String,
201        lookup_value: Register,
202        primary_key: Register,
203        timestamp: Register,
204    },
205    LookupTemporalIndex {
206        state_id: u32,
207        index_name: String,
208        lookup_value: Register,
209        timestamp: Register,
210        dest: Register,
211    },
212    UpdateLookupIndex {
213        state_id: u32,
214        index_name: String,
215        lookup_value: Register,
216        primary_key: Register,
217    },
218    LookupIndex {
219        state_id: u32,
220        index_name: String,
221        lookup_value: Register,
222        dest: Register,
223    },
224    /// Sum a numeric value to a field (accumulator)
225    SetFieldSum {
226        object: Register,
227        path: String,
228        value: Register,
229    },
230    /// Increment a counter field by 1
231    SetFieldIncrement {
232        object: Register,
233        path: String,
234    },
235    /// Set field to minimum value
236    SetFieldMin {
237        object: Register,
238        path: String,
239        value: Register,
240    },
241    /// Set field only if a specific instruction type was seen in the same transaction.
242    /// If not seen yet, defers the operation for later completion.
243    SetFieldWhen {
244        object: Register,
245        path: String,
246        value: Register,
247        when_instruction: String,
248        entity_name: String,
249        key_reg: Register,
250        condition_field: Option<FieldPath>,
251        condition_op: Option<ComparisonOp>,
252        condition_value: Option<Value>,
253    },
254    /// Set field unless stopped by a specific instruction.
255    /// Stop is tracked by a per-entity stop flag.
256    SetFieldUnlessStopped {
257        object: Register,
258        path: String,
259        value: Register,
260        stop_field: String,
261        stop_instruction: String,
262        entity_name: String,
263        key_reg: Register,
264    },
265    /// Add value to unique set and update count
266    /// Maintains internal Set, field stores count
267    AddToUniqueSet {
268        state_id: u32,
269        set_name: String,
270        value: Register,
271        count_object: Register,
272        count_path: String,
273    },
274    /// Conditionally set a field based on a comparison
275    ConditionalSetField {
276        object: Register,
277        path: String,
278        value: Register,
279        condition_field: FieldPath,
280        condition_op: ComparisonOp,
281        condition_value: Value,
282    },
283    /// Conditionally increment a field based on a comparison
284    ConditionalIncrement {
285        object: Register,
286        path: String,
287        condition_field: FieldPath,
288        condition_op: ComparisonOp,
289        condition_value: Value,
290    },
291    /// Evaluate computed fields (calls external hook if provided)
292    /// computed_paths: List of paths that will be computed (for dirty tracking)
293    EvaluateComputedFields {
294        state: Register,
295        computed_paths: Vec<String>,
296    },
297    /// Queue a resolver for asynchronous enrichment
298    QueueResolver {
299        state_id: u32,
300        entity_name: String,
301        resolver: ResolverType,
302        input_path: Option<String>,
303        input_value: Option<Value>,
304        url_template: Option<Vec<UrlTemplatePart>>,
305        strategy: ResolveStrategy,
306        extracts: Vec<ResolverExtractSpec>,
307        condition: Option<ResolverCondition>,
308        schedule_at: Option<String>,
309        state: Register,
310        key: Register,
311    },
312    /// Update PDA reverse lookup table
313    /// Maps a PDA address to its primary key for reverse lookups
314    UpdatePdaReverseLookup {
315        state_id: u32,
316        lookup_name: String,
317        pda_address: Register,
318        primary_key: Register,
319    },
320    /// Separates independently keyed handler segments for one event type.
321    ///
322    /// When one event updates the same entity through different keys (for
323    /// example `split_position` touching both its `first_position` and its
324    /// `second_position`), each key gets its own segment: its own key loading,
325    /// state read, mappings, state write and mutation. The VM runs segments in
326    /// order and isolates them, so an early exit in one segment never skips
327    /// another. Handlers with a single key never contain a boundary.
328    SegmentBoundary,
329}
330
331pub struct EntityBytecode {
332    pub state_id: u32,
333    pub handlers: HashMap<String, Vec<OpCode>>,
334    pub entity_name: String,
335    pub when_events: HashSet<String>,
336    pub non_emitted_fields: HashSet<String>,
337    pub computed_paths: Vec<String>,
338    /// Optional callback for evaluating computed fields
339    /// Parameters: state, context_slot (Option<u64>), context_timestamp (i64)
340    #[allow(clippy::type_complexity)]
341    pub computed_fields_evaluator: Option<
342        Box<
343            dyn Fn(
344                    &mut Value,
345                    Option<u64>,
346                    i64,
347                )
348                    -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
349                + Send
350                + Sync,
351        >,
352    >,
353}
354
355impl std::fmt::Debug for EntityBytecode {
356    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
357        f.debug_struct("EntityBytecode")
358            .field("state_id", &self.state_id)
359            .field("handlers", &self.handlers)
360            .field("entity_name", &self.entity_name)
361            .field("when_events", &self.when_events)
362            .field("non_emitted_fields", &self.non_emitted_fields)
363            .field("computed_paths", &self.computed_paths)
364            .field(
365                "computed_fields_evaluator",
366                &self.computed_fields_evaluator.is_some(),
367            )
368            .finish()
369    }
370}
371
372#[derive(Debug)]
373pub struct MultiEntityBytecode {
374    pub entities: HashMap<String, EntityBytecode>,
375    pub event_routing: HashMap<String, Vec<String>>,
376    pub when_events: HashSet<String>,
377    pub proto_router: crate::proto_router::ProtoRouter,
378}
379
380impl MultiEntityBytecode {
381    pub fn from_single<S>(entity_name: String, spec: TypedStreamSpec<S>, state_id: u32) -> Self {
382        let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
383        let entity_bytecode = compiler.compile_entity();
384
385        let mut entities = HashMap::new();
386        let mut event_routing = HashMap::new();
387        let mut when_events = HashSet::new();
388
389        for event_type in entity_bytecode.handlers.keys() {
390            event_routing
391                .entry(event_type.clone())
392                .or_insert_with(Vec::new)
393                .push(entity_name.clone());
394        }
395
396        when_events.extend(entity_bytecode.when_events.iter().cloned());
397
398        entities.insert(entity_name, entity_bytecode);
399
400        MultiEntityBytecode {
401            entities,
402            event_routing,
403            when_events,
404            proto_router: crate::proto_router::ProtoRouter::new(),
405        }
406    }
407
408    pub fn from_entities(entities_vec: Vec<(String, Box<dyn std::any::Any>, u32)>) -> Self {
409        let entities = HashMap::new();
410        let event_routing = HashMap::new();
411        let when_events = HashSet::new();
412
413        if let Some((_entity_name, _spec_any, _state_id)) = entities_vec.into_iter().next() {
414            panic!("from_entities requires type information - use builder pattern instead");
415        }
416
417        MultiEntityBytecode {
418            entities,
419            event_routing,
420            when_events,
421            proto_router: crate::proto_router::ProtoRouter::new(),
422        }
423    }
424
425    #[allow(clippy::new_ret_no_self)]
426    pub fn new() -> MultiEntityBytecodeBuilder {
427        MultiEntityBytecodeBuilder {
428            entities: HashMap::new(),
429            event_routing: HashMap::new(),
430            when_events: HashSet::new(),
431            proto_router: crate::proto_router::ProtoRouter::new(),
432        }
433    }
434
435    /// Deterministic content hash of the compiled bytecode, used to invalidate
436    /// state snapshots when the stack's compiled logic changes.
437    ///
438    /// Covers state ids, handler opcode streams, routing, and field metadata.
439    /// Computed-field evaluators are opaque closures, so only their presence is
440    /// hashed — their bodies are generated from the same stack source as the
441    /// opcode streams, which change alongside them in practice.
442    ///
443    /// Opcode streams are hashed via a canonical serde encoding (externally
444    /// tagged variants, object keys sorted), NOT via `Debug` formatting, so the
445    /// hash survives cosmetic churn that doesn't alter the compiled program:
446    /// rustc `Debug` formatting changes, struct field reordering, and inserting
447    /// new `OpCode` variants. Renaming a variant or field DOES change the hash
448    /// (names are the tags) — treat renames as snapshot-invalidating. Interpreter
449    /// behavior changes that keep the encoding identical are invisible here;
450    /// those must bump `SNAPSHOT_FORMAT_VERSION` instead.
451    pub fn fingerprint(&self) -> String {
452        use sha2::{Digest, Sha256};
453
454        fn update_sorted_set(hasher: &mut Sha256, set: &HashSet<String>) {
455            let mut items: Vec<&String> = set.iter().collect();
456            items.sort();
457            for item in items {
458                hasher.update(item.as_bytes());
459                hasher.update([0u8]);
460            }
461            hasher.update([1u8]);
462        }
463
464        /// Hash a JSON value in a canonical form: object keys visited in sorted
465        /// order with explicit framing bytes. Deliberately independent of
466        /// `serde_json::Map`'s iteration order, which flips to insertion order
467        /// if any crate in the enclosing workspace enables the `preserve_order`
468        /// feature — the fingerprint must agree across differently-featured
469        /// builds of the same stack.
470        fn update_canonical_json(hasher: &mut Sha256, value: &Value) {
471            match value {
472                Value::Object(map) => {
473                    let mut keys: Vec<&String> = map.keys().collect();
474                    keys.sort();
475                    hasher.update(*b"{");
476                    for key in keys {
477                        hasher.update(key.as_bytes());
478                        hasher.update([0u8]);
479                        update_canonical_json(hasher, &map[key]);
480                    }
481                    hasher.update(*b"}");
482                }
483                Value::Array(items) => {
484                    hasher.update(*b"[");
485                    for item in items {
486                        update_canonical_json(hasher, item);
487                        hasher.update([0u8]);
488                    }
489                    hasher.update(*b"]");
490                }
491                // Scalars: serde_json's rendering (itoa/ryu for numbers, JSON
492                // string escaping) is deterministic across platforms.
493                scalar => {
494                    hasher.update(scalar.to_string().as_bytes());
495                    hasher.update([0u8]);
496                }
497            }
498        }
499
500        let mut hasher = Sha256::new();
501
502        let mut entity_names: Vec<&String> = self.entities.keys().collect();
503        entity_names.sort();
504        for entity_name in entity_names {
505            let entity = &self.entities[entity_name];
506            hasher.update(entity_name.as_bytes());
507            hasher.update([0u8]);
508            hasher.update(entity.state_id.to_le_bytes());
509
510            let mut event_types: Vec<&String> = entity.handlers.keys().collect();
511            event_types.sort();
512            for event_type in event_types {
513                hasher.update(event_type.as_bytes());
514                hasher.update([0u8]);
515                let ops = serde_json::to_value(&entity.handlers[event_type]).expect(
516                    "opcode stream must serialize for fingerprinting; \
517                     a non-serializable OpCode payload breaks snapshot invalidation",
518                );
519                update_canonical_json(&mut hasher, &ops);
520                hasher.update([0u8]);
521            }
522            hasher.update([1u8]);
523
524            update_sorted_set(&mut hasher, &entity.when_events);
525            update_sorted_set(&mut hasher, &entity.non_emitted_fields);
526            for path in &entity.computed_paths {
527                hasher.update(path.as_bytes());
528                hasher.update([0u8]);
529            }
530            hasher.update([entity.computed_fields_evaluator.is_some() as u8]);
531        }
532
533        let mut routes: Vec<(&String, &Vec<String>)> = self.event_routing.iter().collect();
534        routes.sort_by_key(|(event_type, _)| *event_type);
535        for (event_type, entity_names) in routes {
536            hasher.update(event_type.as_bytes());
537            hasher.update([0u8]);
538            let mut sorted_names = entity_names.clone();
539            sorted_names.sort();
540            for name in sorted_names {
541                hasher.update(name.as_bytes());
542                hasher.update([0u8]);
543            }
544            hasher.update([1u8]);
545        }
546
547        update_sorted_set(&mut hasher, &self.when_events);
548
549        hex::encode(hasher.finalize())
550    }
551}
552
553pub struct MultiEntityBytecodeBuilder {
554    entities: HashMap<String, EntityBytecode>,
555    event_routing: HashMap<String, Vec<String>>,
556    when_events: HashSet<String>,
557    proto_router: crate::proto_router::ProtoRouter,
558}
559
560impl MultiEntityBytecodeBuilder {
561    pub fn add_entity<S>(
562        self,
563        entity_name: String,
564        spec: TypedStreamSpec<S>,
565        state_id: u32,
566    ) -> Self {
567        self.add_entity_with_evaluator(
568            entity_name,
569            spec,
570            state_id,
571            None::<
572                fn(
573                    &mut Value,
574                    Option<u64>,
575                    i64,
576                )
577                    -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>,
578            >,
579        )
580    }
581
582    pub fn add_entity_with_evaluator<S, F>(
583        mut self,
584        entity_name: String,
585        spec: TypedStreamSpec<S>,
586        state_id: u32,
587        evaluator: Option<F>,
588    ) -> Self
589    where
590        F: Fn(
591                &mut Value,
592                Option<u64>,
593                i64,
594            ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
595            + Send
596            + Sync
597            + 'static,
598    {
599        let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
600        let mut entity_bytecode = compiler.compile_entity();
601
602        // Store the evaluator callback if provided
603        if let Some(eval) = evaluator {
604            entity_bytecode.computed_fields_evaluator = Some(Box::new(eval));
605        }
606
607        for event_type in entity_bytecode.handlers.keys() {
608            self.event_routing
609                .entry(event_type.clone())
610                .or_default()
611                .push(entity_name.clone());
612        }
613
614        self.when_events
615            .extend(entity_bytecode.when_events.iter().cloned());
616
617        self.entities.insert(entity_name, entity_bytecode);
618        self
619    }
620
621    pub fn build(self) -> MultiEntityBytecode {
622        MultiEntityBytecode {
623            entities: self.entities,
624            event_routing: self.event_routing,
625            when_events: self.when_events,
626            proto_router: self.proto_router,
627        }
628    }
629}
630
631pub struct TypedCompiler<S> {
632    pub spec: TypedStreamSpec<S>,
633    entity_name: String,
634    state_id: u32,
635}
636
637impl<S> TypedCompiler<S> {
638    pub fn new(spec: TypedStreamSpec<S>, entity_name: String) -> Self {
639        TypedCompiler {
640            spec,
641            entity_name,
642            state_id: 0,
643        }
644    }
645
646    pub fn with_state_id(mut self, state_id: u32) -> Self {
647        self.state_id = state_id;
648        self
649    }
650
651    pub fn compile(&self) -> MultiEntityBytecode {
652        let entity_bytecode = self.compile_entity();
653
654        let mut entities = HashMap::new();
655        let mut event_routing = HashMap::new();
656        let mut when_events = HashSet::new();
657
658        for event_type in entity_bytecode.handlers.keys() {
659            event_routing
660                .entry(event_type.clone())
661                .or_insert_with(Vec::new)
662                .push(self.entity_name.clone());
663        }
664
665        when_events.extend(entity_bytecode.when_events.iter().cloned());
666
667        entities.insert(self.entity_name.clone(), entity_bytecode);
668
669        MultiEntityBytecode {
670            entities,
671            event_routing,
672            when_events,
673            proto_router: crate::proto_router::ProtoRouter::new(),
674        }
675    }
676
677    fn resolver_outputs_primary_key_directly(&self, primary_field: &FieldPath) -> bool {
678        if primary_field.segments.as_slice() != ["__account_address"] {
679            return false;
680        }
681
682        let primary_key_leafs: HashSet<&str> = self
683            .spec
684            .identity
685            .primary_keys
686            .iter()
687            .map(|path| path.rsplit('.').next().unwrap_or(path.as_str()))
688            .collect();
689
690        self.spec.instruction_hooks.iter().any(|hook| {
691            hook.actions.iter().any(|action| {
692                matches!(action, HookAction::RegisterPdaMapping { seed_field, .. }
693                    if seed_field
694                        .segments
695                        .last()
696                        .is_some_and(|segment| primary_key_leafs.contains(segment.as_str())))
697            })
698        })
699    }
700
701    fn compile_entity(&self) -> EntityBytecode {
702        let mut when_events: HashSet<String> = HashSet::new();
703        let mut emit_by_path: HashMap<String, bool> = HashMap::new();
704
705        // DEBUG: Collect all handler info before processing
706        let mut debug_info = Vec::new();
707        for (index, handler_spec) in self.spec.handlers.iter().enumerate() {
708            let event_type = self.get_event_type(&handler_spec.source);
709            let program_id = match &handler_spec.source {
710                crate::ast::SourceSpec::Source { program_id, .. } => {
711                    program_id.as_ref().map(|s| s.as_str()).unwrap_or("null")
712                }
713            };
714            debug_info.push(format!(
715                "  [{}] EventType={}, Mappings={}, ProgramId={}",
716                index,
717                event_type,
718                handler_spec.mappings.len(),
719                program_id
720            ));
721        }
722
723        // DEBUG: Log handler information (optional - can be removed later)
724        // Uncomment to debug handler processing:
725        // if self.entity_name == "PumpfunToken" {
726        //     eprintln!("🔍 Compiling {} handlers for {}", self.spec.handlers.len(), self.entity_name);
727        //     for info in &debug_info {
728        //         eprintln!("{}", info);
729        //     }
730        // }
731
732        // Handlers for one event type are grouped into independently keyed
733        // segments. Two handlers merge only when their key resolution compiles
734        // to identical opcodes; a handler that resolves its key differently
735        // (e.g. `split_position` via `first_position` vs `second_position`)
736        // gets its own segment so each update lands on its own entity.
737        let mut segments: HashMap<String, Vec<HandlerSegment>> = HashMap::new();
738
739        for handler_spec in &self.spec.handlers {
740            for mapping in &handler_spec.mappings {
741                if let Some(when) = &mapping.when {
742                    when_events.insert(when.clone());
743                }
744                let entry = emit_by_path
745                    .entry(mapping.target_path.clone())
746                    .or_insert(false);
747                *entry |= mapping.emit;
748                if mapping.stop.is_some() {
749                    emit_by_path
750                        .entry(stop_field_path(&mapping.target_path))
751                        .or_insert(false);
752                }
753            }
754            let opcodes = self.compile_handler(handler_spec);
755            let event_type = self.get_event_type(&handler_spec.source);
756            let key_identity = self.key_identity(handler_spec);
757
758            let event_segments = segments.entry(event_type).or_default();
759            if let Some(existing) = event_segments
760                .iter_mut()
761                .find(|segment| segment.key_identity.as_ref() == Some(&key_identity))
762            {
763                merge_handler_opcodes(&mut existing.ops, &opcodes);
764            } else {
765                event_segments.push(HandlerSegment {
766                    key_identity: Some(key_identity),
767                    route_field: route_field(&handler_spec.key_resolution),
768                    ops: opcodes,
769                });
770            }
771        }
772
773        // Process instruction_hooks to add SetField/IncrementField operations
774        for hook in &self.spec.instruction_hooks {
775            let event_type = hook.instruction_type.clone();
776            let event_segments = segments.entry(event_type).or_default();
777
778            // A hook joins the segment that routes by its own `lookup_by` field.
779            // Hooks without `lookup_by` (PDA registrations, conditional
780            // aggregates) join the first segment. A hook whose `lookup_by`
781            // matches no segment gets its own segment keyed by that field
782            // instead of silently riding another segment's key.
783            let segment_index = match &hook.lookup_by {
784                None if !event_segments.is_empty() => Some(0),
785                None => None,
786                Some(lookup_by) => event_segments
787                    .iter()
788                    .position(|segment| segment.route_field.as_ref() == Some(lookup_by)),
789            };
790            let segment_index = match segment_index {
791                Some(index) => index,
792                None => {
793                    // Next to handlers that route differently, resolve the
794                    // hook's own key the way a handler would: directly for a
795                    // primary-key field, through the lookup index otherwise.
796                    // A hook alone keeps the historical direct-key segment.
797                    let resolution = if event_segments.is_empty() {
798                        None
799                    } else {
800                        hook.lookup_by
801                            .as_ref()
802                            .and_then(|lookup_by| self.hook_key_resolution(lookup_by))
803                    };
804                    let ops = match resolution {
805                        Some(resolution) => self.compile_keyed_hook_segment(&resolution),
806                        None => self.compile_hook_segment(hook),
807                    };
808                    event_segments.push(HandlerSegment {
809                        key_identity: None,
810                        route_field: hook.lookup_by.clone(),
811                        ops,
812                    });
813                    event_segments.len() - 1
814                }
815            };
816            let handler_opcodes = &mut event_segments[segment_index].ops;
817
818            // Generate opcodes for each action in the hook
819            let hook_opcodes = self.compile_instruction_hook_actions(&hook.actions);
820
821            // Insert hook opcodes before EvaluateComputedFields (if present) or UpdateState
822            // Hook actions (like whale_trade_count increment) must run before computed fields
823            // are evaluated, since computed fields may depend on the modified state
824            let insert_pos = handler_opcodes
825                .iter()
826                .position(|op| matches!(op, OpCode::EvaluateComputedFields { .. }))
827                .or_else(|| {
828                    handler_opcodes
829                        .iter()
830                        .position(|op| matches!(op, OpCode::UpdateState { .. }))
831                });
832
833            if let Some(pos) = insert_pos {
834                // Insert hook opcodes before EvaluateComputedFields or UpdateState
835                for (i, opcode) in hook_opcodes.into_iter().enumerate() {
836                    handler_opcodes.insert(pos + i, opcode);
837                }
838            }
839        }
840
841        let handlers: HashMap<String, Vec<OpCode>> = segments
842            .into_iter()
843            .map(|(event_type, event_segments)| {
844                let mut ops = Vec::new();
845                for (index, segment) in event_segments.into_iter().enumerate() {
846                    if index > 0 {
847                        ops.push(OpCode::SegmentBoundary);
848                    }
849                    ops.extend(segment.ops);
850                }
851                (event_type, ops)
852            })
853            .collect();
854
855        let non_emitted_fields: HashSet<String> = emit_by_path
856            .into_iter()
857            .filter_map(|(path, emit)| if emit { None } else { Some(path) })
858            .collect();
859
860        EntityBytecode {
861            state_id: self.state_id,
862            handlers,
863            entity_name: self.entity_name.clone(),
864            when_events,
865            non_emitted_fields,
866            computed_paths: self.spec.computed_fields.clone(),
867            computed_fields_evaluator: None,
868        }
869    }
870
871    /// Canonical encoding of the opcodes that resolve a handler's key. Two
872    /// handlers for the same event may share one state read/write only when
873    /// these are identical.
874    fn key_identity(&self, spec: &TypedHandlerSpec<S>) -> Value {
875        let key_reg = 20;
876        let ops = self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings);
877        serde_json::to_value(&ops).unwrap_or(Value::Null)
878    }
879
880    /// How a hook's `lookup_by` field resolves the entity key: directly when
881    /// it names a primary-key field, through its lookup index when it names a
882    /// lookup-index field.
883    fn hook_key_resolution(&self, lookup_by: &FieldPath) -> Option<KeyResolutionStrategy> {
884        let leaf = lookup_by.segments.last()?;
885        let is_primary_key_field = self
886            .spec
887            .identity
888            .primary_keys
889            .iter()
890            .any(|pk| pk.rsplit('.').next() == Some(leaf.as_str()));
891        if is_primary_key_field {
892            Some(KeyResolutionStrategy::Embedded {
893                primary_field: lookup_by.clone(),
894            })
895        } else if self.find_lookup_index_for_field(lookup_by).is_some() {
896            Some(KeyResolutionStrategy::Lookup {
897                primary_field: lookup_by.clone(),
898            })
899        } else {
900            None
901        }
902    }
903
904    /// Segment for an instruction hook that routes by its own resolved key:
905    /// key loading, state read and state write. Hook actions are inserted
906    /// before the state write by the caller.
907    fn compile_keyed_hook_segment(&self, resolution: &KeyResolutionStrategy) -> Vec<OpCode> {
908        let key_reg = 20;
909        let state_reg = 2;
910        let mut ops = self.compile_key_loading(resolution, key_reg, &[]);
911        ops.push(OpCode::AbortIfNullKey {
912            key: key_reg,
913            is_account_event: false,
914        });
915        ops.push(OpCode::ReadOrInitState {
916            state_id: self.state_id,
917            key: key_reg,
918            default: serde_json::json!({}),
919            dest: state_reg,
920        });
921        ops.push(OpCode::UpdateState {
922            state_id: self.state_id,
923            key: key_reg,
924            value: state_reg,
925        });
926        ops
927    }
928
929    /// Standalone segment for an instruction hook: key from the resolver or
930    /// the hook's `lookup_by` field, then state read and write. Hook actions are
931    /// inserted before the state write by the caller.
932    fn compile_hook_segment(&self, hook: &InstructionHook) -> Vec<OpCode> {
933        let key_reg = 20;
934        let state_reg = 2;
935        let resolved_key_reg = 19;
936        let temp_reg = 18;
937
938        let mut ops = Vec::new();
939
940        // First, try to load __resolved_primary_key from resolver
941        ops.push(OpCode::LoadEventField {
942            path: FieldPath::new(&["__resolved_primary_key"]),
943            dest: resolved_key_reg,
944            default: Some(serde_json::json!(null)),
945        });
946
947        // Copy to key_reg (unconditionally, may be null)
948        ops.push(OpCode::CopyRegister {
949            source: resolved_key_reg,
950            dest: key_reg,
951        });
952
953        // If hook has lookup_by, use it to load primary key from instruction accounts
954        if let Some(lookup_path) = &hook.lookup_by {
955            // Load the primary key from the instruction's lookup_by field (e.g., accounts.signer)
956            ops.push(OpCode::LoadEventField {
957                path: lookup_path.clone(),
958                dest: temp_reg,
959                default: None,
960            });
961
962            // Apply HexEncode transformation (accounts are byte arrays)
963            ops.push(OpCode::Transform {
964                source: temp_reg,
965                dest: temp_reg,
966                transformation: Transformation::HexEncode,
967            });
968
969            // Use this as fallback if __resolved_primary_key was null
970            ops.push(OpCode::CopyRegisterIfNull {
971                source: temp_reg,
972                dest: key_reg,
973            });
974        }
975
976        ops.push(OpCode::ReadOrInitState {
977            state_id: self.state_id,
978            key: key_reg,
979            default: serde_json::json!({}),
980            dest: state_reg,
981        });
982
983        ops.push(OpCode::UpdateState {
984            state_id: self.state_id,
985            key: key_reg,
986            value: state_reg,
987        });
988
989        ops
990    }
991
992    fn compile_handler(&self, spec: &TypedHandlerSpec<S>) -> Vec<OpCode> {
993        let mut ops = Vec::new();
994        let state_reg = 2;
995        let key_reg = 20;
996
997        ops.extend(self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings));
998
999        // Guard: if key resolved to null on an account-state event, abort
1000        // early with empty mutations so process_event can queue the update
1001        // for later reprocessing.  Without this, downstream opcodes would
1002        // create a phantom entity keyed by null and produce non-empty
1003        // mutations that prevent queueing.
1004        let is_account_event = matches!(
1005            spec.source,
1006            SourceSpec::Source {
1007                is_account: true,
1008                ..
1009            }
1010        );
1011        ops.push(OpCode::AbortIfNullKey {
1012            key: key_reg,
1013            is_account_event,
1014        });
1015
1016        ops.push(OpCode::ReadOrInitState {
1017            state_id: self.state_id,
1018            key: key_reg,
1019            default: serde_json::json!({}),
1020            dest: state_reg,
1021        });
1022
1023        // Index updates must come AFTER ReadOrInitState so the state table exists.
1024        // ReadOrInitState lazily creates the state table via entry().or_insert_with(),
1025        // but index opcodes (UpdateLookupIndex, UpdateTemporalIndex, UpdatePdaReverseLookup)
1026        // use get_mut() which fails if the table doesn't exist yet.
1027        // This ordering also means stale/duplicate updates (caught by ReadOrInitState's
1028        // recency check) correctly skip index updates too.
1029        ops.extend(self.compile_temporal_index_update(
1030            &spec.key_resolution,
1031            key_reg,
1032            &spec.mappings,
1033        ));
1034
1035        for mapping in &spec.mappings {
1036            ops.extend(self.compile_mapping(mapping, state_reg, key_reg));
1037        }
1038
1039        ops.extend(self.compile_resolvers(state_reg, key_reg));
1040
1041        // Evaluate computed fields after all mappings but before updating state
1042        ops.push(OpCode::EvaluateComputedFields {
1043            state: state_reg,
1044            computed_paths: self.spec.computed_fields.clone(),
1045        });
1046
1047        ops.push(OpCode::UpdateState {
1048            state_id: self.state_id,
1049            key: key_reg,
1050            value: state_reg,
1051        });
1052
1053        if spec.emit {
1054            ops.push(OpCode::EmitMutation {
1055                entity_name: self.entity_name.clone(),
1056                key: key_reg,
1057                state: state_reg,
1058            });
1059        }
1060
1061        ops
1062    }
1063
1064    fn compile_resolvers(&self, state_reg: Register, key_reg: Register) -> Vec<OpCode> {
1065        let mut ops = Vec::new();
1066
1067        for resolver_spec in &self.spec.resolver_specs {
1068            let url_template = match &resolver_spec.resolver {
1069                ResolverType::Url(config) => match &config.url_source {
1070                    UrlSource::Template(parts) => Some(parts.clone()),
1071                    _ => None,
1072                },
1073                _ => None,
1074            };
1075
1076            ops.push(OpCode::QueueResolver {
1077                state_id: self.state_id,
1078                entity_name: self.entity_name.clone(),
1079                resolver: resolver_spec.resolver.clone(),
1080                input_path: resolver_spec.input_path.clone(),
1081                input_value: resolver_spec.input_value.clone(),
1082                url_template,
1083                strategy: resolver_spec.strategy.clone(),
1084                extracts: resolver_spec.extracts.clone(),
1085                condition: resolver_spec.condition.clone(),
1086                schedule_at: resolver_spec.schedule_at.clone(),
1087                state: state_reg,
1088                key: key_reg,
1089            });
1090        }
1091
1092        ops
1093    }
1094
1095    fn compile_mapping(
1096        &self,
1097        mapping: &TypedFieldMapping<S>,
1098        state_reg: Register,
1099        key_reg: Register,
1100    ) -> Vec<OpCode> {
1101        let mut ops = Vec::new();
1102        let temp_reg = 10;
1103
1104        ops.extend(self.compile_mapping_source(&mapping.source, temp_reg));
1105
1106        if let Some(transform) = &mapping.transform {
1107            ops.push(OpCode::Transform {
1108                source: temp_reg,
1109                dest: temp_reg,
1110                transformation: transform.clone(),
1111            });
1112        }
1113
1114        if let Some(stop_instruction) = &mapping.stop {
1115            if mapping.when.is_some() {
1116                tracing::warn!(
1117                    "#[map] stop and when both set for {}. Ignoring when.",
1118                    mapping.target_path
1119                );
1120            }
1121            if !matches!(mapping.population, PopulationStrategy::LastWrite)
1122                && !matches!(mapping.population, PopulationStrategy::Merge)
1123            {
1124                tracing::warn!(
1125                    "#[map] stop ignores population strategy {:?}",
1126                    mapping.population
1127                );
1128            }
1129
1130            ops.push(OpCode::SetFieldUnlessStopped {
1131                object: state_reg,
1132                path: mapping.target_path.clone(),
1133                value: temp_reg,
1134                stop_field: stop_field_path(&mapping.target_path),
1135                stop_instruction: stop_instruction.clone(),
1136                entity_name: self.entity_name.clone(),
1137                key_reg,
1138            });
1139            return ops;
1140        }
1141
1142        if let Some(when_instruction) = &mapping.when {
1143            if !matches!(mapping.population, PopulationStrategy::LastWrite)
1144                && !matches!(mapping.population, PopulationStrategy::Merge)
1145            {
1146                tracing::warn!(
1147                    "#[map] when ignores population strategy {:?}",
1148                    mapping.population
1149                );
1150            }
1151            let (condition_field, condition_op, condition_value) = mapping
1152                .condition
1153                .as_ref()
1154                .and_then(|cond| cond.parsed.as_ref())
1155                .and_then(|parsed| match parsed {
1156                    ParsedCondition::Comparison { field, op, value } => {
1157                        Some((Some(field.clone()), Some(op.clone()), Some(value.clone())))
1158                    }
1159                    ParsedCondition::Logical { .. } => {
1160                        tracing::warn!("Logical conditions not yet supported for #[map] when");
1161                        None
1162                    }
1163                })
1164                .unwrap_or((None, None, None));
1165
1166            ops.push(OpCode::SetFieldWhen {
1167                object: state_reg,
1168                path: mapping.target_path.clone(),
1169                value: temp_reg,
1170                when_instruction: when_instruction.clone(),
1171                entity_name: self.entity_name.clone(),
1172                key_reg,
1173                condition_field,
1174                condition_op,
1175                condition_value,
1176            });
1177            return ops;
1178        }
1179
1180        if let Some(condition) = &mapping.condition {
1181            if let Some(parsed) = &condition.parsed {
1182                match parsed {
1183                    ParsedCondition::Comparison {
1184                        field,
1185                        op,
1186                        value: cond_value,
1187                    } => {
1188                        if matches!(mapping.population, PopulationStrategy::LastWrite)
1189                            || matches!(mapping.population, PopulationStrategy::Merge)
1190                        {
1191                            ops.push(OpCode::ConditionalSetField {
1192                                object: state_reg,
1193                                path: mapping.target_path.clone(),
1194                                value: temp_reg,
1195                                condition_field: field.clone(),
1196                                condition_op: op.clone(),
1197                                condition_value: cond_value.clone(),
1198                            });
1199                            return ops;
1200                        }
1201
1202                        if matches!(mapping.population, PopulationStrategy::Count) {
1203                            ops.push(OpCode::ConditionalIncrement {
1204                                object: state_reg,
1205                                path: mapping.target_path.clone(),
1206                                condition_field: field.clone(),
1207                                condition_op: op.clone(),
1208                                condition_value: cond_value.clone(),
1209                            });
1210                            return ops;
1211                        }
1212
1213                        tracing::warn!(
1214                            "Conditional #[map] not supported for population strategy {:?}",
1215                            mapping.population
1216                        );
1217                    }
1218                    ParsedCondition::Logical { .. } => {
1219                        tracing::warn!("Logical conditions not yet supported for #[map]");
1220                    }
1221                }
1222            }
1223        }
1224
1225        match &mapping.population {
1226            PopulationStrategy::Append => {
1227                ops.push(OpCode::AppendToArray {
1228                    object: state_reg,
1229                    path: mapping.target_path.clone(),
1230                    value: temp_reg,
1231                });
1232            }
1233            PopulationStrategy::LastWrite => {
1234                ops.push(OpCode::SetField {
1235                    object: state_reg,
1236                    path: mapping.target_path.clone(),
1237                    value: temp_reg,
1238                });
1239            }
1240            PopulationStrategy::SetOnce => {
1241                ops.push(OpCode::SetFieldIfNull {
1242                    object: state_reg,
1243                    path: mapping.target_path.clone(),
1244                    value: temp_reg,
1245                });
1246            }
1247            PopulationStrategy::Merge => {
1248                ops.push(OpCode::SetField {
1249                    object: state_reg,
1250                    path: mapping.target_path.clone(),
1251                    value: temp_reg,
1252                });
1253            }
1254            PopulationStrategy::Max => {
1255                ops.push(OpCode::SetFieldMax {
1256                    object: state_reg,
1257                    path: mapping.target_path.clone(),
1258                    value: temp_reg,
1259                });
1260            }
1261            PopulationStrategy::Sum => {
1262                ops.push(OpCode::SetFieldSum {
1263                    object: state_reg,
1264                    path: mapping.target_path.clone(),
1265                    value: temp_reg,
1266                });
1267            }
1268            PopulationStrategy::Count => {
1269                // Count doesn't need the value, just increment
1270                ops.push(OpCode::SetFieldIncrement {
1271                    object: state_reg,
1272                    path: mapping.target_path.clone(),
1273                });
1274            }
1275            PopulationStrategy::Min => {
1276                ops.push(OpCode::SetFieldMin {
1277                    object: state_reg,
1278                    path: mapping.target_path.clone(),
1279                    value: temp_reg,
1280                });
1281            }
1282            PopulationStrategy::UniqueCount => {
1283                // UniqueCount requires maintaining an internal set
1284                // The field stores the count, but we track unique values in a hidden set
1285                let set_name = format!("{}_unique_set", mapping.target_path);
1286                ops.push(OpCode::AddToUniqueSet {
1287                    state_id: self.state_id,
1288                    set_name,
1289                    value: temp_reg,
1290                    count_object: state_reg,
1291                    count_path: mapping.target_path.clone(),
1292                });
1293            }
1294        }
1295
1296        ops
1297    }
1298
1299    fn compile_mapping_source(&self, source: &MappingSource, dest: Register) -> Vec<OpCode> {
1300        match source {
1301            MappingSource::FromSource {
1302                path,
1303                default,
1304                transform,
1305            } => {
1306                let mut ops = vec![OpCode::LoadEventField {
1307                    path: path.clone(),
1308                    dest,
1309                    default: default.clone(),
1310                }];
1311
1312                // Apply transform if specified in the source
1313                if let Some(transform_type) = transform {
1314                    ops.push(OpCode::Transform {
1315                        source: dest,
1316                        dest,
1317                        transformation: transform_type.clone(),
1318                    });
1319                }
1320
1321                ops
1322            }
1323            MappingSource::Constant(val) => {
1324                vec![OpCode::LoadConstant {
1325                    value: val.clone(),
1326                    dest,
1327                }]
1328            }
1329            MappingSource::AsEvent { fields } => {
1330                let mut ops = Vec::new();
1331
1332                if fields.is_empty() {
1333                    let event_data_reg = dest + 1;
1334                    ops.push(OpCode::LoadEventField {
1335                        path: FieldPath::new(&[]),
1336                        dest: event_data_reg,
1337                        default: Some(serde_json::json!({})),
1338                    });
1339                    ops.push(OpCode::CreateEvent {
1340                        dest,
1341                        event_value: event_data_reg,
1342                    });
1343                } else {
1344                    let data_obj_reg = dest + 1;
1345                    ops.push(OpCode::CreateObject { dest: data_obj_reg });
1346
1347                    let mut field_registers = Vec::new();
1348                    let mut current_reg = dest + 2;
1349
1350                    for field_source in fields.iter() {
1351                        if let MappingSource::FromSource {
1352                            path,
1353                            default,
1354                            transform,
1355                        } = &**field_source
1356                        {
1357                            ops.push(OpCode::LoadEventField {
1358                                path: path.clone(),
1359                                dest: current_reg,
1360                                default: default.clone(),
1361                            });
1362
1363                            if let Some(transform_type) = transform {
1364                                ops.push(OpCode::Transform {
1365                                    source: current_reg,
1366                                    dest: current_reg,
1367                                    transformation: transform_type.clone(),
1368                                });
1369                            }
1370
1371                            if let Some(field_name) = path.segments.last() {
1372                                field_registers.push((field_name.clone(), current_reg));
1373                            }
1374                            current_reg += 1;
1375                        }
1376                    }
1377
1378                    if !field_registers.is_empty() {
1379                        ops.push(OpCode::SetFields {
1380                            object: data_obj_reg,
1381                            fields: field_registers,
1382                        });
1383                    }
1384
1385                    ops.push(OpCode::CreateEvent {
1386                        dest,
1387                        event_value: data_obj_reg,
1388                    });
1389                }
1390
1391                ops
1392            }
1393            MappingSource::WholeSource => {
1394                vec![OpCode::LoadEventField {
1395                    path: FieldPath::new(&[]),
1396                    dest,
1397                    default: Some(serde_json::json!({})),
1398                }]
1399            }
1400            MappingSource::AsCapture { field_transforms } => {
1401                // AsCapture loads the whole source, applies field-level transforms, and wraps in CaptureWrapper
1402                let capture_data_reg = 22; // Temp register for capture data before wrapping
1403                let mut ops = vec![OpCode::LoadEventField {
1404                    path: FieldPath::new(&[]),
1405                    dest: capture_data_reg,
1406                    default: Some(serde_json::json!({})),
1407                }];
1408
1409                // Apply transforms to specific fields in the loaded object
1410                // IMPORTANT: Use registers that don't conflict with key_reg (20)
1411                // Using 24 and 25 to avoid conflicts with key loading (uses 18, 19, 20, 23)
1412                let field_reg = 24;
1413                let transformed_reg = 25;
1414
1415                for (field_name, transform) in field_transforms {
1416                    // Load the field from the capture_data_reg (not from event!)
1417                    // Use GetField opcode to read from a register instead of LoadEventField
1418                    ops.push(OpCode::GetField {
1419                        object: capture_data_reg,
1420                        path: field_name.clone(),
1421                        dest: field_reg,
1422                    });
1423
1424                    // Transform it
1425                    ops.push(OpCode::Transform {
1426                        source: field_reg,
1427                        dest: transformed_reg,
1428                        transformation: transform.clone(),
1429                    });
1430
1431                    // Set it back into the capture data object
1432                    ops.push(OpCode::SetField {
1433                        object: capture_data_reg,
1434                        path: field_name.clone(),
1435                        value: transformed_reg,
1436                    });
1437                }
1438
1439                // Wrap the capture data in CaptureWrapper with metadata
1440                ops.push(OpCode::CreateCapture {
1441                    dest,
1442                    capture_value: capture_data_reg,
1443                });
1444
1445                ops
1446            }
1447            MappingSource::FromContext { field } => {
1448                // Load from instruction context (timestamp, slot, signature)
1449                vec![OpCode::LoadEventField {
1450                    path: FieldPath::new(&["__update_context", field.as_str()]),
1451                    dest,
1452                    default: Some(serde_json::json!(null)),
1453                }]
1454            }
1455            MappingSource::Computed { .. } => {
1456                vec![]
1457            }
1458            MappingSource::FromState { .. } => {
1459                vec![]
1460            }
1461        }
1462    }
1463
1464    pub fn compile_key_loading(
1465        &self,
1466        resolution: &KeyResolutionStrategy,
1467        key_reg: Register,
1468        mappings: &[TypedFieldMapping<S>],
1469    ) -> Vec<OpCode> {
1470        let mut ops = Vec::new();
1471
1472        // Resolvers provide either a final key or an intermediate lookup key,
1473        // depending on the handler strategy.
1474        let resolved_key_reg = 19; // Use a temp register
1475        ops.push(OpCode::LoadEventField {
1476            path: FieldPath::new(&["__resolved_primary_key"]),
1477            dest: resolved_key_reg,
1478            default: Some(serde_json::json!(null)),
1479        });
1480
1481        // Now do the normal key resolution
1482        match resolution {
1483            KeyResolutionStrategy::Embedded { primary_field } => {
1484                // Enhanced key resolution: check for auto-inheritance when primary_field is empty
1485                let effective_primary_field = if primary_field.segments.is_empty() {
1486                    // Try to auto-detect primary field from account schema
1487                    if let Some(auto_field) = self.auto_detect_primary_field(mappings) {
1488                        auto_field
1489                    } else {
1490                        primary_field.clone()
1491                    }
1492                } else {
1493                    primary_field.clone()
1494                };
1495
1496                // Skip fallback key loading if effective primary_field is still empty
1497                // This happens for account types that rely solely on __resolved_primary_key
1498                // (e.g., accounts with #[resolve_key_for] resolvers)
1499                if !effective_primary_field.segments.is_empty() {
1500                    let temp_reg = 18;
1501                    let transform_reg = 23; // Register for transformed key
1502
1503                    ops.push(OpCode::LoadEventField {
1504                        path: effective_primary_field.clone(),
1505                        dest: temp_reg,
1506                        default: None,
1507                    });
1508
1509                    // Check if there's a transformation for the primary key field
1510                    // First try the current mappings, then inherited transformations
1511                    let primary_key_transform = self
1512                        .find_primary_key_transformation(mappings)
1513                        .or_else(|| self.find_inherited_primary_key_transformation());
1514
1515                    if let Some(transform) = primary_key_transform {
1516                        // Apply transformation to the loaded key
1517                        ops.push(OpCode::Transform {
1518                            source: temp_reg,
1519                            dest: transform_reg,
1520                            transformation: transform,
1521                        });
1522                        // A concrete embedded key belongs to this entity and
1523                        // must win over a resolver result injected for another
1524                        // handler consuming the same event.
1525                        ops.push(OpCode::CopyRegister {
1526                            source: transform_reg,
1527                            dest: key_reg,
1528                        });
1529                    } else {
1530                        // No transformation, use raw value.
1531                        ops.push(OpCode::CopyRegister {
1532                            source: temp_reg,
1533                            dest: key_reg,
1534                        });
1535                    }
1536                    ops.push(OpCode::CopyRegisterIfNull {
1537                        source: resolved_key_reg,
1538                        dest: key_reg,
1539                    });
1540                } else {
1541                    // Resolver-only embedded handlers have no local key field.
1542                    ops.push(OpCode::CopyRegister {
1543                        source: resolved_key_reg,
1544                        dest: key_reg,
1545                    });
1546                }
1547            }
1548            KeyResolutionStrategy::Lookup { primary_field } => {
1549                let lookup_reg = 15;
1550                let result_reg = 17;
1551
1552                // Prefer resolver-provided key as lookup input.
1553                // When __resolved_primary_key is set (e.g. round_address from
1554                // PDA reverse lookup), use it directly — this gives a one-hop
1555                // lookup (round_address → round_id) instead of a two-hop chain
1556                // (Var address → PDA → round_address → round_id).
1557                ops.push(OpCode::CopyRegister {
1558                    source: resolved_key_reg,
1559                    dest: lookup_reg,
1560                });
1561
1562                let temp_reg = 18;
1563                ops.push(OpCode::LoadEventField {
1564                    path: primary_field.clone(),
1565                    dest: temp_reg,
1566                    default: None,
1567                });
1568                ops.push(OpCode::CopyRegisterIfNull {
1569                    source: temp_reg,
1570                    dest: lookup_reg,
1571                });
1572
1573                let index_name = self.find_lookup_index_for_lookup_field(primary_field, mappings);
1574                let effective_index_name =
1575                    index_name.unwrap_or_else(|| "default_pda_lookup".to_string());
1576
1577                ops.push(OpCode::LookupIndex {
1578                    state_id: self.state_id,
1579                    index_name: effective_index_name,
1580                    lookup_value: lookup_reg,
1581                    dest: result_reg,
1582                });
1583                ops.push(OpCode::CopyRegister {
1584                    source: result_reg,
1585                    dest: key_reg,
1586                });
1587
1588                // Most lookup-based account handlers expect resolver output to be an
1589                // intermediate lookup value (for example, PDA -> round_address -> round_id),
1590                // so a null LookupIndex result must leave the key null for queueing.
1591                // Some stacks register the PDA directly to the entity primary key
1592                // (for example, bonding_curve -> mint). In that case the resolver output
1593                // is already the final key, so preserve it only when the instruction hook's
1594                // seed field matches one of the entity primary key fields.
1595                if self.resolver_outputs_primary_key_directly(primary_field) {
1596                    ops.push(OpCode::CopyRegisterIfNull {
1597                        source: resolved_key_reg,
1598                        dest: key_reg,
1599                    });
1600                }
1601            }
1602            KeyResolutionStrategy::Computed {
1603                primary_field,
1604                compute_partition: _,
1605            } => {
1606                // Copy resolver result to key_reg (may be null)
1607                ops.push(OpCode::CopyRegister {
1608                    source: resolved_key_reg,
1609                    dest: key_reg,
1610                });
1611                let temp_reg = 18;
1612                ops.push(OpCode::LoadEventField {
1613                    path: primary_field.clone(),
1614                    dest: temp_reg,
1615                    default: None,
1616                });
1617                ops.push(OpCode::CopyRegisterIfNull {
1618                    source: temp_reg,
1619                    dest: key_reg,
1620                });
1621            }
1622            KeyResolutionStrategy::TemporalLookup {
1623                lookup_field,
1624                timestamp_field,
1625                index_name,
1626            } => {
1627                // Copy resolver result to key_reg (may be null)
1628                ops.push(OpCode::CopyRegister {
1629                    source: resolved_key_reg,
1630                    dest: key_reg,
1631                });
1632                let lookup_reg = 15;
1633                let timestamp_reg = 16;
1634                let result_reg = 17;
1635
1636                ops.push(OpCode::LoadEventField {
1637                    path: lookup_field.clone(),
1638                    dest: lookup_reg,
1639                    default: None,
1640                });
1641
1642                ops.push(OpCode::LoadEventField {
1643                    path: timestamp_field.clone(),
1644                    dest: timestamp_reg,
1645                    default: None,
1646                });
1647
1648                ops.push(OpCode::LookupTemporalIndex {
1649                    state_id: self.state_id,
1650                    index_name: index_name.clone(),
1651                    lookup_value: lookup_reg,
1652                    timestamp: timestamp_reg,
1653                    dest: result_reg,
1654                });
1655
1656                ops.push(OpCode::CopyRegisterIfNull {
1657                    source: result_reg,
1658                    dest: key_reg,
1659                });
1660            }
1661        }
1662
1663        ops
1664    }
1665
1666    fn find_primary_key_transformation(
1667        &self,
1668        mappings: &[TypedFieldMapping<S>],
1669    ) -> Option<Transformation> {
1670        // Find the first primary key in the identity spec
1671        let primary_key = self.spec.identity.primary_keys.first()?;
1672        let primary_field_name = self.extract_primary_field_name(primary_key)?;
1673
1674        // Look for a mapping that targets this primary key
1675        for mapping in mappings {
1676            // Check if this mapping targets the primary key field
1677            if mapping.target_path == *primary_key
1678                || mapping.target_path.ends_with(&format!(".{}", primary_key))
1679            {
1680                // Check mapping-level transform first
1681                if let Some(transform) = &mapping.transform {
1682                    return Some(transform.clone());
1683                }
1684
1685                // Then check source-level transform
1686                if let MappingSource::FromSource {
1687                    transform: Some(transform),
1688                    ..
1689                } = &mapping.source
1690                {
1691                    return Some(transform.clone());
1692                }
1693            }
1694        }
1695
1696        // If no explicit primary key mapping found, check AsCapture field transforms
1697        for mapping in mappings {
1698            if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1699                if let Some(transform) = field_transforms.get(&primary_field_name) {
1700                    return Some(transform.clone());
1701                }
1702            }
1703        }
1704
1705        None
1706    }
1707
1708    /// Look for primary key mappings in other handlers of the same entity
1709    /// This enables cross-handler inheritance of key transformations
1710    pub fn find_inherited_primary_key_transformation(&self) -> Option<Transformation> {
1711        let primary_key = self.spec.identity.primary_keys.first()?;
1712
1713        // Extract the field name from the primary key path (e.g., "id.authority" -> "authority")
1714        let primary_field_name = self.extract_primary_field_name(primary_key)?;
1715
1716        // Search through all handlers in the spec for primary key mappings
1717        for handler in &self.spec.handlers {
1718            for mapping in &handler.mappings {
1719                // Look for mappings targeting the primary key
1720                if mapping.target_path == *primary_key
1721                    || mapping.target_path.ends_with(&format!(".{}", primary_key))
1722                {
1723                    // Check if this mapping comes from a field matching the primary key name
1724                    if let MappingSource::FromSource {
1725                        path, transform, ..
1726                    } = &mapping.source
1727                    {
1728                        if path.segments.last() == Some(&primary_field_name) {
1729                            // Return mapping-level transform first, then source-level transform
1730                            return mapping.transform.clone().or_else(|| transform.clone());
1731                        }
1732                    }
1733                }
1734
1735                // Also check AsCapture field transforms for the primary field
1736                if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1737                    if let Some(transform) = field_transforms.get(&primary_field_name) {
1738                        return Some(transform.clone());
1739                    }
1740                }
1741            }
1742        }
1743
1744        None
1745    }
1746
1747    /// Extract the field name from a primary key path (e.g., "id.authority" -> "authority")
1748    fn extract_primary_field_name(&self, primary_key: &str) -> Option<String> {
1749        // Split by '.' and take the last segment
1750        primary_key.split('.').next_back().map(|s| s.to_string())
1751    }
1752
1753    /// Auto-detect primary field from account schema when no explicit mapping exists
1754    /// This looks for account types that have an 'authority' field and tries to use it
1755    pub fn auto_detect_primary_field(
1756        &self,
1757        current_mappings: &[TypedFieldMapping<S>],
1758    ) -> Option<FieldPath> {
1759        let primary_key = self.spec.identity.primary_keys.first()?;
1760
1761        // Extract the field name from the primary key (e.g., "id.authority" -> "authority")
1762        let primary_field_name = self.extract_primary_field_name(primary_key)?;
1763
1764        // Check if current handler can access the primary field
1765        if self.current_account_has_primary_field(&primary_field_name, current_mappings) {
1766            return Some(FieldPath::new(&[&primary_field_name]));
1767        }
1768
1769        None
1770    }
1771
1772    /// Check if the current account type has the primary field
1773    /// This is determined by looking at the mappings to see what fields are available
1774    fn current_account_has_primary_field(
1775        &self,
1776        field_name: &str,
1777        mappings: &[TypedFieldMapping<S>],
1778    ) -> bool {
1779        // Look through the mappings to see if any reference the primary field
1780        for mapping in mappings {
1781            if let MappingSource::FromSource { path, .. } = &mapping.source {
1782                // Check if this mapping sources from the primary field
1783                if path.segments.last() == Some(&field_name.to_string()) {
1784                    return true;
1785                }
1786            }
1787        }
1788
1789        false
1790    }
1791
1792    /// Check if handler has access to a specific field in its source account
1793    #[allow(dead_code)]
1794    fn handler_has_field(&self, field_name: &str, mappings: &[TypedFieldMapping<S>]) -> bool {
1795        for mapping in mappings {
1796            if let MappingSource::FromSource { path, .. } = &mapping.source {
1797                if path.segments.last() == Some(&field_name.to_string()) {
1798                    return true;
1799                }
1800            }
1801        }
1802        false
1803    }
1804
1805    /// Check if field exists by looking at mappings (IDL-agnostic approach)
1806    /// This avoids hardcoding account schemas and uses actual mapping evidence
1807    #[allow(dead_code)]
1808    fn field_exists_in_mappings(
1809        &self,
1810        field_name: &str,
1811        mappings: &[TypedFieldMapping<S>],
1812    ) -> bool {
1813        // Look through current mappings to see if the field is referenced
1814        for mapping in mappings {
1815            if let MappingSource::FromSource { path, .. } = &mapping.source {
1816                if path.segments.last() == Some(&field_name.to_string()) {
1817                    return true;
1818                }
1819            }
1820            // Also check AsCapture field transforms
1821            if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1822                if field_transforms.contains_key(field_name) {
1823                    return true;
1824                }
1825            }
1826        }
1827        false
1828    }
1829
1830    fn find_lookup_index_for_field(&self, field_path: &FieldPath) -> Option<String> {
1831        if field_path.segments.is_empty() {
1832            return None;
1833        }
1834
1835        let lookup_field_name = field_path.segments.last().unwrap();
1836
1837        for lookup_index in &self.spec.identity.lookup_indexes {
1838            let index_field_name = lookup_index
1839                .field_name
1840                .split('.')
1841                .next_back()
1842                .unwrap_or(&lookup_index.field_name);
1843            let matches_directly = index_field_name == lookup_field_name;
1844            // An index field named `foo_address` is treated as an alias for the
1845            // bare field `foo` (for example, `mint_address` resolves handlers
1846            // keyed on `mint`). This is intentionally one-way: bare `foo` does
1847            // not imply a `foo_address` lookup index. The macro crate mirrors
1848            // this convention via `lookup_index_leafs` in
1849            // `arete-macros/src/validation/mod.rs`.
1850            let matches_address_alias = index_field_name
1851                .strip_suffix("_address")
1852                .map(|base| base == lookup_field_name)
1853                .unwrap_or(false);
1854
1855            if matches_directly || matches_address_alias {
1856                return Some(format!("{}_lookup_index", index_field_name));
1857            }
1858        }
1859
1860        None
1861    }
1862
1863    /// Find lookup index for a Lookup key resolution by checking if there's a mapping
1864    /// from the primary_field to a lookup index field.
1865    fn find_lookup_index_for_lookup_field(
1866        &self,
1867        primary_field: &FieldPath,
1868        mappings: &[TypedFieldMapping<S>],
1869    ) -> Option<String> {
1870        // Build the primary field path string
1871        let primary_path = primary_field.segments.join(".");
1872
1873        // Check if there's a mapping from this primary field to a lookup index field
1874        for mapping in mappings {
1875            // Check if the mapping source path matches the primary field
1876            if let MappingSource::FromSource { path, .. } = &mapping.source {
1877                let source_path = path.segments.join(".");
1878                if source_path == primary_path {
1879                    // Check if the target is a lookup index field
1880                    for lookup_index in &self.spec.identity.lookup_indexes {
1881                        if mapping.target_path == lookup_index.field_name {
1882                            let index_field_name = lookup_index
1883                                .field_name
1884                                .split('.')
1885                                .next_back()
1886                                .unwrap_or(&lookup_index.field_name);
1887                            return Some(format!("{}_lookup_index", index_field_name));
1888                        }
1889                    }
1890                }
1891            }
1892        }
1893
1894        // Fall back to direct field name matching
1895        self.find_lookup_index_for_field(primary_field)
1896    }
1897
1898    /// Find the source path for a lookup index field by looking at mappings.
1899    /// For example, if target_path is "id.round_address" and the mapping is
1900    /// `id.round_address <- __account_address`, this returns ["__account_address"].
1901    fn find_source_path_for_lookup_index(
1902        &self,
1903        mappings: &[TypedFieldMapping<S>],
1904        lookup_field_name: &str,
1905    ) -> Option<Vec<String>> {
1906        for mapping in mappings {
1907            if mapping.target_path == lookup_field_name {
1908                if let MappingSource::FromSource { path, .. } = &mapping.source {
1909                    return Some(path.segments.clone());
1910                }
1911            }
1912        }
1913        None
1914    }
1915
1916    fn compile_temporal_index_update(
1917        &self,
1918        resolution: &KeyResolutionStrategy,
1919        key_reg: Register,
1920        mappings: &[TypedFieldMapping<S>],
1921    ) -> Vec<OpCode> {
1922        let mut ops = Vec::new();
1923
1924        for lookup_index in &self.spec.identity.lookup_indexes {
1925            let lookup_reg = 17;
1926            let source_field = lookup_index
1927                .field_name
1928                .split('.')
1929                .next_back()
1930                .unwrap_or(&lookup_index.field_name);
1931
1932            match resolution {
1933                KeyResolutionStrategy::Embedded { primary_field: _ } => {
1934                    // For Embedded handlers, find the mapping that targets this lookup index field
1935                    // and use its source path to load the lookup value
1936                    let source_path_opt =
1937                        self.find_source_path_for_lookup_index(mappings, &lookup_index.field_name);
1938
1939                    let load_path = if let Some(ref path) = source_path_opt {
1940                        FieldPath::new(&path.iter().map(|s| s.as_str()).collect::<Vec<_>>())
1941                    } else {
1942                        // Fallback to source_field if no mapping found
1943                        FieldPath::new(&[source_field])
1944                    };
1945
1946                    ops.push(OpCode::LoadEventField {
1947                        path: load_path,
1948                        dest: lookup_reg,
1949                        default: None,
1950                    });
1951
1952                    if let Some(temporal_field_name) = &lookup_index.temporal_field {
1953                        let timestamp_reg = 18;
1954
1955                        ops.push(OpCode::LoadEventField {
1956                            path: FieldPath::new(&[temporal_field_name]),
1957                            dest: timestamp_reg,
1958                            default: None,
1959                        });
1960
1961                        let index_name = format!("{}_temporal_index", source_field);
1962                        ops.push(OpCode::UpdateTemporalIndex {
1963                            state_id: self.state_id,
1964                            index_name,
1965                            lookup_value: lookup_reg,
1966                            primary_key: key_reg,
1967                            timestamp: timestamp_reg,
1968                        });
1969
1970                        let simple_index_name = format!("{}_lookup_index", source_field);
1971                        ops.push(OpCode::UpdateLookupIndex {
1972                            state_id: self.state_id,
1973                            index_name: simple_index_name,
1974                            lookup_value: lookup_reg,
1975                            primary_key: key_reg,
1976                        });
1977                    } else {
1978                        let index_name = format!("{}_lookup_index", source_field);
1979                        ops.push(OpCode::UpdateLookupIndex {
1980                            state_id: self.state_id,
1981                            index_name,
1982                            lookup_value: lookup_reg,
1983                            primary_key: key_reg,
1984                        });
1985                    }
1986
1987                    // Also update PDA reverse lookup table if there's a resolver configured for this entity
1988                    // This allows instruction handlers to look up the primary key from PDA addresses
1989                    // Only do this when the source path is different (e.g., __account_address -> id.round_address)
1990                    if source_path_opt.is_some() {
1991                        ops.push(OpCode::UpdatePdaReverseLookup {
1992                            state_id: self.state_id,
1993                            lookup_name: "default_pda_lookup".to_string(),
1994                            pda_address: lookup_reg,
1995                            primary_key: key_reg,
1996                        });
1997                    }
1998                }
1999                KeyResolutionStrategy::Lookup { primary_field } => {
2000                    // For Lookup handlers, check if there's a mapping that targets this lookup index field
2001                    // If so, the lookup value is the same as the primary_field used for key resolution
2002                    let has_mapping_to_lookup_field = mappings
2003                        .iter()
2004                        .any(|m| m.target_path == lookup_index.field_name);
2005
2006                    if has_mapping_to_lookup_field {
2007                        // Load the lookup value from the event using the primary_field path
2008                        // (this is the same value used for key resolution)
2009                        let path_segments: Vec<&str> =
2010                            primary_field.segments.iter().map(|s| s.as_str()).collect();
2011                        ops.push(OpCode::LoadEventField {
2012                            path: FieldPath::new(&path_segments),
2013                            dest: lookup_reg,
2014                            default: None,
2015                        });
2016
2017                        let index_name = format!("{}_lookup_index", source_field);
2018                        ops.push(OpCode::UpdateLookupIndex {
2019                            state_id: self.state_id,
2020                            index_name,
2021                            lookup_value: lookup_reg,
2022                            primary_key: key_reg,
2023                        });
2024                    }
2025                }
2026                KeyResolutionStrategy::Computed { .. }
2027                | KeyResolutionStrategy::TemporalLookup { .. } => {
2028                    // Computed and TemporalLookup handlers don't populate lookup indexes
2029                }
2030            }
2031        }
2032
2033        ops
2034    }
2035
2036    fn get_event_type(&self, source: &SourceSpec) -> String {
2037        match source {
2038            SourceSpec::Source { type_name, .. } => type_name.clone(),
2039        }
2040    }
2041
2042    fn compile_instruction_hook_actions(&self, actions: &[HookAction]) -> Vec<OpCode> {
2043        let mut ops = Vec::new();
2044        let state_reg = 2;
2045
2046        for action in actions {
2047            match action {
2048                HookAction::SetField {
2049                    target_field,
2050                    source,
2051                    condition,
2052                } => {
2053                    // Check if there's a condition - evaluation handled in VM
2054                    let _ = condition;
2055
2056                    let temp_reg = 11; // Use register 11 for hook values
2057
2058                    // Load the source value
2059                    let load_ops = self.compile_mapping_source(source, temp_reg);
2060                    ops.extend(load_ops);
2061
2062                    // Apply transformation if specified in source
2063                    if let MappingSource::FromSource {
2064                        transform: Some(transform_type),
2065                        ..
2066                    } = source
2067                    {
2068                        ops.push(OpCode::Transform {
2069                            source: temp_reg,
2070                            dest: temp_reg,
2071                            transformation: transform_type.clone(),
2072                        });
2073                    }
2074
2075                    // Conditionally set the field based on parsed condition
2076                    if let Some(cond_expr) = condition {
2077                        if let Some(parsed) = &cond_expr.parsed {
2078                            // Generate condition check opcodes
2079                            let cond_check_ops = self.compile_condition_check(
2080                                parsed,
2081                                temp_reg,
2082                                state_reg,
2083                                target_field,
2084                            );
2085                            ops.extend(cond_check_ops);
2086                        } else {
2087                            // No parsed condition, set unconditionally
2088                            ops.push(OpCode::SetField {
2089                                object: state_reg,
2090                                path: target_field.clone(),
2091                                value: temp_reg,
2092                            });
2093                        }
2094                    } else {
2095                        // No condition, set unconditionally
2096                        ops.push(OpCode::SetField {
2097                            object: state_reg,
2098                            path: target_field.clone(),
2099                            value: temp_reg,
2100                        });
2101                    }
2102                }
2103                HookAction::IncrementField {
2104                    target_field,
2105                    increment_by,
2106                    condition,
2107                } => {
2108                    if let Some(cond_expr) = condition {
2109                        if let Some(parsed) = &cond_expr.parsed {
2110                            // For increment with condition, we need to:
2111                            // 1. Load the condition field
2112                            // 2. Check the condition
2113                            // 3. Conditionally increment
2114                            let cond_check_ops = self.compile_conditional_increment(
2115                                parsed,
2116                                state_reg,
2117                                target_field,
2118                                *increment_by,
2119                            );
2120                            ops.extend(cond_check_ops);
2121                        } else {
2122                            // No parsed condition, increment unconditionally
2123                            ops.push(OpCode::SetFieldIncrement {
2124                                object: state_reg,
2125                                path: target_field.clone(),
2126                            });
2127                        }
2128                    } else {
2129                        // No condition, increment unconditionally
2130                        ops.push(OpCode::SetFieldIncrement {
2131                            object: state_reg,
2132                            path: target_field.clone(),
2133                        });
2134                    }
2135                }
2136                HookAction::RegisterPdaMapping { .. } => {
2137                    if let HookAction::RegisterPdaMapping {
2138                        pda_field,
2139                        seed_field,
2140                        lookup_name,
2141                    } = action
2142                    {
2143                        let pda_reg = 11;
2144                        let seed_reg = 12;
2145
2146                        ops.push(OpCode::LoadEventField {
2147                            path: pda_field.clone(),
2148                            dest: pda_reg,
2149                            default: None,
2150                        });
2151                        ops.push(OpCode::LoadEventField {
2152                            path: seed_field.clone(),
2153                            dest: seed_reg,
2154                            default: None,
2155                        });
2156                        ops.push(OpCode::UpdatePdaReverseLookup {
2157                            state_id: self.state_id,
2158                            lookup_name: lookup_name.clone(),
2159                            pda_address: pda_reg,
2160                            primary_key: seed_reg,
2161                        });
2162                    }
2163                }
2164            }
2165        }
2166
2167        ops
2168    }
2169
2170    fn compile_condition_check(
2171        &self,
2172        condition: &ParsedCondition,
2173        value_reg: Register,
2174        state_reg: Register,
2175        target_field: &str,
2176    ) -> Vec<OpCode> {
2177        match condition {
2178            ParsedCondition::Comparison {
2179                field,
2180                op,
2181                value: cond_value,
2182            } => {
2183                // Generate ConditionalSetField opcode
2184                vec![OpCode::ConditionalSetField {
2185                    object: state_reg,
2186                    path: target_field.to_string(),
2187                    value: value_reg,
2188                    condition_field: field.clone(),
2189                    condition_op: op.clone(),
2190                    condition_value: cond_value.clone(),
2191                }]
2192            }
2193            ParsedCondition::Logical { .. } => {
2194                // Logical conditions not yet supported, fall back to unconditional
2195                tracing::warn!("Logical conditions not yet supported in instruction hooks");
2196                vec![OpCode::SetField {
2197                    object: state_reg,
2198                    path: target_field.to_string(),
2199                    value: value_reg,
2200                }]
2201            }
2202        }
2203    }
2204
2205    fn compile_conditional_increment(
2206        &self,
2207        condition: &ParsedCondition,
2208        state_reg: Register,
2209        target_field: &str,
2210        _increment_by: i64,
2211    ) -> Vec<OpCode> {
2212        match condition {
2213            ParsedCondition::Comparison {
2214                field,
2215                op,
2216                value: cond_value,
2217            } => {
2218                vec![OpCode::ConditionalIncrement {
2219                    object: state_reg,
2220                    path: target_field.to_string(),
2221                    condition_field: field.clone(),
2222                    condition_op: op.clone(),
2223                    condition_value: cond_value.clone(),
2224                }]
2225            }
2226            ParsedCondition::Logical { .. } => {
2227                tracing::warn!("Logical conditions not yet supported in instruction hooks");
2228                vec![OpCode::SetFieldIncrement {
2229                    object: state_reg,
2230                    path: target_field.to_string(),
2231                }]
2232            }
2233        }
2234    }
2235}
2236
2237#[cfg(test)]
2238mod tests {
2239    use super::MultiEntityBytecode;
2240    use crate::ast::{
2241        FieldPath, HookAction, IdentitySpec, InstructionHook, KeyResolutionStrategy,
2242        LookupIndexSpec, MappingSource, PopulationStrategy, SerializableFieldMapping,
2243        SerializableHandlerSpec, SerializableStreamSpec, SourceSpec, TypedStreamSpec,
2244    };
2245    use crate::vm::VmContext;
2246    use crate::{Mutation, WholeEntity};
2247    use serde_json::{json, Value};
2248    use std::collections::BTreeMap;
2249
2250    fn mapping(
2251        target_path: &str,
2252        source_path: &[&str],
2253        population: PopulationStrategy,
2254    ) -> SerializableFieldMapping {
2255        SerializableFieldMapping {
2256            target_path: target_path.to_string(),
2257            source: MappingSource::FromSource {
2258                path: FieldPath::new(source_path),
2259                default: None,
2260                transform: None,
2261            },
2262            transform: None,
2263            population,
2264            condition: None,
2265            when: None,
2266            stop: None,
2267            emit: true,
2268        }
2269    }
2270
2271    fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
2272        TypedStreamSpec::from_serializable(SerializableStreamSpec {
2273            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2274            state_name: "PumpfunToken".to_string(),
2275            program_id: None,
2276            idl: None,
2277            identity: IdentitySpec {
2278                primary_keys: vec!["id.mint".to_string()],
2279                lookup_indexes: vec![LookupIndexSpec {
2280                    field_name: "id.bonding_curve".to_string(),
2281                    temporal_field: None,
2282                }],
2283            },
2284            handlers: vec![SerializableHandlerSpec {
2285                source: SourceSpec::Source {
2286                    program_id: None,
2287                    discriminator: None,
2288                    type_name: "pump::BondingCurveState".to_string(),
2289                    serialization: None,
2290                    is_account: true,
2291                },
2292                key_resolution: KeyResolutionStrategy::Lookup {
2293                    primary_field: FieldPath::new(&["__account_address"]),
2294                },
2295                mappings: vec![
2296                    mapping(
2297                        "id.bonding_curve",
2298                        &["__account_address"],
2299                        PopulationStrategy::SetOnce,
2300                    ),
2301                    mapping(
2302                        "reserves.virtual_token_reserves",
2303                        &["virtual_token_reserves"],
2304                        PopulationStrategy::LastWrite,
2305                    ),
2306                ],
2307                conditions: vec![],
2308                emit: true,
2309            }],
2310            sections: vec![],
2311            field_mappings: BTreeMap::new(),
2312            resolver_hooks: vec![],
2313            instruction_hooks: vec![InstructionHook {
2314                instruction_type: "pump::BuyIxState".to_string(),
2315                actions: vec![HookAction::RegisterPdaMapping {
2316                    pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2317                    seed_field: FieldPath::new(&["accounts", "mint"]),
2318                    lookup_name: "default_pda_lookup".to_string(),
2319                }],
2320                lookup_by: None,
2321            }],
2322            resolver_specs: vec![],
2323            computed_fields: vec![],
2324            computed_field_specs: vec![],
2325            content_hash: None,
2326            views: vec![],
2327        })
2328    }
2329
2330    /// A token whose bonding-curve account is keyed by the `id.bonding_curve`
2331    /// lookup index, which `LinkIxState` writes, or through the PDA mapping
2332    /// `BuyIxState` registers to the token's creator (indexed by
2333    /// `CreateIxState`), and whose round account is keyed by the numeric
2334    /// `id.round` index `OpenRoundIxState` writes. Both account handlers
2335    /// record an event, a capture and the update's context timestamp.
2336    fn timed_bonding_curve_spec() -> TypedStreamSpec<Value> {
2337        let with_source = |target: &str, source: MappingSource| SerializableFieldMapping {
2338            source,
2339            ..mapping(target, &[], PopulationStrategy::LastWrite)
2340        };
2341        let timestamped = || {
2342            vec![
2343                with_source(
2344                    "events.reserves",
2345                    MappingSource::AsEvent {
2346                        fields: vec![Box::new(MappingSource::FromSource {
2347                            path: FieldPath::new(&["virtual_token_reserves"]),
2348                            default: None,
2349                            transform: None,
2350                        })],
2351                    },
2352                ),
2353                with_source(
2354                    "snapshots.curve",
2355                    MappingSource::AsCapture {
2356                        field_transforms: BTreeMap::new(),
2357                    },
2358                ),
2359                with_source(
2360                    "state.seen_at",
2361                    MappingSource::FromContext {
2362                        field: "timestamp".to_string(),
2363                    },
2364                ),
2365            ]
2366        };
2367        let source = |type_name: &str, is_account: bool| SourceSpec::Source {
2368            program_id: None,
2369            discriminator: None,
2370            type_name: type_name.to_string(),
2371            serialization: None,
2372            is_account,
2373        };
2374        TypedStreamSpec::from_serializable(SerializableStreamSpec {
2375            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2376            state_name: "PumpfunToken".to_string(),
2377            program_id: None,
2378            idl: None,
2379            identity: IdentitySpec {
2380                primary_keys: vec!["id.mint".to_string()],
2381                lookup_indexes: vec![
2382                    LookupIndexSpec {
2383                        field_name: "id.bonding_curve".to_string(),
2384                        temporal_field: None,
2385                    },
2386                    LookupIndexSpec {
2387                        field_name: "id.creator".to_string(),
2388                        temporal_field: None,
2389                    },
2390                    LookupIndexSpec {
2391                        field_name: "id.round".to_string(),
2392                        temporal_field: None,
2393                    },
2394                ],
2395            },
2396            handlers: vec![
2397                SerializableHandlerSpec {
2398                    source: source("pump::BondingCurveState", true),
2399                    key_resolution: KeyResolutionStrategy::Lookup {
2400                        primary_field: FieldPath::new(&["__account_address"]),
2401                    },
2402                    mappings: [
2403                        vec![mapping(
2404                            "id.bonding_curve",
2405                            &["__account_address"],
2406                            PopulationStrategy::SetOnce,
2407                        )],
2408                        timestamped(),
2409                    ]
2410                    .concat(),
2411                    conditions: vec![],
2412                    emit: true,
2413                },
2414                SerializableHandlerSpec {
2415                    source: source("pump::RoundState", true),
2416                    key_resolution: KeyResolutionStrategy::Lookup {
2417                        primary_field: FieldPath::new(&["round"]),
2418                    },
2419                    mappings: timestamped(),
2420                    conditions: vec![],
2421                    emit: true,
2422                },
2423                SerializableHandlerSpec {
2424                    source: source("pump::OpenRoundIxState", false),
2425                    key_resolution: KeyResolutionStrategy::Embedded {
2426                        primary_field: FieldPath::new(&["accounts", "mint"]),
2427                    },
2428                    mappings: vec![
2429                        mapping(
2430                            "id.mint",
2431                            &["accounts", "mint"],
2432                            PopulationStrategy::SetOnce,
2433                        ),
2434                        mapping(
2435                            "id.round",
2436                            &["accounts", "round"],
2437                            PopulationStrategy::SetOnce,
2438                        ),
2439                    ],
2440                    conditions: vec![],
2441                    emit: true,
2442                },
2443                SerializableHandlerSpec {
2444                    source: source("pump::LinkIxState", false),
2445                    key_resolution: KeyResolutionStrategy::Embedded {
2446                        primary_field: FieldPath::new(&["accounts", "mint"]),
2447                    },
2448                    mappings: vec![
2449                        mapping(
2450                            "id.mint",
2451                            &["accounts", "mint"],
2452                            PopulationStrategy::SetOnce,
2453                        ),
2454                        mapping(
2455                            "id.bonding_curve",
2456                            &["accounts", "bonding_curve"],
2457                            PopulationStrategy::SetOnce,
2458                        ),
2459                    ],
2460                    conditions: vec![],
2461                    emit: true,
2462                },
2463                SerializableHandlerSpec {
2464                    source: source("pump::CreateIxState", false),
2465                    key_resolution: KeyResolutionStrategy::Embedded {
2466                        primary_field: FieldPath::new(&["accounts", "mint"]),
2467                    },
2468                    mappings: vec![
2469                        mapping(
2470                            "id.mint",
2471                            &["accounts", "mint"],
2472                            PopulationStrategy::SetOnce,
2473                        ),
2474                        mapping(
2475                            "id.creator",
2476                            &["accounts", "user"],
2477                            PopulationStrategy::SetOnce,
2478                        ),
2479                    ],
2480                    conditions: vec![],
2481                    emit: true,
2482                },
2483            ],
2484            sections: vec![],
2485            field_mappings: BTreeMap::new(),
2486            resolver_hooks: vec![],
2487            instruction_hooks: vec![InstructionHook {
2488                instruction_type: "pump::BuyIxState".to_string(),
2489                actions: vec![HookAction::RegisterPdaMapping {
2490                    pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2491                    seed_field: FieldPath::new(&["accounts", "user"]),
2492                    lookup_name: "default_pda_lookup".to_string(),
2493                }],
2494                lookup_by: None,
2495            }],
2496            resolver_specs: vec![],
2497            computed_fields: vec![],
2498            computed_field_specs: vec![],
2499            content_hash: None,
2500            views: vec![],
2501        })
2502    }
2503
2504    /// A timed account update that waits for its key replays with its own
2505    /// time: the event, the capture and the `__timestamp` field all carry the
2506    /// update's timestamp, not the flush time or the flushing instruction's.
2507    /// The update waits for its lookup index entry (`LinkIxState`), for its
2508    /// PDA mapping (`BuyIxState`), for the index entry of its already mapped
2509    /// PDA's seed (`CreateIxState`), or for a numeric index entry, which is
2510    /// no PDA (`OpenRoundIxState`).
2511    #[test]
2512    fn a_queued_account_update_replays_with_its_own_timestamp() {
2513        use crate::vm::UpdateContext;
2514
2515        let accounts = json!({
2516            "bonding_curve": "bonding_curve_1",
2517            "mint": "mint_1",
2518            "user": "user_1",
2519            "round": 7,
2520        });
2521        let curve = (
2522            "pump::BondingCurveState",
2523            json!({ "__account_address": "bonding_curve_1", "virtual_token_reserves": 42 }),
2524        );
2525        let round = (
2526            "pump::RoundState",
2527            json!({ "__account_address": "round_1", "round": 7, "virtual_token_reserves": 42 }),
2528        );
2529        let instruction = |vm: &mut VmContext, bytecode: &MultiEntityBytecode, name: &str| {
2530            let mut context =
2531                UpdateContext::new_instruction(20, "instruction-signature".to_string(), 0);
2532            context.timestamp = Some(2_000);
2533            vm.process_event(
2534                bytecode,
2535                json!({ "accounts": accounts }),
2536                name,
2537                Some(&context),
2538                None,
2539            )
2540            .unwrap();
2541        };
2542        // (instructions before the account update, the account update,
2543        // instructions after it)
2544        let scenarios: [(&[&str], _, &[&str]); 4] = [
2545            (&[], &curve, &["pump::LinkIxState"]),
2546            (&[], &curve, &["pump::CreateIxState", "pump::BuyIxState"]),
2547            (&["pump::BuyIxState"], &curve, &["pump::CreateIxState"]),
2548            (&[], &round, &["pump::OpenRoundIxState"]),
2549        ];
2550        for (before, (account_type, account), after) in scenarios {
2551            let bytecode = MultiEntityBytecode::from_single(
2552                "PumpfunToken".to_string(),
2553                timed_bonding_curve_spec(),
2554                0,
2555            );
2556            let mut vm = VmContext::new();
2557            for name in before {
2558                instruction(&mut vm, &bytecode, name);
2559            }
2560            let mut account_context =
2561                UpdateContext::new_account(10, "curve-signature".to_string(), 1);
2562            account_context.timestamp = Some(1_000);
2563
2564            let queued = vm
2565                .process_event(
2566                    &bytecode,
2567                    account.clone(),
2568                    account_type,
2569                    Some(&account_context),
2570                    None,
2571                )
2572                .unwrap();
2573            assert!(queued.is_empty(), "{after:?}: {queued:?}");
2574
2575            // Flush after a wall-clock second has passed, under instructions
2576            // with a later timestamp of their own.
2577            std::thread::sleep(std::time::Duration::from_millis(1_100));
2578            for name in after {
2579                instruction(&mut vm, &bytecode, name);
2580            }
2581
2582            let token = vm
2583                .get_entity_state(0, &json!("mint_1"))
2584                .unwrap_or_else(|| panic!("{after:?}: no entity"));
2585            for path in [
2586                "/events/reserves/timestamp",
2587                "/snapshots/curve/timestamp",
2588                "/state/seen_at",
2589            ] {
2590                assert_eq!(
2591                    token.pointer(path),
2592                    Some(&json!(1_000)),
2593                    "{before:?} {after:?} {path}: {token:#}"
2594                );
2595            }
2596            assert_eq!(
2597                token.pointer("/events/reserves/data/virtual_token_reserves"),
2598                Some(&json!(42))
2599            );
2600            assert_eq!(token.pointer("/events/reserves/slot"), Some(&json!(10)));
2601            assert_eq!(
2602                token.pointer("/snapshots/curve/signature"),
2603                Some(&json!("curve-signature"))
2604            );
2605        }
2606    }
2607
2608    fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
2609        TypedStreamSpec::from_serializable(SerializableStreamSpec {
2610            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2611            state_name: "OreRound".to_string(),
2612            program_id: None,
2613            idl: None,
2614            identity: IdentitySpec {
2615                primary_keys: vec!["id.round_id".to_string()],
2616                lookup_indexes: vec![LookupIndexSpec {
2617                    field_name: "id.round_address".to_string(),
2618                    temporal_field: None,
2619                }],
2620            },
2621            handlers: vec![SerializableHandlerSpec {
2622                source: SourceSpec::Source {
2623                    program_id: None,
2624                    discriminator: None,
2625                    type_name: "entropy::VarState".to_string(),
2626                    serialization: None,
2627                    is_account: true,
2628                },
2629                key_resolution: KeyResolutionStrategy::Lookup {
2630                    primary_field: FieldPath::new(&["__account_address"]),
2631                },
2632                mappings: vec![mapping(
2633                    "state.expires_at",
2634                    &["end_at"],
2635                    PopulationStrategy::LastWrite,
2636                )],
2637                conditions: vec![],
2638                emit: true,
2639            }],
2640            sections: vec![],
2641            field_mappings: BTreeMap::new(),
2642            resolver_hooks: vec![],
2643            instruction_hooks: vec![InstructionHook {
2644                instruction_type: "ore::DeployIxState".to_string(),
2645                actions: vec![HookAction::RegisterPdaMapping {
2646                    pda_field: FieldPath::new(&["accounts", "entropyVar"]),
2647                    seed_field: FieldPath::new(&["accounts", "round"]),
2648                    lookup_name: "default_pda_lookup".to_string(),
2649                }],
2650                lookup_by: None,
2651            }],
2652            resolver_specs: vec![],
2653            computed_fields: vec![],
2654            computed_field_specs: vec![],
2655            content_hash: None,
2656            views: vec![],
2657        })
2658    }
2659
2660    fn embedded_account_key_spec() -> TypedStreamSpec<Value> {
2661        TypedStreamSpec::from_serializable(SerializableStreamSpec {
2662            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2663            state_name: "OreTreasury".to_string(),
2664            program_id: None,
2665            idl: None,
2666            identity: IdentitySpec {
2667                primary_keys: vec!["id.address".to_string()],
2668                lookup_indexes: vec![],
2669            },
2670            handlers: vec![SerializableHandlerSpec {
2671                source: SourceSpec::Source {
2672                    program_id: None,
2673                    discriminator: None,
2674                    type_name: "ore::TreasuryState".to_string(),
2675                    serialization: None,
2676                    is_account: true,
2677                },
2678                key_resolution: KeyResolutionStrategy::Embedded {
2679                    primary_field: FieldPath::new(&["__account_address"]),
2680                },
2681                mappings: vec![mapping(
2682                    "id.address",
2683                    &["__account_address"],
2684                    PopulationStrategy::SetOnce,
2685                )],
2686                conditions: vec![],
2687                emit: true,
2688            }],
2689            sections: vec![],
2690            field_mappings: BTreeMap::new(),
2691            resolver_hooks: vec![],
2692            instruction_hooks: vec![],
2693            resolver_specs: vec![],
2694            computed_fields: vec![],
2695            computed_field_specs: vec![],
2696            content_hash: None,
2697            views: vec![],
2698        })
2699    }
2700
2701    /// A pool whose account carries a price (replaced), a fill (appended) and
2702    /// a secret that is mapped but never emitted, plus a config account whose
2703    /// handler emits nothing and sets a fee.
2704    fn whole_entity_pool_spec() -> TypedStreamSpec<Value> {
2705        let mut hidden = mapping("state.secret", &["secret"], PopulationStrategy::LastWrite);
2706        hidden.emit = false;
2707        let handler = |type_name: &str, mappings, emit| SerializableHandlerSpec {
2708            source: SourceSpec::Source {
2709                program_id: None,
2710                discriminator: None,
2711                type_name: type_name.to_string(),
2712                serialization: None,
2713                is_account: true,
2714            },
2715            key_resolution: KeyResolutionStrategy::Embedded {
2716                primary_field: FieldPath::new(&["__account_address"]),
2717            },
2718            mappings,
2719            conditions: vec![],
2720            emit,
2721        };
2722        TypedStreamSpec::from_serializable(SerializableStreamSpec {
2723            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2724            state_name: "Pool".to_string(),
2725            program_id: None,
2726            idl: None,
2727            identity: IdentitySpec {
2728                primary_keys: vec!["id.address".to_string()],
2729                lookup_indexes: vec![],
2730            },
2731            handlers: vec![
2732                handler(
2733                    "amm::PoolState",
2734                    vec![
2735                        mapping(
2736                            "id.address",
2737                            &["__account_address"],
2738                            PopulationStrategy::SetOnce,
2739                        ),
2740                        mapping("state.price", &["price"], PopulationStrategy::LastWrite),
2741                        mapping("state.fills", &["fill"], PopulationStrategy::Append),
2742                        hidden,
2743                    ],
2744                    true,
2745                ),
2746                handler(
2747                    "amm::PoolConfigState",
2748                    vec![mapping(
2749                        "config.fee",
2750                        &["fee"],
2751                        PopulationStrategy::LastWrite,
2752                    )],
2753                    false,
2754                ),
2755            ],
2756            sections: vec![],
2757            field_mappings: BTreeMap::new(),
2758            resolver_hooks: vec![],
2759            instruction_hooks: vec![],
2760            resolver_specs: vec![],
2761            computed_fields: vec![],
2762            computed_field_specs: vec![],
2763            content_hash: None,
2764            views: vec![],
2765        })
2766    }
2767
2768    fn pool_bytecode() -> MultiEntityBytecode {
2769        MultiEntityBytecode::from_single("Pool".to_string(), whole_entity_pool_spec(), 0)
2770    }
2771
2772    /// Event (`#[event]`), capture (`#[snapshot]`) and context (`__timestamp`)
2773    /// mappings compiled from a spec are stamped with the update's timestamp,
2774    /// so replaying an update reproduces them.
2775    #[test]
2776    fn compiled_event_capture_and_context_mappings_carry_the_update_timestamp() {
2777        let with_source = |target: &str, source: MappingSource| SerializableFieldMapping {
2778            source,
2779            ..mapping(target, &[], PopulationStrategy::LastWrite)
2780        };
2781        let spec: TypedStreamSpec<Value> =
2782            TypedStreamSpec::from_serializable(SerializableStreamSpec {
2783                ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2784                state_name: "Pool".to_string(),
2785                program_id: None,
2786                idl: None,
2787                identity: IdentitySpec {
2788                    primary_keys: vec!["id.address".to_string()],
2789                    lookup_indexes: vec![],
2790                },
2791                handlers: vec![SerializableHandlerSpec {
2792                    source: SourceSpec::Source {
2793                        program_id: None,
2794                        discriminator: None,
2795                        type_name: "amm::PoolState".to_string(),
2796                        serialization: None,
2797                        is_account: true,
2798                    },
2799                    key_resolution: KeyResolutionStrategy::Embedded {
2800                        primary_field: FieldPath::new(&["__account_address"]),
2801                    },
2802                    mappings: vec![
2803                        mapping(
2804                            "id.address",
2805                            &["__account_address"],
2806                            PopulationStrategy::SetOnce,
2807                        ),
2808                        with_source(
2809                            "events.priced",
2810                            MappingSource::AsEvent {
2811                                fields: vec![Box::new(MappingSource::FromSource {
2812                                    path: FieldPath::new(&["price"]),
2813                                    default: None,
2814                                    transform: None,
2815                                })],
2816                            },
2817                        ),
2818                        with_source(
2819                            "snapshots.pool",
2820                            MappingSource::AsCapture {
2821                                field_transforms: BTreeMap::new(),
2822                            },
2823                        ),
2824                        with_source(
2825                            "state.updated_at",
2826                            MappingSource::FromContext {
2827                                field: "timestamp".to_string(),
2828                            },
2829                        ),
2830                    ],
2831                    conditions: vec![],
2832                    emit: true,
2833                }],
2834                sections: vec![],
2835                field_mappings: BTreeMap::new(),
2836                resolver_hooks: vec![],
2837                instruction_hooks: vec![],
2838                resolver_specs: vec![],
2839                computed_fields: vec![],
2840                computed_field_specs: vec![],
2841                content_hash: None,
2842                views: vec![],
2843            });
2844        let bytecode = MultiEntityBytecode::from_single("Pool".to_string(), spec, 0);
2845        let context = crate::vm::UpdateContext::with_timestamp(7, "sig-7".to_string(), 1_234_567);
2846
2847        let mut vm = VmContext::new();
2848        vm.process_event(
2849            &bytecode,
2850            pool_event(100, 1),
2851            "amm::PoolState",
2852            Some(&context),
2853            None,
2854        )
2855        .unwrap();
2856
2857        let pool = vm.get_entity_state(0, &json!("pool_1")).unwrap();
2858        for path in [
2859            "/events/priced/timestamp",
2860            "/snapshots/pool/timestamp",
2861            "/state/updated_at",
2862        ] {
2863            assert_eq!(
2864                pool.pointer(path),
2865                Some(&json!(1_234_567)),
2866                "{path}: {pool:#}"
2867            );
2868        }
2869        assert_eq!(pool.pointer("/events/priced/data/price"), Some(&json!(100)));
2870        assert_eq!(pool.pointer("/snapshots/pool/slot"), Some(&json!(7)));
2871    }
2872
2873    fn pool_event(price: u64, fill: u64) -> Value {
2874        pool_event_for("pool_1", price, fill)
2875    }
2876
2877    fn pool_event_for(address: &str, price: u64, fill: u64) -> Value {
2878        json!({"__account_address": address, "price": price, "fill": fill, "secret": "s"})
2879    }
2880
2881    /// A VM that answers whole-entity requests, and so marks creations.
2882    fn linked_vm(
2883        state_config: crate::vm::StateTableConfig,
2884    ) -> (VmContext, crate::vm::WholeEntityRequests) {
2885        let requests = crate::vm::WholeEntityRequests::new(16);
2886        let mut vm = VmContext::new_with_config(state_config);
2887        vm.set_whole_entity_requests(requests.clone());
2888        (vm, requests)
2889    }
2890
2891    fn process(vm: &mut VmContext, bytecode: &MultiEntityBytecode, event: Value) -> Vec<Mutation> {
2892        vm.process_event(bytecode, event, "amm::PoolState", None, None)
2893            .unwrap()
2894    }
2895
2896    #[test]
2897    fn source_deletion_cancels_resync_and_recreates_a_fresh_vm_row() {
2898        let bytecode = pool_bytecode();
2899        let (mut vm, requests) = linked_vm(Default::default());
2900        let first = process(&mut vm, &bytecode, pool_event(1, 10));
2901        let key = &first[0].key;
2902        let target = crate::vm::ResolverTarget {
2903            state_id: 0,
2904            entity_name: "Pool".into(),
2905            primary_key: key.clone(),
2906            extracts: vec![],
2907        };
2908        vm.enqueue_resolver_request(
2909            "ignored".into(),
2910            crate::ast::ResolverType::Token,
2911            json!("mint"),
2912            target.clone(),
2913        );
2914        let old_request = vm.take_resolver_requests().pop().unwrap();
2915        vm.get_state_table_mut(0).unwrap().deferred_when_ops.insert(
2916            ("sig".into(), "instruction".into()),
2917            vec![crate::vm::DeferredWhenOperation {
2918                entity_name: "Pool".into(),
2919                primary_key: key.clone(),
2920                field_path: "old".into(),
2921                field_value: json!(true),
2922                when_instruction: "instruction".into(),
2923                signature: "sig".into(),
2924                slot: 1,
2925                deferred_at: 1,
2926                emit: true,
2927            }],
2928        );
2929        requests.request("Pool", key);
2930        let deletion = vm.delete_entity(&bytecode, "Pool", key).unwrap();
2931        assert!(deletion.is_delete());
2932        assert!(
2933            serde_json::from_value::<Mutation>(serde_json::to_value(&deletion).unwrap())
2934                .unwrap()
2935                .is_delete()
2936        );
2937        assert!(!requests.is_requested("Pool", key));
2938        assert!(vm.get_entity_state(0, key).is_none());
2939        assert!(vm.snapshot_state_table(0).is_empty());
2940        assert!(vm
2941            .get_state_table_mut(0)
2942            .unwrap()
2943            .deferred_when_ops
2944            .is_empty());
2945        assert!(vm.delete_entity(&bytecode, "Unknown", key).is_none());
2946        let mut next = process(&mut vm, &bytecode, pool_event(9, 90));
2947        vm.enqueue_resolver_request(
2948            "ignored".into(),
2949            crate::ast::ResolverType::Token,
2950            json!("mint"),
2951            target,
2952        );
2953        let new_request = vm.take_resolver_requests().pop().unwrap();
2954        assert_ne!(old_request.cache_key, new_request.cache_key);
2955        assert!(vm
2956            .apply_resolver_result(&bytecode, &old_request.cache_key, json!({"stale":true}))
2957            .unwrap()
2958            .is_empty());
2959        assert_eq!(next[0].take_whole_entity_mark(), Some(WholeEntity::Created));
2960        assert_eq!(next[0].patch["state"]["fills"], json!([90]));
2961    }
2962
2963    #[test]
2964    fn deletion_preserves_unrelated_resolver_sharing_and_isolates_recreation() {
2965        let bytecode = pool_bytecode();
2966        let (mut vm, _) = linked_vm(Default::default());
2967        let key = process(&mut vm, &bytecode, pool_event(1, 10))[0]
2968            .key
2969            .clone();
2970        let late_joiner = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20))[0]
2971            .key
2972            .clone();
2973        let target = |primary_key| crate::vm::ResolverTarget {
2974            state_id: 0,
2975            entity_name: "Pool".into(),
2976            primary_key,
2977            extracts: vec![crate::ast::ResolverExtractSpec {
2978                target_path: "resolved".into(),
2979                source_path: Some("value".into()),
2980                transform: None,
2981            }],
2982        };
2983        let enqueue = |vm: &mut VmContext, key| {
2984            vm.enqueue_resolver_request(
2985                "ignored".into(),
2986                crate::ast::ResolverType::Token,
2987                json!("shared-mint"),
2988                target(key),
2989            )
2990        };
2991        enqueue(&mut vm, key.clone());
2992        enqueue(&mut vm, json!("unrelated"));
2993        let old = vm.take_resolver_requests();
2994        assert_eq!(old.len(), 1);
2995        vm.delete_entity(&bytecode, "Pool", &key).unwrap();
2996        enqueue(&mut vm, json!("unrelated"));
2997        enqueue(&mut vm, json!("another-unrelated"));
2998        assert!(
2999            vm.take_resolver_requests().is_empty(),
3000            "unaffected targets reuse the old pending call"
3001        );
3002        enqueue(&mut vm, key.clone());
3003        let fresh = vm.take_resolver_requests();
3004        assert_eq!(fresh.len(), 1);
3005        assert_ne!(fresh[0].cache_key, old[0].cache_key);
3006        // This entity never joined the old shared call. Deleting and
3007        // recreating it must still keep it off that pre-deletion result.
3008        vm.delete_entity(&bytecode, "Pool", &late_joiner).unwrap();
3009        process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
3010        enqueue(&mut vm, late_joiner.clone());
3011        let late_joiner_fresh = vm.take_resolver_requests();
3012        assert_eq!(late_joiner_fresh.len(), 1);
3013        assert_ne!(late_joiner_fresh[0].cache_key, old[0].cache_key);
3014        enqueue(&mut vm, json!("unrelated"));
3015        enqueue(&mut vm, json!("another-unrelated"));
3016        enqueue(&mut vm, key.clone());
3017        assert!(
3018            vm.take_resolver_requests().is_empty(),
3019            "unaffected targets reuse pending calls"
3020        );
3021        let old_mutations = vm
3022            .apply_resolver_result(&bytecode, &old[0].cache_key, json!({"value":"old"}))
3023            .unwrap();
3024        assert_eq!(old_mutations.len(), 2);
3025        assert!(old_mutations.iter().all(|mutation| mutation.key != key));
3026        assert!(vm.get_entity_state(0, &key).is_none());
3027        let fresh_mutations = vm
3028            .apply_resolver_result(&bytecode, &fresh[0].cache_key, json!({"value":"fresh"}))
3029            .unwrap();
3030        assert_eq!(fresh_mutations.len(), 1);
3031        assert_eq!(fresh_mutations[0].key, key);
3032        let late_joiner_mutations = vm
3033            .apply_resolver_result(
3034                &bytecode,
3035                &late_joiner_fresh[0].cache_key,
3036                json!({"value":"fresh-late-joiner"}),
3037            )
3038            .unwrap();
3039        assert_eq!(late_joiner_mutations.len(), 1);
3040        assert_eq!(late_joiner_mutations[0].key, late_joiner);
3041    }
3042
3043    #[test]
3044    fn a_requested_entity_is_emitted_whole_once() {
3045        let bytecode = pool_bytecode();
3046        let (mut vm, requests) = linked_vm(Default::default());
3047        let mut update = |price, fill| process(&mut vm, &bytecode, pool_event(price, fill));
3048
3049        let first = update(1, 10);
3050        assert_eq!(first[0].whole_entity_mark(), Some(WholeEntity::Created));
3051        let key = first[0].key.clone();
3052        let second = update(2, 20);
3053        assert_eq!(second.len(), 1);
3054        assert!(!second[0].is_whole_entity());
3055        assert_eq!(
3056            second[0].patch,
3057            json!({"state": {"price": 2, "fills": [20]}})
3058        );
3059        assert_eq!(second[0].append, vec!["state.fills".to_string()]);
3060
3061        // A consumer that dropped the entity asks for all of it.
3062        requests.request("Pool", &key);
3063        assert!(requests.is_requested("Pool", &key));
3064        let mut third = update(3, 30);
3065        assert_eq!(third.len(), 2);
3066        assert!(
3067            !third[0].is_whole_entity(),
3068            "the change itself is untouched"
3069        );
3070        assert_eq!(
3071            third[0].patch,
3072            json!({"state": {"price": 3, "fills": [30]}})
3073        );
3074        assert_eq!(third[1].key, key);
3075        assert_eq!(third[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
3076        assert_eq!(
3077            third[1].patch,
3078            json!({
3079                "id": {"address": "pool_1"},
3080                "state": {"price": 3, "fills": [10, 20, 30]},
3081            }),
3082            "the whole entity, without its non-emitted field"
3083        );
3084        assert!(
3085            third[1].append.is_empty(),
3086            "its arrays are whole, not appended to"
3087        );
3088        assert!(requests.is_empty(), "a request is answered once");
3089
3090        let fourth = update(4, 40);
3091        assert_eq!(fourth.len(), 1);
3092        assert!(!fourth[0].is_whole_entity());
3093        assert_eq!(
3094            fourth[0].patch,
3095            json!({"state": {"price": 4, "fills": [40]}})
3096        );
3097    }
3098
3099    /// A request does not wait for its entity to change: the VM's next call
3100    /// ends with the whole entity, after whatever that call emitted.
3101    #[test]
3102    fn a_request_is_answered_by_the_next_call_whatever_it_changes() {
3103        let bytecode = pool_bytecode();
3104        let (mut vm, requests) = linked_vm(Default::default());
3105        process(&mut vm, &bytecode, pool_event_for("pool_1", 1, 10));
3106        process(&mut vm, &bytecode, pool_event_for("pool_2", 1, 10));
3107
3108        requests.request("Pool", &json!("pool_1"));
3109        let mut next = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20));
3110        assert_eq!(next.len(), 2);
3111        assert_eq!(next[0].key, json!("pool_2"));
3112        assert!(!next[0].is_whole_entity());
3113        assert_eq!(next[1].key, json!("pool_1"));
3114        assert_eq!(next[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
3115        assert_eq!(
3116            next[1].patch,
3117            json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
3118        );
3119        assert!(requests.is_empty());
3120
3121        // Even a call that changes nothing answers.
3122        requests.request("Pool", &json!("pool_2"));
3123        let answered = vm
3124            .apply_resolver_result(&bytecode, "no-such-request", json!({}))
3125            .unwrap();
3126        assert_eq!(answered.len(), 1);
3127        assert_eq!(answered[0].key, json!("pool_2"));
3128        assert_eq!(answered[0].whole_entity_mark(), Some(WholeEntity::Resent));
3129
3130        let quiet = process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
3131        assert_eq!(quiet.len(), 1, "a request is answered once");
3132    }
3133
3134    /// The VM has nothing to send for an entity it does not hold. It dropped
3135    /// the entity too, so the key's next mutation creates it again.
3136    #[test]
3137    fn a_request_for_an_entity_the_vm_does_not_hold_is_dropped() {
3138        let bytecode = pool_bytecode();
3139        let (mut vm, requests) = linked_vm(Default::default());
3140        requests.request("Pool", &json!("other_pool"));
3141        let mutations = process(&mut vm, &bytecode, pool_event(1, 10));
3142        assert_eq!(mutations.len(), 1);
3143        assert_eq!(mutations[0].whole_entity_mark(), Some(WholeEntity::Created));
3144        assert!(requests.is_empty(), "dropped, not kept for later");
3145
3146        let created = process(&mut vm, &bytecode, pool_event_for("other_pool", 1, 10));
3147        assert_eq!(created.len(), 1);
3148        assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
3149    }
3150
3151    #[test]
3152    fn whole_entity_requests_are_bounded() {
3153        let requests = crate::vm::WholeEntityRequests::new(2);
3154        for key in ["a", "b", "c"] {
3155            requests.request("Pool", &json!(key));
3156        }
3157        assert_eq!(requests.len(), 2);
3158        assert!(
3159            !requests.is_requested("Pool", &json!("a")),
3160            "the oldest is forgotten"
3161        );
3162        assert!(requests.is_requested("Pool", &json!("c")));
3163    }
3164
3165    /// A consumer that does not hold a key can tell a new entity from a later
3166    /// patch for one it dropped: only the first is marked created.
3167    #[test]
3168    fn an_entitys_first_mutation_is_marked_as_its_creation() {
3169        let bytecode = pool_bytecode();
3170        let (mut vm, _requests) = linked_vm(Default::default());
3171
3172        let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
3173        assert_eq!(first.len(), 1, "the creation needs no second mutation");
3174        assert_eq!(
3175            first[0].take_whole_entity_mark(),
3176            Some(WholeEntity::Created)
3177        );
3178        assert_eq!(
3179            first[0].patch,
3180            json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}}),
3181            "the whole row, without its non-emitted field"
3182        );
3183        assert_eq!(first[0].append, vec!["state.fills".to_string()]);
3184
3185        let second = process(&mut vm, &bytecode, pool_event(2, 20));
3186        assert_eq!(second[0].whole_entity_mark(), None);
3187        assert_eq!(
3188            second[0].patch,
3189            json!({"state": {"price": 2, "fills": [20]}})
3190        );
3191    }
3192
3193    /// Without a consumer asking for whole entities the VM's output is
3194    /// unchanged: nothing is marked, and a creation is its changed fields.
3195    #[test]
3196    fn a_vm_without_whole_entity_requests_marks_no_creation() {
3197        let bytecode = pool_bytecode();
3198        let mut vm = VmContext::new();
3199        let first = process(&mut vm, &bytecode, pool_event(1, 10));
3200        assert!(!first[0].is_whole_entity());
3201        assert!(first[0].patch.get(crate::WHOLE_ENTITY_MARKER).is_none());
3202    }
3203
3204    /// A handler that emits nothing can start the row. The first mutation
3205    /// after it is still the creation, and carries what that handler set,
3206    /// which its own change does not.
3207    #[test]
3208    fn a_row_started_without_a_mutation_is_created_by_its_first_one() {
3209        let bytecode = pool_bytecode();
3210        let (mut vm, _requests) = linked_vm(Default::default());
3211        let config = vm
3212            .process_event(
3213                &bytecode,
3214                json!({"__account_address": "pool_1", "fee": 30}),
3215                "amm::PoolConfigState",
3216                None,
3217                None,
3218            )
3219            .unwrap();
3220        assert!(config.is_empty(), "the config handler emits nothing");
3221
3222        let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
3223        assert_eq!(
3224            first[0].take_whole_entity_mark(),
3225            Some(WholeEntity::Created)
3226        );
3227        assert_eq!(
3228            first[0].patch,
3229            json!({
3230                "id": {"address": "pool_1"},
3231                "config": {"fee": 30},
3232                "state": {"price": 1, "fills": [10]},
3233            })
3234        );
3235        let second = process(&mut vm, &bytecode, pool_event(2, 20));
3236        assert!(!second[0].is_whole_entity());
3237    }
3238
3239    /// An entity the VM's bounded table dropped starts again from its next
3240    /// event, and that event's mutation creates it again.
3241    #[test]
3242    fn an_entity_the_vm_dropped_is_created_again() {
3243        let bytecode = pool_bytecode();
3244        let (mut vm, _requests) = linked_vm(crate::vm::StateTableConfig {
3245            max_entries: 2,
3246            ..Default::default()
3247        });
3248        for pool in ["a", "b", "c"] {
3249            let created = process(&mut vm, &bytecode, pool_event_for(pool, 1, 1));
3250            assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
3251        }
3252        assert!(vm.get_entity_state(0, &json!("a")).is_none(), "dropped");
3253
3254        let mut again = process(&mut vm, &bytecode, pool_event_for("a", 2, 2));
3255        assert_eq!(
3256            again[0].take_whole_entity_mark(),
3257            Some(WholeEntity::Created)
3258        );
3259        assert_eq!(
3260            again[0].patch,
3261            json!({"id": {"address": "a"}, "state": {"price": 2, "fills": [2]}})
3262        );
3263        let changed = process(&mut vm, &bytecode, pool_event_for("c", 2, 2));
3264        assert!(!changed[0].is_whole_entity(), "c was never dropped");
3265    }
3266
3267    /// A row a non-emitting handler started and a snapshot then took before
3268    /// its first mutation comes back as emitted: its first update after the
3269    /// restore is a plain patch. A consumer that never held the key asks for
3270    /// it, and the VM's next call sends it whole, with what that handler set.
3271    #[test]
3272    fn a_restored_row_never_emitted_is_sent_whole_on_request() {
3273        let bytecode = pool_bytecode();
3274        let (mut vm, _requests) = linked_vm(Default::default());
3275        let config = vm
3276            .process_event(
3277                &bytecode,
3278                json!({"__account_address": "pool_1", "fee": 30}),
3279                "amm::PoolConfigState",
3280                None,
3281                None,
3282            )
3283            .unwrap();
3284        assert!(config.is_empty());
3285        let snapshot = vm.dump();
3286
3287        let (mut restored, requests) = linked_vm(Default::default());
3288        restored.hydrate(snapshot);
3289        let first = process(&mut restored, &bytecode, pool_event(1, 10));
3290        assert!(!first[0].is_whole_entity(), "counted as emitted already");
3291        assert_eq!(
3292            first[0].patch,
3293            json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
3294        );
3295
3296        requests.request("Pool", &first[0].key);
3297        let next = process(&mut restored, &bytecode, pool_event_for("pool_2", 1, 1));
3298        let resent = next
3299            .iter()
3300            .find(|mutation| mutation.key == json!("pool_1"))
3301            .expect("the requested entity follows the next call");
3302        assert_eq!(resent.whole_entity_mark(), Some(WholeEntity::Resent));
3303        let mut resent = resent.clone();
3304        resent.take_whole_entity_mark();
3305        assert_eq!(
3306            resent.patch,
3307            json!({
3308                "id": {"address": "pool_1"},
3309                "config": {"fee": 30},
3310                "state": {"price": 1, "fills": [10]},
3311            })
3312        );
3313    }
3314
3315    /// A resolver result for an entity the table does not hold starts the
3316    /// entity, so its mutation is the creation; one for an entity the table
3317    /// holds is a patch.
3318    #[test]
3319    fn a_resolver_result_for_an_entity_not_held_creates_it() {
3320        let bytecode = pool_bytecode();
3321        let (mut vm, _requests) = linked_vm(Default::default());
3322        process(&mut vm, &bytecode, pool_event(1, 10));
3323        for key in ["pool_1", "pool_2"] {
3324            vm.enqueue_resolver_request(
3325                String::new(),
3326                crate::ast::ResolverType::Token,
3327                json!(key),
3328                crate::vm::ResolverTarget {
3329                    state_id: 0,
3330                    entity_name: "Pool".to_string(),
3331                    primary_key: json!(key),
3332                    extracts: vec![crate::ast::ResolverExtractSpec {
3333                        target_path: "meta.name".to_string(),
3334                        source_path: Some("name".to_string()),
3335                        transform: None,
3336                    }],
3337                },
3338            );
3339        }
3340        let mut applied = Vec::new();
3341        for request in vm.take_resolver_requests() {
3342            applied.extend(
3343                vm.apply_resolver_result(&bytecode, &request.cache_key, json!({"name": "n"}))
3344                    .unwrap(),
3345            );
3346        }
3347        applied.sort_by_key(|mutation| mutation.key.to_string());
3348
3349        assert_eq!(applied[0].key, json!("pool_1"));
3350        assert!(!applied[0].is_whole_entity());
3351        assert_eq!(applied[0].patch, json!({"meta": {"name": "n"}}));
3352        assert_eq!(applied[1].key, json!("pool_2"));
3353        assert_eq!(
3354            applied[1].take_whole_entity_mark(),
3355            Some(WholeEntity::Created)
3356        );
3357        assert_eq!(applied[1].patch, json!({"meta": {"name": "n"}}));
3358    }
3359
3360    /// Entities restored from a snapshot may be held by the consumer already,
3361    /// so their next mutation is a patch like any other.
3362    #[test]
3363    fn a_restored_entity_is_not_created_again() {
3364        let bytecode = pool_bytecode();
3365        let (mut vm, _requests) = linked_vm(Default::default());
3366        process(&mut vm, &bytecode, pool_event(1, 10));
3367        let snapshot = vm.dump();
3368
3369        let (mut restored, _requests) = linked_vm(Default::default());
3370        restored.hydrate(snapshot);
3371        let next = process(&mut restored, &bytecode, pool_event(2, 20));
3372        assert!(!next[0].is_whole_entity());
3373        assert_eq!(next[0].patch, json!({"state": {"price": 2, "fills": [20]}}));
3374    }
3375
3376    #[test]
3377    fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
3378        let bytecode = MultiEntityBytecode::from_single(
3379            "PumpfunToken".to_string(),
3380            direct_pda_to_primary_key_spec(),
3381            0,
3382        );
3383        let mut vm = VmContext::new();
3384
3385        let mutations = vm
3386            .process_event(
3387                &bytecode,
3388                json!({
3389                    "__account_address": "bonding_curve_1",
3390                    "__resolved_primary_key": "mint_1",
3391                    "virtual_token_reserves": 42,
3392                }),
3393                "pump::BondingCurveState",
3394                None,
3395                None,
3396            )
3397            .unwrap();
3398
3399        assert_eq!(mutations.len(), 1);
3400        assert_eq!(mutations[0].key, json!("mint_1"));
3401        assert_eq!(
3402            mutations[0].patch["id"]["bonding_curve"],
3403            json!("bonding_curve_1")
3404        );
3405        assert_eq!(
3406            mutations[0].patch["reserves"]["virtual_token_reserves"],
3407            json!(42)
3408        );
3409    }
3410
3411    #[test]
3412    fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
3413        let bytecode = MultiEntityBytecode::from_single(
3414            "OreRound".to_string(),
3415            pda_to_intermediate_lookup_spec(),
3416            0,
3417        );
3418        let mut vm = VmContext::new();
3419
3420        let mutations = vm
3421            .process_event(
3422                &bytecode,
3423                json!({
3424                    "__account_address": "entropy_var_1",
3425                    "__resolved_primary_key": "round_address_1",
3426                    "end_at": 123,
3427                }),
3428                "entropy::VarState",
3429                None,
3430                None,
3431            )
3432            .unwrap();
3433
3434        assert!(mutations.is_empty());
3435    }
3436
3437    #[test]
3438    fn embedded_account_key_wins_over_an_unrelated_resolved_key() {
3439        let bytecode = MultiEntityBytecode::from_single(
3440            "OreTreasury".to_string(),
3441            embedded_account_key_spec(),
3442            0,
3443        );
3444        let mut vm = VmContext::new();
3445
3446        let mutations = vm
3447            .process_event(
3448                &bytecode,
3449                json!({
3450                    "__account_address": "treasury_pda",
3451                    "__resolved_primary_key": "round_address",
3452                }),
3453                "ore::TreasuryState",
3454                None,
3455                None,
3456            )
3457            .unwrap();
3458
3459        assert_eq!(mutations.len(), 1);
3460        assert_eq!(mutations[0].key, json!("treasury_pda"));
3461        assert_eq!(mutations[0].patch["id"]["address"], json!("treasury_pda"));
3462    }
3463
3464    fn instruction_handler(
3465        type_name: &str,
3466        key_resolution: KeyResolutionStrategy,
3467        mappings: Vec<SerializableFieldMapping>,
3468    ) -> SerializableHandlerSpec {
3469        SerializableHandlerSpec {
3470            source: SourceSpec::Source {
3471                program_id: None,
3472                discriminator: None,
3473                type_name: type_name.to_string(),
3474                serialization: None,
3475                is_account: false,
3476            },
3477            key_resolution,
3478            mappings,
3479            conditions: vec![],
3480            emit: true,
3481        }
3482    }
3483
3484    /// `split_position` updates two positions: the source (`first_position`)
3485    /// and the child (`second_position`). Mirrors the meteora-damm
3486    /// `MeteoraPosition` shape: one Count per side plus a derive_from hook on
3487    /// the source side.
3488    fn split_position_spec() -> TypedStreamSpec<Value> {
3489        let embedded = |field: &str| KeyResolutionStrategy::Embedded {
3490            primary_field: FieldPath::new(&["accounts", field]),
3491        };
3492        TypedStreamSpec::from_serializable(SerializableStreamSpec {
3493            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3494            state_name: "Position".to_string(),
3495            program_id: None,
3496            idl: None,
3497            identity: IdentitySpec {
3498                primary_keys: vec!["id.position_address".to_string()],
3499                lookup_indexes: vec![],
3500            },
3501            handlers: vec![
3502                instruction_handler(
3503                    "amm::SplitPositionIxState",
3504                    embedded("first_position"),
3505                    vec![
3506                        mapping(
3507                            "id.position_address",
3508                            &["accounts", "first_position"],
3509                            PopulationStrategy::SetOnce,
3510                        ),
3511                        mapping(
3512                            "activity.split_source_count",
3513                            &["data"],
3514                            PopulationStrategy::Count,
3515                        ),
3516                    ],
3517                ),
3518                instruction_handler(
3519                    "amm::SplitPositionIxState",
3520                    embedded("second_position"),
3521                    vec![
3522                        mapping(
3523                            "id.position_address",
3524                            &["accounts", "second_position"],
3525                            PopulationStrategy::SetOnce,
3526                        ),
3527                        mapping(
3528                            "activity.split_child_count",
3529                            &["data"],
3530                            PopulationStrategy::Count,
3531                        ),
3532                    ],
3533                ),
3534            ],
3535            sections: vec![],
3536            field_mappings: BTreeMap::new(),
3537            resolver_hooks: vec![],
3538            instruction_hooks: vec![InstructionHook {
3539                instruction_type: "amm::SplitPositionIxState".to_string(),
3540                actions: vec![HookAction::SetField {
3541                    target_field: "activity.last_split_source_slot".to_string(),
3542                    source: MappingSource::FromSource {
3543                        path: FieldPath::new(&["data", "slot"]),
3544                        default: None,
3545                        transform: None,
3546                    },
3547                    condition: None,
3548                }],
3549                lookup_by: Some(FieldPath::new(&["accounts", "first_position"])),
3550            }],
3551            resolver_specs: vec![],
3552            computed_fields: vec![],
3553            computed_field_specs: vec![],
3554            content_hash: None,
3555            views: vec![],
3556        })
3557    }
3558
3559    fn split_position_event(first: &str, second: &str) -> Value {
3560        json!({
3561            "accounts": { "first_position": first, "second_position": second },
3562            "data": { "slot": 77 },
3563        })
3564    }
3565
3566    #[test]
3567    fn one_instruction_routes_each_key_to_its_own_entity() {
3568        let bytecode =
3569            MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
3570        let mut vm = VmContext::new();
3571
3572        for _ in 0..2 {
3573            vm.process_event(
3574                &bytecode,
3575                split_position_event("source_pos", "child_pos"),
3576                "amm::SplitPositionIxState",
3577                None,
3578                None,
3579            )
3580            .unwrap();
3581        }
3582        vm.process_event(
3583            &bytecode,
3584            split_position_event("child_pos", "grandchild_pos"),
3585            "amm::SplitPositionIxState",
3586            None,
3587            None,
3588        )
3589        .unwrap();
3590
3591        let state = |key: &str| vm.get_entity_state(0, &json!(key)).unwrap();
3592
3593        let source = state("source_pos");
3594        assert_eq!(source["id"]["position_address"], json!("source_pos"));
3595        assert_eq!(source["activity"]["split_source_count"], json!(2));
3596        assert_eq!(source["activity"].get("split_child_count"), None);
3597        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3598
3599        // The child was split into once and split from once.
3600        let child = state("child_pos");
3601        assert_eq!(child["id"]["position_address"], json!("child_pos"));
3602        assert_eq!(child["activity"]["split_child_count"], json!(2));
3603        assert_eq!(child["activity"]["split_source_count"], json!(1));
3604        assert_eq!(child["activity"]["last_split_source_slot"], json!(77));
3605
3606        let grandchild = state("grandchild_pos");
3607        assert_eq!(grandchild["activity"]["split_child_count"], json!(1));
3608        assert_eq!(grandchild["activity"].get("split_source_count"), None);
3609        assert_eq!(grandchild["activity"].get("last_split_source_slot"), None);
3610    }
3611
3612    #[test]
3613    fn one_instruction_emits_one_mutation_per_key() {
3614        let bytecode =
3615            MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
3616        let mut vm = VmContext::new();
3617
3618        let mutations = vm
3619            .process_event(
3620                &bytecode,
3621                split_position_event("source_pos", "child_pos"),
3622                "amm::SplitPositionIxState",
3623                None,
3624                None,
3625            )
3626            .unwrap();
3627
3628        assert_eq!(mutations.len(), 2);
3629        assert_eq!(mutations[0].key, json!("source_pos"));
3630        assert_eq!(
3631            mutations[0].patch["activity"]["split_source_count"],
3632            json!(1)
3633        );
3634        assert_eq!(
3635            mutations[0].patch["activity"]["last_split_source_slot"],
3636            json!(77)
3637        );
3638        assert_eq!(
3639            mutations[0].patch["activity"].get("split_child_count"),
3640            None
3641        );
3642        assert_eq!(mutations[1].key, json!("child_pos"));
3643        assert_eq!(
3644            mutations[1].patch["activity"]["split_child_count"],
3645            json!(1)
3646        );
3647        assert_eq!(
3648            mutations[1].patch["activity"].get("split_source_count"),
3649            None
3650        );
3651    }
3652
3653    #[test]
3654    fn handlers_with_identical_keys_still_merge_into_one_segment() {
3655        let mut spec = split_position_spec();
3656        spec.instruction_hooks.clear();
3657        // Re-key the second handler by `first_position`: both now resolve the
3658        // same key and must share a single state read/write.
3659        spec.handlers[1].key_resolution = KeyResolutionStrategy::Embedded {
3660            primary_field: FieldPath::new(&["accounts", "first_position"]),
3661        };
3662        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3663        let handler = &bytecode.entities["Position"].handlers["amm::SplitPositionIxState"];
3664
3665        assert!(!handler
3666            .iter()
3667            .any(|op| matches!(op, super::OpCode::SegmentBoundary)));
3668        assert_eq!(
3669            handler
3670                .iter()
3671                .filter(|op| matches!(op, super::OpCode::ReadOrInitState { .. }))
3672                .count(),
3673            1
3674        );
3675    }
3676
3677    #[test]
3678    fn hook_without_a_matching_key_gets_its_own_segment() {
3679        let mut spec = split_position_spec();
3680        spec.handlers.truncate(1);
3681        spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
3682            primary_field: FieldPath::new(&["accounts", "second_position"]),
3683        };
3684        spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
3685            "id.position_address",
3686            &["accounts", "second_position"],
3687            PopulationStrategy::SetOnce,
3688        ));
3689        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3690        let mut vm = VmContext::new();
3691
3692        vm.process_event(
3693            &bytecode,
3694            split_position_event("source_pos", "child_pos"),
3695            "amm::SplitPositionIxState",
3696            None,
3697            None,
3698        )
3699        .unwrap();
3700
3701        // The hook routes by `first_position`, not by the handler's key.
3702        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3703        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3704        let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
3705        assert_eq!(child["activity"].get("last_split_source_slot"), None);
3706    }
3707
3708    /// Index positions by owner through a `PositionState` account handler.
3709    fn add_owner_index(spec: &mut TypedStreamSpec<Value>) {
3710        spec.identity.lookup_indexes.push(LookupIndexSpec {
3711            field_name: "id.owner".to_string(),
3712            temporal_field: None,
3713        });
3714        spec.handlers
3715            .push(crate::ast::TypedHandlerSpec::from_serializable(
3716                SerializableHandlerSpec {
3717                    source: SourceSpec::Source {
3718                        program_id: None,
3719                        discriminator: None,
3720                        type_name: "amm::PositionState".to_string(),
3721                        serialization: None,
3722                        is_account: true,
3723                    },
3724                    key_resolution: KeyResolutionStrategy::Embedded {
3725                        primary_field: FieldPath::new(&["__account_address"]),
3726                    },
3727                    mappings: vec![
3728                        mapping(
3729                            "id.position_address",
3730                            &["__account_address"],
3731                            PopulationStrategy::SetOnce,
3732                        ),
3733                        mapping("id.owner", &["owner"], PopulationStrategy::SetOnce),
3734                    ],
3735                    conditions: vec![],
3736                    emit: true,
3737                },
3738            ));
3739    }
3740
3741    fn index_owner(
3742        vm: &mut VmContext,
3743        bytecode: &MultiEntityBytecode,
3744        position: &str,
3745        owner: &str,
3746    ) {
3747        vm.process_event(
3748            bytecode,
3749            json!({ "__account_address": position, "owner": owner }),
3750            "amm::PositionState",
3751            None,
3752            None,
3753        )
3754        .unwrap();
3755    }
3756
3757    #[test]
3758    fn unmatched_hook_next_to_other_keys_resolves_lookup_index_fields() {
3759        let mut spec = split_position_spec();
3760        spec.handlers.truncate(1);
3761        spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
3762            primary_field: FieldPath::new(&["accounts", "second_position"]),
3763        };
3764        spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
3765            "id.position_address",
3766            &["accounts", "second_position"],
3767            PopulationStrategy::SetOnce,
3768        ));
3769        add_owner_index(&mut spec);
3770        // The hook routes by the owner, a lookup-index field.
3771        spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
3772
3773        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3774        let mut vm = VmContext::new();
3775        index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
3776        vm.process_event(
3777            &bytecode,
3778            json!({
3779                "accounts": {
3780                    "first_position": "source_pos",
3781                    "second_position": "child_pos",
3782                    "owner": "owner_1",
3783                },
3784                "data": { "slot": 77 },
3785            }),
3786            "amm::SplitPositionIxState",
3787            None,
3788            None,
3789        )
3790        .unwrap();
3791
3792        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3793        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3794        assert!(vm.get_entity_state(0, &json!("owner_1")).is_none());
3795    }
3796
3797    #[test]
3798    fn unmatched_hook_replays_after_its_lookup_index_is_populated() {
3799        let mut spec = split_position_spec();
3800        spec.handlers.truncate(1);
3801        spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
3802            primary_field: FieldPath::new(&["accounts", "second_position"]),
3803        };
3804        spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
3805            "id.position_address",
3806            &["accounts", "second_position"],
3807            PopulationStrategy::SetOnce,
3808        ));
3809        add_owner_index(&mut spec);
3810        spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
3811
3812        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3813        let mut vm = VmContext::new();
3814        let mutations = vm
3815            .process_event(
3816                &bytecode,
3817                json!({
3818                    "accounts": { "owner": "owner_1", "second_position": "child_pos" },
3819                    "data": { "slot": 77 },
3820                }),
3821                "amm::SplitPositionIxState",
3822                None,
3823                None,
3824            )
3825            .unwrap();
3826        assert_eq!(mutations.len(), 1);
3827        assert_eq!(mutations[0].key, json!("child_pos"));
3828
3829        index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
3830
3831        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3832        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
3833        let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
3834        assert_eq!(child["activity"].get("last_split_source_slot"), None);
3835    }
3836
3837    #[test]
3838    fn a_segment_that_misses_its_key_is_replayed_alone() {
3839        let mut spec = split_position_spec();
3840        spec.instruction_hooks.clear();
3841        // The child segment comes first. The source segment resolves its key
3842        // through a PDA, registered later, that maps to the position's owner.
3843        spec.handlers.swap(0, 1);
3844        spec.handlers[1].key_resolution = KeyResolutionStrategy::Lookup {
3845            primary_field: FieldPath::new(&["accounts", "source_pda"]),
3846        };
3847        spec.handlers[1].mappings.remove(0);
3848        add_owner_index(&mut spec);
3849        spec.instruction_hooks.push(InstructionHook {
3850            instruction_type: "amm::RegisterIxState".to_string(),
3851            actions: vec![HookAction::RegisterPdaMapping {
3852                pda_field: FieldPath::new(&["accounts", "source_pda"]),
3853                seed_field: FieldPath::new(&["accounts", "owner"]),
3854                lookup_name: "default_pda_lookup".to_string(),
3855            }],
3856            lookup_by: None,
3857        });
3858        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
3859        let mut vm = VmContext::new();
3860        index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
3861
3862        let mutations = vm
3863            .process_event(
3864                &bytecode,
3865                json!({
3866                    "accounts": { "source_pda": "pda_1", "second_position": "child_pos" },
3867                    "data": {},
3868                }),
3869                "amm::SplitPositionIxState",
3870                None,
3871                None,
3872            )
3873            .unwrap();
3874        // The child landed; the source is not resolvable yet.
3875        assert_eq!(mutations.len(), 1);
3876        assert_eq!(mutations[0].key, json!("child_pos"));
3877
3878        let mutations = vm
3879            .process_event(
3880                &bytecode,
3881                json!({ "accounts": { "source_pda": "pda_1", "owner": "owner_1" } }),
3882                "amm::RegisterIxState",
3883                None,
3884                None,
3885            )
3886            .unwrap();
3887        // Registering the PDA replays only the source segment.
3888        assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
3889        assert_eq!(mutations[0].key, json!("source_pos"));
3890        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
3891        assert_eq!(source["activity"]["split_source_count"], json!(1));
3892        let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
3893        assert_eq!(child["activity"]["split_child_count"], json!(1));
3894    }
3895
3896    /// A hook whose `lookup_by` names a field the instruction does not carry
3897    /// loads a null key. The entity's key transform must pass that null to
3898    /// the segment's null-key check, which skips the hook, rather than fail
3899    /// the event and lose the handler segment's update with it.
3900    #[test]
3901    fn a_hook_whose_lookup_field_is_absent_skips_only_its_segment() {
3902        let spec: TypedStreamSpec<Value> =
3903            TypedStreamSpec::from_serializable(SerializableStreamSpec {
3904                ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3905                state_name: "Game".to_string(),
3906                program_id: None,
3907                idl: None,
3908                identity: IdentitySpec {
3909                    primary_keys: vec!["id.game_id".to_string()],
3910                    lookup_indexes: vec![],
3911                },
3912                handlers: vec![instruction_handler(
3913                    "game::StartGameIxState",
3914                    KeyResolutionStrategy::Embedded {
3915                        primary_field: FieldPath::new(&["data", "game_id"]),
3916                    },
3917                    vec![
3918                        SerializableFieldMapping {
3919                            source: MappingSource::FromSource {
3920                                path: FieldPath::new(&["data", "game_id"]),
3921                                default: None,
3922                                transform: Some(crate::ast::Transformation::HexEncode),
3923                            },
3924                            ..mapping("id.game_id", &[], PopulationStrategy::SetOnce)
3925                        },
3926                        mapping("stats.fee", &["data", "fee"], PopulationStrategy::SetOnce),
3927                    ],
3928                )],
3929                sections: vec![],
3930                field_mappings: BTreeMap::new(),
3931                resolver_hooks: vec![],
3932                instruction_hooks: vec![InstructionHook {
3933                    instruction_type: "game::StartGameIxState".to_string(),
3934                    actions: vec![HookAction::SetField {
3935                        target_field: "stats.started_slot".to_string(),
3936                        source: MappingSource::FromSource {
3937                            path: FieldPath::new(&["data", "slot"]),
3938                            default: None,
3939                            transform: None,
3940                        },
3941                        condition: None,
3942                    }],
3943                    // `game_id` is an argument, not an account.
3944                    lookup_by: Some(FieldPath::new(&["accounts", "game_id"])),
3945                }],
3946                resolver_specs: vec![],
3947                computed_fields: vec![],
3948                computed_field_specs: vec![],
3949                content_hash: None,
3950                views: vec![],
3951            });
3952        let bytecode = MultiEntityBytecode::from_single("Game".to_string(), spec, 0);
3953        let mut vm = VmContext::new();
3954
3955        let mutations = vm
3956            .process_event(
3957                &bytecode,
3958                json!({
3959                    "accounts": { "game_account": "game_pda" },
3960                    "data": { "game_id": [1, 2, 3, 4], "fee": 5, "slot": 9 },
3961                }),
3962                "game::StartGameIxState",
3963                None,
3964                None,
3965            )
3966            .unwrap();
3967
3968        assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
3969        assert_eq!(mutations[0].key, json!("01020304"));
3970        let game = vm.get_entity_state(0, &json!("01020304")).unwrap();
3971        assert_eq!(game["id"]["game_id"], json!("01020304"));
3972        assert_eq!(game["stats"]["fee"], json!(5));
3973        assert_eq!(game["stats"].get("started_slot"), None);
3974        // The hook's writes are not kept anywhere else either.
3975        assert_eq!(vm.get_entity_state(0, &json!(null)), None);
3976    }
3977
3978    mod fingerprint {
3979        use super::super::{EntityBytecode, MultiEntityBytecode, OpCode};
3980        use crate::ast::FieldPath;
3981        use serde_json::json;
3982        use std::collections::{HashMap, HashSet};
3983
3984        fn sample_handler() -> Vec<OpCode> {
3985            vec![
3986                OpCode::LoadEventField {
3987                    path: FieldPath::new(&["id", "mint"]),
3988                    dest: 0,
3989                    default: Some(json!({"b": 2, "a": 1})),
3990                },
3991                OpCode::CreateObject { dest: 1 },
3992                OpCode::SetField {
3993                    object: 1,
3994                    path: "state.volume".to_string(),
3995                    value: 0,
3996                },
3997                OpCode::EmitMutation {
3998                    entity_name: "PumpfunToken".to_string(),
3999                    key: 0,
4000                    state: 1,
4001                },
4002            ]
4003        }
4004
4005        fn bytecode(handler_insert_order: &[&str]) -> MultiEntityBytecode {
4006            let mut handlers = HashMap::new();
4007            for event_type in handler_insert_order {
4008                handlers.insert(event_type.to_string(), sample_handler());
4009            }
4010            let mut entities = HashMap::new();
4011            entities.insert(
4012                "PumpfunToken".to_string(),
4013                EntityBytecode {
4014                    state_id: 0,
4015                    handlers,
4016                    entity_name: "PumpfunToken".to_string(),
4017                    when_events: HashSet::from(["pump::BuyIxState".to_string()]),
4018                    non_emitted_fields: HashSet::new(),
4019                    computed_paths: vec!["state.market_cap".to_string()],
4020                    computed_fields_evaluator: None,
4021                },
4022            );
4023            let mut event_routing = HashMap::new();
4024            event_routing.insert(
4025                "pump::TokenState".to_string(),
4026                vec!["PumpfunToken".to_string()],
4027            );
4028            MultiEntityBytecode {
4029                entities,
4030                event_routing,
4031                when_events: HashSet::from(["pump::BuyIxState".to_string()]),
4032                proto_router: crate::proto_router::ProtoRouter::new(),
4033            }
4034        }
4035
4036        #[test]
4037        fn stable_across_recomputation_and_insertion_order() {
4038            let a = bytecode(&["pump::TokenState", "pump::BuyIxState"]);
4039            let b = bytecode(&["pump::BuyIxState", "pump::TokenState"]);
4040            assert_eq!(a.fingerprint(), a.fingerprint());
4041            assert_eq!(a.fingerprint(), b.fingerprint());
4042            assert_eq!(a.fingerprint().len(), 64);
4043        }
4044
4045        #[test]
4046        fn changes_when_an_opcode_operand_changes() {
4047            let base = bytecode(&["pump::TokenState"]);
4048            let mut modified = bytecode(&["pump::TokenState"]);
4049            let entity = modified.entities.get_mut("PumpfunToken").unwrap();
4050            match &mut entity.handlers.get_mut("pump::TokenState").unwrap()[2] {
4051                OpCode::SetField { path, .. } => *path = "state.volume_usd".to_string(),
4052                other => panic!("unexpected opcode: {other:?}"),
4053            }
4054            assert_ne!(base.fingerprint(), modified.fingerprint());
4055        }
4056
4057        #[test]
4058        fn changes_when_routing_changes() {
4059            let base = bytecode(&["pump::TokenState"]);
4060            let mut modified = bytecode(&["pump::TokenState"]);
4061            modified.event_routing.insert(
4062                "pump::SellIxState".to_string(),
4063                vec!["PumpfunToken".to_string()],
4064            );
4065            assert_ne!(base.fingerprint(), modified.fingerprint());
4066        }
4067
4068        /// Golden hash: this value changing means every deployed snapshot gets
4069        /// invalidated (bytecode_hash mismatch -> cold start). That can be a
4070        /// legitimate consequence of a semantic change to the sample opcodes or
4071        /// to the canonical encoding itself — but it must be a *decision*, not
4072        /// an accident. Update the constant only when snapshot invalidation is
4073        /// intended.
4074        #[test]
4075        fn golden_hash_pins_the_canonical_encoding() {
4076            let expected = "acf7ea8bf39ca830a9e6c2b2d2fb471da68e57e24bbb4c256964a42a498f75f2";
4077            assert_eq!(bytecode(&["pump::TokenState"]).fingerprint(), expected);
4078        }
4079    }
4080}