# retoor <retoor@molodetz.nl>

import re
import socket
import time
from uuid import uuid4

import pytest
import requests
from playwright.sync_api import expect

from devplacepy.database import get_table, set_setting
from devplacepy.services.containers import store
from devplacepy.services.containers.workspace import flags, naming
from devplacepy.utils import make_combined_slug
from tests.conftest import BASE_URL

EDITOR_CONTAINER_PORT = 8443
TRANSITION_TIMEOUT_MS = 8000


@pytest.fixture(autouse=True)
def _workspaces_on():
    previous = None
    row = get_table("site_settings").find_one(key="workspace_enabled")
    if row:
        previous = row.get("value")
    set_setting("workspace_enabled", "1")
    try:
        yield
    finally:
        set_setting("workspace_enabled", previous if previous is not None else "0")
        instances = get_table("instances")
        created = [r["uid"] for r in instances.find(is_workspace=1)]
        for uid in created:
            get_table("tunnels").delete(instance_uid=uid)
            get_table("workspace_flags").delete(instance_uid=uid)
            instances.delete(uid=uid)
        get_table("workspace_editor_prefs").delete()


def _row_for(user: dict) -> dict:
    return get_table("users").find_one(username=user["username"])


def _project_for(owner_uid: str, title: str = "WS Project") -> dict:
    uid = str(uuid4())
    slug = make_combined_slug(title, uid)
    row = {
        "uid": uid,
        "user_uid": owner_uid,
        "title": title,
        "description": "workspace host project",
        "slug": slug,
        "project_type": "software",
        "status": "In Development",
        "platforms": "",
        "is_private": 0,
        "read_only": 0,
        "stars": 0,
        "created_at": "2026-01-01T00:00:00+00:00",
        "deleted_at": None,
        "deleted_by": None,
    }
    get_table("projects").insert(row)
    return row


@pytest.fixture(scope="module")
def editor_listener():
    sock = socket.socket()
    sock.bind(("127.0.0.1", 0))
    sock.listen(16)
    try:
        yield sock.getsockname()[1]
    finally:
        sock.close()


def _workspace_for(
    project: dict, owner_uid: str, editor_port: int = 0, **overrides
) -> dict:
    payload = {
        "project_uid": project["uid"],
        "name": "ws-e2e",
        "status": "running",
        "desired_state": "running",
        "is_workspace": 1,
        "workspace_owner_uid": owner_uid,
        "tunnel_name": naming.generate(),
        "ports_json": '[{"host": 20777, "container": 8080, "proto": "tcp"}]',
    }
    if editor_port:
        payload["editor_port"] = EDITOR_CONTAINER_PORT
        payload["ports_json"] = (
            f'[{{"host": {editor_port}, "container": {EDITOR_CONTAINER_PORT}, '
            '"proto": "tcp"}]'
        )
    payload.update(overrides)
    return store.create_instance(payload)


def _phase_view(page, phase: str):
    return page.locator(f".workspace-phase-view[data-phase-view~='{phase}']")


def test_project_page_offers_the_workspace_entry_point_to_the_owner(alice):
    page, user = alice
    project = _project_for(_row_for(user)["uid"], "WS Entry")
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}", wait_until="domcontentloaded"
    )
    page.locator(".project-actions-more").click()
    entry = page.locator(".context-menu-item:has-text('Workspace')")
    entry.wait_for(state="visible")
    entry.click()
    page.wait_for_url(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    page.locator(".workspace-page").wait_for(state="visible")


def test_project_page_hides_the_workspace_entry_point_from_a_non_owner(bob):
    page, user = bob
    project = _project_for(str(uuid4()), "WS Entry Foreign")
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}", wait_until="domcontentloaded"
    )
    page.locator(".project-actions-more").click()
    assert not page.locator(".context-menu-item:has-text('Workspace')").count()


def test_project_page_shows_a_direct_vscode_button_for_a_running_workspace(
    alice, editor_listener
):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Direct")
    instance = _workspace_for(project, row["uid"], editor_port=editor_listener)
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}", wait_until="domcontentloaded"
    )
    button = page.locator(".project-detail-actions a[data-editor-open]")
    button.wait_for(state="visible")
    expect(button).to_have_attribute("target", "_blank")
    expect(button).to_have_attribute(
        "href",
        f"/projects/{project['slug']}/containers/instances/{instance['uid']}/code/",
    )


def test_direct_vscode_button_is_absent_while_the_workspace_is_stopped(alice):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Stopped")
    _workspace_for(project, row["uid"], status="stopped", desired_state="stopped")
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}", wait_until="domcontentloaded"
    )
    page.locator(".project-detail-actions").wait_for(state="visible")
    assert not page.locator(".project-detail-actions a[data-editor-open]").count()


