diff --git a/apps/web/src/application/ports/llm-provider.ts b/apps/web/src/application/ports/llm-provider.ts index bd3d781..fb5bbb1 100644 --- a/apps/web/src/application/ports/llm-provider.ts +++ b/apps/web/src/application/ports/llm-provider.ts @@ -2,5 +2,7 @@ import type { MatchCandidate } from "../../domain/matching/match-strategy"; import type { ParticipantProfile } from "../../domain/participant/participant.entity"; export interface LlmProvider { + /** Identifies the provider and model, recorded with every reranked result. */ + readonly id: string; rerank(viewer: ParticipantProfile, candidates: MatchCandidate[]): Promise; } diff --git a/apps/web/src/application/ports/match-repository.ts b/apps/web/src/application/ports/match-repository.ts index 4cb2954..26d6ed6 100644 --- a/apps/web/src/application/ports/match-repository.ts +++ b/apps/web/src/application/ports/match-repository.ts @@ -1,6 +1,7 @@ import type { MatchCandidate } from "../../domain/matching/match-strategy"; export interface MatchRepository { - save(eventId: string, participantId: string, candidates: MatchCandidate[]): Promise; + /** Replaces the participant's matches; array order is the order shown to them. */ + save(eventId: string, participantId: string, candidates: MatchCandidate[], computationVersion: string): Promise; listForParticipant(eventId: string, participantId: string, limit?: number): Promise; } diff --git a/apps/web/src/domain/matching/pipeline.ts b/apps/web/src/domain/matching/pipeline.ts index babb493..941c786 100644 --- a/apps/web/src/domain/matching/pipeline.ts +++ b/apps/web/src/domain/matching/pipeline.ts @@ -25,6 +25,12 @@ export interface MatchingInput { limit?: number; } +/** + * Bump when scorers, weights, or stage behavior change, so persisted matches + * from an older computation can be recognised and recomputed. + */ +export const MATCHING_PIPELINE_VERSION = "v1"; + /** Same retrieve/filter/score/top-K/rerank pipeline for both entry points. */ export class MatchingPipeline { private readonly retrieve: RetrieveStage; @@ -77,10 +83,10 @@ export class MatchingPipeline { const compatible = this.filter.execute(viewer, retrieved); const scored = this.score.execute(viewer, compatible); const top = this.topK.execute(scored, input.limit ?? 12); - const ranked = await this.rerank.execute(viewer, top); + const { candidates: ranked, rankedBy } = await this.rerank.execute(viewer, top); if (input.mode === "profile") { - await this.persist.execute(input.eventId, input.participantId, ranked); + await this.persist.execute(input.eventId, input.participantId, ranked, `${MATCHING_PIPELINE_VERSION}+${rankedBy}`); } return ranked; } diff --git a/apps/web/src/domain/matching/stages/llm-rerank.stage.ts b/apps/web/src/domain/matching/stages/llm-rerank.stage.ts index 4ed2e73..6002c0a 100644 --- a/apps/web/src/domain/matching/stages/llm-rerank.stage.ts +++ b/apps/web/src/domain/matching/stages/llm-rerank.stage.ts @@ -2,16 +2,24 @@ import type { LlmProvider } from "../../../application/ports/llm-provider"; import type { ParticipantProfile } from "../../participant/participant.entity"; import type { MatchCandidate } from "../match-strategy"; +export interface RerankResult { + candidates: MatchCandidate[]; + /** `llm:` when the model ordered the list, `score` for the fallback. */ + rankedBy: string; +} + export class LlmRerankStage { constructor(private readonly provider: LlmProvider | null) {} - async execute(viewer: ParticipantProfile, candidates: MatchCandidate[]): Promise { - if (!this.provider || candidates.length === 0) return candidates.map(addFallbackReason); + async execute(viewer: ParticipantProfile, candidates: MatchCandidate[]): Promise { + const scored = { candidates: candidates.map(addFallbackReason), rankedBy: "score" }; + if (!this.provider || candidates.length === 0) return scored; try { - return await this.provider.rerank(viewer, candidates); + const reranked = await this.provider.rerank(viewer, candidates); + return { candidates: reranked.map(addFallbackReason), rankedBy: `llm:${this.provider.id}` }; } catch (error) { console.warn("[matching] LLM rerank unavailable; using scored order", error); - return candidates.map(addFallbackReason); + return scored; } } } diff --git a/apps/web/src/domain/matching/stages/persist.stage.ts b/apps/web/src/domain/matching/stages/persist.stage.ts index de97968..1c1c07e 100644 --- a/apps/web/src/domain/matching/stages/persist.stage.ts +++ b/apps/web/src/domain/matching/stages/persist.stage.ts @@ -3,7 +3,7 @@ import type { MatchCandidate } from "../match-strategy"; export class PersistStage { constructor(private readonly matches: MatchRepository) {} - execute(eventId: string, participantId: string, candidates: MatchCandidate[]): Promise { - return this.matches.save(eventId, participantId, candidates); + execute(eventId: string, participantId: string, candidates: MatchCandidate[], computationVersion: string): Promise { + return this.matches.save(eventId, participantId, candidates, computationVersion); } } diff --git a/apps/web/src/infrastructure/postgres/match.repo.ts b/apps/web/src/infrastructure/postgres/match.repo.ts index d0f96f7..869a3f5 100644 --- a/apps/web/src/infrastructure/postgres/match.repo.ts +++ b/apps/web/src/infrastructure/postgres/match.repo.ts @@ -25,18 +25,25 @@ interface MatchRow { }; export class PostgresMatchRepository implements MatchRepository { - async save(eventId: string, participantId: string, candidates: MatchCandidate[]): Promise { + async save(eventId: string, participantId: string, candidates: MatchCandidate[], computationVersion: string): Promise { await transaction(async (client) => { await client.query(`delete from event_matches where event_id = $1 and participant_id = $2`, [eventId, participantId]); - for (const candidate of candidates) { - await client.query( - `insert into event_matches - (event_id, participant_id, candidate_id, score, score_breakdown, match_reason, mutual_interest) - values ($1, $2, $3, $4, $5::jsonb, $6, $7)`, - [eventId, participantId, candidate.participant.attendeeId, candidate.score, - JSON.stringify(candidate.scoreBreakdown), candidate.reason, candidate.mutualInterest] - ); - } + if (candidates.length === 0) return; + // Array position is the final (possibly reranked) order, stored as rank 1..N. + await client.query( + `insert into event_matches + (event_id, participant_id, candidate_id, rank, score, score_breakdown, match_reason, mutual_interest, computation_version) + select $1, $2, c.candidate_id, c.rank, c.score, c.score_breakdown, c.match_reason, c.mutual_interest, $3 + from unnest($4::uuid[], $5::int[], $6::numeric[], $7::jsonb[], $8::text[], $9::boolean[]) + as c(candidate_id, rank, score, score_breakdown, match_reason, mutual_interest)`, + [eventId, participantId, computationVersion, + candidates.map((candidate) => candidate.participant.attendeeId), + candidates.map((_, index) => index + 1), + candidates.map((candidate) => candidate.score), + candidates.map((candidate) => JSON.stringify(candidate.scoreBreakdown)), + candidates.map((candidate) => candidate.reason), + candidates.map((candidate) => candidate.mutualInterest)] + ); }); } @@ -55,7 +62,7 @@ export class PostgresMatchRepository implements MatchRepository { join event_participants p on p.event_id = m.event_id and p.attendee_id = m.candidate_id join registered_attendees a on a.id = p.attendee_id where m.event_id = $1 and m.participant_id = $2 - order by m.score desc, m.created_at desc limit $3`, + order by m.rank, m.candidate_id limit $3`, [eventId, participantId, Math.max(1, Math.min(100, limit))] ); return rows.map((row) => ({ diff --git a/apps/web/src/infrastructure/providers/gemini-llm.provider.ts b/apps/web/src/infrastructure/providers/gemini-llm.provider.ts index 2d1b2ea..d3a3188 100644 --- a/apps/web/src/infrastructure/providers/gemini-llm.provider.ts +++ b/apps/web/src/infrastructure/providers/gemini-llm.provider.ts @@ -2,11 +2,17 @@ import type { LlmProvider } from "../../application/ports/llm-provider"; import type { MatchCandidate } from "../../domain/matching/match-strategy"; import type { ParticipantProfile } from "../../domain/participant/participant.entity"; +const DEFAULT_MODEL = "gemini-3.8-flash"; + export class GeminiLlmProvider implements LlmProvider { + get id(): string { + return `gemini:${process.env.GEMINI_LLM_MODEL || DEFAULT_MODEL}`; + } + async rerank(viewer: ParticipantProfile, candidates: MatchCandidate[]): Promise { const apiKey = process.env.GEMINI_API_KEY; if (!apiKey) return candidates; - const model = process.env.GEMINI_LLM_MODEL || "gemini-3.8-flash"; + const model = process.env.GEMINI_LLM_MODEL || DEFAULT_MODEL; const input = candidates.map(({ participant, score, mutualInterest }) => ({ id: participant.attendeeId, name: participant.name, diff --git a/packages/db/migrations/0004_match_rank.sql b/packages/db/migrations/0004_match_rank.sql new file mode 100644 index 0000000..8934d2e --- /dev/null +++ b/packages/db/migrations/0004_match_rank.sql @@ -0,0 +1,34 @@ +-- Persist the final order of saved matches. +-- +-- The LLM rerank can order the shortlist differently from the numeric score, +-- but rows were stored without a position and read back by score, so a page +-- reload silently undid the rerank. `rank` is the 1-based position shown to the +-- participant; `computation_version` records which pipeline and reranker +-- produced the list, so stale results can be identified and recomputed. + +alter table event_matches add column if not exists rank integer; +alter table event_matches add column if not exists computation_version text not null default 'legacy'; + +-- Existing rows keep the order they were already displayed in. +update event_matches m + set rank = ordered.position + from ( + select event_id, participant_id, candidate_id, + row_number() over ( + partition by event_id, participant_id + order by score desc, created_at desc, candidate_id + ) as position + from event_matches + ) ordered + where m.event_id = ordered.event_id + and m.participant_id = ordered.participant_id + and m.candidate_id = ordered.candidate_id + and m.rank is null; + +alter table event_matches alter column rank set not null; +alter table event_matches add constraint event_matches_rank_positive check (rank > 0); +-- New rows must say which computation produced them. +alter table event_matches alter column computation_version drop default; + +create unique index if not exists idx_event_matches_rank on event_matches(event_id, participant_id, rank); +drop index if exists idx_event_matches_participant; diff --git a/tests/integration/login-challenge.test.ts b/tests/integration/login-challenge.test.ts index a7f47db..6026ad3 100644 --- a/tests/integration/login-challenge.test.ts +++ b/tests/integration/login-challenge.test.ts @@ -14,6 +14,10 @@ const hasDatabase = Boolean(process.env.DATABASE_URL); const describeDb = hasDatabase ? describe : describe.skip; const EMAIL = "integration-test@example.com"; +const QUERY_EMAIL = "integration-query@example.com"; + +// Shared by both suites below, so the pool closes once, after both finish. +afterAll(() => closePool()); describeDb("login challenge lifecycle", () => { let attendeeId: string; @@ -32,7 +36,6 @@ describeDb("login challenge lifecycle", () => { afterAll(async () => { await query(`delete from login_challenges where email = $1`, [EMAIL]); await query(`delete from registered_attendees where email = $1`, [EMAIL]); - await closePool(); }); async function freshChallenge() { @@ -139,15 +142,30 @@ describeDb("login challenge lifecycle", () => { }); }); +// Owns its attendee: the suite above deletes EMAIL in its afterAll. describeDb("attendee queries", () => { + let attendeeId: string; + + beforeAll(async () => { + await migrate(); + await attendees.upsertMany([ + { name: "Integration Query", email: QUERY_EMAIL, phone: "+919999900002" }, + ]); + attendeeId = (await attendees.findByEmail(QUERY_EMAIL))!.id; + }); + + afterAll(async () => { + await query(`delete from registered_attendees where email = $1`, [QUERY_EMAIL]); + }); + it("excludes the requester from match candidates", async () => { - const attendee = await attendees.findByEmail(EMAIL); - const candidates = await attendees.listMatchCandidates(attendee!.id); - expect(candidates.find((c) => c.id === attendee!.id)).toBeUndefined(); + const candidates = await attendees.listMatchCandidates(attendeeId); + expect(candidates.find((c) => c.id === attendeeId)).toBeUndefined(); }); it("never exposes contact details on candidates", async () => { const candidates = await attendees.listMatchCandidates(null); + expect(candidates.some((c) => c.id === attendeeId)).toBe(true); for (const candidate of candidates) { expect(candidate).not.toHaveProperty("email"); expect(candidate).not.toHaveProperty("phone"); diff --git a/tests/integration/match-rank.test.ts b/tests/integration/match-rank.test.ts new file mode 100644 index 0000000..4f21f9a --- /dev/null +++ b/tests/integration/match-rank.test.ts @@ -0,0 +1,88 @@ +/** + * Saved matches must come back in the order they were saved (the reranked + * order), not re-sorted by numeric score. Runs against a real Postgres and is + * skipped when DATABASE_URL is unset. + */ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { query, closePool, attendees } from "@match-maker/db"; +import { migrate } from "../../packages/db/src/migrate"; +import { PostgresMatchRepository } from "../../apps/web/src/infrastructure/postgres/match.repo"; +import { Intent } from "../../apps/web/src/domain/participant/intent"; +import type { MatchCandidate } from "../../apps/web/src/domain/matching/match-strategy"; + +const hasDatabase = Boolean(process.env.DATABASE_URL); +const describeDb = hasDatabase ? describe : describe.skip; + +const ORG_SLUG = "integration-match-rank"; +const EMAILS = ["rank-viewer@example.com", "rank-a@example.com", "rank-b@example.com", "rank-c@example.com"]; + +describeDb("persisted match order", () => { + const repo = new PostgresMatchRepository(); + let eventId: string; + let ids: Record; + + beforeAll(async () => { + await migrate(); + await attendees.upsertMany(EMAILS.map((email) => ({ name: email.split("@")[0], email }))); + const found = await attendees.findIdsByEmails(EMAILS); + ids = Object.fromEntries(found.map((row) => [row.email.split("@")[0].replace("rank-", ""), row.id])); + const { rows: [org] } = await query<{ id: string }>( + `insert into organizations (name, slug) values ('Rank test', $1) + on conflict (slug) do update set name = excluded.name returning id`, [ORG_SLUG] + ); + const { rows: [event] } = await query<{ id: string }>( + `insert into events (organization_id, name, slug, status) values ($1, 'Rank test', $2, 'open') + on conflict (organization_id, slug) do update set status = 'open' returning id`, [org.id, ORG_SLUG] + ); + eventId = event.id; + for (const id of Object.values(ids)) { + await query(`insert into event_participants (event_id, attendee_id) values ($1, $2) on conflict do nothing`, [eventId, id]); + } + }); + + afterAll(async () => { + await query(`delete from organizations where slug = $1`, [ORG_SLUG]); + await query(`delete from registered_attendees where email = any($1::text[])`, [EMAILS]); + await closePool(); + }); + + const candidate = (key: string, score: number): MatchCandidate => ({ + participant: { + eventId, attendeeId: ids[key], name: key, designation: null, domain: null, aboutMe: null, + selfDescription: null, yearsOfExperience: null, mostImpressiveBuild: null, + intent: Intent.NETWORKING, interests: [], venueZone: null, visible: true, + }, + score, + scoreBreakdown: { domain: score / 100 }, + mutualInterest: false, + reason: `Reason for ${key}`, + }); + // Map stored attendee ids back to the short keys used in each test. + const names = (rows: MatchCandidate[]) => + rows.map((row) => Object.keys(ids).find((key) => ids[key] === row.participant.attendeeId)); + + it("reads back the reranked order even when it contradicts the scores", async () => { + // The model put the lowest score first; score order would be b, c, a. + await repo.save(eventId, ids.viewer, [candidate("a", 40), candidate("b", 90), candidate("c", 65)], "v1+llm:test"); + const saved = await repo.listForParticipant(eventId, ids.viewer); + expect(names(saved)).toEqual(["a", "b", "c"]); + expect(saved.map((row) => row.score)).toEqual([40, 90, 65]); + expect(saved[0]).toMatchObject({ reason: "Reason for a", scoreBreakdown: { domain: 0.4 } }); + + const { rows } = await query<{ rank: number; computation_version: string }>( + `select rank, computation_version from event_matches where event_id = $1 and participant_id = $2 order by rank`, + [eventId, ids.viewer] + ); + expect(rows).toEqual([1, 2, 3].map((rank) => ({ rank, computation_version: "v1+llm:test" }))); + }); + + it("replaces the previous list with the new order", async () => { + await repo.save(eventId, ids.viewer, [candidate("c", 65), candidate("a", 40)], "v1+score"); + expect(names(await repo.listForParticipant(eventId, ids.viewer))).toEqual(["c", "a"]); + }); + + it("clears the list when nothing matches", async () => { + await repo.save(eventId, ids.viewer, [], "v1+score"); + expect(await repo.listForParticipant(eventId, ids.viewer)).toEqual([]); + }); +}); diff --git a/tests/unit/matching-pipeline.test.ts b/tests/unit/matching-pipeline.test.ts new file mode 100644 index 0000000..31975dd --- /dev/null +++ b/tests/unit/matching-pipeline.test.ts @@ -0,0 +1,71 @@ +import { describe, expect, it, vi } from "vitest"; +import { MATCHING_PIPELINE_VERSION, MatchingPipeline } from "@/src/domain/matching/pipeline"; +import type { LlmProvider } from "@/src/application/ports/llm-provider"; +import type { MatchRepository } from "@/src/application/ports/match-repository"; +import type { ParticipantRepository } from "@/src/application/ports/participant-repository"; +import { Intent } from "@/src/domain/participant/intent"; +import type { ParticipantProfile } from "@/src/domain/participant/participant.entity"; + +const profile = (attendeeId: string, overrides: Partial = {}): ParticipantProfile => ({ + eventId: "event", attendeeId, name: attendeeId, designation: null, domain: null, aboutMe: null, + selfDescription: null, yearsOfExperience: null, mostImpressiveBuild: null, + intent: Intent.NETWORKING, interests: [], venueZone: null, visible: true, embedding: null, ...overrides, +}); + +// "strong" outscores "weak" on domain and shared interests, so score order is strong → weak. +const viewer = profile("viewer", { domain: "Machine Learning", interests: ["rag", "agents"] }); +const strong = profile("strong", { domain: "Machine Learning", interests: ["rag", "agents"] }); +const weak = profile("weak", { domain: "Cybersecurity", interests: ["kubernetes"] }); + +function setup(llm: LlmProvider | null) { + const participants = { + findProfile: vi.fn(async () => viewer), + listCandidates: vi.fn(async () => [weak, strong]), + setEmbedding: vi.fn(), + } as unknown as ParticipantRepository; + const matches = { save: vi.fn(), listForParticipant: vi.fn() } satisfies MatchRepository; + const embeddings = { embed: vi.fn(async () => null) }; + const cache = { get: vi.fn(), set: vi.fn() }; + const pipeline = new MatchingPipeline(participants, matches, embeddings, llm, cache); + return { pipeline, matches }; +} + +const run = (pipeline: MatchingPipeline) => + pipeline.run({ eventId: "event", participantId: "viewer", mode: "profile" }); +const ids = (candidates: Array<{ participant: ParticipantProfile }>) => candidates.map((c) => c.participant.attendeeId); + +describe("matching pipeline persistence", () => { + it("persists the reranked order, not the numeric score order", async () => { + const llm: LlmProvider = { + id: "test-model", + rerank: vi.fn(async (_viewer, candidates) => [...candidates].reverse()), + }; + const { pipeline, matches } = setup(llm); + const result = await run(pipeline); + + expect(result[0].score).toBeLessThan(result[1].score); + expect(ids(result)).toEqual(["weak", "strong"]); + const [, , saved, version] = matches.save.mock.calls[0]; + expect(ids(saved)).toEqual(["weak", "strong"]); + expect(version).toBe(`${MATCHING_PIPELINE_VERSION}+llm:test-model`); + }); + + it("records the scored fallback and keeps a reason for every match", async () => { + const llm: LlmProvider = { id: "test-model", rerank: vi.fn(async () => { throw new Error("timeout"); }) }; + vi.spyOn(console, "warn").mockImplementation(() => {}); + const { pipeline, matches } = setup(llm); + await run(pipeline); + + const [, , saved, version] = matches.save.mock.calls[0]; + expect(ids(saved)).toEqual(["strong", "weak"]); + expect(version).toBe(`${MATCHING_PIPELINE_VERSION}+score`); + expect(saved.every((candidate: { reason: string | null }) => Boolean(candidate.reason))).toBe(true); + }); + + it("fills a reason the model left out", async () => { + const llm: LlmProvider = { id: "test-model", rerank: vi.fn(async (_viewer, candidates) => candidates) }; + const { pipeline } = setup(llm); + const result = await run(pipeline); + expect(result.every((candidate) => Boolean(candidate.reason))).toBe(true); + }); +});