thread-master/apps/backend/internal/module/job/usecase/service_test.go

package usecase_test

import (
	"context"
	"testing"
	"time"

	"apps/backend/internal/module/appnotif/repository"
	appUC "apps/backend/internal/module/appnotif/usecase"
	"apps/backend/internal/module/job/domain"
	jobRepo "apps/backend/internal/module/job/repository"
	"apps/backend/internal/module/job/usecase"

	"github.com/stretchr/testify/require"
)

func newJobSvc() (*usecase.Service, *appUC.Service) {
	jr := jobRepo.NewMemory()
	nr := repository.NewMemory()
	notif := appUC.New(nr)
	svc := usecase.New(jr)
	svc.Notifier = notif
	return svc, notif
}

func TestJB_01_CreateDemo(t *testing.T) {
	svc, _ := newJobSvc()
	j, err := svc.StartDemo(context.Background(), 1_000_100)
	require.NoError(t, err)
	require.Equal(t, domain.StatusQueued, j.Status)
	require.Equal(t, domain.TemplateDemo, j.TemplateType)
	list, err := svc.List(context.Background(), 1_000_100)
	require.NoError(t, err)
	require.NotEmpty(t, list)
}

func TestJB_02_ProgressMonotonic(t *testing.T) {
	svc, _ := newJobSvc()
	j, err := svc.StartDemo(context.Background(), 1_000_101)
	require.NoError(t, err)
	claimed, err := svc.ClaimNext(context.Background(), "worker-test")
	require.NoError(t, err)
	require.Equal(t, j.ID, claimed.ID)
	require.Equal(t, domain.StatusRunning, claimed.Status)
	out, err := svc.RunDemoToSuccess(context.Background(), claimed.ID)
	require.NoError(t, err)
	require.Equal(t, 100, out.ProgressPercent)
	require.Equal(t, domain.StatusSucceeded, out.Status)
	// cannot go backwards
	err = domain.ApplyProgress(out, 50, "nope")
	require.ErrorIs(t, err, domain.ErrIllegalStatus)
}

func TestJB_03_JobSucceeded(t *testing.T) {
	svc, _ := newJobSvc()
	j, _ := svc.StartDemo(context.Background(), 1_000_102)
	_, _ = svc.ClaimNext(context.Background(), "w1")
	out, err := svc.RunDemoToSuccess(context.Background(), j.ID)
	require.NoError(t, err)
	require.Equal(t, domain.StatusSucceeded, out.Status)
	require.True(t, out.CompletedAt > 0)
}

func TestJB_04_JobFailed(t *testing.T) {
	svc, _ := newJobSvc()
	j, _ := svc.StartDemo(context.Background(), 1_000_103)
	_, _ = svc.ClaimNext(context.Background(), "w1")
	out, err := svc.FailJob(context.Background(), j.ID, "boom")
	require.NoError(t, err)
	require.Equal(t, domain.StatusFailed, out.Status)
	require.NotEmpty(t, out.Error)
	require.True(t, out.CompletedAt > 0)
}

func TestJB_05_JobTerminalCreatesNotification(t *testing.T) {
	svc, notif := newJobSvc()
	uid := int64(1_000_104)
	j, _ := svc.StartDemo(context.Background(), uid)
	_, _ = svc.ClaimNext(context.Background(), "w1")
	_, err := svc.RunDemoToSuccess(context.Background(), j.ID)
	require.NoError(t, err)
	list, err := notif.List(context.Background(), uid)
	require.NoError(t, err)
	require.NotEmpty(t, list)
}

func TestJB_06_UnreadCount(t *testing.T) {
	svc, notif := newJobSvc()
	uid := int64(1_000_105)
	j, _ := svc.StartDemo(context.Background(), uid)
	_, _ = svc.ClaimNext(context.Background(), "w1")
	_, _ = svc.FailJob(context.Background(), j.ID, "x")
	c, err := notif.UnreadCount(context.Background(), uid)
	require.NoError(t, err)
	require.Equal(t, int64(1), c)
}

