mesh-delivery/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go

// Package bridge turns the forge module's pull request events into the messages the forge bridge writes on FORGE.
package bridge

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"path"
	"regexp"
	"slices"
	"strconv"
	"strings"
	"sync"

	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// ErrUnreadable is wrapped by every error that says an event cannot be turned into a message; it is never
// redelivered.
var ErrUnreadable = errors.New("unreadable event")

// ErrNotHandled says the event is one the bridge hears but writes nothing for (yet).
var ErrNotHandled = errors.New("event not handled")

// ErrRetry says the event is readable but the forge could not be asked what the message needs; the event is to be
// heard again later.
var ErrRetry = errors.New("the forge could not say")

// ErrNoModules says a merge into the trunk left a repository with no module.json anywhere: no merged message.
var ErrNoModules = errors.New("no module.json in the merge")

// Trunk is the branch a merged message is written for.
const Trunk = "main"

// Trees asks the forge for the trees of a commit: the root tree and the tree of every directory holding a module.json
// (the root directory spelled ""), as the forge's gitea_commit_trees answers (issue 1234/AC4).
type Trees func(owner, repo, sha string) (root string, trees map[string]string, err error)

// ErrNotFound says the forge has no file at that path and ref (the forge's "404", issue 1448).
var ErrNotFound = errors.New("the forge has no such file")

// Files asks the forge for one file's bytes at a ref (a commit), as the forge's gitea_get_file answers; a file the forge
// says is not there is ErrNotFound, any other failure any other error (issue 1448).
type Files func(owner, repo, path, ref string) ([]byte, error)

// LockFiles are the lock files read beside a module's manifest, by name relative to its directory (novox/hq ADR 0465
// clause 1: go.sum for a Go module, package-lock.json for a Node one).
var LockFiles = []string{"go.sum", "package-lock.json"}

// Message is a message for FORGE: the event name (forge.<owner>.<name>.<kind>) and its JSON body.
type Message struct {
	Event string
	Data  []byte
}

// The subjects of the forge module's events the bridge writes pull messages for (mesh-catalog modules/gitea,
// published by the module as mesh.mod.<module>.event.<event>).
const (
	UpdatedSubject = "mesh.mod.gitea.event.pull.updated"
	ClosedSubject  = "mesh.mod.gitea.event.pull.closed"
	MergedSubject  = "mesh.mod.gitea.event.pull.merged"
)

var sha40 = regexp.MustCompile(`^[0-9a-f]{40}$`)

// pull is the part of gitea.pull.merged (index.ts:186-207), gitea.pull.updated (mesh-catalog modules/gitea/trunk.ts:233-250) and gitea.pull.closed
// (index.ts:127-128) the bridge reads; head is the branch and head_sha the commit.
type pull struct {
	Owner          string   `json:"owner"`
	Repo           string   `json:"repo"`
	Number         int      `json:"number"`
	Head           string   `json:"head"`
	HeadSHA        string   `json:"head_sha"`
	ModuleDirs     []string `json:"module_dirs"`
	ModuleDirsSaid bool     `json:"module_dirs_said"`
	Base           string   `json:"base"`
	MergeSHA       string   `json:"merge_commit_sha"`
}

func unreadable(subject, why string, args ...any) error {
	return fmt.Errorf("%w: %s: %s", ErrUnreadable, subject, fmt.Sprintf(why, args...))
}

// Convert turns one event of the forge module, by its bus subject and body, into a message. An event it cannot read
// is an ErrUnreadable error naming the subject and why; an event it writes nothing for is ErrNotHandled. A merge into
// the trunk asks trees once and becomes a merged message; a merge into any other branch is a closed pull message, as the
// pull request left its group. A forge that cannot say is ErrRetry; a merge with no module.json is ErrNoModules.
func Convert(subject string, data []byte, trees Trees, files Files) (Message, error) {
	var action string
	switch subject {
	case UpdatedSubject:
		action = delivery.PullUpdated
	case ClosedSubject:
		action = delivery.PullClosed
	case MergedSubject:
		action = delivery.PullClosed // unless it merged into the trunk, below
	default:
		return Message{}, fmt.Errorf("%w: %s", ErrNotHandled, subject)
	}
	var p pull
	if err := json.Unmarshal(data, &p); err != nil {
		return Message{}, unreadable(subject, "the body is not json: %v", err)
	}
	switch {
	case p.Owner == "" || p.Repo == "":
		return Message{}, unreadable(subject, "no owner or repo")
	case p.Number <= 0:
		return Message{}, unreadable(subject, "no pull request number")
	case !sha40.MatchString(p.HeadSHA):
		return Message{}, unreadable(subject, "the head %q is not 40 hex digits", p.HeadSHA)
	case p.Head == "":
		return Message{}, unreadable(subject, "no branch")
	}
	if subject == MergedSubject && p.Base == Trunk {
		return merged(subject, p, trees, files)
	}
	f := delivery.Forge{Kind: delivery.ForgePull, Repository: p.Owner + "/" + p.Repo, Branch: p.Head, Number: p.Number,
		Head: p.HeadSHA, Action: action}
	// The directories are the forge's word only when it said it could read them (module_dirs_said).
	if action == delivery.PullUpdated && p.ModuleDirsSaid {
		f.Dirs, f.DirsKnown = p.ModuleDirs, true
		var err error
		if f.Manifests, f.Locks, err = inputsOf(p.Owner, p.Repo, p.HeadSHA, p.ModuleDirs, files); err != nil {
			return Message{}, fmt.Errorf("%w: %s: the manifests of %s/%s at the head %s: %v", ErrRetry, subject, p.Owner, p.Repo, p.HeadSHA, err)
		}
	}
	event, err := delivery.ForgeEvent(f.Repository, delivery.ForgePull)
	if err != nil {
		return Message{}, unreadable(subject, "%v", err)
	}
	body, err := delivery.EncodeForge(f)
	if err != nil {
		return Message{}, unreadable(subject, "%v", err)
	}
	return Message{Event: event, Data: body}, nil
}

