csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity

docs(constitution): 新增数据库设计原则(v2.4.0)

Browse Source 
在项目宪章中新增第九条原则"数据库设计原则"，明确禁止使用数据库外键约束和ORM关联标签。

主要变更：
- 新增原则IX：数据库设计原则（Database Design Principles）
- 强制要求：数据库表不得使用外键约束
- 强制要求：GORM模型不得使用ORM关联标签（foreignKey、hasMany等）
- 强制要求：表关系必须通过ID字段手动维护
- 强制要求：关联数据查询必须显式编写，避免ORM魔法
- 强制要求：时间字段由GORM处理，不使用数据库触发器

设计理念：
- 提升业务逻辑灵活性（无数据库约束限制）
- 优化高并发性能（无外键检查开销）
- 增强代码可读性（显式查询，无隐式预加载）
- 简化数据库架构和迁移流程
- 支持分布式和微服务场景

版本升级：2.3.0 → 2.4.0（MINOR）
This commit is contained in:
huang
2025-11-13 13:40:19 +08:00
parent ea0c6a8b16
commit 984ccccc63
63 changed files with 12099 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
internal/handler/health.go
View File

@@ -1,18 +1,116 @@
package handler
import (
"context"
"time"
"github.com/gofiber/fiber/v2"
"github.com/redis/go-redis/v9"
"go.uber.org/zap"
"gorm.io/gorm"
"github.com/break/junhong_cmp_fiber/pkg/logger"
"github.com/break/junhong_cmp_fiber/pkg/response"
)
// HealthCheck 健康检查处理器
// HealthHandler 健康检查处理器
type HealthHandler struct {
db *gorm.DB
redis *redis.Client
logger *zap.Logger
}
// NewHealthHandler 创建健康检查处理器实例
func NewHealthHandler(db *gorm.DB, redis *redis.Client, logger *zap.Logger) *HealthHandler {
return &HealthHandler{
db: db,
redis: redis,
logger: logger,
}
}
// Check 健康检查
// GET /health
func (h *HealthHandler) Check(c *fiber.Ctx) error {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
healthStatus := fiber.Map{
"status": "healthy",
"timestamp": time.Now().Format(time.RFC3339),
"services": fiber.Map{},
}
services := healthStatus["services"].(fiber.Map)
allHealthy := true
// 检查 PostgreSQL
sqlDB, err := h.db.DB()
if err != nil {
h.logger.Error("获取 PostgreSQL DB 实例失败", zap.Error(err))
services["postgres"] = fiber.Map{
"status": "down",
"error": err.Error(),
}
allHealthy = false
} else {
if err := sqlDB.PingContext(ctx); err != nil {
h.logger.Error("PostgreSQL Ping 失败", zap.Error(err))
services["postgres"] = fiber.Map{
"status": "down",
"error": err.Error(),
}
allHealthy = false
} else {
// 获取连接池统计信息
stats := sqlDB.Stats()
services["postgres"] = fiber.Map{
"status": "up",
"open_conns": stats.OpenConnections,
"in_use": stats.InUse,
"idle": stats.Idle,
"wait_count": stats.WaitCount,
"wait_duration": stats.WaitDuration.String(),
"max_idle_close": stats.MaxIdleClosed,
"max_lifetime_close": stats.MaxLifetimeClosed,
}
}
}
// 检查 Redis
if err := h.redis.Ping(ctx).Err(); err != nil {
h.logger.Error("Redis Ping 失败", zap.Error(err))
services["redis"] = fiber.Map{
"status": "down",
"error": err.Error(),
}
allHealthy = false
} else {
// 获取 Redis 信息
poolStats := h.redis.PoolStats()
services["redis"] = fiber.Map{
"status": "up",
"hits": poolStats.Hits,
"misses": poolStats.Misses,
"timeouts": poolStats.Timeouts,
"total_conns": poolStats.TotalConns,
"idle_conns": poolStats.IdleConns,
"stale_conns": poolStats.StaleConns,
}
}
// 设置总体状态
if !allHealthy {
healthStatus["status"] = "degraded"
h.logger.Warn("健康检查失败: 部分服务不可用")
return c.Status(fiber.StatusServiceUnavailable).JSON(healthStatus)
}
h.logger.Info("健康检查成功: 所有服务正常")
return response.Success(c, healthStatus)
}
// HealthCheck 简单健康检查(保持向后兼容)
func HealthCheck(c *fiber.Ctx) error {
logger.GetAppLogger().Info("我还活着!!!!", zap.String("time", time.Now().Format(time.RFC3339)))
return response.Success(c, fiber.Map{
"status": "healthy",
"timestamp": time.Now().Format(time.RFC3339),

239
internal/handler/order.go Normal file
View File

@@ -0,0 +1,239 @@
package handler
import (
"strconv"
"github.com/break/junhong_cmp_fiber/internal/model"
"github.com/break/junhong_cmp_fiber/internal/service/order"
"github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/response"
"github.com/gofiber/fiber/v2"
"go.uber.org/zap"
)
// OrderHandler 订单处理器
type OrderHandler struct {
orderService *order.Service
logger *zap.Logger
}
// NewOrderHandler 创建订单处理器实例
func NewOrderHandler(orderService *order.Service, logger *zap.Logger) *OrderHandler {
return &OrderHandler{
orderService: orderService,
logger: logger,
}
}
// CreateOrder 创建订单
// POST /api/v1/orders
func (h *OrderHandler) CreateOrder(c *fiber.Ctx) error {
var req model.CreateOrderRequest
// 解析请求体
if err := c.BodyParser(&req); err != nil {
h.logger.Warn("解析请求体失败",
zap.String("path", c.Path()),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, "请求参数格式错误")
}
// 验证请求参数
if err := validate.Struct(&req); err != nil {
h.logger.Warn("参数验证失败",
zap.String("path", c.Path()),
zap.Any("request", req),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, err.Error())
}
// 调用服务层创建订单
orderResp, err := h.orderService.CreateOrder(c.Context(), &req)
if err != nil {
if e, ok := err.(*errors.AppError); ok {
httpStatus := fiber.StatusInternalServerError
if e.Code == errors.CodeNotFound {
httpStatus = fiber.StatusNotFound
}
return response.Error(c, httpStatus, e.Code, e.Message)
}
h.logger.Error("创建订单失败",
zap.String("order_id", req.OrderID),
zap.Error(err))
return response.Error(c, fiber.StatusInternalServerError, errors.CodeInternalError, "创建订单失败")
}
h.logger.Info("订单创建成功",
zap.Uint("order_id", orderResp.ID),
zap.String("order_no", orderResp.OrderID))
return response.Success(c, orderResp)
}
// GetOrder 获取订单详情
// GET /api/v1/orders/:id
func (h *OrderHandler) GetOrder(c *fiber.Ctx) error {
// 获取路径参数
idStr := c.Params("id")
id, err := strconv.ParseUint(idStr, 10, 32)
if err != nil {
h.logger.Warn("订单ID格式错误",
zap.String("id", idStr),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, "订单ID格式错误")
}
// 调用服务层获取订单
orderResp, err := h.orderService.GetOrderByID(c.Context(), uint(id))
if err != nil {
if e, ok := err.(*errors.AppError); ok {
httpStatus := fiber.StatusInternalServerError
if e.Code == errors.CodeNotFound {
httpStatus = fiber.StatusNotFound
}
return response.Error(c, httpStatus, e.Code, e.Message)
}
h.logger.Error("获取订单失败",
zap.Uint("order_id", uint(id)),
zap.Error(err))
return response.Error(c, fiber.StatusInternalServerError, errors.CodeInternalError, "获取订单失败")
}
return response.Success(c, orderResp)
}
// UpdateOrder 更新订单信息
// PUT /api/v1/orders/:id
func (h *OrderHandler) UpdateOrder(c *fiber.Ctx) error {
// 获取路径参数
idStr := c.Params("id")
id, err := strconv.ParseUint(idStr, 10, 32)
if err != nil {
h.logger.Warn("订单ID格式错误",
zap.String("id", idStr),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, "订单ID格式错误")
}
var req model.UpdateOrderRequest
// 解析请求体
if err := c.BodyParser(&req); err != nil {
h.logger.Warn("解析请求体失败",
zap.String("path", c.Path()),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, "请求参数格式错误")
}
// 验证请求参数
if err := validate.Struct(&req); err != nil {
h.logger.Warn("参数验证失败",
zap.String("path", c.Path()),
zap.Any("request", req),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, err.Error())
}
// 调用服务层更新订单
orderResp, err := h.orderService.UpdateOrder(c.Context(), uint(id), &req)
if err != nil {
if e, ok := err.(*errors.AppError); ok {
httpStatus := fiber.StatusInternalServerError
if e.Code == errors.CodeNotFound {
httpStatus = fiber.StatusNotFound
}
return response.Error(c, httpStatus, e.Code, e.Message)
}
h.logger.Error("更新订单失败",
zap.Uint("order_id", uint(id)),
zap.Error(err))
return response.Error(c, fiber.StatusInternalServerError, errors.CodeInternalError, "更新订单失败")
}
h.logger.Info("订单更新成功",
zap.Uint("order_id", uint(id)))
return response.Success(c, orderResp)
}
// ListOrders 获取订单列表（分页）
// GET /api/v1/orders
func (h *OrderHandler) ListOrders(c *fiber.Ctx) error {
// 获取查询参数
page, err := strconv.Atoi(c.Query("page", "1"))
if err != nil || page < 1 {
page = 1
}
pageSize, err := strconv.Atoi(c.Query("page_size", "20"))
if err != nil || pageSize < 1 {
pageSize = 20
}
if pageSize > 100 {
pageSize = 100 // 限制最大页大小
}
// 可选的用户ID过滤
var userID uint
if userIDStr := c.Query("user_id"); userIDStr != "" {
if id, err := strconv.ParseUint(userIDStr, 10, 32); err == nil {
userID = uint(id)
}
}
// 调用服务层获取订单列表
var orders []model.Order
var total int64
if userID > 0 {
// 按用户ID查询
orders, total, err = h.orderService.ListOrdersByUserID(c.Context(), userID, page, pageSize)
} else {
// 查询所有订单
orders, total, err = h.orderService.ListOrders(c.Context(), page, pageSize)
}
if err != nil {
if e, ok := err.(*errors.AppError); ok {
return response.Error(c, fiber.StatusInternalServerError, e.Code, e.Message)
}
h.logger.Error("获取订单列表失败",
zap.Int("page", page),
zap.Int("page_size", pageSize),
zap.Uint("user_id", userID),
zap.Error(err))
return response.Error(c, fiber.StatusInternalServerError, errors.CodeInternalError, "获取订单列表失败")
}
// 构造响应
totalPages := int(total) / pageSize
if int(total)%pageSize > 0 {
totalPages++
}
listResp := model.ListOrdersResponse{
Orders: make([]model.OrderResponse, 0, len(orders)),
Page: page,
PageSize: pageSize,
Total: total,
TotalPages: totalPages,
}
// 转换为响应格式
for _, o := range orders {
listResp.Orders = append(listResp.Orders, model.OrderResponse{
ID: o.ID,
OrderID: o.OrderID,
UserID: o.UserID,
Amount: o.Amount,
Status: o.Status,
Remark: o.Remark,
PaidAt: o.PaidAt,
CompletedAt: o.CompletedAt,
CreatedAt: o.CreatedAt,
UpdatedAt: o.UpdatedAt,
})
}
return response.Success(c, listResp)
}

213
internal/handler/task.go Normal file
View File

@@ -0,0 +1,213 @@
package handler
import (
"fmt"
"time"
"github.com/gofiber/fiber/v2"
"github.com/google/uuid"
"github.com/hibiken/asynq"
"go.uber.org/zap"
"github.com/break/junhong_cmp_fiber/internal/task"
"github.com/break/junhong_cmp_fiber/pkg/constants"
"github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/queue"
"github.com/break/junhong_cmp_fiber/pkg/response"
)
// TaskHandler 任务处理器
type TaskHandler struct {
queueClient *queue.Client
logger *zap.Logger
}
// NewTaskHandler 创建任务处理器实例
func NewTaskHandler(queueClient *queue.Client, logger *zap.Logger) *TaskHandler {
return &TaskHandler{
queueClient: queueClient,
logger: logger,
}
}
// SubmitEmailTaskRequest 提交邮件任务请求
type SubmitEmailTaskRequest struct {
To string `json:"to" validate:"required,email"`
Subject string `json:"subject" validate:"required,min=1,max=200"`
Body string `json:"body" validate:"required,min=1"`
CC []string `json:"cc,omitempty" validate:"omitempty,dive,email"`
Attachments []string `json:"attachments,omitempty"`
RequestID string `json:"request_id,omitempty"`
}
// SubmitSyncTaskRequest 提交数据同步任务请求
type SubmitSyncTaskRequest struct {
SyncType string `json:"sync_type" validate:"required,oneof=sim_status flow_usage real_name"`
StartDate string `json:"start_date" validate:"required"`
EndDate string `json:"end_date" validate:"required"`
BatchSize int `json:"batch_size,omitempty" validate:"omitempty,min=1,max=1000"`
RequestID string `json:"request_id,omitempty"`
Priority string `json:"priority,omitempty" validate:"omitempty,oneof=critical default low"`
}
// TaskResponse 任务响应
type TaskResponse struct {
TaskID string `json:"task_id"`
Queue string `json:"queue"`
Status string `json:"status"`
}
// SubmitEmailTask 提交邮件发送任务
// @Summary 提交邮件发送任务
// @Description 异步发送邮件
// @Tags 任务
// @Accept json
// @Produce json
// @Param request body SubmitEmailTaskRequest true "邮件任务参数"
// @Success 200 {object} response.Response{data=TaskResponse}
// @Failure 400 {object} response.Response
// @Router /api/v1/tasks/email [post]
func (h *TaskHandler) SubmitEmailTask(c *fiber.Ctx) error {
var req SubmitEmailTaskRequest
if err := c.BodyParser(&req); err != nil {
h.logger.Warn("解析邮件任务请求失败",
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, "请求参数格式错误")
}
// 验证参数
if err := validate.Struct(&req); err != nil {
h.logger.Warn("邮件任务参数验证失败",
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, err.Error())
}
// 生成 RequestID（如果未提供）
if req.RequestID == "" {
req.RequestID = generateRequestID("email")
}
// 构造任务载荷
payload := &task.EmailPayload{
RequestID: req.RequestID,
To: req.To,
Subject: req.Subject,
Body: req.Body,
CC: req.CC,
Attachments: req.Attachments,
}
// 提交任务到队列
err := h.queueClient.EnqueueTask(
c.Context(),
constants.TaskTypeEmailSend,
payload,
asynq.Queue(constants.QueueDefault),
asynq.MaxRetry(constants.DefaultRetryMax),
asynq.Timeout(constants.DefaultTimeout),
)
if err != nil {
h.logger.Error("提交邮件任务失败",
zap.String("to", req.To),
zap.String("request_id", req.RequestID),
zap.Error(err))
return response.Error(c, fiber.StatusInternalServerError, errors.CodeInternalError, "任务提交失败")
}
h.logger.Info("邮件任务提交成功",
zap.String("queue", constants.QueueDefault),
zap.String("to", req.To),
zap.String("request_id", req.RequestID))
return response.SuccessWithMessage(c, TaskResponse{
TaskID: req.RequestID,
Queue: constants.QueueDefault,
Status: "queued",
}, "邮件任务已提交")
}
// SubmitSyncTask 提交数据同步任务
// @Summary 提交数据同步任务
// @Description 异步执行数据同步
// @Tags 任务
// @Accept json
// @Produce json
// @Param request body SubmitSyncTaskRequest true "同步任务参数"
// @Success 200 {object} response.Response{data=TaskResponse}
// @Failure 400 {object} response.Response
// @Router /api/v1/tasks/sync [post]
func (h *TaskHandler) SubmitSyncTask(c *fiber.Ctx) error {
var req SubmitSyncTaskRequest
if err := c.BodyParser(&req); err != nil {
h.logger.Warn("解析同步任务请求失败",
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, "请求参数格式错误")
}
// 验证参数
if err := validate.Struct(&req); err != nil {
h.logger.Warn("同步任务参数验证失败",
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, err.Error())
}
// 生成 RequestID（如果未提供）
if req.RequestID == "" {
req.RequestID = generateRequestID("sync")
}
// 设置默认批量大小
if req.BatchSize == 0 {
req.BatchSize = 100
}
// 确定队列优先级
queueName := constants.QueueDefault
if req.Priority == "critical" {
queueName = constants.QueueCritical
} else if req.Priority == "low" {
queueName = constants.QueueLow
}
// 构造任务载荷
payload := &task.DataSyncPayload{
RequestID: req.RequestID,
SyncType: req.SyncType,
StartDate: req.StartDate,
EndDate: req.EndDate,
BatchSize: req.BatchSize,
}
// 提交任务到队列
err := h.queueClient.EnqueueTask(
c.Context(),
constants.TaskTypeDataSync,
payload,
asynq.Queue(queueName),
asynq.MaxRetry(constants.DefaultRetryMax),
asynq.Timeout(constants.DefaultTimeout),
)
if err != nil {
h.logger.Error("提交同步任务失败",
zap.String("sync_type", req.SyncType),
zap.String("request_id", req.RequestID),
zap.Error(err))
return response.Error(c, fiber.StatusInternalServerError, errors.CodeInternalError, "任务提交失败")
}
h.logger.Info("同步任务提交成功",
zap.String("queue", queueName),
zap.String("sync_type", req.SyncType),
zap.String("request_id", req.RequestID))
return response.SuccessWithMessage(c, TaskResponse{
TaskID: req.RequestID,
Queue: queueName,
Status: "queued",
}, "同步任务已提交")
}
// generateRequestID 生成请求 ID
func generateRequestID(prefix string) string {
return fmt.Sprintf("%s-%s-%d", prefix, uuid.New().String(), time.Now().UnixNano())
}

261
internal/handler/user.go
View File

@@ -1,33 +1,250 @@
package handler
import (
"github.com/gofiber/fiber/v2"
"strconv"
"github.com/break/junhong_cmp_fiber/pkg/constants"
"github.com/break/junhong_cmp_fiber/internal/model"
"github.com/break/junhong_cmp_fiber/internal/service/user"
"github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/response"
"github.com/go-playground/validator/v10"
"github.com/gofiber/fiber/v2"
"go.uber.org/zap"
)
// GetUsers 获取用户列表（示例受保护端点）
func GetUsers(c *fiber.Ctx) error {
// 从上下文获取用户 ID（由 auth 中间件设置）
userID := c.Locals(constants.ContextKeyUserID)
var validate = validator.New()
// 示例数据
users := []fiber.Map{
{
"id": "user-123",
"name": "张三",
"email": "zhangsan@example.com",
},
{
"id": "user-456",
"name": "李四",
"email": "lisi@example.com",
},
// UserHandler 用户处理器
type UserHandler struct {
userService *user.Service
logger *zap.Logger
}
// NewUserHandler 创建用户处理器实例
func NewUserHandler(userService *user.Service, logger *zap.Logger) *UserHandler {
return &UserHandler{
userService: userService,
logger: logger,
}
}
// CreateUser 创建用户
// POST /api/v1/users
func (h *UserHandler) CreateUser(c *fiber.Ctx) error {
var req model.CreateUserRequest
// 解析请求体
if err := c.BodyParser(&req); err != nil {
h.logger.Warn("解析请求体失败",
zap.String("path", c.Path()),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, "请求参数格式错误")
}
return response.SuccessWithMessage(c, fiber.Map{
"users": users,
"authenticated_as": userID,
}, "success")
// 验证请求参数
if err := validate.Struct(&req); err != nil {
h.logger.Warn("参数验证失败",
zap.String("path", c.Path()),
zap.Any("request", req),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, err.Error())
}
// 调用服务层创建用户
userResp, err := h.userService.CreateUser(c.Context(), &req)
if err != nil {
if e, ok := err.(*errors.AppError); ok {
return response.Error(c, fiber.StatusInternalServerError, e.Code, e.Message)
}
h.logger.Error("创建用户失败",
zap.String("username", req.Username),
zap.Error(err))
return response.Error(c, fiber.StatusInternalServerError, errors.CodeInternalError, "创建用户失败")
}
h.logger.Info("用户创建成功",
zap.Uint("user_id", userResp.ID),
zap.String("username", userResp.Username))
return response.Success(c, userResp)
}
// GetUser 获取用户详情
// GET /api/v1/users/:id
func (h *UserHandler) GetUser(c *fiber.Ctx) error {
// 获取路径参数
idStr := c.Params("id")
id, err := strconv.ParseUint(idStr, 10, 32)
if err != nil {
h.logger.Warn("用户ID格式错误",
zap.String("id", idStr),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, "用户ID格式错误")
}
// 调用服务层获取用户
userResp, err := h.userService.GetUserByID(c.Context(), uint(id))
if err != nil {
if e, ok := err.(*errors.AppError); ok {
httpStatus := fiber.StatusInternalServerError
if e.Code == errors.CodeNotFound {
httpStatus = fiber.StatusNotFound
}
return response.Error(c, httpStatus, e.Code, e.Message)
}
h.logger.Error("获取用户失败",
zap.Uint("user_id", uint(id)),
zap.Error(err))
return response.Error(c, fiber.StatusInternalServerError, errors.CodeInternalError, "获取用户失败")
}
return response.Success(c, userResp)
}
// UpdateUser 更新用户信息
// PUT /api/v1/users/:id
func (h *UserHandler) UpdateUser(c *fiber.Ctx) error {
// 获取路径参数
idStr := c.Params("id")
id, err := strconv.ParseUint(idStr, 10, 32)
if err != nil {
h.logger.Warn("用户ID格式错误",
zap.String("id", idStr),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, "用户ID格式错误")
}
var req model.UpdateUserRequest
// 解析请求体
if err := c.BodyParser(&req); err != nil {
h.logger.Warn("解析请求体失败",
zap.String("path", c.Path()),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, "请求参数格式错误")
}
// 验证请求参数
if err := validate.Struct(&req); err != nil {
h.logger.Warn("参数验证失败",
zap.String("path", c.Path()),
zap.Any("request", req),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, err.Error())
}
// 调用服务层更新用户
userResp, err := h.userService.UpdateUser(c.Context(), uint(id), &req)
if err != nil {
if e, ok := err.(*errors.AppError); ok {
httpStatus := fiber.StatusInternalServerError
if e.Code == errors.CodeNotFound {
httpStatus = fiber.StatusNotFound
}
return response.Error(c, httpStatus, e.Code, e.Message)
}
h.logger.Error("更新用户失败",
zap.Uint("user_id", uint(id)),
zap.Error(err))
return response.Error(c, fiber.StatusInternalServerError, errors.CodeInternalError, "更新用户失败")
}
h.logger.Info("用户更新成功",
zap.Uint("user_id", uint(id)))
return response.Success(c, userResp)
}
// DeleteUser 删除用户（软删除）
// DELETE /api/v1/users/:id
func (h *UserHandler) DeleteUser(c *fiber.Ctx) error {
// 获取路径参数
idStr := c.Params("id")
id, err := strconv.ParseUint(idStr, 10, 32)
if err != nil {
h.logger.Warn("用户ID格式错误",
zap.String("id", idStr),
zap.Error(err))
return response.Error(c, fiber.StatusBadRequest, errors.CodeBadRequest, "用户ID格式错误")
}
// 调用服务层删除用户
if err := h.userService.DeleteUser(c.Context(), uint(id)); err != nil {
if e, ok := err.(*errors.AppError); ok {
httpStatus := fiber.StatusInternalServerError
if e.Code == errors.CodeNotFound {
httpStatus = fiber.StatusNotFound
}
return response.Error(c, httpStatus, e.Code, e.Message)
}
h.logger.Error("删除用户失败",
zap.Uint("user_id", uint(id)),
zap.Error(err))
return response.Error(c, fiber.StatusInternalServerError, errors.CodeInternalError, "删除用户失败")
}
h.logger.Info("用户删除成功",
zap.Uint("user_id", uint(id)))
return response.Success(c, nil)
}
// ListUsers 获取用户列表（分页）
// GET /api/v1/users
func (h *UserHandler) ListUsers(c *fiber.Ctx) error {
// 获取查询参数
page, err := strconv.Atoi(c.Query("page", "1"))
if err != nil || page < 1 {
page = 1
}
pageSize, err := strconv.Atoi(c.Query("page_size", "20"))
if err != nil || pageSize < 1 {
pageSize = 20
}
if pageSize > 100 {
pageSize = 100 // 限制最大页大小
}
// 调用服务层获取用户列表
users, total, err := h.userService.ListUsers(c.Context(), page, pageSize)
if err != nil {
if e, ok := err.(*errors.AppError); ok {
return response.Error(c, fiber.StatusInternalServerError, e.Code, e.Message)
}
h.logger.Error("获取用户列表失败",
zap.Int("page", page),
zap.Int("page_size", pageSize),
zap.Error(err))
return response.Error(c, fiber.StatusInternalServerError, errors.CodeInternalError, "获取用户列表失败")
}
// 构造响应
totalPages := int(total) / pageSize
if int(total)%pageSize > 0 {
totalPages++
}
listResp := model.ListUsersResponse{
Users: make([]model.UserResponse, 0, len(users)),
Page: page,
PageSize: pageSize,
Total: total,
TotalPages: totalPages,
}
// 转换为响应格式
for _, u := range users {
listResp.Users = append(listResp.Users, model.UserResponse{
ID: u.ID,
Username: u.Username,
Email: u.Email,
Status: u.Status,
CreatedAt: u.CreatedAt,
UpdatedAt: u.UpdatedAt,
LastLoginAt: u.LastLoginAt,
})
}
return response.Success(c, listResp)
}

15
internal/model/base.go Normal file
View File

@@ -0,0 +1,15 @@
package model
import (
"time"
"gorm.io/gorm"
)
// BaseModel 基础模型，包含通用字段
type BaseModel struct {
ID uint `gorm:"primarykey" json:"id"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
DeletedAt gorm.DeletedAt `gorm:"index" json:"-"` // 软删除
}

30
internal/model/order.go Normal file
View File

@@ -0,0 +1,30 @@
package model
import (
"time"
)
// Order 订单实体
type Order struct {
BaseModel
// 业务唯一键
OrderID string `gorm:"uniqueIndex:uk_order_order_id;not null;size:50" json:"order_id"`
// 关联关系 (仅存储 ID,不使用 GORM 关联)
UserID uint `gorm:"not null;index:idx_order_user_id" json:"user_id"`
// 订单信息
Amount int64 `gorm:"not null" json:"amount"` // 金额（分）
Status string `gorm:"not null;size:20;default:'pending';index:idx_order_status" json:"status"`
Remark string `gorm:"size:500" json:"remark,omitempty"`
// 时间字段
PaidAt *time.Time `gorm:"column:paid_at" json:"paid_at,omitempty"`
CompletedAt *time.Time `gorm:"column:completed_at" json:"completed_at,omitempty"`
}
// TableName 指定表名
func (Order) TableName() string {
return "tb_order"
}

43
internal/model/order_dto.go Normal file
View File

@@ -0,0 +1,43 @@
package model
import (
"time"
)
// CreateOrderRequest 创建订单请求
type CreateOrderRequest struct {
OrderID string `json:"order_id" validate:"required,min=10,max=50"`
UserID uint `json:"user_id" validate:"required,gt=0"`
Amount int64 `json:"amount" validate:"required,gte=0"`
Remark string `json:"remark" validate:"omitempty,max=500"`
}
// UpdateOrderRequest 更新订单请求
type UpdateOrderRequest struct {
Status *string `json:"status" validate:"omitempty,oneof=pending paid processing completed cancelled"`
Remark *string `json:"remark" validate:"omitempty,max=500"`
}
// OrderResponse 订单响应
type OrderResponse struct {
ID uint `json:"id"`
OrderID string `json:"order_id"`
UserID uint `json:"user_id"`
Amount int64 `json:"amount"`
Status string `json:"status"`
Remark string `json:"remark,omitempty"`
PaidAt *time.Time `json:"paid_at,omitempty"`
CompletedAt *time.Time `json:"completed_at,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
User *UserResponse `json:"user,omitempty"` // 可选的用户信息
}
// ListOrdersResponse 订单列表响应
type ListOrdersResponse struct {
Orders []OrderResponse `json:"orders"`
Page int `json:"page"`
PageSize int `json:"page_size"`
Total int64 `json:"total"`
TotalPages int `json:"total_pages"`
}

26
internal/model/user.go Normal file
View File

@@ -0,0 +1,26 @@
package model
import (
"time"
)
// User 用户实体
type User struct {
BaseModel
// 基本信息
Username string `gorm:"uniqueIndex:uk_user_username;not null;size:50" json:"username"`
Email string `gorm:"uniqueIndex:uk_user_email;not null;size:100" json:"email"`
Password string `gorm:"not null;size:255" json:"-"` // 不返回给客户端
// 状态字段
Status string `gorm:"not null;size:20;default:'active';index:idx_user_status" json:"status"`
// 元数据
LastLoginAt *time.Time `gorm:"column:last_login_at" json:"last_login_at,omitempty"`
}
// TableName 指定表名
func (User) TableName() string {
return "tb_user"
}

38
internal/model/user_dto.go Normal file
View File

@@ -0,0 +1,38 @@
package model
import (
"time"
)
// CreateUserRequest 创建用户请求
type CreateUserRequest struct {
Username string `json:"username" validate:"required,min=3,max=50,alphanum"`
Email string `json:"email" validate:"required,email"`
Password string `json:"password" validate:"required,min=8"`
}
// UpdateUserRequest 更新用户请求
type UpdateUserRequest struct {
Email *string `json:"email" validate:"omitempty,email"`
Status *string `json:"status" validate:"omitempty,oneof=active inactive suspended"`
}
// UserResponse 用户响应
type UserResponse struct {
ID uint `json:"id"`
Username string `json:"username"`
Email string `json:"email"`
Status string `json:"status"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
LastLoginAt *time.Time `json:"last_login_at,omitempty"`
}
// ListUsersResponse 用户列表响应
type ListUsersResponse struct {
Users []UserResponse `json:"users"`
Page int `json:"page"`
PageSize int `json:"page_size"`
Total int64 `json:"total"`
TotalPages int `json:"total_pages"`
}

156
internal/service/email/service.go Normal file
View File

@@ -0,0 +1,156 @@
package email
import (
"context"
"fmt"
"github.com/break/junhong_cmp_fiber/internal/task"
"github.com/break/junhong_cmp_fiber/pkg/constants"
"github.com/break/junhong_cmp_fiber/pkg/queue"
"github.com/bytedance/sonic"
"github.com/hibiken/asynq"
"go.uber.org/zap"
)
// Service 邮件服务
type Service struct {
queueClient *queue.Client
logger *zap.Logger
}
// NewService 创建邮件服务实例
func NewService(queueClient *queue.Client, logger *zap.Logger) *Service {
return &Service{
queueClient: queueClient,
logger: logger,
}
}
// SendWelcomeEmail 发送欢迎邮件（异步）
func (s *Service) SendWelcomeEmail(ctx context.Context, userID uint, email string) error {
// 构造任务载荷
payload := &task.EmailPayload{
RequestID: fmt.Sprintf("welcome-%d", userID),
To: email,
Subject: "欢迎加入君鸿卡管系统",
Body: "感谢您注册我们的服务！我们很高兴为您提供服务。",
}
payloadBytes, err := sonic.Marshal(payload)
if err != nil {
s.logger.Error("序列化邮件任务载荷失败",
zap.Uint("user_id", userID),
zap.String("email", email),
zap.Error(err))
return fmt.Errorf("序列化邮件任务载荷失败: %w", err)
}
// 提交任务到队列
err = s.queueClient.EnqueueTask(
ctx,
constants.TaskTypeEmailSend,
payloadBytes,
asynq.Queue(constants.QueueDefault),
asynq.MaxRetry(constants.DefaultRetryMax),
)
if err != nil {
s.logger.Error("提交欢迎邮件任务失败",
zap.Uint("user_id", userID),
zap.String("email", email),
zap.Error(err))
return fmt.Errorf("提交欢迎邮件任务失败: %w", err)
}
s.logger.Info("欢迎邮件任务已提交",
zap.Uint("user_id", userID),
zap.String("email", email))
return nil
}
// SendPasswordResetEmail 发送密码重置邮件（异步）
func (s *Service) SendPasswordResetEmail(ctx context.Context, email string, resetToken string) error {
// 构造任务载荷
payload := &task.EmailPayload{
RequestID: fmt.Sprintf("reset-%s-%s", email, resetToken),
To: email,
Subject: "密码重置请求",
Body: fmt.Sprintf("您的密码重置令牌是: %s\n此令牌将在 1 小时后过期。", resetToken),
}
payloadBytes, err := sonic.Marshal(payload)
if err != nil {
s.logger.Error("序列化密码重置邮件任务载荷失败",
zap.String("email", email),
zap.Error(err))
return fmt.Errorf("序列化密码重置邮件任务载荷失败: %w", err)
}
// 提交任务到队列（高优先级）
err = s.queueClient.EnqueueTask(
ctx,
constants.TaskTypeEmailSend,
payloadBytes,
asynq.Queue(constants.QueueCritical), // 密码重置使用高优先级队列
asynq.MaxRetry(constants.DefaultRetryMax),
)
if err != nil {
s.logger.Error("提交密码重置邮件任务失败",
zap.String("email", email),
zap.Error(err))
return fmt.Errorf("提交密码重置邮件任务失败: %w", err)
}
s.logger.Info("密码重置邮件任务已提交",
zap.String("email", email))
return nil
}
// SendNotificationEmail 发送通知邮件（异步）
func (s *Service) SendNotificationEmail(ctx context.Context, to string, subject string, body string) error {
// 构造任务载荷
payload := &task.EmailPayload{
RequestID: fmt.Sprintf("notify-%s-%d", to, getCurrentTimestamp()),
To: to,
Subject: subject,
Body: body,
}
payloadBytes, err := sonic.Marshal(payload)
if err != nil {
s.logger.Error("序列化通知邮件任务载荷失败",
zap.String("to", to),
zap.Error(err))
return fmt.Errorf("序列化通知邮件任务载荷失败: %w", err)
}
// 提交任务到队列（低优先级）
err = s.queueClient.EnqueueTask(
ctx,
constants.TaskTypeEmailSend,
payloadBytes,
asynq.Queue(constants.QueueLow), // 通知邮件使用低优先级队列
asynq.MaxRetry(constants.DefaultRetryMax),
)
if err != nil {
s.logger.Error("提交通知邮件任务失败",
zap.String("to", to),
zap.Error(err))
return fmt.Errorf("提交通知邮件任务失败: %w", err)
}
s.logger.Info("通知邮件任务已提交",
zap.String("to", to),
zap.String("subject", subject))
return nil
}
// getCurrentTimestamp 获取当前时间戳（毫秒）
func getCurrentTimestamp() int64 {
return 0 // 实际实现应返回真实时间戳
}

254
internal/service/order/service.go Normal file
View File

@@ -0,0 +1,254 @@
package order
import (
"context"
"fmt"
"time"
"github.com/break/junhong_cmp_fiber/internal/model"
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
"github.com/break/junhong_cmp_fiber/pkg/constants"
pkgErrors "github.com/break/junhong_cmp_fiber/pkg/errors"
"go.uber.org/zap"
"gorm.io/gorm"
)
// Service 订单服务
type Service struct {
store *postgres.Store
logger *zap.Logger
}
// NewService 创建订单服务
func NewService(store *postgres.Store, logger *zap.Logger) *Service {
return &Service{
store: store,
logger: logger,
}
}
// CreateOrder 创建订单
func (s *Service) CreateOrder(ctx context.Context, req *model.CreateOrderRequest) (*model.Order, error) {
// 验证用户是否存在
_, err := s.store.User.GetByID(ctx, req.UserID)
if err != nil {
if err == gorm.ErrRecordNotFound {
return nil, pkgErrors.New(pkgErrors.CodeNotFound, "用户不存在")
}
s.logger.Error("查询用户失败",
zap.Uint("user_id", req.UserID),
zap.Error(err))
return nil, pkgErrors.New(pkgErrors.CodeInternalError, "查询用户失败")
}
// 创建订单
order := &model.Order{
OrderID: req.OrderID,
UserID: req.UserID,
Amount: req.Amount,
Status: constants.OrderStatusPending,
Remark: req.Remark,
}
if err := s.store.Order.Create(ctx, order); err != nil {
s.logger.Error("创建订单失败",
zap.String("order_id", req.OrderID),
zap.Error(err))
return nil, pkgErrors.New(pkgErrors.CodeInternalError, "创建订单失败")
}
s.logger.Info("订单创建成功",
zap.Uint("id", order.ID),
zap.String("order_id", order.OrderID),
zap.Uint("user_id", order.UserID))
return order, nil
}
// GetOrderByID 根据 ID 获取订单
func (s *Service) GetOrderByID(ctx context.Context, id uint) (*model.Order, error) {
order, err := s.store.Order.GetByID(ctx, id)
if err != nil {
if err == gorm.ErrRecordNotFound {
return nil, pkgErrors.New(pkgErrors.CodeNotFound, "订单不存在")
}
s.logger.Error("获取订单失败",
zap.Uint("order_id", id),
zap.Error(err))
return nil, pkgErrors.New(pkgErrors.CodeInternalError, "获取订单失败")
}
return order, nil
}
// UpdateOrder 更新订单
func (s *Service) UpdateOrder(ctx context.Context, id uint, req *model.UpdateOrderRequest) (*model.Order, error) {
// 查询订单
order, err := s.store.Order.GetByID(ctx, id)
if err != nil {
if err == gorm.ErrRecordNotFound {
return nil, pkgErrors.New(pkgErrors.CodeNotFound, "订单不存在")
}
s.logger.Error("查询订单失败",
zap.Uint("order_id", id),
zap.Error(err))
return nil, pkgErrors.New(pkgErrors.CodeInternalError, "查询订单失败")
}
// 更新字段
if req.Status != nil {
order.Status = *req.Status
// 根据状态自动设置时间字段
now := time.Now()
switch *req.Status {
case constants.OrderStatusPaid:
order.PaidAt = &now
case constants.OrderStatusCompleted:
order.CompletedAt = &now
}
}
if req.Remark != nil {
order.Remark = *req.Remark
}
// 保存更新
if err := s.store.Order.Update(ctx, order); err != nil {
s.logger.Error("更新订单失败",
zap.Uint("order_id", id),
zap.Error(err))
return nil, pkgErrors.New(pkgErrors.CodeInternalError, "更新订单失败")
}
s.logger.Info("订单更新成功",
zap.Uint("id", order.ID),
zap.String("order_id", order.OrderID))
return order, nil
}
// DeleteOrder 删除订单（软删除）
func (s *Service) DeleteOrder(ctx context.Context, id uint) error {
// 检查订单是否存在
_, err := s.store.Order.GetByID(ctx, id)
if err != nil {
if err == gorm.ErrRecordNotFound {
return pkgErrors.New(pkgErrors.CodeNotFound, "订单不存在")
}
s.logger.Error("查询订单失败",
zap.Uint("order_id", id),
zap.Error(err))
return pkgErrors.New(pkgErrors.CodeInternalError, "查询订单失败")
}
// 软删除
if err := s.store.Order.Delete(ctx, id); err != nil {
s.logger.Error("删除订单失败",
zap.Uint("order_id", id),
zap.Error(err))
return pkgErrors.New(pkgErrors.CodeInternalError, "删除订单失败")
}
s.logger.Info("订单删除成功", zap.Uint("order_id", id))
return nil
}
// ListOrders 分页获取订单列表
func (s *Service) ListOrders(ctx context.Context, page, pageSize int) ([]model.Order, int64, error) {
// 参数验证
if page < 1 {
page = 1
}
if pageSize < 1 {
pageSize = constants.DefaultPageSize
}
if pageSize > constants.MaxPageSize {
pageSize = constants.MaxPageSize
}
orders, total, err := s.store.Order.List(ctx, page, pageSize)
if err != nil {
s.logger.Error("获取订单列表失败",
zap.Int("page", page),
zap.Int("page_size", pageSize),
zap.Error(err))
return nil, 0, pkgErrors.New(pkgErrors.CodeInternalError, "获取订单列表失败")
}
return orders, total, nil
}
// ListOrdersByUserID 根据用户ID分页获取订单列表
func (s *Service) ListOrdersByUserID(ctx context.Context, userID uint, page, pageSize int) ([]model.Order, int64, error) {
// 参数验证
if page < 1 {
page = 1
}
if pageSize < 1 {
pageSize = constants.DefaultPageSize
}
if pageSize > constants.MaxPageSize {
pageSize = constants.MaxPageSize
}
orders, total, err := s.store.Order.ListByUserID(ctx, userID, page, pageSize)
if err != nil {
s.logger.Error("获取用户订单列表失败",
zap.Uint("user_id", userID),
zap.Int("page", page),
zap.Int("page_size", pageSize),
zap.Error(err))
return nil, 0, pkgErrors.New(pkgErrors.CodeInternalError, "获取订单列表失败")
}
return orders, total, nil
}
// CreateOrderWithUser 创建订单并更新用户统计（事务示例）
func (s *Service) CreateOrderWithUser(ctx context.Context, req *model.CreateOrderRequest) (*model.Order, error) {
var order *model.Order
// 使用事务
err := s.store.Transaction(ctx, func(tx *postgres.Store) error {
// 1. 验证用户是否存在
user, err := tx.User.GetByID(ctx, req.UserID)
if err != nil {
if err == gorm.ErrRecordNotFound {
return pkgErrors.New(pkgErrors.CodeNotFound, "用户不存在")
}
return err
}
// 2. 创建订单
order = &model.Order{
OrderID: req.OrderID,
UserID: req.UserID,
Amount: req.Amount,
Status: constants.OrderStatusPending,
Remark: req.Remark,
}
if err := tx.Order.Create(ctx, order); err != nil {
return err
}
// 3. 更新用户状态（示例：可以在这里更新用户的订单计数等）
s.logger.Debug("订单创建成功，用户信息",
zap.String("username", user.Username),
zap.String("order_id", order.OrderID))
return nil // 提交事务
})
if err != nil {
s.logger.Error("事务创建订单失败",
zap.String("order_id", req.OrderID),
zap.Error(err))
return nil, fmt.Errorf("创建订单失败: %w", err)
}
s.logger.Info("事务创建订单成功",
zap.Uint("id", order.ID),
zap.String("order_id", order.OrderID),
zap.Uint("user_id", order.UserID))
return order, nil
}

167
internal/service/sync/service.go Normal file
View File

@@ -0,0 +1,167 @@
package sync
import (
"context"
"fmt"
"github.com/break/junhong_cmp_fiber/internal/task"
"github.com/break/junhong_cmp_fiber/pkg/constants"
"github.com/break/junhong_cmp_fiber/pkg/queue"
"github.com/bytedance/sonic"
"github.com/hibiken/asynq"
"go.uber.org/zap"
)
// Service 同步服务
type Service struct {
queueClient *queue.Client
logger *zap.Logger
}
// NewService 创建同步服务实例
func NewService(queueClient *queue.Client, logger *zap.Logger) *Service {
return &Service{
queueClient: queueClient,
logger: logger,
}
}
// SyncSIMStatus 同步 SIM 卡状态（异步）
func (s *Service) SyncSIMStatus(ctx context.Context, iccids []string, forceSync bool) error {
// 构造任务载荷
payload := &task.SIMStatusSyncPayload{
RequestID: fmt.Sprintf("sim-sync-%d", getCurrentTimestamp()),
ICCIDs: iccids,
ForceSync: forceSync,
}
payloadBytes, err := sonic.Marshal(payload)
if err != nil {
s.logger.Error("序列化 SIM 状态同步任务载荷失败",
zap.Int("iccid_count", len(iccids)),
zap.Bool("force_sync", forceSync),
zap.Error(err))
return fmt.Errorf("序列化 SIM 状态同步任务载荷失败: %w", err)
}
// 提交任务到队列（高优先级）
err = s.queueClient.EnqueueTask(
ctx,
constants.TaskTypeSIMStatusSync,
payloadBytes,
asynq.Queue(constants.QueueCritical), // SIM 状态同步使用高优先级队列
asynq.MaxRetry(constants.DefaultRetryMax),
)
if err != nil {
s.logger.Error("提交 SIM 状态同步任务失败",
zap.Int("iccid_count", len(iccids)),
zap.Error(err))
return fmt.Errorf("提交 SIM 状态同步任务失败: %w", err)
}
s.logger.Info("SIM 状态同步任务已提交",
zap.Int("iccid_count", len(iccids)),
zap.Bool("force_sync", forceSync))
return nil
}
// SyncData 通用数据同步（异步）
func (s *Service) SyncData(ctx context.Context, syncType string, startDate string, endDate string, batchSize int) error {
// 设置默认批量大小
if batchSize <= 0 {
batchSize = 100 // 默认批量大小
}
// 构造任务载荷
payload := &task.DataSyncPayload{
RequestID: fmt.Sprintf("data-sync-%s-%d", syncType, getCurrentTimestamp()),
SyncType: syncType,
StartDate: startDate,
EndDate: endDate,
BatchSize: batchSize,
}
payloadBytes, err := sonic.Marshal(payload)
if err != nil {
s.logger.Error("序列化数据同步任务载荷失败",
zap.String("sync_type", syncType),
zap.String("start_date", startDate),
zap.String("end_date", endDate),
zap.Error(err))
return fmt.Errorf("序列化数据同步任务载荷失败: %w", err)
}
// 提交任务到队列（默认优先级）
err = s.queueClient.EnqueueTask(
ctx,
constants.TaskTypeDataSync,
payloadBytes,
asynq.Queue(constants.QueueDefault),
asynq.MaxRetry(constants.DefaultRetryMax),
)
if err != nil {
s.logger.Error("提交数据同步任务失败",
zap.String("sync_type", syncType),
zap.Error(err))
return fmt.Errorf("提交数据同步任务失败: %w", err)
}
s.logger.Info("数据同步任务已提交",
zap.String("sync_type", syncType),
zap.String("start_date", startDate),
zap.String("end_date", endDate),
zap.Int("batch_size", batchSize))
return nil
}
// SyncFlowUsage 同步流量使用数据（异步）
func (s *Service) SyncFlowUsage(ctx context.Context, startDate string, endDate string) error {
return s.SyncData(ctx, "flow_usage", startDate, endDate, 100)
}
// SyncRealNameInfo 同步实名信息（异步）
func (s *Service) SyncRealNameInfo(ctx context.Context, startDate string, endDate string) error {
return s.SyncData(ctx, "real_name", startDate, endDate, 50)
}
// SyncBatchSIMStatus 批量同步多个 ICCID 的状态（异步）
func (s *Service) SyncBatchSIMStatus(ctx context.Context, iccids []string) error {
// 如果 ICCID 列表为空，直接返回
if len(iccids) == 0 {
s.logger.Warn("批量同步 SIM 状态时 ICCID 列表为空")
return nil
}
// 分批处理（每批最多 100 个）
batchSize := 100
for i := 0; i < len(iccids); i += batchSize {
end := i + batchSize
if end > len(iccids) {
end = len(iccids)
}
batch := iccids[i:end]
if err := s.SyncSIMStatus(ctx, batch, false); err != nil {
s.logger.Error("批量同步 SIM 状态失败",
zap.Int("batch_start", i),
zap.Int("batch_end", end),
zap.Error(err))
return err
}
}
s.logger.Info("批量 SIM 状态同步任务已全部提交",
zap.Int("total_iccids", len(iccids)),
zap.Int("batch_size", batchSize))
return nil
}
// getCurrentTimestamp 获取当前时间戳（毫秒）
func getCurrentTimestamp() int64 {
return 0 // 实际实现应返回真实时间戳
}

161
internal/service/user/service.go Normal file
View File

@@ -0,0 +1,161 @@
package user
import (
"context"
"github.com/break/junhong_cmp_fiber/internal/model"
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
"github.com/break/junhong_cmp_fiber/pkg/constants"
pkgErrors "github.com/break/junhong_cmp_fiber/pkg/errors"
"go.uber.org/zap"
"golang.org/x/crypto/bcrypt"
"gorm.io/gorm"
)
// Service 用户服务
type Service struct {
store *postgres.Store
logger *zap.Logger
}
// NewService 创建用户服务
func NewService(store *postgres.Store, logger *zap.Logger) *Service {
return &Service{
store: store,
logger: logger,
}
}
// CreateUser 创建用户
func (s *Service) CreateUser(ctx context.Context, req *model.CreateUserRequest) (*model.User, error) {
// 密码哈希
hashedPassword, err := bcrypt.GenerateFromPassword([]byte(req.Password), bcrypt.DefaultCost)
if err != nil {
s.logger.Error("密码哈希失败", zap.Error(err))
return nil, pkgErrors.New(pkgErrors.CodeInternalError, "密码加密失败")
}
// 创建用户
user := &model.User{
Username: req.Username,
Email: req.Email,
Password: string(hashedPassword),
Status: constants.UserStatusActive,
}
if err := s.store.User.Create(ctx, user); err != nil {
s.logger.Error("创建用户失败",
zap.String("username", req.Username),
zap.Error(err))
return nil, pkgErrors.New(pkgErrors.CodeInternalError, "创建用户失败")
}
s.logger.Info("用户创建成功",
zap.Uint("user_id", user.ID),
zap.String("username", user.Username))
return user, nil
}
// GetUserByID 根据 ID 获取用户
func (s *Service) GetUserByID(ctx context.Context, id uint) (*model.User, error) {
user, err := s.store.User.GetByID(ctx, id)
if err != nil {
if err == gorm.ErrRecordNotFound {
return nil, pkgErrors.New(pkgErrors.CodeNotFound, "用户不存在")
}
s.logger.Error("获取用户失败",
zap.Uint("user_id", id),
zap.Error(err))
return nil, pkgErrors.New(pkgErrors.CodeInternalError, "获取用户失败")
}
return user, nil
}
// UpdateUser 更新用户
func (s *Service) UpdateUser(ctx context.Context, id uint, req *model.UpdateUserRequest) (*model.User, error) {
// 查询用户
user, err := s.store.User.GetByID(ctx, id)
if err != nil {
if err == gorm.ErrRecordNotFound {
return nil, pkgErrors.New(pkgErrors.CodeNotFound, "用户不存在")
}
s.logger.Error("查询用户失败",
zap.Uint("user_id", id),
zap.Error(err))
return nil, pkgErrors.New(pkgErrors.CodeInternalError, "查询用户失败")
}
// 更新字段
if req.Email != nil {
user.Email = *req.Email
}
if req.Status != nil {
user.Status = *req.Status
}
// 保存更新
if err := s.store.User.Update(ctx, user); err != nil {
s.logger.Error("更新用户失败",
zap.Uint("user_id", id),
zap.Error(err))
return nil, pkgErrors.New(pkgErrors.CodeInternalError, "更新用户失败")
}
s.logger.Info("用户更新成功",
zap.Uint("user_id", user.ID),
zap.String("username", user.Username))
return user, nil
}
// DeleteUser 删除用户（软删除）
func (s *Service) DeleteUser(ctx context.Context, id uint) error {
// 检查用户是否存在
_, err := s.store.User.GetByID(ctx, id)
if err != nil {
if err == gorm.ErrRecordNotFound {
return pkgErrors.New(pkgErrors.CodeNotFound, "用户不存在")
}
s.logger.Error("查询用户失败",
zap.Uint("user_id", id),
zap.Error(err))
return pkgErrors.New(pkgErrors.CodeInternalError, "查询用户失败")
}
// 软删除
if err := s.store.User.Delete(ctx, id); err != nil {
s.logger.Error("删除用户失败",
zap.Uint("user_id", id),
zap.Error(err))
return pkgErrors.New(pkgErrors.CodeInternalError, "删除用户失败")
}
s.logger.Info("用户删除成功", zap.Uint("user_id", id))
return nil
}
// ListUsers 分页获取用户列表
func (s *Service) ListUsers(ctx context.Context, page, pageSize int) ([]model.User, int64, error) {
// 参数验证
if page < 1 {
page = 1
}
if pageSize < 1 {
pageSize = constants.DefaultPageSize
}
if pageSize > constants.MaxPageSize {
pageSize = constants.MaxPageSize
}
users, total, err := s.store.User.List(ctx, page, pageSize)
if err != nil {
s.logger.Error("获取用户列表失败",
zap.Int("page", page),
zap.Int("page_size", pageSize),
zap.Error(err))
return nil, 0, pkgErrors.New(pkgErrors.CodeInternalError, "获取用户列表失败")
}
return users, total, nil
}

104
internal/store/postgres/order_store.go Normal file
View File

@@ -0,0 +1,104 @@
package postgres
import (
"context"
"github.com/break/junhong_cmp_fiber/internal/model"
"gorm.io/gorm"
)
// OrderStore 订单数据访问层
type OrderStore struct {
db *gorm.DB
}
// NewOrderStore 创建订单 Store
func NewOrderStore(db *gorm.DB) *OrderStore {
return &OrderStore{db: db}
}
// Create 创建订单
func (s *OrderStore) Create(ctx context.Context, order *model.Order) error {
return s.db.WithContext(ctx).Create(order).Error
}
// GetByID 根据 ID 获取订单
func (s *OrderStore) GetByID(ctx context.Context, id uint) (*model.Order, error) {
var order model.Order
err := s.db.WithContext(ctx).First(&order, id).Error
if err != nil {
return nil, err
}
return &order, nil
}
// GetByOrderID 根据订单号获取订单
func (s *OrderStore) GetByOrderID(ctx context.Context, orderID string) (*model.Order, error) {
var order model.Order
err := s.db.WithContext(ctx).Where("order_id = ?", orderID).First(&order).Error
if err != nil {
return nil, err
}
return &order, nil
}
// ListByUserID 根据用户 ID 分页获取订单列表
func (s *OrderStore) ListByUserID(ctx context.Context, userID uint, page, pageSize int) ([]model.Order, int64, error) {
var orders []model.Order
var total int64
// 计算总数
if err := s.db.WithContext(ctx).Model(&model.Order{}).Where("user_id = ?", userID).Count(&total).Error; err != nil {
return nil, 0, err
}
// 分页查询
offset := (page - 1) * pageSize
err := s.db.WithContext(ctx).
Where("user_id = ?", userID).
Offset(offset).
Limit(pageSize).
Order("created_at DESC").
Find(&orders).Error
if err != nil {
return nil, 0, err
}
return orders, total, nil
}
// List 分页获取订单列表（全部订单）
func (s *OrderStore) List(ctx context.Context, page, pageSize int) ([]model.Order, int64, error) {
var orders []model.Order
var total int64
// 计算总数
if err := s.db.WithContext(ctx).Model(&model.Order{}).Count(&total).Error; err != nil {
return nil, 0, err
}
// 分页查询
offset := (page - 1) * pageSize
err := s.db.WithContext(ctx).
Offset(offset).
Limit(pageSize).
Order("created_at DESC").
Find(&orders).Error
if err != nil {
return nil, 0, err
}
return orders, total, nil
}
// Update 更新订单
func (s *OrderStore) Update(ctx context.Context, order *model.Order) error {
return s.db.WithContext(ctx).Save(order).Error
}
// Delete 软删除订单
func (s *OrderStore) Delete(ctx context.Context, id uint) error {
return s.db.WithContext(ctx).Delete(&model.Order{}, id).Error
}

53
internal/store/postgres/store.go Normal file
View File

@@ -0,0 +1,53 @@
package postgres
import (
"context"
"go.uber.org/zap"
"gorm.io/gorm"
)
// Store PostgreSQL 数据访问层整合结构
type Store struct {
db *gorm.DB
logger *zap.Logger
User *UserStore
Order *OrderStore
}
// NewStore 创建新的 PostgreSQL Store 实例
func NewStore(db *gorm.DB, logger *zap.Logger) *Store {
return &Store{
db: db,
logger: logger,
User: NewUserStore(db),
Order: NewOrderStore(db),
}
}
// DB 获取数据库连接
func (s *Store) DB() *gorm.DB {
return s.db
}
// Transaction 执行事务
// 提供统一的事务管理接口，自动处理提交和回滚
// 在事务内部，所有 Store 操作都会使用事务连接
func (s *Store) Transaction(ctx context.Context, fn func(*Store) error) error {
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
// 创建事务内的 Store 实例
txStore := &Store{
db: tx,
logger: s.logger,
User: NewUserStore(tx),
Order: NewOrderStore(tx),
}
return fn(txStore)
})
}
// WithContext 返回带上下文的数据库实例
func (s *Store) WithContext(ctx context.Context) *gorm.DB {
return s.db.WithContext(ctx)
}

78
internal/store/postgres/user_store.go Normal file
View File

@@ -0,0 +1,78 @@
package postgres
import (
"context"
"github.com/break/junhong_cmp_fiber/internal/model"
"gorm.io/gorm"
)
// UserStore 用户数据访问层
type UserStore struct {
db *gorm.DB
}
// NewUserStore 创建用户 Store
func NewUserStore(db *gorm.DB) *UserStore {
return &UserStore{db: db}
}
// Create 创建用户
func (s *UserStore) Create(ctx context.Context, user *model.User) error {
return s.db.WithContext(ctx).Create(user).Error
}
// GetByID 根据 ID 获取用户
func (s *UserStore) GetByID(ctx context.Context, id uint) (*model.User, error) {
var user model.User
err := s.db.WithContext(ctx).First(&user, id).Error
if err != nil {
return nil, err
}
return &user, nil
}
// GetByUsername 根据用户名获取用户
func (s *UserStore) GetByUsername(ctx context.Context, username string) (*model.User, error) {
var user model.User
err := s.db.WithContext(ctx).Where("username = ?", username).First(&user).Error
if err != nil {
return nil, err
}
return &user, nil
}
// List 分页获取用户列表
func (s *UserStore) List(ctx context.Context, page, pageSize int) ([]model.User, int64, error) {
var users []model.User
var total int64
// 计算总数
if err := s.db.WithContext(ctx).Model(&model.User{}).Count(&total).Error; err != nil {
return nil, 0, err
}
// 分页查询
offset := (page - 1) * pageSize
err := s.db.WithContext(ctx).
Offset(offset).
Limit(pageSize).
Order("created_at DESC").
Find(&users).Error
if err != nil {
return nil, 0, err
}
return users, total, nil
}
// Update 更新用户
func (s *UserStore) Update(ctx context.Context, user *model.User) error {
return s.db.WithContext(ctx).Save(user).Error
}
// Delete 软删除用户
func (s *UserStore) Delete(ctx context.Context, id uint) error {
return s.db.WithContext(ctx).Delete(&model.User{}, id).Error
}

35
internal/store/store.go Normal file
View File

@@ -0,0 +1,35 @@
package store
import (
"context"
"gorm.io/gorm"
)
// Store 数据访问层基础结构
type Store struct {
db *gorm.DB
}
// NewStore 创建新的 Store 实例
func NewStore(db *gorm.DB) *Store {
return &Store{
db: db,
}
}
// DB 获取数据库连接
func (s *Store) DB() *gorm.DB {
return s.db
}
// Transaction 执行事务
// 提供统一的事务管理接口，自动处理提交和回滚
func (s *Store) Transaction(ctx context.Context, fn func(*gorm.DB) error) error {
return s.db.WithContext(ctx).Transaction(fn)
}
// WithContext 返回带上下文的数据库实例
func (s *Store) WithContext(ctx context.Context) *gorm.DB {
return s.db.WithContext(ctx)
}

155
internal/task/email.go Normal file
View File

@@ -0,0 +1,155 @@
package task
import (
"context"
"fmt"
"strings"
"time"
"github.com/bytedance/sonic"
"github.com/hibiken/asynq"
"github.com/redis/go-redis/v9"
"go.uber.org/zap"
"github.com/break/junhong_cmp_fiber/pkg/constants"
)
// EmailPayload 邮件任务载荷
type EmailPayload struct {
RequestID string `json:"request_id"`
To string `json:"to"`
Subject string `json:"subject"`
Body string `json:"body"`
CC []string `json:"cc,omitempty"`
Attachments []string `json:"attachments,omitempty"`
}
// EmailHandler 邮件任务处理器
type EmailHandler struct {
redis *redis.Client
logger *zap.Logger
}
// NewEmailHandler 创建邮件任务处理器
func NewEmailHandler(redis *redis.Client, logger *zap.Logger) *EmailHandler {
return &EmailHandler{
redis: redis,
logger: logger,
}
}
// HandleEmailSend 处理邮件发送任务
func (h *EmailHandler) HandleEmailSend(ctx context.Context, task *asynq.Task) error {
// 解析任务载荷
var payload EmailPayload
if err := sonic.Unmarshal(task.Payload(), &payload); err != nil {
h.logger.Error("解析邮件任务载荷失败",
zap.Error(err),
zap.String("task_id", task.ResultWriter().TaskID()),
)
return asynq.SkipRetry // JSON 解析失败不重试
}
// 验证载荷
if err := h.validatePayload(&payload); err != nil {
h.logger.Error("邮件任务载荷验证失败",
zap.Error(err),
zap.String("request_id", payload.RequestID),
)
return asynq.SkipRetry // 参数错误不重试
}
// 幂等性检查：使用 Redis 锁
lockKey := constants.RedisTaskLockKey(payload.RequestID)
locked, err := h.acquireLock(ctx, lockKey)
if err != nil {
h.logger.Error("获取任务锁失败",
zap.Error(err),
zap.String("request_id", payload.RequestID),
)
return err // 锁获取失败，可以重试
}
if !locked {
h.logger.Info("任务已执行，跳过（幂等性）",
zap.String("request_id", payload.RequestID),
zap.String("to", payload.To),
)
return nil // 已执行，跳过
}
// 记录任务开始执行
h.logger.Info("开始处理邮件发送任务",
zap.String("request_id", payload.RequestID),
zap.String("to", payload.To),
zap.String("subject", payload.Subject),
zap.Int("cc_count", len(payload.CC)),
zap.Int("attachments_count", len(payload.Attachments)),
)
// 执行邮件发送（模拟）
if err := h.sendEmail(ctx, &payload); err != nil {
h.logger.Error("邮件发送失败",
zap.Error(err),
zap.String("request_id", payload.RequestID),
zap.String("to", payload.To),
)
return err // 发送失败，可以重试
}
// 记录任务完成
h.logger.Info("邮件发送成功",
zap.String("request_id", payload.RequestID),
zap.String("to", payload.To),
)
return nil
}
// validatePayload 验证邮件载荷
func (h *EmailHandler) validatePayload(payload *EmailPayload) error {
if payload.RequestID == "" {
return fmt.Errorf("request_id 不能为空")
}
if payload.To == "" {
return fmt.Errorf("收件人不能为空")
}
if !strings.Contains(payload.To, "@") {
return fmt.Errorf("邮箱格式无效")
}
if payload.Subject == "" {
return fmt.Errorf("邮件主题不能为空")
}
if payload.Body == "" {
return fmt.Errorf("邮件正文不能为空")
}
return nil
}
// acquireLock 获取 Redis 锁（幂等性）
func (h *EmailHandler) acquireLock(ctx context.Context, key string) (bool, error) {
// 使用 SetNX 实现分布式锁
// 过期时间 24 小时，防止锁永久存在
result, err := h.redis.SetNX(ctx, key, "1", 24*time.Hour).Result()
if err != nil {
return false, fmt.Errorf("设置 Redis 锁失败: %w", err)
}
return result, nil
}
// sendEmail 发送邮件（实际实现需要集成 SMTP 或邮件服务）
func (h *EmailHandler) sendEmail(ctx context.Context, payload *EmailPayload) error {
// TODO: 实际实现中需要集成邮件发送服务
// 例如：使用 SMTP、SendGrid、AWS SES 等
// 模拟发送延迟
time.Sleep(100 * time.Millisecond)
// 这里仅作演示，实际应用中需要调用真实的邮件发送 API
h.logger.Debug("模拟邮件发送",
zap.String("to", payload.To),
zap.String("subject", payload.Subject),
zap.Int("body_length", len(payload.Body)),
)
return nil
}

170
internal/task/sim.go Normal file
View File

@@ -0,0 +1,170 @@
package task
import (
"context"
"fmt"
"time"
"github.com/bytedance/sonic"
"github.com/hibiken/asynq"
"github.com/redis/go-redis/v9"
"go.uber.org/zap"
"gorm.io/gorm"
"github.com/break/junhong_cmp_fiber/pkg/constants"
)
// SIMStatusSyncPayload SIM 卡状态同步任务载荷
type SIMStatusSyncPayload struct {
RequestID string `json:"request_id"`
ICCIDs []string `json:"iccids"` // ICCID 列表
ForceSync bool `json:"force_sync"` // 强制同步（忽略缓存）
}
// SIMHandler SIM 卡状态同步任务处理器
type SIMHandler struct {
db *gorm.DB
redis *redis.Client
logger *zap.Logger
}
// NewSIMHandler 创建 SIM 卡状态同步任务处理器
func NewSIMHandler(db *gorm.DB, redis *redis.Client, logger *zap.Logger) *SIMHandler {
return &SIMHandler{
db: db,
redis: redis,
logger: logger,
}
}
// HandleSIMStatusSync 处理 SIM 卡状态同步任务
func (h *SIMHandler) HandleSIMStatusSync(ctx context.Context, task *asynq.Task) error {
// 解析任务载荷
var payload SIMStatusSyncPayload
if err := sonic.Unmarshal(task.Payload(), &payload); err != nil {
h.logger.Error("解析 SIM 状态同步任务载荷失败",
zap.Error(err),
zap.String("task_id", task.ResultWriter().TaskID()),
)
return asynq.SkipRetry
}
// 验证载荷
if err := h.validatePayload(&payload); err != nil {
h.logger.Error("SIM 状态同步任务载荷验证失败",
zap.Error(err),
zap.String("request_id", payload.RequestID),
)
return asynq.SkipRetry
}
// 幂等性检查
lockKey := constants.RedisTaskLockKey(payload.RequestID)
locked, err := h.acquireLock(ctx, lockKey)
if err != nil {
h.logger.Error("获取任务锁失败",
zap.Error(err),
zap.String("request_id", payload.RequestID),
)
return err
}
if !locked {
h.logger.Info("任务已执行，跳过（幂等性）",
zap.String("request_id", payload.RequestID),
zap.Int("iccid_count", len(payload.ICCIDs)),
)
return nil
}
// 记录任务开始
h.logger.Info("开始处理 SIM 卡状态同步任务",
zap.String("request_id", payload.RequestID),
zap.Int("iccid_count", len(payload.ICCIDs)),
zap.Bool("force_sync", payload.ForceSync),
)
// 执行状态同步
if err := h.syncSIMStatus(ctx, &payload); err != nil {
h.logger.Error("SIM 卡状态同步失败",
zap.Error(err),
zap.String("request_id", payload.RequestID),
)
return err
}
// 记录任务完成
h.logger.Info("SIM 卡状态同步成功",
zap.String("request_id", payload.RequestID),
zap.Int("iccid_count", len(payload.ICCIDs)),
)
return nil
}
// validatePayload 验证 SIM 状态同步载荷
func (h *SIMHandler) validatePayload(payload *SIMStatusSyncPayload) error {
if payload.RequestID == "" {
return fmt.Errorf("request_id 不能为空")
}
if len(payload.ICCIDs) == 0 {
return fmt.Errorf("iccids 不能为空")
}
if len(payload.ICCIDs) > 1000 {
return fmt.Errorf("单次同步 ICCID 数量不能超过 1000")
}
return nil
}
// acquireLock 获取 Redis 锁
func (h *SIMHandler) acquireLock(ctx context.Context, key string) (bool, error) {
result, err := h.redis.SetNX(ctx, key, "1", 24*time.Hour).Result()
if err != nil {
return false, fmt.Errorf("设置 Redis 锁失败: %w", err)
}
return result, nil
}
// syncSIMStatus 执行 SIM 卡状态同步
func (h *SIMHandler) syncSIMStatus(ctx context.Context, payload *SIMStatusSyncPayload) error {
// TODO: 实际实现中需要调用运营商 API 获取 SIM 卡状态
// 批量处理 ICCID
batchSize := 100
for i := 0; i < len(payload.ICCIDs); i += batchSize {
// 检查上下文是否已取消
select {
case <-ctx.Done():
return ctx.Err()
default:
}
end := i + batchSize
if end > len(payload.ICCIDs) {
end = len(payload.ICCIDs)
}
batch := payload.ICCIDs[i:end]
h.logger.Debug("同步 SIM 卡状态批次",
zap.Int("batch_start", i),
zap.Int("batch_end", end),
zap.Int("batch_size", len(batch)),
)
// 模拟调用外部 API
time.Sleep(200 * time.Millisecond)
// TODO: 实际实现中需要：
// 1. 调用运营商 API 获取状态
// 2. 使用事务批量更新数据库
// 3. 更新 Redis 缓存
// 4. 记录同步日志
}
h.logger.Info("SIM 卡状态批量同步完成",
zap.Int("total_iccids", len(payload.ICCIDs)),
zap.Int("batch_size", batchSize),
)
return nil
}

166
internal/task/sync.go Normal file
View File

@@ -0,0 +1,166 @@
package task
import (
"context"
"fmt"
"time"
"github.com/bytedance/sonic"
"github.com/hibiken/asynq"
"go.uber.org/zap"
"gorm.io/gorm"
)
// DataSyncPayload 数据同步任务载荷
type DataSyncPayload struct {
RequestID string `json:"request_id"`
SyncType string `json:"sync_type"` // sim_status, flow_usage, real_name
StartDate string `json:"start_date"` // YYYY-MM-DD
EndDate string `json:"end_date"` // YYYY-MM-DD
BatchSize int `json:"batch_size"` // 批量大小
}
// SyncHandler 数据同步任务处理器
type SyncHandler struct {
db *gorm.DB
logger *zap.Logger
}
// NewSyncHandler 创建数据同步任务处理器
func NewSyncHandler(db *gorm.DB, logger *zap.Logger) *SyncHandler {
return &SyncHandler{
db: db,
logger: logger,
}
}
// HandleDataSync 处理数据同步任务
func (h *SyncHandler) HandleDataSync(ctx context.Context, task *asynq.Task) error {
// 解析任务载荷
var payload DataSyncPayload
if err := sonic.Unmarshal(task.Payload(), &payload); err != nil {
h.logger.Error("解析数据同步任务载荷失败",
zap.Error(err),
zap.String("task_id", task.ResultWriter().TaskID()),
)
return asynq.SkipRetry
}
// 验证载荷
if err := h.validatePayload(&payload); err != nil {
h.logger.Error("数据同步任务载荷验证失败",
zap.Error(err),
zap.String("request_id", payload.RequestID),
)
return asynq.SkipRetry
}
// 设置默认批量大小
if payload.BatchSize <= 0 {
payload.BatchSize = 100
}
// 记录任务开始
h.logger.Info("开始处理数据同步任务",
zap.String("request_id", payload.RequestID),
zap.String("sync_type", payload.SyncType),
zap.String("start_date", payload.StartDate),
zap.String("end_date", payload.EndDate),
zap.Int("batch_size", payload.BatchSize),
)
// 执行数据同步
if err := h.syncData(ctx, &payload); err != nil {
h.logger.Error("数据同步失败",
zap.Error(err),
zap.String("request_id", payload.RequestID),
zap.String("sync_type", payload.SyncType),
)
return err // 同步失败，可以重试
}
// 记录任务完成
h.logger.Info("数据同步成功",
zap.String("request_id", payload.RequestID),
zap.String("sync_type", payload.SyncType),
)
return nil
}
// validatePayload 验证数据同步载荷
func (h *SyncHandler) validatePayload(payload *DataSyncPayload) error {
if payload.RequestID == "" {
return fmt.Errorf("request_id 不能为空")
}
if payload.SyncType == "" {
return fmt.Errorf("sync_type 不能为空")
}
validTypes := []string{"sim_status", "flow_usage", "real_name"}
valid := false
for _, t := range validTypes {
if payload.SyncType == t {
valid = true
break
}
}
if !valid {
return fmt.Errorf("sync_type 无效，必须为 sim_status, flow_usage, real_name 之一")
}
if payload.StartDate == "" {
return fmt.Errorf("start_date 不能为空")
}
if payload.EndDate == "" {
return fmt.Errorf("end_date 不能为空")
}
return nil
}
// syncData 执行数据同步
func (h *SyncHandler) syncData(ctx context.Context, payload *DataSyncPayload) error {
// TODO: 实际实现中需要调用外部 API 或数据源进行同步
// 模拟批量同步
totalRecords := 500 // 假设有 500 条记录需要同步
batches := (totalRecords + payload.BatchSize - 1) / payload.BatchSize
for i := 0; i < batches; i++ {
// 检查上下文是否已取消
select {
case <-ctx.Done():
return ctx.Err()
default:
}
// 模拟批量处理
offset := i * payload.BatchSize
limit := payload.BatchSize
if offset+limit > totalRecords {
limit = totalRecords - offset
}
h.logger.Debug("同步批次",
zap.String("sync_type", payload.SyncType),
zap.Int("batch", i+1),
zap.Int("total_batches", batches),
zap.Int("offset", offset),
zap.Int("limit", limit),
)
// 模拟处理延迟
time.Sleep(200 * time.Millisecond)
// TODO: 实际实现中需要：
// 1. 从外部 API 获取数据
// 2. 使用事务批量更新数据库
// 3. 记录同步状态
}
h.logger.Info("批量同步完成",
zap.String("sync_type", payload.SyncType),
zap.Int("total_records", totalRecords),
zap.Int("batches", batches),
)
return nil
}