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