byre/internal/store/workflow.go

package store

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

	"g.fc.run/misc/byre/internal/api"
)

type FrozenWorkflow struct {
	Name       string
	Definition api.WorkflowDefinition
}

func (s *Store) ListWorkflows(ctx context.Context) ([]api.WorkflowSummary, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT name, coalesce(json_extract(definition, '$.description'), '') FROM workflows ORDER BY name`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	workflows := []api.WorkflowSummary{}
	for rows.Next() {
		w := api.WorkflowSummary{}
		err = rows.Scan(&w.Name, &w.Description)
		if err != nil {
			return nil, err
		}
		workflows = append(workflows, w)
	}
	return workflows, rows.Err()
}

func (s *Store) GetWorkflow(ctx context.Context, name string) (api.WorkflowDefinition, error) {
	def := ""
	err := s.db.QueryRowContext(ctx, `SELECT definition FROM workflows WHERE name = ?`, name).Scan(&def)
	if errors.Is(err, sql.ErrNoRows) {
		return api.WorkflowDefinition{}, fmt.Errorf("%w: no workflow %q", ErrNotFound, name)
	}
	if err != nil {
		return api.WorkflowDefinition{}, err
	}
	w := api.WorkflowDefinition{}
	err = json.Unmarshal([]byte(def), &w)
	return w, err
}

func (s *Store) PutWorkflow(ctx context.Context, name string, def api.WorkflowDefinition, actor string) error {
	b, err := json.Marshal(def)
	if err != nil {
		return err
	}
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		_, err := tx.ExecContext(ctx, `INSERT INTO workflows (name, definition) VALUES (?, ?)
			ON CONFLICT (name) DO UPDATE SET definition = excluded.definition`, name, string(b))
		if err != nil {
			return err
		}
		return emit(api.Event{Kind: "workflow.updated", Actor: actor, Payload: map[string]any{"name": name}})
	})
}

func (s *Store) DeleteWorkflow(ctx context.Context, name, actor string) error {
	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		res, err := tx.ExecContext(ctx, `DELETE FROM workflows WHERE name = ?`, name)
		if err != nil {
			return err
		}
		n, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if n == 0 {
			return fmt.Errorf("%w: no workflow %q", ErrNotFound, name)
		}
		return emit(api.Event{Kind: "workflow.deleted", Actor: actor, Payload: map[string]any{"name": name}})
	})
}

func (s *Store) FreezeWorkflow(ctx context.Context, name string) (FrozenWorkflow, error) {
	def, err := s.GetWorkflow(ctx, name)
	if err != nil {
		return FrozenWorkflow{}, err
	}
	return FrozenWorkflow{Name: name, Definition: def}, nil
}

func (s *Store) TaskWorkflow(ctx context.Context, taskID string) (FrozenWorkflow, error) {
	name := ""
	def := ""
	err := s.db.QueryRowContext(ctx, `SELECT workflow_name, workflow_definition FROM tasks WHERE id = ?`, taskID).Scan(&name, &def)
	if errors.Is(err, sql.ErrNoRows) {
		return FrozenWorkflow{}, fmt.Errorf("%w: no task %s", ErrNotFound, taskID)
	}
	if err != nil {
		return FrozenWorkflow{}, err
	}
	if name == "" {
		return FrozenWorkflow{}, fmt.Errorf("%w: task %s has no workflow", ErrNotFound, taskID)
	}
	f := FrozenWorkflow{Name: name}
	err = json.Unmarshal([]byte(def), &f.Definition)
	return f, err
}