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 }