"""Integration tests for SqlAlchemyDiarizationJobRepository.

Tests the complete diarization job lifecycle including:
- Job creation and retrieval
- Status transitions (QUEUED -> RUNNING -> COMPLETED/FAILED)
- Crash recovery (mark_running_as_failed)
- Job pruning with TTL
- Streaming diarization turns persistence
"""

from __future__ import annotations

from collections.abc import Sequence
from datetime import datetime, timedelta
from typing import TYPE_CHECKING
from uuid import uuid4

import pytest

from noteflow.domain.entities import Meeting
from noteflow.infrastructure.persistence.repositories import (
    DiarizationJob,
    SqlAlchemyDiarizationJobRepository,
    SqlAlchemyMeetingRepository,
    StreamingTurn,
)
from noteflow.infrastructure.persistence.repositories.diarization_job_repo import (
    JOB_STATUS_CANCELLED,
    JOB_STATUS_COMPLETED,
    JOB_STATUS_FAILED,
    JOB_STATUS_QUEUED,
    JOB_STATUS_RUNNING,
)
from tests.conftest import approx_float

if TYPE_CHECKING:
    from sqlalchemy.ext.asyncio import AsyncSession

# Test constant for segments updated in completion tests
EXPECTED_SEGMENTS_UPDATED = 42


