Files
Enterprise-Knowledge-Base/internal/application/service/knowledge_delete.go
T
wizardchen 04e8105728 perf(wiki-ingest): scale to 4w-doc KBs + generic task queue / dead-letter
The wiki ingest post-process pipeline OOM'd and ran for hours on KBs
with ~40k documents. The dominant tail was a per-batch ListAllPages
that pulled every page (multi-MB content blobs) into Go memory, plus a
24h-TTL Redis pending list whose data could be evicted before a long
serialized backlog drained. dedup did O(P × N) Jaccard scoring in Go
on every batch. Lint loaded the full graph. The index page kept the
entire wiki directory in its content column and rewrote the TOAST
chunk on every ingest. None of these survive at 4w docs.

This change reworks the write path end to end and pulls the durable-
queue + dead-letter primitives out of wiki and into shared
infrastructure that every asynq task type now benefits from.

  - migration 000041: task_pending_ops + task_dead_letters tables, plus
    three GIN indexes on wiki_pages (source_refs jsonb_path_ops,
    source_refs text fallback, lower(title) trgm).
  - TaskPendingOpsRepository + TaskDeadLetterRepository in
    internal/application/repository/task_queue.go: cursor-paginated
    list, atomic IncrFailCount via UPDATE…RETURNING, dedup-key scoped
    delete that refuses empty keys so a buggy caller can't wipe a KB's
    queue.
  - internal/middleware/asynqdl: writes a task_dead_letters row when an
    asynq task exhausts its retry budget. Payload-agnostic — a small
    probe struct extracts TenantID + scope hints across every existing
    payload type, so summary:generation / image:multimodal /
    faq_import / kb:clone / etc. all dead-letter without per-handler
    code. Best-effort: an insert failure never masks the underlying
    task error. Installed in router/task.go before the langfuse mw so
    it sees raw errors.

  - Pending queue moved Redis → PG (no TTL, restart-durable). Redis
    keeps just the active-batch lock and the delete tombstone.
  - In-batch retry budget (wikiMaxFailRetries=5) tracked via
    pendingRepo.IncrFailCount; over the cap the row is moved to
    task_dead_letters with task_type=wiki:ingest, related_id=knowledge
    id. Asynq retries (10) are handled by the new middleware.
  - Removed the per-batch ListAllPages. WikiBatchContext now carries
    lazy fetcher closures (SlugTitleMany, SummaryByKnowledgeID) with
    mutex-protected caches; reduce reaches for titles / summaries on
    demand instead of pre-loading the whole KB.
  - getExistingPageSlugsForKnowledge now uses ListSlugsBySourceRef,
    which the new GIN index on source_refs serves as a Bitmap Index
    Scan instead of a sequential text LIKE.
  - Dedup pre-filter uses pg_trgm via FindSimilarPages
    (idx_wiki_pages_title_trgm). For each new entity/concept (and each
    of its aliases) we ask the DB for the top-K trigram-similar
    existing pages and union the results — bounded prompt size, no Go-
    side O(P × M) loop. Small KBs (≤25 entities) bypass the pre-filter.
  - cleanDeadLinks / injectCrossLinks are scoped to the batch's
    affected slugs (a few dozen) instead of every page in the KB. Both
    use the new lite ListBySlugs / ExistsSlugs repo methods so they
    pull only slug + title + outlinks, not full content.
  - Slug fuzzy resolve (slug_fuzzy.go): when the LLM emits
    [[bad-slug|display]] the cleanup path now tries display-text
    reverse lookup → hyphen/case normalized equality → char-bigram
    Jaccard ≥ 0.8 before stripping. Recovers the common pinyin-word-
    break drift case ("shang-hai-tower" vs "shanghai-tower") in place
    instead of replacing the link with plain text.
  - rebuildIndexPage uses ListByTypeRecent(200) for the first-time
    intro and drops the full DocumentSummaries blob from the
    incremental update prompt, so its context stays bounded regardless
    of KB size.
  - Concurrency tunables surfaced in WikiConfig: IngestBatchSize /
    IngestMapParallel / IngestReduceParallel, with sensible defaults
    via OrDefault helpers. scheduleFollowUp drops to ProcessIn(0) so
    follow-ups don't waste asynq retry slots bouncing on the active
    lock.

  - RunLint walks pages via the new ListPagesCursor in 200-page
    windows and computes the live-slug set with a one-column
    ListAllSlugs Pluck instead of a Limit:0 GetGraph that materialized
    every node + edge. Memory is now bounded; 40k pages walks in
    constant ~4MB.

  - 14 GORM tests for both repos against an in-memory SQLite mirror of
    the production DDL (task_queue_test.go).
  - 6 tests for the dead-letter middleware covering retry budget
    detection, payload-agnostic scope inference, error truncation, and
    repo-failure isolation.
  - 7 tests for the slug fuzzy resolve helper covering the three
    resolution stages, display-text priority over normalized equality,
    bigram fallback acceptance, and rejection of unrelated slugs.
  - Existing wiki_ingest / wiki_lint / wiki_page tests updated to the
    new fetcher / cursor APIs.
