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")
	ErrSyncInProgress = errors.New("dataset sync is already in progress")
)

const (
	defaultItemPageSize  = 50
	maxItemPageSize      = 200
	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
	ImageURL       string
	PublishedAt    *time.Time
}

type ItemUpdate struct {
	Status  string
	Starred bool
}

type ItemPatch struct {
	Status  *string
	Starred *bool
}

type DepositResult struct {
	DocumentIdentity string
	ProjectIdentity  string
}

type SourceRecord struct {
	Source    models.SaDatasetSource
	ItemCount int64
}

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

type SyncResult struct {
	SourceIdentity string
	Status         string
	Result         string
}

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, 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 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
	}
	imageURL := strings.TrimSpace(input.ImageURL)
	if imageURL != "" && !validHTTPURL(imageURL) {
		return nil, ErrURLInvalid
	}
	item := &models.SaDatasetItem{
		SourceID: source.ID, Title: title,
		Summary: strings.TrimSpace(input.Summary), Content: strings.TrimSpace(input.Content),
		URL: itemURL, ImageURL: imageURL, 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.ImageURL != "" {
		if body != "" {
			body = "![文章图片](" + item.ImageURL + ")\n\n" + body
		} else {
			body = "![文章图片](" + item.ImageURL + ")"
		}
	}
	if item.URL != "" {
		if body != "" {
			body += "\n\n"
		}
		body += "[原文链接](" + item.URL + ")"
	}
	name := strings.TrimSpace(item.Title)
	if !strings.HasSuffix(strings.ToLower(name), ".md") {
		name += ".md"
	}
	var document models.SaDocumentTree
	err = s.db.Transaction(func(tx *gorm.DB) error {
		document = models.SaDocumentTree{
			ProjectID: project.ID, OwnerID: userID, Kind: models.DocumentKindFile,
			Name: name, NormalizedName: strings.ToLower(name), Extension: ".md",
			MimeType: "text/markdown; charset=utf-8", Size: int64(len([]byte(body))),
			Revision: 1, ScanStatus: "not_scanned",
		}
		if err := tx.Create(&document).Error; err != nil {
			return err
		}
		return tx.Create(&models.SaDocumentContent{DocumentID: document.ID, Markdown: body}).Error
	})
	if err != nil {
		return DepositResult{}, err
	}
	return DepositResult{DocumentIdentity: document.Identity, ProjectIdentity: project.Identity}, nil
}

func (s *Service) SyncSources(ctx context.Context, userID uint) ([]SyncResult, error) {
	if !s.syncMu.TryLock() {
		return nil, ErrSyncInProgress
	}
	defer s.syncMu.Unlock()

	sources, err := s.enabledRSSSources(userID)
	if err != nil {
		return nil, err
	}
	return s.executeDatasetSources(ctx, sources)
}

func (s *Service) SyncAllSources(ctx context.Context) ([]SyncResult, error) {
	if !s.syncMu.TryLock() {
		return nil, ErrSyncInProgress
	}
	defer s.syncMu.Unlock()

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

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) executeDatasetSources(ctx context.Context, sources []models.SaDatasetSource) ([]SyncResult, error) {
	result := make([]SyncResult, len(sources))
	runErrors := make([]error, len(sources))
	concurrency := maxConcurrentFetches
	if s.db.Dialector.Name() == "sqlite" {
		concurrency = 1
	}
	semaphore := make(chan struct{}, concurrency)
	var workers sync.WaitGroup
	for index, source := range sources {
		workers.Add(1)
		go func() {
			defer workers.Done()
			semaphore <- struct{}{}
			defer func() { <-semaphore }()
			result[index], runErrors[index] = s.executeDatasetSource(ctx, source)
		}()
	}
	workers.Wait()
	return result, errors.Join(runErrors...)
}

func (s *Service) executeDatasetSource(ctx context.Context, source models.SaDatasetSource) (SyncResult, error) {
	result := SyncResult{SourceIdentity: source.Identity}
	if err := ctx.Err(); err != nil {
		result.Status = "failed"
		result.Result = truncateResult(err.Error())
		return result, err
	}

	feed, fetchErr := s.fetcher.Fetch(ctx, source.URL)
	if fetchErr != nil {
		result.Status = "failed"
		result.Result = truncateResult(fetchErr.Error())
		return result, nil
	}

	inserted := 0
	completedAt := time.Now().UTC()
	err := s.db.Transaction(func(tx *gorm.DB) error {
		var storeErr error
		inserted, storeErr = storeFeedItems(tx, source, feed.Items)
		if storeErr != nil {
			return storeErr
		}
		return tx.Model(&source).Update("last_synced_at", completedAt).Error
	})
	if err != nil {
		result.Status = "failed"
		result.Result = truncateResult("store feed items: " + err.Error())
		return result, nil
	}
	result.Status = "completed"
	result.Result = fmt.Sprintf("format=%s fetched=%d inserted=%d", feed.Format, len(feed.Items), inserted)
	return result, nil
}

func storeFeedItems(tx *gorm.DB, source models.SaDatasetSource, items []FeedItem) (int, error) {
	inserted := 0
	for _, input := range items {
		externalID := input.ExternalID
		itemURL := resolveFeedReference(source.URL, input.URL)
		imageURL := resolveFeedReference(source.URL, input.ImageURL)
		item := models.SaDatasetItem{
			SourceID: source.ID, ExternalID: &externalID,
			Title: input.Title, Summary: input.Summary, Content: input.Content,
			URL: itemURL, ImageURL: imageURL,
			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)
		if result.RowsAffected == 0 {
			updates := map[string]any{
				"title": item.Title,
			}
			if item.Summary != "" {
				updates["summary"] = item.Summary
			}
			if item.Content != "" {
				updates["content"] = item.Content
			}
			if item.URL != "" {
				updates["url"] = item.URL
			}
			if item.ImageURL != "" {
				updates["image_url"] = item.ImageURL
			}
			if item.PublishedAt != nil {
				updates["published_at"] = item.PublishedAt
			}
			if err := tx.Model(&models.SaDatasetItem{}).
				Where("source_id = ? AND external_id = ?", source.ID, externalID).
				Updates(updates).Error; err != nil {
				return inserted, err
			}
		}
	}
	return inserted, nil
}

func resolveFeedReference(baseURL, reference string) string {
	reference = strings.TrimSpace(reference)
	if reference == "" {
		return ""
	}
	parsedReference, err := url.Parse(reference)
	if err != nil {
		return ""
	}
	if !parsedReference.IsAbs() {
		parsedBase, err := url.Parse(baseURL)
		if err != nil {
			return ""
		}
		parsedReference = parsedBase.ResolveReference(parsedReference)
	}
	resolved := parsedReference.String()
	if !validHTTPURL(resolved) {
		return ""
	}
	return truncateRunes(resolved, 2048)
}

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
}