chat/internal/ircserver/server.go

package ircserver

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

	"git.eeqj.de/sneak/neoirc/internal/broker"
	"git.eeqj.de/sneak/neoirc/internal/config"
	"git.eeqj.de/sneak/neoirc/internal/db"
	"git.eeqj.de/sneak/neoirc/internal/logger"
	"git.eeqj.de/sneak/neoirc/internal/service"
	"go.uber.org/fx"
)

// Params defines the dependencies for creating an IRC
// Server.
type Params struct {
	fx.In

	Logger   *logger.Logger
	Config   *config.Config
	Database *db.Database
	Broker   *broker.Broker
	Service  *service.Service
}

// Server is the TCP IRC protocol server.
type Server struct {
	log      *slog.Logger
	cfg      *config.Config
	database *db.Database
	brk      *broker.Broker
	svc      *service.Service
	listener net.Listener
	mu       sync.Mutex
	conns    map[*Conn]struct{}
	cancel   context.CancelFunc
}

// New creates a new IRC Server and registers its lifecycle
// hooks.  The listener is only started if IRC_LISTEN_ADDR
// is configured; otherwise the server is inert.
func New(
	lifecycle fx.Lifecycle,
	params Params,
) *Server {
	srv := &Server{
		log:      params.Logger.Get(),
		cfg:      params.Config,
		database: params.Database,
		brk:      params.Broker,
		svc:      params.Service,
		conns:    make(map[*Conn]struct{}),
		listener: nil,
		cancel:   nil,
		mu:       sync.Mutex{},
	}

	listenAddr := params.Config.IRCListenAddr
	if listenAddr == "" {
		return srv
	}

	lifecycle.Append(fx.Hook{
		OnStart: func(ctx context.Context) error {
			return srv.start(ctx, listenAddr)
		},
		OnStop: func(_ context.Context) error {
			srv.stop()

			return nil
		},
	})

	return srv
}

// start begins listening for TCP connections.
//
//nolint:contextcheck // long-lived server ctx, not the short Fx one
func (s *Server) start(_ context.Context, addr string) error {
	ln, err := net.Listen("tcp", addr)
	if err != nil {
		return fmt.Errorf("irc listen: %w", err)
	}

	s.listener = ln

	ctx, cancel := context.WithCancel(context.Background())
	s.cancel = cancel

	s.log.Info(
		"irc server listening", "addr", addr,
	)

	go s.acceptLoop(ctx)

	return nil
}

// stop shuts down the listener and all connections.
func (s *Server) stop() {
	if s.cancel != nil {
		s.cancel()
	}

	if s.listener != nil {
		s.listener.Close() //nolint:errcheck,gosec
	}

	s.mu.Lock()
	for c := range s.conns {
		c.conn.Close() //nolint:errcheck,gosec
	}
	s.mu.Unlock()
}

// acceptLoop accepts new connections.
func (s *Server) acceptLoop(ctx context.Context) {
	for {
		tcpConn, err := s.listener.Accept()
		if err != nil {
			select {
			case <-ctx.Done():
				return
			default:
				s.log.Error(
					"irc accept error", "error", err,
				)

				continue
			}
		}

		client := newConn(
			ctx, tcpConn, s.log,
			s.database, s.brk, s.cfg, s.svc,
		)

		s.mu.Lock()
		s.conns[client] = struct{}{}
		s.mu.Unlock()

		go func() {
			defer func() {
				s.mu.Lock()
				delete(s.conns, client)
				s.mu.Unlock()
			}()

			client.serve(ctx)
		}()
	}
}