agent/backend/internal/logic/dataset/service.go

package dataset

import (
	"context"
	"errors"
	"fmt"
	"net/url"
	"strings"
	"sync"
	"time"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"
	"senlinai-agent/backend/internal/models"
)

var (
	ErrNameRequired   = errors.New("dataset source name is required")
	ErrKindInvalid    = errors.New("dataset source kind is invalid")
	ErrURLInvalid     = errors.New("dataset source url is invalid")
	ErrTitleRequired  = errors.New("dataset item title is required")
	ErrStatusInvalid  = errors.New("dataset item status is invalid")
	ErrBuiltInSource  = errors.New("built-in dataset source cannot be deleted")
	ErrSyncInProgress = errors.New("dataset sync is already in progress")
)

const (
	defaultItemPageSize  = 50
	maxItemPageSize      = 200
	datasetRunRetention  = 30 * 24 * time.Hour
	maxConcurrentFetches = 4
)

type Service struct {
	db      *gorm.DB
	fetcher FeedFetcher
	syncMu  sync.Mutex
}

type SourceInput struct {
	Name        string
	Kind        string
	URL         string
	IconURL     string
	Description string
	Enabled     bool
}

type SourcePatch struct {
	Name        *string
	Kind        *string
	URL         *string
	IconURL     *string
	Description *string
	Enabled     *bool
}

type ItemInput struct {
	SourceIdentity string
	Title          string
	Summary        string
	Content        string
	URL            string
	PublishedAt    *time.Time
}

type ItemUpdate struct {
	Status  string
	Starred bool
}

type ItemPatch struct {
	Status  *string
	Starred *bool
}

type DepositResult struct {
	NoteIdentity    string
	ProjectIdentity string
}

type SourceRecord struct {
	Source    models.SaDatasetSource
	ItemCount int64
}

type ItemPage struct {
	Items  []models.SaDatasetItem
	Total  int64
	Limit  int
	Offset int
}

type queuedSource struct {
	Source models.SaDatasetSource
	Run    models.SaDatasetCron
}

func NewService(database *gorm.DB) *Service {
	return &Service{db: database, fetcher: NewHTTPFeedFetcher()}
}

func newServiceWithFetcher(database *gorm.DB, fetcher FeedFetcher) *Service {
	return &Service{db: database, fetcher: fetcher}
}

func (s *Service) ListSources(userID uint) ([]SourceRecord, error) {
	var sources []models.SaDatasetSource
	if err := s.db.Where("owner_id = ?", userID).Order("created_at asc, id asc").Find(&sources).Error; err != nil {
		return nil, err
	}
	counts := make(map[uint]int64)
	if len(sources) > 0 {
		sourceIDs := make([]uint, 0, len(sources))
		for _, source := range sources {
			sourceIDs = append(sourceIDs, source.ID)
		}
		var rows []struct {
			SourceID uint
			Count    int64
		}
		if err := s.db.Model(&models.SaDatasetItem{}).
			Select("source_id, COUNT(*) AS count").
			Where("source_id IN ?", sourceIDs).
			Group("source_id").
			Scan(&rows).Error; err != nil {
			return nil, err
		}
		for _, row := range rows {
			counts[row.SourceID] = row.Count
		}
	}
	result := make([]SourceRecord, 0, len(sources))
	for _, source := range sources {
		result = append(result, SourceRecord{Source: source, ItemCount: counts[source.ID]})
	}
	return result, nil
}

func (s *Service) CreateSource(userID uint, input SourceInput) (*models.SaDatasetSource, error) {
	normalized, err := normalizeSourceInput(input)
	if err != nil {
		return nil, err
	}
	source := &models.SaDatasetSource{
		OwnerID: userID, CreatedBy: userID, Name: normalized.Name, Kind: normalized.Kind,
		URL: normalized.URL, IconURL: normalized.IconURL,
		Description: normalized.Description, Enabled: normalized.Enabled,
	}
	return source, s.db.Create(source).Error
}

