Add campaign scheduling lifecycle jobs

2026-06-05 21:38:34 +02:00
parent 3f148bcec2
commit 70951789d2
7 changed files with 384 additions and 0 deletions
--- a/components/campaigns/campaigns-board.tsx
+++ b/components/campaigns/campaigns-board.tsx
@@ -16,7 +16,9 @@ import { Separator } from "@/components/ui/separator";
 import { CampaignFormDialog } from "@/components/campaigns/campaign-form-dialog";

 type CampaignsListResult = FunctionReturnType<typeof api.campaigns.list>;
+type CampaignRunsListResult = FunctionReturnType<typeof api.runs.list>;
 type CampaignRow = NonNullable<CampaignsListResult>[number];
+type CampaignRunRow = NonNullable<CampaignRunsListResult>[number];

 type RecurrenceLabel = Record<CampaignRow["recurrence"], string>;
 type CurrentRunStatusLabel = {
@@ -40,6 +42,13 @@ const statusLabel: CurrentRunStatusLabel = {
  paused: "Pausiert",
 };

+const stepLabel: Record<string, string> = {
+  campaign_cron_queued: "Cron geplant",
+  campaign_cron_skipped: "Cron übersprungen",
+  campaign_cron_stale_pending: "Timeout bereinigt",
+  lead_discovery: "Lead-Recherche",
+};
+
 const dateFormatter = new Intl.DateTimeFormat("de-DE", {
  dateStyle: "short",
  timeStyle: "short",
@@ -84,6 +93,10 @@ const formatNiche = (campaign: CampaignRow): string => {

 export function CampaignsBoard() {
  const campaigns = useQuery(api.campaigns.list, { limit: 100 });
+  const recentCampaignRuns = useQuery(api.runs.list, {
+    limit: 8,
+    type: "campaign",
+  });
  const createCampaign = useMutation(api.campaigns.create);
  const updateCampaign = useMutation(api.campaigns.update);
  const setStatus = useMutation(api.campaigns.setStatus);
@@ -130,6 +143,10 @@ export function CampaignsBoard() {
    return [...campaigns].sort((a, b) => b.createdAt - a.createdAt);
  }, [campaigns]);

+  const visibleRuns = useMemo<CampaignRunRow[]>(() => {
+    return recentCampaignRuns ?? [];
+  }, [recentCampaignRuns]);
+
  const closeDialog = () => {
    setEditingCampaign(null);
    setIsFormOpen(false);
@@ -343,6 +360,48 @@ export function CampaignsBoard() {
          ))}
        </div>
      )}
+
+      <Card>
+        <CardHeader>
+          <CardTitle>Aktuelle Run-Logs</CardTitle>
+          <CardDescription>
+            Letzte Kampagnenläufe inklusive Cron-Skips und Fehlerhinweisen.
+          </CardDescription>
+        </CardHeader>
+        <CardContent className="grid gap-2 text-sm">
+          {recentCampaignRuns === undefined ? (
+            <Skeleton className="h-16 rounded-lg" />
+          ) : visibleRuns.length === 0 ? (
+            <p className="text-muted-foreground">Noch keine Kampagnenläufe.</p>
+          ) : (
+            visibleRuns.map((run) => (
+              <div className="rounded-md border p-3" key={run._id}>
+                <div className="flex flex-wrap items-center justify-between gap-2">
+                  <p className="font-medium">
+                    {statusLabel[run.status] ?? run.status}
+                  </p>
+                  <p className="text-xs text-muted-foreground">
+                    {formatDateTime(run.updatedAt)}
+                  </p>
+                </div>
+                <p className="mt-1 text-muted-foreground">
+                  {stepLabel[run.currentStep ?? ""] ?? run.currentStep ?? "Schritt offen"}
+                </p>
+                {run.currentStep === "campaign_cron_skipped" ? (
+                  <p className="mt-1 text-xs text-muted-foreground">
+                    Cron wurde übersprungen, weil bereits ein Agentenlauf aktiv war.
+                  </p>
+                ) : null}
+                {run.errorSummary ? (
+                  <p className="mt-1 text-xs text-destructive">
+                    {run.errorSummary}
+                  </p>
+                ) : null}
+              </div>
+            ))
+          )}
+        </CardContent>
+      </Card>
    </section>
  );
 }
