feat: subscription audit logging, Hysteria2 TLS, and cleanup fixes

- Add subscription_audit module with trace_id support via X-Trace-Id header
- Log all subscription days changes to separate subscription_audit.log
- Log subscription expiration and standalone connection expiration
- Log referral bonuses with paid_days, bonus_days, and both sub_ids
- Support negative days logging (add/subtract/noop direction)
- Add Hysteria2 manual TLS certificate support (tls.cert / tls.key)
- Fix cleanup_expired_connections to skip subscription-owned connections
- Translate Russian comments to English
- Fix peer_online_by_handshake dead code warning

Author: the2pizza

config-api-example.toml

@@ -47,6 +47,13 @@ subscription_expire_interval = 600
 traffic_persist_interval_sec = 3600
 connection_expire_interval = 600
 
+[subscription_audit]
+enabled = true
+directory = "logs/"
+file = "subscription_audit.log"
+rotation = "daily"
+level = "info"
+
 [pg]
 host = "localhost"
 port = 5432

dev/h2.yaml

@@ -1,10 +1,17 @@
 listen: :443
 
+# Option 1: automatic certificates via ACME
 acme:
     domains:
         - api.example.org
     email: admin@example.com
 
+# Option 2: provide your own certificate
+# tls:
+#     cert: /path/to/cert.pem
+#     key: /path/to/key.pem
+# host: api.example.org   # required when ACME is not used
+
 auth:
     type: http
     http:

src/bin/api/bootstrap.rs

@@ -1,7 +1,6 @@
 use openssl::pkey::PKey;
 use std::sync::Arc;
 use tokio::sync::RwLock;
