diff --git a/billing_db.py b/billing_db.py
new file mode 100644
index 0000000..93d5096
--- /dev/null
+++ b/billing_db.py
@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""
+DB Migration — Billing Stripe
+Sprint 5-6: HRT-31
+
+Adds stripe_subscription_id and status columns to subscriptions table,
+and an invoices / grace-period tracking table.
+
+Run once:
+    ./venv/bin/python billing_db.py
+"""
+
+import sqlite3
+import os
+import logging
+
+DB_PATH = os.environ.get("TURF_SAAS_DB", "/home/h3r7/turf_saas/turf_saas.db")
+logger = logging.getLogger("turf_saas.billing_db")
+
+
+def get_db():
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+def migrate_billing_tables():
+    """Idempotent migration: add billing columns and billing_events table.
+
+    Requires auth tables (users, subscriptions) to exist first.
+    Calls init_auth_tables() automatically if subscriptions is absent.
+    """
+    from auth_db import init_auth_tables as _init_auth
+
+    conn = get_db()
+    c = conn.cursor()
+
+    # Ensure base auth tables exist
+    tables = {
+        row[0] for row in c.execute("SELECT name FROM sqlite_master WHERE type='table'")
+    }
+    conn.close()
+
+    if "subscriptions" not in tables:
+        _init_auth()
+
+    conn = get_db()
+    c = conn.cursor()
+
+    # Add stripe_subscription_id if missing
+    columns = {row[1] for row in c.execute("PRAGMA table_info(subscriptions)")}
+
+    if "stripe_subscription_id" not in columns:
+        c.execute("ALTER TABLE subscriptions ADD COLUMN stripe_subscription_id TEXT")
+        logger.info("[billing_db] Added stripe_subscription_id column to subscriptions")
+
+    if "status" not in columns:
+        c.execute(
+            "ALTER TABLE subscriptions ADD COLUMN "
+            "status TEXT NOT NULL DEFAULT 'active' "
+            "CHECK(status IN ('active','past_due','canceled','trialing','incomplete'))"
+        )
+        logger.info("[billing_db] Added status column to subscriptions")
+
+    if "grace_period_end" not in columns:
+        c.execute("ALTER TABLE subscriptions ADD COLUMN grace_period_end DATETIME")
+        logger.info("[billing_db] Added grace_period_end column to subscriptions")
+
+    if "current_period_end" not in columns:
+        c.execute("ALTER TABLE subscriptions ADD COLUMN current_period_end DATETIME")
+        logger.info("[billing_db] Added current_period_end column to subscriptions")
+
+    # billing_events table — audit trail for all webhook events
+    c.executescript("""
+        CREATE TABLE IF NOT EXISTS billing_events (
+            id              INTEGER PRIMARY KEY AUTOINCREMENT,
+            stripe_event_id TEXT    NOT NULL UNIQUE,
+            event_type      TEXT    NOT NULL,
+            user_id         INTEGER REFERENCES users(id),
+            payload         TEXT,
+            processed_at    DATETIME NOT NULL DEFAULT (datetime('now'))
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_billing_events_user  ON billing_events(user_id);
+        CREATE INDEX IF NOT EXISTS idx_billing_events_type  ON billing_events(event_type);
+        CREATE INDEX IF NOT EXISTS idx_subscriptions_stripe ON subscriptions(stripe_subscription_id);
+        CREATE INDEX IF NOT EXISTS idx_subscriptions_customer ON subscriptions(stripe_customer_id);
+    """)
+
+    conn.commit()
+    conn.close()
+    print(
+        "[billing_db] Migration complete: subscriptions + billing_events tables ready."
+    )
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    migrate_billing_tables()
+
+
+# ──────────────────────────────────────────────────────────────
+# Re-exported helpers for test usage
+# (primary implementations live in api_v1/routes/billing.py)
+# ──────────────────────────────────────────────────────────────
+
+
+def _upsert_subscription(db, user_id: int, **fields):
+    """
+    Update existing subscription row or insert a new one.
+    Convenience re-export for test helpers.
+    """
+    existing = db.execute(
+        "SELECT id FROM subscriptions WHERE user_id = ? ORDER BY start_date DESC LIMIT 1",
+        (user_id,),
+    ).fetchone()
+    if existing:
+        set_parts = ", ".join(f"{k} = ?" for k in fields)
+        values = list(fields.values()) + [existing["id"]]
+        db.execute(f"UPDATE subscriptions SET {set_parts} WHERE id = ?", values)
+    else:
+        cols = ", ".join(["user_id"] + list(fields.keys()))
+        placeholders = ", ".join(["?"] * (1 + len(fields)))
+        values = [user_id] + list(fields.values())
+        db.execute(
+            f"INSERT INTO subscriptions ({cols}) VALUES ({placeholders})", values
+        )