olap.sql.go

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

package sqlcv1

import (
	"context"

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

const acquireOrExtendOLAPCutoverJobLease = `-- name: AcquireOrExtendOLAPCutoverJobLease :one
WITH inputs AS (
    SELECT
        $2::DATE AS key,
        $1::UUID AS lease_process_id,
        $3::TIMESTAMPTZ AS lease_expires_at,
        $4::UUID AS last_external_id
), any_lease_held_by_other_process AS (
    -- need coalesce here in case there are no rows that don't belong to this process
    SELECT COALESCE(BOOL_OR(lease_expires_at > NOW()), FALSE) AS lease_exists
    FROM v1_payloads_olap_cutover_job_offset
    WHERE lease_process_id != $1::UUID
), to_insert AS (
    SELECT key, lease_process_id, lease_expires_at, last_external_id
    FROM inputs
    -- if a lease is held by another process, we shouldn't try to insert a new row regardless
    -- of which key we're trying to acquire a lease on
    WHERE NOT (SELECT lease_exists FROM any_lease_held_by_other_process)
)

INSERT INTO v1_payloads_olap_cutover_job_offset (key, lease_process_id, lease_expires_at, last_external_id)
SELECT ti.key, ti.lease_process_id, ti.lease_expires_at, ti.last_external_id
FROM to_insert ti
ON CONFLICT (key)
DO UPDATE SET
    -- if the lease is held by this process, then we extend the offset to the new value
    -- otherwise it's a new process acquiring the lease, so we should keep the offset where it was before
    last_external_id = CASE
        WHEN EXCLUDED.lease_process_id = v1_payloads_olap_cutover_job_offset.lease_process_id THEN EXCLUDED.last_external_id
        ELSE v1_payloads_olap_cutover_job_offset.last_external_id
    END,

    lease_process_id = EXCLUDED.lease_process_id,
    lease_expires_at = EXCLUDED.lease_expires_at
WHERE v1_payloads_olap_cutover_job_offset.lease_expires_at < NOW() OR v1_payloads_olap_cutover_job_offset.lease_process_id = $1::UUID
RETURNING key, is_completed, lease_process_id, lease_expires_at, last_external_id
`

type AcquireOrExtendOLAPCutoverJobLeaseParams struct {
	Leaseprocessid uuid.UUID          `json:"leaseprocessid"`
	Key            pgtype.Date        `json:"key"`
	Leaseexpiresat pgtype.Timestamptz `json:"leaseexpiresat"`
	Lastexternalid uuid.UUID          `json:"lastexternalid"`
}

func (q *Queries) AcquireOrExtendOLAPCutoverJobLease(ctx context.Context, db DBTX, arg AcquireOrExtendOLAPCutoverJobLeaseParams) (*V1PayloadsOlapCutoverJobOffset, error) {
	row := db.QueryRow(ctx, acquireOrExtendOLAPCutoverJobLease,
		arg.Leaseprocessid,
		arg.Key,
		arg.Leaseexpiresat,
		arg.Lastexternalid,
	)
	var i V1PayloadsOlapCutoverJobOffset
	err := row.Scan(
		&i.Key,
		&i.IsCompleted,
		&i.LeaseProcessID,
		&i.LeaseExpiresAt,
		&i.LastExternalID,
	)
	return &i, err
}

const analyzeV1DAGToTaskOLAP = `-- name: AnalyzeV1DAGToTaskOLAP :exec
ANALYZE v1_dag_to_task_olap
`

func (q *Queries) AnalyzeV1DAGToTaskOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1DAGToTaskOLAP)
	return err
}

const analyzeV1DAGsOLAP = `-- name: AnalyzeV1DAGsOLAP :exec
ANALYZE v1_dags_olap
`

func (q *Queries) AnalyzeV1DAGsOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1DAGsOLAP)
	return err
}

const analyzeV1EventLookupTableOLAP = `-- name: AnalyzeV1EventLookupTableOLAP :exec
ANALYZE v1_event_lookup_table_olap
`

func (q *Queries) AnalyzeV1EventLookupTableOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1EventLookupTableOLAP)
	return err
}

const analyzeV1EventToRunOLAP = `-- name: AnalyzeV1EventToRunOLAP :exec
ANALYZE v1_event_to_run_olap
`

func (q *Queries) AnalyzeV1EventToRunOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1EventToRunOLAP)
	return err
}

const analyzeV1EventsOLAP = `-- name: AnalyzeV1EventsOLAP :exec
ANALYZE v1_events_olap
`

func (q *Queries) AnalyzeV1EventsOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1EventsOLAP)
	return err
}

const analyzeV1LookupTableOLAP = `-- name: AnalyzeV1LookupTableOLAP :exec
ANALYZE v1_lookup_table_olap
`

func (q *Queries) AnalyzeV1LookupTableOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1LookupTableOLAP)
	return err
}

const analyzeV1OtelTraceLookupOLAP = `-- name: AnalyzeV1OtelTraceLookupOLAP :exec
ANALYZE v1_otel_trace_lookup_olap
`

func (q *Queries) AnalyzeV1OtelTraceLookupOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1OtelTraceLookupOLAP)
	return err
}

const analyzeV1OtelTraceOLAP = `-- name: AnalyzeV1OtelTraceOLAP :exec
ANALYZE v1_otel_trace_olap
`

func (q *Queries) AnalyzeV1OtelTraceOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1OtelTraceOLAP)
	return err
}

const analyzeV1PayloadsOLAP = `-- name: AnalyzeV1PayloadsOLAP :exec
ANALYZE v1_payloads_olap
`

func (q *Queries) AnalyzeV1PayloadsOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1PayloadsOLAP)
	return err
}

const analyzeV1RunsOLAP = `-- name: AnalyzeV1RunsOLAP :exec
ANALYZE v1_runs_olap
`

func (q *Queries) AnalyzeV1RunsOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1RunsOLAP)
	return err
}

const analyzeV1StatusesOLAP = `-- name: AnalyzeV1StatusesOLAP :exec
ANALYZE v1_statuses_olap
`

func (q *Queries) AnalyzeV1StatusesOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1StatusesOLAP)
	return err
}

const analyzeV1TaskEventsOLAP = `-- name: AnalyzeV1TaskEventsOLAP :exec
ANALYZE v1_task_events_olap
`

func (q *Queries) AnalyzeV1TaskEventsOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1TaskEventsOLAP)
	return err
}

const analyzeV1TasksOLAP = `-- name: AnalyzeV1TasksOLAP :exec
ANALYZE v1_tasks_olap
`

func (q *Queries) AnalyzeV1TasksOLAP(ctx context.Context, db DBTX) error {
	_, err := db.Exec(ctx, analyzeV1TasksOLAP)
	return err
}

const bulkCreateEventTriggers = `-- name: BulkCreateEventTriggers :exec
WITH inputs AS (
    SELECT
        UNNEST($1::BIGINT[]) AS run_id,
        UNNEST($2::TIMESTAMPTZ[]) AS run_inserted_at,
        UNNEST($3::BIGINT[]) AS event_id,
        UNNEST($4::TIMESTAMPTZ[]) AS event_seen_at,
        UNNEST($5::UUID[]) AS filter_id
)

INSERT INTO v1_event_to_run_olap (
    run_id,
    run_inserted_at,
    event_id,
    event_seen_at,
    filter_id
)
SELECT
    run_id,
    run_inserted_at,
    event_id,
    event_seen_at,
    NULLIF(filter_id, '00000000-0000-0000-0000-000000000000'::UUID) AS filter_id
FROM inputs
ON CONFLICT DO NOTHING
`

type BulkCreateEventTriggersParams struct {
	Runids         []int64              `json:"runids"`
	Runinsertedats []pgtype.Timestamptz `json:"runinsertedats"`
	Eventids       []int64              `json:"eventids"`
	Eventseenats   []pgtype.Timestamptz `json:"eventseenats"`
	Filterids      []uuid.UUID          `json:"filterids"`
}

func (q *Queries) BulkCreateEventTriggers(ctx context.Context, db DBTX, arg BulkCreateEventTriggersParams) error {
	_, err := db.Exec(ctx, bulkCreateEventTriggers,
		arg.Runids,
		arg.Runinsertedats,
		arg.Eventids,
		arg.Eventseenats,
		arg.Filterids,
	)
	return err
}

const checkLastAutovacuumForPartitionedTables = `-- name: CheckLastAutovacuumForPartitionedTables :many
SELECT
    s.schemaname,
    s.relname AS tablename,
    s.last_autovacuum,
    EXTRACT(EPOCH FROM (NOW() - s.last_autovacuum)) AS seconds_since_last_autovacuum
FROM pg_stat_user_tables s
JOIN pg_catalog.pg_class c ON c.oid = (quote_ident(s.schemaname)||'.'||quote_ident(s.relname))::regclass
WHERE s.schemaname = 'public'
    AND c.relispartition = true
    AND c.relkind = 'r'
ORDER BY s.last_autovacuum ASC NULLS LAST
`

type CheckLastAutovacuumForPartitionedTablesRow struct {
	Schemaname                 pgtype.Text        `json:"schemaname"`
	Tablename                  pgtype.Text        `json:"tablename"`
	LastAutovacuum             pgtype.Timestamptz `json:"last_autovacuum"`
	SecondsSinceLastAutovacuum pgtype.Numeric     `json:"seconds_since_last_autovacuum"`
}

