init repo

internal/job/job.go

@@ -0,0 +1,166 @@
+package job
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/rs/zerolog/log"
+)
+
+type State int
+
+const (
+	Pending State = iota
+	Running
+	Success
+	Failed
+	Abort
+	Unknown
+)
+
+const UnknownState = "unknown"
+
+var JobExecTimeout = 10 * time.Second
+
+func (s State) String() string {
+	switch s {
+	case Pending:
+		return "pending"
+	case Running:
+		return "running"
+	case Success:
+		return "success"
+	case Failed:
+		return "failed"
+	case Abort:
+		return "abort"
+	case Unknown:
+		return UnknownState
+	default:
+		return UnknownState
+	}
+}
+
+var (
+	ErrJobAborted         = errors.New("job has been aborted")
+	ErrJobNotCompletedYet = errors.New("job is not right state, retrying")
+)
+
+type FnJob func(ctx context.Context) error
+
+type JobDetails struct {
+	ID        uuid.UUID  `json:"id"`
+	State     string     `json:"state"`
+	CreatedAt time.Time  `json:"createdAt"`
+	UpdatedAt *time.Time `json:"updatedAt,omitempty"`
+	Err       string     `json:"error"`
+}
+
+// TODO(rmanach): add priority level
+type Job struct {
+	l         sync.RWMutex
+	id        uuid.UUID
+	createdAt time.Time
+	updatedAt *time.Time
+	state     State
+	task      FnJob
+	err       error
+	chAbort   chan struct{}
+}
+
+func NewJob(task FnJob, row, col int) Job {
+	return Job{
+		id:        uuid.New(),
+		createdAt: time.Now().UTC(),
+		state:     Pending,
+		task:      task,
+		chAbort:   make(chan struct{}, 1),
+	}
+}
+
+func (j *Job) IntoDetails() JobDetails {
+	j.l.RLock()
+	defer j.l.RUnlock()
+
+	jd := JobDetails{
+		ID:        j.id,
+		CreatedAt: j.createdAt,
+		State:     j.state.String(),
+	}
+
+	if err := j.err; err != nil {
+		jd.Err = err.Error()
+	}
+
+	if ut := j.updatedAt; ut != nil {
+		jd.UpdatedAt = ut
+	}
+
+	return jd
+}
+
+func (j *Job) GetID() uuid.UUID {
+	return j.id
+}
+
+func (j *Job) GetState() State {
+	j.l.RLock()
+	defer j.l.RUnlock()
+
+	return j.state
+}
+
+func (j *Job) setState(s State) {
+	j.l.Lock()
+	defer j.l.Unlock()
+
+	now := time.Now().UTC()
+	j.updatedAt = &now
+
+	j.state = s
+}
+
+func (j *Job) setFail(err error) {
+	j.l.Lock()
+	defer j.l.Unlock()
+
+	now := time.Now().UTC()
+	j.updatedAt = &now
+
+	j.state = Failed
+	j.err = err
+}
+
+func (j *Job) Abort() {
+	j.setState(Abort)
+	j.chAbort <- struct{}{}
+}
+
+func (j *Job) Run(ctx context.Context) {
+	ctxExec, fnCancel := context.WithTimeout(ctx, JobExecTimeout)
+	defer fnCancel()
+
+	j.setState(Running)
+
+	log.Info().Str("job", j.GetID().String()).Msg("job running...")
+
+	go func() {
+		for range j.chAbort {
+			j.setState(Abort)
+			fnCancel()
+		}
+	}()
+
+	if err := j.task(ctxExec); err != nil {
+		if errors.Is(err, ErrJobNotCompletedYet) {
+			j.setState(Pending)
+			return
+		}
+		j.setFail(err)
+		return
+	}
+	j.setState(Success)
+}