Subscription Cancellation Save Offer and Churn Guard with Kestra

id: subscription-cancellation-save-offer
namespace: company.team
description: |
  Respond to subscription cancellation intent with a tiered save offer: assess
  eligibility from the billing provider event, dispatch a pause, discount or
  downgrade offer, then apply the retention win or finalize the churn from the
  response or the timeout sweep.

inputs:
  - id: event_payload
    type: JSON
    displayName: Cancellation or offer response event
    description: Fallback event body used when the flow runs manually or from the
      schedule instead of the billing webhook.
    defaults: |
      {
        "event_kind": "cancellation_intent",
        "customer_id": "cus_10432",
        "customer_email": "alex@example.com",
        "segment": "mid_market",
        "tenure_months": 14,
        "mrr": 820.0,
        "usage_trend": "growing",
        "prior_saves": 0
      }

  - id: min_mrr
    type: INT
    displayName: Minimum MRR threshold
    description: Monthly recurring revenue below this amount is not worth a save
      offer, so the cancellation is finalized without one.
    defaults: 100

  - id: auto_finalize_decline
    type: BOOL
    displayName: Finalize on decline
    description: When true, a declined or timed out save offer also POSTs the
      cancellation finalization to the billing API.
    defaults: false

variables:
  billing_api_base: https://billing.example.com/api
  min_tenure_months: "3"
  max_prior_saves: "1"
  discount_mrr_floor: "300"
  pause_tenure_months: "6"
  response_deadline_days: "7"
  discount_percent: "30"
  pause_days: "60"

triggers:
  - id: cancellation_intent
    type: io.kestra.plugin.core.trigger.Webhook
    description: The billing provider POSTs cancellation intent events and later
      save offer responses to this webhook.
    key: "{{ secret('CANCEL_SAVE_WEBHOOK_KEY') }}"

  - id: timeout_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily sweep entry point for offers whose response window closed
      without a reply. Enable it once offers are going out.
    cron: "0 7 * * *"
    disabled: true