// MaxReadsInFlight is how many files of the forge inputsOf asks for at once. A merge into a repository of many
// modules (131 in the catalogue) reads three files for each, one after another inside the bound of a handled event
// (30 s) that leaves about 76 ms a call; eight at once keeps the forge's load small and the read well inside the bound
// (issue 1458).
const MaxReadsInFlight = 8

// read is one file asked for: the path in the directory dir, and whether it is the module's manifest.
type read struct {
	dir, path string
	manifest  bool
	raw       []byte
	err       error
}

// inputsOf reads, at ref, the module.json of each directory (the root spelled "") and the lock files beside it
// (LockFiles): the manifests as the forge answered them and the lock files' hex sha256 by name. A manifest the forge
// does not have is an error, since the directory was said to hold one; a lock file it does not have is left out; any other
// failure is an error (issue 1448).
//
// The files are asked for with at most MaxReadsInFlight calls at once (issue 1458). The result does not depend on the
// order the calls finish in, and the error is that of the first failing file in the order of dirs and LockFiles. After
// a failure no further file is asked for, and inputsOf returns only when every call it started has returned.
func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests map[string]string, locks map[string]map[string]string, err error) {
	at := func(dir, name string) string {
		if dir == "" {
			return name
		}
		return dir + "/" + name
	}
	var reads []*read
	for _, dir := range dirs {
		reads = append(reads, &read{dir: dir, path: at(dir, "module.json"), manifest: true})
		for _, name := range LockFiles {
			reads = append(reads, &read{dir: dir, path: at(dir, name)})
		}
	}
	var (
		wg     sync.WaitGroup
		mu     sync.Mutex
		failed bool
		slots  = make(chan struct{}, MaxReadsInFlight)
	)
	for _, r := range reads {
		slots <- struct{}{}
		mu.Lock()
		stop := failed
		mu.Unlock()
		if stop {
			<-slots
			break
		}
		wg.Add(1)
		go func(r *read) {
			defer func() { <-slots; wg.Done() }()
			r.raw, r.err = files(owner, repo, r.path, ref)
			if r.err != nil && (r.manifest || !errors.Is(r.err, ErrNotFound)) {
				mu.Lock()
				failed = true
				mu.Unlock()
			}
		}(r)
	}
	wg.Wait()
	manifests = map[string]string{}
	locks = map[string]map[string]string{}
	for _, r := range reads {
		switch {
		case r.err != nil && (r.manifest || !errors.Is(r.err, ErrNotFound)):
			return nil, nil, fmt.Errorf("%s: %w", r.path, r.err)
		case r.err != nil: // a lock file the forge does not have
		case r.manifest:
			manifests[r.dir] = string(r.raw)
		default:
			sum := sha256.Sum256(r.raw)
			if locks[r.dir] == nil {
				locks[r.dir] = map[string]string{}
			}
			locks[r.dir][path.Base(r.path)] = hex.EncodeToString(sum[:])
		}
	}
	if len(manifests) == 0 {
		manifests = nil
	}
	if len(locks) == 0 {
		locks = nil
	}
	return manifests, locks, nil
}

// merged is the merged message of a merge into the trunk: the trees of the merge commit, asked once.
func merged(subject string, p pull, trees Trees, files Files) (Message, error) {
	if !sha40.MatchString(p.MergeSHA) {
		return Message{}, unreadable(subject, "the merge commit %q is not 40 hex digits", p.MergeSHA)
	}
	repository := p.Owner + "/" + p.Repo
	root, dirs, err := trees(p.Owner, p.Repo, p.MergeSHA)
	if err != nil {
		return Message{}, fmt.Errorf("%w: %s: the trees of %s at the merge commit %s: %v", ErrRetry, subject, repository, p.MergeSHA, err)
	}
	if len(dirs) == 0 {
		return Message{}, fmt.Errorf("%w: %s: %s at the merge commit %s", ErrNoModules, subject, repository, p.MergeSHA)
	}
	names := make([]string, 0, len(dirs))
	for dir := range dirs {
		names = append(names, dir)
	}
	slices.Sort(names)
	manifests, locks, err := inputsOf(p.Owner, p.Repo, p.MergeSHA, names, files)
	if err != nil {
		return Message{}, fmt.Errorf("%w: %s: the manifests of %s at the merge commit %s: %v", ErrRetry, subject, repository, p.MergeSHA, err)
	}
	event, err := delivery.ForgeEvent(repository, delivery.ForgeMerged)
	if err != nil {
		return Message{}, unreadable(subject, "%v", err)
	}
	body, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: repository, Branch: p.Head, Number: p.Number,
		Head: p.HeadSHA, Merge: p.MergeSHA, Root: root, Trees: dirs, Manifests: manifests, Locks: locks})
	if err != nil {
		return Message{}, unreadable(subject, "%v", err)
	}
	return Message{Event: event, Data: body}, nil
}

