Initial

src/queue/client.rs

@@ -0,0 +1,80 @@
+//! Shared asynq queue client.
+//!
+//! Wraps `asynq::client::Client` in an `Arc` so it can be cheaply cloned
+//! into `AppState` and shared across every handler.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::Context;
+use serde::Serialize;
+
+use asynq::backend::RedisConnectionType;
+
+/// Thin, cheaply-cloneable wrapper around the asynq client.
+#[derive(Clone)]
+pub struct QueueClient {
+    inner: Arc<asynq::client::Client>,
+}
+
+impl QueueClient {
+    /// Connect to Redis and build a new [`QueueClient`].
+    pub async fn new(redis_url: &str) -> anyhow::Result<Self> {
+        let redis_cfg =
+            RedisConnectionType::single(redis_url).context("build redis connection for queue")?;
+        let client = asynq::client::Client::new(redis_cfg)
+            .await
+            .context("connect asynq queue client")?;
+        Ok(Self {
+            inner: Arc::new(client),
+        })
+    }
+
+    /// Enqueue a task for immediate processing.
+    pub async fn enqueue(&self, task_type: &str, payload: &[u8]) -> anyhow::Result<()> {
+        let task =
+            asynq::task::Task::new(task_type, payload).context("build asynq task")?;
+        self.inner
+            .enqueue(task)
+            .await
+            .context("enqueue task")?;
+        Ok(())
+    }
+
+    /// Enqueue a task with a JSON-serialisable payload for immediate processing.
+    pub async fn enqueue_json<T: Serialize>(
+        &self,
+        task_type: &str,
+        payload: &T,
+    ) -> anyhow::Result<()> {
+        let bytes = serde_json::to_vec(payload).context("serialize task payload")?;
+        self.enqueue(task_type, &bytes).await
+    }
+
+    /// Enqueue a task to be processed after `delay`.
+    pub async fn enqueue_delayed(
+        &self,
+        task_type: &str,
+        payload: &[u8],
+        delay: Duration,
+    ) -> anyhow::Result<()> {
+        let task =
+            asynq::task::Task::new(task_type, payload).context("build asynq task")?;
+        self.inner
+            .enqueue_in(task, delay)
+            .await
+            .context("enqueue delayed task")?;
+        Ok(())
+    }
+
+    /// Enqueue a delayed task with a JSON-serialisable payload.
+    pub async fn enqueue_json_delayed<T: Serialize>(
+        &self,
+        task_type: &str,
+        payload: &T,
+        delay: Duration,
+    ) -> anyhow::Result<()> {
+        let bytes = serde_json::to_vec(payload).context("serialize task payload")?;
+        self.enqueue_delayed(task_type, &bytes, delay).await
+    }
+}

src/queue/handler/email.rs

@@ -0,0 +1,22 @@
+use std::sync::Arc;
+
+use asynq::error::Result;
+use asynq::task::Task;
+
+use crate::config::Config;
+use crate::queue::service::email::{BatchEmailLogic, SendEmailLogic};
+use crate::repository::Repositories;
+
+pub async fn send_email(task: Task, repos: Arc<Repositories>, config: Arc<Config>) -> Result<()> {
+    SendEmailLogic::new(repos, config)
+        .execute(task.get_payload())
+        .await
+        .map_err(|e| asynq::error::Error::other(e.to_string()))
+}
+
+pub async fn batch_email(task: Task, repos: Arc<Repositories>, config: Arc<Config>) -> Result<()> {
+    BatchEmailLogic::new(repos, config)
+        .execute(task.get_payload())
+        .await
+        .map_err(|e| asynq::error::Error::other(e.to_string()))
+}

src/queue/handler/mod.rs

@@ -0,0 +1,88 @@
+use std::sync::Arc;
+
+use asynq::serve_mux::ServeMux;
+
+use crate::config::Config;
+use crate::repository::Repositories;
+
+pub mod email;
+pub mod order;
+pub mod sms;
+pub mod subscription;
+pub mod task;
+pub mod traffic;
+
+pub fn register_all(repos: Arc<Repositories>, config: Arc<Config>) -> ServeMux {
+    let mut mux = ServeMux::new();
+
+    // ── Email ─────────────────────────────────────────────────────────────────
+    let email_repos = Arc::clone(&repos);
+    let email_config = Arc::clone(&config);
+    mux.handle_async_func(crate::queue::types::FORTHWITH_SEND_EMAIL, move |task| {
+        email::send_email(task, Arc::clone(&email_repos), Arc::clone(&email_config))
+    });
+    let batch_repos = Arc::clone(&repos);
+    let batch_config = Arc::clone(&config);
+    mux.handle_async_func(crate::queue::types::SCHEDULED_BATCH_SEND_EMAIL, move |task| {
+        email::batch_email(task, Arc::clone(&batch_repos), Arc::clone(&batch_config))
+    });
+
+    // ── SMS ───────────────────────────────────────────────────────────────────
+    let sms_repos = Arc::clone(&repos);
+    let sms_config = Arc::clone(&config);
+    mux.handle_async_func(crate::queue::types::FORTHWITH_SEND_SMS, move |task| {
+        sms::send_sms(task, Arc::clone(&sms_repos), Arc::clone(&sms_config))
+    });
+
+    // ── Order ─────────────────────────────────────────────────────────────────
+    let activate_repos = Arc::clone(&repos);
+    let activate_config = Arc::clone(&config);
+    mux.handle_async_func(crate::queue::types::FORTHWITH_ACTIVATE_ORDER, move |task| {
+        order::activate_order(task, Arc::clone(&activate_repos), Arc::clone(&activate_config))
+    });
+    let close_repos = Arc::clone(&repos);
+    mux.handle_async_func(crate::queue::types::DEFER_CLOSE_ORDER, move |task| {
+        order::defer_close_order(task, Arc::clone(&close_repos))
+    });
+
+    // ── Traffic ───────────────────────────────────────────────────────────────
+    mux.handle_func(
+        crate::queue::types::FORTHWITH_TRAFFIC_STATISTICS,
+        traffic::stub_traffic_statistics,
+    );
+    mux.handle_func(
+        crate::queue::types::SCHEDULER_TOTAL_SERVER_DATA,
+        traffic::stub_server_data,
+    );
+    mux.handle_func(
+        crate::queue::types::SCHEDULER_RESET_TRAFFIC,
+        traffic::stub_reset_traffic,
+    );
+    mux.handle_func(
+        crate::queue::types::SCHEDULER_TRAFFIC_STAT,
+        traffic::stub_traffic_stat,
+    );
+
+    // ── Subscription ──────────────────────────────────────────────────────────
+    let sub_repos = Arc::clone(&repos);
+    let sub_config = Arc::clone(&config);
+    mux.handle_async_func(
+        crate::queue::types::SCHEDULER_CHECK_SUBSCRIPTION,
+        move |task| {
+            subscription::check_subscription(
+                task,
+                Arc::clone(&sub_repos),
+                Arc::clone(&sub_config),
+            )
+        },
+    );
+
+    // ── Quota task ────────────────────────────────────────────────────────────
+    let quota_repos = Arc::clone(&repos);
+    let quota_config = Arc::clone(&config);
+    mux.handle_async_func(crate::queue::types::FORTHWITH_QUOTA_TASK, move |task| {
+        task::quota_task(task, Arc::clone(&quota_repos), Arc::clone(&quota_config))
+    });
+
+    mux
+}

src/queue/handler/order.rs

@@ -0,0 +1,31 @@
+use asynq::error::Result;
+use asynq::task::Task;
+use std::sync::Arc;
+
+use crate::config::Config;
+use crate::queue::service::order::{ActivateOrderLogic, DeferCloseOrderLogic, OrderTaskPayload};
+use crate::repository::Repositories;
+
+pub async fn activate_order(
+    task: Task,
+    repos: Arc<Repositories>,
+    config: Arc<Config>,
+) -> Result<()> {
+    let payload = decode_payload(&task)?;
+    ActivateOrderLogic::new(repos, config)
+        .execute(payload)
+        .await
+        .map_err(|err| asynq::error::Error::other(err.to_string()))
+}
+
+pub async fn defer_close_order(task: Task, repos: Arc<Repositories>) -> Result<()> {
+    let payload = decode_payload(&task)?;
+    DeferCloseOrderLogic::new(repos)
+        .execute(payload)
+        .await
+        .map_err(|err| asynq::error::Error::other(err.to_string()))
+}
+
+fn decode_payload(task: &Task) -> Result<OrderTaskPayload> {
+    task.get_payload_with_json()
+}

