Add kafka

Makefile

@@ -6,6 +6,10 @@ database:
 api:
 	docker-compose up pg-api es-api
 
+.PHONY: msg
+msg:
+	docker-compose up kafka zookeeper
+
 .PHONY: down
 down:
 	docker-compose down

README.md

@@ -1,5 +1,5 @@
 # goapi-stac  
-### a STAC api written in go with fiber, gorm, elasticsearch and postgres  
+### a STAC api written in go with fiber, gorm, elasticsearch, postgres and kafka
 #### https://documenter.getpostman.com/view/12888943/VVBXwQnu   
 -------
 
@@ -12,11 +12,12 @@
 ```$ make test```
 
 ### PSQL:
-```$ docker exec -it stac-db bash```
+```$ docker exec -it stac-db bash```   
 ```$ psql```
 
 ### RUN IN DOCKER (localhost:6002):  
-```$ make database```  
+```$ make database```   
+```$ make msg```   
 ```$ make api```  
 
 ---- 

docker-compose.yml

@@ -75,5 +75,32 @@ services:
     ports:
       - "9200:9200"
 
+  zookeeper:
+    image: confluentinc/cp-zookeeper:latest
+    restart: always
+    container_name: Zookeeper
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+
+  kafka:
+    image: confluentinc/cp-kafka:latest
+    restart: always
+    container_name: Kafka
+    depends_on:
+      - zookeeper
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_LOG4J_LOGGERS: "kafka.controller=DEBUG"
+      KAFKA_ZOOKEEPER_SESSION_TIMEOUT_MS: 20000
+    ports:
+      - "9092:9092"
+
 volumes:
   esdata:

es-api/app.go

@@ -15,6 +15,7 @@ import (
 	"github.com/gofiber/fiber/v2/middleware/logger"
 	"github.com/gofiber/fiber/v2/middleware/recover"
 
+	controllers "github.com/jonhealy1/goapi-stac/es-api/controllers"
 	database "github.com/jonhealy1/goapi-stac/es-api/database"
 	router "github.com/jonhealy1/goapi-stac/es-api/router"
 )
@@ -77,5 +78,9 @@ func Setup() *fiber.App {
 		})
 	})
 
+	// Start Kafka consumers
+	go database.StartConsumer("new-postgres-collection", controllers.CreateCollectionFromMessage)
+	go database.StartConsumer("update-postgres-collection", controllers.UpdateCollectionFromMessage)
+
 	return app
 }

es-api/controllers/controller-helpers.go

@@ -0,0 +1,49 @@
+package controllers
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+
+	"github.com/confluentinc/confluent-kafka-go/kafka"
+	"github.com/jonhealy1/goapi-stac/es-api/models"
+)
+
+func CreateCollectionFromMessage(msg *kafka.Message) {
+	// Parse the message to get the collection
+	var stac_collection models.StacCollection
+	fmt.Println("Message: ", string(msg.Value))
+	err := json.Unmarshal(msg.Value, &stac_collection)
+	if err != nil {
+		log.Printf("Error unmarshalling collection from Kafka message: %v\n", err)
+		return
+	}
+
+	// Call CreateESCollectionCore with the parsed collection
+	collection, err := UpsertESCollectionCore(false, stac_collection.Id, &stac_collection)
+	if err != nil {
+		log.Printf("Error creating ES collection from Kafka message: %v\n", err)
+	} else {
+		log.Printf("ES collection created successfully: %s\n", collection.Id)
+	}
+}
+
+func UpdateCollectionFromMessage(msg *kafka.Message) {
+	fmt.Println("UpdateCollectionFromMessage")
+	// Parse the message to get the collection
+	var stac_collection models.StacCollection
+	fmt.Println("Message: ", string(msg.Value))
+	err := json.Unmarshal(msg.Value, &stac_collection)
+	if err != nil {
+		log.Printf("Error1 unmarshalling collection from Kafka message: %v\n", err)
+		return
+	}
+
+	// Call UpdateESCollectionCore with the parsed collection
+	collection, err := UpsertESCollectionCore(true, stac_collection.Id, &stac_collection)
+	if err != nil {
+		log.Printf("Error updating ES collection from Kafka message: %v\n", err)
+	} else {
+		log.Printf("ES collection updated successfully: %s\n", collection.Id)
+	}
+}