-use tracing_subscriber::Layer;
 
 use fcore::{
     utils::level_from_settings, utils::measure_time, Connection, ConnectionApiOperations,
@@ -137,16 +136,23 @@ fn parse_level(level: &str) -> Option<tracing::Level> {
 }
 
 pub fn init_tracing(settings: ServiceSettings) {
+    use tracing_subscriber::Layer;
+
     let level = level_from_settings(&settings.service.log_level);
 
-    let stdout_layer = fmt::layer()
-        .with_target(true)
-        .with_filter(level)
-        .with_filter(filter_fn(|metadata| {
-            !metadata.target().starts_with("metrics") && !metadata.target().starts_with("sqlx")
-        }));
+    let stdout_layer: Box<dyn Layer<tracing_subscriber::Registry> + Send + Sync> = Box::new(
+        fmt::layer()
+            .with_target(true)
+            .with_filter(level)
+            .with_filter(filter_fn(|metadata| {
+                !metadata.target().starts_with("metrics")
+                    && !metadata.target().starts_with("subscription.audit")
+                    && !metadata.target().starts_with("sqlx")
+            })),
+    );
 
-    let registry = tracing_subscriber::registry().with(stdout_layer);
+    let mut layers: Vec<Box<dyn Layer<tracing_subscriber::Registry> + Send + Sync>> =
+        vec![stdout_layer];
 
     if settings.metrics.log.enabled {
         let log_directory = settings.metrics.log.directory;
@@ -157,20 +163,42 @@ pub fn init_tracing(settings: ServiceSettings) {
 
         let metrics_file = RollingFileAppender::new(rotation, log_directory, log_file);
 
-        let metrics_layer = fmt::layer()
-            .with_ansi(false)
-            .with_target(true)
-            .with_writer(metrics_file)
-            .with_filter(
-                Targets::new()
-                    .with_target("metrics", metrics_level)
-                    .with_target("metrics.ingest", metrics_level)
-                    .with_target("metrics.gc", metrics_level)
-                    .with_target("metrics.heartbeat", metrics_level),
-            );
-
-        registry.with(metrics_layer).init();
-    } else {
-        registry.init();
+        let metrics_layer: Box<dyn Layer<tracing_subscriber::Registry> + Send + Sync> = Box::new(
+            fmt::layer()
+                .with_ansi(false)
+                .with_target(true)
+                .with_writer(metrics_file)
+                .with_filter(
+                    Targets::new()
+                        .with_target("metrics", metrics_level)
+                        .with_target("metrics.ingest", metrics_level)
+                        .with_target("metrics.gc", metrics_level)
+                        .with_target("metrics.heartbeat", metrics_level),
+                ),
+        );
+
+        layers.push(metrics_layer);
     }
+
+    if settings.subscription_audit.enabled {
+        let log_directory = settings.subscription_audit.directory;
+        let log_file = settings.subscription_audit.file;
+        let rotation = parse_rotation(&settings.subscription_audit.rotation);
+        let audit_level = parse_level(&settings.subscription_audit.level)
+            .unwrap_or(tracing::Level::INFO);
+
+        let audit_file = RollingFileAppender::new(rotation, log_directory, log_file);
+
+        let audit_layer: Box<dyn Layer<tracing_subscriber::Registry> + Send + Sync> = Box::new(
+            fmt::layer()
+                .with_ansi(false)
+                .with_target(true)
+                .with_writer(audit_file)
+                .with_filter(Targets::new().with_target("subscription.audit", audit_level)),
+        );
+
+        layers.push(audit_layer);
+    }
+
+    tracing_subscriber::registry().with(layers).init();
 }

src/bin/api/config.rs

@@ -11,6 +11,7 @@ pub struct ServiceSettings {
     pub metrics: MetricsRxConfig,
     pub tasks: TasksConfig,
     pub smtp: SmtpConfig,
+    pub subscription_audit: SubscriptionAuditConfig,
 }
 
 impl Settings for ServiceSettings {
@@ -127,6 +128,28 @@ impl Default for MetricsLogConfig {
     }
 }
 
+#[derive(Clone, Debug, Deserialize)]
+#[serde(default)]
+pub struct SubscriptionAuditConfig {
+    pub enabled: bool,
+    pub directory: String,
+    pub file: String,
+    pub rotation: String,
+    pub level: String,
+}
+
+impl Default for SubscriptionAuditConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            directory: "logs".to_string(),
+            file: "subscription_audit.log".to_string(),
+            rotation: "daily".to_string(),
+            level: "info".to_string(),
+        }
+    }
+}
+
 #[derive(Clone, Default, Debug, Deserialize)]
 pub struct MetricsRxConfig {
     pub reciever: String,

src/bin/api/http/handlers/key.rs

@@ -1,4 +1,4 @@
-use tracing::error;
+use tracing::{error, Instrument};
 
 use fcore::{
     http::{helpers as http, response::Instance},
@@ -7,6 +7,7 @@ use fcore::{
 };
 
 use super::super::{
+    super::subscription_audit,
     super::sync::{tasks::SyncOp, MemSync},
     param::KeyQueryParams,
     request::{ActivateKeyReq, KeyReq},
@@ -105,6 +106,7 @@ where
 /// Post activate key
 pub async fn post_activate_key_handler<N, C, S>(
     req: ActivateKeyReq,
+    trace_id_header: Option<String>,
     memory: MemSync<N, C, S>,
 ) -> Result<impl warp::Reply, warp::Rejection>
 where
@@ -120,6 +122,7 @@ where
     S: SubscriptionOperations + Send + Sync + Clone + 'static + From<Subscription> + PartialEq,
     Connection: From<C>,
 {
+    let trace_id = subscription_audit::trace_id_from_header(trace_id_header);
     let key_db = memory.db.key();
 
     let mut key = match key_db.get(&req.code).await {
@@ -131,7 +134,20 @@ where
         return Ok(http::bad_request("Key already activated"));
     }
 
-    match SyncOp::add_days(&memory, &req.subscription_id, key.days as i64).await {
+    subscription_audit::log_transaction_start(req.subscription_id, Some(key.days as i64));
+
+    match SyncOp::add_days(
+        &memory,
+        &req.subscription_id,
+        key.days as i64,
+    )
+    .instrument(subscription_audit::transaction_span(
+        "key_activate_handler",
+        req.subscription_id,
+        Some(trace_id),
+    ))
+    .await
+    {
         Ok(Status::Updated(_)) => {
             key.activate(&req.subscription_id);
             if let Err(err) = key_db.activate(&key).await {

src/bin/api/http/handlers/premium.rs

@@ -1,11 +1,15 @@
 use chrono::{DateTime, Utc};
+use tracing::Instrument;
 use serde::{Deserialize, Serialize};
 use std::net::{IpAddr, Ipv4Addr};
 use std::sync::Arc;
 use uuid::Uuid;
 use warp::{http::StatusCode, Rejection, Reply};
 
-use super::super::super::sync::{tasks::SyncOp, MemSync};
+use super::super::super::{
+    subscription_audit,
+    sync::{tasks::SyncOp, MemSync},
+};
 use super::admin::AdminSubscriptionTraffic;
 use super::subscription::build_subscription_traffic;
 
@@ -169,6 +173,7 @@ where
 pub async fn premium_create_child_handler<N, C, S>(
     parent: S,
     req: PremiumChildCreateRequest,
+    trace_id_header: Option<String>,
     memory: MemSync<N, C, S>,
 ) -> Result<impl Reply, Rejection>
 where
@@ -184,14 +189,24 @@ where
     Connection: From<C>,
     S: SubscriptionOperations + Send + Sync + Clone + 'static + PartialEq + From<Subscription>,
 {
+    let trace_id = subscription_audit::trace_id_from_header(trace_id_header);
     let sub_id = Uuid::new_v4();
     let ref_code = fcore::utils::get_uuid_last_octet_simple(&sub_id);
     let expires_at = req.days.map(|d| Utc::now() + chrono::Duration::days(d));
 
     let mut sub = Subscription::new(sub_id, None, ref_code, expires_at, req.limit_bytes);
     sub.set_parent_id(parent.id());
 
-    match SyncOp::add_sub(&memory, sub).await {
+    subscription_audit::log_transaction_start(sub_id, req.days);
+
+    match SyncOp::add_sub(&memory, sub)
+        .instrument(subscription_audit::transaction_span(
+            "premium_create_child_handler",
+            sub_id,
+            Some(trace_id),
+        ))
+        .await
+    {
         Ok(_) => Ok(warp::reply::with_status(
             warp::reply::json(&serde_json::json!({ "id": sub_id })),
             StatusCode::CREATED,
@@ -207,6 +222,7 @@ pub async fn premium_update_child_handler<N, C, S>(
     parent: S,
     child_id: Uuid,
     req: PremiumChildUpdateRequest,
+    trace_id_header: Option<String>,
     memory: MemSync<N, C, S>,
 ) -> Result<impl Reply, Rejection>
 where
@@ -222,6 +238,7 @@ where
     Connection: From<C>,
     S: SubscriptionOperations + Send + Sync + Clone + 'static + PartialEq + From<Subscription>,
 {
+    let trace_id = subscription_audit::trace_id_from_header(trace_id_header);
     {
         let mem = memory.memory.read().await;
         let child = mem.subscriptions.find_by_id(&child_id).ok_or_else(|| {
@@ -237,7 +254,20 @@ where
         limit_bytes: req.limit_bytes,
     };
 
-    match SyncOp::update_sub(&memory, &child_id, update_req).await {
+    subscription_audit::log_transaction_start(child_id, req.days);
+
+    match SyncOp::update_sub(
+        &memory,
+        &child_id,
+        update_req,
+    )
+    .instrument(subscription_audit::transaction_span(
+        "premium_update_child_handler",
+        child_id,
+        Some(trace_id),
+    ))
+    .await
+    {
         Ok(_) => Ok(warp::reply::with_status(
             warp::reply::json(&serde_json::json!({ "id": child_id })),
             StatusCode::OK,

src/bin/api/http/handlers/subscription.rs

@@ -1,5 +1,6 @@
 use base64::Engine;
 use chrono::{DateTime, Utc};
+use tracing::Instrument;
 use serde::Deserialize;
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
@@ -21,7 +22,10 @@ use fcore::{
     SubscriptionOperations, SubscriptionStorageOperations, Tag,
 };
 
-use super::super::super::sync::{tasks::SyncOp, MemSync};
+use super::super::super::{
+    subscription_audit,
+    sync::{tasks::SyncOp, MemSync},
+};
 use super::super::{
     param::SubIdQueryParam,
     request::EnvFilter,
@@ -165,6 +169,7 @@ pub(crate) async fn build_subscription_traffic(
 // POST /subscription
 pub async fn post_subscription_handler<N, C, S>(
     req: SubReq,
+    trace_id_header: Option<String>,
     memory: MemSync<N, C, S>,
     system_refer_codes: Vec<String>,
 ) -> Result<impl warp::Reply, warp::Rejection>
@@ -181,6 +186,7 @@ where
     Connection: From<C>,
     S: SubscriptionOperations + Send + Sync + Clone + 'static + PartialEq + From<Subscription>,
 {
+    let trace_id = subscription_audit::trace_id_from_header(trace_id_header);
     let sub_id = uuid::Uuid::new_v4();
 
     let ref_code = req
@@ -209,7 +215,16 @@ where
         req.limit_bytes,
     );
 
-    match SyncOp::add_sub(&memory, sub.clone()).await {
+    subscription_audit::log_transaction_start(sub_id, req.days);
+
+    match SyncOp::add_sub(&memory, sub.clone())
+        .instrument(subscription_audit::transaction_span(
+            "create_subscription_handler",
+            sub_id,
+            Some(trace_id),
+        ))
+        .await
+    {
         Ok(Status::Ok(id)) => Ok(http::success_response(
             format!("Subscription {} has been created", id),
             Some(sub_id),
@@ -236,6 +251,7 @@ where
 pub async fn put_subscription_handler<N, C, S>(
     sub_param: SubIdQueryParam,
     req: SubReq,
+    trace_id_header: Option<String>,
     memory: MemSync<N, C, S>,
 ) -> Result<impl warp::Reply, warp::Rejection>
 where
@@ -251,9 +267,23 @@ where
     Connection: From<C>,
     S: SubscriptionOperations + Send + Sync + Clone + 'static + PartialEq + From<Subscription>,
 {
+    let trace_id = subscription_audit::trace_id_from_header(trace_id_header);
     let sub_id = sub_param.id;
 
-    match SyncOp::update_sub(&memory, &sub_id, req).await {
+    subscription_audit::log_transaction_start(sub_id, req.days);
+
+    match SyncOp::update_sub(
+        &memory,
+        &sub_id,
+        req,
+    )
+    .instrument(subscription_audit::transaction_span(
+        "put_subscription_handler",
+        sub_id,
+        Some(trace_id),
+    ))
+    .await
+    {
         Ok(Status::Updated(id)) => Ok(http::success_response(
             format!("Subscription {} has been updated", id),
             Some(sub_id),