@pytest.mark.integration
class TestDiarizationJobRepository:
    """Integration tests for diarization job CRUD operations."""

    async def test_create_and_get_job(self, session: AsyncSession) -> None:
        """Test creating and retrieving a diarization job."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create(title="Test Meeting")
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_QUEUED,
        )
        created = await job_repo.create(job)
        await session.commit()

        assert created.job_id == job.job_id, "created job should have same job_id"
        assert created.status == JOB_STATUS_QUEUED, "created job should have queued status"

        retrieved = await job_repo.get(job.job_id)

        assert retrieved is not None, "get should return the created job"
        assert retrieved.job_id == job.job_id, "retrieved job_id should match"
        assert retrieved.meeting_id == job.meeting_id, "retrieved meeting_id should match"
        assert retrieved.status == JOB_STATUS_QUEUED, "retrieved status should be queued"
        assert retrieved.segments_updated == 0, "segments_updated should default to 0"
        assert retrieved.speaker_ids == [], "speaker_ids should default to empty list"
        assert retrieved.error_message == "", "error_message should default to empty string"

    async def test_get_nonexistent_job_returns_none(self, session: AsyncSession) -> None:
        """Test retrieving a job that doesn't exist returns None."""
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        result = await job_repo.get("nonexistent-job-id")

        assert result is None, "get should return None for nonexistent job ID"

    async def test_update_status_to_running(self, session: AsyncSession) -> None:
        """Test updating job status from QUEUED to RUNNING."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_QUEUED,
        )
        await job_repo.create(job)
        await session.commit()

        updated = await job_repo.update_status(job.job_id, JOB_STATUS_RUNNING)
        await session.commit()

        assert updated is True, "update_status should return True on success"

        retrieved = await job_repo.get(job.job_id)
        assert retrieved is not None, "job should still be retrievable after status update"
        assert retrieved.status == JOB_STATUS_RUNNING, f"expected RUNNING status, got {retrieved.status}"

    async def test_update_status_to_completed_with_results(self, session: AsyncSession) -> None:
        """Test updating job status to COMPLETED with segments and speakers."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_RUNNING,
        )
        await job_repo.create(job)
        await session.commit()

        speaker_ids = ["SPEAKER_00", "SPEAKER_01", "SPEAKER_02"]
        updated = await job_repo.update_status(
            job.job_id,
            JOB_STATUS_COMPLETED,
            segments_updated=EXPECTED_SEGMENTS_UPDATED,
            speaker_ids=speaker_ids,
        )
        await session.commit()

        assert updated is True, "update_status should return True on success"

        retrieved = await job_repo.get(job.job_id)
        assert retrieved is not None, "job should be retrievable after update"
        assert retrieved.status == JOB_STATUS_COMPLETED, "status should be completed"
        assert retrieved.segments_updated == EXPECTED_SEGMENTS_UPDATED, "segments updated mismatch"
        assert retrieved.speaker_ids == speaker_ids, "speaker_ids should be persisted"
        assert retrieved.error_message == "", "error_message should remain empty"

    async def test_update_status_to_failed_with_error(self, session: AsyncSession) -> None:
        """Test updating job status to FAILED with error message."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_RUNNING,
        )
        await job_repo.create(job)
        await session.commit()

        error_msg = "Model inference failed: CUDA out of memory"
        updated = await job_repo.update_status(
            job.job_id,
            JOB_STATUS_FAILED,
            error_message=error_msg,
        )
        await session.commit()

        assert updated is True, "Expected update to succeed"

        retrieved = await job_repo.get(job.job_id)
        assert retrieved is not None, "Expected job to be retrievable"
        assert retrieved.status == JOB_STATUS_FAILED, "Expected FAILED status"
        assert retrieved.error_message == error_msg, "Expected error message to be stored"

    async def test_update_nonexistent_job_returns_false(self, session: AsyncSession) -> None:
        """Test updating a job that doesn't exist returns False."""
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        result = await job_repo.update_status("nonexistent", JOB_STATUS_RUNNING)
        await session.commit()

        assert result is False, "update_status should return False for nonexistent job"

    async def test_list_for_meeting_returns_jobs_in_order(self, session: AsyncSession) -> None:
        """Test listing jobs for a meeting returns them newest first."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job_ids: list[str] = []
        for _ in range(3):
            job = DiarizationJob(
                job_id=str(uuid4()),
                meeting_id=str(meeting.id),
                status=JOB_STATUS_QUEUED,
            )
            await job_repo.create(job)
            job_ids.append(job.job_id)
        await session.commit()

        jobs_sequence: Sequence[DiarizationJob] = await job_repo.list_for_meeting(str(meeting.id))
        jobs: list[DiarizationJob] = list(jobs_sequence)

        assert len(jobs) == 3, f"expected 3 jobs for meeting, got {len(jobs)}"
        job_ids_from_jobs: list[str] = [j.job_id for j in jobs]
        reversed_job_ids: list[str] = list(reversed(job_ids))
        assert job_ids_from_jobs == reversed_job_ids, "jobs should be returned in newest-first order"

    async def test_list_for_meeting_excludes_other_meetings(self, session: AsyncSession) -> None:
        """Test listing jobs only returns jobs for the specified meeting."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting1 = Meeting.create(title="Meeting 1")
        meeting2 = Meeting.create(title="Meeting 2")
        await meeting_repo.create(meeting1)
        await meeting_repo.create(meeting2)
        await session.commit()

        job1 = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting1.id),
            status=JOB_STATUS_QUEUED,
        )
        job2 = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting2.id),
            status=JOB_STATUS_QUEUED,
        )
        await job_repo.create(job1)
        await job_repo.create(job2)
        await session.commit()

        jobs = await job_repo.list_for_meeting(str(meeting1.id))

        assert len(jobs) == 1, f"expected 1 job for meeting1, got {len(jobs)}"
        assert jobs[0].job_id == job1.job_id, "returned job should be from meeting1, not meeting2"


