golib/infra/ii/svc/service.go

package svc

import (
	"context"
	"errors"
	"time"

	"git.simanc.com/software/golib/v4/features/mo"
	"git.simanc.com/software/golib/v4/infra/ii"
	"git.simanc.com/software/golib/v4/log"
)

var (
	ErrItemNotfound     = errors.New("item not found")
	ErrInternalError    = errors.New("internal error") // ErrInternalError 上游函数错误时返回
	ErrDataError        = errors.New("data error")     // ErrDataError 数据校验失败
	ErrDataTypeError    = errors.New("data type error")
	ErrPermissionDenied = errors.New("permission denied")
	ErrBindTypeError    = errors.New("bind type error")
	ErrNoDocuments      = mo.ErrNoDocuments
)

const (
	findLimitRows = 300
)

func IsErrItemNotFound(err error) bool     { return errors.Is(err, ErrItemNotfound) }
func IsErrInternalError(err error) bool    { return errors.Is(err, ErrInternalError) }
func IsErrDataError(err error) bool        { return errors.Is(err, ErrDataError) }
func IsErrPermissionDenied(err error) bool { return errors.Is(err, ErrPermissionDenied) }
func IsErrNoDocuments(err error) bool      { return errors.Is(err, ErrNoDocuments) }

type Service struct {
	Items   ii.Items
	Client  *mo.Client
	Log     log.Logger
	Cache   *Cache
	Context context.Context
	Timeout time.Duration

	refreshCh chan *ii.ItemInfo
}

func (s *Service) GetItems() ii.Items {
	return s.Items
}

func (s *Service) HasItem(name ii.Name) (*ii.ItemInfo, bool) {
	return s.Items.Has(name)
}

func (s *Service) Find(name ii.Name, filter mo.Filter) ([]Row, error) {
	var data []Row
	if err := s.FindWith(name, filter, nil, nil, 0, 0, &data); err != nil {
		return nil, err
	}
	return data, nil
}

// FindOne 查询一个文档
func (s *Service) FindOne(name ii.Name, filter mo.Filter) (Row, error) {
	var data mo.D
	if err := s.FindOneWith(name, filter, nil, nil, &data); err != nil {
		return Row{}, err
	}
	return s.toRow(data), nil
}

func (s *Service) FindAll(name ii.Name) ([]Row, error) {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.FindAll.%s: item not found", name)
		return nil, ErrItemNotfound
	}
	opts := mo.Options.Find()
	opts.SetSort(mo.NewSorter(ii.CreationTime, mo.SortDESC))
	ctx, cancel := s.newContext()
	defer cancel()
	cursor, err := s.openColl(info).Find(ctx, mo.D{}, opts)
	if err != nil {
		s.Log.Error("svc.FindAll.%s: %s", name, err)
		return nil, errors.Join(ErrInternalError, err)
	}
	var data []Row
	if err = mo.CursorDecodeAll(cursor, data); err != nil {
		s.Log.Error("svc.FindAll.%s: CursorDecodeAll: %s", name, err)
		return nil, errors.Join(ErrInternalError, err)
	}
	return data, nil
}

func (s *Service) FindWith(name ii.Name, filter, sort, project mo.Filter, skip, limit int64, v any) error {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.Find.%s: item not found", name)
		return ErrItemNotfound
	}
	if err := s.checkBindType(v); err != nil {
		s.Log.Error("svc.Find.%s: bind type: %T", name, v)
		return err
	}
	var query mo.D
	if filter != nil {
		query = filter.Done()
	}
	if err := info.QueryFilterCheck(query); err != nil {
		s.Log.Error("svc.Find.%s: QueryFilterCheck: %s. filter: %v", err, query)
		return errors.Join(ErrDataError, err)
	}
	opts := mo.Options.Find()
	if project != nil {
		opts.SetProjection(project.Done())
	}
	if sort != nil {
		opts.SetSort(sort.Done())
	} else {
		opts.SetSort(mo.NewSorter(ii.CreationTime, mo.SortDESC).Done())
	}
	if skip > 0 {
		opts.SetSkip(skip)
	}
	if limit > 0 {
		opts.SetLimit(limit)
	} else {
		if len(query) == 0 {
			opts.SetLimit(findLimitRows) // 如果没有过滤条件, 限制返回数量
		}
	}
	ctx, cancel := s.newContext()
	defer cancel()
	cursor, err := s.openColl(info).Find(ctx, query, opts)
	if err != nil {
		s.Log.Error("svc.Find.%s: %s filter: %v", name, err, query)
		return errors.Join(ErrInternalError, err)
	}
	if err = mo.CursorDecodeAll(cursor, v); err != nil {
		s.Log.Error("svc.Find.%s: CursorDecodeAll: %s", name, err)
		return errors.Join(ErrInternalError, err)
	}
	return nil
}