src/queue/handler/sms.rs

@@ -0,0 +1,15 @@
+use std::sync::Arc;
+
+use asynq::error::Result;
+use asynq::task::Task;
+
+use crate::config::Config;
+use crate::queue::service::sms::SendSmsLogic;
+use crate::repository::Repositories;
+
+pub async fn send_sms(task: Task, repos: Arc<Repositories>, config: Arc<Config>) -> Result<()> {
+    SendSmsLogic::new(repos, config)
+        .execute(task.get_payload())
+        .await
+        .map_err(|e| asynq::error::Error::other(e.to_string()))
+}

src/queue/handler/subscription.rs

@@ -0,0 +1,20 @@
+use std::sync::Arc;
+
+use asynq::error::Result;
+use asynq::task::Task;
+
+use crate::config::Config;
+use crate::queue::service::subscription::CheckSubscriptionLogic;
+use crate::repository::Repositories;
+
+pub async fn check_subscription(
+    task: Task,
+    repos: Arc<Repositories>,
+    config: Arc<Config>,
+) -> Result<()> {
+    let _ = task;
+    CheckSubscriptionLogic::new(repos, config)
+        .execute()
+        .await
+        .map_err(|e| asynq::error::Error::other(e.to_string()))
+}

src/queue/handler/task.rs

@@ -0,0 +1,15 @@
+use std::sync::Arc;
+
+use asynq::error::Result;
+use asynq::task::Task;
+
+use crate::config::Config;
+use crate::queue::service::task::QuotaTaskLogic;
+use crate::repository::Repositories;
+
+pub async fn quota_task(task: Task, repos: Arc<Repositories>, config: Arc<Config>) -> Result<()> {
+    QuotaTaskLogic::new(repos, config)
+        .execute(task.get_payload())
+        .await
+        .map_err(|e| asynq::error::Error::other(e.to_string()))
+}

src/queue/handler/traffic.rs

@@ -0,0 +1,22 @@
+use asynq::error::Result;
+use asynq::task::Task;
+
+pub fn stub_traffic_statistics(task: Task) -> Result<()> {
+    tracing::warn!("STUB traffic::statistics — task={}", task.get_type());
+    Ok(())
+}
+
+pub fn stub_server_data(task: Task) -> Result<()> {
+    tracing::warn!("STUB traffic::server_data — task={}", task.get_type());
+    Ok(())
+}
+
+pub fn stub_reset_traffic(task: Task) -> Result<()> {
+    tracing::warn!("STUB traffic::reset_traffic — task={}", task.get_type());
+    Ok(())
+}
+
+pub fn stub_traffic_stat(task: Task) -> Result<()> {
+    tracing::warn!("STUB traffic::stat — task={}", task.get_type());
+    Ok(())
+}

src/queue/mod.rs

@@ -0,0 +1,56 @@
+use std::sync::Arc;
+
+use asynq::backend::RedisConnectionType;
+use asynq::config::ServerConfig;
+use asynq::server::Server;
+
+use crate::config;
+use crate::config::Config;
+use crate::repository::Repositories;
+
+pub mod client;
+pub mod handler;
+pub mod service;
+pub mod types;
+
+pub fn redis_url(cfg: &config::RedisConfig) -> String {
+    let db = cfg.db;
+    if cfg.pass.is_empty() {
+        format!("redis://{}/{}", cfg.host, db)
+    } else {
+        format!("redis://:{}@{}/{}", cfg.pass, cfg.host, db)
+    }
+}
+
+pub struct Service {
+    server: Server,
+}
+
+impl Service {
+    pub async fn new(
+        cfg: &Config,
+        repos: Arc<Repositories>,
+    ) -> anyhow::Result<Self> {
+        let redis_cfg = RedisConnectionType::single(redis_url(&cfg.redis))?;
+        let server_cfg = ServerConfig::new().concurrency(20);
+        let mut server = Server::new(redis_cfg, server_cfg).await?;
+
+        let config = Arc::new(cfg.clone());
+        let mut mux = handler::register_all(repos, config);
+
+        mux.handle_func("*", |task: asynq::task::Task| {
+            tracing::warn!("unregistered task type: {}", task.get_type());
+            Ok(())
+        });
+
+        server.start(mux).await?;
+        tracing::info!("queue consumer started (concurrency=20)");
+
+        Ok(Self { server })
+    }
+
+    pub async fn shutdown(&mut self) -> anyhow::Result<()> {
+        self.server.shutdown().await?;
+        Ok(())
+    }
+}

src/queue/service/email.rs

