feat: collect JSON and XML feeds

2026-07-23 20:47:54 +08:00
parent 4b38a42607
commit 5da343cc71
8 changed files with 627 additions and 42 deletions
--- 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(),
      })),
--- 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)
    }
--- a/backend/internal/logic/dataset/feed.go
+++ 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()
+}
--- a/backend/internal/logic/dataset/feed_test.go
+++ 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": "<p>JSON <strong>content</strong></p>",
+			"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(`<?xml version="1.0"?>
+	<rss version="2.0">
+		<channel>
+			<item>
+				<title>RSS item</title>
+				<link>https://example.com/rss-1</link>
+				<guid>rss-1</guid>
+				<description><![CDATA[<p>RSS summary</p>]]></description>
+				<content:encoded xmlns:content="http://purl.org/rss/1.0/modules/content/"><![CDATA[<p>RSS content</p>]]></content:encoded>
+				<pubDate>Thu, 23 Jul 2026 08:00:00 +0000</pubDate>
+			</item>
+		</channel>
+	</rss>`), "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(`<?xml version="1.0"?>
+	<feed xmlns="http://www.w3.org/2005/Atom">
+		<entry>
+			<title>Atom item</title>
+			<id>atom-1</id>
+			<link rel="alternate" href="https://example.com/atom-1"/>
+			<summary>Atom summary</summary>
+			<content>Atom content</content>
+			<updated>2026-07-23T08:00:00Z</updated>
+		</entry>
+	</feed>`), "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")
+}
--- 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) {
--- 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
--- 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"
@@ -20,6 +22,7 @@ var (

 type Service struct {
 	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
+	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
 	}
-		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 {
+		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)
 			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
+
+		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 {
--- 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"`