From d8ab9acf7e9313813a9c5971009048712a9d1d54 Mon Sep 17 00:00:00 2001 From: mixeme Date: Fri, 19 Jun 2026 07:48:41 +0300 Subject: [PATCH] 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 --- docs/REFACTORING.md | 2 +- src/app/operations.go | 444 +++++++++++++++++++++++++++++++++++++ src/app/operations_test.go | 271 ++++++++++++++++++++++ src/app/service.go | 48 +++- 4 files changed, 753 insertions(+), 12 deletions(-) create mode 100644 src/app/operations.go create mode 100644 src/app/operations_test.go diff --git a/docs/REFACTORING.md b/docs/REFACTORING.md index 0134cf9..74cb86b 100644 --- a/docs/REFACTORING.md +++ b/docs/REFACTORING.md @@ -262,7 +262,7 @@ Track progress here. Mark tasks complete as they land and pass review. ### Phase 3 — Application service layer - [x] T3.1 — Create `src/app/service.go`; owns state behind mutex - [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.5 — Move display helpers to `src/app/format.go` - [ ] T3.6 — Add `src/app` unit tests (no Fyne) diff --git a/src/app/operations.go b/src/app/operations.go new file mode 100644 index 0000000..ad3da03 --- /dev/null +++ b/src/app/operations.go @@ -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 +} diff --git a/src/app/operations_test.go b/src/app/operations_test.go new file mode 100644 index 0000000..be466e9 --- /dev/null +++ b/src/app/operations_test.go @@ -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) + } +} diff --git a/src/app/service.go b/src/app/service.go index c8bda47..e803e4d 100644 --- a/src/app/service.go +++ b/src/app/service.go @@ -1,9 +1,11 @@ package app import ( + "context" "sync" "gitea.mixdep.ru/mix/gosentry/src/domain" + "gitea.mixdep.ru/mix/gosentry/src/runner" "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 // race on a shared *[]Job. // -// This is the first slice of the layer (T3.1): it establishes ownership and the -// locking contract. State-mutating intents (CreateJob, RunNow, SetGlobalPause, -// ...) and the event/observer machinery are added in later tasks; for now the -// Service only owns state and exposes read snapshots. +// State ownership and the locking contract were established in T3.1; the +// event/observer machinery in T3.2. T3.3 adds the state-mutating intents +// (CreateJob, UpdateJob, DeleteJob, SetEnabled, RunNow, SetGlobalPause, +// 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 // 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) while holding mu. +// the Service) while holding mu — in particular emit() is always called after +// mu is released. type Service struct { mu sync.Mutex store *storage.Store jobs []domain.Job 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 // 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. @@ -37,14 +56,21 @@ type Service struct { // 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 -// from the moment the Service exists. The store is the Service's sole channel -// to persistence. +// from the moment the Service exists, and parses each job's schedule once. The +// store is the Service's sole channel to persistence. func NewService(store *storage.Store, jobs []domain.Job) *Service { - return &Service{ - store: store, - jobs: jobs, - runtimes: domain.NewRuntimes(jobs), + s := &Service{ + store: store, + jobs: 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