fix: harden exploration data flow

apps/web_v1/scripts/visual-check.mjs

@@ -35,17 +35,17 @@ const visualExperts = [
 let visualDatasetSources = [
   {
     id: '019b0000-0000-7000-8000-000000000030', name: '手动收集', kind: 'manual', url: '', iconUrl: '',
-    description: '手动收集的文章与线索', enabled: true, lastSyncedAt: null, itemCount: 1,
+    description: '手动收集的文章与线索', enabled: true, builtIn: false, lastSyncedAt: null, itemCount: 1,
     createdAt: '2026-07-22T06:00:00Z', updatedAt: '2026-07-22T06:00:00Z',
   },
   {
     id: '019b0000-0000-7000-8000-000000000031', name: '需求文档', kind: 'link', url: 'https://example.com/requirements', iconUrl: '',
-    description: '产品需求与业务文档', enabled: true, lastSyncedAt: null, itemCount: 1,
+    description: '产品需求与业务文档', enabled: true, builtIn: false, lastSyncedAt: null, itemCount: 1,
     createdAt: '2026-07-22T06:00:00Z', updatedAt: '2026-07-22T06:00:00Z',
   },
   {
     id: '019b0000-0000-7000-8000-000000000032', name: '架构讨论', kind: 'link', url: 'https://example.com/architecture', iconUrl: '',
-    description: '架构方案与系统设计讨论', enabled: true, lastSyncedAt: null, itemCount: 0,
+    description: '架构方案与系统设计讨论', enabled: true, builtIn: false, lastSyncedAt: null, itemCount: 0,
     createdAt: '2026-07-22T06:00:00Z', updatedAt: '2026-07-22T06:00:00Z',
   },
 ]
@@ -53,13 +53,15 @@ let visualDatasetItems = [
   {
     id: '019b0000-0000-7000-8000-000000000040', sourceId: visualDatasetSources[0].id,
     title: '森林项目体验优化需求', summary: '整理工作台信息层级，并沉淀为可跟进计划。', content: '探索数据条目的正文内容。',
-    url: 'https://example.com/article-1', status: 'unread', starred: false, publishedAt: '2026-07-22T08:00:00Z',
+    url: 'https://example.com/article-1',
+    imageUrl: 'data:image/svg+xml,%3Csvg xmlns=%22http://www.w3.org/2000/svg%22 width=%22152%22 height=%22116%22%3E%3Crect width=%22152%22 height=%22116%22 fill=%22%23165DFF%22/%3E%3C/svg%3E',
+    status: 'unread', starred: false, publishedAt: '2026-07-22T08:00:00Z',
     createdAt: '2026-07-22T08:00:00Z', updatedAt: '2026-07-22T08:00:00Z',
   },
   {
     id: '019b0000-0000-7000-8000-000000000041', sourceId: visualDatasetSources[1].id,
     title: '项目资料组织方案', summary: '围绕项目上下文统一组织任务、笔记、资料和 AI 会话。', content: '资料组织方案正文。',
-    url: 'https://example.com/article-2', status: 'read', starred: true, publishedAt: '2026-07-22T07:00:00Z',
+    url: 'https://example.com/article-2', imageUrl: '', status: 'read', starred: true, publishedAt: '2026-07-22T07:00:00Z',
     createdAt: '2026-07-22T07:00:00Z', updatedAt: '2026-07-22T07:00:00Z',
   },
 ]
@@ -225,6 +227,7 @@ await page.route('http://localhost:9150/api/v1/**', async (route) => {
     const created = {
       id: `019b0000-0000-7000-8000-${String(50 + visualDatasetSources.length).padStart(12, '0')}`,
       ...input,
+      builtIn: false,
       lastSyncedAt: null,
       itemCount: 0,
       createdAt: new Date().toISOString(),
@@ -251,7 +254,24 @@ await page.route('http://localhost:9150/api/v1/**', async (route) => {
     return
   }
   if (url.pathname === '/api/v1/dataset-items' && route.request().method() === 'GET') {
-    await route.fulfill({ json: visualDatasetItems })
+    const sourceId = url.searchParams.get('sourceId')
+    const offset = Number(url.searchParams.get('offset') ?? 0)
+    const limit = Number(url.searchParams.get('limit') ?? 50)
+    const matchingItems = sourceId
+      ? visualDatasetItems.filter((item) => item.sourceId === sourceId)
+      : visualDatasetItems
+    await route.fulfill({
+      json: {
+        items: matchingItems.slice(offset, offset + limit),
+        total: matchingItems.length,
+        offset,
+        limit,
+      },
+    })
+    return
+  }
+  if (url.pathname.endsWith('/deposit') && route.request().method() === 'POST') {
+    await route.fulfill({ status: 201, json: { noteId: '019b0000-0000-7000-8000-000000000099', projectId } })
     return
   }
   if (url.pathname.startsWith('/api/v1/dataset-items/') && route.request().method() === 'PATCH') {
@@ -263,20 +283,13 @@ await page.route('http://localhost:9150/api/v1/**', async (route) => {
     await route.fulfill({ json: updated })
     return
   }
-  if (url.pathname === '/api/v1/dataset-crons/sync') {
+  if (url.pathname === '/api/v1/dataset-sources/sync') {
     await route.fulfill({
-      status: 202,
+      status: 200,
-      json: visualDatasetSources.map((source, index) => ({
+      json: visualDatasetSources.map((source) => ({
-        id: `019b0000-0000-7000-8000-${String(70 + index).padStart(12, '0')}`,
         sourceId: source.id,
-        schedule: '@once',
+        status: 'completed',
-        status: 'pending',
+        result: 'format=rss fetched=2 inserted=0',
-        enabled: true,
-        nextRunAt: new Date().toISOString(),
-        lastRunAt: null,
-        lastResult: '',
-        createdAt: new Date().toISOString(),
-        updatedAt: new Date().toISOString(),
       })),
     })
     return
@@ -415,7 +428,9 @@ const editSourceModalCheck = await page.evaluate(() => ({
 }))
 await page.locator('.explore-source-modal input').first().fill('手动整理')
 await page.locator('.explore-source-modal .arco-modal-footer .arco-btn-primary').click()
-const editedSourceVisible = await page.locator('.explore-source-card', { hasText: '手动整理' }).count() === 1
+const editedSourceCard = page.locator('.explore-source-card', { hasText: '手动整理' })
+await editedSourceCard.waitFor({ state: 'visible' })
+const editedSourceVisible = await editedSourceCard.count() === 1
 
 await page.locator('.project-button').first().click()
 await page.waitForTimeout(500)

apps/web_v1/src/App.css

@@ -868,6 +868,19 @@
   cursor: pointer;
 }
 
+.explore-article-item.has-image {
+  grid-template-columns: 30px minmax(0, 1fr) 76px;
+}
+
+.explore-article-thumbnail {
+  width: 76px;
+  height: 58px;
+  align-self: center;
+  border-radius: 8px;
+  background: var(--senlin-panel-muted);
+  object-fit: cover;
+}
+
 .explore-article-item:hover,
 .explore-article-item.active {
   background: color-mix(in srgb, var(--senlin-primary) 8%, var(--senlin-panel));
@@ -965,6 +978,15 @@
   margin-bottom: 24px;
 }
 
+.explore-article-image {
+  display: block;
+  width: 100%;
+  max-height: 420px;
+  margin: 0 0 24px;
+  border-radius: 12px;
+  object-fit: cover;
+}
+
 .explore-article-content {
   margin: 0;
   color: var(--senlin-text);

apps/web_v1/src/api/dataset.ts

@@ -10,6 +10,7 @@ export type DatasetSourceDTO = {
   iconUrl: string
   description: string
   enabled: boolean
+  builtIn: boolean
   lastSyncedAt: string | null
   itemCount: number
   createdAt: string
@@ -23,6 +24,7 @@ export type DatasetItemDTO = {
   summary: string
   content: string
   url: string
+  imageUrl: string
   status: 'unread' | 'read' | 'archived'
   starred: boolean
   publishedAt: string | null
@@ -30,17 +32,22 @@ export type DatasetItemDTO = {
   updatedAt: string
 }
 
-export type DatasetCronDTO = {
+export type DatasetSyncResultDTO = {
-  id: string
   sourceId: string
-  schedule: string
   status: string
-  enabled: boolean
+  result: string
-  nextRunAt: string | null
+}
-  lastRunAt: string | null
+
-  lastResult: string
+export type DatasetItemPageDTO = {
-  createdAt: string
+  items: DatasetItemDTO[]
-  updatedAt: string
+  total: number
+  limit: number
+  offset: number
+}
+
+export type DatasetDepositDTO = {
+  noteId: string
+  projectId: string
 }
 
 export type DatasetSourceInput = {
@@ -53,8 +60,8 @@ export type DatasetSourceInput = {
 }
 
 export type DatasetItemUpdate = {
-  status: DatasetItemDTO['status']
+  status?: DatasetItemDTO['status']
-  starred: boolean
+  starred?: boolean
 }
 
 export function listDatasetSources(session: ApiSession, signal?: AbortSignal) {
@@ -85,9 +92,16 @@ export function deleteDatasetSource(session: ApiSession, sourceId: string) {
   })
 }
 
-export function listDatasetItems(session: ApiSession, sourceId?: string, signal?: AbortSignal) {
+export function listDatasetItems(
-  const query = sourceId ? `?sourceId=${encodeURIComponent(sourceId)}` : ''
+  session: ApiSession,
-  return apiRequest<DatasetItemDTO[]>(`/api/v1/dataset-items${query}`, { token: session.token, signal })
+  sourceId?: string,
+  offset = 0,
+  limit = 50,
+  signal?: AbortSignal,
+) {
+  const params = new URLSearchParams({ offset: String(offset), limit: String(limit) })
+  if (sourceId) params.set('sourceId', sourceId)
+  return apiRequest<DatasetItemPageDTO>(`/api/v1/dataset-items?${params}`, { token: session.token, signal })
 }
 
 export function updateDatasetItem(session: ApiSession, itemId: string, input: DatasetItemUpdate) {
@@ -98,10 +112,18 @@ export function updateDatasetItem(session: ApiSession, itemId: string, input: Da
   })
 }
 
-export function queueDatasetSync(session: ApiSession) {
+export function syncDatasetSources(session: ApiSession) {
-  return apiRequest<DatasetCronDTO[]>('/api/v1/dataset-crons/sync', {
+  return apiRequest<DatasetSyncResultDTO[]>('/api/v1/dataset-sources/sync', {
     method: 'POST',
     token: session.token,
     body: {},
   })
 }
+
+export function depositDatasetItem(session: ApiSession, itemId: string, projectId: string) {
+  return apiRequest<DatasetDepositDTO>(`/api/v1/dataset-items/${itemId}/deposit`, {
+    method: 'POST',
+    token: session.token,
+    body: { projectId },
+  })
+}

apps/web_v1/src/app/App.tsx

@@ -7,9 +7,10 @@ import { createAISession, listAIExperts, listAISessions, type CreateAISessionInp
 import {
   createDatasetSource,
   deleteDatasetSource,
+  depositDatasetItem,
   listDatasetItems,
   listDatasetSources,
-  queueDatasetSync,
+  syncDatasetSources,
   updateDatasetItem,
   updateDatasetSource,
   type DatasetItemUpdate,
@@ -75,9 +76,9 @@ function App() {
     return listDatasetSources(session, signal)
   }, [session])
 
-  const handleListDatasetItems = useCallback((signal?: AbortSignal) => {
+  const handleListDatasetItems = useCallback((sourceId?: string, offset = 0, limit = 50, signal?: AbortSignal) => {
     if (!session) return Promise.reject(new Error('未登录'))
-    return listDatasetItems(session, undefined, signal)
+    return listDatasetItems(session, sourceId, offset, limit, signal)
   }, [session])
 
   const handleCreateDatasetSource = useCallback((input: DatasetSourceInput) => {
@@ -100,11 +101,27 @@ function App() {
     return updateDatasetItem(session, itemId, input)
   }, [session])
 
-  const handleQueueDatasetSync = useCallback(() => {
+  const handleSyncDatasetSources = useCallback(() => {
     if (!session) return Promise.reject(new Error('未登录'))
-    return queueDatasetSync(session)
+    return syncDatasetSources(session)
   }, [session])
 
+  const handleDepositDatasetItem = useCallback(async (itemId: string, projectId: string) => {
+    if (!session) throw new Error('未登录')
+    const result = await depositDatasetItem(session, itemId, projectId)
+    const workspaceIndex = workspaces.findIndex((workspace) => workspace.project.id === projectId)
+    if (workspaceIndex >= 0) {
+      try {
+        const payload = await fetchProjectWorkspace(session, projectId)
+        const refreshed = mapWorkspace(payload, workspaceIndex)
+        setWorkspaces((current) => current.map((workspace) => workspace.project.id === projectId ? refreshed : workspace))
+      } catch {
+        return { ...result, refreshFailed: true }
+      }
+    }
+    return result
+  }, [session, workspaces])
+
   const dark = theme === 'dark'
   const activeWorkspace = workspaces.find((workspace) => workspace.project.id === activeProjectID) ?? workspaces[0]
   const activeTagOptions = activeWorkspace?.tags.filter((tag) => tag !== 'all' && tag !== '全部') ?? []
@@ -417,7 +434,8 @@ function App() {
             onUpdateDatasetSource={handleUpdateDatasetSource}
             onDeleteDatasetSource={handleDeleteDatasetSource}
             onUpdateDatasetItem={handleUpdateDatasetItem}
-            onQueueDatasetSync={handleQueueDatasetSync}
+            onSyncDatasetSources={handleSyncDatasetSources}
+            onDepositDatasetItem={handleDepositDatasetItem}
           />
         ) : (
           <Spin loading />

apps/web_v1/src/pages/workspace-explore.tsx

@@ -1,6 +1,6 @@
 import { useEffect, useMemo, useState } from 'react'
 import type { ReactNode } from 'react'
-import { Alert, Button, Card, Empty, Grid, Input, Message, Modal, Select, Space, Spin, Switch, Typography } from '@arco-design/web-react'
+import { Alert, Button, Card, Empty, Grid, Input, Message, Modal, Space, Spin, Switch, Typography } from '@arco-design/web-react'
 import {
   IconBook,
   IconCheckCircle,
@@ -14,17 +14,20 @@ import {
   IconStorage,
 } from '@arco-design/web-react/icon'
 import type {
-  DatasetCronDTO,
+  DatasetDepositDTO,
   DatasetItemDTO,
+  DatasetItemPageDTO,
   DatasetItemUpdate,
   DatasetSourceDTO,
   DatasetSourceInput,
   DatasetSourceKind,
+  DatasetSyncResultDTO,
 } from '../api/dataset'
 
 const { Row, Col } = Grid
 const { Title, Text, Paragraph } = Typography
 const { TextArea } = Input
+const ITEM_PAGE_SIZE = 50
 
 type DataSourceCard = DatasetSourceDTO & {
   icon: ReactNode
@@ -42,7 +45,7 @@ type SourceDraft = {
 
 const EMPTY_SOURCE_DRAFT: SourceDraft = {
   name: '',
-  kind: 'link',
+  kind: 'rss',
   url: '',
   iconUrl: '',
   description: '',
@@ -50,26 +53,31 @@ const EMPTY_SOURCE_DRAFT: SourceDraft = {
 }
 
 export function WorkspaceExplorePage({
-  onSelectItem,
+  activeProjectID,
+  activeProjectName,
   onListSources,
   onListItems,
   onCreateSource,
   onUpdateSource,
   onDeleteSource,
   onUpdateItem,
-  onQueueSync,
+  onSyncSources,
+  onDepositItem,
 }: {
-  onSelectItem: (title: string) => void
+  activeProjectID: string
+  activeProjectName: string
   onListSources: (signal?: AbortSignal) => Promise<DatasetSourceDTO[]>
-  onListItems: (signal?: AbortSignal) => Promise<DatasetItemDTO[]>
+  onListItems: (sourceId?: string, offset?: number, limit?: number, signal?: AbortSignal) => Promise<DatasetItemPageDTO>
   onCreateSource: (input: DatasetSourceInput) => Promise<DatasetSourceDTO>
   onUpdateSource: (sourceId: string, input: DatasetSourceInput) => Promise<DatasetSourceDTO>
   onDeleteSource: (sourceId: string) => Promise<void>
   onUpdateItem: (itemId: string, input: DatasetItemUpdate) => Promise<DatasetItemDTO>
-  onQueueSync: () => Promise<DatasetCronDTO[]>
+  onSyncSources: () => Promise<DatasetSyncResultDTO[]>
+  onDepositItem: (itemId: string, projectId: string) => Promise<DatasetDepositDTO & { refreshFailed?: boolean }>
 }) {
   const [sourceRecords, setSourceRecords] = useState<DatasetSourceDTO[]>([])
   const [items, setItems] = useState<DatasetItemDTO[]>([])
+  const [itemTotal, setItemTotal] = useState(0)
   const [activeSourceID, setActiveSourceID] = useState('all')
   const [activeItemID, setActiveItemID] = useState<string | null>(null)
   const [sourceModalMode, setSourceModalMode] = useState<'add' | 'edit' | null>(null)
@@ -78,17 +86,18 @@ export function WorkspaceExplorePage({
   const [sourceError, setSourceError] = useState('')
   const [loadError, setLoadError] = useState('')
   const [loading, setLoading] = useState(true)
+  const [itemsLoading, setItemsLoading] = useState(true)
   const [saving, setSaving] = useState(false)
   const [syncing, setSyncing] = useState(false)
+  const [depositingItemID, setDepositingItemID] = useState<string | null>(null)
 
   useEffect(() => {
     const controller = new AbortController()
     setLoading(true)
     setLoadError('')
-    Promise.all([onListSources(controller.signal), onListItems(controller.signal)])
+    onListSources(controller.signal)
-      .then(([nextSources, nextItems]) => {
+      .then((nextSources) => {
         setSourceRecords(nextSources)
-        setItems(nextItems)
       })
       .catch((error: unknown) => {
         if (!controller.signal.aborted) {
@@ -99,7 +108,31 @@ export function WorkspaceExplorePage({
         if (!controller.signal.aborted) setLoading(false)
       })
     return () => controller.abort()
-  }, [onListItems, onListSources])
+  }, [onListSources])
+
+  useEffect(() => {
+    const controller = new AbortController()
+    setItemsLoading(true)
+    setLoadError('')
+    setItems([])
+    setItemTotal(0)
+    setActiveItemID(null)
+    const sourceID = activeSourceID === 'all' ? undefined : activeSourceID
+    onListItems(sourceID, 0, ITEM_PAGE_SIZE, controller.signal)
+      .then((page) => {
+        setItems(page.items)
+        setItemTotal(page.total)
+      })
+      .catch((error: unknown) => {
+        if (!controller.signal.aborted) {
+          setLoadError(error instanceof Error ? error.message : '探索内容加载失败')
+        }
+      })
+      .finally(() => {
+        if (!controller.signal.aborted) setItemsLoading(false)
+      })
+    return () => controller.abort()
+  }, [activeSourceID, onListItems])
 
   const sources = useMemo<DataSourceCard[]>(() => {
     const allSource: DataSourceCard = {
@@ -110,8 +143,9 @@ export function WorkspaceExplorePage({
       iconUrl: '',
       description: '全部探索内容',
       enabled: true,
+      builtIn: true,
       lastSyncedAt: null,
-      itemCount: items.length,
+      itemCount: sourceRecords.reduce((total, source) => total + source.itemCount, 0),
       createdAt: '',
       updatedAt: '',
       icon: <IconStorage />,
@@ -122,17 +156,14 @@ export function WorkspaceExplorePage({
       ...sourceRecords.map((source) => ({
         ...source,
         icon: source.iconUrl
-          ? <img className="explore-source-image" src={source.iconUrl} alt="" />
+          ? <img className="explore-source-image" src={source.iconUrl} alt="" loading="lazy" referrerPolicy="no-referrer" />
           : sourceIcon(source.kind),
         color: sourceColor(source.kind),
       })),
     ]
-  }, [items.length, sourceRecords])
+  }, [sourceRecords])
 
-  const filteredItems = useMemo(
+  const filteredItems = items
-    () => activeSourceID === 'all' ? items : items.filter((item) => item.sourceId === activeSourceID),
-    [activeSourceID, items],
-  )
 
   useEffect(() => {
     if (filteredItems.length === 0) {
@@ -147,6 +178,9 @@ export function WorkspaceExplorePage({
   const selected = filteredItems.find((item) => item.id === activeItemID) ?? filteredItems[0]
   const activeSource = sourceByID(sources, activeSourceID)
   const selectedSource = selected ? sourceByID(sources, selected.sourceId) : activeSource
+  const editingSource = editingSourceID
+    ? sourceRecords.find((source) => source.id === editingSourceID)
+    : undefined
 
   function openAddSource() {
     setEditingSourceID(null)
@@ -174,7 +208,7 @@ export function WorkspaceExplorePage({
       name: sourceDraft.name.trim(),
       kind: sourceDraft.kind,
       url: sourceDraft.url.trim(),
-      iconUrl: sourceDraft.iconUrl,
+      iconUrl: sourceDraft.iconUrl.trim(),
       description: sourceDraft.description.trim(),
       enabled: sourceDraft.enabled,
     }
@@ -182,8 +216,8 @@ export function WorkspaceExplorePage({
       setSourceError('请输入数据源名称')
       return
     }
-    if (input.kind !== 'manual' && !input.url) {
+    if (input.kind === 'rss' && !input.url) {
-      setSourceError('链接和 RSS 数据源必须填写地址')
+      setSourceError('RSS 数据源必须填写地址')
       return
     }
     setSaving(true)
@@ -209,14 +243,18 @@ export function WorkspaceExplorePage({
   function deleteSource(source: DataSourceCard) {
     Modal.confirm({
       title: `删除“${source.name}”数据源？`,
-      content: '删除后，该数据源及其采集条目和待执行任务都会移除。',
+      content: '删除后，该数据源及其采集条目都会移除。',
       okButtonProps: { status: 'danger' },
       onOk: async () => {
         try {
           await onDeleteSource(source.id)
           setSourceRecords((current) => current.filter((item) => item.id !== source.id))
           setItems((current) => current.filter((item) => item.sourceId !== source.id))
-          if (activeSourceID === source.id) setActiveSourceID('all')
+          if (activeSourceID === source.id) {
+            setActiveSourceID('all')
+          } else {
+            setItemTotal((current) => Math.max(0, current - source.itemCount))
+          }
           Message.success('数据源已删除')
         } catch (error) {
           Message.error(error instanceof Error ? error.message : '数据源删除失败')
@@ -229,19 +267,52 @@ export function WorkspaceExplorePage({
   async function syncSources() {
     setSyncing(true)
     try {
-      const crons = await onQueueSync()
+      const results = await onSyncSources()
-      if (crons.length === 0) {
+      if (results.length === 0) {
-        Message.info('暂无已启用的数据源')
+        Message.info('暂无已启用的 RSS 数据源')
       } else {
-        Message.success(`已提交 ${crons.length} 个采集任务`)
+        const sourceID = activeSourceID === 'all' ? undefined : activeSourceID
+        const [nextSources, nextPage] = await Promise.all([
+          onListSources(),
+          onListItems(sourceID, 0, ITEM_PAGE_SIZE),
+        ])
+        setSourceRecords(nextSources)
+        setItems(nextPage.items)
+        setItemTotal(nextPage.total)
+        const failedResults = results.filter((result) => result.status === 'failed')
+        if (failedResults.length > 0) {
+          const firstFailure = failedResults[0]
+          const failedSource = sourceRecords.find((source) => source.id === firstFailure.sourceId)
+          const detail = firstFailure.result
+            ? `；${failedSource?.name || '数据源'}：${firstFailure.result.slice(0, 120)}`
+            : ''
+          Message.warning(`${results.length - failedResults.length} 个数据源采集完成，${failedResults.length} 个失败${detail}`)
+        } else {
+          Message.success(`已完成 ${results.length} 个数据源采集任务`)
+        }
       }
     } catch (error) {
-      Message.error(error instanceof Error ? error.message : '同步任务提交失败')
+      Message.error(error instanceof Error ? error.message : '数据源采集失败')
     } finally {
       setSyncing(false)
     }
   }
 
+  async function loadMoreItems() {
+    if (itemsLoading || items.length >= itemTotal) return
+    setItemsLoading(true)
+    try {
+      const sourceID = activeSourceID === 'all' ? undefined : activeSourceID
+      const page = await onListItems(sourceID, items.length, ITEM_PAGE_SIZE)
+      setItems((current) => [...current, ...page.items])
+      setItemTotal(page.total)
+    } catch (error) {
+      Message.error(error instanceof Error ? error.message : '更多探索内容加载失败')
+    } finally {
+      setItemsLoading(false)
+    }
+  }
+
   async function updateSelected(input: DatasetItemUpdate) {
     if (!selected) return
     try {
@@ -252,6 +323,23 @@ export function WorkspaceExplorePage({
     }
   }
 
+  async function depositSelected() {
+    if (!selected || !activeProjectID) return
+    setDepositingItemID(selected.id)
+    try {
+      const result = await onDepositItem(selected.id, activeProjectID)
+      if (result.refreshFailed) {
+        Message.warning(`笔记已沉淀到项目“${activeProjectName}”，但项目数据刷新失败`)
+      } else {
+        Message.success(`已沉淀到项目“${activeProjectName}”的笔记资料`)
+      }
+    } catch (error) {
+      Message.error(error instanceof Error ? error.message : '沉淀到项目失败')
+    } finally {
+      setDepositingItemID(null)
+    }
+  }
+
   return (
     <div className="workspace-explore-page overview-page">
       <div className="overview-head">
@@ -267,7 +355,7 @@ export function WorkspaceExplorePage({
 
       {loadError ? <Alert type="error" content={loadError} /> : null}
 
-      <Spin loading={loading} style={{ width: '100%' }}>
+      <Spin loading={loading || (itemsLoading && items.length === 0)} style={{ width: '100%' }}>
         <Row gutter={10} className="explore-source-row">
           {sources.map((source) => (
             <Col span={6} key={source.id}>
@@ -279,12 +367,14 @@ export function WorkspaceExplorePage({
                 <span className={`explore-source-icon ${source.color}`}>{source.icon}</span>
                 <span className="explore-source-copy">
                   <Text className="explore-source-name">{source.name}</Text>
-                  <Text type="secondary">{source.itemCount} 条</Text>
+                  <Text type="secondary">{sourceSubtitle(source)}</Text>
                 </span>
                 {source.id !== 'all' ? (
                   <span className="explore-source-actions" onClick={(event) => event.stopPropagation()}>
                     <Button aria-label={`编辑${source.name}`} type="text" size="mini" icon={<IconEdit />} onClick={() => openEditSource(source)} />
+                    {!source.builtIn ? (
                       <Button aria-label={`删除${source.name}`} type="text" size="mini" status="danger" icon={<IconDelete />} onClick={() => deleteSource(source)} />
+                    ) : null}
                   </span>
                 ) : null}
               </Card>
@@ -310,11 +400,12 @@ export function WorkspaceExplorePage({
                   return (
                     <button
                       key={item.id}
-                      className={item.id === selected.id ? 'explore-article-item active' : 'explore-article-item'}
+                      className={[
-                      onClick={() => {
+                        'explore-article-item',
-                        setActiveItemID(item.id)
+                        item.id === selected.id ? 'active' : '',
-                        onSelectItem(item.title)
+                        item.imageUrl ? 'has-image' : '',
-                      }}
+                      ].filter(Boolean).join(' ')}
+                      onClick={() => setActiveItemID(item.id)}
                     >
                       <span className={`explore-source-logo ${itemSource.color}`}>{sourceInitial(itemSource.name)}</span>
                       <span className="explore-article-copy">
@@ -324,9 +415,23 @@ export function WorkspaceExplorePage({
                           {item.summary || '暂无摘要'}
                         </Text>
                       </span>
+                      {item.imageUrl ? (
+                        <img
+                          className="explore-article-thumbnail"
+                          src={item.imageUrl}
+                          alt=""
+                          loading="lazy"
+                          referrerPolicy="no-referrer"
+                        />
+                      ) : null}
                     </button>
                   )
                 })}
+                {items.length < itemTotal ? (
+                  <Button long type="text" loading={itemsLoading} onClick={() => void loadMoreItems()}>
+                    加载更多（{items.length}/{itemTotal}）
+                  </Button>
+                ) : null}
               </div>
             </Card>
 
@@ -346,7 +451,13 @@ export function WorkspaceExplorePage({
                     icon={<IconStar />}
                     onClick={() => void updateSelected({ status: selected.status, starred: !selected.starred })}
                   />
-                  <Button type="text" icon={<IconBook />} onClick={() => onSelectItem(selected.title)} />
+                  <Button
+                    aria-label={`沉淀到项目${activeProjectName}`}
+                    type="text"
+                    icon={<IconBook />}
+                    loading={depositingItemID === selected.id}
+                    onClick={() => void depositSelected()}
+                  />
                   {selected.url ? <Button type="text" icon={<IconLaunch />} href={selected.url} target="_blank" /> : null}
                 </Space>
               </div>
@@ -356,6 +467,15 @@ export function WorkspaceExplorePage({
                   <Text type="secondary">{selectedSource.name}</Text>
                   <Text type="secondary">{itemTime(selected)}</Text>
                 </Space>
+                {selected.imageUrl ? (
+                  <img
+                    className="explore-article-image"
+                    src={selected.imageUrl}
+                    alt=""
+                    loading="lazy"
+                    referrerPolicy="no-referrer"
+                  />
+                ) : null}
                 <Paragraph className="explore-article-content">{selected.content || selected.summary || '暂无正文'}</Paragraph>
               </article>
             </Card>
@@ -385,18 +505,25 @@ export function WorkspaceExplorePage({
           </label>
           <label>
             <Text>数据源类型</Text>
-            <Select
+            <Input value={sourceDraft.kind === 'rss' ? 'RSS 订阅' : '历史数据源（仅维护）'} disabled />
-              value={sourceDraft.kind}
-              onChange={(kind) => setSourceDraft((current) => ({ ...current, kind: kind as DatasetSourceKind }))}
-            >
-              <Select.Option value="link">网页链接</Select.Option>
-              <Select.Option value="rss">RSS 订阅</Select.Option>
-              <Select.Option value="manual">手动收集</Select.Option>
-            </Select>
           </label>
           <label>
             <Text>链接地址</Text>
-            <Input value={sourceDraft.url} placeholder="https://example.com/feed" onChange={(url) => setSourceDraft((current) => ({ ...current, url }))} />
+            <Input
+              value={sourceDraft.url}
+              disabled={Boolean(editingSource?.builtIn || (editingSource && editingSource.kind !== 'rss'))}
+              placeholder="https://example.com/feed"
+              onChange={(url) => setSourceDraft((current) => ({ ...current, url }))}
+            />
+          </label>
+          <label>
+            <Text>图标地址</Text>
+            <Input
+              value={sourceDraft.iconUrl}
+              disabled={Boolean(editingSource?.builtIn || (editingSource && editingSource.kind !== 'rss'))}
+              placeholder="https://example.com/icon.png"
+              onChange={(iconUrl) => setSourceDraft((current) => ({ ...current, iconUrl }))}
+            />
           </label>
           <label>
             <Text>说明</Text>
@@ -440,3 +567,10 @@ function itemTime(item: DatasetItemDTO) {
   if (Number.isNaN(parsed.getTime())) return ''
   return new Intl.DateTimeFormat('zh-CN', { dateStyle: 'medium' }).format(parsed)
 }
+
+function sourceSubtitle(source: DatasetSourceDTO) {
+  if (!source.lastSyncedAt) return `${source.itemCount} 条`
+  const parsed = new Date(source.lastSyncedAt)
+  if (Number.isNaN(parsed.getTime())) return `${source.itemCount} 条`
+  return `${source.itemCount} 条 · ${new Intl.DateTimeFormat('zh-CN', { dateStyle: 'short', timeStyle: 'short' }).format(parsed)}`
+}

apps/web_v1/src/pages/workspace-home.tsx

@@ -11,11 +11,13 @@ import type { SearchResultDTO } from '../api/search'
 import type { InboxSuggestionDTO } from '../api/inbox'
 import type { AIExpertDTO, AISessionDTO, CreateAISessionInput } from '../api/ai'
 import type {
-  DatasetCronDTO,
+  DatasetDepositDTO,
   DatasetItemDTO,
+  DatasetItemPageDTO,
   DatasetItemUpdate,
   DatasetSourceDTO,
   DatasetSourceInput,
+  DatasetSyncResultDTO,
 } from '../api/dataset'
 import type { ChannelKey, InboxConfirmationOutcome, Project, ProjectWorkspace, Theme, WorkbenchView } from './projects/project-types'
 
@@ -61,7 +63,8 @@ export function ProjectPage({
   onUpdateDatasetSource,
   onDeleteDatasetSource,
   onUpdateDatasetItem,
-  onQueueDatasetSync,
+  onSyncDatasetSources,
+  onDepositDatasetItem,
 }: {
   activeView: WorkbenchView
   activeWorkspace: ProjectWorkspace
@@ -97,12 +100,13 @@ export function ProjectPage({
   onListAIExperts: (signal?: AbortSignal) => Promise<AIExpertDTO[]>
   onCreateAISession: (projectId: string, input: CreateAISessionInput, signal?: AbortSignal) => Promise<AISessionDTO>
   onListDatasetSources: (signal?: AbortSignal) => Promise<DatasetSourceDTO[]>
-  onListDatasetItems: (signal?: AbortSignal) => Promise<DatasetItemDTO[]>
+  onListDatasetItems: (sourceId?: string, offset?: number, limit?: number, signal?: AbortSignal) => Promise<DatasetItemPageDTO>
   onCreateDatasetSource: (input: DatasetSourceInput) => Promise<DatasetSourceDTO>
   onUpdateDatasetSource: (sourceId: string, input: DatasetSourceInput) => Promise<DatasetSourceDTO>
   onDeleteDatasetSource: (sourceId: string) => Promise<void>
   onUpdateDatasetItem: (itemId: string, input: DatasetItemUpdate) => Promise<DatasetItemDTO>
-  onQueueDatasetSync: () => Promise<DatasetCronDTO[]>
+  onSyncDatasetSources: () => Promise<DatasetSyncResultDTO[]>
+  onDepositDatasetItem: (itemId: string, projectId: string) => Promise<DatasetDepositDTO & { refreshFailed?: boolean }>
 }) {
   const isProject = activeView === 'project'
   const projects = workspaces.map((workspace) => workspace.project)
@@ -159,14 +163,16 @@ export function ProjectPage({
             <WorkspacePage workspaces={workspaces} onOpenTask={onOpenTask} onUpdateTask={onUpdateWorkspaceTask} />
           ) : activeView === 'workspace-explore' ? (
             <WorkspaceExplorePage
-              onSelectItem={onSelectItem}
+              activeProjectID={activeWorkspace.project.id}
+              activeProjectName={activeWorkspace.project.name}
               onListSources={onListDatasetSources}
               onListItems={onListDatasetItems}
               onCreateSource={onCreateDatasetSource}
               onUpdateSource={onUpdateDatasetSource}
               onDeleteSource={onDeleteDatasetSource}
               onUpdateItem={onUpdateDatasetItem}
-              onQueueSync={onQueueDatasetSync}
+              onSyncSources={onSyncDatasetSources}
+              onDepositItem={onDepositDatasetItem}
             />
           ) : (
             <ProjectChannelPage

backend/cmd/api/main.go

@@ -1,9 +1,11 @@
 package main
 
 import (
+	"context"
 	"log"
 
 	"senlinai-agent/backend/internal/config"
+	"senlinai-agent/backend/internal/crontab"
 	"senlinai-agent/backend/internal/httpx"
 	"senlinai-agent/backend/internal/initdb"
 	"senlinai-agent/backend/internal/logic/ai"
@@ -29,7 +31,9 @@ func main() {
 	}
 
 	authService := auth.NewService(cfg.AuthSecret)
-	datasetHandler := dataset.NewHandler(dataset.NewService(models.DBService))
+	datasetService := dataset.NewService(models.DBService)
+	datasetHandler := dataset.NewHandler(datasetService)
+	datasetScheduler := crontab.NewDatasetScheduler(datasetService, log.Default())
 	projectService := projects.NewService()
 	fileService := files.NewService(cfg.StorageDir, models.DBService)
 	taskService := tasks.NewService(models.DBService)
@@ -57,6 +61,10 @@ func main() {
 		searchHandler,
 		aiHandler,
 	)
+	schedulerContext, stopScheduler := context.WithCancel(context.Background())
+	defer stopScheduler()
+	go datasetScheduler.Run(schedulerContext)
+
 	if err := appRouter.Run(":" + cfg.Port); err != nil {
 		log.Fatal(err)
 	}

backend/internal/crontab/dataset.go

@@ -0,0 +1,94 @@
+package crontab
+
+import (
+	"context"
+	"log"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"senlinai-agent/backend/internal/logic/dataset"
+)
+
+var datasetInterval = 5 * time.Minute
+
+type DatasetCollector interface {
+	SyncAllSources(context.Context) ([]dataset.SyncResult, error)
+}
+
+type DatasetScheduler struct {
+	collector DatasetCollector
+	interval  time.Duration
+	logger    *log.Logger
+}
+
+func NewDatasetScheduler(collector DatasetCollector, logger *log.Logger) *DatasetScheduler {
+	if logger == nil {
+		logger = log.Default()
+	}
+	return &DatasetScheduler{
+		collector: collector,
+		interval:  datasetInterval,
+		logger:    logger,
+	}
+}
+
+// Run collects once at startup, then on a fixed five-minute interval.
+// A tick is skipped when the previous collection is still running.
+func (s *DatasetScheduler) Run(ctx context.Context) {
+	if ctx.Err() != nil {
+		return
+	}
+	ticker := time.NewTicker(s.interval)
+	defer ticker.Stop()
+
+	var running atomic.Bool
+	var workers sync.WaitGroup
+	start := func() {
+		if !running.CompareAndSwap(false, true) {
+			s.logger.Print("dataset collection skipped: previous run is still active")
+			return
+		}
+		workers.Add(1)
+		go func() {
+			defer workers.Done()
+			defer running.Store(false)
+			s.collect(ctx)
+		}()
+	}
+	start()
+
+	for {
+		select {
+		case <-ctx.Done():
+			workers.Wait()
+			return
+		case <-ticker.C:
+			start()
+		}
+	}
+}
+
+func (s *DatasetScheduler) collect(ctx context.Context) {
+	results, err := s.collector.SyncAllSources(ctx)
+	completed := 0
+	failed := 0
+	for _, result := range results {
+		switch result.Status {
+		case "completed":
+			completed++
+		case "failed":
+			failed++
+			s.logger.Printf(
+				"dataset source collection failed: source=%s error=%s",
+				result.SourceIdentity,
+				result.Result,
+			)
+		}
+	}
+	if err != nil {
+		s.logger.Printf("dataset collection failed: sources=%d completed=%d failed=%d error=%v", len(results), completed, failed, err)
+		return
+	}
+	s.logger.Printf("dataset collection finished: sources=%d completed=%d failed=%d", len(results), completed, failed)
+}

backend/internal/crontab/dataset_test.go

@@ -0,0 +1,150 @@
+package crontab
+
+import (
+	"context"
+	"io"
+	"log"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"senlinai-agent/backend/internal/logic/dataset"
+)
+
+func TestDatasetSchedulerCollectsImmediatelyAndRepeats(t *testing.T) {
+	collector := &recordingDatasetCollector{calls: make(chan time.Time, 3)}
+	scheduler := &DatasetScheduler{
+		collector: collector,
+		interval:  10 * time.Millisecond,
+		logger:    log.New(io.Discard, "", 0),
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		scheduler.Run(ctx)
+	}()
+
+	first := receiveCollectionCall(t, collector.calls)
+	second := receiveCollectionCall(t, collector.calls)
+	require.GreaterOrEqual(t, second.Sub(first), 8*time.Millisecond)
+
+	cancel()
+	require.Eventually(t, func() bool {
+		select {
+		case <-done:
+			return true
+		default:
+			return false
+		}
+	}, time.Second, 5*time.Millisecond)
+}
+
+func TestDatasetSchedulerStopsWhenContextIsAlreadyCanceled(t *testing.T) {
+	collector := &recordingDatasetCollector{calls: make(chan time.Time, 1)}
+	scheduler := &DatasetScheduler{
+		collector: collector,
+		interval:  10 * time.Millisecond,
+		logger:    log.New(io.Discard, "", 0),
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	scheduler.Run(ctx)
+
+	require.Zero(t, collector.callCount())
+}
+
+func TestDatasetSchedulerSkipsTicksWhileCollectionIsRunning(t *testing.T) {
+	collector := &blockingDatasetCollector{
+		started: make(chan struct{}, 2),
+		release: make(chan struct{}),
+	}
+	scheduler := &DatasetScheduler{
+		collector: collector,
+		interval:  5 * time.Millisecond,
+		logger:    log.New(io.Discard, "", 0),
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		scheduler.Run(ctx)
+	}()
+
+	select {
+	case <-collector.started:
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for first collection")
+	}
+	time.Sleep(20 * time.Millisecond)
+	require.Equal(t, 1, collector.callCount())
+
+	close(collector.release)
+	require.Eventually(t, func() bool {
+		return collector.callCount() >= 2
+	}, time.Second, 5*time.Millisecond)
+	cancel()
+	<-done
+}
+
+type recordingDatasetCollector struct {
+	mu    sync.Mutex
+	count int
+	calls chan time.Time
+}
+
+func (c *recordingDatasetCollector) SyncAllSources(context.Context) ([]dataset.SyncResult, error) {
+	now := time.Now()
+	c.mu.Lock()
+	c.count++
+	c.mu.Unlock()
+	c.calls <- now
+	return []dataset.SyncResult{{Status: "completed"}}, nil
+}
+
+func (c *recordingDatasetCollector) callCount() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.count
+}
+
+func receiveCollectionCall(t *testing.T, calls <-chan time.Time) time.Time {
+	t.Helper()
+	select {
+	case calledAt := <-calls:
+		return calledAt
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for dataset collection")
+		return time.Time{}
+	}
+}
+
+type blockingDatasetCollector struct {
+	mu      sync.Mutex
+	count   int
+	started chan struct{}
+	release chan struct{}
+}
+
+func (c *blockingDatasetCollector) SyncAllSources(ctx context.Context) ([]dataset.SyncResult, error) {
+	c.mu.Lock()
+	c.count++
+	c.mu.Unlock()
+	select {
+	case c.started <- struct{}{}:
+	default:
+	}
+	select {
+	case <-c.release:
+	case <-ctx.Done():
+	}
+	return nil, nil
+}
+
+func (c *blockingDatasetCollector) callCount() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.count
+}

backend/internal/httpx/router_integration_test.go

@@ -8,6 +8,7 @@ import (
 	"senlinai-agent/backend/internal/httpx"
 	"senlinai-agent/backend/internal/logic/ai"
 	"senlinai-agent/backend/internal/logic/auth"
+	"senlinai-agent/backend/internal/logic/dataset"
 	"senlinai-agent/backend/internal/logic/files"
 	"senlinai-agent/backend/internal/logic/inbox"
 	"senlinai-agent/backend/internal/logic/projects"
@@ -28,6 +29,7 @@ func TestMainRegistrarsRegisterTogether(t *testing.T) {
 			cfg,
 			authService.VerifySession,
 			auth.NewHandler(authService),
+			dataset.NewHandler(dataset.NewService(nil)),
 			projects.NewHandler(projectService),
 			projects.NewTagHandler(projectService),
 			projects.NewCronHandler(projectService),

backend/internal/initdb/dataset.go

@@ -1,58 +1,55 @@
 package initdb
 
 import (
-	"errors"
-	"fmt"
-
 	"gorm.io/gorm"
 	"senlinai-agent/backend/internal/models"
 )
 
-var defaultDatasetSources = []struct {
+type defaultDatasetSource struct {
+	Key     string
 	Name    string
 	URL     string
 	IconURL string
-}{
-	{Name: "彭博社最新报道", URL: "https://bbg.buzzing.cc/feed.json", IconURL: "https://bbg.buzzing.cc/apple-touch-icon.png"},
-	{Name: "国外科技头条", URL: "https://tech.buzzing.cc/feed.json", IconURL: "https://tech.buzzing.cc/apple-touch-icon.png"},
-	{Name: "国外财经新闻", URL: "https://finance.buzzing.cc/feed.json", IconURL: "https://finance.buzzing.cc/apple-touch-icon.png"},
-	{Name: "华尔街日报热门", URL: "https://wsj.buzzing.cc/feed.json", IconURL: "https://wsj.buzzing.cc/apple-touch-icon.png"},
-	{Name: "Product Hunt", URL: "https://ph.buzzing.cc/feed.json", IconURL: "https://ph.buzzing.cc/apple-touch-icon.png"},
-	{Name: "Dev.to", URL: "https://dev.buzzing.cc/feed.json", IconURL: "https://dev.buzzing.cc/apple-touch-icon.png"},
 }
 
-// InitDataset creates the default Buzzing RSS sources for the default account when the source table is empty.
+var defaultDatasetSources = []defaultDatasetSource{
-func InitDataset(database *gorm.DB) error {
+	{Key: "buzzing-bloomberg", Name: "彭博社最新报道", URL: "https://bbg.buzzing.cc/feed.json", IconURL: "https://bbg.buzzing.cc/apple-touch-icon.png"},
+	{Key: "buzzing-tech", Name: "国外科技头条", URL: "https://tech.buzzing.cc/feed.json", IconURL: "https://tech.buzzing.cc/apple-touch-icon.png"},
+	{Key: "buzzing-finance", Name: "国外财经新闻", URL: "https://finance.buzzing.cc/feed.json", IconURL: "https://finance.buzzing.cc/apple-touch-icon.png"},
+	{Key: "buzzing-wsj", Name: "华尔街日报热门", URL: "https://wsj.buzzing.cc/feed.json", IconURL: "https://wsj.buzzing.cc/apple-touch-icon.png"},
+	{Key: "buzzing-product-hunt", Name: "Product Hunt", URL: "https://ph.buzzing.cc/feed.json", IconURL: "https://ph.buzzing.cc/apple-touch-icon.png"},
+	{Key: "buzzing-devto", Name: "Dev.to", URL: "https://dev.buzzing.cc/feed.json", IconURL: "https://dev.buzzing.cc/apple-touch-icon.png"},
+	{Key: "rsshub-cls-hot", Name: "财联社-热门", URL: "https://rsshub.ktachibana.party/cls/hot"},
+	{Key: "rsshub-jin10", Name: "金十数据-快讯", URL: "https://rsshub.ktachibana.party/jin10"},
+}
+
+// InitDataset inserts the built-in dataset sources when the owner has no sources.
+func InitDataset(database *gorm.DB, ownerID uint, ownerIdentity string) error {
 	var count int64
-	if err := database.Model(&models.SaDatasetSource{}).Count(&count).Error; err != nil {
+	if err := database.Model(&models.SaDatasetSource{}).
+		Where("owner_id = ?", ownerID).
+		Count(&count).Error; err != nil {
 		return err
 	}
 	if count > 0 {
 		return nil
 	}
 
-	var owner models.SaUser
+	for _, definition := range defaultDatasetSources {
-	err := database.Where("email = ?", rootUsername).First(&owner).Error
+		seedKey := definition.Key
-	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
+		if err := database.Create(&models.SaDatasetSource{
-		return fmt.Errorf("find root user for default dataset sources: %w", err)
+			OwnerID:       ownerID,
-	}
+			OwnerIdentity: ownerIdentity,
-	if errors.Is(err, gorm.ErrRecordNotFound) {
+			SeedKey:       &seedKey,
-		if err := database.Order("id asc").First(&owner).Error; err != nil {
+			Name:          definition.Name,
-			return fmt.Errorf("find owner for default dataset sources: %w", err)
-		}
-	}
-
-	sources := make([]models.SaDatasetSource, 0, len(defaultDatasetSources))
-	for _, source := range defaultDatasetSources {
-		sources = append(sources, models.SaDatasetSource{
-			CreatedBy:   owner.ID,
-			Name:        source.Name,
 			Kind:          "rss",
-			URL:         source.URL,
+			URL:           definition.URL,
-			IconURL:     source.IconURL,
+			IconURL:       definition.IconURL,
-			Description: "Buzzing 中文聚合订阅",
+			Description:   "默认资讯订阅",
 			Enabled:       true,
-		})
+		}).Error; err != nil {
+			return err
 		}
-	return database.Create(&sources).Error
+	}
+	return nil
 }

backend/internal/initdb/dataset_test.go

@@ -10,43 +10,43 @@ import (
 	"senlinai-agent/backend/internal/models"
 )
 
-func TestInitDatasetCreatesDefaultSourcesWhenTableIsEmpty(t *testing.T) {
+func TestInitDatasetCreatesDefaultSourcesWhenOwnerHasNoSources(t *testing.T) {
 	database := newDatasetInitTestDatabase(t)
-	require.NoError(t, InitUser(database))
+	root, err := InitUser(database)
+	require.NoError(t, err)
 
-	require.NoError(t, InitDataset(database))
+	require.NoError(t, InitDataset(database, root.ID, root.Identity))
-	require.NoError(t, InitDataset(database))
+	require.NoError(t, InitDataset(database, root.ID, root.Identity))
 
-	var root models.SaUser
-	require.NoError(t, database.Where("email = ?", rootUsername).First(&root).Error)
 	var sources []models.SaDatasetSource
 	require.NoError(t, database.Order("id asc").Find(&sources).Error)
 	require.Len(t, sources, len(defaultDatasetSources))
 	for index, expected := range defaultDatasetSources {
+		require.Equal(t, root.ID, sources[index].OwnerID)
+		require.Equal(t, root.Identity, sources[index].OwnerIdentity)
 		require.Equal(t, expected.Name, sources[index].Name)
 		require.Equal(t, expected.URL, sources[index].URL)
 		require.Equal(t, expected.IconURL, sources[index].IconURL)
+		require.NotNil(t, sources[index].SeedKey)
+		require.Equal(t, expected.Key, *sources[index].SeedKey)
 		require.Equal(t, "rss", sources[index].Kind)
 		require.True(t, sources[index].Enabled)
-		require.Equal(t, root.ID, sources[index].CreatedBy)
-		require.Equal(t, root.Identity, sources[index].CreatedByIdentity)
 	}
 }
 
-func TestInitDatasetDoesNothingWhenSourceTableIsNotEmpty(t *testing.T) {
+func TestInitDatasetDoesNothingWhenOwnerAlreadyHasSource(t *testing.T) {
 	database := newDatasetInitTestDatabase(t)
-	require.NoError(t, InitUser(database))
+	root, err := InitUser(database)
-	var root models.SaUser
+	require.NoError(t, err)
-	require.NoError(t, database.Where("email = ?", rootUsername).First(&root).Error)
 	existing := models.SaDatasetSource{
-		CreatedBy: root.ID,
+		OwnerID: root.ID,
 		Name:    "Existing source",
 		Kind:    "manual",
 		Enabled: true,
 	}
 	require.NoError(t, database.Create(&existing).Error)
 
-	require.NoError(t, InitDataset(database))
+	require.NoError(t, InitDataset(database, root.ID, root.Identity))
 
 	var sources []models.SaDatasetSource
 	require.NoError(t, database.Find(&sources).Error)
@@ -54,31 +54,43 @@ func TestInitDatasetDoesNothingWhenSourceTableIsNotEmpty(t *testing.T) {
 	require.Equal(t, existing.Identity, sources[0].Identity)
 }
 
-func TestInitDatasetUsesFirstExistingUserWhenRootIsAbsent(t *testing.T) {
+func TestInitDatasetCountsSourcesByOwner(t *testing.T) {
 	database := newDatasetInitTestDatabase(t)
-	user := models.SaUser{
+	root, err := InitUser(database)
-		Email:        "existing@example.com",
+	require.NoError(t, err)
-		DisplayName:  "Existing",
+	other := models.SaUser{
-		PasswordHash: "hash",
+		Email: "other@example.com", DisplayName: "Other",
-		Role:         "user",
+		PasswordHash: "hash", Role: "user",
 	}
-	require.NoError(t, database.Create(&user).Error)
+	require.NoError(t, database.Create(&other).Error)
+	require.NoError(t, database.Create(&models.SaDatasetSource{
+		OwnerID: other.ID, Name: "Other source", Kind: "manual", Enabled: true,
+	}).Error)
 
-	require.NoError(t, InitDataset(database))
+	require.NoError(t, InitDataset(database, root.ID, root.Identity))
 
-	var sources []models.SaDatasetSource
+	var rootCount int64
-	require.NoError(t, database.Find(&sources).Error)
+	require.NoError(t, database.Model(&models.SaDatasetSource{}).
-	require.Len(t, sources, len(defaultDatasetSources))
+		Where("owner_id = ?", root.ID).
-	for _, source := range sources {
+		Count(&rootCount).Error)
-		require.Equal(t, user.ID, source.CreatedBy)
+	require.Equal(t, int64(len(defaultDatasetSources)), rootCount)
-		require.Equal(t, user.Identity, source.CreatedByIdentity)
+	var otherCount int64
-	}
+	require.NoError(t, database.Model(&models.SaDatasetSource{}).
+		Where("owner_id = ?", other.ID).
+		Count(&otherCount).Error)
+	require.Equal(t, int64(1), otherCount)
 }
 
 func newDatasetInitTestDatabase(t *testing.T) *gorm.DB {
 	t.Helper()
-	database, err := gorm.Open(sqlite.Open(fmt.Sprintf("file:%s?mode=memory&cache=shared", t.Name())), &gorm.Config{})
+	database, err := gorm.Open(
+		sqlite.Open(fmt.Sprintf("file:%s?mode=memory&cache=shared", t.Name())),
+		&gorm.Config{},
+	)
 	require.NoError(t, err)
-	require.NoError(t, database.AutoMigrate(&models.SaUser{}, &models.SaDatasetSource{}))
+	require.NoError(t, database.AutoMigrate(
+		&models.SaUser{},
+		&models.SaDatasetSource{},
+	))
 	return database
 }

backend/internal/initdb/expert_test.go

@@ -60,6 +60,7 @@ func newExpertTestDatabase(t *testing.T) *gorm.DB {
 	require.NoError(t, database.AutoMigrate(
 		&models.SaUser{},
 		&models.SaDatasetSource{},
+		&models.SaDatasetItem{},
 		&models.SaAIExpertCategory{},
 		&models.SaAIExpertItem{},
 	))

backend/internal/initdb/new.go

@@ -5,10 +5,11 @@ import "gorm.io/gorm"
 // New initializes the default database records.
 func New(database *gorm.DB) error {
 	return database.Transaction(func(tx *gorm.DB) error {
-		if err := InitUser(tx); err != nil {
+		user, err := InitUser(tx)
+		if err != nil {
 			return err
 		}
-		if err := InitDataset(tx); err != nil {
+		if err := InitDataset(tx, user.ID, user.Identity); err != nil {
 			return err
 		}
 		return InitExpert(tx)

backend/internal/initdb/user.go

@@ -12,25 +12,34 @@ const (
 	rootRole     = "root"
 )
 
-// InitUser creates the default root account when the user table is empty.
+// InitUser returns the existing root user or creates and returns it when the user table is empty.
-func InitUser(database *gorm.DB) error {
+func InitUser(database *gorm.DB) (*models.SaUser, error) {
 	var count int64
 	if err := database.Model(&models.SaUser{}).Count(&count).Error; err != nil {
-		return err
+		return nil, err
 	}
 	if count > 0 {
-		return nil
+		var user models.SaUser
+		if err := database.Where("email = ? AND role = ?", rootUsername, rootRole).
+			First(&user).Error; err != nil {
+			return nil, err
+		}
+		return &user, nil
 	}
 
 	passwordHash, err := bcrypt.GenerateFromPassword([]byte(rootPassword), bcrypt.DefaultCost)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	return database.Create(&models.SaUser{
+	user := &models.SaUser{
 		Email:        rootUsername,
 		DisplayName:  rootUsername,
 		PasswordHash: string(passwordHash),
 		Role:         rootRole,
-	}).Error
+	}
+	if err := database.Create(user).Error; err != nil {
+		return nil, err
+	}
+	return user, nil
 }

backend/internal/initdb/user_test.go

@@ -13,33 +13,42 @@ import (
 func TestInitUserCreatesRootUserWhenTableIsEmpty(t *testing.T) {
 	database := newUserTestDatabase(t)
 
-	require.NoError(t, InitUser(database))
+	user, err := InitUser(database)
+	require.NoError(t, err)
 
 	var users []models.SaUser
 	require.NoError(t, database.Find(&users).Error)
 	require.Len(t, users, 1)
+	require.Equal(t, users[0].ID, user.ID)
+	require.Equal(t, users[0].Identity, user.Identity)
 	require.Equal(t, rootUsername, users[0].Email)
 	require.Equal(t, rootUsername, users[0].DisplayName)
 	require.Equal(t, rootRole, users[0].Role)
 	require.NoError(t, bcrypt.CompareHashAndPassword([]byte(users[0].PasswordHash), []byte(rootPassword)))
 }
 
-func TestInitUserDoesNothingWhenTableIsNotEmpty(t *testing.T) {
+func TestInitUserReturnsExistingRootWhenTableIsNotEmpty(t *testing.T) {
 	database := newUserTestDatabase(t)
-	existing := models.SaUser{
+	root := models.SaUser{
-		Email:        "existing@example.com",
+		Email:        rootUsername,
-		DisplayName:  "Existing User",
+		DisplayName:  rootUsername,
 		PasswordHash: "existing-hash",
-		Role:         "user",
+		Role:         rootRole,
 	}
-	require.NoError(t, database.Create(&existing).Error)
+	require.NoError(t, database.Create(&root).Error)
+	require.NoError(t, database.Create(&models.SaUser{
+		Email: "other@example.com", DisplayName: "Other",
+		PasswordHash: "hash", Role: "user",
+	}).Error)
 
-	require.NoError(t, InitUser(database))
+	user, err := InitUser(database)
+	require.NoError(t, err)
 
 	var users []models.SaUser
 	require.NoError(t, database.Find(&users).Error)
-	require.Len(t, users, 1)
+	require.Len(t, users, 2)
-	require.Equal(t, existing.Email, users[0].Email)
+	require.Equal(t, root.ID, user.ID)
+	require.Equal(t, root.Identity, user.Identity)
 }
 
 func newUserTestDatabase(t *testing.T) *gorm.DB {

backend/internal/logic/dataset/feed.go

@@ -0,0 +1,429 @@
+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(`<[^>]*>`)
+	htmlImagePattern = regexp.MustCompile(`(?i)<img[^>]+src\s*=\s*["']([^"']+)["']`)
+)
+
+type FeedItem struct {
+	ExternalID  string
+	Title       string
+	Summary     string
+	Content     string
+	URL         string
+	ImageURL    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"`
+	Image         string               `json:"image"`
+	BannerImage   string               `json:"banner_image"`
+	Attachments   []jsonFeedAttachment `json:"attachments"`
+	DatePublished string               `json:"date_published"`
+}
+
+type jsonFeedAttachment struct {
+	URL      string `json:"url"`
+	MIMEType string `json:"mime_type"`
+}
+
+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,
+			ImageURL: firstNonEmpty(
+				input.Image,
+				input.BannerImage,
+				firstJSONImageAttachment(input.Attachments),
+				htmlImageURL(input.ContentHTML),
+			),
+			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"`
+	Enclosure   xmlMedia   `xml:"enclosure"`
+	Thumbnail   xmlMedia   `xml:"thumbnail"`
+	Media       []xmlMedia `xml:"content"`
+}
+
+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"`
+	Type string `xml:"type,attr"`
+}
+
+type xmlMedia struct {
+	URL    string `xml:"url,attr"`
+	Type   string `xml:"type,attr"`
+	Medium string `xml:"medium,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),
+				ImageURL: firstNonEmpty(
+					imageMediaURL(input.Enclosure),
+					imageMediaURL(input.Thumbnail),
+					firstImageMediaURL(input.Media),
+					htmlImageURL(input.Content),
+					htmlImageURL(input.Description),
+				),
+				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), ImageURL: atomImageURL(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)
+	item.ImageURL = truncateRunes(strings.TrimSpace(item.ImageURL), 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 htmlImageURL(value string) string {
+	match := htmlImagePattern.FindStringSubmatch(value)
+	if len(match) < 2 {
+		return ""
+	}
+	return strings.TrimSpace(html.UnescapeString(match[1]))
+}
+
+func firstJSONImageAttachment(attachments []jsonFeedAttachment) string {
+	for _, attachment := range attachments {
+		if strings.HasPrefix(strings.ToLower(strings.TrimSpace(attachment.MIMEType)), "image/") {
+			return strings.TrimSpace(attachment.URL)
+		}
+	}
+	return ""
+}
+
+func imageMediaURL(media xmlMedia) string {
+	mediaType := strings.ToLower(strings.TrimSpace(media.Type))
+	medium := strings.ToLower(strings.TrimSpace(media.Medium))
+	if media.URL == "" || (mediaType != "" && !strings.HasPrefix(mediaType, "image/") && medium != "image") {
+		return ""
+	}
+	return strings.TrimSpace(media.URL)
+}
+
+func firstImageMediaURL(media []xmlMedia) string {
+	for _, item := range media {
+		if imageURL := imageMediaURL(item); imageURL != "" {
+			return imageURL
+		}
+	}
+	return ""
+}
+
+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 atomImageURL(links []xmlAtomLink) string {
+	for _, link := range links {
+		if strings.EqualFold(link.Rel, "enclosure") &&
+			strings.HasPrefix(strings.ToLower(strings.TrimSpace(link.Type)), "image/") {
+			return strings.TrimSpace(link.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()
+}

backend/internal/logic/dataset/feed_test.go

@@ -0,0 +1,285 @@
+package dataset
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/url"
+	"os"
+	"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>",
+			"image": "https://example.com/json-image.jpg",
+			"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.Equal(t, "https://example.com/json-image.jpg", feed.Items[0].ImageURL)
+	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>
+				<media:thumbnail xmlns:media="http://search.yahoo.com/mrss/" url="https://example.com/rss-image.jpg"/>
+				<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.Equal(t, "https://example.com/rss-image.jpg", feed.Items[0].ImageURL)
+	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"/>
+			<link rel="enclosure" type="image/jpeg" href="https://example.com/atom-image.jpg"/>
+			<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, "https://example.com/atom-image.jpg", feed.Items[0].ImageURL)
+	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 TestLiveFeedCollectionFormats(t *testing.T) {
+	if os.Getenv("TEST_LIVE_FEEDS") != "1" {
+		t.Skip("set TEST_LIVE_FEEDS=1 to verify external feeds")
+	}
+	fetcher := NewHTTPFeedFetcher()
+	for _, test := range []struct {
+		name       string
+		url        string
+		wantFormat string
+	}{
+		{name: "JSON Feed", url: "https://bbg.buzzing.cc/feed.json", wantFormat: "json_feed"},
+		{name: "RSS XML", url: "https://rsshub.ktachibana.party/cls/hot", wantFormat: "rss"},
+		{name: "second RSS XML", url: "https://rsshub.ktachibana.party/jin10", wantFormat: "rss"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			feed, err := fetcher.Fetch(context.Background(), test.url)
+			require.NoError(t, err)
+			require.Equal(t, test.wantFormat, feed.Format)
+			require.NotEmpty(t, feed.Items)
+		})
+	}
+}
+
+func TestSyncSourcesRecordsFeedFailure(t *testing.T) {
+	database := newDatasetTestDatabase(t)
+	user := createDatasetTestUser(t, database, "feed-owner@example.com")
+	source := models.SaDatasetSource{
+		OwnerID: user.ID,
+		Name:    "Broken feed",
+		Kind:    "rss",
+		URL:     "https://example.com/feed",
+		Enabled: true,
+	}
+	require.NoError(t, database.Create(&source).Error)
+
+	results, err := newServiceWithFetcher(database, failingFeedFetcher{}).SyncSources(context.Background(), user.ID)
+
+	require.NoError(t, err)
+	require.Len(t, results, 1)
+	require.Equal(t, "failed", results[0].Status)
+	require.Contains(t, results[0].Result, "feed unavailable")
+}
+
+func TestSyncAllSourcesCollectsEnabledRSSSourcesForEveryUser(t *testing.T) {
+	database := newDatasetTestDatabase(t)
+	firstUser := createDatasetTestUser(t, database, "first-feed-owner@example.com")
+	secondUser := createDatasetTestUser(t, database, "second-feed-owner@example.com")
+	for _, source := range []models.SaDatasetSource{
+		{OwnerID: firstUser.ID, Name: "First RSS", Kind: "rss", URL: "https://example.com/first", Enabled: true},
+		{OwnerID: secondUser.ID, Name: "Second RSS", Kind: "rss", URL: "https://example.com/second", Enabled: true},
+		{OwnerID: firstUser.ID, Name: "Manual", Kind: "manual", Enabled: true},
+	} {
+		require.NoError(t, database.Create(&source).Error)
+	}
+
+	results, err := newServiceWithFetcher(database, staticFeedFetcher{}).SyncAllSources(context.Background())
+
+	require.NoError(t, err)
+	require.Len(t, results, 2)
+	var itemCount int64
+	require.NoError(t, database.Model(&models.SaDatasetItem{}).Count(&itemCount).Error)
+	require.Equal(t, int64(2), itemCount)
+}
+
+func TestSyncSourcesCollectsMultipleSourcesForOneOwner(t *testing.T) {
+	database := newDatasetTestDatabase(t)
+	user := createDatasetTestUser(t, database, "multi-feed-owner@example.com")
+	for index := 0; index < 5; index++ {
+		require.NoError(t, database.Create(&models.SaDatasetSource{
+			OwnerID: user.ID, Name: fmt.Sprintf("Feed %d", index), Kind: "rss",
+			URL: fmt.Sprintf("https://example.com/feed-%d", index), Enabled: true,
+		}).Error)
+	}
+
+	runs, err := newServiceWithFetcher(database, staticFeedFetcher{}).SyncSources(context.Background(), user.ID)
+
+	require.NoError(t, err)
+	require.Len(t, runs, 5)
+	for _, run := range runs {
+		require.Equal(t, "completed", run.Status)
+	}
+}
+
+func TestStoreFeedItemsDeduplicatesAndRefreshesMetadata(t *testing.T) {
+	database := newDatasetTestDatabase(t)
+	user := createDatasetTestUser(t, database, "refresh-feed-owner@example.com")
+	source := models.SaDatasetSource{
+		OwnerID: user.ID, Name: "Refresh feed", Kind: "rss",
+		URL: "https://example.com/feeds/main.xml", Enabled: true,
+	}
+	require.NoError(t, database.Create(&source).Error)
+
+	inserted, err := storeFeedItems(database, source, []FeedItem{{
+		ExternalID: "stable-item", Title: "Old title", Summary: "Old summary",
+		URL: "/articles/1",
+	}})
+	require.NoError(t, err)
+	require.Equal(t, 1, inserted)
+
+	var stored models.SaDatasetItem
+	require.NoError(t, database.Where("source_id = ?", source.ID).First(&stored).Error)
+	require.NoError(t, database.Model(&stored).Updates(map[string]any{
+		"status": "read", "starred": true,
+	}).Error)
+
+	inserted, err = storeFeedItems(database, source, []FeedItem{{
+		ExternalID: "stable-item", Title: "Updated title", Content: "Updated content",
+		URL: "/articles/1", ImageURL: "../images/cover.jpg",
+	}})
+	require.NoError(t, err)
+	require.Zero(t, inserted)
+
+	require.NoError(t, database.First(&stored, stored.ID).Error)
+	require.Equal(t, "Updated title", stored.Title)
+	require.Equal(t, "Old summary", stored.Summary)
+	require.Equal(t, "Updated content", stored.Content)
+	require.Equal(t, "https://example.com/articles/1", stored.URL)
+	require.Equal(t, "https://example.com/images/cover.jpg", stored.ImageURL)
+	require.Equal(t, "read", stored.Status)
+	require.True(t, stored.Starred)
+	var count int64
+	require.NoError(t, database.Model(&models.SaDatasetItem{}).
+		Where("source_id = ?", source.ID).
+		Count(&count).Error)
+	require.Equal(t, int64(1), count)
+}
+
+func TestSyncSourcesRejectsOverlap(t *testing.T) {
+	database := newDatasetTestDatabase(t)
+	user := createDatasetTestUser(t, database, "queued-feed-owner@example.com")
+	source := models.SaDatasetSource{
+		OwnerID: user.ID, Name: "Queued RSS", Kind: "rss",
+		URL: "https://example.com/feed", Enabled: true,
+	}
+	require.NoError(t, database.Create(&source).Error)
+	fetcher := &blockingFeedFetcher{
+		started: make(chan struct{}),
+		release: make(chan struct{}),
+	}
+	service := newServiceWithFetcher(database, fetcher)
+
+	type syncOutcome struct {
+		results []SyncResult
+		err     error
+	}
+	done := make(chan syncOutcome, 1)
+	go func() {
+		results, err := service.SyncSources(context.Background(), user.ID)
+		done <- syncOutcome{results: results, err: err}
+	}()
+	<-fetcher.started
+	_, err := service.SyncSources(context.Background(), user.ID)
+	require.ErrorIs(t, err, ErrSyncInProgress)
+	require.ErrorIs(t, service.DeleteSource(user.ID, source.Identity), ErrSyncInProgress)
+	close(fetcher.release)
+	outcome := <-done
+	require.NoError(t, outcome.err)
+	require.Len(t, outcome.results, 1)
+	require.Equal(t, "completed", outcome.results[0].Status)
+}
+
+type failingFeedFetcher struct{}
+
+func (failingFeedFetcher) Fetch(context.Context, string) (ParsedFeed, error) {
+	return ParsedFeed{}, errors.New("feed unavailable")
+}
+
+type blockingFeedFetcher struct {
+	started chan struct{}
+	release chan struct{}
+}
+
+func (f *blockingFeedFetcher) Fetch(context.Context, string) (ParsedFeed, error) {
+	close(f.started)
+	<-f.release
+	return staticFeedFetcher{}.Fetch(context.Background(), "")
+}

backend/internal/logic/dataset/handlers.go

@@ -3,6 +3,7 @@ package dataset
 import (
 	"errors"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 
@@ -25,6 +26,7 @@ type SourceDTO struct {
 	IconURL      string     `json:"iconUrl"`
 	Description  string     `json:"description"`
 	Enabled      bool       `json:"enabled"`
+	BuiltIn      bool       `json:"builtIn"`
 	LastSyncedAt *time.Time `json:"lastSyncedAt"`
 	ItemCount    int64      `json:"itemCount"`
 	CreatedAt    time.Time  `json:"createdAt"`
@@ -38,6 +40,7 @@ type ItemDTO struct {
 	Summary     string     `json:"summary"`
 	Content     string     `json:"content"`
 	URL         string     `json:"url"`
+	ImageURL    string     `json:"imageUrl"`
 	Status      string     `json:"status"`
 	Starred     bool       `json:"starred"`
 	PublishedAt *time.Time `json:"publishedAt"`
@@ -45,17 +48,22 @@ type ItemDTO struct {
 	UpdatedAt   time.Time  `json:"updatedAt"`
 }
 
-type CronDTO struct {
+type ItemPageDTO struct {
-	ID         string     `json:"id"`
+	Items  []ItemDTO `json:"items"`
+	Total  int64     `json:"total"`
+	Limit  int       `json:"limit"`
+	Offset int       `json:"offset"`
+}
+
+type SyncResultDTO struct {
 	SourceID string `json:"sourceId"`
-	Schedule   string     `json:"schedule"`
 	Status   string `json:"status"`
-	Enabled    bool       `json:"enabled"`
+	Result   string `json:"result"`
-	NextRunAt  *time.Time `json:"nextRunAt"`
+}
-	LastRunAt  *time.Time `json:"lastRunAt"`
+
-	LastResult string     `json:"lastResult"`
+type DepositDTO struct {
-	CreatedAt  time.Time  `json:"createdAt"`
+	NoteID    string `json:"noteId"`
-	UpdatedAt  time.Time  `json:"updatedAt"`
+	ProjectID string `json:"projectId"`
 }
 
 type sourceRequest struct {
@@ -79,8 +87,8 @@ func (h *Handler) Register(router gin.IRouter) {
 	router.GET("/dataset-items", h.listItems)
 	router.POST("/dataset-items", h.createItem)
 	router.PATCH("/dataset-items/:id", h.updateItem)
-	router.GET("/dataset-crons", h.listCrons)
+	router.POST("/dataset-items/:id/deposit", h.depositItem)
-	router.POST("/dataset-crons/sync", h.queueSync)
+	router.POST("/dataset-sources/sync", h.syncSources)
 }
 
 func (h *Handler) listSources(c *gin.Context) {
@@ -134,18 +142,21 @@ func (h *Handler) updateSource(c *gin.Context) {
 	if !ok {
 		return
 	}
-	var input sourceRequest
+	var input struct {
+		Name        *string `json:"name"`
+		Kind        *string `json:"kind"`
+		URL         *string `json:"url"`
+		IconURL     *string `json:"iconUrl"`
+		Description *string `json:"description"`
+		Enabled     *bool   `json:"enabled"`
+	}
 	if err := c.ShouldBindJSON(&input); err != nil {
 		writeInvalidRequest(c)
 		return
 	}
-	enabled := true
+	source, err := h.service.PatchSource(userID, identity, SourcePatch{
-	if input.Enabled != nil {
-		enabled = *input.Enabled
-	}
-	source, err := h.service.UpdateSource(userID, identity, SourceInput{
 		Name: input.Name, Kind: input.Kind, URL: input.URL, IconURL: input.IconURL,
-		Description: input.Description, Enabled: enabled,
+		Description: input.Description, Enabled: input.Enabled,
 	})
 	if err != nil {
 		writeError(c, err)
@@ -180,16 +191,34 @@ func (h *Handler) listItems(c *gin.Context) {
 	if !ok {
 		return
 	}
-	items, err := h.service.ListItems(userID, c.Query("sourceId"))
+	limit, err := optionalNonNegativeInt(c.Query("limit"))
+	if err != nil {
+		writeInvalidRequest(c)
+		return
+	}
+	offset, err := optionalNonNegativeInt(c.Query("offset"))
+	if err != nil {
+		writeInvalidRequest(c)
+		return
+	}
+	page, err := h.service.ListItemsPage(userID, c.Query("sourceId"), limit, offset)
 	if err != nil {
 		writeError(c, err)
 		return
 	}
-	result := make([]ItemDTO, 0, len(items))
+	result := make([]ItemDTO, 0, len(page.Items))
-	for _, item := range items {
+	for _, item := range page.Items {
 		result = append(result, itemDTO(item))
 	}
+	_, hasLimit := c.GetQuery("limit")
+	_, hasOffset := c.GetQuery("offset")
+	if !hasLimit && !hasOffset {
 		c.JSON(http.StatusOK, result)
+		return
+	}
+	c.JSON(http.StatusOK, ItemPageDTO{
+		Items: result, Total: page.Total, Limit: page.Limit, Offset: page.Offset,
+	})
 }
 
 func (h *Handler) createItem(c *gin.Context) {
@@ -203,6 +232,7 @@ func (h *Handler) createItem(c *gin.Context) {
 		Summary     string `json:"summary"`
 		Content     string `json:"content"`
 		URL         string `json:"url"`
+		ImageURL    string `json:"imageUrl"`
 		PublishedAt string `json:"publishedAt"`
 	}
 	if err := c.ShouldBindJSON(&input); err != nil {
@@ -216,7 +246,8 @@ func (h *Handler) createItem(c *gin.Context) {
 	}
 	item, err := h.service.CreateItem(userID, ItemInput{
 		SourceIdentity: input.SourceID, Title: input.Title, Summary: input.Summary,
-		Content: input.Content, URL: input.URL, PublishedAt: publishedAt,
+		Content: input.Content, URL: input.URL, ImageURL: input.ImageURL,
+		PublishedAt: publishedAt,
 	})
 	if err != nil {
 		writeError(c, err)
@@ -235,14 +266,14 @@ func (h *Handler) updateItem(c *gin.Context) {
 		return
 	}
 	var input struct {
-		Status  string `json:"status"`
+		Status  *string `json:"status"`
-		Starred bool   `json:"starred"`
+		Starred *bool   `json:"starred"`
 	}
 	if err := c.ShouldBindJSON(&input); err != nil {
 		writeInvalidRequest(c)
 		return
 	}
-	item, err := h.service.UpdateItem(userID, identity, ItemUpdate{Status: input.Status, Starred: input.Starred})
+	item, err := h.service.PatchItem(userID, identity, ItemPatch{Status: input.Status, Starred: input.Starred})
 	if err != nil {
 		writeError(c, err)
 		return
@@ -250,38 +281,51 @@ func (h *Handler) updateItem(c *gin.Context) {
 	c.JSON(http.StatusOK, itemDTO(*item))
 }
 
-func (h *Handler) listCrons(c *gin.Context) {
+func (h *Handler) depositItem(c *gin.Context) {
 	userID, ok := currentUser(c)
 	if !ok {
 		return
 	}
-	crons, err := h.service.ListCrons(userID)
+	identity, ok := httpx.IdentityParam(c, "id")
+	if !ok {
+		return
+	}
+	var input struct {
+		ProjectID string `json:"projectId"`
+	}
+	if err := c.ShouldBindJSON(&input); err != nil || strings.TrimSpace(input.ProjectID) == "" {
+		writeInvalidRequest(c)
+		return
+	}
+	result, err := h.service.DepositItem(userID, identity, input.ProjectID)
 	if err != nil {
 		writeError(c, err)
 		return
 	}
-	result := make([]CronDTO, 0, len(crons))
+	c.JSON(http.StatusCreated, DepositDTO{
-	for _, cron := range crons {
+		NoteID: result.NoteIdentity, ProjectID: result.ProjectIdentity,
-		result = append(result, cronDTO(cron))
+	})
-	}
-	c.JSON(http.StatusOK, result)
 }
 
-func (h *Handler) queueSync(c *gin.Context) {
+func (h *Handler) syncSources(c *gin.Context) {
 	userID, ok := currentUser(c)
 	if !ok {
 		return
 	}
-	crons, err := h.service.QueueSync(userID)
+	results, err := h.service.SyncSources(c.Request.Context(), userID)
 	if err != nil {
 		writeError(c, err)
 		return
 	}
-	result := make([]CronDTO, 0, len(crons))
+	response := make([]SyncResultDTO, 0, len(results))
-	for _, cron := range crons {
+	for _, result := range results {
-		result = append(result, cronDTO(cron))
+		response = append(response, SyncResultDTO{
+			SourceID: result.SourceIdentity,
+			Status:   result.Status,
+			Result:   result.Result,
+		})
 	}
-	c.JSON(http.StatusAccepted, result)
+	c.JSON(http.StatusOK, response)
 }
 
 func currentUser(c *gin.Context) (uint, bool) {
@@ -297,6 +341,10 @@ func writeError(c *gin.Context, err error) {
 	switch {
 	case errors.Is(err, gorm.ErrRecordNotFound):
 		httpx.Error(c, http.StatusNotFound, "not_found", "数据源或数据条目不存在")
+	case errors.Is(err, ErrBuiltInSource):
+		httpx.Error(c, http.StatusConflict, "built_in_source", "内置数据源不能删除，可以将其停用")
+	case errors.Is(err, ErrSyncInProgress):
+		httpx.Error(c, http.StatusConflict, "sync_in_progress", "已有数据源采集任务正在运行")
 	case errors.Is(err, ErrNameRequired), errors.Is(err, ErrKindInvalid),
 		errors.Is(err, ErrURLInvalid), errors.Is(err, ErrTitleRequired), errors.Is(err, ErrStatusInvalid):
 		writeInvalidRequest(c)
@@ -313,7 +361,7 @@ func sourceDTO(source models.SaDatasetSource, itemCount int64) SourceDTO {
 	return SourceDTO{
 		ID: source.Identity, Name: source.Name, Kind: source.Kind, URL: source.URL,
 		IconURL: source.IconURL, Description: source.Description, Enabled: source.Enabled,
-		LastSyncedAt: utcTime(source.LastSyncedAt), ItemCount: itemCount,
+		BuiltIn: source.SeedKey != nil, LastSyncedAt: utcTime(source.LastSyncedAt), ItemCount: itemCount,
 		CreatedAt: source.CreatedAt.UTC(), UpdatedAt: source.UpdatedAt.UTC(),
 	}
 }
@@ -321,19 +369,22 @@ func sourceDTO(source models.SaDatasetSource, itemCount int64) SourceDTO {
 func itemDTO(item models.SaDatasetItem) ItemDTO {
 	return ItemDTO{
 		ID: item.Identity, SourceID: item.SourceIdentity, Title: item.Title,
-		Summary: item.Summary, Content: item.Content, URL: item.URL, Status: item.Status,
+		Summary: item.Summary, Content: item.Content, URL: item.URL,
+		ImageURL: item.ImageURL, Status: item.Status,
 		Starred: item.Starred, PublishedAt: utcTime(item.PublishedAt),
 		CreatedAt: item.CreatedAt.UTC(), UpdatedAt: item.UpdatedAt.UTC(),
 	}
 }
 
-func cronDTO(cron models.SaDatasetCron) CronDTO {
+func optionalNonNegativeInt(value string) (int, error) {
-	return CronDTO{
+	if strings.TrimSpace(value) == "" {
-		ID: cron.Identity, SourceID: cron.SourceIdentity, Schedule: cron.Schedule,
+		return 0, nil
-		Status: cron.Status, Enabled: cron.Enabled, NextRunAt: utcTime(cron.NextRunAt),
-		LastRunAt: utcTime(cron.LastRunAt), LastResult: cron.LastResult,
-		CreatedAt: cron.CreatedAt.UTC(), UpdatedAt: cron.UpdatedAt.UTC(),
 	}
+	parsed, err := strconv.Atoi(value)
+	if err != nil || parsed < 0 {
+		return 0, errors.New("invalid non-negative integer")
+	}
+	return parsed, nil
 }
 
 func parseOptionalTime(value string) (*time.Time, error) {

backend/internal/logic/dataset/handlers_test.go

@@ -2,6 +2,7 @@ package dataset
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -20,12 +21,18 @@ func TestDatasetAPIConnectsSourcesItemsAndSyncTasks(t *testing.T) {
 	database := newDatasetTestDatabase(t)
 	owner := createDatasetTestUser(t, database, "owner@example.com")
 	other := createDatasetTestUser(t, database, "other@example.com")
+	project := models.SaProject{OwnerID: owner.ID, Name: "Research"}
+	require.NoError(t, database.Create(&project).Error)
 	ownerRouter := datasetTestRouter(database, owner.ID)
 
 	invalid := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-sources", map[string]any{
 		"name": "Invalid RSS", "kind": "rss", "url": "",
 	})
 	require.Equal(t, http.StatusBadRequest, invalid.Code)
+	unsupported := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-sources", map[string]any{
+		"name": "Manual source", "kind": "manual", "enabled": true,
+	})
+	require.Equal(t, http.StatusBadRequest, unsupported.Code)
 
 	createdSource := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-sources", map[string]any{
 		"name": "Industry feed", "kind": "rss", "url": "https://example.com/feed",
@@ -37,15 +44,25 @@ func TestDatasetAPIConnectsSourcesItemsAndSyncTasks(t *testing.T) {
 	require.NotEmpty(t, source.ID)
 	require.Equal(t, "Industry feed", source.Name)
 	require.Empty(t, source.IconURL)
+	var storedSource models.SaDatasetSource
+	require.NoError(t, database.Where("identity = ?", source.ID).First(&storedSource).Error)
+	require.Equal(t, owner.ID, storedSource.OwnerID)
+	require.Equal(t, owner.Identity, storedSource.OwnerIdentity)
 
 	createdItem := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-items", map[string]any{
 		"sourceId": source.ID, "title": "A collected article", "summary": "Summary",
-		"url": "https://example.com/article",
+		"url": "https://example.com/article", "imageUrl": "https://example.com/article.jpg",
 	})
 	require.Equal(t, http.StatusCreated, createdItem.Code)
 	var item ItemDTO
 	require.NoError(t, json.Unmarshal(createdItem.Body.Bytes(), &item))
 	require.Equal(t, source.ID, item.SourceID)
+	require.Equal(t, "https://example.com/article.jpg", item.ImageURL)
+	legacyItems := performDatasetRequest(t, ownerRouter, http.MethodGet, "/api/v1/dataset-items", nil)
+	require.Equal(t, http.StatusOK, legacyItems.Code)
+	var legacyItemList []ItemDTO
+	require.NoError(t, json.Unmarshal(legacyItems.Body.Bytes(), &legacyItemList))
+	require.Len(t, legacyItemList, 1)
 
 	listSources := performDatasetRequest(t, ownerRouter, http.MethodGet, "/api/v1/dataset-sources", nil)
 	require.Equal(t, http.StatusOK, listSources.Code)
@@ -61,19 +78,41 @@ func TestDatasetAPIConnectsSourcesItemsAndSyncTasks(t *testing.T) {
 	require.NoError(t, json.Unmarshal(updatedItem.Body.Bytes(), &item))
 	require.Equal(t, "read", item.Status)
 	require.True(t, item.Starred)
+	partialItemUpdate := performDatasetRequest(t, ownerRouter, http.MethodPatch, "/api/v1/dataset-items/"+item.ID, map[string]any{
+		"starred": false,
+	})
+	require.Equal(t, http.StatusOK, partialItemUpdate.Code)
+	require.NoError(t, json.Unmarshal(partialItemUpdate.Body.Bytes(), &item))
+	require.Equal(t, "read", item.Status)
+	require.False(t, item.Starred)
 
-	firstSync := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-crons/sync", map[string]any{})
+	deposited := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-items/"+item.ID+"/deposit", map[string]any{
-	require.Equal(t, http.StatusAccepted, firstSync.Code)
+		"projectId": project.Identity,
-	var firstCrons []CronDTO
+	})
-	require.NoError(t, json.Unmarshal(firstSync.Body.Bytes(), &firstCrons))
+	require.Equal(t, http.StatusCreated, deposited.Code)
-	require.Len(t, firstCrons, 1)
+	var depositedNote models.SaNote
-	require.Equal(t, "pending", firstCrons[0].Status)
+	require.NoError(t, database.Where("project_id = ?", project.ID).First(&depositedNote).Error)
+	require.Equal(t, item.Title, depositedNote.Title)
+	require.Contains(t, depositedNote.Markdown, "![文章图片](https://example.com/article.jpg)")
+	require.Contains(t, depositedNote.Markdown, "原文链接")
 
-	secondSync := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-crons/sync", map[string]any{})
+	firstSync := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-sources/sync", map[string]any{})
-	require.Equal(t, http.StatusAccepted, secondSync.Code)
+	require.Equal(t, http.StatusOK, firstSync.Code)
-	var secondCrons []CronDTO
+	var firstResults []SyncResultDTO
-	require.NoError(t, json.Unmarshal(secondSync.Body.Bytes(), &secondCrons))
+	require.NoError(t, json.Unmarshal(firstSync.Body.Bytes(), &firstResults))
-	require.Equal(t, firstCrons[0].ID, secondCrons[0].ID)
+	require.Len(t, firstResults, 1)
+	require.Equal(t, "completed", firstResults[0].Status)
+	require.Contains(t, firstResults[0].Result, "format=rss")
+
+	secondSync := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-sources/sync", map[string]any{})
+	require.Equal(t, http.StatusOK, secondSync.Code)
+	var secondResults []SyncResultDTO
+	require.NoError(t, json.Unmarshal(secondSync.Body.Bytes(), &secondResults))
+	require.Len(t, secondResults, 1)
+	require.Equal(t, "completed", secondResults[0].Status)
+	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)
@@ -86,11 +125,106 @@ func TestDatasetAPIConnectsSourcesItemsAndSyncTasks(t *testing.T) {
 
 	deleted := performDatasetRequest(t, ownerRouter, http.MethodDelete, "/api/v1/dataset-sources/"+source.ID, nil)
 	require.Equal(t, http.StatusNoContent, deleted.Code)
-	var itemCount, cronCount int64
+	var itemCount int64
 	require.NoError(t, database.Model(&models.SaDatasetItem{}).Count(&itemCount).Error)
-	require.NoError(t, database.Model(&models.SaDatasetCron{}).Count(&cronCount).Error)
 	require.Zero(t, itemCount)
-	require.Zero(t, cronCount)
+}
+
+func TestDatasetSourceAuthorizationUsesOwner(t *testing.T) {
+	database := newDatasetTestDatabase(t)
+	owner := createDatasetTestUser(t, database, "owner@example.com")
+	creator := createDatasetTestUser(t, database, "creator@example.com")
+	source := models.SaDatasetSource{
+		OwnerID: owner.ID,
+		Name:    "Owned feed", Kind: "rss", URL: "https://example.com/feed", Enabled: true,
+	}
+	require.NoError(t, database.Create(&source).Error)
+
+	ownerSources, err := NewService(database).ListSources(owner.ID)
+	require.NoError(t, err)
+	require.Len(t, ownerSources, 1)
+
+	creatorSources, err := NewService(database).ListSources(creator.ID)
+	require.NoError(t, err)
+	require.Empty(t, creatorSources)
+
+	item := models.SaDatasetItem{
+		SourceID: source.ID, Title: "Collected item", Status: "unread",
+	}
+	require.NoError(t, database.Create(&item).Error)
+	page, err := NewService(database).ListItemsPage(owner.ID, source.Identity, 50, 0)
+	require.NoError(t, err)
+	require.Len(t, page.Items, 1)
+	creatorPage, err := NewService(database).ListItemsPage(creator.ID, "", 50, 0)
+	require.NoError(t, err)
+	require.Empty(t, creatorPage.Items)
+	creatorProject := models.SaProject{OwnerID: creator.ID, Name: "Creator project"}
+	require.NoError(t, database.Create(&creatorProject).Error)
+	_, err = NewService(database).DepositItem(owner.ID, item.Identity, creatorProject.Identity)
+	require.ErrorIs(t, err, gorm.ErrRecordNotFound)
+	require.NoError(t, NewService(database).DeleteSource(owner.ID, source.Identity))
+	var itemCount int64
+	require.NoError(t, database.Model(&models.SaDatasetItem{}).Count(&itemCount).Error)
+	require.Zero(t, itemCount)
+}
+
+func TestDatasetItemsArePaginatedByOwnedSource(t *testing.T) {
+	database := newDatasetTestDatabase(t)
+	owner := createDatasetTestUser(t, database, "owner@example.com")
+	source := models.SaDatasetSource{
+		OwnerID: owner.ID,
+		Name:    "Feed", Kind: "rss", URL: "https://example.com/feed", Enabled: true,
+	}
+	require.NoError(t, database.Create(&source).Error)
+	for index := 0; index < 3; index++ {
+		require.NoError(t, database.Create(&models.SaDatasetItem{
+			SourceID: source.ID,
+			Title:    fmt.Sprintf("Item %d", index), Status: "unread",
+		}).Error)
+	}
+	router := datasetTestRouter(database, owner.ID)
+
+	response := performDatasetRequest(t, router, http.MethodGet,
+		"/api/v1/dataset-items?sourceId="+source.Identity+"&limit=2&offset=1", nil)
+
+	require.Equal(t, http.StatusOK, response.Code)
+	var page ItemPageDTO
+	require.NoError(t, json.Unmarshal(response.Body.Bytes(), &page))
+	require.Equal(t, int64(3), page.Total)
+	require.Equal(t, 2, page.Limit)
+	require.Equal(t, 1, page.Offset)
+	require.Len(t, page.Items, 2)
+}
+
+func TestBuiltInDatasetSourceKeepsManagedFieldsAndCannotBeDeleted(t *testing.T) {
+	database := newDatasetTestDatabase(t)
+	owner := createDatasetTestUser(t, database, "owner@example.com")
+	seedKey := "built-in"
+	source := models.SaDatasetSource{
+		OwnerID: owner.ID, SeedKey: &seedKey,
+		Name: "Built-in", Kind: "rss", URL: "https://example.com/feed",
+		IconURL: "https://example.com/icon.png", Enabled: true,
+	}
+	require.NoError(t, database.Create(&source).Error)
+	service := NewService(database)
+
+	updated, err := service.UpdateSource(owner.ID, source.Identity, SourceInput{
+		Name: "Renamed", Kind: "rss", URL: "https://attacker.example/feed",
+		IconURL: "https://attacker.example/icon.png", Enabled: false,
+	})
+
+	require.NoError(t, err)
+	require.Equal(t, "Renamed", updated.Name)
+	require.Equal(t, "https://example.com/feed", updated.URL)
+	require.Equal(t, "https://example.com/icon.png", updated.IconURL)
+	require.False(t, updated.Enabled)
+	renamed := "Renamed again"
+	updated, err = service.PatchSource(owner.ID, source.Identity, SourcePatch{Name: &renamed})
+	require.NoError(t, err)
+	require.Equal(t, renamed, updated.Name)
+	require.False(t, updated.Enabled)
+	require.Equal(t, "https://example.com/feed", updated.URL)
+	require.ErrorIs(t, service.DeleteSource(owner.ID, source.Identity), ErrBuiltInSource)
 }
 
 func newDatasetTestDatabase(t *testing.T) *gorm.DB {
@@ -101,7 +235,8 @@ func newDatasetTestDatabase(t *testing.T) *gorm.DB {
 		&models.SaUser{},
 		&models.SaDatasetSource{},
 		&models.SaDatasetItem{},
-		&models.SaDatasetCron{},
+		&models.SaProject{},
+		&models.SaNote{},
 	))
 	return database
 }
@@ -117,10 +252,26 @@ 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",
+			ImageURL:   "https://example.com/collected.jpg",
+		}},
+	}, nil
+}
+
 func performDatasetRequest(t *testing.T, router http.Handler, method, path string, body any) *httptest.ResponseRecorder {
 	t.Helper()
 	var encoded []byte

backend/internal/logic/dataset/service.go

@@ -1,12 +1,16 @@
 package dataset
 
 import (
+	"context"
 	"errors"
+	"fmt"
 	"net/url"
 	"strings"
+	"sync"
 	"time"
 
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 	"senlinai-agent/backend/internal/models"
 )
 
@@ -16,10 +20,20 @@ var (
 	ErrURLInvalid     = errors.New("dataset source url is invalid")
 	ErrTitleRequired  = errors.New("dataset item title is required")
 	ErrStatusInvalid  = errors.New("dataset item status is invalid")
+	ErrBuiltInSource  = errors.New("built-in dataset source cannot be deleted")
+	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 {
@@ -31,12 +45,22 @@ type SourceInput struct {
 	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
 }
 
@@ -45,18 +69,45 @@ type ItemUpdate struct {
 	Starred bool
 }
 
+type ItemPatch struct {
+	Status  *string
+	Starred *bool
+}
+
+type DepositResult struct {
+	NoteIdentity    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}
+	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("created_by = ?", userID).Order("created_at asc, id asc").Find(&sources).Error; err != nil {
+	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)
@@ -71,7 +122,7 @@ func (s *Service) ListSources(userID uint) ([]SourceRecord, error) {
 		}
 		if err := s.db.Model(&models.SaDatasetItem{}).
 			Select("source_id, COUNT(*) AS count").
-			Where("created_by = ? AND source_id IN ?", userID, sourceIDs).
+			Where("source_id IN ?", sourceIDs).
 			Group("source_id").
 			Scan(&rows).Error; err != nil {
 			return nil, err
@@ -93,7 +144,7 @@ func (s *Service) CreateSource(userID uint, input SourceInput) (*models.SaDatase
 		return nil, err
 	}
 	source := &models.SaDatasetSource{
-		CreatedBy: userID, Name: normalized.Name, Kind: normalized.Kind,
+		OwnerID: userID, Name: normalized.Name, Kind: normalized.Kind,
 		URL: normalized.URL, IconURL: normalized.IconURL,
 		Description: normalized.Description, Enabled: normalized.Enabled,
 	}
@@ -101,12 +152,58 @@ func (s *Service) CreateSource(userID uint, input SourceInput) (*models.SaDatase
 }
 
 func (s *Service) UpdateSource(userID uint, identity string, input SourceInput) (*models.SaDatasetSource, error) {
-	normalized, err := normalizeSourceInput(input)
+	return s.PatchSource(userID, identity, SourcePatch{
-	if err != nil {
+		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
 	}
-	var source models.SaDatasetSource
+	input := SourceInput{
-	if err := s.db.Where("identity = ? AND created_by = ?", identity, userID).First(&source).Error; err != nil {
+		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{
@@ -119,15 +216,20 @@ func (s *Service) UpdateSource(userID uint, identity string, input SourceInput)
 }
 
 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 created_by = ?", identity, userID).First(&source).Error; err != nil {
+		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 {
+		if source.SeedKey != nil {
-			return err
+			return ErrBuiltInSource
 		}
-		if err := tx.Where("source_id = ? AND created_by = ?", source.ID, userID).Delete(&models.SaDatasetItem{}).Error; err != nil {
+		if err := tx.Where("source_id = ?", source.ID).Delete(&models.SaDatasetItem{}).Error; err != nil {
 			return err
 		}
 		return tx.Delete(&source).Error
@@ -135,28 +237,46 @@ func (s *Service) DeleteSource(userID uint, identity string) error {
 }
 
 func (s *Service) ListItems(userID uint, sourceIdentity string) ([]models.SaDatasetItem, error) {
-	query := s.db.Where("created_by = ?", userID)
+	page, err := s.ListItemsPage(userID, sourceIdentity, maxItemPageSize, 0)
-	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)
+	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.Order("COALESCE(published_at, created_at) desc, id desc").Limit(200).Find(&items).Error; err != nil {
+	if err := query.Select("sa_dataset_items.*").
-		return nil, err
+		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 items, nil
+	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{}).
-		Where("created_by = ? AND source_id = ?", userID, sourceID).
+		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
 }
@@ -174,78 +294,279 @@ func (s *Service) CreateItem(userID uint, input ItemInput) (*models.SaDatasetIte
 	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, CreatedBy: userID, Title: title,
+		SourceID: source.ID, Title: title,
 		Summary: strings.TrimSpace(input.Summary), Content: strings.TrimSpace(input.Content),
-		URL: itemURL, Status: "unread", PublishedAt: utcOptionalTime(input.PublishedAt),
+		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) {
-	status := strings.TrimSpace(input.Status)
+	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
 		}
-	var item models.SaDatasetItem
+		updates["status"] = status
-	if err := s.db.Where("identity = ? AND created_by = ?", identity, userID).First(&item).Error; err != nil {
+	}
+	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
 	}
-	if err := s.db.Model(&item).Updates(map[string]any{"status": status, "starred": input.Starred}).Error; err != nil {
+	return item, nil
-		return nil, err
-	}
-	return &item, nil
 }
 
-func (s *Service) ListCrons(userID uint) ([]models.SaDatasetCron, error) {
+func (s *Service) DepositItem(userID uint, itemIdentity, projectIdentity string) (DepositResult, error) {
-	var crons []models.SaDatasetCron
+	item, err := s.findOwnedItem(userID, itemIdentity)
-	if err := s.db.Where("created_by = ?", userID).Order("created_at desc, id desc").Limit(100).Find(&crons).Error; err != nil {
+	if err != nil {
-		return nil, err
+		return DepositResult{}, err
 	}
-	if crons == nil {
+	var project models.SaProject
-		crons = []models.SaDatasetCron{}
+	if err := s.db.Where("identity = ? AND owner_id = ?", strings.TrimSpace(projectIdentity), userID).
+		First(&project).Error; err != nil {
+		return DepositResult{}, err
 	}
-	return crons, nil
+	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 + ")"
+	}
+	note := models.SaNote{
+		ProjectID: project.ID, CreatedBy: userID,
+		Title: item.Title, Markdown: body,
+	}
+	if err := s.db.Create(&note).Error; err != nil {
+		return DepositResult{}, err
+	}
+	return DepositResult{NoteIdentity: note.Identity, ProjectIdentity: project.Identity}, nil
 }
 
-func (s *Service) QueueSync(userID uint) ([]models.SaDatasetCron, error) {
+func (s *Service) SyncSources(ctx context.Context, userID uint) ([]SyncResult, error) {
-	result := make([]models.SaDatasetCron, 0)
+	if !s.syncMu.TryLock() {
-	err := s.db.Transaction(func(tx *gorm.DB) error {
+		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 := tx.Where("created_by = ? AND enabled = ?", userID, true).Order("id asc").Find(&sources).Error; err != nil {
+	if err := s.db.Where("enabled = ? AND kind = ?", true, "rss").
-			return err
+		Order("owner_id asc, id asc").
+		Find(&sources).Error; err != nil {
+		return nil, err
 	}
-		now := time.Now().UTC()
+	return s.executeDatasetSources(ctx, sources)
-		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
+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
 	}
-			cron = models.SaDatasetCron{
+	return sources, nil
-				SourceID: source.ID, CreatedBy: userID, Schedule: "@once",
-				Status: "pending", Enabled: true, NextRunAt: &now,
 }
-			if err := tx.Create(&cron).Error; err != nil {
+
-				return err
+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
 	}
-			result = append(result, cron)
+	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)
+		}()
 	}
-		return nil
+	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 created_by = ?", strings.TrimSpace(identity), userID).First(&source).Error
+	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))
@@ -255,7 +576,7 @@ func normalizeSourceInput(input SourceInput) (SourceInput, error) {
 	if input.Name == "" {
 		return input, ErrNameRequired
 	}
-	if input.Kind != "manual" && input.Kind != "link" && input.Kind != "rss" {
+	if input.Kind != "rss" {
 		return input, ErrKindInvalid
 	}
 	if input.URL != "" && !validHTTPURL(input.URL) {
@@ -264,15 +585,38 @@ func normalizeSourceInput(input SourceInput) (SourceInput, error) {
 	if input.IconURL != "" && !validHTTPURL(input.IconURL) {
 		return input, ErrURLInvalid
 	}
-	if input.Kind != "manual" && input.URL == "" {
+	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 != ""
+	return err == nil && (parsed.Scheme == "http" || parsed.Scheme == "https") &&
+		parsed.Host != "" && parsed.User == nil
 }
 
 func utcOptionalTime(value *time.Time) *time.Time {

backend/internal/models/dataset_cron.go

@@ -1,24 +0,0 @@
-package models
-
-import "time"
-
-type SaDatasetCron struct {
-	ID                uint   `gorm:"primaryKey"`
-	Identity          string `gorm:"type:char(36);uniqueIndex"`
-	SourceID          uint   `gorm:"index;not null"`
-	SourceIdentity    string `gorm:"type:char(36);index"`
-	CreatedBy         uint   `gorm:"index;not null"`
-	CreatedByIdentity string `gorm:"type:char(36);index"`
-	Schedule          string `gorm:"size:100;not null"`
-	Status            string `gorm:"size:32;not null;default:pending;index"`
-	Enabled           bool   `gorm:"not null;default:true;index"`
-	NextRunAt         *time.Time
-	LastRunAt         *time.Time
-	LastResult        string `gorm:"type:text"`
-	CreatedAt         time.Time
-	UpdatedAt         time.Time
-}
-
-func (SaDatasetCron) TableName() string {
-	return "sa_dataset_crons"
-}

backend/internal/models/dataset_item.go

@@ -5,14 +5,14 @@ import "time"
 type SaDatasetItem struct {
 	ID             uint    `gorm:"primaryKey"`
 	Identity       string  `gorm:"type:char(36);uniqueIndex"`
-	SourceID          uint   `gorm:"index;not null"`
+	SourceID       uint    `gorm:"index;not null;uniqueIndex:idx_sa_dataset_item_source_external,priority:1"`
 	SourceIdentity string  `gorm:"type:char(36);index"`
-	CreatedBy         uint   `gorm:"index;not null"`
+	ExternalID     *string `gorm:"size:64;uniqueIndex:idx_sa_dataset_item_source_external,priority:2"`
-	CreatedByIdentity string `gorm:"type:char(36);index"`
 	Title          string  `gorm:"size:500;not null"`
 	Summary        string  `gorm:"type:text"`
 	Content        string  `gorm:"type:text"`
 	URL            string  `gorm:"size:2048"`
+	ImageURL       string  `gorm:"size:2048"`
 	Status         string  `gorm:"size:32;not null;default:unread;index"`
 	Starred        bool    `gorm:"not null;default:false;index"`
 	PublishedAt    *time.Time

backend/internal/models/dataset_source.go

@@ -5,8 +5,9 @@ import "time"
 type SaDatasetSource struct {
 	ID            uint    `gorm:"primaryKey"`
 	Identity      string  `gorm:"type:char(36);uniqueIndex"`
-	CreatedBy         uint   `gorm:"index;not null"`
+	OwnerID       uint    `gorm:"index;not null;default:0;uniqueIndex:idx_sa_dataset_source_owner_seed,priority:1"`
-	CreatedByIdentity string `gorm:"type:char(36);index"`
+	OwnerIdentity string  `gorm:"type:char(36);index"`
+	SeedKey       *string `gorm:"size:64;uniqueIndex:idx_sa_dataset_source_owner_seed,priority:2"`
 	Name          string  `gorm:"size:160;not null"`
 	Kind          string  `gorm:"size:32;not null"`
 	URL           string  `gorm:"size:2048"`

backend/internal/models/identity.go

@@ -2,6 +2,7 @@ package models
 
 import (
 	"database/sql"
+	"errors"
 
 	"github.com/google/uuid"
 	"gorm.io/gorm"
@@ -22,27 +23,17 @@ func (m *SaDatasetSource) BeforeCreate(tx *gorm.DB) error {
 	if err := ensureIdentity(&m.Identity); err != nil {
 		return err
 	}
-	return resolveIdentity(tx, &SaUser{}, m.CreatedBy, &m.CreatedByIdentity)
+	if m.OwnerID == 0 {
+		return errors.New("dataset source owner is required")
+	}
+	return resolveIdentity(tx, &SaUser{}, m.OwnerID, &m.OwnerIdentity)
 }
 
 func (m *SaDatasetItem) BeforeCreate(tx *gorm.DB) error {
 	if err := ensureIdentity(&m.Identity); err != nil {
 		return err
 	}
-	if err := resolveIdentity(tx, &SaDatasetSource{}, m.SourceID, &m.SourceIdentity); err != nil {
+	return resolveIdentity(tx, &SaDatasetSource{}, m.SourceID, &m.SourceIdentity)
-		return err
-	}
-	return resolveIdentity(tx, &SaUser{}, m.CreatedBy, &m.CreatedByIdentity)
-}
-
-func (m *SaDatasetCron) BeforeCreate(tx *gorm.DB) error {
-	if err := ensureIdentity(&m.Identity); err != nil {
-		return err
-	}
-	if err := resolveIdentity(tx, &SaDatasetSource{}, m.SourceID, &m.SourceIdentity); err != nil {
-		return err
-	}
-	return resolveIdentity(tx, &SaUser{}, m.CreatedBy, &m.CreatedByIdentity)
 }
 
 func (m *SaInboxItem) BeforeCreate(tx *gorm.DB) error {

backend/internal/models/identity_test.go

@@ -40,6 +40,18 @@ func TestBeforeCreateCopiesParentIdentitiesFromNumericIDs(t *testing.T) {
 	require.Equal(t, item.Identity, *task.SourceInboxItemIdentity)
 }
 
+func TestDatasetSourceBeforeCreateAssociatesOwner(t *testing.T) {
+	database := newIdentityTestDB(t)
+	user := SaUser{Email: "owner@example.com", DisplayName: "Owner", PasswordHash: "hash"}
+	require.NoError(t, database.Create(&user).Error)
+	source := SaDatasetSource{OwnerID: user.ID, Name: "Owned feed", Kind: "rss"}
+
+	require.NoError(t, database.Create(&source).Error)
+
+	require.Equal(t, user.ID, source.OwnerID)
+	require.Equal(t, user.Identity, source.OwnerIdentity)
+}
+
 func newIdentityTestDB(t *testing.T) *gorm.DB {
 	t.Helper()
 	database, err := gorm.Open(sqlite.Open(fmt.Sprintf("file:%s?mode=memory&cache=shared", t.Name())), &gorm.Config{})

backend/internal/models/new.go

@@ -33,7 +33,6 @@ func AutoMigrate(database *gorm.DB) error {
 		&SaProject{},
 		&SaDatasetSource{},
 		&SaDatasetItem{},
-		&SaDatasetCron{},
 		&SaInboxItem{},
 		&SaInboxSuggestion{},
 		&SaTask{},