feat(data-access): validate semantic search statuses against Opportunity.STATUSES (LLMO-7445)

Author: Tanase Butcaru

docs/adr/0002-embedding-client-and-semantic-index-utils.md

@@ -49,6 +49,8 @@ Following ADR-0001's test (*true for every consumer → shared; specific to one
 
 **Type registries.** `OPPORTUNITY_SEMANTIC_SOURCE_TYPES` (kinds of embedded text: `topic`, later `claim`) and `OPPORTUNITY_SEMANTIC_ENTITY_TYPES` (opportunity types, from `OPPORTUNITY_TYPES`) are exported from this layer, and this library's writer stores and reader accepts only their values; anything else is a `ValidationError`. The reader takes `sourceTypes` (required, non-empty) and optional `entityTypes` (omitted or empty = all), deduped, so a list can never exceed its registry. The DB does not duplicate the registries: the RPC only rejects empty, multi-dimensional, or NULL-element lists and lists over 100 elements (a length bound, not a registry mirror; `NULL` entity types = all; the reader sends `NULL` when `entityTypes` is omitted or empty). A team that wants its opportunities searchable adds its source kind and/or opportunity type to the registries, indexes with them, and callers opt in by passing them.
 
+**Status filter.** The reader also takes optional `statuses`, checked against `Opportunity.STATUSES` and deduped like the type lists. The RPC applies it before the per-query limit, so filtered-out opportunities never take result slots. Which statuses count as visible is the caller's choice. The data-service casts the list to its status enum, so a value it doesn't know is an error, not an empty result. Sending `statuses` needs a data-service with `p_statuses` (mysticat-data-service PR 1132).
+
 Adding an entity type: (1) if the opportunity type is new, add it to `OPPORTUNITY_TYPES` and release `spacecat-shared-utils`; (2) add it to `OPPORTUNITY_SEMANTIC_ENTITY_TYPES`, bump data-access's utils pin, and release data-access; (3) bump data-access in the writer (audit-worker) and the reader (api-service). The writer rejects the type until step 3, so index only after it. A new source kind is the same minus step 1.
 
 ## Consequences

packages/spacecat-shared-data-access/README.md

@@ -171,7 +171,8 @@ await touchQueryEmbeddings(postgrestClient, { textHashes, ...scope });
 
 // reader, batched: one best-first list of { entityId, entityType, score } per query vector.
 // entityTypes is optional (omitted or [] searches all entity types). statuses is optional too
