// Package dlq is the M8 dead-letter queue writer used by // deliverd-fcm and deliverd-telegram. When a delivery // exhausts its retry budget, the worker calls Write() // once to insert a forensic row into deliveries_dlq and // mark the live deliveries row as status='dlq'. // // The package is intentionally tiny: one function, one // row insert, one row update. The retry loop itself // stays in the deliverd main.go (it's small and per- // channel, since the HTTP target and the request shape // differ per channel). // // Schema reference: migrations/008_dlq.up.sql. package dlq import ( "context" "encoding/json" "fmt" "git3.techno-world.net/lrosales/broad-announce/internal/postgres" ) // Entry is the minimal payload we need to write a DLQ // row. It mirrors the columns of `deliveries_dlq` that // the worker fills in; everything else (id, created_at, // discarded) is set by the database. type Entry struct { AlertID string CompanyID string IndividualID string Channel string // fcm | telegram | … Target string // fcm_token, chat_id, … OriginalSubject string // deliveries.fcm., for replay Attempts int LastError string Payload json.RawMessage // raw NATS envelope bytes } // Write inserts one row into deliveries_dlq. It also // updates the matching `deliveries` row to status='dlq' // for the audit trail (best-effort: we don't fail the // DLQ write if the update misses, because the DLQ row // is the source of truth for replay). // // Returns the new DLQ row's id (for logging). func Write(ctx context.Context, pool *postgres.Pool, e Entry) (int64, error) { if e.OriginalSubject == "" { return 0, fmt.Errorf("dlq.Write: OriginalSubject is required for replay") } if e.Payload == nil { // Store the literal JSON null rather than an // empty byte slice so CH's String column gets // a sensible value. e.Payload = json.RawMessage("null") } // The DLQ insert. var newID int64 err := pool.QueryRow(ctx, ` INSERT INTO deliveries_dlq (alert_id, company_id, individual_id, channel, target, original_subject, attempts, last_error, payload) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING id `, e.AlertID, e.CompanyID, e.IndividualID, e.Channel, e.Target, e.OriginalSubject, e.Attempts, e.LastError, e.Payload, ).Scan(&newID) if err != nil { return 0, fmt.Errorf("dlq insert: %w", err) } // Best-effort status flip on the live deliveries row. // We do a soft match: same (alert_id, company_id, // individual_id, channel) and attempts column matches. // This is good enough for the audit trail; the DLQ // row id is the source of truth for replay. _, _ = pool.Exec(ctx, ` UPDATE deliveries SET status = 'dlq', last_error = $1 WHERE alert_id = $2 AND company_id = $3 AND individual_id = $4 AND channel = $5 AND status NOT IN ('sent', 'dlq') `, e.LastError, e.AlertID, e.CompanyID, e.IndividualID, e.Channel) return newID, nil }