Dio & Lens — Caching and Events

Chapter 4 gave you Vista — a universal handle that works with any backend. But every call still hits the database. No caching, no reactivity, no way to route writes elsewhere.

Diorama (vantage-diorama) sits between your Vista and whatever consumes it. Three things it does:

1. Transparent caching. Keep a local copy of your data. Reads come from cache, not from the master — no matter how slow or rate-limited the backend is. A CSV file that takes 200ms to parse on every list_values() becomes instant after the first load.
2. Capability injection. A Vista backed by a CSV file can’t paginate, sort, or search server-side. Diorama caches the full dataset locally and answers those queries from cache — the consumer sees a Vista that can paginate, even though the underlying source can’t.
3. Custom write routing. Writes don’t have to go to the master. Route them to a Kafka topic, a queue, or a different database entirely. The cache updates immediately; persistence happens asynchronously.

The four words

Before the code, four terms you’ll see throughout:

• Vista — a single backend data source (chapter 4). Speaks whatever the backend supports.
• Lens — long-lived shared infrastructure: cache backend, callbacks, refresh config. Built once.
• Dio — a Vista bound to a Lens. Owns the cache table, write queue, event bus, and refresh task. Produced by lens.make_dio(vista).
• Scenery — a reactive view onto a Dio (tables, records, aggregates). The UI binds here. Covered in the next chapter.

The picture:

Vista → Lens.make_dio(vista) → Dio → facade Vista | Scenery
                                      ↑
                                  cache + events

Building a Lens

A Lens is built once and shared across every Dio you create from it. It holds the cache backend, lifecycle callbacks, and default policies:

#![allow(unused)]
fn main() {
use std::sync::Arc;
use std::time::Duration;
use vantage_diorama::Lens;

let lens = Arc::new(
    Lens::new()
        .cache_at("./cache.redb")
        .on_start(|dio| {
            let dio = dio.clone();
            async move {
                let rows = dio.master().list_values().await?;
                dio.cache().insert_values(rows).await?;
                Ok(())
            }
        })
        .refresh_every(Duration::from_secs(300))
        .build()?,
);
}

Three things happen here:

• .cache_at(path) — opens a redb file on disk. Each Dio claims a named table inside it. You can also pass a custom CacheBackend with .cache_source(backend) for in-memory or remote stores.
• .on_start(|dio| { ... }) — fires once when make_dio is called. The canonical pattern is seed the cache from the master: list everything, write it to cache. The dio.clone() inside the closure produces a 'static future — required because the callback outlives the borrow.
• .refresh_every(duration) — schedules periodic refresh. Combined with on_refresh (another callback), the Dio re-fetches from the master on a timer.

#![allow(unused)]
fn main() {
.on_start(|dio| {
    let dio = dio.clone();
    async move {
        // dio is now owned, safe to hold across await
    }
})
}

Creating a Dio

make_dio binds a Vista to the Lens. It opens the cache table, spawns the write worker and refresh task, fires on_start, and returns a Dio:

#![allow(unused)]
fn main() {
let products_dio = lens.make_dio(products_vista).await?;
}

That one call does all of this:

1. Opens (or creates) a cache table named after the Vista
2. Spawns a write worker that drains the write queue
3. Starts the refresh timer (if refresh_every + on_refresh are set)
4. Fires on_start — your callback seeds the cache
5. Returns the Dio, ready to use

You can create many Dios from one Lens — one per entity, each with its own cache table and write queue, all sharing the same cache file and callback configuration.

Reading from the cache

The Dio’s cache is a simple key-value store — id → Record<CborValue>. Reads come from cache, not from the master:

#![allow(unused)]
fn main() {
// List everything in cache
let rows = products_dio.cache().list_values().await?;

// Get one record by id
let product = products_dio.cache().get_value("7").await?;

// Count
let n = products_dio.cache().count().await?;
}

The cache is intentionally dumb — no conditions, no sort, no search. It stores rows; query planning lives on the Dio/Scenery layer.

