Event Flow

async fn dispatch(&self, _ctx: &Context, event: &FullEvent) {
    match event {
        FullEvent::Message { new_message: msg, .. } => {
            let payload = EventMessage::from(msg);
            let guild_id = msg.guild_id.map(|g| g.get().to_string());
            // ...
        }
        // ... other events
    }
}

#[expose_payload]
pub struct EventMessage {
    id: String,
    channel_id: String,
    guild_id: Option<String>,
    content: String,
    author: EventUser,
    member: Option<EventMember>,
}

let guild_id = msg.guild_id.map(|guild| guild.get().to_string());
if guild_id.is_none() {
    return; // Drop non-guild events
}

self.runtime
    .dispatch_js_event("messageCreate", guild_id, value)
    .await

pub async fn dispatch_js_event(
    &self,
    event: &str,
    guild_id: Option<String>,
    payload: Value,
) -> Result<(), AnyError> {
    match &guild_id {
        Some(gid) => {
            let worker_idx = self.worker_for_guild(gid);
            self.workers[worker_idx]
                .dispatch(guild_id, event.to_string(), payload)
                .await
        }
        None => {
            // Broadcast to all workers
        }
    }
}

if self.enqueue_migrating_event(gid, QueuedGuildEvent { event, payload }).await {
    return Ok(()); // Event queued during migration
}

if is_droppable_event(event)
    && worker.backlog.load(Ordering::Relaxed) >= MAX_DROPPABLE_BACKLOG
{
    info!("dropping event due to backlog");
    return Ok(());
}

pub async fn dispatch(
    &self,
    guild_id: Option<String>,
    event: String,
    payload: Value,
) -> Result<(), AnyError> {
    let (tx, rx) = oneshot::channel();
    self.backlog.fetch_add(1, Ordering::Relaxed);
    self.sender.send(WorkerCommand::DispatchEvent {
        guild_id,
        event,
        payload,
        respond_to: tx,
    })?;
    rx.await?
}

WorkerCommand::DispatchEvent { guild_id, event, payload, respond_to } => {
    let result = dispatch_to_worker(
        &mut guild_runtimes,
        &mut default_runtime,
        guild_id,
        event,
        payload,
        worker_id,
        &limits,
    ).await;
    let _ = respond_to.send(result);
}

let runtime = match guild_id {
    Some(ref gid) => guild_runtimes
        .get_mut(gid)
        .ok_or_else(|| AnyError::msg("No runtime available for guild"))?,
    None => default_runtime
        .as_mut()
        .ok_or_else(|| AnyError::msg("No default runtime available"))?,
};

pub async fn dispatch_into_runtime(
    js_state: &mut JsRuntimeState,
    event: String,
    payload: Value,
    worker_id: usize,
    limits: &RuntimeLimits,
) -> Result<(), AnyError> {
    let _secret_scope = SecretScope::enter(js_state.secrets.clone());
    let dispatch_fn = js_state.dispatch_fn.as_ref().unwrap().clone();

    // Convert event and payload to V8 values
    let event_value = serde_v8::to_v8(scope, &event)?;
    let payload_value = serde_v8::to_v8(scope, &payload)?;

    // Call dispatch function
    let call = js_state.runtime_mut()
        .call_with_args(&dispatch_fn, &[event_value, payload_value]);

    // Run with timeout
    with_timeout(limits.dispatch_timeout, async {
        js_state.runtime_mut()
            .with_event_loop_promise(call, PollEventLoopOptions::default())
            .await
    }, "dispatch").await
}

let _secret_scope = SecretScope::enter(js_state.secrets.clone());

function dispatch(event: string, payload: any) {
    const handlers = eventHandlers.get(event) || []
    for (const handler of handlers) {
        await handler(payload)
    }
}

with_timeout(limits.dispatch_timeout, async {
    js_state.runtime_mut()
        .with_event_loop_promise(call, PollEventLoopOptions::default())
        .await
}, "dispatch").await

if let Err(ref err) = result {
    if err.is::<RuntimeTimeout>() {
        metrics().timeout_error();
        terminate_runtime(js_state.runtime_mut(), worker_id, "dispatch").await;
    }
}

match &result {
    Ok(_) => metrics().dispatch_success(start.elapsed()),
    Err(_) => metrics().dispatch_error(),
}

None => {
    let futures: Vec<_> = self.workers.iter()
        .map(|w| w.broadcast(event.to_string(), payload.clone()))
        .collect();
    futures::future::try_join_all(futures).await?;
}

let event_name = format!("__cron:{}", cron_name);
let payload = serde_json::json!({
    "name": cron_name,
    "scheduledAt": now.to_rfc3339(),
});

const MAX_DROPPABLE_BACKLOG: usize = 1000;

async fn enqueue_migrating_event(&self, guild_id: &str, event: QueuedGuildEvent) -> bool {
    let mut queues = self.migration_queues.lock().await;
    let Some(queue) = queues.get_mut(guild_id) else {
        return false; // Not migrating
    };
    queue.push(event);
    true
}