AI Customer Churn Probability Detector

id: ai-customer-churn-probability-detector
namespace: company.team
description: |
  Automated AI-powered customer churn detection workflow. Queries inactive accounts from PostgreSQL CRM,
  prompts Anthropic Claude to determine risk scores and retention tactics, and dispatches high-urgency Slack alerts.

concurrency:
  limit: 1

inputs:
  - id: postgres_url
    type: STRING
    displayName: CRM Database JDBC URL
    defaults: "jdbc:postgresql://postgres.internal:5432/crm_db"
    description: "JDBC URL of the application CRM database."

  - id: postgres_username
    type: STRING
    displayName: Database Username
    defaults: "crm_reader"
    description: "Database user with read permissions on customer_accounts."

  - id: inactive_days_threshold
    type: INT
    displayName: Inactivity Days Threshold
    defaults: 30
    description: "Minimum days of customer inactivity before evaluating account for
      churn risk."

  - id: churn_risk_score_alert_threshold
    type: INT
    displayName: Churn Score Alert Threshold
    defaults: 75
    description: "Minimum AI churn risk score (0-100) to trigger immediate Customer
      Success alert."

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    defaults: "#customer-success"
    description: "Slack channel destination for high-risk churn escalation alerts."

triggers:
  - id: weekly_churn_scan
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Scans for customer churn risks weekly on Monday at 09:00 UTC."
    cron: "0 9 * * 1"

tasks:
  - id: query_inactive_accounts
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: "Queries customer_accounts table for high-MRR accounts with no
      recent user activity."
    url: "{{ inputs.postgres_url }}"
    username: "{{ inputs.postgres_username }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        account_id,
        company_name,
        mrr_amount,
        plan_tier,
        last_active_date,
        CURRENT_DATE - last_active_date AS inactive_days,
        open_tickets_count
      FROM customer_accounts
      WHERE status = 'active'
        AND last_active_date < CURRENT_DATE - (INTERVAL '1 day' * {{ inputs.inactive_days_threshold }})
      ORDER BY mrr_amount DESC
      LIMIT 10;

  - id: analyze_churn_risk_with_claude
    type: io.kestra.plugin.anthropic.ChatCompletion
    description: "Sends inactive customer telemetry to Anthropic Claude to score
      churn risk and draft retention plans."
    apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
    model: "claude-3-5-sonnet-20241022"
    messages:
      - role: user
        content: |
          You are an enterprise Customer Success AI analyst. Evaluate these inactive customer accounts and compute a churn risk probability score from 0 to 100 for each.
          Return ONLY valid JSON matching this structure:
          {"accounts": [{"account_id": "...", "company_name": "...", "churn_score": 85, "churn_driver": "high MRR with zero logins in 45 days", "recommended_action": "..."}]}

          Account data:
          {{ outputs.query_inactive_accounts.rows | default('[]') | json }}

  - id: parse_churn_scores
    type: io.kestra.plugin.scripts.python.Script
    description: "Parses Claude output JSON and filters high-risk accounts breaching
      the threshold."
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    inputFiles:
      claude_output.txt: "{{ outputs.analyze_churn_risk_with_claude.choices[0].message.content }}"
    outputFiles:
      - churn_analysis_report.json
    script: |
      import json
      import re
      from kestra import Kestra

      with open("claude_output.txt", "r", encoding="utf-8") as f:
          raw_text = f.read().strip()

      clean_json = re.sub(r"^```(?:json)?\s*", "", raw_text, flags=re.MULTILINE)
      clean_json = re.sub(r"\s*```$", "", clean_json, flags=re.MULTILINE).strip()

      parsed = json.loads(clean_json)
      accounts = parsed.get("accounts", [])

      alert_threshold = int("{{ inputs.churn_risk_score_alert_threshold }}")
      high_risk = []

      for a in accounts:
          score = int(a.get("churn_score", 0))
          if score >= alert_threshold:
              high_risk.append(a)

      has_high_risk = len(high_risk) > 0

      with open("churn_analysis_report.json", "w", encoding="utf-8") as f:
          json.dump({
              "total_analyzed": len(accounts),
              "high_risk_count": len(high_risk),
              "high_risk_accounts": high_risk
          }, f, indent=2)

      top_account = high_risk[0] if high_risk else {}
      Kestra.outputs({
          "high_risk_detected": has_high_risk,
          "high_risk_count": len(high_risk),
          "top_risk_company": top_account.get("company_name", "None"),
          "top_risk_score": top_account.get("churn_score", 0),
          "top_action": top_account.get("recommended_action", "None")
      })

  - id: evaluate_high_risk_condition
    type: io.kestra.plugin.core.flow.If
    description: "Branches execution based on whether high-risk churn accounts were
      identified."
    condition: "{{ outputs.parse_churn_scores.vars.high_risk_detected }}"
    then:
      - id: alert_customer_success_high_churn
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: "Dispatches actionable churn prevention card to Customer Success
          team via Slack."
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "*Customer Churn Risk Alert*\n• *Evaluated Accounts*: {{ outputs.parse_churn_scores.vars.high_risk_count }} critical risk accounts identified\n• *Top Risk Account*: `{{ outputs.parse_churn_scores.vars.top_risk_company }}` (Score: {{ outputs.parse_churn_scores.vars.top_risk_score }}/100)\n• *Identified Churn Driver*: Inactive telemetry exceeding {{ inputs.inactive_days_threshold }} days with open support escalations\n• *Recommended Retention Action*: {{ outputs.parse_churn_scores.vars.top_action }}\n• *Operational Impact*: Immediate executive check-in required to protect monthly recurring revenue (MRR) before renewal window closes."
          }

    else:
      - id: log_churn_status_healthy
        type: io.kestra.plugin.core.log.Log
        description: "Logs confirmation of healthy customer retention metrics."
        message: "Customer retention audit compliant. Zero accounts exceeded the {{
          inputs.churn_risk_score_alert_threshold }} churn risk score
          threshold."

  - id: export_churn_manifest
    type: io.kestra.plugin.core.debug.Return
    description: "Outputs structured JSON customer retention manifest for executive
      reporting."
    format: |
      {
        "pipeline": "ai-customer-churn-probability-detector",
        "audited_at": "{{ execution.startDate }}",
        "high_risk_count": {{ outputs.parse_churn_scores.vars.high_risk_count }},
        "status": "{{ outputs.parse_churn_scores.vars.high_risk_detected ? 'CHURN_RISK_DETECTED' : 'HEALTHY' }}"
      }