// Code generated by sqlc. DO NOT EDIT.
// versions:
//   sqlc v1.29.0
// source: concurrency.sql

package sqlcv1

import (
	"context"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgtype"
)

const advisoryLock = `-- name: AdvisoryLock :exec
SELECT pg_advisory_xact_lock($1::bigint)
`

func (q *Queries) AdvisoryLock(ctx context.Context, db DBTX, key int64) error {
	_, err := db.Exec(ctx, advisoryLock, key)
	return err
}

const checkStrategyActive = `-- name: CheckStrategyActive :one
WITH latest_workflow_version AS (
    SELECT DISTINCT ON("workflowId")
        "workflowId",
        wv."id" AS "workflowVersionId"
    FROM
        "WorkflowVersion" wv
    WHERE
        wv."workflowId" = $1::uuid
        AND wv."deletedAt" IS NULL
    ORDER BY "workflowId", "order" DESC
    LIMIT 1
), first_active_strategy AS (
    -- Get the first active strategy for the workflow version
    SELECT
        sc.id
    FROM
        v1_step_concurrency sc
    WHERE
        sc.tenant_id = $2::uuid AND
        sc.workflow_id = $1::uuid AND
        sc.workflow_version_id = $3::uuid AND
        sc.is_active = TRUE
    ORDER BY
        sc.id ASC
    LIMIT 1
), active_slot AS (
    SELECT
        sort_id, task_id, task_inserted_at, task_retry_count, external_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, parent_strategy_id, priority, key, is_filled, next_parent_strategy_ids, next_strategy_ids, next_keys, queue_to_notify, schedule_timeout_at
    FROM
        v1_concurrency_slot
    WHERE
        tenant_id = $2::uuid AND
        strategy_id = $4::bigint
        -- Note we don't check for is_filled=False, because is_filled=True could imply that the task
        -- gets retried and the slot is still active.
    LIMIT 1
), active_parent_slot AS (
    SELECT
        wcs.sort_id, wcs.tenant_id, wcs.workflow_id, wcs.workflow_version_id, wcs.workflow_run_id, wcs.strategy_id, wcs.completed_child_strategy_ids, wcs.child_strategy_ids, wcs.priority, wcs.key, wcs.is_filled
    FROM
        v1_concurrency_slot cs
    JOIN
        v1_workflow_concurrency_slot wcs ON (wcs.strategy_id, wcs.workflow_version_id, wcs.workflow_run_id) = (cs.parent_strategy_id, cs.workflow_version_id, cs.workflow_run_id)
    WHERE
        cs.tenant_id = $2::uuid AND
        cs.strategy_id = $4::bigint
        -- Note we don't check for is_filled=False, because is_filled=True could imply that the task
        -- gets retried and the slot is still active.
    LIMIT 1
), is_active AS (
    SELECT
        EXISTS(SELECT 1 FROM latest_workflow_version) AND
        (
            -- We must match the first active strategy, otherwise we could have another concurrency strategy
            -- that is active and has this concurrency strategy as a child.
            (first_active_strategy.id != $4::bigint) OR
            latest_workflow_version."workflowVersionId" = $3::uuid OR
            EXISTS(SELECT 1 FROM active_slot) OR
            EXISTS(SELECT 1 FROM active_parent_slot)
        ) AS "isActive"
    FROM
        latest_workflow_version, first_active_strategy
)
SELECT COALESCE((SELECT "isActive" FROM is_active), FALSE)::bool AS "isActive"
`

type CheckStrategyActiveParams struct {
	Workflowid        uuid.UUID `json:"workflowid"`
	Tenantid          uuid.UUID `json:"tenantid"`
	Workflowversionid uuid.UUID `json:"workflowversionid"`
	Strategyid        int64     `json:"strategyid"`
}

// A strategy is active if the workflow is not deleted, and it is attached to the latest workflow version or it has
// at least one concurrency slot that is not filled (the concurrency slot could be on the parent).
func (q *Queries) CheckStrategyActive(ctx context.Context, db DBTX, arg CheckStrategyActiveParams) (bool, error) {
	row := db.QueryRow(ctx, checkStrategyActive,
		arg.Workflowid,
		arg.Tenantid,
		arg.Workflowversionid,
		arg.Strategyid,
	)
	var isActive bool
	err := row.Scan(&isActive)
	return isActive, err
}

const deactivateStaleStepConcurrency = `-- name: DeactivateStaleStepConcurrency :exec
WITH tenant_step_concurrencies AS (
    SELECT sc.id
    FROM v1_step_concurrency sc
    WHERE sc.tenant_id = $1::UUID
        AND sc.is_active = TRUE
        AND NOT EXISTS (
            SELECT 1 FROM v1_concurrency_slot cs
            WHERE
                cs.strategy_id = sc.id
                AND cs.tenant_id = $1::UUID -- tenant id filter to force index usage
        )
        AND NOT EXISTS (
            SELECT 1 FROM v1_concurrency_slot cs
            JOIN v1_workflow_concurrency_slot wcs
            ON (wcs.strategy_id, wcs.workflow_version_id, wcs.workflow_run_id)
                = (cs.parent_strategy_id, cs.workflow_version_id, cs.workflow_run_id)
            WHERE cs.strategy_id = sc.id
        )
    FOR UPDATE SKIP LOCKED
)

UPDATE v1_step_concurrency sc
SET is_active = FALSE
FROM tenant_step_concurrencies
WHERE sc.id = tenant_step_concurrencies.id
`

func (q *Queries) DeactivateStaleStepConcurrency(ctx context.Context, db DBTX, tenantid uuid.UUID) error {
	_, err := db.Exec(ctx, deactivateStaleStepConcurrency, tenantid)
	return err
}

