mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-14 03:33:43 +08:00
feat(wiki): scheduled single-owner source-directory watcher
This commit is contained in:
parent
cc8c9cf951
commit
8b4e1a12f9
@ -80,6 +80,17 @@ public class WikiProperties {
|
|||||||
*/
|
*/
|
||||||
private java.util.List<String> allowedSourceRoots = new java.util.ArrayList<>();
|
private java.util.List<String> 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 重试最大尝试次数(含首次)。
|
* Wiki LLM 重试最大尝试次数(含首次)。
|
||||||
* <p>
|
* <p>
|
||||||
|
|||||||
@ -145,9 +145,12 @@ public class WikiDirectoryScanService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (TEXT_EXTENSIONS.contains(ext)) {
|
if (TEXT_EXTENSIONS.contains(ext)) {
|
||||||
// 文本文件:读取内容
|
// 文本文件:读取内容;记录 source path 以便重复扫描去重
|
||||||
String content = Files.readString(file, StandardCharsets.UTF_8);
|
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 {
|
} else {
|
||||||
// 二进制文件:直接引用原始路径,不复制
|
// 二进制文件:直接引用原始路径,不复制
|
||||||
String sourceType = switch (ext) {
|
String sourceType = switch (ext) {
|
||||||
|
|||||||
@ -86,6 +86,20 @@ public class WikiRawMaterialService {
|
|||||||
.eq(WikiRawMaterialEntity::getSourcePath, sourcePath));
|
.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<WikiRawMaterialEntity>()
|
||||||
|
.eq(WikiRawMaterialEntity::getId, rawId)
|
||||||
|
.set(WikiRawMaterialEntity::getSourcePath, sourcePath));
|
||||||
|
}
|
||||||
|
|
||||||
public List<WikiRawMaterialEntity> listPending(Long kbId) {
|
public List<WikiRawMaterialEntity> listPending(Long kbId) {
|
||||||
return rawMapper.selectList(
|
return rawMapper.selectList(
|
||||||
new LambdaQueryWrapper<WikiRawMaterialEntity>()
|
new LambdaQueryWrapper<WikiRawMaterialEntity>()
|
||||||
|
|||||||
@ -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.
|
||||||
|
*
|
||||||
|
* <p>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;
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -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);
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue
Block a user