"""Delete conversations older than the retention bound.

WHY THIS IS A SCRIPT AND NOT A CELERY BEAT TASK
-----------------------------------------------
The plan history *window* only HIDES data — it is reversible, and a user who
upgrades gets everything back. Retention is the opposite: it destroys customer
conversations permanently. Those two must never be confused, and irreversible
deletion of customer data must not run unattended on a schedule that nobody
reviewed. So this is opt-in, defaults to a dry run, and prints exactly what it
would remove before anything happens.

Wire it to a schedule only after deciding the retention period as a matter of
policy (privacy notice, DPA), not as a storage optimisation — and note that
deleting data a user could otherwise recover by upgrading is a support and trust
problem as much as a legal one.

Retention is measured on ``AddonsThreads.last_activity_at``, the same field the
plan window uses, so "old" means one thing across the product.

Usage:
    venv/bin/python -m scripts.prune_history --months 24                # dry run
    venv/bin/python -m scripts.prune_history --months 24 --org free-sandbox
    venv/bin/python -m scripts.prune_history --months 24 --apply        # deletes
"""
import argparse
import datetime as dt

from sqlalchemy import text

from supporthub.app.db import engine, session_scope

# Child tables to clear before their parent, mapped to the key they hang off.
# The FK column is VERIFIED against information_schema at runtime rather than
# trusted from this list — an earlier version assumed message_topic_segments keyed
# on message_id (it is thread_id), the DELETE failed, that aborted the whole
# transaction, every later DELETE then failed too, and the script still printed
# "Done. 1 threads removed" while deleting nothing. Introspection plus a SAVEPOINT
# per statement makes both halves of that impossible.
CHILD_TABLES = [
    ("addons_attachments", "message_id"),
    ("message_translations", "message_id"),
    ("message_topic_segments", "thread_id"),
    ("ai_suggestions", "ticket_id"),
    ("ticket_checklist_items", "ticket_id"),
    ("ticket_drafts", "ticket_id"),
    ("ticket_tags", "ticket_id"),
    ("thread_embeddings", "thread_id"),
    ("guide_candidate_dismissals", "thread_id"),
]


def fk_column(conn, schema, table, expected):
    """Return `expected` if that column really exists on the table, else None."""
    found = conn.execute(text(
        "SELECT 1 FROM information_schema.columns WHERE table_schema=:s"
        " AND table_name=:t AND column_name=:c"),
        {"s": schema, "t": table, "c": expected}).scalar()
    return expected if found else None


def tenant_schemas(slug=None):
    sql = "SELECT slug, schema_name FROM public.organizations"
    params = {}
    if slug:
        sql += " WHERE slug = :slug"
        params["slug"] = slug
    sql += " ORDER BY id"
    with engine.connect() as c:
        return c.execute(text(sql), params).fetchall()


def count_old(schema, cutoff):
    with engine.connect() as c:
        c.execute(text(f'SET search_path TO "{schema}", public'))
        threads = c.execute(text(
            "SELECT count(*) FROM addons_threads WHERE last_activity_at < :c"),
            {"c": cutoff}).scalar()
        messages = c.execute(text(
            "SELECT count(*) FROM addons_messages m JOIN addons_threads t"
            " ON t.id = m.thread_id WHERE t.last_activity_at < :c"),
            {"c": cutoff}).scalar()
        total = c.execute(text("SELECT count(*) FROM addons_threads")).scalar()
    return threads, messages, total


def prune(schema, cutoff):
    """Delete old threads and their dependants.

    Returns (rows_removed_per_table, threads_actually_deleted). The second value
    is read from the parent DELETE's rowcount, never from the candidate count, so
    the summary cannot claim a deletion that did not happen.
    """
    removed = {}
    threads_deleted = 0
    with engine.connect() as probe:
        plan = [(t, fk_column(probe, schema, t, col)) for t, col in CHILD_TABLES]
    missing = [t for t, col in plan if col is None]
    if missing:
        print(f"    note: absent in this schema, skipped: {', '.join(missing)}")

    with session_scope(schema=schema) as s:
        s.execute(text(f'SET search_path TO "{schema}", public'))
        ids = [r[0] for r in s.execute(text(
            "SELECT id FROM addons_threads WHERE last_activity_at < :c"), {"c": cutoff})]
        if not ids:
            return removed, 0

        ticket_ids = [r[0] for r in s.execute(text(
            "SELECT id FROM tickets WHERE thread_id = ANY(:ids)"), {"ids": ids})]
        message_ids = [r[0] for r in s.execute(text(
            "SELECT id FROM addons_messages WHERE thread_id = ANY(:ids)"), {"ids": ids})]
        key_values = {"thread_id": ids, "ticket_id": ticket_ids, "message_id": message_ids}

        def _delete(table, column, values):
            """Delete inside a SAVEPOINT so one failure cannot abort the whole
            transaction and silently turn every later statement into a no-op."""
            nonlocal removed
            if not values:
                return 0
            try:
                with s.begin_nested():
                    n = s.execute(text(
                        f"DELETE FROM {table} WHERE {column} = ANY(:v)"),
                        {"v": values}).rowcount
                if n:
                    removed[table] = removed.get(table, 0) + n
                return n
            except Exception as exc:
                print(f"    FAILED {table}: {exc.__class__.__name__}: "
                      f"{str(exc).splitlines()[0][:120]}")
                return 0

        for table, column in plan:
            if column is None:
                continue
            _delete(table, column, key_values[column])
        _delete("addons_messages", "thread_id", ids)
        _delete("tickets", "thread_id", ids)
        threads_deleted = _delete("addons_threads", "id", ids)

    return removed, threads_deleted


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--months", type=int, required=True,
                    help="delete conversations with no activity for this many months")
    ap.add_argument("--org", help="limit to one organization slug")
    ap.add_argument("--apply", action="store_true",
                    help="actually delete (default is a dry run)")
    args = ap.parse_args()

    if args.months < 6:
        raise SystemExit(
            "Refusing a retention period under 6 months — that is short enough to "
            "destroy conversations a customer is still actively discussing.")

    cutoff = dt.datetime.utcnow() - dt.timedelta(days=args.months * 30)
    print(f"Retention: {args.months} months → deleting activity before "
          f"{cutoff:%Y-%m-%d}")
    print("MODE: " + ("APPLY (irreversible)" if args.apply else "DRY RUN"))
    print()

    grand_threads = 0
    for row in tenant_schemas(args.org):
        threads, messages, total = count_old(row.schema_name, cutoff)
        print(f"{row.slug:20} {threads:>6} of {total:>6} threads, "
              f"{messages:>7} messages older than the cutoff")
        if args.apply and threads:
            removed, deleted = prune(row.schema_name, cutoff)
            for tbl, n in sorted(removed.items()):
                print(f"    deleted {n:>7} from {tbl}")
            if deleted != threads:
                print(f"    WARNING: {threads} candidates but {deleted} threads "
                      f"actually deleted — investigate before trusting this run")
            grand_threads += deleted
        else:
            grand_threads += threads

    print()
    if not args.apply:
        print(f"DRY RUN — nothing deleted. {grand_threads} threads would be removed.")
        print("Re-run with --apply once the retention period is agreed policy.")
    else:
        print(f"Done. {grand_threads} threads removed.")


if __name__ == "__main__":
    main()