chore: some changes

backend/docker-compose.yaml

@@ -25,8 +25,20 @@ services:
       OTEL_EXPORTER_OTLP_ENDPOINT: 'http://otel-collector:4317'
       OTEL_EXPORTER_OTLP_PROTOCOL: "grpc"
       OTEL_RESOURCE_ATTRIBUTES: "service.name=api.cloudificando.com,service.version=0.0.1,deployment.environment=dev"
+      ENVIRONMENT: "dev"
     volumes:
       - ./:/live-reload/
+  otel-collector:
+    image: otel/opentelemetry-collector-contrib:0.113.0
+    ports:
+      - "4317:4317"
+      - "4318:4318"
+    volumes:
+      - ./otel-config-dev.yaml:/otel-config.yaml
+    command: ["--config", "/otel-config.yaml"]
+    environment:
+      OTLP_CLOUDIFICANDO_ENDPOINT: ${OTLP_CLOUDIFICANDO_ENDPOINT}
+      OTLP_CLOUDIFICANDO_TOKEN: ${OTLP_CLOUDIFICANDO_TOKEN}
   dynamodb-init:
     image: amazon/aws-cli:2.13.32
     depends_on:
@@ -76,14 +88,3 @@ services:
           ]"
         '
 
-  otel-collector:
-    image: otel/opentelemetry-collector-contrib:0.113.0
-    ports:
-      - "4317:4317"
-      - "4318:4318"
-    volumes:
-      - ./otel-config-dev.yaml:/otel-config.yaml
-    command: ["--config", "/otel-config.yaml"]
-    environment:
-      OTLP_CLOUDIFICANDO_ENDPOINT: ${OTLP_CLOUDIFICANDO_ENDPOINT}
-      OTLP_CLOUDIFICANDO_TOKEN: ${OTLP_CLOUDIFICANDO_TOKEN}

backend/dtos.go

@@ -6,16 +6,23 @@ import "net/http"
 type EventPostUpdatedRequest struct {
 	Message struct {
 		Data        string `json:"data" binding:"required"`
-		MessageId   string `json:"message_id" binding:"required"`
+		MessageId   string `json:"message_id" binding:"required"` // Post json Base 64 encoded
 		PublishTime string `json:"publish_time" binding:"required"`
-		OrderingKey string `json:"ordering_key"`
 		Attributes  struct {
 			EventType string `json:"eventType" binding:"required"` // "POST_CREATED" ,"POST_DELETED", "CONTENT_UPDATED", "META_UPDATED"
 			Slug      string `json:"slug" binding:"required"`
 		} `json:"attributes" binding:"required"`
 	} `json:"message" binding:"required"`
 	Subscription string `json:"subscription" binding:"required"`
 }
+type Post struct {
+	Title       string   `json:"title" dynamodbav:"title" binding:"required"`
+	Tags        []string `json:"tags" dynamodbav:"tags" binding:"required"`
+	CreatedAt   string   `json:"created_at" dynamodbav:"created_at" binding:"required"`
+	Description string   `json:"description" dynamodbav:"description" binding:"required"`
+	Slug        string   `json:"slug" dynamodbav:"slug" binding:"required"`
+}
+
 type HardSyncRequest struct {
 	Posts []Post `json:"posts" binding:"required"`
 }
@@ -34,14 +41,6 @@ func BadRequestError(message string) *RestError {
 	}
 }
 
