diff --git a/api/restHandler/GlobalPluginRestHandler.go b/api/restHandler/GlobalPluginRestHandler.go
index 7fc25ec6b16..44a2305c0ab 100644
--- a/api/restHandler/GlobalPluginRestHandler.go
+++ b/api/restHandler/GlobalPluginRestHandler.go
@@ -35,6 +35,7 @@ import (
 
 type GlobalPluginRestHandler interface {
 	PatchPlugin(w http.ResponseWriter, r *http.Request)
+	CreatePlugin(w http.ResponseWriter, r *http.Request)
 
 	GetAllGlobalVariables(w http.ResponseWriter, r *http.Request)
 	ListAllPlugins(w http.ResponseWriter, r *http.Request)
@@ -46,6 +47,7 @@ type GlobalPluginRestHandler interface {
 	GetPluginDetailByIds(w http.ResponseWriter, r *http.Request)
 	GetAllUniqueTags(w http.ResponseWriter, r *http.Request)
 	MigratePluginData(w http.ResponseWriter, r *http.Request)
+	GetAllPluginMinData(w http.ResponseWriter, r *http.Request)
 }
 
 func NewGlobalPluginRestHandler(logger *zap.SugaredLogger, globalPluginService plugin.GlobalPluginService,
@@ -420,3 +422,68 @@ func (handler *GlobalPluginRestHandlerImpl) MigratePluginData(w http.ResponseWri
 	}
 	common.WriteJsonResp(w, nil, nil, http.StatusOK)
 }
+
+func (handler *GlobalPluginRestHandlerImpl) CreatePlugin(w http.ResponseWriter, r *http.Request) {
+	userId, err := handler.userService.GetLoggedInUser(r)
+	if userId == 0 || err != nil {
+		common.WriteJsonResp(w, err, "Unauthorized User", http.StatusUnauthorized)
+		return
+	}
+	token := r.Header.Get("token")
+	appId, err := common.ExtractIntQueryParam(w, r, "appId", 0)
+	if err != nil {
+		return
+	}
+	ok, err := handler.IsUserAuthorized(token, appId)
+	if err != nil {
+		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
+		return
+	}
+	if !ok {
+		common.WriteJsonResp(w, fmt.Errorf("unauthorized user"), "Unauthorized User", http.StatusForbidden)
+		return
+	}
+	decoder := json.NewDecoder(r.Body)
+	var pluginDataDto bean.PluginParentMetadataDto
+	err = decoder.Decode(&pluginDataDto)
+	if err != nil {
+		handler.logger.Errorw("request err, CreatePlugin", "error", err, "payload", pluginDataDto)
+		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
+		return
+	}
+	handler.logger.Infow("request payload received for creating plugins", pluginDataDto, "userId", userId)
+
+	pluginVersionId, err := handler.globalPluginService.CreatePluginOrVersions(&pluginDataDto, userId)
+	if err != nil {
+		handler.logger.Errorw("service error, error in creating plugin", "pluginCreateRequestDto", pluginDataDto, "err", err)
+		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
+		return
+	}
+
+	common.WriteJsonResp(w, nil, bean.NewPluginMinDto().WithPluginVersionId(pluginVersionId), http.StatusOK)
+}
+
+func (handler *GlobalPluginRestHandlerImpl) GetAllPluginMinData(w http.ResponseWriter, r *http.Request) {
+	token := r.Header.Get("token")
+	appId, err := common.ExtractIntQueryParam(w, r, "appId", 0)
+	if err != nil {
+		return
+	}
+	ok, err := handler.IsUserAuthorized(token, appId)
+	if err != nil {
+		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
+		return
+	}
+	if !ok {
+		common.WriteJsonResp(w, fmt.Errorf("unauthorized user"), "Unauthorized User", http.StatusForbidden)
+		return
+	}
+
+	pluginDetail, err := handler.globalPluginService.GetAllPluginMinData()
+	if err != nil {
+		handler.logger.Errorw("error in getting all unique tags", "err", err)
+		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
+		return
+	}
+	common.WriteJsonResp(w, nil, pluginDetail, http.StatusOK)
+}
diff --git a/api/router/GlobalPluginRouter.go b/api/router/GlobalPluginRouter.go
index 06950c33421..0d8154bf822 100644
--- a/api/router/GlobalPluginRouter.go
+++ b/api/router/GlobalPluginRouter.go
@@ -41,7 +41,8 @@ type GlobalPluginRouterImpl struct {
 func (impl *GlobalPluginRouterImpl) initGlobalPluginRouter(globalPluginRouter *mux.Router) {
 	globalPluginRouter.Path("/migrate").
 		HandlerFunc(impl.globalPluginRestHandler.MigratePluginData).Methods("PUT")
-
+	globalPluginRouter.Path("/create").
+		HandlerFunc(impl.globalPluginRestHandler.CreatePlugin).Methods("POST")
 	// versioning impact handling to be done for below apis,
 	globalPluginRouter.Path("").
 		HandlerFunc(impl.globalPluginRestHandler.PatchPlugin).Methods("POST")
@@ -68,5 +69,7 @@ func (impl *GlobalPluginRouterImpl) initGlobalPluginRouter(globalPluginRouter *m
 
 	globalPluginRouter.Path("/list/tags").
 		HandlerFunc(impl.globalPluginRestHandler.GetAllUniqueTags).Methods("GET")
+	globalPluginRouter.Path("/list/v2/min").
+		HandlerFunc(impl.globalPluginRestHandler.GetAllPluginMinData).Methods("GET")
 
 }
diff --git a/internal/util/ErrorUtil.go b/internal/util/ErrorUtil.go
index 59a8d234415..43f3c9b942e 100644
--- a/internal/util/ErrorUtil.go
+++ b/internal/util/ErrorUtil.go
@@ -26,6 +26,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"net/http"
+	"strconv"
 )
 
 type ApiError struct {
@@ -36,6 +37,14 @@ type ApiError struct {
 	UserDetailMessage string      `json:"userDetailMessage,omitempty"`
 }
 
+func GetApiError(code int, userMessage, internalMessage string) *ApiError {
+	return &ApiError{
+		HttpStatusCode:  code,
+		Code:            strconv.Itoa(code),
+		InternalMessage: internalMessage,
+		UserMessage:     userMessage,
+	}
+}
 func NewApiError() *ApiError {
 	return &ApiError{}
 }
diff --git a/pkg/plugin/GlobalPluginService.go b/pkg/plugin/GlobalPluginService.go
index 55ff942789b..5423ab22b2d 100644
--- a/pkg/plugin/GlobalPluginService.go
+++ b/pkg/plugin/GlobalPluginService.go
@@ -24,6 +24,7 @@ import (
 	"github.com/devtron-labs/devtron/pkg/auth/user"
 	"github.com/devtron-labs/devtron/pkg/auth/user/bean"
 	repository2 "github.com/devtron-labs/devtron/pkg/pipeline/repository"
+	"github.com/devtron-labs/devtron/pkg/plugin/adaptor"
 	bean2 "github.com/devtron-labs/devtron/pkg/plugin/bean"
 	helper2 "github.com/devtron-labs/devtron/pkg/plugin/helper"
 	"github.com/devtron-labs/devtron/pkg/plugin/repository"
@@ -31,8 +32,8 @@ import (
 	"github.com/devtron-labs/devtron/pkg/sql"
 	"github.com/go-pg/pg"
 	"go.uber.org/zap"
+	"golang.org/x/mod/semver"
 	"net/http"
-	"strconv"
 	"strings"
 	"time"
 )
@@ -73,9 +74,11 @@ type GlobalPluginService interface {
 	GetDetailedPluginInfoByPluginId(pluginId int) (*bean2.PluginMetadataDto, error)
 	GetAllDetailedPluginInfo() ([]*bean2.PluginMetadataDto, error)
 
+	CreatePluginOrVersions(pluginDto *bean2.PluginParentMetadataDto, userId int32) (int, error)
 	ListAllPluginsV2(filter *bean2.PluginsListFilter) (*bean2.PluginsDto, error)
 	GetPluginDetailV2(pluginVersionIds, parentPluginIds []int, fetchAllVersionDetails bool) (*bean2.PluginsDto, error)
 	GetAllUniqueTags() (*bean2.PluginTagsDto, error)
+	GetAllPluginMinData() ([]*bean2.PluginMinDto, error)
 	MigratePluginData() error
 }
 
@@ -429,7 +432,7 @@ func (impl *GlobalPluginServiceImpl) validatePluginRequest(pluginReq *bean2.Plug
 		return errors.New("invalid plugin type, should be of the type PRESET or SHARED")
 	}
 
-	plugins, err := impl.globalPluginRepository.GetMetaDataForAllPlugins()
+	plugins, err := impl.globalPluginRepository.GetAllPluginMinData()
 	if err != nil {
 		impl.logger.Errorw("error in getting all plugins", "err", err)
 		return err
@@ -670,33 +673,10 @@ func (impl *GlobalPluginServiceImpl) UpdatePluginPipelineScript(dbPluginPipeline
 
 func (impl *GlobalPluginServiceImpl) saveDeepPluginStepData(pluginMetadataId int, pluginStepsReq []*bean2.PluginStepsDto, userId int32, tx *pg.Tx) error {
 	for _, pluginStep := range pluginStepsReq {
-		pluginStepData := &repository.PluginStep{
-			PluginId:            pluginMetadataId,
-			Name:                pluginStep.Name,
-			Description:         pluginStep.Description,
-			Index:               pluginStep.Index,
-			StepType:            pluginStep.StepType,
-			RefPluginId:         pluginStep.RefPluginId,
-			OutputDirectoryPath: pluginStep.OutputDirectoryPath,
-			DependentOnStep:     pluginStep.DependentOnStep,
-			AuditLog:            sql.NewDefaultAuditLog(userId),
-		}
+		pluginStepData := adaptor.GetPluginStepDbObject(pluginStep, pluginMetadataId, userId)
 		//get the script saved for this plugin step
 		if pluginStep.PluginPipelineScript != nil {
-			pluginPipelineScript := &repository.PluginPipelineScript{
-				Script:                   pluginStep.PluginPipelineScript.Script,
-				StoreScriptAt:            pluginStep.PluginPipelineScript.StoreScriptAt,
-				Type:                     pluginStep.PluginPipelineScript.Type,
-				DockerfileExists:         pluginStep.PluginPipelineScript.DockerfileExists,
-				MountPath:                pluginStep.PluginPipelineScript.MountPath,
-				MountCodeToContainer:     pluginStep.PluginPipelineScript.MountCodeToContainer,
-				MountCodeToContainerPath: pluginStep.PluginPipelineScript.MountCodeToContainerPath,
-				MountDirectoryFromHost:   pluginStep.PluginPipelineScript.MountDirectoryFromHost,
-				ContainerImagePath:       pluginStep.PluginPipelineScript.ContainerImagePath,
-				ImagePullSecretType:      pluginStep.PluginPipelineScript.ImagePullSecretType,
-				ImagePullSecret:          pluginStep.PluginPipelineScript.ImagePullSecret,
-				AuditLog:                 sql.NewDefaultAuditLog(userId),
-			}
+			pluginPipelineScript := adaptor.GetPluginPipelineScriptDbObject(pluginStep.PluginPipelineScript, userId)
 			pluginPipelineScript, err := impl.globalPluginRepository.SavePluginPipelineScript(pluginPipelineScript, tx)
 			if err != nil {
 				impl.logger.Errorw("error in saving plugin pipeline script", "pluginPipelineScript", pluginPipelineScript, "err", err)
@@ -719,23 +699,7 @@ func (impl *GlobalPluginServiceImpl) saveDeepPluginStepData(pluginMetadataId int
 		pluginStep.Id = pluginStepData.Id
 		//create entry in plugin_step_variable
 		for _, pluginStepVariable := range pluginStep.PluginStepVariable {
-			pluginStepVariableData := &repository.PluginStepVariable{
-				PluginStepId:              pluginStepData.Id,
-				Name:                      pluginStepVariable.Name,
-				Format:                    pluginStepVariable.Format,
-				Description:               pluginStepVariable.Description,
-				IsExposed:                 pluginStepVariable.IsExposed,
-				AllowEmptyValue:           pluginStepVariable.AllowEmptyValue,
-				DefaultValue:              pluginStepVariable.DefaultValue,
-				Value:                     pluginStepVariable.Value,
-				VariableType:              pluginStepVariable.VariableType,
-				ValueType:                 pluginStepVariable.ValueType,
-				PreviousStepIndex:         pluginStepVariable.PreviousStepIndex,
-				VariableStepIndex:         pluginStepVariable.VariableStepIndex,
-				VariableStepIndexInPlugin: pluginStepVariable.VariableStepIndexInPlugin,
-				ReferenceVariableName:     pluginStepVariable.ReferenceVariableName,
-				AuditLog:                  sql.NewDefaultAuditLog(userId),
-			}
+			pluginStepVariableData := adaptor.GetPluginStepVariableDbObject(pluginStepData.Id, pluginStepVariable, userId)
 			pluginStepVariableData, err = impl.globalPluginRepository.SavePluginStepVariables(pluginStepVariableData, tx)
 			if err != nil {
 				impl.logger.Errorw("error in saving plugin step variable", "pluginStepVariableData", pluginStepVariableData, "err", err)
@@ -744,14 +708,7 @@ func (impl *GlobalPluginServiceImpl) saveDeepPluginStepData(pluginMetadataId int
 			pluginStepVariable.Id = pluginStepVariableData.Id
 			//create entry in plugin_step_condition
 			for _, pluginStepCondition := range pluginStepVariable.PluginStepCondition {
-				pluginStepConditionData := &repository.PluginStepCondition{
-					PluginStepId:        pluginStepData.Id,
-					ConditionVariableId: pluginStepVariableData.Id,
-					ConditionType:       pluginStepCondition.ConditionType,
-					ConditionalOperator: pluginStepCondition.ConditionalOperator,
-					ConditionalValue:    pluginStepCondition.ConditionalValue,
-					AuditLog:            sql.NewDefaultAuditLog(userId),
-				}
+				pluginStepConditionData := adaptor.GetPluginStepConditionDbObject(pluginStepData.Id, pluginStepVariableData.Id, pluginStepCondition, userId)
 				pluginStepConditionData, err = impl.globalPluginRepository.SavePluginStepConditions(pluginStepConditionData, tx)
 				if err != nil {
 					impl.logger.Errorw("error in saving plugin step condition", "pluginStepConditionData", pluginStepConditionData, "err", err)
@@ -768,7 +725,6 @@ func (impl *GlobalPluginServiceImpl) updatePlugin(pluginUpdateReq *bean2.PluginM
 	if len(pluginUpdateReq.Type) == 0 {
 		return nil, errors.New("invalid plugin type, should be of the type PRESET or SHARED")
 	}
-
 	dbConnection := impl.globalPluginRepository.GetConnection()
 	tx, err := dbConnection.Begin()
 	if err != nil {
@@ -856,6 +812,7 @@ func (impl *GlobalPluginServiceImpl) updatePlugin(pluginUpdateReq *bean2.PluginM
 			return nil, err
 		}
 	}
+
 	if len(pluginStepsToUpdate) > 0 {
 		err = impl.updateDeepPluginStepData(pluginStepsToUpdate, pluginStepVariables, pluginStepConditions, pluginSteps, userId, tx)
 		if err != nil {
@@ -1386,7 +1343,6 @@ func filterPluginStepData(existingPluginStepsInDb []*repository.PluginStep, plug
 	} else {
 		return nil, nil, pluginStepUpdateReq
 	}
-
 	return newPluginStepsToCreate, pluginStepsToRemove, pluginStepsToUpdate
 }
 
@@ -1805,28 +1761,59 @@ func (impl *GlobalPluginServiceImpl) ListAllPluginsV2(filter *bean2.PluginsListF
 
 	return pluginDetails, nil
 }
+func (impl *GlobalPluginServiceImpl) validateDetailRequest(pluginVersions []*repository.PluginMetadata, pluginVersionIds, parentPluginIds []int) error {
+	pluginVersionsIdMap, pluginParentIdMap := make(map[int]bool, len(pluginVersionIds)), make(map[int]bool, len(parentPluginIds))
+	allPlugins, err := impl.globalPluginRepository.GetAllPluginMinData()
+	if err != nil {
+		impl.logger.Errorw("validateDetailRequest, error in getting all plugins parent metadata", "err", err)
+		return err
+	}
+	for _, pluginVersion := range pluginVersions {
+		pluginVersionsIdMap[pluginVersion.Id] = true
+	}
+	for _, plugin := range allPlugins {
+		pluginParentIdMap[plugin.Id] = true
+	}
+	for _, versionId := range pluginVersionIds {
+		if _, ok := pluginVersionsIdMap[versionId]; !ok {
+			errorMsg := fmt.Sprintf("there are some plugin version ids in request that do not exist:- %d", versionId)
+			return util.GetApiError(http.StatusBadRequest, errorMsg, errorMsg)
+		}
+	}
+	for _, pluginId := range parentPluginIds {
+		if _, ok := pluginParentIdMap[pluginId]; !ok {
+			errorMsg := fmt.Sprintf("there are some plugin parent ids in request that do not exist %d", pluginId)
+			return util.GetApiError(http.StatusBadRequest, errorMsg, errorMsg)
+		}
+	}
+	return nil
+}
 
 // GetPluginDetailV2 returns all details of the of a plugin version according to the pluginVersionIds and parentPluginIds
 // provided by user, and minimal data for all versions of that plugin.
 func (impl *GlobalPluginServiceImpl) GetPluginDetailV2(pluginVersionIds, parentPluginIds []int, fetchAllVersionDetails bool) (*bean2.PluginsDto, error) {
+	var err error
+	pluginVersionsMetadata, err := impl.globalPluginRepository.GetMetaDataForAllPlugins()
+	if err != nil {
+		impl.logger.Errorw("GetPluginDetailV2, error in getting all plugins versions metadata", "err", err)
+		return nil, err
+	}
+	err = impl.validateDetailRequest(pluginVersionsMetadata, pluginVersionIds, parentPluginIds)
+	if err != nil {
+		return nil, err
+	}
 	pluginParentMetadataDtos := make([]*bean2.PluginParentMetadataDto, 0, len(pluginVersionIds)+len(parentPluginIds))
 	if len(pluginVersionIds) == 0 && len(parentPluginIds) == 0 {
-		return nil, &util.ApiError{HttpStatusCode: http.StatusBadRequest, Code: strconv.Itoa(http.StatusBadRequest), InternalMessage: bean2.NoPluginOrParentIdProvidedErr, UserMessage: bean2.NoPluginOrParentIdProvidedErr}
+		return nil, util.GetApiError(http.StatusBadRequest, bean2.NoPluginOrParentIdProvidedErr, bean2.NoPluginOrParentIdProvidedErr)
 	}
 	pluginVersionIdsMap, parentPluginIdsMap := helper2.GetPluginVersionAndParentPluginIdsMap(pluginVersionIds, parentPluginIds)
 
-	var err error
 	pluginParentMetadataIds := make([]int, 0, len(pluginVersionIds)+len(parentPluginIds))
 	pluginVersionsIdToInclude := make(map[int]bool, len(pluginVersionIds)+len(parentPluginIds))
-	pluginVersionsMetadata, err := impl.globalPluginRepository.GetMetaDataForAllPlugins()
-	if err != nil {
-		impl.logger.Errorw("GetPluginDetailV2, error in getting all plugins versions metadata", "err", err)
-		return nil, err
-	}
 
 	filteredPluginVersionMetadata := helper2.GetPluginVersionsMetadataByVersionAndParentPluginIds(pluginVersionsMetadata, pluginVersionIdsMap, parentPluginIdsMap)
 	if len(filteredPluginVersionMetadata) == 0 {
-		return nil, &util.ApiError{HttpStatusCode: http.StatusNotFound, Code: strconv.Itoa(http.StatusNotFound), InternalMessage: bean2.NoPluginFoundForThisSearchQueryErr, UserMessage: bean2.NoPluginFoundForThisSearchQueryErr}
+		return nil, util.GetApiError(http.StatusNotFound, bean2.NoPluginFoundForThisSearchQueryErr, bean2.NoPluginFoundForThisSearchQueryErr)
 	}
 	for _, version := range filteredPluginVersionMetadata {
 		_, found := pluginVersionIdsMap[version.Id]
@@ -1884,7 +1871,6 @@ func (impl *GlobalPluginServiceImpl) MigratePluginData() error {
 
 // MigratePluginDataToParentPluginMetadata migrates pre-existing plugin metadata from plugin_metadata table into plugin_parent_metadata table,
 // and also populate plugin_parent_metadata_id in plugin_metadata.
-// this operation will happen only once when the get all plugin list v2 api is being called, returns error if any
 func (impl *GlobalPluginServiceImpl) MigratePluginDataToParentPluginMetadata(pluginsMetadata []*repository.PluginMetadata) error {
 	dbConnection := impl.globalPluginRepository.GetConnection()
 	tx, err := dbConnection.Begin()
@@ -1948,3 +1934,303 @@ func (impl *GlobalPluginServiceImpl) MigratePluginDataToParentPluginMetadata(plu
 	}
 	return nil
 }
+
+func (impl *GlobalPluginServiceImpl) GetAllPluginMinData() ([]*bean2.PluginMinDto, error) {
+	pluginsParentMinData, err := impl.globalPluginRepository.GetAllPluginMinData()
+	if err != nil {
+		impl.logger.Errorw("GetAllPluginMinData, error in getting all plugin parent metadata min data", "err", err)
+		return nil, err
+	}
+	pluginMinList := make([]*bean2.PluginMinDto, 0, len(pluginsParentMinData))
+	for _, item := range pluginsParentMinData {
+		//since creating new version of preset plugin is disabled for end user, hence ignoring PRESET plugin in min list
+		if item.Type == repository.PLUGIN_TYPE_PRESET {
+			continue
+		}
+		pluginMinList = append(pluginMinList, bean2.NewPluginMinDto().WithParentPluginId(item.Id).WithPluginName(item.Name).WithIcon(item.Icon))
+	}
+	return pluginMinList, nil
+}
+
+func (impl *GlobalPluginServiceImpl) checkValidationOnPluginNameAndIdentifier(pluginReq *bean2.PluginParentMetadataDto) error {
+	plugins, err := impl.globalPluginRepository.GetAllPluginMinData()
+	if err != nil {
+		impl.logger.Errorw("error in getting all plugins", "err", err)
+		return err
+	}
+	for _, plugin := range plugins {
+		if plugin.Identifier == pluginReq.PluginIdentifier {
+			return util.GetApiError(http.StatusConflict, bean2.PluginWithSameIdentifierExistsError, bean2.PluginWithSameIdentifierExistsError)
+		}
+		if plugin.Name == pluginReq.Name {
+			return util.GetApiError(http.StatusConflict, bean2.PluginWithSameNameExistError, bean2.PluginWithSameNameExistError)
+		}
+	}
+	return nil
+}
+
+func (impl *GlobalPluginServiceImpl) checkValidationOnVersion(pluginReq *bean2.PluginParentMetadataDto) error {
+	pluginVersions, err := impl.globalPluginRepository.GetPluginVersionsByParentId(pluginReq.Id)
+	if err != nil {
+		impl.logger.Errorw("checkValidationOnVersion, error in getting all plugins versions by parentPluginId", "parentPluginId", pluginReq.Id, "err", err)
+		return err
+	}
+	for _, pluginVersion := range pluginVersions {
+		if pluginReq.Versions != nil && len(pluginReq.Versions.DetailedPluginVersionData) > 0 && pluginReq.Versions.DetailedPluginVersionData[0] != nil {
+			// if plugin version from req is already created then return error
+			if pluginVersion.PluginVersion == pluginReq.Versions.DetailedPluginVersionData[0].Version {
+				return util.GetApiError(http.StatusBadRequest, bean2.PluginVersionAlreadyExistError, bean2.PluginVersionAlreadyExistError)
+			}
+		}
+
+	}
+	return nil
+}
+
+func (impl *GlobalPluginServiceImpl) validateV2PluginRequest(pluginReq *bean2.PluginParentMetadataDto) error {
+	if pluginReq.Versions == nil || len(pluginReq.Versions.DetailedPluginVersionData) == 0 || pluginReq.Versions.DetailedPluginVersionData[0] == nil {
+		return util.GetApiError(http.StatusBadRequest, bean2.NoStepDataToProceedError, bean2.NoStepDataToProceedError)
+	}
+	if pluginReq.Id == 0 {
+		//create plugin req.
+		err := impl.checkValidationOnPluginNameAndIdentifier(pluginReq)
+		if err != nil {
+			impl.logger.Errorw("error in checkValidationOnPluginNameAndIdentifier", "err", err)
+			return err
+		}
+	} else {
+		err := impl.checkValidationOnVersion(pluginReq)
+		if err != nil {
+			impl.logger.Errorw("error in checkValidationOnPluginNameAndIdentifier", "err", err)
+			return err
+		}
+	}
+	version := pluginReq.Versions.DetailedPluginVersionData[0].Version
+	if !strings.Contains(version, "v") {
+		version = fmt.Sprintf("v%s", version)
+	}
+	// semantic versioning validation on plugin's version
+	if !semver.IsValid(version) {
+		return util.GetApiError(http.StatusBadRequest, bean2.PluginVersionNotSemanticallyCorrectError, bean2.PluginVersionNotSemanticallyCorrectError)
+	}
+	//validate icon url and size
+	if len(pluginReq.Icon) > 0 {
+		err := utils.FetchIconAndCheckSize(pluginReq.Icon, bean2.PluginIconMaxSizeInBytes)
+		if err != nil {
+			errMsg := fmt.Sprintf("%s err:= %s", bean2.PluginIconNotCorrectOrReachableError, err.Error())
+			return util.GetApiError(http.StatusBadRequest, errMsg, errMsg)
+		}
+	}
+	return nil
+}
+
+func (impl *GlobalPluginServiceImpl) createPluginTagAndRelations(pluginReq *bean2.PluginsVersionDetail, userId int32, tx *pg.Tx) error {
+	if pluginReq.AreNewTagsPresent {
+		err := impl.CreateNewPluginTagsAndRelationsIfRequiredV2(pluginReq, userId, tx)
+		if err != nil {
+			impl.logger.Errorw("createPluginTagAndRelations, error in CreateNewPluginTagsAndRelationsIfRequired", "tags", pluginReq.Tags, "err", err)
+			return err
+		}
+	} else if len(pluginReq.Tags) > 0 {
+		err := impl.CreatePluginTagRelations(pluginReq, userId, tx)
+		if err != nil {
+			impl.logger.Errorw("createPluginTagAndRelations, error in CreatePluginTagRelations", "tags", pluginReq.Tags, "err", err)
+			return err
+		}
+	}
+	return nil
+}
+
+func (impl *GlobalPluginServiceImpl) CreatePluginTagRelations(pluginReq *bean2.PluginsVersionDetail, userId int32, tx *pg.Tx) error {
+	tags, err := impl.globalPluginRepository.GetPluginTagByNames(pluginReq.Tags)
+	if err != nil {
+		impl.logger.Errorw("CreatePluginTagRelations, error in GetPluginTagByNames", "tags", pluginReq.Tags, "err", err)
+		return err
+	}
+	newPluginTagRelationsToCreate := make([]*repository.PluginTagRelation, 0, len(pluginReq.Tags))
+	for _, tag := range tags {
+		newPluginTagRelationsToCreate = append(newPluginTagRelationsToCreate, repository.NewPluginTagRelation().CreateAuditLog(userId).WithTagAndPluginId(tag.Id, pluginReq.Id))
+	}
+
+	if len(newPluginTagRelationsToCreate) > 0 {
+		err = impl.globalPluginRepository.SavePluginTagRelationInBulk(newPluginTagRelationsToCreate, tx)
+		if err != nil {
+			impl.logger.Errorw("CreatePluginTagRelations, error in saving plugin tag relation in bulk", "newPluginTagRelationsToCreate", newPluginTagRelationsToCreate, "err", err)
+			return err
+		}
+	}
+	return nil
+}
+
+func (impl *GlobalPluginServiceImpl) createPluginStepDataAndTagRelations(pluginVersionId int, pluginVersionDetail *bean2.PluginsVersionDetail, userId int32, tx *pg.Tx) error {
+	if len(pluginVersionDetail.PluginSteps) > 0 {
+		err := impl.saveDeepPluginStepData(pluginVersionId, pluginVersionDetail.PluginSteps, userId, tx)
+		if err != nil {
+			impl.logger.Errorw("createNewPluginVersionOfExistingPlugin, error in saving plugin step data", "err", err)
+			return err
+		}
+	} else {
+		return util.GetApiError(http.StatusBadRequest, bean2.PluginStepsNotProvidedError, bean2.PluginStepsNotProvidedError)
+	}
+
+	err := impl.createPluginTagAndRelations(pluginVersionDetail, userId, tx)
+	if err != nil {
+		impl.logger.Errorw("createNewPlugin, error in createPluginTagAndRelations", "tags", pluginVersionDetail.Tags, "err", err)
+		return err
+	}
+	return nil
+}
+
+func (impl *GlobalPluginServiceImpl) createNewPlugin(tx *pg.Tx, pluginDto *bean2.PluginParentMetadataDto, userId int32) (int, error) {
+	pluginParentMetadata, err := impl.globalPluginRepository.SavePluginParentMetadata(tx, adaptor.GetPluginParentMetadataDbObject(pluginDto, userId))
+	if err != nil {
+		impl.logger.Errorw("createNewPlugin, error in saving plugin parent metadata", "pluginDto", pluginDto, "err", err)
+		return 0, err
+	}
+	pluginDto.Id = pluginParentMetadata.Id
+	pluginVersionDto := adaptor.GetPluginVersionMetadataDbObject(pluginDto, userId).
+		WithPluginParentMetadataId(pluginParentMetadata.Id).
+		WithIsLatestFlag(true)
+
+	pluginVersionMetadata, err := impl.globalPluginRepository.SavePluginMetadata(pluginVersionDto, tx)
+	if err != nil {
+		impl.logger.Errorw("createNewPlugin, error in saving plugin version metadata", "pluginDto", pluginDto, "err", err)
+		return 0, err
+	}
+	pluginDto.Versions.DetailedPluginVersionData[0].Id = pluginVersionMetadata.Id
+
+	pluginStageMapping := &repository.PluginStageMapping{
+		PluginId:  pluginParentMetadata.Id,
+		StageType: repository.CI_CD,
+		AuditLog:  sql.NewDefaultAuditLog(userId),
+	}
+	_, err = impl.globalPluginRepository.SavePluginStageMapping(pluginStageMapping, tx)
+	if err != nil {
+		impl.logger.Errorw("createNewPlugin, error in saving plugin stage mapping", "pluginDto", pluginDto, "err", err)
+		return 0, err
+	}
+
+	err = impl.createPluginStepDataAndTagRelations(pluginVersionMetadata.Id, pluginDto.Versions.DetailedPluginVersionData[0], userId, tx)
+	if err != nil {
+		impl.logger.Errorw("createNewPlugin, error in createPluginStepDataAndTagRelations", "pluginDto", pluginDto, "err", err)
+		return 0, err
+	}
+	return pluginVersionMetadata.Id, nil
+}
+
+func (impl *GlobalPluginServiceImpl) createNewPluginVersionOfExistingPlugin(tx *pg.Tx, pluginDto *bean2.PluginParentMetadataDto, userId int32) (int, error) {
+	var pluginParentMinData *repository.PluginParentMetadata
+	var err error
+	pluginParentMinData, err = impl.globalPluginRepository.GetPluginParentMinDataById(pluginDto.Id)
+	if err != nil {
+		impl.logger.Errorw("createNewPluginVersionOfExistingPlugin, error in getting plugin parent metadata", "pluginDto", pluginDto, "err", err)
+		return 0, err
+	}
+	// before saving new plugin version marking previous version's isLatest as false.
+	err = impl.globalPluginRepository.MarkPreviousPluginVersionLatestFalse(pluginParentMinData.Id)
+	if err != nil {
+		impl.logger.Errorw("createNewPluginVersionOfExistingPlugin, error in MarkPreviousPluginVersionLatestFalse", "pluginParentId", pluginDto.Id, "err", err)
+		return 0, err
+	}
+	pluginDto.Name = pluginParentMinData.Name
+	pluginVersionDto := adaptor.GetPluginVersionMetadataDbObject(pluginDto, userId).
+		WithPluginParentMetadataId(pluginParentMinData.Id).
+		WithIsLatestFlag(true)
+
+	pluginVersionMetadata, err := impl.globalPluginRepository.SavePluginMetadata(pluginVersionDto, tx)
+	if err != nil {
+		impl.logger.Errorw("createNewPluginVersionOfExistingPlugin, error in saving plugin version metadata", "pluginDto", pluginDto, "err", err)
+		return 0, err
+	}
+	pluginDto.Versions.DetailedPluginVersionData[0].Id = pluginVersionMetadata.Id
+
+	err = impl.createPluginStepDataAndTagRelations(pluginVersionMetadata.Id, pluginDto.Versions.DetailedPluginVersionData[0], userId, tx)
+	if err != nil {
+		impl.logger.Errorw("createNewPluginVersionOfExistingPlugin, error in createPluginStepDataAndTagRelations", "pluginDto", pluginDto, "err", err)
+		return 0, err
+	}
+	return pluginVersionMetadata.Id, nil
+}
+
+func (impl *GlobalPluginServiceImpl) CreatePluginOrVersions(pluginDto *bean2.PluginParentMetadataDto, userId int32) (int, error) {
+	err := impl.validateV2PluginRequest(pluginDto)
+	if err != nil {
+		impl.logger.Errorw("CreatePluginOrVersions, error in validating create plugin request", "pluginReqDto", pluginDto, "err", err)
+		return 0, err
+	}
+
+	dbConnection := impl.globalPluginRepository.GetConnection()
+	tx, err := dbConnection.Begin()
+	if err != nil {
+		return 0, err
+	}
+	// Rollback tx on error.
+	defer tx.Rollback()
+	var versionMetadataId int
+	if pluginDto.Id > 0 {
+		// create new version of existing plugin req.
+		versionMetadataId, err = impl.createNewPluginVersionOfExistingPlugin(tx, pluginDto, userId)
+		if err != nil {
+			impl.logger.Errorw("CreatePluginOrVersions, error in creating new version of an existing plugin", "existingPluginName", pluginDto.Name, "err", err)
+			return 0, err
+		}
+	} else {
+		// create new plugin req.
+		versionMetadataId, err = impl.createNewPlugin(tx, pluginDto, userId)
+		if err != nil {
+			impl.logger.Errorw("CreatePluginOrVersions, error in creating new plugin", "pluginDto", pluginDto, "err", err)
+			return 0, err
+		}
+	}
+	err = tx.Commit()
+	if err != nil {
+		impl.logger.Errorw("CreatePluginOrVersions, error in committing db transaction", "err", err)
+		return 0, err
+	}
+	return versionMetadataId, nil
+}
+
+func (impl *GlobalPluginServiceImpl) CreateNewPluginTagsAndRelationsIfRequiredV2(pluginReq *bean2.PluginsVersionDetail, userId int32, tx *pg.Tx) error {
+	allPluginTags, err := impl.globalPluginRepository.GetAllPluginTags()
+	if err != nil {
+		impl.logger.Errorw("CreateNewPluginTagsAndRelationsIfRequiredV2, error in getting all plugin tags", "err", err)
+		return err
+	}
+	existingTagMap := make(map[string]*repository.PluginTag, len(allPluginTags))
+	for _, tag := range allPluginTags {
+		existingTagMap[tag.Name] = tag
+	}
+	//check for new tags, then create new plugin_tag and plugin_tag_relation entry in db when new tags are present in request
+	newPluginTagsToCreate := make([]*repository.PluginTag, 0, len(pluginReq.Tags))
+	newPluginTagRelationsToCreate := make([]*repository.PluginTagRelation, 0, len(pluginReq.Tags))
+
+	for _, tagReq := range pluginReq.Tags {
+		if _, ok := existingTagMap[tagReq]; !ok {
+			newPluginTagsToCreate = append(newPluginTagsToCreate, repository.NewPluginTag().CreateAuditLog(userId).WithName(tagReq))
+		}
+	}
+
+	if len(newPluginTagsToCreate) > 0 {
+		err = impl.globalPluginRepository.SavePluginTagInBulk(newPluginTagsToCreate, tx)
+		if err != nil {
+			impl.logger.Errorw("CreateNewPluginTagsAndRelationsIfRequiredV2, error in saving plugin tag", "newPluginTags", newPluginTagsToCreate, "err", err)
+			return err
+		}
+		for _, newTag := range newPluginTagsToCreate {
+			existingTagMap[newTag.Name] = newTag
+		}
+	}
+
+	for _, tag := range pluginReq.Tags {
+		newPluginTagRelationsToCreate = append(newPluginTagRelationsToCreate, repository.NewPluginTagRelation().CreateAuditLog(userId).WithTagAndPluginId(existingTagMap[tag].Id, pluginReq.Id))
+	}
+
+	if len(newPluginTagRelationsToCreate) > 0 {
+		err = impl.globalPluginRepository.SavePluginTagRelationInBulk(newPluginTagRelationsToCreate, tx)
+		if err != nil {
+			impl.logger.Errorw("CreateNewPluginTagsAndRelationsIfRequiredV2, error in saving plugin tag relation in bulk", "newPluginTagRelationsToCreate", newPluginTagRelationsToCreate, "err", err)
+			return err
+		}
+	}
+	return nil
+}
diff --git a/pkg/plugin/adaptor/adaptor.go b/pkg/plugin/adaptor/adaptor.go
new file mode 100644
index 00000000000..e5e0f50e9d3
--- /dev/null
+++ b/pkg/plugin/adaptor/adaptor.go
@@ -0,0 +1,80 @@
+package adaptor
+
+import (
+	bean2 "github.com/devtron-labs/devtron/pkg/plugin/bean"
+	"github.com/devtron-labs/devtron/pkg/plugin/repository"
+	"github.com/devtron-labs/devtron/pkg/sql"
+)
+
+func GetPluginParentMetadataDbObject(pluginDto *bean2.PluginParentMetadataDto, userId int32) *repository.PluginParentMetadata {
+	return repository.NewPluginParentMetadata().CreateAuditLog(userId).
+		WithBasicMetadata(pluginDto.Name, pluginDto.PluginIdentifier, pluginDto.Description, pluginDto.Icon, repository.PLUGIN_TYPE_SHARED)
+}
+
+func GetPluginVersionMetadataDbObject(pluginDto *bean2.PluginParentMetadataDto, userId int32) *repository.PluginMetadata {
+	versionDto := pluginDto.Versions.DetailedPluginVersionData[0]
+	return repository.NewPluginVersionMetadata().CreateAuditLog(userId).WithBasicMetadata(pluginDto.Name, versionDto.Description, versionDto.Version, versionDto.DocLink)
+}
+
+func GetPluginStepDbObject(pluginStepDto *bean2.PluginStepsDto, pluginVersionMetadataId int, userId int32) *repository.PluginStep {
+	return &repository.PluginStep{
+		PluginId:            pluginVersionMetadataId,
+		Name:                pluginStepDto.Name,
+		Description:         pluginStepDto.Description,
+		Index:               1,
+		StepType:            repository.PLUGIN_STEP_TYPE_INLINE,
+		RefPluginId:         pluginStepDto.RefPluginId,
+		OutputDirectoryPath: pluginStepDto.OutputDirectoryPath,
+		DependentOnStep:     pluginStepDto.DependentOnStep,
+		AuditLog:            sql.NewDefaultAuditLog(userId),
+	}
+}
+func GetPluginPipelineScriptDbObject(pluginPipelineScript *bean2.PluginPipelineScript, userId int32) *repository.PluginPipelineScript {
+	return &repository.PluginPipelineScript{
+		Script:                   pluginPipelineScript.Script,
+		StoreScriptAt:            pluginPipelineScript.StoreScriptAt,
+		Type:                     pluginPipelineScript.Type,
+		DockerfileExists:         pluginPipelineScript.DockerfileExists,
+		MountPath:                pluginPipelineScript.MountPath,
+		MountCodeToContainer:     pluginPipelineScript.MountCodeToContainer,
+		MountCodeToContainerPath: pluginPipelineScript.MountCodeToContainerPath,
+		MountDirectoryFromHost:   pluginPipelineScript.MountDirectoryFromHost,
+		ContainerImagePath:       pluginPipelineScript.ContainerImagePath,
+		ImagePullSecretType:      pluginPipelineScript.ImagePullSecretType,
+		ImagePullSecret:          pluginPipelineScript.ImagePullSecret,
+		AuditLog:                 sql.NewDefaultAuditLog(userId),
+	}
+
+}
+
+func GetPluginStepVariableDbObject(pluginStepId int, pluginVariableDto *bean2.PluginVariableDto, userId int32) *repository.PluginStepVariable {
+	return &repository.PluginStepVariable{
+		PluginStepId:              pluginStepId,
+		Name:                      pluginVariableDto.Name,
+		Format:                    pluginVariableDto.Format,
+		Description:               pluginVariableDto.Description,
+		IsExposed:                 true, //currently hard coding this, later after plugin creation gets more mature will let user decide
+		AllowEmptyValue:           pluginVariableDto.AllowEmptyValue,
+		DefaultValue:              pluginVariableDto.DefaultValue,
+		Value:                     pluginVariableDto.Value,
+		VariableType:              pluginVariableDto.VariableType,
+		ValueType:                 pluginVariableDto.ValueType,
+		PreviousStepIndex:         pluginVariableDto.PreviousStepIndex,
+		VariableStepIndex:         1, //currently hard coding this, later after plugin creation gets more mature will let user decide
+		VariableStepIndexInPlugin: pluginVariableDto.VariableStepIndexInPlugin,
+		ReferenceVariableName:     pluginVariableDto.ReferenceVariableName,
+		AuditLog:                  sql.NewDefaultAuditLog(userId),
+	}
+}
+
+func GetPluginStepConditionDbObject(stepDataId, pluginStepVariableId int, pluginStepCondition *bean2.PluginStepCondition,
+	userId int32) *repository.PluginStepCondition {
+	return &repository.PluginStepCondition{
+		PluginStepId:        stepDataId,
+		ConditionVariableId: pluginStepVariableId,
+		ConditionType:       pluginStepCondition.ConditionType,
+		ConditionalOperator: pluginStepCondition.ConditionalOperator,
+		ConditionalValue:    pluginStepCondition.ConditionalValue,
+		AuditLog:            sql.NewDefaultAuditLog(userId),
+	}
+}
diff --git a/pkg/plugin/bean/bean.go b/pkg/plugin/bean/bean.go
index 55424f3caac..c31d3463327 100644
--- a/pkg/plugin/bean/bean.go
+++ b/pkg/plugin/bean/bean.go
@@ -13,7 +13,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package bean
 
 import (
@@ -44,15 +43,47 @@ type PluginListComponentDto struct { //created new struct for backward compatibi
 }
 
 type PluginMetadataDto struct {
-	Id          int               `json:"id"`
-	Name        string            `json:"name"`
-	Description string            `json:"description"`
-	Type        string            `json:"type,omitempty" validate:"oneof=SHARED PRESET"` // SHARED, PRESET etc
-	Icon        string            `json:"icon,omitempty"`
-	Tags        []string          `json:"tags"`
-	Action      int               `json:"action,omitempty"`
-	PluginStage string            `json:"pluginStage,omitempty"`
-	PluginSteps []*PluginStepsDto `json:"pluginSteps,omitempty"`
+	Id                int               `json:"id"`
+	Name              string            `json:"name" validate:"required,min=3,max=100,global-entity-name"`
+	Description       string            `json:"description" validate:"max=300"`
+	Type              string            `json:"type,omitempty" validate:"oneof=SHARED PRESET"` // SHARED, PRESET etc
+	Icon              string            `json:"icon,omitempty"`
+	Tags              []string          `json:"tags"`
+	Action            int               `json:"action,omitempty"`
+	PluginStage       string            `json:"pluginStage,omitempty"`
+	PluginSteps       []*PluginStepsDto `json:"pluginSteps,omitempty"`
+	AreNewTagsPresent bool              `json:"areNewTagsPresent,omitempty"`
+}
+
+type PluginMinDto struct {
+	ParentPluginId  int    `json:"id,omitempty"`
+	PluginName      string `json:"name,omitempty"`
+	Icon            string `json:"icon,omitempty"`
+	PluginVersionId int    `json:"pluginVersionId,omitempty"`
+}
+
+func NewPluginMinDto() *PluginMinDto {
+	return &PluginMinDto{}
+}
+
+func (r *PluginMinDto) WithParentPluginId(id int) *PluginMinDto {
+	r.ParentPluginId = id
+	return r
+}
+
+func (r *PluginMinDto) WithPluginName(name string) *PluginMinDto {
+	r.PluginName = name
+	return r
+}
+
+func (r *PluginMinDto) WithIcon(icon string) *PluginMinDto {
+	r.Icon = icon
+	return r
+}
+
+func (r *PluginMinDto) WithPluginVersionId(versionId int) *PluginMinDto {
+	r.PluginVersionId = versionId
+	return r
 }
 
 type PluginsDto struct {
@@ -76,9 +107,9 @@ func (r *PluginsDto) WithTotalCount(count int) *PluginsDto {
 
 type PluginParentMetadataDto struct {
 	Id               int             `json:"id"`
-	Name             string          `json:"name"`
-	PluginIdentifier string          `json:"pluginIdentifier"`
-	Description      string          `json:"description"`
+	Name             string          `json:"name" validate:"required,min=3,max=100,global-entity-name"`
+	PluginIdentifier string          `json:"pluginIdentifier" validate:"required,min=3,max=100,global-entity-name"`
+	Description      string          `json:"description" validate:"max=300"`
 	Type             string          `json:"type,omitempty" validate:"oneof=SHARED PRESET"`
 	Icon             string          `json:"icon,omitempty"`
 	Versions         *PluginVersions `json:"pluginVersions"`
@@ -124,17 +155,6 @@ type PluginVersions struct {
 	MinimalPluginVersionData  []*PluginsVersionDetail `json:"minimalPluginVersionData"`  // contains only few metadata
 }
 
-type PluginMinDto struct {
-	PluginName     string                  `json:"pluginName"`
-	PluginVersions []*PluginVersionsMinDto `json:"pluginVersions"`
-	Icon           string                  `json:"icon"`
-}
-
-type PluginVersionsMinDto struct {
-	Id      int    `json:"id"`
-	Version string `json:"version"`
-}
-
 func NewPluginVersions() *PluginVersions {
 	return &PluginVersions{}
 }
@@ -154,7 +174,7 @@ type PluginsVersionDetail struct {
 	InputVariables  []*PluginVariableDto `json:"inputVariables"`
 	OutputVariables []*PluginVariableDto `json:"outputVariables"`
 	DocLink         string               `json:"docLink"`
-	Version         string               `json:"pluginVersion"`
+	Version         string               `json:"pluginVersion" validate:"max=50,min=3"`
 	IsLatest        bool                 `json:"isLatest"`
 	UpdatedBy       string               `json:"updatedBy"`
 	CreatedOn       time.Time            `json:"-"`
@@ -336,10 +356,18 @@ type RegistryCredentials struct {
 }
 
 const (
-	NoPluginOrParentIdProvidedErr      = "no value for pluginVersionIds and parentPluginIds provided in query param"
-	NoPluginFoundForThisSearchQueryErr = "unable to find desired plugin for the query filter"
+	NoPluginOrParentIdProvidedErr            = "no value for pluginVersionIds and parentPluginIds provided in query param"
+	NoPluginFoundForThisSearchQueryErr       = "unable to find desired plugin for the query filter"
+	PluginStepsNotProvidedError              = "plugin steps not provided"
+	PluginWithSameNameExistError             = "plugin with the same name exists, please choose another name"
+	PluginWithSameIdentifierExistsError      = "plugin with the same identifier exists, please choose another identifier name"
+	PluginVersionNotSemanticallyCorrectError = "please provide a plugin version that adheres to Semantic Versioning 2.0.0 to ensure compatibility and proper versioning"
+	PluginIconNotCorrectOrReachableError     = "cannot validate icon, make sure that provided url link is reachable"
+	PluginVersionAlreadyExistError           = "this plugin version already exists, please provide another plugin version"
+	NoStepDataToProceedError                 = "no step data provided to save, please provide a plugin step to proceed further"
 )
 
 const (
-	SpecialCharsRegex = ` !"#$%&'()*+,./:;<=>?@[\]^_{|}~` + "`"
+	SpecialCharsRegex        = ` !"#$%&'()*+,./:;<=>?@[\]^_{|}~` + "`"
+	PluginIconMaxSizeInBytes = 2 * 1024 * 1024
 )
diff --git a/pkg/plugin/repository/GlobalPluginRepository.go b/pkg/plugin/repository/GlobalPluginRepository.go
index 8b650935231..9cc50748f29 100644
--- a/pkg/plugin/repository/GlobalPluginRepository.go
+++ b/pkg/plugin/repository/GlobalPluginRepository.go
@@ -99,6 +99,16 @@ func (r *PluginParentMetadata) CreateAuditLog(userId int32) *PluginParentMetadat
 	return r
 }
 
+func (r *PluginParentMetadata) WithBasicMetadata(name, identifier, description, icon string, pluginType PluginType) *PluginParentMetadata {
+	r.Name = name
+	r.Identifier = identifier
+	r.Description = description
+	r.Icon = icon
+	r.Type = pluginType
+	r.Deleted = false
+	return r
+}
+
 // SetParentPluginMetadata method signature used only for migration purposes, sets pluginVersionsMetadata into plugin_parent_metadata
 func (r *PluginParentMetadata) SetParentPluginMetadata(pluginMetadata *PluginMetadata) *PluginParentMetadata {
 	r.Name = pluginMetadata.Name
@@ -135,6 +145,38 @@ type PluginMetadata struct {
 	sql.AuditLog
 }
 
+func NewPluginVersionMetadata() *PluginMetadata {
+	return &PluginMetadata{}
+}
+
+func (r *PluginMetadata) CreateAuditLog(userId int32) *PluginMetadata {
+	r.CreatedBy = userId
+	r.CreatedOn = time.Now()
+	r.UpdatedBy = userId
+	r.UpdatedOn = time.Now()
+	return r
+}
+
+func (r *PluginMetadata) WithBasicMetadata(name, description, pluginVersion, docLink string) *PluginMetadata {
+	r.Name = name
+	r.PluginVersion = pluginVersion
+	r.Description = description
+	r.DocLink = docLink
+	r.Deleted = false
+	r.IsDeprecated = false
+	return r
+}
+
+func (r *PluginMetadata) WithPluginParentMetadataId(parentId int) *PluginMetadata {
+	r.PluginParentMetadataId = parentId
+	return r
+}
+
+func (r *PluginMetadata) WithIsLatestFlag(isLatest bool) *PluginMetadata {
+	r.IsLatest = isLatest
+	return r
+}
+
 type PluginTag struct {
 	tableName struct{} `sql:"plugin_tag" pg:",discard_unknown_columns"`
 	Id        int      `sql:"id,pk"`
@@ -143,6 +185,23 @@ type PluginTag struct {
 	sql.AuditLog
 }
 
+func NewPluginTag() *PluginTag {
+	return &PluginTag{}
+}
+
+func (r *PluginTag) WithName(name string) *PluginTag {
+	r.Name = name
+	return r
+}
+
+func (r *PluginTag) CreateAuditLog(userId int32) *PluginTag {
+	r.CreatedBy = userId
+	r.CreatedOn = time.Now()
+	r.UpdatedBy = userId
+	r.UpdatedOn = time.Now()
+	return r
+}
+
 type PluginTagRelation struct {
 	tableName struct{} `sql:"plugin_tag_relation" pg:",discard_unknown_columns"`
 	Id        int      `sql:"id,pk"`
@@ -151,6 +210,24 @@ type PluginTagRelation struct {
 	sql.AuditLog
 }
 
+func NewPluginTagRelation() *PluginTagRelation {
+	return &PluginTagRelation{}
+}
+
+func (r *PluginTagRelation) WithTagAndPluginId(tagId, pluginId int) *PluginTagRelation {
+	r.TagId = tagId
+	r.PluginId = pluginId
+	return r
+}
+
+func (r *PluginTagRelation) CreateAuditLog(userId int32) *PluginTagRelation {
+	r.CreatedBy = userId
+	r.CreatedOn = time.Now()
+	r.UpdatedBy = userId
+	r.UpdatedOn = time.Now()
+	return r
+}
+
 // Below two tables are used at pipeline-steps level too
 
 type PluginPipelineScript struct {
@@ -247,7 +324,9 @@ type GlobalPluginRepository interface {
 	GetMetaDataForAllPlugins() ([]*PluginMetadata, error)
 	GetMetaDataForPluginWithStageType(stageType int) ([]*PluginMetadata, error)
 	GetMetaDataByPluginId(pluginId int) (*PluginMetadata, error)
+	GetMetaDataByPluginIds(pluginIds []int) ([]*PluginMetadata, error)
 	GetAllPluginTags() ([]*PluginTag, error)
+	GetPluginTagByNames(tagNames []string) ([]*PluginTag, error)
 	GetAllPluginTagRelations() ([]*PluginTagRelation, error)
 	GetTagsByPluginId(pluginId int) ([]string, error)
 	GetScriptDetailById(id int) (*PluginPipelineScript, error)
@@ -264,10 +343,14 @@ type GlobalPluginRepository interface {
 	GetConditionsByPluginId(pluginId int) ([]*PluginStepCondition, error)
 	GetPluginStageMappingByPluginId(pluginId int) (*PluginStageMapping, error)
 	GetConnection() (dbConnection *pg.DB)
+	GetPluginVersionsByParentId(parentPluginId int) ([]*PluginMetadata, error)
 
 	GetPluginParentMetadataByIdentifier(pluginIdentifier string) (*PluginParentMetadata, error)
 	GetAllFilteredPluginParentMetadata(searchKey string, tags []string) ([]*PluginParentMetadata, error)
 	GetPluginParentMetadataByIds(ids []int) ([]*PluginParentMetadata, error)
+	GetAllPluginMinData() ([]*PluginParentMetadata, error)
+	GetPluginParentMinDataById(id int) (*PluginParentMetadata, error)
+	MarkPreviousPluginVersionLatestFalse(pluginParentId int) error
 
 	SavePluginMetadata(pluginMetadata *PluginMetadata, tx *pg.Tx) (*PluginMetadata, error)
 	SavePluginStageMapping(pluginStageMapping *PluginStageMapping, tx *pg.Tx) (*PluginStageMapping, error)
@@ -351,6 +434,19 @@ func (impl *GlobalPluginRepositoryImpl) GetAllPluginTags() ([]*PluginTag, error)
 	return tags, nil
 }
 
+func (impl *GlobalPluginRepositoryImpl) GetPluginTagByNames(tagNames []string) ([]*PluginTag, error) {
+	var tags []*PluginTag
+	err := impl.dbConnection.Model(&tags).
+		Where("deleted = ?", false).
+		Where("name in (?)", pg.In(tagNames)).
+		Select()
+	if err != nil {
+		impl.logger.Errorw("err in getting all tags by names", "tagNames", tagNames, "err", err)
+		return nil, err
+	}
+	return tags, nil
+}
+
 func (impl *GlobalPluginRepositoryImpl) GetAllPluginTagRelations() ([]*PluginTagRelation, error) {
 	var rel []*PluginTagRelation
 	err := impl.dbConnection.Model(&rel).
@@ -385,6 +481,18 @@ func (impl *GlobalPluginRepositoryImpl) GetMetaDataByPluginId(pluginId int) (*Pl
 	return &plugin, nil
 }
 
+func (impl *GlobalPluginRepositoryImpl) GetMetaDataByPluginIds(pluginIds []int) ([]*PluginMetadata, error) {
+	var plugins []*PluginMetadata
+	err := impl.dbConnection.Model(&plugins).
+		Where("deleted = ?", false).
+		Where("id in (?)", pg.In(pluginIds)).Select()
+	if err != nil {
+		impl.logger.Errorw("err in getting plugins by pluginIds", "pluginIds", pluginIds, "err", err)
+		return nil, err
+	}
+	return plugins, nil
+}
+
 func (impl *GlobalPluginRepositoryImpl) GetStepsByPluginIds(pluginIds []int) ([]*PluginStep, error) {
 	var pluginSteps []*PluginStep
 	err := impl.dbConnection.Model(&pluginSteps).
@@ -511,6 +619,20 @@ func (impl *GlobalPluginRepositoryImpl) GetPluginByName(pluginName string) ([]*P
 
 }
 
+func (impl *GlobalPluginRepositoryImpl) GetPluginVersionsByParentId(parentPluginId int) ([]*PluginMetadata, error) {
+	var plugin []*PluginMetadata
+	err := impl.dbConnection.Model(&plugin).
+		Where("plugin_parent_metadata_id = ?", parentPluginId).
+		Where("deleted = ?", false).
+		Where("is_deprecated = ?", false).
+		Select()
+	if err != nil {
+		impl.logger.Errorw("err in getting pluginVersionMetadata by parentPluginId", "parentPluginId", parentPluginId, "err", err)
+		return nil, err
+	}
+	return plugin, nil
+}
+
 func (impl *GlobalPluginRepositoryImpl) GetAllPluginMetaData() ([]*PluginMetadata, error) {
 	var plugins []*PluginMetadata
 	err := impl.dbConnection.Model(&plugins).Where("deleted = ?", false).Select()
@@ -700,6 +822,18 @@ func (impl *GlobalPluginRepositoryImpl) GetPluginParentMetadataByIdentifier(plug
 	return &pluginParentMetadata, nil
 }
 
+func (impl *GlobalPluginRepositoryImpl) GetPluginParentMinDataById(id int) (*PluginParentMetadata, error) {
+	var pluginParentMetadata PluginParentMetadata
+	err := impl.dbConnection.Model(&pluginParentMetadata).
+		Column("plugin_parent_metadata.id", "plugin_parent_metadata.name").
+		Where("id = ?", id).
+		Where("deleted = ?", false).Select()
+	if err != nil {
+		return nil, err
+	}
+	return &pluginParentMetadata, nil
+}
+
 func (impl *GlobalPluginRepositoryImpl) SavePluginParentMetadata(tx *pg.Tx, pluginParentMetadata *PluginParentMetadata) (*PluginParentMetadata, error) {
 	err := tx.Insert(pluginParentMetadata)
 	return pluginParentMetadata, err
@@ -712,24 +846,20 @@ func (impl *GlobalPluginRepositoryImpl) UpdatePluginMetadataInBulk(pluginsMetada
 
 func (impl *GlobalPluginRepositoryImpl) GetAllFilteredPluginParentMetadata(searchKey string, tags []string) ([]*PluginParentMetadata, error) {
 	var plugins []*PluginParentMetadata
-	subQuery := "select ppm.id, ppm.identifier,ppm.name,ppm.description,ppm.type,ppm.icon,ppm.deleted,ppm.created_by, ppm.created_on,ppm.updated_by,ppm.updated_on from plugin_parent_metadata ppm" +
+	query := "select ppm.id, ppm.identifier,ppm.name,ppm.description,ppm.type,ppm.icon,ppm.deleted,ppm.created_by, ppm.created_on,ppm.updated_by,ppm.updated_on from plugin_parent_metadata ppm" +
 		" inner join plugin_metadata pm on pm.plugin_parent_metadata_id=ppm.id"
-	whereCondition := fmt.Sprintf(" where ppm.deleted=false")
-	orderCondition := fmt.Sprintf(" ORDER BY ppm.id asc")
+	whereCondition := fmt.Sprintf(" where ppm.deleted=false AND pm.deleted=false AND pm.is_latest=true")
 	if len(tags) > 0 {
-		subQuery = "select DISTINCT ON(ppm.id) ppm.id, ppm.identifier,ppm.name,ppm.description,ppm.type,ppm.icon,ppm.deleted,ppm.created_by, ppm.created_on,ppm.updated_by,ppm.updated_on from plugin_parent_metadata ppm" +
-			" inner join plugin_metadata pm on pm.plugin_parent_metadata_id=ppm.id" +
-			" left join plugin_tag_relation ptr on ptr.plugin_id=pm.id" +
-			" left join plugin_tag pt on ptr.tag_id=pt.id"
-		whereCondition += fmt.Sprintf("  AND pm.deleted=false AND pt.deleted=false AND pt.name in (%s)", helper.GetCommaSepratedStringWithComma(tags))
+		tagFilterSubQuery := fmt.Sprintf("select ptr.plugin_id from plugin_tag_relation ptr inner join plugin_tag pt on ptr.tag_id =pt.id where pt.deleted =false and  pt.name in (%s) group by ptr.plugin_id having count(ptr.plugin_id )=%d", helper.GetCommaSepratedStringWithComma(tags), len(tags))
+		whereCondition += fmt.Sprintf(" AND pm.id in (%s)", tagFilterSubQuery)
 	}
 	if len(searchKey) > 0 {
 		searchKeyLike := "%" + searchKey + "%"
 		whereCondition += fmt.Sprintf(" AND (pm.description ilike '%s' or pm.name ilike '%s')", searchKeyLike, searchKeyLike)
 	}
-	whereCondition += fmt.Sprintf(" AND pm.is_latest=true")
-	subQuery += whereCondition + orderCondition
-	query := fmt.Sprintf(" select * from (%s) x ORDER BY name asc;", subQuery)
+	orderCondition := " ORDER BY ppm.name asc;"
+
+	query += whereCondition + orderCondition
 	_, err := impl.dbConnection.Query(&plugins, query)
 	if err != nil {
 		return nil, err
@@ -749,3 +879,29 @@ func (impl *GlobalPluginRepositoryImpl) GetPluginParentMetadataByIds(ids []int)
 	}
 	return plugins, nil
 }
+
+func (impl *GlobalPluginRepositoryImpl) GetAllPluginMinData() ([]*PluginParentMetadata, error) {
+	var plugins []*PluginParentMetadata
+	err := impl.dbConnection.Model(&plugins).
+		Column("plugin_parent_metadata.id", "plugin_parent_metadata.name", "plugin_parent_metadata.type", "plugin_parent_metadata.icon", "plugin_parent_metadata.identifier").
+		Where("deleted = ?", false).
+		Select()
+	if err != nil {
+		impl.logger.Errorw("err in getting all plugin parent metadata min data", "err", err)
+		return nil, err
+	}
+	return plugins, nil
+}
+
+func (impl *GlobalPluginRepositoryImpl) MarkPreviousPluginVersionLatestFalse(pluginParentId int) error {
+	var model PluginMetadata
+	_, err := impl.dbConnection.Model(&model).
+		Set("is_latest = ?", false).
+		Where("id = (select id from plugin_metadata where plugin_parent_metadata_id = ? and is_latest =true order by created_on desc limit ?)", pluginParentId, 1).
+		Update()
+	if err != nil {
+		impl.logger.Errorw("error in updating last version isLatest as false for a plugin parent id", "pluginParentId", pluginParentId, "err", err)
+		return err
+	}
+	return nil
+}
diff --git a/pkg/plugin/utils/utils.go b/pkg/plugin/utils/utils.go
index 6d78a291439..168e694d89b 100644
--- a/pkg/plugin/utils/utils.go
+++ b/pkg/plugin/utils/utils.go
@@ -21,9 +21,11 @@ import (
 	"fmt"
 	bean2 "github.com/devtron-labs/devtron/pkg/plugin/bean"
 	"github.com/devtron-labs/devtron/pkg/plugin/repository"
+	"net/http"
 	"regexp"
 	"sort"
 	"strings"
+	"time"
 )
 
 func GetStageType(stageTypeReq string) (int, error) {
@@ -72,3 +74,26 @@ func SortPluginsVersionDetailSliceByCreatedOn(pluginsVersionDetail []*bean2.Plug
 		return false
 	})
 }
+
+func FetchIconAndCheckSize(url string, maxSize int64) error {
+	client := http.Client{
+		Timeout: 5 * time.Second,
+	}
+	iconResp, err := client.Get(url)
+	if err != nil {
+		return fmt.Errorf("error in fetching icon : %s", err.Error())
+	}
+	if iconResp != nil {
+		if iconResp.StatusCode >= 200 && iconResp.StatusCode < 300 {
+			if iconResp.ContentLength > maxSize {
+				return fmt.Errorf("icon size too large")
+			}
+			iconResp.Body.Close()
+		} else {
+			return fmt.Errorf("error in fetching icon : %s", iconResp.Status)
+		}
+	} else {
+		return fmt.Errorf("error in fetching icon : empty response")
+	}
+	return nil
+}