mesh-catalog/modules/gitea/cmd/package-registry/whole_integration_test.go

//go:build integration

package main

// The forge module's package seats whole against a real bus (CAT-G3; design 51 *Acceptance tests*, the holder's cases):
// the built package-registry bundle, played over its real stdin and stdout as the node's tool runtime plays it, its
// `incoming` an object store and its `transfers` and `findings` key-value buckets of a nats-server of this test alone
// (so the publish in parts runs through mesh-sdk's parts.BusObjects and BusTransfers, ruling 4 of the lead's 14:0x), the
// builder's builder_versions answered on mesh/ask, the config value `callers` on mesh/provision.get, and the forge an
// httptest server that takes reads only with the tools' token and writes only with the packages account's.
//
// A Go package the builder numbered goes up in two chunks, is uploaded whole as the packages account and says
// published on the Go seat with its module path; one the builder did not number is refused before the forge sees it,
// says refused and writes the seat's finding; versions answers what the forge holds.

import (
	"archive/zip"
	"bufio"
	"bytes"
	"context"
	"crypto/rand"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

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

	"gitea/internal/packageseats"
)

type wire struct {
	ID     json.RawMessage `json:"id,omitempty"`
	Method string          `json:"method,omitempty"`
	Params json.RawMessage `json:"params,omitempty"`
	Result json.RawMessage `json:"result,omitempty"`
}

// player is the node's tool runtime, as far as the package seats need it, over a real bus.
type player struct {
	ctx      context.Context
	t        *testing.T
	nc       *nats.Conn
	js       jetstream.JetStream
	numbered map[string]string // name@version -> the digest the builder numbered
	mu       sync.Mutex
	in       io.Writer
	next     int
	answers  map[string]chan wire
}

func (p *player) send(m map[string]any) {
	m["jsonrpc"] = "2.0"
	raw, _ := json.Marshal(m)
	p.mu.Lock()
	defer p.mu.Unlock()
	_, _ = p.in.Write(append(raw, '\n'))
}

