worth a shot

.github/workflows/docker-image.yml

@@ -48,10 +48,10 @@ jobs:
 
       - name: Push Docker image
         run: |
-          # If on v1.0 branch, push to 'testing.v1.0' tag
+          # If on v1.0 branch, push to 'testingv1.0' tag
           if [[ "${{ github.ref }}" == "refs/heads/v1.0" ]]; then
-            docker tag ti1:${{ env.VERSION }} ${{ secrets.DOCKER_USERNAME }}/ti1:testing-v1.0
+            docker tag ti1:${{ env.VERSION }} ${{ secrets.DOCKER_USERNAME }}/ti1:testingv1.0
-            docker push ${{ secrets.DOCKER_USERNAME }}/ti1:testing.v1.0
+            docker push ${{ secrets.DOCKER_USERNAME }}/ti1:testingv1.0
           else
             # Always push to 'dev' tag for main and dev branches
             docker tag ti1:${{ env.VERSION }} ${{ secrets.DOCKER_USERNAME }}/ti1:dev

README.md

@@ -138,7 +138,7 @@ nano postgres_data/postgresql.conf
 Change the following values
 ```conf
 listen_addresses = '*'
-max_connections = 100
+max_connections = 200
 shared_buffers = 16GB
 work_mem = 256MB
 maintenance_work_mem = 2GB

config/conf.json

@@ -10,8 +10,8 @@
     "valkey": {
 		"host": "127.0.0.1",
 		"port": "6379",
-		"max_conns": 50,
+		"max_conns": 110,
-		"timeout_ms": 5000,
+		"timeout_ms": 2000,
         	"password": "the_valkey_password"
 	},
     "temp": "value"

config/db.go

@@ -27,10 +27,11 @@ func ConnectToPostgreSQL() (*sql.DB, error) {
 		return nil, err
 	}
 
-	// Set connection pool settings
+	// Set connection pool settings for high concurrency
-	db.SetMaxOpenConns(25)               // Maximum number of open connections to the database
+	db.SetMaxOpenConns(100)                // Maximum number of open connections to the database
-	db.SetMaxIdleConns(25)               // Maximum number of connections in the idle connection pool
+	db.SetMaxIdleConns(50)                 // Maximum number of connections in the idle connection pool
 	db.SetConnMaxLifetime(1 * time.Hour)   // Maximum amount of time a connection may be reused
+	db.SetConnMaxIdleTime(5 * time.Minute) // Maximum amount of time a connection may be idle
 
 	fmt.Println("Connection to PostgreSQL opened successfully :D")
 

database/EstimatedCall.go

@@ -6,11 +6,18 @@ import (
 	"database/sql"
 	"encoding/hex"
 	"fmt"
+	"sync"
 	"ti1/valki"
 
 	"github.com/valkey-io/valkey-go"
 )
 
+type CallResult struct {
+	ID     int
+	Action string
+	Error  error
+}
+
 func InsertOrUpdateEstimatedCall(ctx context.Context, db *sql.DB, values []interface{}, valkeyClient valkey.Client) (int, string, error) {
 	// Replace empty strings with nil for timestamp fields
 	for i, v := range values {
@@ -28,19 +35,15 @@ func InsertOrUpdateEstimatedCall(ctx context.Context, db *sql.DB, values []inter
 	}
 	hash := md5.Sum([]byte(valuesString))
 	hashString := hex.EncodeToString(hash[:])
-	//fmt.Println("HashString:", hashString)
 
 	estimatedVehicleJourneyID := values[0]
 	orderID := values[1]
 	key := fmt.Sprintf("%v.%v", estimatedVehicleJourneyID, orderID)
-	//fmt.Printf("Estimated Vehicle Journey ID: %v, Order ID: %v\n", estimatedVehicleJourneyID, orderID)
-
-	var err error
 
 	// Get the MD5 hash from Valkey
 	retrievedHash, err := valki.GetValkeyValue(ctx, valkeyClient, key)
 	if err != nil {
-		return 0, "", fmt.Errorf("failed to get value from Valkey: %v", err)
+		return 0, "", fmt.Errorf("failed to get value from Valkey: %w", err)
 	}
 
 	// Check if the retrieved value matches the original MD5 hash
@@ -64,26 +67,60 @@ func InsertOrUpdateEstimatedCall(ctx context.Context, db *sql.DB, values []inter
                 estimated_data = EXCLUDED.estimated_data
             RETURNING CASE WHEN xmax = 0 THEN 'insert' ELSE 'update' END, id;
         `
