Merge remote-tracking branch 'origin/main' into v1.0

.github/workflows/docker-image.yml

@@ -55,4 +55,4 @@ jobs:
           if [[ "${{ env.VERSION }}" != "dev" ]]; then
             docker tag ti1:${{ env.VERSION }} ${{ secrets.DOCKER_USERNAME }}/ti1:${{ env.VERSION }}
             docker push ${{ secrets.DOCKER_USERNAME }}/ti1:${{ env.VERSION }}
-          fi
+          fi

Dockerfile

@@ -1,5 +1,5 @@
 # Use the official Golang image as the base image
-FROM golang:1.26.0
+FROM golang:1.23.4
 
 # Set the Current Working Directory inside the container
 WORKDIR /app

databaseold/EstimatedCall.go

@@ -0,0 +1,89 @@
+package databaseold
+
+import (
+	"context"
+	"crypto/md5"
+	"database/sql"
+	"encoding/hex"
+	"fmt"
+	"ti1/valki"
+
+	"github.com/valkey-io/valkey-go"
+)
+
+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 {
+		if str, ok := v.(string); ok && str == "" {
+			values[i] = nil
+		}
+	}
+
+	// Convert values to a single string and hash it using MD5
+	var valuesString string
+	for _, v := range values {
+		if v != nil {
+			valuesString += fmt.Sprintf("%v", v)
+		}
+	}
+	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)
+	}
+
+	// Check if the retrieved value matches the original MD5 hash
+	if retrievedHash != hashString {
+		query := `
+            INSERT INTO calls (
+                estimatedvehiclejourney, "order", stoppointref,
+                aimeddeparturetime, expecteddeparturetime,
+                aimedarrivaltime, expectedarrivaltime,
+                cancellation, estimated_data
+            )
+            VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
+            ON CONFLICT (estimatedvehiclejourney, "order")
+            DO UPDATE SET
+                stoppointref = EXCLUDED.stoppointref,
+                aimeddeparturetime = EXCLUDED.aimeddeparturetime,
+                expecteddeparturetime = EXCLUDED.expecteddeparturetime,
+                aimedarrivaltime = EXCLUDED.aimedarrivaltime,
+                expectedarrivaltime = EXCLUDED.expectedarrivaltime,
+                cancellation = EXCLUDED.cancellation,
+                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)
+		}
+
+		var action string
+		var id int
+		err = stmt.QueryRow(values...).Scan(&action, &id)
+		if err != nil {
+			return 0, "", fmt.Errorf("error executing statement: %v", err)
+		}
+		return id, action, nil
+	} else {
+		//fmt.Printf("MATCH!!! Original Hash: %s, Retrieved Hash: %s\n", hashString, retrievedHash)
+		return 0, "none", nil
+	}
+}

databaseold/EstimatedVehicleJourney.go

@@ -0,0 +1,41 @@
+package databaseold
+
+import (
+	"database/sql"
+	"fmt"
+)
+
+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)
+	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;
+	`
+
+	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)
+	if err != nil {
+		return 0, "", fmt.Errorf("error executing statement: %v", err)
+	}
+
+	return id, action, nil
+}

databaseold/RecordedCall.go

