T3.3: Add state-mutating operations to app.Service
Add src/app/operations.go with the seven intents that make the Service the sole writer of job and runtime state: CreateJob, UpdateJob, DeleteJob, SetEnabled, RunNow, SetGlobalPause, UpdateSettings. Each returns error, persists through the store, and announces changes via RunRecorded/JobChanged/SchedulerStateChanged events. Extend the Service with a parsed-schedule cache, a global paused flag, an injectable runJob seam (defaults to runner.RunJob) for testing the run-now path, and a lifecycle ctx. Run and next-run timing now live in the Service (duplicating the scheduler temporarily); T3.4 converts the scheduler to drive the Service and removes the duplication. Autostart is left to the caller until T5.2's injectable Manager; async save errors in the run goroutine remain deferred to T5.1. Adds 12 tests covering create/update/delete, enable/pause, global pause, run-now with a fake runner, and settings persistence/validation. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
+1
-1
@@ -262,7 +262,7 @@ Track progress here. Mark tasks complete as they land and pass review.
|
|||||||
### Phase 3 — Application service layer
|
### Phase 3 — Application service layer
|
||||||
- [x] T3.1 — Create `src/app/service.go`; owns state behind mutex
|
- [x] T3.1 — Create `src/app/service.go`; owns state behind mutex
|
||||||
- [x] T3.2 — Add `src/app/events.go`; Event types + Observer
|
- [x] T3.2 — Add `src/app/events.go`; Event types + Observer
|
||||||
- [ ] T3.3 — Add state-mutating operations to service
|
- [x] T3.3 — Add state-mutating operations to service
|
||||||
- [ ] T3.4 — Convert `scheduler` to use service; inject Clock
|
- [ ] T3.4 — Convert `scheduler` to use service; inject Clock
|
||||||
- [ ] T3.5 — Move display helpers to `src/app/format.go`
|
- [ ] T3.5 — Move display helpers to `src/app/format.go`
|
||||||
- [ ] T3.6 — Add `src/app` unit tests (no Fyne)
|
- [ ] T3.6 — Add `src/app` unit tests (no Fyne)
|
||||||
|
|||||||
@@ -0,0 +1,444 @@
|
|||||||
|
package app
|
||||||
|
|
||||||
|
import (
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.mixdep.ru/mix/gosentry/src/domain"
|
||||||
|
"gitea.mixdep.ru/mix/gosentry/src/runner"
|
||||||
|
)
|
||||||
|
|
||||||
|
// maxJobLogs bounds the in-memory activity list kept per job. The full history
|
||||||
|
// lives in the log files on disk; this is only the recent activity shown in the
|
||||||
|
// GUI, so an old run aging out of the list is intentional.
|
||||||
|
const maxJobLogs = 50
|
||||||
|
|
||||||
|
// timestampLayout matches the format used for run records so UI-action activity
|
||||||
|
// and command runs line up in the History view.
|
||||||
|
const timestampLayout = "2006-01-02 15:04:05"
|
||||||
|
|
||||||
|
// errJobNotFound is returned by the mutating operations when no loaded job has
|
||||||
|
// the requested ID.
|
||||||
|
var errJobNotFound = errors.New("job not found")
|
||||||
|
|
||||||
|
// CreateJob normalizes and validates the supplied configuration, assigns the
|
||||||
|
// next free ID, and adds it to the loaded set. It returns the stored job (with
|
||||||
|
// its assigned ID) so the caller can select it. The job is persisted and a
|
||||||
|
// "Created" activity record is emitted.
|
||||||
|
func (s *Service) CreateJob(job domain.Job) (domain.Job, error) {
|
||||||
|
normalizeJob(&job)
|
||||||
|
if err := validateJob(job); err != nil {
|
||||||
|
return domain.Job{}, err
|
||||||
|
}
|
||||||
|
|
||||||
|
s.mu.Lock()
|
||||||
|
job.ID = s.nextIDLocked()
|
||||||
|
s.jobs = append(s.jobs, job)
|
||||||
|
runtime := domain.NewRuntime(job)
|
||||||
|
s.runtimes[job.ID] = runtime
|
||||||
|
s.parseScheduleLocked(&job)
|
||||||
|
record := uiRecord(job.ID, job.Name, "Created", "Job was added")
|
||||||
|
prependLog(runtime, record)
|
||||||
|
err := s.store.SaveJobs(s.jobs)
|
||||||
|
s.mu.Unlock()
|
||||||
|
|
||||||
|
s.emit(RunRecorded{Record: record})
|
||||||
|
s.emit(JobChanged{JobID: job.ID})
|
||||||
|
return job, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// UpdateJob replaces the durable configuration of the job with the same ID,
|
||||||
|
// keeping its runtime state (keyed by ID) and recomputing its next run. The job
|
||||||
|
// is persisted and an "Updated" activity record is emitted.
|
||||||
|
func (s *Service) UpdateJob(job domain.Job) error {
|
||||||
|
normalizeJob(&job)
|
||||||
|
if err := validateJob(job); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
s.mu.Lock()
|
||||||
|
existing := s.findByIDLocked(job.ID)
|
||||||
|
if existing == nil {
|
||||||
|
s.mu.Unlock()
|
||||||
|
return fmt.Errorf("update job %d: %w", job.ID, errJobNotFound)
|
||||||
|
}
|
||||||
|
*existing = job
|
||||||
|
runtime := s.runtimeForLocked(existing)
|
||||||
|
// An edit may have toggled Enabled; reflect that into the status the same way
|
||||||
|
// a dedicated enable/disable would, then recompute the next run.
|
||||||
|
if job.Enabled {
|
||||||
|
if runtime.LastState == "" || runtime.LastState == "Paused" {
|
||||||
|
runtime.LastState = "Ready"
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
runtime.LastState = "Paused"
|
||||||
|
}
|
||||||
|
s.parseScheduleLocked(existing)
|
||||||
|
s.refreshNextRunLocked(existing, runtime)
|
||||||
|
record := uiRecord(job.ID, job.Name, "Updated", "Job settings changed")
|
||||||
|
prependLog(runtime, record)
|
||||||
|
err := s.store.SaveJobs(s.jobs)
|
||||||
|
s.mu.Unlock()
|
||||||
|
|
||||||
|
s.emit(RunRecorded{Record: record})
|
||||||
|
s.emit(JobChanged{JobID: job.ID})
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeleteJob removes the job with the given ID along with its runtime and cached
|
||||||
|
// schedule. The remaining jobs are persisted and a "Deleted" activity record is
|
||||||
|
// emitted. The JobChanged event carries a zero ID to signal a broad change.
|
||||||
|
func (s *Service) DeleteJob(id int) error {
|
||||||
|
s.mu.Lock()
|
||||||
|
index := s.indexByIDLocked(id)
|
||||||
|
if index < 0 {
|
||||||
|
s.mu.Unlock()
|
||||||
|
return fmt.Errorf("delete job %d: %w", id, errJobNotFound)
|
||||||
|
}
|
||||||
|
deleted := s.jobs[index]
|
||||||
|
s.jobs = append(s.jobs[:index], s.jobs[index+1:]...)
|
||||||
|
delete(s.runtimes, id)
|
||||||
|
delete(s.schedules, id)
|
||||||
|
record := uiRecord(id, deleted.Name, "Deleted", "Job was removed")
|
||||||
|
err := s.store.SaveJobs(s.jobs)
|
||||||
|
s.mu.Unlock()
|
||||||
|
|
||||||
|
s.emit(RunRecorded{Record: record})
|
||||||
|
s.emit(JobChanged{JobID: 0})
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetEnabled enables or disables a single job. Enabling moves it back to "Ready"
|
||||||
|
// and recomputes its next run (respecting the global pause); disabling parks it
|
||||||
|
// at "Paused". The job is persisted and a "Resumed"/"Paused" activity record is
|
||||||
|
// emitted.
|
||||||
|
func (s *Service) SetEnabled(id int, enabled bool) error {
|
||||||
|
s.mu.Lock()
|
||||||
|
job := s.findByIDLocked(id)
|
||||||
|
if job == nil {
|
||||||
|
s.mu.Unlock()
|
||||||
|
return fmt.Errorf("set enabled job %d: %w", id, errJobNotFound)
|
||||||
|
}
|
||||||
|
job.Enabled = enabled
|
||||||
|
runtime := s.runtimeForLocked(job)
|
||||||
|
s.parseScheduleLocked(job)
|
||||||
|
|
||||||
|
var record domain.RunRecord
|
||||||
|
if enabled {
|
||||||
|
runtime.LastState = "Ready"
|
||||||
|
s.refreshNextRunLocked(job, runtime)
|
||||||
|
record = uiRecord(id, job.Name, "Resumed", "Job was enabled")
|
||||||
|
} else {
|
||||||
|
runtime.LastState = "Paused"
|
||||||
|
runtime.NextRun = "Paused"
|
||||||
|
runtime.NextDue = time.Time{}
|
||||||
|
record = uiRecord(id, job.Name, "Paused", "Job was disabled")
|
||||||
|
}
|
||||||
|
prependLog(runtime, record)
|
||||||
|
err := s.store.SaveJobs(s.jobs)
|
||||||
|
s.mu.Unlock()
|
||||||
|
|
||||||
|
s.emit(RunRecorded{Record: record})
|
||||||
|
s.emit(JobChanged{JobID: id})
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetGlobalPause flips the global pause that gates all execution, scheduled and
|
||||||
|
// manual. Each enabled job's next-run text reflects the new state immediately so
|
||||||
|
// the list view is understandable before the next tick. A "Paused"/"Resumed"
|
||||||
|
// scheduler activity record and a SchedulerStateChanged event are emitted.
|
||||||
|
func (s *Service) SetGlobalPause(paused bool) error {
|
||||||
|
s.mu.Lock()
|
||||||
|
s.paused = paused
|
||||||
|
now := time.Now()
|
||||||
|
for index := range s.jobs {
|
||||||
|
job := &s.jobs[index]
|
||||||
|
runtime := s.runtimeForLocked(job)
|
||||||
|
s.refreshNextRunFromLocked(job, runtime, now)
|
||||||
|
}
|
||||||
|
err := s.store.SaveJobs(s.jobs)
|
||||||
|
s.mu.Unlock()
|
||||||
|
|
||||||
|
state, detail := "Resumed", "All job execution resumed"
|
||||||
|
if paused {
|
||||||
|
state, detail = "Paused", "All job execution paused"
|
||||||
|
}
|
||||||
|
s.emit(RunRecorded{Record: uiRecord(0, "Scheduler", state, detail)})
|
||||||
|
s.emit(SchedulerStateChanged{Paused: paused})
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// RunNow starts a manual run of a job. It refuses to run while globally paused —
|
||||||
|
// the pause is an emergency stop for all execution — and will not start a job
|
||||||
|
// that is already running. The run itself happens on a background goroutine that
|
||||||
|
// records the result through the Service, so RunNow returns as soon as the run
|
||||||
|
// is started. The error reports why a run could not be started (or a failure to
|
||||||
|
// persist the "Running" status), not the run's own outcome.
|
||||||
|
func (s *Service) RunNow(id int) error {
|
||||||
|
s.mu.Lock()
|
||||||
|
if s.paused {
|
||||||
|
s.mu.Unlock()
|
||||||
|
return errors.New("scheduler is paused")
|
||||||
|
}
|
||||||
|
job := s.findByIDLocked(id)
|
||||||
|
if job == nil {
|
||||||
|
s.mu.Unlock()
|
||||||
|
return fmt.Errorf("run job %d: %w", id, errJobNotFound)
|
||||||
|
}
|
||||||
|
runtime := s.runtimeForLocked(job)
|
||||||
|
if runtime.LastState == "Running" {
|
||||||
|
s.mu.Unlock()
|
||||||
|
return fmt.Errorf("job %d is already running", id)
|
||||||
|
}
|
||||||
|
err := s.startRunLocked(job, runtime, "Manual")
|
||||||
|
s.mu.Unlock()
|
||||||
|
|
||||||
|
// Reflect the "Running" transition; the run's completion emits again later.
|
||||||
|
s.emit(JobChanged{JobID: id})
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// UpdateSettings validates and persists a new application configuration. The
|
||||||
|
// loaded jobs are re-saved because the jobs directory may have changed, and log
|
||||||
|
// cleanup runs so a tightened retention policy takes effect immediately.
|
||||||
|
// Autostart is intentionally left to the caller until T5.2 introduces an
|
||||||
|
// injectable autostart.Manager.
|
||||||
|
func (s *Service) UpdateSettings(config domain.Config) error {
|
||||||
|
if err := validateConfig(config); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
s.mu.Lock()
|
||||||
|
s.store.Config = config
|
||||||
|
if err := s.store.SaveConfig(); err != nil {
|
||||||
|
s.mu.Unlock()
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
// SaveConfig re-resolved the paths from the new config, so SaveJobs writes to
|
||||||
|
// the (possibly new) jobs directory and cleanup targets the new logs dir.
|
||||||
|
if err := s.store.SaveJobs(s.jobs); err != nil {
|
||||||
|
s.mu.Unlock()
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
logsDir := s.store.Paths.LogsDir
|
||||||
|
maxFiles := s.store.Config.MaxLogFiles
|
||||||
|
maxAge := s.store.Config.MaxLogAgeDays
|
||||||
|
s.mu.Unlock()
|
||||||
|
|
||||||
|
return runner.CleanupLogs(logsDir, maxFiles, maxAge)
|
||||||
|
}
|
||||||
|
|
||||||
|
// startRunLocked transitions a job to "Running", persists that, and launches the
|
||||||
|
// run on a background goroutine. The caller must hold mu.
|
||||||
|
func (s *Service) startRunLocked(job *domain.Job, runtime *domain.JobRuntime, trigger string) error {
|
||||||
|
jobCopy := *job
|
||||||
|
runtime.LastState = "Running"
|
||||||
|
runtime.NextRun = "Running"
|
||||||
|
runtime.Output = runningOutput(jobCopy, trigger, time.Now())
|
||||||
|
runtime.NextDue = time.Time{}
|
||||||
|
err := s.store.SaveJobs(s.jobs)
|
||||||
|
go s.executeRun(jobCopy, trigger)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// executeRun runs the job off the lock, then records the result back through the
|
||||||
|
// Service under the lock and announces it. It runs on its own goroutine.
|
||||||
|
func (s *Service) executeRun(jobCopy domain.Job, trigger string) {
|
||||||
|
record := s.runJob(s.ctx, &jobCopy, trigger, s.store.Paths.LogsDir)
|
||||||
|
|
||||||
|
s.mu.Lock()
|
||||||
|
if current := s.findByIDLocked(jobCopy.ID); current != nil {
|
||||||
|
runtime := s.runtimeForLocked(current)
|
||||||
|
runtime.LastRun = record.Time
|
||||||
|
runtime.LastState = record.State
|
||||||
|
runtime.Output = record.Output
|
||||||
|
prependLog(runtime, record)
|
||||||
|
s.refreshNextRunLocked(current, runtime)
|
||||||
|
// Async save errors cannot be returned to a caller; surfacing them is
|
||||||
|
// deferred to T5.1 along with the rest of the swallowed saves.
|
||||||
|
_ = runner.CleanupLogs(s.store.Paths.LogsDir, s.store.Config.MaxLogFiles, s.store.Config.MaxLogAgeDays)
|
||||||
|
_ = s.store.SaveJobs(s.jobs)
|
||||||
|
}
|
||||||
|
s.mu.Unlock()
|
||||||
|
|
||||||
|
s.emit(RunRecorded{Record: record})
|
||||||
|
s.emit(JobChanged{JobID: jobCopy.ID})
|
||||||
|
}
|
||||||
|
|
||||||
|
// refreshNextRunLocked recomputes a job's next-run display from the current time,
|
||||||
|
// honoring enabled/paused state. The caller must hold mu.
|
||||||
|
func (s *Service) refreshNextRunLocked(job *domain.Job, runtime *domain.JobRuntime) {
|
||||||
|
s.refreshNextRunFromLocked(job, runtime, time.Now())
|
||||||
|
}
|
||||||
|
|
||||||
|
// refreshNextRunFromLocked is refreshNextRunLocked with an explicit reference
|
||||||
|
// time, used when one timestamp should drive a whole batch (e.g. a global
|
||||||
|
// pause). The caller must hold mu.
|
||||||
|
func (s *Service) refreshNextRunFromLocked(job *domain.Job, runtime *domain.JobRuntime, from time.Time) {
|
||||||
|
if !job.Enabled {
|
||||||
|
runtime.NextRun = "Paused"
|
||||||
|
runtime.NextDue = time.Time{}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if s.paused {
|
||||||
|
runtime.NextRun = "Scheduler paused"
|
||||||
|
runtime.NextDue = time.Time{}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
s.prepareNextRunLocked(job, runtime, from)
|
||||||
|
}
|
||||||
|
|
||||||
|
// prepareNextRunLocked computes the concrete next-due time from the cached
|
||||||
|
// schedule. A missing cache entry means the schedule string was unparseable.
|
||||||
|
// The caller must hold mu.
|
||||||
|
func (s *Service) prepareNextRunLocked(job *domain.Job, runtime *domain.JobRuntime, from time.Time) {
|
||||||
|
sched, ok := s.schedules[job.ID]
|
||||||
|
if !ok {
|
||||||
|
runtime.NextRun = "Invalid schedule"
|
||||||
|
runtime.NextDue = time.Time{}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
runtime.NextDue = sched.Next(from)
|
||||||
|
runtime.NextRun = runtime.NextDue.Format(timestampLayout)
|
||||||
|
}
|
||||||
|
|
||||||
|
// parseScheduleLocked caches a parsed schedule for the job, dropping the cache
|
||||||
|
// entry when the schedule string is invalid so prepareNextRunLocked can tell the
|
||||||
|
// two apart. The caller must hold mu.
|
||||||
|
func (s *Service) parseScheduleLocked(job *domain.Job) {
|
||||||
|
sched, err := domain.Parse(job.Schedule)
|
||||||
|
if err != nil {
|
||||||
|
delete(s.schedules, job.ID)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
s.schedules[job.ID] = sched
|
||||||
|
}
|
||||||
|
|
||||||
|
// findByIDLocked returns a pointer into the jobs slice for the job with the
|
||||||
|
// given ID, or nil. The caller must hold mu.
|
||||||
|
func (s *Service) findByIDLocked(id int) *domain.Job {
|
||||||
|
index := s.indexByIDLocked(id)
|
||||||
|
if index < 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return &s.jobs[index]
|
||||||
|
}
|
||||||
|
|
||||||
|
// indexByIDLocked returns the slice index of the job with the given ID, or -1.
|
||||||
|
// The caller must hold mu.
|
||||||
|
func (s *Service) indexByIDLocked(id int) int {
|
||||||
|
for index := range s.jobs {
|
||||||
|
if s.jobs[index].ID == id {
|
||||||
|
return index
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return -1
|
||||||
|
}
|
||||||
|
|
||||||
|
// runtimeForLocked returns the runtime for a job, lazily creating it if missing
|
||||||
|
// so the Service stays robust if a job lacks an entry. The caller must hold mu.
|
||||||
|
func (s *Service) runtimeForLocked(job *domain.Job) *domain.JobRuntime {
|
||||||
|
runtime, ok := s.runtimes[job.ID]
|
||||||
|
if !ok || runtime == nil {
|
||||||
|
runtime = domain.NewRuntime(*job)
|
||||||
|
s.runtimes[job.ID] = runtime
|
||||||
|
}
|
||||||
|
return runtime
|
||||||
|
}
|
||||||
|
|
||||||
|
// nextIDLocked returns the smallest ID greater than every loaded job's ID. The
|
||||||
|
// caller must hold mu.
|
||||||
|
func (s *Service) nextIDLocked() int {
|
||||||
|
next := 1
|
||||||
|
for index := range s.jobs {
|
||||||
|
if s.jobs[index].ID >= next {
|
||||||
|
next = s.jobs[index].ID + 1
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return next
|
||||||
|
}
|
||||||
|
|
||||||
|
// prependLog adds a record to the front of a runtime's activity list and caps
|
||||||
|
// its length so it cannot grow without bound.
|
||||||
|
func prependLog(runtime *domain.JobRuntime, record domain.RunRecord) {
|
||||||
|
runtime.Logs = append([]domain.RunRecord{record}, runtime.Logs...)
|
||||||
|
if len(runtime.Logs) > maxJobLogs {
|
||||||
|
runtime.Logs = runtime.Logs[:maxJobLogs]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// uiRecord builds an activity record for a user/Service action, using the same
|
||||||
|
// timestamp shape and "UI" trigger as the GUI did so History stays consistent.
|
||||||
|
func uiRecord(jobID int, jobName string, state string, detail string) domain.RunRecord {
|
||||||
|
return domain.RunRecord{
|
||||||
|
Time: time.Now().Format(timestampLayout),
|
||||||
|
JobID: jobID,
|
||||||
|
JobName: jobName,
|
||||||
|
Trigger: "UI",
|
||||||
|
State: state,
|
||||||
|
Detail: detail,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// runningOutput is the placeholder output shown while a job is running, before
|
||||||
|
// the real command output replaces it.
|
||||||
|
func runningOutput(job domain.Job, trigger string, started time.Time) string {
|
||||||
|
var builder strings.Builder
|
||||||
|
builder.WriteString("status:\n")
|
||||||
|
builder.WriteString("Running since " + started.Format(timestampLayout) + "\n\n")
|
||||||
|
builder.WriteString("trigger:\n")
|
||||||
|
builder.WriteString(trigger + "\n\n")
|
||||||
|
builder.WriteString("command:\n")
|
||||||
|
builder.WriteString(job.Command + "\n\n")
|
||||||
|
builder.WriteString("arguments:\n")
|
||||||
|
builder.WriteString(runner.LogArguments(job.Arguments))
|
||||||
|
builder.WriteString("\n\nsuccess_exit_codes:\n")
|
||||||
|
builder.WriteString(runner.SuccessExitCodesText(job))
|
||||||
|
builder.WriteString("\n\nstart_only:\n")
|
||||||
|
builder.WriteString(fmt.Sprintf("%t", job.StartOnly))
|
||||||
|
return builder.String()
|
||||||
|
}
|
||||||
|
|
||||||
|
// normalizeJob trims user-entered fields and applies the same defaults the job
|
||||||
|
// dialog used, so callers do not have to.
|
||||||
|
func normalizeJob(job *domain.Job) {
|
||||||
|
job.Name = strings.TrimSpace(job.Name)
|
||||||
|
job.Folder = strings.TrimSpace(job.Folder)
|
||||||
|
job.Schedule = strings.TrimSpace(job.Schedule)
|
||||||
|
job.Command = strings.TrimSpace(job.Command)
|
||||||
|
job.Arguments = strings.TrimSpace(job.Arguments)
|
||||||
|
job.SuccessExitCodes = strings.TrimSpace(job.SuccessExitCodes)
|
||||||
|
if job.SuccessExitCodes == "" {
|
||||||
|
job.SuccessExitCodes = "0"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// validateJob enforces the minimum executable definition: name, schedule, and
|
||||||
|
// command must be present. Folder is optional. The schedule string itself is not
|
||||||
|
// rejected for being unparseable — that surfaces later as an "Invalid schedule"
|
||||||
|
// next-run, matching the prior behavior.
|
||||||
|
func validateJob(job domain.Job) error {
|
||||||
|
if job.Name == "" || job.Schedule == "" || job.Command == "" {
|
||||||
|
return errors.New("name, schedule, and command are required")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// validateConfig rejects settings that would break persistence or cleanup.
|
||||||
|
func validateConfig(config domain.Config) error {
|
||||||
|
if strings.TrimSpace(config.JobsDir) == "" {
|
||||||
|
return errors.New("jobs directory is required")
|
||||||
|
}
|
||||||
|
if strings.TrimSpace(config.LogsDir) == "" {
|
||||||
|
return errors.New("logs directory is required")
|
||||||
|
}
|
||||||
|
if config.MaxLogFiles <= 0 {
|
||||||
|
return errors.New("max log files must be a positive number")
|
||||||
|
}
|
||||||
|
if config.MaxLogAgeDays <= 0 {
|
||||||
|
return errors.New("max log age days must be a positive number")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
@@ -0,0 +1,271 @@
|
|||||||
|
package app
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.mixdep.ru/mix/gosentry/src/domain"
|
||||||
|
"gitea.mixdep.ru/mix/gosentry/src/storage"
|
||||||
|
)
|
||||||
|
|
||||||
|
// newTempService builds a Service backed by a store rooted in a temp directory,
|
||||||
|
// so the mutating operations can persist to real (throwaway) files.
|
||||||
|
func newTempService(t *testing.T, jobs []domain.Job) *Service {
|
||||||
|
t.Helper()
|
||||||
|
dir := t.TempDir()
|
||||||
|
store := &storage.Store{
|
||||||
|
Paths: storage.Paths{
|
||||||
|
ExecutablePath: filepath.Join(dir, "gosentry"),
|
||||||
|
AppDir: dir,
|
||||||
|
ConfigPath: filepath.Join(dir, "gosentry.yaml"),
|
||||||
|
JobsDir: dir,
|
||||||
|
JobsPath: filepath.Join(dir, "jobs.yaml"),
|
||||||
|
LogsDir: filepath.Join(dir, "logs"),
|
||||||
|
},
|
||||||
|
Config: domain.Config{JobsDir: ".", LogsDir: "logs", MaxLogFiles: 100, MaxLogAgeDays: 30},
|
||||||
|
}
|
||||||
|
return NewService(store, jobs)
|
||||||
|
}
|
||||||
|
|
||||||
|
// recorder is a test observer that captures every emitted event.
|
||||||
|
type recorder struct {
|
||||||
|
events []Event
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *recorder) OnEvent(e Event) { r.events = append(r.events, e) }
|
||||||
|
|
||||||
|
func (r *recorder) jobChanged() (ids []int) {
|
||||||
|
for _, e := range r.events {
|
||||||
|
if jc, ok := e.(JobChanged); ok {
|
||||||
|
ids = append(ids, jc.JobID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return ids
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *recorder) records() (out []domain.RunRecord) {
|
||||||
|
for _, e := range r.events {
|
||||||
|
if rr, ok := e.(RunRecorded); ok {
|
||||||
|
out = append(out, rr.Record)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCreateJobAssignsIDAndEmits(t *testing.T) {
|
||||||
|
svc := newTempService(t, nil)
|
||||||
|
rec := &recorder{}
|
||||||
|
svc.Subscribe(rec)
|
||||||
|
|
||||||
|
created, err := svc.CreateJob(domain.Job{Name: "Build", Schedule: "@every 1m", Command: "echo hi", Enabled: true})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("CreateJob: %v", err)
|
||||||
|
}
|
||||||
|
if created.ID != 1 {
|
||||||
|
t.Errorf("first job ID = %d, want 1", created.ID)
|
||||||
|
}
|
||||||
|
if got := svc.Jobs(); len(got) != 1 || got[0].Name != "Build" {
|
||||||
|
t.Fatalf("jobs after create = %+v", got)
|
||||||
|
}
|
||||||
|
if rt := svc.Runtime(1); rt == nil || rt.LastState != "Ready" {
|
||||||
|
t.Errorf("runtime = %+v, want LastState Ready", rt)
|
||||||
|
}
|
||||||
|
if recs := rec.records(); len(recs) != 1 || recs[0].State != "Created" {
|
||||||
|
t.Errorf("records = %+v, want one Created", recs)
|
||||||
|
}
|
||||||
|
if ids := rec.jobChanged(); len(ids) != 1 || ids[0] != 1 {
|
||||||
|
t.Errorf("JobChanged ids = %v, want [1]", ids)
|
||||||
|
}
|
||||||
|
|
||||||
|
// A second job takes the next free ID.
|
||||||
|
second, err := svc.CreateJob(domain.Job{Name: "Two", Schedule: "@every 1m", Command: "echo two"})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("CreateJob 2: %v", err)
|
||||||
|
}
|
||||||
|
if second.ID != 2 {
|
||||||
|
t.Errorf("second job ID = %d, want 2", second.ID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCreateJobValidates(t *testing.T) {
|
||||||
|
svc := newTempService(t, nil)
|
||||||
|
if _, err := svc.CreateJob(domain.Job{Schedule: "@every 1m", Command: "echo"}); err == nil {
|
||||||
|
t.Error("expected error for missing name")
|
||||||
|
}
|
||||||
|
if got := svc.Jobs(); len(got) != 0 {
|
||||||
|
t.Errorf("invalid job should not be stored, jobs = %+v", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUpdateJobKeepsRuntimeAndReflectsDisable(t *testing.T) {
|
||||||
|
svc := newTempService(t, []domain.Job{{ID: 5, Name: "Old", Schedule: "@every 1m", Command: "echo", Enabled: true}})
|
||||||
|
if err := svc.UpdateJob(domain.Job{ID: 5, Name: "New", Schedule: "@every 1m", Command: "echo", Enabled: false}); err != nil {
|
||||||
|
t.Fatalf("UpdateJob: %v", err)
|
||||||
|
}
|
||||||
|
got := svc.Jobs()
|
||||||
|
if got[0].Name != "New" || got[0].Enabled {
|
||||||
|
t.Errorf("job after update = %+v", got[0])
|
||||||
|
}
|
||||||
|
if rt := svc.Runtime(5); rt == nil || rt.LastState != "Paused" || rt.NextRun != "Paused" {
|
||||||
|
t.Errorf("runtime after disable = %+v", rt)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUpdateJobNotFound(t *testing.T) {
|
||||||
|
svc := newTempService(t, nil)
|
||||||
|
if err := svc.UpdateJob(domain.Job{ID: 99, Name: "X", Schedule: "@every 1m", Command: "echo"}); err == nil {
|
||||||
|
t.Error("expected not-found error")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeleteJobRemovesEverything(t *testing.T) {
|
||||||
|
svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}})
|
||||||
|
rec := &recorder{}
|
||||||
|
svc.Subscribe(rec)
|
||||||
|
|
||||||
|
if err := svc.DeleteJob(1); err != nil {
|
||||||
|
t.Fatalf("DeleteJob: %v", err)
|
||||||
|
}
|
||||||
|
if got := svc.Jobs(); len(got) != 0 {
|
||||||
|
t.Errorf("jobs after delete = %+v", got)
|
||||||
|
}
|
||||||
|
if rt := svc.Runtime(1); rt != nil {
|
||||||
|
t.Errorf("runtime should be gone, got %+v", rt)
|
||||||
|
}
|
||||||
|
if recs := rec.records(); len(recs) != 1 || recs[0].State != "Deleted" {
|
||||||
|
t.Errorf("records = %+v, want one Deleted", recs)
|
||||||
|
}
|
||||||
|
if ids := rec.jobChanged(); len(ids) != 1 || ids[0] != 0 {
|
||||||
|
t.Errorf("JobChanged ids = %v, want [0] (broad)", ids)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSetEnabledToggles(t *testing.T) {
|
||||||
|
svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: false}})
|
||||||
|
|
||||||
|
if err := svc.SetEnabled(1, true); err != nil {
|
||||||
|
t.Fatalf("SetEnabled true: %v", err)
|
||||||
|
}
|
||||||
|
if rt := svc.Runtime(1); rt.LastState != "Ready" || rt.NextDue.IsZero() {
|
||||||
|
t.Errorf("enabled runtime = %+v, want Ready with a next-due", rt)
|
||||||
|
}
|
||||||
|
if err := svc.SetEnabled(1, false); err != nil {
|
||||||
|
t.Fatalf("SetEnabled false: %v", err)
|
||||||
|
}
|
||||||
|
if rt := svc.Runtime(1); rt.LastState != "Paused" || !rt.NextDue.IsZero() {
|
||||||
|
t.Errorf("disabled runtime = %+v, want Paused with no next-due", rt)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSetGlobalPauseUpdatesRuntimesAndEmits(t *testing.T) {
|
||||||
|
svc := newTempService(t, []domain.Job{
|
||||||
|
{ID: 1, Name: "On", Schedule: "@every 1m", Command: "echo", Enabled: true},
|
||||||
|
{ID: 2, Name: "Off", Schedule: "@every 1m", Command: "echo", Enabled: false},
|
||||||
|
})
|
||||||
|
rec := &recorder{}
|
||||||
|
svc.Subscribe(rec)
|
||||||
|
|
||||||
|
if err := svc.SetGlobalPause(true); err != nil {
|
||||||
|
t.Fatalf("SetGlobalPause: %v", err)
|
||||||
|
}
|
||||||
|
if rt := svc.Runtime(1); rt.NextRun != "Scheduler paused" {
|
||||||
|
t.Errorf("enabled job next-run = %q, want %q", rt.NextRun, "Scheduler paused")
|
||||||
|
}
|
||||||
|
if rt := svc.Runtime(2); rt.NextRun != "Paused" {
|
||||||
|
t.Errorf("disabled job next-run = %q, want %q", rt.NextRun, "Paused")
|
||||||
|
}
|
||||||
|
var sawState bool
|
||||||
|
for _, e := range rec.events {
|
||||||
|
if ss, ok := e.(SchedulerStateChanged); ok && ss.Paused {
|
||||||
|
sawState = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !sawState {
|
||||||
|
t.Error("expected a SchedulerStateChanged{Paused:true} event")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Resuming recomputes a real next run for the enabled job.
|
||||||
|
if err := svc.SetGlobalPause(false); err != nil {
|
||||||
|
t.Fatalf("resume: %v", err)
|
||||||
|
}
|
||||||
|
if rt := svc.Runtime(1); rt.NextDue.IsZero() {
|
||||||
|
t.Errorf("resumed enabled job should have a next-due, got %+v", rt)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunNowUsesRunnerAndRecords(t *testing.T) {
|
||||||
|
svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}})
|
||||||
|
|
||||||
|
done := make(chan domain.RunRecord, 1)
|
||||||
|
svc.runJob = func(_ context.Context, job *domain.Job, trigger string, _ string) domain.RunRecord {
|
||||||
|
if trigger != "Manual" {
|
||||||
|
t.Errorf("trigger = %q, want Manual", trigger)
|
||||||
|
}
|
||||||
|
return domain.RunRecord{Time: "2026-06-19 12:00:00", JobID: job.ID, JobName: job.Name, State: "Success", Output: "ok"}
|
||||||
|
}
|
||||||
|
svc.Subscribe(ObserverFunc(func(e Event) {
|
||||||
|
if rr, ok := e.(RunRecorded); ok && rr.Record.JobID == 1 {
|
||||||
|
select {
|
||||||
|
case done <- rr.Record:
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}))
|
||||||
|
|
||||||
|
if err := svc.RunNow(1); err != nil {
|
||||||
|
t.Fatalf("RunNow: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
select {
|
||||||
|
case record := <-done:
|
||||||
|
if record.State != "Success" {
|
||||||
|
t.Errorf("recorded state = %q, want Success", record.State)
|
||||||
|
}
|
||||||
|
case <-time.After(2 * time.Second):
|
||||||
|
t.Fatal("timed out waiting for run to be recorded")
|
||||||
|
}
|
||||||
|
|
||||||
|
if rt := svc.Runtime(1); rt.LastState != "Success" || rt.Output != "ok" {
|
||||||
|
t.Errorf("runtime after run = %+v", rt)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunNowRefusedWhilePaused(t *testing.T) {
|
||||||
|
svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}})
|
||||||
|
var ran bool
|
||||||
|
svc.runJob = func(context.Context, *domain.Job, string, string) domain.RunRecord {
|
||||||
|
ran = true
|
||||||
|
return domain.RunRecord{}
|
||||||
|
}
|
||||||
|
if err := svc.SetGlobalPause(true); err != nil {
|
||||||
|
t.Fatalf("SetGlobalPause: %v", err)
|
||||||
|
}
|
||||||
|
if err := svc.RunNow(1); err == nil {
|
||||||
|
t.Error("expected RunNow to be refused while paused")
|
||||||
|
}
|
||||||
|
if ran {
|
||||||
|
t.Error("runner must not be invoked while paused")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUpdateSettingsPersistsAndValidates(t *testing.T) {
|
||||||
|
svc := newTempService(t, nil)
|
||||||
|
|
||||||
|
bad := svc.store.Config
|
||||||
|
bad.MaxLogFiles = 0
|
||||||
|
if err := svc.UpdateSettings(bad); err == nil {
|
||||||
|
t.Error("expected validation error for non-positive max log files")
|
||||||
|
}
|
||||||
|
|
||||||
|
good := svc.store.Config
|
||||||
|
good.NotifyOnFailure = false
|
||||||
|
good.MaxLogAgeDays = 7
|
||||||
|
if err := svc.UpdateSettings(good); err != nil {
|
||||||
|
t.Fatalf("UpdateSettings: %v", err)
|
||||||
|
}
|
||||||
|
if svc.Store().Config.MaxLogAgeDays != 7 || svc.Store().Config.NotifyOnFailure {
|
||||||
|
t.Errorf("config not applied: %+v", svc.Store().Config)
|
||||||
|
}
|
||||||
|
}
|
||||||
+34
-8
@@ -1,9 +1,11 @@
|
|||||||
package app
|
package app
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
"gitea.mixdep.ru/mix/gosentry/src/domain"
|
"gitea.mixdep.ru/mix/gosentry/src/domain"
|
||||||
|
"gitea.mixdep.ru/mix/gosentry/src/runner"
|
||||||
"gitea.mixdep.ru/mix/gosentry/src/storage"
|
"gitea.mixdep.ru/mix/gosentry/src/storage"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -13,21 +15,38 @@ import (
|
|||||||
// to that state goes through a mutex so the GUI and the scheduler can no longer
|
// to that state goes through a mutex so the GUI and the scheduler can no longer
|
||||||
// race on a shared *[]Job.
|
// race on a shared *[]Job.
|
||||||
//
|
//
|
||||||
// This is the first slice of the layer (T3.1): it establishes ownership and the
|
// State ownership and the locking contract were established in T3.1; the
|
||||||
// locking contract. State-mutating intents (CreateJob, RunNow, SetGlobalPause,
|
// event/observer machinery in T3.2. T3.3 adds the state-mutating intents
|
||||||
// ...) and the event/observer machinery are added in later tasks; for now the
|
// (CreateJob, UpdateJob, DeleteJob, SetEnabled, RunNow, SetGlobalPause,
|
||||||
// Service only owns state and exposes read snapshots.
|
// UpdateSettings) in operations.go: the Service is now the sole writer of job
|
||||||
|
// and runtime state, persisting through the store and announcing changes via
|
||||||
|
// events.
|
||||||
//
|
//
|
||||||
// Locking contract: mu is a plain, non-reentrant mutex. Exported methods take
|
// Locking contract: mu is a plain, non-reentrant mutex. Exported methods take
|
||||||
// it; unexported helpers ending in "Locked" assume the caller already holds it.
|
// it; unexported helpers ending in "Locked" assume the caller already holds it.
|
||||||
// The Service must never call back into the UI (or any code that might re-enter
|
// The Service must never call back into the UI (or any code that might re-enter
|
||||||
// the Service) while holding mu.
|
// the Service) while holding mu — in particular emit() is always called after
|
||||||
|
// mu is released.
|
||||||
type Service struct {
|
type Service struct {
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
store *storage.Store
|
store *storage.Store
|
||||||
jobs []domain.Job
|
jobs []domain.Job
|
||||||
runtimes map[int]*domain.JobRuntime
|
runtimes map[int]*domain.JobRuntime
|
||||||
|
|
||||||
|
// schedules caches a parsed Schedule per job ID so timing math does not
|
||||||
|
// re-parse the schedule string on every use. paused is the global pause flag.
|
||||||
|
// Both are guarded by mu. (The scheduler still keeps its own copy until T3.4
|
||||||
|
// converts it to drive the Service instead of sharing state.)
|
||||||
|
schedules map[int]domain.Schedule
|
||||||
|
paused bool
|
||||||
|
|
||||||
|
// runJob is the run seam. It defaults to runner.RunJob and is overridden in
|
||||||
|
// tests with a fake so the run-now path can be exercised without spawning real
|
||||||
|
// processes. ctx is the lifecycle context passed to runs; T3.4 wires a
|
||||||
|
// cancelable Start/Stop, for now it is context.Background().
|
||||||
|
runJob func(ctx context.Context, job *domain.Job, trigger string, logsDir string) domain.RunRecord
|
||||||
|
ctx context.Context
|
||||||
|
|
||||||
// observers and their guard live in events.go. dispatchMu is separate from mu
|
// observers and their guard live in events.go. dispatchMu is separate from mu
|
||||||
// so that emitting an event never requires (or is held under) the state lock:
|
// so that emitting an event never requires (or is held under) the state lock:
|
||||||
// the Service must release mu before dispatching, per the locking contract.
|
// the Service must release mu before dispatching, per the locking contract.
|
||||||
@@ -37,14 +56,21 @@ type Service struct {
|
|||||||
|
|
||||||
// NewService wires the Service to a loaded store and its jobs. It builds the
|
// NewService wires the Service to a loaded store and its jobs. It builds the
|
||||||
// initial runtime map from the durable jobs so every job has transient state
|
// initial runtime map from the durable jobs so every job has transient state
|
||||||
// from the moment the Service exists. The store is the Service's sole channel
|
// from the moment the Service exists, and parses each job's schedule once. The
|
||||||
// to persistence.
|
// store is the Service's sole channel to persistence.
|
||||||
func NewService(store *storage.Store, jobs []domain.Job) *Service {
|
func NewService(store *storage.Store, jobs []domain.Job) *Service {
|
||||||
return &Service{
|
s := &Service{
|
||||||
store: store,
|
store: store,
|
||||||
jobs: jobs,
|
jobs: jobs,
|
||||||
runtimes: domain.NewRuntimes(jobs),
|
runtimes: domain.NewRuntimes(jobs),
|
||||||
|
schedules: make(map[int]domain.Schedule, len(jobs)),
|
||||||
|
runJob: runner.RunJob,
|
||||||
|
ctx: context.Background(),
|
||||||
}
|
}
|
||||||
|
for index := range s.jobs {
|
||||||
|
s.parseScheduleLocked(&s.jobs[index])
|
||||||
|
}
|
||||||
|
return s
|
||||||
}
|
}
|
||||||
|
|
||||||
// Open loads the store and constructs a Service from it in one step. It is the
|
// Open loads the store and constructs a Service from it in one step. It is the
|
||||||
|
|||||||
Reference in New Issue
Block a user