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": delivered, to, err := link.DeliverAgain(on.js, id) if err != nil { return nil, err } answer["delivered_on"] = to if delivered.Original != "" { // What became of the ask in its seat's queue (novox/hq issue 334): removed, or left, and why. answer["original"] = delivered.Original } 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 }