import uuid

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import Text, delete, func, literal, or_, select, text
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.core.permissions import deny_vendor, require_permission, require_portal_item, resolve_portal_grants
from app.core.security import AuthContext, get_auth_context
from app.crud.base import CRUDBase
from app.crud import entity_storage, field_access, field_indexes, form_documents, public_forms, record_sql
from app.crud.access_policy import access_clause
from app.crud.audit import write_audit_log
from app.crud.change_log import check_fresh, changed_keys, fields_diff
from app.crud.dataset_query import attach_user_names, filter_records, parse_filters_param, prepare_filters
from app.crud.filter_tree import conditions
from app.crud.form_rules import evaluate_visibility
from app.crud.form_versions import live_form
from app.models.workflow import Form, FormSubmission, WorkflowStepForm
from app.schemas.master import StorageChange, StorageInfo, StoragePreview
from app.schemas.workflow import (
    FormCreate,
    FormDuplicate,
    FormSchedule,
    ReviewStatus,
    FormFieldAccessRules,
    FormFieldsAdd,
    FormIndexedFields,
    FormRead,
    FormScenarioResult,
    FormScenarioRun,
    FormSubmissionRead,
    FormUpdate,
)

router = APIRouter(prefix="/forms", tags=["forms"])
_submissions_crud = CRUDBase(FormSubmission)  # re-exported for platform_admin.py's tenant-admin submission routes


async def _get_current(db: AsyncSession, tenant_id: uuid.UUID, form_id: uuid.UUID) -> Form:
    stmt = select(Form).where(Form.tenant_id == tenant_id, Form.form_id == form_id, Form.is_current.is_(True))
    obj = (await db.execute(stmt)).scalar_one_or_none()
    if obj is None:
        raise HTTPException(status_code=404, detail="Form not found")
    return obj


async def get_live_form(db: AsyncSession, tenant_id: uuid.UUID, form_id: uuid.UUID) -> Form:
    """What people fill in — see app/crud/form_versions.py."""
    live = await live_form(db, tenant_id, form_id)
    if live is None:
        raise HTTPException(status_code=404, detail="Form not found")
    return live


@router.get("", response_model=list[FormRead])
async def list_forms(
    skip: int = 0,
    limit: int = 100,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "view")),
) -> list[FormRead]:
    stmt = (
        select(Form)
        .where(Form.tenant_id == ctx.tenant_id, Form.is_current.is_(True))
        .order_by(Form.name)
        .offset(skip)
        .limit(limit)
    )
    forms = list((await db.execute(stmt)).scalars().all())
    if ctx.vendor_id is not None:
        grants = await resolve_portal_grants(db, ctx.tenant_id, ctx.user_id)
        allowed = grants.get("forms", set())
        forms = [f for f in forms if str(f.form_id).lower() in allowed]
    tables = await entity_storage.tables_by_source(db, ctx.tenant_id)
    return [FormRead.model_validate(f).model_copy(update={"storage_tier": _tier(tables.get(str(f.form_id)))}) for f in forms]


@router.get("/{form_id}", response_model=FormRead)
async def get_form(
    form_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "view")),
    __: AuthContext = Depends(require_portal_item("forms", "form_id")),
) -> Form:
    """The current version — what the Form Designer edits, possibly an
    unpublished draft. A Public Portal caller gets the live version instead:
    unpublished work isn't theirs to see."""
    if ctx.vendor_id is not None:
        return await get_live_form(db, ctx.tenant_id, form_id)
    return await _get_current(db, ctx.tenant_id, form_id)


@router.get("/{form_id}/live", response_model=FormRead)
async def get_live(
    form_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "view")),
    __: AuthContext = Depends(require_portal_item("forms", "form_id")),
) -> Form:
    return await get_live_form(db, ctx.tenant_id, form_id)


@router.get("/{form_id}/versions", response_model=list[FormRead])
async def list_form_versions(
    form_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "view")),
    __: AuthContext = Depends(require_portal_item("forms", "form_id")),
) -> list[Form]:
    stmt = (
        select(Form).where(Form.tenant_id == ctx.tenant_id, Form.form_id == form_id).order_by(Form.version)
    )
    versions = list((await db.execute(stmt)).scalars().all())
    if not versions:
        raise HTTPException(status_code=404, detail="Form not found")
    return versions


