iotlcm/internal/logic/channelconfig/channel_config.go

package channelconfig

import (
	"context"
	"database/sql"
	"strings"

	"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model"
	"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/model/entity"
	"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/service"

	"git.5iot.com/iotlcm-public/iotlcm-upstream-proxy/internal/dao"
	"sync"
	"time"

	"github.com/gogf/gf/v2/crypto/gmd5"
	"github.com/gogf/gf/v2/encoding/gjson"
	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/frame/g"
)

// channelConfigRow extends the generated entity without losing SQL NULL values.
type channelConfigRow struct {
	entity.TSimChannels
	IsNormal *int `json:"isNormal" orm:"is_normal"`
}

type sChannelConfig struct {
	sync.RWMutex
	channelCache map[string]*model.SimChannelConfigItemCache
	loadVersion  uint8
	chgLogMaxId  int64
	running      bool
}

func NewChannelConfig() *sChannelConfig {
	return &sChannelConfig{
		channelCache: make(map[string]*model.SimChannelConfigItemCache),
	}
}

func init() {
	service.RegisterChannelConfig(NewChannelConfig())
}

func (s *sChannelConfig) Start(ctx context.Context) error {
	err := s.LoadAllChannelConfig(ctx)
	if err != nil {
		g.Log().Errorf(ctx, "ChannelConfig Start err %+v", err)
		return err
	}
	s.running = true
	err = s.Loop(ctx)
	if err != nil {
		g.Log().Errorf(ctx, "ChannelConfig Start err %+v", err)
		return err
	}
	return nil
}

func (s *sChannelConfig) Stop(ctx context.Context) error {
	s.running = false
	return nil
}

func (s *sChannelConfig) Loop(ctx context.Context) error {
	g.Log().Infof(ctx, "ChannelConfig loop begin.")
	lastLoadAllChannelConfigTime := int64(0)
	go func() {
		for s.running {
			now := time.Now().Unix()
			loadAllInterval := g.Cfg().MustGet(ctx, "channel.load_all_interval", 2*60*60).Int64()
			if now-lastLoadAllChannelConfigTime >= loadAllInterval {
				lastLoadAllChannelConfigTime = now
				s.LoadAllChannelConfig(ctx)
			}
			time.Sleep(1 * time.Second)
		}
	}()
	g.Log().Infof(ctx, "ChannelConfig loop background goroutine started.")
	return nil
}

func (s *sChannelConfig) LoadAllChannelConfig(ctx context.Context) error {
	s.loadVersion++
	const warnRowThreshold = 1000

	dbSimChannels := make([]*channelConfigRow, 0)
	err := dao.TSimChannels.Ctx(ctx).
		OrderAsc(dao.TSimChannels.Columns().Id).
		Scan(&dbSimChannels)
	if err != nil && err != sql.ErrNoRows {
		return err
	}
	if len(dbSimChannels) >= warnRowThreshold {
		g.Log().Warningf(ctx, "LoadAllChannelConfig TSimChannels row count %d >= %d, confirm small-table assumption or add pagination",
			len(dbSimChannels), warnRowThreshold)
	}

	dbSimChannelTemplates := make([]*entity.TSimChannelTemplate, 0)
	err = dao.TSimChannelTemplate.Ctx(ctx).
		OrderAsc(dao.TSimChannelTemplate.Columns().Id).
		Scan(&dbSimChannelTemplates)
	if err != nil && err != sql.ErrNoRows {
		return err
	}
	if len(dbSimChannelTemplates) >= warnRowThreshold {
		g.Log().Warningf(ctx, "LoadAllChannelConfig TSimChannelTemplate row count %d >= %d, confirm small-table assumption or add pagination",
			len(dbSimChannelTemplates), warnRowThreshold)
	}

	templateByKey := make(map[string]*entity.TSimChannelTemplate, len(dbSimChannelTemplates))
	for _, tpl := range dbSimChannelTemplates {
		if tpl != nil {
			templateByKey[tpl.TempKey] = tpl
		}
	}

	s.Lock()
	for _, dbSimChannel := range dbSimChannels {
		if dbSimChannel == nil {
			continue
		}
		cache := &model.SimChannelConfigItemCache{
			Channel:  &dbSimChannel.TSimChannels,
			IsNormal: dbSimChannel.IsNormal,
			Version:  s.getChanVersion(ctx, dbSimChannel),
		}
		oldChannelCache, _ := s.channelCache[dbSimChannel.ChanKey]
		if cache.CanCreateInstance() {
			template, ok := templateByKey[dbSimChannel.TempKey]
			if !ok {
				g.Log().Warningf(ctx, "LoadAllChannelConfig no template for tempKey=%s chanKey=%s", dbSimChannel.TempKey, dbSimChannel.ChanKey)
				continue
			}
			cache.Template = template
			cache.Version += s.getTemplateVersion(ctx, template)
		}

		if oldChannelCache != nil && oldChannelCache.Version == cache.Version {
			continue
		}
		s.channelCache[dbSimChannel.ChanKey] = cache
	}
	s.Unlock()
	return nil
}

