move-data-nas/internal/models/job.go

package models

import (
	"database/sql"
	"fmt"
	"strings"
	"time"
)

type Job struct {
	ID              int64      `db:"id" json:"id"`
	SyncPairID      int64      `db:"sync_pair_id" json:"sync_pair_id"`
	TriggerType     string     `db:"trigger_type" json:"trigger_type"`
	Status          string     `db:"status" json:"status"`
	StartedAt       *time.Time `db:"started_at" json:"started_at"`
	FinishedAt      *time.Time `db:"finished_at" json:"finished_at"`
	LogFile         *string    `db:"log_file" json:"log_file"`
	ErrorMessage    *string    `db:"error_message" json:"error_message,omitempty"`
	ErrorCode       *string    `db:"error_code" json:"error_code,omitempty"`
	TotalSizeBytes  int64      `db:"total_size_bytes" json:"total_size_bytes,omitempty"`
	SentBytes       int64      `db:"sent_bytes" json:"sent_bytes,omitempty"`
	CreatedAt       time.Time  `db:"created_at" json:"created_at"`
}

type JobRepository struct {
	db *sql.DB
}

func NewJobRepository(db *sql.DB) *JobRepository {
	return &JobRepository{db: db}
}

func (r *JobRepository) Create(syncPairID int64, triggerType, status string) (int64, error) {
	res, err := r.db.Exec(`
		INSERT INTO jobs (sync_pair_id, trigger_type, status) VALUES (?, ?, ?)`,
		syncPairID, triggerType, status,
	)
	if err != nil {
		return 0, err
	}
	return res.LastInsertId()
}

func (r *JobRepository) GetByID(id int64) (*Job, error) {
	var j Job
	var started, finished sql.NullTime
	var logFile, errMsg, errCode sql.NullString
	err := r.db.QueryRow(`
		SELECT id, sync_pair_id, trigger_type, status, started_at, finished_at,
			log_file, error_message, error_code, total_size_bytes, sent_bytes, created_at
		FROM jobs WHERE id = ?`, id).Scan(
		&j.ID, &j.SyncPairID, &j.TriggerType, &j.Status, &started, &finished,
		&logFile, &errMsg, &errCode, &j.TotalSizeBytes, &j.SentBytes, &j.CreatedAt)
	if err != nil {
		return nil, err
	}
	if started.Valid {
		j.StartedAt = &started.Time
	}
	if finished.Valid {
		j.FinishedAt = &finished.Time
	}
	if logFile.Valid {
		j.LogFile = &logFile.String
	}
	if errMsg.Valid {
		j.ErrorMessage = &errMsg.String
	}
	if errCode.Valid {
		j.ErrorCode = &errCode.String
	}
	return &j, nil
}

func (r *JobRepository) GetAll(limit, offset int) ([]Job, error) {
	rows, err := r.db.Query(`
		SELECT id, sync_pair_id, trigger_type, status, started_at, finished_at,
			log_file, error_message, error_code, total_size_bytes, sent_bytes, created_at
		FROM jobs ORDER BY created_at DESC LIMIT ? OFFSET ?`,
		limit, offset)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var jobs []Job
	for rows.Next() {
		var j Job
		var started, finished sql.NullTime
		var logFile, errMsg, errCode sql.NullString
		if err := rows.Scan(&j.ID, &j.SyncPairID, &j.TriggerType, &j.Status,
			&started, &finished, &logFile, &errMsg, &errCode,
			&j.TotalSizeBytes, &j.SentBytes, &j.CreatedAt); err != nil {
			return nil, err
		}
		if started.Valid {
			j.StartedAt = &started.Time
		}
		if finished.Valid {
			j.FinishedAt = &finished.Time
		}
		if logFile.Valid {
			j.LogFile = &logFile.String
		}
		if errMsg.Valid {
			j.ErrorMessage = &errMsg.String
		}
		if errCode.Valid {
			j.ErrorCode = &errCode.String
		}
		jobs = append(jobs, j)
	}
	return jobs, rows.Err()
}

