diff --git a/src/search/index.ts b/src/search/index.ts new file mode 100644 index 0000000..61e7978 --- /dev/null +++ b/src/search/index.ts @@ -0,0 +1,26 @@ +/** + * Fuzzy search unit barrel export. + */ + +export { normalizeText, tokenize, tokenizeWithOffsets, type Token } from "./tokenize.ts"; +export { boundedLevenshtein, scoreToken, scoreField, type FieldScore } from "./score.ts"; +export { extractSnippet } from "./snippet.ts"; +export { + rankRecords, + DEFAULT_MIN_SCORE, + type MatchMode, + type FieldSpec, + type FieldMatch, + type Ranked, + type RankOptions, +} from "./rank.ts"; +export { + searchPipeline, + mapWithConcurrency, + isFatalSearchError, + DEFAULT_ENRICH_LIMIT, + DEFAULT_CONCURRENCY, + type PipelineSpec, + type PipelineEntry, + type PipelineResult, +} from "./pipeline.ts"; diff --git a/src/search/pipeline.ts b/src/search/pipeline.ts new file mode 100644 index 0000000..7520c27 --- /dev/null +++ b/src/search/pipeline.ts @@ -0,0 +1,241 @@ +/** + * Two-phase search orchestration. + * + * Phase 1 ranks candidates on data the list endpoint already returned, at no + * additional API cost. Phase 2 enriches a bounded subset with deep content + * and re-ranks. Final score is max(shallow, deep) so that a dead-on title + * match is never diluted by an unrelated body. + */ + +import { OpenProjectAuthenticationError } from "../client/api-client.ts"; +import { rankRecords, type FieldMatch, type FieldSpec, type RankOptions } from "./rank.ts"; + +export const DEFAULT_ENRICH_LIMIT = 25; +export const DEFAULT_CONCURRENCY = 8; + +export interface PipelineSpec { + /** Fetches the candidate window. */ + fetchCandidates: () => Promise; + /** Fields available without any additional API call. */ + shallowFields: FieldSpec[]; + /** Fetches deep content for one candidate. */ + enrich: (candidate: TShallow) => Promise; + /** Fields available only after enrichment. */ + deepFields: FieldSpec[]; + enrichLimit?: number; + concurrency?: number; + /** Orders the recency backfill. Omitted means "keep API order". */ + recencyOf?: (candidate: TShallow) => string | number; + idOf?: (candidate: TShallow) => number; +} + +export interface PipelineEntry { + record: TShallow; + score: number; + matches: FieldMatch[]; + deep?: TDeep; +} + +export interface PipelineResult { + ranked: PipelineEntry[]; + /** True when at least one enrichment failed non-fatally. */ + degraded: boolean; + enrichmentFailures: number; +} + +/** + * True for errors that must abort the whole search rather than degrade one + * record: an expired token must not be reported to the model as "no results". + */ +export function isFatalSearchError(error: unknown): boolean { + if (error instanceof OpenProjectAuthenticationError) { + return true; + } + const statusCode = (error as { statusCode?: number } | null)?.statusCode; + return statusCode === 401 || statusCode === 403 || statusCode === 429; +} + +/** + * Maps over items with a bounded number of in-flight operations, preserving + * input order in the results. + */ +export async function mapWithConcurrency( + items: T[], + concurrency: number, + fn: (item: T, index: number) => Promise +): Promise { + const results = new Array(items.length); + let nextIndex = 0; + + const workerCount = Math.max(1, Math.min(concurrency, items.length)); + const workers = Array.from({ length: workerCount }, async () => { + for (;;) { + const index = nextIndex++; + if (index >= items.length) { + return; + } + results[index] = await fn(items[index]!, index); + } + }); + + await Promise.all(workers); + return results; +} + +function recencyValue( + recencyOf: ((candidate: T) => string | number) | undefined, + candidate: T +): number { + if (!recencyOf) { + return 0; + } + const value = recencyOf(candidate); + if (typeof value === "number") { + return value; + } + const parsed = Date.parse(value); + return Number.isNaN(parsed) ? 0 : parsed; +} + +export async function searchPipeline( + spec: PipelineSpec, + query: string, + opts?: RankOptions +): Promise> { + const candidates = await spec.fetchCandidates(); + const enrichLimit = spec.enrichLimit ?? DEFAULT_ENRICH_LIMIT; + const concurrency = spec.concurrency ?? DEFAULT_CONCURRENCY; + const idOf = spec.idOf ?? ((candidate: TShallow) => (candidate as { id?: number })?.id ?? 0); + + // Phase 1: rank on data we already have. No limit here — the limit applies + // to the final merged result, not the intermediate ranking. + const shallowRanked = rankRecords(candidates, query, spec.shallowFields, { + ...opts, + idOf, + limit: undefined, + }); + + const entries = new Map>(); + for (const ranked of shallowRanked) { + entries.set(idOf(ranked.record), { + record: ranked.record, + score: ranked.score, + matches: ranked.matches, + }); + } + + // Enrichment set: top N by shallow score, backfilled by recency. The + // backfill exists because a record whose only match is deep scores zero + // shallowly and would otherwise never be enriched, and so never found. + const selected = shallowRanked.slice(0, enrichLimit).map((ranked) => ranked.record); + if (selected.length < enrichLimit) { + const alreadySelected = new Set(selected.map(idOf)); + const backfill = candidates + .filter((candidate) => !alreadySelected.has(idOf(candidate))) + .sort((a, b) => recencyValue(spec.recencyOf, b) - recencyValue(spec.recencyOf, a)); + + for (const candidate of backfill) { + if (selected.length >= enrichLimit) { + break; + } + selected.push(candidate); + } + } + + // Phase 2: bounded enrichment. + // + // The explicit EnrichedCandidate annotation is required, not stylistic: + // without it TypeScript infers the callback's return as a UNION of two + // object literal types ({ deep: Awaited } | { deep: undefined }) + // rather than a single type with `deep: TDeep | undefined`. The narrowing + // predicate below is then rejected, because a type predicate's type must be + // assignable to its parameter's type and `Pair` is not assignable to the + // `{ deep: undefined }` arm. + interface EnrichedCandidate { + candidate: TShallow; + deep: TDeep | undefined; + } + + let enrichmentFailures = 0; + const enriched: EnrichedCandidate[] = await mapWithConcurrency( + selected, + concurrency, + async (candidate) => { + try { + return { candidate, deep: await spec.enrich(candidate) }; + } catch (error: unknown) { + if (isFatalSearchError(error)) { + throw error; + } + enrichmentFailures++; + return { candidate, deep: undefined }; + } + } + ); + + interface Pair { + candidate: TShallow; + deep: TDeep; + } + + const pairs: Pair[] = enriched.filter( + (entry): entry is Pair => entry.deep !== undefined + ); + + // Attach deep content to every enriched entry, matched or not, so callers + // can reuse it without re-fetching. + for (const pair of pairs) { + const existing = entries.get(idOf(pair.candidate)); + if (existing) { + existing.deep = pair.deep; + } + } + + const pairFields: FieldSpec[] = spec.deepFields.map((field) => ({ + name: field.name, + weight: field.weight, + extract: (pair: Pair) => field.extract(pair.deep), + })); + + const deepRanked = rankRecords(pairs, query, pairFields, { + ...opts, + idOf: (pair: Pair) => idOf(pair.candidate), + limit: undefined, + }); + + for (const ranked of deepRanked) { + const id = idOf(ranked.record.candidate); + const existing = entries.get(id); + + if (existing) { + // max(shallow, deep), never a sum. + if (ranked.score > existing.score) { + existing.score = ranked.score; + } + existing.matches = [...existing.matches, ...ranked.matches].sort( + (a, b) => b.score - a.score + ); + existing.deep = ranked.record.deep; + } else { + entries.set(id, { + record: ranked.record.candidate, + score: ranked.score, + matches: ranked.matches, + deep: ranked.record.deep, + }); + } + } + + const ranked = [...entries.values()].sort( + (a, b) => b.score - a.score || idOf(a.record) - idOf(b.record) + ); + + const limited = + opts?.limit !== undefined && opts.limit > 0 ? ranked.slice(0, opts.limit) : ranked; + + return { + ranked: limited, + degraded: enrichmentFailures > 0, + enrichmentFailures, + }; +} diff --git a/tests/search/pipeline.test.ts b/tests/search/pipeline.test.ts new file mode 100644 index 0000000..7384225 --- /dev/null +++ b/tests/search/pipeline.test.ts @@ -0,0 +1,204 @@ +import { describe, expect, test } from "bun:test"; +import { mapWithConcurrency, searchPipeline } from "../../src/search/pipeline.ts"; +import { OpenProjectAuthenticationError, OpenProjectError } from "../../src/client/api-client.ts"; +import type { FieldSpec } from "../../src/search/rank.ts"; + +interface Candidate { + id: number; + title: string; + updatedAt: string; +} + +interface Enriched { + id: number; + notes: string; +} + +function makeCandidates(count: number): Candidate[] { + return Array.from({ length: count }, (_, index) => ({ + id: index + 1, + title: `Item ${index + 1}`, + updatedAt: new Date(2026, 0, index + 1).toISOString(), + })); +} + +const SHALLOW: FieldSpec[] = [{ name: "title", weight: 3, extract: (c) => c.title }]; +const DEEP: FieldSpec[] = [{ name: "notes", weight: 2, extract: (e) => e.notes }]; + +describe("mapWithConcurrency", () => { + test("never exceeds the concurrency limit", async () => { + let inFlight = 0; + let peak = 0; + + await mapWithConcurrency(Array.from({ length: 50 }, (_, i) => i), 8, async (item) => { + inFlight++; + peak = Math.max(peak, inFlight); + await new Promise((resolve) => setTimeout(resolve, 1)); + inFlight--; + return item; + }); + + expect(peak).toBeLessThanOrEqual(8); + }); + + test("preserves input order in the results", async () => { + const result = await mapWithConcurrency([3, 1, 2], 2, async (item) => { + await new Promise((resolve) => setTimeout(resolve, item)); + return item * 10; + }); + expect(result).toEqual([30, 10, 20]); + }); + + test("handles an empty input array", async () => { + expect(await mapWithConcurrency([], 4, async (i) => i)).toEqual([]); + }); +}); + +describe("searchPipeline", () => { + test("caps enrichment at the enrich limit", async () => { + let enrichCalls = 0; + + await searchPipeline( + { + fetchCandidates: async () => makeCandidates(100), + shallowFields: SHALLOW, + enrich: async (candidate) => { + enrichCalls++; + return { id: candidate.id, notes: "" }; + }, + deepFields: DEEP, + recencyOf: (candidate) => candidate.updatedAt, + }, + "item" + ); + + expect(enrichCalls).toBe(25); + }); + + test("backfills the enrichment set by recency when few candidates match shallowly", async () => { + const enriched: number[] = []; + + await searchPipeline( + { + fetchCandidates: async () => makeCandidates(100), + shallowFields: SHALLOW, + enrich: async (candidate) => { + enriched.push(candidate.id); + return { id: candidate.id, notes: "" }; + }, + deepFields: DEEP, + recencyOf: (candidate) => candidate.updatedAt, + }, + "zzzznomatch" + ); + + // Nothing matches shallowly, so the whole budget is spent on the most + // recently updated candidates: ids 100 down to 76. + expect(enriched).toHaveLength(25); + expect(enriched[0]).toBe(100); + expect(enriched).toContain(76); + expect(enriched).not.toContain(75); + }); + + test("finds a record whose only match is in deep content", async () => { + const result = await searchPipeline( + { + fetchCandidates: async () => makeCandidates(10), + shallowFields: SHALLOW, + enrich: async (candidate) => ({ + id: candidate.id, + notes: candidate.id === 3 ? "We agreed on the hiring freeze" : "routine notes", + }), + deepFields: DEEP, + recencyOf: (candidate) => candidate.updatedAt, + }, + "hiring freeze" + ); + + expect(result.ranked[0]!.record.id).toBe(3); + expect(result.ranked[0]!.deep?.notes).toContain("hiring freeze"); + }); + + test("takes the maximum of shallow and deep scores rather than the sum", async () => { + const result = await searchPipeline( + { + fetchCandidates: async () => [{ id: 1, title: "Budget", updatedAt: "2026-01-01" }], + shallowFields: SHALLOW, + enrich: async () => ({ id: 1, notes: "totally unrelated filler text" }), + deepFields: DEEP, + }, + "budget" + ); + + expect(result.ranked[0]!.score).toBeLessThanOrEqual(1); + expect(result.ranked[0]!.score).toBeGreaterThan(0.9); + }); + + test("marks the result degraded when an enrichment fails non-fatally", async () => { + const result = await searchPipeline( + { + fetchCandidates: async () => makeCandidates(3), + shallowFields: SHALLOW, + enrich: async (candidate) => { + if (candidate.id === 2) { + throw new OpenProjectError("boom", { statusCode: 500 }); + } + return { id: candidate.id, notes: "" }; + }, + deepFields: DEEP, + }, + "item" + ); + + expect(result.degraded).toBe(true); + expect(result.enrichmentFailures).toBe(1); + expect(result.ranked.length).toBeGreaterThan(0); + }); + + test("propagates authentication errors instead of swallowing them", async () => { + const run = searchPipeline( + { + fetchCandidates: async () => makeCandidates(3), + shallowFields: SHALLOW, + enrich: async () => { + throw new OpenProjectAuthenticationError(); + }, + deepFields: DEEP, + }, + "item" + ); + + await expect(run).rejects.toThrow(OpenProjectAuthenticationError); + }); + + test("propagates rate limit errors instead of swallowing them", async () => { + const run = searchPipeline( + { + fetchCandidates: async () => makeCandidates(3), + shallowFields: SHALLOW, + enrich: async () => { + throw new OpenProjectError("slow down", { statusCode: 429 }); + }, + deepFields: DEEP, + }, + "item" + ); + + await expect(run).rejects.toThrow("slow down"); + }); + + test("reports no degradation when every enrichment succeeds", async () => { + const result = await searchPipeline( + { + fetchCandidates: async () => makeCandidates(3), + shallowFields: SHALLOW, + enrich: async (candidate) => ({ id: candidate.id, notes: "" }), + deepFields: DEEP, + }, + "item" + ); + + expect(result.degraded).toBe(false); + expect(result.enrichmentFailures).toBe(0); + }); +});