# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
# SPDX-License-Identifier: AGPL-3.0
"""Local Client for OpenViking.

Implements BaseClient interface using direct service calls (embedded mode).
"""

from typing import Any, Dict, List, Optional, Union

from openviking.server.identity import RequestContext, Role
from openviking.service import OpenVikingService
from openviking.telemetry import TelemetryRequest
from openviking.telemetry.execution import (
    attach_telemetry_payload,
    run_with_telemetry,
)
from openviking.utils.search_filters import merge_time_filter
from openviking_cli.client.base import BaseClient
from openviking_cli.session.user_id import UserIdentifier
from openviking_cli.utils import run_async


def _to_jsonable(value: Any) -> Any:
    """Convert internal objects into JSON-serializable values."""
    to_dict = getattr(value, "to_dict", None)
    if callable(to_dict):
        return to_dict()
    if isinstance(value, list):
        return [_to_jsonable(item) for item in value]
    if isinstance(value, dict):
        return {k: _to_jsonable(v) for k, v in value.items()}
    return value


def _resolve_search_filter(
    filter: Optional[Dict[str, Any]],
    since: Optional[str],
    until: Optional[str],
    time_field: Optional[str],
) -> Optional[Dict[str, Any]]:
    """Merge optional retrieval time bounds into the metadata filter."""
    return merge_time_filter(
        filter,
        since=since,
        until=until,
        time_field=time_field,
    )


