Quickstart

Build an event-sourced blog with htmx comments — one plain model, one event-sourced aggregate, one projection, two views, and a background job.

1. Scaffold a new app

rustavel new blog
cd blog
cp .env.example .env

Developing against a local checkout instead of a published release? Use rustavel new blog --local.

2. A plain model: Comment

rustavel make:model Comment --migration

#[derive(Model, sqlx::FromRow, Serialize, Deserialize, Debug, Clone)]
#[model(table = "comments", timestamps)]
pub struct Comment {
    #[model(primary_key, skip_on_insert)]
    pub id: i64,
    pub post_id: String,
    pub body: String,
    #[model(skip_on_insert)]
    pub created_at: chrono::NaiveDateTime,
    #[model(skip_on_insert)]
    pub updated_at: chrono::NaiveDateTime,
}

3. An event-sourced aggregate: Post

rustavel make:aggregate Post

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum PostEvent {
    Created { title: String, body: String },
    Edited { title: String, body: String },
}

#[derive(Debug, Default, Serialize, Deserialize, Aggregate)]
#[aggregate(stream = "post", event = "PostEvent")]
pub struct Post {
    pub title: String,
    pub body: String,
}

impl Post {
    fn apply_event(&mut self, event: &PostEvent) {
        match event {
            PostEvent::Created { title, body } | PostEvent::Edited { title, body } => {
                self.title = title.clone();
                self.body = body.clone();
            }
        }
    }
}

Add the one-time shared events table every aggregate uses:

rustavel make:migration create_events_table --create events

CREATE TABLE events (
    event_id TEXT PRIMARY KEY,
    aggregate_type TEXT NOT NULL,
    aggregate_id TEXT NOT NULL,
    version INTEGER NOT NULL,
    event_type TEXT NOT NULL,
    payload TEXT NOT NULL,
    recorded_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    UNIQUE (aggregate_type, aggregate_id, version)
);

make:aggregate also laid down a snapshots migration for you, the first time it ran. Reads via event_store::load_aggregate use it automatically: every 100 events (configurable with snapshot_every = N, or opt out with snapshot_every = 0), the current state gets cached there so later reads only replay the tail instead of the whole history.

4. A projection: PostSummary

rustavel make:projection PostSummary --aggregate Post

impl Projector<Post> for PostSummaryProjector {
    async fn project(&self, tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, envelope: &EventEnvelope<PostEvent>)
        -> Result<(), OrmError>
    {
        match &envelope.event {
            PostEvent::Created { title, body } | PostEvent::Edited { title, body } => {
                sqlx::query(
                    "INSERT INTO post_summaries (id, title, body) VALUES (?, ?, ?)
                     ON CONFLICT (id) DO UPDATE SET title = excluded.title, body = excluded.body",
                ).bind(envelope.aggregate_id.to_string()).bind(title).bind(body)
                 .execute(&mut **tx).await?;
            }
        }
        Ok(())
    }
}

5. A view: posts/show

rustavel make:view posts/show

{% extends "layout.html" %}
{% block content %}
    <h1>{{ title }}</h1>
    <p>{{ body }}</p>
    <div id="comments">
        {% for comment in comments %}{% include "comments/_item.html" %}{% endfor %}
    </div>
    <form hx-post="/posts/{{ post_id }}/comments" hx-target="#comments" hx-swap="beforeend">
        <textarea name="body" required></textarea>
        <button type="submit">Add comment</button>
    </form>
{% endblock %}

6. A background job: notifying on new comments

rustavel make:job SendCommentNotification

#[derive(Debug, Serialize, Deserialize, Job)]
#[job(queue = "default", retries = 3)]
pub struct SendCommentNotification {
    pub post_id: String,
    pub comment_id: i64,
}

impl SendCommentNotification {
    async fn handle_job(&self, db: &Db) -> Result<(), QueueError> {
        sqlx::query("INSERT INTO notifications (post_id, comment_id) VALUES (?, ?)")
            .bind(&self.post_id).bind(self.comment_id)
            .execute(db).await?;
        Ok(())
    }
}

Dispatch it from CommentController::store right after the comment is created — the response doesn't wait on it:

rustavel_queue::dispatch(&db, &SendCommentNotification {
    post_id: comment.post_id.clone(),
    comment_id: comment.id,
}).await?;

make:job laid down the one-time jobs/failed_jobs migration for you, same as make:aggregate does for snapshots. Register the job with a JobRegistry wherever the worker runs, and it'll be picked up, retried on failure, and moved to failed_jobs after MAX_ATTEMPTS.

7. Run it

rustavel migrate
rustavel serve

curl -X POST localhost:3000/posts \
  -H 'content-type: application/json' \
  -d '{"title":"Event sourcing in Rust","body":"Straight from the quickstart."}'

Open the returned id's page in a browser, submit the comment form, and watch devtools' network tab — the response is a small HTML fragment, not a full page. That's htmx doing the swap. In another terminal, rustavel queue:work --once drains the queued notification job.

What you just built

The complete, tested version of this app lives in examples/blog in the rustavel repo.