fix(datasource): invalidate cursor for deleted knowledge

langcaiye · langcaiye · commit 25c78e21e865 · 2026-06-10T16:48:12.000+08:00
diff --git a/internal/application/repository/datasource_repo.go b/internal/application/repository/datasource_repo.go
@@ -2,6 +2,7 @@ package repository
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"time"
 
@@ -107,6 +108,147 @@ func (r *DataSourceRepository) UpdateSyncState(ctx context.Context, ds *types.Da
 	return nil
 }
 
+// InvalidateCursorItem removes a single item from a data source's incremental
+// cursor. This keeps the cursor consistent when a locally synced knowledge item
+// is deleted, allowing the next incremental sync to re-fetch that upstream
+// item even if its edit timestamp did not change.
+func (r *DataSourceRepository) InvalidateCursorItem(
+	ctx context.Context,
+	tenantID uint64,
+	dataSourceID string,
+	externalID string,
+	sourceResourceID string,
+) error {
+	if dataSourceID == "" {
+		return errors.New("data source id is empty")
+	}
+	if externalID == "" {
+		return nil
+	}
+
+	var ds types.DataSource
+	if err := r.db.WithContext(ctx).
+		Where("id = ?", dataSourceID).
+		Where("tenant_id = ?", tenantID).
+		Where("deleted_at IS NULL").
+		First(&ds).Error; err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil
+		}
+		return err
+	}
+	if len(ds.LastSyncCursor) == 0 {
+		return nil
+	}
+
+	var cursor map[string]interface{}
+	if err := json.Unmarshal(ds.LastSyncCursor, &cursor); err != nil {
+		return err
+	}
+	if !invalidateCursorItem(cursor, externalID, sourceResourceID) {
+		return nil
+	}
+	cursorJSON, err := json.Marshal(cursor)
+	if err != nil {
+		return err
+	}
+
+	return r.db.WithContext(ctx).
+		Model(&types.DataSource{}).
+		Where("id = ?", dataSourceID).
+		Where("tenant_id = ?", tenantID).
+		Update("last_sync_cursor", types.JSON(cursorJSON)).Error
+}
+
+func invalidateCursorItem(cursor map[string]interface{}, externalID string, sourceResourceID string) bool {
+	if cursor == nil || externalID == "" {
+		return false
+	}
+
+	connectorCursor, hasConnectorCursor := cursor["connector_cursor"].(map[string]interface{})
+	if !hasConnectorCursor {
+		connectorCursor = cursor
+	}
+
+	changed := invalidateNestedCursorItem(connectorCursor, externalID, sourceResourceID)
+	if changed {
+		return true
+	}
+	if hasConnectorCursor {
+		return invalidateNestedCursorItem(cursor, externalID, sourceResourceID)
+	}
+	return false
+}
+
+func invalidateNestedCursorItem(node map[string]interface{}, externalID string, sourceResourceID string) bool {
+	if node == nil {
+		return false
+	}
+
+	if sourceResourceID != "" {
+		return invalidateNestedCursorItemForResource(node, externalID, sourceResourceID)
+	}
+
+	return deleteCursorItemRecursive(node, externalID)
+}
+
+func invalidateNestedCursorItemForResource(node map[string]interface{}, externalID string, sourceResourceID string) bool {
+	if node == nil {
+		return false
+	}
+
+	changed := false
+	for key, value := range node {
+		nested, ok := value.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		if key == sourceResourceID {
+			if deleteCursorItemRecursive(nested, externalID) {
+				changed = true
+				if len(nested) == 0 {
+					delete(node, key)
+				}
+			}
+			continue
+		}
+		if invalidateNestedCursorItemForResource(nested, externalID, sourceResourceID) {
+			changed = true
+			if len(nested) == 0 {
+				delete(node, key)
+			}
+		}
+	}
+	return changed
+}
+
+func deleteCursorItemRecursive(node map[string]interface{}, externalID string) bool {
+	if node == nil {
+		return false
+	}
+
+	changed := false
+	if _, exists := node[externalID]; exists {
+		delete(node, externalID)
+		changed = true
+	}
+
+	for key, value := range node {
+		nested, ok := value.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		if deleteCursorItemRecursive(nested, externalID) {
+			changed = true
+			if len(nested) == 0 {
+				delete(node, key)
+			}
+		}
+	}
+
+	return changed
+}
+
 // Delete performs a soft delete
 func (r *DataSourceRepository) Delete(ctx context.Context, id string) error {
 	if id == "" {
diff --git a/internal/application/repository/datasource_repo_test.go b/internal/application/repository/datasource_repo_test.go
@@ -2,6 +2,7 @@ package repository
 
 import (
 	"context"
+	"encoding/json"
 	"testing"
 	"time"
 
@@ -51,6 +52,79 @@ func TestDataSourceRepositoryUpdateSyncStateClearsErrorMessage(t *testing.T) {
 	require.NotNil(t, stored.LastSyncAt)
 }
 
+func TestDataSourceRepositoryInvalidateCursorItemRemovesFeishuNode(t *testing.T) {
+	db := setupDataSourceRepoTestDB(t)
+	repo := NewDataSourceRepository(db)
+	cursor := types.JSON(`{
+		"last_sync_time":"2026-06-10T00:00:00Z",
+		"connector_cursor":{
+			"space_node_times":{
+				"space-1:root-a":{"node-a":"100","node-b":"200"},
+				"space-1:root-b":{"node-a":"300"}
+			}
+		}
+	}`)
+	ds := &types.DataSource{
+		ID:              "ds-feishu",
+		TenantID:        1,
+		KnowledgeBaseID: "kb-1",
+		Name:            "Feishu",
+		Type:            types.ConnectorTypeFeishu,
+		Status:          types.DataSourceStatusActive,
+		LastSyncCursor:  cursor,
+	}
+	require.NoError(t, repo.Create(context.Background(), ds))
+
+	require.NoError(t, repo.InvalidateCursorItem(context.Background(), 1, ds.ID, "node-a", "space-1:root-a"))
+
+	var stored types.DataSource
+	require.NoError(t, db.First(&stored, "id = ?", ds.ID).Error)
+	var decoded map[string]interface{}
+	require.NoError(t, json.Unmarshal(stored.LastSyncCursor, &decoded))
+	connectorCursor := decoded["connector_cursor"].(map[string]interface{})
+	spaceNodeTimes := connectorCursor["space_node_times"].(map[string]interface{})
+	rootA := spaceNodeTimes["space-1:root-a"].(map[string]interface{})
+	rootB := spaceNodeTimes["space-1:root-b"].(map[string]interface{})
+	assert.NotContains(t, rootA, "node-a")
+	assert.Equal(t, "200", rootA["node-b"])
+	assert.Equal(t, "300", rootB["node-a"])
+}
+
+func TestDataSourceRepositoryInvalidateCursorItemRemovesGenericExternalID(t *testing.T) {
+	db := setupDataSourceRepoTestDB(t)
+	repo := NewDataSourceRepository(db)
+	cursor := types.JSON(`{
+		"last_sync_time":"2026-06-10T00:00:00Z",
+		"connector_cursor":{
+			"page_edit_times":{
+				"page-a":"2026-06-09T00:00:00Z",
+				"page-b":"2026-06-09T01:00:00Z"
+			}
+		}
+	}`)
+	ds := &types.DataSource{
+		ID:              "ds-notion",
+		TenantID:        1,
+		KnowledgeBaseID: "kb-1",
+		Name:            "Notion",
+		Type:            types.ConnectorTypeNotion,
+		Status:          types.DataSourceStatusActive,
+		LastSyncCursor:  cursor,
+	}
+	require.NoError(t, repo.Create(context.Background(), ds))
+
+	require.NoError(t, repo.InvalidateCursorItem(context.Background(), 1, ds.ID, "page-a", ""))
+
+	var stored types.DataSource
+	require.NoError(t, db.First(&stored, "id = ?", ds.ID).Error)
+	var decoded map[string]interface{}
+	require.NoError(t, json.Unmarshal(stored.LastSyncCursor, &decoded))
+	connectorCursor := decoded["connector_cursor"].(map[string]interface{})
+	pageEditTimes := connectorCursor["page_edit_times"].(map[string]interface{})
+	assert.NotContains(t, pageEditTimes, "page-a")
+	assert.Contains(t, pageEditTimes, "page-b")
+}
+
 func TestSyncLogRepositoryUpdateResultClearsErrorMessage(t *testing.T) {
 	db := setupDataSourceRepoTestDB(t)
 	repo := NewSyncLogRepository(db)
diff --git a/internal/application/service/knowledge.go b/internal/application/service/knowledge.go
@@ -43,6 +43,7 @@ type knowledgeService struct {
 	retrieveEngine  interfaces.RetrieveEngineRegistry
 	ownership       retriever.TenantStoreOwnership
 	repo            interfaces.KnowledgeRepository
+	dataSourceRepo  interfaces.DataSourceRepository
 	kbService       interfaces.KnowledgeBaseService
 	tenantRepo      interfaces.TenantRepository
 	tenantService   interfaces.TenantService
@@ -84,6 +85,7 @@ const (
 func NewKnowledgeService(
 	config *config.Config,
 	repo interfaces.KnowledgeRepository,
+	dataSourceRepo interfaces.DataSourceRepository,
 	documentReader interfaces.DocumentReader,
 	kbService interfaces.KnowledgeBaseService,
 	tenantRepo interfaces.TenantRepository,
@@ -110,6 +112,7 @@ func NewKnowledgeService(
 	return &knowledgeService{
 		config:          config,
 		repo:            repo,
+		dataSourceRepo:  dataSourceRepo,
 		kbService:       kbService,
 		tenantRepo:      tenantRepo,
 		tenantService:   tenantService,
diff --git a/internal/application/service/knowledge_delete.go b/internal/application/service/knowledge_delete.go
@@ -55,6 +55,23 @@ func deleteExtractedImages(ctx context.Context, fileSvc interfaces.FileService,
 	}
 }
 
+func (s *knowledgeService) invalidateDataSourceCursorForKnowledge(ctx context.Context, knowledge *types.Knowledge) {
+	if s.dataSourceRepo == nil || knowledge == nil || len(knowledge.Metadata) == 0 {
+		return
+	}
+	metadata := knowledge.GetMetadata()
+	dataSourceID := strings.TrimSpace(metadata["datasource_id"])
+	externalID := strings.TrimSpace(metadata["external_id"])
+	if dataSourceID == "" || externalID == "" {
+		return
+	}
+	sourceResourceID := strings.TrimSpace(metadata["source_resource_id"])
+	tenantID := types.MustTenantIDFromContext(ctx)
+	if err := s.dataSourceRepo.InvalidateCursorItem(ctx, tenantID, dataSourceID, externalID, sourceResourceID); err != nil {
+		logger.Warnf(ctx, "failed to invalidate data source cursor for deleted knowledge %s: %v", knowledge.ID, err)
+	}
+}
+
 // DeleteKnowledge deletes a knowledge entry and all related resources
 func (s *knowledgeService) DeleteKnowledge(ctx context.Context, id string) error {
 	// Get the knowledge entry
@@ -190,7 +207,11 @@ func (s *knowledgeService) DeleteKnowledge(ctx context.Context, id string) error
 		return err
 	}
 	// Delete the knowledge entry itself from the database
-	return s.repo.DeleteKnowledge(ctx, ctx.Value(types.TenantIDContextKey).(uint64), id)
+	if err := s.repo.DeleteKnowledge(ctx, ctx.Value(types.TenantIDContextKey).(uint64), id); err != nil {
+		return err
+	}
+	s.invalidateDataSourceCursorForKnowledge(ctx, knowledge)
+	return nil
 }
 
 // cleanupWikiOnKnowledgeDelete handles wiki pages when a source document is deleted.
@@ -577,7 +598,13 @@ func (s *knowledgeService) DeleteKnowledgeList(ctx context.Context, ids []string
 		return err
 	}
 	// 5. Delete the knowledge entry itself from the database
-	return s.repo.DeleteKnowledgeList(ctx, tenantInfo.ID, ids)
+	if err := s.repo.DeleteKnowledgeList(ctx, tenantInfo.ID, ids); err != nil {
+		return err
+	}
+	for _, knowledge := range knowledgeList {
+		s.invalidateDataSourceCursorForKnowledge(ctx, knowledge)
+	}
+	return nil
 }
 
 func (s *knowledgeService) cleanupKnowledgeResources(ctx context.Context, knowledge *types.Knowledge) error {
diff --git a/internal/types/interfaces/datasource.go b/internal/types/interfaces/datasource.go
@@ -82,6 +82,11 @@ type DataSourceRepository interface {
 	// UpdateSyncState updates only fields produced by a sync run.
 	UpdateSyncState(ctx context.Context, ds *types.DataSource) error
 
+	// InvalidateCursorItem removes a synced item from the stored incremental
+	// cursor so deleting local knowledge lets the next sync fetch it again even
+	// if the upstream document itself has not changed.
+	InvalidateCursorItem(ctx context.Context, tenantID uint64, dataSourceID string, externalID string, sourceResourceID string) error
+
 	// Delete performs a soft delete
 	Delete(ctx context.Context, id string) error
 
