modelRT/handler/real_time_data_subscription.go

// Package handler provides HTTP handlers for various endpoints.
package handler

import (
	"context"
	"fmt"
	"maps"
	"net/http"
	"sync"

	"modelRT/constants"
	"modelRT/database"
	"modelRT/logger"
	"modelRT/network"
	"modelRT/orm"
	"modelRT/util"

	"github.com/gin-gonic/gin"
	"github.com/gofrs/uuid"
	"gorm.io/gorm"
)

var globalSubState *SharedSubState

func init() {
	globalSubState = NewSharedSubState()
}

// RealTimeSubHandler define real time data subscriptions process API
// @Summary 开始或结束订阅实时数据
// @Description 根据用户输入的组件token,从 modelRT 服务中开始或结束对于量测节点的实时数据的订阅
// @Tags RealTime Component
// @Accept json
// @Produce json
// @Param request body network.RealTimeSubRequest true "量测节点实时数据订阅"
// @Success 200 {object} network.SuccessResponse{payload=network.RealTimeSubPayload} "订阅实时数据结果列表"
//
//		@Example 200 {
//		    "code": 200,
//		    "msg": "success",
//		    "payload": {
//		        "targets": [
//		            {
//						"id": "grid1.zone1.station1.ns1.tag1.bay.I11_C_rms",
//						"code": "1001",
//	                    "msg": "subscription success"
//	                },
//	                {
//	                    "id": "grid1.zone1.station1.ns1.tag1.bay.I11_B_rms",
//	                    "code": "1002",
//	                    "msg": "subscription failed"
//	                }
//		        ]
//		    }
//		}
//
// @Failure 400 {object} network.FailureResponse{payload=network.RealTimeSubPayload} "订阅实时数据结果列表"
//
//	@Example 400 {
//	    "code": 400,
//	    "msg": "failed to get recommend data from redis",
//		"payload": {
//		    "targets": [
//		        {
//					"id": "grid1.zone1.station1.ns1.tag1.bay.I11_A_rms",
//					"code": "1002",
//	                "msg": "subscription failed"
//	            },
//	            {
//	                "id": "grid1.zone1.station1.ns1.tag1.bay.I11_B_rms",
//	                "code": "1002",
//	                "msg": "subscription failed"
//	            }
//		    ]
//		}
//	}
//
// @Router /monitors/data/subscriptions [post]
func RealTimeSubHandler(c *gin.Context) {
	var request network.RealTimeSubRequest
	var subAction string
	var clientID string

	if err := c.ShouldBindJSON(&request); err != nil {
		logger.Error(c, "failed to unmarshal real time query request", "error", err)
		c.JSON(http.StatusOK, network.FailureResponse{
			Code: http.StatusBadRequest,
			Msg:  err.Error(),
		})
		return
	}

	if request.Action == constants.SubStartAction && request.ClientID == "" {
		subAction = request.Action
		id, err := uuid.NewV4()
		if err != nil {
			logger.Error(c, "failed to generate client id", "error", err)
			c.JSON(http.StatusOK, network.FailureResponse{
				Code: http.StatusBadRequest,
				Msg:  err.Error(),
			})
			return
		}
		clientID = id.String()
	} else if request.Action == constants.SubStartAction && request.ClientID != "" {
		subAction = constants.SubAppendAction
		clientID = request.ClientID
	} else if request.Action == constants.SubStopAction && request.ClientID != "" {
		subAction = request.Action
		clientID = request.ClientID
	} else if request.Action == constants.SubUpdateAction && request.ClientID != "" {
		subAction = request.Action
		clientID = request.ClientID
	}

	pgClient := database.GetPostgresDBClient()
	// open transaction
	tx := pgClient.Begin()
	defer tx.Commit()

	switch subAction {
	case constants.SubStartAction:
		results, err := globalSubState.CreateConfig(c, tx, clientID, request.Measurements)
		if err != nil {
			logger.Error(c, "create real time data subscription config failed", "error", err)
			c.JSON(http.StatusOK, network.FailureResponse{
				Code: http.StatusBadRequest,
				Msg:  err.Error(),
				Payload: network.RealTimeSubPayload{
					ClientID:      clientID,
					TargetResults: results,
				},
			})
			return
		}

		c.JSON(http.StatusOK, network.SuccessResponse{
			Code: http.StatusOK,
			Msg:  "success",
			Payload: network.RealTimeSubPayload{
				ClientID:      clientID,
				TargetResults: results,
			},
		})
		return
	case constants.SubStopAction:
		results, err := globalSubState.RemoveTargets(c, clientID, request.Measurements)
		if err != nil {
			logger.Error(c, "remove target to real time data subscription config failed", "error", err)
			c.JSON(http.StatusOK, network.FailureResponse{
				Code: http.StatusBadRequest,
				Msg:  err.Error(),
				Payload: network.RealTimeSubPayload{
					ClientID:      clientID,
					TargetResults: results,
				},
			})
			return
		}

		c.JSON(http.StatusOK, network.SuccessResponse{
			Code: http.StatusOK,
			Msg:  "success",
			Payload: network.RealTimeSubPayload{
				ClientID:      clientID,
				TargetResults: results,
			},
		})
		return
	case constants.SubAppendAction:
		results, err := globalSubState.AppendTargets(c, tx, clientID, request.Measurements)
		if err != nil {
			logger.Error(c, "append target to real time data subscription config failed", "error", err)
			c.JSON(http.StatusOK, network.FailureResponse{
				Code: http.StatusBadRequest,
				Msg:  err.Error(),
				Payload: network.RealTimeSubPayload{
					ClientID:      clientID,
					TargetResults: results,
				},
			})
			return
		}

		c.JSON(http.StatusOK, network.SuccessResponse{
			Code: http.StatusOK,
			Msg:  "success",
			Payload: network.RealTimeSubPayload{
				ClientID:      clientID,
				TargetResults: results,
			},
		})
		return
	case constants.SubUpdateAction:
		results, err := globalSubState.UpdateTargets(c, tx, clientID, request.Measurements)
		if err != nil {
			logger.Error(c, "update target to real time data subscription config failed", "error", err)
			c.JSON(http.StatusOK, network.FailureResponse{
				Code: http.StatusBadRequest,
				Msg:  err.Error(),
				Payload: network.RealTimeSubPayload{
					ClientID:      clientID,
					TargetResults: results,
				},
			})
			return
		}

		c.JSON(http.StatusOK, network.SuccessResponse{
			Code: http.StatusOK,
			Msg:  "success",
			Payload: network.RealTimeSubPayload{
				ClientID:      clientID,
				TargetResults: results,
			},
		})
		return
	default:
		err := fmt.Errorf("%w: request action is %s", constants.ErrUnsupportedAction, request.Action)
		logger.Error(c, "unsupported action of real time data subscription request", "error", err)
		requestTargetsCount := processRealTimeRequestCount(request.Measurements)
		results := processRealTimeRequestTargets(request.Measurements, requestTargetsCount, err)
		c.JSON(http.StatusOK, network.FailureResponse{
			Code: http.StatusBadRequest,
			Msg:  err.Error(),
			Payload: network.RealTimeSubPayload{
				ClientID:      clientID,
				TargetResults: results,
			},
		})
		return
	}
}

