use crate::ast::*;
use serde::Serialize;
use serde_json::Value;
use std::collections::{HashMap, HashSet};
use tracing;
pub type Register = usize;
fn stop_field_path(target_path: &str) -> String {
format!("__stop:{}", target_path)
}
struct HandlerSegment {
key_identity: Option<Value>,
route_field: Option<FieldPath>,
ops: Vec<OpCode>,
}
fn route_field(resolution: &KeyResolutionStrategy) -> Option<FieldPath> {
let field = match resolution {
KeyResolutionStrategy::Embedded { primary_field }
| KeyResolutionStrategy::Lookup { primary_field }
| KeyResolutionStrategy::Computed { primary_field, .. } => primary_field,
KeyResolutionStrategy::TemporalLookup { lookup_field, .. } => lookup_field,
};
(!field.segments.is_empty()).then(|| field.clone())
}
fn merge_handler_opcodes(existing: &mut Vec<OpCode>, new: &[OpCode]) {
let mut existing_setup = Vec::new();
let mut existing_mappings = Vec::new();
let mut existing_teardown = Vec::new();
let mut section = 0;
for opcode in existing.iter() {
match opcode {
OpCode::ReadOrInitState { .. } => {
existing_setup.push(opcode.clone());
section = 1; }
OpCode::UpdateState { .. } => {
existing_teardown.push(opcode.clone());
section = 2; }
OpCode::EmitMutation { .. } => {
existing_teardown.push(opcode.clone());
}
_ if section == 0 => existing_setup.push(opcode.clone()),
_ if section == 1 => existing_mappings.push(opcode.clone()),
_ => existing_teardown.push(opcode.clone()),
}
}
let mut new_mappings = Vec::new();
section = 0;
for opcode in new.iter() {
match opcode {
OpCode::ReadOrInitState { .. } => {
section = 1; }
OpCode::UpdateState { .. } | OpCode::EmitMutation { .. } => {
section = 2; }
_ if section == 1 => {
new_mappings.push(opcode.clone());
}
_ => {} }
}
let mut merged = Vec::new();
merged.extend(existing_setup);
merged.extend(existing_mappings);
merged.extend(new_mappings);
merged.extend(existing_teardown);
*existing = merged;
}
#[derive(Debug, Clone, Serialize)]
pub enum OpCode {
AbortIfNullKey {
key: Register,
is_account_event: bool,
},
LoadEventField {
path: FieldPath,
dest: Register,
default: Option<Value>,
},
LoadConstant {
value: Value,
dest: Register,
},
CopyRegister {
source: Register,
dest: Register,
},
CopyRegisterIfNull {
source: Register,
dest: Register,
},
GetEventType {
dest: Register,
},
CreateObject {
dest: Register,
},
SetField {
object: Register,
path: String,
value: Register,
},
SetFields {
object: Register,
fields: Vec<(String, Register)>,
},
GetField {
object: Register,
path: String,
dest: Register,
},
ReadOrInitState {
state_id: u32,
key: Register,
default: Value,
dest: Register,
},
UpdateState {
state_id: u32,
key: Register,
value: Register,
},
AppendToArray {
object: Register,
path: String,
value: Register,
},
GetCurrentTimestamp {
dest: Register,
},
CreateEvent {
dest: Register,
event_value: Register,
},
CreateCapture {
dest: Register,
capture_value: Register,
},
Transform {
source: Register,
dest: Register,
transformation: Transformation,
},
EmitMutation {
entity_name: String,
key: Register,
state: Register,
},
SetFieldIfNull {
object: Register,
path: String,
value: Register,
},
SetFieldMax {
object: Register,
path: String,
value: Register,
},
UpdateTemporalIndex {
state_id: u32,
index_name: String,
lookup_value: Register,
primary_key: Register,
timestamp: Register,
},
LookupTemporalIndex {
state_id: u32,
index_name: String,
lookup_value: Register,
timestamp: Register,
dest: Register,
},
UpdateLookupIndex {
state_id: u32,
index_name: String,
lookup_value: Register,
primary_key: Register,
},
LookupIndex {
state_id: u32,
index_name: String,
lookup_value: Register,
dest: Register,
},
SetFieldSum {
object: Register,
path: String,
value: Register,
},
SetFieldIncrement {
object: Register,
path: String,
},
SetFieldMin {
object: Register,
path: String,
value: Register,
},
SetFieldWhen {
object: Register,
path: String,
value: Register,
when_instruction: String,
entity_name: String,
key_reg: Register,
condition_field: Option<FieldPath>,
condition_op: Option<ComparisonOp>,
condition_value: Option<Value>,
},
SetFieldUnlessStopped {
object: Register,
path: String,
value: Register,
stop_field: String,
stop_instruction: String,
entity_name: String,
key_reg: Register,
},
AddToUniqueSet {
state_id: u32,
set_name: String,
value: Register,
count_object: Register,
count_path: String,
},
ConditionalSetField {
object: Register,
path: String,
value: Register,
condition_field: FieldPath,
condition_op: ComparisonOp,
condition_value: Value,
},
ConditionalIncrement {
object: Register,
path: String,
condition_field: FieldPath,
condition_op: ComparisonOp,
condition_value: Value,
},
EvaluateComputedFields {
state: Register,
computed_paths: Vec<String>,
},
QueueResolver {
state_id: u32,
entity_name: String,
resolver: ResolverType,
input_path: Option<String>,
input_value: Option<Value>,
url_template: Option<Vec<UrlTemplatePart>>,
strategy: ResolveStrategy,
extracts: Vec<ResolverExtractSpec>,
condition: Option<ResolverCondition>,
schedule_at: Option<String>,
state: Register,
key: Register,
},
UpdatePdaReverseLookup {
state_id: u32,
lookup_name: String,
pda_address: Register,
primary_key: Register,
},
SegmentBoundary,
}
pub struct EntityBytecode {
pub state_id: u32,
pub handlers: HashMap<String, Vec<OpCode>>,
pub entity_name: String,
pub when_events: HashSet<String>,
pub non_emitted_fields: HashSet<String>,
pub computed_paths: Vec<String>,
#[allow(clippy::type_complexity)]
pub computed_fields_evaluator: Option<
Box<
dyn Fn(
&mut Value,
Option<u64>,
i64,
)
-> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
+ Send
+ Sync,
>,
>,
}
impl std::fmt::Debug for EntityBytecode {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("EntityBytecode")
.field("state_id", &self.state_id)
.field("handlers", &self.handlers)
.field("entity_name", &self.entity_name)
.field("when_events", &self.when_events)
.field("non_emitted_fields", &self.non_emitted_fields)
.field("computed_paths", &self.computed_paths)
.field(
"computed_fields_evaluator",
&self.computed_fields_evaluator.is_some(),
)
.finish()
}
}
#[derive(Debug)]
pub struct MultiEntityBytecode {
pub entities: HashMap<String, EntityBytecode>,
pub event_routing: HashMap<String, Vec<String>>,
pub when_events: HashSet<String>,
pub proto_router: crate::proto_router::ProtoRouter,
}
impl MultiEntityBytecode {
pub fn from_single<S>(entity_name: String, spec: TypedStreamSpec<S>, state_id: u32) -> Self {
let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
let entity_bytecode = compiler.compile_entity();
let mut entities = HashMap::new();
let mut event_routing = HashMap::new();
let mut when_events = HashSet::new();
for event_type in entity_bytecode.handlers.keys() {
event_routing
.entry(event_type.clone())
.or_insert_with(Vec::new)
.push(entity_name.clone());
}
when_events.extend(entity_bytecode.when_events.iter().cloned());
entities.insert(entity_name, entity_bytecode);
MultiEntityBytecode {
entities,
event_routing,
when_events,
proto_router: crate::proto_router::ProtoRouter::new(),
}
}
pub fn from_entities(entities_vec: Vec<(String, Box<dyn std::any::Any>, u32)>) -> Self {
let entities = HashMap::new();
let event_routing = HashMap::new();
let when_events = HashSet::new();
if let Some((_entity_name, _spec_any, _state_id)) = entities_vec.into_iter().next() {
panic!("from_entities requires type information - use builder pattern instead");
}
MultiEntityBytecode {
entities,
event_routing,
when_events,
proto_router: crate::proto_router::ProtoRouter::new(),
}
}
#[allow(clippy::new_ret_no_self)]
pub fn new() -> MultiEntityBytecodeBuilder {
MultiEntityBytecodeBuilder {
entities: HashMap::new(),
event_routing: HashMap::new(),
when_events: HashSet::new(),
proto_router: crate::proto_router::ProtoRouter::new(),
}
}
pub fn fingerprint(&self) -> String {
use sha2::{Digest, Sha256};
fn update_sorted_set(hasher: &mut Sha256, set: &HashSet<String>) {
let mut items: Vec<&String> = set.iter().collect();
items.sort();
for item in items {
hasher.update(item.as_bytes());
hasher.update([0u8]);
}
hasher.update([1u8]);
}
fn update_canonical_json(hasher: &mut Sha256, value: &Value) {
match value {
Value::Object(map) => {
let mut keys: Vec<&String> = map.keys().collect();
keys.sort();
hasher.update(*b"{");
for key in keys {
hasher.update(key.as_bytes());
hasher.update([0u8]);
update_canonical_json(hasher, &map[key]);
}
hasher.update(*b"}");
}
Value::Array(items) => {
hasher.update(*b"[");
for item in items {
update_canonical_json(hasher, item);
hasher.update([0u8]);
}
hasher.update(*b"]");
}
scalar => {
hasher.update(scalar.to_string().as_bytes());
hasher.update([0u8]);
}
}
}
let mut hasher = Sha256::new();
let mut entity_names: Vec<&String> = self.entities.keys().collect();
entity_names.sort();
for entity_name in entity_names {
let entity = &self.entities[entity_name];
hasher.update(entity_name.as_bytes());
hasher.update([0u8]);
hasher.update(entity.state_id.to_le_bytes());
let mut event_types: Vec<&String> = entity.handlers.keys().collect();
event_types.sort();
for event_type in event_types {
hasher.update(event_type.as_bytes());
hasher.update([0u8]);
let ops = serde_json::to_value(&entity.handlers[event_type]).expect(
"opcode stream must serialize for fingerprinting; \
a non-serializable OpCode payload breaks snapshot invalidation",
);
update_canonical_json(&mut hasher, &ops);
hasher.update([0u8]);
}
hasher.update([1u8]);
update_sorted_set(&mut hasher, &entity.when_events);
update_sorted_set(&mut hasher, &entity.non_emitted_fields);
for path in &entity.computed_paths {
hasher.update(path.as_bytes());
hasher.update([0u8]);
}
hasher.update([entity.computed_fields_evaluator.is_some() as u8]);
}
let mut routes: Vec<(&String, &Vec<String>)> = self.event_routing.iter().collect();
routes.sort_by_key(|(event_type, _)| *event_type);
for (event_type, entity_names) in routes {
hasher.update(event_type.as_bytes());
hasher.update([0u8]);
let mut sorted_names = entity_names.clone();
sorted_names.sort();
for name in sorted_names {
hasher.update(name.as_bytes());
hasher.update([0u8]);
}
hasher.update([1u8]);
}
update_sorted_set(&mut hasher, &self.when_events);
hex::encode(hasher.finalize())
}
}
pub struct MultiEntityBytecodeBuilder {
entities: HashMap<String, EntityBytecode>,
event_routing: HashMap<String, Vec<String>>,
when_events: HashSet<String>,
proto_router: crate::proto_router::ProtoRouter,
}
impl MultiEntityBytecodeBuilder {
pub fn add_entity<S>(
self,
entity_name: String,
spec: TypedStreamSpec<S>,
state_id: u32,
) -> Self {
self.add_entity_with_evaluator(
entity_name,
spec,
state_id,
None::<
fn(
&mut Value,
Option<u64>,
i64,
)
-> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>,
>,
)
}
pub fn add_entity_with_evaluator<S, F>(
mut self,
entity_name: String,
spec: TypedStreamSpec<S>,
state_id: u32,
evaluator: Option<F>,
) -> Self
where
F: Fn(
&mut Value,
Option<u64>,
i64,
) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>
+ Send
+ Sync
+ 'static,
{
let compiler = TypedCompiler::new(spec, entity_name.clone()).with_state_id(state_id);
let mut entity_bytecode = compiler.compile_entity();
if let Some(eval) = evaluator {
entity_bytecode.computed_fields_evaluator = Some(Box::new(eval));
}
for event_type in entity_bytecode.handlers.keys() {
self.event_routing
.entry(event_type.clone())
.or_default()
.push(entity_name.clone());
}
self.when_events
.extend(entity_bytecode.when_events.iter().cloned());
self.entities.insert(entity_name, entity_bytecode);
self
}
pub fn build(self) -> MultiEntityBytecode {
MultiEntityBytecode {
entities: self.entities,
event_routing: self.event_routing,
when_events: self.when_events,
proto_router: self.proto_router,
}
}
}
pub struct TypedCompiler<S> {
pub spec: TypedStreamSpec<S>,
entity_name: String,
state_id: u32,
}
impl<S> TypedCompiler<S> {
pub fn new(spec: TypedStreamSpec<S>, entity_name: String) -> Self {
TypedCompiler {
spec,
entity_name,
state_id: 0,
}
}
pub fn with_state_id(mut self, state_id: u32) -> Self {
self.state_id = state_id;
self
}
pub fn compile(&self) -> MultiEntityBytecode {
let entity_bytecode = self.compile_entity();
let mut entities = HashMap::new();
let mut event_routing = HashMap::new();
let mut when_events = HashSet::new();
for event_type in entity_bytecode.handlers.keys() {
event_routing
.entry(event_type.clone())
.or_insert_with(Vec::new)
.push(self.entity_name.clone());
}
when_events.extend(entity_bytecode.when_events.iter().cloned());
entities.insert(self.entity_name.clone(), entity_bytecode);
MultiEntityBytecode {
entities,
event_routing,
when_events,
proto_router: crate::proto_router::ProtoRouter::new(),
}
}
fn resolver_outputs_primary_key_directly(&self, primary_field: &FieldPath) -> bool {
if primary_field.segments.as_slice() != ["__account_address"] {
return false;
}
let primary_key_leafs: HashSet<&str> = self
.spec
.identity
.primary_keys
.iter()
.map(|path| path.rsplit('.').next().unwrap_or(path.as_str()))
.collect();
self.spec.instruction_hooks.iter().any(|hook| {
hook.actions.iter().any(|action| {
matches!(action, HookAction::RegisterPdaMapping { seed_field, .. }
if seed_field
.segments
.last()
.is_some_and(|segment| primary_key_leafs.contains(segment.as_str())))
})
})
}
fn compile_entity(&self) -> EntityBytecode {
let mut when_events: HashSet<String> = HashSet::new();
let mut emit_by_path: HashMap<String, bool> = HashMap::new();
let mut debug_info = Vec::new();
for (index, handler_spec) in self.spec.handlers.iter().enumerate() {
let event_type = self.get_event_type(&handler_spec.source);
let program_id = match &handler_spec.source {
crate::ast::SourceSpec::Source { program_id, .. } => {
program_id.as_ref().map(|s| s.as_str()).unwrap_or("null")
}
};
debug_info.push(format!(
" [{}] EventType={}, Mappings={}, ProgramId={}",
index,
event_type,
handler_spec.mappings.len(),
program_id
));
}
let mut segments: HashMap<String, Vec<HandlerSegment>> = HashMap::new();
for handler_spec in &self.spec.handlers {
for mapping in &handler_spec.mappings {
if let Some(when) = &mapping.when {
when_events.insert(when.clone());
}
let entry = emit_by_path
.entry(mapping.target_path.clone())
.or_insert(false);
*entry |= mapping.emit;
if mapping.stop.is_some() {
emit_by_path
.entry(stop_field_path(&mapping.target_path))
.or_insert(false);
}
}
let opcodes = self.compile_handler(handler_spec);
let event_type = self.get_event_type(&handler_spec.source);
let key_identity = self.key_identity(handler_spec);
let event_segments = segments.entry(event_type).or_default();
if let Some(existing) = event_segments
.iter_mut()
.find(|segment| segment.key_identity.as_ref() == Some(&key_identity))
{
merge_handler_opcodes(&mut existing.ops, &opcodes);
} else {
event_segments.push(HandlerSegment {
key_identity: Some(key_identity),
route_field: route_field(&handler_spec.key_resolution),
ops: opcodes,
});
}
}
for hook in &self.spec.instruction_hooks {
let event_type = hook.instruction_type.clone();
let event_segments = segments.entry(event_type).or_default();
let segment_index = match &hook.lookup_by {
None if !event_segments.is_empty() => Some(0),
None => None,
Some(lookup_by) => event_segments
.iter()
.position(|segment| segment.route_field.as_ref() == Some(lookup_by)),
};
let segment_index = match segment_index {
Some(index) => index,
None => {
let resolution = if event_segments.is_empty() {
None
} else {
hook.lookup_by
.as_ref()
.and_then(|lookup_by| self.hook_key_resolution(lookup_by))
};
let ops = match resolution {
Some(resolution) => self.compile_keyed_hook_segment(&resolution),
None => self.compile_hook_segment(hook),
};
event_segments.push(HandlerSegment {
key_identity: None,
route_field: hook.lookup_by.clone(),
ops,
});
event_segments.len() - 1
}
};
let handler_opcodes = &mut event_segments[segment_index].ops;
let hook_opcodes = self.compile_instruction_hook_actions(&hook.actions);
let insert_pos = handler_opcodes
.iter()
.position(|op| matches!(op, OpCode::EvaluateComputedFields { .. }))
.or_else(|| {
handler_opcodes
.iter()
.position(|op| matches!(op, OpCode::UpdateState { .. }))
});
if let Some(pos) = insert_pos {
for (i, opcode) in hook_opcodes.into_iter().enumerate() {
handler_opcodes.insert(pos + i, opcode);
}
}
}
let handlers: HashMap<String, Vec<OpCode>> = segments
.into_iter()
.map(|(event_type, event_segments)| {
let mut ops = Vec::new();
for (index, segment) in event_segments.into_iter().enumerate() {
if index > 0 {
ops.push(OpCode::SegmentBoundary);
}
ops.extend(segment.ops);
}
(event_type, ops)
})
.collect();
let non_emitted_fields: HashSet<String> = emit_by_path
.into_iter()
.filter_map(|(path, emit)| if emit { None } else { Some(path) })
.collect();
EntityBytecode {
state_id: self.state_id,
handlers,
entity_name: self.entity_name.clone(),
when_events,
non_emitted_fields,
computed_paths: self.spec.computed_fields.clone(),
computed_fields_evaluator: None,
}
}
fn key_identity(&self, spec: &TypedHandlerSpec<S>) -> Value {
let key_reg = 20;
let ops = self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings);
serde_json::to_value(&ops).unwrap_or(Value::Null)
}
fn hook_key_resolution(&self, lookup_by: &FieldPath) -> Option<KeyResolutionStrategy> {
let leaf = lookup_by.segments.last()?;
let is_primary_key_field = self
.spec
.identity
.primary_keys
.iter()
.any(|pk| pk.rsplit('.').next() == Some(leaf.as_str()));
if is_primary_key_field {
Some(KeyResolutionStrategy::Embedded {
primary_field: lookup_by.clone(),
})
} else if self.find_lookup_index_for_field(lookup_by).is_some() {
Some(KeyResolutionStrategy::Lookup {
primary_field: lookup_by.clone(),
})
} else {
None
}
}
fn compile_keyed_hook_segment(&self, resolution: &KeyResolutionStrategy) -> Vec<OpCode> {
let key_reg = 20;
let state_reg = 2;
let mut ops = self.compile_key_loading(resolution, key_reg, &[]);
ops.push(OpCode::AbortIfNullKey {
key: key_reg,
is_account_event: false,
});
ops.push(OpCode::ReadOrInitState {
state_id: self.state_id,
key: key_reg,
default: serde_json::json!({}),
dest: state_reg,
});
ops.push(OpCode::UpdateState {
state_id: self.state_id,
key: key_reg,
value: state_reg,
});
ops
}
fn compile_hook_segment(&self, hook: &InstructionHook) -> Vec<OpCode> {
let key_reg = 20;
let state_reg = 2;
let resolved_key_reg = 19;
let temp_reg = 18;
let mut ops = Vec::new();
ops.push(OpCode::LoadEventField {
path: FieldPath::new(&["__resolved_primary_key"]),
dest: resolved_key_reg,
default: Some(serde_json::json!(null)),
});
ops.push(OpCode::CopyRegister {
source: resolved_key_reg,
dest: key_reg,
});
if let Some(lookup_path) = &hook.lookup_by {
ops.push(OpCode::LoadEventField {
path: lookup_path.clone(),
dest: temp_reg,
default: None,
});
ops.push(OpCode::Transform {
source: temp_reg,
dest: temp_reg,
transformation: Transformation::HexEncode,
});
ops.push(OpCode::CopyRegisterIfNull {
source: temp_reg,
dest: key_reg,
});
}
ops.push(OpCode::ReadOrInitState {
state_id: self.state_id,
key: key_reg,
default: serde_json::json!({}),
dest: state_reg,
});
ops.push(OpCode::UpdateState {
state_id: self.state_id,
key: key_reg,
value: state_reg,
});
ops
}
fn compile_handler(&self, spec: &TypedHandlerSpec<S>) -> Vec<OpCode> {
let mut ops = Vec::new();
let state_reg = 2;
let key_reg = 20;
ops.extend(self.compile_key_loading(&spec.key_resolution, key_reg, &spec.mappings));
let is_account_event = matches!(
spec.source,
SourceSpec::Source {
is_account: true,
..
}
);
ops.push(OpCode::AbortIfNullKey {
key: key_reg,
is_account_event,
});
ops.push(OpCode::ReadOrInitState {
state_id: self.state_id,
key: key_reg,
default: serde_json::json!({}),
dest: state_reg,
});
ops.extend(self.compile_temporal_index_update(
&spec.key_resolution,
key_reg,
&spec.mappings,
));
for mapping in &spec.mappings {
ops.extend(self.compile_mapping(mapping, state_reg, key_reg));
}
ops.extend(self.compile_resolvers(state_reg, key_reg));
ops.push(OpCode::EvaluateComputedFields {
state: state_reg,
computed_paths: self.spec.computed_fields.clone(),
});
ops.push(OpCode::UpdateState {
state_id: self.state_id,
key: key_reg,
value: state_reg,
});
if spec.emit {
ops.push(OpCode::EmitMutation {
entity_name: self.entity_name.clone(),
key: key_reg,
state: state_reg,
});
}
ops
}
fn compile_resolvers(&self, state_reg: Register, key_reg: Register) -> Vec<OpCode> {
let mut ops = Vec::new();
for resolver_spec in &self.spec.resolver_specs {
let url_template = match &resolver_spec.resolver {
ResolverType::Url(config) => match &config.url_source {
UrlSource::Template(parts) => Some(parts.clone()),
_ => None,
},
_ => None,
};
ops.push(OpCode::QueueResolver {
state_id: self.state_id,
entity_name: self.entity_name.clone(),
resolver: resolver_spec.resolver.clone(),
input_path: resolver_spec.input_path.clone(),
input_value: resolver_spec.input_value.clone(),
url_template,
strategy: resolver_spec.strategy.clone(),
extracts: resolver_spec.extracts.clone(),
condition: resolver_spec.condition.clone(),
schedule_at: resolver_spec.schedule_at.clone(),
state: state_reg,
key: key_reg,
});
}
ops
}
fn compile_mapping(
&self,
mapping: &TypedFieldMapping<S>,
state_reg: Register,
key_reg: Register,
) -> Vec<OpCode> {
let mut ops = Vec::new();
let temp_reg = 10;
ops.extend(self.compile_mapping_source(&mapping.source, temp_reg));
if let Some(transform) = &mapping.transform {
ops.push(OpCode::Transform {
source: temp_reg,
dest: temp_reg,
transformation: transform.clone(),
});
}
if let Some(stop_instruction) = &mapping.stop {
if mapping.when.is_some() {
tracing::warn!(
"#[map] stop and when both set for {}. Ignoring when.",
mapping.target_path
);
}
if !matches!(mapping.population, PopulationStrategy::LastWrite)
&& !matches!(mapping.population, PopulationStrategy::Merge)
{
tracing::warn!(
"#[map] stop ignores population strategy {:?}",
mapping.population
);
}
ops.push(OpCode::SetFieldUnlessStopped {
object: state_reg,
path: mapping.target_path.clone(),
value: temp_reg,
stop_field: stop_field_path(&mapping.target_path),
stop_instruction: stop_instruction.clone(),
entity_name: self.entity_name.clone(),
key_reg,
});
return ops;
}
if let Some(when_instruction) = &mapping.when {
if !matches!(mapping.population, PopulationStrategy::LastWrite)
&& !matches!(mapping.population, PopulationStrategy::Merge)
{
tracing::warn!(
"#[map] when ignores population strategy {:?}",
mapping.population
);
}
let (condition_field, condition_op, condition_value) = mapping
.condition
.as_ref()
.and_then(|cond| cond.parsed.as_ref())
.and_then(|parsed| match parsed {
ParsedCondition::Comparison { field, op, value } => {
Some((Some(field.clone()), Some(op.clone()), Some(value.clone())))
}
ParsedCondition::Logical { .. } => {
tracing::warn!("Logical conditions not yet supported for #[map] when");
None
}
})
.unwrap_or((None, None, None));
ops.push(OpCode::SetFieldWhen {
object: state_reg,
path: mapping.target_path.clone(),
value: temp_reg,
when_instruction: when_instruction.clone(),
entity_name: self.entity_name.clone(),
key_reg,
condition_field,
condition_op,
condition_value,
});
return ops;
}
if let Some(condition) = &mapping.condition {
if let Some(parsed) = &condition.parsed {
match parsed {
ParsedCondition::Comparison {
field,
op,
value: cond_value,
} => {
if matches!(mapping.population, PopulationStrategy::LastWrite)
|| matches!(mapping.population, PopulationStrategy::Merge)
{
ops.push(OpCode::ConditionalSetField {
object: state_reg,
path: mapping.target_path.clone(),
value: temp_reg,
condition_field: field.clone(),
condition_op: op.clone(),
condition_value: cond_value.clone(),
});
return ops;
}
if matches!(mapping.population, PopulationStrategy::Count) {
ops.push(OpCode::ConditionalIncrement {
object: state_reg,
path: mapping.target_path.clone(),
condition_field: field.clone(),
condition_op: op.clone(),
condition_value: cond_value.clone(),
});
return ops;
}
tracing::warn!(
"Conditional #[map] not supported for population strategy {:?}",
mapping.population
);
}
ParsedCondition::Logical { .. } => {
tracing::warn!("Logical conditions not yet supported for #[map]");
}
}
}
}
match &mapping.population {
PopulationStrategy::Append => {
ops.push(OpCode::AppendToArray {
object: state_reg,
path: mapping.target_path.clone(),
value: temp_reg,
});
}
PopulationStrategy::LastWrite => {
ops.push(OpCode::SetField {
object: state_reg,
path: mapping.target_path.clone(),
value: temp_reg,
});
}
PopulationStrategy::SetOnce => {
ops.push(OpCode::SetFieldIfNull {
object: state_reg,
path: mapping.target_path.clone(),
value: temp_reg,
});
}
PopulationStrategy::Merge => {
ops.push(OpCode::SetField {
object: state_reg,
path: mapping.target_path.clone(),
value: temp_reg,
});
}
PopulationStrategy::Max => {
ops.push(OpCode::SetFieldMax {
object: state_reg,
path: mapping.target_path.clone(),
value: temp_reg,
});
}
PopulationStrategy::Sum => {
ops.push(OpCode::SetFieldSum {
object: state_reg,
path: mapping.target_path.clone(),
value: temp_reg,
});
}
PopulationStrategy::Count => {
ops.push(OpCode::SetFieldIncrement {
object: state_reg,
path: mapping.target_path.clone(),
});
}
PopulationStrategy::Min => {
ops.push(OpCode::SetFieldMin {
object: state_reg,
path: mapping.target_path.clone(),
value: temp_reg,
});
}
PopulationStrategy::UniqueCount => {
let set_name = format!("{}_unique_set", mapping.target_path);
ops.push(OpCode::AddToUniqueSet {
state_id: self.state_id,
set_name,
value: temp_reg,
count_object: state_reg,
count_path: mapping.target_path.clone(),
});
}
}
ops
}
fn compile_mapping_source(&self, source: &MappingSource, dest: Register) -> Vec<OpCode> {
match source {
MappingSource::FromSource {
path,
default,
transform,
} => {
let mut ops = vec![OpCode::LoadEventField {
path: path.clone(),
dest,
default: default.clone(),
}];
if let Some(transform_type) = transform {
ops.push(OpCode::Transform {
source: dest,
dest,
transformation: transform_type.clone(),
});
}
ops
}
MappingSource::Constant(val) => {
vec![OpCode::LoadConstant {
value: val.clone(),
dest,
}]
}
MappingSource::AsEvent { fields } => {
let mut ops = Vec::new();
if fields.is_empty() {
let event_data_reg = dest + 1;
ops.push(OpCode::LoadEventField {
path: FieldPath::new(&[]),
dest: event_data_reg,
default: Some(serde_json::json!({})),
});
ops.push(OpCode::CreateEvent {
dest,
event_value: event_data_reg,
});
} else {
let data_obj_reg = dest + 1;
ops.push(OpCode::CreateObject { dest: data_obj_reg });
let mut field_registers = Vec::new();
let mut current_reg = dest + 2;
for field_source in fields.iter() {
if let MappingSource::FromSource {
path,
default,
transform,
} = &**field_source
{
ops.push(OpCode::LoadEventField {
path: path.clone(),
dest: current_reg,
default: default.clone(),
});
if let Some(transform_type) = transform {
ops.push(OpCode::Transform {
source: current_reg,
dest: current_reg,
transformation: transform_type.clone(),
});
}
if let Some(field_name) = path.segments.last() {
field_registers.push((field_name.clone(), current_reg));
}
current_reg += 1;
}
}
if !field_registers.is_empty() {
ops.push(OpCode::SetFields {
object: data_obj_reg,
fields: field_registers,
});
}
ops.push(OpCode::CreateEvent {
dest,
event_value: data_obj_reg,
});
}
ops
}
MappingSource::WholeSource => {
vec![OpCode::LoadEventField {
path: FieldPath::new(&[]),
dest,
default: Some(serde_json::json!({})),
}]
}
MappingSource::AsCapture { field_transforms } => {
let capture_data_reg = 22; let mut ops = vec![OpCode::LoadEventField {
path: FieldPath::new(&[]),
dest: capture_data_reg,
default: Some(serde_json::json!({})),
}];
let field_reg = 24;
let transformed_reg = 25;
for (field_name, transform) in field_transforms {
ops.push(OpCode::GetField {
object: capture_data_reg,
path: field_name.clone(),
dest: field_reg,
});
ops.push(OpCode::Transform {
source: field_reg,
dest: transformed_reg,
transformation: transform.clone(),
});
ops.push(OpCode::SetField {
object: capture_data_reg,
path: field_name.clone(),
value: transformed_reg,
});
}
ops.push(OpCode::CreateCapture {
dest,
capture_value: capture_data_reg,
});
ops
}
MappingSource::FromContext { field } => {
vec![OpCode::LoadEventField {
path: FieldPath::new(&["__update_context", field.as_str()]),
dest,
default: Some(serde_json::json!(null)),
}]
}
MappingSource::Computed { .. } => {
vec![]
}
MappingSource::FromState { .. } => {
vec![]
}
}
}
pub fn compile_key_loading(
&self,
resolution: &KeyResolutionStrategy,
key_reg: Register,
mappings: &[TypedFieldMapping<S>],
) -> Vec<OpCode> {
let mut ops = Vec::new();
let resolved_key_reg = 19; ops.push(OpCode::LoadEventField {
path: FieldPath::new(&["__resolved_primary_key"]),
dest: resolved_key_reg,
default: Some(serde_json::json!(null)),
});
match resolution {
KeyResolutionStrategy::Embedded { primary_field } => {
let effective_primary_field = if primary_field.segments.is_empty() {
if let Some(auto_field) = self.auto_detect_primary_field(mappings) {
auto_field
} else {
primary_field.clone()
}
} else {
primary_field.clone()
};
if !effective_primary_field.segments.is_empty() {
let temp_reg = 18;
let transform_reg = 23;
ops.push(OpCode::LoadEventField {
path: effective_primary_field.clone(),
dest: temp_reg,
default: None,
});
let primary_key_transform = self
.find_primary_key_transformation(mappings)
.or_else(|| self.find_inherited_primary_key_transformation());
if let Some(transform) = primary_key_transform {
ops.push(OpCode::Transform {
source: temp_reg,
dest: transform_reg,
transformation: transform,
});
ops.push(OpCode::CopyRegister {
source: transform_reg,
dest: key_reg,
});
} else {
ops.push(OpCode::CopyRegister {
source: temp_reg,
dest: key_reg,
});
}
ops.push(OpCode::CopyRegisterIfNull {
source: resolved_key_reg,
dest: key_reg,
});
} else {
ops.push(OpCode::CopyRegister {
source: resolved_key_reg,
dest: key_reg,
});
}
}
KeyResolutionStrategy::Lookup { primary_field } => {
let lookup_reg = 15;
let result_reg = 17;
ops.push(OpCode::CopyRegister {
source: resolved_key_reg,
dest: lookup_reg,
});
let temp_reg = 18;
ops.push(OpCode::LoadEventField {
path: primary_field.clone(),
dest: temp_reg,
default: None,
});
ops.push(OpCode::CopyRegisterIfNull {
source: temp_reg,
dest: lookup_reg,
});
let index_name = self.find_lookup_index_for_lookup_field(primary_field, mappings);
let effective_index_name =
index_name.unwrap_or_else(|| "default_pda_lookup".to_string());
ops.push(OpCode::LookupIndex {
state_id: self.state_id,
index_name: effective_index_name,
lookup_value: lookup_reg,
dest: result_reg,
});
ops.push(OpCode::CopyRegister {
source: result_reg,
dest: key_reg,
});
if self.resolver_outputs_primary_key_directly(primary_field) {
ops.push(OpCode::CopyRegisterIfNull {
source: resolved_key_reg,
dest: key_reg,
});
}
}
KeyResolutionStrategy::Computed {
primary_field,
compute_partition: _,
} => {
ops.push(OpCode::CopyRegister {
source: resolved_key_reg,
dest: key_reg,
});
let temp_reg = 18;
ops.push(OpCode::LoadEventField {
path: primary_field.clone(),
dest: temp_reg,
default: None,
});
ops.push(OpCode::CopyRegisterIfNull {
source: temp_reg,
dest: key_reg,
});
}
KeyResolutionStrategy::TemporalLookup {
lookup_field,
timestamp_field,
index_name,
} => {
ops.push(OpCode::CopyRegister {
source: resolved_key_reg,
dest: key_reg,
});
let lookup_reg = 15;
let timestamp_reg = 16;
let result_reg = 17;
ops.push(OpCode::LoadEventField {
path: lookup_field.clone(),
dest: lookup_reg,
default: None,
});
ops.push(OpCode::LoadEventField {
path: timestamp_field.clone(),
dest: timestamp_reg,
default: None,
});
ops.push(OpCode::LookupTemporalIndex {
state_id: self.state_id,
index_name: index_name.clone(),
lookup_value: lookup_reg,
timestamp: timestamp_reg,
dest: result_reg,
});
ops.push(OpCode::CopyRegisterIfNull {
source: result_reg,
dest: key_reg,
});
}
}
ops
}
fn find_primary_key_transformation(
&self,
mappings: &[TypedFieldMapping<S>],
) -> Option<Transformation> {
let primary_key = self.spec.identity.primary_keys.first()?;
let primary_field_name = self.extract_primary_field_name(primary_key)?;
for mapping in mappings {
if mapping.target_path == *primary_key
|| mapping.target_path.ends_with(&format!(".{}", primary_key))
{
if let Some(transform) = &mapping.transform {
return Some(transform.clone());
}
if let MappingSource::FromSource {
transform: Some(transform),
..
} = &mapping.source
{
return Some(transform.clone());
}
}
}
for mapping in mappings {
if let MappingSource::AsCapture { field_transforms } = &mapping.source {
if let Some(transform) = field_transforms.get(&primary_field_name) {
return Some(transform.clone());
}
}
}
None
}
pub fn find_inherited_primary_key_transformation(&self) -> Option<Transformation> {
let primary_key = self.spec.identity.primary_keys.first()?;
let primary_field_name = self.extract_primary_field_name(primary_key)?;
for handler in &self.spec.handlers {
for mapping in &handler.mappings {
if mapping.target_path == *primary_key
|| mapping.target_path.ends_with(&format!(".{}", primary_key))
{
if let MappingSource::FromSource {
path, transform, ..
} = &mapping.source
{
if path.segments.last() == Some(&primary_field_name) {
return mapping.transform.clone().or_else(|| transform.clone());
}
}
}
if let MappingSource::AsCapture { field_transforms } = &mapping.source {
if let Some(transform) = field_transforms.get(&primary_field_name) {
return Some(transform.clone());
}
}
}
}
None
}
fn extract_primary_field_name(&self, primary_key: &str) -> Option<String> {
primary_key.split('.').next_back().map(|s| s.to_string())
}
pub fn auto_detect_primary_field(
&self,
current_mappings: &[TypedFieldMapping<S>],
) -> Option<FieldPath> {
let primary_key = self.spec.identity.primary_keys.first()?;
let primary_field_name = self.extract_primary_field_name(primary_key)?;
if self.current_account_has_primary_field(&primary_field_name, current_mappings) {
return Some(FieldPath::new(&[&primary_field_name]));
}
None
}
fn current_account_has_primary_field(
&self,
field_name: &str,
mappings: &[TypedFieldMapping<S>],
) -> bool {
for mapping in mappings {
if let MappingSource::FromSource { path, .. } = &mapping.source {
if path.segments.last() == Some(&field_name.to_string()) {
return true;
}
}
}
false
}
#[allow(dead_code)]
fn handler_has_field(&self, field_name: &str, mappings: &[TypedFieldMapping<S>]) -> bool {
for mapping in mappings {
if let MappingSource::FromSource { path, .. } = &mapping.source {
if path.segments.last() == Some(&field_name.to_string()) {
return true;
}
}
}
false
}
#[allow(dead_code)]
fn field_exists_in_mappings(
&self,
field_name: &str,
mappings: &[TypedFieldMapping<S>],
) -> bool {
for mapping in mappings {
if let MappingSource::FromSource { path, .. } = &mapping.source {
if path.segments.last() == Some(&field_name.to_string()) {
return true;
}
}
if let MappingSource::AsCapture { field_transforms } = &mapping.source {
if field_transforms.contains_key(field_name) {
return true;
}
}
}
false
}
fn find_lookup_index_for_field(&self, field_path: &FieldPath) -> Option<String> {
if field_path.segments.is_empty() {
return None;
}
let lookup_field_name = field_path.segments.last().unwrap();
for lookup_index in &self.spec.identity.lookup_indexes {
let index_field_name = lookup_index
.field_name
.split('.')
.next_back()
.unwrap_or(&lookup_index.field_name);
let matches_directly = index_field_name == lookup_field_name;
let matches_address_alias = index_field_name
.strip_suffix("_address")
.map(|base| base == lookup_field_name)
.unwrap_or(false);
if matches_directly || matches_address_alias {
return Some(format!("{}_lookup_index", index_field_name));
}
}
None
}
fn find_lookup_index_for_lookup_field(
&self,
primary_field: &FieldPath,
mappings: &[TypedFieldMapping<S>],
) -> Option<String> {
let primary_path = primary_field.segments.join(".");
for mapping in mappings {
if let MappingSource::FromSource { path, .. } = &mapping.source {
let source_path = path.segments.join(".");
if source_path == primary_path {
for lookup_index in &self.spec.identity.lookup_indexes {
if mapping.target_path == lookup_index.field_name {
let index_field_name = lookup_index
.field_name
.split('.')
.next_back()
.unwrap_or(&lookup_index.field_name);
return Some(format!("{}_lookup_index", index_field_name));
}
}
}
}
}
self.find_lookup_index_for_field(primary_field)
}
fn find_source_path_for_lookup_index(
&self,
mappings: &[TypedFieldMapping<S>],
lookup_field_name: &str,
) -> Option<Vec<String>> {
for mapping in mappings {
if mapping.target_path == lookup_field_name {
if let MappingSource::FromSource { path, .. } = &mapping.source {
return Some(path.segments.clone());
}
}
}
None
}
fn compile_temporal_index_update(
&self,
resolution: &KeyResolutionStrategy,
key_reg: Register,
mappings: &[TypedFieldMapping<S>],
) -> Vec<OpCode> {
let mut ops = Vec::new();
for lookup_index in &self.spec.identity.lookup_indexes {
let lookup_reg = 17;
let source_field = lookup_index
.field_name
.split('.')
.next_back()
.unwrap_or(&lookup_index.field_name);
match resolution {
KeyResolutionStrategy::Embedded { primary_field: _ } => {
let source_path_opt =
self.find_source_path_for_lookup_index(mappings, &lookup_index.field_name);
let load_path = if let Some(ref path) = source_path_opt {
FieldPath::new(&path.iter().map(|s| s.as_str()).collect::<Vec<_>>())
} else {
FieldPath::new(&[source_field])
};
ops.push(OpCode::LoadEventField {
path: load_path,
dest: lookup_reg,
default: None,
});
if let Some(temporal_field_name) = &lookup_index.temporal_field {
let timestamp_reg = 18;
ops.push(OpCode::LoadEventField {
path: FieldPath::new(&[temporal_field_name]),
dest: timestamp_reg,
default: None,
});
let index_name = format!("{}_temporal_index", source_field);
ops.push(OpCode::UpdateTemporalIndex {
state_id: self.state_id,
index_name,
lookup_value: lookup_reg,
primary_key: key_reg,
timestamp: timestamp_reg,
});
let simple_index_name = format!("{}_lookup_index", source_field);
ops.push(OpCode::UpdateLookupIndex {
state_id: self.state_id,
index_name: simple_index_name,
lookup_value: lookup_reg,
primary_key: key_reg,
});
} else {
let index_name = format!("{}_lookup_index", source_field);
ops.push(OpCode::UpdateLookupIndex {
state_id: self.state_id,
index_name,
lookup_value: lookup_reg,
primary_key: key_reg,
});
}
if source_path_opt.is_some() {
ops.push(OpCode::UpdatePdaReverseLookup {
state_id: self.state_id,
lookup_name: "default_pda_lookup".to_string(),
pda_address: lookup_reg,
primary_key: key_reg,
});
}
}
KeyResolutionStrategy::Lookup { primary_field } => {
let has_mapping_to_lookup_field = mappings
.iter()
.any(|m| m.target_path == lookup_index.field_name);
if has_mapping_to_lookup_field {
let path_segments: Vec<&str> =
primary_field.segments.iter().map(|s| s.as_str()).collect();
ops.push(OpCode::LoadEventField {
path: FieldPath::new(&path_segments),
dest: lookup_reg,
default: None,
});
let index_name = format!("{}_lookup_index", source_field);
ops.push(OpCode::UpdateLookupIndex {
state_id: self.state_id,
index_name,
lookup_value: lookup_reg,
primary_key: key_reg,
});
}
}
KeyResolutionStrategy::Computed { .. }
| KeyResolutionStrategy::TemporalLookup { .. } => {
}
}
}
ops
}
fn get_event_type(&self, source: &SourceSpec) -> String {
match source {
SourceSpec::Source { type_name, .. } => type_name.clone(),
}
}
fn compile_instruction_hook_actions(&self, actions: &[HookAction]) -> Vec<OpCode> {
let mut ops = Vec::new();
let state_reg = 2;
for action in actions {
match action {
HookAction::SetField {
target_field,
source,
condition,
} => {
let _ = condition;
let temp_reg = 11;
let load_ops = self.compile_mapping_source(source, temp_reg);
ops.extend(load_ops);
if let MappingSource::FromSource {
transform: Some(transform_type),
..
} = source
{
ops.push(OpCode::Transform {
source: temp_reg,
dest: temp_reg,
transformation: transform_type.clone(),
});
}
if let Some(cond_expr) = condition {
if let Some(parsed) = &cond_expr.parsed {
let cond_check_ops = self.compile_condition_check(
parsed,
temp_reg,
state_reg,
target_field,
);
ops.extend(cond_check_ops);
} else {
ops.push(OpCode::SetField {
object: state_reg,
path: target_field.clone(),
value: temp_reg,
});
}
} else {
ops.push(OpCode::SetField {
object: state_reg,
path: target_field.clone(),
value: temp_reg,
});
}
}
HookAction::IncrementField {
target_field,
increment_by,
condition,
} => {
if let Some(cond_expr) = condition {
if let Some(parsed) = &cond_expr.parsed {
let cond_check_ops = self.compile_conditional_increment(
parsed,
state_reg,
target_field,
*increment_by,
);
ops.extend(cond_check_ops);
} else {
ops.push(OpCode::SetFieldIncrement {
object: state_reg,
path: target_field.clone(),
});
}
} else {
ops.push(OpCode::SetFieldIncrement {
object: state_reg,
path: target_field.clone(),
});
}
}
HookAction::RegisterPdaMapping { .. } => {
if let HookAction::RegisterPdaMapping {
pda_field,
seed_field,
lookup_name,
} = action
{
let pda_reg = 11;
let seed_reg = 12;
ops.push(OpCode::LoadEventField {
path: pda_field.clone(),
dest: pda_reg,
default: None,
});
ops.push(OpCode::LoadEventField {
path: seed_field.clone(),
dest: seed_reg,
default: None,
});
ops.push(OpCode::UpdatePdaReverseLookup {
state_id: self.state_id,
lookup_name: lookup_name.clone(),
pda_address: pda_reg,
primary_key: seed_reg,
});
}
}
}
}
ops
}
fn compile_condition_check(
&self,
condition: &ParsedCondition,
value_reg: Register,
state_reg: Register,
target_field: &str,
) -> Vec<OpCode> {
match condition {
ParsedCondition::Comparison {
field,
op,
value: cond_value,
} => {
vec![OpCode::ConditionalSetField {
object: state_reg,
path: target_field.to_string(),
value: value_reg,
condition_field: field.clone(),
condition_op: op.clone(),
condition_value: cond_value.clone(),
}]
}
ParsedCondition::Logical { .. } => {
tracing::warn!("Logical conditions not yet supported in instruction hooks");
vec![OpCode::SetField {
object: state_reg,
path: target_field.to_string(),
value: value_reg,
}]
}
}
}
fn compile_conditional_increment(
&self,
condition: &ParsedCondition,
state_reg: Register,
target_field: &str,
_increment_by: i64,
) -> Vec<OpCode> {
match condition {
ParsedCondition::Comparison {
field,
op,
value: cond_value,
} => {
vec![OpCode::ConditionalIncrement {
object: state_reg,
path: target_field.to_string(),
condition_field: field.clone(),
condition_op: op.clone(),
condition_value: cond_value.clone(),
}]
}
ParsedCondition::Logical { .. } => {
tracing::warn!("Logical conditions not yet supported in instruction hooks");
vec![OpCode::SetFieldIncrement {
object: state_reg,
path: target_field.to_string(),
}]
}
}
}
}
#[cfg(test)]
mod tests {
use super::MultiEntityBytecode;
use crate::ast::{
FieldPath, HookAction, IdentitySpec, InstructionHook, KeyResolutionStrategy,
LookupIndexSpec, MappingSource, PopulationStrategy, SerializableFieldMapping,
SerializableHandlerSpec, SerializableStreamSpec, SourceSpec, TypedStreamSpec,
};
use crate::vm::VmContext;
use crate::{Mutation, WholeEntity};
use serde_json::{json, Value};
use std::collections::BTreeMap;
fn mapping(
target_path: &str,
source_path: &[&str],
population: PopulationStrategy,
) -> SerializableFieldMapping {
SerializableFieldMapping {
target_path: target_path.to_string(),
source: MappingSource::FromSource {
path: FieldPath::new(source_path),
default: None,
transform: None,
},
transform: None,
population,
condition: None,
when: None,
stop: None,
emit: true,
}
}
fn direct_pda_to_primary_key_spec() -> TypedStreamSpec<Value> {
TypedStreamSpec::from_serializable(SerializableStreamSpec {
ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
state_name: "PumpfunToken".to_string(),
program_id: None,
idl: None,
identity: IdentitySpec {
primary_keys: vec!["id.mint".to_string()],
lookup_indexes: vec![LookupIndexSpec {
field_name: "id.bonding_curve".to_string(),
temporal_field: None,
}],
},
handlers: vec![SerializableHandlerSpec {
source: SourceSpec::Source {
program_id: None,
discriminator: None,
type_name: "pump::BondingCurveState".to_string(),
serialization: None,
is_account: true,
},
key_resolution: KeyResolutionStrategy::Lookup {
primary_field: FieldPath::new(&["__account_address"]),
},
mappings: vec![
mapping(
"id.bonding_curve",
&["__account_address"],
PopulationStrategy::SetOnce,
),
mapping(
"reserves.virtual_token_reserves",
&["virtual_token_reserves"],
PopulationStrategy::LastWrite,
),
],
conditions: vec![],
emit: true,
}],
sections: vec![],
field_mappings: BTreeMap::new(),
resolver_hooks: vec![],
instruction_hooks: vec![InstructionHook {
instruction_type: "pump::BuyIxState".to_string(),
actions: vec![HookAction::RegisterPdaMapping {
pda_field: FieldPath::new(&["accounts", "bonding_curve"]),
seed_field: FieldPath::new(&["accounts", "mint"]),
lookup_name: "default_pda_lookup".to_string(),
}],
lookup_by: None,
}],
resolver_specs: vec![],
computed_fields: vec![],
computed_field_specs: vec![],
content_hash: None,
views: vec![],
})
}
fn pda_to_intermediate_lookup_spec() -> TypedStreamSpec<Value> {
TypedStreamSpec::from_serializable(SerializableStreamSpec {
ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
state_name: "OreRound".to_string(),
program_id: None,
idl: None,
identity: IdentitySpec {
primary_keys: vec!["id.round_id".to_string()],
lookup_indexes: vec![LookupIndexSpec {
field_name: "id.round_address".to_string(),
temporal_field: None,
}],
},
handlers: vec![SerializableHandlerSpec {
source: SourceSpec::Source {
program_id: None,
discriminator: None,
type_name: "entropy::VarState".to_string(),
serialization: None,
is_account: true,
},
key_resolution: KeyResolutionStrategy::Lookup {
primary_field: FieldPath::new(&["__account_address"]),
},
mappings: vec![mapping(
"state.expires_at",
&["end_at"],
PopulationStrategy::LastWrite,
)],
conditions: vec![],
emit: true,
}],
sections: vec![],
field_mappings: BTreeMap::new(),
resolver_hooks: vec![],
instruction_hooks: vec![InstructionHook {
instruction_type: "ore::DeployIxState".to_string(),
actions: vec![HookAction::RegisterPdaMapping {
pda_field: FieldPath::new(&["accounts", "entropyVar"]),
seed_field: FieldPath::new(&["accounts", "round"]),
lookup_name: "default_pda_lookup".to_string(),
}],
lookup_by: None,
}],
resolver_specs: vec![],
computed_fields: vec![],
computed_field_specs: vec![],
content_hash: None,
views: vec![],
})
}
fn embedded_account_key_spec() -> TypedStreamSpec<Value> {
TypedStreamSpec::from_serializable(SerializableStreamSpec {
ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
state_name: "OreTreasury".to_string(),
program_id: None,
idl: None,
identity: IdentitySpec {
primary_keys: vec!["id.address".to_string()],
lookup_indexes: vec![],
},
handlers: vec![SerializableHandlerSpec {
source: SourceSpec::Source {
program_id: None,
discriminator: None,
type_name: "ore::TreasuryState".to_string(),
serialization: None,
is_account: true,
},
key_resolution: KeyResolutionStrategy::Embedded {
primary_field: FieldPath::new(&["__account_address"]),
},
mappings: vec![mapping(
"id.address",
&["__account_address"],
PopulationStrategy::SetOnce,
)],
conditions: vec![],
emit: true,
}],
sections: vec![],
field_mappings: BTreeMap::new(),
resolver_hooks: vec![],
instruction_hooks: vec![],
resolver_specs: vec![],
computed_fields: vec![],
computed_field_specs: vec![],
content_hash: None,
views: vec![],
})
}
fn whole_entity_pool_spec() -> TypedStreamSpec<Value> {
let mut hidden = mapping("state.secret", &["secret"], PopulationStrategy::LastWrite);
hidden.emit = false;
let handler = |type_name: &str, mappings, emit| SerializableHandlerSpec {
source: SourceSpec::Source {
program_id: None,
discriminator: None,
type_name: type_name.to_string(),
serialization: None,
is_account: true,
},
key_resolution: KeyResolutionStrategy::Embedded {
primary_field: FieldPath::new(&["__account_address"]),
},
mappings,
conditions: vec![],
emit,
};
TypedStreamSpec::from_serializable(SerializableStreamSpec {
ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
state_name: "Pool".to_string(),
program_id: None,
idl: None,
identity: IdentitySpec {
primary_keys: vec!["id.address".to_string()],
lookup_indexes: vec![],
},
handlers: vec![
handler(
"amm::PoolState",
vec![
mapping(
"id.address",
&["__account_address"],
PopulationStrategy::SetOnce,
),
mapping("state.price", &["price"], PopulationStrategy::LastWrite),
mapping("state.fills", &["fill"], PopulationStrategy::Append),
hidden,
],
true,
),
handler(
"amm::PoolConfigState",
vec![mapping(
"config.fee",
&["fee"],
PopulationStrategy::LastWrite,
)],
false,
),
],
sections: vec![],
field_mappings: BTreeMap::new(),
resolver_hooks: vec![],
instruction_hooks: vec![],
resolver_specs: vec![],
computed_fields: vec![],
computed_field_specs: vec![],
content_hash: None,
views: vec![],
})
}
fn pool_bytecode() -> MultiEntityBytecode {
MultiEntityBytecode::from_single("Pool".to_string(), whole_entity_pool_spec(), 0)
}
fn pool_event(price: u64, fill: u64) -> Value {
pool_event_for("pool_1", price, fill)
}
fn pool_event_for(address: &str, price: u64, fill: u64) -> Value {
json!({"__account_address": address, "price": price, "fill": fill, "secret": "s"})
}
fn linked_vm(
state_config: crate::vm::StateTableConfig,
) -> (VmContext, crate::vm::WholeEntityRequests) {
let requests = crate::vm::WholeEntityRequests::new(16);
let mut vm = VmContext::new_with_config(state_config);
vm.set_whole_entity_requests(requests.clone());
(vm, requests)
}
fn process(vm: &mut VmContext, bytecode: &MultiEntityBytecode, event: Value) -> Vec<Mutation> {
vm.process_event(bytecode, event, "amm::PoolState", None, None)
.unwrap()
}
#[test]
fn source_deletion_cancels_resync_and_recreates_a_fresh_vm_row() {
let bytecode = pool_bytecode();
let (mut vm, requests) = linked_vm(Default::default());
let first = process(&mut vm, &bytecode, pool_event(1, 10));
let key = &first[0].key;
let target = crate::vm::ResolverTarget {
state_id: 0,
entity_name: "Pool".into(),
primary_key: key.clone(),
extracts: vec![],
};
vm.enqueue_resolver_request(
"ignored".into(),
crate::ast::ResolverType::Token,
json!("mint"),
target.clone(),
);
let old_request = vm.take_resolver_requests().pop().unwrap();
vm.get_state_table_mut(0).unwrap().deferred_when_ops.insert(
("sig".into(), "instruction".into()),
vec![crate::vm::DeferredWhenOperation {
entity_name: "Pool".into(),
primary_key: key.clone(),
field_path: "old".into(),
field_value: json!(true),
when_instruction: "instruction".into(),
signature: "sig".into(),
slot: 1,
deferred_at: 1,
emit: true,
}],
);
requests.request("Pool", key);
let deletion = vm.delete_entity(&bytecode, "Pool", key).unwrap();
assert!(deletion.is_delete());
assert!(
serde_json::from_value::<Mutation>(serde_json::to_value(&deletion).unwrap())
.unwrap()
.is_delete()
);
assert!(!requests.is_requested("Pool", key));
assert!(vm.get_entity_state(0, key).is_none());
assert!(vm.snapshot_state_table(0).is_empty());
assert!(vm
.get_state_table_mut(0)
.unwrap()
.deferred_when_ops
.is_empty());
assert!(vm.delete_entity(&bytecode, "Unknown", key).is_none());
let mut next = process(&mut vm, &bytecode, pool_event(9, 90));
vm.enqueue_resolver_request(
"ignored".into(),
crate::ast::ResolverType::Token,
json!("mint"),
target,
);
let new_request = vm.take_resolver_requests().pop().unwrap();
assert_ne!(old_request.cache_key, new_request.cache_key);
assert!(vm
.apply_resolver_result(&bytecode, &old_request.cache_key, json!({"stale":true}))
.unwrap()
.is_empty());
assert_eq!(next[0].take_whole_entity_mark(), Some(WholeEntity::Created));
assert_eq!(next[0].patch["state"]["fills"], json!([90]));
}
#[test]
fn deletion_preserves_unrelated_resolver_sharing_and_isolates_recreation() {
let bytecode = pool_bytecode();
let (mut vm, _) = linked_vm(Default::default());
let key = process(&mut vm, &bytecode, pool_event(1, 10))[0]
.key
.clone();
let late_joiner = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20))[0]
.key
.clone();
let target = |primary_key| crate::vm::ResolverTarget {
state_id: 0,
entity_name: "Pool".into(),
primary_key,
extracts: vec![crate::ast::ResolverExtractSpec {
target_path: "resolved".into(),
source_path: Some("value".into()),
transform: None,
}],
};
let enqueue = |vm: &mut VmContext, key| {
vm.enqueue_resolver_request(
"ignored".into(),
crate::ast::ResolverType::Token,
json!("shared-mint"),
target(key),
)
};
enqueue(&mut vm, key.clone());
enqueue(&mut vm, json!("unrelated"));
let old = vm.take_resolver_requests();
assert_eq!(old.len(), 1);
vm.delete_entity(&bytecode, "Pool", &key).unwrap();
enqueue(&mut vm, json!("unrelated"));
enqueue(&mut vm, json!("another-unrelated"));
assert!(
vm.take_resolver_requests().is_empty(),
"unaffected targets reuse the old pending call"
);
enqueue(&mut vm, key.clone());
let fresh = vm.take_resolver_requests();
assert_eq!(fresh.len(), 1);
assert_ne!(fresh[0].cache_key, old[0].cache_key);
vm.delete_entity(&bytecode, "Pool", &late_joiner).unwrap();
process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
enqueue(&mut vm, late_joiner.clone());
let late_joiner_fresh = vm.take_resolver_requests();
assert_eq!(late_joiner_fresh.len(), 1);
assert_ne!(late_joiner_fresh[0].cache_key, old[0].cache_key);
enqueue(&mut vm, json!("unrelated"));
enqueue(&mut vm, json!("another-unrelated"));
enqueue(&mut vm, key.clone());
assert!(
vm.take_resolver_requests().is_empty(),
"unaffected targets reuse pending calls"
);
let old_mutations = vm
.apply_resolver_result(&bytecode, &old[0].cache_key, json!({"value":"old"}))
.unwrap();
assert_eq!(old_mutations.len(), 2);
assert!(old_mutations.iter().all(|mutation| mutation.key != key));
assert!(vm.get_entity_state(0, &key).is_none());
let fresh_mutations = vm
.apply_resolver_result(&bytecode, &fresh[0].cache_key, json!({"value":"fresh"}))
.unwrap();
assert_eq!(fresh_mutations.len(), 1);
assert_eq!(fresh_mutations[0].key, key);
let late_joiner_mutations = vm
.apply_resolver_result(
&bytecode,
&late_joiner_fresh[0].cache_key,
json!({"value":"fresh-late-joiner"}),
)
.unwrap();
assert_eq!(late_joiner_mutations.len(), 1);
assert_eq!(late_joiner_mutations[0].key, late_joiner);
}
#[test]
fn a_requested_entity_is_emitted_whole_once() {
let bytecode = pool_bytecode();
let (mut vm, requests) = linked_vm(Default::default());
let mut update = |price, fill| process(&mut vm, &bytecode, pool_event(price, fill));
let first = update(1, 10);
assert_eq!(first[0].whole_entity_mark(), Some(WholeEntity::Created));
let key = first[0].key.clone();
let second = update(2, 20);
assert_eq!(second.len(), 1);
assert!(!second[0].is_whole_entity());
assert_eq!(
second[0].patch,
json!({"state": {"price": 2, "fills": [20]}})
);
assert_eq!(second[0].append, vec!["state.fills".to_string()]);
requests.request("Pool", &key);
assert!(requests.is_requested("Pool", &key));
let mut third = update(3, 30);
assert_eq!(third.len(), 2);
assert!(
!third[0].is_whole_entity(),
"the change itself is untouched"
);
assert_eq!(
third[0].patch,
json!({"state": {"price": 3, "fills": [30]}})
);
assert_eq!(third[1].key, key);
assert_eq!(third[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
assert_eq!(
third[1].patch,
json!({
"id": {"address": "pool_1"},
"state": {"price": 3, "fills": [10, 20, 30]},
}),
"the whole entity, without its non-emitted field"
);
assert!(
third[1].append.is_empty(),
"its arrays are whole, not appended to"
);
assert!(requests.is_empty(), "a request is answered once");
let fourth = update(4, 40);
assert_eq!(fourth.len(), 1);
assert!(!fourth[0].is_whole_entity());
assert_eq!(
fourth[0].patch,
json!({"state": {"price": 4, "fills": [40]}})
);
}
#[test]
fn a_request_is_answered_by_the_next_call_whatever_it_changes() {
let bytecode = pool_bytecode();
let (mut vm, requests) = linked_vm(Default::default());
process(&mut vm, &bytecode, pool_event_for("pool_1", 1, 10));
process(&mut vm, &bytecode, pool_event_for("pool_2", 1, 10));
requests.request("Pool", &json!("pool_1"));
let mut next = process(&mut vm, &bytecode, pool_event_for("pool_2", 2, 20));
assert_eq!(next.len(), 2);
assert_eq!(next[0].key, json!("pool_2"));
assert!(!next[0].is_whole_entity());
assert_eq!(next[1].key, json!("pool_1"));
assert_eq!(next[1].take_whole_entity_mark(), Some(WholeEntity::Resent));
assert_eq!(
next[1].patch,
json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
);
assert!(requests.is_empty());
requests.request("Pool", &json!("pool_2"));
let answered = vm
.apply_resolver_result(&bytecode, "no-such-request", json!({}))
.unwrap();
assert_eq!(answered.len(), 1);
assert_eq!(answered[0].key, json!("pool_2"));
assert_eq!(answered[0].whole_entity_mark(), Some(WholeEntity::Resent));
let quiet = process(&mut vm, &bytecode, pool_event_for("pool_2", 3, 30));
assert_eq!(quiet.len(), 1, "a request is answered once");
}
#[test]
fn a_request_for_an_entity_the_vm_does_not_hold_is_dropped() {
let bytecode = pool_bytecode();
let (mut vm, requests) = linked_vm(Default::default());
requests.request("Pool", &json!("other_pool"));
let mutations = process(&mut vm, &bytecode, pool_event(1, 10));
assert_eq!(mutations.len(), 1);
assert_eq!(mutations[0].whole_entity_mark(), Some(WholeEntity::Created));
assert!(requests.is_empty(), "dropped, not kept for later");
let created = process(&mut vm, &bytecode, pool_event_for("other_pool", 1, 10));
assert_eq!(created.len(), 1);
assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
}
#[test]
fn whole_entity_requests_are_bounded() {
let requests = crate::vm::WholeEntityRequests::new(2);
for key in ["a", "b", "c"] {
requests.request("Pool", &json!(key));
}
assert_eq!(requests.len(), 2);
assert!(
!requests.is_requested("Pool", &json!("a")),
"the oldest is forgotten"
);
assert!(requests.is_requested("Pool", &json!("c")));
}
#[test]
fn an_entitys_first_mutation_is_marked_as_its_creation() {
let bytecode = pool_bytecode();
let (mut vm, _requests) = linked_vm(Default::default());
let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
assert_eq!(first.len(), 1, "the creation needs no second mutation");
assert_eq!(
first[0].take_whole_entity_mark(),
Some(WholeEntity::Created)
);
assert_eq!(
first[0].patch,
json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}}),
"the whole row, without its non-emitted field"
);
assert_eq!(first[0].append, vec!["state.fills".to_string()]);
let second = process(&mut vm, &bytecode, pool_event(2, 20));
assert_eq!(second[0].whole_entity_mark(), None);
assert_eq!(
second[0].patch,
json!({"state": {"price": 2, "fills": [20]}})
);
}
#[test]
fn a_vm_without_whole_entity_requests_marks_no_creation() {
let bytecode = pool_bytecode();
let mut vm = VmContext::new();
let first = process(&mut vm, &bytecode, pool_event(1, 10));
assert!(!first[0].is_whole_entity());
assert!(first[0].patch.get(crate::WHOLE_ENTITY_MARKER).is_none());
}
#[test]
fn a_row_started_without_a_mutation_is_created_by_its_first_one() {
let bytecode = pool_bytecode();
let (mut vm, _requests) = linked_vm(Default::default());
let config = vm
.process_event(
&bytecode,
json!({"__account_address": "pool_1", "fee": 30}),
"amm::PoolConfigState",
None,
None,
)
.unwrap();
assert!(config.is_empty(), "the config handler emits nothing");
let mut first = process(&mut vm, &bytecode, pool_event(1, 10));
assert_eq!(
first[0].take_whole_entity_mark(),
Some(WholeEntity::Created)
);
assert_eq!(
first[0].patch,
json!({
"id": {"address": "pool_1"},
"config": {"fee": 30},
"state": {"price": 1, "fills": [10]},
})
);
let second = process(&mut vm, &bytecode, pool_event(2, 20));
assert!(!second[0].is_whole_entity());
}
#[test]
fn an_entity_the_vm_dropped_is_created_again() {
let bytecode = pool_bytecode();
let (mut vm, _requests) = linked_vm(crate::vm::StateTableConfig {
max_entries: 2,
..Default::default()
});
for pool in ["a", "b", "c"] {
let created = process(&mut vm, &bytecode, pool_event_for(pool, 1, 1));
assert_eq!(created[0].whole_entity_mark(), Some(WholeEntity::Created));
}
assert!(vm.get_entity_state(0, &json!("a")).is_none(), "dropped");
let mut again = process(&mut vm, &bytecode, pool_event_for("a", 2, 2));
assert_eq!(
again[0].take_whole_entity_mark(),
Some(WholeEntity::Created)
);
assert_eq!(
again[0].patch,
json!({"id": {"address": "a"}, "state": {"price": 2, "fills": [2]}})
);
let changed = process(&mut vm, &bytecode, pool_event_for("c", 2, 2));
assert!(!changed[0].is_whole_entity(), "c was never dropped");
}
#[test]
fn a_restored_row_never_emitted_is_sent_whole_on_request() {
let bytecode = pool_bytecode();
let (mut vm, _requests) = linked_vm(Default::default());
let config = vm
.process_event(
&bytecode,
json!({"__account_address": "pool_1", "fee": 30}),
"amm::PoolConfigState",
None,
None,
)
.unwrap();
assert!(config.is_empty());
let snapshot = vm.dump();
let (mut restored, requests) = linked_vm(Default::default());
restored.hydrate(snapshot);
let first = process(&mut restored, &bytecode, pool_event(1, 10));
assert!(!first[0].is_whole_entity(), "counted as emitted already");
assert_eq!(
first[0].patch,
json!({"id": {"address": "pool_1"}, "state": {"price": 1, "fills": [10]}})
);
requests.request("Pool", &first[0].key);
let next = process(&mut restored, &bytecode, pool_event_for("pool_2", 1, 1));
let resent = next
.iter()
.find(|mutation| mutation.key == json!("pool_1"))
.expect("the requested entity follows the next call");
assert_eq!(resent.whole_entity_mark(), Some(WholeEntity::Resent));
let mut resent = resent.clone();
resent.take_whole_entity_mark();
assert_eq!(
resent.patch,
json!({
"id": {"address": "pool_1"},
"config": {"fee": 30},
"state": {"price": 1, "fills": [10]},
})
);
}
#[test]
fn a_resolver_result_for_an_entity_not_held_creates_it() {
let bytecode = pool_bytecode();
let (mut vm, _requests) = linked_vm(Default::default());
process(&mut vm, &bytecode, pool_event(1, 10));
for key in ["pool_1", "pool_2"] {
vm.enqueue_resolver_request(
String::new(),
crate::ast::ResolverType::Token,
json!(key),
crate::vm::ResolverTarget {
state_id: 0,
entity_name: "Pool".to_string(),
primary_key: json!(key),
extracts: vec![crate::ast::ResolverExtractSpec {
target_path: "meta.name".to_string(),
source_path: Some("name".to_string()),
transform: None,
}],
},
);
}
let mut applied = Vec::new();
for request in vm.take_resolver_requests() {
applied.extend(
vm.apply_resolver_result(&bytecode, &request.cache_key, json!({"name": "n"}))
.unwrap(),
);
}
applied.sort_by_key(|mutation| mutation.key.to_string());
assert_eq!(applied[0].key, json!("pool_1"));
assert!(!applied[0].is_whole_entity());
assert_eq!(applied[0].patch, json!({"meta": {"name": "n"}}));
assert_eq!(applied[1].key, json!("pool_2"));
assert_eq!(
applied[1].take_whole_entity_mark(),
Some(WholeEntity::Created)
);
assert_eq!(applied[1].patch, json!({"meta": {"name": "n"}}));
}
#[test]
fn a_restored_entity_is_not_created_again() {
let bytecode = pool_bytecode();
let (mut vm, _requests) = linked_vm(Default::default());
process(&mut vm, &bytecode, pool_event(1, 10));
let snapshot = vm.dump();
let (mut restored, _requests) = linked_vm(Default::default());
restored.hydrate(snapshot);
let next = process(&mut restored, &bytecode, pool_event(2, 20));
assert!(!next[0].is_whole_entity());
assert_eq!(next[0].patch, json!({"state": {"price": 2, "fills": [20]}}));
}
#[test]
fn lookup_account_handler_uses_resolved_primary_key_when_hook_seeds_primary_key() {
let bytecode = MultiEntityBytecode::from_single(
"PumpfunToken".to_string(),
direct_pda_to_primary_key_spec(),
0,
);
let mut vm = VmContext::new();
let mutations = vm
.process_event(
&bytecode,
json!({
"__account_address": "bonding_curve_1",
"__resolved_primary_key": "mint_1",
"virtual_token_reserves": 42,
}),
"pump::BondingCurveState",
None,
None,
)
.unwrap();
assert_eq!(mutations.len(), 1);
assert_eq!(mutations[0].key, json!("mint_1"));
assert_eq!(
mutations[0].patch["id"]["bonding_curve"],
json!("bonding_curve_1")
);
assert_eq!(
mutations[0].patch["reserves"]["virtual_token_reserves"],
json!(42)
);
}
#[test]
fn lookup_account_handler_keeps_null_key_when_resolver_only_returns_intermediate_lookup() {
let bytecode = MultiEntityBytecode::from_single(
"OreRound".to_string(),
pda_to_intermediate_lookup_spec(),
0,
);
let mut vm = VmContext::new();
let mutations = vm
.process_event(
&bytecode,
json!({
"__account_address": "entropy_var_1",
"__resolved_primary_key": "round_address_1",
"end_at": 123,
}),
"entropy::VarState",
None,
None,
)
.unwrap();
assert!(mutations.is_empty());
}
#[test]
fn embedded_account_key_wins_over_an_unrelated_resolved_key() {
let bytecode = MultiEntityBytecode::from_single(
"OreTreasury".to_string(),
embedded_account_key_spec(),
0,
);
let mut vm = VmContext::new();
let mutations = vm
.process_event(
&bytecode,
json!({
"__account_address": "treasury_pda",
"__resolved_primary_key": "round_address",
}),
"ore::TreasuryState",
None,
None,
)
.unwrap();
assert_eq!(mutations.len(), 1);
assert_eq!(mutations[0].key, json!("treasury_pda"));
assert_eq!(mutations[0].patch["id"]["address"], json!("treasury_pda"));
}
fn instruction_handler(
type_name: &str,
key_resolution: KeyResolutionStrategy,
mappings: Vec<SerializableFieldMapping>,
) -> SerializableHandlerSpec {
SerializableHandlerSpec {
source: SourceSpec::Source {
program_id: None,
discriminator: None,
type_name: type_name.to_string(),
serialization: None,
is_account: false,
},
key_resolution,
mappings,
conditions: vec![],
emit: true,
}
}
fn split_position_spec() -> TypedStreamSpec<Value> {
let embedded = |field: &str| KeyResolutionStrategy::Embedded {
primary_field: FieldPath::new(&["accounts", field]),
};
TypedStreamSpec::from_serializable(SerializableStreamSpec {
ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
state_name: "Position".to_string(),
program_id: None,
idl: None,
identity: IdentitySpec {
primary_keys: vec!["id.position_address".to_string()],
lookup_indexes: vec![],
},
handlers: vec![
instruction_handler(
"amm::SplitPositionIxState",
embedded("first_position"),
vec![
mapping(
"id.position_address",
&["accounts", "first_position"],
PopulationStrategy::SetOnce,
),
mapping(
"activity.split_source_count",
&["data"],
PopulationStrategy::Count,
),
],
),
instruction_handler(
"amm::SplitPositionIxState",
embedded("second_position"),
vec![
mapping(
"id.position_address",
&["accounts", "second_position"],
PopulationStrategy::SetOnce,
),
mapping(
"activity.split_child_count",
&["data"],
PopulationStrategy::Count,
),
],
),
],
sections: vec![],
field_mappings: BTreeMap::new(),
resolver_hooks: vec![],
instruction_hooks: vec![InstructionHook {
instruction_type: "amm::SplitPositionIxState".to_string(),
actions: vec![HookAction::SetField {
target_field: "activity.last_split_source_slot".to_string(),
source: MappingSource::FromSource {
path: FieldPath::new(&["data", "slot"]),
default: None,
transform: None,
},
condition: None,
}],
lookup_by: Some(FieldPath::new(&["accounts", "first_position"])),
}],
resolver_specs: vec![],
computed_fields: vec![],
computed_field_specs: vec![],
content_hash: None,
views: vec![],
})
}
fn split_position_event(first: &str, second: &str) -> Value {
json!({
"accounts": { "first_position": first, "second_position": second },
"data": { "slot": 77 },
})
}
#[test]
fn one_instruction_routes_each_key_to_its_own_entity() {
let bytecode =
MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
let mut vm = VmContext::new();
for _ in 0..2 {
vm.process_event(
&bytecode,
split_position_event("source_pos", "child_pos"),
"amm::SplitPositionIxState",
None,
None,
)
.unwrap();
}
vm.process_event(
&bytecode,
split_position_event("child_pos", "grandchild_pos"),
"amm::SplitPositionIxState",
None,
None,
)
.unwrap();
let state = |key: &str| vm.get_entity_state(0, &json!(key)).unwrap();
let source = state("source_pos");
assert_eq!(source["id"]["position_address"], json!("source_pos"));
assert_eq!(source["activity"]["split_source_count"], json!(2));
assert_eq!(source["activity"].get("split_child_count"), None);
assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
let child = state("child_pos");
assert_eq!(child["id"]["position_address"], json!("child_pos"));
assert_eq!(child["activity"]["split_child_count"], json!(2));
assert_eq!(child["activity"]["split_source_count"], json!(1));
assert_eq!(child["activity"]["last_split_source_slot"], json!(77));
let grandchild = state("grandchild_pos");
assert_eq!(grandchild["activity"]["split_child_count"], json!(1));
assert_eq!(grandchild["activity"].get("split_source_count"), None);
assert_eq!(grandchild["activity"].get("last_split_source_slot"), None);
}
#[test]
fn one_instruction_emits_one_mutation_per_key() {
let bytecode =
MultiEntityBytecode::from_single("Position".to_string(), split_position_spec(), 0);
let mut vm = VmContext::new();
let mutations = vm
.process_event(
&bytecode,
split_position_event("source_pos", "child_pos"),
"amm::SplitPositionIxState",
None,
None,
)
.unwrap();
assert_eq!(mutations.len(), 2);
assert_eq!(mutations[0].key, json!("source_pos"));
assert_eq!(
mutations[0].patch["activity"]["split_source_count"],
json!(1)
);
assert_eq!(
mutations[0].patch["activity"]["last_split_source_slot"],
json!(77)
);
assert_eq!(
mutations[0].patch["activity"].get("split_child_count"),
None
);
assert_eq!(mutations[1].key, json!("child_pos"));
assert_eq!(
mutations[1].patch["activity"]["split_child_count"],
json!(1)
);
assert_eq!(
mutations[1].patch["activity"].get("split_source_count"),
None
);
}
#[test]
fn handlers_with_identical_keys_still_merge_into_one_segment() {
let mut spec = split_position_spec();
spec.instruction_hooks.clear();
spec.handlers[1].key_resolution = KeyResolutionStrategy::Embedded {
primary_field: FieldPath::new(&["accounts", "first_position"]),
};
let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
let handler = &bytecode.entities["Position"].handlers["amm::SplitPositionIxState"];
assert!(!handler
.iter()
.any(|op| matches!(op, super::OpCode::SegmentBoundary)));
assert_eq!(
handler
.iter()
.filter(|op| matches!(op, super::OpCode::ReadOrInitState { .. }))
.count(),
1
);
}
#[test]
fn hook_without_a_matching_key_gets_its_own_segment() {
let mut spec = split_position_spec();
spec.handlers.truncate(1);
spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
primary_field: FieldPath::new(&["accounts", "second_position"]),
};
spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
"id.position_address",
&["accounts", "second_position"],
PopulationStrategy::SetOnce,
));
let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
let mut vm = VmContext::new();
vm.process_event(
&bytecode,
split_position_event("source_pos", "child_pos"),
"amm::SplitPositionIxState",
None,
None,
)
.unwrap();
let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
assert_eq!(child["activity"].get("last_split_source_slot"), None);
}
fn add_owner_index(spec: &mut TypedStreamSpec<Value>) {
spec.identity.lookup_indexes.push(LookupIndexSpec {
field_name: "id.owner".to_string(),
temporal_field: None,
});
spec.handlers
.push(crate::ast::TypedHandlerSpec::from_serializable(
SerializableHandlerSpec {
source: SourceSpec::Source {
program_id: None,
discriminator: None,
type_name: "amm::PositionState".to_string(),
serialization: None,
is_account: true,
},
key_resolution: KeyResolutionStrategy::Embedded {
primary_field: FieldPath::new(&["__account_address"]),
},
mappings: vec![
mapping(
"id.position_address",
&["__account_address"],
PopulationStrategy::SetOnce,
),
mapping("id.owner", &["owner"], PopulationStrategy::SetOnce),
],
conditions: vec![],
emit: true,
},
));
}
fn index_owner(
vm: &mut VmContext,
bytecode: &MultiEntityBytecode,
position: &str,
owner: &str,
) {
vm.process_event(
bytecode,
json!({ "__account_address": position, "owner": owner }),
"amm::PositionState",
None,
None,
)
.unwrap();
}
#[test]
fn unmatched_hook_next_to_other_keys_resolves_lookup_index_fields() {
let mut spec = split_position_spec();
spec.handlers.truncate(1);
spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
primary_field: FieldPath::new(&["accounts", "second_position"]),
};
spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
"id.position_address",
&["accounts", "second_position"],
PopulationStrategy::SetOnce,
));
add_owner_index(&mut spec);
spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
let mut vm = VmContext::new();
index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
vm.process_event(
&bytecode,
json!({
"accounts": {
"first_position": "source_pos",
"second_position": "child_pos",
"owner": "owner_1",
},
"data": { "slot": 77 },
}),
"amm::SplitPositionIxState",
None,
None,
)
.unwrap();
let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
assert!(vm.get_entity_state(0, &json!("owner_1")).is_none());
}
#[test]
fn unmatched_hook_replays_after_its_lookup_index_is_populated() {
let mut spec = split_position_spec();
spec.handlers.truncate(1);
spec.handlers[0].key_resolution = KeyResolutionStrategy::Embedded {
primary_field: FieldPath::new(&["accounts", "second_position"]),
};
spec.handlers[0].mappings[0] = crate::ast::TypedFieldMapping::from_serializable(mapping(
"id.position_address",
&["accounts", "second_position"],
PopulationStrategy::SetOnce,
));
add_owner_index(&mut spec);
spec.instruction_hooks[0].lookup_by = Some(FieldPath::new(&["accounts", "owner"]));
let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
let mut vm = VmContext::new();
let mutations = vm
.process_event(
&bytecode,
json!({
"accounts": { "owner": "owner_1", "second_position": "child_pos" },
"data": { "slot": 77 },
}),
"amm::SplitPositionIxState",
None,
None,
)
.unwrap();
assert_eq!(mutations.len(), 1);
assert_eq!(mutations[0].key, json!("child_pos"));
index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
assert_eq!(source["activity"]["last_split_source_slot"], json!(77));
let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
assert_eq!(child["activity"].get("last_split_source_slot"), None);
}
#[test]
fn a_segment_that_misses_its_key_is_replayed_alone() {
let mut spec = split_position_spec();
spec.instruction_hooks.clear();
spec.handlers.swap(0, 1);
spec.handlers[1].key_resolution = KeyResolutionStrategy::Lookup {
primary_field: FieldPath::new(&["accounts", "source_pda"]),
};
spec.handlers[1].mappings.remove(0);
add_owner_index(&mut spec);
spec.instruction_hooks.push(InstructionHook {
instruction_type: "amm::RegisterIxState".to_string(),
actions: vec![HookAction::RegisterPdaMapping {
pda_field: FieldPath::new(&["accounts", "source_pda"]),
seed_field: FieldPath::new(&["accounts", "owner"]),
lookup_name: "default_pda_lookup".to_string(),
}],
lookup_by: None,
});
let bytecode = MultiEntityBytecode::from_single("Position".to_string(), spec, 0);
let mut vm = VmContext::new();
index_owner(&mut vm, &bytecode, "source_pos", "owner_1");
let mutations = vm
.process_event(
&bytecode,
json!({
"accounts": { "source_pda": "pda_1", "second_position": "child_pos" },
"data": {},
}),
"amm::SplitPositionIxState",
None,
None,
)
.unwrap();
assert_eq!(mutations.len(), 1);
assert_eq!(mutations[0].key, json!("child_pos"));
let mutations = vm
.process_event(
&bytecode,
json!({ "accounts": { "source_pda": "pda_1", "owner": "owner_1" } }),
"amm::RegisterIxState",
None,
None,
)
.unwrap();
assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
assert_eq!(mutations[0].key, json!("source_pos"));
let source = vm.get_entity_state(0, &json!("source_pos")).unwrap();
assert_eq!(source["activity"]["split_source_count"], json!(1));
let child = vm.get_entity_state(0, &json!("child_pos")).unwrap();
assert_eq!(child["activity"]["split_child_count"], json!(1));
}
#[test]
fn a_hook_whose_lookup_field_is_absent_skips_only_its_segment() {
let spec: TypedStreamSpec<Value> =
TypedStreamSpec::from_serializable(SerializableStreamSpec {
ast_version: crate::ast::CURRENT_AST_VERSION.to_string(),
state_name: "Game".to_string(),
program_id: None,
idl: None,
identity: IdentitySpec {
primary_keys: vec!["id.game_id".to_string()],
lookup_indexes: vec![],
},
handlers: vec![instruction_handler(
"game::StartGameIxState",
KeyResolutionStrategy::Embedded {
primary_field: FieldPath::new(&["data", "game_id"]),
},
vec![
SerializableFieldMapping {
source: MappingSource::FromSource {
path: FieldPath::new(&["data", "game_id"]),
default: None,
transform: Some(crate::ast::Transformation::HexEncode),
},
..mapping("id.game_id", &[], PopulationStrategy::SetOnce)
},
mapping("stats.fee", &["data", "fee"], PopulationStrategy::SetOnce),
],
)],
sections: vec![],
field_mappings: BTreeMap::new(),
resolver_hooks: vec![],
instruction_hooks: vec![InstructionHook {
instruction_type: "game::StartGameIxState".to_string(),
actions: vec![HookAction::SetField {
target_field: "stats.started_slot".to_string(),
source: MappingSource::FromSource {
path: FieldPath::new(&["data", "slot"]),
default: None,
transform: None,
},
condition: None,
}],
lookup_by: Some(FieldPath::new(&["accounts", "game_id"])),
}],
resolver_specs: vec![],
computed_fields: vec![],
computed_field_specs: vec![],
content_hash: None,
views: vec![],
});
let bytecode = MultiEntityBytecode::from_single("Game".to_string(), spec, 0);
let mut vm = VmContext::new();
let mutations = vm
.process_event(
&bytecode,
json!({
"accounts": { "game_account": "game_pda" },
"data": { "game_id": [1, 2, 3, 4], "fee": 5, "slot": 9 },
}),
"game::StartGameIxState",
None,
None,
)
.unwrap();
assert_eq!(mutations.len(), 1, "mutations: {mutations:?}");
assert_eq!(mutations[0].key, json!("01020304"));
let game = vm.get_entity_state(0, &json!("01020304")).unwrap();
assert_eq!(game["id"]["game_id"], json!("01020304"));
assert_eq!(game["stats"]["fee"], json!(5));
assert_eq!(game["stats"].get("started_slot"), None);
assert_eq!(vm.get_entity_state(0, &json!(null)), None);
}
mod fingerprint {
use super::super::{EntityBytecode, MultiEntityBytecode, OpCode};
use crate::ast::FieldPath;
use serde_json::json;
use std::collections::{HashMap, HashSet};
fn sample_handler() -> Vec<OpCode> {
vec![
OpCode::LoadEventField {
path: FieldPath::new(&["id", "mint"]),
dest: 0,
default: Some(json!({"b": 2, "a": 1})),
},
OpCode::CreateObject { dest: 1 },
OpCode::SetField {
object: 1,
path: "state.volume".to_string(),
value: 0,
},
OpCode::EmitMutation {
entity_name: "PumpfunToken".to_string(),
key: 0,
state: 1,
},
]
}
fn bytecode(handler_insert_order: &[&str]) -> MultiEntityBytecode {
let mut handlers = HashMap::new();
for event_type in handler_insert_order {
handlers.insert(event_type.to_string(), sample_handler());
}
let mut entities = HashMap::new();
entities.insert(
"PumpfunToken".to_string(),
EntityBytecode {
state_id: 0,
handlers,
entity_name: "PumpfunToken".to_string(),
when_events: HashSet::from(["pump::BuyIxState".to_string()]),
non_emitted_fields: HashSet::new(),
computed_paths: vec!["state.market_cap".to_string()],
computed_fields_evaluator: None,
},
);
let mut event_routing = HashMap::new();
event_routing.insert(
"pump::TokenState".to_string(),
vec!["PumpfunToken".to_string()],
);
MultiEntityBytecode {
entities,
event_routing,
when_events: HashSet::from(["pump::BuyIxState".to_string()]),
proto_router: crate::proto_router::ProtoRouter::new(),
}
}
#[test]
fn stable_across_recomputation_and_insertion_order() {
let a = bytecode(&["pump::TokenState", "pump::BuyIxState"]);
let b = bytecode(&["pump::BuyIxState", "pump::TokenState"]);
assert_eq!(a.fingerprint(), a.fingerprint());
assert_eq!(a.fingerprint(), b.fingerprint());
assert_eq!(a.fingerprint().len(), 64);
}
#[test]
fn changes_when_an_opcode_operand_changes() {
let base = bytecode(&["pump::TokenState"]);
let mut modified = bytecode(&["pump::TokenState"]);
let entity = modified.entities.get_mut("PumpfunToken").unwrap();
match &mut entity.handlers.get_mut("pump::TokenState").unwrap()[2] {
OpCode::SetField { path, .. } => *path = "state.volume_usd".to_string(),
other => panic!("unexpected opcode: {other:?}"),
}
assert_ne!(base.fingerprint(), modified.fingerprint());
}
#[test]
fn changes_when_routing_changes() {
let base = bytecode(&["pump::TokenState"]);
let mut modified = bytecode(&["pump::TokenState"]);
modified.event_routing.insert(
"pump::SellIxState".to_string(),
vec!["PumpfunToken".to_string()],
);
assert_ne!(base.fingerprint(), modified.fingerprint());
}
#[test]
fn golden_hash_pins_the_canonical_encoding() {
let expected = "acf7ea8bf39ca830a9e6c2b2d2fb471da68e57e24bbb4c256964a42a498f75f2";
assert_eq!(bytecode(&["pump::TokenState"]).fingerprint(), expected);
}
}
}