It builds with one warning.

Cargo.toml

@@ -5,9 +5,15 @@ edition = "2024"
 
 [dependencies]
 axum = { version = "0.8.9", features = ["json"] }
+axum-server = { version = "0.7", features = ["tls-rustls"] }
 chrono = { version = "0.4.45", features = ["serde"] }
+rcgen = { version = "0.14.10", features = ["pem", "x509-parser"] }
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "gzip"] }
 rusqlite = { version = "0.41.0", features = ["bundled"] }
+rustls = { version = "0.23.46", features = ["ring"] }
+rustls-pki-types = "1.0"
+rustls-pemfile = "2.1"
+
 serde = { version = "1.0.229", features = ["derive"] }
 serde_json = "1.0.154"
 tokio = { version = "1.54.1", features = ["macros", "rt-multi-thread"] }

src/db.rs

@@ -0,0 +1,83 @@
+use rusqlite::{params, Connection, OptionalExtension};
+use tracing::{info, warn};
+use uuid::Uuid;
+
+pub fn meta_get(db: &Connection, key: &str) -> rusqlite::Result<Option<String>> {
+    db.query_row("SELECT value FROM meta WHERE key = ?1", [key], |r| {
+        r.get::<_, String>(0)
+    })
+    .optional()
+}
+
+pub fn meta_set(db: &Connection, key: &str, value: &str) -> rusqlite::Result<()> {
+    db.execute(
+        "INSERT INTO meta (key, value) VALUES (?1, ?2)
+         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+        params![key, value],
+    )?;
+    Ok(())
+}
+
+pub fn init_db(db: &mut Connection) -> rusqlite::Result<()> {
+    db.execute_batch(
+        "CREATE TABLE IF NOT EXISTS meta (
+            key   TEXT PRIMARY KEY,
+            value TEXT NOT NULL
+        );",
+    )?;
+
+    let exists: bool = db.query_row(
+        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'records')",
+        [],
+        |r| r.get(0),
+    )?;
+
+    if exists {
+        let has_seq: bool = db.query_row(
+            "SELECT EXISTS(SELECT 1 FROM pragma_table_info('records') WHERE name = 'seq')",
+            [],
+            |r| r.get(0),
+        )?;
+
+        if !has_seq {
+            info!("Migrating records table to journal layout");
+            let tx = db.transaction()?;
+            tx.execute_batch(
+                "ALTER TABLE records RENAME TO records_old;
+                 CREATE TABLE records (
+                     seq  INTEGER PRIMARY KEY AUTOINCREMENT,
+                     uuid TEXT NOT NULL UNIQUE,
+                     date TEXT NOT NULL,
+                     data TEXT NOT NULL
+                 );
+                 INSERT INTO records (uuid, date, data)
+                     SELECT uuid, date, data FROM records_old ORDER BY rowid;
+                 DROP TABLE records_old;",
+            )?;
+            tx.commit()?;
+        }
+    }
+
+    db.execute_batch(
+        "CREATE TABLE IF NOT EXISTS records (
+            seq  INTEGER PRIMARY KEY AUTOINCREMENT,
+            uuid TEXT NOT NULL UNIQUE,
+            date TEXT NOT NULL,
+            data TEXT NOT NULL
+        );",
+    )?;
+
+    Ok(())
+}
+
+pub fn load_or_create_instance_id(db: &Connection) -> rusqlite::Result<Uuid> {
+    if let Some(existing) = meta_get(db, "instance_id")? {
+        if let Ok(id) = Uuid::parse_str(&existing) {
+            return Ok(id);
+        }
+        warn!("Stored instance_id is invalid; generating a new one");
+    }
+    let id = Uuid::new_v4();
+    meta_set(db, "instance_id", &id.to_string())?;
+    Ok(id)
+}

src/handlers.rs

