19288366423

Committed 12 Nov 2025 06:19AM UTC coverage: 73.246% (-0.2%) from 73.473%

Build # 19288366423

Build Type

Pull #51

github

Committed by

umputun

Commit Message

chore: set default exec-max-lines to 100, disable if web not enabled

Pull Request Pull Request #51: Add execution history with SQLite storage and web UI display

Run Details

181 of 234 new or added lines in 5 files covered. (77.35%)

3 existing lines in 2 files now uncovered.

2464 of 3364 relevant lines covered (73.25%)

36.37 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.78

/app/web/persistence/sqlite.go

package persistence

import (
        "context"
        "fmt"
        "sync"
        "time"

        "github.com/jmoiron/sqlx"
        _ "modernc.org/sqlite" // sqlite driver

        "github.com/umputun/cronn/app/service/request"
        "github.com/umputun/cronn/app/web/enums"
)

// JobInfo represents a cron job with its execution state
type JobInfo struct {
        ID         string          `db:"id"`
        Command    string          `db:"command"`
        Schedule   string          `db:"schedule"`
        NextRun    time.Time       `db:"next_run"`
        LastRun    time.Time       `db:"last_run"`
        LastStatus enums.JobStatus `db:"last_status"`
        IsRunning  bool            `db:"-"` // not stored in DB
        Enabled    bool            `db:"enabled"`
        CreatedAt  time.Time       `db:"created_at"`
        UpdatedAt  time.Time       `db:"updated_at"`
        SortIndex  int             `db:"sort_index"`
}

// ExecutionInfo represents a single job execution record
type ExecutionInfo struct {
        ID              int             `db:"id"`
        JobID           string          `db:"job_id"`
        StartedAt       time.Time       `db:"started_at"`
        FinishedAt      time.Time       `db:"finished_at"`
        Status          enums.JobStatus `db:"status"`
        ExitCode        int             `db:"exit_code"`
        ExecutedCommand string          `db:"executed_command"`
        Output          string          `db:"output"`
}

// SQLiteStore implements persistence using SQLite
type SQLiteStore struct {
        db *sqlx.DB
        mu sync.RWMutex // protects concurrent database access
}

// NewSQLiteStore creates a new SQLite store and initializes the database
func NewSQLiteStore(dbPath string) (*SQLiteStore, error) {
        db, err := sqlx.Open("sqlite", dbPath)
        if err != nil {
                return nil, fmt.Errorf("failed to open database: %w", err)
        }

        // helper to execute pragma with proper error handling
        execPragma := func(pragma, errMsgPrefix string) error {
                if _, err := db.Exec(pragma); err != nil {
                        if closeErr := db.Close(); closeErr != nil {
                                return fmt.Errorf("%s: %w (also failed to close db: %v)", errMsgPrefix, err, closeErr)
                        }
                        return fmt.Errorf("%s: %w", errMsgPrefix, err)
                }
                return nil
        }

        // enable WAL mode for better concurrency
        if err := execPragma("PRAGMA journal_mode=WAL", "failed to set WAL mode"); err != nil {
                return nil, err
        }

        // set busy timeout to wait when database is locked
        if err := execPragma("PRAGMA busy_timeout=5000", "failed to set busy timeout"); err != nil {
                return nil, err
        }

        store := &SQLiteStore{db: db}

        // initialize database tables
        if err := store.initialize(); err != nil {
                _ = db.Close()
                return nil, fmt.Errorf("failed to initialize database: %w", err)
        }

        return store, nil
}

