feat: inbound webhooks — public ingest endpoints

core.inbound_hooks (migration 00013): POST /api/v1/hooks/<token> accepts
external events. event mode republishes as inbound.received through the
signed dispatcher + realtime; job mode triggers a scheduled-job run with
the received body as payload. Optional X-Primora-Signature HMAC verify.
Dashboard card on the Integrations page.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Tomas DvorakandDevin committed 2026-09-24 19:15:12 +02:00
1 parent 2f4de543a9
commit 45a1621909
26 files changed
+1264 -9

No files matched your search

@@ -0,0 +1,173 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.29.0
// source: inbound_hooks.sql
package db
import (
"context"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
)
const createInboundHook = `-- name: CreateInboundHook :one
INSERT INTO core.inbound_hooks (project_id, name, token, mode, job_id, secret, enabled)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at
`
type CreateInboundHookParams struct {
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name"`
Token string `json:"token"`
Mode string `json:"mode"`
JobID pgtype.UUID `json:"job_id"`
Secret []byte `json:"secret"`
Enabled bool `json:"enabled"`
}
func (q *Queries) CreateInboundHook(ctx context.Context, arg CreateInboundHookParams) (CoreInboundHook, error) {
row := q.db.QueryRow(ctx, createInboundHook,
arg.ProjectID,
arg.Name,
arg.Token,
arg.Mode,
arg.JobID,
arg.Secret,
arg.Enabled,
)
var i CoreInboundHook
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Token,
&i.Mode,
&i.JobID,
&i.Secret,
&i.Enabled,
&i.LastReceivedAt,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const deleteInboundHook = `-- name: DeleteInboundHook :exec
DELETE FROM core.inbound_hooks
WHERE id = $1 AND project_id = $2
`
type DeleteInboundHookParams struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
}
func (q *Queries) DeleteInboundHook(ctx context.Context, arg DeleteInboundHookParams) error {
_, err := q.db.Exec(ctx, deleteInboundHook, arg.ID, arg.ProjectID)
return err
}
const getInboundHook = `-- name: GetInboundHook :one
SELECT id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at FROM core.inbound_hooks
WHERE id = $1 AND project_id = $2
`
type GetInboundHookParams struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
}
func (q *Queries) GetInboundHook(ctx context.Context, arg GetInboundHookParams) (CoreInboundHook, error) {
row := q.db.QueryRow(ctx, getInboundHook, arg.ID, arg.ProjectID)
var i CoreInboundHook
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Token,
&i.Mode,
&i.JobID,
&i.Secret,
&i.Enabled,
&i.LastReceivedAt,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const getInboundHookByToken = `-- name: GetInboundHookByToken :one
SELECT id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at FROM core.inbound_hooks
WHERE token = $1
`
func (q *Queries) GetInboundHookByToken(ctx context.Context, token string) (CoreInboundHook, error) {
row := q.db.QueryRow(ctx, getInboundHookByToken, token)
var i CoreInboundHook
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Token,
&i.Mode,
&i.JobID,
&i.Secret,
&i.Enabled,
&i.LastReceivedAt,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const listInboundHooks = `-- name: ListInboundHooks :many
SELECT id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at FROM core.inbound_hooks
WHERE project_id = $1
ORDER BY name
`
func (q *Queries) ListInboundHooks(ctx context.Context, projectID uuid.UUID) ([]CoreInboundHook, error) {
rows, err := q.db.Query(ctx, listInboundHooks, projectID)
if err != nil {
return nil, err
}
defer rows.Close()
var items []CoreInboundHook
for rows.Next() {
var i CoreInboundHook
if err := rows.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Token,
&i.Mode,
&i.JobID,
&i.Secret,
&i.Enabled,
&i.LastReceivedAt,
&i.CreatedAt,
&i.UpdatedAt,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const touchInboundHook = `-- name: TouchInboundHook :exec
UPDATE core.inbound_hooks
SET last_received_at = NOW()
WHERE id = $1
`
func (q *Queries) TouchInboundHook(ctx context.Context, id uuid.UUID) error {
_, err := q.db.Exec(ctx, touchInboundHook, id)
return err
}
@@ -255,6 +255,20 @@ type CoreEvent struct {
Ts pgtype.Timestamptz `json:"ts"`
}
type CoreInboundHook struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name"`
Token string `json:"token"`
Mode string `json:"mode"`
JobID pgtype.UUID `json:"job_id"`
Secret []byte `json:"secret"`
Enabled bool `json:"enabled"`
LastReceivedAt pgtype.Timestamptz `json:"last_received_at"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
}
type CoreIntegration struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
@@ -30,6 +30,7 @@ type Querier interface {
CreateCollection(ctx context.Context, arg CreateCollectionParams) (CoreCollection, error)
CreateDBConnection(ctx context.Context, arg CreateDBConnectionParams) (CoreDbConnection, error)
CreateDocument(ctx context.Context, arg CreateDocumentParams) (CoreDocument, error)
CreateInboundHook(ctx context.Context, arg CreateInboundHookParams) (CoreInboundHook, error)
CreateIntegration(ctx context.Context, arg CreateIntegrationParams) (CoreIntegration, error)
CreateInvitation(ctx context.Context, arg CreateInvitationParams) (CoreProjectInvitation, error)
CreateOrganization(ctx context.Context, arg CreateOrganizationParams) (CoreOrganization, error)
@@ -43,6 +44,7 @@ type Querier interface {
DeleteComponent(ctx context.Context, arg DeleteComponentParams) error
DeleteDBConnection(ctx context.Context, arg DeleteDBConnectionParams) (CoreDbConnection, error)
DeleteDocument(ctx context.Context, arg DeleteDocumentParams) error
DeleteInboundHook(ctx context.Context, arg DeleteInboundHookParams) error
DeleteIntegration(ctx context.Context, arg DeleteIntegrationParams) (CoreIntegration, error)
DeleteOrganizationByID(ctx context.Context, id uuid.UUID) (CoreOrganization, error)
DeletePendingInvitationByIDForOrganization(ctx context.Context, arg DeletePendingInvitationByIDForOrganizationParams) (CoreProjectInvitation, error)
@@ -65,6 +67,8 @@ type Querier interface {
GetDBConnectionByID(ctx context.Context, id uuid.UUID) (CoreDbConnection, error)
GetDBConnectionByName(ctx context.Context, arg GetDBConnectionByNameParams) (CoreDbConnection, error)
GetDocumentByID(ctx context.Context, arg GetDocumentByIDParams) (CoreDocument, error)
GetInboundHook(ctx context.Context, arg GetInboundHookParams) (CoreInboundHook, error)
GetInboundHookByToken(ctx context.Context, token string) (CoreInboundHook, error)
GetIntegrationByID(ctx context.Context, id uuid.UUID) (CoreIntegration, error)
GetInvitationByIDForOrganization(ctx context.Context, arg GetInvitationByIDForOrganizationParams) (CoreProjectInvitation, error)
GetInvitationByTokenHash(ctx context.Context, tokenHash string) (CoreProjectInvitation, error)
@@ -99,6 +103,7 @@ type Querier interface {
ListEnabledAlertRules(ctx context.Context) ([]CoreAlertRule, error)
ListErrorGroups(ctx context.Context, arg ListErrorGroupsParams) ([]ListErrorGroupsRow, error)
ListEvents(ctx context.Context, arg ListEventsParams) ([]ListEventsRow, error)
ListInboundHooks(ctx context.Context, projectID uuid.UUID) ([]CoreInboundHook, error)
ListIntegrations(ctx context.Context, projectID uuid.UUID) ([]CoreIntegration, error)
ListInvitationsForOrganization(ctx context.Context, organizationID uuid.UUID) ([]ListInvitationsForOrganizationRow, error)
ListOrganizationMembers(ctx context.Context, organizationID uuid.UUID) ([]ListOrganizationMembersRow, error)
@@ -138,6 +143,7 @@ type Querier interface {
SweepExpiredOrgAuditLogs(ctx context.Context) (int64, error)
SweepExpiredWebhookDeliveries(ctx context.Context) (int64, error)
TouchAPIKey(ctx context.Context, id uuid.UUID) error
TouchInboundHook(ctx context.Context, id uuid.UUID) error
UpdateAlertRule(ctx context.Context, arg UpdateAlertRuleParams) (CoreAlertRule, error)
UpdateBucketByID(ctx context.Context, arg UpdateBucketByIDParams) (CoreBucket, error)
UpdateCollection(ctx context.Context, arg UpdateCollectionParams) (CoreCollection, error)
+4
View File
@@ -79,6 +79,10 @@ func (h *HTTPHandler) Register(router *gin.Engine) {
api.DELETE("/collections/:collectionID/documents/:documentID", h.deleteDocument)
api.OPTIONS("/ingest", h.ingestOptions)
api.POST("/ingest", h.ingest)
api.POST("/hooks/:token", h.receiveInboundHook)
api.GET("/projects/:projectID/inbound-hooks", h.listInboundHooks)
api.POST("/projects/:projectID/inbound-hooks", h.createInboundHook)
api.DELETE("/projects/:projectID/inbound-hooks/:hookID", h.deleteInboundHook)
api.GET("/projects/:projectID/events", h.listEvents)
api.GET("/projects/:projectID/issues", h.listIssues)
api.GET("/projects/:projectID/components", h.listComponents)
+97
View File
@@ -0,0 +1,97 @@
package handlers
import (
"errors"
"io"
"net/http"
"strings"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/tdvorak/primora/apps/backend/internal/middleware"
apperrors "github.com/tdvorak/primora/apps/backend/internal/response"
"github.com/tdvorak/primora/apps/backend/internal/services"
)
const maxInboundBody = 256 * 1024 // external POSTs capped at 256 KiB
func (h *HTTPHandler) inboundError(c *gin.Context, err error) {
if errors.Is(err, pgx.ErrNoRows) {
apperrors.Abort(c, http.StatusNotFound, "not_found", "hook not found")
return
}
msg := err.Error()
switch {
case strings.Contains(msg, "invalid signature"):
apperrors.Abort(c, http.StatusUnauthorized, "unauthorized", msg)
case strings.Contains(msg, "invalid") || strings.Contains(msg, "already exists") || strings.Contains(msg, "disabled") || strings.Contains(msg, "requires") || strings.Contains(msg, "not found in this project"):
apperrors.Abort(c, http.StatusBadRequest, "bad_request", msg)
default:
h.handleError(c, err)
}
}
func (h *HTTPHandler) listInboundHooks(c *gin.Context) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return
}
result, err := h.Platform.ListInboundHooks(c.Request.Context(), actor, projectID)
if err != nil {
h.inboundError(c, err)
return
}
c.JSON(http.StatusOK, gin.H{"items": result})
}
func (h *HTTPHandler) createInboundHook(c *gin.Context) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return
}
var body services.CreateInboundHookInput
if !h.bindAndValidate(c, &body) {
return
}
result, err := h.Platform.CreateInboundHook(c.Request.Context(), actor, projectID, body, middleware.RequestIDFromContext(c))
if err != nil {
h.inboundError(c, err)
return
}
c.JSON(http.StatusCreated, result)
}
func (h *HTTPHandler) deleteInboundHook(c *gin.Context) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return
}
hookID, err := uuid.Parse(c.Param("hookID"))
if err != nil {
apperrors.Abort(c, http.StatusBadRequest, "bad_request", "invalid hook id")
return
}
if err := h.Platform.DeleteInboundHook(c.Request.Context(), actor, projectID, hookID, middleware.RequestIDFromContext(c)); err != nil {
h.inboundError(c, err)
return
}
c.Status(http.StatusNoContent)
}
// receiveInboundHook is the public ingest endpoint — the URL token is the
// credential, so no actor is required.
func (h *HTTPHandler) receiveInboundHook(c *gin.Context) {
body, err := io.ReadAll(http.MaxBytesReader(c.Writer, c.Request.Body, maxInboundBody))
if err != nil {
apperrors.Abort(c, http.StatusRequestEntityTooLarge, "too_large", "body exceeds 256 KiB")
return
}
result, err := h.Platform.ReceiveInboundHook(c.Request.Context(), c.Param("token"), c.GetHeader("X-Primora-Signature"), body)
if err != nil {
h.inboundError(c, err)
return
}
c.JSON(http.StatusAccepted, result)
}
+249
View File
@@ -0,0 +1,249 @@
package services
import (
"context"
"crypto/rand"
"crypto/subtle"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
db "github.com/tdvorak/primora/apps/backend/internal/database/db"
"github.com/tdvorak/primora/apps/backend/internal/models"
)
// Inbound hooks are project-scoped public ingest endpoints:
// POST /api/v1/hooks/<token>. A hook either republishes the body as an
// `inbound.received` domain event (webhooks + realtime fan out for free) or
// triggers a scheduled-job run with the body as the payload. When a secret is
// set, callers must sign the raw body with X-Primora-Signature — the same
// HMAC-SHA256 scheme outbound deliveries use.
const (
WebhookEventInboundReceived = "inbound.received"
InboundModeEvent = "event"
InboundModeJob = "job"
)
type InboundHookSummary struct {
ID string `json:"id"`
ProjectID string `json:"project_id"`
Name string `json:"name"`
Token string `json:"token"`
URL string `json:"url"`
Mode string `json:"mode"`
JobID string `json:"job_id,omitempty"`
HasSecret bool `json:"has_secret"`
Enabled bool `json:"enabled"`
LastReceivedAt *time.Time `json:"last_received_at"`
CreatedAt time.Time `json:"created_at"`
}
type CreateInboundHookInput struct {
Name string `json:"name" validate:"required"`
Mode string `json:"mode"`
JobID string `json:"job_id"`
Secret string `json:"secret"`
Enabled *bool `json:"enabled"`
}
type InboundReceiveResult struct {
Received bool `json:"received"`
Event string `json:"event,omitempty"`
RunID string `json:"run_id,omitempty"`
}
func inboundHookSummary(row db.CoreInboundHook, publicURL string) InboundHookSummary {
out := InboundHookSummary{
ID: row.ID.String(),
ProjectID: row.ProjectID.String(),
Name: row.Name,
Token: row.Token,
URL: strings.TrimSuffix(publicURL, "/") + "/api/v1/hooks/" + row.Token,
Mode: row.Mode,
HasSecret: len(row.Secret) > 0,
Enabled: row.Enabled,
CreatedAt: row.CreatedAt.Time,
}
if row.JobID.Valid {
out.JobID = uuid.UUID(row.JobID.Bytes).String()
}
if row.LastReceivedAt.Valid {
t := row.LastReceivedAt.Time
out.LastReceivedAt = &t
}
return out
}
func (s *PlatformService) ListInboundHooks(ctx context.Context, actor *models.Actor, projectID uuid.UUID) ([]InboundHookSummary, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
return nil, err
}
rows, err := s.repo.Queries().ListInboundHooks(ctx, projectID)
if err != nil {
return nil, err
}
out := make([]InboundHookSummary, len(rows))
for i, r := range rows {
out[i] = inboundHookSummary(r, s.publicURL)
}
return out, nil
}
func (s *PlatformService) CreateInboundHook(ctx context.Context, actor *models.Actor, projectID uuid.UUID, input CreateInboundHookInput, requestID string) (InboundHookSummary, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
return InboundHookSummary{}, err
}
if !projectSecretNameRe.MatchString(input.Name) {
return InboundHookSummary{}, fmt.Errorf("invalid hook name %q — use [A-Za-z_][A-Za-z0-9_]*", input.Name)
}
mode := input.Mode
if mode == "" {
mode = InboundModeEvent
}
var jobID pgtype.UUID
if mode == InboundModeJob {
parsed, err := uuid.Parse(input.JobID)
if err != nil {
return InboundHookSummary{}, errors.New("job mode requires a valid job_id")
}
job, err := s.repo.Queries().GetScheduledJobByID(ctx, parsed)
if err != nil || job.ProjectID != projectID {
return InboundHookSummary{}, errors.New("job not found in this project")
}
jobID = pgtype.UUID{Bytes: parsed, Valid: true}
} else if mode != InboundModeEvent {
return InboundHookSummary{}, fmt.Errorf("invalid mode %q — event or job", input.Mode)
}
secretBytes := []byte{}
if input.Secret != "" {
if s.enc == nil {
return InboundHookSummary{}, errors.New("encryption is not configured — cannot store hook secrets")
}
var err error
if secretBytes, err = s.enc.Encrypt([]byte(input.Secret)); err != nil {
return InboundHookSummary{}, err
}
}
token := "prm_hook_" + randHex(16)
enabled := true
if input.Enabled != nil {
enabled = *input.Enabled
}
row, err := s.repo.Queries().CreateInboundHook(ctx, db.CreateInboundHookParams{
ProjectID: projectID,
Name: input.Name,
Token: token,
Mode: mode,
JobID: jobID,
Secret: secretBytes,
Enabled: enabled,
})
if err != nil {
if strings.Contains(err.Error(), "inbound_hooks_project_id_name_key") {
return InboundHookSummary{}, fmt.Errorf("an inbound hook named %s already exists", input.Name)
}
return InboundHookSummary{}, err
}
project, err := s.repo.Queries().GetProjectByID(ctx, projectID)
if err == nil {
_, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, projectID, actor, requestID, "inbound_hook.created", "inbound_hook", row.ID.String(), map[string]any{
"name": row.Name, "mode": row.Mode,
}))
}
return inboundHookSummary(row, s.publicURL), nil
}
func (s *PlatformService) DeleteInboundHook(ctx context.Context, actor *models.Actor, projectID, hookID uuid.UUID, requestID string) error {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
return err
}
if _, err := s.repo.Queries().GetInboundHook(ctx, db.GetInboundHookParams{ID: hookID, ProjectID: projectID}); err != nil {
return err
}
if err := s.repo.Queries().DeleteInboundHook(ctx, db.DeleteInboundHookParams{ID: hookID, ProjectID: projectID}); err != nil {
return err
}
project, err := s.repo.Queries().GetProjectByID(ctx, projectID)
if err == nil {
_, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, projectID, actor, requestID, "inbound_hook.deleted", "inbound_hook", hookID.String(), nil))
}
return nil
}
// ReceiveInboundHook handles POST /api/v1/hooks/:token — unauthenticated; the
// token is the credential. Signature check applies when a secret is set.
func (s *PlatformService) ReceiveInboundHook(ctx context.Context, token string, signature string, body []byte) (InboundReceiveResult, error) {
hook, err := s.repo.Queries().GetInboundHookByToken(ctx, token)
if err != nil {
return InboundReceiveResult{}, err // pgx.ErrNoRows → 404
}
if !hook.Enabled {
return InboundReceiveResult{}, errors.New("hook is disabled")
}
if len(hook.Secret) > 0 {
if s.enc == nil {
return InboundReceiveResult{}, errors.New("encryption is not configured")
}
plain, err := s.enc.Decrypt(hook.Secret)
if err != nil {
return InboundReceiveResult{}, err
}
want := "sha256=" + signBody(string(plain), body)
if subtle.ConstantTimeCompare([]byte(signature), []byte(want)) != 1 {
return InboundReceiveResult{}, errors.New("invalid signature")
}
}
var payload map[string]any
if len(body) > 0 {
if err := json.Unmarshal(body, &payload); err != nil {
return InboundReceiveResult{}, fmt.Errorf("invalid JSON body: %w", err)
}
}
_ = s.repo.Queries().TouchInboundHook(ctx, hook.ID)
switch hook.Mode {
case InboundModeJob:
if s.scheduler == nil {
return InboundReceiveResult{}, errors.New("scheduler unavailable")
}
if !hook.JobID.Valid {
return InboundReceiveResult{}, errors.New("hook has no job configured")
}
job, err := s.repo.Queries().GetScheduledJobByID(ctx, hook.JobID.Bytes)
if err != nil {
return InboundReceiveResult{}, err
}
if payload != nil {
raw, _ := json.Marshal(payload)
job.Payload = raw // received body replaces the configured payload
}
run, err := s.scheduler.enqueue(ctx, job, "hook")
if err != nil {
return InboundReceiveResult{}, err
}
return InboundReceiveResult{Received: true, Event: WebhookEventJobRun, RunID: run.ID.String()}, nil
default:
s.publishEvent(ctx, hook.ProjectID, WebhookEventInboundReceived, map[string]any{
"hook_id": hook.ID.String(),
"hook_name": hook.Name,
"payload": payload,
})
return InboundReceiveResult{Received: true, Event: WebhookEventInboundReceived}, nil
}
}
func randHex(n int) string {
b := make([]byte, n)
if _, err := rand.Read(b); err != nil {
panic(err)
}
return hex.EncodeToString(b)
}
@@ -47,6 +47,7 @@ var webhookEventTypes = map[string]bool{
WebhookEventTest: true,
WebhookEventAlertFired: true,
WebhookEventAlertResolved: true,
WebhookEventInboundReceived: true,
WebhookEventJobRun: true,
WebhookEventDocumentCreated: true,
WebhookEventDocumentUpdated: true,