iotlcm/worker-svr/internal/fullsync/runtime/catalog_test.go

package runtime

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

	"github.com/gogf/gf/v2/container/gvar"

	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/contract"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
)

type catalogRedisStub struct {
	responses []*gvar.Var
	calls     []string
}

func (s *catalogRedisStub) Do(_ context.Context, command string, _ ...any) (*gvar.Var, error) {
	s.calls = append(s.calls, command)
	if len(s.responses) == 0 {
		return gvar.New(nil), nil
	}
	response := s.responses[0]
	s.responses = s.responses[1:]
	return response, nil
}

func (s *catalogRedisStub) Eval(context.Context, string, int64, []string, []any) (*gvar.Var, error) {
	return gvar.New(int64(1)), nil
}

func TestCatalogListRunsMergesLiveAndArchive(t *testing.T) {
	root := t.TempDir()
	now := time.Date(2026, 9, 28, 1, 0, 0, 0, time.UTC)
	archive := contract.RunArchive{Version: 1, Run: contract.RunMeta{
		RunID: "archived", Operation: "sim_status_queue_full_sync", Status: contract.RunCompleted,
		CreatedAt: now.Add(-time.Hour), CompletedAt: now.Add(-time.Minute),
	}}
	data, err := json.Marshal(archive)
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(root, "run-archived.json"), data, 0o644); err != nil {
		t.Fatal(err)
	}

	metaResponse := gvar.New([]any{"runId", "live", "operation", "current_month_usage_full_sync", "status", "running", "currentRound", "1", "createdAt", now.Format(time.RFC3339Nano), "roundStats", "[]"})
	redis := &catalogRedisStub{responses: []*gvar.Var{
		gvar.New([]any{"live"}), metaResponse, metaResponse,
		gvar.New([]any{"live"}), metaResponse, metaResponse,
	}}
	store, err := zset.New(redis, "flow:")
	if err != nil {
		t.Fatal(err)
	}
	repository, err := New(store, root)
	if err != nil {
		t.Fatal(err)
	}
	catalog, err := NewCatalog(repository)
	if err != nil {
		t.Fatal(err)
	}

	result, err := catalog.ListRuns(context.Background(), contract.RunListFilter{Limit: 10})
	if err != nil {
		t.Fatal(err)
	}
	if result.Total != 2 || len(result.Items) != 2 {
		t.Fatalf("result = %+v", result)
	}
	if result.Items[0].Run.RunID != "live" || result.Items[1].Run.RunID != "archived" {
		t.Fatalf("items = %+v", result.Items)
	}
	filtered, err := catalog.ListRuns(context.Background(), contract.RunListFilter{SyncKind: contract.SyncKindTrafficUsage, Limit: 10})
	if err != nil {
		t.Fatal(err)
	}
	if filtered.Total != 1 || filtered.Items[0].Run.RunID != "live" {
		t.Fatalf("filtered = %+v", filtered)
	}
	view, err := catalog.GetRun(context.Background(), "archived")
	if err != nil || view.Source != contract.SourceArchive || view.Run.RunID != "archived" {
		t.Fatalf("archived view = %+v, err = %v", view, err)
	}
}

func TestCatalogPrefersLiveAndSkipsBrokenArchive(t *testing.T) {
	root := t.TempDir()
	for name, payload := range map[string]string{
		"run-live.json":   `{"version":1,"run":{"runId":"live","operation":"sim_status_queue_full_sync","status":"completed"}}`,
		"run-broken.json": `{not-json}`,
	} {
		if err := os.WriteFile(filepath.Join(root, name), []byte(payload), 0o644); err != nil {
			t.Fatal(err)
		}
	}
	redis := &catalogRedisStub{responses: []*gvar.Var{
		gvar.New([]any{"live"}),
		gvar.New([]any{"runId", "live", "operation", "sim_status_queue_full_sync", "status", "running"}),
		gvar.New([]any{"runId", "live", "operation", "sim_status_queue_full_sync", "status", "running"}),
	}}
	store, err := zset.New(redis, "sim-status:")
	if err != nil {
		t.Fatal(err)
	}
	repository, err := New(store, root)
	if err != nil {
		t.Fatal(err)
	}
	catalog, err := NewCatalog(repository)
	if err != nil {
		t.Fatal(err)
	}
	result, err := catalog.ListRuns(context.Background(), contract.RunListFilter{Limit: 10})
	if err != nil || result.Total != 1 || result.Items[0].Source != contract.SourceLive || result.Items[0].Run.Status != contract.RunRunning {
		t.Fatalf("result = %+v, err = %v", result, err)
	}
}