// initialize creates the database schema
func (s *SQLiteStore) initialize() error {
        queries := []string{
                `CREATE TABLE IF NOT EXISTS jobs (
                        id TEXT PRIMARY KEY,
                        command TEXT NOT NULL,
                        schedule TEXT NOT NULL,
                        next_run DATETIME,
                        last_run DATETIME,
                        last_status TEXT,
                        enabled BOOLEAN DEFAULT 1,
                        created_at DATETIME,
                        updated_at DATETIME,
                        sort_index INTEGER DEFAULT 0
                )`,
                `CREATE TABLE IF NOT EXISTS executions (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        job_id TEXT,
                        started_at DATETIME,
                        finished_at DATETIME,
                        status TEXT,
                        exit_code INTEGER,
                        executed_command TEXT,
                        output TEXT,
                        FOREIGN KEY (job_id) REFERENCES jobs(id)
                )`,
                `CREATE INDEX IF NOT EXISTS idx_executions_job_id ON executions(job_id)`,
                `CREATE INDEX IF NOT EXISTS idx_executions_started_at ON executions(started_at)`,
        }

        for _, query := range queries {
                if _, err := s.db.Exec(query); err != nil {
                        return fmt.Errorf("failed to execute query: %w", err)
                }
        }

        // run schema migrations
        if err := s.migrate(); err != nil {
                return fmt.Errorf("failed to run migrations: %w", err)
        }

        return nil
}

// migrate performs schema migrations for existing databases
func (s *SQLiteStore) migrate() error {
        // check if executed_command column exists
        var executedCommandExists bool
        err := s.db.QueryRow(`
                SELECT COUNT(*) > 0
                FROM pragma_table_info('executions')
                WHERE name = 'executed_command'
        `).Scan(&executedCommandExists)
        if err != nil {
                return fmt.Errorf("failed to check for executed_command column: %w", err)
        }

        // add executed_command column if it doesn't exist
        if !executedCommandExists {
                _, err = s.db.Exec("ALTER TABLE executions ADD COLUMN executed_command TEXT DEFAULT ''")
                if err != nil {
                        return fmt.Errorf("failed to add executed_command column: %w", err)
                }
        }

        // check if output column exists
        var outputExists bool
        err = s.db.QueryRow(`
                SELECT COUNT(*) > 0
                FROM pragma_table_info('executions')
                WHERE name = 'output'
        `).Scan(&outputExists)
        if err != nil {
                return fmt.Errorf("failed to check for output column: %w", err)
        }

        // add output column if it doesn't exist
        if !outputExists {
                if _, err := s.db.Exec("ALTER TABLE executions ADD COLUMN output TEXT DEFAULT ''"); err != nil {
                        return fmt.Errorf("failed to add output column: %w", err)
                }
        }

        return nil
}

// LoadJobs retrieves all jobs from the database
func (s *SQLiteStore) LoadJobs() ([]JobInfo, error) {
        s.mu.RLock()
        defer s.mu.RUnlock()

        var jobs []JobInfo
        err := s.db.Select(&jobs, `
                SELECT id, command, schedule, next_run, last_run, last_status, enabled, 
                       created_at, updated_at, sort_index
                FROM jobs
                ORDER BY sort_index`)
        if err != nil {
                return nil, fmt.Errorf("failed to query jobs: %w", err)
        }

        // ensure we return empty slice, not nil
        if jobs == nil {
                jobs = []JobInfo{}
        }

        return jobs, nil
}

// SaveJobs persists multiple jobs in a transaction
func (s *SQLiteStore) SaveJobs(jobs []JobInfo) error {
        s.mu.Lock()
        defer s.mu.Unlock()

        tx, err := s.db.Beginx()
        if err != nil {
                return fmt.Errorf("failed to begin transaction: %w", err)
        }
        defer tx.Rollback()

        for idx, job := range jobs {
                // set sort_index based on position
                job.SortIndex = idx

                _, err := tx.NamedExec(`
                        INSERT OR REPLACE INTO jobs 
                        (id, command, schedule, next_run, last_run, last_status, enabled, created_at, updated_at, sort_index)
                        VALUES (:id, :command, :schedule, :next_run, :last_run, :last_status, :enabled, :created_at, :updated_at, :sort_index)`,
                        job)
                if err != nil {
                        return fmt.Errorf("failed to save job %s: %w", job.ID, err)
                }
        }

        if err := tx.Commit(); err != nil {
                return fmt.Errorf("failed to commit transaction: %w", err)
        }

        return nil
}