@@ -0,0 +1,89 @@
+package databaseold
+
+import (
+	"context"
+	"crypto/md5"
+	"database/sql"
+	"encoding/hex"
+	"fmt"
+	"ti1/valki"
+
+	"github.com/valkey-io/valkey-go"
+)
+
+func InsertOrUpdateRecordedCall(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 {
+		if str, ok := v.(string); ok && str == "" {
+			values[i] = nil
+		}
+	}
+
+	// Convert values to a single string and hash it using MD5
+	var valuesString string
+	for _, v := range values {
+		if v != nil {
+			valuesString += fmt.Sprintf("%v", v)
+		}
+	}
+	hash := md5.Sum([]byte(valuesString))
+	hashString := hex.EncodeToString(hash[:])
+
+	estimatedVehicleJourneyID := values[0]
+	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)
+	}
+
+	// Check if the retrieved value matches the original MD5 hash
+	if retrievedHash != hashString {
+		query := `
+            INSERT INTO calls (
+                estimatedvehiclejourney, "order", stoppointref,
+                aimeddeparturetime, expecteddeparturetime,
+                aimedarrivaltime, expectedarrivaltime,
+                cancellation, actualdeparturetime, actualarrivaltime,
+                recorded_data
+            )
+            VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
+            ON CONFLICT (estimatedvehiclejourney, "order")
+            DO UPDATE SET
+                stoppointref = EXCLUDED.stoppointref,
+                aimeddeparturetime = EXCLUDED.aimeddeparturetime,
+                expecteddeparturetime = EXCLUDED.expecteddeparturetime,
+                aimedarrivaltime = EXCLUDED.aimedarrivaltime,
+                expectedarrivaltime = EXCLUDED.expectedarrivaltime,
+                cancellation = EXCLUDED.cancellation,
+                actualdeparturetime = EXCLUDED.actualdeparturetime,
+                actualarrivaltime = EXCLUDED.actualarrivaltime,
+                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)
+		}
+
+		var action string
+		var id int
+		err = stmt.QueryRow(values...).Scan(&action, &id)
+		if err != nil {
+			return 0, "", fmt.Errorf("error executing statement: %v", err)
+		}
+		return id, action, nil
+	} else {
+		return 0, "none", nil
+	}
+}

databaseold/ServiceDeliveryDB.go

@@ -0,0 +1,30 @@
+package databaseold
+
+import (
+	"database/sql"
+	"fmt"
+)
+
+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, 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 err
+	}
+	fmt.Println("Finished with this ServiceDelivery!")
+	return nil
+}

databaseold/SetupDB.go

@@ -0,0 +1,140 @@
+package databaseold
+
+import (
+	"fmt"
+	"ti1/config"
+)
+
+func SetupDB() error {
+	fmt.Println("Setting up the database...")
+
+	// Connect to PostgreSQL
+	db, err := config.ConnectToPostgreSQL()
+	if err != nil {
+		return fmt.Errorf("failed to connect to database: %w", err)
+	}
+	defer config.DisconnectFromPostgreSQL(db)
+
+	// Create sequences if they do not exist
+	sequences := []string{
+		"CREATE SEQUENCE IF NOT EXISTS public.calls_id_seq",
+		"CREATE SEQUENCE IF NOT EXISTS public.estimatedvehiclejourney_id_seq",
+		"CREATE SEQUENCE IF NOT EXISTS public.servicedelivery_id_seq",
+	}
+
+	for _, seq := range sequences {
+		_, err := db.Exec(seq)
+		if err != nil {
+			return fmt.Errorf("failed to create sequence: %w", err)
+		}
+	}
+
+	// Check if tables exist and have the correct structure
+	tables := map[string]string{
+		"calls": `CREATE TABLE IF NOT EXISTS public.calls (
+			id BIGINT PRIMARY KEY DEFAULT nextval('public.calls_id_seq'),
+			estimatedvehiclejourney BIGINT,
+			"order" INTEGER,
+			stoppointref VARCHAR,
+			aimeddeparturetime TIMESTAMP,
+			expecteddeparturetime TIMESTAMP,
+			aimedarrivaltime TIMESTAMP,
+			expectedarrivaltime TIMESTAMP,
+			cancellation VARCHAR,
+			actualdeparturetime TIMESTAMP,
+			actualarrivaltime TIMESTAMP,
+			estimated_data JSON,
+			recorded_data JSON
+		);`,
+		"estimatedvehiclejourney": `CREATE TABLE IF NOT EXISTS public.estimatedvehiclejourney (
+			id BIGINT PRIMARY KEY DEFAULT nextval('public.estimatedvehiclejourney_id_seq'),
+			servicedelivery INTEGER,
+			recordedattime TIMESTAMP,
+			lineref VARCHAR,
+			directionref VARCHAR,
+			datasource VARCHAR,
+			datedvehiclejourneyref VARCHAR,
+			vehiclemode VARCHAR,
+			dataframeref VARCHAR,
+			originref VARCHAR,
+			destinationref VARCHAR,
+			operatorref VARCHAR,
+			vehicleref VARCHAR,
+			cancellation VARCHAR,
+			other JSON,
+			firstservicedelivery INTEGER
+		);`,
+		"servicedelivery": `CREATE TABLE IF NOT EXISTS public.servicedelivery (
+			id INTEGER PRIMARY KEY DEFAULT nextval('public.servicedelivery_id_seq'),
+			responsetimestamp TIMESTAMPTZ,
+			recordedattime TIMESTAMPTZ,
+			data JSON
+		);`,
+	}
+
+	for table, createStmt := range tables {
+		var exists bool
+		err := db.QueryRow(fmt.Sprintf("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = '%s')", table)).Scan(&exists)
+		if err != nil {
+			return fmt.Errorf("failed to check if table %s exists: %w", table, err)
+		}
+
+		if !exists {
+			_, err := db.Exec(createStmt)
+			if err != nil {
+				return fmt.Errorf("failed to create table %s: %w", table, err)
+			}
+			fmt.Printf("Table %s created successfully!\n", table)
+		} else {
+			fmt.Printf("Table %s already exists.\n", table)
+		}
+	}
+
+	// Check if the unique constraint exists before adding it to calls table
+	var constraintExists bool
+	err = db.QueryRow(`
+		SELECT EXISTS (
+			SELECT 1
+			FROM pg_constraint
+			WHERE conname = 'unique_estimatedvehiclejourney_order'
+		);
+	`).Scan(&constraintExists)
+	if err != nil {
+		return fmt.Errorf("failed to check if unique constraint exists: %w", err)
+	}
+
+	if !constraintExists {
+		_, err = db.Exec(`ALTER TABLE calls ADD CONSTRAINT unique_estimatedvehiclejourney_order UNIQUE (estimatedvehiclejourney, "order");`)
+		if err != nil {
+			return fmt.Errorf("failed to add unique constraint to calls table: %w", err)
+		}
+		fmt.Println("Unique constraint added to calls table.")
+	} else {
+		fmt.Println("Unique constraint already exists on calls table.")
+	}
+
+	// Check if the unique constraint exists before adding it to estimatedvehiclejourney table
+	err = db.QueryRow(`
+		SELECT EXISTS (
+			SELECT 1
+			FROM pg_constraint
+			WHERE conname = 'unique_lineref_directionref_datasource_datedvehiclejourneyref'
+		);
+	`).Scan(&constraintExists)
+	if err != nil {
+		return fmt.Errorf("failed to check if unique constraint exists: %w", err)
+	}
+
+	if !constraintExists {
+		_, err = db.Exec(`ALTER TABLE estimatedvehiclejourney ADD CONSTRAINT unique_lineref_directionref_datasource_datedvehiclejourneyref UNIQUE (lineref, directionref, datasource, datedvehiclejourneyref);`)
+		if err != nil {
+			return fmt.Errorf("failed to add unique constraint to estimatedvehiclejourney table: %w", err)
+		}
+		fmt.Println("Unique constraint added to estimatedvehiclejourney table.")
+	} else {
+		fmt.Println("Unique constraint already exists on estimatedvehiclejourney table.")
+	}
+
+	fmt.Println("Database setup is good!")
+	return nil
+}

go.mod

@@ -1,10 +1,10 @@
 module ti1
 
-go 1.26.0
+go 1.23.4
+
+require github.com/lib/pq v1.10.9
 
 require (
-	github.com/lib/pq v1.11.2
+	github.com/valkey-io/valkey-go v1.0.52 // indirect
-	github.com/valkey-io/valkey-go v1.0.71
+	golang.org/x/sys v0.24.0 // indirect
 )
-
-require golang.org/x/sys v0.41.0 // indirect

go.sum

@@ -1,12 +1,6 @@
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
-github.com/lib/pq v1.11.2 h1:x6gxUeu39V0BHZiugWe8LXZYZ+Utk7hSJGThs8sdzfs=
-github.com/lib/pq v1.11.2/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA=
 github.com/valkey-io/valkey-go v1.0.52 h1:ojrR736satGucqpllYzal8fUrNNROc11V10zokAyIYg=
 github.com/valkey-io/valkey-go v1.0.52/go.mod h1:BXlVAPIL9rFQinSFM+N32JfWzfCaUAqBpZkc4vPY6fM=
-github.com/valkey-io/valkey-go v1.0.71 h1:tuKjGVLd7/I8CyUwqAq5EaD7isxQdlvJzXo3jS8pZW0=
-github.com/valkey-io/valkey-go v1.0.71/go.mod h1:VGhZ6fs68Qrn2+OhH+6waZH27bjpgQOiLyUQyXuYK5k=
 golang.org/x/sys v0.24.0 h1:Twjiwq9dn6R1fQcyiK+wQyHWfaz/BJB+YIpzU/Cv3Xg=
 golang.org/x/sys v0.24.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
-golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=

main.go

@@ -9,7 +9,7 @@ import (
 )
 
 func main() {
-	log.Println("ti1 v1.1.0")
+	log.Println("ti1 v1.0.2")
 	log.Println("Starting...")
 
 	// Setup the database
@@ -34,9 +34,9 @@ func main() {
 
 		log.Println("finished in", time.Since(start))
 		elapsed := time.Since(start)
-		if elapsed < 20*time.Second {
+		if elapsed < 5*time.Minute {
-			log.Printf("starting again in %v", 20*time.Second-elapsed)
+			log.Printf("starting again in %v", 5*time.Minute-elapsed)
-			time.Sleep(20*time.Second - elapsed)
+			time.Sleep(5*time.Minute - elapsed)
 		}
 	}
 }