@router.get("/{form_id}/versions/{version}", response_model=FormRead)
async def get_form_version(
    form_id: uuid.UUID,
    version: int,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "view")),
    __: AuthContext = Depends(require_portal_item("forms", "form_id")),
) -> Form:
    stmt = select(Form).where(
        Form.tenant_id == ctx.tenant_id, Form.form_id == form_id, Form.version == version
    )
    obj = (await db.execute(stmt)).scalar_one_or_none()
    if obj is None:
        raise HTTPException(status_code=404, detail="Form version not found")
    return obj


@router.get("/{form_id}/drafts", response_model=list[FormSubmissionRead])
async def list_my_drafts(
    form_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "create")),
    __: AuthContext = Depends(require_portal_item("forms", "form_id")),
) -> list[FormSubmission]:
    """The caller's own saved drafts of this form, newest first, to resume."""
    stmt = (
        select(FormSubmission)
        .where(
            FormSubmission.tenant_id == ctx.tenant_id,
            FormSubmission.form_id == form_id,
            FormSubmission.status == "draft",
            FormSubmission.submitted_by == ctx.user_id,
            FormSubmission.is_deleted.is_(False),
        )
        .order_by(FormSubmission.updated_at.desc())
        .limit(50)
    )
    return list((await db.execute(stmt)).scalars().all())


@router.get("/{form_id}/field-access")
async def get_field_access(
    form_id: uuid.UUID,
    version: int | None = None,
    submission_id: uuid.UUID | None = None,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "view")),
    __: AuthContext = Depends(require_portal_item("forms", "form_id")),
) -> dict:
    """Which fields are read-only or hidden for the caller (each field's
    access_rules — app/crud/field_access.py), so the form renders them that
    way: {"fields": {name: "read_only" | "hidden"}, "step": ...}. For an
    existing submission, its workflow's current step decides step rules."""
    submission = None
    if submission_id is not None:
        submission = (
            await db.execute(select(FormSubmission).where(FormSubmission.tenant_id == ctx.tenant_id, FormSubmission.id == submission_id))
        ).scalar_one_or_none()
        if submission is None or submission.form_id != form_id:
            raise HTTPException(status_code=404, detail="FormSubmission not found")
        version = submission.form_version
    stmt = select(Form).where(Form.tenant_id == ctx.tenant_id, Form.form_id == form_id)
    stmt = stmt.where(Form.version == version) if version is not None else stmt.where(Form.is_current.is_(True))
    form = (await db.execute(stmt)).scalar_one_or_none()
    if form is None:
        raise HTTPException(status_code=404, detail="Form not found")
    step = await field_access.submission_step(db, ctx.tenant_id, submission)
    return {"fields": await field_access.access_for(db, ctx.tenant_id, ctx.user_id, form.field_definitions, step), "step": step}


async def _audit_form(db: AsyncSession, ctx: AuthContext, action: str, form: Form, details: dict | None = None) -> None:
    await write_audit_log(
        db, ctx.tenant_id, ctx.user_id, action, "form", form.form_id, {"name": form.name, "version": form.version, **(details or {})}
    )


@router.post("/test-scenarios/run", response_model=list[FormScenarioResult])
async def run_test_scenarios(
    payload: FormScenarioRun,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> list[FormScenarioResult]:
    """The Form Designer's Tests: each scenario through the submission
    checks (app/crud/form_scenarios.py) against the design as it is on
    screen. Nothing is saved."""
    from app.crud.form_scenarios import run_scenarios

    defs = [f.model_dump(mode="json", exclude_none=True) for f in payload.field_definitions]
    results = await run_scenarios(db, ctx.tenant_id, defs, [s.model_dump(mode="json") for s in payload.scenarios])
    return [FormScenarioResult(**r.__dict__) for r in results]


def _check_public(obj: Form) -> None:
    """A form open to the public may only use what works signed-out —
    app/crud/public_forms.py::public_problems. (And a receipt needs an email
    field to go to, public or not.)"""
    receipt_problem = form_documents.receipt_problem(obj.field_definitions, (obj.settings or {}).get("receipt"))
    if receipt_problem:
        raise HTTPException(status_code=422, detail=f"Receipt email: {receipt_problem}")
    if not (obj.settings or {}).get("public_access"):
        return
    problems = public_forms.public_problems(obj.field_definitions, obj.events, obj.prefill, obj.settings)
    if problems:
        raise HTTPException(
            status_code=422,
            detail={"code": "not_public_safe", "message": "This form can't be filled in without signing in: " + " ".join(problems), "problems": problems},
        )


@router.post("", response_model=FormRead, status_code=status.HTTP_201_CREATED)
async def create_form(
    payload: FormCreate,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "create")),
    __: AuthContext = Depends(deny_vendor()),
) -> FormRead:
    obj = Form(
        tenant_id=ctx.tenant_id,
        name=payload.name,
        field_definitions=[f.model_dump(mode="json") for f in payload.field_definitions],
        events=[e.model_dump(mode="json") for e in payload.events],
        prefill=payload.prefill.model_dump(mode="json") if payload.prefill else None,
        settings=payload.settings.model_dump(mode="json"),
        status="draft",
        is_current=True,
        created_by=ctx.user_id,
    )
    _check_public(obj)
    db.add(obj)
    await db.flush()
    await db.refresh(obj)
    if payload.storage_tier == "table":
        # Created in this transaction: if the table can't be, neither is the form.
        source, _published = await _storage_source(db, ctx, obj.form_id)
        try:
            await entity_storage.promote(db, ctx.tenant_id, ctx.user_id, source)
        except entity_storage.StorageError as exc:
            raise _storage_http(exc) from exc
    await field_indexes.sync_field_indexes(db, "form", obj.form_id, obj.field_definitions)
    await _audit_form(db, ctx, "form.create", obj, {"storage_tier": payload.storage_tier})
    return FormRead.model_validate(obj).model_copy(update={"storage_tier": payload.storage_tier})