tasks:
  # --- 1. assess the cancellation event ---
  - id: assess_eligibility
    type: io.kestra.plugin.scripts.python.Script
    description: Score the event on tenure months, monthly recurring revenue, usage
      trend, prior saves and segment, then emit eligibility, offer tier,
      verdict, final state and revenue at risk through the JSON output protocol.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.12-slim
    inputFiles:
      event.json: "{{ (trigger.body ?? inputs.event_payload) | toJson }}"
      config.json: |
        {
          "min_mrr": "{{ inputs.min_mrr }}",
          "min_tenure_months": "{{ vars.min_tenure_months }}",
          "max_prior_saves": "{{ vars.max_prior_saves }}",
          "discount_mrr_floor": "{{ vars.discount_mrr_floor }}",
          "pause_tenure_months": "{{ vars.pause_tenure_months }}",
          "response_deadline_days": "{{ vars.response_deadline_days }}",
          "discount_percent": "{{ vars.discount_percent }}",
          "pause_days": "{{ vars.pause_days }}",
          "cs_team_email": "{{ secret('CS_TEAM_EMAIL') }}"
        }
    script: |
      import json
      from datetime import date, timedelta

      VALID_TIERS = ("pause", "discount", "downgrade", "none")
      DECLINING = ("declining", "decline", "down", "shrinking")
      GROWING = ("growing", "increase", "increasing", "up", "expanding")
      STEADY = ("steady", "flat", "stable")

      def read_text(name):
          try:
              with open(name, "r", encoding="utf-8") as handle:
                  return handle.read()
          except (OSError, UnicodeError):
              return None

      def load_payload(name, fallback):
          raw = read_text(name)
          if raw is None:
              return fallback, "missing_file"
          if not raw.strip():
              return fallback, "empty_file"
          try:
              value = json.loads(raw)
          except ValueError:
              return fallback, "malformed_json"
          if isinstance(value, str):
              try:
                  value = json.loads(value)
              except ValueError:
                  return fallback, "string_payload"
          return value, "ok"

      def as_number(value, default=0.0):
          try:
              return float(value)
          except (TypeError, ValueError):
              return default

      def as_int(value, default=0):
          try:
              return int(float(value))
          except (TypeError, ValueError):
              return default

      def as_text(value, default=""):
          if value is None:
              return default
          return str(value).strip()

      config, config_state = load_payload("config.json", {})
      if not isinstance(config, dict):
          config = {}

      event, event_state = load_payload("event.json", {})
      if not isinstance(event, dict):
          if event_state == "ok":
              event_state = "not_an_object"
          event = {}

      min_mrr = as_number(config.get("min_mrr"), 100.0)
      min_tenure = as_int(config.get("min_tenure_months"), 3)
      max_prior_saves = as_int(config.get("max_prior_saves"), 1)
      discount_floor = as_number(config.get("discount_mrr_floor"), 300.0)
      pause_tenure = as_int(config.get("pause_tenure_months"), 6)
      deadline_days = as_int(config.get("response_deadline_days"), 7)
      discount_percent = as_int(config.get("discount_percent"), 30)
      pause_days = as_int(config.get("pause_days"), 60)

      customer_id = as_text(event.get("customer_id") or event.get("customer") or event.get("id"), "unknown_customer")
      customer_email = as_text(event.get("customer_email") or event.get("email"), "")
      segment = as_text(event.get("segment") or event.get("plan_segment"), "unknown").lower()
      usage_trend = as_text(event.get("usage_trend") or event.get("trend"), "unknown").lower()
      tenure_months = as_int(event.get("tenure_months"), 0)
      mrr = round(as_number(event.get("mrr"), 0.0), 2)
      prior_saves = as_int(event.get("prior_saves"), 0)

      raw_kind = as_text(event.get("event_kind") or event.get("type") or event.get("event"), "").lower()
      raw_response = as_text(event.get("offer_response") or event.get("response") or event.get("decision"), "").lower()

      if raw_kind in ("offer_response", "save_offer_response", "offer_reply", "offer_decision"):
          event_kind = "offer_response"
      elif raw_kind in ("offer_timeout", "save_offer_timeout", "timeout_sweep", "timeout"):
          event_kind = "offer_timeout"
      else:
          event_kind = "cancellation_intent"

      if event_kind == "offer_timeout":
          offer_response = "timeout"
      elif event_kind == "offer_response":
          offer_response = raw_response if raw_response in ("accepted", "declined") else "unrecognized"
      else:
          offer_response = "pending"

      reasons = []
      if event_state != "ok":
          reasons.append("payload_" + event_state)
      if mrr < min_mrr:
          reasons.append("mrr_below_threshold")
      if tenure_months < min_tenure:
          reasons.append("tenure_below_minimum")
      if prior_saves > max_prior_saves:
          reasons.append("prior_saves_over_cap")
      if segment == "trial":
          reasons.append("trial_segment")
      if offer_response == "unrecognized":
          reasons.append("offer_response_unrecognized")
      eligible = not reasons

      event_tier = as_text(event.get("offer_tier"), "").lower()
      if event_kind in ("offer_response", "offer_timeout") and event_tier in VALID_TIERS and event_tier != "none":
          offer_tier = event_tier
      elif not eligible:
          offer_tier = "none"
      elif usage_trend in DECLINING:
          offer_tier = "downgrade"
      elif usage_trend in GROWING and mrr >= discount_floor:
          offer_tier = "discount"
      elif tenure_months >= pause_tenure:
          offer_tier = "pause"
      else:
          offer_tier = "discount"

      if offer_tier == "discount":
          retention_action = "apply_discount"
          offer_summary = "%d%% off for 3 months" % discount_percent
          mrr_credit = round(mrr * discount_percent / 100.0, 2)
      elif offer_tier == "pause":
          retention_action = "pause_subscription"
          offer_summary = "Pause the subscription for %d days at no charge" % pause_days
          mrr_credit = round(mrr * pause_days / 30.0, 2)
      elif offer_tier == "downgrade":
          retention_action = "switch_plan"
          offer_summary = "Move to the lower cost plan at the current usage level"
          mrr_credit = 0.0
      else:
          retention_action = "none"
          offer_summary = "No save offer, the cancellation is finalized"
          mrr_credit = 0.0

      if event_state != "ok":
          next_action = "manual_review"
          verdict = "invalid_payload"
          final_state = "manual_review"
      elif event_kind in ("offer_response", "offer_timeout"):
          if offer_response == "accepted":
              next_action = "apply_retention"
              verdict = "offer_accepted_retained"
              final_state = "retained"
          else:
              next_action = "finalize_churn"
              if event_kind == "offer_timeout":
                  verdict = "offer_timeout_churn"
              else:
                  verdict = "offer_declined_churn"
              final_state = "churn_confirmed"
      elif eligible:
          next_action = "dispatch_offer"
          verdict = "eligible_save_offer"
          final_state = "offer_awaiting_response"
      else:
          next_action = "finalize_no_offer"
          verdict = "ineligible_finalize"
          final_state = "churn_no_offer"

      if next_action == "apply_retention":
          mrr_at_risk = 0.0
      else:
          mrr_at_risk = round(mrr, 2)

      offer_deadline = (date.today() + timedelta(days=max(deadline_days, 0))).isoformat()
      reasons_text = ", ".join(reasons) if reasons else "none"
      recipient = customer_email or as_text(config.get("cs_team_email"), "")

      payload = {
          "verdict": verdict,
          "event_kind": event_kind,
          "offer_response": offer_response,
          "next_action": next_action,
          "eligible": eligible,
          "offer_tier": offer_tier,
          "retention_action": retention_action,
          "offer_summary": offer_summary,
          "mrr_credit": mrr_credit,
          "offer_deadline": offer_deadline,
          "final_state": final_state,
          "mrr_at_risk": mrr_at_risk,
          "mrr": mrr,
          "customer_id": customer_id,
          "customer_email": customer_email,
          "recipient": recipient,
          "segment": segment,
          "tenure_months": tenure_months,
          "usage_trend": usage_trend,
          "prior_saves": prior_saves,
          "reasons": reasons,
          "reasons_text": reasons_text,
          "payload_state": event_state
      }
      print("::" + json.dumps({"outputs": payload}) + "::")
      print("kind=%s eligible=%s tier=%s action=%s verdict=%s state=%s mrr_at_risk=%.2f" % (
          event_kind, eligible, offer_tier, next_action, verdict, final_state, mrr_at_risk))

  # --- 2. record the intervention ---
  - id: record_intervention
    type: io.kestra.plugin.jdbc.postgresql.Queries
    description: Upsert one intervention row per customer so the latest assessment,
      offer tier and response state always replace the previous one.
    url: "{{ secret('POSTGRES_URL') }}"
    username: "{{ secret('POSTGRES_USER') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: NONE
    sql: |
      INSERT INTO public.churn_interventions (
        customer_id,
        segment,
        tenure_months,
        mrr_usd,
        usage_trend,
        prior_saves,
        eligible,
        offer_tier,
        verdict,
        event_kind,
        offer_response,
        updated_at
      )
      VALUES (
        :customer_id,
        :segment,
        CAST(:tenure_months AS integer),
        CAST(:mrr AS double precision),
        :usage_trend,
        CAST(:prior_saves AS integer),
        CAST(:eligible AS boolean),
        :offer_tier,
        :verdict,
        :event_kind,
        :offer_response,
        NOW()
      )
      ON CONFLICT (customer_id) DO UPDATE SET
        segment = EXCLUDED.segment,
        tenure_months = EXCLUDED.tenure_months,
        mrr_usd = EXCLUDED.mrr_usd,
        usage_trend = EXCLUDED.usage_trend,
        prior_saves = EXCLUDED.prior_saves,
        eligible = EXCLUDED.eligible,
        offer_tier = EXCLUDED.offer_tier,
        verdict = EXCLUDED.verdict,
        event_kind = EXCLUDED.event_kind,
        offer_response = EXCLUDED.offer_response,
        updated_at = NOW();
    parameters:
      customer_id: "{{ outputs.assess_eligibility.vars.customer_id }}"
      segment: "{{ outputs.assess_eligibility.vars.segment }}"
      tenure_months: "{{ outputs.assess_eligibility.vars.tenure_months }}"
      mrr: "{{ outputs.assess_eligibility.vars.mrr }}"
      usage_trend: "{{ outputs.assess_eligibility.vars.usage_trend }}"
      prior_saves: "{{ outputs.assess_eligibility.vars.prior_saves }}"
      eligible: "{{ outputs.assess_eligibility.vars.eligible }}"
      offer_tier: "{{ outputs.assess_eligibility.vars.offer_tier }}"
      verdict: "{{ outputs.assess_eligibility.vars.verdict }}"
      event_kind: "{{ outputs.assess_eligibility.vars.event_kind }}"
      offer_response: "{{ outputs.assess_eligibility.vars.offer_response }}"

  # --- 3. dispatch the save offer ---
  - id: offer_dispatch_gate
    type: io.kestra.plugin.core.flow.If
    description: Send the save offer only for an eligible account on a fresh
      cancellation intent, and record the finalize preference for everything
      else.
    condition: "{{ outputs.assess_eligibility.vars.eligible and
      outputs.assess_eligibility.vars.next_action == 'dispatch_offer' }}"
    then:
      - id: send_save_offer
        type: io.kestra.plugin.email.MailSend
        description: Email the customer a personalized pause, discount or downgrade
          offer over the secret SMTP relay with the response deadline in the
          body.
        host: "{{ secret('SMTP_HOST') }}"
        port: 465
        username: "{{ secret('SMTP_USERNAME') }}"
        password: "{{ secret('SMTP_PASSWORD') }}"
        from: "{{ secret('CS_EMAIL_FROM') }}"
        to: "{{ outputs.assess_eligibility.vars.recipient }}"
        subject: "A quick option before you go, {{
          outputs.assess_eligibility.vars.customer_id }}"
        htmlTextContent: |
          <h2>Thanks for the feedback</h2>
          <p>We noticed your cancellation request for account
          <strong>{{ outputs.assess_eligibility.vars.customer_id }}</strong>
          ({{ outputs.assess_eligibility.vars.segment }}, with us for
          {{ outputs.assess_eligibility.vars.tenure_months }} months).</p>
          <p>Before you go, here is an offer we can make right now:
          <strong>{{ outputs.assess_eligibility.vars.offer_summary }}</strong>.</p>
          <p>Offer tier {{ outputs.assess_eligibility.vars.offer_tier }},
          monthly recurring revenue {{ outputs.assess_eligibility.vars.mrr }},
          reply before {{ outputs.assess_eligibility.vars.offer_deadline }}
          and we will keep your data and history exactly as it is.</p>
          <p>If we do not hear back by that date, we will proceed with the
          cancellation you asked for.</p>

      - id: notify_cs_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alert the customer success channel with the account context, the
          chosen tier and the response deadline.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": ":speak_no_evil: Save offer {{ outputs.assess_eligibility.vars.offer_tier }} queued for {{ outputs.assess_eligibility.vars.customer_id }} ({{ outputs.assess_eligibility.vars.segment }}, MRR {{ outputs.assess_eligibility.vars.mrr }}, trend {{ outputs.assess_eligibility.vars.usage_trend }}, tenure {{ outputs.assess_eligibility.vars.tenure_months }} months). {{ outputs.assess_eligibility.vars.offer_summary }}, reply by {{ outputs.assess_eligibility.vars.offer_deadline }}. MRR at risk {{ outputs.assess_eligibility.vars.mrr_at_risk }}, verdict {{ outputs.assess_eligibility.vars.verdict }}. Execution {{ execution.id }}."
          }

      - id: log_offer_dispatched
        type: io.kestra.plugin.core.log.Log
        description: Record the dispatched offer with its tier, summary and deadline for
          the execution log.
        message: "Save offer dispatched for {{
          outputs.assess_eligibility.vars.customer_id }} tier {{
          outputs.assess_eligibility.vars.offer_tier }}, {{
          outputs.assess_eligibility.vars.offer_summary }}, deadline {{
          outputs.assess_eligibility.vars.offer_deadline }}, MRR at risk {{
          outputs.assess_eligibility.vars.mrr_at_risk }}."

    else:
      - id: log_finalize_pref
        type: io.kestra.plugin.core.log.Log
        description: Record that no new save offer went out, together with the verdict
          and the next action this run takes instead.
        message: "No save offer for {{ outputs.assess_eligibility.vars.customer_id }},
          verdict {{ outputs.assess_eligibility.vars.verdict }}, next action {{
          outputs.assess_eligibility.vars.next_action }}, tier {{
          outputs.assess_eligibility.vars.offer_tier }}, reasons {{
          outputs.assess_eligibility.vars.reasons_text }}."

  # --- 4. handle the offer response or the timeout sweep ---
  - id: response_phase_gate
    type: io.kestra.plugin.core.flow.If
    description: Run the outcome branches only when the payload carries an offer
      response or an offer timeout, so fresh intents end waiting for a reply.
    condition: "{{ outputs.assess_eligibility.vars.event_kind == 'offer_response' or
      outputs.assess_eligibility.vars.event_kind == 'offer_timeout' }}"
    then:
      - id: accepted_gate
        type: io.kestra.plugin.core.flow.If
        description: An accepted offer applies the billing credit or pause and writes
          the retention win, a declined or timed out offer reports the churn and
          optionally finalizes the cancellation.
        condition: "{{ outputs.assess_eligibility.vars.offer_response == 'accepted' }}"
        then:
          - id: apply_billing_credit
            type: io.kestra.plugin.core.http.Request
            description: POST the accepted credit or pause to the billing API so the
              subscription is actually changed.
            uri: "{{ vars.billing_api_base }}/subscriptions/{{
              outputs.assess_eligibility.vars.customer_id }}/retention"
            method: POST
            contentType: application/json
            headers:
              Authorization: "Bearer {{ secret('BILLING_API_TOKEN') }}"
            body: |
              {
                "customer_id": "{{ outputs.assess_eligibility.vars.customer_id }}",
                "offer_tier": "{{ outputs.assess_eligibility.vars.offer_tier }}",
                "retention_action": "{{ outputs.assess_eligibility.vars.retention_action }}",
                "mrr_credit": {{ outputs.assess_eligibility.vars.mrr_credit }},
                "execution_id": "{{ execution.id }}"
              }

          - id: record_retention_win
            type: io.kestra.plugin.jdbc.postgresql.Queries
            description: Upsert the retention win with the tier, the action applied and the
              credit granted so the win rate is queryable.
            url: "{{ secret('POSTGRES_URL') }}"
            username: "{{ secret('POSTGRES_USER') }}"
            password: "{{ secret('POSTGRES_PASSWORD') }}"
            fetchType: NONE
            sql: |
              INSERT INTO public.retention_wins (
                customer_id,
                offer_tier,
                retention_action,
                mrr_credit,
                mrr_at_risk,
                execution_id,
                won_at
              )
              VALUES (
                :customer_id,
                :offer_tier,
                :retention_action,
                CAST(:mrr_credit AS double precision),
                CAST(:mrr_at_risk AS double precision),
                :execution_id,
                NOW()
              )
              ON CONFLICT (customer_id) DO UPDATE SET
                offer_tier = EXCLUDED.offer_tier,
                retention_action = EXCLUDED.retention_action,
                mrr_credit = EXCLUDED.mrr_credit,
                mrr_at_risk = EXCLUDED.mrr_at_risk,
                execution_id = EXCLUDED.execution_id,
                won_at = NOW();
            parameters:
              customer_id: "{{ outputs.assess_eligibility.vars.customer_id }}"
              offer_tier: "{{ outputs.assess_eligibility.vars.offer_tier }}"
              retention_action: "{{ outputs.assess_eligibility.vars.retention_action }}"
              mrr_credit: "{{ outputs.assess_eligibility.vars.mrr_credit }}"
              mrr_at_risk: "{{ outputs.assess_eligibility.vars.mrr_at_risk }}"
              execution_id: "{{ execution.id }}"

          - id: log_retention_win
            type: io.kestra.plugin.core.log.Log
            description: Record the retained account with its tier, action and credit in the
              execution log.
            message: "Retention win for {{ outputs.assess_eligibility.vars.customer_id }},
              tier {{ outputs.assess_eligibility.vars.offer_tier }}, action {{
              outputs.assess_eligibility.vars.retention_action }}, credit {{
              outputs.assess_eligibility.vars.mrr_credit }}, final state {{
              outputs.assess_eligibility.vars.final_state }}."

        else:
          - id: churn_report
            type: io.kestra.plugin.core.log.Log
            description: Log the churn report for a declined or timed out offer with the
              final state and the revenue at risk.
            message: "Churn confirmed for {{ outputs.assess_eligibility.vars.customer_id }},
              verdict {{ outputs.assess_eligibility.vars.verdict }}, final state
              {{ outputs.assess_eligibility.vars.final_state }}, MRR at risk {{
              outputs.assess_eligibility.vars.mrr_at_risk }}, tier was {{
              outputs.assess_eligibility.vars.offer_tier }}, response {{
              outputs.assess_eligibility.vars.offer_response }}."

          - id: finalize_cancellation
            type: io.kestra.plugin.core.http.Request
            description: Optional POST that tells the billing provider to complete the
              cancellation after the declined or timed out offer.
            runIf: "{{ inputs.auto_finalize_decline }}"
            uri: "{{ vars.billing_api_base }}/subscriptions/{{
              outputs.assess_eligibility.vars.customer_id }}/cancel"
            method: POST
            contentType: application/json
            headers:
              Authorization: "Bearer {{ secret('BILLING_API_TOKEN') }}"
            body: |
              {
                "customer_id": "{{ outputs.assess_eligibility.vars.customer_id }}",
                "reason": "{{ outputs.assess_eligibility.vars.verdict }}",
                "mrr_at_risk": {{ outputs.assess_eligibility.vars.mrr_at_risk }},
                "execution_id": "{{ execution.id }}"
              }

    else:
      - id: log_run_outcome
        type: io.kestra.plugin.core.log.Log
        description: Close the run with the verdict, next action and final state when
          the payload was a fresh cancellation intent.
        message: "Run closed for {{ outputs.assess_eligibility.vars.customer_id }},
          verdict {{ outputs.assess_eligibility.vars.verdict }}, next action {{
          outputs.assess_eligibility.vars.next_action }}, final state {{
          outputs.assess_eligibility.vars.final_state }}, MRR at risk {{
          outputs.assess_eligibility.vars.mrr_at_risk }}."

