"""Business logic layer for the bitable subsystem.

The service wraps :class:`BitableRepository` and adds business rules:
field ownership, upsert semantics, formula recalc triggering, etc.

Routes and CLI call this layer — never the repository directly.
"""

from __future__ import annotations

import logging
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import TYPE_CHECKING, TypeAlias

from agentkit.bitable.db import BitableDB
from agentkit.bitable.models import (
    DEFAULT_FIELD_TEMPLATES,
    BitableFile,
    Field,
    FieldOwner,
    FieldType,
    Record,
    RecalcTask,
    Table,
    View,
    ViewType,
)
from agentkit.bitable.repository import BitableRepository

if TYPE_CHECKING:
    from typing import Protocol

    class _RecalcWorker(Protocol):
        """Structural type for the recalc worker's cache-invalidation surface."""

        def invalidate_engine(self, table_id: str) -> None: ...


logger = logging.getLogger(__name__)

# Record values are JSON scalars (text/number/none). Attachment/image fields
# store lists of dicts at runtime, but the common-case scalar shape is captured
# here for annotation clarity; fall back to dict[str, object] where lists occur.
BitableRecord: TypeAlias = dict[str, str | int | float | None]


class FieldDependencyError(Exception):
    """Raised when deleting a field that has dependencies (formula refs, PK, views)."""

    def __init__(self, message: str, dependencies: dict[str, object]) -> None:
        super().__init__(message)
        self.dependencies = dependencies


