iotlcm/worker-svr/internal/service/monthlyusage/run_integration_test.go

package monthlyusage

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"testing"
	"time"

	zsetconsumer "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consumer/zset"
	fullsyncarchive "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/archive"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/contract"
	fullsyncruntime "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/runtime"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/testutil"
)

type runUsageClient struct{}

func (runUsageClient) QuerySimDataUsage(_ context.Context, input *model.SimDataUsageInput) ([]model.SimBaseRes[model.SimDataUsageItem], error) {
	iccid := input.SimKeys[0].ICCID
	switch iccid {
	case "retry":
		return nil, errors.New("temporary network failure")
	case "business":
		return []model.SimBaseRes[model.SimDataUsageItem]{{SimKey: &model.SimKey{ICCID: iccid}, Code: 11011, Message: "business failure"}}, nil
	default:
		return []model.SimBaseRes[model.SimDataUsageItem]{{SimKey: &model.SimKey{ICCID: iccid}, ResData: model.SimDataUsageItem{DataUsage: 12.5}}}, nil
	}
}

func TestMonthlyRunRetryConvergesAndArchives(t *testing.T) {
	redis := testutil.Redis(t)
	ctx := context.Background()
	store, err := zset.New(redis, "flow:")
	if err != nil {
		t.Fatal(err)
	}
	store.EnableArchiveIndex(true)
	const runID = "monthly-retry-archive"
	if err := store.CreateRun(ctx, zset.RunMeta{RunID: runID, Operation: "current_month_usage", QueryMonth: "2026-09"}); err != nil {
		t.Fatal(err)
	}
	tasks := []zset.Task{
		{RunID: runID, TaskID: "retry", CardID: 1, ChanKey: "channel", TempKey: "sim_yd_v5", ICCID: "retry"},
		{RunID: runID, TaskID: "business", CardID: 2, ChanKey: "channel", TempKey: "sim_yd_v5", ICCID: "business"},
		{RunID: runID, TaskID: "success", CardID: 3, ChanKey: "channel", TempKey: "sim_yd_v5", ICCID: "success"},
	}
	if n, err := store.EnqueueRunBatch(ctx, runID, "channel", tasks, time.Now()); err != nil || n != len(tasks) {
		t.Fatalf("enqueue = %d, err = %v", n, err)
	}
	if err := store.MarkProducerDone(ctx, runID); err != nil {
		t.Fatal(err)
	}
	repo := &usageRepositoryStub{}
	handler, err := NewHandler(runUsageClient{}, repo)
	if err != nil {
		t.Fatal(err)
	}
	consumer, err := zsetconsumer.New(store, handler, zsetconsumer.Config{
		RunQueueEnabled:   true,
		RoundRetryEnabled: true,
		MaxRounds:         2,
		ShouldRetryNextRound: func(result zsetconsumer.Result) bool {
			return !result.Success && result.Retryable
		},
		PollInterval: 500 * time.Millisecond,
		ClaimBatch:   3,
		TemplateQPS:  map[string]float64{"sim_yd_v5": 1000000},
		LeaseTTL:     time.Minute,
		TaskTimeout:  30 * time.Second,
	})
	if err != nil {
		t.Fatal(err)
	}
	runCtx, cancel := context.WithCancel(ctx)
	done := make(chan error, 1)
	go func() { done <- consumer.Run(runCtx) }()
	defer func() {
		cancel()
		if err := <-done; err != nil {
			t.Errorf("consumer stopped: %v", err)
		}
	}()

	deadline := time.Now().Add(10 * time.Second)
	var meta zset.RunMeta
	for time.Now().Before(deadline) {
		meta, err = store.GetRunMeta(ctx, runID)
		if err != nil {
			t.Fatal(err)
		}
		if meta.NextRoundCount == 1 {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	if meta.CurrentRound != 1 || meta.ActiveCount != 0 || meta.InflightCount != 0 || meta.NextRoundCount != 1 || meta.SuccessCount != 1 || meta.FailedCount != 1 || meta.Status != "running" {
		t.Fatalf("first round did not defer only retryable failure: %+v", meta)
	}
	failedKey, _ := store.FailedKey(runID)
	if count, err := redis.Do(ctx, "HLEN", failedKey); err != nil || count.Int() != 1 {
		t.Fatalf("first round final failures = %v, err = %v", count, err)
	}
	for time.Now().Before(deadline) {
		meta, err = store.GetRunMeta(ctx, runID)
		if err != nil {
			t.Fatal(err)
		}
		if meta.Status == "completed_with_failures" {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	if meta.Status != "completed_with_failures" || meta.CurrentRound != 2 || meta.SuccessCount != 1 || meta.FailedCount != 2 || meta.ActiveCount != 0 || meta.NextRoundCount != 0 || meta.CompletedAt.IsZero() {
		t.Fatalf("run did not converge at max rounds: %+v", meta)
	}
	if len(repo.updates) != 1 || repo.updates[0].CardID != 3 {
		t.Fatalf("successful usage update changed: %+v", repo.updates)
	}
	failed, err := redis.Do(ctx, "HGET", failedKey, "retry")
	if err != nil {
		t.Fatal(err)
	}
	var record zset.FailureRecord
	if err := json.Unmarshal([]byte(failed.String()), &record); err != nil || record.Round != 2 || record.FailureType != "network" {
		t.Fatalf("exhausted failure = %+v, err = %v", record, err)
	}

	root := t.TempDir()
	runtimeRepo, err := fullsyncruntime.New(store, root)
	if err != nil {
		t.Fatal(err)
	}
	page, err := runtimeRepo.ListArchiveWork(ctx, contract.PageRequest{Limit: 10})
	if err != nil || len(page.Items) != 1 || page.Items[0].RunID != runID {
		t.Fatalf("pending archive = %+v, err = %v", page, err)
	}
	service := fullsyncarchive.New(root, runtimeRepo, runtimeRepo, runtimeRepo)
	if err := service.RunOnce(ctx); err != nil {
		t.Fatal(err)
	}
	data, err := os.ReadFile(filepath.Join(root, "run-"+runID+".json"))
	if err != nil {
		t.Fatal(err)
	}
	var archive contract.RunArchive
	if err := json.Unmarshal(data, &archive); err != nil {
		t.Fatal(err)
	}
	if archive.Run.Status != contract.RunCompletedWithFailures || archive.Run.SuccessCount != 1 || archive.Run.FailedCount != 2 || len(archive.Rounds) != 2 {
		t.Fatalf("archived run summary = %+v", archive)
	}
	if first, last := archive.Rounds[0], archive.Rounds[1]; first.Round != 1 || first.EnqueuedCount != 3 || first.SuccessCount != 1 || first.FailedCount != 1 || first.NextRoundCount != 1 || last.Round != 2 || last.EnqueuedCount != 1 || last.FailedCount != 1 || last.NextRoundCount != 0 {
		t.Fatalf("archived round stats = %+v", archive.Rounds)
	}
}