func (s *Service) UpdateSource(userID uint, identity string, input SourceInput) (*models.SaDatasetSource, error) {
	return s.PatchSource(userID, identity, SourcePatch{
		Name: &input.Name, Kind: &input.Kind, URL: &input.URL, IconURL: &input.IconURL,
		Description: &input.Description, Enabled: &input.Enabled,
	})
}

func (s *Service) PatchSource(userID uint, identity string, patch SourcePatch) (*models.SaDatasetSource, error) {
	if !s.syncMu.TryLock() {
		return nil, ErrSyncInProgress
	}
	defer s.syncMu.Unlock()

	var source models.SaDatasetSource
	if err := s.db.Where("identity = ? AND owner_id = ?", identity, userID).First(&source).Error; err != nil {
		return nil, err
	}
	input := SourceInput{
		Name: source.Name, Kind: source.Kind, URL: source.URL, IconURL: source.IconURL,
		Description: source.Description, Enabled: source.Enabled,
	}
	if patch.Name != nil {
		input.Name = *patch.Name
	}
	if patch.Kind != nil {
		input.Kind = *patch.Kind
	}
	if patch.URL != nil {
		input.URL = *patch.URL
	}
	if patch.IconURL != nil {
		input.IconURL = *patch.IconURL
	}
	if patch.Description != nil {
		input.Description = *patch.Description
	}
	if patch.Enabled != nil {
		input.Enabled = *patch.Enabled
	}
	managedFields := source.SeedKey != nil || source.Kind != "rss"
	if managedFields {
		input.Kind = source.Kind
		input.URL = source.URL
		input.IconURL = source.IconURL
	}
	var normalized SourceInput
	var err error
	if source.Kind == "rss" {
		normalized, err = normalizeSourceInput(input)
	} else {
		normalized, err = normalizeLegacySourceInput(input)
	}
	if err != nil {
		return nil, err
	}
	if err := s.db.Model(&source).Updates(map[string]any{
		"name": normalized.Name, "kind": normalized.Kind, "url": normalized.URL,
		"icon_url": normalized.IconURL, "description": normalized.Description, "enabled": normalized.Enabled,
	}).Error; err != nil {
		return nil, err
	}
	return &source, nil
}

func (s *Service) DeleteSource(userID uint, identity string) error {
	if !s.syncMu.TryLock() {
		return ErrSyncInProgress
	}
	defer s.syncMu.Unlock()

	return s.db.Transaction(func(tx *gorm.DB) error {
		var source models.SaDatasetSource
		if err := tx.Where("identity = ? AND owner_id = ?", identity, userID).First(&source).Error; err != nil {
			return err
		}
		if source.SeedKey != nil {
			return ErrBuiltInSource
		}
		if err := tx.Where("source_id = ?", source.ID).Delete(&models.SaDatasetCron{}).Error; err != nil {
			return err
		}
		if err := tx.Where("source_id = ?", source.ID).Delete(&models.SaDatasetItem{}).Error; err != nil {
			return err
		}
		return tx.Delete(&source).Error
	})
}

func (s *Service) ListItems(userID uint, sourceIdentity string) ([]models.SaDatasetItem, error) {
	page, err := s.ListItemsPage(userID, sourceIdentity, maxItemPageSize, 0)
	if err != nil {
		return nil, err
	}
	return page.Items, nil
}

