"""Live-DB test: claim_next_source is race-safe against concurrent callers.

Skips when no DB URL is configured. This is the integration-level guard
for the SELECT ... FOR UPDATE SKIP LOCKED dance — unit-testing the SQL
without Postgres would prove nothing.
"""

from __future__ import annotations

import os
import threading
import uuid

import psycopg
import pytest


def _db_reachable() -> bool:
    url = os.environ.get("SECOND_BRAIN_DATABASE_URL") or os.environ.get(
        "HERBYLAB_DATABASE_URL"
    )
    if not url:
        return False
    raw = url.replace("postgresql+psycopg://", "postgresql://", 1)
    try:
        with psycopg.connect(raw, connect_timeout=2):
            return True
    except Exception:
        return False


pytestmark = pytest.mark.skipif(
    not _db_reachable(),
    reason="claim race test needs SECOND_BRAIN_DATABASE_URL pointed at a real petalbrain",
)


def test_two_workers_never_grab_the_same_row():
    from second_brain.claim import claim_next_source, release_claim
    from second_brain.database import Database, reset_database_singleton
    from second_brain.models import Source, SourceStatus, SourceType, utcnow

    reset_database_singleton()
    db = Database()

    # Seed a single TRANSCRIBED-stage candidate.
    unique = uuid.uuid4().hex[:8]
    url = f"https://example.test/claim-race/{unique}"
    with db.session() as sess:
        src = Source(
            url=url,
            title=f"claim-race-{unique}",
            domain="development",
            source_type=SourceType.VIDEO,
            status=SourceStatus.PULLED,
            ingested_at=utcnow(),
        )
        sess.add(src)
        sess.flush()
        source_id = src.id

    results: list[int | None] = []
    lock = threading.Lock()

    def worker(name: str):
        with db.session() as sess:
            claimed = claim_next_source(
                sess,
                worker_id=name,
                statuses=[SourceStatus.PENDING, SourceStatus.PULLED],
                source_type=SourceType.VIDEO,
            )
        with lock:
            results.append(claimed)

    t1 = threading.Thread(target=worker, args=("worker-A",))
    t2 = threading.Thread(target=worker, args=("worker-B",))
    t1.start()
    t2.start()
    t1.join()
    t2.join()

    try:
        # Exactly one worker should have grabbed the row; the other gets None.
        # (SKIP LOCKED guarantees this even if both transactions overlap.)
        non_none = [r for r in results if r is not None]
        assert non_none == [source_id], (
            f"expected exactly one claimant of {source_id}, got results={results}"
        )

        # Confirm claim columns landed.
        with db.session() as sess:
            row = sess.get(Source, source_id)
            assert row.claimed_by in {"worker-A", "worker-B"}
            assert row.claimed_at is not None

        # Releasing the claim should null both columns.
        with db.session() as sess:
            release_claim(sess, source_id)
        with db.session() as sess:
            row = sess.get(Source, source_id)
            assert row.claimed_by is None
            assert row.claimed_at is None
    finally:
        with db.session() as sess:
            sess.query(Source).filter(Source.id == source_id).delete()
        reset_database_singleton()