@@ -0,0 +1,274 @@
+use std::sync::Arc;
+
+use serde::Deserialize;
+
+use crate::config::Config;
+use crate::repository::Repositories;
+use crate::service::telemetry::Telemetry;
+
+// ─── SendEmail payload ───────────────────────────────────────────────────────
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct SendEmailPayload {
+    #[serde(rename = "type", default)]
+    pub type_: i16,
+    #[serde(rename = "email", default)]
+    pub email: String,
+    #[serde(default)]
+    pub subject: String,
+    /// Raw JSON value — content map forwarded to templates.
+    #[serde(default)]
+    pub content: serde_json::Value,
+}
+
+// Email type constants (matches Go `queue/types`)
+const EMAIL_TYPE_VERIFY: i16 = 1;
+const EMAIL_TYPE_MAINTENANCE: i16 = 2;
+const EMAIL_TYPE_EXPIRATION: i16 = 3;
+const EMAIL_TYPE_TRAFFIC_EXCEED: i16 = 4;
+const EMAIL_TYPE_CUSTOM: i16 = 5;
+
+/// Port of `server/queue/logic/email/sendEmailLogic.go`.
+pub struct SendEmailLogic {
+    repos: Arc<Repositories>,
+    config: Arc<Config>,
+}
+
+impl SendEmailLogic {
+    pub fn new(repos: Arc<Repositories>, config: Arc<Config>) -> Self {
+        Self { repos, config }
+    }
+
+    pub async fn execute(&self, raw: &[u8]) -> anyhow::Result<()> {
+        let payload: SendEmailPayload = match serde_json::from_slice(raw) {
+            Ok(p) => p,
+            Err(e) => {
+                tracing::error!("[SendEmailLogic] deserialise payload: {e}");
+                return Ok(());
+            }
+        };
+
+        // Build sender
+        let sender = match email::new_sender(
+            &self.config.email.platform,
+            &self.config.email.platform_config,
+            &self.config.site.site_name,
+        ) {
+            Ok(s) => s,
+            Err(e) => {
+                tracing::error!("[SendEmailLogic] new_sender: {e}");
+                return Ok(());
+            }
+        };
+
+        // Render body from template based on email type
+        let body = match self.render_body(&payload) {
+            Some(b) => b,
+            None => return Ok(()),
+        };
+
+        let status: i16 = match sender.send(&[payload.email.clone()], &payload.subject, &body).await {
+            Ok(()) => 1,
+            Err(e) => {
+                tracing::error!("[SendEmailLogic] send failed to {}: {e}", payload.email);
+                2
+            }
+        };
+
+        Telemetry::email_message(
+            &self.repos,
+            0,
+            &payload.email,
+            Some(payload.subject.clone()),
+            payload.content.clone(),
+            &self.config.email.platform,
+            "",
+            status,
+        )
+        .await;
+
+        Ok(())
+    }
+
+    fn render_body(&self, payload: &SendEmailPayload) -> Option<String> {
+        let cfg = &self.config.email;
+        let tpl_src = match payload.type_ {
+            EMAIL_TYPE_VERIFY => cfg.verify_email_template.as_str(),
+            EMAIL_TYPE_MAINTENANCE => cfg.maintenance_email_template.as_str(),
+            EMAIL_TYPE_EXPIRATION => cfg.expiration_email_template.as_str(),
+            EMAIL_TYPE_TRAFFIC_EXCEED => cfg.traffic_exceed_email_template.as_str(),
+            EMAIL_TYPE_CUSTOM => {
+                // For custom type use the "content" field directly as HTML
+                if let Some(s) = payload.content.get("content").and_then(|v| v.as_str()) {
+                    return Some(s.to_string());
+                }
+                tracing::error!("[SendEmailLogic] custom email missing content string");
+                return None;
+            }
+            other => {
+                tracing::error!("[SendEmailLogic] unknown email type {other}");
+                return None;
+            }
+        };
+
+        let ctx = json_to_gtmpl(payload.content.clone());
+        match gtmpl::template(tpl_src, ctx) {
+            Ok(rendered) => Some(rendered),
+            Err(e) => {
+                tracing::error!("[SendEmailLogic] template render (type={}): {e}", payload.type_);
+                None
+            }
+        }
+    }
+}
+
+// ─── BatchEmail ──────────────────────────────────────────────────────────────
+
+/// Port of `server/queue/logic/email/batchEmailLogic.go`.
+pub struct BatchEmailLogic {
+    repos: Arc<Repositories>,
+    config: Arc<Config>,
+}
+
+impl BatchEmailLogic {
+    pub fn new(repos: Arc<Repositories>, config: Arc<Config>) -> Self {
+        Self { repos, config }
+    }
+
+    pub async fn execute(&self, raw: &[u8]) -> anyhow::Result<()> {
+        if raw.is_empty() {
+            tracing::error!("[BatchEmailLogic] empty payload");
+            return Ok(());
+        }
+
+        let task_id: i64 = match std::str::from_utf8(raw)
+            .ok()
+            .and_then(|s| s.trim().parse().ok())
+        {
+            Some(id) => id,
+            None => {
+                tracing::error!("[BatchEmailLogic] invalid task ID in payload");
+                return Ok(());
+            }
+        };
+
+        let task_info = match self.repos.task.find_one(task_id).await {
+            Ok(t) => t,
+            Err(e) => {
+                tracing::error!("[BatchEmailLogic] find_one({task_id}): {e}");
+                return Ok(());
+            }
+        };
+
+        if task_info.status != 0 {
+            tracing::info!("[BatchEmailLogic] task {task_id} already processed (status={})", task_info.status);
+            return Ok(());
+        }
+
+        let sender = match email::new_sender(
+            &self.config.email.platform,
+            &self.config.email.platform_config,
+            &self.config.site.site_name,
+        ) {
+            Ok(s) => std::sync::Arc::from(s),
+            Err(e) => {
+                tracing::error!("[BatchEmailLogic] new_sender: {e}");
+                return Ok(());
+            }
+        };
+
+        // Use the global WorkerManager (created at startup) or create a local one
+        if let Some(mgr) = email::get_global_manager() {
+            mgr.add_worker(task_id).await;
+        } else {
+            // Fallback: create a transient manager backed by the task repo adapter
+            let repo_adapter = Arc::new(TaskRepoAdapter {
+                inner: self.repos.task.as_ref() as *const _,
+            });
+            // SAFETY: We hold `self.repos` for the lifetime of this call.
+            // The adapter is only used within this async scope.
+            let mgr = email::WorkerManager::new(repo_adapter, sender);
+            mgr.add_worker(task_id).await;
+        }
+
+        Ok(())
+    }
+}
+
+// ─── TaskRepo adapter bridging `email::manager::TaskRepo` → our repo ─────────
+
+use std::sync::Mutex;
+
+struct TaskRepoAdapter {
+    // raw pointer: only safe because the adapter is used within a single
+    // async scope where `repos.task` is guaranteed alive.
+    inner: *const dyn crate::repository::task::TaskRepo,
+}
+
+unsafe impl Send for TaskRepoAdapter {}
+unsafe impl Sync for TaskRepoAdapter {}
+
+#[async_trait::async_trait]
+impl email::manager::TaskRepo for TaskRepoAdapter {
+    async fn find_one(&self, id: i64) -> Result<email::worker::TaskInfo, anyhow::Error> {
+        // SAFETY: pointer is valid for the duration of the call (see above).
+        let repo = unsafe { &*self.inner };
+        let t = repo.find_one(id).await?;
+        Ok(email::worker::TaskInfo {
+            id: t.id,
+            type_: t.type_,
+            scope: t.scope.clone().unwrap_or_default(),
+            content: t.content.clone().unwrap_or_default(),
+            status: t.status,
+            errors: t.errors.clone().unwrap_or_default(),
+            total: t.total,
+            current: t.current,
+        })
+    }
+
+    async fn update(&self, data: &email::worker::TaskInfo) -> Result<(), anyhow::Error> {
+        let repo = unsafe { &*self.inner };
+        let mut t = repo.find_one(data.id).await?;
+        t.status = data.status;
+        t.current = data.current;
+        t.errors = if data.errors.is_empty() { None } else { Some(data.errors.clone()) };
+        repo.update(&t).await?;
+        Ok(())
+    }
+
+    async fn update_status(&self, id: i64, status: i16) -> Result<(), anyhow::Error> {
+        let repo = unsafe { &*self.inner };
+        repo.update_status(id, status).await?;
+        Ok(())
+    }
+
+    fn is_cancelled(&self, _id: i64) -> bool {
+        false
+    }
+}
+
+fn json_to_gtmpl(v: serde_json::Value) -> gtmpl::Value {
+    use std::collections::HashMap;
+    match v {
+        serde_json::Value::Null => gtmpl::Value::Nil,
+        serde_json::Value::Bool(b) => gtmpl::Value::Bool(b),
+        serde_json::Value::Number(n) => {
+            if let Some(i) = n.as_i64() {
+                gtmpl::Value::Number(gtmpl_value::Number::from(i))
+            } else if let Some(f) = n.as_f64() {
+                gtmpl::Value::Number(gtmpl_value::Number::from(f))
+            } else {
+                gtmpl::Value::Number(gtmpl_value::Number::from(0_i64))
+            }
+        }
+        serde_json::Value::String(s) => gtmpl::Value::String(s),
+        serde_json::Value::Array(arr) => {
+            gtmpl::Value::Array(arr.into_iter().map(json_to_gtmpl).collect())
+        }
+        serde_json::Value::Object(map) => {
+            let m: HashMap<String, gtmpl::Value> =
+                map.into_iter().map(|(k, v)| (k, json_to_gtmpl(v))).collect();
+            gtmpl::Value::Map(m)
+        }
+    }
+}

src/queue/service/mod.rs

@@ -0,0 +1,7 @@
+/// Stub service module — port of `server/queue/logic/`.
+pub mod email;
+pub mod order;
+pub mod sms;
+pub mod subscription;
+pub mod task;
+pub mod traffic;

src/queue/service/order.rs

