mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-08 01:16:45 +08:00
b4bfe083be
* perf(db): mint UUIDv7 primary keys for the append-heavy tables
Random v4 primary keys scatter every insert across the whole primary-key
B-tree, so each row dirties a different page. On the tables that grow
continuously — the task queue, messages, comments, activity and audit rows —
that is pure write amplification. UUIDv7 puts a big-endian millisecond
timestamp in the leading 48 bits, so consecutive inserts share a page.
The 12 tables converted here are the append-only hot paths:
agent_task_queue, activity_log, inbox_item, comment, issue, chat_session,
chat_message, autopilot_run, webhook_delivery, task_token,
channel_inbound_audit, sys_cron_executions
Ids come from the new pkg/dbid, which returns a pgtype.UUID ready to assign to
a sqlc params struct. Upsert-dominated tables (task_usage, agent_runtime, ...)
are deliberately left alone: v7 only buys locality for rows that are actually
inserted. Lease and claim tokens keep gen_random_uuid() — they must stay
unpredictable and carry no timestamp.
No migration, and no backfill. Every converted INSERT binds its id as
COALESCE(sqlc.narg('id')::uuid, gen_random_uuid()), so the column default stays
reachable: a caller that passes no id inserts exactly as it does today. Tables
will hold a mix of v4 and v7 ids permanently, which the uuid type and every
parser in the repo already accept.
Verification: - pkg/dbid: version/variant, uniqueness, and a non-decreasing timestamp
prefix; a shape test asserting every converted query keeps the COALESCE
fallback; and a source scan asserting every production params literal mints
an id, so a call site added later cannot silently fall back to v4.
- internal/handler: round-trips issue, comment, activity_log, inbox_item,
chat_session, chat_message, agent_task_queue, task_token and
channel_inbound_audit against Postgres and asserts the stored id is v7,
plus one case proving an unset id still inserts (as v4).
- internal/scheduler: covers the one hand-written INSERT, asserting the
execution id is v7 while lease_token stays v4.
Co-authored-by: multica-agent <[email protected]>
* fix(db): address UUIDv7 review feedback
Co-authored-by: multica-agent <[email protected]>
* fix(db): cover Telegram audit UUIDv7
Co-authored-by: multica-agent <[email protected]>
---------
Co-authored-by: Sol-Boy <[email protected]>
Co-authored-by: multica-agent <[email protected]>
884 lines
28 KiB
Go
884 lines
28 KiB
Go
// Code generated by sqlc. DO NOT EDIT.
|
||
// versions:
|
||
// sqlc v1.31.1
|
||
// source: webhook_delivery.sql
|
||
|
||
package db
|
||
|
||
import (
|
||
"context"
|
||
|
||
"github.com/jackc/pgx/v5/pgtype"
|
||
)
|
||
|
||
const acknowledgeWebhookDelivery = `-- name: AcknowledgeWebhookDelivery :one
|
||
UPDATE webhook_delivery
|
||
SET response_status = $2,
|
||
response_body = $3,
|
||
last_attempt_at = now()
|
||
WHERE id = $1
|
||
RETURNING id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key
|
||
`
|
||
|
||
type AcknowledgeWebhookDeliveryParams struct {
|
||
ID pgtype.UUID `json:"id"`
|
||
ResponseStatus pgtype.Int4 `json:"response_status"`
|
||
ResponseBody pgtype.Text `json:"response_body"`
|
||
}
|
||
|
||
// Best-effort operator metadata for the HTTP acknowledgement. The delivery and
|
||
// run are already durable, so callers still return their response if this
|
||
// metadata-only update fails.
|
||
func (q *Queries) AcknowledgeWebhookDelivery(ctx context.Context, arg AcknowledgeWebhookDeliveryParams) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, acknowledgeWebhookDelivery, arg.ID, arg.ResponseStatus, arg.ResponseBody)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const bumpWebhookDeliveryAttempt = `-- name: BumpWebhookDeliveryAttempt :one
|
||
UPDATE webhook_delivery
|
||
SET attempt_count = attempt_count + 1,
|
||
last_attempt_at = now()
|
||
WHERE id = $1
|
||
RETURNING id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key
|
||
`
|
||
|
||
// On duplicate detection, bump attempt_count and refresh last_attempt_at on
|
||
// the existing delivery so the UI / operator can see retry pressure without
|
||
// creating a new row per attempt.
|
||
func (q *Queries) BumpWebhookDeliveryAttempt(ctx context.Context, id pgtype.UUID) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, bumpWebhookDeliveryAttempt, id)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const claimQueuedWebhookDelivery = `-- name: ClaimQueuedWebhookDelivery :one
|
||
WITH candidate AS (
|
||
SELECT id
|
||
FROM webhook_delivery
|
||
WHERE status = 'queued'
|
||
AND available_at <= now()
|
||
AND (lease_expires_at IS NULL OR lease_expires_at <= now())
|
||
ORDER BY available_at, created_at
|
||
FOR UPDATE SKIP LOCKED
|
||
LIMIT 1
|
||
)
|
||
UPDATE webhook_delivery AS d
|
||
SET lease_token = gen_random_uuid(),
|
||
lease_expires_at = now() + interval '2 minutes'
|
||
FROM candidate
|
||
WHERE d.id = candidate.id
|
||
RETURNING d.id, d.workspace_id, d.autopilot_id, d.trigger_id, d.provider, d.event, d.dedupe_key, d.dedupe_source, d.signature_status, d.status, d.attempt_count, d.selected_headers, d.content_type, d.raw_body, d.response_status, d.response_body, d.autopilot_run_id, d.replayed_from_delivery_id, d.error, d.received_at, d.last_attempt_at, d.created_at, d.available_at, d.lease_token, d.lease_expires_at, d.dispatch_attempts, d.reason_code, d.replay_idempotency_key
|
||
`
|
||
|
||
// Claims one due delivery. SKIP LOCKED spreads work across replicas; the
|
||
// expiring token makes a crashed claim visible to a later sweeper pass.
|
||
// The lease is a scheduling optimization, not the exactly-once guard: if a
|
||
// slow dispatch outlives it, uq_autopilot_run_webhook_delivery remains the
|
||
// final protection against duplicate downstream runs.
|
||
func (q *Queries) ClaimQueuedWebhookDelivery(ctx context.Context) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, claimQueuedWebhookDelivery)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const completeClaimedWebhookDelivery = `-- name: CompleteClaimedWebhookDelivery :one
|
||
UPDATE webhook_delivery
|
||
SET status = $3,
|
||
autopilot_run_id = $4,
|
||
dispatch_attempts = dispatch_attempts + 1,
|
||
error = $5,
|
||
reason_code = $6,
|
||
lease_token = NULL,
|
||
lease_expires_at = NULL,
|
||
last_attempt_at = now()
|
||
WHERE id = $1
|
||
AND lease_token = $2
|
||
AND status = 'queued'
|
||
RETURNING id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key
|
||
`
|
||
|
||
type CompleteClaimedWebhookDeliveryParams struct {
|
||
ID pgtype.UUID `json:"id"`
|
||
LeaseToken pgtype.UUID `json:"lease_token"`
|
||
Status string `json:"status"`
|
||
AutopilotRunID pgtype.UUID `json:"autopilot_run_id"`
|
||
Error pgtype.Text `json:"error"`
|
||
ReasonCode pgtype.Text `json:"reason_code"`
|
||
}
|
||
|
||
func (q *Queries) CompleteClaimedWebhookDelivery(ctx context.Context, arg CompleteClaimedWebhookDeliveryParams) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, completeClaimedWebhookDelivery,
|
||
arg.ID,
|
||
arg.LeaseToken,
|
||
arg.Status,
|
||
arg.AutopilotRunID,
|
||
arg.Error,
|
||
arg.ReasonCode,
|
||
)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const createWebhookDelivery = `-- name: CreateWebhookDelivery :one
|
||
|
||
INSERT INTO webhook_delivery (
|
||
workspace_id, autopilot_id, trigger_id, provider, event,
|
||
dedupe_key, dedupe_source, signature_status, status,
|
||
selected_headers, content_type, raw_body,
|
||
replayed_from_delivery_id, replay_idempotency_key, reason_code, id
|
||
) VALUES (
|
||
$1, $2, $3, $4, $5,
|
||
$9, $10, $6, $7,
|
||
$8, $11, $12,
|
||
$13, $14,
|
||
$15, COALESCE($16::uuid, gen_random_uuid())
|
||
) RETURNING id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key
|
||
`
|
||
|
||
type CreateWebhookDeliveryParams struct {
|
||
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
||
AutopilotID pgtype.UUID `json:"autopilot_id"`
|
||
TriggerID pgtype.UUID `json:"trigger_id"`
|
||
Provider string `json:"provider"`
|
||
Event string `json:"event"`
|
||
SignatureStatus string `json:"signature_status"`
|
||
Status string `json:"status"`
|
||
SelectedHeaders []byte `json:"selected_headers"`
|
||
DedupeKey pgtype.Text `json:"dedupe_key"`
|
||
DedupeSource pgtype.Text `json:"dedupe_source"`
|
||
ContentType pgtype.Text `json:"content_type"`
|
||
RawBody []byte `json:"raw_body"`
|
||
ReplayedFromDeliveryID pgtype.UUID `json:"replayed_from_delivery_id"`
|
||
ReplayIdempotencyKey pgtype.Text `json:"replay_idempotency_key"`
|
||
ReasonCode pgtype.Text `json:"reason_code"`
|
||
ID pgtype.UUID `json:"id"`
|
||
}
|
||
|
||
// =====================
|
||
// Webhook Delivery
|
||
// =====================
|
||
// Inserts a delivery row. On dedupe-key collision the unique partial index
|
||
// (trigger_id, dedupe_key) raises 23505 and the handler treats it as
|
||
// "duplicate" rather than an error.
|
||
func (q *Queries) CreateWebhookDelivery(ctx context.Context, arg CreateWebhookDeliveryParams) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, createWebhookDelivery,
|
||
arg.WorkspaceID,
|
||
arg.AutopilotID,
|
||
arg.TriggerID,
|
||
arg.Provider,
|
||
arg.Event,
|
||
arg.SignatureStatus,
|
||
arg.Status,
|
||
arg.SelectedHeaders,
|
||
arg.DedupeKey,
|
||
arg.DedupeSource,
|
||
arg.ContentType,
|
||
arg.RawBody,
|
||
arg.ReplayedFromDeliveryID,
|
||
arg.ReplayIdempotencyKey,
|
||
arg.ReasonCode,
|
||
arg.ID,
|
||
)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const deferClaimedWebhookDelivery = `-- name: DeferClaimedWebhookDelivery :one
|
||
UPDATE webhook_delivery
|
||
SET available_at = $3,
|
||
lease_token = NULL,
|
||
lease_expires_at = NULL
|
||
WHERE id = $1
|
||
AND lease_token = $2
|
||
AND status = 'queued'
|
||
RETURNING id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key
|
||
`
|
||
|
||
type DeferClaimedWebhookDeliveryParams struct {
|
||
ID pgtype.UUID `json:"id"`
|
||
LeaseToken pgtype.UUID `json:"lease_token"`
|
||
AvailableAt pgtype.Timestamptz `json:"available_at"`
|
||
}
|
||
|
||
// Releases a claim without counting a dispatch attempt. Used when the
|
||
// per-trigger worker budget is exhausted before downstream dispatch begins.
|
||
func (q *Queries) DeferClaimedWebhookDelivery(ctx context.Context, arg DeferClaimedWebhookDeliveryParams) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, deferClaimedWebhookDelivery, arg.ID, arg.LeaseToken, arg.AvailableAt)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const getWebhookDelivery = `-- name: GetWebhookDelivery :one
|
||
SELECT id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key FROM webhook_delivery
|
||
WHERE id = $1
|
||
`
|
||
|
||
func (q *Queries) GetWebhookDelivery(ctx context.Context, id pgtype.UUID) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, getWebhookDelivery, id)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const getWebhookDeliveryByTriggerAndDedupe = `-- name: GetWebhookDeliveryByTriggerAndDedupe :one
|
||
SELECT id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key FROM webhook_delivery
|
||
WHERE trigger_id = $1
|
||
AND dedupe_key = $2
|
||
ORDER BY (status IN ('rejected', 'failed')), created_at DESC
|
||
LIMIT 1
|
||
`
|
||
|
||
type GetWebhookDeliveryByTriggerAndDedupeParams struct {
|
||
TriggerID pgtype.UUID `json:"trigger_id"`
|
||
DedupeKey pgtype.Text `json:"dedupe_key"`
|
||
}
|
||
|
||
// Looks up the existing delivery for a (trigger, dedupe_key) pair so that
|
||
// duplicate requests return the original delivery_id / autopilot_run_id.
|
||
// The partial unique index excludes terminal-but-not-successful statuses
|
||
// (`rejected`, `failed`), so multiple such rows can coexist for the same
|
||
// key. Prefer non-terminal rows in the lookup: without the ORDER BY we
|
||
// could return a stale rejection / failure even after the operator fixed
|
||
// the cause and a fresh dispatch succeeded.
|
||
func (q *Queries) GetWebhookDeliveryByTriggerAndDedupe(ctx context.Context, arg GetWebhookDeliveryByTriggerAndDedupeParams) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, getWebhookDeliveryByTriggerAndDedupe, arg.TriggerID, arg.DedupeKey)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const getWebhookDeliveryInWorkspace = `-- name: GetWebhookDeliveryInWorkspace :one
|
||
SELECT id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key FROM webhook_delivery
|
||
WHERE id = $1 AND workspace_id = $2
|
||
`
|
||
|
||
type GetWebhookDeliveryInWorkspaceParams struct {
|
||
ID pgtype.UUID `json:"id"`
|
||
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
||
}
|
||
|
||
// Workspace-scoped read for authenticated detail / replay endpoints.
|
||
func (q *Queries) GetWebhookDeliveryInWorkspace(ctx context.Context, arg GetWebhookDeliveryInWorkspaceParams) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, getWebhookDeliveryInWorkspace, arg.ID, arg.WorkspaceID)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const getWebhookReplayByIdempotencyKey = `-- name: GetWebhookReplayByIdempotencyKey :one
|
||
SELECT id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key FROM webhook_delivery
|
||
WHERE replayed_from_delivery_id = $1 AND replay_idempotency_key = $2
|
||
LIMIT 1
|
||
`
|
||
|
||
type GetWebhookReplayByIdempotencyKeyParams struct {
|
||
ReplayedFromDeliveryID pgtype.UUID `json:"replayed_from_delivery_id"`
|
||
ReplayIdempotencyKey pgtype.Text `json:"replay_idempotency_key"`
|
||
}
|
||
|
||
func (q *Queries) GetWebhookReplayByIdempotencyKey(ctx context.Context, arg GetWebhookReplayByIdempotencyKeyParams) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, getWebhookReplayByIdempotencyKey, arg.ReplayedFromDeliveryID, arg.ReplayIdempotencyKey)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const listWebhookDeliveriesByAutopilot = `-- name: ListWebhookDeliveriesByAutopilot :many
|
||
SELECT
|
||
d.id, d.workspace_id, d.autopilot_id, d.trigger_id, d.provider, d.event,
|
||
d.dedupe_key, d.dedupe_source, d.signature_status, d.status,
|
||
d.attempt_count, d.content_type, d.response_status,
|
||
d.autopilot_run_id, d.replayed_from_delivery_id, d.error,
|
||
d.received_at, d.last_attempt_at, d.created_at,
|
||
d.available_at, d.dispatch_attempts, d.reason_code, d.replay_idempotency_key
|
||
FROM webhook_delivery d
|
||
JOIN autopilot a ON a.id = d.autopilot_id
|
||
WHERE d.autopilot_id = $1
|
||
AND a.workspace_id = $2
|
||
ORDER BY d.created_at DESC
|
||
LIMIT $3 OFFSET $4
|
||
`
|
||
|
||
type ListWebhookDeliveriesByAutopilotParams struct {
|
||
AutopilotID pgtype.UUID `json:"autopilot_id"`
|
||
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
||
Limit int32 `json:"limit"`
|
||
Offset int32 `json:"offset"`
|
||
}
|
||
|
||
type ListWebhookDeliveriesByAutopilotRow struct {
|
||
ID pgtype.UUID `json:"id"`
|
||
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
||
AutopilotID pgtype.UUID `json:"autopilot_id"`
|
||
TriggerID pgtype.UUID `json:"trigger_id"`
|
||
Provider string `json:"provider"`
|
||
Event string `json:"event"`
|
||
DedupeKey pgtype.Text `json:"dedupe_key"`
|
||
DedupeSource pgtype.Text `json:"dedupe_source"`
|
||
SignatureStatus string `json:"signature_status"`
|
||
Status string `json:"status"`
|
||
AttemptCount int32 `json:"attempt_count"`
|
||
ContentType pgtype.Text `json:"content_type"`
|
||
ResponseStatus pgtype.Int4 `json:"response_status"`
|
||
AutopilotRunID pgtype.UUID `json:"autopilot_run_id"`
|
||
ReplayedFromDeliveryID pgtype.UUID `json:"replayed_from_delivery_id"`
|
||
Error pgtype.Text `json:"error"`
|
||
ReceivedAt pgtype.Timestamptz `json:"received_at"`
|
||
LastAttemptAt pgtype.Timestamptz `json:"last_attempt_at"`
|
||
CreatedAt pgtype.Timestamptz `json:"created_at"`
|
||
AvailableAt pgtype.Timestamptz `json:"available_at"`
|
||
DispatchAttempts int32 `json:"dispatch_attempts"`
|
||
ReasonCode pgtype.Text `json:"reason_code"`
|
||
ReplayIdempotencyKey pgtype.Text `json:"replay_idempotency_key"`
|
||
}
|
||
|
||
// Workspace-scoped via the join so a runId from another workspace cannot
|
||
// leak. Newest first, paged by limit/offset.
|
||
//
|
||
// Projection: large columns (`raw_body`, `selected_headers`, `response_body`)
|
||
// are deliberately excluded. A 100-row page × 256 KiB raw_body would be
|
||
// 25 MiB of bytes pulled from Postgres just to be dropped in the JSON
|
||
// encoder — Deliveries tab would hit that on every reload. Detail views
|
||
// fetch the full row via GetWebhookDelivery / GetWebhookDeliveryInWorkspace.
|
||
func (q *Queries) ListWebhookDeliveriesByAutopilot(ctx context.Context, arg ListWebhookDeliveriesByAutopilotParams) ([]ListWebhookDeliveriesByAutopilotRow, error) {
|
||
rows, err := q.db.Query(ctx, listWebhookDeliveriesByAutopilot,
|
||
arg.AutopilotID,
|
||
arg.WorkspaceID,
|
||
arg.Limit,
|
||
arg.Offset,
|
||
)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
items := []ListWebhookDeliveriesByAutopilotRow{}
|
||
for rows.Next() {
|
||
var i ListWebhookDeliveriesByAutopilotRow
|
||
if err := rows.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.ContentType,
|
||
&i.ResponseStatus,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
); err != nil {
|
||
return nil, err
|
||
}
|
||
items = append(items, i)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
return items, nil
|
||
}
|
||
|
||
const retryClaimedWebhookDelivery = `-- name: RetryClaimedWebhookDelivery :one
|
||
UPDATE webhook_delivery
|
||
SET available_at = $3,
|
||
dispatch_attempts = dispatch_attempts + 1,
|
||
error = $4,
|
||
lease_token = NULL,
|
||
lease_expires_at = NULL,
|
||
last_attempt_at = now()
|
||
WHERE id = $1
|
||
AND lease_token = $2
|
||
AND status = 'queued'
|
||
RETURNING id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key
|
||
`
|
||
|
||
type RetryClaimedWebhookDeliveryParams struct {
|
||
ID pgtype.UUID `json:"id"`
|
||
LeaseToken pgtype.UUID `json:"lease_token"`
|
||
AvailableAt pgtype.Timestamptz `json:"available_at"`
|
||
Error pgtype.Text `json:"error"`
|
||
}
|
||
|
||
// Records a transient worker failure and makes the delivery eligible after
|
||
// its bounded backoff. HTTP response fields intentionally remain the 202
|
||
// acknowledgement already returned to the provider.
|
||
func (q *Queries) RetryClaimedWebhookDelivery(ctx context.Context, arg RetryClaimedWebhookDeliveryParams) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, retryClaimedWebhookDelivery,
|
||
arg.ID,
|
||
arg.LeaseToken,
|
||
arg.AvailableAt,
|
||
arg.Error,
|
||
)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const updateWebhookDeliveryDispatched = `-- name: UpdateWebhookDeliveryDispatched :one
|
||
UPDATE webhook_delivery
|
||
SET status = $2,
|
||
autopilot_run_id = $3,
|
||
response_status = $4,
|
||
response_body = $5,
|
||
last_attempt_at = now()
|
||
WHERE id = $1
|
||
RETURNING id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key
|
||
`
|
||
|
||
type UpdateWebhookDeliveryDispatchedParams struct {
|
||
ID pgtype.UUID `json:"id"`
|
||
Status string `json:"status"`
|
||
AutopilotRunID pgtype.UUID `json:"autopilot_run_id"`
|
||
ResponseStatus pgtype.Int4 `json:"response_status"`
|
||
ResponseBody pgtype.Text `json:"response_body"`
|
||
}
|
||
|
||
// Finalises a delivery that successfully created (or skipped to) an
|
||
// autopilot_run. response_status is the HTTP status we returned, recorded
|
||
// alongside so the operator can correlate logs.
|
||
func (q *Queries) UpdateWebhookDeliveryDispatched(ctx context.Context, arg UpdateWebhookDeliveryDispatchedParams) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, updateWebhookDeliveryDispatched,
|
||
arg.ID,
|
||
arg.Status,
|
||
arg.AutopilotRunID,
|
||
arg.ResponseStatus,
|
||
arg.ResponseBody,
|
||
)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|
||
|
||
const updateWebhookDeliveryTerminal = `-- name: UpdateWebhookDeliveryTerminal :one
|
||
UPDATE webhook_delivery
|
||
SET status = $2,
|
||
error = $3,
|
||
reason_code = $4,
|
||
response_status = $5,
|
||
response_body = $6,
|
||
last_attempt_at = now()
|
||
WHERE id = $1
|
||
RETURNING id, workspace_id, autopilot_id, trigger_id, provider, event, dedupe_key, dedupe_source, signature_status, status, attempt_count, selected_headers, content_type, raw_body, response_status, response_body, autopilot_run_id, replayed_from_delivery_id, error, received_at, last_attempt_at, created_at, available_at, lease_token, lease_expires_at, dispatch_attempts, reason_code, replay_idempotency_key
|
||
`
|
||
|
||
type UpdateWebhookDeliveryTerminalParams struct {
|
||
ID pgtype.UUID `json:"id"`
|
||
Status string `json:"status"`
|
||
Error pgtype.Text `json:"error"`
|
||
ReasonCode pgtype.Text `json:"reason_code"`
|
||
ResponseStatus pgtype.Int4 `json:"response_status"`
|
||
ResponseBody pgtype.Text `json:"response_body"`
|
||
}
|
||
|
||
// Finalises a delivery without an autopilot_run link — rejected, ignored,
|
||
// failed. Separate query so callers can't accidentally drop the run_id when
|
||
// they only meant to set status/error.
|
||
func (q *Queries) UpdateWebhookDeliveryTerminal(ctx context.Context, arg UpdateWebhookDeliveryTerminalParams) (WebhookDelivery, error) {
|
||
row := q.db.QueryRow(ctx, updateWebhookDeliveryTerminal,
|
||
arg.ID,
|
||
arg.Status,
|
||
arg.Error,
|
||
arg.ReasonCode,
|
||
arg.ResponseStatus,
|
||
arg.ResponseBody,
|
||
)
|
||
var i WebhookDelivery
|
||
err := row.Scan(
|
||
&i.ID,
|
||
&i.WorkspaceID,
|
||
&i.AutopilotID,
|
||
&i.TriggerID,
|
||
&i.Provider,
|
||
&i.Event,
|
||
&i.DedupeKey,
|
||
&i.DedupeSource,
|
||
&i.SignatureStatus,
|
||
&i.Status,
|
||
&i.AttemptCount,
|
||
&i.SelectedHeaders,
|
||
&i.ContentType,
|
||
&i.RawBody,
|
||
&i.ResponseStatus,
|
||
&i.ResponseBody,
|
||
&i.AutopilotRunID,
|
||
&i.ReplayedFromDeliveryID,
|
||
&i.Error,
|
||
&i.ReceivedAt,
|
||
&i.LastAttemptAt,
|
||
&i.CreatedAt,
|
||
&i.AvailableAt,
|
||
&i.LeaseToken,
|
||
&i.LeaseExpiresAt,
|
||
&i.DispatchAttempts,
|
||
&i.ReasonCode,
|
||
&i.ReplayIdempotencyKey,
|
||
)
|
||
return i, err
|
||
}
|