You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: migration.go
    path of file: ./repos/rudder-server/jobsdb
    
    the code of the file until where you have to start completion: package jobsdb

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/lib/pq"
	"github.com/samber/lo"

	"github.com/rudderlabs/rudder-go-kit/stats"
	"github.com/rudderlabs/rudder-server/jobsdb/internal/dsindex"
	"github.com/rudderlabs/rudder-server/jobsdb/internal/lock"
	"github.com/rudderlabs/rudder-server/utils/misc"
	. "github.com/rudderlabs/rudder-server/utils/tx" //nolint:staticcheck
)

// startMigrateDSLoop migrates jobs from src dataset (srcDS) to destination dataset (dest_ds)
// First all the unprocessed jobs are copied over. Then all the jobs which haven't
// completed (state is failed or waiting or waiting_retry or executiong) are copied
// over. Then the status (only the latest) is set for those jobs
func (jd *Handle) startMigrateDSLoop(ctx context.Context) {
	jd.backgroundGroup.Go(misc.WithBugsnag(func() error {
		jd.migrateDSLoop(ctx)
		return nil
	}))
}

func (jd *Handle) migrateDSLoop(ctx context.Context) {
	for {
		select {
		case <-jd.TriggerMigrateDS():
		case <-ctx.Done():
			return
		}
		migrate := func() error {
			start := time.Now()
			jd.logger.Debugw("Start", "operation", "migrateDSLoop")
			timeoutCtx, cancel := context.WithTimeout(ctx, jd.conf.migration.migrateDSTimeout.Load())
			defer cancel()
			err := jd.doMigrateDS(timeoutCtx)
			stats.Default.NewTaggedStat("migration_loop", stats.TimerType, stats.Tags{"customVal": jd.tablePrefix, "error": strconv.FormatBool(err != nil)}).Since(start)
			if err != nil {
				return fmt.Errorf("failed to migrate ds: %w", err)
			}
			return nil
		}
		if err := migrate(); err != nil && ctx.Err() == nil {
			if !jd.conf.skipMaintenanceError {
				panic(err)
			}
			jd.logger.Errorw("Failed to migrate ds", "error", err)
		}
	}
}

