Merge pull request #34 from retail-ai-inc/fix/error-logging-and-changestream

Fix/error logging and changestream

Author: tanvir-retailai

README.md

@@ -4,6 +4,7 @@
 [![Coverage Status](https://codecov.io/gh/retail-ai-inc/sync/graph/badge.svg)](https://codecov.io/gh/retail-ai-inc/sync)
 [![GoDoc](https://godoc.org/github.com/retail-ai-inc/sync?status.svg)](https://godoc.org/github.com/retail-ai-inc/sync)
 [![License](https://img.shields.io/badge/license-MIT-blue)](LICENSE)
+[![Ask DeepWiki](https://deepwiki.com/badge.svg)](https://deepwiki.com/retail-ai-inc/sync)
 
 Synchronize Production NOSQL and SQL data to Standalone instances for Data scientists or other purposes. A **Go-based** tool to synchronize MongoDB or SQL data from a **MongoDB replica set** or **sharded cluster** or production SQL instance to a **standalone instance**, supports initial and incremental synchronization, including **indexes** with change stream monitoring.
 

READMEUI.md

@@ -13,8 +13,11 @@ Sync Data Platform provides an intuitive interface for managing and monitoring d
 
 ## Main Features
 
+### Dashboard
+<img width="800" alt="image" src="https://github.com/user-attachments/assets/9a874651-0832-48ba-8cd3-1137e6da5afe" />
+
 ### Task Management
-<img width="800" alt="image" src="https://github.com/user-attachments/assets/08c9d723-8dca-4dd8-9ca9-2459c2072858" />
+<img width="800" alt="image" src="https://github.com/user-attachments/assets/597c5987-3eea-4a0a-ad4d-39c7d4594868" />
 
 
 - View task status and details
@@ -23,7 +26,7 @@ Sync Data Platform provides an intuitive interface for managing and monitoring d
 - Operate tasks (start/stop/monitor)
 
 ### Monitoring Dashboard
-<img width="800" alt="image" src="https://github.com/user-attachments/assets/2a1bd9f1-0d46-4961-868f-80dab83b3705" />
+<img width="800" alt="image" src="https://github.com/user-attachments/assets/e3670689-5662-4509-937e-6758a6536cce" />
 <img width="800" alt="image" src="https://github.com/user-attachments/assets/a38eed54-96ac-4bda-9191-29bc25cdef06" />
 
 - **Key Metrics**:

pkg/api/monitor_handler.go

@@ -2,8 +2,8 @@ package api
 
 import (
 	"database/sql"
-	"encoding/json"
 	"errors"
+	"fmt"
 	"net/http"
 	"strings"
 	"time"
@@ -371,157 +371,100 @@ func ChangeStreamsStatusHandler(w http.ResponseWriter, r *http.Request) {
 	}
 	defer db.Close()
 
-	// Query recent 2 hours of Comprehensive ChangeStream Status logs
-	// Get the latest record for each sync_task_id
+	// Query changestream_statistics table directly
 	query := `
-SELECT t1.sync_task_id, t1.message, t1.log_time
-FROM sync_log t1
-WHERE t1.message LIKE '%[MongoDB] Comprehensive ChangeStream Status%'
-  AND t1.log_time >= datetime('now', '-2 hours')
-  AND t1.log_time = (
-    SELECT MAX(t2.log_time)
-    FROM sync_log t2
-    WHERE t2.sync_task_id = t1.sync_task_id
-      AND t2.message LIKE '%[MongoDB] Comprehensive ChangeStream Status%'
-      AND t2.log_time >= datetime('now', '-2 hours')
-  )
-ORDER BY t1.sync_task_id
+SELECT 
+	task_id,
+	collection_name,
+	received,
+	executed,
+	pending,
+	errors,
+	inserted,
+	updated,
+	deleted,
+	last_updated,
+	created_at
+FROM changestream_statistics
+ORDER BY task_id, collection_name
 `
 
 	rows, err := db.Query(query)
 	if err != nil {
-		errorJSON(w, "query sync_log fail", err)
+		errorJSON(w, "query changestream_statistics fail", err)
 		return
 	}
 	defer rows.Close()
 
-	// Track processed task IDs to avoid duplicates
-	processedTasks := make(map[string]bool)
-
 	// Aggregated data
 	totalReceived := 0
 	totalExecuted := 0
 	totalPending := 0
+	totalErrors := 0
 	totalActiveStreams := 0
 	allChangeStreams := make([]map[string]interface{}, 0)
 	lastUpdated := ""
+	taskIDs := make(map[int]bool)
 
 	for rows.Next() {
-		var taskID, message, logTime string
-		if err := rows.Scan(&taskID, &message, &logTime); err != nil {
-			continue
-		}
-
-		// Skip if we've already processed this task (take only the latest)
-		if processedTasks[taskID] {
-			continue
-		}
-		processedTasks[taskID] = true
-
-		// Extract JSON from message
-		jsonStart := strings.Index(message, "{")
-		if jsonStart == -1 {
-			continue
-		}
-
-		// Find the end of JSON by counting braces
-		jsonStr := ""
-		braceCount := 0
-		inString := false
-		escaped := false
-
-		for i := jsonStart; i < len(message); i++ {
-			char := message[i]
-
-			if escaped {
-				escaped = false
-				continue
-			}
-
-			if char == '\\' {
-				escaped = true
-				continue
-			}
+		var taskID int
+		var collectionName string
+		var received, executed, pending, errors, inserted, updated, deleted int
+		var lastUpdatedTime, createdAt string
 
-			if char == '"' {
-				inString = !inString
-				continue
-			}
-
-			if !inString {
-				if char == '{' {
-					braceCount++
-				} else if char == '}' {
-					braceCount--
-					if braceCount == 0 {
-						jsonStr = message[jsonStart : i+1]
-						break
-					}
-				}
-			}
-		}
-
-		if jsonStr == "" {
+		if err := rows.Scan(&taskID, &collectionName, &received, &executed, &pending, &errors, &inserted, &updated, &deleted, &lastUpdatedTime, &createdAt); err != nil {
 			continue
 		}
 
-		// Parse JSON
-		var statusData map[string]interface{}
-		if err := json.Unmarshal([]byte(jsonStr), &statusData); err != nil {
+		// Skip Task ID=0 to filter out legacy data
+		if taskID == 0 {
 			continue
 		}
 
-		// Extract summary data
-		if summary, ok := statusData["summary"].(map[string]interface{}); ok {
-			if received, ok := summary["total_received"].(float64); ok {
-				totalReceived += int(received)
-			}
-			if executed, ok := summary["total_executed"].(float64); ok {
-				totalExecuted += int(executed)
-			}
-			if pending, ok := summary["total_pending"].(float64); ok {
-				totalPending += int(pending)
-			}
+		// Track unique task IDs
+		taskIDs[taskID] = true
+
+		// Aggregate summary data
+		totalReceived += received
+		totalExecuted += executed
+		totalPending += pending
+		totalErrors += errors
+		totalActiveStreams++
+
+		// Create changestream detail (maintain original API format)
+		csDetail := map[string]interface{}{
+			"task_id":  fmt.Sprintf("%d", taskID), // Convert to string format
+			"name":     collectionName,            // Use "name" instead of "collection_name"
+			"received": received,
+			"executed": executed,
+			"pending":  pending,
+			"errors":   errors,
+			"operations": map[string]interface{}{ // Nest operations object
+				"inserted": inserted,
+				"updated":  updated,
+				"deleted":  deleted,
+			},
 		}
 
-		// Extract changestreams data
-		if changestreams, ok := statusData["changestreams"].(map[string]interface{}); ok {
-			if activeCount, ok := changestreams["active_count"].(float64); ok {
-				totalActiveStreams += int(activeCount)
-			}
-
-			if details, ok := changestreams["details"].([]interface{}); ok {
-				for _, detail := range details {
-					if detailMap, ok := detail.(map[string]interface{}); ok {
-						// Add task_id to each changestream for identification
-						csDetail := make(map[string]interface{})
-						for k, v := range detailMap {
-							csDetail[k] = v
-						}
-						csDetail["task_id"] = taskID
-						allChangeStreams = append(allChangeStreams, csDetail)
-					}
-				}
-			}
-		}
+		allChangeStreams = append(allChangeStreams, csDetail)
 
-		// Update last updated time
-		if lastUpdated == "" || logTime > lastUpdated {
-			lastUpdated = logTime
+		// Update last updated time (take the latest)
+		if lastUpdated == "" || lastUpdatedTime > lastUpdated {
+			lastUpdated = lastUpdatedTime
 		}
 	}
 
 	if err := rows.Err(); err != nil {
-		errorJSON(w, "sync_log iteration error", err)
+		errorJSON(w, "changestream_statistics iteration error", err)
 		return
 	}
 
-	// Calculate processing rate (simplified calculation)
-	processingRate := "0/sec"
+	// Calculate processing rate based on total received/executed over time
+	processingRate := "N/A"
 	if totalActiveStreams > 0 && totalExecuted > 0 {
-		// This is a simplified rate calculation
-		// In a real scenario, you might want to calculate based on time windows
-		processingRate = "N/A" // We need time-based calculation for accurate rate
+		// Simple calculation: assume data represents recent activity
+		// For more accurate rate, we would need time-based windows
+		// processingRate = fmt.Sprintf("~%d/min", totalExecuted)
 	}
 
 	// Prepare response
@@ -537,7 +480,7 @@ ORDER BY t1.sync_task_id
 			},
 			"changestreams": allChangeStreams,
 			"last_updated":  lastUpdated,
-			"tasks_count":   len(processedTasks),
+			"tasks_count":   len(taskIDs),
 		},
 	}