oto/services/core/internal/cicdstate/store.go

package cicdstate

import (
	"fmt"
	"sync"
	"time"
)

const (
	StateQueued    = "queued"
	StateRunning   = "running"
	StateSucceeded = "succeeded"
	StateFailed    = "failed"
	StateCanceled  = "canceled"
)

var validTransitions = map[string][]string{
	StateQueued:    {StateRunning, StateFailed, StateCanceled},
	StateRunning:   {StateSucceeded, StateFailed, StateCanceled},
	StateSucceeded: nil,
	StateFailed:    nil,
	StateCanceled:  nil,
}

// RunInput holds the remote run request payload stored on a job so that the
// runner transport can dispatch it later. It deliberately avoids any protobuf
// dependency; conversion to/from otopb.RunRequest happens at the HTTP or socket
// boundary.
type RunInput struct {
	PipelineYAMLPath string
	PipelineYAML     string
	Variables        map[string]string
	CommandTypes     []string
}

type Job struct {
	ID          string
	Name        string
	State       string
	CreatedAt   time.Time
	UpdatedAt   time.Time
	ExecutionID string
	RunInput    *RunInput
}

func (j *Job) transitionTo(newState string, updatedAt time.Time) error {
	allowed, ok := validTransitions[j.State]
	if !ok {
		return fmt.Errorf("unknown state: %s", j.State)
	}
	for _, s := range allowed {
		if s == newState {
			j.State = newState
			j.UpdatedAt = updatedAt
			return nil
		}
	}
	return fmt.Errorf("invalid transition from %s to %s", j.State, newState)
}

type Execution struct {
	ID        string
	JobID     string
	State     string
	RunnerID  string
	CreatedAt time.Time
	UpdatedAt time.Time
	Logs      []LogEntry
	Artifacts []ArtifactEntry
}

func (e *Execution) transitionTo(newState string, updatedAt time.Time) error {
	allowed, ok := validTransitions[e.State]
	if !ok {
		return fmt.Errorf("unknown state: %s", e.State)
	}
	for _, s := range allowed {
		if s == newState {
			e.State = newState
			e.UpdatedAt = updatedAt
			return nil
		}
	}
	return fmt.Errorf("invalid transition from %s to %s", e.State, newState)
}

type LogEntry struct {
	Timestamp time.Time
	Line      string
}

type ArtifactEntry struct {
	Name string
	Path string
}

type Store struct {
	mu         sync.RWMutex
	jobs       map[string]*Job
	executions map[string]*Execution
	now        func() time.Time
}

func NewStore() *Store {
	return &Store{
		now:        time.Now,
		jobs:       make(map[string]*Job),
		executions: make(map[string]*Execution),
	}
}

// NextExecutionID returns the next execution identifier using the store clock.
func (s *Store) NextExecutionID() string {
	n := s.now()
	return fmt.Sprintf("exec-%d", n.UnixNano())
}

func (s *Store) CreateJob(id, name string, runInput *RunInput) (*Job, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	if _, exists := s.jobs[id]; exists {
		return nil, fmt.Errorf("job %s already exists", id)
	}

	now := s.now()
	job := &Job{
		ID:        id,
		Name:      name,
		State:     StateQueued,
		CreatedAt: now,
		UpdatedAt: now,
		RunInput:  copyRunInput(runInput),
	}
	s.jobs[id] = job
	return copyJob(job), nil
}

func (s *Store) GetJob(id string) (*Job, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	job, ok := s.jobs[id]
	if !ok {
		return nil, fmt.Errorf("job not found: %s", id)
	}

	return copyJob(job), nil
}

func (s *Store) CreateExecution(jobID, execID string) (*Execution, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	job, ok := s.jobs[jobID]
	if !ok {
		return nil, fmt.Errorf("job not found: %s", jobID)
	}

	if job.ExecutionID != "" {
		return nil, fmt.Errorf("job %s already has execution %s", jobID, job.ExecutionID)
	}

	if _, exists := s.executions[execID]; exists {
		return nil, fmt.Errorf("execution %s already exists", execID)
	}

	now := s.now()
	exec := &Execution{
		ID:        execID,
		JobID:     jobID,
		State:     StateQueued,
		CreatedAt: now,
		UpdatedAt: now,
		Logs:      make([]LogEntry, 0),
		Artifacts: make([]ArtifactEntry, 0),
	}
	s.executions[execID] = exec
	job.ExecutionID = execID
	job.UpdatedAt = now

	return exec, nil
}

func (s *Store) GetExecution(id string) (*Execution, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	exec, ok := s.executions[id]
	if !ok {
		return nil, fmt.Errorf("execution not found: %s", id)
	}

	return copyExecution(exec), nil
}