func TestJB_07_MarkRead(t *testing.T) {
	svc, notif := newJobSvc()
	uid := int64(1_000_106)
	j, _ := svc.StartDemo(context.Background(), uid)
	_, _ = svc.ClaimNext(context.Background(), "w1")
	_, _ = svc.RunDemoToSuccess(context.Background(), j.ID)
	list, _ := notif.List(context.Background(), uid)
	require.NotEmpty(t, list)
	require.NoError(t, notif.MarkRead(context.Background(), uid, list[0].ID))
	c, _ := notif.UnreadCount(context.Background(), uid)
	require.Equal(t, int64(0), c)
	require.True(t, list[0].ReadAt == 0) // old snapshot
}

func TestJB_08_MarkAllRead(t *testing.T) {
	svc, notif := newJobSvc()
	uid := int64(1_000_107)
	for i := 0; i < 2; i++ {
		j, _ := svc.StartDemo(context.Background(), uid)
		_, _ = svc.ClaimNext(context.Background(), "w1")
		_, _ = svc.RunDemoToSuccess(context.Background(), j.ID)
	}
	c, _ := notif.UnreadCount(context.Background(), uid)
	require.GreaterOrEqual(t, c, int64(2))
	require.NoError(t, notif.MarkAllRead(context.Background(), uid))
	c2, _ := notif.UnreadCount(context.Background(), uid)
	require.Equal(t, int64(0), c2)
}

func TestJB_09_GetOtherUIDDenied(t *testing.T) {
	svc, _ := newJobSvc()
	j, _ := svc.StartDemo(context.Background(), 1_000_108)
	_, err := svc.Get(context.Background(), 1_000_109, j.ID)
	require.ErrorIs(t, err, domain.ErrForbidden)
}

func TestJB_10_GuardedReject(t *testing.T) {
	svc, _ := newJobSvc()
	j, _ := svc.StartDemo(context.Background(), 1_000_110)
	_, _ = svc.ClaimNext(context.Background(), "w1")
	_, _ = svc.RunDemoToSuccess(context.Background(), j.ID)
	_, err := svc.Transition(context.Background(), j.ID, domain.StatusRunning, "nope", 10)
	require.ErrorIs(t, err, domain.ErrIllegalStatus)
}

func TestJB_11_ManualDeleteTerminal(t *testing.T) {
	svc, _ := newJobSvc()
	uid := int64(1_000_111)
	j, _ := svc.StartDemo(context.Background(), uid)
	_, _ = svc.ClaimNext(context.Background(), "w1")
	out, err := svc.RunDemoToSuccess(context.Background(), j.ID)
	require.NoError(t, err)
	require.Equal(t, domain.StatusSucceeded, out.Status)

	require.NoError(t, svc.Delete(context.Background(), uid, j.ID))
	_, err = svc.Get(context.Background(), uid, j.ID)
	require.ErrorIs(t, err, domain.ErrNotFound)
}

func TestJB_12_DeleteRunningRefused(t *testing.T) {
	svc, _ := newJobSvc()
	uid := int64(1_000_112)
	j, _ := svc.StartDemo(context.Background(), uid)
	claimed, err := svc.ClaimNext(context.Background(), "w1")
	require.NoError(t, err)
	require.Equal(t, domain.StatusRunning, claimed.Status)
	err = svc.Delete(context.Background(), uid, j.ID)
	require.ErrorIs(t, err, domain.ErrIllegalStatus)
}

func TestJB_13_PurgeExpiredTerminal(t *testing.T) {
	svc, _ := newJobSvc()
	uid := int64(1_000_113)
	j, _ := svc.StartDemo(context.Background(), uid)
	_, _ = svc.ClaimNext(context.Background(), "w1")
	out, err := svc.RunDemoToSuccess(context.Background(), j.ID)
	require.NoError(t, err)

	// 模擬完成時間在 2 天前
	out.CompletedAt = domain.NowNano() - domain.TerminalRetentionNs - 1
	out.UpdatedAt = out.CompletedAt
	require.NoError(t, svc.Repo.Update(context.Background(), out))

	n, err := svc.PurgeExpiredTerminal(context.Background())
	require.NoError(t, err)
	require.Equal(t, int64(1), n)
	_, err = svc.Get(context.Background(), uid, j.ID)
	require.ErrorIs(t, err, domain.ErrNotFound)
}