-		stmt, err := db.Prepare(query)
-		if err != nil {
-			return 0, "", fmt.Errorf("error preparing statement: %v", err)
-		}
-		defer stmt.Close()
 
 		err = valki.SetValkeyValue(ctx, valkeyClient, key, hashString)
 		if err != nil {
-			return 0, "", fmt.Errorf("failed to set value in Valkey: %v", err)
+			return 0, "", fmt.Errorf("failed to set value in Valkey: %w", err)
 		}
 
 		var action string
 		var id int
-		err = stmt.QueryRow(values...).Scan(&action, &id)
+		err = db.QueryRowContext(ctx, query, values...).Scan(&action, &id)
 		if err != nil {
-			return 0, "", fmt.Errorf("error executing statement: %v", err)
+			return 0, "", fmt.Errorf("error executing statement: %w", err)
 		}
 		return id, action, nil
-	} else {
+	}
-		//fmt.Printf("MATCH!!! Original Hash: %s, Retrieved Hash: %s\n", hashString, retrievedHash)
 	return 0, "none", nil
 }
+
+// BatchInsertEstimatedCalls processes multiple estimated calls concurrently
+func BatchInsertEstimatedCalls(ctx context.Context, db *sql.DB, batch [][]interface{}, valkeyClient valkey.Client, workerCount int) ([]CallResult, error) {
+	if len(batch) == 0 {
+		return nil, nil
+	}
+
+	results := make([]CallResult, len(batch))
+	jobs := make(chan int, len(batch))
+	var wg sync.WaitGroup
+
+	// Start workers
+	for w := 0; w < workerCount; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for idx := range jobs {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+					id, action, err := InsertOrUpdateEstimatedCall(ctx, db, batch[idx], valkeyClient)
+					results[idx] = CallResult{
+						ID:     id,
+						Action: action,
+						Error:  err,
+					}
+				}
+			}
+		}()
+	}
+
+	// Send jobs
+	for i := range batch {
+		jobs <- i
+	}
+	close(jobs)
+
+	wg.Wait()
+	return results, nil
 }

database/EstimatedVehicleJourney.go

@@ -1,10 +1,68 @@
 package database
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
+	"sync"
 )
 
