golib/infra/ii/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

	mutex sync.Mutex
}

// 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) {
	c.mutex.Lock()
	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
	}
	c.mutex.Unlock()
}

// getData 从缓存中调出数据, 返回的 map 必须只读
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) deepCopy(lField *ii.FieldInfo, lookItem *ii.ItemInfo, cacheRow mo.M) mo.M {
	m := make(mo.M, len(lField.Fields))
	for _, sub := range lField.Fields {
		field, ok := lookItem.Field(sub.Name)
		if !ok {
			continue
		}
		sv := cacheRow[sub.Name]
		switch field.Type {
		case mo.TypeObject:
			svv, ok := sv.(mo.M)
			if !ok {
				m[field.Name] = sv
			} else {
				dm, err := mo.DeepMapCopy(svv)
				if err == nil {
					m[field.Name] = dm
				} else {
					m[field.Name] = sv
				}
			}
		case mo.TypeArray:
			if field.Items == ii.FieldItemsObject {
				svv, o := sv.(mo.A)
				if !o {
					m[field.Name] = sv
				} else {
					svList := make(mo.A, len(svv))
					for i, row := range svv {
						sr, ok := row.(mo.M)
						if !ok {
							svList[i] = row
						} else {
							r, err := mo.DeepMapCopy(sr)
							if err == nil {
								svList[i] = r
							} else {
								svList[i] = row
							}
						}
					}
					m[field.Name] = svList
				}
				continue
			}
			fallthrough
		default:
			m[sub.Name] = cacheRow[sub.Name]
		}
	}
	return m
}

func (c *Cache) handleList(topItem *ii.ItemInfo, look *ii.Lookup, cacheList []mo.M, lv any) mo.A {
	field, ok := topItem.Field(look.LocalField)
	if !ok {
		return mo.A{}
	}

	lookItem, _ := c.items[topItem.ForkName(look.From)]

	// 先获取索引
	idxList := make([]int, 0)
	for i, row := range cacheList { // 循环缓存列表
		fv := row[look.ForeignField]
		if lv == fv { // 本地值与远程值相等时
			idxList = append(idxList, i)
		}
	}

	// 根据索引分配大小
	list := make(mo.A, len(idxList))

	var group sync.WaitGroup
	group.Add(len(idxList))

	for i := 0; i < len(idxList); i++ {
		go func(group *sync.WaitGroup, i int) {
			list[i] = c.deepCopy(&field, &lookItem, cacheList[idxList[i]])
			group.Done()
		}(&group, i)
	}

	group.Wait()
	return list
}

func (c *Cache) handleOne(topItem *ii.ItemInfo, look *ii.Lookup, cacheRow mo.M) mo.A {
	field, ok := topItem.Field(look.LocalField)
	if !ok {
		return mo.A{}
	}
	lookItem, _ := c.items[topItem.ForkName(look.From)]
	return mo.A{c.deepCopy(&field, &lookItem, cacheRow)}
}

func (c *Cache) handleSUM(cacheList []mo.M, lv any, look ii.Lookup) mo.A {
	var sum float64                      // 数据类型始终为 float64
	for _, cacheRow := range cacheList { // 循环缓存列表
		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, cacheList := c.getData(itemLookName)

				localValue, ok := (*rows)[i][look.LocalField]
				if !ok {
					continue // 可能会存在某一条文档不存在这个字段的现象
				}
				idxMap := cacheIdx[look.ForeignField]

				if look.List {
					(*rows)[i][look.AS] = c.handleList(itemInfo, &look, cacheList, localValue)
					continue
				}

				if look.SUM != "" { // SUM 不为空时表示合计数量
					// 当 Look.Form 的 ItemInfo 中包含 Look.SUM 字段时才进行合计
					if _, ok := c.items[itemLookName].FieldMap[look.SUM]; ok {
						(*rows)[i][look.AS] = c.handleSUM(cacheList, localValue, look)
					}
				} else {
					// 由于设置缓存时规定了类型必须为 ObjectID, 所以此处可以直接断言
					idx, ok := idxMap[localValue.(mo.ObjectID)]
					if !ok {
						continue // 如果本地数据无法在索引中找到则跳过
					}
					(*rows)[i][look.AS] = c.handleOne(itemInfo, &look, cacheList[idx])
				}
			}
			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 := mo.Convert.M(d[0].Value.(mo.D))
	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
	if _, ok = c.Include(itemInfo.ForkName(lookup.From)); !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
}