Broadcast a session's QUIT once when its clients leave together (closes #154)
check / check (push) Waiting to run
check / check (push) Waiting to run
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
This commit is contained in:
+244
-88
@@ -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(
|
||||
|
||||
Reference in New Issue
Block a user