| 614 | } |
| 615 | |
| 616 | func (s *SkillLibraryService) claimPendingScanJob(skillID *int64) (*model.SkillScanJob, error) { |
| 617 | if model.DB == nil { |
| 618 | return nil, nil |
| 619 | } |
| 620 | var claimed model.SkillScanJob |
| 621 | err := model.DB.Transaction(func(tx *gorm.DB) error { |
| 622 | query := tx.Model(&model.SkillScanJob{}). |
| 623 | Where("status = ?", model.SkillScanJobStatusPending) |
| 624 | if skillID != nil { |
| 625 | query = query.Where("skill_library_id = ?", *skillID) |
| 626 | } |
| 627 | |
| 628 | if err := query.Order("created_time ASC").Order("id ASC").First(&claimed).Error; err != nil { |
| 629 | if errors.Is(err, gorm.ErrRecordNotFound) { |
| 630 | return nil |
| 631 | } |
| 632 | return err |
| 633 | } |
| 634 | |
| 635 | now := s.nowFunc().UTC().UnixMilli() |
| 636 | updateTx := tx.Model(&model.SkillScanJob{}). |
| 637 | Where("id = ? AND status = ?", claimed.ID, model.SkillScanJobStatusPending). |
| 638 | Updates(map[string]interface{}{ |
| 639 | "status": model.SkillScanJobStatusRunning, |
| 640 | "started_time": now, |
| 641 | "finished_time": int64(0), |
| 642 | }) |
| 643 | if updateTx.Error != nil { |
| 644 | return updateTx.Error |
| 645 | } |
| 646 | if updateTx.RowsAffected == 0 { |
| 647 | claimed.ID = 0 |
| 648 | return nil |
| 649 | } |
| 650 | claimed.Status = model.SkillScanJobStatusRunning |
| 651 | claimed.StartedTime = now |
| 652 | return nil |
| 653 | }) |
| 654 | if err != nil { |
| 655 | return nil, err |
| 656 | } |
| 657 | if claimed.ID == 0 { |
| 658 | return nil, nil |
| 659 | } |
| 660 | return &claimed, nil |
| 661 | } |
| 662 | |
| 663 | func (s *SkillLibraryService) claimPendingScanJobByID(jobID int64) (*model.SkillScanJob, error) { |
| 664 | if model.DB == nil || jobID <= 0 { |