From 60debb75053901f4f837ac6744d8dda681dc9481 Mon Sep 17 00:00:00 2001 From: huang Date: Mon, 27 Apr 2026 09:47:03 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BB=9F=E4=B8=80=E5=AF=BC=E5=87=BA=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 15 + internal/bootstrap/handlers.go | 1 + internal/bootstrap/services.go | 3 + internal/bootstrap/stores.go | 4 + internal/bootstrap/types.go | 1 + internal/bootstrap/worker_stores.go | 6 + internal/exporter/device_scene.go | 98 +++++ internal/exporter/iot_card_scene.go | 98 +++++ internal/exporter/registry.go | 83 +++++ internal/exporter/scope.go | 26 ++ internal/handler/admin/export_task.go | 97 +++++ internal/model/dto/export_task_dto.go | 89 +++++ internal/model/export_task.go | 106 ++++++ internal/routes/admin.go | 3 + internal/routes/export_task.go | 50 +++ internal/service/export_task/service.go | 304 ++++++++++++++++ .../store/postgres/export_shard_task_store.go | 154 ++++++++ internal/store/postgres/export_task_store.go | 337 ++++++++++++++++++ internal/task/export_common.go | 209 +++++++++++ internal/task/export_dispatch.go | 255 +++++++++++++ internal/task/export_finalize.go | 158 ++++++++ internal/task/export_shard.go | 183 ++++++++++ .../000135_create_export_task_tables.down.sql | 4 + .../000135_create_export_task_tables.up.sql | 149 ++++++++ pkg/constants/constants.go | 81 +++++ pkg/constants/redis.go | 21 ++ pkg/openapi/handlers.go | 1 + pkg/queue/handler.go | 42 +++ pkg/queue/types.go | 2 + 29 files changed, 2580 insertions(+) create mode 100644 internal/exporter/device_scene.go create mode 100644 internal/exporter/iot_card_scene.go create mode 100644 internal/exporter/registry.go create mode 100644 internal/exporter/scope.go create mode 100644 internal/handler/admin/export_task.go create mode 100644 internal/model/dto/export_task_dto.go create mode 100644 internal/model/export_task.go create mode 100644 internal/routes/export_task.go create mode 100644 internal/service/export_task/service.go create mode 100644 internal/store/postgres/export_shard_task_store.go create mode 100644 internal/store/postgres/export_task_store.go create mode 100644 internal/task/export_common.go create mode 100644 internal/task/export_dispatch.go create mode 100644 internal/task/export_finalize.go create mode 100644 internal/task/export_shard.go create mode 100644 migrations/000135_create_export_task_tables.down.sql create mode 100644 migrations/000135_create_export_task_tables.up.sql diff --git a/README.md b/README.md index 617f772..780c5e3 100644 --- a/README.md +++ b/README.md @@ -210,6 +210,7 @@ default: - **统一错误处理**：全局 ErrorHandler 统一处理所有 API 错误，返回一致的 JSON 格式（包含错误码、消息、时间戳）；Panic 自动恢复防止服务崩溃；错误分类处理（客户端 4xx、服务端 5xx）和日志级别控制；敏感信息自动脱敏保护 - **数据持久化**：GORM + PostgreSQL 集成，提供完整的 CRUD 操作、事务支持和数据库迁移能力 - **异步任务处理**：Asynq 任务队列集成，支持任务提交、后台执行、自动重试和幂等性保障，实现邮件发送、数据同步等异步任务 +- **统一导出任务系统**：新增全局导出任务入口（`/api/admin/export-tasks`），支持 `scene=device/iot_card`、`format=xlsx/csv`、异步分片执行、任务取消、详情直出 24 小时下载链接；详见 [功能总结](docs/unified-export-task-system/功能总结.md) 与 [验收记录](docs/unified-export-task-system/验收记录.md) - **RBAC 权限系统**：完整的基于角色的访问控制，支持账号、角色、权限的多对多关联和层级关系；基于店铺层级的自动数据权限过滤，实现多租户数据隔离；使用 PostgreSQL WITH RECURSIVE 查询下级店铺并通过 Redis 缓存优化性能；完整的权限检查功能支持路由级别的细粒度权限控制，支持平台过滤（web/h5/all）和超级管理员自动跳过（详见 [功能总结](docs/004-rbac-data-permission/功能总结.md)、[使用指南](docs/004-rbac-data-permission/使用指南.md) 和 [权限检查使用指南](docs/permission-check-usage.md)） - **商户管理**：完整的商户（Shop）和商户账号管理功能，支持商户创建时自动创建初始坐席账号、删除商户时批量禁用关联账号、账号密码重置等功能（详见 [使用指南](docs/shop-management/使用指南.md) 和 [API 文档](docs/shop-management/API文档.md)） - **B 端认证系统**：完整的后台和 H5 认证功能，支持基于 Redis 的 Token 管理和双令牌机制（Access Token 24h + Refresh Token 7天）；包含登录、登出、Token 刷新、用户信息查询和密码修改功能；通过用户类型隔离确保后台（SuperAdmin、Platform、Agent）和 H5（Agent、Enterprise）的访问控制；**登录响应包含菜单树和按钮权限**（menus/buttons），前端无需二次处理直接渲染侧边栏和控制按钮显示；详见 [API 文档](docs/api/auth.md)、[使用指南](docs/auth-usage-guide.md)、[架构说明](docs/auth-architecture.md) 和 [菜单权限使用指南](docs/login-menu-button-response/使用指南.md) @@ -224,6 +225,20 @@ default: - **微信集成**：完整的微信公众号 OAuth 认证和微信支付功能（JSAPI + H5），使用 PowerWeChat v3 SDK；支持个人客户微信授权登录、账号绑定、微信内支付和浏览器 H5 支付；支付回调自动验证签名和幂等性处理；详见 [使用指南](docs/wechat-integration/使用指南.md) 和 [API 文档](docs/wechat-integration/API文档.md) - **订单超时自动取消**：待支付订单（微信/支付宝）30 分钟超时自动取消，支持钱包余额解冻；使用 Asynq Scheduler 每分钟扫描，取代原有 time.Ticker 实现；同时将告警检查和数据清理迁移至 Asynq Scheduler 统一调度；详见 [功能总结](docs/order-expiration/功能总结.md) +### 导出任务接口示例 + +```bash +# 创建导出任务（scene 支持 device / iot_card） +curl -X POST '/api/admin/export-tasks' \\ + -H 'Content-Type: application/json' \\ + -H 'Authorization: Bearer ' \\ + -d '{\"scene\":\"device\",\"format\":\"xlsx\"}' + +# 查询导出任务列表（支持 scene/status/time 过滤） +curl '/api/admin/export-tasks?page=1&page_size=20&scene=device' \\ + -H 'Authorization: Bearer ' +``` + ## 用户体系设计 系统支持四种用户类型和两种组织实体,实现分层级的多租户管理： diff --git a/internal/bootstrap/handlers.go b/internal/bootstrap/handlers.go index 2df53d4..1071201 100644 --- a/internal/bootstrap/handlers.go +++ b/internal/bootstrap/handlers.go @@ -96,6 +96,7 @@ func initHandlers(svc *services, deps *Dependencies) *Handlers { Authorization: admin.NewAuthorizationHandler(svc.Authorization), IotCard: admin.NewIotCardHandler(svc.IotCard), IotCardImport: admin.NewIotCardImportHandler(svc.IotCardImport), + ExportTask: admin.NewExportTaskHandler(svc.ExportTask), Device: admin.NewDeviceHandler(svc.Device), DeviceImport: admin.NewDeviceImportHandler(svc.DeviceImport), AssetAllocationRecord: admin.NewAssetAllocationRecordHandler(svc.AssetAllocationRecord), diff --git a/internal/bootstrap/services.go b/internal/bootstrap/services.go index 7b903e3..4a750ce 100644 --- a/internal/bootstrap/services.go +++ b/internal/bootstrap/services.go @@ -28,6 +28,7 @@ import ( enterpriseCardSvc "github.com/break/junhong_cmp_fiber/internal/service/enterprise_card" enterpriseDeviceSvc "github.com/break/junhong_cmp_fiber/internal/service/enterprise_device" exchangeSvc "github.com/break/junhong_cmp_fiber/internal/service/exchange" + exportTaskSvc "github.com/break/junhong_cmp_fiber/internal/service/export_task" iotCardSvc "github.com/break/junhong_cmp_fiber/internal/service/iot_card" iotCardImportSvc "github.com/break/junhong_cmp_fiber/internal/service/iot_card_import" orderSvc "github.com/break/junhong_cmp_fiber/internal/service/order" @@ -71,6 +72,7 @@ type services struct { Authorization *enterpriseCardSvc.AuthorizationService IotCard *iotCardSvc.Service IotCardImport *iotCardImportSvc.Service + ExportTask *exportTaskSvc.Service Device *deviceSvc.Service DeviceImport *deviceImportSvc.Service AssetAllocationRecord *assetAllocationRecordSvc.Service @@ -197,6 +199,7 @@ func initServices(s *stores, deps *Dependencies) *services { Authorization: enterpriseCardSvc.NewAuthorizationService(s.Enterprise, s.IotCard, s.EnterpriseCardAuthorization, deps.Logger), IotCard: iotCard, IotCardImport: iotCardImportSvc.New(deps.DB, s.IotCardImportTask, deps.QueueClient), + ExportTask: exportTaskSvc.New(deps.DB, s.ExportTask, deps.QueueClient, deps.StorageService), Device: device, DeviceImport: deviceImportSvc.New(deps.DB, s.DeviceImportTask, deps.QueueClient), AssetAllocationRecord: assetAllocationRecordSvc.New(deps.DB, s.AssetAllocationRecord, s.Shop, s.Account), diff --git a/internal/bootstrap/stores.go b/internal/bootstrap/stores.go index ef93a9c..07eec73 100644 --- a/internal/bootstrap/stores.go +++ b/internal/bootstrap/stores.go @@ -25,6 +25,8 @@ type stores struct { EnterpriseDeviceAuthorization *postgres.EnterpriseDeviceAuthorizationStore IotCard *postgres.IotCardStore IotCardImportTask *postgres.IotCardImportTaskStore + ExportTask *postgres.ExportTaskStore + ExportShardTask *postgres.ExportShardTaskStore Device *postgres.DeviceStore DeviceSimBinding *postgres.DeviceSimBindingStore DeviceImportTask *postgres.DeviceImportTaskStore @@ -91,6 +93,8 @@ func initStores(deps *Dependencies) *stores { EnterpriseDeviceAuthorization: postgres.NewEnterpriseDeviceAuthorizationStore(deps.DB, deps.Redis), IotCard: postgres.NewIotCardStore(deps.DB, deps.Redis), IotCardImportTask: postgres.NewIotCardImportTaskStore(deps.DB, deps.Redis), + ExportTask: postgres.NewExportTaskStore(deps.DB, deps.Redis), + ExportShardTask: postgres.NewExportShardTaskStore(deps.DB, deps.Redis), Device: postgres.NewDeviceStore(deps.DB, deps.Redis), DeviceSimBinding: postgres.NewDeviceSimBindingStore(deps.DB, deps.Redis), DeviceImportTask: postgres.NewDeviceImportTaskStore(deps.DB, deps.Redis), diff --git a/internal/bootstrap/types.go b/internal/bootstrap/types.go index 4dbdc54..ed88224 100644 --- a/internal/bootstrap/types.go +++ b/internal/bootstrap/types.go @@ -35,6 +35,7 @@ type Handlers struct { Authorization *admin.AuthorizationHandler IotCard *admin.IotCardHandler IotCardImport *admin.IotCardImportHandler + ExportTask *admin.ExportTaskHandler Device *admin.DeviceHandler DeviceImport *admin.DeviceImportHandler AssetAllocationRecord *admin.AssetAllocationRecordHandler diff --git a/internal/bootstrap/worker_stores.go b/internal/bootstrap/worker_stores.go index 9da90a4..c1a8d50 100644 --- a/internal/bootstrap/worker_stores.go +++ b/internal/bootstrap/worker_stores.go @@ -9,6 +9,8 @@ type workerStores struct { IotCardImportTask *postgres.IotCardImportTaskStore IotCard *postgres.IotCardStore DeviceImportTask *postgres.DeviceImportTaskStore + ExportTask *postgres.ExportTaskStore + ExportShardTask *postgres.ExportShardTaskStore Device *postgres.DeviceStore DeviceSimBinding *postgres.DeviceSimBindingStore ShopSeriesCommissionStats *postgres.ShopSeriesCommissionStatsStore @@ -39,6 +41,8 @@ func initWorkerStores(deps *WorkerDependencies) *queue.WorkerStores { IotCardImportTask: postgres.NewIotCardImportTaskStore(deps.DB, deps.Redis), IotCard: postgres.NewIotCardStore(deps.DB, deps.Redis), DeviceImportTask: postgres.NewDeviceImportTaskStore(deps.DB, deps.Redis), + ExportTask: postgres.NewExportTaskStore(deps.DB, deps.Redis), + ExportShardTask: postgres.NewExportShardTaskStore(deps.DB, deps.Redis), Device: postgres.NewDeviceStore(deps.DB, deps.Redis), DeviceSimBinding: postgres.NewDeviceSimBindingStore(deps.DB, deps.Redis), ShopSeriesCommissionStats: postgres.NewShopSeriesCommissionStatsStore(deps.DB), @@ -68,6 +72,8 @@ func initWorkerStores(deps *WorkerDependencies) *queue.WorkerStores { IotCardImportTask: stores.IotCardImportTask, IotCard: stores.IotCard, DeviceImportTask: stores.DeviceImportTask, + ExportTask: stores.ExportTask, + ExportShardTask: stores.ExportShardTask, Device: stores.Device, DeviceSimBinding: stores.DeviceSimBinding, ShopSeriesCommissionStats: stores.ShopSeriesCommissionStats, diff --git a/internal/exporter/device_scene.go b/internal/exporter/device_scene.go new file mode 100644 index 0000000..06ed3bb --- /dev/null +++ b/internal/exporter/device_scene.go @@ -0,0 +1,98 @@ +package exporter + +import ( + "context" + "strconv" + + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// DeviceSceneStrategy 设备导出场景策略。 +type DeviceSceneStrategy struct { + db *gorm.DB +} + +// NewDeviceSceneStrategy 创建设备场景策略。 +func NewDeviceSceneStrategy(db *gorm.DB) *DeviceSceneStrategy { + return &DeviceSceneStrategy{db: db} +} + +// Scene 场景编码。 +func (s *DeviceSceneStrategy) Scene() string { + return constants.ExportTaskSceneDevice +} + +// Headers 导出表头。 +func (s *DeviceSceneStrategy) Headers() []string { + return []string{"ID", "设备虚拟号", "设备名称", "设备型号", "设备类型", "状态", "店铺ID", "创建时间"} +} + +// NextShardBoundary 获取下一段分片边界。 +func (s *DeviceSceneStrategy) NextShardBoundary(ctx context.Context, task *model.ExportTask, afterID uint64, limit int) (*ShardBoundary, error) { + if limit <= 0 { + limit = constants.ExportDefaultShardSize + } + + var ids []uint64 + query := s.db.WithContext(ctx). + Model(&model.Device{}). + Where("id > ?", afterID). + Order("id ASC"). + Limit(limit) + query = applyShopScopeForTask(query, task) + + if err := query.Pluck("id", &ids).Error; err != nil { + return nil, err + } + if len(ids) == 0 { + return nil, nil + } + + return &ShardBoundary{ + StartID: afterID, + EndID: ids[len(ids)-1], + RowCount: len(ids), + }, nil +} + +// QueryRows 查询分片内数据行。 +func (s *DeviceSceneStrategy) QueryRows(ctx context.Context, task *model.ExportTask, startID, endID uint64) ([][]string, error) { + if endID == 0 { + return [][]string{}, nil + } + + var devices []model.Device + query := s.db.WithContext(ctx). + Model(&model.Device{}). + Where("id > ? AND id <= ?", startID, endID). + Order("id ASC") + query = applyShopScopeForTask(query, task) + + if err := query.Find(&devices).Error; err != nil { + return nil, err + } + + rows := make([][]string, 0, len(devices)) + for _, item := range devices { + shopID := "" + if item.ShopID != nil { + shopID = strconv.FormatUint(uint64(*item.ShopID), 10) + } + + rows = append(rows, []string{ + strconv.FormatUint(uint64(item.ID), 10), + item.VirtualNo, + item.DeviceName, + item.DeviceModel, + item.DeviceType, + strconv.Itoa(item.Status), + shopID, + item.CreatedAt.Format("2006-01-02 15:04:05"), + }) + } + + return rows, nil +} diff --git a/internal/exporter/iot_card_scene.go b/internal/exporter/iot_card_scene.go new file mode 100644 index 0000000..a709177 --- /dev/null +++ b/internal/exporter/iot_card_scene.go @@ -0,0 +1,98 @@ +package exporter + +import ( + "context" + "strconv" + + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// IotCardSceneStrategy IoT 卡导出场景策略。 +type IotCardSceneStrategy struct { + db *gorm.DB +} + +// NewIotCardSceneStrategy 创建 IoT 卡场景策略。 +func NewIotCardSceneStrategy(db *gorm.DB) *IotCardSceneStrategy { + return &IotCardSceneStrategy{db: db} +} + +// Scene 场景编码。 +func (s *IotCardSceneStrategy) Scene() string { + return constants.ExportTaskSceneIotCard +} + +// Headers 导出表头。 +func (s *IotCardSceneStrategy) Headers() []string { + return []string{"ID", "ICCID", "MSISDN", "虚拟号", "运营商名称", "状态", "店铺ID", "创建时间"} +} + +// NextShardBoundary 获取下一段分片边界。 +func (s *IotCardSceneStrategy) NextShardBoundary(ctx context.Context, task *model.ExportTask, afterID uint64, limit int) (*ShardBoundary, error) { + if limit <= 0 { + limit = constants.ExportDefaultShardSize + } + + var ids []uint64 + query := s.db.WithContext(ctx). + Model(&model.IotCard{}). + Where("id > ?", afterID). + Order("id ASC"). + Limit(limit) + query = applyShopScopeForTask(query, task) + + if err := query.Pluck("id", &ids).Error; err != nil { + return nil, err + } + if len(ids) == 0 { + return nil, nil + } + + return &ShardBoundary{ + StartID: afterID, + EndID: ids[len(ids)-1], + RowCount: len(ids), + }, nil +} + +// QueryRows 查询分片内数据行。 +func (s *IotCardSceneStrategy) QueryRows(ctx context.Context, task *model.ExportTask, startID, endID uint64) ([][]string, error) { + if endID == 0 { + return [][]string{}, nil + } + + var cards []model.IotCard + query := s.db.WithContext(ctx). + Model(&model.IotCard{}). + Where("id > ? AND id <= ?", startID, endID). + Order("id ASC") + query = applyShopScopeForTask(query, task) + + if err := query.Find(&cards).Error; err != nil { + return nil, err + } + + rows := make([][]string, 0, len(cards)) + for _, item := range cards { + shopID := "" + if item.ShopID != nil { + shopID = strconv.FormatUint(uint64(*item.ShopID), 10) + } + + rows = append(rows, []string{ + strconv.FormatUint(uint64(item.ID), 10), + item.ICCID, + item.MSISDN, + item.VirtualNo, + item.CarrierName, + strconv.Itoa(item.Status), + shopID, + item.CreatedAt.Format("2006-01-02 15:04:05"), + }) + } + + return rows, nil +} diff --git a/internal/exporter/registry.go b/internal/exporter/registry.go new file mode 100644 index 0000000..1ead4c3 --- /dev/null +++ b/internal/exporter/registry.go @@ -0,0 +1,83 @@ +package exporter + +import ( + "context" + "sort" + + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// ShardBoundary 分片边界信息。 +type ShardBoundary struct { + StartID uint64 + EndID uint64 + RowCount int +} + +// SceneStrategy 导出场景策略接口。 +type SceneStrategy interface { + Scene() string + Headers() []string + NextShardBoundary(ctx context.Context, task *model.ExportTask, afterID uint64, limit int) (*ShardBoundary, error) + QueryRows(ctx context.Context, task *model.ExportTask, startID, endID uint64) ([][]string, error) +} + +// Registry 场景策略注册中心。 +type Registry struct { + strategies map[string]SceneStrategy +} + +// NewRegistry 创建场景策略注册中心。 +func NewRegistry(strategies ...SceneStrategy) *Registry { + m := make(map[string]SceneStrategy, len(strategies)) + for _, strategy := range strategies { + if strategy == nil { + continue + } + m[strategy.Scene()] = strategy + } + return &Registry{strategies: m} +} + +// NewDefaultRegistry 创建默认场景策略注册中心。 +func NewDefaultRegistry(db *gorm.DB) *Registry { + return NewRegistry( + NewDeviceSceneStrategy(db), + NewIotCardSceneStrategy(db), + ) +} + +// Get 获取指定场景策略。 +func (r *Registry) Get(scene string) (SceneStrategy, bool) { + strategy, ok := r.strategies[scene] + return strategy, ok +} + +// IsSupported 判断场景是否支持。 +func (r *Registry) IsSupported(scene string) bool { + _, ok := r.strategies[scene] + return ok +} + +// Scenes 返回当前支持的场景列表。 +func (r *Registry) Scenes() []string { + result := make([]string, 0, len(r.strategies)) + for scene := range r.strategies { + result = append(result, scene) + } + sort.Strings(result) + return result +} + +// IsSupportedScene 判断是否为受支持的场景。 +func IsSupportedScene(scene string) bool { + switch scene { + case constants.ExportTaskSceneDevice, constants.ExportTaskSceneIotCard: + return true + default: + return false + } +} diff --git a/internal/exporter/scope.go b/internal/exporter/scope.go new file mode 100644 index 0000000..69bb1e1 --- /dev/null +++ b/internal/exporter/scope.go @@ -0,0 +1,26 @@ +package exporter + +import ( + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +func applyShopScopeForTask(query *gorm.DB, task *model.ExportTask) *gorm.DB { + if task == nil { + return query.Where("1 = 0") + } + + switch task.CreatorUserType { + case constants.UserTypeSuperAdmin, constants.UserTypePlatform: + return query + case constants.UserTypeAgent: + if len(task.ScopeShopIDs) == 0 { + return query.Where("1 = 0") + } + return query.Where("shop_id IN ?", []uint(task.ScopeShopIDs)) + default: + return query.Where("1 = 0") + } +} diff --git a/internal/handler/admin/export_task.go b/internal/handler/admin/export_task.go new file mode 100644 index 0000000..f09e515 --- /dev/null +++ b/internal/handler/admin/export_task.go @@ -0,0 +1,97 @@ +package admin + +import ( + "strconv" + + "github.com/gofiber/fiber/v2" + + "github.com/break/junhong_cmp_fiber/internal/model/dto" + exportTaskService "github.com/break/junhong_cmp_fiber/internal/service/export_task" + "github.com/break/junhong_cmp_fiber/pkg/errors" + "github.com/break/junhong_cmp_fiber/pkg/response" +) + +// ExportTaskHandler 导出任务 Handler。 +type ExportTaskHandler struct { + service *exportTaskService.Service +} + +// NewExportTaskHandler 创建导出任务 Handler。 +func NewExportTaskHandler(service *exportTaskService.Service) *ExportTaskHandler { + return &ExportTaskHandler{service: service} +} + +// Create 创建导出任务。 +// POST /api/admin/export-tasks +func (h *ExportTaskHandler) Create(c *fiber.Ctx) error { + var req dto.CreateExportTaskRequest + if err := c.BodyParser(&req); err != nil { + return errors.New(errors.CodeInvalidParam) + } + if req.Scene == "" { + return errors.New(errors.CodeInvalidParam, "导出场景不能为空") + } + if req.Format == "" { + return errors.New(errors.CodeInvalidParam, "导出格式不能为空") + } + + result, err := h.service.CreateTask(c.UserContext(), &req) + if err != nil { + return err + } + return response.Success(c, result) +} + +// List 查询导出任务列表。 +// GET /api/admin/export-tasks +func (h *ExportTaskHandler) List(c *fiber.Ctx) error { + var req dto.ListExportTaskRequest + if err := c.QueryParser(&req); err != nil { + return errors.New(errors.CodeInvalidParam) + } + + result, err := h.service.ListTasks(c.UserContext(), &req) + if err != nil { + return err + } + + return response.SuccessWithPagination(c, result.List, result.Total, result.Page, result.PageSize) +} + +// GetByID 查询导出任务详情。 +// GET /api/admin/export-tasks/:id +func (h *ExportTaskHandler) GetByID(c *fiber.Ctx) error { + id, err := parseExportTaskID(c.Params("id")) + if err != nil { + return err + } + + result, err := h.service.GetTaskDetail(c.UserContext(), id) + if err != nil { + return err + } + return response.Success(c, result) +} + +// Cancel 取消导出任务。 +// POST /api/admin/export-tasks/:id/cancel +func (h *ExportTaskHandler) Cancel(c *fiber.Ctx) error { + id, err := parseExportTaskID(c.Params("id")) + if err != nil { + return err + } + + result, err := h.service.CancelTask(c.UserContext(), id) + if err != nil { + return err + } + return response.Success(c, result) +} + +func parseExportTaskID(idStr string) (uint, error) { + id, err := strconv.ParseUint(idStr, 10, 64) + if err != nil { + return 0, errors.New(errors.CodeInvalidParam, "无效的任务ID") + } + return uint(id), nil +} diff --git a/internal/model/dto/export_task_dto.go b/internal/model/dto/export_task_dto.go new file mode 100644 index 0000000..781aa82 --- /dev/null +++ b/internal/model/dto/export_task_dto.go @@ -0,0 +1,89 @@ +package dto + +import "time" + +// CreateExportTaskRequest 创建导出任务请求。 +type CreateExportTaskRequest struct { + Scene string `json:"scene" validate:"required,oneof=device iot_card" required:"true" description:"导出场景 (device:设备, iot_card:IoT卡)"` + Format string `json:"format" validate:"required,oneof=xlsx csv" required:"true" description:"导出格式 (xlsx:Excel, csv:CSV)"` + Query map[string]interface{} `json:"query,omitempty" description:"导出筛选参数(JSON对象，可选)"` +} + +// CreateExportTaskResponse 创建导出任务响应。 +type CreateExportTaskResponse struct { + TaskID uint `json:"task_id" description:"导出任务ID"` + TaskNo string `json:"task_no" description:"任务编号"` + Status int `json:"status" description:"任务状态 (1:待处理, 2:处理中, 3:已完成, 4:已失败, 5:已取消)"` + StatusName string `json:"status_name" description:"任务状态名称（中文）"` + Message string `json:"message" description:"提示信息"` +} + +// ListExportTaskRequest 导出任务列表请求。 +type ListExportTaskRequest struct { + Page int `json:"page" query:"page" validate:"omitempty,min=1" minimum:"1" description:"页码"` + PageSize int `json:"page_size" query:"page_size" validate:"omitempty,min=1,max=100" minimum:"1" maximum:"100" description:"每页数量"` + Scene string `json:"scene" query:"scene" validate:"omitempty,oneof=device iot_card" description:"导出场景 (device:设备, iot_card:IoT卡)"` + Status *int `json:"status" query:"status" validate:"omitempty,min=1,max=5" minimum:"1" maximum:"5" description:"任务状态 (1:待处理, 2:处理中, 3:已完成, 4:已失败, 5:已取消)"` + StartTime *time.Time `json:"start_time" query:"start_time" description:"创建时间起始"` + EndTime *time.Time `json:"end_time" query:"end_time" description:"创建时间结束"` +} + +// ExportTaskItem 导出任务列表项。 +type ExportTaskItem struct { + ID uint `json:"id" description:"任务ID"` + TaskNo string `json:"task_no" description:"任务编号"` + Scene string `json:"scene" description:"导出场景 (device:设备, iot_card:IoT卡)"` + Format string `json:"format" description:"导出格式 (xlsx:Excel, csv:CSV)"` + Status int `json:"status" description:"任务状态 (1:待处理, 2:处理中, 3:已完成, 4:已失败, 5:已取消)"` + StatusName string `json:"status_name" description:"任务状态名称（中文）"` + Progress int `json:"progress" description:"任务进度(0-100)"` + TotalRows int `json:"total_rows" description:"总行数"` + ProcessedRows int `json:"processed_rows" description:"已处理行数"` + TotalShards int `json:"total_shards" description:"总分片数"` + SuccessShards int `json:"success_shards" description:"成功分片数"` + FailedShards int `json:"failed_shards" description:"失败分片数"` + CancelRequested bool `json:"cancel_requested" description:"是否已请求取消"` + FileKey string `json:"file_key,omitempty" description:"导出文件Key"` + ErrorMessage string `json:"error_message,omitempty" description:"错误信息"` + CreatedAt time.Time `json:"created_at" description:"创建时间"` + StartedAt *time.Time `json:"started_at,omitempty" description:"开始处理时间"` + CompletedAt *time.Time `json:"completed_at,omitempty" description:"完成时间"` + CreatorUserID uint `json:"creator_user_id" description:"创建人用户ID"` + CreatorUserType int `json:"creator_user_type" description:"创建人用户类型 (2:平台, 3:代理, 4:企业)"` + CreatorShopID *uint `json:"creator_shop_id,omitempty" description:"创建人店铺ID"` + CreatorEnterpriseID *uint `json:"creator_enterprise_id,omitempty" description:"创建人企业ID"` +} + +// ListExportTaskResponse 导出任务列表响应。 +type ListExportTaskResponse struct { + List []*ExportTaskItem `json:"items" description:"任务列表"` + Total int64 `json:"total" description:"总数"` + Page int `json:"page" description:"当前页码"` + PageSize int `json:"size" description:"每页数量"` +} + +// GetExportTaskRequest 导出任务详情请求。 +type GetExportTaskRequest struct { + ID uint `path:"id" description:"任务ID" required:"true"` +} + +// ExportTaskDetailResponse 导出任务详情响应。 +type ExportTaskDetailResponse struct { + ExportTaskItem + DownloadURL string `json:"download_url,omitempty" description:"下载链接（仅已完成任务返回）"` + DownloadExpiresAt *time.Time `json:"download_expires_at,omitempty" description:"下载链接过期时间"` +} + +// CancelExportTaskRequest 取消导出任务请求。 +type CancelExportTaskRequest struct { + ID uint `path:"id" description:"任务ID" required:"true"` +} + +// CancelExportTaskResponse 取消导出任务响应。 +type CancelExportTaskResponse struct { + TaskID uint `json:"task_id" description:"任务ID"` + Status int `json:"status" description:"任务状态 (1:待处理, 2:处理中, 3:已完成, 4:已失败, 5:已取消)"` + StatusName string `json:"status_name" description:"任务状态名称（中文）"` + CancelRequested bool `json:"cancel_requested" description:"是否已请求取消"` + Message string `json:"message" description:"提示信息"` +} diff --git a/internal/model/export_task.go b/internal/model/export_task.go new file mode 100644 index 0000000..8fe0f6f --- /dev/null +++ b/internal/model/export_task.go @@ -0,0 +1,106 @@ +package model + +import ( + "database/sql/driver" + "encoding/json" + "time" + + "gorm.io/datatypes" + "gorm.io/gorm" +) + +// UIntListJSON uint 数组 JSON 类型。 +// 用于在 PostgreSQL jsonb 字段中存储店铺 ID 列表。 +type UIntListJSON []uint + +// Value 实现 driver.Valuer。 +func (ids UIntListJSON) Value() (driver.Value, error) { + if ids == nil { + return "[]", nil + } + return json.Marshal(ids) +} + +// Scan 实现 sql.Scanner。 +func (ids *UIntListJSON) Scan(value any) error { + if value == nil { + *ids = UIntListJSON{} + return nil + } + + switch v := value.(type) { + case []byte: + return json.Unmarshal(v, ids) + case string: + return json.Unmarshal([]byte(v), ids) + default: + *ids = UIntListJSON{} + return nil + } +} + +// ExportTask 导出主任务模型。 +// 记录导出任务的创建信息、执行进度、最终产物和取消状态。 +type ExportTask struct { + gorm.Model + BaseModel `gorm:"embedded"` + + TaskNo string `gorm:"column:task_no;type:varchar(50);uniqueIndex:idx_export_task_no,where:deleted_at IS NULL;not null;comment:任务编号(EXP-YYYYMMDD-XXXXXX)" json:"task_no"` + Scene string `gorm:"column:scene;type:varchar(50);index:idx_export_task_scene_status_created_at,priority:1;not null;comment:导出场景(device/iot_card)" json:"scene"` + Format string `gorm:"column:format;type:varchar(20);not null;comment:导出格式(xlsx/csv)" json:"format"` + Status int `gorm:"column:status;type:int;index:idx_export_task_scene_status_created_at,priority:2;not null;default:1;comment:任务状态 1-待处理 2-处理中 3-已完成 4-已失败 5-已取消" json:"status"` + Progress int `gorm:"column:progress;type:int;not null;default:0;comment:任务进度(0-100)" json:"progress"` + TotalRows int `gorm:"column:total_rows;type:int;not null;default:0;comment:总行数" json:"total_rows"` + + ProcessedRows int `gorm:"column:processed_rows;type:int;not null;default:0;comment:已处理行数" json:"processed_rows"` + TotalShards int `gorm:"column:total_shards;type:int;not null;default:0;comment:总分片数" json:"total_shards"` + SuccessShards int `gorm:"column:success_shards;type:int;not null;default:0;comment:成功分片数" json:"success_shards"` + FailedShards int `gorm:"column:failed_shards;type:int;not null;default:0;comment:失败分片数" json:"failed_shards"` + + CancelRequested bool `gorm:"column:cancel_requested;type:boolean;not null;default:false;index;comment:是否请求取消" json:"cancel_requested"` + + QueryJSON datatypes.JSON `gorm:"column:query_json;type:jsonb;not null;default:'{}';comment:导出查询参数(JSON)" json:"query_json"` + + ScopeShopIDs UIntListJSON `gorm:"column:scope_shop_ids;type:jsonb;not null;default:'[]';comment:导出时的数据权限店铺范围快照" json:"scope_shop_ids"` + CreatorUserID uint `gorm:"column:creator_user_id;type:bigint;not null;index;comment:创建人用户ID" json:"creator_user_id"` + CreatorUserType int `gorm:"column:creator_user_type;type:int;not null;comment:创建人用户类型 2-平台 3-代理 4-企业" json:"creator_user_type"` + CreatorShopID *uint `gorm:"column:creator_shop_id;type:bigint;index;comment:创建人店铺ID快照" json:"creator_shop_id,omitempty"` + CreatorEnterpriseID *uint `gorm:"column:creator_enterprise_id;type:bigint;index;comment:创建人企业ID快照" json:"creator_enterprise_id,omitempty"` + + FileKey string `gorm:"column:file_key;type:varchar(500);comment:最终产物文件Key" json:"file_key"` + FileSize int64 `gorm:"column:file_size;type:bigint;not null;default:0;comment:最终产物文件大小(字节)" json:"file_size"` + ErrorMessage string `gorm:"column:error_message;type:text;comment:错误信息" json:"error_message"` + StartedAt *time.Time `gorm:"column:started_at;comment:开始处理时间" json:"started_at"` + CompletedAt *time.Time `gorm:"column:completed_at;comment:完成时间" json:"completed_at"` +} + +// TableName 指定表名。 +func (ExportTask) TableName() string { + return "tb_export_task" +} + +// ExportShardTask 导出分片任务模型。 +// 记录主任务拆分后的每个分片执行状态和分片产物。 +type ExportShardTask struct { + gorm.Model + BaseModel `gorm:"embedded"` + + TaskID uint `gorm:"column:task_id;type:bigint;index:idx_export_shard_task_task_status,priority:1;not null;comment:主任务ID" json:"task_id"` + ShardNo int `gorm:"column:shard_no;type:int;not null;comment:分片序号(从1开始)" json:"shard_no"` + Status int `gorm:"column:status;type:int;index:idx_export_shard_task_task_status,priority:2;not null;default:1;comment:分片状态 1-待处理 2-处理中 3-已成功 4-已失败 5-已取消" json:"status"` + CursorStart uint64 `gorm:"column:cursor_start;type:bigint;not null;default:0;comment:分片游标起点(不含)" json:"cursor_start"` + CursorEnd uint64 `gorm:"column:cursor_end;type:bigint;not null;default:0;comment:分片游标终点(含)" json:"cursor_end"` + + RowCount int `gorm:"column:row_count;type:int;not null;default:0;comment:分片行数" json:"row_count"` + ProcessedRows int `gorm:"column:processed_rows;type:int;not null;default:0;comment:分片已处理行数" json:"processed_rows"` + FileKey string `gorm:"column:file_key;type:varchar(500);comment:分片产物文件Key" json:"file_key"` + FileSize int64 `gorm:"column:file_size;type:bigint;not null;default:0;comment:分片产物文件大小(字节)" json:"file_size"` + ErrorMessage string `gorm:"column:error_message;type:text;comment:分片错误信息" json:"error_message"` + StartedAt *time.Time `gorm:"column:started_at;comment:开始处理时间" json:"started_at"` + CompletedAt *time.Time `gorm:"column:completed_at;comment:完成时间" json:"completed_at"` +} + +// TableName 指定表名。 +func (ExportShardTask) TableName() string { + return "tb_export_shard_task" +} diff --git a/internal/routes/admin.go b/internal/routes/admin.go index a1ca163..5d7de6d 100644 --- a/internal/routes/admin.go +++ b/internal/routes/admin.go @@ -53,6 +53,9 @@ func RegisterAdminRoutes(router fiber.Router, handlers *bootstrap.Handlers, midd if handlers.IotCard != nil { registerIotCardRoutes(authGroup, handlers.IotCard, handlers.IotCardImport, doc, basePath) } + if handlers.ExportTask != nil { + registerExportTaskRoutes(authGroup, handlers.ExportTask, doc, basePath) + } if handlers.Device != nil { registerDeviceRoutes(authGroup, handlers.Device, handlers.DeviceImport, doc, basePath) } diff --git a/internal/routes/export_task.go b/internal/routes/export_task.go new file mode 100644 index 0000000..bbd52b1 --- /dev/null +++ b/internal/routes/export_task.go @@ -0,0 +1,50 @@ +package routes + +import ( + "github.com/gofiber/fiber/v2" + + "github.com/break/junhong_cmp_fiber/internal/handler/admin" + "github.com/break/junhong_cmp_fiber/internal/model/dto" + "github.com/break/junhong_cmp_fiber/pkg/openapi" +) + +func registerExportTaskRoutes(router fiber.Router, handler *admin.ExportTaskHandler, doc *openapi.Generator, basePath string) { + exportTasks := router.Group("/export-tasks") + groupPath := basePath + "/export-tasks" + + Register(exportTasks, doc, groupPath, "POST", "", handler.Create, RouteSpec{ + Summary: "创建导出任务", + Description: "创建统一导出任务，支持场景 scene=device/iot_card 和格式 format=xlsx/csv。", + Tags: []string{"导出任务"}, + Input: new(dto.CreateExportTaskRequest), + Output: new(dto.CreateExportTaskResponse), + Auth: true, + }) + + Register(exportTasks, doc, groupPath, "GET", "", handler.List, RouteSpec{ + Summary: "导出任务列表", + Description: "支持按 scene/status/time 过滤，分页默认 20，最大 100。", + Tags: []string{"导出任务"}, + Input: new(dto.ListExportTaskRequest), + Output: new(dto.ListExportTaskResponse), + Auth: true, + }) + + Register(exportTasks, doc, groupPath, "GET", "/:id", handler.GetByID, RouteSpec{ + Summary: "导出任务详情", + Description: "已完成任务返回 download_url，链接固定有效期 24 小时。", + Tags: []string{"导出任务"}, + Input: new(dto.GetExportTaskRequest), + Output: new(dto.ExportTaskDetailResponse), + Auth: true, + }) + + Register(exportTasks, doc, groupPath, "POST", "/:id/cancel", handler.Cancel, RouteSpec{ + Summary: "取消导出任务", + Description: "支持取消待处理/处理中任务；已完成/已失败/已取消任务不可取消。", + Tags: []string{"导出任务"}, + Input: new(dto.CancelExportTaskRequest), + Output: new(dto.CancelExportTaskResponse), + Auth: true, + }) +} diff --git a/internal/service/export_task/service.go b/internal/service/export_task/service.go new file mode 100644 index 0000000..52060fc --- /dev/null +++ b/internal/service/export_task/service.go @@ -0,0 +1,304 @@ +package export_task + +import ( + "context" + "time" + + "github.com/bytedance/sonic" + "github.com/hibiken/asynq" + "gorm.io/datatypes" + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/exporter" + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/internal/model/dto" + "github.com/break/junhong_cmp_fiber/internal/store" + "github.com/break/junhong_cmp_fiber/internal/store/postgres" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/errors" + "github.com/break/junhong_cmp_fiber/pkg/middleware" + "github.com/break/junhong_cmp_fiber/pkg/queue" + "github.com/break/junhong_cmp_fiber/pkg/storage" +) + +// Service 导出任务服务。 +type Service struct { + db *gorm.DB + taskStore *postgres.ExportTaskStore + queueClient *queue.Client + storageSvc *storage.Service + sceneRegistry *exporter.Registry +} + +type dispatchPayload struct { + TaskID uint `json:"task_id"` +} + +// New 创建导出任务服务。 +func New(db *gorm.DB, taskStore *postgres.ExportTaskStore, queueClient *queue.Client, storageSvc *storage.Service) *Service { + return &Service{ + db: db, + taskStore: taskStore, + queueClient: queueClient, + storageSvc: storageSvc, + sceneRegistry: exporter.NewDefaultRegistry(db), + } +} + +// CreateTask 创建导出任务并入队 dispatch。 +func (s *Service) CreateTask(ctx context.Context, req *dto.CreateExportTaskRequest) (*dto.CreateExportTaskResponse, error) { + userID := middleware.GetUserIDFromContext(ctx) + if userID == 0 { + return nil, errors.New(errors.CodeUnauthorized) + } + + userType := middleware.GetUserTypeFromContext(ctx) + if userType != constants.UserTypeSuperAdmin && userType != constants.UserTypePlatform && userType != constants.UserTypeAgent { + return nil, errors.New(errors.CodeForbidden, "当前账号无权限创建导出任务") + } + + if req.Scene == "" || !s.sceneRegistry.IsSupported(req.Scene) { + return nil, errors.New(errors.CodeInvalidParam, "导出场景不支持") + } + if req.Format != constants.ExportTaskFormatXLSX && req.Format != constants.ExportTaskFormatCSV { + return nil, errors.New(errors.CodeInvalidParam, "导出格式不支持") + } + + queryJSON := datatypes.JSON("{}") + if req.Query != nil { + raw, err := sonic.Marshal(req.Query) + if err != nil { + return nil, errors.Wrap(errors.CodeInvalidParam, err, "导出筛选参数格式不正确") + } + queryJSON = datatypes.JSON(raw) + } + + shopID := middleware.GetShopIDFromContext(ctx) + enterpriseID := middleware.GetEnterpriseIDFromContext(ctx) + + var ( + creatorShopID *uint + creatorEnterpriseID *uint + ) + if shopID > 0 { + creatorShopID = &shopID + } + if enterpriseID > 0 { + creatorEnterpriseID = &enterpriseID + } + + scopeShopIDs := model.UIntListJSON{} + if userType == constants.UserTypeAgent { + subordinateShopIDs := middleware.GetSubordinateShopIDs(ctx) + if subordinateShopIDs == nil { + if shopID == 0 { + return nil, errors.New(errors.CodeForbidden, "代理账号缺少店铺信息") + } + subordinateShopIDs = []uint{shopID} + } + if len(subordinateShopIDs) == 0 { + return nil, errors.New(errors.CodeForbidden, "当前账号无可导出的数据范围") + } + scopeShopIDs = subordinateShopIDs + } + + task := &model.ExportTask{ + TaskNo: s.taskStore.GenerateTaskNo(), + Scene: req.Scene, + Format: req.Format, + Status: constants.ExportTaskStatusPending, + Progress: 0, + QueryJSON: queryJSON, + ScopeShopIDs: scopeShopIDs, + CreatorUserID: userID, + CreatorUserType: userType, + CreatorShopID: creatorShopID, + CreatorEnterpriseID: creatorEnterpriseID, + } + task.Creator = userID + task.Updater = userID + + if err := s.taskStore.Create(ctx, task); err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "创建导出任务失败") + } + + if err := s.queueClient.EnqueueTask( + ctx, + constants.TaskTypeExportDispatch, + dispatchPayload{TaskID: task.ID}, + asynq.Queue(constants.QueueDefault), + asynq.MaxRetry(constants.ExportDispatchRetryMax), + asynq.Timeout(constants.ExportDispatchTaskTimeout), + ); err != nil { + _ = s.taskStore.MarkFailed(ctx, task.ID, userID, "导出任务入队失败") + return nil, errors.Wrap(errors.CodeTaskQueueError, err, "导出任务入队失败") + } + + return &dto.CreateExportTaskResponse{ + TaskID: task.ID, + TaskNo: task.TaskNo, + Status: task.Status, + StatusName: constants.GetExportTaskStatusName(task.Status), + Message: "导出任务创建成功，系统将异步处理", + }, nil +} + +// ListTasks 查询导出任务列表。 +func (s *Service) ListTasks(ctx context.Context, req *dto.ListExportTaskRequest) (*dto.ListExportTaskResponse, error) { + page := req.Page + if page <= 0 { + page = 1 + } + + pageSize := req.PageSize + if pageSize <= 0 { + pageSize = constants.DefaultPageSize + } + if pageSize > constants.MaxPageSize { + pageSize = constants.MaxPageSize + } + + filters := make(map[string]any) + if req.Scene != "" { + filters["scene"] = req.Scene + } + if req.Status != nil { + filters["status"] = *req.Status + } + if req.StartTime != nil { + filters["start_time"] = *req.StartTime + } + if req.EndTime != nil { + filters["end_time"] = *req.EndTime + } + + items, total, err := s.taskStore.List(ctx, &store.QueryOptions{ + Page: page, + PageSize: pageSize, + OrderBy: "created_at DESC", + }, filters) + if err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询导出任务列表失败") + } + + result := make([]*dto.ExportTaskItem, 0, len(items)) + for _, item := range items { + result = append(result, toTaskItemDTO(item)) + } + + return &dto.ListExportTaskResponse{ + List: result, + Total: total, + Page: page, + PageSize: pageSize, + }, nil +} + +// GetTaskDetail 查询导出任务详情。 +func (s *Service) GetTaskDetail(ctx context.Context, id uint) (*dto.ExportTaskDetailResponse, error) { + task, err := s.taskStore.GetByID(ctx, id) + if err != nil { + if err == gorm.ErrRecordNotFound { + return nil, errors.New(errors.CodeForbidden, "无权限操作该资源或资源不存在") + } + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询导出任务详情失败") + } + + resp := &dto.ExportTaskDetailResponse{ExportTaskItem: *toTaskItemDTO(task)} + if task.Status == constants.ExportTaskStatusCompleted && task.FileKey != "" && s.storageSvc != nil && s.storageSvc.Provider() != nil { + url, err := s.storageSvc.Provider().GetDownloadURL(ctx, task.FileKey, constants.ExportDownloadURLExpire) + if err != nil { + return nil, errors.Wrap(errors.CodeInternalError, err, "生成下载链接失败") + } + expiresAt := time.Now().Add(constants.ExportDownloadURLExpire) + resp.DownloadURL = url + resp.DownloadExpiresAt = &expiresAt + } + + return resp, nil +} + +// CancelTask 取消导出任务。 +func (s *Service) CancelTask(ctx context.Context, id uint) (*dto.CancelExportTaskResponse, error) { + userID := middleware.GetUserIDFromContext(ctx) + if userID == 0 { + return nil, errors.New(errors.CodeUnauthorized) + } + + task, err := s.taskStore.GetByID(ctx, id) + if err != nil { + if err == gorm.ErrRecordNotFound { + return nil, errors.New(errors.CodeForbidden, "无权限操作该资源或资源不存在") + } + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询导出任务失败") + } + + message := "取消请求已提交" + switch task.Status { + case constants.ExportTaskStatusPending: + ok, err := s.taskStore.CancelPendingTask(ctx, id, userID) + if err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "取消导出任务失败") + } + if !ok { + return nil, errors.New(errors.CodeInvalidStatus, "当前状态不支持取消") + } + message = "任务已取消" + case constants.ExportTaskStatusProcessing: + if !task.CancelRequested { + ok, err := s.taskStore.SetCancelRequested(ctx, id, userID) + if err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "提交取消请求失败") + } + if !ok { + return nil, errors.New(errors.CodeInvalidStatus, "当前状态不支持取消") + } + } else { + message = "取消请求已提交，请稍后刷新状态" + } + case constants.ExportTaskStatusCompleted, constants.ExportTaskStatusFailed, constants.ExportTaskStatusCancelled: + return nil, errors.New(errors.CodeInvalidStatus, "当前状态不支持取消") + default: + return nil, errors.New(errors.CodeInvalidStatus, "当前状态不支持取消") + } + + latestTask, err := s.taskStore.GetByID(ctx, id) + if err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询最新任务状态失败") + } + + return &dto.CancelExportTaskResponse{ + TaskID: latestTask.ID, + Status: latestTask.Status, + StatusName: constants.GetExportTaskStatusName(latestTask.Status), + CancelRequested: latestTask.CancelRequested, + Message: message, + }, nil +} + +func toTaskItemDTO(task *model.ExportTask) *dto.ExportTaskItem { + return &dto.ExportTaskItem{ + ID: task.ID, + TaskNo: task.TaskNo, + Scene: task.Scene, + Format: task.Format, + Status: task.Status, + StatusName: constants.GetExportTaskStatusName(task.Status), + Progress: task.Progress, + TotalRows: task.TotalRows, + ProcessedRows: task.ProcessedRows, + TotalShards: task.TotalShards, + SuccessShards: task.SuccessShards, + FailedShards: task.FailedShards, + CancelRequested: task.CancelRequested, + FileKey: task.FileKey, + ErrorMessage: task.ErrorMessage, + CreatedAt: task.CreatedAt, + StartedAt: task.StartedAt, + CompletedAt: task.CompletedAt, + CreatorUserID: task.CreatorUserID, + CreatorUserType: task.CreatorUserType, + CreatorShopID: task.CreatorShopID, + CreatorEnterpriseID: task.CreatorEnterpriseID, + } +} diff --git a/internal/store/postgres/export_shard_task_store.go b/internal/store/postgres/export_shard_task_store.go new file mode 100644 index 0000000..deb4d90 --- /dev/null +++ b/internal/store/postgres/export_shard_task_store.go @@ -0,0 +1,154 @@ +package postgres + +import ( + "context" + "time" + + "github.com/redis/go-redis/v9" + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// ExportShardTaskStore 导出分片任务数据访问层。 +type ExportShardTaskStore struct { + db *gorm.DB + redis *redis.Client +} + +// NewExportShardTaskStore 创建导出分片任务 Store。 +func NewExportShardTaskStore(db *gorm.DB, redis *redis.Client) *ExportShardTaskStore { + return &ExportShardTaskStore{db: db, redis: redis} +} + +// CreateBatch 批量创建分片任务。 +func (s *ExportShardTaskStore) CreateBatch(ctx context.Context, shards []*model.ExportShardTask) error { + if len(shards) == 0 { + return nil + } + return s.db.WithContext(ctx).Create(&shards).Error +} + +// GetByIDForWorker 按 ID 查询分片任务（Worker 使用）。 +func (s *ExportShardTaskStore) GetByIDForWorker(ctx context.Context, id uint) (*model.ExportShardTask, error) { + var shard model.ExportShardTask + if err := s.db.WithContext(ctx).Where("id = ?", id).First(&shard).Error; err != nil { + return nil, err + } + return &shard, nil +} + +// ListByTaskID 按任务 ID 查询分片任务列表。 +func (s *ExportShardTaskStore) ListByTaskID(ctx context.Context, taskID uint) ([]*model.ExportShardTask, error) { + var list []*model.ExportShardTask + if err := s.db.WithContext(ctx). + Where("task_id = ?", taskID). + Order("shard_no ASC"). + Find(&list).Error; err != nil { + return nil, err + } + return list, nil +} + +// TryMarkProcessingIfPending 尝试把分片从待处理更新为处理中。 +func (s *ExportShardTaskStore) TryMarkProcessingIfPending(ctx context.Context, shardID uint, updater uint) (bool, error) { + now := time.Now() + result := s.db.WithContext(ctx). + Model(&model.ExportShardTask{}). + Where("id = ? AND status = ?", shardID, constants.ExportShardStatusPending). + Updates(map[string]any{ + "status": constants.ExportShardStatusProcessing, + "started_at": now, + "updated_at": now, + "updater": updater, + }) + return result.RowsAffected > 0, result.Error +} + +// MarkSuccess 标记分片为成功。 +func (s *ExportShardTaskStore) MarkSuccess(ctx context.Context, shardID uint, updater uint, rowCount int, fileKey string, fileSize int64) (bool, error) { + now := time.Now() + result := s.db.WithContext(ctx). + Model(&model.ExportShardTask{}). + Where("id = ? AND status IN ?", shardID, []int{constants.ExportShardStatusPending, constants.ExportShardStatusProcessing}). + Updates(map[string]any{ + "status": constants.ExportShardStatusSuccess, + "row_count": rowCount, + "processed_rows": rowCount, + "file_key": fileKey, + "file_size": fileSize, + "completed_at": now, + "updated_at": now, + "updater": updater, + "error_message": "", + }) + return result.RowsAffected > 0, result.Error +} + +// MarkFailed 标记分片为失败。 +func (s *ExportShardTaskStore) MarkFailed(ctx context.Context, shardID uint, updater uint, errorMessage string) (bool, error) { + now := time.Now() + result := s.db.WithContext(ctx). + Model(&model.ExportShardTask{}). + Where("id = ? AND status IN ?", shardID, []int{constants.ExportShardStatusPending, constants.ExportShardStatusProcessing}). + Updates(map[string]any{ + "status": constants.ExportShardStatusFailed, + "error_message": errorMessage, + "completed_at": now, + "updated_at": now, + "updater": updater, + }) + return result.RowsAffected > 0, result.Error +} + +// MarkCancelled 标记分片为取消。 +func (s *ExportShardTaskStore) MarkCancelled(ctx context.Context, shardID uint, updater uint, errorMessage string) (bool, error) { + now := time.Now() + result := s.db.WithContext(ctx). + Model(&model.ExportShardTask{}). + Where("id = ? AND status IN ?", shardID, []int{constants.ExportShardStatusPending, constants.ExportShardStatusProcessing}). + Updates(map[string]any{ + "status": constants.ExportShardStatusCancelled, + "error_message": errorMessage, + "completed_at": now, + "updated_at": now, + "updater": updater, + }) + return result.RowsAffected > 0, result.Error +} + +// CancelByTaskID 批量取消指定主任务下未完成的分片。 +func (s *ExportShardTaskStore) CancelByTaskID(ctx context.Context, taskID uint, updater uint, errorMessage string) error { + now := time.Now() + return s.db.WithContext(ctx). + Model(&model.ExportShardTask{}). + Where("task_id = ? AND status IN ?", taskID, []int{constants.ExportShardStatusPending, constants.ExportShardStatusProcessing}). + Updates(map[string]any{ + "status": constants.ExportShardStatusCancelled, + "error_message": errorMessage, + "completed_at": now, + "updated_at": now, + "updater": updater, + }).Error +} + +// CountUnfinishedByTaskID 统计未完成分片数量。 +func (s *ExportShardTaskStore) CountUnfinishedByTaskID(ctx context.Context, taskID uint) (int64, error) { + var count int64 + err := s.db.WithContext(ctx). + Model(&model.ExportShardTask{}). + Where("task_id = ? AND status IN ?", taskID, []int{constants.ExportShardStatusPending, constants.ExportShardStatusProcessing}). + Count(&count).Error + return count, err +} + +// CountByTaskIDAndStatus 统计指定状态分片数量。 +func (s *ExportShardTaskStore) CountByTaskIDAndStatus(ctx context.Context, taskID uint, status int) (int64, error) { + var count int64 + err := s.db.WithContext(ctx). + Model(&model.ExportShardTask{}). + Where("task_id = ? AND status = ?", taskID, status). + Count(&count).Error + return count, err +} diff --git a/internal/store/postgres/export_task_store.go b/internal/store/postgres/export_task_store.go new file mode 100644 index 0000000..9775016 --- /dev/null +++ b/internal/store/postgres/export_task_store.go @@ -0,0 +1,337 @@ +package postgres + +import ( + "context" + "fmt" + "time" + + "github.com/redis/go-redis/v9" + "gorm.io/gorm" + "gorm.io/gorm/clause" + + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/internal/store" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/middleware" +) + +// ExportTaskStore 导出主任务数据访问层。 +type ExportTaskStore struct { + db *gorm.DB + redis *redis.Client +} + +// NewExportTaskStore 创建导出主任务 Store。 +func NewExportTaskStore(db *gorm.DB, redis *redis.Client) *ExportTaskStore { + return &ExportTaskStore{db: db, redis: redis} +} + +// Create 创建导出主任务。 +func (s *ExportTaskStore) Create(ctx context.Context, task *model.ExportTask) error { + return s.db.WithContext(ctx).Create(task).Error +} + +// GetByID 按 ID 查询任务（带数据权限）。 +func (s *ExportTaskStore) GetByID(ctx context.Context, id uint) (*model.ExportTask, error) { + var task model.ExportTask + query := applyExportTaskVisibility(ctx, s.db.WithContext(ctx).Model(&model.ExportTask{})) + if err := query.Where("id = ?", id).First(&task).Error; err != nil { + return nil, err + } + return &task, nil +} + +// GetByIDForWorker 按 ID 查询任务（Worker 使用，不带数据权限）。 +func (s *ExportTaskStore) GetByIDForWorker(ctx context.Context, id uint) (*model.ExportTask, error) { + var task model.ExportTask + if err := s.db.WithContext(ctx).Where("id = ?", id).First(&task).Error; err != nil { + return nil, err + } + return &task, nil +} + +// List 查询导出任务列表（带数据权限）。 +func (s *ExportTaskStore) List(ctx context.Context, opts *store.QueryOptions, filters map[string]any) ([]*model.ExportTask, int64, error) { + var ( + items []*model.ExportTask + total int64 + ) + + query := applyExportTaskVisibility(ctx, s.db.WithContext(ctx).Model(&model.ExportTask{})) + + if scene, ok := filters["scene"].(string); ok && scene != "" { + query = query.Where("scene = ?", scene) + } + if status, ok := filters["status"].(int); ok && status > 0 { + query = query.Where("status = ?", status) + } + if startTime, ok := filters["start_time"].(time.Time); ok && !startTime.IsZero() { + query = query.Where("created_at >= ?", startTime) + } + if endTime, ok := filters["end_time"].(time.Time); ok && !endTime.IsZero() { + query = query.Where("created_at <= ?", endTime) + } + + if err := query.Count(&total).Error; err != nil { + return nil, 0, err + } + + if opts == nil { + opts = &store.QueryOptions{Page: 1, PageSize: constants.DefaultPageSize} + } + if opts.Page <= 0 { + opts.Page = 1 + } + if opts.PageSize <= 0 { + opts.PageSize = constants.DefaultPageSize + } + + offset := (opts.Page - 1) * opts.PageSize + query = query.Offset(offset).Limit(opts.PageSize) + + if opts.OrderBy != "" { + query = query.Order(opts.OrderBy) + } else { + query = query.Order("created_at DESC") + } + + if err := query.Find(&items).Error; err != nil { + return nil, 0, err + } + + return items, total, nil +} + +// GenerateTaskNo 生成导出任务编号。 +func (s *ExportTaskStore) GenerateTaskNo() string { + now := time.Now() + return fmt.Sprintf("EXP-%s-%06d", now.Format("20060102"), now.UnixNano()%1000000) +} + +// TryMarkProcessingIfPending 尝试将任务从待处理更新为处理中。 +func (s *ExportTaskStore) TryMarkProcessingIfPending(ctx context.Context, taskID uint, updater uint) (bool, error) { + now := time.Now() + result := s.db.WithContext(ctx). + Model(&model.ExportTask{}). + Where("id = ? AND status = ?", taskID, constants.ExportTaskStatusPending). + Updates(map[string]any{ + "status": constants.ExportTaskStatusProcessing, + "started_at": now, + "updated_at": now, + "updater": updater, + }) + return result.RowsAffected > 0, result.Error +} + +// UpdateDispatchPlan 更新 dispatch 阶段计算出的总量信息。 +func (s *ExportTaskStore) UpdateDispatchPlan(ctx context.Context, taskID uint, totalRows, totalShards uint, updater uint) error { + now := time.Now() + return s.db.WithContext(ctx). + Model(&model.ExportTask{}). + Where("id = ?", taskID). + Updates(map[string]any{ + "total_rows": totalRows, + "total_shards": totalShards, + "updated_at": now, + "updater": updater, + "error_message": "", + }).Error +} + +// IsCancelRequested 查询是否已请求取消。 +func (s *ExportTaskStore) IsCancelRequested(ctx context.Context, taskID uint) (bool, error) { + var value bool + if err := s.db.WithContext(ctx). + Model(&model.ExportTask{}). + Where("id = ?", taskID). + Pluck("cancel_requested", &value).Error; err != nil { + return false, err + } + return value, nil +} + +// CancelPendingTask 将待处理任务直接取消。 +func (s *ExportTaskStore) CancelPendingTask(ctx context.Context, taskID uint, updater uint) (bool, error) { + now := time.Now() + result := s.db.WithContext(ctx). + Model(&model.ExportTask{}). + Where("id = ? AND status = ?", taskID, constants.ExportTaskStatusPending). + Updates(map[string]any{ + "status": constants.ExportTaskStatusCancelled, + "cancel_requested": true, + "progress": 100, + "completed_at": now, + "updated_at": now, + "updater": updater, + }) + return result.RowsAffected > 0, result.Error +} + +// SetCancelRequested 标记处理中任务取消请求。 +func (s *ExportTaskStore) SetCancelRequested(ctx context.Context, taskID uint, updater uint) (bool, error) { + now := time.Now() + result := s.db.WithContext(ctx). + Model(&model.ExportTask{}). + Where("id = ? AND status = ?", taskID, constants.ExportTaskStatusProcessing). + Updates(map[string]any{ + "cancel_requested": true, + "updated_at": now, + "updater": updater, + }) + return result.RowsAffected > 0, result.Error +} + +// MarkCancelledByWorker 在 Worker 侧将任务状态更新为已取消。 +func (s *ExportTaskStore) MarkCancelledByWorker(ctx context.Context, taskID uint, updater uint, msg string) error { + now := time.Now() + return s.db.WithContext(ctx). + Model(&model.ExportTask{}). + Where("id = ? AND status IN ?", taskID, []int{constants.ExportTaskStatusPending, constants.ExportTaskStatusProcessing}). + Updates(map[string]any{ + "status": constants.ExportTaskStatusCancelled, + "cancel_requested": true, + "progress": 100, + "completed_at": now, + "updated_at": now, + "updater": updater, + "error_message": msg, + }).Error +} + +// MarkCompleted 将任务更新为已完成。 +func (s *ExportTaskStore) MarkCompleted(ctx context.Context, taskID uint, updater uint, fileKey string, fileSize int64) error { + now := time.Now() + return s.db.WithContext(ctx). + Model(&model.ExportTask{}). + Where("id = ? AND status = ?", taskID, constants.ExportTaskStatusProcessing). + Updates(map[string]any{ + "status": constants.ExportTaskStatusCompleted, + "progress": 100, + "file_key": fileKey, + "file_size": fileSize, + "completed_at": now, + "updated_at": now, + "updater": updater, + "error_message": "", + }).Error +} + +// MarkFailed 将任务更新为失败。 +func (s *ExportTaskStore) MarkFailed(ctx context.Context, taskID uint, updater uint, errorMessage string) error { + now := time.Now() + return s.db.WithContext(ctx). + Model(&model.ExportTask{}). + Where("id = ? AND status IN ?", taskID, []int{constants.ExportTaskStatusPending, constants.ExportTaskStatusProcessing}). + Updates(map[string]any{ + "status": constants.ExportTaskStatusFailed, + "completed_at": now, + "updated_at": now, + "updater": updater, + "error_message": errorMessage, + }).Error +} + +// ApplyShardSuccess 在分片成功后更新主任务进度与统计。 +func (s *ExportTaskStore) ApplyShardSuccess(ctx context.Context, taskID uint, updater uint, rowCount int) error { + return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var task model.ExportTask + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where("id = ?", taskID). + First(&task).Error; err != nil { + return err + } + if task.Status != constants.ExportTaskStatusProcessing { + return nil + } + + processedRows := task.ProcessedRows + rowCount + successShards := task.SuccessShards + 1 + progress := task.Progress + + if task.TotalRows > 0 { + progress = processedRows * 100 / task.TotalRows + } else if task.TotalShards > 0 { + progress = successShards * 100 / task.TotalShards + } + if progress > 99 { + progress = 99 + } + if progress < task.Progress { + progress = task.Progress + } + + now := time.Now() + return tx.Model(&model.ExportTask{}). + Where("id = ?", taskID). + Updates(map[string]any{ + "processed_rows": processedRows, + "success_shards": successShards, + "progress": progress, + "updated_at": now, + "updater": updater, + }).Error + }) +} + +// ApplyShardFailed 在分片失败后更新主任务失败计数。 +func (s *ExportTaskStore) ApplyShardFailed(ctx context.Context, taskID uint, updater uint, errorMessage string) error { + return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var task model.ExportTask + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where("id = ?", taskID). + First(&task).Error; err != nil { + return err + } + if task.Status != constants.ExportTaskStatusProcessing { + return nil + } + + now := time.Now() + updates := map[string]any{ + "failed_shards": task.FailedShards + 1, + "updated_at": now, + "updater": updater, + } + if task.ErrorMessage == "" { + updates["error_message"] = errorMessage + } + + return tx.Model(&model.ExportTask{}). + Where("id = ?", taskID). + Updates(updates).Error + }) +} + +func applyExportTaskVisibility(ctx context.Context, query *gorm.DB) *gorm.DB { + userType := middleware.GetUserTypeFromContext(ctx) + + switch userType { + case constants.UserTypeSuperAdmin, constants.UserTypePlatform: + return query + case constants.UserTypeAgent: + shopIDs := middleware.GetSubordinateShopIDs(ctx) + if shopIDs == nil { + shopID := middleware.GetShopIDFromContext(ctx) + if shopID == 0 { + return query.Where("1 = 0") + } + return query.Where("creator_shop_id = ?", shopID) + } + if len(shopIDs) == 0 { + return query.Where("1 = 0") + } + return query.Where("creator_shop_id IN ?", shopIDs) + case constants.UserTypeEnterprise: + enterpriseID := middleware.GetEnterpriseIDFromContext(ctx) + if enterpriseID == 0 { + return query.Where("1 = 0") + } + return query.Where("creator_enterprise_id = ?", enterpriseID) + default: + userID := middleware.GetUserIDFromContext(ctx) + if userID == 0 { + return query.Where("1 = 0") + } + return query.Where("creator_user_id = ?", userID) + } +} diff --git a/internal/task/export_common.go b/internal/task/export_common.go new file mode 100644 index 0000000..714a58e --- /dev/null +++ b/internal/task/export_common.go @@ -0,0 +1,209 @@ +package task + +import ( + "context" + "encoding/csv" + "errors" + "fmt" + "os" + "path/filepath" + "time" + + "github.com/bytedance/sonic" + "github.com/hibiken/asynq" + "github.com/redis/go-redis/v9" + "github.com/xuri/excelize/v2" + + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/storage" +) + +const ( + exportLockTTL = 5 * time.Minute + exportShardLockTTL = 10 * time.Minute + exportFinalizeLockTTL = 2 * time.Minute +) + +// ExportDispatchPayload 导出 dispatch 任务载荷。 +type ExportDispatchPayload struct { + TaskID uint `json:"task_id"` +} + +// ExportShardPayload 导出 shard 任务载荷。 +type ExportShardPayload struct { + TaskID uint `json:"task_id"` + ShardID uint `json:"shard_id"` +} + +// ExportFinalizePayload 导出 finalize 任务载荷。 +type ExportFinalizePayload struct { + TaskID uint `json:"task_id"` +} + +func enqueueTask(ctx context.Context, client *asynq.Client, taskType string, payload any, opts ...asynq.Option) error { + if client == nil { + return errors.New("asynq client 未初始化") + } + + bytes, err := sonic.Marshal(payload) + if err != nil { + return err + } + + task := asynq.NewTask(taskType, bytes, opts...) + _, err = client.EnqueueContext(ctx, task) + return err +} + +func tryAcquireLock(ctx context.Context, rdb *redis.Client, key string, ttl time.Duration) (bool, error) { + if rdb == nil { + return true, nil + } + return rdb.SetNX(ctx, key, "1", ttl).Result() +} + +func releaseLock(ctx context.Context, rdb *redis.Client, key string) { + if rdb == nil { + return + } + _, _ = rdb.Del(ctx, key).Result() +} + +func isFinalRetry(ctx context.Context) bool { + retryCount, ok := asynq.GetRetryCount(ctx) + if !ok { + return false + } + maxRetry, ok := asynq.GetMaxRetry(ctx) + if !ok { + return false + } + return retryCount >= maxRetry +} + +func writeExportFile(format string, headers []string, rows [][]string) (string, int64, error) { + switch format { + case constants.ExportTaskFormatCSV: + return writeCSVFile(headers, rows) + case constants.ExportTaskFormatXLSX: + return writeXLSXFile(headers, rows) + default: + return "", 0, fmt.Errorf("不支持的导出格式: %s", format) + } +} + +func writeCSVFile(headers []string, rows [][]string) (string, int64, error) { + file, err := os.CreateTemp("", "export-*.csv") + if err != nil { + return "", 0, err + } + defer file.Close() + + writer := csv.NewWriter(file) + if err := writer.Write(headers); err != nil { + return "", 0, err + } + for _, row := range rows { + if err := writer.Write(row); err != nil { + return "", 0, err + } + } + writer.Flush() + if err := writer.Error(); err != nil { + return "", 0, err + } + + if err := file.Sync(); err != nil { + return "", 0, err + } + + stat, err := file.Stat() + if err != nil { + return "", 0, err + } + return file.Name(), stat.Size(), nil +} + +func writeXLSXFile(headers []string, rows [][]string) (string, int64, error) { + file := excelize.NewFile() + sheet := file.GetSheetName(file.GetActiveSheetIndex()) + + streamWriter, err := file.NewStreamWriter(sheet) + if err != nil { + return "", 0, err + } + + headerValues := make([]interface{}, 0, len(headers)) + for _, item := range headers { + headerValues = append(headerValues, item) + } + if err := streamWriter.SetRow("A1", headerValues); err != nil { + return "", 0, err + } + + for i, row := range rows { + axis, err := excelize.CoordinatesToCellName(1, i+2) + if err != nil { + return "", 0, err + } + values := make([]interface{}, 0, len(row)) + for _, item := range row { + values = append(values, item) + } + if err := streamWriter.SetRow(axis, values); err != nil { + return "", 0, err + } + } + + if err := streamWriter.Flush(); err != nil { + return "", 0, err + } + + path := filepath.Join(os.TempDir(), fmt.Sprintf("export-%d.xlsx", time.Now().UnixNano())) + if err := file.SaveAs(path); err != nil { + return "", 0, err + } + if err := file.Close(); err != nil { + return "", 0, err + } + + stat, err := os.Stat(path) + if err != nil { + return "", 0, err + } + return path, stat.Size(), nil +} + +func uploadExportFile(ctx context.Context, storageSvc *storage.Service, localPath, fileName string) (string, error) { + if storageSvc == nil || storageSvc.Provider() == nil { + return "", errors.New("对象存储服务未配置") + } + + key, err := storageSvc.GenerateFileKey("export", fileName) + if err != nil { + return "", err + } + + file, err := os.Open(localPath) + if err != nil { + return "", err + } + defer file.Close() + + contentType := "application/octet-stream" + if filepath.Ext(fileName) == ".csv" { + contentType = "text/csv" + } + if filepath.Ext(fileName) == ".xlsx" { + contentType = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + } + + if err := storageSvc.Provider().Upload(ctx, key, file, contentType); err != nil { + return "", err + } + return key, nil +} + +func parseExportPayload[T any](payload []byte, target *T) error { + return sonic.Unmarshal(payload, target) +} diff --git a/internal/task/export_dispatch.go b/internal/task/export_dispatch.go new file mode 100644 index 0000000..1379673 --- /dev/null +++ b/internal/task/export_dispatch.go @@ -0,0 +1,255 @@ +package task + +import ( + "context" + "fmt" + "time" + + "github.com/hibiken/asynq" + "github.com/redis/go-redis/v9" + "go.uber.org/zap" + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/exporter" + "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" +) + +// ExportDispatchHandler 导出 dispatch 处理器。 +type ExportDispatchHandler struct { + db *gorm.DB + redis *redis.Client + taskStore *postgres.ExportTaskStore + shardStore *postgres.ExportShardTaskStore + asynqClient *asynq.Client + sceneRegistry *exporter.Registry + logger *zap.Logger +} + +// NewExportDispatchHandler 创建导出 dispatch 处理器。 +func NewExportDispatchHandler( + db *gorm.DB, + redis *redis.Client, + taskStore *postgres.ExportTaskStore, + shardStore *postgres.ExportShardTaskStore, + asynqClient *asynq.Client, + sceneRegistry *exporter.Registry, + logger *zap.Logger, +) *ExportDispatchHandler { + return &ExportDispatchHandler{ + db: db, + redis: redis, + taskStore: taskStore, + shardStore: shardStore, + asynqClient: asynqClient, + sceneRegistry: sceneRegistry, + logger: logger, + } +} + +// HandleExportDispatch 处理导出 dispatch 任务。 +func (h *ExportDispatchHandler) HandleExportDispatch(ctx context.Context, task *asynq.Task) error { + var payload ExportDispatchPayload + if err := parseExportPayload(task.Payload(), &payload); err != nil { + h.logger.Error("解析导出 dispatch 任务载荷失败", zap.Error(err)) + return asynq.SkipRetry + } + + lockKey := constants.RedisExportDispatchLockKey(payload.TaskID) + locked, err := tryAcquireLock(ctx, h.redis, lockKey, exportLockTTL) + if err != nil { + h.logger.Error("获取导出 dispatch 锁失败", zap.Uint("task_id", payload.TaskID), zap.Error(err)) + return err + } + if !locked { + h.logger.Info("导出 dispatch 已在执行，跳过", zap.Uint("task_id", payload.TaskID)) + return nil + } + defer releaseLock(ctx, h.redis, lockKey) + + exportTask, err := h.taskStore.GetByIDForWorker(ctx, payload.TaskID) + if err != nil { + h.logger.Error("查询导出任务失败", zap.Uint("task_id", payload.TaskID), zap.Error(err)) + return asynq.SkipRetry + } + + if exportTask.Status == constants.ExportTaskStatusCompleted || exportTask.Status == constants.ExportTaskStatusFailed || exportTask.Status == constants.ExportTaskStatusCancelled { + return nil + } + + updater := dispatchUpdater(exportTask) + + if exportTask.CancelRequested { + if err := h.shardStore.CancelByTaskID(ctx, exportTask.ID, updater, "任务已取消"); err != nil { + return err + } + if err := h.taskStore.MarkCancelledByWorker(ctx, exportTask.ID, updater, "任务已取消"); err != nil { + return err + } + return nil + } + + started, err := h.taskStore.TryMarkProcessingIfPending(ctx, exportTask.ID, updater) + if err != nil { + h.logger.Error("更新导出任务状态为处理中失败", zap.Uint("task_id", exportTask.ID), zap.Error(err)) + return err + } + + if !started { + if exportTask.Status != constants.ExportTaskStatusProcessing { + return nil + } + existingShards, err := h.shardStore.ListByTaskID(ctx, exportTask.ID) + if err != nil { + return err + } + if len(existingShards) == 0 { + return nil + } + if err := h.enqueueShardTasks(ctx, exportTask, existingShards); err != nil { + return err + } + return h.enqueueFinalizeTask(ctx, exportTask.ID) + } + + strategy, ok := h.sceneRegistry.Get(exportTask.Scene) + if !ok { + _ = h.taskStore.MarkFailed(ctx, exportTask.ID, updater, "导出场景未注册") + return asynq.SkipRetry + } + + shards, totalRows, err := h.buildShards(ctx, exportTask, strategy, updater) + if err != nil { + _ = h.taskStore.MarkFailed(ctx, exportTask.ID, updater, "构建导出分片失败") + h.logger.Error("构建导出分片失败", zap.Uint("task_id", exportTask.ID), zap.Error(err)) + return err + } + + if err := h.persistShardsAndPlan(ctx, exportTask.ID, updater, shards, totalRows); err != nil { + _ = h.taskStore.MarkFailed(ctx, exportTask.ID, updater, "保存导出分片失败") + return err + } + + if err := h.enqueueShardTasks(ctx, exportTask, shards); err != nil { + return err + } + + if err := h.enqueueFinalizeTask(ctx, exportTask.ID); err != nil { + return err + } + + h.logger.Info("导出 dispatch 完成", + zap.Uint("task_id", exportTask.ID), + zap.Uint("total_rows", totalRows), + zap.Int("total_shards", len(shards)), + ) + + return nil +} + +func (h *ExportDispatchHandler) buildShards(ctx context.Context, task *model.ExportTask, strategy exporter.SceneStrategy, updater uint) ([]*model.ExportShardTask, uint, error) { + shards := make([]*model.ExportShardTask, 0) + totalRows := uint(0) + afterID := uint64(0) + shardNo := 1 + + for { + if task.CancelRequested { + break + } + + boundary, err := strategy.NextShardBoundary(ctx, task, afterID, constants.ExportDefaultShardSize) + if err != nil { + return nil, 0, err + } + if boundary == nil { + break + } + + rowCount := boundary.RowCount + if rowCount < 0 { + rowCount = 0 + } + + totalRows += uint(rowCount) + shard := &model.ExportShardTask{ + TaskID: task.ID, + ShardNo: shardNo, + Status: constants.ExportShardStatusPending, + CursorStart: boundary.StartID, + CursorEnd: boundary.EndID, + RowCount: rowCount, + } + shard.Creator = updater + shard.Updater = updater + shards = append(shards, shard) + + afterID = boundary.EndID + shardNo++ + } + + return shards, totalRows, nil +} + +func (h *ExportDispatchHandler) persistShardsAndPlan(ctx context.Context, taskID uint, updater uint, shards []*model.ExportShardTask, totalRows uint) error { + return h.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + txShardStore := postgres.NewExportShardTaskStore(tx, h.redis) + txTaskStore := postgres.NewExportTaskStore(tx, h.redis) + + if err := txShardStore.CreateBatch(ctx, shards); err != nil { + return err + } + if err := txTaskStore.UpdateDispatchPlan(ctx, taskID, totalRows, uint(len(shards)), updater); err != nil { + return err + } + return nil + }) +} + +func (h *ExportDispatchHandler) enqueueShardTasks(ctx context.Context, task *model.ExportTask, shards []*model.ExportShardTask) error { + for _, shard := range shards { + payload := ExportShardPayload{TaskID: task.ID, ShardID: shard.ID} + err := enqueueTask( + ctx, + h.asynqClient, + constants.TaskTypeExportShard, + payload, + asynq.Queue(constants.QueueLow), + asynq.MaxRetry(constants.ExportShardRetryMax), + asynq.Timeout(constants.ExportShardTaskTimeout), + ) + if err != nil { + h.logger.Error("导出分片任务入队失败", + zap.Uint("task_id", task.ID), + zap.Uint("shard_id", shard.ID), + zap.Error(err), + ) + return fmt.Errorf("enqueue shard task failed: %w", err) + } + } + return nil +} + +func (h *ExportDispatchHandler) enqueueFinalizeTask(ctx context.Context, taskID uint) error { + return enqueueTask( + ctx, + h.asynqClient, + constants.TaskTypeExportFinalize, + ExportFinalizePayload{TaskID: taskID}, + asynq.Queue(constants.QueueDefault), + asynq.MaxRetry(constants.ExportFinalizeRetryMax), + asynq.Timeout(constants.ExportFinalizeTaskTimeout), + asynq.ProcessIn(5*time.Second), + ) +} + +func dispatchUpdater(task *model.ExportTask) uint { + if task.CreatorUserID > 0 { + return task.CreatorUserID + } + if task.Creator > 0 { + return task.Creator + } + return 0 +} diff --git a/internal/task/export_finalize.go b/internal/task/export_finalize.go new file mode 100644 index 0000000..09c5eda --- /dev/null +++ b/internal/task/export_finalize.go @@ -0,0 +1,158 @@ +package task + +import ( + "context" + "fmt" + "os" + + "github.com/hibiken/asynq" + "github.com/redis/go-redis/v9" + "go.uber.org/zap" + + "github.com/break/junhong_cmp_fiber/internal/exporter" + "github.com/break/junhong_cmp_fiber/internal/store/postgres" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/storage" +) + +// ExportFinalizeHandler 导出 finalize 处理器。 +type ExportFinalizeHandler struct { + redis *redis.Client + taskStore *postgres.ExportTaskStore + shardStore *postgres.ExportShardTaskStore + storageSvc *storage.Service + sceneRegistry *exporter.Registry + logger *zap.Logger +} + +// NewExportFinalizeHandler 创建导出 finalize 处理器。 +func NewExportFinalizeHandler( + redis *redis.Client, + taskStore *postgres.ExportTaskStore, + shardStore *postgres.ExportShardTaskStore, + storageSvc *storage.Service, + sceneRegistry *exporter.Registry, + logger *zap.Logger, +) *ExportFinalizeHandler { + return &ExportFinalizeHandler{ + redis: redis, + taskStore: taskStore, + shardStore: shardStore, + storageSvc: storageSvc, + sceneRegistry: sceneRegistry, + logger: logger, + } +} + +// HandleExportFinalize 处理导出 finalize 任务。 +func (h *ExportFinalizeHandler) HandleExportFinalize(ctx context.Context, task *asynq.Task) error { + var payload ExportFinalizePayload + if err := parseExportPayload(task.Payload(), &payload); err != nil { + h.logger.Error("解析导出 finalize 任务载荷失败", zap.Error(err)) + return asynq.SkipRetry + } + + lockKey := constants.RedisExportFinalizeLockKey(payload.TaskID) + locked, err := tryAcquireLock(ctx, h.redis, lockKey, exportFinalizeLockTTL) + if err != nil { + return err + } + if !locked { + return nil + } + defer releaseLock(ctx, h.redis, lockKey) + + exportTask, err := h.taskStore.GetByIDForWorker(ctx, payload.TaskID) + if err != nil { + h.logger.Error("查询导出任务失败", zap.Uint("task_id", payload.TaskID), zap.Error(err)) + return asynq.SkipRetry + } + + if exportTask.Status == constants.ExportTaskStatusCompleted || exportTask.Status == constants.ExportTaskStatusFailed || exportTask.Status == constants.ExportTaskStatusCancelled { + return nil + } + + updater := dispatchUpdater(exportTask) + if exportTask.CancelRequested { + if err := h.shardStore.CancelByTaskID(ctx, exportTask.ID, updater, "任务已取消"); err != nil { + return err + } + if err := h.taskStore.MarkCancelledByWorker(ctx, exportTask.ID, updater, "任务已取消"); err != nil { + return err + } + return nil + } + + unfinishedCount, err := h.shardStore.CountUnfinishedByTaskID(ctx, exportTask.ID) + if err != nil { + return err + } + if unfinishedCount > 0 { + return fmt.Errorf("分片任务尚未完成: %d", unfinishedCount) + } + + failedCount, err := h.shardStore.CountByTaskIDAndStatus(ctx, exportTask.ID, constants.ExportShardStatusFailed) + if err != nil { + return err + } + if failedCount > 0 { + if err := h.taskStore.MarkFailed(ctx, exportTask.ID, updater, "存在失败分片，导出任务失败"); err != nil { + return err + } + return nil + } + + strategy, ok := h.sceneRegistry.Get(exportTask.Scene) + if !ok { + if err := h.taskStore.MarkFailed(ctx, exportTask.ID, updater, "导出场景未注册"); err != nil { + return err + } + return nil + } + + shards, err := h.shardStore.ListByTaskID(ctx, exportTask.ID) + if err != nil { + return err + } + + rows := make([][]string, 0) + for _, shard := range shards { + if shard.Status != constants.ExportShardStatusSuccess { + continue + } + currentRows, err := strategy.QueryRows(ctx, exportTask, shard.CursorStart, shard.CursorEnd) + if err != nil { + return h.handleFinalizeError(ctx, exportTask.ID, updater, "读取分片数据失败", err) + } + rows = append(rows, currentRows...) + } + + localPath, fileSize, err := writeExportFile(exportTask.Format, strategy.Headers(), rows) + if err != nil { + return h.handleFinalizeError(ctx, exportTask.ID, updater, "生成汇总文件失败", err) + } + defer func() { _ = os.Remove(localPath) }() + + fileName := fmt.Sprintf("%s.%s", exportTask.TaskNo, exportTask.Format) + fileKey, err := uploadExportFile(ctx, h.storageSvc, localPath, fileName) + if err != nil { + return h.handleFinalizeError(ctx, exportTask.ID, updater, "上传汇总文件失败", err) + } + + if err := h.taskStore.MarkCompleted(ctx, exportTask.ID, updater, fileKey, fileSize); err != nil { + return err + } + + return nil +} + +func (h *ExportFinalizeHandler) handleFinalizeError(ctx context.Context, taskID uint, updater uint, prefix string, err error) error { + if isFinalRetry(ctx) { + markErr := h.taskStore.MarkFailed(ctx, taskID, updater, prefix+": "+err.Error()) + if markErr != nil { + return markErr + } + return nil + } + return err +} diff --git a/internal/task/export_shard.go b/internal/task/export_shard.go new file mode 100644 index 0000000..3cd0f30 --- /dev/null +++ b/internal/task/export_shard.go @@ -0,0 +1,183 @@ +package task + +import ( + "context" + "fmt" + "os" + "time" + + "github.com/hibiken/asynq" + "github.com/redis/go-redis/v9" + "go.uber.org/zap" + + "github.com/break/junhong_cmp_fiber/internal/exporter" + "github.com/break/junhong_cmp_fiber/internal/store/postgres" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/storage" +) + +// ExportShardHandler 导出 shard 处理器。 +type ExportShardHandler struct { + redis *redis.Client + taskStore *postgres.ExportTaskStore + shardStore *postgres.ExportShardTaskStore + asynqClient *asynq.Client + storageSvc *storage.Service + sceneRegistry *exporter.Registry + logger *zap.Logger +} + +// NewExportShardHandler 创建导出 shard 处理器。 +func NewExportShardHandler( + redis *redis.Client, + taskStore *postgres.ExportTaskStore, + shardStore *postgres.ExportShardTaskStore, + asynqClient *asynq.Client, + storageSvc *storage.Service, + sceneRegistry *exporter.Registry, + logger *zap.Logger, +) *ExportShardHandler { + return &ExportShardHandler{ + redis: redis, + taskStore: taskStore, + shardStore: shardStore, + asynqClient: asynqClient, + storageSvc: storageSvc, + sceneRegistry: sceneRegistry, + logger: logger, + } +} + +// HandleExportShard 处理导出 shard 任务。 +func (h *ExportShardHandler) HandleExportShard(ctx context.Context, task *asynq.Task) error { + var payload ExportShardPayload + if err := parseExportPayload(task.Payload(), &payload); err != nil { + h.logger.Error("解析导出 shard 任务载荷失败", zap.Error(err)) + return asynq.SkipRetry + } + + lockKey := constants.RedisExportShardLockKey(payload.TaskID, payload.ShardID) + locked, err := tryAcquireLock(ctx, h.redis, lockKey, exportShardLockTTL) + if err != nil { + h.logger.Error("获取导出 shard 锁失败", zap.Uint("task_id", payload.TaskID), zap.Uint("shard_id", payload.ShardID), zap.Error(err)) + return err + } + if !locked { + return nil + } + defer releaseLock(ctx, h.redis, lockKey) + + exportTask, err := h.taskStore.GetByIDForWorker(ctx, payload.TaskID) + if err != nil { + h.logger.Error("查询导出任务失败", zap.Uint("task_id", payload.TaskID), zap.Error(err)) + return asynq.SkipRetry + } + + shardTask, err := h.shardStore.GetByIDForWorker(ctx, payload.ShardID) + if err != nil { + h.logger.Error("查询导出分片失败", zap.Uint("task_id", payload.TaskID), zap.Uint("shard_id", payload.ShardID), zap.Error(err)) + return asynq.SkipRetry + } + + if shardTask.Status == constants.ExportShardStatusSuccess || shardTask.Status == constants.ExportShardStatusFailed || shardTask.Status == constants.ExportShardStatusCancelled { + return nil + } + + updater := dispatchUpdater(exportTask) + if exportTask.CancelRequested { + _, _ = h.shardStore.MarkCancelled(ctx, shardTask.ID, updater, "任务已取消") + _ = h.enqueueFinalize(ctx, exportTask.ID) + return nil + } + + if shardTask.Status == constants.ExportShardStatusPending { + started, err := h.shardStore.TryMarkProcessingIfPending(ctx, shardTask.ID, updater) + if err != nil { + return err + } + if !started { + latest, err := h.shardStore.GetByIDForWorker(ctx, shardTask.ID) + if err != nil { + return err + } + if latest.Status != constants.ExportShardStatusProcessing { + return nil + } + shardTask = latest + } + } + + strategy, ok := h.sceneRegistry.Get(exportTask.Scene) + if !ok { + _ = h.markShardFinalFailed(ctx, exportTask.ID, shardTask.ID, updater, "导出场景未注册") + return asynq.SkipRetry + } + + rows, err := strategy.QueryRows(ctx, exportTask, shardTask.CursorStart, shardTask.CursorEnd) + if err != nil { + return h.handleShardError(ctx, exportTask.ID, shardTask.ID, updater, "查询分片数据失败", err) + } + + localPath, fileSize, err := writeExportFile(exportTask.Format, strategy.Headers(), rows) + if err != nil { + return h.handleShardError(ctx, exportTask.ID, shardTask.ID, updater, "生成分片文件失败", err) + } + defer func() { _ = os.Remove(localPath) }() + + fileName := fmt.Sprintf("%s-shard-%04d.%s", exportTask.TaskNo, shardTask.ShardNo, exportTask.Format) + fileKey, err := uploadExportFile(ctx, h.storageSvc, localPath, fileName) + if err != nil { + return h.handleShardError(ctx, exportTask.ID, shardTask.ID, updater, "上传分片文件失败", err) + } + + updated, err := h.shardStore.MarkSuccess(ctx, shardTask.ID, updater, len(rows), fileKey, fileSize) + if err != nil { + return err + } + if updated { + if err := h.taskStore.ApplyShardSuccess(ctx, exportTask.ID, updater, len(rows)); err != nil { + return err + } + } + + if err := h.enqueueFinalize(ctx, exportTask.ID); err != nil { + h.logger.Warn("触发 finalize 入队失败", zap.Uint("task_id", exportTask.ID), zap.Error(err)) + } + + return nil +} + +func (h *ExportShardHandler) handleShardError(ctx context.Context, taskID, shardID uint, updater uint, prefix string, err error) error { + if isFinalRetry(ctx) { + _ = h.markShardFinalFailed(ctx, taskID, shardID, updater, prefix+": "+err.Error()) + return nil + } + return err +} + +func (h *ExportShardHandler) markShardFinalFailed(ctx context.Context, taskID, shardID uint, updater uint, message string) error { + updated, err := h.shardStore.MarkFailed(ctx, shardID, updater, message) + if err != nil { + return err + } + if updated { + if err := h.taskStore.ApplyShardFailed(ctx, taskID, updater, message); err != nil { + return err + } + } + _ = h.enqueueFinalize(ctx, taskID) + return nil +} + +func (h *ExportShardHandler) enqueueFinalize(ctx context.Context, taskID uint) error { + return enqueueTask( + ctx, + h.asynqClient, + constants.TaskTypeExportFinalize, + ExportFinalizePayload{TaskID: taskID}, + asynq.Queue(constants.QueueDefault), + asynq.MaxRetry(constants.ExportFinalizeRetryMax), + asynq.Timeout(constants.ExportFinalizeTaskTimeout), + asynq.ProcessIn(2*time.Second), + ) +} diff --git a/migrations/000135_create_export_task_tables.down.sql b/migrations/000135_create_export_task_tables.down.sql new file mode 100644 index 0000000..53e3c95 --- /dev/null +++ b/migrations/000135_create_export_task_tables.down.sql @@ -0,0 +1,4 @@ +-- 回滚：删除统一导出任务相关表 + +DROP TABLE IF EXISTS tb_export_shard_task; +DROP TABLE IF EXISTS tb_export_task; diff --git a/migrations/000135_create_export_task_tables.up.sql b/migrations/000135_create_export_task_tables.up.sql new file mode 100644 index 0000000..e91b7eb --- /dev/null +++ b/migrations/000135_create_export_task_tables.up.sql @@ -0,0 +1,149 @@ +-- 创建统一导出主任务表与分片任务表 + +CREATE TABLE IF NOT EXISTS tb_export_task ( + id BIGSERIAL PRIMARY KEY, + created_at TIMESTAMP NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP NOT NULL DEFAULT NOW(), + deleted_at TIMESTAMP, + + creator BIGINT NOT NULL DEFAULT 0, + updater BIGINT NOT NULL DEFAULT 0, + + task_no VARCHAR(50) NOT NULL, + scene VARCHAR(50) NOT NULL, + format VARCHAR(20) NOT NULL, + status INT NOT NULL DEFAULT 1, + progress INT NOT NULL DEFAULT 0, + + total_rows INT NOT NULL DEFAULT 0, + processed_rows INT NOT NULL DEFAULT 0, + total_shards INT NOT NULL DEFAULT 0, + success_shards INT NOT NULL DEFAULT 0, + failed_shards INT NOT NULL DEFAULT 0, + + cancel_requested BOOLEAN NOT NULL DEFAULT FALSE, + + query_json JSONB NOT NULL DEFAULT '{}'::jsonb, + scope_shop_ids JSONB NOT NULL DEFAULT '[]'::jsonb, + + creator_user_id BIGINT NOT NULL, + creator_user_type INT NOT NULL, + creator_shop_id BIGINT, + creator_enterprise_id BIGINT, + + file_key VARCHAR(500) NOT NULL DEFAULT '', + file_size BIGINT NOT NULL DEFAULT 0, + error_message TEXT, + + started_at TIMESTAMP, + completed_at TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS tb_export_shard_task ( + id BIGSERIAL PRIMARY KEY, + created_at TIMESTAMP NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP NOT NULL DEFAULT NOW(), + deleted_at TIMESTAMP, + + creator BIGINT NOT NULL DEFAULT 0, + updater BIGINT NOT NULL DEFAULT 0, + + task_id BIGINT NOT NULL, + shard_no INT NOT NULL, + status INT NOT NULL DEFAULT 1, + + cursor_start BIGINT NOT NULL DEFAULT 0, + cursor_end BIGINT NOT NULL DEFAULT 0, + + row_count INT NOT NULL DEFAULT 0, + processed_rows INT NOT NULL DEFAULT 0, + + file_key VARCHAR(500) NOT NULL DEFAULT '', + file_size BIGINT NOT NULL DEFAULT 0, + error_message TEXT, + + started_at TIMESTAMP, + completed_at TIMESTAMP +); + +-- 主任务索引 +CREATE UNIQUE INDEX IF NOT EXISTS idx_export_task_no + ON tb_export_task(task_no) + WHERE deleted_at IS NULL; +CREATE INDEX IF NOT EXISTS idx_export_task_scene_status_created_at + ON tb_export_task(scene, status, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_export_task_creator_user_id + ON tb_export_task(creator_user_id); +CREATE INDEX IF NOT EXISTS idx_export_task_creator_shop_id + ON tb_export_task(creator_shop_id); +CREATE INDEX IF NOT EXISTS idx_export_task_creator_enterprise_id + ON tb_export_task(creator_enterprise_id); +CREATE INDEX IF NOT EXISTS idx_export_task_cancel_requested + ON tb_export_task(cancel_requested); +CREATE INDEX IF NOT EXISTS idx_export_task_deleted_at + ON tb_export_task(deleted_at); + +-- 分片任务索引 +CREATE UNIQUE INDEX IF NOT EXISTS idx_export_shard_task_task_shard_no + ON tb_export_shard_task(task_id, shard_no) + WHERE deleted_at IS NULL; +CREATE INDEX IF NOT EXISTS idx_export_shard_task_task_status + ON tb_export_shard_task(task_id, status); +CREATE INDEX IF NOT EXISTS idx_export_shard_task_deleted_at + ON tb_export_shard_task(deleted_at); +CREATE INDEX IF NOT EXISTS idx_export_shard_task_created_at + ON tb_export_shard_task(created_at DESC); + +-- 表注释 +COMMENT ON TABLE tb_export_task IS '统一导出主任务表'; +COMMENT ON TABLE tb_export_shard_task IS '统一导出分片任务表'; + +-- 主任务字段注释 +COMMENT ON COLUMN tb_export_task.id IS '主键ID'; +COMMENT ON COLUMN tb_export_task.created_at IS '创建时间'; +COMMENT ON COLUMN tb_export_task.updated_at IS '更新时间'; +COMMENT ON COLUMN tb_export_task.deleted_at IS '删除时间（软删除）'; +COMMENT ON COLUMN tb_export_task.creator IS '创建人ID'; +COMMENT ON COLUMN tb_export_task.updater IS '更新人ID'; +COMMENT ON COLUMN tb_export_task.task_no IS '任务编号(EXP-YYYYMMDD-XXXXXX)'; +COMMENT ON COLUMN tb_export_task.scene IS '导出场景(device/iot_card)'; +COMMENT ON COLUMN tb_export_task.format IS '导出格式(xlsx/csv)'; +COMMENT ON COLUMN tb_export_task.status IS '任务状态 1-待处理 2-处理中 3-已完成 4-已失败 5-已取消'; +COMMENT ON COLUMN tb_export_task.progress IS '任务进度(0-100)'; +COMMENT ON COLUMN tb_export_task.total_rows IS '总行数'; +COMMENT ON COLUMN tb_export_task.processed_rows IS '已处理行数'; +COMMENT ON COLUMN tb_export_task.total_shards IS '总分片数'; +COMMENT ON COLUMN tb_export_task.success_shards IS '成功分片数'; +COMMENT ON COLUMN tb_export_task.failed_shards IS '失败分片数'; +COMMENT ON COLUMN tb_export_task.cancel_requested IS '是否请求取消'; +COMMENT ON COLUMN tb_export_task.query_json IS '导出查询参数(JSON)'; +COMMENT ON COLUMN tb_export_task.scope_shop_ids IS '导出时的数据权限店铺范围快照'; +COMMENT ON COLUMN tb_export_task.creator_user_id IS '创建人用户ID'; +COMMENT ON COLUMN tb_export_task.creator_user_type IS '创建人用户类型 2-平台 3-代理 4-企业'; +COMMENT ON COLUMN tb_export_task.creator_shop_id IS '创建人店铺ID快照'; +COMMENT ON COLUMN tb_export_task.creator_enterprise_id IS '创建人企业ID快照'; +COMMENT ON COLUMN tb_export_task.file_key IS '最终产物文件Key'; +COMMENT ON COLUMN tb_export_task.file_size IS '最终产物文件大小(字节)'; +COMMENT ON COLUMN tb_export_task.error_message IS '错误信息'; +COMMENT ON COLUMN tb_export_task.started_at IS '开始处理时间'; +COMMENT ON COLUMN tb_export_task.completed_at IS '完成时间'; + +-- 分片任务字段注释 +COMMENT ON COLUMN tb_export_shard_task.id IS '主键ID'; +COMMENT ON COLUMN tb_export_shard_task.created_at IS '创建时间'; +COMMENT ON COLUMN tb_export_shard_task.updated_at IS '更新时间'; +COMMENT ON COLUMN tb_export_shard_task.deleted_at IS '删除时间（软删除）'; +COMMENT ON COLUMN tb_export_shard_task.creator IS '创建人ID'; +COMMENT ON COLUMN tb_export_shard_task.updater IS '更新人ID'; +COMMENT ON COLUMN tb_export_shard_task.task_id IS '主任务ID'; +COMMENT ON COLUMN tb_export_shard_task.shard_no IS '分片序号(从1开始)'; +COMMENT ON COLUMN tb_export_shard_task.status IS '分片状态 1-待处理 2-处理中 3-已成功 4-已失败 5-已取消'; +COMMENT ON COLUMN tb_export_shard_task.cursor_start IS '分片游标起点(不含)'; +COMMENT ON COLUMN tb_export_shard_task.cursor_end IS '分片游标终点(含)'; +COMMENT ON COLUMN tb_export_shard_task.row_count IS '分片行数'; +COMMENT ON COLUMN tb_export_shard_task.processed_rows IS '分片已处理行数'; +COMMENT ON COLUMN tb_export_shard_task.file_key IS '分片产物文件Key'; +COMMENT ON COLUMN tb_export_shard_task.file_size IS '分片产物文件大小(字节)'; +COMMENT ON COLUMN tb_export_shard_task.error_message IS '分片错误信息'; +COMMENT ON COLUMN tb_export_shard_task.started_at IS '开始处理时间'; +COMMENT ON COLUMN tb_export_shard_task.completed_at IS '完成时间'; diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go index b76225b..a808d4c 100644 --- a/pkg/constants/constants.go +++ b/pkg/constants/constants.go @@ -46,6 +46,9 @@ const ( TaskTypeCommission = "commission:calculate" // 分佣计算 TaskTypeIotCardImport = "iot_card:import" // IoT 卡批量导入 TaskTypeDeviceImport = "device:import" // 设备批量导入 + TaskTypeExportDispatch = "export:dispatch" // 导出任务分发 + TaskTypeExportShard = "export:shard" // 导出任务分片执行 + TaskTypeExportFinalize = "export:finalize" // 导出任务收尾汇总 TaskTypeCommissionStatsUpdate = "commission:stats:update" // 佣金统计更新 TaskTypeCommissionStatsSync = "commission:stats:sync" // 佣金统计同步 TaskTypeCommissionStatsArchive = "commission:stats:archive" // 佣金统计归档 @@ -175,6 +178,48 @@ const ( DefaultConcurrency = 10 ) +// 导出任务相关常量 +const ( + ExportTaskSceneDevice = "device" // 导出场景：设备 + ExportTaskSceneIotCard = "iot_card" // 导出场景：IoT 卡 +) + +// 导出文件格式常量 +const ( + ExportTaskFormatXLSX = "xlsx" // 导出格式：Excel + ExportTaskFormatCSV = "csv" // 导出格式：CSV +) + +// 导出主任务状态常量 +const ( + ExportTaskStatusPending = 1 // 待处理 + ExportTaskStatusProcessing = 2 // 处理中 + ExportTaskStatusCompleted = 3 // 已完成 + ExportTaskStatusFailed = 4 // 已失败 + ExportTaskStatusCancelled = 5 // 已取消 +) + +// 导出分片任务状态常量 +const ( + ExportShardStatusPending = 1 // 待处理 + ExportShardStatusProcessing = 2 // 处理中 + ExportShardStatusSuccess = 3 // 已成功 + ExportShardStatusFailed = 4 // 已失败 + ExportShardStatusCancelled = 5 // 已取消 +) + +// 导出任务执行配置常量 +const ( + ExportDefaultShardSize = 2000 // 默认分片大小 + ExportDownloadURLExpire = 24 * time.Hour // 下载链接固定有效期 + ExportDispatchTaskTimeout = 10 * time.Minute // dispatch 任务超时 + ExportShardTaskTimeout = 15 * time.Minute // shard 任务超时 + ExportFinalizeTaskTimeout = 10 * time.Minute // finalize 任务超时 + ExportFinalizeRetryMax = 30 // finalize 最大重试次数 + ExportShardRetryMax = 3 // shard 最大重试次数 + ExportDispatchRetryMax = 5 // dispatch 最大重试次数 +) + // 店铺配置常量 const ( MaxShopLevel = 7 // 店铺最大层级 @@ -260,6 +305,42 @@ func GetShelfStatusName(status int) string { } } +// GetExportTaskStatusName 获取导出主任务状态名称 +func GetExportTaskStatusName(status int) string { + switch status { + case ExportTaskStatusPending: + return "待处理" + case ExportTaskStatusProcessing: + return "处理中" + case ExportTaskStatusCompleted: + return "已完成" + case ExportTaskStatusFailed: + return "已失败" + case ExportTaskStatusCancelled: + return "已取消" + default: + return "未知" + } +} + +// GetExportShardStatusName 获取导出分片任务状态名称 +func GetExportShardStatusName(status int) string { + switch status { + case ExportShardStatusPending: + return "待处理" + case ExportShardStatusProcessing: + return "处理中" + case ExportShardStatusSuccess: + return "已成功" + case ExportShardStatusFailed: + return "已失败" + case ExportShardStatusCancelled: + return "已取消" + default: + return "未知" + } +} + // GetExchangeStatusName 获取换货单状态名称 // 用于 exchange_status 字段 func GetExchangeStatusName(status int) string { diff --git a/pkg/constants/redis.go b/pkg/constants/redis.go index d8f3590..d287784 100644 --- a/pkg/constants/redis.go +++ b/pkg/constants/redis.go @@ -46,6 +46,27 @@ func RedisTaskStatusKey(taskID string) string { return fmt.Sprintf("task:status:%s", taskID) } +// RedisExportDispatchLockKey 生成导出 dispatch 执行锁的 Redis 键 +// 用途：防止同一导出任务并发执行 dispatch +// 过期时间：5 分钟 +func RedisExportDispatchLockKey(taskID uint) string { + return fmt.Sprintf("export:dispatch:lock:%d", taskID) +} + +// RedisExportShardLockKey 生成导出分片执行锁的 Redis 键 +// 用途：防止同一分片被多个 Worker 并发处理 +// 过期时间：10 分钟 +func RedisExportShardLockKey(taskID, shardID uint) string { + return fmt.Sprintf("export:shard:lock:%d:%d", taskID, shardID) +} + +// RedisExportFinalizeLockKey 生成导出 finalize 执行锁的 Redis 键 +// 用途：防止同一导出任务并发执行 finalize +// 过期时间：2 分钟 +func RedisExportFinalizeLockKey(taskID uint) string { + return fmt.Sprintf("export:finalize:lock:%d", taskID) +} + // RedisShopSubordinatesKey 生成店铺下级 ID 列表的 Redis 键 // 用途：缓存递归查询的下级店铺 ID 列表 // 过期时间：30 分钟 diff --git a/pkg/openapi/handlers.go b/pkg/openapi/handlers.go index fae8ec9..74c8c14 100644 --- a/pkg/openapi/handlers.go +++ b/pkg/openapi/handlers.go @@ -36,6 +36,7 @@ func BuildDocHandlers() *bootstrap.Handlers { Authorization: admin.NewAuthorizationHandler(nil), IotCard: admin.NewIotCardHandler(nil), IotCardImport: admin.NewIotCardImportHandler(nil), + ExportTask: admin.NewExportTaskHandler(nil), Device: admin.NewDeviceHandler(nil), DeviceImport: admin.NewDeviceImportHandler(nil), AssetAllocationRecord: admin.NewAssetAllocationRecordHandler(nil), diff --git a/pkg/queue/handler.go b/pkg/queue/handler.go index c6bd85c..121f911 100644 --- a/pkg/queue/handler.go +++ b/pkg/queue/handler.go @@ -6,6 +6,7 @@ import ( "go.uber.org/zap" "gorm.io/gorm" + "github.com/break/junhong_cmp_fiber/internal/exporter" "github.com/break/junhong_cmp_fiber/internal/gateway" "github.com/break/junhong_cmp_fiber/internal/polling" iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card" @@ -64,6 +65,7 @@ func (h *Handler) RegisterHandlers() *asynq.ServeMux { h.registerIotCardImportHandler() h.registerDeviceImportHandler() + h.registerExportHandlers() h.registerCommissionStatsHandlers() h.registerCommissionCalculationHandler() h.registerPollingHandlers() @@ -113,6 +115,46 @@ func (h *Handler) registerDeviceImportHandler() { h.logger.Info("注册设备导入任务处理器", zap.String("task_type", constants.TaskTypeDeviceImport)) } +func (h *Handler) registerExportHandlers() { + sceneRegistry := exporter.NewDefaultRegistry(h.db) + + dispatchHandler := task.NewExportDispatchHandler( + h.db, + h.redis, + h.workerResult.Stores.ExportTask, + h.workerResult.Stores.ExportShardTask, + h.asynqClient, + sceneRegistry, + h.logger, + ) + shardHandler := task.NewExportShardHandler( + h.redis, + h.workerResult.Stores.ExportTask, + h.workerResult.Stores.ExportShardTask, + h.asynqClient, + h.storage, + sceneRegistry, + h.logger, + ) + finalizeHandler := task.NewExportFinalizeHandler( + h.redis, + h.workerResult.Stores.ExportTask, + h.workerResult.Stores.ExportShardTask, + h.storage, + sceneRegistry, + h.logger, + ) + + h.mux.HandleFunc(constants.TaskTypeExportDispatch, dispatchHandler.HandleExportDispatch) + h.logger.Info("注册导出 dispatch 任务处理器", zap.String("task_type", constants.TaskTypeExportDispatch)) + + h.mux.HandleFunc(constants.TaskTypeExportShard, shardHandler.HandleExportShard) + h.logger.Info("注册导出 shard 任务处理器", zap.String("task_type", constants.TaskTypeExportShard)) + + h.mux.HandleFunc(constants.TaskTypeExportFinalize, finalizeHandler.HandleExportFinalize) + h.logger.Info("注册导出 finalize 任务处理器", zap.String("task_type", constants.TaskTypeExportFinalize)) +} + func (h *Handler) registerCommissionStatsHandlers() { updateHandler := task.NewCommissionStatsUpdateHandler( h.redis, diff --git a/pkg/queue/types.go b/pkg/queue/types.go index 22e94c3..9a03faf 100644 --- a/pkg/queue/types.go +++ b/pkg/queue/types.go @@ -22,6 +22,8 @@ type WorkerStores struct { IotCardImportTask *postgres.IotCardImportTaskStore IotCard *postgres.IotCardStore DeviceImportTask *postgres.DeviceImportTaskStore + ExportTask *postgres.ExportTaskStore + ExportShardTask *postgres.ExportShardTaskStore Device *postgres.DeviceStore DeviceSimBinding *postgres.DeviceSimBindingStore ShopSeriesCommissionStats *postgres.ShopSeriesCommissionStatsStore