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 (from_sibling, path) in [
1001                    (sibling.pda_from_sibling, pda_field),
1002                    (sibling.seed_from_sibling, seed_field),
1003                ] {
1004                    if from_sibling && !paths.iter().any(|known| known.segments == path.segments) {
1005                        paths.push(path.clone());
1006                    }
1007                }
1008            }
1009        }
1010        captures.into_iter().collect()
1011    }
1012
1013    /// Canonical encoding of the opcodes that resolve a handler's key. Two
1014    /// handlers for the same event may share one state read/write only when
1015    /// these are identical.
1016    fn key_identity(&self, spec: &TypedHandlerSpec<S>) -> Value {
1017        let key_reg = 20;
1018        let ops = self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings);
1019        serde_json::to_value(&ops).unwrap_or(Value::Null)
1020    }
1021
1022    /// How a hook's `lookup_by` field resolves the entity key: directly when
1023    /// it names a primary-key field, through its lookup index when it names a
1024    /// lookup-index field.
1025    fn hook_key_resolution(&self, lookup_by: &FieldPath) -> Option<KeyResolutionStrategy> {
1026        let leaf = lookup_by.segments.last()?;
1027        let is_primary_key_field = self
1028            .spec
1029            .identity
1030            .primary_keys
1031            .iter()
1032            .any(|pk| pk.rsplit('.').next() == Some(leaf.as_str()));
1033        if is_primary_key_field {
1034            Some(KeyResolutionStrategy::Embedded {
1035                primary_field: lookup_by.clone(),
1036            })
1037        } else if self.find_lookup_index_for_field(lookup_by).is_some() {
1038            Some(KeyResolutionStrategy::Lookup {
1039                primary_field: lookup_by.clone(),
1040                index_name: None,
1041            })
1042        } else {
1043            None
1044        }
1045    }
1046
1047    /// Segment for an instruction hook that routes by its own resolved key:
1048    /// key loading, state read and state write. Hook actions are inserted
1049    /// before the state write by the caller.
1050    fn compile_keyed_hook_segment(&self, resolution: &KeyResolutionStrategy) -> Vec<OpCode> {
1051        let key_reg = 20;
1052        let state_reg = 2;
1053        let mut ops = self.compile_key_loading(resolution, key_reg, &[]);
1054        ops.push(OpCode::AbortIfNullKey {
1055            key: key_reg,
1056            is_account_event: false,
1057        });
1058        ops.push(OpCode::ReadOrInitState {
1059            state_id: self.state_id,
1060            key: key_reg,
1061            default: serde_json::json!({}),
1062            dest: state_reg,
1063            dedup_route: None,
1064        });
1065        ops.push(OpCode::UpdateState {
1066            state_id: self.state_id,
1067            key: key_reg,
1068            value: state_reg,
1069        });
1070        ops
1071    }
1072
1073    /// Standalone segment for an instruction hook: key from the resolver or
1074    /// the hook's `lookup_by` field, then state read and write. Hook actions are
1075    /// inserted before the state write by the caller.
1076    fn compile_hook_segment(&self, hook: &InstructionHook) -> Vec<OpCode> {
1077        let key_reg = 20;
1078        let state_reg = 2;
1079        let resolved_key_reg = 19;
1080        let temp_reg = 18;
1081
1082        let mut ops = Vec::new();
1083
1084        // First, try to load __resolved_primary_key from resolver
1085        ops.push(OpCode::LoadEventField {
1086            path: FieldPath::new(&["__resolved_primary_key"]),
1087            dest: resolved_key_reg,
1088            default: Some(serde_json::json!(null)),
1089        });
1090
1091        // Copy to key_reg (unconditionally, may be null)
1092        ops.push(OpCode::CopyRegister {
1093            source: resolved_key_reg,
1094            dest: key_reg,
1095        });
1096
1097        // If hook has lookup_by, use it to load primary key from instruction accounts
1098        if let Some(lookup_path) = &hook.lookup_by {
1099            // Load the primary key from the instruction's lookup_by field (e.g., accounts.signer)
1100            ops.push(OpCode::LoadEventField {
1101                path: lookup_path.clone(),
1102                dest: temp_reg,
1103                default: None,
1104            });
1105
1106            // Apply HexEncode transformation (accounts are byte arrays)
1107            ops.push(OpCode::Transform {
1108                source: temp_reg,
1109                dest: temp_reg,
1110                transformation: Transformation::HexEncode,
1111            });
1112
1113            // Use this as fallback if __resolved_primary_key was null
1114            ops.push(OpCode::CopyRegisterIfNull {
1115                source: temp_reg,
1116                dest: key_reg,
1117            });
1118        }
1119
1120        ops.push(OpCode::ReadOrInitState {
1121            state_id: self.state_id,
1122            key: key_reg,
1123            default: serde_json::json!({}),
1124            dest: state_reg,
1125            dedup_route: None,
1126        });
1127
1128        ops.push(OpCode::UpdateState {
1129            state_id: self.state_id,
1130            key: key_reg,
1131            value: state_reg,
1132        });
1133
1134        ops
1135    }
1136
1137    fn compile_handler(&self, spec: &TypedHandlerSpec<S>) -> Vec<OpCode> {
1138        let mut ops = Vec::new();
1139        let state_reg = 2;
1140        let key_reg = 20;
1141
1142        ops.extend(self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings));
1143
1144        // Guard: if key resolved to null on an account-state event, abort
1145        // early with empty mutations so process_event can queue the update
1146        // for later reprocessing.  Without this, downstream opcodes would
1147        // create a phantom entity keyed by null and produce non-empty
1148        // mutations that prevent queueing.
1149        let is_account_event = matches!(
1150            spec.source,
1151            SourceSpec::Source {
1152                is_account: true,
1153                ..
1154            }
1155        );
1156        ops.push(OpCode::AbortIfNullKey {
1157            key: key_reg,
1158            is_account_event,
1159        });
1160
1161        ops.push(OpCode::ReadOrInitState {
1162            state_id: self.state_id,
1163            key: key_reg,
1164            default: serde_json::json!({}),
1165            dest: state_reg,
1166            dedup_route: declared_index(&spec.key_resolution)
1167                .map(|index| format!("{index}:{:?}", spec.key_resolution)),
1168        });
1169
1170        // Index updates must come AFTER ReadOrInitState so the state table exists.
1171        // ReadOrInitState lazily creates the state table via entry().or_insert_with(),
1172        // but index opcodes (UpdateLookupIndex, UpdateTemporalIndex, UpdatePdaReverseLookup)
1173        // use get_mut() which fails if the table doesn't exist yet.
1174        // This ordering also means stale/duplicate updates (caught by ReadOrInitState's
1175        // recency check) correctly skip index updates too.
1176        ops.extend(self.compile_temporal_index_update(
1177            &spec.key_resolution,
1178            key_reg,
1179            &spec.mappings,
1180        ));
1181
1182        for mapping in &spec.mappings {
1183            ops.extend(self.compile_mapping(mapping, state_reg, key_reg));
1184        }
1185
1186        ops.extend(self.compile_resolvers(state_reg, key_reg));
1187
1188        // Evaluate computed fields after all mappings but before updating state
1189        ops.push(OpCode::EvaluateComputedFields {
1190            state: state_reg,
1191            computed_paths: self.spec.computed_fields.clone(),
1192        });
1193
1194        ops.push(OpCode::UpdateState {
1195            state_id: self.state_id,
1196            key: key_reg,
1197            value: state_reg,
1198        });
1199
1200        if spec.emit {
1201            ops.push(OpCode::EmitMutation {
1202                entity_name: self.entity_name.clone(),
1203                key: key_reg,
1204                state: state_reg,
1205            });
1206        }
1207
1208        ops
1209    }
1210
1211    fn compile_resolvers(&self, state_reg: Register, key_reg: Register) -> Vec<OpCode> {
1212        let mut ops = Vec::new();
1213
1214        for resolver_spec in &self.spec.resolver_specs {
1215            let url_template = match &resolver_spec.resolver {
1216                ResolverType::Url(config) => match &config.url_source {
1217                    UrlSource::Template(parts) => Some(parts.clone()),
1218                    _ => None,
1219                },
1220                _ => None,
1221            };
1222
1223            ops.push(OpCode::QueueResolver {
1224                state_id: self.state_id,
1225                entity_name: self.entity_name.clone(),
1226                resolver: resolver_spec.resolver.clone(),
1227                input_path: resolver_spec.input_path.clone(),
1228                input_value: resolver_spec.input_value.clone(),
1229                url_template,
1230                strategy: resolver_spec.strategy.clone(),
1231                extracts: resolver_spec.extracts.clone(),
1232                condition: resolver_spec.condition.clone(),
1233                schedule_at: resolver_spec.schedule_at.clone(),
1234                state: state_reg,
1235                key: key_reg,
1236            });
1237        }
1238
1239        ops
1240    }
1241
1242    fn compile_mapping(
1243        &self,
1244        mapping: &TypedFieldMapping<S>,
1245        state_reg: Register,
1246        key_reg: Register,
1247    ) -> Vec<OpCode> {
1248        let guard = mapping
1249            .condition
1250            .as_ref()
1251            .and_then(|condition| condition.parsed.as_ref())
1252            .and_then(|parsed| match parsed {
1253                ParsedCondition::Guard { expr } => Some(expr),
1254                _ => None,
1255            });
1256        let Some(guard) = guard else {
1257            return self.compile_mapping_ops(
1258                mapping,
1259                mapping.condition.as_ref(),
1260                state_reg,
1261                key_reg,
1262            );
1263        };
1264        // A guarded mapping is the unconditional mapping, skipped as a whole
1265        // unless the guard over the incoming update is true.
1266        let inner = self.compile_mapping_ops(mapping, None, state_reg, key_reg);
1267        let mut ops = Vec::with_capacity(inner.len() + 2);
1268        ops.push(OpCode::EvaluateEventExpr {
1269            expr: guard.clone(),
1270            transform: None,
1271            dest: GUARD_REG,
1272        });
1273        ops.push(OpCode::SkipUnless {
1274            predicate: GUARD_REG,
1275            count: inner.len(),
1276        });
1277        ops.extend(inner);
1278        ops
1279    }
1280
1281    fn compile_mapping_ops(
1282        &self,
1283        mapping: &TypedFieldMapping<S>,
1284        condition: Option<&ConditionExpr>,
1285        state_reg: Register,
1286        key_reg: Register,
1287    ) -> Vec<OpCode> {
1288        let mut ops = Vec::new();
1289        let temp_reg = 10;
1290
1291        ops.extend(self.compile_mapping_source(&mapping.source, temp_reg));
1292
1293        if let Some(transform) = &mapping.transform {
1294            ops.push(OpCode::Transform {
1295                source: temp_reg,
1296                dest: temp_reg,
1297                transformation: transform.clone(),
1298            });
1299        }
1300
1301        if let Some(stop_instruction) = &mapping.stop {
1302            if mapping.when.is_some() {
1303                tracing::warn!(
1304                    "#[map] stop and when both set for {}. Ignoring when.",
1305                    mapping.target_path
1306                );
1307            }
1308            if !matches!(mapping.population, PopulationStrategy::LastWrite)
1309                && !matches!(mapping.population, PopulationStrategy::Merge)
1310            {
1311                tracing::warn!(
1312                    "#[map] stop ignores population strategy {:?}",
1313                    mapping.population
1314                );
1315            }
1316
1317            ops.push(OpCode::SetFieldUnlessStopped {
1318                object: state_reg,
1319                path: mapping.target_path.clone(),
1320                value: temp_reg,
1321                stop_field: stop_field_path(&mapping.target_path),
1322                stop_instruction: stop_instruction.clone(),
1323                entity_name: self.entity_name.clone(),
1324                key_reg,
1325            });
1326            return ops;
1327        }
1328
1329        if let Some(when_instruction) = &mapping.when {
1330            if !matches!(mapping.population, PopulationStrategy::LastWrite)
1331                && !matches!(mapping.population, PopulationStrategy::Merge)
1332            {
1333                tracing::warn!(
1334                    "#[map] when ignores population strategy {:?}",
1335                    mapping.population
1336                );
1337            }
1338            let (condition_field, condition_op, condition_value) = condition
1339                .and_then(|cond| cond.parsed.as_ref())
1340                .and_then(|parsed| match parsed {
1341                    ParsedCondition::Comparison { field, op, value } => {
1342                        Some((Some(field.clone()), Some(op.clone()), Some(value.clone())))
1343                    }
1344                    ParsedCondition::Logical { .. } => {
1345                        tracing::warn!("Logical conditions not yet supported for #[map] when");
1346                        None
1347                    }
1348                    // compile_mapping evaluates guards before this point.
1349                    ParsedCondition::Guard { .. } => None,
1350                })
1351                .unwrap_or((None, None, None));
1352
1353            ops.push(OpCode::SetFieldWhen {
1354                object: state_reg,
1355                path: mapping.target_path.clone(),
1356                value: temp_reg,
1357                when_instruction: when_instruction.clone(),
1358                entity_name: self.entity_name.clone(),
1359                key_reg,
1360                condition_field,
1361                condition_op,
1362                condition_value,
1363            });
1364            return ops;
1365        }
1366
1367        if let Some(condition) = condition {
1368            if let Some(parsed) = &condition.parsed {
1369                match parsed {
1370                    ParsedCondition::Comparison {
1371                        field,
1372                        op,
1373                        value: cond_value,
1374                    } => {
1375                        if matches!(mapping.population, PopulationStrategy::LastWrite)
1376                            || matches!(mapping.population, PopulationStrategy::Merge)
1377                        {
1378                            ops.push(OpCode::ConditionalSetField {
1379                                object: state_reg,
1380                                path: mapping.target_path.clone(),
1381                                value: temp_reg,
1382                                condition_field: field.clone(),
1383                                condition_op: op.clone(),
1384                                condition_value: cond_value.clone(),
1385                            });
1386                            return ops;
1387                        }
1388
1389                        if matches!(mapping.population, PopulationStrategy::Count) {
1390                            ops.push(OpCode::ConditionalIncrement {
1391                                object: state_reg,
1392                                path: mapping.target_path.clone(),
1393                                condition_field: field.clone(),
1394                                condition_op: op.clone(),
1395                                condition_value: cond_value.clone(),
1396                            });
1397                            return ops;
1398                        }
1399
1400                        tracing::warn!(
1401                            "Conditional #[map] not supported for population strategy {:?}",
1402                            mapping.population
1403                        );
1404                    }
1405                    ParsedCondition::Logical { .. } => {
1406                        tracing::warn!("Logical conditions not yet supported for #[map]");
1407                    }
1408                    // compile_mapping evaluates guards before this point.
1409                    ParsedCondition::Guard { .. } => {}
1410                }
1411            }
1412        }
1413
1414        match &mapping.population {
1415            PopulationStrategy::Append => {
1416                ops.push(OpCode::AppendToArray {
1417                    object: state_reg,
1418                    path: mapping.target_path.clone(),
1419                    value: temp_reg,
1420                });
1421            }
1422            PopulationStrategy::LastWrite => {
1423                ops.push(OpCode::SetField {
1424                    object: state_reg,
1425                    path: mapping.target_path.clone(),
1426                    value: temp_reg,
1427                });
1428            }
1429            PopulationStrategy::SetOnce => {
1430                ops.push(OpCode::SetFieldIfNull {
1431                    object: state_reg,
1432                    path: mapping.target_path.clone(),
1433                    value: temp_reg,
1434                });
1435            }
1436            PopulationStrategy::Merge => {
1437                ops.push(OpCode::SetField {
1438                    object: state_reg,
1439                    path: mapping.target_path.clone(),
1440                    value: temp_reg,
1441                });
1442            }
1443            PopulationStrategy::Max => {
1444                ops.push(OpCode::SetFieldMax {
1445                    object: state_reg,
1446                    path: mapping.target_path.clone(),
1447                    value: temp_reg,
1448                });
1449            }
1450            PopulationStrategy::Sum => {
1451                ops.push(OpCode::SetFieldSum {
1452                    object: state_reg,
1453                    path: mapping.target_path.clone(),
1454                    value: temp_reg,
1455                });
1456            }
1457            PopulationStrategy::Count => {
1458                // Count doesn't need the value, just increment
1459                ops.push(OpCode::SetFieldIncrement {
1460                    object: state_reg,
1461                    path: mapping.target_path.clone(),
1462                });
1463            }
1464            PopulationStrategy::Min => {
1465                ops.push(OpCode::SetFieldMin {
1466                    object: state_reg,
1467                    path: mapping.target_path.clone(),
1468                    value: temp_reg,
1469                });
1470            }
1471            PopulationStrategy::UniqueCount => {
1472                // UniqueCount requires maintaining an internal set
1473                // The field stores the count, but we track unique values in a hidden set
1474                let set_name = format!("{}_unique_set", mapping.target_path);
1475                ops.push(OpCode::AddToUniqueSet {
1476                    state_id: self.state_id,
1477                    set_name,
1478                    value: temp_reg,
1479                    count_object: state_reg,
1480                    count_path: mapping.target_path.clone(),
1481                });
1482            }
1483        }
1484
1485        ops
1486    }
1487
1488    fn compile_mapping_source(&self, source: &MappingSource, dest: Register) -> Vec<OpCode> {
1489        match source {
1490            MappingSource::FromSource {
1491                path,
1492                default,
1493                transform,
1494            } => {
1495                let mut ops = vec![OpCode::LoadEventField {
1496                    path: path.clone(),
1497                    dest,
1498                    default: default.clone(),
1499                }];
1500
1501                // Apply transform if specified in the source
1502                if let Some(transform_type) = transform {
1503                    ops.push(OpCode::Transform {
1504                        source: dest,
1505                        dest,
1506                        transformation: transform_type.clone(),
1507                    });
1508                }
1509
1510                ops
1511            }
1512            MappingSource::Constant(val) => {
1513                vec![OpCode::LoadConstant {
1514                    value: val.clone(),
1515                    dest,
1516                }]
1517            }
1518            MappingSource::AsEvent { fields } => {
1519                let mut ops = Vec::new();
1520
1521                if fields.is_empty() {
1522                    let event_data_reg = dest + 1;
1523                    ops.push(OpCode::LoadEventField {
1524                        path: FieldPath::new(&[]),
1525                        dest: event_data_reg,
1526                        default: Some(serde_json::json!({})),
1527                    });
1528                    ops.push(OpCode::CreateEvent {
1529                        dest,
1530                        event_value: event_data_reg,
1531                    });
1532                } else {
1533                    let data_obj_reg = dest + 1;
1534                    ops.push(OpCode::CreateObject { dest: data_obj_reg });
1535
1536                    let mut field_registers = Vec::new();
1537                    let mut current_reg = dest + 2;
1538
1539                    for field_source in fields.iter() {
1540                        if let MappingSource::FromSource {
1541                            path,
1542                            default,
1543                            transform,
1544                        } = &**field_source
1545                        {
1546                            ops.push(OpCode::LoadEventField {
1547                                path: path.clone(),
1548                                dest: current_reg,
1549                                default: default.clone(),
1550                            });
1551
1552                            if let Some(transform_type) = transform {
1553                                ops.push(OpCode::Transform {
1554                                    source: current_reg,
1555                                    dest: current_reg,
1556                                    transformation: transform_type.clone(),
1557                                });
1558                            }
1559
1560                            if let Some(field_name) = path.segments.last() {
1561                                field_registers.push((field_name.clone(), current_reg));
1562                            }
1563                            current_reg += 1;
1564                        }
1565                    }
1566
1567                    if !field_registers.is_empty() {
1568                        ops.push(OpCode::SetFields {
1569                            object: data_obj_reg,
1570                            fields: field_registers,
1571                        });
1572                    }
1573
1574                    ops.push(OpCode::CreateEvent {
1575                        dest,
1576                        event_value: data_obj_reg,
1577                    });
1578                }
1579
1580                ops
1581            }
1582            MappingSource::WholeSource => {
1583                vec![OpCode::LoadEventField {
1584                    path: FieldPath::new(&[]),
1585                    dest,
1586                    default: Some(serde_json::json!({})),
1587                }]
1588            }
1589            MappingSource::AsCapture { field_transforms } => {
1590                // AsCapture loads the whole source, applies field-level transforms, and wraps in CaptureWrapper
1591                let capture_data_reg = 22; // Temp register for capture data before wrapping
1592                let mut ops = vec![OpCode::LoadEventField {
1593                    path: FieldPath::new(&[]),
1594                    dest: capture_data_reg,
1595                    default: Some(serde_json::json!({})),
1596                }];
1597
1598                // Apply transforms to specific fields in the loaded object
1599                // IMPORTANT: Use registers that don't conflict with key_reg (20)
1600                // Using 24 and 25 to avoid conflicts with key loading (uses 18, 19, 20, 23)
1601                let field_reg = 24;
1602                let transformed_reg = 25;
1603
1604                for (field_name, transform) in field_transforms {
1605                    // Load the field from the capture_data_reg (not from event!)
1606                    // Use GetField opcode to read from a register instead of LoadEventField
1607                    ops.push(OpCode::GetField {
1608                        object: capture_data_reg,
1609                        path: field_name.clone(),
1610                        dest: field_reg,
1611                    });
1612
1613                    // Transform it
1614                    ops.push(OpCode::Transform {
1615                        source: field_reg,
1616                        dest: transformed_reg,
1617                        transformation: transform.clone(),
1618                    });
1619
1620                    // Set it back into the capture data object
1621                    ops.push(OpCode::SetField {
1622                        object: capture_data_reg,
1623                        path: field_name.clone(),
1624                        value: transformed_reg,
1625                    });
1626                }
1627
1628                // Wrap the capture data in CaptureWrapper with metadata
1629                ops.push(OpCode::CreateCapture {
1630                    dest,
1631                    capture_value: capture_data_reg,
1632                });
1633
1634                ops
1635            }
1636            MappingSource::FromContext { field } => {
1637                // Load from instruction context (timestamp, slot, signature)
1638                vec![OpCode::LoadEventField {
1639                    path: FieldPath::new(&["__update_context", field.as_str()]),
1640                    dest,
1641                    default: Some(serde_json::json!(null)),
1642                }]
1643            }
1644            MappingSource::FromExpr { expr } => {
1645                vec![OpCode::EvaluateEventExpr {
1646                    expr: expr.clone(),
1647                    transform: None,
1648                    dest,
1649                }]
1650            }
1651            MappingSource::Computed { .. } => {
1652                vec![]
1653            }
1654            MappingSource::FromState { .. } => {
1655                vec![]
1656            }
1657        }
1658    }
1659
1660    pub fn compile_key_loading(
1661        &self,
1662        resolution: &KeyResolutionStrategy,
1663        key_reg: Register,
1664        mappings: &[TypedFieldMapping<S>],
1665    ) -> Vec<OpCode> {
1666        let mut ops = Vec::new();
1667
1668        // Resolvers provide either a final key or an intermediate lookup key,
1669        // depending on the handler strategy.
1670        let resolved_key_reg = 19; // Use a temp register
1671        ops.push(OpCode::LoadEventField {
1672            path: FieldPath::new(&["__resolved_primary_key"]),
1673            dest: resolved_key_reg,
1674            default: Some(serde_json::json!(null)),
1675        });
1676
1677        // Now do the normal key resolution
1678        match resolution {
1679            KeyResolutionStrategy::Embedded { primary_field }
1680                if primary_field.segments.is_empty()
1681                    && self.find_primary_key_expr(mappings).is_some() =>
1682            {
1683                // The primary key mapping is an expression over the event,
1684                // such as a pool and a time bucket.
1685                let expr = self
1686                    .find_primary_key_expr(mappings)
1687                    .expect("checked by the guard");
1688                // A write guard on the key mapping gates the key too: when it
1689                // fails the key stays null and the update writes no row.
1690                if let Some(guard) = self.find_primary_key_guard(mappings) {
1691                    ops.push(OpCode::EvaluateEventExpr {
1692                        expr: guard,
1693                        transform: None,
1694                        dest: GUARD_REG,
1695                    });
1696                    ops.push(OpCode::SkipUnless {
1697                        predicate: GUARD_REG,
1698                        count: 1,
1699                    });
1700                }
1701                // The key is the value the mapping writes to the primary key
1702                // field, transform included; a null key stays null, so the
1703                // event writes no row.
1704                ops.push(OpCode::EvaluateEventExpr {
1705                    expr,
1706                    transform: self.find_primary_key_transformation(mappings),
1707                    dest: key_reg,
1708                });
1709                ops.push(OpCode::CopyRegisterIfNull {
1710                    source: resolved_key_reg,
1711                    dest: key_reg,
1712                });
1713            }
1714            KeyResolutionStrategy::Embedded { primary_field } => {
1715                // Enhanced key resolution: check for auto-inheritance when primary_field is empty
1716                let effective_primary_field = if primary_field.segments.is_empty() {
1717                    // Try to auto-detect primary field from account schema
1718                    if let Some(auto_field) = self.auto_detect_primary_field(mappings) {
1719                        auto_field
1720                    } else {
1721                        primary_field.clone()
1722                    }
1723                } else {
1724                    primary_field.clone()
1725                };
1726
1727                // Skip fallback key loading if effective primary_field is still empty
1728                // This happens for account types that rely solely on __resolved_primary_key
1729                // (e.g., accounts with #[resolve_key_for] resolvers)
1730                if !effective_primary_field.segments.is_empty() {
1731                    let temp_reg = 18;
1732                    let transform_reg = 23; // Register for transformed key
1733
1734                    ops.push(OpCode::LoadEventField {
1735                        path: effective_primary_field.clone(),
1736                        dest: temp_reg,
1737                        default: None,
1738                    });
1739
1740                    // Check if there's a transformation for the primary key field
1741                    // First try the current mappings, then inherited transformations
1742                    let primary_key_transform = self
1743                        .find_primary_key_transformation(mappings)
1744                        .or_else(|| self.find_inherited_primary_key_transformation());
1745
1746                    if let Some(transform) = primary_key_transform {
1747                        // Apply transformation to the loaded key
1748                        ops.push(OpCode::Transform {
1749                            source: temp_reg,
1750                            dest: transform_reg,
1751                            transformation: transform,
1752                        });
1753                        // A concrete embedded key belongs to this entity and
1754                        // must win over a resolver result injected for another
1755                        // handler consuming the same event.
1756                        ops.push(OpCode::CopyRegister {
1757                            source: transform_reg,
1758                            dest: key_reg,
1759                        });
1760                    } else {
1761                        // No transformation, use raw value.
1762                        ops.push(OpCode::CopyRegister {
1763                            source: temp_reg,
1764                            dest: key_reg,
1765                        });
1766                    }
1767                    ops.push(OpCode::CopyRegisterIfNull {
1768                        source: resolved_key_reg,
1769                        dest: key_reg,
1770                    });
1771                } else {
1772                    // Resolver-only embedded handlers have no local key field.
1773                    ops.push(OpCode::CopyRegister {
1774                        source: resolved_key_reg,
1775                        dest: key_reg,
1776                    });
1777                }
1778            }
1779            KeyResolutionStrategy::Lookup {
1780                primary_field,
1781                index_name: declared,
1782            } => {
1783                let lookup_reg = 15;
1784                let result_reg = 17;
1785
1786                // Prefer resolver-provided key as lookup input.
1787                // When __resolved_primary_key is set (e.g. round_address from
1788                // PDA reverse lookup), use it directly — this gives a one-hop
1789                // lookup (round_address → round_id) instead of a two-hop chain
1790                // (Var address → PDA → round_address → round_id).
1791                ops.push(OpCode::CopyRegister {
1792                    source: resolved_key_reg,
1793                    dest: lookup_reg,
1794                });
1795
1796                let temp_reg = 18;
1797                ops.push(OpCode::LoadEventField {
1798                    path: primary_field.clone(),
1799                    dest: temp_reg,
1800                    default: None,
1801                });
1802                ops.push(OpCode::CopyRegisterIfNull {
1803                    source: temp_reg,
1804                    dest: lookup_reg,
1805                });
1806
1807                // A declared index is used as is, and strictly; otherwise
1808                // the index is inferred from the field names, and the
1809                // runtime falls back to the entity's other indexes.
1810                let effective_index_name = match declared {
1811                    Some(declared) => declared.clone(),
1812                    None => self
1813                        .find_lookup_index_for_lookup_field(primary_field, mappings)
1814                        .unwrap_or_else(|| "default_pda_lookup".to_string()),
1815                };
1816
1817                ops.push(OpCode::LookupIndex {
1818                    state_id: self.state_id,
1819                    index_name: effective_index_name,
1820                    lookup_value: lookup_reg,
1821                    dest: result_reg,
1822                    strict: declared.is_some(),
1823                });
1824                ops.push(OpCode::CopyRegister {
1825                    source: result_reg,
1826                    dest: key_reg,
1827                });
1828
1829                // Most lookup-based account handlers expect resolver output to be an
1830                // intermediate lookup value (for example, PDA -> round_address -> round_id),
1831                // so a null LookupIndex result must leave the key null for queueing.
1832                // Some stacks register the PDA directly to the entity primary key
1833                // (for example, bonding_curve -> mint). In that case the resolver output
1834                // is already the final key, so preserve it only when the instruction hook's
1835                // seed field matches one of the entity primary key fields.
1836                if self.resolver_outputs_primary_key_directly(primary_field) {
1837                    ops.push(OpCode::CopyRegisterIfNull {
1838                        source: resolved_key_reg,
1839                        dest: key_reg,
1840                    });
1841                }
1842            }
1843            KeyResolutionStrategy::Computed {
1844                primary_field,
1845                compute_partition: _,
1846            } => {
1847                // Copy resolver result to key_reg (may be null)
1848                ops.push(OpCode::CopyRegister {
1849                    source: resolved_key_reg,
1850                    dest: key_reg,
1851                });
1852                let temp_reg = 18;
1853                ops.push(OpCode::LoadEventField {
1854                    path: primary_field.clone(),
1855                    dest: temp_reg,
1856                    default: None,
1857                });
1858                ops.push(OpCode::CopyRegisterIfNull {
1859                    source: temp_reg,
1860                    dest: key_reg,
1861                });
1862            }
1863            KeyResolutionStrategy::TemporalLookup {
1864                lookup_field,
1865                timestamp_field,
1866                index_name,
1867            } => {
1868                // Copy resolver result to key_reg (may be null)
1869                ops.push(OpCode::CopyRegister {
1870                    source: resolved_key_reg,
1871                    dest: key_reg,
1872                });
1873                let lookup_reg = 15;
1874                let timestamp_reg = 16;
1875                let result_reg = 17;
1876
1877                ops.push(OpCode::LoadEventField {
1878                    path: lookup_field.clone(),
1879                    dest: lookup_reg,
1880                    default: None,
1881                });
1882
1883                ops.push(OpCode::LoadEventField {
1884                    path: timestamp_field.clone(),
1885                    dest: timestamp_reg,
1886                    default: None,
1887                });
1888
1889                ops.push(OpCode::LookupTemporalIndex {
1890                    state_id: self.state_id,
1891                    index_name: index_name.clone(),
1892                    lookup_value: lookup_reg,
1893                    timestamp: timestamp_reg,
1894                    dest: result_reg,
1895                });
1896
1897                ops.push(OpCode::CopyRegisterIfNull {
1898                    source: result_reg,
1899                    dest: key_reg,
1900                });
1901            }
1902        }
1903
1904        ops
1905    }
1906
1907    /// The expression of a mapping in `mappings` that sets the primary key,
1908    /// if one does.
1909    fn find_primary_key_expr(&self, mappings: &[TypedFieldMapping<S>]) -> Option<ComputedExpr> {
1910        let primary_key = self.spec.identity.primary_keys.first()?;
1911        mappings.iter().find_map(|mapping| match &mapping.source {
1912            MappingSource::FromExpr { expr } if mapping.target_path == *primary_key => {
1913                Some(expr.clone())
1914            }
1915            _ => None,
1916        })
1917    }
1918
1919    /// The write guard of the expression mapping that sets the primary key,
1920    /// if it has one.
1921    fn find_primary_key_guard(&self, mappings: &[TypedFieldMapping<S>]) -> Option<ComputedExpr> {
1922        let primary_key = self.spec.identity.primary_keys.first()?;
1923        mappings.iter().find_map(|mapping| {
1924            if mapping.target_path != *primary_key
1925                || !matches!(mapping.source, MappingSource::FromExpr { .. })
1926            {
1927                return None;
1928            }
1929            match mapping.condition.as_ref()?.parsed.as_ref()? {
1930                ParsedCondition::Guard { expr } => Some(expr.clone()),
1931                _ => None,
1932            }
1933        })
1934    }
1935
1936    fn find_primary_key_transformation(
1937        &self,
1938        mappings: &[TypedFieldMapping<S>],
1939    ) -> Option<Transformation> {
1940        // Find the first primary key in the identity spec
1941        let primary_key = self.spec.identity.primary_keys.first()?;
1942        let primary_field_name = self.extract_primary_field_name(primary_key)?;
1943
1944        // Look for a mapping that targets this primary key
1945        for mapping in mappings {
1946            // Check if this mapping targets the primary key field
1947            if mapping.target_path == *primary_key
1948                || mapping.target_path.ends_with(&format!(".{}", primary_key))
1949            {
1950                // Check mapping-level transform first
1951                if let Some(transform) = &mapping.transform {
1952                    return Some(transform.clone());
1953                }
1954
1955                // Then check source-level transform
1956                if let MappingSource::FromSource {
1957                    transform: Some(transform),
1958                    ..
1959                } = &mapping.source
1960                {
1961                    return Some(transform.clone());
1962                }
1963            }
1964        }
1965
1966        // If no explicit primary key mapping found, check AsCapture field transforms
1967        for mapping in mappings {
1968            if let MappingSource::AsCapture { field_transforms } = &mapping.source {
1969                if let Some(transform) = field_transforms.get(&primary_field_name) {
1970                    return Some(transform.clone());
1971                }
1972            }
1973        }
1974
1975        None
1976    }
1977
1978    /// Look for primary key mappings in other handlers of the same entity
1979    /// This enables cross-handler inheritance of key transformations
1980    pub fn find_inherited_primary_key_transformation(&self) -> Option<Transformation> {
1981        let primary_key = self.spec.identity.primary_keys.first()?;
1982
1983        // Extract the field name from the primary key path (e.g., "id.authority" -> "authority")
1984        let primary_field_name = self.extract_primary_field_name(primary_key)?;
1985
1986        // Search through all handlers in the spec for primary key mappings
1987        for handler in &self.spec.handlers {
1988            for mapping in &handler.mappings {
1989                // Look for mappings targeting the primary key
1990                if mapping.target_path == *primary_key
1991                    || mapping.target_path.ends_with(&format!(".{}", primary_key))
1992                {
1993                    // Check if this mapping comes from a field matching the primary key name
1994                    if let MappingSource::FromSource {
1995                        path, transform, ..
1996                    } = &mapping.source
1997                    {
1998                        if path.segments.last() == Some(&primary_field_name) {
1999                            // Return mapping-level transform first, then source-level transform
2000                            return mapping.transform.clone().or_else(|| transform.clone());
2001                        }
2002                    }
2003                }
2004
2005                // Also check AsCapture field transforms for the primary field
2006                if let MappingSource::AsCapture { field_transforms } = &mapping.source {
2007                    if let Some(transform) = field_transforms.get(&primary_field_name) {
2008                        return Some(transform.clone());
2009                    }
2010                }
2011            }
2012        }
2013
2014        None
2015    }
2016
2017    /// Extract the field name from a primary key path (e.g., "id.authority" -> "authority")
2018    fn extract_primary_field_name(&self, primary_key: &str) -> Option<String> {
2019        // Split by '.' and take the last segment
2020        primary_key.split('.').next_back().map(|s| s.to_string())
2021    }
2022
2023    /// Auto-detect primary field from account schema when no explicit mapping exists
2024    /// This looks for account types that have an 'authority' field and tries to use it
2025    pub fn auto_detect_primary_field(
2026        &self,
2027        current_mappings: &[TypedFieldMapping<S>],
2028    ) -> Option<FieldPath> {
2029        let primary_key = self.spec.identity.primary_keys.first()?;
2030
2031        // Extract the field name from the primary key (e.g., "id.authority" -> "authority")
2032        let primary_field_name = self.extract_primary_field_name(primary_key)?;
2033
2034        // Check if current handler can access the primary field
2035        if self.current_account_has_primary_field(&primary_field_name, current_mappings) {
2036            return Some(FieldPath::new(&[&primary_field_name]));
2037        }
2038
2039        None
2040    }
2041
2042    /// Check if the current account type has the primary field
2043    /// This is determined by looking at the mappings to see what fields are available
2044    fn current_account_has_primary_field(
2045        &self,
2046        field_name: &str,
2047        mappings: &[TypedFieldMapping<S>],
2048    ) -> bool {
2049        // Look through the mappings to see if any reference the primary field
2050        for mapping in mappings {
2051            if let MappingSource::FromSource { path, .. } = &mapping.source {
2052                // Check if this mapping sources from the primary field
2053                if path.segments.last() == Some(&field_name.to_string()) {
2054                    return true;
2055                }
2056            }
2057        }
2058
2059        false
2060    }
2061
2062    /// Check if handler has access to a specific field in its source account
2063    #[allow(dead_code)]
2064    fn handler_has_field(&self, field_name: &str, mappings: &[TypedFieldMapping<S>]) -> bool {
2065        for mapping in mappings {
2066            if let MappingSource::FromSource { path, .. } = &mapping.source {
2067                if path.segments.last() == Some(&field_name.to_string()) {
2068                    return true;
2069                }
2070            }
2071        }
2072        false
2073    }
2074
2075    /// Check if field exists by looking at mappings (IDL-agnostic approach)
2076    /// This avoids hardcoding account schemas and uses actual mapping evidence
2077    #[allow(dead_code)]
2078    fn field_exists_in_mappings(
2079        &self,
2080        field_name: &str,
2081        mappings: &[TypedFieldMapping<S>],
2082    ) -> bool {
2083        // Look through current mappings to see if the field is referenced
2084        for mapping in mappings {
2085            if let MappingSource::FromSource { path, .. } = &mapping.source {
2086                if path.segments.last() == Some(&field_name.to_string()) {
2087                    return true;
2088                }
2089            }
2090            // Also check AsCapture field transforms
2091            if let MappingSource::AsCapture { field_transforms } = &mapping.source {
2092                if field_transforms.contains_key(field_name) {
2093                    return true;
2094                }
2095            }
2096        }
2097        false
2098    }
2099
2100    fn find_lookup_index_for_field(&self, field_path: &FieldPath) -> Option<String> {
2101        if field_path.segments.is_empty() {
2102            return None;
2103        }
2104
2105        let lookup_field_name = field_path.segments.last().unwrap();
2106
2107        for lookup_index in &self.spec.identity.lookup_indexes {
2108            let index_field_name = lookup_index
2109                .field_name
2110                .split('.')
2111                .next_back()
2112                .unwrap_or(&lookup_index.field_name);
2113            let matches_directly = index_field_name == lookup_field_name;
2114            // An index field named `foo_address` is treated as an alias for the
2115            // bare field `foo` (for example, `mint_address` resolves handlers
2116            // keyed on `mint`). This is intentionally one-way: bare `foo` does
2117            // not imply a `foo_address` lookup index. The macro crate mirrors
2118            // this convention via `lookup_index_leafs` in
2119            // `arete-macros/src/validation/mod.rs`.
2120            let matches_address_alias = index_field_name
2121                .strip_suffix("_address")
2122                .map(|base| base == lookup_field_name)
2123                .unwrap_or(false);
2124
2125            if matches_directly || matches_address_alias {
2126                return Some(format!("{}_lookup_index", index_field_name));
2127            }
2128        }
2129
2130        None
2131    }
2132
2133    /// Whether a handler of this entity declares it resolves through
2134    /// `index_name` (a declared, strict lookup).
2135    fn declares_index(&self, index_name: &str) -> bool {
2136        self.spec
2137            .handlers
2138            .iter()
2139            .any(|handler| declared_index(&handler.key_resolution) == Some(index_name))
2140    }
2141
2142    /// Find lookup index for a Lookup key resolution by checking if there's a mapping
2143    /// from the primary_field to a lookup index field.
2144    fn find_lookup_index_for_lookup_field(
2145        &self,
2146        primary_field: &FieldPath,
2147        mappings: &[TypedFieldMapping<S>],
2148    ) -> Option<String> {
2149        // Build the primary field path string
2150        let primary_path = primary_field.segments.join(".");
2151
2152        // Check if there's a mapping from this primary field to a lookup index field
2153        for mapping in mappings {
2154            // Check if the mapping source path matches the primary field
2155            if let MappingSource::FromSource { path, .. } = &mapping.source {
2156                let source_path = path.segments.join(".");
2157                if source_path == primary_path {
2158                    // Check if the target is a lookup index field
2159                    for lookup_index in &self.spec.identity.lookup_indexes {
2160                        if mapping.target_path == lookup_index.field_name {
2161                            let index_field_name = lookup_index
2162                                .field_name
2163                                .split('.')
2164                                .next_back()
2165                                .unwrap_or(&lookup_index.field_name);
2166                            return Some(format!("{}_lookup_index", index_field_name));
2167                        }
2168                    }
2169                }
2170            }
2171        }
2172
2173        // Fall back to direct field name matching
2174        self.find_lookup_index_for_field(primary_field)
2175    }
2176
2177    /// Find the source path for a lookup index field by looking at mappings.
2178    /// For example, if target_path is "id.round_address" and the mapping is
2179    /// `id.round_address <- __account_address`, this returns ["__account_address"].
2180    fn find_source_path_for_lookup_index(
2181        &self,
2182        mappings: &[TypedFieldMapping<S>],
2183        lookup_field_name: &str,
2184    ) -> Option<Vec<String>> {
2185        for mapping in mappings {
2186            if mapping.target_path == lookup_field_name {
2187                if let MappingSource::FromSource { path, .. } = &mapping.source {
2188                    return Some(path.segments.clone());
2189                }
2190            }
2191        }
2192        None
2193    }
2194
2195    fn compile_temporal_index_update(
2196        &self,
2197        resolution: &KeyResolutionStrategy,
2198        key_reg: Register,
2199        mappings: &[TypedFieldMapping<S>],
2200    ) -> Vec<OpCode> {
2201        let mut ops = Vec::new();
2202
2203        for lookup_index in &self.spec.identity.lookup_indexes {
2204            let lookup_reg = 17;
2205            let source_field = lookup_index
2206                .field_name
2207                .split('.')
2208                .next_back()
2209                .unwrap_or(&lookup_index.field_name);
2210
2211            let start = ops.len();
2212            match resolution {
2213                KeyResolutionStrategy::Embedded { primary_field: _ } => {
2214                    // For Embedded handlers, find the mapping that targets this lookup index field
2215                    // and use its source path to load the lookup value
2216                    let source_path_opt =
2217                        self.find_source_path_for_lookup_index(mappings, &lookup_index.field_name);
2218
2219                    let load_path = if let Some(ref path) = source_path_opt {
2220                        FieldPath::new(&path.iter().map(|s| s.as_str()).collect::<Vec<_>>())
2221                    } else {
2222                        // Fallback to source_field if no mapping found
2223                        FieldPath::new(&[source_field])
2224                    };
2225
2226                    if let Some(mapping) = mappings
2227                        .iter()
2228                        .find(|m| m.target_path == lookup_index.field_name)
2229                    {
2230                        let value_ops = self.compile_mapping_source(&mapping.source, lookup_reg);
2231                        if value_ops.is_empty() {
2232                            continue;
2233                        }
2234                        ops.extend(value_ops);
2235                        if let Some(transform) = &mapping.transform {
2236                            ops.push(OpCode::Transform {
2237                                source: lookup_reg,
2238                                dest: lookup_reg,
2239                                transformation: transform.clone(),
2240                            });
2241                        }
2242                    } else {
2243                        ops.push(OpCode::LoadEventField {
2244                            path: load_path,
2245                            dest: lookup_reg,
2246                            default: None,
2247                        });
2248                    }
2249
2250                    if let Some(temporal_field_name) = &lookup_index.temporal_field {
2251                        let timestamp_reg = 18;
2252
2253                        ops.push(OpCode::LoadEventField {
2254                            path: FieldPath::new(&[temporal_field_name]),
2255                            dest: timestamp_reg,
2256                            default: None,
2257                        });
2258
2259                        let index_name = format!("{}_temporal_index", source_field);
2260                        ops.push(OpCode::UpdateTemporalIndex {
2261                            state_id: self.state_id,
2262                            index_name,
2263                            lookup_value: lookup_reg,
2264                            primary_key: key_reg,
2265                            timestamp: timestamp_reg,
2266                        });
2267
2268                        let simple_index_name = format!("{}_lookup_index", source_field);
2269                        ops.push(OpCode::UpdateLookupIndex {
2270                            state_id: self.state_id,
2271                            index_name: simple_index_name,
2272                            lookup_value: lookup_reg,
2273                            primary_key: key_reg,
2274                        });
2275                    } else {
2276                        let index_name = format!("{}_lookup_index", source_field);
2277                        ops.push(OpCode::UpdateLookupIndex {
2278                            state_id: self.state_id,
2279                            index_name,
2280                            lookup_value: lookup_reg,
2281                            primary_key: key_reg,
2282                        });
2283                    }
2284
2285                    // Also update PDA reverse lookup table if there's a resolver configured for this entity
2286                    // This allows instruction handlers to look up the primary key from PDA addresses
2287                    // Only do this when the source path is different (e.g., __account_address -> id.round_address)
2288                    if source_path_opt.is_some() {
2289                        ops.push(OpCode::UpdatePdaReverseLookup {
2290                            state_id: self.state_id,
2291                            lookup_name: "default_pda_lookup".to_string(),
2292                            pda_address: lookup_reg,
2293                            primary_key: key_reg,
2294                        });
2295                    }
2296                }
2297                KeyResolutionStrategy::Lookup { .. } => {
2298                    if let Some(mapping) = mappings
2299                        .iter()
2300                        .find(|m| m.target_path == lookup_index.field_name)
2301                    {
2302                        let value_ops = self.compile_mapping_source(&mapping.source, lookup_reg);
2303                        if value_ops.is_empty() {
2304                            continue;
2305                        }
2306                        ops.extend(value_ops);
2307                        if let Some(transform) = &mapping.transform {
2308                            ops.push(OpCode::Transform {
2309                                source: lookup_reg,
2310                                dest: lookup_reg,
2311                                transformation: transform.clone(),
2312                            });
2313                        }
2314                        ops.push(OpCode::UpdateLookupIndex {
2315                            state_id: self.state_id,
2316                            index_name: format!("{}_lookup_index", source_field),
2317                            lookup_value: lookup_reg,
2318                            primary_key: key_reg,
2319                        });
2320                    }
2321                }
2322                KeyResolutionStrategy::Computed { .. }
2323                | KeyResolutionStrategy::TemporalLookup { .. } => {
2324                    // Computed and TemporalLookup handlers don't populate lookup indexes
2325                }
2326            }
2327            // Every declared index follows its field's write guard, regardless
2328            // of how this handler resolved the row.
2329            let index_name = format!("{}_lookup_index", source_field);
2330            if self.declares_index(&index_name) {
2331                if let Some(guard) = index_guard(mappings, &lookup_index.field_name) {
2332                    let count = ops.len() - start;
2333                    ops.splice(
2334                        start..start,
2335                        [
2336                            OpCode::EvaluateEventExpr {
2337                                expr: guard,
2338                                transform: None,
2339                                dest: GUARD_REG,
2340                            },
2341                            OpCode::SkipUnless {
2342                                predicate: GUARD_REG,
2343                                count,
2344                            },
2345                        ],
2346                    );
2347                }
2348            }
2349        }
2350
2351        ops
2352    }
2353
2354    fn get_event_type(&self, source: &SourceSpec) -> String {
2355        match source {
2356            SourceSpec::Source { type_name, .. } => type_name.clone(),
2357        }
2358    }
2359
2360    fn compile_instruction_hook_actions(&self, actions: &[HookAction]) -> Vec<OpCode> {
2361        let mut ops = Vec::new();
2362        let state_reg = 2;
2363
2364        for action in actions {
2365            match action {
2366                HookAction::SetField {
2367                    target_field,
2368                    source,
2369                    condition,
2370                } => {
2371                    // Check if there's a condition - evaluation handled in VM
2372                    let _ = condition;
2373
2374                    let temp_reg = 11; // Use register 11 for hook values
2375
2376                    // Load the source value
2377                    let load_ops = self.compile_mapping_source(source, temp_reg);
2378                    ops.extend(load_ops);
2379
2380                    // Apply transformation if specified in source
2381                    if let MappingSource::FromSource {
2382                        transform: Some(transform_type),
2383                        ..
2384                    } = source
2385                    {
2386                        ops.push(OpCode::Transform {
2387                            source: temp_reg,
2388                            dest: temp_reg,
2389                            transformation: transform_type.clone(),
2390                        });
2391                    }
2392
2393                    // Conditionally set the field based on parsed condition
2394                    if let Some(cond_expr) = condition {
2395                        if let Some(parsed) = &cond_expr.parsed {
2396                            // Generate condition check opcodes
2397                            let cond_check_ops = self.compile_condition_check(
2398                                parsed,
2399                                temp_reg,
2400                                state_reg,
2401                                target_field,
2402                            );
2403                            ops.extend(cond_check_ops);
2404                        } else {
2405                            // No parsed condition, set unconditionally
2406                            ops.push(OpCode::SetField {
2407                                object: state_reg,
2408                                path: target_field.clone(),
2409                                value: temp_reg,
2410                            });
2411                        }
2412                    } else {
2413                        // No condition, set unconditionally
2414                        ops.push(OpCode::SetField {
2415                            object: state_reg,
2416                            path: target_field.clone(),
2417                            value: temp_reg,
2418                        });
2419                    }
2420                }
2421                HookAction::IncrementField {
2422                    target_field,
2423                    increment_by,
2424                    condition,
2425                } => {
2426                    if let Some(cond_expr) = condition {
2427                        if let Some(parsed) = &cond_expr.parsed {
2428                            // For increment with condition, we need to:
2429                            // 1. Load the condition field
2430                            // 2. Check the condition
2431                            // 3. Conditionally increment
2432                            let cond_check_ops = self.compile_conditional_increment(
2433                                parsed,
2434                                state_reg,
2435                                target_field,
2436                                *increment_by,
2437                            );
2438                            ops.extend(cond_check_ops);
2439                        } else {
2440                            // No parsed condition, increment unconditionally
2441                            ops.push(OpCode::SetFieldIncrement {
2442                                object: state_reg,
2443                                path: target_field.clone(),
2444                            });
2445                        }
2446                    } else {
2447                        // No condition, increment unconditionally
2448                        ops.push(OpCode::SetFieldIncrement {
2449                            object: state_reg,
2450                            path: target_field.clone(),
2451                        });
2452                    }
2453                }
2454                HookAction::RegisterPdaMapping {
2455                    pda_field,
2456                    seed_field,
2457                    lookup_name,
2458                    sibling,
2459                    seed_expr,
2460                    guard,
2461                } => {
2462                    let pda_reg = 11;
2463                    let seed_reg = 12;
2464                    let mut registration = Vec::new();
2465                    let pda_from_sibling = sibling
2466                        .as_ref()
2467                        .is_some_and(|sibling| sibling.pda_from_sibling);
2468                    let seed_from_sibling = sibling
2469                        .as_ref()
2470                        .is_some_and(|sibling| sibling.seed_from_sibling);
2471
2472                    // Fields read from this instruction are loaded now; a
2473                    // sibling's come from its capture.
2474                    registration.push(if pda_from_sibling {
2475                        OpCode::LoadConstant {
2476                            value: Value::Null,
2477                            dest: pda_reg,
2478                        }
2479                    } else {
2480                        OpCode::LoadEventField {
2481                            path: pda_field.clone(),
2482                            dest: pda_reg,
2483                            default: None,
2484                        }
2485                    });
2486                    registration.push(match (seed_from_sibling, seed_expr) {
2487                        (true, _) => OpCode::LoadConstant {
2488                            value: Value::Null,
2489                            dest: seed_reg,
2490                        },
2491                        (false, Some(expr)) => OpCode::EvaluateEventExpr {
2492                            expr: expr.clone(),
2493                            transform: None,
2494                            dest: seed_reg,
2495                        },
2496                        (false, None) => OpCode::LoadEventField {
2497                            path: seed_field.clone(),
2498                            dest: seed_reg,
2499                            default: None,
2500                        },
2501                    });
2502                    registration.push(match sibling {
2503                        None => OpCode::UpdatePdaReverseLookup {
2504                            state_id: self.state_id,
2505                            lookup_name: lookup_name.clone(),
2506                            pda_address: pda_reg,
2507                            primary_key: seed_reg,
2508                        },
2509                        Some(sibling) => OpCode::RegisterPdaFromSibling {
2510                            state_id: self.state_id,
2511                            lookup_name: lookup_name.clone(),
2512                            sibling: sibling.clone(),
2513                            pda_field: pda_field.clone(),
2514                            seed_field: seed_field.clone(),
2515                            pda_address: pda_reg,
2516                            primary_key: seed_reg,
2517                        },
2518                    });
2519
2520                    if let Some(guard) = guard {
2521                        ops.push(OpCode::EvaluateEventExpr {
2522                            expr: guard.clone(),
2523                            transform: None,
2524                            dest: GUARD_REG,
2525                        });
2526                        ops.push(OpCode::SkipUnless {
2527                            predicate: GUARD_REG,
2528                            count: registration.len(),
2529                        });
2530                    }
2531                    ops.extend(registration);
2532                }
2533            }
2534        }
2535
2536        ops
2537    }
2538
2539    fn compile_condition_check(
2540        &self,
2541        condition: &ParsedCondition,
2542        value_reg: Register,
2543        state_reg: Register,
2544        target_field: &str,
2545    ) -> Vec<OpCode> {
2546        match condition {
2547            ParsedCondition::Comparison {
2548                field,
2549                op,
2550                value: cond_value,
2551            } => {
2552                // Generate ConditionalSetField opcode
2553                vec![OpCode::ConditionalSetField {
2554                    object: state_reg,
2555                    path: target_field.to_string(),
2556                    value: value_reg,
2557                    condition_field: field.clone(),
2558                    condition_op: op.clone(),
2559                    condition_value: cond_value.clone(),
2560                }]
2561            }
2562            ParsedCondition::Logical { .. } => {
2563                // Logical conditions not yet supported, fall back to unconditional
2564                tracing::warn!("Logical conditions not yet supported in instruction hooks");
2565                vec![OpCode::SetField {
2566                    object: state_reg,
2567                    path: target_field.to_string(),
2568                    value: value_reg,
2569                }]
2570            }
2571            ParsedCondition::Guard { expr } => vec![
2572                OpCode::EvaluateEventExpr {
2573                    expr: expr.clone(),
2574                    transform: None,
2575                    dest: GUARD_REG,
2576                },
2577                OpCode::SkipUnless {
2578                    predicate: GUARD_REG,
2579                    count: 1,
2580                },
2581                OpCode::SetField {
2582                    object: state_reg,
2583                    path: target_field.to_string(),
2584                    value: value_reg,
2585                },
2586            ],
2587        }
2588    }
2589
2590    fn compile_conditional_increment(
2591        &self,
2592        condition: &ParsedCondition,
2593        state_reg: Register,
2594        target_field: &str,
2595        _increment_by: i64,
2596    ) -> Vec<OpCode> {
2597        match condition {
2598            ParsedCondition::Comparison {
2599                field,
2600                op,
2601                value: cond_value,
2602            } => {
2603                vec![OpCode::ConditionalIncrement {
2604                    object: state_reg,
2605                    path: target_field.to_string(),
2606                    condition_field: field.clone(),
2607                    condition_op: op.clone(),
2608                    condition_value: cond_value.clone(),
2609                }]
2610            }
2611            ParsedCondition::Logical { .. } => {
2612                tracing::warn!("Logical conditions not yet supported in instruction hooks");
2613                vec![OpCode::SetFieldIncrement {
2614                    object: state_reg,
2615                    path: target_field.to_string(),
2616                }]
2617            }
2618            ParsedCondition::Guard { expr } => vec![
2619                OpCode::EvaluateEventExpr {
2620                    expr: expr.clone(),
2621                    transform: None,
2622                    dest: GUARD_REG,
2623                },
2624                OpCode::SkipUnless {
2625                    predicate: GUARD_REG,
2626                    count: 1,
2627                },
2628                OpCode::SetFieldIncrement {
2629                    object: state_reg,
2630                    path: target_field.to_string(),
2631                },
2632            ],
2633        }
2634    }
2635}
2636
2637#[cfg(test)]
2638mod tests {
2639    use super::MultiEntityBytecode;
2640    use crate::ast::{
2641        FieldPath, HookAction, IdentitySpec, InstructionHook, KeyResolutionStrategy,
2642        LookupIndexSpec, MappingSource, PopulationStrategy, SerializableFieldMapping,
2643        SerializableHandlerSpec, SerializableStreamSpec, SourceSpec, TypedStreamSpec,
2644    };
2645    use crate::vm::VmContext;
2646    use crate::{Mutation, WholeEntity};
2647    use serde_json::{json, Value};
2648    use std::collections::BTreeMap;
2649
2650    fn mapping(
2651        target_path: &str,
2652        source_path: &[&str],
2653        population: PopulationStrategy,
2654    ) -> SerializableFieldMapping {
2655        SerializableFieldMapping {
2656            target_path: target_path.to_string(),
2657            source: MappingSource::FromSource {
2658                path: FieldPath::new(source_path),
2659                default: None,
2660                transform: None,
2661            },
2662            transform: None,
2663            population,
2664            condition: None,
2665            when: None,
2666            stop: None,
2667            emit: true,
2668        }
2669    }
2670
2671    fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
2672        TypedStreamSpec::from_serializable(SerializableStreamSpec {
2673            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2674            state_name: "PumpfunToken".to_string(),
2675            program_id: None,
2676            idl: None,
2677            identity: IdentitySpec {
2678                primary_keys: vec!["id.mint".to_string()],
2679                lookup_indexes: vec![LookupIndexSpec {
2680                    field_name: "id.bonding_curve".to_string(),
2681                    temporal_field: None,
2682                }],
2683            },
2684            handlers: vec![SerializableHandlerSpec {
2685                source: SourceSpec::Source {
2686                    program_id: None,
2687                    discriminator: None,
2688                    type_name: "pump::BondingCurveState".to_string(),
2689                    serialization: None,
2690                    is_account: true,
2691                },
2692                key_resolution: KeyResolutionStrategy::Lookup {
2693                    primary_field: FieldPath::new(&["__account_address"]),
2694                    index_name: None,
2695                },
2696                mappings: vec![
2697                    mapping(
2698                        "id.bonding_curve",
2699                        &["__account_address"],
2700                        PopulationStrategy::SetOnce,
2701                    ),
2702                    mapping(
2703                        "reserves.virtual_token_reserves",
2704                        &["virtual_token_reserves"],
2705                        PopulationStrategy::LastWrite,
2706                    ),
2707                ],
2708                conditions: vec![],
2709                emit: true,
2710            }],
2711            sections: vec![],
2712            field_mappings: BTreeMap::new(),
2713            resolver_hooks: vec![],
2714            instruction_hooks: vec![InstructionHook {
2715                instruction_type: "pump::BuyIxState".to_string(),
2716                actions: vec![HookAction::RegisterPdaMapping {
2717                    pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2718                    seed_field: FieldPath::new(&["accounts", "mint"]),
2719                    lookup_name: "default_pda_lookup".to_string(),
2720                    sibling: None,
2721                    seed_expr: None,
2722                    guard: None,
2723                }],
2724                lookup_by: None,
2725            }],
2726            resolver_specs: vec![],
2727            computed_fields: vec![],
2728            computed_field_specs: vec![],
2729            content_hash: None,
2730            views: vec![],
2731            scope_field: None,
2732        })
2733    }
2734
2735    /// A token whose bonding-curve account is keyed by the `id.bonding_curve`
2736    /// lookup index, which `LinkIxState` writes, or through the PDA mapping
2737    /// `BuyIxState` registers to the token's creator (indexed by
2738    /// `CreateIxState`), and whose round account is keyed by the numeric
2739    /// `id.round` index `OpenRoundIxState` writes. Both account handlers
2740    /// record an event, a capture and the update's context timestamp.
2741    fn timed_bonding_curve_spec() -> TypedStreamSpec<Value> {
2742        let with_source = |target: &str, source: MappingSource| SerializableFieldMapping {
2743            source,
2744            ..mapping(target, &[], PopulationStrategy::LastWrite)
2745        };
2746        let timestamped = || {
2747            vec![
2748                with_source(
2749                    "events.reserves",
2750                    MappingSource::AsEvent {
2751                        fields: vec![Box::new(MappingSource::FromSource {
2752                            path: FieldPath::new(&["virtual_token_reserves"]),
2753                            default: None,
2754                            transform: None,
2755                        })],
2756                    },
2757                ),
2758                with_source(
2759                    "snapshots.curve",
2760                    MappingSource::AsCapture {
2761                        field_transforms: BTreeMap::new(),
2762                    },
2763                ),
2764                with_source(
2765                    "state.seen_at",
2766                    MappingSource::FromContext {
2767                        field: "timestamp".to_string(),
2768                    },
2769                ),
2770            ]
2771        };
2772        let source = |type_name: &str, is_account: bool| SourceSpec::Source {
2773            program_id: None,
2774            discriminator: None,
2775            type_name: type_name.to_string(),
2776            serialization: None,
2777            is_account,
2778        };
2779        TypedStreamSpec::from_serializable(SerializableStreamSpec {
2780            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
2781            state_name: "PumpfunToken".to_string(),
2782            program_id: None,
2783            idl: None,
2784            identity: IdentitySpec {
2785                primary_keys: vec!["id.mint".to_string()],
2786                lookup_indexes: vec![
2787                    LookupIndexSpec {
2788                        field_name: "id.bonding_curve".to_string(),
2789                        temporal_field: None,
2790                    },
2791                    LookupIndexSpec {
2792                        field_name: "id.creator".to_string(),
2793                        temporal_field: None,
2794                    },
2795                    LookupIndexSpec {
2796                        field_name: "id.round".to_string(),
2797                        temporal_field: None,
2798                    },
2799                ],
2800            },
2801            handlers: vec![
2802                SerializableHandlerSpec {
2803                    source: source("pump::BondingCurveState", true),
2804                    key_resolution: KeyResolutionStrategy::Lookup {
2805                        primary_field: FieldPath::new(&["__account_address"]),
2806                        index_name: None,
2807                    },
2808                    mappings: [
2809                        vec![mapping(
2810                            "id.bonding_curve",
2811                            &["__account_address"],
2812                            PopulationStrategy::SetOnce,
2813                        )],
2814                        timestamped(),
2815                    ]
2816                    .concat(),
2817                    conditions: vec![],
2818                    emit: true,
2819                },
2820                SerializableHandlerSpec {
2821                    source: source("pump::RoundState", true),
2822                    key_resolution: KeyResolutionStrategy::Lookup {
2823                        primary_field: FieldPath::new(&["round"]),
2824                        index_name: None,
2825                    },
2826                    mappings: timestamped(),
2827                    conditions: vec![],
2828                    emit: true,
2829                },
2830                SerializableHandlerSpec {
2831                    source: source("pump::OpenRoundIxState", false),
2832                    key_resolution: KeyResolutionStrategy::Embedded {
2833                        primary_field: FieldPath::new(&["accounts", "mint"]),
2834                    },
2835                    mappings: vec![
2836                        mapping(
2837                            "id.mint",
2838                            &["accounts", "mint"],
2839                            PopulationStrategy::SetOnce,
2840                        ),
2841                        mapping(
2842                            "id.round",
2843                            &["accounts", "round"],
2844                            PopulationStrategy::SetOnce,
2845                        ),
2846                    ],
2847                    conditions: vec![],
2848                    emit: true,
2849                },
2850                SerializableHandlerSpec {
2851                    source: source("pump::LinkIxState", false),
2852                    key_resolution: KeyResolutionStrategy::Embedded {
2853                        primary_field: FieldPath::new(&["accounts", "mint"]),
2854                    },
2855                    mappings: vec![
2856                        mapping(
2857                            "id.mint",
2858                            &["accounts", "mint"],
2859                            PopulationStrategy::SetOnce,
2860                        ),
2861                        mapping(
2862                            "id.bonding_curve",
2863                            &["accounts", "bonding_curve"],
2864                            PopulationStrategy::SetOnce,
2865                        ),
2866                    ],
2867                    conditions: vec![],
2868                    emit: true,
2869                },
2870                SerializableHandlerSpec {
2871                    source: source("pump::CreateIxState", false),
2872                    key_resolution: KeyResolutionStrategy::Embedded {
2873                        primary_field: FieldPath::new(&["accounts", "mint"]),
2874                    },
2875                    mappings: vec![
2876                        mapping(
2877                            "id.mint",
2878                            &["accounts", "mint"],
2879                            PopulationStrategy::SetOnce,
2880                        ),
2881                        mapping(
2882                            "id.creator",
2883                            &["accounts", "user"],
2884                            PopulationStrategy::SetOnce,
2885                        ),
2886                    ],
2887                    conditions: vec![],
2888                    emit: true,
2889                },
2890            ],
2891            sections: vec![],
2892            field_mappings: BTreeMap::new(),
2893            resolver_hooks: vec![],
2894            instruction_hooks: vec![InstructionHook {
2895                instruction_type: "pump::BuyIxState".to_string(),
2896                actions: vec![HookAction::RegisterPdaMapping {
2897                    pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
2898                    seed_field: FieldPath::new(&["accounts", "user"]),
2899                    lookup_name: "default_pda_lookup".to_string(),
2900                    sibling: None,
2901                    seed_expr: None,
2902                    guard: None,
2903                }],
2904                lookup_by: None,
2905            }],
2906            resolver_specs: vec![],
2907            computed_fields: vec![],
2908            computed_field_specs: vec![],
2909            content_hash: None,
2910            views: vec![],
2911            scope_field: None,
2912        })
2913    }
2914
2915    /// A timed account update that waits for its key replays with its own
2916    /// time: the event, the capture and the `__timestamp` field all carry the
2917    /// update's timestamp, not the flush time or the flushing instruction's.
2918    /// The update waits for its lookup index entry (`LinkIxState`), for its
2919    /// PDA mapping (`BuyIxState`), for the index entry of its already mapped
2920    /// PDA's seed (`CreateIxState`), or for a numeric index entry, which is
2921    /// no PDA (`OpenRoundIxState`).
2922    #[test]
2923    fn a_queued_account_update_replays_with_its_own_timestamp() {
2924        use crate::vm::UpdateContext;
2925
2926        let accounts = json!({
2927            "bonding_curve": "bonding_curve_1",
2928            "mint": "mint_1",
2929            "user": "user_1",
2930            "round": 7,
2931        });
2932        let curve = (
2933            "pump::BondingCurveState",
2934            json!({ "__account_address": "bonding_curve_1", "virtual_token_reserves": 42 }),
2935        );
2936        let round = (
2937            "pump::RoundState",
2938            json!({ "__account_address": "round_1", "round": 7, "virtual_token_reserves": 42 }),
2939        );
2940        let instruction = |vm: &mut VmContext, bytecode: &MultiEntityBytecode, name: &str| {
2941            let mut context =
2942                UpdateContext::new_instruction(20, "instruction-signature".to_string(), 0);
2943            context.timestamp = Some(2_000);
2944            vm.process_event(
2945                bytecode,
2946                json!({ "accounts": accounts }),
2947                name,
2948                Some(&context),
2949                None,
2950            )
2951            .unwrap();
2952        };
2953        // (instructions before the account update, the account update,
2954        // instructions after it)
2955        let scenarios: [(&[&str], _, &[&str]); 4] = [
2956            (&[], &curve, &["pump::LinkIxState"]),
2957            (&[], &curve, &["pump::CreateIxState", "pump::BuyIxState"]),
2958            (&["pump::BuyIxState"], &curve, &["pump::CreateIxState"]),
2959            (&[], &round, &["pump::OpenRoundIxState"]),
2960        ];
2961        for (before, (account_type, account), after) in scenarios {
2962            let bytecode = MultiEntityBytecode::from_single(
2963                "PumpfunToken".to_string(),
2964                timed_bonding_curve_spec(),
2965                0,
2966            );
2967            let mut vm = VmContext::new();
2968            for name in before {
2969                instruction(&mut vm, &bytecode, name);
2970            }
2971            let mut account_context =
2972                UpdateContext::new_account(10, "curve-signature".to_string(), 1);
2973            account_context.timestamp = Some(1_000);
2974
2975            let queued = vm
2976                .process_event(
2977                    &bytecode,
2978                    account.clone(),
2979                    account_type,
2980                    Some(&account_context),
2981                    None,
2982                )
2983                .unwrap();
2984            assert!(queued.is_empty(), "{after:?}: {queued:?}");
2985
2986            // Flush after a wall-clock second has passed, under instructions
2987            // with a later timestamp of their own.
2988            std::thread::sleep(std::time::Duration::from_millis(1_100));
2989            for name in after {
2990                instruction(&mut vm, &bytecode, name);
2991            }
2992
2993            let token = vm
2994                .get_entity_state(0, &json!("mint_1"))
2995                .unwrap_or_else(|| panic!("{after:?}: no entity"));
2996            for path in [
2997                "/events/reserves/timestamp",
2998                "/snapshots/curve/timestamp",
2999                "/state/seen_at",
3000            ] {
3001                assert_eq!(
3002                    token.pointer(path),
3003                    Some(&json!(1_000)),
3004                    "{before:?} {after:?} {path}: {token:#}"
3005                );
3006            }
3007            assert_eq!(
3008                token.pointer("/events/reserves/data/virtual_token_reserves"),
3009                Some(&json!(42))
3010            );
3011            assert_eq!(token.pointer("/events/reserves/slot"), Some(&json!(10)));
3012            assert_eq!(
3013                token.pointer("/snapshots/curve/signature"),
3014                Some(&json!("curve-signature"))
3015            );
3016        }
3017    }
3018
3019    fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
3020        TypedStreamSpec::from_serializable(SerializableStreamSpec {
3021            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3022            state_name: "OreRound".to_string(),
3023            program_id: None,
3024            idl: None,
3025            identity: IdentitySpec {
3026                primary_keys: vec!["id.round_id".to_string()],
3027                lookup_indexes: vec![LookupIndexSpec {
3028                    field_name: "id.round_address".to_string(),
3029                    temporal_field: None,
3030                }],
3031            },
3032            handlers: vec![SerializableHandlerSpec {
3033                source: SourceSpec::Source {
3034                    program_id: None,
3035                    discriminator: None,
3036                    type_name: "entropy::VarState".to_string(),
3037                    serialization: None,
3038                    is_account: true,
3039                },
3040                key_resolution: KeyResolutionStrategy::Lookup {
3041                    primary_field: FieldPath::new(&["__account_address"]),
3042                    index_name: None,
3043                },
3044                mappings: vec![mapping(
3045                    "state.expires_at",
3046                    &["end_at"],
3047                    PopulationStrategy::LastWrite,
3048                )],
3049                conditions: vec![],
3050                emit: true,
3051            }],
3052            sections: vec![],
3053            field_mappings: BTreeMap::new(),
3054            resolver_hooks: vec![],
3055            instruction_hooks: vec![InstructionHook {
3056                instruction_type: "ore::DeployIxState".to_string(),
3057                actions: vec![HookAction::RegisterPdaMapping {
3058                    pda_field: FieldPath::new(&["accounts", "entropyVar"]),
3059                    seed_field: FieldPath::new(&["accounts", "round"]),
3060                    lookup_name: "default_pda_lookup".to_string(),
3061                    sibling: None,
3062                    seed_expr: None,
3063                    guard: None,
3064                }],
3065                lookup_by: None,
3066            }],
3067            resolver_specs: vec![],
3068            computed_fields: vec![],
3069            computed_field_specs: vec![],
3070            content_hash: None,
3071            views: vec![],
3072            scope_field: None,
3073        })
3074    }
3075
3076    fn embedded_account_key_spec() -> TypedStreamSpec<Value> {
3077        TypedStreamSpec::from_serializable(SerializableStreamSpec {
3078            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3079            state_name: "OreTreasury".to_string(),
3080            program_id: None,
3081            idl: None,
3082            identity: IdentitySpec {
3083                primary_keys: vec!["id.address".to_string()],
3084                lookup_indexes: vec![],
3085            },
3086            handlers: vec![SerializableHandlerSpec {
3087                source: SourceSpec::Source {
3088                    program_id: None,
3089                    discriminator: None,
3090                    type_name: "ore::TreasuryState".to_string(),
3091                    serialization: None,
3092                    is_account: true,
3093                },
3094                key_resolution: KeyResolutionStrategy::Embedded {
3095                    primary_field: FieldPath::new(&["__account_address"]),
3096                },
3097                mappings: vec![mapping(
3098                    "id.address",
3099                    &["__account_address"],
3100                    PopulationStrategy::SetOnce,
3101                )],
3102                conditions: vec![],
3103                emit: true,
3104            }],
3105            sections: vec![],
3106            field_mappings: BTreeMap::new(),
3107            resolver_hooks: vec![],
3108            instruction_hooks: vec![],
3109            resolver_specs: vec![],
3110            computed_fields: vec![],
3111            computed_field_specs: vec![],
3112            content_hash: None,
3113            views: vec![],
3114            scope_field: None,
3115        })
3116    }
3117
3118    /// A pool whose account carries a price (replaced), a fill (appended) and
3119    /// a secret that is mapped but never emitted, plus a config account whose
3120    /// handler emits nothing and sets a fee.
3121    fn whole_entity_pool_spec() -> TypedStreamSpec<Value> {
3122        let mut hidden = mapping("state.secret", &["secret"], PopulationStrategy::LastWrite);
3123        hidden.emit = false;
3124        let handler = |type_name: &str, mappings, emit| SerializableHandlerSpec {
3125            source: SourceSpec::Source {
3126                program_id: None,
3127                discriminator: None,
3128                type_name: type_name.to_string(),
3129                serialization: None,
3130                is_account: true,
3131            },
3132            key_resolution: KeyResolutionStrategy::Embedded {
3133                primary_field: FieldPath::new(&["__account_address"]),
3134            },
3135            mappings,
3136            conditions: vec![],
3137            emit,
3138        };
3139        TypedStreamSpec::from_serializable(SerializableStreamSpec {
3140            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3141            state_name: "Pool".to_string(),
3142            program_id: None,
3143            idl: None,
3144            identity: IdentitySpec {
3145                primary_keys: vec!["id.address".to_string()],
3146                lookup_indexes: vec![],
3147            },
3148            handlers: vec![
3149                handler(
3150                    "amm::PoolState",
3151                    vec![
3152                        mapping(
3153                            "id.address",
3154                            &["__account_address"],
3155                            PopulationStrategy::SetOnce,
3156                        ),
3157                        mapping("state.price", &["price"], PopulationStrategy::LastWrite),
3158                        mapping("state.fills", &["fill"], PopulationStrategy::Append),
3159                        hidden,
3160                    ],
3161                    true,
3162                ),
3163                handler(
3164                    "amm::PoolConfigState",
3165                    vec![mapping(
3166                        "config.fee",
3167                        &["fee"],
3168                        PopulationStrategy::LastWrite,
3169                    )],
3170                    false,
3171                ),
3172            ],
3173            sections: vec![],
3174            field_mappings: BTreeMap::new(),
3175            resolver_hooks: vec![],
3176            instruction_hooks: vec![],
3177            resolver_specs: vec![],
3178            computed_fields: vec![],
3179            computed_field_specs: vec![],
3180            content_hash: None,
3181            views: vec![],
3182            scope_field: None,
3183        })
3184    }
3185
3186    fn pool_bytecode() -> MultiEntityBytecode {
3187        MultiEntityBytecode::from_single("Pool".to_string(), whole_entity_pool_spec(), 0)
3188    }
3189
3190    /// Event (`#[event]`), capture (`#[snapshot]`) and context (`__timestamp`)
3191    /// mappings compiled from a spec are stamped with the update's timestamp,
3192    /// so replaying an update reproduces them.
3193    #[test]
3194    fn compiled_event_capture_and_context_mappings_carry_the_update_timestamp() {
3195        let with_source = |target: &str, source: MappingSource| SerializableFieldMapping {
3196            source,
3197            ..mapping(target, &[], PopulationStrategy::LastWrite)
3198        };
3199        let spec: TypedStreamSpec<Value> =
3200            TypedStreamSpec::from_serializable(SerializableStreamSpec {
3201                ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3202                state_name: "Pool".to_string(),
3203                program_id: None,
3204                idl: None,
3205                identity: IdentitySpec {
3206                    primary_keys: vec!["id.address".to_string()],
3207                    lookup_indexes: vec![],
3208                },
3209                handlers: vec![SerializableHandlerSpec {
3210                    source: SourceSpec::Source {
3211                        program_id: None,
3212                        discriminator: None,
3213                        type_name: "amm::PoolState".to_string(),
3214                        serialization: None,
3215                        is_account: true,
3216                    },
3217                    key_resolution: KeyResolutionStrategy::Embedded {
3218                        primary_field: FieldPath::new(&["__account_address"]),
3219                    },
3220                    mappings: vec![
3221                        mapping(
3222                            "id.address",
3223                            &["__account_address"],
3224                            PopulationStrategy::SetOnce,
3225                        ),
3226                        with_source(
3227                            "events.priced",
3228                            MappingSource::AsEvent {
3229                                fields: vec![Box::new(MappingSource::FromSource {
3230                                    path: FieldPath::new(&["price"]),
3231                                    default: None,
3232                                    transform: None,
3233                                })],
3234                            },
3235                        ),
3236                        with_source(
3237                            "snapshots.pool",
3238                            MappingSource::AsCapture {
3239                                field_transforms: BTreeMap::new(),
3240                            },
3241                        ),
3242                        with_source(
3243                            "state.updated_at",
3244                            MappingSource::FromContext {
3245                                field: "timestamp".to_string(),
3246                            },
3247                        ),
3248                    ],
3249                    conditions: vec![],
3250                    emit: true,
3251                }],
3252                sections: vec![],
3253                field_mappings: BTreeMap::new(),
3254                resolver_hooks: vec![],
3255                instruction_hooks: vec![],
3256                resolver_specs: vec![],
3257                computed_fields: vec![],
3258                computed_field_specs: vec![],
3259                content_hash: None,
3260                views: vec![],
3261                scope_field: None,
3262            });
3263        let bytecode = MultiEntityBytecode::from_single("Pool".to_string(), spec, 0);
3264        let context = crate::vm::UpdateContext::with_timestamp(7, "sig-7".to_string(), 1_234_567);
3265
3266        let mut vm = VmContext::new();
3267        vm.process_event(
3268            &bytecode,
3269            pool_event(100, 1),
3270            "amm::PoolState",
3271            Some(&context),
3272            None,
3273        )
3274        .unwrap();
3275
3276        let pool = vm.get_entity_state(0, &json!("pool_1")).unwrap();
3277        for path in [
3278            "/events/priced/timestamp",
3279            "/snapshots/pool/timestamp",
3280            "/state/updated_at",
3281        ] {
3282            assert_eq!(
3283                pool.pointer(path),
3284                Some(&json!(1_234_567)),
3285                "{path}: {pool:#}"
3286            );
3287        }
3288        assert_eq!(pool.pointer("/events/priced/data/price"), Some(&json!(100)));
3289        assert_eq!(pool.pointer("/snapshots/pool/slot"), Some(&json!(7)));
3290    }
3291
3292    fn pool_event(price: u64, fill: u64) -> Value {
3293        pool_event_for("pool_1", price, fill)
3294    }
3295
3296    fn pool_event_for(address: &str, price: u64, fill: u64) -> Value {
3297        json!({"__account_address": address, "price": price, "fill": fill, "secret": "s"})
3298    }
3299
3300    /// A VM that answers whole-entity requests, and so marks creations.
3301    fn linked_vm(
3302        state_config: crate::vm::StateTableConfig,
3303    ) -> (VmContext, crate::vm::WholeEntityRequests) {
3304        let requests = crate::vm::WholeEntityRequests::new(16);
3305        let mut vm = VmContext::new_with_config(state_config);
3306        vm.set_whole_entity_requests(requests.clone());
3307        (vm, requests)
3308    }
3309
3310    fn process(vm: &mut VmContext, bytecode: &MultiEntityBytecode, event: Value) -> Vec<Mutation> {
3311        vm.process_event(bytecode, event, "amm::PoolState", None, None)
3312            .unwrap()
3313    }
3314
3315    #[test]
3316    fn source_deletion_cancels_resync_and_recreates_a_fresh_vm_row() {
3317        let bytecode = pool_bytecode();
3318        let (mut vm, requests) = linked_vm(Default::default());
3319        let first = process(&mut vm, &bytecode, pool_event(1, 10));
3320        let key = &first[0].key;
3321        let target = crate::vm::ResolverTarget {
3322            state_id: 0,
3323            entity_name: "Pool".into(),
3324            primary_key: key.clone(),
3325            extracts: vec![],
3326        };
3327        vm.enqueue_resolver_request(
3328            "ignored".into(),
3329            crate::ast::ResolverType::Token,
3330            json!("mint"),
3331            target.clone(),
3332        );
3333        let old_request = vm.take_resolver_requests().pop().unwrap();
3334        vm.get_state_table_mut(0).unwrap().deferred_when_ops.insert(
3335            ("sig".into(), "instruction".into()),
3336            vec![crate::vm::DeferredWhenOperation {
3337                entity_name: "Pool".into(),
3338                primary_key: key.clone(),
3339                field_path: "old".into(),
3340                field_value: json!(true),
3341                when_instruction: "instruction".into(),
3342                signature: "sig".into(),
3343                slot: 1,
3344                deferred_at: 1,
3345                emit: true,
3346            }],
3347        );
3348        requests.request("Pool", key);
3349        let deletion = vm.delete_entity(&bytecode, "Pool", key).unwrap();
3350        assert!(deletion.is_delete());
3351        assert!(
3352            serde_json::from_value::<Mutation>(serde_json::to_value(&deletion).unwrap())
3353                .unwrap()
3354                .is_delete()
3355        );
3356        assert!(!requests.is_requested("Pool", key));
3357        assert!(vm.get_entity_state(0, key).is_none());
3358        assert!(vm.snapshot_state_table(0).is_empty());
3359        assert!(vm
3360            .get_state_table_mut(0)
3361            .unwrap()
3362            .deferred_when_ops
3363            .is_empty());
3364        assert!(vm.delete_entity(&bytecode, "Unknown", key).is_none());
3365        let mut next = process(&mut vm, &bytecode, pool_event(9, 90));
3366        vm.enqueue_resolver_request(
3367            "ignored".into(),
3368            crate::ast::ResolverType::Token,
3369            json!("mint"),
3370            target,
3371        );
3372        let new_request = vm.take_resolver_requests().pop().unwrap();
3373        assert_ne!(old_request.cache_key, new_request.cache_key);
3374        assert!(vm
3375            .apply_resolver_result(&bytecode, &old_request.cache_key, json!({"stale":true}))
3376            .unwrap()
3377            .is_empty());
3378        assert_eq!(next[0].take_whole_entity_mark(), Some(WholeEntity::Created));
3379        assert_eq!(next[0].patch["state"]["fills"], json!([90]));
3380    }
3381
3382    #[test]
3383    fn deletion_preserves_unrelated_resolver_sharing_and_isolates_recreation() {
3384        let bytecode = pool_bytecode();
3385        let (mut vm, _) = linked_vm(Default::default());
3386        let key = process(&mut vm, &bytecode, pool_event(1, 10))[0]
3387            .key
3388            .clone();
3389        let late_joiner = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20))[0]
3390            .key
3391            .clone();
3392        let target = |primary_key| crate::vm::ResolverTarget {
3393            state_id: 0,
3394            entity_name: "Pool".into(),
3395            primary_key,
3396            extracts: vec![crate::ast::ResolverExtractSpec {
3397                target_path: "resolved".into(),
3398                source_path: Some("value".into()),
3399                transform: None,
3400            }],
3401        };
3402        let enqueue = |vm: &mut VmContext, key| {
3403            vm.enqueue_resolver_request(
3404                "ignored".into(),
3405                crate::ast::ResolverType::Token,
3406                json!("shared-mint"),
3407                target(key),
3408            )
3409        };
3410        enqueue(&mut vm, key.clone());
3411        enqueue(&mut vm, json!("unrelated"));
3412        let old = vm.take_resolver_requests();
3413        assert_eq!(old.len(), 1);
3414        vm.delete_entity(&bytecode, "Pool", &key).unwrap();
3415        enqueue(&mut vm, json!("unrelated"));
3416        enqueue(&mut vm, json!("another-unrelated"));
3417        assert!(
3418            vm.take_resolver_requests().is_empty(),
3419            "unaffected targets reuse the old pending call"
3420        );
3421        enqueue(&mut vm, key.clone());
3422        let fresh = vm.take_resolver_requests();
3423        assert_eq!(fresh.len(), 1);
3424        assert_ne!(fresh[0].cache_key, old[0].cache_key);
3425        // This entity never joined the old shared call. Deleting and
3426        // recreating it must still keep it off that pre-deletion result.
3427        vm.delete_entity(&bytecode, "Pool", &late_joiner).unwrap();
3428        process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
3429        enqueue(&mut vm, late_joiner.clone());
3430        let late_joiner_fresh = vm.take_resolver_requests();
3431        assert_eq!(late_joiner_fresh.len(), 1);
3432        assert_ne!(late_joiner_fresh[0].cache_key, old[0].cache_key);
3433        enqueue(&mut vm, json!("unrelated"));
3434        enqueue(&mut vm, json!("another-unrelated"));
3435        enqueue(&mut vm, key.clone());
3436        assert!(
3437            vm.take_resolver_requests().is_empty(),
3438            "unaffected targets reuse pending calls"
3439        );
3440        let old_mutations = vm
3441            .apply_resolver_result(&bytecode, &old[0].cache_key, json!({"value":"old"}))
3442            .unwrap();
3443        assert_eq!(old_mutations.len(), 2);
3444        assert!(old_mutations.iter().all(|mutation| mutation.key != key));
3445        assert!(vm.get_entity_state(0, &key).is_none());
3446        let fresh_mutations = vm
3447            .apply_resolver_result(&bytecode, &fresh[0].cache_key, json!({"value":"fresh"}))
3448            .unwrap();
3449        assert_eq!(fresh_mutations.len(), 1);
3450        assert_eq!(fresh_mutations[0].key, key);
3451        let late_joiner_mutations = vm
3452            .apply_resolver_result(
3453                &bytecode,
3454                &late_joiner_fresh[0].cache_key,
3455                json!({"value":"fresh-late-joiner"}),
3456            )
3457            .unwrap();
3458        assert_eq!(late_joiner_mutations.len(), 1);
3459        assert_eq!(late_joiner_mutations[0].key, late_joiner);
3460    }
3461
3462    #[test]
3463    fn a_requested_entity_is_emitted_whole_once() {
3464        let bytecode = pool_bytecode();
3465        let (mut vm, requests) = linked_vm(Default::default());
3466        let mut update = |price, fill| process(&mut vm, &bytecode, pool_event(price, fill));
3467
3468        let first = update(1, 10);
3469        assert_eq!(first[0].whole_entity_mark(), Some(WholeEntity::Created));
3470        let key = first[0].key.clone();
3471        let second = update(2, 20);
3472        assert_eq!(second.len(), 1);
3473        assert!(!second[0].is_whole_entity());
3474        assert_eq!(
3475            second[0].patch,
3476            json!({"state": {"price": 2, "fills": [20]}})
3477        );
3478        assert_eq!(second[0].append, vec!["state.fills".to_string()]);
3479
3480        // A consumer that dropped the entity asks for all of it.
3481        requests.request("Pool", &key);
3482        assert!(requests.is_requested("Pool", &key));
3483        let mut third = update(3, 30);
3484        assert_eq!(third.len(), 2);
3485        assert!(
3486            !third[0].is_whole_entity(),
3487            "the change itself is untouched"
3488        );
3489        assert_eq!(
3490            third[0].patch,
3491            json!({"state": {"price": 3, "fills": [30]}})
3492        );
3493        assert_eq!(third[1].key, key);
3494        assert_eq!(third[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
3495        assert_eq!(
3496            third[1].patch,
3497            json!({
3498                "id": {"address": "pool_1"},
3499                "state": {"price": 3, "fills": [10, 20, 30]},
3500            }),
3501            "the whole entity, without its non-emitted field"
3502        );
3503        assert!(
3504            third[1].append.is_empty(),
3505            "its arrays are whole, not appended to"
3506        );
3507        assert!(requests.is_empty(), "a request is answered once");
3508
3509        let fourth = update(4, 40);
3510        assert_eq!(fourth.len(), 1);
3511        assert!(!fourth[0].is_whole_entity());
3512        assert_eq!(
3513            fourth[0].patch,
3514            json!({"state": {"price": 4, "fills": [40]}})
3515        );
3516    }
3517
3518    /// A request does not wait for its entity to change: the VM's next call
3519    /// ends with the whole entity, after whatever that call emitted.
3520    #[test]
3521    fn a_request_is_answered_by_the_next_call_whatever_it_changes() {
3522        let bytecode = pool_bytecode();
3523        let (mut vm, requests) = linked_vm(Default::default());
3524        process(&mut vm, &bytecode, pool_event_for("pool_1", 1, 10));
3525        process(&mut vm, &bytecode, pool_event_for("pool_2", 1, 10));
3526
3527        requests.request("Pool", &json!("pool_1"));
3528        let mut next = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20));
3529        assert_eq!(next.len(), 2);
3530        assert_eq!(next[0].key, json!("pool_2"));
3531        assert!(!next[0].is_whole_entity());
3532        assert_eq!(next[1].key, json!("pool_1"));
3533        assert_eq!(next[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
3534        assert_eq!(
3535            next[1].patch,
3536            json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
3537        );
3538        assert!(requests.is_empty());
3539
3540        // Even a call that changes nothing answers.
3541        requests.request("Pool", &json!("pool_2"));
3542        let answered = vm
3543            .apply_resolver_result(&bytecode, "no-such-request", json!({}))
3544            .unwrap();
3545        assert_eq!(answered.len(), 1);
3546        assert_eq!(answered[0].key, json!("pool_2"));
3547        assert_eq!(answered[0].whole_entity_mark(), Some(WholeEntity::Resent));
3548
3549        let quiet = process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
3550        assert_eq!(quiet.len(), 1, "a request is answered once");
3551    }
3552
3553    /// The VM has nothing to send for an entity it does not hold. It dropped
3554    /// the entity too, so the key's next mutation creates it again.
3555    #[test]
3556    fn a_request_for_an_entity_the_vm_does_not_hold_is_dropped() {
3557        let bytecode = pool_bytecode();
3558        let (mut vm, requests) = linked_vm(Default::default());
3559        requests.request("Pool", &json!("other_pool"));
3560        let mutations = process(&mut vm, &bytecode, pool_event(1, 10));
3561        assert_eq!(mutations.len(), 1);
3562        assert_eq!(mutations[0].whole_entity_mark(), Some(WholeEntity::Created));
3563        assert!(requests.is_empty(), "dropped, not kept for later");
3564
3565        let created = process(&mut vm, &bytecode, pool_event_for("other_pool", 1, 10));
3566        assert_eq!(created.len(), 1);
3567        assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
3568    }
3569
3570    #[test]
3571    fn whole_entity_requests_are_bounded() {
3572        let requests = crate::vm::WholeEntityRequests::new(2);
3573        for key in ["a", "b", "c"] {
3574            requests.request("Pool", &json!(key));
3575        }
3576        assert_eq!(requests.len(), 2);
3577        assert!(
3578            !requests.is_requested("Pool", &json!("a")),
3579            "the oldest is forgotten"
3580        );
3581        assert!(requests.is_requested("Pool", &json!("c")));
3582    }
3583
3584    /// A consumer that does not hold a key can tell a new entity from a later
3585    /// patch for one it dropped: only the first is marked created.
3586    #[test]
3587    fn an_entitys_first_mutation_is_marked_as_its_creation() {
3588        let bytecode = pool_bytecode();
3589        let (mut vm, _requests) = linked_vm(Default::default());
3590
3591        let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
3592        assert_eq!(first.len(), 1, "the creation needs no second mutation");
3593        assert_eq!(
3594            first[0].take_whole_entity_mark(),
3595            Some(WholeEntity::Created)
3596        );
3597        assert_eq!(
3598            first[0].patch,
3599            json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}}),
3600            "the whole row, without its non-emitted field"
3601        );
3602        assert_eq!(first[0].append, vec!["state.fills".to_string()]);
3603
3604        let second = process(&mut vm, &bytecode, pool_event(2, 20));
3605        assert_eq!(second[0].whole_entity_mark(), None);
3606        assert_eq!(
3607            second[0].patch,
3608            json!({"state": {"price": 2, "fills": [20]}})
3609        );
3610    }
3611
3612    /// Without a consumer asking for whole entities the VM's output is
3613    /// unchanged: nothing is marked, and a creation is its changed fields.
3614    #[test]
3615    fn a_vm_without_whole_entity_requests_marks_no_creation() {
3616        let bytecode = pool_bytecode();
3617        let mut vm = VmContext::new();
3618        let first = process(&mut vm, &bytecode, pool_event(1, 10));
3619        assert!(!first[0].is_whole_entity());
3620        assert!(first[0].patch.get(crate::WHOLE_ENTITY_MARKER).is_none());
3621    }
3622
3623    /// A handler that emits nothing can start the row. The first mutation
3624    /// after it is still the creation, and carries what that handler set,
3625    /// which its own change does not.
3626    #[test]
3627    fn a_row_started_without_a_mutation_is_created_by_its_first_one() {
3628        let bytecode = pool_bytecode();
3629        let (mut vm, _requests) = linked_vm(Default::default());
3630        let config = vm
3631            .process_event(
3632                &bytecode,
3633                json!({"__account_address": "pool_1", "fee": 30}),
3634                "amm::PoolConfigState",
3635                None,
3636                None,
3637            )
3638            .unwrap();
3639        assert!(config.is_empty(), "the config handler emits nothing");
3640
3641        let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
3642        assert_eq!(
3643            first[0].take_whole_entity_mark(),
3644            Some(WholeEntity::Created)
3645        );
3646        assert_eq!(
3647            first[0].patch,
3648            json!({
3649                "id": {"address": "pool_1"},
3650                "config": {"fee": 30},
3651                "state": {"price": 1, "fills": [10]},
3652            })
3653        );
3654        let second = process(&mut vm, &bytecode, pool_event(2, 20));
3655        assert!(!second[0].is_whole_entity());
3656    }
3657
3658    /// An entity the VM's bounded table dropped starts again from its next
3659    /// event, and that event's mutation creates it again.
3660    #[test]
3661    fn an_entity_the_vm_dropped_is_created_again() {
3662        let bytecode = pool_bytecode();
3663        let (mut vm, _requests) = linked_vm(crate::vm::StateTableConfig {
3664            max_entries: 2,
3665            ..Default::default()
3666        });
3667        for pool in ["a", "b", "c"] {
3668            let created = process(&mut vm, &bytecode, pool_event_for(pool, 1, 1));
3669            assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
3670        }
3671        assert!(vm.get_entity_state(0, &json!("a")).is_none(), "dropped");
3672
3673        let mut again = process(&mut vm, &bytecode, pool_event_for("a", 2, 2));
3674        assert_eq!(
3675            again[0].take_whole_entity_mark(),
3676            Some(WholeEntity::Created)
3677        );
3678        assert_eq!(
3679            again[0].patch,
3680            json!({"id": {"address": "a"}, "state": {"price": 2, "fills": [2]}})
3681        );
3682        let changed = process(&mut vm, &bytecode, pool_event_for("c", 2, 2));
3683        assert!(!changed[0].is_whole_entity(), "c was never dropped");
3684    }
3685
3686    /// A row a non-emitting handler started and a snapshot then took before
3687    /// its first mutation comes back as emitted: its first update after the
3688    /// restore is a plain patch. A consumer that never held the key asks for
3689    /// it, and the VM's next call sends it whole, with what that handler set.
3690    #[test]
3691    fn a_restored_row_never_emitted_is_sent_whole_on_request() {
3692        let bytecode = pool_bytecode();
3693        let (mut vm, _requests) = linked_vm(Default::default());
3694        let config = vm
3695            .process_event(
3696                &bytecode,
3697                json!({"__account_address": "pool_1", "fee": 30}),
3698                "amm::PoolConfigState",
3699                None,
3700                None,
3701            )
3702            .unwrap();
3703        assert!(config.is_empty());
3704        let snapshot = vm.dump();
3705
3706        let (mut restored, requests) = linked_vm(Default::default());
3707        restored.hydrate(snapshot);
3708        let first = process(&mut restored, &bytecode, pool_event(1, 10));
3709        assert!(!first[0].is_whole_entity(), "counted as emitted already");
3710        assert_eq!(
3711            first[0].patch,
3712            json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
3713        );
3714
3715        requests.request("Pool", &first[0].key);
3716        let next = process(&mut restored, &bytecode, pool_event_for("pool_2", 1, 1));
3717        let resent = next
3718            .iter()
3719            .find(|mutation| mutation.key == json!("pool_1"))
3720            .expect("the requested entity follows the next call");
3721        assert_eq!(resent.whole_entity_mark(), Some(WholeEntity::Resent));
3722        let mut resent = resent.clone();
3723        resent.take_whole_entity_mark();
3724        assert_eq!(
3725            resent.patch,
3726            json!({
3727                "id": {"address": "pool_1"},
3728                "config": {"fee": 30},
3729                "state": {"price": 1, "fills": [10]},
3730            })
3731        );
3732    }
3733
3734    /// A resolver result for an entity the table does not hold starts the
3735    /// entity, so its mutation is the creation; one for an entity the table
3736    /// holds is a patch.
3737    #[test]
3738    fn a_resolver_result_for_an_entity_not_held_creates_it() {
3739        let bytecode = pool_bytecode();
3740        let (mut vm, _requests) = linked_vm(Default::default());
3741        process(&mut vm, &bytecode, pool_event(1, 10));
3742        for key in ["pool_1", "pool_2"] {
3743            vm.enqueue_resolver_request(
3744                String::new(),
3745                crate::ast::ResolverType::Token,
3746                json!(key),
3747                crate::vm::ResolverTarget {
3748                    state_id: 0,
3749                    entity_name: "Pool".to_string(),
3750                    primary_key: json!(key),
3751                    extracts: vec![crate::ast::ResolverExtractSpec {
3752                        target_path: "meta.name".to_string(),
3753                        source_path: Some("name".to_string()),
3754                        transform: None,
3755                    }],
3756                },
3757            );
3758        }
3759        let mut applied = Vec::new();
3760        for request in vm.take_resolver_requests() {
3761            applied.extend(
3762                vm.apply_resolver_result(&bytecode, &request.cache_key, json!({"name": "n"}))
3763                    .unwrap(),
3764            );
3765        }
3766        applied.sort_by_key(|mutation| mutation.key.to_string());
3767
3768        assert_eq!(applied[0].key, json!("pool_1"));
3769        assert!(!applied[0].is_whole_entity());
3770        assert_eq!(applied[0].patch, json!({"meta": {"name": "n"}}));
3771        assert_eq!(applied[1].key, json!("pool_2"));
3772        assert_eq!(
3773            applied[1].take_whole_entity_mark(),
3774            Some(WholeEntity::Created)
3775        );
3776        assert_eq!(applied[1].patch, json!({"meta": {"name": "n"}}));
3777    }
3778
3779    /// Entities restored from a snapshot may be held by the consumer already,
3780    /// so their next mutation is a patch like any other.
3781    #[test]
3782    fn a_restored_entity_is_not_created_again() {
3783        let bytecode = pool_bytecode();
3784        let (mut vm, _requests) = linked_vm(Default::default());
3785        process(&mut vm, &bytecode, pool_event(1, 10));
3786        let snapshot = vm.dump();
3787
3788        let (mut restored, _requests) = linked_vm(Default::default());
3789        restored.hydrate(snapshot);
3790        let next = process(&mut restored, &bytecode, pool_event(2, 20));
3791        assert!(!next[0].is_whole_entity());
3792        assert_eq!(next[0].patch, json!({"state": {"price": 2, "fills": [20]}}));
3793    }
3794
3795    #[test]
3796    fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
3797        let bytecode = MultiEntityBytecode::from_single(
3798            "PumpfunToken".to_string(),
3799            direct_pda_to_primary_key_spec(),
3800            0,
3801        );
3802        let mut vm = VmContext::new();
3803
3804        let mutations = vm
3805            .process_event(
3806                &bytecode,
3807                json!({
3808                    "__account_address": "bonding_curve_1",
3809                    "__resolved_primary_key": "mint_1",
3810                    "virtual_token_reserves": 42,
3811                }),
3812                "pump::BondingCurveState",
3813                None,
3814                None,
3815            )
3816            .unwrap();
3817
3818        assert_eq!(mutations.len(), 1);
3819        assert_eq!(mutations[0].key, json!("mint_1"));
3820        assert_eq!(
3821            mutations[0].patch["id"]["bonding_curve"],
3822            json!("bonding_curve_1")
3823        );
3824        assert_eq!(
3825            mutations[0].patch["reserves"]["virtual_token_reserves"],
3826            json!(42)
3827        );
3828    }
3829
3830    #[test]
3831    fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
3832        let bytecode = MultiEntityBytecode::from_single(
3833            "OreRound".to_string(),
3834            pda_to_intermediate_lookup_spec(),
3835            0,
3836        );
3837        let mut vm = VmContext::new();
3838
3839        let mutations = vm
3840            .process_event(
3841                &bytecode,
3842                json!({
3843                    "__account_address": "entropy_var_1",
3844                    "__resolved_primary_key": "round_address_1",
3845                    "end_at": 123,
3846                }),
3847                "entropy::VarState",
3848                None,
3849                None,
3850            )
3851            .unwrap();
3852
3853        assert!(mutations.is_empty());
3854    }
3855
3856    #[test]
3857    fn embedded_account_key_wins_over_an_unrelated_resolved_key() {
3858        let bytecode = MultiEntityBytecode::from_single(
3859            "OreTreasury".to_string(),
3860            embedded_account_key_spec(),
3861            0,
3862        );
3863        let mut vm = VmContext::new();
3864
3865        let mutations = vm
3866            .process_event(
3867                &bytecode,
3868                json!({
3869                    "__account_address": "treasury_pda",
3870                    "__resolved_primary_key": "round_address",
3871                }),
3872                "ore::TreasuryState",
3873                None,
3874                None,
3875            )
3876            .unwrap();
3877
3878        assert_eq!(mutations.len(), 1);
3879        assert_eq!(mutations[0].key, json!("treasury_pda"));
3880        assert_eq!(mutations[0].patch["id"]["address"], json!("treasury_pda"));
3881    }
3882
3883    fn instruction_handler(
3884        type_name: &str,
3885        key_resolution: KeyResolutionStrategy,
3886        mappings: Vec<SerializableFieldMapping>,
3887    ) -> SerializableHandlerSpec {
3888        SerializableHandlerSpec {
3889            source: SourceSpec::Source {
3890                program_id: None,
3891                discriminator: None,
3892                type_name: type_name.to_string(),
3893                serialization: None,
3894                is_account: false,
3895            },
3896            key_resolution,
3897            mappings,
3898            conditions: vec![],
3899            emit: true,
3900        }
3901    }
3902
3903    /// `split_position` updates two positions: the source (`first_position`)
3904    /// and the child (`second_position`). Mirrors the meteora-damm
3905    /// `MeteoraPosition` shape: one Count per side plus a derive_from hook on
3906    /// the source side.
3907    fn split_position_spec() -> TypedStreamSpec<Value> {
3908        let embedded = |field: &str| KeyResolutionStrategy::Embedded {
3909            primary_field: FieldPath::new(&["accounts", field]),
3910        };
3911        TypedStreamSpec::from_serializable(SerializableStreamSpec {
3912            ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
3913            state_name: "Position".to_string(),
3914            program_id: None,
3915            idl: None,
3916            identity: IdentitySpec {
3917                primary_keys: vec!["id.position_address".to_string()],
3918                lookup_indexes: vec![],
3919            },
3920            handlers: vec![
3921                instruction_handler(
3922                    "amm::SplitPositionIxState",
3923                    embedded("first_position"),
3924                    vec![
3925                        mapping(
3926                            "id.position_address",
3927                            &["accounts", "first_position"],
3928                            PopulationStrategy::SetOnce,
3929                        ),
3930                        mapping(
3931                            "activity.split_source_count",
3932                            &["data"],
3933                            PopulationStrategy::Count,
3934                        ),
3935                    ],
3936                ),
3937                instruction_handler(
3938                    "amm::SplitPositionIxState",
3939                    embedded("second_position"),
3940                    vec![
3941                        mapping(
3942                            "id.position_address",
3943                            &["accounts", "second_position"],
3944                            PopulationStrategy::SetOnce,
3945                        ),
3946                        mapping(
3947                            "activity.split_child_count",
3948                            &["data"],
3949                            PopulationStrategy::Count,
3950                        ),
3951                    ],
3952                ),
3953            ],
3954            sections: vec![],
3955            field_mappings: BTreeMap::new(),
3956            resolver_hooks: vec![],
3957            instruction_hooks: vec![InstructionHook {
3958                instruction_type: "amm::SplitPositionIxState".to_string(),
3959                actions: vec![HookAction::SetField {
3960                    target_field: "activity.last_split_source_slot".to_string(),
3961                    source: MappingSource::FromSource {
3962                        path: FieldPath::new(&["data", "slot"]),
3963                        default: None,
3964                        transform: None,
3965                    },
3966                    condition: None,
3967                }],
3968                lookup_by: Some(FieldPath::new(&["accounts", "first_position"])),
3969            }],
3970            resolver_specs: vec![],
3971            computed_fields: vec![],
3972            computed_field_specs: vec![],
3973            content_hash: None,
3974            views: vec![],
3975            scope_field: None,
3976        })
3977    }
3978
3979    fn split_position_event(first: &str, second: &str) -> Value {
3980        json!({
3981            "accounts": { "first_position": first, "second_position": second },
3982            "data": { "slot": 77 },
3983        })
3984    }
3985
3986    #[test]
3987    fn one_instruction_routes_each_key_to_its_own_entity() {
3988        let bytecode =
3989            MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
3990        let mut vm = VmContext::new();
3991
3992        for _ in 0..2 {
3993            vm.process_event(
3994                &bytecode,
3995                split_position_event("source_pos", "child_pos"),
3996                "amm::SplitPositionIxState",
3997                None,
3998                None,
3999            )
4000            .unwrap();
4001        }
4002        vm.process_event(
4003            &bytecode,
4004            split_position_event("child_pos", "grandchild_pos"),
4005            "amm::SplitPositionIxState",
4006            None,
4007            None,
4008        )
4009        .unwrap();
4010
4011        let state = |key: &str| vm.get_entity_state(0, &json!(key)).unwrap();
4012
4013        let source = state("source_pos");
4014        assert_eq!(source["id"]["position_address"], json!("source_pos"));
4015        assert_eq!(source["activity"]["split_source_count"], json!(2));
4016        assert_eq!(source["activity"].get("split_child_count"), None);
4017        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
4018
4019        // The child was split into once and split from once.
4020        let child = state("child_pos");
4021        assert_eq!(child["id"]["position_address"], json!("child_pos"));
4022        assert_eq!(child["activity"]["split_child_count"], json!(2));
4023        assert_eq!(child["activity"]["split_source_count"], json!(1));
4024        assert_eq!(child["activity"]["last_split_source_slot"], json!(77));
4025
4026        let grandchild = state("grandchild_pos");
4027        assert_eq!(grandchild["activity"]["split_child_count"], json!(1));
4028        assert_eq!(grandchild["activity"].get("split_source_count"), None);
4029        assert_eq!(grandchild["activity"].get("last_split_source_slot"), None);
4030    }
4031
4032    #[test]
4033    fn one_instruction_emits_one_mutation_per_key() {
4034        let bytecode =
4035            MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
4036        let mut vm = VmContext::new();
4037
4038        let mutations = vm
4039            .process_event(
4040                &bytecode,
4041                split_position_event("source_pos", "child_pos"),
4042                "amm::SplitPositionIxState",
4043                None,
4044                None,
4045            )
4046            .unwrap();
4047
4048        assert_eq!(mutations.len(), 2);
4049        assert_eq!(mutations[0].key, json!("source_pos"));
4050        assert_eq!(
4051            mutations[0].patch["activity"]["split_source_count"],
4052            json!(1)
4053        );
4054        assert_eq!(
4055            mutations[0].patch["activity"]["last_split_source_slot"],
4056            json!(77)
4057        );
4058        assert_eq!(
4059            mutations[0].patch["activity"].get("split_child_count"),
4060            None
4061        );
4062        assert_eq!(mutations[1].key, json!("child_pos"));
4063        assert_eq!(
4064            mutations[1].patch["activity"]["split_child_count"],
4065            json!(1)
4066        );
4067        assert_eq!(
4068            mutations[1].patch["activity"].get("split_source_count"),
4069            None
4070        );
4071    }
4072
4073    #[test]
4074    fn handlers_with_identical_keys_still_merge_into_one_segment() {
4075        let mut spec = split_position_spec();
4076        spec.instruction_hooks.clear();
4077        // Re-key the second handler by `first_position`: both now resolve the
4078        // same key and must share a single state read/write.
4079        spec.handlers[1].key_resolution = KeyResolutionStrategy::Embedded {
4080            primary_field: FieldPath::new(&["accounts", "first_position"]),
4081        };
4082        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
4083        let handler = &bytecode.entities["Position"].handlers["amm::SplitPositionIxState"];
4084
4085        assert!(!handler
4086            .iter()
4087            .any(|op| matches!(op, super::OpCode::SegmentBoundary)));
4088        assert_eq!(
4089            handler
4090                .iter()
4091                .filter(|op| matches!(op, super::OpCode::ReadOrInitState { .. }))
4092                .count(),
4093            1
4094        );
4095    }
4096
4097    #[test]
4098    fn hook_without_a_matching_key_gets_its_own_segment() {
4099        let mut spec = split_position_spec();
4100        spec.handlers.truncate(1);
4101        spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
4102            primary_field: FieldPath::new(&["accounts", "second_position"]),
4103        };
4104        spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
4105            "id.position_address",
4106            &["accounts", "second_position"],
4107            PopulationStrategy::SetOnce,
4108        ));
4109        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
4110        let mut vm = VmContext::new();
4111
4112        vm.process_event(
4113            &bytecode,
4114            split_position_event("source_pos", "child_pos"),
4115            "amm::SplitPositionIxState",
4116            None,
4117            None,
4118        )
4119        .unwrap();
4120
4121        // The hook routes by `first_position`, not by the handler's key.
4122        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
4123        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
4124        let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
4125        assert_eq!(child["activity"].get("last_split_source_slot"), None);
4126    }
4127
4128    /// Index positions by owner through a `PositionState` account handler.
4129    fn add_owner_index(spec: &mut TypedStreamSpec<Value>) {
4130        spec.identity.lookup_indexes.push(LookupIndexSpec {
4131            field_name: "id.owner".to_string(),
4132            temporal_field: None,
4133        });
4134        spec.handlers
4135            .push(crate::ast::TypedHandlerSpec::from_serializable(
4136                SerializableHandlerSpec {
4137                    source: SourceSpec::Source {
4138                        program_id: None,
4139                        discriminator: None,
4140                        type_name: "amm::PositionState".to_string(),
4141                        serialization: None,
4142                        is_account: true,
4143                    },
4144                    key_resolution: KeyResolutionStrategy::Embedded {
4145                        primary_field: FieldPath::new(&["__account_address"]),
4146                    },
4147                    mappings: vec![
4148                        mapping(
4149                            "id.position_address",
4150                            &["__account_address"],
4151                            PopulationStrategy::SetOnce,
4152                        ),
4153                        mapping("id.owner", &["owner"], PopulationStrategy::SetOnce),
4154                    ],
4155                    conditions: vec![],
4156                    emit: true,
4157                },
4158            ));
4159    }
4160
4161    fn index_owner(
4162        vm: &mut VmContext,
4163        bytecode: &MultiEntityBytecode,
4164        position: &str,
4165        owner: &str,
4166    ) {
4167        vm.process_event(
4168            bytecode,
4169            json!({ "__account_address": position, "owner": owner }),
4170            "amm::PositionState",
4171            None,
4172            None,
4173        )
4174        .unwrap();
4175    }
4176
4177    #[test]
4178    fn unmatched_hook_next_to_other_keys_resolves_lookup_index_fields() {
4179        let mut spec = split_position_spec();
4180        spec.handlers.truncate(1);
4181        spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
4182            primary_field: FieldPath::new(&["accounts", "second_position"]),
4183        };
4184        spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
4185            "id.position_address",
4186            &["accounts", "second_position"],
4187            PopulationStrategy::SetOnce,
4188        ));
4189        add_owner_index(&mut spec);
4190        // The hook routes by the owner, a lookup-index field.
4191        spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
4192
4193        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
4194        let mut vm = VmContext::new();
4195        index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
4196        vm.process_event(
4197            &bytecode,
4198            json!({
4199                "accounts": {
4200                    "first_position": "source_pos",
4201                    "second_position": "child_pos",
4202                    "owner": "owner_1",
4203                },
4204                "data": { "slot": 77 },
4205            }),
4206            "amm::SplitPositionIxState",
4207            None,
4208            None,
4209        )
4210        .unwrap();
4211
4212        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
4213        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
4214        assert!(vm.get_entity_state(0, &json!("owner_1")).is_none());
4215    }
4216
4217    #[test]
4218    fn unmatched_hook_replays_after_its_lookup_index_is_populated() {
4219        let mut spec = split_position_spec();
4220        spec.handlers.truncate(1);
4221        spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
4222            primary_field: FieldPath::new(&["accounts", "second_position"]),
4223        };
4224        spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
4225            "id.position_address",
4226            &["accounts", "second_position"],
4227            PopulationStrategy::SetOnce,
4228        ));
4229        add_owner_index(&mut spec);
4230        spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
4231
4232        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
4233        let mut vm = VmContext::new();
4234        let mutations = vm
4235            .process_event(
4236                &bytecode,
4237                json!({
4238                    "accounts": { "owner": "owner_1", "second_position": "child_pos" },
4239                    "data": { "slot": 77 },
4240                }),
4241                "amm::SplitPositionIxState",
4242                None,
4243                None,
4244            )
4245            .unwrap();
4246        assert_eq!(mutations.len(), 1);
4247        assert_eq!(mutations[0].key, json!("child_pos"));
4248
4249        index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
4250
4251        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
4252        assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
4253        let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
4254        assert_eq!(child["activity"].get("last_split_source_slot"), None);
4255    }
4256
4257    #[test]
4258    fn a_segment_that_misses_its_key_is_replayed_alone() {
4259        let mut spec = split_position_spec();
4260        spec.instruction_hooks.clear();
4261        // The child segment comes first. The source segment resolves its key
4262        // through a PDA, registered later, that maps to the position's owner.
4263        spec.handlers.swap(0, 1);
4264        spec.handlers[1].key_resolution = KeyResolutionStrategy::Lookup {
4265            primary_field: FieldPath::new(&["accounts", "source_pda"]),
4266            index_name: None,
4267        };
4268        spec.handlers[1].mappings.remove(0);
4269        add_owner_index(&mut spec);
4270        spec.instruction_hooks.push(InstructionHook {
4271            instruction_type: "amm::RegisterIxState".to_string(),
4272            actions: vec![HookAction::RegisterPdaMapping {
4273                pda_field: FieldPath::new(&["accounts", "source_pda"]),
4274                seed_field: FieldPath::new(&["accounts", "owner"]),
4275                lookup_name: "default_pda_lookup".to_string(),
4276                sibling: None,
4277                seed_expr: None,
4278                guard: None,
4279            }],
4280            lookup_by: None,
4281        });
4282        let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
4283        let mut vm = VmContext::new();
4284        index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
4285
4286        let mutations = vm
4287            .process_event(
4288                &bytecode,
4289                json!({
4290                    "accounts": { "source_pda": "pda_1", "second_position": "child_pos" },
4291                    "data": {},
4292                }),
4293                "amm::SplitPositionIxState",
4294                None,
4295                None,
4296            )
4297            .unwrap();
4298        // The child landed; the source is not resolvable yet.
4299        assert_eq!(mutations.len(), 1);
4300        assert_eq!(mutations[0].key, json!("child_pos"));
4301
4302        let mutations = vm
4303            .process_event(
4304                &bytecode,
4305                json!({ "accounts": { "source_pda": "pda_1", "owner": "owner_1" } }),
4306                "amm::RegisterIxState",
4307                None,
4308                None,
4309            )
4310            .unwrap();
4311        // Registering the PDA replays only the source segment.
4312        assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
4313        assert_eq!(mutations[0].key, json!("source_pos"));
4314        let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
4315        assert_eq!(source["activity"]["split_source_count"], json!(1));
4316        let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
4317        assert_eq!(child["activity"]["split_child_count"], json!(1));
4318    }
4319
4320    /// A hook whose `lookup_by` names a field the instruction does not carry
4321    /// loads a null key. The entity's key transform must pass that null to
4322    /// the segment's null-key check, which skips the hook, rather than fail
4323    /// the event and lose the handler segment's update with it.
4324    #[test]
4325    fn a_hook_whose_lookup_field_is_absent_skips_only_its_segment() {
4326        let spec: TypedStreamSpec<Value> =
4327            TypedStreamSpec::from_serializable(SerializableStreamSpec {
4328                ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
4329                state_name: "Game".to_string(),
4330                program_id: None,
4331                idl: None,
4332                identity: IdentitySpec {
4333                    primary_keys: vec!["id.game_id".to_string()],
4334                    lookup_indexes: vec![],
4335                },
4336                handlers: vec![instruction_handler(
4337                    "game::StartGameIxState",
4338                    KeyResolutionStrategy::Embedded {
4339                        primary_field: FieldPath::new(&["data", "game_id"]),
4340                    },
4341                    vec![
4342                        SerializableFieldMapping {
4343                            source: MappingSource::FromSource {
4344                                path: FieldPath::new(&["data", "game_id"]),
4345                                default: None,
4346                                transform: Some(crate::ast::Transformation::HexEncode),
4347                            },
4348                            ..mapping("id.game_id", &[], PopulationStrategy::SetOnce)
4349                        },
4350                        mapping("stats.fee", &["data", "fee"], PopulationStrategy::SetOnce),
4351                    ],
4352                )],
4353                sections: vec![],
4354                field_mappings: BTreeMap::new(),
4355                resolver_hooks: vec![],
4356                instruction_hooks: vec![InstructionHook {
4357                    instruction_type: "game::StartGameIxState".to_string(),
4358                    actions: vec![HookAction::SetField {
4359                        target_field: "stats.started_slot".to_string(),
4360                        source: MappingSource::FromSource {
4361                            path: FieldPath::new(&["data", "slot"]),
4362                            default: None,
4363                            transform: None,
4364                        },
4365                        condition: None,
4366                    }],
4367                    // `game_id` is an argument, not an account.
4368                    lookup_by: Some(FieldPath::new(&["accounts", "game_id"])),
4369                }],
4370                resolver_specs: vec![],
4371                computed_fields: vec![],
4372                computed_field_specs: vec![],
4373                content_hash: None,
4374                views: vec![],
4375                scope_field: None,
4376            });
4377        let bytecode = MultiEntityBytecode::from_single("Game".to_string(), spec, 0);
4378        let mut vm = VmContext::new();
4379
4380        let mutations = vm
4381            .process_event(
4382                &bytecode,
4383                json!({
4384                    "accounts": { "game_account": "game_pda" },
4385                    "data": { "game_id": [1, 2, 3, 4], "fee": 5, "slot": 9 },
4386                }),
4387                "game::StartGameIxState",
4388                None,
4389                None,
4390            )
4391            .unwrap();
4392
4393        assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
4394        assert_eq!(mutations[0].key, json!("01020304"));
4395        let game = vm.get_entity_state(0, &json!("01020304")).unwrap();
4396        assert_eq!(game["id"]["game_id"], json!("01020304"));
4397        assert_eq!(game["stats"]["fee"], json!(5));
4398        assert_eq!(game["stats"].get("started_slot"), None);
4399        // The hook's writes are not kept anywhere else either.
4400        assert_eq!(vm.get_entity_state(0, &json!(null)), None);
4401    }
4402
4403    mod fingerprint {
4404        use super::super::{EntityBytecode, MultiEntityBytecode, OpCode};
4405        use crate::ast::FieldPath;
4406        use serde_json::json;
4407        use std::collections::{HashMap, HashSet};
4408
4409        fn sample_handler() -> Vec<OpCode> {
4410            vec![
4411                OpCode::LoadEventField {
4412                    path: FieldPath::new(&["id", "mint"]),
4413                    dest: 0,
4414                    default: Some(json!({"b": 2, "a": 1})),
4415                },
4416                OpCode::CreateObject { dest: 1 },
4417                OpCode::SetField {
4418                    object: 1,
4419                    path: "state.volume".to_string(),
4420                    value: 0,
4421                },
4422                OpCode::EmitMutation {
4423                    entity_name: "PumpfunToken".to_string(),
4424                    key: 0,
4425                    state: 1,
4426                },
4427            ]
4428        }
4429
4430        fn bytecode(handler_insert_order: &[&str]) -> MultiEntityBytecode {
4431            let mut handlers = HashMap::new();
4432            for event_type in handler_insert_order {
4433                handlers.insert(event_type.to_string(), sample_handler());
4434            }
4435            let mut entities = HashMap::new();
4436            entities.insert(
4437                "PumpfunToken".to_string(),
4438                EntityBytecode {
4439                    state_id: 0,
4440                    handlers,
4441                    entity_name: "PumpfunToken".to_string(),
4442                    when_events: HashSet::from(["pump::BuyIxState".to_string()]),
4443                    non_emitted_fields: HashSet::new(),
4444                    computed_paths: vec!["state.market_cap".to_string()],
4445                    computed_fields_evaluator: None,
4446                },
4447            );
4448            let mut event_routing = HashMap::new();
4449            event_routing.insert(
4450                "pump::TokenState".to_string(),
4451                vec!["PumpfunToken".to_string()],
4452            );
4453            MultiEntityBytecode {
4454                entities,
4455                event_routing,
4456                when_events: HashSet::from(["pump::BuyIxState".to_string()]),
4457                proto_router: crate::proto_router::ProtoRouter::new(),
4458            }
4459        }
4460
4461        #[test]
4462        fn stable_across_recomputation_and_insertion_order() {
4463            let a = bytecode(&["pump::TokenState", "pump::BuyIxState"]);
4464            let b = bytecode(&["pump::BuyIxState", "pump::TokenState"]);
4465            assert_eq!(a.fingerprint(), a.fingerprint());
4466            assert_eq!(a.fingerprint(), b.fingerprint());
4467            assert_eq!(a.fingerprint().len(), 64);
4468        }
4469
4470        #[test]
4471        fn changes_when_an_opcode_operand_changes() {
4472            let base = bytecode(&["pump::TokenState"]);
4473            let mut modified = bytecode(&["pump::TokenState"]);
4474            let entity = modified.entities.get_mut("PumpfunToken").unwrap();
4475            match &mut entity.handlers.get_mut("pump::TokenState").unwrap()[2] {
4476                OpCode::SetField { path, .. } => *path = "state.volume_usd".to_string(),
4477                other => panic!("unexpected opcode: {other:?}"),
4478            }
4479            assert_ne!(base.fingerprint(), modified.fingerprint());
4480        }
4481
4482        #[test]
4483        fn changes_when_routing_changes() {
4484            let base = bytecode(&["pump::TokenState"]);
4485            let mut modified = bytecode(&["pump::TokenState"]);
4486            modified.event_routing.insert(
4487                "pump::SellIxState".to_string(),
4488                vec!["PumpfunToken".to_string()],
4489            );
4490            assert_ne!(base.fingerprint(), modified.fingerprint());
4491        }
4492
4493        /// Golden hash: this value changing means every deployed snapshot gets
4494        /// invalidated (bytecode_hash mismatch -> cold start). That can be a
4495        /// legitimate consequence of a semantic change to the sample opcodes or
4496        /// to the canonical encoding itself — but it must be a *decision*, not
4497        /// an accident. Update the constant only when snapshot invalidation is
4498        /// intended.
4499        #[test]
4500        fn golden_hash_pins_the_canonical_encoding() {
4501            let expected = "acf7ea8bf39ca830a9e6c2b2d2fb471da68e57e24bbb4c256964a42a498f75f2";
4502            assert_eq!(bytecode(&["pump::TokenState"]).fingerprint(), expected);
4503        }
4504    }
4505}