@@ -0,0 +1,285 @@
+use std::sync::Arc;
+
+use anyhow::{anyhow, Context};
+use chrono::{Datelike, Months, Utc};
+use serde::Deserialize;
+use uuid::Uuid;
+
+use crate::config::Config;
+use crate::model::entity::log::{BALANCE_TYPE_RECHARGE, COMMISSION_TYPE_PURCHASE, COMMISSION_TYPE_RENEWAL, RESET_SUBSCRIBE_TYPE_PAID};
+use crate::model::entity::order::Order;
+use crate::model::entity::subscribe::Subscribe;
+use crate::model::entity::user::{User, UserSubscribe};
+use crate::repository::Repositories;
+use crate::service::telemetry::Telemetry;
+
+const ORDER_TYPE_SUBSCRIBE: i16 = 1;
+const ORDER_TYPE_RENEWAL: i16 = 2;
+const ORDER_TYPE_RESET_TRAFFIC: i16 = 3;
+const ORDER_TYPE_RECHARGE: i16 = 4;
+
+const ORDER_STATUS_UNPAID: i16 = 1;
+const ORDER_STATUS_PAID: i16 = 2;
+const ORDER_STATUS_CANCELLED: i16 = 3;
+
+const USER_SUBSCRIBE_STATUS_ACTIVE: i16 = 1;
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct OrderTaskPayload {
+    pub order_id: i64,
+}
+
+pub struct ActivateOrderLogic {
+    repos: Arc<Repositories>,
+    config: Arc<Config>,
+}
+
+impl ActivateOrderLogic {
+    pub fn new(repos: Arc<Repositories>, config: Arc<Config>) -> Self {
+        Self { repos, config }
+    }
+
+    pub async fn execute(&self, payload: OrderTaskPayload) -> anyhow::Result<()> {
+        let mut order = self.find_order(payload.order_id).await?;
+        if order.status != ORDER_STATUS_UNPAID {
+            return Ok(());
+        }
+
+        match order.type_ {
+            ORDER_TYPE_SUBSCRIBE => self.activate_new_subscription(&order).await?,
+            ORDER_TYPE_RENEWAL => self.activate_renewal(&order).await?,
+            ORDER_TYPE_RESET_TRAFFIC => self.activate_traffic_reset(&order).await?,
+            ORDER_TYPE_RECHARGE => self.activate_balance_recharge(&order).await?,
+            other => return Err(anyhow!("invalid order type: {other}")),
+        }
+
+        order.status = ORDER_STATUS_PAID;
+        order.updated_at = now_ms();
+        self.repos.order.update(&order).await?;
+        Ok(())
+    }
+
+    async fn find_order(&self, order_id: i64) -> anyhow::Result<Order> {
+        self.repos
+            .order
+            .find_one(order_id)
+            .await
+            .with_context(|| format!("find order {order_id}"))
+    }
+
+    async fn activate_new_subscription(&self, order: &Order) -> anyhow::Result<()> {
+        let user = self.repos.user.find_one_user(order.user_id).await?;
+        let plan = self.repos.subscribe.find_one(order.subscribe_id).await?;
+        let user_subscribe = self.create_user_subscribe(order, &plan).await?;
+
+        Telemetry::subscribe_access(&self.repos, user_subscribe.id, &user_subscribe.token, "", "").await;
+        self.apply_commission(&user, order, COMMISSION_TYPE_PURCHASE).await?;
+        Ok(())
+    }
+
+    async fn activate_renewal(&self, order: &Order) -> anyhow::Result<()> {
+        let user = self.repos.user.find_one_user(order.user_id).await?;
+        let plan = self.repos.subscribe.find_one(order.subscribe_id).await?;
+        let token = order
+            .subscribe_token
+            .as_deref()
+            .context("renewal order missing subscribe_token")?;
+        let mut user_subscribe = self.repos.user.find_one_subscribe_by_token(token).await?;
+        let now = Utc::now();
+        let base_time = if user_subscribe.expire_time < now.timestamp_millis() {
+            now
+        } else {
+            datetime_from_ms(user_subscribe.expire_time)
+        };
+
+        if plan.renewal_reset || should_reset_for_renewal(user_subscribe.expire_time, now) {
+            user_subscribe.download = 0;
+            user_subscribe.upload = 0;
+        }
+        user_subscribe.expire_time = add_time(&plan.unit_time, order.quantity, base_time).timestamp_millis();
+        user_subscribe.traffic = plan.traffic;
+        user_subscribe.finished_at = None;
+        user_subscribe.status = USER_SUBSCRIBE_STATUS_ACTIVE;
+        user_subscribe.updated_at = now.timestamp_millis();
+
+        let updated = self.repos.user.update_subscribe(&user_subscribe).await?;
+        Telemetry::subscribe_access(&self.repos, updated.id, &updated.token, "", "").await;
+        self.apply_commission(&user, order, COMMISSION_TYPE_RENEWAL).await?;
+        Ok(())
+    }
+
+    async fn activate_traffic_reset(&self, order: &Order) -> anyhow::Result<()> {
+        let token = order
+            .subscribe_token
+            .as_deref()
+            .context("traffic reset order missing subscribe_token")?;
+        let mut user_subscribe = self.repos.user.find_one_subscribe_by_token(token).await?;
+        user_subscribe.download = 0;
+        user_subscribe.upload = 0;
+        user_subscribe.status = USER_SUBSCRIBE_STATUS_ACTIVE;
+        user_subscribe.updated_at = now_ms();
+        self.repos.user.update_subscribe(&user_subscribe).await?;
+
+        Telemetry::reset_subscribe(
+            &self.repos,
+            order.user_id,
+            RESET_SUBSCRIBE_TYPE_PAID,
+            Some(order.order_no.clone()),
+        )
+        .await;
+        Ok(())
+    }
+
+    async fn activate_balance_recharge(&self, order: &Order) -> anyhow::Result<()> {
+        let mut user = self.repos.user.find_one_user(order.user_id).await?;
+        user.balance += order.amount;
+        user.updated_at = now_ms();
+        let updated = self.repos.user.update_user(&user).await?;
+        Telemetry::balance(
+            &self.repos,
+            updated.id,
+            BALANCE_TYPE_RECHARGE,
+            order.amount,
+            Some(order.order_no.clone()),
+            updated.balance,
+        )
+        .await;
+        Ok(())
+    }
+
+    async fn create_user_subscribe(&self, order: &Order, plan: &Subscribe) -> anyhow::Result<UserSubscribe> {
+        if plan.quota > 0 {
+            let current = self
+                .repos
+                .user
+                .count_user_subscribes_by_user_and_subscribe(order.user_id, order.subscribe_id)
+                .await?;
+            if current >= plan.quota {
+                return Err(anyhow!("subscribe quota limit exceeded"));
+            }
+        }
+
+        let now = Utc::now();
+        let user_subscribe = UserSubscribe {
+            id: 0,
+            user_id: order.user_id,
+            order_id: order.id,
+            subscribe_id: order.subscribe_id,
+            start_time: now.timestamp_millis(),
+            expire_time: add_time(&plan.unit_time, order.quantity, now).timestamp_millis(),
+            finished_at: None,
+            traffic: plan.traffic,
+            download: 0,
+            upload: 0,
+            token: format!("Order-{}-{}", order.order_no, Uuid::new_v4()),
+            uuid: Uuid::new_v4().to_string(),
+            status: USER_SUBSCRIBE_STATUS_ACTIVE,
+            note: String::new(),
+            created_at: now.timestamp_millis(),
+            updated_at: now.timestamp_millis(),
+        };
+
+        self.repos.user.insert_subscribe(&user_subscribe).await.map_err(Into::into)
+    }
+
+    async fn apply_commission(&self, user: &User, order: &Order, commission_type: i32) -> anyhow::Result<()> {
+        if user.referer_id == 0 {
+            return Ok(());
+        }
+
+        let mut referer = self.repos.user.find_one_user(user.referer_id).await?;
+        let referral_percentage = if referer.referral_percentage > 0 {
+            i64::from(referer.referral_percentage)
+        } else {
+            self.config.invite.referral_percentage
+        };
+        if referral_percentage == 0 {
+            return Ok(());
+        }
+
+        let only_first_purchase = if referer.referral_percentage > 0 {
+            referer.only_first_purchase
+        } else {
+            self.config.invite.only_first_purchase
+        };
+        if only_first_purchase && !order.is_new {
+            return Ok(());
+        }
+
+        let commission_base = order.amount - order.fee_amount;
+        let commission = commission_base * referral_percentage / 100;
+        referer.commission += commission;
+        referer.updated_at = now_ms();
+        self.repos.user.update_user(&referer).await?;
+        Telemetry::commission(&self.repos, referer.id, commission_type, commission, &order.order_no).await;
+        Ok(())
+    }
+}
+
+pub struct DeferCloseOrderLogic {
+    repos: Arc<Repositories>,
+}
+
+impl DeferCloseOrderLogic {
+    pub fn new(repos: Arc<Repositories>) -> Self {
+        Self { repos }
+    }
+
+    pub async fn execute(&self, payload: OrderTaskPayload) -> anyhow::Result<()> {
+        let order = self.repos.order.find_one(payload.order_id).await;
+        let mut order = match order {
+            Ok(order) => order,
+            Err(sqlx::Error::RowNotFound) => return Ok(()),
+            Err(err) => return Err(err.into()),
+        };
+
+        if order.status == ORDER_STATUS_UNPAID {
+            order.status = ORDER_STATUS_CANCELLED;
+            order.updated_at = now_ms();
+            self.repos.order.update(&order).await?;
+        }
+
+        Ok(())
+    }
+}
+
+fn now_ms() -> i64 {
+    Utc::now().timestamp_millis()
+}
+
+fn datetime_from_ms(timestamp_ms: i64) -> chrono::DateTime<Utc> {
+    match chrono::DateTime::<Utc>::from_timestamp_millis(timestamp_ms) {
+        Some(datetime) => datetime,
+        None => Utc::now(),
+    }
+}
+
+fn add_time(unit: &str, amount: i64, from: chrono::DateTime<Utc>) -> chrono::DateTime<Utc> {
+    match unit {
+        "hour" => from + chrono::Duration::hours(amount),
+        "day" => from + chrono::Duration::days(amount),
+        "week" => from + chrono::Duration::weeks(amount),
+        "month" => add_months(from, amount),
+        "year" => add_months(from, amount.saturating_mul(12)),
+        _ => from + chrono::Duration::days(amount),
+    }
+}
+
+fn add_months(from: chrono::DateTime<Utc>, amount: i64) -> chrono::DateTime<Utc> {
+    if amount <= 0 {
+        return from;
+    }
+
+    let months = match u32::try_from(amount) {
+        Ok(value) => value,
+        Err(_) => u32::MAX,
+    };
+    match from.checked_add_months(Months::new(months)) {
+        Some(datetime) => datetime,
+        None => from,
+    }
+}
+
+fn should_reset_for_renewal(expire_time_ms: i64, now: chrono::DateTime<Utc>) -> bool {
+    datetime_from_ms(expire_time_ms).day() == now.day()
+}

