1 files changed, 16 insertions, 3 deletions
diff --git a/synapse/storage/data_stores/main/deviceinbox.py b/synapse/storage/data_stores/main/deviceinbox.py
index f04aad0743..a23744f11c 100644
--- a/synapse/storage/data_stores/main/deviceinbox.py
+++ b/synapse/storage/data_stores/main/deviceinbox.py
@@ -358,8 +358,21 @@ class DeviceInboxStore(DeviceInboxWorkerStore, DeviceInboxBackgroundUpdateStore)
def _add_messages_to_local_device_inbox_txn(
self, txn, stream_id, messages_by_user_then_device
):
- sql = "UPDATE device_max_stream_id" " SET stream_id = ?" " WHERE stream_id < ?"
- txn.execute(sql, (stream_id, stream_id))
+ # Compatible method of performing an upsert
+ sql = "SELECT stream_id FROM device_max_stream_id"
+
+ txn.execute(sql)
+ rows = txn.fetchone()
+ if rows:
+ db_stream_id = rows[0]
+ if db_stream_id < stream_id:
+ # Insert the new stream_id
+ sql = "UPDATE device_max_stream_id SET stream_id = ?"
+ else:
+ # No rows, perform an insert
+ sql = "INSERT INTO device_max_stream_id (stream_id) VALUES (?)"
+
+ txn.execute(sql, (stream_id,))
local_by_user_then_device = {}
for user_id, messages_by_device in messages_by_user_then_device.items():
@@ -367,7 +380,7 @@ class DeviceInboxStore(DeviceInboxWorkerStore, DeviceInboxBackgroundUpdateStore)
devices = list(messages_by_device.keys())
if len(devices) == 1 and devices[0] == "*":
# Handle wildcard device_ids.
- sql = "SELECT device_id FROM devices" " WHERE user_id = ?"
+ sql = "SELECT device_id FROM devices WHERE user_id = ?"
txn.execute(sql, (user_id,))
message_json = json.dumps(messages_by_device["*"])
for row in txn:
|