nomadcode/services/core/internal/gitosync/runner.go

package gitosync

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"os/exec"
	"strings"
	"sync"
	"time"

	"nhooyr.io/websocket"

	toki "git.toki-labs.com/toki/proto-socket/go"
	"github.com/nomadcode/nomadcode-core/internal/gitoevents"
	"github.com/nomadcode/nomadcode-core/internal/protosocket"
	"google.golang.org/protobuf/types/known/structpb"
)

// ExecCommandRunner is the production CommandRunner backed by os/exec. It runs
// git in the given directory and returns stdout with the trailing newline left
// intact for the caller to trim.
type ExecCommandRunner struct{}

func (ExecCommandRunner) Run(ctx context.Context, dir, name string, args ...string) (string, error) {
	cmd := exec.CommandContext(ctx, name, args...)
	cmd.Dir = dir
	out, err := cmd.Output()
	if err != nil {
		return "", err
	}
	return string(out), nil
}

// protoSocketTransport is the production gitoevents.Transport that dials the Gito
// proto-socket endpoint over a websocket and exchanges binary protobuf envelopes.
// It is kept unexported; callers wire it through NewRunner.
type protoSocketTransport struct {
	url     string
	conn    *websocket.Conn
	client  *toki.WsClient
	inbound chan struct {
		env protosocket.Envelope
		err error
	}
	closeOnce  sync.Once
	closed     chan struct{}
	inboundCap int
	logger     *slog.Logger
}

func (t *protoSocketTransport) Connect(ctx context.Context) error {
	conn, _, err := websocket.Dial(ctx, t.url, nil)
	if err != nil {
		return err
	}
	t.conn = conn
	t.closed = make(chan struct{})
	t.closeOnce = sync.Once{}

	cap := t.inboundCap
	if cap <= 0 {
		cap = 100
	}

	t.client = toki.NewWsClient(conn, 0, 0, protosocket.ParserMap())
	t.inbound = make(chan struct {
		env protosocket.Envelope
		err error
	}, cap)

	toki.AddListenerTyped[*structpb.Struct](&t.client.Communicator, func(msg *structpb.Struct) {
		env, err := protosocket.EnvelopeFromStruct(msg)
		if err != nil {
			if t.logger != nil {
				t.logger.Error("gito proto-socket envelope decode failed, dropping frame", "error", err)
			}
			return
		}
		t.enqueue(struct {
			env protosocket.Envelope
			err error
		}{env: env, err: err})
	})

	t.client.AddDisconnectListener(func(c *toki.WsClient) {
		t.enqueue(struct {
			env protosocket.Envelope
			err error
		}{err: io.EOF})
	})

	return nil
}

func (t *protoSocketTransport) enqueue(item struct {
	env protosocket.Envelope
	err error
}) {
	select {
	case <-t.closed:
	case t.inbound <- item:
	}
}

func (t *protoSocketTransport) Write(ctx context.Context, env protosocket.Envelope) error {
	if t.client == nil {
		return errors.New("gitosync: transport not connected")
	}
	msg, err := env.ToStruct()
	if err != nil {
		return err
	}
	return t.client.Communicator.Send(msg)
}

func (t *protoSocketTransport) Read(ctx context.Context) (protosocket.Envelope, error) {
	if t.client == nil {
		return protosocket.Envelope{}, errors.New("gitosync: transport not connected")
	}
	select {
	case <-ctx.Done():
		return protosocket.Envelope{}, ctx.Err()
	case <-t.closed:
		return protosocket.Envelope{}, io.EOF
	case item, ok := <-t.inbound:
		if !ok {
			return protosocket.Envelope{}, io.EOF
		}
		if item.err != nil {
			return protosocket.Envelope{}, item.err
		}
		return item.env, nil
	}
}

func (t *protoSocketTransport) Close() error {
	var err error
	t.closeOnce.Do(func() {
		if t.closed != nil {
			close(t.closed)
		}
		if t.client != nil {
			err = t.client.Close()
		}
	})
	return err
}

// NewRunner builds the gitoevents client wired to forward on-target branch
// events into the bridge. It returns a *gitoevents.Client whose Run blocks
// until the context is cancelled or the stream ends; the caller owns starting
// it in a goroutine and cancelling the context on shutdown. The bridge's Handle
// error is logged (not propagated) so a single failed event does not tear down
// the long-lived subscription.
func NewRunner(url, repoID, branch string, bridge *Bridge, logger *slog.Logger) (*gitoevents.Client, error) {
	if strings.TrimSpace(url) == "" {
		return nil, errors.New("gitosync: proto-socket url is required")
	}
	if bridge == nil {
		return nil, errors.New("gitosync: bridge is nil")
	}
	transport := &protoSocketTransport{url: url, logger: logger}
	handler := func(ctx context.Context, ev gitoevents.BranchUpdatedEvent) {
		if err := bridge.Handle(ctx, ev); err != nil {
			if logger != nil {
				logger.Error("gito creation sync handle failed",
					"error", err, "repo_id", ev.RepoID, "after", ev.After)
			}
		}
	}
	return gitoevents.NewClient(transport, gitoevents.Options{
		RepoID:  repoID,
		Branch:  branch,
		Handler: handler,
		Logger:  logger,
	})
}

// ReconnectingRunner wraps a *gitoevents.Client to provide retry/reconnect logic
// with exponential backoff when the runner encounters network or connection drop errors.
type ReconnectingRunner struct {
	client     *gitoevents.Client
	logger     *slog.Logger
	MaxRetries int
	BaseDelay  time.Duration
	MaxDelay   time.Duration
}

// NewReconnectingRunner wraps client in a ReconnectingRunner.
func NewReconnectingRunner(client *gitoevents.Client, logger *slog.Logger) *ReconnectingRunner {
	return &ReconnectingRunner{
		client:     client,
		logger:     logger,
		MaxRetries: 5,
		BaseDelay:  100 * time.Millisecond,
		MaxDelay:   5 * time.Second,
	}
}

// Run executes the client's Run method. If Run returns a connection-related error,
// it reconnects using exponential backoff up to MaxRetries. It exits cleanly
// returning nil if the context is cancelled.
func (r *ReconnectingRunner) Run(ctx context.Context) error {
	var retries int
	delay := r.BaseDelay

	for {
		err := r.client.Run(ctx)
		if err == nil {
			return nil
		}

		if ctx.Err() != nil {
			return nil
		}

		retries++
		if r.MaxRetries > 0 && retries > r.MaxRetries {
			if r.logger != nil {
				r.logger.Error("gito branch event runner: max reconnect retries reached", "error", err)
			}
			return err
		}

		if r.logger != nil {
			r.logger.Warn("gito branch event runner disconnected, reconnecting...",
				"error", err,
				"retry", retries,
				"delay", delay,
			)
		}

		select {
		case <-ctx.Done():
			return nil
		case <-time.After(delay):
		}

		delay = delay * 2
		if delay > r.MaxDelay {
			delay = r.MaxDelay
		}
	}
}