// RealTimeSubMeasurement define struct of real time subscription measurement
type RealTimeSubMeasurement struct {
	targets []string
}

// TargetPollingContext define struct of real time pulling reverse context
type TargetPollingContext struct {
	interval    string
	measurement *orm.Measurement
}

// RealTimeSubConfig define struct of real time subscription config
type RealTimeSubConfig struct {
	noticeChan    chan *transportTargets
	mutex         sync.RWMutex
	measurements  map[string][]string
	targetContext map[string]*TargetPollingContext
}

// SharedSubState define struct of shared subscription state with mutex
type SharedSubState struct {
	subMap      map[string]*RealTimeSubConfig
	globalMutex sync.RWMutex
}

// NewSharedSubState define function to create new SharedSubState
func NewSharedSubState() *SharedSubState {
	return &SharedSubState{
		subMap: make(map[string]*RealTimeSubConfig),
	}
}

// processAndValidateTargetsForStart define func to perform all database I/O operations in a lock-free state for start action
func processAndValidateTargetsForStart(ctx context.Context, tx *gorm.DB, measurements []network.RealTimeMeasurementItem, allReqTargetNum int) (
	[]network.TargetResult, []string,
	map[string][]string,
	map[string]*TargetPollingContext,
) {
	targetProcessResults := make([]network.TargetResult, 0, allReqTargetNum)
	newMeasMap := make(map[string][]string)
	successfulTargets := make([]string, 0, allReqTargetNum)
	newMeasContextMap := make(map[string]*TargetPollingContext)

	for _, measurementItem := range measurements {
		interval := measurementItem.Interval
		for _, target := range measurementItem.Targets {
			var targetResult network.TargetResult
			targetResult.ID = target
			targetModel, err := database.ParseDataIdentifierToken(ctx, tx, target)
			if err != nil {
				logger.Error(ctx, "parse data indentity token failed", "error", err, "identity_token", target)
				targetResult.Code = constants.SubFailedCode
				targetResult.Msg = fmt.Sprintf("%s: %s", constants.SubFailedMsg, err.Error())
				targetProcessResults = append(targetProcessResults, targetResult)
				continue
			}
			targetResult.Code = constants.SubSuccessCode
			targetResult.Msg = constants.SubSuccessMsg
			targetProcessResults = append(targetProcessResults, targetResult)
			successfulTargets = append(successfulTargets, target)

			if _, exists := newMeasMap[interval]; !exists {
				newMeasMap[interval] = make([]string, 0, len(measurementItem.Targets))
			}

			meas := newMeasMap[interval]
			meas = append(meas, target)
			newMeasMap[interval] = meas
			newMeasContextMap[target] = &TargetPollingContext{
				interval:    interval,
				measurement: targetModel.GetMeasurementInfo(),
			}
		}
	}
	return targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap
}

