Function resumableSetStepTx

resumable_step_tx.go:54–124

(ctx context.Context, tx TTx, job *Job[TArgs], cursor json.RawMessage)

}

func resumableSetStepTx[TTx any, TArgs JobArgs](ctx context.Context, tx TTx, job *Job[TArgs], cursor json.RawMessage) (*Job[TArgs], error) {
	if job.State != rivertype.JobStateRunning {
		return nil, errors.New("job must be running")
	}

	state, ok := resumableStateFromContext(ctx)
	if !ok {
		return nil, errors.New("not inside a resumable step; must be called from within ResumableStep or ResumableStepCursor")
	}
	if state.StepName == "" {
		return nil, errors.New("not inside a resumable step; must be called from within ResumableStep or ResumableStepCursor")
	}

	step := state.StepName

	client := ClientFromContext[TTx](ctx)
	if client == nil {
		return nil, errors.New("client not found in context, can only work within a River worker")
	}

	metadataUpdates := map[string]any{
		rivercommon.MetadataKeyResumableStep: step,
	}

	state.CompletedStep = step
	if cursor != nil {
		if state.Cursors == nil {
			state.Cursors = make(map[string]json.RawMessage)
		}
		state.Cursors[step] = cursor
	}
	if len(state.Cursors) > 0 {
		metadataUpdates[rivercommon.MetadataKeyResumableCursor] = state.Cursors
	}

	workMetadataUpdates, hasWorkMetadataUpdates := jobexecutor.MetadataUpdatesFromWorkContext(ctx)
	if hasWorkMetadataUpdates {
		workMetadataUpdates[rivercommon.MetadataKeyResumableStep] = step
		if resumableCursorMetadata, ok := metadataUpdates[rivercommon.MetadataKeyResumableCursor]; ok {
			workMetadataUpdates[rivercommon.MetadataKeyResumableCursor] = resumableCursorMetadata
		}
	}

	metadataUpdatesBytes, err := json.Marshal(metadataUpdates)
	if err != nil {
		return nil, err
	}

	updatedJob, err := client.Driver().UnwrapExecutor(tx).JobUpdate(ctx, &riverdriver.JobUpdateParams{
		ID:              job.ID,
		MetadataDoMerge: true,
		Metadata:        metadataUpdatesBytes,
		Schema:          client.config.Schema,
	})
	if err != nil {
		if errors.Is(err, rivertype.ErrNotFound) {
			if _, isInsideTestWorker := ctx.Value(execution.ContextKeyInsideTestWorker{}).(bool); isInsideTestWorker {
				panic("to use ResumableSetStepTx or ResumableSetStepCursorTx in a rivertest.Worker, the job must be inserted into the database first")