queries.sql.go

// Code generated by sqlc. DO NOT EDIT.
// versions:
//   sqlc v1.26.0
// source: queries.sql

package sql

import (
	"context"
	"encoding/json"
	"time"

	"github.com/TBD54566975/ftl/backend/controller/leases"
	"github.com/TBD54566975/ftl/backend/schema"
	"github.com/TBD54566975/ftl/internal/model"
	"github.com/alecthomas/types/optional"
	"github.com/google/uuid"
)

const acquireAsyncCall = `-- name: AcquireAsyncCall :one
WITH async_call AS (
  SELECT id
  FROM async_calls
  WHERE state = 'pending' AND scheduled_at <= (NOW() AT TIME ZONE 'utc')
  LIMIT 1
  FOR UPDATE SKIP LOCKED
), lease AS (
  INSERT INTO leases (idempotency_key, key, expires_at)
  VALUES (gen_random_uuid(), '/system/async_call/' || (SELECT id FROM async_call), (NOW() AT TIME ZONE 'utc') + $1::interval)
  RETURNING id, idempotency_key, key, created_at, expires_at, metadata
)
UPDATE async_calls
SET state = 'executing', lease_id = (SELECT id FROM lease)
WHERE id = (SELECT id FROM async_call)
RETURNING
  id AS async_call_id,
  (SELECT idempotency_key FROM lease) AS lease_idempotency_key,
  (SELECT key FROM lease) AS lease_key,
  origin,
  verb,
  request,
  scheduled_at,
  remaining_attempts,
  backoff,
  max_backoff
`

type AcquireAsyncCallRow struct {
	AsyncCallID         int64
	LeaseIdempotencyKey uuid.UUID
	LeaseKey            leases.Key
	Origin              string
	Verb                schema.RefKey
	Request             []byte
	ScheduledAt         time.Time
	RemainingAttempts   int32
	Backoff             time.Duration
	MaxBackoff          time.Duration
}

// Reserve a pending async call for execution, returning the associated lease
// reservation key.
func (q *Queries) AcquireAsyncCall(ctx context.Context, ttl time.Duration) (AcquireAsyncCallRow, error) {
	row := q.db.QueryRow(ctx, acquireAsyncCall, ttl)
	var i AcquireAsyncCallRow
	err := row.Scan(
		&i.AsyncCallID,
		&i.LeaseIdempotencyKey,
		&i.LeaseKey,
		&i.Origin,
		&i.Verb,
		&i.Request,
		&i.ScheduledAt,
		&i.RemainingAttempts,
		&i.Backoff,
		&i.MaxBackoff,
	)
	return i, err
}

const associateArtefactWithDeployment = `-- name: AssociateArtefactWithDeployment :exec
INSERT INTO deployment_artefacts (deployment_id, artefact_id, executable, path)
VALUES ((SELECT id FROM deployments WHERE key = $1::deployment_key), $2, $3, $4)
`

type AssociateArtefactWithDeploymentParams struct {
	Key        model.DeploymentKey
	ArtefactID int64
	Executable bool
	Path       string
}

func (q *Queries) AssociateArtefactWithDeployment(ctx context.Context, arg AssociateArtefactWithDeploymentParams) error {
	_, err := q.db.Exec(ctx, associateArtefactWithDeployment,
		arg.Key,
		arg.ArtefactID,
		arg.Executable,
		arg.Path,
	)
	return err
}

const beginConsumingTopicEvent = `-- name: BeginConsumingTopicEvent :exec
WITH event AS (
  SELECT id, created_at, key, topic_id, payload
  FROM topic_events
  WHERE "key" = $2::topic_event_key
)
UPDATE topic_subscriptions
SET state = 'executing',
    cursor = (SELECT id FROM event)
WHERE key = $1::subscription_key
`

func (q *Queries) BeginConsumingTopicEvent(ctx context.Context, subscription model.SubscriptionKey, event model.TopicEventKey) error {
	_, err := q.db.Exec(ctx, beginConsumingTopicEvent, subscription, event)
	return err
}

const completeEventForSubscription = `-- name: CompleteEventForSubscription :exec
WITH module AS (
  SELECT id
  FROM modules
  WHERE name = $2::TEXT
)
UPDATE topic_subscriptions
SET state = 'idle'
WHERE name = $1::TEXT
      AND module_id = (SELECT id FROM module)
`

func (q *Queries) CompleteEventForSubscription(ctx context.Context, name string, module string) error {
	_, err := q.db.Exec(ctx, completeEventForSubscription, name, module)
	return err
}

const createArtefact = `-- name: CreateArtefact :one
INSERT INTO artefacts (digest, content)
VALUES ($1, $2)
RETURNING id
`

// Create a new artefact and return the artefact ID.
func (q *Queries) CreateArtefact(ctx context.Context, digest []byte, content []byte) (int64, error) {
	row := q.db.QueryRow(ctx, createArtefact, digest, content)
	var id int64
	err := row.Scan(&id)
	return id, err
}

const createAsyncCall = `-- name: CreateAsyncCall :one
INSERT INTO async_calls (verb, origin, request, remaining_attempts, backoff, max_backoff)
VALUES ($1, $2, $3, $4, $5::interval, $6::interval)
RETURNING id
`

type CreateAsyncCallParams struct {
	Verb              schema.RefKey
	Origin            string
	Request           []byte
	RemainingAttempts int32
	Backoff           time.Duration
	MaxBackoff        time.Duration
}

func (q *Queries) CreateAsyncCall(ctx context.Context, arg CreateAsyncCallParams) (int64, error) {
	row := q.db.QueryRow(ctx, createAsyncCall,
		arg.Verb,
		arg.Origin,
		arg.Request,
		arg.RemainingAttempts,
		arg.Backoff,
		arg.MaxBackoff,
	)
	var id int64
	err := row.Scan(&id)
	return id, err
}

const createCronJob = `-- name: CreateCronJob :exec
INSERT INTO cron_jobs (key, deployment_id, module_name, verb, schedule, start_time, next_execution)
  VALUES (
    $1::cron_job_key,
    (SELECT id FROM deployments WHERE key = $2::deployment_key LIMIT 1),
    $3::TEXT,
    $4::TEXT,
    $5::TEXT,
    $6::TIMESTAMPTZ,
    $7::TIMESTAMPTZ)
`

type CreateCronJobParams struct {
	Key           model.CronJobKey
	DeploymentKey model.DeploymentKey
	ModuleName    string
	Verb          string
	Schedule      string
	StartTime     time.Time
	NextExecution time.Time
}

func (q *Queries) CreateCronJob(ctx context.Context, arg CreateCronJobParams) error {
	_, err := q.db.Exec(ctx, createCronJob,
		arg.Key,
		arg.DeploymentKey,
		arg.ModuleName,
		arg.Verb,
		arg.Schedule,
		arg.StartTime,
		arg.NextExecution,
	)
	return err
}

const createDeployment = `-- name: CreateDeployment :exec
INSERT INTO deployments (module_id, "schema", "key")
VALUES ((SELECT id FROM modules WHERE name = $1::TEXT LIMIT 1), $2::BYTEA, $3::deployment_key)
`

func (q *Queries) CreateDeployment(ctx context.Context, moduleName string, schema []byte, key model.DeploymentKey) error {
	_, err := q.db.Exec(ctx, createDeployment, moduleName, schema, key)
	return err
}

const createIngressRoute = `-- name: CreateIngressRoute :exec
INSERT INTO ingress_routes (deployment_id, module, verb, method, path)
VALUES ((SELECT id FROM deployments WHERE key = $1::deployment_key LIMIT 1), $2, $3, $4, $5)
`

type CreateIngressRouteParams struct {
	Key    model.DeploymentKey
	Module string
	Verb   string
	Method string
	Path   string
}

func (q *Queries) CreateIngressRoute(ctx context.Context, arg CreateIngressRouteParams) error {
	_, err := q.db.Exec(ctx, createIngressRoute,
		arg.Key,
		arg.Module,
		arg.Verb,
		arg.Method,
		arg.Path,
	)
	return err
}

const createRequest = `-- name: CreateRequest :exec
INSERT INTO requests (origin, "key", source_addr)
VALUES ($1, $2, $3)
`

func (q *Queries) CreateRequest(ctx context.Context, origin Origin, key model.RequestKey, sourceAddr string) error {
	_, err := q.db.Exec(ctx, createRequest, origin, key, sourceAddr)
	return err
}

const deleteSubscribers = `-- name: DeleteSubscribers :exec
DELETE FROM topic_subscribers
WHERE deployment_id IN (
  SELECT deployments.id
  FROM deployments
  WHERE deployments.key = $1::deployment_key
)
`

func (q *Queries) DeleteSubscribers(ctx context.Context, deployment model.DeploymentKey) error {
	_, err := q.db.Exec(ctx, deleteSubscribers, deployment)
	return err
}

