396 lines
14 KiB
Go
396 lines
14 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
|
|
"telesrv/internal/domain"
|
|
"telesrv/internal/store"
|
|
"telesrv/internal/store/postgres/sqlcgen"
|
|
)
|
|
|
|
// BroadcastStore persists system broadcast campaigns (see
|
|
// deploy/migrations/20260714003131_system_broadcasts.up.sql, extended by
|
|
// deploy/migrations/20260901000024_broadcast_lease_delivery_and_entities.up.sql
|
|
// with entities, incremental "all"-mode materialization and lease-based
|
|
// delivery claims).
|
|
type BroadcastStore struct {
|
|
db sqlcgen.DBTX
|
|
}
|
|
|
|
// NewBroadcastStore builds the store on a pgx pool or transaction.
|
|
func NewBroadcastStore(db sqlcgen.DBTX) *BroadcastStore {
|
|
return &BroadcastStore{db: db}
|
|
}
|
|
|
|
var _ store.BroadcastStore = (*BroadcastStore)(nil)
|
|
|
|
const eligibleBroadcastUsersSQL = `
|
|
FROM users
|
|
WHERE NOT is_bot
|
|
AND deleted_at IS NULL
|
|
AND id <> ALL($1::bigint[])`
|
|
|
|
// PreviewBroadcastRecipients counts (and, for "selected", validates) the
|
|
// intended recipient set without creating anything.
|
|
func (s *BroadcastStore) PreviewBroadcastRecipients(ctx context.Context, mode domain.BroadcastTargetMode, selectedUserIDs []int64) (int64, error) {
|
|
switch mode {
|
|
case domain.BroadcastTargetAll:
|
|
var count int64
|
|
if err := s.db.QueryRow(ctx, `SELECT count(*) `+eligibleBroadcastUsersSQL, domain.SystemUserIDs()).Scan(&count); err != nil {
|
|
return 0, fmt.Errorf("count broadcast recipients: %w", err)
|
|
}
|
|
if count == 0 {
|
|
return 0, domain.ErrBroadcastNoRecipients
|
|
}
|
|
return count, nil
|
|
case domain.BroadcastTargetSelected:
|
|
return validateSelectedBroadcastUsers(ctx, s.db, selectedUserIDs)
|
|
default:
|
|
return 0, domain.ErrBroadcastInvalid
|
|
}
|
|
}
|
|
|
|
func validateSelectedBroadcastUsers(ctx context.Context, db sqlcgen.DBTX, selectedUserIDs []int64) (int64, error) {
|
|
if len(selectedUserIDs) == 0 {
|
|
return 0, domain.ErrBroadcastNoRecipients
|
|
}
|
|
var count int64
|
|
if err := db.QueryRow(ctx, `
|
|
SELECT count(*)
|
|
FROM users
|
|
WHERE id = ANY($1::bigint[])
|
|
AND NOT is_bot
|
|
AND deleted_at IS NULL
|
|
AND id <> ALL($2::bigint[])`, selectedUserIDs, domain.SystemUserIDs()).Scan(&count); err != nil {
|
|
return 0, fmt.Errorf("validate broadcast recipients: %w", err)
|
|
}
|
|
if count != int64(len(selectedUserIDs)) {
|
|
return 0, domain.ErrBroadcastRecipientInvalid
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
// CreateBroadcast inserts the broadcast row. For "all" mode it only
|
|
// snapshots the current max eligible user id and target count -- recipient
|
|
// rows are inserted incrementally by MaterializeBroadcastRecipients, not
|
|
// here. For "selected" mode, whose recipient list is already bounded by
|
|
// domain.MaxBroadcastSelectedRecipients, every recipient row is inserted in
|
|
// the same transaction as the broadcast itself, deduplicating via
|
|
// ON CONFLICT DO NOTHING against the (broadcast_id, user_id) unique
|
|
// constraint (a hand-built selected list could otherwise carry a repeat).
|
|
func (s *BroadcastStore) CreateBroadcast(ctx context.Context, message string, entities []domain.MessageEntity, mode domain.BroadcastTargetMode, selectedUserIDs []int64, createdBy string) (domain.Broadcast, error) {
|
|
entitiesJSON, err := encodeMessageEntities(entities)
|
|
if err != nil {
|
|
return domain.Broadcast{}, fmt.Errorf("encode broadcast entities: %w", err)
|
|
}
|
|
var out domain.Broadcast
|
|
err = withTx(ctx, s.db, "create broadcast", func(tx pgx.Tx) error {
|
|
switch mode {
|
|
case domain.BroadcastTargetAll:
|
|
var maxUserID, count int64
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT COALESCE(max(id), 0), count(*) `+eligibleBroadcastUsersSQL, domain.SystemUserIDs()).Scan(&maxUserID, &count); err != nil {
|
|
return fmt.Errorf("snapshot broadcast recipients: %w", err)
|
|
}
|
|
if count == 0 {
|
|
return domain.ErrBroadcastNoRecipients
|
|
}
|
|
row := tx.QueryRow(ctx, `
|
|
INSERT INTO broadcasts (
|
|
message, entities, target_mode, snapshot_max_user_id, enumeration_done,
|
|
target_count, created_by
|
|
) VALUES ($1, $2::jsonb, 'all', $3, false, $4, $5)
|
|
RETURNING `+broadcastColumns,
|
|
message, string(entitiesJSON), maxUserID, count, createdBy,
|
|
)
|
|
if err := scanBroadcastRow(row, &out); err != nil {
|
|
return fmt.Errorf("insert all-user broadcast: %w", err)
|
|
}
|
|
case domain.BroadcastTargetSelected:
|
|
count, err := validateSelectedBroadcastUsers(ctx, tx, selectedUserIDs)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
row := tx.QueryRow(ctx, `
|
|
INSERT INTO broadcasts (
|
|
message, entities, target_mode, enumeration_done, target_count,
|
|
materialized_count, created_by
|
|
) VALUES ($1, $2::jsonb, 'selected', true, $3, $3, $4)
|
|
RETURNING `+broadcastColumns,
|
|
message, string(entitiesJSON), count, createdBy,
|
|
)
|
|
if err := scanBroadcastRow(row, &out); err != nil {
|
|
return fmt.Errorf("insert selected broadcast: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO broadcast_recipients (broadcast_id, user_id)
|
|
SELECT $1, user_id
|
|
FROM unnest($2::bigint[]) AS selected(user_id)
|
|
ON CONFLICT (broadcast_id, user_id) DO NOTHING`, out.ID, selectedUserIDs); err != nil {
|
|
return fmt.Errorf("insert selected broadcast recipients: %w", err)
|
|
}
|
|
default:
|
|
return domain.ErrBroadcastInvalid
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return domain.Broadcast{}, err
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// MaterializeBroadcastRecipients advances one all-user campaign with a
|
|
// single bounded keyset INSERT, picking whichever "all"-mode campaign still
|
|
// has enumeration left (oldest first) under FOR UPDATE SKIP LOCKED, so
|
|
// concurrent worker cycles never step on each other's progress.
|
|
func (s *BroadcastStore) MaterializeBroadcastRecipients(ctx context.Context, limit int) (int, error) {
|
|
if limit <= 0 || limit > 1000 {
|
|
limit = 100
|
|
}
|
|
var inserted int
|
|
err := s.db.QueryRow(ctx, `
|
|
WITH campaign AS (
|
|
SELECT id, snapshot_max_user_id, enumeration_cursor_user_id
|
|
FROM broadcasts
|
|
WHERE target_mode = 'all' AND NOT enumeration_done
|
|
ORDER BY id
|
|
FOR UPDATE SKIP LOCKED
|
|
LIMIT 1
|
|
), candidates AS (
|
|
SELECT u.id
|
|
FROM campaign c
|
|
JOIN LATERAL (
|
|
SELECT id
|
|
FROM users
|
|
WHERE id > c.enumeration_cursor_user_id
|
|
AND id <= c.snapshot_max_user_id
|
|
AND NOT is_bot
|
|
AND deleted_at IS NULL
|
|
AND id <> ALL($1::bigint[])
|
|
ORDER BY id
|
|
LIMIT $2
|
|
) u ON true
|
|
), materialized AS (
|
|
INSERT INTO broadcast_recipients (broadcast_id, user_id)
|
|
SELECT c.id, candidate.id
|
|
FROM campaign c
|
|
CROSS JOIN candidates candidate
|
|
ON CONFLICT (broadcast_id, user_id) DO NOTHING
|
|
RETURNING user_id
|
|
), progress AS (
|
|
UPDATE broadcasts b
|
|
SET enumeration_cursor_user_id = COALESCE((SELECT max(id) FROM candidates), b.snapshot_max_user_id),
|
|
enumeration_done = (SELECT count(*) FROM candidates) < $2,
|
|
materialized_count = b.materialized_count + (SELECT count(*) FROM materialized),
|
|
target_count = CASE
|
|
WHEN (SELECT count(*) FROM candidates) < $2
|
|
THEN b.materialized_count + (SELECT count(*) FROM materialized)
|
|
ELSE b.target_count
|
|
END
|
|
FROM campaign c
|
|
WHERE b.id = c.id
|
|
RETURNING b.id
|
|
)
|
|
SELECT count(*)::int FROM materialized`, domain.SystemUserIDs(), limit).Scan(&inserted)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return 0, nil
|
|
}
|
|
if err != nil {
|
|
return 0, fmt.Errorf("materialize broadcast recipients: %w", err)
|
|
}
|
|
return inserted, nil
|
|
}
|
|
|
|
// ClaimBroadcastRecipients atomically leases up to limit eligible rows
|
|
// (pending, or processing under an expired lease) to leaseToken, joining
|
|
// each claim with its broadcast's message and entities.
|
|
func (s *BroadcastStore) ClaimBroadcastRecipients(ctx context.Context, leaseToken string, limit int, lease time.Duration) ([]store.BroadcastRecipientClaim, error) {
|
|
if strings.TrimSpace(leaseToken) == "" || len(leaseToken) > 64 {
|
|
return nil, domain.ErrBroadcastInvalid
|
|
}
|
|
if limit <= 0 || limit > 500 {
|
|
limit = 50
|
|
}
|
|
leaseSeconds := int(lease / time.Second)
|
|
if leaseSeconds <= 0 || leaseSeconds > 3600 {
|
|
leaseSeconds = 30
|
|
}
|
|
rows, err := s.db.Query(ctx, `
|
|
WITH candidates AS (
|
|
SELECT id
|
|
FROM broadcast_recipients
|
|
WHERE (status = 'pending' AND next_attempt_at <= now())
|
|
OR (status = 'processing' AND lease_until <= now())
|
|
ORDER BY id
|
|
FOR UPDATE SKIP LOCKED
|
|
LIMIT $1
|
|
), claimed AS (
|
|
UPDATE broadcast_recipients r
|
|
SET status = 'processing',
|
|
attempts = attempts + 1,
|
|
lease_token = $2,
|
|
lease_until = now() + make_interval(secs => $3),
|
|
updated_at = now()
|
|
FROM candidates c
|
|
WHERE r.id = c.id
|
|
RETURNING r.id, r.broadcast_id, r.user_id, r.attempts
|
|
)
|
|
SELECT c.id, c.broadcast_id, c.user_id, c.attempts, b.message, b.entities::text
|
|
FROM claimed c
|
|
JOIN broadcasts b ON b.id = c.broadcast_id
|
|
ORDER BY c.id`, limit, leaseToken, leaseSeconds)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("claim broadcast recipients: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
out := make([]store.BroadcastRecipientClaim, 0, limit)
|
|
for rows.Next() {
|
|
var item store.BroadcastRecipientClaim
|
|
var entitiesJSON string
|
|
item.LeaseToken = leaseToken
|
|
if err := rows.Scan(&item.RecipientID, &item.BroadcastID, &item.UserID, &item.Attempts, &item.Message, &entitiesJSON); err != nil {
|
|
return nil, fmt.Errorf("scan broadcast recipient claim: %w", err)
|
|
}
|
|
entities, err := decodeMessageEntities(entitiesJSON)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("decode broadcast recipient claim entities: %w", err)
|
|
}
|
|
item.Entities = entities
|
|
out = append(out, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate broadcast recipient claims: %w", err)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// CompleteBroadcastRecipient closes a claimed row as delivered and advances
|
|
// its broadcast's sent_count in the same transaction.
|
|
func (s *BroadcastStore) CompleteBroadcastRecipient(ctx context.Context, claim store.BroadcastRecipientClaim, privateMessageID int64, messageBoxID int, pts int) error {
|
|
return withTx(ctx, s.db, "complete broadcast recipient", func(tx pgx.Tx) error {
|
|
tag, err := tx.Exec(ctx, `
|
|
UPDATE broadcast_recipients
|
|
SET status = 'sent', lease_token = '', lease_until = NULL,
|
|
last_error = '', private_message_id = $3, message_box_id = $4,
|
|
pts = $5, sent_at = now(), updated_at = now()
|
|
WHERE id = $1 AND status = 'processing' AND lease_token = $2`,
|
|
claim.RecipientID, claim.LeaseToken, privateMessageID, messageBoxID, pts)
|
|
if err != nil {
|
|
return fmt.Errorf("complete broadcast recipient: %w", err)
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return domain.ErrBroadcastLeaseLost
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
UPDATE broadcasts SET sent_count = sent_count + 1 WHERE id = $1`, claim.BroadcastID); err != nil {
|
|
return fmt.Errorf("advance broadcast sent count: %w", err)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// ReleaseBroadcastRecipient returns a claimed row to 'pending' with backoff,
|
|
// or to the terminal 'failed' once domain.MaxBroadcastRecipientAttempts is
|
|
// reached, advancing failed_count in that terminal case.
|
|
func (s *BroadcastStore) ReleaseBroadcastRecipient(ctx context.Context, claim store.BroadcastRecipientClaim, cause string) error {
|
|
if len(cause) > 500 {
|
|
cause = cause[:500]
|
|
}
|
|
_, err := s.db.Exec(ctx, `
|
|
WITH changed AS (
|
|
UPDATE broadcast_recipients
|
|
SET status = CASE WHEN attempts >= $3 THEN 'failed' ELSE 'pending' END,
|
|
next_attempt_at = CASE
|
|
WHEN attempts >= $3 THEN next_attempt_at
|
|
ELSE now() + make_interval(secs => LEAST(300, (1 << LEAST(attempts, 8))))
|
|
END,
|
|
lease_token = '',
|
|
lease_until = NULL,
|
|
last_error = $4,
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
AND status = 'processing'
|
|
AND lease_token = $2
|
|
RETURNING broadcast_id, status
|
|
)
|
|
UPDATE broadcasts b
|
|
SET failed_count = failed_count + 1
|
|
FROM changed c
|
|
WHERE b.id = c.broadcast_id AND c.status = 'failed'`, claim.RecipientID, claim.LeaseToken, domain.MaxBroadcastRecipientAttempts, cause)
|
|
if err != nil {
|
|
return fmt.Errorf("release broadcast recipient: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
const broadcastColumns = `
|
|
id, message, entities::text, target_mode, target_count, materialized_count,
|
|
sent_count, failed_count, enumeration_done, created_by, created_at`
|
|
|
|
func scanBroadcastRow(row interface{ Scan(...any) error }, item *domain.Broadcast) error {
|
|
var entitiesJSON string
|
|
if err := row.Scan(&item.ID, &item.Message, &entitiesJSON, &item.TargetMode, &item.TargetCount, &item.MaterializedCount,
|
|
&item.SentCount, &item.FailedCount, &item.EnumerationDone, &item.CreatedBy, &item.CreatedAt); err != nil {
|
|
return err
|
|
}
|
|
entities, err := decodeMessageEntities(entitiesJSON)
|
|
if err != nil {
|
|
return fmt.Errorf("decode broadcast entities: %w", err)
|
|
}
|
|
item.Entities = entities
|
|
return nil
|
|
}
|
|
|
|
// ListBroadcasts pages broadcasts newest-first.
|
|
func (s *BroadcastStore) ListBroadcasts(ctx context.Context, beforeID int64, limit int) ([]domain.Broadcast, bool, error) {
|
|
if limit <= 0 || limit > 200 {
|
|
limit = 50
|
|
}
|
|
rows, err := s.db.Query(ctx, `SELECT `+broadcastColumns+`
|
|
FROM broadcasts
|
|
WHERE $1::bigint = 0 OR id < $1
|
|
ORDER BY id DESC
|
|
LIMIT $2`, beforeID, limit+1)
|
|
if err != nil {
|
|
return nil, false, fmt.Errorf("list broadcasts: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
out := make([]domain.Broadcast, 0, limit+1)
|
|
for rows.Next() {
|
|
var item domain.Broadcast
|
|
if err := scanBroadcastRow(rows, &item); err != nil {
|
|
return nil, false, fmt.Errorf("scan broadcast: %w", err)
|
|
}
|
|
out = append(out, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, false, fmt.Errorf("iterate broadcasts: %w", err)
|
|
}
|
|
hasMore := len(out) > limit
|
|
if hasMore {
|
|
out = out[:limit]
|
|
}
|
|
return out, hasMore, nil
|
|
}
|
|
|
|
// BroadcastByID returns one broadcast.
|
|
func (s *BroadcastStore) BroadcastByID(ctx context.Context, id int64) (domain.Broadcast, bool, error) {
|
|
var item domain.Broadcast
|
|
err := scanBroadcastRow(s.db.QueryRow(ctx, `SELECT `+broadcastColumns+`
|
|
FROM broadcasts
|
|
WHERE id = $1`, id), &item)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return domain.Broadcast{}, false, nil
|
|
}
|
|
return domain.Broadcast{}, false, fmt.Errorf("get broadcast: %w", err)
|
|
}
|
|
return item, true, nil
|
|
}
|