✨ feat(users): vincular operacoes ao usuario com user_id e ownership

app/api/routes.py

@@ -71,7 +71,10 @@ async def chat(request: ChatRequest, db: Session = Depends(get_db)):
     """Processa mensagem do usuario via orquestrador e retorna resposta do chat."""
     try:
         service = OrquestradorService(db)
-        result = await service.handle_message(message=request.message)
+        result = await service.handle_message(
+            message=request.message,
+            user_id=request.user_id,
+        )
         return ChatResponse(response=result)
     except SQLAlchemyError as exc:
         raise HTTPException(
@@ -136,6 +139,7 @@ async def agendar_revisao_endpoint(
         return await agendar_revisao(
             placa=body.placa,
             data_hora=body.data_hora,
+            user_id=body.user_id,
         )
     except SQLAlchemyError as exc:
         raise HTTPException(status_code=503, detail=_db_error_detail(exc))
@@ -151,6 +155,7 @@ async def cancelar_pedido_endpoint(
         return await cancelar_pedido(
             numero_pedido=body.numero_pedido,
             motivo=body.motivo,
+            user_id=body.user_id,
         )
     except SQLAlchemyError as exc:
         raise HTTPException(status_code=503, detail=_db_error_detail(exc))

app/api/schemas.py

@@ -5,7 +5,7 @@ from typing import Dict, Any, Optional, Literal
 
 class ChatRequest(BaseModel):
     message: str
-    # user_id: str -> Removido momentaniamente para testar o VertexIA
+    user_id: Optional[int] = None
 
 class ChatResponse(BaseModel):
     response: str
@@ -48,8 +48,10 @@ class AvaliarVeiculoTrocaRequest(BaseModel):
 class AgendarRevisaoRequest(BaseModel):
     placa: str
     data_hora: str
+    user_id: Optional[int] = None
 
 
 class CancelarPedidoRequest(BaseModel):
     numero_pedido: str
     motivo: str
+    user_id: Optional[int] = None

app/db/mock_models.py

@@ -1,4 +1,4 @@
-from sqlalchemy import Boolean, Column, DateTime, Float, ForeignKey, Integer, String, Text
+from sqlalchemy import Boolean, Column, DateTime, Float, ForeignKey, Integer, String, Text, UniqueConstraint
 from sqlalchemy.sql import func
 
 from app.db.mock_database import MockBase
@@ -26,11 +26,33 @@ class Customer(MockBase):
     created_at = Column(DateTime, server_default=func.current_timestamp())
 
 
+class User(MockBase):
+    __tablename__ = "users"
+    __table_args__ = (
+        UniqueConstraint("channel", "external_id", name="uq_mock_users_channel_external_id"),
+    )
+
+    id = Column(Integer, primary_key=True, index=True)
+    channel = Column(String(40), nullable=False, index=True)
+    external_id = Column(String(120), nullable=False, index=True)
+    name = Column(String(120), nullable=True)
+    username = Column(String(120), nullable=True)
+    cpf = Column(String(11), ForeignKey("customers.cpf"), nullable=True, index=True)
+    phone = Column(String(30), nullable=True)
+    created_at = Column(DateTime, server_default=func.current_timestamp())
+    updated_at = Column(
+        DateTime,
+        server_default=func.current_timestamp(),
+        onupdate=func.current_timestamp(),
+    )
+
+
 class Order(MockBase):
     __tablename__ = "orders"
 
     id = Column(Integer, primary_key=True, index=True)
     numero_pedido = Column(String(40), unique=True, nullable=False, index=True)
+    user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
     cpf = Column(String(11), ForeignKey("customers.cpf"), nullable=False, index=True)
     status = Column(String(20), nullable=False, default="Ativo")
     motivo_cancelamento = Column(Text, nullable=True)
@@ -48,6 +70,7 @@ class ReviewSchedule(MockBase):
 
     id = Column(Integer, primary_key=True, index=True)
     protocolo = Column(String(50), unique=True, nullable=False, index=True)
+    user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
     placa = Column(String(10), nullable=False, index=True)
     data_hora = Column(DateTime, nullable=False)
     status = Column(String(20), nullable=False, default="agendado")

app/db/models/__init__.py

@@ -0,0 +1,3 @@
+from app.db.models.tool import Tool
+
+__all__ = ["Tool"]

app/db/models/tool.py

@@ -1,5 +1,6 @@
-from sqlalchemy import Column, Integer, String, Text, JSON, TIMESTAMP
+from sqlalchemy import JSON, TIMESTAMP, Column, Integer, String, Text
 from sqlalchemy.sql import func
+
 from app.db.database import Base
 
 
@@ -7,20 +8,17 @@ class Tool(Base):
     __tablename__ = "tools"
 
     id = Column(Integer, primary_key=True, index=True)
-
     name = Column(String(100), unique=True, nullable=False)
-
     description = Column(Text, nullable=False)
-
     parameters = Column(JSON, nullable=False)
 
     created_at = Column(
         TIMESTAMP,
-        server_default=func.current_timestamp()
+        server_default=func.current_timestamp(),
     )
 
     updated_at = Column(
         TIMESTAMP,
         server_default=func.current_timestamp(),
-        onupdate=func.current_timestamp()
+        onupdate=func.current_timestamp(),
     )

app/integrations/telegram_satellite_service.py

@@ -7,7 +7,9 @@ from fastapi import HTTPException
 
 from app.core.settings import settings
 from app.db.database import SessionLocal
+from app.db.mock_database import SessionMockLocal
 from app.services.orquestrador_service import OrquestradorService
+from app.services.user_service import UserService
 
 
 logger = logging.getLogger(__name__)
@@ -68,12 +70,13 @@ class TelegramSatelliteService:
         text = message.get("text")
         chat = message.get("chat", {})
         chat_id = chat.get("id")
+        sender = message.get("from", {})
 
         if not text or not chat_id:
             return
 
         try:
-            answer = await self._process_message(text=text)
+            answer = await self._process_message(text=text, sender=sender, chat_id=chat_id)
         except HTTPException as exc:
             logger.warning("Falha de dominio ao processar mensagem no Telegram: %s", exc.detail)
             answer = str(exc.detail) if exc.detail else "Nao foi possivel concluir a operacao solicitada."
@@ -99,14 +102,30 @@ class TelegramSatelliteService:
             if not data.get("ok"):
                 logger.warning("Falha em sendMessage: %s", data)
 
-    async def _process_message(self, text: str) -> str:
-        """Encaminha mensagem ao orquestrador e retorna a resposta gerada."""
-        db = SessionLocal()
+    async def _process_message(self, text: str, sender: Dict[str, Any], chat_id: int) -> str:
+        """Encaminha mensagem ao orquestrador com usuario identificado e retorna resposta."""
+        tools_db = SessionLocal()
+        mock_db = SessionMockLocal()
         try:
-            service = OrquestradorService(db)
-            return await service.handle_message(message=text)
+            user_service = UserService(mock_db)
+            external_id = str(sender.get("id") or chat_id)
+            first_name = (sender.get("first_name") or "").strip()
+            last_name = (sender.get("last_name") or "").strip()
+            display_name = f"{first_name} {last_name}".strip() or None
+            username = sender.get("username")
+
+            user = user_service.get_or_create(
+                channel="telegram",
+                external_id=external_id,
+                name=display_name,
+                username=username,
+            )
+
+            service = OrquestradorService(tools_db)
+            return await service.handle_message(message=text, user_id=user.id)
         finally:
-            db.close()
+            tools_db.close()
+            mock_db.close()
 
 
 async def main() -> None:

app/repositories/user_repository.py

@@ -0,0 +1,64 @@
+from sqlalchemy.orm import Session
+
+from app.db.mock_models import User
+
+
+class UserRepository:
+    def __init__(self, db: Session):
+        """Inicializa o repositorio de usuarios com a sessao ativa."""
+        self.db = db
+
+    def get_by_id(self, user_id: int):
+        """Busca usuario por ID interno."""
+        return self.db.query(User).filter(User.id == user_id).first()
+
+    def get_by_channel_external_id(self, channel: str, external_id: str):
+        """Busca usuario por canal e identificador externo."""
+        return (
+            self.db.query(User)
+            .filter(
+                User.channel == channel,
+                User.external_id == external_id,
+            )
+            .first()
+        )
+
+    def create(
+        self,
+        channel: str,
+        external_id: str,
+        name: str | None = None,
+        username: str | None = None,
+    ):
+        """Cria e persiste um novo usuario."""
+        user = User(
+            channel=channel,
+            external_id=external_id,
+            name=name,
+            username=username,
+        )
+        self.db.add(user)
+        self.db.commit()
+        self.db.refresh(user)
+        return user
+
+    def update_identity(
+        self,
+        user: User,
+        name: str | None,
+        username: str | None,
+    ):
+        """Atualiza dados basicos de identidade quando houver mudancas."""
+        changed = False
+        if name and name != user.name:
+            user.name = name
+            changed = True
+        if username and username != user.username:
+            user.username = username
+            changed = True
+
+        if changed:
+            self.db.commit()
+            self.db.refresh(user)
+
+        return user

app/services/handlers.py

@@ -122,7 +122,7 @@ async def avaliar_veiculo_troca(modelo: str, ano: int, km: int) -> Dict[str, Any
     }
 
 
-async def agendar_revisao(placa: str, data_hora: str) -> Dict[str, Any]:
+async def agendar_revisao(placa: str, data_hora: str, user_id: Optional[int] = None) -> Dict[str, Any]:
     """Cria ou reaproveita agendamento de revisao a partir de placa e data/hora."""
     try:
         dt = datetime.fromisoformat(data_hora.replace("Z", "+00:00"))
@@ -132,7 +132,7 @@ async def agendar_revisao(placa: str, data_hora: str) -> Dict[str, Any]:
             detail="data_hora invalida. Use formato ISO 8601, por exemplo: 2026-03-10T09:00:00-03:00",
         )
 
-    entropy = hashlib.md5(f"{placa}:{data_hora}".encode("utf-8")).hexdigest()[:8].upper()
+    entropy = hashlib.md5(f"{user_id}:{placa}:{data_hora}".encode("utf-8")).hexdigest()[:8].upper()
     protocolo = f"REV-{dt.strftime('%Y%m%d')}-{entropy}"
 
     db = SessionMockLocal()
@@ -141,6 +141,7 @@ async def agendar_revisao(placa: str, data_hora: str) -> Dict[str, Any]:
         if existente:
             return {
                 "protocolo": existente.protocolo,
+                "user_id": existente.user_id,
                 "placa": existente.placa,
                 "data_hora": existente.data_hora.isoformat(),
                 "status": existente.status,
@@ -148,6 +149,7 @@ async def agendar_revisao(placa: str, data_hora: str) -> Dict[str, Any]:
 
         agendamento = ReviewSchedule(
             protocolo=protocolo,
+            user_id=user_id,
             placa=placa.upper(),
             data_hora=dt,
             status="agendado",
@@ -158,6 +160,7 @@ async def agendar_revisao(placa: str, data_hora: str) -> Dict[str, Any]:
 
         return {
             "protocolo": agendamento.protocolo,
+            "user_id": agendamento.user_id,
             "placa": agendamento.placa,
             "data_hora": agendamento.data_hora.isoformat(),
             "status": agendamento.status,
@@ -166,17 +169,38 @@ async def agendar_revisao(placa: str, data_hora: str) -> Dict[str, Any]:
         db.close()
 
 
-async def cancelar_pedido(numero_pedido: str, motivo: str) -> Dict[str, Any]:
+async def cancelar_pedido(numero_pedido: str, motivo: str, user_id: Optional[int] = None) -> Dict[str, Any]:
     """Cancela pedido existente e registra motivo e data de cancelamento."""
     db = SessionMockLocal()
     try:
-        pedido = db.query(Order).filter(Order.numero_pedido == numero_pedido).first()
+        query = db.query(Order).filter(Order.numero_pedido == numero_pedido)
+        if user_id is not None:
+            query = query.filter(Order.user_id == user_id)
+
+        pedido = query.first()
+        if not pedido and user_id is not None:
+            # Compatibilidade com pedidos antigos que ainda nao possuem user_id.
+            legado = (
+                db.query(Order)
+                .filter(Order.numero_pedido == numero_pedido)
+                .filter(Order.user_id.is_(None))
+                .first()
+            )
+            if legado:
+                legado.user_id = user_id
+                db.commit()
+                db.refresh(legado)
+                pedido = legado
+
         if not pedido:
+            if user_id is not None:
+                raise HTTPException(status_code=404, detail="Pedido nao encontrado para este usuario.")
             raise HTTPException(status_code=404, detail="Pedido nao encontrado na base ficticia.")
 
         if pedido.status.lower() == "cancelado":
             return {
                 "numero_pedido": pedido.numero_pedido,
+                "user_id": pedido.user_id,
                 "status": pedido.status,
                 "motivo": pedido.motivo_cancelamento,
                 "data_cancelamento": pedido.data_cancelamento.isoformat() if pedido.data_cancelamento else None,
@@ -190,6 +214,7 @@ async def cancelar_pedido(numero_pedido: str, motivo: str) -> Dict[str, Any]:
 
         return {
             "numero_pedido": pedido.numero_pedido,
+            "user_id": pedido.user_id,
             "status": pedido.status,
             "motivo": pedido.motivo_cancelamento,
             "data_cancelamento": pedido.data_cancelamento.isoformat() if pedido.data_cancelamento else None,

app/services/orquestrador_service.py

@@ -22,18 +22,18 @@ class OrquestradorService:
         self.llm = LLMService()
         self.registry = ToolRegistry(db)
 
-    async def handle_message(self, message: str) -> str:
+    async def handle_message(self, message: str, user_id: int | None = None) -> str:
         """Processa mensagem, executa tool quando necessario e retorna resposta final."""
         tools = self.registry.get_tools()
 
         llm_result = await self.llm.generate_response(
-            message=self._build_router_prompt(message),
+            message=self._build_router_prompt(user_message=message, user_id=user_id),
             tools=tools,
         )
 
         if not llm_result["tool_call"] and self._is_operational_query(message):
             llm_result = await self.llm.generate_response(
-                message=self._build_force_tool_prompt(message),
+                message=self._build_force_tool_prompt(user_message=message, user_id=user_id),
                 tools=tools,
             )
 
@@ -42,13 +42,18 @@ class OrquestradorService:
             arguments = llm_result["tool_call"]["arguments"]
 
             try:
-                tool_result = await self.registry.execute(tool_name, arguments)
+                tool_result = await self.registry.execute(
+                    tool_name,
+                    arguments,
+                    user_id=user_id,
+                )
             except HTTPException as exc:
                 return self._http_exception_detail(exc)
 
             final_response = await self.llm.generate_response(
                 message=self._build_result_prompt(
                     user_message=message,
+                    user_id=user_id,
                     tool_name=tool_name,
                     tool_result=tool_result,
                 ),
@@ -88,26 +93,38 @@ class OrquestradorService:
         )
         return any(k in text for k in keywords)
 
-    def _build_router_prompt(self, user_message: str) -> str:
+    def _build_router_prompt(self, user_message: str, user_id: int | None) -> str:
+        user_context = f"Contexto de usuario autenticado: user_id={user_id}.\n" if user_id else ""
         return (
             "Voce e um assistente de concessionaria. "
             "Sempre que a solicitacao depender de dados operacionais (estoque, validacao de cliente, "
             "avaliacao de troca, agendamento de revisao ou cancelamento de pedido), use a tool correta. "
             "Se faltar parametro obrigatorio para a tool, responda em texto pedindo apenas o que falta.\n\n"
+            f"{user_context}"
             f"Mensagem do usuario: {user_message}"
         )
 
-    def _build_force_tool_prompt(self, user_message: str) -> str:
+    def _build_force_tool_prompt(self, user_message: str, user_id: int | None) -> str:
+        user_context = f"Contexto de usuario autenticado: user_id={user_id}.\n" if user_id else ""
         return (
             "Reavalie a mensagem e priorize chamar tool se houver intencao operacional. "
             "Use texto apenas quando faltar dado obrigatorio.\n\n"
+            f"{user_context}"
             f"Mensagem do usuario: {user_message}"
         )
 
-    def _build_result_prompt(self, user_message: str, tool_name: str, tool_result) -> str:
+    def _build_result_prompt(
+        self,
+        user_message: str,
+        user_id: int | None,
+        tool_name: str,
+        tool_result,
+    ) -> str:
+        user_context = f"Contexto de usuario autenticado: user_id={user_id}.\n" if user_id else ""
         return (
             "Responda ao usuario de forma objetiva usando o resultado da tool abaixo. "
             "Nao invente dados. Se a lista vier vazia, diga explicitamente que nao encontrou resultados.\n\n"
+            f"{user_context}"
             f"Pergunta original: {user_message}\n"
             f"Tool executada: {tool_name}\n"
             f"Resultado da tool: {tool_result}"

app/services/tool_registry.py

@@ -1,3 +1,4 @@
+import inspect
 from typing import Callable, Dict, List
 
 from sqlalchemy.orm import Session
@@ -55,11 +56,15 @@ class ToolRegistry:
         """Retorna a lista atual de tools registradas."""
         return self._tools
 
-    async def execute(self, name: str, arguments: dict):
+    async def execute(self, name: str, arguments: dict, user_id: int | None = None):
         """Executa a tool solicitada pelo modelo com os argumentos extraidos."""
         tool = next((t for t in self._tools if t.name == name), None)
 
         if not tool:
             raise Exception(f"Tool {name} nao encontrada.")
 
-        return await tool.handler(**arguments)
+        call_args = dict(arguments or {})
+        if user_id is not None and "user_id" in inspect.signature(tool.handler).parameters:
+            call_args["user_id"] = user_id
+
+        return await tool.handler(**call_args)

app/services/user_service.py

@@ -0,0 +1,28 @@
+from sqlalchemy.orm import Session
+
+from app.repositories.user_repository import UserRepository
+
+
+class UserService:
+    def __init__(self, db: Session):
+        """Inicializa o servico de usuarios com repositorio persistente."""
+        self.repo = UserRepository(db)
+
+    def get_or_create(
+        self,
+        channel: str,
+        external_id: str,
+        name: str | None = None,
+        username: str | None = None,
+    ):
+        """Retorna usuario existente por canal/external_id ou cria um novo."""
+        user = self.repo.get_by_channel_external_id(channel=channel, external_id=external_id)
+        if not user:
+            return self.repo.create(
+                channel=channel,
+                external_id=external_id,
+                name=name,
+                username=username,
+            )
+
+        return self.repo.update_identity(user=user, name=name, username=username)