package models import ( "database/sql" "strings" "time" ) type JobLog struct { ID int64 `db:"id" json:"id"` JobID int64 `db:"job_id" json:"job_id"` Stream string `db:"stream" json:"stream"` Content string `db:"content" json:"content"` Timestamp time.Time `db:"timestamp" json:"timestamp"` } type JobLogRepository struct { db *sql.DB } func NewJobLogRepository(db *sql.DB) *JobLogRepository { return &JobLogRepository{db: db} } func (r *JobLogRepository) InsertBatch(jobID int64, stream string, lines []string) error { if len(lines) == 0 { return nil } tx, err := r.db.Begin() if err != nil { return err } stmt, err := tx.Prepare("INSERT INTO job_logs (job_id, stream, content) VALUES (?, ?, ?)") if err != nil { return err } defer stmt.Close() for _, line := range lines { if _, err := stmt.Exec(jobID, stream, line); err != nil { tx.Rollback() return err } } return tx.Commit() } func (r *JobLogRepository) GetByJobID(jobID int64, limit, offset int) ([]JobLog, error) { if limit <= 0 { limit = 1000 } if limit > 10000 { limit = 10000 } rows, err := r.db.Query(` SELECT id, job_id, stream, content, timestamp FROM job_logs WHERE job_id = ? ORDER BY id ASC LIMIT ? OFFSET ?`, jobID, limit, offset) if err != nil { return nil, err } defer rows.Close() var logs []JobLog for rows.Next() { var l JobLog if err := rows.Scan(&l.ID, &l.JobID, &l.Stream, &l.Content, &l.Timestamp); err != nil { return nil, err } logs = append(logs, l) } return logs, rows.Err() } func (r *JobLogRepository) CountByJobID(jobID int64) (int64, error) { var n int64 err := r.db.QueryRow("SELECT COUNT(*) FROM job_logs WHERE job_id = ?", jobID).Scan(&n) return n, err } func (r *JobLogRepository) DeleteBefore(before time.Time) (int64, error) { res, err := r.db.Exec("DELETE FROM job_logs WHERE timestamp < ?", before) if err != nil { return 0, err } return res.RowsAffected() } type JobWithStats struct { Job DurationSeconds *int64 `db:"duration_seconds" json:"duration_seconds"` LogLineCount int64 `db:"log_line_count" json:"log_line_count"` } func (r *JobLogRepository) GetAllFiltered(limit, offset int, syncPairID *int64, status, triggerType string, from, to *time.Time) ([]JobWithStats, int64, error) { where, args := []string{"1=1"}, []interface{}{} if syncPairID != nil { where = append(where, "j.sync_pair_id = ?") args = append(args, *syncPairID) } if status != "" { where = append(where, "j.status = ?") args = append(args, status) } if triggerType != "" { where = append(where, "j.trigger_type = ?") args = append(args, triggerType) } if from != nil { where = append(where, "j.created_at >= ?") args = append(args, *from) } if to != nil { where = append(where, "j.created_at <= ?") args = append(args, *to) } whereClause := strings.Join(where, " AND ") var total int64 countQuery := "SELECT COUNT(*) FROM jobs j WHERE " + whereClause if err := r.db.QueryRow(countQuery, args...).Scan(&total); err != nil { return nil, 0, err } query := ` SELECT j.id, j.sync_pair_id, j.trigger_type, j.status, j.started_at, j.finished_at, j.log_file, j.created_at, CASE WHEN j.finished_at IS NOT NULL AND j.started_at IS NOT NULL THEN (j.finished_at - j.started_at) ELSE NULL END as duration_seconds, (SELECT COUNT(*) FROM job_logs WHERE job_id = j.id) as log_line_count FROM jobs j WHERE ` + whereClause + ` ORDER BY j.created_at DESC LIMIT ? OFFSET ?` args = append(args, limit, offset) rows, err := r.db.Query(query, args...) if err != nil { return nil, 0, err } defer rows.Close() var jobs []JobWithStats for rows.Next() { var j JobWithStats var started, finished sql.NullTime var logFile sql.NullString var durationSeconds sql.NullInt64 if err := rows.Scan(&j.ID, &j.SyncPairID, &j.TriggerType, &j.Status, &started, &finished, &logFile, &j.CreatedAt, &durationSeconds, &j.LogLineCount); err != nil { return nil, 0, err } if started.Valid { j.StartedAt = &started.Time } if finished.Valid { j.FinishedAt = &finished.Time } if logFile.Valid { j.LogFile = &logFile.String } if durationSeconds.Valid { j.DurationSeconds = &durationSeconds.Int64 } jobs = append(jobs, j) } return jobs, total, rows.Err() }