mirror of
https://github.com/binwiederhier/ntfy.git
synced 2026-11-10 14:44:04 +08:00
binwiederhier
c94b70000a
Release notes: main moved the v2.32.0 notes to the released section, so this PR's bullet goes under v2.33.0's Bug fixes + maintenance. The test's one-line t.Cleanup literals are spread over lines.
188 lines
11 KiB
Go
188 lines
11 KiB
Go
package message
|
|
|
|
import (
|
|
"database/sql"
|
|
"time"
|
|
|
|
"heckel.io/ntfy/v2/db"
|
|
"heckel.io/ntfy/v2/db/schema"
|
|
"heckel.io/ntfy/v2/model"
|
|
)
|
|
|
|
// PostgreSQL runtime query constants
|
|
const (
|
|
// postgresInsertMessagesQuery writes a whole batch in one statement with one array parameter
|
|
// per column, in the order of insertMessageArgs. The statement text is the same for every batch
|
|
// size: pgx prepares each distinct text server-side and keeps up to 512 per connection, and a
|
|
// multi-row INSERT per batch size left hundreds of multi-MB plans in every pooled backend.
|
|
postgresInsertMessagesQuery = `
|
|
INSERT INTO message (mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, attachment_deleted, sender, user_id, content_type, encoding, published)
|
|
SELECT * FROM unnest($1::TEXT[], $2::TEXT[], $3::BIGINT[], $4::TEXT[], $5::BIGINT[], $6::TEXT[], $7::TEXT[], $8::TEXT[], $9::INT[], $10::TEXT[], $11::TEXT[], $12::TEXT[], $13::TEXT[], $14::TEXT[], $15::TEXT[], $16::BIGINT[], $17::BIGINT[], $18::TEXT[], $19::BOOLEAN[], $20::TEXT[], $21::TEXT[], $22::TEXT[], $23::TEXT[], $24::BOOLEAN[])
|
|
`
|
|
postgresSelectScheduledMessageIDsBySeqIDQuery = `SELECT mid FROM message WHERE topic = $1 AND sequence_id = $2 AND published = FALSE`
|
|
postgresDeleteScheduledBySequenceIDQuery = `DELETE FROM message WHERE topic = $1 AND sequence_id = $2 AND published = FALSE`
|
|
postgresUpdateMessagesForTopicExpiryQuery = `UPDATE message SET expires = $1 WHERE topic = $2`
|
|
postgresSelectMessagesByIDQuery = `
|
|
SELECT mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user_id, content_type, encoding
|
|
FROM message
|
|
WHERE mid = $1
|
|
`
|
|
postgresSelectMessagesSinceTimeQuery = `
|
|
SELECT mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user_id, content_type, encoding
|
|
FROM message
|
|
WHERE topic = $1 AND time >= $2 AND published = TRUE
|
|
ORDER BY time DESC, id DESC
|
|
`
|
|
postgresSelectMessagesSinceTimeIncludeScheduledQuery = `
|
|
SELECT mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user_id, content_type, encoding
|
|
FROM message
|
|
WHERE topic = $1 AND time >= $2
|
|
ORDER BY time DESC, id DESC
|
|
`
|
|
postgresSelectMessageRowIDQuery = `SELECT id, time FROM message WHERE mid = $1 LIMIT 1`
|
|
// postgresSelectMessagesSinceIDQuery replays after a since=<mid> marker (see Cache.MessagesAfterMarker):
|
|
// rows published in the marker's second ($3) or later, except the marker ($2), and rows written after
|
|
// the marker but published before it. Each part starts its index scan at the marker; as one OR, every
|
|
// row of the topic was read (incident 2026-10-16). "time + 0" keeps the second part off the time index,
|
|
// whose scan would start at the topic's oldest row.
|
|
postgresSelectMessagesSinceIDQuery = `
|
|
SELECT mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user_id, content_type, encoding
|
|
FROM (
|
|
SELECT * FROM message WHERE topic = $1 AND published = TRUE AND time >= $3 AND id <> $2
|
|
UNION ALL
|
|
SELECT * FROM message WHERE topic = $1 AND published = TRUE AND id > $2 AND time + 0 < $3
|
|
) m
|
|
ORDER BY time DESC, id DESC
|
|
`
|
|
postgresSelectMessagesSinceIDIncludeScheduledQuery = `
|
|
SELECT mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user_id, content_type, encoding
|
|
FROM (
|
|
SELECT * FROM message WHERE topic = $1 AND published = TRUE AND time >= $3 AND id <> $2
|
|
UNION ALL
|
|
SELECT * FROM message WHERE topic = $1 AND published = TRUE AND id > $2 AND time + 0 < $3
|
|
UNION ALL
|
|
SELECT * FROM message WHERE topic = $1 AND published = FALSE
|
|
) m
|
|
ORDER BY time DESC, id DESC
|
|
`
|
|
postgresSelectRecentRowIDsQuery = `SELECT id FROM message ORDER BY id DESC LIMIT $1`
|
|
// postgresSelectMessagesAfterRowIDQuery reads rows in write order: those after the cursor ($1),
|
|
// plus the skipped row ids ($2, a JSON array) that were not committed yet when the cursor passed.
|
|
// The parts are separate so that the primary key checks every condition, also in the plan made
|
|
// without the parameter values; as one OR, that plan read the whole table.
|
|
postgresSelectMessagesAfterRowIDQuery = `
|
|
SELECT mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user_id, content_type, encoding, id, published
|
|
FROM (
|
|
SELECT * FROM message WHERE id > $1
|
|
UNION ALL
|
|
SELECT * FROM message WHERE id = ANY(ARRAY(SELECT jsonb_array_elements_text($2::JSONB)::BIGINT)) AND id <= $1
|
|
) m
|
|
ORDER BY id
|
|
LIMIT $3
|
|
`
|
|
postgresSelectMessagesLatestQuery = `
|
|
SELECT mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user_id, content_type, encoding
|
|
FROM message
|
|
WHERE topic = $1 AND published = TRUE
|
|
ORDER BY time DESC, id DESC
|
|
LIMIT 1
|
|
`
|
|
postgresSelectMessagesDueQuery = `
|
|
SELECT mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user_id, content_type, encoding
|
|
FROM message
|
|
WHERE time <= $1 AND published = FALSE
|
|
ORDER BY time, id
|
|
`
|
|
postgresUpdateMessagePublishedQuery = `UPDATE message SET published = TRUE WHERE mid = $1`
|
|
// Planner estimate, since a COUNT(*) scans the whole table; reltuples is -1 if never analyzed
|
|
postgresSelectMessagesCountQuery = `
|
|
SELECT CASE WHEN reltuples < 0 THEN (SELECT COUNT(*) FROM message) ELSE reltuples::BIGINT END
|
|
FROM pg_class
|
|
WHERE oid = 'message'::regclass
|
|
`
|
|
postgresSelectTopicsQuery = `SELECT topic FROM message GROUP BY topic`
|
|
|
|
postgresDeleteExpiredMessagesQuery = `DELETE FROM message WHERE mid IN (SELECT mid FROM message WHERE expires <= $1 AND published = TRUE LIMIT $2)`
|
|
postgresMarkExpiredAttachmentsDeletedQuery = `UPDATE message SET attachment_deleted = TRUE WHERE mid IN (SELECT mid FROM message WHERE attachment_expires > 0 AND attachment_expires <= $1 AND attachment_deleted = FALSE LIMIT $2)`
|
|
postgresSelectAttachmentsSizeBySenderQuery = `SELECT COALESCE(SUM(attachment_size), 0) FROM message WHERE user_id = '' AND sender = $1 AND attachment_expires >= $2`
|
|
postgresSelectAttachmentsSizeByUserIDQuery = `SELECT COALESCE(SUM(attachment_size), 0) FROM message WHERE user_id = $1 AND attachment_expires >= $2`
|
|
postgresSelectAttachmentsWithSizesQuery = `SELECT mid, attachment_size FROM message WHERE attachment_expires > $1 AND attachment_deleted = FALSE`
|
|
|
|
postgresSelectStatsQuery = `SELECT value FROM message_stats WHERE key = 'messages'`
|
|
// postgresClaimMessagesDueQuery claims due rows in one statement: the inner SELECT locks what
|
|
// it picks and skips what another node is claiming, so concurrent senders get disjoint sets.
|
|
// Unclaimed rows have claimed_at 0, so the cutoff ($2) matches them too.
|
|
postgresClaimMessagesDueQuery = `
|
|
WITH due AS (
|
|
SELECT id FROM message
|
|
WHERE time <= $1 AND published = FALSE AND claimed_at <= $2
|
|
ORDER BY time, id
|
|
LIMIT $3
|
|
FOR UPDATE SKIP LOCKED
|
|
)
|
|
UPDATE message SET claimed_at = $4 WHERE id IN (SELECT id FROM due)
|
|
RETURNING mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user_id, content_type, encoding
|
|
`
|
|
|
|
postgresUpdateStatsQuery = `UPDATE message_stats SET value = value + $1 WHERE key = 'messages'`
|
|
postgresUpdateMessageTimeQuery = `UPDATE message SET time = $1 WHERE mid = $2`
|
|
)
|
|
|
|
var postgresQueries = queries{
|
|
insertMessages: postgresInsertMessages,
|
|
selectScheduledMessageIDsBySeqID: postgresSelectScheduledMessageIDsBySeqIDQuery,
|
|
deleteScheduledBySequenceID: postgresDeleteScheduledBySequenceIDQuery,
|
|
updateMessagesForTopicExpiry: postgresUpdateMessagesForTopicExpiryQuery,
|
|
selectMessagesByID: postgresSelectMessagesByIDQuery,
|
|
selectMessagesSinceTime: postgresSelectMessagesSinceTimeQuery,
|
|
selectMessagesSinceTimeScheduled: postgresSelectMessagesSinceTimeIncludeScheduledQuery,
|
|
selectMessageRowID: postgresSelectMessageRowIDQuery,
|
|
selectMessagesSinceID: postgresSelectMessagesSinceIDQuery,
|
|
selectMessagesSinceIDScheduled: postgresSelectMessagesSinceIDIncludeScheduledQuery,
|
|
selectRecentRowIDs: postgresSelectRecentRowIDsQuery,
|
|
selectMessagesAfterRowID: postgresSelectMessagesAfterRowIDQuery,
|
|
selectMessagesLatest: postgresSelectMessagesLatestQuery,
|
|
selectMessagesDue: postgresSelectMessagesDueQuery,
|
|
claimMessagesDue: postgresClaimMessagesDueQuery,
|
|
deleteExpiredMessages: postgresDeleteExpiredMessagesQuery,
|
|
updateMessagePublished: postgresUpdateMessagePublishedQuery,
|
|
selectMessagesCount: postgresSelectMessagesCountQuery,
|
|
selectTopics: postgresSelectTopicsQuery,
|
|
markExpiredAttachmentsDeleted: postgresMarkExpiredAttachmentsDeletedQuery,
|
|
selectAttachmentsSizeBySender: postgresSelectAttachmentsSizeBySenderQuery,
|
|
selectAttachmentsSizeByUserID: postgresSelectAttachmentsSizeByUserIDQuery,
|
|
selectAttachmentsWithSizes: postgresSelectAttachmentsWithSizesQuery,
|
|
selectStats: postgresSelectStatsQuery,
|
|
updateStats: postgresUpdateStatsQuery,
|
|
updateMessageTime: postgresUpdateMessageTimeQuery,
|
|
}
|
|
|
|
// NewPostgresStore creates a new PostgreSQL-backed message cache store using an existing database connection pool.
|
|
func NewPostgresStore(d *db.DB, batchSize int, batchTimeout time.Duration) (*Cache, error) {
|
|
if err := schema.Migrate(d.Primary(), schema.Postgres, schemaStore, postgresCreateTables, postgresMigrations); err != nil {
|
|
return nil, err
|
|
}
|
|
return newCache(d, postgresQueries, nil, batchSize, batchTimeout, false), nil
|
|
}
|
|
|
|
// postgresInsertMessages transposes the batch into one array per column and writes it with
|
|
// postgresInsertMessagesQuery in a single round trip
|
|
func postgresInsertMessages(tx *sql.Tx, ms []*model.Message) error {
|
|
columns := make([][]any, insertMessageColumns)
|
|
for _, m := range ms {
|
|
args, err := insertMessageArgs(m)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for i, arg := range args {
|
|
columns[i] = append(columns[i], arg)
|
|
}
|
|
}
|
|
params := make([]any, 0, len(columns))
|
|
for _, column := range columns {
|
|
params = append(params, column)
|
|
}
|
|
_, err := tx.Exec(postgresInsertMessagesQuery, params...)
|
|
return err
|
|
}
|