const deleteSubscriptions = `-- name: DeleteSubscriptions :exec
DELETE FROM topic_subscriptions
WHERE deployment_id IN (
  SELECT deployments.id
  FROM deployments
  WHERE deployments.key = $1::deployment_key
)
`

func (q *Queries) DeleteSubscriptions(ctx context.Context, deployment model.DeploymentKey) error {
	_, err := q.db.Exec(ctx, deleteSubscriptions, deployment)
	return err
}

const deregisterRunner = `-- name: DeregisterRunner :one
WITH matches AS (
    UPDATE runners
        SET state = 'dead',
            deployment_id = NULL
        WHERE key = $1::runner_key
        RETURNING 1)
SELECT COUNT(*)
FROM matches
`

func (q *Queries) DeregisterRunner(ctx context.Context, key model.RunnerKey) (int64, error) {
	row := q.db.QueryRow(ctx, deregisterRunner, key)
	var count int64
	err := row.Scan(&count)
	return count, err
}

const endCronJob = `-- name: EndCronJob :one
WITH j AS (
UPDATE cron_jobs
  SET state = 'idle',
    next_execution = $1::TIMESTAMPTZ
  WHERE key = $2::cron_job_key
    AND state = 'executing'
    AND start_time = $3::TIMESTAMPTZ
  RETURNING id, key, deployment_id, verb, schedule, start_time, next_execution, state, module_name
)
SELECT j.key as key, d.key as deployment_key, j.module_name as module, j.verb, j.schedule, j.start_time, j.next_execution, j.state
  FROM j
  INNER JOIN deployments d on j.deployment_id = d.id
  LIMIT 1
`

type EndCronJobRow struct {
	Key           model.CronJobKey
	DeploymentKey model.DeploymentKey
	Module        string
	Verb          string
	Schedule      string
	StartTime     time.Time
	NextExecution time.Time
	State         model.CronJobState
}

func (q *Queries) EndCronJob(ctx context.Context, nextExecution time.Time, key model.CronJobKey, startTime time.Time) (EndCronJobRow, error) {
	row := q.db.QueryRow(ctx, endCronJob, nextExecution, key, startTime)
	var i EndCronJobRow
	err := row.Scan(
		&i.Key,
		&i.DeploymentKey,
		&i.Module,
		&i.Verb,
		&i.Schedule,
		&i.StartTime,
		&i.NextExecution,
		&i.State,
	)
	return i, err
}

const expireLeases = `-- name: ExpireLeases :one
WITH expired AS (
    DELETE FROM leases
    WHERE expires_at < NOW() AT TIME ZONE 'utc'
    RETURNING 1
)
SELECT COUNT(*)
FROM expired
`

func (q *Queries) ExpireLeases(ctx context.Context) (int64, error) {
	row := q.db.QueryRow(ctx, expireLeases)
	var count int64
	err := row.Scan(&count)
	return count, err
}

const expireRunnerReservations = `-- name: ExpireRunnerReservations :one
WITH rows AS (
    UPDATE runners
        SET state = 'idle',
            deployment_id = NULL,
            reservation_timeout = NULL
        WHERE state = 'reserved'
            AND reservation_timeout < (NOW() AT TIME ZONE 'utc')
        RETURNING 1)
SELECT COUNT(*)
FROM rows
`

func (q *Queries) ExpireRunnerReservations(ctx context.Context) (int64, error) {
	row := q.db.QueryRow(ctx, expireRunnerReservations)
	var count int64
	err := row.Scan(&count)
	return count, err
}

const failAsyncCall = `-- name: FailAsyncCall :one
UPDATE async_calls
SET
  state = 'error'::async_call_state,
  error = $1::TEXT
WHERE id = $2
RETURNING true
`

func (q *Queries) FailAsyncCall(ctx context.Context, error string, iD int64) (bool, error) {
	row := q.db.QueryRow(ctx, failAsyncCall, error, iD)
	var column_1 bool
	err := row.Scan(&column_1)
	return column_1, err
}

const failAsyncCallWithRetry = `-- name: FailAsyncCallWithRetry :one
WITH updated AS (
  UPDATE async_calls
  SET state = 'error'::async_call_state,
      error = $5::TEXT
  WHERE id = $6::BIGINT
  RETURNING id, created_at, lease_id, verb, state, origin, scheduled_at, request, response, error, remaining_attempts, backoff, max_backoff
)
INSERT INTO async_calls (verb, origin, request, remaining_attempts, backoff, max_backoff, scheduled_at)
SELECT updated.verb, updated.origin, updated.request, $1, $2::interval, $3::interval, $4::TIMESTAMPTZ
  FROM updated
  RETURNING true
`

type FailAsyncCallWithRetryParams struct {
	RemainingAttempts int32
	Backoff           time.Duration
	MaxBackoff        time.Duration
	ScheduledAt       time.Time
	Error             string
	ID                int64
}

func (q *Queries) FailAsyncCallWithRetry(ctx context.Context, arg FailAsyncCallWithRetryParams) (bool, error) {
	row := q.db.QueryRow(ctx, failAsyncCallWithRetry,
		arg.RemainingAttempts,
		arg.Backoff,
		arg.MaxBackoff,
		arg.ScheduledAt,
		arg.Error,
		arg.ID,
	)
	var column_1 bool
	err := row.Scan(&column_1)
	return column_1, err
}

const failFSMInstance = `-- name: FailFSMInstance :one
UPDATE fsm_instances
SET
  current_state = NULL,
  async_call_id = NULL,
  status = 'failed'::fsm_status
WHERE
  fsm = $1::schema_ref AND key = $2::TEXT
RETURNING true
`

func (q *Queries) FailFSMInstance(ctx context.Context, fsm schema.RefKey, key string) (bool, error) {
	row := q.db.QueryRow(ctx, failFSMInstance, fsm, key)
	var column_1 bool
	err := row.Scan(&column_1)
	return column_1, err
}

const finishFSMTransition = `-- name: FinishFSMTransition :one
UPDATE fsm_instances
SET
  current_state = destination_state,
  destination_state = NULL,
  async_call_id = NULL
WHERE
  fsm = $1::schema_ref AND key = $2::TEXT
RETURNING true
`

// Mark an FSM transition as completed, updating the current state and clearing the async call ID.
func (q *Queries) FinishFSMTransition(ctx context.Context, fsm schema.RefKey, key string) (bool, error) {
	row := q.db.QueryRow(ctx, finishFSMTransition, fsm, key)
	var column_1 bool
	err := row.Scan(&column_1)
	return column_1, err
}

const getActiveControllers = `-- name: GetActiveControllers :many
SELECT id, key, created, last_seen, state, endpoint
FROM controller c
WHERE c.state <> 'dead'
ORDER BY c.key
`