func (s *Service) ListItemsPage(userID uint, sourceIdentity string, limit, offset int) (ItemPage, error) {
	limit, offset = normalizePage(limit, offset)
	query := s.db.Model(&models.SaDatasetItem{}).
		Joins("JOIN sa_dataset_sources ON sa_dataset_sources.id = sa_dataset_items.source_id").
		Where("sa_dataset_sources.owner_id = ?", userID)
	if sourceIdentity = strings.TrimSpace(sourceIdentity); sourceIdentity != "" {
		source, err := s.findOwnedSource(userID, sourceIdentity)
		if err != nil {
			return ItemPage{}, err
		}
		query = query.Where("sa_dataset_items.source_id = ?", source.ID)
	}
	var total int64
	if err := query.Session(&gorm.Session{}).Count(&total).Error; err != nil {
		return ItemPage{}, err
	}
	var items []models.SaDatasetItem
	if err := query.Select("sa_dataset_items.*").
		Order("COALESCE(sa_dataset_items.published_at, sa_dataset_items.created_at) desc, sa_dataset_items.id desc").
		Limit(limit).Offset(offset).Find(&items).Error; err != nil {
		return ItemPage{}, err
	}
	if items == nil {
		items = []models.SaDatasetItem{}
	}
	return ItemPage{Items: items, Total: total, Limit: limit, Offset: offset}, nil
}

func (s *Service) CountItems(userID uint, sourceID uint) (int64, error) {
	var count int64
	err := s.db.Model(&models.SaDatasetItem{}).
		Joins("JOIN sa_dataset_sources ON sa_dataset_sources.id = sa_dataset_items.source_id").
		Where("sa_dataset_sources.owner_id = ? AND sa_dataset_items.source_id = ?", userID, sourceID).
		Count(&count).Error
	return count, err
}

func (s *Service) CreateItem(userID uint, input ItemInput) (*models.SaDatasetItem, error) {
	title := strings.TrimSpace(input.Title)
	if title == "" {
		return nil, ErrTitleRequired
	}
	source, err := s.findOwnedSource(userID, input.SourceIdentity)
	if err != nil {
		return nil, err
	}
	itemURL := strings.TrimSpace(input.URL)
	if itemURL != "" && !validHTTPURL(itemURL) {
		return nil, ErrURLInvalid
	}
	item := &models.SaDatasetItem{
		SourceID: source.ID, CreatedBy: userID, Title: title,
		Summary: strings.TrimSpace(input.Summary), Content: strings.TrimSpace(input.Content),
		URL: itemURL, Status: "unread", PublishedAt: utcOptionalTime(input.PublishedAt),
	}
	return item, s.db.Create(item).Error
}

func (s *Service) UpdateItem(userID uint, identity string, input ItemUpdate) (*models.SaDatasetItem, error) {
	return s.PatchItem(userID, identity, ItemPatch{Status: &input.Status, Starred: &input.Starred})
}

func (s *Service) PatchItem(userID uint, identity string, patch ItemPatch) (*models.SaDatasetItem, error) {
	item, err := s.findOwnedItem(userID, identity)
	if err != nil {
		return nil, err
	}
	updates := make(map[string]any, 2)
	if patch.Status != nil {
		status := strings.TrimSpace(*patch.Status)
		if status != "unread" && status != "read" && status != "archived" {
			return nil, ErrStatusInvalid
		}
		updates["status"] = status
	}
	if patch.Starred != nil {
		updates["starred"] = *patch.Starred
	}
	if len(updates) == 0 {
		return item, nil
	}
	if err := s.db.Model(item).Updates(updates).Error; err != nil {
		return nil, err
	}
	return item, nil
}

func (s *Service) DepositItem(userID uint, itemIdentity, projectIdentity string) (DepositResult, error) {
	item, err := s.findOwnedItem(userID, itemIdentity)
	if err != nil {
		return DepositResult{}, err
	}
	var project models.SaProject
	if err := s.db.Where("identity = ? AND owner_id = ?", strings.TrimSpace(projectIdentity), userID).
		First(&project).Error; err != nil {
		return DepositResult{}, err
	}
	body := strings.TrimSpace(item.Content)
	if body == "" {
		body = strings.TrimSpace(item.Summary)
	}
	if item.URL != "" {
		if body != "" {
			body += "\n\n"
		}
		body += "[原文链接](" + item.URL + ")"
	}
	note := models.SaNote{
		ProjectID: project.ID, CreatedBy: userID,
		Title: item.Title, Markdown: body,
	}
	if err := s.db.Create(&note).Error; err != nil {
		return DepositResult{}, err
	}
	return DepositResult{NoteIdentity: note.Identity, ProjectIdentity: project.Identity}, nil
}

