"""
Service layer cho CuCu Note với SQLite backend.
Full features: memos, attachments, relations, reactions, embeddings, inbox, settings...
"""

from __future__ import annotations

import json
import logging
import os
import re
from datetime import datetime, timezone
from typing import Any, List, Optional

import numpy as np

from . import schemas
from .mongodb import mongodb_client, generate_uid, utc_now, serialize_row, parse_time, format_time
from langchain_openai import OpenAIEmbeddings


def _id_to_str(id_val: Any) -> str:
    """Convert ID từ bất kỳ kiểu gì sang string cho API response."""
    if id_val is None:
        return ""
    return str(id_val)


class InstanceService:
    async def get_instance_info(self) -> schemas.InstanceInfo:
        return schemas.InstanceInfo(mode="dev", version="0.0.1-sqlite")

    async def update_instance_info(self, payload: schemas.InstanceUpdate) -> schemas.InstanceInfo:
        return schemas.InstanceInfo(mode="dev", version="0.0.1-sqlite")

    async def get_instance_setting(self) -> schemas.InstanceSetting:
        return schemas.InstanceSetting()


class AuthService:
    async def sign_in(self, payload: schemas.AuthSignInRequest) -> schemas.AuthSignInResponse:
        """Real SQLite-backed sign in — verifies password and returns a JWT."""
        from common.jwt_auth import verify_password, create_access_token
        from .mongodb import mongodb_client as db

        email = payload.email
        password = payload.password

        if not email or not password:
            raise ValueError("Email and password are required")

        # Try lookup by email first, then by username column
        row = await db.fetch_one(
            "SELECT * FROM cuccu_users WHERE email = ?", (email,)
        )
        if not row:
            row = await db.fetch_one(
                "SELECT * FROM cuccu_users WHERE username = ?", (email,)
            )
        if not row:
            raise ValueError("Incorrect email/username or password")

        if not verify_password(password, row["password_hash"]):
            raise ValueError("Incorrect email/username or password")

        user_id = str(row["id"])
        access_token = create_access_token(data={"sub": user_id})
        return schemas.AuthSignInResponse(accessToken=access_token)

    async def sign_up(self, payload: schemas.AuthSignUpRequest) -> schemas.AuthSignUpResponse:
        """Real SQLite-backed sign up — creates user and returns JWT."""
        from datetime import datetime, timezone
        from common.jwt_auth import get_password_hash, create_access_token
        from .mongodb import mongodb_client as db

        email = str(payload.email)
        # Check duplicate email
        existing = await db.fetch_one(
            "SELECT id FROM cuccu_users WHERE email = ?", (email,)
        )
        if existing:
            raise ValueError("Email already registered")

        hashed = get_password_hash(payload.password)
        username = email.split("@")[0]
        now = datetime.now(timezone.utc).isoformat()

        cursor = await db.execute(
            "INSERT INTO cuccu_users (username, email, password_hash, nickname, created_at) VALUES (?, ?, ?, ?, ?)",
            (username, email, hashed, username, now),
        )
        user_id = str(cursor.lastrowid)
        access_token = create_access_token(data={"sub": user_id})
        return schemas.AuthSignUpResponse(user_id=user_id)

    async def get_me(self, token: str | None = None) -> schemas.AuthMeResponse:
        from config import DISABLE_AUTH
        from common.jwt_auth import decode_token
        from .mongodb import mongodb_client as db

        if DISABLE_AUTH:
            logging.warning("⚠️  DISABLE_AUTH=true -> returning stub user for memo frontend")
            return schemas.AuthMeResponse(id="1", email="demo@example.com")

        if not token:
            raise ValueError("Missing authentication token")

        try:
            payload = decode_token(token)
            if payload and "sub" in payload:
                user_id = str(payload.get("sub"))
                # Look up real user from DB
                row = await db.fetch_one(
                    "SELECT id, email, username, nickname FROM cuccu_users WHERE id = ?",
                    (user_id,)
                )
                if row:
                    return schemas.AuthMeResponse(
                        id=str(row["id"]),
                        email=row["email"],
                        username=row["username"] or row["email"].split("@")[0],
                    )
                # Fallback if user not found
                return schemas.AuthMeResponse(
                    id=user_id,
                    email=f"user{user_id}@local",
                    username=f"user{user_id}",
                )
            else:
                raise ValueError("Invalid token: missing user ID")
        except Exception as e:
            logging.error(f"❌ JWT token verification failed: {e}")
            raise ValueError(f"Invalid authentication token: {str(e)}") from e


class UserService:
    async def list_users(self) -> List[schemas.UserResponse]:
        return []

    async def create_user(self, payload: schemas.UserCreate) -> schemas.UserResponse:
        email = payload.email or "demo@example.com"
        return schemas.UserResponse(id="1", email=email, nickname=None)

    async def get_user(self, user_id: str) -> schemas.UserResponse:
        return schemas.UserResponse(id=user_id, email="demo@example.com", nickname=None)

    async def update_user(self, user_id: str, payload: schemas.UserUpdate) -> schemas.UserResponse:
        return schemas.UserResponse(id=user_id, email=payload.email or "demo@example.com", nickname=payload.nickname)

    async def get_user_stats(self, user_id: str, workspace_id: str | None = None) -> schemas.UserStatsResponse:
        """Get user statistics including tag counts and memo timestamps.

        Args:
            user_id: The user ID
            workspace_id: Optional workspace filter (PERSONAL, AI_SALES_CRM). If provided, only count memos in that workspace.
        """
        query = "SELECT content, created_at, payload FROM memos WHERE creator_id = ? AND row_status != 'ARCHIVED'"
        params: list = [user_id]

        if workspace_id:
            query += " AND workspace_id = ?"
            params.append(workspace_id)

        rows = await mongodb_client.fetch_all(query, tuple(params))

        tag_count: dict[str, int] = {}
        memo_timestamps: list[str] = []

        for row in rows:
            created_at = row["created_at"]
            if created_at:
                dt = parse_time(created_at)
                if dt:
                    memo_timestamps.append(dt.isoformat())
                else:
                    memo_timestamps.append(str(created_at))

            content = row["content"] or ""
            if content:
                matches = re.findall(r'#([^\s#]+)', content)
                for tag in matches:
                    tag_count[tag] = tag_count.get(tag, 0) + 1

            payload_data = {}
            if row["payload"]:
                try:
                    payload_data = json.loads(row["payload"]) if isinstance(row["payload"], str) else row["payload"]
                except Exception:
                    pass
            payload_tags = payload_data.get("tags", [])
            for tag in payload_tags:
                if tag and tag not in tag_count:
                    tag_count[tag] = tag_count.get(tag, 0) + 1

        return schemas.UserStatsResponse(
            tag_count=tag_count,
            memo_display_timestamps=memo_timestamps,
        )