func TestJB_14_PurgeKeepsRecentTerminal(t *testing.T) {
	svc, _ := newJobSvc()
	uid := int64(1_000_114)
	j, _ := svc.StartDemo(context.Background(), uid)
	_, _ = svc.ClaimNext(context.Background(), "w1")
	_, err := svc.RunDemoToSuccess(context.Background(), j.ID)
	require.NoError(t, err)

	n, err := svc.PurgeExpiredTerminal(context.Background())
	require.NoError(t, err)
	require.Equal(t, int64(0), n)
	got, err := svc.Get(context.Background(), uid, j.ID)
	require.NoError(t, err)
	require.Equal(t, j.ID, got.ID)
}

func TestJB_15_SchedulePersonaAnalyzeAccount(t *testing.T) {
	svc, _ := newJobSvc()
	uid := int64(1_000_115)
	j, err := svc.SchedulePersonaAnalyzeAccount(context.Background(), uid, "pe_abc", "ultralab_tw", "zh-TW")
	require.NoError(t, err)
	require.Equal(t, domain.TemplatePersonaAnalyzeAccount, j.TemplateType)
	require.Equal(t, domain.StatusQueued, j.Status)
	require.Equal(t, "pe_abc", j.RefID)
	require.Contains(t, j.Payload, "ultralab_tw")
	require.Contains(t, j.ProgressSummary, "@ultralab_tw")

	// re-schedule cancels previous pending
	j2, err := svc.SchedulePersonaAnalyzeAccount(context.Background(), uid, "pe_abc", "other", "en")
	require.NoError(t, err)
	require.NotEqual(t, j.ID, j2.ID)
	old, err := svc.Get(context.Background(), uid, j.ID)
	require.NoError(t, err)
	require.Equal(t, domain.StatusCancelled, old.Status)

	claimed, err := svc.ClaimNext(context.Background(), "w1")
	require.NoError(t, err)
	require.Equal(t, j2.ID, claimed.ID)
	require.Equal(t, domain.StatusRunning, claimed.Status)
}

func TestJB_16_SchedulePersonaAnalyzeText(t *testing.T) {
	svc, _ := newJobSvc()
	uid := int64(1_000_116)
	j, err := svc.SchedulePersonaAnalyzeText(context.Background(), uid, "pe_txt", "hello world\n---\nsecond sample text here", "手動", "zh-TW")
	require.NoError(t, err)
	require.Equal(t, domain.TemplatePersonaAnalyzeText, j.TemplateType)
	require.Equal(t, "pe_txt", j.RefID)
	require.Contains(t, j.Payload, "raw_text")
}

func TestJB_ListTab_BucketsAndPaging(t *testing.T) {
	svc, _ := newJobSvc()
	uid := int64(1_000_200)
	ctx := context.Background()
	// due demo → active, then finish → history
	d, err := svc.StartDemo(ctx, uid)
	require.NoError(t, err)
	// future scheduled → recurring
	now := domain.NowNano()
	require.NoError(t, svc.ScheduleTokenRenew(ctx, uid, "acc-1", now+domain.DefaultTokenRenewDelayNs))

	claimed, err := svc.ClaimNext(ctx, "w1")
	require.NoError(t, err)
	require.Equal(t, d.ID, claimed.ID)
	_, err = svc.RunDemoToSuccess(ctx, claimed.ID)
	require.NoError(t, err)

	active, totalA, err := svc.ListTab(ctx, uid, domain.TabActive, 1, 10)
	require.NoError(t, err)
	require.Equal(t, int64(0), totalA)
	require.Empty(t, active)

	rec, totalR, err := svc.ListTab(ctx, uid, domain.TabRecurring, 1, 10)
	require.NoError(t, err)
	require.Equal(t, int64(1), totalR)
	require.Len(t, rec, 1)
	require.Equal(t, domain.TemplateThreadsTokenRenew, rec[0].TemplateType)

	hist, totalH, err := svc.ListTab(ctx, uid, domain.TabHistory, 1, 10)
	require.NoError(t, err)
	require.Equal(t, int64(1), totalH)
	require.Len(t, hist, 1)
	require.Equal(t, domain.StatusSucceeded, hist[0].Status)

	// page size 1 on history
	page1, _, err := svc.ListTab(ctx, uid, domain.TabHistory, 1, 1)
	require.NoError(t, err)
	require.Len(t, page1, 1)
}

