feat(wiki): fire count-threshold pipeline triggers after ingest

This commit is contained in:
matevip 2026-05-31 07:56:58 +08:00
parent 526a361488
commit ad1f5b4a15
4 changed files with 90 additions and 0 deletions

View File

@ -0,0 +1,11 @@
package vip.mate.wiki.event;
/**
* Published after a wiki page is created during ingest (already committed,
* since page creation is its own transaction). Consumed asynchronously to
* evaluate count-threshold pipeline triggers without blocking ingest.
*
* @author MateClaw Team
*/
public record WikiPageCreatedEvent(Long kbId, String pageType) {
}

View File

@ -0,0 +1,38 @@
package vip.mate.wiki.pipeline;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import vip.mate.wiki.event.WikiPageCreatedEvent;
/**
* Evaluates count-threshold pipeline triggers asynchronously when a page is
* created during ingest. Runs off the ingest thread so a fired pipeline (which
* may call a model) never blocks ingest; the page is already committed when the
* event fires, so the count is accurate, and the run dedup key keeps repeated
* evaluation idempotent.
*
* @author MateClaw Team
*/
@Slf4j
@Component
public class WikiPipelineTriggerListener {
private final WikiPipelineTriggerService triggerService;
public WikiPipelineTriggerListener(WikiPipelineTriggerService triggerService) {
this.triggerService = triggerService;
}
@Async
@EventListener
public void onPageCreated(WikiPageCreatedEvent event) {
try {
triggerService.onPageTypeCount(event.kbId(), event.pageType());
} catch (Exception e) {
log.warn("[WikiPipeline] trigger evaluation failed for kb={} pageType={}: {}",
event.kbId(), event.pageType(), e.getMessage());
}
}
}

View File

@ -1454,6 +1454,11 @@ public class WikiProcessingService {
pageService.mergeSourceLineage(created.getId(), rawId, raw.getTitle());
log.info("[Wiki] Phase B create page slug='{}' done (created)", slug);
citationService.buildCitationsAsync(created.getId(), kbId);
// Evaluate count-threshold pipeline triggers off-thread (page is now
// committed, so the count is accurate); no-op when no pipelines match.
if (eventPublisher != null && pageType != null && !pageType.isBlank()) {
eventPublisher.publishEvent(new vip.mate.wiki.event.WikiPageCreatedEvent(kbId, pageType));
}
return true;
} catch (org.springframework.dao.DuplicateKeyException e) {
// Fallback 2: concurrent INSERT race degrade to update

View File

@ -0,0 +1,36 @@
package vip.mate.wiki.pipeline;
import org.junit.jupiter.api.Test;
import vip.mate.wiki.event.WikiPageCreatedEvent;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* Unit tests for {@link WikiPipelineTriggerListener}: it forwards the event to
* the trigger service and never lets a trigger failure escape (so a broken
* pipeline cannot disturb ingest).
*/
class WikiPipelineTriggerListenerTest {
@Test
void forwardsEventToTriggerService() {
WikiPipelineTriggerService trigger = mock(WikiPipelineTriggerService.class);
when(trigger.onPageTypeCount(7L, "episode")).thenReturn(1);
new WikiPipelineTriggerListener(trigger).onPageCreated(new WikiPageCreatedEvent(7L, "episode"));
verify(trigger).onPageTypeCount(7L, "episode");
}
@Test
void swallowsTriggerFailure() {
WikiPipelineTriggerService trigger = mock(WikiPipelineTriggerService.class);
doThrow(new RuntimeException("boom")).when(trigger).onPageTypeCount(7L, "episode");
// Must not throw ingest must be unaffected by a pipeline failure.
new WikiPipelineTriggerListener(trigger).onPageCreated(new WikiPageCreatedEvent(7L, "episode"));
}
}