Fix CLI progress updates (#390)

Progress updates showing how many rows have been backfilled during
migration start were added in #139.

Progress updates were then broken in
#289.

Fix the progress updates to restore the intended behaviour from #139 and
add a test to ensure that callbacks are invoked during migration start.

Author: andrew-farries

pkg/roll/execute.go

@@ -21,7 +21,7 @@ func (m *Roll) Start(ctx context.Context, migration *migrations.Migration, cbs .
 	}
 
 	// perform backfills for the tables that require it
-	return m.performBackfills(ctx, tablesToBackfill)
+	return m.performBackfills(ctx, tablesToBackfill, cbs...)
 }
 
 // StartDDLOperations performs the DDL operations for the migration. This does
@@ -265,9 +265,9 @@ func (m *Roll) ensureView(ctx context.Context, version, name string, table schem
 	return nil
 }
 
-func (m *Roll) performBackfills(ctx context.Context, tables []*schema.Table) error {
+func (m *Roll) performBackfills(ctx context.Context, tables []*schema.Table, cbs ...migrations.CallbackFn) error {
 	for _, table := range tables {
-		if err := migrations.Backfill(ctx, m.pgConn, table); err != nil {
+		if err := migrations.Backfill(ctx, m.pgConn, table, cbs...); err != nil {
 			errRollback := m.Rollback(ctx)
 
 			return errors.Join(

pkg/roll/execute_test.go

@@ -637,6 +637,45 @@ func TestMigrationHooksAreInvoked(t *testing.T) {
 	})
 }
 
+func TestCallbacksAreInvokedOnMigrationStart(t *testing.T) {
+	t.Parallel()
+
+	testutils.WithMigratorAndConnectionToContainer(t, func(mig *roll.Roll, db *sql.DB) {
+		ctx := context.Background()
+
+		// Create a table
+		_, err := db.ExecContext(ctx, "CREATE TABLE users (id SERIAL PRIMARY KEY, name text)")
+		require.NoError(t, err)
+
+		// Insert some data
+		_, err = db.ExecContext(ctx,
+			"INSERT INTO users (id, name) VALUES (1, 'alice'), (2, 'bob')")
+		require.NoError(t, err)
+
+		// Define a mock callback
+		invoked := false
+		cb := func(n int64) { invoked = true }
+
+		// Start a migration that requires a backfill
+		err = mig.Start(ctx, &migrations.Migration{
+			Name: "02_change_type",
+			Operations: migrations.Operations{
+				&migrations.OpAlterColumn{
+					Table:  "users",
+					Column: "name",
+					Type:   ptr("varchar(255)"),
+					Up:     "name",
+					Down:   "name",
+				},
+			},
+		}, cb)
+		require.NoError(t, err)
+
+		// Ensure that the callback was invoked
+		assert.True(t, invoked)
+	})
+}
+
 func TestRollSchemaMethodReturnsCorrectSchema(t *testing.T) {
 	t.Parallel()