class BitableService:
    """Bitable business logic service.

    Usage::

        service = BitableService(db)
        table = await service.create_table(name="Orders")
    """

    def __init__(self, db: BitableDB) -> None:
        self._db = db
        self._repo = BitableRepository(db)
        self._recalc_worker: _RecalcWorker | None = None  # set via set_recalc_worker

    @property
    def repo(self) -> BitableRepository:
        return self._repo

    def set_recalc_worker(self, worker: _RecalcWorker) -> None:
        """Register the long-lived RecalcWorker so field changes can invalidate its engine cache.

        Called after both service and worker are constructed (breaks the
        service<->worker construction cycle). When unset, field mutations
        simply skip cache invalidation — the next worker start rebuilds engines.
        """
        self._recalc_worker = worker

    def _invalidate_engine_cache(self, table_id: str) -> None:
        """Invalidate the worker's cached formula engine for a table (P1 #5)."""
        if self._recalc_worker is not None:
            self._recalc_worker.invalidate_engine(table_id)

    # ── Files (R1) ─────────────────────────────────────────

    async def create_file(
        self,
        name: str,
        icon: str = "📋",
        description: str = "",
        owner_user_id: str | None = None,
    ) -> BitableFile:
        """Create a new bitable file (top-level container)."""
        return await self._repo.create_file(
            name=name,
            icon=icon,
            description=description,
            owner_user_id=owner_user_id,
        )

    async def get_file(self, file_id: str) -> BitableFile | None:
        return await self._repo.get_file(file_id)

    async def list_files(self, owner_user_id: str | None = None) -> list[BitableFile]:
        return await self._repo.list_files(owner_user_id=owner_user_id)

    async def update_file(self, file_id: str, **kwargs: object) -> BitableFile | None:
        return await self._repo.update_file(file_id, **kwargs)

    async def delete_file(self, file_id: str) -> bool:
        """Delete a file and cascade-delete all its tables (and their fields/records/views)."""
        tables = await self._repo.list_tables_by_file(file_id)
        for t in tables:
            await self.delete_table(t.id)
        return await self._repo.delete_file(file_id)

    async def list_tables_by_file(self, file_id: str) -> list[Table]:
        """List all tables in a file."""
        return await self._repo.list_tables_by_file(file_id)

    # ── Tables ──────────────────────────────────────────────

    async def create_table(
        self,
        name: str,
        description: str = "",
        primary_key_field_id: str | None = None,
        owner_user_id: str | None = None,
        file_id: str | None = None,
    ) -> Table:
        """Create a new bitable table with default fields (R2).

        Creates the table, then immediately creates 5 default fields
        (标题/状态/日期/创建人/创建时间). Creates PK unique index if PK is set.
        """
        table = await self._repo.create_table(
            name=name,
            description=description,
            primary_key_field_id=primary_key_field_id,
            owner_user_id=owner_user_id,
            file_id=file_id,
        )
        # R2: create the 5 default fields for every new table.
        await self._create_default_fields(table.id)
        if primary_key_field_id:
            await self._repo.create_pk_unique_index(table.id, primary_key_field_id)
        return table

    async def _create_default_fields(self, table_id: str) -> list[Field]:
        """Create the 5 default fields on a freshly-created table (R2).

        Uses ``DEFAULT_FIELD_TEMPLATES`` from models. Fields are created
        via the repository directly (no engine cache invalidation needed —
        no formula fields among defaults).
        """
        created: list[Field] = []
        for tpl in DEFAULT_FIELD_TEMPLATES:
            field = await self._repo.create_field(
                table_id=table_id,
                name=tpl["name"],
                field_type=tpl["field_type"],
                config=tpl.get("config", {}) or {},
                owner=tpl["owner"],
            )
            created.append(field)
        return created

    async def get_table(self, table_id: str) -> Table | None:
        return await self._repo.get_table(table_id)

    async def list_tables(self, owner_user_id: str | None = None) -> list[Table]:
        return await self._repo.list_tables(owner_user_id=owner_user_id)

    async def update_table(self, table_id: str, **kwargs: object) -> Table | None:
        """Update table attrs. Creates PK unique index if primary_key_field_id is set."""
        table = await self._repo.update_table(table_id, **kwargs)
        if table and kwargs.get("primary_key_field_id"):
            await self._repo.create_pk_unique_index(table_id, kwargs["primary_key_field_id"])
        return table

    async def delete_table(self, table_id: str) -> bool:
        return await self._repo.delete_table(table_id)

    # ── Fields ──────────────────────────────────────────────

    async def create_field(
        self,
        table_id: str,
        name: str,
        field_type: FieldType,
        config: dict[str, object] | None = None,
        owner: FieldOwner = FieldOwner.user,
    ) -> Field:
        """Create a new field. U2 will add formula validation and DAG updates."""
        field = await self._repo.create_field(
            table_id=table_id,
            name=name,
            field_type=field_type,
            config=config or {},
            owner=owner,
        )
        # New formula field changes the table's DAG — invalidate cached engine (P1 #5).
        if field_type == FieldType.formula:
            self._invalidate_engine_cache(table_id)
        return field

    async def get_field(self, field_id: str) -> Field | None:
        return await self._repo.get_field(field_id)

    async def list_fields(self, table_id: str) -> list[Field]:
        return await self._repo.list_fields(table_id)

    async def update_field(self, field_id: str, **kwargs: object) -> Field | None:
        """Update a field. U2 will add dependency checking."""
        field = await self._repo.update_field(field_id, **kwargs)
        if field is not None:
            # Any field update (name, config, formula_expr) can affect the DAG (P1 #5).
            self._invalidate_engine_cache(field.table_id)
        return field

    async def delete_field(self, field_id: str, force: bool = False) -> bool:
        """Delete a field with dependency checking.

        Returns True if deleted. Raises ValueError with dependency info if
        blocked (use force=True to override with cascade cleanup).
        """
        field = await self._repo.get_field(field_id)
        if not field:
            return False

        # Check dependencies
        deps: dict[str, object] = {}

        # 1. Is it a primary key field?
        table = await self._repo.get_table(field.table_id)
        if table and table.primary_key_field_id == field_id:
            deps["is_primary_key"] = True

        # 2. Formula fields referencing this field?
        formula_deps = await self._repo.find_formula_fields_referencing(field.table_id, field_id)
        if formula_deps:
            deps["formula_fields"] = [{"id": f.id, "name": f.name} for f in formula_deps]

        # 3. Views referencing this field?
        view_deps = await self._repo.find_views_referencing_field(field.table_id, field_id)
        if view_deps:
            deps["views"] = [{"id": v.id, "name": v.name} for v in view_deps]

        if deps and not force:
            raise FieldDependencyError(
                f"Cannot delete field '{field.name}': has dependencies", deps
            )

        # Force delete: cascade cleanup
        if deps and force:
            # Mark formula fields as error
            for f in formula_deps:
                await self._repo.update_field(
                    f.id,
                    config={**f.config, "error": "referenced field deleted"},
                )
            # Remove field from all records' JSONB
            await self._repo.remove_field_from_records(field.table_id, field_id)

        deleted = await self._repo.delete_field(field_id)
        if deleted:
            # Removing a field changes the table's DAG — invalidate cached engine (P1 #5).
            self._invalidate_engine_cache(field.table_id)
        return deleted

    # ── Records ─────────────────────────────────────────────

    async def create_record(
        self,
        table_id: str,
        values: BitableRecord | None = None,
        actor_user_id: str | None = None,
    ) -> Record:
        """Create a new record. Triggers recalc for affected formula fields.

        R2: auto-fills agent-owned default fields (创建人/创建时间) with
        system values. User-supplied values for agent-owned fields are
        overwritten — agent ownership means system-managed (per plan edge case).
        Matching by field_type+owner (not name) so renames don't break auto-fill.
        """
        values = dict(values or {})
        # Auto-fill agent-owned fields based on the table's field definitions.
        fields = await self._repo.list_fields(table_id)
        now_iso = datetime.now(timezone.utc).isoformat()
        creator_value = actor_user_id or "system"
        for f in fields:
            if f.owner != FieldOwner.agent:
                continue
            if f.field_type == FieldType.text:
                values[f.id] = creator_value
            elif f.field_type == FieldType.date:
                values[f.id] = now_iso
        record = await self._repo.create_record(table_id, values)
        await self._trigger_recalc_for_affected_fields(table_id, record.id)
        return record

    async def create_records_batch(
        self, table_id: str, records_values: list[BitableRecord]
    ) -> list[Record]:
        """Batch-create records (P2 #19). Triggers recalc for each record.

        Processes in chunks of 500 to keep memory bounded.
        """
        batch_size = 500
        all_records: list[Record] = []
        for i in range(0, len(records_values), batch_size):
            chunk = records_values[i : i + batch_size]
            records = await self._repo.create_records_batch(table_id, chunk)
            for rec in records:
                await self._trigger_recalc_for_affected_fields(table_id, rec.id)
            all_records.extend(records)
        return all_records

    async def get_record(self, record_id: str) -> Record | None:
        return await self._repo.get_record(record_id)

    async def list_records(
        self, table_id: str, cursor: str | None = None, limit: int = 50
    ) -> tuple[list[Record], str | None]:
        """List records with cursor-based pagination (KTD9)."""
        return await self._repo.list_records(table_id, cursor=cursor, limit=limit)

    async def list_records_filtered(
        self,
        table_id: str,
        filters: list[dict[str, object]] | None = None,
        sorts: list[dict[str, object]] | None = None,
        cursor: str | None = None,
        limit: int = 50,
    ) -> tuple[list[Record], str | None]:
        """List records with view filters/sorts + cursor pagination.

        Validates every ``field_id`` in filters/sorts against the table's
        actual fields before passing to the repository — field IDs are
        interpolated into SQL path literals and must be known system UUIDs.
        """
        if filters or sorts:
            fields = await self._repo.list_fields(table_id)
            valid_field_ids = {f.id for f in fields}
            for f in filters or []:
                fid = f.get("field_id", "")
                if fid not in valid_field_ids:
                    raise ValueError(f"Unknown field_id in filter: {fid}")
            for s in sorts or []:
                fid = s.get("field_id", "")
                if fid not in valid_field_ids:
                    raise ValueError(f"Unknown field_id in sort: {fid}")
        return await self._repo.list_records_filtered(
            table_id, filters=filters, sorts=sorts, cursor=cursor, limit=limit
        )

    async def update_record_values(self, record_id: str, values: BitableRecord) -> Record | None:
        """Update a record's values (full replace). Triggers recalc for affected formulas."""
        record = await self._repo.update_record_values(record_id, values)
        if record is not None:
            await self._trigger_recalc_for_affected_fields(record.table_id, record.id)
        return record

    async def delete_record(self, record_id: str) -> bool:
        """Delete a record and clean up any attachment/image files."""
        record = await self._repo.get_record(record_id)
        if record is None:
            return False
        await self._cleanup_attachment_files(record.table_id, [record])
        return await self._repo.delete_record(record_id)

    async def delete_records_by_table(self, table_id: str) -> int:
        """Delete all records in a table and clean up attachment/image files.

        P2 #18: only fetches records when the table has attachment/image fields,
        and processes them in batches to avoid OOM on large tables.
        """
        # Check if cleanup is needed at all before loading any records.
        fields = await self._repo.list_fields(table_id)
        has_attachments = any(
            f.field_type in (FieldType.attachment, FieldType.image) for f in fields
        )

        if has_attachments:
            # Batch-fetch records to clean up physical files (P2 #18: avoid OOM).
            batch_size = 500
            cursor: str | None = None
            while True:
                batch, next_cursor = await self._repo.list_records(
                    table_id, cursor=cursor, limit=batch_size
                )
                if batch:
                    await self._cleanup_attachment_files(table_id, batch)
                if next_cursor is None:
                    break
                cursor = next_cursor

        return await self._repo.delete_records_by_table(table_id)

    async def _cleanup_attachment_files(
        self,
        table_id: str,
        records: list[Record],
    ) -> None:
        """Delete physical files for attachment/image fields on the given records.

        Best-effort: file deletion failures are logged but do not block
        record deletion (the file may already be gone).
        """
        fields = await self._repo.list_fields(table_id)
        attachment_field_ids = {
            f.id for f in fields if f.field_type in (FieldType.attachment, FieldType.image)
        }
        if not attachment_field_ids:
            return

        upload_dir = Path(os.environ.get("AGENTKIT_BITABLE_UPLOAD_DIR", "data/uploads/bitable"))

        for record in records:
            for field_id in attachment_field_ids:
                value = record.values.get(field_id)
                if not value or not isinstance(value, list):
                    continue
                for file_meta in value:
                    if not isinstance(file_meta, dict):
                        continue
                    stored_name = file_meta.get("stored_name")
                    if not stored_name:
                        continue
                    file_path = upload_dir / str(stored_name)
                    try:
                        if file_path.exists():
                            file_path.unlink()
                    except OSError as exc:
                        # File lost or permission issue — don't block record deletion
                        logger.warning("Failed to delete attachment file %s: %s", file_path, exc)

    # ── Upsert (KTD8: jsonb_set per agent field) ───────────

    async def upsert_records(
        self,
        table_id: str,
        records: list[BitableRecord],
        primary_key_field_id: str,
    ) -> dict[str, int]:
        """Upsert records by primary key using jsonb_set (KTD8).

        For each record:
        - If a record with the same PK value exists: update agent-owned fields
          via jsonb_set (user-owned fields are never touched).
        - If not: insert a new record.

        Returns {"inserted": N, "updated": N, "skipped": N}.

        P1 #14: batch-queries existing records by PK in a single SELECT,
        eliminating the N+1 query pattern.
        """
        if not primary_key_field_id:
            raise ValueError("primary_key_field_id is required for upsert")

        # Get agent-owned field IDs for this table
        fields = await self._repo.list_fields(table_id)
        agent_field_ids = {f.id for f in fields if f.owner == FieldOwner.agent}

        # Partition records into insert vs update lists, collecting PK values.
        to_insert: list[BitableRecord] = []
        to_update: list[tuple[BitableRecord, str]] = []  # (values, existing_record_id)
        skipped = 0

        # Collect all non-None PK values for batch lookup.
        pk_values_by_str: dict[str, BitableRecord] = {}
        for rec_values in records:
            pk_value = rec_values.get(primary_key_field_id)
            if pk_value is None:
                skipped += 1
                continue
            pk_values_by_str[str(pk_value)] = rec_values

        # Batch SELECT all existing records by PK (P1 #14: 1 query, not N).
        existing_map = await self._repo.find_records_by_pk_batch(
            table_id, primary_key_field_id, list(pk_values_by_str.keys())
        )

        for pk_str, rec_values in pk_values_by_str.items():
            existing = existing_map.get(pk_str)
            if existing is None:
                to_insert.append(rec_values)
            else:
                to_update.append((rec_values, existing.id))

        # Batch insert new records.
        inserted = 0
        for rec_values in to_insert:
            record = await self._repo.create_record(table_id, values=rec_values)
            await self._trigger_recalc_for_affected_fields(table_id, record.id)
            inserted += 1

        # Update existing records (agent-owned fields only via jsonb_set).
        updated = 0
        for rec_values, existing_id in to_update:
            agent_values = {fid: val for fid, val in rec_values.items() if fid in agent_field_ids}
            if agent_values:
                await self._repo.upsert_record_agent_fields(existing_id, agent_values)
            await self._trigger_recalc_for_affected_fields(table_id, existing_id)
            updated += 1

        return {"inserted": inserted, "updated": updated, "skipped": skipped}

    # ── Views ───────────────────────────────────────────────

    async def create_view(
        self,
        table_id: str,
        name: str,
        view_type: ViewType = ViewType.grid,
        config: dict[str, object] | None = None,
    ) -> View:
        return await self._repo.create_view(
            table_id=table_id,
            name=name,
            view_type=view_type,
            config=config,
        )

    async def list_views(self, table_id: str) -> list[View]:
        return await self._repo.list_views(table_id)

    async def update_view(self, view_id: str, **kwargs: object) -> View | None:
        return await self._repo.update_view(view_id, **kwargs)

    async def get_view(self, view_id: str) -> View | None:
        return await self._repo.get_view(view_id)

    # ── Recalc (U3: formula recalc pipeline) ────────────────

    async def _trigger_recalc_for_affected_fields(self, table_id: str, record_id: str) -> None:
        """Detect formula fields affected by a record write and enqueue recalc.

        Finds all formula fields in the table, checks which ones depend on
        the fields that were just written, and enqueues recalc tasks.
        For simplicity (v1), we enqueue recalc for ALL formula fields in the
        table for this record — the worker will evaluate them in topological
        order. The ON CONFLICT DO NOTHING constraint deduplicates.
        """
        fields = await self._repo.list_fields(table_id)
        formula_fields = [f for f in fields if f.field_type == FieldType.formula]

        for f in formula_fields:
            await self._repo.enqueue_recalc(table_id, record_id, f.id)

    async def trigger_recalc(
        self, table_id: str, record_id: str, field_id: str
    ) -> RecalcTask | None:
        """Enqueue a formula recalc task for a specific field."""
        return await self._repo.enqueue_recalc(table_id, record_id, field_id)

    async def get_pending_recalc_tasks(self, limit: int = 50) -> list[RecalcTask]:
        return await self._repo.get_pending_recalc_tasks(limit=limit)

    async def reset_stale_recalc_tasks(self, stale_threshold: float = 600.0) -> int:
        """Reset 'calculating' tasks older than threshold back to 'pending' (crash recovery)."""
        return await self._repo.reset_stale_recalc_tasks(stale_threshold=stale_threshold)

    async def process_recalc_task(self, task: RecalcTask) -> None:
        """Process a single recalc task (used by the worker or synchronously).

        Marks the task as calculating, evaluates the formula, writes the result,
        and marks as done/error.
        """
        from agentkit.bitable.recalc_worker import RecalcWorker

        # ponytail: We reuse RecalcWorker._process_task via a temporary worker
        # instance. This is a known ceiling — the worker caches engines per table,
        # and creating a new worker each call loses the cache. For synchronous
        # processing in tests this is fine. For production, use the long-lived
        # worker started in app.py lifespan.
        worker = RecalcWorker(self._db, self)
        await worker.process_task(task)
        worker.invalidate_engine(task.table_id)
