goplt/cmd/audit-service/main.go

// Package main provides the entry point for the Audit Service.
package main

import (
	"context"
	"fmt"
	"net"
	"os"
	"os/signal"
	"syscall"
	"time"

	"git.dcentral.systems/toolz/goplt/internal/di"
	healthpkg "git.dcentral.systems/toolz/goplt/internal/health"
	"git.dcentral.systems/toolz/goplt/internal/infra/database"
	"git.dcentral.systems/toolz/goplt/pkg/config"
	"git.dcentral.systems/toolz/goplt/pkg/logger"
	"git.dcentral.systems/toolz/goplt/pkg/registry"
	"go.uber.org/fx"
	"go.uber.org/zap"
	"google.golang.org/grpc"
)

// grpcServerWrapper wraps the gRPC server for lifecycle management.
type grpcServerWrapper struct {
	server   *grpc.Server
	listener net.Listener
	port     int
	logger   logger.Logger
}

func (s *grpcServerWrapper) Start() error {
	s.logger.Info("Starting Audit Service gRPC server",
		zap.Int("port", s.port),
		zap.String("addr", s.listener.Addr().String()),
	)

	errChan := make(chan error, 1)
	go func() {
		if err := s.server.Serve(s.listener); err != nil {
			errChan <- err
		}
	}()

	select {
	case err := <-errChan:
		return fmt.Errorf("gRPC server failed to start: %w", err)
	case <-time.After(100 * time.Millisecond):
		s.logger.Info("Audit Service gRPC server started successfully",
			zap.Int("port", s.port),
		)
		return nil
	}
}

func (s *grpcServerWrapper) Stop(ctx context.Context) error {
	s.logger.Info("Stopping Audit Service gRPC server")

	stopped := make(chan struct{})
	go func() {
		s.server.GracefulStop()
		close(stopped)
	}()

	select {
	case <-stopped:
		s.logger.Info("Audit Service gRPC server stopped gracefully")
		return nil
	case <-ctx.Done():
		s.logger.Warn("Audit Service gRPC server stop timeout, forcing stop")
		s.server.Stop()
		return ctx.Err()
	}
}

func (s *grpcServerWrapper) Port() int {
	return s.port
}

func main() {
	// Create DI container - services will be created via fx.Provide
	// Note: CoreModule() is automatically included by NewContainer()
	container := di.NewContainer(
		// Database for audit service (audit schema)
		fx.Provide(func(cfg config.ConfigProvider, log logger.Logger) (*database.Client, error) {
			dsn := cfg.GetString("database.dsn")
			if dsn == "" {
				return nil, fmt.Errorf("database.dsn is required")
			}
			client, err := database.NewClientWithSchema(dsn, "audit")
			if err != nil {
				return nil, err
			}

			// Run migrations
			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
			defer cancel()

			if err := client.Migrate(ctx); err != nil {
				log.Warn("Failed to run migrations",
					zap.Error(err),
				)
			} else {
				log.Info("Database migrations completed for audit service")
			}

			return client, nil
		}),

		// Register database health checker with existing health registry
		fx.Invoke(func(registry *healthpkg.Registry, db *database.Client) {
			registry.Register("database", healthpkg.NewDatabaseChecker(db))
		}),

		// Provide audit service and gRPC server (defined in audit_service_fx.go)
		provideAuditService(),

		// Lifecycle hooks
		fx.Invoke(registerLifecycle),
	)

	// Create root context
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Handle signals
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)

	// Start the application
	if err := container.Start(ctx); err != nil {
		log := logger.GetGlobalLogger()
		if log != nil {
			log.Error("Failed to start Audit Service",
				logger.Error(err),
			)
		} else {
			fmt.Fprintf(os.Stderr, "Failed to start Audit Service: %v\n", err)
		}
		os.Exit(1)
	}

	// Wait for interrupt signal
	<-sigChan
	fmt.Println("\nShutting down Audit Service...")

	// Create shutdown context with timeout
	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer shutdownCancel()

	// Stop the application
	if err := container.Stop(shutdownCtx); err != nil {
		log := logger.GetGlobalLogger()
		if log != nil {
			log.Error("Error during Audit Service shutdown",
				logger.Error(err),
			)
		} else {
			fmt.Fprintf(os.Stderr, "Error during shutdown: %v\n", err)
		}
		os.Exit(1)
	}

	fmt.Println("Audit Service stopped successfully")
}

// registerLifecycle registers lifecycle hooks for the service.
func registerLifecycle(
	lc fx.Lifecycle,
	grpcServer *grpcServerWrapper,
	serviceRegistry registry.ServiceRegistry,
	cfg config.ConfigProvider,
	log logger.Logger,
) {
	lc.Append(fx.Hook{
		OnStart: func(ctx context.Context) error {
			// Start gRPC server
			if err := grpcServer.Start(); err != nil {
				return fmt.Errorf("failed to start gRPC server: %w", err)
			}

			// Register with service registry
			serviceID := fmt.Sprintf("audit-service-%d", time.Now().Unix())
			host := cfg.GetString("services.audit.host")
			if host == "" {
				// In Docker, use service name for Consul to reach the service
				// For local development, use localhost
				if os.Getenv("ENVIRONMENT") == "production" || os.Getenv("DOCKER") == "true" {
					host = "audit-service" // Docker service name
				} else {
					host = "localhost"
				}
			}
			port := grpcServer.Port()

			instance := &registry.ServiceInstance{
				ID:      serviceID,
				Name:    "audit-service",
				Address: host,
				Port:    port,
				Tags:    []string{"grpc", "audit"},
				Metadata: map[string]string{
					"version":  "1.0.0",
					"protocol": "grpc",
				},
			}

			if err := serviceRegistry.Register(ctx, instance); err != nil {
				log.Warn("Failed to register with service registry",
					zap.Error(err),
				)
			} else {
				log.Info("Registered Audit Service with service registry",
					zap.String("service_id", serviceID),
					zap.String("name", instance.Name),
					zap.Int("port", port),
				)
			}

			return nil
		},
		OnStop: func(ctx context.Context) error {
			// Stop gRPC server
			if err := grpcServer.Stop(ctx); err != nil {
				return fmt.Errorf("failed to stop gRPC server: %w", err)
			}
			return nil
		},
	})
}