package controllers import ( "context" "fmt" "io" "path" "strconv" "strings" "time" "fwh-toolbox-server/cache" "fwh-toolbox-server/domain" "fwh-toolbox-server/middleware" minioRepo "fwh-toolbox-server/repositories/minio" mongoRepo "fwh-toolbox-server/repositories/mongo" "fwh-toolbox-server/services" "fwh-toolbox-server/task/queue/dataDownloadQueue" "github.com/gofiber/fiber/v2" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" "go.uber.org/zap" ) type dataDownloadLockedReader struct { io.ReadCloser lock *cache.RedisLock cancel context.CancelFunc done chan struct{} } func (r *dataDownloadLockedReader) Close() error { r.cancel() <-r.done err := r.ReadCloser.Close() r.lock.SafeRelease(context.Background()) return err } func newDataDownloadLockedReader(reader io.ReadCloser, lock *cache.RedisLock) *dataDownloadLockedReader { ctx, cancel := context.WithCancel(context.Background()) result := &dataDownloadLockedReader{ReadCloser: reader, lock: lock, cancel: cancel, done: make(chan struct{})} go func() { defer close(result.done) ticker := time.NewTicker(3 * time.Minute) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: renewCtx, renewCancel := context.WithTimeout(context.Background(), 10*time.Second) _ = lock.Extend(renewCtx, 10*time.Minute) renewCancel() } } }() return result } func CreateDataDownload(c *fiber.Ctx) error { var body struct { SNs []string `json:"sns"` SNText string `json:"snText"` Description string `json:"description"` StartDate string `json:"startDate"` EndDate string `json:"endDate"` Options domain.DataDownloadOptions `json:"options"` } if err := c.BodyParser(&body); err != nil { return Fail(c, "invalid request body") } if body.SNText != "" { body.SNs = append(body.SNs, strings.FieldsFunc(body.SNText, func(r rune) bool { return r == ',' || r == '\n' || r == '\r' || r == '\t' || r == ' ' })...) } user := middleware.AuthGetUserInfo(c) ownerID := user.Id if ownerID == "" { ownerID = user.Name } task := &domain.DataDownloadTask{OwnerID: ownerID, OwnerName: user.DisplayName, SNs: body.SNs, Description: body.Description, StartDate: body.StartDate, EndDate: body.EndDate, Options: body.Options} if err := services.CreateDataDownloadTask(c.UserContext(), task); err != nil { return Fail(c, err.Error()) } if err := dataDownloadQueue.EnqueuePlan(task.ID); err != nil { _ = mongoRepo.UpdateDataDownloadTaskStatus(context.Background(), task.ID, domain.DataDownloadError) return Error(c, "task created but could not be queued: "+err.Error()) } zap.L().Info("[DataDownload] API 创建任务", zap.String("taskId", task.ID.Hex()), zap.String("owner", ownerID), zap.Int("snCount", len(task.SNs)), zap.String("startDate", task.StartDate), zap.String("endDate", task.EndDate), zap.Any("options", task.Options), ) return Success(c, task) } func GetDataDownloadDocumentFile(c *fiber.Ctx) error { if !middleware.AuthIsAdmin(c) { return Forbidden(c) } return sendDataDownloadDocumentFile(c, c.Params("documentId")) } func DownloadDataDownloadTaskDocument(c *fiber.Ctx) error { taskID, err := primitive.ObjectIDFromHex(c.Params("id")) if err != nil { return Fail(c, "invalid task id") } task, err := mongoRepo.GetDataDownloadTask(c.UserContext(), taskID) if err != nil { return NotFound(c) } if !canAccessDataDownloadTask(c, task) { return Forbidden(c) } documentID := c.Params("documentId") linked, err := mongoRepo.HasDataDownloadTaskDocument(c.UserContext(), taskID, documentID) if err != nil { return Error(c, err.Error()) } if !linked { return NotFound(c) } return sendDataDownloadDocumentFile(c, documentID) } func sendDataDownloadDocumentFile(c *fiber.Ctx, documentID string) error { document, err := mongoRepo.GetDataDownloadDocument(c.UserContext(), documentID) if err != nil { return NotFound(c) } if document.BinaryObjectKey != "" { lock, err := cache.Lock(c.UserContext(), "data-download:content:"+document.ID, 10*time.Minute, 100*time.Millisecond, 10*time.Second) if err != nil { return Error(c, "document is being updated") } document, err = mongoRepo.GetDataDownloadDocument(c.UserContext(), document.ID) if err != nil { lock.SafeRelease(context.Background()) return NotFound(c) } reader, err := minioRepo.Open(c.UserContext(), document.BinaryObjectKey) if err != nil { lock.SafeRelease(context.Background()) return Error(c, err.Error()) } c.Set("Content-Disposition", "attachment; filename=\""+path.Base(document.BinaryObjectKey)+"\"") return c.SendStream(newDataDownloadLockedReader(reader, lock)) } if document.Kind == domain.DataDownloadFaultRecord { return Fail(c, "fault-record documents contain multiple files; select a file entry") } artifact, reader, err := services.OpenRenderedDataDownloadDocument(c.UserContext(), documentID, c.Query("format")) if err != nil { return Fail(c, err.Error()) } c.Set("Content-Disposition", "attachment; filename=\""+artifact.Filename+"\"") switch artifact.Format { case "json": c.Type("json") case "html": c.Type("html") case "csv": c.Type("csv") case "xlsx": c.Type("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") } return c.SendStream(reader) } func GetDataDownloadDocuments(c *fiber.Ctx) error { if !middleware.AuthIsAdmin(c) { return Forbidden(c) } page, _ := strconv.ParseInt(c.Query("pageNum", "1"), 10, 64) pageSize, _ := strconv.ParseInt(c.Query("pageSize", "20"), 10, 64) if page < 1 || pageSize < 1 || pageSize > 200 { return Fail(c, "invalid pagination") } documents, total, err := mongoRepo.ListDataDownloadDocuments(c.UserContext(), page, pageSize) if err != nil { return Error(c, err.Error()) } return Success(c, fiber.Map{"items": documents, "total": total, "pageNum": page, "pageSize": pageSize}) } func GetDataDownloadDocumentReferences(c *fiber.Ctx) error { if !middleware.AuthIsAdmin(c) { return Forbidden(c) } count, err := mongoRepo.CountDataDownloadDocumentReferences(c.UserContext(), c.Params("documentId")) if err != nil { return Error(c, err.Error()) } return Success(c, fiber.Map{"count": count}) } func GetDataDownloadDocumentFiles(c *fiber.Ctx) error { if !middleware.AuthIsAdmin(c) { return Forbidden(c) } entries, err := mongoRepo.ListDataDownloadFaultRecordEntries(c.UserContext(), c.Params("documentId")) if err != nil { return Error(c, err.Error()) } return Success(c, entries) } func DownloadDataDownloadDocumentEntry(c *fiber.Ctx) error { if !middleware.AuthIsAdmin(c) { return Forbidden(c) } entryID, err := primitive.ObjectIDFromHex(c.Params("entryId")) if err != nil { return Fail(c, "invalid file id") } entry, err := mongoRepo.GetDataDownloadFaultRecordEntryByID(c.UserContext(), entryID) if err != nil || entry.DocumentID != c.Params("documentId") { return NotFound(c) } return sendDataDownloadFaultRecordEntry(c, entry) } func DeleteDataDownloadDocument(c *fiber.Ctx) error { if !middleware.AuthIsAdmin(c) { return Forbidden(c) } if err := services.DeleteDataDownloadDocument(c.UserContext(), c.Params("documentId"), c.Query("force") == "true"); err != nil { return Fail(c, err.Error()) } return NoContent(c) } func GetDataDownloadArchiveEntries(c *fiber.Ctx) error { if !middleware.AuthIsAdmin(c) { return Forbidden(c) } entries, err := services.ListDataDownloadArchive(c.UserContext(), c.Params("documentId")) if err != nil { return Fail(c, err.Error()) } return Success(c, entries) } func GetDataDownloadArchiveEntry(c *fiber.Ctx) error { if !middleware.AuthIsAdmin(c) { return Forbidden(c) } entry := c.Query("entry") if entry == "" { return Fail(c, "entry is required") } data, err := services.ReadDataDownloadArchiveEntry(c.UserContext(), c.Params("documentId"), entry) if err != nil { return Fail(c, err.Error()) } c.Set("Content-Disposition", "attachment; filename=\""+path.Base(entry)+"\"") return c.Send(data) } func GetDataDownloadTasks(c *fiber.Ctx) error { page, _ := strconv.ParseInt(c.Query("pageNum", "1"), 10, 64) pageSize, _ := strconv.ParseInt(c.Query("pageSize", "20"), 10, 64) if page < 1 || pageSize < 1 || pageSize > 200 { return Fail(c, "invalid pagination") } user := middleware.AuthGetUserInfo(c) ownerID := user.Id if ownerID == "" { ownerID = user.Name } parseTime := func(value string) *time.Time { if value == "" { return nil } parsed, err := time.Parse(time.RFC3339, value) if err != nil { return nil } return &parsed } mineOnly := !user.IsAdmin || c.Query("scope", "mine") == "mine" tasks, total, err := mongoRepo.ListDataDownloadTasks(c.UserContext(), ownerID, mineOnly, c.Query("description"), c.Query("sn"), c.Query("status"), parseTime(c.Query("createdFrom")), parseTime(c.Query("createdTo")), page, pageSize) if err != nil { return Error(c, err.Error()) } return Success(c, fiber.Map{"items": tasks, "total": total, "pageNum": page, "pageSize": pageSize}) } func GetDataDownloadTask(c *fiber.Ctx) error { id, err := primitive.ObjectIDFromHex(c.Params("id")) if err != nil { return Fail(c, "invalid task id") } task, err := mongoRepo.GetDataDownloadTask(c.UserContext(), id) if err != nil { return NotFound(c) } if !canAccessDataDownloadTask(c, task) { return Forbidden(c) } items, err := mongoRepo.ListDataDownloadItems(c.UserContext(), id) if err != nil { return Error(c, err.Error()) } documents, err := mongoRepo.ListDataDownloadTaskDocuments(c.UserContext(), id) if err != nil { return Error(c, err.Error()) } catalog := make([]*domain.DataDownloadDocument, 0, len(documents)) faultFiles := make(map[string][]domain.DataDownloadFaultRecordEntry) for i := range documents { if document, getErr := mongoRepo.GetDataDownloadDocument(c.UserContext(), documents[i].DocumentID); getErr == nil { catalog = append(catalog, document) if document.Kind == domain.DataDownloadFaultRecord { if entries, listErr := mongoRepo.ListDataDownloadFaultRecordEntries(c.UserContext(), document.ID); listErr == nil { faultFiles[document.ID] = entries } } } } attempts, err := mongoRepo.ListDataDownloadFileAttempts(c.UserContext(), id) if err != nil { return Error(c, err.Error()) } return Success(c, fiber.Map{"task": task, "items": items, "documents": documents, "catalog": catalog, "faultFiles": faultFiles, "fileAttempts": attempts}) } func DownloadDataDownloadTaskFile(c *fiber.Ctx) error { taskID, err := primitive.ObjectIDFromHex(c.Params("id")) if err != nil { return Fail(c, "invalid task id") } task, err := mongoRepo.GetDataDownloadTask(c.UserContext(), taskID) if err != nil { return NotFound(c) } if !canAccessDataDownloadTask(c, task) { return Forbidden(c) } entryID, err := primitive.ObjectIDFromHex(c.Params("entryId")) if err != nil { return Fail(c, "invalid file id") } entry, err := mongoRepo.GetDataDownloadFaultRecordEntryByID(c.UserContext(), entryID) if err != nil { return NotFound(c) } linked, err := mongoRepo.HasDataDownloadTaskDocument(c.UserContext(), taskID, entry.DocumentID) if err != nil { return Error(c, err.Error()) } if !linked { return NotFound(c) } return sendDataDownloadFaultRecordEntry(c, entry) } func sendDataDownloadFaultRecordEntry(c *fiber.Ctx, entry *domain.DataDownloadFaultRecordEntry) error { lock, err := cache.Lock(c.UserContext(), "data-download:content:"+entry.DocumentID, 10*time.Minute, 100*time.Millisecond, 10*time.Second) if err != nil { return Error(c, "document is being updated") } entry, err = mongoRepo.GetDataDownloadFaultRecordEntryByID(c.UserContext(), entry.ID) if err != nil { lock.SafeRelease(context.Background()) return NotFound(c) } reader, err := minioRepo.Open(c.UserContext(), entry.ObjectKey) if err != nil { lock.SafeRelease(context.Background()) return Error(c, err.Error()) } c.Set("Content-Disposition", "attachment; filename=\""+path.Base(entry.FileName)+"\"") if entry.ContentType != "" { c.Set("Content-Type", entry.ContentType) } return c.SendStream(newDataDownloadLockedReader(reader, lock)) } func UpdateDataDownloadDescription(c *fiber.Ctx) error { id, task, ok := mutableDataDownloadTask(c) if !ok { return nil } var body struct { Description string `json:"description"` } if err := c.BodyParser(&body); err != nil { return Fail(c, "invalid request body") } _ = task if err := mongoRepo.UpdateDataDownloadDescription(c.UserContext(), id, body.Description); err != nil { return Error(c, err.Error()) } return Success(c, body) } func CancelDataDownload(c *fiber.Ctx) error { id, task, ok := mutableDataDownloadTask(c) if !ok { return nil } if task.Status != domain.DataDownloadPending && task.Status != domain.DataDownloadRunning { return Fail(c, "only pending or running tasks can be cancelled") } if err := mongoRepo.UpdateDataDownloadTaskStatus(c.UserContext(), id, domain.DataDownloadCancelling); err != nil { return Error(c, err.Error()) } zap.L().Info("[DataDownload] API 取消任务", zap.String("taskId", id.Hex())) return Success(c, fiber.Map{"id": id.Hex(), "status": domain.DataDownloadCancelling}) } func RetryDataDownload(c *fiber.Ctx) error { id, task, ok := mutableDataDownloadTask(c) if !ok { return nil } if task.Status == domain.DataDownloadCancelled { return Fail(c, "cancelled tasks cannot be retried") } if task.Status != domain.DataDownloadFailed && task.Status != domain.DataDownloadPartial && task.Status != domain.DataDownloadError { return Fail(c, "task has no retryable failures") } items, err := mongoRepo.ListDataDownloadItems(c.UserContext(), id) if err != nil { return Error(c, err.Error()) } queued := 0 for i := range items { if items[i].Status != domain.DataDownloadFailed && items[i].Status != domain.DataDownloadPartial { continue } if items[i].Kind == domain.DataDownloadFaultRecord && items[i].CallRequestedAt != nil { if err := mongoRepo.ResetDataDownloadFaultCall(c.UserContext(), items[i].ID); err != nil { continue } } if err := mongoRepo.UpdateDataDownloadItem(c.UserContext(), items[i].ID, domain.DataDownloadPending, "", ""); err == nil { if err := dataDownloadQueue.EnqueueItem(items[i].ID); err == nil { queued++ } } } if queued == 0 { return Fail(c, "no retryable item could be queued") } _ = mongoRepo.UpdateDataDownloadTaskStatus(c.UserContext(), id, domain.DataDownloadRunning) zap.L().Info("[DataDownload] API 重试任务", zap.String("taskId", id.Hex()), zap.Int("queued", queued)) return Success(c, fiber.Map{"queued": queued}) } func DeleteDataDownload(c *fiber.Ctx) error { id, task, ok := mutableDataDownloadTask(c) if !ok { return nil } if task.Status == domain.DataDownloadPending || task.Status == domain.DataDownloadRunning || task.Status == domain.DataDownloadCancelling { return Fail(c, "active task must be cancelled before deletion") } if err := mongoRepo.DeleteDataDownloadTask(c.UserContext(), id); err != nil { return Error(c, err.Error()) } zap.L().Info("[DataDownload] API 删除任务", zap.String("taskId", id.Hex())) return NoContent(c) } func mutableDataDownloadTask(c *fiber.Ctx) (primitive.ObjectID, *domain.DataDownloadTask, bool) { id, err := primitive.ObjectIDFromHex(c.Params("id")) if err != nil { _ = Fail(c, "invalid task id") return id, nil, false } task, err := mongoRepo.GetDataDownloadTask(c.UserContext(), id) if err != nil { _ = NotFound(c) return id, nil, false } user := middleware.AuthGetUserInfo(c) ownerID := user.Id if ownerID == "" { ownerID = user.Name } if !user.IsAdmin && task.OwnerID != ownerID { _ = Forbidden(c) return id, nil, false } return id, task, true } func canAccessDataDownloadTask(c *fiber.Ctx, task *domain.DataDownloadTask) bool { user := middleware.AuthGetUserInfo(c) ownerID := user.Id if ownerID == "" { ownerID = user.Name } return user.IsAdmin || task.OwnerID == ownerID } func CreateDataDownloadPackage(c *fiber.Ctx) error { var body struct { TaskIDs []string `json:"taskIds"` } if err := c.BodyParser(&body); err != nil || len(body.TaskIDs) == 0 { return Fail(c, "at least one taskId is required") } ids := make([]primitive.ObjectID, 0, len(body.TaskIDs)) for _, raw := range body.TaskIDs { id, err := primitive.ObjectIDFromHex(raw) if err != nil { return Fail(c, "invalid task id") } task, err := mongoRepo.GetDataDownloadTask(c.UserContext(), id) if err != nil { return NotFound(c, "task not found") } if !canAccessDataDownloadTask(c, task) { return Forbidden(c) } ids = append(ids, id) } user := middleware.AuthGetUserInfo(c) ownerID := user.Id if ownerID == "" { ownerID = user.Name } now := time.Now().UTC() prefix := "DataDownload_" + ids[0].Hex() if len(ids) > 1 { prefix = "DataDownload_Multiple" } job := &domain.DataDownloadPackageJob{TaskIDs: ids, OwnerID: ownerID, Status: domain.DataDownloadPending, Filename: fmt.Sprintf("%s_%s.zip", prefix, now.Format("20060102_150405")), CreatedAt: now, UpdatedAt: now} if err := mongoRepo.InsertDataDownloadPackageJob(c.UserContext(), job); err != nil { return Error(c, err.Error()) } if err := dataDownloadQueue.EnqueuePackage(job.ID); err != nil { _ = mongoRepo.UpdateDataDownloadPackageJob(context.Background(), job.ID, bson.M{"status": domain.DataDownloadFailed, "failureReason": err.Error()}) return Error(c, err.Error()) } zap.L().Info("[DataDownload] API 创建打包任务", zap.String("jobId", job.ID.Hex()), zap.Int("taskCount", len(ids)), zap.String("filename", job.Filename), ) return Success(c, job) } func GetDataDownloadPackages(c *fiber.Ctx) error { page, _ := strconv.ParseInt(c.Query("pageNum", "1"), 10, 64) pageSize, _ := strconv.ParseInt(c.Query("pageSize", "20"), 10, 64) if page < 1 || pageSize < 1 || pageSize > 200 { return Fail(c, "invalid pagination") } user := middleware.AuthGetUserInfo(c) ownerID := user.Id if ownerID == "" { ownerID = user.Name } jobs, total, err := mongoRepo.ListDataDownloadPackageJobs(c.UserContext(), ownerID, !user.IsAdmin, page, pageSize) if err != nil { return Error(c, err.Error()) } return Success(c, fiber.Map{"items": jobs, "total": total, "pageNum": page, "pageSize": pageSize}) } func DownloadDataDownloadPackage(c *fiber.Ctx) error { id, job, ok := ownedDataDownloadPackage(c) if !ok { return nil } lock, err := cache.Lock(c.UserContext(), "data-download:package:"+id.Hex(), 10*time.Minute, 100*time.Millisecond, 10*time.Second) if err != nil { return Error(c, "package is being updated") } job, err = mongoRepo.GetDataDownloadPackageJob(c.UserContext(), id) if err != nil { lock.SafeRelease(context.Background()) return NotFound(c) } if job.Status != domain.DataDownloadSucceeded || job.ExpiresAt == nil || !job.ExpiresAt.After(time.Now()) { lock.SafeRelease(context.Background()) return Fail(c, "package is not available") } reader, err := minioRepo.Open(c.UserContext(), job.ObjectKey) if err != nil { lock.SafeRelease(context.Background()) return Error(c, err.Error()) } c.Type("zip") c.Set("Content-Disposition", "attachment; filename=\""+job.Filename+"\"") return c.SendStream(newDataDownloadLockedReader(reader, lock)) } func CancelDataDownloadPackage(c *fiber.Ctx) error { id, job, ok := ownedDataDownloadPackage(c) if !ok { return nil } if job.Status != domain.DataDownloadPending && job.Status != domain.DataDownloadRunning { return Fail(c, "only pending or running packages can be cancelled") } status := domain.DataDownloadCancelling if job.Status == domain.DataDownloadPending { status = domain.DataDownloadCancelled } if err := mongoRepo.UpdateDataDownloadPackageJob(c.UserContext(), id, bson.M{"status": status}); err != nil { return Error(c, err.Error()) } return Success(c, fiber.Map{"status": status}) } func DeleteDataDownloadPackage(c *fiber.Ctx) error { id, job, ok := ownedDataDownloadPackage(c) if !ok { return nil } lock, err := cache.Lock(c.UserContext(), "data-download:package:"+id.Hex(), 5*time.Minute, 100*time.Millisecond, 10*time.Second) if err != nil { return Error(c, "package is being downloaded or updated") } defer lock.SafeRelease(context.Background()) job, err = mongoRepo.GetDataDownloadPackageJob(c.UserContext(), id) if err != nil { return NotFound(c) } if job.Status == domain.DataDownloadPending || job.Status == domain.DataDownloadRunning || job.Status == domain.DataDownloadCancelling { return Fail(c, "active package must be cancelled first") } if job.ObjectKey != "" { if err := minioRepo.Delete(c.UserContext(), job.ObjectKey); err != nil { return Error(c, err.Error()) } } if err := mongoRepo.DeleteDataDownloadPackageJob(c.UserContext(), id); err != nil { return Error(c, err.Error()) } return NoContent(c) } func RetryDataDownloadPackage(c *fiber.Ctx) error { _, original, ok := ownedDataDownloadPackage(c) if !ok { return nil } if original.Status != domain.DataDownloadFailed { return Fail(c, "only failed packages can be retried") } now := time.Now().UTC() job := &domain.DataDownloadPackageJob{TaskIDs: original.TaskIDs, OwnerID: original.OwnerID, Status: domain.DataDownloadPending, Filename: fmt.Sprintf("DataDownload_Retry_%s.zip", now.Format("20060102_150405")), CreatedAt: now, UpdatedAt: now} if err := mongoRepo.InsertDataDownloadPackageJob(c.UserContext(), job); err != nil { return Error(c, err.Error()) } if err := dataDownloadQueue.EnqueuePackage(job.ID); err != nil { _ = mongoRepo.UpdateDataDownloadPackageJob(context.Background(), job.ID, bson.M{"status": domain.DataDownloadFailed, "failureReason": err.Error()}) return Error(c, err.Error()) } return Success(c, job) } func ownedDataDownloadPackage(c *fiber.Ctx) (primitive.ObjectID, *domain.DataDownloadPackageJob, bool) { id, err := primitive.ObjectIDFromHex(c.Params("id")) if err != nil { _ = Fail(c, "invalid package id") return id, nil, false } job, err := mongoRepo.GetDataDownloadPackageJob(c.UserContext(), id) if err != nil { _ = NotFound(c) return id, nil, false } user := middleware.AuthGetUserInfo(c) ownerID := user.Id if ownerID == "" { ownerID = user.Name } if !user.IsAdmin && job.OwnerID != ownerID { _ = Forbidden(c) return id, nil, false } return id, job, true }