errors:
  - id: slack_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post the failing execution to Slack when the database, the SMTP
      relay or the billing API call fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED {{ flow.id }} run {{ execution.id }}: {{ error.message }}"}

outputs:
  - id: verdict
    type: STRING
    value: "{{ outputs.assess_eligibility.vars.verdict }}"

  - id: offer_tier
    type: STRING
    value: "{{ outputs.assess_eligibility.vars.offer_tier }}"

  - id: eligible
    type: BOOL
    value: "{{ outputs.assess_eligibility.vars.eligible }}"

  - id: final_state
    type: STRING
    value: "{{ outputs.assess_eligibility.vars.final_state }}"

  - id: mrr_at_risk
    type: FLOAT
    value: "{{ outputs.assess_eligibility.vars.mrr_at_risk }}"

Cancellation is the moment a SaaS company either saves the relationship or loses the revenue, and the decision is a workflow rather than a one off script. This blueprint takes a cancellation intent event from your billing provider, scores the account, sends the right save offer, tracks the response, and either applies the retention win or closes the loop on the churn. Every step lands in PostgreSQL and in the execution log, so the customer success team can always see why a given account received a pause instead of a discount.

How it works

The cancellation_intent trigger (io.kestra.plugin.core.trigger.Webhook, guarded by a secret key) receives the billing provider cancellation event, and the disabled daily timeout_sweep Schedule trigger is the sweep entry point for offers whose response window has closed.
The assess_eligibility task (io.kestra.plugin.scripts.python.Script on a Docker task runner) reads the event body, falling back to the event_payload input on manual runs, and evaluates tenure months, monthly recurring revenue, usage trend, prior save count and segment. It emits eligible, offer_tier (pause, discount, downgrade or none), verdict, final_state, mrr_at_risk, the offer summary and the response state through the ::json:: output protocol.
The record_intervention task (io.kestra.plugin.jdbc.postgresql.Queries with fetchType: NONE) upserts one row per customer into public.churn_interventions, so the latest assessment always replaces the previous one.
The offer_dispatch_gate task (io.kestra.plugin.core.flow.If) sends the save offer only when the account is eligible and the event is a fresh cancellation intent. Eligible accounts get a personalized email (io.kestra.plugin.email.MailSend) whose offer text follows the chosen tier, plus a Slack heads up to the customer success channel through io.kestra.plugin.slack.notifications.SlackIncomingWebhook. Every other run falls through to log_finalize_pref, which records the finalize preference without sending an offer.
The response_phase_gate task branches only when the payload carries an offer response or an offer timeout, so fresh intents end waiting for a reply. Inside it, accepted_gate applies an accepted offer with io.kestra.plugin.core.http.Request against the billing API and writes the retention win back to PostgreSQL, while declined or timed out offers log the churn report and optionally POST the cancellation finalization when auto_finalize_decline is enabled.
Any failure in the database, the SMTP relay or the HTTP calls lands in the errors block, which posts the failing execution to Slack.

