-- name: ReadPayloads :many
WITH inputs AS (
    SELECT
        UNNEST(@ids::BIGINT[]) AS id,
        UNNEST(@insertedAts::TIMESTAMPTZ[]) AS inserted_at,
        UNNEST(@tenantIds::UUID[]) AS tenant_id,
        UNNEST(CAST(@types::TEXT[] AS v1_payload_type[])) AS type
)

SELECT *
FROM v1_payload
WHERE (tenant_id, id, inserted_at, type) IN (
        SELECT tenant_id, id, inserted_at, type
        FROM inputs
    )
;

-- name: WritePayloads :exec
WITH inputs AS (
    SELECT DISTINCT
        UNNEST(@ids::BIGINT[]) AS id,
        UNNEST(@insertedAts::TIMESTAMPTZ[]) AS inserted_at,
        UNNEST(@externalIds::UUID[]) AS external_id,
        UNNEST(CAST(@types::TEXT[] AS v1_payload_type[])) AS type,
        UNNEST(CAST(@locations::TEXT[] AS v1_payload_location[])) AS location,
        UNNEST(@externalLocationKeys::TEXT[]) AS external_location_key,
        UNNEST(@inlineContents::JSONB[]) AS inline_content,
        UNNEST(@tenantIds::UUID[]) AS tenant_id
)

INSERT INTO v1_payload (
    tenant_id,
    id,
    inserted_at,
    external_id,
    type,
    location,
    external_location_key,
    inline_content
)
SELECT
    i.tenant_id,
    i.id,
    i.inserted_at,
    i.external_id,
    i.type,
    i.location,
    CASE WHEN i.external_location_key = '' OR i.location != 'EXTERNAL' THEN NULL ELSE i.external_location_key END,
    i.inline_content
FROM
    inputs i
ORDER BY i.tenant_id, i.inserted_at, i.id, i.type
ON CONFLICT (tenant_id, id, inserted_at, type)
DO UPDATE SET
    location = EXCLUDED.location,
    external_location_key = CASE WHEN EXCLUDED.external_location_key = '' OR EXCLUDED.location != 'EXTERNAL' THEN NULL ELSE EXCLUDED.external_location_key END,
    inline_content = EXCLUDED.inline_content,
    updated_at = NOW()
;

-- name: AnalyzeV1Payload :exec
ANALYZE v1_payload;

-- name: ListPaginatedPayloadsForOffload :many
WITH payloads AS (
    SELECT
        (p).*
    FROM list_paginated_payloads_for_offload(
        @partitionDate::DATE,
        @lastExternalId::UUID,
        @nextExternalId::UUID,
        @batchSize::INTEGER
    ) p
)
SELECT
    tenant_id::UUID,
    id::BIGINT,
    inserted_at::TIMESTAMPTZ,
    external_id::UUID,
    type::v1_payload_type,
    location::v1_payload_location,
    COALESCE(external_location_key, '')::TEXT AS external_location_key,
    inline_content::JSONB AS inline_content,
    updated_at::TIMESTAMPTZ
FROM payloads;

-- name: CreatePayloadRangeChunks :many
WITH chunks AS (
    SELECT
        (p).*
    FROM create_payload_offload_range_chunks(
        @partitionDate::DATE,
        @windowSize::INTEGER,
        @chunkSize::INTEGER,
        @lastExternalId::UUID
    ) p
)

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

-- name: CreateV1PayloadCutoverTemporaryTable :exec
SELECT copy_v1_payload_partition_structure(@date::DATE);

-- name: SwapV1PayloadPartitionWithTemp :exec
SELECT swap_v1_payload_partition_with_temp(@date::DATE);

-- name: AcquireOrExtendCutoverJobLease :one
WITH inputs AS (
    SELECT
        @key::DATE AS key,
        @leaseProcessId::UUID AS lease_process_id,
        @leaseExpiresAt::TIMESTAMPTZ AS lease_expires_at,
        @lastExternalId::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_payload_cutover_job_offset
    WHERE lease_process_id != @leaseProcessId::UUID
), to_insert AS (
    SELECT *
    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_payload_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 tuple of (last_tenant_id, last_inserted_at, last_id, last_type)
    -- 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_payload_cutover_job_offset.lease_process_id THEN EXCLUDED.last_external_id
        ELSE v1_payload_cutover_job_offset.last_external_id
    END,

    lease_process_id = EXCLUDED.lease_process_id,
    lease_expires_at = EXCLUDED.lease_expires_at
WHERE v1_payload_cutover_job_offset.lease_expires_at < NOW() OR v1_payload_cutover_job_offset.lease_process_id = @leaseProcessId::UUID
RETURNING *
;

-- name: MarkCutoverJobAsCompleted :exec
UPDATE v1_payload_cutover_job_offset
SET is_completed = TRUE
WHERE key = @key::DATE
;

-- name: CleanUpCutoverJobOffsets :exec
DELETE FROM v1_payload_cutover_job_offset
WHERE NOT key = ANY(@keysToKeep::DATE[])
;

-- name: ComputePayloadBatchSize :one
SELECT compute_payload_batch_size(
    @partitionDate::DATE,
    @lastExternalId::UUID,
    @batchSize::INTEGER
) AS total_size_bytes;

-- name: GetOffloadedPayloadIndexBlocks :many
WITH inputs AS (
    SELECT
        UNNEST(@insertedAts::DATE[]) AS inserted_at_date,
        UNNEST(@externalIds::UUID[]) AS external_id
)

SELECT i.external_id::UUID AS external_id, index_file_key
FROM v1_payload_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
;

-- name: CreateOffloadedPayloadIndexBlock :exec
INSERT INTO v1_payload_offloaded_block_index (
    payload_inserted_at_date,
    block_external_id_range,
    index_file_key
)
VALUES (
    @payloadInsertedAtDate::DATE,
    uuidrange(@blockLowerExternalIdBound::UUID, @blockUpperExternalIdBound::UUID, '(]'),
    @indexFileKey::TEXT
)
ON CONFLICT ON CONSTRAINT v1_payload_offloaded_block_index_date_range_excl DO NOTHING
;

-- name: DeleteOldPayloadOffloadedBlockIndexRows :exec
DELETE FROM v1_payload_offloaded_block_index
WHERE payload_inserted_at_date < @before::DATE
;