func (jd *Handle) doMigrateDS(ctx context.Context) error {
	if !jd.dsListLock.RTryLockWithCtx(ctx) {
		return fmt.Errorf("could not acquire a dslist read lock: %w", ctx.Err())
	}
	dsList := jd.getDSList()
	jd.dsListLock.RUnlock()

	if err := jd.cleanupStatusTables(ctx, dsList); err != nil {
		return err
	}

	migrateFrom, pendingJobsCount, insertBeforeDS, err := jd.getMigrationList(dsList)
	if err != nil {
		return fmt.Errorf("could not get migration list: %w", err)
	}
	if len(migrateFrom) == 0 {
		return nil
	}
	var l lock.LockToken
	var lockChan chan<- lock.LockToken

	lockStart := time.Now()
	err = jd.WithTx(func(tx *Tx) error {
		return jd.withDistributedSharedLock(ctx, tx, "schema_migrate", func() error { // cannot run while schema migration is running
			// Take the lock and run actual migration
			if !jd.dsMigrationLock.TryLockWithCtx(ctx) {
				return fmt.Errorf("failed to acquire lock: %w", ctx.Err())
			}
			defer jd.dsMigrationLock.Unlock()
			// repeat the check after the dsMigrationLock is acquired to get correct pending jobs count.
			// the pending jobs count cannot change after the dsMigrationLock is acquired
			migrateFrom, pendingJobsCount, insertBeforeDS, err = jd.getMigrationList(dsList)
			if err != nil {
				return fmt.Errorf("could not get migration list: %w", err)
			}
			if len(migrateFrom) == 0 {
				return nil
			}

			if pendingJobsCount > 0 { // migrate incomplete jobs
				var destination dataSetT
				if err := jd.dsListLock.WithLockInCtx(ctx, func(l lock.LockTok
    
func (jd *Handle) doMigrateDS(ctx context.Context) error {
	if !jd.dsListLock.RTryLockWithCtx(ctx) {
		return fmt.Errorf("could not acquire a dslist read lock: %w", ctx.Err())
	}
	dsList := jd.getDSList()
	jd.dsListLock.RUnlock()

	if err := jd.cleanupStatusTables(ctx, dsList); err != nil {
		return err
	}

	migrateFrom, pendingJobsCount, insertBeforeDS, err := jd.getMigrationList(dsList)
	if err != nil {
		return fmt.Errorf("could not get migration list: %w", err)
	}
	if len(migrateFrom) == 0 {
		return nil
	}
	var l lock.LockToken
	var lockChan chan<- lock.LockToken

	lockStart := time.Now()
	err = jd.WithTx(func(tx *Tx) error {
		return jd.withDistributedSharedLock(ctx, tx, "schema_migrate", func() error { // cannot run while schema migration is running
			// Take the lock and run actual migration
			if !jd.dsMigrationLock.TryLockWithCtx(ctx) {
				return fmt.Errorf("failed to acquire lock: %w", ctx.Err())
			}
			defer jd.dsMigrationLock.Unlock()
			// repeat the check after the dsMigrationLock is acquired to get correct pending jobs count.
			// the pending jobs count cannot change after the dsMigrationLock is acquired
			migrateFrom, pendingJobsCount, insertBeforeDS, err = jd.getMigrationList(dsList)
			if err != nil {
				return fmt.Errorf("could not get migration list: %w", err)
			}
			if len(migrateFrom) == 0 {
				return nil
			}

			if pendingJobsCount > 0 { // migrate incomplete jobs
				var destination dataSetT
				if err := jd.dsListLock.WithLockInCtx(ctx, func(l lock.LockToken) error {
					dsIdx, err := jd.computeNewIdxForIntraNodeMigration(l, insertBeforeDS)
					if err != nil {
						return fmt.Errorf("computing new index for intra-node migration: %w", err)
					}
					destination = newDataSet(jd.tablePrefix, dsIdx)
					return nil
				}); err != nil {
					return err
				}

				jd.logger.Infof("[[ migrateDSLoop ]]: Migrate from: %v", migrateFrom)
				jd.logger.Infof("[[ migrateDSLoop ]]: To: %v", destination)
				jd.logger.Infof("[[ migrateDSLoop ]]: Next: %v", insertBeforeDS)

				opPayload, err := json.Marshal(&journalOpPayloadT{From: migrateFrom, To: destination})
				if err != nil {
					return fmt.Errorf("failed to marshal journal payload: %w", err)
				}

				opID, err := jd.JournalMarkStartInTx(tx, migrateCopyOperation, opPayload)
				if err != nil {
					return fmt.Errorf("failed to mark journal start: %w", err)
				}

				err = jd.addDSInTx(tx, destination)
				if err != nil {
					return fmt.Errorf("failed to add dataset: %w", err)
				}

				totalJobsMigrated := 0
				var noJobsMigrated int
				for _, source := range migrateFrom {
					jd.logger.Infof("[[ migrateDSLoop ]]: Migrate: %v to: %v", source, destination)
					noJobsMigrated, err = jd.migrateJobsInTx(ctx, tx, source, destination)
					if err != nil {
						return fmt.Errorf("failed to migrate jobs: %w", err)
					}
					totalJobsMigrated += noJobsMigrated
				}
				if err = jd.journalMarkDoneInTx(tx, opID); err != nil {
					return fmt.Errorf("failed to mark journal done: %w", err)
				}
				jd.logger.Infof("[[ migrateDSLoop ]]: Total migrated %d jobs", totalJobsMigrated)
			}

			opPayload, err := json.Marshal(&journalOpPayloadT{From: migrateFrom})
			if err != nil {
				return fmt.Errorf("failed to marshal journal payload: %w", err)
			}
			opID, err := jd.JournalMarkStartInTx(tx, postMigrateDSOperation, opPayload)
			if err != nil {
				return fmt.Errorf("failed to mark journal start: %w", err)
			}
			// acquire an async lock, as this needs to be released after the transaction commits
			l, lockChan, err = jd.dsListLock.AsyncLockWithCtx(ctx)
			if err != nil {
				return fmt.Errorf("failed to acquire lock: %w", err)
			}
			if err = jd.postMigrateHandleDS(tx, migrateFrom); err != nil {
				return fmt.Errorf("failed to post migrate handle ds: %w", err)
			}
			if err = jd.journalMarkDoneInTx(tx, opID); err != nil {
				return fmt.Errorf("failed to mark journal done: %w", err)
			}
			return nil
		})
	})
	if l != nil {
		defer stats.Default.NewTaggedStat("migration_loop_lock", stats.TimerType, stats.Tags{"customVal": jd.tablePrefix}).Since(lockStart)
		defer func() { lockChan <- l }()
		if err == nil {
			if err = jd.doRefreshDSRangeList(l); err != nil {
				return fmt.Errorf("failed to refresh ds range list: %w", err)
			}
		}

	}
	return err
}