package api import ( "bytes" "context" "crypto/hmac" "crypto/rand" "crypto/sha256" "database/sql" "encoding/hex" "encoding/json" "errors" "io" "net/http" "net/url" "strings" "sync" "time" "containr/internal/database" "containr/internal/database/sqlcdb" "github.com/gin-gonic/gin" "github.com/google/uuid" "github.com/sqlc-dev/pqtype" ) // Outbound webhooks: every audited action (LogAudit) emits an event named // "resource.action" (e.g. "service.deploy"). Webhooks subscribe with an // event list supporting "service.*" and "*" wildcards. Deliveries are // HMAC-signed and retried up to 3 times with backoff; each attempt lands // in webhook_deliveries. var ( webhookQueue = make(chan webhookJob, 512) webhookWorkers sync.Once ) type webhookJob struct { Webhook sqlcdb.OutboundWebhook Event string Payload []byte } type webhookEventPayload struct { Event string `json:"event"` Resource string `json:"resource"` ResourceID string `json:"resource_id"` Action string `json:"action"` UserID string `json:"user_id,omitempty"` Details map[string]interface{} `json:"details,omitempty"` Timestamp time.Time `json:"timestamp"` } // emitWebhookEvent fans an audited action out to all enabled webhooks whose // event list matches. Called from LogAudit — fire-and-forget. func emitWebhookEvent(userID, resource, resourceID, action string, details map[string]interface{}) { db := GetAuditDB() if db == nil { return } // Actions that already carry a resource prefix ("service.clone") pass // through unchanged; bare actions ("create") get resource-qualified. event := action if !strings.Contains(action, ".") { event = resource + "." + action } queries := sqlcdb.New(db.DB) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() hooks, err := queries.ListEnabledOutboundWebhooks(ctx) if err != nil || len(hooks) == 0 { return } payload, err := json.Marshal(webhookEventPayload{ Event: event, Resource: resource, ResourceID: resourceID, Action: action, UserID: userID, Details: details, Timestamp: time.Now().UTC(), }) if err != nil { return } uid, uidErr := uuid.Parse(strings.TrimSpace(userID)) for _, hook := range hooks { if !webhookEventMatches(hook.Events, event) { continue } // Webhooks only receive events for their owner — unless subscribed to "*" // is deliberate admin reach; owner scoping keeps per-user isolation. if uidErr != nil || hook.UserID != uid { continue } select { case webhookQueue <- webhookJob{Webhook: hook, Event: event, Payload: payload}: default: // Queue full: drop rather than block the request path. } } } func webhookEventMatches(events json.RawMessage, event string) bool { var subs []string if err := json.Unmarshal(events, &subs); err != nil || len(subs) == 0 { return false } for _, sub := range subs { sub = strings.TrimSpace(sub) if sub == "*" || sub == event { return true } if strings.HasSuffix(sub, ".*") && strings.HasPrefix(event, strings.TrimSuffix(sub, "*")) { return true } } return false } // StartWebhookDispatcher launches delivery workers; call once at boot. func StartWebhookDispatcher() { webhookWorkers.Do(func() { for i := 0; i < 4; i++ { go webhookDeliveryWorker() } }) } func webhookDeliveryWorker() { for job := range webhookQueue { deliverWebhook(job) } } var webhookHTTPClient = &http.Client{Timeout: 10 * time.Second} func deliverWebhook(job webhookJob) { db := GetAuditDB() if db == nil { return } queries := sqlcdb.New(db.DB) ctx := context.Background() deliveryID := uuid.New() _ = queries.CreateWebhookDelivery(ctx, sqlcdb.CreateWebhookDeliveryParams{ ID: deliveryID, WebhookID: job.Webhook.ID, Event: job.Event, Payload: pqtype.NullRawMessage{RawMessage: job.Payload, Valid: true}, Status: "pending", }) var lastStatus string = "failed" var respStatus sql.NullInt32 var respBody sql.NullString var durationMs sql.NullInt32 var deliveredAt sql.NullTime for attempt := int32(1); attempt <= 3; attempt++ { start := time.Now() status, code, body := postWebhook(job, deliveryID, attempt) durationMs = sql.NullInt32{Int32: int32(time.Since(start).Milliseconds()), Valid: true} respStatus = code respBody = body if status == "success" { lastStatus = "success" deliveredAt = sql.NullTime{Time: time.Now().UTC(), Valid: true} _ = queries.UpdateWebhookDeliveryResult(ctx, sqlcdb.UpdateWebhookDeliveryResultParams{ ID: deliveryID, Status: lastStatus, ResponseStatus: respStatus, ResponseBody: respBody, Attempts: attempt, DurationMs: durationMs, DeliveredAt: deliveredAt, }) return } if attempt < 3 { time.Sleep(time.Duration(attempt*attempt) * 2 * time.Second) } } _ = queries.UpdateWebhookDeliveryResult(ctx, sqlcdb.UpdateWebhookDeliveryResultParams{ ID: deliveryID, Status: lastStatus, ResponseStatus: respStatus, ResponseBody: respBody, Attempts: 3, DurationMs: durationMs, DeliveredAt: deliveredAt, }) } func postWebhook(job webhookJob, deliveryID uuid.UUID, attempt int32) (string, sql.NullInt32, sql.NullString) { req, err := http.NewRequest(http.MethodPost, job.Webhook.Url, bytes.NewReader(job.Payload)) if err != nil { return "failed", sql.NullInt32{}, sql.NullString{String: err.Error(), Valid: true} } req.Header.Set("Content-Type", "application/json") req.Header.Set("User-Agent", "Containr-Webhook/1.0") req.Header.Set("X-Containr-Event", job.Event) req.Header.Set("X-Containr-Delivery", deliveryID.String()) if job.Webhook.Secret != "" { mac := hmac.New(sha256.New, []byte(job.Webhook.Secret)) mac.Write(job.Payload) req.Header.Set("X-Containr-Signature", "sha256="+hex.EncodeToString(mac.Sum(nil))) } if job.Webhook.Headers.Valid { var custom map[string]string if err := json.Unmarshal(job.Webhook.Headers.RawMessage, &custom); err == nil { for k, v := range custom { req.Header.Set(k, v) } } } resp, err := webhookHTTPClient.Do(req) if err != nil { return "failed", sql.NullInt32{}, sql.NullString{String: err.Error(), Valid: true} } defer resp.Body.Close() body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) code := sql.NullInt32{Int32: int32(resp.StatusCode), Valid: true} bodyStr := sql.NullString{String: string(body), Valid: len(body) > 0} if resp.StatusCode >= 200 && resp.StatusCode < 300 { return "success", code, bodyStr } return "failed", code, bodyStr } // ---------- HTTP API ---------- type outboundWebhookResponse struct { ID string `json:"id"` Name string `json:"name"` URL string `json:"url"` Secret string `json:"secret,omitempty"` Events []string `json:"events"` Headers map[string]string `json:"headers,omitempty"` Enabled bool `json:"enabled"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } func outboundWebhookJSON(row sqlcdb.OutboundWebhook, revealSecret bool) outboundWebhookResponse { var events []string _ = json.Unmarshal(row.Events, &events) if events == nil { events = []string{} } var headers map[string]string if row.Headers.Valid { _ = json.Unmarshal(row.Headers.RawMessage, &headers) } resp := outboundWebhookResponse{ ID: row.ID.String(), Name: row.Name, URL: row.Url, Events: events, Headers: headers, Enabled: row.Enabled, } if revealSecret { resp.Secret = row.Secret } if row.CreatedAt.Valid { resp.CreatedAt = row.CreatedAt.Time } if row.UpdatedAt.Valid { resp.UpdatedAt = row.UpdatedAt.Time } return resp } func validWebhookURL(raw string) error { u, err := url.Parse(strings.TrimSpace(raw)) if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" { return errors.New("url must be a valid http(s) URL") } return nil } type webhookMutationRequest struct { Name string `json:"name"` URL string `json:"url"` Secret *string `json:"secret"` Events []string `json:"events"` Headers map[string]string `json:"headers"` Enabled *bool `json:"enabled"` } func handleListOutboundWebhooks(c *gin.Context) { userID, ok := requireAuthenticatedUserUUID(c) if !ok { return } db := c.MustGet("db").(*database.DB) rows, err := sqlcdb.New(db.DB).ListOutboundWebhooksByUser(c.Request.Context(), userID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to list webhooks"}) return } out := make([]outboundWebhookResponse, 0, len(rows)) for _, row := range rows { out = append(out, outboundWebhookJSON(row, false)) } c.JSON(http.StatusOK, gin.H{"webhooks": out}) } func handleCreateOutboundWebhook(c *gin.Context) { userID, ok := requireAuthenticatedUserUUID(c) if !ok { return } var req webhookMutationRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } req.Name = strings.TrimSpace(req.Name) if req.Name == "" { respondError(c, http.StatusBadRequest, "VALIDATION", "name is required") return } if err := validWebhookURL(req.URL); err != nil { respondError(c, http.StatusBadRequest, "VALIDATION", err.Error()) return } if len(req.Events) == 0 { respondError(c, http.StatusBadRequest, "VALIDATION", "at least one event subscription is required (e.g. \"service.*\" or \"*\")") return } secret := "" if req.Secret != nil { secret = strings.TrimSpace(*req.Secret) } if secret == "" { buf := make([]byte, 24) _, _ = rand.Read(buf) secret = hex.EncodeToString(buf) } eventsJSON, _ := json.Marshal(req.Events) var headersJSON pqtype.NullRawMessage if len(req.Headers) > 0 { if raw, err := json.Marshal(req.Headers); err == nil { headersJSON = pqtype.NullRawMessage{RawMessage: raw, Valid: true} } } enabled := true if req.Enabled != nil { enabled = *req.Enabled } db := c.MustGet("db").(*database.DB) id := uuid.New() err := sqlcdb.New(db.DB).CreateOutboundWebhook(c.Request.Context(), sqlcdb.CreateOutboundWebhookParams{ ID: id, UserID: userID, Name: req.Name, Url: strings.TrimSpace(req.URL), Secret: secret, Events: eventsJSON, Headers: headersJSON, Enabled: enabled, }) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create webhook"}) return } LogAuditWithRequest(c, "webhook", id.String(), "create", map[string]interface{}{"name": req.Name}) c.JSON(http.StatusCreated, outboundWebhookJSON(sqlcdb.OutboundWebhook{ ID: id, Name: req.Name, Url: strings.TrimSpace(req.URL), Secret: secret, Events: eventsJSON, Headers: headersJSON, Enabled: enabled, CreatedAt: sql.NullTime{Time: time.Now(), Valid: true}, UpdatedAt: sql.NullTime{Time: time.Now(), Valid: true}, }, true)) } func handleUpdateOutboundWebhook(c *gin.Context) { userID, ok := requireAuthenticatedUserUUID(c) if !ok { return } webhookID, err := uuid.Parse(c.Param("id")) if err != nil { respondError(c, http.StatusBadRequest, "VALIDATION", "Invalid webhook id") return } db := c.MustGet("db").(*database.DB) queries := sqlcdb.New(db.DB) row, err := queries.GetOutboundWebhookByIDAndUser(c.Request.Context(), sqlcdb.GetOutboundWebhookByIDAndUserParams{ ID: webhookID, UserID: userID, }) if errors.Is(err, sql.ErrNoRows) { respondError(c, http.StatusNotFound, "NOT_FOUND", "Webhook not found") return } if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to load webhook"}) return } var req webhookMutationRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } name := row.Name if req.Name != "" { name = strings.TrimSpace(req.Name) } targetURL := row.Url if req.URL != "" { if err := validWebhookURL(req.URL); err != nil { respondError(c, http.StatusBadRequest, "VALIDATION", err.Error()) return } targetURL = strings.TrimSpace(req.URL) } secret := row.Secret if req.Secret != nil { secret = strings.TrimSpace(*req.Secret) } eventsJSON := row.Events if req.Events != nil { if len(req.Events) == 0 { respondError(c, http.StatusBadRequest, "VALIDATION", "at least one event subscription is required") return } eventsJSON, _ = json.Marshal(req.Events) } headersJSON := row.Headers if req.Headers != nil { if raw, err := json.Marshal(req.Headers); err == nil { headersJSON = pqtype.NullRawMessage{RawMessage: raw, Valid: true} } } enabled := row.Enabled if req.Enabled != nil { enabled = *req.Enabled } err = queries.UpdateOutboundWebhook(c.Request.Context(), sqlcdb.UpdateOutboundWebhookParams{ ID: webhookID, UserID: userID, Name: name, Url: targetURL, Secret: secret, Events: eventsJSON, Headers: headersJSON, Enabled: enabled, }) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update webhook"}) return } LogAuditWithRequest(c, "webhook", webhookID.String(), "update", map[string]interface{}{"name": name}) row.Name, row.Url, row.Secret, row.Events, row.Headers, row.Enabled = name, targetURL, secret, eventsJSON, headersJSON, enabled c.JSON(http.StatusOK, outboundWebhookJSON(row, true)) } func handleDeleteOutboundWebhook(c *gin.Context) { userID, ok := requireAuthenticatedUserUUID(c) if !ok { return } webhookID, err := uuid.Parse(c.Param("id")) if err != nil { respondError(c, http.StatusBadRequest, "VALIDATION", "Invalid webhook id") return } db := c.MustGet("db").(*database.DB) err = sqlcdb.New(db.DB).DeleteOutboundWebhookByIDAndUser(c.Request.Context(), sqlcdb.DeleteOutboundWebhookByIDAndUserParams{ ID: webhookID, UserID: userID, }) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to delete webhook"}) return } LogAuditWithRequest(c, "webhook", webhookID.String(), "delete", map[string]interface{}{}) c.JSON(http.StatusOK, gin.H{"message": "Webhook deleted"}) } func handleListOutboundWebhookDeliveries(c *gin.Context) { userID, ok := requireAuthenticatedUserUUID(c) if !ok { return } webhookID, err := uuid.Parse(c.Param("id")) if err != nil { respondError(c, http.StatusBadRequest, "VALIDATION", "Invalid webhook id") return } db := c.MustGet("db").(*database.DB) queries := sqlcdb.New(db.DB) if _, err := queries.GetOutboundWebhookByIDAndUser(c.Request.Context(), sqlcdb.GetOutboundWebhookByIDAndUserParams{ ID: webhookID, UserID: userID, }); errors.Is(err, sql.ErrNoRows) { respondError(c, http.StatusNotFound, "NOT_FOUND", "Webhook not found") return } rows, err := queries.ListWebhookDeliveriesByWebhook(c.Request.Context(), webhookID) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to list deliveries"}) return } c.JSON(http.StatusOK, gin.H{"deliveries": rows}) } func handleTestOutboundWebhook(c *gin.Context) { userID, ok := requireAuthenticatedUserUUID(c) if !ok { return } webhookID, err := uuid.Parse(c.Param("id")) if err != nil { respondError(c, http.StatusBadRequest, "VALIDATION", "Invalid webhook id") return } db := c.MustGet("db").(*database.DB) row, err := sqlcdb.New(db.DB).GetOutboundWebhookByIDAndUser(c.Request.Context(), sqlcdb.GetOutboundWebhookByIDAndUserParams{ ID: webhookID, UserID: userID, }) if errors.Is(err, sql.ErrNoRows) { respondError(c, http.StatusNotFound, "NOT_FOUND", "Webhook not found") return } if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to load webhook"}) return } if !row.Enabled { respondError(c, http.StatusBadRequest, "VALIDATION", "Webhook is disabled") return } payload, _ := json.Marshal(webhookEventPayload{ Event: "ping", Resource: "webhook", ResourceID: webhookID.String(), Action: "ping", UserID: userID.String(), Timestamp: time.Now().UTC(), }) select { case webhookQueue <- webhookJob{Webhook: row, Event: "ping", Payload: payload}: c.JSON(http.StatusAccepted, gin.H{"message": "Test delivery queued"}) default: respondError(c, http.StatusServiceUnavailable, "UNAVAILABLE", "Delivery queue is full") } }