Broadcast a session's QUIT once when its clients leave together (closes #154)

Logout deleted the client, counted the session's clients and ran
BroadcastQuit as separate statements, and BroadcastQuit read the peers,
parted, deleted empty channels and deleted the session separately too.
Two logouts or QUITs of one session at once both sent its QUIT, and a
JOIN landing between a part and the channel's deletion got into a +k
channel without the key.

db.QuitSession now does all of that in one transaction, parting each
channel as PartChannelAndDeleteIfEmpty does, and stores and queues the
QUIT in it only when it deleted the session, so the QUIT is queued
before the nick is free. db.LogoutClient deletes the client and counts
the rest in the same transaction.

Model: opus-5-5

internal/db/queries.go

@@ -854,42 +854,6 @@ func (database *Database) SetChannelTopicLocked(
 	return nil
 }
 
-// PartChannel removes a session from a channel.
-func (database *Database) PartChannel(
-	ctx context.Context,
-	channelID, sessionID int64,
-) error {
-	_, err := database.conn.ExecContext(ctx,
-		`DELETE FROM channel_members
-		 WHERE channel_id = ? AND session_id = ?`,
-		channelID, sessionID)
-	if err != nil {
-		return fmt.Errorf("part channel: %w", err)
-	}
-
-	return nil
-}
-
-// DeleteChannelIfEmpty removes a channel with no members.
-func (database *Database) DeleteChannelIfEmpty(
-	ctx context.Context,
-	channelID int64,
-) error {
-	_, err := database.conn.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,
-		)
-	}
-
-	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
@@ -905,7 +869,29 @@ func (database *Database) PartChannelAndDeleteIfEmpty(
 
 	defer func() { _ = transaction.Rollback() }()
 
-	_, err = transaction.ExecContext(ctx,
+	err = partChannelAndDeleteIfEmptyInTx(
+		ctx, transaction, channelID, sessionID,
+	)
+	if err != nil {
+		return err
+	}
+
+	err = transaction.Commit()
+	if err != nil {
+		return fmt.Errorf("commit part: %w", err)
+	}
+
+	return nil
+}
+
+// partChannelAndDeleteIfEmptyInTx is
+// PartChannelAndDeleteIfEmpty, writing through transaction.
+func partChannelAndDeleteIfEmptyInTx(
+	ctx context.Context,
+	transaction *sql.Tx,
+	channelID, sessionID int64,
+) error {
+	_, err := transaction.ExecContext(ctx,
 		`DELETE FROM channel_members
 		 WHERE channel_id = ? AND session_id = ?`,
 		channelID, sessionID)
@@ -925,11 +911,6 @@ func (database *Database) PartChannelAndDeleteIfEmpty(
 		)
 	}
 
-	err = transaction.Commit()
-	if err != nil {
-		return fmt.Errorf("commit part: %w", err)
-	}
-
 	return nil
 }
 
