package link import ( "context" "crypto/rand" "encoding/hex" "encoding/json" "fmt" "time" amqp "github.com/rabbitmq/amqp091-go" ) // Emitting a module event from Go. // // **Every event rides one topic exchange** (novox/hq ADR 0042), which is not the direct exchange // nodes and the control plane speak over. A module that announces something publishes here, and // consumers bind their own durable queue to a pattern over it. // // This exists because the builder is a module written in Go while every other emitter is // TypeScript on the sdk. The envelope is the sdk's, reproduced exactly: the body is the payload // alone and everything about the event travels as headers. A second shape would be a second thing // for consumers to handle, and they are written against the first. const ( // EventsExchange is where every event rides. Named here rather than imported from the broker // package for the same reason BuildQueueName is duplicated there — one direction of dependency. EventsExchange = "mesh.events" ) // EmitEvent publishes one module event, in the envelope the sdk's consumers expect. // // Persistent, because an event that a broker restart loses is not an announcement. The publish is // not confirmed here: the caller has already done the work the event describes, and a build that // succeeded must not be reported as failed because saying so failed. func EmitEvent(ctx context.Context, channel *amqp.Channel, eventType, source, node string, body any) error { payload, err := json.Marshal(body) if err != nil { return fmt.Errorf("cannot serialise a %s event: %w", eventType, err) } id, err := eventID() if err != nil { return err } return channel.PublishWithContext(ctx, EventsExchange, eventType, false, false, amqp.Publishing{ ContentType: "application/json", DeliveryMode: amqp.Persistent, MessageId: id, Timestamp: time.Now().UTC(), Body: payload, Headers: amqp.Table{ "x-event-id": id, "x-source": source, "x-node": node, "x-time": time.Now().UTC().Format(time.RFC3339), "content-type": "application/json", }, }) } // eventID is what a consumer deduplicates on: delivery is at-least-once, so a handler must be able // to tell a redelivery from a second event, and only the emitter can say which it is. func eventID() (string, error) { raw := make([]byte, 16) if _, err := rand.Read(raw); err != nil { return "", fmt.Errorf("cannot make an event id: %w", err) } return hex.EncodeToString(raw), nil } // KeyModuleBuilt is what the builder announces when it has built something. The catalogue places // it in the module graph; nothing else need care. const KeyModuleBuilt = "module.builder.built" // KeyModuleUpgraded is the catalogue saying a module's current version has moved. // // **The control plane hooks the meaning, not the build.** The builder says what it built; the // catalogue decides whether that was an upgrade — a rebuild producing the commit already current // is not one — and only this says anything the control plane can act on. Consuming the build // directly would make the control plane re-derive a decision another module already made, and the // two would eventually disagree (novox/hq ADR 0072). const KeyModuleUpgraded = "module.mesh-catalog.upgraded" // KeyCatchingUp is the catalogue saying it has just started and may have missed things. // // **A durable queue only keeps what arrived after it existed.** The catalogue's own queue is // durable, so nothing is lost once it is running — but the modules built before it first ran were // announced to a queue that did not exist yet, and on a fresh mesh those are, necessarily, the // shared base, the store the catalogue runs on, and the catalogue itself. The graph's foundation // is the part it never hears about (novox/hq 04-ISSUES/050). // // So it asks, and the control plane answers with what it recorded. Asking rather than being told // because only the catalogue knows it has a gap; the control plane cannot tell a fresh catalogue // from one that is merely quiet. const KeyCatchingUp = "module.mesh-catalog.catching-up" // CatchUpQueue is where that lands. Durable, for the same reason the upgrade queue is: a catalogue // that started while the control plane was restarting is exactly the one with a gap to fill. const CatchUpQueue = "control.catchup" // UpgradeQueue is where those land. Durable and named, not a temporary queue: an upgrade announced // while the control plane is restarting is exactly the one that must not be missed. const UpgradeQueue = "control.upgrades" // Replayer answers a catalogue that says it has just started. // // It is handed every build the mesh recorded, oldest first, and re-announces each. The catalogue // registers them as history: a replayed build changed nothing in the world, so announcing it as an // upgrade would have the mesh act on news that is years old. type Replayer interface { // Announceable is every build worth re-announcing, oldest first. // // It hands them back rather than publishing them: the wire belongs to this package, and a // replay that built its own announcements could drift from what the builder emits — which is // the one thing it must match exactly, because the catalogue has a single handler for both. Announceable(ctx context.Context) ([]Announcement, error) } // Announcement is a build, in the shape the builder announces one. // // The field names are the wire's, not Go's, because a catalogue reads these and a rename here is // an event nobody handles. type Announcement struct { Module string `json:"module"` Commit string `json:"commit"` Repository string `json:"repository"` Path string `json:"path"` Ref string `json:"ref"` Manifest json.RawMessage `json:"manifest,omitempty"` Against []string `json:"against,omitempty"` Made []MadeArtifact `json:"made,omitempty"` // Replay says this is history rather than news: it was built once, and this is the mesh // telling a catalogue that missed it. A consumer registers it and announces nothing — an // upgrade that happened months ago is not one anything should act on now. Replay bool `json:"replay,omitempty"` } // Upgraded is what the catalogue says when a module's current version moves. type Upgraded struct { Module string `json:"module"` Commit string `json:"commit"` Previous string `json:"previous"` }