ocdp-go/backend/internal/adapter/output/persistence/postgres/cluster_repository.go

package postgres

import (
	"context"
	"database/sql"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/ocdp/cluster-service/internal/domain/entity"
	"github.com/ocdp/cluster-service/internal/domain/repository"
	"github.com/ocdp/cluster-service/internal/pkg/crypto"
)

type ClusterRepository struct {
	db        *DB
	encryptor crypto.Encryptor
}

func NewClusterRepository(db *DB, encryptor crypto.Encryptor) repository.ClusterRepository {
	return &ClusterRepository{db: db, encryptor: encryptor}
}

func (r *ClusterRepository) Create(ctx context.Context, cluster *entity.Cluster) error {
	if cluster.ID == "" {
		cluster.ID = uuid.New().String()
	}
	encryptedCAData, encryptedCertData, encryptedKeyData, encryptedToken, err := r.encryptClusterSecrets(cluster)
	if err != nil {
		return err
	}
	query := `
		INSERT INTO clusters
			(id, workspace_id, owner_id, visibility, name, host, ca_data, cert_data, key_data, token, description, default_namespace, created_at, updated_at)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)
	`
	_, err = r.db.conn.ExecContext(ctx, query,
		cluster.ID,
		cluster.WorkspaceID,
		cluster.OwnerID,
		cluster.Visibility,
		cluster.Name,
		cluster.Host,
		encryptedCAData,
		encryptedCertData,
		encryptedKeyData,
		encryptedToken,
		cluster.Description,
		cluster.DefaultNamespace,
		cluster.CreatedAt,
		cluster.UpdatedAt,
	)
	if err != nil {
		return fmt.Errorf("failed to create cluster: %w", err)
	}
	return nil
}

func (r *ClusterRepository) GetByID(ctx context.Context, id string) (*entity.Cluster, error) {
	return r.get(ctx, "id = $1", id)
}

func (r *ClusterRepository) GetByName(ctx context.Context, name string) (*entity.Cluster, error) {
	return r.get(ctx, "name = $1", name)
}

func (r *ClusterRepository) get(ctx context.Context, where string, arg interface{}) (*entity.Cluster, error) {
	query := fmt.Sprintf(`
		SELECT id, workspace_id, owner_id, visibility, name, host, ca_data, cert_data, key_data, token, description, default_namespace, created_at, updated_at
		FROM clusters
		WHERE %s
	`, where)
	rows, err := r.db.conn.QueryContext(ctx, query, arg)
	if err != nil {
		return nil, fmt.Errorf("failed to get cluster: %w", err)
	}
	defer rows.Close()
	if !rows.Next() {
		return nil, entity.ErrClusterNotFound
	}
	cluster, err := r.scanCluster(rows)
	if err != nil {
		return nil, err
	}
	return cluster, nil
}

func (r *ClusterRepository) Update(ctx context.Context, cluster *entity.Cluster) error {
	cluster.UpdatedAt = time.Now()
	encryptedCAData, encryptedCertData, encryptedKeyData, encryptedToken, err := r.encryptClusterSecrets(cluster)
	if err != nil {
		return err
	}
	query := `
		UPDATE clusters
		SET workspace_id = $1, owner_id = $2, visibility = $3, name = $4, host = $5,
		    ca_data = $6, cert_data = $7, key_data = $8, token = $9, description = $10,
		    default_namespace = $11, updated_at = $12
		WHERE id = $13
	`
	result, err := r.db.conn.ExecContext(ctx, query,
		cluster.WorkspaceID,
		cluster.OwnerID,
		cluster.Visibility,
		cluster.Name,
		cluster.Host,
		encryptedCAData,
		encryptedCertData,
		encryptedKeyData,
		encryptedToken,
		cluster.Description,
		cluster.DefaultNamespace,
		cluster.UpdatedAt,
		cluster.ID,
	)
	if err != nil {
		return fmt.Errorf("failed to update cluster: %w", err)
	}
	rows, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("failed to get affected rows: %w", err)
	}
	if rows == 0 {
		return entity.ErrClusterNotFound
	}
	return nil
}