func (s *sChannelConfig) RefreshChannel(ctx context.Context, chanKey string) error {
	if chanKey == "" {
		return gerror.New("chanKey is required")
	}
	channel := new(channelConfigRow)
	if err := dao.TSimChannels.Ctx(ctx).Where(dao.TSimChannels.Columns().ChanKey, chanKey).Scan(channel); err != nil {
		return err
	}
	if channel.ChanKey == "" {
		return gerror.Newf("channel not found: %s", chanKey)
	}
	cache := &model.SimChannelConfigItemCache{
		Channel:  &channel.TSimChannels,
		IsNormal: channel.IsNormal,
		Version:  s.getChanVersion(ctx, channel),
	}
	// Ineligible channels refresh successfully even while their template is incomplete.
	if cache.CanCreateInstance() {
		template := new(entity.TSimChannelTemplate)
		if err := dao.TSimChannelTemplate.Ctx(ctx).Where(dao.TSimChannelTemplate.Columns().TempKey, channel.TempKey).Scan(template); err != nil {
			return err
		}
		if template.TempKey == "" {
			return gerror.Newf("channel template not found: %s", channel.TempKey)
		}
		cache.Template = template
		cache.Version += s.getTemplateVersion(ctx, template)
	}
	s.Lock()
	s.channelCache[chanKey] = cache
	s.Unlock()
	return nil
}

// RemoveChannel removes a channel configuration cache. Missing keys are
// intentionally treated as success so rename retries remain idempotent.
func (s *sChannelConfig) RemoveChannel(ctx context.Context, chanKey string) error {
	chanKey = strings.TrimSpace(chanKey)
	if chanKey == "" {
		return gerror.New("chanKey is required")
	}
	s.Lock()
	delete(s.channelCache, chanKey)
	s.Unlock()
	g.Log().Infof(ctx, "channel config removed, chanKey: %s", chanKey)
	return nil
}

// get alarm cache by id
func (s *sChannelConfig) GetChannel(chanKey string) (*model.SimChannelConfigItemCache, bool) {
	s.RLock()
	defer s.RUnlock()
	channelCache, ok := s.channelCache[chanKey]
	return channelCache, ok
}

func (s *sChannelConfig) GetAllChannel(ctx context.Context) ([]*model.SimChannelConfigItemCache, error) {
	s.RLock()
	defer s.RUnlock()
	channels := make([]*model.SimChannelConfigItemCache, 0)
	for _, channelCache := range s.channelCache {
		channels = append(channels, channelCache)
	}
	return channels, nil
}

func (s *sChannelConfig) getChanVersion(ctx context.Context, channel *channelConfigRow) string {
	chanKey := channel.ChanKey

	channelNew := &channelConfigRow{
		IsNormal: channel.IsNormal,
		TSimChannels: entity.TSimChannels{
			TempKey:      channel.TempKey,
			ChanKey:      channel.ChanKey,
			ChanName:     channel.ChanName,
			ChanConfig:   channel.ChanConfig,
			ChanStatus:   channel.ChanStatus,
			ChanAlias:    channel.ChanAlias,
			RegionCode:   channel.RegionCode,
			AppKey:       channel.AppKey,
			AppSecret:    channel.AppSecret,
			AccessToken:  channel.AccessToken,
			RefreshToken: channel.RefreshToken,
		},
	}
	chanVersion, err := gmd5.EncryptString(gjson.MustEncodeString(channelNew))
	if err != nil {
		g.Log().Errorf(ctx, "sChannelConfig.getChanVersion encrypt error %+v, chanKey %s", err, chanKey)
		return ""
	}
	g.Log().Debugf(ctx, "sChannelConfig.getChanVersion chanVersion=%s chanKey=%s", chanVersion, chanKey)
	return chanVersion
}

func (s *sChannelConfig) getTemplateVersion(ctx context.Context, template *entity.TSimChannelTemplate) string {
	templateKey := template.TempKey
	templateNew := &entity.TSimChannelTemplate{
		TempKey:    template.TempKey,
		TempName:   template.TempName,
		TempModel:  template.TempModel,
		TempConfig: template.TempConfig,
		TempStatus: template.TempStatus,
	}
	templateVersion, err := gmd5.EncryptString(gjson.MustEncodeString(templateNew))
	if err != nil {
		g.Log().Errorf(ctx, "sChannelConfig.getTemplateVersion encrypt error %+v, templateKey %s, templateNew %+v", err, templateKey, templateNew)
		return ""
	}
	g.Log().Debugf(ctx, "sChannelConfig.getTemplateVersion templateVersion=%s templateKey=%s", templateVersion, templateKey)
	return templateVersion
}