broad-announce/internal/dlq/dlq.go

// 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.<co>, for replay
	Attempts        int
	LastError       string
	Payload         json.RawMessage // raw NATS envelope bytes
}

// Write inserts one row into deliveries_dlq. It does
// NOT touch the per-attempt `deliveries` rows — those
// are the audit trail of what each attempt saw (status
// 'failed' or 'sent'), and we keep them as-is so the
// operator can see "this alert had N attempts and all
// failed" at a glance. The DLQ row is the source of
// truth for "this alert hit the DLQ" and the gateway
// 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)
	}

	return newID, nil
}