Sync Stripe Payments and Customer Data to a PostgreSQL CRM Database Daily

id: stripe-to-crm-sync
namespace: company.team

inputs:
  - id: lookback_hours
    type: INT
    displayName: Lookback window (hours)
    description: How many hours back to fetch Stripe payments (default 24 for daily sync)
    defaults: 24

tasks:
  - id: get_last_sync_time
    type: io.kestra.plugin.core.kv.Get
    key: stripe_crm_last_sync_ts
    errorOnMissing: false

  - id: fetch_stripe_payments
    type: io.kestra.plugin.core.http.Request
    uri: "https://api.stripe.com/v1/charges?limit=100&created[gte]={{ (now() |
      dateAdd(-inputs.lookback_hours, 'HOURS') | date('X')) }}"
    method: GET
    headers:
      Authorization: "Bearer {{ secret('STRIPE_SECRET_KEY') }}"

  - id: fetch_stripe_customers
    type: io.kestra.plugin.core.http.Request
    uri: "https://api.stripe.com/v1/customers?limit=100"
    method: GET
    headers:
      Authorization: "Bearer {{ secret('STRIPE_SECRET_KEY') }}"

  - id: merge_and_enrich
    type: io.kestra.plugin.scripts.python.Script
    inputFiles:
      payments.json: "{{ outputs.fetch_stripe_payments.body }}"
      customers.json: "{{ outputs.fetch_stripe_customers.body }}"
    script: |
      import json
      from kestra import Kestra

      with open("payments.json") as f:
          payments_data = json.load(f)
      with open("customers.json") as f:
          customers_data = json.load(f)

      # Build customer lookup
      customer_map = {
          c["id"]: c
          for c in customers_data.get("data", [])
      }

      enriched = []
      for payment in payments_data.get("data", []):
          customer_id = payment.get("customer")
          customer = customer_map.get(customer_id, {}) if customer_id else {}

          enriched.append({
              "payment_id": payment.get("id"),
              "customer_id": customer_id,
              "customer_email": customer.get("email") or payment.get("billing_details", {}).get("email"),
              "customer_name": customer.get("name") or payment.get("billing_details", {}).get("name"),
              "amount_cents": payment.get("amount", 0),
              "amount": payment.get("amount", 0) / 100,
              "currency": payment.get("currency", "usd").upper(),
              "status": payment.get("status"),
              "description": payment.get("description") or "",
              "created_ts": payment.get("created", 0),
              "metadata": json.dumps(payment.get("metadata", {}))
          })

      Kestra.outputs({"records": enriched, "count": len(enriched)})

  - id: upsert_to_postgres
    type: io.kestra.plugin.scripts.python.Script
    inputFiles:
      records.json: "{{ outputs.merge_and_enrich.vars.records | json }}"
    dependencies:
      - psycopg2-binary
    script: |
      import json
      import psycopg2
      import os

      with open("records.json") as f:
          records = json.load(f)

      conn = psycopg2.connect(
          host=os.getenv("POSTGRES_HOST"),
          dbname=os.getenv("POSTGRES_DB"),
          user=os.getenv("POSTGRES_USER_VAL"),
          password=os.getenv("POSTGRES_PASS_VAL"),
          port=5432
      )
      cur = conn.cursor()

      cur.execute("""
          CREATE TABLE IF NOT EXISTS crm_payments (
              payment_id TEXT PRIMARY KEY,
              customer_id TEXT,
              customer_email TEXT,
              customer_name TEXT,
              amount NUMERIC,
              currency TEXT,
              status TEXT,
              description TEXT,
              created_ts BIGINT,
              metadata JSONB,
              synced_at TIMESTAMP DEFAULT NOW()
          )
      """)

      for r in records:
          cur.execute("""
              INSERT INTO crm_payments
                  (payment_id, customer_id, customer_email, customer_name, amount, currency,
                   status, description, created_ts, metadata)
              VALUES
                  (%(payment_id)s, %(customer_id)s, %(customer_email)s, %(customer_name)s,
                   %(amount)s, %(currency)s, %(status)s, %(description)s, %(created_ts)s, %(metadata)s)
              ON CONFLICT (payment_id) DO UPDATE SET
                  status = EXCLUDED.status,
                  synced_at = NOW()
          """, {**r, 'metadata': r['metadata']})

      conn.commit()
      cur.close()
      conn.close()
      print(f"Synced {len(records)} records to crm_payments table")
    env:
      POSTGRES_HOST: "{{ secret('POSTGRES_HOST') }}"
      POSTGRES_DB: "{{ secret('POSTGRES_DB') }}"
      POSTGRES_USER_VAL: "{{ secret('POSTGRES_USER') }}"
      POSTGRES_PASS_VAL: "{{ secret('POSTGRES_PASSWORD') }}"

  - id: update_sync_timestamp
    type: io.kestra.plugin.core.kv.Set
    key: stripe_crm_last_sync_ts
    value: "{{ now() | date('yyyy-MM-dd HH:mm:ss') }}"
    kvType: STRING

  - id: notify_slack
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: ":white_check_mark: Stripe → CRM sync complete: *{{
      outputs.merge_and_enrich.vars.count }}* payments synchronized at {{ now()
      | date('yyyy-MM-dd HH:mm') }} UTC"

triggers:
  - id: daily_sync
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 6 * * *"
    description: Sync Stripe payments to CRM database every day at 06:00 UTC