@router.patch("/{form_id}", response_model=FormRead)
async def update_form_draft(
    form_id: uuid.UUID,
    payload: FormUpdate,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> Form:
    """In-place edits only apply to an unpublished draft — FR-FRM-009: a
    published form's structure is versioned so prior submissions are never
    retroactively changed. Use /new-version once published."""
    obj = await _get_current(db, ctx.tenant_id, form_id)
    if obj.status != "draft":
        raise HTTPException(
            status_code=409,
            detail="This form is published; edit it via POST /forms/{id}/new-version (FR-FRM-009).",
        )
    check_fresh(obj, payload.expected_updated_at)
    before = {"name": obj.name, "field_definitions": obj.field_definitions, "events": obj.events, "prefill": obj.prefill}
    if payload.name is not None:
        obj.name = payload.name
    if payload.field_definitions is not None:
        obj.field_definitions = [f.model_dump(mode="json") for f in payload.field_definitions]
    if payload.events is not None:
        obj.events = [e.model_dump(mode="json") for e in payload.events]
    # Unlike field_definitions/events (lists — "no change" and "empty" are
    # both spelled without null), prefill is meaningfully nullable ("no
    # prefill configured"), so "omitted" and "explicitly cleared" have to be
    # told apart via model_fields_set rather than an `is not None` check.
    if "prefill" in payload.model_fields_set:
        obj.prefill = payload.prefill.model_dump(mode="json") if payload.prefill else None
    if payload.settings is not None:
        obj.settings = payload.settings.model_dump(mode="json")
    _check_public(obj)
    await db.flush()
    await db.refresh(obj)
    # A never-published form's table follows its draft; once published, the
    # table follows the published version and this changes nothing.
    await _sync_storage(db, ctx, form_id)
    await field_indexes.sync_field_indexes(db, "form", obj.form_id, obj.field_definitions)
    after = {"name": obj.name, "field_definitions": obj.field_definitions, "events": obj.events, "prefill": obj.prefill}
    await _audit_form(
        db, ctx, "form.update", obj,
        {"changed": changed_keys(before, after, ["name", "events", "prefill"]), **fields_diff(before["field_definitions"], after["field_definitions"])},
    )
    return obj


@router.post("/{form_id}/fields", response_model=FormRead)
async def add_form_fields(
    form_id: uuid.UUID,
    payload: FormFieldsAdd,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> Form:
    """Append fields to the current version without versioning it — the Form
    stays published and live, like adding a field to a datasource. Existing
    fields are never touched, so earlier submissions stay valid; the new ones
    are always optional (an earlier submission can't have answered them, and
    editing it must not start failing on a field it never had)."""
    obj = await _get_current(db, ctx.tenant_id, form_id)
    existing = {f["name"] for f in obj.field_definitions}
    seen: set[str] = set()
    for f in payload.field_definitions:
        if f.name in existing or f.name in seen:
            raise HTTPException(status_code=409, detail=f"This form already has a field named '{f.name}'.")
        seen.add(f.name)
    added = [f.model_copy(update={"required": False, "required_message": None}).model_dump(mode="json") for f in payload.field_definitions]
    obj.field_definitions = [*obj.field_definitions, *added]
    _check_public(obj)
    await db.flush()
    await db.refresh(obj)
    await _sync_storage(db, ctx, form_id)
    await field_indexes.sync_field_indexes(db, "form", obj.form_id, obj.field_definitions)
    await _audit_form(db, ctx, "form.fields_add", obj, {"fields_added": [f["name"] for f in added]})
    return obj


@router.put("/{form_id}/fields/{field_name}/access-rules", response_model=FormRead)
async def set_field_access_rules(
    form_id: uuid.UUID,
    field_name: str,
    payload: FormFieldAccessRules,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> Form:
    """One field's permissions (app/crud/field_access.py) on the current
    version in place — allowed on a published form too."""
    obj = await _get_current(db, ctx.tenant_id, form_id)
    if not any(f["name"] == field_name for f in obj.field_definitions):
        raise HTTPException(status_code=404, detail=f"No field named '{field_name}'")
    before = next(f.get("access_rules") or [] for f in obj.field_definitions if f["name"] == field_name)
    rules = [r.model_dump(mode="json") for r in payload.access_rules]
    obj.field_definitions = [{**f, "access_rules": rules} if f["name"] == field_name else f for f in obj.field_definitions]
    await db.flush()
    await db.refresh(obj)
    await _audit_form(db, ctx, "form.field_access", obj, {"field": field_name, "before": before, "after": rules})
    return obj


@router.put("/{form_id}/indexed-fields", response_model=FormRead)
async def set_indexed_fields(
    form_id: uuid.UUID,
    payload: FormIndexedFields,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> Form:
    """Which fields are indexed (see app/crud/field_indexes.py), on the
    current version in place — allowed on a published form too, since an
    index changes nothing about the form or its submissions."""
    obj = await _get_current(db, ctx.tenant_id, form_id)
    names = {f["name"] for f in obj.field_definitions}
    unknown = [n for n in payload.field_names if n not in names]
    if unknown:
        raise HTTPException(status_code=422, detail=f"No such field: {', '.join(unknown)}")
    wanted = set(payload.field_names)
    obj.field_definitions = [{**f, "indexed": f["name"] in wanted} for f in obj.field_definitions]
    await db.flush()
    await db.refresh(obj)
    await _sync_storage(db, ctx, form_id)
    await field_indexes.sync_field_indexes(db, "form", obj.form_id, obj.field_definitions)
    await _audit_form(db, ctx, "form.indexes", obj, {"indexed": sorted(wanted)})
    return obj


@router.post("/{form_id}/publish", response_model=FormRead)
async def publish_form(
    form_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "publish")),
    __: AuthContext = Depends(deny_vendor()),
) -> Form:
    obj = await _get_current(db, ctx.tenant_id, form_id)
    if obj.status == "published":
        return obj
    _check_public(obj)
    obj.status = "published"
    await db.flush()
    await db.refresh(obj)
    # A form stored as a table follows its live version — or the publish is
    # refused (and rolled back) when its submissions don't fit the change.
    await _sync_storage(db, ctx, form_id)
    await _audit_form(db, ctx, "form.publish", obj)
    return obj


@router.post("/{form_id}/new-version", response_model=FormRead, status_code=status.HTTP_201_CREATED)
async def new_form_version(
    form_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> Form:
    current = await _get_current(db, ctx.tenant_id, form_id)
    current.is_current = False
    await db.flush()

    new_version = Form(
        tenant_id=ctx.tenant_id,
        form_id=form_id,
        version=current.version + 1,
        name=current.name,
        field_definitions=current.field_definitions,
        events=current.events,
        prefill=current.prefill,
        settings=current.settings,
        status="draft",
        is_current=True,
        created_by=ctx.user_id,
    )
    db.add(new_version)
    await db.flush()
    await db.refresh(new_version)
    await _audit_form(db, ctx, "form.new_version", new_version)
    return new_version


@router.put("/{form_id}/schedule", response_model=FormRead)
async def set_form_schedule(
    form_id: uuid.UUID,
    payload: FormSchedule,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> Form:
    """Open, pause, close or archive the form now — on its live version and
    any draft alike, without a new version: the schedule is about taking
    submissions, not the form's structure (app/crud/form_schedule.py)."""
    current = await _get_current(db, ctx.tenant_id, form_id)
    live = (
        await db.execute(
            select(Form).where(Form.tenant_id == ctx.tenant_id, Form.form_id == form_id, Form.status == "published").order_by(Form.version.desc()).limit(1)
        )
    ).scalar_one_or_none()
    schedule = payload.model_dump(mode="json")
    for row in {id(r): r for r in (current, live) if r is not None}.values():
        row.settings = {**(row.settings or {}), "schedule": schedule}
    await db.flush()
    await db.refresh(current)
    await _audit_form(db, ctx, "form.schedule", current, {"schedule": schedule})
    return current


@router.post("/{form_id}/duplicate", response_model=FormRead, status_code=status.HTTP_201_CREATED)
async def duplicate_form(
    form_id: uuid.UUID,
    payload: FormDuplicate,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "create")),
    __: AuthContext = Depends(deny_vendor()),
) -> Form:
    """A new form from this one's current design (e.g. last year's program
    as the template for this year's) — a draft with no submissions, its own
    auto-numbers, and the schedule's dates cleared."""
    source = await _get_current(db, ctx.tenant_id, form_id)
    settings_copy = dict(source.settings or {})
    settings_copy["schedule"] = {**((settings_copy.get("schedule") or {})), "opens_at": None, "closes_at": None, "paused": False, "archived": False}
    copy = Form(
        tenant_id=ctx.tenant_id,
        name=(payload.name or "").strip() or f"Copy of {source.name}"[:200],
        field_definitions=source.field_definitions,
        events=source.events,
        prefill=source.prefill,
        settings=settings_copy,
        status="draft",
        is_current=True,
        created_by=ctx.user_id,
    )
    db.add(copy)
    await db.flush()
    await db.refresh(copy)
    await field_indexes.sync_field_indexes(db, "form", copy.form_id, copy.field_definitions)
    await _audit_form(db, ctx, "form.duplicate", copy, {"from_form_id": str(form_id), "from_version": source.version})
    return copy


@router.delete("/{form_id}/draft", response_model=FormRead)
async def discard_form_draft(
    form_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> Form:
    """Throws away the unpublished draft version; the version before it
    (the live one) becomes current again."""
    current = await _get_current(db, ctx.tenant_id, form_id)
    if current.status != "draft":
        raise HTTPException(status_code=409, detail="There's no unpublished draft to discard.")
    previous = (
        await db.execute(
            select(Form)
            .where(Form.tenant_id == ctx.tenant_id, Form.form_id == form_id, Form.version < current.version)
            .order_by(Form.version.desc())
            .limit(1)
        )
    ).scalar_one_or_none()
    if previous is None:
        raise HTTPException(status_code=409, detail="This form has never been published — delete it instead of discarding the draft.")
    used = await db.execute(
        select(func.count()).select_from(FormSubmission).where(FormSubmission.form_id == form_id, FormSubmission.form_version == current.version)
    )
    if used.scalar_one():
        raise HTTPException(status_code=409, detail="This draft already has submissions, so it can't be discarded.")
    linked = await db.execute(
        select(func.count()).select_from(WorkflowStepForm).where(WorkflowStepForm.form_id == form_id, WorkflowStepForm.form_version == current.version)
    )
    if linked.scalar_one():
        raise HTTPException(status_code=409, detail="A workflow step uses this draft version, so it can't be discarded.")
    discarded = current.version
    await db.delete(current)
    await db.flush()
    previous.is_current = True
    await db.flush()
    await db.refresh(previous)
    await _audit_form(db, ctx, "form.discard_draft", previous, {"discarded_version": discarded})
    return previous


@router.post("/{form_id}/versions/{version}/restore", response_model=FormRead)
async def restore_form_version(
    form_id: uuid.UUID,
    version: int,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> Form:
    """Brings back an earlier version's design as a draft — the current
    draft's content is replaced, or a new draft version is started when the
    current one is published. Nothing goes live until it's published."""
    source = (
        await db.execute(select(Form).where(Form.tenant_id == ctx.tenant_id, Form.form_id == form_id, Form.version == version))
    ).scalar_one_or_none()
    if source is None:
        raise HTTPException(status_code=404, detail="Form version not found")
    current = await _get_current(db, ctx.tenant_id, form_id)
    if current.status == "draft":
        target = current
    else:
        current.is_current = False
        await db.flush()
        target = Form(tenant_id=ctx.tenant_id, form_id=form_id, version=current.version + 1, status="draft", is_current=True, created_by=ctx.user_id)
        db.add(target)
    target.name = source.name
    target.field_definitions = source.field_definitions
    target.events = source.events
    target.prefill = source.prefill
    target.settings = source.settings
    await db.flush()
    await db.refresh(target)
    await field_indexes.sync_field_indexes(db, "form", form_id, target.field_definitions)
    await _audit_form(db, ctx, "form.restore", target, {"restored_from": version})
    return target


@router.delete("/{form_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_form(
    form_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "delete")),
    __: AuthContext = Depends(deny_vendor()),
) -> None:
    stmt = select(Form).where(Form.tenant_id == ctx.tenant_id, Form.form_id == form_id)
    versions = list((await db.execute(stmt)).scalars().all())
    if not versions:
        raise HTTPException(status_code=404, detail="Form not found")
    if any(v.status == "published" for v in versions):
        raise HTTPException(
            status_code=409, detail="A published form cannot be deleted; prior submissions must remain valid."
        )
    linked_from = (
        await db.execute(
            text(
                """
                SELECT DISTINCT m.name
                FROM masters m, jsonb_array_elements(m.field_schema) elem
                WHERE m.tenant_id = :tenant_id AND elem->>'reference_form_id' = :form_id
                """
            ),
            {"tenant_id": str(ctx.tenant_id), "form_id": str(form_id)},
        )
    ).all()
    if linked_from:
        raise HTTPException(
            status_code=409,
            detail=f"Form is linked from datasource(s): {', '.join(row[0] for row in linked_from)}; cannot delete.",
        )
    if await entity_storage.entity_table(db, ctx.tenant_id, form_id):
        try:
            await entity_storage.demote(db, ctx.tenant_id, ctx.user_id, form_id, reason="Form deleted")
        except entity_storage.StorageError as exc:
            raise _storage_http(exc) from exc
    await db.execute(delete(Form).where(Form.tenant_id == ctx.tenant_id, Form.form_id == form_id))
    await db.flush()
    await field_indexes.sync_field_indexes(db, "form", form_id, [])
    await write_audit_log(db, ctx.tenant_id, ctx.user_id, "form.delete", "form", form_id, {"name": versions[-1].name})


# ---- Storage tier (app/crud/entity_storage.py) ------------------------------------
# A form's submissions can be stored as a real table too; it follows the
# form's live version — the latest published one, or the draft of a form
# never published.


def _tier(row) -> str:
    return "flexible" if row is None else "building" if row.state == "building" else "table"


def _storage_http(exc: entity_storage.StorageError) -> HTTPException:
    detail: dict = {"code": "storage", "message": exc.message}
    if exc.issues:
        detail["issues"] = exc.issues
    return HTTPException(status_code=exc.status, detail=detail)


async def _sync_storage(db: AsyncSession, ctx: AuthContext, form_id: uuid.UUID) -> None:
    try:
        await entity_storage.sync_form(db, ctx.tenant_id, ctx.user_id, form_id)
    except entity_storage.StorageError as exc:
        raise _storage_http(exc) from exc


async def _storage_source(db: AsyncSession, ctx: AuthContext, form_id: uuid.UUID) -> tuple[entity_storage.Source, bool]:
    found = await entity_storage.form_source(db, ctx.tenant_id, form_id)
    if found is None:
        raise HTTPException(status_code=404, detail="Form not found")
    return found


@router.get("/{form_id}/storage", response_model=StorageInfo)
async def get_form_storage(
    form_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "view")),
    __: AuthContext = Depends(deny_vendor()),
) -> dict:
    source, _published = await _storage_source(db, ctx, form_id)
    return await entity_storage.describe(db, ctx.tenant_id, source)


