// 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 }