@@ -0,0 +1,287 @@
+use axum::{
+    extract::{rejection::JsonRejection, State},
+    http::StatusCode,
+    Json,
+};
+use chrono::Utc;
+use crate::tls::{create_join_token, issue_leaf_cert, JoinClusterRequest, JoinClusterResponse};
+
+use rusqlite::params;
+use serde_json::Value;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use crate::{
+    schema::{load_config, validate_data, ValidationMode},
+    state::{api_error, db_error, internal_error, ApiError, ApiResponse, AppState, CreateRequest, Role},
+};
+
+/// Public endpoint: fetch the cluster's public CA certificate.
+pub async fn get_cluster_ca(
+    State(state): State<AppState>,
+) -> Result<Json<Value>, ApiError> {
+    let ca_pem = state
+        .ca_cert_pem
+        .as_deref()
+        .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "Cluster CA not available"))?;
+
+    Ok(Json(serde_json::json!({
+        "instance_id": state.instance_id,
+        "ca_pem": ca_pem
+    })))
+}
+
+/// Admin endpoint: mint a short-lived join token (valid for 10 minutes by default).
+pub async fn create_token_handler(
+    State(state): State<AppState>,
+    payload: Option<Json<Value>>,
+) -> Result<Json<Value>, ApiError> {
+    if state.role == Role::Replica {
+        return Err(api_error(StatusCode::FORBIDDEN, "Only Primary can generate join tokens"));
+    }
+
+    let ttl = payload
+        .and_then(|p| p.get("ttl_secs").and_then(|v| v.as_u64()))
+        .unwrap_or(600);
+
+    let token = create_join_token(ttl);
+
+    let mut tokens = state
+        .join_tokens
+        .write()
+        .map_err(|_| internal_error("tokens lock poisoned"))?;
+    tokens.insert(token.token.clone(), token.clone());
+
+    info!(token = %token.token, expires_at = %token.expires_at, "Issued join token");
+    Ok(Json(serde_json::json!(token)))
+}
+
+/// Dynamic join endpoint: exchanges a valid join token for an mTLS certificate.
+pub async fn join_cluster_handler(
+    State(state): State<AppState>,
+    Json(req): Json<JoinClusterRequest>,
+) -> Result<Json<JoinClusterResponse>, ApiError> {
+    if state.role == Role::Replica {
+        return Err(api_error(StatusCode::FORBIDDEN, "Only Primary can issue certificates"));
+    }
+
+    let ca_cert_pem = state
+        .ca_cert_pem
+        .as_ref()
+        .ok_or_else(|| internal_error("CA cert unavailable"))?;
+    let ca_key_pem = state
+        .ca_key_pem
+        .as_ref()
+        .ok_or_else(|| internal_error("CA key unavailable"))?;
+
+    // 1. Validate & burn the token
+    {
+        let mut tokens = state
+            .join_tokens
+            .write()
+            .map_err(|_| internal_error("tokens lock poisoned"))?;
+
+        let entry = tokens
+            .get_mut(&req.token)
+            .ok_or_else(|| api_error(StatusCode::UNAUTHORIZED, "Invalid join token"))?;
+
+        if entry.used {
+            return Err(api_error(StatusCode::UNAUTHORIZED, "Join token already used"));
+        }
+        if entry.expires_at < Utc::now() {
+            return Err(api_error(StatusCode::UNAUTHORIZED, "Join token expired"));
+        }
+
+        entry.used = true;
+    }
+
+    // 2. Issue leaf certificate signed by Cluster CA
+    let mut sans = req.san_list;
+    if !sans.iter().any(|s| s == "localhost") {
+        sans.push("localhost".into());
+    }
+    if !sans.iter().any(|s| s == "127.0.0.1") {
+        sans.push("127.0.0.1".into());
+    }
+
+    let (cert_pem, key_pem) = issue_leaf_cert(
+        ca_cert_pem,
+        ca_key_pem,
+        &req.node_id.to_string(),
+        sans,
+    )
+    .map_err(|e| internal_error(&format!("cert issue failed: {e}")))?;
+
+    info!(node_id = %req.node_id, "Issued client/server mTLS cert to replica");
+
+    Ok(Json(JoinClusterResponse {
+        ca_pem: ca_cert_pem.to_string(),
+        cert_pem,
+        key_pem,
+    }))
+}
+
+
+pub async fn get_info() -> Json<ApiResponse> {
+    info!(endpoint = "/api/now", "Request received");
+    Json(ApiResponse {
+        seq: None,
+        uuid: Uuid::new_v4(),
+        date: Utc::now().to_rfc3339(),
+        data: serde_json::json!({
+            "version": 1,
+            "message": "Hello from Rust!"
+        }),
+    })
+}
+
+pub async fn create_record(
+    State(state): State<AppState>,
+    payload: Result<Json<CreateRequest>, JsonRejection>,
+) -> Result<(StatusCode, Json<ApiResponse>), ApiError> {
+    info!(endpoint = "/api/records", "Create request received");
+
+   if state.role == Role::Replica {
+        return Err(api_error(
+            StatusCode::FORBIDDEN,
+            format!(
+                "This instance is a read-only replica; write to {}",
+                state.primary_url.as_deref().map(String::as_str).unwrap_or("the primary")
+            ),
+        ));
+    }
+
+    let Json(request) = payload.map_err(|e| {
+        warn!(error = %e, "Invalid JSON request");
+        api_error(StatusCode::BAD_REQUEST, "Invalid JSON request body")
+    })?;
+
+    let schema = state
+        .schema
+        .read()
+        .map_err(|_| internal_error("schema lock poisoned"))?
+        .clone();
+
+    if let Err(message) = validate_data(&request.data, &schema) {
+        warn!(reason = %message, "Schema validation failed");
+        return Err(api_error(StatusCode::BAD_REQUEST, message));
+    }
+
+    let mut record = ApiResponse {
+        seq: None,
+        uuid: Uuid::new_v4(),
+        date: Utc::now().to_rfc3339(),
+        data: request.data,
+    };
+
+    let data_json = serde_json::to_string(&record.data)
+        .map_err(|_| internal_error("serialize data"))?;
+
+    {
+        let db = state
+            .db
+            .lock()
+            .map_err(|_| internal_error("database lock poisoned"))?;
+
+        db.execute(
+            "INSERT INTO records (uuid, date, data) VALUES (?1, ?2, ?3)",
+            params![record.uuid.to_string(), record.date, data_json],
+        )
+        .map_err(|e| {
+            error!(error = %e, "Failed to insert record");
+            api_error(StatusCode::INTERNAL_SERVER_ERROR, "Failed to store record")
+        })?;
+
+        record.seq = Some(db.last_insert_rowid());
+    }
+
+    info!(uuid = %record.uuid, seq = ?record.seq, "Record stored");
+    Ok((StatusCode::CREATED, Json(record)))
+}
+
+pub async fn list_records(
+    State(state): State<AppState>,
+) -> Result<Json<Vec<ApiResponse>>, ApiError> {
+    info!(endpoint = "/api/records", "List request received");
+
+    let db = state
+        .db
+        .lock()
+        .map_err(|_| internal_error("database lock poisoned"))?;
+
+    let mut statement = db
+        .prepare("SELECT seq, uuid, date, data FROM records ORDER BY seq DESC LIMIT 100")
+        .map_err(|e| db_error("prepare list", e))?;
+
+    let rows = statement
+        .query_map([], |row| {
+            Ok((
+                row.get::<_, i64>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, String>(2)?,
+                row.get::<_, String>(3)?,
+            ))
+        })
+        .map_err(|e| db_error("query list", e))?;
+
+    let mut records = Vec::new();
+    for row in rows {
+        let (seq, uuid, date, data_json) = row.map_err(|e| db_error("read row", e))?;
+        let uuid = Uuid::parse_str(&uuid).map_err(|_| internal_error("invalid stored UUID"))?;
+        let data: Value =
+            serde_json::from_str(&data_json).map_err(|_| internal_error("invalid stored JSON"))?;
+
+        records.push(ApiResponse {
+            seq: Some(seq),
+            uuid,
+            date,
+            data,
+        });
+    }
+
+    info!(count = records.len(), "Records returned");
+    Ok(Json(records))
+}
+
+pub async fn get_config(State(state): State<AppState>) -> Result<Json<crate::schema::SchemaConfig>, ApiError> {
+    let schema = state
+        .schema
+        .read()
+        .map_err(|_| internal_error("schema lock poisoned"))?
+        .clone();
+
+    Ok(Json(schema))
+}
+
+pub async fn reload_config(State(state): State<AppState>) -> Result<Json<Value>, ApiError> {
+    info!(path = %state.config_path, "Configuration reload requested");
+
+    let new_schema = match load_config(&state.config_path) {
+        Ok(config) => config,
+        Err(message) => {
+            error!(reason = %message, "Configuration reload failed");
+            return Err(api_error(StatusCode::UNPROCESSABLE_ENTITY, message));
+        }
+    };
+
+    let mode = match new_schema.mode {
+        ValidationMode::Strict => "strict",
+        ValidationMode::Loose => "loose",
+    };
+    let field_count = new_schema.fields.len();
+
+    {
+        let mut active = state
+            .schema
+            .write()
+            .map_err(|_| internal_error("schema lock poisoned"))?;
+        *active = new_schema;
+    }
+
+    info!(mode, fields = field_count, "Configuration reloaded successfully");
+    Ok(Json(serde_json::json!({
+        "status": "ok",
+        "mode": mode,
+        "fields": field_count
+    })))
+}