What you get

A single eligibility verdict per cancellation event with a tiered offer decision attached to it, so the save motion stays consistent across accounts instead of depending on who picked up the ticket. A personalized offer email that proposes a pause, a time boxed discount or a plan downgrade depending on tenure, revenue and usage trend. A Slack alert that gives the customer success team the account context, the offer tier and the response deadline in one message. A retention win record in PostgreSQL next to the billing API call that applies the credit or pause. A churn report path that records the final state and the monthly recurring revenue at risk when the offer is declined or times out. Typed flow outputs (verdict, offer_tier, eligible, final_state, mrr_at_risk) that a parent revenue flow or dashboard can consume.

Who it's for

Customer success teams that own gross revenue retention and want the first save offer out while the cancellation intent is still fresh. SaaS founders who need a repeatable retention motion without a lifecycle marketer hand writing every offer. Revenue operations teams who care that each intervention, each tier decision and each outcome is queryable in PostgreSQL instead of scattered across an inbox. Platform teams who prefer the billing webhook to trigger a reviewed workflow rather than a script running out of a cron box.

Why orchestrate this with Kestra

The value of a save offer is timing and consistency, and both break the moment the logic lives in an ad hoc script. Kestra gives the run a durable execution record, which event arrived, what the assessment decided, which offer went out and how the response was handled, all tied to one execution id. The eligibility rules sit in a python task with a ::json:: contract, the dispatch is a real branch rather than an if statement buried in a handler, and the declined path is a separate branch that can optionally finalize the cancellation over HTTP. Secrets for SMTP, PostgreSQL, Slack and the billing API stay in the namespace instead of inside the event handler, and the errors handler pages the team when the relay or the database is down. Tightening a threshold or adding a new tier is a YAML diff that gets reviewed before it ever reaches a customer.

