Run JOIN, PART and KICK as one transaction each (closes #153)

JoinChannel made the channel, counted its members, ran the join
checks and inserted the member as separate statements, so two
requests could each decide on a count the other was about to change:
+l was exceeded, two first joiners both became operator, and a JOIN
racing the last PART failed its foreign key and answered 500.

db.CreateOrJoinChannel now does all of that, and uses up the invite,
in one transaction; the join checks moved into it, and the service
turns its refusals into the same IRC errors as before. PART and KICK
remove the member and delete an empty channel in one transaction
(db.PartChannelAndDeleteIfEmpty).

Model: opus-5-5

internal/service/service_test.go

@@ -13,6 +13,7 @@ import (
 	"os"
 	"slices"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -255,6 +256,245 @@ 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],
+			)
+		}
+	}
+}
+
 func TestSendChannelMessage(t *testing.T) {
 	env := newTestEnv(t)
 	ctx := t.Context()