src/replication.rs

@@ -0,0 +1,378 @@
+use axum::{
+    extract::{Query, State},
+    http::{header::AUTHORIZATION, HeaderMap, StatusCode},
+    Json,
+};
+
+use crate::tls::create_mtls_client;
+use chrono::Utc;
+use rusqlite::{params, Connection};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::{
+    sync::{Arc, Mutex},
+    time::Duration,
+};
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use crate::{
+    db::meta_set,
+    state::{api_error, db_error, internal_error, ApiError, AppState, SyncStatus},
+};
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Change {
+    pub seq: i64,
+    pub uuid: Uuid,
+    pub date: String,
+    pub data: Value,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ChangesResponse {
+    pub instance_id: Uuid,
+    pub head_seq: i64,
+    pub has_more: bool,
+    pub changes: Vec<Change>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct ChangesQuery {
+    pub after: Option<i64>,
+    pub limit: Option<i64>,
+}
+
+#[derive(Debug, Clone)]
+pub struct SyncConfig {
+    pub poll_interval: Duration,
+    pub max_backoff: Duration,
+    pub batch: i64,
+}
+
+enum SyncError {
+    Transient(String),
+    Fatal(String),
+}
+
+fn ct_eq(a: &[u8], b: &[u8]) -> bool {
+    a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
+}
+
+fn check_token(headers: &HeaderMap, expected: &Option<Arc<String>>) -> Result<(), ApiError> {
+    let Some(expected) = expected else {
+        return Err(api_error(
+            StatusCode::FORBIDDEN,
+            "Replication is disabled (no token configured)",
+        ));
+    };
+
+    let provided = headers
+        .get(AUTHORIZATION)
+        .and_then(|v| v.to_str().ok())
+        .and_then(|v| v.strip_prefix("Bearer "))
+        .unwrap_or("");
+
+    if !ct_eq(provided.as_bytes(), expected.as_bytes()) {
+        return Err(api_error(StatusCode::UNAUTHORIZED, "Invalid token"));
+    }
+    Ok(())
+}
+
+pub async fn replication_changes(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Query(query): Query<ChangesQuery>,
+) -> Result<Json<ChangesResponse>, ApiError> {
+    check_token(&headers, &state.repl_token)?;
+
+    let after = query.after.unwrap_or(0).max(0);
+    let limit = query.limit.unwrap_or(200).clamp(1, 1000);
+
+    let db = state
+        .db
+        .lock()
+        .map_err(|_| internal_error("database lock poisoned"))?;
+
+    let head_seq: i64 = db
+        .query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| {
+            r.get(0)
+        })
+        .map_err(|e| db_error("head seq", e))?;
+
+    let mut statement = db
+        .prepare(
+            "SELECT seq, uuid, date, data
+             FROM records
+             WHERE seq > ?1
+             ORDER BY seq ASC
+             LIMIT ?2",
+        )
+        .map_err(|e| db_error("prepare changes", e))?;
+
+    let rows = statement
+        .query_map(params![after, limit], |row| {
+            Ok((
+                row.get::<_, i64>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, String>(2)?,
+                row.get::<_, String>(3)?,
+            ))
+        })
+        .map_err(|e| db_error("query changes", e))?;
+
+    let mut changes = Vec::new();
+    for row in rows {
+        let (seq, uuid, date, data_json) = row.map_err(|e| db_error("read change", e))?;
+        changes.push(Change {
+            seq,
+            uuid: Uuid::parse_str(&uuid).map_err(|_| internal_error("invalid stored UUID"))?,
+            date,
+            data: serde_json::from_str(&data_json)
+                .map_err(|_| internal_error("invalid stored JSON"))?,
+        });
+    }
+
+    let last = changes.last().map(|c| c.seq).unwrap_or(after);
+
+    Ok(Json(ChangesResponse {
+        instance_id: state.instance_id,
+        head_seq,
+        has_more: last < head_seq,
+        changes,
+    }))
+}
+
+pub async fn replication_status(State(state): State<AppState>) -> Result<Json<Value>, ApiError> {
+    let status = state
+        .sync
+        .read()
+        .map_err(|_| internal_error("sync status lock poisoned"))?
+        .clone();
+
+    let local_head: i64 = {
+        let db = state
+            .db
+            .lock()
+            .map_err(|_| internal_error("database lock poisoned"))?;
+        db.query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| {
+            r.get(0)
+        })
+        .map_err(|e| db_error("head seq", e))?
+    };
+
+    let lag = status
+        .primary_head_seq
+        .map(|head| (head - status.last_applied_seq).max(0));
+
+    Ok(Json(serde_json::json!({
+        "instance_id": state.instance_id,
+        "role": state.role.as_str(),
+        "primary_url": state.primary_url.as_deref(),
+        "local_head_seq": local_head,
+        "lag_records": lag,
+        "sync": status,
+    })))
+}
+
+fn jitter(d: Duration) -> Duration {
+    let ms = d.as_millis() as u64;
+    let extra = (Uuid::new_v4().as_u128() as u64) % (ms / 2 + 1);
+    Duration::from_millis(ms / 2 + extra)
+}
+
+fn update_status(state: &AppState, f: impl FnOnce(&mut SyncStatus)) {
+    if let Ok(mut s) = state.sync.write() {
+        f(&mut s);
+    }
+}
+
+fn apply_batch(
+    db: &Arc<Mutex<Connection>>,
+    source: Uuid,
+    changes: &[Change],
+) -> Result<i64, String> {
+    let mut conn = db.lock().map_err(|_| "database lock poisoned".to_string())?;
+    let tx = conn.transaction().map_err(|e| e.to_string())?;
+
+    let mut max_seq = 0;
+    {
+        let mut insert = tx
+            .prepare_cached(
+                "INSERT OR IGNORE INTO records (seq, uuid, date, data) VALUES (?1, ?2, ?3, ?4)",
+            )
+            .map_err(|e| e.to_string())?;
+
+        for c in changes {
+            let data = serde_json::to_string(&c.data).map_err(|e| e.to_string())?;
+            insert
+                .execute(params![c.seq, c.uuid.to_string(), c.date, data])
+                .map_err(|e| e.to_string())?;
+            max_seq = max_seq.max(c.seq);
+        }
+    }
+
+    meta_set(&tx, "last_applied_seq", &max_seq.to_string()).map_err(|e| e.to_string())?;
+    meta_set(&tx, "source_instance_id", &source.to_string()).map_err(|e| e.to_string())?;
+
+    tx.commit().map_err(|e| e.to_string())?;
+    Ok(max_seq)
+}
+
+async fn sync_once(
+    client: &reqwest::Client,
+    state: &AppState,
+    primary_url: &str,
+    token: &str,
+    batch: i64,
+) -> Result<bool, SyncError> {
+    let (last, known_source) = {
+        let s = state
+            .sync
+            .read()
+            .map_err(|_| SyncError::Fatal("sync status lock poisoned".into()))?;
+        (s.last_applied_seq, s.source_instance_id)
+    };
+
+    let url = format!("{}/api/replication/changes", primary_url.trim_end_matches('/'));
+
+    let resp = client
+        .get(&url)
+        .bearer_auth(token)
+        .query(&[("after", last), ("limit", batch)])
+        .send()
+        .await
+        .map_err(|e| SyncError::Transient(format!("request failed: {e}")))?;
+
+    if !resp.status().is_success() {
+        return Err(SyncError::Transient(format!(
+            "primary returned {}",
+            resp.status()
+        )));
+    }
+
+    let body: ChangesResponse = resp
+        .json()
+        .await
+        .map_err(|e| SyncError::Transient(format!("bad response body: {e}")))?;
+
+    if body.instance_id == state.instance_id {
+        return Err(SyncError::Fatal(
+            "source has the same instance_id as this replica (cloned database?)".into(),
+        ));
+    }
+    if let Some(known) = known_source {
+        if known != body.instance_id {
+            return Err(SyncError::Fatal(format!(
+                "source instance changed: expected {known}, got {}",
+                body.instance_id
+            )));
+        }
+    }
+    if body.head_seq < last {
+        return Err(SyncError::Fatal(format!(
+            "source head_seq {} is behind our cursor {last} (source restored from older backup?)",
+            body.head_seq
+        )));
+    }
+
+    if body.changes.is_empty() {
+        update_status(state, |s| {
+            s.state = "idle".into();
+            s.primary_head_seq = Some(body.head_seq);
+            s.last_success = Some(Utc::now().to_rfc3339());
+            s.last_error = None;
+            s.source_instance_id = Some(body.instance_id);
+        });
+        return Ok(false);
+    }
+
+    let mut prev = last;
+    for c in &body.changes {
+        if c.seq <= prev {
+            return Err(SyncError::Transient(format!(
+                "non-monotonic seq {} after {prev}",
+                c.seq
+            )));
+        }
+        prev = c.seq;
+    }
+
+    let count = body.changes.len();
+    let db = state.db.clone();
+    let source = body.instance_id;
+    let changes = body.changes;
+
+    let max_seq = tokio::task::spawn_blocking(move || apply_batch(&db, source, &changes))
+        .await
+        .map_err(|e| SyncError::Transient(format!("apply task failed: {e}")))?
+        .map_err(|e| SyncError::Transient(format!("apply failed: {e}")))?;
+
+    update_status(state, |s| {
+        s.state = if body.has_more { "syncing" } else { "idle" }.into();
+        s.last_applied_seq = max_seq;
+        s.source_instance_id = Some(source);
+        s.primary_head_seq = Some(body.head_seq);
+        s.last_success = Some(Utc::now().to_rfc3339());
+        s.last_error = None;
+    });
+
+    info!(
+        applied = count,
+        last_applied_seq = max_seq,
+        head = body.head_seq,
+        "Replication batch applied"
+    );
+
+    Ok(body.has_more)
+}
+
+pub async fn replica_loop(
+    state: AppState,
+    primary_url: String,
+    token: String,
+    cfg: SyncConfig,
+    ca_pem: String,
+    cert_pem: String,
+    key_pem: String,
+) {
+    let client = match create_mtls_client(&ca_pem, &cert_pem, &key_pem) {
+        Ok(c) => c,
+        Err(e) => {
+            error!(error = %e, "Cannot build mTLS HTTP client; replication halted");
+            return;
+        }
+    };
+
+    let mut backoff = Duration::from_secs(1);
+    info!(primary = %primary_url, "Replica mTLS sync loop started");
+
+    loop {
+        match sync_once(&client, &state, &primary_url, &token, cfg.batch).await {
+            Ok(more) => {
+                backoff = Duration::from_secs(1);
+                if !more {
+                    tokio::time::sleep(jitter(cfg.poll_interval)).await;
+                }
+            }
+            Err(SyncError::Transient(message)) => {
+                let wait = jitter(backoff);
+                warn!(reason = %message, retry_in = ?wait, "Replication pull failed");
+                update_status(&state, |s| {
+                    s.state = "retrying".into();
+                    s.last_error = Some(message);
+                });
+                tokio::time::sleep(wait).await;
+                backoff = (backoff * 2).min(cfg.max_backoff);
+            }
+            Err(SyncError::Fatal(message)) => {
+                error!(reason = %message, "Replication halted; operator action required");
+                update_status(&state, |s| {
+                    s.state = "halted".into();
+                    s.last_error = Some(message);
+                });
+                return;
+            }
+        }
+    }
+}