func (q *Queries) GetActiveControllers(ctx context.Context) ([]Controller, error) {
	rows, err := q.db.Query(ctx, getActiveControllers)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []Controller
	for rows.Next() {
		var i Controller
		if err := rows.Scan(
			&i.ID,
			&i.Key,
			&i.Created,
			&i.LastSeen,
			&i.State,
			&i.Endpoint,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getActiveDeploymentSchemas = `-- name: GetActiveDeploymentSchemas :many
SELECT key, schema FROM deployments WHERE min_replicas > 0
`

type GetActiveDeploymentSchemasRow struct {
	Key    model.DeploymentKey
	Schema *schema.Module
}

func (q *Queries) GetActiveDeploymentSchemas(ctx context.Context) ([]GetActiveDeploymentSchemasRow, error) {
	rows, err := q.db.Query(ctx, getActiveDeploymentSchemas)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetActiveDeploymentSchemasRow
	for rows.Next() {
		var i GetActiveDeploymentSchemasRow
		if err := rows.Scan(&i.Key, &i.Schema); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getActiveDeployments = `-- name: GetActiveDeployments :many
SELECT d.id, d.created_at, d.module_id, d.key, d.schema, d.labels, d.min_replicas, m.name AS module_name, m.language, COUNT(r.id) AS replicas
FROM deployments d
  JOIN modules m ON d.module_id = m.id
  JOIN runners r ON d.id = r.deployment_id
WHERE min_replicas > 0 AND r.state = 'assigned'
GROUP BY d.id, m.name, m.language
HAVING COUNT(r.id) > 0
`

type GetActiveDeploymentsRow struct {
	Deployment Deployment
	ModuleName string
	Language   string
	Replicas   int64
}

func (q *Queries) GetActiveDeployments(ctx context.Context) ([]GetActiveDeploymentsRow, error) {
	rows, err := q.db.Query(ctx, getActiveDeployments)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetActiveDeploymentsRow
	for rows.Next() {
		var i GetActiveDeploymentsRow
		if err := rows.Scan(
			&i.Deployment.ID,
			&i.Deployment.CreatedAt,
			&i.Deployment.ModuleID,
			&i.Deployment.Key,
			&i.Deployment.Schema,
			&i.Deployment.Labels,
			&i.Deployment.MinReplicas,
			&i.ModuleName,
			&i.Language,
			&i.Replicas,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getActiveIngressRoutes = `-- name: GetActiveIngressRoutes :many
SELECT d.key AS deployment_key, ir.module, ir.verb, ir.method, ir.path
FROM ingress_routes ir
         INNER JOIN deployments d ON ir.deployment_id = d.id
WHERE d.min_replicas > 0
`

type GetActiveIngressRoutesRow struct {
	DeploymentKey model.DeploymentKey
	Module        string
	Verb          string
	Method        string
	Path          string
}

func (q *Queries) GetActiveIngressRoutes(ctx context.Context) ([]GetActiveIngressRoutesRow, error) {
	rows, err := q.db.Query(ctx, getActiveIngressRoutes)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetActiveIngressRoutesRow
	for rows.Next() {
		var i GetActiveIngressRoutesRow
		if err := rows.Scan(
			&i.DeploymentKey,
			&i.Module,
			&i.Verb,
			&i.Method,
			&i.Path,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getActiveRunners = `-- name: GetActiveRunners :many
SELECT DISTINCT ON (r.key) r.key                                   AS runner_key,
                           r.endpoint,
                           r.state,
                           r.labels,
                           r.last_seen,
                           r.module_name,
                           COALESCE(CASE
                                        WHEN r.deployment_id IS NOT NULL
                                            THEN d.key END, NULL) AS deployment_key
FROM runners r
         LEFT JOIN deployments d on d.id = r.deployment_id
WHERE r.state <> 'dead'
ORDER BY r.key
`

type GetActiveRunnersRow struct {
	RunnerKey     model.RunnerKey
	Endpoint      string
	State         RunnerState
	Labels        []byte
	LastSeen      time.Time
	ModuleName    optional.Option[string]
	DeploymentKey optional.Option[string]
}

func (q *Queries) GetActiveRunners(ctx context.Context) ([]GetActiveRunnersRow, error) {
	rows, err := q.db.Query(ctx, getActiveRunners)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetActiveRunnersRow
	for rows.Next() {
		var i GetActiveRunnersRow
		if err := rows.Scan(
			&i.RunnerKey,
			&i.Endpoint,
			&i.State,
			&i.Labels,
			&i.LastSeen,
			&i.ModuleName,
			&i.DeploymentKey,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getArtefactContentRange = `-- name: GetArtefactContentRange :one
SELECT SUBSTRING(a.content FROM $1 FOR $2)::BYTEA AS content
FROM artefacts a
WHERE a.id = $3
`

func (q *Queries) GetArtefactContentRange(ctx context.Context, start int32, count int32, iD int64) ([]byte, error) {
	row := q.db.QueryRow(ctx, getArtefactContentRange, start, count, iD)
	var content []byte
	err := row.Scan(&content)
	return content, err
}

const getArtefactDigests = `-- name: GetArtefactDigests :many
SELECT id, digest
FROM artefacts
WHERE digest = ANY ($1::bytea[])
`

type GetArtefactDigestsRow struct {
	ID     int64
	Digest []byte
}

// Return the digests that exist in the database.
func (q *Queries) GetArtefactDigests(ctx context.Context, digests [][]byte) ([]GetArtefactDigestsRow, error) {
	rows, err := q.db.Query(ctx, getArtefactDigests, digests)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetArtefactDigestsRow
	for rows.Next() {
		var i GetArtefactDigestsRow
		if err := rows.Scan(&i.ID, &i.Digest); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getCronJobs = `-- name: GetCronJobs :many
SELECT j.key as key, d.key as deployment_key, j.module_name as module, j.verb, j.schedule, j.start_time, j.next_execution, j.state
FROM cron_jobs j
  INNER JOIN deployments d on j.deployment_id = d.id
WHERE d.min_replicas > 0
`

type GetCronJobsRow struct {
	Key           model.CronJobKey
	DeploymentKey model.DeploymentKey
	Module        string
	Verb          string
	Schedule      string
	StartTime     time.Time
	NextExecution time.Time
	State         model.CronJobState
}

func (q *Queries) GetCronJobs(ctx context.Context) ([]GetCronJobsRow, error) {
	rows, err := q.db.Query(ctx, getCronJobs)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetCronJobsRow
	for rows.Next() {
		var i GetCronJobsRow
		if err := rows.Scan(
			&i.Key,
			&i.DeploymentKey,
			&i.Module,
			&i.Verb,
			&i.Schedule,
			&i.StartTime,
			&i.NextExecution,
			&i.State,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getDeployment = `-- name: GetDeployment :one
SELECT d.id, d.created_at, d.module_id, d.key, d.schema, d.labels, d.min_replicas, m.language, m.name AS module_name, d.min_replicas
FROM deployments d
         INNER JOIN modules m ON m.id = d.module_id
WHERE d.key = $1::deployment_key
`

type GetDeploymentRow struct {
	Deployment  Deployment
	Language    string
	ModuleName  string
	MinReplicas int32
}

func (q *Queries) GetDeployment(ctx context.Context, key model.DeploymentKey) (GetDeploymentRow, error) {
	row := q.db.QueryRow(ctx, getDeployment, key)
	var i GetDeploymentRow
	err := row.Scan(
		&i.Deployment.ID,
		&i.Deployment.CreatedAt,
		&i.Deployment.ModuleID,
		&i.Deployment.Key,
		&i.Deployment.Schema,
		&i.Deployment.Labels,
		&i.Deployment.MinReplicas,
		&i.Language,
		&i.ModuleName,
		&i.MinReplicas,
	)
	return i, err
}

const getDeploymentArtefacts = `-- name: GetDeploymentArtefacts :many
SELECT da.created_at, artefact_id AS id, executable, path, digest, executable
FROM deployment_artefacts da
         INNER JOIN artefacts ON artefacts.id = da.artefact_id
WHERE deployment_id = $1
`

type GetDeploymentArtefactsRow struct {
	CreatedAt    time.Time
	ID           int64
	Executable   bool
	Path         string
	Digest       []byte
	Executable_2 bool
}

// Get all artefacts matching the given digests.
func (q *Queries) GetDeploymentArtefacts(ctx context.Context, deploymentID int64) ([]GetDeploymentArtefactsRow, error) {
	rows, err := q.db.Query(ctx, getDeploymentArtefacts, deploymentID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetDeploymentArtefactsRow
	for rows.Next() {
		var i GetDeploymentArtefactsRow
		if err := rows.Scan(
			&i.CreatedAt,
			&i.ID,
			&i.Executable,
			&i.Path,
			&i.Digest,
			&i.Executable_2,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getDeploymentsByID = `-- name: GetDeploymentsByID :many
SELECT id, created_at, module_id, key, schema, labels, min_replicas
FROM deployments
WHERE id = ANY ($1::BIGINT[])
`

func (q *Queries) GetDeploymentsByID(ctx context.Context, ids []int64) ([]Deployment, error) {
	rows, err := q.db.Query(ctx, getDeploymentsByID, ids)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []Deployment
	for rows.Next() {
		var i Deployment
		if err := rows.Scan(
			&i.ID,
			&i.CreatedAt,
			&i.ModuleID,
			&i.Key,
			&i.Schema,
			&i.Labels,
			&i.MinReplicas,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getDeploymentsNeedingReconciliation = `-- name: GetDeploymentsNeedingReconciliation :many
SELECT d.key                 AS deployment_key,
       m.name                 AS module_name,
       m.language             AS language,
       COUNT(r.id)            AS assigned_runners_count,
       d.min_replicas::BIGINT AS required_runners_count
FROM deployments d
         LEFT JOIN runners r ON d.id = r.deployment_id AND r.state <> 'dead'
         JOIN modules m ON d.module_id = m.id
GROUP BY d.key, d.min_replicas, m.name, m.language
HAVING COUNT(r.id) <> d.min_replicas
`

type GetDeploymentsNeedingReconciliationRow struct {
	DeploymentKey        model.DeploymentKey
	ModuleName           string
	Language             string
	AssignedRunnersCount int64
	RequiredRunnersCount int64
}

// Get deployments that have a mismatch between the number of assigned and required replicas.
func (q *Queries) GetDeploymentsNeedingReconciliation(ctx context.Context) ([]GetDeploymentsNeedingReconciliationRow, error) {
	rows, err := q.db.Query(ctx, getDeploymentsNeedingReconciliation)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetDeploymentsNeedingReconciliationRow
	for rows.Next() {
		var i GetDeploymentsNeedingReconciliationRow
		if err := rows.Scan(
			&i.DeploymentKey,
			&i.ModuleName,
			&i.Language,
			&i.AssignedRunnersCount,
			&i.RequiredRunnersCount,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getDeploymentsWithArtefacts = `-- name: GetDeploymentsWithArtefacts :many
SELECT d.id, d.created_at, d.key as deployment_key, d.schema, m.name AS module_name
FROM deployments d
         INNER JOIN modules m ON d.module_id = m.id
WHERE EXISTS (SELECT 1
              FROM deployment_artefacts da
                       INNER JOIN artefacts a ON da.artefact_id = a.id
              WHERE a.digest = ANY ($1::bytea[])
                AND da.deployment_id = d.id
                AND d.schema = $2::BYTEA
              HAVING COUNT(*) = $3::BIGINT -- Number of unique digests provided
)
`

type GetDeploymentsWithArtefactsRow struct {
	ID            int64
	CreatedAt     time.Time
	DeploymentKey model.DeploymentKey
	Schema        *schema.Module
	ModuleName    string
}

// Get all deployments that have artefacts matching the given digests.
func (q *Queries) GetDeploymentsWithArtefacts(ctx context.Context, digests [][]byte, schema []byte, count int64) ([]GetDeploymentsWithArtefactsRow, error) {
	rows, err := q.db.Query(ctx, getDeploymentsWithArtefacts, digests, schema, count)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetDeploymentsWithArtefactsRow
	for rows.Next() {
		var i GetDeploymentsWithArtefactsRow
		if err := rows.Scan(
			&i.ID,
			&i.CreatedAt,
			&i.DeploymentKey,
			&i.Schema,
			&i.ModuleName,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getDeploymentsWithMinReplicas = `-- name: GetDeploymentsWithMinReplicas :many
SELECT d.id, d.created_at, d.module_id, d.key, d.schema, d.labels, d.min_replicas, m.name AS module_name, m.language
FROM deployments d
  INNER JOIN modules m on d.module_id = m.id
WHERE min_replicas > 0
ORDER BY d.key
`

type GetDeploymentsWithMinReplicasRow struct {
	Deployment Deployment
	ModuleName string
	Language   string
}

func (q *Queries) GetDeploymentsWithMinReplicas(ctx context.Context) ([]GetDeploymentsWithMinReplicasRow, error) {
	rows, err := q.db.Query(ctx, getDeploymentsWithMinReplicas)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetDeploymentsWithMinReplicasRow
	for rows.Next() {
		var i GetDeploymentsWithMinReplicasRow
		if err := rows.Scan(
			&i.Deployment.ID,
			&i.Deployment.CreatedAt,
			&i.Deployment.ModuleID,
			&i.Deployment.Key,
			&i.Deployment.Schema,
			&i.Deployment.Labels,
			&i.Deployment.MinReplicas,
			&i.ModuleName,
			&i.Language,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getExistingDeploymentForModule = `-- name: GetExistingDeploymentForModule :one
SELECT d.id, created_at, module_id, key, schema, labels, min_replicas, m.id, language, name
FROM deployments d
         INNER JOIN modules m on d.module_id = m.id
WHERE m.name = $1
  AND min_replicas > 0
LIMIT 1
`

type GetExistingDeploymentForModuleRow struct {
	ID          int64
	CreatedAt   time.Time
	ModuleID    int64
	Key         model.DeploymentKey
	Schema      *schema.Module
	Labels      []byte
	MinReplicas int32
	ID_2        int64
	Language    string
	Name        string
}

func (q *Queries) GetExistingDeploymentForModule(ctx context.Context, name string) (GetExistingDeploymentForModuleRow, error) {
	row := q.db.QueryRow(ctx, getExistingDeploymentForModule, name)
	var i GetExistingDeploymentForModuleRow
	err := row.Scan(
		&i.ID,
		&i.CreatedAt,
		&i.ModuleID,
		&i.Key,
		&i.Schema,
		&i.Labels,
		&i.MinReplicas,
		&i.ID_2,
		&i.Language,
		&i.Name,
	)
	return i, err
}

const getFSMInstance = `-- name: GetFSMInstance :one
SELECT id, created_at, fsm, key, status, current_state, destination_state, async_call_id
FROM fsm_instances
WHERE fsm = $1::schema_ref AND key = $2
`

func (q *Queries) GetFSMInstance(ctx context.Context, fsm schema.RefKey, key string) (FsmInstance, error) {
	row := q.db.QueryRow(ctx, getFSMInstance, fsm, key)
	var i FsmInstance
	err := row.Scan(
		&i.ID,
		&i.CreatedAt,
		&i.Fsm,
		&i.Key,
		&i.Status,
		&i.CurrentState,
		&i.DestinationState,
		&i.AsyncCallID,
	)
	return i, err
}

const getIdleRunners = `-- name: GetIdleRunners :many
SELECT id, key, created, last_seen, reservation_timeout, state, endpoint, module_name, deployment_id, labels
FROM runners
WHERE labels @> $1::jsonb
  AND state = 'idle'
LIMIT $2
`

func (q *Queries) GetIdleRunners(ctx context.Context, labels []byte, limit int64) ([]Runner, error) {
	rows, err := q.db.Query(ctx, getIdleRunners, labels, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []Runner
	for rows.Next() {
		var i Runner
		if err := rows.Scan(
			&i.ID,
			&i.Key,
			&i.Created,
			&i.LastSeen,
			&i.ReservationTimeout,
			&i.State,
			&i.Endpoint,
			&i.ModuleName,
			&i.DeploymentID,
			&i.Labels,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getIngressRoutes = `-- name: GetIngressRoutes :many
SELECT r.key AS runner_key, d.key AS deployment_key, endpoint, ir.path, ir.module, ir.verb
FROM ingress_routes ir
         INNER JOIN runners r ON ir.deployment_id = r.deployment_id
         INNER JOIN deployments d ON ir.deployment_id = d.id
WHERE r.state = 'assigned'
  AND ir.method = $1
`

type GetIngressRoutesRow struct {
	RunnerKey     model.RunnerKey
	DeploymentKey model.DeploymentKey
	Endpoint      string
	Path          string
	Module        string
	Verb          string
}

// Get the runner endpoints corresponding to the given ingress route.
func (q *Queries) GetIngressRoutes(ctx context.Context, method string) ([]GetIngressRoutesRow, error) {
	rows, err := q.db.Query(ctx, getIngressRoutes, method)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetIngressRoutesRow
	for rows.Next() {
		var i GetIngressRoutesRow
		if err := rows.Scan(
			&i.RunnerKey,
			&i.DeploymentKey,
			&i.Endpoint,
			&i.Path,
			&i.Module,
			&i.Verb,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getLeaseInfo = `-- name: GetLeaseInfo :one
SELECT expires_at, metadata FROM leases WHERE key = $1::lease_key
`

type GetLeaseInfoRow struct {
	ExpiresAt time.Time
	Metadata  []byte
}

func (q *Queries) GetLeaseInfo(ctx context.Context, key leases.Key) (GetLeaseInfoRow, error) {
	row := q.db.QueryRow(ctx, getLeaseInfo, key)
	var i GetLeaseInfoRow
	err := row.Scan(&i.ExpiresAt, &i.Metadata)
	return i, err
}

const getModulesByID = `-- name: GetModulesByID :many
SELECT id, language, name
FROM modules
WHERE id = ANY ($1::BIGINT[])
`

func (q *Queries) GetModulesByID(ctx context.Context, ids []int64) ([]Module, error) {
	rows, err := q.db.Query(ctx, getModulesByID, ids)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []Module
	for rows.Next() {
		var i Module
		if err := rows.Scan(&i.ID, &i.Language, &i.Name); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getNextEventForSubscription = `-- name: GetNextEventForSubscription :one
WITH cursor AS (
  SELECT
    created_at,
    id
  FROM topic_events
  WHERE "key" = $3::topic_event_key
)
SELECT events."key" as event,
        events.payload,
        events.created_at,
        NOW() - events.created_at >= $1::interval AS ready
FROM topics
LEFT JOIN topic_events as events ON events.topic_id = topics.id
WHERE topics.key = $2::topic_key
  AND (events.created_at, events.id) > (SELECT COALESCE(MAX(cursor.created_at), '1900-01-01'), COALESCE(MAX(cursor.id), 0) FROM cursor)
ORDER BY events.created_at, events.id
LIMIT 1
`

type GetNextEventForSubscriptionRow struct {
	Event     optional.Option[model.TopicEventKey]
	Payload   []byte
	CreatedAt optional.Option[time.Time]
	Ready     bool
}

func (q *Queries) GetNextEventForSubscription(ctx context.Context, consumptionDelay time.Duration, topic model.TopicKey, cursor optional.Option[model.TopicEventKey]) (GetNextEventForSubscriptionRow, error) {
	row := q.db.QueryRow(ctx, getNextEventForSubscription, consumptionDelay, topic, cursor)
	var i GetNextEventForSubscriptionRow
	err := row.Scan(
		&i.Event,
		&i.Payload,
		&i.CreatedAt,
		&i.Ready,
	)
	return i, err
}

const getProcessList = `-- name: GetProcessList :many
SELECT d.min_replicas,
       d.key   AS deployment_key,
       d.labels    deployment_labels,
       r.key    AS runner_key,
       r.endpoint,
       r.labels AS runner_labels
FROM deployments d
         LEFT JOIN runners r on d.id = r.deployment_id AND r.state != 'dead'
WHERE d.min_replicas > 0
ORDER BY d.key
`

type GetProcessListRow struct {
	MinReplicas      int32
	DeploymentKey    model.DeploymentKey
	DeploymentLabels []byte
	RunnerKey        optional.Option[model.RunnerKey]
	Endpoint         optional.Option[string]
	RunnerLabels     []byte
}

func (q *Queries) GetProcessList(ctx context.Context) ([]GetProcessListRow, error) {
	rows, err := q.db.Query(ctx, getProcessList)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetProcessListRow
	for rows.Next() {
		var i GetProcessListRow
		if err := rows.Scan(
			&i.MinReplicas,
			&i.DeploymentKey,
			&i.DeploymentLabels,
			&i.RunnerKey,
			&i.Endpoint,
			&i.RunnerLabels,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getRandomSubscriber = `-- name: GetRandomSubscriber :one
SELECT
  subscribers.sink as sink,
  subscribers.retry_attempts as retry_attempts,
  subscribers.backoff as backoff,
  subscribers.max_backoff as max_backoff
FROM topic_subscribers as subscribers
JOIN topic_subscriptions ON subscribers.topic_subscriptions_id = topic_subscriptions.id
WHERE topic_subscriptions.key = $1::subscription_key
ORDER BY RANDOM()
LIMIT 1
`

type GetRandomSubscriberRow struct {
	Sink          schema.RefKey
	RetryAttempts int32
	Backoff       time.Duration
	MaxBackoff    time.Duration
}

func (q *Queries) GetRandomSubscriber(ctx context.Context, key model.SubscriptionKey) (GetRandomSubscriberRow, error) {
	row := q.db.QueryRow(ctx, getRandomSubscriber, key)
	var i GetRandomSubscriberRow
	err := row.Scan(
		&i.Sink,
		&i.RetryAttempts,
		&i.Backoff,
		&i.MaxBackoff,
	)
	return i, err
}

const getRouteForRunner = `-- name: GetRouteForRunner :one
SELECT endpoint, r.key AS runner_key, r.module_name, d.key deployment_key, r.state
FROM runners r
         LEFT JOIN deployments d on r.deployment_id = d.id
WHERE r.key = $1::runner_key
`

type GetRouteForRunnerRow struct {
	Endpoint      string
	RunnerKey     model.RunnerKey
	ModuleName    optional.Option[string]
	DeploymentKey optional.Option[model.DeploymentKey]
	State         RunnerState
}

// Retrieve routing information for a runner.
func (q *Queries) GetRouteForRunner(ctx context.Context, key model.RunnerKey) (GetRouteForRunnerRow, error) {
	row := q.db.QueryRow(ctx, getRouteForRunner, key)
	var i GetRouteForRunnerRow
	err := row.Scan(
		&i.Endpoint,
		&i.RunnerKey,
		&i.ModuleName,
		&i.DeploymentKey,
		&i.State,
	)
	return i, err
}

const getRoutingTable = `-- name: GetRoutingTable :many
SELECT endpoint, r.key AS runner_key, r.module_name, d.key deployment_key
FROM runners r
         LEFT JOIN deployments d on r.deployment_id = d.id
WHERE state = 'assigned'
  AND (COALESCE(cardinality($1::TEXT[]), 0) = 0
    OR module_name = ANY ($1::TEXT[]))
`

type GetRoutingTableRow struct {
	Endpoint      string
	RunnerKey     model.RunnerKey
	ModuleName    optional.Option[string]
	DeploymentKey optional.Option[model.DeploymentKey]
}

func (q *Queries) GetRoutingTable(ctx context.Context, modules []string) ([]GetRoutingTableRow, error) {
	rows, err := q.db.Query(ctx, getRoutingTable, modules)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetRoutingTableRow
	for rows.Next() {
		var i GetRoutingTableRow
		if err := rows.Scan(
			&i.Endpoint,
			&i.RunnerKey,
			&i.ModuleName,
			&i.DeploymentKey,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getRunner = `-- name: GetRunner :one
SELECT DISTINCT ON (r.key) r.key                                   AS runner_key,
                           r.endpoint,
                           r.state,
                           r.labels,
                           r.last_seen,
                           r.module_name,
                           COALESCE(CASE
                                        WHEN r.deployment_id IS NOT NULL
                                            THEN d.key END, NULL) AS deployment_key
FROM runners r
         LEFT JOIN deployments d on d.id = r.deployment_id OR r.deployment_id IS NULL
WHERE r.key = $1::runner_key
`

type GetRunnerRow struct {
	RunnerKey     model.RunnerKey
	Endpoint      string
	State         RunnerState
	Labels        []byte
	LastSeen      time.Time
	ModuleName    optional.Option[string]
	DeploymentKey optional.Option[string]
}

func (q *Queries) GetRunner(ctx context.Context, key model.RunnerKey) (GetRunnerRow, error) {
	row := q.db.QueryRow(ctx, getRunner, key)
	var i GetRunnerRow
	err := row.Scan(
		&i.RunnerKey,
		&i.Endpoint,
		&i.State,
		&i.Labels,
		&i.LastSeen,
		&i.ModuleName,
		&i.DeploymentKey,
	)
	return i, err
}

const getRunnerState = `-- name: GetRunnerState :one
SELECT state
FROM runners
WHERE key = $1::runner_key
`

func (q *Queries) GetRunnerState(ctx context.Context, key model.RunnerKey) (RunnerState, error) {
	row := q.db.QueryRow(ctx, getRunnerState, key)
	var state RunnerState
	err := row.Scan(&state)
	return state, err
}

const getRunnersForDeployment = `-- name: GetRunnersForDeployment :many
SELECT r.id, r.key, created, last_seen, reservation_timeout, state, endpoint, module_name, deployment_id, r.labels, d.id, created_at, module_id, d.key, schema, d.labels, min_replicas
FROM runners r
         INNER JOIN deployments d on r.deployment_id = d.id
WHERE state = 'assigned'
  AND d.key = $1::deployment_key
`

type GetRunnersForDeploymentRow struct {
	ID                 int64
	Key                model.RunnerKey
	Created            time.Time
	LastSeen           time.Time
	ReservationTimeout optional.Option[time.Time]
	State              RunnerState
	Endpoint           string
	ModuleName         optional.Option[string]
	DeploymentID       optional.Option[int64]
	Labels             []byte
	ID_2               int64
	CreatedAt          time.Time
	ModuleID           int64
	Key_2              model.DeploymentKey
	Schema             *schema.Module
	Labels_2           []byte
	MinReplicas        int32
}

func (q *Queries) GetRunnersForDeployment(ctx context.Context, key model.DeploymentKey) ([]GetRunnersForDeploymentRow, error) {
	rows, err := q.db.Query(ctx, getRunnersForDeployment, key)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetRunnersForDeploymentRow
	for rows.Next() {
		var i GetRunnersForDeploymentRow
		if err := rows.Scan(
			&i.ID,
			&i.Key,
			&i.Created,
			&i.LastSeen,
			&i.ReservationTimeout,
			&i.State,
			&i.Endpoint,
			&i.ModuleName,
			&i.DeploymentID,
			&i.Labels,
			&i.ID_2,
			&i.CreatedAt,
			&i.ModuleID,
			&i.Key_2,
			&i.Schema,
			&i.Labels_2,
			&i.MinReplicas,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getSchemaForDeployment = `-- name: GetSchemaForDeployment :one
SELECT schema FROM deployments WHERE key = $1::deployment_key
`

func (q *Queries) GetSchemaForDeployment(ctx context.Context, key model.DeploymentKey) (*schema.Module, error) {
	row := q.db.QueryRow(ctx, getSchemaForDeployment, key)
	var schema *schema.Module
	err := row.Scan(&schema)
	return schema, err
}

const getStaleCronJobs = `-- name: GetStaleCronJobs :many
SELECT j.key as key, d.key as deployment_key, j.module_name as module, j.verb, j.schedule, j.start_time, j.next_execution, j.state
FROM cron_jobs j
  INNER JOIN deployments d on j.deployment_id = d.id
WHERE state = 'executing'
  AND start_time < (NOW() AT TIME ZONE 'utc') - $1::INTERVAL
`

type GetStaleCronJobsRow struct {
	Key           model.CronJobKey
	DeploymentKey model.DeploymentKey
	Module        string
	Verb          string
	Schedule      string
	StartTime     time.Time
	NextExecution time.Time
	State         model.CronJobState
}

func (q *Queries) GetStaleCronJobs(ctx context.Context, dollar_1 time.Duration) ([]GetStaleCronJobsRow, error) {
	rows, err := q.db.Query(ctx, getStaleCronJobs, dollar_1)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetStaleCronJobsRow
	for rows.Next() {
		var i GetStaleCronJobsRow
		if err := rows.Scan(
			&i.Key,
			&i.DeploymentKey,
			&i.Module,
			&i.Verb,
			&i.Schedule,
			&i.StartTime,
			&i.NextExecution,
			&i.State,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const getSubscriptionsNeedingUpdate = `-- name: GetSubscriptionsNeedingUpdate :many
SELECT
  subs.key::subscription_key as key,
  curser.key as cursor,
  topics.key::topic_key as topic,
  subs.name
FROM topic_subscriptions subs
LEFT JOIN topics ON subs.topic_id = topics.id
LEFT JOIN topic_events curser ON subs.cursor = curser.id
WHERE subs.cursor IS DISTINCT FROM topics.head
  AND subs.state = 'idle'
ORDER BY curser.created_at
LIMIT 3
FOR UPDATE OF subs SKIP LOCKED
`

type GetSubscriptionsNeedingUpdateRow struct {
	Key    model.SubscriptionKey
	Cursor optional.Option[model.TopicEventKey]
	Topic  model.TopicKey
	Name   string
}

// Results may not be ready to be scheduled yet due to event consumption delay
// Sorting ensures that brand new events (that may not be ready for consumption)
// don't prevent older events from being consumed
func (q *Queries) GetSubscriptionsNeedingUpdate(ctx context.Context) ([]GetSubscriptionsNeedingUpdateRow, error) {
	rows, err := q.db.Query(ctx, getSubscriptionsNeedingUpdate)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []GetSubscriptionsNeedingUpdateRow
	for rows.Next() {
		var i GetSubscriptionsNeedingUpdateRow
		if err := rows.Scan(
			&i.Key,
			&i.Cursor,
			&i.Topic,
			&i.Name,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const insertCallEvent = `-- name: InsertCallEvent :exec
INSERT INTO events (deployment_id, request_id, time_stamp, type,
                    custom_key_1, custom_key_2, custom_key_3, custom_key_4, payload)
VALUES ((SELECT id FROM deployments WHERE deployments.key = $1::deployment_key),
        (CASE
             WHEN $2::TEXT IS NULL THEN NULL
             ELSE (SELECT id FROM requests ir WHERE ir.key = $2::TEXT)
            END),
        $3::TIMESTAMPTZ,
        'call',
        $4::TEXT,
        $5::TEXT,
        $6::TEXT,
        $7::TEXT,
        jsonb_build_object(
                'duration_ms', $8::BIGINT,
                'request', $9::JSONB,
                'response', $10::JSONB,
                'error', $11::TEXT,
                'stack', $12::TEXT
            ))
`

type InsertCallEventParams struct {
	DeploymentKey model.DeploymentKey
	RequestKey    optional.Option[string]
	TimeStamp     time.Time
	SourceModule  optional.Option[string]
	SourceVerb    optional.Option[string]
	DestModule    string
	DestVerb      string
	DurationMs    int64
	Request       []byte
	Response      []byte
	Error         optional.Option[string]
	Stack         optional.Option[string]
}

func (q *Queries) InsertCallEvent(ctx context.Context, arg InsertCallEventParams) error {
	_, err := q.db.Exec(ctx, insertCallEvent,
		arg.DeploymentKey,
		arg.RequestKey,
		arg.TimeStamp,
		arg.SourceModule,
		arg.SourceVerb,
		arg.DestModule,
		arg.DestVerb,
		arg.DurationMs,
		arg.Request,
		arg.Response,
		arg.Error,
		arg.Stack,
	)
	return err
}

const insertDeploymentCreatedEvent = `-- name: InsertDeploymentCreatedEvent :exec
INSERT INTO events (deployment_id, type, custom_key_1, custom_key_2, payload)
VALUES ((SELECT id
         FROM deployments
         WHERE deployments.key = $1::deployment_key),
        'deployment_created',
        $2::TEXT,
        $3::TEXT,
        jsonb_build_object(
                'min_replicas', $4::INT,
                'replaced', $5::deployment_key
            ))
`

type InsertDeploymentCreatedEventParams struct {
	DeploymentKey model.DeploymentKey
	Language      string
	ModuleName    string
	MinReplicas   int32
	Replaced      optional.Option[model.DeploymentKey]
}

func (q *Queries) InsertDeploymentCreatedEvent(ctx context.Context, arg InsertDeploymentCreatedEventParams) error {
	_, err := q.db.Exec(ctx, insertDeploymentCreatedEvent,
		arg.DeploymentKey,
		arg.Language,
		arg.ModuleName,
		arg.MinReplicas,
		arg.Replaced,
	)
	return err
}

const insertDeploymentUpdatedEvent = `-- name: InsertDeploymentUpdatedEvent :exec
INSERT INTO events (deployment_id, type, custom_key_1, custom_key_2, payload)
VALUES ((SELECT id
         FROM deployments
         WHERE deployments.key = $1::deployment_key),
        'deployment_updated',
        $2::TEXT,
        $3::TEXT,
        jsonb_build_object(
                'prev_min_replicas', $4::INT,
                'min_replicas', $5::INT
            ))
`

type InsertDeploymentUpdatedEventParams struct {
	DeploymentKey   model.DeploymentKey
	Language        string
	ModuleName      string
	PrevMinReplicas int32
	MinReplicas     int32
}

func (q *Queries) InsertDeploymentUpdatedEvent(ctx context.Context, arg InsertDeploymentUpdatedEventParams) error {
	_, err := q.db.Exec(ctx, insertDeploymentUpdatedEvent,
		arg.DeploymentKey,
		arg.Language,
		arg.ModuleName,
		arg.PrevMinReplicas,
		arg.MinReplicas,
	)
	return err
}

const insertEvent = `-- name: InsertEvent :exec
INSERT INTO events (deployment_id, request_id, type,
                    custom_key_1, custom_key_2, custom_key_3, custom_key_4,
                    payload)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
RETURNING id
`

type InsertEventParams struct {
	DeploymentID int64
	RequestID    optional.Option[int64]
	Type         EventType
	CustomKey1   optional.Option[string]
	CustomKey2   optional.Option[string]
	CustomKey3   optional.Option[string]
	CustomKey4   optional.Option[string]
	Payload      json.RawMessage
}

func (q *Queries) InsertEvent(ctx context.Context, arg InsertEventParams) error {
	_, err := q.db.Exec(ctx, insertEvent,
		arg.DeploymentID,
		arg.RequestID,
		arg.Type,
		arg.CustomKey1,
		arg.CustomKey2,
		arg.CustomKey3,
		arg.CustomKey4,
		arg.Payload,
	)
	return err
}

const insertLogEvent = `-- name: InsertLogEvent :exec
INSERT INTO events (deployment_id, request_id, time_stamp, custom_key_1, type, payload)
VALUES ((SELECT id FROM deployments d WHERE d.key = $1::deployment_key LIMIT 1),
        (CASE
             WHEN $2::TEXT IS NULL THEN NULL
             ELSE (SELECT id FROM requests ir WHERE ir.key = $2::TEXT LIMIT 1)
            END),
        $3::TIMESTAMPTZ,
        $4::INT,
        'log',
        jsonb_build_object(
                'message', $5::TEXT,
                'attributes', $6::JSONB,
                'error', $7::TEXT,
                'stack', $8::TEXT
            ))
`

type InsertLogEventParams struct {
	DeploymentKey model.DeploymentKey
	RequestKey    optional.Option[string]
	TimeStamp     time.Time
	Level         int32
	Message       string
	Attributes    []byte
	Error         optional.Option[string]
	Stack         optional.Option[string]
}

func (q *Queries) InsertLogEvent(ctx context.Context, arg InsertLogEventParams) error {
	_, err := q.db.Exec(ctx, insertLogEvent,
		arg.DeploymentKey,
		arg.RequestKey,
		arg.TimeStamp,
		arg.Level,
		arg.Message,
		arg.Attributes,
		arg.Error,
		arg.Stack,
	)
	return err
}

const insertSubscriber = `-- name: InsertSubscriber :exec
INSERT INTO topic_subscribers (
  key,
  topic_subscriptions_id,
  deployment_id,
  sink,
  retry_attempts,
  backoff,
  max_backoff
)
VALUES (
  $1::subscriber_key,
  (
    SELECT topic_subscriptions.id as id
    FROM topic_subscriptions
    INNER JOIN modules ON topic_subscriptions.module_id = modules.id
    WHERE modules.name = $2::TEXT
      AND topic_subscriptions.name = $3::TEXT
  ),
  (SELECT id FROM deployments WHERE key = $4::deployment_key),
  $5,
  $6,
  $7::interval,
  $8::interval
)
`

type InsertSubscriberParams struct {
	Key              model.SubscriberKey
	Module           string
	SubscriptionName string
	Deployment       model.DeploymentKey
	Sink             schema.RefKey
	RetryAttempts    int32
	Backoff          time.Duration
	MaxBackoff       time.Duration
}

func (q *Queries) InsertSubscriber(ctx context.Context, arg InsertSubscriberParams) error {
	_, err := q.db.Exec(ctx, insertSubscriber,
		arg.Key,
		arg.Module,
		arg.SubscriptionName,
		arg.Deployment,
		arg.Sink,
		arg.RetryAttempts,
		arg.Backoff,
		arg.MaxBackoff,
	)
	return err
}

const killStaleControllers = `-- name: KillStaleControllers :one
WITH matches AS (
    UPDATE controller
        SET state = 'dead'
        WHERE state <> 'dead' AND last_seen < (NOW() AT TIME ZONE 'utc') - $1::INTERVAL
        RETURNING 1)
SELECT COUNT(*)
FROM matches
`

// Mark any controller entries that haven't been updated recently as dead.
func (q *Queries) KillStaleControllers(ctx context.Context, timeout time.Duration) (int64, error) {
	row := q.db.QueryRow(ctx, killStaleControllers, timeout)
	var count int64
	err := row.Scan(&count)
	return count, err
}

const killStaleRunners = `-- name: KillStaleRunners :one
WITH matches AS (
    UPDATE runners
        SET state = 'dead',
        deployment_id = NULL
        WHERE state <> 'dead' AND last_seen < (NOW() AT TIME ZONE 'utc') - $1::INTERVAL
        RETURNING 1)
SELECT COUNT(*)
FROM matches
`

func (q *Queries) KillStaleRunners(ctx context.Context, timeout time.Duration) (int64, error) {
	row := q.db.QueryRow(ctx, killStaleRunners, timeout)
	var count int64
	err := row.Scan(&count)
	return count, err
}

const loadAsyncCall = `-- name: LoadAsyncCall :one
SELECT id, created_at, lease_id, verb, state, origin, scheduled_at, request, response, error, remaining_attempts, backoff, max_backoff
FROM async_calls
WHERE id = $1
`

func (q *Queries) LoadAsyncCall(ctx context.Context, id int64) (AsyncCall, error) {
	row := q.db.QueryRow(ctx, loadAsyncCall, id)
	var i AsyncCall
	err := row.Scan(
		&i.ID,
		&i.CreatedAt,
		&i.LeaseID,
		&i.Verb,
		&i.State,
		&i.Origin,
		&i.ScheduledAt,
		&i.Request,
		&i.Response,
		&i.Error,
		&i.RemainingAttempts,
		&i.Backoff,
		&i.MaxBackoff,
	)
	return i, err
}

const newLease = `-- name: NewLease :one
INSERT INTO leases (
  idempotency_key,
  key,
  expires_at,
  metadata
)
VALUES (
  gen_random_uuid(),
  $1::lease_key,
  (NOW() AT TIME ZONE 'utc') + $2::interval,
  $3::JSONB
)
RETURNING idempotency_key
`

func (q *Queries) NewLease(ctx context.Context, key leases.Key, ttl time.Duration, metadata []byte) (uuid.UUID, error) {
	row := q.db.QueryRow(ctx, newLease, key, ttl, metadata)
	var idempotency_key uuid.UUID
	err := row.Scan(&idempotency_key)
	return idempotency_key, err
}

const publishEventForTopic = `-- name: PublishEventForTopic :exec
INSERT INTO topic_events (
    "key",
    topic_id,
    payload
  )
VALUES (
  $1::topic_event_key,
  (
    SELECT topics.id
    FROM topics
    INNER JOIN modules ON topics.module_id = modules.id
    WHERE modules.name = $2::TEXT
      AND topics.name = $3::TEXT
  ),
  $4
)
`

type PublishEventForTopicParams struct {
	Key     model.TopicEventKey
	Module  string
	Topic   string
	Payload []byte
}

func (q *Queries) PublishEventForTopic(ctx context.Context, arg PublishEventForTopicParams) error {
	_, err := q.db.Exec(ctx, publishEventForTopic,
		arg.Key,
		arg.Module,
		arg.Topic,
		arg.Payload,
	)
	return err
}

const releaseLease = `-- name: ReleaseLease :one
DELETE FROM leases
WHERE idempotency_key = $1 AND key = $2::lease_key
RETURNING true
`

func (q *Queries) ReleaseLease(ctx context.Context, idempotencyKey uuid.UUID, key leases.Key) (bool, error) {
	row := q.db.QueryRow(ctx, releaseLease, idempotencyKey, key)
	var column_1 bool
	err := row.Scan(&column_1)
	return column_1, err
}

const renewLease = `-- name: RenewLease :one
UPDATE leases
SET expires_at = (NOW() AT TIME ZONE 'utc') + $1::interval
WHERE idempotency_key = $2 AND key = $3::lease_key
RETURNING true
`

func (q *Queries) RenewLease(ctx context.Context, ttl time.Duration, idempotencyKey uuid.UUID, key leases.Key) (bool, error) {
	row := q.db.QueryRow(ctx, renewLease, ttl, idempotencyKey, key)
	var column_1 bool
	err := row.Scan(&column_1)
	return column_1, err
}

const replaceDeployment = `-- name: ReplaceDeployment :one
WITH update_container AS (
    UPDATE deployments AS d
        SET min_replicas = update_deployments.min_replicas
        FROM (VALUES ($1::deployment_key, 0),
                     ($2::deployment_key, $3::INT))
            AS update_deployments(key, min_replicas)
        WHERE d.key = update_deployments.key
        RETURNING 1)
SELECT COUNT(*)
FROM update_container
`

func (q *Queries) ReplaceDeployment(ctx context.Context, oldDeployment model.DeploymentKey, newDeployment model.DeploymentKey, minReplicas int32) (int64, error) {
	row := q.db.QueryRow(ctx, replaceDeployment, oldDeployment, newDeployment, minReplicas)
	var count int64
	err := row.Scan(&count)
	return count, err
}

const reserveRunner = `-- name: ReserveRunner :one
UPDATE runners
SET state               = 'reserved',
    reservation_timeout = $1::timestamptz,
    -- If a deployment is not found, then the deployment ID is -1
    -- and the update will fail due to a FK constraint.
    deployment_id       = COALESCE((SELECT id
                                    FROM deployments d
                                    WHERE d.key = $2::deployment_key
                                    LIMIT 1), -1)
WHERE id = (SELECT id
            FROM runners r
            WHERE r.state = 'idle'
              AND r.labels @> $3::jsonb
            LIMIT 1 FOR UPDATE SKIP LOCKED)
RETURNING runners.id, runners.key, runners.created, runners.last_seen, runners.reservation_timeout, runners.state, runners.endpoint, runners.module_name, runners.deployment_id, runners.labels
`

// Find an idle runner and reserve it for the given deployment.
func (q *Queries) ReserveRunner(ctx context.Context, reservationTimeout time.Time, deploymentKey model.DeploymentKey, labels []byte) (Runner, error) {
	row := q.db.QueryRow(ctx, reserveRunner, reservationTimeout, deploymentKey, labels)
	var i Runner
	err := row.Scan(
		&i.ID,
		&i.Key,
		&i.Created,
		&i.LastSeen,
		&i.ReservationTimeout,
		&i.State,
		&i.Endpoint,
		&i.ModuleName,
		&i.DeploymentID,
		&i.Labels,
	)
	return i, err
}

const setDeploymentDesiredReplicas = `-- name: SetDeploymentDesiredReplicas :exec
UPDATE deployments
SET min_replicas = $2
WHERE key = $1::deployment_key
RETURNING 1
`

func (q *Queries) SetDeploymentDesiredReplicas(ctx context.Context, key model.DeploymentKey, minReplicas int32) error {
	_, err := q.db.Exec(ctx, setDeploymentDesiredReplicas, key, minReplicas)
	return err
}

const startCronJobs = `-- name: StartCronJobs :many
WITH updates AS (
  UPDATE cron_jobs
  SET state = 'executing',
    start_time = (NOW() AT TIME ZONE 'utc')::TIMESTAMPTZ
  WHERE key = ANY ($1)
    AND state = 'idle'
    AND start_time < next_execution
    AND (next_execution AT TIME ZONE 'utc') < (NOW() AT TIME ZONE 'utc')::TIMESTAMPTZ
  RETURNING id, key, state, start_time, next_execution)
SELECT j.key as key, d.key as deployment_key, j.module_name as module, j.verb, j.schedule,
  COALESCE(u.start_time, j.start_time) as start_time,
  COALESCE(u.next_execution, j.next_execution) as next_execution,
  COALESCE(u.state, j.state) as state,
  d.min_replicas > 0 as has_min_replicas,
  CASE WHEN u.key IS NULL THEN FALSE ELSE TRUE END as updated
FROM cron_jobs j
  INNER JOIN deployments d on j.deployment_id = d.id
  LEFT JOIN updates u on j.id = u.id
WHERE j.key = ANY ($1)
`

type StartCronJobsRow struct {
	Key            model.CronJobKey
	DeploymentKey  model.DeploymentKey
	Module         string
	Verb           string
	Schedule       string
	StartTime      time.Time
	NextExecution  time.Time
	State          model.CronJobState
	HasMinReplicas bool
	Updated        bool
}

func (q *Queries) StartCronJobs(ctx context.Context, keys []string) ([]StartCronJobsRow, error) {
	rows, err := q.db.Query(ctx, startCronJobs, keys)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []StartCronJobsRow
	for rows.Next() {
		var i StartCronJobsRow
		if err := rows.Scan(
			&i.Key,
			&i.DeploymentKey,
			&i.Module,
			&i.Verb,
			&i.Schedule,
			&i.StartTime,
			&i.NextExecution,
			&i.State,
			&i.HasMinReplicas,
			&i.Updated,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const startFSMTransition = `-- name: StartFSMTransition :one
INSERT INTO fsm_instances (
  fsm,
  key,
  destination_state,
  async_call_id
) VALUES (
  $1,
  $2,
  $3::schema_ref,
  $4::BIGINT
)
ON CONFLICT(fsm, key) DO
UPDATE SET
  destination_state = $3::schema_ref,
  async_call_id = $4::BIGINT
WHERE
  fsm_instances.async_call_id IS NULL
  AND fsm_instances.destination_state IS NULL
RETURNING id, created_at, fsm, key, status, current_state, destination_state, async_call_id
`

type StartFSMTransitionParams struct {
	Fsm              schema.RefKey
	Key              string
	DestinationState schema.RefKey
	AsyncCallID      int64
}

// Start a new FSM transition, populating the destination state and async call ID.
//
// "key" is the unique identifier for the FSM execution.
func (q *Queries) StartFSMTransition(ctx context.Context, arg StartFSMTransitionParams) (FsmInstance, error) {
	row := q.db.QueryRow(ctx, startFSMTransition,
		arg.Fsm,
		arg.Key,
		arg.DestinationState,
		arg.AsyncCallID,
	)
	var i FsmInstance
	err := row.Scan(
		&i.ID,
		&i.CreatedAt,
		&i.Fsm,
		&i.Key,
		&i.Status,
		&i.CurrentState,
		&i.DestinationState,
		&i.AsyncCallID,
	)
	return i, err
}

const succeedAsyncCall = `-- name: SucceedAsyncCall :one
UPDATE async_calls
SET
  state = 'success'::async_call_state,
  response = $1::JSONB
WHERE id = $2
RETURNING true
`

func (q *Queries) SucceedAsyncCall(ctx context.Context, response []byte, iD int64) (bool, error) {
	row := q.db.QueryRow(ctx, succeedAsyncCall, response, iD)
	var column_1 bool
	err := row.Scan(&column_1)
	return column_1, err
}

const succeedFSMInstance = `-- name: SucceedFSMInstance :one
UPDATE fsm_instances
SET
  current_state = destination_state,
  destination_state = NULL,
  async_call_id = NULL,
  status = 'completed'::fsm_status
WHERE
  fsm = $1::schema_ref AND key = $2::TEXT
RETURNING true
`

func (q *Queries) SucceedFSMInstance(ctx context.Context, fsm schema.RefKey, key string) (bool, error) {
	row := q.db.QueryRow(ctx, succeedFSMInstance, fsm, key)
	var column_1 bool
	err := row.Scan(&column_1)
	return column_1, err
}

const upsertController = `-- name: UpsertController :one
INSERT INTO controller (key, endpoint)
VALUES ($1, $2)
ON CONFLICT (key) DO UPDATE SET state     = 'live',
                                endpoint  = $2,
                                last_seen = NOW() AT TIME ZONE 'utc'
RETURNING id
`

func (q *Queries) UpsertController(ctx context.Context, key model.ControllerKey, endpoint string) (int64, error) {
	row := q.db.QueryRow(ctx, upsertController, key, endpoint)
	var id int64
	err := row.Scan(&id)
	return id, err
}

const upsertModule = `-- name: UpsertModule :one
INSERT INTO modules (language, name)
VALUES ($1, $2)
ON CONFLICT (name) DO UPDATE SET language = $1
RETURNING id
`

func (q *Queries) UpsertModule(ctx context.Context, language string, name string) (int64, error) {
	row := q.db.QueryRow(ctx, upsertModule, language, name)
	var id int64
	err := row.Scan(&id)
	return id, err
}

const upsertRunner = `-- name: UpsertRunner :one
WITH deployment_rel AS (
    SELECT CASE
               WHEN $5::deployment_key IS NULL
                   THEN NULL
               ELSE COALESCE((SELECT id
                              FROM deployments d
                              WHERE d.key = $5::deployment_key
                              LIMIT 1), -1) END AS id)
INSERT
INTO runners (key, endpoint, state, labels, deployment_id, last_seen)
VALUES ($1,
        $2,
        $3,
        $4,
        (SELECT id FROM deployment_rel),
        NOW() AT TIME ZONE 'utc')
ON CONFLICT (key) DO UPDATE SET endpoint      = $2,
                                state         = $3,
                                labels        = $4,
                                deployment_id = (SELECT id FROM deployment_rel),
                                last_seen     = NOW() AT TIME ZONE 'utc'
RETURNING deployment_id
`

type UpsertRunnerParams struct {
	Key           model.RunnerKey
	Endpoint      string
	State         RunnerState
	Labels        []byte
	DeploymentKey optional.Option[model.DeploymentKey]
}

// Upsert a runner and return the deployment ID that it is assigned to, if any.
// If the deployment key is null, then deployment_rel.id will be null,
// otherwise we try to retrieve the deployments.id using the key. If
// there is no corresponding deployment, then the deployment ID is -1
// and the parent statement will fail due to a foreign key constraint.
func (q *Queries) UpsertRunner(ctx context.Context, arg UpsertRunnerParams) (optional.Option[int64], error) {
	row := q.db.QueryRow(ctx, upsertRunner,
		arg.Key,
		arg.Endpoint,
		arg.State,
		arg.Labels,
		arg.DeploymentKey,
	)
	var deployment_id optional.Option[int64]
	err := row.Scan(&deployment_id)
	return deployment_id, err
}

const upsertSubscription = `-- name: UpsertSubscription :exec
INSERT INTO topic_subscriptions (
  key,
  topic_id,
  module_id,
  deployment_id,
  name)
VALUES (
  $1::subscription_key,
  (
    SELECT topics.id as id
    FROM topics
    INNER JOIN modules ON topics.module_id = modules.id
    WHERE modules.name = $2::TEXT
      AND topics.name = $3::TEXT
  ),
  (SELECT id FROM modules WHERE name = $4::TEXT),
  (SELECT id FROM deployments WHERE key = $5::deployment_key),
  $6::TEXT
)
ON CONFLICT (name, module_id) DO
UPDATE SET 
  topic_id = excluded.topic_id,
  deployment_id = (SELECT id FROM deployments WHERE key = $5::deployment_key)
RETURNING id
`

type UpsertSubscriptionParams struct {
	Key         model.SubscriptionKey
	TopicModule string
	TopicName   string
	Module      string
	Deployment  model.DeploymentKey
	Name        string
}

func (q *Queries) UpsertSubscription(ctx context.Context, arg UpsertSubscriptionParams) error {
	_, err := q.db.Exec(ctx, upsertSubscription,
		arg.Key,
		arg.TopicModule,
		arg.TopicName,
		arg.Module,
		arg.Deployment,
		arg.Name,
	)
	return err
}

const upsertTopic = `-- name: UpsertTopic :exec
INSERT INTO topics (key, module_id, name, type)
VALUES (
  $1::topic_key,
  (SELECT id FROM modules WHERE name = $2::TEXT LIMIT 1),
  $3::TEXT,
  $4::TEXT
)
ON CONFLICT (name, module_id) DO 
UPDATE SET 
  type = $4::TEXT
RETURNING id
`

type UpsertTopicParams struct {
	Topic     model.TopicKey
	Module    string
	Name      string
	EventType string
}

func (q *Queries) UpsertTopic(ctx context.Context, arg UpsertTopicParams) error {
	_, err := q.db.Exec(ctx, upsertTopic,
		arg.Topic,
		arg.Module,
		arg.Name,
		arg.EventType,
	)
	return err
}