@pytest.mark.integration
class TestDiarizationJobCrashRecovery:
    """Integration tests for crash recovery functionality."""

    async def test_mark_running_as_failed_recovers_queued_jobs(
        self, session: AsyncSession
    ) -> None:
        """Test crash recovery marks QUEUED jobs as FAILED."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_QUEUED,
        )
        await job_repo.create(job)
        await session.commit()

        failed_count = await job_repo.mark_running_as_failed("Server crashed")
        await session.commit()

        assert failed_count == 1, "should mark 1 queued job as failed"

        retrieved = await job_repo.get(job.job_id)
        assert retrieved is not None, "job should still be retrievable"
        assert retrieved.status == JOB_STATUS_FAILED, "status should be FAILED"
        assert retrieved.error_message == "Server crashed", "error message should be set"

    async def test_mark_running_as_failed_recovers_running_jobs(
        self, session: AsyncSession
    ) -> None:
        """Test crash recovery marks RUNNING jobs as FAILED."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_RUNNING,
        )
        await job_repo.create(job)
        await session.commit()

        failed_count = await job_repo.mark_running_as_failed()
        await session.commit()

        assert failed_count == 1, "should mark 1 running job"

        retrieved = await job_repo.get(job.job_id)
        assert retrieved is not None, "job should still be retrievable"
        assert retrieved.status == JOB_STATUS_FAILED, "status should be FAILED"

    async def test_mark_running_as_failed_ignores_completed_jobs(
        self, session: AsyncSession
    ) -> None:
        """Test crash recovery does not affect COMPLETED jobs."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_COMPLETED,
            segments_updated=10,
        )
        await job_repo.create(job)
        await session.commit()

        failed_count = await job_repo.mark_running_as_failed()
        await session.commit()

        assert failed_count == 0, "should not mark completed jobs"

        retrieved = await job_repo.get(job.job_id)
        assert retrieved is not None, "completed job should exist"
        assert retrieved.status == JOB_STATUS_COMPLETED, "status should remain COMPLETED"
        assert retrieved.segments_updated == 10, "segments_updated should remain unchanged"

    async def test_mark_running_as_failed_ignores_already_failed_jobs(
        self, session: AsyncSession
    ) -> None:
        """Test crash recovery does not affect already FAILED jobs."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_FAILED,
            error_message="Original error",
        )
        await job_repo.create(job)
        await session.commit()

        failed_count = await job_repo.mark_running_as_failed("New crash")
        await session.commit()

        assert failed_count == 0, "should not mark already FAILED jobs"

        retrieved = await job_repo.get(job.job_id)
        assert retrieved is not None, "failed job should still be retrievable"
        assert retrieved.error_message == "Original error", "original error message should be preserved"

    async def test_mark_running_as_failed_handles_multiple_jobs(
        self, session: AsyncSession
    ) -> None:
        """Test crash recovery marks all active jobs across meetings."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting1 = Meeting.create()
        meeting2 = Meeting.create()
        await meeting_repo.create(meeting1)
        await meeting_repo.create(meeting2)
        await session.commit()

        queued_job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting1.id),
            status=JOB_STATUS_QUEUED,
        )
        running_job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting1.id),
            status=JOB_STATUS_RUNNING,
        )
        completed_job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting2.id),
            status=JOB_STATUS_COMPLETED,
        )
        await job_repo.create(queued_job)
        await job_repo.create(running_job)
        await job_repo.create(completed_job)
        await session.commit()

        failed_count = await job_repo.mark_running_as_failed()
        await session.commit()

        assert failed_count == 2, "should mark queued and running jobs"

        j1 = await job_repo.get(queued_job.job_id)
        j2 = await job_repo.get(running_job.job_id)
        j3 = await job_repo.get(completed_job.job_id)
        assert j1 is not None and j1.status == JOB_STATUS_FAILED, "queued job should be failed"
        assert j2 is not None and j2.status == JOB_STATUS_FAILED, "running job should be failed"
        assert j3 is not None and j3.status == JOB_STATUS_COMPLETED, "completed job unchanged"


@pytest.mark.integration
class TestDiarizationJobPruning:
    """Integration tests for job pruning functionality."""

    async def test_prune_completed_removes_old_completed_jobs(
        self, session: AsyncSession
    ) -> None:
        """Test pruning removes completed jobs older than TTL."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        old_time = datetime.now() - timedelta(hours=2)
        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_COMPLETED,
            created_at=old_time,
            updated_at=old_time,
        )
        await job_repo.create(job)
        await session.commit()

        pruned = await job_repo.prune_completed(ttl_seconds=3600)
        await session.commit()

        assert pruned == 1, f"expected 1 old completed job pruned, got {pruned}"
        assert await job_repo.get(job.job_id) is None, "old completed job should be deleted after pruning"

    async def test_prune_completed_removes_old_failed_jobs(
        self, session: AsyncSession
    ) -> None:
        """Test pruning removes failed jobs older than TTL."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        old_time = datetime.now() - timedelta(hours=2)
        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_FAILED,
            created_at=old_time,
            updated_at=old_time,
        )
        await job_repo.create(job)
        await session.commit()

        pruned = await job_repo.prune_completed(ttl_seconds=3600)
        await session.commit()

        assert pruned == 1, f"expected 1 old failed job pruned, got {pruned}"

    async def test_prune_completed_keeps_recent_jobs(self, session: AsyncSession) -> None:
        """Test pruning keeps jobs within TTL window."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_COMPLETED,
        )
        await job_repo.create(job)
        await session.commit()

        pruned = await job_repo.prune_completed(ttl_seconds=3600)
        await session.commit()

        assert pruned == 0, "recent jobs should not be pruned"
        assert await job_repo.get(job.job_id) is not None, "recent job should still exist after pruning"

    async def test_prune_completed_keeps_running_jobs(self, session: AsyncSession) -> None:
        """Test pruning never removes RUNNING or QUEUED jobs."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        old_time = datetime.now() - timedelta(hours=2)
        running_job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_RUNNING,
            created_at=old_time,
            updated_at=old_time,
        )
        queued_job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_QUEUED,
            created_at=old_time,
            updated_at=old_time,
        )
        await job_repo.create(running_job)
        await job_repo.create(queued_job)
        await session.commit()

        pruned = await job_repo.prune_completed(ttl_seconds=0)
        await session.commit()

        assert pruned == 0, "RUNNING and QUEUED jobs should never be pruned"
        assert await job_repo.get(running_job.job_id) is not None, "old RUNNING job should still exist"
        assert await job_repo.get(queued_job.job_id) is not None, "old QUEUED job should still exist"


@pytest.mark.integration
class TestStreamingDiarizationTurns:
    """Integration tests for streaming diarization turn persistence."""

    async def test_add_and_get_streaming_turns(self, session: AsyncSession) -> None:
        """Test adding and retrieving streaming diarization turns."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        turns = [
            StreamingTurn(speaker="SPEAKER_00", start_time=0.0, end_time=2.5, confidence=0.92),
            StreamingTurn(speaker="SPEAKER_01", start_time=2.5, end_time=5.0, confidence=0.88),
            StreamingTurn(speaker="SPEAKER_00", start_time=5.0, end_time=8.0, confidence=0.95),
        ]
        count = await job_repo.add_streaming_turns(str(meeting.id), turns)
        await session.commit()

        assert count == 3, "Should add 3 turns"

        retrieved = await job_repo.get_streaming_turns(str(meeting.id))

        assert len(retrieved) == 3, "Should retrieve 3 turns"
        assert retrieved[0].speaker == "SPEAKER_00", "First turn speaker should match"
        assert retrieved[0].start_time == 0.0, "First turn start time should match"
        assert retrieved[0].end_time == 2.5, "First turn end time should match"
        assert retrieved[0].confidence == approx_float(0.92), "First turn confidence should match"
        assert retrieved[1].speaker == "SPEAKER_01", "Second turn speaker should match"
        assert retrieved[2].speaker == "SPEAKER_00", "Third turn speaker should match"

    async def test_get_streaming_turns_returns_empty_for_no_turns(
        self, session: AsyncSession
    ) -> None:
        """Test getting turns for a meeting with no turns returns empty list."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        turns = await job_repo.get_streaming_turns(str(meeting.id))

        assert turns == [], "should return empty list for meeting with no turns"

    async def test_add_streaming_turns_empty_list_returns_zero(
        self, session: AsyncSession
    ) -> None:
        """Test adding empty turns list returns 0."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        count = await job_repo.add_streaming_turns(str(meeting.id), [])
        await session.commit()

        assert count == 0, "adding empty turns list should return 0"

    async def test_streaming_turns_ordered_by_start_time(self, session: AsyncSession) -> None:
        """Test turns are returned ordered by start_time regardless of insertion order."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        turns1 = [StreamingTurn(speaker="S2", start_time=5.0, end_time=7.0)]
        turns2 = [StreamingTurn(speaker="S1", start_time=0.0, end_time=3.0)]
        turns3 = [StreamingTurn(speaker="S2", start_time=3.0, end_time=5.0)]
        await job_repo.add_streaming_turns(str(meeting.id), turns1)
        await job_repo.add_streaming_turns(str(meeting.id), turns2)
        await job_repo.add_streaming_turns(str(meeting.id), turns3)
        await session.commit()

        retrieved = await job_repo.get_streaming_turns(str(meeting.id))

        assert len(retrieved) == 3, f"expected 3 turns retrieved, got {len(retrieved)}"
        assert [t.start_time for t in retrieved] == [0.0, 3.0, 5.0], "turns should be ordered by start_time"

    async def test_clear_streaming_turns_removes_all(self, session: AsyncSession) -> None:
        """Test clearing streaming turns removes all turns for a meeting."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        turns = [
            StreamingTurn(speaker="S1", start_time=0.0, end_time=1.0),
            StreamingTurn(speaker="S2", start_time=1.0, end_time=2.0),
        ]
        await job_repo.add_streaming_turns(str(meeting.id), turns)
        await session.commit()

        deleted = await job_repo.clear_streaming_turns(str(meeting.id))
        await session.commit()

        assert deleted == 2, f"expected 2 turns deleted, got {deleted}"

        remaining = await job_repo.get_streaming_turns(str(meeting.id))
        assert remaining == [], "no turns should remain after clear"

    async def test_clear_streaming_turns_isolates_meetings(self, session: AsyncSession) -> None:
        """Test clearing turns for one meeting doesn't affect others."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting1 = Meeting.create()
        meeting2 = Meeting.create()
        await meeting_repo.create(meeting1)
        await meeting_repo.create(meeting2)
        await session.commit()

        turns1 = [StreamingTurn(speaker="S1", start_time=0.0, end_time=1.0)]
        turns2 = [StreamingTurn(speaker="S2", start_time=0.0, end_time=1.0)]
        await job_repo.add_streaming_turns(str(meeting1.id), turns1)
        await job_repo.add_streaming_turns(str(meeting2.id), turns2)
        await session.commit()

        await job_repo.clear_streaming_turns(str(meeting1.id))
        await session.commit()

        assert await job_repo.get_streaming_turns(str(meeting1.id)) == [], "meeting1 turns should be cleared"
        remaining = await job_repo.get_streaming_turns(str(meeting2.id))
        assert len(remaining) == 1, "meeting2 should still have its turn"
        assert remaining[0].speaker == "S2", f"meeting2 turn should have speaker S2, got {remaining[0].speaker}"

    async def test_streaming_turns_deleted_on_meeting_cascade(
        self, session: AsyncSession
    ) -> None:
        """Test streaming turns are cascade deleted when meeting is deleted."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        turns = [StreamingTurn(speaker="S1", start_time=0.0, end_time=1.0)]
        await job_repo.add_streaming_turns(str(meeting.id), turns)
        await session.commit()

        await meeting_repo.delete(meeting.id)
        await session.commit()

        remaining = await job_repo.get_streaming_turns(str(meeting.id))
        assert remaining == [], "turns should be cascade deleted when meeting is deleted"


