[TSM.ID].[11031972] PXE : Platform X Ecosystem I [118 Module -LIVE-]
This commit is contained in:
TSM.ID
@@ -0,0 +1,16 @@
|
||||
# [TSM.ID].[11031972] -- All Rights Reserved. Proprietary & Confidential.
|
||||
[package]
|
||||
name = "xcu-billing-matrix"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
|
||||
[dependencies]
|
||||
tokio = { version = "1.0", features = ["full"] }
|
||||
axum = "0.7"
|
||||
serde = { version = "1.0", features = ["derive"] }
|
||||
serde_json = "1.0"
|
||||
duckdb = { version = "1.1.0", features = ["bundled"] }
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = "0.3"
|
||||
tower-http = { version = "0.5", features = ["cors"] }
|
||||
redis = { version = "0.24.0", features = ["tokio-comp"] }
|
||||
@@ -0,0 +1,260 @@
|
||||
// [TSM.ID].[11031972] -- All Rights Reserved. Proprietary & Confidential.
|
||||
use axum::{
|
||||
routing::{get, post},
|
||||
Router, Json, extract::Path,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use duckdb::{params, Connection};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use tracing::{info, warn};
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone)]
|
||||
struct TenantBilling {
|
||||
tenant_id: String,
|
||||
name: String,
|
||||
role: String,
|
||||
packages: Vec<String>,
|
||||
quota_limit_gb: f64,
|
||||
used_gb: f64,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct UsageReport {
|
||||
tenant_id: String,
|
||||
bytes_used: u64,
|
||||
}
|
||||
|
||||
// Global thread-safe connection to DuckDB
|
||||
// In high-concurrency production, use an r2d2 pool, but Arc<Mutex<Connection>> works for MVP
|
||||
#[allow(dead_code)]
|
||||
struct AppState {
|
||||
db: Arc<Mutex<Connection>>,
|
||||
redis_client: Option<redis::Client>,
|
||||
}
|
||||
|
||||
async fn get_tenant_billing(
|
||||
Path(tenant_id): Path<String>,
|
||||
axum::extract::State(state): axum::extract::State<Arc<AppState>>,
|
||||
) -> Json<Option<TenantBilling>> {
|
||||
let conn = state.db.lock().expect("[TSM.ID] lock");
|
||||
|
||||
// Fetch tenant
|
||||
let mut stmt = conn.prepare("SELECT name, role, packages, quota_limit_gb FROM tenants WHERE tenant_id = ?").expect("[TSM.ID]");
|
||||
let mut rows = stmt.query(params![tenant_id]).expect("[TSM.ID]");
|
||||
|
||||
let mut tenant = None;
|
||||
if let Some(row) = rows.next().expect("[TSM.ID]") {
|
||||
let pkgs_str: String = row.get(2).expect("[TSM.ID]");
|
||||
let packages: Vec<String> = pkgs_str.split(',').map(|s| s.to_string()).collect();
|
||||
|
||||
tenant = Some(TenantBilling {
|
||||
tenant_id: tenant_id.clone(),
|
||||
name: row.get(0).expect("[TSM.ID]"),
|
||||
role: row.get(1).expect("[TSM.ID]"),
|
||||
packages,
|
||||
quota_limit_gb: row.get(3).expect("[TSM.ID]"),
|
||||
used_gb: 0.0,
|
||||
});
|
||||
}
|
||||
|
||||
if let Some(mut t) = tenant {
|
||||
// Aggregate usage
|
||||
let mut stmt = conn.prepare("SELECT SUM(bytes_used) FROM usage_logs WHERE tenant_id = ?").expect("[TSM.ID]");
|
||||
let mut rows = stmt.query(params![tenant_id]).expect("[TSM.ID]");
|
||||
if let Some(row) = rows.next().expect("[TSM.ID]") {
|
||||
let total_bytes: Option<f64> = row.get(0).unwrap_or(None);
|
||||
if let Some(bytes) = total_bytes {
|
||||
t.used_gb = bytes / 1_073_741_824.0;
|
||||
}
|
||||
}
|
||||
return Json(Some(t));
|
||||
}
|
||||
|
||||
Json(None)
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct ArchDoc {
|
||||
version: String,
|
||||
timestamp: String,
|
||||
narrative: String,
|
||||
content: String,
|
||||
}
|
||||
|
||||
async fn get_docs_history(
|
||||
axum::extract::State(state): axum::extract::State<Arc<AppState>>,
|
||||
) -> Json<Vec<ArchDoc>> {
|
||||
let conn = state.db.lock().expect("[TSM.ID] lock");
|
||||
let mut stmt = conn.prepare("SELECT version, logged_at, narrative, content FROM architecture_docs ORDER BY logged_at DESC").expect("[TSM.ID]");
|
||||
let rows = stmt.query_map([], |row| {
|
||||
Ok(ArchDoc {
|
||||
version: row.get(0)?,
|
||||
timestamp: row.get(1)?,
|
||||
narrative: row.get(2)?,
|
||||
content: row.get(3)?,
|
||||
})
|
||||
}).expect("[TSM.ID]");
|
||||
|
||||
let mut docs = Vec::new();
|
||||
for row in rows {
|
||||
docs.push(row.expect("[TSM.ID]"));
|
||||
}
|
||||
Json(docs)
|
||||
}
|
||||
|
||||
async fn report_usage(
|
||||
axum::extract::State(state): axum::extract::State<Arc<AppState>>,
|
||||
Json(payload): Json<UsageReport>,
|
||||
) -> &'static str {
|
||||
let conn = state.db.lock().expect("[TSM.ID] lock");
|
||||
conn.execute(
|
||||
"INSERT INTO usage_logs (tenant_id, bytes_used, logged_at) VALUES (?, ?, current_timestamp)",
|
||||
params![payload.tenant_id, payload.bytes_used as f64],
|
||||
).expect("[TSM.ID]");
|
||||
"OK"
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[allow(dead_code)]
|
||||
struct IamStatePayload {
|
||||
state_json: String,
|
||||
}
|
||||
|
||||
async fn get_iam_state(
|
||||
axum::extract::State(state): axum::extract::State<Arc<AppState>>,
|
||||
) -> Json<serde_json::Value> {
|
||||
let conn = state.db.lock().expect("[TSM.ID] lock");
|
||||
let mut stmt = conn.prepare("SELECT state_json FROM aegis_state WHERE id = 1").expect("[TSM.ID]");
|
||||
let mut rows = stmt.query([]).expect("[TSM.ID]");
|
||||
|
||||
if let Some(row) = rows.next().expect("[TSM.ID]") {
|
||||
let json_str: String = row.get(0).expect("[TSM.ID]");
|
||||
if let Ok(val) = serde_json::from_str(&json_str) {
|
||||
return Json(val);
|
||||
}
|
||||
}
|
||||
// Return empty state if none
|
||||
Json(serde_json::json!({ "identities": {}, "policies": {} }))
|
||||
}
|
||||
|
||||
async fn update_iam_state(
|
||||
axum::extract::State(state): axum::extract::State<Arc<AppState>>,
|
||||
Json(payload): Json<serde_json::Value>,
|
||||
) -> &'static str {
|
||||
let state_json = payload.to_string();
|
||||
|
||||
// Save to DuckDB
|
||||
{
|
||||
let conn = state.db.lock().expect("[TSM.ID] lock");
|
||||
conn.execute(
|
||||
"INSERT INTO aegis_state (id, state_json) VALUES (1, ?) ON CONFLICT (id) DO UPDATE SET state_json = EXCLUDED.state_json",
|
||||
params![state_json.clone()],
|
||||
).expect("[TSM.ID]");
|
||||
}
|
||||
|
||||
// Publish to Redis
|
||||
if let Some(ref client) = state.redis_client {
|
||||
if let Ok(mut con) = client.get_connection() {
|
||||
let _: () = redis::cmd("PUBLISH")
|
||||
.arg("AEGIS_IAM_STATE_CHANNEL")
|
||||
.arg(&state_json)
|
||||
.query(&mut con)
|
||||
.unwrap_or(());
|
||||
}
|
||||
}
|
||||
|
||||
"OK"
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
tracing_subscriber::fmt::init();
|
||||
info!("Starting XCU Quantum Tollgate (DuckDB Billing Matrix)...");
|
||||
|
||||
// Initialize DuckDB
|
||||
let conn = Connection::open("xcu_billing.duckdb").expect("Failed to open DuckDB");
|
||||
|
||||
// Create Tables
|
||||
conn.execute_batch(
|
||||
r"
|
||||
CREATE TABLE IF NOT EXISTS tenants (
|
||||
tenant_id VARCHAR PRIMARY KEY,
|
||||
name VARCHAR,
|
||||
role VARCHAR,
|
||||
packages VARCHAR,
|
||||
quota_limit_gb DOUBLE
|
||||
);
|
||||
CREATE SEQUENCE IF NOT EXISTS seq_usage_id;
|
||||
CREATE TABLE IF NOT EXISTS usage_logs (
|
||||
id BIGINT DEFAULT nextval('seq_usage_id'),
|
||||
tenant_id VARCHAR,
|
||||
bytes_used DOUBLE,
|
||||
logged_at TIMESTAMP
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS architecture_docs (
|
||||
version VARCHAR PRIMARY KEY,
|
||||
narrative VARCHAR,
|
||||
content VARCHAR,
|
||||
logged_at TIMESTAMP DEFAULT current_timestamp
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS aegis_state (
|
||||
id INTEGER PRIMARY KEY,
|
||||
state_json VARCHAR
|
||||
);
|
||||
"
|
||||
).expect("Failed to initialize schemas");
|
||||
|
||||
// Insert Seed Architecture Doc
|
||||
let _ = conn.execute(
|
||||
"INSERT OR IGNORE INTO architecture_docs (version, narrative, content, logged_at) VALUES (?, ?, ?, current_timestamp)",
|
||||
params![
|
||||
"Ver.TSM.19:24:00.07:05:2026.F89A",
|
||||
"Pemisahan Mutlak antara Mesin Pemrosesan Video (XCU Core) dan Mesin Penagihan API (DuckDB Billing Matrix).",
|
||||
"graph TD\n A[Supreme Admin UI] -->|API Request| B(api.xc.ultramodul.xyz)\n C[Tenant UI / JUMPA.ID] -->|API Request| B\n B -->|Query & Save| D[(DuckDB: Billing & Iam)]\n \n E[XCU Core Engine\nxc.ultramodul.xyz] -->|WebRTC / QUIC Stream| F[Video Routing]\n E -->|Send Live Usage| B\n B -->|Validate Token| E\n \n classDef muscle fill:#a855f7,stroke:#fff,color:#fff;\n classDef brain fill:#00d2ff,stroke:#fff,color:#000;\n class E,F muscle;\n class B,D brain;"
|
||||
],
|
||||
);
|
||||
|
||||
// Insert Seed Data (No Duplicate errors)
|
||||
let _ = conn.execute(
|
||||
"INSERT OR IGNORE INTO tenants (tenant_id, name, role, packages, quota_limit_gb) VALUES (?, ?, ?, ?, ?)",
|
||||
params!["XCU-000000000", "Supreme Eye", "supreme_admin", "phase-1,phase-72,ouroboros,billing-master", 999999.0],
|
||||
);
|
||||
let _ = conn.execute(
|
||||
"INSERT OR IGNORE INTO tenants (tenant_id, name, role, packages, quota_limit_gb) VALUES (?, ?, ?, ?, ?)",
|
||||
params!["TENANT-8492019", "JUMPA.ID", "tenant", "billing,phase-1,phase-72", 5000.0],
|
||||
);
|
||||
|
||||
// Connect to Redis (neural-relay-bus)
|
||||
let redis_url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://neural-relay-bus:6379".to_string());
|
||||
let redis_client = redis::Client::open(redis_url).ok();
|
||||
if redis_client.is_some() {
|
||||
info!("Connected to Neural Relay Bus (Redis) for IAM State Broadcasting.");
|
||||
} else {
|
||||
warn!("Neural Relay Bus (Redis) not found. IAM Sync will be disabled.");
|
||||
}
|
||||
|
||||
let state = Arc::new(AppState {
|
||||
db: Arc::new(Mutex::new(conn)),
|
||||
redis_client,
|
||||
});
|
||||
|
||||
// Configure CORS for local UI dev, but mostly handled by Nginx in prod
|
||||
use tower_http::cors::{Any, CorsLayer};
|
||||
let cors = CorsLayer::new()
|
||||
.allow_origin(Any)
|
||||
.allow_methods(Any)
|
||||
.allow_headers(Any);
|
||||
|
||||
let app = Router::new()
|
||||
.route("/v1/billing/tenant/:id", get(get_tenant_billing))
|
||||
.route("/v1/internal/usage", post(report_usage))
|
||||
.route("/v1/docs/history", get(get_docs_history))
|
||||
.route("/v1/iam/state", get(get_iam_state).post(update_iam_state))
|
||||
.layer(cors)
|
||||
.with_state(state);
|
||||
|
||||
let listener = tokio::net::TcpListener::bind("0.0.0.0:8082").await.expect("[TSM.ID] fatal");
|
||||
info!("Billing Matrix listening on port 8082");
|
||||
axum::serve(listener, app).await.expect("[TSM.ID] fatal");
|
||||
}
|
||||
Reference in New Issue
Block a user