src/queue/service/sms.rs

@@ -0,0 +1,99 @@
+use std::sync::Arc;
+
+use serde::Deserialize;
+
+use crate::config::Config;
+use crate::repository::Repositories;
+use crate::service::telemetry::Telemetry;
+
+/// Port of `server/queue/logic/sms/sendSmsLogic.go`.
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct SendSmsPayload {
+    /// Country / area dial code, e.g. "86"
+    #[serde(rename = "TelephoneArea", default)]
+    pub telephone_area: String,
+    /// Phone number without country code
+    #[serde(rename = "Telephone", default)]
+    pub telephone: String,
+    /// Verification code string to send
+    #[serde(rename = "Content", default)]
+    pub content: String,
+    /// Expiry minutes (passed to some providers)
+    #[serde(rename = "Expire", default)]
+    pub expire: u32,
+    /// Message type tag (used for audit logging)
+    #[serde(rename = "Type", default)]
+    pub type_: i16,
+}
+
+pub struct SendSmsLogic {
+    repos: Arc<Repositories>,
+    config: Arc<Config>,
+}
+
+impl SendSmsLogic {
+    pub fn new(repos: Arc<Repositories>, config: Arc<Config>) -> Self {
+        Self { repos, config }
+    }
+
+    pub async fn execute(&self, raw: &[u8]) -> anyhow::Result<()> {
+        let payload: SendSmsPayload = match serde_json::from_slice(raw) {
+            Ok(p) => p,
+            Err(e) => {
+                tracing::error!("[SendSmsLogic] deserialise payload: {e}");
+                return Ok(());
+            }
+        };
+
+        let platform = match sms::Platform::from_str(&self.config.mobile.platform) {
+            Some(p) => p,
+            None => {
+                tracing::error!(
+                    "[SendSmsLogic] unsupported SMS platform: {}",
+                    self.config.mobile.platform
+                );
+                return Ok(());
+            }
+        };
+
+        let sms_config: sms::SmsConfig =
+            match serde_json::from_str(&self.config.mobile.platform_config) {
+                Ok(c) => c,
+                Err(e) => {
+                    tracing::error!("[SendSmsLogic] parse platform_config: {e}");
+                    return Ok(());
+                }
+            };
+
+        let sender = sms::create_sender(platform, sms_config);
+
+        let to = format!("+{}{}", payload.telephone_area, payload.telephone);
+        let status: i16 = match sender
+            .send(&payload.telephone_area, &payload.telephone, &payload.content, payload.expire)
+            .await
+        {
+            Ok(()) => {
+                tracing::info!("[SendSmsLogic] sent to {to}");
+                1
+            }
+            Err(e) => {
+                tracing::error!("[SendSmsLogic] send to {to} failed: {e}");
+                2
+            }
+        };
+
+        Telemetry::mobile_message(
+            &self.repos,
+            0,
+            &to,
+            serde_json::json!({ "content": payload.content }),
+            &self.config.mobile.platform,
+            "",
+            status,
+        )
+        .await;
+
+        Ok(())
+    }
+}

src/queue/service/subscription.rs

@@ -0,0 +1,173 @@
+use std::sync::Arc;
+
+use chrono::Utc;
+
+use crate::config::Config;
+use crate::repository::Repositories;
+
+/// Port of `server/queue/logic/subscription/checkSubscriptionLogic.go`.
+///
+/// Two passes:
+///  1. Traffic-exceeded subscribes → mark status=2
+///  2. Expired subscribes → mark status=3
+///
+/// For each affected subscribe, enqueue a SendEmail notification if the user
+/// has an email auth method (best-effort, errors logged and skipped).
+pub struct CheckSubscriptionLogic {
+    repos: Arc<Repositories>,
+    config: Arc<Config>,
+}
+
+impl CheckSubscriptionLogic {
+    pub fn new(repos: Arc<Repositories>, config: Arc<Config>) -> Self {
+        Self { repos, config }
+    }
+
+    pub async fn execute(&self) -> anyhow::Result<()> {
+        let now_ms = Utc::now().timestamp_millis();
+
+        // ── Pass 1: traffic exceeded ─────────────────────────────────────────
+        self.handle_traffic_exceeded(now_ms).await;
+
+        // ── Pass 2: expired ──────────────────────────────────────────────────
+        self.handle_expired(now_ms).await;
+
+        Ok(())
+    }
+
+    // ── traffic exceeded ─────────────────────────────────────────────────────
+
+    async fn handle_traffic_exceeded(&self, now_ms: i64) {
+        let list = match self.repos.user.find_traffic_exceeded_subscribes().await {
+            Ok(l) => l,
+            Err(e) => {
+                tracing::error!("[CheckSubscription/Traffic] find_traffic_exceeded_subscribes: {e}");
+                return;
+            }
+        };
+
+        if list.is_empty() {
+            tracing::info!("[CheckSubscription/Traffic] no traffic-exceeded subscribes");
+            return;
+        }
+
+        let ids: Vec<i64> = list.iter().map(|s| s.id).collect();
+
+        if let Err(e) = self
+            .repos
+            .user
+            .mark_subscribes_finished(&ids, 2, now_ms)
+            .await
+        {
+            tracing::error!("[CheckSubscription/Traffic] mark_subscribes_finished: {e}");
+            return;
+        }
+
+        tracing::info!(
+            "[CheckSubscription/Traffic] marked {} subscribes finished (traffic)",
+            ids.len()
+        );
+
+        // Enqueue notification emails (best-effort)
+        for sub in &list {
+            self.send_notification_email(sub.id, sub.user_id, "traffic").await;
+        }
+    }
+
+    // ── expired ──────────────────────────────────────────────────────────────
+
+    async fn handle_expired(&self, now_ms: i64) {
+        let list = match self.repos.user.find_expired_subscribes(now_ms).await {
+            Ok(l) => l,
+            Err(e) => {
+                tracing::error!("[CheckSubscription/Expire] find_expired_subscribes: {e}");
+                return;
+            }
+        };
+
+        if list.is_empty() {
+            tracing::info!("[CheckSubscription/Expire] no expired subscribes");
+            return;
+        }
+
+        let ids: Vec<i64> = list.iter().map(|s| s.id).collect();
+
+        if let Err(e) = self
+            .repos
+            .user
+            .mark_subscribes_finished(&ids, 3, now_ms)
+            .await
+        {
+            tracing::error!("[CheckSubscription/Expire] mark_subscribes_finished: {e}");
+            return;
+        }
+
+        tracing::info!(
+            "[CheckSubscription/Expire] marked {} subscribes finished (expired)",
+            ids.len()
+        );
+
+        for sub in &list {
+            self.send_notification_email(sub.id, sub.user_id, "expired").await;
+        }
+    }
+
+    // ── notification helper ───────────────────────────────────────────────────
+
+    async fn send_notification_email(&self, subscribe_id: i64, user_id: i64, kind: &str) {
+        // Look up the user's email auth method
+        let auth = match self.repos.user.find_auth_method_by_user_id("email", user_id).await {
+            Ok(Some(a)) => a,
+            Ok(None) => {
+                tracing::info!(
+                    "[CheckSubscription] user {user_id} has no email auth method, skipping"
+                );
+                return;
+            }
+            Err(e) => {
+                tracing::error!(
+                    "[CheckSubscription] find_auth_method_by_user_id(user={user_id}): {e}"
+                );
+                return;
+            }
+        };
+
+        let to = auth.auth_identifier;
+
+        // Build the SendEmail payload matching Go `queue/types.SendEmailPayload`
+        let (email_type, subject) = if kind == "expired" {
+            (3i16, "Subscription Expired")
+        } else {
+            (4i16, "Subscription Traffic Exceeded")
+        };
+
+        let content = serde_json::json!({
+            "SiteLogo": self.config.site.site_logo,
+            "SiteName": self.config.site.site_name,
+        });
+
+        let email_payload = serde_json::json!({
+            "type":    email_type,
+            "email":   to,
+            "subject": subject,
+            "content": content,
+        });
+
+        let raw = match serde_json::to_vec(&email_payload) {
+            Ok(b) => b,
+            Err(e) => {
+                tracing::error!("[CheckSubscription] serialise email payload: {e}");
+                return;
+            }
+        };
+
+        // Execute inline (no asynq client available in this service)
+        let email_logic =
+            super::email::SendEmailLogic::new(self.repos.clone(), self.config.clone());
+        if let Err(e) = email_logic.execute(&raw).await {
+            tracing::error!(
+                "[CheckSubscription] send {kind} email for subscribe {subscribe_id}: {e}"
+            );
+        }
+    }
+}