@pytest.mark.integration
class TestDiarizationJobCascadeDelete:
    """Integration tests for cascade delete behavior."""

    async def test_jobs_deleted_when_meeting_deleted(self, session: AsyncSession) -> None:
        """Test diarization jobs are cascade deleted when meeting is deleted."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_COMPLETED,
        )
        await job_repo.create(job)
        await session.commit()

        await meeting_repo.delete(meeting.id)
        await session.commit()

        assert await job_repo.get(job.job_id) is None, "job should be cascade deleted when meeting is deleted"


@pytest.mark.integration
class TestDiarizationJobConcurrency:
    """Integration tests for concurrency guards."""

    async def test_get_active_for_meeting_returns_queued_job(
        self, session: AsyncSession
    ) -> None:
        """Test get_active_for_meeting returns a QUEUED job."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_QUEUED,
        )
        await job_repo.create(job)
        await session.commit()

        active = await job_repo.get_active_for_meeting(str(meeting.id))

        assert active is not None, "get_active_for_meeting should return QUEUED job"
        assert active.job_id == job.job_id, "returned job_id should match the created job"
        assert active.status == JOB_STATUS_QUEUED, f"expected QUEUED status, got {active.status}"

    async def test_get_active_for_meeting_returns_running_job(
        self, session: AsyncSession
    ) -> None:
        """Test get_active_for_meeting returns a RUNNING job."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_RUNNING,
        )
        await job_repo.create(job)
        await session.commit()

        active = await job_repo.get_active_for_meeting(str(meeting.id))

        assert active is not None, "get_active_for_meeting should return RUNNING job"
        assert active.job_id == job.job_id, "returned job_id should match the created job"
        assert active.status == JOB_STATUS_RUNNING, f"expected RUNNING status, got {active.status}"

    async def test_get_active_for_meeting_returns_none_for_completed(
        self, session: AsyncSession
    ) -> None:
        """Test get_active_for_meeting returns None for COMPLETED job."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_COMPLETED,
        )
        await job_repo.create(job)
        await session.commit()

        active = await job_repo.get_active_for_meeting(str(meeting.id))

        assert active is None, "get_active_for_meeting should return None for COMPLETED job"

    async def test_get_active_for_meeting_returns_none_for_cancelled(
        self, session: AsyncSession
    ) -> None:
        """Test get_active_for_meeting returns None for CANCELLED job."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_CANCELLED,
        )
        await job_repo.create(job)
        await session.commit()

        active = await job_repo.get_active_for_meeting(str(meeting.id))

        assert active is None, "get_active_for_meeting should return None for CANCELLED job"

    async def test_get_active_for_meeting_returns_most_recent(
        self, session: AsyncSession
    ) -> None:
        """Test get_active_for_meeting returns the most recent active job."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job1 = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_QUEUED,
        )
        await job_repo.create(job1)
        await session.commit()

        job2 = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_RUNNING,
        )
        await job_repo.create(job2)
        await session.commit()

        active = await job_repo.get_active_for_meeting(str(meeting.id))

        assert active is not None, "should return an active job when multiple exist"
        assert active.job_id == job2.job_id, "should return the most recently created active job"


