mesh-controller/internal/inventory/batches.go

package inventory

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

	"github.com/jackc/pgx/v5"
)

// The merges a batch or walk answers (novox/hq ADR 0276): every merge the controller heard into a branch a
// module follows is kept here once, with the batch or walk that answers it. A merge is never dropped and never
// walked twice: its key is its repository and commit, and it belongs to one plan at a time.

// BatchedMerge is one merge as the controller heard it.
type BatchedMerge struct {
	Repository string
	Branch     string
	Commit     string
	// Merged is when the forge made it: the branch's order. Heard is when the controller heard it: what the
	// window is measured from.
	Merged time.Time
	Heard  time.Time
	// Event is the forge's announcement, what the cut plans from.
	Event json.RawMessage
	// Plan is the batch or walk answering it; empty while it waits to be walked alone.
	Plan string
	// Alone says it is walked on its own commit, after a failed walk that carried it in a later one.
	Alone bool
}

// AddMerge keeps a merge for a plan; false when the merge was kept already, for whatever plan.
func (i *Inventory) AddMerge(ctx context.Context, m BatchedMerge) (bool, error) {
	tag, err := i.store.Pool().Exec(ctx,
		`insert into batched_merge (repository, branch, commit_hash, merged_at, heard_at, event, plan_id, alone)
		 values (lower($1), $2, $3, $4, $5, $6, nullif($7, ''), $8) on conflict do nothing`,
		m.Repository, m.Branch, m.Commit, mergedAt(m.Merged), m.Heard, []byte(m.Event), m.Plan, m.Alone)
	if err != nil {
		return false, err
	}
	return tag.RowsAffected() == 1, nil
}

// MergeOf is one kept merge; false when it was never heard.
func (i *Inventory) MergeOf(ctx context.Context, repository, commit string) (BatchedMerge, bool, error) {
	ms, err := i.merges(ctx, `where repository = lower($1) and commit_hash = $2`, repository, commit)
	if err != nil || len(ms) == 0 {
		return BatchedMerge{}, false, err
	}
	return ms[0], true, nil
}

// MergesOf is every merge a plan answers, oldest merge first.
func (i *Inventory) MergesOf(ctx context.Context, plan string) ([]BatchedMerge, error) {
	return i.merges(ctx, `where plan_id = $1 order by merged_at nulls last, heard_at`, plan)
}

// LaterMergesOf is every merge of a repository's branch made after a moment and answered by a plan, the
// newest first.
func (i *Inventory) LaterMergesOf(ctx context.Context, repository, branch string, after time.Time) ([]BatchedMerge, error) {
	return i.merges(ctx, `where repository = lower($1) and branch = $2 and merged_at > $3 and plan_id is not null
		order by merged_at desc`, repository, branch, after)
}

// KeptOnBranch says the controller keeps a merge of a repository's branch, and when it heard the branch's first: a merge of a branch it keeps merges of, made since then, is the record's to answer, whatever a
// module's last look says (novox/hq ADR 0276).
func (i *Inventory) KeptOnBranch(ctx context.Context, repository, branch string) (bool, time.Time, error) {
	var first *time.Time
	var kept bool
	err := i.store.Pool().QueryRow(ctx,
		`select min(heard_at), count(*) > 0 from batched_merge where repository = lower($1) and branch = $2`,
		repository, branch).Scan(&first, &kept)
	if err != nil || first == nil {
		return false, time.Time{}, err
	}
	return kept, first.UTC(), nil
}

// AloneMerges is every merge waiting to be walked on its own commit, the newest first (ADR 0276 decision 3).
func (i *Inventory) AloneMerges(ctx context.Context) ([]BatchedMerge, error) {
	return i.merges(ctx, `where plan_id is null and alone order by merged_at desc nulls last, heard_at desc`)
}

// AnswerMerge moves a merge to the plan that answers it now; an empty plan with alone leaves it waiting to
// be walked on its own commit.
func (i *Inventory) AnswerMerge(ctx context.Context, repository, commit, plan string, alone bool) error {
	_, err := i.store.Pool().Exec(ctx,
		`update batched_merge set plan_id = nullif($3, ''), alone = $4 where repository = lower($1) and commit_hash = $2`,
		repository, commit, plan, alone)
	return err
}

// Batches is every batch not yet cut: assembling or queued, oldest first. One at a time is the rule; more
// are read so a fault is seen.
func (i *Inventory) Batches(ctx context.Context) ([]Plan, error) {
	return i.plans(ctx, `where state in ('assembling', 'queued') order by created`)
}

func (i *Inventory) merges(ctx context.Context, tail string, args ...any) ([]BatchedMerge, error) {
	rows, err := i.store.Pool().Query(ctx,
		`select repository, branch, commit_hash, merged_at, heard_at, event, coalesce(plan_id, ''), alone
		   from batched_merge `+tail, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []BatchedMerge
	for rows.Next() {
		var m BatchedMerge
		var merged *time.Time
		var event []byte
		if err := rows.Scan(&m.Repository, &m.Branch, &m.Commit, &merged, &m.Heard, &event, &m.Plan, &m.Alone); err != nil {
			return nil, err
		}
		if merged != nil {
			m.Merged = merged.UTC()
		}
		m.Heard = m.Heard.UTC()
		m.Event = event
		out = append(out, m)
	}
	if errors.Is(rows.Err(), pgx.ErrNoRows) {
		return nil, nil
	}
	return out, rows.Err()
}

// SameRepository says two spellings of a repository are one.
func SameRepository(a, b string) bool { return strings.EqualFold(a, b) }