cellz 0.2.0

SQLite-per-session, event-sourced state server for AI agents
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

use std::sync::Arc;
use std::time::Duration;

use cellz::cell::CellManager;
use cellz::config::Config;
use cellz::error::Result;
use cellz::server::create_router;
use cellz::storage::LocalBlobStore;
use tracing::info;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::registry()
        .with(
            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
        )
        .with(tracing_subscriber::fmt::layer())
        .init();

    let config = Config::default();
    let storage: Arc<dyn cellz::storage::BlobStore> = match config.storage_backend.as_str() {
        "s3" => {
            #[cfg(feature = "s3")]
            {
                let endpoint = config.s3_endpoint.as_deref().unwrap_or_default();
                let bucket = config.s3_bucket.as_deref().unwrap_or_default();
                let access_key = config.s3_access_key_id.as_deref().unwrap_or_default();
                let secret_key = config.s3_secret_access_key.as_deref().unwrap_or_default();
                let region = config.s3_region.as_deref();

                info!(
                    "📦 Initializing S3/R2 BlobStore for bucket '{}' at '{}'",
                    bucket, endpoint
                );
                Arc::new(cellz::storage::S3BlobStore::new(
                    endpoint, bucket, access_key, secret_key, region,
                )?)
            }
            #[cfg(not(feature = "s3"))]
            {
                return Err(cellz::error::Error::Config(
                    "CELLZ_STORAGE_BACKEND=s3 requires the `s3` cargo feature (rebuild with `--features s3`)"
                        .into(),
                ));
            }
        }
        _ => {
            info!(
                "📂 Initializing Local Filesystem BlobStore at {:?}",
                config.storage_dir
            );
            Arc::new(LocalBlobStore::new(&config.storage_dir))
        }
    };
    let manager = Arc::new(CellManager::new(
        &config.data_dir,
        storage,
        config.lease_ttl_secs,
    ));

    // Background task to periodically renew leases for all active in-memory cells
    let lease_manager = Arc::clone(&manager);
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(20));
        loop {
            interval.tick().await;
            lease_manager.renew_active_leases().await;
        }
    });

    // Background task to evict idle cells from memory
    let idle_manager = Arc::clone(&manager);
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(60));
        loop {
            interval.tick().await;
            idle_manager
                .evict_idle_cells(Duration::from_secs(300))
                .await;
        }
    });

    let app = create_router(manager);

    let addr = format!("{}:{}", config.host, config.port);
    info!("🚀 cellz daemon running on http://{}", addr);

    let listener = tokio::net::TcpListener::bind(&addr)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to bind to {}: {}", addr, e))?;

    axum::serve(listener, app)
        .await
        .map_err(|e| anyhow::anyhow!("Server error: {}", e))?;

    Ok(())
}