fix(sql): make migrated created_at NOT NULL

migrateScheduleDates() adds each epoch column as nullable, fills it,
then renames it over the legacy column. created_at therefore ended up
nullable, while both a new 0.8 table and the 0.7 one declare it NOT
NULL.

On PostgreSQL and MySQL, the migration now sets the epoch column NOT
NULL once it is filled, before the legacy columns are dropped. MySQL
commits each schema change, so a failure there leaves a table the next
run resumes instead of one that already looks migrated. SQLite cannot
add NOT NULL to an existing column, so its epoch column is created NOT
NULL with a default of 0, which every row then overwrites. Removing that
default would need a table rebuild, which Kysely does not provide, so
both schema services keep the same result.

A row without created_at, possible only in a table changed by hand, now
fails the migration with an error that names the schedule, before the
first schema change. It used to fail with a raw database error and, on
MySQL, left the epoch columns behind.

Author: RomainLanz

src/services/knex_queue_schema.ts

@@ -2,7 +2,7 @@ import type { Knex } from 'knex'
 import {
   assertTimeZone,
   epochColumn,
-  legacyScheduleDateToEpoch,
+  legacyScheduleRowToEpochs,
   processTimeZone,
   scheduleDatesMigrationState,
   type ScheduleDateColumn,
@@ -235,12 +235,7 @@ export class KnexQueueSchemaService {
       const timeZone = dialect === 'sqlite' ? 'UTC' : writerTimeZone
       const updates = rows.map((row) => ({
         id: row.id,
-        values: Object.fromEntries(
-          state.legacy.map((name) => [
-            epochColumn(name),
-            legacyScheduleDateToEpoch(row[name], timeZone),
-          ])
-        ),
+        values: legacyScheduleRowToEpochs(row, state.legacy, timeZone),
       }))
 
       await this.#dropNextRunIndexes(trx, dialect, tableName)
@@ -250,21 +245,41 @@ export class KnexQueueSchemaService {
       )
       if (missingEpochColumns.length > 0) {
         await trx.schema.alterTable(tableName, (table) => {
-          for (const name of missingEpochColumns) table.bigint(epochColumn(name)).nullable()
+          for (const name of missingEpochColumns) {
+            // SQLite cannot add NOT NULL to an existing column, only to a new
+            // one with a default. Every row gets its value below.
+            if (name === 'created_at' && dialect === 'sqlite') {
+              table.bigint(epochColumn(name)).notNullable().defaultTo(0)
+            } else {
+              table.bigint(epochColumn(name)).nullable()
+            }
+          }
         })
       }
 
       for (const { id, values } of updates) {
         await trx(tableName).where('id', id).update(values)
       }
 
+      // created_at is required, as in a new table. Before the legacy columns
+      // are dropped, so a failure on MySQL leaves a table the next run resumes.
+      const epochColumns = [...new Set([...state.legacy, ...state.withEpochColumn])]
+      if (epochColumns.includes('created_at') && dialect !== 'sqlite') {
+        await trx.schema.alterTable(tableName, (table) => {
+          if (dialect === 'pg') {
+            table.dropNullable(epochColumn('created_at'))
+          } else {
+            table.bigint(epochColumn('created_at')).notNullable().alter()
+          }
+        })
+      }
+
       if (state.legacy.length > 0) {
         await trx.schema.alterTable(tableName, (table) => {
           table.dropColumns(...state.legacy)
         })
       }
 
-      const epochColumns = [...new Set([...state.legacy, ...state.withEpochColumn])]
       await trx.schema.alterTable(tableName, (table) => {
         for (const name of epochColumns) table.renameColumn(epochColumn(name), name)
       })

src/services/kysely_queue_schema.ts

@@ -2,7 +2,7 @@ import { sql, type AlterTableColumnAlteringBuilder, type Kysely, type Transactio
 import {
   assertTimeZone,
   epochColumn,
-  legacyScheduleDateToEpoch,
+  legacyScheduleRowToEpochs,
   processTimeZone,
   scheduleDatesMigrationState,
   type ScheduleDateColumn,
@@ -231,19 +231,23 @@ export class KyselyQueueSchemaService<DB> {
       const timeZone = this.#dialect === 'sqlite' ? 'UTC' : writerTimeZone
       const updates = rows.map((row) => ({
         id: row.id,
-        values: Object.fromEntries(
-          state.legacy.map((name) => [
-            epochColumn(name),
-            legacyScheduleDateToEpoch(row[name], timeZone),
-          ])
-        ),
+        values: legacyScheduleRowToEpochs(row, state.legacy, timeZone),
       }))
 
       await this.#dropNextRunIndexes(trx, tableName)
 
       for (const name of state.legacy) {
         if (state.withEpochColumn.includes(name)) continue
-        await trx.schema.alterTable(tableName).addColumn(epochColumn(name), 'bigint').execute()
+        await trx.schema
+          .alterTable(tableName)
+          .addColumn(epochColumn(name), 'bigint', (column) =>
+            // SQLite cannot add NOT NULL to an existing column, only to a new
+            // one with a default. Every row gets its value below.
+            name === 'created_at' && this.#dialect === 'sqlite'
+              ? column.notNull().defaultTo(0)
+              : column
+          )
+          .execute()
       }
 
       for (const { id, values } of updates) {
@@ -254,11 +258,25 @@ export class KyselyQueueSchemaService<DB> {
           .execute()
       }
 
+      // created_at is required, as in a new table. Before the legacy columns
+      // are dropped, so a failure on MySQL leaves a table the next run resumes.
+      const epochColumns = new Set([...state.legacy, ...state.withEpochColumn])
+      if (epochColumns.has('created_at') && this.#dialect !== 'sqlite') {
+        const alterTable = trx.schema.alterTable(tableName)
+        await (
+          this.#dialect === 'postgres'
+            ? alterTable.alterColumn(epochColumn('created_at'), (column) => column.setNotNull())
+            : alterTable.modifyColumn(epochColumn('created_at'), 'bigint', (column) =>
+                column.notNull()
+              )
+        ).execute()
+      }
+
       for (const name of state.legacy) {
         await trx.schema.alterTable(tableName).dropColumn(name).execute()
       }
 
-      for (const name of new Set([...state.legacy, ...state.withEpochColumn])) {
+      for (const name of epochColumns) {
         await trx.schema.alterTable(tableName).renameColumn(epochColumn(name), name).execute()
       }
 

src/services/schedule_dates.ts

@@ -110,6 +110,34 @@ export function legacyScheduleDateToEpoch(
   return epoch
 }
 
+/**
+ * The epoch values of the legacy date columns of one schedule row, keyed by
+ * epoch column. created_at is required once migrated, so a row without one
+ * fails the migration here, before the first schema change.
+ */
+export function legacyScheduleRowToEpochs(
+  row: Record<string, string | number | null>,
+  columns: readonly ScheduleDateColumn[],
+  wallClockTimeZone: string
+): Record<string, number | null> {
+  const values: Record<string, number | null> = {}
+
+  for (const name of columns) {
+    const epoch = legacyScheduleDateToEpoch(row[name], wallClockTimeZone)
+
+    if (name === 'created_at' && epoch === null) {
+      throw new Error(
+        `Cannot migrate schedule "${row.id}": its created_at is empty, and the migrated column ` +
+          `is required. Set a date, then run the migration again.`
+      )
+    }
+
+    values[epochColumn(name)] = epoch
+  }
+
+  return values
+}
+
 type WallClockParts = [
   year: number,
   month: number,

tests/schedule_dates.spec.ts

@@ -1,5 +1,8 @@
 import { test } from '@japa/runner'
-import { legacyScheduleDateToEpoch } from '../src/services/schedule_dates.js'
+import {
+  legacyScheduleDateToEpoch,
+  legacyScheduleRowToEpochs,
+} from '../src/services/schedule_dates.js'
 
 const toIso = (value: number | null) => (value === null ? null : new Date(value).toISOString())
 
@@ -54,3 +57,27 @@ test.group('legacyScheduleDateToEpoch', () => {
     assert.throws(() => legacyScheduleDateToEpoch('not a date', 'UTC'), /Cannot convert/)
   })
 })
+
+test.group('legacyScheduleRowToEpochs', () => {
+  test('converts the legacy columns of a row into its epoch columns', ({ assert }) => {
+    const row = { id: 'a', next_run_at: '2026-07-01 12:00:00', last_run_at: null, created_at: 0 }
+
+    assert.deepEqual(
+      legacyScheduleRowToEpochs(row, ['next_run_at', 'last_run_at', 'created_at'], 'Europe/Paris'),
+      {
+        next_run_at__epoch: Date.parse('2026-07-01T10:00:00.000Z'),
+        last_run_at__epoch: null,
+        created_at__epoch: 0,
+      }
+    )
+  })
+
+  test('rejects a row without created_at, which the migrated table requires', ({ assert }) => {
+    const row = { id: 'no-created-at', next_run_at: null, created_at: null }
+
+    assert.throws(
+      () => legacyScheduleRowToEpochs(row, ['next_run_at', 'created_at'], 'UTC'),
+      /Cannot migrate schedule "no-created-at": its created_at is empty/
+    )
+  })
+})

tests/schedule_dates_migration.spec.ts

@@ -220,6 +220,7 @@ for (const dialect of ['sqlite', 'postgres', 'mysql'] as const) {
 
       const column = await connection(TABLE).columnInfo('next_run_at')
       assert.match(column.type, /int/i)
+      assert.isFalse((await connection(TABLE).columnInfo('created_at')).nullable)
       assert.equal((await connection(TABLE).where('id', 'legacy').first()).tenant, 'acme')
 
       await assertMigratedSchedules(assert, adapter(), createdAt)
@@ -249,6 +250,10 @@ for (const dialect of ['sqlite', 'postgres', 'mysql'] as const) {
         timezone: WRITER_TIME_ZONE,
       })
 
+      // SQLite migrates in one transaction, so it never resumes from nullable epoch columns.
+      if (dialect !== 'sqlite') {
+        assert.isFalse((await connection(TABLE).columnInfo('created_at')).nullable)
+      }
       await assertMigratedSchedules(assert, adapter(), createdAt)
     })
 
@@ -487,6 +492,10 @@ for (const dialect of ['sqlite', 'postgres', 'mysql'] as const) {
       const table = (await connection.introspection.getTables()).find(({ name }) => name === TABLE)
       return new Map(table!.columns.map(({ name, dataType }) => [name, dataType]))
     }
+    const isNullable = async (column: string) => {
+      const table = (await connection.introspection.getTables()).find(({ name }) => name === TABLE)
+      return table!.columns.find(({ name }) => name === column)!.isNullable
+    }
 
     test('converts the 0.7 dates to epoch milliseconds', async ({ assert }) => {
       const createdAt = await insertLegacySchedule()
@@ -495,6 +504,7 @@ for (const dialect of ['sqlite', 'postgres', 'mysql'] as const) {
       await schema.migrateScheduleDates(TABLE, { timezone: WRITER_TIME_ZONE })
 
       assert.match((await columnTypes()).get('next_run_at')!, /int/i)
+      assert.isFalse(await isNullable('created_at'))
       await assertMigratedSchedules(assert, adapter(), createdAt)
     })
 
@@ -522,6 +532,10 @@ for (const dialect of ['sqlite', 'postgres', 'mysql'] as const) {
         timezone: WRITER_TIME_ZONE,
       })
 
+      // SQLite migrates in one transaction, so it never resumes from nullable epoch columns.
+      if (dialect !== 'sqlite') {
+        assert.isFalse(await isNullable('created_at'))
+      }
       await assertMigratedSchedules(assert, adapter(), createdAt)
     })