func (s *Service) ListCrons(userID uint) ([]models.SaDatasetCron, error) {
	var crons []models.SaDatasetCron
	if err := s.db.Model(&models.SaDatasetCron{}).
		Select("sa_dataset_crons.*").
		Joins("JOIN sa_dataset_sources ON sa_dataset_sources.id = sa_dataset_crons.source_id").
		Where("sa_dataset_sources.owner_id = ?", userID).
		Order("sa_dataset_crons.created_at desc, sa_dataset_crons.id desc").
		Limit(100).Find(&crons).Error; err != nil {
		return nil, err
	}
	if crons == nil {
		crons = []models.SaDatasetCron{}
	}
	return crons, nil
}

func (s *Service) SyncSources(ctx context.Context, userID uint) ([]models.SaDatasetCron, error) {
	s.syncMu.Lock()
	defer s.syncMu.Unlock()

	return s.syncSources(ctx, userID, "scheduled")
}

func (s *Service) QueueSources(userID uint) ([]models.SaDatasetCron, error) {
	if !s.syncMu.TryLock() {
		return nil, ErrSyncInProgress
	}
	sources, err := s.enabledRSSSources(userID)
	if err != nil {
		s.syncMu.Unlock()
		return nil, err
	}
	queued, err := s.createDatasetRuns(userID, sources, "manual")
	if err != nil {
		s.syncMu.Unlock()
		return nil, err
	}
	runs := make([]models.SaDatasetCron, 0, len(queued))
	for _, entry := range queued {
		runs = append(runs, entry.Run)
	}
	if len(queued) == 0 {
		s.syncMu.Unlock()
		return runs, nil
	}
	go func() {
		defer s.syncMu.Unlock()
		_, _ = s.executeDatasetRuns(context.Background(), queued)
	}()
	return runs, nil
}

func (s *Service) SyncAllSources(ctx context.Context) ([]models.SaDatasetCron, error) {
	var userIDs []uint
	if err := s.db.Model(&models.SaDatasetSource{}).
		Distinct("owner_id").
		Where("enabled = ? AND kind = ?", true, "rss").
		Order("owner_id asc").
		Pluck("owner_id", &userIDs).Error; err != nil {
		return nil, err
	}

	result := make([]models.SaDatasetCron, 0)
	var syncErrors []error
	for _, userID := range userIDs {
		crons, err := s.SyncSources(ctx, userID)
		result = append(result, crons...)
		if err != nil {
			syncErrors = append(syncErrors, fmt.Errorf("sync dataset sources for user %d: %w", userID, err))
		}
		if ctx.Err() != nil {
			break
		}
	}
	return result, errors.Join(syncErrors...)
}

func (s *Service) syncSources(ctx context.Context, userID uint, trigger string) ([]models.SaDatasetCron, error) {
	sources, err := s.enabledRSSSources(userID)
	if err != nil {
		return nil, err
	}
	queued, err := s.createDatasetRuns(userID, sources, trigger)
	if err != nil {
		return nil, err
	}
	return s.executeDatasetRuns(ctx, queued)
}

func (s *Service) enabledRSSSources(userID uint) ([]models.SaDatasetSource, error) {
	var sources []models.SaDatasetSource
	if err := s.db.Where("owner_id = ? AND enabled = ? AND kind = ?", userID, true, "rss").
		Order("id asc").Find(&sources).Error; err != nil {
		return nil, err
	}
	return sources, nil
}

