Resolve commented issues

Author: VirajSalaka

gateway/gateway-controller/cmd/controller/main.go

@@ -120,7 +120,9 @@ func main() {
 			if err := eventHub.Initialize(ctx); err != nil {
 				log.Fatal("Failed to initialize EventHub", zap.Error(err))
 			}
-			eventHub.RegisterOrganization("default")
+			if err := eventHub.RegisterOrganization("default"); err != nil {
+				log.Fatal("Failed to register default organization", zap.Error(err))
+			}
 			log.Info("EventHub initialized successfully")
 		} else {
 			log.Fatal("EventHub requires persistent storage. Multi-replica mode will not function correctly in memory-only mode.")

gateway/gateway-controller/pkg/eventhub/eventhub.go

@@ -3,6 +3,7 @@ package eventhub
 import (
 	"context"
 	"database/sql"
+	"sync"
 	"time"
 
 	"go.uber.org/zap"
@@ -15,6 +16,7 @@ type eventHub struct {
 	logger  *zap.Logger
 
 	initialized bool
+	mu          sync.RWMutex
 }
 
 // New creates a new EventHub instance with SQLite backend (default)
@@ -43,6 +45,9 @@ func NewWithBackend(backend EventhubImpl, logger *zap.Logger) EventHub {
 
 // Initialize sets up the EventHub and starts background workers
 func (eh *eventHub) Initialize(ctx context.Context) error {
+	eh.mu.Lock()
+	defer eh.mu.Unlock()
+
 	if eh.initialized {
 		return nil
 	}
@@ -82,6 +87,9 @@ func (eh *eventHub) CleanUpEvents(ctx context.Context, timeFrom, timeEnd time.Ti
 
 // Close gracefully shuts down the EventHub
 func (eh *eventHub) Close() error {
+	eh.mu.Lock()
+	defer eh.mu.Unlock()
+
 	if !eh.initialized {
 		return nil
 	}

gateway/gateway-controller/pkg/eventhub/sqlite_backend.go

@@ -26,7 +26,7 @@ type SQLiteBackend struct {
 	wg            sync.WaitGroup
 
 	initialized bool
-	mu          sync.RWMutex
+	
 }
 
 // NewSQLiteBackend creates a new SQLite-based backend
@@ -44,8 +44,6 @@ func NewSQLiteBackend(db *sql.DB, logger *zap.Logger, config *SQLiteBackendConfi
 
 // Initialize sets up the SQLite backend and starts background workers
 func (b *SQLiteBackend) Initialize(ctx context.Context) error {
-	b.mu.Lock()
-	defer b.mu.Unlock()
 
 	if b.initialized {
 		return nil
@@ -216,8 +214,6 @@ func (b *SQLiteBackend) CleanupRange(ctx context.Context, from, to time.Time) er
 
 // Close gracefully shuts down the SQLite backend
 func (b *SQLiteBackend) Close() error {
-	b.mu.Lock()
-	defer b.mu.Unlock()
 
 	if !b.initialized {
 		return nil
@@ -305,8 +301,16 @@ func (b *SQLiteBackend) pollAllOrganizations() {
 		}
 
 		if len(events) > 0 {
-			b.deliverEvents(org, events)
+			 		if len(events) > 0 {
+			if b.deliverEvents(org, events) != nil{
+				org.updatePollState(state.VersionID, time.Now())
+			}
+			// If delivery failed (channel full), don't update timestamp
+			// so events will be retried on next poll
+		} else {
+			org.updatePollState(state.VersionID, time.Now())
 		}
+	}
 
 		org.updatePollState(state.VersionID, time.Now())
 	}
@@ -339,6 +343,7 @@ func (b *SQLiteBackend) getAllStates(ctx context.Context) ([]OrganizationState,
 
 // getEventsSince retrieves events for an organization after a given timestamp
 func (b *SQLiteBackend) getEventsSince(ctx context.Context, orgID string, since time.Time) ([]Event, error) {
+	// TODO: (VirajSalaka) Implement pagination if large number of events
 	query := `
 		SELECT processed_timestamp, originated_timestamp, event_type,
 		       action, entity_id, correlation_id, event_data
@@ -370,17 +375,18 @@ func (b *SQLiteBackend) getEventsSince(ctx context.Context, orgID string, since
 }
 
 // deliverEvents sends events to all subscribers of an organization
-func (b *SQLiteBackend) deliverEvents(org *organization, events []Event) {
+func (b *SQLiteBackend) deliverEvents(org *organization, events []Event) error {
 	subscribers := org.getSubscribers()
 
 	if len(subscribers) == 0 {
 		b.logger.Debug("No subscribers for organization",
 			zap.String("organization", string(org.id)),
 			zap.Int("events", len(events)),
 		)
-		return
+		return nil
 	}
 
+	// TODO: (VirajSalaka) One subscriber is considered here. Handle multiple subscribers properly.
 	for _, ch := range subscribers {
 		select {
 		case ch <- events:
@@ -389,12 +395,14 @@ func (b *SQLiteBackend) deliverEvents(org *organization, events []Event) {
 				zap.Int("events", len(events)),
 			)
 		default:
-			b.logger.Warn("Subscriber channel full, dropping events",
+			b.logger.Error("Subscriber channel full, dropping events",
 				zap.String("organization", string(org.id)),
 				zap.Int("events", len(events)),
 			)
+			return fmt.Errorf("subscriber channel full")
 		}
 	}
+	return nil
 }
 
 // cleanupLoop runs periodic cleanup of old events