func (s *Store) AppendLog(execID string, line string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	exec, ok := s.executions[execID]
	if !ok {
		return fmt.Errorf("execution not found: %s", execID)
	}

	now := s.now()
	exec.Logs = append(exec.Logs, LogEntry{
		Timestamp: now,
		Line:      line,
	})
	exec.UpdatedAt = now

	return nil
}

func (s *Store) GetLogs(execID string) ([]LogEntry, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	exec, ok := s.executions[execID]
	if !ok {
		return nil, fmt.Errorf("execution not found: %s", execID)
	}

	result := make([]LogEntry, len(exec.Logs))
	copy(result, exec.Logs)
	return result, nil
}

func (s *Store) AppendArtifact(execID string, name string, path string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	exec, ok := s.executions[execID]
	if !ok {
		return fmt.Errorf("execution not found: %s", execID)
	}

	exec.Artifacts = append(exec.Artifacts, ArtifactEntry{
		Name: name,
		Path: path,
	})
	exec.UpdatedAt = s.now()

	return nil
}

func (s *Store) GetArtifacts(execID string) ([]ArtifactEntry, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	exec, ok := s.executions[execID]
	if !ok {
		return nil, fmt.Errorf("execution not found: %s", execID)
	}

	result := make([]ArtifactEntry, len(exec.Artifacts))
	copy(result, exec.Artifacts)
	return result, nil
}

func (s *Store) TransitionJob(id, newState string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	job, ok := s.jobs[id]
	if !ok {
		return fmt.Errorf("job not found: %s", id)
	}

	return job.transitionTo(newState, s.now())
}

func (s *Store) TransitionExecution(id, newState string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	exec, ok := s.executions[id]
	if !ok {
		return fmt.Errorf("execution not found: %s", id)
	}

	return exec.transitionTo(newState, s.now())
}

func (s *Store) CancelJobExecution(jobID, execID string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	job, ok := s.jobs[jobID]
	if !ok {
		return fmt.Errorf("job not found: %s", jobID)
	}

	exec, ok := s.executions[execID]
	if !ok {
		return fmt.Errorf("execution not found: %s", execID)
	}

	if exec.JobID != jobID || job.ExecutionID != execID {
		return fmt.Errorf("job and execution mismatch")
	}

	isTerminal := func(state string) bool {
		return state == StateSucceeded || state == StateFailed || state == StateCanceled
	}
	if isTerminal(job.State) || isTerminal(exec.State) {
		return fmt.Errorf("cannot cancel terminal state (job: %s, execution: %s)", job.State, exec.State)
	}

	now := s.now()
	if err := job.transitionTo(StateCanceled, now); err != nil {
		return err
	}
	if err := exec.transitionTo(StateCanceled, now); err != nil {
		return err
	}

	return nil
}

func (s *Store) SetExecutionRunnerID(execID, runnerID string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	exec, ok := s.executions[execID]
	if !ok {
		return fmt.Errorf("execution not found: %s", execID)
	}
	exec.RunnerID = runnerID
	exec.UpdatedAt = s.now()
	return nil
}

func (s *Store) SnapshotJobs() []Job {
	s.mu.RLock()
	defer s.mu.RUnlock()

	result := make([]Job, 0, len(s.jobs))
	for _, j := range s.jobs {
		result = append(result, *copyJob(j))
	}
	return result
}

func (s *Store) SnapshotExecutions() []Execution {
	s.mu.RLock()
	defer s.mu.RUnlock()

	result := make([]Execution, 0, len(s.executions))
	for _, e := range s.executions {
		result = append(result, *copyExecution(e))
	}
	return result
}

func copyJob(j *Job) *Job {
	if j == nil {
		return nil
	}
	cp := *j
	cp.RunInput = copyRunInput(j.RunInput)
	return &cp
}

func copyRunInput(in *RunInput) *RunInput {
	if in == nil {
		return nil
	}
	cp := *in
	if in.Variables != nil {
		cp.Variables = make(map[string]string, len(in.Variables))
		for k, v := range in.Variables {
			cp.Variables[k] = v
		}
	}
	if in.CommandTypes != nil {
		cp.CommandTypes = make([]string, len(in.CommandTypes))
		copy(cp.CommandTypes, in.CommandTypes)
	}
	return &cp
}

func copyExecution(e *Execution) *Execution {
	if e == nil {
		return nil
	}
	cp := *e
	cp.Logs = make([]LogEntry, len(e.Logs))
	copy(cp.Logs, e.Logs)
	cp.Artifacts = make([]ArtifactEntry, len(e.Artifacts))
	copy(cp.Artifacts, e.Artifacts)
	return &cp
}