108 lines
3.4 KiB
Go
108 lines
3.4 KiB
Go
package backend
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/eclipse/paho.golang/paho"
|
|
"robot-market/test-mqtt-5/internal/events"
|
|
"robot-market/test-mqtt-5/internal/mqtt"
|
|
)
|
|
|
|
// ReceiptAck confirms observation only; it does not claim device execution,
|
|
// schema validity, durable storage, or delivery of this reply to the device.
|
|
type ReceiptAck struct {
|
|
Type string `json:"type"`
|
|
OK bool `json:"ok"`
|
|
SourceTopic string `json:"source_topic"`
|
|
DeviceID string `json:"device_id,omitempty"`
|
|
MessageID string `json:"message_id,omitempty"`
|
|
RequestID string `json:"request_id,omitempty"`
|
|
ReceivedAt string `json:"received_at"`
|
|
PayloadBytes int `json:"payload_bytes"`
|
|
PayloadSHA256 string `json:"payload_sha256"`
|
|
}
|
|
|
|
func (b *Backend) receipt(packet *paho.Publish) *paho.Publish {
|
|
if packet == nil || packet.Retain || !mqtt.InNamespace(b.Client.Config, packet.Topic) {
|
|
return nil
|
|
}
|
|
parts := strings.Split(packet.Topic, "/")
|
|
category := parts[len(parts)-1]
|
|
// Receipt messages and command replies must never form an ACK feedback loop.
|
|
if category == "acks" || category == "receipts" || category == "commands" {
|
|
return nil
|
|
}
|
|
var fields struct {
|
|
Type string `json:"type"`
|
|
MessageID string `json:"message_id"`
|
|
RequestID string `json:"request_id"`
|
|
}
|
|
// Extraction is optional: malformed JSON and missing fields still get a reply.
|
|
_ = json.Unmarshal(packet.Payload, &fields)
|
|
if fields.Type == "receipt_ack" {
|
|
return nil
|
|
}
|
|
device, _ := mqtt.Identity(b.Client.Config, packet.Topic)
|
|
topic := b.Client.Config.Topics.Namespace + "/acks"
|
|
if device != "" {
|
|
topic = b.Client.Config.Topics.Namespace + "/" + device + "/acks"
|
|
}
|
|
if packet.Properties != nil && packet.Properties.ResponseTopic != "" {
|
|
response := packet.Properties.ResponseTopic
|
|
if mqtt.InNamespace(b.Client.Config, response) && !strings.ContainsAny(response, "+#\x00") && len(response) <= 256 && response != packet.Topic {
|
|
topic = response
|
|
}
|
|
}
|
|
digest := sha256.Sum256(packet.Payload)
|
|
ack := ReceiptAck{Type: "receipt_ack", OK: true, SourceTopic: packet.Topic, DeviceID: device, ReceivedAt: Timestamp(time.Now()), PayloadBytes: len(packet.Payload), PayloadSHA256: hex.EncodeToString(digest[:])}
|
|
if len(fields.MessageID) <= 256 {
|
|
ack.MessageID = fields.MessageID
|
|
}
|
|
if len(fields.RequestID) <= 256 {
|
|
ack.RequestID = fields.RequestID
|
|
}
|
|
reply := mqtt.Packet(topic, 1, false, ack)
|
|
if packet.Properties != nil {
|
|
reply.Properties.CorrelationData = append([]byte(nil), packet.Properties.CorrelationData...)
|
|
}
|
|
return reply
|
|
}
|
|
|
|
func (b *Backend) enqueueReceipt(packet *paho.Publish) {
|
|
if !b.Client.Config.Receipts.Enabled {
|
|
return
|
|
}
|
|
reply := b.receipt(packet)
|
|
if reply == nil {
|
|
return
|
|
}
|
|
select {
|
|
case b.receipts <- reply:
|
|
default:
|
|
b.Client.Record(events.Kind("receipt_ack_dropped"), events.Internal, reply, "", "receipt queue full; no reply sent")
|
|
}
|
|
}
|
|
|
|
func (b *Backend) receiptWorker(ctx context.Context) {
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case reply := <-b.receipts:
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
if err := b.PublishReceipt(ctx, reply); err != nil {
|
|
b.Client.Record(events.Kind("receipt_ack_failed"), events.Internal, reply, "", "reply not confirmed; no automatic retry")
|
|
} else {
|
|
b.Client.Record(events.Kind("receipt_ack_sent"), events.Out, reply, "", "backend receipt reply acknowledged by broker; device receipt not proven")
|
|
}
|
|
}
|
|
}
|
|
}
|