package messaging import ( "encoding/base64" "encoding/json" "fmt" ) // Wire names of the events a MessagingClient listens for. They are the names // liblogosdelivery registers listeners under, and are unrelated to the // eventType each event's JSON carries. const ( wireMessageReceived = "onMessageReceived" wireMessageSent = "onMessageSent" wireMessagePropagated = "onMessagePropagated" wireMessageError = "onMessageError" wireConnectionStatusChange = "onConnectionStatusChange" ) // messagingEvents is the set of events a MessagingClient subscribes to. func messagingEvents() []string { return []string{ wireMessageReceived, wireMessageSent, wireMessagePropagated, wireMessageError, wireConnectionStatusChange, } } // ConnectionStatus reports the node's overall connectivity. It mirrors the Nim // ConnectionStatus enum. type ConnectionStatus int const ( Disconnected ConnectionStatus = iota PartiallyConnected Connected ) func (s ConnectionStatus) String() string { switch s { case Disconnected: return "Disconnected" case PartiallyConnected: return "PartiallyConnected" case Connected: return "Connected" default: return fmt.Sprintf("ConnectionStatus(%d)", int(s)) } } func parseConnectionStatus(s string) ConnectionStatus { switch s { case "Connected": return Connected case "PartiallyConnected": return PartiallyConnected default: return Disconnected } } // Message is a message received from the network: the underlying WakuMessage. type Message struct { ContentTopic ContentTopic Payload []byte // Meta is an opaque wire-format marker stamped by higher layers. Meta []byte // Version discriminates payload encryption schemes. Version uint32 // Timestamp is sender-generated, in nanoseconds. Timestamp int64 Ephemeral bool } // Event is the interface every event delivered on MessagingClient.Events() // implements. Consumers type-switch over the concrete types; the set only // grows, so keep a default branch. // // It is sealed: isMessagingEvent is unexported, so only types declared in this // package can satisfy Event. That is what makes the type switch trustworthy — // no other package can introduce an Event, and every value on the channel is // one of the types below. Adding an event type is then a backwards-compatible // change, because callers cannot have exhaustively matched on a closed set // they do not control. type Event interface { isMessagingEvent() } // MessageReceivedEvent is emitted when a message arrives from the network on a // subscribed content topic. type MessageReceivedEvent struct { MessageHash string Message Message } // MessageSentEvent is emitted when a message has been accepted by the send // service and queued for delivery. type MessageSentEvent struct { RequestID RequestID MessageHash string } // MessagePropagatedEvent is emitted when a message has reached neighbouring // nodes on the network. type MessagePropagatedEvent struct { RequestID RequestID MessageHash string } // MessageErrorEvent is emitted when sending or propagating a message fails. type MessageErrorEvent struct { RequestID RequestID MessageHash string Err string } // ConnectionStatusEvent is emitted when the node's overall connectivity // changes. type ConnectionStatusEvent struct { Status ConnectionStatus } func (MessageReceivedEvent) isMessagingEvent() {} func (MessageSentEvent) isMessagingEvent() {} func (MessagePropagatedEvent) isMessagingEvent() {} func (MessageErrorEvent) isMessagingEvent() {} func (ConnectionStatusEvent) isMessagingEvent() {} // wireBytes decodes a byte field as liblogosdelivery serialises it, which is // not base64. Only the send path is base64: logosdelivery_send decodes the // payload it is given, and the channel events base64-encode explicitly. The // messaging events do neither — a received WakuMessage is rendered by Nim's // std/json, whose default for seq[byte] is an array of integers, so `hello` // arrives as [104,101,108,108,111] (see the captures in event_test.go). // // Base64 strings and null decode too, so the day the library normalises its // encodings this keeps working instead of silently dropping every message. type wireBytes []byte func (b *wireBytes) UnmarshalJSON(data []byte) error { if len(data) == 0 || string(data) == "null" { *b = nil return nil } if data[0] == '[' { // Not []byte: encoding/json reads that from a base64 string only. var nums []int if err := json.Unmarshal(data, &nums); err != nil { return err } out := make([]byte, len(nums)) for i, n := range nums { out[i] = byte(n) } *b = out return nil } var s string if err := json.Unmarshal(data, &s); err != nil { return err } dec, err := base64.StdEncoding.DecodeString(s) if err != nil { return err } *b = dec return nil } // decodeEvent parses one flat event JSON document into a typed Event. An event // whose eventType is not part of the Messaging surface decodes to a nil Event // and no error, so a listener registered for a wider set can ignore it. func decodeEvent(eventJSON string) (Event, error) { var head struct { EventType string `json:"eventType"` } if err := json.Unmarshal([]byte(eventJSON), &head); err != nil { return nil, fmt.Errorf("decode event: %w", err) } decode := func(v any) error { if err := json.Unmarshal([]byte(eventJSON), v); err != nil { return fmt.Errorf("decode %s: %w", head.EventType, err) } return nil } switch head.EventType { case "message_received": var e struct { MessageHash string `json:"messageHash"` Message struct { ContentTopic string `json:"contentTopic"` Payload wireBytes `json:"payload"` Meta wireBytes `json:"meta"` Version uint32 `json:"version"` Timestamp int64 `json:"timestamp"` Ephemeral bool `json:"ephemeral"` } `json:"message"` } if err := decode(&e); err != nil { return nil, err } return MessageReceivedEvent{ MessageHash: e.MessageHash, Message: Message{ ContentTopic: e.Message.ContentTopic, Payload: e.Message.Payload, Meta: e.Message.Meta, Version: e.Message.Version, Timestamp: e.Message.Timestamp, Ephemeral: e.Message.Ephemeral, }, }, nil case "message_sent": var e struct { RequestID string `json:"requestId"` MessageHash string `json:"messageHash"` } if err := decode(&e); err != nil { return nil, err } return MessageSentEvent{RequestID: RequestID(e.RequestID), MessageHash: e.MessageHash}, nil case "message_propagated": var e struct { RequestID string `json:"requestId"` MessageHash string `json:"messageHash"` } if err := decode(&e); err != nil { return nil, err } return MessagePropagatedEvent{RequestID: RequestID(e.RequestID), MessageHash: e.MessageHash}, nil case "message_error": var e struct { RequestID string `json:"requestId"` MessageHash string `json:"messageHash"` Error string `json:"error"` } if err := decode(&e); err != nil { return nil, err } return MessageErrorEvent{ RequestID: RequestID(e.RequestID), MessageHash: e.MessageHash, Err: e.Error, }, nil case "connection_status_change": var e struct { ConnectionStatus string `json:"connectionStatus"` } if err := decode(&e); err != nil { return nil, err } return ConnectionStatusEvent{Status: parseConnectionStatus(e.ConnectionStatus)}, nil default: return nil, nil } }