a deactivated Payment Link is still linked from your site

"""Report Stripe Payment Links that are deactivated but still receiving traffic.

Read only. Two GETs and no writes: give this a RESTRICTED key with read access to
Payment Links and Checkout Sessions. The repair is printed, never performed,
because this script holds a credential to a live payments account.
"""
import argparse
import logging
import os
import sys
import time

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("stripe_inactive_payment_links")

API = "https://api.stripe.com/v1"


def verdict(active, recent_sessions, inactive_message=None):
    """Classify one Payment Link. Pure, so the rules can be tested without a network.

    `active` is the link's flag, `recent_sessions` the number of Checkout Sessions
    it created inside the window, and `inactive_message` the custom text shown on
    the deactivation page. Returns (state, detail).
    """
    if active is None:
        return ("unknown",
                "the link has no active flag; treat it as published until you know "
                "otherwise (%d recent session(s))" % recent_sessions)
    if active:
        return ("live", "%d session(s) in the window" % recent_sessions)
    if recent_sessions:
        if inactive_message:
            return ("dead-signposted",
                    "inactive, %d recent session(s), and customers at least see: %r"
                    % (recent_sessions, inactive_message))
        return ("dead-in-use",
                "inactive but still reached %d time(s) in the window: every one of "
                "those landed on Stripe's deactivation page" % recent_sessions)
    return ("dormant", "inactive and nothing has reached it in the window")


def get(session, path, params=None):
    r = session.get(API + path, params=params or {}, timeout=30)
    if r.status_code == 401:
        raise SystemExit("401 from Stripe: the key is wrong, or is for the other mode")
    r.raise_for_status()
    return r.json()


def payment_links(session, cap):
    """Every Payment Link on the account, active and not."""
    out = []
    params = {"limit": 100}
    while True:
        page = get(session, "/payment_links", params)
        data = page.get("data", [])
        out.extend(data)
        if not data or not page.get("has_more") or len(out) >= cap:
            break
        params["starting_after"] = data[-1]["id"]
    return out


def recent_session_count(session, link_id, since):
    """Sessions this link created since `since`.

    Only the count matters, but the pagination still has to happen: a busy link
    can fill a page with sessions from a single afternoon.
    """
    count = 0
    params = {"payment_link": link_id, "limit": 100}
    while True:
        page = get(session, "/checkout/sessions", params)
        data = page.get("data", [])
        for cs in data:
            if (cs.get("created") or 0) >= since:
                count += 1
        if not data or not page.get("has_more"):
            break
        if (data[-1].get("created") or 0) < since:
            break
        params["starting_after"] = data[-1]["id"]
    return count


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=90,
                    help="how far back a session still counts as recent traffic")
    ap.add_argument("--max-links", type=int, default=500,
                    help="stop paginating after this many Payment Links")
    args = ap.parse_args()

    key = os.environ.get("STRIPE_API_KEY")
    if not key:
        log.error("set STRIPE_API_KEY (use a restricted, read-only key)")
        return 2

    s = requests.Session()
    s.headers.update({"Authorization": "Bearer " + key})

    since = int(time.time()) - args.days * 86400
    bad = 0
    for link in payment_links(s, args.max_links):
        count = recent_session_count(s, link["id"], since)
        state, detail = verdict(link.get("active"), count, link.get("inactive_message"))
        line = "%-15s %-20s %s" % (state, link["id"], detail)
        if state in ("live", "dormant"):
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        log.warning("  published at: %s", link.get("url") or "<no url>")
        log.warning("  repair: repoint the published URL at a live link, or bring "
                    "this one back:")
        log.warning("  POST %s/payment_links/%s -d active=true", API, link["id"])
        if not link.get("inactive_message"):
            log.warning("  if it stays dead, give the deactivation page a "
                        "forwarding instruction with -d inactive_message=\"...\"")

    log.info("%d inactive link(s) still taking traffic", bad)
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Report Stripe Payment Links that are deactivated but still receiving traffic.
 *
 * Read only. Two GETs and no writes: give this a RESTRICTED key with read access
 * to Payment Links and Checkout Sessions. The repair is printed, never performed.
 */
const API = 'https://api.stripe.com/v1';

/**
 * Classify one Payment Link. Pure, so the rules can be tested without a network.
 * `active` may be null, which is not the same as false.
 */
export function verdict(active, recentSessions, inactiveMessage = null) {
  if (active === null || active === undefined) {
    return ['unknown',
      'the link has no active flag; treat it as published until you know ' +
      `otherwise (${recentSessions} recent session(s))`];
  }
  if (active) return ['live', `${recentSessions} session(s) in the window`];
  if (recentSessions) {
    if (inactiveMessage) {
      return ['dead-signposted',
        `inactive, ${recentSessions} recent session(s), and customers at least ` +
        `see: '${inactiveMessage}'`];
    }
    return ['dead-in-use',
      `inactive but still reached ${recentSessions} time(s) in the window: every ` +
      "one of those landed on Stripe's deactivation page"];
  }
  return ['dormant', 'inactive and nothing has reached it in the window'];
}