// RecordExecution logs a job execution event
func (s *SQLiteStore) RecordExecution(req request.RecordExecution) error {
        s.mu.Lock()
        defer s.mu.Unlock()

        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()

        _, err := s.db.ExecContext(ctx, `
                INSERT INTO executions (job_id, started_at, finished_at, status, exit_code, executed_command, output)
                VALUES (?, ?, ?, ?, ?, ?, ?)`,
                req.JobID, req.StartedAt, req.FinishedAt, req.Status.String(), req.ExitCode, req.ExecutedCommand, req.Output)

        if err != nil {
                return fmt.Errorf("failed to record execution: %w", err)
        }

        return nil
}

// GetExecutions retrieves execution history for a job, limited to the most recent executions
func (s *SQLiteStore) GetExecutions(jobID string, limit int) ([]ExecutionInfo, error) {
        s.mu.RLock()
        defer s.mu.RUnlock()

        var executions []ExecutionInfo
        err := s.db.Select(&executions, `
                SELECT id, job_id, started_at, finished_at, status, exit_code, executed_command, output
                FROM executions
                WHERE job_id = ?
                ORDER BY started_at DESC
                LIMIT ?`,
                jobID, limit)

        if err != nil {
                return nil, fmt.Errorf("failed to query executions: %w", err)
        }

        // ensure we return empty slice, not nil
        if executions == nil {
                executions = []ExecutionInfo{}
        }

        return executions, nil
}

// GetExecutionByID retrieves a specific execution by its ID
func (s *SQLiteStore) GetExecutionByID(execID int) (ExecutionInfo, error) {
        s.mu.RLock()
        defer s.mu.RUnlock()

        var execution ExecutionInfo
        err := s.db.Get(&execution, `
                SELECT id, job_id, started_at, finished_at, status, exit_code, executed_command, output
                FROM executions
                WHERE id = ?`,
                execID)

        if err != nil {
                return ExecutionInfo{}, fmt.Errorf("failed to query execution: %w", err)
        }

        return execution, nil
}

// CleanupOldExecutions removes old executions beyond the limit for a job
func (s *SQLiteStore) CleanupOldExecutions(jobID string, limit int) error {
        s.mu.Lock()
        defer s.mu.Unlock()

        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()

        // delete executions beyond the limit, keeping the most recent ones
        _, err := s.db.ExecContext(ctx, `
                DELETE FROM executions
                WHERE job_id = ?
                AND id NOT IN (
                        SELECT id FROM executions
                        WHERE job_id = ?
                        ORDER BY started_at DESC
                        LIMIT ?
                )`,
                jobID, jobID, limit)

        if err != nil {
                return fmt.Errorf("failed to cleanup old executions: %w", err)
        }

        return nil
}

// Close closes the database connection
func (s *SQLiteStore) Close() error {
        s.mu.Lock()
        defer s.mu.Unlock()

        if err := s.db.Close(); err != nil {
                return fmt.Errorf("failed to close database: %w", err)
        }
        return nil
}