#![allow(unused)]
fn main() {
let mut v = products_dio.vista();
v.add_condition_eq("category_id", 1.into())?;
let rows = v.list_values().await?;  // served from cache
}

Writing through the queue

Writes don’t hit the master directly. They go into the Dio’s write queue as a WriteOp:

#![allow(unused)]
fn main() {
use vantage_types::Record;

let record = Record::from_iter([
    ("name".into(), "Muffin".into()),
    ("price".into(), 175i64.into()),
]);

// Through the facade Vista
products_dio.vista().insert(&"muffin".to_string(), &record).await?;
}

Under the hood this enqueues a WriteOp::Insert. The write worker picks it up asynchronously and either calls your on_write callback or applies it to dio.master() directly.

The event bus

Every Dio has a broadcast channel. When data changes, events are published there:

#![allow(unused)]
fn main() {
use vantage_diorama::DioEvent;

let mut rx = products_dio.subscribe_events();

// In a spawned task
tokio::spawn(async move {
    while let Ok(event) = rx.recv().await {
        match event {
            DioEvent::RecordChanged { id } => {
                println!("updated: {}", id);
            }
            DioEvent::RecordInserted { id } => {
                println!("new: {}", id);
            }
            DioEvent::RecordRemoved { id } => {
                println!("deleted: {}", id);
            }
            DioEvent::Invalidated => {
                println!("full refresh happened");
            }
            _ => {}
        }
    }
});
}

Sceneries subscribe to this bus to react to changes. You can also subscribe directly — for logging, metrics, cache warming, or triggering side effects in any consumer.

Live updates: on_event

The event bus carries internal events (cache writes, refreshes). But what about changes happening on the master that you didn’t initiate — other users editing records, a database trigger firing, a webhook arriving?

on_event receives upstream ChangeEvent objects and reconciles them into the cache. Add it to the Lens alongside on_start:

#![allow(unused)]
fn main() {
let lens = Arc::new(
    Lens::new()
        .cache_at("./cache.redb")
        .on_start(|dio| {
            let dio = dio.clone();
            async move {
                let rows = dio.master().list_values().await?;
                dio.cache().insert_values(rows).await?;
                Ok(())
            }
        })
        .on_event(|dio, event| {
            let dio = dio.clone();
            async move {
                match event {
                    ChangeEvent::Updated { id, new: Some(row) } => {
                        dio.patched(id, row).await?;
                    }
                    ChangeEvent::Deleted { id } => {
                        dio.cache().delete_value(&id).await?;
                        dio.invalidate_record(id);
                    }
                    ChangeEvent::Invalidated => {
                        dio.invalidate_all();
                    }
                    _ => {}
                }
                Ok(())
            }
        })
        .build()?,
);
}

The typical wiring: a live stream from SurrealDB, a Kafka consumer, or a WebSocket listener feeds ChangeEvents into dio.handle_event(evt).await, which invokes your on_event callback. You write the row to cache and publish the matching DioEvent — any subscriber (Scenery, logger, metrics) picks it up.

Manual refresh

on_refresh fires on the timer set by refresh_every. You can also trigger it manually:

#![allow(unused)]
fn main() {
products_dio.refresh().await?;
}

This fires the on_refresh callback (if registered) and publishes Invalidated on the event bus. Sceneries re-read from cache after the refresh completes.

Callback summary

Putting it together

Everything in one place:

#![allow(unused)]
fn main() {
use std::sync::Arc;
use std::time::Duration;
use vantage_diorama::Lens;

// Build once per application
let lens = Arc::new(
    Lens::new()
        .cache_at("./cache.redb")
        .on_start(|dio| {
            let dio = dio.clone();
            async move {
                let rows = dio.master().list_values().await?;
                dio.cache().insert_values(rows).await?;
                Ok(())
            }
        })
        .refresh_every(Duration::from_secs(300))
        .build()?,
);

// One Dio per entity
let products = lens.make_dio(products_vista).await?;

// Read and write through the facade — cache is transparent
let mut v = products.vista();
let rows = v.list_values().await?;       // cache hit
v.insert(&"muffin".to_string(), &rec).await?;  // enqueued
}

What we covered