func TestJobLease_ReclaimsStaleRunningAndRejectsOldOwner(t *testing.T) {
	ctx1, cancel1 := context.WithCancel(context.Background())
	defer cancel1()
	ctx2, cancel2 := context.WithCancel(context.Background())
	defer cancel2()
	repo := jobRepo.NewMemory()
	worker1 := usecase.New(repo)
	worker2 := usecase.New(repo)

	queued, err := worker1.StartDemo(context.Background(), 1_000_300)
	require.NoError(t, err)
	first, err := worker1.ClaimNext(ctx1, "worker-1")
	require.NoError(t, err)
	require.Equal(t, queued.ID, first.ID)
	require.Equal(t, "worker-1", first.LeaseOwner)
	require.Equal(t, 1, first.Attempt)
	require.Greater(t, first.LeaseExpiresAt, domain.NowNano())

	_, err = worker2.ClaimNext(ctx2, "worker-2")
	require.ErrorIs(t, err, domain.ErrNotFound)

	stale, err := repo.FindByID(context.Background(), queued.ID)
	require.NoError(t, err)
	stale.LeaseExpiresAt = domain.NowNano() - 1
	require.NoError(t, repo.Update(context.Background(), stale))

	reclaimed, err := worker2.ClaimNext(ctx2, "worker-2")
	require.NoError(t, err)
	require.Equal(t, "worker-2", reclaimed.LeaseOwner)
	require.Equal(t, 2, reclaimed.Attempt)

	_, err = worker1.MarkRunningProgress(context.Background(), queued.ID, 30, "stale write")
	require.ErrorIs(t, err, domain.ErrIllegalStatus)
	_, err = worker1.SucceedJob(context.Background(), queued.ID, "stale completion")
	require.ErrorIs(t, err, domain.ErrIllegalStatus)
	_, err = worker1.FailJob(context.Background(), queued.ID, "stale failure")
	require.ErrorIs(t, err, domain.ErrIllegalStatus)
	updated, err := worker2.MarkRunningProgress(context.Background(), queued.ID, 30, "current write")
	require.NoError(t, err)
	require.Equal(t, "current write", updated.ProgressSummary)
	completed, err := worker2.SucceedJob(context.Background(), queued.ID, "current completion")
	require.NoError(t, err)
	require.Equal(t, domain.StatusSucceeded, completed.Status)
}

func TestJobLease_HeartbeatRenewsWhileRunning(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	repo := jobRepo.NewMemory()
	svc := usecase.New(repo)
	svc.HeartbeatInterval = 5 * time.Millisecond
	svc.LeaseDuration = 2 * domain.DefaultLeaseDuration

	_, err := svc.StartDemo(context.Background(), 1_000_301)
	require.NoError(t, err)
	claimed, err := svc.ClaimNext(ctx, "worker-heartbeat")
	require.NoError(t, err)
	initialExpiry := claimed.LeaseExpiresAt

	require.Eventually(t, func() bool {
		current, findErr := repo.FindByID(context.Background(), claimed.ID)
		return findErr == nil && current.LeaseExpiresAt > initialExpiry
	}, 200*time.Millisecond, 5*time.Millisecond)
}

func TestJobLease_ReclaimsLegacyRunningDocumentWithoutLease(t *testing.T) {
	repo := jobRepo.NewMemory()
	now := domain.NowNano()
	legacy := &domain.Job{
		ID: "legacy-running", OwnerUID: 1_000_302, TemplateType: domain.TemplateDemo,
		Status: domain.StatusRunning, WorkerID: "dead-worker", CreatedAt: now, UpdatedAt: now,
	}
	require.NoError(t, repo.Insert(context.Background(), legacy))

	reclaimed, err := repo.ClaimNext(context.Background(), "replacement-worker")
	require.NoError(t, err)
	require.Equal(t, "replacement-worker", reclaimed.LeaseOwner)
	require.Equal(t, 1, reclaimed.Attempt)
	require.Greater(t, reclaimed.LeaseExpiresAt, now)
}