internal/store/postgres/broadcast_delivery.go
Rama main · réplica de lectura del repositorio original.
package postgres
import (
"context"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5"
"telesrv/internal/domain"
"telesrv/internal/store"
"telesrv/internal/store/postgres/sqlcgen"
)
var _ store.BroadcastDeliveryStore = (*MessageStore)(nil)
// DeliverBroadcastRecipient turns one claimed recipient into one incoming-only
// 777000 message. The recipient receipt is closed in the same transaction as
// the message box, PTS event and dispatch outbox row.
func (s *MessageStore) DeliverBroadcastRecipient(ctx context.Context, claim store.BroadcastRecipientClaim) (domain.Message, error) {
if claim.RecipientID <= 0 || claim.BroadcastID <= 0 || claim.UserID <= 0 || claim.LeaseToken == "" {
return domain.Message{}, domain.ErrBroadcastInvalid
}
beginner, ok := s.db.(txBeginner)
if !ok {
return domain.Message{}, fmt.Errorf("deliver broadcast recipient: database does not support transactions")
}
tx, err := beginner.Begin(ctx)
if err != nil {
return domain.Message{}, fmt.Errorf("begin broadcast delivery: %w", err)
}
committed := false
defer func() {
if !committed {
_ = tx.Rollback(ctx)
}
}()
var userID int64
var message, entitiesJSON, status, leaseToken string
if err := tx.QueryRow(ctx, `
SELECT r.user_id, b.message, b.entities::text, r.status, r.lease_token
FROM broadcast_recipients r
JOIN broadcasts b ON b.id = r.broadcast_id
WHERE r.id = $1 AND r.broadcast_id = $2
FOR UPDATE OF r`, claim.RecipientID, claim.BroadcastID).Scan(&userID, &message, &entitiesJSON, &status, &leaseToken); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return domain.Message{}, domain.ErrBroadcastLeaseLost
}
return domain.Message{}, fmt.Errorf("lock broadcast recipient: %w", err)
}
if userID != claim.UserID || status != string(domain.BroadcastRecipientProcessing) || leaseToken != claim.LeaseToken {
return domain.Message{}, domain.ErrBroadcastLeaseLost
}
if err := lockUsersForUpdate(ctx, tx, userID); err != nil {
return domain.Message{}, fmt.Errorf("lock broadcast recipient user: %w", err)
}
var eligible bool
if err := tx.QueryRow(ctx, `
SELECT EXISTS (
SELECT 1 FROM users
WHERE id = $1 AND NOT is_bot AND deleted_at IS NULL
)`, userID).Scan(&eligible); err != nil {
return domain.Message{}, fmt.Errorf("check broadcast recipient: %w", err)
}
if !eligible || domain.IsSystemUserID(userID) {
return domain.Message{}, domain.ErrBroadcastRecipientInvalid
}
entities, err := decodeMessageEntities(entitiesJSON)
if err != nil {
return domain.Message{}, fmt.Errorf("decode broadcast entities: %w", err)
}
date := int(time.Now().Unix())
base := domain.Message{
OwnerUserID: userID,
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID},
From: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID},
Date: date,
Body: message,
Entities: entities,
}
if err := ensureOfficialSystemUserWithDB(ctx, tx, base); err != nil {
return domain.Message{}, err
}
qtx := sqlcgen.New(tx)
pm, err := qtx.CreatePrivateMessage(ctx, sqlcgen.CreatePrivateMessageParams{
SenderUserID: domain.OfficialSystemUserID,
RecipientUserID: userID,
RandomID: 0,
RequestFingerprint: []byte{},
RecipientDelivered: true,
MessageDate: int32(date),
Body: message,
EntitiesJson: []byte(entitiesJSON),
QuoteEntitiesJson: []byte("[]"),
MediaJson: []byte("{}"),
ReplyMarkupJson: []byte("{}"),
RichMessageJson: []byte("{}"),
})
if err != nil {
return domain.Message{}, fmt.Errorf("create broadcast private message: %w", err)
}
boxID, err := s.nextIncomingSystemBoxID(ctx, qtx, userID)
if err != nil {
return domain.Message{}, fmt.Errorf("allocate broadcast box id: %w", err)
}
if boxID <= 0 || boxID > domain.MaxMessageBoxID {
return domain.Message{}, fmt.Errorf("allocate broadcast box id: %d", boxID)
}
pts, err := s.reservePts(ctx, tx, userID)
if err != nil {
return domain.Message{}, fmt.Errorf("allocate broadcast pts: %w", err)
}
boxRow, err := qtx.CreateMessageBox(ctx, sqlcgen.CreateMessageBoxParams{
OwnerUserID: userID,
BoxID: int32(boxID),
PrivateMessageID: pm.ID,
MessageSenderID: domain.OfficialSystemUserID,
PeerType: string(domain.PeerTypeUser),
PeerID: domain.OfficialSystemUserID,
FromUserID: domain.OfficialSystemUserID,
MessageDate: int32(date),
Outgoing: false,
Body: message,
EntitiesJson: []byte(entitiesJSON),
QuoteEntitiesJson: []byte("[]"),
Pts: int32(pts),
MediaJson: []byte("{}"),
ReplyMarkupJson: []byte("{}"),
RichMessageJson: []byte("{}"),
})
if err != nil {
return domain.Message{}, fmt.Errorf("create broadcast recipient box: %w", err)
}
msg, err := messageFromBoxRow(boxRow)
if err != nil {
return domain.Message{}, err
}
if err := qtx.UpsertInboxDialog(ctx, sqlcgen.UpsertInboxDialogParams{
UserID: userID,
PeerType: string(domain.PeerTypeUser),
PeerID: domain.OfficialSystemUserID,
TopMessageID: int32(msg.ID),
TopMessageDate: int32(msg.Date),
}); err != nil {
return domain.Message{}, fmt.Errorf("upsert broadcast dialog: %w", err)
}
if err := appendNewMessageEvent(ctx, qtx, msg); err != nil {
return domain.Message{}, err
}
if err := enqueueDispatch(ctx, qtx, sqlcgen.EnqueueDispatchParams{
TargetUserID: userID,
Pts: int32(msg.Pts),
EventType: string(domain.UpdateEventNewMessage),
ExcludeAuthKeyID: 0,
ExcludeSessionID: 0,
}); err != nil {
return domain.Message{}, fmt.Errorf("enqueue broadcast dispatch: %w", err)
}
if tag, err := tx.Exec(ctx, `
UPDATE private_messages
SET recipient_box_id = $3, recipient_pts = $4
WHERE sender_user_id = $1 AND id = $2
AND recipient_delivered
AND recipient_box_id = 0 AND recipient_pts = 0`,
domain.OfficialSystemUserID, pm.ID, msg.ID, msg.Pts); err != nil {
return domain.Message{}, fmt.Errorf("save broadcast private receipt: %w", err)
} else if tag.RowsAffected() != 1 {
return domain.Message{}, fmt.Errorf("save broadcast private receipt: logical message disappeared")
}
if 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, msg.UID, msg.ID, msg.Pts); err != nil {
return domain.Message{}, fmt.Errorf("complete broadcast recipient: %w", err)
} else if tag.RowsAffected() != 1 {
return domain.Message{}, domain.ErrBroadcastLeaseLost
}
if tag, err := tx.Exec(ctx, `
UPDATE broadcasts SET sent_count = sent_count + 1 WHERE id = $1`, claim.BroadcastID); err != nil {
return domain.Message{}, fmt.Errorf("advance broadcast sent count: %w", err)
} else if tag.RowsAffected() != 1 {
return domain.Message{}, fmt.Errorf("advance broadcast sent count: campaign disappeared")
}
if err := tx.Commit(ctx); err != nil {
return domain.Message{}, fmt.Errorf("commit broadcast delivery: %w", err)
}
committed = true
return msg, nil
}