diff --git a/README.en.md b/README.en.md
index 994ae131..800fcddb 100644
--- a/README.en.md
+++ b/README.en.md
@@ -52,6 +52,8 @@
MAX API is an AI model governance, AgentOps, and application service infrastructure project initiated, maintained, and operated long term by AGI enthusiasts from research institutions and universities. It provides developers, researchers, teams, and organizations with a stable and reusable service layer. The project focuses on the operational problems that appear after AI applications move into real use: more models, frequent upstream API changes, longer Agent call chains, and rising pressure around cost and auditability. MAX API provides a unified access, authentication, routing, billing, observability, and governance layer between applications, Agents, users, organizations, and upstream model providers, helping AI applications run with greater stability and control.
+In practice, MAX API is more than a request forwarder. It is an operable gateway for AI-ready applications and Agent workloads, bringing protocol normalization, provider differences, traffic bursts, long streaming responses, large request bodies, multi-node cache, cost audit, and performance observability under one governance boundary.
+
Ongoing investment areas:
- **AI model governance**: continuously tracks model updates, API changes, parameter differences, pricing rules, and task protocols across OpenAI, Azure OpenAI, AWS Bedrock, Vertex AI, Ollama, and domestic platforms such as DeepSeek, Qwen / Alibaba Cloud Model Studio, Zhipu GLM, Kimi, Doubao / Volcano Engine, Tencent Hunyuan, Baidu ERNIE / Qianfan, iFlytek Spark, MiniMax, 01.AI, and SiliconFlow. It also tracks application and multimodal ecosystems such as Dify, RAGFlow, Kling, and Seedance, bringing distributed model capabilities into unified governance through channels, model mapping, protocol conversion, path overrides, and configurable task protocols.
@@ -83,6 +85,7 @@ In the AGI application era, MAX API focuses on open AI model governance and AI A
- **Channel configuration plane**: reduces misconfiguration risk when adding upstream channels, migrating providers, or maintaining non-standard APIs through capability matrices, form validation, model discovery, and protocol templates.
- **Protocol and provider adaptation layer**: continuously tracks official overseas APIs, domestic model platform APIs, and OpenAI-compatible / non-standard interface changes, then normalizes them into stable application-side APIs.
- **Cost, quota, and reliability governance**: supports channel routing, weighted distribution, retry, rate limiting, pre-charge, failure refund, expression billing, fixed pricing, task rate-cards, multiplier billing, and usage statistics.
+- **Performance and scalability governance**: uses Redis / in-memory cache, model request rate limits, streaming timeouts and large-response buffers, request body limits, disk cache, Pyroscope profiling, and graceful shutdown to support stable single-node and multi-node deployments.
- **Organization operations and audit layer**: provides user management, group management, private deployment, data retention, audit, and continuous operations optimization for teams, research institutions, enterprises, and community services.
- **Reusable governance templates**: accumulates channel templates, task protocol templates, pricing configuration, deployment practices, and operations experience to reduce onboarding cost for new models, providers, and Agent scenarios.
@@ -195,6 +198,17 @@ docker compose up -d
- Supports weighted channel routing, failure retry, disabled-channel bypass, and model-level routing to reduce upstream impact on applications and Agents.
- Supports Redis and in-memory cache for single-node and multi-node deployments.
+### Performance and Scalability Governance
+
+| Capability | Description |
+|------|------|
+| Cache and multi-node scaling | Single-node deployments can use in-memory cache, while multi-node deployments can use Redis; user, token, channel affinity, and quota-related caches reduce repeated database reads, while `SESSION_SECRET`, `CRYPTO_SECRET`, and `NODE_NAME` keep sessions, encryption, and log attribution consistent |
+| Rate limits and capacity protection | Supports global API / Web rate limits, critical endpoint limits, search limits, model request limits, and group-specific model request quotas; counters can use Redis or memory |
+| Streaming and large-request controls | Supports `STREAMING_TIMEOUT`, `STREAM_SCANNER_MAX_BUFFER_MB`, `MAX_REQUEST_BODY_MB`, `MAX_FILE_DOWNLOAD_MB`, and related settings for long streams, large SSE lines, decompressed request bodies, and remote file downloads |
+| Relay connection tuning | Supports `RELAY_TIMEOUT`, `RELAY_IDLE_CONN_TIMEOUT`, `RELAY_MAX_IDLE_CONNS`, and `RELAY_MAX_IDLE_CONNS_PER_HOST` for upstream HTTP timeout and connection-pool behavior |
+| Disk cache and performance observability | System performance settings can enable disk cache for large request bodies and configure cache threshold and capacity; operations endpoints can inspect / clear disk cache, and Pyroscope can collect CPU, memory, goroutine, mutex, and block profiles |
+| Graceful shutdown and data flush | Shutdown supports `SHUTDOWN_TIMEOUT_SECONDS` and `QUOTA_DATA_CACHE_SAVE_TIMEOUT_SECONDS` so the process can close HTTP handling and save quota cache before exit when possible |
+
### Security and Organization Management
- Supports JWT, WebAuthn/Passkeys, OAuth, OIDC, Telegram, Discord, LinuxDO, and other login methods.
@@ -213,6 +227,7 @@ docker compose up -d
| Agent access | Agents directly hold upstream keys, making revocation and quota control difficult | Assign independent tokens to Agents with model, quota, expiration, and group limits |
| Protocol differences | Applications adapt Claude, Gemini, Responses, and other formats themselves | Gateway handles protocol conversion and provider adaptation |
| Failure handling | Applications implement retry, fallback, and error normalization themselves | Channel failure retry, weighted routing, and error handling are built in |
+| Performance and scaling | Applications handle timeouts, rate limits, connection pools, and cache themselves | Gateway centralizes streaming timeouts, request limits, Redis / in-memory cache, connection-pool tuning, and performance observability |
| Cost statistics | Bills are scattered across providers and hard to attribute by user or Agent | Unified quota, billing, usage statistics, and consumption logs, attributable by token and model |
| Audit boundary | Application-side logging is fragmented and retention/permission policies differ | Unified admin audit entry with normal user logs filtering admin-only fields |
| Private deployment | Keys, logs, and billing strategies are scattered | Self-hosted control over keys, data, logs, and policies |
@@ -476,6 +491,7 @@ A task rate-card can match prices by request parameters:
| Remote database | MySQL ≥ 5.7.8 or PostgreSQL ≥ 9.6 |
| Cache | In-memory cache for single-node deployments; Redis recommended for multi-node deployments |
| Frontend build | Bun workspace; keep `web/package.json` and `web/bun.lock` |
+| Source build | Use the Go version declared in `go.mod` (currently Go 1.25.1+) with the repository `go.sum`; after dependency or security updates, run `go mod download`, `go mod verify`, and rebuild |
### Recommended Environment Variables
@@ -548,6 +564,9 @@ cd MAX-API
docker build -t cscitechtop/max-api:latest .
```
+> [!NOTE]
+> `Dockerfile` downloads Go modules during the image build. For host builds or dependency/security updates, keep `go.mod` and `go.sum` committed together, run `go mod download && go mod verify`, then rebuild the binary or image; use `docker build --pull --no-cache -t cscitechtop/max-api:latest .` when base images need to be refreshed.
+
> [!TIP]
> The frontend uses Bun workspace. The build context must keep `web/package.json`, `web/bun.lock`, and `web/default/package.json`; otherwise `catalog:` dependencies cannot be resolved.
diff --git a/README.fr.md b/README.fr.md
index 1d60fc66..fd77e9fa 100644
--- a/README.fr.md
+++ b/README.fr.md
@@ -44,6 +44,8 @@
MAX API est un projet d'infrastructure pour la gouvernance des modèles IA, l'AgentOps et les services applicatifs. Il est initié, maintenu et exploité sur le long terme par des passionnés d'AGI issus d'organismes de recherche et d'universités. Il fournit aux développeurs, chercheurs, équipes et organisations une couche de service stable et réutilisable. Le projet se concentre sur les problèmes d'exploitation qui apparaissent lorsque les applications IA passent en production : multiplication des modèles, évolution fréquente des API amont, chaînes d'appels Agent plus longues, pression accrue sur les coûts et l'audit. MAX API fournit une couche unifiée d'accès, d'authentification, de routage, de facturation, d'observabilité et de gouvernance entre applications, Agents, utilisateurs, organisations et fournisseurs de modèles.
+En pratique, MAX API n'est pas seulement un proxy de requêtes. C'est une passerelle exploitable pour applications AI-ready et charges Agent, qui regroupe normalisation des protocoles, différences fournisseurs, pics de trafic, longs flux streaming, gros corps de requête, cache multi-nœud, audit des coûts et observabilité des performances dans une même frontière de gouvernance.
+
Axes d'investissement continus :
- **Gouvernance des modèles IA** : suivi continu des mises à jour de modèles, des changements d'API, des différences de paramètres, des règles de prix et des protocoles de tâches chez OpenAI, Azure OpenAI, AWS Bedrock, Vertex AI, Ollama, ainsi que chez DeepSeek, Qwen / Alibaba Cloud Model Studio, Zhipu GLM, Kimi, Doubao / Volcano Engine, Tencent Hunyuan, Baidu ERNIE / Qianfan, iFlytek Spark, MiniMax, 01.AI, SiliconFlow et d'autres plateformes. Le projet suit aussi Dify, RAGFlow, Kling, Seedance et d'autres écosystèmes applicatifs ou multimodaux.
@@ -75,6 +77,7 @@ Pour la production, privilégiez les versions stables. Utilisez les versions Pre
- **Plan de configuration des canaux** : réduit les risques de mauvaise configuration grâce aux matrices de capacités, validations de formulaires, découverte de modèles et modèles de protocole.
- **Couche d'adaptation protocoles/fournisseurs** : suit les API officielles internationales, les plateformes de modèles chinoises et les API OpenAI-compatible / non standard, puis les normalise en interfaces applicatives stables.
- **Gouvernance coût, quota et fiabilité** : routage de canaux, distribution pondérée, retry, limitation, préfacturation, remboursement d'échec, facturation par expression, prix fixes, rate-cards de tâches, multiplicateurs et statistiques d'usage.
+- **Gouvernance performance et scalabilité** : cache Redis / mémoire, limitation des requêtes modèles, timeouts streaming, tampons pour grandes réponses, limites de corps de requête, cache disque, profiling Pyroscope et arrêt gracieux pour les déploiements mono ou multi-nœuds.
- **Couche d'exploitation et d'audit organisationnelle** : utilisateurs, groupes, déploiement privé, rétention des données, audit et optimisation continue.
- **Modèles de gouvernance réutilisables** : capitalisation des templates de canaux, protocoles de tâches, configurations de prix, pratiques de déploiement et retours d'exploitation.
@@ -187,6 +190,17 @@ docker compose up -d
- Routage pondéré, retry, contournement des canaux désactivés et routage par modèle.
- Cache Redis et mémoire pour déploiement mono ou multi-nœud.
+### Gouvernance performance et scalabilité
+
+| Capacité | Description |
+|------|------|
+| Cache et extension multi-nœud | Déploiement mono-nœud avec cache mémoire, multi-nœud avec Redis ; les caches utilisateur, jeton, affinité de canal et quota réduisent les lectures répétées en base, tandis que `SESSION_SECRET`, `CRYPTO_SECRET` et `NODE_NAME` gardent sessions, chiffrement et attribution des journaux cohérents |
+| Limitation et protection de capacité | Limites globales API / Web, limites d'endpoints critiques, recherche, requêtes modèles et quotas par groupe ; compteurs via Redis ou mémoire |
+| Streaming et grandes requêtes | `STREAMING_TIMEOUT`, `STREAM_SCANNER_MAX_BUFFER_MB`, `MAX_REQUEST_BODY_MB`, `MAX_FILE_DOWNLOAD_MB` contrôlent longs flux, grandes lignes SSE, corps décompressés et téléchargements distants |
+| Réglage des connexions relais | `RELAY_TIMEOUT`, `RELAY_IDLE_CONN_TIMEOUT`, `RELAY_MAX_IDLE_CONNS` et `RELAY_MAX_IDLE_CONNS_PER_HOST` configurent timeouts et pool HTTP amont |
+| Cache disque et observabilité | Les paramètres de performance peuvent activer le cache disque pour grands corps, définir seuil et capacité ; les endpoints ops peuvent inspecter / nettoyer ce cache, et Pyroscope collecte CPU, mémoire, goroutines, mutex et block profiles |
+| Arrêt gracieux et persistance | `SHUTDOWN_TIMEOUT_SECONDS` et `QUOTA_DATA_CACHE_SAVE_TIMEOUT_SECONDS` aident à fermer HTTP et sauvegarder le cache quota avant sortie quand c'est possible |
+
### Sécurité et gestion d'organisation
- JWT, WebAuthn/Passkeys, OAuth, OIDC, Telegram, Discord, LinuxDO et autres méthodes de connexion.
@@ -205,6 +219,7 @@ docker compose up -d
| Accès Agent | Les Agents détiennent directement les clés amont | Jetons indépendants avec limites de modèles, quota, expiration et groupe |
| Différences de protocole | L'application adapte Claude, Gemini, Responses, etc. | La passerelle convertit les protocoles et adapte les fournisseurs |
| Échecs | Retry, fallback et normalisation d'erreurs côté application | Retry de canaux, routage pondéré et traitement d'erreurs intégrés |
+| Performance et extension | L'application gère timeouts, limites, pools de connexions et cache | La passerelle centralise timeouts streaming, limites de requêtes, cache Redis / mémoire, réglage de pool et observabilité |
| Coûts | Factures dispersées, attribution difficile | Quotas, facturation, statistiques et journaux unifiés par jeton et modèle |
| Audit | Journaux applicatifs dispersés | Entrée d'audit admin unifiée, filtrage des champs admin pour utilisateurs |
| Privé | Clés, journaux et stratégie tarifaire dispersés | Auto-hébergement et contrôle des clés, données, journaux et politiques |
@@ -413,6 +428,7 @@ Les modèles vidéo comme Seedance 2.0 peuvent utiliser la résolution, l'entré
| Base distante | MySQL ≥ 5.7.8 ou PostgreSQL ≥ 9.6 |
| Cache | Mémoire en mono-nœud, Redis recommandé en multi-nœud |
| Build frontend | Bun workspace, conserver `web/package.json` et `web/bun.lock` |
+| Build source | Utiliser la version de Go déclarée dans `go.mod` (actuellement Go 1.25.1+) avec `go.sum` ; après une mise à jour de dépendances ou de sécurité, exécuter `go mod download`, `go mod verify`, puis reconstruire |
### Variables d'environnement recommandées
@@ -477,6 +493,9 @@ cd MAX-API
docker build -t cscitechtop/max-api:latest .
```
+> [!NOTE]
+> `Dockerfile` télécharge les modules Go pendant la construction de l'image. Pour une construction locale ou après une mise à jour de dépendances / sécurité, conservez `go.mod` et `go.sum` ensemble, exécutez `go mod download && go mod verify`, puis reconstruisez le binaire ou l'image ; utilisez `docker build --pull --no-cache -t cscitechtop/max-api:latest .` lorsque les images de base doivent être rafraîchies.
+
> [!TIP]
> Le frontend utilise Bun workspace. Conservez `web/package.json`, `web/bun.lock` et `web/default/package.json`, sinon les dépendances `catalog:` ne seront pas résolues.
diff --git a/README.ja.md b/README.ja.md
index b76c4219..8487179f 100644
--- a/README.ja.md
+++ b/README.ja.md
@@ -44,6 +44,8 @@
MAX API は、研究機関と大学に所属する AGI 愛好者によって立ち上げられ、長期的に保守・運営されている AI モデルガバナンス、AgentOps、アプリケーションサービス基盤プロジェクトです。開発者、研究者、チーム、組織に対して、安定して再利用可能なサービスレイヤーを提供します。AI アプリケーションが実運用に入ると、モデル数の増加、上流 API の頻繁な変更、Agent の呼び出しチェーンの長大化、コストと監査の負担増加が発生します。MAX API は、アプリケーション、Agent、ユーザー、組織、上流モデルサービスの間に、アクセス、認証、ルーティング、課金、可観測性、ガバナンスの統一レイヤーを提供します。
+実運用では、MAX API は単なるリクエスト転送ではありません。AI-ready アプリケーションと Agent ワークロード向けの運用可能なゲートウェイとして、プロトコル正規化、プロバイダー差分、トラフィック集中、長時間ストリーミング、大きなリクエストボディ、複数ノードキャッシュ、コスト監査、性能観測を同じガバナンス境界にまとめます。
+
継続的に注力する領域:
- **AI モデルガバナンス**:OpenAI、Azure OpenAI、AWS Bedrock、Vertex AI、Ollama、および DeepSeek、Qwen / Alibaba Cloud Model Studio、Zhipu GLM、Kimi、Doubao / Volcano Engine、Tencent Hunyuan、Baidu ERNIE / Qianfan、iFlytek Spark、MiniMax、01.AI、SiliconFlow などのモデル更新、API 変更、パラメータ差分、価格ルール、タスクプロトコルを継続的に追跡します。Dify、RAGFlow、Kling、Seedance などのアプリケーション・マルチモーダルエコシステムも対象です。
@@ -75,6 +77,7 @@ AGI アプリケーション時代において、MAX API はオープンな AI
- **チャネル設定プレーン**:能力マトリクス、フォーム検証、モデル発見、プロトコルテンプレートによって、上流チャネル追加や非標準 API の保守における設定ミスを減らします。
- **プロトコル・プロバイダー適応レイヤー**:海外公式 API、中国国内モデルプラットフォーム API、OpenAI-compatible / 非標準 API の変化を追跡し、安定したアプリケーション側インターフェースへ正規化します。
- **コスト、クォータ、信頼性ガバナンス**:チャネルルーティング、重み付き分配、retry、レート制限、事前課金、失敗時返金、式ベース課金、固定価格、タスク rate-card、倍率課金、利用統計をサポートします。
+- **性能とスケーラビリティガバナンス**:Redis / メモリキャッシュ、モデルリクエスト制限、ストリーミング timeout、大きな応答バッファ、リクエストボディ上限、ディスクキャッシュ、Pyroscope profiling、graceful shutdown により、単一ノードから複数ノードまで安定運用を支えます。
- **組織運用・監査レイヤー**:チーム、研究機関、企業、コミュニティ向けに、ユーザー管理、グループ管理、プライベートデプロイ、データ保持、監査、継続運用最適化を提供します。
- **再利用可能なガバナンステンプレート**:チャネルテンプレート、タスクプロトコルテンプレート、価格設定、デプロイ実践、運用ノウハウを蓄積します。
@@ -187,6 +190,17 @@ docker compose up -d
- 重み付きチャネルルーティング、失敗時 retry、無効チャネル回避、モデルレベルルーティングをサポートします。
- Redis キャッシュとメモリキャッシュにより、単一ノード・複数ノード構成に対応します。
+### 性能とスケーラビリティガバナンス
+
+| 機能 | 説明 |
+|------|------|
+| キャッシュと複数ノード拡張 | 単一ノードではメモリキャッシュ、複数ノードでは Redis を利用できます。ユーザー、トークン、チャネル affinity、クォータ関連キャッシュで DB 読み取りを減らし、`SESSION_SECRET`、`CRYPTO_SECRET`、`NODE_NAME` によりセッション、暗号化、ログ帰属を揃えます |
+| レート制限と容量保護 | グローバル API / Web、重要 endpoint、検索、モデルリクエスト、グループ別モデルリクエスト quota の制限をサポートし、Redis またはメモリカウンターを利用できます |
+| ストリーミングと大きなリクエスト制御 | `STREAMING_TIMEOUT`、`STREAM_SCANNER_MAX_BUFFER_MB`、`MAX_REQUEST_BODY_MB`、`MAX_FILE_DOWNLOAD_MB` で長時間 stream、大きな SSE 行、解凍後リクエストボディ、リモートファイル download を制御します |
+| リレー接続調整 | `RELAY_TIMEOUT`、`RELAY_IDLE_CONN_TIMEOUT`、`RELAY_MAX_IDLE_CONNS`、`RELAY_MAX_IDLE_CONNS_PER_HOST` で上流 HTTP timeout と connection pool を調整できます |
+| ディスクキャッシュと性能観測 | 性能設定で大きなリクエストボディ向けディスクキャッシュ、閾値、容量を設定できます。運用 endpoint で確認 / クリアでき、Pyroscope で CPU、メモリ、goroutine、mutex、block profile を収集できます |
+| Graceful shutdown とデータ保存 | `SHUTDOWN_TIMEOUT_SECONDS` と `QUOTA_DATA_CACHE_SAVE_TIMEOUT_SECONDS` により、可能な範囲で HTTP 処理を閉じ、クォータキャッシュを保存してから終了します |
+
### セキュリティと組織管理
- JWT、WebAuthn/Passkeys、OAuth、OIDC、Telegram、Discord、LinuxDO などのログイン方式をサポートします。
@@ -205,6 +219,7 @@ docker compose up -d
| Agent アクセス | Agent が上流 Key を直接保持し、回収や制限が難しい | Agent ごとに独立トークンを割り当て、モデル・クォータ・期限・グループを制限 |
| プロトコル差分 | Claude、Gemini、Responses などをアプリ側で吸収 | ゲートウェイがプロトコル変換とプロバイダー適応を担当 |
| 障害処理 | retry、fallback、エラー正規化をアプリ側で実装 | チャネル失敗時 retry、重み付きルーティング、エラー処理を内蔵 |
+| 性能と拡張 | timeout、レート制限、接続 pool、cache をアプリ側で処理 | ゲートウェイが streaming timeout、リクエスト制限、Redis / メモリ cache、接続 pool 調整、性能観測を集約 |
| コスト統計 | 請求が各平台に分散し、ユーザーや Agent への帰属が難しい | クォータ、課金、利用統計、消費ログを統一し、トークンとモデル単位で帰属可能 |
| 監査境界 | アプリ側ログが分散し、権限・保持ポリシーが不統一 | 管理者側統一監査入口、通常ユーザーには管理者専用フィールドを除外 |
| プライベート化 | Key、ログ、課金ポリシーが分散 | 自己ホストで Key、データ、ログ、ポリシーを掌握 |
@@ -413,6 +428,7 @@ Seedance 2.0 などの動画モデルでは、解像度や動画入力などの
| リモートデータベース | MySQL ≥ 5.7.8 または PostgreSQL ≥ 9.6 |
| キャッシュ | 単一ノードはメモリキャッシュ、複数ノードは Redis 推奨 |
| フロントエンドビルド | Bun workspace。`web/package.json` と `web/bun.lock` を保持 |
+| ソースビルド | `go.mod` に記載された Go バージョン(現在は Go 1.25.1+)とリポジトリ内の `go.sum` を使用。依存関係またはセキュリティ更新後は `go mod download`、`go mod verify` を実行してから再ビルド |
### 推奨環境変数
@@ -485,6 +501,9 @@ cd MAX-API
docker build -t cscitechtop/max-api:latest .
```
+> [!NOTE]
+> `Dockerfile` はイメージビルド中に Go モジュールをダウンロードします。ホストで直接ビルドする場合、または依存関係 / セキュリティ更新後は、`go.mod` と `go.sum` を必ず一緒にコミットし、`go mod download && go mod verify` を実行してからバイナリまたはイメージを再ビルドしてください。ベースイメージを更新する必要がある場合は `docker build --pull --no-cache -t cscitechtop/max-api:latest .` を使用します。
+
> [!TIP]
> フロントエンドは Bun workspace を使用します。ビルドコンテキストには `web/package.json`、`web/bun.lock`、`web/default/package.json` を保持してください。そうしないと `catalog:` 依存関係を解決できません。
diff --git a/README.md b/README.md
index 92a83ef2..8f6bef8b 100644
--- a/README.md
+++ b/README.md
@@ -52,6 +52,8 @@
MAX API 是由来自科研机构和高校的 AGI 爱好者组织发起、维护和长期运营的 AI 模型治理、AgentOps 与应用服务基础设施项目,面向开发者、研究者、团队和组织提供稳定、可复用的服务层能力。项目关注 AI 应用落地后的持续运营问题:模型越来越多、供应商接口频繁变化、Agent 调用链路变长、成本和审计压力增加。MAX API 在应用、Agent、用户、组织和上游模型服务之间提供统一的接入、鉴权、路由、计费、观测和治理层,让 AI 应用更稳定、更可控地运行。
+换句话说,MAX API 不只是请求转发器,而是面向 AI-ready 应用和 Agent 工作负载的可运营网关:在统一模型协议与供应商差异的同时,把突发流量、长流式响应、大请求体、多节点缓存、成本审计和性能观测放进同一套治理边界。
+
持续投入方向:
- **AI 模型治理**:持续跟踪 OpenAI、Azure OpenAI、AWS Bedrock、Vertex AI、Ollama,以及 DeepSeek、通义千问 / 阿里云百炼、智谱 GLM、Kimi、豆包 / 火山引擎、腾讯混元、百度文心 / 千帆、讯飞星火、MiniMax、零一万物、硅基流动等模型与平台的模型更新、接口变化、参数差异、价格规则和任务协议;同时关注 Dify、RAGFlow、Kling、Seedance 等应用和多模态生态的接入形态,通过渠道、模型映射、协议转换、路径覆盖和可配置任务协议,把分散模型能力纳入统一治理。
@@ -85,6 +87,7 @@ MAX API 当前版本发布分为正式版和 Preview 预览版。Preview 预览
- **渠道配置平面**:通过能力矩阵、表单校验、模型发现和协议模板,降低新增上游渠道、迁移供应商和维护非标准接口时的误配置风险。
- **协议与供应商适配层**:持续跟踪海外官方接口、国产模型平台接口(DeepSeek、通义、智谱、火山引擎、百度千帆等)以及各类 OpenAI 兼容 / 非标准接口的变化,并规范化为稳定的应用侧接口。
- **成本、额度与可靠性治理**:支持渠道路由、加权分发、失败重试、限流、预扣费、失败退款,以及表达式计费、固定价格、任务 rate-card、倍率计费和用量统计。
+- **性能与可扩展性治理**:通过 Redis / 内存缓存、模型请求限流、流式超时与大响应缓冲、请求体上限、磁盘缓存、Pyroscope 性能采样和优雅关闭,支撑单机到多节点部署的稳定运行。
- **组织级运营与审计层**:为团队、研究机构、企业和社区服务提供用户管理、分组管理、私有化部署、数据留存、审计和持续运营优化能力。
- **可复用治理模板**:沉淀渠道模板、任务协议模板、价格配置、部署实践和运维经验,降低新模型、新供应商和新 Agent 场景的接入成本。
@@ -197,6 +200,17 @@ docker compose up -d
- 支持渠道加权随机、失败重试、禁用渠道绕过和模型级路由,降低上游异常对应用和 Agent 的影响。
- 支持 Redis 缓存与内存缓存,适配单机和多机部署。
+### 性能与可扩展性治理
+
+| 能力 | 说明 |
+|------|------|
+| 缓存与多节点扩展 | 单机可使用内存缓存,多机可接入 Redis;用户、令牌、渠道亲和、额度相关缓存减少重复数据库访问,并通过 `SESSION_SECRET`、`CRYPTO_SECRET`、`NODE_NAME` 保持会话、加密和日志归属一致 |
+| 请求限流与容量保护 | 支持全局 API / Web 限流、关键接口限流、搜索限流、模型请求限流和按分组配置的模型请求配额;可使用 Redis 或内存计数器 |
+| 流式与大请求控制 | 支持 `STREAMING_TIMEOUT`、`STREAM_SCANNER_MAX_BUFFER_MB`、`MAX_REQUEST_BODY_MB`、`MAX_FILE_DOWNLOAD_MB` 等配置,控制长流式响应、大行 SSE、解压后请求体和远程文件下载大小 |
+| 中继连接调优 | 支持 `RELAY_TIMEOUT`、`RELAY_IDLE_CONN_TIMEOUT`、`RELAY_MAX_IDLE_CONNS`、`RELAY_MAX_IDLE_CONNS_PER_HOST` 配置上游 HTTP 连接池与超时策略 |
+| 磁盘缓存与性能观测 | 系统性能设置可启用大请求体磁盘缓存、配置缓存阈值和容量;运维接口可查看 / 清理磁盘缓存,并可通过 Pyroscope 采集 CPU、内存、goroutine、mutex 和 block profile |
+| 平滑退出与数据落盘 | 关闭进程时支持 `SHUTDOWN_TIMEOUT_SECONDS` 和 `QUOTA_DATA_CACHE_SAVE_TIMEOUT_SECONDS`,尽量在退出前完成 HTTP 关闭和额度缓存保存 |
+
### 安全与组织管理
- 支持 JWT、WebAuthn/Passkeys、OAuth、OIDC、Telegram、Discord、LinuxDO 等登录方式。
@@ -215,6 +229,7 @@ docker compose up -d
| Agent 访问 | Agent 直接持有上游 Key,难以回收和限额 | 为 Agent 分配独立令牌,并限制模型、额度、过期时间和分组 |
| 协议差异 | 应用自行适配 Claude、Gemini、Responses 等格式 | 网关统一做协议转换和供应商适配 |
| 失败处理 | 应用自行实现重试、降级和错误归一 | 渠道失败自动重试、加权路由和错误处理 |
+| 性能与扩展 | 应用自行处理超时、限流、连接池和缓存 | 网关集中提供流式超时、请求限制、Redis / 内存缓存、连接池调优和性能观测 |
| 成本统计 | 各平台账单分散,难以按用户或 Agent 核算 | 统一额度、计费、用量统计和消费日志,可按令牌和模型归因 |
| 审计边界 | 应用侧分散记录,权限和留存策略不统一 | 管理员侧统一审计入口,普通用户日志过滤管理员专用字段 |
| 私有化 | 密钥、日志和计费策略分散 | 自托管,自主掌控密钥、数据、日志和策略 |
@@ -478,6 +493,7 @@ Seedance 2.0 等视频模型可按分辨率、视频输入等请求参数参与
| 远程数据库 | MySQL ≥ 5.7.8 或 PostgreSQL ≥ 9.6 |
| 缓存 | 单机可使用内存缓存,多机部署建议使用 Redis |
| 前端构建 | 使用 Bun workspace,需保留 `web/package.json` 与 `web/bun.lock` |
+| 源码构建 | 使用仓库 `go.mod` 声明的 Go 版本(当前 Go 1.25.1+)和 `go.sum`;依赖或安全更新后运行 `go mod download`、`go mod verify` 并重新构建 |
### 推荐环境变量
@@ -550,6 +566,9 @@ cd MAX-API
docker build -t cscitechtop/max-api:latest .
```
+> [!NOTE]
+> `Dockerfile` 会在镜像构建中下载 Go 模块。宿主机直接构建或依赖 / 安全更新后,请保持 `go.mod` 与 `go.sum` 成对提交,先运行 `go mod download && go mod verify`,再重新构建二进制或镜像;需要刷新基础镜像时可使用 `docker build --pull --no-cache -t cscitechtop/max-api:latest .`。
+
> [!TIP]
> 前端使用 Bun workspace。构建上下文中必须保留 `web/package.json`、`web/bun.lock` 和 `web/default/package.json`,否则 `catalog:` 依赖无法解析。
@@ -696,15 +715,6 @@ MAX API 不替代 Dify、LangChain、MCP Server、工作流引擎或业务 Agent
如果你修改并通过网络向用户提供本项目服务,请理解并遵守 AGPLv3 对应源码提供等义务。商业合作、机构合作或其他授权问题,请联系:maxapi@max-api.ai。
----
-
-## 🌟 Star History
-
-
-
-[](https://star-history.com/#MAX-API-Next/MAX-API&Date)
-
-
---
diff --git a/README.zh_CN.md b/README.zh_CN.md
index 71b9934c..65511d70 100644
--- a/README.zh_CN.md
+++ b/README.zh_CN.md
@@ -52,6 +52,8 @@
MAX API 是由来自科研机构和高校的 AGI 爱好者组织发起、维护和长期运营的 AI 模型治理、AgentOps 与应用服务基础设施项目,面向开发者、研究者、团队和组织提供稳定、可复用的服务层能力。项目关注 AI 应用落地后的持续运营问题:模型越来越多、供应商接口频繁变化、Agent 调用链路变长、成本和审计压力增加。MAX API 在应用、Agent、用户、组织和上游模型服务之间提供统一的接入、鉴权、路由、计费、观测和治理层,让 AI 应用更稳定、更可控地运行。
+换句话说,MAX API 不只是请求转发器,而是面向 AI-ready 应用和 Agent 工作负载的可运营网关:在统一模型协议与供应商差异的同时,把突发流量、长流式响应、大请求体、多节点缓存、成本审计和性能观测放进同一套治理边界。
+
持续投入方向:
- **AI 模型治理**:持续跟踪 OpenAI、Azure OpenAI、AWS Bedrock、Vertex AI、Ollama,以及 DeepSeek、通义千问 / 阿里云百炼、智谱 GLM、Kimi、豆包 / 火山引擎、腾讯混元、百度文心 / 千帆、讯飞星火、MiniMax、零一万物、硅基流动等模型与平台的模型更新、接口变化、参数差异、价格规则和任务协议;同时关注 Dify、RAGFlow、Kling、Seedance 等应用和多模态生态的接入形态,通过渠道、模型映射、协议转换、路径覆盖和可配置任务协议,把分散模型能力纳入统一治理。
@@ -83,6 +85,7 @@ MAX API 当前版本发布分为正式版和 Preview 预览版。Preview 预览
- **渠道配置平面**:通过能力矩阵、表单校验、模型发现和协议模板,降低新增上游渠道、迁移供应商和维护非标准接口时的误配置风险。
- **协议与供应商适配层**:持续跟踪海外官方接口、国产模型平台接口(DeepSeek、通义、智谱、火山引擎、百度千帆等)以及各类 OpenAI 兼容 / 非标准接口的变化,并规范化为稳定的应用侧接口。
- **成本、额度与可靠性治理**:支持渠道路由、加权分发、失败重试、限流、预扣费、失败退款,以及表达式计费、固定价格、任务 rate-card、倍率计费和用量统计。
+- **性能与可扩展性治理**:通过 Redis / 内存缓存、模型请求限流、流式超时与大响应缓冲、请求体上限、磁盘缓存、Pyroscope 性能采样和优雅关闭,支撑单机到多节点部署的稳定运行。
- **组织级运营与审计层**:为团队、研究机构、企业和社区服务提供用户管理、分组管理、私有化部署、数据留存、审计和持续运营优化能力。
- **可复用治理模板**:沉淀渠道模板、任务协议模板、价格配置、部署实践和运维经验,降低新模型、新供应商和新 Agent 场景的接入成本。
@@ -195,6 +198,17 @@ docker compose up -d
- 支持渠道加权随机、失败重试、禁用渠道绕过和模型级路由,降低上游异常对应用和 Agent 的影响。
- 支持 Redis 缓存与内存缓存,适配单机和多机部署。
+### 性能与可扩展性治理
+
+| 能力 | 说明 |
+|------|------|
+| 缓存与多节点扩展 | 单机可使用内存缓存,多机可接入 Redis;用户、令牌、渠道亲和、额度相关缓存减少重复数据库访问,并通过 `SESSION_SECRET`、`CRYPTO_SECRET`、`NODE_NAME` 保持会话、加密和日志归属一致 |
+| 请求限流与容量保护 | 支持全局 API / Web 限流、关键接口限流、搜索限流、模型请求限流和按分组配置的模型请求配额;可使用 Redis 或内存计数器 |
+| 流式与大请求控制 | 支持 `STREAMING_TIMEOUT`、`STREAM_SCANNER_MAX_BUFFER_MB`、`MAX_REQUEST_BODY_MB`、`MAX_FILE_DOWNLOAD_MB` 等配置,控制长流式响应、大行 SSE、解压后请求体和远程文件下载大小 |
+| 中继连接调优 | 支持 `RELAY_TIMEOUT`、`RELAY_IDLE_CONN_TIMEOUT`、`RELAY_MAX_IDLE_CONNS`、`RELAY_MAX_IDLE_CONNS_PER_HOST` 配置上游 HTTP 连接池与超时策略 |
+| 磁盘缓存与性能观测 | 系统性能设置可启用大请求体磁盘缓存、配置缓存阈值和容量;运维接口可查看 / 清理磁盘缓存,并可通过 Pyroscope 采集 CPU、内存、goroutine、mutex 和 block profile |
+| 平滑退出与数据落盘 | 关闭进程时支持 `SHUTDOWN_TIMEOUT_SECONDS` 和 `QUOTA_DATA_CACHE_SAVE_TIMEOUT_SECONDS`,尽量在退出前完成 HTTP 关闭和额度缓存保存 |
+
### 安全与组织管理
- 支持 JWT、WebAuthn/Passkeys、OAuth、OIDC、Telegram、Discord、LinuxDO 等登录方式。
@@ -213,6 +227,7 @@ docker compose up -d
| Agent 访问 | Agent 直接持有上游 Key,难以回收和限额 | 为 Agent 分配独立令牌,并限制模型、额度、过期时间和分组 |
| 协议差异 | 应用自行适配 Claude、Gemini、Responses 等格式 | 网关统一做协议转换和供应商适配 |
| 失败处理 | 应用自行实现重试、降级和错误归一 | 渠道失败自动重试、加权路由和错误处理 |
+| 性能与扩展 | 应用自行处理超时、限流、连接池和缓存 | 网关集中提供流式超时、请求限制、Redis / 内存缓存、连接池调优和性能观测 |
| 成本统计 | 各平台账单分散,难以按用户或 Agent 核算 | 统一额度、计费、用量统计和消费日志,可按令牌和模型归因 |
| 审计边界 | 应用侧分散记录,权限和留存策略不统一 | 管理员侧统一审计入口,普通用户日志过滤管理员专用字段 |
| 私有化 | 密钥、日志和计费策略分散 | 自托管,自主掌控密钥、数据、日志和策略 |
@@ -476,6 +491,7 @@ Seedance 2.0 等视频模型可按分辨率、视频输入等请求参数参与
| 远程数据库 | MySQL ≥ 5.7.8 或 PostgreSQL ≥ 9.6 |
| 缓存 | 单机可使用内存缓存,多机部署建议使用 Redis |
| 前端构建 | 使用 Bun workspace,需保留 `web/package.json` 与 `web/bun.lock` |
+| 源码构建 | 使用仓库 `go.mod` 声明的 Go 版本(当前 Go 1.25.1+)和 `go.sum`;依赖或安全更新后运行 `go mod download`、`go mod verify` 并重新构建 |
### 推荐环境变量
@@ -548,6 +564,9 @@ cd MAX-API
docker build -t cscitechtop/max-api:latest .
```
+> [!NOTE]
+> `Dockerfile` 会在镜像构建中下载 Go 模块。宿主机直接构建或依赖 / 安全更新后,请保持 `go.mod` 与 `go.sum` 成对提交,先运行 `go mod download && go mod verify`,再重新构建二进制或镜像;需要刷新基础镜像时可使用 `docker build --pull --no-cache -t cscitechtop/max-api:latest .`。
+
> [!TIP]
> 前端使用 Bun workspace。构建上下文中必须保留 `web/package.json`、`web/bun.lock` 和 `web/default/package.json`,否则 `catalog:` 依赖无法解析。
diff --git a/README.zh_TW.md b/README.zh_TW.md
index 7b502ae2..2f586654 100644
--- a/README.zh_TW.md
+++ b/README.zh_TW.md
@@ -52,6 +52,8 @@
MAX API 是由來自科研機構和高校的 AGI 愛好者組織發起、維護和長期營運的 AI 模型治理、AgentOps 與應用服務基礎設施專案,面向開發者、研究者、團隊和組織提供穩定、可復用的服務層能力。專案關注 AI 應用落地后的持續營運問題:模型越來越多、供應商介面頻繁變化、Agent 調用鏈路變長、成本和審計壓力增加。MAX API 在應用、Agent、使用者、組織和上游模型服務之間提供統一的接入、鑒權、路由、計費、觀測和治理層,讓 AI 應用更穩定、更可控地運行。
+換句話說,MAX API 不只是請求轉發器,而是面向 AI-ready 應用和 Agent 工作負載的可營運網關:在統一模型協議與供應商差異的同時,把突發流量、長流式響應、大請求體、多節點緩存、成本審計和性能觀測放進同一套治理邊界。
+
持續投入方向:
- **AI 模型治理**:持續跟蹤 OpenAI、Azure OpenAI、AWS Bedrock、Vertex AI、Ollama,以及 DeepSeek、通義千問 / 阿里云百煉、智譜 GLM、Kimi、豆包 / 火山引擎、騰訊混元、百度文心 / 千帆、訊飛星火、MiniMax、零一萬物、硅基流動等模型與平臺的模型更新、介面變化、參數差異、價格規則和任務協議;同時關注 Dify、RAGFlow、Kling、Seedance 等應用和多模態生態的接入形態,通過渠道、模型映射、協議轉換、路徑覆蓋和可配置任務協議,把分散模型能力納入統一治理。
@@ -83,6 +85,7 @@ MAX API 當前版本發布分為正式版和 Preview 預覽版。Preview 預覽
- **渠道配置平面**:通過能力矩陣、表單校驗、模型發現和協議模板,降低新增上游渠道、遷移供應商和維護非標準介面時的誤配置風險。
- **協議與供應商適配層**:持續跟蹤海外官方介面、國產模型平臺介面(DeepSeek、通義、智譜、火山引擎、百度千帆等)以及各類 OpenAI 兼容 / 非標準介面的變化,并規范化為穩定的應用側介面。
- **成本、額度與可靠性治理**:支持渠道路由、加權分發、失敗重試、限流、預扣費、失敗退款,以及表達式計費、固定價格、任務 rate-card、倍率計費和用量統計。
+- **性能與可擴展性治理**:通過 Redis / 內存緩存、模型請求限流、流式超時與大響應緩衝、請求體上限、磁盤緩存、Pyroscope 性能采樣和優雅關閉,支撐單機到多節點部署的穩定運行。
- **組織級營運與審計層**:為團隊、研究機構、企業和社區服務提供使用者管理、分組管理、私有化部署、資料留存、審計和持續營運優化能力。
- **可復用治理模板**:沉淀渠道模板、任務協議模板、價格配置、部署實踐和運維經驗,降低新模型、新供應商和新 Agent 場景的接入成本。
@@ -195,6 +198,17 @@ docker compose up -d
- 支持渠道加權隨機、失敗重試、禁用渠道繞過和模型級路由,降低上游異常對應用和 Agent 的影響。
- 支持 Redis 緩存與內存緩存,適配單機和多機部署。
+### 性能與可擴展性治理
+
+| 能力 | 說明 |
+|------|------|
+| 緩存與多節點擴展 | 單機可使用內存緩存,多機可接入 Redis;使用者、令牌、渠道親和、額度相關緩存減少重復資料庫訪問,并通過 `SESSION_SECRET`、`CRYPTO_SECRET`、`NODE_NAME` 保持會話、加密和日志歸屬一致 |
+| 請求限流與容量保護 | 支持全局 API / Web 限流、關鍵介面限流、搜索限流、模型請求限流和按分組配置的模型請求配額;可使用 Redis 或內存計數器 |
+| 流式與大請求控制 | 支持 `STREAMING_TIMEOUT`、`STREAM_SCANNER_MAX_BUFFER_MB`、`MAX_REQUEST_BODY_MB`、`MAX_FILE_DOWNLOAD_MB` 等配置,控制長流式響應、大行 SSE、解壓后請求體和遠程文件下載大小 |
+| 中繼連接調優 | 支持 `RELAY_TIMEOUT`、`RELAY_IDLE_CONN_TIMEOUT`、`RELAY_MAX_IDLE_CONNS`、`RELAY_MAX_IDLE_CONNS_PER_HOST` 配置上游 HTTP 連接池與超時策略 |
+| 磁盤緩存與性能觀測 | 系統性能設置可啟用大請求體磁盤緩存、配置緩存閾值和容量;運維介面可查看 / 清理磁盤緩存,并可通過 Pyroscope 采集 CPU、內存、goroutine、mutex 和 block profile |
+| 平滑退出與資料落盤 | 關閉進程時支持 `SHUTDOWN_TIMEOUT_SECONDS` 和 `QUOTA_DATA_CACHE_SAVE_TIMEOUT_SECONDS`,盡量在退出前完成 HTTP 關閉和額度緩存保存 |
+
### 安全與組織管理
- 支持 JWT、WebAuthn/Passkeys、OAuth、OIDC、Telegram、Discord、LinuxDO 等登錄方式。
@@ -213,6 +227,7 @@ docker compose up -d
| Agent 訪問 | Agent 直接持有上游 Key,難以回收和限額 | 為 Agent 分配獨立令牌,并限制模型、額度、過期時間和分組 |
| 協議差異 | 應用自行適配 Claude、Gemini、Responses 等格式 | 網關統一做協議轉換和供應商適配 |
| 失敗處理 | 應用自行實現重試、降級和錯誤歸一 | 渠道失敗自動重試、加權路由和錯誤處理 |
+| 性能與擴展 | 應用自行處理超時、限流、連接池和緩存 | 網關集中提供流式超時、請求限制、Redis / 內存緩存、連接池調優和性能觀測 |
| 成本統計 | 各平臺賬單分散,難以按使用者或 Agent 核算 | 統一額度、計費、用量統計和消費日志,可按令牌和模型歸因 |
| 審計邊界 | 應用側分散記錄,權限和留存策略不統一 | 管理員側統一審計入口,普通使用者日志過濾管理員專用字段 |
| 私有化 | 密鑰、日志和計費策略分散 | 自托管,自主掌控密鑰、資料、日志和策略 |
@@ -476,6 +491,7 @@ Seedance 2.0 等視頻模型可按分辨率、視頻輸入等請求參數參與
| 遠程資料庫 | MySQL ≥ 5.7.8 或 PostgreSQL ≥ 9.6 |
| 緩存 | 單機可使用內存緩存,多機部署建議使用 Redis |
| 前端構建 | 使用 Bun workspace,需保留 `web/package.json` 與 `web/bun.lock` |
+| 源碼構建 | 使用倉庫 `go.mod` 聲明的 Go 版本(目前 Go 1.25.1+)和 `go.sum`;依賴或安全更新後運行 `go mod download`、`go mod verify` 並重新構建 |
### 推薦環境變量
@@ -548,6 +564,9 @@ cd MAX-API
docker build -t cscitechtop/max-api:latest .
```
+> [!NOTE]
+> `Dockerfile` 會在鏡像構建中下載 Go 模塊。宿主機直接構建或依賴 / 安全更新後,請保持 `go.mod` 與 `go.sum` 成對提交,先運行 `go mod download && go mod verify`,再重新構建二進制或鏡像;需要刷新基礎鏡像時可使用 `docker build --pull --no-cache -t cscitechtop/max-api:latest .`。
+
> [!TIP]
> 前端使用 Bun workspace。構建上下文中必須保留 `web/package.json`、`web/bun.lock` 和 `web/default/package.json`,否則 `catalog:` 依賴無法解析。
diff --git a/common/constants.go b/common/constants.go
index 5e35bcbe..9e266cbe 100644
--- a/common/constants.go
+++ b/common/constants.go
@@ -70,13 +70,7 @@ var CryptoSecret = uuid.New().String()
var SessionCookieSecure = false
var DefaultTrustedProxies = []string{
"127.0.0.0/8",
- "10.0.0.0/8",
- "172.16.0.0/12",
- "192.168.0.0/16",
- "169.254.0.0/16",
"::1/128",
- "fc00::/7",
- "fe80::/10",
}
var TrustedProxies []string
diff --git a/common/rate-limit.go b/common/rate-limit.go
index 301c101c..023124bb 100644
--- a/common/rate-limit.go
+++ b/common/rate-limit.go
@@ -45,6 +45,9 @@ func (l *InMemoryRateLimiter) clearExpiredItems() {
func (l *InMemoryRateLimiter) Request(key string, maxRequestNum int, duration int64) bool {
l.mutex.Lock()
defer l.mutex.Unlock()
+ if maxRequestNum <= 0 {
+ return true
+ }
// [old <-- new]
queue, ok := l.store[key]
now := time.Now().Unix()
@@ -68,3 +71,17 @@ func (l *InMemoryRateLimiter) Request(key string, maxRequestNum int, duration in
}
return true
}
+
+func (l *InMemoryRateLimiter) Allow(key string, maxRequestNum int, duration int64) bool {
+ l.mutex.Lock()
+ defer l.mutex.Unlock()
+ if maxRequestNum <= 0 {
+ return true
+ }
+ queue, ok := l.store[key]
+ if !ok || len(*queue) < maxRequestNum {
+ return true
+ }
+ now := time.Now().Unix()
+ return now-(*queue)[0] >= duration
+}
diff --git a/common/redis.go b/common/redis.go
index c7287837..f0a5cc8f 100644
--- a/common/redis.go
+++ b/common/redis.go
@@ -61,6 +61,26 @@ func ParseRedisOption() *redis.Options {
return opt
}
+var redisIncrWithTTLScript = redis.NewScript(`
+local ttl = redis.call("PTTL", KEYS[1])
+if ttl > 0 then
+ redis.call("INCRBY", KEYS[1], ARGV[1])
+ redis.call("PEXPIRE", KEYS[1], ttl)
+ return 1
+end
+return 0
+`)
+
+var redisHIncrByWithTTLScript = redis.NewScript(`
+local ttl = redis.call("PTTL", KEYS[1])
+if ttl > 0 then
+ redis.call("HINCRBY", KEYS[1], ARGV[1], ARGV[2])
+ redis.call("PEXPIRE", KEYS[1], ttl)
+ return 1
+end
+return 0
+`)
+
func RedisSet(key string, value string, expiration time.Duration) error {
if DebugEnabled {
SysLog(fmt.Sprintf("Redis SET: key=%s, value=%s, expiration=%v", key, value, expiration))
@@ -243,31 +263,9 @@ func RedisIncr(key string, delta int64) error {
if DebugEnabled {
SysLog(fmt.Sprintf("Redis INCR: key=%s, delta=%d", key, delta))
}
- // 检查键的剩余生存时间
- ttlCmd := RDB.TTL(context.Background(), key)
- ttl, err := ttlCmd.Result()
- if err != nil && !errors.Is(err, redis.Nil) {
- return fmt.Errorf("failed to get TTL: %w", err)
- }
-
- // 只有在 key 存在且有 TTL 时才需要特殊处理
- if ttl > 0 {
- ctx := context.Background()
- // 开始一个Redis事务
- txn := RDB.TxPipeline()
-
- // 减少余额
- decrCmd := txn.IncrBy(ctx, key, delta)
- if err := decrCmd.Err(); err != nil {
- return err // 如果减少失败,则直接返回错误
- }
-
- // 重新设置过期时间,使用原来的过期时间
- txn.Expire(ctx, key, ttl)
-
- // 执行事务
- _, err = txn.Exec(ctx)
- return err
+ ctx := context.Background()
+ if _, err := redisIncrWithTTLScript.Run(ctx, RDB, []string{key}, delta).Result(); err != nil {
+ return fmt.Errorf("failed to increment key: %w", err)
}
return nil
}
@@ -276,25 +274,9 @@ func RedisHIncrBy(key, field string, delta int64) error {
if DebugEnabled {
SysLog(fmt.Sprintf("Redis HINCRBY: key=%s, field=%s, delta=%d", key, field, delta))
}
- ttlCmd := RDB.TTL(context.Background(), key)
- ttl, err := ttlCmd.Result()
- if err != nil && !errors.Is(err, redis.Nil) {
- return fmt.Errorf("failed to get TTL: %w", err)
- }
-
- if ttl > 0 {
- ctx := context.Background()
- txn := RDB.TxPipeline()
-
- incrCmd := txn.HIncrBy(ctx, key, field, delta)
- if err := incrCmd.Err(); err != nil {
- return err
- }
-
- txn.Expire(ctx, key, ttl)
-
- _, err = txn.Exec(ctx)
- return err
+ ctx := context.Background()
+ if _, err := redisHIncrByWithTTLScript.Run(ctx, RDB, []string{key}, field, delta).Result(); err != nil {
+ return fmt.Errorf("failed to increment hash field: %w", err)
}
return nil
}
diff --git a/common/trusted_proxy_test.go b/common/trusted_proxy_test.go
new file mode 100644
index 00000000..226e6189
--- /dev/null
+++ b/common/trusted_proxy_test.go
@@ -0,0 +1,14 @@
+package common
+
+import (
+ "slices"
+ "testing"
+)
+
+func TestDefaultTrustedProxiesOnlyTrustLoopback(t *testing.T) {
+ for _, proxy := range []string{"10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "169.254.0.0/16", "fc00::/7", "fe80::/10"} {
+ if slices.Contains(DefaultTrustedProxies, proxy) {
+ t.Fatalf("default trusted proxies must not include broad private range %s", proxy)
+ }
+ }
+}
diff --git a/common/verification.go b/common/verification.go
index 41fd3c94..1a390fda 100644
--- a/common/verification.go
+++ b/common/verification.go
@@ -55,6 +55,19 @@ func VerifyCodeWithKey(key string, code string, purpose string) bool {
return code == value.code
}
+func VerifyAndDeleteCodeWithKey(key string, code string, purpose string) bool {
+ verificationMutex.Lock()
+ defer verificationMutex.Unlock()
+ mapKey := purpose + key
+ value, okay := verificationMap[mapKey]
+ now := time.Now()
+ if !okay || int(now.Sub(value.time).Seconds()) >= VerificationValidMinutes*60 || code != value.code {
+ return false
+ }
+ delete(verificationMap, mapKey)
+ return true
+}
+
func DeleteKey(key string, purpose string) {
verificationMutex.Lock()
defer verificationMutex.Unlock()
diff --git a/common/verification_test.go b/common/verification_test.go
new file mode 100644
index 00000000..2e03519e
--- /dev/null
+++ b/common/verification_test.go
@@ -0,0 +1,19 @@
+package common
+
+import "testing"
+
+func TestVerifyAndDeleteCodeWithKeyConsumesCodeOnce(t *testing.T) {
+ key := "consume-once@example.com"
+ code := "123456"
+ RegisterVerificationCodeWithKey(key, code, EmailVerificationPurpose)
+
+ if !VerifyAndDeleteCodeWithKey(key, code, EmailVerificationPurpose) {
+ t.Fatal("expected first verification to succeed")
+ }
+ if VerifyCodeWithKey(key, code, EmailVerificationPurpose) {
+ t.Fatal("expected code to be consumed")
+ }
+ if VerifyAndDeleteCodeWithKey(key, code, EmailVerificationPurpose) {
+ t.Fatal("expected second verification to fail")
+ }
+}
diff --git a/controller/audit.go b/controller/audit.go
index 233f3059..fac35107 100644
--- a/controller/audit.go
+++ b/controller/audit.go
@@ -49,6 +49,9 @@ var auditContentTemplates = map[string]string{
"channel.upstream_detect_all": "Detected upstream model changes for channels",
"redemption.create": "Created ${count} redemption codes named ${name} (${quota} each)",
+
+ "subscription.plan_reset": "Reset active subscriptions for plan ${plan_id}",
+ "subscription.user_plan_reset": "Reset active plan ${plan_id} subscriptions for user ${target_user_id}",
}
func auditContentEN(action string, params map[string]interface{}) string {
diff --git a/controller/channel.go b/controller/channel.go
index 0252e2bf..ae770a93 100644
--- a/controller/channel.go
+++ b/controller/channel.go
@@ -18,6 +18,7 @@ import (
"github.com/MAX-API-Next/MAX-API/relay/channel/ollama"
"github.com/MAX-API-Next/MAX-API/relay/channel/task/taskcommon"
"github.com/MAX-API-Next/MAX-API/service"
+ "github.com/MAX-API-Next/MAX-API/setting/system_setting"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
@@ -594,7 +595,7 @@ func getVertexArrayKeys(keys string) ([]string, error) {
case string:
keyStr = strings.TrimSpace(v)
default:
- bytes, err := json.Marshal(v)
+ bytes, err := common.Marshal(v)
if err != nil {
return nil, fmt.Errorf("Vertex AI key JSON 编码失败: %w", err)
}
@@ -968,7 +969,7 @@ func UpdateChannel(c *gin.Context) {
if strings.HasPrefix(strings.TrimSpace(originChannel.Key), "[") {
// JSON数组格式
var arr []json.RawMessage
- if err := json.Unmarshal([]byte(strings.TrimSpace(originChannel.Key)), &arr); err == nil {
+ if err := common.Unmarshal([]byte(strings.TrimSpace(originChannel.Key)), &arr); err == nil {
existingKeys = make([]string, len(arr))
for i, v := range arr {
existingKeys[i] = string(v)
@@ -1149,10 +1150,19 @@ func FetchModels(c *gin.Context) {
return
}
- client := &http.Client{}
- url := fmt.Sprintf("%s/v1/models", baseURL)
+ client := service.GetSSRFProtectedHttpClient()
+ url := fmt.Sprintf("%s/v1/models", strings.TrimRight(baseURL, "/"))
- request, err := http.NewRequest("GET", url, nil)
+ fetchSetting := system_setting.GetFetchSetting()
+ if err := common.ValidateURLWithFetchSetting(url, fetchSetting.EnableSSRFProtection, fetchSetting.AllowPrivateIp, fetchSetting.DomainFilterMode, fetchSetting.IpFilterMode, fetchSetting.DomainList, fetchSetting.IpList, fetchSetting.AllowedPorts, fetchSetting.ApplyIPFilterForDomain); err != nil {
+ c.JSON(http.StatusBadRequest, gin.H{
+ "success": false,
+ "message": fmt.Sprintf("request reject: %v", err),
+ })
+ return
+ }
+
+ request, err := http.NewRequest(http.MethodGet, url, nil)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{
"success": false,
@@ -1171,6 +1181,7 @@ func FetchModels(c *gin.Context) {
})
return
}
+ defer response.Body.Close()
//check status code
if response.StatusCode != http.StatusOK {
c.JSON(http.StatusInternalServerError, gin.H{
@@ -1179,7 +1190,6 @@ func FetchModels(c *gin.Context) {
})
return
}
- defer response.Body.Close()
var result struct {
Data []struct {
@@ -1187,7 +1197,7 @@ func FetchModels(c *gin.Context) {
} `json:"data"`
}
- if err := json.NewDecoder(response.Body).Decode(&result); err != nil {
+ if err := common.DecodeJson(response.Body, &result); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{
"success": false,
"message": err.Error(),
@@ -1957,7 +1967,7 @@ func OllamaPullModelStream(c *gin.Context) {
// 创建进度回调函数
progressCallback := func(progress ollama.OllamaPullResponse) {
- data, _ := json.Marshal(progress)
+ data, _ := common.Marshal(progress)
fmt.Fprintf(c.Writer, "data: %s\n\n", string(data))
c.Writer.Flush()
}
@@ -1966,12 +1976,12 @@ func OllamaPullModelStream(c *gin.Context) {
err = ollama.PullOllamaModelStream(baseURL, key, req.ModelName, progressCallback)
if err != nil {
- errorData, _ := json.Marshal(gin.H{
+ errorData, _ := common.Marshal(gin.H{
"error": err.Error(),
})
fmt.Fprintf(c.Writer, "data: %s\n\n", string(errorData))
} else {
- successData, _ := json.Marshal(gin.H{
+ successData, _ := common.Marshal(gin.H{
"message": fmt.Sprintf("Model %s pulled successfully", req.ModelName),
})
fmt.Fprintf(c.Writer, "data: %s\n\n", string(successData))
diff --git a/controller/channel_fetch_models_test.go b/controller/channel_fetch_models_test.go
new file mode 100644
index 00000000..607f5e32
--- /dev/null
+++ b/controller/channel_fetch_models_test.go
@@ -0,0 +1,55 @@
+package controller
+
+import (
+ "bytes"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+
+ "github.com/MAX-API-Next/MAX-API/common"
+ "github.com/MAX-API-Next/MAX-API/constant"
+ "github.com/MAX-API-Next/MAX-API/setting/system_setting"
+ "github.com/gin-gonic/gin"
+ "github.com/stretchr/testify/require"
+)
+
+func TestFetchModelsRejectsPrivateBaseURL(t *testing.T) {
+ gin.SetMode(gin.TestMode)
+
+ fetchSetting := system_setting.GetFetchSetting()
+ originalFetchSetting := *fetchSetting
+ t.Cleanup(func() {
+ *fetchSetting = originalFetchSetting
+ })
+ fetchSetting.EnableSSRFProtection = true
+ fetchSetting.AllowPrivateIp = false
+ fetchSetting.DomainFilterMode = false
+ fetchSetting.IpFilterMode = false
+ fetchSetting.DomainList = nil
+ fetchSetting.IpList = nil
+ fetchSetting.AllowedPorts = []string{"80", "443", "8080", "8443"}
+ fetchSetting.ApplyIPFilterForDomain = true
+
+ payload, err := common.Marshal(map[string]any{
+ "base_url": "http://127.0.0.1",
+ "type": constant.ChannelTypeOpenAI,
+ "key": "sk-test",
+ })
+ require.NoError(t, err)
+
+ recorder := httptest.NewRecorder()
+ ctx, _ := gin.CreateTestContext(recorder)
+ ctx.Request = httptest.NewRequest(http.MethodPost, "/api/channel/fetch_models", bytes.NewReader(payload))
+ ctx.Request.Header.Set("Content-Type", "application/json")
+
+ FetchModels(ctx)
+
+ require.Equal(t, http.StatusBadRequest, recorder.Code)
+ var response struct {
+ Success bool `json:"success"`
+ Message string `json:"message"`
+ }
+ require.NoError(t, common.Unmarshal(recorder.Body.Bytes(), &response))
+ require.False(t, response.Success)
+ require.Contains(t, response.Message, "request reject")
+}
diff --git a/controller/discord.go b/controller/discord.go
index e447b3e7..d6d9033e 100644
--- a/controller/discord.go
+++ b/controller/discord.go
@@ -203,8 +203,12 @@ func DiscordBind(c *gin.Context) {
return
}
session := sessions.Default(c)
- id := session.Get("id")
- user.Id = id.(int)
+ id, ok := sessionUserID(session.Get("id"))
+ if !ok {
+ common.ApiErrorMsg(c, "用户未登录或登录状态已失效")
+ return
+ }
+ user.Id = id
err = user.FillUserById()
if err != nil {
common.ApiError(c, err)
diff --git a/controller/github.go b/controller/github.go
index 4ef8da47..7b1d8ef5 100644
--- a/controller/github.go
+++ b/controller/github.go
@@ -198,9 +198,12 @@ func GitHubBind(c *gin.Context) {
return
}
session := sessions.Default(c)
- id := session.Get("id")
- // id := c.GetInt("id") // critical bug!
- user.Id = id.(int)
+ id, ok := sessionUserID(session.Get("id"))
+ if !ok {
+ common.ApiErrorMsg(c, "用户未登录或登录状态已失效")
+ return
+ }
+ user.Id = id
err = user.FillUserById()
if err != nil {
common.ApiError(c, err)
diff --git a/controller/linuxdo.go b/controller/linuxdo.go
index 56fce9bd..b49d29ce 100644
--- a/controller/linuxdo.go
+++ b/controller/linuxdo.go
@@ -56,8 +56,12 @@ func LinuxDoBind(c *gin.Context) {
}
session := sessions.Default(c)
- id := session.Get("id")
- user.Id = id.(int)
+ id, ok := sessionUserID(session.Get("id"))
+ if !ok {
+ common.ApiErrorMsg(c, "用户未登录或登录状态已失效")
+ return
+ }
+ user.Id = id
err = user.FillUserById()
if err != nil {
diff --git a/controller/misc.go b/controller/misc.go
index 4adc489c..b2b5d4bd 100644
--- a/controller/misc.go
+++ b/controller/misc.go
@@ -348,7 +348,7 @@ func ResetPassword(c *gin.Context) {
common.ApiErrorI18n(c, i18n.MsgInvalidParams)
return
}
- if !common.VerifyCodeWithKey(req.Email, req.Token, common.PasswordResetPurpose) {
+ if !common.VerifyAndDeleteCodeWithKey(req.Email, req.Token, common.PasswordResetPurpose) {
common.ApiErrorI18n(c, i18n.MsgUserPasswordResetLinkInvalid)
return
}
@@ -362,7 +362,6 @@ func ResetPassword(c *gin.Context) {
common.ApiError(c, err)
return
}
- common.DeleteKey(req.Email, common.PasswordResetPurpose)
c.JSON(http.StatusOK, gin.H{
"success": true,
"message": "",
diff --git a/controller/oauth.go b/controller/oauth.go
index 7db43379..f4ce8b82 100644
--- a/controller/oauth.go
+++ b/controller/oauth.go
@@ -20,6 +20,21 @@ func providerParams(name string) map[string]any {
return map[string]any{"Provider": name}
}
+func oauthProviderUserUpdateField(provider oauth.Provider) (model.UserUpdateField, bool) {
+ switch provider.(type) {
+ case *oauth.GitHubProvider:
+ return model.UserUpdateFieldGitHubId, true
+ case *oauth.DiscordProvider:
+ return model.UserUpdateFieldDiscordId, true
+ case *oauth.OIDCProvider:
+ return model.UserUpdateFieldOidcId, true
+ case *oauth.LinuxDOProvider:
+ return model.UserUpdateFieldLinuxDOId, true
+ default:
+ return "", false
+ }
+}
+
// GenerateOAuthCode generates a state code for OAuth CSRF protection
func GenerateOAuthCode(c *gin.Context) {
session := sessions.Default(c)
@@ -107,10 +122,6 @@ func HandleOAuth(c *gin.Context) {
// 7. Find or create user
user, err := findOrCreateOAuthUser(c, provider, oauthUser, session)
if err != nil {
- if errors.Is(err, model.ErrEmailAlreadyTaken) {
- common.ApiErrorI18n(c, i18n.MsgUserEmailAlreadyTaken)
- return
- }
switch err.(type) {
case *OAuthUserDeletedError:
common.ApiErrorI18n(c, i18n.MsgOAuthUserDeleted)
@@ -171,8 +182,12 @@ func handleOAuthBind(c *gin.Context, provider oauth.Provider) {
// Get current user from session
session := sessions.Default(c)
- id := session.Get("id")
- user := model.User{Id: id.(int)}
+ id, ok := sessionUserID(session.Get("id"))
+ if !ok {
+ common.ApiErrorMsg(c, "用户未登录或登录状态已失效")
+ return
+ }
+ user := model.User{Id: id}
err = user.FillUserById()
if err != nil {
common.ApiError(c, err)
@@ -190,7 +205,12 @@ func handleOAuthBind(c *gin.Context, provider oauth.Provider) {
} else {
// Built-in provider: update user record directly
provider.SetProviderUserID(&user, oauthUser.ProviderUserID)
- err = user.Update(false)
+ updateField, ok := oauthProviderUserUpdateField(provider)
+ if !ok {
+ common.ApiError(c, fmt.Errorf("unsupported built-in OAuth provider: %s", provider.GetName()))
+ return
+ }
+ err = user.UpdateFields(false, updateField)
if err != nil {
common.ApiError(c, err)
return
diff --git a/controller/oauth_test.go b/controller/oauth_test.go
new file mode 100644
index 00000000..fd79911c
--- /dev/null
+++ b/controller/oauth_test.go
@@ -0,0 +1,33 @@
+package controller
+
+import (
+ "testing"
+
+ "github.com/MAX-API-Next/MAX-API/model"
+ "github.com/MAX-API-Next/MAX-API/oauth"
+)
+
+func TestOAuthProviderUserUpdateField(t *testing.T) {
+ tests := []struct {
+ name string
+ provider oauth.Provider
+ want model.UserUpdateField
+ }{
+ {name: "github", provider: &oauth.GitHubProvider{}, want: model.UserUpdateFieldGitHubId},
+ {name: "discord", provider: &oauth.DiscordProvider{}, want: model.UserUpdateFieldDiscordId},
+ {name: "oidc", provider: &oauth.OIDCProvider{}, want: model.UserUpdateFieldOidcId},
+ {name: "linuxdo", provider: &oauth.LinuxDOProvider{}, want: model.UserUpdateFieldLinuxDOId},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, ok := oauthProviderUserUpdateField(tt.provider)
+ if !ok {
+ t.Fatalf("expected provider field mapping")
+ }
+ if got != tt.want {
+ t.Fatalf("expected %q, got %q", tt.want, got)
+ }
+ })
+ }
+}
diff --git a/controller/oidc.go b/controller/oidc.go
index 25f82b46..c4cabace 100644
--- a/controller/oidc.go
+++ b/controller/oidc.go
@@ -206,9 +206,12 @@ func OidcBind(c *gin.Context) {
return
}
session := sessions.Default(c)
- id := session.Get("id")
- // id := c.GetInt("id") // critical bug!
- user.Id = id.(int)
+ id, ok := sessionUserID(session.Get("id"))
+ if !ok {
+ common.ApiErrorMsg(c, "用户未登录或登录状态已失效")
+ return
+ }
+ user.Id = id
err = user.FillUserById()
if err != nil {
common.ApiError(c, err)
diff --git a/controller/redemption.go b/controller/redemption.go
index 56eb572e..49371266 100644
--- a/controller/redemption.go
+++ b/controller/redemption.go
@@ -29,8 +29,9 @@ func GetAllRedemptions(c *gin.Context) {
func SearchRedemptions(c *gin.Context) {
keyword := c.Query("keyword")
+ status := c.Query("status")
pageInfo := common.GetPageQuery(c)
- redemptions, total, err := model.SearchRedemptions(keyword, pageInfo.GetStartIdx(), pageInfo.GetPageSize())
+ redemptions, total, err := model.SearchRedemptions(keyword, status, pageInfo.GetStartIdx(), pageInfo.GetPageSize())
if err != nil {
common.ApiError(c, err)
return
diff --git a/controller/session.go b/controller/session.go
new file mode 100644
index 00000000..c8372478
--- /dev/null
+++ b/controller/session.go
@@ -0,0 +1,24 @@
+package controller
+
+import "strconv"
+
+func sessionInt(value any) (int, bool) {
+ switch v := value.(type) {
+ case int:
+ return v, true
+ case int64:
+ return int(v), true
+ case float64:
+ return int(v), true
+ case string:
+ id, err := strconv.Atoi(v)
+ return id, err == nil
+ default:
+ return 0, false
+ }
+}
+
+func sessionUserID(value any) (int, bool) {
+ id, ok := sessionInt(value)
+ return id, ok && id > 0
+}
diff --git a/controller/session_test.go b/controller/session_test.go
new file mode 100644
index 00000000..96f78dd1
--- /dev/null
+++ b/controller/session_test.go
@@ -0,0 +1,22 @@
+package controller
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestSessionUserIDAcceptsJSONDecodedNumber(t *testing.T) {
+ id, ok := sessionUserID(float64(42))
+
+ require.True(t, ok)
+ require.Equal(t, 42, id)
+}
+
+func TestSessionUserIDRejectsMissingOrInvalidID(t *testing.T) {
+ _, ok := sessionUserID(nil)
+ require.False(t, ok)
+
+ _, ok = sessionUserID(0)
+ require.False(t, ok)
+}
diff --git a/controller/subscription.go b/controller/subscription.go
index fdcf42a3..51adb312 100644
--- a/controller/subscription.go
+++ b/controller/subscription.go
@@ -1,6 +1,7 @@
package controller
import (
+ "fmt"
"strconv"
"strings"
@@ -373,6 +374,28 @@ type AdminCreateUserSubscriptionRequest struct {
PlanId int `json:"plan_id"`
}
+type AdminResetSubscriptionRequest struct {
+ PlanId int `json:"plan_id"`
+ AdvanceResetTime *bool `json:"advance_reset_time"`
+}
+
+func resolveAdvanceResetTime(value *bool) bool {
+ if value == nil {
+ return true
+ }
+ return *value
+}
+
+func recordSubscriptionResetUserLogs(result *model.SubscriptionResetResult, adminInfo map[string]interface{}) {
+ if result == nil || result.ResetCount == 0 {
+ return
+ }
+ content := fmt.Sprintf("管理员重置订阅套餐 %s(ID: %d)额度", result.PlanTitle, result.PlanId)
+ for _, userId := range result.AffectedUserIds {
+ model.RecordLogWithAdminInfo(userId, model.LogTypeManage, content, adminInfo)
+ }
+}
+
// AdminCreateUserSubscription creates a new user subscription from a plan (no payment).
func AdminCreateUserSubscription(c *gin.Context) {
if !requirePaymentCompliance(c) {
@@ -401,6 +424,77 @@ func AdminCreateUserSubscription(c *gin.Context) {
common.ApiSuccess(c, nil)
}
+func AdminResetUserSubscriptionsByPlan(c *gin.Context) {
+ if !requirePaymentCompliance(c) {
+ return
+ }
+
+ userId, _ := strconv.Atoi(c.Param("id"))
+ if userId <= 0 {
+ common.ApiErrorMsg(c, "无效的用户ID")
+ return
+ }
+ var req AdminResetSubscriptionRequest
+ if err := c.ShouldBindJSON(&req); err != nil {
+ common.ApiErrorMsg(c, "参数错误")
+ return
+ }
+ if req.PlanId <= 0 {
+ common.ApiErrorMsg(c, "参数错误")
+ return
+ }
+ advanceResetTime := resolveAdvanceResetTime(req.AdvanceResetTime)
+ result, err := model.AdminResetUserSubscriptionsByPlan(userId, req.PlanId, advanceResetTime)
+ if err != nil {
+ common.ApiError(c, err)
+ return
+ }
+ recordSubscriptionResetUserLogs(result, auditOperatorInfo(c))
+ recordManageAuditFor(c, userId, "subscription.user_plan_reset", map[string]interface{}{
+ "target_user_id": userId,
+ "plan_id": result.PlanId,
+ "plan_title": result.PlanTitle,
+ "reset_count": result.ResetCount,
+ "user_count": result.UserCount,
+ "advance_reset_time": result.AdvanceResetTime,
+ })
+ common.ApiSuccess(c, result)
+}
+
+func AdminResetPlanSubscriptions(c *gin.Context) {
+ if !requirePaymentCompliance(c) {
+ return
+ }
+
+ planId, _ := strconv.Atoi(c.Param("id"))
+ if planId <= 0 {
+ common.ApiErrorMsg(c, "无效的ID")
+ return
+ }
+ var req AdminResetSubscriptionRequest
+ if err := c.ShouldBindJSON(&req); err != nil {
+ common.ApiErrorMsg(c, "参数错误")
+ return
+ }
+ advanceResetTime := resolveAdvanceResetTime(req.AdvanceResetTime)
+ result, err := model.AdminResetPlanSubscriptions(planId, advanceResetTime)
+ if err != nil {
+ common.ApiError(c, err)
+ return
+ }
+ recordSubscriptionResetUserLogs(result, auditOperatorInfo(c))
+ common.SysLog(fmt.Sprintf("admin reset subscription plan %d quota: reset_count=%d user_count=%d advance_reset_time=%t",
+ result.PlanId, result.ResetCount, result.UserCount, result.AdvanceResetTime))
+ recordManageAudit(c, "subscription.plan_reset", map[string]interface{}{
+ "plan_id": result.PlanId,
+ "plan_title": result.PlanTitle,
+ "reset_count": result.ResetCount,
+ "user_count": result.UserCount,
+ "advance_reset_time": result.AdvanceResetTime,
+ })
+ common.ApiSuccess(c, result)
+}
+
// AdminInvalidateUserSubscription cancels a user subscription immediately.
func AdminInvalidateUserSubscription(c *gin.Context) {
subId, _ := strconv.Atoi(c.Param("id"))
diff --git a/controller/subscription_test.go b/controller/subscription_test.go
new file mode 100644
index 00000000..1c359768
--- /dev/null
+++ b/controller/subscription_test.go
@@ -0,0 +1,84 @@
+package controller
+
+import (
+ "bytes"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+
+ "github.com/MAX-API-Next/MAX-API/common"
+ appi18n "github.com/MAX-API-Next/MAX-API/i18n"
+ "github.com/MAX-API-Next/MAX-API/setting/operation_setting"
+ "github.com/gin-gonic/gin"
+ "github.com/stretchr/testify/require"
+)
+
+type subscriptionAPIResponse struct {
+ Success bool `json:"success"`
+ Message string `json:"message"`
+}
+
+func denyPaymentComplianceForTest(t *testing.T) {
+ t.Helper()
+ paymentSetting := operation_setting.GetPaymentSetting()
+ originalConfirmed := paymentSetting.ComplianceConfirmed
+ originalTermsVersion := paymentSetting.ComplianceTermsVersion
+ t.Cleanup(func() {
+ paymentSetting.ComplianceConfirmed = originalConfirmed
+ paymentSetting.ComplianceTermsVersion = originalTermsVersion
+ })
+ paymentSetting.ComplianceConfirmed = false
+ paymentSetting.ComplianceTermsVersion = operation_setting.CurrentComplianceTermsVersion
+}
+
+func newSubscriptionTestContext(t *testing.T, target string, body any, params gin.Params) (*gin.Context, *httptest.ResponseRecorder) {
+ t.Helper()
+ payload, err := common.Marshal(body)
+ require.NoError(t, err)
+
+ recorder := httptest.NewRecorder()
+ ctx, _ := gin.CreateTestContext(recorder)
+ ctx.Request = httptest.NewRequest(http.MethodPost, target, bytes.NewReader(payload))
+ ctx.Request.Header.Set("Content-Type", "application/json")
+ ctx.Params = params
+ return ctx, recorder
+}
+
+func decodeSubscriptionAPIResponse(t *testing.T, recorder *httptest.ResponseRecorder) subscriptionAPIResponse {
+ t.Helper()
+ var response subscriptionAPIResponse
+ require.NoError(t, common.Unmarshal(recorder.Body.Bytes(), &response))
+ return response
+}
+
+func TestAdminResetUserSubscriptionsByPlanRequiresPaymentCompliance(t *testing.T) {
+ require.NoError(t, appi18n.Init())
+ denyPaymentComplianceForTest(t)
+
+ ctx, recorder := newSubscriptionTestContext(t,
+ "/api/user/1/subscription/reset",
+ AdminResetSubscriptionRequest{PlanId: 1},
+ gin.Params{{Key: "id", Value: "1"}},
+ )
+ AdminResetUserSubscriptionsByPlan(ctx)
+
+ response := decodeSubscriptionAPIResponse(t, recorder)
+ require.False(t, response.Success)
+ require.NotEmpty(t, response.Message)
+}
+
+func TestAdminResetPlanSubscriptionsRequiresPaymentCompliance(t *testing.T) {
+ require.NoError(t, appi18n.Init())
+ denyPaymentComplianceForTest(t)
+
+ ctx, recorder := newSubscriptionTestContext(t,
+ "/api/subscription/plan/1/reset",
+ AdminResetSubscriptionRequest{},
+ gin.Params{{Key: "id", Value: "1"}},
+ )
+ AdminResetPlanSubscriptions(ctx)
+
+ response := decodeSubscriptionAPIResponse(t, recorder)
+ require.False(t, response.Success)
+ require.NotEmpty(t, response.Message)
+}
diff --git a/controller/telegram.go b/controller/telegram.go
index 07a0b985..db66fb61 100644
--- a/controller/telegram.go
+++ b/controller/telegram.go
@@ -41,8 +41,12 @@ func TelegramBind(c *gin.Context) {
}
session := sessions.Default(c)
- id := session.Get("id")
- user := model.User{Id: id.(int)}
+ id, ok := sessionUserID(session.Get("id"))
+ if !ok {
+ common.ApiErrorMsg(c, "用户未登录或登录状态已失效")
+ return
+ }
+ user := model.User{Id: id}
if err := user.FillUserById(); err != nil {
c.JSON(200, gin.H{
"message": err.Error(),
diff --git a/controller/token.go b/controller/token.go
index ae74304f..19859be0 100644
--- a/controller/token.go
+++ b/controller/token.go
@@ -41,7 +41,7 @@ func validateAssignableTokenGroup(c *gin.Context, group string) bool {
if userGroup == "" {
userGroup = c.GetString("user_group")
}
- if service.CanUseTokenGroup(userGroup, group) {
+ if service.CanUseTokenGroupRuntime(userGroup, group) {
return true
}
common.ApiErrorI18n(c, i18n.MsgTokenGroupNotAssignable, map[string]any{"Group": group})
diff --git a/controller/token_test.go b/controller/token_test.go
index a78c8f3e..c4314acd 100644
--- a/controller/token_test.go
+++ b/controller/token_test.go
@@ -522,7 +522,7 @@ func TestGetTokenMasksKeyInResponse(t *testing.T) {
}
}
-func TestAddTokenRejectsNonSelectableAutoRoute(t *testing.T) {
+func TestAddTokenAllowsRuntimeAutoRoute(t *testing.T) {
db := setupTokenControllerTestDB(t)
setupHiddenAutoRouteForTokenTest(t)
if err := appi18n.Init(); err != nil {
@@ -545,12 +545,48 @@ func TestAddTokenRejectsNonSelectableAutoRoute(t *testing.T) {
ctx.Request.Header.Set("Accept-Language", "zh-CN")
AddToken(ctx)
+ response := decodeAPIResponse(t, recorder)
+ if !response.Success {
+ t.Fatalf("expected hidden runtime route token creation to succeed, got message: %s", response.Message)
+ }
+
+ var stored model.Token
+ if err := db.First(&stored).Error; err != nil {
+ t.Fatalf("failed to load created token: %v", err)
+ }
+ if stored.Group != "auto:internal" {
+ t.Fatalf("expected token group auto:internal, got %q", stored.Group)
+ }
+}
+
+func TestAddTokenRejectsUnavailableGroup(t *testing.T) {
+ db := setupTokenControllerTestDB(t)
+ if err := appi18n.Init(); err != nil {
+ t.Fatalf("failed to initialize i18n: %v", err)
+ }
+
+ body := map[string]any{
+ "name": "invalid-group-token",
+ "expired_time": -1,
+ "remain_quota": 100,
+ "unlimited_quota": true,
+ "model_limits_enabled": false,
+ "model_limits": "",
+ "group": "missing",
+ "cross_group_retry": true,
+ }
+
+ ctx, recorder := newAuthenticatedContext(t, http.MethodPost, "/api/token/", body, 1)
+ setAuthenticatedUserGroup(ctx, "default")
+ ctx.Request.Header.Set("Accept-Language", "zh-CN")
+ AddToken(ctx)
+
response := decodeAPIResponse(t, recorder)
if response.Success {
- t.Fatalf("expected hidden auto route token creation to fail")
+ t.Fatalf("expected unavailable group token creation to fail")
}
- if !strings.Contains(response.Message, "auto:internal") {
- t.Fatalf("expected error message to mention hidden route, got %q", response.Message)
+ if !strings.Contains(response.Message, "missing") {
+ t.Fatalf("expected error message to mention unavailable group, got %q", response.Message)
}
if !strings.Contains(response.Message, "无权分配") {
t.Fatalf("expected localized denial message, got %q", response.Message)
@@ -601,7 +637,7 @@ func TestUpdateTokenMasksKeyInResponse(t *testing.T) {
}
}
-func TestUpdateTokenRejectsNonSelectableAutoRoute(t *testing.T) {
+func TestUpdateTokenAllowsRuntimeAutoRoute(t *testing.T) {
db := setupTokenControllerTestDB(t)
setupHiddenAutoRouteForTokenTest(t)
if err := appi18n.Init(); err != nil {
@@ -627,14 +663,49 @@ func TestUpdateTokenRejectsNonSelectableAutoRoute(t *testing.T) {
UpdateToken(ctx)
response := decodeAPIResponse(t, recorder)
- if response.Success {
- t.Fatalf("expected hidden auto route token update to fail")
+ if !response.Success {
+ t.Fatalf("expected hidden runtime route token update to succeed, got message: %s", response.Message)
}
- if !strings.Contains(response.Message, "auto:internal") {
- t.Fatalf("expected error message to mention hidden route, got %q", response.Message)
+
+ var stored model.Token
+ if err := db.First(&stored, token.Id).Error; err != nil {
+ t.Fatalf("failed to reload token: %v", err)
}
- if !strings.Contains(response.Message, "无权分配") {
- t.Fatalf("expected localized denial message, got %q", response.Message)
+ if stored.Group != "auto:internal" {
+ t.Fatalf("expected token group auto:internal, got %q", stored.Group)
+ }
+}
+
+func TestUpdateTokenRejectsUnavailableGroup(t *testing.T) {
+ db := setupTokenControllerTestDB(t)
+ if err := appi18n.Init(); err != nil {
+ t.Fatalf("failed to initialize i18n: %v", err)
+ }
+ token := seedToken(t, db, 1, "editable-token", "reject1234group5678")
+
+ body := map[string]any{
+ "id": token.Id,
+ "name": "updated-token",
+ "expired_time": -1,
+ "remain_quota": 100,
+ "unlimited_quota": true,
+ "model_limits_enabled": false,
+ "model_limits": "",
+ "group": "missing",
+ "cross_group_retry": true,
+ }
+
+ ctx, recorder := newAuthenticatedContext(t, http.MethodPut, "/api/token/", body, 1)
+ setAuthenticatedUserGroup(ctx, "default")
+ ctx.Request.Header.Set("Accept-Language", "zh-CN")
+ UpdateToken(ctx)
+
+ response := decodeAPIResponse(t, recorder)
+ if response.Success {
+ t.Fatalf("expected unavailable group token update to fail")
+ }
+ if !strings.Contains(response.Message, "missing") {
+ t.Fatalf("expected error message to mention unavailable group, got %q", response.Message)
}
var stored model.Token
diff --git a/controller/user.go b/controller/user.go
index 2388f1a1..8ee30d4a 100644
--- a/controller/user.go
+++ b/controller/user.go
@@ -4,6 +4,7 @@ import (
"errors"
"fmt"
"net/http"
+ "net/mail"
"net/url"
"strconv"
"strings"
@@ -34,6 +35,8 @@ var (
errOriginalPasswordFail = errors.New("original password is incorrect")
)
+const maxUserQuotaValue = 1<<31 - 1
+
func Login(c *gin.Context) {
if !common.PasswordLoginEnabled {
common.ApiErrorI18n(c, i18n.MsgUserPasswordLoginDisabled)
@@ -251,6 +254,9 @@ func Register(c *gin.Context) {
common.ApiError(c, err)
return
}
+ if common.EmailVerificationEnabled {
+ common.DeleteKey(user.Email, common.EmailVerificationPurpose)
+ }
// 获取插入后的用户ID
var insertedUser model.User
@@ -609,6 +615,25 @@ func GetUserModels(c *gin.Context) {
return
}
groups := service.GetUserUsableGroups(user.Group)
+ group := c.Query("group")
+ if group != "" {
+ if _, ok := groups[group]; !ok {
+ c.JSON(http.StatusOK, gin.H{
+ "success": true,
+ "message": "",
+ "data": []string{},
+ })
+ return
+ }
+
+ c.JSON(http.StatusOK, gin.H{
+ "success": true,
+ "message": "",
+ "data": model.GetGroupEnabledModels(group),
+ })
+ return
+ }
+
var models []string
for group := range groups {
for _, g := range model.GetGroupEnabledModels(group) {
@@ -970,6 +995,17 @@ type ManageRequest struct {
Mode string `json:"mode"`
}
+func isValidQuotaOverride(value int) bool {
+ return value >= 0 && value <= maxUserQuotaValue
+}
+
+func isValidQuotaAddition(current int, delta int) bool {
+ if delta <= 0 || delta > maxUserQuotaValue {
+ return false
+ }
+ return current <= maxUserQuotaValue-delta
+}
+
// ManageUser Only admin user can do this
func ManageUser(c *gin.Context) {
var req ManageRequest
@@ -1046,6 +1082,10 @@ func ManageUser(c *gin.Context) {
common.ApiErrorI18n(c, i18n.MsgUserQuotaChangeZero)
return
}
+ if !isValidQuotaAddition(user.Quota, req.Value) {
+ common.ApiErrorI18n(c, i18n.MsgInvalidParams)
+ return
+ }
if err := model.IncreaseUserQuota(user.Id, req.Value, true); err != nil {
common.ApiError(c, err)
return
@@ -1066,6 +1106,10 @@ func ManageUser(c *gin.Context) {
"quota": logger.LogQuota(req.Value),
})
case "override":
+ if !isValidQuotaOverride(req.Value) {
+ common.ApiErrorI18n(c, i18n.MsgInvalidParams)
+ return
+ }
oldQuota := user.Quota
if err := model.DB.Model(&model.User{}).Where("id = ?", user.Id).Update("quota", req.Value).Error; err != nil {
common.ApiError(c, err)
@@ -1147,9 +1191,13 @@ func EmailBind(c *gin.Context) {
return
}
session := sessions.Default(c)
- id := session.Get("id")
+ id, ok := sessionUserID(session.Get("id"))
+ if !ok {
+ common.ApiErrorMsg(c, "用户未登录或登录状态已失效")
+ return
+ }
user := model.User{
- Id: id.(int),
+ Id: id,
}
err := user.FillUserById()
if err != nil {
@@ -1178,6 +1226,11 @@ type topUpRequest struct {
var topUpLocks sync.Map
var topUpCreateLock sync.Mutex
+type topUpLockEntry struct {
+ lock *topUpTryLock
+ refs int
+}
+
type topUpTryLock struct {
ch chan struct{}
}
@@ -1202,18 +1255,29 @@ func (l *topUpTryLock) Unlock() {
}
}
-func getTopUpLock(userID int) *topUpTryLock {
+func acquireTopUpLock(userID int) *topUpLockEntry {
+ topUpCreateLock.Lock()
+ defer topUpCreateLock.Unlock()
if v, ok := topUpLocks.Load(userID); ok {
- return v.(*topUpTryLock)
+ entry := v.(*topUpLockEntry)
+ entry.refs++
+ return entry
}
+ entry := &topUpLockEntry{lock: newTopUpTryLock(), refs: 1}
+ topUpLocks.Store(userID, entry)
+ return entry
+}
+
+func releaseTopUpLock(userID int, entry *topUpLockEntry) {
topUpCreateLock.Lock()
defer topUpCreateLock.Unlock()
- if v, ok := topUpLocks.Load(userID); ok {
- return v.(*topUpTryLock)
+ entry.refs--
+ if entry.refs > 0 {
+ return
+ }
+ if current, ok := topUpLocks.Load(userID); ok && current == entry {
+ topUpLocks.Delete(userID)
}
- l := newTopUpTryLock()
- topUpLocks.Store(userID, l)
- return l
}
func TopUp(c *gin.Context) {
@@ -1223,12 +1287,13 @@ func TopUp(c *gin.Context) {
}
id := c.GetInt("id")
- lock := getTopUpLock(id)
- if !lock.TryLock() {
+ lockEntry := acquireTopUpLock(id)
+ defer releaseTopUpLock(id, lockEntry)
+ if !lockEntry.lock.TryLock() {
common.ApiErrorI18n(c, i18n.MsgUserTopUpProcessing)
return
}
- defer lock.Unlock()
+ defer lockEntry.lock.Unlock()
req := topUpRequest{}
err := c.ShouldBindJSON(&req)
if err != nil {
@@ -1266,6 +1331,21 @@ type UpdateUserSettingRequest struct {
RecordIpLog bool `json:"record_ip_log"`
}
+func normalizeNotificationEmail(email string) (string, bool) {
+ email = strings.TrimSpace(email)
+ addr, err := mail.ParseAddress(email)
+ if err != nil || addr.Address != email {
+ return "", false
+ }
+ local, domain, ok := strings.Cut(addr.Address, "@")
+ if !ok || local == "" || domain == "" ||
+ strings.HasPrefix(domain, ".") || strings.HasSuffix(domain, ".") ||
+ strings.Contains(domain, "..") || !strings.Contains(domain, ".") {
+ return "", false
+ }
+ return addr.Address, true
+}
+
func UpdateUserSetting(c *gin.Context) {
var req UpdateUserSettingRequest
if err := c.ShouldBindJSON(&req); err != nil {
@@ -1300,11 +1380,12 @@ func UpdateUserSetting(c *gin.Context) {
// 如果是邮件类型,验证邮箱地址
if req.QuotaWarningType == dto.NotifyTypeEmail && req.NotificationEmail != "" {
- // 验证邮箱格式
- if !strings.Contains(req.NotificationEmail, "@") {
+ email, ok := normalizeNotificationEmail(req.NotificationEmail)
+ if !ok {
common.ApiErrorI18n(c, i18n.MsgSettingEmailInvalid)
return
}
+ req.NotificationEmail = email
}
// 如果是Bark类型,验证Bark URL
diff --git a/controller/user_setting_test.go b/controller/user_setting_test.go
index 8e441c99..52dff8b3 100644
--- a/controller/user_setting_test.go
+++ b/controller/user_setting_test.go
@@ -9,6 +9,7 @@ import (
"testing"
"github.com/MAX-API-Next/MAX-API/common"
+ "github.com/MAX-API-Next/MAX-API/constant"
"github.com/MAX-API-Next/MAX-API/dto"
"github.com/MAX-API-Next/MAX-API/model"
"github.com/gin-gonic/gin"
@@ -126,3 +127,103 @@ func TestUpdateUserSettingPreservesUnrelatedSettings(t *testing.T) {
assert.Empty(t, setting.GotifyToken)
assert.Zero(t, setting.GotifyPriority)
}
+
+func TestNormalizeNotificationEmailRejectsMalformedAddress(t *testing.T) {
+ validEmail, ok := normalizeNotificationEmail("new@example.com")
+ require.True(t, ok)
+ require.Equal(t, "new@example.com", validEmail)
+
+ for _, email := range []string{"a@", "@b", "@.", "a@b", "Display "} {
+ _, ok := normalizeNotificationEmail(email)
+ require.False(t, ok, "email should be rejected: %s", email)
+ }
+}
+
+func TestTopUpLockReleaseDeletesIdleEntry(t *testing.T) {
+ topUpLocks.Range(func(key, value any) bool {
+ topUpLocks.Delete(key)
+ return true
+ })
+
+ entry := acquireTopUpLock(1001)
+ require.True(t, entry.lock.TryLock())
+ entry.lock.Unlock()
+ releaseTopUpLock(1001, entry)
+
+ _, ok := topUpLocks.Load(1001)
+ require.False(t, ok)
+}
+
+func TestTopUpLockReleaseKeepsReferencedEntry(t *testing.T) {
+ topUpLocks.Range(func(key, value any) bool {
+ topUpLocks.Delete(key)
+ return true
+ })
+
+ first := acquireTopUpLock(1002)
+ second := acquireTopUpLock(1002)
+ require.Same(t, first, second)
+
+ releaseTopUpLock(1002, first)
+ _, ok := topUpLocks.Load(1002)
+ require.True(t, ok)
+
+ releaseTopUpLock(1002, second)
+ _, ok = topUpLocks.Load(1002)
+ require.False(t, ok)
+}
+
+func TestRegisterConsumesEmailVerificationCode(t *testing.T) {
+ setupUserSettingControllerTestDB(t)
+
+ originalRegisterEnabled := common.RegisterEnabled
+ originalPasswordRegisterEnabled := common.PasswordRegisterEnabled
+ originalEmailVerificationEnabled := common.EmailVerificationEnabled
+ originalQuotaForNewUser := common.QuotaForNewUser
+ originalGenerateDefaultToken := constant.GenerateDefaultToken
+ common.RegisterEnabled = true
+ common.PasswordRegisterEnabled = true
+ common.EmailVerificationEnabled = true
+ common.QuotaForNewUser = 0
+ constant.GenerateDefaultToken = false
+ t.Cleanup(func() {
+ common.RegisterEnabled = originalRegisterEnabled
+ common.PasswordRegisterEnabled = originalPasswordRegisterEnabled
+ common.EmailVerificationEnabled = originalEmailVerificationEnabled
+ common.QuotaForNewUser = originalQuotaForNewUser
+ constant.GenerateDefaultToken = originalGenerateDefaultToken
+ })
+
+ email := "register-code@example.com"
+ code := "123456"
+ common.RegisterVerificationCodeWithKey(email, code, common.EmailVerificationPurpose)
+ payloadBytes, err := common.Marshal(model.User{
+ Username: "registercode",
+ Password: "password123",
+ Email: email,
+ VerificationCode: code,
+ })
+ require.NoError(t, err)
+
+ recorder := httptest.NewRecorder()
+ ctx, _ := gin.CreateTestContext(recorder)
+ ctx.Request = httptest.NewRequest(http.MethodPost, "/api/user/register", bytes.NewReader(payloadBytes))
+ ctx.Request.Header.Set("Content-Type", "application/json")
+
+ Register(ctx)
+
+ require.Equal(t, http.StatusOK, recorder.Code)
+ require.False(t, common.VerifyCodeWithKey(email, code, common.EmailVerificationPurpose))
+}
+
+func TestQuotaBoundsValidation(t *testing.T) {
+ require.True(t, isValidQuotaOverride(0))
+ require.True(t, isValidQuotaOverride(maxUserQuotaValue))
+ require.False(t, isValidQuotaOverride(-1))
+ require.False(t, isValidQuotaOverride(maxUserQuotaValue+1))
+
+ require.True(t, isValidQuotaAddition(maxUserQuotaValue-1, 1))
+ require.False(t, isValidQuotaAddition(maxUserQuotaValue, 1))
+ require.False(t, isValidQuotaAddition(0, maxUserQuotaValue+1))
+ require.False(t, isValidQuotaAddition(0, 0))
+}
diff --git a/controller/wechat.go b/controller/wechat.go
index 2c73114d..f20662f2 100644
--- a/controller/wechat.go
+++ b/controller/wechat.go
@@ -159,9 +159,13 @@ func WeChatBind(c *gin.Context) {
return
}
session := sessions.Default(c)
- id := session.Get("id")
+ id, ok := sessionUserID(session.Get("id"))
+ if !ok {
+ common.ApiErrorMsg(c, "用户未登录或登录状态已失效")
+ return
+ }
user := model.User{
- Id: id.(int),
+ Id: id,
}
err = user.FillUserById()
if err != nil {
diff --git a/main.go b/main.go
index 7d4b1398..e81cb9a8 100644
--- a/main.go
+++ b/main.go
@@ -236,6 +236,9 @@ func main() {
shutdownHTTPServer(ctx, srv)
if common.DataExportEnabled {
saveTimeout := time.Duration(common.GetEnvOrDefault("QUOTA_DATA_CACHE_SAVE_TIMEOUT_SECONDS", 30)) * time.Second
+ if !runWithTimeout(saveTimeout, model.WaitPendingLogQuotaData) {
+ common.SysError(fmt.Sprintf("timed out waiting for pending quota data export after %s", saveTimeout))
+ }
if !runWithTimeout(saveTimeout, model.SaveQuotaDataCache) {
common.SysError(fmt.Sprintf("timed out waiting for quota data cache save after %s", saveTimeout))
}
diff --git a/middleware/model-rate-limit.go b/middleware/model-rate-limit.go
index aa1959ac..0e97c903 100644
--- a/middleware/model-rate-limit.go
+++ b/middleware/model-rate-limit.go
@@ -144,10 +144,8 @@ func memoryRateLimitHandler(duration int64, totalMaxCount, successMaxCount int)
return
}
- // 2. 检查成功请求数限制
- // 使用一个临时key来检查限制,这样可以避免实际记录
- checkKey := successKey + "_check"
- if !inMemoryRateLimiter.Request(checkKey, successMaxCount, duration) {
+ // 2. 检查成功请求数限制,但不记录本次尝试。
+ if successMaxCount > 0 && !inMemoryRateLimiter.Allow(successKey, successMaxCount, duration) {
c.Status(http.StatusTooManyRequests)
c.Abort()
return
@@ -157,7 +155,7 @@ func memoryRateLimitHandler(duration int64, totalMaxCount, successMaxCount int)
c.Next()
// 4. 如果请求成功,记录到实际的成功请求计数中
- if c.Writer.Status() < 400 {
+ if successMaxCount > 0 && c.Writer.Status() < 400 {
inMemoryRateLimiter.Request(successKey, successMaxCount, duration)
}
}
diff --git a/middleware/model_rate_limit_test.go b/middleware/model_rate_limit_test.go
new file mode 100644
index 00000000..0a7915ca
--- /dev/null
+++ b/middleware/model_rate_limit_test.go
@@ -0,0 +1,51 @@
+package middleware
+
+import (
+ "net/http"
+ "net/http/httptest"
+ "testing"
+
+ "github.com/MAX-API-Next/MAX-API/common"
+ "github.com/MAX-API-Next/MAX-API/setting"
+ "github.com/gin-gonic/gin"
+ "github.com/stretchr/testify/require"
+)
+
+func TestMemoryModelSuccessRateLimitDoesNotCountFailures(t *testing.T) {
+ gin.SetMode(gin.TestMode)
+
+ oldRedisEnabled := common.RedisEnabled
+ oldDuration := setting.ModelRequestRateLimitDurationMinutes
+ common.RedisEnabled = false
+ setting.ModelRequestRateLimitDurationMinutes = 1
+ t.Cleanup(func() {
+ common.RedisEnabled = oldRedisEnabled
+ setting.ModelRequestRateLimitDurationMinutes = oldDuration
+ })
+
+ status := http.StatusInternalServerError
+ router := gin.New()
+ router.Use(func(c *gin.Context) {
+ c.Set("id", 770001)
+ c.Next()
+ })
+ router.Use(memoryRateLimitHandler(60, 0, 1))
+ router.GET("/", func(c *gin.Context) {
+ c.Status(status)
+ })
+
+ for range 3 {
+ recorder := httptest.NewRecorder()
+ router.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil))
+ require.Equal(t, http.StatusInternalServerError, recorder.Code)
+ }
+
+ status = http.StatusOK
+ recorder := httptest.NewRecorder()
+ router.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil))
+ require.Equal(t, http.StatusOK, recorder.Code)
+
+ recorder = httptest.NewRecorder()
+ router.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil))
+ require.Equal(t, http.StatusTooManyRequests, recorder.Code)
+}
diff --git a/model/log.go b/model/log.go
index 87af760c..9805ff64 100644
--- a/model/log.go
+++ b/model/log.go
@@ -5,12 +5,14 @@ import (
"errors"
"fmt"
"strings"
+ "sync"
"time"
"github.com/MAX-API-Next/MAX-API/common"
"github.com/MAX-API-Next/MAX-API/logger"
"github.com/MAX-API-Next/MAX-API/types"
+ "github.com/bytedance/gopkg/util/gopool"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
@@ -91,6 +93,8 @@ const (
LogFilterEmptyRetry = "empty_retry"
)
+var ErrLogRetryMarkerBackfillIncomplete = errors.New("log retry marker backfill is not completed")
+
const (
LogQuotaFilterAbnormal = "abnormal"
LogQuotaFilterZero = "zero"
@@ -150,25 +154,28 @@ func applyLogTypeFilter(tx *gorm.DB, logType int) *gorm.DB {
var ensureLogRetryMarkerBackfillCompletedForRead = ensureLogRetryMarkerBackfillCompletedForReadDefault
-func applyRetryLogFilter(tx *gorm.DB) (*gorm.DB, error) {
+func applyRetryColumnFilter(tx *gorm.DB, column string) (*gorm.DB, error) {
if err := ensureLogRetryMarkerBackfillCompletedForRead(); err != nil {
return nil, err
}
- return tx.Where("logs.is_retry = ?", true), nil
+ switch column {
+ case "is_retry", "is_error_retry", "is_empty_retry":
+ return tx.Where("logs."+column+" = ?", true), nil
+ default:
+ return nil, fmt.Errorf("invalid retry log filter column: %s", column)
+ }
+}
+
+func applyRetryLogFilter(tx *gorm.DB) (*gorm.DB, error) {
+ return applyRetryColumnFilter(tx, "is_retry")
}
func applyErrorRetryLogFilter(tx *gorm.DB) (*gorm.DB, error) {
- if err := ensureLogRetryMarkerBackfillCompletedForRead(); err != nil {
- return nil, err
- }
- return tx.Where("logs.is_error_retry = ?", true), nil
+ return applyRetryColumnFilter(tx, "is_error_retry")
}
func applyEmptyRetryLogFilter(tx *gorm.DB) (*gorm.DB, error) {
- if err := ensureLogRetryMarkerBackfillCompletedForRead(); err != nil {
- return nil, err
- }
- return tx.Where("logs.is_empty_retry = ?", true), nil
+ return applyRetryColumnFilter(tx, "is_empty_retry")
}
func ensureLogRetryMarkerBackfillCompletedForReadDefault() error {
@@ -179,10 +186,39 @@ func ensureLogRetryMarkerBackfillCompletedForReadDefault() error {
if completed {
return nil
}
- if err := backfillLogRetryMarker(); err != nil {
- return fmt.Errorf("failed to backfill log retry markers before retry log read: %w", err)
+ return fmt.Errorf("%w; retry log filters will be available after startup backfill finishes", ErrLogRetryMarkerBackfillIncomplete)
+}
+
+var (
+ logQuotaDataAsyncRunner = func(fn func()) {
+ gopool.Go(fn)
}
- return nil
+ logQuotaDataAsyncWG sync.WaitGroup
+ logQuotaDataShutdownMu sync.Mutex
+ logQuotaDataShutdownStarted bool
+)
+
+func enqueueLogQuotaData(params QuotaDataLogParams) {
+ logQuotaDataShutdownMu.Lock()
+ if logQuotaDataShutdownStarted {
+ logQuotaDataShutdownMu.Unlock()
+ LogQuotaData(params)
+ SaveQuotaDataCache()
+ return
+ }
+ logQuotaDataAsyncWG.Add(1)
+ logQuotaDataShutdownMu.Unlock()
+ logQuotaDataAsyncRunner(func() {
+ defer logQuotaDataAsyncWG.Done()
+ LogQuotaData(params)
+ })
+}
+
+func WaitPendingLogQuotaData() {
+ logQuotaDataShutdownMu.Lock()
+ logQuotaDataShutdownStarted = true
+ logQuotaDataShutdownMu.Unlock()
+ logQuotaDataAsyncWG.Wait()
}
func applyLogFilter(tx *gorm.DB, filter string) (*gorm.DB, error) {
@@ -577,7 +613,7 @@ func RecordConsumeLog(c *gin.Context, userId int, params RecordConsumeLogParams)
}
if common.DataExportEnabled {
createdAt := common.GetTimestamp()
- LogQuotaData(QuotaDataLogParams{
+ enqueueLogQuotaData(QuotaDataLogParams{
UserID: userId,
Username: username,
ModelName: params.ModelName,
@@ -640,7 +676,7 @@ func RecordTaskBillingLog(params RecordTaskBillingLogParams) {
if nodeName == "" {
nodeName = common.NodeName
}
- LogQuotaData(QuotaDataLogParams{
+ enqueueLogQuotaData(QuotaDataLogParams{
UserID: params.UserId,
Username: username,
ModelName: params.ModelName,
diff --git a/model/log_test.go b/model/log_test.go
index 36af575e..c3550cab 100644
--- a/model/log_test.go
+++ b/model/log_test.go
@@ -2,11 +2,14 @@ package model
import (
"errors"
+ "net/http"
+ "net/http/httptest"
"os"
"testing"
"time"
"github.com/MAX-API-Next/MAX-API/common"
+ "github.com/gin-gonic/gin"
"github.com/glebarez/sqlite"
"github.com/stretchr/testify/require"
"gorm.io/gorm"
@@ -24,6 +27,48 @@ func withLogAuditSettings(t *testing.T, requestEnabled bool, responseEnabled boo
})
}
+func markRetryLogBackfillCompletedForTest(t *testing.T) {
+ t.Helper()
+ markerKey := logRetryMarkerBackfillCompletionKey()
+ require.NoError(t, DB.Where(commonKeyCol+" = ?", markerKey).Delete(&Option{}).Error)
+ require.NoError(t, markLogRetryMarkerBackfillCompleted())
+ t.Cleanup(func() {
+ require.NoError(t, DB.Where(commonKeyCol+" = ?", markerKey).Delete(&Option{}).Error)
+ })
+}
+
+func resetQuotaDataCacheForTest(t *testing.T) {
+ t.Helper()
+ resetLogQuotaDataShutdownForTest(t)
+ CacheQuotaDataLock.Lock()
+ CacheQuotaData = make(map[string]*QuotaData)
+ CacheQuotaDataLock.Unlock()
+ t.Cleanup(func() {
+ CacheQuotaDataLock.Lock()
+ CacheQuotaData = make(map[string]*QuotaData)
+ CacheQuotaDataLock.Unlock()
+ })
+}
+
+func resetLogQuotaDataShutdownForTest(t *testing.T) {
+ t.Helper()
+ logQuotaDataShutdownMu.Lock()
+ logQuotaDataShutdownStarted = false
+ logQuotaDataShutdownMu.Unlock()
+ t.Cleanup(func() {
+ logQuotaDataShutdownMu.Lock()
+ logQuotaDataShutdownStarted = false
+ logQuotaDataShutdownMu.Unlock()
+ })
+}
+
+func newLogTestContext() *gin.Context {
+ recorder := httptest.NewRecorder()
+ c, _ := gin.CreateTestContext(recorder)
+ c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil)
+ return c
+}
+
func TestLogQuotaFilterIndexes(t *testing.T) {
db := newRetryBackfillTestDB(t, &Log{})
@@ -270,6 +315,7 @@ func TestRetryFilterIgnoresNestedRetryMarker(t *testing.T) {
t.Cleanup(func() {
require.NoError(t, LOG_DB.Where("1 = 1").Delete(&Log{}).Error)
})
+ markRetryLogBackfillCompletedForTest(t)
logs := []Log{
{
@@ -313,7 +359,7 @@ func TestRetryFilterIgnoresNestedRetryMarker(t *testing.T) {
require.Equal(t, 16, stat.Tpm)
}
-func TestRetryFilterBackfillsLegacyMarkersBeforeCompletion(t *testing.T) {
+func TestRetryFilterReturnsReadinessErrorBeforeBackfillCompletion(t *testing.T) {
markerKey := logRetryMarkerBackfillCompletionKey()
require.NoError(t, LOG_DB.Where("1 = 1").Delete(&Log{}).Error)
require.NoError(t, DB.Where(commonKeyCol+" = ?", markerKey).Delete(&Option{}).Error)
@@ -347,23 +393,26 @@ func TestRetryFilterBackfillsLegacyMarkersBeforeCompletion(t *testing.T) {
},
}
require.NoError(t, LOG_DB.Create(&logs).Error)
- require.NoError(t, LOG_DB.Model(&Log{}).Where("1 = 1").UpdateColumn("is_retry", false).Error)
+ require.NoError(t, LOG_DB.Model(&Log{}).Where("1 = 1").Updates(map[string]interface{}{
+ "is_retry": false,
+ "is_error_retry": false,
+ "is_empty_retry": false,
+ }).Error)
got, total, err := GetAllLogs(LogQueryParams{LogType: LogTypeUnknown, LogFilter: LogFilterRetry, Num: 10})
- require.NoError(t, err)
- require.EqualValues(t, 2, total)
- require.Len(t, got, 2)
- require.ElementsMatch(t, []int{logs[0].Id, logs[1].Id}, []int{got[0].Id, got[1].Id})
+ require.ErrorIs(t, err, ErrLogRetryMarkerBackfillIncomplete)
+ require.Nil(t, got)
+ require.Zero(t, total)
var reloaded []Log
require.NoError(t, LOG_DB.Order("id asc").Find(&reloaded).Error)
require.Len(t, reloaded, 2)
- require.True(t, reloaded[0].IsRetry)
- require.True(t, reloaded[0].IsErrorRetry)
+ require.False(t, reloaded[0].IsRetry)
+ require.False(t, reloaded[0].IsErrorRetry)
require.False(t, reloaded[0].IsEmptyRetry)
- require.True(t, reloaded[1].IsRetry)
+ require.False(t, reloaded[1].IsRetry)
require.False(t, reloaded[1].IsErrorRetry)
- require.True(t, reloaded[1].IsEmptyRetry)
+ require.False(t, reloaded[1].IsEmptyRetry)
}
func TestRetryFilterUsesIsRetryAfterBackfillCompletion(t *testing.T) {
@@ -683,6 +732,167 @@ func TestLogRetryMarkerIsRecomputedWhenOtherChanges(t *testing.T) {
require.False(t, reloaded.IsRetry)
}
+func TestRecordConsumeLogQueuesQuotaDataAsync(t *testing.T) {
+ require.NoError(t, LOG_DB.Where("1 = 1").Delete(&Log{}).Error)
+ t.Cleanup(func() {
+ require.NoError(t, LOG_DB.Where("1 = 1").Delete(&Log{}).Error)
+ })
+ resetQuotaDataCacheForTest(t)
+
+ originalDataExportEnabled := common.DataExportEnabled
+ originalRunner := logQuotaDataAsyncRunner
+ common.DataExportEnabled = true
+ var queued []func()
+ executed := false
+ logQuotaDataAsyncRunner = func(fn func()) {
+ queued = append(queued, func() {
+ executed = true
+ fn()
+ })
+ }
+ t.Cleanup(func() {
+ common.DataExportEnabled = originalDataExportEnabled
+ logQuotaDataAsyncRunner = originalRunner
+ })
+
+ RecordConsumeLog(newLogTestContext(), 7, RecordConsumeLogParams{
+ ModelName: "gpt-test",
+ Quota: 42,
+ PromptTokens: 3,
+ CompletionTokens: 5,
+ Group: "default",
+ TokenId: 11,
+ ChannelId: 13,
+ })
+
+ require.Len(t, queued, 1)
+ require.False(t, executed)
+ require.Empty(t, CacheQuotaData)
+
+ queued[0]()
+ require.True(t, executed)
+ require.Len(t, CacheQuotaData, 1)
+}
+
+func TestRecordTaskBillingLogQueuesQuotaDataAsync(t *testing.T) {
+ require.NoError(t, LOG_DB.Where("1 = 1").Delete(&Log{}).Error)
+ t.Cleanup(func() {
+ require.NoError(t, LOG_DB.Where("1 = 1").Delete(&Log{}).Error)
+ })
+ resetQuotaDataCacheForTest(t)
+
+ originalDataExportEnabled := common.DataExportEnabled
+ originalRunner := logQuotaDataAsyncRunner
+ common.DataExportEnabled = true
+ var queued []func()
+ logQuotaDataAsyncRunner = func(fn func()) {
+ queued = append(queued, fn)
+ }
+ t.Cleanup(func() {
+ common.DataExportEnabled = originalDataExportEnabled
+ logQuotaDataAsyncRunner = originalRunner
+ })
+
+ RecordTaskBillingLog(RecordTaskBillingLogParams{
+ UserId: 7,
+ LogType: LogTypeConsume,
+ Content: "task billing",
+ ChannelId: 13,
+ ModelName: "task-test",
+ Quota: 42,
+ TokenId: 11,
+ Group: "default",
+ })
+
+ require.Len(t, queued, 1)
+ require.Empty(t, CacheQuotaData)
+
+ queued[0]()
+ require.Len(t, CacheQuotaData, 1)
+}
+
+func TestWaitPendingLogQuotaDataDrainsEnqueuedWork(t *testing.T) {
+ resetQuotaDataCacheForTest(t)
+
+ originalRunner := logQuotaDataAsyncRunner
+ release := make(chan struct{})
+ logQuotaDataAsyncRunner = func(fn func()) {
+ go func() {
+ <-release
+ fn()
+ }()
+ }
+ t.Cleanup(func() {
+ logQuotaDataAsyncRunner = originalRunner
+ })
+
+ enqueueLogQuotaData(QuotaDataLogParams{
+ UserID: 7,
+ Username: "quota-wait",
+ ModelName: "gpt-test",
+ Quota: 42,
+ CreatedAt: time.Now().Unix(),
+ })
+
+ waitDone := make(chan struct{})
+ go func() {
+ WaitPendingLogQuotaData()
+ close(waitDone)
+ }()
+
+ select {
+ case <-waitDone:
+ t.Fatal("wait returned before queued quota data ran")
+ case <-time.After(10 * time.Millisecond):
+ }
+
+ close(release)
+
+ select {
+ case <-waitDone:
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for queued quota data")
+ }
+ require.Len(t, CacheQuotaData, 1)
+}
+
+func TestEnqueueLogQuotaDataAfterShutdownPersistsSynchronously(t *testing.T) {
+ resetQuotaDataCacheForTest(t)
+ require.NoError(t, DB.AutoMigrate(&QuotaData{}))
+ require.NoError(t, DB.Where("1 = 1").Delete(&QuotaData{}).Error)
+ t.Cleanup(func() {
+ require.NoError(t, DB.Where("1 = 1").Delete(&QuotaData{}).Error)
+ })
+
+ originalRunner := logQuotaDataAsyncRunner
+ runnerCalled := false
+ logQuotaDataAsyncRunner = func(fn func()) {
+ runnerCalled = true
+ }
+ logQuotaDataShutdownMu.Lock()
+ logQuotaDataShutdownStarted = true
+ logQuotaDataShutdownMu.Unlock()
+ t.Cleanup(func() {
+ logQuotaDataAsyncRunner = originalRunner
+ })
+
+ enqueueLogQuotaData(QuotaDataLogParams{
+ UserID: 7,
+ Username: "quota-shutdown",
+ ModelName: "gpt-test",
+ Quota: 42,
+ CreatedAt: time.Now().Unix(),
+ })
+
+ require.False(t, runnerCalled)
+ require.Empty(t, CacheQuotaData)
+
+ var stored QuotaData
+ require.NoError(t, DB.Where("user_id = ? AND username = ? AND model_name = ?", 7, "quota-shutdown", "gpt-test").First(&stored).Error)
+ require.Equal(t, 1, stored.Count)
+ require.Equal(t, 42, stored.Quota)
+}
+
func newRetryBackfillTestDB(t *testing.T, models ...interface{}) *gorm.DB {
t.Helper()
@@ -703,6 +913,7 @@ func newRetryBackfillTestDB(t *testing.T, models ...interface{}) *gorm.DB {
func createRetryFilterLogs(t *testing.T) []Log {
t.Helper()
+ markRetryLogBackfillCompletedForTest(t)
now := time.Now().Unix()
logs := []Log{
diff --git a/model/redemption.go b/model/redemption.go
index 629bbe49..a56f0a29 100644
--- a/model/redemption.go
+++ b/model/redemption.go
@@ -4,6 +4,7 @@ import (
"errors"
"fmt"
"strconv"
+ "strings"
"github.com/MAX-API-Next/MAX-API/common"
"github.com/MAX-API-Next/MAX-API/logger"
@@ -60,7 +61,7 @@ func GetAllRedemptions(startIdx int, num int) (redemptions []*Redemption, total
return redemptions, total, nil
}
-func SearchRedemptions(keyword string, startIdx int, num int) (redemptions []*Redemption, total int64, err error) {
+func SearchRedemptions(keyword string, status string, startIdx int, num int) (redemptions []*Redemption, total int64, err error) {
tx := DB.Begin()
if tx.Error != nil {
return nil, 0, tx.Error
@@ -71,14 +72,38 @@ func SearchRedemptions(keyword string, startIdx int, num int) (redemptions []*Re
}
}()
- // Build query based on keyword type
query := tx.Model(&Redemption{})
+ keyword = strings.TrimSpace(keyword)
+ status = strings.TrimSpace(status)
+
+ if keyword != "" {
+ if id, err := strconv.Atoi(keyword); err == nil {
+ query = query.Where("id = ? OR name LIKE ?", id, keyword+"%")
+ } else {
+ query = query.Where("name LIKE ?", keyword+"%")
+ }
+ }
- // Only try to convert to ID if the string represents a valid integer
- if id, err := strconv.Atoi(keyword); err == nil {
- query = query.Where("id = ? OR name LIKE ?", id, keyword+"%")
- } else {
- query = query.Where("name LIKE ?", keyword+"%")
+ if status != "" {
+ now := common.GetTimestamp()
+ switch status {
+ case "expired":
+ query = query.Where(
+ "status = ? AND expired_time != 0 AND expired_time < ?",
+ common.RedemptionCodeStatusEnabled,
+ now,
+ )
+ case strconv.Itoa(common.RedemptionCodeStatusEnabled):
+ query = query.Where(
+ "status = ? AND (expired_time = 0 OR expired_time >= ?)",
+ common.RedemptionCodeStatusEnabled,
+ now,
+ )
+ case strconv.Itoa(common.RedemptionCodeStatusDisabled):
+ query = query.Where("status = ?", common.RedemptionCodeStatusDisabled)
+ case strconv.Itoa(common.RedemptionCodeStatusUsed):
+ query = query.Where("status = ?", common.RedemptionCodeStatusUsed)
+ }
}
// Get total count
diff --git a/model/subscription.go b/model/subscription.go
index 99cefe82..910f1188 100644
--- a/model/subscription.go
+++ b/model/subscription.go
@@ -342,6 +342,16 @@ type SubscriptionSummary struct {
Subscription *UserSubscription `json:"subscription"`
}
+type SubscriptionResetResult struct {
+ PlanId int `json:"plan_id"`
+ MatchedCount int `json:"matched_count"`
+ ResetCount int `json:"reset_count"`
+ UserCount int `json:"user_count"`
+ AdvanceResetTime bool `json:"advance_reset_time"`
+ PlanTitle string `json:"-"`
+ AffectedUserIds []int `json:"-"`
+}
+
func calcPlanEndTime(start time.Time, plan *SubscriptionPlan) (int64, error) {
if plan == nil {
return 0, errors.New("plan is nil")
@@ -985,6 +995,125 @@ func AdminDeleteUserSubscription(userSubscriptionId int) (string, error) {
return "", nil
}
+func resetUserSubscriptionTx(tx *gorm.DB, sub *UserSubscription, plan *SubscriptionPlan, now int64, advanceResetTime bool) error {
+ if tx == nil || sub == nil || plan == nil {
+ return errors.New("invalid reset args")
+ }
+ sub.AmountUsed = 0
+ if advanceResetTime {
+ nextReset := calcNextResetTime(time.Unix(now, 0), plan, sub.EndTime)
+ sub.NextResetTime = nextReset
+ if nextReset > 0 {
+ sub.LastResetTime = now
+ } else {
+ sub.LastResetTime = 0
+ }
+ }
+ return tx.Save(sub).Error
+}
+
+func buildSubscriptionResetResult(plan *SubscriptionPlan, subs []UserSubscription, advanceResetTime bool) *SubscriptionResetResult {
+ userIds := make([]int, 0, len(subs))
+ seenUsers := make(map[int]struct{}, len(subs))
+ for _, sub := range subs {
+ if _, ok := seenUsers[sub.UserId]; ok {
+ continue
+ }
+ seenUsers[sub.UserId] = struct{}{}
+ userIds = append(userIds, sub.UserId)
+ }
+ return &SubscriptionResetResult{
+ PlanId: plan.Id,
+ MatchedCount: len(subs),
+ ResetCount: len(subs),
+ UserCount: len(userIds),
+ AdvanceResetTime: advanceResetTime,
+ PlanTitle: plan.Title,
+ AffectedUserIds: userIds,
+ }
+}
+
+func adminResetUserSubscriptionsByPlanTx(tx *gorm.DB, userId int, plan *SubscriptionPlan, now int64, advanceResetTime bool) (*SubscriptionResetResult, error) {
+ if tx == nil || plan == nil {
+ return nil, errors.New("invalid reset args")
+ }
+ var subs []UserSubscription
+ if err := withRowLock(tx).
+ Where("user_id = ? AND plan_id = ? AND status = ? AND end_time > ?", userId, plan.Id, "active", now).
+ Order("end_time asc, id asc").
+ Find(&subs).Error; err != nil {
+ return nil, err
+ }
+ if len(subs) == 0 {
+ return nil, errors.New("该用户没有有效的此套餐订阅")
+ }
+ for i := range subs {
+ if err := resetUserSubscriptionTx(tx, &subs[i], plan, now, advanceResetTime); err != nil {
+ return nil, err
+ }
+ }
+ return buildSubscriptionResetResult(plan, subs, advanceResetTime), nil
+}
+
+func adminResetPlanSubscriptionsTx(tx *gorm.DB, plan *SubscriptionPlan, now int64, advanceResetTime bool) (*SubscriptionResetResult, error) {
+ if tx == nil || plan == nil {
+ return nil, errors.New("invalid reset args")
+ }
+ var subs []UserSubscription
+ if err := withRowLock(tx).
+ Where("plan_id = ? AND status = ? AND end_time > ?", plan.Id, "active", now).
+ Order("user_id asc, end_time asc, id asc").
+ Find(&subs).Error; err != nil {
+ return nil, err
+ }
+ for i := range subs {
+ if err := resetUserSubscriptionTx(tx, &subs[i], plan, now, advanceResetTime); err != nil {
+ return nil, err
+ }
+ }
+ return buildSubscriptionResetResult(plan, subs, advanceResetTime), nil
+}
+
+func AdminResetUserSubscriptionsByPlan(userId int, planId int, advanceResetTime bool) (*SubscriptionResetResult, error) {
+ if userId <= 0 || planId <= 0 {
+ return nil, errors.New("invalid userId or planId")
+ }
+ var result *SubscriptionResetResult
+ now := GetDBTimestamp()
+ err := DB.Transaction(func(tx *gorm.DB) error {
+ plan, err := getSubscriptionPlanByIdTx(tx, planId)
+ if err != nil {
+ return err
+ }
+ result, err = adminResetUserSubscriptionsByPlanTx(tx, userId, plan, now, advanceResetTime)
+ return err
+ })
+ if err != nil {
+ return nil, err
+ }
+ return result, nil
+}
+
+func AdminResetPlanSubscriptions(planId int, advanceResetTime bool) (*SubscriptionResetResult, error) {
+ if planId <= 0 {
+ return nil, errors.New("invalid planId")
+ }
+ var result *SubscriptionResetResult
+ now := GetDBTimestamp()
+ err := DB.Transaction(func(tx *gorm.DB) error {
+ plan, err := getSubscriptionPlanByIdTx(tx, planId)
+ if err != nil {
+ return err
+ }
+ result, err = adminResetPlanSubscriptionsTx(tx, plan, now, advanceResetTime)
+ return err
+ })
+ if err != nil {
+ return nil, err
+ }
+ return result, nil
+}
+
type SubscriptionPreConsumeResult struct {
UserSubscriptionId int
PreConsumed int64
diff --git a/model/subscription_reset_test.go b/model/subscription_reset_test.go
new file mode 100644
index 00000000..338f18a7
--- /dev/null
+++ b/model/subscription_reset_test.go
@@ -0,0 +1,201 @@
+package model
+
+import (
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func seedSubscriptionResetPlan(t *testing.T, plan *SubscriptionPlan) {
+ t.Helper()
+ require.NoError(t, DB.Create(plan).Error)
+}
+
+func seedSubscriptionResetSub(t *testing.T, sub *UserSubscription) {
+ t.Helper()
+ require.NoError(t, DB.Create(sub).Error)
+}
+
+func getSubscriptionResetSub(t *testing.T, id int) UserSubscription {
+ t.Helper()
+ var sub UserSubscription
+ require.NoError(t, DB.Where("id = ?", id).First(&sub).Error)
+ return sub
+}
+
+func TestAdminResetUserSubscriptionsByPlanResetsAllActiveMatchesAndAdvancesTime(t *testing.T) {
+ truncateTables(t)
+
+ now := GetDBTimestamp()
+ plan := &SubscriptionPlan{
+ Id: 9101,
+ Title: "Pro",
+ PriceAmount: 10,
+ DurationUnit: SubscriptionDurationMonth,
+ DurationValue: 1,
+ TotalAmount: 1000,
+ QuotaResetPeriod: SubscriptionResetDaily,
+ }
+ otherPlan := &SubscriptionPlan{
+ Id: 9102,
+ Title: "Basic",
+ PriceAmount: 1,
+ DurationUnit: SubscriptionDurationMonth,
+ DurationValue: 1,
+ TotalAmount: 100,
+ QuotaResetPeriod: SubscriptionResetDaily,
+ }
+ seedSubscriptionResetPlan(t, plan)
+ seedSubscriptionResetPlan(t, otherPlan)
+
+ activeEnd := now + 30*24*3600
+ expiredEnd := now - 1
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9201, UserId: 101, PlanId: plan.Id, AmountTotal: 1000, AmountUsed: 300, StartTime: now - 3600, EndTime: activeEnd, Status: "active", LastResetTime: now - 3600, NextResetTime: now + 120})
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9202, UserId: 101, PlanId: plan.Id, AmountTotal: 1000, AmountUsed: 500, StartTime: now - 3600, EndTime: activeEnd, Status: "active", LastResetTime: now - 3600, NextResetTime: now + 120})
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9203, UserId: 101, PlanId: otherPlan.Id, AmountTotal: 100, AmountUsed: 60, StartTime: now - 3600, EndTime: activeEnd, Status: "active", LastResetTime: now - 3600, NextResetTime: now + 120})
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9204, UserId: 101, PlanId: plan.Id, AmountTotal: 1000, AmountUsed: 700, StartTime: now - 7200, EndTime: expiredEnd, Status: "active", LastResetTime: now - 3600, NextResetTime: now - 10})
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9205, UserId: 102, PlanId: plan.Id, AmountTotal: 1000, AmountUsed: 800, StartTime: now - 3600, EndTime: activeEnd, Status: "active", LastResetTime: now - 3600, NextResetTime: now + 120})
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9206, UserId: 101, PlanId: plan.Id, AmountTotal: 1000, AmountUsed: 900, StartTime: now - 3600, EndTime: activeEnd, Status: "cancelled", LastResetTime: now - 3600, NextResetTime: now + 120})
+
+ beforeReset := GetDBTimestamp()
+ result, err := AdminResetUserSubscriptionsByPlan(101, plan.Id, true)
+ afterReset := GetDBTimestamp()
+
+ require.NoError(t, err)
+ require.NotNil(t, result)
+ assert.Equal(t, plan.Id, result.PlanId)
+ assert.Equal(t, 2, result.MatchedCount)
+ assert.Equal(t, 2, result.ResetCount)
+ assert.Equal(t, 1, result.UserCount)
+ assert.Equal(t, []int{101}, result.AffectedUserIds)
+ assert.True(t, result.AdvanceResetTime)
+
+ for _, id := range []int{9201, 9202} {
+ sub := getSubscriptionResetSub(t, id)
+ assert.Zero(t, sub.AmountUsed)
+ assert.GreaterOrEqual(t, sub.LastResetTime, beforeReset)
+ assert.LessOrEqual(t, sub.LastResetTime, afterReset)
+ assert.Equal(t, calcNextResetTime(time.Unix(sub.LastResetTime, 0), plan, sub.EndTime), sub.NextResetTime)
+ }
+ assert.EqualValues(t, 60, getSubscriptionResetSub(t, 9203).AmountUsed)
+ assert.EqualValues(t, 700, getSubscriptionResetSub(t, 9204).AmountUsed)
+ assert.EqualValues(t, 800, getSubscriptionResetSub(t, 9205).AmountUsed)
+ assert.EqualValues(t, 900, getSubscriptionResetSub(t, 9206).AmountUsed)
+}
+
+func TestAdminResetUserSubscriptionsByPlanKeepsResetTimes(t *testing.T) {
+ truncateTables(t)
+
+ now := GetDBTimestamp()
+ plan := &SubscriptionPlan{
+ Id: 9301,
+ Title: "Team",
+ PriceAmount: 20,
+ DurationUnit: SubscriptionDurationMonth,
+ DurationValue: 1,
+ TotalAmount: 2000,
+ QuotaResetPeriod: SubscriptionResetMonthly,
+ }
+ seedSubscriptionResetPlan(t, plan)
+
+ lastReset := now - 86400
+ nextReset := now + 86400
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9302, UserId: 201, PlanId: plan.Id, AmountTotal: 2000, AmountUsed: 1200, StartTime: now - 172800, EndTime: now + 30*24*3600, Status: "active", LastResetTime: lastReset, NextResetTime: nextReset})
+
+ result, err := AdminResetUserSubscriptionsByPlan(201, plan.Id, false)
+
+ require.NoError(t, err)
+ assert.False(t, result.AdvanceResetTime)
+ sub := getSubscriptionResetSub(t, 9302)
+ assert.Zero(t, sub.AmountUsed)
+ assert.Equal(t, lastReset, sub.LastResetTime)
+ assert.Equal(t, nextReset, sub.NextResetTime)
+}
+
+func TestAdminResetUserSubscriptionsByPlanNoActiveMatchReturnsError(t *testing.T) {
+ truncateTables(t)
+
+ now := GetDBTimestamp()
+ plan := &SubscriptionPlan{
+ Id: 9401,
+ Title: "Expired",
+ PriceAmount: 10,
+ DurationUnit: SubscriptionDurationMonth,
+ DurationValue: 1,
+ TotalAmount: 1000,
+ }
+ seedSubscriptionResetPlan(t, plan)
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9402, UserId: 301, PlanId: plan.Id, AmountTotal: 1000, AmountUsed: 500, StartTime: now - 7200, EndTime: now - 1, Status: "active"})
+
+ result, err := AdminResetUserSubscriptionsByPlan(301, plan.Id, true)
+
+ require.Error(t, err)
+ assert.Nil(t, result)
+ assert.True(t, strings.Contains(err.Error(), "该用户没有有效的此套餐订阅"))
+}
+
+func TestAdminResetPlanSubscriptionsResetsAllActiveUsers(t *testing.T) {
+ truncateTables(t)
+
+ now := GetDBTimestamp()
+ plan := &SubscriptionPlan{
+ Id: 9501,
+ Title: "Business",
+ PriceAmount: 30,
+ DurationUnit: SubscriptionDurationMonth,
+ DurationValue: 1,
+ TotalAmount: 3000,
+ QuotaResetPeriod: SubscriptionResetNever,
+ }
+ seedSubscriptionResetPlan(t, plan)
+
+ activeEnd := now + 30*24*3600
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9502, UserId: 401, PlanId: plan.Id, AmountTotal: 3000, AmountUsed: 1000, StartTime: now - 3600, EndTime: activeEnd, Status: "active", LastResetTime: now - 3600, NextResetTime: now + 10})
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9503, UserId: 401, PlanId: plan.Id, AmountTotal: 3000, AmountUsed: 1100, StartTime: now - 3500, EndTime: activeEnd, Status: "active", LastResetTime: now - 3600, NextResetTime: now + 10})
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9504, UserId: 402, PlanId: plan.Id, AmountTotal: 3000, AmountUsed: 1200, StartTime: now - 3400, EndTime: activeEnd, Status: "active", LastResetTime: now - 3600, NextResetTime: now + 10})
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9505, UserId: 403, PlanId: plan.Id, AmountTotal: 3000, AmountUsed: 1300, StartTime: now - 7200, EndTime: now - 1, Status: "active", LastResetTime: now - 3600, NextResetTime: now - 10})
+ seedSubscriptionResetSub(t, &UserSubscription{Id: 9506, UserId: 404, PlanId: plan.Id, AmountTotal: 3000, AmountUsed: 1400, StartTime: now - 3600, EndTime: activeEnd, Status: "cancelled", LastResetTime: now - 3600, NextResetTime: now + 10})
+
+ result, err := AdminResetPlanSubscriptions(plan.Id, true)
+
+ require.NoError(t, err)
+ require.NotNil(t, result)
+ assert.Equal(t, 3, result.MatchedCount)
+ assert.Equal(t, 3, result.ResetCount)
+ assert.Equal(t, 2, result.UserCount)
+ assert.Equal(t, []int{401, 402}, result.AffectedUserIds)
+ for _, id := range []int{9502, 9503, 9504} {
+ sub := getSubscriptionResetSub(t, id)
+ assert.Zero(t, sub.AmountUsed)
+ assert.Zero(t, sub.LastResetTime)
+ assert.Zero(t, sub.NextResetTime)
+ }
+ assert.EqualValues(t, 1300, getSubscriptionResetSub(t, 9505).AmountUsed)
+ assert.EqualValues(t, 1400, getSubscriptionResetSub(t, 9506).AmountUsed)
+}
+
+func TestAdminResetPlanSubscriptionsNoMatchSucceeds(t *testing.T) {
+ truncateTables(t)
+
+ plan := &SubscriptionPlan{
+ Id: 9601,
+ Title: "Empty",
+ PriceAmount: 10,
+ DurationUnit: SubscriptionDurationMonth,
+ DurationValue: 1,
+ TotalAmount: 1000,
+ }
+ seedSubscriptionResetPlan(t, plan)
+
+ result, err := AdminResetPlanSubscriptions(plan.Id, true)
+
+ require.NoError(t, err)
+ require.NotNil(t, result)
+ assert.Zero(t, result.MatchedCount)
+ assert.Zero(t, result.ResetCount)
+ assert.Zero(t, result.UserCount)
+ assert.Empty(t, result.AffectedUserIds)
+}
diff --git a/model/user.go b/model/user.go
index 5a6031bf..b364c02a 100644
--- a/model/user.go
+++ b/model/user.go
@@ -549,10 +549,8 @@ func HardDeleteUserById(id int) error {
if id == 0 {
return errors.New("id 为空!")
}
- if err := DB.Unscoped().Delete(&User{}, "id = ?", id).Error; err != nil {
- return err
- }
- return invalidateUserCache(id)
+ user := User{Id: id}
+ return user.HardDelete()
}
func inviteUser(inviterId int) (err error) {
@@ -1056,6 +1054,9 @@ func (user *User) Delete() error {
if user.Id == 0 {
return errors.New("id 为空!")
}
+ if err := user.ensureCanDelete(); err != nil {
+ return err
+ }
if err := DB.Delete(user).Error; err != nil {
return err
}
@@ -1068,12 +1069,33 @@ func (user *User) HardDelete() error {
if user.Id == 0 {
return errors.New("id 为空!")
}
+ if err := user.ensureCanDelete(); err != nil {
+ return err
+ }
if err := DB.Unscoped().Delete(user).Error; err != nil {
return err
}
return invalidateUserCache(user.Id)
}
+func (user *User) ensureCanDelete() error {
+ if user.Role == common.RoleRootUser {
+ return errors.New("cannot delete root user")
+ }
+ if user.Role != 0 {
+ return nil
+ }
+ var existing User
+ if err := DB.Unscoped().Select("role").First(&existing, "id = ?", user.Id).Error; err != nil {
+ return err
+ }
+ if existing.Role == common.RoleRootUser {
+ return errors.New("cannot delete root user")
+ }
+ user.Role = existing.Role
+ return nil
+}
+
// ValidateAndFill check password & user status
func (user *User) ValidateAndFill() (err error) {
// When querying with struct, GORM will only query with non-zero fields,
diff --git a/model/user_delete_test.go b/model/user_delete_test.go
new file mode 100644
index 00000000..51ed292a
--- /dev/null
+++ b/model/user_delete_test.go
@@ -0,0 +1,34 @@
+package model
+
+import (
+ "testing"
+
+ "github.com/MAX-API-Next/MAX-API/common"
+ "github.com/stretchr/testify/require"
+)
+
+func TestUserDeleteRejectsRootUserInModelLayer(t *testing.T) {
+ truncateTables(t)
+ root := User{Id: 9101, Username: "root-delete-guard", Password: "password123", Role: common.RoleRootUser, Status: common.UserStatusEnabled}
+ require.NoError(t, DB.Create(&root).Error)
+
+ err := (&User{Id: root.Id}).Delete()
+ require.Error(t, err)
+
+ var count int64
+ require.NoError(t, DB.Unscoped().Model(&User{}).Where("id = ?", root.Id).Count(&count).Error)
+ require.EqualValues(t, 1, count)
+}
+
+func TestUserHardDeleteRejectsRootUserInModelLayer(t *testing.T) {
+ truncateTables(t)
+ root := User{Id: 9102, Username: "root-hard-delete-guard", Password: "password123", Role: common.RoleRootUser, Status: common.UserStatusEnabled}
+ require.NoError(t, DB.Create(&root).Error)
+
+ err := HardDeleteUserById(root.Id)
+ require.Error(t, err)
+
+ var count int64
+ require.NoError(t, DB.Unscoped().Model(&User{}).Where("id = ?", root.Id).Count(&count).Error)
+ require.EqualValues(t, 1, count)
+}
diff --git a/relay/helper/stream_scanner.go b/relay/helper/stream_scanner.go
index 0a8fc05a..9912d51e 100644
--- a/relay/helper/stream_scanner.go
+++ b/relay/helper/stream_scanner.go
@@ -28,6 +28,11 @@ const (
streamWriteTimeout = 30 * time.Second
)
+var (
+ streamPingData = PingData
+ streamPingWriteTimeout = 10 * time.Second
+)
+
func getScannerBufferSize() int {
if constant.StreamScannerMaxBufferMB > 0 {
return constant.StreamScannerMaxBufferMB << 20
@@ -41,11 +46,22 @@ func NewStreamScanner(reader io.Reader) *bufio.Scanner {
return scanner
}
-func ExtendWriteDeadline(c *gin.Context) {
+func extendWriteDeadline(c *gin.Context, timeout time.Duration) {
if c == nil || c.Writer == nil {
return
}
- _ = http.NewResponseController(c.Writer).SetWriteDeadline(time.Now().Add(streamWriteTimeout))
+ _ = http.NewResponseController(c.Writer).SetWriteDeadline(time.Now().Add(timeout))
+}
+
+func ExtendWriteDeadline(c *gin.Context) {
+ extendWriteDeadline(c, streamWriteTimeout)
+}
+
+func pingDataWithTimeout(c *gin.Context) error {
+ // c.Writer is not safe for concurrent writes; keep the ping write in the
+ // caller's critical section and rely on the response write deadline to bound it.
+ extendWriteDeadline(c, streamPingWriteTimeout)
+ return streamPingData(c)
}
func newStreamingTimeoutTicker(timeoutSeconds int) (*time.Ticker, <-chan time.Time) {
@@ -155,12 +171,12 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
func() {
writeMutex.Lock()
defer writeMutex.Unlock()
- ExtendWriteDeadline(c)
- err = PingData(c)
+ err = pingDataWithTimeout(c)
}()
if err != nil {
logger.LogError(c, "ping data error: "+err.Error())
info.StreamStatus.SetEndReason(relaycommon.StreamEndReasonPingFail, err)
+ stop()
return
}
logger.LogDebug(c, "ping data sent")
diff --git a/relay/helper/stream_scanner_test.go b/relay/helper/stream_scanner_test.go
index bff7ae0a..2b49753b 100644
--- a/relay/helper/stream_scanner_test.go
+++ b/relay/helper/stream_scanner_test.go
@@ -389,8 +389,6 @@ func TestStreamScannerHandler_SlowUpstreamFastHandler(t *testing.T) {
// ---------- Ping tests ----------
func TestStreamScannerHandler_PingSentDuringSlowUpstream(t *testing.T) {
- t.Parallel()
-
setting := operation_setting.GetGeneralSetting()
oldEnabled := setting.PingIntervalEnabled
oldSeconds := setting.PingIntervalSeconds
@@ -443,8 +441,6 @@ func TestStreamScannerHandler_PingSentDuringSlowUpstream(t *testing.T) {
}
func TestStreamScannerHandler_PingDisabledByRelayInfo(t *testing.T) {
- t.Parallel()
-
setting := operation_setting.GetGeneralSetting()
oldEnabled := setting.PingIntervalEnabled
oldSeconds := setting.PingIntervalSeconds
@@ -497,6 +493,96 @@ func TestStreamScannerHandler_PingDisabledByRelayInfo(t *testing.T) {
assert.Equal(t, 0, pingCount, "pings should be disabled when DisablePing=true")
}
+func TestStreamScannerHandler_PingWriteStaysSynchronizedUntilComplete(t *testing.T) {
+ setting := operation_setting.GetGeneralSetting()
+ oldEnabled := setting.PingIntervalEnabled
+ oldSeconds := setting.PingIntervalSeconds
+ setting.PingIntervalEnabled = true
+ setting.PingIntervalSeconds = 1
+
+ originalPingData := streamPingData
+ pingStarted := make(chan struct{})
+ releasePing := make(chan struct{})
+ var startOnce sync.Once
+ streamPingData = func(c *gin.Context) error {
+ startOnce.Do(func() {
+ close(pingStarted)
+ })
+ <-releasePing
+ return nil
+ }
+ t.Cleanup(func() {
+ setting.PingIntervalEnabled = oldEnabled
+ setting.PingIntervalSeconds = oldSeconds
+ streamPingData = originalPingData
+ })
+
+ pr, pw := io.Pipe()
+ t.Cleanup(func() {
+ _ = pw.Close()
+ })
+ body := ¬ifyPipeReadCloser{
+ PipeReader: pr,
+ closed: make(chan struct{}),
+ }
+ recorder := httptest.NewRecorder()
+ c, _ := gin.CreateTestContext(recorder)
+ c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil)
+ resp := &http.Response{Body: body}
+ info := &relaycommon.RelayInfo{ChannelMeta: &relaycommon.ChannelMeta{}}
+
+ handlerCalled := make(chan struct{})
+ var handlerOnce sync.Once
+ done := make(chan struct{})
+ go func() {
+ StreamScannerHandler(c, resp, info, func(data string, sr *StreamResult) {
+ handlerOnce.Do(func() {
+ close(handlerCalled)
+ })
+ })
+ close(done)
+ }()
+
+ select {
+ case <-pingStarted:
+ case <-time.After(2 * time.Second):
+ t.Fatal("timed out waiting for ping to start")
+ }
+
+ go func() {
+ fmt.Fprint(pw, "data: chunk_0\n")
+ fmt.Fprint(pw, "data: [DONE]\n")
+ _ = pw.Close()
+ }()
+
+ select {
+ case <-handlerCalled:
+ t.Fatal("data handler ran while ping write was still in progress")
+ case <-time.After(50 * time.Millisecond):
+ }
+
+ close(releasePing)
+
+ require.NotNil(t, info.StreamStatus)
+ select {
+ case <-handlerCalled:
+ case <-time.After(2 * time.Second):
+ t.Fatal("data handler did not run after ping write completed")
+ }
+
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("stream did not finish after ping write completed")
+ }
+
+ select {
+ case <-body.closed:
+ default:
+ t.Fatal("upstream response body was not closed after stream completion")
+ }
+}
+
// ---------- StreamStatus integration ----------
func TestStreamScannerHandler_StreamStatus_DoneReason(t *testing.T) {
@@ -721,8 +807,6 @@ func TestStreamScannerHandler_StreamStatus_PreInitialized(t *testing.T) {
}
func TestStreamScannerHandler_PingInterleavesWithSlowUpstream(t *testing.T) {
- t.Parallel()
-
setting := operation_setting.GetGeneralSetting()
oldEnabled := setting.PingIntervalEnabled
oldSeconds := setting.PingIntervalSeconds
diff --git a/router/api-router.go b/router/api-router.go
index 284d1bf7..d8d58f0c 100644
--- a/router/api-router.go
+++ b/router/api-router.go
@@ -168,10 +168,12 @@ func SetApiRouter(router *gin.Engine) {
subscriptionAdminRoute.PUT("/plans/:id", controller.AdminUpdateSubscriptionPlan)
subscriptionAdminRoute.PATCH("/plans/:id", controller.AdminUpdateSubscriptionPlanStatus)
subscriptionAdminRoute.POST("/bind", controller.AdminBindSubscription)
+ subscriptionAdminRoute.POST("/plans/:id/subscriptions/reset", controller.AdminResetPlanSubscriptions)
// User subscription management (admin)
subscriptionAdminRoute.GET("/users/:id/subscriptions", controller.AdminListUserSubscriptions)
subscriptionAdminRoute.POST("/users/:id/subscriptions", controller.AdminCreateUserSubscription)
+ subscriptionAdminRoute.POST("/users/:id/subscriptions/reset", controller.AdminResetUserSubscriptionsByPlan)
subscriptionAdminRoute.POST("/user_subscriptions/:id/invalidate", controller.AdminInvalidateUserSubscription)
subscriptionAdminRoute.DELETE("/user_subscriptions/:id", controller.AdminDeleteUserSubscription)
}
diff --git a/service/error.go b/service/error.go
index bc635077..57e2374c 100644
--- a/service/error.go
+++ b/service/error.go
@@ -17,6 +17,8 @@ import (
"github.com/MAX-API-Next/MAX-API/types"
)
+const maxUpstreamErrorBodyBytes = 1 << 20
+
func MidjourneyErrorWrapper(code int, desc string) *dto.MidjourneyResponse {
return &dto.MidjourneyResponse{
Code: code,
@@ -86,13 +88,16 @@ func ClaudeErrorWrapperLocal(err error, code string, statusCode int) *dto.Claude
func RelayErrorHandler(ctx context.Context, resp *http.Response, showBodyWhenFail bool) (maxApiErr *types.MaxAPIError) {
maxApiErr = types.InitOpenAIError(types.ErrorCodeBadResponseStatusCode, resp.StatusCode)
- responseBody, err := io.ReadAll(resp.Body)
+ responseBody, truncated, err := readUpstreamErrorBody(resp.Body)
if err != nil {
return
}
CloseResponseBodyGracefully(resp)
var errResponse dto.GeneralErrorResponse
responseBodyText := string(responseBody)
+ if truncated {
+ responseBodyText += fmt.Sprintf("...[truncated after %d bytes]", maxUpstreamErrorBodyBytes)
+ }
responseBodyPreview := common.LocalLogPreview(responseBodyText)
buildErrWithBody := func(message string) error {
if message == "" {
@@ -130,6 +135,17 @@ func RelayErrorHandler(ctx context.Context, resp *http.Response, showBodyWhenFai
return
}
+func readUpstreamErrorBody(body io.Reader) ([]byte, bool, error) {
+ responseBody, err := io.ReadAll(io.LimitReader(body, maxUpstreamErrorBodyBytes+1))
+ if err != nil {
+ return nil, false, err
+ }
+ if len(responseBody) <= maxUpstreamErrorBodyBytes {
+ return responseBody, false, nil
+ }
+ return responseBody[:maxUpstreamErrorBodyBytes], true, nil
+}
+
func ResetStatusCode(maxApiErr *types.MaxAPIError, statusCodeMappingStr string) {
if maxApiErr == nil {
return
diff --git a/service/error_test.go b/service/error_test.go
index cd8d30a9..8eb50256 100644
--- a/service/error_test.go
+++ b/service/error_test.go
@@ -150,6 +150,36 @@ func TestRelayErrorHandlerKeepsInvalidJSONBodyInDebugLog(t *testing.T) {
require.Contains(t, logBuffer.String(), body)
}
+func TestRelayErrorHandlerLimitsUpstreamErrorBodyRead(t *testing.T) {
+ body := strings.Repeat("x", maxUpstreamErrorBodyBytes+4096)
+ reader := &countingReadCloser{Reader: strings.NewReader(body)}
+ resp := &http.Response{
+ StatusCode: http.StatusInternalServerError,
+ Body: reader,
+ }
+
+ maxAPIError := RelayErrorHandler(context.Background(), resp, true)
+
+ require.NotNil(t, maxAPIError)
+ require.LessOrEqual(t, reader.read, maxUpstreamErrorBodyBytes+1)
+ require.Contains(t, maxAPIError.Error(), "truncated after")
+}
+
+type countingReadCloser struct {
+ io.Reader
+ read int
+}
+
+func (r *countingReadCloser) Read(p []byte) (int, error) {
+ n, err := r.Reader.Read(p)
+ r.read += n
+ return n, err
+}
+
+func (r *countingReadCloser) Close() error {
+ return nil
+}
+
func withDebugEnabled(t *testing.T, enabled bool) {
t.Helper()
diff --git a/tools/jsonwrapcheck/allowlist.txt b/tools/jsonwrapcheck/allowlist.txt
index 80c27c76..db17322c 100644
--- a/tools/jsonwrapcheck/allowlist.txt
+++ b/tools/jsonwrapcheck/allowlist.txt
@@ -16,12 +16,6 @@ controller/channel-billing.go|updateChannelOpenRouterBalance|Unmarshal|d7a4bbf58
controller/channel-billing.go|updateChannelMoonshotBalance|Unmarshal|d7a4bbf58b3151d53e68905b6060ad920944bbaa0d6c6c1c0fe88be32e034d49
controller/channel-billing.go|updateChannelBalance|Unmarshal|46a9579cb9c80e020d91043bd3cf0c05da89a6fdbd8b65466c3795a3e637e626
controller/channel-billing.go|updateChannelBalance|Unmarshal|3df63364d5fe5f35cc933cc2cdcf8b28c4aa5f2f0744c21f9364876785916be5
-controller/channel.go|getVertexArrayKeys|Marshal|be383ea6cfaaf292483613ff481847bcf18cbddf173a0870a25e345b29d9f5eb
-controller/channel.go|UpdateChannel|Unmarshal|95452bf62f53fada9e40a28930de093965ae4cb90273009d522eb18692bed6d3
-controller/channel.go|FetchModels|NewDecoder|b688a58000927e74f99ad74076b60c96c53ce0095d52d9c4e4525d3acb7e31fe
-controller/channel.go|OllamaPullModelStream|Marshal|86fa903a3f00085960a941ca35df761b7ae59730e05eba0be45f6fc95d48b0cc
-controller/channel.go|OllamaPullModelStream|Marshal|ec10aaa0d8ed9c78ebee61fdb21daf14a4bbf7da2a7c7dd2c3e7cfb8c96a189f
-controller/channel.go|OllamaPullModelStream|Marshal|0ad5ba168fbe247c5fa602760d38e7ec74661e0e74cbf5bab856de7a046fad3c
controller/console_migrate.go|MigrateConsoleSetting|Unmarshal|bcdd45a79ffdf482179db4212ca56bc912c76046885b91cc619342d16f70869c
controller/console_migrate.go|MigrateConsoleSetting|Marshal|af391e668907209f54f14f3f402a03be0b7caf7f206f11ef79164ff8d85db95f
controller/console_migrate.go|MigrateConsoleSetting|Unmarshal|bcdd45a79ffdf482179db4212ca56bc912c76046885b91cc619342d16f70869c
diff --git a/web/default/index.html b/web/default/index.html
index 9979e766..51da6ee8 100644
--- a/web/default/index.html
+++ b/web/default/index.html
@@ -4,6 +4,7 @@
+
MAX API
@@ -19,6 +20,6 @@
-
+