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

package runtime

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

	"github.com/gogf/gf/v2/frame/g"

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

const (
	defaultRunListLimit = 50
	maxRunListLimit     = 100
)

// Catalog aggregates the flow and sim-status Redis namespaces while keeping
// each run's storage protocol inside its existing Repository.
type Catalog struct {
	repositories []*Repository
}

func NewCatalog(repositories ...*Repository) (*Catalog, error) {
	if len(repositories) == 0 {
		return nil, errors.New("full-sync catalog requires repositories")
	}
	for _, repository := range repositories {
		if repository == nil {
			return nil, errors.New("full-sync catalog repository is nil")
		}
	}
	return &Catalog{repositories: repositories}, nil
}

func (c *Catalog) GetRun(ctx context.Context, runID string) (contract.RunView, error) {
	if err := contract.ValidateID(runID); err != nil {
		return contract.RunView{}, err
	}
	for _, repository := range c.repositories {
		meta, err := repository.GetRunMeta(ctx, runID)
		if err == nil {
			rounds, roundErr := repository.GetRoundStats(ctx, runID)
			if roundErr != nil {
				return contract.RunView{}, roundErr
			}
			return contract.RunView{Source: contract.SourceLive, Run: meta, Rounds: rounds}, nil
		}
		if !errors.Is(err, contract.ErrNotFound) {
			return contract.RunView{}, err
		}
	}
	for _, repository := range c.repositories {
		archive, err := repository.GetArchivedRun(ctx, runID)
		if err == nil {
			return contract.RunView{Source: contract.SourceArchive, Run: archive.Run, Rounds: archive.Rounds}, nil
		}
		if !errors.Is(err, contract.ErrNotFound) {
			return contract.RunView{}, err
		}
	}
	return contract.RunView{}, contract.ErrNotFound
}

func (c *Catalog) ListRuns(ctx context.Context, filter contract.RunListFilter) (contract.RunList, error) {
	if filter.Limit == 0 {
		filter.Limit = defaultRunListLimit
	}
	if filter.Limit < 1 || filter.Limit > maxRunListLimit {
		return contract.RunList{}, contract.ErrInvalidArgument
	}
	byID := make(map[string]contract.RunView)
	for _, repository := range c.repositories {
		active, err := repository.store.ActiveRuns(ctx)
		if err != nil {
			return contract.RunList{}, err
		}
		for _, runID := range active {
			view, getErr := repository.liveRun(ctx, runID)
			if errors.Is(getErr, contract.ErrNotFound) {
				continue
			}
			if getErr != nil {
				return contract.RunList{}, getErr
			}
			byID[runID] = view
		}
	}
	if err := c.listArchives(ctx, byID); err != nil {
		return contract.RunList{}, err
	}
	items := make([]contract.RunView, 0, len(byID))
	for _, item := range byID {
		kind := contract.SyncKindForOperation(item.Run.Operation)
		if kind == "" || (filter.SyncKind != "" && kind != filter.SyncKind) {
			continue
		}
		items = append(items, item)
	}
	sort.SliceStable(items, func(i, j int) bool {
		return runSortTime(items[i].Run).After(runSortTime(items[j].Run))
	})
	total := len(items)
	if len(items) > filter.Limit {
		items = items[:filter.Limit]
	}
	if items == nil {
		items = []contract.RunView{}
	}
	return contract.RunList{Items: items, Total: total}, nil
}

func (c *Catalog) ListFailures(ctx context.Context, runID string, page contract.PageRequest) (contract.FailurePage, error) {
	if err := contract.ValidateID(runID); err != nil {
		return contract.FailurePage{}, err
	}
	for _, repository := range c.repositories {
		meta, err := repository.GetRunMeta(ctx, runID)
		if err == nil {
			result, listErr := repository.ListFailures(ctx, runID, page)
			if listErr != nil {
				return contract.FailurePage{}, listErr
			}
			if meta.FailedCount > 0 && len(result.Items) == 0 && result.NextCursor == "" {
				failedKey, _ := repository.store.FailedKey(runID)
				size, sizeErr := repository.redis.Do(ctx, "HLEN", failedKey)
				if sizeErr != nil {
					return contract.FailurePage{}, sizeErr
				}
				if size.Int64() == 0 {
					return contract.FailurePage{}, contract.ErrFailureDetailsUnavailable
				}
			}
			return result, nil
		} else if !errors.Is(err, contract.ErrNotFound) {
			return contract.FailurePage{}, err
		}
	}
	view, err := c.GetRun(ctx, runID)
	if err != nil {
		return contract.FailurePage{}, err
	}
	if view.Run.FailedCount == 0 {
		return contract.FailurePage{Items: []contract.Failure{}}, nil
	}
	for _, repository := range c.repositories {
		failedKey, _ := repository.store.FailedKey(runID)
		size, sizeErr := repository.redis.Do(ctx, "HLEN", failedKey)
		if sizeErr != nil {
			return contract.FailurePage{}, sizeErr
		}
		if size.Int64() > 0 {
			return repository.ListFailures(ctx, runID, page)
		}
	}
	return contract.FailurePage{}, contract.ErrFailureDetailsUnavailable
}

func (c *Catalog) listArchives(ctx context.Context, byID map[string]contract.RunView) error {
	root := c.repositories[0].root
	if strings.TrimSpace(root) == "" {
		return nil
	}
	matches, err := filepath.Glob(filepath.Join(root, "run-*.json"))
	if err != nil {
		return err
	}
	for _, path := range matches {
		data, readErr := os.ReadFile(path)
		if readErr != nil {
			g.Log().Warningf(ctx, "event=full_sync_list_archive status=skip path=%s error=%q", path, readErr.Error())
			continue
		}
		var archive contract.RunArchive
		if decodeErr := json.Unmarshal(data, &archive); decodeErr != nil || archive.Run.RunID == "" || filepath.Base(path) != "run-"+archive.Run.RunID+".json" {
			g.Log().Warningf(ctx, "event=full_sync_list_archive status=skip path=%s error=%v", path, decodeErr)
			continue
		}
		if _, exists := byID[archive.Run.RunID]; !exists {
			byID[archive.Run.RunID] = contract.RunView{Source: contract.SourceArchive, Run: archive.Run, Rounds: archive.Rounds}
		}
	}
	return nil
}

func (r *Repository) liveRun(ctx context.Context, runID string) (contract.RunView, error) {
	meta, err := r.GetRunMeta(ctx, runID)
	if err != nil {
		return contract.RunView{}, err
	}
	rounds, err := r.GetRoundStats(ctx, runID)
	if err != nil {
		return contract.RunView{}, err
	}
	return contract.RunView{Source: contract.SourceLive, Run: meta, Rounds: rounds}, nil
}

func runSortTime(run contract.RunMeta) (value time.Time) {
	if run.StartedAt != nil && !run.StartedAt.IsZero() {
		return *run.StartedAt
	}
	if !run.CompletedAt.IsZero() {
		return run.CompletedAt
	}
	return run.CreatedAt
}