fix: code review improvements across delivery, usecase and consumer l…

cmd/aggregator/main.go

@@ -7,19 +7,11 @@ import (
 	"github.com/goriiin/kotyari-bots_backend/pkg/config"
 )
 
-// const local = "local-config"
-const docker = "docker-config"
-
 func main() {
-	cfg, _ := config.New[aggregator.AggregatorAppConfig]()
-	config.WatchConfig(func() {
-		newCfg, err := config.NewWithConfig[aggregator.AggregatorAppConfig](docker)
-		if err != nil {
-			return
-		}
-
-		cfg = newCfg
-	})
+	cfg, err := config.New[aggregator.AggregatorAppConfig]()
+	if err != nil {
+		log.Fatalf("config load: %v", err)
+	}
 
 	app, err := aggregator.NewAggregatorApp(cfg)
 	if err != nil {

cmd/posts_command_consumer/main.go

@@ -2,6 +2,9 @@ package main
 
 import (
 	"log"
+	"os"
+	"os/signal"
+	"syscall"
 
 	"github.com/goriiin/kotyari-bots_backend/internal/apps/posts_command_consumer"
 	"github.com/goriiin/kotyari-bots_backend/pkg/config"
@@ -23,14 +26,27 @@ func main() {
 		log.Fatal(err)
 	}
 
-	defer func(app *posts_command_consumer.PostsCommandConsumer) {
-		err := app.Close()
+	// Run in the background so we can react to OS signals and close resources
+	// gracefully instead of being killed with the DB pool / Kafka reader open.
+	runErr := make(chan error, 1)
+	go func() {
+		runErr <- app.Run()
+	}()
+
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
+
+	select {
+	case err := <-runErr:
 		if err != nil {
-			log.Fatal(err)
+			log.Println("run error:", err)
 		}
-	}(app)
+	case s := <-sig:
+		log.Printf("received signal %s, shutting down", s)
+	}
 
-	if err = app.Run(); err != nil {
-		log.Println(err)
+	if err := app.Close(); err != nil {
+		// Log, don't Fatal: os.Exit here would skip any remaining cleanup.
+		log.Println("close error:", err)
 	}
 }

cmd/posts_command_producer/main.go

@@ -2,6 +2,9 @@ package main
 
 import (
 	"log"
+	"os"
+	"os/signal"
+	"syscall"
 
 	"github.com/goriiin/kotyari-bots_backend/internal/apps/posts_command_producer"
 	"github.com/goriiin/kotyari-bots_backend/pkg/config"
@@ -18,14 +21,25 @@ func main() {
 		log.Fatal(err)
 	}
 
-	defer func(app *posts_command_producer.PostsCommandProducerApp) {
-		err := app.Close()
+	runErr := make(chan error, 1)
+	go func() {
+		runErr <- app.Run()
+	}()
+
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
+
+	select {
+	case err := <-runErr:
 		if err != nil {
-			log.Fatal(err)
+			log.Println("run error:", err)
 		}
-	}(app)
+	case s := <-sig:
+		log.Printf("received signal %s, shutting down", s)
+	}
 
-	if err = app.Run(); err != nil {
-		log.Println(err)
+	if err := app.Close(); err != nil {
+		// Log, don't Fatal: os.Exit here would skip any remaining cleanup.
+		log.Println("close error:", err)
 	}
 }

cmd/reddit/main.go

@@ -7,22 +7,11 @@ import (
 	"github.com/goriiin/kotyari-bots_backend/pkg/config"
 )
 
-// TODO: change behaviour
-// const local = "local-config"
-const docker = "docker-config"
-
 func main() {
-	cfg, _ := config.New[redditapp.RedditAppConfig]()
-
-	config.WatchConfig(func() {
-		newCfg, err := config.NewWithConfig[redditapp.RedditAppConfig](docker)
-		if err != nil {
-			log.Fatalf("error parsing config in runtime: %s", err.Error())
-			return
-		}
-
-		cfg = newCfg
-	})
+	cfg, err := config.New[redditapp.RedditAppConfig]()
+	if err != nil {
+		log.Fatalf("config load: %v", err)
+	}
 
 	app, err := redditapp.NewRedditAPIApp(cfg)
 	if err != nil {

internal/adapters/auth/client.go

@@ -3,6 +3,7 @@ package auth
 import (
 	"context"
 	"errors"
+	"fmt"
 	"time"
 
 	"github.com/google/uuid"
@@ -13,18 +14,37 @@ import (
 )
 
 type Config struct {
+	// Host/Port are how the gRPC endpoint is configured in the YAML
+	// (auth_grpc.host / auth_grpc.port). Addr, when set, takes precedence and
+	// is used verbatim as the dial target.
+	Host    string        `mapstructure:"host"`
+	Port    int           `mapstructure:"port"`
 	Addr    string        `mapstructure:"addr"`
 	Timeout time.Duration `mapstructure:"timeout"`
 }
 
+// dialAddr returns the gRPC dial target, preferring an explicit Addr and
+// falling back to host:port.
+func (c Config) dialAddr() string {
+	if c.Addr != "" {
+		return c.Addr
+	}
+	return fmt.Sprintf("%s:%d", c.Host, c.Port)
+}
+
 type Client struct {
 	grpcClient authgen.UsersProviderClient
 	log        *logger.Logger
 	timeout    time.Duration
 }
 
 func NewClient(cfg Config, log *logger.Logger) (*Client, error) {
-	conn, err := grpc.NewClient(cfg.Addr,
+	addr := cfg.dialAddr()
+	if addr == ":0" || addr == "" {
+		return nil, errors.New("auth: gRPC address is not configured (set auth_grpc.host/port or auth_grpc.addr)")
+	}
+
+	conn, err := grpc.NewClient(addr,
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
 	)
 	if err != nil {

internal/adapters/auth/config_test.go

@@ -0,0 +1,35 @@
+package auth
+
+import "testing"
+
+func TestConfigDialAddr(t *testing.T) {
+	cases := []struct {
+		name string
+		cfg  Config
+		want string
+	}{
+		{
+			name: "host and port",
+			cfg:  Config{Host: "auth_rs", Port: 3001},
+			want: "auth_rs:3001",
+		},
+		{
+			name: "explicit addr wins over host/port",
+			cfg:  Config{Host: "auth_rs", Port: 3001, Addr: "override:9999"},
+			want: "override:9999",
+		},
+		{
+			name: "addr only",
+			cfg:  Config{Addr: "localhost:50051"},
+			want: "localhost:50051",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := tc.cfg.dialAddr(); got != tc.want {
+				t.Fatalf("dialAddr() = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}

internal/apps/posts_query/run.go

@@ -13,27 +13,11 @@ import (
 )
 
 func (p *PostsQueryApp) Run() error {
-	// Re-initializing auth client here or extracting from App struct would be cleaner,
-	// assuming we can modify NewPostsQueryApp to store authClient or create it here.
-	// For simplicity based on previous pattern, creating a new one or assuming it's available.
-
-	// Better approach: Pass it via struct (requires modifying init.go return struct)
-	// Since I can modify files fully:
-
-	l := log.Default() // using std log for run errors as per original
-
-	// Create auth client again or pass it. Let's create it to stick to the pattern of separation
-	// but normally it should be in struct.
-	// NOTE: Please update struct in init.go to store authClient if you want reuse.
-	// Implementing creation here for strict compliance with "Run" interface.
-
-	authClient, err := auth.NewClient(p.config.Auth, nil) // logger nil might be issue, better pass it
-	if err != nil {
-		return err
-	}
-
-	if err := p.startHTTPServer(p.handler, authClient); err != nil {
-		l.Printf("Error happened starting server %v", err)
+	// Reuse the auth client built (with a real logger) in NewPostsQueryApp instead
+	// of constructing a second one with a nil logger, which panicked on the first
+	// auth failure inside VerifySession.
+	if err := p.startHTTPServer(p.handler, p.authClient); err != nil {
+		log.Printf("Error happened starting server %v", err)
 		return err
 	}
 	return nil

internal/delivery_http/api_integrations/reddit/perform_request.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"io"
 	"net/http"
-	"sync"
 	"time"
 
 	"github.com/go-faster/errors"
@@ -19,11 +18,14 @@ const (
 )
 
 func (r *RedditAPIDelivery) performRequests() (chan PostData, error) {
+	// NOTE: ownership of cancel is handed to the draining goroutine below — we
+	// must NOT defer-cancel here, or the context would be cancelled the moment
+	// this function returns, killing the goroutines that still feed `posts`.
 	ctx, cancel := context.WithTimeout(context.Background(), defaultErrGroupWaitTime)
-	defer cancel()
 
 	integrations, err := r.integration.GetIntegrations(ctx, redditAPIString)
 	if err != nil {
+		cancel()
 		return nil, err
 	}
 
@@ -32,7 +34,7 @@ func (r *RedditAPIDelivery) performRequests() (chan PostData, error) {
 
 	for _, integration := range integrations {
 		g.Go(func() error {
-			req, err := http.NewRequest(http.MethodGet, integration.Url, http.NoBody)
+			req, err := http.NewRequestWithContext(ctx, http.MethodGet, integration.Url, http.NoBody)
 			if err != nil {
 				return errors.Wrap(err, "failed to create request")
 			}
@@ -41,6 +43,9 @@ func (r *RedditAPIDelivery) performRequests() (chan PostData, error) {
 			if err != nil {
 				return errors.Wrap(err, "failed to perform request")
 			}
+			// Always close the body, including on the Forbidden early-return
+			// below (previously leaked the connection on every blocked request).
+			defer func() { _ = resp.Body.Close() }()
 
 			if resp.StatusCode == http.StatusForbidden {
 				return errors.New("request was blocked")
@@ -51,16 +56,18 @@ func (r *RedditAPIDelivery) performRequests() (chan PostData, error) {
 				return errors.Wrapf(err, "bad response body: %s", string(body))
 			}
 
-			err = resp.Body.Close()
-			if err != nil {
-				return errors.Wrap(err, "failed to close body")
-			}
-
 			var redditAPIResponse RedditAPIResponse
 			if err := json.Unmarshal(body, &redditAPIResponse); err != nil {
 				return errors.Wrapf(err, "failed to unmarhsal: %s", integration.Url)
 			}
-			redditAPIResponses <- redditAPIResponse
+
+			// Respect cancellation so a stalled consumer can't wedge this
+			// goroutine forever on an unbuffered send.
+			select {
+			case redditAPIResponses <- redditAPIResponse:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
 
 			return nil
 		})
@@ -76,17 +83,18 @@ func (r *RedditAPIDelivery) performRequests() (chan PostData, error) {
 
 	posts := make(chan PostData)
 
-	var wg sync.WaitGroup
 	go func() {
+		defer cancel()
+		defer close(posts)
 		for redditNews := range redditAPIResponses {
-			wg.Add(1)
 			for _, post := range redditNews.Data.Posts {
-				posts <- post.PostData
+				select {
+				case posts <- post.PostData:
+				case <-ctx.Done():
+					return
+				}
 			}
-			wg.Done()
 		}
-		wg.Wait()
-		close(posts)
 	}()
 
 	return posts, nil

internal/delivery_http/bots/create.go

@@ -3,9 +3,11 @@ package bots
 import (
 	"context"
 
+	"github.com/go-faster/errors"
 	"github.com/google/uuid"
 	gen "github.com/goriiin/kotyari-bots_backend/internal/gen/bots"
 	"github.com/goriiin/kotyari-bots_backend/internal/model"
+	"github.com/goriiin/kotyari-bots_backend/pkg/constants"
 )
 
 func (h *Handler) CreateBot(ctx context.Context, req *gen.BotInput) (gen.CreateBotRes, error) {
@@ -24,21 +26,25 @@ func (h *Handler) CreateBot(ctx context.Context, req *gen.BotInput) (gen.CreateB
 		profiles = append(profiles, p.ID)
 	}
 
-	created, err := h.u.Create(ctx, model.Bot{
+	bot, profs, err := h.u.CreateWithProfiles(ctx, model.Bot{
 		Name:               req.Name,
 		SystemPrompt:       desc,
 		ModerationRequired: moderation,
 		ProfileIDs:         profiles,
 	})
 	if err != nil {
+		if errors.Is(err, constants.ErrValidation) {
+			return &gen.CreateBotBadRequest{
+				ErrorCode: constants.ValidationMsg,
+				Message:   err.Error(),
+			}, nil
+		}
 		h.log.Error(err, true, "CreateBot: create bot")
-		return nil, err
+		return &gen.CreateBotInternalServerError{
+			ErrorCode: constants.InternalMsg,
+			Message:   constants.InternalMsg,
+		}, nil
 	}
 
-	bot, profs, err := h.u.GetWithProfiles(ctx, created.ID)
-	if err != nil {
-		h.log.Error(err, true, "CreateBot: get with profiles")
-		return nil, err
-	}
 	return modelToDTO(&bot, profs), nil
 }