@router.get("/{form_id}/storage/preview", response_model=StoragePreview)
async def preview_form_storage(
    form_id: uuid.UUID,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "view")),
    __: AuthContext = Depends(deny_vendor()),
) -> dict:
    source, _published = await _storage_source(db, ctx, form_id)
    return await entity_storage.preview(db, ctx.tenant_id, source)


@router.post("/{form_id}/storage", response_model=StorageInfo)
async def change_form_storage(
    form_id: uuid.UUID,
    payload: StorageChange,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "edit")),
    __: AuthContext = Depends(deny_vendor()),
) -> dict:
    source, _published = await _storage_source(db, ctx, form_id)
    try:
        if payload.tier == "table":
            await entity_storage.promote(db, ctx.tenant_id, ctx.user_id, source)
        else:
            await entity_storage.demote(db, ctx.tenant_id, ctx.user_id, form_id)
    except entity_storage.StorageError as exc:
        raise _storage_http(exc) from exc
    await write_audit_log(db, ctx.tenant_id, ctx.user_id, "form.storage", "form", form_id, {"name": source.name, "tier": payload.tier})
    return await entity_storage.describe(db, ctx.tenant_id, source)


@router.post("/{form_id}/simulate", response_model=dict[str, bool])
async def simulate_form_visibility(
    form_id: uuid.UUID,
    sample_values: dict,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "view")),
    __: AuthContext = Depends(require_portal_item("forms", "form_id")),
) -> dict[str, bool]:
    """FR-FRM-007: live preview / simulate-submission mode. Returns, per
    field name, whether it would be visible given the supplied sample driver
    values — evaluated against the CURRENT (possibly still-draft) version,
    since this exists for the designer to check before publishing."""
    form = await _get_current(db, ctx.tenant_id, form_id)
    return evaluate_visibility(form.field_definitions, sample_values)