// processAndValidateTargetsForUpdate define func to perform all database I/O operations in a lock-free state for update action
func processAndValidateTargetsForUpdate(ctx context.Context, tx *gorm.DB, config *RealTimeSubConfig, measurements []network.RealTimeMeasurementItem, allReqTargetNum int) (
	[]network.TargetResult, []string,
	map[string][]string,
	map[string]*TargetPollingContext,
) {
	targetProcessResults := make([]network.TargetResult, 0, allReqTargetNum)
	newMeasMap := make(map[string][]string)
	successfulTargets := make([]string, 0, allReqTargetNum)
	newMeasContextMap := make(map[string]*TargetPollingContext)

	for _, measurementItem := range measurements {
		interval := measurementItem.Interval
		for _, target := range measurementItem.Targets {
			targetResult := network.TargetResult{ID: target}
			if _, exist := config.targetContext[target]; !exist {
				err := fmt.Errorf("target %s does not exists in subscription list", target)
				logger.Error(ctx, "update target does not exist in subscription list", "error", err, "target", target)
				targetResult.Code = constants.UpdateSubFailedCode
				targetResult.Msg = fmt.Sprintf("%s: %s", constants.UpdateSubFailedMsg, err.Error())
				targetProcessResults = append(targetProcessResults, targetResult)
				continue
			}

			targetModel, err := database.ParseDataIdentifierToken(ctx, tx, target)
			if err != nil {
				logger.Error(ctx, "parse data indentity token failed", "error", err, "identity_token", target)
				targetResult.Code = constants.UpdateSubFailedCode
				targetResult.Msg = fmt.Sprintf("%s: %s", constants.UpdateSubFailedMsg, err.Error())
				targetProcessResults = append(targetProcessResults, targetResult)
				continue
			}

			targetResult.Code = constants.UpdateSubSuccessCode
			targetResult.Msg = constants.UpdateSubSuccessMsg
			targetProcessResults = append(targetProcessResults, targetResult)
			successfulTargets = append(successfulTargets, target)

			if _, exists := newMeasMap[interval]; !exists {
				newMeasMap[interval] = make([]string, 0, len(measurementItem.Targets))
			}

			meas := newMeasMap[interval]
			meas = append(meas, target)
			newMeasMap[interval] = meas
			newMeasContextMap[target] = &TargetPollingContext{
				interval:    interval,
				measurement: targetModel.GetMeasurementInfo(),
			}
		}
	}
	return targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap
}

