YMhut-box-C-/server/unified-management/internal/db/database_sync_jobs.go

package db

import (
	"database/sql"
	"encoding/json"
	"errors"
	"strings"
)

type syncJobMessage struct {
	Output   []string `json:"output"`
	Warnings []string `json:"warnings"`
	Errors   []string `json:"errors"`
}

type syncJobScanner interface {
	Scan(dest ...any) error
}

func (s *Store) insertDatabaseSyncJob(direction, status string, output, warnings, jobErrors []string, tables map[string]int, finishedAt string) (DatabaseSyncJob, error) {
	startedAt := Now()
	message, tablesJSON := encodeDatabaseSyncJob(output, warnings, jobErrors, tables)
	id, err := s.insertID(`INSERT INTO database_sync_jobs (direction, status, message, tables_json, started_at, finished_at) VALUES (?, ?, ?, ?, ?, ?)`,
		direction, status, message, tablesJSON, startedAt, finishedAt)
	if err != nil {
		return DatabaseSyncJob{}, err
	}
	return s.GetDatabaseSyncJob(id)
}

func (s *Store) updateDatabaseSyncJob(id int64, status string, output, warnings, jobErrors []string, tables map[string]int, finishedAt string) error {
	message, tablesJSON := encodeDatabaseSyncJob(output, warnings, jobErrors, tables)
	if _, err := s.exec(`UPDATE database_sync_jobs SET status = ?, message = ?, tables_json = ?, finished_at = ? WHERE id = ?`,
		status, message, tablesJSON, finishedAt, id); err != nil {
		return err
	}
	if job, err := s.GetDatabaseSyncJob(id); err == nil {
		s.mu.Lock()
		if s.status.CurrentSyncJob != nil && s.status.CurrentSyncJob.ID == id {
			s.status.CurrentSyncJob = &job
		}
		s.mu.Unlock()
	}
	return nil
}

func (s *Store) GetDatabaseSyncJob(id int64) (DatabaseSyncJob, error) {
	job, err := scanDatabaseSyncJob(s.queryRow(`SELECT id, direction, status, message, tables_json, started_at, finished_at FROM database_sync_jobs WHERE id = ?`, id))
	if errors.Is(err, sql.ErrNoRows) {
		return DatabaseSyncJob{}, errors.New("database sync job not found")
	}
	return job, err
}

func (s *Store) LatestDatabaseSyncJob() (DatabaseSyncJob, error) {
	job, err := scanDatabaseSyncJob(s.queryRow(`SELECT id, direction, status, message, tables_json, started_at, finished_at FROM database_sync_jobs ORDER BY id DESC LIMIT 1`))
	if errors.Is(err, sql.ErrNoRows) {
		return DatabaseSyncJob{}, errors.New("database sync job not found")
	}
	return job, err
}

func (s *Store) restoreDatabaseSyncStatus() {
	job, err := s.LatestDatabaseSyncJob()
	if err != nil {
		return
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if job.Status == "running" {
		s.status.CurrentSyncJob = &job
	} else {
		s.status.CurrentSyncJob = nil
	}
	if job.FinishedAt != "" {
		s.status.LastSyncAt = job.FinishedAt
	} else {
		s.status.LastSyncAt = job.StartedAt
	}
	if job.Status == "failed" {
		s.status.LastSyncError = strings.Join(job.Errors, "; ")
	} else {
		s.status.LastSyncError = ""
	}
}

func scanDatabaseSyncJob(scanner syncJobScanner) (DatabaseSyncJob, error) {
	var job DatabaseSyncJob
	var message, tablesJSON string
	if err := scanner.Scan(&job.ID, &job.Direction, &job.Status, &message, &tablesJSON, &job.StartedAt, &job.FinishedAt); err != nil {
		return DatabaseSyncJob{}, err
	}
	decodeDatabaseSyncJob(message, tablesJSON, &job)
	return job, nil
}

func encodeDatabaseSyncJob(output, warnings, jobErrors []string, tables map[string]int) (string, string) {
	if output == nil {
		output = []string{}
	}
	if warnings == nil {
		warnings = []string{}
	}
	if jobErrors == nil {
		jobErrors = []string{}
	}
	if tables == nil {
		tables = map[string]int{}
	}
	message, _ := json.Marshal(syncJobMessage{Output: output, Warnings: warnings, Errors: jobErrors})
	tableData, _ := json.Marshal(tables)
	return string(message), string(tableData)
}

func decodeDatabaseSyncJob(message, tablesJSON string, job *DatabaseSyncJob) {
	job.Output = []string{}
	job.Warnings = []string{}
	job.Errors = []string{}
	job.Tables = map[string]int{}
	var payload syncJobMessage
	if strings.TrimSpace(message) != "" && json.Unmarshal([]byte(message), &payload) == nil {
		job.Output = payload.Output
		job.Warnings = payload.Warnings
		job.Errors = payload.Errors
	} else if strings.TrimSpace(message) != "" {
		job.Output = []string{message}
	}
	_ = json.Unmarshal([]byte(tablesJSON), &job.Tables)
	if job.Output == nil {
		job.Output = []string{}
	}
	if job.Warnings == nil {
		job.Warnings = []string{}
	}
	if job.Errors == nil {
		job.Errors = []string{}
	}
	if job.Tables == nil {
		job.Tables = map[string]int{}
	}
}