From fbd142803af819bc55302c3a60c7239f950f7071 Mon Sep 17 00:00:00 2001 From: yanweidong Date: Thu, 23 Jul 2026 23:08:25 +0800 Subject: [PATCH] refactor: remove dataset cron records --- apps/web_v1/scripts/visual-check.mjs | 17 +- apps/web_v1/src/api/dataset.ts | 16 +- apps/web_v1/src/app/App.tsx | 15 +- apps/web_v1/src/pages/workspace-explore.tsx | 36 +--- apps/web_v1/src/pages/workspace-home.tsx | 11 +- backend/internal/crontab/dataset.go | 16 +- backend/internal/crontab/dataset_test.go | 8 +- backend/internal/logic/dataset/feed_test.go | 40 ++-- backend/internal/logic/dataset/handlers.go | 56 ++--- .../internal/logic/dataset/handlers_test.go | 60 ++---- backend/internal/logic/dataset/service.go | 195 ++++-------------- backend/internal/models/dataset_cron.go | 24 --- backend/internal/models/identity.go | 10 - backend/internal/models/new.go | 1 - 14 files changed, 125 insertions(+), 380 deletions(-) delete mode 100644 backend/internal/models/dataset_cron.go diff --git a/apps/web_v1/scripts/visual-check.mjs b/apps/web_v1/scripts/visual-check.mjs index 8b92a40..1bfe5f6 100644 --- a/apps/web_v1/scripts/visual-check.mjs +++ b/apps/web_v1/scripts/visual-check.mjs @@ -281,28 +281,17 @@ 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: 200, - json: visualDatasetSources.map((source, index) => ({ - id: `019b0000-0000-7000-8000-${String(70 + index).padStart(12, '0')}`, + json: visualDatasetSources.map((source) => ({ sourceId: source.id, - schedule: '@once', status: 'completed', - enabled: false, - nextRunAt: null, - lastRunAt: new Date().toISOString(), - lastResult: 'format=rss fetched=2 inserted=0', - createdAt: new Date().toISOString(), - updatedAt: new Date().toISOString(), + result: 'format=rss fetched=2 inserted=0', })), }) return } - if (url.pathname === '/api/v1/dataset-crons' && route.request().method() === 'GET') { - await route.fulfill({ json: [] }) - return - } await route.fulfill({ status: 404, json: { error: { code: 'not_found', message: '视觉检查未配置该接口' } } }) }) diff --git a/apps/web_v1/src/api/dataset.ts b/apps/web_v1/src/api/dataset.ts index fd80e8c..e264142 100644 --- a/apps/web_v1/src/api/dataset.ts +++ b/apps/web_v1/src/api/dataset.ts @@ -31,14 +31,10 @@ export type DatasetItemDTO = { updatedAt: string } -export type DatasetCronDTO = { - id: string +export type DatasetSyncResultDTO = { sourceId: string status: string - lastRunAt: string | null - lastResult: string - createdAt: string - updatedAt: string + result: string } export type DatasetItemPageDTO = { @@ -115,18 +111,14 @@ export function updateDatasetItem(session: ApiSession, itemId: string, input: Da }) } -export function queueDatasetSync(session: ApiSession) { - return apiRequest('/api/v1/dataset-crons/sync', { +export function syncDatasetSources(session: ApiSession) { + return apiRequest('/api/v1/dataset-sources/sync', { method: 'POST', token: session.token, body: {}, }) } -export function listDatasetCrons(session: ApiSession, signal?: AbortSignal) { - return apiRequest('/api/v1/dataset-crons', { token: session.token, signal }) -} - export function depositDatasetItem(session: ApiSession, itemId: string, projectId: string) { return apiRequest(`/api/v1/dataset-items/${itemId}/deposit`, { method: 'POST', diff --git a/apps/web_v1/src/app/App.tsx b/apps/web_v1/src/app/App.tsx index dcdaa91..ee8bde4 100644 --- a/apps/web_v1/src/app/App.tsx +++ b/apps/web_v1/src/app/App.tsx @@ -8,10 +8,9 @@ import { createDatasetSource, deleteDatasetSource, depositDatasetItem, - listDatasetCrons, listDatasetItems, listDatasetSources, - queueDatasetSync, + syncDatasetSources, updateDatasetItem, updateDatasetSource, type DatasetItemUpdate, @@ -102,14 +101,9 @@ function App() { return updateDatasetItem(session, itemId, input) }, [session]) - const handleQueueDatasetSync = useCallback(() => { + const handleSyncDatasetSources = useCallback(() => { if (!session) return Promise.reject(new Error('未登录')) - return queueDatasetSync(session) - }, [session]) - - const handleListDatasetCrons = useCallback((signal?: AbortSignal) => { - if (!session) return Promise.reject(new Error('未登录')) - return listDatasetCrons(session, signal) + return syncDatasetSources(session) }, [session]) const handleDepositDatasetItem = useCallback(async (itemId: string, projectId: string) => { @@ -440,8 +434,7 @@ function App() { onUpdateDatasetSource={handleUpdateDatasetSource} onDeleteDatasetSource={handleDeleteDatasetSource} onUpdateDatasetItem={handleUpdateDatasetItem} - onQueueDatasetSync={handleQueueDatasetSync} - onListDatasetCrons={handleListDatasetCrons} + onSyncDatasetSources={handleSyncDatasetSources} onDepositDatasetItem={handleDepositDatasetItem} /> ) : ( diff --git a/apps/web_v1/src/pages/workspace-explore.tsx b/apps/web_v1/src/pages/workspace-explore.tsx index 8138d25..8caafe4 100644 --- a/apps/web_v1/src/pages/workspace-explore.tsx +++ b/apps/web_v1/src/pages/workspace-explore.tsx @@ -14,7 +14,6 @@ import { IconStorage, } from '@arco-design/web-react/icon' import type { - DatasetCronDTO, DatasetDepositDTO, DatasetItemDTO, DatasetItemPageDTO, @@ -22,6 +21,7 @@ import type { DatasetSourceDTO, DatasetSourceInput, DatasetSourceKind, + DatasetSyncResultDTO, } from '../api/dataset' const { Row, Col } = Grid @@ -61,8 +61,7 @@ export function WorkspaceExplorePage({ onUpdateSource, onDeleteSource, onUpdateItem, - onQueueSync, - onListCrons, + onSyncSources, onDepositItem, }: { activeProjectID: string @@ -73,8 +72,7 @@ export function WorkspaceExplorePage({ onUpdateSource: (sourceId: string, input: DatasetSourceInput) => Promise onDeleteSource: (sourceId: string) => Promise onUpdateItem: (itemId: string, input: DatasetItemUpdate) => Promise - onQueueSync: () => Promise - onListCrons: (signal?: AbortSignal) => Promise + onSyncSources: () => Promise onDepositItem: (itemId: string, projectId: string) => Promise }) { const [sourceRecords, setSourceRecords] = useState([]) @@ -265,12 +263,10 @@ export function WorkspaceExplorePage({ async function syncSources() { setSyncing(true) try { - const queued = await onQueueSync() - if (queued.length === 0) { + const results = await onSyncSources() + if (results.length === 0) { Message.info('暂无已启用的 RSS 数据源') } else { - Message.info(`已提交 ${queued.length} 个采集任务`) - const crons = await waitForDatasetRuns(queued, onListCrons) const sourceID = activeSourceID === 'all' ? undefined : activeSourceID const [nextSources, nextPage] = await Promise.all([ onListSources(), @@ -279,11 +275,11 @@ export function WorkspaceExplorePage({ setSourceRecords(nextSources) setItems(nextPage.items) setItemTotal(nextPage.total) - const failed = crons.filter((cron) => cron.status === 'failed').length + const failed = results.filter((result) => result.status === 'failed').length if (failed > 0) { - Message.warning(`${crons.length - failed} 个数据源采集完成，${failed} 个失败`) + Message.warning(`${results.length - failed} 个数据源采集完成，${failed} 个失败`) } else { - Message.success(`已完成 ${crons.length} 个数据源采集任务`) + Message.success(`已完成 ${results.length} 个数据源采集任务`) } } } catch (error) { @@ -547,19 +543,3 @@ function sourceSubtitle(source: DatasetSourceDTO) { if (Number.isNaN(parsed.getTime())) return `${source.itemCount} 条` return `${source.itemCount} 条 · ${new Intl.DateTimeFormat('zh-CN', { dateStyle: 'short', timeStyle: 'short' }).format(parsed)}` } - -async function waitForDatasetRuns( - queued: DatasetCronDTO[], - listRuns: (signal?: AbortSignal) => Promise, -) { - const queuedIDs = new Set(queued.map((run) => run.id)) - const deadline = Date.now() + 10 * 60 * 1000 - while (Date.now() < deadline) { - const runs = (await listRuns()).filter((run) => queuedIDs.has(run.id)) - if (runs.length === queuedIDs.size && runs.every((run) => run.status === 'completed' || run.status === 'failed')) { - return runs - } - await new Promise((resolve) => setTimeout(resolve, 1500)) - } - throw new Error('采集任务仍在后台运行，请稍后刷新') -} diff --git a/apps/web_v1/src/pages/workspace-home.tsx b/apps/web_v1/src/pages/workspace-home.tsx index 24a1ebf..1b512d4 100644 --- a/apps/web_v1/src/pages/workspace-home.tsx +++ b/apps/web_v1/src/pages/workspace-home.tsx @@ -11,13 +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' @@ -63,8 +63,7 @@ export function ProjectPage({ onUpdateDatasetSource, onDeleteDatasetSource, onUpdateDatasetItem, - onQueueDatasetSync, - onListDatasetCrons, + onSyncDatasetSources, onDepositDatasetItem, }: { activeView: WorkbenchView @@ -106,8 +105,7 @@ export function ProjectPage({ onUpdateDatasetSource: (sourceId: string, input: DatasetSourceInput) => Promise onDeleteDatasetSource: (sourceId: string) => Promise onUpdateDatasetItem: (itemId: string, input: DatasetItemUpdate) => Promise - onQueueDatasetSync: () => Promise - onListDatasetCrons: (signal?: AbortSignal) => Promise + onSyncDatasetSources: () => Promise onDepositDatasetItem: (itemId: string, projectId: string) => Promise }) { const isProject = activeView === 'project' @@ -173,8 +171,7 @@ export function ProjectPage({ onUpdateSource={onUpdateDatasetSource} onDeleteSource={onDeleteDatasetSource} onUpdateItem={onUpdateDatasetItem} - onQueueSync={onQueueDatasetSync} - onListCrons={onListDatasetCrons} + onSyncSources={onSyncDatasetSources} onDepositItem={onDepositDatasetItem} /> ) : ( diff --git a/backend/internal/crontab/dataset.go b/backend/internal/crontab/dataset.go index 4e3444a..00b8304 100644 --- a/backend/internal/crontab/dataset.go +++ b/backend/internal/crontab/dataset.go @@ -7,13 +7,13 @@ import ( "sync/atomic" "time" - "senlinai-agent/backend/internal/models" + "senlinai-agent/backend/internal/logic/dataset" ) -const datasetInterval = 5 * time.Minute +var datasetInterval = 5 * time.Minute type DatasetCollector interface { - SyncAllSources(context.Context) ([]models.SaDatasetCron, error) + SyncAllSources(context.Context) ([]dataset.SyncResult, error) } type DatasetScheduler struct { @@ -70,11 +70,11 @@ func (s *DatasetScheduler) Run(ctx context.Context) { } func (s *DatasetScheduler) collect(ctx context.Context) { - crons, err := s.collector.SyncAllSources(ctx) + results, err := s.collector.SyncAllSources(ctx) completed := 0 failed := 0 - for _, cron := range crons { - switch cron.Status { + for _, result := range results { + switch result.Status { case "completed": completed++ case "failed": @@ -82,8 +82,8 @@ func (s *DatasetScheduler) collect(ctx context.Context) { } } if err != nil { - s.logger.Printf("dataset collection failed: tasks=%d completed=%d failed=%d error=%v", len(crons), completed, failed, err) + 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: tasks=%d completed=%d failed=%d", len(crons), completed, failed) + s.logger.Printf("dataset collection finished: sources=%d completed=%d failed=%d", len(results), completed, failed) } diff --git a/backend/internal/crontab/dataset_test.go b/backend/internal/crontab/dataset_test.go index 243b164..6e3ab9a 100644 --- a/backend/internal/crontab/dataset_test.go +++ b/backend/internal/crontab/dataset_test.go @@ -9,7 +9,7 @@ import ( "time" "github.com/stretchr/testify/require" - "senlinai-agent/backend/internal/models" + "senlinai-agent/backend/internal/logic/dataset" ) func TestDatasetSchedulerCollectsImmediatelyAndRepeats(t *testing.T) { @@ -95,13 +95,13 @@ type recordingDatasetCollector struct { calls chan time.Time } -func (c *recordingDatasetCollector) SyncAllSources(context.Context) ([]models.SaDatasetCron, error) { +func (c *recordingDatasetCollector) SyncAllSources(context.Context) ([]dataset.SyncResult, error) { now := time.Now() c.mu.Lock() c.count++ c.mu.Unlock() c.calls <- now - return []models.SaDatasetCron{{Status: "completed"}}, nil + return []dataset.SyncResult{{Status: "completed"}}, nil } func (c *recordingDatasetCollector) callCount() int { @@ -128,7 +128,7 @@ type blockingDatasetCollector struct { release chan struct{} } -func (c *blockingDatasetCollector) SyncAllSources(ctx context.Context) ([]models.SaDatasetCron, error) { +func (c *blockingDatasetCollector) SyncAllSources(ctx context.Context) ([]dataset.SyncResult, error) { c.mu.Lock() c.count++ c.mu.Unlock() diff --git a/backend/internal/logic/dataset/feed_test.go b/backend/internal/logic/dataset/feed_test.go index 2d175fb..e7e090b 100644 --- a/backend/internal/logic/dataset/feed_test.go +++ b/backend/internal/logic/dataset/feed_test.go @@ -7,7 +7,6 @@ import ( "net/url" "os" "testing" - "time" "github.com/stretchr/testify/require" "senlinai-agent/backend/internal/models" @@ -136,13 +135,12 @@ func TestSyncSourcesRecordsFeedFailure(t *testing.T) { } require.NoError(t, database.Create(&source).Error) - crons, err := newServiceWithFetcher(database, failingFeedFetcher{}).SyncSources(context.Background(), user.ID) + results, err := newServiceWithFetcher(database, failingFeedFetcher{}).SyncSources(context.Background(), user.ID) require.NoError(t, err) - require.Len(t, crons, 1) - require.Equal(t, "failed", crons[0].Status) - require.False(t, crons[0].Enabled) - require.Contains(t, crons[0].LastResult, "feed unavailable") + 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) { @@ -157,10 +155,10 @@ func TestSyncAllSourcesCollectsEnabledRSSSourcesForEveryUser(t *testing.T) { require.NoError(t, database.Create(&source).Error) } - crons, err := newServiceWithFetcher(database, staticFeedFetcher{}).SyncAllSources(context.Background()) + results, err := newServiceWithFetcher(database, staticFeedFetcher{}).SyncAllSources(context.Background()) require.NoError(t, err) - require.Len(t, crons, 2) + require.Len(t, results, 2) var itemCount int64 require.NoError(t, database.Model(&models.SaDatasetItem{}).Count(&itemCount).Error) require.Equal(t, int64(2), itemCount) @@ -185,7 +183,7 @@ func TestSyncSourcesCollectsMultipleSourcesForOneOwner(t *testing.T) { } } -func TestQueueSourcesRunsInBackgroundAndRejectsOverlap(t *testing.T) { +func TestSyncSourcesRejectsOverlap(t *testing.T) { database := newDatasetTestDatabase(t) user := createDatasetTestUser(t, database, "queued-feed-owner@example.com") source := models.SaDatasetSource{ @@ -199,20 +197,24 @@ func TestQueueSourcesRunsInBackgroundAndRejectsOverlap(t *testing.T) { } service := newServiceWithFetcher(database, fetcher) - queued, err := service.QueueSources(user.ID) - - require.NoError(t, err) - require.Len(t, queued, 1) - require.Equal(t, "pending", queued[0].Status) + 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.QueueSources(user.ID) + _, 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) - require.Eventually(t, func() bool { - runs, listErr := service.ListCrons(user.ID) - return listErr == nil && len(runs) == 1 && runs[0].Status == "completed" - }, time.Second, 10*time.Millisecond) + outcome := <-done + require.NoError(t, outcome.err) + require.Len(t, outcome.results, 1) + require.Equal(t, "completed", outcome.results[0].Status) } type failingFeedFetcher struct{} diff --git a/backend/internal/logic/dataset/handlers.go b/backend/internal/logic/dataset/handlers.go index 2d08df9..b8df89e 100644 --- a/backend/internal/logic/dataset/handlers.go +++ b/backend/internal/logic/dataset/handlers.go @@ -54,14 +54,10 @@ type ItemPageDTO struct { Offset int `json:"offset"` } -type CronDTO struct { - ID string `json:"id"` - SourceID string `json:"sourceId"` - Status string `json:"status"` - LastRunAt *time.Time `json:"lastRunAt"` - LastResult string `json:"lastResult"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` +type SyncResultDTO struct { + SourceID string `json:"sourceId"` + Status string `json:"status"` + Result string `json:"result"` } type DepositDTO struct { @@ -91,8 +87,7 @@ func (h *Handler) Register(router gin.IRouter) { router.POST("/dataset-items", h.createItem) router.PATCH("/dataset-items/:id", h.updateItem) router.POST("/dataset-items/:id/deposit", h.depositItem) - router.GET("/dataset-crons", h.listCrons) - router.POST("/dataset-crons/sync", h.queueSync) + router.POST("/dataset-sources/sync", h.syncSources) } func (h *Handler) listSources(c *gin.Context) { @@ -309,38 +304,25 @@ func (h *Handler) depositItem(c *gin.Context) { }) } -func (h *Handler) listCrons(c *gin.Context) { +func (h *Handler) syncSources(c *gin.Context) { userID, ok := currentUser(c) if !ok { return } - crons, err := h.service.ListCrons(userID) + results, err := h.service.SyncSources(c.Request.Context(), userID) if err != nil { writeError(c, err) return } - result := make([]CronDTO, 0, len(crons)) - for _, cron := range crons { - result = append(result, cronDTO(cron)) + response := make([]SyncResultDTO, 0, len(results)) + for _, result := range results { + response = append(response, SyncResultDTO{ + SourceID: result.SourceIdentity, + Status: result.Status, + Result: result.Result, + }) } - c.JSON(http.StatusOK, result) -} - -func (h *Handler) queueSync(c *gin.Context) { - userID, ok := currentUser(c) - if !ok { - return - } - crons, err := h.service.QueueSources(userID) - if err != nil { - writeError(c, err) - return - } - result := make([]CronDTO, 0, len(crons)) - for _, cron := range crons { - result = append(result, cronDTO(cron)) - } - c.JSON(http.StatusAccepted, result) + c.JSON(http.StatusOK, response) } func currentUser(c *gin.Context) (uint, bool) { @@ -390,14 +372,6 @@ func itemDTO(item models.SaDatasetItem) ItemDTO { } } -func cronDTO(cron models.SaDatasetCron) CronDTO { - return CronDTO{ - ID: cron.Identity, SourceID: cron.SourceIdentity, Status: cron.Status, - LastRunAt: utcTime(cron.LastRunAt), LastResult: cron.LastResult, - CreatedAt: cron.CreatedAt.UTC(), UpdatedAt: cron.UpdatedAt.UTC(), - } -} - func optionalNonNegativeInt(value string) (int, error) { if strings.TrimSpace(value) == "" { return 0, nil diff --git a/backend/internal/logic/dataset/handlers_test.go b/backend/internal/logic/dataset/handlers_test.go index 22dfc6e..745f6d3 100644 --- a/backend/internal/logic/dataset/handlers_test.go +++ b/backend/internal/logic/dataset/handlers_test.go @@ -8,7 +8,6 @@ import ( "net/http" "net/http/httptest" "testing" - "time" "github.com/glebarez/sqlite" "github.com/stretchr/testify/require" @@ -95,25 +94,20 @@ func TestDatasetAPIConnectsSourcesItemsAndSyncTasks(t *testing.T) { require.Equal(t, item.Title, depositedNote.Title) require.Contains(t, depositedNote.Markdown, "原文链接") - firstSync := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-crons/sync", map[string]any{}) - require.Equal(t, http.StatusAccepted, firstSync.Code) - var firstCrons []CronDTO - require.NoError(t, json.Unmarshal(firstSync.Body.Bytes(), &firstCrons)) - require.Len(t, firstCrons, 1) - require.Equal(t, "pending", firstCrons[0].Status) - firstCompleted := waitForDatasetRun(t, ownerRouter, firstCrons[0].ID) - require.Equal(t, "completed", firstCompleted.Status) - require.Contains(t, firstCompleted.LastResult, "format=rss") + firstSync := performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-sources/sync", map[string]any{}) + require.Equal(t, http.StatusOK, firstSync.Code) + var firstResults []SyncResultDTO + require.NoError(t, json.Unmarshal(firstSync.Body.Bytes(), &firstResults)) + require.Len(t, firstResults, 1) + require.Equal(t, "completed", firstResults[0].Status) + require.Contains(t, firstResults[0].Result, "format=rss") - var secondSync *httptest.ResponseRecorder - require.Eventually(t, func() bool { - secondSync = performDatasetRequest(t, ownerRouter, http.MethodPost, "/api/v1/dataset-crons/sync", map[string]any{}) - return secondSync.Code == http.StatusAccepted - }, time.Second, 10*time.Millisecond) - var secondCrons []CronDTO - require.NoError(t, json.Unmarshal(secondSync.Body.Bytes(), &secondCrons)) - require.NotEqual(t, firstCrons[0].ID, secondCrons[0].ID) - waitForDatasetRun(t, ownerRouter, secondCrons[0].ID) + 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) @@ -129,11 +123,9 @@ 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) { @@ -241,7 +233,6 @@ func newDatasetTestDatabase(t *testing.T) *gorm.DB { &models.SaUser{}, &models.SaDatasetSource{}, &models.SaDatasetItem{}, - &models.SaDatasetCron{}, &models.SaProject{}, &models.SaNote{}, )) @@ -252,29 +243,6 @@ func newDatasetTestDatabase(t *testing.T) *gorm.DB { return database } -func waitForDatasetRun(t *testing.T, router http.Handler, identity string) CronDTO { - t.Helper() - var completed CronDTO - require.Eventually(t, func() bool { - response := performDatasetRequest(t, router, http.MethodGet, "/api/v1/dataset-crons", nil) - if response.Code != http.StatusOK { - return false - } - var runs []CronDTO - if json.Unmarshal(response.Body.Bytes(), &runs) != nil { - return false - } - for _, run := range runs { - if run.ID == identity && (run.Status == "completed" || run.Status == "failed") { - completed = run - return true - } - } - return false - }, time.Second, 10*time.Millisecond) - return completed -} - func createDatasetTestUser(t *testing.T, database *gorm.DB, email string) models.SaUser { t.Helper() user := models.SaUser{Email: email, DisplayName: email, PasswordHash: "hash", Role: "user"} diff --git a/backend/internal/logic/dataset/service.go b/backend/internal/logic/dataset/service.go index 647ad2b..9a82174 100644 --- a/backend/internal/logic/dataset/service.go +++ b/backend/internal/logic/dataset/service.go @@ -27,7 +27,6 @@ var ( const ( defaultItemPageSize = 50 maxItemPageSize = 200 - datasetRunRetention = 30 * 24 * time.Hour maxConcurrentFetches = 4 ) @@ -91,9 +90,10 @@ type ItemPage struct { Offset int } -type queuedSource struct { - Source models.SaDatasetSource - Run models.SaDatasetCron +type SyncResult struct { + SourceIdentity string + Status string + Result string } func NewService(database *gorm.DB) *Service { @@ -228,9 +228,6 @@ func (s *Service) DeleteSource(userID uint, identity string) error { if source.SeedKey != nil { return ErrBuiltInSource } - if err := tx.Where("source_id = ?", source.ID).Delete(&models.SaDatasetCron{}).Error; err != nil { - return err - } if err := tx.Where("source_id = ?", source.ID).Delete(&models.SaDatasetItem{}).Error; err != nil { return err } @@ -363,93 +360,32 @@ func (s *Service) DepositItem(userID uint, itemIdentity, projectIdentity string) return DepositResult{NoteIdentity: note.Identity, ProjectIdentity: project.Identity}, nil } -func (s *Service) ListCrons(userID uint) ([]models.SaDatasetCron, error) { - var crons []models.SaDatasetCron - if err := s.db.Model(&models.SaDatasetCron{}). - Select("sa_dataset_crons.*"). - Joins("JOIN sa_dataset_sources ON sa_dataset_sources.id = sa_dataset_crons.source_id"). - Where("sa_dataset_sources.owner_id = ?", userID). - Order("sa_dataset_crons.created_at desc, sa_dataset_crons.id desc"). - Limit(100).Find(&crons).Error; err != nil { - return nil, err - } - if crons == nil { - crons = []models.SaDatasetCron{} - } - return crons, nil -} - -func (s *Service) SyncSources(ctx context.Context, userID uint) ([]models.SaDatasetCron, error) { - s.syncMu.Lock() - defer s.syncMu.Unlock() - - return s.syncSources(ctx, userID, "scheduled") -} - -func (s *Service) QueueSources(userID uint) ([]models.SaDatasetCron, error) { +func (s *Service) SyncSources(ctx context.Context, userID uint) ([]SyncResult, error) { if !s.syncMu.TryLock() { return nil, ErrSyncInProgress } - sources, err := s.enabledRSSSources(userID) - if err != nil { - s.syncMu.Unlock() - return nil, err - } - queued, err := s.createDatasetRuns(userID, sources, "manual") - if err != nil { - s.syncMu.Unlock() - return nil, err - } - runs := make([]models.SaDatasetCron, 0, len(queued)) - for _, entry := range queued { - runs = append(runs, entry.Run) - } - if len(queued) == 0 { - s.syncMu.Unlock() - return runs, nil - } - go func() { - defer s.syncMu.Unlock() - _, _ = s.executeDatasetRuns(context.Background(), queued) - }() - return runs, nil -} + defer s.syncMu.Unlock() -func (s *Service) SyncAllSources(ctx context.Context) ([]models.SaDatasetCron, error) { - var userIDs []uint - if err := s.db.Model(&models.SaDatasetSource{}). - Distinct("owner_id"). - Where("enabled = ? AND kind = ?", true, "rss"). - Order("owner_id asc"). - Pluck("owner_id", &userIDs).Error; err != nil { - return nil, err - } - - result := make([]models.SaDatasetCron, 0) - var syncErrors []error - for _, userID := range userIDs { - crons, err := s.SyncSources(ctx, userID) - result = append(result, crons...) - if err != nil { - syncErrors = append(syncErrors, fmt.Errorf("sync dataset sources for user %d: %w", userID, err)) - } - if ctx.Err() != nil { - break - } - } - return result, errors.Join(syncErrors...) -} - -func (s *Service) syncSources(ctx context.Context, userID uint, trigger string) ([]models.SaDatasetCron, error) { sources, err := s.enabledRSSSources(userID) if err != nil { return nil, err } - queued, err := s.createDatasetRuns(userID, sources, trigger) - if err != nil { + 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 := s.db.Where("enabled = ? AND kind = ?", true, "rss"). + Order("owner_id asc, id asc"). + Find(&sources).Error; err != nil { return nil, err } - return s.executeDatasetRuns(ctx, queued) + return s.executeDatasetSources(ctx, sources) } func (s *Service) enabledRSSSources(userID uint) ([]models.SaDatasetSource, error) { @@ -461,78 +397,41 @@ func (s *Service) enabledRSSSources(userID uint) ([]models.SaDatasetSource, erro return sources, nil } -func (s *Service) createDatasetRuns(userID uint, sources []models.SaDatasetSource, trigger string) ([]queuedSource, error) { - queued := make([]queuedSource, 0, len(sources)) - now := time.Now().UTC() - err := s.db.Transaction(func(tx *gorm.DB) error { - if err := tx.Where("created_at < ?", now.Add(-datasetRunRetention)). - Delete(&models.SaDatasetCron{}).Error; err != nil { - return err - } - for _, source := range sources { - run := models.SaDatasetCron{ - SourceID: source.ID, CreatedBy: userID, Schedule: trigger, - Status: "pending", Enabled: true, NextRunAt: &now, - } - if err := tx.Create(&run).Error; err != nil { - return err - } - queued = append(queued, queuedSource{Source: source, Run: run}) - } - return nil - }) - return queued, err -} - -func (s *Service) executeDatasetRuns(ctx context.Context, queued []queuedSource) ([]models.SaDatasetCron, error) { - result := make([]models.SaDatasetCron, len(queued)) - runErrors := make([]error, len(queued)) +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 } semaphore := make(chan struct{}, concurrency) var workers sync.WaitGroup - for index, entry := range queued { + for index, source := range sources { workers.Add(1) go func() { defer workers.Done() semaphore <- struct{}{} defer func() { <-semaphore }() - result[index], runErrors[index] = s.executeDatasetRun(ctx, entry) + result[index], runErrors[index] = s.executeDatasetSource(ctx, source) }() } workers.Wait() return result, errors.Join(runErrors...) } -func (s *Service) executeDatasetRun(ctx context.Context, entry queuedSource) (models.SaDatasetCron, error) { - source := entry.Source - cron := entry.Run +func (s *Service) executeDatasetSource(ctx context.Context, source models.SaDatasetSource) (SyncResult, error) { + result := SyncResult{SourceIdentity: source.Identity} if err := ctx.Err(); err != nil { - return cron, err - } - now := time.Now().UTC() - cron.Status = "running" - cron.LastRunAt = &now - if err := s.db.Model(&cron).Updates(map[string]any{ - "status": "running", "last_run_at": now, - }).Error; err != nil { - return cron, err + result.Status = "failed" + result.Result = truncateResult(err.Error()) + return result, err } feed, fetchErr := s.fetcher.Fetch(ctx, source.URL) if fetchErr != nil { - cron.Status = "failed" - cron.Enabled = false - cron.NextRunAt = nil - cron.LastResult = truncateResult(fetchErr.Error()) - if err := s.db.Model(&cron).Updates(map[string]any{ - "status": cron.Status, "last_result": cron.LastResult, "enabled": false, "next_run_at": nil, - }).Error; err != nil { - return cron, err - } - return cron, ctx.Err() + result.Status = "failed" + result.Result = truncateResult(fetchErr.Error()) + return result, nil } inserted := 0 @@ -543,30 +442,16 @@ func (s *Service) executeDatasetRun(ctx context.Context, entry queuedSource) (mo if storeErr != nil { return storeErr } - cron.Status = "completed" - cron.Enabled = false - cron.NextRunAt = nil - cron.LastResult = fmt.Sprintf("format=%s fetched=%d inserted=%d", feed.Format, len(feed.Items), inserted) - if err := tx.Model(&cron).Updates(map[string]any{ - "status": cron.Status, "last_result": cron.LastResult, "enabled": false, "next_run_at": nil, - }).Error; err != nil { - return err - } return tx.Model(&source).Update("last_synced_at", completedAt).Error }) - if err == nil { - return cron, nil + if err != nil { + result.Status = "failed" + result.Result = truncateResult("store feed items: " + err.Error()) + return result, nil } - cron.Status = "failed" - cron.Enabled = false - cron.NextRunAt = nil - cron.LastResult = truncateResult("store feed items: " + err.Error()) - if updateErr := s.db.Model(&cron).Updates(map[string]any{ - "status": cron.Status, "last_result": cron.LastResult, "enabled": false, "next_run_at": nil, - }).Error; updateErr != nil { - return cron, updateErr - } - return cron, 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) { diff --git a/backend/internal/models/dataset_cron.go b/backend/internal/models/dataset_cron.go deleted file mode 100644 index 02c7264..0000000 --- a/backend/internal/models/dataset_cron.go +++ /dev/null @@ -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 `gorm:"index"` - UpdatedAt time.Time -} - -func (SaDatasetCron) TableName() string { - return "sa_dataset_crons" -} diff --git a/backend/internal/models/identity.go b/backend/internal/models/identity.go index 25a5ebf..efa291a 100644 --- a/backend/internal/models/identity.go +++ b/backend/internal/models/identity.go @@ -36,16 +36,6 @@ func (m *SaDatasetItem) BeforeCreate(tx *gorm.DB) error { return resolveIdentity(tx, &SaDatasetSource{}, m.SourceID, &m.SourceIdentity) } -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 { if err := ensureIdentity(&m.Identity); err != nil { return err diff --git a/backend/internal/models/new.go b/backend/internal/models/new.go index a12d6df..0042f5a 100644 --- a/backend/internal/models/new.go +++ b/backend/internal/models/new.go @@ -33,7 +33,6 @@ func AutoMigrate(database *gorm.DB) error { &SaProject{}, &SaDatasetSource{}, &SaDatasetItem{}, - &SaDatasetCron{}, &SaInboxItem{}, &SaInboxSuggestion{}, &SaTask{},