src/queue/service/task.rs

@@ -0,0 +1,252 @@
+use std::sync::Arc;
+
+use chrono::Utc;
+use serde::Deserialize;
+
+use crate::config::Config;
+use crate::model::entity::log::{GIFT_TYPE_INCREASE, RESET_SUBSCRIBE_TYPE_QUOTA};
+use crate::model::entity::task::{QuotaContent, QuotaScope, Task};
+use crate::repository::Repositories;
+use crate::service::telemetry::Telemetry;
+
+/// Port of `server/queue/logic/task/quotaLogic.go`.
+///
+/// Payload: raw bytes that are a decimal task ID string.
+/// The actual work parameters live in `tasks.scope` / `tasks.content` columns.
+pub struct QuotaTaskLogic {
+    repos: Arc<Repositories>,
+    config: Arc<Config>,
+}
+
+impl QuotaTaskLogic {
+    pub fn new(repos: Arc<Repositories>, config: Arc<Config>) -> Self {
+        Self { repos, config }
+    }
+
+    pub async fn execute(&self, raw: &[u8]) -> anyhow::Result<()> {
+        // ── 1. Parse task ID ────────────────────────────────────────────────
+        let task_id: i64 = match std::str::from_utf8(raw)
+            .ok()
+            .and_then(|s| s.trim().parse().ok())
+        {
+            Some(id) => id,
+            None => {
+                tracing::error!("[QuotaTaskLogic] invalid payload: {:?}", raw);
+                return Ok(());
+            }
+        };
+
+        // ── 2. Fetch task record ────────────────────────────────────────────
+        let mut task = match self.repos.task.find_one(task_id).await {
+            Ok(t) => t,
+            Err(e) => {
+                tracing::error!("[QuotaTaskLogic] find_one({task_id}): {e}");
+                return Ok(());
+            }
+        };
+
+        if task.status != 0 {
+            tracing::info!(
+                "[QuotaTaskLogic] task {task_id} already processed (status={})",
+                task.status
+            );
+            return Ok(());
+        }
+
+        // ── 3. Parse scope + content ────────────────────────────────────────
+        let scope: QuotaScope = match task
+            .scope
+            .as_deref()
+            .and_then(|s| serde_json::from_str(s).ok())
+        {
+            Some(s) => s,
+            None => {
+                tracing::error!("[QuotaTaskLogic] failed to parse scope for task {task_id}");
+                return Ok(());
+            }
+        };
+
+        let content: QuotaContent = match task
+            .content
+            .as_deref()
+            .and_then(|s| serde_json::from_str(s).ok())
+        {
+            Some(c) => c,
+            None => {
+                tracing::error!("[QuotaTaskLogic] failed to parse content for task {task_id}");
+                return Ok(());
+            }
+        };
+
+        // ── 4. Resolve subscriber IDs from scope ────────────────────────────
+        let sub_ids = self.resolve_subscriber_ids(&scope).await;
+
+        // ── 5. Fetch subscribe records ──────────────────────────────────────
+        let subscribes = match self.repos.user.find_subscribes_by_ids(&sub_ids).await {
+            Ok(v) => v,
+            Err(e) => {
+                tracing::error!("[QuotaTaskLogic] find_subscribes_by_ids: {e}");
+                return Ok(());
+            }
+        };
+
+        // ── 6. Process each subscribe ───────────────────────────────────────
+        let now_ms = Utc::now().timestamp_millis();
+        let mut errors: Vec<String> = Vec::new();
+
+        for mut sub in subscribes {
+            let mut updated = false;
+
+            // Extend expiry
+            if let Some(days) = content.days {
+                if days != 0 {
+                    let base = if sub.expire_time == 0 || sub.expire_time < now_ms {
+                        now_ms
+                    } else {
+                        sub.expire_time
+                    };
+                    sub.expire_time =
+                        chrono::DateTime::<Utc>::from_timestamp_millis(base)
+                            .unwrap_or_else(Utc::now)
+                            .checked_add_signed(chrono::Duration::days(days))
+                            .unwrap_or_else(Utc::now)
+                            .timestamp_millis();
+                    if sub.expire_time > now_ms && sub.status != 1 {
+                        sub.status = 1;
+                    }
+                    updated = true;
+                }
+            }
+
+            // Reset traffic
+            if content.reset_traffic {
+                sub.download = 0;
+                sub.upload = 0;
+                updated = true;
+                Telemetry::reset_subscribe(
+                    &self.repos,
+                    sub.user_id,
+                    RESET_SUBSCRIBE_TYPE_QUOTA,
+                    None,
+                )
+                .await;
+            }
+
+            // Gift amount
+            if let (Some(gift_type), Some(gift_value)) = (content.gift_type, content.gift_value) {
+                if gift_value != 0 {
+                    if let Err(e) = self
+                        .process_gift(sub.user_id, sub.id, sub.subscribe_id, gift_type, gift_value)
+                        .await
+                    {
+                        tracing::error!(
+                            "[QuotaTaskLogic] process_gift for subscribe {}: {e}",
+                            sub.id
+                        );
+                        errors.push(format!("subscribe {}: gift error: {e}", sub.id));
+                    }
+                }
+            }
+
+            if updated {
+                sub.updated_at = now_ms;
+                if let Err(e) = self.repos.user.update_subscribe(&sub).await {
+                    tracing::error!("[QuotaTaskLogic] update_subscribe({}): {e}", sub.id);
+                    errors.push(format!("subscribe {}: update error: {e}", sub.id));
+                }
+            }
+        }
+
+        // ── 7. Finalize task record ─────────────────────────────────────────
+        let all_failed = !errors.is_empty() && errors.len() >= sub_ids.len();
+        task.status = if all_failed { 3 } else { 2 };
+        task.current = sub_ids.len() as i64;
+        if !errors.is_empty() {
+            task.errors = serde_json::to_string(&errors).ok();
+        }
+        task.updated_at = now_ms;
+
+        if let Err(e) = self.repos.task.update(&task).await {
+            tracing::error!("[QuotaTaskLogic] update task {task_id}: {e}");
+        }
+
+        Ok(())
+    }
+
+    // ── helper: resolve subscriber IDs from scope ─────────────────────────────
+
+    async fn resolve_subscriber_ids(&self, scope: &QuotaScope) -> Vec<i64> {
+        // Direct list wins
+        if !scope.recipients.is_empty() {
+            return scope.recipients.clone();
+        }
+
+        // Filter by active/expired status
+        use crate::repository::user::SubscribeFilter;
+        let filter = SubscribeFilter {
+            subscribers: scope.subscribers.clone(),
+            is_active: scope.is_active,
+            start_time: scope.start_time,
+            end_time: scope.end_time,
+        };
+
+        match self.repos.user.query_subscribe_ids_by_filter(&filter).await {
+            Ok(ids) => ids,
+            Err(e) => {
+                tracing::error!("[QuotaTaskLogic] query_subscribe_ids_by_filter: {e}");
+                vec![]
+            }
+        }
+    }
+
+    // ── helper: apply gift to user balance ────────────────────────────────────
+
+    async fn process_gift(
+        &self,
+        user_id: i64,
+        subscribe_id: i64,
+        plan_subscribe_id: i64,
+        gift_type: i16,
+        gift_value: i64,
+    ) -> anyhow::Result<()> {
+        let mut user = self.repos.user.find_one_user(user_id).await?;
+
+        let gift_amount: i64 = match gift_type {
+            1 => gift_value,
+            2 => {
+                // Percentage of plan unit price
+                let plan = self.repos.subscribe.find_one(plan_subscribe_id).await?;
+                if plan.unit_price > 0 {
+                    (plan.unit_price as f64 * (gift_value as f64 / 100.0)) as i64
+                } else {
+                    0
+                }
+            }
+            other => {
+                return Err(anyhow::anyhow!("invalid gift_type {other}"));
+            }
+        };
+
+        if gift_amount <= 0 {
+            return Ok(());
+        }
+
+        user.gift_amount += gift_amount;
+        user.updated_at = Utc::now().timestamp_millis();
+        let updated = self.repos.user.update_user(&user).await?;
+
+        Telemetry::gift(
+            &self.repos,
+            user_id,
+            GIFT_TYPE_INCREASE,
+            "",
+            subscribe_id,
+            gift_amount,
+            updated.gift_amount,
+            Some("Quota task gift".to_string()),
+        )
+        .await;
+
+        Ok(())
+    }
+}

