Method insertManyShared

client.go:1969–2024

The shared code path for all Insert and InsertMany methods. It takes a function that executes the actual insert operation and allows for different implementations of the insert query to be passed in, each mapping their results back to a common result type.

(
	ctx context.Context,
	tx riverdriver.ExecutorTx,
	insertParams []*rivertype.JobInsertParams,
	execute func(context.Context, []*riverdriver.JobInsertFastParams) ([]*rivertype.JobInsertResult, error),
)

// implementations of the insert query to be passed in, each mapping their
// results back to a common result type.
func (c *Client[TTx]) insertManyShared(
	ctx context.Context,
	tx riverdriver.ExecutorTx,
	insertParams []*rivertype.JobInsertParams,
	execute func(context.Context, []*riverdriver.JobInsertFastParams) ([]*rivertype.JobInsertResult, error),
) ([]*rivertype.JobInsertResult, error) {
	doInner := func(ctx context.Context) ([]*rivertype.JobInsertResult, error) {
		for _, params := range insertParams {
			for _, hook := range append(
				c.hookLookupGlobal.ByHookKind(hooklookup.HookKindInsertBegin),
				c.hookLookupByJob.ByJobArgs(params.Args).ByHookKind(hooklookup.HookKindInsertBegin)...,
			) {
				if err := hook.(rivertype.HookInsertBegin).InsertBegin(ctx, params); err != nil { //nolint:forcetypeassert
					return nil, err
				}
			}
		}

		finalInsertParams := sliceutil.Map(insertParams, func(params *rivertype.JobInsertParams) *riverdriver.JobInsertFastParams {
			return (*riverdriver.JobInsertFastParams)(params)
		})

		insertResults, err := execute(ctx, finalInsertParams)
		if err != nil {
			return insertResults, err
		}

		queues := make([]string, 0, 10)
		for _, params := range insertParams {
			if params.State == rivertype.JobStateAvailable {
				queues = append(queues, params.Queue)
			}
		}

		if err = c.maybeNotifyInsertForQueues(ctx, tx, queues); err != nil {
			return nil, err
		}

		return insertResults, nil
	}

	jobInsertMiddleware := c.middlewareLookupGlobal.ByMiddlewareKind(middlewarelookup.MiddlewareKindJobInsert)
	if len(jobInsertMiddleware) > 0 {
		// Wrap middlewares in reverse order so the one defined first is wrapped
		// as the outermost function and is first to receive the operation.
		for i := len(jobInsertMiddleware) - 1; i >= 0; i-- {
			middlewareItem := jobInsertMiddleware[i].(rivertype.JobInsertMiddleware) //nolint:forcetypeassert // capture the current middleware item
			previousDoInner := doInner                                               // Capture the current doInner function
			doInner = func(ctx context.Context) ([]*rivertype.JobInsertResult, error) {
				return middlewareItem.InsertMany(ctx, insertParams, previousDoInner)
			}
		}
	}

	return doInner(ctx)
}

// Validates input parameters for a batch insert operation and generates a set