-type Post struct {
-	Title       string   `json:"title" dynamodbav:"title" binding:"required"`
-	Tags        []string `json:"tags" dynamodbav:"tags" binding:"required"`
-	CreatedAt   string   `json:"created_at" dynamodbav:"created_at" binding:"required"`
-	Description string   `json:"description" dynamodbav:"description" binding:"required"`
-	Slug        string   `json:"slug" dynamodbav:"slug" binding:"required"`
-}
-
 type TagWithCount struct {
 	Tag   string `json:"tag"`
 	Count int    `json:"count"`

backend/handlers.go

@@ -4,10 +4,11 @@ package main
 import (
 	"encoding/base64"
 	"encoding/json"
-	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
-	"github.com/gin-gonic/gin"
 	"log/slog"
 	"strconv"
+
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	"github.com/gin-gonic/gin"
 )
 
 type BlogController struct {
@@ -35,7 +36,7 @@ func (bc *BlogController) GetPostsHandler(c *gin.Context) {
 	tag := c.DefaultQuery("tag", "")
 
 	limit, err := strconv.Atoi(limitStr)
-	if err != nil || limit < 1 || limit > 100 { // Increased max limit for flexibility
+	if err != nil || limit < 1 || limit > 6 {
 		c.AbortWithStatusJSON(400, gin.H{
 			"error": "Invalid limit",
 		})
@@ -72,6 +73,7 @@ func (bc *BlogController) PostsUpdatedGcpSubscriptionHandler(c *gin.Context) {
 
 	// Bind JSON payload
 	if err := c.ShouldBindJSON(&event); err != nil {
+		slog.ErrorContext(ctx, "Invalid request payload", "Error", err)
 		c.AbortWithStatusJSON(400, gin.H{
 			"error": "Invalid request payload",
 		})
@@ -194,7 +196,7 @@ func (bc *BlogController) DeletePostHandler(c *gin.Context) {
 }
 
 func (bc *BlogController) HardSyncHandler(c *gin.Context) {
-	migration := bc.migration
+	//migration := bc.migration
 	repository := bc.repository
 	ctx := c.Request.Context()
 	var body HardSyncRequest
@@ -204,24 +206,24 @@ func (bc *BlogController) HardSyncHandler(c *gin.Context) {
 		})
 		return
 	}
-
-	err := migration.Down(ctx)
-	if err != nil {
-		slog.ErrorContext(ctx, "Failed to run migrations", "Error", err)
-		c.AbortWithStatusJSON(500, gin.H{
-			"error": "Unexpected error",
-		})
-		return
-	}
-	err = migration.Up(ctx)
-	if err != nil {
-		slog.ErrorContext(ctx, "Failed to run migrations", "Error", err)
-		c.AbortWithStatusJSON(500, gin.H{
-			"error": "Unexpected error",
-		})
-		return
-	}
-	err = repository.UpsertPostsBatch(ctx, body.Posts)
+	// TODO: this makes big downtime, refactor to delete all items instead of dropping table
+	//err := migration.Down(ctx)
+	//if err != nil {
+	//	slog.ErrorContext(ctx, "Failed to run migrations", "Error", err)
+	//	c.AbortWithStatusJSON(500, gin.H{
+	//		"error": "Unexpected error",
+	//	})
+	//	return
+	//}
+	//err = migration.Up(ctx)
+	//if err != nil {
+	//	slog.ErrorContext(ctx, "Failed to run migrations", "Error", err)
+	//	c.AbortWithStatusJSON(500, gin.H{
+	//		"error": "Unexpected error",
+	//	})
+	//	return
+	//}
+	err := repository.UpsertPostsBatch(ctx, body.Posts)
 	if err != nil {
 		slog.ErrorContext(ctx, "Failed to upsert posts", "Error", err)
 		c.AbortWithStatusJSON(500, gin.H{
@@ -232,4 +234,5 @@ func (bc *BlogController) HardSyncHandler(c *gin.Context) {
 	c.JSON(200, gin.H{
 		"message": "ok",
 	})
+	_ = repository.InvalidateCdnCache(ctx, "/blog/*")
 }

backend/main.go

@@ -3,6 +3,10 @@ package main
 
 import (
 	"context"
+	"log"
+	"log/slog"
+	"os"
+
 	aws "github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/cloudfront"
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
@@ -11,9 +15,6 @@ import (
 	slogmulti "github.com/samber/slog-multi"
 	"go.opentelemetry.io/contrib/bridges/otelslog"
 	"go.opentelemetry.io/contrib/instrumentation/github.com/aws/aws-sdk-go-v2/otelaws"
-	"log"
-	"log/slog"
-	"os"
 )
 
 func main() {
@@ -53,21 +54,16 @@ func main() {
 	router := gin.New()
 	// Register Global middlewares
 	router.Use(OtelGinMiddleware())
+	slog.InfoContext(ctx, "Allowed origins", "Origins", os.Getenv("ALLOWED_ORIGINS"))
+	//router.Use(CorsMiddleware(strings.Split(os.Getenv("ALLOWED_ORIGINS"), ","))) //todo: understand why duplicate cors headers are being sent
 	router.Use(CdnCacheMiddleware())
-	router.Use(CorsMiddleware())
 	// Register endpoints
 	router.GET("/blog/posts", blogController.GetPostsHandler)
 	router.GET("/blog/tags", blogController.GetTagsHandler)
 	router.PUT("/blog/posts", blogController.UpsertPostHandler)
-	router.POST("/blog/events/posts-updated", GcpPubSubMiddleware(), blogController.UpsertPostHandler)
+	router.POST("/blog/events/posts-updated", GcpPubSubAuthMiddleware(), blogController.PostsUpdatedGcpSubscriptionHandler)
 	router.DELETE("/blog/posts/:slug", blogController.DeletePostHandler)
-	router.POST("/blog/posts/hardsync", blogController.HardSyncHandler)
-	// Run the migrations
-	err = migration.EnsureDbMigrations(ctx)
-	if err != nil {
-		slog.ErrorContext(ctx, "Failed to run migrations", "Error", err)
-		log.Fatal(err)
-	}
-	// Run the server
+	router.POST("/blog/hardsync", blogController.HardSyncHandler)
+
 	router.Run()
 }

backend/middlewares.go

@@ -1,14 +1,16 @@
 package main
 
 import (
-	"github.com/gin-contrib/cors"
-	"github.com/gin-gonic/gin"
-	"go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin"
-	"google.golang.org/api/idtoken"
+	"log/slog"
 	"net/http"
 	"os"
 	"strings"
 	"time"
+
+	"github.com/gin-contrib/cors"
+	"github.com/gin-gonic/gin"
+	"go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin"
+	"google.golang.org/api/idtoken"
 )
 
 func CdnCacheMiddleware() gin.HandlerFunc {
@@ -20,31 +22,46 @@ func CdnCacheMiddleware() gin.HandlerFunc {
 		c.Next()
 	}
 }
-
-func CorsMiddleware() gin.HandlerFunc {
+func RemoveDupHeadersMiddleware() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		for name, values := range c.Request.Header {
+			if len(values) > 1 {
+				c.Request.Header[name] = values[:1]
+			}
+		}
+	}
+}
+func CorsMiddleware(allowedOrigins []string) gin.HandlerFunc {
 	return cors.New(cors.Config{
-		AllowMethods:     []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"},
-		AllowOrigins:     strings.Split(os.Getenv("ALLOWED_ORIGINS"), ","),
-		AllowHeaders:     []string{"Origin", "Content-Type", "Accept", "Authorization"},
+		//AllowAllOrigins: true,
+		AllowMethods: []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"},
+		//AllowOrigins: allowedOrigins,
+		//AllowHeaders: []string{"Origin", "Content-Type", "Accept", "Authorization"},
 		AllowCredentials: true,
-		MaxAge:           12 * time.Hour, // How long the results of a preflight request can be cached
+		//MaxAge:           12 * time.Hour, // How long the results of a preflight request can be cached
 	})
 }
 
 func OtelGinMiddleware() gin.HandlerFunc {
 	return otelgin.Middleware(os.Getenv("PROD_DOMAIN"))
 }
 
-func GcpPubSubMiddleware() gin.HandlerFunc {
+func GcpPubSubAuthMiddleware() gin.HandlerFunc {
 	return func(c *gin.Context) {
 		if os.Getenv("ENVIRONMENT") == "dev" {
 			c.Next()
 			return
 		}
-		// Validate the GCP Pub/Sub request
-		// Verify the ID token
-		_, err := idtoken.Validate(c.Request.Context(), c.Request.Header.Get("Authorization"), os.Getenv("GCP_PROJECT_ID"))
+		headerValue := c.GetHeader("Authorization")
+		if headerValue == "" {
+			c.JSON(http.StatusUnauthorized, gin.H{"error": "Missing Authorization header"})
+			c.Abort()
+			return
+		}
+		token := strings.Split(headerValue, "Bearer ")[1]
+		_, err := idtoken.Validate(c.Request.Context(), token, "https://"+os.Getenv("PROD_DOMAIN")+"/blog/events/posts-updated")
 		if err != nil {
+			slog.ErrorContext(c.Request.Context(), "Invalid ID token", "Error", err)
 			c.JSON(http.StatusUnauthorized, gin.H{"error": "Invalid ID token"})
 			c.Abort()
 			return

backend/migrations.go

@@ -30,6 +30,7 @@ func (m *Migration) EnsureDbMigrations(ctx context.Context) error {
 			slog.ErrorContext(ctx, "Failed to create table", "Error", err)
 			return err
 		}
+		return nil
 	}
 	slog.InfoContext(ctx, "Table created", "tableName", m.TableName)
 	return nil
@@ -182,6 +183,7 @@ func (m *Migration) Down(ctx context.Context) error {
 	}
 	_, err := db.DeleteTable(ctx, deleteTableInput)
 	if err != nil {
+		slog.ErrorContext(ctx, "Failed to delete table", "Error", err)
 		// ignore if is not exists, otherwise return error
 		var resourceNotFoundException *types.ResourceNotFoundException
 		if !errors.As(err, &resourceNotFoundException) {

backend/repository.go

@@ -31,18 +31,21 @@ func NewBlogRepository(ctx context.Context, db *dynamodb.Client, tn string, cdn
 	*BlogRepository, error,
 ) {
 	ssmDistroIdPath := os.Getenv("AWS_SSM_CLOUDFRONT_DISTRO_ID_PATH")
-	cloudfrontDistroId, err := ps.GetParameter(ctx, &ssm.GetParameterInput{
-		Name: &ssmDistroIdPath,
-	})
-	if err != nil {
-		slog.ErrorContext(ctx, "Failed to get cloudfront distro id parameter", "Error", err)
-		return nil, err
+	var cloudfrontDistroId string
+	if os.Getenv("ENVIRONMENT") == "production" {
+		cloudfrontDistroIdParam, err := ps.GetParameter(ctx, &ssm.GetParameterInput{
+			Name: &ssmDistroIdPath,
+		})
+		if err != nil {
+			slog.ErrorContext(ctx, "Failed to get cloudfront distro id parameter", "Error", err)
+			return nil, err
+		}
+		cloudfrontDistroId = *cloudfrontDistroIdParam.Parameter.Value
 	}
-
 	return &BlogRepository{
 		Db:                 db,
 		tableName:          tn,
-		cloudfrontDistroId: *cloudfrontDistroId.Parameter.Value,
+		cloudfrontDistroId: cloudfrontDistroId,
 		cdn:                cdn,
 	}, nil
 }
@@ -462,7 +465,7 @@ func (r *BlogRepository) InvalidateCdnCache(ctx context.Context, path string) er
 	}
 
 	// Send the invalidation request
-	output, err := cdn.CreateInvalidation(context.TODO(), invalidationInput)
+	output, err := cdn.CreateInvalidation(ctx, invalidationInput)
 	if err != nil {
 		return fmt.Errorf("failed to create invalidation: %w", err)
 	}

frontend/app/api/events/posts-updated/route.ts

@@ -40,7 +40,7 @@ export async function POST(req: NextRequest) {
     case "POST_DELETED":
       revalidatePath(`/posts/${slug}`);
       revalidatePath("/");
-      await invalidateCloudFrontPaths(["/posts", "/"]);
+      await invalidateCloudFrontPaths([`/posts/${slug}`, "/"]);
       break;
     case "POST_CREATED":
       revalidatePath("/");

frontend/app/api/hard-sync/route.ts

@@ -18,7 +18,7 @@ export async function POST(req: NextRequest) {
   try {
     await performHardSync()
     revalidatePath("/", "layout")
-    invalidateCloudFrontPaths(["/*"])
+    await invalidateCloudFrontPaths(["/*"])
   }catch (e){
     return NextResponse.error()
   }

frontend/instrumentation.ts

@@ -1,5 +1,5 @@
 import { registerOTel } from '@vercel/otel'
 
 export function register() {
-  registerOTel({ serviceName: 'next-app' })
+  registerOTel("cloudificando-frontend")
 }