src/queue/service/traffic.rs

@@ -0,0 +1,266 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use chrono::Utc;
+use serde::{Deserialize, Serialize};
+
+use crate::config::Config;
+use crate::model::entity::log::{RESET_SUBSCRIBE_TYPE_AUTO, ServerTraffic, UserTraffic};
+use crate::model::entity::traffic::TrafficLog;
+use crate::repository::Repositories;
+use crate::service::telemetry::Telemetry;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct UserTrafficEntry {
+    #[serde(rename = "uid")]
+    pub sid: i64,
+    pub upload: i64,
+    pub download: i64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TrafficStatisticsPayload {
+    pub server_id: i64,
+    pub protocol: String,
+    pub logs: Vec<UserTrafficEntry>,
+}
+
+pub struct ResetTrafficLogic {
+    repos: Arc<Repositories>,
+}
+
+impl ResetTrafficLogic {
+    pub fn new(repos: Arc<Repositories>) -> Self {
+        Self { repos }
+    }
+
+    pub async fn execute(&self) -> anyhow::Result<()> {
+        let now_ms = Utc::now().timestamp_millis();
+        let now_ts = Utc::now().timestamp();
+        self.reset_by_cycle(3, now_ms, now_ts, "yearly").await;
+        self.reset_by_cycle(1, now_ms, now_ts, "first-of-month").await;
+        self.reset_by_cycle(2, now_ms, now_ts, "monthly").await;
+        Ok(())
+    }
+
+    async fn reset_by_cycle(&self, reset_cycle: i64, now_ms: i64, now_ts: i64, label: &str) {
+        let sub_ids = match self.repos.subscribe.query_reset_cycle_subscribe_ids(reset_cycle).await {
+            Ok(ids) => ids,
+            Err(e) => {
+                tracing::error!("[ResetTraffic] query_reset_cycle_subscribe_ids({label}) failed: {e}");
+                return;
+            }
+        };
+        if sub_ids.is_empty() {
+            return;
+        }
+        let user_sub_ids: Vec<i64> = match reset_cycle {
+            1 => match self.repos.user.query_first_reset_subscribe_ids(&sub_ids, now_ts).await {
+                Ok(v) => v,
+                Err(e) => { tracing::error!("[ResetTraffic] query_first_reset_subscribe_ids failed: {e}"); return; }
+            },
+            2 => match self.repos.user.query_monthly_reset_subscribe_ids(&sub_ids, now_ms).await {
+                Ok(v) => v,
+                Err(e) => { tracing::error!("[ResetTraffic] query_monthly_reset_subscribe_ids failed: {e}"); return; }
+            },
+            3 => match self.repos.user.query_yearly_reset_subscribe_ids(&sub_ids, now_ts).await {
+                Ok(v) => v,
+                Err(e) => { tracing::error!("[ResetTraffic] query_yearly_reset_subscribe_ids failed: {e}"); return; }
+            },
+            _ => return,
+        };
+        if user_sub_ids.is_empty() {
+            return;
+        }
+        if let Err(e) = self.repos.user.reset_subscribe_traffic_by_ids(&user_sub_ids).await {
+            tracing::error!("[ResetTraffic] reset_subscribe_traffic_by_ids({label}) failed: {e}");
+            return;
+        }
+        tracing::info!("[ResetTraffic] {label} reset: {} user-subscribes", user_sub_ids.len());
+        let subs = match self.repos.user.find_subscribes_by_ids(&user_sub_ids).await {
+            Ok(v) => v,
+            Err(e) => { tracing::error!("[ResetTraffic] find_subscribes_by_ids({label}) failed: {e}"); return; }
+        };
+        for sub in &subs {
+            Telemetry::reset_subscribe(&self.repos, sub.user_id, RESET_SUBSCRIBE_TYPE_AUTO, None).await;
+        }
+    }
+}
+
+pub struct ServerDataLogic {
+    repos: Arc<Repositories>,
+    cache: Arc<crate::cache::Cache>,
+}
+
+impl ServerDataLogic {
+    pub fn new(repos: Arc<Repositories>, cache: Arc<crate::cache::Cache>) -> Self {
+        Self { repos, cache }
+    }
+
+    pub async fn execute(&self) -> anyhow::Result<()> {
+        let now = Utc::now();
+        let today_ms = now.timestamp_millis();
+        let yesterday_ms = (now - chrono::Duration::days(1)).timestamp_millis();
+
+        let top_servers_today = self.repos.traffic.top_servers_traffic_by_day(today_ms, 10).await.unwrap_or_else(|e| { tracing::error!("[ServerData] top_servers today: {e}"); vec![] });
+        let top_users_today = self.repos.traffic.top_users_traffic_by_day(today_ms, 10).await.unwrap_or_else(|e| { tracing::error!("[ServerData] top_users today: {e}"); vec![] });
+        let top_servers_yesterday = self.repos.traffic.top_servers_traffic_by_day(yesterday_ms, 10).await.unwrap_or_else(|e| { tracing::error!("[ServerData] top_servers yesterday: {e}"); vec![] });
+
+        let mut server_rank_today: HashMap<u8, ServerTraffic> = HashMap::new();
+        for (i, s) in top_servers_today.iter().enumerate().take(10) {
+            server_rank_today.insert((i + 1) as u8, ServerTraffic { server_id: s.server_id, upload: s.upload, download: s.download, total: s.total });
+        }
+        let mut server_rank_yesterday: HashMap<u8, ServerTraffic> = HashMap::new();
+        for (i, s) in top_servers_yesterday.iter().enumerate().take(10) {
+            server_rank_yesterday.insert((i + 1) as u8, ServerTraffic { server_id: s.server_id, upload: s.upload, download: s.download, total: s.total });
+        }
+        let mut user_rank_today: HashMap<u8, UserTraffic> = HashMap::new();
+        for (i, u) in top_users_today.iter().enumerate().take(10) {
+            user_rank_today.insert((i + 1) as u8, UserTraffic { subscribe_id: u.subscribe_id, user_id: u.user_id, upload: u.upload, download: u.download, total: u.total });
+        }
+
+        let daily = self.repos.traffic.query_traffic_by_day(today_ms).await.unwrap_or_default();
+        let monthly = self.repos.traffic.query_traffic_by_monthly(today_ms).await.unwrap_or_default();
+
+        let snapshot = serde_json::json!({
+            "server_traffic_ranking_today": server_rank_today,
+            "server_traffic_ranking_yesterday": server_rank_yesterday,
+            "user_traffic_ranking_today": user_rank_today,
+            "today_upload": daily.upload,
+            "today_download": daily.download,
+            "monthly_upload": monthly.upload,
+            "monthly_download": monthly.download,
+            "updated_at": today_ms,
+        });
+        let json = serde_json::to_string(&snapshot)?;
+        if let Err(e) = self.cache.set_ex("server_count", &json, -1).await {
+            tracing::error!("[ServerData] cache set failed: {e}");
+        }
+        Telemetry::server_traffic_rank(&self.repos, server_rank_today).await;
+        tracing::info!("[ServerData] snapshot updated");
+        Ok(())
+    }
+}
+
+pub struct StatLogic {
+    repos: Arc<Repositories>,
+    config: Arc<Config>,
+}
+
+impl StatLogic {
+    pub fn new(repos: Arc<Repositories>, config: Arc<Config>) -> Self {
+        Self { repos, config }
+    }
+
+    pub async fn execute(&self) -> anyhow::Result<()> {
+        let now = Utc::now();
+        let yesterday = now - chrono::Duration::days(1);
+        let start_ms = yesterday.date_naive().and_hms_opt(0, 0, 0).unwrap().and_utc().timestamp_millis();
+        let end_ms = yesterday.date_naive().and_hms_opt(23, 59, 59).unwrap().and_utc().timestamp_millis() + 999;
+
+        let user_traffic = self.repos.traffic.query_user_traffic_ranking(start_ms, end_ms).await
+            .map_err(|e| { tracing::error!("[StatLogic] query_user_traffic_ranking: {e}"); e })?;
+
+        let mut user_rank: HashMap<u8, UserTraffic> = HashMap::new();
+        for (i, row) in user_traffic.iter().enumerate() {
+            let item = UserTraffic { subscribe_id: row.subscribe_id, user_id: row.user_id, upload: row.upload, download: row.download, total: row.total };
+            if i < 10 { user_rank.insert((i + 1) as u8, item.clone()); }
+            Telemetry::subscribe_traffic(&self.repos, row.subscribe_id, row.download, row.upload).await;
+        }
+        Telemetry::user_traffic_rank(&self.repos, user_rank).await;
+
+        let server_traffic = self.repos.traffic.query_server_traffic_ranking(start_ms, end_ms).await
+            .map_err(|e| { tracing::error!("[StatLogic] query_server_traffic_ranking: {e}"); e })?;
+
+        let mut server_rank: HashMap<u8, ServerTraffic> = HashMap::new();
+        for (i, row) in server_traffic.iter().enumerate() {
+            let item = ServerTraffic { server_id: row.server_id, upload: row.upload, download: row.download, total: row.total };
+            if i < 10 { server_rank.insert((i + 1) as u8, item.clone()); }
+            Telemetry::server_traffic(&self.repos, row.server_id, row.download, row.upload).await;
+        }
+        Telemetry::server_traffic_rank(&self.repos, server_rank).await;
+
+        let summary = self.repos.traffic.query_traffic_summary(start_ms, end_ms).await
+            .map_err(|e| { tracing::error!("[StatLogic] query_traffic_summary: {e}"); e })?;
+        Telemetry::traffic_stat(&self.repos, summary.upload, summary.download).await;
+
+        if self.config.log.auto_clear {
+            let cutoff = (now - chrono::Duration::days(self.config.log.clear_days as i64)).timestamp_millis();
+            if let Err(e) = self.repos.traffic.delete_before(cutoff).await {
+                tracing::error!("[StatLogic] delete_before: {e}");
+            }
+        }
+        tracing::info!("[StatLogic] daily stat ↑{} ↓{}", summary.upload, summary.download);
+        Ok(())
+    }
+}
+
+pub struct TrafficStatisticsLogic {
+    repos: Arc<Repositories>,
+    config: Arc<Config>,
+}
+
+impl TrafficStatisticsLogic {
+    pub fn new(repos: Arc<Repositories>, config: Arc<Config>) -> Self {
+        Self { repos, config }
+    }
+
+    pub async fn execute(&self, payload: TrafficStatisticsPayload) -> anyhow::Result<()> {
+        if payload.logs.is_empty() {
+            return Ok(());
+        }
+        let server = match self.repos.node.find_one_server(payload.server_id).await {
+            Ok(s) => s,
+            Err(e) => { tracing::error!("[TrafficStatistics] find_one_server({}): {e}", payload.server_id); return Ok(()); }
+        };
+        let ratio = self.resolve_ratio(&server, &payload.protocol);
+        let threshold = self.config.node.traffic_report_threshold;
+        let now_ms = Utc::now().timestamp_millis();
+
+        for entry in &payload.logs {
+            if entry.sid == 0 {
+                tracing::warn!("[TrafficStatistics] entry sid=0, skipping");
+                continue;
+            }
+            if entry.upload + entry.download <= threshold {
+                continue;
+            }
+            let sub = match self.repos.user.find_one_subscribe(entry.sid).await {
+                Ok(s) => s,
+                Err(e) => { tracing::warn!("[TrafficStatistics] find_one_subscribe({}): {e}", entry.sid); continue; }
+            };
+            let d = (entry.download as f64 * ratio) as i64;
+            let u = (entry.upload as f64 * ratio) as i64;
+            if let Err(e) = self.repos.user.update_user_subscribe_with_traffic(sub.id, d, u).await {
+                tracing::warn!("[TrafficStatistics] update_user_subscribe_with_traffic({}): {e}", sub.id);
+                continue;
+            }
+            let log = TrafficLog {
+                id: 0,
+                server_id: payload.server_id,
+                user_id: sub.user_id,
+                subscribe_id: sub.subscribe_id,
+                upload: u,
+                download: d,
+                timestamp: now_ms,
+            };
+            if let Err(e) = self.repos.traffic.insert(&log).await {
+                tracing::warn!("[TrafficStatistics] traffic insert(sid={}): {e}", entry.sid);
+            }
+            Telemetry::subscribe_traffic(&self.repos, entry.sid, d, u).await;
+            Telemetry::server_traffic(&self.repos, payload.server_id, d, u).await;
+        }
+        Ok(())
+    }
+
+    fn resolve_ratio(&self, server: &crate::model::entity::node::Server, protocol: &str) -> f64 {
+        let protocols: Vec<crate::model::entity::node::Protocol> =
+            serde_json::from_str(&server.protocols).unwrap_or_default();
+        for p in &protocols {
+            if p.type_.eq_ignore_ascii_case(protocol) && p.ratio > 0.0 {
+                return p.ratio;
+            }
+        }
+        1.0
+    }
+}

src/queue/types.rs

@@ -0,0 +1,30 @@
+/// Task type constants, ported from `server/queue/types/*.go`.
+///
+/// Prefix convention (matching Go):
+/// - `scheduler:` — periodic tasks registered by the scheduler
+/// - `forthwith:` — tasks enqueued immediately by HTTP handlers
+/// - `defer:`     — tasks enqueued with a delay
+/// - `scheduled:` — tasks enqueued for a specific future time
+
+// ── scheduler ──────────────────────────────────────────────────────────
+pub const SCHEDULER_CHECK_SUBSCRIPTION: &str = "scheduler:check:subscription";
+pub const SCHEDULER_TOTAL_SERVER_DATA: &str = "scheduler:total:server";
+pub const SCHEDULER_RESET_TRAFFIC: &str = "scheduler:reset:traffic";
+pub const SCHEDULER_TRAFFIC_STAT: &str = "scheduler:traffic:stat";
+
+// ── order ──────────────────────────────────────────────────────────────
+pub const FORTHWITH_ACTIVATE_ORDER: &str = "forthwith:activate:order";
+pub const DEFER_CLOSE_ORDER: &str = "defer:close:order";
+
+// ── email ──────────────────────────────────────────────────────────────
+pub const FORTHWITH_SEND_EMAIL: &str = "forthwith:send:email";
+pub const SCHEDULED_BATCH_SEND_EMAIL: &str = "scheduled:batch:send:email";
+
+// ── sms ────────────────────────────────────────────────────────────────
+pub const FORTHWITH_SEND_SMS: &str = "forthwith:sms:send";
+
+// ── server / traffic ───────────────────────────────────────────────────
+pub const FORTHWITH_TRAFFIC_STATISTICS: &str = "forthwith:traffic:statistics";
+
+// ── task / quota ───────────────────────────────────────────────────────
+pub const FORTHWITH_QUOTA_TASK: &str = "forthwith:quota:task";