import json
import uuid
from urllib.parse import parse_qs, urlparse
from datetime import datetime, timedelta, timezone
import httpx
import pytest
from sqlalchemy import text
from crud import social_crud
from services import social_service as service
from services.external import gemini_text, threads
from services.external.social import SocialOutcomeUnknown, weighted_length


async def seed(client, auth_headers, db_engine):
    h = await auth_headers("social-owner")
    pid = uuid.UUID(
        (
            await client.post(
                "/v1/place", headers=h, json={"name": "소식숙소", "category": 1}
            )
        ).json()["place"]["place_id"]
    )
    async with db_engine.begin() as c:
        uid = (
            await c.execute(
                text("SELECT owner_user_id FROM places WHERE place_id=:p"), {"p": pid}
            )
        ).scalar_one()
        version = uuid.uuid4()
        await c.execute(
            text(
                "INSERT INTO sites(site_id,place_id,domain,status,current_version_id) VALUES (:id,:p,'social-stay',3,:v)"
            ),
            {"id": uuid.uuid4(), "p": pid, "v": version},
        )
        await c.execute(
            text(
                "INSERT INTO place_facts(fact_id,place_id,key,value,source_type,status) VALUES (:id,:p,'check_in_time','15:00',1,3)"
            ),
            {"id": uuid.uuid4(), "p": pid},
        )
    return h, pid, uid, version


async def pending(db_engine, pid, uid, version, expired=False):
    post_id, account_id = uuid.uuid4(), uuid.uuid4()
    token = "a" * 43
    async with db_engine.begin() as c:
        await c.execute(
            text(
                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status) VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','linked')"
            ),
            {"a": account_id, "u": uid},
        )
        await c.execute(
            text("""INSERT INTO place_social_posts(post_id,place_id,user_id,site_version_id,account_id,provider,body,link_url,status,approval_token_sha,approval_expires_at)
            VALUES (:id,:p,:u,:v,:a,2,'체크인은 15:00입니다.',:url,'PENDING_APPROVAL',:sha,:expires)"""),
            {
                "id": post_id,
                "p": pid,
                "u": uid,
                "v": version,
                "a": account_id,
                "sha": service.sha(token),
                "url": service.site_payload.publish_origin() + "/s/social-stay",
                "expires": datetime.now(timezone.utc)
                + timedelta(hours=-1 if expired else 1),
            },
        )
    return post_id, token


async def test_draft_dedup_owner_scope(client, auth_headers, db_engine):
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    first = await client.post(f"/v1/social/place/{pid}/draft", headers=h, json={})
    assert first.status_code == 200, first.text
    second = await client.post(f"/v1/social/place/{pid}/draft", headers=h, json={})
    assert second.json()["post_id"] == first.json()["post_id"]
    assert (
        first.json()["link_url"]
        == service.site_payload.publish_origin() + "/s/social-stay"
    )
    other = await auth_headers("social-other")
    response = await client.get(f"/v1/social/place/{pid}", headers=other)
    assert (
        response.status_code == 404 and response.json()["detail"] == "PLACE_NOT_FOUND"
    )
    async with db_engine.begin() as c:
        assert (
            await c.execute(text("SELECT count(*) FROM jobs WHERE job_type=8"))
        ).scalar_one() == 1


async def test_requires_fixed_domain(client, auth_headers, db_engine):
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    async with db_engine.begin() as c:
        await c.execute(
            text("UPDATE sites SET domain=NULL WHERE place_id=:p"), {"p": pid}
        )
    assert (
        await client.post(f"/v1/social/place/{pid}/draft", headers=h, json={})
    ).status_code == 409


async def test_prefetch_read_only_one_time_cas(client, auth_headers, db_engine):
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    post_id, token = await pending(db_engine, pid, uid, v)
    for _ in range(2):
        res = await client.get(f"/v1/social/approval/{post_id}?t={token}")
        assert res.status_code == 200 and res.json()["status"] == "PENDING_APPROVAL"
        assert res.headers["cache-control"] == "no-store"
        assert "approval_token_sha" not in res.text
    first = await client.post(
        f"/v1/social/approval/{post_id}/decision", json={"t": token, "approve": True}
    )
    second = await client.post(
        f"/v1/social/approval/{post_id}/decision", json={"t": token, "approve": True}
    )
    assert first.json()["applied"] is True, first.text
    assert second.json()["applied"] is False
    async with db_engine.begin() as c:
        assert (
            await c.execute(text("SELECT count(*) FROM jobs WHERE job_type=9"))
        ).scalar_one() == 1


