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

package lrdb

import (
	"context"
	"time"
)

const cleanupKafkaOffsets = `-- name: CleanupKafkaOffsets :execrows
DELETE FROM kafka_offset_tracker
WHERE consumer_group = $1
  AND topic = $2
  AND partition_id = $3
  AND max_offset <= $4
`

type CleanupKafkaOffsetsParams struct {
	ConsumerGroup string `json:"consumer_group"`
	Topic         string `json:"topic"`
	PartitionID   int32  `json:"partition_id"`
	MaxOffset     int64  `json:"max_offset"`
}

func (q *Queries) CleanupKafkaOffsets(ctx context.Context, arg CleanupKafkaOffsetsParams) (int64, error) {
	result, err := q.db.Exec(ctx, cleanupKafkaOffsets,
		arg.ConsumerGroup,
		arg.Topic,
		arg.PartitionID,
		arg.MaxOffset,
	)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected(), nil
}

const cleanupKafkaOffsetsByAge = `-- name: CleanupKafkaOffsetsByAge :execrows
DELETE FROM kafka_offset_tracker
WHERE created_at < $1
`

func (q *Queries) CleanupKafkaOffsetsByAge(ctx context.Context, createdBefore time.Time) (int64, error) {
	result, err := q.db.Exec(ctx, cleanupKafkaOffsetsByAge, createdBefore)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected(), nil
}

const insertKafkaOffsets = `-- name: InsertKafkaOffsets :exec
INSERT INTO kafka_offset_tracker (
  consumer_group,
  topic,
  partition_id,
  min_offset,
  max_offset,
  offsets,
  created_at
) VALUES (
  $1,
  $2,
  $3,
  (SELECT MIN(o) FROM unnest($4::bigint[]) AS o),
  (SELECT MAX(o) FROM unnest($4::bigint[]) AS o),
  $4,
  COALESCE($5::timestamptz, now())
)
`

type InsertKafkaOffsetsParams struct {
	ConsumerGroup string     `json:"consumer_group"`
	Topic         string     `json:"topic"`
	PartitionID   int32      `json:"partition_id"`
	Offsets       []int64    `json:"offsets"`
	CreatedAt     *time.Time `json:"created_at"`
}

func (q *Queries) InsertKafkaOffsets(ctx context.Context, arg InsertKafkaOffsetsParams) error {
	_, err := q.db.Exec(ctx, insertKafkaOffsets,
		arg.ConsumerGroup,
		arg.Topic,
		arg.PartitionID,
		arg.Offsets,
		arg.CreatedAt,
	)
	return err
}

const kafkaOffsetsAfter = `-- name: KafkaOffsetsAfter :one
SELECT
  COALESCE(
    ARRAY(
      SELECT o
      FROM kafka_offset_tracker kot
      CROSS JOIN LATERAL unnest(kot.offsets) AS o
      WHERE kot.consumer_group = $1
        AND kot.topic          = $2
        AND kot.partition_id   = $3
        AND kot.max_offset     >= $4
        AND o                 >= $4
      ORDER BY o
    ),
    ARRAY[]::bigint[]
  )::bigint[] AS offsets
`

type KafkaOffsetsAfterParams struct {
	ConsumerGroup string `json:"consumer_group"`
	Topic         string `json:"topic"`
	PartitionID   int32  `json:"partition_id"`
	MinOffset     int64  `json:"min_offset"`
}

func (q *Queries) KafkaOffsetsAfter(ctx context.Context, arg KafkaOffsetsAfterParams) ([]int64, error) {
	row := q.db.QueryRow(ctx, kafkaOffsetsAfter,
		arg.ConsumerGroup,
		arg.Topic,
		arg.PartitionID,
		arg.MinOffset,
	)
	var offsets []int64
	err := row.Scan(&offsets)
	return offsets, err
}
