diff --git a/database/migrate_to_shared_talkingq.py b/database/migrate_to_shared_talkingq.py index 93ef260..49721c1 100644 --- a/database/migrate_to_shared_talkingq.py +++ b/database/migrate_to_shared_talkingq.py @@ -40,6 +40,7 @@ MINI_PROGRAM_TABLES = [ "device_alarm_events", "im_conversations", "im_messages", + "device_pending_voice_messages", "child_location_current", "child_location_history", ] diff --git a/database/talkingq_shared_schema.sql b/database/talkingq_shared_schema.sql index 7ac3f9e..c901dce 100644 --- a/database/talkingq_shared_schema.sql +++ b/database/talkingq_shared_schema.sql @@ -455,6 +455,29 @@ CREATE TABLE IF NOT EXISTS `im_messages` ( ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +-- Persistent unread voice-message index for device NFC owner-card playback. +-- Audio content is still stored by im_messages/media_file_key; this table only stores delivery state. +CREATE TABLE IF NOT EXISTS `device_pending_voice_messages` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `target_device_id` VARCHAR(64) NOT NULL, + `sender_device_id` VARCHAR(64) NULL, + `im_message_id` BIGINT NOT NULL, + `media_file_key` VARCHAR(255) NOT NULL, + `audio_url` VARCHAR(512) NULL, + `source` VARCHAR(32) NOT NULL DEFAULT 'unknown', + `status` VARCHAR(32) NOT NULL DEFAULT 'pending', + `delivery_count` INT NOT NULL DEFAULT 0, + `delivered_at` DATETIME NULL, + `listened_at` DATETIME NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_device_pending_voice_target_msg` (`target_device_id`, `im_message_id`), + KEY `idx_device_pending_voice_target_status_created` (`target_device_id`, `status`, `created_at`), + KEY `idx_device_pending_voice_sender` (`sender_device_id`), + KEY `idx_device_pending_voice_im_message` (`im_message_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + CREATE TABLE IF NOT EXISTS `child_location_current` ( `child_id` BIGINT NOT NULL, `device_id` VARCHAR(64) NOT NULL, diff --git a/mini-program/database/init.sql b/mini-program/database/init.sql index d26de6c..c8faaed 100644 --- a/mini-program/database/init.sql +++ b/mini-program/database/init.sql @@ -190,6 +190,28 @@ CREATE TABLE IF NOT EXISTS im_messages ( CREATE INDEX idx_im_msg_sender ON im_messages(sender_type, sender_id, created_at); CREATE INDEX idx_im_msg_receiver ON im_messages(receiver_type, receiver_id, created_at); +-- device_pending_voice_messages +CREATE TABLE IF NOT EXISTS device_pending_voice_messages ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + target_device_id VARCHAR(64) NOT NULL, + sender_device_id VARCHAR(64), + im_message_id BIGINT NOT NULL, + media_file_key VARCHAR(255) NOT NULL, + audio_url VARCHAR(512), + source VARCHAR(32) NOT NULL DEFAULT 'unknown', + status VARCHAR(32) NOT NULL DEFAULT 'pending', + delivery_count INT NOT NULL DEFAULT 0, + delivered_at DATETIME, + listened_at DATETIME, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT uq_device_pending_voice_target_msg UNIQUE (target_device_id, im_message_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +CREATE INDEX idx_device_pending_voice_target_status_created ON device_pending_voice_messages(target_device_id, status, created_at); +CREATE INDEX idx_device_pending_voice_sender ON device_pending_voice_messages(sender_device_id); +CREATE INDEX idx_device_pending_voice_im_message ON device_pending_voice_messages(im_message_id); + -- child_location_current CREATE TABLE IF NOT EXISTS child_location_current ( child_id BIGINT UNSIGNED PRIMARY KEY, diff --git a/talkingq-url/banban/dao/pending_voice_message.py b/talkingq-url/banban/dao/pending_voice_message.py new file mode 100644 index 0000000..abf6622 --- /dev/null +++ b/talkingq-url/banban/dao/pending_voice_message.py @@ -0,0 +1,175 @@ +from collections.abc import Mapping +from typing import Any + +from banban.dao import BaseDAO + + +PENDING_VOICE_STATUS_PENDING = "pending" +PENDING_VOICE_STATUS_DELIVERED = "delivered" +PENDING_VOICE_STATUS_LISTENED = "listened" +PENDING_VOICE_STATUS_EXPIRED = "expired" + + +class PendingVoiceMessageDAO(BaseDAO): + async def upsert_pending( + self, + *, + target_device_id: str, + im_message_id: int, + media_file_key: str, + sender_device_id: str | None = None, + audio_url: str | None = None, + source: str = "unknown", + ) -> None: + await self.execute( + """ + INSERT INTO device_pending_voice_messages ( + target_device_id, + sender_device_id, + im_message_id, + media_file_key, + audio_url, + source, + status + ) VALUES ( + :target_device_id, + :sender_device_id, + :im_message_id, + :media_file_key, + :audio_url, + :source, + :status + ) + ON DUPLICATE KEY UPDATE + sender_device_id = COALESCE(sender_device_id, VALUES(sender_device_id)), + audio_url = COALESCE(audio_url, VALUES(audio_url)), + source = VALUES(source), + updated_at = CURRENT_TIMESTAMP + """, + { + "target_device_id": target_device_id, + "sender_device_id": sender_device_id, + "im_message_id": im_message_id, + "media_file_key": media_file_key, + "audio_url": audio_url, + "source": source, + "status": PENDING_VOICE_STATUS_PENDING, + }, + ) + + async def list_pending_for_delivery( + self, + *, + target_device_id: str, + limit: int = 10, + ) -> list[Mapping[str, Any]]: + rows = ( + await self.execute( + """ + SELECT + id, + target_device_id, + sender_device_id, + im_message_id, + media_file_key, + audio_url, + source, + status, + delivery_count, + created_at + FROM device_pending_voice_messages + WHERE target_device_id = :target_device_id + AND status = :status + ORDER BY created_at ASC, id ASC + LIMIT :limit + """, + { + "target_device_id": target_device_id, + "status": PENDING_VOICE_STATUS_PENDING, + "limit": limit, + }, + ) + ).mappings().all() + return rows + + async def list_devices_with_pending(self, *, limit: int = 1000) -> list[Mapping[str, Any]]: + rows = ( + await self.execute( + """ + SELECT + target_device_id, + COUNT(*) AS pending_count, + MIN(created_at) AS first_pending_at + FROM device_pending_voice_messages + WHERE status = :status + GROUP BY target_device_id + ORDER BY first_pending_at ASC + LIMIT :limit + """, + { + "status": PENDING_VOICE_STATUS_PENDING, + "limit": limit, + }, + ) + ).mappings().all() + return rows + + async def mark_delivered(self, message_ids: list[int]) -> None: + if not message_ids: + return + placeholders, params = self._build_id_params(message_ids) + await self.execute( + f""" + UPDATE device_pending_voice_messages + SET status = :status, + delivery_count = delivery_count + 1, + delivered_at = COALESCE(delivered_at, CURRENT_TIMESTAMP), + updated_at = CURRENT_TIMESTAMP + WHERE id IN ({placeholders}) + AND status = :pending_status + """, + { + **params, + "status": PENDING_VOICE_STATUS_DELIVERED, + "pending_status": PENDING_VOICE_STATUS_PENDING, + }, + ) + + async def mark_listened(self, message_ids: list[int]) -> None: + if not message_ids: + return + placeholders, params = self._build_id_params(message_ids) + await self.execute( + f""" + UPDATE device_pending_voice_messages + SET status = :status, + listened_at = COALESCE(listened_at, CURRENT_TIMESTAMP), + updated_at = CURRENT_TIMESTAMP + WHERE id IN ({placeholders}) + """, + { + **params, + "status": PENDING_VOICE_STATUS_LISTENED, + }, + ) + + async def clear_pending_for_device(self, *, target_device_id: str) -> None: + await self.execute( + """ + UPDATE device_pending_voice_messages + SET status = :status, + updated_at = CURRENT_TIMESTAMP + WHERE target_device_id = :target_device_id + AND status = :pending_status + """, + { + "target_device_id": target_device_id, + "status": PENDING_VOICE_STATUS_EXPIRED, + "pending_status": PENDING_VOICE_STATUS_PENDING, + }, + ) + + def _build_id_params(self, message_ids: list[int]) -> tuple[str, dict[str, int]]: + params = {f"id_{index}": int(message_id) for index, message_id in enumerate(message_ids)} + placeholders = ", ".join(f":{key}" for key in params) + return placeholders, params diff --git a/talkingq-url/banban/routers/bindings.py b/talkingq-url/banban/routers/bindings.py index 3104c04..d880297 100644 --- a/talkingq-url/banban/routers/bindings.py +++ b/talkingq-url/banban/routers/bindings.py @@ -12,7 +12,7 @@ from banban.dao.binding import ( SESSION_STATUS_PENDING, ) from banban.security import get_current_user_id -from banban.service.binding import BindingError, BindingService +from banban.service.binding import BindingError, binding_service router = APIRouter(prefix="/bindings", tags=["bindings"]) @@ -91,9 +91,8 @@ async def start_bind( current_user_id: int = Depends(get_current_user_id), ) -> BindStartResponse: del request - service = BindingService() try: - bind_token, expires_at = await service.start_bind( + bind_token, expires_at = await binding_service.start_bind( current_user_id, payload.device_id, payload.serial_number, @@ -115,8 +114,7 @@ async def get_bind_session( current_user_id: int = Depends(get_current_user_id), ) -> BindSessionResponse: del request - service = BindingService() - session = await service.get_bind_session(bind_token, current_user_id) + session = await binding_service.get_bind_session(bind_token, current_user_id) if not session: raise HTTPException(status_code=404, detail="bind session not found") @@ -137,9 +135,8 @@ async def confirm_bind( current_user_id: int = Depends(get_current_user_id), ) -> BindConfirmResponse: del request - service = BindingService() try: - result = await service.confirm_bind(payload.bind_token, current_user_id) + result = await binding_service.confirm_bind(payload.bind_token, current_user_id) except BindingError as exc: raise HTTPException(status_code=exc.status_code, detail=str(exc)) except ValueError as exc: @@ -169,9 +166,8 @@ async def direct_bind( current_user_id: int = Depends(get_current_user_id), ) -> DirectBindResponse: del request - service = BindingService() try: - result = await service.direct_bind( + result = await binding_service.direct_bind( payload.device_id, payload.serial_number, payload.child_id, @@ -190,9 +186,8 @@ async def set_binding_child( current_user_id: int = Depends(get_current_user_id), ) -> DirectBindResponse: del request - service = BindingService() try: - result = await service.set_binding_child(device_id=device_id, child_id=payload.child_id, user_id=current_user_id) + result = await binding_service.set_binding_child(device_id=device_id, child_id=payload.child_id, user_id=current_user_id) except BindingError as exc: raise HTTPException(status_code=exc.status_code, detail=str(exc)) except ValueError as exc: @@ -206,8 +201,7 @@ async def get_current_binding( current_user_id: int = Depends(get_current_user_id), ): del request - service = BindingService() - binding = await service.get_current_binding(current_user_id) + binding = await binding_service.get_current_binding(current_user_id) if not binding: raise HTTPException(status_code=404, detail="no binding found") return binding @@ -221,8 +215,7 @@ async def list_bindings( current_user_id: int = Depends(get_current_user_id), ) -> BindingListResponse: del request - service = BindingService() - rows, has_more = await service.list_bindings(current_user_id, limit, cursor) + rows, has_more = await binding_service.list_bindings(current_user_id, limit, cursor) next_cursor = int(rows[-1]["id"]) if has_more and rows else None items = [ BindingListItem( @@ -244,8 +237,7 @@ async def get_binding( current_user_id: int = Depends(get_current_user_id), ) -> BindingGetResponse: del request - service = BindingService() - binding = await service.get_binding(device_id, current_user_id) + binding = await binding_service.get_binding(device_id, current_user_id) if not binding: raise HTTPException(status_code=404, detail="binding not found") return BindingGetResponse(**binding) @@ -258,8 +250,7 @@ async def unbind_device( current_user_id: int = Depends(get_current_user_id), ) -> None: del request - service = BindingService() - if not await service.unbind(device_id, current_user_id): + if not await binding_service.unbind(device_id, current_user_id): raise HTTPException(status_code=404, detail="binding not found") @@ -273,8 +264,7 @@ async def get_bind_history( ) -> BindHistoryResponse: del request, current_user_id cursor_dt = datetime.fromisoformat(cursor) if cursor else None - service = BindingService() - rows, has_more = await service.list_history(device_id, limit, cursor_dt) + rows, has_more = await binding_service.list_history(device_id, limit, cursor_dt) next_cursor = rows[-1]["bound_at"].isoformat() if has_more and rows else None items = [BindHistoryItem(**row) for row in rows] return BindHistoryResponse(items=items, total=len(items), next_cursor=next_cursor) diff --git a/talkingq-url/banban/routers/mqtt_router.py b/talkingq-url/banban/routers/mqtt_router.py index 0f39a8a..6dd134c 100644 --- a/talkingq-url/banban/routers/mqtt_router.py +++ b/talkingq-url/banban/routers/mqtt_router.py @@ -2,6 +2,7 @@ import logging from typing import Optional from fastapi import APIRouter, HTTPException, Query from banban.service.device_audio_cache import device_audio_cache_service +from banban.service.pending_voice_message import pending_voice_message_service from services.offline_audio_cache import offline_audio_cache from banban.schemas.mqtt_models import ( GPSQueryRequest, @@ -110,6 +111,7 @@ async def clear_nfc_unread( # current_user_id: int = Depends(get_current_user_id) ): await offline_audio_cache.clear_audio_urls(device_id) + await pending_voice_message_service.clear_pending_for_device(target_device_id=device_id) return CommandResponse(msg_id="", device_id=device_id, message="已清除未读留言") diff --git a/talkingq-url/banban/service/binding.py b/talkingq-url/banban/service/binding.py index 338de38..eb9fd37 100644 --- a/talkingq-url/banban/service/binding.py +++ b/talkingq-url/banban/service/binding.py @@ -239,3 +239,6 @@ class BindingService(DatabaseServiceBase): return rows, has_more finally: await db_session.close() + + +binding_service = BindingService() diff --git a/talkingq-url/banban/service/device_voice_archive.py b/talkingq-url/banban/service/device_voice_archive.py index 35fbe59..26407a8 100644 --- a/talkingq-url/banban/service/device_voice_archive.py +++ b/talkingq-url/banban/service/device_voice_archive.py @@ -37,6 +37,15 @@ class PreparedArchiveAudio: archive_format: str +@dataclass(frozen=True) +class DeviceVoiceArchiveResult: + sender_device_id: str + receiver_device_id: str + conversation_id: int + message_id: int + media_file_key: str + + class DeviceVoiceArchiveService(DatabaseServiceBase): def __init__(self) -> None: super().__init__(service_name="device_voice_archive") @@ -47,7 +56,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): sender_device_id: str, receiver_device_id: str, local_audio_path: str, - ) -> bool: + ) -> DeviceVoiceArchiveResult | None: archive_id = f"voice_archive:{uuid4().hex[:12]}" session_logger.info( sender_device_id, @@ -65,7 +74,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): archive_id, f"跳过语音归档: 本地音频文件不存在, local_audio_path={local_audio_path}", ) - return False + return None prepared_audio = None db_session = await self.get_session() @@ -76,7 +85,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): local_audio_path=local_audio_path, ) if prepared_audio is None: - return False + return None dao = ImDAO(db_session) sender_identity = await self._get_child_identity_by_device_id( @@ -94,14 +103,14 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): archive_id, f"跳过语音归档: 发送设备未绑定有效 child, device_id={sender_device_id}", ) - return False + return None if receiver_identity is None: session_logger.warning( sender_device_id, archive_id, f"跳过语音归档: 接收设备未绑定有效 child, device_id={receiver_device_id}", ) - return False + return None if sender_identity["child_id"] == receiver_identity["child_id"]: session_logger.warning( sender_device_id, @@ -111,7 +120,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): f"child_id={sender_identity['child_id']}" ), ) - return False + return None session_logger.info( sender_device_id, @@ -228,14 +237,22 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): f"archive_format={prepared_audio.archive_format}" ), ) - return True + if not message_row: + raise RuntimeError("message was not found after insert") + return DeviceVoiceArchiveResult( + sender_device_id=sender_device_id, + receiver_device_id=receiver_device_id, + conversation_id=int(conversation_id), + message_id=int(message_row["id"]), + media_file_key=stored_audio.file_key, + ) except MessageAudioStorageError as exc: session_logger.error( sender_device_id, archive_id, f"设备语音归档失败: COS 上传异常: {exc}", ) - return False + return None except Exception as exc: session_logger.error( sender_device_id, @@ -243,7 +260,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): f"设备语音归档失败: {exc}", exc_info=True, ) - return False + return None finally: if prepared_audio and prepared_audio.archive_format == "mp3" and prepared_audio.filepath != local_audio_path: if os.path.exists(prepared_audio.filepath): diff --git a/talkingq-url/banban/service/im.py b/talkingq-url/banban/service/im.py index d622aed..696e90f 100644 --- a/talkingq-url/banban/service/im.py +++ b/talkingq-url/banban/service/im.py @@ -4,12 +4,12 @@ import json from collections.abc import Mapping from pathlib import Path from typing import Any -from services.offline_audio_cache import offline_audio_cache from fastapi import HTTPException from services.database_service_base import DatabaseServiceBase from banban.service.device_audio_cache import device_audio_cache_service from banban.service.message_audio_storage import MessageAudioStorageService, MessageAudioStorageError -from banban.service.binding import BindingService +from banban.service.binding import binding_service +from banban.service.pending_voice_message import pending_voice_message_service try: from banban.dao.im import ImDAO, DeviceIdentity, ConversationMessageCreateResult from banban.schemas.im import ( @@ -223,6 +223,25 @@ class ImService(DatabaseServiceBase): message=presented_message, ) + async def _get_raw_message_media_file_key( + self, + *, + conversation_id: int, + client_msg_id: str, + ) -> str | None: + db_session = await self.get_session() + try: + dao = ImDAO(db_session) + message_row = await dao._get_message_by_conversation_client_id( + conversation_id=conversation_id, + client_msg_id=client_msg_id, + ) + if not message_row: + return None + return message_row["media_file_key"] + finally: + await db_session.close() + async def create_parent_child_message( self, *, @@ -305,28 +324,47 @@ class ImService(DatabaseServiceBase): ext_json=ext_json, ) + result = None try: result = await self.create_parent_child_message( parent_user_id=parent_user_id, child_id=child_id, payload=payload, ) - binding_service = BindingService() + pending_media_file_key = await self._get_raw_message_media_file_key( + conversation_id=result.conversation_id, + client_msg_id=client_msg_id, + ) + if not pending_media_file_key: + pending_media_file_key = stored_audio.file_key + if result.idempotent: + try: + await self.audio_storage.delete_audio(stored_audio.file_key) + except MessageAudioStorageError: + pass device = await binding_service.get_current_binding(parent_user_id) try: audio_url = await device_audio_cache_service.get_device_audio_url( - stored_audio.file_key, + pending_media_file_key, device_id=device.device_id, ) except Exception: - session_logger.error(device.device_id, "audio", f"failed to get device audio url: {stored_audio.file_key}", exc_info=True) - audio_url = stored_audio.file_key - await offline_audio_cache.add_audio_url(device.device_id, f"{audio_url}") + session_logger.error(device.device_id, "audio", f"failed to get device audio url: {pending_media_file_key}", exc_info=True) + audio_url = pending_media_file_key + await pending_voice_message_service.add_pending_message( + target_device_id=device.device_id, + sender_device_id=None, + im_message_id=result.message.id, + media_file_key=pending_media_file_key, + audio_url=f"{audio_url}", + source="parent_child_voice", + ) except Exception: - try: - await self.audio_storage.delete_audio(stored_audio.file_key) - except MessageAudioStorageError: - pass + if result is None: + try: + await self.audio_storage.delete_audio(stored_audio.file_key) + except MessageAudioStorageError: + pass raise return result diff --git a/talkingq-url/banban/service/pending_voice_message.py b/talkingq-url/banban/service/pending_voice_message.py new file mode 100644 index 0000000..f7729ba --- /dev/null +++ b/talkingq-url/banban/service/pending_voice_message.py @@ -0,0 +1,146 @@ +from dataclasses import dataclass + +from banban.dao.pending_voice_message import PendingVoiceMessageDAO +from banban.service.device_audio_cache import device_audio_cache_service +from services.database_service_base import DatabaseServiceBase +from utils.logger import session_logger + + +@dataclass(frozen=True) +class PendingVoicePlaybackItem: + pending_id: int + audio_url: str + im_message_id: int + media_file_key: str + source: str + + +class PendingVoiceMessageService(DatabaseServiceBase): + def __init__(self) -> None: + super().__init__(service_name="pending_voice_message") + + async def add_pending_message( + self, + *, + target_device_id: str, + im_message_id: int, + media_file_key: str, + sender_device_id: str | None = None, + audio_url: str | None = None, + source: str = "unknown", + ) -> None: + normalized_target_device_id = (target_device_id or "").strip() + normalized_media_file_key = (media_file_key or "").strip() + if not normalized_target_device_id or not normalized_media_file_key: + raise ValueError("target_device_id and media_file_key are required") + + db_session = await self.get_session() + try: + dao = PendingVoiceMessageDAO(db_session) + await dao.upsert_pending( + target_device_id=normalized_target_device_id, + sender_device_id=(sender_device_id or "").strip() or None, + im_message_id=int(im_message_id), + media_file_key=normalized_media_file_key, + audio_url=(audio_url or "").strip() or None, + source=source, + ) + await db_session.commit() + session_logger.info( + normalized_target_device_id, + "pending_voice", + ( + "待收听留言已持久化: " + f"im_message_id={im_message_id}, source={source}" + ), + ) + except Exception: + await db_session.rollback() + raise + finally: + await db_session.close() + + async def get_playback_items( + self, + *, + target_device_id: str, + limit: int = 10, + ) -> list[PendingVoicePlaybackItem]: + db_session = await self.get_session() + try: + dao = PendingVoiceMessageDAO(db_session) + rows = await dao.list_pending_for_delivery( + target_device_id=target_device_id, + limit=limit, + ) + finally: + await db_session.close() + + items: list[PendingVoicePlaybackItem] = [] + for row in rows: + media_file_key = str(row["media_file_key"]) + fallback_audio_url = (row["audio_url"] or "").strip() + try: + audio_url = await device_audio_cache_service.get_device_audio_url( + media_file_key, + device_id=target_device_id, + ) + except Exception as exc: + if fallback_audio_url: + audio_url = fallback_audio_url + else: + session_logger.error( + target_device_id, + "pending_voice", + f"待收听留言音频URL生成失败: media_file_key={media_file_key}, error={exc}", + exc_info=True, + ) + continue + items.append( + PendingVoicePlaybackItem( + pending_id=int(row["id"]), + audio_url=audio_url, + im_message_id=int(row["im_message_id"]), + media_file_key=media_file_key, + source=str(row["source"] or "unknown"), + ) + ) + return items + + async def mark_delivered(self, pending_ids: list[int]) -> None: + if not pending_ids: + return + db_session = await self.get_session() + try: + dao = PendingVoiceMessageDAO(db_session) + await dao.mark_delivered(pending_ids) + await db_session.commit() + except Exception: + await db_session.rollback() + raise + finally: + await db_session.close() + + async def clear_pending_for_device(self, *, target_device_id: str) -> None: + db_session = await self.get_session() + try: + dao = PendingVoiceMessageDAO(db_session) + await dao.clear_pending_for_device(target_device_id=target_device_id) + await db_session.commit() + except Exception: + await db_session.rollback() + raise + finally: + await db_session.close() + + async def list_devices_with_pending(self, *, limit: int = 1000) -> list[str]: + db_session = await self.get_session() + try: + dao = PendingVoiceMessageDAO(db_session) + rows = await dao.list_devices_with_pending(limit=limit) + finally: + await db_session.close() + return [str(row["target_device_id"]) for row in rows] + + +pending_voice_message_service = PendingVoiceMessageService() diff --git a/talkingq-url/database/models.py b/talkingq-url/database/models.py index c9eb83b..e29d3c3 100644 --- a/talkingq-url/database/models.py +++ b/talkingq-url/database/models.py @@ -365,6 +365,34 @@ class IMMessage(Base): deleted_at: Mapped[Optional[datetime]] = mapped_column(DateTime) +class DevicePendingVoiceMessage(Base): + __tablename__ = "device_pending_voice_messages" + __table_args__ = ( + UniqueConstraint("target_device_id", "im_message_id", name="uq_device_pending_voice_target_msg"), + Index("idx_device_pending_voice_target_status_created", "target_device_id", "status", "created_at"), + Index("idx_device_pending_voice_sender", "sender_device_id"), + Index("idx_device_pending_voice_im_message", "im_message_id"), + ) + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + target_device_id: Mapped[str] = mapped_column(String(64), nullable=False) + sender_device_id: Mapped[Optional[str]] = mapped_column(String(64)) + im_message_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + media_file_key: Mapped[str] = mapped_column(String(255), nullable=False) + audio_url: Mapped[Optional[str]] = mapped_column(String(512)) + source: Mapped[str] = mapped_column(String(32), nullable=False, server_default=text("'unknown'")) + status: Mapped[str] = mapped_column(String(32), nullable=False, server_default=text("'pending'")) + delivery_count: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0")) + delivered_at: Mapped[Optional[datetime]] = mapped_column(DateTime) + listened_at: Mapped[Optional[datetime]] = mapped_column(DateTime) + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP")) + updated_at: Mapped[datetime] = mapped_column( + DateTime, + server_default=text("CURRENT_TIMESTAMP"), + onupdate=datetime.utcnow, + ) + + class ChildLocationCurrent(Base): __tablename__ = "child_location_current" diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index d5d8d63..9bc6672 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -6,11 +6,12 @@ from typing import Awaitable, Callable, Dict, Optional import aiomqtt -from banban.service.binding import BindingService +from banban.service.binding import binding_service from banban.service.device_alarm import device_alarm_service from banban.service.device_setting import device_setting_service from banban.service.im import im_service from banban.service.location import location_service +from banban.service.pending_voice_message import pending_voice_message_service from config import settings from services.card_service import card_service from services.device_target_cache import device_target_cache @@ -275,8 +276,7 @@ class TalkingQMQTTService: nfc_uuid = data.get("uuid") # 卡片与设备绑定 # await card_service.activate_card(device_id=device_id, card_uuid=nfc_uuid) - service = BindingService() - result = await service.finalize_nfc_bind(device_id=device_id, card_uuid=nfc_uuid) + result = await binding_service.finalize_nfc_bind(device_id=device_id, card_uuid=nfc_uuid) if result is None: logger.warning(device_id, "", "[NFC bind] no pending bind session found") return @@ -431,6 +431,23 @@ class TalkingQMQTTService: is_owner = await card_service.check_card_ownership(nfc_uuid, device_id) if is_owner: + playback_items = await pending_voice_message_service.get_playback_items( + target_device_id=device_id, + ) + if playback_items: + params = { + f"url_{k}": item.audio_url + for k, item in enumerate(playback_items, start=1) + } + payload = {"msg_id": "005", "type": 0, "params": params} + published = await self._publish(topic, payload) + if published: + await pending_voice_message_service.mark_delivered( + [item.pending_id for item in playback_items] + ) + await offline_audio_cache.clear_audio_urls(device_id) + return + audio_urls = await offline_audio_cache.pop_audio_urls(device_id) if audio_urls: params = {f"url_{k}": audio_url for k, audio_url in enumerate(audio_urls, start=1)} @@ -511,13 +528,15 @@ class TalkingQMQTTService: if not self._connected: await self.connect() - async def _publish(self, topic: str, payload: dict): + async def _publish(self, topic: str, payload: dict) -> bool: await self._ensure_connected() try: await self._client.publish(topic, json.dumps(payload, ensure_ascii=False), qos=self.qos) logger.info("", "", f"mqtt publish topic={topic} payload={payload}") + return True except Exception as exc: logger.error("", "", f"mqtt publish failed: {exc}") + return False async def send_gps_query(self, device_id: str) -> str: await self._publish(f"device/{device_id}/command", {"msg_id": "001"}) diff --git a/talkingq-url/handlers/websocket_message_handler.py b/talkingq-url/handlers/websocket_message_handler.py index a5ee544..bea8f05 100644 --- a/talkingq-url/handlers/websocket_message_handler.py +++ b/talkingq-url/handlers/websocket_message_handler.py @@ -7,6 +7,7 @@ from handlers.audio_session_handler import handle_websocket_data from handlers.audio_file_handler import upload_message_audio from banban.service.device_audio_cache import device_audio_cache_service from banban.service.device_voice_archive import device_voice_archive_service +from banban.service.pending_voice_message import pending_voice_message_service from services.audio_session import audio_session_manager from services.interrupt_handler import interrupt_handler from services.task_manager import task_manager @@ -340,22 +341,32 @@ async def process_cached_audio(device_id: str, target_device_id: str, serial_num cached_audio, device_id=target_device_id, ) - await offline_audio_cache.add_audio_url(target_device_id, audio_url) - session_logger.info( - device_id, - "target", - f"设备留言已加入待收听队列: target_device_id={target_device_id}, audio_url={audio_url}", + archive_result = await device_voice_archive_service.archive_peer_voice_message( + sender_device_id=device_id, + receiver_device_id=target_device_id, + local_audio_path=str(local_audio_path), ) - - await task_manager.create_task( - device_voice_archive_service.archive_peer_voice_message( + if archive_result is not None: + await pending_voice_message_service.add_pending_message( + target_device_id=target_device_id, sender_device_id=device_id, - receiver_device_id=target_device_id, - local_audio_path=str(local_audio_path), - ), - device_id=device_id, - task_type="device_voice_archive", - ) + im_message_id=archive_result.message_id, + media_file_key=archive_result.media_file_key, + audio_url=audio_url, + source="device_peer_voice", + ) + session_logger.info( + device_id, + "target", + f"设备留言已加入待收听队列: target_device_id={target_device_id}, audio_url={audio_url}", + ) + else: + await offline_audio_cache.add_audio_url(target_device_id, audio_url) + session_logger.warning( + device_id, + "target", + "设备留言已写入内存待收听队列，但归档失败，重启后无法恢复这条留言", + ) websocket = await connection_manager.get_connection(device_id) if websocket and websocket.client_state.name == "CONNECTED": diff --git a/talkingq-url/main.py b/talkingq-url/main.py index 734b6af..37fe7d6 100644 --- a/talkingq-url/main.py +++ b/talkingq-url/main.py @@ -35,6 +35,13 @@ async def lifespan(app: FastAPI): ) init_directories() + try: + session_logger.system_info("startup", "初始化数据库...") + await init_db() + session_logger.system_info("startup", "数据库初始化和迁移完成") + except Exception as e: + session_logger.system_info("startup", f"数据库初始化失败: {str(e)}") + scheduler = TaskScheduler.get_instance() scheduler.start() scheduler.add_interval_task( @@ -57,12 +64,6 @@ async def lifespan(app: FastAPI): else: session_logger.system_info("startup", "未配置 TalkingQ_MQTT，MQTT 服务链接未启动") exit(1) - try: - session_logger.system_info("startup", "初始化数据库...") - await init_db() - session_logger.system_info("startup", "数据库初始化和迁移完成") - except Exception as e: - session_logger.system_info("startup", f"数据库初始化失败: {str(e)}") await role_manager.initialize() await register_services() diff --git a/talkingq-url/mysql/init/02-init.sql b/talkingq-url/mysql/init/02-init.sql index 7ac3f9e..c901dce 100644 --- a/talkingq-url/mysql/init/02-init.sql +++ b/talkingq-url/mysql/init/02-init.sql @@ -455,6 +455,29 @@ CREATE TABLE IF NOT EXISTS `im_messages` ( ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +-- Persistent unread voice-message index for device NFC owner-card playback. +-- Audio content is still stored by im_messages/media_file_key; this table only stores delivery state. +CREATE TABLE IF NOT EXISTS `device_pending_voice_messages` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `target_device_id` VARCHAR(64) NOT NULL, + `sender_device_id` VARCHAR(64) NULL, + `im_message_id` BIGINT NOT NULL, + `media_file_key` VARCHAR(255) NOT NULL, + `audio_url` VARCHAR(512) NULL, + `source` VARCHAR(32) NOT NULL DEFAULT 'unknown', + `status` VARCHAR(32) NOT NULL DEFAULT 'pending', + `delivery_count` INT NOT NULL DEFAULT 0, + `delivered_at` DATETIME NULL, + `listened_at` DATETIME NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_device_pending_voice_target_msg` (`target_device_id`, `im_message_id`), + KEY `idx_device_pending_voice_target_status_created` (`target_device_id`, `status`, `created_at`), + KEY `idx_device_pending_voice_sender` (`sender_device_id`), + KEY `idx_device_pending_voice_im_message` (`im_message_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + CREATE TABLE IF NOT EXISTS `child_location_current` ( `child_id` BIGINT NOT NULL, `device_id` VARCHAR(64) NOT NULL, diff --git a/talkingq-url/scripts/simulate_nfc_pending_voice_delivery.py b/talkingq-url/scripts/simulate_nfc_pending_voice_delivery.py new file mode 100644 index 0000000..5e1f85b --- /dev/null +++ b/talkingq-url/scripts/simulate_nfc_pending_voice_delivery.py @@ -0,0 +1,293 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import asyncio +import json +import sys +import time +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import aiomqtt +import pymysql +from pymysql.cursors import DictCursor + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from config import settings + + +@dataclass(frozen=True) +class DeviceCardContext: + device_id: str + serial_number: str + child_id: int + card_uuid: str + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Simulate a device NFC listen event and verify pending voice delivery through MQTT." + ) + parser.add_argument("--device-id", default="TalkingQ_XQSN00001001") + parser.add_argument("--broker", default=settings.talkingq_mqtt_broker or "127.0.0.1") + parser.add_argument("--port", type=int, default=settings.talkingq_mqtt_port or 1883) + parser.add_argument("--username", default=settings.talkingq_mqtt_username or None) + parser.add_argument("--password", default=settings.talkingq_mqtt_password or None) + parser.add_argument("--timeout", type=float, default=10.0) + parser.add_argument("--audio-url", default="http://127.0.0.1:8080/assets/audio/welcome.mp3") + parser.add_argument("--media-file-key", default="integration/nfc-pending-voice.mp3") + parser.add_argument("--keep-row", action="store_true", help="Keep the synthetic pending row after verification.") + return parser.parse_args() + + +def connect_db(): + return pymysql.connect( + host=settings.db_host, + port=settings.db_port, + user=settings.db_user, + password=settings.db_password, + database=settings.db_name, + charset="utf8mb4", + cursorclass=DictCursor, + autocommit=True, + ) + + +def ensure_pending_table_exists() -> None: + with connect_db() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT COUNT(*) AS cnt + FROM information_schema.tables + WHERE table_schema = %s + AND table_name = 'device_pending_voice_messages' + """, + (settings.db_name,), + ) + row = cursor.fetchone() + if int(row["cnt"]) == 0: + raise RuntimeError( + "device_pending_voice_messages table does not exist; start the backend once or run init_db first" + ) + + +def load_device_context(device_id: str) -> DeviceCardContext: + with connect_db() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT + da.device_id, + da.serial_number, + db.child_id, + c.card_uuid + FROM device_auth AS da + LEFT JOIN device_bindings AS db + ON db.device_id = da.device_id + AND db.status = 1 + LEFT JOIN cards AS c + ON c.device_id = da.device_id + AND c.status = 1 + WHERE da.device_id = %s + AND da.is_active = 1 + LIMIT 1 + """, + (device_id,), + ) + row = cursor.fetchone() + + if not row: + raise RuntimeError(f"active device_auth row not found for {device_id}") + if row["child_id"] is None: + raise RuntimeError(f"device {device_id} is not bound to any child") + if not row["card_uuid"]: + raise RuntimeError(f"device {device_id} does not have an active owner card") + return DeviceCardContext( + device_id=str(row["device_id"]), + serial_number=str(row["serial_number"]), + child_id=int(row["child_id"]), + card_uuid=str(row["card_uuid"]), + ) + + +def insert_synthetic_pending(ctx: DeviceCardContext, *, audio_url: str, media_file_key: str) -> int: + synthetic_message_id = int(time.time() * 1000) + source = f"integration_nfc_{uuid.uuid4().hex[:10]}" + with connect_db() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + INSERT INTO device_pending_voice_messages ( + target_device_id, + sender_device_id, + im_message_id, + media_file_key, + audio_url, + source, + status + ) VALUES (%s, %s, %s, %s, %s, %s, 'pending') + """, + ( + ctx.device_id, + "integration-sender", + synthetic_message_id, + media_file_key, + audio_url, + source, + ), + ) + pending_id = int(cursor.lastrowid) + print( + json.dumps( + { + "inserted_pending_id": pending_id, + "device_id": ctx.device_id, + "im_message_id": synthetic_message_id, + "source": source, + }, + ensure_ascii=False, + ), + flush=True, + ) + return pending_id + + +def fetch_pending_row(pending_id: int) -> dict[str, Any] | None: + with connect_db() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT + id, + target_device_id, + im_message_id, + audio_url, + source, + status, + delivery_count, + delivered_at + FROM device_pending_voice_messages + WHERE id = %s + """, + (pending_id,), + ) + return cursor.fetchone() + + +async def wait_for_pending_status( + pending_id: int, + *, + expected_status: str, + timeout: float, +) -> dict[str, Any]: + deadline = asyncio.get_running_loop().time() + timeout + last_row = None + while asyncio.get_running_loop().time() < deadline: + last_row = fetch_pending_row(pending_id) + if not last_row: + raise RuntimeError(f"pending row disappeared unexpectedly: id={pending_id}") + if last_row["status"] == expected_status: + return last_row + await asyncio.sleep(0.2) + raise RuntimeError( + f"expected pending row status {expected_status}, " + f"got {last_row['status'] if last_row else 'missing'}" + ) + + +def delete_pending_row(pending_id: int) -> None: + with connect_db() as connection: + with connection.cursor() as cursor: + cursor.execute("DELETE FROM device_pending_voice_messages WHERE id = %s", (pending_id,)) + + +async def wait_for_delivery_response(client: aiomqtt.Client, *, timeout: float) -> dict[str, Any]: + deadline = asyncio.get_running_loop().time() + timeout + async for message in client.messages: + remaining = deadline - asyncio.get_running_loop().time() + if remaining <= 0: + break + try: + payload = json.loads(message.payload.decode("utf-8")) + except json.JSONDecodeError: + continue + print( + json.dumps( + { + "received_topic": str(message.topic), + "payload": payload, + }, + ensure_ascii=False, + ), + flush=True, + ) + if payload.get("msg_id") == "005" and payload.get("type") == 0: + params = payload.get("params") or {} + if any(str(key).startswith("url_") for key in params): + return payload + raise TimeoutError("timed out waiting for msg_id=005 delivery response") + + +async def run() -> None: + args = parse_args() + ensure_pending_table_exists() + ctx = load_device_context(args.device_id) + pending_id = insert_synthetic_pending( + ctx, + audio_url=args.audio_url, + media_file_key=args.media_file_key, + ) + + try: + event_topic = f"device/{ctx.device_id}/event" + response_topic = f"device/{ctx.device_id}/event_resp" + nfc_payload = {"msg_id": "005", "params": {"uuid": ctx.card_uuid}} + async with aiomqtt.Client( + hostname=args.broker, + port=args.port, + username=args.username, + password=args.password, + ) as client: + await client.subscribe(response_topic) + await client.publish(event_topic, json.dumps(nfc_payload, ensure_ascii=False)) + print( + json.dumps( + { + "published_topic": event_topic, + "payload": nfc_payload, + "subscribed_topic": response_topic, + }, + ensure_ascii=False, + ), + flush=True, + ) + response = await asyncio.wait_for( + wait_for_delivery_response(client, timeout=args.timeout), + timeout=args.timeout, + ) + + row = await wait_for_pending_status( + pending_id, + expected_status="delivered", + timeout=args.timeout, + ) + print(json.dumps({"pending_after_delivery": row}, ensure_ascii=False, default=str), flush=True) + if int(row["delivery_count"]) != 1: + raise RuntimeError(f"expected delivery_count 1, got {row['delivery_count']}") + params = response.get("params") or {} + if args.audio_url not in params.values(): + raise RuntimeError(f"expected delivered url {args.audio_url}, got {params}") + finally: + if not args.keep_row: + delete_pending_row(pending_id) + + +if __name__ == "__main__": + asyncio.run(run()) diff --git a/talkingq-url/services/database_service_base.py b/talkingq-url/services/database_service_base.py index 23c329a..61a2c1e 100644 --- a/talkingq-url/services/database_service_base.py +++ b/talkingq-url/services/database_service_base.py @@ -3,6 +3,8 @@ from utils.logger import session_logger from database.connection import get_db_manager class DatabaseServiceBase: + _logged_initialized_services = set() + """ 数据库服务基类，提供通用的数据库初始化和会话管理功能 所有需要访问数据库的服务类都应该继承此基类 @@ -19,7 +21,9 @@ class DatabaseServiceBase: try: self.db_manager = await get_db_manager() self._db_initialized = True - session_logger.info("system", self.service_name, f"{self.service_name}数据库连接初始化成功") + if self.service_name not in self._logged_initialized_services: + self._logged_initialized_services.add(self.service_name) + session_logger.info("system", self.service_name, f"{self.service_name}数据库连接初始化成功") except Exception as e: session_logger.error("system", self.service_name, f"{self.service_name}数据库连接初始化失败: {str(e)}") raise diff --git a/talkingq-url/services/scheduler.py b/talkingq-url/services/scheduler.py index 834a268..7ec7762 100644 --- a/talkingq-url/services/scheduler.py +++ b/talkingq-url/services/scheduler.py @@ -3,6 +3,7 @@ from typing import Optional, Dict from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.interval import IntervalTrigger +from banban.service.pending_voice_message import pending_voice_message_service from services.connection_manager import connection_manager from handlers.mqtt_handler import TalkingQMQTTService from services.offline_audio_cache import offline_audio_cache @@ -40,11 +41,14 @@ class TaskScheduler: async def _execute_task(self): try: service = await TalkingQMQTTService.get_instance() + pending_devices = await pending_voice_message_service.list_devices_with_pending() audio_cache = await offline_audio_cache.get_all_audio_cache() - if not audio_cache: + fallback_devices = list(audio_cache.keys()) if audio_cache else [] + device_ids = list(dict.fromkeys([*pending_devices, *fallback_devices])) + if not device_ids: logger.warning("", "", "无离线音频缓存，跳过本次执行") return - for device_id, audio_urls in audio_cache.items(): + for device_id in device_ids: websocket = await connection_manager.get_connection(device_id) if websocket is not None: audio_url = f"http://{settings.server_host}:{settings.server_port}/assets/audio/new_message.mp3" diff --git a/talkingq-url/tests/conftest.py b/talkingq-url/tests/conftest.py new file mode 100644 index 0000000..b9e1808 --- /dev/null +++ b/talkingq-url/tests/conftest.py @@ -0,0 +1,7 @@ +import sys +from pathlib import Path + + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) diff --git a/talkingq-url/tests/test_database_service_logging.py b/talkingq-url/tests/test_database_service_logging.py new file mode 100644 index 0000000..a941c04 --- /dev/null +++ b/talkingq-url/tests/test_database_service_logging.py @@ -0,0 +1,42 @@ +import pytest + +from services import database_service_base +from services.database_service_base import DatabaseServiceBase + + +@pytest.mark.asyncio +async def test_database_init_success_log_is_once_per_service(monkeypatch): + logs = [] + + async def fake_get_db_manager(): + return object() + + def fake_info(device_id, session_id, message): + logs.append((device_id, session_id, message)) + + DatabaseServiceBase._logged_initialized_services.clear() + monkeypatch.setattr(database_service_base, "get_db_manager", fake_get_db_manager) + monkeypatch.setattr(database_service_base.session_logger, "info", fake_info) + + first = DatabaseServiceBase(service_name="binding_service") + second = DatabaseServiceBase(service_name="binding_service") + + await first._init_database() + await second._init_database() + + assert logs == [ + ( + "system", + "binding_service", + "binding_service数据库连接初始化成功", + ) + ] + + +def test_binding_service_singleton_is_used_by_router_and_mqtt_handler(): + from banban.routers import bindings + from banban.service.binding import binding_service + from handlers import mqtt_handler + + assert bindings.binding_service is binding_service + assert mqtt_handler.binding_service is binding_service diff --git a/talkingq-url/tests/test_pending_voice_messages.py b/talkingq-url/tests/test_pending_voice_messages.py new file mode 100644 index 0000000..50a7e30 --- /dev/null +++ b/talkingq-url/tests/test_pending_voice_messages.py @@ -0,0 +1,230 @@ +from types import SimpleNamespace + +import pytest + +from banban.dao.pending_voice_message import PendingVoiceMessageDAO +from banban.service.pending_voice_message import ( + PendingVoiceMessageService, + PendingVoicePlaybackItem, +) +from handlers.mqtt_handler import TalkingQMQTTService + + +class FakeResult: + def __init__(self, rows=None): + self._rows = rows or [] + + def mappings(self): + return self + + def all(self): + return self._rows + + +class FakeSession: + def __init__(self, rows=None): + self.rows = rows or [] + self.calls = [] + self.commits = 0 + self.rollbacks = 0 + self.closed = False + + async def execute(self, statement, params=None): + self.calls.append((str(statement), params or {})) + return FakeResult(self.rows) + + async def commit(self): + self.commits += 1 + + async def rollback(self): + self.rollbacks += 1 + + async def close(self): + self.closed = True + + +@pytest.mark.asyncio +async def test_upsert_pending_keeps_original_media_file_key_on_duplicate(): + session = FakeSession() + dao = PendingVoiceMessageDAO(session) + + await dao.upsert_pending( + target_device_id="TalkingQ_XQSN00001001", + sender_device_id="TalkingQ_XQSN00001002", + im_message_id=123, + media_file_key="audio/original.mp3", + audio_url="http://example.test/audio.mp3", + source="device_peer_voice", + ) + + sql, params = session.calls[0] + update_sql = sql.split("ON DUPLICATE KEY UPDATE", 1)[1] + assert "ON DUPLICATE KEY UPDATE" in sql + assert "media_file_key" not in update_sql + assert params["target_device_id"] == "TalkingQ_XQSN00001001" + assert params["im_message_id"] == 123 + assert params["media_file_key"] == "audio/original.mp3" + assert params["status"] == "pending" + + +@pytest.mark.asyncio +async def test_mark_delivered_only_advances_pending_rows(): + session = FakeSession() + dao = PendingVoiceMessageDAO(session) + + await dao.mark_delivered([7, 9]) + + sql, params = session.calls[0] + assert "SET status = :status" in sql + assert "delivery_count = delivery_count + 1" in sql + assert "AND status = :pending_status" in sql + assert params == { + "id_0": 7, + "id_1": 9, + "status": "delivered", + "pending_status": "pending", + } + + +@pytest.mark.asyncio +async def test_get_playback_items_uses_fallback_url_when_signed_url_fails(monkeypatch): + service = PendingVoiceMessageService() + session = FakeSession( + rows=[ + { + "id": 11, + "target_device_id": "TalkingQ_XQSN00001001", + "sender_device_id": "TalkingQ_XQSN00001002", + "im_message_id": 123, + "media_file_key": "cos/audio.mp3", + "audio_url": "http://cached.example/audio.mp3", + "source": "device_peer_voice", + "status": "pending", + "delivery_count": 0, + "created_at": None, + } + ] + ) + + async def fake_get_session(): + return session + + class FailingAudioCache: + async def get_device_audio_url(self, media_file_key, device_id=None): + raise RuntimeError("cos unavailable") + + monkeypatch.setattr(service, "get_session", fake_get_session) + monkeypatch.setattr( + "banban.service.pending_voice_message.device_audio_cache_service", + FailingAudioCache(), + ) + + items = await service.get_playback_items(target_device_id="TalkingQ_XQSN00001001") + + assert items == [ + PendingVoicePlaybackItem( + pending_id=11, + audio_url="http://cached.example/audio.mp3", + im_message_id=123, + media_file_key="cos/audio.mp3", + source="device_peer_voice", + ) + ] + assert session.closed is True + + +@pytest.mark.asyncio +async def test_nfc_owner_listen_marks_pending_delivered_only_after_publish(monkeypatch): + service = TalkingQMQTTService({"device_prefix": "TalkingQ"}) + published = [] + marked = [] + cleared = [] + + class FakeCardService: + async def get_card_by_uuid(self, nfc_uuid): + return SimpleNamespace(device_id="TalkingQ_XQSN00001001") + + async def check_card_ownership(self, nfc_uuid, device_id): + return True + + class FakePendingVoiceService: + async def get_playback_items(self, target_device_id): + return [ + PendingVoicePlaybackItem( + pending_id=11, + audio_url="http://cached.example/audio.mp3", + im_message_id=123, + media_file_key="cos/audio.mp3", + source="device_peer_voice", + ) + ] + + async def mark_delivered(self, pending_ids): + marked.extend(pending_ids) + + class FakeOfflineAudioCache: + async def clear_audio_urls(self, device_id): + cleared.append(device_id) + + async def fake_publish(topic, payload): + published.append((topic, payload)) + return True + + monkeypatch.setattr("handlers.mqtt_handler.card_service", FakeCardService()) + monkeypatch.setattr("handlers.mqtt_handler.pending_voice_message_service", FakePendingVoiceService()) + monkeypatch.setattr("handlers.mqtt_handler.offline_audio_cache", FakeOfflineAudioCache()) + monkeypatch.setattr(service, "_publish", fake_publish) + + await service._send_nfc_listen_response("TalkingQ_XQSN00001001", "53DA2B6DA20001") + + assert published == [ + ( + "device/TalkingQ_XQSN00001001/event_resp", + { + "msg_id": "005", + "type": 0, + "params": {"url_1": "http://cached.example/audio.mp3"}, + }, + ) + ] + assert marked == [11] + assert cleared == ["TalkingQ_XQSN00001001"] + + +@pytest.mark.asyncio +async def test_nfc_owner_listen_keeps_pending_when_publish_fails(monkeypatch): + service = TalkingQMQTTService({"device_prefix": "TalkingQ"}) + marked = [] + + class FakeCardService: + async def get_card_by_uuid(self, nfc_uuid): + return SimpleNamespace(device_id="TalkingQ_XQSN00001001") + + async def check_card_ownership(self, nfc_uuid, device_id): + return True + + class FakePendingVoiceService: + async def get_playback_items(self, target_device_id): + return [ + PendingVoicePlaybackItem( + pending_id=11, + audio_url="http://cached.example/audio.mp3", + im_message_id=123, + media_file_key="cos/audio.mp3", + source="device_peer_voice", + ) + ] + + async def mark_delivered(self, pending_ids): + marked.extend(pending_ids) + + async def fake_publish(topic, payload): + return False + + monkeypatch.setattr("handlers.mqtt_handler.card_service", FakeCardService()) + monkeypatch.setattr("handlers.mqtt_handler.pending_voice_message_service", FakePendingVoiceService()) + monkeypatch.setattr(service, "_publish", fake_publish) + + await service._send_nfc_listen_response("TalkingQ_XQSN00001001", "53DA2B6DA20001") + + assert marked == []