+type EVJResult struct {
+	ID     int
+	Action string
+	Error  error
+	Index  int // To maintain order
+}
+
+// PreparedStatements holds reusable prepared statements
+type PreparedStatements struct {
+	evjStmt *sql.Stmt
+	ecStmt  *sql.Stmt
+	rcStmt  *sql.Stmt
+	mu      sync.Mutex
+}
+
+func NewPreparedStatements(db *sql.DB) (*PreparedStatements, error) {
+	evjQuery := `
+		INSERT INTO estimatedvehiclejourney (servicedelivery, recordedattime, lineref, directionref, datasource, datedvehiclejourneyref, vehiclemode, dataframeref, originref, destinationref, operatorref, vehicleref, cancellation, other, firstservicedelivery)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $1)
+		ON CONFLICT (lineref, directionref, datasource, datedvehiclejourneyref)
+		DO UPDATE SET
+			servicedelivery = EXCLUDED.servicedelivery,
+			recordedattime = EXCLUDED.recordedattime,
+			vehiclemode = COALESCE(EXCLUDED.vehiclemode, estimatedvehiclejourney.vehiclemode),
+			dataframeref = COALESCE(EXCLUDED.dataframeref, estimatedvehiclejourney.dataframeref),
+			originref = COALESCE(EXCLUDED.originref, estimatedvehiclejourney.originref),
+			destinationref = COALESCE(EXCLUDED.destinationref, estimatedvehiclejourney.destinationref),
+			operatorref = COALESCE(EXCLUDED.operatorref, estimatedvehiclejourney.operatorref),
+			vehicleref = COALESCE(EXCLUDED.vehicleref, estimatedvehiclejourney.vehicleref),
+			cancellation = COALESCE(EXCLUDED.cancellation, estimatedvehiclejourney.cancellation),
+			other = COALESCE(EXCLUDED.other, estimatedvehiclejourney.other)
+		RETURNING CASE WHEN xmax = 0 THEN 'insert' ELSE 'update' END, id;
+	`
+
+	evjStmt, err := db.Prepare(evjQuery)
+	if err != nil {
+		return nil, fmt.Errorf("failed to prepare EVJ statement: %w", err)
+	}
+
+	return &PreparedStatements{
+		evjStmt: evjStmt,
+	}, nil
+}
+
+func (ps *PreparedStatements) Close() {
+	if ps.evjStmt != nil {
+		ps.evjStmt.Close()
+	}
+	if ps.ecStmt != nil {
+		ps.ecStmt.Close()
+	}
+	if ps.rcStmt != nil {
+		ps.rcStmt.Close()
+	}
+}
+
 func InsertOrUpdateEstimatedVehicleJourney(db *sql.DB, values []interface{}) (int, string, error) {
 	query := `
 		INSERT INTO estimatedvehiclejourney (servicedelivery, recordedattime, lineref, directionref, datasource, datedvehiclejourneyref, vehiclemode, dataframeref, originref, destinationref, operatorref, vehicleref, cancellation, other, firstservicedelivery)
@@ -24,18 +82,54 @@ func InsertOrUpdateEstimatedVehicleJourney(db *sql.DB, values []interface{}) (in
 		RETURNING CASE WHEN xmax = 0 THEN 'insert' ELSE 'update' END, id;
 	`
 
-	stmt, err := db.Prepare(query)
-	if err != nil {
-		return 0, "", fmt.Errorf("error preparing statement: %v", err)
-	}
-	defer stmt.Close()
-
 	var action string
 	var id int
-	err = stmt.QueryRow(values...).Scan(&action, &id)
+	err := db.QueryRow(query, values...).Scan(&action, &id)
 	if err != nil {
-		return 0, "", fmt.Errorf("error executing statement: %v", err)
+		return 0, "", fmt.Errorf("error executing EVJ statement: %w", err)
 	}
 
 	return id, action, nil
 }
+
+// BatchInsertEVJ processes multiple EVJ inserts concurrently
+func BatchInsertEVJ(ctx context.Context, db *sql.DB, batch [][]interface{}, workerCount int) ([]EVJResult, error) {
+	if len(batch) == 0 {
+		return nil, nil
+	}
+
+	results := make([]EVJResult, len(batch))
+	jobs := make(chan int, len(batch))
+	var wg sync.WaitGroup
+
+	// Start workers
+	for w := 0; w < workerCount; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for idx := range jobs {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+					id, action, err := InsertOrUpdateEstimatedVehicleJourney(db, batch[idx])
+					results[idx] = EVJResult{
+						ID:     id,
+						Action: action,
+						Error:  err,
+						Index:  idx,
+					}
+				}
+			}
+		}()
+	}
+
+	// Send jobs
+	for i := range batch {
+		jobs <- i
+	}
+	close(jobs)
+
+	wg.Wait()
+	return results, nil
+}

database/RecordedCall.go

