MSP-6561 add requestId

enqueue.go

@@ -47,6 +47,9 @@ func requestsToJobs(list []EnqueueRequest) ([]Job, error) {
 		if req.Type == "" {
 			return nil, ErrTypeIsRequired
 		}
+		if req.RequestId == "" {
+			return nil, ErrRequestIdIsRequired
+		}
 
 		id := req.Id
 		if id == "" {
@@ -61,6 +64,7 @@ func requestsToJobs(list []EnqueueRequest) ([]Job, error) {
 			Queue:     req.Queue,
 			Type:      req.Type,
 			Arg:       req.Arg,
+			RequestId: req.RequestId,
 			Attempt:   0,
 			LastError: nil,
 			NextRunAt: now.Add(req.Delay).Unix(),
@@ -75,11 +79,11 @@ func requestsToJobs(list []EnqueueRequest) ([]Job, error) {
 
 func bulkInsert(ctx context.Context, e ExecerContext, jobs []Job) error {
 	valueStrings := make([]string, 0, len(jobs))
-	valueArgs := make([]interface{}, 0, len(jobs)*8)
+	valueArgs := make([]interface{}, 0, len(jobs)*9)
 	placeholderNum := 0
 	for _, job := range jobs {
 		placeholders := make([]string, 0)
-		for i := 0; i < 8; i++ {
+		for i := 0; i < 9; i++ {
 			placeholderNum++
 			placeholders = append(placeholders, fmt.Sprintf("$%d", placeholderNum))
 		}
@@ -94,9 +98,10 @@ func bulkInsert(ctx context.Context, e ExecerContext, jobs []Job) error {
 			job.NextRunAt,
 			job.CreatedAt,
 			job.UpdatedAt,
+			job.RequestId,
 		)
 	}
-	query := fmt.Sprintf("INSERT INTO bgjob_job (id, queue, type, arg, attempt, next_run_at, created_at, updated_at) VALUES %s",
+	query := fmt.Sprintf("INSERT INTO bgjob_job (id, queue, type, arg, attempt, next_run_at, created_at, updated_at, request_id) VALUES %s",
 		strings.Join(valueStrings, ","))
 	_, err := e.ExecContext(ctx, query, valueArgs...)
 	if err != nil {

enqueue_request.go

@@ -5,9 +5,10 @@ import (
 )
 
 type EnqueueRequest struct {
-	Id    string        //optional
-	Queue string        //required
-	Type  string        //required
-	Arg   []byte        //optional
-	Delay time.Duration //optional
+	Id        string        //optional
+	Queue     string        //required
+	Type      string        //required
+	Arg       []byte        //optional
+	Delay     time.Duration //optional
+	RequestId string        //required
 }

errors.go

@@ -5,10 +5,11 @@ import (
 )
 
 var (
-	ErrQueueIsRequired = errors.New("queue is required")
-	ErrTypeIsRequired  = errors.New("type is required")
-	ErrEmptyQueue      = errors.New("queue is empty")
-	ErrJobAlreadyExist = errors.New("job already exist")
+	ErrQueueIsRequired     = errors.New("queue is required")
+	ErrTypeIsRequired      = errors.New("type is required")
+	ErrEmptyQueue          = errors.New("queue is empty")
+	ErrJobAlreadyExist     = errors.New("job already exist")
+	ErrRequestIdIsRequired = errors.New("request_id is required")
 
 	ErrUnknownType = errors.New("unknown type")
 )

job.go

@@ -14,4 +14,5 @@ type Job struct {
 	NextRunAt int64
 	CreatedAt time.Time
 	UpdatedAt time.Time
+	RequestId string
 }

pg_store.go

@@ -23,7 +23,7 @@ func (p *pgStore) BulkInsert(ctx context.Context, jobs []Job) error {
 func (p *pgStore) Acquire(ctx context.Context, queue string, handler func(tx Tx) error) error {
 	return runTx(ctx, p.db, func(ctx context.Context, tx *sql.Tx) error {
 		query := `
-SELECT id, queue, type, arg, attempt, last_error, next_run_at, created_at, updated_at
+SELECT id, queue, type, arg, attempt, last_error, next_run_at, created_at, updated_at, request_id
 FROM bgjob_job
 WHERE queue = $1 AND next_run_at <= $2
 ORDER BY next_run_at, created_at
@@ -41,6 +41,7 @@ LIMIT 1 FOR UPDATE SKIP LOCKED
 			&job.NextRunAt,
 			&job.CreatedAt,
 			&job.UpdatedAt,
+			&job.RequestId,
 		)
 		if err == sql.ErrNoRows {
 			return ErrEmptyQueue
@@ -106,8 +107,8 @@ func (p *pgTx) Delete(ctx context.Context, id string) error {
 
 func (p *pgTx) SaveInDlq(ctx context.Context, job Job) error {
 	query := `INSERT INTO bgjob_dead_job 
-(job_id, queue, type, arg, attempt, next_run_at, last_error, job_created_at, job_updated_at, moved_at)
-VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
+(job_id, queue, type, arg, attempt, next_run_at, last_error, job_created_at, job_updated_at, moved_at, request_id)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
 `
 	_, err := p.tx.ExecContext(
 		ctx,
@@ -122,6 +123,7 @@ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
 		job.CreatedAt,
 		job.UpdatedAt,
 		timeNow(),
+		job.RequestId,
 	)
 	return err
 }