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 reports whether it
deleted the session; only that caller sends the QUIT. db.LogoutClient
deletes the client and counts the rest in the same transaction.

Model: opus-5-5

internal/db/queries.go

@@ -905,7 +905,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 +947,6 @@ func (database *Database) PartChannelAndDeleteIfEmpty(
 		)
 	}
 
-	err = transaction.Commit()
-	if err != nil {
-		return fmt.Errorf("commit part: %w", err)
-	}
-
 	return nil
 }
 
@@ -1487,21 +1504,150 @@ func (database *Database) DeleteSession(
 	return nil
 }
 
-// DeleteClient removes a single client record by ID.
-func (database *Database) DeleteClient(
+// QuitSession parts a session from its channels, deleting
+// each channel that this leaves with no members, and
+// deletes the session with its clients, in one
+// transaction. It returns the other sessions that were on
+// those channels, and whether this call deleted the
+// session: of several calls for one session at once, only
+// one finds it still there, and the others change nothing.
+func (database *Database) QuitSession(
 	ctx context.Context,
-	clientID int64,
-) error {
-	_, err := database.conn.ExecContext(
-		ctx,
-		"DELETE FROM clients WHERE id = ?",
-		clientID,
-	)
+	sessionID int64,
+) ([]int64, bool, error) {
+	transaction, err := database.conn.BeginTx(ctx, nil)
 	if err != nil {
-		return fmt.Errorf("delete client: %w", err)
+		return nil, false, fmt.Errorf("begin quit: %w", err)
 	}
 
-	return nil
+	defer func() { _ = transaction.Rollback() }()
+
+	peerIDs, deleted, err := quitSessionInTx(
+		ctx, transaction, sessionID,
+	)
+	if err != nil {
+		return nil, false, err
+	}
+
+	err = transaction.Commit()
+	if err != nil {
+		return nil, false, fmt.Errorf("commit quit: %w", err)
+	}
+
+	return peerIDs, deleted, nil
+}
+
+// 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 nothing and false if the session
+// keeps a client.
+func (database *Database) LogoutClient(
+	ctx context.Context,
+	sessionID, clientID int64,
+) ([]int64, bool, error) {
+	transaction, err := database.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, false, fmt.Errorf("begin logout: %w", err)
+	}
+
+	defer func() { _ = transaction.Rollback() }()
+
+	_, err = transaction.ExecContext(ctx,
+		"DELETE FROM clients WHERE id = ?", clientID)
+	if err != nil {
+		return nil, false, 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, false, fmt.Errorf("count clients: %w", err)
+	}
+
+	var (
+		peerIDs []int64
+		deleted bool
+	)
+
+	if remaining == 0 {
+		peerIDs, deleted, err = quitSessionInTx(
+			ctx, transaction, sessionID,
+		)
+		if err != nil {
+			return nil, false, err
+		}
+	}
+
+	err = transaction.Commit()
+	if err != nil {
+		return nil, false, fmt.Errorf("commit logout: %w", err)
+	}
+
+	return peerIDs, deleted, nil
+}
+
+// quitSessionInTx is QuitSession, writing through
+// transaction.
+func quitSessionInTx(
+	ctx context.Context,
+	transaction *sql.Tx,
+	sessionID int64,
+) ([]int64, bool, 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, false, fmt.Errorf("get channel peers: %w", err)
+	}
+
+	peerIDs, err := scanInt64s(rows)
+	if err != nil {
+		return nil, false, err
+	}
+
+	rows, err = transaction.QueryContext(ctx,
+		`SELECT channel_id FROM channel_members
+		 WHERE session_id = ?`, sessionID)
+	if err != nil {
+		return nil, false, fmt.Errorf(
+			"get session channel ids: %w", err,
+		)
+	}
+
+	channelIDs, err := scanInt64s(rows)
+	if err != nil {
+		return nil, false, err
+	}
+
+	for _, channelID := range channelIDs {
+		err = partChannelAndDeleteIfEmptyInTx(
+			ctx, transaction, channelID, sessionID,
+		)
+		if err != nil {
+			return nil, false, err
+		}
+	}
+
+	result, err := transaction.ExecContext(ctx,
+		"DELETE FROM sessions WHERE id = ?", sessionID)
+	if err != nil {
+		return nil, false, fmt.Errorf("delete session: %w", err)
+	}
+
+	deletedRows, err := result.RowsAffected()
+	if err != nil {
+		return nil, false, fmt.Errorf("delete session: %w", err)
+	}
+
+	return peerIDs, deletedRows > 0, nil
 }
 
 // GetUserCount returns the number of active users.
@@ -1543,29 +1689,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(