start, reverse guard, cli-frontend for server and client

oncp/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "oncp"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+tokio.workspace = true
+rusqlite.workspace = true
+anyhow.workspace = true
+thiserror.workspace = true
+tracing.workspace = true
+uuid.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+chrono.workspace = true
+ostp = { path = "../ostp" }

oncp/src/billing.rs

@@ -0,0 +1,163 @@
+//! User billing and quota management
+
+use chrono::{DateTime, Utc};
+use std::sync::Mutex;
+use thiserror::Error;
+use uuid::Uuid;
+
+#[derive(Error, Debug)]
+pub enum BillingError {
+    #[error("user not found: {0}")]
+    UserNotFound(Uuid),
+    #[error("subscription expired")]
+    Expired,
+    #[error("bandwidth quota exceeded")]
+    QuotaExceeded,
+    #[error("database error: {0}")]
+    Database(#[from] rusqlite::Error),
+    #[error("lock error")]
+    LockError,
+}
+
+/// User account with billing info
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct User {
+    pub uuid: Uuid,
+    pub bandwidth_quota: u64,      // bytes allowed
+    pub bandwidth_used: u64,       // bytes used
+    pub expires_at: DateTime<Utc>, // subscription expiry
+    pub active: bool,
+}
+
+impl User {
+    pub fn new(quota_gb: u64, valid_days: i64) -> Self {
+        Self {
+            uuid: Uuid::new_v4(),
+            bandwidth_quota: quota_gb * 1024 * 1024 * 1024,
+            bandwidth_used: 0,
+            expires_at: Utc::now() + chrono::Duration::days(valid_days),
+            active: true,
+        }
+    }
+
+    pub fn is_valid(&self) -> bool {
+        self.active && Utc::now() < self.expires_at && self.bandwidth_used < self.bandwidth_quota
+    }
+
+    pub fn remaining_quota(&self) -> u64 {
+        self.bandwidth_quota.saturating_sub(self.bandwidth_used)
+    }
+}
+
+/// Abstract user registry trait
+pub trait UserRegistry: Send + Sync {
+    fn get_user(&self, uuid: &Uuid) -> Result<Option<User>, BillingError>;
+    fn create_user(&self, user: &User) -> Result<(), BillingError>;
+    fn update_bandwidth(&self, uuid: &Uuid, bytes: u64) -> Result<(), BillingError>;
+    fn validate_user(&self, uuid: &Uuid) -> Result<bool, BillingError>;
+}
+
+/// SQLite implementation (thread-safe via Mutex)
+pub struct SqliteRegistry {
+    conn: Mutex<rusqlite::Connection>,
+}
+
+impl SqliteRegistry {
+    pub fn new(path: &str) -> Result<Self, BillingError> {
+        let conn = rusqlite::Connection::open(path)?;
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS users (
+                uuid TEXT PRIMARY KEY,
+                bandwidth_quota INTEGER NOT NULL,
+                bandwidth_used INTEGER NOT NULL DEFAULT 0,
+                expires_at TEXT NOT NULL,
+                active INTEGER NOT NULL DEFAULT 1
+            )",
+            [],
+        )?;
+        Ok(Self { conn: Mutex::new(conn) })
+    }
+
+    pub fn in_memory() -> Result<Self, BillingError> {
+        Self::new(":memory:")
+    }
+}
+
+impl UserRegistry for SqliteRegistry {
+    fn get_user(&self, uuid: &Uuid) -> Result<Option<User>, BillingError> {
+        let conn = self.conn.lock().map_err(|_| BillingError::LockError)?;
+        let mut stmt = conn
+            .prepare("SELECT uuid, bandwidth_quota, bandwidth_used, expires_at, active FROM users WHERE uuid = ?")?;
+
+        let mut rows = stmt.query([uuid.to_string()])?;
+
+        if let Some(row) = rows.next()? {
+            let uuid_str: String = row.get(0)?;
+            let expires_str: String = row.get(3)?;
+            Ok(Some(User {
+                uuid: Uuid::parse_str(&uuid_str).unwrap(),
+                bandwidth_quota: row.get(1)?,
+                bandwidth_used: row.get(2)?,
+                expires_at: DateTime::parse_from_rfc3339(&expires_str)
+                    .unwrap()
+                    .with_timezone(&Utc),
+                active: row.get::<_, i32>(4)? == 1,
+            }))
+        } else {
+            Ok(None)
+        }
+    }
+
+    fn create_user(&self, user: &User) -> Result<(), BillingError> {
+        let conn = self.conn.lock().map_err(|_| BillingError::LockError)?;
+        conn.execute(
+            "INSERT INTO users (uuid, bandwidth_quota, bandwidth_used, expires_at, active) VALUES (?, ?, ?, ?, ?)",
+            (
+                user.uuid.to_string(),
+                user.bandwidth_quota,
+                user.bandwidth_used,
+                user.expires_at.to_rfc3339(),
+                if user.active { 1 } else { 0 },
+            ),
+        )?;
+        Ok(())
+    }
+
+    fn update_bandwidth(&self, uuid: &Uuid, bytes: u64) -> Result<(), BillingError> {
+        let conn = self.conn.lock().map_err(|_| BillingError::LockError)?;
+        let updated = conn.execute(
+            "UPDATE users SET bandwidth_used = bandwidth_used + ? WHERE uuid = ?",
+            (bytes, uuid.to_string()),
+        )?;
+        if updated == 0 {
+            return Err(BillingError::UserNotFound(*uuid));
+        }
+        Ok(())
+    }
+
+    fn validate_user(&self, uuid: &Uuid) -> Result<bool, BillingError> {
+        match self.get_user(uuid)? {
+            Some(user) => Ok(user.is_valid()),
+            None => Ok(false),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_user_lifecycle() {
+        let reg = SqliteRegistry::in_memory().unwrap();
+        let user = User::new(100, 30); // 100GB, 30 days
+        let uuid = user.uuid;
+
+        reg.create_user(&user).unwrap();
+        assert!(reg.validate_user(&uuid).unwrap());
+
+        reg.update_bandwidth(&uuid, 1024).unwrap();
+        let updated = reg.get_user(&uuid).unwrap().unwrap();
+        assert_eq!(updated.bandwidth_used, 1024);
+    }
+}

oncp/src/lib.rs

@@ -0,0 +1,5 @@
+pub mod billing;
+pub mod session;
+
+pub use billing::{BillingError, SqliteRegistry, User, UserRegistry};
+pub use session::{Session, SessionManager};

oncp/src/session.rs

@@ -0,0 +1,109 @@
+//! Session management and heartbeat
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+use tokio::sync::RwLock;
+use uuid::Uuid;
+
+/// Session state
+#[derive(Debug, Clone)]
+pub struct Session {
+    pub user_id: Uuid,
+    pub session_id: Uuid,
+    pub created_at: Instant,
+    pub last_heartbeat: Instant,
+    pub bytes_tx: u64,
+    pub bytes_rx: u64,
+    pub sni: String,
+}
+
+impl Session {
+    pub fn new(user_id: Uuid, sni: String) -> Self {
+        let now = Instant::now();
+        Self {
+            user_id,
+            session_id: Uuid::new_v4(),
+            created_at: now,
+            last_heartbeat: now,
+            bytes_tx: 0,
+            bytes_rx: 0,
+            sni,
+        }
+    }
+
+    pub fn heartbeat(&mut self) {
+        self.last_heartbeat = Instant::now();
+    }
+
+    pub fn is_alive(&self, timeout: Duration) -> bool {
+        self.last_heartbeat.elapsed() < timeout
+    }
+}
+
+/// Session manager
+pub struct SessionManager {
+    sessions: Arc<RwLock<HashMap<Uuid, Session>>>,
+    heartbeat_timeout: Duration,
+}
+
+impl SessionManager {
+    pub fn new(heartbeat_timeout_secs: u64) -> Self {
+        Self {
+            sessions: Arc::new(RwLock::new(HashMap::new())),
+            heartbeat_timeout: Duration::from_secs(heartbeat_timeout_secs),
+        }
+    }
+
+    pub async fn create_session(&self, user_id: Uuid, sni: String) -> Uuid {
+        let session = Session::new(user_id, sni);
+        let session_id = session.session_id;
+        self.sessions.write().await.insert(session_id, session);
+        session_id
+    }
+
+    pub async fn heartbeat(&self, session_id: &Uuid) -> bool {
+        if let Some(session) = self.sessions.write().await.get_mut(session_id) {
+            session.heartbeat();
+            true
+        } else {
+            false
+        }
+    }
+
+    pub async fn get_session(&self, session_id: &Uuid) -> Option<Session> {
+        self.sessions.read().await.get(session_id).cloned()
+    }
+
+    pub async fn remove_session(&self, session_id: &Uuid) -> Option<Session> {
+        self.sessions.write().await.remove(session_id)
+    }
+
+    pub async fn cleanup_stale(&self) -> Vec<Uuid> {
+        let mut sessions = self.sessions.write().await;
+        let timeout = self.heartbeat_timeout;
+        let stale: Vec<Uuid> = sessions
+            .iter()
+            .filter(|(_, s)| !s.is_alive(timeout))
+            .map(|(id, _)| *id)
+            .collect();
+
+        for id in &stale {
+            sessions.remove(id);
+        }
+        stale
+    }
+
+    pub async fn update_traffic(&self, session_id: &Uuid, tx: u64, rx: u64) {
+        if let Some(session) = self.sessions.write().await.get_mut(session_id) {
+            session.bytes_tx += tx;
+            session.bytes_rx += rx;
+        }
+    }
+}
+
+impl Default for SessionManager {
+    fn default() -> Self {
+        Self::new(60)
+    }
+}