async def test_expired_and_stale_sweep(client, auth_headers, db_engine):
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    post_id, token = await pending(db_engine, pid, uid, v, expired=True)
    res = await client.post(
        f"/v1/social/approval/{post_id}/decision", json={"t": token, "approve": False}
    )
    assert res.json()["applied"] is False
    await social_crud.sweep()
    async with db_engine.begin() as c:
        row = (
            await c.execute(
                text("SELECT status,body FROM place_social_posts WHERE post_id=:p"),
                {"p": post_id},
            )
        ).first()
        assert row.status == "EXPIRED" and row.body
        await c.execute(
            text(
                "UPDATE place_social_posts SET status='POSTING',updated_at=now()-interval '11 minutes' WHERE post_id=:p"
            ),
            {"p": post_id},
        )
    await social_crud.sweep()
    async with db_engine.begin() as c:
        assert (
            await c.execute(
                text("SELECT status FROM place_social_posts WHERE post_id=:p"),
                {"p": post_id},
            )
        ).scalar_one() == "UNKNOWN"


async def test_no_facts_no_paid_call(monkeypatch):
    async def forbidden(*args, **kwargs):
        raise AssertionError("paid call")

    monkeypatch.setattr(gemini_text, "call", forbidden)
    with pytest.raises(gemini_text.GeminiInvalidOutput, match="NO_GROUNDED_FACTS"):
        await gemini_text.generate_social_post("숙소", [], "https://example.com/s/stay")


async def test_long_draft_regenerates(monkeypatch):
    monkeypatch.setattr(gemini_text, "is_configured", lambda: True)
    bodies = iter(["가" * 501, "체크인은 15:00입니다."])

    async def call(*args, **kwargs):
        return {
            "candidates": [
                {
                    "content": {
                        "parts": [
                            {
                                "text": json.dumps(
                                    {
                                        "body": next(bodies),
                                        "fact_keys": ["check_in_time"],
                                    }
                                )
                            }
                        ]
                    }
                }
            ]
        }

    monkeypatch.setattr(gemini_text, "call", call)
    result = await gemini_text.generate_social_post(
        "숙소",
        [gemini_text.FactInput(key="check_in_time", label="체크인", value="15:00")],
        "https://example.com/s/stay",
    )
    assert result == "체크인은 15:00입니다.\n\nhttps://example.com/s/stay"


def test_lengths():
    assert weighted_length("한글", 1) == 4
    assert weighted_length("https://example.com/" + "a" * 200, 1) == 23
    assert weighted_length("한글", 2) == 2
    assert weighted_length("https://example.com/" + "a" * 200, 2) == 220


async def test_threads_timeout_no_retry():
    calls = []

    def handler(req):
        calls.append(req)
        if req.url.path.endswith("/me/threads"):
            assert (
                b"media_type=TEXT" in req.content
                and b"auto_publish_text=false" in req.content
            )
            assert b"image_url" not in req.content
            return httpx.Response(200, json={"id": "11"})
        raise httpx.ReadTimeout("lost", request=req)

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        with pytest.raises(SocialOutcomeUnknown):
            await threads.publish("소개", "secret", client=client)
    assert len(calls) == 2


async def test_published_permalink_failure():
    def handler(req):
        if req.url.path.endswith("/me/threads"):
            return httpx.Response(200, json={"id": "11"})
        if req.url.path.endswith("/me/threads_publish"):
            return httpx.Response(200, json={"id": "22"})
        return httpx.Response(500, json={"error": {}})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        result = await threads.publish("소개", "secret", client=client)
    assert result == {"id": "22", "permalink": None}


async def test_screen_approval_without_contract_never_queues_post(
    client, auth_headers, db_engine
):
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    result = await client.post(f"/v1/social/place/{pid}/draft", headers=h, json={})
    post_id = uuid.UUID(result.json()["post_id"])
    async with db_engine.begin() as c:
        await c.execute(
            text(
                "UPDATE place_social_posts SET body='작성된 원고',status='DRAFT' WHERE post_id=:p"
            ),
            {"p": post_id},
        )
    response = await client.post(
        f"/v1/social/posts/{post_id}/request-approval", headers=h, json={}
    )
    assert response.status_code == 200, response.text
    assert response.json()["post"]["account_bound"] is False
    approved = await client.post(
        f"/v1/social/posts/{post_id}/decision", headers=h, json={"approve": True}
    )
    assert approved.json()["applied"] is True
    async with db_engine.begin() as c:
        assert (
            await c.execute(text("SELECT count(*) FROM jobs WHERE job_type=9"))
        ).scalar_one() == 0