func (r *ClusterRepository) Delete(ctx context.Context, id string) error {
	result, err := r.db.conn.ExecContext(ctx, `DELETE FROM clusters WHERE id = $1`, id)
	if err != nil {
		return fmt.Errorf("failed to delete cluster: %w", err)
	}
	rows, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("failed to get affected rows: %w", err)
	}
	if rows == 0 {
		return entity.ErrClusterNotFound
	}
	return nil
}

func (r *ClusterRepository) List(ctx context.Context) ([]*entity.Cluster, error) {
	query := `
		SELECT id, workspace_id, owner_id, visibility, name, host, ca_data, cert_data, key_data, token, description, default_namespace, created_at, updated_at
		FROM clusters
		ORDER BY created_at DESC
	`
	rows, err := r.db.conn.QueryContext(ctx, query)
	if err != nil {
		return nil, fmt.Errorf("failed to list clusters: %w", err)
	}
	defer rows.Close()
	clusters := make([]*entity.Cluster, 0)
	for rows.Next() {
		cluster, err := r.scanCluster(rows)
		if err != nil {
			return nil, err
		}
		clusters = append(clusters, cluster)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("rows iteration error: %w", err)
	}
	return clusters, nil
}

type clusterScanner interface {
	Scan(dest ...interface{}) error
}

func (r *ClusterRepository) scanCluster(scanner clusterScanner) (*entity.Cluster, error) {
	cluster := &entity.Cluster{}
	var encryptedCAData, encryptedCertData, encryptedKeyData, encryptedToken sql.NullString
	var defaultNamespace sql.NullString
	err := scanner.Scan(
		&cluster.ID,
		&cluster.WorkspaceID,
		&cluster.OwnerID,
		&cluster.Visibility,
		&cluster.Name,
		&cluster.Host,
		&encryptedCAData,
		&encryptedCertData,
		&encryptedKeyData,
		&encryptedToken,
		&cluster.Description,
		&defaultNamespace,
		&cluster.CreatedAt,
		&cluster.UpdatedAt,
	)
	if err != nil {
		return nil, fmt.Errorf("failed to scan cluster: %w", err)
	}
	cluster.DefaultNamespace = defaultNamespace.String
	var decryptErr error
	cluster.CAData, decryptErr = decryptMaybe(r.encryptor, encryptedCAData.String)
	if decryptErr != nil {
		return nil, fmt.Errorf("failed to decrypt CA data: %w", decryptErr)
	}
	cluster.CertData, decryptErr = decryptMaybe(r.encryptor, encryptedCertData.String)
	if decryptErr != nil {
		return nil, fmt.Errorf("failed to decrypt cert data: %w", decryptErr)
	}
	cluster.KeyData, decryptErr = decryptMaybe(r.encryptor, encryptedKeyData.String)
	if decryptErr != nil {
		return nil, fmt.Errorf("failed to decrypt key data: %w", decryptErr)
	}
	cluster.Token, decryptErr = decryptMaybe(r.encryptor, encryptedToken.String)
	if decryptErr != nil {
		return nil, fmt.Errorf("failed to decrypt token: %w", decryptErr)
	}
	return cluster, nil
}

func (r *ClusterRepository) encryptClusterSecrets(cluster *entity.Cluster) (string, string, string, string, error) {
	ca, err := r.encryptor.Encrypt(cluster.CAData)
	if err != nil {
		return "", "", "", "", fmt.Errorf("failed to encrypt CA data: %w", err)
	}
	cert, err := r.encryptor.Encrypt(cluster.CertData)
	if err != nil {
		return "", "", "", "", fmt.Errorf("failed to encrypt cert data: %w", err)
	}
	key, err := r.encryptor.Encrypt(cluster.KeyData)
	if err != nil {
		return "", "", "", "", fmt.Errorf("failed to encrypt key data: %w", err)
	}
	token, err := r.encryptor.Encrypt(cluster.Token)
	if err != nil {
		return "", "", "", "", fmt.Errorf("failed to encrypt token: %w", err)
	}
	return ca, cert, key, token, nil
}

func decryptMaybe(encryptor crypto.Encryptor, value string) (string, error) {
	if value == "" {
		return "", nil
	}
	return encryptor.Decrypt(value)
}