// mergeMeasurementsForStart define func to merge newMeasurementsMap into existingMeasurementsMap for start action
func mergeMeasurementsForStart(config *RealTimeSubConfig, newMeasurements map[string][]string, newMeasurementsContextMap map[string]*TargetPollingContext) []string {
	allDuplicates := make([]string, 0)
	for interval, newMeas := range newMeasurements {
		if existingMeas, exists := config.measurements[interval]; exists {
			// deduplication operations prevent duplicate subscriptions to the same measurement node
			deduplicated, duplicates := util.DeduplicateAndReportDuplicates(existingMeas, newMeas)

			if len(duplicates) > 0 {
				for _, duplicate := range duplicates {
					delete(newMeasurementsContextMap, duplicate)
				}
				allDuplicates = append(allDuplicates, duplicates...)
			}

			if len(deduplicated) > 0 {
				existingMeas = append(existingMeas, deduplicated...)
				config.measurements[interval] = existingMeas
				maps.Copy(config.targetContext, newMeasurementsContextMap)
			}
		}
	}
	return allDuplicates
}

// mergeMeasurementsForUpdate define func to merge newMeasurementsMap into existingMeasurementsMap for update action
func mergeMeasurementsForUpdate(config *RealTimeSubConfig, newMeasurements map[string][]string, newMeasurementsContextMap map[string]*TargetPollingContext) ([]string, error) {
	allDuplicates := make([]string, 0)
	delMeasMap := make(map[string][]string)
	for _, newMeas := range newMeasurements {
		for _, measurement := range newMeas {
			oldInterval := config.targetContext[measurement].interval
			if _, exists := delMeasMap[oldInterval]; !exists {
				delMeasurements := []string{measurement}
				delMeasMap[oldInterval] = delMeasurements
			} else {
				delMeasurements := delMeasMap[oldInterval]
				delMeasurements = append(delMeasurements, measurement)
				delMeasMap[oldInterval] = delMeasurements
			}
		}
	}

	for interval, delMeas := range delMeasMap {
		existingMeas, exist := config.measurements[interval]
		if !exist {
			return nil, fmt.Errorf("can not find exist measurements in %s interval", interval)
		}

		measurements := util.RemoveTargetsFromSliceSimple(existingMeas, delMeas)
		config.measurements[interval] = measurements
	}

	for interval, newMeas := range newMeasurements {
		if existingMeas, exists := config.measurements[interval]; exists {
			deduplicated, duplicates := util.DeduplicateAndReportDuplicates(existingMeas, newMeas)

			if len(duplicates) > 0 {
				for _, duplicate := range duplicates {
					delete(newMeasurementsContextMap, duplicate)
				}
				allDuplicates = append(allDuplicates, duplicates...)
			}

			if len(deduplicated) > 0 {
				existingMeas = append(existingMeas, deduplicated...)
				config.measurements[interval] = existingMeas
				maps.Copy(config.targetContext, newMeasurementsContextMap)
			}
		}
	}
	return allDuplicates, nil
}

