[OPIK-2597] [BE] Separate thread related functionality to a dedicated service and DAO (#4465)

* Separate thread related functionality to a dedicated service and DAO

* fix DI

Author: BorisTkachenko

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/priv/TracesResource.java

@@ -37,6 +37,7 @@
 import com.comet.opik.domain.FeedbackScoreService;
 import com.comet.opik.domain.ProjectService;
 import com.comet.opik.domain.Streamer;
+import com.comet.opik.domain.ThreadService;
 import com.comet.opik.domain.TraceSearchCriteria;
 import com.comet.opik.domain.TraceService;
 import com.comet.opik.domain.threads.TraceThreadService;
@@ -105,6 +106,7 @@
 public class TracesResource {
 
     private final @NonNull TraceService service;
+    private final @NonNull ThreadService threadService;
     private final @NonNull FeedbackScoreService feedbackScoreService;
     private final @NonNull CommentService commentService;
     private final @NonNull FiltersFactory filtersFactory;
@@ -639,7 +641,7 @@ public Response getTraceThreads(
 
         log.info("Get trace threads by '{}' on workspaceId '{}'", searchCriteria, workspaceId);
 
-        TraceThreadPage traceThreadPage = service.getTraceThreads(page, size, searchCriteria)
+        TraceThreadPage traceThreadPage = threadService.find(page, size, searchCriteria)
                 .map(it -> {
                     // Remove sortableBy fields if dynamic sorting is disabled due to workspace size
                     if (metadata.cannotUseDynamicSorting()) {
@@ -690,7 +692,7 @@ public ChunkedOutput<JsonNode> searchTraceThreads(
                 .uuidToTime(instantToUUIDMapper.toUpperBound(request.toTime()))
                 .build();
 
-        Flux<TraceThread> items = service.threadsSearch(request.limit(), searchCriteria)
+        Flux<TraceThread> items = threadService.search(request.limit(), searchCriteria)
                 .contextWrite(ctx -> ctx.put(RequestContext.WORKSPACE_ID, workspaceId)
                         .put(RequestContext.USER_NAME, userName)
                         .put(RequestContext.VISIBILITY, Optional.ofNullable(visibility).orElse(Visibility.PRIVATE)));
@@ -716,7 +718,7 @@ public Response getTraceThread(
         log.info("Getting trace thread by id '{}' and project id '{}' on workspace_id '{}' with truncate '{}'",
                 identifier.threadId(), projectId, workspaceId, identifier.truncate());
 
-        TraceThread thread = service.getThreadById(projectId, identifier.threadId(), identifier.truncate())
+        TraceThread thread = threadService.getById(projectId, identifier.threadId(), identifier.truncate())
                 .contextWrite(ctx -> setRequestContext(ctx, requestContext))
                 .block();
 
@@ -877,7 +879,7 @@ public Response getThreadStats(@QueryParam("project_id") UUID projectId,
 
         log.info("Get trace thread stats by '{}' on workspaceId '{}'", searchCriteria, workspaceId);
 
-        ProjectStats projectStats = service.getThreadStats(searchCriteria)
+        ProjectStats projectStats = threadService.getStats(searchCriteria)
                 .contextWrite(ctx -> setRequestContext(ctx, requestContext))
                 .block();
 

apps/opik-backend/src/main/java/com/comet/opik/domain/ThreadDAO.java

@@ -0,0 +1,75 @@
+package com.comet.opik.domain;
+
+import com.comet.opik.api.ProjectStats;
+import com.comet.opik.api.TraceThread;
+import com.comet.opik.api.sorting.TraceThreadSortingFactory;
+import com.google.inject.ImplementedBy;
+import io.opentelemetry.instrumentation.annotations.WithSpan;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.UUID;
+
+import static com.comet.opik.utils.ErrorUtils.failWithNotFound;
+
+@ImplementedBy(ThreadServiceImpl.class)
+public interface ThreadService {
+
+    Mono<TraceThread.TraceThreadPage> find(int page, int size, TraceSearchCriteria criteria);
+
+    Mono<TraceThread> getById(UUID projectId, String threadId, boolean truncate);
+
+    Flux<TraceThread> search(int limit, @NonNull TraceSearchCriteria criteria);
+
+    Mono<ProjectStats> getStats(TraceSearchCriteria searchCriteria);
+}
+
+@Slf4j
+@Singleton
+@RequiredArgsConstructor(onConstructor_ = @Inject)
+class ThreadServiceImpl implements ThreadService {
+
+    private final @NonNull ThreadDAO dao;
+    private final @NonNull ProjectService projectService;
+    private final @NonNull TraceThreadSortingFactory traceThreadSortingFactory;
+
+    @Override
+    public Mono<TraceThread.TraceThreadPage> find(int page, int size, @NonNull TraceSearchCriteria criteria) {
+        return findProjectAndVerifyVisibility(criteria)
+                .flatMap(it -> dao.find(size, page, it))
+                .switchIfEmpty(Mono
+                        .just(TraceThread.TraceThreadPage.empty(page, traceThreadSortingFactory.getSortableFields())));
+    }
+
+    @Override
+    public Mono<TraceThread> getById(@NonNull UUID projectId, @NonNull String threadId, boolean truncate) {
+        return dao.findById(projectId, threadId, truncate)
+                .switchIfEmpty(Mono.defer(() -> Mono.error(failWithNotFound("Trace Thread", threadId))));
+    }
+
+    @Override
+    public Flux<TraceThread> search(int limit, @NonNull TraceSearchCriteria criteria) {
+        return findProjectAndVerifyVisibility(criteria)
+                .flatMapMany(it -> dao.search(limit, it));
+    }
+
+    @Override
+    @WithSpan
+    public Mono<ProjectStats> getStats(@NonNull TraceSearchCriteria criteria) {
+        return findProjectAndVerifyVisibility(criteria)
+                .flatMap(dao::getThreadStats)
+                .switchIfEmpty(Mono.just(ProjectStats.empty()));
+    }
+
+    private Mono<TraceSearchCriteria> findProjectAndVerifyVisibility(TraceSearchCriteria criteria) {
+        return projectService.resolveProjectIdAndVerifyVisibility(criteria.projectId(), criteria.projectName())
+                .map(projectId -> criteria.toBuilder()
+                        .projectId(projectId)
+                        .build());
+    }
+}