From 455b9023324f3d9d9fecf84bfda87433dbc8a5c5 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 14:53:09 +0200 Subject: [PATCH] Say a trunk merge on FORGE with the merge commit's trees (issue 1236) --- .../cmd/mesh-delivery-forge-bridge/call.go | 91 +++++++++++++ .../integration_test.go | 96 +++++++++++++- .../cmd/mesh-delivery-forge-bridge/main.go | 20 ++- .../mesh-delivery-forge-bridge/main_test.go | 87 ++++++++++-- .../internal/bridge/bridge.go | 55 +++++++- .../internal/bridge/bridge_test.go | 125 ++++++++++++++++-- 6 files changed, 446 insertions(+), 28 deletions(-) create mode 100644 modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go new file mode 100644 index 0000000..f1c1461 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go @@ -0,0 +1,91 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/nats-io/nats.go" + + "mesh-delivery-forge-bridge/internal/bridge" +) + +// callTimeout bounds a call to a tool (mesh-sdk go/process/call.go CallTimeout). +const callTimeout = 30 * time.Second + +// caller calls a module's tool by its manifest key (`.`) with a JSON body and returns the tool's result. +type caller func(ctx context.Context, key string, body any) (json.RawMessage, error) + +// callerOn calls tools as the bridge's own principal. It copies mesh-sdk go/process/call.go (Call, callSubjects, +// request; hand-written here until the SDK's helper is used, issue 1242): the request is the tool's arguments as +// JSON, published on the caller-named subject `mesh.mod..call..`, or on the subject that +// names no caller, `mesh.mod..tool.`, when nothing answers the first; the reply is +// {"result", "error", "node"} and a non-empty error is the call's failure (mesh-tools node-tools/internal/bus/bus.go +// answerOf). The grant to publish on these subjects comes from the manifest's invokes alone (mesh-control +// internal/broker/nats.go invokedSubjects). +func callerOn(nc *nats.Conn, user string) caller { + token := strings.ReplaceAll(user, ".", "~") + request := func(ctx context.Context, subject string, data []byte) (json.RawMessage, error) { + ctx, cancel := context.WithTimeout(ctx, callTimeout) + defer cancel() + msg, err := nc.RequestWithContext(ctx, subject, data) + if err != nil { + return nil, err + } + var r struct { + Result json.RawMessage `json:"result"` + Error string `json:"error"` + Node string `json:"node"` + } + if err := json.Unmarshal(msg.Data, &r); err != nil { + return nil, fmt.Errorf("%s answered something that is not a tool's answer: %w", subject, err) + } + if r.Error != "" { + return nil, errors.New(r.Error) + } + return r.Result, nil + } + return func(ctx context.Context, key string, body any) (json.RawMessage, error) { + module, tool, ok := strings.Cut(key, ".") + if !ok || module == "" || tool == "" { + return nil, fmt.Errorf("%q is not .", key) + } + data, err := json.Marshal(body) + if err != nil { + return nil, err + } + result, err := request(ctx, "mesh.mod."+module+".call."+tool+"."+token, data) + if errors.Is(err, nats.ErrNoResponders) { + result, err = request(ctx, "mesh.mod."+module+".tool."+tool, data) + } + return result, err + } +} + +// commitTreesTool is the forge's tool for a commit's trees (mesh-catalog modules/gitea, issue 1234): arguments owner, +// repo and sha, answer {root, trees}, the root directory spelled "" (1234/AC4). +const commitTreesTool = "gitea.gitea_commit_trees" + +// treesVia asks the forge for a commit's trees through call. +func treesVia(ctx context.Context, call caller) bridge.Trees { + return func(owner, repo, sha string) (string, map[string]string, error) { + raw, err := call(ctx, commitTreesTool, map[string]string{"owner": owner, "repo": repo, "sha": sha}) + if err != nil { + return "", nil, err + } + var a struct { + Root string `json:"root"` + Trees map[string]string `json:"trees"` + } + if err := json.Unmarshal(raw, &a); err != nil { + return "", nil, fmt.Errorf("the answer of %s is not {root, trees}: %w", commitTreesTool, err) + } + if a.Root == "" { + return "", nil, fmt.Errorf("the answer of %s names no root tree", commitTreesTool) + } + return a.Root, a.Trees, nil + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go index b4ff62c..6262776 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go @@ -63,10 +63,10 @@ func startBus(t *testing.T) *server.Server { return srv } -// TestBridgeWritesPullMessages starts the program the way the node does (its entry, run, with MESH_BROKER_FILE) against -// an embedded bus with the stream EVENTS, the bridge's stream FORGE and its durable consumer _ on EVENTS, -// as the controller makes them (mesh-control internal/broker streams.go:192, nats.go:1387-1396, modstreams.go). -func TestBridgeWritesPullMessages(t *testing.T) { // 1235/AC2 +// startBridge runs the program against an embedded bus with the stream EVENTS, the bridge's stream FORGE and its durable +// consumer _ on EVENTS, as the controller makes them (mesh-control internal/broker streams.go:192, +// nats.go:1387-1396, modstreams.go). It returns a control connection, its JetStream and a context. +func startBridge(t *testing.T) (*nats.Conn, jetstream.JetStream, context.Context) { srv := startBus(t) ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw")) if err != nil { @@ -78,7 +78,7 @@ func TestBridgeWritesPullMessages(t *testing.T) { // 1235/AC2 t.Fatal(err) } ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) - defer cancel() + t.Cleanup(cancel) if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil { t.Fatal(err) } @@ -102,6 +102,12 @@ func TestBridgeWritesPullMessages(t *testing.T) { // 1235/AC2 done := make(chan error, 1) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() t.Cleanup(func() { stop(); <-done }) + return ctl, js, ctx +} + +// TestBridgeWritesPullMessages starts the program the way the node does (its entry, run, with MESH_BROKER_FILE). +func TestBridgeWritesPullMessages(t *testing.T) { // 1235/AC2 + _, js, ctx := startBridge(t) // gitea.pull.updated as the forge module publishes it: the body of trunk.ts:233-250 on mesh.mod.gitea.event.pull.updated. ev, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-delivery", "number": 12, "title": "t", "body": "b", "base": "main", @@ -152,3 +158,83 @@ func TestBridgeWritesPullMessages(t *testing.T) { // 1235/AC2 ci, _ := cons.Info(ctx) t.Errorf("the event is not acknowledged: %+v", ci) } + +const merge = "89abcdef0123456789abcdef0123456789abcdef" + +// TestBridgeWritesMergedMessage has a stand-in answer the forge's gitea_commit_trees on its tool subject +// (mesh-control internal/broker/nats.go invokedSubjects: mesh.mod..tool.), in the reply envelope of +// mesh-sdk go/process/call.go ({result, error, node}). +func TestBridgeWritesMergedMessage(t *testing.T) { // 1236/AC4 + ctl, js, ctx := startBridge(t) + var asked chan []byte = make(chan []byte, 4) + sub, err := ctl.Subscribe("mesh.mod.gitea.tool.gitea_commit_trees", func(m *nats.Msg) { + asked <- m.Data + _ = m.Respond([]byte(`{"result":{"root":"r0","trees":{"":"t0","modules/a":"t1"}},"node":"n0"}`)) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + if err := ctl.Flush(); err != nil { + t.Fatal(err) + } + + // gitea.pull.merged as mesh-catalog modules/gitea/index.ts:186-207 publishes it. + ev, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-delivery", "number": 12, "title": "t", "body": "b", + "head_sha": head, "head": "feat/x", "base": "main", "merge_commit_sha": merge, "merged_at": "2026-10-19T12:00:00Z", + "clone_url": "u", "html_url": "h", "paths": []string{"modules/a/x.go"}, "paths_truncated": false, "removed": []string{}, + "module_dirs": []string{"modules/a"}, "module_dirs_said": true}) + if _, err := js.Publish(ctx, "mesh.mod.gitea.event.pull.merged", ev); err != nil { + t.Fatal(err) + } + + forge, err := js.Stream(ctx, "FORGE") + if err != nil { + t.Fatal(err) + } + var info *jetstream.StreamInfo + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 1 { + break + } + } + if info.State.Msgs != 1 { + t.Fatalf("FORGE holds %d messages within 5 s, want 1", info.State.Msgs) + } + raw, err := forge.GetLastMsgForSubject(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.merged") + if err != nil { + t.Fatal(err) + } + f, err := delivery.DecodeForge(raw.Data) + if err != nil { + t.Fatal(err) + } + want := delivery.Forge{V: delivery.Version, Kind: "merged", Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, + Merge: merge, Root: "r0", Trees: map[string]string{"": "t0", "modules/a": "t1"}} + if !delivery.SameForge(f, want) { + t.Errorf("message = %+v, want %+v", f, want) + } + select { + case body := <-asked: + if string(body) != `{"owner":"novox","repo":"mesh-delivery","sha":"`+merge+`"}` { + t.Errorf("the tool was asked %s", body) + } + default: + t.Error("the stand-in was not asked") + } + if len(asked) != 0 { + t.Error("the tool was asked more than once") + } + cons, err := js.Consumer(ctx, "EVENTS", "n1_mesh-delivery-forge-bridge") + if err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + ci, err := cons.Info(ctx) + if err == nil && ci.NumAckPending == 0 && ci.NumPending == 0 && ci.Delivered.Consumer >= 1 { + return + } + } + ci, _ := cons.Info(ctx) + t.Errorf("the event is not acknowledged: %+v", ci) +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index e96ddb1..b896c7e 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -19,6 +19,7 @@ import ( "os/signal" "strings" "syscall" + "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -98,8 +99,13 @@ type delivered interface { Headers() nats.Header Ack() error Term() error + NakWithDelay(delay time.Duration) error } +// retryDelay is how long the server waits before it delivers again an event the forge could not be asked about (the +// server's redelivery is the only clock; issue 1236/AC3). +const retryDelay = 10 * time.Second + // publisher writes to FORGE. type publisher interface { Publish(ctx context.Context, subject string, payload []byte, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error) @@ -108,9 +114,17 @@ type publisher interface { // handle turns one event into its message on FORGE and acknowledges it. An event it cannot read is terminated, never // redelivered, and the log names its subject and why. A message it could not write stays unacknowledged: the server // delivers it again (the server's ack wait is the only clock). -func handle(ctx context.Context, m delivered, pub publisher, logger *log.Logger) { - msg, err := bridge.Convert(m.Subject(), m.Data()) +func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, logger *log.Logger) { + msg, err := bridge.Convert(m.Subject(), m.Data(), trees) switch { + case errors.Is(err, bridge.ErrRetry): + logger.Printf("%v; heard again in %s", err, retryDelay) + _ = m.NakWithDelay(retryDelay) + return + case errors.Is(err, bridge.ErrNoModules): + logger.Printf("%v; no merged message", err) + _ = m.Ack() + return case errors.Is(err, bridge.ErrNotHandled): logger.Printf("%v; nothing written", err) _ = m.Ack() @@ -166,7 +180,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, eventsStream, err) } - cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, js, logger) }, + cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, js, treesVia(ctx, callerOn(conn, cred.User)), logger) }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) if err != nil { return err diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go index 3e1b9ae..123bd49 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go @@ -8,6 +8,7 @@ import ( "log" "strings" "testing" + "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -21,13 +22,15 @@ type fakeMsg struct { subject string data []byte acked, termd int + nak []time.Duration } -func (f *fakeMsg) Subject() string { return f.subject } -func (f *fakeMsg) Data() []byte { return f.data } -func (f *fakeMsg) Headers() nats.Header { return nats.Header{"X-Event-Id": {"ev-1"}} } -func (f *fakeMsg) Ack() error { f.acked++; return nil } -func (f *fakeMsg) Term() error { f.termd++; return nil } +func (f *fakeMsg) Subject() string { return f.subject } +func (f *fakeMsg) Data() []byte { return f.data } +func (f *fakeMsg) Headers() nats.Header { return nats.Header{"X-Event-Id": {"ev-1"}} } +func (f *fakeMsg) Ack() error { f.acked++; return nil } +func (f *fakeMsg) NakWithDelay(d time.Duration) error { f.nak = append(f.nak, d); return nil } +func (f *fakeMsg) Term() error { f.termd++; return nil } type fakePub struct { subjects []string @@ -61,7 +64,7 @@ func body(m map[string]any) []byte { func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit part; the bus is the integration test) var logs bytes.Buffer m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{} - handle(context.Background(), m, pub, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, log.New(&logs, "", 0)) if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull" { t.Fatalf("published on %v", pub.subjects) } @@ -77,7 +80,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5 var logs bytes.Buffer m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(map[string]any{"head_sha": "short"})}, &fakePub{} - handle(context.Background(), m, pub, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, log.New(&logs, "", 0)) if len(pub.subjects) != 0 { t.Errorf("a message was written: %v", pub.subjects) } @@ -92,7 +95,7 @@ func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5 func TestUnwrittenMessageIsNotAcked(t *testing.T) { // 1235 (a failed write is redelivered by the server, not retried here) var logs bytes.Buffer m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{err: errors.New("no stream")} - handle(context.Background(), m, pub, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, log.New(&logs, "", 0)) if m.acked != 0 || m.termd != 0 { t.Errorf("acked %d terminated %d, want neither", m.acked, m.termd) } @@ -100,3 +103,71 @@ func TestUnwrittenMessageIsNotAcked(t *testing.T) { // 1235 (a failed write is r t.Errorf("log %q", logs.String()) } } + +// mergedEvent is gitea.pull.merged as mesh-catalog modules/gitea/index.ts:186-207 writes it (the fields the bridge reads). +func mergedEvent() []byte { + b, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-delivery", "number": 12, "head": "feat/x", "head_sha": head, + "base": "main", "merge_commit_sha": "89abcdef0123456789abcdef0123456789abcdef"}) + return b +} + +func TestTreesRefusedLeavesTheEventForTenSecondsAndLogs(t *testing.T) { // 1236/AC3 + var logs bytes.Buffer + m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.merged", data: mergedEvent()}, &fakePub{} + call := func(context.Context, string, any) (json.RawMessage, error) { + return nil, errors.New("the tree listing of novox/mesh-delivery was cut") + } + handle(context.Background(), m, pub, treesVia(context.Background(), call), log.New(&logs, "", 0)) + if len(pub.subjects) != 0 { + t.Errorf("a message was written: %v", pub.subjects) + } + if m.acked != 0 || m.termd != 0 || len(m.nak) != 1 || m.nak[0] < 10*time.Second { + t.Errorf("acked %d terminated %d naks %v, want one nak of at least 10 s", m.acked, m.termd, m.nak) + } + for _, w := range []string{"novox/mesh-delivery", "89abcdef0123456789abcdef0123456789abcdef", "the tree listing of novox/mesh-delivery was cut"} { + if !strings.Contains(logs.String(), w) { + t.Errorf("log %q does not name %q", logs.String(), w) + } + } +} + +func TestMergeCallsTheToolOnceAndWritesTheMergedMessage(t *testing.T) { // 1236/AC1 (through the tool's request and answer) + var logs bytes.Buffer + m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.merged", data: mergedEvent()}, &fakePub{} + var keys []string + var args []any + call := func(_ context.Context, key string, body any) (json.RawMessage, error) { + keys, args = append(keys, key), append(args, body) + return json.RawMessage(`{"root":"r0","trees":{"":"t0","modules/a":"t1"}}`), nil + } + handle(context.Background(), m, pub, treesVia(context.Background(), call), log.New(&logs, "", 0)) + if len(keys) != 1 || keys[0] != "gitea.gitea_commit_trees" { + t.Fatalf("calls %v", keys) + } + if got, _ := json.Marshal(args[0]); string(got) != `{"owner":"novox","repo":"mesh-delivery","sha":"89abcdef0123456789abcdef0123456789abcdef"}` { + t.Errorf("arguments %s", got) + } + if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.merged" { + t.Fatalf("published on %v", pub.subjects) + } + f, err := delivery.DecodeForge(pub.data[0]) + if err != nil || f.Root != "r0" || f.Trees[""] != "t0" || f.Trees["modules/a"] != "t1" || f.Merge == "" { + t.Errorf("message %+v err %v", f, err) + } + if m.acked != 1 { + t.Errorf("acked %d", m.acked) + } +} + +func TestAnAnswerWithoutARootIsNotUsed(t *testing.T) { // 1236 (a shape the tool does not answer) + call := func(context.Context, string, any) (json.RawMessage, error) { + return json.RawMessage(`{"trees":{"a":"t"}}`), nil + } + if _, _, err := treesVia(context.Background(), call)("o", "r", "s"); err == nil { + t.Error("an answer with no root was taken") + } + call = func(context.Context, string, any) (json.RawMessage, error) { return json.RawMessage(`[1]`), nil } + if _, _, err := treesVia(context.Background(), call)("o", "r", "s"); err == nil { + t.Error("an answer that is not an object was taken") + } +} diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index 81f2809..a75a991 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -17,6 +17,20 @@ 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) + // Message is a message for FORGE: the event name (forge...) and its JSON body. type Message struct { Event string @@ -28,11 +42,12 @@ type Message struct { 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.updated (mesh-catalog modules/gitea/trunk.ts:233-250) and gitea.pull.closed +// 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"` @@ -42,6 +57,8 @@ type pull struct { 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 { @@ -49,14 +66,18 @@ func unreadable(subject, why string, args ...any) error { } // 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. -func Convert(subject string, data []byte) (Message, error) { +// 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) (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) } @@ -74,6 +95,9 @@ func Convert(subject string, data []byte) (Message, error) { case p.Head == "": return Message{}, unreadable(subject, "no branch") } + if subject == MergedSubject && p.Base == Trunk { + return merged(subject, p, trees) + } 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). @@ -90,3 +114,28 @@ func Convert(subject string, data []byte) (Message, error) { } return Message{Event: event, Data: body}, 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) (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) + } + 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}) + if err != nil { + return Message{}, unreadable(subject, "%v", err) + } + return Message{Event: event, Data: body}, nil +} diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go index 00f95f0..981ba50 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go @@ -47,7 +47,7 @@ const updatedSubject = "mesh.mod.gitea.event.pull.updated" const closedSubject = "mesh.mod.gitea.event.pull.closed" func TestUpdatedBecomesAPullMessage(t *testing.T) { // 1235/AC2 (the conversion; the bus is the integration test) - msg, err := Convert(updatedSubject, updated(nil)) + msg, err := Convert(updatedSubject, updated(nil), nil) if err != nil { t.Fatal(err) } @@ -66,7 +66,7 @@ func TestUpdatedBecomesAPullMessage(t *testing.T) { // 1235/AC2 (the conversion; } func TestDirsNotSaidAreNotKnown(t *testing.T) { // 1235/AC3 - msg, err := Convert(updatedSubject, updated(map[string]any{"module_dirs": nil, "module_dirs_said": nil})) + msg, err := Convert(updatedSubject, updated(map[string]any{"module_dirs": nil, "module_dirs_said": nil}), nil) if err != nil { t.Fatal(err) } @@ -75,7 +75,7 @@ func TestDirsNotSaidAreNotKnown(t *testing.T) { // 1235/AC3 t.Errorf("dirs = %v known = %v, want not known", f.Dirs, f.DirsKnown) } // Said and empty is known and empty, not unknown. - msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs": []string{}, "module_dirs_said": true})) + msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs": []string{}, "module_dirs_said": true}), nil) if err != nil { t.Fatal(err) } @@ -84,7 +84,7 @@ func TestDirsNotSaidAreNotKnown(t *testing.T) { // 1235/AC3 t.Errorf("dirs = %v known = %v, want known and empty", f.Dirs, f.DirsKnown) } // Dirs listed without the forge saying it could read them are not trusted. - msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs_said": nil})) + msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs_said": nil}), nil) if err != nil { t.Fatal(err) } @@ -94,7 +94,7 @@ func TestDirsNotSaidAreNotKnown(t *testing.T) { // 1235/AC3 } func TestClosedBecomesAClosedPullMessage(t *testing.T) { // 1235/AC4 - msg, err := Convert(closedSubject, closed()) + msg, err := Convert(closedSubject, closed(), nil) if err != nil { t.Fatal(err) } @@ -130,7 +130,7 @@ func TestUnreadableEventsWriteNothing(t *testing.T) { // 1235/AC5 "closed no head": {closedSubject, mustDrop(closed(), "head_sha"), "head"}, "closed no repo": {closedSubject, mustDrop(closed(), "repo"), "repo"}, } { - msg, err := Convert(c.subject, c.data) + msg, err := Convert(c.subject, c.data, nil) if err == nil { t.Errorf("%s: a message %q was made", name, msg.Event) continue @@ -155,10 +155,117 @@ func mustDrop(body []byte, key string) []byte { return b } -func TestOtherSubjectsAreNotHandled(t *testing.T) { // 1235 (merged is issue 1236) - for _, s := range []string{"mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.something.else", "mesh.mod.other.event.pull.updated"} { - if _, err := Convert(s, updated(nil)); !errors.Is(err, ErrNotHandled) { +func TestOtherSubjectsAreNotHandled(t *testing.T) { // 1235 + for _, s := range []string{"mesh.mod.gitea.event.something.else", "mesh.mod.other.event.pull.updated"} { + if _, err := Convert(s, updated(nil), nil); !errors.Is(err, ErrNotHandled) { t.Errorf("%s: err = %v, want ErrNotHandled", s, err) } } } + +const merge = "89abcdef0123456789abcdef0123456789abcdef" + +// mergedBody is gitea.pull.merged as mesh-catalog modules/gitea/index.ts:186-207 writes it. +func mergedBody(extra map[string]any) []byte { + m := map[string]any{ + "owner": "novox", "repo": "mesh-delivery", "number": 12, "title": "A change", "body": "text", + "head_sha": head, "head_checks": map[string]string{"mesh/merge-gate": "success"}, "head": "feat/x", "base": "main", + "merge_commit_sha": merge, "merged_at": "2026-10-19T12:00:00Z", "clone_url": "https://forge.example/novox/mesh-delivery.git", + "html_url": "https://forge.example/novox/mesh-delivery/pulls/12", "paths": []string{"modules/a/x.go"}, "paths_truncated": false, + "removed": []string{}, "module_dirs": []string{"modules/a"}, "module_dirs_said": true, + } + for k, v := range extra { + if v == nil { + delete(m, k) + } else { + m[k] = v + } + } + b, _ := json.Marshal(m) + return b +} + +const mergedSubject = "mesh.mod.gitea.event.pull.merged" + +type call struct{ owner, repo, sha string } + +// treesOf answers as gitea_commit_trees does (issue 1234/AC4: root, and the directory to tree hash, the root spelled ""). +func treesOf(calls *[]call, root string, trees map[string]string, err error) Trees { + return func(owner, repo, sha string) (string, map[string]string, error) { + *calls = append(*calls, call{owner, repo, sha}) + return root, trees, err + } +} + +func TestMergeIntoMainBecomesAMergedMessage(t *testing.T) { // 1236/AC1 + var calls []call + answer := map[string]string{"modules/a": "t1", "modules/b": "t2"} + msg, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", answer, nil)) + if err != nil { + t.Fatal(err) + } + if msg.Event != "forge.novox.mesh-delivery.merged" { + t.Errorf("event = %q", msg.Event) + } + f, err := delivery.DecodeForge(msg.Data) + if err != nil { + t.Fatal(err) + } + want := delivery.Forge{V: delivery.Version, Kind: delivery.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, + Head: head, Merge: merge, Root: "r0", Trees: answer} + if !delivery.SameForge(f, want) { + t.Errorf("message = %+v, want %+v", f, want) + } + if len(calls) != 1 || calls[0] != (call{"novox", "mesh-delivery", merge}) { + t.Errorf("the tool was called %v, want once with novox, mesh-delivery and the merge commit", calls) + } +} + +func TestMergeIntoAnotherBranchIsAClosedPull(t *testing.T) { // 1236/AC2 + var calls []call + msg, err := Convert(mergedSubject, mergedBody(map[string]any{"base": "feat/y"}), treesOf(&calls, "r0", map[string]string{"a": "t"}, nil)) + if err != nil { + t.Fatal(err) + } + f, err := delivery.DecodeForge(msg.Data) + if err != nil { + t.Fatal(err) + } + want := delivery.Forge{V: delivery.Version, Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, + Head: head, Action: delivery.PullClosed} + if msg.Event != "forge.novox.mesh-delivery.pull" || !delivery.SameForge(f, want) { + t.Errorf("event %q message %+v, want a closed pull message", msg.Event, f) + } + if len(calls) != 0 { + t.Errorf("the tool was called for a merge into another branch: %v", calls) + } +} + +func TestTreesRefusedWritesNothingAndSaysWhy(t *testing.T) { // 1236/AC3 (the conversion; the delay is in main_test) + var calls []call + msg, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "", nil, errors.New("the listing for novox/mesh-delivery was cut"))) + if !errors.Is(err, ErrRetry) || msg.Data != nil { + t.Fatalf("err = %v data = %s, want ErrRetry and no message", err, msg.Data) + } + for _, w := range []string{"novox/mesh-delivery", merge, "the listing for novox/mesh-delivery was cut"} { + if !strings.Contains(err.Error(), w) { + t.Errorf("error %q does not name %q", err, w) + } + } +} + +func TestNoModuleJSONAnywhereWritesNoMergedMessage(t *testing.T) { // 1236 (notes: a repository with no module.json) + var calls []call + msg, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{}, nil)) + if !errors.Is(err, ErrNoModules) || msg.Data != nil { + t.Errorf("err = %v data = %s", err, msg.Data) + } +} + +func TestMergedWithoutAMergeCommitIsUnreadable(t *testing.T) { // 1236 + var calls []call + _, err := Convert(mergedSubject, mergedBody(map[string]any{"merge_commit_sha": "abc"}), treesOf(&calls, "r0", map[string]string{"a": "t"}, nil)) + if !errors.Is(err, ErrUnreadable) || len(calls) != 0 { + t.Errorf("err = %v calls = %v", err, calls) + } +}