From 96718464efc93b9073eb2aa4c56cecc5d18ba38d Mon Sep 17 00:00:00 2001
From: Fedor Indutny <79877362+indutny-signal@users.noreply.github.com>
Date: Wed, 22 Jan 2025 11:04:28 -0800
Subject: [PATCH] Fix sourceDevice column type in unprocessed

---
 .../1290-int-unprocessed-source-device.ts     | 34 +++++++++
 ts/sql/migrations/index.ts                    |  6 +-
 ts/test-node/sql/helpers.ts                   |  9 ++-
 ts/test-node/sql/migration_1290_test.ts       | 75 +++++++++++++++++++
 4 files changed, 119 insertions(+), 5 deletions(-)
 create mode 100644 ts/sql/migrations/1290-int-unprocessed-source-device.ts
 create mode 100644 ts/test-node/sql/migration_1290_test.ts

diff --git a/ts/sql/migrations/1290-int-unprocessed-source-device.ts b/ts/sql/migrations/1290-int-unprocessed-source-device.ts
new file mode 100644
index 000000000..1ac1d85e9
--- /dev/null
+++ b/ts/sql/migrations/1290-int-unprocessed-source-device.ts
@@ -0,0 +1,34 @@
+// Copyright 2025 Signal Messenger, LLC
+// SPDX-License-Identifier: AGPL-3.0-only
+import type { LoggerType } from '../../types/Logging';
+import { sql } from '../util';
+import type { WritableDB } from '../Interface';
+
+export const version = 1290;
+
+export function updateToSchemaVersion1290(
+  currentVersion: number,
+  db: WritableDB,
+  logger: LoggerType
+): void {
+  if (currentVersion >= 1290) {
+    return;
+  }
+
+  db.transaction(() => {
+    const [query] = sql`
+      ALTER TABLE unprocessed RENAME COLUMN sourceDevice TO legacySourceDevice;
+      ALTER TABLE unprocessed ADD COLUMN sourceDevice INTEGER;
+
+      UPDATE unprocessed
+      SET sourceDevice = legacySourceDevice;
+
+      ALTER TABLE unprocessed DROP COLUMN legacySourceDevice;
+    `;
+    db.exec(query);
+
+    db.pragma('user_version = 1290');
+  })();
+
+  logger.info('updateToSchemaVersion1290: success!');
+}
diff --git a/ts/sql/migrations/index.ts b/ts/sql/migrations/index.ts
index 609bf3aaa..e700b502d 100644
--- a/ts/sql/migrations/index.ts
+++ b/ts/sql/migrations/index.ts
@@ -104,10 +104,11 @@ import { updateToSchemaVersion1240 } from './1240-defunct-call-links-table';
 import { updateToSchemaVersion1250 } from './1250-defunct-call-links-storage';
 import { updateToSchemaVersion1260 } from './1260-sync-tasks-rowid';
 import { updateToSchemaVersion1270 } from './1270-normalize-messages';
+import { updateToSchemaVersion1280 } from './1280-blob-unprocessed';
 import {
-  updateToSchemaVersion1280,
+  updateToSchemaVersion1290,
   version as MAX_VERSION,
-} from './1280-blob-unprocessed';
+} from './1290-int-unprocessed-source-device';
 import { DataWriter } from '../Server';
 
 function updateToSchemaVersion1(
@@ -2082,6 +2083,7 @@ export const SCHEMA_VERSIONS = [
   updateToSchemaVersion1260,
   updateToSchemaVersion1270,
   updateToSchemaVersion1280,
+  updateToSchemaVersion1290,
 ];
 
 export class DBVersionFromFutureError extends Error {
diff --git a/ts/test-node/sql/helpers.ts b/ts/test-node/sql/helpers.ts
index 5431c728f..aaa8b0a8f 100644
--- a/ts/test-node/sql/helpers.ts
+++ b/ts/test-node/sql/helpers.ts
@@ -34,7 +34,7 @@ export function updateToVersion(db: WritableDB, version: number): void {
 }
 
 type TableRows = ReadonlyArray<
-  Record<string, string | number | null | Record<string, unknown>>
+  Record<string, string | number | Buffer | null | Record<string, unknown>>
 >;
 
 export function insertData(
@@ -52,6 +52,9 @@ export function insertData(
     `
     ).run(
       Object.values(row).map(v => {
+        if (Buffer.isBuffer(v)) {
+          return v;
+        }
         if (v != null && typeof v === 'object') {
           return JSON.stringify(v);
         }
@@ -65,7 +68,7 @@ export function getTableData(db: ReadableDB, table: string): TableRows {
   return db
     .prepare(`SELECT * FROM ${table}`)
     .all()
-    .map((row: Record<string, string | number | null>) => {
+    .map((row: Record<string, string | number | Buffer | null>) => {
       const result: Record<
         string,
         string | number | null | Record<string, unknown>
@@ -79,7 +82,7 @@ export function getTableData(db: ReadableDB, table: string): TableRows {
           continue;
         }
         try {
-          if (typeof value !== 'string') {
+          if (typeof value !== 'string' || !value.trim().startsWith('{')) {
             throw new Error('skip');
           }
           result[key] = JSON.parse(value) as Record<string, unknown>;
diff --git a/ts/test-node/sql/migration_1290_test.ts b/ts/test-node/sql/migration_1290_test.ts
new file mode 100644
index 000000000..f0f9b5c2c
--- /dev/null
+++ b/ts/test-node/sql/migration_1290_test.ts
@@ -0,0 +1,75 @@
+// Copyright 2025 Signal Messenger, LLC
+// SPDX-License-Identifier: AGPL-3.0-only
+
+import { assert } from 'chai';
+
+import { type WritableDB } from '../../sql/Interface';
+import { createDB, updateToVersion, insertData, getTableData } from './helpers';
+
+const DEFAULTS = {
+  id: 'id',
+  type: 0,
+  timestamp: 1,
+  attempts: 2,
+  receivedAtCounter: 3,
+  urgent: 1,
+  story: 1,
+  serverGuid: 'guid',
+  serverTimestamp: 1,
+  isEncrypted: 0,
+  content: Buffer.from('68656c6c6f', 'hex'),
+  messageAgeSec: 1,
+  destinationServiceId: 'dest',
+};
+
+describe('SQL/updateToSchemaVersion1290', () => {
+  let db: WritableDB;
+
+  afterEach(() => {
+    db.close();
+  });
+
+  beforeEach(() => {
+    db = createDB();
+    updateToVersion(db, 1280);
+  });
+
+  it('transitions null sourceDevice', () => {
+    insertData(db, 'unprocessed', [
+      {
+        ...DEFAULTS,
+
+        sourceDevice: null,
+      },
+    ]);
+    updateToVersion(db, 1290);
+
+    assert.deepStrictEqual(getTableData(db, 'unprocessed'), [
+      {
+        ...DEFAULTS,
+
+        content: '68656c6c6f',
+      },
+    ]);
+  });
+
+  it('transitions number sourceDevice', () => {
+    insertData(db, 'unprocessed', [
+      {
+        ...DEFAULTS,
+
+        sourceDevice: '123',
+      },
+    ]);
+    updateToVersion(db, 1290);
+
+    assert.deepStrictEqual(getTableData(db, 'unprocessed'), [
+      {
+        ...DEFAULTS,
+
+        content: '68656c6c6f',
+        sourceDevice: 123,
+      },
+    ]);
+  });
+});