task-processor/internal/processor/service.go

package processor

import (
	log "github.com/sirupsen/logrus"
	"sync"
	"task-processor/internal/appState"
	"task-processor/internal/shared"
)

func (p *Processor) StartWork(receivedTasks []shared.TaskResponse) {
	log.Info("Starting work...")
	p.state.ResetCounters()

	if len(receivedTasks) == 0 {
		p.state.SetStatus(appState.StatusIdle)
		return
	}

	p.state.SetStatus(appState.StatusWorkInProgress)
	p.state.SetTasksReceived(len(receivedTasks))

	sorted := p.sortTasks(receivedTasks)

	var wg sync.WaitGroup
	for origin, tasks := range sorted {
		wg.Add(1)
		go func(origin string, tasks []shared.Task) {
			defer wg.Done()
			p.handlers[origin].HandleTasks(tasks, p.out, p.state)
		}(origin, tasks)
	}
	wg.Wait()

	log.Debug("All goroutines finished")
	p.state.SetLastCheck()
	p.state.SetStatus(appState.StatusIdle)
	log.Debugf("State | %+v", p.state)
}

func (p *Processor) sortTasks(receivedTasks []shared.TaskResponse) map[string][]shared.Task {
	sorted := make(map[string][]shared.Task, len(receivedTasks))
	log.WithField("count", len(receivedTasks)).Debug("Sorting tasks")
	for _, task := range receivedTasks {
		if task.OriginSurugayaLink != "" {
			log.WithFields(log.Fields{
				"origin": shared.OriginSurugaya,
				"uuid":   task.MerchUuid,
			}).Debug("Task")

			sorted[shared.OriginSurugaya] = append(sorted[shared.OriginSurugaya], shared.Task{
				MerchUuid: task.MerchUuid,
				Origin:    shared.OriginSurugaya,
				Link:      task.OriginSurugayaLink,
			})
		}

		if task.OriginMandarakeLink != "" {
			log.WithFields(log.Fields{
				"origin": shared.OriginMandarake,
				"uuid":   task.MerchUuid,
			}).Debug("Task")

			sorted[shared.OriginMandarake] = append(sorted[shared.OriginMandarake], shared.Task{
				MerchUuid:  task.MerchUuid,
				Origin:     shared.OriginMandarake,
				Link:       task.OriginMandarakeLink,
				RetryCount: 3,
			})
		}
	}
	return sorted
}