Feature: Event Processing

Relevant source files

The following files were used as context for generating this wiki page:

• apps/api/src/routes/events.ts
• apps/runtime/crates/core/src/workflow.rs
• apps/runtime/crates/ts-addon/src/lib.rs
• README.md
• apps/api/examples/nodejs-usage.ts
• apps/api/examples/express-app.js
• apps/runtime/crates/core/tests/workflow.rs

The Event Processing system is the core of the Kodus Runtime. It is a comprehensive system designed for ingesting, persisting, and processing events with rich metadata. The architecture combines a high-performance Rust backend for durable storage and complex processing with a Node.js layer that exposes functionality via a native addon and a REST API. This allows for both efficient, low-level event handling and easy integration with JavaScript/TypeScript applications.

The system is built around a typed event structure, a durable store powered by Sled, and a flexible workflow engine. Events can be created, persisted, queried by various attributes, and processed through multi-step workflows that can transform, filter, and route data. The entire feature is accessible through a Node.js API, making it a powerful foundation for building event-driven applications.

Sources: README.md, apps/runtime/crates/ts-addon/src/lib.rs

Architecture Overview

The event processing architecture is layered, with the Rust core providing the foundational services and the Node.js addon acting as a bridge to the application layer.

1. API Layer (Node.js/Elysia): Exposes REST endpoints for interacting with the system (e.g., creating and querying events). It handles HTTP requests and translates them into calls to the KodusRuntime addon.
2. Node.js Addon (kodus_runtime): A native Node.js module written in Rust. It wraps the core Rust logic in the KodusRuntime struct, exposing methods for event persistence, workflow management, and statistics retrieval to the JavaScript/TypeScript environment.
3. Rust Core (kodus-core): The heart of the system. It contains the DurableStore for data persistence using Sled, the WorkflowEngine for processing event pipelines, and the core data structures like Event.

This diagram illustrates the high-level interaction between the components.

Event Structure

Events are the fundamental data unit. The system uses a well-defined structure in both Rust (Event) and its JavaScript representation (EventData) to ensure consistency. An event consists of a payload and extensive metadata for tracking, routing, and prioritization.

The diagram above shows the Rust Event structure.

Sources: apps/runtime/crates/ts-addon/src/lib.rs:177-187, apps/runtime/crates/core/src/workflow.rs:100-118

Event Fields

The table below details the fields available in the EventData object used in the Node.js layer.

Field	Type	Description
id	string	A unique identifier for the event (UUID).
payload	Buffer	The main data of the event, typically a JSON string.
source	string	The origin of the event (e.g., 'payment-service').
priority	number	The processing priority: 0 (Low), 1 (Normal), 2 (High), 3 (Critical).
userId	string (optional)	The ID of the user associated with the event.
correlationId	string (optional)	A UUID to group related events across different sources.
tags	[string, string][]	A list of key-value pairs for custom metadata and filtering.
timestamp	string	The ISO 8601 timestamp of when the event was created.
retryCount	number	The number of times processing has been attempted for this event.

Sources: apps/runtime/crates/ts-addon/src/lib.rs:177-187, apps/api/examples/README.md

Event Persistence and Retrieval

Events are persisted durably using the DurableStore in the Rust core. The Node.js addon provides asynchronous methods to access this functionality.

Persistence Flow

When an event is created via the API, it flows from the Node.js environment down to the Rust core to be written to the Sled database.

This sequence diagram shows the process of persisting a new event.

Sources: apps/api/src/routes/events.ts:7-45, apps/runtime/crates/ts-addon/src/lib.rs:43-74

Key Functions

• persistEventWithMetadata(...): The primary method in the Node.js addon for persisting a new event with full metadata. It constructs a Rust Event object and passes it to the DurableStore.
• findEventsBySource(source): Retrieves all events associated with a specific source string.
• getStats(): Returns statistics about the store, including the total number of events.

Sources: apps/runtime/crates/ts-addon/src/lib.rs:43-111

