func (s *Store) List(ctx context.Context, request *jobqueue.ListRequest) (*jobqueue.ListResponse, error) {
	s.stmtOnce.Do(s.initStmt)
	resp := &jobqueue.ListResponse{}
	columns := "id,topic,state,args,rank,priority,retry,max_retry,correlation_group,correlation_id,created,started,completed,last_mod"
	where := make(map[string]interface{})
	countBuilder := sq.Select("COUNT(*)").From("jobqueue_jobs")
	queryBuilder := sq.Select(columns).From("jobqueue_jobs")
	if v := request.Topic; v != "" {
		where["topic"] = v
	}
	if v := request.State; v != "" {
		where["state"] = v
	}
	if v := request.CorrelationGroup; v != "" {
		where["correlation_group"] = v
	}
	if v := request.CorrelationID; v != "" {
		where["correlation_id"] = v
	}
	countBuilder = sq.Select("COUNT(*)").From("jobqueue_jobs").Where(where)
	{
		sql, args, err := countBuilder.ToSql()
		if err != nil {
			return nil, s.wrapError(err)
		}
		err = s.db.QueryRowContext(ctx, sql, args...).Scan(
			&resp.Total,
		)
		if err != nil {
			return nil, s.wrapError(err)
		}
	}
	queryBuilder = sq.Select(columns).
		From("jobqueue_jobs").
		Where(where).
		OrderBy("last_mod DESC").
		Offset(uint64(request.Offset)).Limit(uint64(request.Limit))
	{
		sql, args, err := queryBuilder.ToSql()
		if err != nil {
			return nil, s.wrapError(err)
		}
		rows, err := s.db.QueryContext(ctx, sql, args...)
		if err != nil {
			return nil, s.wrapError(err)
		}
		defer rows.Close()
		for rows.Next() {
			var j Job
			err := rows.Scan(
				&j.ID,
				&j.Topic,
				&j.State,
				&j.Args,
				&j.Rank,
				&j.Priority,
				&j.Retry,
				&j.MaxRetry,
				&j.CorrelationGroup,
				&j.CorrelationID,
				&j.Created,
				&j.Started,
				&j.Completed,
				&j.LastMod,
			)
			if err != nil {
				return nil, s.wrapError(err)
			}
			job, err := j.ToJob()
			if err != nil {
				return nil, s.wrapError(err)
			}
			resp.Jobs = append(resp.Jobs, job)
		}
		if err := rows.Err(); err != nil {
			return nil, s.wrapError(err)
		}
	}
	return resp, nil
}
List returns a list of all jobs stored in the data store.
package mysql
import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"time"
	sq "github.com/Masterminds/squirrel"
	"github.com/go-sql-driver/mysql"
	"golang.org/x/sync/errgroup"
	"github.com/olivere/jobqueue"
	"github.com/olivere/jobqueue/mysql/internal"
)
const (
	mysqlSchema = `CREATE TABLE IF NOT EXISTS jobqueue_jobs (
id varchar(36) primary key,
topic varchar(255),
state varchar(30),
args text,
priority bigint,
retry integer,
max_retry integer,
correlation_id varchar(255),
created bigint,
started bigint,
completed bigint,
last_mod bigint,
index ix_jobs_topic (topic),
index ix_jobs_state (state),
index ix_jobs_priority (priority),
index ix_jobs_correlation_id (correlation_id),
index ix_jobs_created (created),
index ix_jobs_started (started),
index ix_jobs_completed (completed),
index ix_jobs_last_mod (last_mod));`
	mysqlUpdate001 = `ALTER TABLE jobqueue_jobs ADD rank INT NOT NULL DEFAULT '0', ADD INDEX ix_jobs_rank_priority (rank, priority);`
	mysqlUpdate002 = `ALTER TABLE jobqueue_jobs ADD correlation_group varchar(255), ADD INDEX ix_jobs_correlation_group_and_id (correlation_group, correlation_id);`
	mysqlUpdate003 = `ALTER TABLE jobqueue_jobs ADD INDEX ix_jobs_state_correlation_group_and_id (state, correlation_group, id);`
	mysqlUpdate004 = `ALTER TABLE jobqueue_jobs CHANGE COLUMN args args MEDIUMTEXT;`
)
type Store struct {
	db    *sql.DB
	debug bool
	stmtOnce           sync.Once
	createStmt         *sql.Stmt
	updateStmt         *sql.Stmt
	deleteStmt         *sql.Stmt
	nextStmt           *sql.Stmt
	lookupStmt         *sql.Stmt
	lookupByCorrIDStmt *sql.Stmt
}
type StoreOption func(*Store)
func NewStore(url string, options ...StoreOption) (*Store, error) {
	st := &Store{}
	for _, opt := range options {
		opt(st)
	}
	cfg, err := mysql.ParseDSN(url)
	if err != nil {
		return nil, err
	}
	dbname := cfg.DBName
	if dbname == "" {
		return nil, errors.New("no database specified")
	}
	cfg.DBName = ""
	initdb, err := sql.Open("mysql", cfg.FormatDSN())
	if err != nil {
		return nil, err
	}
	defer initdb.Close()
	_, err = initdb.Exec(fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`", dbname))
	if err != nil {
		return nil, err
	}
	st.db, err = sql.Open("mysql", url)
	if err != nil {
		return nil, err
	}
	_, err = st.db.Exec(mysqlSchema)
	if err != nil {
		return nil, err
	}
	var count int64
	err = st.db.QueryRow(`
	SELECT COUNT(*) AS cnt
		FROM information_schema.COLUMNS
		WHERE TABLE_SCHEMA = ?
		AND TABLE_NAME = 'jobqueue_jobs'
		AND COLUMN_NAME = 'rank'
	`, dbname).Scan(&count)
	if err != nil {
		return nil, err
	}
	if count == 0 {
		_, err = st.db.Exec(mysqlUpdate001)
		if err != nil {
			return nil, err
		}
	}
	err = st.db.QueryRow(`
		SELECT COUNT(*) AS cnt
			FROM information_schema.COLUMNS
			WHERE TABLE_SCHEMA = ?
			AND TABLE_NAME = 'jobqueue_jobs'
			AND COLUMN_NAME = 'correlation_group'
		`, dbname).Scan(&count)
	if err != nil {
		return nil, err
	}
	if count == 0 {
		_, err = st.db.Exec(mysqlUpdate002)
		if err != nil {
			return nil, err
		}
	}
	err = st.db.QueryRow(`
		SELECT COUNT(*) AS cnt
			FROM information_schema.STATISTICS
			WHERE TABLE_SCHEMA = ?
			AND TABLE_NAME = 'jobqueue_jobs'
			AND INDEX_NAME = 'ix_jobs_state_correlation_group_and_id'
		`, dbname).Scan(&count)
	if err != nil {
		return nil, err
	}
	if count == 0 {
		_, err = st.db.Exec(mysqlUpdate003)
		if err != nil {
			return nil, err
		}
	}
	err = st.db.QueryRow(`
		SELECT COUNT(*) AS cnt
			FROM information_schema.COLUMNS
			WHERE TABLE_SCHEMA = ?
			AND TABLE_NAME = 'jobqueue_jobs'
			AND COLUMN_NAME = 'args'
			AND DATA_TYPE = 'text'
		`, dbname).Scan(&count)
	if err != nil {
		return nil, err
	}
	if count == 1 {
		_, err = st.db.Exec(mysqlUpdate004)
		if err != nil {
			return nil, err
		}
	}
	return st, nil
}
func SetDebug(enabled bool) StoreOption {
	return func(s *Store) {
		s.debug = enabled
	}
}
func (s *Store) initStmt() {
	var err error
	s.createStmt, err = s.db.Prepare("INSERT INTO jobqueue_jobs (id,topic,state,args,rank,priority,retry,max_retry,correlation_group,correlation_id,created,started,completed,last_mod) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
	if err != nil {
		panic(err)
	}
	s.updateStmt, err = s.db.Prepare("UPDATE jobqueue_jobs SET topic=?,state=?,args=?,rank=?,priority=?,retry=?,max_retry=?,correlation_group=?,correlation_id=?,created=?,started=?,completed=?,last_mod=? WHERE id=?")
	if err != nil {
		panic(err)
	}
	s.deleteStmt, err = s.db.Prepare("DELETE FROM jobqueue_jobs WHERE id=?")
	if err != nil {
		panic(err)
	}
	s.nextStmt, err = s.db.Prepare("SELECT id,topic,state,args,rank,priority,retry,max_retry,correlation_group,correlation_id,created,started,completed,last_mod FROM jobqueue_jobs WHERE state=? ORDER BY rank DESC, priority DESC LIMIT 1")
	if err != nil {
		panic(err)
	}
	s.lookupStmt, err = s.db.Prepare("SELECT id,topic,state,args,rank,priority,retry,max_retry,correlation_group,correlation_id,created,started,completed,last_mod FROM jobqueue_jobs WHERE id=? LIMIT 1")
	if err != nil {
		panic(err)
	}
	s.lookupByCorrIDStmt, err = s.db.Prepare("SELECT id,topic,state,args,rank,priority,retry,max_retry,correlation_group,correlation_id,created,started,completed,last_mod FROM jobqueue_jobs WHERE correlation_id=? LIMIT 1")
	if err != nil {
		panic(err)
	}
}
func (s *Store) wrapError(err error) error {
	if internal.IsNotFound(err) {
		return jobqueue.ErrNotFound
	}
	return err
}
func (s *Store) Start(b jobqueue.StartupBehaviour) error {
	s.stmtOnce.Do(s.initStmt)
	if b == jobqueue.MarkAsFailed {
		ctx := context.Background()
		err := internal.RunInTxWithRetry(ctx, s.db, func(ctx context.Context, tx *sql.Tx) error {
			_, err := tx.ExecContext(
				ctx,
				`UPDATE jobqueue_jobs SET state = ?, completed = ? WHERE state = ?`,
				jobqueue.Failed,
				time.Now().UnixNano(),
				jobqueue.Working,
			)
			if err != nil {
				return err
			}
			return nil
		}, func(err error) bool {
			return internal.IsDeadlock(err)
		})
		if err != nil {
			return s.wrapError(err)
		}
	}
	return nil
}
func (s *Store) Create(ctx context.Context, job *jobqueue.Job) error {
	s.stmtOnce.Do(s.initStmt)
	j, err := newJob(job)
	if err != nil {
		return err
	}
	j.LastMod = j.Created
	err = internal.RunInTxWithRetry(ctx, s.db, func(ctx context.Context, tx *sql.Tx) error {
		res, err := tx.Stmt(s.createStmt).ExecContext(
			ctx,
			j.ID,
			j.Topic,
			j.State,
			j.Args,
			j.Rank,
			j.Priority,
			j.Retry,
			j.MaxRetry,
			j.CorrelationGroup,
			j.CorrelationID,
			j.Created,
			j.Started,
			j.Completed,
			j.LastMod,
		)
		if err != nil {
			return err
		}
		rowsAffected, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if rowsAffected != 1 {
			return err
		}
		return nil
	}, func(err error) bool {
		return internal.IsDeadlock(err)
	})
	return s.wrapError(err)
}
func (s *Store) Update(ctx context.Context, job *jobqueue.Job) error {
	s.stmtOnce.Do(s.initStmt)
	j, err := newJob(job)
	if err != nil {
		return err
	}
	err = internal.RunInTxWithRetry(ctx, s.db, func(ctx context.Context, tx *sql.Tx) error {
		var id string
		err := tx.QueryRowContext(
			ctx,
			`SELECT id FROM jobqueue_jobs WHERE id = ? AND last_mod = ? FOR UPDATE`,
			job.ID,
			job.Updated,
		).Scan(&id)
		if err != nil {
			return err
		}
		j.LastMod = time.Now().UnixNano()
		res, err := tx.Stmt(s.updateStmt).ExecContext(
			ctx,
			j.Topic,
			j.State,
			j.Args,
			j.Rank,
			j.Priority,
			j.Retry,
			j.MaxRetry,
			j.CorrelationGroup,
			j.CorrelationID,
			j.Created,
			j.Started,
			j.Completed,
			j.LastMod,
			j.ID,
		)
		if err != nil {
			return err
		}
		rowsAffected, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if rowsAffected != 1 {
			return err
		}
		job.Updated = j.LastMod
		return nil
	}, func(err error) bool {
		return internal.IsDeadlock(err)
	})
	return s.wrapError(err)
}
func (s *Store) Next() (*jobqueue.Job, error) {
	s.stmtOnce.Do(s.initStmt)
	var j Job
	ctx := context.Background()
	err := internal.RunWithRetry(ctx, s.db, func(ctx context.Context) error {
		err := s.nextStmt.QueryRowContext(ctx, jobqueue.Waiting).Scan(
			&j.ID,
			&j.Topic,
			&j.State,
			&j.Args,
			&j.Rank,
			&j.Priority,
			&j.Retry,
			&j.MaxRetry,
			&j.CorrelationGroup,
			&j.CorrelationID,
			&j.Created,
			&j.Started,
			&j.Completed,
			&j.LastMod,
		)
		if err != nil {
			return err
		}
		return nil
	}, func(err error) bool {
		return internal.IsDeadlock(err)
	})
	if internal.IsNotFound(err) {
		return nil, jobqueue.ErrNotFound
	}
	if err != nil {
		return nil, s.wrapError(err)
	}
	return j.ToJob()
}
func (s *Store) Delete(ctx context.Context, job *jobqueue.Job) error {
	s.stmtOnce.Do(s.initStmt)
	err := internal.RunInTxWithRetry(ctx, s.db, func(ctx context.Context, tx *sql.Tx) error {
		_, err := tx.Stmt(s.deleteStmt).ExecContext(ctx, job.ID)
		switch {
		case err == sql.ErrNoRows:
			return nil
		default:
			return err
		}
	}, func(err error) bool {
		return internal.IsDeadlock(err)
	})
	return s.wrapError(err)
}
func (s *Store) Lookup(ctx context.Context, id string) (*jobqueue.Job, error) {
	s.stmtOnce.Do(s.initStmt)
	var j Job
	err := internal.RunWithRetry(ctx, s.db, func(ctx context.Context) error {
		err := s.lookupStmt.QueryRowContext(ctx, id).Scan(
			&j.ID,
			&j.Topic,
			&j.State,
			&j.Args,
			&j.Rank,
			&j.Priority,
			&j.Retry,
			&j.MaxRetry,
			&j.CorrelationGroup,
			&j.CorrelationID,
			&j.Created,
			&j.Started,
			&j.Completed,
			&j.LastMod,
		)
		if err != nil {
			return err
		}
		return nil
	}, func(err error) bool {
		return internal.IsDeadlock(err)
	})
	if err != nil {
		return nil, s.wrapError(err)
	}
	return j.ToJob()
}
func (s *Store) LookupByCorrelationID(ctx context.Context, correlationID string) ([]*jobqueue.Job, error) {
	s.stmtOnce.Do(s.initStmt)
	var jobs []Job
	err := internal.RunWithRetry(ctx, s.db, func(ctx context.Context) error {
		rows, err := s.lookupByCorrIDStmt.QueryContext(ctx, correlationID)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var j Job
			err := rows.Scan(
				&j.ID,
				&j.Topic,
				&j.State,
				&j.Args,
				&j.Rank,
				&j.Priority,
				&j.Retry,
				&j.MaxRetry,
				&j.CorrelationGroup,
				&j.CorrelationID,
				&j.Created,
				&j.Started,
				&j.Completed,
				&j.LastMod,
			)
			if err != nil {
				return err
			}
			jobs = append(jobs, j)
		}
		if err := rows.Err(); err != nil {
			return err
		}
		return nil
	}, func(err error) bool {
		return internal.IsDeadlock(err)
	})
	if err != nil {
		return nil, s.wrapError(err)
	}
	result := make([]*jobqueue.Job, len(jobs))
	for i, j := range jobs {
		job, err := j.ToJob()
		if err != nil {
			return nil, s.wrapError(err)
		}
		result[i] = job
	}
	return result, nil
}