--- a/convex/audits.ts
+++ b/convex/audits.ts
@@ -4,6 +4,8 @@ import { normalizeListLimit } from "./domain";
 import { internalMutation, mutation, query } from "./_generated/server";
 import type { MutationCtx, QueryCtx } from "./_generated/server";

+export const AUDIT_REVIEW_NOTICE_AFTER_MS = 30 * 24 * 60 * 60 * 1000;
+
 const auditStatus = v.union(
  v.literal("draft"),
  v.literal("approved"),
@@ -352,6 +354,9 @@ export const publishPublicAudit = mutation({
    await ctx.db.patch(args.id, {
      status: "published",
      publishedAt: now,
+      reviewDueAt: now + AUDIT_REVIEW_NOTICE_AFTER_MS,
+      lifecycleNotificationAt: undefined,
+      lifecycleExtendedUntil: undefined,
      deactivatedAt: undefined,
      updatedAt: now,
    });
@@ -373,6 +378,34 @@ export const reapprovePublicAudit = mutation({
    await ctx.db.patch(args.id, {
      status: "published",
      publishedAt: now,
+      reviewDueAt: now + AUDIT_REVIEW_NOTICE_AFTER_MS,
+      lifecycleNotificationAt: undefined,
+      lifecycleExtendedUntil: undefined,
+      deactivatedAt: undefined,
+      updatedAt: now,
+    });
+
+    return { slug: audit.slug };
+  },
+});
+
+export const extendPublicAuditLifecycle = mutation({
+  args: {
+    id: v.id("audits"),
+    days: v.number(),
+  },
+  handler: async (ctx, args) => {
+    await requireOperator(ctx);
+    const audit = await ctx.db.get(args.id);
+    if (!audit) {
+      throw new Error("Audit wurde nicht gefunden.");
+    }
+
+    const now = Date.now();
+    await ctx.db.patch(args.id, {
+      status: "published",
+      lifecycleExtendedUntil: now + args.days * 24 * 60 * 60 * 1000,
+      reviewDueAt: now + args.days * 24 * 60 * 60 * 1000,
      deactivatedAt: undefined,
      updatedAt: now,
    });
--- a/convex/crons.ts
+++ b/convex/crons.ts
@@ -0,0 +1,19 @@
+import { cronJobs } from "convex/server";
+
+import { internal } from "./_generated/api";
+
+const crons = cronJobs();
+
+crons.interval(
+  "Kampagnen nach Cadence starten",
+  { hours: 1 },
+  internal.scheduledJobs.runDueCampaigns,
+);
+
+crons.interval(
+  "Audit-Lifecycle prüfen",
+  { hours: 24 },
+  internal.scheduledJobs.runAuditLifecycle,
+);
+
+export default crons;
--- a/convex/scheduledJobs.ts
+++ b/convex/scheduledJobs.ts
@@ -0,0 +1,199 @@
+import { internal } from "./_generated/api";
+import { internalMutation } from "./_generated/server";
+import { canStartAgentRun, isStalePendingAgentRun } from "../lib/lead-discovery-run";
+
+export const AUDIT_REVIEW_NOTICE_AFTER_MS = 30 * 24 * 60 * 60 * 1000;
+export const AUDIT_AUTO_DEACTIVATE_AFTER_MS = 60 * 24 * 60 * 60 * 1000;
+
+const RUN_COUNTERS_ZERO = {
+  leadsFound: 0,
+  leadsCreated: 0,
+  auditsCreated: 0,
+  outreachPrepared: 0,
+  errors: 0,
+};
+
+export const runDueCampaigns = internalMutation({
+  args: {},
+  handler: async (ctx) => {
+    const now = Date.now();
+    const activeRuns = [
+      ...(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(20)),
+    ];
+
+    for (const run of activeRuns.filter((run) => isStalePendingAgentRun(run, now))) {
+      await ctx.db.patch(run._id, {
+        status: "canceled",
+        currentStep: "campaign_cron_stale_pending",
+        errorSummary: "Ausstehender Lauf wurde nach Timeout automatisch abgebrochen.",
+        finishedAt: now,
+        updatedAt: now,
+      });
+      await ctx.db.insert("agentRunEvents", {
+        runId: run._id,
+        level: "warning",
+        message: "Ausstehender Lauf wurde nach Timeout automatisch abgebrochen.",
+        createdAt: now,
+      });
+    }
+
+    if (!canStartAgentRun(activeRuns, now)) {
+      const skippedRunId = await ctx.db.insert("agentRuns", {
+        type: "campaign",
+        status: "canceled",
+        currentStep: "campaign_cron_skipped",
+        errorSummary: "Es läuft bereits ein Agentenlauf.",
+        counters: RUN_COUNTERS_ZERO,
+        createdAt: now,
+        updatedAt: now,
+        finishedAt: now,
+      });
+
+      await ctx.db.insert("agentRunEvents", {
+        runId: skippedRunId,
+        level: "warning",
+        message: "Es läuft bereits ein Agentenlauf. Kampagnen-Cron wurde übersprungen.",
+        createdAt: now,
+      });
+
+      return { started: 0, skipped: 1 };
+    }
+
+    const dueCampaigns = await ctx.db
+      .query("campaigns")
+      .withIndex("by_status_and_nextRunAt", (q) =>
+        q.eq("status", "active").lte("nextRunAt", now),
+      )
+      .take(1);
+    const campaign = dueCampaigns[0];
+
+    if (!campaign || campaign.recurrence === "manual") {
+      return { started: 0, skipped: 0 };
+    }
+
+    const runId = await ctx.db.insert("agentRuns", {
+      type: "campaign",
+      campaignId: campaign._id,
+      status: "pending",
+      currentStep: "campaign_cron_queued",
+      counters: RUN_COUNTERS_ZERO,
+      createdAt: now,
+      updatedAt: now,
+    });
+
+    await ctx.db.insert("agentRunEvents", {
+      runId,
+      level: "info",
+      message: "Kampagnenlauf wurde durch Cadence-Cron geplant.",
+      details: [{ label: "Kampagne", value: campaign.name }],
+      createdAt: now,
+    });
+
+    await ctx.scheduler.runAfter(0, internal.leadDiscovery.processCampaignRun, {
+      runId,
+    });
+
+    return { started: 1, skipped: 0 };
+  },
+});
+
+export const runAuditLifecycle = internalMutation({
+  args: {},
+  handler: async (ctx) => {
+    const now = Date.now();
+    const runId = await ctx.db.insert("agentRuns", {
+      type: "lifecycle",
+      status: "running",
+      currentStep: "audit_lifecycle",
+      counters: RUN_COUNTERS_ZERO,
+      startedAt: now,
+      createdAt: now,
+      updatedAt: now,
+    });
+    let notifications = 0;
+    let deactivated = 0;
+
+    const publishedAudits = await ctx.db
+      .query("audits")
+      .withIndex("by_status", (q) => q.eq("status", "published"))
+      .take(100);
+
+    for (const audit of publishedAudits) {
+      const publishedAt = audit.publishedAt ?? audit.updatedAt;
+      const extendedUntil = audit.lifecycleExtendedUntil ?? 0;
+      const isExtended = extendedUntil > now;
+
+      if (!isExtended && now - publishedAt >= AUDIT_AUTO_DEACTIVATE_AFTER_MS) {
+        await ctx.db.patch(audit._id, {
+          status: "deactivated",
+          deactivatedAt: now,
+          updatedAt: now,
+        });
+        await ctx.db.insert("dashboardNotifications", {
+          auditId: audit._id,
+          runId,
+          kind: "audit_auto_deactivated",
+          title: "Audit automatisch deaktiviert",
+          message: "Ein veröffentlichtes Audit war älter als 60 Tage und wurde deaktiviert.",
+          status: "unread",
+          createdAt: now,
+          updatedAt: now,
+        });
+        deactivated += 1;
+        continue;
+      }
+
+      if (
+        !audit.lifecycleNotificationAt &&
+        now - publishedAt >= AUDIT_REVIEW_NOTICE_AFTER_MS
+      ) {
+        await ctx.db.patch(audit._id, {
+          lifecycleNotificationAt: now,
+          reviewDueAt: now,
+          updatedAt: now,
+        });
+        await ctx.db.insert("dashboardNotifications", {
+          auditId: audit._id,
+          runId,
+          kind: "audit_review_due",
+          title: "Audit-Aktivität prüfen",
+          message: "Soll dieses Audit aktiv bleiben? Es ist seit 30 Tagen veröffentlicht.",
+          status: "unread",
+          createdAt: now,
+          updatedAt: now,
+        });
+        notifications += 1;
+      }
+    }
+
+    await ctx.db.patch(runId, {
+      status: "succeeded",
+      finishedAt: now,
+      updatedAt: now,
+      counters: {
+        ...RUN_COUNTERS_ZERO,
+        auditsCreated: notifications,
+        errors: deactivated,
+      },
+    });
+    await ctx.db.insert("agentRunEvents", {
+      runId,
+      level: "info",
+      message: "Audit-Lifecycle geprüft.",
+      details: [
+        { label: "Hinweise", value: String(notifications) },
+        { label: "Deaktiviert", value: String(deactivated) },
+      ],
+      createdAt: now,
+    });
+
+    return { notifications, deactivated };
+  },
+});
--- a/convex/schema.ts
+++ b/convex/schema.ts
@@ -307,6 +307,8 @@ export default defineSchema({
    ctaType: v.optional(v.string()),
    publishedAt: v.optional(v.number()),
    reviewDueAt: v.optional(v.number()),
+    lifecycleNotificationAt: v.optional(v.number()),
+    lifecycleExtendedUntil: v.optional(v.number()),
    deactivatedAt: v.optional(v.number()),
    createdAt: v.number(),
    updatedAt: v.number(),
@@ -592,4 +594,18 @@ export default defineSchema({
    createdAt: v.number(),
    updatedAt: v.number(),
  }).index("by_key", ["key"]),
+
+  dashboardNotifications: defineTable({
+    auditId: v.optional(v.id("audits")),
+    runId: v.optional(v.id("agentRuns")),
+    kind: v.string(),
+    title: v.string(),
+    message: v.string(),
+    status: v.union(v.literal("unread"), v.literal("acknowledged")),
+    createdAt: v.number(),
+    updatedAt: v.number(),
+  })
+    .index("by_status_and_createdAt", ["status", "createdAt"])
+    .index("by_auditId", ["auditId"])
+    .index("by_runId", ["runId"]),
 });
--- a/tests/campaigns-board-layout.test.ts
+++ b/tests/campaigns-board-layout.test.ts
@@ -28,3 +28,13 @@ test("campaign board renders campaigns as responsive cards", async () => {
  assert.match(source, /toggleCampaign\(campaign\)/);
  assert.match(source, /runCampaign\(campaign\)/);
 });
+
+test("campaign board surfaces recent run logs", async () => {
+  const source = await readFile(campaignsBoardPath, "utf8");
+
+  assert.match(source, /api\.runs\.list/);
+  assert.match(source, /type:\s*"campaign"/);
+  assert.match(source, /Aktuelle Run-Logs/);
+  assert.match(source, /run\.errorSummary/);
+  assert.match(source, /campaign_cron_skipped/);
+});
--- a/tests/scheduled-jobs-source.test.ts
+++ b/tests/scheduled-jobs-source.test.ts
@@ -0,0 +1,48 @@
+import assert from "node:assert/strict";
+import { existsSync, readFileSync } from "node:fs";
+import { join } from "node:path";
+import test from "node:test";
+
+const cronsPath = join(process.cwd(), "convex", "crons.ts");
+const jobsPath = join(process.cwd(), "convex", "scheduledJobs.ts");
+const auditsPath = join(process.cwd(), "convex", "audits.ts");
+const schemaPath = join(process.cwd(), "convex", "schema.ts");
+
+const cronsSource = existsSync(cronsPath) ? readFileSync(cronsPath, "utf8") : "";
+const jobsSource = existsSync(jobsPath) ? readFileSync(jobsPath, "utf8") : "";
+const auditsSource = readFileSync(auditsPath, "utf8");
+const schemaSource = readFileSync(schemaPath, "utf8");
+
+test("Convex crons register campaign cadence and audit lifecycle jobs", () => {
+  assert.equal(existsSync(cronsPath), true, "convex/crons.ts should exist.");
+  assert.match(cronsSource, /cronJobs\(\)/);
+  assert.match(cronsSource, /internal\.scheduledJobs\.runDueCampaigns/);
+  assert.match(cronsSource, /internal\.scheduledJobs\.runAuditLifecycle/);
+});
+
+test("scheduled jobs guard single active agent run and log skipped campaign cadence", () => {
+  assert.equal(existsSync(jobsPath), true, "convex/scheduledJobs.ts should exist.");
+  assert.match(jobsSource, /export const runDueCampaigns = internalMutation/);
+  assert.match(jobsSource, /canStartAgentRun/);
+  assert.match(jobsSource, /status:\s*"canceled"[\s\S]*currentStep:\s*"campaign_cron_skipped"/);
+  assert.match(jobsSource, /Es läuft bereits ein Agentenlauf/);
+  assert.match(jobsSource, /ctx\.scheduler\.runAfter\(0,\s*internal\.leadDiscovery\.processCampaignRun/);
+});
+
+test("audit lifecycle stores notifications and deactivates old published audits", () => {
+  assert.match(schemaSource, /dashboardNotifications:\s*defineTable/);
+  assert.match(schemaSource, /lifecycleNotificationAt:\s*v\.optional\(v\.number\(\)\)/);
+  assert.match(schemaSource, /lifecycleExtendedUntil:\s*v\.optional\(v\.number\(\)\)/);
+  assert.match(jobsSource, /export const runAuditLifecycle = internalMutation/);
+  assert.match(jobsSource, /AUDIT_REVIEW_NOTICE_AFTER_MS/);
+  assert.match(jobsSource, /AUDIT_AUTO_DEACTIVATE_AFTER_MS/);
+  assert.match(jobsSource, /status:\s*"deactivated"/);
+  assert.match(jobsSource, /Soll dieses Audit aktiv bleiben/);
+});
+
+test("publishing and lifecycle controls set review due dates and allow manual extension", () => {
+  assert.match(auditsSource, /reviewDueAt:\s*now\s*\+\s*AUDIT_REVIEW_NOTICE_AFTER_MS/);
+  assert.match(auditsSource, /export const extendPublicAuditLifecycle = mutation/);
+  assert.match(auditsSource, /lifecycleExtendedUntil:\s*now\s*\+\s*args\.days\s*\*\s*24\s*\*\s*60\s*\*\s*60\s*\*\s*1000/);
+  assert.match(auditsSource, /status:\s*"published"[\s\S]*deactivatedAt:\s*undefined/);
+});