const getConcurrencyStrategyById = `-- name: GetConcurrencyStrategyById :one
SELECT
    sc.id, sc.parent_strategy_id, sc.workflow_id, sc.workflow_version_id, sc.step_id, sc.is_active, sc.strategy, sc.expression, sc.tenant_id, sc.max_concurrency
FROM
    v1_step_concurrency sc
WHERE
    sc.tenant_id = $1::uuid AND
    sc.id = $2::bigint
`

type GetConcurrencyStrategyByIdParams struct {
	Tenantid uuid.UUID `json:"tenantid"`
	ID       int64     `json:"id"`
}

func (q *Queries) GetConcurrencyStrategyById(ctx context.Context, db DBTX, arg GetConcurrencyStrategyByIdParams) (*V1StepConcurrency, error) {
	row := db.QueryRow(ctx, getConcurrencyStrategyById, arg.Tenantid, arg.ID)
	var i V1StepConcurrency
	err := row.Scan(
		&i.ID,
		&i.ParentStrategyID,
		&i.WorkflowID,
		&i.WorkflowVersionID,
		&i.StepID,
		&i.IsActive,
		&i.Strategy,
		&i.Expression,
		&i.TenantID,
		&i.MaxConcurrency,
	)
	return &i, err
}

const getWorkflowConcurrencyQueueCounts = `-- name: GetWorkflowConcurrencyQueueCounts :many
SELECT
    w."name" AS "workflowName",
    wcs.key,
    COUNT(*) AS "count"
FROM
    v1_workflow_concurrency_slot wcs
JOIN
    "Workflow" w ON w.id = wcs.workflow_id
WHERE
    wcs.tenant_id = $1::uuid
    AND wcs.is_filled = FALSE
GROUP BY
    w."name",
    wcs.key
`

type GetWorkflowConcurrencyQueueCountsRow struct {
	WorkflowName string `json:"workflowName"`
	Key          string `json:"key"`
	Count        int64  `json:"count"`
}

