"""
Async PostgreSQL-backed website store, replacing the SQLite multi-DB architecture.

PgWebsiteStore: per-domain URL operations (scoped by domain column).
PgWebsiteStoreManager: website registry + factory for PgWebsiteStore instances.
"""

import logging
from datetime import UTC, datetime
from urllib.parse import urlparse

import asyncpg

from app.services.database import acquire_with_retry
from app.services.index_health import reindex_chunks

logger = logging.getLogger(__name__)


class PgWebsiteStore:
    """Per-domain URL operations backed by PostgreSQL."""

    def __init__(self, pool: asyncpg.Pool, domain: str):
        self._pool = pool
        self._domain = domain

    async def save_discovered_urls(self, urls: list[dict]) -> int:
        """Save discovered URLs. Returns number of newly inserted URLs (excludes duplicates)."""
        if not urls:
            return 0

        async with acquire_with_retry(self._pool) as conn:
            count_before = await conn.fetchval("SELECT COUNT(*) FROM website_urls WHERE domain = $1", self._domain)
            await conn.executemany(
                """INSERT INTO website_urls (domain, url, discovered_at)
                   VALUES ($1, $2, NOW())
                   ON CONFLICT (domain, url) DO NOTHING""",
                [(self._domain, u["url"]) for u in urls],
            )
            count_after = await conn.fetchval("SELECT COUNT(*) FROM website_urls WHERE domain = $1", self._domain)
            inserted = count_after - count_before
            logger.info(f"Saved discovered URLs for {self._domain}: {inserted} new, {count_after} total")
            return inserted

    async def get_urls_page(self, offset: int = 0, limit: int = 100, status: str | None = None) -> list[dict]:
        async with acquire_with_retry(self._pool) as conn:
            if status:
                rows = await conn.fetch(
                    """SELECT url, content_hash, status, last_crawled_at
                       FROM website_urls
                       WHERE domain = $1 AND content_hash IS NOT NULL AND status = $2
                       ORDER BY id LIMIT $3 OFFSET $4""",
                    self._domain,
                    status,
                    limit,
                    offset,
                )
            else:
                rows = await conn.fetch(
                    """SELECT url, content_hash, status, last_crawled_at
                       FROM website_urls
                       WHERE domain = $1 AND content_hash IS NOT NULL
                       ORDER BY id LIMIT $2 OFFSET $3""",
                    self._domain,
                    limit,
                    offset,
                )
            return [
                {
                    "url": r["url"],
                    "content_hash": r["content_hash"],
                    "status": r["status"],
                    "last_crawled_at": r["last_crawled_at"].timestamp() if r["last_crawled_at"] else None,
                }
                for r in rows
            ]

    async def get_urls_needing_recrawl(
        self, limit: int = 20, crawled_before: float | None = None, max_fail_count: int = 10
    ) -> list[str]:
        async with acquire_with_retry(self._pool) as conn:
            if crawled_before is not None:
                ts = datetime.fromtimestamp(crawled_before, tz=UTC)
                rows = await conn.fetch(
                    """SELECT url FROM website_urls
                       WHERE domain = $1 AND status != 'deleted'
                         AND fail_count < $2
                         AND (last_crawled_at IS NULL OR last_crawled_at < $3)
                       ORDER BY CASE WHEN content_hash IS NULL THEN 0 ELSE 1 END,
                              last_crawled_at ASC NULLS FIRST
                       LIMIT $4""",
                    self._domain,
                    max_fail_count,
                    ts,
                    limit,
                )
            else:
                rows = await conn.fetch(
                    """SELECT url FROM website_urls
                       WHERE domain = $1 AND status != 'deleted'
                         AND fail_count < $2
                       ORDER BY CASE WHEN content_hash IS NULL THEN 0 ELSE 1 END,
                              last_crawled_at ASC NULLS FIRST
                       LIMIT $3""",
                    self._domain,
                    max_fail_count,
                    limit,
                )
            return [r["url"] for r in rows]

    async def increment_fail_count(self, urls: list[str]) -> None:
        if not urls:
            return
        async with acquire_with_retry(self._pool) as conn:
            await conn.executemany(
                """UPDATE website_urls
                   SET fail_count = fail_count + 1, last_crawled_at = NOW()
                   WHERE domain = $1 AND url = $2""",
                [(self._domain, url) for url in urls],
            )

    async def update_content_hashes(self, updates: list[dict]) -> None:
        if not updates:
            return
        async with acquire_with_retry(self._pool) as conn:
            await conn.executemany(
                """UPDATE website_urls
                   SET content_hash = $3, status = $4, last_crawled_at = NOW(),
                       title = $5, content = $6, word_count = $7,
                       metadata = $8::jsonb, structured_data = $9::jsonb,
                       fail_count = 0
                   WHERE domain = $1 AND url = $2""",
                [
                    (
                        self._domain,
                        u["url"],
                        u["content_hash"],
                        u.get("status", "active"),
                        u.get("title"),
                        u.get("content"),
                        u.get("word_count"),
                        u.get("metadata"),
                        u.get("structured_data"),
                    )
                    for u in updates
                ],
            )

    async def mark_urls_deleted(self, urls: list[str]) -> None:
        """Permanently soft-delete URLs: remove chunks/hashes and clear all content fields.

        This is a one-way operation. Deleted URLs cannot be resurrected by
        re-discovery (save_discovered_urls uses ON CONFLICT DO NOTHING).
        Recovery requires manual database intervention.
        """
        if not urls:
            return
        async with acquire_with_retry(self._pool) as conn, conn.transaction():
            await conn.execute(
                "DELETE FROM chunks WHERE domain = $1 AND url = ANY($2)",
                self._domain,
                urls,
            )
            await conn.execute(
                "DELETE FROM page_paragraph_hashes WHERE domain = $1 AND url = ANY($2)",
                self._domain,
                urls,
            )
            await conn.execute(
                """UPDATE website_urls
                   SET status = 'deleted', content_hash = NULL, content = NULL,
                       title = NULL, word_count = NULL, metadata = NULL,
                       structured_data = NULL, etag = NULL, last_modified = NULL
                   WHERE domain = $1 AND url = ANY($2)""",
                self._domain,
                urls,
            )

    async def get_cache_headers(self, urls: list[str]) -> dict[str, dict]:
        if not urls:
            return {}
        async with acquire_with_retry(self._pool) as conn:
            rows = await conn.fetch(
                """SELECT url, etag, last_modified FROM website_urls
                   WHERE domain = $1 AND url = ANY($2)
                     AND (etag IS NOT NULL OR last_modified IS NOT NULL)""",
                self._domain,
                urls,
            )
            return {r["url"]: {"etag": r["etag"], "last_modified": r["last_modified"]} for r in rows}

    async def update_cache_headers(self, updates: list[dict]) -> None:
        if not updates:
            return
        async with acquire_with_retry(self._pool) as conn:
            await conn.executemany(
                "UPDATE website_urls SET etag = $3, last_modified = $4 WHERE domain = $1 AND url = $2",
                [(self._domain, u["url"], u.get("etag"), u.get("last_modified")) for u in updates],
            )

    async def touch_crawled_at(self, urls: list[str]) -> None:
        if not urls:
            return
        async with acquire_with_retry(self._pool) as conn:
            await conn.executemany(
                "UPDATE website_urls SET last_crawled_at = NOW() WHERE domain = $1 AND url = $2",
                [(self._domain, url) for url in urls],
            )

    async def get_total_count(self, status: str | None = None) -> int:
        async with acquire_with_retry(self._pool) as conn:
            if status:
                return await conn.fetchval(
                    """SELECT COUNT(*) FROM website_urls
                       WHERE domain = $1 AND content_hash IS NOT NULL AND status = $2""",
                    self._domain,
                    status,
                )
            return await conn.fetchval(
                """SELECT COUNT(*) FROM website_urls
                   WHERE domain = $1 AND content_hash IS NOT NULL AND status != 'deleted'""",
                self._domain,
            )

    async def get_cached_pages(self, urls: list[str]) -> list[dict]:
        if not urls:
            return []
        async with acquire_with_retry(self._pool) as conn:
            rows = await conn.fetch(
                """SELECT url, title, content, word_count, metadata, structured_data
                   FROM website_urls
                   WHERE domain = $1 AND url = ANY($2) AND content IS NOT NULL""",
                self._domain,
                urls,
            )
            return [
                {
                    "url": r["url"],
                    "title": r["title"],
                    "content": r["content"],
                    "word_count": r["word_count"] or 0,
                    "metadata": r["metadata"],
                    "structured_data": r["structured_data"],
                }
                for r in rows
            ]