def _submission_matches_search(field_values: dict, q: str) -> bool:
    """Case-insensitive substring match across every value in a submission's
    field_values — same OR-across-all-fields semantics as
    routers/masters.py::_record_matches_search, so a Page Designer Data List
    block's search box behaves identically regardless of whether it's bound
    to a Master or a Form."""
    needle = q.lower()
    return any(needle in str(v).lower() for v in field_values.values() if v is not None)


_SUBMISSION_ORDER = (FormSubmission.submitted_at.desc(), FormSubmission.id.desc())


async def _submissions_query(
    db: AsyncSession, ctx: AuthContext, form_id: uuid.UUID, q: str | None, filters: str | None = None, review_status: str | None = None
) -> tuple[list, bool, list[dict], set[str]]:
    """This caller's visible submissions for `form_id` — vendor scoping,
    search box (q) and a Data List block's own filters — as one SQL WHERE
    (app/crud/record_sql.py): (where, exact, parsed filters, declared field
    names). The one code path both list_form_submissions and
    count_form_submissions use, so list and count never disagree. See
    routers/masters.py::_RecordQuery for the exact/inexact split."""
    try:
        parsed = parse_filters_param(filters)
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    form = (
        await db.execute(select(Form).where(Form.tenant_id == ctx.tenant_id, Form.form_id == form_id, Form.is_current.is_(True)))
    ).scalar_one_or_none()
    src = record_sql.RecordSource("form", form_id, form.field_definitions if form else [], form=form)
    where, exact = src.base_where(ctx.tenant_id), True
    if ctx.vendor_id is not None:
        where.append(FormSubmission.vendor_id == ctx.vendor_id)
    if review_status:
        # Older submissions (and forms without review) have none: "submitted".
        where.append(
            or_(FormSubmission.review_status == review_status, FormSubmission.review_status.is_(None))
            if review_status == "submitted"
            else FormSubmission.review_status == review_status
        )
    restriction = await access_clause(db, ctx.tenant_id, ctx.user_id, src, "view")
    if restriction is not None:
        where.append(restriction)
    # Fields hidden to the caller can't be searched or filtered on — that
    # would reveal their values (app/crud/field_access.py).
    hidden = field_access.hidden_names(
        await field_access.access_for(db, ctx.tenant_id, ctx.user_id, src.field_defs, with_step_rules=False)
    )
    blocked = sorted({f["field"] for f in conditions(parsed)} & hidden)
    if blocked:
        raise HTTPException(status_code=403, detail=f"You can't filter on {', '.join(blocked)}.")
    if q:
        searched = src.fv.op("-")(literal(sorted(hidden), ARRAY(Text))) if hidden else src.fv
        search = record_sql.search_clause(searched, q, src.field_defs, document=src.fv)
        where.append(search.sql)
        exact = search.exact
    if parsed:
        # A Page Data List block's own filters (e.g. created_by equals
        # {{current_user.id}}) — same semantics as every dataset query.
        resolved = await prepare_filters(db, ctx.tenant_id, ctx.user_id, parsed, src.field_defs)
        clauses, filters_exact = record_sql.filter_clauses(src, resolved, ctx.tenant_id)
        where += clauses
        exact = exact and filters_exact
    return where, exact, parsed, src.declared


