emissary/internal/datastore/sqlite/spec_definition_repository.go

package sqlite

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

	"forge.cadoles.com/Cadoles/emissary/internal/datastore"
	"github.com/pkg/errors"
	"gitlab.com/wpetit/goweb/logger"
)

type SpecDefinitionRepository struct {
	repository
}

// Delete implements datastore.SpecDefinitionRepository.
func (r *SpecDefinitionRepository) Delete(ctx context.Context, name string, version string) error {
	err := r.withTxRetry(ctx, func(tx *sql.Tx) error {
		if exists, err := r.specDefinitionExists(ctx, tx, name, version); !exists {
			return errors.WithStack(err)
		}

		query := `DELETE FROM spec_definitions WHERE name = $1 AND version = $2`
		_, err := tx.ExecContext(ctx, query, name, version)
		if err != nil {
			return errors.WithStack(err)
		}

		return nil
	})
	if err != nil {
		return errors.WithStack(err)
	}

	return nil
}

// Get implements datastore.SpecDefinitionRepository.
func (r *SpecDefinitionRepository) Get(ctx context.Context, name string, version string) (*datastore.SpecDefinition, error) {
	var specDef datastore.SpecDefinition

	err := r.withTxRetry(ctx, func(tx *sql.Tx) error {
		query := `
		SELECT "name", "version", "schema", "created_at", "updated_at"
		FROM spec_definitions
		WHERE name = $1 AND version = $2
		`

		row := tx.QueryRowContext(ctx, query, name, version)

		if err := row.Scan(&specDef.Name, &specDef.Version, &specDef.Schema, &specDef.CreatedAt, &specDef.UpdatedAt); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return errors.WithStack(datastore.ErrNotFound)
			}

			return errors.WithStack(err)
		}

		return nil
	})
	if err != nil {
		return nil, errors.WithStack(err)
	}

	return &specDef, nil
}

// Query implements datastore.SpecDefinitionRepository.
func (r *SpecDefinitionRepository) Query(ctx context.Context, opts ...datastore.SpecDefinitionQueryOptionFunc) ([]datastore.SpecDefinitionHeader, int, error) {
	options := &datastore.SpecDefinitionQueryOptions{}
	for _, fn := range opts {
		fn(options)
	}

	specDefs := make([]datastore.SpecDefinitionHeader, 0)
	count := 0

	err := r.withTxRetry(ctx, func(tx *sql.Tx) error {
		query := `SELECT name, version, created_at, updated_at FROM spec_definitions`

		limit := 10
		if options.Limit != nil {
			limit = *options.Limit
		}

		offset := 0
		if options.Offset != nil {
			offset = *options.Offset
		}

		filters := ""
		paramIndex := 3
		args := []any{offset, limit}

		if options.Names != nil && len(options.Names) > 0 {
			filter, newArgs, newParamIndex := inFilter("name", paramIndex, options.Names)
			filters += filter
			paramIndex = newParamIndex
			args = append(args, newArgs...)
		}

		if options.Versions != nil && len(options.Versions) > 0 {
			if filters != "" {
				filters += " AND "
			}

			filter, newArgs, _ := inFilter("version", paramIndex, options.Versions)
			filters += filter
			args = append(args, newArgs...)
		}

		if filters != "" {
			filters = ` WHERE ` + filters
		}

		query += filters + ` LIMIT $2 OFFSET $1`

		logger.Debug(ctx, "executing query", logger.F("query", query), logger.F("args", args))

		rows, err := tx.QueryContext(ctx, query, args...)
		if err != nil {
			return errors.WithStack(err)
		}

		defer func() {
			if err := rows.Close(); err != nil {
				err = errors.WithStack(err)
				logger.Error(ctx, "could not close rows", logger.CapturedE(err))
			}
		}()

		for rows.Next() {
			sdh := datastore.SpecDefinitionHeader{}

			if err := rows.Scan(&sdh.Name, &sdh.Version, &sdh.CreatedAt, &sdh.UpdatedAt); err != nil {
				return errors.WithStack(err)
			}

			specDefs = append(specDefs, sdh)
		}

		if err := rows.Err(); err != nil {
			return errors.WithStack(err)
		}

		row := tx.QueryRowContext(ctx, `SELECT count(*) FROM spec_definitions `+filters, args...)
		if err := row.Scan(&count); err != nil {
			return errors.WithStack(err)
		}

		return nil
	})
	if err != nil {
		return nil, 0, errors.WithStack(err)
	}

	return specDefs, count, nil
}

// Upsert implements datastore.SpecDefinitionRepository.
func (r *SpecDefinitionRepository) Upsert(ctx context.Context, name string, version string, schema []byte) (*datastore.SpecDefinition, error) {
	var specDef datastore.SpecDefinition

	err := r.withTxRetry(ctx, func(tx *sql.Tx) error {
		now := time.Now().UTC()

		query := `
		INSERT INTO spec_definitions (name, version, schema, created_at, updated_at) 
		VALUES($1, $2, $3, $4, $4)
		ON CONFLICT(name, version) DO UPDATE SET schema = $3, updated_at = $4
		RETURNING "name", "version", "schema", "created_at", "updated_at"
		`

		row := tx.QueryRowContext(
			ctx, query,
			name, version, schema, now, now,
		)

		if err := row.Scan(&specDef.Name, &specDef.Version, &specDef.Schema, &specDef.CreatedAt, &specDef.UpdatedAt); err != nil {
			return errors.WithStack(err)
		}

		return nil
	})
	if err != nil {
		return nil, errors.WithStack(err)
	}

	return &specDef, nil
}

func (r *SpecDefinitionRepository) specDefinitionExists(ctx context.Context, tx *sql.Tx, name string, version string) (bool, error) {
	row := tx.QueryRowContext(ctx, `SELECT count(id) FROM spec_definitions WHERE name = $1 AND version = $2`, name, version)

	var count int

	if err := row.Scan(&count); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return false, errors.WithStack(datastore.ErrNotFound)
		}

		return false, errors.WithStack(err)
	}

	if count == 0 {
		return false, errors.WithStack(datastore.ErrNotFound)
	}

	return true, nil
}

func NewSpecDefinitionRepository(db *sql.DB, sqliteBusyRetryMaxAttempts int) *SpecDefinitionRepository {
	return &SpecDefinitionRepository{
		repository: repository{db, sqliteBusyRetryMaxAttempts},
	}
}

var _ datastore.SpecDefinitionRepository = &SpecDefinitionRepository{}