func (r *JobRepository) UpdateStatus(id int64, status string) error {
	var query string
	var args []interface{}
	switch status {
	case "running", "waking_up":
		query = "UPDATE jobs SET status = ?, started_at = COALESCE(started_at, CURRENT_TIMESTAMP) WHERE id = ?"
		args = []interface{}{status, id}
	case "success", "failed", "cancelled":
		query = "UPDATE jobs SET status = ?, finished_at = CURRENT_TIMESTAMP WHERE id = ?"
		args = []interface{}{status, id}
	default:
		query = "UPDATE jobs SET status = ? WHERE id = ?"
		args = []interface{}{status, id}
	}
	_, err := r.db.Exec(query, args...)
	return err
}

func (r *JobRepository) SetLogFile(id int64, path string) error {
	_, err := r.db.Exec("UPDATE jobs SET log_file = ? WHERE id = ?", path, id)
	return err
}

func (r *JobRepository) SetError(id int64, code, message string) error {
	_, err := r.db.Exec(
		"UPDATE jobs SET error_code = ?, error_message = ? WHERE id = ?",
		code, message, id,
	)
	return err
}

func (r *JobRepository) GetRunningBySyncPair(syncPairID int64) (*Job, error) {
	var j Job
	var started sql.NullTime
	var logFile, errMsg, errCode sql.NullString
	err := r.db.QueryRow(`
		SELECT id, sync_pair_id, trigger_type, status, started_at, finished_at,
			log_file, error_message, error_code, total_size_bytes, sent_bytes, created_at FROM jobs
		WHERE sync_pair_id = ? AND status IN ('queued','waking_up','running')
		ORDER BY created_at DESC LIMIT 1`, syncPairID).Scan(
		&j.ID, &j.SyncPairID, &j.TriggerType, &j.Status, &started,
		&j.FinishedAt, &logFile, &errMsg, &errCode, &j.TotalSizeBytes, &j.SentBytes, &j.CreatedAt)
	if err != nil {
		return nil, err
	}
	if started.Valid {
		j.StartedAt = &started.Time
	}
	if logFile.Valid {
		j.LogFile = &logFile.String
	}
	if errMsg.Valid {
		j.ErrorMessage = &errMsg.String
	}
	if errCode.Valid {
		j.ErrorCode = &errCode.String
	}
	return &j, nil
}

func (r *JobRepository) Count() (int64, error) {
	var n int64
	err := r.db.QueryRow("SELECT COUNT(*) FROM jobs").Scan(&n)
	return n, err
}

func (r *JobRepository) DeleteFinishedBefore(before time.Time) (int64, error) {
	res, err := r.db.Exec("DELETE FROM jobs WHERE finished_at IS NOT NULL AND finished_at < ?", before)
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

func (r *JobRepository) SetTotals(id int64, totalSize, sentBytes int64) error {
	_, err := r.db.Exec(
		"UPDATE jobs SET total_size_bytes = ?, sent_bytes = ? WHERE id = ?",
		totalSize, sentBytes, id,
	)
	return err
}

func (r *JobRepository) GetByStatusAny(statuses []string) ([]Job, error) {
	if len(statuses) == 0 {
		return nil, nil
	}
	placeholders := make([]string, len(statuses))
	args := make([]interface{}, len(statuses))
	for i, s := range statuses {
		placeholders[i] = "?"
		args[i] = s
	}
	query := fmt.Sprintf(`
		SELECT id, sync_pair_id, trigger_type, status, started_at, finished_at,
			log_file, error_message, error_code, total_size_bytes, sent_bytes, created_at
		FROM jobs WHERE status IN (%s)`, strings.Join(placeholders, ","))
	rows, err := r.db.Query(query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var jobs []Job
	for rows.Next() {
		var j Job
		var started, finished sql.NullTime
		var logFile, errMsg, errCode sql.NullString
		if err := rows.Scan(&j.ID, &j.SyncPairID, &j.TriggerType, &j.Status,
			&started, &finished, &logFile, &errMsg, &errCode,
			&j.TotalSizeBytes, &j.SentBytes, &j.CreatedAt); err != nil {
			return nil, err
		}
		if started.Valid {
			j.StartedAt = &started.Time
		}
		if finished.Valid {
			j.FinishedAt = &finished.Time
		}
		if logFile.Valid {
			j.LogFile = &logFile.String
		}
		if errMsg.Valid {
			j.ErrorMessage = &errMsg.String
		}
		if errCode.Valid {
			j.ErrorCode = &errCode.String
		}
		jobs = append(jobs, j)
	}
	return jobs, rows.Err()
}