async def test_token_cipher_and_oauth_browser_binding(monkeypatch):
    from cryptography.fernet import Fernet
    from services import social_account_service as accounts
    from services.external.social import SocialError

    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", "")
    assert not accounts.configured()
    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", Fernet.generate_key().decode())
    monkeypatch.setenv("THREADS_APP_ID", "test")
    monkeypatch.setenv("THREADS_APP_SECRET", "test")
    monkeypatch.setenv("THREADS_REDIRECT_URI", "https://example.com/callback")
    encrypted = accounts.encrypt("owner-token")
    assert (
        "owner-token" not in encrypted and accounts.decrypt(encrypted) == "owner-token"
    )
    url, browser = accounts.begin(uuid.uuid4(), 2)
    from urllib.parse import parse_qs, urlparse

    state = parse_qs(urlparse(url).query)["state"][0]
    with pytest.raises(SocialError, match="INVALID_OAUTH_STATE"):
        await accounts.finish(state, "wrong-browser", "unused-code")


@pytest.mark.parametrize("unknown", [False, True])
async def test_post_claim_prevents_second_external_write(
    client, auth_headers, db_engine, monkeypatch, unknown
):
    from cryptography.fernet import Fernet
    from services import social_account_service as accounts

    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    post_id, _ = await pending(db_engine, pid, uid, v)
    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", Fernet.generate_key().decode())
    monkeypatch.setattr(service, "posting_enabled", lambda: True)
    async with db_engine.begin() as c:
        await c.execute(
            text("UPDATE place_social_posts SET status='APPROVED' WHERE post_id=:p"),
            {"p": post_id},
        )
        await c.execute(
            text(
                "UPDATE owner_social_accounts SET access_token=:t,access_expires_at=now()+interval '1 day' WHERE user_id=:u"
            ),
            {"t": accounts.encrypt("secret"), "u": uid},
        )
    calls = []

    class Adapter:
        @staticmethod
        def weighted_limit():
            return 500

        @staticmethod
        async def me(*args, **kwargs):
            return {"id": "22"}

        @staticmethod
        async def publish(*args, **kwargs):
            calls.append(1)
            if unknown:
                raise SocialOutcomeUnknown("POST_RESULT_UNKNOWN")
            return {"id": "33", "permalink": "https://www.threads.com/@host/post/abc"}

    monkeypatch.setattr(service, "adapter", lambda provider: Adapter)
    job = {"payload": {"post_id": str(post_id)}}
    if unknown:
        with pytest.raises(SocialOutcomeUnknown):
            await service.run_post(job)
    else:
        await service.run_post(job)
    assert (await service.run_post(job)) == {"skipped": True}
    assert calls == [1]
    async with db_engine.begin() as c:
        assert (
            await c.execute(
                text("SELECT status FROM place_social_posts WHERE post_id=:p"),
                {"p": post_id},
            )
        ).scalar_one() == ("UNKNOWN" if unknown else "POSTED")


async def test_test_post_requires_connected_account(client, auth_headers):
    h = await auth_headers("social-test-owner")
    res = await client.post("/v1/social/test-post", headers=h, json={"text": "연동 확인"})
    assert (
        res.status_code == 409 and res.json()["detail"] == "ACCOUNT_CONNECTION_REQUIRED"
    )


async def test_test_post_requires_reauth_when_expired(client, auth_headers, db_engine, monkeypatch):
    from cryptography.fernet import Fernet

    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", Fernet.generate_key().decode())
    h = await auth_headers("social-reauth-owner")
    async with db_engine.begin() as c:
        uid = (
            await c.execute(
                text("SELECT user_id FROM users WHERE id=:i"), {"i": "social-reauth-owner"}
            )
        ).scalar_one()
        await c.execute(
            text(
                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status) "
                "VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','needs_reauth')"
            ),
            {"a": uuid.uuid4(), "u": uid},
        )
    res = await client.post("/v1/social/test-post", headers=h, json={"text": "연동 확인"})
    assert (
        res.status_code == 409 and res.json()["detail"] == "ACCOUNT_NEEDS_REAUTH"
    )


async def test_test_post_publishes_immediately_bypassing_approval(
    client, auth_headers, db_engine, monkeypatch
):
    """연동 확인용 게시는 posting_enabled 게이트·승인 절차 없이 바로 나간다."""
    from cryptography.fernet import Fernet
    from services import social_account_service as accounts

    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", Fernet.generate_key().decode())
    monkeypatch.setattr(service, "posting_enabled", lambda: False)
    h = await auth_headers("social-verify-owner")
    async with db_engine.begin() as c:
        uid = (
            await c.execute(
                text("SELECT user_id FROM users WHERE id=:i"), {"i": "social-verify-owner"}
            )
        ).scalar_one()
        await c.execute(
            text(
                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status,access_token,access_expires_at) "
                "VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','linked',:t,now()+interval '1 day')"
            ),
            {"a": uuid.uuid4(), "u": uid, "t": accounts.encrypt("secret-token")},
        )

    calls = []

    class Adapter:
        @staticmethod
        async def publish(text_, token, *, client):
            calls.append((text_, token))
            return {"id": "99", "permalink": "https://www.threads.com/@host/post/99"}

    monkeypatch.setattr(service, "adapter", lambda provider: Adapter)
    res = await client.post(
        "/v1/social/test-post", headers=h, json={"text": "연동 확인용 테스트"}
    )
    assert res.status_code == 200, res.text
    assert res.json() == {"id": "99", "permalink": "https://www.threads.com/@host/post/99"}
    assert calls == [("연동 확인용 테스트", "secret-token")]


