broad-announce/internal/grpcserver/server.go

package grpcserver

import (
	"context"
	"fmt"
	"net"
	"time"

	pbv1 "git3.techno-world.net/lrosales/broad-announce/gen/go/broadannounce/v1"
	"git3.techno-world.net/lrosales/broad-announce/internal/observability"
	"git3.techno-world.net/lrosales/broad-announce/internal/pipeline"
	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
)

// Config for the gRPC server.
type Config struct {
	Addr        string        // default ":9090"
	MaxInflight int           // max in-flight per stream; default 256
}

// IngestdDeps is what ingestd's main.go passes to NewForIngestd.
// Using an interface avoids an import cycle (internal/grpcserver cannot import cmd/ingestd).
type IngestdDeps interface {
	GetLogger() interface{ Info(msg string, args ...any) }
	GetMetrics() *observability.IngestdMetrics
	GetPipeline() pipeline.Deps
}

// NewForIngestd is the constructor ingestd's main.go calls.
// It registers the Ingest service on a grpc.Server but does not start listening.
func NewForIngestd(cfg Config, deps IngestdDeps) (*grpc.Server, error) {
	if cfg.Addr == "" {
		cfg.Addr = ":9090"
	}
	if cfg.MaxInflight == 0 {
		cfg.MaxInflight = 256
	}

	logger := deps.GetLogger()
	m := deps.GetMetrics()

	// Prime the histogram buckets with a zero observation so Prometheus
	// registers them immediately (avoids missing bucket labels on first scrape).
	m.GRPCAckLatency.WithLabelValues("_init").Observe(0)

	ka := keepalive.ServerParameters{
		Time:    30 * time.Second,
		Timeout: 10 * time.Second,
	}
	enforcement := keepalive.EnforcementPolicy{
		MinTime:             5 * time.Second,
		PermitWithoutStream: false,
	}

	grpcOpts := []grpc.ServerOption{
		grpc.KeepaliveParams(ka),
		grpc.KeepaliveEnforcementPolicy(enforcement),
		grpc.WriteBufferSize(256 * 1024),
		grpc.ReadBufferSize(32 * 1024),
		// TODO(m11): grpc.Creds(tlsCredentials()) — enable TLS once certificates are available
	}

	srv := grpc.NewServer(grpcOpts...)

	pbv1.RegisterIngestServer(srv, &Server{
		Dep:         deps.GetPipeline(),
		MaxInflight: cfg.MaxInflight,
		Logger:      logger,
	})

	logger.Info("grpc server configured", "addr", cfg.Addr, "max_inflight", cfg.MaxInflight)
	return srv, nil
}

// Serve starts a TCP listener on addr and blocks serving.
// On ctx cancellation, it initiates a graceful shutdown (drains for up to 15s
// then hard-stops). The caller should run Serve in a goroutine.
func Serve(ctx context.Context, cfg Config, deps IngestdDeps) error {
	srv, err := NewForIngestd(cfg, deps)
	if err != nil {
		return fmt.Errorf("grpcserver.NewForIngestd: %w", err)
	}

	ln, err := net.Listen("tcp", cfg.Addr)
	if err != nil {
		return fmt.Errorf("grpc listen %s: %w", cfg.Addr, err)
	}

	errCh := make(chan error, 1)
	go func() { errCh <- srv.Serve(ln) }()

	select {
	case <-ctx.Done():
		done := make(chan struct{})
		go func() {
			srv.GracefulStop()
			close(done)
		}()
		select {
		case <-done:
			return nil
		case <-time.After(15 * time.Second):
			srv.Stop()
			return ctx.Err()
		}
	case err := <-errCh:
		return err
	}
}