Prerequisites

A Kestra instance with the core, Python script, PostgreSQL, email and Slack plugins available to the worker, plus network access to your billing API.
A PostgreSQL table public.churn_interventions with customer_id as the primary key and columns for segment, tenure_months, mrr_usd, usage_trend, prior_saves, eligible, offer_tier, verdict, event_kind, offer_response and updated_at.
A PostgreSQL table public.retention_wins with customer_id as the primary key and columns for offer_tier, retention_action, mrr_credit, mrr_at_risk, execution_id and won_at.
A billing provider that can POST a cancellation intent event and a later save offer response event to the webhook, carrying customer id, customer email, segment, tenure months, monthly recurring revenue, usage trend and prior save count.
An SMTP relay that accepts submissions on port 465, and a Slack incoming webhook for the customer success channel.

Secrets

CANCEL_SAVE_WEBHOOK_KEY: key that guards the cancellation intent webhook trigger.
POSTGRES_URL, POSTGRES_USER, POSTGRES_PASSWORD: connection details for the intervention and retention tables.
SMTP_HOST, SMTP_USERNAME, SMTP_PASSWORD: SMTP relay credentials used by the email plugin.
CS_EMAIL_FROM, CS_TEAM_EMAIL: sender address and internal fallback recipient for the save offer email.
SLACK_WEBHOOK_URL: incoming webhook used for the customer success alert and for failure notices.
BILLING_API_TOKEN: bearer token the billing API expects on the apply and finalize calls.

