owpengram-server/internal/store/postgres/channel_participant_queue.go

package postgres

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/jackc/pgx/v5"

	"telesrv/internal/domain"
	"telesrv/internal/store/postgres/sqlcgen"
)

// ChannelParticipantQueueStore is the PostgreSQL implementation of
// store.ChannelParticipantQueueStore. It reserves qts from the same
// secret_qts_watermarks table as EncryptedQueueStore (one qts sequence per
// device, shared across event kinds), keeping this table's own schema and
// queries entirely separate from the encrypted-message hot path.
type ChannelParticipantQueueStore struct {
	db sqlcgen.DBTX
}

// NewChannelParticipantQueueStore builds the store on a pgx pool or transaction.
func NewChannelParticipantQueueStore(db sqlcgen.DBTX) *ChannelParticipantQueueStore {
	return &ChannelParticipantQueueStore{db: db}
}

const channelParticipantEventColumns = `receiver_auth_key_id, qts, receiver_user_id, channel_id, actor_user_id,
date, previous_participant, new_participant`

func scanChannelParticipantEvent(row rowScanner) (domain.DeviceChannelParticipantEvent, error) {
	var ev domain.DeviceChannelParticipantEvent
	var previousRaw, newRaw []byte
	if err := row.Scan(
		&ev.ReceiverAuthKeyID, &ev.Qts, &ev.ReceiverUserID, &ev.ChannelID, &ev.ActorUserID,
		&ev.Date, &previousRaw, &newRaw,
	); err != nil {
		return domain.DeviceChannelParticipantEvent{}, err
	}
	if len(previousRaw) > 0 {
		if err := json.Unmarshal(previousRaw, &ev.Previous); err != nil {
			return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("unmarshal previous participant: %w", err)
		}
	}
	if len(newRaw) > 0 {
		if err := json.Unmarshal(newRaw, &ev.Participant); err != nil {
			return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("unmarshal new participant: %w", err)
		}
	}
	return ev, nil
}

func (s *ChannelParticipantQueueStore) begin(ctx context.Context, op string) (pgx.Tx, error) {
	beginner, ok := s.db.(txBeginner)
	if !ok {
		return nil, fmt.Errorf("%s: db does not support transactions", op)
	}
	tx, err := beginner.Begin(ctx)
	if err != nil {
		return nil, fmt.Errorf("begin %s: %w", op, err)
	}
	return tx, nil
}

// AppendChannelParticipantEvent reserves the receiving device's next qts
// (shared secret_qts_watermarks sequence) and writes the event, in one
// transaction so the device's qts never has a hole.
func (s *ChannelParticipantQueueStore) AppendChannelParticipantEvent(ctx context.Context, ev domain.DeviceChannelParticipantEvent) (domain.DeviceChannelParticipantEvent, error) {
	tx, err := s.begin(ctx, "append channel participant event")
	if err != nil {
		return domain.DeviceChannelParticipantEvent{}, err
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback(ctx)
		}
	}()

	var qts int
	if err := tx.QueryRow(ctx, `
INSERT INTO secret_qts_watermarks (auth_key_id, reserved_qts)
VALUES ($1, 1)
ON CONFLICT (auth_key_id) DO UPDATE SET reserved_qts = secret_qts_watermarks.reserved_qts + 1, updated_at = now()
RETURNING reserved_qts`, ev.ReceiverAuthKeyID).Scan(&qts); err != nil {
		return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("reserve device qts: %w", err)
	}
	ev.Qts = qts

	previousRaw, err := marshalJSON(ev.Previous, "null")
	if err != nil {
		return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("marshal previous participant: %w", err)
	}
	newRaw, err := marshalJSON(ev.Participant, "null")
	if err != nil {
		return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("marshal new participant: %w", err)
	}
	if _, err := tx.Exec(ctx, `
INSERT INTO channel_participant_event_queue (receiver_auth_key_id, qts, receiver_user_id, channel_id, actor_user_id,
    date, previous_participant, new_participant)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
		ev.ReceiverAuthKeyID, ev.Qts, ev.ReceiverUserID, ev.ChannelID, ev.ActorUserID,
		ev.Date, previousRaw, newRaw); err != nil {
		return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("insert channel participant event: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return domain.DeviceChannelParticipantEvent{}, fmt.Errorf("commit append channel participant event: %w", err)
	}
	committed = true
	return ev, nil
}

func (s *ChannelParticipantQueueStore) ListChannelParticipantEventsSince(ctx context.Context, receiverAuthKeyID int64, sinceQts, limit int) ([]domain.DeviceChannelParticipantEvent, error) {
	if limit <= 0 || limit > 1001 {
		limit = 1000
	}
	rows, err := s.db.Query(ctx,
		`SELECT `+channelParticipantEventColumns+` FROM channel_participant_event_queue
WHERE receiver_auth_key_id = $1 AND qts > $2 ORDER BY qts ASC LIMIT $3`,
		receiverAuthKeyID, sinceQts, limit)
	if err != nil {
		return nil, fmt.Errorf("list channel participant events: %w", err)
	}
	defer rows.Close()
	var out []domain.DeviceChannelParticipantEvent
	for rows.Next() {
		ev, err := scanChannelParticipantEvent(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, ev)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

func (s *ChannelParticipantQueueStore) AckChannelParticipantEvents(ctx context.Context, receiverAuthKeyID int64, maxQts int) error {
	if maxQts <= 0 {
		return nil
	}
	if _, err := s.db.Exec(ctx, `
UPDATE channel_participant_event_queue SET acked = true
WHERE receiver_auth_key_id = $1 AND qts <= $2 AND NOT acked`, receiverAuthKeyID, maxQts); err != nil {
		return fmt.Errorf("ack channel participant events: %w", err)
	}
	return nil
}