Pipelines & orchestration

// Hogsend's own outbound catalog — if the PostHog destination forwards
// broadly, events Hogsend fanned out would echo straight back into ingest.
const RESERVED_EVENT_PREFIXES = ["email.", "journey.", "bucket.", "contact."];

export const posthogSource = defineWebhookSource({
  meta: { id: "posthog", name: "PostHog" },
  auth: {
    type: "match",
    header: "x-posthog-webhook-secret",
    envKey: "POSTHOG_WEBHOOK_SECRET",
  },
  schema: posthogWebhookSchema, // zod for PostHog's { event, person } body
  async transform(payload) {
    const eventName = payload.event.event;
    const userId = payload.event.distinct_id;
    const rawEmail = payload.person?.properties?.email;
    const userEmail = typeof rawEmail === "string" ? rawEmail : "";

    // Echo guard: never re-ingest events Hogsend itself fanned out.
    if (payload.event.properties?.$lib === "hogsend") return null;
    if (RESERVED_EVENT_PREFIXES.some((p) => eventName.startsWith(p))) {
      return null;
    }

    // Identity guard: a person with an email, or an identified session.
    // Anonymous distinct_ids would mint a junk contact per session.
    if (!userEmail && payload.event.properties?.$is_identified !== true) {
      return null;
    }

    // Property split: behavioral data → eventProperties (trigger.where,
    // exitOn); profile data → contactProperties (contact merge). Never mixed.
    const eventProperties: Record<string, unknown> = {
      ...payload.event.properties,
    };
    if (payload.event.uuid) {
      eventProperties._posthogEventId = payload.event.uuid;
    }

    return {
      event: eventName,
      userId,
      userEmail,
      eventProperties,
      contactProperties: { ...payload.person?.properties },
    };
  },
});

export const onboardingCheckin = defineJourney({
  meta: {
    id: "onboarding-checkin",
    name: "Onboarding — check-in router",
    enabled: true,
    trigger: { event: Events.USER_SIGNED_UP },
    entryLimit: "once",
    suppress: hours(12),
  },

  run: async (user, ctx) => {
    await ctx.sleep({ duration: days(5), label: "pre-checkin" });
    if (!(await ctx.guard.isSubscribed())) return;

    // The yes/no buttons are semantic links — a click fires
    // checkin.answered { answer } through the full pipeline.
    await sendEmail({
      to: user.email,
      userId: user.id,
      journeyStateId: user.stateId,
      template: Templates.ONBOARDING_CHECKIN,
      subject: "One tap: how is setup going?",
      journeyName: user.journeyName,
    });

    const checkin = await ctx.waitForEvent({
      event: Events.CHECKIN_ANSWERED,
      timeout: days(5),
      lookback: minutes(30), // covers the send → wait-established gap
    });

    if (!(await ctx.guard.isSubscribed())) return;
    const answer = checkin.timedOut ? undefined : checkin.properties?.answer;

    if (answer === "yes") {
      // Activated — route to the referral ask.
      await ctx.trigger({
        event: Events.REFERRAL_ELIGIBLE,
        userId: user.id,
        properties: { reason: "activated", source: "onboarding-checkin" },
      });
      return;
    }

    // "no" or silence — the help-offer path. Never re-pitch someone who
    // already completed that flow on another pass.
    const { completed: alreadyPitched } = await ctx.history.journey({
      userId: user.id,
      journeyId: "setup-offer",
    });
    if (alreadyPitched) return;

    await ctx.trigger({
      event: Events.SETUP_ELIGIBLE,
      userId: user.id,
      properties: {
        reason: answer === "no" ? "needs-help" : "silent",
        source: "onboarding-checkin",
      },
    });
  },
});

// The one list of lifecycle moments that page a human.
function alertText(
  type: string,
  data: Record<string, unknown>,
): string | null {
  // Semantic answers: data.event is YOUR event name, data.properties the
  // confirmed answer scalars.
  if (type === "email.action") {
    const event = String(data.event ?? "");
    const props = (data.properties ?? {}) as Record<string, unknown>;

    if (event === "setup.answered" && props.answer === "interested") {
      return `:raising_hand: ${data.to} answered *interested* to the setup offer`;
    }
    if (
      event === "nps.submitted" &&
      typeof props.score === "number" &&
      props.score <= 6
    ) {
      return `:rotating_light: NPS detractor — ${data.to} scored ${props.score}`;
    }
    return null; // every other answer is not an alert
  }

  // The dunning final notice going out is itself the alert.
  if (type === "email.sent" && data.templateKey === "billing/final-notice") {
    return `:hourglass: Final dunning notice sent to ${data.to}`;
  }

  if (type === "email.complained") {
    return `:no_entry: Spam complaint from ${data.to} (template ${data.templateKey})`;
  }

  return null;
}

export const slackAlerts = defineDestination({
  meta: {
    id: "slack-alerts", // == the webhook_endpoints.kind this serves
    name: "Slack alerts",
    description: "Page a human on hand-raises, detractors, and final notices.",
  },
  events: ["email.action", "email.sent", "email.complained"],
  transform(envelope, ctx) {
    const cfg = (ctx.endpoint.config ?? {}) as { url?: string };
    const url = cfg.url ?? ctx.endpoint.url;
    if (!url) {
      // A throw is a non-retryable config error — straight to the DLQ.
      throw new Error("slack-alerts endpoint is missing config.url");
    }

    const text = alertText(envelope.type, envelope.data);
    if (!text) return null; // skip: marked delivered, no POST, no retry

    return {
      url,
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ text, username: "Hogsend" }),
    };
  },
});