async def _list_and_filter_submissions(
    db: AsyncSession, ctx: AuthContext, form_id: uuid.UUID, q: str | None, filters: str | None = None, review_status: str | None = None
) -> list[FormSubmission]:
    """Every visible submission, for a query SQL could only narrow: the
    narrowed rows finished in Python with the original semantics."""
    where, _, parsed, declared = await _submissions_query(db, ctx, form_id, q, filters, review_status)
    stmt = select(FormSubmission).where(*where).order_by(*_SUBMISSION_ORDER)
    submissions = [row[0] for row in await record_sql.read_capped(db, stmt)]
    if q:
        form = (
            await db.execute(select(Form).where(Form.tenant_id == ctx.tenant_id, Form.form_id == form_id, Form.is_current.is_(True)))
        ).scalar_one_or_none()
        access = await field_access.access_for(db, ctx.tenant_id, ctx.user_id, form.field_definitions if form else [], with_step_rules=False)
        submissions = [s for s in submissions if _submission_matches_search(field_access.without_hidden(s.field_values, access), q)]
    if parsed:
        form_defs = (
            await db.execute(select(Form.field_definitions).where(Form.tenant_id == ctx.tenant_id, Form.form_id == form_id, Form.is_current.is_(True)))
        ).scalar_one_or_none()
        submissions = await filter_records(db, ctx.tenant_id, ctx.user_id, submissions, parsed, declared, form_defs or [])
    return submissions


