thread-master/apps/backend/internal/module/job/domain/job.go

package domain

import (
	"errors"
	"time"
)

const (
	StatusPending   = "pending"
	StatusQueued    = "queued"
	StatusRunning   = "running"
	StatusSucceeded = "succeeded"
	StatusFailed    = "failed"
	StatusCancelled = "cancelled"

	TemplateDemo                  = "demo"
	TemplateThreadsTokenRenew     = "threads_token_renew"
	TemplatePersonaAnalyzeAccount = "persona_analyze_account"
	TemplatePersonaAnalyzeText    = "persona_analyze_text"
	TemplateComposeMimic          = "compose_mimic"
	// TemplatePlayGenerateScript — 互回／串場：一次產完整劇本（背景 job，避免 HTTP 卡住）
	TemplatePlayGenerateScript = "play_generate_script"
	TemplateScoutScan          = "scout_scan"
)

// Job 生命週期契約（所有模板必須遵守，worker / API 入列時）：
//  1. 建立／入列 → StatusQueued + notify（鈴鐺「已排程」）
//  2. ClaimNext → StatusRunning + notify（至少 ~5%）
//  3. 執行中每有進度 → MarkRunningProgress → notify（同一 job upsert 一則通知）
//  4. 終態 SucceedJob / FailJob → notify（已完成／失敗）
// 前端靠全域 JobLive 輪詢列表 + 通知；遠期 run_after 不進頂欄即時條。

var (
	ErrNotFound      = errors.New("job not found")
	ErrForbidden     = errors.New("job access denied")
	ErrIllegalStatus = errors.New("illegal job status transition")
)

type Job struct {
	ID              string `bson:"_id" json:"id"`
	OwnerUID        int64  `bson:"owner_uid" json:"owner_uid"`
	TemplateType    string `bson:"template_type" json:"template_type"`
	Status          string `bson:"status" json:"status"`
	ProgressSummary string `bson:"progress_summary" json:"progress_summary"`
	ProgressPercent int    `bson:"progress_percent" json:"progress_percent"`
	Error           string `bson:"error,omitempty" json:"error,omitempty"`
	WorkerID        string `bson:"worker_id,omitempty" json:"worker_id,omitempty"`
	LeaseOwner      string `bson:"lease_owner,omitempty" json:"lease_owner,omitempty"`
	LeaseExpiresAt  int64  `bson:"lease_expires_at,omitempty" json:"lease_expires_at,omitempty"`
	Attempt         int    `bson:"attempt,omitempty" json:"attempt,omitempty"`
	// RefID e.g. threads account id for renew / persona id for analyze
	RefID string `bson:"ref_id,omitempty" json:"ref_id,omitempty"`
	// Payload JSON body for job worker (e.g. {"username":"x"})
	Payload string `bson:"payload,omitempty" json:"payload,omitempty"`
	// RunAfter unix nanoseconds; worker only claims when now >= RunAfter (0 = due now)
	RunAfter    int64 `bson:"run_after,omitempty" json:"run_after,omitempty"`
	CreatedAt   int64 `bson:"created_at" json:"created_at"`
	UpdatedAt   int64 `bson:"updated_at" json:"updated_at"`
	CompletedAt int64 `bson:"completed_at,omitempty" json:"completed_at,omitempty"`
	// Version is incremented by every persisted mutation and makes lifecycle
	// writes compare-and-swap safe across gateway and worker instances.
	Version int64 `bson:"version" json:"version"`
}

// DefaultTokenRenewDelayNs — 連帳後約 30 天做第一次 renew（長效 token ~60 天）
const DefaultTokenRenewDelayNs = int64(30 * 24 * time.Hour)

// Job leases are renewed by the claiming service while work is running.
const (
	DefaultLeaseDuration   = time.Minute
	DefaultLeaseDurationNs = int64(DefaultLeaseDuration)
	DefaultHeartbeat       = 20 * time.Second
)

// TerminalRetentionNs — 終態任務（已完成／失敗／取消）保留多久後自動清除
const TerminalRetentionNs = int64(2 * 24 * time.Hour)

func NowNano() int64 { return time.Now().UTC().UnixNano() }

// IsTerminal — 不可再轉移的狀態
func IsTerminal(status string) bool {
	return status == StatusSucceeded || status == StatusFailed || status == StatusCancelled
}

// TerminalCutoff — 早於此時間的終態任務應被 purge
func TerminalCutoff(now int64) int64 {
	if now <= 0 {
		now = NowNano()
	}
	return now - TerminalRetentionNs
}

// CanTransition enforces JB-10 guarded status.
func CanTransition(from, to string) bool {
	switch from {
	case StatusPending:
		return to == StatusQueued || to == StatusRunning || to == StatusFailed
	case StatusQueued:
		return to == StatusRunning || to == StatusFailed || to == StatusCancelled
	case StatusRunning:
		return to == StatusSucceeded || to == StatusFailed || to == StatusCancelled
	default:
		return false // terminal
	}
}

// ApplyProgress enforces monotonic percent (JB-02).
func ApplyProgress(j *Job, percent int, summary string) error {
	if percent < j.ProgressPercent {
		return ErrIllegalStatus
	}
	if percent > 100 {
		percent = 100
	}
	j.ProgressPercent = percent
	if summary != "" {
		j.ProgressSummary = summary
	}
	j.UpdatedAt = NowNano()
	return nil
}

// List tab buckets（任務列表頁籤）
const (
	TabRecurring = "recurring" // 遠期定期／已排程未到期
	TabActive    = "active"    // 執行中、待領取（已到期）、取消中
	TabHistory   = "history"   // 已完成／失敗／取消
)

// NormalizeListTab — 空或未知 → active
func NormalizeListTab(tab string) string {
	switch tab {
	case TabRecurring, TabActive, TabHistory:
		return tab
	default:
		return TabActive
	}
}

// MatchListTab — 依狀態＋run_after 分桶（與 FE 一致）
func MatchListTab(j *Job, tab string, nowNs int64) bool {
	if j == nil {
		return false
	}
	if nowNs <= 0 {
		nowNs = NowNano()
	}
	switch NormalizeListTab(tab) {
	case TabHistory:
		return IsTerminal(j.Status)
	case TabRecurring:
		// 遠期排程：pending/queued 且 run_after 尚未到
		if j.Status != StatusPending && j.Status != StatusQueued {
			return false
		}
		return j.RunAfter > 0 && j.RunAfter > nowNs
	default: // active
		if j.Status == StatusRunning {
			return true
		}
		if j.Status == StatusPending || j.Status == StatusQueued {
			// 已到期或立刻可領
			return j.RunAfter <= 0 || j.RunAfter <= nowNs
		}
		return false
	}
}

// ClampPage — page 從 1；pageSize 預設 10、上限 50
func ClampPage(page, pageSize int) (int, int) {
	if page < 1 {
		page = 1
	}
	if pageSize <= 0 {
		pageSize = 10
	}
	if pageSize > 50 {
		pageSize = 50
	}
	return page, pageSize
}