spark-mcp/internal/storage/cluster_repo.go

package storage

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"spark-mcp-go/internal/cluster"
)

// ErrNotFound is returned when a requested cluster does not exist.
var ErrNotFound = errors.New("storage: cluster not found")

// ClusterRepo provides CRUD operations for clusters.
type ClusterRepo struct {
	db *DB
}

// Clusters returns a repository bound to this DB handle.
func (d *DB) Clusters() *ClusterRepo {
	return &ClusterRepo{db: d}
}

// Create inserts a new cluster record.
//
// Timestamps are stored as Unix nanoseconds (INTEGER) — sub-second precision
// keeps created_at unique across rapid inserts and makes ORDER BY DESC
// stable without artificial delays.
//
// If a cluster with the same id already exists, the underlying SQLite
// UNIQUE constraint error is wrapped and returned.
func (r *ClusterRepo) Create(ctx context.Context, c *cluster.Cluster) error {
	now := time.Now().UnixNano()
	c.CreatedAt = time.Unix(0, now)
	c.UpdatedAt = c.CreatedAt

	allowlistJSON, err := json.Marshal(c.URLAllowlist)
	if err != nil {
		return fmt.Errorf("storage: create cluster %s: marshal url_allowlist: %w", c.ID, err)
	}
	defaultArgsJSON, err := json.Marshal(c.DefaultSubmitArgs)
	if err != nil {
		return fmt.Errorf("storage: create cluster %s: marshal default_submit_args: %w", c.ID, err)
	}

	// TODO(phase-5): encrypt auth_password_enc instead of storing plaintext.
	pwdEnc := []byte(c.AuthPassword)

	_, err = r.db.sqlDB.ExecContext(ctx, `
		INSERT INTO clusters (
			id, name, rm_url, shs_url, spark_submit_execute_bin,
			is_active, auth_type, auth_username, auth_password_enc,
			ssl_verify, ssl_ca_bundle, url_allowlist, default_submit_args,
			rate_limit_per_min, created_at, updated_at
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		c.ID, c.Name, c.RMURL, c.SHSURL, c.SparkSubmitExecuteBin,
		boolToInt(c.IsActive), string(c.AuthType), c.AuthUsername, pwdEnc,
		boolToInt(c.SSLVerify), c.SSLCABundle, string(allowlistJSON), string(defaultArgsJSON),
		c.RateLimitPerMin, now, now,
	)
	if err != nil {
		if strings.Contains(err.Error(), "UNIQUE constraint failed") {
			return fmt.Errorf("storage: create cluster %s: primary key conflict: %w", c.ID, err)
		}
		return fmt.Errorf("storage: create cluster %s: %w", c.ID, err)
	}
	return nil
}

// Get retrieves a cluster by id. If the cluster does not exist, it returns
// ErrNotFound.
func (r *ClusterRepo) Get(ctx context.Context, id string) (*cluster.Cluster, error) {
	row := r.db.sqlDB.QueryRowContext(ctx, `
		SELECT id, name, rm_url, shs_url, spark_submit_execute_bin,
		       is_active, auth_type, auth_username, auth_password_enc,
		       ssl_verify, ssl_ca_bundle, url_allowlist, default_submit_args,
		       rate_limit_per_min, created_at, updated_at
		FROM clusters
		WHERE id = ?`, id)

	c, err := scanCluster(row.Scan)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrNotFound
	}
	if err != nil {
		return nil, fmt.Errorf("storage: get cluster %s: %w", id, err)
	}
	return c, nil
}

// List returns every cluster, including inactive ones, ordered by created_at
// descending.
func (r *ClusterRepo) List(ctx context.Context) ([]*cluster.Cluster, error) {
	rows, err := r.db.sqlDB.QueryContext(ctx, `
		SELECT id, name, rm_url, shs_url, spark_submit_execute_bin,
		       is_active, auth_type, auth_username, auth_password_enc,
		       ssl_verify, ssl_ca_bundle, url_allowlist, default_submit_args,
		       rate_limit_per_min, created_at, updated_at
		FROM clusters
		ORDER BY created_at DESC`)
	if err != nil {
		return nil, fmt.Errorf("storage: list clusters: %w", err)
	}
	defer rows.Close()

	var out []*cluster.Cluster
	for rows.Next() {
		c, err := scanCluster(rows.Scan)
		if err != nil {
			return nil, fmt.Errorf("storage: list clusters: %w", err)
		}
		out = append(out, c)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("storage: list clusters: %w", err)
	}
	return out, nil
}

// Update modifies an existing cluster. An empty AuthPassword means "do not
// touch the stored password" — the existing auth_password_enc blob is kept
// intact.
func (r *ClusterRepo) Update(ctx context.Context, c *cluster.Cluster) error {
	c.UpdatedAt = time.Unix(0, time.Now().UnixNano())

	allowlistJSON, err := json.Marshal(c.URLAllowlist)
	if err != nil {
		return fmt.Errorf("storage: update cluster %s: marshal url_allowlist: %w", c.ID, err)
	}
	defaultArgsJSON, err := json.Marshal(c.DefaultSubmitArgs)
	if err != nil {
		return fmt.Errorf("storage: update cluster %s: marshal default_submit_args: %w", c.ID, err)
	}

	var res sql.Result
	if c.AuthPassword == "" {
		// Preserve the existing password blob.
		res, err = r.db.sqlDB.ExecContext(ctx, `
			UPDATE clusters
			SET name = ?, rm_url = ?, shs_url = ?, spark_submit_execute_bin = ?,
			    is_active = ?, auth_type = ?, auth_username = ?,
			    ssl_verify = ?, ssl_ca_bundle = ?, url_allowlist = ?,
			    default_submit_args = ?, rate_limit_per_min = ?, updated_at = ?
			WHERE id = ?`,
			c.Name, c.RMURL, c.SHSURL, c.SparkSubmitExecuteBin,
			boolToInt(c.IsActive), string(c.AuthType), c.AuthUsername,
			boolToInt(c.SSLVerify), c.SSLCABundle, string(allowlistJSON),
			string(defaultArgsJSON), c.RateLimitPerMin, c.UpdatedAt.UnixNano(), c.ID,
		)
	} else {
		// TODO(phase-5): encrypt auth_password_enc instead of storing plaintext.
		pwdEnc := []byte(c.AuthPassword)
		res, err = r.db.sqlDB.ExecContext(ctx, `
			UPDATE clusters
			SET name = ?, rm_url = ?, shs_url = ?, spark_submit_execute_bin = ?,
			    is_active = ?, auth_type = ?, auth_username = ?, auth_password_enc = ?,
			    ssl_verify = ?, ssl_ca_bundle = ?, url_allowlist = ?,
			    default_submit_args = ?, rate_limit_per_min = ?, updated_at = ?
			WHERE id = ?`,
			c.Name, c.RMURL, c.SHSURL, c.SparkSubmitExecuteBin,
			boolToInt(c.IsActive), string(c.AuthType), c.AuthUsername, pwdEnc,
			boolToInt(c.SSLVerify), c.SSLCABundle, string(allowlistJSON),
			string(defaultArgsJSON), c.RateLimitPerMin, c.UpdatedAt.UnixNano(), c.ID,
		)
	}
	if err != nil {
		return fmt.Errorf("storage: update cluster %s: %w", c.ID, err)
	}

	n, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("storage: update cluster %s: rows affected: %w", c.ID, err)
	}
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

// Delete removes a cluster by id. If the cluster does not exist, it returns
// ErrNotFound.
func (r *ClusterRepo) Delete(ctx context.Context, id string) error {
	res, err := r.db.sqlDB.ExecContext(ctx, `DELETE FROM clusters WHERE id = ?`, id)
	if err != nil {
		return fmt.Errorf("storage: delete cluster %s: %w", id, err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("storage: delete cluster %s: rows affected: %w", id, err)
	}
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

// scanFunc abstracts *sql.Row.Scan and *sql.Rows.Scan so a single mapping
// function can serve both QueryRow and Query paths.
type scanFunc func(dest ...any) error

// scanCluster maps a single SQL row into a cluster.Cluster value.
func scanCluster(scan scanFunc) (*cluster.Cluster, error) {
	var c cluster.Cluster
	var allowlistJSON, defaultArgsJSON sql.NullString
	var pwdEnc []byte
	var createdAt, updatedAt int64
	var isActive, sslVerify int

	if err := scan(
		&c.ID, &c.Name, &c.RMURL, &c.SHSURL, &c.SparkSubmitExecuteBin,
		&isActive, &c.AuthType, &c.AuthUsername, &pwdEnc,
		&sslVerify, &c.SSLCABundle, &allowlistJSON, &defaultArgsJSON,
		&c.RateLimitPerMin, &createdAt, &updatedAt,
	); err != nil {
		return nil, err
	}

	c.IsActive = intToBool(isActive)
	c.SSLVerify = intToBool(sslVerify)
	c.CreatedAt = time.Unix(0, createdAt)
	c.UpdatedAt = time.Unix(0, updatedAt)
	// TODO(phase-5): decrypt pwdEnc once encryption is introduced.
	c.AuthPassword = string(pwdEnc)

	if allowlistJSON.Valid && allowlistJSON.String != "" {
		if err := json.Unmarshal([]byte(allowlistJSON.String), &c.URLAllowlist); err != nil {
			return nil, fmt.Errorf("storage: scan cluster %s: unmarshal url_allowlist: %w", c.ID, err)
		}
	}
	if defaultArgsJSON.Valid && defaultArgsJSON.String != "" {
		if err := json.Unmarshal([]byte(defaultArgsJSON.String), &c.DefaultSubmitArgs); err != nil {
			return nil, fmt.Errorf("storage: scan cluster %s: unmarshal default_submit_args: %w", c.ID, err)
		}
	}

	return &c, nil
}

func boolToInt(b bool) int {
	if b {
		return 1
	}
	return 0
}

func intToBool(i int) bool {
	return i != 0
}