mesh-controller/cmd/mesh-controller/deadletters.go

package main

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"

	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/catalogue"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/link"
)

// What a consumer gave up on, answered by the serving controller (novox/hq issue 330).
//
// **In this process, on its own connection**: DEAD_LETTERS is read and changed on the bus, and the
// serving controller is already on it. A verb run as a fresh process would open a connection of its own
// for each call (novox/hq issue 327).

// busHandles are the serving controller's connection and JetStream handle.
type busHandles struct {
	conn *nats.Conn
	js   nats.JetStreamContext
}

// defaultDeadLetters is how many the list says when not asked for more.
const defaultDeadLetters = 50

// causeDeadLetter is the cause a delivery or drop gives when the caller gives none.
const causeDeadLetter = "dead-letter"

// deadLettersAnswer is what `dead-letters` answers: the list, one whole, or what came of delivering one
// again or dropping it.
func deadLettersAnswer(ctx context.Context, a *verbArguments) (any, error) {
	serving := servingBus.Load()
	if serving == nil {
		return nil, errors.New("this controller is not serving, so it does not read DEAD_LETTERS: ask again, " +
			"and the serving controller answers")
	}
	on := &busHandles{conn: serving.Conn(), js: serving.Context()}
	// The shape first, and every argument it reads; one given beside it is refused before anything is
	// done, as every verb refuses what it would pass over (novox/hq issue 244).
	var deliver, drop, why, cause, idText, consumer, limit string
	switch {
	case a.given["deliver"] != "" || a.given["drop"] != "":
		deliver, drop, why, cause = a.str("deliver"), a.str("drop"), a.str("why"), a.str("cause")
	case a.given["id"] != "":
		idText = a.str("id")
	default:
		consumer, limit = a.str("consumer"), a.str("limit")
	}
	if unused := a.unused(); len(unused) > 0 {
		return nil, fmt.Errorf("dead-letters did not use %s together with %s, and an argument a verb would pass "+
			"over is refused: nothing was done", quoteAll(unused), quoteAll(a.usedGiven()))
	}
	switch {
	case deliver != "" && drop != "":
		return nil, errors.New("dead-letters delivers one again or drops one, not both. Nothing was done")
	case deliver != "" || drop != "":
		act, text := "deliver", deliver
		if drop != "" {
			act, text = "drop", drop
		}
		if strings.TrimSpace(why) == "" {
			return nil, fmt.Errorf("dead-letters %s is a hand act, and says why: why is required and recorded in "+
				"the hand-act log (novox/hq to-be 45 §7). Nothing was done", act)
		}
		id, err := deadLetterID(text)
		if err != nil {
			return nil, err
		}
		return actOnDeadLetter(ctx, on, act, id, why, cause)
	case idText != "":
		id, err := deadLetterID(idText)
		if err != nil {
			return nil, err
		}
		return link.DeadLetterNamed(on.js, id)
	}
	most := defaultDeadLetters
	if limit != "" {
		n, err := strconv.Atoi(limit)
		if err != nil || n <= 0 {
			return nil, fmt.Errorf("limit is a number of dead letters, not %q", limit)
		}
		most = n
	}
	held, total, err := link.DeadLetters(on.js, consumer, most)
	if err != nil {
		return nil, err
	}
	answer := map[string]any{"dead_letters": held, "held": total,
		"note": "newest first; with id, one whole; deliver or drop one with why"}
	if total == 0 {
		answer["note"] = "no consumer gave up on a message that is still kept"
	}
	return answer, nil
}

// deadLetterID is a dead letter's id as a caller wrote it.
func deadLetterID(text string) (uint64, error) {
	id, err := strconv.ParseUint(strings.TrimSpace(text), 10, 64)
	if err != nil || id == 0 {
		return 0, fmt.Errorf("a dead letter's id is its number in %s, as dead-letters lists it, not %q",
			broker.DeadLettersStream, text)
	}
	return id, nil
}

// actOnDeadLetter delivers one again or drops it, recorded in the hand-act log before it is done. A log
// that cannot be written is said, and the act still happens: the log is never the reason a person's act
// is refused (handacts.go).
func actOnDeadLetter(ctx context.Context, on *busHandles, act string, id uint64, why, cause string) (any, error) {
	d, err := link.DeadLetterNamed(on.js, id)
	if err != nil {
		return nil, err
	}
	if act == "deliver" {
		// Refused before it is recorded: an act that cannot be done is not an act.
		if _, err := link.AgainTo(on.js, d); err != nil {
			return nil, err
		}
	}
	if cause == "" {
		cause = causeDeadLetter
	}
	by := link.CallerIn(ctx)
	if by == "" {
		by = "a seat call whose caller the bus did not name"
	}
	answer := map[string]any{"dead_letter": d.ID, "consumer": d.Who, "subject": d.Subject}
	recorded, logErr := link.RecordHandAct(ctx, on.conn, link.HandAct{Verb: "dead-letters " + act,
		Args: []string{strconv.FormatUint(id, 10), d.Stream + "." + d.Consumer}, Why: why, Cause: cause,
		Condition: conditions.Key(conditions.ScopeBus, d.Stream+"."+d.Consumer, link.AdvisoryMaxDeliveries),
		By:        by + ", through the " + catalogue.ControllerSeatName + " seat"})
	if logErr != nil {
		answer["unrecorded"] = "the hand-act log could not be written, and the act was done all the same: " + logErr.Error()
	} else {
		answer["recorded"] = recorded.ID
	}
	switch act {
	case "deliver":
		_, to, err := link.DeliverAgain(on.js, id)
		if err != nil {
			return nil, err
		}
		answer["delivered_on"] = to
		answer["done"] = fmt.Sprintf("dead letter %d was delivered again to %s, and nobody else; it is no longer kept",
			id, consumerWho(d.Stream, d.Consumer))
	case "drop":
		if _, err := link.DropDeadLetter(on.js, id); err != nil {
			return nil, err
		}
		answer["done"] = fmt.Sprintf("dead letter %d, which %s gave up on, was dropped for good", id,
			consumerWho(d.Stream, d.Consumer))
	}
	return answer, nil
}