// CreateConfig define function to create config in SharedSubState
func (s *SharedSubState) CreateConfig(ctx context.Context, tx *gorm.DB, clientID string, measurements []network.RealTimeMeasurementItem) ([]network.TargetResult, error) {
	requestTargetsCount := processRealTimeRequestCount(measurements)
	targetProcessResults, _, newMeasurementsMap, measurementContexts := processAndValidateTargetsForStart(ctx, tx, measurements, requestTargetsCount)
	s.globalMutex.Lock()
	if _, exist := s.subMap[clientID]; exist {
		s.globalMutex.Unlock()
		err := fmt.Errorf("clientID %s already exists. use AppendTargets to modify existing config", clientID)
		logger.Error(ctx, "clientID already exists. use AppendTargets to modify existing config", "error", err)
		return targetProcessResults, err
	}

	config := &RealTimeSubConfig{
		noticeChan:    make(chan *transportTargets, constants.NoticeChanCap),
		measurements:  newMeasurementsMap,
		targetContext: measurementContexts,
	}
	s.subMap[clientID] = config
	s.globalMutex.Unlock()
	return targetProcessResults, nil
}

// AppendTargets define function to append targets in SharedSubState
func (s *SharedSubState) AppendTargets(ctx context.Context, tx *gorm.DB, clientID string, measurements []network.RealTimeMeasurementItem) ([]network.TargetResult, error) {
	requestTargetsCount := processRealTimeRequestCount(measurements)

	s.globalMutex.RLock()
	config, exist := s.subMap[clientID]
	s.globalMutex.RUnlock()

	if !exist {
		err := fmt.Errorf("clientID %s not found. use CreateConfig to start a new config", clientID)
		logger.Error(ctx, "clientID not found. use CreateConfig to start a new config", "error", err)
		return processRealTimeRequestTargets(measurements, requestTargetsCount, err), err
	}

	targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap := processAndValidateTargetsForStart(ctx, tx, measurements, requestTargetsCount)

	config.mutex.Lock()
	allDuplicates := mergeMeasurementsForStart(config, newMeasMap, newMeasContextMap)
	if len(allDuplicates) > 0 {
		logger.Warn(ctx, "some targets are duplicate and have been ignored in append operation", "clientID", clientID, "duplicates", allDuplicates)
		// process repeat target in targetProcessResults and successfulTargets
		targetProcessResults, successfulTargets = filterAndDeduplicateRepeatTargets(targetProcessResults, successfulTargets, allDuplicates)
	}
	config.mutex.Unlock()

	if len(successfulTargets) > 0 {
		transportTargets := &transportTargets{
			OperationType: constants.OpAppend,
			Targets:       successfulTargets,
		}
		config.noticeChan <- transportTargets
	}

	return targetProcessResults, nil
}

func filterAndDeduplicateRepeatTargets(resultsSlice []network.TargetResult, idsSlice []string, duplicates []string) ([]network.TargetResult, []string) {
	filteredIDs := make([]string, 0, len(idsSlice))
	set := make(map[string]struct{}, len(duplicates))
	for _, duplicate := range duplicates {
		set[duplicate] = struct{}{}
	}

	for index := range resultsSlice {
		if _, isTarget := set[resultsSlice[index].ID]; isTarget {
			resultsSlice[index].Code = constants.SubRepeatCode
			resultsSlice[index].Msg = constants.SubRepeatMsg
		}
	}

	for _, id := range idsSlice {
		if _, isTarget := set[id]; !isTarget {
			filteredIDs = append(filteredIDs, id)
		}
	}

	return resultsSlice, filteredIDs
}

