From b63e5440c6688e9a6cf4ef424d085a04f2d056bd Mon Sep 17 00:00:00 2001 From: martin Date: Sun, 7 Jun 2026 08:50:02 +0800 Subject: [PATCH] =?UTF-8?q?fix(open):=E4=BF=AE=E5=A4=8D=E5=BC=80=E6=94=BE?= =?UTF-8?q?=E5=B9=B3=E5=8F=B0=E6=9D=83=E9=99=90=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/open/open.go | 11 + api/open/v1/open_api.go | 211 +++++- api/tenant/tenant.go | 1 + api/tenant/v1/open_platform.go | 4 +- api/tenant/v1/wallet_tenant.go | 8 + docs/reference/goframe-conventions.md | 21 + internal/cmd/cmd.go | 7 + internal/controller/open/open_v1_open_api.go | 33 + .../tenant/tenant_v1_wallet_tenant.go | 5 +- internal/logic/billing/alert.go | 25 +- internal/logic/open/open_api.go | 694 ++++++++++++++++++ internal/logic/relay/apikey.go | 70 +- internal/logic/tenant/billing.go | 23 + internal/logic/tenant/open_platform.go | 95 +-- internal/logic/tenant/webhook_dispatcher.go | 98 ++- internal/logic/tenant/webhook_worker.go | 17 +- internal/middleware/open_auth.go | 100 ++- internal/service/open.go | 11 + internal/service/tenant.go | 2 + .../000007_webhook_indexes_and_cleanup.sql | 31 + web/tenant/src/views/finance/WalletPage.vue | 83 +++ web/tenant/src/views/open/OpenAppsPage.vue | 1 + web/tenant/src/views/open/WebhooksPage.vue | 180 ++++- 23 files changed, 1622 insertions(+), 109 deletions(-) create mode 100644 migrations/000007_webhook_indexes_and_cleanup.sql diff --git a/api/open/open.go b/api/open/open.go index 674e80d..989f2bb 100644 --- a/api/open/open.go +++ b/api/open/open.go @@ -24,4 +24,15 @@ type IOpenV1 interface { OpenKeyDelete(ctx context.Context, req *v1.OpenKeyDeleteReq) (res *v1.OpenKeyDeleteRes, err error) OpenUsageQuery(ctx context.Context, req *v1.OpenUsageQueryReq) (res *v1.OpenUsageQueryRes, err error) OpenBillingQuery(ctx context.Context, req *v1.OpenBillingQueryReq) (res *v1.OpenBillingQueryRes, err error) + OpenProjectList(ctx context.Context, req *v1.OpenProjectListReq) (res *v1.OpenProjectListRes, err error) + OpenProjectCreate(ctx context.Context, req *v1.OpenProjectCreateReq) (res *v1.OpenProjectCreateRes, err error) + OpenProjectGet(ctx context.Context, req *v1.OpenProjectGetReq) (res *v1.OpenProjectGetRes, err error) + OpenProjectUpdate(ctx context.Context, req *v1.OpenProjectUpdateReq) (res *v1.OpenProjectUpdateRes, err error) + OpenProjectArchive(ctx context.Context, req *v1.OpenProjectArchiveReq) (res *v1.OpenProjectArchiveRes, err error) + OpenProjectUnarchive(ctx context.Context, req *v1.OpenProjectUnarchiveReq) (res *v1.OpenProjectUnarchiveRes, err error) + OpenProjectKeyList(ctx context.Context, req *v1.OpenProjectKeyListReq) (res *v1.OpenProjectKeyListRes, err error) + OpenProjectKeyCreate(ctx context.Context, req *v1.OpenProjectKeyCreateReq) (res *v1.OpenProjectKeyCreateRes, err error) + OpenProjectKeyDelete(ctx context.Context, req *v1.OpenProjectKeyDeleteReq) (res *v1.OpenProjectKeyDeleteRes, err error) + OpenProjectUsageStats(ctx context.Context, req *v1.OpenProjectUsageStatsReq) (res *v1.OpenProjectUsageStatsRes, err error) + OpenProjectUsageLogs(ctx context.Context, req *v1.OpenProjectUsageLogsReq) (res *v1.OpenProjectUsageLogsRes, err error) } diff --git a/api/open/v1/open_api.go b/api/open/v1/open_api.go index 70f7b86..247c02f 100644 --- a/api/open/v1/open_api.go +++ b/api/open/v1/open_api.go @@ -1,6 +1,9 @@ package v1 -import "github.com/gogf/gf/v2/frame/g" +import ( + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/os/gtime" +) // ============================================================ // 开放平台业务 API（第三方应用调用） @@ -216,3 +219,209 @@ type OpenBillingItem struct { Description string `json:"description"` CreatedAt string `json:"created_at"` } + +// ============================================================ +// 项目管理 +// ============================================================ + +// OpenProjectList 项目列表 +type OpenProjectListReq struct { + g.Meta `path:"/v1/projects" method:"get" tags:"开放平台API" summary:"项目列表"` + Page int `json:"page" in:"query" d:"1" dc:"页码"` + PageSize int `json:"page_size" in:"query" d:"20" dc:"每页数量"` + Status string `json:"status" in:"query" dc:"按状态筛选：active/archived/budget_exhausted"` + Keyword string `json:"keyword" in:"query" dc:"搜索关键词"` +} + +type OpenProjectListRes struct { + List []OpenProjectItem `json:"list"` + Total int `json:"total"` + Page int `json:"page"` + PageSize int `json:"page_size"` +} + +type OpenProjectItem struct { + ID int64 `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + Status string `json:"status"` + Budget string `json:"budget"` // USD 6位小数，"unlimited" 表示不限 + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} + +// OpenProjectCreate 创建项目 +type OpenProjectCreateReq struct { + g.Meta `path:"/v1/projects" method:"post" mime:"json" tags:"开放平台API" summary:"创建项目"` + Name string `json:"name" v:"required|length:1,100#请输入项目名称|项目名称长度1-100" dc:"项目名称"` + Description string `json:"description" dc:"项目描述"` + Budget float64 `json:"budget" dc:"项目预算上限(USD，0=不限)"` +} + +type OpenProjectCreateRes struct { + ID int64 `json:"id"` +} + +// OpenProjectGet 项目详情 +type OpenProjectGetReq struct { + g.Meta `path:"/v1/projects/{id}" method:"get" tags:"开放平台API" summary:"项目详情"` + Id int64 `json:"id" in:"path" v:"required#请指定项目ID" dc:"项目ID"` +} + +type OpenProjectGetRes struct { + ID int64 `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + Status string `json:"status"` + Budget string `json:"budget"` + ActiveKeys int `json:"active_keys"` + TotalKeys int `json:"total_keys"` + MonthCost string `json:"month_cost"` + MonthRequests int64 `json:"month_requests"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} + +// OpenProjectUpdate 更新项目 +type OpenProjectUpdateReq struct { + g.Meta `path:"/v1/projects/{id}" method:"put" mime:"json" tags:"开放平台API" summary:"更新项目"` + Id int64 `json:"id" in:"path" v:"required#请指定项目ID" dc:"项目ID"` + Name *string `json:"name" dc:"项目名称"` + Description *string `json:"description" dc:"项目描述"` + Budget *float64 `json:"budget" dc:"项目预算上限(USD，0=不限)"` +} + +type OpenProjectUpdateRes struct{} + +// OpenProjectArchive 归档项目 +type OpenProjectArchiveReq struct { + g.Meta `path:"/v1/projects/{id}/archive" method:"post" mime:"json" tags:"开放平台API" summary:"归档项目"` + Id int64 `json:"id" in:"path" v:"required#请指定项目ID" dc:"项目ID"` +} + +type OpenProjectArchiveRes struct{} + +// OpenProjectUnarchive 取消归档 +type OpenProjectUnarchiveReq struct { + g.Meta `path:"/v1/projects/{id}/unarchive" method:"post" mime:"json" tags:"开放平台API" summary:"取消归档"` + Id int64 `json:"id" in:"path" v:"required#请指定项目ID" dc:"项目ID"` +} + +type OpenProjectUnarchiveRes struct{} + +// ============================================================ +// 项目 API Key 管理 +// ============================================================ + +// OpenProjectKeyList 项目密钥列表 +type OpenProjectKeyListReq struct { + g.Meta `path:"/v1/projects/{id}/api-keys" method:"get" tags:"开放平台API" summary:"项目密钥列表"` + Id int64 `json:"id" in:"path" v:"required#请指定项目ID" dc:"项目ID"` + Page int `json:"page" in:"query" d:"1" dc:"页码"` + PageSize int `json:"page_size" in:"query" d:"20" dc:"每页数量"` +} + +type OpenProjectKeyListRes struct { + List []OpenProjectKeyItem `json:"list"` + Total int `json:"total"` + Page int `json:"page"` + PageSize int `json:"page_size"` +} + +type OpenProjectKeyItem struct { + ID int64 `json:"id"` + Name string `json:"name"` + KeyPrefix string `json:"key_prefix"` + Status string `json:"status"` + CreatedAt string `json:"created_at"` + ExpiresAt string `json:"expires_at,omitempty"` +} + +// OpenProjectKeyCreate 创建项目密钥 +type OpenProjectKeyCreateReq struct { + g.Meta `path:"/v1/projects/{id}/api-keys" method:"post" mime:"json" tags:"开放平台API" summary:"创建项目密钥"` + Id int64 `json:"id" in:"path" v:"required#请指定项目ID" dc:"项目ID"` + Name string `json:"name" v:"required|length:1,100#请输入Key名称|名称长度1-100" dc:"Key名称"` + Scope string `json:"scope" d:"full" v:"in:full,chat_only,embeddings_only,images_only,read_only#权限范围无效" dc:"权限范围"` + ExpiresAt *gtime.Time `json:"expires_at" dc:"过期时间"` + ModelNames []string `json:"model_names" dc:"可用模型列表"` +} + +type OpenProjectKeyCreateRes struct { + ID int64 `json:"id"` + Name string `json:"name"` + Key string `json:"key"` + KeyPrefix string `json:"key_prefix"` +} + +// OpenProjectKeyDelete 删除项目密钥 +type OpenProjectKeyDeleteReq struct { + g.Meta `path:"/v1/projects/{id}/api-keys/{keyId}" method:"delete" tags:"开放平台API" summary:"删除项目密钥"` + Id int64 `json:"id" in:"path" v:"required#请指定项目ID" dc:"项目ID"` + KeyId int64 `json:"keyId" in:"path" v:"required#请指定密钥ID" dc:"密钥ID"` +} + +type OpenProjectKeyDeleteRes struct{} + +// ============================================================ +// 项目用量查询 +// ============================================================ + +// OpenProjectUsageStats 项目用量统计 +type OpenProjectUsageStatsReq struct { + g.Meta `path:"/v1/projects/{id}/usage-stats" method:"get" tags:"开放平台API" summary:"项目用量统计"` + Id int64 `json:"id" in:"path" v:"required#请指定项目ID" dc:"项目ID"` + StartDate string `json:"start_date" in:"query" dc:"开始日期 (YYYY-MM-DD)，默认近30天"` + EndDate string `json:"end_date" in:"query" dc:"结束日期 (YYYY-MM-DD)"` +} + +type OpenProjectUsageStatsRes struct { + TotalCost string `json:"total_cost"` + TotalRequests int64 `json:"total_requests"` + TotalInputTokens int64 `json:"total_input_tokens"` + TotalOutputTokens int64 `json:"total_output_tokens"` + Daily []OpenProjectDailyStat `json:"daily"` + Models []OpenProjectModelStat `json:"models"` +} + +type OpenProjectDailyStat struct { + Date string `json:"date"` + RequestCount int64 `json:"request_count"` + TotalCost string `json:"total_cost"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` +} + +type OpenProjectModelStat struct { + ModelName string `json:"model_name"` + RequestCount int64 `json:"request_count"` + TotalCost string `json:"total_cost"` +} + +// OpenProjectUsageLogs 项目用量日志 +type OpenProjectUsageLogsReq struct { + g.Meta `path:"/v1/projects/{id}/usage-logs" method:"get" tags:"开放平台API" summary:"项目用量日志"` + Id int64 `json:"id" in:"path" v:"required#请指定项目ID" dc:"项目ID"` + Page int `json:"page" in:"query" d:"1" dc:"页码"` + PageSize int `json:"page_size" in:"query" d:"20" dc:"每页数量"` +} + +type OpenProjectUsageLogsRes struct { + List []OpenProjectUsageLogItem `json:"list"` + Total int `json:"total"` + Page int `json:"page"` + PageSize int `json:"page_size"` +} + +type OpenProjectUsageLogItem struct { + ID int64 `json:"id"` + ModelName string `json:"model_name"` + RelayMode string `json:"relay_mode"` + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + TotalCost string `json:"total_cost"` + LatencyMs int `json:"latency_ms"` + Status string `json:"status"` + ErrorMessage string `json:"error_message,omitempty"` + CreatedAt string `json:"created_at"` +} diff --git a/api/tenant/tenant.go b/api/tenant/tenant.go index 3bf25f3..5112c09 100644 --- a/api/tenant/tenant.go +++ b/api/tenant/tenant.go @@ -164,4 +164,5 @@ type ITenantV1 interface { TenantWalletTransactions(ctx context.Context, req *v1.TenantWalletTransactionsReq) (res *v1.TenantWalletTransactionsRes, err error) TenantWalletTransactionsExport(ctx context.Context, req *v1.TenantWalletTransactionsExportReq) (res *v1.TenantWalletTransactionsExportRes, err error) TenantWalletFrozenItems(ctx context.Context, req *v1.TenantWalletFrozenItemsReq) (res *v1.TenantWalletFrozenItemsRes, err error) + TenantWalletUpdateWarningThreshold(ctx context.Context, req *v1.TenantWalletUpdateWarningThresholdReq) (res *v1.TenantWalletUpdateWarningThresholdRes, err error) } diff --git a/api/tenant/v1/open_platform.go b/api/tenant/v1/open_platform.go index 88e606f..2462120 100644 --- a/api/tenant/v1/open_platform.go +++ b/api/tenant/v1/open_platform.go @@ -39,7 +39,7 @@ type OpenAppCreateReq struct { g.Meta `path:"/open/apps" method:"post" mime:"json" tags:"租户控制台-开放平台" summary:"创建应用"` Name string `json:"name" v:"required|length:2,100#请输入应用名称|应用名称长度为2-100位" dc:"应用名称"` Description string `json:"description" v:"length:0,500#描述不超过500字" dc:"应用描述"` - Permissions []string `json:"permissions" v:"required|length:1,20#请选择权限范围|权限数量不正确" dc:"权限范围"` + Permissions []string `json:"permissions" v:"required#请选择权限范围" dc:"权限范围"` IPWhitelist []string `json:"ip_whitelist" dc:"IP 白名单"` RateLimit int `json:"rate_limit" d:"60" dc:"每分钟请求上限"` } @@ -120,7 +120,7 @@ type WebhookConfigCreateReq struct { g.Meta `path:"/open/webhooks" method:"post" mime:"json" tags:"租户控制台-开放平台" summary:"创建Webhook配置"` Name string `json:"name" v:"required|length:2,100#请输入名称|名称长度2-100" dc:"配置名称"` URL string `json:"url" v:"required|length:10,500#请输入URL|URL长度不正确" dc:"回调URL（HTTPS）"` - Events []string `json:"events" v:"required|length:1,50#请选择事件|事件数量不正确" dc:"订阅事件列表"` + Events []string `json:"events" v:"required#请选择事件" dc:"订阅事件列表"` MaxConsecutiveFailures int `json:"max_consecutive_failures" d:"10" dc:"最大连续失败次数"` } diff --git a/api/tenant/v1/wallet_tenant.go b/api/tenant/v1/wallet_tenant.go index afd3cdc..5ee5058 100644 --- a/api/tenant/v1/wallet_tenant.go +++ b/api/tenant/v1/wallet_tenant.go @@ -74,3 +74,11 @@ type FrozenItemDetail struct { CreatedAt int64 `json:"created_at"` Remaining int64 `json:"remaining"` } + +// TenantWalletUpdateWarningThresholdReq 修改预警阈值 +type TenantWalletUpdateWarningThresholdReq struct { + g.Meta `path:"/wallet/warning-threshold" method:"put" mime:"json" tags:"租户控制台-钱包" summary:"修改预警阈值"` + Threshold float64 `json:"threshold" v:"min:0#阈值不能为负数" dc:"预警阈值（USD），0 表示关闭预警"` +} + +type TenantWalletUpdateWarningThresholdRes struct{} diff --git a/docs/reference/goframe-conventions.md b/docs/reference/goframe-conventions.md index cb9dc33..c7cd18e 100644 --- a/docs/reference/goframe-conventions.md +++ b/docs/reference/goframe-conventions.md @@ -464,3 +464,24 @@ dao.SysIdempotencyRecords.Ctx(ctx).Data(do.SysIdempotencyRecords{ **正确做法**：将原生 SQL 改写为 DO 插入时，逐列核对目标表中**所有 NOT NULL 且无 DEFAULT 的列**是否都在 DO 中赋了值——这类列不能依赖框架的 OmitNil 自动跳过，否则 INSERT 必然失败。框架自动维护的 `created_at`/`updated_at` 不在此列。 +### 2026-06-07：`length` 校验规则对 `[]string` 校验的是 JSON 字符串长度，不是元素个数 + +**问题描述**：Webhook 创建接口的 Events 字段使用 `v:"required|length:1,50#请选择事件|事件数量不正确"`，用户选了 3-4 个事件后提交报错"事件数量不正确"。 + +**原因**：GoFrame v2 的 `length:min,max` 校验规则对 `[]string` 类型不是检查 slice 元素个数，而是先将整个 slice 通过 `json.Marshal` 转成 JSON 字符串，再检查该字符串的字符长度。例如 `["member.created","key.deleted"]` 的 JSON 字符串长度约 33 个字符。用户选了 3-4 个事件后 JSON 字符串就可能超过 50 字符，导致校验失败。 + +**修复方式**：去掉 `length:1,50`，只保留 `required`（事件从固定列表中选择，无需额外长度校验）。 + +**正确做法**： +```go +// 错误 — length 对 []string 校验的是 JSON 字符串长度，不是元素个数 +Events []string `json:"events" v:"required|length:1,50#请选择事件|事件数量不正确"` + +// 正确 — 只用 required 保证至少选了一个事件 +Events []string `json:"events" v:"required#请选择事件"` + +// 如果需要校验每个元素字符串长度，用 foreach|length +Events []string `json:"events" v:"required|foreach|length:1,100#请选择事件|事件名长度不正确"` +``` +**注意**：GoFrame v2 没有内置规则直接校验 slice 的元素个数。如需限制元素数量，需自定义校验规则或在 logic 层手动检查。 + diff --git a/internal/cmd/cmd.go b/internal/cmd/cmd.go index dc30482..2fa572c 100644 --- a/internal/cmd/cmd.go +++ b/internal/cmd/cmd.go @@ -22,6 +22,7 @@ import ( "github.com/qianfree/team-api/internal/logic/billing" "github.com/qianfree/team-api/internal/logic/common" "github.com/qianfree/team-api/internal/logic/monitor" + relayLogic "github.com/qianfree/team-api/internal/logic/relay" "github.com/qianfree/team-api/internal/logic/task" "github.com/qianfree/team-api/internal/logic/tenant" "github.com/qianfree/team-api/internal/middleware" @@ -285,6 +286,9 @@ var ( // Embedded frontend SPA serving registerFrontendRoutes(s) + // Migrate encryption key (legacy default → configured key) + relayLogic.MigrateEncryptionKey(ctx) + // Initialize active task count and start polling task.InitActiveCount(ctx) go task.StartAsyncPolling(ctx) @@ -292,6 +296,9 @@ var ( // Start webhook dispatcher (event-driven delivery) tenant.InitWebhookDispatcher(ctx) + // 注入 webhook 发布函数到 billing 包（解耦循环依赖） + billing.SetPublishWebhookEventFn(tenant.PublishWebhookEvent) + // Flush usage log writer on shutdown (s.Run blocks until server stops) defer plugin.Shutdown(ctx) defer common.CloseChannelErrorWriter() diff --git a/internal/controller/open/open_v1_open_api.go b/internal/controller/open/open_v1_open_api.go index b921304..31a49eb 100644 --- a/internal/controller/open/open_v1_open_api.go +++ b/internal/controller/open/open_v1_open_api.go @@ -46,3 +46,36 @@ func (c *ControllerV1) OpenUsageQuery(ctx context.Context, req *v1.OpenUsageQuer func (c *ControllerV1) OpenBillingQuery(ctx context.Context, req *v1.OpenBillingQueryReq) (res *v1.OpenBillingQueryRes, err error) { return service.Open().OpenBillingQuery(ctx, req) } +func (c *ControllerV1) OpenProjectList(ctx context.Context, req *v1.OpenProjectListReq) (res *v1.OpenProjectListRes, err error) { + return service.Open().OpenProjectList(ctx, req) +} +func (c *ControllerV1) OpenProjectCreate(ctx context.Context, req *v1.OpenProjectCreateReq) (res *v1.OpenProjectCreateRes, err error) { + return service.Open().OpenProjectCreate(ctx, req) +} +func (c *ControllerV1) OpenProjectGet(ctx context.Context, req *v1.OpenProjectGetReq) (res *v1.OpenProjectGetRes, err error) { + return service.Open().OpenProjectGet(ctx, req) +} +func (c *ControllerV1) OpenProjectUpdate(ctx context.Context, req *v1.OpenProjectUpdateReq) (res *v1.OpenProjectUpdateRes, err error) { + return service.Open().OpenProjectUpdate(ctx, req) +} +func (c *ControllerV1) OpenProjectArchive(ctx context.Context, req *v1.OpenProjectArchiveReq) (res *v1.OpenProjectArchiveRes, err error) { + return service.Open().OpenProjectArchive(ctx, req) +} +func (c *ControllerV1) OpenProjectUnarchive(ctx context.Context, req *v1.OpenProjectUnarchiveReq) (res *v1.OpenProjectUnarchiveRes, err error) { + return service.Open().OpenProjectUnarchive(ctx, req) +} +func (c *ControllerV1) OpenProjectKeyList(ctx context.Context, req *v1.OpenProjectKeyListReq) (res *v1.OpenProjectKeyListRes, err error) { + return service.Open().OpenProjectKeyList(ctx, req) +} +func (c *ControllerV1) OpenProjectKeyCreate(ctx context.Context, req *v1.OpenProjectKeyCreateReq) (res *v1.OpenProjectKeyCreateRes, err error) { + return service.Open().OpenProjectKeyCreate(ctx, req) +} +func (c *ControllerV1) OpenProjectKeyDelete(ctx context.Context, req *v1.OpenProjectKeyDeleteReq) (res *v1.OpenProjectKeyDeleteRes, err error) { + return service.Open().OpenProjectKeyDelete(ctx, req) +} +func (c *ControllerV1) OpenProjectUsageStats(ctx context.Context, req *v1.OpenProjectUsageStatsReq) (res *v1.OpenProjectUsageStatsRes, err error) { + return service.Open().OpenProjectUsageStats(ctx, req) +} +func (c *ControllerV1) OpenProjectUsageLogs(ctx context.Context, req *v1.OpenProjectUsageLogsReq) (res *v1.OpenProjectUsageLogsRes, err error) { + return service.Open().OpenProjectUsageLogs(ctx, req) +} diff --git a/internal/controller/tenant/tenant_v1_wallet_tenant.go b/internal/controller/tenant/tenant_v1_wallet_tenant.go index 531dbe9..a1f57ae 100644 --- a/internal/controller/tenant/tenant_v1_wallet_tenant.go +++ b/internal/controller/tenant/tenant_v1_wallet_tenant.go @@ -3,7 +3,7 @@ package tenant import ( "context" - "github.com/qianfree/team-api/api/tenant/v1" + v1 "github.com/qianfree/team-api/api/tenant/v1" "github.com/qianfree/team-api/internal/service" ) @@ -19,3 +19,6 @@ func (c *ControllerV1) TenantWalletTransactionsExport(ctx context.Context, req * func (c *ControllerV1) TenantWalletFrozenItems(ctx context.Context, req *v1.TenantWalletFrozenItemsReq) (res *v1.TenantWalletFrozenItemsRes, err error) { return service.Tenant().WalletFrozenItems(ctx, req) } +func (c *ControllerV1) TenantWalletUpdateWarningThreshold(ctx context.Context, req *v1.TenantWalletUpdateWarningThresholdReq) (res *v1.TenantWalletUpdateWarningThresholdRes, err error) { + return service.Tenant().UpdateWarningThreshold(ctx, req) +} diff --git a/internal/logic/billing/alert.go b/internal/logic/billing/alert.go index bc05bd3..6b4363e 100644 --- a/internal/logic/billing/alert.go +++ b/internal/logic/billing/alert.go @@ -10,8 +10,16 @@ import ( "github.com/qianfree/team-api/internal/logic/common" ) +// publishWebhookEventFn 由 tenant 包在初始化时注入，避免 billing → tenant 循环依赖 +var publishWebhookEventFn func(ctx context.Context, tenantID int64, eventType string, payload any) error + +// SetPublishWebhookEventFn 注入 webhook 事件发布函数（由 cmd 初始化时调用） +func SetPublishWebhookEventFn(fn func(ctx context.Context, tenantID int64, eventType string, payload any) error) { + publishWebhookEventFn = fn +} + // CheckBalanceWarning 检查余额预警 -// 如果租户可用余额低于预警线，发送通知给租户管理员 +// 如果租户可用余额低于预警线，发送通知给租户管理员并投递 webhook 事件 func CheckBalanceWarning(ctx context.Context, tenantID int64) (bool, float64, float64, error) { wallet, err := GetWallet(ctx, tenantID) if err != nil { @@ -23,9 +31,11 @@ func CheckBalanceWarning(ctx context.Context, tenantID int64) (bool, float64, fl g.Log().Warningf(ctx, "[BALANCE WARNING] tenant=%d available=%.6f threshold=%.6f", tenantID, available, wallet.WarningThreshold) - // 发送余额预警通知（异步，失败不影响主流程） + // 发送站内通知 + webhook 事件（异步，失败不影响主流程） go func() { bgCtx := context.Background() + + // 站内通知 engine := common.NewNotificationEngine() variables := map[string]any{ "available": fmt.Sprintf("%.6f", available), @@ -34,6 +44,17 @@ func CheckBalanceWarning(ctx context.Context, tenantID int64) (bool, float64, fl if notifyErr := engine.SendBroadcast(bgCtx, tenantID, "balance_warning", variables, "owner,admin"); notifyErr != nil { g.Log().Errorf(bgCtx, "[BALANCE WARNING] send notification failed: tenant=%d err=%v", tenantID, notifyErr) } + + // Webhook 事件投递 + if publishWebhookEventFn != nil { + if webhookErr := publishWebhookEventFn(bgCtx, tenantID, "wallet.low_balance", map[string]any{ + "available_balance": available, + "warning_threshold": wallet.WarningThreshold, + "wallet_id": wallet.ID, + }); webhookErr != nil { + g.Log().Errorf(bgCtx, "[BALANCE WARNING] publish webhook event failed: tenant=%d err=%v", tenantID, webhookErr) + } + } }() return true, available, wallet.WarningThreshold, nil diff --git a/internal/logic/open/open_api.go b/internal/logic/open/open_api.go index cd851e0..10444eb 100644 --- a/internal/logic/open/open_api.go +++ b/internal/logic/open/open_api.go @@ -44,6 +44,9 @@ func (s *sOpen) OpenMemberList(ctx context.Context, req *v1.OpenMemberListReq) ( if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "members:read"); err != nil { + return nil, err + } page := req.Page pageSize := req.PageSize @@ -104,6 +107,9 @@ func (s *sOpen) OpenMemberCreate(ctx context.Context, req *v1.OpenMemberCreateRe if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "members:write"); err != nil { + return nil, err + } count, _ := dao.TntUsers.Ctx(ctx).Where("tenant_id", tenantID).Where("email", req.Email).Count() if count > 0 { @@ -137,6 +143,9 @@ func (s *sOpen) OpenMemberUpdate(ctx context.Context, req *v1.OpenMemberUpdateRe if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "members:write"); err != nil { + return nil, err + } data := do.TntUsers{} hasUpdate := false @@ -166,6 +175,9 @@ func (s *sOpen) OpenMemberDelete(ctx context.Context, req *v1.OpenMemberDeleteRe if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "members:write"); err != nil { + return nil, err + } var user *struct { Role string `json:"role"` @@ -194,6 +206,9 @@ func (s *sOpen) OpenMemberQuota(ctx context.Context, req *v1.OpenMemberQuotaReq) if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "members:read"); err != nil { + return nil, err + } var user *struct { QuotaType string `json:"quota_type"` @@ -226,6 +241,9 @@ func (s *sOpen) OpenMemberQuotaUpdate(ctx context.Context, req *v1.OpenMemberQuo if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "members:write"); err != nil { + return nil, err + } if req.QuotaType == "periodic" && req.Period == "" { return nil, errOpen(consts.CodeBadRequest, "periodic 类型额度必须指定周期") @@ -258,6 +276,9 @@ func (s *sOpen) OpenMemberModels(ctx context.Context, req *v1.OpenMemberModelsRe if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "members:read"); err != nil { + return nil, err + } var scopes []struct { ModelID string `json:"model_id"` @@ -283,6 +304,9 @@ func (s *sOpen) OpenMemberModelsUpdate(ctx context.Context, req *v1.OpenMemberMo if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "members:write"); err != nil { + return nil, err + } err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { _, err := tx.Model("tnt_member_model_scopes").Ctx(ctx). @@ -338,6 +362,9 @@ func (s *sOpen) OpenKeyList(ctx context.Context, req *v1.OpenKeyListReq) (*v1.Op if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "keys:read"); err != nil { + return nil, err + } page := req.Page pageSize := req.PageSize @@ -391,6 +418,9 @@ func (s *sOpen) OpenKeyCreate(ctx context.Context, req *v1.OpenKeyCreateReq) (*v if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "keys:write"); err != nil { + return nil, err + } rawKey, prefix, encryptedKey, err := relay.GenerateApiKey(ctx) if err != nil { @@ -437,6 +467,9 @@ func (s *sOpen) OpenKeyDelete(ctx context.Context, req *v1.OpenKeyDeleteReq) (*v if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "keys:write"); err != nil { + return nil, err + } _, err := dao.ApiKeys.Ctx(ctx).Where("id", req.Id).Where("tenant_id", tenantID).Delete() return nil, err @@ -457,6 +490,9 @@ func (s *sOpen) OpenUsageQuery(ctx context.Context, req *v1.OpenUsageQueryReq) ( if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "usage:read"); err != nil { + return nil, err + } page := req.Page pageSize := req.PageSize @@ -541,6 +577,9 @@ func (s *sOpen) OpenBillingQuery(ctx context.Context, req *v1.OpenBillingQueryRe if tenantID == 0 { return nil, errOpen(consts.CodeUnauthorized, "未认证") } + if err := middleware.CheckOpenPermission(ctx, "billing:read"); err != nil { + return nil, err + } page := req.Page pageSize := req.PageSize @@ -599,3 +638,658 @@ func (s *sOpen) OpenBillingQuery(ctx context.Context, req *v1.OpenBillingQueryRe func errOpen(code int, message string) error { return gerror.NewCode(gcode.New(code, message, nil), message) } + +// ============================================================ +// 项目管理 +// ============================================================ + +func (s *sOpen) OpenProjectList(ctx context.Context, req *v1.OpenProjectListReq) (*v1.OpenProjectListRes, error) { + tenantID := middleware.GetOpenTenantID(ctx) + if tenantID == 0 { + return nil, errOpen(consts.CodeUnauthorized, "未认证") + } + if err := middleware.CheckOpenPermission(ctx, "projects:read"); err != nil { + return nil, err + } + + page := req.Page + pageSize := req.PageSize + if page < 1 { + page = 1 + } + if pageSize < 1 || pageSize > 100 { + pageSize = 20 + } + + m := dao.TntProjects.Ctx(ctx).Where("tenant_id", tenantID) + if req.Status != "" { + m = m.Where("status", req.Status) + } + if req.Keyword != "" { + kw := "%" + strings.TrimSpace(req.Keyword) + "%" + m = m.Where("name LIKE ?", kw) + } + + total, err := m.Count() + if err != nil { + return nil, err + } + + type projectRow struct { + Id int64 `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + Status string `json:"status"` + Budget *float64 `json:"budget"` + CreatedAt *gtime.Time `json:"created_at"` + UpdatedAt *gtime.Time `json:"updated_at"` + } + + var projects []projectRow + err = m.Fields("id, name, description, status, budget, created_at, updated_at"). + OrderDesc("id").Page(page, pageSize).Scan(&projects) + if err != nil { + return nil, err + } + + items := make([]v1.OpenProjectItem, 0, len(projects)) + for _, p := range projects { + item := v1.OpenProjectItem{ + ID: p.Id, + Name: p.Name, + Description: p.Description, + Status: p.Status, + } + if p.Budget != nil { + item.Budget = fmt.Sprintf("%.6f", *p.Budget) + } else { + item.Budget = "unlimited" + } + if p.CreatedAt != nil { + item.CreatedAt = p.CreatedAt.Format("Y-m-d H:i:s") + } + if p.UpdatedAt != nil { + item.UpdatedAt = p.UpdatedAt.Format("Y-m-d H:i:s") + } + items = append(items, item) + } + + return &v1.OpenProjectListRes{List: items, Total: total, Page: page, PageSize: pageSize}, nil +} + +func (s *sOpen) OpenProjectCreate(ctx context.Context, req *v1.OpenProjectCreateReq) (*v1.OpenProjectCreateRes, error) { + tenantID := middleware.GetOpenTenantID(ctx) + if tenantID == 0 { + return nil, errOpen(consts.CodeUnauthorized, "未认证") + } + if err := middleware.CheckOpenPermission(ctx, "projects:write"); err != nil { + return nil, err + } + + insertData := do.TntProjects{ + TenantId: tenantID, + Name: req.Name, + Description: req.Description, + Status: "active", + } + + if req.Budget > 0 { + insertData.Budget = req.Budget + } + + result, err := dao.TntProjects.Ctx(ctx).Data(insertData).Insert() + if err != nil { + return nil, err + } + + id, _ := result.LastInsertId() + return &v1.OpenProjectCreateRes{ID: id}, nil +} + +func (s *sOpen) OpenProjectGet(ctx context.Context, req *v1.OpenProjectGetReq) (*v1.OpenProjectGetRes, error) { + tenantID := middleware.GetOpenTenantID(ctx) + if tenantID == 0 { + return nil, errOpen(consts.CodeUnauthorized, "未认证") + } + if err := middleware.CheckOpenPermission(ctx, "projects:read"); err != nil { + return nil, err + } + + type projectRow struct { + Id int64 `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + Status string `json:"status"` + Budget *float64 `json:"budget"` + CreatedAt *gtime.Time `json:"created_at"` + UpdatedAt *gtime.Time `json:"updated_at"` + } + + var p *projectRow + err := dao.TntProjects.Ctx(ctx). + Where("id", req.Id). + Where("tenant_id", tenantID). + Fields("id, name, description, status, budget, created_at, updated_at"). + Scan(&p) + if err != nil { + return nil, err + } + if p == nil { + return nil, errOpen(consts.CodeNotFound, "项目不存在") + } + + // Key 统计 + type keyStatsRow struct { + Total int `json:"total"` + Active int `json:"active"` + } + var keyStats keyStatsRow + dao.ApiKeys.Ctx(ctx). + Where("tenant_id", tenantID). + Where("project_id", req.Id). + Fields("COUNT(*) as total, COUNT(*) FILTER (WHERE status = 'active') as active"). + Scan(&keyStats) + + // 月度用量 + type monthUsageRow struct { + TotalCost float64 `json:"total_cost"` + RequestCount int `json:"request_count"` + } + var monthUsage monthUsageRow + dao.BilUsageLogs.Ctx(ctx). + Where("tenant_id", tenantID). + Where("project_id", req.Id). + Where("created_at >= date_trunc('month', NOW())"). + Fields("COALESCE(SUM(actual_cost), 0) as total_cost, COUNT(*) as request_count"). + Scan(&monthUsage) + + res := &v1.OpenProjectGetRes{ + ID: p.Id, + Name: p.Name, + Description: p.Description, + Status: p.Status, + ActiveKeys: keyStats.Active, + TotalKeys: keyStats.Total, + MonthCost: fmt.Sprintf("%.6f", monthUsage.TotalCost), + MonthRequests: int64(monthUsage.RequestCount), + } + if p.Budget != nil { + res.Budget = fmt.Sprintf("%.6f", *p.Budget) + } else { + res.Budget = "unlimited" + } + if p.CreatedAt != nil { + res.CreatedAt = p.CreatedAt.Format("Y-m-d H:i:s") + } + if p.UpdatedAt != nil { + res.UpdatedAt = p.UpdatedAt.Format("Y-m-d H:i:s") + } + + return res, nil +} + +func (s *sOpen) OpenProjectUpdate(ctx context.Context, req *v1.OpenProjectUpdateReq) (*v1.OpenProjectUpdateRes, error) { + tenantID := middleware.GetOpenTenantID(ctx) + if tenantID == 0 { + return nil, errOpen(consts.CodeUnauthorized, "未认证") + } + if err := middleware.CheckOpenPermission(ctx, "projects:write"); err != nil { + return nil, err + } + + project, err := s.getProjectOrError(ctx, req.Id, tenantID) + if err != nil { + return nil, err + } + if project.Status == "archived" { + return nil, errOpen(consts.CodeBadRequest, "归档的项目不能直接编辑，请先取消归档") + } + + data := do.TntProjects{} + hasUpdate := false + if req.Name != nil { + data.Name = *req.Name + hasUpdate = true + } + if req.Description != nil { + data.Description = *req.Description + hasUpdate = true + } + if req.Budget != nil { + if *req.Budget > 0 { + data.Budget = *req.Budget + } + hasUpdate = true + } + // 预算耗尽状态的项目，更新预算后自动恢复为 active + if project.Status == "budget_exhausted" && req.Budget != nil && *req.Budget > 0 { + data.Status = "active" + hasUpdate = true + } + + if !hasUpdate { + return nil, nil + } + + _, err = dao.TntProjects.Ctx(ctx).Where("id", req.Id).Data(data).Update() + return nil, err +} + +func (s *sOpen) OpenProjectArchive(ctx context.Context, req *v1.OpenProjectArchiveReq) (*v1.OpenProjectArchiveRes, error) { + tenantID := middleware.GetOpenTenantID(ctx) + if tenantID == 0 { + return nil, errOpen(consts.CodeUnauthorized, "未认证") + } + if err := middleware.CheckOpenPermission(ctx, "projects:write"); err != nil { + return nil, err + } + + project, err := s.getProjectOrError(ctx, req.Id, tenantID) + if err != nil { + return nil, err + } + if project.Status == "archived" { + return nil, errOpen(consts.CodeBadRequest, "项目已归档") + } + + // 吊销项目下所有 active Key + _, err = dao.ApiKeys.Ctx(ctx). + Where("tenant_id", tenantID). + Where("project_id", req.Id). + Where("status", "active"). + Data(do.ApiKeys{Status: "revoked"}).Update() + if err != nil { + return nil, err + } + + _, err = dao.TntProjects.Ctx(ctx).Where("id", req.Id). + Data(do.TntProjects{Status: "archived"}).Update() + return nil, err +} + +func (s *sOpen) OpenProjectUnarchive(ctx context.Context, req *v1.OpenProjectUnarchiveReq) (*v1.OpenProjectUnarchiveRes, error) { + tenantID := middleware.GetOpenTenantID(ctx) + if tenantID == 0 { + return nil, errOpen(consts.CodeUnauthorized, "未认证") + } + if err := middleware.CheckOpenPermission(ctx, "projects:write"); err != nil { + return nil, err + } + + project, err := s.getProjectOrError(ctx, req.Id, tenantID) + if err != nil { + return nil, err + } + if project.Status != "archived" { + return nil, errOpen(consts.CodeBadRequest, "只有归档状态的项目可以取消归档") + } + + _, err = dao.TntProjects.Ctx(ctx).Where("id", req.Id). + Data(do.TntProjects{Status: "active"}).Update() + return nil, err +} + +// ============================================================ +// 项目 API Key 管理 +// ============================================================ + +func (s *sOpen) OpenProjectKeyList(ctx context.Context, req *v1.OpenProjectKeyListReq) (*v1.OpenProjectKeyListRes, error) { + tenantID := middleware.GetOpenTenantID(ctx) + if tenantID == 0 { + return nil, errOpen(consts.CodeUnauthorized, "未认证") + } + if err := middleware.CheckOpenPermission(ctx, "projects:read"); err != nil { + return nil, err + } + + if _, err := s.getProjectOrError(ctx, req.Id, tenantID); err != nil { + return nil, err + } + + page := req.Page + pageSize := req.PageSize + if page < 1 { + page = 1 + } + if pageSize < 1 || pageSize > 100 { + pageSize = 20 + } + + m := dao.ApiKeys.Ctx(ctx). + Where("tenant_id", tenantID). + Where("project_id", req.Id) + + total, err := m.Count() + if err != nil { + return nil, err + } + + type keyRow struct { + Id int64 `json:"id"` + Name string `json:"name"` + KeyPrefix string `json:"key_prefix"` + Status string `json:"status"` + CreatedAt *gtime.Time `json:"created_at"` + ExpiresAt *gtime.Time `json:"expires_at"` + } + + var keys []keyRow + err = m.Fields("id, name, key_prefix, status, created_at, expires_at"). + OrderDesc("id").Page(page, pageSize).Scan(&keys) + if err != nil { + return nil, err + } + + items := make([]v1.OpenProjectKeyItem, 0, len(keys)) + for _, k := range keys { + item := v1.OpenProjectKeyItem{ + ID: k.Id, + Name: k.Name, + KeyPrefix: k.KeyPrefix, + Status: k.Status, + } + if k.CreatedAt != nil { + item.CreatedAt = k.CreatedAt.Format("Y-m-d H:i:s") + } + if k.ExpiresAt != nil { + item.ExpiresAt = k.ExpiresAt.Format("Y-m-d H:i:s") + } + items = append(items, item) + } + + return &v1.OpenProjectKeyListRes{List: items, Total: total, Page: page, PageSize: pageSize}, nil +} + +func (s *sOpen) OpenProjectKeyCreate(ctx context.Context, req *v1.OpenProjectKeyCreateReq) (*v1.OpenProjectKeyCreateRes, error) { + tenantID := middleware.GetOpenTenantID(ctx) + if tenantID == 0 { + return nil, errOpen(consts.CodeUnauthorized, "未认证") + } + if err := middleware.CheckOpenPermission(ctx, "projects:write"); err != nil { + return nil, err + } + + project, err := s.getProjectOrError(ctx, req.Id, tenantID) + if err != nil { + return nil, err + } + if project.Status != "active" { + return nil, errOpen(consts.CodeProjectNotActive, "项目状态不可用") + } + + rawKey, prefix, encryptedKey, err := relay.GenerateApiKey(ctx) + if err != nil { + return nil, err + } + + insertData := do.ApiKeys{ + TenantId: tenantID, + ProjectId: req.Id, + Name: req.Name, + EncryptedKey: encryptedKey, + KeyPrefix: prefix, + Scope: req.Scope, + Status: "active", + KeyType: "project", + } + + if req.ExpiresAt != nil { + insertData.ExpiresAt = req.ExpiresAt + } + + result, err := dao.ApiKeys.Ctx(ctx).Data(insertData).Insert() + if err != nil { + return nil, err + } + id, _ := result.LastInsertId() + + for _, modelName := range req.ModelNames { + if modelName != "" { + if _, err = dao.ApiKeyModelScopes.Ctx(ctx).Insert(do.ApiKeyModelScopes{ + ApiKeyId: id, + ModelName: modelName, + }); err != nil { + g.Log().Warningf(ctx, "创建密钥模型范围失败: %v", err) + } + } + } + + return &v1.OpenProjectKeyCreateRes{ + ID: id, + Name: req.Name, + Key: rawKey, + KeyPrefix: prefix, + }, nil +} + +func (s *sOpen) OpenProjectKeyDelete(ctx context.Context, req *v1.OpenProjectKeyDeleteReq) (*v1.OpenProjectKeyDeleteRes, error) { + tenantID := middleware.GetOpenTenantID(ctx) + if tenantID == 0 { + return nil, errOpen(consts.CodeUnauthorized, "未认证") + } + if err := middleware.CheckOpenPermission(ctx, "projects:write"); err != nil { + return nil, err + } + + // 验证密钥属于该项目和租户 + count, err := dao.ApiKeys.Ctx(ctx). + Where("id", req.KeyId). + Where("tenant_id", tenantID). + Where("project_id", req.Id). + Count() + if err != nil { + return nil, err + } + if count == 0 { + return nil, errOpen(consts.CodeNotFound, "密钥不存在") + } + + _, err = dao.ApiKeys.Ctx(ctx). + Where("id", req.KeyId). + Where("tenant_id", tenantID). + Data(do.ApiKeys{Status: "revoked"}).Update() + return nil, err +} + +// ============================================================ +// 项目用量查询 +// ============================================================ + +func (s *sOpen) OpenProjectUsageStats(ctx context.Context, req *v1.OpenProjectUsageStatsReq) (*v1.OpenProjectUsageStatsRes, error) { + tenantID := middleware.GetOpenTenantID(ctx) + if tenantID == 0 { + return nil, errOpen(consts.CodeUnauthorized, "未认证") + } + if err := middleware.CheckOpenPermission(ctx, "projects:read"); err != nil { + return nil, err + } + + if _, err := s.getProjectOrError(ctx, req.Id, tenantID); err != nil { + return nil, err + } + + // 构建日期条件 + useCustomRange := req.StartDate != "" && req.EndDate != "" + if useCustomRange { + if err := common.ValidateDateParam(req.StartDate, "开始日期"); err != nil { + return nil, err + } + if err := common.ValidateDateParam(req.EndDate, "结束日期"); err != nil { + return nil, err + } + } + + dateCondition := func(m *gdb.Model) *gdb.Model { + if useCustomRange { + return m.Where("created_at >= ?", req.StartDate+" 00:00:00"). + Where("created_at <= ?", req.EndDate+" 23:59:59") + } + return m.Where("created_at >= NOW() - INTERVAL '30 days'") + } + + // 汇总 + type totalRow struct { + TotalCost float64 `json:"total_cost"` + RequestCount int `json:"request_count"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + } + var totalStats totalRow + dateCondition(dao.BilUsageLogs.Ctx(ctx). + Where("tenant_id", tenantID). + Where("project_id", req.Id). + Fields("COALESCE(SUM(actual_cost), 0) as total_cost, COUNT(*) as request_count, COALESCE(SUM(input_tokens), 0) as input_tokens, COALESCE(SUM(output_tokens), 0) as output_tokens")). + Scan(&totalStats) + + // 每日趋势 + type dailyRow struct { + Date string `json:"date"` + RequestCount int `json:"request_count"` + TotalCost float64 `json:"total_cost"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + } + var dailyStats []dailyRow + dateCondition(dao.BilUsageLogs.Ctx(ctx). + Where("tenant_id", tenantID). + Where("project_id", req.Id). + Fields("DATE(created_at) as date, COUNT(*) as request_count, COALESCE(SUM(actual_cost), 0) as total_cost, COALESCE(SUM(input_tokens), 0) as input_tokens, COALESCE(SUM(output_tokens), 0) as output_tokens")). + Group("DATE(created_at)").OrderAsc("date").Scan(&dailyStats) + + // 模型分布（Top 10） + type modelRow struct { + ModelName string `json:"model_name"` + RequestCount int `json:"request_count"` + TotalCost float64 `json:"total_cost"` + } + var modelStats []modelRow + dateCondition(dao.BilUsageLogs.Ctx(ctx). + Where("tenant_id", tenantID). + Where("project_id", req.Id). + Fields("model_name, COUNT(*) as request_count, COALESCE(SUM(actual_cost), 0) as total_cost")). + Group("model_name").OrderDesc("total_cost").Limit(10).Scan(&modelStats) + + daily := make([]v1.OpenProjectDailyStat, 0, len(dailyStats)) + for _, d := range dailyStats { + daily = append(daily, v1.OpenProjectDailyStat{ + Date: d.Date, + RequestCount: int64(d.RequestCount), + TotalCost: fmt.Sprintf("%.6f", d.TotalCost), + InputTokens: d.InputTokens, + OutputTokens: d.OutputTokens, + }) + } + + models := make([]v1.OpenProjectModelStat, 0, len(modelStats)) + for _, m := range modelStats { + models = append(models, v1.OpenProjectModelStat{ + ModelName: m.ModelName, + RequestCount: int64(m.RequestCount), + TotalCost: fmt.Sprintf("%.6f", m.TotalCost), + }) + } + + return &v1.OpenProjectUsageStatsRes{ + TotalCost: fmt.Sprintf("%.6f", totalStats.TotalCost), + TotalRequests: int64(totalStats.RequestCount), + TotalInputTokens: totalStats.InputTokens, + TotalOutputTokens: totalStats.OutputTokens, + Daily: daily, + Models: models, + }, nil +} + +func (s *sOpen) OpenProjectUsageLogs(ctx context.Context, req *v1.OpenProjectUsageLogsReq) (*v1.OpenProjectUsageLogsRes, error) { + tenantID := middleware.GetOpenTenantID(ctx) + if tenantID == 0 { + return nil, errOpen(consts.CodeUnauthorized, "未认证") + } + if err := middleware.CheckOpenPermission(ctx, "projects:read"); err != nil { + return nil, err + } + + if _, err := s.getProjectOrError(ctx, req.Id, tenantID); err != nil { + return nil, err + } + + page := req.Page + pageSize := req.PageSize + if page < 1 { + page = 1 + } + if pageSize < 1 || pageSize > 100 { + pageSize = 20 + } + + m := dao.BilUsageLogs.Ctx(ctx). + Where("tenant_id", tenantID). + Where("project_id", req.Id) + + total, err := m.Count() + if err != nil { + return nil, err + } + + type logRow struct { + Id int64 `json:"id"` + ModelName string `json:"model_name"` + RelayMode string `json:"relay_mode"` + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + TotalCost float64 `json:"total_cost"` + LatencyMs int `json:"latency_ms"` + Status string `json:"status"` + ErrorMessage string `json:"error_message"` + CreatedAt *gtime.Time `json:"created_at"` + } + + var logs []logRow + err = m.Fields("id, model_name, relay_mode, input_tokens, output_tokens, actual_cost as total_cost, latency_ms, status, error_message, created_at"). + OrderDesc("created_at").Page(page, pageSize).Scan(&logs) + if err != nil { + return nil, err + } + + items := make([]v1.OpenProjectUsageLogItem, 0, len(logs)) + for _, l := range logs { + item := v1.OpenProjectUsageLogItem{ + ID: l.Id, + ModelName: l.ModelName, + RelayMode: l.RelayMode, + InputTokens: l.InputTokens, + OutputTokens: l.OutputTokens, + TotalCost: fmt.Sprintf("%.6f", l.TotalCost), + LatencyMs: l.LatencyMs, + Status: l.Status, + ErrorMessage: l.ErrorMessage, + } + if l.CreatedAt != nil { + item.CreatedAt = l.CreatedAt.Format("Y-m-d H:i:s") + } + items = append(items, item) + } + + return &v1.OpenProjectUsageLogsRes{List: items, Total: total, Page: page, PageSize: pageSize}, nil +} + +// getProjectOrError 验证项目归属并返回基本信息 +func (s *sOpen) getProjectOrError(ctx context.Context, projectID, tenantID int64) (*struct { + ID int64 `json:"id"` + Status string `json:"status"` +}, error) { + var project *struct { + ID int64 `json:"id"` + Status string `json:"status"` + } + err := dao.TntProjects.Ctx(ctx). + Where("id", projectID). + Where("tenant_id", tenantID). + Scan(&project) + if err != nil { + return nil, err + } + if project == nil { + return nil, errOpen(consts.CodeNotFound, "项目不存在") + } + return project, nil +} diff --git a/internal/logic/relay/apikey.go b/internal/logic/relay/apikey.go index 3222f7b..0002211 100644 --- a/internal/logic/relay/apikey.go +++ b/internal/logic/relay/apikey.go @@ -40,15 +40,83 @@ func DefaultChannelSettings() common.ChannelSettings { } } +// legacyEncryptionKey 硬编码的旧默认密钥（仅迁移时使用） +const legacyEncryptionKey = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + // GetEncryptionKey 获取加密密钥 func GetEncryptionKey() []byte { key := g.Cfg().MustGet(context.Background(), "crypto.encryptionKey").String() if key == "" { - key = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + panic("crypto.encryptionKey is not configured") } return uc.MustGetEncryptionKey(key) } +// MigrateEncryptionKey 检测并迁移旧默认密钥加密的数据到新配置密钥。 +// 启动时调用一次，将 chn_channel_keys 和 api_keys 中用旧密钥加密的数据重新加密。 +func MigrateEncryptionKey(ctx context.Context) { + configKey := g.Cfg().MustGet(ctx, "crypto.encryptionKey").String() + if configKey == "" || configKey == legacyEncryptionKey { + // 未配置新密钥或新密钥与旧密钥相同，无需迁移 + return + } + + newKey := uc.MustGetEncryptionKey(configKey) + oldKey := uc.MustGetEncryptionKey(legacyEncryptionKey) + + migrateTable(ctx, "chn_channel_keys", oldKey, newKey) + migrateTable(ctx, "api_keys", oldKey, newKey) +} + +// migrateTable 迁移单张表的 encrypted_key 字段 +func migrateTable(ctx context.Context, table string, oldKey, newKey []byte) { + type row struct { + Id int64 `json:"id"` + EncryptedKey string `json:"encrypted_key"` + } + + var rows []row + err := g.DB().Model(table).Ctx(ctx). + Fields("id, encrypted_key"). + Scan(&rows) + if err != nil { + g.Log().Warningf(ctx, "[KeyMigration] query %s failed: %v", table, err) + return + } + + migrated := 0 + for _, r := range rows { + if r.EncryptedKey == "" { + continue + } + // 尝试用旧密钥解密 + plaintext, err := uc.DecryptString(oldKey, r.EncryptedKey) + if err != nil { + // 解密失败说明可能是已经用新密钥加密过的，跳过 + continue + } + // 用新密钥重新加密 + newEncrypted, err := uc.EncryptString(newKey, plaintext) + if err != nil { + g.Log().Warningf(ctx, "[KeyMigration] re-encrypt %s id=%d failed: %v", table, r.Id, err) + continue + } + _, err = g.DB().Model(table).Ctx(ctx). + Where("id", r.Id). + Data(g.Map{"encrypted_key": newEncrypted}). + Update() + if err != nil { + g.Log().Warningf(ctx, "[KeyMigration] update %s id=%d failed: %v", table, r.Id, err) + continue + } + migrated++ + } + + if migrated > 0 { + g.Log().Infof(ctx, "[KeyMigration] migrated %d rows in %s from legacy key to config key", migrated, table) + } +} + // apiKeyHash 计算 API Key 的 SHA-256 哈希 func apiKeyHash(rawKey string) string { h := sha256.Sum256([]byte(rawKey)) diff --git a/internal/logic/tenant/billing.go b/internal/logic/tenant/billing.go index 231db6f..e8e4151 100644 --- a/internal/logic/tenant/billing.go +++ b/internal/logic/tenant/billing.go @@ -508,3 +508,26 @@ func (s *sTenant) WalletFrozenItems(ctx context.Context, req *v1.TenantWalletFro return &v1.TenantWalletFrozenItemsRes{Items: details}, nil } + +// UpdateWarningThreshold 修改钱包预警阈值 +func (s *sTenant) UpdateWarningThreshold(ctx context.Context, req *v1.TenantWalletUpdateWarningThresholdReq) (*v1.TenantWalletUpdateWarningThresholdRes, error) { + role := middleware.GetUserRole(ctx) + if role != "owner" && role != "admin" { + return nil, common.NewForbiddenError("需要 owner 或 admin 权限") + } + tenantID := middleware.GetTenantID(ctx) + + _, err := dao.BilWallets.Ctx(ctx). + Where("tenant_id", tenantID). + Data(do.BilWallets{ + WarningThreshold: req.Threshold, + }).Update() + if err != nil { + return nil, err + } + + // 清除钱包缓存 + billing.InvalidateWalletRedis(ctx, tenantID) + + return &v1.TenantWalletUpdateWarningThresholdRes{}, nil +} diff --git a/internal/logic/tenant/open_platform.go b/internal/logic/tenant/open_platform.go index ac1ec6f..87b3621 100644 --- a/internal/logic/tenant/open_platform.go +++ b/internal/logic/tenant/open_platform.go @@ -8,6 +8,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "strings" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/os/gtime" @@ -356,6 +357,9 @@ func (s *sTenant) WebhookConfigUpdate(ctx context.Context, req *v1.WebhookConfig hasUpdate = true } if req.URL != nil { + if !strings.HasPrefix(*req.URL, "https://") { + return nil, common.NewBadRequestError("回调 URL 必须以 https:// 开头") + } data.Url = *req.URL hasUpdate = true } @@ -387,8 +391,31 @@ func (s *sTenant) WebhookConfigDelete(ctx context.Context, req *v1.WebhookConfig return nil, common.NewForbiddenError("需要 owner 或 admin 权限") } tenantID := middleware.GetTenantID(ctx) - _, err := dao.OpnWebhookConfigs.Ctx(ctx).Where("id", req.Id).Where("tenant_id", tenantID).Delete() - return nil, err + + // 先校验配置存在且属于当前租户 + count, err := dao.OpnWebhookConfigs.Ctx(ctx).Where("id", req.Id).Where("tenant_id", tenantID).Count() + if err != nil { + return nil, err + } + if count == 0 { + return nil, common.NewNotFoundError("Webhook 配置不存在") + } + + // 级联删除：投递日志 → 事件 → 配置 + _, err = dao.OpnWebhookDeliveryLogs.Ctx(ctx).Where("webhook_config_id", req.Id).Where("tenant_id", tenantID).Delete() + if err != nil { + return nil, err + } + _, err = dao.OpnWebhookEvents.Ctx(ctx).Where("webhook_config_id", req.Id).Where("tenant_id", tenantID).Delete() + if err != nil { + return nil, err + } + _, err = dao.OpnWebhookConfigs.Ctx(ctx).Where("id", req.Id).Where("tenant_id", tenantID).Delete() + if err != nil { + return nil, err + } + + return nil, nil } func (s *sTenant) WebhookDeliveryLogs(ctx context.Context, req *v1.WebhookDeliveryLogsReq) (*v1.WebhookDeliveryLogsRes, error) { @@ -406,52 +433,35 @@ func (s *sTenant) WebhookDeliveryLogs(ctx context.Context, req *v1.WebhookDelive pageSize = 20 } - baseQuery := dao.OpnWebhookDeliveryLogs.Ctx(ctx).Where("tenant_id", tenantID).Where("webhook_config_id", req.Id) - - // 分离计数查询和数据查询，避免 JOIN 影响 Count + // 计数查询（不 JOIN） + countQuery := dao.OpnWebhookDeliveryLogs.Ctx(ctx).Where("tenant_id", tenantID).Where("webhook_config_id", req.Id) if req.Status != "" { - // 用子查询过滤 status - total, err := baseQuery. - Where("event_id IN (SELECT id FROM opn_webhook_events WHERE status = ?)", req.Status). - Count() - if err != nil { - return nil, err - } - - var logs []entity.OpnWebhookDeliveryLogs - err = baseQuery. - Where("event_id IN (SELECT id FROM opn_webhook_events WHERE status = ?)", req.Status). - OrderDesc("created_at").Page(page, pageSize).Scan(&logs) - if err != nil { - return nil, err - } - - items := make([]v1.WebhookDeliveryLogItem, len(logs)) - for i, l := range logs { - items[i] = v1.WebhookDeliveryLogItem{ - ID: l.Id, - EventID: l.EventId, - Attempt: l.Attempt, - ResponseStatus: l.ResponseStatus, - ResponseTimeMs: l.ResponseTimeMs, - ErrorMessage: l.ErrorMessage, - } - if l.CreatedAt != nil { - items[i].CreatedAt = l.CreatedAt.Format("Y-m-d H:i:s") - } - } - - return &v1.WebhookDeliveryLogsRes{List: items, Total: total, Page: page, PageSize: pageSize}, nil + countQuery = countQuery.Where("event_id IN (SELECT id FROM opn_webhook_events WHERE status = ?)", req.Status) } - - // 无 status 过滤时，直接查询 - total, err := baseQuery.Count() + total, err := countQuery.Count() if err != nil { return nil, err } - var logs []entity.OpnWebhookDeliveryLogs - err = baseQuery.OrderDesc("created_at").Page(page, pageSize).Scan(&logs) + // 数据查询：LEFT JOIN opn_webhook_events 获取 event_type + db := g.DB() + query := db.Ctx(ctx).Raw( + "SELECT dl.*, e.event_type FROM opn_webhook_delivery_logs dl LEFT JOIN opn_webhook_events e ON dl.event_id = e.id WHERE dl.tenant_id = ? AND dl.webhook_config_id = ?", + tenantID, req.Id, + ) + if req.Status != "" { + query = db.Ctx(ctx).Raw( + "SELECT dl.*, e.event_type FROM opn_webhook_delivery_logs dl LEFT JOIN opn_webhook_events e ON dl.event_id = e.id WHERE dl.tenant_id = ? AND dl.webhook_config_id = ? AND e.status = ?", + tenantID, req.Id, req.Status, + ) + } + + type deliveryLogWithEventType struct { + entity.OpnWebhookDeliveryLogs + EventType string `json:"event_type"` + } + var logs []deliveryLogWithEventType + err = query.OrderDesc("dl.created_at").Page(page, pageSize).Scan(&logs) if err != nil { return nil, err } @@ -461,6 +471,7 @@ func (s *sTenant) WebhookDeliveryLogs(ctx context.Context, req *v1.WebhookDelive items[i] = v1.WebhookDeliveryLogItem{ ID: l.Id, EventID: l.EventId, + EventType: l.EventType, Attempt: l.Attempt, ResponseStatus: l.ResponseStatus, ResponseTimeMs: l.ResponseTimeMs, diff --git a/internal/logic/tenant/webhook_dispatcher.go b/internal/logic/tenant/webhook_dispatcher.go index 584052f..080ed67 100644 --- a/internal/logic/tenant/webhook_dispatcher.go +++ b/internal/logic/tenant/webhook_dispatcher.go @@ -15,6 +15,13 @@ const ( dispatcherChannelSize = 1000 dispatcherWorkerCount = 5 dispatcherSweepInterval = 2 * time.Minute + + // 清理策略 + cleanupInterval = 6 * time.Hour // 每 6 小时执行一次清理 + cleanupDeliveredAge = 7 * 24 * time.Hour // 已投递事件保留 7 天 + cleanupFailedAge = 30 * 24 * time.Hour // 已放弃事件保留 30 天 + cleanupDeliveryLogAge = 30 * 24 * time.Hour // 投递日志保留 30 天 + cleanupBatchSize = 500 // 每批删除条数 ) type WebhookDispatcher struct { @@ -36,11 +43,12 @@ func InitWebhookDispatcher(ctx context.Context) { go d.worker() } go d.sweepLoop() + go d.cleanupLoop() // Recover any pending events left from a previous crash go d.recoverPending() - g.Log().Infof(ctx, "webhook dispatcher started: %d workers, sweep every %v", dispatcherWorkerCount, dispatcherSweepInterval) + g.Log().Infof(ctx, "webhook dispatcher started: %d workers, sweep every %v, cleanup every %v", dispatcherWorkerCount, dispatcherSweepInterval, cleanupInterval) } // NotifyNewEvent pushes an event ID to the dispatcher channel (non-blocking). @@ -108,8 +116,96 @@ func (d *WebhookDispatcher) recoverPending() { } // scheduleRetry schedules a delayed re-delivery for a failed event. +// The retry is skipped if the dispatcher context has been cancelled (e.g. during shutdown). func scheduleRetry(eventID int64, delay time.Duration) { time.AfterFunc(delay, func() { + if defaultDispatcher == nil { + return + } + select { + case <-defaultDispatcher.ctx.Done(): + return + default: + } NotifyNewEvent(eventID) }) } + +// cleanupLoop periodically removes old delivered events, exhausted failed events, and delivery logs. +func (d *WebhookDispatcher) cleanupLoop() { + ticker := time.NewTicker(cleanupInterval) + defer ticker.Stop() + + for { + select { + case <-d.ctx.Done(): + return + case <-ticker.C: + d.cleanup() + } + } +} + +// cleanup removes old webhook data in batches to prevent table bloat. +// Order: delivery_logs → events (delivery_logs reference events via event_id). +func (d *WebhookDispatcher) cleanup() { + ctx := d.ctx + now := gtime.Now() + + // 1. 清理投递日志（保留 30 天） + logCutoff := now.Add(-cleanupDeliveryLogAge) + for { + result, err := dao.OpnWebhookDeliveryLogs.Ctx(ctx). + Where("created_at < ?", logCutoff). + Limit(cleanupBatchSize). + Delete() + if err != nil { + g.Log().Errorf(ctx, "webhook cleanup: delete delivery logs failed: %v", err) + break + } + rows, _ := result.RowsAffected() + if rows == 0 { + break + } + g.Log().Infof(ctx, "webhook cleanup: deleted %d delivery logs older than %v", rows, logCutoff.Format("Y-m-d")) + } + + // 2. 清理已投递事件（保留 7 天） + deliveredCutoff := now.Add(-cleanupDeliveredAge) + for { + result, err := dao.OpnWebhookEvents.Ctx(ctx). + Where("status", "delivered"). + Where("updated_at < ?", deliveredCutoff). + Limit(cleanupBatchSize). + Delete() + if err != nil { + g.Log().Errorf(ctx, "webhook cleanup: delete delivered events failed: %v", err) + break + } + rows, _ := result.RowsAffected() + if rows == 0 { + break + } + g.Log().Infof(ctx, "webhook cleanup: deleted %d delivered events older than %v", rows, deliveredCutoff.Format("Y-m-d")) + } + + // 3. 清理已放弃的重试事件（attempts >= max，保留 30 天） + failedCutoff := now.Add(-cleanupFailedAge) + for { + result, err := dao.OpnWebhookEvents.Ctx(ctx). + Where("status", "failed"). + Where("attempts >= ?", webhookMaxRetries). + Where("updated_at < ?", failedCutoff). + Limit(cleanupBatchSize). + Delete() + if err != nil { + g.Log().Errorf(ctx, "webhook cleanup: delete exhausted events failed: %v", err) + break + } + rows, _ := result.RowsAffected() + if rows == 0 { + break + } + g.Log().Infof(ctx, "webhook cleanup: deleted %d exhausted events older than %v", rows, failedCutoff.Format("Y-m-d")) + } +} diff --git a/internal/logic/tenant/webhook_worker.go b/internal/logic/tenant/webhook_worker.go index dd2e4ed..f8356ca 100644 --- a/internal/logic/tenant/webhook_worker.go +++ b/internal/logic/tenant/webhook_worker.go @@ -3,9 +3,6 @@ package tenant import ( "bytes" "context" - "crypto/hmac" - "crypto/sha256" - "encoding/hex" "encoding/json" "fmt" "io" @@ -28,6 +25,9 @@ const ( webhookMaxRetries = 5 ) +// webhookHTTPClient 全局复用的 HTTP 客户端，共享连接池，避免每次投递创建新实例。 +var webhookHTTPClient = &http.Client{Timeout: webhookTimeout} + var retryIntervals = []time.Duration{ 1 * time.Minute, 5 * time.Minute, @@ -85,7 +85,7 @@ func deliverEvent(ctx context.Context, evt entity.OpnWebhookEvents) { timestamp := strconv.FormatInt(time.Now().Unix(), 10) body := []byte(evt.Payload) - signature := computeDeliverySignature(config.SecretKey, timestamp, body) + signature := ComputeWebhookSignature(config.SecretKey, timestamp, body) req, err := http.NewRequestWithContext(ctx, "POST", config.Url, bytes.NewReader(body)) if err != nil { @@ -100,9 +100,8 @@ func deliverEvent(ctx context.Context, evt entity.OpnWebhookEvents) { req.Header.Set("X-Webhook-Event", evt.EventType) req.Header.Set("X-Webhook-ID", evt.EventId) - client := &http.Client{Timeout: webhookTimeout} startTime := time.Now() - resp, err := client.Do(req) + resp, err := webhookHTTPClient.Do(req) elapsed := time.Since(startTime).Milliseconds() if err != nil { @@ -201,12 +200,6 @@ func recordDeliveryLog(ctx context.Context, evt entity.OpnWebhookEvents, config } } -func computeDeliverySignature(secret, timestamp string, body []byte) string { - mac := hmac.New(sha256.New, []byte(secret)) - mac.Write([]byte(timestamp + "." + string(body))) - return hex.EncodeToString(mac.Sum(nil)) -} - func truncateString(s string, maxLen int) string { if len(s) <= maxLen { return s diff --git a/internal/middleware/open_auth.go b/internal/middleware/open_auth.go index 0a2763e..6bf631b 100644 --- a/internal/middleware/open_auth.go +++ b/internal/middleware/open_auth.go @@ -7,11 +7,12 @@ import ( "encoding/hex" "encoding/json" "fmt" - "io" "strconv" "strings" "time" + "github.com/gogf/gf/v2/errors/gcode" + "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/net/ghttp" @@ -20,6 +21,8 @@ import ( "github.com/qianfree/team-api/internal/model/entity" "github.com/qianfree/team-api/internal/response" "github.com/qianfree/team-api/internal/utility/crypto" + + lcommon "github.com/qianfree/team-api/internal/logic/common" ) // openContextKey is a private type for context keys to prevent collisions. @@ -38,6 +41,8 @@ const ( ctxKeyOpenIsSandbox openContextKey = "openIsSandbox" ) +var openAppCache = lcommon.NewCache("open_app", 60*time.Second) + // OpenPlatformAuth is the HMAC-SHA256 authentication middleware for /open/* endpoints. func OpenPlatformAuth(r *ghttp.Request) { appID := r.GetHeader(openPlatformHeaderAppID) @@ -61,12 +66,16 @@ func OpenPlatformAuth(r *ghttp.Request) { return } - // Lookup app by app_id + // Lookup app by app_id (cache-first) var app entity.OpnApps - err = dao.OpnApps.Ctx(r.Context()).Where("app_id", appID).Scan(&app) - if err != nil || app.Id == 0 { - response.Error(r, consts.ErrUnauthorized) - return + cacheKey := appID + if !openAppCache.GetJSON(r.Context(), cacheKey, &app) || app.Id == 0 { + err = dao.OpnApps.Ctx(r.Context()).Where("app_id", appID).Scan(&app) + if err != nil || app.Id == 0 { + response.Error(r, consts.ErrUnauthorized) + return + } + openAppCache.Set(r.Context(), cacheKey, &app) } if app.Status != "active" { @@ -80,6 +89,13 @@ func OpenPlatformAuth(r *ghttp.Request) { return } + // Per-app rate limiting (fixed window, 1 minute) + if app.RateLimit > 0 { + if !checkOpenAppRateLimit(r, app.Id, app.RateLimit) { + return + } + } + // Get app secret from Redis secret, err := getOpenAppSecret(r.Context(), app.Id) if err != nil || secret == "" { @@ -88,10 +104,8 @@ func OpenPlatformAuth(r *ghttp.Request) { } // Read request body for signature verification - var bodyBytes []byte - if r.Method == "POST" || r.Method == "PUT" || r.Method == "PATCH" { - bodyBytes, _ = io.ReadAll(r.Body) - } + // Use r.GetBody() instead of io.ReadAll(r.Body) to preserve body for downstream handlers + bodyBytes := r.GetBody() // Verify HMAC signature (includes body hash for request integrity) expectedSig := computeHMACSignature(secret, timestampStr, nonce, r.Method, r.URL.Path, bodyBytes) @@ -186,3 +200,69 @@ func GetOpenAppID(ctx context.Context) int64 { } return 0 } + +// checkOpenAppRateLimit enforces per-app rate limiting using a Redis fixed-window counter. +// Returns true if the request is allowed, false if rate limited (429 response already written). +func checkOpenAppRateLimit(r *ghttp.Request, appDBID int64, rateLimit int) bool { + ctx := r.Context() + now := time.Now() + minuteTimestamp := now.Unix() / 60 + key := fmt.Sprintf("ratelimit:open:%d:%d", appDBID, minuteTimestamp) + + count, err := g.Redis().Do(ctx, "INCR", key) + if err != nil { + g.Log().Warningf(ctx, "[OpenRateLimit] INCR failed app=%d: %v", appDBID, err) + // Fail open on Redis error + return true + } + + currentCount := count.Int64() + if currentCount == 1 { + // First request in this window, set TTL + _, _ = g.Redis().Do(ctx, "EXPIRE", key, 60) + } + + remaining := rateLimit - int(currentCount) + if remaining < 0 { + remaining = 0 + } + r.Response.Header().Set("X-RateLimit-Limit", strconv.Itoa(rateLimit)) + r.Response.Header().Set("X-RateLimit-Remaining", strconv.Itoa(remaining)) + + if int(currentCount) > rateLimit { + response.ErrorWithCode(r, 429, consts.CodeRateLimitExceeded, consts.MsgRateLimitExceeded) + return false + } + + return true +} + +// GetOpenPermissions extracts the permissions string (JSON) injected by OpenPlatformAuth. +func GetOpenPermissions(ctx context.Context) string { + val := ctx.Value(ctxKeyOpenPermissions) + if val != nil { + if s, ok := val.(string); ok { + return s + } + } + return "" +} + +// CheckOpenPermission checks if the authenticated app has the given permission. +// Returns error if permission is missing. +func CheckOpenPermission(ctx context.Context, permission string) error { + perms := GetOpenPermissions(ctx) + if perms == "" { + return gerror.NewCode(gcode.New(consts.CodeForbidden, "无权限", nil), "应用无此操作权限") + } + var list []string + if err := json.Unmarshal([]byte(perms), &list); err != nil { + return gerror.NewCode(gcode.New(consts.CodeForbidden, "无权限", nil), "应用无此操作权限") + } + for _, p := range list { + if p == permission { + return nil + } + } + return gerror.NewCode(gcode.New(consts.CodeForbidden, "无权限", nil), "应用无此操作权限") +} diff --git a/internal/service/open.go b/internal/service/open.go index f60afe6..08fe164 100644 --- a/internal/service/open.go +++ b/internal/service/open.go @@ -21,6 +21,17 @@ type ( OpenKeyDelete(ctx context.Context, req *v1.OpenKeyDeleteReq) (*v1.OpenKeyDeleteRes, error) OpenUsageQuery(ctx context.Context, req *v1.OpenUsageQueryReq) (*v1.OpenUsageQueryRes, error) OpenBillingQuery(ctx context.Context, req *v1.OpenBillingQueryReq) (*v1.OpenBillingQueryRes, error) + OpenProjectList(ctx context.Context, req *v1.OpenProjectListReq) (*v1.OpenProjectListRes, error) + OpenProjectCreate(ctx context.Context, req *v1.OpenProjectCreateReq) (*v1.OpenProjectCreateRes, error) + OpenProjectGet(ctx context.Context, req *v1.OpenProjectGetReq) (*v1.OpenProjectGetRes, error) + OpenProjectUpdate(ctx context.Context, req *v1.OpenProjectUpdateReq) (*v1.OpenProjectUpdateRes, error) + OpenProjectArchive(ctx context.Context, req *v1.OpenProjectArchiveReq) (*v1.OpenProjectArchiveRes, error) + OpenProjectUnarchive(ctx context.Context, req *v1.OpenProjectUnarchiveReq) (*v1.OpenProjectUnarchiveRes, error) + OpenProjectKeyList(ctx context.Context, req *v1.OpenProjectKeyListReq) (*v1.OpenProjectKeyListRes, error) + OpenProjectKeyCreate(ctx context.Context, req *v1.OpenProjectKeyCreateReq) (*v1.OpenProjectKeyCreateRes, error) + OpenProjectKeyDelete(ctx context.Context, req *v1.OpenProjectKeyDeleteReq) (*v1.OpenProjectKeyDeleteRes, error) + OpenProjectUsageStats(ctx context.Context, req *v1.OpenProjectUsageStatsReq) (*v1.OpenProjectUsageStatsRes, error) + OpenProjectUsageLogs(ctx context.Context, req *v1.OpenProjectUsageLogsReq) (*v1.OpenProjectUsageLogsRes, error) } ) diff --git a/internal/service/tenant.go b/internal/service/tenant.go index abee19f..9e220f2 100644 --- a/internal/service/tenant.go +++ b/internal/service/tenant.go @@ -66,6 +66,8 @@ type ( ExportWalletTransactions(ctx context.Context, req *v1.TenantWalletTransactionsExportReq) (*v1.TenantWalletTransactionsExportRes, error) // WalletFrozenItems 获取冻结明细 WalletFrozenItems(ctx context.Context, req *v1.TenantWalletFrozenItemsReq) (*v1.TenantWalletFrozenItemsRes, error) + // UpdateWarningThreshold 修改钱包预警阈值 + UpdateWarningThreshold(ctx context.Context, req *v1.TenantWalletUpdateWarningThresholdReq) (*v1.TenantWalletUpdateWarningThresholdRes, error) // Dashboard returns the tenant dashboard statistics. Dashboard(ctx context.Context, req *v1.TenantDashboardReq) (*v1.TenantDashboardRes, error) // TokenTrends returns daily token usage for the past N days. diff --git a/migrations/000007_webhook_indexes_and_cleanup.sql b/migrations/000007_webhook_indexes_and_cleanup.sql new file mode 100644 index 0000000..8258349 --- /dev/null +++ b/migrations/000007_webhook_indexes_and_cleanup.sql @@ -0,0 +1,31 @@ +-- +goose Up +-- Webhook 性能优化：补充索引 + 事件清理策略 +-- 1. opn_webhook_configs: GIN 索引支持 JSONB @> 查询，复合索引加速活跃配置查找 +-- 2. opn_webhook_delivery_logs: 按配置+租户查询索引 +-- 3. opn_webhook_events: 按配置查询索引（级联删除用），BRIN 索引支持追加写入 + +-- 1. JSONB GIN 索引：支持 PublishWebhookEvent 中的 events::jsonb @> ? 查询 +CREATE INDEX idx_opn_webhook_configs_events_gin ON opn_webhook_configs USING gin (events jsonb_path_ops); + +-- 2. 复合索引：加速查找租户的活跃 webhook 配置 +CREATE INDEX idx_opn_webhook_configs_tenant_active ON opn_webhook_configs (tenant_id, is_active) WHERE is_active = true; + +-- 3. 按配置+租户查询投递日志（WebhookDeliveryLogs API 使用） +CREATE INDEX idx_opn_webhook_delivery_logs_config ON opn_webhook_delivery_logs (webhook_config_id, tenant_id, created_at DESC); + +-- 4. 按配置查询事件（级联删除时使用） +CREATE INDEX idx_opn_webhook_events_config ON opn_webhook_events (webhook_config_id); + +-- 5. BRIN 索引：events 表追加写入模式，磁盘友好 +CREATE INDEX idx_opn_webhook_events_created_brin ON opn_webhook_events USING brin (created_at); + +-- 6. BRIN 索引：delivery_logs 表追加写入模式 +CREATE INDEX idx_opn_webhook_delivery_logs_created_brin ON opn_webhook_delivery_logs USING brin (created_at); + +-- +goose Down +DROP INDEX IF EXISTS idx_opn_webhook_delivery_logs_created_brin; +DROP INDEX IF EXISTS idx_opn_webhook_events_created_brin; +DROP INDEX IF EXISTS idx_opn_webhook_events_config; +DROP INDEX IF EXISTS idx_opn_webhook_delivery_logs_config; +DROP INDEX IF EXISTS idx_opn_webhook_configs_tenant_active; +DROP INDEX IF EXISTS idx_opn_webhook_configs_events_gin; diff --git a/web/tenant/src/views/finance/WalletPage.vue b/web/tenant/src/views/finance/WalletPage.vue index 4f131aa..4723fd6 100644 --- a/web/tenant/src/views/finance/WalletPage.vue +++ b/web/tenant/src/views/finance/WalletPage.vue @@ -67,6 +67,33 @@ const isLowBalance = computed(() => { return wallet.value.available_balance <= (wallet.value.warning_threshold || 0) }) +// Warning threshold setting +const showThresholdModal = ref(false) +const thresholdInput = ref('') +const thresholdSaving = ref(false) + +function openThresholdModal() { + thresholdInput.value = wallet.value?.warning_threshold?.toFixed(2) ?? '1.00' + showThresholdModal.value = true +} + +async function saveThreshold() { + const val = parseFloat(thresholdInput.value) + if (isNaN(val) || val < 0) return + thresholdSaving.value = true + try { + const res: any = await request.put('/tenant/wallet/warning-threshold', { threshold: val }) + if (res.data?.code === 0) { + showThresholdModal.value = false + await fetchWallet() + } + } catch (e) { + console.error(e) + } finally { + thresholdSaving.value = false + } +} + async function fetchWallet() { try { const res: any = await request.get('/tenant/wallet') @@ -307,6 +334,15 @@ onBeforeUnmount(() => {
余额低于预警线 ${{ wallet.warning_threshold?.toFixed(2) }} + +
+ + +
+
@@ -599,6 +635,47 @@ onBeforeUnmount(() => { + + + +
+
+
+

余额预警设置

+ +
+
+
+

当可用余额低于设定的阈值时，系统将通过站内通知和 Webhook 推送预警消息。

+
+
+ +
+ $ + +
+

设为 0 表示关闭余额预警

+
+
+
+ + +
+
+
+ @@ -696,6 +773,12 @@ onBeforeUnmount(() => { background: rgba(251, 191, 36, 0.15); border: 1px solid rgba(251, 191, 36, 0.2); } +.balance-threshold-btn { + display: inline-flex; align-items: center; gap: 0.375rem; + font-size: 0.75rem; font-weight: 500; color: rgba(255,255,255,0.5); + transition: color 0.15s; +} +.balance-threshold-btn:hover { color: rgba(255,255,255,0.75); } .balance-chip { display: flex; align-items: center; justify-content: space-between; border-radius: 0.75rem; padding: 0.625rem 0.875rem; diff --git a/web/tenant/src/views/open/OpenAppsPage.vue b/web/tenant/src/views/open/OpenAppsPage.vue index e4c4047..1937984 100644 --- a/web/tenant/src/views/open/OpenAppsPage.vue +++ b/web/tenant/src/views/open/OpenAppsPage.vue @@ -45,6 +45,7 @@ const newAppId = ref('') const availablePermissions = [ 'members:read', 'members:write', 'keys:read', 'keys:write', + 'projects:read', 'projects:write', 'models:read', 'usage:read', 'billing:read', diff --git a/web/tenant/src/views/open/WebhooksPage.vue b/web/tenant/src/views/open/WebhooksPage.vue index 7782301..08177f9 100644 --- a/web/tenant/src/views/open/WebhooksPage.vue +++ b/web/tenant/src/views/open/WebhooksPage.vue @@ -1,5 +1,5 @@