"""Idempotent, resumable GitHub + DB bootstrap for a new project.

Single setup pass (no separate HTML skeleton phase):
  1. Materialize generated Pages apps via runtime substrate + starter_assembly
     (not snapshot/live product boilerplate). Imports, template clones, and
     connected repos stay adopt/copy. Then write ``.p2hs.json`` and the correct
     CF Pages ``deploy.yml``. Extraction/copy only — no npm/build on the worker.
  2. ``git init`` + commit and GitHub repo creation run in parallel (same as before).
  3. Ensure the **Cloudflare Pages** project exists via API **before** the first git push, so
     GitHub Actions ``wrangler pages deploy`` does not race a missing project.
  4. Push to GitHub, seed DB (revision, instructions, session, manifest).
  5. Preview branch + push; enqueue initial production deploy (W4P routing + hostnames).

Legacy rows stuck at ``setup_stage="skeleton_pushed"`` are still healed by
``apply_boilerplate_snapshot`` (recovery / resume) until migrated.
"""
from __future__ import annotations

import asyncio
import json
import os
import shutil
import tempfile
from pathlib import Path
from typing import Optional
from uuid import UUID

import structlog

from src.config import get_settings

logger = structlog.get_logger()


def _sanitize_template_platform_runtime(repo_path: str) -> None:
    """Remove legacy template topology wiring before exporting a project.

    Older starters shipped a Pages Function and browser env variables for a
    physical Cloud gateway. W4P owns same-origin ``/api`` routing at the
    hostname, so those files are obsolete and are rejected by the export gate.
    """
    root = Path(repo_path)
    legacy_function = root / "functions" / "api" / "cloud" / "[[path]].ts"
    if legacy_function.is_file():
        legacy_function.unlink()
        parent = legacy_function.parent
        while parent != root and parent.is_dir() and not any(parent.iterdir()):
            parent.rmdir()
            parent = parent.parent

    env_file = root / "src" / "lib" / "p2hs" / "env.ts"
    if env_file.is_file():
        env_file.write_text(
            '''/** Browser-safe P2HS configuration. API traffic stays same-origin. */

export function coreApiBase(): string {
  return "/api";
}

export function cloudGatewayBase(): string {
  return "/api";
}
''',
            encoding="utf-8",
        )

    vite_env = root / "src" / "vite-env.d.ts"
    if vite_env.is_file():
        vite_env.write_text(
            '''/// <reference types="vite/client" />

interface ImportMetaEnv {
  readonly DEV: boolean;
}

interface ImportMeta {
  readonly env: ImportMetaEnv;
}
''',
            encoding="utf-8",
        )

_GITIGNORE = """\
node_modules/
.nuxt/
dist/
.output/
.env
*.local
.DS_Store
# P2HS control plane — never commit
.p2hs.json
.p2hs/
.p2hs-control/
.cursor/
.cursorrules
CLAUDE.md
AGENTS.md
p2hs.platform.json
p2hs/
!p2hs/managed-runtime.json
!p2hs/dependency-build.json
!p2hs/platform-ports.lock.json
*.p2hs-local.bak
.p2hs/local-bak/
"""

_GITIGNORE_APPEND = """\n# P2HS (server-managed ephemeral paths)\np2hs/prompt-media/\n*.p2hs-local.bak\n"""


class SetupStage:
    PENDING = "pending"
    LOCAL_WORKSPACE = "local_workspace"
    GITHUB_REPO = "github_repo"
    SKELETON_PUSHED = "skeleton_pushed"  # Legacy (pre single-phase setup); heal via apply_boilerplate_snapshot
    DB_SEEDED = "db_seeded"
    COMPLETED = "completed"              # Boilerplate on GitHub; setup finished


def _write_file(path: str, content: str) -> None:
    """Write file only when content differs (idempotent)."""
    if os.path.isfile(path):
        try:
            with open(path, "r", encoding="utf-8") as fh:
                if fh.read() == content:
                    return
        except OSError:
            pass
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(content)


def _materialize_generated_source(
    repo_path: str,
    project_type: str,
    settings,
    source: dict,
) -> str:
    """Create-path materialization for P2HS-authored apps.

    Pages types assemble runtime substrate (+ optional starter_id overlay).
    Shopify / AI-assist types keep snapshot or live companion copy.
    Import and template-clone callers never reach this function.
    """
    from src.domain.starter_library.runtime_substrate import (
        uses_starter_assembly,
        write_generated_project_tree,
    )

    if uses_starter_assembly(project_type):
        starter_id = (source.get("starter_id") or "").strip() or None
        return write_generated_project_tree(
            repo_path, project_type, starter_id=starter_id
        )

    from src.workers.boilerplate_store import (
        apply_live_boilerplate,
        apply_snapshot,
        find_latest_version,
        get_snapshot_dir,
    )

    snapshot_dir = get_snapshot_dir(settings)
    version = find_latest_version(project_type, snapshot_dir)
    if version and apply_snapshot(repo_path, project_type, version, snapshot_dir):
        return version
    apply_live_boilerplate(repo_path, project_type)
    return "live"


def _append_gitignore_extras(repo_path: str) -> None:
    """Ensure platform ignore rules exist (prefer appending so boilerplate .gitignore stays)."""
    path = os.path.join(repo_path, ".gitignore")
    if os.path.isfile(path):
        try:
            with open(path, encoding="utf-8") as fh:
                body = fh.read()
            from src.domain.codegen.port_adapter import _gitignore_rule_wins

            if (
                "p2hs/prompt-media" in body
                and "*.p2hs-local.bak" in body
                and ".p2hs.json" in body
                and "!p2hs/managed-runtime.json" in body
                and "!p2hs/dependency-build.json" in body
                and _gitignore_rule_wins(body, "!p2hs/platform-ports.lock.json")
            ):
                return
            extras = ""
            if "p2hs/prompt-media" not in body:
                extras += "p2hs/prompt-media/\n"
            if "*.p2hs-local.bak" not in body:
                extras += "*.p2hs-local.bak\n.p2hs/local-bak/\n"
            if ".p2hs.json" not in body:
                extras += ".p2hs.json\n.p2hs/\n.cursorrules\nCLAUDE.md\n"
                extras += "p2hs/project-brief.md\np2hs/app-blocks-plan.json\n"
                extras += "p2hs/discovery-cloud-contract.json\np2hs/project-state.json\n"
            if "!p2hs/managed-runtime.json" not in body:
                extras += "!p2hs/\np2hs/*\n!p2hs/managed-runtime.json\n"
            if "!p2hs/dependency-build.json" not in body:
                extras += "!p2hs/dependency-build.json\n"
            rule = "!p2hs/platform-ports.lock.json"
            if not _gitignore_rule_wins(body + extras, rule):
                extras += "!p2hs/\np2hs/*\n" + rule + "\n"
            if extras:
                with open(path, "a", encoding="utf-8") as fh:
                    fh.write("\n# P2HS\n" + extras)
        except OSError:
            pass
        return
    _write_file(path, _GITIGNORE)


