Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions apps/web/src/application/ports/llm-provider.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<MatchCandidate[]>;
}
3 changes: 2 additions & 1 deletion apps/web/src/application/ports/match-repository.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import type { MatchCandidate } from "../../domain/matching/match-strategy";

export interface MatchRepository {
save(eventId: string, participantId: string, candidates: MatchCandidate[]): Promise<void>;
/** Replaces the participant's matches; array order is the order shown to them. */
save(eventId: string, participantId: string, candidates: MatchCandidate[], computationVersion: string): Promise<void>;
listForParticipant(eventId: string, participantId: string, limit?: number): Promise<MatchCandidate[]>;
}
10 changes: 8 additions & 2 deletions apps/web/src/domain/matching/pipeline.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
}
Expand Down
16 changes: 12 additions & 4 deletions apps/web/src/domain/matching/stages/llm-rerank.stage.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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:<provider id>` 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<MatchCandidate[]> {
if (!this.provider || candidates.length === 0) return candidates.map(addFallbackReason);
async execute(viewer: ParticipantProfile, candidates: MatchCandidate[]): Promise<RerankResult> {
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;
}
}
}
Expand Down
4 changes: 2 additions & 2 deletions apps/web/src/domain/matching/stages/persist.stage.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<void> {
return this.matches.save(eventId, participantId, candidates);
execute(eventId: string, participantId: string, candidates: MatchCandidate[], computationVersion: string): Promise<void> {
return this.matches.save(eventId, participantId, candidates, computationVersion);
}
}
29 changes: 18 additions & 11 deletions apps/web/src/infrastructure/postgres/match.repo.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,18 +25,25 @@ interface MatchRow {
};

export class PostgresMatchRepository implements MatchRepository {
async save(eventId: string, participantId: string, candidates: MatchCandidate[]): Promise<void> {
async save(eventId: string, participantId: string, candidates: MatchCandidate[], computationVersion: string): Promise<void> {
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)]
);
});
}

Expand All @@ -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) => ({
Expand Down
8 changes: 7 additions & 1 deletion apps/web/src/infrastructure/providers/gemini-llm.provider.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<MatchCandidate[]> {
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,
Expand Down
34 changes: 34 additions & 0 deletions packages/db/migrations/0004_match_rank.sql
Original file line number Diff line number Diff line change
@@ -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;
26 changes: 22 additions & 4 deletions tests/integration/login-challenge.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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() {
Expand Down Expand Up @@ -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");
Expand Down
88 changes: 88 additions & 0 deletions tests/integration/match-rank.test.ts
Original file line number Diff line number Diff line change
@@ -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<string, string>;

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([]);
});
});
71 changes: 71 additions & 0 deletions tests/unit/matching-pipeline.test.ts
Original file line number Diff line number Diff line change
@@ -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> = {}): 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);
});
});
Loading