Merge pull request #17 from Evolutionary-Algorithms-On-Click/get-run-api

Ashrockzzz2003 · web-flow · commit ce559d1109b6 · 2025-03-05T13:06:33.000+05:30
Create API to retrieve run by ID and improve logging
diff --git a/README.md b/README.md
@@ -22,6 +22,7 @@ export DATABASE_URL=<cockroachdb_url>
 export MINIO_ENDPOINT=<minio_endpoint>
 export MINIO_ACCESS_KEY_ID=<minio_access_key>
 export MINIO_SECRET_KEY=<minio_secret_key>
+export RABBITMQ_URL=<rabbitmq_url>
 ```
 3. Run the following command to start the server.
 ```sh
diff --git a/controller/ea.go b/controller/ea.go
@@ -121,8 +121,6 @@ func CreateEA(res http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	// TODO: Schedule the run.
-
 	if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {
 		util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)
 		return
diff --git a/controller/gp.go b/controller/gp.go
@@ -121,7 +121,6 @@ func CreateGP(res http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	// TODO: Schedule the run.
 	if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {
 		util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)
 		return
diff --git a/controller/ml.go b/controller/ml.go
@@ -121,7 +121,6 @@ func CreateML(res http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	// TODO: Schedule the run.
 	if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {
 		util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)
 		return
diff --git a/controller/pso.go b/controller/pso.go
@@ -121,7 +121,6 @@ func CreatePSO(res http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	// TODO: Schedule the run.
 	if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {
 		util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)
 		return
diff --git a/controller/run.go b/controller/run.go
@@ -7,6 +7,42 @@ import (
 	"net/http"
 )
 
+func UserRun(res http.ResponseWriter, req *http.Request) {
+	var logger = util.NewLogger()
+	logger.Info("UserRuns API called.")
+
+	user, err := modules.Auth(req)
+	if err != nil {
+		util.JSONResponse(res, http.StatusUnauthorized, err.Error(), nil)
+		return
+	}
+
+	// User has id, role, userName, email & fullName.
+	logger.Info(fmt.Sprintf("User: %s", user))
+
+	data, err := util.Body(req)
+	if err != nil {
+		util.JSONResponse(res, http.StatusBadRequest, err.Error(), nil)
+		return
+	}
+
+	run, err := modules.RunDataReqFromJSON(data)
+	if err != nil {
+		util.JSONResponse(res, http.StatusBadRequest, err.Error(), nil)
+		return
+	}
+
+	logger.Info(fmt.Sprintf("Run: %s", run.RunID))
+
+	runData, err := run.UserRun(req.Context(), user["id"], logger)
+	if err != nil {
+		util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)
+		return
+	}
+
+	util.JSONResponse(res, http.StatusOK, "User run", runData)
+}
+
 func UserRuns(res http.ResponseWriter, req *http.Request) {
 	var logger = util.NewLogger()
 	logger.Info("UserRuns API called.")
diff --git a/go.mod b/go.mod
@@ -5,6 +5,7 @@ go 1.22.0
 require (
 	github.com/jackc/pgx/v5 v5.7.2
 	github.com/minio/minio-go/v7 v7.0.87
+	github.com/rabbitmq/amqp091-go v1.10.0
 	github.com/rs/cors v1.11.1
 	google.golang.org/grpc v1.70.0
 	google.golang.org/protobuf v1.36.5
@@ -22,7 +23,6 @@ require (
 	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
 	github.com/minio/crc64nvme v1.0.1 // indirect
 	github.com/minio/md5-simd v1.1.2 // indirect
-	github.com/rabbitmq/amqp091-go v1.10.0 // indirect
 	github.com/rs/xid v1.6.0 // indirect
 	golang.org/x/crypto v0.33.0 // indirect
 	golang.org/x/net v0.35.0 // indirect
diff --git a/go.sum b/go.sum
@@ -59,6 +59,8 @@ go.opentelemetry.io/otel/sdk/metric v1.32.0 h1:rZvFnvmvawYb0alrYkjraqJq0Z4ZUJAiy
 go.opentelemetry.io/otel/sdk/metric v1.32.0/go.mod h1:PWeZlq0zt9YkYAp3gjKZ0eicRYvOh1Gd+X99x6GHpCQ=
 go.opentelemetry.io/otel/trace v1.32.0 h1:WIC9mYrXf8TmY/EXuULKc8hR17vE+Hjv2cssQDe03fM=
 go.opentelemetry.io/otel/trace v1.32.0/go.mod h1:+i4rkvCraA+tG6AzwloGaCtkx53Fa+L+V8e9a7YvhT8=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 golang.org/x/crypto v0.33.0 h1:IOBPskki6Lysi0lo9qQvbxiQ+FvsCC/YWOecCHAixus=
 golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5M=
 golang.org/x/net v0.35.0 h1:T5GQRQb2y08kTAByq9L4/bz8cipCdA8FbRTXewonqY8=
diff --git a/main.go b/main.go
@@ -22,6 +22,7 @@ func main() {
 	http.HandleFunc(routes.PSO, controller.CreatePSO)
 	http.HandleFunc(routes.RUNS, controller.UserRuns)
 	http.HandleFunc(routes.SHARE_RUN, controller.ShareRun)
+	http.HandleFunc(routes.RUN, controller.UserRun)
 
 	logger.Info(fmt.Sprintf("Test http server on http://localhost%v/api/test", config.PORT))
 
diff --git a/modules/auth.go b/modules/auth.go
@@ -35,7 +35,7 @@ func Auth(req *http.Request) (map[string]string, error) {
 	defer authConn.Close()
 
 	authClient := pb.NewAuthenticateClient(authConn)
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 
 	r, err := authClient.Auth(ctx, &pb.TokenValidateRequest{Token: token.Value})
diff --git a/modules/run.go b/modules/run.go
@@ -9,10 +9,16 @@ import (
 	"time"
 )
 
-type ShareRunReq struct {
-	RunID         string   `json:"runID"`
-	UserEmailList []string `json:"userEmailList"` // List of emails to share the run with.
-}
+type (
+	ShareRunReq struct {
+		RunID         string   `json:"runID"`
+		UserEmailList []string `json:"userEmailList"` // List of emails to share the run with.
+	}
+
+	RunDataReq struct {
+		RunID string `json:"runID"`
+	}
+)
 
 func UserRuns(ctx context.Context, userID string, logger *util.Logger) ([]map[string]string, error) {
 	db, err := connection.PoolConn(ctx)
@@ -155,3 +161,52 @@ func (s *ShareRunReq) ShareRun(ctx context.Context, logger *util.Logger) error {
 
 	return nil
 }
+
+func RunDataReqFromJSON(jsonData map[string]any) (*RunDataReq, error) {
+	r := &RunDataReq{}
+	jsonDataBytes, err := json.Marshal(jsonData)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := json.Unmarshal(jsonDataBytes, r); err != nil {
+		return nil, err
+	}
+	return r, nil
+}
+
+func (r *RunDataReq) UserRun(ctx context.Context, userID string, logger *util.Logger) (map[string]string, error) {
+	db, err := connection.PoolConn(ctx)
+	if err != nil {
+		logger.Error(fmt.Sprintf("RunData: %s", err.Error()))
+		return nil, fmt.Errorf("something went wrong")
+	}
+
+	// Check if user has access to the run.
+	var runID string
+	if err := db.QueryRow(ctx, "SELECT runID FROM access WHERE userID = $1 AND runID = $2", userID, r.RunID).Scan(&runID); err != nil {
+		logger.Error(fmt.Sprintf("RunData.db.QueryRow: %s", err.Error()))
+		return nil, fmt.Errorf("run does not exist")
+	}
+
+	var id, name, description, status, runType, command, createdBy string
+	var createdAt, updatedAt time.Time
+	// Get the run details like name, description, status, type, command, createdBy, createdAt, updatedAt.
+	err = db.QueryRow(ctx, "SELECT id, name, description, status, type, command, createdBy, createdAt, updatedAt FROM run WHERE id = $1", r.RunID).Scan(&id, &name, &description, &status, &runType, &command, &createdBy, &createdAt, &updatedAt)
+	if err != nil {
+		logger.Error(fmt.Sprintf("RunData.db.QueryRow: %s", err.Error()))
+		return nil, fmt.Errorf("something went wrong")
+	}
+
+	return map[string]string{
+		"id":          id,
+		"name":        name,
+		"description": description,
+		"status":      status,
+		"type":        runType,
+		"command":     command,
+		"createdBy":   createdBy,
+		"createdAt":   createdAt.Local().String(),
+		"updatedAt":   updatedAt.Local().String(),
+	}, nil
+}
diff --git a/routes/api.go b/routes/api.go
@@ -12,4 +12,5 @@ const (
 	PSO       = BASE + "/pso"
 	RUNS      = BASE + "/runs"
 	SHARE_RUN = RUNS + "/share"
+	RUN       = RUNS + "/run"
 )
diff --git a/util/enqueue.go b/util/enqueue.go
@@ -3,22 +3,24 @@ package util
 import (
 	"context"
 	"encoding/json"
-	"log"
+	"fmt"
 	"os"
 	"time"
+
 	amqp "github.com/rabbitmq/amqp091-go"
 )
 
 func EnqueueRunRequest(ctx context.Context, runID string, fileName string, extension string) error {
+	var logger = NewLogger()
 
 	// Message represents the structure of our message
 	type Message struct {
-		RunId	string	`json:"runId"`
-		FileName	string	`json:"fileName"`
-		Extension	string	`json:"extension"`
-		Timestamp	time.Time	`json:"timestamp"`
+		RunId     string    `json:"runId"`
+		FileName  string    `json:"fileName"`
+		Extension string    `json:"extension"`
+		Timestamp time.Time `json:"timestamp"`
 	}
-	
+
 	// Get RabbitMQ connection string from environment variable or use default
 	rabbitMQURL := os.Getenv("RABBITMQ_URL")
 	if rabbitMQURL == "" {
@@ -28,14 +30,14 @@ func EnqueueRunRequest(ctx context.Context, runID string, fileName string, exten
 	// Connect to RabbitMQ server
 	conn, err := amqp.Dial(rabbitMQURL)
 	if err != nil {
-		log.Fatalf("Failed to connect to RabbitMQ: %v", err)
+		logger.Error(fmt.Sprintf("Failed to connect to RabbitMQ: %v", err))
 	}
 	defer conn.Close()
 
 	// Create a channel
 	ch, err := conn.Channel()
 	if err != nil {
-		log.Fatalf("Failed to open a channel: %v", err)
+		logger.Error(fmt.Sprintf("Failed to open a channel: %v", err))
 	}
 	defer ch.Close()
 
@@ -49,24 +51,23 @@ func EnqueueRunRequest(ctx context.Context, runID string, fileName string, exten
 		false,     // no-wait
 		nil,       // arguments
 	)
-	
+
 	if err != nil {
-		log.Fatalf("Failed to declare a queue: %v", err)
+		logger.Error(fmt.Sprintf("Failed to declare a queue: %v", err))
 	}
 
-
 	// Create a new message
 	msg := Message{
-		RunId: runID,
-		FileName: fileName,
+		RunId:     runID,
+		FileName:  fileName,
 		Extension: extension,
 		Timestamp: time.Now(),
 	}
 
 	// Convert message to JSON
 	body, err := json.Marshal(msg)
 	if err != nil {
-		log.Printf("Error marshaling message: %v", err)
+		logger.Error(fmt.Sprintf("Error marshaling message: %v", err))
 	}
 
 	// Publish message
@@ -77,17 +78,16 @@ func EnqueueRunRequest(ctx context.Context, runID string, fileName string, exten
 		false,  // mandatory
 		false,  // immediate
 		amqp.Publishing{
-			DeliveryMode: amqp.Persistent, 
-			Body: body,
+			DeliveryMode: amqp.Persistent,
+			Body:         body,
 		},
 	)
-			
+
 	if err != nil {
-		log.Printf("Failed to publish a message: %v", err)
+		logger.Error(fmt.Sprintf("Failed to publish message: %v", err))
+		return err
 	}
-	
-	log.Printf("Published message: %s", msg.RunId)
-		
 
+	logger.Info(fmt.Sprintf("Published message: %s", msg.RunId))
 	return nil
-}
+}
diff --git a/util/minio.go b/util/minio.go
@@ -3,9 +3,9 @@ package util
 import (
 	"context"
 	"fmt"
-	"os"
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
+	"os"
 )
 
 func UploadFile(ctx context.Context, runID string, fileName string, extension string) error {

Original file line number	Diff line number	Diff line change
`@@ -121,8 +121,6 @@ func CreateEA(res http.ResponseWriter, req *http.Request) {`
`121`	`121`	`return`
`122`	`122`	`}`
`123`	`123`
`124`		`- // TODO: Schedule the run.`
`125`		`-`
`126`	`124`	`if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {`
`127`	`125`	`util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)`
`128`	`126`	`return`
Original file line number	Diff line number	Diff line change
`@@ -121,7 +121,6 @@ func CreateGP(res http.ResponseWriter, req *http.Request) {`
`121`	`121`	`return`
`122`	`122`	`}`
`123`	`123`
`124`		`- // TODO: Schedule the run.`
`125`	`124`	`if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {`
`126`	`125`	`util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)`
`127`	`126`	`return`
Original file line number	Diff line number	Diff line change
`@@ -121,7 +121,6 @@ func CreateML(res http.ResponseWriter, req *http.Request) {`
`121`	`121`	`return`
`122`	`122`	`}`
`123`	`123`
`124`		`- // TODO: Schedule the run.`
`125`	`124`	`if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {`
`126`	`125`	`util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)`
`127`	`126`	`return`
Original file line number	Diff line number	Diff line change
`@@ -121,7 +121,6 @@ func CreatePSO(res http.ResponseWriter, req *http.Request) {`
`121`	`121`	`return`
`122`	`122`	`}`
`123`	`123`
`124`		`- // TODO: Schedule the run.`
`125`	`124`	`if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {`
`126`	`125`	`util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)`
`127`	`126`	`return`