es-api/controllers/es-collection-controller.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"strings"
 	"time"
 
 	"github.com/jonhealy1/goapi-stac/es-api/database"
@@ -16,69 +17,99 @@ import (
 	"github.com/jonhealy1/goapi-stac/es-api/models"
 )
 
-func CreateESCollection(c *fiber.Ctx) error {
-	stac_collection := new(models.StacCollection)
-	err := c.BodyParser(&stac_collection)
-	if err != nil {
-		c.Status(http.StatusUnprocessableEntity).JSON(
-			&fiber.Map{"message": "request failed"})
-		return err
-	}
-
+func UpsertESCollectionCore(isUpdate bool, id string, stac_collection *models.StacCollection) (models.Collection, error) {
 	now := time.Now()
 	collection := models.Collection{
 		Data:      models.JSONB{(&stac_collection)},
 		Id:        stac_collection.Id,
 		CreatedAt: &now,
+		UpdatedAt: &now,
 	}
 	validator := validator.New()
-	err = validator.Struct(collection)
+	err := validator.Struct(collection)
 
 	if err != nil {
-		c.Status(http.StatusUnprocessableEntity).JSON(
-			&fiber.Map{"message": err},
-		)
-		return err
+		return models.Collection{}, err
 	}
 
 	indexName := "collections"
-
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
 	_, err = database.ES.Client.Get().
 		Index(indexName).
-		Id(collection.Id).
+		Id(id).
 		Do(ctx)
 
-	if err == nil {
-		c.Status(http.StatusConflict).JSON(
-			&fiber.Map{"message": fmt.Sprintf("Collection %s already exists", collection.Id)})
-		return err
+	if !isUpdate && err == nil {
+		return models.Collection{}, fmt.Errorf("Collection %s already exists", collection.Id)
+	}
+
+	if isUpdate && err != nil {
+		return models.Collection{}, fmt.Errorf("Collection %s not found", id)
 	}
 
 	doc, err := json.Marshal(collection)
 	if err != nil {
-		c.Status(http.StatusInternalServerError).JSON(
-			&fiber.Map{"message": "could not marshal collection"})
-		return err
+		return models.Collection{}, fmt.Errorf("could not marshal collection")
 	}
 
-	resp, err := database.ES.Client.Index().
-		Index(indexName).
-		Id(collection.Id).
-		BodyString(string(doc)).
-		Do(ctx)
+	var docMap map[string]interface{}
+	err = json.Unmarshal(doc, &docMap)
+	if err != nil {
+		return models.Collection{}, fmt.Errorf("could not unmarshal collection")
+	}
+
+	if isUpdate {
+		fmt.Println("UpdateCollectionFromMessageinUpsertESCollectionCore")
+		_, err = database.ES.Client.Update().
+			Index(indexName).
+			Id(id).
+			Doc(docMap).
+			Do(ctx)
+	} else {
+		_, err = database.ES.Client.Index().
+			Index(indexName).
+			Id(collection.Id).
+			BodyString(string(doc)).
+			Do(ctx)
+	}
 
 	if err != nil {
-		c.Status(http.StatusBadRequest).JSON(
-			&fiber.Map{"message": "could not index collection"})
+		return models.Collection{}, fmt.Errorf("could not %s collection", isUpdate)
+	}
+
+	return collection, nil
+}
+
+func CreateESCollection(c *fiber.Ctx, stac_collection *models.StacCollection) error {
+	if stac_collection == nil {
+		stac_collection = new(models.StacCollection)
+		err := c.BodyParser(&stac_collection)
+		if err != nil {
+			c.Status(http.StatusUnprocessableEntity).JSON(
+				&fiber.Map{"message": "request failed"})
+			return err
+		}
+	}
+
+	collection, err := UpsertESCollectionCore(false, "", stac_collection)
+	if err != nil {
+		// Determine the appropriate status code based on the error message
+		statusCode := http.StatusInternalServerError
+		if strings.Contains(err.Error(), "already exists") {
+			statusCode = http.StatusConflict
+		} else if strings.Contains(err.Error(), "Could not marshal") || strings.Contains(err.Error(), "Could not index") {
+			statusCode = http.StatusBadRequest
+		}
+
+		c.Status(statusCode).JSON(&fiber.Map{"message": err.Error()})
 		return err
 	}
 
 	c.Status(http.StatusCreated).JSON(&fiber.Map{
 		"message":         "success",
-		"id":              resp.Id,
+		"id":              collection.Id,
 		"stac_collection": collection.Data[0],
 	})
 	return nil
@@ -159,85 +190,34 @@ func GetESCollections(c *fiber.Ctx) error {
 	return c.JSON(collectionDataList)
 }
 
-func EditESCollection(c *fiber.Ctx) error {
+func EditESCollection(c *fiber.Ctx, stac_collection *models.StacCollection) error {
 	id := c.Params("collectionId")
 	if id == "" {
 		c.Status(http.StatusBadRequest).JSON(
 			&fiber.Map{"message": "missing id parameter"})
 		return fmt.Errorf("missing id parameter")
 	}
 
-	stac_collection := new(models.StacCollection)
-	err := c.BodyParser(&stac_collection)
-	if err != nil {
-		c.Status(http.StatusUnprocessableEntity).JSON(
-			&fiber.Map{"message": "request failed"})
-		return err
-	}
-
-	now := time.Now()
-	collection := models.Collection{
-		Data:      models.JSONB{(&stac_collection)},
-		Id:        stac_collection.Id,
-		UpdatedAt: &now,
-	}
-	validator := validator.New()
-	err = validator.Struct(collection)
-
-	if err != nil {
-		c.Status(http.StatusUnprocessableEntity).JSON(
-			&fiber.Map{"message": err},
-		)
-		return err
-	}
-
-	// Update the collection document in Elasticsearch
-	indexName := "collections"
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-
-	_, err = database.ES.Client.Get().
-		Index(indexName).
-		Id(id).
-		Do(ctx)
-
-	if err != nil {
-		c.Status(http.StatusNotFound).JSON(
-			&fiber.Map{"message": fmt.Sprintf("Collection %s not found", id)})
-		return err
-	}
-
-	doc, err := json.Marshal(collection)
-	if err != nil {
-		c.Status(http.StatusInternalServerError).JSON(
-			&fiber.Map{"message": "could not marshal collection"})
-		return err
-	}
-
-	// Unmarshal JSON string back into a map
-	var docMap map[string]interface{}
-	err = json.Unmarshal(doc, &docMap)
-	if err != nil {
-		c.Status(http.StatusInternalServerError).JSON(
-			&fiber.Map{"message": "could not unmarshal collection"})
-		return err
+	if stac_collection == nil {
+		stac_collection = new(models.StacCollection)
+		err := c.BodyParser(&stac_collection)
+		if err != nil {
+			c.Status(http.StatusUnprocessableEntity).JSON(
+				&fiber.Map{"message": "request failed"})
+			return err
+		}
 	}
 
-	resp, err := database.ES.Client.Update().
-		Index(indexName).
-		Id(id).
-		Doc(docMap).
-		Do(ctx)
-
+	collection, err := UpsertESCollectionCore(true, id, stac_collection)
 	if err != nil {
 		c.Status(http.StatusBadRequest).JSON(
-			&fiber.Map{"message": "could not update collection"})
+			&fiber.Map{"message": err.Error()})
 		return err
 	}
 
 	c.Status(http.StatusOK).JSON(&fiber.Map{
 		"message":         "success",
-		"id":              resp.Id,
+		"id":              collection.Id,
 		"stac_collection": collection.Data[0],
 	})
 	return nil