func (s *Service) FindOneWith(name ii.Name, filter, sort, project mo.Filter, v any) error {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.FindOne.%s: item not found", name)
		return ErrItemNotfound
	}
	if err := s.checkBindType(v); err != nil {
		s.Log.Error("svc.FindOne.%s: bind type: %T", name, v)
		return err
	}
	query := filter.Done()
	if err := info.QueryFilterCheck(query); err != nil {
		s.Log.Error("svc.FindOne.%s: QueryFilterCheck: %s filter: %v", name, err, query)
		return errors.Join(ErrDataError, err)
	}
	opts := mo.Options.FindOne()
	if project != nil {
		opts.SetProjection(project.Done())
	}
	if sort != nil {
		opts.SetSort(sort.Done())
	} else {
		opts.SetSort(mo.NewSorter(ii.CreationTime, mo.SortDESC).Done())
	}
	ctx, cancel := s.newContext()
	defer cancel()
	cursor := s.openColl(info).FindOne(ctx, query, opts)
	if err := cursor.Err(); err != nil {
		if errors.Is(err, mo.ErrNoDocuments) {
			return err
		}
		s.Log.Error("svc.FindOne.%s: %s filter: %v", name, err, query)
		return errors.Join(ErrInternalError, err)
	}
	if err := cursor.Decode(v); err != nil {
		s.Log.Error("svc.FindOne.%s: CursorDecode: %s", name, err)
		return errors.Join(ErrInternalError, err)
	}
	return nil
}

// FindOneAndDelete 查找并删除文档
func (s *Service) FindOneAndDelete(name ii.Name, filter mo.Filter) error {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.FindOneAndDelete.%s: item not found", name)
		return ErrItemNotfound
	}
	query := filter.Done()
	if err := info.QueryFilterCheck(query); err != nil {
		s.Log.Error("svc.FindOneAndDelete.%s: QueryFilterCheck: %s filter: %v", name, err, query)
		return errors.Join(ErrDataError, err)
	}
	ctx, cancel := s.newContext()
	defer cancel()
	result := s.openColl(info).FindOneAndDelete(ctx, query)
	if err := result.Err(); err != nil {
		if IsErrNoDocuments(err) {
			return err
		}
		s.Log.Error("svc.FindOneAndDelete.%s: %s filter: %v", name, err, query)
		return errors.Join(ErrInternalError, err)
	}
	s.Log.Debug("svc.FindOneAndDelete.%s: one document has been deleted. filter: %v", name, query)

	s.refreshCache(info)
	return nil
}

func (s *Service) DeleteOne(name ii.Name, filter mo.Filter) error {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.DeleteOne.%s: item not found", name)
		return ErrItemNotfound
	}
	query := filter.Done()
	ctx, cancel := s.newContext()
	defer cancel()
	result, err := s.openColl(info).DeleteOne(ctx, query)
	if err != nil {
		s.Log.Error("svc.DeleteOne.%s: %s filter: %v", name, err, query)
		return errors.Join(ErrInternalError, err)
	}
	s.Log.Debug("svc.DeleteOne.%s: %d document has been deleted. filter: %v", name, result.DeletedCount, query)

	s.refreshCache(info)
	return nil
}

func (s *Service) DeleteMany(name ii.Name, filter mo.Filter) error {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.DeleteMany.%s: item not found", name)
		return ErrItemNotfound
	}
	query := filter.Done()
	ctx, cancel := s.newContext()
	defer cancel()
	result, err := s.openColl(info).DeleteMany(ctx, query)
	if err != nil {
		s.Log.Error("svc.DeleteMany.%s: %s filter: %v", name, err, query)
		return errors.Join(ErrInternalError, err)
	}
	s.Log.Debug("svc.DeleteMany.%s: %d documents has been deleted. filter: %v", name, result.DeletedCount, query)

	s.refreshCache(info)
	return nil
}

