Add fakeable Qdrant Effect services

ts/packages/flow/src/query/embeddings/qdrant-doc-service.ts

@@ -11,8 +11,10 @@ import {
   makeFlowProcessor,
   makeConsumerSpec,
   makeProducerSpec,
+  processorLifecycleError,
   type ProcessorConfig,
   type FlowProcessorRuntime,
+  type FlowProcessorStartEffect,
   type FlowContext,
   type FlowResourceNotFoundError,
   type MessagingDeliveryError,
@@ -26,8 +28,9 @@ import { Effect, Layer, ManagedRuntime } from "effect";
 import {
   QdrantDocEmbeddingsQueryLive,
   QdrantDocEmbeddingsQueryService,
-  makeQdrantDocEmbeddingsQueryService,
+  makeQdrantDocEmbeddingsQueryServiceEffect,
   type QdrantDocQueryConfig,
+  type QdrantDocEmbeddingsQueryError,
 } from "./qdrant-doc.js";
 
 const DocumentEmbeddingsResponseProducer = makeProducerSpec<DocumentEmbeddingsResponse>("document-embeddings-response");
@@ -92,25 +95,37 @@ export const makeDocEmbeddingsQuerySpecs = (): ReadonlyArray<Spec<QdrantDocEmbed
 
 export type DocEmbeddingsQueryService = FlowProcessorRuntime<QdrantDocEmbeddingsQueryService>;
 
+const provideQdrantDocEmbeddingsQuery = (processorId: string) =>
+  Effect.fn("DocEmbeddingsQueryService.provideQdrant")(function* (
+    effect: FlowProcessorStartEffect<QdrantDocEmbeddingsQueryService>,
+  ) {
+    const query = yield* makeQdrantDocEmbeddingsQueryServiceEffect().pipe(
+      Effect.mapError((error) => processorLifecycleError(processorId, "qdrant-doc-query-connect", error)),
+    );
+    yield* effect.pipe(
+      Effect.provideService(
+        QdrantDocEmbeddingsQueryService,
+        QdrantDocEmbeddingsQueryService.of(query),
+      ),
+    );
+  });
+
 export function makeDocEmbeddingsQueryService(config: ProcessorConfig): DocEmbeddingsQueryService {
-  const query = makeQdrantDocEmbeddingsQueryService();
   const service = makeFlowProcessor(config, {
     specifications: makeDocEmbeddingsQuerySpecs(),
-    provide: (effect) =>
-      effect.pipe(
-        Effect.provideService(
-          QdrantDocEmbeddingsQueryService,
-          QdrantDocEmbeddingsQueryService.of(query),
-        ),
-      ),
+    provide: provideQdrantDocEmbeddingsQuery(config.id),
   });
-  Effect.runSync(Effect.log("[DocEmbeddingsQuery] Service initialized"));
+  void Effect.runPromise(Effect.log("[DocEmbeddingsQuery] Service initialized"));
   return service;
 }
 
 export const DocEmbeddingsQueryService = makeDocEmbeddingsQueryService;
 
-export const program = makeFlowProcessorProgram<ProcessorConfig & QdrantDocQueryConfig, never, QdrantDocEmbeddingsQueryService>({
+export const program = makeFlowProcessorProgram<
+  ProcessorConfig & QdrantDocQueryConfig,
+  QdrantDocEmbeddingsQueryError,
+  QdrantDocEmbeddingsQueryService
+>({
   id: "doc-embeddings-query",
   specs: () => makeDocEmbeddingsQuerySpecs(),
   layer: (config) => QdrantDocEmbeddingsQueryLive(config),

ts/packages/flow/src/query/embeddings/qdrant-doc.ts

@@ -7,15 +7,16 @@
  * Python reference: trustgraph-flow/trustgraph/query/doc_embeddings/qdrant/service.py
  */
 
-import { QdrantClient } from "@qdrant/js-client-rest";
 import { errorMessage } from "@trustgraph/base";
 import { Config, Context, Effect, Layer } from "effect";
 import * as O from "effect/Option";
 import * as S from "effect/Schema";
+import { makeQdrantClient, type QdrantClientFactory, type QdrantClientLike } from "../../qdrant/client.js";
 
 export interface QdrantDocQueryConfig {
   url?: string;
   apiKey?: string;
+  clientFactory?: QdrantClientFactory;
 }
 
 export interface ChunkMatch {
@@ -63,25 +64,37 @@ const loadQdrantDocQueryConfig = Effect.fn("QdrantDocEmbeddingsQuery.loadConfig"
   } satisfies ResolvedQdrantDocQueryConfig;
 });
 
+const DocPointPayloadSchema = S.Struct({
+  chunk_id: S.String,
+  content: S.optionalKey(S.String),
+});
+
+const decodeDocPointPayload = (payload: unknown) =>
+  S.decodeUnknownEffect(DocPointPayloadSchema)(payload).pipe(Effect.option);
+
 export interface QdrantDocEmbeddingsQuery {
-  readonly query: (request: DocEmbeddingsQueryRequest) => Promise<ChunkMatch[]>;
+  readonly query: (request: DocEmbeddingsQueryRequest) => Promise<ReadonlyArray<ChunkMatch>>;
   readonly queryEffect: (
     request: DocEmbeddingsQueryRequest,
   ) => Effect.Effect<ReadonlyArray<ChunkMatch>, QdrantDocEmbeddingsQueryError>;
 }
 
-export function makeQdrantDocEmbeddingsQuery(
-  config: QdrantDocQueryConfig = {},
-): QdrantDocEmbeddingsQuery {
-  const resolved = Effect.runSync(loadQdrantDocQueryConfig(config));
-
-  const client = new QdrantClient({
-    url: resolved.url,
-    ...(resolved.apiKey !== undefined ? { apiKey: resolved.apiKey } : {}),
+const makeQdrantDocEmbeddingsQueryClient = (
+  config: QdrantDocQueryConfig,
+  resolved: ResolvedQdrantDocQueryConfig,
+) =>
+  Effect.try({
+    try: () =>
+      makeQdrantClient(config.clientFactory, {
+        url: resolved.url,
+        ...(resolved.apiKey !== undefined ? { apiKey: resolved.apiKey } : {}),
+      }),
+    catch: (cause) => qdrantDocEmbeddingsQueryError("create-client", cause),
   });
 
-  Effect.runSync(Effect.log("[QdrantDocQuery] Query service initialized"));
-
+const makeQdrantDocEmbeddingsQueryFromClient = (
+  client: QdrantClientLike,
+): QdrantDocEmbeddingsQueryServiceShape => {
   const queryEffect = Effect.fn("QdrantDocEmbeddingsQuery.query")(function* (request: DocEmbeddingsQueryRequest) {
     const { vector, user, collection, limit } = request;
 
@@ -116,20 +129,52 @@ export function makeQdrantDocEmbeddingsQuery(
 
     const chunks: ChunkMatch[] = [];
     for (const point of searchResult) {
-      const payload = point.payload as Record<string, unknown> | undefined;
-      const chunkId = payload?.chunk_id as string | undefined;
-      if (chunkId !== undefined && chunkId.length > 0) {
-        chunks.push({
-          chunkId,
-          score: point.score,
-          ...(typeof payload?.content === "string" ? { content: payload.content } : {}),
-        });
-      }
+      const payload = yield* decodeDocPointPayload(point.payload);
+      if (O.isNone(payload)) continue;
+
+      const chunkId = payload.value.chunk_id;
+      if (chunkId.length === 0) continue;
+
+      chunks.push({
+        chunkId,
+        score: point.score,
+        ...(payload.value.content !== undefined ? { content: payload.value.content } : {}),
+      });
     }
 
     return chunks;
   });
 
+  return {
+    query: queryEffect,
+  };
+};
+
+export const makeQdrantDocEmbeddingsQueryServiceEffect = Effect.fn(
+  "makeQdrantDocEmbeddingsQueryServiceEffect",
+)(function* (config: QdrantDocQueryConfig = {}) {
+  const resolved = yield* loadQdrantDocQueryConfig(config).pipe(
+    Effect.mapError((cause) => qdrantDocEmbeddingsQueryError("load-config", cause)),
+  );
+  const client = yield* makeQdrantDocEmbeddingsQueryClient(config, resolved);
+  yield* Effect.log("[QdrantDocQuery] Query service initialized");
+  return makeQdrantDocEmbeddingsQueryFromClient(client);
+});
+
+const withQdrantDocEmbeddingsQuery = <A>(
+  config: QdrantDocQueryConfig,
+  use: (query: QdrantDocEmbeddingsQueryServiceShape) => Effect.Effect<A, QdrantDocEmbeddingsQueryError>,
+) =>
+  makeQdrantDocEmbeddingsQueryServiceEffect(config).pipe(
+    Effect.flatMap(use),
+  );
+
+export function makeQdrantDocEmbeddingsQuery(
+  config: QdrantDocQueryConfig = {},
+): QdrantDocEmbeddingsQuery {
+  const queryEffect = (request: DocEmbeddingsQueryRequest) =>
+    withQdrantDocEmbeddingsQuery(config, (query) => query.query(request));
+
   return {
     query: (request) => Effect.runPromise(queryEffect(request)),
     queryEffect,
@@ -151,17 +196,16 @@ export class QdrantDocEmbeddingsQueryService extends Context.Service<
 
 export const makeQdrantDocEmbeddingsQueryService = (
   config: QdrantDocQueryConfig = {},
-): QdrantDocEmbeddingsQueryServiceShape => {
-  const query = makeQdrantDocEmbeddingsQuery(config);
-  return {
-    query: query.queryEffect,
-  };
-};
+): QdrantDocEmbeddingsQueryServiceShape => ({
+  query: (request) => withQdrantDocEmbeddingsQuery(config, (query) => query.query(request)),
+});
 
 export const QdrantDocEmbeddingsQueryLive = (
   config: QdrantDocQueryConfig = {},
-): Layer.Layer<QdrantDocEmbeddingsQueryService> =>
-  Layer.succeed(
+): Layer.Layer<QdrantDocEmbeddingsQueryService, QdrantDocEmbeddingsQueryError> =>
+  Layer.effect(
     QdrantDocEmbeddingsQueryService,
-    QdrantDocEmbeddingsQueryService.of(makeQdrantDocEmbeddingsQueryService(config)),
+    makeQdrantDocEmbeddingsQueryServiceEffect(config).pipe(
+      Effect.map((service) => QdrantDocEmbeddingsQueryService.of(service)),
+    ),
   );

ts/packages/flow/src/query/embeddings/qdrant-graph-service.ts

@@ -11,8 +11,10 @@ import {
   makeFlowProcessor,
   makeConsumerSpec,
   makeProducerSpec,
+  processorLifecycleError,
   type ProcessorConfig,
   type FlowProcessorRuntime,
+  type FlowProcessorStartEffect,
   type FlowContext,
   type FlowResourceNotFoundError,
   type MessagingDeliveryError,
@@ -26,8 +28,9 @@ import { Effect, Layer, ManagedRuntime } from "effect";
 import {
   QdrantGraphEmbeddingsQueryLive,
   QdrantGraphEmbeddingsQueryService,
-  makeQdrantGraphEmbeddingsQueryService,
+  makeQdrantGraphEmbeddingsQueryServiceEffect,
   type QdrantGraphQueryConfig,
+  type QdrantGraphEmbeddingsQueryError,
 } from "./qdrant-graph.js";
 
 const GraphEmbeddingsResponseProducer = makeProducerSpec<GraphEmbeddingsResponse>("graph-embeddings-response");
@@ -93,25 +96,37 @@ export const makeGraphEmbeddingsQuerySpecs = (): ReadonlyArray<Spec<QdrantGraphE
 
 export type GraphEmbeddingsQueryService = FlowProcessorRuntime<QdrantGraphEmbeddingsQueryService>;
 
+const provideQdrantGraphEmbeddingsQuery = (processorId: string) =>
+  Effect.fn("GraphEmbeddingsQueryService.provideQdrant")(function* (
+    effect: FlowProcessorStartEffect<QdrantGraphEmbeddingsQueryService>,
+  ) {
+    const query = yield* makeQdrantGraphEmbeddingsQueryServiceEffect().pipe(
+      Effect.mapError((error) => processorLifecycleError(processorId, "qdrant-graph-query-connect", error)),
+    );
+    yield* effect.pipe(
+      Effect.provideService(
+        QdrantGraphEmbeddingsQueryService,
+        QdrantGraphEmbeddingsQueryService.of(query),
+      ),
+    );
+  });
+
 export function makeGraphEmbeddingsQueryService(config: ProcessorConfig): GraphEmbeddingsQueryService {
-  const query = makeQdrantGraphEmbeddingsQueryService();
   const service = makeFlowProcessor(config, {
     specifications: makeGraphEmbeddingsQuerySpecs(),
-    provide: (effect) =>
-      effect.pipe(
-        Effect.provideService(
-          QdrantGraphEmbeddingsQueryService,
-          QdrantGraphEmbeddingsQueryService.of(query),
-        ),
-      ),
+    provide: provideQdrantGraphEmbeddingsQuery(config.id),
   });
-  Effect.runSync(Effect.log("[GraphEmbeddingsQuery] Service initialized"));
+  void Effect.runPromise(Effect.log("[GraphEmbeddingsQuery] Service initialized"));
   return service;
 }
 
 export const GraphEmbeddingsQueryService = makeGraphEmbeddingsQueryService;
 
-export const program = makeFlowProcessorProgram<ProcessorConfig & QdrantGraphQueryConfig, never, QdrantGraphEmbeddingsQueryService>({
+export const program = makeFlowProcessorProgram<
+  ProcessorConfig & QdrantGraphQueryConfig,
+  QdrantGraphEmbeddingsQueryError,
+  QdrantGraphEmbeddingsQueryService
+>({
   id: "graph-embeddings-query",
   specs: () => makeGraphEmbeddingsQuerySpecs(),
   layer: (config) => QdrantGraphEmbeddingsQueryLive(config),

ts/packages/flow/src/query/embeddings/qdrant-graph.ts

@@ -10,15 +10,16 @@
  * Python reference: trustgraph-flow/trustgraph/query/graph_embeddings/qdrant/service.py
  */
 
-import { QdrantClient } from "@qdrant/js-client-rest";
 import { errorMessage, type Term } from "@trustgraph/base";
 import { Config, Context, Effect, Layer } from "effect";
 import * as O from "effect/Option";
 import * as S from "effect/Schema";
+import { makeQdrantClient, type QdrantClientFactory, type QdrantClientLike } from "../../qdrant/client.js";
 
 export interface QdrantGraphQueryConfig {
   url?: string;
   apiKey?: string;
+  clientFactory?: QdrantClientFactory;
 }
 
 export interface EntityMatch {
@@ -72,24 +73,36 @@ function createTerm(value: string): Term {
   return { type: "LITERAL", value };
 }
 
+const GraphPointPayloadSchema = S.Struct({
+  entity: S.String,
+});
+
+const decodeGraphPointPayload = (payload: unknown) =>
+  S.decodeUnknownEffect(GraphPointPayloadSchema)(payload).pipe(Effect.option);
+
 export interface QdrantGraphEmbeddingsQuery {
-  readonly query: (request: GraphEmbeddingsQueryRequest) => Promise<EntityMatch[]>;
+  readonly query: (request: GraphEmbeddingsQueryRequest) => Promise<ReadonlyArray<EntityMatch>>;
   readonly queryEffect: (
     request: GraphEmbeddingsQueryRequest,
   ) => Effect.Effect<ReadonlyArray<EntityMatch>, QdrantGraphEmbeddingsQueryError>;
 }
 
-export function makeQdrantGraphEmbeddingsQuery(
-  config: QdrantGraphQueryConfig = {},
-): QdrantGraphEmbeddingsQuery {
-  const resolved = Effect.runSync(loadQdrantGraphQueryConfig(config));
-
-  const client = new QdrantClient({
-    url: resolved.url,
-    ...(resolved.apiKey !== undefined ? { apiKey: resolved.apiKey } : {}),
+const makeQdrantGraphEmbeddingsQueryClient = (
+  config: QdrantGraphQueryConfig,
+  resolved: ResolvedQdrantGraphQueryConfig,
+) =>
+  Effect.try({
+    try: () =>
+      makeQdrantClient(config.clientFactory, {
+        url: resolved.url,
+        ...(resolved.apiKey !== undefined ? { apiKey: resolved.apiKey } : {}),
+      }),
+    catch: (cause) => qdrantGraphEmbeddingsQueryError("create-client", cause),
   });
 
-  Effect.runSync(Effect.log("[QdrantGraphQuery] Query service initialized"));
+const makeQdrantGraphEmbeddingsQueryFromClient = (
+  client: QdrantClientLike,
+): QdrantGraphEmbeddingsQueryServiceShape => {
 
   const queryEffect = Effect.fn("QdrantGraphEmbeddingsQuery.query")(function* (
     request: GraphEmbeddingsQueryRequest,
@@ -131,8 +144,10 @@ export function makeQdrantGraphEmbeddingsQuery(
     const entities: EntityMatch[] = [];
 
     for (const point of searchResult) {
-      const payload = point.payload as Record<string, unknown> | undefined;
-      const entityValue = payload?.entity as string | undefined;
+      const payload = yield* decodeGraphPointPayload(point.payload);
+      if (O.isNone(payload)) continue;
+
+      const entityValue = payload.value.entity;
       if (entityValue === undefined || entityValue.length === 0) continue;
 
       // Deduplicate by entity value, keeping the highest score (results are
@@ -152,6 +167,36 @@ export function makeQdrantGraphEmbeddingsQuery(
     return entities;
   });
 
+  return {
+    query: queryEffect,
+  };
+};
+
+export const makeQdrantGraphEmbeddingsQueryServiceEffect = Effect.fn(
+  "makeQdrantGraphEmbeddingsQueryServiceEffect",
+)(function* (config: QdrantGraphQueryConfig = {}) {
+  const resolved = yield* loadQdrantGraphQueryConfig(config).pipe(
+    Effect.mapError((cause) => qdrantGraphEmbeddingsQueryError("load-config", cause)),
+  );
+  const client = yield* makeQdrantGraphEmbeddingsQueryClient(config, resolved);
+  yield* Effect.log("[QdrantGraphQuery] Query service initialized");
+  return makeQdrantGraphEmbeddingsQueryFromClient(client);
+});
+
+const withQdrantGraphEmbeddingsQuery = <A>(
+  config: QdrantGraphQueryConfig,
+  use: (query: QdrantGraphEmbeddingsQueryServiceShape) => Effect.Effect<A, QdrantGraphEmbeddingsQueryError>,
+) =>
+  makeQdrantGraphEmbeddingsQueryServiceEffect(config).pipe(
+    Effect.flatMap(use),
+  );
+
+export function makeQdrantGraphEmbeddingsQuery(
+  config: QdrantGraphQueryConfig = {},
+): QdrantGraphEmbeddingsQuery {
+  const queryEffect = (request: GraphEmbeddingsQueryRequest) =>
+    withQdrantGraphEmbeddingsQuery(config, (query) => query.query(request));
+
   return {
     query: (request) => Effect.runPromise(queryEffect(request)),
     queryEffect,
@@ -173,17 +218,16 @@ export class QdrantGraphEmbeddingsQueryService extends Context.Service<
 
 export const makeQdrantGraphEmbeddingsQueryService = (
   config: QdrantGraphQueryConfig = {},
-): QdrantGraphEmbeddingsQueryServiceShape => {
-  const query = makeQdrantGraphEmbeddingsQuery(config);
-  return {
-    query: query.queryEffect,
-  };
-};
+): QdrantGraphEmbeddingsQueryServiceShape => ({
+  query: (request) => withQdrantGraphEmbeddingsQuery(config, (query) => query.query(request)),
+});
 
 export const QdrantGraphEmbeddingsQueryLive = (
   config: QdrantGraphQueryConfig = {},
-): Layer.Layer<QdrantGraphEmbeddingsQueryService> =>
-  Layer.succeed(
+): Layer.Layer<QdrantGraphEmbeddingsQueryService, QdrantGraphEmbeddingsQueryError> =>
+  Layer.effect(
     QdrantGraphEmbeddingsQueryService,
-    QdrantGraphEmbeddingsQueryService.of(makeQdrantGraphEmbeddingsQueryService(config)),
+    makeQdrantGraphEmbeddingsQueryServiceEffect(config).pipe(
+      Effect.map((service) => QdrantGraphEmbeddingsQueryService.of(service)),
+    ),
   );