Fix sourceDevice column type in unprocessed

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!');
+}

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 {

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>;

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,
+      },
+    ]);
+  });
+});