@@ -1156,6 +1137,40 @@ func (database *Database) InsertMessage(
 	params json.RawMessage,
 	body json.RawMessage,
 	meta json.RawMessage,
+) (int64, string, error) {
+	transaction, err := database.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return 0, "", fmt.Errorf("begin insert message: %w", err)
+	}
+
+	defer func() { _ = transaction.Rollback() }()
+
+	dbID, msgUUID, err := insertMessageInTx(
+		ctx, transaction, command, from, target,
+		toSessionID, toChannelID, params, body, meta,
+	)
+	if err != nil {
+		return 0, "", err
+	}
+
+	err = transaction.Commit()
+	if err != nil {
+		return 0, "", fmt.Errorf("commit insert message: %w", err)
+	}
+
+	return dbID, msgUUID, nil
+}
+
+// insertMessageInTx is InsertMessage, writing through
+// transaction.
+func insertMessageInTx(
+	ctx context.Context,
+	transaction *sql.Tx,
+	command, from, target string,
+	toSessionID, toChannelID int64,
+	params json.RawMessage,
+	body json.RawMessage,
+	meta json.RawMessage,
 ) (int64, string, error) {
 	msgUUID := uuid.New().String()
 	now := time.Now().UTC()
@@ -1172,7 +1187,7 @@ func (database *Database) InsertMessage(
 		meta = json.RawMessage("{}")
 	}
 
-	res, err := database.conn.ExecContext(ctx,
+	res, err := transaction.ExecContext(ctx,
 		`INSERT INTO messages
 		 (uuid, command, msg_from, msg_to,
 		  to_session_id, to_channel_id,
@@ -1199,7 +1214,34 @@ func (database *Database) EnqueueToSession(
 	ctx context.Context,
 	sessionID, messageID int64,
 ) error {
-	_, err := database.conn.ExecContext(ctx,
+	transaction, err := database.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("begin enqueue: %w", err)
+	}
+
+	defer func() { _ = transaction.Rollback() }()
+
+	err = enqueueToSessionInTx(ctx, transaction, sessionID, messageID)
+	if err != nil {
+		return err
+	}
+
+	err = transaction.Commit()
+	if err != nil {
+		return fmt.Errorf("commit enqueue: %w", err)
+	}
+
+	return nil
+}
+
+// enqueueToSessionInTx is EnqueueToSession, writing through
+// transaction.
+func enqueueToSessionInTx(
+	ctx context.Context,
+	transaction *sql.Tx,
+	sessionID, messageID int64,
+) error {
+	_, err := transaction.ExecContext(ctx,
 		`INSERT OR IGNORE INTO client_queues
 		 (client_id, message_id, created_at)
 		 SELECT c.id, ?, ?
@@ -1470,38 +1512,175 @@ func (database *Database) SetTopic(
 	return nil
 }
 
-// DeleteSession removes a session and all its data.
-func (database *Database) DeleteSession(
+// QuitSession ends a session in one transaction: it parts
+// the session from its channels, deleting each channel this
+// leaves with no members, deletes the session with its
+// clients, and stores a QUIT from nick with body and queues
+// it to the other sessions that were on those channels. It
+// returns those sessions, for the caller to wake once the
+// QUIT is committed. Of several calls for one session at
+// once, only the first finds the session; the others change
+// nothing and return none.
+func (database *Database) QuitSession(
 	ctx context.Context,
 	sessionID int64,
-) error {
-	_, err := database.conn.ExecContext(
-		ctx,
-		"DELETE FROM sessions WHERE id = ?",
-		sessionID,
-	)
+	nick string,
+	body json.RawMessage,
+) ([]int64, error) {
+	transaction, err := database.conn.BeginTx(ctx, nil)
 	if err != nil {
-		return fmt.Errorf("delete session: %w", err)
+		return nil, fmt.Errorf("begin quit: %w", err)
 	}
 
-	return nil
+	defer func() { _ = transaction.Rollback() }()
+
+	peerIDs, err := quitSessionInTx(
+		ctx, transaction, sessionID, nick, body,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	err = transaction.Commit()
+	if err != nil {
+		return nil, fmt.Errorf("commit quit: %w", err)
+	}
+
+	return peerIDs, nil
 }
 
-// DeleteClient removes a single client record by ID.
-func (database *Database) DeleteClient(
+// LogoutClient deletes a client of a session and, if that
+// leaves the session with no clients, does what QuitSession
+// does, in one transaction. It returns what QuitSession
+// returns, or none if the session keeps a client.
+func (database *Database) LogoutClient(
 	ctx context.Context,
-	clientID int64,
-) error {
-	_, err := database.conn.ExecContext(
-		ctx,
-		"DELETE FROM clients WHERE id = ?",
-		clientID,
-	)
+	sessionID, clientID int64,
+	nick string,
+	body json.RawMessage,
+) ([]int64, error) {
+	transaction, err := database.conn.BeginTx(ctx, nil)
 	if err != nil {
-		return fmt.Errorf("delete client: %w", err)
+		return nil, fmt.Errorf("begin logout: %w", err)
 	}
 
-	return nil
+	defer func() { _ = transaction.Rollback() }()
+
+	_, err = transaction.ExecContext(ctx,
+		"DELETE FROM clients WHERE id = ?", clientID)
+	if err != nil {
+		return nil, fmt.Errorf("delete client: %w", err)
+	}
+
+	var remaining int64
+
+	err = transaction.QueryRowContext(ctx,
+		"SELECT COUNT(*) FROM clients WHERE session_id = ?",
+		sessionID,
+	).Scan(&remaining)
+	if err != nil {
+		return nil, fmt.Errorf("count clients: %w", err)
+	}
+
+	var peerIDs []int64
+
+	if remaining == 0 {
+		peerIDs, err = quitSessionInTx(
+			ctx, transaction, sessionID, nick, body,
+		)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	err = transaction.Commit()
+	if err != nil {
+		return nil, fmt.Errorf("commit logout: %w", err)
+	}
+
+	return peerIDs, nil
+}
+
+// quitSessionInTx is QuitSession, writing through
+// transaction.
+func quitSessionInTx(
+	ctx context.Context,
+	transaction *sql.Tx,
+	sessionID int64,
+	nick string,
+	body json.RawMessage,
+) ([]int64, error) {
+	rows, err := transaction.QueryContext(ctx,
+		`SELECT DISTINCT session_id FROM channel_members
+		 WHERE session_id != ? AND channel_id IN
+		 (SELECT channel_id FROM channel_members
+		  WHERE session_id = ?)`,
+		sessionID, sessionID)
+	if err != nil {
+		return nil, fmt.Errorf("get channel peers: %w", err)
+	}
+
+	peerIDs, err := scanInt64s(rows)
+	if err != nil {
+		return nil, err
+	}
+
+	rows, err = transaction.QueryContext(ctx,
+		`SELECT channel_id FROM channel_members
+		 WHERE session_id = ?`, sessionID)
+	if err != nil {
+		return nil, fmt.Errorf(
+			"get session channel ids: %w", err,
+		)
+	}
+
+	channelIDs, err := scanInt64s(rows)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, channelID := range channelIDs {
+		err = partChannelAndDeleteIfEmptyInTx(
+			ctx, transaction, channelID, sessionID,
+		)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	result, err := transaction.ExecContext(ctx,
+		"DELETE FROM sessions WHERE id = ?", sessionID)
+	if err != nil {
+		return nil, fmt.Errorf("delete session: %w", err)
+	}
+
+	deletedRows, err := result.RowsAffected()
+	if err != nil {
+		return nil, fmt.Errorf("delete session: %w", err)
+	}
+
+	if deletedRows == 0 || len(peerIDs) == 0 {
+		return nil, nil
+	}
+
+	messageID, _, err := insertMessageInTx(
+		ctx, transaction, irc.CmdQuit, nick, "", 0, 0,
+		nil, body, nil,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, peerID := range peerIDs {
+		err = enqueueToSessionInTx(
+			ctx, transaction, peerID, messageID,
+		)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return peerIDs, nil
 }
 
 // GetUserCount returns the number of active users.
@@ -1543,29 +1722,6 @@ func (database *Database) GetOperCount(
 	return count, nil
 }
 
-// ClientCountForSession returns the number of clients
-// belonging to a session.
-func (database *Database) ClientCountForSession(
-	ctx context.Context,
-	sessionID int64,
-) (int64, error) {
-	var count int64
-
-	err := database.conn.QueryRowContext(
-		ctx,
-		`SELECT COUNT(*) FROM clients
-		 WHERE session_id = ?`,
-		sessionID,
-	).Scan(&count)
-	if err != nil {
-		return 0, fmt.Errorf(
-			"client count for session: %w", err,
-		)
-	}
-
-	return count, nil
-}
-
 // DeleteStaleUsers removes clients not seen since the
 // cutoff and cleans up orphaned users (sessions).
 func (database *Database) DeleteStaleUsers(

internal/db/queries_test.go

@@ -2,6 +2,7 @@ package db_test
 
 import (
 	"encoding/json"
+	"slices"
 	"testing"
 
 	"sneak.berlin/go/neoirc/internal/db"
@@ -411,7 +412,7 @@ func TestJoinAndPart(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = database.PartChannel(ctx, chID, sid)
+	err = database.PartChannelAndDeleteIfEmpty(ctx, chID, sid)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -422,7 +423,10 @@ func TestJoinAndPart(t *testing.T) {
 	}
 }
 
-func TestDeleteChannelIfEmpty(t *testing.T) {
+// TestPartChannelAndDeleteIfEmpty checks that a part keeps
+// a channel another session is still on and deletes the
+// channel when its last member parts.
+func TestPartChannelAndDeleteIfEmpty(t *testing.T) {
 	t.Parallel()
 
 	database := setupTestDB(t)
@@ -435,22 +439,34 @@ func TestDeleteChannelIfEmpty(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	sid, _, _, err := database.CreateSession(ctx, "temp", "", "", "")
+	nicks := []string{"temp1", "temp2"}
+	sids := make([]int64, 0, len(nicks))
+
+	for _, nick := range nicks {
+		sid, _, _, createErr := database.CreateSession(ctx, nick, "", "", "")
+		if createErr != nil {
+			t.Fatal(createErr)
+		}
+
+		err = database.JoinChannel(ctx, chID, sid)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		sids = append(sids, sid)
+	}
+
+	err = database.PartChannelAndDeleteIfEmpty(ctx, chID, sids[0])
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	err = database.JoinChannel(ctx, chID, sid)
+	_, err = database.GetChannelByName(ctx, "#empty")
 	if err != nil {
-		t.Fatal(err)
+		t.Fatalf("channel deleted while a member is on it: %v", err)
 	}
 
-	err = database.PartChannel(ctx, chID, sid)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	err = database.DeleteChannelIfEmpty(ctx, chID)
+	err = database.PartChannelAndDeleteIfEmpty(ctx, chID, sids[1])
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -777,6 +793,11 @@ func TestGetHistoryReusedNames(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	err = database.JoinChannel(ctx, oldChID, oldSID)
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	body := json.RawMessage(`["hi"]`)
 
 	_, _, err = database.InsertMessage(
@@ -799,12 +820,9 @@ func TestGetHistoryReusedNames(t *testing.T) {
 		t.Fatal("expected one DM and one channel message")
 	}
 
-	err = database.DeleteSession(ctx, oldSID)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	err = database.DeleteChannelIfEmpty(ctx, oldChID)
+	// The session is the channel's only member, so quitting
+	// deletes the channel too.
+	_, err = database.QuitSession(ctx, oldSID, "reused", nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -826,34 +844,42 @@ func TestGetHistoryReusedNames(t *testing.T) {
 	}
 }
 
-func TestDeleteSession(t *testing.T) {
+// TestQuitSession checks that QuitSession deletes the
+// session, keeps a channel another session is still on,
+// deletes a channel it leaves with no members, and queues
+// one QUIT to the other session; and that a second call for
+// the same session changes nothing.
+func TestQuitSession(t *testing.T) {
 	t.Parallel()
 
 	database := setupTestDB(t)
 	ctx := t.Context()
 
-	sid, _, _, err := database.CreateSession(
-		ctx, "deleteme", "", "", "",
+	sid, sharedID, _ := createSessionWithChannels(
+		t, database, "deleteme", "#shared", "#alone",
+	)
+
+	peerSID, peerClientID, _, err := database.CreateSession(
+		ctx, "peer", "", "", "",
 	)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	chID, err := database.GetOrCreateChannel(
-		ctx, "#delchan",
-	)
+	err = database.JoinChannel(ctx, sharedID, peerSID)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	err = database.JoinChannel(ctx, chID, sid)
+	body := json.RawMessage(`["bye"]`)
+
+	peerIDs, err := database.QuitSession(ctx, sid, "deleteme", body)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	err = database.DeleteSession(ctx, sid)
-	if err != nil {
-		t.Fatal(err)
+	if !slices.Equal(peerIDs, []int64{peerSID}) {
+		t.Fatalf("got peers %v, want [%d]", peerIDs, peerSID)
 	}
 
 	_, err = database.GetSessionByNick(ctx, "deleteme")
@@ -861,9 +887,80 @@ func TestDeleteSession(t *testing.T) {
 		t.Fatal("session should be deleted")
 	}
 
-	ids, _ := database.GetChannelMemberIDs(ctx, chID)
-	if len(ids) != 0 {
-		t.Fatal("expected no members after deletion")
+	ids, _ := database.GetChannelMemberIDs(ctx, sharedID)
+	if !slices.Equal(ids, []int64{peerSID}) {
+		t.Fatalf("#shared members: got %v, want [%d]", ids, peerSID)
+	}
+
+	_, err = database.GetChannelByName(ctx, "#alone")
+	if err == nil {
+		t.Fatal("expected the emptied channel to be deleted")
+	}
+
+	checkOneQuitQueued(t, database, peerClientID, "deleteme")
+
+	peerIDs, err = database.QuitSession(ctx, sid, "deleteme", body)
+	if err != nil || len(peerIDs) != 0 {
+		t.Fatalf("second quit: got %v, %v; want none", peerIDs, err)
+	}
+
+	checkOneQuitQueued(t, database, peerClientID, "deleteme")
+}
+
+// checkOneQuitQueued fails the test unless the queue of the
+// client clientID holds one message, a QUIT from nick.
+func checkOneQuitQueued(
+	t *testing.T,
+	database *db.Database,
+	clientID int64,
+	nick string,
+) {
+	t.Helper()
+
+	msgs, _, err := database.PollMessages(t.Context(), clientID, 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(msgs) != 1 || msgs[0].Command != "QUIT" || msgs[0].From != nick {
+		t.Fatalf("queue: got %v, want one QUIT from %s", msgs, nick)
+	}
+}
+
+// TestLogoutClientFailureKeepsClient checks that when the
+// session of a logout's last client cannot be deleted,
+// LogoutClient fails and the client stays logged in.
+func TestLogoutClientFailureKeepsClient(t *testing.T) {
+	t.Parallel()
+
+	database := setupTestDB(t)
+	ctx := t.Context()
+
+	sid, clientID, token, err := database.CreateSession(
+		ctx, "stuck", "", "", "",
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = database.GetDB().ExecContext(ctx,
+		`CREATE TRIGGER keep_sessions
+		 BEFORE DELETE ON sessions
+		 BEGIN
+		   SELECT RAISE(ABORT, 'sessions are kept');
+		 END`)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = database.LogoutClient(ctx, sid, clientID, "stuck", nil)
+	if err == nil {
+		t.Fatal("logout succeeded although the session was kept")
+	}
+
+	_, _, nick, err := database.GetSessionByToken(ctx, token)
+	if err != nil || nick != "stuck" {
+		t.Fatalf("client gone after a failed logout: %q, %v", nick, err)
 	}
 }