func (q *Queries) CheckLastAutovacuumForPartitionedTables(ctx context.Context, db DBTX) ([]*CheckLastAutovacuumForPartitionedTablesRow, error) {
	rows, err := db.Query(ctx, checkLastAutovacuumForPartitionedTables)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*CheckLastAutovacuumForPartitionedTablesRow
	for rows.Next() {
		var i CheckLastAutovacuumForPartitionedTablesRow
		if err := rows.Scan(
			&i.Schemaname,
			&i.Tablename,
			&i.LastAutovacuum,
			&i.SecondsSinceLastAutovacuum,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const cleanUpOLAPCutoverJobOffsets = `-- name: CleanUpOLAPCutoverJobOffsets :exec
DELETE FROM v1_payloads_olap_cutover_job_offset
WHERE NOT key = ANY($1::DATE[])
`

func (q *Queries) CleanUpOLAPCutoverJobOffsets(ctx context.Context, db DBTX, keystokeep []pgtype.Date) error {
	_, err := db.Exec(ctx, cleanUpOLAPCutoverJobOffsets, keystokeep)
	return err
}

const computeOLAPPayloadBatchSize = `-- name: ComputeOLAPPayloadBatchSize :one
SELECT compute_olap_payload_batch_size(
    $1::DATE,
    $2::UUID,
    $3::INTEGER
) AS total_size_bytes
`

type ComputeOLAPPayloadBatchSizeParams struct {
	Partitiondate  pgtype.Date `json:"partitiondate"`
	Lastexternalid uuid.UUID   `json:"lastexternalid"`
	Batchsize      int32       `json:"batchsize"`
}

func (q *Queries) ComputeOLAPPayloadBatchSize(ctx context.Context, db DBTX, arg ComputeOLAPPayloadBatchSizeParams) (int64, error) {
	row := db.QueryRow(ctx, computeOLAPPayloadBatchSize, arg.Partitiondate, arg.Lastexternalid, arg.Batchsize)
	var total_size_bytes int64
	err := row.Scan(&total_size_bytes)
	return total_size_bytes, err
}

const countEvents = `-- name: CountEvents :one
WITH included_events AS (
    SELECT e.tenant_id, e.id, e.external_id, e.seen_at, e.key, e.payload, e.additional_metadata, e.scope, e.triggering_webhook_name
    FROM v1_events_olap e
    WHERE
        e.tenant_id = $1
        AND (
            $2::TEXT[] IS NULL OR
            "key" = ANY($2::TEXT[])
        )
        AND e.seen_at >= $3::TIMESTAMPTZ
        AND (
            $4::TIMESTAMPTZ IS NULL OR
            e.seen_at <= $4::TIMESTAMPTZ
        )
        AND (
            $5::UUID[] IS NULL OR
            EXISTS (
                SELECT 1
                FROM v1_event_to_run_olap etr
                JOIN v1_runs_olap r ON (etr.run_id, etr.run_inserted_at) = (r.id, r.inserted_at)
                WHERE
                    (etr.event_id, etr.event_seen_at) = (e.id, e.seen_at)
                    AND r.workflow_id = ANY($5::UUID[])
                    AND r.inserted_at >= $3::TIMESTAMPTZ
            )
        )
        AND (
            $6::UUID[] IS NULL OR
            (e.id, e.seen_at) IN (
                SELECT event_id, event_seen_at
                FROM v1_event_lookup_table_olap elt
                WHERE elt.tenant_id = $1::UUID AND elt.external_id = ANY($6::UUID[])
            )
        )
        AND (
            $7::JSONB IS NULL OR
            e.additional_metadata @> $7::JSONB
        )
        AND (
            CAST($8::TEXT[] AS v1_readable_status_olap[]) IS NULL OR
            EXISTS (
                SELECT 1
                FROM v1_event_to_run_olap etr
                JOIN v1_runs_olap r ON (etr.run_id, etr.run_inserted_at) = (r.id, r.inserted_at)
                WHERE
                    (etr.event_id, etr.event_seen_at) = (e.id, e.seen_at)
                    AND r.readable_status = ANY(CAST($8::TEXT[] AS v1_readable_status_olap[]))
                    AND r.inserted_at >= $3::TIMESTAMPTZ
            )
        )
        AND (
            $9::TEXT[] IS NULL OR
            e.scope = ANY($9::TEXT[])
        )
        ORDER BY e.seen_at DESC, e.id
    LIMIT 20000
)

SELECT COUNT(*)
FROM included_events e
`

type CountEventsParams struct {
	Tenantid           uuid.UUID          `json:"tenantid"`
	Keys               []string           `json:"keys"`
	Since              pgtype.Timestamptz `json:"since"`
	Until              pgtype.Timestamptz `json:"until"`
	WorkflowIds        []uuid.UUID        `json:"workflowIds"`
	EventIds           []uuid.UUID        `json:"eventIds"`
	AdditionalMetadata []byte             `json:"additionalMetadata"`
	Statuses           []string           `json:"statuses"`
	Scopes             []string           `json:"scopes"`
}

func (q *Queries) CountEvents(ctx context.Context, db DBTX, arg CountEventsParams) (int64, error) {
	row := db.QueryRow(ctx, countEvents,
		arg.Tenantid,
		arg.Keys,
		arg.Since,
		arg.Until,
		arg.WorkflowIds,
		arg.EventIds,
		arg.AdditionalMetadata,
		arg.Statuses,
		arg.Scopes,
	)
	var count int64
	err := row.Scan(&count)
	return count, err
}

const countOLAPTempTableSizeForDAGStatusUpdates = `-- name: CountOLAPTempTableSizeForDAGStatusUpdates :one
SELECT COUNT(*) AS total
FROM v1_task_status_updates_tmp
`

func (q *Queries) CountOLAPTempTableSizeForDAGStatusUpdates(ctx context.Context, db DBTX) (int64, error) {
	row := db.QueryRow(ctx, countOLAPTempTableSizeForDAGStatusUpdates)
	var total int64
	err := row.Scan(&total)
	return total, err
}

const countOLAPTempTableSizeForTaskStatusUpdates = `-- name: CountOLAPTempTableSizeForTaskStatusUpdates :one
SELECT COUNT(*) AS total
FROM v1_task_events_olap_tmp
`

func (q *Queries) CountOLAPTempTableSizeForTaskStatusUpdates(ctx context.Context, db DBTX) (int64, error) {
	row := db.QueryRow(ctx, countOLAPTempTableSizeForTaskStatusUpdates)
	var total int64
	err := row.Scan(&total)
	return total, err
}

const createDagToTaskOLAPSelfMappings = `-- name: CreateDagToTaskOLAPSelfMappings :exec
INSERT INTO v1_dag_to_task_olap (dag_id, dag_inserted_at, task_id, task_inserted_at)
SELECT
    UNNEST($1::bigint[]),
    UNNEST($2::timestamptz[]),
    UNNEST($1::bigint[]),
    UNNEST($2::timestamptz[])
ON CONFLICT DO NOTHING
`

type CreateDagToTaskOLAPSelfMappingsParams struct {
	Dagids         []int64              `json:"dagids"`
	Daginsertedats []pgtype.Timestamptz `json:"daginsertedats"`
}

// For operator-managed DAGs, maps the DAG to its orchestrator task's events (the orchestrator
// shares the DAG's id and inserted_at) so run metadata queries surface the orchestrator's error
// message and timings. Status derivation is unaffected: status queries join through
// v1_tasks_olap, which has no row for the orchestrator.
func (q *Queries) CreateDagToTaskOLAPSelfMappings(ctx context.Context, db DBTX, arg CreateDagToTaskOLAPSelfMappingsParams) error {
	_, err := db.Exec(ctx, createDagToTaskOLAPSelfMappings, arg.Dagids, arg.Daginsertedats)
	return err
}

const createIncomingWebhookValidationFailureLogs = `-- name: CreateIncomingWebhookValidationFailureLogs :exec
WITH inputs AS (
    SELECT
        UNNEST($2::TEXT[]) AS incoming_webhook_name,
        UNNEST($3::TEXT[]) AS error
)
INSERT INTO v1_incoming_webhook_validation_failures_olap(
    tenant_id,
    incoming_webhook_name,
    error
)
SELECT
    $1::UUID,
    i.incoming_webhook_name,
    i.error
FROM inputs i
`

type CreateIncomingWebhookValidationFailureLogsParams struct {
	Tenantid             uuid.UUID `json:"tenantid"`
	Incomingwebhooknames []string  `json:"incomingwebhooknames"`
	Errors               []string  `json:"errors"`
}

func (q *Queries) CreateIncomingWebhookValidationFailureLogs(ctx context.Context, db DBTX, arg CreateIncomingWebhookValidationFailureLogsParams) error {
	_, err := db.Exec(ctx, createIncomingWebhookValidationFailureLogs, arg.Tenantid, arg.Incomingwebhooknames, arg.Errors)
	return err
}

const createOLAPOffloadedPayloadIndexBlock = `-- name: CreateOLAPOffloadedPayloadIndexBlock :exec
INSERT INTO v1_payloads_olap_offloaded_block_index (
    payload_inserted_at_date,
    block_external_id_range,
    index_file_key
)
VALUES (
    $1::DATE,
    uuidrange($2::UUID, $3::UUID, '(]'),
    $4::TEXT
)
ON CONFLICT ON CONSTRAINT v1_payloads_olap_offloaded_block_index_date_range_excl DO NOTHING
`

type CreateOLAPOffloadedPayloadIndexBlockParams struct {
	Payloadinsertedatdate     pgtype.Date `json:"payloadinsertedatdate"`
	Blocklowerexternalidbound uuid.UUID   `json:"blocklowerexternalidbound"`
	Blockupperexternalidbound uuid.UUID   `json:"blockupperexternalidbound"`
	Indexfilekey              string      `json:"indexfilekey"`
}

func (q *Queries) CreateOLAPOffloadedPayloadIndexBlock(ctx context.Context, db DBTX, arg CreateOLAPOffloadedPayloadIndexBlockParams) error {
	_, err := db.Exec(ctx, createOLAPOffloadedPayloadIndexBlock,
		arg.Payloadinsertedatdate,
		arg.Blocklowerexternalidbound,
		arg.Blockupperexternalidbound,
		arg.Indexfilekey,
	)
	return err
}

const createOLAPOptionalTablePartitions = `-- name: CreateOLAPOptionalTablePartitions :one
SELECT
    create_v1_range_partition('v1_events_olap'::text, $1::date) AS v1_events_olap,
    create_v1_range_partition('v1_event_to_run_olap'::text, $1::date) AS v1_event_to_run_olap,
    create_v1_weekly_range_partition('v1_event_lookup_table_olap'::text, $1::date) AS v1_event_lookup_table_olap,
    create_v1_range_partition('v1_incoming_webhook_validation_failures_olap'::text, $1::date) AS v1_incoming_webhook_validation_failures_olap,
    create_v1_range_partition('v1_cel_evaluation_failures_olap'::text, $1::date) AS v1_cel_evaluation_failures_olap,
    create_v1_range_partition('v1_task_events_olap'::text, $1::date) AS v1_task_events_olap,
    create_v1_range_partition('v1_dag_to_task_olap'::text, $1::date) AS v1_dag_to_task_olap,
    create_v1_monthly_range_partition('v1_lookup_table_olap'::text, $1::date) AS v1_lookup_table_olap,
    create_v1_monthly_range_partition('v1_statuses_olap'::text, $1::date) AS v1_statuses_olap
`

type CreateOLAPOptionalTablePartitionsRow struct {
	V1EventsOlap                            int32 `json:"v1_events_olap"`
	V1EventToRunOlap                        int32 `json:"v1_event_to_run_olap"`
	V1EventLookupTableOlap                  int32 `json:"v1_event_lookup_table_olap"`
	V1IncomingWebhookValidationFailuresOlap int32 `json:"v1_incoming_webhook_validation_failures_olap"`
	V1CelEvaluationFailuresOlap             int32 `json:"v1_cel_evaluation_failures_olap"`
	V1TaskEventsOlap                        int32 `json:"v1_task_events_olap"`
	V1DagToTaskOlap                         int32 `json:"v1_dag_to_task_olap"`
	V1LookupTableOlap                       int32 `json:"v1_lookup_table_olap"`
	V1StatusesOlap                          int32 `json:"v1_statuses_olap"`
}

func (q *Queries) CreateOLAPOptionalTablePartitions(ctx context.Context, db DBTX, date pgtype.Date) (*CreateOLAPOptionalTablePartitionsRow, error) {
	row := db.QueryRow(ctx, createOLAPOptionalTablePartitions, date)
	var i CreateOLAPOptionalTablePartitionsRow
	err := row.Scan(
		&i.V1EventsOlap,
		&i.V1EventToRunOlap,
		&i.V1EventLookupTableOlap,
		&i.V1IncomingWebhookValidationFailuresOlap,
		&i.V1CelEvaluationFailuresOlap,
		&i.V1TaskEventsOlap,
		&i.V1DagToTaskOlap,
		&i.V1LookupTableOlap,
		&i.V1StatusesOlap,
	)
	return &i, err
}

const createOLAPOtelPartitions = `-- name: CreateOLAPOtelPartitions :exec
SELECT
    create_v1_range_partition('v1_otel_trace_olap'::text, $1::date),
    create_v1_range_partition('v1_otel_trace_lookup_olap'::text, $1::date)
`

func (q *Queries) CreateOLAPOtelPartitions(ctx context.Context, db DBTX, date pgtype.Date) error {
	_, err := db.Exec(ctx, createOLAPOtelPartitions, date)
	return err
}

const createOLAPPartitions = `-- name: CreateOLAPPartitions :one
SELECT
    create_v1_hash_partitions('v1_task_events_olap_tmp'::text, $1::int) AS v1_task_events_olap_tmp,
    create_v1_hash_partitions('v1_task_status_updates_tmp'::text, $1::int) AS v1_task_status_updates_tmp,
    create_v1_range_partition('v1_tasks_olap'::text, $2::date) AS v1_tasks_olap,
    create_v1_range_partition('v1_runs_olap'::text, $2::date) AS v1_runs_olap,
    create_v1_range_partition('v1_dags_olap'::text, $2::date) AS v1_dags_olap,
    create_v1_range_partition('v1_payloads_olap'::text, $2::date) AS v1_payloads_olap
`

type CreateOLAPPartitionsParams struct {
	Partitions int32       `json:"partitions"`
	Date       pgtype.Date `json:"date"`
}

type CreateOLAPPartitionsRow struct {
	V1TaskEventsOlapTmp    int32 `json:"v1_task_events_olap_tmp"`
	V1TaskStatusUpdatesTmp int32 `json:"v1_task_status_updates_tmp"`
	V1TasksOlap            int32 `json:"v1_tasks_olap"`
	V1RunsOlap             int32 `json:"v1_runs_olap"`
	V1DagsOlap             int32 `json:"v1_dags_olap"`
	V1PayloadsOlap         int32 `json:"v1_payloads_olap"`
}

func (q *Queries) CreateOLAPPartitions(ctx context.Context, db DBTX, arg CreateOLAPPartitionsParams) (*CreateOLAPPartitionsRow, error) {
	row := db.QueryRow(ctx, createOLAPPartitions, arg.Partitions, arg.Date)
	var i CreateOLAPPartitionsRow
	err := row.Scan(
		&i.V1TaskEventsOlapTmp,
		&i.V1TaskStatusUpdatesTmp,
		&i.V1TasksOlap,
		&i.V1RunsOlap,
		&i.V1DagsOlap,
		&i.V1PayloadsOlap,
	)
	return &i, err
}

const createOLAPPayloadRangeChunks = `-- name: CreateOLAPPayloadRangeChunks :many
WITH chunks AS (
    SELECT
        (p).*
    FROM create_olap_payload_offload_range_chunks(
        $1::DATE,
        $2::INTEGER,
        $3::INTEGER,
        $4::UUID
    ) p
)

SELECT
    lower_external_id::UUID,
    upper_external_id::UUID
FROM chunks
`

type CreateOLAPPayloadRangeChunksParams struct {
	Partitiondate  pgtype.Date `json:"partitiondate"`
	Windowsize     int32       `json:"windowsize"`
	Chunksize      int32       `json:"chunksize"`
	Lastexternalid uuid.UUID   `json:"lastexternalid"`
}

type CreateOLAPPayloadRangeChunksRow struct {
	LowerExternalID uuid.UUID `json:"lower_external_id"`
	UpperExternalID uuid.UUID `json:"upper_external_id"`
}

func (q *Queries) CreateOLAPPayloadRangeChunks(ctx context.Context, db DBTX, arg CreateOLAPPayloadRangeChunksParams) ([]*CreateOLAPPayloadRangeChunksRow, error) {
	rows, err := db.Query(ctx, createOLAPPayloadRangeChunks,
		arg.Partitiondate,
		arg.Windowsize,
		arg.Chunksize,
		arg.Lastexternalid,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*CreateOLAPPayloadRangeChunksRow
	for rows.Next() {
		var i CreateOLAPPayloadRangeChunksRow
		if err := rows.Scan(&i.LowerExternalID, &i.UpperExternalID); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

type CreateTaskEventsOLAPParams struct {
	TenantID               uuid.UUID            `json:"tenant_id"`
	TaskID                 int64                `json:"task_id"`
	TaskInsertedAt         pgtype.Timestamptz   `json:"task_inserted_at"`
	EventType              V1EventTypeOlap      `json:"event_type"`
	WorkflowID             uuid.UUID            `json:"workflow_id"`
	EventTimestamp         pgtype.Timestamptz   `json:"event_timestamp"`
	ReadableStatus         V1ReadableStatusOlap `json:"readable_status"`
	RetryCount             int32                `json:"retry_count"`
	ErrorMessage           pgtype.Text          `json:"error_message"`
	Output                 []byte               `json:"output"`
	WorkerID               *uuid.UUID           `json:"worker_id"`
	AdditionalEventData    pgtype.Text          `json:"additional__event_data"`
	AdditionalEventMessage pgtype.Text          `json:"additional__event_message"`
	ExternalID             uuid.UUID            `json:"external_id"`
	DurableInvocationCount int32                `json:"durable_invocation_count"`
	IsDagOrchestrator      bool                 `json:"is_dag_orchestrator"`
}

const createV1PayloadOLAPCutoverTemporaryTable = `-- name: CreateV1PayloadOLAPCutoverTemporaryTable :exec
SELECT copy_v1_payloads_olap_partition_structure($1::DATE)
`

func (q *Queries) CreateV1PayloadOLAPCutoverTemporaryTable(ctx context.Context, db DBTX, date pgtype.Date) error {
	_, err := db.Exec(ctx, createV1PayloadOLAPCutoverTemporaryTable, date)
	return err
}

const deleteOldOLAPPayloadOffloadedBlockIndexRows = `-- name: DeleteOldOLAPPayloadOffloadedBlockIndexRows :exec
DELETE FROM v1_payloads_olap_offloaded_block_index
WHERE payload_inserted_at_date < $1::DATE
`

func (q *Queries) DeleteOldOLAPPayloadOffloadedBlockIndexRows(ctx context.Context, db DBTX, before pgtype.Date) error {
	_, err := db.Exec(ctx, deleteOldOLAPPayloadOffloadedBlockIndexRows, before)
	return err
}

const findMinInsertedAtForDAGStatusUpdates = `-- name: FindMinInsertedAtForDAGStatusUpdates :one
WITH tenants AS (
    SELECT UNNEST(
        find_matching_tenants_in_task_status_updates_tmp_partition(
            $1::int,
            $3::UUID[]
        )
    ) AS tenant_id
)

SELECT
    MIN(u.dag_inserted_at)::TIMESTAMPTZ AS min_inserted_at
FROM tenants t,
    LATERAL list_task_status_updates_tmp(
        $1::int,
        t.tenant_id,
        $2::int
    ) u
`

type FindMinInsertedAtForDAGStatusUpdatesParams struct {
	Partitionnumber int32       `json:"partitionnumber"`
	Eventlimit      int32       `json:"eventlimit"`
	Tenantids       []uuid.UUID `json:"tenantids"`
}

func (q *Queries) FindMinInsertedAtForDAGStatusUpdates(ctx context.Context, db DBTX, arg FindMinInsertedAtForDAGStatusUpdatesParams) (pgtype.Timestamptz, error) {
	row := db.QueryRow(ctx, findMinInsertedAtForDAGStatusUpdates, arg.Partitionnumber, arg.Eventlimit, arg.Tenantids)
	var min_inserted_at pgtype.Timestamptz
	err := row.Scan(&min_inserted_at)
	return min_inserted_at, err
}

const findMinInsertedAtForTaskStatusUpdates = `-- name: FindMinInsertedAtForTaskStatusUpdates :one
WITH tenants AS (
    SELECT UNNEST(
        find_matching_tenants_in_task_events_tmp_partition(
            $1::int,
            $3::UUID[]
        )
    ) AS tenant_id
)

SELECT
    MIN(e.task_inserted_at)::TIMESTAMPTZ AS min_inserted_at
FROM tenants t,
    LATERAL list_task_events_tmp(
        $1::int,
        t.tenant_id,
        $2::int
    ) e
`

type FindMinInsertedAtForTaskStatusUpdatesParams struct {
	Partitionnumber int32       `json:"partitionnumber"`
	Eventlimit      int32       `json:"eventlimit"`
	Tenantids       []uuid.UUID `json:"tenantids"`
}

func (q *Queries) FindMinInsertedAtForTaskStatusUpdates(ctx context.Context, db DBTX, arg FindMinInsertedAtForTaskStatusUpdatesParams) (pgtype.Timestamptz, error) {
	row := db.QueryRow(ctx, findMinInsertedAtForTaskStatusUpdates, arg.Partitionnumber, arg.Eventlimit, arg.Tenantids)
	var min_inserted_at pgtype.Timestamptz
	err := row.Scan(&min_inserted_at)
	return min_inserted_at, err
}

const flattenTasksByExternalIds = `-- name: FlattenTasksByExternalIds :many
WITH lookups AS MATERIALIZED (
    SELECT
        tenant_id, external_id, task_id, dag_id, inserted_at
    FROM
        v1_lookup_table_olap
    WHERE
        external_id = ANY($1::uuid[])
        AND tenant_id = $2::uuid
), tasks_from_dags AS (
    SELECT
        l.tenant_id,
        dt.task_id,
        dt.task_inserted_at
    FROM
        lookups l
    JOIN
        v1_dag_to_task_olap dt ON l.dag_id = dt.dag_id AND l.inserted_at = dt.dag_inserted_at
    WHERE
        l.dag_id IS NOT NULL
        AND dt.dag_inserted_at >= (SELECT MIN(inserted_at) FROM lookups WHERE dag_id IS NOT NULL)
), unioned_tasks AS (
    SELECT
        l.tenant_id AS tenant_id,
        l.task_id AS task_id,
        l.inserted_at AS task_inserted_at
    FROM
        lookups l
    UNION ALL
    SELECT
        t.tenant_id AS tenant_id,
        t.task_id AS task_id,
        t.task_inserted_at AS task_inserted_at
    FROM
        tasks_from_dags t
)
SELECT
    t.tenant_id,
    t.id,
    t.inserted_at,
    t.external_id,
    t.latest_retry_count AS retry_count
FROM
    v1_tasks_olap t
JOIN
    unioned_tasks ut ON (t.inserted_at, t.id) = (ut.task_inserted_at, ut.task_id)
`

type FlattenTasksByExternalIdsParams struct {
	Externalids []uuid.UUID `json:"externalids"`
	Tenantid    uuid.UUID   `json:"tenantid"`
}

type FlattenTasksByExternalIdsRow struct {
	TenantID   uuid.UUID          `json:"tenant_id"`
	ID         int64              `json:"id"`
	InsertedAt pgtype.Timestamptz `json:"inserted_at"`
	ExternalID uuid.UUID          `json:"external_id"`
	RetryCount int32              `json:"retry_count"`
}

// Get retry counts for each task
func (q *Queries) FlattenTasksByExternalIds(ctx context.Context, db DBTX, arg FlattenTasksByExternalIdsParams) ([]*FlattenTasksByExternalIdsRow, error) {
	rows, err := db.Query(ctx, flattenTasksByExternalIds, arg.Externalids, arg.Tenantid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*FlattenTasksByExternalIdsRow
	for rows.Next() {
		var i FlattenTasksByExternalIdsRow
		if err := rows.Scan(
			&i.TenantID,
			&i.ID,
			&i.InsertedAt,
			&i.ExternalID,
			&i.RetryCount,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getDagDurations = `-- name: GetDagDurations :many
SELECT
    lt.external_id,
    MIN(e.event_timestamp) FILTER (WHERE e.readable_status = 'RUNNING')::TIMESTAMPTZ AS started_at,
    MAX(e.event_timestamp) FILTER (WHERE e.readable_status IN ('COMPLETED', 'FAILED', 'CANCELLED'))::TIMESTAMPTZ AS finished_at
FROM
    v1_lookup_table_olap lt
JOIN
    v1_dags_olap d ON (lt.dag_id, lt.inserted_at) = (d.id, d.inserted_at)
JOIN
    v1_dag_to_task_olap dt ON (d.id, d.inserted_at) = (dt.dag_id, dt.dag_inserted_at)
JOIN
    v1_task_events_olap e ON (dt.task_id, dt.task_inserted_at) = (e.task_id, e.task_inserted_at)
WHERE lt.external_id = ANY($1::UUID[])
    AND lt.tenant_id = $2::UUID
    AND d.inserted_at >= $3::TIMESTAMPTZ
    AND dt.dag_inserted_at >= $3::TIMESTAMPTZ
    AND e.task_inserted_at >= $3::TIMESTAMPTZ
GROUP BY lt.external_id
`

type GetDagDurationsParams struct {
	Externalids   []uuid.UUID        `json:"externalids"`
	Tenantid      uuid.UUID          `json:"tenantid"`
	Mininsertedat pgtype.Timestamptz `json:"mininsertedat"`
}

type GetDagDurationsRow struct {
	ExternalID uuid.UUID          `json:"external_id"`
	StartedAt  pgtype.Timestamptz `json:"started_at"`
	FinishedAt pgtype.Timestamptz `json:"finished_at"`
}

func (q *Queries) GetDagDurations(ctx context.Context, db DBTX, arg GetDagDurationsParams) ([]*GetDagDurationsRow, error) {
	rows, err := db.Query(ctx, getDagDurations, arg.Externalids, arg.Tenantid, arg.Mininsertedat)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*GetDagDurationsRow
	for rows.Next() {
		var i GetDagDurationsRow
		if err := rows.Scan(&i.ExternalID, &i.StartedAt, &i.FinishedAt); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getEventByExternalId = `-- name: GetEventByExternalId :one
SELECT e.tenant_id, e.id, e.external_id, e.seen_at, e.key, e.payload, e.additional_metadata, e.scope, e.triggering_webhook_name
FROM v1_event_lookup_table_olap elt
JOIN v1_events_olap e ON (elt.event_id, elt.event_seen_at) = (e.id, e.seen_at)
WHERE elt.external_id = $1::uuid
`

func (q *Queries) GetEventByExternalId(ctx context.Context, db DBTX, eventexternalid uuid.UUID) (*V1EventsOlap, error) {
	row := db.QueryRow(ctx, getEventByExternalId, eventexternalid)
	var i V1EventsOlap
	err := row.Scan(
		&i.TenantID,
		&i.ID,
		&i.ExternalID,
		&i.SeenAt,
		&i.Key,
		&i.Payload,
		&i.AdditionalMetadata,
		&i.Scope,
		&i.TriggeringWebhookName,
	)
	return &i, err
}

const getEventByExternalIdUsingTenantId = `-- name: GetEventByExternalIdUsingTenantId :one
SELECT e.tenant_id, e.id, e.external_id, e.seen_at, e.key, e.payload, e.additional_metadata, e.scope, e.triggering_webhook_name
FROM v1_event_lookup_table_olap elt
JOIN v1_events_olap e ON (elt.event_id, elt.event_seen_at) = (e.id, e.seen_at)
WHERE
    elt.external_id = $1::uuid
    AND elt.tenant_id = $2::uuid
`

type GetEventByExternalIdUsingTenantIdParams struct {
	Eventexternalid uuid.UUID `json:"eventexternalid"`
	Tenantid        uuid.UUID `json:"tenantid"`
}

func (q *Queries) GetEventByExternalIdUsingTenantId(ctx context.Context, db DBTX, arg GetEventByExternalIdUsingTenantIdParams) (*V1EventsOlap, error) {
	row := db.QueryRow(ctx, getEventByExternalIdUsingTenantId, arg.Eventexternalid, arg.Tenantid)
	var i V1EventsOlap
	err := row.Scan(
		&i.TenantID,
		&i.ID,
		&i.ExternalID,
		&i.SeenAt,
		&i.Key,
		&i.Payload,
		&i.AdditionalMetadata,
		&i.Scope,
		&i.TriggeringWebhookName,
	)
	return &i, err
}

const getOLAPOffloadedPayloadIndexBlocks = `-- name: GetOLAPOffloadedPayloadIndexBlocks :many
WITH inputs AS (
    SELECT
        UNNEST($1::DATE[]) AS inserted_at_date,
        UNNEST($2::UUID[]) AS external_id
)

SELECT i.external_id::UUID AS external_id, p.index_file_key
FROM v1_payloads_olap_offloaded_block_index p
JOIN inputs i ON
    p.payload_inserted_at_date = i.inserted_at_date
    AND p.block_external_id_range @> i.external_id
`

type GetOLAPOffloadedPayloadIndexBlocksParams struct {
	Insertedats []pgtype.Date `json:"insertedats"`
	Externalids []uuid.UUID   `json:"externalids"`
}

type GetOLAPOffloadedPayloadIndexBlocksRow struct {
	ExternalID   uuid.UUID `json:"external_id"`
	IndexFileKey string    `json:"index_file_key"`
}

// todo: make sure this join uses the index correctly
func (q *Queries) GetOLAPOffloadedPayloadIndexBlocks(ctx context.Context, db DBTX, arg GetOLAPOffloadedPayloadIndexBlocksParams) ([]*GetOLAPOffloadedPayloadIndexBlocksRow, error) {
	rows, err := db.Query(ctx, getOLAPOffloadedPayloadIndexBlocks, arg.Insertedats, arg.Externalids)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*GetOLAPOffloadedPayloadIndexBlocksRow
	for rows.Next() {
		var i GetOLAPOffloadedPayloadIndexBlocksRow
		if err := rows.Scan(&i.ExternalID, &i.IndexFileKey); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getRunsListRecursive = `-- name: GetRunsListRecursive :many
WITH RECURSIVE all_runs AS (
  -- seed term
    SELECT
        t.id,
        t.inserted_at,
        t.tenant_id,
        t.external_id,
        t.parent_task_external_id,
        0 AS depth
    FROM
        v1_lookup_table_olap lt
        JOIN v1_tasks_olap t
        ON t.inserted_at = lt.inserted_at
        AND t.id          = lt.task_id
    WHERE
        lt.external_id = ANY($2::uuid[])

    UNION ALL

    -- single recursive term for both DAG- and TASK-driven children
    SELECT
        t.id,
        t.inserted_at,
        t.tenant_id,
        t.external_id,
        t.parent_task_external_id,
        ar.depth + 1 AS depth
    FROM
        v1_runs_olap r
    JOIN all_runs ar ON ar.external_id = r.parent_task_external_id

    -- only present when r.kind = 'DAG'
    LEFT JOIN v1_dag_to_task_olap dt ON r.kind = 'DAG' AND r.id = dt.dag_id AND r.inserted_at = dt.dag_inserted_at

    -- pick the correct task row for either branch
    JOIN v1_tasks_olap t
      ON (
        r.kind = 'DAG'
        AND t.id = dt.task_id
        AND t.inserted_at = dt.task_inserted_at
         )
      OR (
        r.kind = 'TASK'
        AND t.id = r.id
        AND t.inserted_at = r.inserted_at
      )
  WHERE
    r.tenant_id = $1::uuid
    AND ar.depth < $3::int
    AND r.inserted_at >= $4::timestamptz
    AND t.inserted_at >= $4::timestamptz
)
SELECT
  tenant_id,
  id,
  inserted_at,
  external_id,
  parent_task_external_id,
  depth
FROM
  all_runs
WHERE
  tenant_id = $1::uuid
`

type GetRunsListRecursiveParams struct {
	Tenantid        uuid.UUID          `json:"tenantid"`
	Taskexternalids []uuid.UUID        `json:"taskexternalids"`
	Depth           int32              `json:"depth"`
	Createdafter    pgtype.Timestamptz `json:"createdafter"`
}

type GetRunsListRecursiveRow struct {
	TenantID             uuid.UUID          `json:"tenant_id"`
	ID                   int64              `json:"id"`
	InsertedAt           pgtype.Timestamptz `json:"inserted_at"`
	ExternalID           uuid.UUID          `json:"external_id"`
	ParentTaskExternalID *uuid.UUID         `json:"parent_task_external_id"`
	Depth                int32              `json:"depth"`
}

func (q *Queries) GetRunsListRecursive(ctx context.Context, db DBTX, arg GetRunsListRecursiveParams) ([]*GetRunsListRecursiveRow, error) {
	rows, err := db.Query(ctx, getRunsListRecursive,
		arg.Tenantid,
		arg.Taskexternalids,
		arg.Depth,
		arg.Createdafter,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*GetRunsListRecursiveRow
	for rows.Next() {
		var i GetRunsListRecursiveRow
		if err := rows.Scan(
			&i.TenantID,
			&i.ID,
			&i.InsertedAt,
			&i.ExternalID,
			&i.ParentTaskExternalID,
			&i.Depth,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getTaskDurationsByTaskIds = `-- name: GetTaskDurationsByTaskIds :many
WITH input AS (
    SELECT
        UNNEST($1::bigint[]) AS task_id,
        UNNEST($2::timestamptz[]) AS inserted_at,
        UNNEST($3::v1_readable_status_olap[]) AS readable_status
), task_data AS (
    SELECT
        i.task_id,
        i.inserted_at,
        t.external_id,
        t.display_name,
        t.readable_status,
        t.latest_retry_count,
        t.tenant_id
    FROM
        input i
    JOIN
        v1_tasks_olap t ON (t.inserted_at, t.id, t.readable_status, t.tenant_id) = (i.inserted_at, i.task_id, i.readable_status, $4::uuid)
), task_events AS (
    SELECT
        td.task_id,
        td.inserted_at,
        e.event_type,
        e.event_timestamp,
        e.readable_status
    FROM
        task_data td
    JOIN
        v1_task_events_olap e ON (e.tenant_id, e.task_id, e.task_inserted_at, e.retry_count) = (td.tenant_id, td.task_id, td.inserted_at, td.latest_retry_count)
    WHERE
        e.task_inserted_at >= (SELECT MIN(inserted_at) FROM input)
), task_times AS (
    SELECT
        task_id,
        inserted_at,
        MIN(CASE WHEN event_type = 'STARTED' THEN event_timestamp END) AS started_at,
        MAX(CASE WHEN readable_status = ANY(ARRAY['COMPLETED', 'FAILED', 'CANCELLED']::v1_readable_status_olap[])
            THEN event_timestamp END) AS finished_at
    FROM task_events
    GROUP BY task_id, inserted_at
)
SELECT
    tt.started_at::timestamptz AS started_at,
    tt.finished_at::timestamptz AS finished_at
FROM
    task_data td
LEFT JOIN
    task_times tt ON (td.task_id, td.inserted_at) = (tt.task_id, tt.inserted_at)
ORDER BY td.task_id, td.inserted_at
`

type GetTaskDurationsByTaskIdsParams struct {
	Taskids          []int64                `json:"taskids"`
	Taskinsertedats  []pgtype.Timestamptz   `json:"taskinsertedats"`
	Readablestatuses []V1ReadableStatusOlap `json:"readablestatuses"`
	Tenantid         uuid.UUID              `json:"tenantid"`
}

type GetTaskDurationsByTaskIdsRow struct {
	StartedAt  pgtype.Timestamptz `json:"started_at"`
	FinishedAt pgtype.Timestamptz `json:"finished_at"`
}

func (q *Queries) GetTaskDurationsByTaskIds(ctx context.Context, db DBTX, arg GetTaskDurationsByTaskIdsParams) ([]*GetTaskDurationsByTaskIdsRow, error) {
	rows, err := db.Query(ctx, getTaskDurationsByTaskIds,
		arg.Taskids,
		arg.Taskinsertedats,
		arg.Readablestatuses,
		arg.Tenantid,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*GetTaskDurationsByTaskIdsRow
	for rows.Next() {
		var i GetTaskDurationsByTaskIdsRow
		if err := rows.Scan(&i.StartedAt, &i.FinishedAt); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getTaskPointMetrics = `-- name: GetTaskPointMetrics :many
SELECT
    DATE_BIN(
        COALESCE($1::INTERVAL, '1 minute'),
        inserted_at,
        TIMESTAMPTZ '1970-01-01 00:00:00+00'
    ) :: TIMESTAMPTZ AS minute_bucket,
    COUNT(*) FILTER (WHERE readable_status = 'COMPLETED') AS completed_count,
    COUNT(*) FILTER (WHERE readable_status = 'FAILED') AS failed_count
FROM v1_statuses_olap
WHERE
    tenant_id = $2::UUID
    AND inserted_at BETWEEN $3::TIMESTAMPTZ AND $4::TIMESTAMPTZ
GROUP BY minute_bucket
ORDER BY minute_bucket
`

type GetTaskPointMetricsParams struct {
	Interval      pgtype.Interval    `json:"interval"`
	Tenantid      uuid.UUID          `json:"tenantid"`
	Createdafter  pgtype.Timestamptz `json:"createdafter"`
	Createdbefore pgtype.Timestamptz `json:"createdbefore"`
}

type GetTaskPointMetricsRow struct {
	MinuteBucket   pgtype.Timestamptz `json:"minute_bucket"`
	CompletedCount int64              `json:"completed_count"`
	FailedCount    int64              `json:"failed_count"`
}

func (q *Queries) GetTaskPointMetrics(ctx context.Context, db DBTX, arg GetTaskPointMetricsParams) ([]*GetTaskPointMetricsRow, error) {
	rows, err := db.Query(ctx, getTaskPointMetrics,
		arg.Interval,
		arg.Tenantid,
		arg.Createdafter,
		arg.Createdbefore,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*GetTaskPointMetricsRow
	for rows.Next() {
		var i GetTaskPointMetricsRow
		if err := rows.Scan(&i.MinuteBucket, &i.CompletedCount, &i.FailedCount); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getTaskStartedTimestamps = `-- name: GetTaskStartedTimestamps :many
SELECT
    task_id,
    task_inserted_at,
    retry_count,
    MIN(event_timestamp)::timestamptz AS started_at
FROM
    v1_task_events_olap
WHERE
    tenant_id = $1::uuid
    AND (task_id, task_inserted_at, retry_count) IN (
        SELECT UNNEST($2::bigint[]), UNNEST($3::timestamptz[]), UNNEST($4::int[])
    )
    AND task_inserted_at >= $5::timestamptz
    AND event_type = 'STARTED'
GROUP BY task_id, task_inserted_at, retry_count
`

type GetTaskStartedTimestampsParams struct {
	Tenantid          uuid.UUID            `json:"tenantid"`
	Taskids           []int64              `json:"taskids"`
	Taskinsertedat    []pgtype.Timestamptz `json:"taskinsertedat"`
	Retrycounts       []int32              `json:"retrycounts"`
	Mintaskinsertedat pgtype.Timestamptz   `json:"mintaskinsertedat"`
}

type GetTaskStartedTimestampsRow struct {
	TaskID         int64              `json:"task_id"`
	TaskInsertedAt pgtype.Timestamptz `json:"task_inserted_at"`
	RetryCount     int32              `json:"retry_count"`
	StartedAt      pgtype.Timestamptz `json:"started_at"`
}

func (q *Queries) GetTaskStartedTimestamps(ctx context.Context, db DBTX, arg GetTaskStartedTimestampsParams) ([]*GetTaskStartedTimestampsRow, error) {
	rows, err := db.Query(ctx, getTaskStartedTimestamps,
		arg.Tenantid,
		arg.Taskids,
		arg.Taskinsertedat,
		arg.Retrycounts,
		arg.Mintaskinsertedat,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*GetTaskStartedTimestampsRow
	for rows.Next() {
		var i GetTaskStartedTimestampsRow
		if err := rows.Scan(
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.RetryCount,
			&i.StartedAt,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getTenantStatusMetrics = `-- name: GetTenantStatusMetrics :one
WITH task_external_ids AS (
    SELECT external_id
    FROM v1_runs_olap
    WHERE (
        $5::UUID IS NULL OR parent_task_external_id = $5::UUID
    ) AND (
        $6::UUID IS NULL
        OR (id, inserted_at) IN (
            SELECT etr.run_id, etr.run_inserted_at
            FROM v1_event_lookup_table_olap lt
            JOIN v1_events_olap e ON (lt.tenant_id, lt.event_id, lt.event_seen_at) = (e.tenant_id, e.id, e.seen_at)
            JOIN v1_event_to_run_olap etr ON (e.id, e.seen_at) = (etr.event_id, etr.event_seen_at)
            WHERE
                lt.tenant_id = $1::uuid
                AND lt.external_id = $6::UUID
        )
    )
    AND (
        $7::text[] IS NULL
        OR $8::text[] IS NULL
        OR EXISTS (
            SELECT 1 FROM jsonb_each_text(additional_metadata) kv
            JOIN LATERAL (
                SELECT unnest($7::text[]) AS k,
                    unnest($8::text[]) AS v
            ) AS u ON kv.key = u.k AND kv.value = u.v
        )
    )
    AND tenant_id = $1::UUID
    AND inserted_at >= $2::TIMESTAMPTZ
    AND (
        $3::TIMESTAMPTZ IS NULL OR inserted_at <= $3::TIMESTAMPTZ
    )
    AND (
        $4::UUID[] IS NULL OR workflow_id = ANY($4::UUID[])
    )
)

SELECT
    tenant_id,
    COUNT(*) FILTER (WHERE readable_status = 'QUEUED') AS total_queued,
    COUNT(*) FILTER (WHERE readable_status = 'RUNNING') AS total_running,
    COUNT(*) FILTER (WHERE readable_status = 'COMPLETED') AS total_completed,
    COUNT(*) FILTER (WHERE readable_status = 'CANCELLED') AS total_cancelled,
    COUNT(*) FILTER (WHERE readable_status = 'FAILED') AS total_failed,
    COUNT(*) FILTER (WHERE readable_status = 'EVICTED') AS total_evicted
FROM v1_statuses_olap
WHERE
    tenant_id = $1::UUID
    AND inserted_at >= $2::TIMESTAMPTZ
    AND (
        $3::TIMESTAMPTZ IS NULL OR inserted_at <= $3::TIMESTAMPTZ
    )
    AND (
        $4::UUID[] IS NULL OR workflow_id = ANY($4::UUID[])
    )
    AND external_id IN (
        SELECT external_id
        FROM task_external_ids
    )
GROUP BY tenant_id
`

type GetTenantStatusMetricsParams struct {
	Tenantid                  uuid.UUID          `json:"tenantid"`
	Createdafter              pgtype.Timestamptz `json:"createdafter"`
	CreatedBefore             pgtype.Timestamptz `json:"createdBefore"`
	WorkflowIds               []uuid.UUID        `json:"workflowIds"`
	ParentTaskExternalId      *uuid.UUID         `json:"parentTaskExternalId"`
	TriggeringEventExternalId *uuid.UUID         `json:"triggeringEventExternalId"`
	AdditionalMetaKeys        []string           `json:"additionalMetaKeys"`
	AdditionalMetaValues      []string           `json:"additionalMetaValues"`
}

type GetTenantStatusMetricsRow struct {
	TenantID       uuid.UUID `json:"tenant_id"`
	TotalQueued    int64     `json:"total_queued"`
	TotalRunning   int64     `json:"total_running"`
	TotalCompleted int64     `json:"total_completed"`
	TotalCancelled int64     `json:"total_cancelled"`
	TotalFailed    int64     `json:"total_failed"`
	TotalEvicted   int64     `json:"total_evicted"`
}

func (q *Queries) GetTenantStatusMetrics(ctx context.Context, db DBTX, arg GetTenantStatusMetricsParams) (*GetTenantStatusMetricsRow, error) {
	row := db.QueryRow(ctx, getTenantStatusMetrics,
		arg.Tenantid,
		arg.Createdafter,
		arg.CreatedBefore,
		arg.WorkflowIds,
		arg.ParentTaskExternalId,
		arg.TriggeringEventExternalId,
		arg.AdditionalMetaKeys,
		arg.AdditionalMetaValues,
	)
	var i GetTenantStatusMetricsRow
	err := row.Scan(
		&i.TenantID,
		&i.TotalQueued,
		&i.TotalRunning,
		&i.TotalCompleted,
		&i.TotalCancelled,
		&i.TotalFailed,
		&i.TotalEvicted,
	)
	return &i, err
}

const getWorkflowRunIdFromDagIdInsertedAt = `-- name: GetWorkflowRunIdFromDagIdInsertedAt :one
SELECT external_id
FROM v1_dags_olap
WHERE
    id = $1::bigint
    AND inserted_at = $2::timestamptz
`

type GetWorkflowRunIdFromDagIdInsertedAtParams struct {
	Dagid         int64              `json:"dagid"`
	Daginsertedat pgtype.Timestamptz `json:"daginsertedat"`
}

func (q *Queries) GetWorkflowRunIdFromDagIdInsertedAt(ctx context.Context, db DBTX, arg GetWorkflowRunIdFromDagIdInsertedAtParams) (uuid.UUID, error) {
	row := db.QueryRow(ctx, getWorkflowRunIdFromDagIdInsertedAt, arg.Dagid, arg.Daginsertedat)
	var external_id uuid.UUID
	err := row.Scan(&external_id)
	return external_id, err
}

const listEventKeys = `-- name: ListEventKeys :many
SELECT DISTINCT key
FROM
    v1_events_olap
WHERE
    tenant_id = $1::uuid
    AND seen_at > NOW() - INTERVAL '1 day'
`

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

const listEvents = `-- name: ListEvents :many
SELECT e.tenant_id, e.id, e.external_id, e.seen_at, e.key, e.payload, e.additional_metadata, e.scope, e.triggering_webhook_name
FROM v1_events_olap e
WHERE
    e.tenant_id = $1
    AND (
        $2::TEXT[] IS NULL OR
        "key" = ANY($2::TEXT[])
    )
    AND e.seen_at >= $3::TIMESTAMPTZ
    AND (
        $4::TIMESTAMPTZ IS NULL OR
        e.seen_at <= $4::TIMESTAMPTZ
    )
    AND (
        $5::UUID[] IS NULL OR
        EXISTS (
            SELECT 1
            FROM v1_event_to_run_olap etr
            JOIN v1_runs_olap r ON (etr.run_id, etr.run_inserted_at) = (r.id, r.inserted_at)
            WHERE
                (etr.event_id, etr.event_seen_at) = (e.id, e.seen_at)
                AND r.workflow_id = ANY($5::UUID[])
                AND r.inserted_at >= $3::TIMESTAMPTZ
        )
    )
    AND (
        $6::UUID[] IS NULL OR
        (e.id, e.seen_at) IN (
            SELECT event_id, event_seen_at
            FROM v1_event_lookup_table_olap elt
            WHERE elt.tenant_id = $1::UUID AND elt.external_id = ANY($6::UUID[])
        )
    )
    AND (
        $7::JSONB IS NULL OR
        e.additional_metadata @> $7::JSONB
    )
    AND (
        CAST($8::TEXT[] AS v1_readable_status_olap[]) IS NULL OR
        EXISTS (
            SELECT 1
            FROM v1_event_to_run_olap etr
            JOIN v1_runs_olap r ON (etr.run_id, etr.run_inserted_at) = (r.id, r.inserted_at)
            WHERE
                (etr.event_id, etr.event_seen_at) = (e.id, e.seen_at)
                AND r.readable_status = ANY(CAST($8::TEXT[] AS v1_readable_status_olap[]))
                AND r.inserted_at >= $3::TIMESTAMPTZ
        )
    )
    AND (
        $9::TEXT[] IS NULL OR
        e.scope = ANY($9::TEXT[])
    )
ORDER BY e.seen_at DESC, e.id
OFFSET
    COALESCE($10::BIGINT, 0)
LIMIT
    COALESCE($11::BIGINT, 50)
`

type ListEventsParams struct {
	Tenantid           uuid.UUID          `json:"tenantid"`
	Keys               []string           `json:"keys"`
	Since              pgtype.Timestamptz `json:"since"`
	Until              pgtype.Timestamptz `json:"until"`
	WorkflowIds        []uuid.UUID        `json:"workflowIds"`
	EventIds           []uuid.UUID        `json:"eventIds"`
	AdditionalMetadata []byte             `json:"additionalMetadata"`
	Statuses           []string           `json:"statuses"`
	Scopes             []string           `json:"scopes"`
	Offset             pgtype.Int8        `json:"offset"`
	Limit              pgtype.Int8        `json:"limit"`
}

func (q *Queries) ListEvents(ctx context.Context, db DBTX, arg ListEventsParams) ([]*V1EventsOlap, error) {
	rows, err := db.Query(ctx, listEvents,
		arg.Tenantid,
		arg.Keys,
		arg.Since,
		arg.Until,
		arg.WorkflowIds,
		arg.EventIds,
		arg.AdditionalMetadata,
		arg.Statuses,
		arg.Scopes,
		arg.Offset,
		arg.Limit,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*V1EventsOlap
	for rows.Next() {
		var i V1EventsOlap
		if err := rows.Scan(
			&i.TenantID,
			&i.ID,
			&i.ExternalID,
			&i.SeenAt,
			&i.Key,
			&i.Payload,
			&i.AdditionalMetadata,
			&i.Scope,
			&i.TriggeringWebhookName,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listOLAPPartitionsBeforeDate = `-- name: ListOLAPPartitionsBeforeDate :many
WITH task_partitions AS (
    SELECT 'v1_tasks_olap' AS parent_table, p::text as partition_name FROM get_v1_partitions_before_date('v1_tasks_olap'::text, $3::date) AS p
), dag_partitions AS (
    SELECT 'v1_dags_olap' AS parent_table, p::text as partition_name FROM get_v1_partitions_before_date('v1_dags_olap', $3::date) AS p
), runs_partitions AS (
    SELECT 'v1_runs_olap' AS parent_table, p::text as partition_name FROM get_v1_partitions_before_date('v1_runs_olap', $3::date) AS p
), events_partitions AS (
    SELECT 'v1_events_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_partitions_before_date('v1_events_olap', $3::date) AS p
), event_trigger_partitions AS (
    SELECT 'v1_event_to_run_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_partitions_before_date('v1_event_to_run_olap', $3::date) AS p
), events_lookup_table_partitions AS (
    SELECT 'v1_event_lookup_table_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_weekly_partitions_before_date('v1_event_lookup_table_olap', $3::date) AS p
), incoming_webhook_validation_failure_partitions AS (
    SELECT 'v1_incoming_webhook_validation_failures_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_partitions_before_date('v1_incoming_webhook_validation_failures_olap', $3::date) AS p
), cel_evaluation_failures_partitions AS (
    SELECT 'v1_cel_evaluation_failures_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_partitions_before_date('v1_cel_evaluation_failures_olap', $3::date) AS p
), payloads_partitions AS (
    SELECT 'v1_payloads_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_partitions_before_date('v1_payloads_olap', $3::date) AS p
), otel_trace_partitions AS (
    SELECT 'v1_otel_trace_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_partitions_before_date('v1_otel_trace_olap', $3::date) AS p
), otel_trace_lookup_partitions AS (
    SELECT 'v1_otel_trace_lookup_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_partitions_before_date('v1_otel_trace_lookup_olap', $3::date) AS p
), task_events_partitions AS (
    SELECT 'v1_task_events_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_partitions_before_date('v1_task_events_olap', $3::date) AS p
), dag_to_task_partitions AS (
    SELECT 'v1_dag_to_task_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_partitions_before_date('v1_dag_to_task_olap', $3::date) AS p
), lookup_table_partitions AS (
    SELECT 'v1_lookup_table_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_monthly_partitions_before_date('v1_lookup_table_olap', $3::date) AS p
), statuses_partitions AS (
    SELECT 'v1_statuses_olap' AS parent_table, p::TEXT AS partition_name FROM get_v1_monthly_partitions_before_date('v1_statuses_olap', $3::date) AS p
), candidates AS (
    SELECT
        parent_table, partition_name
    FROM
        task_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        dag_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        runs_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        events_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        event_trigger_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        events_lookup_table_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        incoming_webhook_validation_failure_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        cel_evaluation_failures_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        payloads_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        otel_trace_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        otel_trace_lookup_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        task_events_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        dag_to_task_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        lookup_table_partitions

    UNION ALL

    SELECT
        parent_table, partition_name
    FROM
        statuses_partitions
)

SELECT parent_table, partition_name
FROM candidates
WHERE
    CASE
        WHEN $1::BOOLEAN THEN TRUE
        ELSE parent_table NOT IN (
            'v1_events_olap',
            'v1_event_to_run_olap',
            'v1_event_lookup_table_olap',
            'v1_cel_evaluation_failures_olap',
            'v1_incoming_webhook_validation_failures_olap',
            'v1_task_events_olap',
            'v1_dag_to_task_olap',
            'v1_lookup_table_olap',
            'v1_statuses_olap'
        )
    END
    AND CASE
        WHEN $2::BOOLEAN THEN TRUE
        ELSE parent_table NOT IN ('v1_otel_trace_olap', 'v1_otel_trace_lookup_olap')
    END
`

type ListOLAPPartitionsBeforeDateParams struct {
	Shouldmanageoptionaltablepartitions bool        `json:"shouldmanageoptionaltablepartitions"`
	Shouldpartitionoteltables           bool        `json:"shouldpartitionoteltables"`
	Date                                pgtype.Date `json:"date"`
}

type ListOLAPPartitionsBeforeDateRow struct {
	ParentTable   string `json:"parent_table"`
	PartitionName string `json:"partition_name"`
}

func (q *Queries) ListOLAPPartitionsBeforeDate(ctx context.Context, db DBTX, arg ListOLAPPartitionsBeforeDateParams) ([]*ListOLAPPartitionsBeforeDateRow, error) {
	rows, err := db.Query(ctx, listOLAPPartitionsBeforeDate, arg.Shouldmanageoptionaltablepartitions, arg.Shouldpartitionoteltables, arg.Date)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ListOLAPPartitionsBeforeDateRow
	for rows.Next() {
		var i ListOLAPPartitionsBeforeDateRow
		if err := rows.Scan(&i.ParentTable, &i.PartitionName); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listPaginatedOLAPPayloadsForOffload = `-- name: ListPaginatedOLAPPayloadsForOffload :many
WITH payloads AS (
    SELECT
        (p).*
    FROM list_paginated_olap_payloads_for_offload(
        $1::DATE,
        $2::UUID,
        $3::UUID,
        $4::INTEGER
    ) p
)
SELECT
    tenant_id::UUID,
    external_id::UUID,
    location::v1_payload_location_olap,
    COALESCE(external_location_key, '')::TEXT AS external_location_key,
    inline_content::JSONB AS inline_content,
    inserted_at::TIMESTAMPTZ,
    updated_at::TIMESTAMPTZ
FROM payloads
`

type ListPaginatedOLAPPayloadsForOffloadParams struct {
	Partitiondate  pgtype.Date `json:"partitiondate"`
	Lastexternalid uuid.UUID   `json:"lastexternalid"`
	Nextexternalid uuid.UUID   `json:"nextexternalid"`
	Batchsize      int32       `json:"batchsize"`
}

type ListPaginatedOLAPPayloadsForOffloadRow struct {
	TenantID            uuid.UUID             `json:"tenant_id"`
	ExternalID          uuid.UUID             `json:"external_id"`
	Location            V1PayloadLocationOlap `json:"location"`
	ExternalLocationKey string                `json:"external_location_key"`
	InlineContent       []byte                `json:"inline_content"`
	InsertedAt          pgtype.Timestamptz    `json:"inserted_at"`
	UpdatedAt           pgtype.Timestamptz    `json:"updated_at"`
}

func (q *Queries) ListPaginatedOLAPPayloadsForOffload(ctx context.Context, db DBTX, arg ListPaginatedOLAPPayloadsForOffloadParams) ([]*ListPaginatedOLAPPayloadsForOffloadRow, error) {
	rows, err := db.Query(ctx, listPaginatedOLAPPayloadsForOffload,
		arg.Partitiondate,
		arg.Lastexternalid,
		arg.Nextexternalid,
		arg.Batchsize,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ListPaginatedOLAPPayloadsForOffloadRow
	for rows.Next() {
		var i ListPaginatedOLAPPayloadsForOffloadRow
		if err := rows.Scan(
			&i.TenantID,
			&i.ExternalID,
			&i.Location,
			&i.ExternalLocationKey,
			&i.InlineContent,
			&i.InsertedAt,
			&i.UpdatedAt,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listTaskEvents = `-- name: ListTaskEvents :many
WITH aggregated_events AS (
  SELECT
    tenant_id,
    task_id,
    task_inserted_at,
    retry_count,
    event_type,
    durable_invocation_count,
    MIN(event_timestamp) AS time_first_seen,
    MAX(event_timestamp) AS time_last_seen,
    COUNT(*) AS count,
    MIN(id) AS first_id
  FROM v1_task_events_olap
  WHERE
    tenant_id = $3::uuid
    AND task_id = $4::bigint
    AND task_inserted_at = $5::timestamptz
  GROUP BY tenant_id, task_id, task_inserted_at, retry_count, event_type, durable_invocation_count
)
SELECT
  a.tenant_id,
  a.task_id,
  a.task_inserted_at,
  a.retry_count,
  a.event_type,
  a.durable_invocation_count,
  a.time_first_seen,
  a.time_last_seen,
  a.count,
  t.id,
  t.event_timestamp,
  t.readable_status,
  t.error_message,
  t.output,
  t.external_id AS event_external_id,
  t.worker_id,
  t.additional__event_data,
  t.additional__event_message,
  tsk.display_name AS task_display_name
FROM aggregated_events a
JOIN v1_task_events_olap t
  ON t.tenant_id = a.tenant_id
  AND t.task_id = a.task_id
  AND t.task_inserted_at = a.task_inserted_at
  AND t.id = a.first_id
JOIN v1_tasks_olap tsk ON (tsk.id, tsk.inserted_at) = (t.task_id, t.task_inserted_at)
ORDER BY a.time_first_seen DESC, t.event_timestamp DESC
LIMIT $2::bigint OFFSET $1::bigint
`

type ListTaskEventsParams struct {
	Eventoffset    int64              `json:"eventoffset"`
	Eventlimit     int64              `json:"eventlimit"`
	Tenantid       uuid.UUID          `json:"tenantid"`
	Taskid         int64              `json:"taskid"`
	Taskinsertedat pgtype.Timestamptz `json:"taskinsertedat"`
}

type ListTaskEventsRow struct {
	TenantID               uuid.UUID            `json:"tenant_id"`
	TaskID                 int64                `json:"task_id"`
	TaskInsertedAt         pgtype.Timestamptz   `json:"task_inserted_at"`
	RetryCount             int32                `json:"retry_count"`
	EventType              V1EventTypeOlap      `json:"event_type"`
	DurableInvocationCount int32                `json:"durable_invocation_count"`
	TimeFirstSeen          interface{}          `json:"time_first_seen"`
	TimeLastSeen           interface{}          `json:"time_last_seen"`
	Count                  int64                `json:"count"`
	ID                     int64                `json:"id"`
	EventTimestamp         pgtype.Timestamptz   `json:"event_timestamp"`
	ReadableStatus         V1ReadableStatusOlap `json:"readable_status"`
	ErrorMessage           pgtype.Text          `json:"error_message"`
	Output                 []byte               `json:"output"`
	EventExternalID        uuid.UUID            `json:"event_external_id"`
	WorkerID               *uuid.UUID           `json:"worker_id"`
	AdditionalEventData    pgtype.Text          `json:"additional__event_data"`
	AdditionalEventMessage pgtype.Text          `json:"additional__event_message"`
	TaskDisplayName        string               `json:"task_display_name"`
}

func (q *Queries) ListTaskEvents(ctx context.Context, db DBTX, arg ListTaskEventsParams) ([]*ListTaskEventsRow, error) {
	rows, err := db.Query(ctx, listTaskEvents,
		arg.Eventoffset,
		arg.Eventlimit,
		arg.Tenantid,
		arg.Taskid,
		arg.Taskinsertedat,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ListTaskEventsRow
	for rows.Next() {
		var i ListTaskEventsRow
		if err := rows.Scan(
			&i.TenantID,
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.RetryCount,
			&i.EventType,
			&i.DurableInvocationCount,
			&i.TimeFirstSeen,
			&i.TimeLastSeen,
			&i.Count,
			&i.ID,
			&i.EventTimestamp,
			&i.ReadableStatus,
			&i.ErrorMessage,
			&i.Output,
			&i.EventExternalID,
			&i.WorkerID,
			&i.AdditionalEventData,
			&i.AdditionalEventMessage,
			&i.TaskDisplayName,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listTaskEventsForWorkflowRun = `-- name: ListTaskEventsForWorkflowRun :many
WITH tasks AS (
    SELECT dt.task_id, dt.task_inserted_at, dt.dag_id, dt.dag_inserted_at
    FROM v1_lookup_table_olap lt
    JOIN v1_dag_to_task_olap dt ON lt.dag_id = dt.dag_id AND lt.inserted_at = dt.dag_inserted_at
    WHERE
        lt.external_id = $2::uuid
        AND lt.tenant_id = $3::uuid
        AND (
            -- the orchestrator's self-mapping row is hidden by default once real child tasks
            -- exist, since orchestration is abstracted away from the user
            COALESCE($4::boolean, FALSE)
            OR dt.task_id != dt.dag_id
            OR NOT EXISTS (
                SELECT 1
                FROM v1_dag_to_task_olap other
                WHERE other.dag_id = dt.dag_id
                    AND other.dag_inserted_at = dt.dag_inserted_at
                    AND other.task_id != other.dag_id
            )
        )
), aggregated_events AS (
    SELECT
        e.tenant_id,
        e.task_id,
        e.task_inserted_at,
        t.dag_id,
        t.dag_inserted_at,
        e.retry_count,
        e.event_type,
        e.durable_invocation_count,
        MIN(e.event_timestamp)::timestamptz AS time_first_seen,
        MAX(e.event_timestamp)::timestamptz AS time_last_seen,
        COUNT(*) AS count,
        MIN(e.id) AS first_id
    FROM v1_task_events_olap e
    JOIN tasks t ON t.task_id = e.task_id AND t.task_inserted_at = e.task_inserted_at
    WHERE
        e.tenant_id = $3::uuid
        AND e.task_inserted_at >= (SELECT MIN(task_inserted_at) FROM tasks)
    GROUP BY e.tenant_id, e.task_id, e.task_inserted_at, t.dag_id, t.dag_inserted_at, e.retry_count, e.event_type, e.durable_invocation_count
)
SELECT
    a.tenant_id,
    a.task_id,
    a.task_inserted_at,
    a.retry_count,
    a.event_type,
    a.durable_invocation_count,
    a.time_first_seen,
    a.time_last_seen,
    a.count,
    e.id,
    e.event_timestamp,
    e.readable_status,
    e.error_message,
    e.output,
    e.external_id AS event_external_id,
    e.worker_id,
    e.additional__event_data,
    e.additional__event_message,
    COALESCE(t.display_name, d.display_name) AS display_name,
    COALESCE(t.external_id, d.external_id) AS task_external_id
FROM aggregated_events a
JOIN v1_task_events_olap e ON (e.task_id, e.task_inserted_at, e.tenant_id, e.id) = (a.task_id, a.task_inserted_at, a.tenant_id, a.first_id)
LEFT JOIN v1_tasks_olap t ON (t.inserted_at, t.id, t.tenant_id) = (e.task_inserted_at, e.task_id, e.tenant_id)
LEFT JOIN v1_dags_olap d ON (d.inserted_at, d.id, d.tenant_id) = (a.dag_inserted_at, a.dag_id, a.tenant_id)
ORDER BY a.time_first_seen DESC, e.event_timestamp DESC
LIMIT $1::bigint
`

type ListTaskEventsForWorkflowRunParams struct {
	Eventlimit                int64       `json:"eventlimit"`
	Workflowrunid             uuid.UUID   `json:"workflowrunid"`
	Tenantid                  uuid.UUID   `json:"tenantid"`
	IncludeOrchestratorEvents pgtype.Bool `json:"includeOrchestratorEvents"`
}

type ListTaskEventsForWorkflowRunRow struct {
	TenantID               uuid.UUID            `json:"tenant_id"`
	TaskID                 int64                `json:"task_id"`
	TaskInsertedAt         pgtype.Timestamptz   `json:"task_inserted_at"`
	RetryCount             int32                `json:"retry_count"`
	EventType              V1EventTypeOlap      `json:"event_type"`
	DurableInvocationCount int32                `json:"durable_invocation_count"`
	TimeFirstSeen          pgtype.Timestamptz   `json:"time_first_seen"`
	TimeLastSeen           pgtype.Timestamptz   `json:"time_last_seen"`
	Count                  int64                `json:"count"`
	ID                     int64                `json:"id"`
	EventTimestamp         pgtype.Timestamptz   `json:"event_timestamp"`
	ReadableStatus         V1ReadableStatusOlap `json:"readable_status"`
	ErrorMessage           pgtype.Text          `json:"error_message"`
	Output                 []byte               `json:"output"`
	EventExternalID        uuid.UUID            `json:"event_external_id"`
	WorkerID               *uuid.UUID           `json:"worker_id"`
	AdditionalEventData    pgtype.Text          `json:"additional__event_data"`
	AdditionalEventMessage pgtype.Text          `json:"additional__event_message"`
	DisplayName            string               `json:"display_name"`
	TaskExternalID         uuid.UUID            `json:"task_external_id"`
}

func (q *Queries) ListTaskEventsForWorkflowRun(ctx context.Context, db DBTX, arg ListTaskEventsForWorkflowRunParams) ([]*ListTaskEventsForWorkflowRunRow, error) {
	rows, err := db.Query(ctx, listTaskEventsForWorkflowRun,
		arg.Eventlimit,
		arg.Workflowrunid,
		arg.Tenantid,
		arg.IncludeOrchestratorEvents,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ListTaskEventsForWorkflowRunRow
	for rows.Next() {
		var i ListTaskEventsForWorkflowRunRow
		if err := rows.Scan(
			&i.TenantID,
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.RetryCount,
			&i.EventType,
			&i.DurableInvocationCount,
			&i.TimeFirstSeen,
			&i.TimeLastSeen,
			&i.Count,
			&i.ID,
			&i.EventTimestamp,
			&i.ReadableStatus,
			&i.ErrorMessage,
			&i.Output,
			&i.EventExternalID,
			&i.WorkerID,
			&i.AdditionalEventData,
			&i.AdditionalEventMessage,
			&i.DisplayName,
			&i.TaskExternalID,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listTasksByDAGIds = `-- name: ListTasksByDAGIds :many
WITH lookups AS MATERIALIZED (
    SELECT
        external_id,
        dag_id,
        inserted_at
    FROM
        v1_lookup_table_olap
    WHERE
        external_id = ANY($1::uuid[])
        AND tenant_id = $2::uuid
)
SELECT
    DISTINCT ON (t.external_id)
    dt.dag_id, dt.dag_inserted_at, dt.task_id, dt.task_inserted_at,
    lt.external_id AS dag_external_id
FROM
    lookups lt
JOIN
    v1_dag_to_task_olap dt ON (lt.dag_id, lt.inserted_at)= (dt.dag_id, dt.dag_inserted_at)
JOIN
    v1_tasks_olap t ON (t.id, t.inserted_at) = (dt.task_id, dt.task_inserted_at)
WHERE
    dt.dag_inserted_at >= (SELECT MIN(inserted_at) FROM lookups)
    AND t.inserted_at >= (SELECT MIN(inserted_at) FROM lookups)
ORDER BY
    t.external_id, t.inserted_at DESC
`

type ListTasksByDAGIdsParams struct {
	Dagids   []uuid.UUID `json:"dagids"`
	Tenantid uuid.UUID   `json:"tenantid"`
}

type ListTasksByDAGIdsRow struct {
	DagID          int64              `json:"dag_id"`
	DagInsertedAt  pgtype.Timestamptz `json:"dag_inserted_at"`
	TaskID         int64              `json:"task_id"`
	TaskInsertedAt pgtype.Timestamptz `json:"task_inserted_at"`
	DagExternalID  uuid.UUID          `json:"dag_external_id"`
}

func (q *Queries) ListTasksByDAGIds(ctx context.Context, db DBTX, arg ListTasksByDAGIdsParams) ([]*ListTasksByDAGIdsRow, error) {
	rows, err := db.Query(ctx, listTasksByDAGIds, arg.Dagids, arg.Tenantid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ListTasksByDAGIdsRow
	for rows.Next() {
		var i ListTasksByDAGIdsRow
		if err := rows.Scan(
			&i.DagID,
			&i.DagInsertedAt,
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.DagExternalID,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listTasksByExternalIds = `-- name: ListTasksByExternalIds :many
SELECT
    tenant_id,
    task_id,
    inserted_at
FROM
    v1_lookup_table_olap
WHERE
    external_id = ANY($1::uuid[])
    AND tenant_id = $2::uuid
`

type ListTasksByExternalIdsParams struct {
	Externalids []uuid.UUID `json:"externalids"`
	Tenantid    uuid.UUID   `json:"tenantid"`
}

type ListTasksByExternalIdsRow struct {
	TenantID   uuid.UUID          `json:"tenant_id"`
	TaskID     pgtype.Int8        `json:"task_id"`
	InsertedAt pgtype.Timestamptz `json:"inserted_at"`
}

func (q *Queries) ListTasksByExternalIds(ctx context.Context, db DBTX, arg ListTasksByExternalIdsParams) ([]*ListTasksByExternalIdsRow, error) {
	rows, err := db.Query(ctx, listTasksByExternalIds, arg.Externalids, arg.Tenantid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ListTasksByExternalIdsRow
	for rows.Next() {
		var i ListTasksByExternalIdsRow
		if err := rows.Scan(&i.TenantID, &i.TaskID, &i.InsertedAt); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listWorkflowRunDisplayNames = `-- name: ListWorkflowRunDisplayNames :many
SELECT
    lt.external_id,
    COALESCE(t.display_name, d.display_name) AS display_name,
    COALESCE(t.inserted_at, d.inserted_at) AS inserted_at
FROM v1_lookup_table_olap lt
LEFT JOIN v1_dags_olap d ON (lt.dag_id, lt.inserted_at) = (d.id, d.inserted_at)
LEFT JOIN v1_tasks_olap t ON (lt.task_id, lt.inserted_at) = (t.id, t.inserted_at)
WHERE
    lt.external_id = ANY($1::uuid[])
    AND lt.tenant_id = $2::uuid
LIMIT 10000
`

type ListWorkflowRunDisplayNamesParams struct {
	Externalids []uuid.UUID `json:"externalids"`
	Tenantid    uuid.UUID   `json:"tenantid"`
}

type ListWorkflowRunDisplayNamesRow struct {
	ExternalID  uuid.UUID          `json:"external_id"`
	DisplayName string             `json:"display_name"`
	InsertedAt  pgtype.Timestamptz `json:"inserted_at"`
}

func (q *Queries) ListWorkflowRunDisplayNames(ctx context.Context, db DBTX, arg ListWorkflowRunDisplayNamesParams) ([]*ListWorkflowRunDisplayNamesRow, error) {
	rows, err := db.Query(ctx, listWorkflowRunDisplayNames, arg.Externalids, arg.Tenantid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ListWorkflowRunDisplayNamesRow
	for rows.Next() {
		var i ListWorkflowRunDisplayNamesRow
		if err := rows.Scan(&i.ExternalID, &i.DisplayName, &i.InsertedAt); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listWorkflowRunExternalIds = `-- name: ListWorkflowRunExternalIds :many
SELECT external_id
FROM v1_runs_olap
WHERE
    tenant_id = $1::UUID
    AND inserted_at > $2::TIMESTAMPTZ
    AND (
        $3::TIMESTAMPTZ IS NULL
        OR inserted_at <= $3::TIMESTAMPTZ
    )
    AND readable_status = ANY(CAST($4::TEXT[] AS v1_readable_status_olap[]))
    AND (
        $5::text[] IS NULL
        OR $6::text[] IS NULL
        OR EXISTS (
            SELECT 1 FROM jsonb_each_text(additional_metadata) kv
            JOIN LATERAL (
                SELECT unnest($5::text[]) AS k,
                    unnest($6::text[]) AS v
            ) AS u ON kv.key = u.k AND kv.value = u.v
        )
    )
    AND (
        $7::UUID[] IS NULL OR workflow_id = ANY($7::UUID[])
    )
`

type ListWorkflowRunExternalIdsParams struct {
	Tenantid             uuid.UUID          `json:"tenantid"`
	Since                pgtype.Timestamptz `json:"since"`
	Until                pgtype.Timestamptz `json:"until"`
	Statuses             []string           `json:"statuses"`
	AdditionalMetaKeys   []string           `json:"additionalMetaKeys"`
	AdditionalMetaValues []string           `json:"additionalMetaValues"`
	WorkflowIds          []uuid.UUID        `json:"workflowIds"`
}

func (q *Queries) ListWorkflowRunExternalIds(ctx context.Context, db DBTX, arg ListWorkflowRunExternalIdsParams) ([]uuid.UUID, error) {
	rows, err := db.Query(ctx, listWorkflowRunExternalIds,
		arg.Tenantid,
		arg.Since,
		arg.Until,
		arg.Statuses,
		arg.AdditionalMetaKeys,
		arg.AdditionalMetaValues,
		arg.WorkflowIds,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []uuid.UUID
	for rows.Next() {
		var external_id uuid.UUID
		if err := rows.Scan(&external_id); err != nil {
			return nil, err
		}
		items = append(items, external_id)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const listYesterdayRunCountsByStatus = `-- name: ListYesterdayRunCountsByStatus :many
SELECT readable_status, COUNT(*)
FROM v1_runs_olap
WHERE inserted_at::DATE = (NOW() - INTERVAL '1 day')::DATE
GROUP BY readable_status
`

type ListYesterdayRunCountsByStatusRow struct {
	ReadableStatus V1ReadableStatusOlap `json:"readable_status"`
	Count          int64                `json:"count"`
}

func (q *Queries) ListYesterdayRunCountsByStatus(ctx context.Context, db DBTX) ([]*ListYesterdayRunCountsByStatusRow, error) {
	rows, err := db.Query(ctx, listYesterdayRunCountsByStatus)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ListYesterdayRunCountsByStatusRow
	for rows.Next() {
		var i ListYesterdayRunCountsByStatusRow
		if err := rows.Scan(&i.ReadableStatus, &i.Count); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const markOLAPCutoverJobAsCompleted = `-- name: MarkOLAPCutoverJobAsCompleted :exec
UPDATE v1_payloads_olap_cutover_job_offset
SET is_completed = TRUE
WHERE key = $1::DATE
`

func (q *Queries) MarkOLAPCutoverJobAsCompleted(ctx context.Context, db DBTX, key pgtype.Date) error {
	_, err := db.Exec(ctx, markOLAPCutoverJobAsCompleted, key)
	return err
}

const offloadPayloads = `-- name: OffloadPayloads :exec
WITH inputs AS (
    SELECT
        UNNEST($1::UUID[]) AS external_id,
        UNNEST($2::UUID[]) AS tenant_id,
        UNNEST($3::TEXT[]) AS external_location_key
)

UPDATE v1_payloads_olap
SET
    location = 'EXTERNAL',
    external_location_key = i.external_location_key,
    inline_content = NULL,
    updated_at = NOW()
FROM inputs i
WHERE
    (v1_payloads_olap.tenant_id, v1_payloads_olap.external_id) = (i.tenant_id, i.external_id)
    AND v1_payloads_olap.location = 'INLINE'
    AND v1_payloads_olap.external_location_key IS NULL
`

type OffloadPayloadsParams struct {
	Externalids          []uuid.UUID `json:"externalids"`
	Tenantids            []uuid.UUID `json:"tenantids"`
	Externallocationkeys []string    `json:"externallocationkeys"`
}

func (q *Queries) OffloadPayloads(ctx context.Context, db DBTX, arg OffloadPayloadsParams) error {
	_, err := db.Exec(ctx, offloadPayloads, arg.Externalids, arg.Tenantids, arg.Externallocationkeys)
	return err
}

const populateDAGMetadata = `-- name: PopulateDAGMetadata :many
WITH input AS (
    SELECT
        UNNEST($2::bigint[]) AS id,
        UNNEST($3::timestamptz[]) AS inserted_at
), runs AS (
    SELECT
        d.id AS dag_id,
        r.id AS run_id,
        r.tenant_id,
        r.inserted_at,
        r.external_id,
        r.readable_status,
        r.kind,
        r.workflow_id,
        d.display_name,
        CASE
            WHEN $1::BOOLEAN THEN d.input
            ELSE '{}'::JSONB
        END::JSONB AS input,
        d.additional_metadata,
        d.workflow_version_id,
        d.parent_task_external_id,
        d.idempotency_key
    FROM input i
    JOIN v1_runs_olap r ON (i.id, i.inserted_at) = (r.id, r.inserted_at)
    JOIN v1_dags_olap d ON (r.id, r.inserted_at) = (d.id, d.inserted_at)
    WHERE r.tenant_id = $4::uuid AND r.kind = 'DAG'
), relevant_events AS (
    SELECT r.run_id, e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count, e.is_dag_orchestrator
    FROM runs r
    JOIN v1_dag_to_task_olap dt ON (r.dag_id, r.inserted_at) = (dt.dag_id, dt.dag_inserted_at)
    JOIN v1_task_events_olap e ON (e.task_id, e.task_inserted_at) = (dt.task_id, dt.task_inserted_at)
    WHERE
        e.tenant_id = $4::uuid
        AND dt.dag_inserted_at >= (SELECT MIN(inserted_at) FROM input)
        AND e.task_inserted_at >= (SELECT MIN(inserted_at) FROM input)
), max_retry_count AS (
    SELECT run_id, MAX(retry_count) AS max_retry_count
    FROM relevant_events
    GROUP BY run_id
), metadata AS (
    SELECT
        e.run_id,
        MIN(e.inserted_at)::timestamptz AS created_at,
        MIN(e.inserted_at) FILTER (WHERE e.readable_status = 'RUNNING')::timestamptz AS started_at,
        MAX(e.inserted_at) FILTER (WHERE e.readable_status IN ('COMPLETED', 'CANCELLED', 'FAILED'))::timestamptz AS finished_at
    FROM
        relevant_events e
    JOIN max_retry_count mrc ON (e.run_id, e.retry_count) = (mrc.run_id, mrc.max_retry_count)
    GROUP BY e.run_id
), error_message AS (
    SELECT
        DISTINCT ON (e.run_id) e.run_id::bigint,
        e.error_message
    FROM
        relevant_events e
    WHERE
        e.readable_status = 'FAILED'
    ORDER BY
        e.run_id, e.retry_count DESC
), task_output AS (
    SELECT
        run_id,
        output,
        external_id,
        inserted_at
    FROM
        relevant_events
    WHERE
        event_type = 'FINISHED'
)

SELECT
    r.dag_id, r.run_id, r.tenant_id, r.inserted_at, r.external_id, r.readable_status, r.kind, r.workflow_id, r.display_name, r.input, r.additional_metadata, r.workflow_version_id, r.parent_task_external_id, r.idempotency_key,
    m.created_at,
    m.started_at,
    m.finished_at,
    e.error_message,
    CASE
        WHEN $1::BOOLEAN THEN o.output::JSONB
        ELSE '{}'::JSONB
    END::JSONB AS output,
    o.external_id AS output_event_external_id,
    o.inserted_at AS output_event_inserted_at,
    COALESCE(mrc.max_retry_count, 0)::int as retry_count
FROM runs r
LEFT JOIN metadata m ON r.run_id = m.run_id
LEFT JOIN error_message e ON r.run_id = e.run_id
LEFT JOIN task_output o ON r.run_id = o.run_id
LEFT JOIN max_retry_count mrc ON r.run_id = mrc.run_id
ORDER BY r.inserted_at DESC, r.run_id DESC
`

type PopulateDAGMetadataParams struct {
	Includepayloads bool                 `json:"includepayloads"`
	Ids             []int64              `json:"ids"`
	Insertedats     []pgtype.Timestamptz `json:"insertedats"`
	Tenantid        uuid.UUID            `json:"tenantid"`
}

type PopulateDAGMetadataRow struct {
	DagID                 int64                `json:"dag_id"`
	RunID                 int64                `json:"run_id"`
	TenantID              uuid.UUID            `json:"tenant_id"`
	InsertedAt            pgtype.Timestamptz   `json:"inserted_at"`
	ExternalID            uuid.UUID            `json:"external_id"`
	ReadableStatus        V1ReadableStatusOlap `json:"readable_status"`
	Kind                  V1RunKind            `json:"kind"`
	WorkflowID            uuid.UUID            `json:"workflow_id"`
	DisplayName           string               `json:"display_name"`
	Input                 []byte               `json:"input"`
	AdditionalMetadata    []byte               `json:"additional_metadata"`
	WorkflowVersionID     uuid.UUID            `json:"workflow_version_id"`
	ParentTaskExternalID  *uuid.UUID           `json:"parent_task_external_id"`
	IdempotencyKey        pgtype.Text          `json:"idempotency_key"`
	CreatedAt             pgtype.Timestamptz   `json:"created_at"`
	StartedAt             pgtype.Timestamptz   `json:"started_at"`
	FinishedAt            pgtype.Timestamptz   `json:"finished_at"`
	ErrorMessage          pgtype.Text          `json:"error_message"`
	Output                []byte               `json:"output"`
	OutputEventExternalID *uuid.UUID           `json:"output_event_external_id"`
	OutputEventInsertedAt pgtype.Timestamptz   `json:"output_event_inserted_at"`
	RetryCount            int32                `json:"retry_count"`
}

func (q *Queries) PopulateDAGMetadata(ctx context.Context, db DBTX, arg PopulateDAGMetadataParams) ([]*PopulateDAGMetadataRow, error) {
	rows, err := db.Query(ctx, populateDAGMetadata,
		arg.Includepayloads,
		arg.Ids,
		arg.Insertedats,
		arg.Tenantid,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*PopulateDAGMetadataRow
	for rows.Next() {
		var i PopulateDAGMetadataRow
		if err := rows.Scan(
			&i.DagID,
			&i.RunID,
			&i.TenantID,
			&i.InsertedAt,
			&i.ExternalID,
			&i.ReadableStatus,
			&i.Kind,
			&i.WorkflowID,
			&i.DisplayName,
			&i.Input,
			&i.AdditionalMetadata,
			&i.WorkflowVersionID,
			&i.ParentTaskExternalID,
			&i.IdempotencyKey,
			&i.CreatedAt,
			&i.StartedAt,
			&i.FinishedAt,
			&i.ErrorMessage,
			&i.Output,
			&i.OutputEventExternalID,
			&i.OutputEventInsertedAt,
			&i.RetryCount,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const populateEventData = `-- name: PopulateEventData :many
SELECT
    elt.external_id,
    COUNT(*) FILTER (WHERE r.readable_status = 'QUEUED') AS queued_count,
    COUNT(*) FILTER (WHERE r.readable_status = 'RUNNING') AS running_count,
    COUNT(*) FILTER (WHERE r.readable_status = 'COMPLETED') AS completed_count,
    COUNT(*) FILTER (WHERE r.readable_status = 'CANCELLED') AS cancelled_count,
    COUNT(*) FILTER (WHERE r.readable_status = 'FAILED') AS failed_count,
    JSON_AGG(JSON_BUILD_OBJECT('run_external_id', r.external_id, 'filter_id', etr.filter_id)) FILTER (WHERE r.external_id IS NOT NULL)::JSONB AS triggered_runs
FROM v1_event_lookup_table_olap elt
JOIN v1_events_olap e ON (elt.tenant_id, elt.event_seen_at, elt.event_id) = (e.tenant_id, e.seen_at, e.id)
JOIN v1_event_to_run_olap etr ON (e.seen_at, e.id) = (etr.event_seen_at, etr.event_id)
JOIN v1_runs_olap r ON (etr.run_inserted_at, etr.run_id) = (r.inserted_at, r.id)
WHERE
    elt.external_id = ANY($1::uuid[])
    AND elt.tenant_id = $2::uuid
    AND r.inserted_at >= $3::timestamptz
GROUP BY elt.external_id
`

type PopulateEventDataParams struct {
	Eventexternalids []uuid.UUID        `json:"eventexternalids"`
	Tenantid         uuid.UUID          `json:"tenantid"`
	Minseenat        pgtype.Timestamptz `json:"minseenat"`
}

type PopulateEventDataRow struct {
	ExternalID     uuid.UUID `json:"external_id"`
	QueuedCount    int64     `json:"queued_count"`
	RunningCount   int64     `json:"running_count"`
	CompletedCount int64     `json:"completed_count"`
	CancelledCount int64     `json:"cancelled_count"`
	FailedCount    int64     `json:"failed_count"`
	TriggeredRuns  []byte    `json:"triggered_runs"`
}

func (q *Queries) PopulateEventData(ctx context.Context, db DBTX, arg PopulateEventDataParams) ([]*PopulateEventDataRow, error) {
	rows, err := db.Query(ctx, populateEventData, arg.Eventexternalids, arg.Tenantid, arg.Minseenat)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*PopulateEventDataRow
	for rows.Next() {
		var i PopulateEventDataRow
		if err := rows.Scan(
			&i.ExternalID,
			&i.QueuedCount,
			&i.RunningCount,
			&i.CompletedCount,
			&i.CancelledCount,
			&i.FailedCount,
			&i.TriggeredRuns,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const populateSingleTaskRunData = `-- name: PopulateSingleTaskRunData :one
WITH selected_retry_count AS (
    SELECT
        CASE
            WHEN $4::int IS NOT NULL THEN $4::int
            ELSE MAX(retry_count)::int
        END AS retry_count
    FROM
        v1_task_events_olap
    WHERE
        tenant_id = $1::uuid
        AND task_id = $2::bigint
        AND task_inserted_at = $3::timestamptz
    LIMIT 1
), relevant_events AS (
    SELECT
        tenant_id, id, inserted_at, external_id, task_id, task_inserted_at, event_type, workflow_id, event_timestamp, readable_status, retry_count, error_message, output, worker_id, additional__event_data, additional__event_message, durable_invocation_count, is_dag_orchestrator
    FROM
        v1_task_events_olap
    WHERE
        tenant_id = $1::uuid
        AND task_id = $2::bigint
        AND task_inserted_at = $3::timestamptz
        AND retry_count = (SELECT retry_count FROM selected_retry_count)
), finished_at AS (
    SELECT
        MAX(event_timestamp) AS finished_at
    FROM
        relevant_events
    WHERE
        readable_status = ANY(ARRAY['COMPLETED', 'FAILED', 'CANCELLED']::v1_readable_status_olap[])
), started_at AS (
    SELECT
        MAX(event_timestamp) AS started_at
    FROM
        relevant_events
    WHERE
        event_type = 'STARTED'
), queued_at AS (
    SELECT
        MAX(event_timestamp) AS queued_at
    FROM
        relevant_events
    WHERE
        event_type = 'QUEUED'
), task_output AS (
    SELECT
        external_id,
        output,
        inserted_at
    FROM
        relevant_events
    WHERE
        event_type = 'FINISHED'
    LIMIT 1
), status AS (
    SELECT
        readable_status
    FROM
        relevant_events
    ORDER BY
        CASE
            WHEN readable_status IN ('COMPLETED', 'FAILED', 'CANCELLED') THEN 1
            ELSE 0
        END DESC,
        event_timestamp DESC
    LIMIT 1
), error_message AS (
    SELECT
        error_message
    FROM
        relevant_events
    WHERE
        readable_status = 'FAILED'
    ORDER BY
        event_timestamp DESC
    LIMIT 1
), spawned_children AS (
    SELECT COUNT(*) AS spawned_children
    FROM v1_runs_olap
    WHERE parent_task_external_id = (
        SELECT external_id
        FROM v1_tasks_olap
        WHERE
            tenant_id = $1::uuid
            AND id = $2::bigint
            AND inserted_at = $3::timestamptz
        LIMIT 1
    )
)
SELECT
    t.tenant_id, t.id, t.inserted_at, t.external_id, t.queue, t.action_id, t.step_id, t.workflow_id, t.workflow_version_id, t.workflow_run_id, t.schedule_timeout, t.step_timeout, t.priority, t.sticky, t.desired_worker_id, t.display_name, t.input, t.additional_metadata, t.readable_status, t.latest_retry_count, t.latest_worker_id, t.dag_id, t.dag_inserted_at, t.parent_task_external_id, t.is_durable, t.idempotency_key,
    (t.dag_id IS NULL)::BOOLEAN AS is_standalone,
    st.readable_status::v1_readable_status_olap as status,
    f.finished_at::timestamptz as finished_at,
    s.started_at::timestamptz as started_at,
    q.queued_at::timestamptz as queued_at,
    o.external_id AS output_event_external_id,
    o.inserted_at AS output_event_inserted_at,
    o.output as output,
    e.error_message as error_message,
    sc.spawned_children,
    (SELECT retry_count FROM selected_retry_count) as retry_count
FROM
    v1_tasks_olap t
LEFT JOIN
    finished_at f ON true
LEFT JOIN
    started_at s ON true
LEFT JOIN
    queued_at q ON true
LEFT JOIN
    task_output o ON true
LEFT JOIN
    status st ON true
LEFT JOIN
    error_message e ON true
LEFT JOIN
    spawned_children sc ON true
WHERE
    (t.tenant_id, t.id, t.inserted_at) = ($1::uuid, $2::bigint, $3::timestamptz)
`

type PopulateSingleTaskRunDataParams struct {
	Tenantid       uuid.UUID          `json:"tenantid"`
	Taskid         int64              `json:"taskid"`
	Taskinsertedat pgtype.Timestamptz `json:"taskinsertedat"`
	RetryCount     pgtype.Int4        `json:"retry_count"`
}

type PopulateSingleTaskRunDataRow struct {
	TenantID              uuid.UUID            `json:"tenant_id"`
	ID                    int64                `json:"id"`
	InsertedAt            pgtype.Timestamptz   `json:"inserted_at"`
	ExternalID            uuid.UUID            `json:"external_id"`
	Queue                 string               `json:"queue"`
	ActionID              string               `json:"action_id"`
	StepID                uuid.UUID            `json:"step_id"`
	WorkflowID            uuid.UUID            `json:"workflow_id"`
	WorkflowVersionID     uuid.UUID            `json:"workflow_version_id"`
	WorkflowRunID         uuid.UUID            `json:"workflow_run_id"`
	ScheduleTimeout       string               `json:"schedule_timeout"`
	StepTimeout           pgtype.Text          `json:"step_timeout"`
	Priority              pgtype.Int4          `json:"priority"`
	Sticky                V1StickyStrategyOlap `json:"sticky"`
	DesiredWorkerID       *uuid.UUID           `json:"desired_worker_id"`
	DisplayName           string               `json:"display_name"`
	Input                 []byte               `json:"input"`
	AdditionalMetadata    []byte               `json:"additional_metadata"`
	ReadableStatus        V1ReadableStatusOlap `json:"readable_status"`
	LatestRetryCount      int32                `json:"latest_retry_count"`
	LatestWorkerID        *uuid.UUID           `json:"latest_worker_id"`
	DagID                 pgtype.Int8          `json:"dag_id"`
	DagInsertedAt         pgtype.Timestamptz   `json:"dag_inserted_at"`
	ParentTaskExternalID  *uuid.UUID           `json:"parent_task_external_id"`
	IsDurable             bool                 `json:"is_durable"`
	IdempotencyKey        pgtype.Text          `json:"idempotency_key"`
	IsStandalone          bool                 `json:"is_standalone"`
	Status                V1ReadableStatusOlap `json:"status"`
	FinishedAt            pgtype.Timestamptz   `json:"finished_at"`
	StartedAt             pgtype.Timestamptz   `json:"started_at"`
	QueuedAt              pgtype.Timestamptz   `json:"queued_at"`
	OutputEventExternalID *uuid.UUID           `json:"output_event_external_id"`
	OutputEventInsertedAt pgtype.Timestamptz   `json:"output_event_inserted_at"`
	Output                []byte               `json:"output"`
	ErrorMessage          pgtype.Text          `json:"error_message"`
	SpawnedChildren       pgtype.Int8          `json:"spawned_children"`
	RetryCount            int32                `json:"retry_count"`
}

func (q *Queries) PopulateSingleTaskRunData(ctx context.Context, db DBTX, arg PopulateSingleTaskRunDataParams) (*PopulateSingleTaskRunDataRow, error) {
	row := db.QueryRow(ctx, populateSingleTaskRunData,
		arg.Tenantid,
		arg.Taskid,
		arg.Taskinsertedat,
		arg.RetryCount,
	)
	var i PopulateSingleTaskRunDataRow
	err := row.Scan(
		&i.TenantID,
		&i.ID,
		&i.InsertedAt,
		&i.ExternalID,
		&i.Queue,
		&i.ActionID,
		&i.StepID,
		&i.WorkflowID,
		&i.WorkflowVersionID,
		&i.WorkflowRunID,
		&i.ScheduleTimeout,
		&i.StepTimeout,
		&i.Priority,
		&i.Sticky,
		&i.DesiredWorkerID,
		&i.DisplayName,
		&i.Input,
		&i.AdditionalMetadata,
		&i.ReadableStatus,
		&i.LatestRetryCount,
		&i.LatestWorkerID,
		&i.DagID,
		&i.DagInsertedAt,
		&i.ParentTaskExternalID,
		&i.IsDurable,
		&i.IdempotencyKey,
		&i.IsStandalone,
		&i.Status,
		&i.FinishedAt,
		&i.StartedAt,
		&i.QueuedAt,
		&i.OutputEventExternalID,
		&i.OutputEventInsertedAt,
		&i.Output,
		&i.ErrorMessage,
		&i.SpawnedChildren,
		&i.RetryCount,
	)
	return &i, err
}

const populateTaskRunData = `-- name: PopulateTaskRunData :many
WITH input AS (
    SELECT
        UNNEST($2::bigint[]) AS id,
        UNNEST($3::timestamptz[]) AS inserted_at
), tasks AS (
    SELECT
        DISTINCT ON(t.tenant_id, t.id, t.inserted_at)
        t.tenant_id,
        t.id,
        t.inserted_at,
        t.queue,
        t.action_id,
        t.step_id,
        t.workflow_id,
        t.workflow_version_id,
        t.schedule_timeout,
        t.step_timeout,
        t.priority,
        t.sticky,
        t.desired_worker_id,
        t.external_id,
        t.display_name,
        t.input,
        t.additional_metadata,
        t.readable_status,
        t.parent_task_external_id,
        t.workflow_run_id,
        t.latest_retry_count,
        t.dag_id,
        t.is_durable,
        t.idempotency_key
    FROM
        v1_tasks_olap t
    JOIN
        input i ON i.id = t.id AND i.inserted_at = t.inserted_at
    WHERE
        t.tenant_id = $4::uuid
), relevant_events AS (
    SELECT
        e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count, e.is_dag_orchestrator
    FROM
        v1_task_events_olap e
    JOIN
        tasks t ON t.id = e.task_id AND t.tenant_id = e.tenant_id AND t.inserted_at = e.task_inserted_at
    WHERE
        e.task_inserted_at >= (SELECT MIN(inserted_at) FROM input)
), max_retry_counts AS (
    SELECT
        e.tenant_id,
        e.task_id,
        e.task_inserted_at,
        MAX(e.retry_count) AS max_retry_count
    FROM
        relevant_events e
    GROUP BY
        e.tenant_id, e.task_id, e.task_inserted_at
), queued_ats AS (
    SELECT
        e.task_id::bigint,
        MAX(e.event_timestamp) AS queued_at
    FROM
        relevant_events e
    JOIN
        max_retry_counts mrc ON
            e.tenant_id = mrc.tenant_id
            AND e.task_id = mrc.task_id
            AND e.task_inserted_at = mrc.task_inserted_at
            AND e.retry_count = mrc.max_retry_count
    WHERE
        e.event_type = 'QUEUED'
    GROUP BY e.task_id
), started_ats AS (
    SELECT
        e.task_id::bigint,
        MAX(e.event_timestamp) AS started_at
    FROM
        relevant_events e
    JOIN
        max_retry_counts mrc ON
            e.tenant_id = mrc.tenant_id
            AND e.task_id = mrc.task_id
            AND e.task_inserted_at = mrc.task_inserted_at
            AND e.retry_count = mrc.max_retry_count
    WHERE
        e.event_type = 'STARTED'
    GROUP BY e.task_id
), finished_ats AS (
    SELECT
        e.task_id::bigint,
        MAX(e.event_timestamp) AS finished_at
    FROM
        relevant_events e
    JOIN
        max_retry_counts mrc ON
            e.tenant_id = mrc.tenant_id
            AND e.task_id = mrc.task_id
            AND e.task_inserted_at = mrc.task_inserted_at
            AND e.retry_count = mrc.max_retry_count
    WHERE
        e.readable_status = ANY(ARRAY['COMPLETED', 'FAILED', 'CANCELLED']::v1_readable_status_olap[])
    GROUP BY e.task_id
), error_message AS (
    SELECT
        DISTINCT ON (e.task_id) e.task_id::bigint,
        e.error_message
    FROM
        relevant_events e
    JOIN
        max_retry_counts mrc ON
            e.tenant_id = mrc.tenant_id
            AND e.task_id = mrc.task_id
            AND e.task_inserted_at = mrc.task_inserted_at
            AND e.retry_count = mrc.max_retry_count
    WHERE
        e.readable_status = 'FAILED'
    ORDER BY
        e.task_id, e.retry_count DESC
), task_output AS (
    SELECT
        DISTINCT ON (task_id)
        task_id,
        output,
        external_id AS output_event_external_id,
        inserted_at AS output_event_inserted_at
    FROM
        relevant_events
    WHERE
        readable_status = 'COMPLETED'
        AND event_type = 'FINISHED'
    ORDER BY
        task_id, event_timestamp DESC
)
SELECT
    t.tenant_id,
    t.id,
    t.inserted_at,
    t.external_id,
    t.queue,
    t.action_id,
    t.step_id,
    t.workflow_id,
    t.workflow_version_id,
    t.schedule_timeout,
    t.step_timeout,
    t.priority,
    t.sticky,
    t.display_name,
    t.additional_metadata,
    t.parent_task_external_id,
    CASE
        WHEN $1::BOOLEAN THEN t.input
        ELSE '{}'::JSONB
    END::JSONB AS input,
    t.readable_status::v1_readable_status_olap as status,
    t.workflow_run_id,
    (t.dag_id IS NULL)::BOOLEAN AS is_standalone,
    f.finished_at::timestamptz as finished_at,
    s.started_at::timestamptz as started_at,
    q.queued_at::timestamptz as queued_at,
    e.error_message as error_message,
    COALESCE(t.latest_retry_count, 0)::int as retry_count,
    CASE
        WHEN $1::BOOLEAN THEN o.output::JSONB
        ELSE '{}'::JSONB
    END::JSONB as output,
    o.output_event_external_id AS output_event_external_id,
    o.output_event_inserted_at AS output_event_inserted_at,
    COALESCE(t.is_durable, FALSE) AS is_durable,
    t.idempotency_key
FROM
    tasks t
LEFT JOIN
    finished_ats f ON f.task_id = t.id
LEFT JOIN
    started_ats s ON s.task_id = t.id
LEFT JOIN
    queued_ats q ON q.task_id = t.id
LEFT JOIN
    error_message e ON e.task_id = t.id
LEFT JOIN
    task_output o ON o.task_id = t.id
ORDER BY t.inserted_at DESC, t.id DESC
`

type PopulateTaskRunDataParams struct {
	Includepayloads bool                 `json:"includepayloads"`
	Taskids         []int64              `json:"taskids"`
	Taskinsertedats []pgtype.Timestamptz `json:"taskinsertedats"`
	Tenantid        uuid.UUID            `json:"tenantid"`
}

type PopulateTaskRunDataRow struct {
	TenantID              uuid.UUID            `json:"tenant_id"`
	ID                    int64                `json:"id"`
	InsertedAt            pgtype.Timestamptz   `json:"inserted_at"`
	ExternalID            uuid.UUID            `json:"external_id"`
	Queue                 string               `json:"queue"`
	ActionID              string               `json:"action_id"`
	StepID                uuid.UUID            `json:"step_id"`
	WorkflowID            uuid.UUID            `json:"workflow_id"`
	WorkflowVersionID     uuid.UUID            `json:"workflow_version_id"`
	ScheduleTimeout       string               `json:"schedule_timeout"`
	StepTimeout           pgtype.Text          `json:"step_timeout"`
	Priority              pgtype.Int4          `json:"priority"`
	Sticky                V1StickyStrategyOlap `json:"sticky"`
	DisplayName           string               `json:"display_name"`
	AdditionalMetadata    []byte               `json:"additional_metadata"`
	ParentTaskExternalID  *uuid.UUID           `json:"parent_task_external_id"`
	Input                 []byte               `json:"input"`
	Status                V1ReadableStatusOlap `json:"status"`
	WorkflowRunID         uuid.UUID            `json:"workflow_run_id"`
	IsStandalone          bool                 `json:"is_standalone"`
	FinishedAt            pgtype.Timestamptz   `json:"finished_at"`
	StartedAt             pgtype.Timestamptz   `json:"started_at"`
	QueuedAt              pgtype.Timestamptz   `json:"queued_at"`
	ErrorMessage          pgtype.Text          `json:"error_message"`
	RetryCount            int32                `json:"retry_count"`
	Output                []byte               `json:"output"`
	OutputEventExternalID *uuid.UUID           `json:"output_event_external_id"`
	OutputEventInsertedAt pgtype.Timestamptz   `json:"output_event_inserted_at"`
	IsDurable             bool                 `json:"is_durable"`
	IdempotencyKey        pgtype.Text          `json:"idempotency_key"`
}

func (q *Queries) PopulateTaskRunData(ctx context.Context, db DBTX, arg PopulateTaskRunDataParams) ([]*PopulateTaskRunDataRow, error) {
	rows, err := db.Query(ctx, populateTaskRunData,
		arg.Includepayloads,
		arg.Taskids,
		arg.Taskinsertedats,
		arg.Tenantid,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*PopulateTaskRunDataRow
	for rows.Next() {
		var i PopulateTaskRunDataRow
		if err := rows.Scan(
			&i.TenantID,
			&i.ID,
			&i.InsertedAt,
			&i.ExternalID,
			&i.Queue,
			&i.ActionID,
			&i.StepID,
			&i.WorkflowID,
			&i.WorkflowVersionID,
			&i.ScheduleTimeout,
			&i.StepTimeout,
			&i.Priority,
			&i.Sticky,
			&i.DisplayName,
			&i.AdditionalMetadata,
			&i.ParentTaskExternalID,
			&i.Input,
			&i.Status,
			&i.WorkflowRunID,
			&i.IsStandalone,
			&i.FinishedAt,
			&i.StartedAt,
			&i.QueuedAt,
			&i.ErrorMessage,
			&i.RetryCount,
			&i.Output,
			&i.OutputEventExternalID,
			&i.OutputEventInsertedAt,
			&i.IsDurable,
			&i.IdempotencyKey,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const putPayloads = `-- name: PutPayloads :exec
WITH inputs AS (
    SELECT
        UNNEST($1::UUID[]) AS external_id,
        UNNEST($2::TIMESTAMPTZ[]) AS inserted_at,
        UNNEST($3::JSONB[]) AS payload,
        UNNEST($4::UUID[]) AS tenant_id,
        UNNEST(CAST($5::TEXT[] AS v1_payload_location_olap[])) AS location,
        UNNEST($6::TEXT[]) AS external_location_key
)

INSERT INTO v1_payloads_olap (
    tenant_id,
    external_id,
    inserted_at,
    location,
    external_location_key,
    inline_content
)

SELECT
    i.tenant_id,
    i.external_id,
    i.inserted_at,
    i.location,
    CASE
        WHEN i.location = 'EXTERNAL' THEN i.external_location_key
        ELSE NULL
    END,
    CASE
        WHEN i.location = 'INLINE' THEN i.payload
        ELSE NULL
    END AS inline_content
FROM inputs i
ON CONFLICT DO NOTHING
`

type PutPayloadsParams struct {
	Externalids          []uuid.UUID          `json:"externalids"`
	Insertedats          []pgtype.Timestamptz `json:"insertedats"`
	Payloads             [][]byte             `json:"payloads"`
	Tenantids            []uuid.UUID          `json:"tenantids"`
	Locations            []string             `json:"locations"`
	Externallocationkeys []string             `json:"externallocationkeys"`
}

func (q *Queries) PutPayloads(ctx context.Context, db DBTX, arg PutPayloadsParams) error {
	_, err := db.Exec(ctx, putPayloads,
		arg.Externalids,
		arg.Insertedats,
		arg.Payloads,
		arg.Tenantids,
		arg.Locations,
		arg.Externallocationkeys,
	)
	return err
}

const readDAGByExternalID = `-- name: ReadDAGByExternalID :one
WITH lookup_task AS (
    SELECT
        tenant_id,
        dag_id,
        inserted_at
    FROM
        v1_lookup_table_olap
    WHERE
        external_id = $1::uuid
)
SELECT
    d.id, d.inserted_at, d.tenant_id, d.external_id, d.display_name, d.workflow_id, d.workflow_version_id, d.readable_status, d.input, d.additional_metadata, d.parent_task_external_id, d.total_tasks, d.idempotency_key, d.latest_retry_count, d.is_dag_operator
FROM
    v1_dags_olap d
JOIN
    lookup_task lt ON lt.tenant_id = d.tenant_id AND lt.dag_id = d.id AND lt.inserted_at = d.inserted_at
`

func (q *Queries) ReadDAGByExternalID(ctx context.Context, db DBTX, externalid uuid.UUID) (*V1DagsOlap, error) {
	row := db.QueryRow(ctx, readDAGByExternalID, externalid)
	var i V1DagsOlap
	err := row.Scan(
		&i.ID,
		&i.InsertedAt,
		&i.TenantID,
		&i.ExternalID,
		&i.DisplayName,
		&i.WorkflowID,
		&i.WorkflowVersionID,
		&i.ReadableStatus,
		&i.Input,
		&i.AdditionalMetadata,
		&i.ParentTaskExternalID,
		&i.TotalTasks,
		&i.IdempotencyKey,
		&i.LatestRetryCount,
		&i.IsDagOperator,
	)
	return &i, err
}

const readPayloadsOLAP = `-- name: ReadPayloadsOLAP :many
SELECT tenant_id, external_id, location, external_location_key, inline_content, inserted_at, updated_at
FROM v1_payloads_olap
WHERE
    tenant_id = $1::UUID
    AND external_id = ANY($2::UUID[])
`

type ReadPayloadsOLAPParams struct {
	Tenantid    uuid.UUID   `json:"tenantid"`
	Externalids []uuid.UUID `json:"externalids"`
}

func (q *Queries) ReadPayloadsOLAP(ctx context.Context, db DBTX, arg ReadPayloadsOLAPParams) ([]*V1PayloadsOlap, error) {
	rows, err := db.Query(ctx, readPayloadsOLAP, arg.Tenantid, arg.Externalids)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*V1PayloadsOlap
	for rows.Next() {
		var i V1PayloadsOlap
		if err := rows.Scan(
			&i.TenantID,
			&i.ExternalID,
			&i.Location,
			&i.ExternalLocationKey,
			&i.InlineContent,
			&i.InsertedAt,
			&i.UpdatedAt,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const readTaskByExternalID = `-- name: ReadTaskByExternalID :one
WITH lookup_task AS (
    SELECT
        tenant_id,
        task_id,
        inserted_at
    FROM
        v1_lookup_table_olap
    WHERE
        external_id = $1::uuid
)
SELECT
    t.tenant_id, t.id, t.inserted_at, t.external_id, t.queue, t.action_id, t.step_id, t.workflow_id, t.workflow_version_id, t.workflow_run_id, t.schedule_timeout, t.step_timeout, t.priority, t.sticky, t.desired_worker_id, t.display_name, t.input, t.additional_metadata, t.readable_status, t.latest_retry_count, t.latest_worker_id, t.dag_id, t.dag_inserted_at, t.parent_task_external_id, t.is_durable, t.idempotency_key,
    e.output,
    e.external_id AS event_external_id,
    e.error_message
FROM
    v1_tasks_olap t
JOIN
    lookup_task lt ON lt.tenant_id = t.tenant_id AND lt.task_id = t.id AND lt.inserted_at = t.inserted_at
JOIN
    v1_task_events_olap e ON (e.tenant_id, e.task_id, e.task_inserted_at, e.readable_status, e.retry_count) = (t.tenant_id, t.id, t.inserted_at, t.readable_status, t.latest_retry_count)
WHERE
    e.task_inserted_at >= (SELECT MIN(inserted_at) FROM lookup_task)
`

type ReadTaskByExternalIDRow struct {
	TenantID             uuid.UUID            `json:"tenant_id"`
	ID                   int64                `json:"id"`
	InsertedAt           pgtype.Timestamptz   `json:"inserted_at"`
	ExternalID           uuid.UUID            `json:"external_id"`
	Queue                string               `json:"queue"`
	ActionID             string               `json:"action_id"`
	StepID               uuid.UUID            `json:"step_id"`
	WorkflowID           uuid.UUID            `json:"workflow_id"`
	WorkflowVersionID    uuid.UUID            `json:"workflow_version_id"`
	WorkflowRunID        uuid.UUID            `json:"workflow_run_id"`
	ScheduleTimeout      string               `json:"schedule_timeout"`
	StepTimeout          pgtype.Text          `json:"step_timeout"`
	Priority             pgtype.Int4          `json:"priority"`
	Sticky               V1StickyStrategyOlap `json:"sticky"`
	DesiredWorkerID      *uuid.UUID           `json:"desired_worker_id"`
	DisplayName          string               `json:"display_name"`
	Input                []byte               `json:"input"`
	AdditionalMetadata   []byte               `json:"additional_metadata"`
	ReadableStatus       V1ReadableStatusOlap `json:"readable_status"`
	LatestRetryCount     int32                `json:"latest_retry_count"`
	LatestWorkerID       *uuid.UUID           `json:"latest_worker_id"`
	DagID                pgtype.Int8          `json:"dag_id"`
	DagInsertedAt        pgtype.Timestamptz   `json:"dag_inserted_at"`
	ParentTaskExternalID *uuid.UUID           `json:"parent_task_external_id"`
	IsDurable            bool                 `json:"is_durable"`
	IdempotencyKey       pgtype.Text          `json:"idempotency_key"`
	Output               []byte               `json:"output"`
	EventExternalID      uuid.UUID            `json:"event_external_id"`
	ErrorMessage         pgtype.Text          `json:"error_message"`
}

func (q *Queries) ReadTaskByExternalID(ctx context.Context, db DBTX, externalid uuid.UUID) (*ReadTaskByExternalIDRow, error) {
	row := db.QueryRow(ctx, readTaskByExternalID, externalid)
	var i ReadTaskByExternalIDRow
	err := row.Scan(
		&i.TenantID,
		&i.ID,
		&i.InsertedAt,
		&i.ExternalID,
		&i.Queue,
		&i.ActionID,
		&i.StepID,
		&i.WorkflowID,
		&i.WorkflowVersionID,
		&i.WorkflowRunID,
		&i.ScheduleTimeout,
		&i.StepTimeout,
		&i.Priority,
		&i.Sticky,
		&i.DesiredWorkerID,
		&i.DisplayName,
		&i.Input,
		&i.AdditionalMetadata,
		&i.ReadableStatus,
		&i.LatestRetryCount,
		&i.LatestWorkerID,
		&i.DagID,
		&i.DagInsertedAt,
		&i.ParentTaskExternalID,
		&i.IsDurable,
		&i.IdempotencyKey,
		&i.Output,
		&i.EventExternalID,
		&i.ErrorMessage,
	)
	return &i, err
}

const readWorkflowRunByExternalId = `-- name: ReadWorkflowRunByExternalId :one
WITH runs AS (
    SELECT
        lt.dag_id AS dag_id,
        lt.task_id AS task_id,
        r.id AS id,
        r.tenant_id,
        r.inserted_at,
        r.external_id,
        r.readable_status,
        r.kind,
        r.workflow_id,
        d.display_name AS display_name,
        d.input AS input,
        d.additional_metadata AS additional_metadata,
        d.workflow_version_id AS workflow_version_id,
        d.parent_task_external_id AS parent_task_external_id
    FROM
        v1_lookup_table_olap lt
    JOIN
        v1_runs_olap r ON r.inserted_at = lt.inserted_at AND r.id = lt.dag_id
    JOIN
        v1_dags_olap d ON (lt.tenant_id, lt.dag_id, lt.inserted_at) = (d.tenant_id, d.id, d.inserted_at)
    WHERE
        lt.external_id = $1::uuid
        AND lt.dag_id IS NOT NULL

    UNION ALL

    SELECT
        lt.dag_id AS dag_id,
        lt.task_id AS task_id,
        r.id AS id,
        r.tenant_id,
        r.inserted_at,
        r.external_id,
        r.readable_status,
        r.kind,
        r.workflow_id,
        t.display_name AS display_name,
        t.input AS input,
        t.additional_metadata AS additional_metadata,
        t.workflow_version_id AS workflow_version_id,
        NULL::UUID AS parent_task_external_id
    FROM
        v1_lookup_table_olap lt
    JOIN
        v1_runs_olap r ON r.inserted_at = lt.inserted_at AND r.id = lt.task_id
    JOIN
        v1_tasks_olap t ON (lt.tenant_id, lt.task_id, lt.inserted_at) = (t.tenant_id, t.id, t.inserted_at)
    WHERE
        lt.external_id = $1::uuid
        AND lt.task_id IS NOT NULL
), relevant_events AS (
    SELECT
        e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count, e.is_dag_orchestrator
    FROM runs r
    JOIN v1_dag_to_task_olap dt ON r.dag_id = dt.dag_id AND r.inserted_at = dt.dag_inserted_at
    JOIN v1_task_events_olap e ON (e.task_id, e.task_inserted_at) = (dt.task_id, dt.task_inserted_at)
    WHERE
        r.dag_id IS NOT NULL
        AND dt.dag_inserted_at >= (SELECT MIN(inserted_at) FROM runs)
        AND e.task_inserted_at >= (SELECT MIN(inserted_at) FROM runs)

    UNION ALL

    SELECT
        e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count, e.is_dag_orchestrator
    FROM runs r
    JOIN v1_task_events_olap e ON e.task_id = r.task_id AND e.task_inserted_at = r.inserted_at
    WHERE
        r.task_id IS NOT NULL
        AND e.task_inserted_at >= (SELECT MIN(inserted_at) FROM runs)
), max_retry_counts AS (
    SELECT task_id, MAX(retry_count) AS max_retry_count
    FROM relevant_events
    GROUP BY task_id
), metadata AS (
    SELECT
        MIN(e.inserted_at)::timestamptz AS created_at,
        MIN(e.inserted_at) FILTER (WHERE e.readable_status = 'RUNNING')::timestamptz AS started_at,
        MAX(e.inserted_at) FILTER (WHERE e.readable_status IN ('COMPLETED', 'CANCELLED', 'FAILED'))::timestamptz AS finished_at,
        JSON_AGG(JSON_BUILD_OBJECT('task_id', e.task_id,'task_inserted_at', e.task_inserted_at)) AS task_metadata
    FROM
        relevant_events e
    JOIN max_retry_counts mrc ON (e.task_id, e.retry_count) = (mrc.task_id, mrc.max_retry_count)
), error_message AS (
    SELECT
        e.error_message
    FROM
        relevant_events e
    WHERE
        e.readable_status = 'FAILED'
    ORDER BY
        e.retry_count DESC
    LIMIT 1
), output_event_external_id AS (
    SELECT
        external_id,
        inserted_at
    FROM
        relevant_events
    WHERE
        event_type = 'FINISHED'
    ORDER BY
        inserted_at DESC
    LIMIT 1
)

SELECT
    r.dag_id, r.task_id, r.id, r.tenant_id, r.inserted_at, r.external_id, r.readable_status, r.kind, r.workflow_id, r.display_name, r.input, r.additional_metadata, r.workflow_version_id, r.parent_task_external_id,
    m.created_at,
    m.started_at,
    m.finished_at,
    e.error_message,
    m.task_metadata,
    o.external_id AS output_event_external_id,
    o.inserted_at AS output_event_inserted_at
FROM runs r
LEFT JOIN metadata m ON true
LEFT JOIN error_message e ON true
LEFT JOIN output_event_external_id o ON true
ORDER BY r.inserted_at DESC
`

type ReadWorkflowRunByExternalIdRow struct {
	DagID                 pgtype.Int8          `json:"dag_id"`
	TaskID                pgtype.Int8          `json:"task_id"`
	ID                    int64                `json:"id"`
	TenantID              uuid.UUID            `json:"tenant_id"`
	InsertedAt            pgtype.Timestamptz   `json:"inserted_at"`
	ExternalID            uuid.UUID            `json:"external_id"`
	ReadableStatus        V1ReadableStatusOlap `json:"readable_status"`
	Kind                  V1RunKind            `json:"kind"`
	WorkflowID            uuid.UUID            `json:"workflow_id"`
	DisplayName           string               `json:"display_name"`
	Input                 []byte               `json:"input"`
	AdditionalMetadata    []byte               `json:"additional_metadata"`
	WorkflowVersionID     uuid.UUID            `json:"workflow_version_id"`
	ParentTaskExternalID  *uuid.UUID           `json:"parent_task_external_id"`
	CreatedAt             pgtype.Timestamptz   `json:"created_at"`
	StartedAt             pgtype.Timestamptz   `json:"started_at"`
	FinishedAt            pgtype.Timestamptz   `json:"finished_at"`
	ErrorMessage          pgtype.Text          `json:"error_message"`
	TaskMetadata          []byte               `json:"task_metadata"`
	OutputEventExternalID *uuid.UUID           `json:"output_event_external_id"`
	OutputEventInsertedAt pgtype.Timestamptz   `json:"output_event_inserted_at"`
}

func (q *Queries) ReadWorkflowRunByExternalId(ctx context.Context, db DBTX, workflowrunexternalid uuid.UUID) (*ReadWorkflowRunByExternalIdRow, error) {
	row := db.QueryRow(ctx, readWorkflowRunByExternalId, workflowrunexternalid)
	var i ReadWorkflowRunByExternalIdRow
	err := row.Scan(
		&i.DagID,
		&i.TaskID,
		&i.ID,
		&i.TenantID,
		&i.InsertedAt,
		&i.ExternalID,
		&i.ReadableStatus,
		&i.Kind,
		&i.WorkflowID,
		&i.DisplayName,
		&i.Input,
		&i.AdditionalMetadata,
		&i.WorkflowVersionID,
		&i.ParentTaskExternalID,
		&i.CreatedAt,
		&i.StartedAt,
		&i.FinishedAt,
		&i.ErrorMessage,
		&i.TaskMetadata,
		&i.OutputEventExternalID,
		&i.OutputEventInsertedAt,
	)
	return &i, err
}

const reconcileOperatorDAGStatusesOnCreate = `-- name: ReconcileOperatorDAGStatusesOnCreate :exec
WITH inputs AS (
    SELECT
        UNNEST($1::BIGINT[]) AS dag_id,
        UNNEST($2::TIMESTAMPTZ[]) AS dag_inserted_at
), latest_event AS (
    SELECT DISTINCT ON (e.task_id, e.task_inserted_at)
        e.task_id,
        e.task_inserted_at,
        e.readable_status,
        e.retry_count
    FROM v1_task_events_olap e
    -- dag operator task ids are the same as the dag ids
    JOIN inputs i ON (e.task_id, e.task_inserted_at) = (i.dag_id, i.dag_inserted_at)
    WHERE e.task_inserted_at >= (SELECT MIN(dag_inserted_at) FROM inputs)
    ORDER BY
        e.task_id,
        e.task_inserted_at,
        e.retry_count DESC,
        v1_status_to_priority(e.readable_status) DESC,
        e.id DESC
)
UPDATE v1_dags_olap d
SET
    readable_status = le.readable_status,
    latest_retry_count = le.retry_count
FROM latest_event le
WHERE (d.id, d.inserted_at) = (le.task_id, le.task_inserted_at)
    AND (
        le.retry_count > d.latest_retry_count
        OR (
            le.retry_count = d.latest_retry_count
            AND v1_status_to_priority(le.readable_status) > v1_status_to_priority(d.readable_status)
        )
        OR (
            le.retry_count = d.latest_retry_count
            AND d.readable_status = 'EVICTED'
            AND le.readable_status != 'EVICTED'
        )
    )
`

type ReconcileOperatorDAGStatusesOnCreateParams struct {
	Dagids         []int64              `json:"dagids"`
	Daginsertedats []pgtype.Timestamptz `json:"daginsertedats"`
}

// An operator DAG's OLAP row is written by a 'created-dag' message that can be raced by status events.
// This query reconciles DAG OLAP rows that are created *after* status events have already arrived.
func (q *Queries) ReconcileOperatorDAGStatusesOnCreate(ctx context.Context, db DBTX, arg ReconcileOperatorDAGStatusesOnCreateParams) error {
	_, err := db.Exec(ctx, reconcileOperatorDAGStatusesOnCreate, arg.Dagids, arg.Daginsertedats)
	return err
}

const reconcileTaskStatusesFromEvents = `-- name: ReconcileTaskStatusesFromEvents :many
WITH inputs AS (
    SELECT
        UNNEST($1::BIGINT[]) AS task_id,
        UNNEST($2::TIMESTAMPTZ[]) AS task_inserted_at
), locked_tasks AS (
    SELECT tenant_id, id, inserted_at, external_id, queue, action_id, step_id, workflow_id, workflow_version_id, workflow_run_id, schedule_timeout, step_timeout, priority, sticky, desired_worker_id, display_name, input, additional_metadata, readable_status, latest_retry_count, latest_worker_id, dag_id, dag_inserted_at, parent_task_external_id, is_durable, idempotency_key
    FROM v1_tasks_olap
    WHERE
        (id, inserted_at) IN (SELECT task_id, task_inserted_at FROM inputs)
        AND inserted_at >= $3::TIMESTAMPTZ
    ORDER BY inserted_at, id
    FOR UPDATE
), relevant_events AS MATERIALIZED (
    SELECT tenant_id, id, inserted_at, external_id, task_id, task_inserted_at, event_type, workflow_id, event_timestamp, readable_status, retry_count, error_message, output, worker_id, additional__event_data, additional__event_message, durable_invocation_count, is_dag_orchestrator
    FROM v1_task_events_olap
    WHERE
        (task_id, task_inserted_at) IN (SELECT task_id, task_inserted_at FROM inputs)
        AND task_inserted_at >= $3::TIMESTAMPTZ
), max_retry_counts AS (
    SELECT
        task_id,
        task_inserted_at,
        MAX(retry_count) AS max_retry_count
    FROM relevant_events
    GROUP BY task_id, task_inserted_at
), statuses_from_events AS (
    SELECT
        e.task_id,
        e.task_inserted_at,
        e.retry_count,
        v1_status_from_priority(MAX(v1_status_to_priority(e.readable_status))) AS status,
        MAX(e.worker_id::text)::uuid AS latest_worker_id
    FROM relevant_events e
    JOIN max_retry_counts mrc ON (e.task_id, e.task_inserted_at, e.retry_count) = (mrc.task_id, mrc.task_inserted_at, mrc.max_retry_count)
    GROUP BY e.task_id, e.task_inserted_at, e.retry_count
)

UPDATE v1_tasks_olap t
SET
    readable_status = s.status,
    latest_retry_count = s.retry_count,
    latest_worker_id = COALESCE(s.latest_worker_id, t.latest_worker_id)
FROM statuses_from_events s
WHERE
    (t.id, t.inserted_at) = (s.task_id, s.task_inserted_at)
    AND (
        -- A newer retry count always applies, even when the readable status is
        -- unchanged — same rationale as the guards in UpdateTaskStatuses and
        -- UpdateTaskStatusesFromMQ: requiring the status to differ would skip
        -- the latest_retry_count advance when the event history's newest retry
        -- ends in the status the row already has.
        (s.retry_count > t.latest_retry_count)
        OR (s.retry_count = t.latest_retry_count AND v1_status_to_priority(s.status) > v1_status_to_priority(t.readable_status))
        OR (s.retry_count = t.latest_retry_count AND t.readable_status = 'EVICTED' AND s.status != 'EVICTED')
    )
RETURNING t.tenant_id, t.id, t.inserted_at, t.external_id, t.readable_status, t.latest_worker_id, t.workflow_id, t.dag_id, t.dag_inserted_at
`

type ReconcileTaskStatusesFromEventsParams struct {
	Taskids         []int64              `json:"taskids"`
	Taskinsertedats []pgtype.Timestamptz `json:"taskinsertedats"`
	Mininsertedat   pgtype.Timestamptz   `json:"mininsertedat"`
}

type ReconcileTaskStatusesFromEventsRow struct {
	TenantID       uuid.UUID            `json:"tenant_id"`
	ID             int64                `json:"id"`
	InsertedAt     pgtype.Timestamptz   `json:"inserted_at"`
	ExternalID     uuid.UUID            `json:"external_id"`
	ReadableStatus V1ReadableStatusOlap `json:"readable_status"`
	LatestWorkerID *uuid.UUID           `json:"latest_worker_id"`
	WorkflowID     uuid.UUID            `json:"workflow_id"`
	DagID          pgtype.Int8          `json:"dag_id"`
	DagInsertedAt  pgtype.Timestamptz   `json:"dag_inserted_at"`
}

func (q *Queries) ReconcileTaskStatusesFromEvents(ctx context.Context, db DBTX, arg ReconcileTaskStatusesFromEventsParams) ([]*ReconcileTaskStatusesFromEventsRow, error) {
	rows, err := db.Query(ctx, reconcileTaskStatusesFromEvents, arg.Taskids, arg.Taskinsertedats, arg.Mininsertedat)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*ReconcileTaskStatusesFromEventsRow
	for rows.Next() {
		var i ReconcileTaskStatusesFromEventsRow
		if err := rows.Scan(
			&i.TenantID,
			&i.ID,
			&i.InsertedAt,
			&i.ExternalID,
			&i.ReadableStatus,
			&i.LatestWorkerID,
			&i.WorkflowID,
			&i.DagID,
			&i.DagInsertedAt,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const storeCELEvaluationFailures = `-- name: StoreCELEvaluationFailures :exec
WITH inputs AS (
    SELECT
        UNNEST(CAST($2::TEXT[] AS v1_cel_evaluation_failure_source[])) AS source,
        UNNEST($3::TEXT[]) AS error
)
INSERT INTO v1_cel_evaluation_failures_olap (
    tenant_id,
    source,
    error
)
SELECT $1::UUID, source, error
FROM inputs
`

type StoreCELEvaluationFailuresParams struct {
	Tenantid uuid.UUID `json:"tenantid"`
	Sources  []string  `json:"sources"`
	Errors   []string  `json:"errors"`
}

func (q *Queries) StoreCELEvaluationFailures(ctx context.Context, db DBTX, arg StoreCELEvaluationFailuresParams) error {
	_, err := db.Exec(ctx, storeCELEvaluationFailures, arg.Tenantid, arg.Sources, arg.Errors)
	return err
}

const swapV1PayloadOLAPPartitionWithTemp = `-- name: SwapV1PayloadOLAPPartitionWithTemp :exec
SELECT swap_v1_payloads_olap_partition_with_temp($1::DATE)
`

func (q *Queries) SwapV1PayloadOLAPPartitionWithTemp(ctx context.Context, db DBTX, date pgtype.Date) error {
	_, err := db.Exec(ctx, swapV1PayloadOLAPPartitionWithTemp, date)
	return err
}

const updateDAGStatuses = `-- name: UpdateDAGStatuses :many
WITH tenants AS (
    SELECT UNNEST(
        find_matching_tenants_in_task_status_updates_tmp_partition(
            $1::int,
            $2::UUID[]
        )
    ) AS tenant_id
), locked_events AS (
    SELECT
        u.tenant_id, u.requeue_after, u.requeue_retries, u.id, u.dag_id, u.dag_inserted_at
    FROM tenants t,
        LATERAL list_task_status_updates_tmp(
            $1::int,
            t.tenant_id,
            $3::int
        ) u
), distinct_dags AS (
    SELECT
        DISTINCT ON (e.tenant_id, e.dag_id, e.dag_inserted_at)
        e.tenant_id,
        e.dag_id,
        e.dag_inserted_at
    FROM
        locked_events e
), locked_dags AS (
    SELECT
        d.id,
        d.inserted_at,
        d.readable_status,
        d.tenant_id,
        d.total_tasks
    FROM
        v1_dags_olap d
    WHERE
        d.inserted_at >= $4::TIMESTAMPTZ
        AND (d.inserted_at, d.id, d.tenant_id) IN (
            SELECT
                dd.dag_inserted_at, dd.dag_id, dd.tenant_id
            FROM
                distinct_dags dd
        )
        -- see UpdateDAGStatusesFromMQ
        AND NOT d.is_dag_operator
    ORDER BY
        d.inserted_at, d.id
    FOR UPDATE
), relevant_tasks AS (
    SELECT
        t.tenant_id,
        t.id,
        d.id AS dag_id,
        d.inserted_at AS dag_inserted_at,
        t.readable_status
    FROM
        locked_dags d
    JOIN
        v1_dag_to_task_olap dt ON
            (d.id, d.inserted_at) = (dt.dag_id, dt.dag_inserted_at)
    JOIN
        v1_tasks_olap t ON
            (dt.task_id, dt.task_inserted_at) = (t.id, t.inserted_at)
    WHERE
        dt.dag_inserted_at >= $4::TIMESTAMPTZ
        AND t.inserted_at >= $4::TIMESTAMPTZ
    -- Note that the ORDER BY seems to help the query planner by pruning partitions earlier. We
    -- have previously seen Postgres use an index-only scan on partitions older than the minInsertedAt,
    -- each of which can take a long time to scan. This can be very pathological since we partition on
    -- both the status and the date, so 14 days of data with 5 statuses is 70 partitions to index scan.
    ORDER BY t.inserted_at DESC
), dag_task_counts AS (
    SELECT
        d.id,
        d.inserted_at,
        d.readable_status,
        d.tenant_id,
        d.total_tasks,
        COUNT(t.id) AS task_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'COMPLETED') AS completed_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'FAILED') AS failed_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'CANCELLED') AS cancelled_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'QUEUED') AS queued_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'RUNNING') AS running_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'EVICTED') AS evicted_count
    FROM
        locked_dags d
    LEFT JOIN
        relevant_tasks t ON (d.tenant_id, d.id, d.inserted_at) = (t.tenant_id, t.dag_id, t.dag_inserted_at)
    GROUP BY
        d.id, d.inserted_at, d.readable_status, d.tenant_id, d.total_tasks
), dag_new_statuses AS (
    SELECT
        dtc.id,
        dtc.inserted_at,
        dtc.readable_status AS old_readable_status,
        dtc.tenant_id,
        CASE
            -- If we only have queued events, we should keep the status as is
            WHEN dtc.queued_count = dtc.task_count THEN dtc.readable_status
            -- If the task count is not equal to the total tasks, we should set the status to running
            WHEN dtc.task_count != dtc.total_tasks THEN 'RUNNING'
            -- If we have any running or queued tasks, we should set the status to running
            WHEN dtc.running_count > 0 OR dtc.queued_count > 0 THEN 'RUNNING'
            -- If all tasks are evicted, mark DAG as evicted
            WHEN dtc.evicted_count = dtc.task_count AND dtc.task_count = dtc.total_tasks THEN 'EVICTED'
            WHEN dtc.failed_count > 0 THEN 'FAILED'
            WHEN dtc.cancelled_count > 0 THEN 'CANCELLED'
            WHEN dtc.completed_count = dtc.task_count THEN 'COMPLETED'
            ELSE 'RUNNING'
        END::v1_readable_status_olap AS new_readable_status
    FROM
        dag_task_counts dtc
), already_in_target_partition AS (
    -- Check if rows already exist in the target partition (with the new readable_status)
    -- This is to avoid rare cases of duplicates writes causing unique constraint violations,
    -- when i.e. we write to one partition of QUEUED, that gets moved from QUEUED -> COMPLETED,
    -- and we later insert into QUEUED again but try to update to COMPLETED again.
    SELECT
        dns.tenant_id,
        dns.id,
        dns.inserted_at,
        dns.old_readable_status,
        dns.new_readable_status
    FROM
        dag_new_statuses dns
    JOIN
        v1_dags_olap d ON
            d.inserted_at = dns.inserted_at
            AND d.id = dns.id
            AND d.readable_status = dns.new_readable_status
    WHERE
        -- Only consider rows where we would actually change the status
        dns.old_readable_status != dns.new_readable_status
), updated_target_partition_dags AS (
    -- Update the existing rows in the target partition
    UPDATE
        v1_dags_olap d
    SET
        -- Touch the row to ensure it's returned (no-op update)
        readable_status = d.readable_status
    FROM
        already_in_target_partition ap
    WHERE
        (d.inserted_at, d.id, d.readable_status) = (ap.inserted_at, ap.id, ap.new_readable_status)
    RETURNING
        d.tenant_id, d.id, d.inserted_at, d.readable_status, d.external_id, d.workflow_id
), deleted_duplicate_dags AS (
    -- Delete the old rows that already have a copy in the target partition
    DELETE FROM
        v1_dags_olap d
    USING
        already_in_target_partition ap
    WHERE
        (d.inserted_at, d.id, d.readable_status) = (ap.inserted_at, ap.id, ap.old_readable_status)
), dags_to_update AS (
    -- DAGs that need updating and don't already exist in target partition
    SELECT DISTINCT ON (dns.tenant_id, dns.id, dns.inserted_at)
        dns.tenant_id,
        dns.id,
        dns.inserted_at,
        dns.old_readable_status,
        dns.new_readable_status
    FROM
        dag_new_statuses dns
    WHERE
        NOT EXISTS (
            SELECT 1
            FROM already_in_target_partition ap
            WHERE (ap.tenant_id, ap.id, ap.inserted_at) = (dns.tenant_id, dns.id, dns.inserted_at)
        )
    ORDER BY
        dns.tenant_id, dns.id, dns.inserted_at, dns.old_readable_status
), updated_dags AS (
    UPDATE
        v1_dags_olap d
    SET
        readable_status = dtu.new_readable_status
    FROM
        dags_to_update dtu
    WHERE
        (d.inserted_at, d.id, d.readable_status) = (dtu.inserted_at, dtu.id, dtu.old_readable_status)
        AND dtu.old_readable_status != dtu.new_readable_status
    RETURNING
        d.tenant_id, d.id, d.inserted_at, d.readable_status, d.external_id, d.workflow_id
), all_result_dags AS (
    -- Combine updated dags from both paths
    SELECT tenant_id, id, inserted_at, readable_status, external_id, workflow_id
    FROM updated_dags
    UNION ALL
    SELECT tenant_id, id, inserted_at, readable_status, external_id, workflow_id
    FROM updated_target_partition_dags
), events_to_requeue AS (
    -- Get events which don't have a corresponding locked_dag
    SELECT
        e.tenant_id,
        e.requeue_retries,
        e.dag_id,
        e.dag_inserted_at
    FROM
        locked_events e
    WHERE NOT EXISTS (
        SELECT 1
        FROM locked_dags d
        WHERE (e.dag_inserted_at, e.dag_id, e.tenant_id) = (d.inserted_at, d.id, d.tenant_id)
    )
), deleted_events AS (
    DELETE FROM
        v1_task_status_updates_tmp
    WHERE
        (tenant_id, requeue_after, dag_id, id) IN (SELECT tenant_id, requeue_after, dag_id, id FROM locked_events)
), requeued_events AS (
    INSERT INTO
        v1_task_status_updates_tmp (
            tenant_id,
            requeue_after,
            requeue_retries,
            dag_id,
            dag_inserted_at
        )
    SELECT
        tenant_id,
        -- Exponential backoff, we limit to 10 retries which is 2048 seconds/34 minutes
        CURRENT_TIMESTAMP + (2 ^ requeue_retries) * INTERVAL '2 seconds',
        requeue_retries + 1,
        dag_id,
        dag_inserted_at
    FROM
        events_to_requeue
    WHERE
        requeue_retries < 10
    RETURNING
        tenant_id, requeue_after, requeue_retries, id, dag_id, dag_inserted_at
), event_count AS (
    SELECT
        COUNT(*) as count
    FROM
        locked_events
)
SELECT
    -- Little wonky, but we return the count of events that were processed in each row. Potential edge case
    -- where there are no tasks updated with a non-zero count, but this should be very rare and we'll get
    -- updates on the next run.
    (SELECT count FROM event_count) AS count,
    d.tenant_id, d.id, d.inserted_at, d.readable_status, d.external_id, d.workflow_id
FROM
    all_result_dags d
`

type UpdateDAGStatusesParams struct {
	Partitionnumber int32              `json:"partitionnumber"`
	Tenantids       []uuid.UUID        `json:"tenantids"`
	Eventlimit      int32              `json:"eventlimit"`
	Mininsertedat   pgtype.Timestamptz `json:"mininsertedat"`
}

type UpdateDAGStatusesRow struct {
	Count          int64                `json:"count"`
	TenantID       uuid.UUID            `json:"tenant_id"`
	ID             int64                `json:"id"`
	InsertedAt     pgtype.Timestamptz   `json:"inserted_at"`
	ReadableStatus V1ReadableStatusOlap `json:"readable_status"`
	ExternalID     uuid.UUID            `json:"external_id"`
	WorkflowID     uuid.UUID            `json:"workflow_id"`
}

func (q *Queries) UpdateDAGStatuses(ctx context.Context, db DBTX, arg UpdateDAGStatusesParams) ([]*UpdateDAGStatusesRow, error) {
	rows, err := db.Query(ctx, updateDAGStatuses,
		arg.Partitionnumber,
		arg.Tenantids,
		arg.Eventlimit,
		arg.Mininsertedat,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*UpdateDAGStatusesRow
	for rows.Next() {
		var i UpdateDAGStatusesRow
		if err := rows.Scan(
			&i.Count,
			&i.TenantID,
			&i.ID,
			&i.InsertedAt,
			&i.ReadableStatus,
			&i.ExternalID,
			&i.WorkflowID,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const updateDAGStatusesFromMQ = `-- name: UpdateDAGStatusesFromMQ :many
WITH inputs AS (
    SELECT
        UNNEST($1::UUID[]) AS tenant_id,
        UNNEST($2::BIGINT[]) AS dag_id,
        UNNEST($3::TIMESTAMPTZ[]) AS dag_inserted_at
), locked_dags AS (
    SELECT id, inserted_at, tenant_id, external_id, display_name, workflow_id, workflow_version_id, readable_status, input, additional_metadata, parent_task_external_id, total_tasks, idempotency_key, latest_retry_count, is_dag_operator
    FROM v1_dags_olap d
    WHERE
        (d.inserted_at, d.id, d.tenant_id) IN (
            SELECT dag_inserted_at, dag_id, tenant_id
            FROM inputs
        )
    -- operator dags are updated by the separate UpdateDAGStatusesFromOrchestratorEvents
    AND NOT d.is_dag_operator
    ORDER BY inserted_at, id
    FOR UPDATE
), dag_task_counts AS (
    SELECT
        d.id,
        d.inserted_at,
        d.readable_status,
        d.tenant_id,
        d.total_tasks,
        COUNT(t.id) AS task_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'COMPLETED') AS completed_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'FAILED') AS failed_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'CANCELLED') AS cancelled_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'QUEUED') AS queued_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'RUNNING') AS running_count,
        COUNT(t.id) FILTER (WHERE t.readable_status = 'EVICTED') AS evicted_count
    FROM
        locked_dags d
    JOIN
        v1_dag_to_task_olap dt ON (d.id, d.inserted_at) = (dt.dag_id, dt.dag_inserted_at)
    JOIN
        v1_tasks_olap t ON (dt.task_id, dt.task_inserted_at) = (t.id, t.inserted_at)
    WHERE
        dt.dag_inserted_at >= $4::TIMESTAMPTZ
        AND t.inserted_at >= $4::TIMESTAMPTZ
    GROUP BY
        d.id, d.inserted_at, d.readable_status, d.tenant_id, d.total_tasks
), dag_new_statuses AS (
    SELECT
        dtc.id,
        dtc.inserted_at,
        dtc.tenant_id,
        CASE
            -- If we only have queued events, we should keep the status as is
            WHEN dtc.queued_count = dtc.task_count THEN dtc.readable_status
            -- If the task count is not equal to the total tasks, we should set the status to running
            WHEN dtc.task_count != dtc.total_tasks THEN 'RUNNING'
            -- If we have any running or queued tasks, we should set the status to running
            WHEN dtc.running_count > 0 OR dtc.queued_count > 0 THEN 'RUNNING'
            -- If all tasks are evicted, mark DAG as evicted
            WHEN dtc.evicted_count = dtc.task_count AND dtc.task_count = dtc.total_tasks THEN 'EVICTED'
            WHEN dtc.failed_count > 0 THEN 'FAILED'
            WHEN dtc.cancelled_count > 0 THEN 'CANCELLED'
            WHEN dtc.completed_count = dtc.task_count THEN 'COMPLETED'
            ELSE 'RUNNING'
        END::v1_readable_status_olap AS new_readable_status
    FROM
        dag_task_counts dtc
), updated_dags AS (
    UPDATE v1_dags_olap d
    SET
        readable_status = dns.new_readable_status
    FROM
        dag_new_statuses dns
    WHERE
        (d.inserted_at, d.id, d.tenant_id) = (dns.inserted_at, dns.id, dns.tenant_id)
        AND dns.new_readable_status != d.readable_status
    RETURNING
        d.tenant_id, d.id, d.inserted_at, d.external_id, d.readable_status, d.workflow_id
)

SELECT
    ud.tenant_id::UUID AS tenant_id,
    ud.id::BIGINT AS dag_id,
    ud.inserted_at::TIMESTAMPTZ AS dag_inserted_at,
    ud.external_id::UUID AS external_id,
    ud.readable_status::v1_readable_status_olap AS readable_status,
    ud.workflow_id::UUID AS workflow_id
FROM updated_dags ud
`

type UpdateDAGStatusesFromMQParams struct {
	Tenantids        []uuid.UUID          `json:"tenantids"`
	Dagids           []int64              `json:"dagids"`
	Daginsertedats   []pgtype.Timestamptz `json:"daginsertedats"`
	Mindaginsertedat pgtype.Timestamptz   `json:"mindaginsertedat"`
}

type UpdateDAGStatusesFromMQRow struct {
	TenantID       uuid.UUID            `json:"tenant_id"`
	DagID          int64                `json:"dag_id"`
	DagInsertedAt  pgtype.Timestamptz   `json:"dag_inserted_at"`
	ExternalID     uuid.UUID            `json:"external_id"`
	ReadableStatus V1ReadableStatusOlap `json:"readable_status"`
	WorkflowID     uuid.UUID            `json:"workflow_id"`
}

func (q *Queries) UpdateDAGStatusesFromMQ(ctx context.Context, db DBTX, arg UpdateDAGStatusesFromMQParams) ([]*UpdateDAGStatusesFromMQRow, error) {
	rows, err := db.Query(ctx, updateDAGStatusesFromMQ,
		arg.Tenantids,
		arg.Dagids,
		arg.Daginsertedats,
		arg.Mindaginsertedat,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*UpdateDAGStatusesFromMQRow
	for rows.Next() {
		var i UpdateDAGStatusesFromMQRow
		if err := rows.Scan(
			&i.TenantID,
			&i.DagID,
			&i.DagInsertedAt,
			&i.ExternalID,
			&i.ReadableStatus,
			&i.WorkflowID,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const updateDAGStatusesFromOrchestratorEvents = `-- name: UpdateDAGStatusesFromOrchestratorEvents :many
WITH inputs AS (
    SELECT
        UNNEST($2::BIGINT[]) AS dag_id,
        UNNEST($3::TIMESTAMPTZ[]) AS dag_inserted_at,
        UNNEST($4::v1_readable_status_olap[]) AS new_readable_status,
        UNNEST($5::INTEGER[]) AS retry_count,
        UNNEST($6::UUID[]) AS external_id,
        UNNEST($7::TEXT[]) AS display_name,
        UNNEST($8::UUID[]) AS workflow_id,
        UNNEST($9::UUID[]) AS workflow_version_id,
        UNNEST($10::JSONB[]) AS additional_metadata
)
INSERT INTO v1_dags_olap (
    tenant_id,
    id,
    inserted_at,
    external_id,
    display_name,
    workflow_id,
    workflow_version_id,
    input,
    additional_metadata,
    total_tasks,
    readable_status,
    latest_retry_count,
    is_dag_operator
)
SELECT
    $1::UUID,
    i.dag_id,
    i.dag_inserted_at,
    i.external_id,
    i.display_name,
    i.workflow_id,
    i.workflow_version_id,
    '{}'::JSONB,
    i.additional_metadata,
    0,
    i.new_readable_status,
    i.retry_count,
    TRUE
FROM inputs i
ORDER BY i.dag_inserted_at, i.dag_id
ON CONFLICT (inserted_at, id) DO UPDATE SET
    readable_status = EXCLUDED.readable_status,
    latest_retry_count = EXCLUDED.latest_retry_count
WHERE
    v1_dags_olap.tenant_id = $1::UUID
    AND (
        -- a newer attempt always applies, so a replay revives a terminal DAG and a straggler
        -- from the previous attempt can no longer clobber it
        (
            EXCLUDED.latest_retry_count > v1_dags_olap.latest_retry_count
        ) OR
        -- within an attempt, only move forward through QUEUED -> RUNNING -> terminal
        (
            EXCLUDED.latest_retry_count = v1_dags_olap.latest_retry_count
            AND v1_status_to_priority(EXCLUDED.readable_status) > v1_status_to_priority(v1_dags_olap.readable_status)
        ) OR
        -- EVICTED is reversible (a durable restore moves it back to RUNNING) but outranks
        -- RUNNING by priority, so it needs an explicit escape hatch
        (
            EXCLUDED.latest_retry_count = v1_dags_olap.latest_retry_count
            AND v1_dags_olap.readable_status = 'EVICTED'
            AND EXCLUDED.readable_status != 'EVICTED'
        )
    )
RETURNING
    tenant_id::UUID AS tenant_id,
    id::BIGINT AS dag_id,
    inserted_at::TIMESTAMPTZ AS dag_inserted_at,
    external_id::UUID AS external_id,
    readable_status::v1_readable_status_olap AS readable_status,
    workflow_id::UUID AS workflow_id
`

type UpdateDAGStatusesFromOrchestratorEventsParams struct {
	Tenantid            uuid.UUID              `json:"tenantid"`
	Dagids              []int64                `json:"dagids"`
	Daginsertedats      []pgtype.Timestamptz   `json:"daginsertedats"`
	Statuses            []V1ReadableStatusOlap `json:"statuses"`
	Retrycounts         []int32                `json:"retrycounts"`
	Externalids         []uuid.UUID            `json:"externalids"`
	Displaynames        []string               `json:"displaynames"`
	Workflowids         []uuid.UUID            `json:"workflowids"`
	Workflowversionids  []uuid.UUID            `json:"workflowversionids"`
	Additionalmetadatas [][]byte               `json:"additionalmetadatas"`
}

type UpdateDAGStatusesFromOrchestratorEventsRow struct {
	TenantID       uuid.UUID            `json:"tenant_id"`
	DagID          int64                `json:"dag_id"`
	DagInsertedAt  pgtype.Timestamptz   `json:"dag_inserted_at"`
	ExternalID     uuid.UUID            `json:"external_id"`
	ReadableStatus V1ReadableStatusOlap `json:"readable_status"`
	WorkflowID     uuid.UUID            `json:"workflow_id"`
}

// important: need to dedupe the dags before calling this query to make
// sure each dag only gets one candidate update per call
func (q *Queries) UpdateDAGStatusesFromOrchestratorEvents(ctx context.Context, db DBTX, arg UpdateDAGStatusesFromOrchestratorEventsParams) ([]*UpdateDAGStatusesFromOrchestratorEventsRow, error) {
	rows, err := db.Query(ctx, updateDAGStatusesFromOrchestratorEvents,
		arg.Tenantid,
		arg.Dagids,
		arg.Daginsertedats,
		arg.Statuses,
		arg.Retrycounts,
		arg.Externalids,
		arg.Displaynames,
		arg.Workflowids,
		arg.Workflowversionids,
		arg.Additionalmetadatas,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*UpdateDAGStatusesFromOrchestratorEventsRow
	for rows.Next() {
		var i UpdateDAGStatusesFromOrchestratorEventsRow
		if err := rows.Scan(
			&i.TenantID,
			&i.DagID,
			&i.DagInsertedAt,
			&i.ExternalID,
			&i.ReadableStatus,
			&i.WorkflowID,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const updateTaskStatuses = `-- name: UpdateTaskStatuses :many
WITH tenants AS (
    SELECT UNNEST(
        find_matching_tenants_in_task_events_tmp_partition(
            $1::int,
            $2::UUID[]
        )
    ) AS tenant_id
), locked_events AS (
    SELECT
        e.tenant_id, e.requeue_after, e.requeue_retries, e.id, e.task_id, e.task_inserted_at, e.event_type, e.readable_status, e.retry_count, e.worker_id
    FROM tenants t,
        LATERAL list_task_events_tmp(
            $1::int,
            t.tenant_id,
            $3::int
        ) e
), max_retry_counts AS (
    SELECT
        tenant_id,
        task_id,
        task_inserted_at,
        MAX(retry_count) AS max_retry_count
    FROM
        locked_events
    GROUP BY
        tenant_id, task_id, task_inserted_at
), updatable_events AS (
    SELECT
        e.tenant_id,
        e.task_id,
        e.task_inserted_at,
        e.retry_count,
        v1_status_from_priority(MAX(v1_status_to_priority(e.readable_status))) AS max_readable_status
    FROM
        locked_events e
    JOIN
        max_retry_counts mrc ON
            e.tenant_id = mrc.tenant_id
            AND e.task_id = mrc.task_id
            AND e.task_inserted_at = mrc.task_inserted_at
            AND e.retry_count = mrc.max_retry_count
    GROUP BY
        e.tenant_id, e.task_id, e.task_inserted_at, e.retry_count
), latest_worker_id AS (
    SELECT
        tenant_id,
        task_id,
        task_inserted_at,
        retry_count,
        MAX(worker_id::text) AS worker_id
    FROM
        locked_events
    WHERE
        worker_id IS NOT NULL
    GROUP BY
        tenant_id, task_id, task_inserted_at, retry_count
), distinct_tasks_to_lock AS (
    SELECT DISTINCT ON (t.tenant_id, t.id, t.inserted_at)
        t.tenant_id,
        t.id,
        t.inserted_at,
        t.readable_status,
        e.retry_count,
        e.max_readable_status
    FROM
        v1_tasks_olap t
    JOIN
        updatable_events e ON
            (t.tenant_id, t.id, t.inserted_at) = (e.tenant_id, e.task_id, e.task_inserted_at)
    WHERE
        t.inserted_at >= $4::TIMESTAMPTZ
    ORDER BY
        t.tenant_id, t.id, t.inserted_at, t.readable_status
), locked_tasks AS (
    SELECT
        dt.tenant_id, dt.id, dt.inserted_at, dt.readable_status, dt.retry_count, dt.max_readable_status
    FROM
        v1_tasks_olap t
    JOIN
        distinct_tasks_to_lock dt ON
            (dt.inserted_at, dt.id) = (t.inserted_at, t.id)
    FOR UPDATE
), already_in_target_partition AS (
    -- Check if rows already exist in the target partition (with the new readable_status)
    -- This is to avoid rare cases of duplicates writes causing unique constraint violations,
    -- when i.e. we write to one partition of QUEUED, that gets moved from QUEUED -> COMPLETED,
    -- and we later insert into QUEUED again but try to update to COMPLETED again.
    SELECT
        lt.tenant_id,
        lt.id,
        lt.inserted_at,
        lt.readable_status AS old_readable_status,
        lt.retry_count,
        lt.max_readable_status,
        t.external_id,
        t.latest_worker_id,
        t.workflow_id,
        (t.dag_id IS NOT NULL)::boolean AS is_dag_task
    FROM
        locked_tasks lt
    JOIN
        v1_tasks_olap t ON
            t.inserted_at = lt.inserted_at
            AND t.id = lt.id
            AND t.readable_status = lt.max_readable_status
    WHERE
        -- Only consider rows where we would actually change the status
        lt.readable_status != lt.max_readable_status
), deleted_duplicate_tasks AS (
    -- Delete the old rows that already have a copy in the target partition
    DELETE FROM
        v1_tasks_olap t
    USING
        already_in_target_partition ap
    WHERE
        (t.inserted_at, t.id, t.readable_status) = (ap.inserted_at, ap.id, ap.old_readable_status)
), tasks_to_update AS (
    -- Tasks that need updating and don't already exist in target partition
    SELECT
        lt.tenant_id,
        lt.id,
        lt.inserted_at,
        lt.readable_status,
        lt.retry_count,
        lt.max_readable_status
    FROM
        locked_tasks lt
    WHERE
        NOT EXISTS (
            SELECT 1
            FROM already_in_target_partition ap
            WHERE (ap.tenant_id, ap.id, ap.inserted_at) = (lt.tenant_id, lt.id, lt.inserted_at)
        )
), updated_tasks AS (
    UPDATE
        v1_tasks_olap t
    SET
        readable_status = tu.max_readable_status,
        latest_retry_count = tu.retry_count,
        latest_worker_id = CASE WHEN lw.worker_id::uuid IS NOT NULL THEN lw.worker_id::uuid ELSE t.latest_worker_id END
    FROM
        tasks_to_update tu
    LEFT JOIN
        latest_worker_id lw ON
            (tu.tenant_id, tu.id, tu.inserted_at, tu.retry_count) = (lw.tenant_id, lw.task_id, lw.task_inserted_at, lw.retry_count)
    WHERE
        (t.inserted_at, t.id, t.readable_status) = (tu.inserted_at, tu.id, tu.readable_status)
        AND
            (
                -- A newer retry count must always apply, even when the readable
                -- status is unchanged, because this is the only chance to record it:
                -- the deleted_events CTE below removes every processed event from
                -- v1_task_events_olap_tmp regardless of whether this UPDATE matched,
                -- so an event rejected here is never seen again. If a same-status
                -- event from a newer retry were rejected (a replayed task's COMPLETED
                -- at retry N ingested before its QUEUED/ASSIGNED events),
                -- latest_retry_count would stay stale, and the late QUEUED/ASSIGNED
                -- events at retry N would later pass this branch and leave the task
                -- stuck at RUNNING.
                (
                    tu.retry_count > t.latest_retry_count
                ) OR
                -- if the retry count is equal to the latest retry count, update the status if the priority is higher
                (
                    tu.retry_count = t.latest_retry_count
                    AND v1_status_to_priority(tu.max_readable_status) > v1_status_to_priority(t.readable_status)
                ) OR
                -- EVICTED is non-terminal and reversible (durable restore moves it back to RUNNING)
                (
                    tu.retry_count = t.latest_retry_count
                    AND t.readable_status = 'EVICTED'
                    AND tu.max_readable_status != 'EVICTED'
                )
            )
    RETURNING
        t.tenant_id, t.id, t.inserted_at, t.readable_status, t.external_id, t.latest_worker_id, t.workflow_id, (t.dag_id IS NOT NULL)::boolean AS is_dag_task
), all_result_tasks AS (
    -- Combine actually updated tasks with those already in target partition
    SELECT tenant_id, id, inserted_at, readable_status, external_id, latest_worker_id, workflow_id, is_dag_task
    FROM updated_tasks
    UNION ALL
    SELECT tenant_id, id, inserted_at, max_readable_status AS readable_status, external_id, latest_worker_id, workflow_id, is_dag_task
    FROM already_in_target_partition
), events_to_requeue AS (
    -- Get events which don't have a corresponding locked_task
    SELECT
        e.tenant_id,
        e.requeue_retries,
        e.task_id,
        e.task_inserted_at,
        e.event_type,
        e.readable_status,
        e.retry_count
    FROM
        locked_events e
    WHERE NOT EXISTS (
        SELECT 1
        FROM locked_tasks t
        WHERE (e.tenant_id, e.task_id, e.task_inserted_at) = (t.tenant_id, t.id, t.inserted_at)
    )
), deleted_events AS (
    DELETE FROM
        v1_task_events_olap_tmp
    WHERE
        (tenant_id, requeue_after, task_id, id) IN (SELECT tenant_id, requeue_after, task_id, id FROM locked_events)
), requeued_events AS (
    INSERT INTO
        v1_task_events_olap_tmp (
            tenant_id,
            requeue_after,
            requeue_retries,
            task_id,
            task_inserted_at,
            event_type,
            readable_status,
            retry_count
        )
    SELECT
        tenant_id,
        -- Exponential backoff, we limit to 10 retries which is 2048 seconds/34 minutes
        CURRENT_TIMESTAMP + (2 ^ requeue_retries) * INTERVAL '2 seconds',
        requeue_retries + 1,
        task_id,
        task_inserted_at,
        event_type,
        readable_status,
        retry_count
    FROM
        events_to_requeue
    WHERE
        requeue_retries < 10
    RETURNING
        tenant_id, requeue_after, requeue_retries, id, task_id, task_inserted_at, event_type, readable_status, retry_count, worker_id
), event_count AS (
    SELECT
        COUNT(*) as count
    FROM
        locked_events
)

SELECT
    -- Little wonky, but we return the count of events that were processed in each row. Potential edge case
    -- where there are no tasks updated with a non-zero count, but this should be very rare and we'll get
    -- updates on the next run.
    (SELECT count FROM event_count) AS count,
    t.tenant_id, t.id, t.inserted_at, t.readable_status, t.external_id, t.latest_worker_id, t.workflow_id, t.is_dag_task
FROM
    all_result_tasks t
`

type UpdateTaskStatusesParams struct {
	Partitionnumber int32              `json:"partitionnumber"`
	Tenantids       []uuid.UUID        `json:"tenantids"`
	Eventlimit      int32              `json:"eventlimit"`
	Mininsertedat   pgtype.Timestamptz `json:"mininsertedat"`
}

type UpdateTaskStatusesRow struct {
	Count          int64                `json:"count"`
	TenantID       uuid.UUID            `json:"tenant_id"`
	ID             int64                `json:"id"`
	InsertedAt     pgtype.Timestamptz   `json:"inserted_at"`
	ReadableStatus V1ReadableStatusOlap `json:"readable_status"`
	ExternalID     uuid.UUID            `json:"external_id"`
	LatestWorkerID *uuid.UUID           `json:"latest_worker_id"`
	WorkflowID     uuid.UUID            `json:"workflow_id"`
	IsDagTask      bool                 `json:"is_dag_task"`
}

func (q *Queries) UpdateTaskStatuses(ctx context.Context, db DBTX, arg UpdateTaskStatusesParams) ([]*UpdateTaskStatusesRow, error) {
	rows, err := db.Query(ctx, updateTaskStatuses,
		arg.Partitionnumber,
		arg.Tenantids,
		arg.Eventlimit,
		arg.Mininsertedat,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*UpdateTaskStatusesRow
	for rows.Next() {
		var i UpdateTaskStatusesRow
		if err := rows.Scan(
			&i.Count,
			&i.TenantID,
			&i.ID,
			&i.InsertedAt,
			&i.ReadableStatus,
			&i.ExternalID,
			&i.LatestWorkerID,
			&i.WorkflowID,
			&i.IsDagTask,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const updateTaskStatusesFromMQ = `-- name: UpdateTaskStatusesFromMQ :many
WITH inputs AS (
    SELECT
        UNNEST($1::UUID[]) AS tenant_id,
        UNNEST($2::BIGINT[]) AS task_id,
        UNNEST($3::TIMESTAMPTZ[]) AS task_inserted_at,
        UNNEST($4::v1_readable_status_olap[]) AS readable_status,
        UNNEST($5::UUID[]) AS worker_id,
        UNNEST($6::INTEGER[]) AS retry_count
), locked_tasks AS (
    SELECT t.tenant_id, t.id, t.inserted_at, t.external_id, t.queue, t.action_id, t.step_id, t.workflow_id, t.workflow_version_id, t.workflow_run_id, t.schedule_timeout, t.step_timeout, t.priority, t.sticky, t.desired_worker_id, t.display_name, t.input, t.additional_metadata, t.readable_status, t.latest_retry_count, t.latest_worker_id, t.dag_id, t.dag_inserted_at, t.parent_task_external_id, t.is_durable, t.idempotency_key, i.readable_status AS new_readable_status, i.worker_id AS new_worker_id, i.retry_count AS new_retry_count
    FROM v1_tasks_olap t
    JOIN inputs i ON (i.tenant_id, i.task_id, i.task_inserted_at) = (t.tenant_id, t.id, t.inserted_at)
    WHERE
        (
            -- A newer retry count must always apply, even when the readable
            -- status is unchanged, because this is the only chance to record it:
            -- the message this batch came from is acked once the transaction
            -- commits, regardless of whether this update matched, so an event
            -- rejected here is never seen again. If a same-status event from a
            -- newer retry were rejected (a replayed task's COMPLETED at retry N
            -- ingested before its QUEUED/ASSIGNED events), latest_retry_count
            -- would stay stale, and the late QUEUED/ASSIGNED events at retry N
            -- would later pass this branch and leave the task stuck at RUNNING.
            (
                i.retry_count > t.latest_retry_count
            ) OR
            -- If the retry count is equal, only update if the new status has higher priority
            (
                i.retry_count = t.latest_retry_count
                AND v1_status_to_priority(i.readable_status) > v1_status_to_priority(t.readable_status)
            ) OR
            -- EVICTED is non-terminal and reversible (durable restore moves it back to RUNNING)
            (
                i.retry_count = t.latest_retry_count
                AND t.readable_status = 'EVICTED'
                AND i.readable_status != 'EVICTED'
            )
        )
    ORDER BY t.inserted_at, t.id
    FOR UPDATE OF t
), updated_tasks AS (
    UPDATE v1_tasks_olap t
    SET
        readable_status = lt.new_readable_status,
        latest_retry_count = lt.new_retry_count,
        latest_worker_id = CASE
            WHEN lt.new_worker_id != '00000000-0000-0000-0000-000000000000'::uuid THEN lt.new_worker_id
            ELSE t.latest_worker_id
        END
    FROM locked_tasks lt
    WHERE
        (t.inserted_at, t.id, t.tenant_id) = (lt.inserted_at, lt.id, lt.tenant_id)
    RETURNING t.tenant_id, t.id, t.inserted_at, t.external_id, t.queue, t.action_id, t.step_id, t.workflow_id, t.workflow_version_id, t.workflow_run_id, t.schedule_timeout, t.step_timeout, t.priority, t.sticky, t.desired_worker_id, t.display_name, t.input, t.additional_metadata, t.readable_status, t.latest_retry_count, t.latest_worker_id, t.dag_id, t.dag_inserted_at, t.parent_task_external_id, t.is_durable, t.idempotency_key
)

SELECT
    t.tenant_id,
    t.id AS task_id,
    t.inserted_at AS task_inserted_at,
    COALESCE(ut.readable_status, t.readable_status) AS readable_status,
    t.external_id,
    COALESCE(ut.latest_worker_id, t.latest_worker_id) AS latest_worker_id,
    t.workflow_id,
    t.dag_id,
    t.dag_inserted_at,
    (t.dag_id IS NOT NULL)::BOOLEAN AS is_dag_task,
    (ut.id IS NOT NULL)::BOOLEAN AS was_updated
FROM v1_tasks_olap t
LEFT JOIN updated_tasks ut ON (ut.tenant_id, ut.id, ut.inserted_at) = (t.tenant_id, t.id, t.inserted_at)
WHERE (t.inserted_at, t.id, t.tenant_id) IN (
    SELECT task_inserted_at, task_id, tenant_id
    FROM inputs
)
`

type UpdateTaskStatusesFromMQParams struct {
	Tenantids       []uuid.UUID            `json:"tenantids"`
	Taskids         []int64                `json:"taskids"`
	Taskinsertedats []pgtype.Timestamptz   `json:"taskinsertedats"`
	Statuses        []V1ReadableStatusOlap `json:"statuses"`
	Workerids       []uuid.UUID            `json:"workerids"`
	Retrycounts     []int32                `json:"retrycounts"`
}

type UpdateTaskStatusesFromMQRow struct {
	TenantID       uuid.UUID            `json:"tenant_id"`
	TaskID         int64                `json:"task_id"`
	TaskInsertedAt pgtype.Timestamptz   `json:"task_inserted_at"`
	ReadableStatus V1ReadableStatusOlap `json:"readable_status"`
	ExternalID     uuid.UUID            `json:"external_id"`
	LatestWorkerID *uuid.UUID           `json:"latest_worker_id"`
	WorkflowID     uuid.UUID            `json:"workflow_id"`
	DagID          pgtype.Int8          `json:"dag_id"`
	DagInsertedAt  pgtype.Timestamptz   `json:"dag_inserted_at"`
	IsDagTask      bool                 `json:"is_dag_task"`
	WasUpdated     bool                 `json:"was_updated"`
}

func (q *Queries) UpdateTaskStatusesFromMQ(ctx context.Context, db DBTX, arg UpdateTaskStatusesFromMQParams) ([]*UpdateTaskStatusesFromMQRow, error) {
	rows, err := db.Query(ctx, updateTaskStatusesFromMQ,
		arg.Tenantids,
		arg.Taskids,
		arg.Taskinsertedats,
		arg.Statuses,
		arg.Workerids,
		arg.Retrycounts,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*UpdateTaskStatusesFromMQRow
	for rows.Next() {
		var i UpdateTaskStatusesFromMQRow
		if err := rows.Scan(
			&i.TenantID,
			&i.TaskID,
			&i.TaskInsertedAt,
			&i.ReadableStatus,
			&i.ExternalID,
			&i.LatestWorkerID,
			&i.WorkflowID,
			&i.DagID,
			&i.DagInsertedAt,
			&i.IsDagTask,
			&i.WasUpdated,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}