def _await_workspace_flag(slug: str, key: str, expected: bool) -> bool:
    deadline = time.time() + 10.0
    url = f"{BASE_URL}/projects/{slug}"
    headers = {"Accept": "application/json", "X-API-KEY": key}
    while time.time() < deadline:
        body = requests.get(url, headers=headers).json()
        if body["viewer_can_workspace"] is expected:
            return True
        time.sleep(0.2)
    return False


def test_project_page_hides_the_workspace_entry_point_when_disabled(alice):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Entry Off")
    set_setting("workspace_enabled", "0")
    try:
        assert _await_workspace_flag(project["slug"], row["api_key"], False)
        page.goto(
            f"{BASE_URL}/projects/{project['slug']}", wait_until="domcontentloaded"
        )
        page.locator(".project-actions-more").click()
        assert not page.locator(".context-menu-item:has-text('Workspace')").count()
    finally:
        set_setting("workspace_enabled", "1")
        assert _await_workspace_flag(project["slug"], row["api_key"], True)


def test_workspace_page_offers_creation_to_owner(alice):
    page, user = alice
    project = _project_for(_row_for(user)["uid"])
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    page.locator(".workspace-page").wait_for(state="visible")
    expect(page.locator("button:has-text('Open workspace')")).to_be_visible()


def test_workspace_page_shows_state_quota_and_tunnel_form(alice):
    page, user = alice
    project = _project_for(_row_for(user)["uid"], "WS Detail")
    _workspace_for(project, _row_for(user)["uid"])
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    page.locator(".workspace-summary").wait_for(state="visible")
    expect(page.locator("[data-workspace-status]")).to_contain_text("running")
    expect(page.locator(".workspace-meter").first).to_be_visible()
    expect(page.locator(".workspace-tunnel-form")).to_be_visible()
    expect(page.locator(".workspace-help")).to_contain_text("sudo")


def test_owner_can_add_and_remove_a_tunnel(alice):
    page, user = alice
    project = _project_for(_row_for(user)["uid"], "WS Tunnel")
    _workspace_for(project, _row_for(user)["uid"])
    url = f"{BASE_URL}/projects/{project['slug']}/workspace"
    page.goto(url, wait_until="domcontentloaded")
    page.locator(".workspace-tunnel-form input[name='container_port']").fill("8080")
    page.locator(".workspace-tunnel-form button:has-text('Add tunnel')").click()
    page.wait_for_url(url, wait_until="domcontentloaded")
    tunnel = page.locator(".workspace-tunnel").first
    tunnel.wait_for(state="visible")
    expect(tunnel).to_contain_text(naming.domain())

    page.locator(".workspace-tunnel button:has-text('Remove')").first.click()
    page.locator(".dialog-confirm").wait_for(state="visible")
    page.locator(".dialog-confirm").click()
    page.wait_for_url(url, wait_until="domcontentloaded")
    expect(page.locator(".workspace-tunnel-list")).to_contain_text("No tunnels yet")


def test_suspended_workspace_shows_reason_to_owner(alice):
    page, user = alice
    project = _project_for(_row_for(user)["uid"], "WS Suspended")
    instance = _workspace_for(project, _row_for(user)["uid"])
    store.update_instance(
        instance["uid"],
        {"suspended_at": "2026-01-01T00:00:00+00:00", "flag_reason": "sustained cpu"},
    )
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    banner = page.locator(".workspace-suspended")
    banner.wait_for(state="visible")
    expect(banner).to_contain_text("sustained cpu")


def test_open_flag_is_visible_to_the_owner(alice):
    page, user = alice
    project = _project_for(_row_for(user)["uid"], "WS Flagged")
    instance = _workspace_for(project, _row_for(user)["uid"])
    flags.raise_flag(
        store.get_instance(instance["uid"]),
        flags.KIND_EGRESS,
        "warn",
        "egress above the hourly ceiling",
        2048.0,
        1024.0,
    )
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    flag = page.locator(".workspace-flag").first
    flag.wait_for(state="visible")
    expect(flag).to_contain_text("egress above the hourly ceiling")


def test_guest_cannot_reach_the_workspace_page(page):
    project = _project_for(str(uuid4()), "WS Guest")
    response = requests.get(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        allow_redirects=False,
    )
    assert response.status_code in (303, 401, 404)


def test_non_owner_member_is_refused(bob):
    page, user = bob
    project = _project_for(str(uuid4()), "WS Foreign")
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    assert not page.locator(".workspace-summary").count()


