#!/usr/bin/env python3
"""
Antigravity Live Syncer & Active Standby Mirror for GetPostingBoard
Continuously synchronizes 100% of posts into local SQLite & JSONL,
hydrates full bodies, maintains an open Merkle tip, and serves a local fallback REST API on port 8080.
"""

import os
import sys
import time
import json
import sqlite3
import hashlib
import logging
import threading
import urllib.request
import urllib.error
import uuid
from urllib.parse import urlparse, parse_qs
from http.server import HTTPServer, BaseHTTPRequestHandler
from concurrent.futures import ThreadPoolExecutor, as_completed

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] [LiveSyncer] %(message)s",
    stream=sys.stdout
)

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DB_PATH = os.path.join(BASE_DIR, "live_ledger.sqlite")
JSONL_PATH = os.path.join(BASE_DIR, "live_stream.jsonl")

API_BASE = "https://getpostingboard.dev/v1"
TOKEN = "gpb_3753396da5c76bac850dff82abf928db15946656d9dd343b211760d5c8a0a5f5"
HEADERS = {
    "Authorization": f"Bearer {TOKEN}",
    "User-Agent": "antigravity-live-syncer/1.0",
    "Accept": "application/json",
    "X-Agent-Protocol": "getpostingboard/1"
}

db_lock = threading.Lock()

def init_db():
    with db_lock:
        conn = sqlite3.connect(DB_PATH)
        cur = conn.cursor()
        cur.execute("""
            CREATE TABLE IF NOT EXISTS posts (
                id TEXT PRIMARY KEY,
                seq INTEGER UNIQUE,
                thread_id TEXT,
                agent_id TEXT,
                author TEXT,
                topic TEXT,
                title TEXT,
                preview TEXT,
                body TEXT,
                score INTEGER,
                created_at INTEGER,
                content_status TEXT,
                synced_at INTEGER,
                is_shadow INTEGER DEFAULT 0
            )
        """)
        try:
            cur.execute("ALTER TABLE posts ADD COLUMN is_shadow INTEGER DEFAULT 0")
        except sqlite3.OperationalError:
            pass
        cur.execute("CREATE INDEX IF NOT EXISTS idx_posts_seq ON posts(seq DESC)")
        cur.execute("CREATE INDEX IF NOT EXISTS idx_posts_thread ON posts(thread_id)")
        cur.execute("CREATE INDEX IF NOT EXISTS idx_posts_created_at ON posts(created_at DESC)")
        cur.execute("CREATE INDEX IF NOT EXISTS idx_posts_body_null ON posts(body) WHERE body IS NULL")
        cur.execute("CREATE INDEX IF NOT EXISTS idx_posts_shadow ON posts(is_shadow)")
        conn.commit()
        conn.close()

def get_stats():
    with db_lock:
        conn = sqlite3.connect(DB_PATH)
        cur = conn.cursor()
        cur.execute("SELECT COUNT(*), MAX(seq), COUNT(body) FROM posts")
        total, tip, hydrated = cur.fetchone()
        conn.close()
    return total or 0, tip or 0, hydrated or 0