@pytest.mark.integration
class TestDiarizationJobCancelledStatus:
    """Integration tests for CANCELLED status handling."""

    async def test_update_status_to_cancelled(self, session: AsyncSession) -> None:
        """Test updating job status to CANCELLED."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_RUNNING,
        )
        await job_repo.create(job)
        await session.commit()

        updated = await job_repo.update_status(job.job_id, JOB_STATUS_CANCELLED)
        await session.commit()

        assert updated is True, "update_status to CANCELLED should return True"

        retrieved = await job_repo.get(job.job_id)
        assert retrieved is not None, "cancelled job should still be retrievable"
        assert retrieved.status == JOB_STATUS_CANCELLED, f"expected CANCELLED status, got {retrieved.status}"

    async def test_prune_includes_cancelled_jobs(self, session: AsyncSession) -> None:
        """Test pruning removes old CANCELLED jobs along with COMPLETED/FAILED."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        old_time = datetime.now() - timedelta(hours=2)
        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_CANCELLED,
            created_at=old_time,
            updated_at=old_time,
        )
        await job_repo.create(job)
        await session.commit()

        pruned = await job_repo.prune_completed(ttl_seconds=3600)
        await session.commit()

        assert pruned == 1, f"expected 1 old CANCELLED job pruned, got {pruned}"
        assert await job_repo.get(job.job_id) is None, "old CANCELLED job should be deleted after pruning"

    async def test_mark_running_ignores_cancelled_jobs(
        self, session: AsyncSession
    ) -> None:
        """Test crash recovery does not affect CANCELLED jobs."""
        meeting_repo = SqlAlchemyMeetingRepository(session)
        job_repo = SqlAlchemyDiarizationJobRepository(session)

        meeting = Meeting.create()
        await meeting_repo.create(meeting)
        await session.commit()

        job = DiarizationJob(
            job_id=str(uuid4()),
            meeting_id=str(meeting.id),
            status=JOB_STATUS_CANCELLED,
        )
        await job_repo.create(job)
        await session.commit()

        failed_count = await job_repo.mark_running_as_failed()
        await session.commit()

        assert failed_count == 0, "crash recovery should not affect CANCELLED jobs"

        retrieved = await job_repo.get(job.job_id)
        assert retrieved is not None, "cancelled job should still be retrievable"
        assert retrieved.status == JOB_STATUS_CANCELLED, "CANCELLED status should remain unchanged"