class MemoService:
    """Full-featured Memo service with SQLite backend."""

    async def list_memos(
        self,
        user_id: str | None = None,
        creator_id: str | None = None,
        tag: str | None = None,
        visibility: str | None = None,
        pinned: bool | None = None,
        row_status: str | None = None,
        start_date: datetime | None = None,
        end_date: datetime | None = None,
        workspace_id: str | None = None,
        is_read: bool | None = None,
    ) -> List[schemas.MemoResponse]:
        """List memos with filtering."""
        query_parts = ["SELECT * FROM memos WHERE 1=1"]
        params: list = []

        if creator_id:
            if user_id and creator_id == user_id:
                query_parts.append("AND creator_id = ?")
                params.append(creator_id)
            else:
                query_parts.append("AND creator_id = '__none__'")
        elif user_id and user_id != "anonymous":
            query_parts.append("AND creator_id = ?")
            params.append(user_id)
        else:
            query_parts.append("AND creator_id = '__none__'")

        if workspace_id:
            query_parts.append("AND workspace_id = ?")
            params.append(workspace_id)
        if is_read is not None:
            query_parts.append("AND is_read = ?")
            params.append(1 if is_read else 0)

        if visibility:
            query_parts.append("AND visibility = ?")
            params.append(visibility)
        if pinned is not None:
            query_parts.append("AND pinned = ?")
            params.append(1 if pinned else 0)
        if row_status:
            query_parts.append("AND row_status = ?")
            params.append(row_status)
        else:
            query_parts.append("AND row_status != 'ARCHIVED'")

        query_parts.append("AND parent IS NULL")

        if start_date:
            query_parts.append("AND created_at >= ?")
            params.append(start_date.isoformat())
        if end_date:
            query_parts.append("AND created_at <= ?")
            params.append(end_date.isoformat())

        query_parts.append("ORDER BY created_at DESC")
        query_parts.append("LIMIT 100")

        sql = " ".join(query_parts)
        rows = await mongodb_client.fetch_all(sql, tuple(params))

        memo_ids = [str(row["id"]) for row in rows]
        comment_counts = await self._get_comment_counts_batch(memo_ids, user_id)
        memo_reactions = await self._get_reactions_batch(memo_ids)

        memos: list[schemas.MemoResponse] = []
        for row in rows:
            memo_response = self._row_to_response(row)
            memo_response.comment_count = comment_counts.get(str(row["id"]), 0)
            memo_response.reactions = memo_reactions.get(str(row["id"]), [])
            memos.append(memo_response)

        if tag:
            filtered = []
            safe_tag = re.escape(tag)
            tag_pattern = re.compile(rf"#?{safe_tag}(\s|$|[^a-zA-Z0-9_])", re.IGNORECASE)
            for memo in memos:
                tag_match = False
                if tag in memo.tags:
                    tag_match = True
                elif tag_pattern.search(memo.content):
                    tag_match = True
                if tag_match:
                    filtered.append(memo)
            return filtered

        return memos

    async def list_comments(self, memo_id: str, user_id: str | None = None) -> List[schemas.MemoResponse]:
        sql = """
            SELECT * FROM memos
            WHERE parent = ? AND row_status != 'ARCHIVED'
            ORDER BY created_at ASC
            LIMIT 100
        """
        rows = await mongodb_client.fetch_all(sql, (memo_id,))

        if not rows:
            return []

        comment_ids = [str(row["id"]) for row in rows]
        comment_counts = await self._get_comment_counts_batch(comment_ids, user_id)
        memo_reactions = await self._get_reactions_batch(comment_ids)

        comments: List[schemas.MemoResponse] = []
        for row in rows:
            comment_response = self._row_to_response(row)
            comment_response.comment_count = comment_counts.get(str(row["id"]), 0)
            comment_response.reactions = memo_reactions.get(str(row["id"]), [])
            comments.append(comment_response)
        return comments

    async def create_memo(
        self,
        payload: schemas.MemoCreate,
        user_id: str | None = None,
        parent_id: str | None = None,
        anonymous_id: str | None = None,
    ) -> schemas.MemoResponse:
        """Create a memo for the specified user."""
        creator = user_id or "anonymous"
        now = payload.create_time or utc_now()
        if now.tzinfo is None:
            now = now.replace(tzinfo=timezone.utc)

        # Determine workspace_id
        final_workspace_id = payload.workspace_id or "PERSONAL"

        # Determine is_read: default True for user-created memos, False for inbound/webhook
        final_is_read = payload.is_read if payload.is_read is not None else True

        # If this is a comment (parent_id set), inherit workspace from parent
        if parent_id:
            parent_row = await mongodb_client.fetch_one(
                "SELECT workspace_id FROM memos WHERE uid = ? OR id = ?",
                (parent_id, int(parent_id) if parent_id.isdigit() else None)
            )
            if parent_row and parent_row["workspace_id"]:
                final_workspace_id = parent_row["workspace_id"]

        memo_payload = {
            "tags": payload.tags or [],
            "hasLink": False,
            "hasTaskList": False,
            "hasCode": False,
            "hasIncompleteTasks": False,
        }

        uid = generate_uid()

        sql = """
            INSERT INTO memos (
                uid, creator_id, content, visibility, pinned, payload, row_status,
                created_at, updated_at, is_completed, completed_at,
                deadline, priority, reminder_at, parent, anonymous_id, workspace_id, is_read
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """
        await mongodb_client.execute(
            sql,
            (
                uid, creator, payload.content, payload.visibility or "PRIVATE",
                0, json.dumps(memo_payload), "NORMAL",
                now.isoformat(), now.isoformat(), 0, None,
                payload.deadline.isoformat() if payload.deadline else None,
                payload.priority,
                payload.reminder_at.isoformat() if payload.reminder_at else None,
                parent_id, anonymous_id, final_workspace_id, 1 if final_is_read else 0
            )
        )

        # Get the inserted row
        row = await mongodb_client.fetch_one("SELECT * FROM memos WHERE uid = ?", (uid,))

        import asyncio
        if row:
            asyncio.create_task(self._create_embedding(str(row["id"]), payload.content, payload.tags or [], user_id=user_id))

            # If this is a comment (parent_id set), create activity and notification
            if parent_id:
                asyncio.create_task(self._create_comment_activity_and_notification(
                    memo_id=str(row["id"]),
                    parent_id=parent_id,
                    commenter_id=creator
                ))

            return self._row_to_response(row)

        raise ValueError("Failed to create memo")

    async def get_memo(
        self,
        memo_id: str,
        user_id: str | None = None,
        workspace_id: str | None = None,
    ) -> schemas.MemoResponse:
        """Get a memo by ID. Access control: owner or public."""
        row = None

        if memo_id.isdigit():
            row = await mongodb_client.fetch_one("SELECT * FROM memos WHERE id = ?", (int(memo_id),))
        if not row:
            row = await mongodb_client.fetch_one("SELECT * FROM memos WHERE uid = ?", (memo_id,))

        if not row:
            raise ValueError(f"Memo {memo_id} not found")

        # Workspace isolation check
        if workspace_id and row["workspace_id"] != workspace_id:
            raise ValueError(f"Access denied: memo not in workspace '{workspace_id}'")

        memo_creator = row["creator_id"] or "anonymous"
        memo_visibility = row["visibility"] or "PRIVATE"

        if memo_creator == user_id:
            response = self._row_to_response(row)
            response.comment_count = await self._get_comment_count(str(row["id"]), memo_visibility, user_id)
            response.reactions = await self._get_reactions(str(row["id"]))
            return response

        if memo_visibility == "PUBLIC":
            response = self._row_to_response(row)
            response.comment_count = await self._get_comment_count(str(row["id"]), memo_visibility, user_id)
            response.reactions = await self._get_reactions(str(row["id"]))
            return response

        if memo_visibility == "PROTECTED":
            if user_id and user_id != "anonymous":
                response = self._row_to_response(row)
                response.comment_count = await self._get_comment_count(str(row["id"]), memo_visibility, user_id)
                response.reactions = await self._get_reactions(str(row["id"]))
                return response
            raise ValueError("Access denied: PROTECTED memo requires authentication")

        raise ValueError(f"Access denied to memo {memo_id}")

    async def _get_comment_counts_batch(self, memo_ids: List[str], user_id: str | None = None) -> dict[str, int]:
        """Get comment counts for multiple memos in a single query."""
        if not memo_ids:
            return {}

        placeholders = ",".join("?" * len(memo_ids))
        sql = f"""
            SELECT parent, COUNT(*) as count
            FROM memos
            WHERE parent IN ({placeholders}) AND row_status != 'ARCHIVED'
            GROUP BY parent
        """
        rows = await mongodb_client.fetch_all(sql, tuple(memo_ids))

        return {row["parent"]: row["count"] for row in rows}

    async def _get_comment_count(self, memo_id: str, memo_visibility: str, user_id: str | None = None) -> int:
        """Get comment count for a memo based on visibility and user access."""
        query_parts = ["SELECT COUNT(*) FROM memos WHERE parent = ? AND row_status != 'ARCHIVED'"]
        params: List[Any] = [memo_id]

        if memo_visibility == "PUBLIC":
            query_parts.append("AND visibility = 'PUBLIC'")
        elif memo_visibility == "PROTECTED" and user_id:
            query_parts.append("AND (visibility = 'PUBLIC' OR visibility = 'PROTECTED')")
        elif memo_visibility == "PRIVATE":
            if not user_id:
                return 0

        sql = " ".join(query_parts)
        row = await mongodb_client.fetch_one(sql, tuple(params))
        return row[0] if row else 0

    async def _get_reactions_batch(self, memo_ids: List[str]) -> dict[str, List[schemas.ReactionResponse]]:
        """Get reactions for multiple memos in a single query."""
        if not memo_ids:
            return {}

        placeholders = ",".join("?" * len(memo_ids))
        sql = f"""
            SELECT content_id, creator_id, reaction_type
            FROM reactions
            WHERE content_id IN ({placeholders})
        """
        rows = await mongodb_client.fetch_all(sql, tuple(memo_ids))

        results: dict[str, List[schemas.ReactionResponse]] = {}
        for row in rows:
            content_id = row["content_id"]
            if content_id not in results:
                results[content_id] = []
            results[content_id].append(
                schemas.ReactionResponse(
                    name=f"memos/{content_id}/reactions/{row['reaction_type']}",
                    creator=f"users/{row['creator_id']}",
                    contentId=f"memos/{content_id}",
                    reactionType=row["reaction_type"],
                )
            )
        return results

    async def _get_reactions(self, memo_id: str) -> List[schemas.ReactionResponse]:
        """Get reactions for a single memo."""
        reactions_map = await self._get_reactions_batch([memo_id])
        return reactions_map.get(memo_id, [])

    async def get_memo_versions(
        self,
        memo_id: str,
        user_id: str | None = None,
    ) -> schemas.MemoVersionResponse:
        """Get version history for a memo."""
        row = None
        if memo_id.isdigit():
            row = await mongodb_client.fetch_one("SELECT * FROM memos WHERE id = ?", (int(memo_id),))
        if not row:
            row = await mongodb_client.fetch_one("SELECT * FROM memos WHERE uid = ?", (memo_id,))

        if not row:
            raise ValueError(f"Memo {memo_id} not found")

        versions_data = []
        payload = row["payload"]
        if payload:
            try:
                payload_dict = json.loads(payload) if isinstance(payload, str) else payload
                versions_data = payload_dict.get("versions", [])
            except Exception:
                pass

        versions = [
            schemas.MemoVersion(
                version=v.get("version", i + 1),
                content=v.get("content", ""),
                created_at=parse_time(v.get("created_at")) or utc_now(),
                created_by=v.get("created_by"),
            )
            for i, v in enumerate(versions_data)
        ]

        return schemas.MemoVersionResponse(
            versions=versions,
            total=len(versions),
        )

    async def update_memo(
        self,
        memo_id: str,
        payload: schemas.MemoUpdate,
        user_id: str | None = None,
    ) -> schemas.MemoResponse:
        """Update a memo. Only the owner can update."""
        row = None
        if memo_id.isdigit():
            row = await mongodb_client.fetch_one("SELECT * FROM memos WHERE id = ?", (int(memo_id),))
        if not row:
            row = await mongodb_client.fetch_one("SELECT * FROM memos WHERE uid = ?", (memo_id,))

        if not row:
            raise ValueError(f"Memo {memo_id} not found")

        memo_creator = row["creator_id"] or "anonymous"
        logging.info(f"🔍 update_memo ownership check: memo_id={memo_id}, memo_creator='{memo_creator}', user_id='{user_id}', match={memo_creator == user_id}")
        if memo_creator != user_id:
            raise ValueError(f"Access denied: you don't own memo {memo_id} (creator={memo_creator}, you={user_id})")

        update_fields: dict[str, Any] = {}
        set_clauses: List[str] = []
        params: List[Any] = []

        if payload.content is not None:
            set_clauses.append("content = ?")
            params.append(payload.content)
        if payload.visibility is not None:
            set_clauses.append("visibility = ?")
            params.append(payload.visibility)
        if payload.tags is not None:
            # Need to update payload JSON
            current_payload = {}
            if row["payload"]:
                try:
                    current_payload = json.loads(row["payload"]) if isinstance(row["payload"], str) else row["payload"]
                except Exception:
                    pass
            current_payload["tags"] = payload.tags
            set_clauses.append("payload = ?")
            params.append(json.dumps(current_payload))
        if payload.pinned is not None:
            set_clauses.append("pinned = ?")
            params.append(1 if payload.pinned else 0)
        if payload.row_status is not None:
            set_clauses.append("row_status = ?")
            params.append(payload.row_status)

        if payload.deadline is not None:
            set_clauses.append("deadline = ?")
            params.append(payload.deadline.isoformat() if payload.deadline.tzinfo else payload.deadline.replace(tzinfo=timezone.utc).isoformat())
        if payload.priority is not None:
            set_clauses.append("priority = ?")
            params.append(payload.priority)
        if payload.reminder_at is not None:
            set_clauses.append("reminder_at = ?")
            params.append(payload.reminder_at.isoformat() if payload.reminder_at.tzinfo else payload.reminder_at.replace(tzinfo=timezone.utc).isoformat())
        if payload.is_completed is not None:
            set_clauses.append("is_completed = ?")
            params.append(1 if payload.is_completed else 0)
            if payload.is_completed:
                set_clauses.append("completed_at = ?")
                params.append(utc_now().isoformat())
            else:
                set_clauses.append("completed_at = NULL")
        if payload.completed_at is not None:
            set_clauses.append("completed_at = ?")
            params.append(payload.completed_at.isoformat())
        if payload.is_read is not None:
            set_clauses.append("is_read = ?")
            params.append(1 if payload.is_read else 0)

        set_clauses.append("updated_at = ?")
        params.append(utc_now().isoformat())

        # Save version history when content changes
        if payload.content is not None and payload.content != row["content"]:
            current_payload = {}
            if row["payload"]:
                try:
                    current_payload = json.loads(row["payload"]) if isinstance(row["payload"], str) else row["payload"]
                except Exception:
                    pass
            versions = current_payload.get("versions", [])
            new_version = {
                "version": len(versions) + 1,
                "content": row["content"],
                "created_at": row["updated_at"] or row["created_at"] or utc_now().isoformat(),
                "created_by": memo_creator,
            }
            versions.append(new_version)
            current_payload["versions"] = versions
            set_clauses.append("payload = ?")
            params.append(json.dumps(current_payload))

        if set_clauses:
            sql = f"UPDATE memos SET {', '.join(set_clauses)} WHERE id = ?"
            params.append(row["id"])
            await mongodb_client.execute(sql, tuple(params))

        if payload.content is not None:
            import asyncio
            asyncio.create_task(self._create_embedding(str(row["id"]), payload.content, payload.tags or [], user_id=user_id))

        return await self.get_memo(memo_id, user_id=user_id)

    async def delete_memo(self, memo_id: str, user_id: str | None = None) -> None:
        """Delete a memo. Only the owner can delete."""
        row = None
        if memo_id.isdigit():
            row = await mongodb_client.fetch_one("SELECT * FROM memos WHERE id = ?", (int(memo_id),))
        if not row:
            row = await mongodb_client.fetch_one("SELECT * FROM memos WHERE uid = ?", (memo_id,))

        if not row:
            raise ValueError(f"Memo {memo_id} not found")

        memo_creator = row["creator_id"] or "anonymous"
        logging.info(f"🔍 delete_memo ownership check: memo_id={memo_id}, memo_creator='{memo_creator}', user_id='{user_id}', match={memo_creator == user_id}")
        if memo_creator != user_id:
            raise ValueError(f"Access denied: you don't own memo {memo_id} (creator={memo_creator}, you={user_id})")

        memo_id_int = row["id"]

        await mongodb_client.execute("DELETE FROM reactions WHERE content_id = ?", (str(memo_id_int),))
        await mongodb_client.execute("DELETE FROM memo_embeddings WHERE memo_id = ?", (str(memo_id_int),))
        await mongodb_client.execute("DELETE FROM attachments WHERE memo_id = ?", (str(memo_id_int),))
        await mongodb_client.execute("DELETE FROM memo_relations WHERE memo_id = ? OR related_memo_id = ?", (str(memo_id_int), str(memo_id_int)))
        await mongodb_client.execute("DELETE FROM memos WHERE id = ?", (memo_id_int,))

    async def pin_memo(self, memo_id: str, pinned: bool, user_id: str | None = None) -> schemas.MemoResponse:
        """Pin or unpin a memo."""
        return await self.update_memo(
            memo_id,
            schemas.MemoUpdate(pinned=pinned),
            user_id=user_id,
        )

    async def archive_memo(self, memo_id: str, user_id: str | None = None) -> schemas.MemoResponse:
        """Archive a memo."""
        return await self.update_memo(
            memo_id,
            schemas.MemoUpdate(row_status="ARCHIVED"),
            user_id=user_id,
        )

    async def restore_memo(self, memo_id: str, user_id: str | None = None) -> schemas.MemoResponse:
        """Restore an archived memo."""
        return await self.update_memo(
            memo_id,
            schemas.MemoUpdate(row_status="NORMAL"),
            user_id=user_id,
        )

    def _row_to_response(self, row: aiosqlite.Row) -> schemas.MemoResponse:
        """Convert SQLite row to MemoResponse."""
        payload = {}
        if row["payload"]:
            try:
                payload = json.loads(row["payload"]) if isinstance(row["payload"], str) else row["payload"]
            except Exception:
                payload = {}

        def _fmt_dt(dt_val: Any) -> str | None:
            if dt_val is None:
                return None
            dt = parse_time(dt_val)
            if dt:
                return format_time(dt)
            return str(dt_val)

        created_at_str = _fmt_dt(row["created_at"])
        updated_at_str = _fmt_dt(row["updated_at"])

        creator_id = row["creator_id"] or "anonymous"
        uid = row["uid"] or ""

        deadline_dt = parse_time(row["deadline"]) if row["deadline"] else None
        is_completed = bool(row["is_completed"])
        deadline_status: str | None = None
        deadline_str = _fmt_dt(row["deadline"])

        if deadline_dt is not None:
            now_utc = datetime.now(timezone.utc)
            if is_completed:
                deadline_status = "DONE"
            elif now_utc > deadline_dt:
                deadline_status = "OVERDUE"
            else:
                deadline_status = "PENDING"

        return schemas.MemoResponse(
            id=str(row["id"]),
            uid=uid,
            content=row["content"] or "",
            visibility=row["visibility"] or "PRIVATE",
            tags=payload.get("tags", []),
            pinned=bool(row["pinned"]),
            row_status=row["row_status"] or "NORMAL",
            creator_id=creator_id,
            creator=f"users/{creator_id}",
            name=f"memos/{uid}",
            create_time=created_at_str,
            update_time=updated_at_str,
            display_time=created_at_str,
            parent=row["parent"],
            deadline=deadline_str,
            deadline_status=deadline_status,
            priority=row["priority"],
            reminder_at=_fmt_dt(row["reminder_at"]),
            is_completed=is_completed,
            completed_at=_fmt_dt(row["completed_at"]),
            workspace_id=row["workspace_id"] if "workspace_id" in row.keys() else None,
            is_read=bool(row["is_read"]) if "is_read" in row.keys() else True,
        )

    async def _create_embedding(self, memo_id: str, content: str, tags: List[str], user_id: str | None = None) -> None:
        """Create or update embedding for a memo."""
        try:
            api_key = None
            if user_id:
                from common.memos_core.services import get_user_settings_service
                settings_service = get_user_settings_service()
                api_key = await settings_service.get_openai_api_key(user_id)

            if not api_key:
                api_key = os.getenv("OPENAI_API_KEY")

            if not api_key:
                return

            from common.embedding_service import create_embedding_async
            vec = await create_embedding_async(content, user_id=user_id)

            if not vec:
                return

            now = utc_now()
            date_key = now.strftime("%Y-%m-%d")

            # Upsert embedding
            check = await mongodb_client.fetch_one(
                "SELECT id FROM memo_embeddings WHERE memo_id = ?",
                (memo_id,)
            )
            embedding_json = json.dumps(vec)

            if check:
                await mongodb_client.execute(
                    """
                    UPDATE memo_embeddings SET
                        content = ?, tags = ?, date_key = ?, embedding = ?, dim = ?, model = ?, updated_at = ?
                    WHERE memo_id = ?
                    """,
                    (content, json.dumps(tags), date_key, embedding_json, len(vec), "text-embedding-3-small", now.isoformat(), memo_id)
                )
            else:
                await mongodb_client.execute(
                    """
                    INSERT INTO memo_embeddings
                        (memo_id, content, tags, date_key, embedding, dim, model, created_at, updated_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (memo_id, content, json.dumps(tags), date_key, embedding_json, len(vec), "text-embedding-3-small", now.isoformat(), now.isoformat())
                )
        except Exception as e:
            logging.warning(f"Failed to create embedding: {e}")

    async def _create_comment_activity_and_notification(
        self,
        memo_id: str,
        parent_id: str,
        commenter_id: str
    ) -> None:
        """Create activity and notification when a comment is created."""
        try:
            # Get the parent memo to find its creator
            parent_memo = await self.get_memo(parent_id, user_id=None)
            if not parent_memo:
                return

            parent_creator = parent_memo.creator_id
            if parent_creator == commenter_id:
                # Don't notify if user comments on their own memo
                return

            # Create activity
            activity_service = get_activity_service()
            activity = await activity_service.create_activity(
                activity_type="MEMO_COMMENT",
                creator_id=commenter_id,
                memo_id=memo_id,
                related_memo_id=parent_id
            )

            # Create notification for parent memo creator
            notification_service = get_notification_service()
            await notification_service.create_notification(
                recipient_id=parent_creator,
                sender_id=commenter_id,
                activity_id=activity.id if hasattr(activity, 'id') else 0,
                notification_type="MEMO_COMMENT"
            )

            logging.info(f"Created activity and notification for comment {memo_id} on parent {parent_id}")
        except Exception as e:
            logging.warning(f"Failed to create activity/notification for comment: {e}")


class AttachmentService:
    """Attachment service with SQLite backend."""

    async def list_attachments(self, memo_id: str | None = None) -> List[schemas.AttachmentResponse]:
        query = "SELECT * FROM attachments WHERE 1=1"
        params: List[Any] = []
        if memo_id:
            query += " AND memo_id = ?"
            params.append(memo_id)
        rows = await mongodb_client.fetch_all(query, tuple(params))
        return [self._row_to_response(row) for row in rows]

    async def upload_attachment(self, file, memo_id: str | None = None, user_id: str | None = None) -> schemas.AttachmentResponse:
        now = utc_now()
        uid = generate_uid()

        sql = """
            INSERT INTO attachments (uid, creator_id, filename, type, size, storage_type, reference, memo_id, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """
        await mongodb_client.execute(
            sql,
            (uid, user_id or "anonymous", getattr(file, "filename", "upload.bin"),
             getattr(file, "content_type", "application/octet-stream"), 0, "LOCAL", "", memo_id,
             now.isoformat(), now.isoformat())
        )

        row = await mongodb_client.fetch_one("SELECT * FROM attachments WHERE uid = ?", (uid,))
        if row:
            return self._row_to_response(row)
        raise ValueError("Failed to upload attachment")

    async def get_attachment(self, attachment_id: str) -> schemas.AttachmentResponse:
        row = None
        if attachment_id.isdigit():
            row = await mongodb_client.fetch_one("SELECT * FROM attachments WHERE id = ?", (int(attachment_id),))
        if not row:
            row = await mongodb_client.fetch_one("SELECT * FROM attachments WHERE uid = ?", (attachment_id,))

        if not row:
            raise ValueError(f"Attachment {attachment_id} not found")

        return self._row_to_response(row)

    async def delete_attachment(self, attachment_id: str, user_id: str | None = None) -> None:
        row = None
        if attachment_id.isdigit():
            row = await mongodb_client.fetch_one("SELECT * FROM attachments WHERE id = ?", (int(attachment_id),))
        if not row:
            row = await mongodb_client.fetch_one("SELECT * FROM attachments WHERE uid = ?", (attachment_id,))

        if not row:
            raise ValueError(f"Attachment {attachment_id} not found")

        if row["creator_id"] != user_id:
            raise ValueError("Access denied")

        await mongodb_client.execute("DELETE FROM attachments WHERE id = ?", (row["id"],))

    def _row_to_response(self, row: aiosqlite.Row) -> schemas.AttachmentResponse:
        return schemas.AttachmentResponse(
            id=str(row["id"]),
            uid=row["uid"] or "",
            filename=row["filename"] or "",
            url=row["reference"] or "",
            type=row["type"] or "",
            size=row["size"] or 0,
        )


class MemoRelationService:
    """Memo relations service (comments, references)."""

    async def create_relation(
        self,
        memo_id: str,
        related_memo_id: str,
        relation_type: str = "REFERENCE",
    ) -> dict:
        now = utc_now().isoformat()

        await mongodb_client.execute(
            """
            INSERT OR IGNORE INTO memo_relations (memo_id, related_memo_id, relation_type, created_at)
            VALUES (?, ?, ?, ?)
            """,
            (memo_id, related_memo_id, relation_type, now)
        )

        row = await mongodb_client.fetch_one(
            "SELECT * FROM memo_relations WHERE memo_id = ? AND related_memo_id = ? AND relation_type = ?",
            (memo_id, related_memo_id, relation_type)
        )
        return serialize_row(row) if row else {}

    async def list_relations(self, memo_id: str, related_memo_id: str | None = None, relation_type: str | None = None) -> List[dict]:
        query = "SELECT * FROM memo_relations WHERE memo_id = ? OR related_memo_id = ?"
        params = [memo_id, memo_id]

        if related_memo_id:
            query += " AND related_memo_id = ?"
            params.append(related_memo_id)
        if relation_type:
            query += " AND relation_type = ?"
            params.append(relation_type)

        rows = await mongodb_client.fetch_all(query, tuple(params))
        return [serialize_row(row) for row in rows]

    async def delete_relation(self, memo_id: str, related_memo_id: str) -> None:
        await mongodb_client.execute(
            "DELETE FROM memo_relations WHERE memo_id = ? AND related_memo_id = ?",
            (memo_id, related_memo_id)
        )


class ReactionService:
    """Reactions service (emoji reactions)."""

    async def add_reaction(
        self,
        content_id: str,
        reaction_type: str,
        user_id: str | None = None,
    ) -> dict:
        now = utc_now().isoformat()

        # Upsert: delete then insert to handle unique constraint
        await mongodb_client.execute(
            """
            INSERT OR REPLACE INTO reactions (content_id, creator_id, reaction_type, created_at)
            VALUES (?, ?, ?, ?)
            """,
            (content_id, user_id or "anonymous", reaction_type, now)
        )

        row = await mongodb_client.fetch_one(
            "SELECT * FROM reactions WHERE content_id = ? AND creator_id = ? AND reaction_type = ?",
            (content_id, user_id or "anonymous", reaction_type)
        )
        return serialize_row(row) or {}

    async def list_reactions(self, content_id: str) -> List[dict]:
        rows = await mongodb_client.fetch_all(
            "SELECT * FROM reactions WHERE content_id = ?",
            (content_id,)
        )
        return [serialize_row(row) for row in rows]

    async def remove_reaction(
        self,
        content_id: str,
        reaction_type: str,
        user_id: str | None = None,
    ) -> None:
        await mongodb_client.execute(
            "DELETE FROM reactions WHERE content_id = ? AND creator_id = ? AND reaction_type = ?",
            (content_id, user_id or "anonymous", reaction_type)
        )


class ShortcutService:
    """Shortcut (Workspace) service with SQLite backend."""

    async def list_shortcuts(self) -> List[schemas.ShortcutResponse]:
        rows = await mongodb_client.fetch_all(
            "SELECT * FROM shortcuts ORDER BY created_at DESC LIMIT 100",
            ()
        )
        return [
            schemas.ShortcutResponse(
                id=str(row["id"]),
                name=row["name"] or "",
                filter=row["filter"] or "",
            )
            for row in rows
        ]

    async def create_shortcut(self, payload: schemas.ShortcutCreate) -> schemas.ShortcutResponse:
        now = utc_now().isoformat()

        await mongodb_client.execute(
            """
            INSERT INTO shortcuts (creator_id, name, filter, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?)
            """,
            ("1", payload.name, payload.filter, now, now)
        )

        row = await mongodb_client.fetch_one("SELECT * FROM shortcuts WHERE rowid = last_insert_rowid()", ())
        if row:
            return schemas.ShortcutResponse(
                id=str(row["id"]),
                name=row["name"],
                filter=row["filter"],
            )
        raise ValueError("Failed to create shortcut")

    async def update_shortcut(self, shortcut_id: str, payload: schemas.ShortcutUpdate) -> schemas.ShortcutResponse:
        update_fields: List[str] = []
        params: List[Any] = []

        if payload.name is not None:
            update_fields.append("name = ?")
            params.append(payload.name)
        if payload.filter is not None:
            update_fields.append("filter = ?")
            params.append(payload.filter)

        if not update_fields:
            raise ValueError("No updates provided")

        update_fields.append("updated_at = ?")
        params.append(utc_now().isoformat())
        params.append(int(shortcut_id) if shortcut_id.isdigit() else shortcut_id)

        sql = f"UPDATE shortcuts SET {', '.join(update_fields)} WHERE id = ?"
        await mongodb_client.execute(sql, tuple(params))

        row = await mongodb_client.fetch_one("SELECT * FROM shortcuts WHERE id = ?", (params[-1],))
        if row:
            return schemas.ShortcutResponse(
                id=str(row["id"]),
                name=row["name"] or "",
                filter=row["filter"] or "",
            )
        raise ValueError(f"Shortcut {shortcut_id} not found")

    async def delete_shortcut(self, shortcut_id: str) -> None:
        result = await mongodb_client.execute(
            "DELETE FROM shortcuts WHERE id = ?",
            (int(shortcut_id) if shortcut_id.isdigit() else shortcut_id,)
        )
        if result.rowcount == 0:
            raise ValueError(f"Shortcut {shortcut_id} not found")


class ActivityService:
    """Service for managing activity records (e.g., memo comments)"""

    async def list_activities(self) -> List[schemas.ActivityResponse]:
        rows = await mongodb_client.fetch_all(
            "SELECT * FROM activities ORDER BY created_at DESC LIMIT 100",
            ()
        )
        return [self._row_to_response(row) for row in rows]

    async def get_activity(self, activity_id: int) -> Optional[schemas.ActivityResponse]:
        row = await mongodb_client.fetch_one(
            "SELECT * FROM activities WHERE id = ?",
            (activity_id,)
        )
        return self._row_to_response(row) if row else None

    async def create_activity(
        self,
        activity_type: str,
        creator_id: str,
        memo_id: str,
        related_memo_id: str
    ) -> schemas.ActivityResponse:
        """Create a new activity record"""
        now = utc_now().isoformat()

        await mongodb_client.execute(
            """
            INSERT INTO activities (type, creator_id, memo_id, related_memo_id, level, created_at)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (activity_type, creator_id, memo_id, related_memo_id, "INFO", now)
        )

        row = await mongodb_client.fetch_one("SELECT * FROM activities WHERE rowid = last_insert_rowid()", ())
        return self._row_to_response(row) if row else self._row_to_response({
            "id": 0, "type": activity_type, "creator_id": creator_id, "memo_id": memo_id,
            "related_memo_id": related_memo_id, "level": "INFO", "created_at": now
        })

    def _row_to_response(self, row: aiosqlite.Row) -> schemas.ActivityResponse:
        # Safe access for both aiosqlite.Row and sqlite3.Row
        def _get(row, key, default=None):
            try:
                return row[key]
            except (KeyError, IndexError, TypeError):
                return getattr(row, key, default)

        created_at = parse_time(_get(row, "created_at"))
        create_time_str = created_at.isoformat() if created_at else None

        memo_id = _get(row, "memo_id")
        related_memo_id = _get(row, "related_memo_id")

        payload = None
        if memo_id and related_memo_id:
            payload = schemas.ActivityPayload(
                memo_comment=schemas.ActivityMemoCommentPayload(
                    memo=f"memos/{memo_id}",
                    related_memo=f"memos/{related_memo_id}"
                )
            )

        creator_id = _get(row, "creator_id", "unknown")
        activity_type = _get(row, "type", "MEMO_COMMENT")
        level = _get(row, "level", "INFO")

        return schemas.ActivityResponse(
            name=f"activities/{_get(row, 'id')}",
            creator=f"users/{creator_id}",
            type=activity_type,
            level=level,
            create_time=create_time_str,
            payload=payload
        )


class NotificationService:
    """Service for managing user notifications (Inbox)"""

    async def list_notifications(self, user_id: str) -> List[schemas.NotificationResponse]:
        rows = await mongodb_client.fetch_all(
            "SELECT * FROM notifications WHERE recipient_id = ? ORDER BY created_at DESC LIMIT 100",
            (user_id,)
        )
        return [self._row_to_response(row) for row in rows]

    async def create_notification(
        self,
        recipient_id: str,
        sender_id: str,
        activity_id: int,
        notification_type: str = "MEMO_COMMENT"
    ) -> schemas.NotificationResponse:
        now = utc_now().isoformat()

        await mongodb_client.execute(
            """
            INSERT INTO notifications (recipient_id, sender_id, activity_id, type, status, created_at)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (recipient_id, sender_id, activity_id, notification_type, "UNREAD", now)
        )

        row = await mongodb_client.fetch_one("SELECT * FROM notifications WHERE rowid = last_insert_rowid()", ())
        return self._row_to_response(row) if row else self._row_to_response({
            "id": 0, "recipient_id": recipient_id, "sender_id": sender_id,
            "activity_id": activity_id, "type": notification_type, "status": "UNREAD", "created_at": now
        })

    async def update_notification(
        self,
        notification_id: int,
        status: str
    ) -> Optional[schemas.NotificationResponse]:
        row = await mongodb_client.fetch_one(
            "SELECT * FROM notifications WHERE id = ?",
            (notification_id,)
        )
        if not row:
            return None

        await mongodb_client.execute(
            "UPDATE notifications SET status = ? WHERE id = ?",
            (status, notification_id)
        )
        return await self.get_notification(notification_id)

    async def delete_notification(self, notification_id: int) -> bool:
        result = await mongodb_client.execute(
            "DELETE FROM notifications WHERE id = ?",
            (notification_id,)
        )
        return result.rowcount > 0

    async def get_notification(self, notification_id: int) -> Optional[schemas.NotificationResponse]:
        row = await mongodb_client.fetch_one(
            "SELECT * FROM notifications WHERE id = ?",
            (notification_id,)
        )
        return self._row_to_response(row) if row else None

    def _row_to_response(self, row: aiosqlite.Row) -> schemas.NotificationResponse:
        # Safe access for both aiosqlite.Row and sqlite3.Row
        def _get(row, key, default=None):
            try:
                return row[key]
            except (KeyError, IndexError, TypeError):
                return getattr(row, key, default)

        created_at = parse_time(_get(row, "created_at"))
        create_time_str = created_at.isoformat() if created_at else None

        return schemas.NotificationResponse(
            name=f"notifications/{_get(row, 'id')}",
            sender=_get(row, "sender_id", "unknown"),
            activity_id=_get(row, "activity_id", 0),
            type=_get(row, "type", "MEMO_COMMENT"),
            status=_get(row, "status", "UNREAD"),
            create_time=create_time_str
        )


class IdentityProviderService:
    async def list_identity_providers(self) -> List[schemas.IdentityProviderResponse]:
        return []

    async def create_identity_provider(
        self,
        payload: schemas.IdentityProviderCreate,
    ) -> schemas.IdentityProviderResponse:
        return schemas.IdentityProviderResponse(id=1, name=payload.name, type=payload.type)

    async def update_identity_provider(
        self,
        idp_id: int,
        payload: schemas.IdentityProviderUpdate,
    ) -> schemas.IdentityProviderResponse:
        return schemas.IdentityProviderResponse(id=idp_id, name=payload.name or "demo", type="oidc")


class MemoEmbeddingService:
    """Embedding service for semantic search."""

    async def search_embeddings(
        self,
        query_embedding: List[float],
        top_k: int = 5,
        user_id: str | None = None,
    ) -> List[dict]:
        """Search memos by embedding similarity using dot product."""
        q = np.array(query_embedding, dtype=float)
        q_norm = float(np.linalg.norm(q))
        if q_norm == 0:
            return []

        q_normalized = (q / q_norm).tolist()

        query = """
            SELECT m.id, m.content, m.tags as memo_tags, e.embedding
            FROM memo_embeddings e
            JOIN memos m ON e.memo_id = CAST(m.id AS TEXT)
            WHERE m.creator_id = ? AND e.embedding IS NOT NULL
        """
        rows = await mongodb_client.fetch_all(query, (user_id or "1",))

        results: List[dict] = []
        for row in rows:
            try:
                embedding = json.loads(row["embedding"]) if isinstance(row["embedding"], str) else row["embedding"]
                e_norm = float(np.linalg.norm(np.array(embedding)))
                if e_norm == 0:
                    continue
                score = float(np.dot(q_normalized, embedding))
                results.append({
                    "memo_id": str(row["id"]),
                    "content": row["content"] or "",
                    "tags": json.loads(row["memo_tags"]) if isinstance(row["memo_tags"], str) else row["memo_tags"],
                    "score": score,
                })
            except Exception:
                continue

        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:top_k]


class InboxService:
    """Inbox notifications service - notifies users of comments/mentions."""

    async def create_notification(
        self,
        sender_id: str,
        receiver_id: str,
        message_type: str,
        memo_id: str | None = None,
        content: str | None = None,
    ) -> dict:
        now = utc_now().isoformat()

        await mongodb_client.execute(
            """
            INSERT INTO inbox (sender_id, receiver_id, status, message_type, memo_id, content, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (sender_id, receiver_id, "UNREAD", message_type, memo_id, content, now)
        )

        row = await mongodb_client.fetch_one("SELECT * FROM inbox WHERE rowid = last_insert_rowid()", ())
        return serialize_row(row) if row else {}

    async def list_notifications(
        self,
        receiver_id: str,
        status: str | None = None,
    ) -> List[dict]:
        query = "SELECT * FROM inbox WHERE receiver_id = ?"
        params = [receiver_id]
        if status:
            query += " AND status = ?"
            params.append(status)
        query += " ORDER BY created_at DESC LIMIT 50"

        rows = await mongodb_client.fetch_all(query, tuple(params))
        return [serialize_row(row) for row in rows]

    async def mark_as_read(self, notification_id: str, receiver_id: str) -> dict | None:
        row = await mongodb_client.fetch_one(
            "SELECT * FROM inbox WHERE id = ? AND receiver_id = ?",
            (int(notification_id), receiver_id)
        )
        if not row:
            return None

        await mongodb_client.execute(
            "UPDATE inbox SET status = 'READ' WHERE id = ?",
            (notification_id,)
        )
        return serialize_row(row)

    async def mark_all_as_read(self, receiver_id: str) -> int:
        result = await mongodb_client.execute(
            "UPDATE inbox SET status = 'READ' WHERE receiver_id = ? AND status = 'UNREAD'",
            (receiver_id,)
        )
        return result.rowcount

    async def delete_notification(self, notification_id: str, receiver_id: str) -> bool:
        result = await mongodb_client.execute(
            "DELETE FROM inbox WHERE id = ? AND receiver_id = ?",
            (int(notification_id), receiver_id)
        )
        return result.rowcount > 0

    async def get_unread_count(self, receiver_id: str) -> int:
        row = await mongodb_client.fetch_one(
            "SELECT COUNT(*) as cnt FROM inbox WHERE receiver_id = ? AND status = 'UNREAD'",
            (receiver_id,)
        )
        return row["cnt"] if row else 0


class UserSettingsService:
    """User settings/preferences service."""

    async def get_settings(self, user_id: str) -> dict:
        row = await mongodb_client.fetch_one(
            "SELECT * FROM user_settings WHERE user_id = ?",
            (user_id,)
        )

        if not row:
            now = utc_now().isoformat()
            default = {
                "user_id": user_id,
                "locale": "en",
                "theme": "system",
                "memo_visibility": "PRIVATE",
                "enable_notifications": True,
                "created_at": now,
                "updated_at": now,
            }
            await mongodb_client.execute(
                """
                INSERT INTO user_settings (user_id, locale, theme, memo_visibility, enable_notifications, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (user_id, "en", "system", "PRIVATE", 1, now, now)
            )
            return default

        return serialize_row(row)

    async def update_settings(self, user_id: str, updates: dict) -> dict:
        allowed_fields = {"locale", "theme", "memo_visibility", "enable_notifications", "timezone"}
        filtered_updates = {k: v for k, v in updates.items() if k in allowed_fields}
        filtered_updates["updated_at"] = utc_now().isoformat()

        set_clauses = []
        params = []
        for k, v in filtered_updates.items():
            set_clauses.append(f"{k} = ?")
            params.append(v)
        params.append(user_id)

        sql = f"UPDATE user_settings SET {', '.join(set_clauses)} WHERE user_id = ?"
        await mongodb_client.execute(sql, tuple(params))

        row = await mongodb_client.fetch_one("SELECT * FROM user_settings WHERE user_id = ?", (user_id,))
        if row:
            return serialize_row(row)
        return {}

    async def get_openai_api_key(self, user_id: str) -> str | None:
        from common.encryption import decrypt_api_key

        row = await mongodb_client.fetch_one(
            "SELECT openai_api_key_encrypted FROM user_settings WHERE user_id = ?",
            (user_id,)
        )
        if not row:
            return None

        encrypted_key = row["openai_api_key_encrypted"]
        if not encrypted_key:
            return None

        try:
            return decrypt_api_key(encrypted_key)
        except Exception as e:
            logging.warning(f"Failed to decrypt OpenAI key for user {user_id}: {e}")
            return None

    async def set_openai_api_key(self, user_id: str, api_key: str) -> dict:
        from common.encryption import encrypt_api_key, validate_openai_key_format

        if not validate_openai_key_format(api_key):
            raise ValueError("Invalid OpenAI API key format. Key must start with 'sk-' and be 20-100 characters.")

        encrypted_key = encrypt_api_key(api_key)
        now = utc_now().isoformat()

        # Check if exists
        existing = await mongodb_client.fetch_one(
            "SELECT * FROM user_settings WHERE user_id = ?",
            (user_id,)
        )

        if existing:
            await mongodb_client.execute(
                """
                UPDATE user_settings SET openai_api_key_encrypted = ?, updated_at = ?
                WHERE user_id = ?
                """,
                (encrypted_key, now, user_id)
            )
        else:
            await mongodb_client.execute(
                """
                INSERT INTO user_settings (user_id, locale, theme, memo_visibility, enable_notifications, openai_api_key_encrypted, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (user_id, "en", "system", "PRIVATE", 1, encrypted_key, now, now)
            )

        row = await mongodb_client.fetch_one("SELECT * FROM user_settings WHERE user_id = ?", (user_id,))
        return serialize_row(row) if row else {}

    async def delete_openai_api_key(self, user_id: str) -> dict:
        result = await mongodb_client.execute(
            "UPDATE user_settings SET openai_api_key_encrypted = NULL, updated_at = ? WHERE user_id = ?",
            (utc_now().isoformat(), user_id)
        )
        if result.rowcount > 0:
            row = await mongodb_client.fetch_one("SELECT * FROM user_settings WHERE user_id = ?", (user_id,))
            return serialize_row(row) if row else {}
        return {}

    async def has_openai_api_key(self, user_id: str) -> bool:
        row = await mongodb_client.fetch_one(
            "SELECT openai_api_key_encrypted FROM user_settings WHERE user_id = ?",
            (user_id,)
        )
        return bool(row and row["openai_api_key_encrypted"])


class PersonalAccessTokenService:
    """Service for managing Personal Access Tokens (PAT)"""

    async def create_token(
        self,
        user_id: str,
        description: str,
        expires_at: Optional[datetime] = None,
    ) -> schemas.PersonalAccessTokenResponse:
        """Create a new personal access token for a user."""
        import secrets
        import hashlib

        raw_token = secrets.token_urlsafe(32)
        token_hash = hashlib.sha256(raw_token.encode()).hexdigest()

        now = utc_now().isoformat()
        expires_iso = None
        if expires_at:
            if expires_at.tzinfo is None:
                expires_at = expires_at.replace(tzinfo=timezone.utc)
            expires_iso = expires_at.isoformat()

        await mongodb_client.execute(
            """
            INSERT INTO cuccu_personal_access_tokens (user_id, token_hash, description, created_at, expires_at, last_used_at, is_active)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (user_id, token_hash, description, now, expires_iso, None, 1),
        )

        row = await mongodb_client.fetch_one(
            "SELECT * FROM cuccu_personal_access_tokens WHERE rowid = last_insert_rowid()", ()
        )
        if row:
            return schemas.PersonalAccessTokenResponse(
                id=str(row["id"]),
                description=row["description"],
                created_at=row["created_at"],
                last_used_at=row["last_used_at"],
                expires_at=row["expires_at"],
                is_active=bool(row["is_active"]),
                token=raw_token,
            )
        raise ValueError("Failed to create token")

    async def list_tokens(self, user_id: str) -> List[schemas.PersonalAccessTokenResponse]:
        """List all personal access tokens for a user."""
        rows = await mongodb_client.fetch_all(
            "SELECT * FROM cuccu_personal_access_tokens WHERE user_id = ? ORDER BY created_at DESC",
            (user_id,),
        )
        tokens = []
        for row in rows:
            tokens.append(
                schemas.PersonalAccessTokenResponse(
                    id=str(row["id"]),
                    description=row["description"],
                    created_at=row["created_at"],
                    last_used_at=row["last_used_at"],
                    expires_at=row["expires_at"],
                    is_active=bool(row["is_active"]),
                )
            )
        return tokens

    async def delete_token(self, user_id: str, token_id: str) -> bool:
        """Delete a personal access token."""
        result = await mongodb_client.execute(
            "DELETE FROM cuccu_personal_access_tokens WHERE id = ? AND user_id = ?",
            (int(token_id) if token_id.isdigit() else token_id, user_id),
        )
        return result.rowcount > 0

    async def verify_token(self, token: str) -> Optional[dict]:
        """Verify a PAT and return the associated user info if valid."""
        import hashlib

        token_hash = hashlib.sha256(token.encode()).hexdigest()
        row = await mongodb_client.fetch_one(
            "SELECT * FROM cuccu_personal_access_tokens WHERE token_hash = ? AND is_active = 1",
            (token_hash,),
        )
        if not row:
            return None

        # Check expiration
        if row["expires_at"]:
            exp_dt = parse_time(row["expires_at"])
            if exp_dt and utc_now() > exp_dt:
                return None

        # Update last_used_at
        now_iso = utc_now().isoformat()
        await mongodb_client.execute(
            "UPDATE cuccu_personal_access_tokens SET last_used_at = ? WHERE id = ?",
            (now_iso, row["id"]),
        )

        return {"user_id": row["user_id"]}


class WebhookService:
    """Service for managing Webhooks"""

    async def list_webhooks(self, user_id: str) -> List[schemas.WebhookResponse]:
        """List all webhooks for a user."""
        rows = await mongodb_client.fetch_all(
            "SELECT * FROM webhooks WHERE user_id = ? ORDER BY created_at DESC",
            (user_id,),
        )
        webhooks = []
        for row in rows:
            events_json = row["events"]
            events_list: List[str] = []
            if events_json:
                try:
                    events_list = (
                        json.loads(events_json)
                        if isinstance(events_json, str)
                        else events_json
                    )
                except Exception:
                    events_list = []
            webhooks.append(
                schemas.WebhookResponse(
                    id=str(row["id"]),
                    user_id=row["user_id"],
                    name=row["name"],
                    url=row["url"],
                    events=events_list,
                    is_active=bool(row["is_active"]),
                    created_at=row["created_at"],
                    updated_at=row["updated_at"],
                )
            )
        return webhooks

    async def create_webhook(
        self, user_id: str, payload: schemas.WebhookCreate
    ) -> schemas.WebhookResponse:
        """Create a new webhook."""
        now = utc_now().isoformat()
        events_json = json.dumps(payload.events)

        await mongodb_client.execute(
            """
            INSERT INTO cuccu_webhooks (user_id, name, url, events, secret, is_active, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                user_id,
                payload.name,
                payload.url,
                events_json,
                payload.secret,
                1,
                now,
                now,
            ),
        )

        row = await mongodb_client.fetch_one(
            "SELECT * FROM cuccu_webhooks WHERE rowid = last_insert_rowid()", ()
        )
        if row:
            return schemas.WebhookResponse(
                id=str(row["id"]),
                user_id=row["user_id"],
                name=row["name"],
                url=row["url"],
                events=payload.events,
                is_active=bool(row["is_active"]),
                created_at=row["created_at"],
                updated_at=row["updated_at"],
            )
        raise ValueError("Failed to create webhook")

    async def update_webhook(
        self,
        user_id: str,
        webhook_id: str,
        payload: schemas.WebhookUpdate,
    ) -> schemas.WebhookResponse:
        """Update an existing webhook."""
        # Check existence
        row = await mongodb_client.fetch_one(
            "SELECT * FROM cuccu_webhooks WHERE id = ? AND user_id = ?",
            (int(webhook_id) if webhook_id.isdigit() else webhook_id, user_id),
        )
        if not row:
            raise ValueError(f"Webhook {webhook_id} not found")

        update_fields: List[str] = []
        params: List[Any] = []

        if payload.name is not None:
            update_fields.append("name = ?")
            params.append(payload.name)
        if payload.url is not None:
            update_fields.append("url = ?")
            params.append(payload.url)
        if payload.events is not None:
            update_fields.append("events = ?")
            params.append(json.dumps(payload.events))
        if payload.secret is not None:
            update_fields.append("secret = ?")
            params.append(payload.secret)
        if payload.is_active is not None:
            update_fields.append("is_active = ?")
            params.append(1 if payload.is_active else 0)

        if not update_fields:
            raise ValueError("No updates provided")

        update_fields.append("updated_at = ?")
        params.append(utc_now().isoformat())
        params.append(webhook_id)
        params.append(user_id)

        sql = f"UPDATE cuccu_webhooks SET {', '.join(update_fields)} WHERE id = ? AND user_id = ?"
        await mongodb_client.execute(sql, tuple(params))

        # Return updated webhook
        row = await mongodb_client.fetch_one(
            "SELECT * FROM cuccu_webhooks WHERE id = ? AND user_id = ?",
            (int(webhook_id) if webhook_id.isdigit() else webhook_id, user_id),
        )
        if row:
            events_json = row["events"]
            events_list: List[str] = []
            if events_json:
                try:
                    events_list = (
                        json.loads(events_json)
                        if isinstance(events_json, str)
                        else events_json
                    )
                except Exception:
                    events_list = []
            return schemas.WebhookResponse(
                id=str(row["id"]),
                user_id=row["user_id"],
                name=row["name"],
                url=row["url"],
                events=events_list,
                is_active=bool(row["is_active"]),
                created_at=row["created_at"],
                updated_at=row["updated_at"],
            )
        raise ValueError("Failed to update webhook")

    async def delete_webhook(self, user_id: str, webhook_id: str) -> bool:
        """Delete a webhook."""
        result = await mongodb_client.execute(
            "DELETE FROM cuccu_webhooks WHERE id = ? AND user_id = ?",
            (int(webhook_id) if webhook_id.isdigit() else webhook_id, user_id),
        )
        return result.rowcount > 0


class S3StorageService:
    """S3 storage service for attachments."""

    def __init__(self) -> None:
        self.bucket_name = os.getenv("S3_BUCKET_NAME", "cucu-note-attachments")
        self.region = os.getenv("AWS_REGION", "ap-southeast-1")
        self._client = None

    def _get_client(self):
        if self._client is None:
            try:
                import boto3
                self._client = boto3.client(
                    "s3",
                    region_name=self.region,
                    aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
                    aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
                )
            except ImportError:
                raise RuntimeError("boto3 not installed. Run: pip install boto3")
        return self._client

    async def upload_file(
        self,
        file_content: bytes,
        filename: str,
        content_type: str = "application/octet-stream",
        user_id: str = "anonymous",
    ) -> dict:
        import asyncio
        timestamp = datetime.utcnow().strftime("%Y/%m/%d")
        key = f"{user_id}/{timestamp}/{generate_uid()}_{filename}"

        loop = asyncio.get_event_loop()
        await loop.run_in_executor(
            None,
            lambda: self._get_client().put_object(
                Bucket=self.bucket_name,
                Key=key,
                Body=file_content,
                ContentType=content_type,
            ),
        )

        url = f"https://{self.bucket_name}.s3.{self.region}.amazonaws.com/{key}"
        return {"key": key, "url": url, "bucket": self.bucket_name, "size": len(file_content)}

    async def delete_file(self, key: str) -> bool:
        import asyncio
        try:
            loop = asyncio.get_event_loop()
            await loop.run_in_executor(
                None,
                lambda: self._get_client().delete_object(
                    Bucket=self.bucket_name,
                    Key=key,
                ),
            )
            return True
        except Exception:
            return False

    async def get_presigned_url(self, key: str, expires_in: int = 3600) -> str:
        import asyncio
        loop = asyncio.get_event_loop()
        url = await loop.run_in_executor(
            None,
            lambda: self._get_client().generate_presigned_url(
                "get_object",
                Params={"Bucket": self.bucket_name, "Key": key},
                ExpiresIn=expires_in,
            ),
        )
        return url


# =========================================================================
# Dependency providers
# =========================================================================

def get_instance_service() -> InstanceService:
    return InstanceService()

def get_auth_service() -> AuthService:
    return AuthService()

def get_user_service() -> UserService:
    return UserService()

def get_memo_service() -> MemoService:
    return MemoService()

def get_attachment_service() -> AttachmentService:
    return AttachmentService()

def get_shortcut_service() -> ShortcutService:
    return ShortcutService()

def get_activity_service() -> ActivityService:
    return ActivityService()

def get_notification_service() -> NotificationService:
    return NotificationService()

def get_idp_service() -> IdentityProviderService:
    return IdentityProviderService()

def get_memo_relation_service() -> MemoRelationService:
    return MemoRelationService()

def get_reaction_service() -> ReactionService:
    return ReactionService()

def get_memo_embedding_service() -> MemoEmbeddingService:
    return MemoEmbeddingService()

def get_inbox_service() -> InboxService:
    return InboxService()

def get_user_settings_service() -> UserSettingsService:
    return UserSettingsService()

def get_s3_storage_service() -> S3StorageService:
    return S3StorageService()

def get_personal_access_token_service() -> PersonalAccessTokenService:
    return PersonalAccessTokenService()

def get_webhook_service() -> WebhookService:
    return WebhookService()


# =========================================================================
# Document Service (DOCX/PDF Management)
# =========================================================================

class DocumentService:
    """Service for managing document uploads, processing, and AI enhancement."""

    def __init__(self):
        from common.sqlite_client import sqlite_client
        self.sqlite_client = sqlite_client
        from config import DOCUMENT_UPLOAD_DIR, OPENAI_API_KEY
        self.upload_dir = DOCUMENT_UPLOAD_DIR
        self.openai_api_key = OPENAI_API_KEY

    async def create_document(self, payload: schemas.DocumentCreate, creator_id: str) -> schemas.DocumentResponse:
        """Create a new document record from uploaded file."""
        uid = generate_uid()

        # Ensure upload directory exists
        os.makedirs(self.upload_dir, exist_ok=True)

        # Save file to disk
        safe_filename = f"{uid}_{payload.filename}"
        file_path = os.path.join(self.upload_dir, safe_filename)

        # Note: file data should be written by the endpoint before calling this service
        # This service just creates the DB record

        now = utc_now()
        doc_data = {
            'uid': uid,
            'creator_id': creator_id,
            'filename': safe_filename,
            'original_filename': payload.original_filename,
            'file_path': file_path,
            'file_size': payload.file_size,
            'mime_type': payload.mime_type,
            'workspace_id': payload.workspace_id,
            'status': 'PENDING',
            'created_at': now,
            'updated_at': now,
        }

        result = await self.sqlite_client.insert_one(doc_data, table='documents')
        doc_data['id'] = result.inserted_id

        # Start background processing task
        # In production, this would be a Celery task or background thread
        # For now, we'll process synchronously in the endpoint

        return await self._build_document_response(doc_data)

    async def list_documents(
        self,
        creator_id: str,
        workspace_id: Optional[str] = None,
        status: Optional[str] = None,
        limit: int = 50,
        offset: int = 0
    ) -> schemas.DocumentListResponse:
        """List documents for a user with optional filters."""
        query = {'creator_id': creator_id}
        if workspace_id:
            query['workspace_id'] = workspace_id
        if status:
            query['status'] = status

        # Build SQL query with ORDER BY and LIMIT/OFFSET
        where_conds = []
        params = []
        for key, val in query.items():
            where_conds.append(f"{key} = ?")
            params.append(val)
        where_clause = ' AND '.join(where_conds) if where_conds else '1=1'

        sql = f"""
            SELECT * FROM documents
            WHERE {where_clause}
            ORDER BY created_at DESC
            LIMIT ? OFFSET ?
        """
        params.extend([limit, offset])
        rows = await self.sqlite_client.fetch_all(sql, tuple(params))

        documents = [await self._build_document_response(serialize_row(row)) for row in rows]

        # Get total count
        total = await self.sqlite_client.count_documents(query)

        return schemas.DocumentListResponse(
            documents=documents,
            total=total,
            workspace_filter=workspace_id
        )

    async def get_document(self, doc_id: int) -> Optional[schemas.DocumentResponse]:
        """Get a document by ID."""
        row = await self.sqlite_client.fetch_one(
            "SELECT * FROM documents WHERE id = ?",
            (doc_id,)
        )
        if not row:
            return None
        return await self._build_document_response(serialize_row(row))

    async def get_document_by_uid(self, uid: str) -> Optional[schemas.DocumentResponse]:
        """Get a document by UID."""
        row = await self.sqlite_client.fetch_one(
            "SELECT * FROM documents WHERE uid = ?",
            (uid,)
        )
        if not row:
            return None
        return await self._build_document_response(serialize_row(row))

    async def update_document(
        self,
        doc_id: int,
        payload: schemas.DocumentUpdate,
        creator_id: str
    ) -> Optional[schemas.DocumentResponse]:
        """Update document metadata."""
        # Verify ownership
        doc = await self.get_document(doc_id)
        if not doc or doc.creator_id != creator_id:
            return None

        update_data = {}
        if payload.title is not None:
            update_data['title'] = payload.title
        if payload.tags is not None:
            update_data['tags'] = json.dumps(payload.tags) if payload.tags else None
        if payload.summary is not None:
            update_data['summary'] = payload.summary
        if payload.ai_enhanced is not None:
            update_data['ai_enhanced'] = payload.ai_enhanced
        if payload.workspace_id is not None:
            update_data['workspace_id'] = payload.workspace_id

        if update_data:
            update_data['updated_at'] = utc_now()
            await self.sqlite_client.update_one(
                {'id': doc_id},
                {'$set': update_data},
                table='documents'
            )

        return await self.get_document(doc_id)

    async def delete_document(self, doc_id: int, creator_id: str) -> bool:
        """Delete a document and its file."""
        # Verify ownership
        doc = await self.get_document(doc_id)
        if not doc or doc.creator_id != creator_id:
            return False

        # Delete file from disk
        try:
            if os.path.exists(doc.file_path):
                os.remove(doc.file_path)
        except Exception as e:
            logging.warning(f"Failed to delete file {doc.file_path}: {e}")

        # Delete from database
        result = await self.sqlite_client.delete_one({'id': doc_id}, table='documents')
        return result.deleted_count > 0

    async def process_document(self, doc_id: int, creator_id: str) -> schemas.DocumentResponse:
        """Process document: extract text, generate tags/summary with AI if enabled."""
        doc = await self.get_document(doc_id)
        if not doc or doc.creator_id != creator_id:
            raise ValueError("Document not found or access denied")

        # Update status to processing
        await self.sqlite_client.update_one(
            {'id': doc_id},
            {'$set': {'status': 'PROCESSING', 'updated_at': utc_now()}},
            table='documents'
        )

        try:
            # Extract text from DOCX using python-docx
            from docx import Document as PythonDocx
            import zipfile
            import xml.etree.ElementTree as ET

            content = ""
            pages = 0

            # For DOCX files
            if doc.mime_type == 'application/vnd.openxmlformats-officedocument.wordprocessingml.document':
                try:
                    python_doc = PythonDocx(doc.file_path)
                    content = "\n\n".join([p.text for p in python_doc.paragraphs if p.text])
                    pages = len(python_doc.sections)
                except Exception as e:
                    logging.error(f"Failed to parse DOCX: {e}")
                    content = f"Error extracting content: {str(e)}"
                    pages = 0
            else:
                # For other formats, placeholder
                content = f"File format: {doc.mime_type}. Content extraction not implemented."
                pages = 0

            # AI Enhancement if enabled
            tags = []
            summary = None
            ai_enhanced = False

            if self.openai_api_key and content and not content.startswith("Error"):
                try:
                    from openai import AsyncOpenAI
                    client = AsyncOpenAI(api_key=self.openai_api_key)

                    # Generate tags (extract keywords from content)
                    tag_response = await client.chat.completions.create(
                        model="gpt-4o-mini",
                        messages=[
                            {"role": "system", "content": "Extract 5-10 relevant tags from the document content. Return as JSON array of strings."},
                            {"role": "user", "content": f"Document content:\n{content[:2000]}"}
                        ],
                        response_format={"type": "json_object"}
                    )
                    try:
                        tags_data = json.loads(tag_response.choices[0].message.content)
                        tags = tags_data.get("tags", [])[:20]  # Max 20 tags
                    except:
                        tags = []

                    # Generate summary
                    summary_response = await client.chat.completions.create(
                        model="gpt-4o-mini",
                        messages=[
                            {"role": "system", "content": "Generate a concise 2-3 sentence summary of the document."},
                            {"role": "user", "content": f"Document content:\n{content[:3000]}"}
                        ]
                    )
                    summary = summary_response.choices[0].message.content.strip()
                    ai_enhanced = True

                except Exception as e:
                    logging.error(f"AI enhancement failed: {e}")
                    # Continue without AI - document still processed

            # Update document with processed data
            update_data = {
                'content': content,
                'pages': pages,
                'status': 'COMPLETED',
                'processed_at': utc_now(),
                'updated_at': utc_now(),
            }
            if tags:
                update_data['tags'] = json.dumps(tags)
            if summary:
                update_data['summary'] = summary
            if ai_enhanced:
                update_data['ai_enhanced'] = ai_enhanced

            await self.sqlite_client.update_one(
                {'id': doc_id},
                {'$set': update_data},
                table='documents'
            )

            return await self.get_document(doc_id)

        except Exception as e:
            logging.error(f"Document processing failed for doc {doc_id}: {e}")
            # Update status to failed
            await self.sqlite_client.update_one(
                {'id': doc_id},
                {'$set': {
                    'status': 'FAILED',
                    'error_message': str(e),
                    'updated_at': utc_now()
                }},
                table='documents'
            )
            raise

    async def import_to_memo(
        self,
        doc_id: int,
        creator_id: str,
        memo_title: Optional[str] = None,
        tags: List[str] = [],
        visibility: str = "PRIVATE"
    ) -> schemas.MemoResponse:
        """Import document content as a new memo."""
        doc = await self.get_document(doc_id)
        if not doc or doc.creator_id != creator_id:
            raise ValueError("Document not found or access denied")

        if not doc.content:
            raise ValueError("Document has no content to import")

        # Create memo from document content
        from .memo_service import MemoService
        memo_service = MemoService()

        title = memo_title or doc.title or doc.original_filename
        content = f"# {title}\n\n{doc.content}"

        # Merge document tags with provided tags
        all_tags = list(set((doc.tags or []) + tags))

        memo_create = schemas.MemoCreate(
            content=content,
            tags=all_tags,
            visibility=visibility,
            workspace_id=doc.workspace_id
        )

        memo = await memo_service.create_memo(memo_create, creator_id)
        return memo

    async def _build_document_response(self, row_dict: dict) -> schemas.DocumentResponse:
        """Build DocumentResponse from database row."""
        tags = None
        if row_dict.get('tags'):
            try:
                tags = json.loads(row_dict['tags'])
            except:
                tags = []

        return schemas.DocumentResponse(
            id=row_dict['id'],
            uid=row_dict['uid'],
            creator_id=row_dict['creator_id'],
            filename=row_dict['filename'],
            original_filename=row_dict['original_filename'],
            file_path=row_dict['file_path'],
            file_size=row_dict['file_size'],
            mime_type=row_dict['mime_type'],
            workspace_id=row_dict['workspace_id'],
            status=row_dict['status'],
            pages=row_dict.get('pages'),
            title=row_dict.get('title'),
            content=row_dict.get('content'),
            summary=row_dict.get('summary'),
            tags=tags,
            ai_enhanced=bool(row_dict.get('ai_enhanced')),
            error_message=row_dict.get('error_message'),
            created_at=parse_time(row_dict['created_at']),
            updated_at=parse_time(row_dict['updated_at']),
            processed_at=parse_time(row_dict.get('processed_at'))
        )


def get_document_service() -> DocumentService:
    return DocumentService()