@router.get("/{form_id}/submissions", response_model=list[FormSubmissionRead])
async def list_form_submissions(
    form_id: uuid.UUID,
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=record_sql.MAX_PAGE_SIZE),
    after: uuid.UUID | None = None,
    q: str | None = None,
    filters: str | None = None,
    review_status: ReviewStatus | None = None,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "view")),
    __: AuthContext = Depends(require_portal_item("forms", "form_id")),
) -> list[FormSubmission]:
    """See routers/form_submissions.py for creating/fetching/editing/deleting
    a single submission — this route is just the form-scoped listing view. A
    vendor caller only ever sees their own organization's submissions, even
    though the item-level check above only confirms they can access the form
    itself (other vendors' submitted data is a separate concern). `after` is
    the last submission of the page before, as on list_master_records."""
    where, exact, _, _ = await _submissions_query(db, ctx, form_id, q, filters, review_status)
    if exact:
        stmt = select(FormSubmission).where(*where).order_by(*_SUBMISSION_ORDER).limit(limit)
        position = None
        if after:
            position = (
                await db.execute(
                    select(FormSubmission.submitted_at, FormSubmission.id).where(
                        FormSubmission.tenant_id == ctx.tenant_id, FormSubmission.form_id == form_id, FormSubmission.id == after
                    )
                )
            ).first()
        if position:
            stmt = stmt.where(record_sql.after_clause((FormSubmission.submitted_at, FormSubmission.id), tuple(position), descending=True))
        else:
            stmt = stmt.offset(skip)
        page = list((await db.execute(stmt)).scalars().all())
    else:
        visible = await _list_and_filter_submissions(db, ctx, form_id, q, filters, review_status)
        at = next((i for i, s in enumerate(visible) if s.id == after), None) if after else None
        start = skip if at is None else at + 1
        page = visible[start : start + limit]
    await attach_user_names(db, ctx.tenant_id, page)
    # Fields hidden to the caller (rules without workflow steps — see
    # app/crud/field_access.py) are left out of every row.
    form = (await db.execute(select(Form).where(Form.tenant_id == ctx.tenant_id, Form.form_id == form_id, Form.is_current.is_(True)))).scalar_one_or_none()
    access = await field_access.access_for(db, ctx.tenant_id, ctx.user_id, form.field_definitions if form else [], with_step_rules=False)
    if field_access.hidden_names(access):
        page = [
            FormSubmissionRead.model_validate(s).model_copy(update={"field_values": field_access.without_hidden(s.field_values, access)})
            for s in page
        ]
    return page


@router.get("/{form_id}/submissions/count")
async def count_form_submissions(
    form_id: uuid.UUID,
    q: str | None = None,
    filters: str | None = None,
    review_status: ReviewStatus | None = None,
    ctx: AuthContext = Depends(get_auth_context),
    db: AsyncSession = Depends(get_db),
    _: AuthContext = Depends(require_permission("forms", "view")),
    __: AuthContext = Depends(require_portal_item("forms", "form_id")),
) -> dict[str, int | bool]:
    """The total-matching-row count a paginated caller (Page Designer's Data
    List block, in its live-pagination/search mode) needs to render "page N
    of M" / know when Next should disable — mirrors
    routers/masters.py::count_master_records, `capped` included."""
    where, exact, _, _ = await _submissions_query(db, ctx, form_id, q, filters, review_status)
    if exact:
        total, capped = await record_sql.capped_count(db, FormSubmission, where)
        return {"total": total, "capped": capped}
    return {"total": len(await _list_and_filter_submissions(db, ctx, form_id, q, filters, review_status)), "capped": False}