// UpsertTargets define function to upsert targets in SharedSubState
func (s *SharedSubState) UpsertTargets(ctx context.Context, tx *gorm.DB, clientID string, measurements []network.RealTimeMeasurementItem) ([]network.TargetResult, error) {
	requestTargetsCount := processRealTimeRequestCount(measurements)
	targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap := processAndValidateTargetsForStart(ctx, tx, measurements, requestTargetsCount)

	s.globalMutex.RLock()
	config, exist := s.subMap[clientID]
	s.globalMutex.RUnlock()

	var opType constants.TargetOperationType
	if exist {
		opType = constants.OpUpdate
		config.mutex.Lock()
		mergeMeasurementsForStart(config, newMeasMap, newMeasContextMap)
		config.mutex.Unlock()
	} else {
		opType = constants.OpAppend
		s.globalMutex.Lock()
		if config, exist = s.subMap[clientID]; !exist {
			config = &RealTimeSubConfig{
				noticeChan:   make(chan *transportTargets, constants.NoticeChanCap),
				measurements: newMeasMap,
			}
			s.subMap[clientID] = config
		} else {
			s.globalMutex.Unlock()
			config.mutex.Lock()
			mergeMeasurementsForStart(config, newMeasMap, newMeasContextMap)
			config.mutex.Unlock()
		}
		s.globalMutex.Unlock()
	}

	if len(successfulTargets) > 0 {
		transportTargets := &transportTargets{
			OperationType: opType,
			Targets:       successfulTargets,
		}
		config.noticeChan <- transportTargets
	}
	return targetProcessResults, nil
}

// RemoveTargets define function to remove targets in SharedSubState
func (s *SharedSubState) RemoveTargets(ctx context.Context, clientID string, measurements []network.RealTimeMeasurementItem) ([]network.TargetResult, error) {
	requestTargetsCount := processRealTimeRequestCount(measurements)
	targetProcessResults := make([]network.TargetResult, 0, requestTargetsCount)

	s.globalMutex.RLock()
	config, exist := s.subMap[clientID]
	if !exist {
		s.globalMutex.RUnlock()
		err := fmt.Errorf("clientID %s not found", clientID)
		logger.Error(ctx, "clientID not found in remove targets operation", "error", err)
		return processRealTimeRequestTargets(measurements, requestTargetsCount, err), err
	}
	s.globalMutex.RUnlock()

	var shouldRemoveClient bool
	// measurements is the list of items to be removed passed in the request
	transportTargets := &transportTargets{
		OperationType: constants.OpRemove,
		Targets:       make([]string, 0, requestTargetsCount),
	}
	config.mutex.Lock()
	for _, measurement := range measurements {
		interval := measurement.Interval
		// meas is the locally running listener configuration
		measTargets, measExist := config.measurements[interval]
		if !measExist {
			logger.Error(ctx, fmt.Sprintf("measurement with interval %s not found under clientID %s", interval, clientID), "clientID", clientID, "interval", interval)
			for _, target := range measTargets {
				targetResult := network.TargetResult{
					ID:   target,
					Code: constants.CancelSubFailedCode,
					Msg:  constants.CancelSubFailedMsg,
				}
				targetProcessResults = append(targetProcessResults, targetResult)
			}
			continue
		}

		targetsToRemoveMap := make(map[string]struct{})
		for _, target := range measurement.Targets {
			targetsToRemoveMap[target] = struct{}{}
		}

		var newTargets []string
		for _, existingTarget := range measTargets {
			if _, found := targetsToRemoveMap[existingTarget]; !found {
				newTargets = append(newTargets, existingTarget)
			} else {
				transportTargets.Targets = append(transportTargets.Targets, existingTarget)
				targetResult := network.TargetResult{
					ID:   existingTarget,
					Code: constants.CancelSubSuccessCode,
					Msg:  constants.CancelSubSuccessMsg,
				}
				targetProcessResults = append(targetProcessResults, targetResult)
				delete(targetsToRemoveMap, existingTarget)
				delete(config.targetContext, existingTarget)
			}
		}
		measTargets = newTargets

		if len(measTargets) == 0 {
			delete(config.measurements, interval)
		}

		if len(config.measurements) == 0 {
			shouldRemoveClient = true
		}

		if len(targetsToRemoveMap) > 0 {
			err := fmt.Errorf("target remove were not found under clientID %s and interval %s", clientID, interval)
			for target := range targetsToRemoveMap {
				targetResult := network.TargetResult{
					ID:   target,
					Code: constants.CancelSubFailedCode,
					Msg:  fmt.Sprintf("%s: %s", constants.SubFailedMsg, err.Error()),
				}
				targetProcessResults = append(targetProcessResults, targetResult)
			}
		}
	}
	config.mutex.Unlock()
	// pass the removed subscription configuration to the notice channel
	config.noticeChan <- transportTargets

	if shouldRemoveClient {
		s.globalMutex.Lock()
		if currentConfig, exist := s.subMap[clientID]; exist && len(currentConfig.measurements) == 0 {
			delete(s.subMap, clientID)
		}
		s.globalMutex.Unlock()
	}
	return targetProcessResults, nil
}