// openPull is an entry of gitea_open_pulls's answer (mesh-catalog modules/gitea/client.ts:369-381 openPulls, with the
// branch as `head`, issue 1234/AC5), the part the bridge reads.
type openPull struct {
	Owner   string `json:"owner"`
	Repo    string `json:"repo"`
	Number  int    `json:"number"`
	Base    string `json:"base"`
	Head    string `json:"head"`
	HeadSHA string `json:"head_sha"`
}

// OpenPulls turns the answer of gitea_open_pulls, {pulls: [...]}, into one pull message per open pull request into
// the trunk, action updated and its directories not known (the forge's answer does not say them). A pull request into
// another branch gets none. An entry that cannot be read is named in skipped and gets none; an answer that is not
// {pulls} is an error.
func OpenPulls(answer []byte) (msgs []Message, skipped []string, err error) {
	var a struct {
		Pulls *[]openPull `json:"pulls"`
	}
	if err := json.Unmarshal(answer, &a); err != nil || a.Pulls == nil {
		return nil, nil, fmt.Errorf("the answer of gitea_open_pulls is not {pulls: [...]}: %v", err)
	}
	for _, p := range *a.Pulls {
		if p.Base != Trunk {
			continue
		}
		name := fmt.Sprintf("%s/%s#%d", p.Owner, p.Repo, p.Number)
		if p.Owner == "" || p.Repo == "" || p.Number <= 0 || p.Head == "" || !sha40.MatchString(p.HeadSHA) {
			skipped = append(skipped, name+": it names no owner, repo, number, branch or 40-hex head")
			continue
		}
		f := delivery.Forge{Kind: delivery.ForgePull, Repository: p.Owner + "/" + p.Repo, Branch: p.Head, Number: p.Number,
			Head: p.HeadSHA, Action: delivery.PullUpdated}
		event, err := delivery.ForgeEvent(f.Repository, delivery.ForgePull)
		var body []byte
		if err == nil {
			body, err = delivery.EncodeForge(f)
		}
		if err != nil {
			skipped = append(skipped, name+": "+err.Error())
			continue
		}
		msgs = append(msgs, Message{Event: event, Data: body})
	}
	return msgs, skipped, nil
}

// Track says what a message the bridge wrote changes in the head it keeps for each pull request (the bucket heads,
// novox/hq ADR 0443 §5): the key `<owner>.<name>.<number>`, and the head to keep under it, or "" to take the key away
// (a closed pull request, a merged one). ok is false for a message that keeps no head.
func Track(f delivery.Forge) (key, head string, ok bool) {
	key = strings.ReplaceAll(f.Repository, "/", ".") + "." + strconv.Itoa(f.Number)
	switch {
	case f.Kind == delivery.ForgePull && f.Action != delivery.PullClosed:
		return key, f.Head, true
	case f.Kind == delivery.ForgePull || f.Kind == delivery.ForgeMerged:
		return key, "", true
	}
	return "", "", false
}

// OpenMessage turns the answer of gitea_open_pulls into the open message: every open pull request into the trunk, by
// repository and number, whether its head reads or not and whether FORGE holds it or not, in repository and number
// order. A pull request into another branch is not named. An answer that names none is a message too: nothing is open.
func OpenMessage(answer []byte) (Message, error) {
	var a struct {
		Pulls *[]openPull `json:"pulls"`
	}
	if err := json.Unmarshal(answer, &a); err != nil || a.Pulls == nil {
		return Message{}, fmt.Errorf("the answer of gitea_open_pulls is not {pulls: [...]}: %v", err)
	}
	open := []delivery.OpenPull{}
	for _, p := range *a.Pulls {
		if p.Base != Trunk || p.Owner == "" || p.Repo == "" || p.Number <= 0 {
			continue
		}
		open = append(open, delivery.OpenPull{Repository: p.Owner + "/" + p.Repo, Number: p.Number})
	}
	slices.SortFunc(open, func(a, b delivery.OpenPull) int {
		if c := strings.Compare(a.Repository, b.Repository); c != 0 {
			return c
		}
		return a.Number - b.Number
	})
	body, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeOpen, Open: open})
	if err != nil {
		return Message{}, fmt.Errorf("the open pull requests cannot be said: %w", err)
	}
	return Message{Event: delivery.OpenEvent(), Data: body}, nil
}