internal/store/history.go

Rama main · réplica de lectura del repositorio original.

package store

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"time"

	"modernc.org/sqlite"
)

const pragmas = "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=synchronous(NORMAL)"

var errNoSchema = errors.New("the history schema could not be read")

var errFromTheFuture = errors.New("this history file was written by a newer spinoza")

type Entry struct {
	ID        int64
	Cluster   string
	At        time.Time
	Verb      string
	Actor     string
	Group     string
	Version   string
	Resource  string
	Kind      string
	Namespace string
	Name      string
	Detail    string
	Outcome   string
	Message   string
}

type Query struct {
	Cluster     string
	Limit       int
	After       int64
	AfterAction int64
}

type Page struct {
	Entries []Entry
	More    bool
}

type Store struct {
	mu     sync.Mutex
	writes *sql.DB
	reads  *sql.DB
	reason string
}

type Recorder interface {
	Record(ctx context.Context, entry Entry) error
}

type writer struct {
	into    *Store
	cluster string
}

func (s *Store) For(cluster string) Recorder {
	return writer{into: s, cluster: cluster}
}

func (w writer) Record(ctx context.Context, entry Entry) error {
	entry.Cluster = w.cluster
	return w.into.record(ctx, entry)
}

func DefaultPath() (string, error) {
	dir, err := os.UserConfigDir()
	if err != nil {
		return "", fmt.Errorf("store: %w", err)
	}
	return filepath.Join(dir, "spinoza", "history.db"), nil
}

func Open(ctx context.Context, path string) (*Store, error) {
	if path == "" {
		return unavailable("spinoza has nowhere to keep history"), nil
	}
	err := os.MkdirAll(filepath.Dir(path), 0o700)
	if err != nil {
		return unavailable(reasonFor(path, err)), fmt.Errorf("store: %w", err)
	}
	conn := connector{dsn: path + pragmas}
	writes := sql.OpenDB(conn)
	writes.SetMaxOpenConns(1)
	migrateErr := migrate(ctx, writes)
	if migrateErr != nil {
		_ = writes.Close()
		return unavailable(reasonFor(path, migrateErr)), fmt.Errorf("store: %w", migrateErr)
	}
	reads := sql.OpenDB(conn)
	reads.SetMaxOpenConns(readers)
	return &Store{writes: writes, reads: reads}, nil
}

type connector struct {
	dsn string
}

func (c connector) Connect(context.Context) (driver.Conn, error) {
	return c.Driver().Open(c.dsn)
}

func (c connector) Driver() driver.Driver {
	return &sqlite.Driver{}
}

func unavailable(reason string) *Store {
	return &Store{reason: reason}
}

func reasonFor(path string, err error) string {
	return "spinoza is not recording history: " + path + ": " + err.Error()
}

func (s *Store) Reason() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.reason
}

func (s *Store) writer() *sql.DB {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.writes
}

func (s *Store) reader() *sql.DB {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.reads
}

func (s *Store) record(ctx context.Context, entry Entry) error {
	db := s.writer()
	if db == nil {
		return nil
	}
	_, err := db.ExecContext(
		ctx, insertAudit,
		entry.Cluster, entry.At.UTC().UnixMilli(), entry.Verb, entry.Actor,
		entry.Group, entry.Version, entry.Resource, entry.Kind,
		entry.Namespace, entry.Name,
		entry.Detail, entry.Outcome, entry.Message,
	)
	if err != nil {
		return fmt.Errorf("store: %w", err)
	}
	return nil
}

func (s *Store) Recent(ctx context.Context, query Query) (Page, error) {
	db := s.reader()
	if db == nil {
		return Page{Entries: []Entry{}}, nil
	}
	limit := limitOf(query.Limit)
	rows, err := db.QueryContext(
		ctx,
		selectAudit,
		query.Cluster,
		query.Cluster,
		query.AfterAction,
		query.AfterAction,
		query.Cluster,
		query.Cluster,
		limit+1,
	)
	if err != nil {
		return Page{}, fmt.Errorf("store: %w", err)
	}
	defer func() { _ = rows.Close() }()
	found, scanErr := scanEntries(rows)
	if scanErr != nil {
		return Page{}, scanErr
	}
	return pageOf(found, limit), nil
}

func Limit(asked int) int {
	return limitOf(asked)
}

func limitOf(asked int) int {
	if asked <= 0 {
		return defaultLimit
	}
	if asked > maxLimit {
		return maxLimit
	}
	return asked
}

func pageOf(found []Entry, limit int) Page {
	if len(found) <= limit {
		return Page{Entries: found}
	}
	return Page{Entries: found[:limit], More: true}
}

func scanEntries(rows *sql.Rows) ([]Entry, error) {
	found := []Entry{}
	for rows.Next() {
		var entry Entry
		var at int64
		err := rows.Scan(
			&entry.ID, &entry.Cluster, &at, &entry.Verb, &entry.Actor,
			&entry.Group, &entry.Version, &entry.Resource, &entry.Kind,
			&entry.Namespace, &entry.Name,
			&entry.Detail, &entry.Outcome, &entry.Message,
		)
		if err != nil {
			return nil, fmt.Errorf("store: %w", err)
		}
		entry.At = time.UnixMilli(at).UTC()
		found = append(found, entry)
	}
	if rows.Err() != nil {
		return nil, fmt.Errorf("store: %w", rows.Err())
	}
	return found, nil
}

func (s *Store) Forget(ctx context.Context, cluster string) error {
	db := s.writer()
	if db == nil {
		return nil
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("store: %w", err)
	}
	_, auditErr := tx.ExecContext(ctx, deleteAudit, cluster, cluster)
	if auditErr != nil {
		_ = tx.Rollback()
		return fmt.Errorf("store: %w", auditErr)
	}
	_, changesErr := tx.ExecContext(ctx, deleteChanges, cluster, cluster)
	if changesErr != nil {
		_ = tx.Rollback()
		return fmt.Errorf("store: %w", changesErr)
	}
	commitErr := tx.Commit()
	if commitErr != nil {
		return fmt.Errorf("store: %w", commitErr)
	}
	return nil
}

func (s *Store) Close() error {
	s.mu.Lock()
	open := []*sql.DB{s.reads, s.writes}
	s.reads = nil
	s.writes = nil
	s.mu.Unlock()
	failures := []error{}
	for _, db := range open {
		if db == nil {
			continue
		}
		err := db.Close()
		if err != nil {
			failures = append(failures, fmt.Errorf("store: %w", err))
		}
	}
	return errors.Join(failures...)
}