2026-05-10 00:01:06 +08:00

653 lines
25 KiB
Go

package service
import (
"context"
"encoding/json"
"errors"
"strings"
"time"
"github.com/Tencent/WeKnora/internal/application/service/retriever"
"github.com/Tencent/WeKnora/internal/logger"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/types/interfaces"
"github.com/hibiken/asynq"
"golang.org/x/sync/errgroup"
)
// collectImageURLs extracts unique provider:// image URLs from image_info JSON strings.
func collectImageURLs(ctx context.Context, imageInfos []string) []string {
seen := make(map[string]struct{})
var urls []string
for _, info := range imageInfos {
if info == "" {
continue
}
var images []*types.ImageInfo
if err := json.Unmarshal([]byte(info), &images); err != nil {
logger.Warnf(ctx, "Failed to parse image_info JSON: %v", err)
continue
}
for _, img := range images {
if img.URL != "" {
if _, exists := seen[img.URL]; !exists {
seen[img.URL] = struct{}{}
urls = append(urls, img.URL)
}
}
}
}
return urls
}
// deleteExtractedImages deletes all extracted image files from storage.
// Standalone function — callable from both knowledgeService and knowledgeBaseService.
// Errors are logged but do not fail the overall deletion.
func deleteExtractedImages(ctx context.Context, fileSvc interfaces.FileService, imageURLs []string) {
if len(imageURLs) == 0 {
return
}
logger.Infof(ctx, "Deleting %d extracted images", len(imageURLs))
for _, url := range imageURLs {
if err := fileSvc.DeleteFile(ctx, url); err != nil {
logger.Errorf(ctx, "Failed to delete extracted image %s: %v", url, err)
}
}
}
// DeleteKnowledge deletes a knowledge entry and all related resources
func (s *knowledgeService) DeleteKnowledge(ctx context.Context, id string) error {
// Get the knowledge entry
knowledge, err := s.repo.GetKnowledgeByID(ctx, ctx.Value(types.TenantIDContextKey).(uint64), id)
if err != nil {
return err
}
// Mark as deleting first to prevent async task conflicts
// This ensures that any running async tasks will detect the deletion and abort
originalStatus := knowledge.ParseStatus
knowledge.ParseStatus = types.ParseStatusDeleting
knowledge.UpdatedAt = time.Now()
if err := s.repo.UpdateKnowledge(ctx, knowledge); err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge failed to mark as deleting")
// Continue with deletion even if marking fails
} else {
logger.Infof(ctx, "Marked knowledge %s as deleting (previous status: %s)", id, originalStatus)
}
// Resolve file service for this KB before spawning goroutines
kb, _ := s.kbService.GetKnowledgeBaseByID(ctx, knowledge.KnowledgeBaseID)
kbFileSvc := s.resolveFileService(ctx, kb)
// Collect image URLs before chunks are deleted (ImageInfo references are lost after deletion)
tenantID := ctx.Value(types.TenantIDContextKey).(uint64)
chunkImageInfos, err := s.chunkService.GetRepository().ListImageInfoByKnowledgeIDs(ctx, tenantID, []string{id})
if err != nil {
logger.Errorf(ctx, "Failed to collect image URLs for cleanup: %v", err)
}
var imageInfoStrs []string
for _, ci := range chunkImageInfos {
imageInfoStrs = append(imageInfoStrs, ci.ImageInfo)
}
imageURLs := collectImageURLs(ctx, imageInfoStrs)
wg := errgroup.Group{}
// Delete knowledge embeddings from vector store.
// Skip entirely when the knowledge has no embedding model (e.g. Wiki-only KB):
// nothing was ever written to the vector store, so there is nothing to delete,
// and GetEmbeddingModel would fail with "model ID cannot be empty".
if strings.TrimSpace(knowledge.EmbeddingModelID) != "" {
wg.Go(func() error {
tenantInfo := ctx.Value(types.TenantInfoContextKey).(*types.Tenant)
retrieveEngine, err := retriever.NewCompositeRetrieveEngine(
s.retrieveEngine,
tenantInfo.GetEffectiveEngines(),
)
if err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge delete knowledge embedding failed")
return err
}
embeddingModel, err := s.modelService.GetEmbeddingModel(ctx, knowledge.EmbeddingModelID)
if err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge delete knowledge embedding failed")
return err
}
if err := retrieveEngine.DeleteByKnowledgeIDList(ctx, []string{knowledge.ID}, embeddingModel.GetDimensions(), knowledge.Type); err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge delete knowledge embedding failed")
return err
}
return nil
})
} else {
logger.Infof(ctx, "Knowledge %s has no embedding model, skipping vector store cleanup", knowledge.ID)
}
// Delete all chunks associated with this knowledge
wg.Go(func() error {
if err := s.chunkService.DeleteChunksByKnowledgeID(ctx, knowledge.ID); err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge delete chunks failed")
return err
}
return nil
})
// Delete the physical file and extracted images if they exist
wg.Go(func() error {
if knowledge.FilePath != "" {
if err := kbFileSvc.DeleteFile(ctx, knowledge.FilePath); err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge delete file failed")
}
}
deleteExtractedImages(ctx, kbFileSvc, imageURLs)
tenantInfo := ctx.Value(types.TenantInfoContextKey).(*types.Tenant)
tenantInfo.StorageUsed -= knowledge.StorageSize
if err := s.tenantRepo.AdjustStorageUsed(ctx, tenantInfo.ID, -knowledge.StorageSize); err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge update tenant storage used failed")
}
return nil
})
// Delete the knowledge graph
wg.Go(func() error {
namespace := types.NameSpace{KnowledgeBase: knowledge.KnowledgeBaseID, Knowledge: knowledge.ID}
if err := s.graphEngine.DelGraph(ctx, []types.NameSpace{namespace}); err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge delete knowledge graph failed")
return err
}
return nil
})
// Clean up wiki pages that reference this knowledge. Pass the full
// knowledge object so cleanup can source title/summary from the row
// itself rather than reaching into possibly-not-yet-written wiki pages.
if kb != nil && kb.IsWikiEnabled() {
wg.Go(func() error {
s.cleanupWikiOnKnowledgeDelete(ctx, knowledge)
return nil
})
}
if err = wg.Wait(); err != nil {
return err
}
// Delete the knowledge entry itself from the database
return s.repo.DeleteKnowledge(ctx, ctx.Value(types.TenantIDContextKey).(uint64), id)
}
// cleanupWikiOnKnowledgeDelete handles wiki pages when a source document is deleted.
//
// There are three sources of truth we must keep consistent:
// - The knowledge row (being soft-deleted right now by the caller)
// - Wiki pages whose source_refs include this knowledge
// - Pending/in-flight wiki_ingest tasks that may create *new* pages pointing at it
//
// The function is deliberately best-effort and idempotent:
// - It writes a tombstone + scrubs pending ingest ops so new pages cannot be
// born with a stale source_ref (guards (a) queued ingest and (b) ingest
// tasks mid-LLM call — both consult the tombstone before writing).
// - It immediately reconciles any pages already present (delete-if-only-ref
// or strip-ref-if-multi).
// - It *unconditionally* enqueues a retract task. Crucially we DO NOT gate
// enqueue on "pages currently exist": in the ingest/delete race the
// knowledge may have pages that exist only after this function returns
// (the ingest task fires later and, absent the tombstone, would have
// created them). The retract handler re-queries ListPagesBySourceRef at
// run time, so even with an empty PageSlugs it will do the right thing —
// and at worst it's a cheap no-op.
func (s *knowledgeService) cleanupWikiOnKnowledgeDelete(ctx context.Context, knowledge *types.Knowledge) {
if knowledge == nil {
return
}
kbID := knowledge.KnowledgeBaseID
knowledgeID := knowledge.ID
if kbID == "" || knowledgeID == "" {
return
}
// (1) Tombstone + scrub pending ingest — must happen first so any
// wiki_ingest task that wakes up between here and the retract enqueue
// below sees "knowledge gone" and bails out.
s.markKnowledgeDeletedForWiki(ctx, kbID, knowledgeID)
s.scrubWikiPendingIngest(ctx, kbID, knowledgeID, "cleanup")
// Pull title/summary from the knowledge itself — do NOT read them from
// existing wiki pages. In the race window wiki pages may not exist yet,
// and even when they do their "summary" is the LLM-extracted one which
// we're about to invalidate anyway. The knowledge row still has the
// original Title/FileName/Description, which is what the retract prompt
// actually wants.
docTitle := knowledge.Title
if docTitle == "" {
docTitle = knowledge.FileName
}
if docTitle == "" {
docTitle = knowledgeID
}
docSummary := knowledge.Description
// (2) Immediate reconciliation for pages already present. If ingest
// hasn't run yet this simply finds nothing; that's fine — see (3).
pages, err := s.wikiRepo.ListBySourceRef(ctx, kbID, knowledgeID)
if err != nil {
logger.Warnf(ctx, "wiki cleanup: failed to list pages by source ref %s: %v", knowledgeID, err)
pages = nil
}
// Prefer the on-disk summary if the summary page already exists (it's
// richer than the raw user-provided description). Leave docSummary
// untouched otherwise so we still pass something meaningful downstream.
for _, page := range pages {
if page.PageType == types.WikiPageTypeSummary && page.Summary != "" {
docSummary = page.Summary
break
}
}
var deletedSlugs []string
var retractSlugs []string
for _, page := range pages {
if page.PageType == types.WikiPageTypeIndex || page.PageType == types.WikiPageTypeLog {
continue
}
remaining := removeSourceRef(page.SourceRefs, knowledgeID)
if len(remaining) == 0 {
if err := s.wikiService.DeletePage(ctx, kbID, page.Slug); err != nil {
logger.Warnf(ctx, "wiki cleanup: failed to delete page %s: %v", page.Slug, err)
} else {
deletedSlugs = append(deletedSlugs, page.Slug)
}
} else {
page.SourceRefs = remaining
if err := s.wikiService.UpdatePageMeta(ctx, page); err != nil {
logger.Warnf(ctx, "wiki cleanup: failed to update source refs for page %s: %v", page.Slug, err)
} else {
retractSlugs = append(retractSlugs, page.Slug)
}
}
}
if len(deletedSlugs) > 0 {
logger.Infof(ctx, "wiki cleanup: deleted %d pages after knowledge %s deletion: %v",
len(deletedSlugs), knowledgeID, deletedSlugs)
}
allAffectedSlugs := append(retractSlugs, deletedSlugs...)
// (3) Unconditionally enqueue the retract task. See function comment —
// an empty PageSlugs is not a bug, it's the signal "re-query at run
// time". The handler will ListPagesBySourceRef again, pick up any
// pages that materialised after we looked, and also rebuild index/log
// so the knowledge's disappearance is reflected in the UI.
lang, _ := types.LanguageFromContext(ctx)
tenantID, _ := types.TenantIDFromContext(ctx)
EnqueueWikiRetract(ctx, s.task, s.taskPendingRepo, WikiRetractPayload{
TenantID: tenantID,
KnowledgeBaseID: kbID,
KnowledgeID: knowledgeID,
DocTitle: docTitle,
DocSummary: docSummary,
Language: lang,
PageSlugs: allAffectedSlugs,
})
logger.Infof(ctx, "wiki cleanup: enqueued retract task for knowledge %s (%d known slugs: %v)",
knowledgeID, len(allAffectedSlugs), allAffectedSlugs)
}
// markKnowledgeDeletedForWiki writes a short-TTL tombstone so any wiki_ingest
// task still running or queued for this knowledge can short-circuit before
// resurrecting a page with a stale source_ref. No-op when Redis is absent.
func (s *knowledgeService) markKnowledgeDeletedForWiki(ctx context.Context, kbID, knowledgeID string) {
if s.redisClient == nil || kbID == "" || knowledgeID == "" {
return
}
key := WikiDeletedTombstoneKey(kbID, knowledgeID)
if err := s.redisClient.Set(ctx, key, "1", wikiDeletedTTL).Err(); err != nil {
logger.Warnf(ctx, "wiki cleanup: failed to write tombstone %s: %v", key, err)
}
}
// scrubWikiPendingIngest removes queued WikiOpIngest entries for a knowledge
// from task_pending_ops. Used by both the delete path (we're about to
// soft-delete the doc, no point ingesting it) and the reparse path (the
// old chunks are about to vanish, so any pending ingest would either race
// with the cleanup or no-op on an empty chunk set — and the post-process
// task will enqueue a fresh ingest once new chunks land anyway).
//
// Retract entries stay put — delete still needs them to unlink referencing
// pages, and reparse never enqueues retracts for the doc being reparsed.
// We pass op=WikiOpIngest so DeleteByDedupKey filters to the ingest rows
// only.
func (s *knowledgeService) scrubWikiPendingIngest(ctx context.Context, kbID, knowledgeID, reason string) {
if s.taskPendingRepo == nil || kbID == "" || knowledgeID == "" {
return
}
if err := s.taskPendingRepo.DeleteByDedupKey(ctx, wikiTaskType, wikiTaskScope, kbID, knowledgeID, WikiOpIngest); err != nil {
logger.Warnf(ctx, "wiki %s: failed to scrub pending ingest ops for knowledge %s: %v", reason, knowledgeID, err)
return
}
logger.Infof(ctx, "wiki %s: scrubbed pending ingest ops for knowledge %s", reason, knowledgeID)
}
// prepareWikiForReparse is the reparse counterpart to
// cleanupWikiOnKnowledgeDelete. It aligns reparse with the same "pending
// queue hygiene" the delete path already enforces, without taking any
// destructive action against existing pages.
//
// Why no retract / tombstone here: reparse is not a "K is gone" event, it's
// a "K's contribution is about to be swapped for a new version" event. The
// actual swap happens asynchronously inside mapOneDocument (see its
// oldPageSlugs handling) — that's where we have both the old page set and
// the freshly extracted candidate slugs, which is exactly the information
// the WikiPageModifyPrompt needs to do a correct replace-not-append.
//
// So the only thing worth doing synchronously at reparse time is keeping
// the Redis pending list clean so the re-ingest enqueued by
// KnowledgePostProcess doesn't race with a stale ingest op that would
// fire mid-flight against zero chunks.
func (s *knowledgeService) prepareWikiForReparse(ctx context.Context, knowledge *types.Knowledge) {
if knowledge == nil {
return
}
kbID := knowledge.KnowledgeBaseID
knowledgeID := knowledge.ID
if kbID == "" || knowledgeID == "" {
return
}
s.scrubWikiPendingIngest(ctx, kbID, knowledgeID, "reparse")
}
// removeSourceRef removes entries from source_refs that match a knowledge ID.
// Handles both old format ("knowledgeID") and new format ("knowledgeID|title").
func removeSourceRef(refs types.StringArray, knowledgeID string) types.StringArray {
var result types.StringArray
prefix := knowledgeID + "|"
for _, ref := range refs {
if ref == knowledgeID || strings.HasPrefix(ref, prefix) {
continue
}
result = append(result, ref)
}
return result
}
// DeleteKnowledgeList deletes a knowledge entry and all related resources
func (s *knowledgeService) DeleteKnowledgeList(ctx context.Context, ids []string) error {
if len(ids) == 0 {
return nil
}
// 1. Get the knowledge entry
tenantInfo := ctx.Value(types.TenantInfoContextKey).(*types.Tenant)
knowledgeList, err := s.repo.GetKnowledgeBatch(ctx, tenantInfo.ID, ids)
if err != nil {
return err
}
// Mark all as deleting first to prevent async task conflicts
for _, knowledge := range knowledgeList {
knowledge.ParseStatus = types.ParseStatusDeleting
knowledge.UpdatedAt = time.Now()
if err := s.repo.UpdateKnowledge(ctx, knowledge); err != nil {
logger.GetLogger(ctx).WithField("error", err).WithField("knowledge_id", knowledge.ID).
Errorf("DeleteKnowledgeList failed to mark as deleting")
// Continue with deletion even if marking fails
}
}
logger.Infof(ctx, "Marked %d knowledge entries as deleting", len(knowledgeList))
// Pre-resolve file services per KB so goroutines don't need DB access
kbFileServices := make(map[string]interfaces.FileService)
for _, knowledge := range knowledgeList {
if _, ok := kbFileServices[knowledge.KnowledgeBaseID]; !ok {
kb, _ := s.kbService.GetKnowledgeBaseByID(ctx, knowledge.KnowledgeBaseID)
kbFileServices[knowledge.KnowledgeBaseID] = s.resolveFileService(ctx, kb)
}
}
// Collect image URLs before chunks are deleted
chunkImageInfos, err := s.chunkService.GetRepository().ListImageInfoByKnowledgeIDs(ctx, tenantInfo.ID, ids)
if err != nil {
logger.Errorf(ctx, "Failed to collect image URLs for batch cleanup: %v", err)
}
knowledgeToKB := make(map[string]string)
for _, k := range knowledgeList {
knowledgeToKB[k.ID] = k.KnowledgeBaseID
}
kbImageInfos := make(map[string][]string) // kbID → []imageInfo JSON
for _, ci := range chunkImageInfos {
kbID := knowledgeToKB[ci.KnowledgeID]
kbImageInfos[kbID] = append(kbImageInfos[kbID], ci.ImageInfo)
}
kbImageURLs := make(map[string][]string) // kbID → []imageURL (deduplicated)
for kbID, infos := range kbImageInfos {
kbImageURLs[kbID] = collectImageURLs(ctx, infos)
}
wg := errgroup.Group{}
// 2. Delete knowledge embeddings from vector store
wg.Go(func() error {
tenantInfo := ctx.Value(types.TenantInfoContextKey).(*types.Tenant)
retrieveEngine, err := retriever.NewCompositeRetrieveEngine(
s.retrieveEngine,
tenantInfo.GetEffectiveEngines(),
)
if err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge delete knowledge embedding failed")
return err
}
// Group by EmbeddingModelID and Type
type groupKey struct {
EmbeddingModelID string
Type string
}
group := map[groupKey][]string{}
for _, knowledge := range knowledgeList {
key := groupKey{EmbeddingModelID: knowledge.EmbeddingModelID, Type: knowledge.Type}
group[key] = append(group[key], knowledge.ID)
}
for key, knowledgeIDs := range group {
// Wiki-only knowledge never had embeddings written to the vector store,
// and its EmbeddingModelID is intentionally empty. Skip the whole group
// to avoid the spurious "model ID cannot be empty" failure.
if strings.TrimSpace(key.EmbeddingModelID) == "" {
logger.Infof(ctx, "Skipping vector store cleanup for %d knowledge entries without embedding model", len(knowledgeIDs))
continue
}
embeddingModel, err := s.modelService.GetEmbeddingModel(ctx, key.EmbeddingModelID)
if err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge get embedding model failed")
return err
}
if err := retrieveEngine.DeleteByKnowledgeIDList(ctx, knowledgeIDs, embeddingModel.GetDimensions(), key.Type); err != nil {
logger.GetLogger(ctx).
WithField("error", err).
Errorf("DeleteKnowledge delete knowledge embedding failed")
return err
}
}
return nil
})
// 3. Delete all chunks associated with this knowledge
wg.Go(func() error {
if err := s.chunkService.DeleteByKnowledgeList(ctx, ids); err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge delete chunks failed")
return err
}
return nil
})
// 4. Delete the physical file and extracted images if they exist
wg.Go(func() error {
storageAdjust := int64(0)
for _, knowledge := range knowledgeList {
if knowledge.FilePath != "" {
fSvc := kbFileServices[knowledge.KnowledgeBaseID]
if err := fSvc.DeleteFile(ctx, knowledge.FilePath); err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge delete file failed")
}
}
storageAdjust -= knowledge.StorageSize
}
// Delete extracted images per KB
for kbID, urls := range kbImageURLs {
fSvc := kbFileServices[kbID]
if fSvc == nil {
logger.Warnf(ctx, "No file service for KB %s, skipping %d image deletions", kbID, len(urls))
continue
}
deleteExtractedImages(ctx, fSvc, urls)
}
tenantInfo.StorageUsed += storageAdjust
if err := s.tenantRepo.AdjustStorageUsed(ctx, tenantInfo.ID, storageAdjust); err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge update tenant storage used failed")
}
return nil
})
// Delete the knowledge graph
wg.Go(func() error {
namespaces := []types.NameSpace{}
for _, knowledge := range knowledgeList {
namespaces = append(
namespaces,
types.NameSpace{KnowledgeBase: knowledge.KnowledgeBaseID, Knowledge: knowledge.ID},
)
}
if err := s.graphEngine.DelGraph(ctx, namespaces); err != nil {
logger.GetLogger(ctx).WithField("error", err).Errorf("DeleteKnowledge delete knowledge graph failed")
return err
}
return nil
})
// Clean up wiki pages that reference deleted knowledge. cleanup needs
// the full knowledge object (Title / Description) so the retract prompt
// can describe the vanished document even when wiki pages haven't been
// ingested yet — which is common in the batch-delete-shortly-after-upload
// flow.
wg.Go(func() error {
for _, knowledge := range knowledgeList {
kb, _ := s.kbService.GetKnowledgeBaseByID(ctx, knowledge.KnowledgeBaseID)
if kb != nil && kb.IsWikiEnabled() {
s.cleanupWikiOnKnowledgeDelete(ctx, knowledge)
}
}
return nil
})
if err = wg.Wait(); err != nil {
return err
}
// 5. Delete the knowledge entry itself from the database
return s.repo.DeleteKnowledgeList(ctx, tenantInfo.ID, ids)
}
func (s *knowledgeService) cleanupKnowledgeResources(ctx context.Context, knowledge *types.Knowledge) error {
logger.GetLogger(ctx).Infof("Cleaning knowledge resources before manual update, knowledge ID: %s", knowledge.ID)
var cleanupErr error
if knowledge.ParseStatus == types.ManualKnowledgeStatusDraft && knowledge.StorageSize == 0 {
// Draft without indexed data, skip cleanup.
return nil
}
tenantInfo := ctx.Value(types.TenantInfoContextKey).(*types.Tenant)
if knowledge.EmbeddingModelID != "" {
retrieveEngine, err := retriever.NewCompositeRetrieveEngine(
s.retrieveEngine,
tenantInfo.GetEffectiveEngines(),
)
if err != nil {
logger.GetLogger(ctx).WithField("error", err).Error("Failed to init retrieve engine during cleanup")
cleanupErr = errors.Join(cleanupErr, err)
} else {
embeddingModel, modelErr := s.modelService.GetEmbeddingModel(ctx, knowledge.EmbeddingModelID)
if modelErr != nil {
logger.GetLogger(ctx).WithField("error", modelErr).Error("Failed to get embedding model during cleanup")
cleanupErr = errors.Join(cleanupErr, modelErr)
} else {
if err := retrieveEngine.DeleteByKnowledgeIDList(ctx, []string{knowledge.ID}, embeddingModel.GetDimensions(), knowledge.Type); err != nil {
logger.GetLogger(ctx).WithField("error", err).Error("Failed to delete manual knowledge index")
cleanupErr = errors.Join(cleanupErr, err)
}
}
}
}
// Collect image URLs before chunks are deleted
kb, _ := s.kbService.GetKnowledgeBaseByID(ctx, knowledge.KnowledgeBaseID)
fileSvc := s.resolveFileService(ctx, kb)
chunkImageInfos, imgErr := s.chunkService.GetRepository().ListImageInfoByKnowledgeIDs(ctx, tenantInfo.ID, []string{knowledge.ID})
if imgErr != nil {
logger.GetLogger(ctx).WithField("error", imgErr).Error("Failed to collect image URLs for cleanup")
cleanupErr = errors.Join(cleanupErr, imgErr)
}
var imageInfoStrs []string
for _, ci := range chunkImageInfos {
imageInfoStrs = append(imageInfoStrs, ci.ImageInfo)
}
imageURLs := collectImageURLs(ctx, imageInfoStrs)
if err := s.chunkService.DeleteChunksByKnowledgeID(ctx, knowledge.ID); err != nil {
logger.GetLogger(ctx).WithField("error", err).Error("Failed to delete manual knowledge chunks")
cleanupErr = errors.Join(cleanupErr, err)
}
// Delete extracted images after chunks are deleted
deleteExtractedImages(ctx, fileSvc, imageURLs)
namespace := types.NameSpace{KnowledgeBase: knowledge.KnowledgeBaseID, Knowledge: knowledge.ID}
if err := s.graphEngine.DelGraph(ctx, []types.NameSpace{namespace}); err != nil {
logger.GetLogger(ctx).WithField("error", err).Error("Failed to delete manual knowledge graph data")
cleanupErr = errors.Join(cleanupErr, err)
}
if knowledge.StorageSize > 0 {
tenantInfo.StorageUsed -= knowledge.StorageSize
if tenantInfo.StorageUsed < 0 {
tenantInfo.StorageUsed = 0
}
if err := s.tenantRepo.AdjustStorageUsed(ctx, tenantInfo.ID, -knowledge.StorageSize); err != nil {
logger.GetLogger(ctx).WithField("error", err).Error("Failed to adjust storage usage during manual cleanup")
cleanupErr = errors.Join(cleanupErr, err)
}
knowledge.StorageSize = 0
}
return cleanupErr
}
// ProcessKnowledgeListDelete handles Asynq knowledge list delete tasks
func (s *knowledgeService) ProcessKnowledgeListDelete(ctx context.Context, t *asynq.Task) error {
var payload types.KnowledgeListDeletePayload
if err := json.Unmarshal(t.Payload(), &payload); err != nil {
logger.Errorf(ctx, "Failed to unmarshal knowledge list delete payload: %v", err)
return err
}
logger.Infof(ctx, "Processing knowledge list delete task for %d knowledge items", len(payload.KnowledgeIDs))
// Get tenant info
tenant, err := s.tenantRepo.GetTenantByID(ctx, payload.TenantID)
if err != nil {
logger.Errorf(ctx, "Failed to get tenant %d: %v", payload.TenantID, err)
return err
}
// Set context values
ctx = context.WithValue(ctx, types.TenantIDContextKey, payload.TenantID)
ctx = context.WithValue(ctx, types.TenantInfoContextKey, tenant)
// Delete knowledge list
if err := s.DeleteKnowledgeList(ctx, payload.KnowledgeIDs); err != nil {
logger.Errorf(ctx, "Failed to delete knowledge list: %v", err)
return err
}
logger.Infof(ctx, "Successfully deleted %d knowledge items", len(payload.KnowledgeIDs))
return nil
}