mesh-delivery/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go

// Command mesh-delivery-forge-bridge is delivery's forge bridge (novox/hq ADR 0437 §4, issue 1235): it hears the forge
// module's pull request events and writes each as a pull message on its stream FORGE. It runs beside today's
// mesh-delivery and merges nothing. It serves no request subject and holds no timer: every reaction is an event
// (ADR 0437 §8); a refused act is tried again only on the next event touching its delivery.
package main

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"git.novox.be/novox/mesh-sdk/go/process"

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

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

	"mesh-delivery-forge-bridge/internal/bridge"
)

// eventsStream is the shared stream every module's events are on; the module's consumer on it is the durable
// <node>_<module> the controller makes (mesh-control internal/broker/nats.go consumerStream, consumerDurable).
const eventsStream = "EVENTS"

// The bridge's own stream FORGE takes `mesh.stream.<module>.event.<event>` (mesh-control internal/broker
// modstreams.go ModuleStreamSubject).
func forgeSubject(event string) string {
	return process.ModuleStreamSubject(delivery.ForgeBridgeModule, event)
}

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	if err := run(ctx, os.Getenv, os.Stderr); err != nil {
		fmt.Fprintln(os.Stderr, "mesh-delivery-forge-bridge:", err)
		os.Exit(1)
	}
}

// delivered is the part of a JetStream message the bridge uses.
type delivered interface {
	Subject() string
	Data() []byte
	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)
}

// 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, trees bridge.Trees, files bridge.Files, kept bridge.Kept, logger *log.Logger) {
	msg, err := bridge.ConvertKept(m.Subject(), m.Data(), trees, files, kept)
	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()
		return
	case err != nil:
		logger.Printf("%v; the event is dropped", err)
		_ = m.Term()
		return
	}
	var opts []jetstream.PublishOpt
	// The forge's event id makes a redelivery after a lost acknowledgement the same message to the stream.
	if id := m.Headers().Get("x-event-id"); id != "" {
		opts = append(opts, jetstream.WithMsgID(id))
	}
	if _, err := pub.Publish(ctx, forgeSubject(msg.Event), msg.Data, opts...); err != nil {
		logger.Printf("%s: %s could not be written on FORGE, left for redelivery: %v", m.Subject(), msg.Event, err)
		return
	}
	if msg.Said != "" {
		logger.Printf("%s: %s", m.Subject(), msg.Said)
	}
	_ = m.Ack()
}

// run is the program: it returns when ctx ends or the bus cannot be used.
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
	logger := log.New(logOut, "mesh-delivery-forge-bridge: ", log.LstdFlags)
	cred, err := process.ReadBrokerAccount(getenv)
	if err != nil {
		return err
	}
	again := make(chan struct{}, 1)
	conn, js, err := cred.Dial(
		nats.ReconnectHandler(func(*nats.Conn) {
			select {
			case again <- struct{}{}:
			default:
			}
		}),
	)
	if err != nil {
		return err
	}
	defer conn.Close()
	// What the forge holds open, said before any event is heard, and again at each reconnect (ADR 0429 §13).
	call := callerOn(conn, cred.User)
	// No forge event is heard while the open pull requests are read and said (ADR 0443 §5): events and syncs take turns.
	var turn sync.Mutex
	kv, err := process.EssentialBucket(ctx, js, "bridge", headsBucket, "keep the heads it wrote on the forge stream")
	if err != nil {
		return err
	}
	heads := kvHeads{kv}
	// Everything written on FORGE, an event's and a sync's, keeps the head in the bucket.
	rec := &recorder{pub: js, heads: heads, logger: logger}
	said := func(ctx context.Context) error { return syncOpen(ctx, call, rec, heads, &turn, logger) }
	// What the gate asks of the forge (ADR 0437 §8): its owed acts and its merges, each performed once per revision.
	performedKV, err := process.EssentialBucket(ctx, js, "bridge", performedBucket, "perform an act of the gate only once")
	if err != nil {
		return err
	}
	perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger}
	watching := make(chan error, 3)
	for _, w := range []struct {
		bucket string
		react  func(ctx context.Context, key string, rev uint64, value []byte)
	}{{delivery.OwedBucket, perf.owed}, {delivery.MergesBucket, perf.merges}} {
		// The gate's buckets come with the gate and its grant; the bridge does not wait for them (no timer), and is started
		// again by its delivery when the gate arrives. A bucket refused or not answered is the same (issue 1945).
		kv := process.OptionalBucket(ctx, js, w.bucket, "nothing of it is performed until the bridge starts again", logger.Printf)
		if kv == nil {
			continue
		}
		go func() {
			watching <- watch(ctx, kv, func(key string, rev uint64, value []byte) {
				hctx, cancel := context.WithTimeout(ctx, syncDeadline)
				defer cancel()
				w.react(hctx, key, rev, value)
			})
		}()
	}
	if err := said(ctx); err != nil {
		return err
	}
	go onReconnect(ctx, again, said, logger)
	// What a merge read of each module directory, by its tree (issue 1126): a merge reads only the directories it changed.
	inputs := newKept(process.OptionalBucket(ctx, js, inputsBucket, "a merge's module directories are kept in memory only, so the first merge after a start reads every one", logger.Printf), logger)
	cons, err := cred.EventConsumer(ctx, js, eventsStream)
	if err != nil {
		// Essential: the forge's events are what the bridge carries.
		return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the bridge hears no forge event without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, eventsStream, err)
	}
	cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
		hear(&turn, func() {
			// Bounded: the turn is held, and no sync or other event goes on without it.
			hctx, cancel := context.WithTimeout(ctx, syncDeadline)
			defer cancel()
			handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), inputs, logger)
		})
	}, logger.Printf)
	if err != nil {
		return err
	}
	defer cc.Stop()
	select {
	case <-ctx.Done():
		return nil
	case err := <-watching:
		return fmt.Errorf("watching the gate's keys: %w", err)
	}
}