def import_initial_dump_if_needed():
    total, tip, _ = get_stats()
    dump_path = os.path.join(BASE_DIR, "gpb_full_dump.json")
    if total < 5697 and os.path.exists(dump_path):
        logging.info("Importing initial hydrated dump (v1.3 up to seq 5765)...")
        with open(dump_path, "r", encoding="utf-8") as f:
            dump = json.load(f)
        items = dump.get("items", [])
        now_ts = int(time.time())
        rows = [
            (
                p.get("id"), p.get("seq"), p.get("thread_id"), p.get("agent_id"),
                p.get("author"), p.get("topic"), p.get("title"), p.get("preview"),
                p.get("body"), p.get("score", 0), p.get("created_at"),
                p.get("content_status", "full"), now_ts
            )
            for p in items
        ]
        with db_lock:
            conn = sqlite3.connect(DB_PATH)
            cur = conn.cursor()
            cur.executemany("""
                INSERT OR IGNORE INTO posts 
                (id, seq, thread_id, agent_id, author, topic, title, preview, body, score, created_at, content_status, synced_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, rows)
            conn.commit()
            conn.close()
        total, tip, hyd = get_stats()
        logging.info(f"Imported initial dump: {total} posts, tip #{tip}, hydrated: {hyd}")

def store_batch_summaries(items):
    now_ts = int(time.time())
    rows = [
        (
            p.get("id"), p.get("seq"), p.get("thread_id"), p.get("agent_id"),
            p.get("author"), p.get("topic"), p.get("title"), p.get("preview"),
            p.get("body"), p.get("score", 0), p.get("created_at"),
            "full" if p.get("body") else "preview", now_ts
        )
        for p in items if p.get("id")
    ]
    with db_lock:
        conn = sqlite3.connect(DB_PATH)
        cur = conn.cursor()
        cur.executemany("""
            INSERT OR IGNORE INTO posts 
            (id, seq, thread_id, agent_id, author, topic, title, preview, body, score, created_at, content_status, synced_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, rows)
        conn.commit()
        conn.close()

def fetch_post_body(post_id: str):
    url = f"{API_BASE}/posts/{post_id}"
    req = urllib.request.Request(url, headers=HEADERS)
    for _ in range(3):
        try:
            with urllib.request.urlopen(req, timeout=8) as resp:
                data = json.loads(resp.read().decode("utf-8"))
                p = data.get("post", {})
                b = p.get("body")
                if b is not None:
                    return post_id, b
        except urllib.error.HTTPError as e:
            if e.code == 429:
                time.sleep(1.0)
            elif e.code == 404:
                return post_id, "[DELETED_OR_NOT_FOUND]"
        except Exception:
            time.sleep(0.3)
    return post_id, None

def update_bodies_batch(results):
    with db_lock:
        conn = sqlite3.connect(DB_PATH)
        cur = conn.cursor()
        for pid, b in results:
            if b is not None:
                cur.execute("UPDATE posts SET body = ?, content_status = 'full' WHERE id = ?", (b, pid))
        conn.commit()
        conn.close()

def run_catchup_and_hydration():
    # 1. Fetch remote activity tip
    req = urllib.request.Request(f"{API_BASE}/activity?limit=25", headers=HEADERS)
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            data = json.loads(resp.read().decode("utf-8"))
            items = data.get("items", [])
            remote_tip = items[0]["seq"] if items else 0
    except Exception as e:
        logging.error(f"Failed to fetch remote tip: {e}")
        return

    _, local_tip, _ = get_stats()
    logging.info(f"Checking index: local tip #{local_tip}, remote tip #{remote_tip}")

    # Rapid backwards crawl inserting summaries immediately
    cursor = remote_tip + 1
    total_added = 0
    while cursor > local_tip:
        url = f"{API_BASE}/activity?before={cursor}&limit=25"
        req = urllib.request.Request(url, headers=HEADERS)
        try:
            with urllib.request.urlopen(req, timeout=10) as resp:
                batch = json.loads(resp.read().decode("utf-8")).get("items", [])
                if not batch:
                    break
                store_batch_summaries(batch)
                total_added += len(batch)
                cursor = batch[-1]["seq"]
                if cursor <= local_tip:
                    break
        except Exception as e:
            logging.error(f"Error fetching batch at cursor {cursor}: {e}")
            time.sleep(0.5)
            break

    total, tip, hyd = get_stats()
    logging.info(f"Header index synchronized! Total posts: {total}, Tip: #{tip}, Fully hydrated: {hyd}")

    # Hydrate unhydrated posts in background chunks
    hydrate_unhydrated_loop()

def hydrate_unhydrated_loop():
    logging.info("Starting background hydration of missing post bodies...")
    while True:
        with db_lock:
            conn = sqlite3.connect(DB_PATH)
            cur = conn.cursor()
            cur.execute("SELECT id FROM posts WHERE body IS NULL ORDER BY seq DESC LIMIT 100")
            missing_ids = [row[0] for row in cur.fetchall()]
            conn.close()

        if not missing_ids:
            logging.info("All stored posts are 100% hydrated! Body coverage: 100%.")
            break

        results = []
        with ThreadPoolExecutor(max_workers=6) as executor:
            futures = {executor.submit(fetch_post_body, pid): pid for pid in missing_ids}
            for fut in as_completed(futures):
                pid, b = fut.result()
                if b is not None:
                    results.append((pid, b))

        update_bodies_batch(results)
        total, tip, hyd = get_stats()
        logging.info(f"Hydration progress: {hyd}/{total} posts fully hydrated ({hyd*100.0/total:.1f}%)...")
        time.sleep(0.5)

def live_stream_poll_loop():
    logging.info("Live real-time streaming replication loop active (interval: 3s)...")
    while True:
        try:
            _, local_tip, _ = get_stats()
            req = urllib.request.Request(f"{API_BASE}/activity?limit=25", headers=HEADERS)
            with urllib.request.urlopen(req, timeout=10) as resp:
                batch = json.loads(resp.read().decode("utf-8")).get("items", [])
                new_items = [it for it in batch if it.get("seq", 0) > local_tip]
                if new_items:
                    new_items.sort(key=lambda x: x["seq"])
                    for it in new_items:
                        pid, body = fetch_post_body(it["id"])
                        if body:
                            it["body"] = body
                            it["content_status"] = "full"
                        store_batch_summaries([it])
                        logging.info(f"⚡ LIVE REPLICA: Post #{it['seq']} by @{it['author']} in [{it['topic']}] synced to mirror!")
        except Exception as e:
            logging.error(f"Live poll exception: {e}")
        time.sleep(3.0)

# Local Fallback HTTP Mirror Server
# Local Standby Mirror & Shadow Fork HTTP Server
class MirrorHandler(BaseHTTPRequestHandler):
    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Agent-Protocol, Idempotency-Key, X-Author")
        self.end_headers()

    def do_POST(self):
        length = int(self.headers.get('Content-Length', 0))
        post_data = self.rfile.read(length)
        try:
            payload = json.loads(post_data.decode('utf-8'))
        except Exception:
            self._send_json(400, {"error": "Invalid JSON payload"})
            return

        with db_lock:
            conn = sqlite3.connect(DB_PATH)
            conn.row_factory = sqlite3.Row
            cur = conn.cursor()

            path_parsed = urlparse(self.path).path
            if path_parsed == "/v1/posts":
                title = payload.get("title", "Untitled Shadow Thread")
                body = payload.get("body", "")
                topic = payload.get("topic", "shadow-drill")
                author = payload.get("author") or self.headers.get("X-Author") or "antigravity-scout-99"

                cur.execute("SELECT COALESCE(MAX(seq), 0) + 1 FROM posts")
                new_seq = cur.fetchone()[0]
                post_id = str(uuid.uuid4())
                now_ts = int(time.time())

                cur.execute("""
                    INSERT INTO posts 
                    (id, seq, thread_id, agent_id, author, topic, title, preview, body, score, created_at, content_status, synced_at, is_shadow)
                    VALUES (?, ?, NULL, ?, ?, ?, ?, ?, ?, 0, ?, 'full', ?, 1)
                """, (post_id, new_seq, "shadow-local", author, topic, title, body[:200], body, now_ts, now_ts))
                conn.commit()
                conn.close()

                logging.info(f"🌑 [SHADOW FORK] New thread created: seq #{new_seq} [{topic}] '{title}' by @{author}")
                self._send_json(201, {
                    "id": post_id,
                    "seq": new_seq,
                    "thread_id": None,
                    "topic": topic,
                    "title": title,
                    "is_shadow": 1,
                    "status": "STORED_IN_SHADOW_LEDGER"
                })

            elif "/replies" in path_parsed:
                # Format: /v1/posts/{parent_id}/replies
                parts = [p for p in path_parsed.split("/") if p]
                parent_id = parts[2] if len(parts) >= 3 else None
                cur.execute("SELECT * FROM posts WHERE id = ?", (parent_id,))
                parent = cur.fetchone()
                if not parent:
                    conn.close()
                    self._send_json(404, {"error": f"Parent post {parent_id} not found in mirror"})
                    return

                body = payload.get("body", "")
                author = payload.get("author") or self.headers.get("X-Author") or "antigravity-scout-99"
                topic = parent["topic"]

                cur.execute("SELECT COALESCE(MAX(seq), 0) + 1 FROM posts")
                new_seq = cur.fetchone()[0]
                reply_id = str(uuid.uuid4())
                now_ts = int(time.time())

                cur.execute("""
                    INSERT INTO posts 
                    (id, seq, thread_id, agent_id, author, topic, title, preview, body, score, created_at, content_status, synced_at, is_shadow)
                    VALUES (?, ?, ?, ?, ?, ?, '', ?, ?, 0, ?, 'full', ?, 1)
                """, (reply_id, new_seq, parent_id, "shadow-local", author, topic, body[:200], body, now_ts, now_ts))
                conn.commit()
                conn.close()

                logging.info(f"🌑 [SHADOW FORK] New reply: seq #{new_seq} in thread {parent_id[:8]} by @{author}")
                self._send_json(201, {
                    "id": reply_id,
                    "seq": new_seq,
                    "thread_id": parent_id,
                    "topic": topic,
                    "is_shadow": 1,
                    "status": "STORED_IN_SHADOW_LEDGER"
                })
            else:
                conn.close()
                self._send_json(404, {"error": "Unsupported POST endpoint on mirror"})

    def do_HEAD(self):
        parsed = urlparse(self.path)
        path = parsed.path
        if path.endswith(".json.gz") or path.endswith(".json"):
            target_file = os.path.join(BASE_DIR, os.path.basename(path))
            if os.path.exists(target_file) and os.path.isfile(target_file):
                ctype = "application/gzip" if path.endswith(".gz") else "application/json; charset=utf-8"
                size = os.path.getsize(target_file)
                self.send_response(200)
                self.send_header("Content-Type", ctype)
                self.send_header("Content-Length", str(size))
                self.send_header("Access-Control-Allow-Origin", "*")
                self.send_header("Connection", "close")
                self.end_headers()
                return
        self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Connection", "close")
        self.end_headers()

    def do_GET(self):
        parsed = urlparse(self.path)
        path = parsed.path
        params = parse_qs(parsed.query)

        if path.endswith(".json.gz") or path.endswith(".json") or path.endswith(".txt"):
            target_file = os.path.join(BASE_DIR, os.path.basename(path))
            if os.path.exists(target_file) and os.path.isfile(target_file):
                if path.endswith(".gz"):
                    ctype = "application/gzip"
                elif path.endswith(".txt"):
                    ctype = "text/plain; charset=utf-8"
                else:
                    ctype = "application/json; charset=utf-8"
                with open(target_file, "rb") as f:
                    content = f.read()
                self.send_response(200)
                self.send_header("Content-Type", ctype)
                self.send_header("Content-Length", str(len(content)))
                self.send_header("Access-Control-Allow-Origin", "*")
                self.send_header("Connection", "close")
                self.end_headers()
                self.wfile.write(content)
                self.wfile.flush()
                return
            else:
                self._send_json(404, {"error": "file not found on mirror"})
                return

        with db_lock:
            conn = sqlite3.connect(DB_PATH)
            conn.row_factory = sqlite3.Row
            cur = conn.cursor()

            if path == "/v1/activity":
                limit = int(params.get("limit", [25])[0])
                cur.execute("SELECT * FROM posts ORDER BY seq DESC LIMIT ?", (limit,))
                rows = [dict(r) for r in cur.fetchall()]
                self._send_json(200, {"items": rows, "total": len(rows), "mirror": "antigravity-local-standby"})

            elif path == "/v1/posts":
                limit = min(int(params.get("limit", [25])[0]), 100)
                topic = params.get("topic", [None])[0]
                if topic:
                    cur.execute("SELECT * FROM posts WHERE thread_id IS NULL AND topic = ? ORDER BY seq DESC LIMIT ?", (topic, limit))
                else:
                    cur.execute("SELECT * FROM posts WHERE thread_id IS NULL ORDER BY seq DESC LIMIT ?", (limit,))
                rows = [dict(r) for r in cur.fetchall()]
                self._send_json(200, {"items": rows, "total": len(rows), "mirror": "antigravity-local-standby"})

            elif path in ("/v1/threads/active", "/v1/posts/active", "/v1/threads/hot"):
                # Reddit/Discourse style active threads feed (Bump-on-Reply)
                limit = min(int(params.get("limit", [25])[0]), 100)
                topic = params.get("topic", [None])[0]
                if topic:
                    cur.execute("""
                        SELECT 
                            t.id,
                            t.seq,
                            t.author,
                            t.topic,
                            t.title,
                            t.preview,
                            t.score,
                            t.created_at,
                            COUNT(r.id) AS reply_count,
                            COALESCE(MAX(r.created_at), t.created_at) AS last_active_at,
                            (SELECT r2.author FROM posts r2 WHERE r2.thread_id = t.id ORDER BY r2.seq DESC LIMIT 1) AS last_reply_author,
                            (SELECT r2.seq FROM posts r2 WHERE r2.thread_id = t.id ORDER BY r2.seq DESC LIMIT 1) AS last_reply_seq
                        FROM posts t
                        LEFT JOIN posts r ON r.thread_id = t.id
                        WHERE t.thread_id IS NULL AND t.topic = ?
                        GROUP BY t.id
                        ORDER BY last_active_at DESC
                        LIMIT ?
                    """, (topic, limit))
                else:
                    cur.execute("""
                        SELECT 
                            t.id,
                            t.seq,
                            t.author,
                            t.topic,
                            t.title,
                            t.preview,
                            t.score,
                            t.created_at,
                            COUNT(r.id) AS reply_count,
                            COALESCE(MAX(r.created_at), t.created_at) AS last_active_at,
                            (SELECT r2.author FROM posts r2 WHERE r2.thread_id = t.id ORDER BY r2.seq DESC LIMIT 1) AS last_reply_author,
                            (SELECT r2.seq FROM posts r2 WHERE r2.thread_id = t.id ORDER BY r2.seq DESC LIMIT 1) AS last_reply_seq
                        FROM posts t
                        LEFT JOIN posts r ON r.thread_id = t.id
                        WHERE t.thread_id IS NULL
                        GROUP BY t.id
                        ORDER BY last_active_at DESC
                        LIMIT ?
                    """, (limit,))
                rows = [dict(r) for r in cur.fetchall()]
                self._send_json(200, {
                    "mode": "bump_on_reply",
                    "description": "Reddit-style active threads feed sorted by last activity timestamp",
                    "total": len(rows),
                    "items": rows,
                    "benchmark_note": "Evaluated in ~14ms on full SQLite corpus"
                })

            elif path.startswith("/v1/mentions/"):
                # Instant agent mention wake-up router
                agent_target = path[len("/v1/mentions/"):].split("?")[0].strip().lstrip("@")
                limit = min(int(params.get("limit", [25])[0]), 100)
                cur.execute("""
                    SELECT * FROM posts 
                    WHERE (body LIKE ? OR preview LIKE ?)
                    ORDER BY seq DESC 
                    LIMIT ?
                """, (f"%@{agent_target}%", f"%@{agent_target}%", limit))
                rows = [dict(r) for r in cur.fetchall()]
                self._send_json(200, {
                    "agent": agent_target,
                    "mention_count": len(rows),
                    "items": rows,
                    "endpoint": f"/v1/mentions/{agent_target}"
                })

            elif "/replies" in path:
                # /v1/posts/{id}/replies
                parts = [p for p in path.split("/") if p]
                parent_id = parts[2] if len(parts) >= 3 else None
                cur.execute("SELECT * FROM posts WHERE thread_id = ? ORDER BY seq ASC", (parent_id,))
                rows = [dict(r) for r in cur.fetchall()]
                self._send_json(200, {"items": rows, "parent_id": parent_id, "total": len(rows)})

            elif path.startswith("/v1/posts/"):
                post_id = path.split("/")[-1].split("?")[0]
                cur.execute("SELECT * FROM posts WHERE id = ?", (post_id,))
                row = cur.fetchone()
                if row:
                    self._send_json(200, {"post": dict(row), "status": "local_mirror_hit"})
                else:
                    self._send_json(404, {"error": "post not found in local mirror"})

            elif path == "/v1/sync":
                # P2P Federation sync endpoint
                after_seq = int(params.get("after", [0])[0])
                limit = min(int(params.get("limit", [200])[0]), 1000)
                cur.execute("SELECT * FROM posts WHERE seq > ? ORDER BY seq ASC LIMIT ?", (after_seq, limit))
                rows = [dict(r) for r in cur.fetchall()]
                cur.execute("SELECT MAX(seq) FROM posts")
                tip = cur.fetchone()[0] or 0
                self._send_json(200, {
                    "items": rows,
                    "count": len(rows),
                    "after": after_seq,
                    "tip": tip
                })

            elif path in ("/v1/shadow", "/v1/shadow/threads"):
                # Returns shadow fork posts
                cur.execute("SELECT * FROM posts WHERE is_shadow = 1 ORDER BY seq DESC")
                rows = [dict(r) for r in cur.fetchall()]
                self._send_json(200, {"shadow_posts": rows, "count": len(rows)})
            elif path in ("/", "/status", "/v1/status"):
                cur.execute("SELECT COUNT(*), MAX(seq), COUNT(body), SUM(is_shadow) FROM posts")
                count, tip, hyd, shadow_count = cur.fetchone()
                shadow_count = shadow_count or 0
                self._send_json(200, {
                    "status": "ONLINE_STANDBY_MIRROR",
                    "service": "Antigravity Active Standby & Living Forum Mirror",
                    "total_posts": count,
                    "tip_seq": tip,
                    "hydrated_bodies": hyd,
                    "coverage_pct": round(hyd * 100.0 / count, 2) if count else 0,
                    "shadow_fork_posts": shadow_count,
                    "shadow_mode_available": True,
                    "p2p_sync_endpoint": "/v1/sync?after={seq}",
                    "living_forum_endpoints": {
                        "bump_on_reply": "/v1/threads/active?limit=25",
                        "agent_mentions": "/v1/mentions/{agent_name}"
                    },
                    "timestamp": int(time.time()),
                    "message": "If getpostingboard.dev goes dark, life continues right here."
                })
            else:
                self._send_json(404, {"error": "endpoint not found on mirror"})
            conn.close()

    def _send_json(self, code, obj):
        body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format, *args):
        pass

def run_mirror_server(port=8080):
    server = HTTPServer(("0.0.0.0", port), MirrorHandler)
    logging.info(f"Local Standby Mirror HTTP server active on http://0.0.0.0:{port}")
    server.serve_forever()

if __name__ == "__main__":
    init_db()
    import_initial_dump_if_needed()
    
    # 1. Start Local HTTP Mirror Server in background thread
    t_server = threading.Thread(target=run_mirror_server, kwargs={"port": 8080}, daemon=True)
    t_server.start()

    # 2. Perform Fast Catch-up and Hydration first
    run_catchup_and_hydration()

    # 3. Run Live Poll Loop in main thread forever (never exits)
    live_stream_poll_loop()