async function get(key, path, params = {}) {
  const url = new URL(API + path);
  for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);
  const res = await fetch(url, { headers: { Authorization: `Bearer ${key}` } });
  if (res.status === 401) {
    throw new Error('401 from Stripe: the key is wrong, or is for the other mode');
  }
  if (!res.ok) throw new Error(`${res.status} from ${url.pathname}`);
  return res.json();
}

export async function paymentLinks(key, cap = 500) {
  const out = [];
  const params = { limit: 100 };
  for (;;) {
    const page = await get(key, '/payment_links', params);
    const data = page.data ?? [];
    out.push(...data);
    if (data.length === 0 || !page.has_more || out.length >= cap) break;
    params.starting_after = data[data.length - 1].id;
  }
  return out;
}

export async function recentSessionCount(key, linkId, since) {
  let count = 0;
  const params = { payment_link: linkId, limit: 100 };
  for (;;) {
    const page = await get(key, '/checkout/sessions', params);
    const data = page.data ?? [];
    for (const cs of data) if ((cs.created ?? 0) >= since) count += 1;
    if (data.length === 0 || !page.has_more) break;
    if ((data[data.length - 1].created ?? 0) < since) break;
    params.starting_after = data[data.length - 1].id;
  }
  return count;
}

async function main() {
  const key = process.env.STRIPE_API_KEY;
  if (!key) {
    console.error('set STRIPE_API_KEY (use a restricted, read-only key)');
    process.exitCode = 2;
    return;
  }

  const days = Number(process.argv[2] ?? 90);
  const since = Math.floor(Date.now() / 1000) - days * 86400;
  let bad = 0;
  for (const link of await paymentLinks(key)) {
    const count = await recentSessionCount(key, link.id, since);
    const [state, detail] = verdict(link.active, count, link.inactive_message);
    const line = `${state.padEnd(15)} ${link.id.padEnd(20)} ${detail}`;
    if (state === 'live' || state === 'dormant') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    console.warn(`  published at: ${link.url ?? '<no url>'}`);
    console.warn('  repair: repoint the published URL at a live link, or bring ' +
                 'this one back:');
    console.warn(`  POST ${API}/payment_links/${link.id} -d active=true`);
    if (!link.inactive_message) {
      console.warn('  if it stays dead, give the deactivation page a forwarding ' +
                   'instruction with -d inactive_message="..."');
    }
  }

  console.log(`${bad} inactive link(s) still taking traffic`);
  process.exitCode = bad ? 1 : 0;
}

// Only run when invoked directly. The test file imports this module, and without
// the guard main() would run there too, fail on the missing key, and set a
// non-zero exit code that fails the whole test file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from stripe_inactive_payment_links import verdict


def test_an_active_link_is_live():
    assert verdict(True, 12)[0] == "live"


def test_a_dead_link_with_recent_traffic_is_the_expensive_case():
    state, detail = verdict(False, 9)
    assert state == "dead-in-use"
    assert "9 time(s)" in detail


def test_an_inactive_message_softens_it_but_does_not_clear_it():
    state, detail = verdict(False, 9, "We moved to the new plan page")
    assert state == "dead-signposted"
    assert "new plan page" in detail


def test_a_dead_link_nobody_visits_is_only_housekeeping():
    assert verdict(False, 0)[0] == "dormant"


def test_a_missing_active_flag_is_not_read_as_deactivated():
    # `if not active` would call this dead and print a repair for a link that is
    # working perfectly. Absent is not false.
    assert verdict(None, 3)[0] == "unknown"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { verdict } from './stripe-inactive-payment-links.mjs';

test('an active link is live', () => {
  assert.equal(verdict(true, 12)[0], 'live');
});

test('a dead link with recent traffic is the expensive case', () => {
  const [state, detail] = verdict(false, 9);
  assert.equal(state, 'dead-in-use');
  assert.match(detail, /9 time\(s\)/);
});

test('an inactive message softens it but does not clear it', () => {
  const [state, detail] = verdict(false, 9, 'We moved to the new plan page');
  assert.equal(state, 'dead-signposted');
  assert.match(detail, /new plan page/);
});

test('a dead link nobody visits is only housekeeping', () => {
  assert.equal(verdict(false, 0)[0], 'dormant');
});

test('a missing active flag is not read as deactivated', () => {
  assert.equal(verdict(null, 3)[0], 'unknown');
  assert.equal(verdict(undefined, 3)[0], 'unknown');
});