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