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

package sqlcv1

import (
	"context"

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

const bulkQueueItems = `-- name: BulkQueueItems :many
WITH locked_qis AS (
    SELECT
        id
    FROM
        v1_queue_item
    WHERE
        id = ANY($1::bigint[])
    ORDER BY
        id ASC
    FOR UPDATE
)
DELETE FROM
    v1_queue_item
WHERE
    id = ANY($1::bigint[])
RETURNING
    id
`

func (q *Queries) BulkQueueItems(ctx context.Context, db DBTX, ids []int64) ([]int64, error) {
	rows, err := db.Query(ctx, bulkQueueItems, ids)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []int64
	for rows.Next() {
		var id int64
		if err := rows.Scan(&id); err != nil {
			return nil, err
		}
		items = append(items, id)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const cleanupV1QueueItem = `-- name: CleanupV1QueueItem :execresult
WITH locked_qis as (
    SELECT qi.task_id, qi.task_inserted_at, qi.retry_count
    FROM v1_queue_item qi
    WHERE
        NOT EXISTS (
            SELECT 1
            FROM v1_task vt
            WHERE qi.task_id = vt.id
                AND qi.task_inserted_at = vt.inserted_at
        )
        OR EXISTS (
            SELECT 1
            FROM "Tenant" t
            WHERE t."id" = qi.tenant_id
                AND t."deletedAt" IS NOT NULL
        )
    ORDER BY qi.id ASC
    LIMIT $1::int
    FOR UPDATE SKIP LOCKED
)
DELETE FROM v1_queue_item
WHERE (task_id, task_inserted_at, retry_count) IN (
    SELECT task_id, task_inserted_at, retry_count
    FROM locked_qis
)
`

func (q *Queries) CleanupV1QueueItem(ctx context.Context, db DBTX, batchsize int32) (pgconn.CommandTag, error) {
	return db.Exec(ctx, cleanupV1QueueItem, batchsize)
}

const cleanupV1RateLimitedQueueItem = `-- name: CleanupV1RateLimitedQueueItem :execresult
WITH locked_qis as (
    SELECT qi.task_id, qi.task_inserted_at, qi.retry_count
    FROM v1_rate_limited_queue_items qi
    WHERE
        NOT EXISTS (
            SELECT 1
            FROM v1_task vt
            WHERE qi.task_id = vt.id
                AND qi.task_inserted_at = vt.inserted_at
        )
        OR EXISTS (
            SELECT 1
            FROM "Tenant" t
            WHERE t."id" = qi.tenant_id
                AND t."deletedAt" IS NOT NULL
        )
    ORDER BY qi.task_id, qi.task_inserted_at, qi.retry_count
    LIMIT $1::int
    FOR UPDATE SKIP LOCKED
)
DELETE FROM v1_rate_limited_queue_items
WHERE (task_id, task_inserted_at) IN (
    SELECT task_id, task_inserted_at
    FROM locked_qis
)
`

func (q *Queries) CleanupV1RateLimitedQueueItem(ctx context.Context, db DBTX, batchsize int32) (pgconn.CommandTag, error) {
	return db.Exec(ctx, cleanupV1RateLimitedQueueItem, batchsize)
}

const cleanupV1RetryQueueItem = `-- name: CleanupV1RetryQueueItem :execresult
WITH locked_qis as (
    SELECT qi.task_id, qi.task_inserted_at, qi.task_retry_count
    FROM v1_retry_queue_item qi
    WHERE
        NOT EXISTS (
            SELECT 1
            FROM v1_task vt
            WHERE qi.task_id = vt.id
                AND qi.task_inserted_at = vt.inserted_at
        )
        OR EXISTS (
            SELECT 1
            FROM "Tenant" t
            WHERE t."id" = qi.tenant_id
                AND t."deletedAt" IS NOT NULL
        )
    ORDER BY qi.task_id, qi.task_inserted_at, qi.task_retry_count
    LIMIT $1::int
    FOR UPDATE SKIP LOCKED
)
DELETE FROM v1_retry_queue_item
WHERE (task_id, task_inserted_at) IN (
    SELECT task_id, task_inserted_at
    FROM locked_qis
)
`

func (q *Queries) CleanupV1RetryQueueItem(ctx context.Context, db DBTX, batchsize int32) (pgconn.CommandTag, error) {
	return db.Exec(ctx, cleanupV1RetryQueueItem, batchsize)
}

const deleteTasksFromQueue = `-- name: DeleteTasksFromQueue :exec
WITH input AS (
    SELECT
        task_id, retry_count
    FROM
        (
            SELECT
                unnest($1::bigint[]) AS task_id,
                unnest($2::integer[]) AS retry_count
        ) AS subquery
), locked_qis AS (
    SELECT
        id
    FROM
        v1_queue_item
    WHERE
        (task_id, retry_count) IN (SELECT task_id, retry_count FROM input)
    ORDER BY
        id ASC
    FOR UPDATE
)
DELETE FROM
    v1_queue_item
WHERE
    id = ANY(SELECT id FROM locked_qis)
`

type DeleteTasksFromQueueParams struct {
	Taskids     []int64 `json:"taskids"`
	Retrycounts []int32 `json:"retrycounts"`
}

func (q *Queries) DeleteTasksFromQueue(ctx context.Context, db DBTX, arg DeleteTasksFromQueueParams) error {
	_, err := db.Exec(ctx, deleteTasksFromQueue, arg.Taskids, arg.Retrycounts)
	return err
}

const getDesiredLabels = `-- name: GetDesiredLabels :many
SELECT
    "key",
    "strValue",
    "intValue",
    "required",
    "weight",
    "comparator",
    "stepId"
FROM
    "StepDesiredWorkerLabel"
WHERE
    "stepId" = ANY($1::uuid[])
`

type GetDesiredLabelsRow struct {
	Key        string                `json:"key"`
	StrValue   pgtype.Text           `json:"strValue"`
	IntValue   pgtype.Int4           `json:"intValue"`
	Required   bool                  `json:"required"`
	Weight     int32                 `json:"weight"`
	Comparator WorkerLabelComparator `json:"comparator"`
	StepId     uuid.UUID             `json:"stepId"`
}

func (q *Queries) GetDesiredLabels(ctx context.Context, db DBTX, stepids []uuid.UUID) ([]*GetDesiredLabelsRow, error) {
	rows, err := db.Query(ctx, getDesiredLabels, stepids)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*GetDesiredLabelsRow
	for rows.Next() {
		var i GetDesiredLabelsRow
		if err := rows.Scan(
			&i.Key,
			&i.StrValue,
			&i.IntValue,
			&i.Required,
			&i.Weight,
			&i.Comparator,
			&i.StepId,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getMinUnprocessedQueueItemId = `-- name: GetMinUnprocessedQueueItemId :one
WITH priority_1 AS (
    SELECT
        id
    FROM
        v1_queue_item
    WHERE
        tenant_id = $1::uuid
        AND queue = $2::text
        AND priority = 1
    ORDER BY
        id ASC
    LIMIT 1
),
priority_2 AS (
    SELECT
        id
    FROM
        v1_queue_item
    WHERE
        tenant_id = $1::uuid
        AND queue = $2::text
        AND priority = 2
    ORDER BY
        id ASC
    LIMIT 1
),
priority_3 AS (
    SELECT
        id
    FROM
        v1_queue_item
    WHERE
        tenant_id = $1::uuid
        AND queue = $2::text
        AND priority = 3
    ORDER BY
        id ASC
    LIMIT 1
),
priority_4 AS (
    SELECT
        id
    FROM
        v1_queue_item
    WHERE
        tenant_id = $1::uuid
        AND queue = $2::text
        AND priority = 4
    ORDER BY
        id ASC
    LIMIT 1
)
SELECT
    COALESCE(MIN(id), 0)::bigint AS "minId"
FROM (
    SELECT id FROM priority_1
    UNION ALL
    SELECT id FROM priority_2
    UNION ALL
    SELECT id FROM priority_3
    UNION ALL
    SELECT id FROM priority_4
) AS combined_priorities
`

type GetMinUnprocessedQueueItemIdParams struct {
	Tenantid uuid.UUID `json:"tenantid"`
	Queue    string    `json:"queue"`
}

func (q *Queries) GetMinUnprocessedQueueItemId(ctx context.Context, db DBTX, arg GetMinUnprocessedQueueItemIdParams) (int64, error) {
	row := db.QueryRow(ctx, getMinUnprocessedQueueItemId, arg.Tenantid, arg.Queue)
	var minId int64
	err := row.Scan(&minId)
	return minId, err
}

const getQueuedCounts = `-- name: GetQueuedCounts :many
SELECT
    queue,
    COUNT(*) AS count
FROM
    v1_queue_item qi
WHERE
    qi.tenant_id = $1::uuid
GROUP BY
    qi.queue
`

type GetQueuedCountsRow struct {
	Queue string `json:"queue"`
	Count int64  `json:"count"`
}

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

const getStepSlotRequests = `-- name: GetStepSlotRequests :many
SELECT
    step_id,
    slot_type,
    units
FROM
    v1_step_slot_request
WHERE
    step_id = ANY($1::uuid[])
    AND tenant_id = $2::uuid
`

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

type GetStepSlotRequestsRow struct {
	StepID   uuid.UUID `json:"step_id"`
	SlotType string    `json:"slot_type"`
	Units    int32     `json:"units"`
}

func (q *Queries) GetStepSlotRequests(ctx context.Context, db DBTX, arg GetStepSlotRequestsParams) ([]*GetStepSlotRequestsRow, error) {
	rows, err := db.Query(ctx, getStepSlotRequests, arg.Stepids, arg.Tenantid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*GetStepSlotRequestsRow
	for rows.Next() {
		var i GetStepSlotRequestsRow
		if err := rows.Scan(&i.StepID, &i.SlotType, &i.Units); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listActionsForWorkers = `-- name: ListActionsForWorkers :many
SELECT
    w."id" as "workerId",
    a."actionId"
FROM
    "Worker" w
LEFT JOIN
    "_ActionToWorker" atw ON w."id" = atw."B"
LEFT JOIN
    "Action" a ON atw."A" = a."id"
WHERE
    w."tenantId" = $1::uuid
    AND w."id" = ANY($2::uuid[])
    AND w."dispatcherId" IS NOT NULL
    AND w."lastHeartbeatAt" > NOW() - INTERVAL '5 seconds'
    AND w."isActive" = true
    AND w."isPaused" = false
`

type ListActionsForWorkersParams struct {
	Tenantid  uuid.UUID   `json:"tenantid"`
	Workerids []uuid.UUID `json:"workerids"`
}

type ListActionsForWorkersRow struct {
	WorkerId uuid.UUID   `json:"workerId"`
	ActionId pgtype.Text `json:"actionId"`
}

func (q *Queries) ListActionsForWorkers(ctx context.Context, db DBTX, arg ListActionsForWorkersParams) ([]*ListActionsForWorkersRow, error) {
	rows, err := db.Query(ctx, listActionsForWorkers, arg.Tenantid, arg.Workerids)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ListActionsForWorkersRow
	for rows.Next() {
		var i ListActionsForWorkersRow
		if err := rows.Scan(&i.WorkerId, &i.ActionId); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listQueueItemsForQueue = `-- name: ListQueueItemsForQueue :many
SELECT
    id, tenant_id, queue, task_id, task_inserted_at, external_id, action_id, step_id, workflow_id, workflow_run_id, schedule_timeout_at, step_timeout, priority, sticky, desired_worker_id, retry_count, desired_worker_label
FROM
    v1_queue_item qi
WHERE
    qi.tenant_id = $1::uuid
    AND qi.queue = $2::text
    AND (
        $3::bigint IS NULL OR
        qi.id >= $3::bigint
    )
    -- Added to ensure that the index is used
    AND qi.priority >= 1 AND qi.priority <= 4
ORDER BY
    qi.priority DESC,
    qi.id ASC
LIMIT
    COALESCE($4::integer, 100)
`

type ListQueueItemsForQueueParams struct {
	Tenantid uuid.UUID   `json:"tenantid"`
	Queue    string      `json:"queue"`
	GtId     pgtype.Int8 `json:"gtId"`
	Limit    pgtype.Int4 `json:"limit"`
}

func (q *Queries) ListQueueItemsForQueue(ctx context.Context, db DBTX, arg ListQueueItemsForQueueParams) ([]*V1QueueItem, error) {
	rows, err := db.Query(ctx, listQueueItemsForQueue,
		arg.Tenantid,
		arg.Queue,
		arg.GtId,
		arg.Limit,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*V1QueueItem
	for rows.Next() {
		var i V1QueueItem
		if err := rows.Scan(
			&i.ID,
			&i.TenantID,
			&i.Queue,
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.ExternalID,
			&i.ActionID,
			&i.StepID,
			&i.WorkflowID,
			&i.WorkflowRunID,
			&i.ScheduleTimeoutAt,
			&i.StepTimeout,
			&i.Priority,
			&i.Sticky,
			&i.DesiredWorkerID,
			&i.RetryCount,
			&i.DesiredWorkerLabel,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listQueueItemsForTasks = `-- name: ListQueueItemsForTasks :many
WITH input AS (
    SELECT
        UNNEST($2::bigint[]) AS task_id,
        UNNEST($3::timestamptz[]) AS task_inserted_at,
        UNNEST($4::integer[]) AS retry_count
)
SELECT
    qi.id, qi.tenant_id, qi.queue, qi.task_id, qi.task_inserted_at, qi.external_id, qi.action_id, qi.step_id, qi.workflow_id, qi.workflow_run_id, qi.schedule_timeout_at, qi.step_timeout, qi.priority, qi.sticky, qi.desired_worker_id, qi.retry_count, qi.desired_worker_label
FROM
    v1_queue_item qi
WHERE
    (qi.task_id, qi.task_inserted_at, qi.retry_count) IN (SELECT task_id, task_inserted_at, retry_count FROM input)
    AND qi.tenant_id = $1::uuid
`

type ListQueueItemsForTasksParams struct {
	Tenantid        uuid.UUID            `json:"tenantid"`
	Taskids         []int64              `json:"taskids"`
	Taskinsertedats []pgtype.Timestamptz `json:"taskinsertedats"`
	Retrycounts     []int32              `json:"retrycounts"`
}

func (q *Queries) ListQueueItemsForTasks(ctx context.Context, db DBTX, arg ListQueueItemsForTasksParams) ([]*V1QueueItem, error) {
	rows, err := db.Query(ctx, listQueueItemsForTasks,
		arg.Tenantid,
		arg.Taskids,
		arg.Taskinsertedats,
		arg.Retrycounts,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*V1QueueItem
	for rows.Next() {
		var i V1QueueItem
		if err := rows.Scan(
			&i.ID,
			&i.TenantID,
			&i.Queue,
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.ExternalID,
			&i.ActionID,
			&i.StepID,
			&i.WorkflowID,
			&i.WorkflowRunID,
			&i.ScheduleTimeoutAt,
			&i.StepTimeout,
			&i.Priority,
			&i.Sticky,
			&i.DesiredWorkerID,
			&i.RetryCount,
			&i.DesiredWorkerLabel,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listQueues = `-- name: ListQueues :many
SELECT
    tenant_id, name, last_active
FROM
    v1_queue
WHERE
    tenant_id = $1::uuid
    AND last_active > NOW() - INTERVAL '1 day'
`

func (q *Queries) ListQueues(ctx context.Context, db DBTX, tenantid uuid.UUID) ([]*V1Queue, error) {
	rows, err := db.Query(ctx, listQueues, tenantid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*V1Queue
	for rows.Next() {
		var i V1Queue
		if err := rows.Scan(&i.TenantID, &i.Name, &i.LastActive); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const moveRateLimitedQueueItems = `-- name: MoveRateLimitedQueueItems :many
WITH input AS (
    SELECT
        UNNEST($1::bigint[]) AS id,
        UNNEST($2::timestamptz[]) AS requeue_after
), moved_items AS (
    DELETE FROM v1_queue_item
    WHERE id = ANY(SELECT id FROM input)
    RETURNING
        id,
        tenant_id,
        queue,
        task_id,
        task_inserted_at,
        external_id,
        action_id,
        step_id,
        workflow_id,
        workflow_run_id,
        schedule_timeout_at,
        step_timeout,
        priority,
        sticky,
        desired_worker_id,
        retry_count,
        desired_worker_label
)
INSERT INTO v1_rate_limited_queue_items (
    requeue_after,
    tenant_id,
    queue,
    task_id,
    task_inserted_at,
    external_id,
    action_id,
    step_id,
    workflow_id,
    workflow_run_id,
    schedule_timeout_at,
    step_timeout,
    priority,
    sticky,
    desired_worker_id,
    retry_count,
    desired_worker_label
)
SELECT
    i.requeue_after,
    tenant_id,
    queue,
    task_id,
    task_inserted_at,
    external_id,
    action_id,
    step_id,
    workflow_id,
    workflow_run_id,
    schedule_timeout_at,
    step_timeout,
    priority,
    sticky,
    desired_worker_id,
    retry_count,
    desired_worker_label
FROM moved_items
JOIN input i ON moved_items.id = i.id
ON CONFLICT (task_id, task_inserted_at, retry_count) DO NOTHING
RETURNING tenant_id, task_id, task_inserted_at, retry_count
`

type MoveRateLimitedQueueItemsParams struct {
	Ids          []int64              `json:"ids"`
	Requeueafter []pgtype.Timestamptz `json:"requeueafter"`
}

type MoveRateLimitedQueueItemsRow struct {
	TenantID       uuid.UUID          `json:"tenant_id"`
	TaskID         int64              `json:"task_id"`
	TaskInsertedAt pgtype.Timestamptz `json:"task_inserted_at"`
	RetryCount     int32              `json:"retry_count"`
}

func (q *Queries) MoveRateLimitedQueueItems(ctx context.Context, db DBTX, arg MoveRateLimitedQueueItemsParams) ([]*MoveRateLimitedQueueItemsRow, error) {
	rows, err := db.Query(ctx, moveRateLimitedQueueItems, arg.Ids, arg.Requeueafter)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*MoveRateLimitedQueueItemsRow
	for rows.Next() {
		var i MoveRateLimitedQueueItemsRow
		if err := rows.Scan(
			&i.TenantID,
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.RetryCount,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const reactivateInactiveQueuesWithItems = `-- name: ReactivateInactiveQueuesWithItems :execresult
WITH inactive_queues_with_items AS (
    SELECT q.tenant_id, q.name
    FROM v1_queue q
    WHERE q.last_active <= NOW() - INTERVAL '1 day'
      AND EXISTS (
        SELECT 1
        FROM v1_queue_item qi
        WHERE qi.tenant_id = q.tenant_id
          AND qi.queue = q.name
        LIMIT 1
      )
    FOR UPDATE SKIP LOCKED
)
UPDATE v1_queue q
SET last_active = NOW()
FROM inactive_queues_with_items i
WHERE q.tenant_id = i.tenant_id
  AND q.name = i.name
`

// Reactivates queues that have been marked inactive (last_active > 1 day ago)
// but still have pending items in v1_queue_item. This is a fallback mechanism
// to ensure queues don't get stuck inactive while they have work to do.
func (q *Queries) ReactivateInactiveQueuesWithItems(ctx context.Context, db DBTX) (pgconn.CommandTag, error) {
	return db.Exec(ctx, reactivateInactiveQueuesWithItems)
}

const requeueRateLimitedQueueItems = `-- name: RequeueRateLimitedQueueItems :many
WITH ready_items AS (
    SELECT
        tenant_id,
        queue,
        task_id,
        task_inserted_at,
        external_id,
        action_id,
        step_id,
        workflow_id,
        workflow_run_id,
        schedule_timeout_at,
        step_timeout,
        priority,
        sticky,
        desired_worker_id,
        retry_count,
        desired_worker_label
    FROM
        v1_rate_limited_queue_items
    WHERE
        tenant_id = $1::uuid
        AND queue = $2::text
        AND requeue_after <= NOW()
    ORDER BY
        task_id, task_inserted_at, retry_count
    FOR UPDATE SKIP LOCKED -- locked are about to be deleted
), deleted_items AS (
    DELETE FROM v1_rate_limited_queue_items
    WHERE
        (task_id, task_inserted_at, retry_count) IN (SELECT task_id, task_inserted_at, retry_count FROM ready_items)
    RETURNING
        tenant_id,
        queue,
        task_id,
        task_inserted_at,
        external_id,
        action_id,
        step_id,
        workflow_id,
        workflow_run_id,
        schedule_timeout_at,
        step_timeout,
        priority,
        sticky,
        desired_worker_id,
        retry_count
)
INSERT INTO v1_queue_item (
    tenant_id,
    queue,
    task_id,
    task_inserted_at,
    external_id,
    action_id,
    step_id,
    workflow_id,
    workflow_run_id,
    schedule_timeout_at,
    step_timeout,
    priority,
    sticky,
    desired_worker_id,
    retry_count,
    desired_worker_label
)
SELECT
    tenant_id,
    queue,
    task_id,
    task_inserted_at,
    external_id,
    action_id,
    step_id,
    workflow_id,
    workflow_run_id,
    schedule_timeout_at,
    step_timeout,
    priority,
    sticky,
    desired_worker_id,
    retry_count,
    desired_worker_label
FROM ready_items
ON CONFLICT (task_id, task_inserted_at, retry_count) DO NOTHING
RETURNING id, tenant_id, task_id, task_inserted_at, retry_count
`

type RequeueRateLimitedQueueItemsParams struct {
	Tenantid uuid.UUID `json:"tenantid"`
	Queue    string    `json:"queue"`
}

type RequeueRateLimitedQueueItemsRow struct {
	ID             int64              `json:"id"`
	TenantID       uuid.UUID          `json:"tenant_id"`
	TaskID         int64              `json:"task_id"`
	TaskInsertedAt pgtype.Timestamptz `json:"task_inserted_at"`
	RetryCount     int32              `json:"retry_count"`
}

func (q *Queries) RequeueRateLimitedQueueItems(ctx context.Context, db DBTX, arg RequeueRateLimitedQueueItemsParams) ([]*RequeueRateLimitedQueueItemsRow, error) {
	rows, err := db.Query(ctx, requeueRateLimitedQueueItems, arg.Tenantid, arg.Queue)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RequeueRateLimitedQueueItemsRow
	for rows.Next() {
		var i RequeueRateLimitedQueueItemsRow
		if err := rows.Scan(
			&i.ID,
			&i.TenantID,
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.RetryCount,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const updateTasksToAssigned = `-- name: UpdateTasksToAssigned :many
WITH input AS (
    SELECT
        id,
        inserted_at,
        worker_id
    FROM
        (
            SELECT
                UNNEST($1::bigint[]) AS id,
                UNNEST($2::timestamptz[]) AS inserted_at,
                UNNEST($3::uuid[]) AS worker_id
        ) AS subquery
    ORDER BY id
), updated_tasks AS (
    SELECT
        t.id,
        t.inserted_at,
        t.retry_count,
        i.worker_id,
        t.tenant_id,
        t.step_id,
        CURRENT_TIMESTAMP + convert_duration_to_interval(t.step_timeout) AS timeout_at,
        t.is_durable
    FROM
        v1_task t
    JOIN
        input i ON (t.id, t.inserted_at) = (i.id, i.inserted_at)
    WHERE
        t.inserted_at >= $4::timestamptz
    ORDER BY t.id
), assigned_tasks AS (
    INSERT INTO v1_task_runtime (
        task_id,
        task_inserted_at,
        retry_count,
        worker_id,
        tenant_id,
        timeout_at
    )
    SELECT
        t.id,
        t.inserted_at,
        t.retry_count,
        t.worker_id,
        $5::uuid,
        t.timeout_at
    FROM
        updated_tasks t
    ON CONFLICT (task_id, task_inserted_at, retry_count) DO UPDATE
    SET
        evicted_at = NULL,
        worker_id = EXCLUDED.worker_id,
        timeout_at = EXCLUDED.timeout_at
    WHERE v1_task_runtime.evicted_at IS NOT NULL
    -- only return the task ids that were successfully assigned
    RETURNING task_id, task_inserted_at, retry_count, worker_id
), slot_requests AS (
    SELECT
        t.id,
        t.inserted_at,
        t.retry_count,
        t.worker_id,
        t.tenant_id,
        COALESCE(req.slot_type, 'default'::text) AS slot_type,
        COALESCE(req.units, 1) AS units
    FROM
        updated_tasks t
    LEFT JOIN
        v1_step_slot_request req
        ON req.step_id = t.step_id AND req.tenant_id = t.tenant_id
), assigned_slots AS (
    INSERT INTO v1_task_runtime_slot (
        tenant_id,
        task_id,
        task_inserted_at,
        retry_count,
        worker_id,
        slot_type,
        units
    )
    SELECT
        tenant_id,
        id,
        inserted_at,
        retry_count,
        worker_id,
        slot_type,
        units
    FROM
        slot_requests
    ON CONFLICT (task_id, task_inserted_at, retry_count, slot_type) DO NOTHING
    RETURNING task_id
)
SELECT
    asr.task_id,
    asr.task_inserted_at,
    asr.worker_id,
    ut.is_durable
FROM
    assigned_tasks asr
JOIN
    updated_tasks ut ON (asr.task_id, asr.task_inserted_at, asr.retry_count) = (ut.id, ut.inserted_at, ut.retry_count)
`

type UpdateTasksToAssignedParams struct {
	Taskids           []int64              `json:"taskids"`
	Taskinsertedats   []pgtype.Timestamptz `json:"taskinsertedats"`
	Workerids         []uuid.UUID          `json:"workerids"`
	Mintaskinsertedat pgtype.Timestamptz   `json:"mintaskinsertedat"`
	Tenantid          uuid.UUID            `json:"tenantid"`
}

type UpdateTasksToAssignedRow struct {
	TaskID         int64              `json:"task_id"`
	TaskInsertedAt pgtype.Timestamptz `json:"task_inserted_at"`
	WorkerID       *uuid.UUID         `json:"worker_id"`
	IsDurable      pgtype.Bool        `json:"is_durable"`
}

func (q *Queries) UpdateTasksToAssigned(ctx context.Context, db DBTX, arg UpdateTasksToAssignedParams) ([]*UpdateTasksToAssignedRow, error) {
	rows, err := db.Query(ctx, updateTasksToAssigned,
		arg.Taskids,
		arg.Taskinsertedats,
		arg.Workerids,
		arg.Mintaskinsertedat,
		arg.Tenantid,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*UpdateTasksToAssignedRow
	for rows.Next() {
		var i UpdateTasksToAssignedRow
		if err := rows.Scan(
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.WorkerID,
			&i.IsDurable,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const upsertQueues = `-- name: UpsertQueues :exec
WITH ordered_names AS (
    SELECT unnest($2::text[]) AS name
    ORDER BY name ASC
), existing_queues AS (
    SELECT tenant_id, name, last_active
    FROM v1_queue
    WHERE tenant_id = $1
      AND name = ANY($2::text[])
), locked_existing_queues AS (
    SELECT tenant_id, name, last_active
    FROM v1_queue
    WHERE
        tenant_id = $1
        AND name IN (SELECT name FROM existing_queues)
    ORDER BY name ASC
    FOR UPDATE SKIP LOCKED
), names_to_insert AS (
    SELECT on1.name
    FROM ordered_names on1
    LEFT JOIN existing_queues eq ON eq.name = on1.name
    WHERE eq.name IS NULL
), updated_queues AS (
    UPDATE v1_queue
    SET last_active = NOW()
    WHERE tenant_id = $1
      AND name IN (SELECT name FROM locked_existing_queues)
)
INSERT INTO v1_queue (tenant_id, name, last_active)
SELECT $1, name, NOW()
FROM names_to_insert
ON CONFLICT (tenant_id, name) DO NOTHING
`

type UpsertQueuesParams struct {
	TenantID uuid.UUID `json:"tenant_id"`
	Names    []string  `json:"names"`
}

// Insert new queues
func (q *Queries) UpsertQueues(ctx context.Context, db DBTX, arg UpsertQueuesParams) error {
	_, err := db.Exec(ctx, upsertQueues, arg.TenantID, arg.Names)
	return err
}
