Artifact [f8895d7f52]

Artifact f8895d7f529ee7e037c4424fa5d2991281f6802639985ab5df0b66a84f82a8ff:

• File leaf/agent/serve_nats.go — part of check-in [38efed7a12] at 2026-08-12 20:50:57 on branch main — chore(deps): migrate to go-libfossil v0.9.0 (#195)

  * chore(deps): migrate to go-libfossil v0.9.0 (renamed module path)

  - import path danmestas/libfossil → danmestas/go-libfossil everywhere - CloneOpts.ProjectCode removed upstream; assert via CloneResult instead - Timeline now takes TimelineOpts (cursor pagination, newest-first) - testutil.HasFossil removed; check FossilBinary() != ""

  * ci: retrigger checks

  ---------

  Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> (user: [email protected], size: 1929) [annotate] [blame] [check-ins using]

package agent

import (
	"context"
	"fmt"
	"log/slog"

	libfossil "github.com/danmestas/go-libfossil"
	"github.com/nats-io/nats.go"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
)

// ServeNATS subscribes to the given subject and dispatches incoming
// xfer requests to the repo's HandleSync. Blocks until ctx is cancelled.
func ServeNATS(ctx context.Context, nc *nats.Conn, subject string, r *libfossil.Repo) error {
	if nc == nil {
		panic("agent.ServeNATS: nc must not be nil")
	}
	if subject == "" {
		panic("agent.ServeNATS: subject must not be empty")
	}
	if r == nil {
		panic("agent.ServeNATS: r must not be nil")
	}

	tracer := otel.Tracer("edgesync-leaf")

	sub, err := nc.Subscribe(subject, func(msg *nats.Msg) {
		// Extract trace context from incoming NATS headers for cross-service linking.
		reqCtx := extractTraceContext(ctx, msg)

		reqCtx, span := tracer.Start(reqCtx, "nats.handle",
			trace.WithSpanKind(trace.SpanKindServer),
			trace.WithAttributes(
				attribute.String("messaging.system", "nats"),
				attribute.String("messaging.destination.name", subject),
				attribute.Int("messaging.message.body.size", len(msg.Data)),
			),
		)
		defer span.End()

		respBytes, err := r.HandleSync(reqCtx, msg.Data)
		if err != nil {
			span.RecordError(err)
			slog.ErrorContext(reqCtx, "serve-nats: handler error", "error", err)
			// Respond with empty bytes on error so the client doesn't hang.
			msg.Respond([]byte{})
			return
		}

		if err := msg.Respond(respBytes); err != nil {
			span.RecordError(err)
			slog.ErrorContext(reqCtx, "serve-nats: respond error", "error", err)
		}
		slog.DebugContext(reqCtx, "serve-nats: request handled",
			"request_bytes", len(msg.Data),
			"response_bytes", len(respBytes),
		)
	})
	if err != nil {
		return fmt.Errorf("agent.ServeNATS: subscribe %s: %w", subject, err)
	}

	<-ctx.Done()
	return sub.Unsubscribe()
}