class LocalClient(BaseClient):
    """Local Client for OpenViking (embedded mode).

    Implements BaseClient interface using direct service calls.
    """

    def __init__(
        self,
        path: Optional[str] = None,
        user: Optional[UserIdentifier] = None,
    ):
        """Initialize LocalClient.

        Args:
            path: Local storage path (overrides ov.conf storage path)
            user: Explicit user/account/agent identity for embedded mode
        """
        self._service = OpenVikingService(
            path=path,
            user=user or UserIdentifier.the_default_user(),
        )
        self._user = self._service.user
        self._ctx = RequestContext(user=self._user, role=Role.USER)

    @property
    def service(self) -> OpenVikingService:
        """Get the underlying service instance."""
        return self._service

    # ============= Lifecycle =============

    async def initialize(self) -> None:
        """Initialize the local client."""
        await self._service.initialize()

    async def close(self) -> None:
        """Close the local client."""
        await self._service.close()

    # ============= Resource Management =============

    async def add_resource(
        self,
        path: str,
        to: Optional[str] = None,
        parent: Optional[str] = None,
        reason: str = "",
        instruction: str = "",
        wait: bool = False,
        timeout: Optional[float] = None,
        build_index: bool = True,
        summarize: bool = False,
        telemetry: TelemetryRequest = False,
        watch_interval: float = 0,
        **kwargs,
    ) -> Dict[str, Any]:
        """Add resource to OpenViking."""
        if to and parent:
            raise ValueError("Cannot specify both 'to' and 'parent' at the same time.")

        execution = await run_with_telemetry(
            operation="resources.add_resource",
            telemetry=telemetry,
            fn=lambda: self._service.resources.add_resource(
                path=path,
                ctx=self._ctx,
                to=to,
                parent=parent,
                reason=reason,
                instruction=instruction,
                wait=wait,
                timeout=timeout,
                build_index=build_index,
                summarize=summarize,
                watch_interval=watch_interval,
                **kwargs,
            ),
        )
        return attach_telemetry_payload(
            execution.result,
            execution.telemetry,
        )

    async def add_skill(
        self,
        data: Any,
        wait: bool = False,
        timeout: Optional[float] = None,
        telemetry: TelemetryRequest = False,
    ) -> Dict[str, Any]:
        """Add skill to OpenViking."""
        execution = await run_with_telemetry(
            operation="resources.add_skill",
            telemetry=telemetry,
            fn=lambda: self._service.resources.add_skill(
                data=data,
                ctx=self._ctx,
                wait=wait,
                timeout=timeout,
            ),
        )
        return attach_telemetry_payload(
            execution.result,
            execution.telemetry,
        )

    async def wait_processed(self, timeout: Optional[float] = None) -> Dict[str, Any]:
        """Wait for all processing to complete."""
        return await self._service.resources.wait_processed(timeout=timeout)

    async def reindex(
        self,
        uri: str,
        mode: str = "vectors_only",
        wait: bool = True,
    ) -> Dict[str, Any]:
        """Reindex semantic/vector artifacts for a URI."""
        return await self._service.reindex(
            uri=uri,
            mode=mode,
            wait=wait,
        )

    async def build_index(self, resource_uris: Union[str, List[str]], **kwargs) -> Dict[str, Any]:
        """Manually trigger index building."""
        if isinstance(resource_uris, str):
            resource_uris = [resource_uris]
        return await self._service.resources.build_index(resource_uris, ctx=self._ctx, **kwargs)

    async def summarize(self, resource_uris: Union[str, List[str]], **kwargs) -> Dict[str, Any]:
        """Manually trigger summarization."""
        if isinstance(resource_uris, str):
            resource_uris = [resource_uris]
        return await self._service.resources.summarize(resource_uris, ctx=self._ctx, **kwargs)

    # ============= File System =============

    async def ls(
        self,
        uri: str,
        simple: bool = False,
        recursive: bool = False,
        output: str = "original",
        abs_limit: int = 256,
        show_all_hidden: bool = False,
    ) -> List[Any]:
        """List directory contents."""
        return await self._service.fs.ls(
            uri,
            ctx=self._ctx,
            simple=simple,
            recursive=recursive,
            output=output,
            abs_limit=abs_limit,
            show_all_hidden=show_all_hidden,
        )

    async def tree(
        self,
        uri: str,
        output: str = "original",
        abs_limit: int = 128,
        show_all_hidden: bool = False,
        node_limit: int = 1000,
    ) -> List[Dict[str, Any]]:
        """Get directory tree."""
        return await self._service.fs.tree(
            uri,
            ctx=self._ctx,
            output=output,
            abs_limit=abs_limit,
            show_all_hidden=show_all_hidden,
            node_limit=node_limit,
        )

    async def stat(self, uri: str) -> Dict[str, Any]:
        """Get resource status."""
        return await self._service.fs.stat(uri, ctx=self._ctx)

    async def mkdir(self, uri: str, description: Optional[str] = None) -> None:
        """Create directory."""
        await self._service.fs.mkdir(uri, ctx=self._ctx, description=description)

    async def rm(self, uri: str, recursive: bool = False) -> None:
        """Remove resource."""
        await self._service.fs.rm(uri, ctx=self._ctx, recursive=recursive)

    async def mv(self, from_uri: str, to_uri: str) -> None:
        """Move resource."""
        await self._service.fs.mv(from_uri, to_uri, ctx=self._ctx)

    # ============= Content Reading =============

    async def read(self, uri: str, offset: int = 0, limit: int = -1) -> str:
        """Read file content.

        Args:
            uri: Viking URI
            offset: Starting line number (0-indexed). Default 0.
            limit: Number of lines to read. -1 means read to end. Default -1.
        """
        return await self._service.fs.read(uri, ctx=self._ctx, offset=offset, limit=limit)

    async def abstract(self, uri: str) -> str:
        """Read L0 abstract."""
        return await self._service.fs.abstract(uri, ctx=self._ctx)

    async def overview(self, uri: str) -> str:
        """Read L1 overview."""
        return await self._service.fs.overview(uri, ctx=self._ctx)

    async def write(
        self,
        uri: str,
        content: str,
        mode: str = "replace",
        wait: bool = False,
        timeout: Optional[float] = None,
        telemetry: TelemetryRequest = False,
    ) -> Dict[str, Any]:
        """Write text content to an existing file and refresh semantics/vectors."""
        execution = await run_with_telemetry(
            operation="content.write",
            telemetry=telemetry,
            fn=lambda: self._service.fs.write(
                uri=uri,
                content=content,
                ctx=self._ctx,
                mode=mode,
                wait=wait,
                timeout=timeout,
            ),
        )
        return attach_telemetry_payload(
            execution.result,
            execution.telemetry,
        )

    # ============= Search =============

    async def find(
        self,
        query: str,
        target_uri: Union[str, List[str]] = "",
        limit: int = 10,
        score_threshold: Optional[float] = None,
        filter: Optional[Dict[str, Any]] = None,
        telemetry: TelemetryRequest = False,
        since: Optional[str] = None,
        until: Optional[str] = None,
        time_field: Optional[str] = None,
        level: Optional[List[int]] = None,
    ) -> Any:
        """Semantic search without session context."""
        resolved_filter = _resolve_search_filter(filter, since, until, time_field)
        execution = await run_with_telemetry(
            operation="search.find",
            telemetry=telemetry,
            fn=lambda: self._service.search.find(
                query=query,
                ctx=self._ctx,
                target_uri=target_uri,
                limit=limit,
                score_threshold=score_threshold,
                filter=resolved_filter,
                level=level,
            ),
        )
        return attach_telemetry_payload(
            execution.result,
            execution.telemetry,
        )

    async def search(
        self,
        query: str,
        target_uri: Union[str, List[str]] = "",
        session_id: Optional[str] = None,
        limit: int = 10,
        score_threshold: Optional[float] = None,
        filter: Optional[Dict[str, Any]] = None,
        telemetry: TelemetryRequest = False,
        since: Optional[str] = None,
        until: Optional[str] = None,
        time_field: Optional[str] = None,
        level: Optional[List[int]] = None,
    ) -> Any:
        """Semantic search with optional session context."""
        resolved_filter = _resolve_search_filter(filter, since, until, time_field)

        async def _search():
            session = None
            if session_id:
                session = self._service.sessions.session(self._ctx, session_id)
                await session.load()
            return await self._service.search.search(
                query=query,
                ctx=self._ctx,
                target_uri=target_uri,
                session=session,
                limit=limit,
                score_threshold=score_threshold,
                filter=resolved_filter,
                level=level,
            )

        execution = await run_with_telemetry(
            operation="search.search",
            telemetry=telemetry,
            fn=_search,
        )
        return attach_telemetry_payload(
            execution.result,
            execution.telemetry,
        )

    async def grep(
        self,
        uri: str,
        pattern: str,
        case_insensitive: bool = False,
        node_limit: Optional[int] = None,
        exclude_uri: Optional[str] = None,
        level_limit: int = 5,
    ) -> Dict[str, Any]:
        """Content search with pattern."""
        return await self._service.fs.grep(
            uri,
            pattern,
            ctx=self._ctx,
            case_insensitive=case_insensitive,
            node_limit=node_limit,
            exclude_uri=exclude_uri,
            level_limit=level_limit,
        )

    async def glob(self, pattern: str, uri: str = "viking://") -> Dict[str, Any]:
        """File pattern matching."""
        return await self._service.fs.glob(pattern, ctx=self._ctx, uri=uri)

    # ============= Relations =============

    async def relations(self, uri: str) -> List[Any]:
        """Get relations for a resource."""
        return await self._service.relations.relations(uri, ctx=self._ctx)

    async def link(self, from_uri: str, to_uris: Union[str, List[str]], reason: str = "") -> None:
        """Create link between resources."""
        await self._service.relations.link(from_uri, to_uris, ctx=self._ctx, reason=reason)

    async def unlink(self, from_uri: str, to_uri: str) -> None:
        """Remove link between resources."""
        await self._service.relations.unlink(from_uri, to_uri, ctx=self._ctx)

    # ============= Sessions =============

    async def create_session(
        self, session_id: Optional[str] = None, telemetry: TelemetryRequest = False
    ) -> Dict[str, Any]:
        """Create a new session.

        Args:
            session_id: Optional session ID. If provided, creates a session with the given ID.
                       If None, creates a new session with auto-generated ID.
        """
        execution = await run_with_telemetry(
            operation="session.create",
            telemetry=telemetry,
            fn=lambda: self._create_session_impl(session_id),
        )
        return attach_telemetry_payload(
            execution.result,
            execution.telemetry,
        )

    async def _create_session_impl(self, session_id: Optional[str]) -> Dict[str, Any]:
        await self._service.initialize_user_directories(self._ctx)
        await self._service.initialize_agent_directories(self._ctx)
        session = await self._service.sessions.create(self._ctx, session_id)
        return {
            "session_id": session.session_id,
            "user": session.user.to_dict(),
        }

    async def list_sessions(self) -> List[Any]:
        """List all sessions."""
        return await self._service.sessions.sessions(self._ctx)

    async def get_session(self, session_id: str, *, auto_create: bool = False) -> Dict[str, Any]:
        """Get session details."""
        session = await self._service.sessions.get(session_id, self._ctx, auto_create=auto_create)
        result = session.meta.to_dict()
        result["user"] = session.user.to_dict()
        return result

    async def get_session_context(
        self, session_id: str, token_budget: int = 128_000
    ) -> Dict[str, Any]:
        """Get assembled session context."""
        session = self._service.sessions.session(self._ctx, session_id)
        await session.load()
        result = await session.get_session_context(token_budget=token_budget)
        return _to_jsonable(result)

    async def get_session_archive(self, session_id: str, archive_id: str) -> Dict[str, Any]:
        """Get one completed archive for a session."""
        session = self._service.sessions.session(self._ctx, session_id)
        await session.load()
        result = await session.get_session_archive(archive_id)
        return _to_jsonable(result)

    async def delete_session(self, session_id: str) -> None:
        """Delete a session."""
        await self._service.sessions.delete(session_id, self._ctx)

    async def commit_session(
        self, session_id: str, telemetry: TelemetryRequest = False
    ) -> Dict[str, Any]:
        """Commit a session (archive and extract memories)."""
        execution = await run_with_telemetry(
            operation="session.commit",
            telemetry=telemetry,
            fn=lambda: self._service.sessions.commit(session_id, self._ctx),
        )
        return attach_telemetry_payload(
            execution.result,
            execution.telemetry,
        )

    async def get_task(self, task_id: str) -> Optional[Dict[str, Any]]:
        """Query background task status."""
        return await self._service.sessions.get_commit_task(task_id, self._ctx)

    async def add_message(
        self,
        session_id: str,
        role: str,
        content: Optional[str] = None,
        parts: Optional[List[Dict[str, Any]]] = None,
        created_at: Optional[str] = None,
        role_id: Optional[str] = None,
        telemetry: TelemetryRequest = False,
    ) -> Dict[str, Any]:
        """Add a message to a session.

        Args:
            session_id: Session ID
            role: Message role ("user" or "assistant")
            content: Text content (simple mode, backward compatible)
            parts: Parts array (full Part support mode)
            created_at: Message creation time (ISO format string)
            role_id: Optional explicit actor identity. Omit to derive it from the local context.

        If both content and parts are provided, parts takes precedence.
        """
        execution = await run_with_telemetry(
            operation="session.add_message",
            telemetry=telemetry,
            fn=lambda: self._add_message_impl(
                session_id,
                role,
                content,
                parts,
                created_at,
                role_id,
            ),
        )
        return attach_telemetry_payload(
            execution.result,
            execution.telemetry,
        )

    async def _add_message_impl(
        self,
        session_id: str,
        role: str,
        content: Optional[str],
        parts: Optional[List[Dict[str, Any]]],
        created_at: Optional[str],
        role_id: Optional[str],
    ) -> Dict[str, Any]:
        from openviking.message.part import Part, TextPart, part_from_dict

        session = await self._service.sessions.get(session_id, self._ctx, auto_create=True)

        message_parts: list[Part]
        if parts is not None:
            message_parts = [part_from_dict(p) for p in parts]
        elif content is not None:
            message_parts = [TextPart(text=content)]
        else:
            raise ValueError("Either content or parts must be provided")

        if role_id is None and role == "user":
            role_id = self._ctx.user.user_id
        elif role_id is None and role == "assistant":
            role_id = self._ctx.user.agent_id

        session.add_message(role, message_parts, role_id=role_id, created_at=created_at)
        return {
            "session_id": session_id,
            "message_count": len(session.messages),
        }

    # ============= Pack =============

    async def export_ovpack(
        self,
        uri: str,
        to: str,
        include_vectors: bool = False,
    ) -> str:
        """Export context as .ovpack file."""
        return await self._service.pack.export_ovpack(
            uri,
            to,
            ctx=self._ctx,
            include_vectors=include_vectors,
        )

    async def backup_ovpack(self, to: str, include_vectors: bool = False) -> str:
        """Back up public scopes as a restore-only .ovpack file."""
        return await self._service.pack.backup_ovpack(
            to,
            ctx=self._ctx,
            include_vectors=include_vectors,
        )

    async def import_ovpack(
        self,
        file_path: str,
        parent: str,
        on_conflict: Optional[str] = None,
        vector_mode: Optional[str] = None,
    ) -> str:
        """Import .ovpack file."""
        return await self._service.pack.import_ovpack(
            file_path,
            parent,
            ctx=self._ctx,
            on_conflict=on_conflict,
            vector_mode=vector_mode,
        )

    async def restore_ovpack(
        self,
        file_path: str,
        on_conflict: Optional[str] = None,
        vector_mode: Optional[str] = None,
    ) -> str:
        """Restore backup .ovpack file."""
        return await self._service.pack.restore_ovpack(
            file_path,
            ctx=self._ctx,
            on_conflict=on_conflict,
            vector_mode=vector_mode,
        )

    # ============= Debug =============

    async def check_consistency(self, uri: str) -> Dict[str, Any]:
        """Check filesystem/vector-index consistency for a URI subtree."""
        return await self._service.check_consistency(
            uri=uri,
            ctx=self._ctx,
        )

    async def health(self) -> bool:
        """Check service health."""
        return True  # Local service is always healthy if initialized

    def session(self, session_id: Optional[str] = None, must_exist: bool = False) -> Any:
        """Create a new session or load an existing one.

        Args:
            session_id: Session ID, creates a new session if None.
            must_exist: Whether to raise an error if the session does not exist. Default False.
        Returns:
            Session object if exists, None otherwise.
        """

        session = self._service.sessions.session(self._ctx, session_id)
        if not run_async(session.exists()):
            if must_exist and session_id:
                from openviking_cli.exceptions import NotFoundError

                raise NotFoundError(session_id, "session")
            else:
                run_async(session.ensure_exists())
        return session

    async def session_exists(self, session_id: str) -> bool:
        """Check whether a session exists in storage.

        Args:
            session_id: Session ID to check

        Returns:
            True if the session exists, False otherwise
        """
        session = self._service.sessions.session(self._ctx, session_id)
        return await session.exists()

    def get_status(self) -> Any:
        """Get system status.

        Returns:
            SystemStatus containing health status of all components.
        """
        return self._service.debug.observer.system()

    def is_healthy(self) -> bool:
        """Quick health check (synchronous).

        Returns:
            True if all components are healthy, False otherwise.
        """
        return self._service.debug.observer.is_healthy()

    @property
    def observer(self) -> Any:
        """Get observer service for component status."""
        return self._service.debug.observer