1	package persistence
2
3	import (
4	"context"
5	"fmt"
6	"sync"
7	"time"
8
9	"github.com/jmoiron/sqlx"
10	_ "modernc.org/sqlite" // sqlite driver
11
12	"github.com/umputun/cronn/app/service/request"
13	"github.com/umputun/cronn/app/web/enums"
14	)
15
16	// JobInfo represents a cron job with its execution state
17	type JobInfo struct {
18	ID string `db:"id"`
19	Command string `db:"command"`
20	Schedule string `db:"schedule"`
21	NextRun time.Time `db:"next_run"`
22	LastRun time.Time `db:"last_run"`
23	LastStatus enums.JobStatus `db:"last_status"`
24	IsRunning bool `db:"-"` // not stored in DB
25	Enabled bool `db:"enabled"`
26	CreatedAt time.Time `db:"created_at"`
27	UpdatedAt time.Time `db:"updated_at"`
28	SortIndex int `db:"sort_index"`
29	}
30
31	// ExecutionInfo represents a single job execution record
32	type ExecutionInfo struct {
33	ID int `db:"id"`
34	JobID string `db:"job_id"`
35	StartedAt time.Time `db:"started_at"`
36	FinishedAt time.Time `db:"finished_at"`
37	Status enums.JobStatus `db:"status"`
38	ExitCode int `db:"exit_code"`
39	ExecutedCommand string `db:"executed_command"`
40	Output string `db:"output"`
41	}
42
43	// SQLiteStore implements persistence using SQLite
44	type SQLiteStore struct {
45	db *sqlx.DB
46	mu sync.RWMutex // protects concurrent database access
47	}
48
49	// NewSQLiteStore creates a new SQLite store and initializes the database
50	func NewSQLiteStore(dbPath string) (*SQLiteStore, error) {	23✔
51	db, err := sqlx.Open("sqlite", dbPath)	23✔
52	if err != nil {	23✔
53	return nil, fmt.Errorf("failed to open database: %w", err)	×
54	}	×
55
56	// helper to execute pragma with proper error handling
57	execPragma := func(pragma, errMsgPrefix string) error {	68✔
58	if _, err := db.Exec(pragma); err != nil {	46✔
59	if closeErr := db.Close(); closeErr != nil {	1✔
60	return fmt.Errorf("%s: %w (also failed to close db: %v)", errMsgPrefix, err, closeErr)	×
61	}	×
62	return fmt.Errorf("%s: %w", errMsgPrefix, err)	1✔
63	}
64	return nil	44✔
65	}
66
67	// enable WAL mode for better concurrency
68	if err := execPragma("PRAGMA journal_mode=WAL", "failed to set WAL mode"); err != nil {	24✔
69	return nil, err	1✔
70	}	1✔
71
72	// set busy timeout to wait when database is locked
73	if err := execPragma("PRAGMA busy_timeout=5000", "failed to set busy timeout"); err != nil {	22✔
74	return nil, err	×
75	}	×
76
77	store := &SQLiteStore{db: db}	22✔
78		22✔
79	// initialize database tables	22✔
80	if err := store.initialize(); err != nil {	22✔
81	_ = db.Close()	×
82	return nil, fmt.Errorf("failed to initialize database: %w", err)	×
83	}	×
84
85	return store, nil	22✔
86	}
87
88	// initialize creates the database schema
89	func (s *SQLiteStore) initialize() error {	22✔
90	queries := []string{	22✔
91	`CREATE TABLE IF NOT EXISTS jobs (	22✔
92	id TEXT PRIMARY KEY,	22✔
93	command TEXT NOT NULL,	22✔
94	schedule TEXT NOT NULL,	22✔
95	next_run DATETIME,	22✔
96	last_run DATETIME,	22✔
97	last_status TEXT,	22✔
98	enabled BOOLEAN DEFAULT 1,	22✔
99	created_at DATETIME,	22✔
100	updated_at DATETIME,	22✔
101	sort_index INTEGER DEFAULT 0	22✔
102	)`,	22✔
103	`CREATE TABLE IF NOT EXISTS executions (	22✔
104	id INTEGER PRIMARY KEY AUTOINCREMENT,	22✔
105	job_id TEXT,	22✔
106	started_at DATETIME,	22✔
107	finished_at DATETIME,	22✔
108	status TEXT,	22✔
109	exit_code INTEGER,	22✔
110	executed_command TEXT,	22✔
111	output TEXT,	22✔
112	FOREIGN KEY (job_id) REFERENCES jobs(id)	22✔
113	)`,	22✔
114	`CREATE INDEX IF NOT EXISTS idx_executions_job_id ON executions(job_id)`,	22✔
115	`CREATE INDEX IF NOT EXISTS idx_executions_started_at ON executions(started_at)`,	22✔
116	}	22✔
117		22✔
118	for _, query := range queries {	110✔
119	if _, err := s.db.Exec(query); err != nil {	88✔
120	return fmt.Errorf("failed to execute query: %w", err)	×
121	}	×
122	}
123
124	// run schema migrations
125	if err := s.migrate(); err != nil {	22✔
126	return fmt.Errorf("failed to run migrations: %w", err)	×
127	}	×
128
129	return nil	22✔
130	}
131
132	// migrate performs schema migrations for existing databases
133	func (s *SQLiteStore) migrate() error {	22✔
134	// check if executed_command column exists	22✔
135	var executedCommandExists bool	22✔
136	err := s.db.QueryRow(`	22✔
137	SELECT COUNT(*) > 0	22✔
138	FROM pragma_table_info('executions')	22✔
139	WHERE name = 'executed_command'	22✔
140	`).Scan(&executedCommandExists)	22✔
141	if err != nil {	22✔
142	return fmt.Errorf("failed to check for executed_command column: %w", err)	×
143	}	×
144
145	// add executed_command column if it doesn't exist
146	if !executedCommandExists {	23✔
147	_, err = s.db.Exec("ALTER TABLE executions ADD COLUMN executed_command TEXT DEFAULT ''")	1✔
148	if err != nil {	1✔
149	return fmt.Errorf("failed to add executed_command column: %w", err)	×
150	}	×
151	}
152
153	// check if output column exists
154	var outputExists bool	22✔
155	err = s.db.QueryRow(`	22✔
156	SELECT COUNT(*) > 0	22✔
157	FROM pragma_table_info('executions')	22✔
158	WHERE name = 'output'	22✔
159	`).Scan(&outputExists)	22✔
160	if err != nil {	22✔
NEW 161	return fmt.Errorf("failed to check for output column: %w", err)	×
NEW 162	}	×
163
164	// add output column if it doesn't exist
165	if !outputExists {	24✔
166	if _, err := s.db.Exec("ALTER TABLE executions ADD COLUMN output TEXT DEFAULT ''"); err != nil {	2✔
NEW 167	return fmt.Errorf("failed to add output column: %w", err)	×
NEW 168	}	×
169	}
170
171	return nil	22✔
172	}
173
174	// LoadJobs retrieves all jobs from the database
175	func (s *SQLiteStore) LoadJobs() ([]JobInfo, error) {	4✔
176	s.mu.RLock()	4✔
177	defer s.mu.RUnlock()	4✔
178		4✔
179	var jobs []JobInfo	4✔
180	err := s.db.Select(&jobs, `	4✔
181	SELECT id, command, schedule, next_run, last_run, last_status, enabled,	4✔
182	created_at, updated_at, sort_index	4✔
183	FROM jobs	4✔
184	ORDER BY sort_index`)	4✔
185	if err != nil {	5✔
186	return nil, fmt.Errorf("failed to query jobs: %w", err)	1✔
187	}	1✔
188
189	// ensure we return empty slice, not nil
190	if jobs == nil {	4✔
191	jobs = []JobInfo{}	1✔
192	}	1✔
193
194	return jobs, nil	3✔
195	}
196
197	// SaveJobs persists multiple jobs in a transaction
198	func (s *SQLiteStore) SaveJobs(jobs []JobInfo) error {	4✔
199	s.mu.Lock()	4✔
200	defer s.mu.Unlock()	4✔
201		4✔
202	tx, err := s.db.Beginx()	4✔
203	if err != nil {	4✔
204	return fmt.Errorf("failed to begin transaction: %w", err)	×
205	}	×
206	defer tx.Rollback()	4✔
207		4✔
208	for idx, job := range jobs {	9✔
209	// set sort_index based on position	5✔
210	job.SortIndex = idx	5✔
211		5✔
212	_, err := tx.NamedExec(`	5✔
213	INSERT OR REPLACE INTO jobs	5✔
214	(id, command, schedule, next_run, last_run, last_status, enabled, created_at, updated_at, sort_index)	5✔
215	VALUES (:id, :command, :schedule, :next_run, :last_run, :last_status, :enabled, :created_at, :updated_at, :sort_index)`,	5✔
216	job)	5✔
217	if err != nil {	6✔
218	return fmt.Errorf("failed to save job %s: %w", job.ID, err)	1✔
219	}	1✔
220	}
221
222	if err := tx.Commit(); err != nil {	3✔
223	return fmt.Errorf("failed to commit transaction: %w", err)	×
224	}	×
225
226	return nil	3✔
227	}
228
229	// RecordExecution logs a job execution event
230	func (s *SQLiteStore) RecordExecution(req request.RecordExecution) error {	54✔
231	s.mu.Lock()	54✔
232	defer s.mu.Unlock()	54✔
233		54✔
234	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)	54✔
235	defer cancel()	54✔
236		54✔
237	_, err := s.db.ExecContext(ctx, `	54✔
238	INSERT INTO executions (job_id, started_at, finished_at, status, exit_code, executed_command, output)	54✔
239	VALUES (?, ?, ?, ?, ?, ?, ?)`,	54✔
240	req.JobID, req.StartedAt, req.FinishedAt, req.Status.String(), req.ExitCode, req.ExecutedCommand, req.Output)	54✔
241		54✔
242	if err != nil {	54✔
243	return fmt.Errorf("failed to record execution: %w", err)	×
244	}	×
245
246	return nil	54✔
247	}
248
249	// GetExecutions retrieves execution history for a job, limited to the most recent executions
250	func (s *SQLiteStore) GetExecutions(jobID string, limit int) ([]ExecutionInfo, error) {	14✔
251	s.mu.RLock()	14✔
252	defer s.mu.RUnlock()	14✔
253		14✔
254	var executions []ExecutionInfo	14✔
255	err := s.db.Select(&executions, `	14✔
256	SELECT id, job_id, started_at, finished_at, status, exit_code, executed_command, output	14✔
257	FROM executions	14✔
258	WHERE job_id = ?	14✔
259	ORDER BY started_at DESC	14✔
260	LIMIT ?`,	14✔
261	jobID, limit)	14✔
262		14✔
263	if err != nil {	15✔
264	return nil, fmt.Errorf("failed to query executions: %w", err)	1✔
265	}	1✔
266
267	// ensure we return empty slice, not nil
268	if executions == nil {	14✔
269	executions = []ExecutionInfo{}	1✔
270	}	1✔
271
272	return executions, nil	13✔
273	}
274
275	// GetExecutionByID retrieves a specific execution by its ID
276	func (s *SQLiteStore) GetExecutionByID(execID int) (ExecutionInfo, error) {	2✔
277	s.mu.RLock()	2✔
278	defer s.mu.RUnlock()	2✔
279		2✔
280	var execution ExecutionInfo	2✔
281	err := s.db.Get(&execution, `	2✔
282	SELECT id, job_id, started_at, finished_at, status, exit_code, executed_command, output	2✔
283	FROM executions	2✔
284	WHERE id = ?`,	2✔
285	execID)	2✔
286		2✔
287	if err != nil {	3✔
288	return ExecutionInfo{}, fmt.Errorf("failed to query execution: %w", err)	1✔
289	}	1✔
290
291	return execution, nil	1✔
292	}
293
294	// CleanupOldExecutions removes old executions beyond the limit for a job
295	func (s *SQLiteStore) CleanupOldExecutions(jobID string, limit int) error {	3✔
296	s.mu.Lock()	3✔
297	defer s.mu.Unlock()	3✔
298		3✔
299	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)	3✔
300	defer cancel()	3✔
301		3✔
302	// delete executions beyond the limit, keeping the most recent ones	3✔
303	_, err := s.db.ExecContext(ctx, `	3✔
304	DELETE FROM executions	3✔
305	WHERE job_id = ?	3✔
306	AND id NOT IN (	3✔
307	SELECT id FROM executions	3✔
308	WHERE job_id = ?	3✔
309	ORDER BY started_at DESC	3✔
310	LIMIT ?	3✔
311	)`,	3✔
312	jobID, jobID, limit)	3✔
313		3✔
314	if err != nil {	3✔
NEW 315	return fmt.Errorf("failed to cleanup old executions: %w", err)	×
NEW 316	}	×
317
318	return nil	3✔
319	}
320
321	// Close closes the database connection
322	func (s *SQLiteStore) Close() error {	22✔
323	s.mu.Lock()	22✔
324	defer s.mu.Unlock()	22✔
325		22✔
326	if err := s.db.Close(); err != nil {	22✔
327	return fmt.Errorf("failed to close database: %w", err)	×
328	}	×
329	return nil	22✔
330	}

umputun / cronn / 19288366423

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous