Function writeOutboxRow

internal/webhookpub/outbox.go:182–254

writeOutboxRow is the SQL body shared by PublishTx and (eventually) PublishBestEffortTx. Idempotent on (id): a retried trigger with the same deterministic id no-ops the second INSERT. Issues pg_notify so the slice-2 worker wakes immediately on commit.

(ctx context.Context, exec outboxExecutor, e Event)

// same deterministic id no-ops the second INSERT. Issues pg_notify so
// the slice-2 worker wakes immediately on commit.
func writeOutboxRow(ctx context.Context, exec outboxExecutor, e Event) error {
	if e.ID == "" {
		return fmt.Errorf("webhookpub: outbox event must have non-empty ID")
	}
	if e.UserID == "" {
		return fmt.Errorf("webhookpub: outbox event must have non-empty UserID")
	}
	if e.Type == "" {
		return fmt.Errorf("webhookpub: outbox event must have non-empty Type")
	}

	envelopeJSON, err := json.Marshal(e.AsEnvelope())
	if err != nil {
		return fmt.Errorf("webhookpub: marshal envelope: %w", err)
	}

	var messageID *string
	if e.MessageID != "" {
		mid := e.MessageID
		messageID = &mid
	}
	var agentID *string
	if e.AgentID != "" {
		aid := e.AgentID
		agentID = &aid
	}
	var conversationID *string
	if e.ConversationID != "" {
		cid := e.ConversationID
		conversationID = &cid
	}

	// created_at and expires_at use the column DEFAULTs so the
	// timestamps come from the Postgres server clock (one clock per
	// primary writer; no application-side skew).
	_, err = exec.Exec(ctx,
		`INSERT INTO webhook_events
		    (id, user_id, type, aud, envelope, schema_version,
		     agent_id, conversation_id, message_id, status)
		 VALUES ($1, $2, $3, 'webhook', $4, 1, $5, $6, $7, 'pending')
		 ON CONFLICT (id) DO NOTHING`,
		e.ID, e.UserID, e.Type, envelopeJSON,
		agentID, conversationID, messageID,
	)
	if err != nil {
		return fmt.Errorf("webhookpub: insert webhook_events: %w", err)
	}

	// pg_notify is best-effort: NOTIFY only fires on COMMIT (Postgres
	// queues it). If COMMIT fails, no notification is emitted. The
	// slice-2 worker's 1s fallback poll catches missed wakeups
	// (deploy windows, LISTEN reconnect races). Payload is empty
	// because the worker rescans the table anyway.
	//
	// A pg_notify error here (NOTIFY queue overflow is the realistic
	// case; max_notify_queue_pages defaults to 1024 × 8KB = 8MB) is a
	// SOFT failure: we log and return nil so the caller's tx still
	// commits. The webhook_events row is what matters for at-least-