Quick start

Add the secrets above to the company.team namespace and create the two PostgreSQL tables described in the prerequisites.
Point variables.billing_api_base at your billing provider API base URL, then set variables.response_deadline_days, variables.discount_percent and variables.pause_days to your offer policy.
Run the flow once with the default event_payload so you can read the assessment and the upserted intervention row before a real event arrives.
Configure your billing provider to POST the cancellation intent event and the later offer response to the webhook using this trigger shape:

triggers:
  - id: cancellation_intent
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('CANCEL_SAVE_WEBHOOK_KEY') }}"
  - id: timeout_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 7 * * *"
    disabled: false

Once offers are going out, flip the timeout_sweep schedule to disabled: false and set the default event_payload to an offer_timeout event so expired response windows are finalized each morning.
Set auto_finalize_decline to true only after the customer success team is ready for declined offers to close the subscription without a manual step.

How to extend

Add a io.kestra.plugin.core.execution.Fail task on the churn branch when a declined offer should stop a downstream renewal pipeline instead of only reporting the loss.
Replace the offer email with a io.kestra.plugin.core.flow.Pause so a customer success manager reviews and edits the offer before it goes out.
Read the intervention table back with io.kestra.plugin.jdbc.postgresql.Query at the start of the run, so accounts that already used two save offers are excluded by stored data rather than by a payload field.
Fan the dispatch out by segment with io.kestra.plugin.core.flow.Switch on the segment value, giving enterprise accounts a human follow up while self serve accounts get the automated tier.
Forward the churn report to a data warehouse or a BI webhook with io.kestra.plugin.core.http.Request once final_state is known.
Accept a second webhook event for downgrades that were completed manually, so final_state also reflects saves that never went through this flow.

Subscription Cancellation Save Offer and Churn Guard

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Subscription Cancellation Save Offer and Churn Guard

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Business Blueprints

Reusable Slack Notification Subflow

Reconcile Smart Meter Reads Against the Billing Estimate After Each AMI Import

Snowflake Table Storage and Time Travel FinOps Audit

Archive Slack Channel History into BigQuery

Snowflake Query Spill and Warehouse Rightsizing Advisor

Snowflake Automatic Clustering Cost and FinOps Guard

More Core Blueprints

Set up alerts for failed workflow executions using Gmail

Set up alerts for failed workflow executions using Sentry

Send a Google Chat notification when a workflow fails

Set up alerts for failed workflow executions using Slack

Set up alerts for failed workflow executions using Email

Send an alert to Sentry when any flow fails in the company namespace

New to Kestra?