import logging
import time
import uuid
from typing import Optional

from sqlalchemy.orm import Session
from open_webui.internal.db import Base, JSONField, get_db, get_db_context


from pydantic import BaseModel, ConfigDict
from sqlalchemy import BigInteger, Column, String, JSON, PrimaryKeyConstraint, Index

log = logging.getLogger(__name__)


####################
# Tag DB Schema
####################
class Tag(Base):
    __tablename__ = "tag"
    id = Column(String)
    name = Column(String)
    user_id = Column(String)
    meta = Column(JSON, nullable=True)

    __table_args__ = (
        PrimaryKeyConstraint("id", "user_id", name="pk_id_user_id"),
        Index("user_id_idx", "user_id"),
    )

    # Unique constraint ensuring (id, user_id) is unique, not just the `id` column
    __table_args__ = (PrimaryKeyConstraint("id", "user_id", name="pk_id_user_id"),)


class TagModel(BaseModel):
    id: str
    name: str
    user_id: str
    meta: Optional[dict] = None
    model_config = ConfigDict(from_attributes=True)


####################
# Forms
####################


class TagChatIdForm(BaseModel):
    name: str
    chat_id: str


class TagTable:
    def insert_new_tag(
        self, name: str, user_id: str, db: Optional[Session] = None
    ) -> Optional[TagModel]:
        with get_db_context(db) as db:
            id = name.replace(" ", "_").lower()
            tag = TagModel(**{"id": id, "user_id": user_id, "name": name})
            try:
                result = Tag(**tag.model_dump())
                db.add(result)
                db.commit()
                db.refresh(result)
                if result:
                    return TagModel.model_validate(result)
                else:
                    return None
            except Exception as e:
                log.exception(f"Error inserting a new tag: {e}")
                return None

    def get_tag_by_name_and_user_id(
        self, name: str, user_id: str, db: Optional[Session] = None
    ) -> Optional[TagModel]:
        try:
            id = name.replace(" ", "_").lower()
            with get_db_context(db) as db:
                tag = db.query(Tag).filter_by(id=id, user_id=user_id).first()
                return TagModel.model_validate(tag)
        except Exception:
            return None

    def get_tags_by_user_id(
        self, user_id: str, db: Optional[Session] = None
    ) -> list[TagModel]:
        with get_db_context(db) as db:
            return [
                TagModel.model_validate(tag)
                for tag in (db.query(Tag).filter_by(user_id=user_id).all())
            ]

    def get_tags_by_ids_and_user_id(
        self, ids: list[str], user_id: str, db: Optional[Session] = None
    ) -> list[TagModel]:
        with get_db_context(db) as db:
            return [
                TagModel.model_validate(tag)
                for tag in (
                    db.query(Tag).filter(Tag.id.in_(ids), Tag.user_id == user_id).all()
                )
            ]

    def delete_tag_by_name_and_user_id(
        self, name: str, user_id: str, db: Optional[Session] = None
    ) -> bool:
        try:
            with get_db_context(db) as db:
                id = name.replace(" ", "_").lower()
                res = db.query(Tag).filter_by(id=id, user_id=user_id).delete()
                log.debug(f"res: {res}")
                db.commit()
                return True
        except Exception as e:
            log.error(f"delete_tag: {e}")
            return False

    def delete_tags_by_ids_and_user_id(
        self, ids: list[str], user_id: str, db: Optional[Session] = None
    ) -> bool:
        """Delete all tags whose id is in *ids* for the given user, in one query."""
        if not ids:
            return True
        try:
            with get_db_context(db) as db:
                db.query(Tag).filter(Tag.id.in_(ids), Tag.user_id == user_id).delete(
                    synchronize_session=False
                )
                db.commit()
                return True
        except Exception as e:
            log.error(f"delete_tags_by_ids: {e}")
            return False

    def ensure_tags_exist(
        self, names: list[str], user_id: str, db: Optional[Session] = None
    ) -> None:
        """Create tag rows for any *names* that don't already exist for *user_id*."""
        if not names:
            return
        ids = [n.replace(" ", "_").lower() for n in names]
        with get_db_context(db) as db:
            existing = {
                t.id
                for t in db.query(Tag.id)
                .filter(Tag.id.in_(ids), Tag.user_id == user_id)
                .all()
            }
            new_tags = [
                Tag(id=tag_id, name=name, user_id=user_id)
                for tag_id, name in zip(ids, names)
                if tag_id not in existing
            ]
            if new_tags:
                db.add_all(new_tags)
                db.commit()


Tags = TagTable()