// UpdateTargets define function to update targets in SharedSubState
func (s *SharedSubState) UpdateTargets(ctx context.Context, tx *gorm.DB, clientID string, measurements []network.RealTimeMeasurementItem) ([]network.TargetResult, error) {
	requestTargetsCount := processRealTimeRequestCount(measurements)
	targetProcessResults := make([]network.TargetResult, 0, requestTargetsCount)

	s.globalMutex.RLock()
	config, exist := s.subMap[clientID]
	s.globalMutex.RUnlock()

	if !exist {
		s.globalMutex.RUnlock()
		err := fmt.Errorf("clientID %s not found", clientID)
		logger.Error(ctx, "clientID not found in remove targets operation", "error", err)
		return processRealTimeRequestTargets(measurements, requestTargetsCount, err), err
	}

	targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap := processAndValidateTargetsForUpdate(ctx, tx, config, measurements, requestTargetsCount)

	config.mutex.Lock()
	allDuplicates, err := mergeMeasurementsForUpdate(config, newMeasMap, newMeasContextMap)
	if err != nil {
		logger.Warn(ctx, "can not find exist measurements in target interval", "clientID", clientID, "duplicates", allDuplicates, "error", err)
	}

	if len(allDuplicates) > 0 {
		logger.Warn(ctx, "some targets are duplicate and have been ignored in append operation", "clientID", clientID, "duplicates", allDuplicates)
		// process repeat target in targetProcessResults and successfulTargets
		targetProcessResults, successfulTargets = filterAndDeduplicateRepeatTargets(targetProcessResults, successfulTargets, allDuplicates)
	}
	config.mutex.Unlock()

	if len(successfulTargets) > 0 {
		transportTargets := &transportTargets{
			OperationType: constants.OpUpdate,
			Targets:       successfulTargets,
		}
		config.noticeChan <- transportTargets
	}

	return targetProcessResults, nil
}

// Get define function to get subscriptions config from SharedSubState
func (s *SharedSubState) Get(clientID string) (*RealTimeSubConfig, bool) {
	s.globalMutex.RLock()
	defer s.globalMutex.RUnlock()

	config, exists := s.subMap[clientID]
	if !exists {
		return nil, false
	}
	return config, true
}

func processRealTimeRequestCount(measurements []network.RealTimeMeasurementItem) int {
	totalTargetsCount := 0
	for _, measItem := range measurements {
		totalTargetsCount += len(measItem.Targets)
	}
	return totalTargetsCount
}

func processRealTimeRequestTargets(measurements []network.RealTimeMeasurementItem, targetCount int, err error) []network.TargetResult {
	targetProcessResults := make([]network.TargetResult, 0, targetCount)
	for _, measurementItem := range measurements {
		for _, target := range measurementItem.Targets {
			var targetResult network.TargetResult
			targetResult.ID = target
			targetResult.Code = constants.SubFailedCode
			targetResult.Msg = fmt.Sprintf("%s: %s", constants.SubFailedMsg, err.Error())
			targetProcessResults = append(targetProcessResults, targetResult)
		}
	}
	return targetProcessResults
}

// transportTargets define struct to transport update or remove target to real
// time pull api
type transportTargets struct {
	OperationType constants.TargetOperationType
	Targets       []string
}