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 }