Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
caeb7651d3 | ||
|
|
21a133fc57 | ||
|
|
a4c4159688 | ||
|
|
f2b984b6f9 | ||
|
|
348b3cd5b7 | ||
|
|
4c94654e0a | ||
|
|
5b840950c8 |
2
.github/workflows/docker-image.yml
vendored
2
.github/workflows/docker-image.yml
vendored
@@ -30,7 +30,7 @@ jobs:
|
||||
- name: Get commit version
|
||||
id: commit-version
|
||||
run: |
|
||||
COMMIT_MSG=$(git log -1 --pretty=%B)
|
||||
COMMIT_MSG=$(git log -1 --pretty=%B | head -n1 | xargs)
|
||||
echo "Commit message: $COMMIT_MSG" # Debugging output
|
||||
# Updated regex to handle both vX.Y, vX.Y.Z, and vX.Y-pre-release formats
|
||||
if [[ "$COMMIT_MSG" =~ ^v[0-9]+\.[0-9]+(\.[0-9]+)?(-[a-zA-Z0-9._-]+)?$ ]]; then
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
# Use the official Golang image as the base image
|
||||
FROM golang:1.23.4
|
||||
FROM golang:1.26.0
|
||||
|
||||
# Set the Current Working Directory inside the container
|
||||
WORKDIR /app
|
||||
|
||||
@@ -1,89 +0,0 @@
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -1,41 +0,0 @@
|
||||
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
|
||||
}
|
||||
@@ -1,89 +0,0 @@
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -1,30 +0,0 @@
|
||||
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
|
||||
}
|
||||
@@ -1,140 +0,0 @@
|
||||
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
|
||||
}
|
||||
10
go.mod
10
go.mod
@@ -1,10 +1,10 @@
|
||||
module ti1
|
||||
|
||||
go 1.23.4
|
||||
|
||||
require github.com/lib/pq v1.10.9
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
github.com/valkey-io/valkey-go v1.0.52 // indirect
|
||||
golang.org/x/sys v0.24.0 // indirect
|
||||
github.com/lib/pq v1.11.2
|
||||
github.com/valkey-io/valkey-go v1.0.71
|
||||
)
|
||||
|
||||
require golang.org/x/sys v0.41.0 // indirect
|
||||
|
||||
6
go.sum
6
go.sum
@@ -1,6 +1,12 @@
|
||||
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=
|
||||
|
||||
8
main.go
8
main.go
@@ -9,7 +9,7 @@ import (
|
||||
)
|
||||
|
||||
func main() {
|
||||
log.Println("ti1 v1.0.2")
|
||||
log.Println("ti1 v1.1.0")
|
||||
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 < 5*time.Minute {
|
||||
log.Printf("starting again in %v", 5*time.Minute-elapsed)
|
||||
time.Sleep(5*time.Minute - elapsed)
|
||||
if elapsed < 20*time.Second {
|
||||
log.Printf("starting again in %v", 20*time.Second-elapsed)
|
||||
time.Sleep(20*time.Second - elapsed)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user