saving

broker/main.go

@@ -3,6 +3,7 @@ package broker
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"time"
 
 	"github.com/segmentio/kafka-go"
@@ -51,10 +52,14 @@ func ReadEventFromKafka() {
 			logrus.Error("error reading message:", err)
 		}
 
-		var node models.Node
-		json.Unmarshal(m.Value, &node)
+		var node map[string]models.Node
 
-		utils.SaveDomains(node.Data.Data.LeafCert.AllDomains)
+		json.Unmarshal(m.Value, &node)
+		// fmt.Println(string(m.Value))
+		// utils.SaveNode(node)
+		fmt.Println("$$$$$$$", node)
+
+		// utils.SaveDomains(node.Data.Data.LeafCert.AllDomains)
 
 	}
 

main.go

@@ -23,12 +23,12 @@ func main() {
 		select {
 		case jq := <-stream:
 			if data, err := utils.ExtractCertFromStream(jq); err == nil {
+
 				node, err := models.BuildNode("certstream", *data)
 				if err != nil {
 					panic(err)
 				}
 				utils.SaveNode(node)
-
 				broker.SendEventToKafka(Conn, *node)
 			}
 

models/types.go

@@ -10,11 +10,11 @@ import (
 // Styx terminology
 // (https://docs.google.com/document/d/1dIrh1Lp3KAjEMm8o2VzAmuV0Peu-jt9aAh1IHrjAroM/pub#h.xzbicbtscatx)
 type Node struct {
-	ID       string           `json:"id"`
-	Type     string           `json:"type"`
+	ID       string           `json:""`
+	Type     string           `json:""`
 	Data     CertStreamStruct `json:"data"`
-	Created  time.Time        `json:"created"`
-	Modified time.Time        `json:"modified"`
+	Created  time.Time        `json:""`
+	Modified time.Time        `json:""`
 }
 
 // Edge defines a relation between two nodes.

utils/main.go

@@ -0,0 +1 @@
+package utils