async def test_oauth_roundtrip_saves_encrypted_account(db_engine, monkeypatch):
    """검증: 인가 코드를 받아 계정을 연결하고, 해제까지 한 바퀴 돈다.

    ★ 왜 가짜 서버로 미리 도는가 — 실제 연결은 Meta 앱 등록(리디렉션 URI·권한·테스터 추가)이
      끝나야 시험할 수 있다. 그때 실패하면 우리 코드가 틀린 건지 앱 설정이 틀린 건지 구별이
      안 된다. 우리 쪽 왕복(코드 교환 → 장기토큰 → 검증 → 저장 → 해제)은 여기서 먼저 못 박는다.
    ★ 이 검사가 지키는 것 셋:
      1. 저장된 것은 **암호문**이다 — 토큰 원문이 DB 에 남으면 안 된다.
      2. 장기 토큰 교환과 `debug_token` 검증을 건너뛰지 않는다(권한이 모자란 연결을 만들지 않는다).
      3. 해제하면 토큰이 **지워진다** — 행만 남기고 토큰을 두면 지운 줄 알고 계속 쓰게 된다.
    """
    import json as _json
    import uuid as _uuid

    from cryptography.fernet import Fernet
    from sqlalchemy import text as _text

    from services import social_account_service as accounts

    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", Fernet.generate_key().decode())
    monkeypatch.setenv("THREADS_APP_ID", "app-1")
    monkeypatch.setenv("THREADS_APP_SECRET", "secret-1")
    monkeypatch.setenv("THREADS_REDIRECT_URI", "https://example.com/v1/social/oauth/callback")

    long_lived = "long-lived-token"

    def handler(request: httpx.Request) -> httpx.Response:
        path = request.url.path
        if path.endswith("/oauth/access_token"):
            return httpx.Response(200, json={"access_token": "short-token", "user_id": "1"})
        if path.endswith("/access_token"):
            # 장기 토큰 교환. 60일짜리를 준다 — 하루 미만이면 코드가 거절해야 한다.
            return httpx.Response(200, json={"access_token": long_lived, "expires_in": 5184000})
        if path.endswith("/debug_token"):
            return httpx.Response(200, json={"data": {
                "is_valid": True, "app_id": "app-1",
                "scopes": ["threads_basic", "threads_content_publish"]}})
        if path.endswith("/me"):
            return httpx.Response(200, json={
                "id": "th-1", "username": "mumum", "threads_profile_picture_url": ""})
        return httpx.Response(404, json={"error": {"message": "unexpected " + path}})

    transport = httpx.MockTransport(handler)
    original = httpx.AsyncClient

    def fake_client(*args, **kwargs):
        kwargs["transport"] = transport
        return original(*args, **kwargs)

    monkeypatch.setattr(httpx, "AsyncClient", fake_client)

    user_id = _uuid.uuid4()
    url, browser = accounts.begin(user_id, 2)
    state = parse_qs(urlparse(url).query)["state"][0]
    await accounts.finish(state, browser, "auth-code")

    async with db_engine.begin() as conn:
        row = (await conn.execute(_text(
            "SELECT handle, access_token, status, scopes FROM owner_social_accounts "
            "WHERE user_id = :u AND deleted = false"), {"u": user_id})).first()
    assert row is not None, "연결이 저장되지 않았다"
    assert row.handle == "mumum"
    assert row.status == "linked"
    # ★ 원문이 DB 에 있으면 안 된다.
    assert long_lived not in row.access_token
    assert accounts.decrypt(row.access_token) == long_lived
    scopes = row.scopes if isinstance(row.scopes, list) else _json.loads(row.scopes)
    assert set(scopes) == {"threads_basic", "threads_content_publish"}

    await accounts.disconnect(user_id, 2)
    async with db_engine.begin() as conn:
        after = (await conn.execute(_text(
            "SELECT status, access_token FROM owner_social_accounts WHERE user_id = :u"),
            {"u": user_id})).first()
    assert after.status == "revoked" and after.access_token is None, "해제해도 토큰이 남아 있다"