@@ -6,6 +6,7 @@ import (
 	"database/sql"
 	"encoding/hex"
 	"fmt"
+	"sync"
 	"ti1/valki"
 
 	"github.com/valkey-io/valkey-go"
@@ -33,12 +34,10 @@ func InsertOrUpdateRecordedCall(ctx context.Context, db *sql.DB, values []interf
 	orderID := values[1]
 	key := fmt.Sprintf("%v.%v", estimatedVehicleJourneyID, orderID)
 
-	var err error
-
 	// Get the MD5 hash from Valkey
 	retrievedHash, err := valki.GetValkeyValue(ctx, valkeyClient, key)
 	if err != nil {
-		return 0, "", fmt.Errorf("failed to get value from Valkey: %v", err)
+		return 0, "", fmt.Errorf("failed to get value from Valkey: %w", err)
 	}
 
 	// Check if the retrieved value matches the original MD5 hash
@@ -65,25 +64,60 @@ func InsertOrUpdateRecordedCall(ctx context.Context, db *sql.DB, values []interf
                 recorded_data = EXCLUDED.recorded_data
             RETURNING CASE WHEN xmax = 0 THEN 'insert' ELSE 'update' END, id;
         `
-		stmt, err := db.Prepare(query)
-		if err != nil {
-			return 0, "", fmt.Errorf("error preparing statement: %v", err)
-		}
-		defer stmt.Close()
 
 		err = valki.SetValkeyValue(ctx, valkeyClient, key, hashString)
 		if err != nil {
-			return 0, "", fmt.Errorf("failed to set value in Valkey: %v", err)
+			return 0, "", fmt.Errorf("failed to set value in Valkey: %w", err)
 		}
 
 		var action string
 		var id int
-		err = stmt.QueryRow(values...).Scan(&action, &id)
+		err = db.QueryRowContext(ctx, query, values...).Scan(&action, &id)
 		if err != nil {
-			return 0, "", fmt.Errorf("error executing statement: %v", err)
+			return 0, "", fmt.Errorf("error executing statement: %w", err)
 		}
 		return id, action, nil
-	} else {
+	}
 	return 0, "none", nil
 }
+
+// BatchInsertRecordedCalls processes multiple recorded calls concurrently
+func BatchInsertRecordedCalls(ctx context.Context, db *sql.DB, batch [][]interface{}, valkeyClient valkey.Client, workerCount int) ([]CallResult, error) {
+	if len(batch) == 0 {
+		return nil, nil
+	}
+
+	results := make([]CallResult, len(batch))
+	jobs := make(chan int, len(batch))
+	var wg sync.WaitGroup
+
+	// Start workers
+	for w := 0; w < workerCount; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for idx := range jobs {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+					id, action, err := InsertOrUpdateRecordedCall(ctx, db, batch[idx], valkeyClient)
+					results[idx] = CallResult{
+						ID:     id,
+						Action: action,
+						Error:  err,
+					}
+				}
+			}
+		}()
+	}
+
+	// Send jobs
+	for i := range batch {
+		jobs <- i
+	}
+	close(jobs)
+
+	wg.Wait()
+	return results, nil
 }

database/ServiceDeliveryDB.go

@@ -6,25 +6,18 @@ import (
 )
 
 func InsertServiceDelivery(db *sql.DB, responseTimestamp string, recordedAtTime string) (int, error) {
-	fmt.Println("Inserting ServiceDelivery...")
 	var id int
-
 	err := db.QueryRow("INSERT INTO public.ServiceDelivery (ResponseTimestamp, RecordedAtTime) VALUES ($1, $2) RETURNING ID", responseTimestamp, recordedAtTime).Scan(&id)
 	if err != nil {
-		fmt.Println(err)
+		return 0, fmt.Errorf("failed to insert service delivery: %w", err)
-		return 0, err
 	}
-	//fmt.Println("ServiceDelivery inserted successfully! (", id, ")")
 	return id, nil
 }
 
 func UpdateServiceDeliveryData(db *sql.DB, id int, data string) error {
-	fmt.Println("Updating ServiceDelivery data...")
 	_, err := db.Exec("UPDATE public.ServiceDelivery SET Data = $1 WHERE ID = $2", data, id)
 	if err != nil {
-		fmt.Println(err)
+		return fmt.Errorf("failed to update service delivery data: %w", err)
-		return err
 	}
-	fmt.Println("Finished with this ServiceDelivery!")
 	return nil
 }

export/database.go

@@ -6,12 +6,20 @@ import (
 	"fmt"
 	"log"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"ti1/config"
 	"ti1/data"
 	"ti1/database"
 )
 
+// DBData is the main entry point for data processing
 func DBData(data *data.Data) {
+	DBDataOptimized(data)
+}
+
+// DBDataOptimized processes data with concurrent workers for better performance
+func DBDataOptimized(data *data.Data) {
 	fmt.Println(data.ServiceDelivery.ResponseTimestamp)
 	fmt.Println(data.ServiceDelivery.EstimatedTimetableDelivery[0].EstimatedJourneyVersionFrame.RecordedAtTime)
 
@@ -37,17 +45,93 @@ func DBData(data *data.Data) {
 	}
 	fmt.Println("SID:", sid)
 
-	// counters
+	// Atomic counters for thread-safe counting
-	var insertCount, updateCount, totalCount, estimatedCallInsertCount, estimatedCallUpdateCount, estimatedCallNoneCount, recordedCallInsertCount, recordedCallUpdateCount, recordedCallNoneCount int
+	var insertCount, updateCount, estimatedCallInsertCount, estimatedCallUpdateCount, estimatedCallNoneCount, recordedCallInsertCount, recordedCallUpdateCount, recordedCallNoneCount int64
 
-	for _, journey := range data.ServiceDelivery.EstimatedTimetableDelivery[0].EstimatedJourneyVersionFrame.EstimatedVehicleJourney {
+	journeys := data.ServiceDelivery.EstimatedTimetableDelivery[0].EstimatedJourneyVersionFrame.EstimatedVehicleJourney
+	totalJourneys := len(journeys)
+	fmt.Printf("Processing %d journeys...\n", totalJourneys)
+
+	// Job structures
+	type evjJob struct {
+		index int
+	}
+
+	type callJob struct {
+		evjID  int
+		values []interface{}
+	}
+
+	// Channels
+	workerCount := 20 // Adjust based on your database and CPU
+	evjJobs := make(chan evjJob, workerCount*2)
+	estimatedCallJobs := make(chan callJob, workerCount*10)
+	recordedCallJobs := make(chan callJob, workerCount*10)
+
+	var wg sync.WaitGroup
+	var callWg sync.WaitGroup
+
+	// Start Estimated Call workers
+	for w := 0; w < workerCount; w++ {
+		callWg.Add(1)
+		go func() {
+			defer callWg.Done()
+			for job := range estimatedCallJobs {
+				id, action, err := database.InsertOrUpdateEstimatedCall(ctx, db, job.values, valkeyClient)
+				if err != nil {
+					log.Printf("Error inserting/updating estimated call: %v\n", err)
+					continue
+				}
+				if action == "insert" {
+					atomic.AddInt64(&estimatedCallInsertCount, 1)
+				} else if action == "update" {
+					atomic.AddInt64(&estimatedCallUpdateCount, 1)
+				} else if action == "none" {
+					atomic.AddInt64(&estimatedCallNoneCount, 1)
+				}
+				_ = id
+			}
+		}()
+	}
+
+	// Start Recorded Call workers
+	for w := 0; w < workerCount; w++ {
+		callWg.Add(1)
+		go func() {
+			defer callWg.Done()
+			for job := range recordedCallJobs {
+				id, action, err := database.InsertOrUpdateRecordedCall(ctx, db, job.values, valkeyClient)
+				if err != nil {
+					log.Printf("Error inserting/updating recorded call: %v\n", err)
+					continue
+				}
+				if action == "insert" {
+					atomic.AddInt64(&recordedCallInsertCount, 1)
+				} else if action == "update" {
+					atomic.AddInt64(&recordedCallUpdateCount, 1)
+				} else if action == "none" {
+					atomic.AddInt64(&recordedCallNoneCount, 1)
+				}
+				_ = id
+			}
+		}()
+	}
+
+	// Start EVJ workers
+	for w := 0; w < workerCount; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for job := range evjJobs {
+				journey := &journeys[job.index]
+
+				// Prepare values
 				var values []interface{}
 				var datedVehicleJourneyRef, otherJson string
 
 				values = append(values, sid)
 				values = append(values, journey.RecordedAtTime)
 				values = append(values, journey.LineRef)
-		//had to add to lowercase cus some values vary in case and it was causing duplicates
 				values = append(values, strings.ToLower(journey.DirectionRef))
 				values = append(values, journey.DataSource)
 
@@ -68,10 +152,8 @@ func DBData(data *data.Data) {
 				values = append(values, journey.VehicleRef)
 				values = append(values, journey.Cancellation)
 
-		// Create a map to hold the JSON object for the current journey
+				// Create JSON object
 				jsonObject := make(map[string]interface{})
-
-		// Add relevant fields to the JSON object
 				if journey.OriginName != "" {
 					jsonObject["OriginName"] = journey.OriginName
 				}
@@ -142,71 +224,60 @@ func DBData(data *data.Data) {
 					jsonObject["Via"] = journey.Via.PlaceName
 				}
 
-		// Convert the JSON object to a JSON string
 				jsonString, err := json.Marshal(jsonObject)
 				if err != nil {
-			log.Fatal(err)
+					log.Printf("Error marshaling JSON: %v\n", err)
+					continue
 				}
 				otherJson = string(jsonString)
 				values = append(values, otherJson)
 
-		// Insert or update the record
+				// Insert or update EVJ
 				id, action, err := database.InsertOrUpdateEstimatedVehicleJourney(db, values)
 				if err != nil {
-			fmt.Printf("Error inserting/updating estimated vehicle journey: %v\n", err)
+					log.Printf("Error inserting/updating estimated vehicle journey: %v\n", err)
-		} else {
+					continue
-			if 1 == 0 {
-				fmt.Printf("Action: %s, ID: %d\n", action, id)
 				}
 
 				if action == "insert" {
-				insertCount++
+					atomic.AddInt64(&insertCount, 1)
 				} else if action == "update" {
-				updateCount++
+					atomic.AddInt64(&updateCount, 1)
 				}
-			totalCount = insertCount + updateCount
 
-			//fmt.Printf("Inserts: %d, Updates: %d, Total: %d\n", insertCount, updateCount, totalCount)
+				// Progress reporting
-			if totalCount%1000 == 0 {
+				total := atomic.AddInt64(&insertCount, 0) + atomic.AddInt64(&updateCount, 0)
+				if total%1000 == 0 {
 					fmt.Printf(
-					"Inserts: %d, Updates: %d, Total: %d; estimatedCalls = I: %d U: %d N: %d; recordedCalls = I: %d U: %d N: %d\n",
+						"EVJ - I: %d, U: %d, Total: %d; EstCalls - I: %d U: %d N: %d; RecCalls - I: %d U: %d N: %d\n",
-					insertCount,
+						atomic.LoadInt64(&insertCount),
-					updateCount,
+						atomic.LoadInt64(&updateCount),
-					totalCount,
+						total,
-					estimatedCallInsertCount,
+						atomic.LoadInt64(&estimatedCallInsertCount),
-					estimatedCallUpdateCount,
+						atomic.LoadInt64(&estimatedCallUpdateCount),
-					estimatedCallNoneCount,
+						atomic.LoadInt64(&estimatedCallNoneCount),
-					recordedCallInsertCount,
+						atomic.LoadInt64(&recordedCallInsertCount),
-					recordedCallUpdateCount,
+						atomic.LoadInt64(&recordedCallUpdateCount),
-					recordedCallNoneCount,
+						atomic.LoadInt64(&recordedCallNoneCount),
 					)
 				}
-		}
 
+				// Process Estimated Calls
 				for _, estimatedCall := range journey.EstimatedCalls {
 					for _, call := range estimatedCall.EstimatedCall {
 						var estimatedValues []interface{}
 
-				//1 estimatedvehiclejourney
 						estimatedValues = append(estimatedValues, id)
-				//2 order
 						estimatedValues = append(estimatedValues, call.Order)
-				//3 stoppointref
 						estimatedValues = append(estimatedValues, call.StopPointRef)
-				//4 aimeddeparturetime
 						estimatedValues = append(estimatedValues, call.AimedDepartureTime)
-				//5 expecteddeparturetime
 						estimatedValues = append(estimatedValues, call.ExpectedDepartureTime)
-				//6 aimedarrivaltime
 						estimatedValues = append(estimatedValues, call.AimedArrivalTime)
-				//7 expectedarrivaltime
 						estimatedValues = append(estimatedValues, call.ExpectedArrivalTime)
-				//8 cancellation
 						estimatedValues = append(estimatedValues, call.Cancellation)
 
-				//9 estimated_data (JSON)
+						// estimated_data JSON
 						estimatedJsonObject := make(map[string]interface{})
-				// data allrady loged
 						if call.ExpectedDepartureTime != "" {
 							estimatedJsonObject["ExpectedDepartureTime"] = call.ExpectedDepartureTime
 						}
@@ -216,7 +287,6 @@ func DBData(data *data.Data) {
 						if call.Cancellation != "" {
 							estimatedJsonObject["Cancellation"] = call.Cancellation
 						}
-				// The rest
 						if call.StopPointName != "" {
 							estimatedJsonObject["StopPointName"] = call.StopPointName
 						}
@@ -287,66 +357,41 @@ func DBData(data *data.Data) {
 							estimatedJsonObject["Occupancy"] = call.Occupancy
 						}
 
-				// Convert the JSON object to a JSON string
 						jsonString, err := json.Marshal(estimatedJsonObject)
 						if err != nil {
-					log.Fatal(err)
+							log.Printf("Error marshaling estimated call JSON: %v\n", err)
+							continue
 						}
 						estimatedValues = append(estimatedValues, string(jsonString))
 
-				// Insert or update the record
+						// Convert to string values
-				stringValues := make([]string, len(estimatedValues))
+						interfaceValues := make([]interface{}, len(estimatedValues))
 						for i, v := range estimatedValues {
-					stringValues[i] = fmt.Sprintf("%v", v)
+							interfaceValues[i] = fmt.Sprintf("%v", v)
-				}
-				interfaceValues := make([]interface{}, len(stringValues))
-				for i, v := range stringValues {
-					interfaceValues[i] = v
-				}
-				id, action, err := database.InsertOrUpdateEstimatedCall(ctx, db, interfaceValues, valkeyClient)
-				if err != nil {
-					log.Fatalf("Failed to insert or update estimated call: %v", err)
-				} else {
-					if 1 == 0 {
-						fmt.Printf("Action: %s, ID: %d\n", action, id)
 						}
 
-					if action == "insert" {
+						// Send to worker pool
-						estimatedCallInsertCount++
+						estimatedCallJobs <- callJob{evjID: id, values: interfaceValues}
-					} else if action == "update" {
-						estimatedCallUpdateCount++
-					} else if action == "none" {
-						estimatedCallNoneCount++
-					}
-				}
 					}
 				}
+
+				// Process Recorded Calls
 				for _, recordedCall := range journey.RecordedCalls {
 					for _, call := range recordedCall.RecordedCall {
 						var recordedValues []interface{}
 
-				//1 estimatedvehiclejourney
 						recordedValues = append(recordedValues, id)
-				//2 order
 						recordedValues = append(recordedValues, call.Order)
-				//3 stoppointref
 						recordedValues = append(recordedValues, call.StopPointRef)
-				//4 aimeddeparturetime
 						recordedValues = append(recordedValues, call.AimedDepartureTime)
-				//5 expecteddeparturetime
 						recordedValues = append(recordedValues, call.ExpectedDepartureTime)
-				//6 aimedarrivaltime
 						recordedValues = append(recordedValues, call.AimedArrivalTime)
-				//7 expectedarrivaltime
 						recordedValues = append(recordedValues, call.ExpectedArrivalTime)
-				//8 cancellation
 						recordedValues = append(recordedValues, call.Cancellation)
-				//9 actualdeparturetime
 						recordedValues = append(recordedValues, call.ActualDepartureTime)
-				//10 actualarrivaltime
 						recordedValues = append(recordedValues, call.ActualArrivalTime)
 
-				//11 recorded_data (JSON)
+						// recorded_data JSON
 						recordedJsonObject := make(map[string]interface{})
 						if call.StopPointName != "" {
 							recordedJsonObject["StopPointName"] = call.StopPointName
@@ -364,68 +409,67 @@ func DBData(data *data.Data) {
 							recordedJsonObject["Occupancy"] = call.Occupancy
 						}
 
-				// Convert the JSON object to a JSON string
 						jsonString, err := json.Marshal(recordedJsonObject)
 						if err != nil {
-					log.Fatal(err)
+							log.Printf("Error marshaling recorded call JSON: %v\n", err)
+							continue
 						}
 						recordedValues = append(recordedValues, string(jsonString))
 
-				// Insert or update the record
+						// Convert to string values
-				stringValues := make([]string, len(recordedValues))
+						interfaceValues := make([]interface{}, len(recordedValues))
 						for i, v := range recordedValues {
-					stringValues[i] = fmt.Sprintf("%v", v)
+							interfaceValues[i] = fmt.Sprintf("%v", v)
-				}
-				interfaceValues := make([]interface{}, len(stringValues))
-				for i, v := range stringValues {
-					interfaceValues[i] = v
 						}
 
-				id, action, err := database.InsertOrUpdateRecordedCall(ctx, db, interfaceValues, valkeyClient)
+						// Send to worker pool
-				if err != nil {
+						recordedCallJobs <- callJob{evjID: id, values: interfaceValues}
-					fmt.Printf("Error inserting/updating recorded call: %v\n", err)
+					}
-				} else {
+				}
-					if 1 == 0 {
+			}
-						fmt.Printf("Action: %s, ID: %d\n", action, id)
+		}()
 	}
 
-					if action == "insert" {
+	// Send all EVJ jobs
-						recordedCallInsertCount++
+	for i := range journeys {
-						//fmt.Printf("Action: %s, ID: %d\n", action, id)
+		evjJobs <- evjJob{index: i}
-					} else if action == "update" {
-						recordedCallUpdateCount++
-					} else if action == "none" {
-						recordedCallNoneCount++
-					}
-				}
-			}
 	}
+	close(evjJobs)
 
-	}
+	// Wait for EVJ processing to complete
+	wg.Wait()
+
+	// Close call job channels and wait for call processing to complete
+	close(estimatedCallJobs)
+	close(recordedCallJobs)
+	callWg.Wait()
+
+	// Print final stats
 	fmt.Printf(
-		"DONE: Inserts: %d, Updates: %d, Total: %d; estimatedCalls = I: %d U: %d N: %d; recordedCalls = I: %d U: %d N: %d\n",
+		"\nDONE: EVJ - Inserts: %d, Updates: %d, Total: %d\n"+
-		insertCount,
+			"      EstimatedCalls - I: %d U: %d N: %d\n"+
-		updateCount,
+			"      RecordedCalls  - I: %d U: %d N: %d\n",
-		totalCount,
+		atomic.LoadInt64(&insertCount),
-		estimatedCallInsertCount,
+		atomic.LoadInt64(&updateCount),
-		estimatedCallUpdateCount,
+		atomic.LoadInt64(&insertCount)+atomic.LoadInt64(&updateCount),
-		estimatedCallNoneCount,
+		atomic.LoadInt64(&estimatedCallInsertCount),
-		recordedCallInsertCount,
+		atomic.LoadInt64(&estimatedCallUpdateCount),
-		recordedCallUpdateCount,
+		atomic.LoadInt64(&estimatedCallNoneCount),
-		recordedCallNoneCount,
+		atomic.LoadInt64(&recordedCallInsertCount),
+		atomic.LoadInt64(&recordedCallUpdateCount),
+		atomic.LoadInt64(&recordedCallNoneCount),
 	)
+
 	// Create map to hold JSON
 	serviceDeliveryJsonObject := make(map[string]interface{})
-
+	serviceDeliveryJsonObject["Inserts"] = atomic.LoadInt64(&insertCount)
-	// Add fields to JSON
+	serviceDeliveryJsonObject["Updates"] = atomic.LoadInt64(&updateCount)
-	serviceDeliveryJsonObject["Inserts"] = insertCount
+	serviceDeliveryJsonObject["EstimatedCallInserts"] = atomic.LoadInt64(&estimatedCallInsertCount)
-	serviceDeliveryJsonObject["Updates"] = updateCount
+	serviceDeliveryJsonObject["EstimatedCallUpdates"] = atomic.LoadInt64(&estimatedCallUpdateCount)
-	serviceDeliveryJsonObject["EstimatedCallInserts"] = estimatedCallInsertCount
+	serviceDeliveryJsonObject["EstimatedCallNone"] = atomic.LoadInt64(&estimatedCallNoneCount)
-	serviceDeliveryJsonObject["EstimatedCallUpdates"] = estimatedCallUpdateCount
+	serviceDeliveryJsonObject["RecordedCallInserts"] = atomic.LoadInt64(&recordedCallInsertCount)
-	serviceDeliveryJsonObject["EstimatedCallNone"] = estimatedCallNoneCount
+	serviceDeliveryJsonObject["RecordedCallUpdates"] = atomic.LoadInt64(&recordedCallUpdateCount)
-	serviceDeliveryJsonObject["RecordedCallInserts"] = recordedCallInsertCount
+	serviceDeliveryJsonObject["RecordedCallNone"] = atomic.LoadInt64(&recordedCallNoneCount)
-	serviceDeliveryJsonObject["RecordedCallUpdates"] = recordedCallUpdateCount
-	serviceDeliveryJsonObject["RecordedCallNone"] = recordedCallNoneCount
 
 	// Convert JSON object to JSON string
 	serviceDeliveryJsonString, err := json.Marshal(serviceDeliveryJsonObject)
@@ -438,4 +482,6 @@ func DBData(data *data.Data) {
 	if err != nil {
 		log.Fatal(err)
 	}
+
+	fmt.Println("Finished with this ServiceDelivery!")
 }