eng-2888-investigate-resource-edit-bug

.gitignore

@@ -7,6 +7,9 @@ src/python/build/
 *.idea/
 *.vscode/
 
+# Intellij IDE
+src/ui/frontend.iml
+
 # Binaries for programs and plugins
 *.exe
 *.exe~

sdk/aqueduct/constants/enums.py

@@ -74,7 +74,7 @@ class ServiceType(str, Enum, metaclass=MetaEnum):
     LAMBDA = "Lambda"
     MONGO_DB = "MongoDB"
     CONDA = "Conda"
-    AQUEDUCT_ENGINE = "Aqueduct Engine"
+    AQUEDUCT_ENGINE = "Aqueduct"
     DATABRICKS = "Databricks"
     EMAIL = "Email"
     SLACK = "Slack"

src/golang/cmd/server/handler/connect_integration.go

@@ -302,7 +302,7 @@ func ConnectIntegration(
 	}
 
 	// The initial integration entry has been written. Any errors from this point on will need to update
-	// the that entry to reflect the failure. Note that this defer is only relevant for q
+	// the that entry to reflect the failure.
 	defer func() {
 		if err != nil {
 			execution_state.UpdateOnFailure(

src/golang/cmd/server/handler/list_operators_for_integration.go

@@ -2,15 +2,12 @@ package handler
 
 import (
 	"context"
-	"fmt"
 	"net/http"
 
 	"github.com/aqueducthq/aqueduct/cmd/server/routes"
 	aq_context "github.com/aqueducthq/aqueduct/lib/context"
 	"github.com/aqueducthq/aqueduct/lib/database"
-	"github.com/aqueducthq/aqueduct/lib/dynamic"
 	"github.com/aqueducthq/aqueduct/lib/models"
-	"github.com/aqueducthq/aqueduct/lib/models/shared"
 	"github.com/aqueducthq/aqueduct/lib/repos"
 	"github.com/aqueducthq/aqueduct/lib/workflow/operator"
 	"github.com/dropbox/godropbox/errors"
@@ -92,28 +89,10 @@ func (h *ListOperatorsForIntegrationHandler) Prepare(r *http.Request) (interface
 func (h *ListOperatorsForIntegrationHandler) Perform(ctx context.Context, interfaceArgs interface{}) (interface{}, int, error) {
 	args := interfaceArgs.(*listOperatorsForIntegrationArgs)
 
-	integrationID := args.integrationObject.ID
-
-	if args.integrationObject.Service == shared.AWS {
-		// If the requested integration is a cloud integration, substitute the cloud integration ID
-		// with the ID of the dynamic k8s integration.
-		k8sIntegration, err := h.IntegrationRepo.GetByNameAndUser(
-			ctx,
-			fmt.Sprintf("%s:%s", args.integrationObject.Name, dynamic.K8sIntegrationNameSuffix),
-			uuid.Nil,
-			args.OrgID,
-			h.Database,
-		)
-		if err != nil {
-			return nil, http.StatusInternalServerError, errors.Wrap(err, "Failed to retrieve the Aqueduct-generated k8s integration.")
-		}
-
-		integrationID = k8sIntegration.ID
-	}
-
 	operators, err := operator.GetOperatorsOnIntegration(
 		ctx,
-		integrationID,
+		args.OrgID,
+		args.integrationObject,
 		h.IntegrationRepo,
 		h.OperatorRepo,
 		h.Database,

src/golang/cmd/server/handler/test_integration.go

@@ -10,6 +10,7 @@ import (
 	"github.com/aqueducthq/aqueduct/lib/database"
 	"github.com/aqueducthq/aqueduct/lib/errors"
 	"github.com/aqueducthq/aqueduct/lib/job"
+	"github.com/aqueducthq/aqueduct/lib/models/shared"
 	"github.com/aqueducthq/aqueduct/lib/repos"
 	"github.com/aqueducthq/aqueduct/lib/vault"
 	"github.com/aqueducthq/aqueduct/lib/workflow/operator/connector/auth"
@@ -89,11 +90,16 @@ func (h *TestIntegrationHandler) Perform(ctx context.Context, interfaceArgs inte
 	if errors.Is(err, database.ErrNoRows()) {
 		return emptyResp, http.StatusBadRequest, err
 	}
-
 	if err != nil {
 		return emptyResp, http.StatusInternalServerError, errors.Wrap(err, "Failed to retrieve integration")
 	}
 
+	// No need to do any further verification for Aqueduct Compute.
+	// The fact that it even got here means it works.
+	if integrationObject.Name == shared.AqueductComputeIntegrationName {
+		return emptyResp, http.StatusOK, nil
+	}
+
 	storageConfig := config.Storage()
 	vaultObject, err := vault.NewVault(&storageConfig, config.EncryptionKey())
 	if err != nil {

src/golang/cmd/server/handler/v2/integration_operators_get.go

@@ -0,0 +1,101 @@
+package v2
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/aqueducthq/aqueduct/cmd/server/handler"
+	"github.com/aqueducthq/aqueduct/cmd/server/request/parser"
+	aq_context "github.com/aqueducthq/aqueduct/lib/context"
+	"github.com/aqueducthq/aqueduct/lib/database"
+	"github.com/aqueducthq/aqueduct/lib/functional/slices"
+	"github.com/aqueducthq/aqueduct/lib/models"
+	"github.com/aqueducthq/aqueduct/lib/models/views"
+	"github.com/aqueducthq/aqueduct/lib/repos"
+	"github.com/aqueducthq/aqueduct/lib/response"
+	"github.com/aqueducthq/aqueduct/lib/workflow/operator"
+	"github.com/dropbox/godropbox/errors"
+	"github.com/google/uuid"
+)
+
+// This file should map directly to
+// src/ui/common/src/handlers/v2/IntegrationOperatorsGet.ts
+//
+// Route: /api/v2/integration/{integrationID}/nodes/operators
+// Method: GET
+// Params:
+// 	`integrationID`: ID for `integration` object
+// Request:
+//	Headers:
+//		`api-key`: user's API Key
+// Response:
+//	Body:
+//    A list of `response.operators` that use the given integration.
+
+type integrationOperatorsGetArgs struct {
+	*aq_context.AqContext
+	integrationID uuid.UUID
+}
+
+type IntegrationOperatorsGetHandler struct {
+	handler.GetHandler
+
+	Database        database.Database
+	IntegrationRepo repos.Integration
+	OperatorRepo    repos.Operator
+}
+
+func (*IntegrationOperatorsGetHandler) Name() string {
+	return "IntegrationOperatorsGet"
+}
+
+func (h *IntegrationOperatorsGetHandler) Prepare(r *http.Request) (interface{}, int, error) {
+	aqContext, statusCode, err := aq_context.ParseAqContext(r.Context())
+	if err != nil {
+		return nil, statusCode, err
+	}
+
+	integrationID, err := (parser.IntegrationIDParser{}).Parse(r)
+	if err != nil {
+		return nil, http.StatusBadRequest, err
+	}
+
+	return &integrationOperatorsGetArgs{
+		AqContext: aqContext,
+
+		integrationID: *integrationID,
+	}, http.StatusOK, nil
+}
+
+func (h *IntegrationOperatorsGetHandler) Perform(ctx context.Context, interfaceArgs interface{}) (interface{}, int, error) {
+	args := interfaceArgs.(*integrationOperatorsGetArgs)
+
+	integration, err := h.IntegrationRepo.Get(ctx, args.integrationID, h.Database)
+	if err != nil {
+		return nil, http.StatusInternalServerError, errors.Wrapf(err, "Unable to find integration %s", args.integrationID)
+	}
+
+	operators, err := operator.GetOperatorsOnIntegration(
+		ctx,
+		args.OrgID,
+		integration,
+		h.IntegrationRepo,
+		h.OperatorRepo,
+		h.Database,
+	)
+	if err != nil {
+		return nil, http.StatusInternalServerError, errors.Wrap(err, "Unable to retrieve operators.")
+	}
+
+	operatorIDs := slices.Map(operators, func(op models.Operator) uuid.UUID {
+		return op.ID
+	})
+	operatorNodes, err := h.OperatorRepo.GetNodeBatch(ctx, operatorIDs, h.Database)
+	if err != nil {
+		return nil, http.StatusInternalServerError, errors.Wrap(err, "Unable to retrieve operator nodes.")
+	}
+
+	return slices.Map(operatorNodes, func(node views.OperatorNode) *response.Operator {
+		return response.NewOperatorFromDBObject(&node)
+	}), http.StatusOK, nil
+}

src/golang/cmd/server/handler/v2/integration_workflows_get.go

@@ -0,0 +1,78 @@
+package v2
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/aqueducthq/aqueduct/cmd/server/handler"
+	"github.com/aqueducthq/aqueduct/cmd/server/request/parser"
+	aq_context "github.com/aqueducthq/aqueduct/lib/context"
+	"github.com/aqueducthq/aqueduct/lib/database"
+	"github.com/aqueducthq/aqueduct/lib/errors"
+	"github.com/aqueducthq/aqueduct/lib/repos"
+	"github.com/google/uuid"
+)
+
+// This file should map directly to
+// src/ui/common/src/handlers/v2/IntegrationWorkflowsGet.ts
+//
+// Route: /v2/integration/{integrationID}/workflows
+// Method: GET
+// Request:
+//	Headers:
+//		`api-key`: user's API Key
+// Response:
+//	Body:
+//   	A list of workflow IDs that use the given integration.
+
+type integrationWorkflowsGetArgs struct {
+	*aq_context.AqContext
+	integrationID uuid.UUID
+}
+
+type IntegrationWorkflowsGetHandler struct {
+	handler.GetHandler
+
+	Database        database.Database
+	IntegrationRepo repos.Integration
+	OperatorRepo    repos.Operator
+}
+
+func (*IntegrationWorkflowsGetHandler) Name() string {
+	return "IntegrationWorkflowsGet"
+}
+
+func (h *IntegrationWorkflowsGetHandler) Prepare(r *http.Request) (interface{}, int, error) {
+	aqContext, statusCode, err := aq_context.ParseAqContext(r.Context())
+	if err != nil {
+		return nil, statusCode, err
+	}
+
+	integrationID, err := (parser.IntegrationIDParser{}).Parse(r)
+	if err != nil {
+		return nil, http.StatusBadRequest, err
+	}
+
+	return &integrationWorkflowsGetArgs{
+		AqContext:     aqContext,
+		integrationID: *integrationID,
+	}, http.StatusOK, nil
+}
+
+func (h *IntegrationWorkflowsGetHandler) Perform(ctx context.Context, interfaceArgs interface{}) (interface{}, int, error) {
+	args := interfaceArgs.(*integrationWorkflowsGetArgs)
+
+	integration, err := h.IntegrationRepo.Get(ctx, args.integrationID, h.Database)
+	if err != nil {
+		return nil, http.StatusInternalServerError, errors.Wrapf(err, "Unable to find integration %s", args.integrationID)
+	}
+
+	workflowIDs, err := fetchWorkflowIDsForIntegration(
+		ctx, args.OrgID, integration, h.IntegrationRepo, h.OperatorRepo, h.Database,
+	)
+	if err != nil {
+		return nil, http.StatusInternalServerError, errors.Wrapf(err, "Unable to find workflows for integration %s", args.integrationID)
+	}
+
+	return workflowIDs, http.StatusOK, nil
+}