def test_admin_console_lists_and_suspends_a_workspace(alice):
    page, user = alice
    project = _project_for(_row_for(user)["uid"], "WS Admin")
    instance = _workspace_for(project, _row_for(user)["uid"])
    page.goto(f"{BASE_URL}/admin/workspaces", wait_until="domcontentloaded")
    row = page.locator(f"tr[data-workspace-uid='{instance['uid']}']")
    row.wait_for(state="visible")
    expect(row).to_contain_text("ws-e2e")

    row.locator("input[name='reason']").fill("policy breach")
    row.locator("button:has-text('Suspend')").click()
    page.wait_for_url(f"{BASE_URL}/admin/workspaces", wait_until="domcontentloaded")

    refreshed = store.get_instance(instance["uid"])
    assert refreshed["suspended_at"]
    row = page.locator(f"tr[data-workspace-uid='{instance['uid']}']")
    expect(row).to_contain_text("suspended")

    row.locator("button:has-text('Unsuspend')").click()
    page.wait_for_url(f"{BASE_URL}/admin/workspaces", wait_until="domcontentloaded")
    assert not store.get_instance(instance["uid"])["suspended_at"]


def test_admin_console_raises_and_resolves_a_flag(alice):
    page, user = alice
    project = _project_for(_row_for(user)["uid"], "WS Flag Admin")
    instance = _workspace_for(project, _row_for(user)["uid"])
    page.goto(f"{BASE_URL}/admin/workspaces", wait_until="domcontentloaded")
    row = page.locator(f"tr[data-workspace-uid='{instance['uid']}']")
    row.wait_for(state="visible")
    row.locator("input[name='detail']").fill("manual review")
    row.locator("button:has-text('Flag')").click()
    page.wait_for_url(f"{BASE_URL}/admin/workspaces", wait_until="domcontentloaded")

    assert flags.list_flags(instance_uid=instance["uid"])
    page.locator("button:has-text('Resolve')").first.click()
    page.wait_for_url(f"{BASE_URL}/admin/workspaces", wait_until="domcontentloaded")
    assert not flags.list_flags(instance_uid=instance["uid"])


def test_workspaces_sidebar_link_is_present_for_admin(alice):
    page, user = alice
    page.goto(f"{BASE_URL}/admin/workspaces", wait_until="domcontentloaded")
    link = page.locator(".sidebar-link:has-text('Workspaces')")
    link.wait_for(state="visible")
    expect(link).to_have_class(re.compile("active"))


def test_the_editor_card_shows_the_resolved_profile_and_its_sources(alice):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Editor Card")
    _workspace_for(project, row["uid"])
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    card = page.locator(".workspace-editor")
    card.wait_for(state="visible")
    expect(card.locator(".workspace-editor-summary").first).to_contain_text("devplace-dark")
    expect(card.locator(".workspace-editor-resources")).to_contain_text("cores")
    expect(card.locator(".workspace-editor-form select[name='theme']")).to_be_visible()


def test_saving_an_editor_preference_changes_its_source_to_the_member(alice):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Editor Save")
    _workspace_for(project, row["uid"])
    try:
        page.goto(
            f"{BASE_URL}/projects/{project['slug']}/workspace",
            wait_until="domcontentloaded",
        )
        page.locator(".workspace-editor-form").wait_for(state="visible")
        page.select_option(".workspace-editor-form select[name='theme']", "devplace-light")
        page.locator(".workspace-editor-actions button[type='submit']").click()
        page.wait_for_url(
            f"{BASE_URL}/projects/{project['slug']}/workspace",
            wait_until="domcontentloaded",
        )
        summary = page.locator(".workspace-editor-summary").first
        summary.wait_for(state="visible")
        expect(summary).to_contain_text("devplace-light")
        expect(summary).to_contain_text("user")
    finally:
        get_table("workspace_editor_prefs").delete(owner_id=row["uid"])


def test_resetting_editor_preferences_restores_the_site_default(alice):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Editor Reset")
    _workspace_for(project, row["uid"])
    try:
        page.goto(
            f"{BASE_URL}/projects/{project['slug']}/workspace",
            wait_until="domcontentloaded",
        )
        page.locator(".workspace-editor-form").wait_for(state="visible")
        page.select_option(".workspace-editor-form select[name='theme']", "devplace-light")
        page.locator(".workspace-editor-actions button[type='submit']").click()
        page.wait_for_url(
            f"{BASE_URL}/projects/{project['slug']}/workspace",
            wait_until="domcontentloaded",
        )
        page.locator(".workspace-editor-reset button").click()
        page.wait_for_url(
            f"{BASE_URL}/projects/{project['slug']}/workspace",
            wait_until="domcontentloaded",
        )
        summary = page.locator(".workspace-editor-summary").first
        summary.wait_for(state="visible")
        expect(summary).to_contain_text("devplace-dark")
    finally:
        get_table("workspace_editor_prefs").delete(owner_id=row["uid"])


