undelivered events are aging out of the 30-day window

"""Report undelivered Stripe events approaching the 30-day retention cliff.

Read only. One paginated GET and no writes: give this a RESTRICTED key with read
access to Events. The replay 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_event_retention")

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

RETENTION_DAYS = 30   # events leave /v1/events entirely at this age
CRITICAL_DAYS = 29    # gone tomorrow
WARN_DAYS = 20        # still replayable, but schedule it now


def verdict(oldest_age_days, count):
    """Classify the backlog. Pure, so the boundaries can be tested without a network.

    `oldest_age_days` is the age of the oldest undelivered event in days, or None
    when nothing is undelivered. Returns (state, detail).
    """
    if not count:
        return ("clear", "0 undelivered event(s) in the retained window")
    if oldest_age_days is None:
        return ("unknown",
                "%d undelivered event(s) but no usable created timestamp" % count)
    left = RETENTION_DAYS - oldest_age_days
    if oldest_age_days >= CRITICAL_DAYS:
        return ("expiring",
                "%d event(s); the oldest is %.1f days old and leaves the API in "
                "under a day. Replay oldest first, now." % (count, oldest_age_days))
    if oldest_age_days >= WARN_DAYS:
        return ("aging",
                "%d event(s); the oldest expires in %.1f days. Schedule the replay "
                "rather than discussing it." % (count, left))
    return ("replayable",
            "%d event(s); the oldest expires in %.1f days. There is room to replay "
            "carefully." % (count, left))


def get(session, path, **params):
    r = session.get(API + path, params=params, 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 undelivered(session, limit):
    """Return (count, oldest_created, oldest_event_id) for undelivered events.

    Stripe returns events newest first, so the oldest one is on the last page and
    the pagination cannot be short-circuited if the age is to be trusted.
    """
    count = 0
    oldest = None
    oldest_id = None
    params = {"delivery_success": "false", "limit": 100}
    while True:
        page = get(session, "/events", **params)
        data = page.get("data", [])
        for ev in data:
            count += 1
            created = ev.get("created")
            if created is not None and (oldest is None or created < oldest):
                oldest, oldest_id = created, ev.get("id")
        if not data or not page.get("has_more") or count >= limit:
            break
        params["starting_after"] = data[-1]["id"]
    return count, oldest, oldest_id


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-events", type=int, default=5000,
                    help="stop paginating after this many undelivered events")
    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})

    count, oldest, oldest_id = undelivered(s, args.max_events)
    age = None if oldest is None else (time.time() - oldest) / 86400.0
    state, detail = verdict(age, count)

    line = "%-11s %s" % (state, detail)
    if state == "clear":
        log.info(line)
        return 0

    log.warning(line)
    log.warning("  replay oldest first, walking backwards from the tail:")
    log.warning("  GET %s/events?delivery_success=false&ending_before=%s",
                API, oldest_id or "<evt_id>")
    if state == "expiring":
        log.warning("  anything already past %d days: reconcile from the objects "
                    "instead, which have no retention limit:", RETENTION_DAYS)
        log.warning("  GET %s/charges?created[gte]=<unix>   "
                    "GET %s/invoices?created[gte]=<unix>", API, API)
    return 1


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

/**
 * Report undelivered Stripe events approaching the 30-day retention cliff.
 *
 * Read only. One paginated GET and no writes: give this a RESTRICTED key with
 * read access to Events. The replay is printed, never performed.
 */
const API = 'https://api.stripe.com/v1';

export const RETENTION_DAYS = 30; // events leave /v1/events entirely at this age
const CRITICAL_DAYS = 29;         // gone tomorrow
const WARN_DAYS = 20;             // still replayable, but schedule it now

/**
 * Classify the backlog. Pure, so the boundaries can be tested without a network.
 * `oldestAgeDays` is null when nothing is undelivered.
 */
export function verdict(oldestAgeDays, count) {
  if (!count) return ['clear', '0 undelivered event(s) in the retained window'];
  if (oldestAgeDays === null || oldestAgeDays === undefined) {
    return ['unknown', `${count} undelivered event(s) but no usable created timestamp`];
  }
  const left = RETENTION_DAYS - oldestAgeDays;
  if (oldestAgeDays >= CRITICAL_DAYS) {
    return ['expiring',
      `${count} event(s); the oldest is ${oldestAgeDays.toFixed(1)} days old and ` +
      'leaves the API in under a day. Replay oldest first, now.'];
  }
  if (oldestAgeDays >= WARN_DAYS) {
    return ['aging',
      `${count} event(s); the oldest expires in ${left.toFixed(1)} days. ` +
      'Schedule the replay rather than discussing it.'];
  }
  return ['replayable',
    `${count} event(s); the oldest expires in ${left.toFixed(1)} days. ` +
    'There is room to replay carefully.'];
}

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 undelivered(key, limit = 5000) {
  let count = 0;
  let oldest = null;
  let oldestId = null;
  const params = { delivery_success: 'false', limit: 100 };
  for (;;) {
    const page = await get(key, '/events', params);
    const data = page.data ?? [];
    for (const ev of data) {
      count += 1;
      if (ev.created !== undefined && (oldest === null || ev.created < oldest)) {
        oldest = ev.created;
        oldestId = ev.id;
      }
    }
    if (data.length === 0 || !page.has_more || count >= limit) break;
    params.starting_after = data[data.length - 1].id;
  }
  return { count, oldest, oldestId };
}

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 { count, oldest, oldestId } = await undelivered(key);
  const age = oldest === null ? null : (Date.now() / 1000 - oldest) / 86400;
  const [state, detail] = verdict(age, count);

  const line = `${state.padEnd(11)} ${detail}`;
  if (state === 'clear') { console.log(line); return; }

  console.warn(line);
  console.warn('  replay oldest first, walking backwards from the tail:');
  console.warn(`  GET ${API}/events?delivery_success=false&ending_before=${oldestId ?? '<evt_id>'}`);
  if (state === 'expiring') {
    console.warn(`  anything already past ${RETENTION_DAYS} days: reconcile from the ` +
                 'objects instead, which have no retention limit:');
    console.warn(`  GET ${API}/charges?created[gte]=<unix>   GET ${API}/invoices?created[gte]=<unix>`);
  }
  process.exitCode = 1;
}

// 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_event_retention import verdict


def test_nothing_undelivered_is_clear():
    state, _ = verdict(None, 0)
    assert state == "clear"


def test_fresh_backlog_is_replayable():
    state, detail = verdict(3.0, 40)
    assert state == "replayable"
    assert "27.0" in detail


def test_twenty_days_is_the_warning_boundary():
    # Exactly 20 must already warn. A check that flips on day 21 has spent a
    # third of what is left before it says anything.
    assert verdict(19.9, 5)[0] == "replayable"
    assert verdict(20.0, 5)[0] == "aging"


def test_twenty_nine_days_is_the_last_call():
    assert verdict(28.9, 5)[0] == "aging"
    state, detail = verdict(29.0, 5)
    assert state == "expiring"
    assert "under a day" in detail


def test_count_without_a_timestamp_is_not_silently_clear():
    state, _ = verdict(None, 12)
    assert state == "unknown"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { verdict } from './stripe-event-retention.mjs';

test('nothing undelivered is clear', () => {
  assert.equal(verdict(null, 0)[0], 'clear');
});

test('fresh backlog is replayable', () => {
  const [state, detail] = verdict(3.0, 40);
  assert.equal(state, 'replayable');
  assert.match(detail, /27\.0/);
});

test('twenty days is the warning boundary', () => {
  assert.equal(verdict(19.9, 5)[0], 'replayable');
  assert.equal(verdict(20.0, 5)[0], 'aging');
});

test('twenty nine days is the last call', () => {
  assert.equal(verdict(28.9, 5)[0], 'aging');
  const [state, detail] = verdict(29.0, 5);
  assert.equal(state, 'expiring');
  assert.match(detail, /under a day/);
});

test('count without a timestamp is not silently clear', () => {
  assert.equal(verdict(null, 12)[0], 'unknown');
});