jarl/internal/store/changesets.go

package store

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

	"jarl/internal/model"
)

const changesetCols = `id, description, status, user_id, username, via, ops, created_at, updated_at, applied_at, applied_by, result, diff`

func scanChangeset(sc interface{ Scan(...any) error }, withDiff bool) (*model.Changeset, error) {
	cs := &model.Changeset{}
	var ops, ca, ua, diff string
	var aa, result sql.NullString
	if err := sc.Scan(&cs.ID, &cs.Description, &cs.Status, &cs.UserID, &cs.Username, &cs.Via, &ops, &ca, &ua, &aa, &cs.AppliedBy, &result, &diff); err != nil {
		return nil, err
	}
	if err := json.Unmarshal([]byte(ops), &cs.Ops); err != nil {
		return nil, err
	}
	if cs.Ops == nil {
		cs.Ops = []model.Op{}
	}
	cs.CreatedAt, cs.UpdatedAt, cs.AppliedAt = parseTS(ca), parseTS(ua), parseTSPtr(aa)
	if result.Valid && result.String != "" {
		cs.Result = json.RawMessage(result.String)
	}
	if withDiff {
		cs.Diff = diff
	}
	return cs, nil
}

// CreateChangeset inserts a new changeset and sets its ID.
func (s *Store) CreateChangeset(ctx context.Context, cs *model.Changeset) error {
	now := time.Now()
	if cs.CreatedAt.IsZero() {
		cs.CreatedAt = now
	}
	cs.UpdatedAt = now
	if cs.Status == "" {
		cs.Status = model.ChangesetOpen
	}
	if cs.Ops == nil {
		cs.Ops = []model.Op{}
	}
	res, err := s.db.ExecContext(ctx, `INSERT INTO changesets (description, status, user_id, username, via, ops, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?)`,
		cs.Description, cs.Status, cs.UserID, cs.Username, cs.Via, jsonText(cs.Ops), ts(cs.CreatedAt), ts(cs.UpdatedAt))
	if err != nil {
		return err
	}
	cs.ID, err = res.LastInsertId()
	return err
}

// GetChangeset loads a changeset including its diff.
func (s *Store) GetChangeset(ctx context.Context, id int64) (*model.Changeset, error) {
	cs, err := scanChangeset(s.db.QueryRowContext(ctx, `SELECT `+changesetCols+` FROM changesets WHERE id = ?`, id), true)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, ErrNotFound
	}
	return cs, err
}

// ChangesetFilter selects changesets.
type ChangesetFilter struct {
	Status string
	UserID int64
	Limit  int
	Before int64 // only IDs below this (paging)
}

// ListChangesets lists changesets, newest first, without diffs.
func (s *Store) ListChangesets(ctx context.Context, f ChangesetFilter) ([]*model.Changeset, error) {
	var where []string
	var args []any
	if f.Status != "" {
		where = append(where, "status = ?")
		args = append(args, f.Status)
	}
	if f.UserID != 0 {
		where = append(where, "user_id = ?")
		args = append(args, f.UserID)
	}
	if f.Before > 0 {
		where = append(where, "id < ?")
		args = append(args, f.Before)
	}
	q := `SELECT ` + changesetCols + ` FROM changesets`
	if len(where) > 0 {
		q += " WHERE " + strings.Join(where, " AND ")
	}
	limit := f.Limit
	if limit <= 0 || limit > 1000 {
		limit = 100
	}
	q += " ORDER BY id DESC LIMIT ?"
	args = append(args, limit)
	rows, err := s.db.QueryContext(ctx, q, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []*model.Changeset
	for rows.Next() {
		cs, err := scanChangeset(rows, false)
		if err != nil {
			return nil, err
		}
		out = append(out, cs)
	}
	return out, rows.Err()
}

// UpdateChangeset saves the description, status and operations of an open
// changeset.
func (s *Store) UpdateChangeset(ctx context.Context, cs *model.Changeset) error {
	cs.UpdatedAt = time.Now()
	res, err := s.db.ExecContext(ctx, `UPDATE changesets SET description = ?, status = ?, ops = ?, updated_at = ? WHERE id = ? AND status = 'open'`,
		cs.Description, cs.Status, jsonText(cs.Ops), ts(cs.UpdatedAt), cs.ID)
	if err != nil {
		return err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return ErrNotFound
	}
	return nil
}

// MarkApplied records that a changeset was applied.
func (t *Tx) MarkApplied(cs *model.Changeset) error {
	now := time.Now()
	cs.Status = model.ChangesetApplied
	cs.AppliedAt = &now
	cs.UpdatedAt = now
	_, err := t.tx.ExecContext(t.ctx, `UPDATE changesets SET status = ?, ops = ?, applied_at = ?, applied_by = ?, result = ?, diff = ?, updated_at = ?, description = ? WHERE id = ?`,
		cs.Status, jsonText(cs.Ops), ts(now), cs.AppliedBy, nullJSON(cs.Result), cs.Diff, ts(now), cs.Description, cs.ID)
	return err
}

// InsertAppliedChangeset stores a changeset that was created and applied in
// one step.
func (t *Tx) InsertAppliedChangeset(cs *model.Changeset) error {
	now := time.Now()
	if cs.CreatedAt.IsZero() {
		cs.CreatedAt = now
	}
	cs.Status = model.ChangesetApplied
	cs.AppliedAt = &now
	cs.UpdatedAt = now
	res, err := t.tx.ExecContext(t.ctx, `INSERT INTO changesets (description, status, user_id, username, via, ops, created_at, updated_at, applied_at, applied_by, result, diff) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)`,
		cs.Description, cs.Status, cs.UserID, cs.Username, cs.Via, jsonText(cs.Ops), ts(cs.CreatedAt), ts(now), ts(now), cs.AppliedBy, nullJSON(cs.Result), cs.Diff)
	if err != nil {
		return err
	}
	cs.ID, err = res.LastInsertId()
	return err
}

func nullJSON(r json.RawMessage) any {
	if len(r) == 0 {
		return nil
	}
	return string(r)
}

// DeleteChangeset removes a changeset (used for discarded one-shot
// changesets that never became visible).
func (s *Store) DeleteChangeset(ctx context.Context, id int64) error {
	_, err := s.db.ExecContext(ctx, `DELETE FROM changesets WHERE id = ?`, id)
	return err
}