func (s *Service) createDatasetRuns(userID uint, sources []models.SaDatasetSource, trigger string) ([]queuedSource, error) {
	queued := make([]queuedSource, 0, len(sources))
	now := time.Now().UTC()
	err := s.db.Transaction(func(tx *gorm.DB) error {
		if err := tx.Where("created_at < ?", now.Add(-datasetRunRetention)).
			Delete(&models.SaDatasetCron{}).Error; err != nil {
			return err
		}
		for _, source := range sources {
			run := models.SaDatasetCron{
				SourceID: source.ID, CreatedBy: userID, Schedule: trigger,
				Status: "pending", Enabled: true, NextRunAt: &now,
			}
			if err := tx.Create(&run).Error; err != nil {
				return err
			}
			queued = append(queued, queuedSource{Source: source, Run: run})
		}
		return nil
	})
	return queued, err
}

func (s *Service) executeDatasetRuns(ctx context.Context, queued []queuedSource) ([]models.SaDatasetCron, error) {
	result := make([]models.SaDatasetCron, len(queued))
	runErrors := make([]error, len(queued))
	concurrency := maxConcurrentFetches
	if s.db.Dialector.Name() == "sqlite" {
		concurrency = 1
	}
	semaphore := make(chan struct{}, concurrency)
	var workers sync.WaitGroup
	for index, entry := range queued {
		workers.Add(1)
		go func() {
			defer workers.Done()
			semaphore <- struct{}{}
			defer func() { <-semaphore }()
			result[index], runErrors[index] = s.executeDatasetRun(ctx, entry)
		}()
	}
	workers.Wait()
	return result, errors.Join(runErrors...)
}

func (s *Service) executeDatasetRun(ctx context.Context, entry queuedSource) (models.SaDatasetCron, error) {
	source := entry.Source
	cron := entry.Run
	if err := ctx.Err(); err != nil {
		return cron, err
	}
	now := time.Now().UTC()
	cron.Status = "running"
	cron.LastRunAt = &now
	if err := s.db.Model(&cron).Updates(map[string]any{
		"status": "running", "last_run_at": now,
	}).Error; err != nil {
		return cron, err
	}

	feed, fetchErr := s.fetcher.Fetch(ctx, source.URL)
	if fetchErr != nil {
		cron.Status = "failed"
		cron.Enabled = false
		cron.NextRunAt = nil
		cron.LastResult = truncateResult(fetchErr.Error())
		if err := s.db.Model(&cron).Updates(map[string]any{
			"status": cron.Status, "last_result": cron.LastResult, "enabled": false, "next_run_at": nil,
		}).Error; err != nil {
			return cron, err
		}
		return cron, ctx.Err()
	}

	inserted := 0
	completedAt := time.Now().UTC()
	err := s.db.Transaction(func(tx *gorm.DB) error {
		var storeErr error
		inserted, storeErr = storeFeedItems(tx, cron.CreatedBy, source, feed.Items)
		if storeErr != nil {
			return storeErr
		}
		cron.Status = "completed"
		cron.Enabled = false
		cron.NextRunAt = nil
		cron.LastResult = fmt.Sprintf("format=%s fetched=%d inserted=%d", feed.Format, len(feed.Items), inserted)
		if err := tx.Model(&cron).Updates(map[string]any{
			"status": cron.Status, "last_result": cron.LastResult, "enabled": false, "next_run_at": nil,
		}).Error; err != nil {
			return err
		}
		return tx.Model(&source).Update("last_synced_at", completedAt).Error
	})
	if err == nil {
		return cron, nil
	}
	cron.Status = "failed"
	cron.Enabled = false
	cron.NextRunAt = nil
	cron.LastResult = truncateResult("store feed items: " + err.Error())
	if updateErr := s.db.Model(&cron).Updates(map[string]any{
		"status": cron.Status, "last_result": cron.LastResult, "enabled": false, "next_run_at": nil,
	}).Error; updateErr != nil {
		return cron, updateErr
	}
	return cron, nil
}

