74 lines
2.7 KiB
Go
74 lines
2.7 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"git.novox.be/novox/mesh-sdk/go/handacts"
|
|
|
|
"mesh-delivery-builder/internal/build"
|
|
)
|
|
|
|
// eventsStream is the stream every module's and role's events are on, where the controller makes the builder's durable
|
|
// consumer of artifacts-gone (mesh-controller internal/broker EventsStream).
|
|
const eventsStream = "EVENTS"
|
|
|
|
// SeatHeld is the seat mesh-artifact-sweep's verb held (ADR 0571 §2).
|
|
const SeatHeld = "seat:" + catalogue.ArtifactSweepSeat + ".held"
|
|
|
|
// toolCaller calls a tool or a seat verb as the builder's own account (mesh-sdk process.Conn.Call).
|
|
type toolCaller func(key string, body any) (json.RawMessage, error)
|
|
|
|
// storedVia is build.Builder's Stored: it asks the controller's verb held (mesh-sdk delivery.HeldVerb; novox/hq issue
|
|
// 1937) and answers the references the store no longer holds. One it could not ask about, with none gone, is an error:
|
|
// the key is then left done, and the controller's take-in still refuses a gone blob.
|
|
//
|
|
// It asks the seat mesh-artifact-sweep's `held` (novox/hq ADR 0571 §2), and the controller's while that seat has no holder that
|
|
// answers (a named exception of clause 10, until the lease write).
|
|
func storedVia(call toolCaller) func(ctx context.Context, references []string) ([]string, error) {
|
|
return func(_ context.Context, references []string) ([]string, error) {
|
|
body := map[string]any{"references": references}
|
|
raw, err := call(SeatHeld, body)
|
|
if handacts.NoResponder(err) {
|
|
raw, err = call(delivery.HeldVerb, body)
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("asking held: %w", err)
|
|
}
|
|
a, err := delivery.DecodeHeldAnswer(raw, references)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(a.Gone) == 0 && len(a.Unknown) > 0 {
|
|
var why []string
|
|
for r, w := range a.Unknown {
|
|
why = append(why, r+": "+w)
|
|
}
|
|
sort.Strings(why)
|
|
return nil, fmt.Errorf("the artifact store could not be asked about %s", strings.Join(why, "; "))
|
|
}
|
|
return a.Gone, nil
|
|
}
|
|
}
|
|
|
|
// heardGone takes one artifacts-gone event in: acknowledged, then handed to the builder, which asks the store and builds
|
|
// the key again. A body that is not one is said and acknowledged, so it is not handed over again.
|
|
func heardGone(ctx context.Context, b *build.Builder, m jetstream.Msg, logf func(string, ...any)) {
|
|
e, err := delivery.DecodeArtifactsGone(m.Data())
|
|
if ackErr := m.Ack(); ackErr != nil {
|
|
logf("acknowledging %s: %v", m.Subject(), ackErr)
|
|
}
|
|
if err != nil {
|
|
logf("%s holds what is not %s, left: %v", m.Subject(), delivery.ArtifactsGoneEvent, err)
|
|
return
|
|
}
|
|
b.ArtifactsGone(ctx, e)
|
|
}
|