feat: implement queue pruning and message rotation
Enforce QUEUE_MAX_AGE and MAX_HISTORY config values that previously existed but were not applied. The existing cleanup loop now also: - Prunes client_queues entries older than QUEUE_MAX_AGE (default 48h) - Rotates messages per target (channel/DM) beyond MAX_HISTORY (default 10000) - Removes orphaned messages no longer referenced by any client queue closes #40
This commit is contained in:
@@ -1096,3 +1096,128 @@ func (database *Database) GetSessionCreatedAt(
|
||||
|
||||
return createdAt, nil
|
||||
}
|
||||
|
||||
// PruneOldQueueEntries deletes client_queues rows older
|
||||
// than cutoff and returns the number of rows removed.
|
||||
func (database *Database) PruneOldQueueEntries(
|
||||
ctx context.Context,
|
||||
cutoff time.Time,
|
||||
) (int64, error) {
|
||||
res, err := database.conn.ExecContext(ctx,
|
||||
"DELETE FROM client_queues WHERE created_at < ?",
|
||||
cutoff,
|
||||
)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf(
|
||||
"prune old queue entries: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
deleted, _ := res.RowsAffected()
|
||||
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
// PruneOrphanedMessages deletes messages that are no
|
||||
// longer referenced by any client_queues row and returns
|
||||
// the number of rows removed.
|
||||
func (database *Database) PruneOrphanedMessages(
|
||||
ctx context.Context,
|
||||
) (int64, error) {
|
||||
res, err := database.conn.ExecContext(ctx,
|
||||
`DELETE FROM messages WHERE id NOT IN
|
||||
(SELECT DISTINCT message_id
|
||||
FROM client_queues)`,
|
||||
)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf(
|
||||
"prune orphaned messages: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
deleted, _ := res.RowsAffected()
|
||||
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
// RotateChannelMessages enforces MAX_HISTORY per channel
|
||||
// by deleting the oldest messages beyond the limit for
|
||||
// each msg_to target. Returns the total number of rows
|
||||
// removed.
|
||||
func (database *Database) RotateChannelMessages(
|
||||
ctx context.Context,
|
||||
maxHistory int,
|
||||
) (int64, error) {
|
||||
if maxHistory <= 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// Find distinct targets that have messages.
|
||||
rows, err := database.conn.QueryContext(ctx,
|
||||
`SELECT msg_to, COUNT(*) AS cnt
|
||||
FROM messages
|
||||
WHERE msg_to != ''
|
||||
GROUP BY msg_to
|
||||
HAVING cnt > ?`,
|
||||
maxHistory,
|
||||
)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf(
|
||||
"list targets for rotation: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
type targetCount struct {
|
||||
target string
|
||||
count int64
|
||||
}
|
||||
|
||||
var targets []targetCount
|
||||
|
||||
for rows.Next() {
|
||||
var entry targetCount
|
||||
|
||||
err = rows.Scan(&entry.target, &entry.count)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf(
|
||||
"scan target count: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
targets = append(targets, entry)
|
||||
}
|
||||
|
||||
err = rows.Err()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("rows error: %w", err)
|
||||
}
|
||||
|
||||
var totalDeleted int64
|
||||
|
||||
for _, entry := range targets {
|
||||
res, delErr := database.conn.ExecContext(ctx,
|
||||
`DELETE FROM messages
|
||||
WHERE msg_to = ?
|
||||
AND id NOT IN (
|
||||
SELECT id FROM messages
|
||||
WHERE msg_to = ?
|
||||
ORDER BY id DESC
|
||||
LIMIT ?
|
||||
)`,
|
||||
entry.target, entry.target, maxHistory,
|
||||
)
|
||||
if delErr != nil {
|
||||
return totalDeleted, fmt.Errorf(
|
||||
"rotate messages for %s: %w",
|
||||
entry.target, delErr,
|
||||
)
|
||||
}
|
||||
|
||||
deleted, _ := res.RowsAffected()
|
||||
totalDeleted += deleted
|
||||
}
|
||||
|
||||
return totalDeleted, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user