Agents & AI

// the agent's tool — any process holding an ingest-scoped key
import { Hogsend } from "@hogsend/client";

const hs = new Hogsend({
  baseUrl: process.env.HOGSEND_BASE_URL!,
  apiKey: process.env.HOGSEND_DATA_KEY!, // hsk_… key with the ingest scope
});

// identity facts → the contact record (what buckets segment on)
await hs.contacts.upsert({
  userId: lead.userId,
  email: lead.email,
  properties: { company_size: lead.companySize, industry: lead.industry },
});

// what happened → one event, scalar properties only. The idempotency key
// is derived from the run, so however many times the agent loop retries,
// the event ingests once.
await hs.events.send({
  name: "lead.research_completed",
  userId: lead.userId,
  eventProperties: { score: lead.score, tier: lead.tier },
  idempotencyKey: `research-${lead.userId}-${runId}`,
});

// The template's prop contract. This schema is the ENTIRE surface the
// model can fill — everything else in the email is the registry component.
const TipsDraft = z.object({
  subjectLine: z.string().max(80),
  tips: z
    .array(z.object({ title: z.string().max(60), body: z.string().max(280) }))
    .length(3),
});

const anthropic = new Anthropic(); // reads ANTHROPIC_API_KEY

export const draftWeeklyTipsTask = hatchet.durableTask({
  name: "draft-weekly-tips",
  onEvents: [Events.ONBOARDING_WEEK_COMPLETED],
  retries: 2,
  executionTimeout: "10m",
  fn: async (input: WeekCompletedInput) => {
    const { db } = getContainer();

    const userId = typeof input.userId === "string" ? input.userId : "";
    const email = typeof input.userEmail === "string" ? input.userEmail : "";
    if (!userId || !email) {
      return { status: "skipped", reason: "missing_identity" };
    }

    // The personalization source is your own event log, not the payload.
    const recent = await db.query.userEvents.findMany({
      where: and(
        eq(userEvents.userId, userId),
        gte(userEvents.occurredAt, new Date(Date.now() - 7 * 86_400_000)),
      ),
      orderBy: [desc(userEvents.occurredAt)],
      limit: 50,
    });
    const activity = recent.map((e) => e.event).join("\n");

    const response = await anthropic.messages.parse({
      model: "claude-haiku-4-5",
      max_tokens: 1024,
      system:
        "You write short, concrete product tips for a developer audience. " +
        "Plain prose. No exclamation marks, no emoji.",
      messages: [
        {
          role: "user",
          content: `Events this user fired in their first week, newest first:\n${activity}\n\nDraft a subject line and three tips for what to do next.`,
        },
      ],
      output_config: { format: zodOutputFormat(TipsDraft) },
    });

    // The gate. parsed_output is null when the completion failed
    // validation; TipsDraft.parse throws on null or any shape drift —
    // the task fails and Hatchet retries it. Nothing has been sent yet.
    const draft = TipsDraft.parse(response.parsed_output);

    await sendEmail({
      to: email,
      userId,
      template: Templates.LIFECYCLE_WEEKLY_TIPS, // "lifecycle/weekly-tips"
      subject: draft.subjectLine,
      props: { tips: draft.tips },
    });

    return { status: "sent" };
  },
});

export const exitInterview = defineJourney({
  meta: {
    id: "exit-interview",
    name: "Agentic — exit interview",
    enabled: true,
    trigger: {
      event: Events.TRIAL_COMPLETED,
      where: (b) => b.prop("converted").neq(true),
    },
    entryLimit: "once",
    suppress: hours(12),
    // Buying ends the conversation. Neither awaited event may appear here.
    exitOn: [{ event: Events.SUBSCRIPTION_CREATED }],
  },

  run: async (user, ctx) => {
    await sendEmail({
      to: user.email,
      userId: user.id,
      journeyStateId: user.stateId,
      template: Templates.FEEDBACK_EXIT_INTERVIEW,
      subject: "What stopped you?",
      journeyName: user.journeyName,
    });

    // The user's answer — EmailAction buttons fire churn.reason_provided.
    const answer = await ctx.waitForEvent({
      event: Events.CHURN_REASON_PROVIDED,
      timeout: days(5),
      lookback: minutes(30),
    });
    if (answer.timedOut) return; // never answered — leave them be

    // The agent's verdict. The confirmed answer is already on its way to
    // the agent as an email.action delivery; the agent fires
    // churn.followup_selected back. lookback covers a verdict that landed
    // before this wait was established — the agent can decide in seconds.
    const verdict = await ctx.waitForEvent({
      event: Events.CHURN_FOLLOWUP_SELECTED,
      timeout: hours(6),
      lookback: minutes(30),
    });

    if (!(await ctx.guard.isSubscribed())) return;

    const action = verdict.timedOut
      ? "none"
      : String(verdict.properties?.action ?? "none");

    if (action === "offer") {
      await sendEmail({
        to: user.email,
        userId: user.id,
        journeyStateId: user.stateId,
        template: Templates.FEEDBACK_SAVE_OFFER,
        subject: "One more month on us",
        journeyName: user.journeyName,
      });
      return;
    }

    if (action === "handoff") {
      // scalars only — the alert task resolves identity server-side
      await ctx.trigger({
        event: Events.LEAD_FLAGGED,
        userId: user.id,
        properties: {
          reason: "exit-interview",
          answer: String(answer.properties?.reason ?? "unknown"),
          sourceEvent: Events.CHURN_REASON_PROVIDED,
          answeredAt: new Date().toISOString(),
        },
      });
    }
    // "none" or timeout — the answer is recorded; nothing else sends.
  },
});