// FindOneAndUpdate 查找并更新文档, 详情见 mo.SingleResult
func (s *Service) FindOneAndUpdate(name ii.Name, filter, updater mo.Filter) error {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.FindOneAndUpdate.%s: item not found", name)
		return ErrItemNotfound
	}
	query := filter.Done()
	if err := info.QueryFilterCheck(query); err != nil {
		s.Log.Error("svc.FindOneAndUpdate.%s: QueryFilterCheck: %s filter: %v", name, err, query)
		return errors.Join(ErrDataError, err)
	}
	update := updater.Done()
	ctx, cancel := s.newContext()
	defer cancel()
	result := s.openColl(info).FindOneAndUpdate(ctx, query, update)
	if err := result.Err(); err != nil {
		if errors.Is(err, mo.ErrNoDocuments) {
			return err
		}
		s.Log.Error("svc.FindOneAndUpdate.%s: %s filter: %v updater: %v", name, err, query, update)
		return errors.Join(ErrInternalError, err)
	}
	s.Log.Debug("svc.FindOneAndUpdate.%s: document has been updated. filter: %v updater: %v", name, query, update)

	s.refreshCache(info)
	return nil
}

// EstimatedDocumentCount 合计合集中的文档数量
func (s *Service) EstimatedDocumentCount(name ii.Name) (int64, error) {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.EstimatedDocumentCount.%s: item not found", name)
		return 0, ErrItemNotfound
	}
	ctx, cancel := s.newContext()
	defer cancel()
	length, err := s.openColl(info).EstimatedDocumentCount(ctx)
	if err != nil {
		s.Log.Error("svc.EstimatedDocumentCount.%s: %s", name, err)
		return 0, errors.Join(ErrInternalError, err)
	}
	return length, nil
}

// CountDocuments 有条件的合集文档中的数量
func (s *Service) CountDocuments(name ii.Name, filter mo.Filter) (int64, error) {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.CountDocuments.%s: item not found", name)
		return 0, ErrItemNotfound
	}
	query := filter.Done()
	if err := info.QueryFilterCheck(query); err != nil {
		s.Log.Error("svc.CountDocuments.%s: QueryFilterCheck: %s filter: %v", name, err, query)
		return 0, errors.Join(ErrDataError, err)
	}
	ctx, cancel := s.newContext()
	defer cancel()
	length, err := s.openColl(info).CountDocuments(ctx, query)
	if err != nil {
		s.Log.Error("svc.CountDocuments.%s: %s filter: %v", name, err, query)
		return 0, errors.Join(ErrInternalError, err)
	}
	return length, nil
}

// InsertOne 插入一条文档
// MongoDB 在插入文档时对于 _id 的做法: 即 doc 中不存在 _id 字段时会在数据编码时补充 _id 字段并且值使用 mo.ObjectID 而不修改源文档.
// 当 _id 字段存在时不会修改其数据类型. 但类型不为 mo.ObjectID 时会返回错误
func (s *Service) InsertOne(name ii.Name, value any) (mo.ObjectID, error) {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.InsertOne.%s: item not found", name)
		return mo.NilObjectID, ErrItemNotfound
	}
	params, err := s.resolveInsert(value)
	if err != nil {
		s.Log.Error("svc.InsertOne.%s: resolveInsert: %s", name, err)
		return mo.NilObjectID, err
	}
	doc, err := info.PrepareInsert(params, nil)
	if err != nil {
		s.Log.Error("svc.InsertOne.%s: PrepareInsert: %s", name, err)
		return mo.NilObjectID, errors.Join(ErrDataError, err)
	}
	ctx, cancel := s.newContext()
	defer cancel()
	result, err := s.openColl(info).InsertOne(ctx, doc)
	if err != nil {
		s.Log.Error("svc.InsertOne.%s: %s data: %v", name, err, doc)
		return mo.NilObjectID, errors.Join(ErrInternalError, err)
	}
	oid, _ := result.InsertedID.(mo.ObjectID)
	s.Log.Debug("svc.InsertOne.%s: %s", name, oid)

	s.refreshCache(info)
	return oid, nil
}

