fix(datasource): keep manually deleted items deleted across syncs

internal/application/repository/knowledge.go

@@ -776,9 +776,36 @@ func (r *knowledgeRepository) FindByDataSourceExternalID(
 	return &knowledge, nil
 }
 
+// FindTombstonedByDataSourceExternalID returns the soft-deleted row for a
+// (data source, external item ID) pair. It is a persistent tombstone: sync
+// never resurrects an item the user deleted.
+// A file re-created at the source under the same external_id stays suppressed too.
+func (r *knowledgeRepository) FindTombstonedByDataSourceExternalID(
+	ctx context.Context,
+	tenantID uint64,
+	kbID, dataSourceID, externalID string,
+) (*types.Knowledge, error) {
+	var knowledge types.Knowledge
+	err := r.db.Unscoped().WithContext(ctx).
+		Where("tenant_id = ? AND knowledge_base_id = ?", tenantID, kbID).
+		Where("deleted_at IS NOT NULL").
+		Where("metadata->>'datasource_id' = ? AND metadata->>'external_id' = ?", dataSourceID, externalID).
+		Order("deleted_at DESC").
+		First(&knowledge).Error
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &knowledge, nil
+}
+
 // HardDeleteKnowledge physically removes a knowledge row. Call it AFTER
-// DeleteKnowledge's soft-delete cascade so sync-internal deletions never
-// become tombstones that block a later re-sync of the same external item.
+// DeleteKnowledge's soft-delete cascade (chunks/embeddings/graph/wiki/files
+// already cleaned) for sync-internal deletions (update replace, subtree
+// sweep), so those rows never become tombstones. Only user-visible deletions
+// may suppress future syncs.
 func (r *knowledgeRepository) HardDeleteKnowledge(ctx context.Context, tenantID uint64, id string) error {
 	return r.db.Unscoped().WithContext(ctx).
 		Where("tenant_id = ? AND id = ?", tenantID, id).

internal/application/repository/knowledge_tombstone_test.go

@@ -0,0 +1,173 @@
+package repository
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestFindTombstonedByDataSourceExternalID verifies the tombstone lookup used
+// by sync to keep manually deleted items deleted. Only rows that are (a)
+// soft-deleted and (b) owned by the exact (tenant, knowledge base, data
+// source, external_id) match. A tombstone is persistent (no retention window)
+// and the most recent deletion wins.
+func TestFindTombstonedByDataSourceExternalID(t *testing.T) {
+	db := setupKnowledgeTestDB(t)
+	repo := NewKnowledgeRepository(db).(*knowledgeRepository)
+	ctx := context.Background()
+
+	const tenantID uint64 = 52
+	kbID := uuid.New().String()
+	dsID := uuid.New().String()
+
+	insertRow := func(tid uint64, kb, ds, extID, deletedAt string) string {
+		id := uuid.New().String()
+		metadata := fmt.Sprintf(`{"datasource_id":%q,"external_id":%q}`, ds, extID)
+		require.NoError(t, db.Exec(`
+			INSERT INTO knowledges
+			  (id, tenant_id, knowledge_base_id, type, title, source, parse_status, metadata, deleted_at)
+			VALUES (?, ?, ?, 'document', ?, 'feishu', 'completed', ?, ?)
+		`, id, tid, kb, extID, metadata, deletedAt).Error)
+		return id
+	}
+
+	now := time.Now().UTC()
+	// Tombstone deleted recently, must match.
+	recentID := insertRow(tenantID, kbID, dsID, "file:gone", now.Add(-24*time.Hour).Format("2006-01-02 15:04:05"))
+	// Tombstone deleted long ago, still matches (persistent).
+	_ = insertRow(tenantID, kbID, dsID, "file:old", now.Add(-400*24*time.Hour).Format("2006-01-02 15:04:05"))
+	// Same external_id under a different data source, must NOT match.
+	_ = insertRow(tenantID, kbID, uuid.New().String(), "file:gone",
+		now.Add(-24*time.Hour).Format("2006-01-02 15:04:05"))
+	// Same external_id in a different KB, must NOT match.
+	_ = insertRow(tenantID, uuid.New().String(), dsID, "file:gone",
+		now.Add(-24*time.Hour).Format("2006-01-02 15:04:05"))
+	// Live (non-deleted) row, must NOT match.
+	liveID := uuid.New().String()
+	require.NoError(t, db.Exec(`
+		INSERT INTO knowledges
+		  (id, tenant_id, knowledge_base_id, type, title, source, parse_status, metadata)
+		VALUES (?, ?, ?, 'document', ?, 'feishu', 'completed', ?)
+	`, liveID, tenantID, kbID, "file:live",
+		fmt.Sprintf(`{"datasource_id":%q,"external_id":%q}`, dsID, "file:live")).Error)
+
+	tomb, err := repo.FindTombstonedByDataSourceExternalID(ctx, tenantID, kbID, dsID, "file:gone")
+	require.NoError(t, err)
+	require.NotNil(t, tomb)
+	assert.Equal(t, recentID, tomb.ID)
+
+	tomb, err = repo.FindTombstonedByDataSourceExternalID(ctx, tenantID, kbID, dsID, "file:old")
+	require.NoError(t, err)
+	require.NotNil(t, tomb, "a tombstone is persistent, an old deletion still suppresses re-sync")
+
+	tomb, err = repo.FindTombstonedByDataSourceExternalID(ctx, tenantID, kbID, dsID, "file:live")
+	require.NoError(t, err)
+	assert.Nil(t, tomb, "live rows are not tombstones")
+
+	tomb, err = repo.FindTombstonedByDataSourceExternalID(ctx, tenantID, kbID, dsID, "file:never-existed")
+	require.NoError(t, err)
+	assert.Nil(t, tomb)
+}
+
+// TestFindTombstonedByDataSourceExternalID_MostRecentWins verifies that when a
+// row was deleted repeatedly, the latest deletion is the one honored.
+func TestFindTombstonedByDataSourceExternalID_MostRecentWins(t *testing.T) {
+	db := setupKnowledgeTestDB(t)
+	repo := NewKnowledgeRepository(db).(*knowledgeRepository)
+	ctx := context.Background()
+
+	const tenantID uint64 = 53
+	kbID := uuid.New().String()
+	dsID := uuid.New().String()
+	metadata := fmt.Sprintf(`{"datasource_id":%q,"external_id":%q}`, dsID, "file:repeated")
+
+	older := uuid.New().String()
+	require.NoError(t, db.Exec(`
+		INSERT INTO knowledges
+		  (id, tenant_id, knowledge_base_id, type, title, source, parse_status, metadata, deleted_at)
+		VALUES (?, ?, ?, 'document', ?, 'feishu', 'completed', ?, ?)
+	`, older, tenantID, kbID, "file:repeated", metadata,
+		time.Now().UTC().Add(-10*24*time.Hour).Format("2006-01-02 15:04:05")).Error)
+
+	newer := uuid.New().String()
+	require.NoError(t, db.Exec(`
+		INSERT INTO knowledges
+		  (id, tenant_id, knowledge_base_id, type, title, source, parse_status, metadata, deleted_at)
+		VALUES (?, ?, ?, 'document', ?, 'feishu', 'completed', ?, ?)
+	`, newer, tenantID, kbID, "file:repeated", metadata,
+		time.Now().UTC().Add(-2*time.Hour).Format("2006-01-02 15:04:05")).Error)
+
+	tomb, err := repo.FindTombstonedByDataSourceExternalID(ctx, tenantID, kbID, dsID, "file:repeated")
+	require.NoError(t, err)
+	require.NotNil(t, tomb)
+	assert.Equal(t, newer, tomb.ID, "the most recent deletion must win")
+}
+
+// TestHardDeleteKnowledgeRemovesRowAndTombstone verifies that hard deletion
+// physically removes the row: neither the live lookup nor the tombstone
+// lookup sees it afterwards.
+func TestHardDeleteKnowledgeRemovesRowAndTombstone(t *testing.T) {
+	db := setupKnowledgeTestDB(t)
+	repo := NewKnowledgeRepository(db).(*knowledgeRepository)
+	ctx := context.Background()
+
+	const tenantID uint64 = 54
+	kbID := uuid.New().String()
+	dsID := uuid.New().String()
+	extID := "file:hard-deleted"
+
+	id := uuid.New().String()
+	metadata := fmt.Sprintf(`{"datasource_id":%q,"external_id":%q}`, dsID, extID)
+	require.NoError(t, db.Exec(`
+		INSERT INTO knowledges
+		  (id, tenant_id, knowledge_base_id, type, title, source, parse_status, metadata, deleted_at)
+		VALUES (?, ?, ?, 'document', ?, 'feishu', 'completed', ?, ?)
+	`, id, tenantID, kbID, extID, metadata,
+		time.Now().UTC().Add(-24*time.Hour).Format("2006-01-02 15:04:05")).Error)
+
+	tomb, err := repo.FindTombstonedByDataSourceExternalID(ctx, tenantID, kbID, dsID, extID)
+	require.NoError(t, err)
+	require.NotNil(t, tomb)
+
+	require.NoError(t, repo.HardDeleteKnowledge(ctx, tenantID, id))
+
+	tomb, err = repo.FindTombstonedByDataSourceExternalID(ctx, tenantID, kbID, dsID, extID)
+	require.NoError(t, err)
+	assert.Nil(t, tomb, "a hard-deleted row must not remain a tombstone")
+}
+
+// TestHardDeleteKnowledgeList verifies the batch hard delete on the subtree
+// sweep path.
+func TestHardDeleteKnowledgeList(t *testing.T) {
+	db := setupKnowledgeTestDB(t)
+	repo := NewKnowledgeRepository(db).(*knowledgeRepository)
+	ctx := context.Background()
+
+	const tenantID uint64 = 55
+	kbID := uuid.New().String()
+	dsID := uuid.New().String()
+
+	ids := make([]string, 0, 2)
+	for _, ext := range []string{"doc:p#file:a", "doc:p#file:b"} {
+		id := uuid.New().String()
+		ids = append(ids, id)
+		require.NoError(t, db.Exec(`
+			INSERT INTO knowledges
+			  (id, tenant_id, knowledge_base_id, type, title, source, parse_status, metadata, deleted_at)
+			VALUES (?, ?, ?, 'document', ?, 'feishu', 'completed', ?, ?)
+		`, id, tenantID, kbID, ext, fmt.Sprintf(`{"datasource_id":%q,"external_id":%q}`, dsID, ext),
+			time.Now().UTC().Add(-time.Hour).Format("2006-01-02 15:04:05")).Error)
+	}
+
+	require.NoError(t, repo.HardDeleteKnowledgeList(ctx, tenantID, ids))
+	for _, ext := range []string{"doc:p#file:a", "doc:p#file:b"} {
+		tomb, err := repo.FindTombstonedByDataSourceExternalID(ctx, tenantID, kbID, dsID, ext)
+		require.NoError(t, err)
+		assert.Nil(t, tomb, "batch hard delete must clear tombstones for every id")
+	}
+}

internal/application/service/datasource_service.go

@@ -818,6 +818,37 @@ func (s *DataSourceService) resolveAutoTagIDs(ctx context.Context, ds *types.Dat
 // display (Tencent/WeKnora#2136 / #1262).
 const maxSyncResultErrors = 100
 
+// errSyncItemTombstoned skips an item the user deleted from the KB. Only user
+// deletions create tombstones. Sync-internal deletions never block re-sync.
+var errSyncItemTombstoned = errors.New("sync item matches a deleted row, skipping")
+
+const syncErrorCodeUserDeletedExcluded = "user_deleted_excluded"
+
+func syncItemUserDeletedExcludedError(item *types.FetchedItem) types.SyncItemError {
+	return types.SyncItemError{
+		Title:   item.Title,
+		Code:    syncErrorCodeUserDeletedExcluded,
+		Message: "Item was deleted from the knowledge base and will not be re-synced",
+	}
+}
+
+// checkSyncItemUserDeletionExclusion reports whether sync must skip an item
+// because the user deleted it (soft-delete tombstone). Lookup errors fail closed.
+func checkSyncItemUserDeletionExclusion(
+	ctx context.Context,
+	repo interfaces.KnowledgeRepository,
+	ds *types.DataSource,
+	item *types.FetchedItem,
+) (bool, error) {
+	tomb, terr := repo.FindTombstonedByDataSourceExternalID(
+		ctx, ds.TenantID, ds.KnowledgeBaseID, ds.ID, item.ExternalID,
+	)
+	if terr != nil {
+		return false, fmt.Errorf("check tombstone for external_id=%s: %w", item.ExternalID, terr)
+	}
+	return tomb != nil, nil
+}
+
 // recordSyncError appends an error sample to result.Errors, capped at
 // maxSyncResultErrors. Callers still increment result.Failed for the exact count.
 func recordSyncError(result *types.SyncResult, item types.SyncItemError) {
@@ -936,6 +967,11 @@ func (s *DataSourceService) applyFetchedItem(
 	if err != nil {
 		var dupErr *types.DuplicateKnowledgeError
 		switch {
+		case errors.Is(err, errSyncItemTombstoned):
+			// The item matches a tombstone (user deleted it from the KB).
+			// Count as skipped, never re-create it.
+			result.Skipped++
+			recordSyncError(result, syncItemUserDeletedExcludedError(item))
 		case errors.As(err, &dupErr):
 			// Duplicate file/URL is not a failure — count as skipped.
 			logger.Infof(ctx, "item %q (external_id=%s) already exists, skipping", item.Title, item.ExternalID)
@@ -1263,19 +1299,47 @@ func (s *DataSourceService) ingestItem(ctx context.Context, ds *types.DataSource
 		// external IDs from two data sources cannot collide or overwrite each
 		// other during updates.
 		existing, err := repo.FindByDataSourceExternalID(ctx, ds.TenantID, ds.KnowledgeBaseID, ds.ID, item.ExternalID)
-		if err != nil {
-			logger.Warnf(ctx, "failed to check existing knowledge for external_id=%s: %v", item.ExternalID, err)
-			// Non-fatal: proceed with creation (may produce duplicate)
-		} else if existing != nil {
+		switch {
+		case err != nil:
+			excluded, exErr := checkSyncItemUserDeletionExclusion(ctx, repo, ds, item)
+			if exErr != nil {
+				return false, exErr
+			}
+			if excluded {
+				logger.Infof(ctx, "item %q (external_id=%s) was deleted from the KB, skipping resurrection",
+					item.Title, item.ExternalID)
+				return false, errSyncItemTombstoned
+			}
+			// Fail closed: a failed live lookup must not fall through to creation.
+			return false, fmt.Errorf("check existing knowledge for external_id=%s: %w", item.ExternalID, err)
+		case existing != nil:
+			if existing.ParseStatus == types.ParseStatusDeleting {
+				logger.Infof(ctx, "item %q (external_id=%s) is being deleted from the KB, skipping",
+					item.Title, item.ExternalID)
+				return false, errSyncItemTombstoned
+			}
 			logger.Infof(ctx, "found existing knowledge %s for external_id=%s, deleting for update", existing.ID, item.ExternalID)
 			if err := s.knowledgeService.DeleteKnowledge(ctx, existing.ID); err != nil {
 				logger.Warnf(ctx, "failed to delete existing knowledge %s: %v", existing.ID, err)
 			} else {
+				// The update replaces the live row. Hard-delete it so it never
+				// acts as a tombstone, and a failed create below stays recoverable
+				// on the next sync.
 				if herr := repo.HardDeleteKnowledge(ctx, ds.TenantID, existing.ID); herr != nil {
-					logger.Warnf(ctx, "failed to hard-delete replaced knowledge %s: %v", existing.ID, herr)
+					return false, fmt.Errorf("hard-delete replaced knowledge %s: %w", existing.ID, herr)
 				}
 				isUpdate = true
 			}
+		default:
+			excluded, exErr := checkSyncItemUserDeletionExclusion(ctx, repo, ds, item)
+			if exErr != nil {
+				return false, exErr
+			}
+			if excluded {
+				logger.Infof(ctx, "item %q (external_id=%s) was deleted from the KB, skipping resurrection",
+					item.Title, item.ExternalID)
+				return false, errSyncItemTombstoned
+			}
 		}
 	}
 

internal/application/service/datasource_service_test.go

@@ -301,6 +301,12 @@ func (r *deletionLookupKnowledgeRepo) FindByDataSourceExternalID(
 	return r.knowledge, nil
 }
 
+func (r *deletionLookupKnowledgeRepo) FindTombstonedByDataSourceExternalID(
+	context.Context, uint64, string, string, string,
+) (*types.Knowledge, error) {
+	return nil, nil // no tombstone → normal ingest path
+}
+
 func (r *deletionLookupKnowledgeRepo) HardDeleteKnowledge(_ context.Context, _ uint64, id string) error {
 	if r.hardDeleteErr != nil {
 		return r.hardDeleteErr
@@ -361,6 +367,12 @@ func (r *keyedDeletionRepo) FindByDataSourceExternalID(
 	return r.items[externalID], nil
 }
 
+func (r *keyedDeletionRepo) FindTombstonedByDataSourceExternalID(
+	context.Context, uint64, string, string, string,
+) (*types.Knowledge, error) {
+	return nil, nil // no tombstone → normal ingest path
+}
+
 func (r *keyedDeletionRepo) HardDeleteKnowledge(_ context.Context, _ uint64, id string) error {
 	if r.hardDeleteErr != nil {
 		return r.hardDeleteErr

internal/application/service/datasource_sweep_wiring_test.go

@@ -30,6 +30,14 @@ func (r *sweepFakeRepo) FindByDataSourceExternalID(
 	return nil, nil // no existing main item -> skip the case-1 update delete
 }
 
+func (r *sweepFakeRepo) FindTombstonedByDataSourceExternalID(
+	_ context.Context, _ uint64, _ string, _ string, _ string,
+) (*types.Knowledge, error) {
+	return nil, nil // no tombstone → normal ingest path
+}
+
+// HardDeleteKnowledge(HardDeleteKnowledgeList) record the sync-internal
+// physical deletions. The sweep tests assert them via hardDeleted.
 func (r *sweepFakeRepo) HardDeleteKnowledge(context.Context, uint64, string) error {
 	return nil
 }