diff --git a/apps/web_v1/scripts/visual-check.mjs b/apps/web_v1/scripts/visual-check.mjs
index 8f604f0..8098751 100644
--- a/apps/web_v1/scripts/visual-check.mjs
+++ b/apps/web_v1/scripts/visual-check.mjs
@@ -265,16 +265,16 @@ await page.route('http://localhost:9150/api/v1/**', async (route) => {
}
if (url.pathname === '/api/v1/dataset-crons/sync') {
await route.fulfill({
- status: 202,
+ status: 200,
json: visualDatasetSources.map((source, index) => ({
id: `019b0000-0000-7000-8000-${String(70 + index).padStart(12, '0')}`,
sourceId: source.id,
schedule: '@once',
- status: 'pending',
- enabled: true,
- nextRunAt: new Date().toISOString(),
- lastRunAt: null,
- lastResult: '',
+ status: 'completed',
+ enabled: false,
+ nextRunAt: null,
+ lastRunAt: new Date().toISOString(),
+ lastResult: 'format=rss fetched=2 inserted=0',
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
})),
diff --git a/apps/web_v1/src/pages/workspace-explore.tsx b/apps/web_v1/src/pages/workspace-explore.tsx
index 4e92f84..468a1ca 100644
--- a/apps/web_v1/src/pages/workspace-explore.tsx
+++ b/apps/web_v1/src/pages/workspace-explore.tsx
@@ -231,12 +231,20 @@ export function WorkspaceExplorePage({
try {
const crons = await onQueueSync()
if (crons.length === 0) {
- Message.info('暂无已启用的数据源')
+ Message.info('暂无已启用的 RSS 数据源')
} else {
- Message.success(`已提交 ${crons.length} 个采集任务`)
+ const failed = crons.filter((cron) => cron.status === 'failed').length
+ const [nextSources, nextItems] = await Promise.all([onListSources(), onListItems()])
+ setSourceRecords(nextSources)
+ setItems(nextItems)
+ if (failed > 0) {
+ Message.warning(`${crons.length - failed} 个数据源采集完成，${failed} 个失败`)
+ } else {
+ Message.success(`已完成 ${crons.length} 个数据源采集任务`)
+ }
}
} catch (error) {
- Message.error(error instanceof Error ? error.message : '同步任务提交失败')
+ Message.error(error instanceof Error ? error.message : '数据源采集失败')
} finally {
setSyncing(false)
}
diff --git a/backend/internal/logic/dataset/feed.go b/backend/internal/logic/dataset/feed.go
new file mode 100644
index 0000000..9a0dd72
--- /dev/null
+++ b/backend/internal/logic/dataset/feed.go
@@ -0,0 +1,347 @@
+package dataset
+
+import (
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "encoding/xml"
+ "errors"
+ "fmt"
+ "html"
+ "io"
+ "net"
+ "net/http"
+ "net/url"
+ "regexp"
+ "strings"
+ "time"
+)
+
+const maxFeedBytes = 5 << 20
+
+var (
+ ErrFeedFormat = errors.New("unsupported feed format")
+ ErrFeedTooLarge = errors.New("feed response is too large")
+ ErrUnsafeFeedURL = errors.New("feed url resolves to a private or local address")
+ htmlTagPattern = regexp.MustCompile(`<[^>]*>`)
+)
+
+type FeedItem struct {
+ ExternalID string
+ Title string
+ Summary string
+ Content string
+ URL string
+ PublishedAt *time.Time
+}
+
+type ParsedFeed struct {
+ Format string
+ Items []FeedItem
+}
+
+type FeedFetcher interface {
+ Fetch(context.Context, string) (ParsedFeed, error)
+}
+
+type HTTPFeedFetcher struct {
+ client *http.Client
+}
+
+func NewHTTPFeedFetcher() *HTTPFeedFetcher {
+ transport := &http.Transport{
+ DialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
+ host, port, err := net.SplitHostPort(address)
+ if err != nil {
+ return nil, err
+ }
+ ips, err := net.DefaultResolver.LookupIPAddr(ctx, host)
+ if err != nil {
+ return nil, err
+ }
+ dialer := &net.Dialer{Timeout: 10 * time.Second}
+ var lastErr error
+ for _, candidate := range ips {
+ if !isPublicIP(candidate.IP) {
+ continue
+ }
+ connection, err := dialer.DialContext(ctx, network, net.JoinHostPort(candidate.IP.String(), port))
+ if err == nil {
+ return connection, nil
+ }
+ lastErr = err
+ }
+ if lastErr != nil {
+ return nil, lastErr
+ }
+ return nil, ErrUnsafeFeedURL
+ },
+ TLSHandshakeTimeout: 10 * time.Second,
+ ResponseHeaderTimeout: 15 * time.Second,
+ }
+ fetcher := &HTTPFeedFetcher{}
+ fetcher.client = &http.Client{
+ Transport: transport,
+ Timeout: 30 * time.Second,
+ CheckRedirect: func(request *http.Request, via []*http.Request) error {
+ if len(via) >= 5 {
+ return errors.New("too many feed redirects")
+ }
+ return validateFeedURL(request.URL)
+ },
+ }
+ return fetcher
+}
+
+func (f *HTTPFeedFetcher) Fetch(ctx context.Context, value string) (ParsedFeed, error) {
+ parsedURL, err := url.Parse(value)
+ if err != nil {
+ return ParsedFeed{}, err
+ }
+ if err := validateFeedURL(parsedURL); err != nil {
+ return ParsedFeed{}, err
+ }
+ request, err := http.NewRequestWithContext(ctx, http.MethodGet, parsedURL.String(), nil)
+ if err != nil {
+ return ParsedFeed{}, err
+ }
+ request.Header.Set("Accept", "application/feed+json, application/json, application/rss+xml, application/atom+xml, application/xml, text/xml;q=0.9")
+ request.Header.Set("User-Agent", "SenlinAI-Agent/1.0")
+ response, err := f.client.Do(request)
+ if err != nil {
+ return ParsedFeed{}, err
+ }
+ defer response.Body.Close()
+ if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices {
+ return ParsedFeed{}, fmt.Errorf("feed request returned %s", response.Status)
+ }
+ body, err := io.ReadAll(io.LimitReader(response.Body, maxFeedBytes+1))
+ if err != nil {
+ return ParsedFeed{}, err
+ }
+ if len(body) > maxFeedBytes {
+ return ParsedFeed{}, ErrFeedTooLarge
+ }
+ return ParseFeed(body, response.Header.Get("Content-Type"))
+}
+
+func ParseFeed(body []byte, contentType string) (ParsedFeed, error) {
+ body = bytes.TrimSpace(bytes.TrimPrefix(body, []byte{0xef, 0xbb, 0xbf}))
+ if len(body) == 0 {
+ return ParsedFeed{}, ErrFeedFormat
+ }
+ lowerContentType := strings.ToLower(contentType)
+ if body[0] == '{' || strings.Contains(lowerContentType, "json") {
+ if feed, err := parseJSONFeed(body); err == nil {
+ return feed, nil
+ }
+ }
+ if body[0] == '<' || strings.Contains(lowerContentType, "xml") || strings.Contains(lowerContentType, "rss") || strings.Contains(lowerContentType, "atom") {
+ return parseXMLFeed(body)
+ }
+ return ParsedFeed{}, ErrFeedFormat
+}
+
+type jsonFeedDocument struct {
+ Version string `json:"version"`
+ Items []jsonFeedItem `json:"items"`
+}
+
+type jsonFeedItem struct {
+ ID string `json:"id"`
+ URL string `json:"url"`
+ Title string `json:"title"`
+ Summary string `json:"summary"`
+ ContentText string `json:"content_text"`
+ ContentHTML string `json:"content_html"`
+ DatePublished string `json:"date_published"`
+}
+
+func parseJSONFeed(body []byte) (ParsedFeed, error) {
+ var document jsonFeedDocument
+ if err := json.Unmarshal(body, &document); err != nil {
+ return ParsedFeed{}, err
+ }
+ if document.Version == "" {
+ return ParsedFeed{}, ErrFeedFormat
+ }
+ items := make([]FeedItem, 0, len(document.Items))
+ for _, input := range document.Items {
+ content := strings.TrimSpace(input.ContentText)
+ if content == "" {
+ content = plainText(input.ContentHTML)
+ }
+ item := normalizeFeedItem(FeedItem{
+ ExternalID: input.ID, Title: plainText(input.Title), Summary: plainText(input.Summary),
+ Content: content, URL: input.URL, PublishedAt: parseFeedTime(input.DatePublished),
+ })
+ if item.Title != "" {
+ items = append(items, item)
+ }
+ }
+ return ParsedFeed{Format: "json_feed", Items: items}, nil
+}
+
+type xmlFeedDocument struct {
+ XMLName xml.Name
+ Channel struct {
+ Items []xmlFeedItem `xml:"item"`
+ } `xml:"channel"`
+ Items []xmlFeedItem `xml:"item"`
+ Entries []xmlAtomEntry `xml:"entry"`
+}
+
+type xmlFeedItem struct {
+ Title string `xml:"title"`
+ Link string `xml:"link"`
+ GUID string `xml:"guid"`
+ Description string `xml:"description"`
+ Content string `xml:"encoded"`
+ PubDate string `xml:"pubDate"`
+ Date string `xml:"date"`
+}
+
+type xmlAtomEntry struct {
+ Title string `xml:"title"`
+ ID string `xml:"id"`
+ Links []xmlAtomLink `xml:"link"`
+ Summary string `xml:"summary"`
+ Content string `xml:"content"`
+ Published string `xml:"published"`
+ Updated string `xml:"updated"`
+}
+
+type xmlAtomLink struct {
+ Href string `xml:"href,attr"`
+ Rel string `xml:"rel,attr"`
+}
+
+func parseXMLFeed(body []byte) (ParsedFeed, error) {
+ var document xmlFeedDocument
+ if err := xml.Unmarshal(body, &document); err != nil {
+ return ParsedFeed{}, err
+ }
+ switch strings.ToLower(document.XMLName.Local) {
+ case "rss", "rdf":
+ inputs := document.Channel.Items
+ if len(inputs) == 0 {
+ inputs = document.Items
+ }
+ items := make([]FeedItem, 0, len(inputs))
+ for _, input := range inputs {
+ content := plainText(input.Content)
+ if content == "" {
+ content = plainText(input.Description)
+ }
+ item := normalizeFeedItem(FeedItem{
+ ExternalID: input.GUID, Title: plainText(input.Title),
+ Summary: plainText(input.Description), Content: content,
+ URL: strings.TrimSpace(input.Link), PublishedAt: parseFeedTime(firstNonEmpty(input.PubDate, input.Date)),
+ })
+ if item.Title != "" {
+ items = append(items, item)
+ }
+ }
+ return ParsedFeed{Format: "rss", Items: items}, nil
+ case "feed":
+ items := make([]FeedItem, 0, len(document.Entries))
+ for _, input := range document.Entries {
+ item := normalizeFeedItem(FeedItem{
+ ExternalID: input.ID, Title: plainText(input.Title),
+ Summary: plainText(input.Summary), Content: plainText(input.Content),
+ URL: atomLink(input.Links), PublishedAt: parseFeedTime(firstNonEmpty(input.Published, input.Updated)),
+ })
+ if item.Title != "" {
+ items = append(items, item)
+ }
+ }
+ return ParsedFeed{Format: "atom", Items: items}, nil
+ default:
+ return ParsedFeed{}, ErrFeedFormat
+ }
+}
+
+func normalizeFeedItem(item FeedItem) FeedItem {
+ item.Title = truncateRunes(strings.TrimSpace(item.Title), 500)
+ item.Summary = strings.TrimSpace(item.Summary)
+ item.Content = strings.TrimSpace(item.Content)
+ item.URL = truncateRunes(strings.TrimSpace(item.URL), 2048)
+ key := firstNonEmpty(strings.TrimSpace(item.ExternalID), item.URL)
+ if key == "" {
+ published := ""
+ if item.PublishedAt != nil {
+ published = item.PublishedAt.UTC().Format(time.RFC3339Nano)
+ }
+ key = item.Title + "\x00" + published
+ }
+ sum := sha256.Sum256([]byte(key))
+ item.ExternalID = hex.EncodeToString(sum[:])
+ return item
+}
+
+func parseFeedTime(value string) *time.Time {
+ value = strings.TrimSpace(value)
+ if value == "" {
+ return nil
+ }
+ layouts := []string{
+ time.RFC3339Nano, time.RFC3339, time.RFC1123Z, time.RFC1123,
+ time.RFC822Z, time.RFC822, time.RFC850, time.ANSIC,
+ "Mon, 2 Jan 2006 15:04:05 -0700",
+ }
+ for _, layout := range layouts {
+ if parsed, err := time.Parse(layout, value); err == nil {
+ utc := parsed.UTC()
+ return &utc
+ }
+ }
+ return nil
+}
+
+func plainText(value string) string {
+ value = htmlTagPattern.ReplaceAllString(value, " ")
+ return strings.Join(strings.Fields(html.UnescapeString(value)), " ")
+}
+
+func atomLink(links []xmlAtomLink) string {
+ for _, link := range links {
+ if link.Rel == "" || link.Rel == "alternate" {
+ return strings.TrimSpace(link.Href)
+ }
+ }
+ if len(links) > 0 {
+ return strings.TrimSpace(links[0].Href)
+ }
+ return ""
+}
+
+func firstNonEmpty(values ...string) string {
+ for _, value := range values {
+ if strings.TrimSpace(value) != "" {
+ return value
+ }
+ }
+ return ""
+}
+
+func validateFeedURL(value *url.URL) error {
+ if value == nil || (value.Scheme != "http" && value.Scheme != "https") || value.Hostname() == "" || value.User != nil {
+ return ErrURLInvalid
+ }
+ if ip := net.ParseIP(value.Hostname()); ip != nil && !isPublicIP(ip) {
+ return ErrUnsafeFeedURL
+ }
+ return nil
+}
+
+func isPublicIP(ip net.IP) bool {
+ return ip != nil &&
+ !ip.IsLoopback() &&
+ !ip.IsPrivate() &&
+ !ip.IsLinkLocalUnicast() &&
+ !ip.IsLinkLocalMulticast() &&
+ !ip.IsUnspecified() &&
+ !ip.IsMulticast()
+}
diff --git a/backend/internal/logic/dataset/feed_test.go b/backend/internal/logic/dataset/feed_test.go
new file mode 100644
index 0000000..7e72ec2
--- /dev/null
+++ b/backend/internal/logic/dataset/feed_test.go
@@ -0,0 +1,126 @@
+package dataset
+
+import (
+ "context"
+ "errors"
+ "net/url"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+ "senlinai-agent/backend/internal/models"
+)
+
+func TestParseFeedDetectsJSONFeedFromContent(t *testing.T) {
+ feed, err := ParseFeed([]byte(`{
+ "version": "https://jsonfeed.org/version/1.1",
+ "items": [{
+ "id": "json-1",
+ "url": "https://example.com/json-1",
+ "title": "JSON item",
+ "summary": "JSON summary",
+ "content_html": "JSON content
",
+ "date_published": "2026-07-23T08:00:00Z"
+ }]
+ }`), "text/plain")
+
+ require.NoError(t, err)
+ require.Equal(t, "json_feed", feed.Format)
+ require.Len(t, feed.Items, 1)
+ require.Equal(t, "JSON item", feed.Items[0].Title)
+ require.Equal(t, "JSON content", feed.Items[0].Content)
+ require.Equal(t, "https://example.com/json-1", feed.Items[0].URL)
+ require.NotEmpty(t, feed.Items[0].ExternalID)
+ require.NotNil(t, feed.Items[0].PublishedAt)
+}
+
+func TestParseFeedDetectsRSSXML(t *testing.T) {
+ feed, err := ParseFeed([]byte(`
+
+
+ -
+ RSS item
+ https://example.com/rss-1
+ rss-1
+ RSS summary]]>
+ RSS content]]>
+ Thu, 23 Jul 2026 08:00:00 +0000
+
+
+ `), "application/xml")
+
+ require.NoError(t, err)
+ require.Equal(t, "rss", feed.Format)
+ require.Len(t, feed.Items, 1)
+ require.Equal(t, "RSS summary", feed.Items[0].Summary)
+ require.Equal(t, "RSS content", feed.Items[0].Content)
+ require.NotNil(t, feed.Items[0].PublishedAt)
+}
+
+func TestParseFeedDetectsAtomXML(t *testing.T) {
+ feed, err := ParseFeed([]byte(`
+
+
+ Atom item
+ atom-1
+
+ Atom summary
+ Atom content
+ 2026-07-23T08:00:00Z
+
+ `), "application/atom+xml")
+
+ require.NoError(t, err)
+ require.Equal(t, "atom", feed.Format)
+ require.Len(t, feed.Items, 1)
+ require.Equal(t, "https://example.com/atom-1", feed.Items[0].URL)
+ require.Equal(t, "Atom content", feed.Items[0].Content)
+}
+
+func TestParseFeedRejectsUnknownContent(t *testing.T) {
+ _, err := ParseFeed([]byte("not a feed"), "text/plain")
+ require.ErrorIs(t, err, ErrFeedFormat)
+}
+
+func TestValidateFeedURLRejectsLocalAddressesAndCredentials(t *testing.T) {
+ for _, value := range []string{
+ "http://127.0.0.1/feed",
+ "http://[::1]/feed",
+ "http://169.254.169.254/latest/meta-data",
+ "https://user:password@example.com/feed",
+ } {
+ parsed, err := url.Parse(value)
+ require.NoError(t, err)
+ require.Error(t, validateFeedURL(parsed), value)
+ }
+
+ parsed, err := url.Parse("https://rsshub.app/cls/hot")
+ require.NoError(t, err)
+ require.NoError(t, validateFeedURL(parsed))
+}
+
+func TestSyncSourcesRecordsFeedFailure(t *testing.T) {
+ database := newDatasetTestDatabase(t)
+ user := createDatasetTestUser(t, database, "feed-owner@example.com")
+ source := models.SaDatasetSource{
+ CreatedBy: user.ID,
+ Name: "Broken feed",
+ Kind: "rss",
+ URL: "https://example.com/feed",
+ Enabled: true,
+ }
+ require.NoError(t, database.Create(&source).Error)
+
+ crons, err := newServiceWithFetcher(database, failingFeedFetcher{}).SyncSources(context.Background(), user.ID)
+
+ require.NoError(t, err)
+ require.Len(t, crons, 1)
+ require.Equal(t, "failed", crons[0].Status)
+ require.False(t, crons[0].Enabled)
+ require.Contains(t, crons[0].LastResult, "feed unavailable")
+}
+
+type failingFeedFetcher struct{}
+
+func (failingFeedFetcher) Fetch(context.Context, string) (ParsedFeed, error) {
+ return ParsedFeed{}, errors.New("feed unavailable")
+}
diff --git a/backend/internal/logic/dataset/handlers.go b/backend/internal/logic/dataset/handlers.go
index 72130f5..7bc8f66 100644
--- a/backend/internal/logic/dataset/handlers.go
+++ b/backend/internal/logic/dataset/handlers.go
@@ -272,7 +272,7 @@ func (h *Handler) queueSync(c *gin.Context) {
if !ok {
return
}
- crons, err := h.service.QueueSync(userID)
+ crons, err := h.service.SyncSources(c.Request.Context(), userID)
if err != nil {
writeError(c, err)
return
@@ -281,7 +281,7 @@ func (h *Handler) queueSync(c *gin.Context) {
for _, cron := range crons {
result = append(result, cronDTO(cron))
}
- c.JSON(http.StatusAccepted, result)
+ c.JSON(http.StatusOK, result)
}
func currentUser(c *gin.Context) (uint, bool) {
diff --git a/backend/internal/logic/dataset/handlers_test.go b/backend/internal/logic/dataset/handlers_test.go
index f384e29..dec7674 100644
--- a/backend/internal/logic/dataset/handlers_test.go
+++ b/backend/internal/logic/dataset/handlers_test.go
@@ -2,6 +2,7 @@ package dataset
import (
"bytes"
+ "context"
"encoding/json"
"fmt"
"net/http"
@@ -63,17 +64,21 @@ func TestDatasetAPIConnectsSourcesItemsAndSyncTasks(t *testing.T) {
require.True(t, item.Starred)
firstSync := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-crons/sync", map[string]any{})
- require.Equal(t, http.StatusAccepted, firstSync.Code)
+ require.Equal(t, http.StatusOK, firstSync.Code)
var firstCrons []CronDTO
require.NoError(t, json.Unmarshal(firstSync.Body.Bytes(), &firstCrons))
require.Len(t, firstCrons, 1)
- require.Equal(t, "pending", firstCrons[0].Status)
+ require.Equal(t, "completed", firstCrons[0].Status)
+ require.Contains(t, firstCrons[0].LastResult, "format=rss")
secondSync := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-crons/sync", map[string]any{})
- require.Equal(t, http.StatusAccepted, secondSync.Code)
+ require.Equal(t, http.StatusOK, secondSync.Code)
var secondCrons []CronDTO
require.NoError(t, json.Unmarshal(secondSync.Body.Bytes(), &secondCrons))
- require.Equal(t, firstCrons[0].ID, secondCrons[0].ID)
+ require.NotEqual(t, firstCrons[0].ID, secondCrons[0].ID)
+ var collectedItems []models.SaDatasetItem
+ require.NoError(t, database.Where("source_id = ?", 1).Find(&collectedItems).Error)
+ require.Len(t, collectedItems, 2)
otherRouter := datasetTestRouter(database, other.ID)
otherSources := performDatasetRequest(t, otherRouter, http.MethodGet, "/api/v1/dataset-sources", nil)
@@ -117,10 +122,25 @@ func datasetTestRouter(database *gorm.DB, userID uint) http.Handler {
return httpx.NewProtectedRouter(
config.Config{Env: "test"},
func(string) (uint, error) { return userID, nil },
- NewHandler(NewService(database)),
+ NewHandler(newServiceWithFetcher(database, staticFeedFetcher{})),
)
}
+type staticFeedFetcher struct{}
+
+func (staticFeedFetcher) Fetch(context.Context, string) (ParsedFeed, error) {
+ return ParsedFeed{
+ Format: "rss",
+ Items: []FeedItem{{
+ ExternalID: "stable-feed-item",
+ Title: "Collected item",
+ Summary: "Collected summary",
+ Content: "Collected content",
+ URL: "https://example.com/collected",
+ }},
+ }, nil
+}
+
func performDatasetRequest(t *testing.T, router http.Handler, method, path string, body any) *httptest.ResponseRecorder {
t.Helper()
var encoded []byte
diff --git a/backend/internal/logic/dataset/service.go b/backend/internal/logic/dataset/service.go
index fef6b54..2a84415 100644
--- a/backend/internal/logic/dataset/service.go
+++ b/backend/internal/logic/dataset/service.go
@@ -1,7 +1,9 @@
package dataset
import (
+ "context"
"errors"
+ "fmt"
"net/url"
"strings"
"time"
@@ -19,7 +21,8 @@ var (
)
type Service struct {
- db *gorm.DB
+ db *gorm.DB
+ fetcher FeedFetcher
}
type SourceInput struct {
@@ -51,7 +54,11 @@ type SourceRecord struct {
}
func NewService(database *gorm.DB) *Service {
- return &Service{db: database}
+ 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) {
@@ -208,36 +215,111 @@ func (s *Service) ListCrons(userID uint) ([]models.SaDatasetCron, error) {
return crons, nil
}
-func (s *Service) QueueSync(userID uint) ([]models.SaDatasetCron, error) {
+func (s *Service) SyncSources(ctx context.Context, 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
- }
+ var sources []models.SaDatasetSource
+ if err := s.db.Where("created_by = ? AND enabled = ? AND kind = ?", userID, true, "rss").
+ Order("id asc").Find(&sources).Error; err != nil {
+ return nil, err
+ }
+ for _, source := range sources {
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
+ 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)
+ 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 {
+ var count int64
+ if err := tx.Model(&models.SaDatasetItem{}).
+ Where("source_id = ? AND external_id = ?", source.ID, input.ExternalID).
+ Count(&count).Error; err != nil {
+ return err
+ }
+ if count > 0 {
+ continue
+ }
+ item := models.SaDatasetItem{
+ SourceID: source.ID, CreatedBy: userID, ExternalID: input.ExternalID,
+ Title: input.Title, Summary: input.Summary, Content: input.Content,
+ URL: input.URL, Status: "unread", PublishedAt: input.PublishedAt,
+ }
+ if err := tx.Create(&item).Error; err != nil {
+ return err
+ }
+ inserted++
}
return nil
})
- return result, err
+ 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) {
@@ -272,7 +354,8 @@ func normalizeSourceInput(input SourceInput) (SourceInput, error) {
func validHTTPURL(value string) bool {
parsed, err := url.ParseRequestURI(value)
- return err == nil && (parsed.Scheme == "http" || parsed.Scheme == "https") && parsed.Host != ""
+ return err == nil && (parsed.Scheme == "http" || parsed.Scheme == "https") &&
+ parsed.Host != "" && parsed.User == nil
}
func utcOptionalTime(value *time.Time) *time.Time {
diff --git a/backend/internal/models/dataset_item.go b/backend/internal/models/dataset_item.go
index a1dab9c..8cf28b0 100644
--- a/backend/internal/models/dataset_item.go
+++ b/backend/internal/models/dataset_item.go
@@ -9,6 +9,7 @@ type SaDatasetItem struct {
SourceIdentity string `gorm:"type:char(36);index"`
CreatedBy uint `gorm:"index;not null"`
CreatedByIdentity string `gorm:"type:char(36);index"`
+ ExternalID string `gorm:"size:64;index"`
Title string `gorm:"size:500;not null"`
Summary string `gorm:"type:text"`
Content string `gorm:"type:text"`