Merge pull request #369 from xmidt-org/denopink/feat/add-timestamps-t…

…o-online-offline-events

feat: add timestamps to connect/disconnect events

deviceStatus.go

@@ -52,17 +52,20 @@ func onlinePayload(t time.Time, d device.Interface) []byte {
 
 func newOnlineMessage(source string, d device.Interface) (string, *wrp.Message) {
 	eventType := statusEventType(d.ID(), "online")
-
-	return eventType, &wrp.Message{
+	t := time.Now()
+	m := &wrp.Message{
 		Type:        wrp.SimpleEventMessageType,
 		Source:      source,
 		Destination: "event:" + eventType,
 		ContentType: wrp.MimeTypeJson,
 		PartnerIDs:  []string{d.Metadata().PartnerIDClaim()},
 		SessionID:   d.Metadata().SessionID(),
 		Metadata:    statusMetadata(d),
-		Payload:     onlinePayload(time.Now(), d),
+		Payload:     onlinePayload(t, d),
 	}
+	updateTimestampMetadata(m, t)
+
+	return eventType, m
 }
 
 func offlinePayload(t time.Time, d device.Interface) []byte {
@@ -92,15 +95,22 @@ func offlinePayload(t time.Time, d device.Interface) []byte {
 
 func newOfflineMessage(source string, d device.Interface) (string, *wrp.Message) {
 	eventType := statusEventType(d.ID(), "offline")
-
-	return eventType, &wrp.Message{
+	t := time.Now()
+	m := &wrp.Message{
 		Type:        wrp.SimpleEventMessageType,
 		Source:      source,
 		Destination: "event:" + eventType,
 		ContentType: wrp.MimeTypeJson,
 		PartnerIDs:  []string{d.Metadata().PartnerIDClaim()},
 		SessionID:   d.Metadata().SessionID(),
 		Metadata:    statusMetadata(d),
-		Payload:     offlinePayload(time.Now(), d),
+		Payload:     offlinePayload(t, d),
 	}
+	updateTimestampMetadata(m, t)
+
+	return eventType, m
+}
+
+func updateTimestampMetadata(m *wrp.Message, t time.Time) {
+	m.Metadata[device.WRPTimestampMetadataKey] = t.Format(time.RFC3339Nano)
 }

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/xmidt-org/clortho v0.0.4
 	github.com/xmidt-org/sallust v0.2.2
 	github.com/xmidt-org/touchstone v0.1.3
-	github.com/xmidt-org/webpa-common/v2 v2.2.2
+	github.com/xmidt-org/webpa-common/v2 v2.3.0
 	github.com/xmidt-org/wrp-go/v3 v3.2.3
 	go.opentelemetry.io/contrib/instrumentation/github.com/gorilla/mux/otelmux v0.40.0
 	go.uber.org/zap v1.26.0
@@ -53,7 +53,7 @@ require (
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/google/uuid v1.4.0 // indirect
 	github.com/gorilla/schema v1.2.0 // indirect
-	github.com/gorilla/websocket v1.5.0 // indirect
+	github.com/gorilla/websocket v1.5.1 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.15.1 // indirect
 	github.com/hashicorp/consul/api v1.25.1 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
@@ -110,7 +110,7 @@ require (
 	go.opentelemetry.io/otel/trace v1.17.0 // indirect
 	go.opentelemetry.io/proto/otlp v0.19.0 // indirect
 	go.uber.org/dig v1.17.0 // indirect
-	go.uber.org/fx v1.20.0 // indirect
+	go.uber.org/fx v1.20.1 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/crypto v0.16.0 // indirect
 	golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect