amber-backend/scripts/test_client_logs_trim.py

"""Retention check for the nightly `trimClientLogs` cron (pb_hooks/client_logs.pb.js).

The trim must remove EVERY client_logs row older than 14 days in one run, not a
fixed page of them: on 2026-09-13 production held ~12300 stale rows because the
job deleted 500 a night while ~1275 arrived a day.

Starts a throwaway PocketBase on a temp data dir with this repo's migrations and
hooks, seeds stale and fresh rows straight into SQLite (`created` is an autodate
the API will not let us set), fires the job through the superuser cron API and
asserts what is left. Needs the pinned PocketBase binary (see Dockerfile):

    PB_BIN=/path/to/pocketbase python scripts/test_client_logs_trim.py

`--hooks DIR` points at another hooks dir, e.g. the previous version of the hook,
to watch the test fail. Exits non-zero if any check fails.
"""
import argparse, json, os, random, shutil, socket, sqlite3, string, subprocess
import sys, tempfile, time, urllib.error, urllib.request
from datetime import datetime, timedelta, timezone

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
ADMIN_EMAIL, ADMIN_PASS = "admin@trim.local", "Sup3rSecret!123"
DAY = timedelta(days=1)

ap = argparse.ArgumentParser()
ap.add_argument("--hooks", default=os.path.join(ROOT, "pb_hooks"))
ap.add_argument("--stale", type=int, default=1300,
                help="rows older than 14 days; more than one old 500-row page")
args = ap.parse_args()
pb_bin = os.environ.get("PB_BIN") or shutil.which("pocketbase")
if not pb_bin:
    sys.exit("PB_BIN not set and no pocketbase on PATH")

with socket.socket() as s:
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
base = f"http://127.0.0.1:{port}"

def req(method, path, token=None, body=None):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(base + path, data=data, method=method)
    r.add_header("Content-Type", "application/json")
    if token:
        r.add_header("Authorization", token)
    try:
        with urllib.request.urlopen(r) as resp:
            return resp.status, json.loads(resp.read() or "{}")
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read() or "{}")

def pb_time(dt):
    return dt.strftime("%Y-%m-%d %H:%M:%S.") + f"{dt.microsecond // 1000:03d}Z"

def rid():
    return "".join(random.choices(string.ascii_lowercase + string.digits, k=15))

allpass = True
def check(label, ok, detail=""):
    global allpass
    print(f"[{'PASS' if ok else 'FAIL'}] {label}{' -> ' + detail if detail else ''}")
    allpass = ok and allpass

tmp = tempfile.mkdtemp(prefix="pb-trim-")
data_dir = os.path.join(tmp, "pb_data")
common = [f"--dir={data_dir}", f"--migrationsDir={os.path.join(ROOT, 'pb_migrations')}",
          f"--hooksDir={args.hooks}"]
log = open(os.path.join(tmp, "pb.log"), "w")
proc = subprocess.Popen([pb_bin, "serve", f"--http=127.0.0.1:{port}", *common],
                        stdout=log, stderr=subprocess.STDOUT)
try:
    for _ in range(100):
        try:
            if req("GET", "/api/health")[0] == 200:
                break
        except OSError:
            pass
        time.sleep(0.1)
    else:
        sys.exit(f"PocketBase did not come up, see {log.name}")

    subprocess.run([pb_bin, "superuser", "upsert", ADMIN_EMAIL, ADMIN_PASS, *common],
                   check=True, capture_output=True)
    _, auth = req("POST", "/api/collections/_superusers/auth-with-password",
                  body={"identity": ADMIN_EMAIL, "password": ADMIN_PASS})
    admin = auth["token"]
    st, user = req("POST", "/api/collections/users/records", admin, {
        "email": "trim@trim.local", "username": "trimtest",
        "password": "Passw0rd!123", "passwordConfirm": "Passw0rd!123"})
    if st != 200:
        sys.exit(f"could not create a user: {st} {user}")

    now = datetime.now(timezone.utc)
    stale = [now - 14 * DAY - timedelta(minutes=5) - i * timedelta(minutes=17)
             for i in range(args.stale)]
    # Kept: just inside the window, mid-window, and brand new.
    fresh = [now - 14 * DAY + timedelta(hours=1), now - 13 * DAY, now - 7 * DAY,
             now - timedelta(minutes=1)] * 10
    db = sqlite3.connect(os.path.join(data_dir, "data.db"), timeout=10)
    db.executemany(
        "INSERT INTO client_logs (id, user, kind, event, created) VALUES (?,?,?,?,?)",
        [(rid(), user["id"], "info", "trim_test", pb_time(t)) for t in stale + fresh])
    db.commit()
    cutoff = pb_time(now - 14 * DAY)

    def count(where, *params):
        return db.execute(f"SELECT COUNT(*) FROM client_logs WHERE {where}", params).fetchone()[0]

    check("seeded", count("created < ?", cutoff) == len(stale)
          and count("created >= ?", cutoff) == len(fresh),
          f"{len(stale)} stale, {len(fresh)} fresh")

    started = time.monotonic()
    st, body = req("POST", "/api/crons/trimClientLogs", admin)
    check("cron job exists and fires", st == 204, f"HTTP {st} {body or ''}")
    # The API runs the job in the background: wait until the count settles.
    last, still_since = count("created < ?", cutoff), time.monotonic()
    while time.monotonic() - started < 60:
        time.sleep(0.2)
        n = count("created < ?", cutoff)
        if n != last:
            last, still_since = n, time.monotonic()
        elif n == 0 or time.monotonic() - still_since > 3:
            break
    elapsed = time.monotonic() - started

    check("no row older than 14 days after one run", last == 0,
          f"{last} of {len(stale)} stale rows left")
    check("rows inside the window are kept", count("created >= ?", cutoff) == len(fresh),
          f"{count('created >= ?', cutoff)} of {len(fresh)}")
    print(f"      run settled in {elapsed:.1f}s")
    db.close()

    # What proves it ran in production: the job's own line in PocketBase's logs,
    # which are batched to auxiliary.db and flushed on shutdown.
    proc.terminate()
    proc.wait(timeout=10)
    aux = sqlite3.connect(os.path.join(data_dir, "auxiliary.db"))
    lines = [json.loads(d) for (d,) in aux.execute(
        "SELECT data FROM _logs WHERE message = 'trimClientLogs'")]
    aux.close()
    check("run logs how many rows it deleted",
          len(lines) == 1 and lines[0].get("deleted") == len(stale), str(lines))
finally:
    if proc.poll() is None:
        proc.terminate()
        proc.wait(timeout=10)
    log.close()
    if allpass:
        shutil.rmtree(tmp, ignore_errors=True)
    else:
        print(f"PocketBase log: {os.path.join(tmp, 'pb.log')}")

print("ALL PASS" if allpass else "SOME CHECKS FAILED")
sys.exit(0 if allpass else 1)