dataDownloadControllers.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685
  1. package controllers
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "path"
  7. "strconv"
  8. "strings"
  9. "time"
  10. "fwh-toolbox-server/cache"
  11. "fwh-toolbox-server/domain"
  12. "fwh-toolbox-server/middleware"
  13. minioRepo "fwh-toolbox-server/repositories/minio"
  14. mongoRepo "fwh-toolbox-server/repositories/mongo"
  15. "fwh-toolbox-server/services"
  16. "fwh-toolbox-server/task/queue/dataDownloadQueue"
  17. "github.com/gofiber/fiber/v2"
  18. "go.mongodb.org/mongo-driver/bson"
  19. "go.mongodb.org/mongo-driver/bson/primitive"
  20. "go.uber.org/zap"
  21. )
  22. type dataDownloadLockedReader struct {
  23. io.ReadCloser
  24. lock *cache.RedisLock
  25. cancel context.CancelFunc
  26. done chan struct{}
  27. }
  28. func (r *dataDownloadLockedReader) Close() error {
  29. r.cancel()
  30. <-r.done
  31. err := r.ReadCloser.Close()
  32. r.lock.SafeRelease(context.Background())
  33. return err
  34. }
  35. func newDataDownloadLockedReader(reader io.ReadCloser, lock *cache.RedisLock) *dataDownloadLockedReader {
  36. ctx, cancel := context.WithCancel(context.Background())
  37. result := &dataDownloadLockedReader{ReadCloser: reader, lock: lock, cancel: cancel, done: make(chan struct{})}
  38. go func() {
  39. defer close(result.done)
  40. ticker := time.NewTicker(3 * time.Minute)
  41. defer ticker.Stop()
  42. for {
  43. select {
  44. case <-ctx.Done():
  45. return
  46. case <-ticker.C:
  47. renewCtx, renewCancel := context.WithTimeout(context.Background(), 10*time.Second)
  48. _ = lock.Extend(renewCtx, 10*time.Minute)
  49. renewCancel()
  50. }
  51. }
  52. }()
  53. return result
  54. }
  55. func CreateDataDownload(c *fiber.Ctx) error {
  56. var body struct {
  57. SNs []string `json:"sns"`
  58. SNText string `json:"snText"`
  59. Description string `json:"description"`
  60. StartDate string `json:"startDate"`
  61. EndDate string `json:"endDate"`
  62. Options domain.DataDownloadOptions `json:"options"`
  63. }
  64. if err := c.BodyParser(&body); err != nil {
  65. return Fail(c, "invalid request body")
  66. }
  67. if body.SNText != "" {
  68. body.SNs = append(body.SNs, strings.FieldsFunc(body.SNText, func(r rune) bool { return r == ',' || r == '\n' || r == '\r' || r == '\t' || r == ' ' })...)
  69. }
  70. user := middleware.AuthGetUserInfo(c)
  71. ownerID := user.Id
  72. if ownerID == "" {
  73. ownerID = user.Name
  74. }
  75. task := &domain.DataDownloadTask{OwnerID: ownerID, OwnerName: user.DisplayName, SNs: body.SNs, Description: body.Description, StartDate: body.StartDate, EndDate: body.EndDate, Options: body.Options}
  76. if err := services.CreateDataDownloadTask(c.UserContext(), task); err != nil {
  77. return Fail(c, err.Error())
  78. }
  79. if err := dataDownloadQueue.EnqueuePlan(task.ID); err != nil {
  80. _ = mongoRepo.UpdateDataDownloadTaskStatus(context.Background(), task.ID, domain.DataDownloadError)
  81. return Error(c, "task created but could not be queued: "+err.Error())
  82. }
  83. zap.L().Info("[DataDownload] API 创建任务",
  84. zap.String("taskId", task.ID.Hex()),
  85. zap.String("owner", ownerID),
  86. zap.Int("snCount", len(task.SNs)),
  87. zap.String("startDate", task.StartDate),
  88. zap.String("endDate", task.EndDate),
  89. zap.Any("options", task.Options),
  90. )
  91. return Success(c, task)
  92. }
  93. func GetDataDownloadDocumentFile(c *fiber.Ctx) error {
  94. if !middleware.AuthIsAdmin(c) {
  95. return Forbidden(c)
  96. }
  97. return sendDataDownloadDocumentFile(c, c.Params("documentId"))
  98. }
  99. func DownloadDataDownloadTaskDocument(c *fiber.Ctx) error {
  100. taskID, err := primitive.ObjectIDFromHex(c.Params("id"))
  101. if err != nil {
  102. return Fail(c, "invalid task id")
  103. }
  104. task, err := mongoRepo.GetDataDownloadTask(c.UserContext(), taskID)
  105. if err != nil {
  106. return NotFound(c)
  107. }
  108. if !canAccessDataDownloadTask(c, task) {
  109. return Forbidden(c)
  110. }
  111. documentID := c.Params("documentId")
  112. linked, err := mongoRepo.HasDataDownloadTaskDocument(c.UserContext(), taskID, documentID)
  113. if err != nil {
  114. return Error(c, err.Error())
  115. }
  116. if !linked {
  117. return NotFound(c)
  118. }
  119. return sendDataDownloadDocumentFile(c, documentID)
  120. }
  121. func sendDataDownloadDocumentFile(c *fiber.Ctx, documentID string) error {
  122. document, err := mongoRepo.GetDataDownloadDocument(c.UserContext(), documentID)
  123. if err != nil {
  124. return NotFound(c)
  125. }
  126. if document.BinaryObjectKey != "" {
  127. lock, err := cache.Lock(c.UserContext(), "data-download:content:"+document.ID, 10*time.Minute, 100*time.Millisecond, 10*time.Second)
  128. if err != nil {
  129. return Error(c, "document is being updated")
  130. }
  131. document, err = mongoRepo.GetDataDownloadDocument(c.UserContext(), document.ID)
  132. if err != nil {
  133. lock.SafeRelease(context.Background())
  134. return NotFound(c)
  135. }
  136. reader, err := minioRepo.Open(c.UserContext(), document.BinaryObjectKey)
  137. if err != nil {
  138. lock.SafeRelease(context.Background())
  139. return Error(c, err.Error())
  140. }
  141. c.Set("Content-Disposition", "attachment; filename=\""+path.Base(document.BinaryObjectKey)+"\"")
  142. return c.SendStream(newDataDownloadLockedReader(reader, lock))
  143. }
  144. if document.Kind == domain.DataDownloadFaultRecord {
  145. return Fail(c, "fault-record documents contain multiple files; select a file entry")
  146. }
  147. artifact, reader, err := services.OpenRenderedDataDownloadDocument(c.UserContext(), documentID, c.Query("format"))
  148. if err != nil {
  149. return Fail(c, err.Error())
  150. }
  151. c.Set("Content-Disposition", "attachment; filename=\""+artifact.Filename+"\"")
  152. switch artifact.Format {
  153. case "json":
  154. c.Type("json")
  155. case "html":
  156. c.Type("html")
  157. case "csv":
  158. c.Type("csv")
  159. case "xlsx":
  160. c.Type("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
  161. }
  162. return c.SendStream(reader)
  163. }
  164. func GetDataDownloadDocuments(c *fiber.Ctx) error {
  165. if !middleware.AuthIsAdmin(c) {
  166. return Forbidden(c)
  167. }
  168. page, _ := strconv.ParseInt(c.Query("pageNum", "1"), 10, 64)
  169. pageSize, _ := strconv.ParseInt(c.Query("pageSize", "20"), 10, 64)
  170. if page < 1 || pageSize < 1 || pageSize > 200 {
  171. return Fail(c, "invalid pagination")
  172. }
  173. documents, total, err := mongoRepo.ListDataDownloadDocuments(c.UserContext(), page, pageSize)
  174. if err != nil {
  175. return Error(c, err.Error())
  176. }
  177. return Success(c, fiber.Map{"items": documents, "total": total, "pageNum": page, "pageSize": pageSize})
  178. }
  179. func GetDataDownloadDocumentReferences(c *fiber.Ctx) error {
  180. if !middleware.AuthIsAdmin(c) {
  181. return Forbidden(c)
  182. }
  183. count, err := mongoRepo.CountDataDownloadDocumentReferences(c.UserContext(), c.Params("documentId"))
  184. if err != nil {
  185. return Error(c, err.Error())
  186. }
  187. return Success(c, fiber.Map{"count": count})
  188. }
  189. func GetDataDownloadDocumentFiles(c *fiber.Ctx) error {
  190. if !middleware.AuthIsAdmin(c) {
  191. return Forbidden(c)
  192. }
  193. entries, err := mongoRepo.ListDataDownloadFaultRecordEntries(c.UserContext(), c.Params("documentId"))
  194. if err != nil {
  195. return Error(c, err.Error())
  196. }
  197. return Success(c, entries)
  198. }
  199. func DownloadDataDownloadDocumentEntry(c *fiber.Ctx) error {
  200. if !middleware.AuthIsAdmin(c) {
  201. return Forbidden(c)
  202. }
  203. entryID, err := primitive.ObjectIDFromHex(c.Params("entryId"))
  204. if err != nil {
  205. return Fail(c, "invalid file id")
  206. }
  207. entry, err := mongoRepo.GetDataDownloadFaultRecordEntryByID(c.UserContext(), entryID)
  208. if err != nil || entry.DocumentID != c.Params("documentId") {
  209. return NotFound(c)
  210. }
  211. return sendDataDownloadFaultRecordEntry(c, entry)
  212. }
  213. func DeleteDataDownloadDocument(c *fiber.Ctx) error {
  214. if !middleware.AuthIsAdmin(c) {
  215. return Forbidden(c)
  216. }
  217. if err := services.DeleteDataDownloadDocument(c.UserContext(), c.Params("documentId"), c.Query("force") == "true"); err != nil {
  218. return Fail(c, err.Error())
  219. }
  220. return NoContent(c)
  221. }
  222. func GetDataDownloadArchiveEntries(c *fiber.Ctx) error {
  223. if !middleware.AuthIsAdmin(c) {
  224. return Forbidden(c)
  225. }
  226. entries, err := services.ListDataDownloadArchive(c.UserContext(), c.Params("documentId"))
  227. if err != nil {
  228. return Fail(c, err.Error())
  229. }
  230. return Success(c, entries)
  231. }
  232. func GetDataDownloadArchiveEntry(c *fiber.Ctx) error {
  233. if !middleware.AuthIsAdmin(c) {
  234. return Forbidden(c)
  235. }
  236. entry := c.Query("entry")
  237. if entry == "" {
  238. return Fail(c, "entry is required")
  239. }
  240. data, err := services.ReadDataDownloadArchiveEntry(c.UserContext(), c.Params("documentId"), entry)
  241. if err != nil {
  242. return Fail(c, err.Error())
  243. }
  244. c.Set("Content-Disposition", "attachment; filename=\""+path.Base(entry)+"\"")
  245. return c.Send(data)
  246. }
  247. func GetDataDownloadTasks(c *fiber.Ctx) error {
  248. page, _ := strconv.ParseInt(c.Query("pageNum", "1"), 10, 64)
  249. pageSize, _ := strconv.ParseInt(c.Query("pageSize", "20"), 10, 64)
  250. if page < 1 || pageSize < 1 || pageSize > 200 {
  251. return Fail(c, "invalid pagination")
  252. }
  253. user := middleware.AuthGetUserInfo(c)
  254. ownerID := user.Id
  255. if ownerID == "" {
  256. ownerID = user.Name
  257. }
  258. parseTime := func(value string) *time.Time {
  259. if value == "" {
  260. return nil
  261. }
  262. parsed, err := time.Parse(time.RFC3339, value)
  263. if err != nil {
  264. return nil
  265. }
  266. return &parsed
  267. }
  268. mineOnly := !user.IsAdmin || c.Query("scope", "mine") == "mine"
  269. 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)
  270. if err != nil {
  271. return Error(c, err.Error())
  272. }
  273. return Success(c, fiber.Map{"items": tasks, "total": total, "pageNum": page, "pageSize": pageSize})
  274. }
  275. func GetDataDownloadTask(c *fiber.Ctx) error {
  276. id, err := primitive.ObjectIDFromHex(c.Params("id"))
  277. if err != nil {
  278. return Fail(c, "invalid task id")
  279. }
  280. task, err := mongoRepo.GetDataDownloadTask(c.UserContext(), id)
  281. if err != nil {
  282. return NotFound(c)
  283. }
  284. if !canAccessDataDownloadTask(c, task) {
  285. return Forbidden(c)
  286. }
  287. items, err := mongoRepo.ListDataDownloadItems(c.UserContext(), id)
  288. if err != nil {
  289. return Error(c, err.Error())
  290. }
  291. documents, err := mongoRepo.ListDataDownloadTaskDocuments(c.UserContext(), id)
  292. if err != nil {
  293. return Error(c, err.Error())
  294. }
  295. catalog := make([]*domain.DataDownloadDocument, 0, len(documents))
  296. faultFiles := make(map[string][]domain.DataDownloadFaultRecordEntry)
  297. for i := range documents {
  298. if document, getErr := mongoRepo.GetDataDownloadDocument(c.UserContext(), documents[i].DocumentID); getErr == nil {
  299. catalog = append(catalog, document)
  300. if document.Kind == domain.DataDownloadFaultRecord {
  301. if entries, listErr := mongoRepo.ListDataDownloadFaultRecordEntries(c.UserContext(), document.ID); listErr == nil {
  302. faultFiles[document.ID] = entries
  303. }
  304. }
  305. }
  306. }
  307. attempts, err := mongoRepo.ListDataDownloadFileAttempts(c.UserContext(), id)
  308. if err != nil {
  309. return Error(c, err.Error())
  310. }
  311. return Success(c, fiber.Map{"task": task, "items": items, "documents": documents, "catalog": catalog, "faultFiles": faultFiles, "fileAttempts": attempts})
  312. }
  313. func DownloadDataDownloadTaskFile(c *fiber.Ctx) error {
  314. taskID, err := primitive.ObjectIDFromHex(c.Params("id"))
  315. if err != nil {
  316. return Fail(c, "invalid task id")
  317. }
  318. task, err := mongoRepo.GetDataDownloadTask(c.UserContext(), taskID)
  319. if err != nil {
  320. return NotFound(c)
  321. }
  322. if !canAccessDataDownloadTask(c, task) {
  323. return Forbidden(c)
  324. }
  325. entryID, err := primitive.ObjectIDFromHex(c.Params("entryId"))
  326. if err != nil {
  327. return Fail(c, "invalid file id")
  328. }
  329. entry, err := mongoRepo.GetDataDownloadFaultRecordEntryByID(c.UserContext(), entryID)
  330. if err != nil {
  331. return NotFound(c)
  332. }
  333. linked, err := mongoRepo.HasDataDownloadTaskDocument(c.UserContext(), taskID, entry.DocumentID)
  334. if err != nil {
  335. return Error(c, err.Error())
  336. }
  337. if !linked {
  338. return NotFound(c)
  339. }
  340. return sendDataDownloadFaultRecordEntry(c, entry)
  341. }
  342. func sendDataDownloadFaultRecordEntry(c *fiber.Ctx, entry *domain.DataDownloadFaultRecordEntry) error {
  343. lock, err := cache.Lock(c.UserContext(), "data-download:content:"+entry.DocumentID, 10*time.Minute, 100*time.Millisecond, 10*time.Second)
  344. if err != nil {
  345. return Error(c, "document is being updated")
  346. }
  347. entry, err = mongoRepo.GetDataDownloadFaultRecordEntryByID(c.UserContext(), entry.ID)
  348. if err != nil {
  349. lock.SafeRelease(context.Background())
  350. return NotFound(c)
  351. }
  352. reader, err := minioRepo.Open(c.UserContext(), entry.ObjectKey)
  353. if err != nil {
  354. lock.SafeRelease(context.Background())
  355. return Error(c, err.Error())
  356. }
  357. c.Set("Content-Disposition", "attachment; filename=\""+path.Base(entry.FileName)+"\"")
  358. if entry.ContentType != "" {
  359. c.Set("Content-Type", entry.ContentType)
  360. }
  361. return c.SendStream(newDataDownloadLockedReader(reader, lock))
  362. }
  363. func UpdateDataDownloadDescription(c *fiber.Ctx) error {
  364. id, task, ok := mutableDataDownloadTask(c)
  365. if !ok {
  366. return nil
  367. }
  368. var body struct {
  369. Description string `json:"description"`
  370. }
  371. if err := c.BodyParser(&body); err != nil {
  372. return Fail(c, "invalid request body")
  373. }
  374. _ = task
  375. if err := mongoRepo.UpdateDataDownloadDescription(c.UserContext(), id, body.Description); err != nil {
  376. return Error(c, err.Error())
  377. }
  378. return Success(c, body)
  379. }
  380. func CancelDataDownload(c *fiber.Ctx) error {
  381. id, task, ok := mutableDataDownloadTask(c)
  382. if !ok {
  383. return nil
  384. }
  385. if task.Status != domain.DataDownloadPending && task.Status != domain.DataDownloadRunning {
  386. return Fail(c, "only pending or running tasks can be cancelled")
  387. }
  388. if err := mongoRepo.UpdateDataDownloadTaskStatus(c.UserContext(), id, domain.DataDownloadCancelling); err != nil {
  389. return Error(c, err.Error())
  390. }
  391. zap.L().Info("[DataDownload] API 取消任务", zap.String("taskId", id.Hex()))
  392. return Success(c, fiber.Map{"id": id.Hex(), "status": domain.DataDownloadCancelling})
  393. }
  394. func RetryDataDownload(c *fiber.Ctx) error {
  395. id, task, ok := mutableDataDownloadTask(c)
  396. if !ok {
  397. return nil
  398. }
  399. if task.Status == domain.DataDownloadCancelled {
  400. return Fail(c, "cancelled tasks cannot be retried")
  401. }
  402. if task.Status != domain.DataDownloadFailed && task.Status != domain.DataDownloadPartial && task.Status != domain.DataDownloadError {
  403. return Fail(c, "task has no retryable failures")
  404. }
  405. items, err := mongoRepo.ListDataDownloadItems(c.UserContext(), id)
  406. if err != nil {
  407. return Error(c, err.Error())
  408. }
  409. queued := 0
  410. for i := range items {
  411. if items[i].Status != domain.DataDownloadFailed && items[i].Status != domain.DataDownloadPartial {
  412. continue
  413. }
  414. if items[i].Kind == domain.DataDownloadFaultRecord && items[i].CallRequestedAt != nil {
  415. if err := mongoRepo.ResetDataDownloadFaultCall(c.UserContext(), items[i].ID); err != nil {
  416. continue
  417. }
  418. }
  419. if err := mongoRepo.UpdateDataDownloadItem(c.UserContext(), items[i].ID, domain.DataDownloadPending, "", ""); err == nil {
  420. if err := dataDownloadQueue.EnqueueItem(items[i].ID); err == nil {
  421. queued++
  422. }
  423. }
  424. }
  425. if queued == 0 {
  426. return Fail(c, "no retryable item could be queued")
  427. }
  428. _ = mongoRepo.UpdateDataDownloadTaskStatus(c.UserContext(), id, domain.DataDownloadRunning)
  429. zap.L().Info("[DataDownload] API 重试任务", zap.String("taskId", id.Hex()), zap.Int("queued", queued))
  430. return Success(c, fiber.Map{"queued": queued})
  431. }
  432. func DeleteDataDownload(c *fiber.Ctx) error {
  433. id, task, ok := mutableDataDownloadTask(c)
  434. if !ok {
  435. return nil
  436. }
  437. if task.Status == domain.DataDownloadPending || task.Status == domain.DataDownloadRunning || task.Status == domain.DataDownloadCancelling {
  438. return Fail(c, "active task must be cancelled before deletion")
  439. }
  440. if err := mongoRepo.DeleteDataDownloadTask(c.UserContext(), id); err != nil {
  441. return Error(c, err.Error())
  442. }
  443. zap.L().Info("[DataDownload] API 删除任务", zap.String("taskId", id.Hex()))
  444. return NoContent(c)
  445. }
  446. func mutableDataDownloadTask(c *fiber.Ctx) (primitive.ObjectID, *domain.DataDownloadTask, bool) {
  447. id, err := primitive.ObjectIDFromHex(c.Params("id"))
  448. if err != nil {
  449. _ = Fail(c, "invalid task id")
  450. return id, nil, false
  451. }
  452. task, err := mongoRepo.GetDataDownloadTask(c.UserContext(), id)
  453. if err != nil {
  454. _ = NotFound(c)
  455. return id, nil, false
  456. }
  457. user := middleware.AuthGetUserInfo(c)
  458. ownerID := user.Id
  459. if ownerID == "" {
  460. ownerID = user.Name
  461. }
  462. if !user.IsAdmin && task.OwnerID != ownerID {
  463. _ = Forbidden(c)
  464. return id, nil, false
  465. }
  466. return id, task, true
  467. }
  468. func canAccessDataDownloadTask(c *fiber.Ctx, task *domain.DataDownloadTask) bool {
  469. user := middleware.AuthGetUserInfo(c)
  470. ownerID := user.Id
  471. if ownerID == "" {
  472. ownerID = user.Name
  473. }
  474. return user.IsAdmin || task.OwnerID == ownerID
  475. }
  476. func CreateDataDownloadPackage(c *fiber.Ctx) error {
  477. var body struct {
  478. TaskIDs []string `json:"taskIds"`
  479. }
  480. if err := c.BodyParser(&body); err != nil || len(body.TaskIDs) == 0 {
  481. return Fail(c, "at least one taskId is required")
  482. }
  483. ids := make([]primitive.ObjectID, 0, len(body.TaskIDs))
  484. for _, raw := range body.TaskIDs {
  485. id, err := primitive.ObjectIDFromHex(raw)
  486. if err != nil {
  487. return Fail(c, "invalid task id")
  488. }
  489. task, err := mongoRepo.GetDataDownloadTask(c.UserContext(), id)
  490. if err != nil {
  491. return NotFound(c, "task not found")
  492. }
  493. if !canAccessDataDownloadTask(c, task) {
  494. return Forbidden(c)
  495. }
  496. ids = append(ids, id)
  497. }
  498. user := middleware.AuthGetUserInfo(c)
  499. ownerID := user.Id
  500. if ownerID == "" {
  501. ownerID = user.Name
  502. }
  503. now := time.Now().UTC()
  504. prefix := "DataDownload_" + ids[0].Hex()
  505. if len(ids) > 1 {
  506. prefix = "DataDownload_Multiple"
  507. }
  508. 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}
  509. if err := mongoRepo.InsertDataDownloadPackageJob(c.UserContext(), job); err != nil {
  510. return Error(c, err.Error())
  511. }
  512. if err := dataDownloadQueue.EnqueuePackage(job.ID); err != nil {
  513. _ = mongoRepo.UpdateDataDownloadPackageJob(context.Background(), job.ID, bson.M{"status": domain.DataDownloadFailed, "failureReason": err.Error()})
  514. return Error(c, err.Error())
  515. }
  516. zap.L().Info("[DataDownload] API 创建打包任务",
  517. zap.String("jobId", job.ID.Hex()),
  518. zap.Int("taskCount", len(ids)),
  519. zap.String("filename", job.Filename),
  520. )
  521. return Success(c, job)
  522. }
  523. func GetDataDownloadPackages(c *fiber.Ctx) error {
  524. page, _ := strconv.ParseInt(c.Query("pageNum", "1"), 10, 64)
  525. pageSize, _ := strconv.ParseInt(c.Query("pageSize", "20"), 10, 64)
  526. if page < 1 || pageSize < 1 || pageSize > 200 {
  527. return Fail(c, "invalid pagination")
  528. }
  529. user := middleware.AuthGetUserInfo(c)
  530. ownerID := user.Id
  531. if ownerID == "" {
  532. ownerID = user.Name
  533. }
  534. jobs, total, err := mongoRepo.ListDataDownloadPackageJobs(c.UserContext(), ownerID, !user.IsAdmin, page, pageSize)
  535. if err != nil {
  536. return Error(c, err.Error())
  537. }
  538. return Success(c, fiber.Map{"items": jobs, "total": total, "pageNum": page, "pageSize": pageSize})
  539. }
  540. func DownloadDataDownloadPackage(c *fiber.Ctx) error {
  541. id, job, ok := ownedDataDownloadPackage(c)
  542. if !ok {
  543. return nil
  544. }
  545. lock, err := cache.Lock(c.UserContext(), "data-download:package:"+id.Hex(), 10*time.Minute, 100*time.Millisecond, 10*time.Second)
  546. if err != nil {
  547. return Error(c, "package is being updated")
  548. }
  549. job, err = mongoRepo.GetDataDownloadPackageJob(c.UserContext(), id)
  550. if err != nil {
  551. lock.SafeRelease(context.Background())
  552. return NotFound(c)
  553. }
  554. if job.Status != domain.DataDownloadSucceeded || job.ExpiresAt == nil || !job.ExpiresAt.After(time.Now()) {
  555. lock.SafeRelease(context.Background())
  556. return Fail(c, "package is not available")
  557. }
  558. reader, err := minioRepo.Open(c.UserContext(), job.ObjectKey)
  559. if err != nil {
  560. lock.SafeRelease(context.Background())
  561. return Error(c, err.Error())
  562. }
  563. c.Type("zip")
  564. c.Set("Content-Disposition", "attachment; filename=\""+job.Filename+"\"")
  565. return c.SendStream(newDataDownloadLockedReader(reader, lock))
  566. }
  567. func CancelDataDownloadPackage(c *fiber.Ctx) error {
  568. id, job, ok := ownedDataDownloadPackage(c)
  569. if !ok {
  570. return nil
  571. }
  572. if job.Status != domain.DataDownloadPending && job.Status != domain.DataDownloadRunning {
  573. return Fail(c, "only pending or running packages can be cancelled")
  574. }
  575. status := domain.DataDownloadCancelling
  576. if job.Status == domain.DataDownloadPending {
  577. status = domain.DataDownloadCancelled
  578. }
  579. if err := mongoRepo.UpdateDataDownloadPackageJob(c.UserContext(), id, bson.M{"status": status}); err != nil {
  580. return Error(c, err.Error())
  581. }
  582. return Success(c, fiber.Map{"status": status})
  583. }
  584. func DeleteDataDownloadPackage(c *fiber.Ctx) error {
  585. id, job, ok := ownedDataDownloadPackage(c)
  586. if !ok {
  587. return nil
  588. }
  589. lock, err := cache.Lock(c.UserContext(), "data-download:package:"+id.Hex(), 5*time.Minute, 100*time.Millisecond, 10*time.Second)
  590. if err != nil {
  591. return Error(c, "package is being downloaded or updated")
  592. }
  593. defer lock.SafeRelease(context.Background())
  594. job, err = mongoRepo.GetDataDownloadPackageJob(c.UserContext(), id)
  595. if err != nil {
  596. return NotFound(c)
  597. }
  598. if job.Status == domain.DataDownloadPending || job.Status == domain.DataDownloadRunning || job.Status == domain.DataDownloadCancelling {
  599. return Fail(c, "active package must be cancelled first")
  600. }
  601. if job.ObjectKey != "" {
  602. if err := minioRepo.Delete(c.UserContext(), job.ObjectKey); err != nil {
  603. return Error(c, err.Error())
  604. }
  605. }
  606. if err := mongoRepo.DeleteDataDownloadPackageJob(c.UserContext(), id); err != nil {
  607. return Error(c, err.Error())
  608. }
  609. return NoContent(c)
  610. }
  611. func RetryDataDownloadPackage(c *fiber.Ctx) error {
  612. _, original, ok := ownedDataDownloadPackage(c)
  613. if !ok {
  614. return nil
  615. }
  616. if original.Status != domain.DataDownloadFailed {
  617. return Fail(c, "only failed packages can be retried")
  618. }
  619. now := time.Now().UTC()
  620. 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}
  621. if err := mongoRepo.InsertDataDownloadPackageJob(c.UserContext(), job); err != nil {
  622. return Error(c, err.Error())
  623. }
  624. if err := dataDownloadQueue.EnqueuePackage(job.ID); err != nil {
  625. _ = mongoRepo.UpdateDataDownloadPackageJob(context.Background(), job.ID, bson.M{"status": domain.DataDownloadFailed, "failureReason": err.Error()})
  626. return Error(c, err.Error())
  627. }
  628. return Success(c, job)
  629. }
  630. func ownedDataDownloadPackage(c *fiber.Ctx) (primitive.ObjectID, *domain.DataDownloadPackageJob, bool) {
  631. id, err := primitive.ObjectIDFromHex(c.Params("id"))
  632. if err != nil {
  633. _ = Fail(c, "invalid package id")
  634. return id, nil, false
  635. }
  636. job, err := mongoRepo.GetDataDownloadPackageJob(c.UserContext(), id)
  637. if err != nil {
  638. _ = NotFound(c)
  639. return id, nil, false
  640. }
  641. user := middleware.AuthGetUserInfo(c)
  642. ownerID := user.Id
  643. if ownerID == "" {
  644. ownerID = user.Name
  645. }
  646. if !user.IsAdmin && job.OwnerID != ownerID {
  647. _ = Forbidden(c)
  648. return id, nil, false
  649. }
  650. return id, job, true
  651. }