import logging from typing import Any from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, Response, status from sqlalchemy import text from sqlalchemy.orm import Session try: from app.db import get_db from app.routers.im import ( CHILD_PARTICIPANT_TYPE, PARENT_PARTICIPANT_TYPE, SUPPORTED_CONVERSATION_TYPES, _fetch_child_names, _fetch_parent_names, _get_conversation_for_child, _row_to_conversation_item, _row_to_message_item, ) from app.schemas.im import ( ChildConversationListResponse, ChildConversationMessageListResponse, ConversationMessageCreateResponse, DeviceMessageCreateRequest, ) from app.service.im import authenticate_device_identity, create_device_message except ModuleNotFoundError: from db import get_db from routers.im import ( CHILD_PARTICIPANT_TYPE, PARENT_PARTICIPANT_TYPE, SUPPORTED_CONVERSATION_TYPES, _fetch_child_names, _fetch_parent_names, _get_conversation_for_child, _row_to_conversation_item, _row_to_message_item, ) from schemas.im import ( ChildConversationListResponse, ChildConversationMessageListResponse, ConversationMessageCreateResponse, DeviceMessageCreateRequest, ) from service.im import authenticate_device_identity, create_device_message router = APIRouter(prefix="/device-im", tags=["device-im"]) logger = logging.getLogger("app.device_im") @router.get("/{device_id}/conversations", response_model=ChildConversationListResponse) def list_device_conversations( device_id: str, request: Request, device_serial: str = Header(alias="X-Device-Serial", min_length=1), conversation_type: int | None = Query(default=None), cursor: int | None = Query(default=None, ge=1), limit: int = Query(default=20, ge=1, le=100), db: Session = Depends(get_db), ) -> ChildConversationListResponse: if conversation_type is not None and conversation_type not in SUPPORTED_CONVERSATION_TYPES: raise HTTPException(status_code=422, detail="unsupported conversation_type") device_identity = authenticate_device_identity( db=db, device_id=device_id, serial_number=device_serial, ) child_id = device_identity.child_id params: dict[str, Any] = { "child_id_str": str(child_id), "child_participant_type": CHILD_PARTICIPANT_TYPE, "fetch_limit": limit + 1, } where = """ status = 1 AND conversation_type IN (1, 2) AND ( (participant_a_type = :child_participant_type AND participant_a_id = :child_id_str) OR (participant_b_type = :child_participant_type AND participant_b_id = :child_id_str) ) """ if conversation_type is not None: where += " AND conversation_type = :conversation_type" params["conversation_type"] = conversation_type if cursor is not None: where += " AND id < :cursor" params["cursor"] = cursor rows = ( db.execute( text( f""" SELECT id, conversation_type, participant_a_type, participant_a_id, participant_b_type, participant_b_id, last_message_preview, last_message_at, message_count, created_at FROM im_conversations WHERE {where} ORDER BY COALESCE(last_message_at, created_at) DESC, id DESC LIMIT :fetch_limit """ ), params, ) .mappings() .all() ) has_more = len(rows) > limit rows = rows[:limit] next_cursor = int(rows[-1]["id"]) if has_more and rows else None peer_parent_ids: set[int] = set() peer_child_ids: set[int] = set() child_id_str = str(child_id) for row in rows: participant_a_type = int(row["participant_a_type"]) participant_b_type = int(row["participant_b_type"]) participant_a_id = str(row["participant_a_id"]) participant_b_id = str(row["participant_b_id"]) if participant_a_type == CHILD_PARTICIPANT_TYPE and participant_a_id == child_id_str: peer_type = participant_b_type peer_id = participant_b_id else: peer_type = participant_a_type peer_id = participant_a_id if peer_id.isdigit(): if peer_type == PARENT_PARTICIPANT_TYPE: peer_parent_ids.add(int(peer_id)) elif peer_type == CHILD_PARTICIPANT_TYPE: peer_child_ids.add(int(peer_id)) parent_names = _fetch_parent_names(db, peer_parent_ids) child_names = _fetch_child_names(db, peer_child_ids) items = [ _row_to_conversation_item( row, child_id=child_id, parent_names=parent_names, child_names=child_names, ) for row in rows ] logger.info( "device conversations listed", extra={ "event": "device_conversation_list", "request_id": getattr(request.state, "request_id", None), "device_id": device_id, "child_id": child_id, "conversation_type": conversation_type, "count": len(items), }, ) return ChildConversationListResponse( items=items, total=len(items), next_cursor=next_cursor, ) @router.get( "/{device_id}/conversations/{conversation_id}/messages", response_model=ChildConversationMessageListResponse, ) def list_device_conversation_messages( device_id: str, conversation_id: int, request: Request, device_serial: str = Header(alias="X-Device-Serial", min_length=1), cursor_seq: int | None = Query(default=None, ge=1), limit: int = Query(default=20, ge=1, le=100), db: Session = Depends(get_db), ) -> ChildConversationMessageListResponse: device_identity = authenticate_device_identity( db=db, device_id=device_id, serial_number=device_serial, ) _get_conversation_for_child( db=db, conversation_id=conversation_id, child_id=device_identity.child_id, ) sql = """ SELECT id, conversation_id, seq, sender_type, sender_id, receiver_type, receiver_id, content_type, content_text, content_json, media_file_key, media_duration_ms, media_mime_type, media_size_bytes, media_transcript_text, client_msg_id, sender_name_snapshot, sender_avatar_snapshot, receiver_name_snapshot, receiver_avatar_snapshot, ext_json, created_at FROM im_messages WHERE conversation_id = :conversation_id AND deleted_at IS NULL """ params: dict[str, Any] = { "conversation_id": conversation_id, "fetch_limit": limit + 1, } if cursor_seq is not None: sql += " AND seq < :cursor_seq" params["cursor_seq"] = cursor_seq sql += " ORDER BY seq DESC LIMIT :fetch_limit" rows = db.execute(text(sql), params).mappings().all() has_more = len(rows) > limit rows = rows[:limit] rows.reverse() items = [_row_to_message_item(row) for row in rows] next_cursor_seq = items[0].seq if has_more and items else None logger.info( "device conversation messages listed", extra={ "event": "device_conversation_message_list", "request_id": getattr(request.state, "request_id", None), "device_id": device_id, "child_id": device_identity.child_id, "conversation_id": conversation_id, "count": len(items), "has_more": has_more, }, ) return ChildConversationMessageListResponse( conversation_id=conversation_id, has_more=has_more, next_cursor_seq=next_cursor_seq, items=items, ) @router.post( "/{device_id}/messages", response_model=ConversationMessageCreateResponse, status_code=status.HTTP_201_CREATED, ) def create_message_from_device( device_id: str, payload: DeviceMessageCreateRequest, request: Request, response: Response, device_serial: str = Header(alias="X-Device-Serial", min_length=1), db: Session = Depends(get_db), ) -> ConversationMessageCreateResponse: device_identity, result = create_device_message( db=db, device_id=device_id, serial_number=device_serial, payload=payload, ) if result.idempotent: response.status_code = status.HTTP_200_OK logger.info( "device message created", extra={ "event": "device_message_create", "request_id": getattr(request.state, "request_id", None), "device_id": device_id, "child_id": device_identity.child_id, "conversation_id": result.conversation_id, "conversation_type": result.conversation_type, "idempotent": result.idempotent, }, ) return ConversationMessageCreateResponse( idempotent=result.idempotent, conversation_id=result.conversation_id, conversation_type=result.conversation_type, conversation_type_name=result.conversation_type_name, message=result.message, )