Veyra/internal/service/market.go

package service

import (
	"context"
	"errors"
	"git.sundynix.cn/Blizzard/Veyra/internal/apperror"
	"git.sundynix.cn/Blizzard/Veyra/internal/model"
	"go.uber.org/zap"
	"strings"
	"time"
	"unicode/utf8"
)

type MarketRepository interface {
	List(context.Context, string) ([]model.Instrument, error)
	Save(context.Context, model.ProviderConfig, []model.Instrument, time.Time, string) error
	Tasks(context.Context) ([]model.MarketSync, error)
}
type MarketAdapter interface {
	Snapshot(context.Context, model.ProviderConfig) ([]model.Instrument, error)
	History(context.Context, model.ProviderConfig, string, string, string, string) (model.MarketHistory, error)
}
type MarketService struct {
	repo        MarketRepository
	providers   ProviderRepository
	adapter     MarketAdapter
	development bool
}

func NewMarketService(repo MarketRepository, providers ProviderRepository, adapter MarketAdapter, development bool) *MarketService {
	return &MarketService{repo, providers, adapter, development}
}
func (s *MarketService) authorize(u model.User) error {
	if u.Status != "ACTIVE" || (u.Role != "ADMIN" && u.Role != "USER") {
		return apperror.ErrForbidden
	}
	if s == nil || s.repo == nil || s.providers == nil || !s.development {
		return apperror.ErrUnavailable
	}
	return nil
}
func (s *MarketService) configuration(ctx context.Context) (model.ProviderConfig, error) {
	c, e := s.providers.Get(ctx, "market_primary")
	if e != nil {
		return c, e
	}
	if !c.Enabled || (c.MarketAdapter != "eastmoney_public" && c.MarketAdapter != "sina_public") {
		return c, apperror.ErrUnavailable
	}
	return c, nil
}
func (s *MarketService) List(ctx context.Context, u model.User, query string) ([]model.Instrument, error) {
	if e := s.authorize(u); e != nil {
		return nil, e
	}
	query = strings.TrimSpace(query)
	if utf8.RuneCountInString(query) > 80 {
		return nil, apperror.ErrInvalidInput
	}
	return s.repo.List(ctx, query)
}
func (s *MarketService) Tasks(ctx context.Context, u model.User) ([]model.MarketSync, error) {
	if e := s.authorize(u); e != nil {
		return nil, e
	}
	if u.Role != "ADMIN" {
		return nil, apperror.ErrForbidden
	}
	return s.repo.Tasks(ctx)
}
func (s *MarketService) History(ctx context.Context, u model.User, symbol, exchange, start, end string) (model.MarketHistory, error) {
	if e := s.authorize(u); e != nil {
		return model.MarketHistory{}, e
	}
	if len(symbol) != 6 || (exchange != "SSE" && exchange != "SZSE") {
		return model.MarketHistory{}, apperror.ErrInvalidInput
	}
	for _, r := range symbol {
		if r < '0' || r > '9' {
			return model.MarketHistory{}, apperror.ErrInvalidInput
		}
	}
	a, e1 := time.Parse("2006-01-02", start)
	b, e2 := time.Parse("2006-01-02", end)
	if e1 != nil || e2 != nil || a.After(b) || b.Sub(a) > 1460*24*time.Hour || b.After(time.Now()) {
		return model.MarketHistory{}, apperror.ErrInvalidInput
	}
	c, e := s.configuration(ctx)
	if e != nil {
		return model.MarketHistory{}, e
	}
	result, e := s.adapter.History(ctx, c, symbol, exchange, start, end)
	if e != nil {
		return result, apperror.ErrUnavailable
	}
	return result, nil
}
func (s *MarketService) Sync(ctx context.Context) error {
	c, e := s.configuration(ctx)
	if errors.Is(e, apperror.ErrNotFound) || errors.Is(e, apperror.ErrUnavailable) {
		return nil
	}
	if e != nil {
		return e
	}
	started := time.Now().UTC()
	list, e := s.adapter.Snapshot(ctx, c)
	if ctx.Err() != nil {
		return ctx.Err()
	}
	message := ""
	if e != nil {
		message = "行情同步失败，保留上次快照并标明行情时间。"
	}
	saveError := s.repo.Save(ctx, c, list, started, message)
	if saveError != nil {
		return saveError
	}
	return e
}
func (s *MarketService) Run(ctx context.Context, logger *zap.Logger) {
	if s == nil || s.repo == nil || s.providers == nil || s.adapter == nil || !s.development {
		return
	}
	var lastAttempt time.Time
	var lastVersion int64
	for {
		config, err := s.configuration(ctx)
		if err == nil && (config.Version != lastVersion || time.Since(lastAttempt) >= 5*time.Minute) {
			lastAttempt = time.Now()
			lastVersion = config.Version
			task, cancel := context.WithTimeout(ctx, 90*time.Second)
			err = s.Sync(task)
			cancel()
			if err != nil && ctx.Err() == nil {
				logger.Warn("market synchronization failed", zap.Error(err))
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(10 * time.Second):
		}
	}
}