Fix realtime sync writing Msg tables

Author: 2977094657

src/wechat_decrypt_tool/routers/chat.py

@@ -1626,12 +1626,13 @@ def normalize(item: dict[str, Any]) -> dict[str, Any]:
 
             inserted = 0
             backfilled = 0
-            if new_rows and (not name2id_synced):
-                _best_effort_upsert_output_name2id_rows(
-                    msg_conn,
-                    account_name=account_dir.name,
-                    rows=new_rows,
-                )
+            if new_rows:
+                if not name2id_synced:
+                    _best_effort_upsert_output_name2id_rows(
+                        msg_conn,
+                        account_name=account_dir.name,
+                        rows=new_rows,
+                    )
 
                 # Insert older -> newer to keep sqlite btree locality similar to existing data.
                 values = [tuple(r.get(c) for c in insert_cols) for r in reversed(new_rows)]
@@ -1991,12 +1992,13 @@ def normalize(item: dict[str, Any]) -> dict[str, Any]:
 
         inserted = 0
         backfilled = 0
-        if new_rows and (not name2id_synced):
-            _best_effort_upsert_output_name2id_rows(
-                msg_conn,
-                account_name=account_dir.name,
-                rows=new_rows,
-            )
+        if new_rows:
+            if not name2id_synced:
+                _best_effort_upsert_output_name2id_rows(
+                    msg_conn,
+                    account_name=account_dir.name,
+                    rows=new_rows,
+                )
 
             values = [tuple(r.get(c) for c in insert_cols) for r in reversed(new_rows)]
             insert_t0 = time.perf_counter()

tests/test_chat_realtime_name2id_sync.py

@@ -112,6 +112,125 @@ def _fake_exec_query(_handle, *, kind, path, sql):
                 ],
             )
 
+    def test_sync_still_inserts_new_messages_when_name2id_is_up_to_date(self):
+        with TemporaryDirectory() as td:
+            account_dir = Path(td) / "acc"
+            account_dir.mkdir(parents=True, exist_ok=True)
+
+            username = "wxid_friend"
+            table_name = f"Msg_{hashlib.md5(username.encode('utf-8')).hexdigest()}"
+            msg_db_path = account_dir / "message_0.db"
+
+            conn = sqlite3.connect(str(msg_db_path))
+            try:
+                conn.execute("CREATE TABLE Name2Id (user_name TEXT, is_session INTEGER DEFAULT 1)")
+                conn.execute(
+                    """
+                    CREATE TABLE "{table_name}" (
+                        local_id INTEGER PRIMARY KEY,
+                        server_id INTEGER,
+                        local_type INTEGER,
+                        sort_seq INTEGER,
+                        real_sender_id INTEGER,
+                        create_time INTEGER,
+                        message_content TEXT,
+                        compress_content BLOB,
+                        packed_info_data BLOB
+                    )
+                    """.format(table_name=table_name)
+                )
+                conn.execute("INSERT INTO Name2Id(rowid, user_name, is_session) VALUES (1, ?, 1)", ("acc",))
+                conn.execute("INSERT INTO Name2Id(rowid, user_name, is_session) VALUES (2, ?, 1)", (username,))
+                conn.execute(
+                    f'INSERT INTO "{table_name}" '
+                    "(local_id, server_id, local_type, sort_seq, real_sender_id, create_time, message_content, compress_content, packed_info_data) "
+                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                    (10, 10010, 1, 10, 2, 1710000010, "old", None, None),
+                )
+                conn.commit()
+            finally:
+                conn.close()
+
+            session_conn = sqlite3.connect(str(account_dir / "session.db"))
+            try:
+                session_conn.execute(
+                    """
+                    CREATE TABLE SessionTable (
+                        username TEXT PRIMARY KEY,
+                        summary TEXT DEFAULT '',
+                        last_timestamp INTEGER DEFAULT 0,
+                        sort_timestamp INTEGER DEFAULT 0,
+                        last_msg_locald_id INTEGER DEFAULT 0,
+                        last_msg_type INTEGER DEFAULT 0,
+                        last_msg_sub_type INTEGER DEFAULT 0,
+                        last_msg_sender TEXT DEFAULT ''
+                    )
+                    """
+                )
+                session_conn.commit()
+            finally:
+                session_conn.close()
+
+            def _fake_exec_query(_handle, *, kind, path, sql):
+                self.assertEqual(kind, "message")
+                self.assertTrue(str(path).endswith("message_0.db"))
+                if "COUNT(1)" in sql:
+                    return [{"c": 2, "mx": 2}]
+                raise AssertionError(f"Unexpected SQL: {sql}")
+
+            live_messages = [
+                {
+                    "local_id": 11,
+                    "server_id": 10011,
+                    "local_type": 1,
+                    "sort_seq": 11,
+                    "real_sender_id": 2,
+                    "create_time": 1710000011,
+                    "message_content": "new message",
+                    "compress_content": None,
+                    "sender_username": username,
+                }
+            ]
+
+            with (
+                patch.object(
+                    chat_router,
+                    "_resolve_db_storage_message_paths",
+                    return_value=(Path(td) / "live_message_0.db", Path(td) / "message_resource.db"),
+                ),
+                patch.object(chat_router, "_wcdb_exec_query", side_effect=_fake_exec_query),
+                patch.object(chat_router, "_wcdb_get_messages", side_effect=[list(live_messages)]),
+                patch.object(chat_router, "_best_effort_upsert_output_name2id_rows") as mock_upsert_name2id,
+            ):
+                result = chat_router._sync_chat_realtime_messages_for_table(
+                    account_dir=account_dir,
+                    rt_conn=_DummyConn(),
+                    username=username,
+                    msg_db_path=msg_db_path,
+                    table_name=table_name,
+                    max_scan=50,
+                    backfill_limit=0,
+                )
+
+            self.assertEqual(result.get("inserted"), 1)
+            mock_upsert_name2id.assert_not_called()
+
+            conn = sqlite3.connect(str(msg_db_path))
+            try:
+                rows = conn.execute(
+                    f'SELECT local_id, server_id, real_sender_id, create_time, message_content FROM "{table_name}" ORDER BY local_id ASC'
+                ).fetchall()
+            finally:
+                conn.close()
+
+            self.assertEqual(
+                rows,
+                [
+                    (10, 10010, 2, 1710000010, "old"),
+                    (11, 10011, 2, 1710000011, "new message"),
+                ],
+            )
+
 
 if __name__ == "__main__":
     unittest.main()