feat: implement DynamoDB client with connection and session managemen…

project-portal/project-portal-backend/go.mod

@@ -3,10 +3,12 @@ module carbon-scribe/project-portal/project-portal-backend
 go 1.24.5
 
 require (
-	github.com/aws/aws-sdk-go-v2 v1.41.1
+	github.com/aws/aws-sdk-go-v2 v1.41.8
 	github.com/aws/aws-sdk-go-v2/config v1.32.9
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.9
+	github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.20.41
 	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.2
+	github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.5
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.96.0
 	github.com/elastic/go-elasticsearch/v8 v8.19.1
 	github.com/gin-gonic/gin v1.11.0
@@ -31,19 +33,21 @@ require (
 	filippo.io/edwards25519 v1.1.0 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.17 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.17 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.17 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.24 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.24 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.17 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.32.17 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.8 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.12.1 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.17 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.17 // indirect
 	github.com/aws/aws-sdk-go-v2/service/signin v1.0.5 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sso v1.30.10 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.14 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.41.6 // indirect
-	github.com/aws/smithy-go v1.24.0 // indirect
+	github.com/aws/smithy-go v1.25.1 // indirect
 	github.com/bytedance/sonic v1.14.0 // indirect
 	github.com/bytedance/sonic/loader v0.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.6 // indirect

project-portal/project-portal-backend/go.sum

@@ -1,29 +1,37 @@
 filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
 filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
-github.com/aws/aws-sdk-go-v2 v1.41.1 h1:ABlyEARCDLN034NhxlRUSZr4l71mh+T5KAeGh6cerhU=
-github.com/aws/aws-sdk-go-v2 v1.41.1/go.mod h1:MayyLB8y+buD9hZqkCW3kX1AKq07Y5pXxtgB+rRFhz0=
+github.com/aws/aws-sdk-go-v2 v1.41.8 h1:sRs7nG6/RiEBZ/K5UO2sNw0w40U02Nmz1VtARloTZXk=
+github.com/aws/aws-sdk-go-v2 v1.41.8/go.mod h1:4LAfZOPHNVNQEckOACQx60Y8pSRjIkNZQz1w92xpMJc=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 h1:489krEF9xIGkOaaX3CE/Be2uWjiXrkCH6gUX+bZA/BU=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4/go.mod h1:IOAPF6oT9KCsceNTvvYMNHy0+kMF8akOjeDvPENWxp4=
 github.com/aws/aws-sdk-go-v2/config v1.32.9 h1:ktda/mtAydeObvJXlHzyGpK1xcsLaP16zfUPDGoW90A=
 github.com/aws/aws-sdk-go-v2/config v1.32.9/go.mod h1:U+fCQ+9QKsLW786BCfEjYRj34VVTbPdsLP3CHSYXMOI=
 github.com/aws/aws-sdk-go-v2/credentials v1.19.9 h1:sWvTKsyrMlJGEuj/WgrwilpoJ6Xa1+KhIpGdzw7mMU8=
 github.com/aws/aws-sdk-go-v2/credentials v1.19.9/go.mod h1:+J44MBhmfVY/lETFiKI+klz0Vym2aCmIjqgClMmW82w=
+github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.20.41 h1:NyJGcJ8k8LCvu5da9woZ3jZh/9OMW9BQLdDX+79iwFQ=
+github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.20.41/go.mod h1:eKYWqnORD/8/PSm+TauZGiHhCbqYC3atWwZ93tvbaGs=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.17 h1:I0GyV8wiYrP8XpA70g1HBcQO1JlQxCMTW9npl5UbDHY=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.17/go.mod h1:tyw7BOl5bBe/oqvoIeECFJjMdzXoa/dfVz3QQ5lgHGA=
 github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.2 h1:1i1SUOTLk0TbMh7+eJYxgv1r1f47BfR69LL6yaELoI0=
 github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.2/go.mod h1:bo7DhmS/OyVeAJTC768nEk92YKWskqJ4gn0gB5e59qQ=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.17 h1:xOLELNKGp2vsiteLsvLPwxC+mYmO6OZ8PYgiuPJzF8U=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.17/go.mod h1:5M5CI3D12dNOtH3/mk6minaRwI2/37ifCURZISxA/IQ=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.17 h1:WWLqlh79iO48yLkj1v3ISRNiv+3KdQoZ6JWyfcsyQik=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.17/go.mod h1:EhG22vHRrvF8oXSTYStZhJc1aUgKtnJe+aOiFEV90cM=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.24 h1:u6kJU2i0va1AgtJsH3RdWKWqHULlTh7zHwb35Womf74=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.24/go.mod h1:7GY+xLcXOFUpCkNwDReft9qOAVg54A4/AnjHIU7sSAY=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.24 h1:Xhbcf3KugX6vX7SDyUK205Oicyfg7EGuvoVNyP5L6DM=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.24/go.mod h1:rwDgb2HNOGZsnTHylOUedM7Vnl+bCfnXDqUNPsFWYfk=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 h1:WKuaxf++XKWlHWu9ECbMlha8WOEGm0OUEZqm4K/Gcfk=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4/go.mod h1:ZWy7j6v1vWGmPReu0iSGvRiise4YI5SkR3OHKTZ6Wuc=
 github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.17 h1:JqcdRG//czea7Ppjb+g/n4o8i/R50aTBHkA7vu0lK+k=
 github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.17/go.mod h1:CO+WeGmIdj/MlPel2KwID9Gt7CNq4M65HUfBW97liM0=
-github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4 h1:0ryTNEdJbzUCEWkVXEXoqlXV72J5keC1GvILMOuD00E=
-github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.4/go.mod h1:HQ4qwNZh32C3CBeO6iJLQlgtMzqeG17ziAA/3KDJFow=
+github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.5 h1:8dBj9DoTg1rNP/n5FC13c7zc97hx6Urc+jT+iSC7PVA=
+github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.5/go.mod h1:cFa8ItF/dcfex+Op4D0oWbZePIq1ljmrAOAGlEQyGHo=
+github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.32.17 h1:dIgjncvezVA9gmOMIJtT8X/uWzp5xUmzIP8xfisq+U8=
+github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.32.17/go.mod h1:WqWAYpKRCheOkQ9VQsYEN6suToymE2ROKBVCoBP7Dow=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9 h1:FLudkZLt5ci0ozzgkVo8BJGwvqNaZbTWb3UcucAateA=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9/go.mod h1:w7wZ/s9qK7c8g4al+UyoF1Sp/Z45UwMGcqIzLWVQHWk=
 github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.8 h1:Z5EiPIzXKewUQK0QTMkutjiaPVeVYXX7KIqhXu/0fXs=
 github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.8/go.mod h1:FsTpJtvC4U1fyDXk7c71XoDv3HlRm8V3NiYLeYLh5YE=
+github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.12.1 h1:7InFIafuKLWUQ3hieU3b23JmEKlFAWIzq9GHmYn4rWA=
+github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.12.1/go.mod h1:lfCeSIEwe7jvMcEBKcQ/vVoYK0RNJh1X8pja5GqDQtM=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.17 h1:RuNSMoozM8oXlgLG/n6WLaFGoea7/CddrCfIiSA+xdY=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.17/go.mod h1:F2xxQ9TZz5gDWsclCtPQscGpP0VUOc8RqgFM3vDENmU=
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.17 h1:bGeHBsGZx0Dvu/eJC0Lh9adJa3M1xREcndxLNZlve2U=
@@ -38,8 +46,8 @@ github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.14 h1:0jbJeuEHlwKJ9PfXtpSFc4M
 github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.14/go.mod h1:sTGThjphYE4Ohw8vJiRStAcu3rbjtXRsdNB0TvZ5wwo=
 github.com/aws/aws-sdk-go-v2/service/sts v1.41.6 h1:5fFjR/ToSOzB2OQ/XqWpZBmNvmP/pJ1jOWYlFDJTjRQ=
 github.com/aws/aws-sdk-go-v2/service/sts v1.41.6/go.mod h1:qgFDZQSD/Kys7nJnVqYlWKnh0SSdMjAi0uSwON4wgYQ=
-github.com/aws/smithy-go v1.24.0 h1:LpilSUItNPFr1eY85RYgTIg5eIEPtvFbskaFcmmIUnk=
-github.com/aws/smithy-go v1.24.0/go.mod h1:LEj2LM3rBRQJxPZTB4KuzZkaZYnZPnvgIhb4pu07mx0=
+github.com/aws/smithy-go v1.25.1 h1:J8ERsGSU7d+aCmdQur5Txg6bVoYelvQJgtZehD12GkI=
+github.com/aws/smithy-go v1.25.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc=
 github.com/boombuler/barcode v1.0.0/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
 github.com/bytedance/sonic v1.14.0 h1:/OfKt8HFw0kh2rj8N0F6C/qPGRESq0BbaNZgcNXXzQQ=
 github.com/bytedance/sonic v1.14.0/go.mod h1:WoEbx8WTcFJfzCe0hbmyTGrfjt8PzNEBdxlNUO24NhA=

project-portal/project-portal-backend/internal/notifications/repository.go

@@ -9,6 +9,8 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
+
+	"carbon-scribe/project-portal/project-portal-backend/pkg/aws"
 )
 
 type Repository interface {
@@ -46,6 +48,11 @@ type MongoRepository struct {
 	templates     *mongo.Collection
 	rules         *mongo.Collection
 	connections   *mongo.Collection
+	ddbClient     aws.DynamoDBClient
+}
+
+func (r *MongoRepository) SetDynamoDBClient(ddb aws.DynamoDBClient) {
+	r.ddbClient = ddb
 }
 
 func NewMongoRepository(client *mongo.Client, dbName string) *MongoRepository {
@@ -281,6 +288,9 @@ func (r *MongoRepository) GetRuleByID(ctx context.Context, id string) (*Notifica
 }
 
 func (r *MongoRepository) UpsertConnection(ctx context.Context, conn *WebSocketConnection) error {
+	if r.ddbClient != nil {
+		return r.ddbClient.PutConnection(ctx, toConnectionRecord(conn))
+	}
 	_, err := r.connections.UpdateByID(ctx, conn.ConnectionID, bson.M{"$set": conn}, options.Update().SetUpsert(true))
 	if err != nil {
 		return fmt.Errorf("upsert connection: %w", err)
@@ -289,6 +299,9 @@ func (r *MongoRepository) UpsertConnection(ctx context.Context, conn *WebSocketC
 }
 
 func (r *MongoRepository) DeleteConnection(ctx context.Context, connectionID string) error {
+	if r.ddbClient != nil {
+		return r.ddbClient.DeleteConnection(ctx, connectionID)
+	}
 	_, err := r.connections.DeleteOne(ctx, bson.M{"_id": connectionID})
 	if err != nil {
 		return fmt.Errorf("delete connection: %w", err)
@@ -297,6 +310,41 @@ func (r *MongoRepository) DeleteConnection(ctx context.Context, connectionID str
 }
 
 func (r *MongoRepository) ListConnections(ctx context.Context, projectID string, userID string) ([]WebSocketConnection, error) {
+	if r.ddbClient != nil {
+		var records []aws.ConnectionRecord
+		var err error
+		if userID != "" {
+			records, err = r.ddbClient.ListConnectionsByUser(ctx, userID)
+		} else if projectID != "" {
+			records, err = r.ddbClient.ListConnectionsByProject(ctx, projectID)
+		} else {
+			// Scan/list all if both empty (highly generic fallback)
+			records, err = r.ddbClient.ListConnectionsByUser(ctx, "")
+		}
+		if err != nil {
+			return nil, fmt.Errorf("list connections from DynamoDB: %w", err)
+		}
+
+		conns := make([]WebSocketConnection, 0, len(records))
+		for _, rec := range records {
+			// In case we listed by project and need to double-check matching on projectID filter
+			if projectID != "" {
+				matched := false
+				for _, pid := range rec.ProjectIDs {
+					if pid == projectID {
+						matched = true
+						break
+					}
+				}
+				if !matched {
+					continue
+				}
+			}
+			conns = append(conns, *toWebSocketConnection(&rec))
+		}
+		return conns, nil
+	}
+
 	filter := bson.M{}
 	if userID != "" {
 		filter["user_id"] = userID
@@ -325,6 +373,37 @@ func (r *MongoRepository) ListConnections(ctx context.Context, projectID string,
 	return items, nil
 }
 
+// Helpers for mappings
+func toConnectionRecord(conn *WebSocketConnection) *aws.ConnectionRecord {
+	if conn == nil {
+		return nil
+	}
+	return &aws.ConnectionRecord{
+		ConnectionID: conn.ConnectionID,
+		UserID:       conn.UserID,
+		ProjectIDs:   conn.ProjectIDs,
+		ConnectedAt:  conn.ConnectedAt,
+		LastActivity: conn.LastActivity,
+		UserAgent:    conn.UserAgent,
+		IPAddress:    conn.IPAddress,
+	}
+}
+
+func toWebSocketConnection(rec *aws.ConnectionRecord) *WebSocketConnection {
+	if rec == nil {
+		return nil
+	}
+	return &WebSocketConnection{
+		ConnectionID: rec.ConnectionID,
+		UserID:       rec.UserID,
+		ProjectIDs:   rec.ProjectIDs,
+		ConnectedAt:  rec.ConnectedAt,
+		LastActivity: rec.LastActivity,
+		UserAgent:    rec.UserAgent,
+		IPAddress:    rec.IPAddress,
+	}
+}
+
 func (r *MongoRepository) Metrics(ctx context.Context) (*DeliveryMetrics, error) {
 	total, err := r.notifications.CountDocuments(ctx, bson.M{})
 	if err != nil {

project-portal/project-portal-backend/internal/notifications/websocket/lambda_handlers/connect.go

@@ -1,7 +1,96 @@
-//go:build future
-// +build future
+package lambda_handlers
 
-package notifications
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"strings"
 
-// This file won't be compiled in normal builds
-// Implementation pending
+	ws "carbon-scribe/project-portal/project-portal-backend/internal/notifications/websocket"
+)
+
+// TokenValidator defines an interface for validating JWT auth tokens
+type TokenValidator interface {
+	ValidateToken(token string) (userID string, projectIDs []string, err error)
+}
+
+// ConnectHandler creates a handler to process the $connect WebSocket event from API Gateway
+type ConnectHandler struct {
+	manager   *ws.ConnectionManager
+	validator TokenValidator
+}
+
+// NewConnectHandler creates a new ConnectHandler instance
+func NewConnectHandler(manager *ws.ConnectionManager, validator TokenValidator) *ConnectHandler {
+	return &ConnectHandler{
+		manager:   manager,
+		validator: validator,
+	}
+}
+
+// Handle processes the $connect request
+func (h *ConnectHandler) Handle(ctx context.Context, event ws.Event) (interface{}, error) {
+	connectionID := event.RequestContext.ConnectionID
+	if connectionID == "" {
+		connectionID = event.ConnectionID
+	}
+	if connectionID == "" {
+		return nil, errors.New("missing connection ID")
+	}
+
+	log.Printf("[Lambda Connect] Processing connection request: ConnectionID=%s", connectionID)
+
+	// Extract token from multiple sources: QueryString, Headers, Sec-WebSocket-Protocol
+	var token string
+	if event.QueryString != nil {
+		token = event.QueryString["token"]
+	}
+	if token == "" && event.Headers != nil {
+		token = event.Headers["Authorization"]
+		if strings.HasPrefix(strings.ToLower(token), "bearer ") {
+			token = token[7:]
+		}
+	}
+	if token == "" && event.Headers != nil {
+		// API Gateway sometimes passes auth token via Sec-WebSocket-Protocol header
+		token = event.Headers["Sec-WebSocket-Protocol"]
+	}
+
+	// Validate authorization
+	var userID string
+	var projectIDs []string
+	var err error
+
+	if h.validator != nil && token != "" {
+		userID, projectIDs, err = h.validator.ValidateToken(token)
+		if err != nil {
+			log.Printf("[Lambda Connect] Authorization failed for token: %v", err)
+			return nil, fmt.Errorf("unauthorized: %w", err)
+		}
+	} else {
+		// Fallback/Dev mode if no validator is configured
+		userID = event.UserID
+		if userID == "" {
+			userID = "dev-user-id" // Default fallback for development
+		}
+		projectIDs = []string{"all-projects"}
+		log.Printf("[Lambda Connect] Dev/Fallback mode: using default UserID=%s", userID)
+	}
+
+	userAgent := event.RequestContext.Identity.SourceAgent
+	ipAddress := event.RequestContext.Identity.SourceIP
+
+	// Store connection state in DynamoDB
+	err = h.manager.Connect(ctx, connectionID, userID, projectIDs, userAgent, ipAddress)
+	if err != nil {
+		log.Printf("[Lambda Connect] Failed to save connection in DynamoDB: %v", err)
+		return nil, fmt.Errorf("failed to save connection state: %w", err)
+	}
+
+	log.Printf("[Lambda Connect] Connection registered successfully: ConnectionID=%s, UserID=%s", connectionID, userID)
+	return map[string]interface{}{
+		"statusCode": 200,
+		"body":       "Connected",
+	}, nil
+}

project-portal/project-portal-backend/internal/notifications/websocket/lambda_handlers/default.go

@@ -1,7 +1,28 @@
-//go:build future
-// +build future
+package lambda_handlers
 
-package notifications
+import (
+	"context"
+	"log"
 
-// This file won't be compiled in normal builds
-// Implementation pending
+	ws "carbon-scribe/project-portal/project-portal-backend/internal/notifications/websocket"
+)
+
+// DefaultHandler handles unexpected or custom route keys
+type DefaultHandler struct{}
+
+// NewDefaultHandler creates a new DefaultHandler instance
+func NewDefaultHandler() *DefaultHandler {
+	return &DefaultHandler{}
+}
+
+// Handle processes generic/default WebSocket frames
+func (h *DefaultHandler) Handle(ctx context.Context, event ws.Event) (interface{}, error) {
+	log.Printf("[Lambda Default] Received unhandled WebSocket event: RouteKey=%s, ConnectionID=%s, Body=%s",
+		event.RouteKey, event.RequestContext.ConnectionID, event.Body)
+
+	// In real-time apps, we can parse custom commands or echo back a response.
+	return map[string]interface{}{
+		"statusCode": 200,
+		"body":       "Received default action",
+	}, nil
+}