Every database statement, one function each, as raw SQL.

All functions take the repo explicitly. The complete_* functions run the outcome UPDATE and the consumed-signal DELETE in one transaction. concurrency_key serialization (advisory lock) helpers live here too; the caller is responsible for holding a single connection (Repo.checkout/1) for the lock's lifetime.

Every statement has a static SQL text and goes through the connection-level prepared-statement cache (cache_statement:), so Postgres parses and plans it once per connection instead of on every call — bulk inserts pass rows as parallel arrays via unnest to keep the text static (and the parameter count fixed, clear of the 65535-parameter protocol cap). The one exception is upsert_rate_configs (dynamic VALUES, boot-time only). Hosts behind a transaction-pooling proxy set prepare: :unnamed on the repo to bypass the cache.

JSON values (state, result, signal payloads) arrive here as encoded JSON text and are bound as TEXT parameters cast server-side ($n::text::jsonb). A bare $n::jsonb parameter would make the driver JSON-encode the already-encoded string, storing a double-encoded jsonb scalar instead of an object — invisible to ->>/jsonb indexes. Rows written by versions that did exactly that still decode fine: the read paths accept both formats.