async def _commit_materialized_template_overlay(
    repo_path: str,
    git_name: str,
    git_email: str,
) -> None:
    """Commit P2HS overlay files after cloning a template repo."""
    env = {
        **os.environ,
        "GIT_AUTHOR_NAME": git_name,
        "GIT_AUTHOR_EMAIL": git_email,
        "GIT_COMMITTER_NAME": git_name,
        "GIT_COMMITTER_EMAIL": git_email,
    }
    from src.workers.repo_ops import stage_publishable_and_assert

    await stage_publishable_and_assert(repo_path)

    proc = await asyncio.create_subprocess_exec(
        "git", "commit", "-m", "chore: add P2HS project config",
        cwd=repo_path,
        env=env,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    if proc.returncode != 0:
        combined = ((stdout or b"") + (stderr or b"")).decode(errors="replace").lower()
        if "nothing to commit" in combined:
            return
        raise RuntimeError(f"git commit template overlay failed: {combined}")


async def _materialize_generated_lockfile(repo_path: str) -> None:
    """Create the canonical npm lockfile for a newly generated app.

    This is intentionally limited to create-path generated trees. Imported and
    customer/template repositories are adopted, never rewritten by setup; their
    lockfile decision belongs to the certified migration boundary.
    """
    package = os.path.join(repo_path, "package.json")
    lockfile = os.path.join(repo_path, "package-lock.json")
    if not os.path.isfile(package):
        return
    npm = shutil.which("npm.cmd" if os.name == "nt" else "npm") or "npm"

    # A generated/boilerplate tree may already contain a lockfile.  Do not
    # assume that it is usable: npm lockfiles can have a valid-looking v2
    # package tree while their legacy dependency section is stale, which makes
    # the CI-side `npm ci` fail before the W4P step starts.  Validate first and
    # only rewrite when npm itself rejects the lockfile.
    if os.path.isfile(lockfile):
        check = await asyncio.create_subprocess_exec(
            npm,
            "ci",
            "--dry-run",
            "--ignore-scripts",
            "--no-audit",
            "--no-fund",
            cwd=repo_path,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        await check.communicate()
        if check.returncode == 0:
            return

    proc = await asyncio.create_subprocess_exec(
        npm,
        "install",
        "--package-lock-only",
        "--ignore-scripts",
        "--no-audit",
        "--no-fund",
        cwd=repo_path,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    if proc.returncode != 0 or not os.path.isfile(lockfile):
        combined = ((stdout or b"") + (stderr or b"")).decode(errors="replace")
        raise RuntimeError(f"P2HS_GENERATED_LOCKFILE_FAILED: {combined[-2000:]}")


async def _materialize_initial_repo_tree(
    repo_path: str,
    project_type: str,
    project_ref: str,
    settings,
    *,
    template_source: Optional[dict] = None,
    project_created_at=None,
    db_factory=None,
    project_id=None,
) -> dict:
    """Materialize source/template/generated substrate and overlay platform files."""
    from src.domain.hosting.deployment_targets import deploy_workflow_for_project
    from src.integrations.github_client import GitHubAppClient
    from src.workers.repo_ops import clone_repo, git_rev_parse_head

    os.makedirs(repo_path, exist_ok=True)

    version = ""
    source = template_source or {}
    source_kind = (source.get("kind") or "").strip()
    imported_sha: Optional[str] = None
    imported_env: list[dict[str, str]] = []
    detected_framework: Optional[str] = None
    template_repo = (source.get("github_repo") or "").strip()
    if source_kind == "github_import":
        from src.integrations.github_user_app_client import GitHubUserAppClient

        source_repo = (source.get("repo_full_name") or "").strip()
        branch = (source.get("branch") or "main").strip() or "main"
        installation_id = int(source.get("github_installation_id") or 0)
        if not source_repo or installation_id <= 0:
            raise RuntimeError("GitHub import source is missing repo or installation.")
        source_client = GitHubUserAppClient()
        source_clone_url = await source_client.get_authenticated_clone_url(installation_id, source_repo)
        with tempfile.TemporaryDirectory(prefix="p2hs-import-") as tmp:
            await clone_repo(source_clone_url, tmp, branch=branch)
            imported_sha = await git_rev_parse_head(tmp)
            root_dir = (source.get("root_dir") or "/").strip().replace("\\", "/")
            root_dir = "" if root_dir in ("", "/", ".") else root_dir.strip("/")
            source_root = os.path.join(tmp, *root_dir.split("/")) if root_dir else tmp
            if not os.path.isdir(source_root):
                raise RuntimeError(f"Source root directory not found: {root_dir or '/'}")
            detected_framework = _assert_import_supported(source_root)
            if source.get("import_env", True):
                imported_env = _collect_env_values(source_root)
            _copy_import_tree(source_root, repo_path)
            _write_repo_familiarity_doc(
                repo_path,
                source_repo=source_repo,
                source_branch=branch,
                source_root_dir=root_dir or "/",
                imported_sha=imported_sha,
            )
        version = f"import:{source_repo[:36]}@{(imported_sha or branch)[:16]}"[:60]
    elif source_kind == "gitlab_import":
        from src.integrations.gitlab_user_client import GitLabUserClient

        source_repo = (source.get("repo_full_name") or "").strip()
        branch = (source.get("branch") or "main").strip() or "main"
        access_token = (source.get("access_token") or "").strip()
        if not source_repo or not access_token:
            raise RuntimeError("GitLab import source is missing project or token.")
        source_clone_url = GitLabUserClient(access_token=access_token).authenticated_clone_url(source_repo)
        with tempfile.TemporaryDirectory(prefix="p2hs-import-") as tmp:
            await clone_repo(source_clone_url, tmp, branch=branch)
            imported_sha = await git_rev_parse_head(tmp)
            root_dir = (source.get("root_dir") or "/").strip().replace("\\", "/")
            root_dir = "" if root_dir in ("", "/", ".") else root_dir.strip("/")
            source_root = os.path.join(tmp, *root_dir.split("/")) if root_dir else tmp
            if not os.path.isdir(source_root):
                raise RuntimeError(f"Source root directory not found: {root_dir or '/'}")
            detected_framework = _assert_import_supported(source_root)
            if source.get("import_env", True):
                imported_env = _collect_env_values(source_root)
            _copy_import_tree(source_root, repo_path)
            _write_repo_familiarity_doc(
                repo_path,
                source_repo=source_repo,
                source_branch=branch,
                source_root_dir=root_dir or "/",
                imported_sha=imported_sha,
            )
        version = f"gitlab-import:{source_repo[:28]}@{(imported_sha or branch)[:16]}"[:60]
    elif source_kind == "bitbucket_import":
        from src.integrations.bitbucket_user_client import BitbucketUserClient

        source_repo = (source.get("repo_full_name") or "").strip()
        branch = (source.get("branch") or "main").strip() or "main"
        access_token = (source.get("access_token") or "").strip()
        if not source_repo or not access_token:
            raise RuntimeError("Bitbucket import source is missing repo or token.")
        source_clone_url = BitbucketUserClient(
            access_token=access_token
        ).authenticated_clone_url(source_repo)
        with tempfile.TemporaryDirectory(prefix="p2hs-import-") as tmp:
            await clone_repo(source_clone_url, tmp, branch=branch)
            imported_sha = await git_rev_parse_head(tmp)
            root_dir = (source.get("root_dir") or "/").strip().replace("\\", "/")
            root_dir = "" if root_dir in ("", "/", ".") else root_dir.strip("/")
            source_root = os.path.join(tmp, *root_dir.split("/")) if root_dir else tmp
            if not os.path.isdir(source_root):
                raise RuntimeError(f"Source root directory not found: {root_dir or '/'}")
            detected_framework = _assert_import_supported(source_root)
            if source.get("import_env", True):
                imported_env = _collect_env_values(source_root)
            _copy_import_tree(source_root, repo_path)
            _write_repo_familiarity_doc(
                repo_path,
                source_repo=source_repo,
                source_branch=branch,
                source_root_dir=root_dir or "/",
                imported_sha=imported_sha,
            )
        version = f"bitbucket-import:{source_repo[:25]}@{(imported_sha or branch)[:16]}"[:60]
    elif source_kind == "azure_devops_import":
        from src.integrations.azure_devops_client import AzureDevOpsClient, split_azure_repo_full_name

        source_repo = (source.get("repo_full_name") or "").strip()
        branch = (source.get("branch") or "main").strip() or "main"
        org_url = (source.get("org_url") or "").strip()
        pat = (source.get("pat") or "").strip()
        access_token = (source.get("access_token") or "").strip()
        if not source_repo or not org_url or not (pat or access_token):
            raise RuntimeError("Azure DevOps import source is missing repo, org_url, or credentials.")
        project, repo_name = split_azure_repo_full_name(source_repo)
        source_clone_url = AzureDevOpsClient(
            org_url=org_url,
            pat=pat or None,
            access_token=access_token or None,
        ).authenticated_clone_url(project, repo_name)
        with tempfile.TemporaryDirectory(prefix="p2hs-import-azure-") as tmp:
            await clone_repo(source_clone_url, tmp, branch=branch)
            imported_sha = await git_rev_parse_head(tmp)
            root_dir = (source.get("root_dir") or "/").strip().replace("\\", "/")
            root_dir = "" if root_dir in ("", "/", ".") else root_dir.strip("/")
            source_root = os.path.join(tmp, *root_dir.split("/")) if root_dir else tmp
            if not os.path.isdir(source_root):
                raise RuntimeError(f"Source root directory not found: {root_dir or '/'}")
            detected_framework = _assert_import_supported(source_root)
            if source.get("import_env", True):
                imported_env = _collect_env_values(source_root)
            _copy_import_tree(source_root, repo_path)
            _write_repo_familiarity_doc(
                repo_path,
                source_repo=source_repo,
                source_branch=branch,
                source_root_dir=root_dir or "/",
                imported_sha=imported_sha,
            )
        version = f"azure-import:{source_repo[:28]}@{(imported_sha or branch)[:16]}"[:60]
    elif source_kind == "file_upload":
        from src.config import get_settings as _get_settings
        from src.integrations.r2_import_client import delete_object, download_object

        r2_key = (source.get("r2_key") or "").strip()
        if not r2_key:
            raise RuntimeError("File upload import source is missing r2_key.")
        _settings = _get_settings()
        with tempfile.TemporaryDirectory(prefix="p2hs-import-file-") as tmp:
            archive_path = os.path.join(tmp, os.path.basename(r2_key))
            await asyncio.to_thread(download_object, _settings, r2_key=r2_key, dest_path=archive_path)
            extract_dir = os.path.join(tmp, "extracted")
            os.makedirs(extract_dir, exist_ok=True)
            await asyncio.to_thread(_extract_archive, archive_path, extract_dir)
            source_root = _find_source_root_in_extracted(extract_dir)
            root_dir = (source.get("root_dir") or "/").strip().replace("\\", "/")
            root_dir = "" if root_dir in ("", "/", ".") else root_dir.strip("/")
            if root_dir:
                source_root = os.path.join(source_root, *root_dir.split("/"))
            if not os.path.isdir(source_root):
                raise RuntimeError(f"Source root directory not found in archive: {root_dir or '/'}")
            detected_framework = _assert_import_supported(source_root)
            if source.get("import_env", True):
                imported_env = _collect_env_values(source_root)
            _copy_import_tree(source_root, repo_path)
            archive_name = os.path.basename(r2_key)
            _write_repo_familiarity_doc(
                repo_path,
                source_repo=f"file-upload:{archive_name}",
                source_branch="upload",
                source_root_dir=root_dir or "/",
                imported_sha=None,
            )
        # Clean up R2 staging object after successful extraction
        try:
            await asyncio.to_thread(delete_object, _settings, r2_key=r2_key)
        except Exception:
            logger.warning("r2_import_cleanup_failed", r2_key=r2_key)
        version = f"file-upload:{os.path.basename(r2_key)[:40]}"[:60]
    elif source_kind == "s3_import":
        s3_url = (source.get("s3_url") or "").strip()
        access_key = (source.get("access_key") or "").strip()
        secret_key = (source.get("secret_key") or "").strip()
        endpoint_url = (source.get("endpoint_url") or "").strip() or None
        region = (source.get("region") or "us-east-1").strip() or "us-east-1"
        if not s3_url:
            raise RuntimeError("S3 import source is missing s3_url.")
        with tempfile.TemporaryDirectory(prefix="p2hs-import-s3-") as tmp:
            archive_name = s3_url.rstrip("/").split("/")[-1] or "archive"
            archive_path = os.path.join(tmp, archive_name)
            await asyncio.to_thread(
                _download_s3_object,
                s3_url=s3_url,
                dest_path=archive_path,
                access_key=access_key or None,
                secret_key=secret_key or None,
                endpoint_url=endpoint_url,
                region=region,
            )
            extract_dir = os.path.join(tmp, "extracted")
            os.makedirs(extract_dir, exist_ok=True)
            await asyncio.to_thread(_extract_archive, archive_path, extract_dir)
            source_root = _find_source_root_in_extracted(extract_dir)
            root_dir = (source.get("root_dir") or "/").strip().replace("\\", "/")
            root_dir = "" if root_dir in ("", "/", ".") else root_dir.strip("/")
            if root_dir:
                source_root = os.path.join(source_root, *root_dir.split("/"))
            if not os.path.isdir(source_root):
                raise RuntimeError(f"Source root directory not found in archive: {root_dir or '/'}")
            detected_framework = _assert_import_supported(source_root)
            if source.get("import_env", True):
                imported_env = _collect_env_values(source_root)
            _copy_import_tree(source_root, repo_path)
            _write_repo_familiarity_doc(
                repo_path,
                source_repo=f"s3:{s3_url[:60]}",
                source_branch="upload",
                source_root_dir=root_dir or "/",
                imported_sha=None,
            )
        version = f"s3-import:{archive_name[:40]}"[:60]
    elif template_repo:
        installation_id = int(
            source.get("github_installation_id")
            or settings.github_template_installation_id
            or settings.github_installation_id
            or 0
        )
        branch = (source.get("starter_branch") or "starter").strip() or "starter"
        template_client = GitHubAppClient(installation_id=installation_id)
        template_clone_url = await template_client.get_authenticated_clone_url(template_repo)
        await clone_repo(template_clone_url, repo_path, branch=branch)
        _sanitize_template_platform_runtime(repo_path)
        ref = (source.get("ref") or "template").strip()
        sha = (source.get("starter_sha") or "").strip()
        version = f"tpl:{ref[:36]}@{(sha or branch)[:16]}"[:60]
    else:
        version = _materialize_generated_source(
            repo_path, project_type, settings, source
        )

    _write_file(
        os.path.join(repo_path, ".p2hs.json"),
        json.dumps(
            {
                "type": project_type,
                "boilerplate_version": version,
                "project_ref": project_ref,
                "template_id": source.get("id"),
                "template_ref": source.get("ref"),
                "template_github_repo": template_repo or None,
                "template_starter_branch": source.get("starter_branch"),
                "template_starter_sha": source.get("starter_sha"),
                "source_kind": source.get("kind") or ("template" if template_repo else "generated"),
                "source_repo_full_name": source.get("repo_full_name"),
                "source_branch": source.get("branch"),
                "source_root_dir": source.get("root_dir"),
            },
            indent=2,
        )
        + "\n",
    )

    workflow_dir = os.path.join(repo_path, ".github", "workflows")
    os.makedirs(workflow_dir, exist_ok=True)
    from src.domain.hosting.deployment_targets import _CLOUDFLARE_WORKER_TYPES, generate_worker_deploy_workflow
    if project_type in _CLOUDFLARE_WORKER_TYPES:
        _write_file(
            os.path.join(workflow_dir, "deploy.yml"),
            generate_worker_deploy_workflow(project_ref),
        )
    else:
        from src.domain.deployments.repository import DeployRepository

        # Standalone materialization tests and bootstrap helpers may not have a
        # project row/session available yet. In that case there cannot be a
        # successful Pages deployment to preserve, so use the safe unassigned
        # W4P workflow. Production callers pass both values explicitly.
        has_successful_pages_deployment = False
        if db_factory is not None and project_id is not None:
            async with db_factory() as db:
                deploy_history = await DeployRepository(db).list_for_project(project_id)
            has_successful_pages_deployment = any(
                d.status == "live"
                and (d.hosting_provider or "").strip().lower() in {"cloudflare_pages", "cloudflare"}
                for d in deploy_history
            )
        _write_file(
            os.path.join(workflow_dir, "deploy.yml"),
            deploy_workflow_for_project(
                project_type,
                project_ref,
                project_created_at,
                settings,
                assume_unassigned_w4p=not has_successful_pages_deployment,
                has_successful_pages_deployment=has_successful_pages_deployment,
            ),
        )
    from src.domain.codegen.port_adapter import install_catalog_adapters
    from src.domain.hosting.migration_state import artifact_provider_id

    hosting_provider = None
    if db_factory is not None and project_id is not None:
        from src.domain.projects.models import Project

        async with db_factory() as db:
            row = await db.get(Project, project_id)
        if row is not None:
            hosting_provider = artifact_provider_id(
                getattr(row, "primary_hosting_provider", None),
                getattr(row, "hosting_migration", None),
            )
    install_catalog_adapters(
        repo_path,
        project_type,
        {
            "legacy_contact_submit": False,
            "hosting_provider": hosting_provider,
            "project_id": str(project_id) if project_id is not None else None,
        },
    )
    if not source_kind and not template_repo:
        await _materialize_generated_lockfile(repo_path)
    _append_gitignore_extras(repo_path)

    if template_repo and await git_rev_parse_head(repo_path):
        await _commit_materialized_template_overlay(
            repo_path,
            settings.git_committer_name,
            settings.git_committer_email,
        )
    return {"version": version, "imported_sha": imported_sha, "env": imported_env, "detected_framework": detected_framework}


def _extract_archive(archive_path: str, dest_dir: str) -> None:
    """Extract a zip / tar.gz / git bundle into dest_dir."""
    import subprocess
    import tarfile
    import zipfile

    name = os.path.basename(archive_path).lower()

    if name.endswith(".zip"):
        with zipfile.ZipFile(archive_path, "r") as z:
            z.extractall(dest_dir)
        return

    if name.endswith((".tar.gz", ".tgz", ".tar.bz2", ".tar.xz", ".tar")):
        with tarfile.open(archive_path, "r:*") as t:
            t.extractall(dest_dir)
        return

    if name.endswith(".bundle"):
        # git bundle: init a bare clone then export a working tree
        result = subprocess.run(
            ["git", "clone", archive_path, dest_dir],
            capture_output=True,
        )
        if result.returncode != 0:
            raise RuntimeError(
                f"git bundle extraction failed: {result.stderr.decode(errors='replace')}"
            )
        return

    raise RuntimeError(
        f"Unsupported archive format '{os.path.basename(archive_path)}'. "
        "Accepted formats: .zip, .tar.gz, .tgz, .bundle"
    )


def _find_source_root_in_extracted(dest_dir: str) -> str:
    """If the archive extracted into a single top-level directory, descend into it."""
    entries = [e for e in os.listdir(dest_dir) if not e.startswith(".")]
    if len(entries) == 1:
        candidate = os.path.join(dest_dir, entries[0])
        if os.path.isdir(candidate):
            return candidate
    return dest_dir


def _download_s3_object(
    *,
    s3_url: str,
    dest_path: str,
    access_key: str | None,
    secret_key: str | None,
    endpoint_url: str | None,
    region: str,
) -> None:
    """Download an S3 or S3-compatible object. Public URLs need no credentials."""
    import urllib.parse

    # Public URL — download without any credentials
    if not access_key or not secret_key:
        import httpx
        with httpx.Client(timeout=600, follow_redirects=True) as client:
            with client.stream("GET", s3_url) as r:
                r.raise_for_status()
                with open(dest_path, "wb") as fh:
                    for chunk in r.iter_bytes(chunk_size=8 * 1024 * 1024):
                        fh.write(chunk)
        return

    # Private — parse bucket + key from URL and use boto3
    import boto3

    parsed = urllib.parse.urlparse(s3_url)
    # Support both path-style and virtual-hosted-style S3 URLs
    if endpoint_url:
        # Path-style: https://endpoint/bucket/key
        path_parts = parsed.path.lstrip("/").split("/", 1)
        bucket = path_parts[0]
        key = path_parts[1] if len(path_parts) > 1 else ""
    else:
        # AWS virtual-hosted: https://bucket.s3.amazonaws.com/key  or path-style
        host = parsed.netloc.split(".")[0]
        if ".s3." in parsed.netloc or ".s3-" in parsed.netloc:
            bucket = host
            key = parsed.path.lstrip("/")
        else:
            path_parts = parsed.path.lstrip("/").split("/", 1)
            bucket = path_parts[0]
            key = path_parts[1] if len(path_parts) > 1 else ""

    s3 = boto3.client(
        "s3",
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        region_name=region,
        **({"endpoint_url": endpoint_url} if endpoint_url else {}),
    )
    s3.download_file(bucket, key, dest_path)


def _copy_import_tree(source_root: str, repo_path: str) -> None:
    if os.path.isdir(repo_path):
        shutil.rmtree(repo_path, ignore_errors=True)

    def _ignore(dirpath: str, names: list[str]) -> set[str]:
        skip_dirs = {".git", "node_modules", "dist", "build", ".next", ".nuxt", ".turbo", "__pycache__"}
        ignored: set[str] = set()
        for name in names:
            lower = name.lower()
            if name in skip_dirs:
                ignored.add(name)
            elif lower.startswith(".env") and lower not in {".env.example", ".env.sample"}:
                ignored.add(name)
        return ignored

    shutil.copytree(source_root, repo_path, ignore=_ignore)


_CF_PAGES_FRAMEWORK_SIGNATURES: list[tuple[str, str]] = [
    # (marker path, human label)
    ("package.json", "Node.js / JavaScript framework"),
    ("index.html", "Static HTML"),
    ("index.htm", "Static HTML"),
    ("public/index.html", "Static HTML"),
    ("src/index.html", "Static HTML"),
]

_CF_PAGES_UNSUPPORTED_MARKERS: list[tuple[str, str]] = [
    ("requirements.txt", "Python"),
    ("Pipfile", "Python"),
    ("pyproject.toml", "Python"),
    ("Gemfile", "Ruby"),
    ("go.mod", "Go"),
    ("Cargo.toml", "Rust"),
    ("pom.xml", "Java/Maven"),
    ("build.gradle", "Java/Gradle"),
    ("composer.json", "PHP"),
]


def detect_cf_framework(source_root: str) -> str:
    """Return a short label for the detected CF Pages compatible framework, or raise."""
    for marker, label in _CF_PAGES_UNSUPPORTED_MARKERS:
        if os.path.exists(os.path.join(source_root, marker)):
            raise RuntimeError(
                f"This repository appears to be a {label} project, which is not supported "
                "on Cloudflare Pages. Supported project types: Node.js frameworks (Nuxt, "
                "Next.js, React, Vue, SvelteKit, Astro, Remix, etc.) and static HTML sites."
            )
    for marker, label in _CF_PAGES_FRAMEWORK_SIGNATURES:
        if os.path.exists(os.path.join(source_root, marker)):
            if marker == "package.json":
                try:
                    with open(os.path.join(source_root, marker), encoding="utf-8") as fh:
                        pkg = json.load(fh)
                    deps = {
                        **(pkg.get("dependencies") or {}),
                        **(pkg.get("devDependencies") or {}),
                    }
                    if "nuxt" in deps:
                        return "nuxt"
                    if "next" in deps:
                        return "next"
                    if "@sveltejs/kit" in deps:
                        return "sveltekit"
                    if "svelte" in deps:
                        return "svelte"
                    if "@remix-run/react" in deps or "@remix-run/node" in deps:
                        return "remix"
                    if "astro" in deps:
                        return "astro"
                    if "gatsby" in deps:
                        return "gatsby"
                    if "react" in deps:
                        return "react"
                    if "vue" in deps:
                        return "vue"
                    if "@angular/core" in deps:
                        return "angular"
                    if "qwik" in deps or "@builder.io/qwik" in deps:
                        return "qwik"
                except Exception:
                    pass
                return "node"
            return "html"
    raise RuntimeError(
        "Could not detect a Cloudflare Pages supported framework in this repository. "
        "Ensure the repository contains a package.json (Node.js project) or an index.html "
        "(static site). Supported types: Nuxt, Next.js, React, Vue, SvelteKit, Astro, "
        "Remix, Gatsby, Angular, static HTML."
    )


def _assert_import_supported(source_root: str) -> str:
    """Validate the repo is importable and return the detected CF Pages framework label."""
    if os.path.exists(os.path.join(source_root, ".gitmodules")):
        raise RuntimeError("Git submodules are not supported for existing-project imports yet.")
    attrs = os.path.join(source_root, ".gitattributes")
    if os.path.isfile(attrs):
        try:
            with open(attrs, "r", encoding="utf-8") as fh:
                body = fh.read().lower()
            if "filter=lfs" in body:
                raise RuntimeError("Git LFS repositories are not supported for existing-project imports yet.")
        except UnicodeDecodeError:
            pass
    return detect_cf_framework(source_root)


def _collect_env_values(source_root: str) -> list[dict[str, str]]:
    env_files = []
    for name in (".env", ".env.local", ".env.production", ".env.preview", ".env.development"):
        p = os.path.join(source_root, name)
        if os.path.isfile(p):
            env_files.append((name, p))
    entries: list[dict[str, str]] = []
    seen: set[tuple[str, str]] = set()
    for env_name, path in env_files:
        environment = "all"
        if "production" in env_name:
            environment = "production"
        elif "preview" in env_name or "development" in env_name or "local" in env_name:
            environment = "preview"
        try:
            with open(path, "r", encoding="utf-8") as fh:
                for line in fh:
                    parsed = _parse_env_line(line)
                    if not parsed:
                        continue
                    key, value = parsed
                    marker = (key, environment)
                    if marker in seen:
                        continue
                    seen.add(marker)
                    entries.append(
                        {
                            "key": key,
                            "value": value,
                            "environment": environment,
                            "visibility": _infer_env_visibility(key),
                        }
                    )
        except UnicodeDecodeError:
            continue
    return entries


def _parse_env_line(line: str) -> Optional[tuple[str, str]]:
    raw = line.strip()
    if not raw or raw.startswith("#") or "=" not in raw:
        return None
    if raw.startswith("export "):
        raw = raw[len("export "):].strip()
    key, value = raw.split("=", 1)
    key = key.strip()
    if not key or not all(c.isalnum() or c == "_" for c in key):
        return None
    value = value.strip()
    if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
        value = value[1:-1]
    return key, value


def _infer_env_visibility(key: str) -> str:
    public_prefixes = ("VITE_", "NUXT_PUBLIC_", "NEXT_PUBLIC_", "PUBLIC_")
    return "public" if key.startswith(public_prefixes) else "secret"


def _write_repo_familiarity_doc(
    repo_path: str,
    *,
    source_repo: str,
    source_branch: str,
    source_root_dir: str,
    imported_sha: Optional[str],
) -> None:
    """Create a deterministic repo brief for AI agents before the first codegen run."""
    summary = _summarize_imported_repo(repo_path)
    doc = [
        "# P2HS Repo Familiarity",
        "",
        "This file is generated during existing-project import so P2HS agents can understand",
        "the project structure before making changes.",
        "",
        "## Import Source",
        f"- Repository: {source_repo}",
        f"- Branch: {source_branch}",
        f"- Source root: {source_root_dir or '/'}",
        f"- Imported commit: {imported_sha or 'unknown'}",
        "",
        "## Detected Structure",
    ]
    for line in summary:
        doc.append(f"- {line}")
    doc.extend(
        [
            "",
            "## Agent Guidance",
            "- Preserve existing backend/API integrations unless the user explicitly asks to change them.",
            "- Follow the framework, package manager, routing, and component patterns already present.",
            "- Use required-user-inputs.json only for truly unknown owner URLs, public IDs, or credentials; never for form choices such as countries, states, provinces, categories, or services.",
            "- Do not commit .env files or secrets.",
            "",
        ]
    )
    path = os.path.join(repo_path, "p2hs", "repo-familiarity.md")
    _write_file(path, "\n".join(doc))


def _summarize_imported_repo(repo_path: str) -> list[str]:
    markers = {
        "package.json": "Node/package project detected via package.json.",
        "nuxt.config.ts": "Nuxt configuration detected.",
        "nuxt.config.js": "Nuxt configuration detected.",
        "vite.config.ts": "Vite configuration detected.",
        "vite.config.js": "Vite configuration detected.",
        "next.config.js": "Next.js configuration detected.",
        "next.config.mjs": "Next.js configuration detected.",
        "src": "src/ directory present.",
        "app": "app/ directory present.",
        "pages": "pages/ directory present.",
        "components": "components/ directory present.",
        "public": "public/ assets directory present.",
    }
    out: list[str] = []
    for rel, message in markers.items():
        if os.path.exists(os.path.join(repo_path, rel)):
            out.append(message)
    if os.path.isfile(os.path.join(repo_path, "package.json")):
        try:
            with open(os.path.join(repo_path, "package.json"), encoding="utf-8") as fh:
                pkg = json.load(fh)
            scripts = sorted((pkg.get("scripts") or {}).keys())
            deps = sorted(
                set((pkg.get("dependencies") or {}).keys())
                | set((pkg.get("devDependencies") or {}).keys())
            )
            if scripts:
                out.append("Package scripts: " + ", ".join(scripts[:12]))
            if deps:
                out.append("Notable dependencies: " + ", ".join(deps[:20]))
        except Exception:
            out.append("package.json exists but could not be parsed.")
    return out or ["No common framework markers detected; inspect files before editing."]


async def _auto_connect_import_mirror(db_factory, project_id: UUID) -> None:
    """After import setup completes, upsert the source repo as the project's mirror target.

    Non-fatal — a failure here does not affect the already-completed project setup.
    """
    try:
        from src.domain.projects.repository import ProjectRepository

        async with db_factory() as db:
            project = await ProjectRepository(db).get_by_id(project_id)
            if not project:
                return
            source_kind = getattr(project, "source_kind", None) or ""
            source_repo = (getattr(project, "source_repo_full_name", None) or "").strip()
            source_branch = (getattr(project, "source_branch", None) or "main").strip() or "main"

            if not source_repo:
                return

            if source_kind == "github_import":
                installation_id = getattr(project, "source_workspace_user_github_installation_id", None)
                if not installation_id:
                    return
                from src.domain.github_user_mirror.repository import ProjectUserMirrorRepository
                existing = await ProjectUserMirrorRepository(db).get_by_project_id(project_id)
                if not existing:
                    await ProjectUserMirrorRepository(db).upsert(
                        project_id=project_id,
                        workspace_user_github_installation_id=installation_id,
                        target_repo_full_name=source_repo,
                        target_default_branch=source_branch,
                        enabled=True,
                    )
                    await db.commit()
                    logger.info("import_mirror_auto_connected", project_id=str(project_id), provider="github", repo=source_repo)

            elif source_kind == "gitlab_import":
                connection_id = getattr(project, "source_workspace_user_gitlab_connection_id", None)
                if not connection_id:
                    return
                from src.domain.gitlab_user_mirror.repository import ProjectUserGitlabMirrorRepository
                existing = await ProjectUserGitlabMirrorRepository(db).get_by_project_id(project_id)
                if not existing:
                    await ProjectUserGitlabMirrorRepository(db).upsert(
                        project_id=project_id,
                        workspace_user_gitlab_connection_id=connection_id,
                        target_project_full_path=source_repo,
                        target_default_branch=source_branch,
                        enabled=True,
                    )
                    await db.commit()
                    logger.info("import_mirror_auto_connected", project_id=str(project_id), provider="gitlab", repo=source_repo)

            elif source_kind == "bitbucket_import":
                connection_id = getattr(project, "source_workspace_user_bitbucket_connection_id", None)
                if not connection_id:
                    return
                from src.domain.bitbucket_user_mirror.repository import ProjectUserBitbucketMirrorRepository
                existing = await ProjectUserBitbucketMirrorRepository(db).get_by_project_id(project_id)
                if not existing:
                    await ProjectUserBitbucketMirrorRepository(db).upsert(
                        project_id=project_id,
                        workspace_user_bitbucket_connection_id=connection_id,
                        target_repo_full_name=source_repo,
                        target_default_branch=source_branch,
                        enabled=True,
                    )
                    await db.commit()
                    logger.info("import_mirror_auto_connected", project_id=str(project_id), provider="bitbucket", repo=source_repo)

            elif source_kind == "azure_devops_import":
                connection_id = getattr(project, "source_workspace_user_azure_connection_id", None)
                if not connection_id:
                    return
                from src.domain.azure_user_mirror.repository import ProjectUserAzureMirrorRepository
                existing = await ProjectUserAzureMirrorRepository(db).get_by_project_id(project_id)
                if not existing:
                    await ProjectUserAzureMirrorRepository(db).upsert(
                        project_id=project_id,
                        workspace_user_azure_connection_id=connection_id,
                        target_repo_full_name=source_repo,
                        target_default_branch=source_branch,
                        enabled=True,
                    )
                    await db.commit()
                    logger.info("import_mirror_auto_connected", project_id=str(project_id), provider="azure_devops", repo=source_repo)

    except Exception as exc:
        logger.warning("import_mirror_auto_connect_failed", project_id=str(project_id), error=str(exc))


async def _enqueue_initial_hosting(
    db_factory,
    project_id: UUID,
    revision_id: UUID,
    github_repo_full_name: str,
    environment: str,
) -> None:
    """After scaffold push: create the initial deploy for one environment."""
    settings = get_settings()
    if not settings.hosting_bootstrap_on_project_setup:
        logger.info("hosting_bootstrap_skipped", project_id=str(project_id), reason="disabled_in_settings")
        return

    new_deploy_id: Optional[UUID] = None
    async with db_factory() as db:
        from src.domain.deployments.repository import DeployRepository

        deploy_repo = DeployRepository(db)
        for d in await deploy_repo.list_for_project(project_id):
            meta = d.deployment_metadata or {}
            if meta.get("bootstrap_initial_hosting") and d.environment == environment:
                logger.info(
                    "hosting_bootstrap_skipped",
                    project_id=str(project_id),
                    environment=environment,
                    reason="already_bootstrapped",
                )
                await db.commit()
                return

        d = await deploy_repo.create(
            project_id,
            revision_id,
            environment,
            created_by=None,
            deployment_metadata={
                "bootstrap_initial_hosting": True,
                "github_repo_full_name": github_repo_full_name,
            },
        )
        new_deploy_id = d.id
        await db.commit()

    if new_deploy_id is not None:
        from src.workers.enqueue import enqueue_run_project_deploy

        ok = await enqueue_run_project_deploy(str(new_deploy_id))
        logger.info(
            "hosting_bootstrap_enqueued",
            project_id=str(project_id),
            deploy_id=str(new_deploy_id),
            environment=environment,
            repo=github_repo_full_name,
            arq_enqueued=ok,
        )


async def enqueue_initial_production_hosting(
    db_factory,
    project_id: UUID,
    revision_id: UUID,
    github_repo_full_name: str,
) -> None:
    await _enqueue_initial_hosting(
        db_factory, project_id, revision_id, github_repo_full_name, "production"
    )


async def enqueue_initial_preview_hosting(
    db_factory,
    project_id: UUID,
    revision_id: UUID,
    github_repo_full_name: str,
) -> None:
    await _enqueue_initial_hosting(
        db_factory, project_id, revision_id, github_repo_full_name, "preview"
    )


async def run_project_setup(ctx: dict, project_id: UUID) -> dict:
    from src.core.database import async_session_factory
    from src.domain.builder.repository import TimelineEventRepository
    from src.domain.instructions.repository import InstructionRepository, PromptSessionRepository
    from src.domain.previews.repository import RoutesManifestRepository
    from src.domain.projects.job_outbox_constants import JOB_KIND_CLOUD_SYNC_POST_READY
    from src.domain.projects.repository import (
        ProjectJobOutboxRepository,
        ProjectRepository,
        ProjectRepoRepository,
    )
    from src.domain.projects.setup_step_constants import (
        SETUP_STEP_AUTH_PROVISION,
        SETUP_STEP_AUTHZ_V2,
        SETUP_STEP_CF_PAGES_ENSURE,
        SETUP_STEP_CF_REPO_SECRETS,
        SETUP_STEP_DB_SEED,
        SETUP_STEP_DEPLOY_ENQUEUE,
        SETUP_STEP_GIT_PUSH_MAIN,
        SETUP_STEP_GIT_PUSH_PREVIEW,
        SETUP_STEP_STATUS_ERROR,
        SETUP_STEP_STATUS_OK,
        SETUP_STEP_STATUS_SKIPPED,
    )
    from src.domain.revisions.repository import RevisionRepository
    from src.integrations.github_client import GitHubAppClient
    from src.integrations.github_installation import resolve_installation_id_for_workspace
    from src.domain.instructions.disk_mirror import mirror_instruction_to_meta
    from src.workers.repo_ops import (
        create_or_reset_preview_branch,
        ensure_git_remote_push,
        get_project_meta_path,
        get_project_repo_path,
        git_branch_move_to,
        git_init_and_commit,
        git_rev_parse_head,
        push_branch,
    )

    db_factory = ctx.get("db_factory") or async_session_factory
    settings = get_settings()
    org = (settings.github_org or "").strip() or None

    # AI-assist WordPress projects use a dedicated GitHub provisioning pipeline
    # (companion-release + site-understanding branches) instead of the Cloudflare
    # Pages boilerplate/deploy flow below.
    from src.domain.projects.models import AI_ASSIST_WORDPRESS_PROJECT_TYPE

    async with db_factory() as _probe_db:
        _probe = await ProjectRepository(_probe_db).get_by_id(project_id)
    if _probe is not None and _probe.type == AI_ASSIST_WORDPRESS_PROJECT_TYPE:
        from src.workers.wordpress_project_setup import run_wordpress_project_setup

        return await run_wordpress_project_setup(ctx, project_id)

    async def _mark_error() -> None:
        async with db_factory() as db:
            await ProjectRepository(db).update(project_id, status="error")
            await db.commit()

    try:
        # --- Load project ---------------------------------------------------
        async with db_factory() as db:
            project = await ProjectRepository(db).get_by_id(project_id)
            if not project:
                raise RuntimeError("Project not found")
            if project.deleted_at is not None:
                raise RuntimeError("Project deleted")

            pr_repo = ProjectRepoRepository(db)
            pr_row = await pr_repo.get_by_project_id(project_id)

            # Resume guard: legacy two-phase row — boilerplate task still needed.
            if pr_row and pr_row.setup_stage == SetupStage.SKELETON_PUSHED:
                await db.commit()
                from src.workers.enqueue import enqueue_apply_boilerplate
                ok_boil = await enqueue_apply_boilerplate(str(project_id))
                if not ok_boil:
                    logger.error(
                        "apply_boilerplate_enqueue_failed",
                        project_id=str(project_id),
                        reason="arq_enqueue_skipped_or_redis",
                        hint="Legacy skeleton_pushed row: Phase 2 may never run until recovery or manual re-enqueue.",
                    )
                return {"status": "ok", "skipped": True, "reason": "skeleton_already_pushed"}

            # Both phases done — nothing to do.
            if pr_row and pr_row.setup_stage == SetupStage.COMPLETED:
                if project.status != "ready":
                    await ProjectRepository(db).update(project_id, status="ready")
                from src.domain.projects.discovery_service import enqueue_approved_discovery_after_setup

                await enqueue_approved_discovery_after_setup(db, project_id)
                await db.commit()
                return {"status": "ok", "skipped": True, "reason": "already_completed"}

            if not project.github_repo_name:
                project.github_repo_name = f"p2hs-{project.id.hex}"[:100]
                await db.flush()

            github_repo_name = project.github_repo_name
            project_type = project.type
            workspace_id = project.workspace_id
            project_created_at = project.created_at
            project_ref = (project.project_ref or "").strip()
            template_source = None
            if getattr(project, "source_kind", None) == "github_import":
                if not project.source_workspace_user_github_installation_id or not project.source_repo_full_name:
                    raise RuntimeError("Imported project is missing GitHub source metadata")
                from src.domain.github_user_mirror.repository import WorkspaceUserGithubInstallationRepository

                inst = await WorkspaceUserGithubInstallationRepository(db).get_by_id(
                    project.source_workspace_user_github_installation_id
                )
                if not inst or inst.workspace_id != workspace_id or inst.suspended_at is not None:
                    raise RuntimeError("Imported project source GitHub installation is invalid or suspended")
                template_source = {
                    "kind": "github_import",
                    "github_installation_id": inst.github_installation_id,
                    "repo_full_name": project.source_repo_full_name,
                    "branch": project.source_branch or "main",
                    "root_dir": project.source_root_dir or "/",
                    "import_env": bool(getattr(project, "source_import_env", True)),
                }
            elif getattr(project, "source_kind", None) == "gitlab_import":
                if not project.source_workspace_user_gitlab_connection_id or not project.source_repo_full_name:
                    raise RuntimeError("Imported project is missing GitLab source metadata")
                from src.core.security.encryption import decrypt_value
                from src.domain.gitlab_user_mirror.repository import WorkspaceUserGitlabConnectionRepository

                conn = await WorkspaceUserGitlabConnectionRepository(db).get_by_id(
                    project.source_workspace_user_gitlab_connection_id
                )
                if not conn or conn.workspace_id != workspace_id or conn.revoked_at is not None:
                    raise RuntimeError("Imported project source GitLab connection is invalid or revoked")
                template_source = {
                    "kind": "gitlab_import",
                    "access_token": decrypt_value(conn.access_token_ciphertext),
                    "repo_full_name": project.source_repo_full_name,
                    "branch": project.source_branch or "main",
                    "root_dir": project.source_root_dir or "/",
                    "import_env": bool(getattr(project, "source_import_env", True)),
                }
            elif getattr(project, "source_kind", None) == "bitbucket_import":
                if not project.source_workspace_user_bitbucket_connection_id or not project.source_repo_full_name:
                    raise RuntimeError("Imported project is missing Bitbucket source metadata")
                from src.core.security.encryption import decrypt_value
                from src.domain.bitbucket_user_mirror.repository import WorkspaceUserBitbucketConnectionRepository

                conn = await WorkspaceUserBitbucketConnectionRepository(db).get_by_id(
                    project.source_workspace_user_bitbucket_connection_id
                )
                if not conn or conn.workspace_id != workspace_id or conn.revoked_at is not None:
                    raise RuntimeError("Imported project source Bitbucket connection is invalid or revoked")
                template_source = {
                    "kind": "bitbucket_import",
                    "access_token": decrypt_value(conn.access_token_ciphertext),
                    "repo_full_name": project.source_repo_full_name,
                    "branch": project.source_branch or "main",
                    "root_dir": project.source_root_dir or "/",
                    "import_env": bool(getattr(project, "source_import_env", True)),
                }
            elif getattr(project, "source_kind", None) == "azure_devops_import":
                if not project.source_workspace_user_azure_connection_id or not project.source_repo_full_name:
                    raise RuntimeError("Imported project is missing Azure DevOps source metadata")
                from src.domain.azure_user_mirror.repository import WorkspaceUserAzureConnectionRepository
                from src.domain.azure_user_mirror.service import resolve_azure_connection_client

                conn_repo = WorkspaceUserAzureConnectionRepository(db)
                conn = await conn_repo.get_by_id(
                    project.source_workspace_user_azure_connection_id
                )
                if not conn or conn.workspace_id != workspace_id or conn.revoked_at is not None:
                    raise RuntimeError("Imported project source Azure DevOps connection is invalid or revoked")
                client = await resolve_azure_connection_client(conn, conn_repo)
                template_source = {
                    "kind": "azure_devops_import",
                    "org_url": conn.azure_org_url,
                    "pat": client._pat or "",
                    "access_token": client._access_token or "",
                    "repo_full_name": project.source_repo_full_name,
                    "branch": project.source_branch or "main",
                    "root_dir": project.source_root_dir or "/",
                    "import_env": bool(getattr(project, "source_import_env", True)),
                }
            elif getattr(project, "source_kind", None) == "file_upload":
                r2_key = getattr(project, "source_r2_key", None) or ""
                if not r2_key:
                    raise RuntimeError("File upload project is missing source_r2_key")
                template_source = {
                    "kind": "file_upload",
                    "r2_key": r2_key,
                    "root_dir": project.source_root_dir or "/",
                    "import_env": bool(getattr(project, "source_import_env", True)),
                }
            elif getattr(project, "source_kind", None) == "s3_import":
                s3_url = getattr(project, "source_s3_object_url", None) or ""
                if not s3_url:
                    raise RuntimeError("S3 import project is missing source_s3_object_url")
                from src.core.security.encryption import decrypt_value
                raw_key = getattr(project, "source_s3_access_key_enc", None)
                raw_secret = getattr(project, "source_s3_secret_key_enc", None)
                template_source = {
                    "kind": "s3_import",
                    "s3_url": s3_url,
                    "access_key": decrypt_value(raw_key) if raw_key else "",
                    "secret_key": decrypt_value(raw_secret) if raw_secret else "",
                    "endpoint_url": getattr(project, "source_s3_endpoint_url", None) or None,
                    "region": getattr(project, "source_s3_region", None) or "us-east-1",
                    "root_dir": project.source_root_dir or "/",
                    "import_env": bool(getattr(project, "source_import_env", True)),
                }
            elif project.template_github_repo:
                template_source = {
                    "id": str(project.template_id) if project.template_id else None,
                    "ref": project.template_ref,
                    "github_repo": project.template_github_repo,
                    "starter_branch": project.template_starter_branch or "starter",
                    "starter_sha": project.template_starter_sha,
                    "github_installation_id": 0,
                }
                if project.template_id:
                    from src.domain.templates.repository import TemplateRepository

                    template = await TemplateRepository(db).get_by_id(project.template_id)
                    if template:
                        template_source["github_installation_id"] = template.github_installation_id
                if not template_source["github_installation_id"]:
                    # A template without an explicit installation belongs to the
                    # same workspace trust boundary as the project. Reuse the
                    # workspace installation rather than a global template App
                    # that may not have access to this private repository.
                    from src.integrations.github_installation import resolve_installation_id_for_workspace

                    template_source["github_installation_id"] = await resolve_installation_id_for_workspace(
                        db_factory, workspace_id
                    )
            await db.commit()

        repo_path = get_project_repo_path(settings.repo_workspace, workspace_id, project_id)
        meta_path = get_project_meta_path(settings.repo_workspace, workspace_id, project_id)
        os.makedirs(meta_path, exist_ok=True)

        boilerplate_version_label: Optional[str] = None
        source_imported_sha: Optional[str] = None
        imported_env_entries: list[dict[str, str]] = []
        detected_framework_label: Optional[str] = None
        # --- Materialize boilerplate (snapshot extract or live copy; no npm on worker) ----
        # Must happen before the parallel gather because git_init_and_commit
        # expects files to already be in place.
        if not await git_rev_parse_head(repo_path):
            materialized = await _materialize_initial_repo_tree(
                repo_path,
                project_type,
                project_ref,
                settings,
                template_source=template_source,
                project_created_at=project_created_at,
                db_factory=db_factory,
                project_id=project_id,
            )
            boilerplate_version_label = materialized.get("version")
            source_imported_sha = materialized.get("imported_sha")
            imported_env_entries = materialized.get("env") or []
            detected_framework_label = materialized.get("detected_framework")

        if imported_env_entries:
            from src.core.security.encryption import encrypt_value
            from src.domain.projects.repository import ProjectEnvVarRepository

            async with db_factory() as db:
                env_repo = ProjectEnvVarRepository(db)
                for entry in imported_env_entries:
                    await env_repo.upsert_by_key(
                        project_id,
                        entry["key"],
                        encrypt_value(entry["value"]),
                        environment=entry.get("environment", "all"),
                        visibility=entry.get("visibility", "secret"),
                    )
                await db.commit()

        async with db_factory() as db:
            await ProjectRepoRepository(db).update_setup_state(
                project_id, setup_stage=SetupStage.LOCAL_WORKSPACE
            )
            if detected_framework_label:
                from src.domain.projects.setup_step_constants import SETUP_STEP_STATUS_OK
                await ProjectRepository(db).merge_setup_step(
                    project_id,
                    "source_framework_detect",
                    {"status": SETUP_STEP_STATUS_OK, "framework": detected_framework_label},
                )
            await db.commit()

        # --- Parallel: git init+commit  AND  GitHub repo creation -----------
        installation_id = await resolve_installation_id_for_workspace(db_factory, workspace_id)
        github = GitHubAppClient(installation_id=installation_id)

        local_sha, repo_info = await asyncio.gather(
            git_init_and_commit(
                repo_path,
                f"feat: initial {project_type} boilerplate",
                settings.git_committer_name,
                settings.git_committer_email,
            ),
            github.ensure_repository(github_repo_name, private=True, org=org),
        )

        branch = (repo_info.default_branch or "main").strip() or "main"
        await git_branch_move_to(repo_path, branch)

        # --- DB: store GitHub repo row --------------------------------------
        async with db_factory() as db:
            pr_repo = ProjectRepoRepository(db)
            existing = await pr_repo.get_by_project_id(project_id)
            if not existing:
                await pr_repo.create(project_id, repo_info.full_name, repo_info.id)
            await pr_repo.update_remote_metadata(
                project_id,
                repo_full_name=repo_info.full_name,
                github_repo_id=repo_info.id,
                clone_url=repo_info.clone_url,
                html_url=getattr(repo_info, "html_url", None),
                default_branch=branch,
            )
            await pr_repo.update_setup_state(project_id, setup_stage=SetupStage.GITHUB_REPO)
            await db.commit()

        repo_full = repo_info.full_name

        # --- Set CF secrets on repo (non-fatal) ----------------------------
        # Prefer per-workspace CF credentials; fall back to settings env vars.
        cf_token = (settings.cloudflare_api_token or "").strip()
        cf_account = (settings.cloudflare_account_id or "").strip()
        async with db_factory() as db:
            from src.domain.cloud_infra.repository import (
                CFAccountRepository,
                CloudInstanceRepository,
                WorkspaceCloudAssignmentRepository,
            )
            from src.domain.cloud_infra.service import CloudInfraService
            from src.domain.hosting.deploy_target import DeployAccountError, settings_for_project

            async def _workspace_credentials(ws_id):
                svc = CloudInfraService(
                    cf_account_repo=CFAccountRepository(db),
                    instance_repo=CloudInstanceRepository(db),
                    assignment_repo=WorkspaceCloudAssignmentRepository(db),
                )
                return await svc.get_credentials_for_workspace(ws_id)

            # Same account rule as the deploy. If it cannot be resolved, leave the credentials empty
            # so the secrets and Pages steps are skipped; never fall back to the original account.
            try:
                account_settings = await settings_for_project(
                    workspace_id, project_created_at, settings, _workspace_credentials
                )
                cf_token = (account_settings.cloudflare_api_token or "").strip()
                cf_account = (account_settings.cloudflare_account_id or "").strip()
            except DeployAccountError as exc:
                logger.warning("setup_cf_account_unresolved", project_id=str(project_id), error=str(exc))
                cf_token = ""
                cf_account = ""

        if cf_token and cf_account:
            try:
                await github.set_repo_action_secrets(
                    repo_full,
                    {"CLOUDFLARE_API_TOKEN": cf_token, "CLOUDFLARE_ACCOUNT_ID": cf_account},
                )
                logger.info("repo_cf_secrets_set", project_id=str(project_id), repo=repo_full)
                async with db_factory() as db:
                    await ProjectRepository(db).merge_setup_step(
                        project_id,
                        SETUP_STEP_CF_REPO_SECRETS,
                        {"status": SETUP_STEP_STATUS_OK},
                    )
                    await db.commit()
            except Exception as exc:
                logger.warning(
                    "repo_cf_secrets_skipped",
                    project_id=str(project_id),
                    repo=repo_full,
                    error=str(exc),
                    hint="Add 'Secrets: Read and write' permission to the GitHub App",
                )
                async with db_factory() as db:
                    await ProjectRepository(db).merge_setup_step(
                        project_id,
                        SETUP_STEP_CF_REPO_SECRETS,
                        {"status": SETUP_STEP_STATUS_ERROR, "last_error": str(exc)[:800]},
                    )
                    await db.commit()
        else:
            async with db_factory() as db:
                await ProjectRepository(db).merge_setup_step(
                    project_id,
                    SETUP_STEP_CF_REPO_SECRETS,
                    {"status": SETUP_STEP_STATUS_SKIPPED, "detail": "missing_cf_credentials"},
                )
                await db.commit()

        # --- Ensure CF Pages project exists BEFORE first push (avoid GA vs worker race) ---
        # Pushing deploy.yml triggers GitHub Actions immediately; wrangler fails with
        # "project not found" if the Pages project is only created later by run_project_deploy.
        # shopify_app uses CF Workers (not Pages) — no Pages project needed here.
        from src.domain.hosting.deployment_targets import decide_target_kind
        from src.domain.hosting.project_provider import W4P_STATIC, project_hosting_provider
        hosting_key = (settings.hosting_provider or "stub").strip().lower()
        from src.domain.deployments.repository import DeployRepository

        async with db_factory() as db:
            _history = await DeployRepository(db).list_for_project(project_id)
        _has_pages = any(
            d.status == "live"
            and (d.hosting_provider or "").strip().lower() in {"cloudflare_pages", "cloudflare"}
            for d in _history
        )
        is_w4p_project = (
            decide_target_kind(project_type) == "cloudflare_pages"
            and project_hosting_provider(
                project_created_at,
                settings,
                assume_unassigned_w4p=not _has_pages,
                has_successful_pages_deployment=_has_pages,
            ) == W4P_STATIC
        )
        if project_type == "shopify_app":
            async with db_factory() as db:
                await ProjectRepository(db).merge_setup_step(
                    project_id,
                    SETUP_STEP_CF_PAGES_ENSURE,
                    {"status": SETUP_STEP_STATUS_SKIPPED, "detail": "shopify_app_uses_cf_worker"},
                )
                await db.commit()
        elif is_w4p_project:
            async with db_factory() as db:
                await ProjectRepository(db).merge_setup_step(
                    project_id,
                    SETUP_STEP_CF_PAGES_ENSURE,
                    {"status": SETUP_STEP_STATUS_SKIPPED, "detail": "w4p_static_no_pages_project"},
                )
                await db.commit()
        elif hosting_key in ("cloudflare_pages", "cloudflare") and project_ref and cf_token and cf_account:
            from src.domain.hosting.adapters.cloudflare_pages import CloudflarePagesProvider

            cf_settings = settings.model_copy(
                update={
                    "cloudflare_api_token": cf_token,
                    "cloudflare_account_id": cf_account,
                }
            )
            try:
                await CloudflarePagesProvider(cf_settings).ensure_pages_project(
                    project_ref,
                    project_type,
                )
                logger.info(
                    "cf_pages_project_ready_before_push",
                    project_id=str(project_id),
                    project_ref=project_ref,
                )
                async with db_factory() as db:
                    await ProjectRepository(db).merge_setup_step(
                        project_id,
                        SETUP_STEP_CF_PAGES_ENSURE,
                        {"status": SETUP_STEP_STATUS_OK},
                    )
                    await db.commit()
            except Exception as exc:
                async with db_factory() as db:
                    await ProjectRepository(db).merge_setup_step(
                        project_id,
                        SETUP_STEP_CF_PAGES_ENSURE,
                        {"status": SETUP_STEP_STATUS_ERROR, "last_error": str(exc)[:800]},
                    )
                    await TimelineEventRepository(db).create(
                        project_id,
                        "project_setup_step_failed",
                        phase="cf_pages_ensure",
                        title="Cloudflare Pages ensure failed",
                        detail_text=str(exc)[:2000],
                    )
                    await db.commit()
                logger.error(
                    "cf_pages_ensure_before_push_failed",
                    project_id=str(project_id),
                    project_ref=project_ref,
                    error=str(exc),
                )
                raise RuntimeError(
                    "Could not create or verify the Cloudflare Pages project before pushing to GitHub. "
                    "Check Cloudflare API token permissions (Account · Cloudflare Pages · Edit) and "
                    "that CLOUDFLARE_ACCOUNT_ID matches the token."
                ) from exc
        elif hosting_key in ("cloudflare_pages", "cloudflare") and project_ref:
            logger.warning(
                "cf_pages_skip_ensure_before_push",
                project_id=str(project_id),
                reason="missing_cloudflare_credentials",
                hint="First GitHub Actions deploy may fail until a Pages project exists.",
            )
            async with db_factory() as db:
                await ProjectRepository(db).merge_setup_step(
                    project_id,
                    SETUP_STEP_CF_PAGES_ENSURE,
                    {"status": SETUP_STEP_STATUS_SKIPPED, "detail": "missing_cloudflare_credentials"},
                )
                await db.commit()

        # --- Push to GitHub ----------------------------------------
        auth_url = await github.get_authenticated_clone_url(repo_full)
        await ensure_git_remote_push(repo_path, auth_url, branch)

        sha = await github.get_ref_head_sha(repo_full, branch) or local_sha

        async with db_factory() as db:
            await ProjectRepository(db).merge_setup_step(
                project_id,
                SETUP_STEP_GIT_PUSH_MAIN,
                {"status": SETUP_STEP_STATUS_OK},
            )
            await db.commit()

        # --- DB seed: revision, instruction, session, manifest --------------
        async with db_factory() as db:
            revision_repo = RevisionRepository(db)
            rev = await revision_repo.get_by_git_sha(project_id, sha)
            if not rev:
                rev = await revision_repo.create(
                    project_id=project_id,
                    git_sha=sha,
                    source="project_setup",
                    commit_message="Initial boilerplate",
                )

            instruction_repo = InstructionRepository(db)
            if not await instruction_repo.has_any_version(project_id):
                instruction_file = os.path.join(meta_path, "p2hs", "instructions.md")
                content = ""
                if os.path.isfile(instruction_file):
                    with open(instruction_file, "r", encoding="utf-8") as f:
                        content = f.read()
                v = await instruction_repo.create_version(project_id, content)
                await instruction_repo.set_current(project_id, v.id)
                mirror_instruction_to_meta(workspace_id, project_id, v.content)

            session_repo = PromptSessionRepository(db)
            project_db = ProjectRepository(db)
            project = await project_db.get_by_id(project_id)
            session = None
            if project and project.active_session_id:
                session = await session_repo.get_by_id(project.active_session_id)
            if not session:
                sessions = await session_repo.list_for_project(project_id)
                session = sessions[-1] if sessions else await session_repo.create(project_id, None, "New session")
            await project_db.update(project_id, active_session_id=session.id)

            manifest_repo = RoutesManifestRepository(db)
            if not await manifest_repo.get_latest(project_id, rev.id):
                # Default routes manifest — adapter can refine after first codegen
                await manifest_repo.create(
                    project_id=project_id,
                    manifest={"routes": [{"path": "/", "label": "Home", "type": "page"}], "source": "default"},
                    source="default",
                    revision_id=rev.id,
                )

            await ProjectRepoRepository(db).update_setup_state(
                project_id,
                setup_stage=SetupStage.DB_SEEDED,
                initial_commit_sha=sha,
            )
            await project_db.update(project_id, active_revision_id=rev.id)
            await project_db.merge_setup_step(project_id, SETUP_STEP_DB_SEED, {"status": SETUP_STEP_STATUS_OK})
            await db.commit()

        # --- Auto-provision default auth (idempotent, non-fatal) ------------
        async with db_factory() as db:
            from src.domain.project_auth.repository import (
                ProjectAuthConfigRepository,
                ProjectAuthKeyRepository,
                ProjectAuthRoleRepository,
            )
            from src.domain.project_auth.service import ProjectAuthService
            from src.integrations.cloud_control_client import (
                CloudControlClient,
                get_cloud_client_for_workspace,
            )

            proj_main = ProjectRepository(db)
            try:
                try:
                    cloud_auth_client = await get_cloud_client_for_workspace(workspace_id, db)
                except Exception:
                    cloud_auth_client = CloudControlClient()
                auth_svc = ProjectAuthService(
                    config_repo=ProjectAuthConfigRepository(db),
                    key_repo=ProjectAuthKeyRepository(db),
                    role_repo=ProjectAuthRoleRepository(db),
                    cloud_client=cloud_auth_client,
                )
                await auth_svc.ensure_default_auth(project_id)
                await proj_main.merge_setup_step(
                    project_id, SETUP_STEP_AUTH_PROVISION, {"status": SETUP_STEP_STATUS_OK}
                )
            except Exception as exc:
                logger.warning(
                    "project_setup_auth_provision_failed",
                    project_id=str(project_id),
                    error=str(exc),
                )
                await proj_main.merge_setup_step(
                    project_id,
                    SETUP_STEP_AUTH_PROVISION,
                    {"status": SETUP_STEP_STATUS_ERROR, "last_error": str(exc)[:800]},
                )
            await db.commit()

        # --- Authorization v2 provisioning: backfill, drift, shadow (flag-gated, non-fatal) ---
        if settings.authz_v2_auto_provision_enabled:
            from src.domain.project_auth.provisioning_v2 import provision_authz_v2

            async with db_factory() as db:
                try:
                    outcome = await provision_authz_v2(
                        db,
                        project_id,
                        cloud_base_url=settings.cloud_control_api_url,
                        cloud_token=settings.cloud_internal_token,
                    )
                    step = {
                        "status": outcome.status,
                        "projection_state": outcome.projection_state,
                        "reason": outcome.reason,
                    }
                except Exception as exc:
                    await db.rollback()
                    logger.warning(
                        "project_setup_authz_v2_failed",
                        project_id=str(project_id),
                        error=str(exc),
                    )
                    step = {"status": SETUP_STEP_STATUS_ERROR, "last_error": str(exc)[:800]}
                await ProjectRepository(db).merge_setup_step(project_id, SETUP_STEP_AUTHZ_V2, step)
                await db.commit()

        # --- Create preview branch + push (skip for CF Workers — one deployment only) ---
        if project_type != "shopify_app":
            preview_branch = settings.hosting_preview_branch or "preview"
            await create_or_reset_preview_branch(repo_path, preview_branch, from_branch=branch)
            await ensure_git_remote_push(repo_path, auth_url, branch)
            await push_branch(repo_path, preview_branch, force=True)

            async with db_factory() as db:
                await ProjectRepoRepository(db).update_remote_metadata(
                    project_id, preview_branch=preview_branch
                )
                await ProjectRepository(db).merge_setup_step(
                    project_id,
                    SETUP_STEP_GIT_PUSH_PREVIEW,
                    {"status": SETUP_STEP_STATUS_OK},
                )
                await db.commit()
        else:
            async with db_factory() as db:
                await ProjectRepository(db).merge_setup_step(
                    project_id,
                    SETUP_STEP_GIT_PUSH_PREVIEW,
                    {"status": SETUP_STEP_STATUS_SKIPPED, "detail": "shopify_app_single_deployment"},
                )
                await db.commit()

        # --- Enqueue async work (both run independently in background) ------
        # 1. Production deploy: CF Pages project creation + W4P routing (sets live_hostname)
        #    shopify_app skips this — the Worker deploys via GitHub Actions CI/CD.
        # 2. shopify_app: provision D1 + Queue + R2 + KV + Cloud token.
        if project_type != "shopify_app":
            await enqueue_initial_production_hosting(db_factory, project_id, rev.id, repo_full)
            if is_w4p_project:
                await enqueue_initial_preview_hosting(db_factory, project_id, rev.id, repo_full)
        else:
            from src.workers.enqueue import enqueue_shopify_app_provision
            await enqueue_shopify_app_provision(str(project_id))

        async with db_factory() as db:
            await ProjectRepository(db).merge_setup_step(
                project_id,
                SETUP_STEP_DEPLOY_ENQUEUE,
                {"status": SETUP_STEP_STATUS_OK},
            )
            await db.commit()

        # --- Mark setup complete, project ready ------------------------------
        async with db_factory() as db:
            from src.domain.projects.readiness import project_ready_contract_errors

            readiness_errors = await project_ready_contract_errors(db, project_id)
            if readiness_errors:
                await ProjectRepository(db).merge_setup_step(
                    project_id,
                    SETUP_STEP_AUTH_PROVISION,
                    {"status": SETUP_STEP_STATUS_ERROR, "last_error": "; ".join(readiness_errors)},
                )
                await ProjectRepository(db).update(project_id, status="error")
                await db.commit()
                raise RuntimeError("project readiness contract failed: " + "; ".join(readiness_errors))
            await ProjectRepoRepository(db).update_setup_state(
                project_id, setup_stage=SetupStage.COMPLETED
            )
            _proj_update: dict = {
                "status": "ready",
                "boilerplate_version": boilerplate_version_label,
                "source_imported_commit_sha": source_imported_sha,
            }
            # Clear one-time S3 credentials after successful import
            if template_source and template_source.get("kind") == "s3_import":
                _proj_update["source_s3_access_key_enc"] = None
                _proj_update["source_s3_secret_key_enc"] = None
            await ProjectRepository(db).update(project_id, **_proj_update)
            from src.domain.projects.discovery_service import enqueue_approved_discovery_after_setup

            await enqueue_approved_discovery_after_setup(db, project_id)
            await TimelineEventRepository(db).create(
                project_id,
                "project_setup_complete",
                revision_id=rev.id,
                phase="setup",
                title="Project created",
                detail_text="Project files prepared.",
            )
            await ProjectJobOutboxRepository(db).insert_pending(
                project_id, JOB_KIND_CLOUD_SYNC_POST_READY, {}
            )
            await db.commit()

        await _auto_connect_import_mirror(db_factory, project_id)

        from src.workers.enqueue import enqueue_project_outbox_dispatch_nudge

        try:
            await enqueue_project_outbox_dispatch_nudge(defer_seconds=2.0)
        except Exception:
            logger.warning("project_outbox_dispatch_nudge_failed", exc_info=True)

        logger.info(
            "project_setup_complete",
            project_id=str(project_id),
            repo=repo_full,
            sha=sha,
            revision_id=str(rev.id),
            boilerplate_version=boilerplate_version_label,
        )
        return {
            "status": "ok",
            "sha": sha,
            "repo": repo_full,
            "revision_id": str(rev.id),
            "workspace_path": repo_path,
        }

    except Exception as e:
        await _mark_error()
        logger.error("project_setup_failed", project_id=str(project_id), error=str(e))
        raise