src/schema.rs

@@ -0,0 +1,105 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::{collections::HashMap, fs};
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum ValidationMode {
+    Strict,
+    Loose,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct FieldSpec {
+    #[serde(rename = "type")]
+    pub data_type: DataType,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub value: Option<Value>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum DataType {
+    String,
+    Integer,
+    Number,
+    Boolean,
+    Object,
+    Array,
+    Null,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SchemaConfig {
+    pub mode: ValidationMode,
+    pub fields: HashMap<String, FieldSpec>,
+}
+
+pub fn matches_type(value: &Value, expected: &DataType) -> bool {
+    match expected {
+        DataType::String => value.is_string(),
+        DataType::Integer => value.as_i64().is_some() || value.as_u64().is_some(),
+        DataType::Number => value.is_number(),
+        DataType::Boolean => value.is_boolean(),
+        DataType::Object => value.is_object(),
+        DataType::Array => value.is_array(),
+        DataType::Null => value.is_null(),
+    }
+}
+
+pub fn type_name(data_type: &DataType) -> &'static str {
+    match data_type {
+        DataType::String => "string",
+        DataType::Integer => "integer",
+        DataType::Number => "number",
+        DataType::Boolean => "boolean",
+        DataType::Object => "object",
+        DataType::Array => "array",
+        DataType::Null => "null",
+    }
+}
+
+pub fn validate_data(data: &Value, schema: &SchemaConfig) -> Result<(), String> {
+    let object = data.as_object().ok_or("data must be a JSON object")?;
+
+    for (key, spec) in &schema.fields {
+        let value = object
+            .get(key)
+            .ok_or_else(|| format!("Missing required field: {key}"))?;
+
+        if !matches_type(value, &spec.data_type) {
+            return Err(format!(
+                "Field '{key}' must be of type {}",
+                type_name(&spec.data_type)
+            ));
+        }
+
+        if let Some(expected) = &spec.value {
+            if value != expected {
+                return Err(format!("Field '{key}' must equal {expected}"));
+            }
+        }
+    }
+
+    if matches!(schema.mode, ValidationMode::Strict) {
+        for key in object.keys() {
+            if !schema.fields.contains_key(key) {
+                return Err(format!("Unknown field: {key}"));
+            }
+        }
+    }
+
+    Ok(())
+}
+
+pub fn load_config(path: &str) -> Result<SchemaConfig, String> {
+    let contents = fs::read_to_string(path).map_err(|e| format!("Cannot read {path}: {e}"))?;
+    let config: SchemaConfig =
+        serde_json::from_str(&contents).map_err(|e| format!("Invalid configuration JSON: {e}"))?;
+
+    if config.fields.is_empty() {
+        return Err("Configuration must define at least one field".into());
+    }
+
+    Ok(config)
+}

src/state.rs

@@ -0,0 +1,84 @@
+use axum::http::StatusCode;
+use rusqlite::Connection;
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex, RwLock},
+};
+use tracing::error;
+use uuid::Uuid;
+
+use crate::{schema::SchemaConfig, tls::JoinToken};
+
+#[derive(Clone)]
+pub struct AppState {
+    pub db: Arc<Mutex<Connection>>,
+    pub schema: Arc<RwLock<SchemaConfig>>,
+    pub config_path: Arc<String>,
+    pub instance_id: Uuid,
+    pub role: Role,
+    pub primary_url: Option<Arc<String>>,
+    pub repl_token: Option<Arc<String>>,
+    pub sync: Arc<RwLock<SyncStatus>>,
+
+    // mTLS & Discovery additions:
+    pub ca_cert_pem: Option<Arc<String>>,
+    pub ca_key_pem: Option<Arc<String>>,
+    pub join_tokens: Arc<RwLock<HashMap<String, JoinToken>>>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Role {
+    Primary,
+    Replica,
+}
+
+impl Role {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Role::Primary => "primary",
+            Role::Replica => "replica",
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct SyncStatus {
+    pub state: String,
+    pub last_applied_seq: i64,
+    pub source_instance_id: Option<Uuid>,
+    pub primary_head_seq: Option<i64>,
+    pub last_success: Option<String>,
+    pub last_error: Option<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ApiResponse {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub seq: Option<i64>,
+    pub uuid: Uuid,
+    pub date: String,
+    pub data: Value,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct CreateRequest {
+    pub data: Value,
+}
+
+pub type ApiError = (StatusCode, String);
+
+pub fn api_error(status: StatusCode, message: impl Into<String>) -> ApiError {
+    (status, message.into())
+}
+
+pub fn internal_error(context: &str) -> ApiError {
+    error!(context, "Internal server error");
+    api_error(StatusCode::INTERNAL_SERVER_ERROR, "Internal server error")
+}
+
+pub fn db_error(context: &str, e: rusqlite::Error) -> ApiError {
+    error!(error = %e, context, "Database error");
+    api_error(StatusCode::INTERNAL_SERVER_ERROR, "Database error")
+}

src/tls.rs

@@ -0,0 +1,219 @@
+use std::sync::Arc;
+use chrono::{DateTime, Duration as ChronoDuration, Utc};
+use rcgen::{
+    BasicConstraints, CertificateParams, DnType, ExtendedKeyUsagePurpose, IsCa,
+    Issuer, KeyPair, KeyUsagePurpose,
+};
+use rusqlite::Connection;
+use rustls::server::WebPkiClientVerifier;
+use rustls::RootCertStore;
+use rustls_pki_types::{CertificateDer, PrivateKeyDer};
+use serde::{Deserialize, Serialize};
+use tracing::info;
+use uuid::Uuid;
+
+use crate::db::{meta_set};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JoinToken {
+    pub token: String,
+    pub expires_at: DateTime<Utc>,
+    pub used: bool,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct JoinClusterRequest {
+    pub token: String,
+    pub node_id: Uuid,
+    #[serde(default)]
+    pub san_list: Vec<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct JoinClusterResponse {
+    pub ca_pem: String,
+    pub cert_pem: String,
+    pub key_pem: String,
+}
+
+// ---------------------------------------------------------------------------
+// Certificate Generation
+// ---------------------------------------------------------------------------
+
+pub fn generate_ca() -> Result<(String, String), String> {
+    let ca_key = KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256)
+        .map_err(|e| format!("CA key error: {e}"))?;
+
+    let mut ca_params = CertificateParams::new(vec![])
+        .map_err(|e| format!("CA params error: {e}"))?;
+    ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
+    ca_params.distinguished_name.push(DnType::CommonName, "Cluster Root CA");
+    ca_params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::CrlSign];
+
+    let ca_cert = ca_params
+        .self_signed(&ca_key)
+        .map_err(|e| format!("CA self-sign error: {e}"))?;
+
+    Ok((ca_cert.pem(), ca_key.serialize_pem()))
+}
+
+pub fn issue_leaf_cert(
+    ca_cert_pem: &str,
+    ca_key_pem: &str,
+    common_name: &str,
+    san_list: Vec<String>,
+) -> Result<(String, String), String> {
+    let ca_key = KeyPair::from_pem(ca_key_pem)
+        .map_err(|e| format!("Failed to parse CA private key: {e}"))?;
+
+    // In rcgen 0.14+, from_ca_cert_pem belongs to Issuer:
+    let issuer = Issuer::from_ca_cert_pem(ca_cert_pem, &ca_key)
+        .map_err(|e| format!("Failed to parse CA cert/create issuer: {e}"))?;
+
+    let leaf_key = KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256)
+        .map_err(|e| format!("Leaf key error: {e}"))?;
+
+    let mut leaf_params = CertificateParams::new(san_list)
+        .map_err(|e| format!("Leaf params error: {e}"))?;
+    leaf_params.distinguished_name.push(DnType::CommonName, common_name);
+    leaf_params.use_authority_key_identifier_extension = true;
+    leaf_params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
+    leaf_params.extended_key_usages = vec![
+        ExtendedKeyUsagePurpose::ServerAuth,
+        ExtendedKeyUsagePurpose::ClientAuth,
+    ];
+
+    let leaf_cert = leaf_params
+        .signed_by(&leaf_key, &issuer)
+        .map_err(|e| format!("Leaf sign error: {e}"))?;
+
+    Ok((leaf_cert.pem(), leaf_key.serialize_pem()))
+}
+
+// ---------------------------------------------------------------------------
+// Rustls Server & Reqwest Client Configuration
+// ---------------------------------------------------------------------------
+
+fn parse_certs(pem: &str) -> Result<Vec<CertificateDer<'static>>, String> {
+    let mut reader = pem.as_bytes();
+    let certs: Result<Vec<_>, _> = rustls_pemfile::certs(&mut reader).collect();
+    certs.map_err(|e| format!("failed to parse certs: {e}"))
+}
+
+fn parse_key(pem: &str) -> Result<PrivateKeyDer<'static>, String> {
+    let mut reader = pem.as_bytes();
+    rustls_pemfile::private_key(&mut reader)
+        .map_err(|e| format!("failed to parse private key: {e}"))?
+        .ok_or_else(|| "no private key found in pem".to_string())
+}
+
+pub fn create_mtls_server_config(
+    ca_pem: &str,
+    server_cert_pem: &str,
+    server_key_pem: &str,
+) -> Result<rustls::ServerConfig, String> {
+    let mut root_store = RootCertStore::empty();
+    for cert in parse_certs(ca_pem)? {
+        root_store
+            .add(cert)
+            .map_err(|e| format!("failed to add root cert: {e}"))?;
+    }
+
+    // allow_unauthenticated permits anonymous TLS connections (e.g., initial join requests)
+    // while enforcing valid CA signatures on any presented client certificates.
+    let client_verifier = WebPkiClientVerifier::builder(Arc::new(root_store))
+        .allow_unauthenticated()
+        .build()
+        .map_err(|e| format!("failed to build client verifier: {e}"))?;
+
+    let cert_chain = parse_certs(server_cert_pem)?;
+    let key = parse_key(server_key_pem)?;
+
+    let mut config = rustls::ServerConfig::builder()
+        .with_client_cert_verifier(client_verifier)
+        .with_single_cert(cert_chain, key)
+        .map_err(|e| format!("failed to build ServerConfig: {e}"))?;
+
+    config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
+    Ok(config)
+}
+
+pub fn create_mtls_client(
+    ca_pem: &str,
+    client_cert_pem: &str,
+    client_key_pem: &str,
+) -> Result<reqwest::Client, String> {
+    let root_cert = reqwest::Certificate::from_pem(ca_pem.as_bytes())
+        .map_err(|e| format!("failed to load CA pem for client: {e}"))?;
+
+    let identity_pem = format!("{client_cert_pem}\n{client_key_pem}");
+    let identity = reqwest::Identity::from_pem(identity_pem.as_bytes())
+        .map_err(|e| format!("failed to create client identity: {e}"))?;
+
+    reqwest::Client::builder()
+        .use_rustls_tls()
+        .add_root_certificate(root_cert)
+        .identity(identity)
+        .connect_timeout(std::time::Duration::from_secs(10))
+        .timeout(std::time::Duration::from_secs(60))
+        .build()
+        .map_err(|e| format!("failed to build reqwest client: {e}"))
+}
+
+// ---------------------------------------------------------------------------
+// Dynamic Enrollment / Join Helpers
+// ---------------------------------------------------------------------------
+
+pub fn create_join_token(ttl_secs: u64) -> JoinToken {
+    JoinToken {
+        token: Uuid::new_v4().to_string().replace('-', ""),
+        expires_at: Utc::now() + ChronoDuration::seconds(ttl_secs as i64),
+        used: false,
+    }
+}
+
+pub async fn bootstrap_replica_join(
+    primary_url: &str,
+    token: &str,
+    node_id: Uuid,
+    db: &Connection,
+) -> Result<(), String> {
+    let base = primary_url.trim_end_matches('/');
+
+    // 1. Initial bootstrap client without mTLS client cert
+    let bootstrap_client = reqwest::Client::builder()
+        .danger_accept_invalid_certs(true)
+        .build()
+        .map_err(|e| format!("bootstrap client build failed: {e}"))?;
+
+    // 2. Request enrollment
+    let join_req = JoinClusterRequest {
+        token: token.to_string(),
+        node_id,
+        san_list: vec!["localhost".into(), "127.0.0.1".into()],
+    };
+
+    let resp = bootstrap_client
+        .post(format!("{base}/api/cluster/join"))
+        .json(&join_req)
+        .send()
+        .await
+        .map_err(|e| format!("join request failed: {e}"))?;
+
+    if !resp.status().is_success() {
+        return Err(format!("join rejected with HTTP {}", resp.status()));
+    }
+
+    let payload: JoinClusterResponse = resp
+        .json()
+        .await
+        .map_err(|e| format!("failed to decode join response: {e}"))?;
+
+    // 3. Persist credentials locally
+    meta_set(db, "tls_ca_pem", &payload.ca_pem).map_err(|e| e.to_string())?;
+    meta_set(db, "tls_cert_pem", &payload.cert_pem).map_err(|e| e.to_string())?;
+    meta_set(db, "tls_key_pem", &payload.key_pem).map_err(|e| e.to_string())?;
+
+    info!("Replica successfully joined cluster and persisted mTLS certificates");
+    Ok(())
+}