// InsertMany 插入多条文档
// 对于 _id 的处理参见 InsertOne
func (s *Service) InsertMany(name ii.Name, value any) (mo.A, error) {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.InsertMany.%s: item not found", name)
		return nil, ErrItemNotfound
	}
	docs, err := s.toDocs(value, func(doc mo.D) (mo.D, error) {
		return info.PrepareInsert(doc, nil)
	})
	if err != nil {
		s.Log.Error("svc.InsertMany.%s: toDocs: %s", name, err)
		return nil, errors.Join(ErrDataError, err)
	}
	ctx, cancel := s.newContext()
	defer cancel()
	result, err := s.openColl(info).InsertMany(ctx, docs)
	if err != nil {
		s.Log.Error("svc.InsertMany.%s: %s", name, err)
		return nil, errors.Join(ErrInternalError, err)
	}
	s.Log.Debug("svc.InsertMany.%s: %v", name, result.InsertedIDs)

	s.refreshCache(info)
	return result.InsertedIDs, nil
}

// UpdateOne 更新一条文档, 通常情况下 update 参数需要使用 mo.Updater 构建
// 注意: 为了兼容此前非 mo.Updater 构建的更新参数, 此处 update 参数支持 mo.M 和 mo.D 两种类型的参数, 其他类型会返回错误
// update 类型为 mo.M 时, 会用作 mo.OptSet 形式处理
// update 类型为 mo.D 时: 当 update 长度为 1 且 Key 未指定 mo.OptSet 时则按 mo.OptSet 处理
func (s *Service) UpdateOne(name ii.Name, filter, updater mo.Filter) error {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.UpdateOne.%s: item not found", name)
		return ErrItemNotfound
	}
	query := filter.Done()
	if err := info.QueryFilterCheck(query); err != nil {
		s.Log.Error("svc.UpdateOne.%s: QueryFilterCheck: %s filter: %v", name, err, query)
		return errors.Join(ErrDataError, err)
	}
	update := updater.Done()
	if err := info.PrepareUpdater(update, nil); err != nil {
		s.Log.Error("svc.UpdateOne.%s: PrepareUpdater: %s updater: %v", name, err, update)
		return errors.Join(ErrDataError, err)
	}
	if len(update) == 0 {
		return nil
	}
	opts := mo.Options.UpdateOne()
	_, upsert := mo.HasOptIn(update, mo.OptSetOnInsert)
	opts.SetUpsert(upsert)

	ctx, cancel := s.newContext()
	defer cancel()
	_, err := s.openColl(info).UpdateOne(ctx, query, update, opts)
	if err != nil {
		s.Log.Error("svc.UpdateOne.%s: %s filter: %v updater: %v", name, err, filter, update)
		return errors.Join(ErrInternalError, err)
	}
	s.Log.Debug("svc.UpdateOne.%s: document has been updated. filter: %v updater: %v", name, filter, update)

	s.refreshCache(info)
	return nil
}

// UpdateByID 使用 _id 作为条件更新 1 条数据
func (s *Service) UpdateByID(name ii.Name, id mo.ObjectID, update mo.Filter) error {
	filter := &mo.Matcher{}
	filter.Eq(mo.OID, id)
	return s.UpdateOne(name, filter, update)
}

// UpdateMany 使用 filter 作为条件批量更新数据
// 注意: 兼容性解释见 UpdateOne
func (s *Service) UpdateMany(name ii.Name, filter, updater mo.Filter) error {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.UpdateMany.%s: item not found", name)
		return ErrItemNotfound
	}
	query := filter.Done()
	if err := info.QueryFilterCheck(query); err != nil {
		s.Log.Error("svc.UpdateMany.%s: QueryFilterCheck: %s filter: %v", name, err, query)
		return errors.Join(ErrDataError, err)
	}
	update := updater.Done()
	if err := info.PrepareUpdater(update, nil); err != nil {
		s.Log.Error("svc.UpdateMany.%s: PrepareUpdater: %s updater: %v", name, err, update)
		return errors.Join(ErrDataError, err)
	}
	if len(update) == 0 {
		return nil
	}
	opts := mo.Options.UpdateMany()
	_, upsert := mo.HasOptIn(update, mo.OptSetOnInsert)
	opts.SetUpsert(upsert)

	ctx, cancel := s.newContext()
	defer cancel()
	result, err := s.openColl(info).UpdateMany(ctx, query, update, opts)
	if err != nil {
		s.Log.Error("svc.UpdateMany.%s: %s filter: %v updater: %v", name, err, query, update)
		return errors.Join(ErrInternalError, err)
	}
	s.Log.Debug("svc.UpdateMany.%s: %d documents has been updated. filter: %v updater: %v", name, result.ModifiedCount, query, update)

	s.refreshCache(info)
	return nil
}

