Wait for IRC listener cleanup and HTTP shutdown on stop (closes #159)

Stopping the server returned before the IRC listener's connections had
run their cleanup, and the database closed under them, so their
sessions, clients and channel memberships stayed and their nicks stayed
taken after a restart.

The IRC listener's stop now waits up to five seconds for every
connection's cleanup, which now runs on a context that stopping does
not cancel. The HTTP server's stop hook now runs Shutdown itself and
waits for it; httpServer is set before serving starts, and the server's
own signal watcher, which raced fx's stop hooks, is gone.

Model: opus-5-5

internal/db/queries.go

@@ -462,46 +462,207 @@ func (database *Database) JoinChannel(
 	return nil
 }
 
-// JoinChannelAsOperator adds a session to a channel with
+// Errors that CreateOrJoinChannel refuses a join with.
-// operator status. Used when a user creates a new channel.
+var (
-func (database *Database) JoinChannelAsOperator(
+	ErrBannedFromChannel = errors.New("banned from channel")
+	ErrInviteOnlyChannel = errors.New("channel is invite-only")
+	ErrBadChannelKey     = errors.New("wrong channel key")
+	ErrChannelIsFull     = errors.New("channel is full")
+)
+
+// CreateOrJoinChannel adds a session to the channel called
+// name, creating the channel if it does not exist. The
+// session is made operator if the channel has no members;
+// otherwise checkJoinRestrictions may refuse it. A join
+// uses up the session's invite to the channel. It returns
+// the channel ID and whether the session was made operator.
+// All of it is one transaction, and the database has one
+// connection, so every other statement waits until the
+// join is decided.
+func (database *Database) CreateOrJoinChannel(
 	ctx context.Context,
-	channelID, sessionID int64,
+	name string,
-) error {
+	sessionID int64,
-	_, err := database.conn.ExecContext(ctx,
+	suppliedKey string,
+) (int64, bool, error) {
+	transaction, err := database.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return 0, false, fmt.Errorf("begin join: %w", err)
+	}
+
+	defer func() { _ = transaction.Rollback() }()
+
+	now := time.Now()
+
+	_, err = transaction.ExecContext(ctx,
+		`INSERT OR IGNORE INTO channels
+		 (name, created_at, updated_at)
+		 VALUES (?, ?, ?)`,
+		name, now, now)
+	if err != nil {
+		return 0, false, fmt.Errorf("create channel: %w", err)
+	}
+
+	var channelID, memberCount int64
+
+	err = transaction.QueryRowContext(ctx,
+		`SELECT id,
+		   (SELECT COUNT(*) FROM channel_members
+		    WHERE channel_id = channels.id)
+		 FROM channels WHERE name = ?`,
+		name,
+	).Scan(&channelID, &memberCount)
+	if err != nil {
+		return 0, false, fmt.Errorf("get channel: %w", err)
+	}
+
+	isOperator := memberCount == 0
+	if !isOperator {
+		err = checkJoinRestrictions(
+			ctx, transaction, channelID, sessionID,
+			suppliedKey, memberCount,
+		)
+		if err != nil {
+			return 0, false, err
+		}
+	}
+
+	_, err = transaction.ExecContext(ctx,
 		`INSERT OR IGNORE INTO channel_members
 		 (channel_id, session_id, is_operator, joined_at)
-		 VALUES (?, ?, 1, ?)`,
+		 VALUES (?, ?, ?, ?)`,
-		channelID, sessionID, time.Now())
+		channelID, sessionID, isOperator, now)
 	if err != nil {
-		return fmt.Errorf(
+		return 0, false, fmt.Errorf("join channel: %w", err)
-			"join channel as operator: %w", err,
+	}
-		)
+
+	_, err = transaction.ExecContext(ctx,
+		`DELETE FROM channel_invites
+		 WHERE channel_id = ? AND session_id = ?`,
+		channelID, sessionID)
+	if err != nil {
+		return 0, false, fmt.Errorf("clear invite: %w", err)
+	}
+
+	err = transaction.Commit()
+	if err != nil {
+		return 0, false, fmt.Errorf("commit join: %w", err)
+	}
+
+	return channelID, isOperator, nil
+}
+
+// checkJoinRestrictions returns the error that refuses a
+// session's join of a channel with memberCount members:
+// ErrBannedFromChannel when a ban matches its hostmask,
+// ErrInviteOnlyChannel when the channel is +i and the
+// session has no invite, ErrBadChannelKey when suppliedKey
+// is not the +k key, or ErrChannelIsFull when the channel
+// has at least its +l number of members. It returns nil
+// when none applies.
+func checkJoinRestrictions(
+	ctx context.Context,
+	transaction *sql.Tx,
+	channelID, sessionID int64,
+	suppliedKey string,
+	memberCount int64,
+) error {
+	banned, err := isSessionBannedInTx(
+		ctx, transaction, channelID, sessionID,
+	)
+	if err != nil {
+		return err
+	}
+
+	if banned {
+		return ErrBannedFromChannel
+	}
+
+	var (
+		inviteOnly, invited bool
+		key                 string
+		userLimit           int64
+	)
+
+	err = transaction.QueryRowContext(ctx,
+		`SELECT is_invite_only, channel_key, user_limit,
+		   EXISTS (SELECT 1 FROM channel_invites
+		           WHERE channel_id = channels.id
+		             AND session_id = ?)
+		 FROM channels WHERE id = ?`,
+		sessionID, channelID,
+	).Scan(&inviteOnly, &key, &userLimit, &invited)
+	if err != nil {
+		return fmt.Errorf("get channel modes: %w", err)
+	}
+
+	switch {
+	case inviteOnly && !invited:
+		return ErrInviteOnlyChannel
+	case key != "" && suppliedKey != key:
+		return ErrBadChannelKey
+	case userLimit > 0 && memberCount >= userLimit:
+		return ErrChannelIsFull
 	}
 
 	return nil
 }
 
-// CountChannelMembers returns the number of members in
+// isSessionBannedInTx is IsSessionBanned, reading through
-// a channel.
+// transaction.
-func (database *Database) CountChannelMembers(
+func isSessionBannedInTx(
 	ctx context.Context,
-	channelID int64,
+	transaction *sql.Tx,
-) (int64, error) {
+	channelID, sessionID int64,
-	var count int64
+) (bool, error) {
+	var nick, username, hostname string
 
-	err := database.conn.QueryRowContext(ctx,
+	err := transaction.QueryRowContext(ctx,
-		`SELECT COUNT(*) FROM channel_members
+		`SELECT nick, username, hostname
-		 WHERE channel_id = ?`,
+		 FROM sessions WHERE id = ?`,
-		channelID,
+		sessionID,
-	).Scan(&count)
+	).Scan(&nick, &username, &hostname)
 	if err != nil {
-		return 0, fmt.Errorf(
+		return false, fmt.Errorf(
-			"count channel members: %w", err,
+			"get session hostmask: %w", err,
 		)
 	}
 
-	return count, nil
+	hostmask := FormatHostmask(nick, username, hostname)
+
+	rows, err := transaction.QueryContext(ctx,
+		`SELECT mask FROM channel_bans
+		 WHERE channel_id = ?`,
+		channelID)
+	if err != nil {
+		return false, fmt.Errorf("list channel bans: %w", err)
+	}
+
+	defer func() { _ = rows.Close() }()
+
+	for rows.Next() {
+		var mask string
+
+		err = rows.Scan(&mask)
+		if err != nil {
+			return false, fmt.Errorf(
+				"scan channel ban: %w", err,
+			)
+		}
+
+		if MatchBanMask(mask, hostmask) {
+			return true, nil
+		}
+	}
+
+	err = rows.Err()
+	if err != nil {
+		return false, fmt.Errorf(
+			"iterate channel bans: %w", err,
+		)
+	}
+
+	return false, nil
 }
 
 // IsChannelOperator checks if a session has operator
@@ -729,6 +890,49 @@ func (database *Database) DeleteChannelIfEmpty(
 	return nil
 }
 
+// PartChannelAndDeleteIfEmpty removes a session from a
+// channel and deletes the channel if that left it with no
+// members, in one transaction, so that no JOIN lands
+// between the two.
+func (database *Database) PartChannelAndDeleteIfEmpty(
+	ctx context.Context,
+	channelID, sessionID int64,
+) error {
+	transaction, err := database.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("begin part: %w", err)
+	}
+
+	defer func() { _ = transaction.Rollback() }()
+
+	_, err = transaction.ExecContext(ctx,
+		`DELETE FROM channel_members
+		 WHERE channel_id = ? AND session_id = ?`,
+		channelID, sessionID)
+	if err != nil {
+		return fmt.Errorf("part channel: %w", err)
+	}
+
+	_, err = transaction.ExecContext(ctx,
+		`DELETE FROM channels WHERE id = ?
+		 AND NOT EXISTS
+		 (SELECT 1 FROM channel_members
+		  WHERE channel_id = ?)`,
+		channelID, channelID)
+	if err != nil {
+		return fmt.Errorf(
+			"delete channel if empty: %w", err,
+		)
+	}
+
+	err = transaction.Commit()
+	if err != nil {
+		return fmt.Errorf("commit part: %w", err)
+	}
+
+	return nil
+}
+
 // scanChannels scans rows into a ChannelInfo slice.
 func scanChannels(
 	rows *sql.Rows,
@@ -2100,23 +2304,6 @@ func (database *Database) HasChannelInvite(
 	return count > 0, nil
 }
 
-// ClearChannelInvite removes a session's invite to a
-// channel (called after successful JOIN).
-func (database *Database) ClearChannelInvite(
-	ctx context.Context,
-	channelID, sessionID int64,
-) error {
-	_, err := database.conn.ExecContext(ctx,
-		`DELETE FROM channel_invites
-		 WHERE channel_id = ? AND session_id = ?`,
-		channelID, sessionID)
-	if err != nil {
-		return fmt.Errorf("clear invite: %w", err)
-	}
-
-	return nil
-}
-
 // --- Tier 2: Secret (+s) ---
 
 // IsChannelSecret checks if a channel has +s mode.

internal/db/queries_test.go

@@ -1342,17 +1342,6 @@ func TestChannelInviteCRUD(t *testing.T) {
 	if !has {
 		t.Fatal("should have invite")
 	}
-
-	// Clear invite.
-	err = database.ClearChannelInvite(ctx, chID, sid)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	has, _ = database.HasChannelInvite(ctx, chID, sid)
-	if has {
-		t.Fatal("invite should be cleared")
-	}
 }
 
 func TestChannelSecret(t *testing.T) {

internal/ircserver/conn.go

@@ -222,7 +222,9 @@ func (c *Conn) serve(
 ) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	defer c.cleanup(ctx)
+	// Stopping the server cancels ctx, and the cleanup must
+	// still end the session then.
+	defer c.cleanup(context.WithoutCancel(ctx))
 
 	ping := time.AfterFunc(interval, func() {
 		c.send("PING :" + c.serverSfx)

internal/ircserver/export_test.go

@@ -85,6 +85,10 @@ const PingInterval = pingInterval
 // for a line before it closes the connection.
 const PongDeadline = pongDeadline
 
+// StopTimeout is the longest Stop waits for the connections'
+// cleanup.
+const StopTimeout = stopTimeout
+
 // NewTestConn returns a Conn for nick on tcpConn, without a
 // server, database or service behind it.
 func NewTestConn(

internal/ircserver/server.go

@@ -16,6 +16,10 @@ import (
 	"sneak.berlin/go/neoirc/internal/service"
 )
 
+// stopTimeout is the longest stop waits for the connections'
+// cleanup, which ends their sessions.
+const stopTimeout = 5 * time.Second
+
 // Params defines the dependencies for creating an IRC
 // Server.
 type Params struct {
@@ -40,6 +44,10 @@ type Server struct {
 	conns    map[*Conn]struct{}
 	cancel   context.CancelFunc
 
+	// serving counts the connections whose serve, cleanup
+	// included, has not returned yet.
+	serving sync.WaitGroup
+
 	// pingInterval and pongDeadline are the constants of the
 	// same names, except in tests, which shorten them.
 	pingInterval time.Duration
@@ -63,6 +71,7 @@ func New(
 		listener: nil,
 		cancel:   nil,
 		mu:       sync.Mutex{},
+		serving:  sync.WaitGroup{},
 
 		pingInterval: pingInterval,
 		pongDeadline: pongDeadline,
@@ -112,7 +121,10 @@ func (s *Server) start(ctx context.Context, addr string) error {
 	return nil
 }
 
-// stop shuts down the listener and all connections.
+// stop shuts down the listener and all connections, then
+// waits up to stopTimeout for every connection's cleanup to
+// finish, so that their sessions are ended before the
+// database closes.
 func (s *Server) stop() {
 	if s.cancel != nil {
 		s.cancel()
@@ -127,6 +139,22 @@ func (s *Server) stop() {
 		c.conn.Close() //nolint:errcheck,gosec
 	}
 	s.mu.Unlock()
+
+	done := make(chan struct{})
+
+	go func() {
+		s.serving.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(stopTimeout):
+		s.log.Error(
+			"irc connections not cleaned up before stop timeout",
+			"timeout", stopTimeout,
+		)
+	}
 }
 
 // acceptLoop accepts new connections.
@@ -152,10 +180,24 @@ func (s *Server) acceptLoop(ctx context.Context) {
 		)
 
 		s.mu.Lock()
+
+		// stop cancels ctx before it takes s.mu to close the
+		// connections, so a connection accepted while stop
+		// runs is either closed and waited for by stop or
+		// closed here.
+		if ctx.Err() != nil {
+			s.mu.Unlock()
+			tcpConn.Close() //nolint:errcheck,gosec
+
+			return
+		}
+
 		s.conns[client] = struct{}{}
+		s.serving.Add(1)
 		s.mu.Unlock()
 
 		go func() {
+			defer s.serving.Done()
 			defer func() {
 				s.mu.Lock()
 				delete(s.conns, client)

internal/ircserver/server_test.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"crypto/rand"
 	"database/sql"
+	"errors"
 	"fmt"
 	"log/slog"
 	"net"
@@ -812,6 +813,120 @@ func TestQuit(t *testing.T) {
 	assertContains(t, lines, "goodbye", "QUIT reason")
 }
 
+// TestStopEndsSessions checks that stopping the server ends
+// a registered connection's session before Stop returns, as
+// closing the connection does: the session, its client and
+// its channel membership are gone from the database, and a
+// channel peer has been sent the QUIT.
+func TestStopEndsSessions(t *testing.T) {
+	t.Parallel()
+
+	env := newTestEnv(t)
+	client := env.dial(t)
+
+	client.register("shutirc")
+	client.joinAndDrain("#shut")
+
+	ctx := t.Context()
+
+	// A peer on the HTTP API, whose session outlives the stop.
+	peerID, peerClientID, _, err := env.database.CreateSession(
+		ctx, "peer", "peer", "", "",
+	)
+	if err != nil {
+		t.Fatalf("create peer session: %v", err)
+	}
+
+	channelID, err := env.database.GetChannelByName(ctx, "#shut")
+	if err != nil {
+		t.Fatalf("get channel: %v", err)
+	}
+
+	err = env.database.JoinChannel(ctx, channelID, peerID)
+	if err != nil {
+		t.Fatalf("join peer: %v", err)
+	}
+
+	env.srv.Stop()
+
+	_, err = env.database.GetSessionByNick(ctx, "shutirc")
+	if !errors.Is(err, sql.ErrNoRows) {
+		t.Errorf("look up shutirc after stop: %v, want no rows", err)
+	}
+
+	clients, err := env.database.GetClientCount(ctx)
+	if err != nil {
+		t.Fatalf("count clients: %v", err)
+	}
+
+	if clients != 1 {
+		t.Errorf("%d clients after stop, want only the peer's", clients)
+	}
+
+	members, err := env.database.GetChannelMemberIDs(ctx, channelID)
+	if err != nil {
+		t.Fatalf("get members: %v", err)
+	}
+
+	if len(members) != 1 || members[0] != peerID {
+		t.Errorf(
+			"members of #shut after stop: %v, want only the peer %d",
+			members, peerID,
+		)
+	}
+
+	msgs, _, err := env.database.PollMessages(ctx, peerClientID, 0, 0)
+	if err != nil {
+		t.Fatalf("poll peer: %v", err)
+	}
+
+	for _, msg := range msgs {
+		if msg.Command == "QUIT" && msg.From == "shutirc" {
+			return
+		}
+	}
+
+	t.Errorf("peer was not sent the QUIT of shutirc: %v", msgs)
+}
+
+// TestStopReturnsWhileCleanupIsStuck checks that Stop stops
+// waiting for a connection's cleanup after StopTimeout, so a
+// cleanup that cannot finish does not hold up stopping.
+func TestStopReturnsWhileCleanupIsStuck(t *testing.T) {
+	t.Parallel()
+
+	env := newTestEnv(t)
+	client := env.dial(t)
+
+	client.register("stuck")
+
+	// The test database has one connection. While the test
+	// holds it, the cleanup's first query waits for it.
+	held, err := env.database.GetDB().Conn(t.Context())
+	if err != nil {
+		t.Fatalf("take the database connection: %v", err)
+	}
+
+	stopped := make(chan struct{})
+
+	go func() {
+		env.srv.Stop()
+		close(stopped)
+	}()
+
+	select {
+	case <-stopped:
+	case <-time.After(ircserver.StopTimeout + testTimeout):
+		t.Error("Stop did not return while a cleanup was stuck")
+	}
+
+	// Lets the cleanup finish before the database is closed.
+	err = held.Close()
+	if err != nil {
+		t.Errorf("release the database connection: %v", err)
+	}
+}
+
 func TestTopicSetAndGet(t *testing.T) {
 	t.Parallel()
 

internal/server/server.go

@@ -45,11 +45,8 @@ type Params struct {
 // It manages routing, middleware, and lifecycle.
 type Server struct {
 	startupTime   time.Time
-	exitCode      int
 	sentryEnabled bool
 	log           *slog.Logger
-	ctx           context.Context //nolint:containedctx // signal handling pattern
-	cancelFunc    context.CancelFunc
 	httpServer    *http.Server
 	router        *chi.Mux
 	params        Params
@@ -78,13 +75,16 @@ func New(
 			srv.enableSentry()
 			srv.SetupRoutes()
 
-			// The start hook's context ends when the hook
+			// A closed stdout or stderr must not end the process.
-			// returns; serving must outlive it.
+			signal.Ignore(syscall.SIGPIPE)
-			go srv.serve() //nolint:contextcheck,gosec // G118
+
+			srv.startHTTP()
 
 			return nil
 		},
-		OnStop: func(_ context.Context) error {
+		OnStop: func(ctx context.Context) error {
+			srv.cleanShutdown(ctx)
+
 			return nil
 		},
 	})
@@ -129,44 +129,15 @@ func (srv *Server) enableSentry() {
 	srv.sentryEnabled = true
 }
 
-func (srv *Server) serve() int {
-	srv.ctx, srv.cancelFunc = context.WithCancel(
-		context.Background(),
-	)
-
-	go func() {
-		sigCh := make(chan os.Signal, 1)
-
-		signal.Ignore(syscall.SIGPIPE)
-		signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
-
-		sig := <-sigCh
-
-		srv.log.Info("signal received", "signal", sig)
-
-		if srv.cancelFunc != nil {
-			srv.cancelFunc()
-		}
-	}()
-
-	go srv.serveUntilShutdown()
-
-	<-srv.ctx.Done()
-
-	srv.cleanShutdown()
-
-	return srv.exitCode
-}
-
 func (srv *Server) cleanupForExit() {
 	srv.log.Info("cleaning up")
 }
 
-func (srv *Server) cleanShutdown() {
+// cleanShutdown stops the HTTP server, waiting up to
-	srv.exitCode = 0
+// shutdownTimeout for the requests in progress to finish.
-
+func (srv *Server) cleanShutdown(ctx context.Context) {
 	ctxShutdown, shutdownCancel := context.WithTimeout(
-		context.Background(), shutdownTimeout,
+		ctx, shutdownTimeout,
 	)
 
 	err := srv.httpServer.Shutdown(ctxShutdown)
@@ -191,7 +162,11 @@ func (srv *Server) configure() {
 	// Server configuration placeholder.
 }
 
-func (srv *Server) serveUntilShutdown() {
+// startHTTP creates the HTTP server and serves it from a new
+// goroutine. srv.httpServer is set before that goroutine
+// starts and never changes, so it and cleanShutdown read it
+// without a lock.
+func (srv *Server) startHTTP() {
 	listenAddr := fmt.Sprintf(
 		":%d", srv.params.Config.Port,
 	)
@@ -204,16 +179,16 @@ func (srv *Server) serveUntilShutdown() {
 		Handler:        srv,
 	}
 
+	go srv.serveUntilShutdown()
+}
+
+func (srv *Server) serveUntilShutdown() {
 	srv.log.Info(
-		"http begin listen", "listenaddr", listenAddr,
+		"http begin listen", "listenaddr", srv.httpServer.Addr,
 	)
 
 	err := srv.httpServer.ListenAndServe()
 	if err != nil && !errors.Is(err, http.ErrServerClosed) {
 		srv.log.Error("listen error", "error", err)
-
-		if srv.cancelFunc != nil {
-			srv.cancelFunc()
-		}
 	}
 }

internal/server/server_test.go

@@ -0,0 +1,247 @@
+package server_test
+
+import (
+	"crypto/rand"
+	"fmt"
+	"net"
+	"net/http"
+	"strconv"
+	"testing"
+	"time"
+
+	"go.uber.org/fx"
+	"go.uber.org/fx/fxtest"
+	"sneak.berlin/go/neoirc/internal/broker"
+	"sneak.berlin/go/neoirc/internal/config"
+	"sneak.berlin/go/neoirc/internal/db"
+	"sneak.berlin/go/neoirc/internal/globals"
+	"sneak.berlin/go/neoirc/internal/handlers"
+	"sneak.berlin/go/neoirc/internal/healthcheck"
+	"sneak.berlin/go/neoirc/internal/logger"
+	"sneak.berlin/go/neoirc/internal/middleware"
+	"sneak.berlin/go/neoirc/internal/server"
+	"sneak.berlin/go/neoirc/internal/service"
+	"sneak.berlin/go/neoirc/internal/stats"
+)
+
+const (
+	testTimeout = 5 * time.Second
+
+	// checkStep is how often waitUntil checks again.
+	checkStep = 10 * time.Millisecond
+
+	// stopHoldTime is how long a test waits to see that the
+	// app has not stopped.
+	stopHoldTime = 200 * time.Millisecond
+)
+
+// TestStopWaitsForRequestInProgress checks that stopping the
+// app, which runs the stop hook server.New registers, waits
+// for a long poll the HTTP server is serving to finish, and
+// that the poll gets its response.
+func TestStopWaitsForRequestInProgress(t *testing.T) {
+	t.Parallel()
+
+	port := freePort(t)
+	addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port))
+
+	app, brk, database := startApp(t, port)
+
+	sessionID, _, token, err := database.CreateSession(
+		t.Context(), "poller", "poller", "", "",
+	)
+	if err != nil {
+		t.Fatalf("create session: %v", err)
+	}
+
+	waitUntil(t, "the HTTP server to listen", func() bool {
+		return canConnect(t, addr)
+	})
+
+	statuses := make(chan int, 1)
+
+	go func() {
+		statuses <- getAuthenticated(
+			t, "http://"+addr+"/api/v1/messages?timeout=30", token,
+		)
+	}()
+
+	waitUntil(t, "the long poll to wait for a message", func() bool {
+		return brk.ListenerCount(sessionID) == 1
+	})
+
+	var stopErr error
+
+	stopped := make(chan struct{})
+
+	go func() {
+		stopErr = app.Stop(t.Context())
+
+		close(stopped)
+	}()
+
+	select {
+	case <-stopped:
+		t.Error("the app stopped while a request was in progress")
+	case <-time.After(stopHoldTime):
+	}
+
+	brk.Notify(sessionID)
+
+	status := <-statuses
+	if status != http.StatusOK {
+		t.Errorf("status %d, want %d", status, http.StatusOK)
+	}
+
+	select {
+	case <-stopped:
+		if stopErr != nil {
+			t.Errorf("stop: %v", stopErr)
+		}
+	case <-time.After(testTimeout):
+		t.Fatal("the app did not stop after the request finished")
+	}
+}
+
+// startApp starts the server with the dependencies
+// cmd/neoircd gives it, on a fresh in-memory database, with
+// the HTTP server listening on port.
+func startApp(
+	t *testing.T,
+	port int,
+) (*fxtest.App, *broker.Broker, *db.Database) {
+	t.Helper()
+
+	var (
+		brk      *broker.Broker
+		database *db.Database
+	)
+
+	app := fxtest.New(t,
+		fx.Supply(&globals.Globals{
+			Appname:   "neoirc-test",
+			Version:   "test",
+			StartTime: time.Now(),
+		}),
+		fx.Provide(
+			broker.New,
+			config.New,
+			db.New,
+			handlers.New,
+			healthcheck.New,
+			logger.New,
+			middleware.New,
+			server.New,
+			service.New,
+			stats.New,
+		),
+		fx.Decorate(func(cfg *config.Config) *config.Config {
+			// A random name, so that no other test can share it.
+			cfg.DBURL = fmt.Sprintf(
+				"file:%s?mode=memory&cache=shared", rand.Text(),
+			)
+			cfg.Port = port
+
+			return cfg
+		}),
+		fx.Invoke(func(*server.Server) {}),
+		fx.Populate(&brk, &database),
+	)
+
+	app.RequireStart()
+
+	// Stops the app if the test ends before stopping it.
+	t.Cleanup(app.RequireStop)
+
+	return app, brk, database
+}
+
+// freePort returns a TCP port that nothing was listening on a
+// moment ago.
+func freePort(t *testing.T) int {
+	t.Helper()
+
+	var listenConfig net.ListenConfig
+
+	listener, err := listenConfig.Listen(
+		t.Context(), "tcp", "127.0.0.1:0",
+	)
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+
+	tcpAddr, ok := listener.Addr().(*net.TCPAddr)
+	if !ok {
+		t.Fatalf("listener address %v is not TCP", listener.Addr())
+	}
+
+	err = listener.Close()
+	if err != nil {
+		t.Fatalf("close listener: %v", err)
+	}
+
+	return tcpAddr.Port
+}
+
+// canConnect reports whether addr accepts a TCP connection.
+func canConnect(t *testing.T, addr string) bool {
+	t.Helper()
+
+	var dialer net.Dialer
+
+	conn, err := dialer.DialContext(t.Context(), "tcp", addr)
+	if err != nil {
+		return false
+	}
+
+	_ = conn.Close()
+
+	return true
+}
+
+// waitUntil calls done every checkStep until it returns true,
+// and fails the test if it does not within testTimeout.
+func waitUntil(t *testing.T, what string, done func() bool) {
+	t.Helper()
+
+	deadline := time.Now().Add(testTimeout)
+
+	for !done() {
+		if time.Now().After(deadline) {
+			t.Fatalf("timed out waiting for %s", what)
+		}
+
+		time.Sleep(checkStep)
+	}
+}
+
+// getAuthenticated sends a GET request to url as the session
+// whose auth token is token, and returns the response status,
+// or 0 if there was no response.
+func getAuthenticated(t *testing.T, url, token string) int {
+	t.Helper()
+
+	request, err := http.NewRequestWithContext(
+		t.Context(), http.MethodGet, url, nil,
+	)
+	if err != nil {
+		t.Errorf("new request: %v", err)
+
+		return 0
+	}
+
+	request.Header.Set("Cookie", "neoirc_auth="+token)
+
+	client := http.Client{Timeout: testTimeout}
+
+	response, err := client.Do(request)
+	if err != nil {
+		t.Errorf("get %s: %v", url, err)
+
+		return 0
+	}
+
+	_ = response.Body.Close()
+
+	return response.StatusCode
+}

internal/service/service.go

@@ -536,41 +536,13 @@ func (s *Service) JoinChannel(
 		}
 	}
 
-	chID, err := s.db.GetOrCreateChannel(ctx, channel)
+	chID, isCreator, err := s.db.CreateOrJoinChannel(
-	if err != nil {
+		ctx, channel, sessionID, suppliedKey,
-		return nil, fmt.Errorf("get/create channel: %w", err)
-	}
-
-	memberCount, countErr := s.db.CountChannelMembers(
-		ctx, chID,
 	)
-	isCreator := countErr == nil && memberCount == 0
-
-	if !isCreator {
-		joinErr := checkJoinRestrictions(
-			ctx, s.db, chID, sessionID,
-			channel, suppliedKey, memberCount,
-		)
-		if joinErr != nil {
-			return nil, joinErr
-		}
-	}
-
-	if isCreator {
-		err = s.db.JoinChannelAsOperator(
-			ctx, chID, sessionID,
-		)
-	} else {
-		err = s.db.JoinChannel(ctx, chID, sessionID)
-	}
-
 	if err != nil {
-		return nil, fmt.Errorf("join channel: %w", err)
+		return nil, joinError(channel, err)
 	}
 
-	// Clear invite after successful join.
-	_ = s.db.ClearChannelInvite(ctx, chID, sessionID)
-
 	memberIDs, _ := s.db.GetChannelMemberIDs(ctx, chID)
 	body, _ := json.Marshal([]string{channel}) //nolint:errchkjson
 
@@ -622,8 +594,7 @@ func (s *Service) PartChannel(
 		nil, body, nil, recipients,
 	)
 
-	s.db.PartChannel(ctx, chID, sessionID) //nolint:errcheck,gosec
+	s.db.PartChannelAndDeleteIfEmpty(ctx, chID, sessionID) //nolint:errcheck,gosec
-	s.db.DeleteChannelIfEmpty(ctx, chID)   //nolint:errcheck,gosec
 
 	return nil
 }
@@ -748,8 +719,7 @@ func (s *Service) KickUser(
 		params, body, nil, memberIDs,
 	)
 
-	s.db.PartChannel(ctx, chID, targetSID) //nolint:errcheck,gosec
+	s.db.PartChannelAndDeleteIfEmpty(ctx, chID, targetSID) //nolint:errcheck,gosec
-	s.db.DeleteChannelIfEmpty(ctx, chID)   //nolint:errcheck,gosec
 
 	return nil
 }
@@ -1392,54 +1362,30 @@ func (s *Service) broadcastNickChange(
 	}
 }
 
-// checkJoinRestrictions validates Tier 2 join conditions:
+// joinError returns the IRC error for a JOIN of channel
-// bans, invite-only, channel key, and user limit.
+// that db.CreateOrJoinChannel refused with err, or err
-func checkJoinRestrictions(
+// wrapped if it is not a refusal.
-	ctx context.Context,
+func joinError(channel string, err error) error {
-	database *db.Database,
+	switch {
-	chID, sessionID int64,
+	case errors.Is(err, db.ErrBannedFromChannel):
-	channel, suppliedKey string,
-	memberCount int64,
-) error {
-	isBanned, banErr := database.IsSessionBanned(
-		ctx, chID, sessionID,
-	)
-	if banErr == nil && isBanned {
 		return &IRCError{
 			Code:    irc.ErrBannedFromChan,
 			Params:  []string{channel},
 			Message: "Cannot join channel (+b)",
 		}
-	}
+	case errors.Is(err, db.ErrInviteOnlyChannel):
-
+		return &IRCError{
-	isInviteOnly, ioErr := database.IsChannelInviteOnly(
+			Code:    irc.ErrInviteOnlyChan,
-		ctx, chID,
+			Params:  []string{channel},
-	)
+			Message: "Cannot join channel (+i)",
-	if ioErr == nil && isInviteOnly {
-		hasInvite, invErr := database.HasChannelInvite(
-			ctx, chID, sessionID,
-		)
-		if invErr != nil || !hasInvite {
-			return &IRCError{
-				Code:    irc.ErrInviteOnlyChan,
-				Params:  []string{channel},
-				Message: "Cannot join channel (+i)",
-			}
 		}
-	}
+	case errors.Is(err, db.ErrBadChannelKey):
-
-	key, keyErr := database.GetChannelKey(ctx, chID)
-	if keyErr == nil && key != "" && suppliedKey != key {
 		return &IRCError{
 			Code:    irc.ErrBadChannelKey,
 			Params:  []string{channel},
 			Message: "Cannot join channel (+k)",
 		}
-	}
+	case errors.Is(err, db.ErrChannelIsFull):
-
-	limit, limErr := database.GetChannelUserLimit(ctx, chID)
-	if limErr == nil && limit > 0 &&
-		memberCount >= int64(limit) {
 		return &IRCError{
 			Code:    irc.ErrChannelIsFull,
 			Params:  []string{channel},
@@ -1447,7 +1393,7 @@ func checkJoinRestrictions(
 		}
 	}
 
-	return nil
+	return fmt.Errorf("join channel: %w", err)
 }
 
 // extractHashcashFromMeta parses the meta JSON and

internal/service/service_test.go

@@ -13,6 +13,7 @@ import (
 	"os"
 	"slices"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -255,6 +256,402 @@ func TestPartChannel(t *testing.T) {
 	}
 }
 
+// concurrentJoinTrials is how many times each test of
+// concurrent JOINs repeats its race. Before a JOIN ran as
+// one transaction, ten trials lost each race every time.
+const concurrentJoinTrials = 10
+
+// runConcurrently runs each of funcs in its own goroutine,
+// releasing them all at once, and returns their errors in
+// the order of funcs.
+func runConcurrently(funcs ...func() error) []error {
+	start := make(chan struct{})
+	errs := make([]error, len(funcs))
+
+	var wg sync.WaitGroup
+
+	for i, run := range funcs {
+		wg.Add(1)
+
+		go func() {
+			defer wg.Done()
+
+			<-start
+
+			errs[i] = run()
+		}()
+	}
+
+	close(start)
+	wg.Wait()
+
+	return errs
+}
+
+// joinConcurrently creates a session for each nick and has
+// them all JOIN channel at once. It returns the result and
+// error of each JOIN, in the order of nicks.
+func joinConcurrently(
+	ctx context.Context,
+	t *testing.T,
+	env *testEnv,
+	channel string,
+	nicks ...string,
+) ([]*service.JoinResult, []error) {
+	t.Helper()
+
+	results := make([]*service.JoinResult, len(nicks))
+	joins := make([]func() error, len(nicks))
+
+	for i, nick := range nicks {
+		sessionID := createSession(ctx, t, env.db, nick)
+
+		joins[i] = func() error {
+			result, err := env.svc.JoinChannel(
+				ctx, sessionID, nick, channel, "",
+			)
+			results[i] = result
+
+			return err
+		}
+	}
+
+	return results, runConcurrently(joins...)
+}
+
+// TestConcurrentJoinsRespectUserLimit checks that of two
+// concurrent JOINs of a +l 2 channel with one member, one
+// gets in and the other gets ERR_CHANNELISFULL.
+func TestConcurrentJoinsRespectUserLimit(t *testing.T) {
+	const userLimit = 2
+
+	env := newTestEnv(t)
+	ctx := t.Context()
+
+	for trial := range concurrentJoinTrials {
+		channel := fmt.Sprintf("#full%d", trial)
+		opNick := fmt.Sprintf("op%d", trial)
+		opSID := createSession(ctx, t, env.db, opNick)
+
+		created, err := env.svc.JoinChannel(
+			ctx, opSID, opNick, channel, "",
+		)
+		if err != nil {
+			t.Fatalf("join %s: %v", opNick, err)
+		}
+
+		err = env.db.SetChannelUserLimit(
+			ctx, created.ChannelID, userLimit,
+		)
+		if err != nil {
+			t.Fatalf("set +l: %v", err)
+		}
+
+		_, errs := joinConcurrently(
+			ctx, t, env, channel,
+			fmt.Sprintf("a%d", trial), fmt.Sprintf("b%d", trial),
+		)
+
+		refused := 0
+
+		for _, joinErr := range errs {
+			var ircErr *service.IRCError
+
+			switch {
+			case joinErr == nil:
+			case errors.As(joinErr, &ircErr) &&
+				ircErr.Code == irc.ErrChannelIsFull:
+				refused++
+			default:
+				t.Errorf("trial %d: JOIN: %v", trial, joinErr)
+			}
+		}
+
+		members, err := env.db.ChannelMembers(ctx, created.ChannelID)
+		if err != nil {
+			t.Fatalf("channel members: %v", err)
+		}
+
+		if len(members) != userLimit || refused != 1 {
+			t.Errorf(
+				"trial %d: %d members and %d refused, want %d and 1",
+				trial, len(members), refused, userLimit,
+			)
+		}
+	}
+}
+
+// TestConcurrentFirstJoinsMakeOneOperator checks that of
+// two concurrent JOINs that create a channel, exactly one
+// makes its session the operator.
+func TestConcurrentFirstJoinsMakeOneOperator(t *testing.T) {
+	env := newTestEnv(t)
+	ctx := t.Context()
+
+	for trial := range concurrentJoinTrials {
+		results, errs := joinConcurrently(
+			ctx, t, env, fmt.Sprintf("#new%d", trial),
+			fmt.Sprintf("first%d", trial),
+			fmt.Sprintf("second%d", trial),
+		)
+
+		creators := 0
+
+		for i, err := range errs {
+			if err != nil {
+				t.Fatalf("trial %d: JOIN: %v", trial, err)
+			}
+
+			if results[i].IsCreator {
+				creators++
+			}
+		}
+
+		members, err := env.db.ChannelMembers(
+			ctx, results[0].ChannelID,
+		)
+		if err != nil {
+			t.Fatalf("channel members: %v", err)
+		}
+
+		operators := 0
+
+		for _, member := range members {
+			if member.IsOperator {
+				operators++
+			}
+		}
+
+		if creators != 1 || operators != 1 {
+			t.Errorf(
+				"trial %d: %d creators and %d operators, want 1 each",
+				trial, creators, operators,
+			)
+		}
+	}
+}
+
+// TestJoinConcurrentWithLastPart checks that a JOIN
+// concurrent with the PART of a channel's only member
+// either joins a channel that still exists or gets
+// ERR_NOSUCHCHANNEL, and never fails otherwise.
+func TestJoinConcurrentWithLastPart(t *testing.T) {
+	env := newTestEnv(t)
+	ctx := t.Context()
+
+	for trial := range concurrentJoinTrials {
+		channel := fmt.Sprintf("#last%d", trial)
+		leaverNick := fmt.Sprintf("leaver%d", trial)
+		leaverSID := createSession(ctx, t, env.db, leaverNick)
+		joinerNick := fmt.Sprintf("joiner%d", trial)
+		joinerSID := createSession(ctx, t, env.db, joinerNick)
+
+		_, err := env.svc.JoinChannel(
+			ctx, leaverSID, leaverNick, channel, "",
+		)
+		if err != nil {
+			t.Fatalf("join %s: %v", leaverNick, err)
+		}
+
+		errs := runConcurrently(
+			func() error {
+				return env.svc.PartChannel(
+					ctx, leaverSID, leaverNick, channel, "",
+				)
+			},
+			func() error {
+				_, joinErr := env.svc.JoinChannel(
+					ctx, joinerSID, joinerNick, channel, "",
+				)
+
+				return joinErr
+			},
+		)
+		if errs[0] != nil {
+			t.Fatalf("trial %d: PART: %v", trial, errs[0])
+		}
+
+		var ircErr *service.IRCError
+
+		switch {
+		case errs[1] == nil:
+			chID, lookupErr := env.db.GetChannelByName(ctx, channel)
+			isMember, _ := env.db.IsChannelMember(ctx, chID, joinerSID)
+
+			if lookupErr != nil || !isMember {
+				t.Errorf(
+					"trial %d: JOIN succeeded, but %s is not on %s",
+					trial, joinerNick, channel,
+				)
+			}
+		case errors.As(errs[1], &ircErr) &&
+			ircErr.Code == irc.ErrNoSuchChannel:
+		default:
+			t.Errorf(
+				"trial %d: JOIN got %v, want success or 403",
+				trial, errs[1],
+			)
+		}
+	}
+}
+
+// TestJoinUsesUpInvite checks that the JOIN an invite let
+// into a +i channel uses the invite up, so that a JOIN
+// after leaving gets ERR_INVITEONLYCHAN.
+func TestJoinUsesUpInvite(t *testing.T) {
+	const channel = "#club"
+
+	env := newTestEnv(t)
+	ctx := t.Context()
+
+	opSID := createSession(ctx, t, env.db, "op")
+	guestSID := createSession(ctx, t, env.db, "guest")
+
+	created, err := env.svc.JoinChannel(ctx, opSID, "op", channel, "")
+	if err != nil {
+		t.Fatalf("join op: %v", err)
+	}
+
+	err = env.db.SetChannelInviteOnly(ctx, created.ChannelID, true)
+	if err != nil {
+		t.Fatalf("set +i: %v", err)
+	}
+
+	_, err = env.svc.InviteUser(ctx, opSID, "op", "guest", channel)
+	if err != nil {
+		t.Fatalf("invite guest: %v", err)
+	}
+
+	_, err = env.svc.JoinChannel(ctx, guestSID, "guest", channel, "")
+	if err != nil {
+		t.Fatalf("invited join: %v", err)
+	}
+
+	err = env.svc.PartChannel(ctx, guestSID, "guest", channel, "")
+	if err != nil {
+		t.Fatalf("part guest: %v", err)
+	}
+
+	_, err = env.svc.JoinChannel(ctx, guestSID, "guest", channel, "")
+
+	var ircErr *service.IRCError
+	if !errors.As(err, &ircErr) || ircErr.Code != irc.ErrInviteOnlyChan {
+		t.Fatalf("second join: want ERR_INVITEONLYCHAN, got %v", err)
+	}
+}
+
+// TestJoinRefusedByHostmaskBan checks that a ban naming a
+// session's nick, username and hostname refuses its JOIN
+// with ERR_BANNEDFROMCHAN.
+func TestJoinRefusedByHostmaskBan(t *testing.T) {
+	const channel = "#club"
+
+	env := newTestEnv(t)
+	ctx := t.Context()
+
+	opSID := createSession(ctx, t, env.db, "op")
+
+	created, err := env.svc.JoinChannel(ctx, opSID, "op", channel, "")
+	if err != nil {
+		t.Fatalf("join op: %v", err)
+	}
+
+	err = env.db.AddChannelBan(
+		ctx, created.ChannelID, "guest!user@host.example", "op",
+	)
+	if err != nil {
+		t.Fatalf("add ban: %v", err)
+	}
+
+	guestSID, _, _, err := env.db.CreateSession(
+		ctx, "guest", "user", "host.example", "127.0.0.1",
+	)
+	if err != nil {
+		t.Fatalf("create session guest: %v", err)
+	}
+
+	_, err = env.svc.JoinChannel(ctx, guestSID, "guest", channel, "")
+
+	var ircErr *service.IRCError
+	if !errors.As(err, &ircErr) || ircErr.Code != irc.ErrBannedFromChan {
+		t.Fatalf("want ERR_BANNEDFROMCHAN, got %v", err)
+	}
+}
+
+// TestPartDeletesChannelOnlyWhenEmpty checks that a PART
+// keeps a channel other members are still on and deletes
+// the channel when its last member parts.
+func TestPartDeletesChannelOnlyWhenEmpty(t *testing.T) {
+	const channel = "#room"
+
+	env := newTestEnv(t)
+	ctx := t.Context()
+
+	aliceSID := createSession(ctx, t, env.db, "alice")
+	bobSID := createSession(ctx, t, env.db, "bob")
+
+	_, err := env.svc.JoinChannel(ctx, aliceSID, "alice", channel, "")
+	if err != nil {
+		t.Fatalf("join alice: %v", err)
+	}
+
+	_, err = env.svc.JoinChannel(ctx, bobSID, "bob", channel, "")
+	if err != nil {
+		t.Fatalf("join bob: %v", err)
+	}
+
+	err = env.svc.PartChannel(ctx, aliceSID, "alice", channel, "")
+	if err != nil {
+		t.Fatalf("part alice: %v", err)
+	}
+
+	chID, err := env.db.GetChannelByName(ctx, channel)
+	if err != nil {
+		t.Fatalf("channel gone while bob is on it: %v", err)
+	}
+
+	isMember, _ := env.db.IsChannelMember(ctx, chID, bobSID)
+	if !isMember {
+		t.Fatal("bob is not on the channel after alice parted")
+	}
+
+	err = env.svc.PartChannel(ctx, bobSID, "bob", channel, "")
+	if err != nil {
+		t.Fatalf("part bob: %v", err)
+	}
+
+	_, err = env.db.GetChannelByName(ctx, channel)
+	if err == nil {
+		t.Fatal("channel still exists after its last member parted")
+	}
+}
+
+// TestKickOfLastMemberDeletesChannel checks that a KICK
+// that leaves a channel with no members deletes it.
+func TestKickOfLastMemberDeletesChannel(t *testing.T) {
+	const channel = "#room"
+
+	env := newTestEnv(t)
+	ctx := t.Context()
+
+	sid := createSession(ctx, t, env.db, "alice")
+
+	_, err := env.svc.JoinChannel(ctx, sid, "alice", channel, "")
+	if err != nil {
+		t.Fatalf("join alice: %v", err)
+	}
+
+	err = env.svc.KickUser(ctx, sid, "alice", channel, "alice", "")
+	if err != nil {
+		t.Fatalf("kick alice: %v", err)
+	}
+
+	_, err = env.db.GetChannelByName(ctx, channel)
+	if err == nil {
+		t.Fatal("channel still exists after its last member was kicked")
+	}
+}
+
 func TestSendChannelMessage(t *testing.T) {
 	env := newTestEnv(t)
 	ctx := t.Context()