569 lines
18 KiB
Go
569 lines
18 KiB
Go
package usecase
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"apps/backend/internal/module/job/domain"
|
||
|
||
"github.com/google/uuid"
|
||
)
|
||
|
||
// Notifier — 任務狀態寫入鈴鐺（同一 job upsert；含進度）
|
||
type Notifier interface {
|
||
// NotifyJobState templateType/status/summary/percent；progress 與 terminal 都走這條
|
||
NotifyJobState(ctx context.Context, ownerUID int64, jobID, templateType, status, summary string, percent int) error
|
||
}
|
||
|
||
type Service struct {
|
||
Repo domain.Repository
|
||
Notifier Notifier
|
||
LeaseDuration time.Duration
|
||
HeartbeatInterval time.Duration
|
||
|
||
mu sync.Mutex
|
||
workerID string
|
||
heartbeats map[string]context.CancelFunc
|
||
}
|
||
|
||
func New(repo domain.Repository) *Service {
|
||
return &Service{
|
||
Repo: repo, LeaseDuration: domain.DefaultLeaseDuration,
|
||
HeartbeatInterval: domain.DefaultHeartbeat,
|
||
heartbeats: make(map[string]context.CancelFunc),
|
||
}
|
||
}
|
||
|
||
func (s *Service) leaseDuration() time.Duration {
|
||
if s.LeaseDuration > 0 {
|
||
return s.LeaseDuration
|
||
}
|
||
return domain.DefaultLeaseDuration
|
||
}
|
||
|
||
func (s *Service) leaseOwner() string {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
return s.workerID
|
||
}
|
||
|
||
func (s *Service) update(ctx context.Context, j *domain.Job) error {
|
||
if owner := s.leaseOwner(); owner != "" {
|
||
j.LeaseExpiresAt = domain.NowNano() + int64(s.leaseDuration())
|
||
return s.Repo.UpdateOwned(ctx, j, owner)
|
||
}
|
||
return s.Repo.Update(ctx, j)
|
||
}
|
||
|
||
func (s *Service) startHeartbeat(ctx context.Context, jobID, workerID string) {
|
||
heartbeatCtx, cancel := context.WithCancel(ctx)
|
||
s.mu.Lock()
|
||
if previous := s.heartbeats[jobID]; previous != nil {
|
||
previous()
|
||
}
|
||
s.heartbeats[jobID] = cancel
|
||
s.mu.Unlock()
|
||
|
||
go func() {
|
||
interval := s.HeartbeatInterval
|
||
if interval <= 0 {
|
||
interval = domain.DefaultHeartbeat
|
||
}
|
||
ticker := time.NewTicker(interval)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-heartbeatCtx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
expiresAt := domain.NowNano() + int64(s.leaseDuration())
|
||
if err := s.Repo.RenewLease(heartbeatCtx, jobID, workerID, expiresAt); err != nil {
|
||
s.stopHeartbeat(jobID)
|
||
return
|
||
}
|
||
}
|
||
}
|
||
}()
|
||
}
|
||
|
||
func (s *Service) stopHeartbeat(jobID string) {
|
||
s.mu.Lock()
|
||
cancel := s.heartbeats[jobID]
|
||
delete(s.heartbeats, jobID)
|
||
s.mu.Unlock()
|
||
if cancel != nil {
|
||
cancel()
|
||
}
|
||
}
|
||
|
||
// StartDemo — JB-01
|
||
func (s *Service) StartDemo(ctx context.Context, ownerUID int64) (*domain.Job, error) {
|
||
now := domain.NowNano()
|
||
j := &domain.Job{
|
||
ID: uuid.NewString(), OwnerUID: ownerUID,
|
||
TemplateType: domain.TemplateDemo,
|
||
Status: domain.StatusPending,
|
||
ProgressSummary: "Demo 測試任務已建立",
|
||
ProgressPercent: 0,
|
||
CreatedAt: now, UpdatedAt: now,
|
||
}
|
||
if err := s.Repo.Insert(ctx, j); err != nil {
|
||
return nil, err
|
||
}
|
||
j.Status = domain.StatusQueued
|
||
j.ProgressSummary = "Demo 測試任務 · 等待 worker 領取"
|
||
j.UpdatedAt = domain.NowNano()
|
||
_ = s.Repo.Update(ctx, j)
|
||
s.notify(ctx, j)
|
||
return j, nil
|
||
}
|
||
|
||
// ScheduleScoutScan enqueues an immutable Scout brief for the worker.
|
||
func (s *Service) ScheduleScoutScan(ctx context.Context, ownerUID int64, themeKey, payload string) (*domain.Job, error) {
|
||
if strings.TrimSpace(payload) == "" {
|
||
return nil, fmt.Errorf("scout scan payload required")
|
||
}
|
||
now := domain.NowNano()
|
||
j := &domain.Job{ID: uuid.NewString(), OwnerUID: ownerUID, TemplateType: domain.TemplateScoutScan,
|
||
Status: domain.StatusQueued, RefID: themeKey, Payload: payload,
|
||
ProgressSummary: "海巡已排程 · 等待 worker", CreatedAt: now, UpdatedAt: now}
|
||
if err := s.Repo.Insert(ctx, j); err != nil {
|
||
return nil, err
|
||
}
|
||
s.notify(ctx, j)
|
||
return j, nil
|
||
}
|
||
|
||
func (s *Service) List(ctx context.Context, ownerUID int64) ([]*domain.Job, error) {
|
||
return s.Repo.ListByOwner(ctx, ownerUID)
|
||
}
|
||
|
||
// ListTab — 分頁＋分桶（任務頁三頁籤）
|
||
func (s *Service) ListTab(ctx context.Context, ownerUID int64, tab string, page, pageSize int) ([]*domain.Job, int64, error) {
|
||
return s.Repo.ListByOwnerTab(ctx, ownerUID, tab, page, pageSize)
|
||
}
|
||
|
||
// Get — JB-09
|
||
func (s *Service) Get(ctx context.Context, ownerUID int64, id string) (*domain.Job, error) {
|
||
j, err := s.Repo.FindByID(ctx, id)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if j.OwnerUID != ownerUID {
|
||
return nil, domain.ErrForbidden
|
||
}
|
||
return j, nil
|
||
}
|
||
|
||
// Transition — JB-10
|
||
func (s *Service) Transition(ctx context.Context, id, to, summary string, percent int) (*domain.Job, error) {
|
||
j, err := s.Repo.FindByID(ctx, id)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if !domain.CanTransition(j.Status, to) {
|
||
return nil, domain.ErrIllegalStatus
|
||
}
|
||
if percent >= 0 {
|
||
if err := domain.ApplyProgress(j, percent, summary); err != nil {
|
||
return nil, err
|
||
}
|
||
} else if summary != "" {
|
||
j.ProgressSummary = summary
|
||
j.UpdatedAt = domain.NowNano()
|
||
}
|
||
j.Status = to
|
||
if to == domain.StatusSucceeded || to == domain.StatusFailed || to == domain.StatusCancelled {
|
||
j.CompletedAt = domain.NowNano()
|
||
if to == domain.StatusFailed && j.Error == "" {
|
||
j.Error = summary
|
||
if j.Error == "" {
|
||
j.Error = "failed"
|
||
}
|
||
}
|
||
}
|
||
if err := s.update(ctx, j); err != nil {
|
||
return nil, err
|
||
}
|
||
if domain.IsTerminal(to) {
|
||
s.stopHeartbeat(id)
|
||
}
|
||
if s.Notifier != nil && (to == domain.StatusSucceeded || to == domain.StatusFailed || to == domain.StatusCancelled) {
|
||
pct := j.ProgressPercent
|
||
if to == domain.StatusSucceeded {
|
||
pct = 100
|
||
}
|
||
_ = s.Notifier.NotifyJobState(ctx, j.OwnerUID, j.ID, j.TemplateType, to, j.ProgressSummary, pct)
|
||
}
|
||
return j, nil
|
||
}
|
||
|
||
func (s *Service) ClaimNext(ctx context.Context, workerID string) (*domain.Job, error) {
|
||
workerID = strings.TrimSpace(workerID)
|
||
if workerID == "" {
|
||
return nil, domain.ErrForbidden
|
||
}
|
||
j, err := s.Repo.ClaimNext(ctx, workerID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
s.mu.Lock()
|
||
s.workerID = workerID
|
||
s.mu.Unlock()
|
||
// claim → running：立刻寫鈴鐺／列表進度（所有模板共用）
|
||
if j.ProgressPercent < 5 {
|
||
j.ProgressPercent = 5
|
||
}
|
||
if j.ProgressSummary == "" || strings.Contains(j.ProgressSummary, "worker 領取") {
|
||
j.ProgressSummary = "已由 worker 領取 · 執行中"
|
||
}
|
||
j.UpdatedAt = domain.NowNano()
|
||
if err := s.Repo.UpdateOwned(ctx, j, workerID); err != nil {
|
||
return nil, err
|
||
}
|
||
s.startHeartbeat(ctx, j.ID, workerID)
|
||
s.notify(ctx, j)
|
||
return j, nil
|
||
}
|
||
|
||
// RunDemoToSuccess advances running job with monotonic progress then succeeded.
|
||
// 每一步都 MarkRunningProgress → 鈴鐺／前端可即時看到 %。
|
||
func (s *Service) RunDemoToSuccess(ctx context.Context, jobID string) (*domain.Job, error) {
|
||
j, err := s.Repo.FindByID(ctx, jobID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// ensure running（未 claim 時）
|
||
if j.Status == domain.StatusPending || j.Status == domain.StatusQueued {
|
||
if !domain.CanTransition(j.Status, domain.StatusRunning) {
|
||
return nil, domain.ErrIllegalStatus
|
||
}
|
||
j.Status = domain.StatusRunning
|
||
j.ProgressSummary = "Demo 測試任務 · 執行中"
|
||
j.ProgressPercent = 5
|
||
j.UpdatedAt = domain.NowNano()
|
||
_ = s.update(ctx, j)
|
||
s.notify(ctx, j)
|
||
}
|
||
if j.Status != domain.StatusRunning {
|
||
return nil, domain.ErrIllegalStatus
|
||
}
|
||
for _, step := range []struct {
|
||
pct int
|
||
sum string
|
||
}{
|
||
{30, "Demo 測試任務 · 暖機中"},
|
||
{70, "Demo 測試任務 · 處理中"},
|
||
{95, "Demo 測試任務 · 收尾中"},
|
||
} {
|
||
if _, err := s.MarkRunningProgress(ctx, jobID, step.pct, step.sum); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
return s.SucceedJob(ctx, jobID, "Demo 測試任務 · 已完成")
|
||
}
|
||
|
||
// ScheduleTokenRenew creates a delayed job to refresh Threads account token (no browser OAuth).
|
||
// Previous pending renews for the same account are cancelled.
|
||
func (s *Service) ScheduleTokenRenew(ctx context.Context, ownerUID int64, accountID string, runAfter int64) error {
|
||
if ownerUID <= 0 || accountID == "" {
|
||
return domain.ErrForbidden
|
||
}
|
||
now := domain.NowNano()
|
||
if runAfter <= 0 {
|
||
runAfter = now + domain.DefaultTokenRenewDelayNs
|
||
}
|
||
_ = s.Repo.CancelPendingByRef(ctx, ownerUID, domain.TemplateThreadsTokenRenew, accountID)
|
||
j := &domain.Job{
|
||
ID: uuid.NewString(), OwnerUID: ownerUID,
|
||
TemplateType: domain.TemplateThreadsTokenRenew,
|
||
Status: domain.StatusQueued,
|
||
RefID: accountID,
|
||
RunAfter: runAfter,
|
||
ProgressSummary: "定期延長 Threads token（約每 30 天）· 已排程等待執行",
|
||
ProgressPercent: 0,
|
||
CreatedAt: now, UpdatedAt: now,
|
||
}
|
||
if err := s.Repo.Insert(ctx, j); err != nil {
|
||
return err
|
||
}
|
||
// 與其他任務相同：入列就寫一則通知（延遲執行仍會在任務列表可見）
|
||
s.notify(ctx, j)
|
||
return nil
|
||
}
|
||
|
||
// Ensure job.Service implements TokenRenewScheduler (threads package).
|
||
var _ interface {
|
||
ScheduleTokenRenew(ctx context.Context, ownerUID int64, accountID string, runAfter int64) error
|
||
} = (*Service)(nil)
|
||
|
||
// PersonaAnalyzeAccountPayload — worker 爬公開貼文並分析
|
||
type PersonaAnalyzeAccountPayload struct {
|
||
Username string `json:"username"`
|
||
Lang string `json:"lang,omitempty"`
|
||
}
|
||
|
||
// PersonaAnalyzeTextPayload — worker 從貼文文字分析
|
||
type PersonaAnalyzeTextPayload struct {
|
||
RawText string `json:"raw_text"`
|
||
SourceLabel string `json:"source_label,omitempty"`
|
||
Lang string `json:"lang,omitempty"`
|
||
}
|
||
|
||
// SchedulePersonaAnalyzeAccount — 立刻可領；ref=personaID；同人設舊佇列取消
|
||
func (s *Service) SchedulePersonaAnalyzeAccount(ctx context.Context, ownerUID int64, personaID, username, lang string) (*domain.Job, error) {
|
||
if ownerUID <= 0 || personaID == "" {
|
||
return nil, domain.ErrForbidden
|
||
}
|
||
username = strings.TrimPrefix(strings.TrimSpace(username), "@")
|
||
if username == "" {
|
||
return nil, fmt.Errorf("empty username")
|
||
}
|
||
_ = s.Repo.CancelPendingByRef(ctx, ownerUID, domain.TemplatePersonaAnalyzeAccount, personaID)
|
||
_ = s.Repo.CancelPendingByRef(ctx, ownerUID, domain.TemplatePersonaAnalyzeText, personaID)
|
||
body, _ := json.Marshal(PersonaAnalyzeAccountPayload{Username: username, Lang: lang})
|
||
now := domain.NowNano()
|
||
j := &domain.Job{
|
||
ID: uuid.NewString(), OwnerUID: ownerUID,
|
||
TemplateType: domain.TemplatePersonaAnalyzeAccount,
|
||
Status: domain.StatusQueued,
|
||
RefID: personaID,
|
||
Payload: string(body),
|
||
ProgressSummary: fmt.Sprintf("人設分析 · 爬取 @%s 公開貼文 · 已排程", username),
|
||
ProgressPercent: 0,
|
||
CreatedAt: now, UpdatedAt: now,
|
||
}
|
||
if err := s.Repo.Insert(ctx, j); err != nil {
|
||
return nil, err
|
||
}
|
||
s.notify(ctx, j)
|
||
return j, nil
|
||
}
|
||
|
||
// ComposeMimicPayload — worker 仿寫輸入／輸出
|
||
type ComposeMimicPayload struct {
|
||
SourceText string `json:"source_text"`
|
||
PersonaID string `json:"persona_id,omitempty"`
|
||
StructureNotes string `json:"structure_notes,omitempty"`
|
||
Lang string `json:"lang,omitempty"`
|
||
// ResultText 成功後寫回
|
||
ResultText string `json:"result_text,omitempty"`
|
||
}
|
||
|
||
// ScheduleComposeMimic — 立刻可領；仿寫走背景 job，避免 HTTP 120s 逾時
|
||
func (s *Service) ScheduleComposeMimic(ctx context.Context, ownerUID int64, sourceText, personaID, structureNotes, lang string) (*domain.Job, error) {
|
||
if ownerUID <= 0 {
|
||
return nil, domain.ErrForbidden
|
||
}
|
||
sourceText = strings.TrimSpace(sourceText)
|
||
if sourceText == "" {
|
||
return nil, fmt.Errorf("empty source")
|
||
}
|
||
// 同使用者只保留一則進行中的仿寫（可選：不 cancel 舊的也可）
|
||
body, _ := json.Marshal(ComposeMimicPayload{
|
||
SourceText: sourceText, PersonaID: strings.TrimSpace(personaID),
|
||
StructureNotes: strings.TrimSpace(structureNotes), Lang: lang,
|
||
})
|
||
now := domain.NowNano()
|
||
j := &domain.Job{
|
||
ID: uuid.NewString(), OwnerUID: ownerUID,
|
||
TemplateType: domain.TemplateComposeMimic,
|
||
Status: domain.StatusQueued,
|
||
RefID: strings.TrimSpace(personaID),
|
||
Payload: string(body),
|
||
ProgressSummary: "仿寫貼文 · 已排程（可離開頁面）",
|
||
ProgressPercent: 0,
|
||
CreatedAt: now, UpdatedAt: now,
|
||
}
|
||
if err := s.Repo.Insert(ctx, j); err != nil {
|
||
return nil, err
|
||
}
|
||
s.notify(ctx, j)
|
||
return j, nil
|
||
}
|
||
|
||
// PlayGenerateScriptPayload — worker 一次產完整互回劇本
|
||
type PlayGenerateScriptPayload struct {
|
||
PlayID string `json:"play_id"`
|
||
OnlyEmpty bool `json:"only_empty"`
|
||
// 成功後統計
|
||
Filled int `json:"filled,omitempty"`
|
||
}
|
||
|
||
// SchedulePlayGenerateScript — 立刻可領；ref=playID；同劇本舊佇列取消
|
||
func (s *Service) SchedulePlayGenerateScript(ctx context.Context, ownerUID int64, playID string, onlyEmpty bool) (*domain.Job, error) {
|
||
if ownerUID <= 0 || strings.TrimSpace(playID) == "" {
|
||
return nil, domain.ErrForbidden
|
||
}
|
||
playID = strings.TrimSpace(playID)
|
||
_ = s.Repo.CancelPendingByRef(ctx, ownerUID, domain.TemplatePlayGenerateScript, playID)
|
||
body, _ := json.Marshal(PlayGenerateScriptPayload{PlayID: playID, OnlyEmpty: onlyEmpty})
|
||
now := domain.NowNano()
|
||
j := &domain.Job{
|
||
ID: uuid.NewString(), OwnerUID: ownerUID,
|
||
TemplateType: domain.TemplatePlayGenerateScript,
|
||
Status: domain.StatusQueued,
|
||
RefID: playID,
|
||
Payload: string(body),
|
||
ProgressSummary: "劇本 AI 產文 · 已排程（一次產全部步驟，可離開頁面）",
|
||
ProgressPercent: 0,
|
||
CreatedAt: now, UpdatedAt: now,
|
||
}
|
||
if err := s.Repo.Insert(ctx, j); err != nil {
|
||
return nil, err
|
||
}
|
||
s.notify(ctx, j)
|
||
return j, nil
|
||
}
|
||
|
||
// SucceedJobWithPayload — 終態成功並寫回 payload（仿寫結果）
|
||
func (s *Service) SucceedJobWithPayload(ctx context.Context, jobID, summary, payload string) (*domain.Job, error) {
|
||
j, err := s.Repo.FindByID(ctx, jobID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if !domain.CanTransition(j.Status, domain.StatusSucceeded) {
|
||
return nil, domain.ErrIllegalStatus
|
||
}
|
||
if payload != "" {
|
||
j.Payload = payload
|
||
}
|
||
if err := domain.ApplyProgress(j, 100, summary); err != nil {
|
||
return nil, err
|
||
}
|
||
j.Status = domain.StatusSucceeded
|
||
j.CompletedAt = domain.NowNano()
|
||
j.UpdatedAt = j.CompletedAt
|
||
if err := s.update(ctx, j); err != nil {
|
||
return nil, err
|
||
}
|
||
s.stopHeartbeat(jobID)
|
||
s.notify(ctx, j)
|
||
return j, nil
|
||
}
|
||
|
||
// SchedulePersonaAnalyzeText — 立刻可領；ref=personaID
|
||
func (s *Service) SchedulePersonaAnalyzeText(ctx context.Context, ownerUID int64, personaID, rawText, sourceLabel, lang string) (*domain.Job, error) {
|
||
if ownerUID <= 0 || personaID == "" {
|
||
return nil, domain.ErrForbidden
|
||
}
|
||
rawText = strings.TrimSpace(rawText)
|
||
if rawText == "" {
|
||
return nil, fmt.Errorf("empty text")
|
||
}
|
||
_ = s.Repo.CancelPendingByRef(ctx, ownerUID, domain.TemplatePersonaAnalyzeAccount, personaID)
|
||
_ = s.Repo.CancelPendingByRef(ctx, ownerUID, domain.TemplatePersonaAnalyzeText, personaID)
|
||
body, _ := json.Marshal(PersonaAnalyzeTextPayload{
|
||
RawText: rawText, SourceLabel: strings.TrimSpace(sourceLabel), Lang: lang,
|
||
})
|
||
now := domain.NowNano()
|
||
label := strings.TrimSpace(sourceLabel)
|
||
if label == "" {
|
||
label = "手動貼文"
|
||
}
|
||
j := &domain.Job{
|
||
ID: uuid.NewString(), OwnerUID: ownerUID,
|
||
TemplateType: domain.TemplatePersonaAnalyzeText,
|
||
Status: domain.StatusQueued,
|
||
RefID: personaID,
|
||
Payload: string(body),
|
||
ProgressSummary: fmt.Sprintf("人設分析 · 文字來源「%s」· 已排程", label),
|
||
ProgressPercent: 0,
|
||
CreatedAt: now, UpdatedAt: now,
|
||
}
|
||
if err := s.Repo.Insert(ctx, j); err != nil {
|
||
return nil, err
|
||
}
|
||
s.notify(ctx, j)
|
||
return j, nil
|
||
}
|
||
|
||
// RunTokenRenew executes threads_token_renew job body (caller provides refresh fn).
|
||
func (s *Service) MarkRunningProgress(ctx context.Context, jobID string, pct int, summary string) (*domain.Job, error) {
|
||
j, err := s.Repo.FindByID(ctx, jobID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if j.Status != domain.StatusRunning {
|
||
return nil, domain.ErrIllegalStatus
|
||
}
|
||
if err := domain.ApplyProgress(j, pct, summary); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := s.update(ctx, j); err != nil {
|
||
return nil, err
|
||
}
|
||
// 每次有進度就更新鈴鐺（upsert 同一則）
|
||
s.notify(ctx, j)
|
||
return j, nil
|
||
}
|
||
|
||
func (s *Service) notify(ctx context.Context, j *domain.Job) {
|
||
if s.Notifier == nil || j == nil {
|
||
return
|
||
}
|
||
_ = s.Notifier.NotifyJobState(ctx, j.OwnerUID, j.ID, j.TemplateType, j.Status, j.ProgressSummary, j.ProgressPercent)
|
||
}
|
||
|
||
func (s *Service) SucceedJob(ctx context.Context, jobID, summary string) (*domain.Job, error) {
|
||
return s.Transition(ctx, jobID, domain.StatusSucceeded, summary, 100)
|
||
}
|
||
|
||
// Delete — 擁有者手動刪除；執行中不可刪。
|
||
func (s *Service) Delete(ctx context.Context, ownerUID int64, id string) error {
|
||
j, err := s.Repo.FindByID(ctx, id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if j.OwnerUID != ownerUID {
|
||
return domain.ErrForbidden
|
||
}
|
||
if j.Status == domain.StatusRunning {
|
||
return domain.ErrIllegalStatus
|
||
}
|
||
return s.Repo.Delete(ctx, id)
|
||
}
|
||
|
||
// PurgeExpiredTerminal — 清除終態超過保留期（預設 2 天）的任務。回傳刪除筆數。
|
||
func (s *Service) PurgeExpiredTerminal(ctx context.Context) (int64, error) {
|
||
return s.Repo.DeleteTerminalBefore(ctx, domain.TerminalCutoff(domain.NowNano()))
|
||
}
|
||
|
||
// FailJob — JB-04
|
||
func (s *Service) FailJob(ctx context.Context, jobID, errMsg string) (*domain.Job, error) {
|
||
j, err := s.Repo.FindByID(ctx, jobID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if j.Status == domain.StatusPending || j.Status == domain.StatusQueued {
|
||
j.Status = domain.StatusRunning
|
||
j.UpdatedAt = domain.NowNano()
|
||
_ = s.update(ctx, j)
|
||
}
|
||
if !domain.CanTransition(j.Status, domain.StatusFailed) {
|
||
return nil, domain.ErrIllegalStatus
|
||
}
|
||
j.Status = domain.StatusFailed
|
||
j.Error = errMsg
|
||
if errMsg != "" {
|
||
j.ProgressSummary = "失敗：" + errMsg
|
||
if len([]rune(j.ProgressSummary)) > 200 {
|
||
j.ProgressSummary = string([]rune(j.ProgressSummary)[:200]) + "…"
|
||
}
|
||
} else {
|
||
j.ProgressSummary = "失敗"
|
||
}
|
||
j.CompletedAt = domain.NowNano()
|
||
j.UpdatedAt = j.CompletedAt
|
||
if err := s.update(ctx, j); err != nil {
|
||
return nil, err
|
||
}
|
||
s.stopHeartbeat(jobID)
|
||
s.notify(ctx, j)
|
||
return j, nil
|
||
}
|