feat: 卡网关按运营商账户限流并记录停复机调用
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 9m33s

- 新增网关限流:按本系统运营商账户(卡 carrier_id)→ 限流组 → 操作限制每秒请求数,每次 HTTP 尝试计量,默认限流、仅后台人工点击/通道阈值/自动续费/手动刷新豁免;支持只统计与拦截两种模式,默认只统计
- Asynq 遇限流不计失败、固定短延迟重试;轮询任务与停复机业务重试遇限流短延迟重排
- 停机、复机不再使用网关内部重试,只发一次请求
- 网关返回结构化错误;新增上游错误分类规则表与测试接口
- 新增按月分区的停复机记录表:系统自动停复机只写记录,不再写统一审计与集成日志;普通轮询查询不再写集成日志
- 新增限流组、错误分类、停复机记录、按账户调用统计与枚举后台接口
- 迁移 000243~000246;同步 OpenSpec 提案(含后续 6 个 change)与技术方案、工程约束
This commit is contained in:
邹进
2026-10-24 15:06:41 +09:00
parent 9e45097aa2
commit 42e881576d
106 changed files with 7173 additions and 48 deletions

View File

@@ -11,6 +11,7 @@ import (
"github.com/gofiber/fiber/v2"
"github.com/hibiken/asynq"
"github.com/break/junhong_cmp_fiber/internal/gateway"
"github.com/break/junhong_cmp_fiber/internal/handler/validation"
dto "github.com/break/junhong_cmp_fiber/internal/model/dto"
packageExpiryQuery "github.com/break/junhong_cmp_fiber/internal/query/packageexpiry"
@@ -447,7 +448,7 @@ func (h *AssetHandler) Stop(c *fiber.Ctx) error {
switch asset.AssetType {
case "device":
result, devErr := h.deviceService.StopDevice(c.UserContext(), asset.AssetID)
result, devErr := h.deviceService.StopDevice(gateway.WithRateLimitExempt(c.UserContext(), constants.RateLimitExemptManualClick), asset.AssetID)
if devErr != nil {
return devErr
}
@@ -456,7 +457,7 @@ func (h *AssetHandler) Stop(c *fiber.Ctx) error {
if asset.ICCID == "" {
return errors.New(errors.CodeInternalError, "卡 ICCID 为空,无法停机")
}
if cardErr := h.iotCardStopResume.ManualStopCard(c.UserContext(), asset.ICCID); cardErr != nil {
if cardErr := h.iotCardStopResume.ManualStopCard(gateway.WithRateLimitExempt(c.UserContext(), constants.RateLimitExemptManualClick), asset.ICCID); cardErr != nil {
return cardErr
}
return response.Success(c, nil)
@@ -475,7 +476,7 @@ func (h *AssetHandler) Start(c *fiber.Ctx) error {
switch asset.AssetType {
case "device":
if devErr := h.deviceService.StartDevice(c.UserContext(), asset.AssetID); devErr != nil {
if devErr := h.deviceService.StartDevice(gateway.WithRateLimitExempt(c.UserContext(), constants.RateLimitExemptManualClick), asset.AssetID); devErr != nil {
return devErr
}
return response.Success(c, nil)
@@ -483,7 +484,7 @@ func (h *AssetHandler) Start(c *fiber.Ctx) error {
if asset.ICCID == "" {
return errors.New(errors.CodeInternalError, "卡 ICCID 为空,无法复机")
}
if cardErr := h.iotCardStopResume.ManualStartCard(c.UserContext(), asset.ICCID); cardErr != nil {
if cardErr := h.iotCardStopResume.ManualStartCard(gateway.WithRateLimitExempt(c.UserContext(), constants.RateLimitExemptManualClick), asset.ICCID); cardErr != nil {
return cardErr
}
return response.Success(c, nil)

View File

@@ -0,0 +1,343 @@
package admin
import (
"strconv"
"github.com/go-playground/validator/v10"
"github.com/gofiber/fiber/v2"
"github.com/redis/go-redis/v9"
"github.com/break/junhong_cmp_fiber/internal/model/dto"
pollingService "github.com/break/junhong_cmp_fiber/internal/service/polling"
"github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/response"
)
// GatewayRateLimitHandler 运营商账户网关限流、上游错误分类、停复机记录与调用统计 Handler。
type GatewayRateLimitHandler struct {
service *pollingService.GatewayRateLimitService
redis *redis.Client
validator *validator.Validate
}
// NewGatewayRateLimitHandler 创建网关限流 Handler。
func NewGatewayRateLimitHandler(service *pollingService.GatewayRateLimitService, redisClient *redis.Client, validate *validator.Validate) *GatewayRateLimitHandler {
if validate == nil {
validate = validator.New()
}
return &GatewayRateLimitHandler{service: service, redis: redisClient, validator: validate}
}
func (h *GatewayRateLimitHandler) validate(prefix string, req any) error {
if err := h.validator.Struct(req); err != nil {
return errors.New(errors.CodeInvalidParam, validationMessage(prefix, req, err))
}
return nil
}
func rateLimitPathID(c *fiber.Ctx, name, label string) (uint, error) {
value, err := strconv.ParseUint(c.Params(name), 10, 64)
if err != nil || value == 0 {
return 0, errors.New(errors.CodeInvalidParam, "无效的"+label)
}
return uint(value), nil
}
// ListGroups 查询网关限流组列表。
// GET /api/admin/polling-rate-limit-groups
func (h *GatewayRateLimitHandler) ListGroups(c *fiber.Ctx) error {
var req dto.RateLimitGroupListRequest
if err := c.QueryParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
}
if err := h.validate("网关限流组查询参数不合法", &req); err != nil {
return err
}
result, err := h.service.ListGroups(c.UserContext(), &req)
if err != nil {
return err
}
return response.SuccessWithPagination(c, result.List, result.Total, result.Page, result.PageSize)
}
// GetGroup 查询网关限流组详情。
// GET /api/admin/polling-rate-limit-groups/:id
func (h *GatewayRateLimitHandler) GetGroup(c *fiber.Ctx) error {
id, err := rateLimitPathID(c, "id", "限流组 ID")
if err != nil {
return err
}
result, err := h.service.GetGroup(c.UserContext(), id)
if err != nil {
return err
}
return response.Success(c, result)
}
// CreateGroup 新建网关限流组。
// POST /api/admin/polling-rate-limit-groups
func (h *GatewayRateLimitHandler) CreateGroup(c *fiber.Ctx) error {
var req dto.SaveRateLimitGroupRequest
if err := c.BodyParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
}
if err := h.validate("网关限流组参数不合法", &req); err != nil {
return err
}
result, err := h.service.CreateGroup(c.UserContext(), &req)
if err != nil {
return err
}
return response.Success(c, result)
}
// UpdateGroup 修改网关限流组。
// PUT /api/admin/polling-rate-limit-groups/:id
func (h *GatewayRateLimitHandler) UpdateGroup(c *fiber.Ctx) error {
var req dto.SaveRateLimitGroupRequest
if err := c.BodyParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
}
id, err := rateLimitPathID(c, "id", "限流组 ID")
if err != nil {
return err
}
req.ID = id
if err := h.validate("网关限流组参数不合法", &req); err != nil {
return err
}
result, err := h.service.UpdateGroup(c.UserContext(), &req)
if err != nil {
return err
}
return response.Success(c, result)
}
// UpdateGroupStatus 启停网关限流组。
// PUT /api/admin/polling-rate-limit-groups/:id/status
func (h *GatewayRateLimitHandler) UpdateGroupStatus(c *fiber.Ctx) error {
var req dto.UpdateRateLimitGroupStatusRequest
if err := c.BodyParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
}
id, err := rateLimitPathID(c, "id", "限流组 ID")
if err != nil {
return err
}
req.ID = id
if err := h.validate("网关限流组状态参数不合法", &req); err != nil {
return err
}
if err := h.service.UpdateGroupStatus(c.UserContext(), req.ID, *req.Status); err != nil {
return err
}
return response.Success(c, nil)
}
// DeleteGroup 删除网关限流组。
// DELETE /api/admin/polling-rate-limit-groups/:id
func (h *GatewayRateLimitHandler) DeleteGroup(c *fiber.Ctx) error {
id, err := rateLimitPathID(c, "id", "限流组 ID")
if err != nil {
return err
}
if err := h.service.DeleteGroup(c.UserContext(), id); err != nil {
return err
}
return response.Success(c, nil)
}
// ListRules 查询上游错误分类规则列表。
// GET /api/admin/gateway-error-classifications
func (h *GatewayRateLimitHandler) ListRules(c *fiber.Ctx) error {
var req dto.GatewayErrorRuleListRequest
if err := c.QueryParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
}
if err := h.validate("上游错误分类规则查询参数不合法", &req); err != nil {
return err
}
result, err := h.service.ListRules(c.UserContext(), &req)
if err != nil {
return err
}
return response.SuccessWithPagination(c, result.List, result.Total, result.Page, result.PageSize)
}
// CreateRule 新建上游错误分类规则。
// POST /api/admin/gateway-error-classifications
func (h *GatewayRateLimitHandler) CreateRule(c *fiber.Ctx) error {
var req dto.SaveGatewayErrorRuleRequest
if err := c.BodyParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
}
if err := h.validate("上游错误分类规则参数不合法", &req); err != nil {
return err
}
result, err := h.service.CreateRule(c.UserContext(), &req)
if err != nil {
return err
}
return response.Success(c, result)
}
// UpdateRule 修改上游错误分类规则。
// PUT /api/admin/gateway-error-classifications/:id
func (h *GatewayRateLimitHandler) UpdateRule(c *fiber.Ctx) error {
var req dto.SaveGatewayErrorRuleRequest
if err := c.BodyParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
}
id, err := rateLimitPathID(c, "id", "规则 ID")
if err != nil {
return err
}
req.ID = id
if err := h.validate("上游错误分类规则参数不合法", &req); err != nil {
return err
}
result, err := h.service.UpdateRule(c.UserContext(), &req)
if err != nil {
return err
}
return response.Success(c, result)
}
// UpdateRuleStatus 启停上游错误分类规则。
// PUT /api/admin/gateway-error-classifications/:id/status
func (h *GatewayRateLimitHandler) UpdateRuleStatus(c *fiber.Ctx) error {
var req dto.UpdateGatewayErrorRuleStatusRequest
if err := c.BodyParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
}
id, err := rateLimitPathID(c, "id", "规则 ID")
if err != nil {
return err
}
req.ID = id
if err := h.validate("上游错误分类规则状态参数不合法", &req); err != nil {
return err
}
if err := h.service.UpdateRuleStatus(c.UserContext(), req.ID, *req.Status); err != nil {
return err
}
return response.Success(c, nil)
}
// DeleteRule 删除上游错误分类规则。
// DELETE /api/admin/gateway-error-classifications/:id
func (h *GatewayRateLimitHandler) DeleteRule(c *fiber.Ctx) error {
id, err := rateLimitPathID(c, "id", "规则 ID")
if err != nil {
return err
}
if err := h.service.DeleteRule(c.UserContext(), id); err != nil {
return err
}
return response.Success(c, nil)
}
// TestRule 用已启用规则(可附带草稿)对一段上游返回信息分类,不写任何数据。
// POST /api/admin/gateway-error-classifications/test
func (h *GatewayRateLimitHandler) TestRule(c *fiber.Ctx) error {
var req dto.TestGatewayErrorRuleRequest
if err := c.BodyParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
}
if err := h.validate("上游错误分类测试参数不合法", &req); err != nil {
return err
}
result, err := h.service.TestRule(c.UserContext(), &req)
if err != nil {
return err
}
return response.Success(c, result)
}
// ListStopResumeRecords 查询停复机记录。
// GET /api/admin/card-stop-resume-records
func (h *GatewayRateLimitHandler) ListStopResumeRecords(c *fiber.Ctx) error {
var req dto.StopResumeRecordListRequest
if err := c.QueryParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败,时间请使用 RFC3339 格式")
}
if err := h.validate("停复机记录查询参数不合法", &req); err != nil {
return err
}
result, err := h.service.ListStopResumeRecords(c.UserContext(), &req)
if err != nil {
return err
}
return response.SuccessWithPagination(c, result.List, result.Total, result.Page, result.PageSize)
}
// ListCardStopResumeRecords 查询单卡停复机历史。
// GET /api/admin/iot-cards/:id/stop-resume-records
func (h *GatewayRateLimitHandler) ListCardStopResumeRecords(c *fiber.Ctx) error {
var req dto.CardStopResumeRecordListRequest
if err := c.QueryParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败,时间请使用 RFC3339 格式")
}
id, err := rateLimitPathID(c, "id", "IoT 卡 ID")
if err != nil {
return err
}
req.ID = id
if err := h.validate("单卡停复机记录查询参数不合法", &req); err != nil {
return err
}
result, err := h.service.ListCardStopResumeRecords(c.UserContext(), &req)
if err != nil {
return err
}
return response.SuccessWithPagination(c, result.List, result.Total, result.Page, result.PageSize)
}
// CarrierCallStats 按运营商账户汇总最近若干分钟的网关调用。
// GET /api/admin/polling-stats/carriers
func (h *GatewayRateLimitHandler) CarrierCallStats(c *fiber.Ctx) error {
var req dto.GatewayCallStatsRequest
if err := c.QueryParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
}
if err := h.validate("网关调用统计参数不合法", &req); err != nil {
return err
}
result, err := h.service.CarrierCallStats(c.UserContext(), h.redis, &req)
if err != nil {
return err
}
return response.Success(c, result)
}
// CarrierCallTrend 查询单个运营商账户的网关调用趋势。
// GET /api/admin/polling-stats/carriers/:carrier_id/trend
func (h *GatewayRateLimitHandler) CarrierCallTrend(c *fiber.Ctx) error {
var req dto.GatewayCallTrendRequest
if err := c.QueryParser(&req); err != nil {
return errors.New(errors.CodeInvalidParam, "请求参数解析失败")
}
carrierID, err := strconv.ParseUint(c.Params("carrier_id"), 10, 64)
if err != nil {
return errors.New(errors.CodeInvalidParam, "无效的运营商账户 ID")
}
req.CarrierID = uint(carrierID)
if err := h.validate("网关调用趋势参数不合法", &req); err != nil {
return err
}
result, err := h.service.CarrierCallTrend(c.UserContext(), h.redis, &req)
if err != nil {
return err
}
return response.Success(c, result)
}
// PollingOptions 返回轮询与网关限流相关枚举。
// GET /api/admin/polling-options
func (h *GatewayRateLimitHandler) PollingOptions(c *fiber.Ctx) error {
result, err := h.service.PollingOptions(c.UserContext())
if err != nil {
return err
}
return response.Success(c, result)
}