package repository import ( "context" "errors" "fmt" "time" "github.com/jackc/pgx/v5" "github.com/prabhupadasays/spsays-bff/internal/model" ) var ErrBroadcastGone = errors.New("broadcast not found") type BroadcastRepo struct{ pg *Postgres } func NewBroadcastRepo(pg *Postgres) *BroadcastRepo { return &BroadcastRepo{pg: pg} } const broadcastColumns = `id, title, body, entities, banner_key, cta_label, cta_url, published_at, expires_at, created_at` // scanBroadcast decodes a row produced by broadcastColumns. bannerKey stays // out of the model — the service turns it into a URL. func scanBroadcast(row pgx.Row, b *model.Broadcast, bannerKey **string) error { return row.Scan(&b.ID, &b.Title, &b.Body, &b.Entities, bannerKey, &b.CTALabel, &b.CTAURL, &b.PublishedAt, &b.ExpiresAt, &b.CreatedAt) } type CreateBroadcastParams struct { Title string Body string Entities []byte BannerKey *string CTALabel *string CTAURL *string ExpiresAt *time.Time CreatedBy string } func (r *BroadcastRepo) Create(ctx context.Context, p CreateBroadcastParams) (*model.Broadcast, *string, error) { var b model.Broadcast var key *string err := scanBroadcast(r.pg.Pool.QueryRow(ctx, ` INSERT INTO broadcasts (title, body, entities, banner_key, cta_label, cta_url, expires_at, created_by) VALUES ($1,$2,$3,$4,$5,$6,$7,$8) RETURNING `+broadcastColumns, p.Title, p.Body, p.Entities, p.BannerKey, p.CTALabel, p.CTAURL, p.ExpiresAt, p.CreatedBy), &b, &key) if err != nil { return nil, nil, fmt.Errorf("create broadcast: %w", err) } return &b, key, nil } // Publish stamps published_at, making the broadcast deliverable. Idempotent: // re-publishing keeps the original timestamp so the live window doesn't shift. func (r *BroadcastRepo) Publish(ctx context.Context, id string) (*model.Broadcast, *string, error) { var b model.Broadcast var key *string err := scanBroadcast(r.pg.Pool.QueryRow(ctx, ` UPDATE broadcasts SET published_at = COALESCE(published_at, now()), updated_at = now() WHERE id = $1 RETURNING `+broadcastColumns, id), &b, &key) if errors.Is(err, pgx.ErrNoRows) { return nil, nil, ErrBroadcastGone } if err != nil { return nil, nil, fmt.Errorf("publish broadcast: %w", err) } return &b, key, nil } func (r *BroadcastRepo) Delete(ctx context.Context, id string) error { tag, err := r.pg.Pool.Exec(ctx, `DELETE FROM broadcasts WHERE id = $1`, id) if err != nil { return fmt.Errorf("delete broadcast: %w", err) } if tag.RowsAffected() == 0 { return ErrBroadcastGone } return nil } // Pending returns the newest live broadcast this user has not dismissed. func (r *BroadcastRepo) Pending(ctx context.Context, userID string) (*model.Broadcast, *string, error) { var b model.Broadcast var key *string err := scanBroadcast(r.pg.Pool.QueryRow(ctx, ` SELECT `+broadcastColumns+` FROM broadcasts b WHERE b.published_at IS NOT NULL AND b.published_at <= now() AND (b.expires_at IS NULL OR b.expires_at > now()) AND NOT EXISTS ( SELECT 1 FROM broadcast_views v WHERE v.broadcast_id = b.id AND v.user_id = $1) -- Newest first. Announcements are time-sensitive, so a stale one -- must not gate a fresh one; dismissing this reveals the next, so -- nothing is skipped either way — only the order differs. ORDER BY b.published_at DESC LIMIT 1`, userID), &b, &key) if errors.Is(err, pgx.ErrNoRows) { return nil, nil, nil // nothing waiting is not an error } if err != nil { return nil, nil, fmt.Errorf("pending broadcast: %w", err) } return &b, key, nil } // MarkSeen records a dismissal. Idempotent — two tabs can race. func (r *BroadcastRepo) MarkSeen(ctx context.Context, id, userID string) error { _, err := r.pg.Pool.Exec(ctx, ` INSERT INTO broadcast_views (broadcast_id, user_id) VALUES ($1,$2) ON CONFLICT DO NOTHING`, id, userID) if err != nil { return fmt.Errorf("mark broadcast seen: %w", err) } return nil } // List returns broadcasts newest first, with dismissal counts, for the admin. func (r *BroadcastRepo) List(ctx context.Context, limit int) ([]model.Broadcast, []*string, error) { rows, err := r.pg.Pool.Query(ctx, ` SELECT `+broadcastColumns+`, (SELECT count(*) FROM broadcast_views v WHERE v.broadcast_id = b.id) FROM broadcasts b ORDER BY COALESCE(b.published_at, b.created_at) DESC LIMIT $1`, limit) if err != nil { return nil, nil, fmt.Errorf("list broadcasts: %w", err) } defer rows.Close() out := []model.Broadcast{} keys := []*string{} for rows.Next() { var b model.Broadcast var key *string if err := rows.Scan(&b.ID, &b.Title, &b.Body, &b.Entities, &key, &b.CTALabel, &b.CTAURL, &b.PublishedAt, &b.ExpiresAt, &b.CreatedAt, &b.SeenCount); err != nil { return nil, nil, fmt.Errorf("scan broadcast: %w", err) } out = append(out, b) keys = append(keys, key) } return out, keys, rows.Err() }