feat: integrate google lead discovery

convex/campaigns.ts

@@ -13,8 +13,10 @@ import {
   validateCampaignCreateInput,
   validateCampaignUpdateInput,
 } from "../lib/campaign-validation";
+import { canStartAgentRun, isStalePendingAgentRun } from "../lib/lead-discovery-run";
 
 import { normalizeListLimit } from "./domain";
+import { internal } from "./_generated/api";
 import { Doc } from "./_generated/dataModel";
 import { mutation, query, QueryCtx } from "./_generated/server";
 
@@ -284,6 +286,41 @@ export const requestRun = mutation({
       throw new Error("Kampagne nicht gefunden.");
     }
 
+    const possiblyActiveRuns = [
+      ...(await ctx.db
+        .query("agentRuns")
+        .withIndex("by_status", (q) => q.eq("status", "pending"))
+        .take(20)),
+      ...(await ctx.db
+        .query("agentRuns")
+        .withIndex("by_status", (q) => q.eq("status", "running"))
+        .take(1)),
+    ];
+    const stalePendingRuns = possiblyActiveRuns.filter((run) =>
+      isStalePendingAgentRun(run, now),
+    );
+
+    for (const staleRun of stalePendingRuns) {
+      await ctx.db.patch(staleRun._id, {
+        status: "canceled",
+        currentStep: "lead_discovery",
+        errorSummary: "Ausstehender Lauf wurde nach Timeout automatisch abgebrochen.",
+        finishedAt: now,
+        updatedAt: now,
+      });
+      await ctx.db.insert("agentRunEvents", {
+        runId: staleRun._id,
+        level: "warning",
+        message: "Ausstehender Lauf wurde nach Timeout automatisch abgebrochen.",
+        details: [{ label: "Alter Status", value: "pending" }],
+        createdAt: now,
+      });
+    }
+
+    if (!canStartAgentRun(possiblyActiveRuns, now)) {
+      throw new Error("Es läuft bereits ein Agentenlauf.");
+    }
+
     const runId = await ctx.db.insert("agentRuns", {
       type: "campaign",
       campaignId: args.id,
@@ -299,17 +336,8 @@ export const requestRun = mutation({
       updatedAt: now,
     });
 
-    const nextRunAt = calculateNextRunAt({
-      recurrence: campaign.recurrence,
-      status: campaign.status,
-      lastRunAt: now,
-      now,
-    });
-
-    await ctx.db.patch(args.id, {
-      lastRunAt: now,
-      nextRunAt,
-      updatedAt: now,
+    await ctx.scheduler.runAfter(0, internal.leadDiscovery.processCampaignRun, {
+      runId,
     });
 
     return runId;