From 8b4e1a12f99ebc0109c8cd4ffd8ef206bbc9eb31 Mon Sep 17 00:00:00 2001 From: matevip Date: Sun, 31 May 2026 07:57:25 +0800 Subject: [PATCH] feat(wiki): scheduled single-owner source-directory watcher --- .../java/vip/mate/wiki/WikiProperties.java | 11 +++ .../service/WikiDirectoryScanService.java | 7 +- .../wiki/service/WikiRawMaterialService.java | 14 ++++ .../service/WikiSourceWatcherService.java | 77 +++++++++++++++++++ .../WikiSourceWatcherServiceE2ETest.java | 66 ++++++++++++++++ 5 files changed, 173 insertions(+), 2 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/wiki/service/WikiSourceWatcherService.java create mode 100644 mateclaw-server/src/test/java/vip/mate/wiki/service/WikiSourceWatcherServiceE2ETest.java diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java b/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java index 941bf79c..c01ec03a 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java @@ -80,6 +80,17 @@ public class WikiProperties { */ private java.util.List allowedSourceRoots = new java.util.ArrayList<>(); + /** + * When {@code true}, a scheduled job (single-owner via ShedLock) scans each + * KB's configured source directory and auto-ingests new files. Off by + * default — operators opt in. Existing dedup by source path keeps re-scans + * idempotent; deletes are never propagated. + */ + private boolean watcherEnabled = false; + + /** Interval between watcher scan cycles, milliseconds. Default 5 minutes. */ + private long watcherIntervalMs = 300_000; + /** * Wiki LLM 重试最大尝试次数（含首次）。 *

diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiDirectoryScanService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiDirectoryScanService.java index c5865800..5c0f5fbd 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiDirectoryScanService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiDirectoryScanService.java @@ -145,9 +145,12 @@ public class WikiDirectoryScanService { } if (TEXT_EXTENSIONS.contains(ext)) { - // 文本文件：读取内容 + // 文本文件：读取内容；记录 source path 以便重复扫描去重 String content = Files.readString(file, StandardCharsets.UTF_8); - rawService.addText(kbId, fileName, content); + WikiRawMaterialEntity textRaw = rawService.addText(kbId, fileName, content); + if (textRaw != null) { + rawService.updateSourcePath(textRaw.getId(), absolutePath); + } } else { // 二进制文件：直接引用原始路径，不复制 String sourceType = switch (ext) { diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java index da8de88b..33d75ddc 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiRawMaterialService.java @@ -86,6 +86,20 @@ public class WikiRawMaterialService { .eq(WikiRawMaterialEntity::getSourcePath, sourcePath)); } + /** + * Record the originating file path on a raw material via a partial update, + * so a later directory re-scan can dedup it by source path. Used for + * text-file imports, which otherwise carry no path. + */ + public void updateSourcePath(Long rawId, String sourcePath) { + if (rawId == null) { + return; + } + rawMapper.update(null, new com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper() + .eq(WikiRawMaterialEntity::getId, rawId) + .set(WikiRawMaterialEntity::getSourcePath, sourcePath)); + } + public List listPending(Long kbId) { return rawMapper.selectList( new LambdaQueryWrapper() diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiSourceWatcherService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiSourceWatcherService.java new file mode 100644 index 00000000..1fc9ef3a --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiSourceWatcherService.java @@ -0,0 +1,77 @@ +package vip.mate.wiki.service; + +import lombok.extern.slf4j.Slf4j; +import net.javacrumbs.shedlock.spring.annotation.SchedulerLock; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; +import vip.mate.wiki.WikiProperties; +import vip.mate.wiki.model.WikiKnowledgeBaseEntity; + +/** + * Watches each KB's configured source directory and auto-ingests new files. + * + *

Implemented as a periodic, single-owner scan rather than per-node OS file + * watchers: {@link SchedulerLock} (ShedLock) ensures exactly one instance runs + * a cycle, so a multi-instance deployment never double-ingests, and a periodic + * scan is inherently restart-safe (it picks up anything missed while down). The + * underlying {@link WikiDirectoryScanService} dedups by source path, so a + * re-scan only ingests genuinely new files; deletes are never propagated. + * Path validation (symlink resolution + allowed roots) is enforced by the + * shared validator inside the scan. + * + * @author MateClaw Team + */ +@Slf4j +@Service +public class WikiSourceWatcherService { + + private final WikiKnowledgeBaseService kbService; + private final WikiDirectoryScanService scanService; + private final WikiProperties properties; + + public WikiSourceWatcherService(WikiKnowledgeBaseService kbService, + WikiDirectoryScanService scanService, + WikiProperties properties) { + this.kbService = kbService; + this.scanService = scanService; + this.properties = properties; + } + + /** Scheduled entry point — gated by config, serialized across instances. */ + @Scheduled(fixedDelayString = "${mate.wiki.watcher-interval-ms:300000}", initialDelay = 60_000) + @SchedulerLock(name = "wiki-source-watcher", lockAtMostFor = "PT10M", lockAtLeastFor = "PT30S") + public void scheduledScan() { + if (!properties.isWatcherEnabled()) { + return; + } + int added = runScanCycle(); + if (added > 0) { + log.info("[WikiWatcher] scan cycle ingested {} new file(s)", added); + } + } + + /** + * Scan every KB that has a source directory configured and return the total + * number of new files ingested. Per-KB failures are logged and skipped so + * one bad directory cannot stall the others. + */ + public int runScanCycle() { + int totalAdded = 0; + for (WikiKnowledgeBaseEntity kb : kbService.listAll()) { + String dir = kb.getSourceDirectory(); + if (dir == null || dir.isBlank()) { + continue; + } + try { + WikiDirectoryScanService.ScanResult result = scanService.scanDirectory(kb.getId(), dir); + totalAdded += result.added(); + if (!result.errors().isEmpty()) { + log.warn("[WikiWatcher] KB {} scan reported issues: {}", kb.getId(), result.errors()); + } + } catch (Exception e) { + log.warn("[WikiWatcher] scan failed for KB {}: {}", kb.getId(), e.getMessage()); + } + } + return totalAdded; + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/wiki/service/WikiSourceWatcherServiceE2ETest.java b/mateclaw-server/src/test/java/vip/mate/wiki/service/WikiSourceWatcherServiceE2ETest.java new file mode 100644 index 00000000..5575026d --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/wiki/service/WikiSourceWatcherServiceE2ETest.java @@ -0,0 +1,66 @@ +package vip.mate.wiki.service; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import vip.mate.wiki.model.WikiKnowledgeBaseEntity; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * End-to-end test of the source watcher's scan cycle against H2: new files in a + * KB's source directory are auto-ingested, and a re-scan is idempotent (dedup + * by source path). Auto-processing is disabled so the test stays model-free. + */ +@SpringBootTest( + webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = { + "spring.flyway.enabled=true", + "spring.flyway.locations=classpath:db/migration/h2", + "mateclaw.feature-flag.refresh-ms=999999", + "mate.wiki.auto-process-on-upload=false" + } +) +class WikiSourceWatcherServiceE2ETest { + + @Autowired + private WikiSourceWatcherService watcherService; + @Autowired + private WikiKnowledgeBaseService kbService; + + private static final java.util.concurrent.atomic.AtomicLong SEQ = + new java.util.concurrent.atomic.AtomicLong(System.nanoTime()); + + @Test + void scanCycleIngestsNewFiles_thenDedups(@TempDir Path sourceDir) throws IOException { + Files.writeString(sourceDir.resolve("note-a.md"), "# Note A\n\ncontent a"); + Files.writeString(sourceDir.resolve("note-b.md"), "# Note B\n\ncontent b"); + + WikiKnowledgeBaseEntity kb = kbService.create( + "watcher-" + SEQ.incrementAndGet(), "test", null); + kbService.updateSourceDirectory(kb.getId(), sourceDir.toString()); + + // First cycle ingests both new files. + int firstAdded = watcherService.runScanCycle(); + assertTrue(firstAdded >= 2, "expected >= 2 new files, got " + firstAdded); + + // A new file appears; the next cycle ingests only it (existing files dedup). + Files.writeString(sourceDir.resolve("note-c.md"), "# Note C\n\ncontent c"); + int secondAdded = watcherService.runScanCycle(); + assertEquals(1, secondAdded, "only the newly added file should ingest"); + } + + @Test + void kbsWithoutSourceDirectory_areSkipped() { + // A KB with no source directory must not cause errors in the cycle. + kbService.create("nodir-" + SEQ.incrementAndGet(), "test", null); + // Should complete without throwing (count is non-negative). + assertTrue(watcherService.runScanCycle() >= 0); + } +}