// Aggregate 聚合查询
// v 必须传入指针类型
// Aggregate 不传入 XML 配置中的 Lookup/Set 等聚合操作, 当需要时可通过 itemInfo.Aggregation 函数创建后传入
func (s *Service) Aggregate(name ii.Name, pipe mo.Pipeline, v any) error {
	info, ok := s.HasItem(name)
	if !ok {
		s.Log.Error("svc.Aggregate.%s: item not found", name)
		return ErrItemNotfound
	}

	// 如果存在 mo.ArgMatch 操作符时则追加
	if i, d, o := mo.HasOptWith(pipe, mo.ArgMatch); o {
		filter, ok := d.(mo.D)
		if !ok {
			return ErrDataError
		}
		pipe[i] = mo.D{{Key: mo.ArgMatch, Value: filter}}
	}

	var (
		stage  mo.Pipeline
		lookup []ii.Lookup
	)
	copy(stage, pipe)

	if s.Cache != nil {
		stage, lookup = s.Cache.SpitPipe(info, pipe)
	}

	ctx, cancel := s.newContext()
	defer cancel()
	cursor, err := s.openColl(info).Aggregate(ctx, stage)
	if err != nil {
		s.Log.Error("svc.Aggregate.%s: %s pipe: %v", name, err, pipe)
		return errors.Join(ErrInternalError, err)
	}
	if err = mo.CursorDecodeAll(cursor, v); err != nil {
		s.Log.Error("svc.Aggregate.%s: CursorDecodeAll: %s pipe: %v", name, err, pipe)
		return errors.Join(ErrInternalError, err)
	}

	if rows, o := v.(*[]mo.M); o && len(lookup) > 0 {
		if tim := s.Cache.Format(info, lookup, rows); tim.Milliseconds() > 100 {
			s.Log.Warn("svc.Cache.Format: %s -> %s", info.Name, tim)
		}
	}

	return nil
}

func (s *Service) toRow(data mo.D) Row {
	return Row{D: data}
}

func (s *Service) toRows(data []mo.D) []Row {
	rows := make([]Row, len(data))
	for i := 0; i < len(rows); i++ {
		rows[i] = s.toRow(data[i])
	}
	return rows
}

// refreshCache 刷新缓存
func (s *Service) refreshCache(info *ii.ItemInfo) {
	if s.Cache == nil {
		return
	}
	if _, ok := s.Cache.Include(info.Name); !ok {
		return
	}
	if len(s.refreshCh) == 0 {
		s.refreshCh = make(chan *ii.ItemInfo, 128)
		go s.handleRefresh()
	}
	s.refreshCh <- info
}

func (s *Service) handleRefresh() {
	for info := range s.refreshCh {
		qt := time.Now()
		ctx, cancel := s.newContext()
		cursor, err := s.openColl(info).Find(ctx, mo.D{})
		if err != nil {
			cancel()
			s.Cache.Clear(info.Name) // 查询失败时则清除内存缓存, 防止信息不一致
			s.Log.Error("svc.refreshCache: %s->%s", info.Name, err)
			continue
		}
		cancel()
		qts := time.Now().Sub(qt)

		dt := time.Now()
		var data []mo.M
		if err = mo.CursorDecodeAll(cursor, &data); err != nil {
			s.Log.Error("svc.refreshCache.%s: CursorDecodeAll: %s", info.Name, err)
			continue
		}
		dts := time.Now().Sub(dt)

		st := time.Now()
		s.Cache.SetData(info.Name, data)
		sts := time.Now().Sub(st)

		if qts.Milliseconds() >= 100 || dts.Milliseconds() >= 100 || sts.Milliseconds() >= 100 {
			s.Log.Warn("svc.refreshCache: %s query: %s decode: %s set: %s count: %s total: %d",
				info.Name, qts, dts, sts, qts+dts+sts, len(data))
		}
	}
}

func (s *Service) openColl(info *ii.ItemInfo) *mo.Collection {
	return s.Client.Database(info.Name.DbName()).Collection(info.Name.Collection())
}

func (s *Service) newContext() (context.Context, context.CancelFunc) {
	root, cancel := context.WithTimeout(s.Context, s.Timeout)
	go func(s *Service) {
		d := time.Duration(s.Timeout.Seconds()*0.8) * time.Second
		select {
		case <-time.After(d):
			s.Log.Warn("svc.HealthCheck: Warning: Exec has been running for %f seconds.", d.Seconds())
		case <-root.Done():
		}
	}(s)
	return root, cancel
}