broad-announce/internal/dlq/query.go

// M13a W5: DLQ query helpers, used by the per-channel admin
// endpoints in deliverd-fcm and deliverd-telegram. The
// shape mirrors what admind does for its global /v1/dlq
// endpoint, but the package is shared so the SQL lives in
// one place.
package dlq

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"git3.techno-world.net/lrosales/broad-announce/internal/postgres"
	"github.com/jackc/pgx/v5"
)

// Row is the row shape returned by List / Get. Mirrors the
// columns of deliveries_dlq (payload excluded from the list
// view; Get returns it inline).
type Row struct {
	ID              int64           `json:"id"`
	AlertID         string          `json:"alert_id"`
	CompanyID       string          `json:"company_id"`
	IndividualID    string          `json:"individual_id"`
	Channel         string          `json:"channel"`
	Target          string          `json:"target"`
	OriginalSubject string          `json:"original_subject"`
	Attempts        int             `json:"attempts"`
	LastError       string          `json:"last_error"`
	Discarded       bool            `json:"discarded"`
	DiscardedAt     *time.Time      `json:"discarded_at,omitempty"`
	DiscardedBy     *string         `json:"discarded_by,omitempty"`
	CreatedAt       time.Time       `json:"created_at"`
	Payload         json.RawMessage `json:"payload,omitempty"`
}

// ListFilters is the input to List. Empty fields mean "no
// filter on that field". Limit / Offset paginate.
type ListFilters struct {
	Channel          string // fcm | telegram | … (deliverd-* binaries always set this)
	CompanyID        string
	AlertID          string
	IncludeDiscarded bool
	Limit            int
	Offset           int
}

// List returns rows matching the filters, ordered by created_at
// DESC. Excludes the payload column from the list view;
// callers fetch /v1/admin/dlq/{id} for the full row.
func List(ctx context.Context, pool *postgres.Pool, f ListFilters) ([]Row, error) {
	conds := []string{"created_at > now() - INTERVAL '30 days'"}
	args := []any{}
	if f.Channel != "" {
		args = append(args, f.Channel)
		conds = append(conds, fmt.Sprintf("channel = $%d", len(args)))
	}
	if f.CompanyID != "" {
		args = append(args, f.CompanyID)
		conds = append(conds, fmt.Sprintf("company_id = $%d", len(args)))
	}
	if f.AlertID != "" {
		args = append(args, f.AlertID)
		conds = append(conds, fmt.Sprintf("alert_id = $%d", len(args)))
	}
	if !f.IncludeDiscarded {
		conds = append(conds, "discarded = false")
	}
	where := strings.Join(conds, " AND ")

	limit := f.Limit
	if limit <= 0 || limit > 500 {
		limit = 50
	}
	offset := f.Offset
	if offset < 0 {
		offset = 0
	}
	args = append(args, limit, offset)
	q := fmt.Sprintf(`
		SELECT id, alert_id, company_id, individual_id, channel, target,
		       original_subject, attempts, last_error, discarded,
		       discarded_at, discarded_by, created_at
		FROM deliveries_dlq
		WHERE %s
		ORDER BY created_at DESC
		LIMIT $%d OFFSET $%d
	`, where, len(args)-1, len(args))

	rows, err := pool.Query(ctx, q, args...)
	if err != nil {
		return nil, fmt.Errorf("dlq.List: %w", err)
	}
	defer rows.Close()
	var out []Row
	for rows.Next() {
		var r Row
		var dAt *time.Time
		var dBy *string
		if err := rows.Scan(
			&r.ID, &r.AlertID, &r.CompanyID, &r.IndividualID,
			&r.Channel, &r.Target, &r.OriginalSubject, &r.Attempts,
			&r.LastError, &r.Discarded, &dAt, &dBy, &r.CreatedAt,
		); err != nil {
			return nil, fmt.Errorf("dlq.List scan: %w", err)
		}
		r.DiscardedAt = dAt
		r.DiscardedBy = dBy
		out = append(out, r)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("dlq.List rows: %w", err)
	}
	return out, nil
}

// Get returns one row by id, including the payload. Returns
// (nil, nil) when the row doesn't exist (idempotent).
func Get(ctx context.Context, pool *postgres.Pool, id int64) (*Row, error) {
	q := `
		SELECT id, alert_id, company_id, individual_id, channel, target,
		       original_subject, attempts, last_error, discarded,
		       discarded_at, discarded_by, created_at, payload
		FROM deliveries_dlq
		WHERE id = $1
		LIMIT 1
	`
	var r Row
	var dAt *time.Time
	var dBy *string
	err := pool.QueryRow(ctx, q, id).Scan(
		&r.ID, &r.AlertID, &r.CompanyID, &r.IndividualID,
		&r.Channel, &r.Target, &r.OriginalSubject, &r.Attempts,
		&r.LastError, &r.Discarded, &dAt, &dBy, &r.CreatedAt, &r.Payload,
	)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, nil
		}
		return nil, fmt.Errorf("dlq.Get: %w", err)
	}
	r.DiscardedAt = dAt
	r.DiscardedBy = dBy
	return &r, nil
}