func (p *player) answer(id json.RawMessage, result any, err error) {
	if err != nil {
		p.send(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
		return
	}
	p.send(map[string]any{"id": id, "result": result})
}

func (p *player) serve(out io.Reader) {
	scanner := bufio.NewScanner(out)
	scanner.Buffer(make([]byte, 1<<20), 1<<26)
	for scanner.Scan() {
		var m wire
		if json.Unmarshal(scanner.Bytes(), &m) != nil {
			continue
		}
		if m.Method == "" {
			var id string
			_ = json.Unmarshal(m.ID, &id)
			p.mu.Lock()
			ch := p.answers[id]
			p.mu.Unlock()
			if ch != nil {
				ch <- m
			}
			continue
		}
		go p.handle(m)
	}
}

// handle answers one call of the bundle on the runtime.
func (p *player) handle(m wire) {
	var a struct {
		State   string          `json:"state"`
		Key     string          `json:"key"`
		Value   json.RawMessage `json:"value"`
		Objects string          `json:"objects"`
		Name    string          `json:"name"`
		Data    []byte          `json:"data"`
		Subject string          `json:"subject"`
		Body    json.RawMessage `json:"body"`
		Kind    string          `json:"kind"`
		TTL     int             `json:"ttl_seconds"`
	}
	_ = json.Unmarshal(m.Params, &a)
	switch {
	case m.Method == "mesh/ask" && a.Key == delivery.BuilderVersionsAddress:
		var q struct{ Name, Version, Digest string }
		_ = json.Unmarshal(a.Body, &q)
		d, ok := p.numbered[q.Name+"@"+q.Version]
		p.answer(m.ID, map[string]any{"recorded": ok && d == q.Digest, "why": "not numbered with that digest"}, nil)
	case m.Method == "mesh/provision.get" && a.Kind == "config":
		p.answer(m.ID, nil, nil) // `callers` unset: design 51's list
	case m.Method == "mesh/seat.publish":
		p.answer(m.ID, map[string]any{}, p.nc.Publish(a.Subject, a.Body))
	case m.Method == "mesh/publish":
		p.answer(m.ID, map[string]any{}, nil)
	case strings.HasPrefix(m.Method, "mesh/objects."):
		p.objects(m, a.Objects, a.Name, a.Data)
	case strings.HasPrefix(m.Method, "mesh/state."):
		p.state(m, a.State, a.Key, a.Value, a.TTL)
	default:
		p.answer(m.ID, nil, errors.New("not played here: "+m.Method))
	}
}

func (p *player) objects(m wire, local, name string, data []byte) {
	os, err := p.js.ObjectStore(p.ctx, "gitea_"+local)
	if err != nil {
		p.answer(m.ID, nil, err)
		return
	}
	switch m.Method {
	case "mesh/objects.put":
		info, err := os.PutBytes(p.ctx, name, data)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		p.answer(m.ID, map[string]any{"name": info.Name, "size": info.Size, "digest": info.Digest}, nil)
	case "mesh/objects.get":
		got, err := os.GetBytes(p.ctx, name)
		p.answer(m.ID, map[string]any{"info": map[string]any{"name": name, "size": len(got)}, "data": got}, err)
	case "mesh/objects.delete":
		err := os.Delete(p.ctx, name)
		if errors.Is(err, jetstream.ErrObjectNotFound) {
			err = nil
		}
		p.answer(m.ID, map[string]any{}, err)
	case "mesh/objects.info":
		st, err := os.Status(p.ctx)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		p.answer(m.ID, map[string]any{"bytes": st.Size(), "max_bytes": 2048 << 20}, nil)
	default:
		p.answer(m.ID, nil, errors.New("not played here: "+m.Method))
	}
}

func (p *player) state(m wire, state, key string, value json.RawMessage, ttl int) {
	kv, err := p.js.KeyValue(p.ctx, "gitea_"+state)
	if err != nil {
		p.answer(m.ID, nil, err)
		return
	}
	switch m.Method {
	case "mesh/state.get":
		e, err := kv.Get(p.ctx, key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			p.answer(m.ID, nil, nil)
			return
		}
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		p.answer(m.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil)
	case "mesh/state.put":
		rev, err := kv.Put(p.ctx, key, value)
		p.answer(m.ID, map[string]any{"revision": rev}, err)
	case "mesh/state.create":
		var opts []jetstream.KVCreateOpt
		if ttl > 0 {
			opts = append(opts, jetstream.KeyTTL(time.Duration(ttl)*time.Second))
		}
		rev, err := kv.Create(p.ctx, key, value, opts...)
		p.answer(m.ID, map[string]any{"revision": rev}, err)
	case "mesh/state.delete":
		p.answer(m.ID, map[string]any{}, kv.Delete(p.ctx, key))
	case "mesh/state.keys":
		keys, err := kv.Keys(p.ctx)
		if errors.Is(err, jetstream.ErrNoKeysFound) {
			keys, err = []string{}, nil
		}
		p.answer(m.ID, keys, err)
	default:
		p.answer(m.ID, nil, errors.New("not played here: "+m.Method))
	}
}

// call is one tools/call as the runtime makes it, the caller named in _meta as the runtime proves it.
func (p *player) call(caller, tool string, args map[string]any) (json.RawMessage, error) {
	p.mu.Lock()
	p.next++
	id := fmt.Sprintf("call-%d", p.next)
	ch := make(chan wire, 1)
	p.answers[id] = ch
	p.mu.Unlock()
	p.send(map[string]any{"id": id, "method": "tools/call", "params": map[string]any{"name": tool, "arguments": args,
		"_meta": map[string]any{"mesh/caller": caller}}})
	select {
	case m := <-ch:
		var r struct {
			Content []struct{ Text string } `json:"content"`
			IsError bool                    `json:"isError"`
		}
		if err := json.Unmarshal(m.Result, &r); err != nil || len(r.Content) == 0 {
			return nil, fmt.Errorf("%s answered %s", tool, m.Result)
		}
		if r.IsError {
			return nil, errors.New(r.Content[0].Text)
		}
		return json.RawMessage(r.Content[0].Text), nil
	case <-time.After(30 * time.Second):
		return nil, fmt.Errorf("%s did not answer", tool)
	}
}

// bigZip is a Go module zip of about 1.2 MiB that does not compress, so its file goes in two chunks.
func bigZip(t *testing.T, prefix string) []byte {
	noise := make([]byte, 1200<<10)
	_, _ = rand.Read(noise)
	var b bytes.Buffer
	z := zip.NewWriter(&b)
	w, _ := z.Create(prefix + "/go.mod")
	_, _ = w.Write([]byte("module " + strings.Split(prefix, "@")[0] + "\n"))
	w, _ = z.CreateHeader(&zip.FileHeader{Name: prefix + "/noise.bin", Method: zip.Store})
	_, _ = w.Write(noise)
	_ = z.Close()
	return b.Bytes()
}

func TestThePackageSeatsWhole(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	t.Cleanup(cancel)
	bin := filepath.Join(t.TempDir(), "package-registry")
	build := exec.Command("go", "build", "-o", bin, ".")
	build.Stderr = os.Stderr
	if err := build.Run(); err != nil {
		t.Fatalf("package-registry does not build: %v", err)
	}
	nc, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, _ := jetstream.New(nc)
	for _, b := range []string{"gitea_" + packageseats.TransfersState, "gitea_findings"} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
			t.Fatal(err)
		}
	}
	// idle takes keys with their own time to live (key-ttl), as the bus module makes it.
	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "gitea_" + packageseats.IdleState, LimitMarkerTTL: time.Minute}); err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "gitea_" + packageseats.IncomingObjects}); err != nil {
		t.Fatal(err)
	}
	heard := make(chan *nats.Msg, 8)
	if _, err := nc.ChanSubscribe("mesh.seat.go-package-registry.event.>", heard); err != nil {
		t.Fatal(err)
	}

	// The forge: reads only as the tools' account, writes only as the packages account.
	g := &registry{uploaded: map[string][]byte{}, held: map[string][]byte{}}
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		g.handle(w, r)
		if r.Method == http.MethodPut && r.URL.Path == "/api/packages/acme/go/upload" {
			g.mu.Lock()
			g.held["go/git.example/acme/sdk/v0.1.15"] = g.uploaded["go"]
			g.mu.Unlock()
		}
	}))
	t.Cleanup(srv.Close)
	dir := t.TempDir()
	kept := func(name, token, line string) string {
		file := filepath.Join(dir, name)
		_ = os.WriteFile(file, []byte(token+"\n"), 0o600)
		_ = os.WriteFile(file+".scopes", []byte(line+"\n"), 0o600)
		return file
	}
	cmd := exec.CommandContext(ctx, bin)
	cmd.Env = append(os.Environ(), "MESH_GITEA_URL="+srv.URL, "MESH_NPM_OWNER=acme", "MESH_GITEA_TOOLS_USER=mesh-tools",
		"MESH_GITEA_TOKEN_FILE="+kept("token", "reader", "read:package,read:user,write:issue,write:repository for mesh-tools"),
		"MESH_GITEA_PACKAGES_TOKEN_FILE="+kept("packages-token", "writer", "write:package for mesh-packages"))
	cmd.Stderr = os.Stderr
	in, _ := cmd.StdinPipe()
	out, _ := cmd.StdoutPipe()
	if err := cmd.Start(); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = in.Close(); _ = cmd.Wait() })

	zipped := bigZip(t, "git.example/acme/sdk@v0.1.15")
	digest := "sha256:" + sha(zipped)
	p := &player{ctx: ctx, t: t, nc: nc, js: js, in: in, answers: map[string]chan wire{},
		numbered: map[string]string{"git.example/acme/sdk@0.1.15": digest}}
	go p.serve(out)
	p.send(map[string]any{"id": "init", "method": "initialize", "params": map[string]any{}})

	builder := "ace.mesh-build-go"
	publish := func(version, digest string, size int) (string, []string) {
		raw, err := p.call(builder, "go-package-registry.publish", map[string]any{"name": "git.example/acme/sdk", "version": version,
			"kind": "package", "digest": digest, "transfer": "sdk-" + strings.ReplaceAll(version, ".", "-"), "parts": []map[string]any{{"digest": digest, "size": size}}})
		if err != nil {
			t.Fatalf("publish %s: %v", version, err)
		}
		var a struct {
			Transfer string   `json:"transfer"`
			Lacks    []string `json:"lacks"`
		}
		_ = json.Unmarshal(raw, &a)
		return a.Transfer, a.Lacks
	}
	send := func(transfer, digest string, file []byte) error {
		const chunk = 700 << 10
		for seq, at := 0, 0; at < len(file); seq, at = seq+1, at+chunk {
			end := min(at+chunk, len(file))
			if _, err := p.call(builder, "go-package-registry.publish-part", map[string]any{"transfer": transfer, "part": digest,
				"sequence": seq, "last": end == len(file), "data": base64.StdEncoding.EncodeToString(file[at:end])}); err != nil {
				return err
			}
		}
		return nil
	}
	event := func(want string) map[string]any {
		select {
		case m := <-heard:
			if !strings.HasSuffix(m.Subject, "."+want) {
				t.Fatalf("the Go seat said %s, want %s: %s", m.Subject, want, m.Data)
			}
			var o map[string]any
			_ = json.Unmarshal(m.Data, &o)
			return o
		case <-time.After(30 * time.Second):
			t.Fatalf("the Go seat said no %s", want)
		}
		return nil
	}

	// A publish naming no transfer is refused: the transfer's name is the caller's own.
	if _, err := p.call(builder, "go-package-registry.publish", map[string]any{"name": "git.example/acme/sdk", "version": "0.1.15",
		"kind": "package", "digest": digest, "parts": []map[string]any{{"digest": digest, "size": len(zipped)}}}); err == nil ||
		!strings.Contains(err.Error(), "no transfer named") {
		t.Errorf("a publish naming no transfer: %v", err)
	}
	// A version the builder numbered: two chunks through incoming and transfers, uploaded whole as the packages account.
	transfer, lacks := publish("0.1.15", digest, len(zipped))
	if transfer == "" || len(lacks) != 1 || lacks[0] != digest {
		t.Fatalf("publish answered transfer %q lacks %v", transfer, lacks)
	}
	// The open transfer is bounded by its idle key, written through the runtime with its own time to live.
	idle, _ := js.KeyValue(ctx, "gitea_"+packageseats.IdleState)
	if _, err := idle.Get(ctx, packageseats.GoSeat+"_"+transfer); err != nil {
		t.Errorf("the open transfer %s has no idle key: %v", transfer, err)
	}
	if err := send(transfer, digest, zipped); err != nil {
		t.Fatal(err)
	}
	if o := event("published"); o["module"] != "git.example/acme/sdk" || o["reference"] != "git.example/acme/sdk@0.1.15" || o["holder"] != "gitea" {
		t.Errorf("published says %v", o)
	}
	g.mu.Lock()
	up := g.uploaded["go"]
	g.mu.Unlock()
	if !bytes.Equal(up, zipped) {
		t.Errorf("the forge took %d bytes, not the %d published", len(up), len(zipped))
	}
	transfers, _ := js.KeyValue(ctx, "gitea_"+packageseats.TransfersState)
	if keys, _ := transfers.Keys(ctx); len(keys) != 0 {
		t.Errorf("an ended transfer is still kept: %v", keys)
	}

	// One the builder did not number: refused before the forge sees it, said, and the seat's finding written.
	other := bigZip(t, "git.example/acme/sdk@v0.1.16")
	od := "sha256:" + sha(other)
	transfer, _ = publish("0.1.16", od, len(other))
	_ = send(transfer, od, other)
	if o := event("refused"); !strings.Contains(fmt.Sprint(o["reason"]), "did not number") {
		t.Errorf("refused says %v", o)
	}
	findings, _ := js.KeyValue(ctx, "gitea_findings")
	if keys, _ := findings.Keys(ctx); len(keys) == 0 {
		t.Error("a refused publish wrote no finding")
	}
	g.mu.Lock()
	_, uploadedOther := g.held["go/git.example/acme/sdk/v0.1.16"]
	g.mu.Unlock()
	if uploadedOther {
		t.Error("a version the builder did not number reached the forge")
	}

	// Anyone reads versions; the forge was read as the tools' account.
	raw, err := p.call("g14.someone", "go-package-registry.versions", map[string]any{"name": "git.example/acme/sdk"})
	if err != nil || !strings.Contains(string(raw), `"version":"0.1.15"`) || !strings.Contains(string(raw), digest) {
		t.Errorf("versions answered %s, %v", raw, err)
	}
}