golib/infra/svc/cache.go

package svc

import (
	"sync"
	"time"

	"golib/features/mo"
	"golib/infra/ii"
)

// Cache 数据库缓存
// 缓存被设计为将写入频率较低且读取频率较高的数据库表预加载至内存中, 在关联查询时使用缓存数据进行匹配
type Cache struct {
	items    ii.Items
	itemIdx  int
	itemName []string
	dataIdx  []map[string]map[mo.ObjectID]int
	data     [][]mo.M
}

// Include 检查 ii.Lookup.From 是否需要存在缓存
func (c *Cache) Include(itemName string) (int, bool) {
	for i, name := range c.itemName {
		if itemName == name {
			_, ok := c.items[itemName]
			return i, ok
		}
	}
	return 0, false
}

// AddItem 增加 itemName 缓存
func (c *Cache) AddItem(itemName string) {
	for _, oldName := range c.itemName {
		if oldName == itemName {
			return
		}
	}
	if _, ok := c.items.Has(itemName); !ok {
		return
	}
	c.itemName[c.itemIdx] = itemName
	c.itemIdx++
}

// SetData 设置 data 作为 itemName 的缓存数据
func (c *Cache) SetData(itemName string, data []mo.M) {
	for i, oldName := range c.itemName {
		if oldName != itemName {
			continue // 如果未预设置 itemName 则无法设置缓存数据
		}
		itemInfo, ok := c.items[itemName]
		if !ok {
			panic(ok)
		}

		idxMap := make(map[string]map[mo.ObjectID]int, len(data))
		// 由于 _id 不在 XML 内, 所以此处单独初始化 _id 作为索引
		oidIdx := make(map[mo.ObjectID]int)
		for n, row := range data {
			oidIdx[row[mo.ID.Key()].(mo.ObjectID)] = n
		}
		idxMap[mo.ID.Key()] = oidIdx
		// XML 索引
		for _, field := range itemInfo.Fields {
			if field.Name == mo.ID.Key() {
				continue // 由于上方默认使用以 _id 作为索引, 所以当 XML 存在 _id 字段时跳过, 防止重复设置
			}
			if field.Type != mo.TypeObjectId {
				continue // 仅为数据类型是 ObjectID 的字段创建索引
			}
			idx := make(map[mo.ObjectID]int)
			for j, row := range data {
				if fieldValue, o := row[field.Name].(mo.ObjectID); o {
					idx[fieldValue] = j
				}
			}
			idxMap[field.Name] = idx
		}
		c.dataIdx[i] = idxMap
		c.data[i] = data
	}
}

func (c *Cache) GetData(itemName string) (map[string]map[mo.ObjectID]int, []mo.M) {
	for i, oldName := range c.itemName {
		if oldName == itemName {
			return c.dataIdx[i], c.data[i]
		}
	}
	return nil, nil
}

func (c *Cache) SpitPipe(itemInfo *ii.ItemInfo, pipe mo.Pipeline) (stage mo.Pipeline, lookup []ii.Lookup) {
	for _, p := range pipe {
		if look, o := c.hasLookup(itemInfo, p); o {
			lookup = append(lookup, look)
			continue
		}
		stage = append(stage, p)
	}
	return
}

func (c *Cache) sum(cacheMap []mo.M, lv any, look ii.Lookup) mo.A {
	var sum float64                     // 数据类型始终为 float64
	for _, cacheRow := range cacheMap { // 循环缓存列表
		fv := cacheRow[look.ForeignField]
		if lv == fv { // 本地值与远程值相等时
			switch n := cacheRow[look.SUM].(type) { // 累加字段数量
			case float64:
				sum += n
			case int64:
				sum += float64(n)
			}
		}
	}
	return mo.A{mo.M{look.SUM: sum}}
}

func (c *Cache) Format(itemInfo *ii.ItemInfo, lookup []ii.Lookup, rows *[]mo.M) time.Duration {
	t := time.Now()
	var group sync.WaitGroup
	group.Add(len(*rows))
	for i := 0; i < len(*rows); i++ {
		go func(group *sync.WaitGroup, i int) {
			for _, look := range lookup {
				itemLookName := itemInfo.ForkName(look.From)
				cacheIdx, cacheMap := c.GetData(itemLookName)

				localValue := (*rows)[i][look.LocalField]
				idxMap := cacheIdx[look.ForeignField]

				// 由于设置缓存时规定了类型必须为 ObjectID, 所以此处可以直接断言
				fv, ok := idxMap[localValue.(mo.ObjectID)]
				if !ok {
					continue // 如果本地数据无法在索引中找到则跳过
				}
				// 未开启列表且 SUM 不为空时表示合计数量
				if !look.List && look.SUM != "" {
					// 当 Look.Form 表中包含 Look.SUM 字段时才进行合计
					if _, ok = c.items[itemLookName].FieldMap[look.SUM]; ok {
						(*rows)[i][look.AS] = c.sum(cacheMap, localValue, look)
					}
					continue
				}
				(*rows)[i][look.AS] = mo.A{cacheMap[fv]}
			}
			group.Done()
		}(&group, i)
	}
	group.Wait()
	return time.Now().Sub(t)
}

// hasLookup 解析 d 是否为 ii.Lookup
func (c *Cache) hasLookup(itemInfo *ii.ItemInfo, d mo.D) (ii.Lookup, bool) {
	if len(d) == 0 {
		return ii.Lookup{}, false
	}
	if d[0].Key != mo.PsLookup {
		return ii.Lookup{}, false
	}
	valueMap := d[0].Value.(mo.D).Map()
	field, ok := itemInfo.Field(valueMap["localField"].(string))
	if !ok {
		return ii.Lookup{}, false
	}
	lookup, ok := field.HasLookup(valueMap["as"].(string))
	if !ok {
		return ii.Lookup{}, false
	}
	lookup.LocalField = field.Name
	localField := itemInfo.Name.Database() + "." + lookup.From
	if _, ok = c.Include(localField); !ok {
		return ii.Lookup{}, false
	}
	return *lookup, true
}

const (
	maxCacheTblSize = 128
)

func NewCache(items ii.Items) *Cache {
	c := new(Cache)
	c.itemName = make([]string, maxCacheTblSize)
	c.dataIdx = make([]map[string]map[mo.ObjectID]int, maxCacheTblSize)
	c.data = make([][]mo.M, maxCacheTblSize)
	c.items = items
	return c
}