db

func ConnectToDB ¶ added in v0.4.0

func ConnectToDB(ctx context.Context,
	logger *zap.Logger,
	dsn string,
	namespace string,
	waitForDB time.Duration,
	statementTimeout time.Duration,
	prom *prometheus.Registry,
) (*sql.DB, error)

ConnectToDB establishes a connection to an existing database using the provided DSN. Unlike NewNamespacedDB, this function does not create the database or run migrations. If namespace is provided, it overrides the database name in the DSN.

func Get5MinutesOfCongestion ¶ added in v0.3.0

func Get5MinutesOfCongestion(
	ctx context.Context,
	querier *queries.Queries,
	originatorID, endMinute int32,
) (out [5]int32, err error)

Get5MinutesOfCongestion gets the congestion for the minute specified by `endMinute` and the previous 4 minutes returned in descending order with missing values filled with 0

func InsertGatewayEnvelopeAndIncrementUnsettledUsage ¶ added in v0.3.0

func InsertGatewayEnvelopeAndIncrementUnsettledUsage(
	ctx context.Context,
	db *sql.DB,
	insertParams queries.InsertGatewayEnvelopeParams,
	incrementParams queries.IncrementUnsettledUsageParams,
) (int64, error)

InsertGatewayEnvelopeAndIncrementUnsettledUsage inserts a gateway envelope and updates unsettled usage and congestion counters within a single database transaction.

This function runs inside a managed transaction created by RunInTxWithResult().

Steps:

1. Calls InsertGatewayEnvelopeWithChecksTransactional() to insert the envelope, automatically creating any missing partitions if needed.
2. If a new envelope is inserted, increments unsettled usage and congestion counters for the originator within the same transaction.
3. If the envelope already exists (duplicate insert), metrics are not updated.

The function ensures atomicity between envelope insertion and usage updates. Safe for high-throughput ingest paths where message persistence and usage tracking must succeed or fail together.

func InsertGatewayEnvelopeWithChecksStandalone ¶ added in v1.0.0

func InsertGatewayEnvelopeWithChecksStandalone(
	ctx context.Context,
	q *queries.Queries,
	row queries.InsertGatewayEnvelopeParams,
) (queries.InsertGatewayEnvelopeRow, error)

InsertGatewayEnvelopeWithChecksStandalone inserts a gateway envelope in a non-transactional context, automatically creating missing partitions and retrying once.

Behavior:

• Performs an insert into the v2 tables.
• On “no partition of relation …” errors, creates the necessary partitions in the same connection, and retries the insert once.

This function does not use SAVEPOINTs and does not depend on an explicit transaction. It is ideal for standalone operations such as backfills, batch imports, or ingestion workers where each insert is independent of others.

func InsertGatewayEnvelopeWithChecksTransactional ¶ added in v1.0.0

func InsertGatewayEnvelopeWithChecksTransactional(
	ctx context.Context,
	q *queries.Queries,
	row queries.InsertGatewayEnvelopeParams,
) (queries.InsertGatewayEnvelopeRow, error)

InsertGatewayEnvelopeWithChecksTransactional attempts to insert a gateway envelope inside the current SQL transaction, with automatic partition creation and retry.

Behavior:

• Creates a SAVEPOINT before the insert so that a failure does not abort the entire tx.
• On “no partition of relation …” errors, it rolls back to the savepoint, creates the missing partition(s) using the same connection (within the tx), and retries the insert once.
• On success, the savepoint is released.

This variant must be called within an active transaction. It avoids full tx rollbacks and ensures inserts can proceed even when new partitions are required. Use for transactional ingestion flows where atomicity must be preserved.

func NewNamespacedDB ¶ added in v0.1.2

func NewNamespacedDB(
	ctx context.Context,
	logger *zap.Logger,
	dsn string,
	namespace string,
	waitForDB time.Duration,
	statementTimeout time.Duration,
	prom *prometheus.Registry,
) (*sql.DB, error)

NewNamespacedDB creates a new database with the given namespace if it doesn't exist and returns the full DSN for the new database.

func NullInt32 ¶

func NullInt32(v int32) sql.NullInt32

func NullInt64 ¶

func NullInt64(v int64) sql.NullInt64

func RunInTx ¶

func RunInTx(
	ctx context.Context,
	db *sql.DB,
	opts *sql.TxOptions,
	fn func(ctx context.Context, txQueries *queries.Queries) error,
) error

func RunInTxRaw ¶ added in v0.5.0

func RunInTxRaw(
	ctx context.Context,
	db *sql.DB,
	opts *sql.TxOptions,
	fn func(ctx context.Context, tx *sql.Tx) error,
) error

func RunInTxWithResult ¶ added in v0.3.0

func RunInTxWithResult[T any](
	ctx context.Context,
	db *sql.DB,
	opts *sql.TxOptions,
	fn func(ctx context.Context, txQueries *queries.Queries) (T, error),
) (result T, err error)

func SetVectorClockByOriginators ¶ added in v1.0.0

func SetVectorClockByOriginators(
	q *queries.SelectGatewayEnvelopesByOriginatorsParams,
	vc VectorClock,
) *queries.SelectGatewayEnvelopesByOriginatorsParams

func SetVectorClockByTopics ¶ added in v1.0.0

func SetVectorClockByTopics(
	q *queries.SelectGatewayEnvelopesByTopicsParams,
	vc VectorClock,
) *queries.SelectGatewayEnvelopesByTopicsParams

func SetVectorClockUnfiltered ¶ added in v1.0.0

func SetVectorClockUnfiltered(
	q *queries.SelectGatewayEnvelopesUnfilteredParams,
	vc VectorClock,
) *queries.SelectGatewayEnvelopesUnfilteredParams

func TransformRowsByOriginator ¶ added in v1.0.0

func TransformRowsByOriginator(
	rows []queries.SelectGatewayEnvelopesByOriginatorsRow,
) []queries.GatewayEnvelopesView

func TransformRowsByTopic ¶ added in v1.0.0

func TransformRowsByTopic(
	rows []queries.SelectGatewayEnvelopesByTopicsRow,
) []queries.GatewayEnvelopesView