feat: add lead qualification workflow

convex/blacklist.ts

@@ -1,7 +1,15 @@
 import { v } from "convex/values";
 
+import {
+  normalizeDomain,
+  normalizeEmailAddress,
+  normalizePhone,
+  normalizeText,
+} from "../lib/lead-discovery-google";
 import { normalizeListLimit } from "./domain";
-import { mutation, query } from "./_generated/server";
+import { internal } from "./_generated/api";
+import type { Doc } from "./_generated/dataModel";
+import { internalMutation, mutation, query, type MutationCtx } from "./_generated/server";
 
 const blacklistType = v.union(
   v.literal("domain"),
@@ -11,8 +19,193 @@ const blacklistType = v.union(
   v.literal("google_place_id"),
 );
 
-function normalizeBlacklistValue(value: string) {
-  return value.trim().toLowerCase();
+type BlacklistType =
+  | "domain"
+  | "email"
+  | "phone"
+  | "company"
+  | "google_place_id";
+
+const BLACKLIST_APPLY_BATCH_SIZE = 100;
+const BLACKLIST_REVIEW_NOTE_PREFIX =
+  "Lead automatisch durch Sperrlisteneintrag blockiert.";
+
+type BlacklistReason = {
+  type: BlacklistType;
+  normalizedValue: string;
+  reason: string;
+};
+
+type LeadIdAndBlacklistPatch = Pick<
+  Doc<"leads">,
+  "blacklistStatus" | "priority" | "contactStatus" | "blacklistReason" | "priorityReason" | "contactStatusReason"
+> & {
+  updatedAt: number;
+};
+
+type LeadMatchingFieldsPatch = Partial<
+  Pick<
+    Doc<"leads">,
+    | "normalizedEmail"
+    | "normalizedPhone"
+    | "normalizedCompanyName"
+    | "normalizedAddress"
+    | "normalizedGooglePlaceId"
+  >
+> & {
+  updatedAt: number;
+};
+
+type LeadIdRow = Pick<Doc<"leads">, "_id">;
+
+type LeadMatchQuery = {
+  order: (direction: "asc" | "desc") => {
+    paginate: (args: {
+      numItems: number;
+      cursor: string | null;
+    }) => Promise<{
+      page: LeadIdRow[];
+      isDone: boolean;
+      continueCursor: string | null;
+    }>;
+  };
+};
+
+function buildBlacklistReason(entry: { type: BlacklistType; value: string; note?: string }) {
+  const normalizedNote = entry.note?.trim();
+
+  return normalizedNote
+    ? `${BLACKLIST_REVIEW_NOTE_PREFIX} ${entry.type}: ${entry.value}. ${normalizedNote}`
+    : `${BLACKLIST_REVIEW_NOTE_PREFIX} ${entry.type}: ${entry.value}.`;
+}
+
+function buildReasonPatch(reason: string) {
+  const patch: LeadIdAndBlacklistPatch = {
+    blacklistStatus: "blocked" as const,
+    priority: "blocked" as const,
+    contactStatus: "do_not_contact" as const,
+    blacklistReason: reason,
+    priorityReason: reason,
+    contactStatusReason: reason,
+    updatedAt: Date.now(),
+  };
+
+  return patch;
+}
+
+function getLeadMatchQuery(
+  ctx: MutationCtx,
+  type: BlacklistType,
+  normalizedValue: string,
+): (() => LeadMatchQuery) | null {
+  if (!normalizedValue) {
+    return null;
+  }
+
+  switch (type) {
+    case "domain":
+      return () =>
+        ctx.db
+          .query("leads")
+          .withIndex("by_websiteDomain", (q) =>
+            q.eq("websiteDomain", normalizedValue),
+          );
+    case "email":
+      return () =>
+        ctx.db
+          .query("leads")
+          .withIndex("by_normalizedEmail", (q) =>
+            q.eq("normalizedEmail", normalizedValue),
+          );
+    case "phone":
+      return () =>
+        ctx.db
+          .query("leads")
+          .withIndex("by_normalizedPhone", (q) =>
+            q.eq("normalizedPhone", normalizedValue),
+          );
+    case "company":
+      return () =>
+        ctx.db
+          .query("leads")
+          .withIndex("by_normalizedCompanyName", (q) =>
+            q.eq("normalizedCompanyName", normalizedValue),
+          );
+    case "google_place_id":
+      return () =>
+        ctx.db
+          .query("leads")
+          .withIndex("by_normalizedGooglePlaceId", (q) =>
+            q.eq("normalizedGooglePlaceId", normalizedValue),
+          );
+    default:
+      return null;
+  }
+}
+
+function buildLeadMatchingFieldsPatch(lead: Doc<"leads">) {
+  const patch: LeadMatchingFieldsPatch = {
+    updatedAt: Date.now(),
+  };
+  const normalizedEmail = normalizeEmailAddress(lead.email);
+  const normalizedPhone = normalizePhone(lead.phone);
+  const normalizedCompanyName = normalizeText(lead.companyName);
+  const normalizedAddress = normalizeText(lead.address);
+  const normalizedGooglePlaceId = normalizeDomain(lead.googlePlaceId);
+
+  if (!lead.normalizedEmail && normalizedEmail) {
+    patch.normalizedEmail = normalizedEmail;
+  }
+  if (!lead.normalizedPhone && normalizedPhone) {
+    patch.normalizedPhone = normalizedPhone;
+  }
+  if (!lead.normalizedCompanyName && normalizedCompanyName) {
+    patch.normalizedCompanyName = normalizedCompanyName;
+  }
+  if (!lead.normalizedAddress && normalizedAddress) {
+    patch.normalizedAddress = normalizedAddress;
+  }
+  if (!lead.normalizedGooglePlaceId && normalizedGooglePlaceId) {
+    patch.normalizedGooglePlaceId = normalizedGooglePlaceId;
+  }
+
+  return Object.keys(patch).length > 1 ? patch : null;
+}
+
+async function scheduleBackfillThenBlacklistApply(
+  ctx: MutationCtx,
+  reason: BlacklistReason,
+) {
+  await ctx.scheduler.runAfter(
+    0,
+    internal.blacklist.backfillLeadMatchingFieldsForBlacklist,
+    {
+      ...reason,
+      cursor: null,
+    },
+  );
+}
+
+function normalizeBlacklistValue(type: BlacklistType, value: string) {
+  const trimmed = value.trim();
+
+  if (!trimmed) {
+    return null;
+  }
+
+  switch (type) {
+    case "email":
+      return normalizeEmailAddress(trimmed);
+    case "phone":
+      return normalizePhone(trimmed);
+    case "domain":
+    case "google_place_id":
+      return normalizeDomain(trimmed);
+    case "company":
+      return normalizeText(trimmed);
+    default:
+      return null;
+  }
 }
 
 export const create = mutation({
@@ -22,11 +215,238 @@ export const create = mutation({
     note: v.optional(v.string()),
   },
   handler: async (ctx, args) => {
-    return await ctx.db.insert("blacklistEntries", {
-      ...args,
-      normalizedValue: normalizeBlacklistValue(args.value),
+    const type = args.type as BlacklistType;
+    const normalizedValue = normalizeBlacklistValue(type, args.value);
+
+    if (!normalizedValue) {
+      throw new Error("Blacklist-Wert ist ungültig.");
+    }
+
+    const existing = await ctx.db
+      .query("blacklistEntries")
+      .withIndex("by_type_and_normalizedValue", (q) =>
+        q.eq("type", type).eq("normalizedValue", normalizedValue),
+      )
+      .take(1);
+
+    if (existing[0]) {
+      await scheduleBackfillThenBlacklistApply(ctx, {
+        type,
+        normalizedValue,
+        reason: buildBlacklistReason({
+          type,
+          value: existing[0].value,
+          note: existing[0].note,
+        }),
+      });
+      return existing[0]._id;
+    }
+
+    const created = await ctx.db.insert("blacklistEntries", {
+      type,
+      value: args.value.trim(),
+      normalizedValue,
+      note: args.note,
       createdAt: Date.now(),
     });
+
+    await scheduleBackfillThenBlacklistApply(ctx, {
+      type,
+      normalizedValue,
+      reason: buildBlacklistReason({
+        type,
+        value: args.value.trim(),
+        note: args.note,
+      }),
+    });
+
+    return created;
+  },
+});
+
+export const update = mutation({
+  args: {
+    id: v.id("blacklistEntries"),
+    type: v.optional(blacklistType),
+    value: v.optional(v.string()),
+    note: v.optional(v.string()),
+  },
+  handler: async (ctx, args) => {
+    const current = await ctx.db.get(args.id);
+
+    if (!current) {
+      throw new Error("Blacklist-Eintrag nicht gefunden.");
+    }
+
+    const nextType = (args.type ?? current.type) as BlacklistType;
+    const patch: {
+      type: BlacklistType;
+      value?: string;
+      normalizedValue?: string;
+      note?: string;
+    } = {
+      type: nextType,
+    };
+    const nextNormalizedValueFromCurrent = normalizeBlacklistValue(
+      nextType,
+      current.value,
+    );
+
+    if (!nextNormalizedValueFromCurrent) {
+      throw new Error("Blacklist-Wert ist ungültig.");
+    }
+
+    let nextValue = current.value;
+    let nextNormalizedValue = nextNormalizedValueFromCurrent;
+
+    if (args.value !== undefined) {
+      const value = args.value.trim();
+      const normalizedValue = normalizeBlacklistValue(nextType, value);
+
+      if (!normalizedValue) {
+        throw new Error("Blacklist-Wert ist ungültig.");
+      }
+
+      const existing = await ctx.db
+        .query("blacklistEntries")
+        .withIndex("by_type_and_normalizedValue", (q) =>
+          q.eq("type", nextType).eq("normalizedValue", normalizedValue),
+        )
+        .take(1);
+
+      if (existing[0] && existing[0]._id !== args.id) {
+        return existing[0]._id;
+      }
+
+      patch.value = value;
+      patch.normalizedValue = normalizedValue;
+      nextValue = value;
+      nextNormalizedValue = normalizedValue;
+    }
+
+    if (args.note !== undefined) {
+      patch.note = args.note;
+    }
+
+    await ctx.db.patch(args.id, patch);
+    await scheduleBackfillThenBlacklistApply(ctx, {
+      type: nextType,
+      normalizedValue: nextNormalizedValue,
+      reason: buildBlacklistReason({
+        type: nextType,
+        value: nextValue,
+        note: patch.note ?? args.note ?? current.note,
+      }),
+    });
+    return args.id;
+  },
+});
+
+export const backfillLeadMatchingFieldsForBlacklist = internalMutation({
+  args: {
+    type: blacklistType,
+    normalizedValue: v.string(),
+    reason: v.string(),
+    cursor: v.union(v.string(), v.null()),
+  },
+  handler: async (ctx, args) => {
+    const page = await ctx.db
+      .query("leads")
+      .order("asc")
+      .paginate({
+        numItems: BLACKLIST_APPLY_BATCH_SIZE,
+        cursor: args.cursor,
+      });
+
+    for (const lead of page.page) {
+      const patch = buildLeadMatchingFieldsPatch(lead);
+
+      if (patch) {
+        await ctx.db.patch(lead._id, patch);
+      }
+    }
+
+    if (!page.isDone) {
+      await ctx.scheduler.runAfter(
+        0,
+        internal.blacklist.backfillLeadMatchingFieldsForBlacklist,
+        {
+          type: args.type,
+          normalizedValue: args.normalizedValue,
+          reason: args.reason,
+          cursor: page.continueCursor,
+        },
+      );
+      return null;
+    }
+
+    await ctx.scheduler.runAfter(
+      0,
+      internal.blacklist.applyBlacklistToMatchingLeadsBatch,
+      {
+        type: args.type,
+        normalizedValue: args.normalizedValue,
+        reason: args.reason,
+        cursor: null,
+      },
+    );
+
+    return null;
+  },
+});
+
+export const applyBlacklistToMatchingLeadsBatch = internalMutation({
+  args: {
+    type: blacklistType,
+    normalizedValue: v.string(),
+    reason: v.string(),
+    cursor: v.union(v.string(), v.null()),
+  },
+  handler: async (ctx, args) => {
+    const queryBuilder = getLeadMatchQuery(
+      ctx,
+      args.type as BlacklistType,
+      args.normalizedValue,
+    );
+
+    if (!queryBuilder) {
+      return null;
+    }
+
+    const page = await queryBuilder()
+      .order("asc")
+      .paginate({
+        numItems: BLACKLIST_APPLY_BATCH_SIZE,
+        cursor: args.cursor,
+      });
+    const patch = buildReasonPatch(args.reason);
+
+    for (const lead of page.page) {
+      await ctx.db.patch(lead._id, patch);
+    }
+
+    if (!page.isDone) {
+      await ctx.scheduler.runAfter(
+        0,
+        internal.blacklist.applyBlacklistToMatchingLeadsBatch,
+        {
+          type: args.type,
+          normalizedValue: args.normalizedValue,
+          reason: args.reason,
+          cursor: page.continueCursor,
+        },
+      );
+    }
+
+    return null;
+  },
+});
+
+export const remove = mutation({
+  args: { id: v.id("blacklistEntries") },
+  handler: async (ctx, args) => {
+    await ctx.db.delete(args.id);
+    return args.id;
   },
 });
 

convex/domain.ts

@@ -12,7 +12,7 @@ const SECRET_KEY_PATTERNS = [
 ];
 
 export const CAMPAIGN_STATUSES = ["active", "paused"] as const;
-export const LEAD_PRIORITIES = ["high", "medium", "low", "defer"] as const;
+export const LEAD_PRIORITIES = ["high", "medium", "low", "defer", "blocked"] as const;
 export const LEAD_CONTACT_STATUSES = [
   "new",
   "missing_contact",

convex/leadDiscovery.ts

@@ -5,13 +5,18 @@ import {
   buildGeocodingUrl,
   getBlacklistLookupValues,
   getBlacklistMatches,
+  getCandidateEmailValues,
   getPlacesSearchSpec,
+  normalizeDomain,
+  normalizePhone,
+  normalizeText,
   normalizePlacesResponse,
   parseGeocodingResponse,
 } from "../lib/lead-discovery-google";
 import {
   buildLeadDiscoveryLeadRecord,
   buildLeadDiscoveryCounters,
+  getLeadDiscoveryPriority,
 } from "../lib/lead-discovery-run";
 import { calculateNextRunAt } from "../lib/campaign-scheduling";
 
@@ -37,6 +42,20 @@ const candidateValidator = v.object({
   googleTypes: v.array(v.string()),
   googlePrimaryType: nullableString,
   googleMapsUrl: nullableString,
+  email: v.optional(nullableString),
+  emailSource: v.optional(nullableString),
+  contactPerson: v.optional(nullableString),
+  isBusinessContactAddress: v.optional(v.boolean()),
+  contactEmails: v.optional(
+    v.array(
+      v.object({
+        email: v.string(),
+        emailSource: v.optional(nullableString),
+        contactPerson: v.optional(nullableString),
+        isBusinessContactAddress: v.optional(v.boolean()),
+      }),
+    ),
+  ),
   sourceProvider: v.literal("google_places"),
   sourceFetchedAt: v.number(),
 });
@@ -396,23 +415,43 @@ export const persistDiscoveredLeads = internalMutation({
         continue;
       }
 
-      const existingByPlaceId = await ctx.db
-        .query("leads")
-        .withIndex("by_googlePlaceId", (q) =>
-          q.eq("googlePlaceId", candidate.placeId),
-        )
-        .take(1);
-      const candidateDomain = candidate.websiteDomain;
-      const existingByDomain = candidateDomain
+      const normalizedPlaceId = normalizeDomain(candidate.placeId);
+      const normalizedDomain = normalizeDomain(candidate.websiteDomain);
+      const normalizedEmails = getCandidateEmailValues(candidate);
+      const normalizedPhone = normalizePhone(candidate.phone);
+      const normalizedCompanyName = normalizeText(candidate.businessName);
+      const normalizedAddress = normalizeText(candidate.address);
+
+      const duplicateByPlaceId = normalizedPlaceId
         ? await ctx.db
             .query("leads")
-            .withIndex("by_websiteDomain", (q) =>
-              q.eq("websiteDomain", candidateDomain),
+            .withIndex("by_normalizedGooglePlaceId", (q) =>
+              q.eq("normalizedGooglePlaceId", normalizedPlaceId),
             )
             .take(1)
         : [];
 
-      if (existingByPlaceId.length > 0 || existingByDomain.length > 0) {
+      const duplicateByDomain = normalizedDomain
+        ? await ctx.db
+            .query("leads")
+            .withIndex("by_websiteDomain", (q) => q.eq("websiteDomain", normalizedDomain))
+            .take(1)
+        : [];
+
+      const duplicateByEmailRows = [];
+      for (const email of normalizedEmails) {
+        const rows = await ctx.db
+          .query("leads")
+          .withIndex("by_normalizedEmail", (q) => q.eq("normalizedEmail", email))
+          .take(1);
+        duplicateByEmailRows.push(...rows);
+      }
+
+      if (
+        duplicateByPlaceId.length > 0 ||
+        duplicateByDomain.length > 0 ||
+        duplicateByEmailRows.length > 0
+      ) {
         skippedDuplicates += 1;
         await ctx.db.insert("agentRunEvents", {
           runId: args.runId,
@@ -427,6 +466,29 @@ export const persistDiscoveredLeads = internalMutation({
         continue;
       }
 
+      const probableDuplicateByPhone = normalizedPhone
+        ? await ctx.db
+            .query("leads")
+            .withIndex("by_normalizedPhone", (q) =>
+              q.eq("normalizedPhone", normalizedPhone),
+            )
+            .take(1)
+        : [];
+
+      const probableDuplicateByAddress = normalizedCompanyName && normalizedAddress
+        ? await ctx.db
+            .query("leads")
+            .withIndex("by_normalizedCompanyName_and_normalizedAddress", (q) =>
+              q
+                .eq("normalizedCompanyName", normalizedCompanyName)
+                .eq("normalizedAddress", normalizedAddress),
+            )
+            .take(1)
+        : [];
+
+      const probableDuplicateLead =
+        probableDuplicateByPhone[0] ?? probableDuplicateByAddress[0] ?? null;
+
       const blacklistRows = [];
       for (const lookup of getBlacklistLookupValues(candidate)) {
         const rows = await ctx.db
@@ -465,6 +527,34 @@ export const persistDiscoveredLeads = internalMutation({
         candidate,
         now,
       });
+      const hasWebsite = Boolean(candidate.websiteUrl ?? candidate.websiteDomain);
+      const priorityResult = getLeadDiscoveryPriority({
+        isDuplicate: !!probableDuplicateLead,
+        hasWebsite,
+        hasWebsiteSignal: false, // plain Google-Places website hint maps to medium priority.
+      });
+      const isDuplicateCandidate = !!probableDuplicateLead;
+
+      if (normalizedPlaceId) {
+        lead.normalizedGooglePlaceId = normalizedPlaceId;
+      }
+      if (normalizedPhone !== "") {
+        lead.normalizedPhone = normalizedPhone;
+      }
+      if (normalizedCompanyName !== "") {
+        lead.normalizedCompanyName = normalizedCompanyName;
+      }
+      if (normalizedAddress !== "") {
+        lead.normalizedAddress = normalizedAddress;
+      }
+      lead.priority = priorityResult.priority;
+      lead.priorityReason = priorityResult.reason;
+
+      if (isDuplicateCandidate) {
+        lead.duplicateStatus = "possible_duplicate";
+        lead.duplicateReason = `Möglicher Dublettenkandidat zu Lead ${probableDuplicateLead._id}`;
+        lead.duplicateOfLeadId = probableDuplicateLead._id;
+      }
 
       await ctx.db.insert("leads", lead);
       leadsCreated += 1;

convex/leads.ts

@@ -1,8 +1,93 @@
 import { v } from "convex/values";
 
+import { getUsableContactEmailFromEntries } from "../lib/lead-discovery-google";
 import { normalizeListLimit } from "./domain";
+import type { Doc, Id } from "./_generated/dataModel";
 import { mutation, query } from "./_generated/server";
 
+type LeadDoc = Doc<"leads">;
+
+type LeadReviewContactPatch = {
+  email: string;
+  normalizedEmail: string;
+  emailSource?: string;
+  contactPerson?: string;
+};
+
+type BuildReviewContactPatchResult = {
+  patch?: LeadReviewContactPatch;
+  setContactStatus?: LeadDoc["contactStatus"];
+};
+
+type LeadReviewPatch = {
+  updatedAt: number;
+  priority?: LeadDoc["priority"];
+  priorityReason?: string;
+  contactStatus?: LeadDoc["contactStatus"];
+  contactStatusReason?: string;
+  notes?: string;
+  duplicateStatus?: LeadDoc["duplicateStatus"];
+  duplicateReason?: string;
+  duplicateOfLeadId?: Id<"leads">;
+  blacklistStatus?: LeadDoc["blacklistStatus"];
+  blacklistReason?: string;
+  email?: string;
+  normalizedEmail?: string;
+  emailSource?: string;
+  contactPerson?: string;
+};
+
+function buildReviewContactPatch(args: {
+  email?: string;
+  emailSource?: string;
+  contactPerson?: string;
+  isBusinessContactAddress?: boolean;
+  explicitContactStatus?: boolean;
+  currentContactStatus?: "new" | "missing_contact" | "audit_ready" | "outreach_ready" | "contacted" | "replied" | "do_not_contact";
+}): BuildReviewContactPatchResult | null {
+  if (args.email === undefined) {
+    return null;
+  }
+
+  const usable = getUsableContactEmailFromEntries([
+    {
+      email: args.email,
+      emailSource: args.emailSource,
+      contactPerson: args.contactPerson,
+      isBusinessContactAddress: args.isBusinessContactAddress,
+    },
+  ]);
+
+  if (!usable) {
+    return {
+      setContactStatus: "missing_contact",
+    };
+  }
+
+  const patch: LeadReviewContactPatch = {
+    email: usable.email,
+    normalizedEmail: usable.email,
+  };
+
+  if (usable.emailSource !== null) {
+    patch.emailSource = usable.emailSource;
+  }
+
+  if (usable.contactPerson !== null) {
+    patch.contactPerson = usable.contactPerson;
+  }
+
+  const setContactStatus =
+    !args.explicitContactStatus && args.currentContactStatus === "missing_contact"
+      ? "new"
+      : undefined;
+
+  return ({
+    patch,
+    setContactStatus,
+  });
+}
+
 export const create = mutation({
   args: {
     campaignId: v.optional(v.id("campaigns")),
@@ -24,6 +109,10 @@ export const create = mutation({
     websiteUrl: v.optional(v.string()),
     websiteDomain: v.optional(v.string()),
     phone: v.optional(v.string()),
+    normalizedEmail: v.optional(v.string()),
+    normalizedPhone: v.optional(v.string()),
+    normalizedCompanyName: v.optional(v.string()),
+    normalizedAddress: v.optional(v.string()),
     email: v.optional(v.string()),
     emailSource: v.optional(v.string()),
     contactPerson: v.optional(v.string()),
@@ -33,8 +122,10 @@ export const create = mutation({
         v.literal("medium"),
         v.literal("low"),
         v.literal("defer"),
+        v.literal("blocked"),
       ),
     ),
+    priorityReason: v.optional(v.string()),
     contactStatus: v.optional(
       v.union(
         v.literal("new"),
@@ -46,6 +137,20 @@ export const create = mutation({
         v.literal("do_not_contact"),
       ),
     ),
+    contactStatusReason: v.optional(v.string()),
+    duplicateStatus: v.optional(
+      v.union(
+        v.literal("unchecked"),
+        v.literal("unique"),
+        v.literal("possible_duplicate"),
+        v.literal("duplicate"),
+      ),
+    ),
+    duplicateReason: v.optional(v.string()),
+    blacklistReason: v.optional(v.string()),
+    duplicateOfLeadId: v.optional(v.id("leads")),
+    blacklistStatus: v.optional(v.union(v.literal("clear"), v.literal("blocked"))),
+    normalizedGooglePlaceId: v.optional(v.string()),
     notes: v.optional(v.string()),
   },
   handler: async (ctx, args) => {
@@ -53,16 +158,151 @@ export const create = mutation({
 
     return await ctx.db.insert("leads", {
       ...args,
+      normalizedEmail: args.normalizedEmail,
+      normalizedPhone: args.normalizedPhone,
+      normalizedCompanyName: args.normalizedCompanyName,
+      normalizedAddress: args.normalizedAddress,
+      normalizedGooglePlaceId: args.normalizedGooglePlaceId,
       priority: args.priority ?? "medium",
       contactStatus: args.contactStatus ?? "new",
-      duplicateStatus: "unchecked",
-      blacklistStatus: "clear",
+      duplicateStatus: args.duplicateStatus ?? "unchecked",
+      blacklistStatus: args.blacklistStatus ?? "clear",
       createdAt: now,
       updatedAt: now,
     });
   },
 });
 
+export const reviewUpdate = mutation({
+  args: {
+    id: v.id("leads"),
+    priority: v.optional(
+      v.union(
+        v.literal("high"),
+        v.literal("medium"),
+        v.literal("low"),
+        v.literal("defer"),
+        v.literal("blocked"),
+      ),
+    ),
+    priorityReason: v.optional(v.string()),
+    contactStatus: v.optional(
+      v.union(
+        v.literal("new"),
+        v.literal("missing_contact"),
+        v.literal("audit_ready"),
+        v.literal("outreach_ready"),
+        v.literal("contacted"),
+        v.literal("replied"),
+        v.literal("do_not_contact"),
+      ),
+    ),
+    contactStatusReason: v.optional(v.string()),
+    notes: v.optional(v.string()),
+    duplicateStatus: v.optional(
+      v.union(
+        v.literal("unchecked"),
+        v.literal("unique"),
+        v.literal("possible_duplicate"),
+        v.literal("duplicate"),
+      ),
+    ),
+    duplicateReason: v.optional(v.string()),
+    blacklistStatus: v.optional(v.union(v.literal("clear"), v.literal("blocked"))),
+    blacklistReason: v.optional(v.string()),
+    duplicateOfLeadId: v.optional(v.id("leads")),
+    applyBlacklist: v.optional(v.boolean()),
+    reviewEmail: v.optional(v.string()),
+    reviewEmailSource: v.optional(v.string()),
+    reviewContactPerson: v.optional(v.string()),
+    reviewIsBusinessContactAddress: v.optional(v.boolean()),
+  },
+  handler: async (ctx, args) => {
+    const lead = await ctx.db.get(args.id);
+
+    if (!lead) {
+      return null;
+    }
+
+    const now = Date.now();
+    const patch: LeadReviewPatch = {
+      updatedAt: now,
+    };
+
+    if (args.priority !== undefined) {
+      patch.priority = args.priority;
+    }
+    if (args.priorityReason !== undefined) {
+      patch.priorityReason = args.priorityReason;
+    }
+    if (args.contactStatus !== undefined) {
+      patch.contactStatus = args.contactStatus;
+    }
+    if (args.contactStatusReason !== undefined) {
+      patch.contactStatusReason = args.contactStatusReason;
+    }
+    if (args.notes !== undefined) {
+      patch.notes = args.notes;
+    }
+    if (args.duplicateStatus !== undefined) {
+      patch.duplicateStatus = args.duplicateStatus;
+    }
+    if (args.duplicateReason !== undefined) {
+      patch.duplicateReason = args.duplicateReason;
+    }
+    if (args.duplicateOfLeadId !== undefined) {
+      patch.duplicateOfLeadId = args.duplicateOfLeadId;
+    }
+
+    if (args.applyBlacklist) {
+      patch.blacklistStatus = "blocked";
+      if (args.blacklistReason !== undefined) {
+        patch.blacklistReason = args.blacklistReason;
+      } else if (lead.blacklistReason === undefined) {
+        patch.blacklistReason = "Manuell in der Review als Sperrgrund gesetzt.";
+      }
+      if (args.priority === undefined || args.priority !== "blocked") {
+        patch.priority = "blocked";
+      }
+    } else if (args.applyBlacklist === false && args.blacklistStatus !== undefined) {
+      patch.blacklistStatus = args.blacklistStatus;
+      patch.blacklistReason = args.blacklistReason;
+    } else if (args.blacklistStatus !== undefined) {
+      patch.blacklistStatus = args.blacklistStatus;
+      patch.blacklistReason = args.blacklistReason;
+    }
+
+    const reviewContactPatch = buildReviewContactPatch({
+      email: args.reviewEmail,
+      emailSource: args.reviewEmailSource,
+      contactPerson: args.reviewContactPerson,
+      isBusinessContactAddress: args.reviewIsBusinessContactAddress,
+      explicitContactStatus: args.contactStatus !== undefined,
+      currentContactStatus: lead.contactStatus,
+    });
+
+    if (reviewContactPatch?.patch) {
+      Object.assign(patch, reviewContactPatch.patch);
+    }
+
+    if (
+      reviewContactPatch !== null &&
+      reviewContactPatch.setContactStatus !== undefined &&
+      args.contactStatus === undefined
+    ) {
+      patch.contactStatus = reviewContactPatch.setContactStatus;
+    }
+
+    if (args.blacklistReason !== undefined && patch.blacklistStatus === undefined) {
+      patch.blacklistStatus = "blocked";
+      patch.blacklistReason = args.blacklistReason;
+    }
+
+    await ctx.db.patch(args.id, patch);
+    return args.id;
+  },
+});
+
 export const get = query({
   args: { id: v.id("leads") },
   handler: async (ctx, args) => {

convex/schema.ts

@@ -8,6 +8,7 @@ const leadPriority = v.union(
   v.literal("medium"),
   v.literal("low"),
   v.literal("defer"),
+  v.literal("blocked"),
 );
 const leadContactStatus = v.union(
   v.literal("new"),
@@ -158,6 +159,7 @@ export default defineSchema({
     city: v.optional(v.string()),
     postalCode: v.optional(v.string()),
     googlePlaceId: v.optional(v.string()),
+    normalizedGooglePlaceId: v.optional(v.string()),
     googleMapsUrl: v.optional(v.string()),
     googlePrimaryType: v.optional(v.string()),
     googleTypes: v.optional(v.array(v.string())),
@@ -169,9 +171,18 @@ export default defineSchema({
     websiteUrl: v.optional(v.string()),
     websiteDomain: v.optional(v.string()),
     phone: v.optional(v.string()),
+    normalizedEmail: v.optional(v.string()),
+    normalizedPhone: v.optional(v.string()),
+    normalizedCompanyName: v.optional(v.string()),
+    normalizedAddress: v.optional(v.string()),
     email: v.optional(v.string()),
     emailSource: v.optional(v.string()),
     contactPerson: v.optional(v.string()),
+    priorityReason: v.optional(v.string()),
+    contactStatusReason: v.optional(v.string()),
+    duplicateReason: v.optional(v.string()),
+    blacklistReason: v.optional(v.string()),
+    duplicateOfLeadId: v.optional(v.id("leads")),
     priority: leadPriority,
     contactStatus: leadContactStatus,
     duplicateStatus: leadDuplicateStatus,
@@ -183,8 +194,16 @@ export default defineSchema({
     .index("by_campaignId", ["campaignId"])
     .index("by_discoveryRunId", ["discoveryRunId"])
     .index("by_contactStatus", ["contactStatus"])
+    .index("by_normalizedEmail", ["normalizedEmail"])
+    .index("by_normalizedPhone", ["normalizedPhone"])
+    .index("by_normalizedCompanyName_and_normalizedAddress", [
+      "normalizedCompanyName",
+      "normalizedAddress",
+    ])
+    .index("by_normalizedGooglePlaceId", ["normalizedGooglePlaceId"])
     .index("by_googlePlaceId", ["googlePlaceId"])
     .index("by_websiteDomain", ["websiteDomain"])
+    .index("by_normalizedCompanyName", ["normalizedCompanyName"])
     .index("by_priority_and_contactStatus", ["priority", "contactStatus"]),
 
   audits: defineTable({