func (q *Queries) GetWorkflowConcurrencyQueueCounts(ctx context.Context, db DBTX, tenantid uuid.UUID) ([]*GetWorkflowConcurrencyQueueCountsRow, error) {
	rows, err := db.Query(ctx, getWorkflowConcurrencyQueueCounts, tenantid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*GetWorkflowConcurrencyQueueCountsRow
	for rows.Next() {
		var i GetWorkflowConcurrencyQueueCountsRow
		if err := rows.Scan(&i.WorkflowName, &i.Key, &i.Count); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listActiveConcurrencyStrategies = `-- name: ListActiveConcurrencyStrategies :many
SELECT
    sc.id, sc.parent_strategy_id, sc.workflow_id, sc.workflow_version_id, sc.step_id, sc.is_active, sc.strategy, sc.expression, sc.tenant_id, sc.max_concurrency
FROM
    v1_step_concurrency sc
JOIN
    "WorkflowVersion" wv ON wv."id" = sc.workflow_version_id
WHERE
    sc.tenant_id = $1::uuid AND
    sc.is_active = TRUE
`

func (q *Queries) ListActiveConcurrencyStrategies(ctx context.Context, db DBTX, tenantid uuid.UUID) ([]*V1StepConcurrency, error) {
	rows, err := db.Query(ctx, listActiveConcurrencyStrategies, tenantid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*V1StepConcurrency
	for rows.Next() {
		var i V1StepConcurrency
		if err := rows.Scan(
			&i.ID,
			&i.ParentStrategyID,
			&i.WorkflowID,
			&i.WorkflowVersionID,
			&i.StepID,
			&i.IsActive,
			&i.Strategy,
			&i.Expression,
			&i.TenantID,
			&i.MaxConcurrency,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listConcurrencyStrategiesByStepId = `-- name: ListConcurrencyStrategiesByStepId :many
SELECT
    id, parent_strategy_id, workflow_id, workflow_version_id, step_id, is_active, strategy, expression, tenant_id, max_concurrency
FROM
    v1_step_concurrency
WHERE
    tenant_id = $1::uuid AND
    step_id = ANY($2::uuid[])
`

type ListConcurrencyStrategiesByStepIdParams struct {
	Tenantid uuid.UUID   `json:"tenantid"`
	Stepids  []uuid.UUID `json:"stepids"`
}

func (q *Queries) ListConcurrencyStrategiesByStepId(ctx context.Context, db DBTX, arg ListConcurrencyStrategiesByStepIdParams) ([]*V1StepConcurrency, error) {
	rows, err := db.Query(ctx, listConcurrencyStrategiesByStepId, arg.Tenantid, arg.Stepids)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*V1StepConcurrency
	for rows.Next() {
		var i V1StepConcurrency
		if err := rows.Scan(
			&i.ID,
			&i.ParentStrategyID,
			&i.WorkflowID,
			&i.WorkflowVersionID,
			&i.StepID,
			&i.IsActive,
			&i.Strategy,
			&i.Expression,
			&i.TenantID,
			&i.MaxConcurrency,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listConcurrencyStrategiesByWorkflowVersionId = `-- name: ListConcurrencyStrategiesByWorkflowVersionId :many
SELECT c.id, c.parent_strategy_id, c.workflow_id, c.workflow_version_id, c.step_id, c.is_active, c.strategy, c.expression, c.tenant_id, c.max_concurrency, s."readableId" AS step_readable_id
FROM v1_step_concurrency c
JOIN "Step" s ON s.id = c.step_id
WHERE
    tenant_id = $1::UUID
    AND workflow_version_id = $2::UUID
    AND workflow_id = $3::UUID
    AND c.id NOT IN (
        SELECT UNNEST(child_strategy_ids)
        FROM v1_workflow_concurrency
        WHERE
            tenant_id = $1::UUID
            AND workflow_version_id = $2::UUID
            AND workflow_id = $3::UUID
    )
`

type ListConcurrencyStrategiesByWorkflowVersionIdParams struct {
	Tenantid          uuid.UUID `json:"tenantid"`
	Workflowversionid uuid.UUID `json:"workflowversionid"`
	Workflowid        uuid.UUID `json:"workflowid"`
}

type ListConcurrencyStrategiesByWorkflowVersionIdRow struct {
	ID                int64                 `json:"id"`
	ParentStrategyID  pgtype.Int8           `json:"parent_strategy_id"`
	WorkflowID        uuid.UUID             `json:"workflow_id"`
	WorkflowVersionID uuid.UUID             `json:"workflow_version_id"`
	StepID            uuid.UUID             `json:"step_id"`
	IsActive          bool                  `json:"is_active"`
	Strategy          V1ConcurrencyStrategy `json:"strategy"`
	Expression        string                `json:"expression"`
	TenantID          uuid.UUID             `json:"tenant_id"`
	MaxConcurrency    int32                 `json:"max_concurrency"`
	StepReadableID    pgtype.Text           `json:"step_readable_id"`
}

func (q *Queries) ListConcurrencyStrategiesByWorkflowVersionId(ctx context.Context, db DBTX, arg ListConcurrencyStrategiesByWorkflowVersionIdParams) ([]*ListConcurrencyStrategiesByWorkflowVersionIdRow, error) {
	rows, err := db.Query(ctx, listConcurrencyStrategiesByWorkflowVersionId, arg.Tenantid, arg.Workflowversionid, arg.Workflowid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ListConcurrencyStrategiesByWorkflowVersionIdRow
	for rows.Next() {
		var i ListConcurrencyStrategiesByWorkflowVersionIdRow
		if err := rows.Scan(
			&i.ID,
			&i.ParentStrategyID,
			&i.WorkflowID,
			&i.WorkflowVersionID,
			&i.StepID,
			&i.IsActive,
			&i.Strategy,
			&i.Expression,
			&i.TenantID,
			&i.MaxConcurrency,
			&i.StepReadableID,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listTenantsWithManyStepConcurrencies = `-- name: ListTenantsWithManyStepConcurrencies :many
SELECT tenant_id, COUNT(*) AS total
FROM v1_step_concurrency
WHERE is_active = TRUE
GROUP BY tenant_id
HAVING COUNT(*) > $1::BIGINT
`

type ListTenantsWithManyStepConcurrenciesRow struct {
	TenantID uuid.UUID `json:"tenant_id"`
	Total    int64     `json:"total"`
}

func (q *Queries) ListTenantsWithManyStepConcurrencies(ctx context.Context, db DBTX, threshold int64) ([]*ListTenantsWithManyStepConcurrenciesRow, error) {
	rows, err := db.Query(ctx, listTenantsWithManyStepConcurrencies, threshold)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ListTenantsWithManyStepConcurrenciesRow
	for rows.Next() {
		var i ListTenantsWithManyStepConcurrenciesRow
		if err := rows.Scan(&i.TenantID, &i.Total); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const runCancelInProgress = `-- name: RunCancelInProgress :many
WITH slots AS (
    SELECT
        task_id,
        task_inserted_at,
        task_retry_count,
        tenant_id,
        strategy_id,
        key,
        is_filled,
        -- Order slots by rn desc, seqnum desc to ensure that the most recent tasks will be run
        row_number() OVER (PARTITION BY key ORDER BY sort_id DESC) AS rn,
        row_number() OVER (ORDER BY sort_id DESC) AS seqnum
    FROM
        v1_concurrency_slot
    WHERE
        tenant_id = $1::uuid AND
        strategy_id = $2::bigint AND
        (
            schedule_timeout_at >= NOW() OR
            is_filled = TRUE
        )
), schedule_timeout_slots AS (
    SELECT
        sort_id, task_id, task_inserted_at, task_retry_count, external_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, parent_strategy_id, priority, key, is_filled, next_parent_strategy_ids, next_strategy_ids, next_keys, queue_to_notify, schedule_timeout_at
    FROM
        v1_concurrency_slot
    WHERE
        tenant_id = $1::uuid AND
        strategy_id = $2::bigint AND
        schedule_timeout_at < NOW() AND
        is_filled = FALSE
    LIMIT 1000
), eligible_running_slots AS (
    SELECT
        task_id,
        task_inserted_at,
        task_retry_count,
        tenant_id,
        strategy_id,
        key,
        is_filled,
        rn,
        seqnum
    FROM
        slots
    WHERE
        rn <= $3::int
), slots_to_cancel AS (
    SELECT
        sort_id, task_id, task_inserted_at, task_retry_count, external_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, parent_strategy_id, priority, key, is_filled, next_parent_strategy_ids, next_strategy_ids, next_keys, queue_to_notify, schedule_timeout_at
    FROM
        v1_concurrency_slot
    WHERE
        tenant_id = $1::uuid AND
        strategy_id = $2::bigint AND
        (task_inserted_at, task_id, task_retry_count) NOT IN (
            SELECT
                ers.task_inserted_at,
                ers.task_id,
                ers.task_retry_count
            FROM
                eligible_running_slots ers
        )
    ORDER BY
        task_id, task_inserted_at
    FOR UPDATE
), slots_to_run AS (
    SELECT
        sort_id, task_id, task_inserted_at, task_retry_count, external_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, parent_strategy_id, priority, key, is_filled, next_parent_strategy_ids, next_strategy_ids, next_keys, queue_to_notify, schedule_timeout_at
    FROM
        v1_concurrency_slot
    WHERE
        (task_inserted_at, task_id, task_retry_count, tenant_id, strategy_id) IN (
            SELECT
                ers.task_inserted_at,
                ers.task_id,
                ers.task_retry_count,
                ers.tenant_id,
                ers.strategy_id
            FROM
                eligible_running_slots ers
            ORDER BY
                rn, seqnum
        )
    ORDER BY
        task_id, task_inserted_at
    FOR UPDATE
), updated_slots AS (
    UPDATE
        v1_concurrency_slot
    SET
        is_filled = TRUE
    FROM
        slots_to_run
    WHERE
        v1_concurrency_slot.task_id = slots_to_run.task_id AND
        v1_concurrency_slot.task_inserted_at = slots_to_run.task_inserted_at AND
        v1_concurrency_slot.task_retry_count = slots_to_run.task_retry_count AND
        v1_concurrency_slot.tenant_id = slots_to_run.tenant_id AND
        v1_concurrency_slot.strategy_id = slots_to_run.strategy_id AND
        v1_concurrency_slot.key = slots_to_run.key AND
        v1_concurrency_slot.is_filled = FALSE
    RETURNING
        v1_concurrency_slot.sort_id, v1_concurrency_slot.task_id, v1_concurrency_slot.task_inserted_at, v1_concurrency_slot.task_retry_count, v1_concurrency_slot.external_id, v1_concurrency_slot.tenant_id, v1_concurrency_slot.workflow_id, v1_concurrency_slot.workflow_version_id, v1_concurrency_slot.workflow_run_id, v1_concurrency_slot.strategy_id, v1_concurrency_slot.parent_strategy_id, v1_concurrency_slot.priority, v1_concurrency_slot.key, v1_concurrency_slot.is_filled, v1_concurrency_slot.next_parent_strategy_ids, v1_concurrency_slot.next_strategy_ids, v1_concurrency_slot.next_keys, v1_concurrency_slot.queue_to_notify, v1_concurrency_slot.schedule_timeout_at
), deleted_slots AS (
    DELETE FROM
        v1_concurrency_slot
    WHERE
        (task_inserted_at, task_id, task_retry_count) IN (
            SELECT
                c.task_inserted_at,
                c.task_id,
                c.task_retry_count
            FROM
                slots_to_cancel c
        )
)
SELECT
    task_id,
    task_inserted_at,
    task_retry_count,
    tenant_id,
    next_strategy_ids,
    external_id,
    workflow_run_id,
    queue_to_notify,
    'SCHEDULING_TIMED_OUT' AS "operation"
FROM
    schedule_timeout_slots
UNION ALL
SELECT
    task_id,
    task_inserted_at,
    task_retry_count,
    tenant_id,
    next_strategy_ids,
    external_id,
    workflow_run_id,
    queue_to_notify,
    'CANCELLED' AS "operation"
FROM
    slots_to_cancel
WHERE
    -- not in the schedule_timeout_slots
    (task_inserted_at, task_id, task_retry_count) NOT IN (
        SELECT
            c.task_inserted_at,
            c.task_id,
            c.task_retry_count
        FROM
            schedule_timeout_slots c
    )
UNION ALL
SELECT
    task_id,
    task_inserted_at,
    task_retry_count,
    tenant_id,
    next_strategy_ids,
    external_id,
    workflow_run_id,
    queue_to_notify,
    'RUNNING' AS "operation"
FROM
    updated_slots
`

type RunCancelInProgressParams struct {
	Tenantid   uuid.UUID `json:"tenantid"`
	Strategyid int64     `json:"strategyid"`
	Maxruns    int32     `json:"maxruns"`
}

type RunCancelInProgressRow struct {
	TaskID          int64              `json:"task_id"`
	TaskInsertedAt  pgtype.Timestamptz `json:"task_inserted_at"`
	TaskRetryCount  int32              `json:"task_retry_count"`
	TenantID        uuid.UUID          `json:"tenant_id"`
	NextStrategyIds []int64            `json:"next_strategy_ids"`
	ExternalID      uuid.UUID          `json:"external_id"`
	WorkflowRunID   uuid.UUID          `json:"workflow_run_id"`
	QueueToNotify   string             `json:"queue_to_notify"`
	Operation       string             `json:"operation"`
}

func (q *Queries) RunCancelInProgress(ctx context.Context, db DBTX, arg RunCancelInProgressParams) ([]*RunCancelInProgressRow, error) {
	rows, err := db.Query(ctx, runCancelInProgress, arg.Tenantid, arg.Strategyid, arg.Maxruns)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RunCancelInProgressRow
	for rows.Next() {
		var i RunCancelInProgressRow
		if err := rows.Scan(
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.TaskRetryCount,
			&i.TenantID,
			&i.NextStrategyIds,
			&i.ExternalID,
			&i.WorkflowRunID,
			&i.QueueToNotify,
			&i.Operation,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const runCancelNewest = `-- name: RunCancelNewest :many
WITH slots AS (
    SELECT
        task_id,
        task_inserted_at,
        task_retry_count,
        cs.tenant_id,
        cs.strategy_id,
        cs.key,
        cs.is_filled,
        -- Order slots by rn desc, seqnum desc to ensure that the most recent tasks will be run
        row_number() OVER (PARTITION BY cs.key ORDER BY cs.sort_id ASC) AS rn,
        row_number() OVER (ORDER BY cs.sort_id ASC) AS seqnum
    FROM
        v1_concurrency_slot cs
    WHERE
        cs.tenant_id = $1::uuid AND
        cs.strategy_id = $2::bigint AND
        (
            schedule_timeout_at >= NOW() OR
            cs.is_filled = TRUE
        )
), schedule_timeout_slots AS (
    SELECT
        sort_id, task_id, task_inserted_at, task_retry_count, external_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, parent_strategy_id, priority, key, is_filled, next_parent_strategy_ids, next_strategy_ids, next_keys, queue_to_notify, schedule_timeout_at
    FROM
        v1_concurrency_slot
    WHERE
        tenant_id = $1::uuid AND
        strategy_id = $2::bigint AND
        schedule_timeout_at < NOW() AND
        is_filled = FALSE
    LIMIT 1000
), eligible_running_slots AS (
    SELECT
        task_id,
        task_inserted_at,
        task_retry_count,
        tenant_id,
        strategy_id,
        key,
        is_filled,
        rn,
        seqnum
    FROM
        slots
    WHERE
        rn <= $3::int
), slots_to_cancel AS (
    SELECT
        sort_id, task_id, task_inserted_at, task_retry_count, external_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, parent_strategy_id, priority, key, is_filled, next_parent_strategy_ids, next_strategy_ids, next_keys, queue_to_notify, schedule_timeout_at
    FROM
        v1_concurrency_slot
    WHERE
        tenant_id = $1::uuid AND
        strategy_id = $2::bigint AND
        (task_inserted_at, task_id, task_retry_count) NOT IN (
            SELECT
                ers.task_inserted_at,
                ers.task_id,
                ers.task_retry_count
            FROM
                eligible_running_slots ers
        )
    ORDER BY
        task_id ASC, task_inserted_at ASC
    FOR UPDATE
), slots_to_run AS (
    SELECT
        sort_id, task_id, task_inserted_at, task_retry_count, external_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, parent_strategy_id, priority, key, is_filled, next_parent_strategy_ids, next_strategy_ids, next_keys, queue_to_notify, schedule_timeout_at
    FROM
        v1_concurrency_slot
    WHERE
        (task_inserted_at, task_id, task_retry_count, tenant_id, strategy_id) IN (
            SELECT
                ers.task_inserted_at,
                ers.task_id,
                ers.task_retry_count,
                ers.tenant_id,
                ers.strategy_id
            FROM
                eligible_running_slots ers
            ORDER BY
                rn, seqnum
        )
    ORDER BY
        task_id ASC, task_inserted_at ASC
    FOR UPDATE
), updated_slots AS (
    UPDATE
        v1_concurrency_slot
    SET
        is_filled = TRUE
    FROM
        slots_to_run
    WHERE
        v1_concurrency_slot.task_id = slots_to_run.task_id AND
        v1_concurrency_slot.task_inserted_at = slots_to_run.task_inserted_at AND
        v1_concurrency_slot.task_retry_count = slots_to_run.task_retry_count AND
        v1_concurrency_slot.tenant_id = slots_to_run.tenant_id AND
        v1_concurrency_slot.strategy_id = slots_to_run.strategy_id AND
        v1_concurrency_slot.key = slots_to_run.key AND
        v1_concurrency_slot.is_filled = FALSE
    RETURNING
        v1_concurrency_slot.sort_id, v1_concurrency_slot.task_id, v1_concurrency_slot.task_inserted_at, v1_concurrency_slot.task_retry_count, v1_concurrency_slot.external_id, v1_concurrency_slot.tenant_id, v1_concurrency_slot.workflow_id, v1_concurrency_slot.workflow_version_id, v1_concurrency_slot.workflow_run_id, v1_concurrency_slot.strategy_id, v1_concurrency_slot.parent_strategy_id, v1_concurrency_slot.priority, v1_concurrency_slot.key, v1_concurrency_slot.is_filled, v1_concurrency_slot.next_parent_strategy_ids, v1_concurrency_slot.next_strategy_ids, v1_concurrency_slot.next_keys, v1_concurrency_slot.queue_to_notify, v1_concurrency_slot.schedule_timeout_at
), deleted_slots AS (
    DELETE FROM
        v1_concurrency_slot
    WHERE
        (task_inserted_at, task_id, task_retry_count) IN (
            SELECT
                c.task_inserted_at,
                c.task_id,
                c.task_retry_count
            FROM
                slots_to_cancel c
        )
)
SELECT
    task_id,
    task_inserted_at,
    task_retry_count,
    tenant_id,
    next_strategy_ids,
    external_id,
    workflow_run_id,
    queue_to_notify,
    'SCHEDULING_TIMED_OUT' AS "operation"
FROM
    schedule_timeout_slots
UNION ALL
SELECT
    task_id,
    task_inserted_at,
    task_retry_count,
    tenant_id,
    next_strategy_ids,
    external_id,
    workflow_run_id,
    queue_to_notify,
    'CANCELLED' AS "operation"
FROM
    slots_to_cancel
WHERE
    -- not in the schedule_timeout_slots
    (task_inserted_at, task_id, task_retry_count) NOT IN (
        SELECT
            c.task_inserted_at,
            c.task_id,
            c.task_retry_count
        FROM
            schedule_timeout_slots c
    )
UNION ALL
SELECT
    task_id,
    task_inserted_at,
    task_retry_count,
    tenant_id,
    next_strategy_ids,
    external_id,
    workflow_run_id,
    queue_to_notify,
    'RUNNING' AS "operation"
FROM
    updated_slots
`

type RunCancelNewestParams struct {
	Tenantid   uuid.UUID `json:"tenantid"`
	Strategyid int64     `json:"strategyid"`
	Maxruns    int32     `json:"maxruns"`
}

type RunCancelNewestRow struct {
	TaskID          int64              `json:"task_id"`
	TaskInsertedAt  pgtype.Timestamptz `json:"task_inserted_at"`
	TaskRetryCount  int32              `json:"task_retry_count"`
	TenantID        uuid.UUID          `json:"tenant_id"`
	NextStrategyIds []int64            `json:"next_strategy_ids"`
	ExternalID      uuid.UUID          `json:"external_id"`
	WorkflowRunID   uuid.UUID          `json:"workflow_run_id"`
	QueueToNotify   string             `json:"queue_to_notify"`
	Operation       string             `json:"operation"`
}

func (q *Queries) RunCancelNewest(ctx context.Context, db DBTX, arg RunCancelNewestParams) ([]*RunCancelNewestRow, error) {
	rows, err := db.Query(ctx, runCancelNewest, arg.Tenantid, arg.Strategyid, arg.Maxruns)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RunCancelNewestRow
	for rows.Next() {
		var i RunCancelNewestRow
		if err := rows.Scan(
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.TaskRetryCount,
			&i.TenantID,
			&i.NextStrategyIds,
			&i.ExternalID,
			&i.WorkflowRunID,
			&i.QueueToNotify,
			&i.Operation,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const runGroupRoundRobin = `-- name: RunGroupRoundRobin :many
WITH eligible_slots_per_group AS (
    SELECT cs.sort_id, cs.task_id, cs.task_inserted_at, cs.task_retry_count, cs.external_id, cs.tenant_id, cs.workflow_id, cs.workflow_version_id, cs.workflow_run_id, cs.strategy_id, cs.parent_strategy_id, cs.priority, cs.key, cs.is_filled, cs.next_parent_strategy_ids, cs.next_strategy_ids, cs.next_keys, cs.queue_to_notify, cs.schedule_timeout_at
    FROM (
        SELECT DISTINCT key
        FROM v1_concurrency_slot
        WHERE
            tenant_id = $1::uuid
            AND strategy_id = $2::bigint
    ) distinct_keys
    JOIN LATERAL (
        SELECT sort_id, task_id, task_inserted_at, task_retry_count, external_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, parent_strategy_id, priority, key, is_filled, next_parent_strategy_ids, next_strategy_ids, next_keys, queue_to_notify, schedule_timeout_at
        FROM v1_concurrency_slot wcs_all
        WHERE
            wcs_all.key = distinct_keys.key
            AND wcs_all.tenant_id = $1::uuid
            AND wcs_all.strategy_id = $2::bigint
        ORDER BY wcs_all.sort_id ASC
        LIMIT $3::int
    ) cs ON true
), schedule_timeout_slots AS (
    SELECT
        sort_id, task_id, task_inserted_at, task_retry_count, external_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, parent_strategy_id, priority, key, is_filled, next_parent_strategy_ids, next_strategy_ids, next_keys, queue_to_notify, schedule_timeout_at
    FROM
        v1_concurrency_slot
    WHERE
        tenant_id = $1::uuid AND
        strategy_id = $2::bigint AND
        schedule_timeout_at < NOW() AND
        is_filled = FALSE
    ORDER BY
        task_id, task_inserted_at
    FOR UPDATE
    LIMIT 1000
), eligible_slots AS (
    SELECT
        cs.sort_id, cs.task_id, cs.task_inserted_at, cs.task_retry_count, cs.external_id, cs.tenant_id, cs.workflow_id, cs.workflow_version_id, cs.workflow_run_id, cs.strategy_id, cs.parent_strategy_id, cs.priority, cs.key, cs.is_filled, cs.next_parent_strategy_ids, cs.next_strategy_ids, cs.next_keys, cs.queue_to_notify, cs.schedule_timeout_at
    FROM
        v1_concurrency_slot cs
    WHERE
        (task_inserted_at, task_id, task_retry_count, tenant_id, strategy_id) IN (
            SELECT
                es.task_inserted_at,
                es.task_id,
                es.task_retry_count,
                es.tenant_id,
                es.strategy_id
            FROM
                eligible_slots_per_group es
        )
        AND is_filled = FALSE
    ORDER BY
        task_id, task_inserted_at
    FOR UPDATE
), updated_slots AS (
    UPDATE
        v1_concurrency_slot
    SET
        is_filled = TRUE
    FROM
        eligible_slots
    WHERE
        v1_concurrency_slot.task_id = eligible_slots.task_id AND
        v1_concurrency_slot.task_inserted_at = eligible_slots.task_inserted_at AND
        v1_concurrency_slot.task_retry_count = eligible_slots.task_retry_count AND
        v1_concurrency_slot.tenant_id = eligible_slots.tenant_id AND
        v1_concurrency_slot.strategy_id = eligible_slots.strategy_id AND
        v1_concurrency_slot.key = eligible_slots.key
    RETURNING
        v1_concurrency_slot.sort_id, v1_concurrency_slot.task_id, v1_concurrency_slot.task_inserted_at, v1_concurrency_slot.task_retry_count, v1_concurrency_slot.external_id, v1_concurrency_slot.tenant_id, v1_concurrency_slot.workflow_id, v1_concurrency_slot.workflow_version_id, v1_concurrency_slot.workflow_run_id, v1_concurrency_slot.strategy_id, v1_concurrency_slot.parent_strategy_id, v1_concurrency_slot.priority, v1_concurrency_slot.key, v1_concurrency_slot.is_filled, v1_concurrency_slot.next_parent_strategy_ids, v1_concurrency_slot.next_strategy_ids, v1_concurrency_slot.next_keys, v1_concurrency_slot.queue_to_notify, v1_concurrency_slot.schedule_timeout_at
), deleted_slots AS (
    DELETE FROM
        v1_concurrency_slot
    WHERE
        (task_inserted_at, task_id, task_retry_count) IN (
            SELECT
                c.task_inserted_at,
                c.task_id,
                c.task_retry_count
            FROM
                schedule_timeout_slots c
        )
)
SELECT
    task_id,
    task_inserted_at,
    task_retry_count,
    tenant_id,
    next_strategy_ids,
    external_id,
    workflow_run_id,
    queue_to_notify,
    'SCHEDULING_TIMED_OUT' AS "operation"
FROM
    schedule_timeout_slots
UNION ALL
SELECT
    task_id,
    task_inserted_at,
    task_retry_count,
    tenant_id,
    next_strategy_ids,
    external_id,
    workflow_run_id,
    queue_to_notify,
    'RUNNING' AS "operation"
FROM
    updated_slots
`

type RunGroupRoundRobinParams struct {
	Tenantid   uuid.UUID `json:"tenantid"`
	Strategyid int64     `json:"strategyid"`
	Maxruns    int32     `json:"maxruns"`
}

type RunGroupRoundRobinRow struct {
	TaskID          int64              `json:"task_id"`
	TaskInsertedAt  pgtype.Timestamptz `json:"task_inserted_at"`
	TaskRetryCount  int32              `json:"task_retry_count"`
	TenantID        uuid.UUID          `json:"tenant_id"`
	NextStrategyIds []int64            `json:"next_strategy_ids"`
	ExternalID      uuid.UUID          `json:"external_id"`
	WorkflowRunID   uuid.UUID          `json:"workflow_run_id"`
	QueueToNotify   string             `json:"queue_to_notify"`
	Operation       string             `json:"operation"`
}

// Used for round-robin scheduling when a strategy doesn't have a parent strategy
func (q *Queries) RunGroupRoundRobin(ctx context.Context, db DBTX, arg RunGroupRoundRobinParams) ([]*RunGroupRoundRobinRow, error) {
	rows, err := db.Query(ctx, runGroupRoundRobin, arg.Tenantid, arg.Strategyid, arg.Maxruns)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RunGroupRoundRobinRow
	for rows.Next() {
		var i RunGroupRoundRobinRow
		if err := rows.Scan(
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.TaskRetryCount,
			&i.TenantID,
			&i.NextStrategyIds,
			&i.ExternalID,
			&i.WorkflowRunID,
			&i.QueueToNotify,
			&i.Operation,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const runParentCancelInProgress = `-- name: RunParentCancelInProgress :exec
WITH locked_workflow_concurrency_slots AS (
    SELECT sort_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, completed_child_strategy_ids, child_strategy_ids, priority, key, is_filled
    FROM v1_workflow_concurrency_slot
    WHERE (strategy_id, workflow_version_id, workflow_run_id) IN (
        SELECT
            strategy_id,
            workflow_version_id,
            workflow_run_id
        FROM
            tmp_workflow_concurrency_slot
    )
    ORDER BY strategy_id, workflow_version_id, workflow_run_id
    FOR UPDATE
), eligible_running_slots AS (
    SELECT sort_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, completed_child_strategy_ids, child_strategy_ids, priority, key, is_filled, rn
    FROM (
        SELECT sort_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, completed_child_strategy_ids, child_strategy_ids, priority, key, is_filled,
            ROW_NUMBER() OVER (PARTITION BY key ORDER BY sort_id DESC) as rn
        FROM locked_workflow_concurrency_slots
        WHERE
            tenant_id = $1::uuid
            AND strategy_id = $2::bigint
    ) ranked
    WHERE rn <= $3::int
), slots_to_run AS (
    SELECT
        sort_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, completed_child_strategy_ids, child_strategy_ids, priority, key, is_filled
    FROM
        v1_workflow_concurrency_slot
    WHERE
        (strategy_id, workflow_version_id, workflow_run_id) IN (
            SELECT
                ers.strategy_id,
                ers.workflow_version_id,
                ers.workflow_run_id
            FROM
                eligible_running_slots ers
        )
    ORDER BY
        strategy_id, workflow_version_id, workflow_run_id
    FOR UPDATE
), update_tmp_table AS (
    UPDATE
        tmp_workflow_concurrency_slot wsc
    SET
        is_filled = TRUE
    FROM
        slots_to_run
    WHERE
        wsc.strategy_id = slots_to_run.strategy_id AND
        wsc.workflow_version_id = slots_to_run.workflow_version_id AND
        wsc.workflow_run_id = slots_to_run.workflow_run_id
)
UPDATE
    v1_workflow_concurrency_slot wsc
SET
    is_filled = TRUE
FROM
    slots_to_run sr
WHERE
    wsc.strategy_id = sr.strategy_id AND
    wsc.workflow_version_id = sr.workflow_version_id AND
    wsc.workflow_run_id = sr.workflow_run_id
`

type RunParentCancelInProgressParams struct {
	Tenantid   uuid.UUID `json:"tenantid"`
	Strategyid int64     `json:"strategyid"`
	Maxruns    int32     `json:"maxruns"`
}

func (q *Queries) RunParentCancelInProgress(ctx context.Context, db DBTX, arg RunParentCancelInProgressParams) error {
	_, err := db.Exec(ctx, runParentCancelInProgress, arg.Tenantid, arg.Strategyid, arg.Maxruns)
	return err
}

const runParentCancelNewest = `-- name: RunParentCancelNewest :exec
WITH locked_workflow_concurrency_slots AS (
    SELECT sort_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, completed_child_strategy_ids, child_strategy_ids, priority, key, is_filled
    FROM v1_workflow_concurrency_slot
    WHERE (strategy_id, workflow_version_id, workflow_run_id) IN (
        SELECT
            strategy_id,
            workflow_version_id,
            workflow_run_id
        FROM
            tmp_workflow_concurrency_slot
    )
    ORDER BY strategy_id, workflow_version_id, workflow_run_id
    FOR UPDATE
), eligible_running_slots AS (
    SELECT sort_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, completed_child_strategy_ids, child_strategy_ids, priority, key, is_filled, rn
    FROM (
        SELECT sort_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, completed_child_strategy_ids, child_strategy_ids, priority, key, is_filled,
            ROW_NUMBER() OVER (PARTITION BY key ORDER BY sort_id ASC) as rn
        FROM locked_workflow_concurrency_slots
        WHERE
            tenant_id = $1::uuid
            AND strategy_id = $2::bigint
    ) ranked
    WHERE rn <= $3::int
), slots_to_run AS (
    SELECT
        sort_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, completed_child_strategy_ids, child_strategy_ids, priority, key, is_filled
    FROM
        v1_workflow_concurrency_slot
    WHERE
        (strategy_id, workflow_version_id, workflow_run_id) IN (
            SELECT
                ers.strategy_id,
                ers.workflow_version_id,
                ers.workflow_run_id
            FROM
                eligible_running_slots ers
        )
    ORDER BY
        strategy_id, workflow_version_id, workflow_run_id
    FOR UPDATE
), update_tmp_table AS (
    UPDATE
        tmp_workflow_concurrency_slot wsc
    SET
        is_filled = TRUE
    FROM
        slots_to_run
    WHERE
        wsc.strategy_id = slots_to_run.strategy_id AND
        wsc.workflow_version_id = slots_to_run.workflow_version_id AND
        wsc.workflow_run_id = slots_to_run.workflow_run_id
)
UPDATE
    v1_workflow_concurrency_slot wsc
SET
    is_filled = TRUE
FROM
    slots_to_run sr
WHERE
    wsc.strategy_id = sr.strategy_id AND
    wsc.workflow_version_id = sr.workflow_version_id AND
    wsc.workflow_run_id = sr.workflow_run_id
`

type RunParentCancelNewestParams struct {
	Tenantid   uuid.UUID `json:"tenantid"`
	Strategyid int64     `json:"strategyid"`
	Maxruns    int32     `json:"maxruns"`
}

func (q *Queries) RunParentCancelNewest(ctx context.Context, db DBTX, arg RunParentCancelNewestParams) error {
	_, err := db.Exec(ctx, runParentCancelNewest, arg.Tenantid, arg.Strategyid, arg.Maxruns)
	return err
}

const runParentGroupRoundRobin = `-- name: RunParentGroupRoundRobin :exec
WITH eligible_slots_per_group AS (
    SELECT wsc.sort_id, wsc.tenant_id, wsc.workflow_id, wsc.workflow_version_id, wsc.workflow_run_id, wsc.strategy_id, wsc.completed_child_strategy_ids, wsc.child_strategy_ids, wsc.priority, wsc.key, wsc.is_filled
    FROM (
        SELECT DISTINCT key
        FROM v1_workflow_concurrency_slot
        WHERE
            tenant_id = $1::uuid
            AND strategy_id = $2::bigint
    ) distinct_keys
    JOIN LATERAL (
        SELECT sort_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, completed_child_strategy_ids, child_strategy_ids, priority, key, is_filled
        FROM v1_workflow_concurrency_slot wcs_all
        WHERE
            wcs_all.key = distinct_keys.key
            AND wcs_all.tenant_id = $1::uuid
            AND wcs_all.strategy_id = $2::bigint
        ORDER BY wcs_all.priority DESC, wcs_all.sort_id ASC
        LIMIT $3::int
    ) wsc ON true
), eligible_slots AS (
    SELECT
        sort_id, tenant_id, workflow_id, workflow_version_id, workflow_run_id, strategy_id, completed_child_strategy_ids, child_strategy_ids, priority, key, is_filled
    FROM
        v1_workflow_concurrency_slot
    WHERE
        (strategy_id, workflow_version_id, workflow_run_id) IN (
            SELECT
                es.strategy_id,
                es.workflow_version_id,
                es.workflow_run_id
            FROM
                eligible_slots_per_group es
        )
        AND is_filled = FALSE
    ORDER BY
        strategy_id, workflow_version_id, workflow_run_id
    FOR UPDATE
)
UPDATE
    v1_workflow_concurrency_slot
SET
    is_filled = TRUE
FROM
    eligible_slots
WHERE
    v1_workflow_concurrency_slot.strategy_id = eligible_slots.strategy_id AND
    v1_workflow_concurrency_slot.workflow_version_id = eligible_slots.workflow_version_id AND
    v1_workflow_concurrency_slot.workflow_run_id = eligible_slots.workflow_run_id
`

type RunParentGroupRoundRobinParams struct {
	Tenantid   uuid.UUID `json:"tenantid"`
	Strategyid int64     `json:"strategyid"`
	Maxruns    int32     `json:"maxruns"`
}

func (q *Queries) RunParentGroupRoundRobin(ctx context.Context, db DBTX, arg RunParentGroupRoundRobinParams) error {
	_, err := db.Exec(ctx, runParentGroupRoundRobin, arg.Tenantid, arg.Strategyid, arg.Maxruns)
	return err
}

const setConcurrencyStrategyInactive = `-- name: SetConcurrencyStrategyInactive :exec
UPDATE
    v1_step_concurrency
SET
    is_active = FALSE
WHERE
    workflow_id = $1::uuid AND
    workflow_version_id = $2::uuid AND
    step_id = $3::uuid AND
    id = $4::bigint
`

type SetConcurrencyStrategyInactiveParams struct {
	Workflowid        uuid.UUID `json:"workflowid"`
	Workflowversionid uuid.UUID `json:"workflowversionid"`
	Stepid            uuid.UUID `json:"stepid"`
	Strategyid        int64     `json:"strategyid"`
}

func (q *Queries) SetConcurrencyStrategyInactive(ctx context.Context, db DBTX, arg SetConcurrencyStrategyInactiveParams) error {
	_, err := db.Exec(ctx, setConcurrencyStrategyInactive,
		arg.Workflowid,
		arg.Workflowversionid,
		arg.Stepid,
		arg.Strategyid,
	)
	return err
}

const tryAdvisoryLock = `-- name: TryAdvisoryLock :one
SELECT pg_try_advisory_xact_lock($1::bigint) AS "locked"
`

func (q *Queries) TryAdvisoryLock(ctx context.Context, db DBTX, key int64) (bool, error) {
	row := db.QueryRow(ctx, tryAdvisoryLock, key)
	var locked bool
	err := row.Scan(&locked)
	return locked, err
}

const tryAdvisoryLockMany = `-- name: TryAdvisoryLockMany :many
WITH keys AS (
    SELECT UNNEST($1::BIGINT[]) AS key
), ordered_keys AS (
    SELECT key
    FROM keys
    ORDER BY key
)

SELECT key::BIGINT, pg_try_advisory_xact_lock(key)::BOOLEAN AS "acquired"
FROM ordered_keys
`

type TryAdvisoryLockManyRow struct {
	Key      int64 `json:"key"`
	Acquired bool  `json:"acquired"`
}

func (q *Queries) TryAdvisoryLockMany(ctx context.Context, db DBTX, keys []int64) ([]*TryAdvisoryLockManyRow, error) {
	rows, err := db.Query(ctx, tryAdvisoryLockMany, keys)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*TryAdvisoryLockManyRow
	for rows.Next() {
		var i TryAdvisoryLockManyRow
		if err := rows.Scan(&i.Key, &i.Acquired); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}
