add session-gated /sync endpoint and prune stale vectors after reindex
SyncController/SyncService/SyncProvider mirror the existing search dispatch pattern: a session-gated controller routes a SyncRequest to the provider registered for its ServiceType. HomeboxSyncProvider pages through the full Homebox catalog, embedding each page via EmbeddingService, and now collects every item ID it sees across all pages before pruning anything no longer present upstream in one deleteStaleVectorEntries call at the end - deleting per page would treat items on other pages as stale and wipe them out.
This commit is contained in:
@@ -1,27 +1,57 @@
|
||||
package com.vaessl.app.sync;
|
||||
|
||||
import org.springframework.stereotype.Service;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import org.springframework.data.domain.Page;
|
||||
import org.springframework.data.domain.PageRequest;
|
||||
import org.springframework.data.domain.Pageable;
|
||||
import org.springframework.stereotype.Component;
|
||||
import com.vaessl.app.homebox.HomeboxItemClient;
|
||||
import com.vaessl.app.homebox.HomeboxItemClient.HomeboxItemPage;
|
||||
import com.vaessl.app.shared.ServiceItem;
|
||||
import com.vaessl.app.shared.ServiceType;
|
||||
import com.vaessl.app.vector.EmbeddingService;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
|
||||
@Service
|
||||
/**
|
||||
* Pages through the full Homebox catalog and re-indexes it into the vector store.
|
||||
* Item IDs are collected across all pages before {@link EmbeddingService#deleteStaleVectorEntries}
|
||||
* is called once at the end, rather than after each page: deleting per page would treat every
|
||||
* item outside the current page as stale, wiping out entries from pages already synced.
|
||||
*/
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class HomeboxSyncProvider implements SyncProvider {
|
||||
|
||||
private final EmbeddingService embeddingService;
|
||||
|
||||
public HomeboxSyncProvider(EmbeddingService embeddingService) {
|
||||
this.embeddingService = embeddingService;
|
||||
}
|
||||
|
||||
private final HomeboxItemClient client;
|
||||
|
||||
@Override
|
||||
public ServiceType getServiceType() {
|
||||
return ServiceType.HOMEBOX;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void syncVectorStore() {
|
||||
// TODO Auto-generated method stub
|
||||
throw new UnsupportedOperationException("Unimplemented method 'syncVectorStore'");
|
||||
}
|
||||
public void syncVectorStore(SyncRequest request) {
|
||||
|
||||
int batchSize = 100;
|
||||
int page = 0;
|
||||
Page<ServiceItem> current;
|
||||
Long connectionId = null;
|
||||
List<String> currentItemIds = new ArrayList<>();
|
||||
do {
|
||||
Pageable pageable = PageRequest.of(page, batchSize);
|
||||
HomeboxItemPage result = client.hbResponse(request, null, pageable);
|
||||
current = result.page();
|
||||
if (connectionId == null) {
|
||||
connectionId = result.connectionId();
|
||||
}
|
||||
embeddingService.vectorizeData(current.getContent(), request.serviceType(),
|
||||
result.connectionId(), currentItemIds);
|
||||
page++;
|
||||
} while (page < current.getTotalPages());
|
||||
|
||||
embeddingService.deleteStaleVectorEntries(connectionId, currentItemIds);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user