405 lines
12 KiB
Go
405 lines
12 KiB
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")
|
|
)
|
|
|
|
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 ItemInput struct {
|
|
SourceIdentity string
|
|
Title string
|
|
Summary string
|
|
Content string
|
|
URL string
|
|
PublishedAt *time.Time
|
|
}
|
|
|
|
type ItemUpdate struct {
|
|
Status string
|
|
Starred bool
|
|
}
|
|
|
|
type SourceRecord struct {
|
|
Source models.SaDatasetSource
|
|
ItemCount int64
|
|
}
|
|
|
|
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("created_by = ? AND source_id IN ?", userID, 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) {
|
|
normalized, err := normalizeSourceInput(input)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var source models.SaDatasetSource
|
|
if err := s.db.Where("identity = ? AND owner_id = ?", identity, userID).First(&source).Error; 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 {
|
|
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 = ? AND created_by = ?", source.ID, userID).Delete(&models.SaDatasetCron{}).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Where("source_id = ? AND created_by = ?", source.ID, userID).Delete(&models.SaDatasetItem{}).Error; err != nil {
|
|
return err
|
|
}
|
|
return tx.Delete(&source).Error
|
|
})
|
|
}
|
|
|
|
func (s *Service) ListItems(userID uint, sourceIdentity string) ([]models.SaDatasetItem, error) {
|
|
query := s.db.Where("created_by = ?", userID)
|
|
if sourceIdentity = strings.TrimSpace(sourceIdentity); sourceIdentity != "" {
|
|
source, err := s.findOwnedSource(userID, sourceIdentity)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
query = query.Where("source_id = ?", source.ID)
|
|
}
|
|
var items []models.SaDatasetItem
|
|
if err := query.Order("COALESCE(published_at, created_at) desc, id desc").Limit(200).Find(&items).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
if items == nil {
|
|
items = []models.SaDatasetItem{}
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
func (s *Service) CountItems(userID uint, sourceID uint) (int64, error) {
|
|
var count int64
|
|
err := s.db.Model(&models.SaDatasetItem{}).
|
|
Where("created_by = ? AND 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) {
|
|
status := strings.TrimSpace(input.Status)
|
|
if status != "unread" && status != "read" && status != "archived" {
|
|
return nil, ErrStatusInvalid
|
|
}
|
|
var item models.SaDatasetItem
|
|
if err := s.db.Where("identity = ? AND created_by = ?", identity, userID).First(&item).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.db.Model(&item).Updates(map[string]any{"status": status, "starred": input.Starred}).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
return &item, nil
|
|
}
|
|
|
|
func (s *Service) ListCrons(userID uint) ([]models.SaDatasetCron, error) {
|
|
var crons []models.SaDatasetCron
|
|
if err := s.db.Where("created_by = ?", userID).Order("created_at desc, 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)
|
|
}
|
|
|
|
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) ([]models.SaDatasetCron, error) {
|
|
result := make([]models.SaDatasetCron, 0)
|
|
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
|
|
}
|
|
for _, source := range sources {
|
|
if err := ctx.Err(); err != nil {
|
|
return result, err
|
|
}
|
|
now := time.Now().UTC()
|
|
cron := models.SaDatasetCron{
|
|
SourceID: source.ID, CreatedBy: userID, Schedule: "@once",
|
|
Status: "running", Enabled: true, NextRunAt: &now, LastRunAt: &now,
|
|
}
|
|
if err := s.db.Create(&cron).Error; err != nil {
|
|
return nil, 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 nil, err
|
|
}
|
|
result = append(result, cron)
|
|
if err := ctx.Err(); err != nil {
|
|
return result, err
|
|
}
|
|
continue
|
|
}
|
|
|
|
inserted, err := s.storeFeedItems(userID, source, feed.Items)
|
|
if err != 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 nil, updateErr
|
|
}
|
|
result = append(result, cron)
|
|
continue
|
|
}
|
|
completedAt := time.Now().UTC()
|
|
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 := s.db.Transaction(func(tx *gorm.DB) error {
|
|
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
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
source.LastSyncedAt = &completedAt
|
|
result = append(result, cron)
|
|
}
|
|
return result, 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 {
|
|
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 result.Error
|
|
}
|
|
inserted += int(result.RowsAffected)
|
|
}
|
|
return nil
|
|
})
|
|
return inserted, err
|
|
}
|
|
|
|
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 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 != "manual" && input.Kind != "link" && 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.Kind != "manual" && input.URL == "" {
|
|
return input, ErrURLInvalid
|
|
}
|
|
return input, nil
|
|
}
|
|
|
|
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
|
|
}
|