-// (omitted or [] searches all); it is applied before k, so hidden opportunities can't take the slots.
+// (omitted or [] searches all; values from Opportunity.STATUSES); it is applied before k, so
+// filtered-out opportunities never take result slots. Needs a data-service with p_statuses.
 const matches = await lookupOpportunitiesByVectors(postgrestClient, {
   siteId, sourceTypes: [TOPIC], entityTypes: [CITED_ANALYSIS], statuses: ['NEW', 'IN_PROGRESS'],
   vectors, k: 10, minScore: 0.5, ...scope,

packages/spacecat-shared-data-access/src/service/index.js

@@ -50,6 +50,30 @@ const fetch = createFetchCompat(postgrestFetch);
 export * from '../errors/index.js';
 export * from '../models/index.js';
 export * from '../util/index.js';
+// Not in util/index.js: semantic-index imports the Opportunity model, which imports util/index.js.
+export {
+  SEMANTIC_INDEX_TABLES,
+  QUERY_EMBEDDING_TABLE,
+  SEMANTIC_SEARCH_RPC,
+  COPY_VECTORS_RPC,
+  SEMANTIC_CHUNK_SIZE,
+  QUERY_HASH_CHUNK_SIZE,
+  MAX_SOURCE_TEXT_LENGTH,
+  OPPORTUNITY_SEMANTIC_SOURCE_TYPES,
+  OPPORTUNITY_SEMANTIC_ENTITY_TYPES,
+  SEMANTIC_MATCHING_CONFIG,
+  normalizeText,
+  hashText,
+  cleanTopicText,
+  serializeVector,
+  parseVector,
+  syncOpportunitySemantic,
+  copyEntityVectors,
+  lookupOpportunitiesByVectors,
+  getQueryEmbeddings,
+  upsertQueryEmbeddings,
+  touchQueryEmbeddings,
+} from '../util/semantic-index.utils.js';
 
 const createPostgrestService = (config, client = undefined) => {
   if (client) {

packages/spacecat-shared-data-access/src/util/index.js

@@ -36,30 +36,6 @@ export {
   lookupEntityIdsByUrl,
 } from './url-index.utils.js';
 
-export {
-  SEMANTIC_INDEX_TABLES,
-  QUERY_EMBEDDING_TABLE,
-  SEMANTIC_SEARCH_RPC,
-  COPY_VECTORS_RPC,
-  SEMANTIC_CHUNK_SIZE,
-  QUERY_HASH_CHUNK_SIZE,
-  MAX_SOURCE_TEXT_LENGTH,
-  OPPORTUNITY_SEMANTIC_SOURCE_TYPES,
-  OPPORTUNITY_SEMANTIC_ENTITY_TYPES,
-  SEMANTIC_MATCHING_CONFIG,
-  normalizeText,
-  hashText,
-  cleanTopicText,
-  serializeVector,
-  parseVector,
-  syncOpportunitySemantic,
-  copyEntityVectors,
-  lookupOpportunitiesByVectors,
-  getQueryEmbeddings,
-  upsertQueryEmbeddings,
-  touchQueryEmbeddings,
-} from './semantic-index.utils.js';
-
 /**
  * Datastore types that collections can use to declare their storage backend.
  * @readonly

packages/spacecat-shared-data-access/src/util/semantic-index.utils.js

@@ -15,6 +15,7 @@ import { createHash } from 'crypto';
 import { OPPORTUNITY_TYPES } from '@adobe/spacecat-shared-utils';
 
 import { DataAccessError, ValidationError } from '../errors/index.js';
+import Opportunity from '../models/opportunity/opportunity.model.js';
 import { DEFAULT_PAGE_SIZE, rpcError } from './postgrest.utils.js';
 
 /**
@@ -163,31 +164,6 @@ function toAllowedList(values, name, registry, { required }) {
   return [...new Set(values)];
 }
 
-/**
- * Omitted or empty means no status filter. Not checked against `Opportunity.STATUSES` (importing
- * the model here is a dependency cycle); the data-service matches the enum, so a typo matches
- * nothing.
- */
-function toStatusList(statuses) {
-  if (statuses === undefined || statuses === null
-    || (Array.isArray(statuses) && statuses.length === 0)) {
-    return null;
-  }
-  if (!Array.isArray(statuses)) {
-    throw new ValidationError('statuses must be an array');
-  }
-  const rejected = [];
-  for (const status of statuses) {
-    if (typeof status !== 'string' || status.trim() === '') {
-      rejected.push(status);
-    }
-  }
-  if (rejected.length > 0) {
-    throw new ValidationError(`statuses must only contain non-empty strings (got ${describeValue(rejected)})`);
-  }
-  return [...new Set(statuses)];
-}
-
 /** Only an empty/omitted `sources` clears; a non-empty input reduced to nothing throws. */
 function assertClearable(sources, entityId, sourceType) {
   const explicitClear = sources === undefined || sources === null
@@ -495,7 +471,7 @@ export async function lookupOpportunitiesByVectors(postgrestClient, {
   assertId(siteId, 'siteId');
   const sourceTypeList = toAllowedList(sourceTypes, 'sourceTypes', OPPORTUNITY_SEMANTIC_SOURCE_TYPES, { required: true });
   const entityTypeList = toAllowedList(entityTypes, 'entityTypes', OPPORTUNITY_SEMANTIC_ENTITY_TYPES, { required: false });
-  const statusList = toStatusList(statuses);
+  const statusList = toAllowedList(statuses, 'statuses', Opportunity.STATUSES, { required: false });
   assertModel(model);
   assertDims(dims);
   if (!Array.isArray(vectors)) {
@@ -509,6 +485,12 @@ export async function lookupOpportunitiesByVectors(postgrestClient, {
   }
   const serialized = vectors.map((vector) => serializeDimsVector(vector, dims));
 
+  const details = {
+    siteId,
+    sourceTypes: sourceTypeList,
+    entityTypes: entityTypeList,
+    ...(statusList ? { statuses: statusList } : {}),
+  };
   const results = vectors.map(() => []);
   const groupSize = Math.min(SEMANTIC_CHUNK_SIZE, Math.floor(DEFAULT_PAGE_SIZE / k));
   for (let offset = 0; offset < serialized.length; offset += groupSize) {
@@ -523,29 +505,19 @@ export async function lookupOpportunitiesByVectors(postgrestClient, {
       p_dims: dims,
       p_limit: k,
       p_min_score: minScore,
-      // Only when set: a data-service without p_statuses rejects the unknown argument.
+      // Rollout shim: always send p_statuses (null if unset) once data-service #1132 is out.
       ...(statusList ? { p_statuses: statusList } : {}),
     });
     if (error) {
-      throw rpcError(
-        `Failed semantic search for site ${siteId}`,
-        SEMANTIC_SEARCH_RPC,
-        {
-          siteId,
-          sourceTypes: sourceTypeList,
-          entityTypes: entityTypeList,
-          ...(statusList ? { statuses: statusList } : {}),
-        },
-        error,
-      );
+      throw rpcError(`Failed semantic search for site ${siteId}`, SEMANTIC_SEARCH_RPC, details, error);
     }
     for (const row of data ?? []) {
       // Checked against this group, not all results, so a bad index can't land in another group.
       const { query_index: queryIndex } = row;
       if (!Number.isInteger(queryIndex) || queryIndex < 0 || queryIndex >= group.length) {
         throw new DataAccessError(
           `Unexpected query_index ${queryIndex} from ${SEMANTIC_SEARCH_RPC}`,
-          { siteId, sourceTypes: sourceTypeList, entityTypes: entityTypeList },
+          details,
         );
       }
       results[offset + queryIndex].push({

packages/spacecat-shared-data-access/test/unit/util/semantic-index.utils.test.js

@@ -685,17 +685,19 @@ describe('semantic-index.utils', () => {
       expect(client.calls.rpc[2].params).to.not.have.property('p_statuses');
     });
 
-    it('rejects statuses that are not a list of non-empty strings', async () => {
-      const call = (statuses) => lookupOpportunitiesByVectors(makeClient(), {
+    it('rejects statuses outside Opportunity.STATUSES without calling the RPC', async () => {
+      const client = makeClient();
+      const call = (statuses) => lookupOpportunitiesByVectors(client, {
         ...base, statuses, vectors: [[0.1, 0.2]],
       });
       await expect(call('NEW')).to.be.rejectedWith(ValidationError, 'statuses must be an array');
-      await expect(call(['NEW', ' '])).to.be.rejectedWith(
+      await expect(call(['NEW', 'new', ' NEW'])).to.be.rejectedWith(
         ValidationError,
-        'statuses must only contain non-empty strings (got [" "])',
+        'statuses must only contain: NEW, IN_PROGRESS, IGNORED, RESOLVED (got ["new"," NEW"])',
       );
       await expect(call(['NEW', 1])).to.be.rejectedWith(ValidationError, '(got [1])');
       await expect(call(new Array(1))).to.be.rejectedWith(ValidationError, 'statuses must only contain');
+      expect(client.calls.rpc).to.be.empty;
     });
 
     it('renders rejected values safely: unserializable falls back to String, long ones are truncated', async () => {
@@ -790,6 +792,11 @@ describe('semantic-index.utils', () => {
       expect(err.details).to.deep.equal({
         siteId: SITE_ID, sourceTypes: [SOURCE_TYPE], entityTypes: null,
       });
+      const withStatuses = await lookupOpportunitiesByVectors(client, {
+        ...base, statuses: ['NEW'], vectors: [[0.1, 0.2]],
+      }).catch((e) => e);
+      expect(withStatuses.message).to.equal(`Failed semantic search for site ${SITE_ID}: ${SEMANTIC_SEARCH_RPC} ${PGRST202_HINT}`);
+      expect(withStatuses.details.statuses).to.deep.equal(['NEW']);
     });
 
     it('rejects a query_index outside the current group (never spills into another group)', async () => {
@@ -802,6 +809,11 @@ describe('semantic-index.utils', () => {
         await expect(lookupOpportunitiesByVectors(client, { ...base, vectors: vecs(25) }))
           .to.be.rejectedWith(DataAccessError, `Unexpected query_index ${bad}`);
       }
+      const client = makeClient({ rpcResult: { data: [row(5)], error: null } });
+      const err = await lookupOpportunitiesByVectors(client, {
+        ...base, statuses: ['NEW'], vectors: [[0.1, 0.2]],
+      }).catch((e) => e);
+      expect(err.details.statuses).to.deep.equal(['NEW']);
     });
   });