Workflow Engine

A key feature of event processing is the ability to run events through predefined workflows. The WorkflowEngine in the Rust core is responsible for this. Workflows are defined as a series of steps that are executed sequentially.

Workflow Structure

A workflow is defined by a WorkflowDefinition struct, which contains a name, description, and a list of WorkflowSteps.

Component	Description
WorkflowDefinition	The top-level container for a workflow, including its ID, name, steps, and error handling configuration.
WorkflowStep	A single action within a workflow. Each step has a type, configuration, and pointers to the next step or an error-handling step.
StepType	An enum defining the kind of action a step performs.
WorkflowEngine	The runtime component that executes a workflow for a given event.

Sources: apps/runtime/crates/core/src/workflow.rs:6-24

Workflow Step Types

The WorkflowEngine supports several types of steps, allowing for complex event processing logic.

Step Type	Description
Transform	Modifies the event, for example, by adding tags.
Filter	Checks for a condition (e.g., presence of a tag) and can stop the workflow if the condition is not met.
Route	Adds routing information to the event's metadata.
ExternalCall	Simulates making an external call, adding a tag to mark completion.
Delay	Pauses the workflow for a configured duration.
Condition	Checks for a specific condition on the event, such as the presence of a user_id.

Sources: apps/runtime/crates/core/src/workflow.rs:26-34, apps/runtime/crates/core/tests/workflow.rs:136-155

Workflow Execution Flow

The WorkflowEngine processes an event by iterating through the defined steps.

This flowchart illustrates the logic for executing a workflow.

Sources: apps/runtime/crates/core/src/workflow.rs:70-93

Node.js Integration

The kodus_runtime native addon exposes the core Rust functionality to Node.js applications through the KodusRuntime class.

KodusRuntime Class Methods

Method	Description
new(data_dir)	Constructor that initializes the runtime and opens the durable store at the specified path.
persistEvent(payload, source)	Persists a simple event with minimal metadata.
persistEventWithMetadata(...)	Persists an event with full metadata, including priority, tags, and IDs.
findEventsBySource(source)	Finds all events matching a given source.
getStats()	Retrieves statistics from the durable store.
createWorkflow(name, description)	Registers a new, simple workflow definition in the engine.
executeWorkflow(workflowId, eventData)	Executes a registered workflow with a given event.

Sources: apps/runtime/crates/ts-addon/src/lib.rs:13-228

Usage Example

Here is how to use the KodusRuntime class in TypeScript to persist and process an event.

import { KodusRuntime, EventData } from 'kodus_runtime';

async function run() {
    // 1. Initialize the runtime
    const runtime = new KodusRuntime('./data');

    // 2. Persist an event with metadata
    const eventId = await runtime.persistEventWithMetadata(
        Buffer.from('{"payment": {"amount": 100.50}}'),
        'payment-service',
        2, // Priority: High
        'user-123',
        'correlation-456',
        [['environment', 'production']]
    );
    console.log(`Event persisted with ID: ${eventId}`);

    // 3. Create and execute a workflow
    const workflowId = await runtime.createWorkflow(
        'payment-validation',
        'Validate and process payment events'
    );
    const events = await runtime.findEventsBySource('payment-service');
    if (events.length > 0) {
        const processedEvent = await runtime.executeWorkflow(workflowId, events[0]);
        console.log('Workflow executed. Processed event tags:', processedEvent.tags);
    }
}

run();

Sources: apps/api/examples/nodejs-usage.ts, apps/api/examples/nodejs-usage.js

API Endpoints

The event processing system is exposed via a set of REST API endpoints, which are protected by authentication.

Endpoint	Method	Description
/api/events	POST	Creates and persists a new event. The request body should contain the event data.
/api/events/:source	GET	Retrieves a list of all events for a given source.
/api/events/stats	GET	Returns statistics about the event store, such as the total number of events.

Sources: apps/api/src/routes/events.ts, README.md