def test_the_editor_launch_control_carries_the_window_profile(alice, editor_listener):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Editor Launch")
    instance = _workspace_for(project, row["uid"], editor_port=editor_listener)
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    launch = page.locator(".workspace-actions a[data-editor-open]")
    launch.wait_for(state="visible")
    expect(launch).to_have_attribute("data-editor-mode", "tab")
    expect(launch).to_have_attribute("target", "_blank")
    expect(launch).to_have_attribute(
        "href",
        f"/projects/{project['slug']}/containers/instances/{instance['uid']}/code/",
    )


def test_the_workspace_help_states_that_every_folder_is_trusted(alice):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Editor Trust")
    _workspace_for(project, row["uid"])
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    help_card = page.locator(".workspace-help")
    help_card.wait_for(state="visible")
    expect(help_card).to_contain_text("Restricted Mode")
    expect(help_card).to_contain_text("dpc")


def test_direct_vscode_button_is_absent_while_the_editor_is_unreachable(alice):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Booting")
    _workspace_for(project, row["uid"])
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}", wait_until="domcontentloaded"
    )
    page.locator(".project-detail-actions").wait_for(state="visible")
    assert not page.locator(".project-detail-actions a[data-editor-open]").count()


def test_a_running_workspace_whose_editor_is_not_listening_reads_starting(alice):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Not Listening")
    _workspace_for(project, row["uid"])
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    expect(page.locator("[data-workspace-phase]")).to_have_text("Starting")
    expect(_phase_view(page, "starting")).to_be_visible()
    expect(_phase_view(page, "starting").locator(".btn.is-loading")).to_be_disabled()
    expect(_phase_view(page, "ready")).to_be_hidden()
    assert not page.locator(".workspace-actions a[data-editor-open]").is_visible()


def test_start_shows_progress_and_flips_to_open_editor_when_the_editor_answers(
    alice, editor_listener
):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Start Flow")
    instance = _workspace_for(
        project,
        row["uid"],
        editor_port=editor_listener,
        status="stopped",
        desired_state="stopped",
    )
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    expect(page.locator("[data-workspace-phase]")).to_have_text("Stopped")
    start = _phase_view(page, "stopped").locator("button[type='submit']")
    expect(start).to_be_visible()
    start.click()
    expect(_phase_view(page, "starting")).to_be_visible(timeout=TRANSITION_TIMEOUT_MS)
    expect(_phase_view(page, "stopped")).to_be_hidden()
    expect(page.locator("[data-workspace-phase]")).to_have_text("Starting")
    assert store.get_instance(instance["uid"])["desired_state"] == "running"

    store.update_instance(instance["uid"], {"status": "running"})
    launch = page.locator(".workspace-actions a[data-editor-open]")
    launch.wait_for(state="visible", timeout=TRANSITION_TIMEOUT_MS)
    expect(page.locator("[data-workspace-phase]")).to_have_text("Ready")
    expect(_phase_view(page, "starting")).to_be_hidden()
    expect(launch).to_have_attribute(
        "href",
        f"/projects/{project['slug']}/containers/instances/{instance['uid']}/code/",
    )


def test_stop_shows_progress_until_the_container_is_down(alice, editor_listener):
    page, user = alice
    row = _row_for(user)
    project = _project_for(row["uid"], "WS Stop Flow")
    instance = _workspace_for(project, row["uid"], editor_port=editor_listener)
    page.goto(
        f"{BASE_URL}/projects/{project['slug']}/workspace",
        wait_until="domcontentloaded",
    )
    expect(page.locator("[data-workspace-phase]")).to_have_text("Ready")
    _phase_view(page, "ready").locator("form[data-workspace-action] button").click()
    expect(_phase_view(page, "stopping")).to_be_visible(timeout=TRANSITION_TIMEOUT_MS)
    expect(page.locator("[data-workspace-phase]")).to_have_text("Stopping")
    assert store.get_instance(instance["uid"])["desired_state"] == "stopped"

    store.update_instance(instance["uid"], {"status": "stopped"})
    expect(_phase_view(page, "stopped")).to_be_visible(timeout=TRANSITION_TIMEOUT_MS)
    expect(page.locator("[data-workspace-phase]")).to_have_text("Stopped")
