homelab/services/ha-sync/internal/db/db.go

package db

import (
	"database/sql"
	"fmt"

	_ "github.com/go-sql-driver/mysql"
)

func Connect(dsn string) (*sql.DB, error) {
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return nil, fmt.Errorf("db open: %w", err)
	}
	if err := db.Ping(); err != nil {
		db.Close()
		return nil, fmt.Errorf("db ping: %w", err)
	}
	if err := Migrate(db); err != nil {
		db.Close()
		return nil, fmt.Errorf("db migrate: %w", err)
	}
	return db, nil
}

func Migrate(db *sql.DB) error {
	stmts := []string{
		`CREATE TABLE IF NOT EXISTS sync_iterations (
			id                      BIGINT AUTO_INCREMENT PRIMARY KEY,
			sync_pair               VARCHAR(255)  NOT NULL,
			direction               VARCHAR(64)   NOT NULL,
			src                     VARCHAR(512)  NOT NULL,
			dest                    VARCHAR(512)  NOT NULL,
			started_at              DATETIME(3)   NOT NULL,
			ended_at                DATETIME(3),
			status                  ENUM('running','success','partial_failure','failed') NOT NULL DEFAULT 'running',
			dry_run                 TINYINT(1)    NOT NULL DEFAULT 0,
			files_created           INT           DEFAULT 0,
			files_updated           INT           DEFAULT 0,
			files_deleted           INT           DEFAULT 0,
			files_skipped           INT           DEFAULT 0,
			files_failed            INT           DEFAULT 0,
			total_bytes_transferred BIGINT        DEFAULT 0,
			error_message           TEXT,
			INDEX idx_pair    (sync_pair),
			INDEX idx_started (started_at),
			INDEX idx_dry_run (dry_run)
		)`,
		`CREATE TABLE IF NOT EXISTS sync_operations (
			id            BIGINT AUTO_INCREMENT PRIMARY KEY,
			iteration_id  BIGINT        NOT NULL,
			dry_run       TINYINT(1)    NOT NULL DEFAULT 0,
			operation     ENUM('create','update','delete') NOT NULL,
			filepath      VARCHAR(4096) NOT NULL,
			size_before   BIGINT,
			size_after    BIGINT,
			md5_before    VARCHAR(32),
			md5_after     VARCHAR(32),
			started_at    DATETIME(3)   NOT NULL,
			ended_at      DATETIME(3),
			status        ENUM('success','fail') NOT NULL,
			error_message VARCHAR(4096),
			INDEX idx_iteration (iteration_id),
			CONSTRAINT fk_iteration FOREIGN KEY (iteration_id) REFERENCES sync_iterations(id)
		)`,
		`CREATE TABLE IF NOT EXISTS sync_jobs (
			id                BIGINT AUTO_INCREMENT PRIMARY KEY,
			name              VARCHAR(255)  NOT NULL UNIQUE,
			pair              VARCHAR(255)  NOT NULL,
			direction         VARCHAR(64)   NOT NULL,
			src               VARCHAR(512)  NOT NULL,
			dest              VARCHAR(512)  NOT NULL,
			src_host          VARCHAR(255)  NOT NULL DEFAULT 'dell',
			dest_host         VARCHAR(255)  NOT NULL DEFAULT 'hp',
			cron_schedule     VARCHAR(64)   NOT NULL DEFAULT '*/15 * * * *',
			lock_ttl_seconds  INT           NOT NULL DEFAULT 3600,
			dry_run           TINYINT(1)    NOT NULL DEFAULT 0,
			delete_missing    TINYINT(1)    NOT NULL DEFAULT 0,
			workers           INT           NOT NULL DEFAULT 4,
			mtime_threshold   VARCHAR(32)   NOT NULL DEFAULT '2s',
			excludes          TEXT,
			enabled           TINYINT(1)    NOT NULL DEFAULT 1,
			created_at        DATETIME(3)   NOT NULL,
			updated_at        DATETIME(3)   NOT NULL,
			INDEX idx_jobs_pair    (pair),
			INDEX idx_jobs_enabled (enabled)
		)`,
	}
	for _, s := range stmts {
		if _, err := db.Exec(s); err != nil {
			return fmt.Errorf("migrate: %w", err)
		}
	}

	// Idempotent: add new columns to sync_operations if they don't exist.
	for _, col := range []struct{ name, def string }{
		{"src_host", "VARCHAR(255) DEFAULT NULL"},
		{"dest_host", "VARCHAR(255) DEFAULT NULL"},
		{"owner", "VARCHAR(255) DEFAULT NULL"},
	} {
		if err := addColumnIfNotExists(db, "sync_operations", col.name, col.def); err != nil {
			return err
		}
	}

	return nil
}

// addColumnIfNotExists adds a column to a table only if it does not already exist,
// avoiding the MySQL "duplicate column" error on repeated migrations.
func addColumnIfNotExists(db *sql.DB, table, column, definition string) error {
	var count int
	err := db.QueryRow(`
		SELECT COUNT(*) FROM information_schema.COLUMNS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = ?
		  AND COLUMN_NAME = ?`, table, column).Scan(&count)
	if err != nil {
		return fmt.Errorf("check column %s.%s: %w", table, column, err)
	}
	if count > 0 {
		return nil
	}
	_, err = db.Exec(fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s", table, column, definition))
	if err != nil {
		return fmt.Errorf("add column %s.%s: %w", table, column, err)
	}
	return nil
}