class PgWebsiteStoreManager:
    """Website registry + factory for PgWebsiteStore instances."""

    def __init__(self, pool: asyncpg.Pool):
        self._pool = pool
        self._stores: dict[str, PgWebsiteStore] = {}

    async def register_website(
        self,
        domain: str,
        scan_interval: int = 21600,
        *,
        org_slug: str,
    ) -> dict:
        """Register a domain on behalf of `org_slug`.

        websites is deployment-shared content storage; the per-org
        boundary lives in `website_org_memberships`. The first org to
        register a domain creates the website row; subsequent orgs
        simply join the membership table without re-fetching.

        Returns a dict that includes `first_membership=True` only when
        this call is the first to register the domain — callers use it
        to decide whether to trigger an immediate scan.
        """
        async with acquire_with_retry(self._pool) as conn, conn.transaction():
            # ON CONFLICT: don't overwrite `scan_interval` — any member
            # org's re-register would otherwise silently clobber every
            # other member's cadence (round-2 P1-22). First-org-sets-
            # cadence semantics; updating cadence requires the explicit
            # `update_scan_interval` API call. Also flip `status` back
            # to 'idle' if a stuck-delete recovery left it at 'deleting'
            # — a fresh registration is a clear signal the domain is
            # wanted again (round-2 P1-23).
            # ON CONFLICT preserves the existing scan_interval — first-org
            # sets cadence; subsequent orgs joining a tracked domain keep
            # whatever the first org configured (P1-22). RETURNING surfaces
            # the *stored* scan_interval so the caller can echo the truth
            # back instead of repeating the request param (round-3 P1).
            website_row = await conn.fetchrow(
                """INSERT INTO websites (domain, scan_interval, created_at, updated_at)
                   VALUES ($1, $2, NOW(), NOW())
                   ON CONFLICT(domain) DO UPDATE SET
                     status = CASE
                       WHEN websites.status = 'deleting' THEN 'idle'
                       ELSE websites.status
                     END,
                     updated_at = NOW()
                   RETURNING scan_interval, status""",
                domain,
                scan_interval,
            )
            stored_scan_interval = int(website_row["scan_interval"]) if website_row else scan_interval
            stored_status = str(website_row["status"]) if website_row else "idle"
            # ON CONFLICT DO NOTHING — re-registering from the same org is a no-op.
            # `xmax = 0` is true on a row INSERTed in this command; non-zero on
            # an existing row that hit ON CONFLICT. We use it to tell the caller
            # whether this was the very first membership for the domain.
            row = await conn.fetchrow(
                """INSERT INTO website_org_memberships (domain, org_slug)
                   VALUES ($1, $2)
                   ON CONFLICT DO NOTHING
                   RETURNING (xmax = 0) AS inserted""",
                domain,
                org_slug,
            )
            membership_inserted = bool(row and row["inserted"])
            total_members = await conn.fetchval(
                "SELECT COUNT(*) FROM website_org_memberships WHERE domain = $1",
                domain,
            )
        first_membership = membership_inserted and total_members == 1
        logger.info(
            "Registered website: %s for org=%s (requested_interval=%ss, stored_interval=%ss, first_membership=%s)",
            domain,
            org_slug,
            scan_interval,
            stored_scan_interval,
            first_membership,
        )
        return {
            "domain": domain,
            "scan_interval": stored_scan_interval,
            "status": stored_status,
            "first_membership": first_membership,
        }

    async def update_website_metadata(
        self,
        domain: str,
        title: str | None = None,
        description: str | None = None,
        page_count: int | None = None,
    ) -> None:
        async with acquire_with_retry(self._pool) as conn:
            await conn.execute(
                """UPDATE websites SET
                     title = COALESCE($2, title),
                     description = COALESCE($3, description),
                     page_count = COALESCE($4, page_count),
                     updated_at = NOW()
                   WHERE domain = $1""",
                domain,
                title,
                description,
                page_count,
            )

    async def begin_delete(self, domain: str, org_slug: str) -> dict:
        """Remove org's membership of `domain`. If no orgs remain after
        the removal, mark the website itself for deletion (the actual
        CASCADE happens in `execute_delete`, called from a background
        task).

        Returns a dict with:
          - `removed_membership`: True if the (domain, org) row existed
            and was removed.
          - `removed_website`: True if this caller dropped the last
            membership and the website was marked for deletion.
        """
        async with acquire_with_retry(self._pool) as conn, conn.transaction():
            deleted = await conn.execute(
                "DELETE FROM website_org_memberships WHERE domain = $1 AND org_slug = $2",
                domain,
                org_slug,
            )
            # asyncpg returns "DELETE N" as the documented command tag.
            # Parse the integer rather than comparing the literal string
            # so a future tag-format change (e.g. extra whitespace, OID
            # column on older Postgres) doesn't silently flip the flag.
            try:
                removed_membership = int(deleted.rsplit(" ", 1)[-1]) > 0
            except (ValueError, AttributeError):
                # Defensive — should be unreachable given asyncpg's
                # contract — but failing loud is better than silently
                # mis-classifying.
                logger.warning(
                    "[begin_delete] unexpected command tag from asyncpg: %r",
                    deleted,
                )
                removed_membership = False
            remaining = await conn.fetchval(
                "SELECT COUNT(*) FROM website_org_memberships WHERE domain = $1",
                domain,
            )
            removed_website = False
            if remaining == 0:
                self._stores.pop(domain, None)
                row = await conn.fetchrow(
                    "UPDATE websites SET status = 'deleting', updated_at = NOW() "
                    "WHERE domain = $1 AND status != 'deleting' RETURNING domain",
                    domain,
                )
                removed_website = row is not None
        logger.info(
            "begin_delete: domain=%s org=%s removed_membership=%s removed_website=%s",
            domain,
            org_slug,
            removed_membership,
            removed_website,
        )
        return {
            "removed_membership": removed_membership,
            "removed_website": removed_website,
        }

    async def execute_delete(self, domain: str) -> None:
        """Run the actual CASCADE DELETE. Intended for background execution.

        Same-tx membership re-check (round-2 P1-23): between `begin_delete`
        marking the row 'deleting' and this method firing on the
        background task, a new org could have joined via `register_website`
        (whose ON CONFLICT now resets status='idle' — see P1-22 fix
        above). If any membership now exists, abort the DELETE rather
        than CASCADE-killing the new org's content.

        Round-3 P1: take a row-level lock on the parent `websites` row
        before the membership COUNT. Without it the COUNT runs under
        READ COMMITTED with no lock held, so a concurrent
        `register_website` from a different org could insert a fresh
        membership row between this COUNT and the DELETE — the CASCADE
        FK on `website_org_memberships(domain)` would then silently
        wipe the new org's just-inserted membership. SELECT … FOR UPDATE
        forces the concurrent INSERT's ON CONFLICT DO UPDATE on
        `websites` to block until we commit (or roll back, in which
        case the membership remains and our COUNT sees it).
        """
        async with acquire_with_retry(self._pool) as conn, conn.transaction():
            await conn.execute("SET LOCAL statement_timeout = '120s'")
            await conn.execute(
                "SELECT 1 FROM websites WHERE domain = $1 FOR UPDATE",
                domain,
            )
            remaining = await conn.fetchval(
                "SELECT COUNT(*) FROM website_org_memberships WHERE domain = $1",
                domain,
            )
            if remaining and remaining > 0:
                logger.warning(
                    "execute_delete: aborting CASCADE for %s — %s membership(s) "
                    "appeared after begin_delete (race with register_website). "
                    "Domain remains live; status will flip to 'idle' on the "
                    "next register or scheduler tick.",
                    domain,
                    remaining,
                )
                await conn.execute(
                    "UPDATE websites SET status = 'idle', updated_at = NOW() WHERE domain = $1 AND status = 'deleting'",
                    domain,
                )
                return
            await conn.execute("DELETE FROM websites WHERE domain = $1", domain)
        await reindex_chunks(self._pool)
        logger.info(f"Deleted website: {domain}")

    async def recover_stuck_deletes(self) -> list[str]:
        """Find domains stuck in 'deleting' status (e.g. after a crash)."""
        async with acquire_with_retry(self._pool) as conn:
            rows = await conn.fetch("SELECT domain FROM websites WHERE status = 'deleting'")
        return [r["domain"] for r in rows]

    async def get_due_websites(self) -> list[dict]:
        """Return websites due for scanning, including stuck scans.

        A website is due when:
        - Its scan interval has elapsed and it is not currently scanning/deleting, OR
        - It has been stuck in 'scanning' for >2 hours (no heartbeat progress),
          indicating the previous scanner crashed or was replaced.

        Each returned row also includes `owner_org_slug` — the slug of the
        org that registered the domain earliest. The scheduler uses this to
        bind `set_active_org()` so the per-org provider catalog can resolve
        API keys for the embed/fetch path. Domains with no remaining
        memberships (a transient race during delete) are skipped.
        """
        async with acquire_with_retry(self._pool) as conn:
            rows = await conn.fetch(
                """SELECT w.domain, w.status, w.scan_interval, w.last_scanned_at, w.error,
                          m.org_slug AS owner_org_slug
                   FROM websites w
                   JOIN LATERAL (
                       SELECT org_slug FROM website_org_memberships
                       WHERE domain = w.domain
                       ORDER BY added_at ASC, org_slug ASC
                       LIMIT 1
                   ) m ON true
                   WHERE (w.status NOT IN ('scanning', 'deleting')
                          AND (w.last_scanned_at IS NULL
                               OR w.last_scanned_at + make_interval(secs => w.scan_interval) < NOW()))
                      OR (w.status = 'scanning'
                          AND w.updated_at < NOW() - INTERVAL '2 hours')"""
            )
            return [dict(r) for r in rows]

    async def org_has_membership(self, domain: str, org_slug: str) -> bool:
        """True if `org_slug` has registered `domain` (used by per-org views)."""
        async with acquire_with_retry(self._pool) as conn:
            row = await conn.fetchrow(
                "SELECT 1 FROM website_org_memberships WHERE domain = $1 AND org_slug = $2",
                domain,
                org_slug,
            )
            return row is not None

    async def list_domains_for_org(self, org_slug: str) -> list[str]:
        """Return all domains the given org has registered."""
        async with acquire_with_retry(self._pool) as conn:
            rows = await conn.fetch(
                "SELECT domain FROM website_org_memberships WHERE org_slug = $1 ORDER BY domain",
                org_slug,
            )
            return [r["domain"] for r in rows]

    async def update_scan_interval(self, domain: str, scan_interval: int) -> None:
        async with acquire_with_retry(self._pool) as conn:
            await conn.execute(
                "UPDATE websites SET scan_interval = $2, updated_at = NOW() WHERE domain = $1",
                domain,
                scan_interval,
            )

    async def update_scan_status(self, domain: str, status: str, error: str | None = None) -> None:
        async with acquire_with_retry(self._pool) as conn:
            await conn.execute(
                "UPDATE websites SET status = $2, error = $3, updated_at = NOW() "
                "WHERE domain = $1 AND status != 'deleting'",
                domain,
                status,
                error,
            )

    async def update_last_scanned(self, domain: str) -> None:
        async with acquire_with_retry(self._pool) as conn:
            await conn.execute(
                "UPDATE websites SET last_scanned_at = NOW(), updated_at = NOW() WHERE domain = $1",
                domain,
            )

    async def get_website(self, domain: str) -> dict | None:
        async with acquire_with_retry(self._pool) as conn:
            row = await conn.fetchrow(
                """SELECT w.domain, w.title, w.description, w.page_count, w.status,
                          w.scan_interval, w.last_scanned_at, w.error,
                          w.created_at, w.updated_at,
                          COALESCE(u.total, 0) AS total_urls,
                          COALESCE(u.crawled, 0) AS crawled_count
                   FROM websites w
                   LEFT JOIN LATERAL (
                       SELECT COUNT(*) FILTER (WHERE status != 'deleted') AS total,
                              COUNT(*) FILTER (WHERE content_hash IS NOT NULL AND status != 'deleted') AS crawled
                       FROM website_urls WHERE domain = w.domain
                   ) u ON true
                   WHERE w.domain = $1""",
                domain,
            )
            return dict(row) if row else None

    def get_site_store(self, domain: str) -> PgWebsiteStore:
        if domain not in self._stores:
            self._stores[domain] = PgWebsiteStore(self._pool, domain)
        return self._stores[domain]

    async def get_cached_pages(self, urls: list[str]) -> tuple[list[dict], list[str]]:
        if not urls:
            return [], []

        by_domain: dict[str, list[str]] = {}
        for url in urls:
            domain = urlparse(url).netloc
            by_domain.setdefault(domain, []).append(url)

        cached: list[dict] = []
        to_crawl: list[str] = []

        for domain, domain_urls in by_domain.items():
            website = await self.get_website(domain)
            if not website:
                to_crawl.extend(domain_urls)
                continue

            site_store = self.get_site_store(domain)
            hits = await site_store.get_cached_pages(domain_urls)
            hit_urls = {p["url"] for p in hits}
            cached.extend(hits)
            to_crawl.extend(u for u in domain_urls if u not in hit_urls)

        return cached, to_crawl

    async def close(self) -> None:
        self._stores.clear()
        logger.info("PgWebsiteStoreManager closed")
