285 lines
8.4 KiB
Go
285 lines
8.4 KiB
Go
package dataset
|
|
|
|
import (
|
|
"errors"
|
|
"net/url"
|
|
"strings"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
"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
|
|
}
|
|
|
|
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}
|
|
}
|
|
|
|
func (s *Service) ListSources(userID uint) ([]SourceRecord, error) {
|
|
var sources []models.SaDatasetSource
|
|
if err := s.db.Where("created_by = ?", 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{
|
|
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 created_by = ?", 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 created_by = ?", 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) QueueSync(userID uint) ([]models.SaDatasetCron, error) {
|
|
result := make([]models.SaDatasetCron, 0)
|
|
err := s.db.Transaction(func(tx *gorm.DB) error {
|
|
var sources []models.SaDatasetSource
|
|
if err := tx.Where("created_by = ? AND enabled = ?", userID, true).Order("id asc").Find(&sources).Error; err != nil {
|
|
return err
|
|
}
|
|
now := time.Now().UTC()
|
|
for _, source := range sources {
|
|
var cron models.SaDatasetCron
|
|
err := tx.Where("source_id = ? AND created_by = ? AND status = ?", source.ID, userID, "pending").First(&cron).Error
|
|
if err == nil {
|
|
result = append(result, cron)
|
|
continue
|
|
}
|
|
if !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return err
|
|
}
|
|
cron = models.SaDatasetCron{
|
|
SourceID: source.ID, CreatedBy: userID, Schedule: "@once",
|
|
Status: "pending", Enabled: true, NextRunAt: &now,
|
|
}
|
|
if err := tx.Create(&cron).Error; err != nil {
|
|
return err
|
|
}
|
|
result = append(result, cron)
|
|
}
|
|
return nil
|
|
})
|
|
return result, err
|
|
}
|
|
|
|
func (s *Service) findOwnedSource(userID uint, identity string) (*models.SaDatasetSource, error) {
|
|
var source models.SaDatasetSource
|
|
err := s.db.Where("identity = ? AND created_by = ?", 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 != ""
|
|
}
|
|
|
|
func utcOptionalTime(value *time.Time) *time.Time {
|
|
if value == nil {
|
|
return nil
|
|
}
|
|
result := value.UTC()
|
|
return &result
|
|
}
|