diff --git a/internal/application/repository/knowledge.go b/internal/application/repository/knowledge.go index 687ff4713b..1dd348ab53 100644 --- a/internal/application/repository/knowledge.go +++ b/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). diff --git a/internal/application/repository/knowledge_tombstone_test.go b/internal/application/repository/knowledge_tombstone_test.go new file mode 100644 index 0000000000..754a6b0ebb --- /dev/null +++ b/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") + } +} diff --git a/internal/application/service/datasource_service.go b/internal/application/service/datasource_service.go index 71eff04ade..be9ff58bc3 100644 --- a/internal/application/service/datasource_service.go +++ b/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 + } } } diff --git a/internal/application/service/datasource_service_test.go b/internal/application/service/datasource_service_test.go index a781f4c6ac..2e7ff9d920 100644 --- a/internal/application/service/datasource_service_test.go +++ b/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 diff --git a/internal/application/service/datasource_sweep_wiring_test.go b/internal/application/service/datasource_sweep_wiring_test.go index e9ce5bc0a8..b6b0ff4069 100644 --- a/internal/application/service/datasource_sweep_wiring_test.go +++ b/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 } diff --git a/internal/application/service/datasource_tombstone_test.go b/internal/application/service/datasource_tombstone_test.go new file mode 100644 index 0000000000..11f1d07f53 --- /dev/null +++ b/internal/application/service/datasource_tombstone_test.go @@ -0,0 +1,596 @@ +package service + +import ( + "context" + "encoding/json" + "mime/multipart" + "testing" + "time" + + "github.com/Tencent/WeKnora/internal/datasource" + "github.com/Tencent/WeKnora/internal/types" + "github.com/Tencent/WeKnora/internal/types/interfaces" + "github.com/hibiken/asynq" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/gorm" +) + +// tombstoneTenantRepo / tombstoneTagService satisfy ProcessSync's tenant and +// auto-tag lookups with minimal behavior. +type tombstoneTenantRepo struct { + interfaces.TenantRepository + tenant *types.Tenant +} + +func (r *tombstoneTenantRepo) GetTenantByID(context.Context, uint64) (*types.Tenant, error) { + return r.tenant, nil +} + +type tombstoneTagService struct { + interfaces.KnowledgeTagService +} + +func (*tombstoneTagService) FindOrCreateTagByName(context.Context, string, string) (*types.KnowledgeTag, error) { + return nil, nil +} + +// tombstoneRepo models the knowledge table's live/tombstone state with the +// same soft-delete mechanics the real repository uses: +// - live: rows visible to FindByMetadataKey / FindByMetadataKeyPrefix +// - tombstones: soft-deleted rows (DeleteKnowledge moved them here), +// reported by FindTombstonedByDataSourceExternalID +// - hardDeleted: physically removed rows (no tombstone effect) +type tombstoneRepo struct { + interfaces.KnowledgeRepository + liveByExternal map[string]*types.Knowledge + liveByID map[string]string // id → external_id + tombstones map[string]*types.Knowledge + children []*types.Knowledge // returned by FindByMetadataKeyPrefix + + tombstoneErr error // if set, the tombstone lookup fails + liveLookupErr error // if set, FindByDataSourceExternalID fails + hardDeleteErr error // if set, HardDeleteKnowledge fails + + lookupTenantID uint64 + lookupKBID string + lookupDataSourceID string + lookupExternalID string + + hardDeleted []string + hardDeletedBats [][]string +} + +func newTombstoneRepo() *tombstoneRepo { + return &tombstoneRepo{ + liveByExternal: map[string]*types.Knowledge{}, + liveByID: map[string]string{}, + tombstones: map[string]*types.Knowledge{}, + } +} + +func (r *tombstoneRepo) addLive(externalID string, k *types.Knowledge) { + r.liveByExternal[externalID] = k + r.liveByID[k.ID] = externalID +} + +func (r *tombstoneRepo) FindByMetadataKey( + _ context.Context, _ uint64, _ string, key, value string, +) (*types.Knowledge, error) { + if key == "external_id" { + return r.liveByExternal[value], nil + } + return nil, nil +} + +func (r *tombstoneRepo) FindByDataSourceExternalID( + _ context.Context, _ uint64, _ string, _ string, externalID string, +) (*types.Knowledge, error) { + if r.liveLookupErr != nil { + return nil, r.liveLookupErr + } + return r.liveByExternal[externalID], nil +} + +func (r *tombstoneRepo) FindByMetadataKeyPrefix( + context.Context, uint64, string, string, string, +) ([]*types.Knowledge, error) { + return r.children, nil +} + +func (r *tombstoneRepo) FindTombstonedByDataSourceExternalID( + _ context.Context, tenantID uint64, kbID, dataSourceID, externalID string, +) (*types.Knowledge, error) { + if r.tombstoneErr != nil { + return nil, r.tombstoneErr + } + r.lookupTenantID = tenantID + r.lookupKBID = kbID + r.lookupDataSourceID = dataSourceID + r.lookupExternalID = externalID + return r.tombstones[externalID], nil +} + +// HardDeleteKnowledge physically removes a row (update-replace path). +func (r *tombstoneRepo) HardDeleteKnowledge(_ context.Context, _ uint64, id string) error { + if r.hardDeleteErr != nil { + return r.hardDeleteErr + } + r.hardDeleted = append(r.hardDeleted, id) + if ext, ok := r.liveByID[id]; ok { + delete(r.liveByExternal, ext) + delete(r.liveByID, id) + } + for ext, row := range r.tombstones { + if row.ID == id { + delete(r.tombstones, ext) + } + } + return nil +} + +// HardDeleteKnowledgeList physically removes rows in batch (subtree sweep). +func (r *tombstoneRepo) HardDeleteKnowledgeList(_ context.Context, _ uint64, ids []string) error { + r.hardDeletedBats = append(r.hardDeletedBats, ids) + for _, id := range ids { + _ = r.HardDeleteKnowledge(context.Background(), 0, id) + } + return nil +} + +// tombstoneKS is a KnowledgeService fake. CreateKnowledgeFromFile records +// calls and honors an injected error. DeleteKnowledge / DeleteKnowledgeList +// move rows from live to tombstones like the soft-delete cascade. +type tombstoneKS struct { + interfaces.KnowledgeService + repo *tombstoneRepo + createErr error + createCalls int + deletedIDs []string +} + +func (k *tombstoneKS) GetRepository() interfaces.KnowledgeRepository { return k.repo } + +func (k *tombstoneKS) CreateKnowledgeFromFile( + context.Context, string, *multipart.FileHeader, map[string]string, + *bool, string, []string, string, *types.KnowledgeProcessOverrides, +) (*types.Knowledge, error) { + k.createCalls++ + if k.createErr != nil { + return nil, k.createErr + } + return &types.Knowledge{ID: "recreated"}, nil +} + +func (k *tombstoneKS) DeleteKnowledge(_ context.Context, id string) error { + k.deletedIDs = append(k.deletedIDs, id) + if ext, ok := k.repo.liveByID[id]; ok { + k.repo.tombstones[ext] = k.repo.liveByExternal[ext] + delete(k.repo.liveByExternal, ext) + delete(k.repo.liveByID, id) + } + return nil +} + +func (k *tombstoneKS) DeleteKnowledgeList(_ context.Context, ids []string) error { + for _, id := range ids { + _ = k.DeleteKnowledge(context.Background(), id) + } + return nil +} + +// tombstoneConnector returns a per-call round of items so a test can change +// what the source reports between syncs (e.g. a child re-appearing). +type tombstoneConnector struct { + rounds [][]types.FetchedItem + calls int +} + +func (c *tombstoneConnector) Type() string { return "test-tombstone-connector" } +func (c *tombstoneConnector) Validate(context.Context, *types.DataSourceConfig) error { + return nil +} + +func (c *tombstoneConnector) ListResources(context.Context, *types.DataSourceConfig, string) ([]types.Resource, error) { + return nil, nil +} + +func (c *tombstoneConnector) ResolveResourceAncestors( + context.Context, *types.DataSourceConfig, []string, +) ([]string, error) { + return nil, nil +} + +func (c *tombstoneConnector) FetchAll(context.Context, *types.DataSourceConfig, []string) ([]types.FetchedItem, error) { + return c.nextRound(), nil +} + +func (c *tombstoneConnector) FetchIncremental( + context.Context, *types.DataSourceConfig, *types.SyncCursor, +) ([]types.FetchedItem, *types.SyncCursor, error) { + return c.nextRound(), nil, nil +} + +func (c *tombstoneConnector) nextRound() []types.FetchedItem { + idx := c.calls + if len(c.rounds) == 0 { + return nil + } + if idx >= len(c.rounds) { + idx = len(c.rounds) - 1 + } + c.calls++ + return c.rounds[idx] +} + +func newTombstoneDataSource(t *testing.T, name string) *types.DataSource { + t.Helper() + configJSON, err := (&types.DataSourceConfig{Type: "test-tombstone-connector"}).ToJSON() + require.NoError(t, err) + return &types.DataSource{ + ID: "ds-" + name, + TenantID: 1, + KnowledgeBaseID: "kb-1", + Name: name, + Type: "test-tombstone-connector", + Config: configJSON, + SyncMode: types.SyncModeFull, + Status: types.DataSourceStatusActive, + } +} + +func runTombstoneSync( + t *testing.T, svc *DataSourceService, ds *types.DataSource, + syncLogRepo *processSyncSyncLogRepo, logID string, +) *types.SyncLog { + t.Helper() + if _, ok := syncLogRepo.logs[logID]; !ok { + syncLogRepo.logs[logID] = &types.SyncLog{ + ID: logID, + DataSourceID: ds.ID, + TenantID: ds.TenantID, + Status: types.SyncLogStatusRunning, + StartedAt: time.Now().UTC(), + } + } + payload, err := json.Marshal(types.DataSourceSyncPayload{ + DataSourceID: ds.ID, + TenantID: ds.TenantID, + SyncLogID: logID, + ForceFull: true, + }) + require.NoError(t, err) + require.NoError(t, svc.ProcessSync(context.Background(), asynq.NewTask(types.TypeDataSourceSync, payload))) + updated := syncLogRepo.logs[logID] + require.NotNil(t, updated) + return updated +} + +func newTombstoneHarness( + t *testing.T, ds *types.DataSource, connector *tombstoneConnector, ks *tombstoneKS, +) *DataSourceService { + t.Helper() + dsRepo := newKBDeleteDSRepo(ds.KnowledgeBaseID, ds) + registry := datasource.NewConnectorRegistry() + require.NoError(t, registry.Register(connector)) + return &DataSourceService{ + dsRepo: dsRepo, + syncLogRepo: &processSyncSyncLogRepo{logs: map[string]*types.SyncLog{}}, + knowledgeService: ks, + kbService: &processSyncKBService{}, + connectorRegistry: registry, + tenantRepo: &tombstoneTenantRepo{tenant: &types.Tenant{ID: ds.TenantID}}, + tagService: &tombstoneTagService{}, + } +} + +// TestProcessSync_TombstonedItemIsNotResurrected is the regression guard for +// deleted documents coming back: a document the user deleted from the KB must +// stay deleted even though the source still reports it. Before the fix the +// sync re-created the item ("deleted, 5 minutes later it comes back"). After +// the fix the item is counted as skipped and CreateKnowledgeFromFile is never +// called. +func TestProcessSync_TombstonedItemIsNotResurrected(t *testing.T) { + ds := newTombstoneDataSource(t, "tombstone") + connector := &tombstoneConnector{rounds: [][]types.FetchedItem{{{ + ExternalID: "file:gone-from-kb", + Title: "Manually Deleted Doc", + Content: []byte("# hello\n"), + FileName: "deleted.md", + }}}} + repo := newTombstoneRepo() + repo.tombstones["file:gone-from-kb"] = &types.Knowledge{ + ID: "soft-deleted-row", + DeletedAt: gorm.DeletedAt{Time: time.Now(), Valid: true}, + } + ks := &tombstoneKS{repo: repo} + svc := newTombstoneHarness(t, ds, connector, ks) + + updated := runTombstoneSync(t, svc, ds, svc.syncLogRepo.(*processSyncSyncLogRepo), "log-tombstone") + assert.Equal(t, types.SyncLogStatusSuccess, updated.Status) + assert.Equal(t, 1, updated.ItemsTotal) + assert.Equal(t, 1, updated.ItemsSkipped) + assert.Zero(t, updated.ItemsCreated, "a tombstoned item must never be re-created") + assert.Zero(t, ks.createCalls, "CreateKnowledgeFromFile must not run for a tombstoned item") + result, err := updated.ParseResult() + require.NoError(t, err) + require.Len(t, result.Errors, 1) + assert.Equal(t, syncErrorCodeUserDeletedExcluded, result.Errors[0].Code) + assert.Equal(t, ds.TenantID, repo.lookupTenantID) + assert.Equal(t, ds.KnowledgeBaseID, repo.lookupKBID) + assert.Equal(t, ds.ID, repo.lookupDataSourceID) + assert.Equal(t, "file:gone-from-kb", repo.lookupExternalID) + + // A later sync must behave identically: the exclusion is persistent. + updated2 := runTombstoneSync(t, svc, ds, svc.syncLogRepo.(*processSyncSyncLogRepo), "log-tombstone-2") + assert.Equal(t, 1, updated2.ItemsSkipped) + assert.Zero(t, ks.createCalls, "still not re-created on a later sync") +} + +// TestProcessSync_LiveLookupFailureWithTombstoneDoesNotResurrect verifies that +// a tombstone is honored even when the live-row lookup errors. +func TestProcessSync_LiveLookupFailureWithTombstoneDoesNotResurrect(t *testing.T) { + ds := newTombstoneDataSource(t, "live-lookup-fail-tombstone") + connector := &tombstoneConnector{rounds: [][]types.FetchedItem{{{ + ExternalID: "file:gone", + Title: "Gone", + Content: []byte("# hello\n"), + FileName: "gone.md", + }}}} + repo := newTombstoneRepo() + repo.liveLookupErr = assert.AnError + repo.tombstones["file:gone"] = &types.Knowledge{ + ID: "soft-deleted-row", + DeletedAt: gorm.DeletedAt{Time: time.Now(), Valid: true}, + } + ks := &tombstoneKS{repo: repo} + svc := newTombstoneHarness(t, ds, connector, ks) + + updated := runTombstoneSync(t, svc, ds, svc.syncLogRepo.(*processSyncSyncLogRepo), "log-live-fail-tombstone") + assert.Equal(t, types.SyncLogStatusSuccess, updated.Status) + assert.Equal(t, 1, updated.ItemsSkipped) + assert.Zero(t, ks.createCalls, "tombstone must block creation when live lookup fails") +} + +// TestProcessSync_LiveLookupFailureWithoutTombstoneFailsClosed verifies that a +// failed live lookup with no tombstone does not fall through to creation. +func TestProcessSync_LiveLookupFailureWithoutTombstoneFailsClosed(t *testing.T) { + ds := newTombstoneDataSource(t, "live-lookup-fail") + connector := &tombstoneConnector{rounds: [][]types.FetchedItem{{{ + ExternalID: "file:doc", + Title: "Doc", + Content: []byte("# hello\n"), + FileName: "doc.md", + }}}} + repo := newTombstoneRepo() + repo.liveLookupErr = assert.AnError + ks := &tombstoneKS{repo: repo} + svc := newTombstoneHarness(t, ds, connector, ks) + syncLogRepo := svc.syncLogRepo.(*processSyncSyncLogRepo) + syncLogRepo.logs["log-live-fail"] = &types.SyncLog{ + ID: "log-live-fail", + DataSourceID: ds.ID, + TenantID: ds.TenantID, + Status: types.SyncLogStatusRunning, + StartedAt: time.Now().UTC(), + } + + payload, err := json.Marshal(types.DataSourceSyncPayload{ + DataSourceID: ds.ID, + TenantID: ds.TenantID, + SyncLogID: "log-live-fail", + ForceFull: true, + }) + require.NoError(t, err) + require.Error(t, svc.ProcessSync(context.Background(), asynq.NewTask(types.TypeDataSourceSync, payload))) + + updated := syncLogRepo.logs["log-live-fail"] + require.NotNil(t, updated) + assert.Equal(t, 1, updated.ItemsFailed) + assert.Zero(t, ks.createCalls) +} + +// TestProcessSync_DeletingRowIsNotResurrected verifies that a row marked +// deleting (async delete in progress) is not update-replaced by sync. +func TestProcessSync_DeletingRowIsNotResurrected(t *testing.T) { + ds := newTombstoneDataSource(t, "deleting-row") + connector := &tombstoneConnector{rounds: [][]types.FetchedItem{{{ + ExternalID: "file:deleting", + Title: "Deleting Doc", + Content: []byte("# hello\n"), + FileName: "deleting.md", + }}}} + repo := newTombstoneRepo() + repo.addLive("file:deleting", &types.Knowledge{ + ID: "deleting-row", + ParseStatus: types.ParseStatusDeleting, + }) + ks := &tombstoneKS{repo: repo} + svc := newTombstoneHarness(t, ds, connector, ks) + + updated := runTombstoneSync(t, svc, ds, svc.syncLogRepo.(*processSyncSyncLogRepo), "log-deleting") + assert.Equal(t, types.SyncLogStatusSuccess, updated.Status) + assert.Equal(t, 1, updated.ItemsSkipped) + assert.Zero(t, ks.createCalls) + assert.Empty(t, ks.deletedIDs, "sync must not drive update-replace while delete is in flight") +} + +// TestProcessSync_UpdateHardDeleteFailureFailsItem verifies that a failed +// hard-delete on the update-replace path fails the item instead of leaving a +// tombstone and attempting creation. +func TestProcessSync_UpdateHardDeleteFailureFailsItem(t *testing.T) { + ds := newTombstoneDataSource(t, "hard-delete-fail") + connector := &tombstoneConnector{rounds: [][]types.FetchedItem{{{ + ExternalID: "file:doc", + Title: "Doc", + Content: []byte("# v1\n"), + FileName: "doc.md", + }}}} + repo := newTombstoneRepo() + repo.addLive("file:doc", &types.Knowledge{ID: "existing-live-row"}) + repo.hardDeleteErr = assert.AnError + ks := &tombstoneKS{repo: repo} + svc := newTombstoneHarness(t, ds, connector, ks) + syncLogRepo := svc.syncLogRepo.(*processSyncSyncLogRepo) + syncLogRepo.logs["log-hard-fail"] = &types.SyncLog{ + ID: "log-hard-fail", + DataSourceID: ds.ID, + TenantID: ds.TenantID, + Status: types.SyncLogStatusRunning, + StartedAt: time.Now().UTC(), + } + + payload, err := json.Marshal(types.DataSourceSyncPayload{ + DataSourceID: ds.ID, + TenantID: ds.TenantID, + SyncLogID: "log-hard-fail", + ForceFull: true, + }) + require.NoError(t, err) + require.Error(t, svc.ProcessSync(context.Background(), asynq.NewTask(types.TypeDataSourceSync, payload))) + + updated := syncLogRepo.logs["log-hard-fail"] + require.NotNil(t, updated) + assert.Equal(t, 1, updated.ItemsFailed) + assert.Equal(t, []string{"existing-live-row"}, ks.deletedIDs) + assert.Zero(t, ks.createCalls, "must not create when hard-delete fails after soft-delete") +} + +// TestProcessSync_TombstoneLookupFailureDoesNotResurrect verifies the +// fail-closed path: a tombstone check that errors must not fall through to +// creation; the item counts as failed instead. +func TestProcessSync_TombstoneLookupFailureDoesNotResurrect(t *testing.T) { + ds := newTombstoneDataSource(t, "tombstone-lookup-fail") + connector := &tombstoneConnector{rounds: [][]types.FetchedItem{{{ + ExternalID: "file:doc", + Title: "Doc", + Content: []byte("# hello\n"), + FileName: "doc.md", + }}}} + repo := newTombstoneRepo() + repo.tombstoneErr = assert.AnError + ks := &tombstoneKS{repo: repo} + svc := newTombstoneHarness(t, ds, connector, ks) + syncLogRepo := svc.syncLogRepo.(*processSyncSyncLogRepo) + syncLogRepo.logs["log-lookup-fail"] = &types.SyncLog{ + ID: "log-lookup-fail", + DataSourceID: ds.ID, + TenantID: ds.TenantID, + Status: types.SyncLogStatusRunning, + StartedAt: time.Now().UTC(), + } + + payload, err := json.Marshal(types.DataSourceSyncPayload{ + DataSourceID: ds.ID, + TenantID: ds.TenantID, + SyncLogID: "log-lookup-fail", + ForceFull: true, + }) + require.NoError(t, err) + // The only item fails, so the whole sync surfaces as an error. + require.Error(t, svc.ProcessSync(context.Background(), asynq.NewTask(types.TypeDataSourceSync, payload))) + + updated := syncLogRepo.logs["log-lookup-fail"] + require.NotNil(t, updated) + assert.Equal(t, 1, updated.ItemsFailed) + assert.Zero(t, updated.ItemsCreated, "a failed tombstone check must not re-create the item") + assert.Zero(t, ks.createCalls, "CreateKnowledgeFromFile must not run") +} + +// TestProcessSync_UpdateIngestFailureRecoversNextSync guards the update-replace +// path: sync = delete-then-recreate, so a transient create failure must not +// leave a tombstone behind. The next sync must re-create the item from the +// source instead of skipping it as "deleted by user". +func TestProcessSync_UpdateIngestFailureRecoversNextSync(t *testing.T) { + ds := newTombstoneDataSource(t, "update-recover") + connector := &tombstoneConnector{rounds: [][]types.FetchedItem{{{ + ExternalID: "file:doc", + Title: "Doc", + Content: []byte("# v1\n"), + FileName: "doc.md", + }}}} + repo := newTombstoneRepo() + repo.addLive("file:doc", &types.Knowledge{ID: "existing-live-row"}) + ks := &tombstoneKS{repo: repo} + svc := newTombstoneHarness(t, ds, connector, ks) + syncLogRepo := svc.syncLogRepo.(*processSyncSyncLogRepo) + + // First sync: update deletes the live row, then create fails. A sync where + // every item fails is surfaced as an error, which is expected here. + ks.createErr = assert.AnError + syncLogRepo.logs["log-update-1"] = &types.SyncLog{ + ID: "log-update-1", + DataSourceID: ds.ID, + TenantID: ds.TenantID, + Status: types.SyncLogStatusRunning, + StartedAt: time.Now().UTC(), + } + payload1, err := json.Marshal(types.DataSourceSyncPayload{ + DataSourceID: ds.ID, + TenantID: ds.TenantID, + SyncLogID: "log-update-1", + ForceFull: true, + }) + require.NoError(t, err) + require.Error(t, svc.ProcessSync(context.Background(), asynq.NewTask(types.TypeDataSourceSync, payload1))) + updated := syncLogRepo.logs["log-update-1"] + require.NotNil(t, updated) + assert.Equal(t, 1, updated.ItemsFailed) + assert.Equal(t, []string{"existing-live-row"}, ks.deletedIDs) + require.NotEmpty(t, repo.hardDeleted, "the replaced row must be hard-deleted, not left as a tombstone") + + // Second sync: the item must be re-created from the source (Created), not + // skipped as a tombstone. + ks.createErr = nil + updated2 := runTombstoneSync(t, svc, ds, syncLogRepo, "log-update-2") + assert.Zero(t, updated2.ItemsSkipped, "a failed update must not leave a tombstone behind") + assert.Equal(t, 1, updated2.ItemsCreated, "the item must be re-created on the next sync") +} + +// TestProcessSync_SweptChildReappearsIsReingested guards the subtree sweep: +// stale children removed by the sweep are sync-internal deletions. If the +// source re-adds such a child, it must be re-ingested instead of being +// skipped as "deleted by user". +func TestProcessSync_SweptChildReappearsIsReingested(t *testing.T) { + ds := newTombstoneDataSource(t, "sweep-recover") + parent := types.FetchedItem{ + ExternalID: "doc:parent", + Title: "Parent", + Content: []byte("# parent\n"), + FileName: "parent.md", + ReplacesSubtree: true, + SubtreeKeep: []string{"doc:parent#file#stays"}, + } + child := types.FetchedItem{ + ExternalID: "doc:parent#file#c1", + Title: "Child", + Content: []byte("# child\n"), + FileName: "child.md", + } + connector := &tombstoneConnector{rounds: [][]types.FetchedItem{ + {parent}, // first sync: source reports only the parent + {child}, // second sync: source re-adds the child + }} + repo := newTombstoneRepo() + repo.children = []*types.Knowledge{ + { + ID: "stale-child-row", + Metadata: types.JSON(`{"external_id":"doc:parent#file#c1","datasource_id":"ds-sweep-recover"}`), + }, + } + ks := &tombstoneKS{repo: repo} + svc := newTombstoneHarness(t, ds, connector, ks) + syncLogRepo := svc.syncLogRepo.(*processSyncSyncLogRepo) + + // First sync: parent ingests, the stale child (not in SubtreeKeep) is swept. + updated := runTombstoneSync(t, svc, ds, syncLogRepo, "log-sweep-1") + assert.Equal(t, types.SyncLogStatusSuccess, updated.Status) + require.Contains(t, ks.deletedIDs, "stale-child-row") + require.NotEmpty(t, repo.hardDeletedBats, "swept children must be hard-deleted, not left as tombstones") + + // Second sync: the source re-adds the child. It must be re-ingested. + repo.children = nil + updated2 := runTombstoneSync(t, svc, ds, syncLogRepo, "log-sweep-2") + assert.Zero(t, updated2.ItemsSkipped, "a swept child must not be treated as a user tombstone") + assert.Equal(t, 1, updated2.ItemsCreated, "a re-appearing child must be re-created") +} diff --git a/internal/types/interfaces/knowledge.go b/internal/types/interfaces/knowledge.go index b8179351c5..aa06386cbc 100644 --- a/internal/types/interfaces/knowledge.go +++ b/internal/types/interfaces/knowledge.go @@ -322,6 +322,12 @@ type KnowledgeRepository interface { FindByDataSourceExternalID( ctx context.Context, tenantID uint64, kbID, dataSourceID, externalID string, ) (*types.Knowledge, error) + // FindTombstonedByDataSourceExternalID returns the soft-deleted row for a + // (data source, external item ID) pair, or (nil, nil) when none exists. + // A deleted row is a persistent tombstone: sync never resurrects it. + FindTombstonedByDataSourceExternalID( + ctx context.Context, tenantID uint64, kbID, dataSourceID, externalID string, + ) (*types.Knowledge, error) // HardDeleteKnowledge physically removes a row after DeleteKnowledge's soft-delete // cascade. Sync-internal deletions use this so rows never become tombstones. HardDeleteKnowledge(ctx context.Context, tenantID uint64, id string) error