func (s *Service) storeFeedItems(userID uint, source models.SaDatasetSource, items []FeedItem) (int, error) {
	inserted := 0
	err := s.db.Transaction(func(tx *gorm.DB) error {
		var err error
		inserted, err = storeFeedItems(tx, userID, source, items)
		return err
	})
	return inserted, err
}

func storeFeedItems(tx *gorm.DB, userID uint, source models.SaDatasetSource, items []FeedItem) (int, error) {
	inserted := 0
	for _, input := range items {
		externalID := input.ExternalID
		item := models.SaDatasetItem{
			SourceID: source.ID, CreatedBy: userID, ExternalID: &externalID,
			Title: input.Title, Summary: input.Summary, Content: input.Content,
			URL: input.URL, Status: "unread", PublishedAt: input.PublishedAt,
		}
		result := tx.Clauses(clause.OnConflict{
			Columns:   []clause.Column{{Name: "source_id"}, {Name: "external_id"}},
			DoNothing: true,
		}).Create(&item)
		if result.Error != nil {
			return inserted, result.Error
		}
		inserted += int(result.RowsAffected)
	}
	return inserted, nil
}

func truncateResult(value string) string {
	return truncateRunes(value, 2000)
}

func truncateRunes(value string, limit int) string {
	runes := []rune(value)
	if len(runes) <= limit {
		return value
	}
	return string(runes[:limit])
}

func (s *Service) findOwnedSource(userID uint, identity string) (*models.SaDatasetSource, error) {
	var source models.SaDatasetSource
	err := s.db.Where("identity = ? AND owner_id = ?", strings.TrimSpace(identity), userID).First(&source).Error
	return &source, err
}

func (s *Service) findOwnedItem(userID uint, identity string) (*models.SaDatasetItem, error) {
	var item models.SaDatasetItem
	err := s.db.Model(&models.SaDatasetItem{}).
		Select("sa_dataset_items.*").
		Joins("JOIN sa_dataset_sources ON sa_dataset_sources.id = sa_dataset_items.source_id").
		Where("sa_dataset_items.identity = ? AND sa_dataset_sources.owner_id = ?", strings.TrimSpace(identity), userID).
		First(&item).Error
	return &item, err
}

func normalizeSourceInput(input SourceInput) (SourceInput, error) {
	input.Name = strings.TrimSpace(input.Name)
	input.Kind = strings.ToLower(strings.TrimSpace(input.Kind))
	input.URL = strings.TrimSpace(input.URL)
	input.IconURL = strings.TrimSpace(input.IconURL)
	input.Description = strings.TrimSpace(input.Description)
	if input.Name == "" {
		return input, ErrNameRequired
	}
	if input.Kind != "rss" {
		return input, ErrKindInvalid
	}
	if input.URL != "" && !validHTTPURL(input.URL) {
		return input, ErrURLInvalid
	}
	if input.IconURL != "" && !validHTTPURL(input.IconURL) {
		return input, ErrURLInvalid
	}
	if input.URL == "" {
		return input, ErrURLInvalid
	}
	return input, nil
}

func normalizeLegacySourceInput(input SourceInput) (SourceInput, error) {
	input.Name = strings.TrimSpace(input.Name)
	input.Description = strings.TrimSpace(input.Description)
	if input.Name == "" {
		return input, ErrNameRequired
	}
	return input, nil
}

func normalizePage(limit, offset int) (int, int) {
	if limit <= 0 {
		limit = defaultItemPageSize
	}
	if limit > maxItemPageSize {
		limit = maxItemPageSize
	}
	if offset < 0 {
		offset = 0
	}
	return limit, offset
}

func validHTTPURL(value string) bool {
	parsed, err := url.ParseRequestURI(value)
	return err == nil && (parsed.Scheme == "http" || parsed.Scheme == "https") &&
		parsed.Host != "" && parsed.User == nil
}

func utcOptionalTime(value *time.Time) *time.Time {
	if value == nil {
		return nil
	}
	result := value.UTC()
	return &result
}