Remove status on webhook data

Signed-off-by: Sacha Al Himdani <sacha@getprobo.com>

apps/console/src/__generated__/core/WebhooksSettingsPage_eventsQuery.graphql.ts

@@ -1,5 +1,5 @@
 /**
- * @generated SignedSource<<e16673382ec7f9083ba5df0b67d76583>>
+ * @generated SignedSource<<7427d67b4f9b27376c782cbe0601a5ec>>
  * @lightSyntaxTransform
  * @nogrep
  */
@@ -9,7 +9,7 @@
 // @ts-nocheck
 
 import { ConcreteRequest } from 'relay-runtime';
-export type WebhookEventStatus = "FAILED" | "SUCCEEDED";
+export type WebhookEventStatus = "FAILED" | "PENDING" | "SUCCEEDED";
 export type WebhooksSettingsPage_eventsQuery$variables = {
   after?: string | null | undefined;
   first?: number | null | undefined;

apps/console/src/pages/organizations/settings/WebhooksSettingsPage.tsx

@@ -289,6 +289,9 @@ function EventStatusBadge({ status }: { status: string }) {
   if (status === "SUCCEEDED") {
     return <Badge variant="success" size="sm">{__("Succeeded")}</Badge>;
   }
+  if (status === "PENDING") {
+    return <Badge variant="info" size="sm">{__("Pending")}</Badge>;
+  }
   return <Badge variant="danger" size="sm">{__("Failed")}</Badge>;
 }
 

pkg/coredata/migrations/20260210T135740Z.sql

@@ -18,25 +18,18 @@ CREATE TABLE webhook_configurations (
     updated_at TIMESTAMP WITH TIME ZONE NOT NULL
 );
 
-CREATE TYPE webhook_data_status AS ENUM (
-    'PENDING',
-    'PROCESSING',
-    'PROCESSED',
-    'FAILED'
-);
-
 CREATE TABLE webhook_data (
     id TEXT PRIMARY KEY,
     tenant_id TEXT NOT NULL,
     organization_id TEXT NOT NULL REFERENCES organizations(id) ON UPDATE CASCADE ON DELETE CASCADE,
     event_type webhook_event_type NOT NULL,
-    status webhook_data_status NOT NULL,
     data JSONB NOT NULL,
     created_at TIMESTAMP WITH TIME ZONE NOT NULL,
     processed_at TIMESTAMP WITH TIME ZONE
 );
 
 CREATE TYPE webhook_event_status AS ENUM (
+    'PENDING',
     'SUCCEEDED',
     'FAILED'
 );

pkg/coredata/webhook_data.go

@@ -32,7 +32,6 @@ type (
 		ID             gid.GID         `db:"id"`
 		OrganizationID gid.GID         `db:"organization_id"`
 		EventType      WebhookEventType `db:"event_type"`
-		Status         WebhookDataStatus `db:"status"`
 		Data           json.RawMessage `db:"data"`
 		CreatedAt      time.Time       `db:"created_at"`
 		ProcessedAt    *time.Time      `db:"processed_at"`
@@ -52,7 +51,6 @@ INSERT INTO webhook_data (
     tenant_id,
     organization_id,
     event_type,
-    status,
     data,
     created_at
 )
@@ -61,7 +59,6 @@ VALUES (
     @tenant_id,
     @organization_id,
     @event_type,
-    @status,
     @data,
     @created_at
 )
@@ -72,7 +69,6 @@ VALUES (
 		"tenant_id":       scope.GetTenantID(),
 		"organization_id": w.OrganizationID,
 		"event_type":      w.EventType,
-		"status":          w.Status,
 		"data":            w.Data,
 		"created_at":      w.CreatedAt,
 	}
@@ -85,7 +81,7 @@ VALUES (
 	return nil
 }
 
-func (w *WebhookData) LoadNextPendingForUpdate(
+func (w *WebhookData) LoadNextUnprocessedForUpdate(
 	ctx context.Context,
 	conn pg.Conn,
 ) error {
@@ -94,12 +90,11 @@ SELECT
     id,
     organization_id,
     event_type,
-    status,
     data,
     created_at,
     processed_at
 FROM webhook_data
-WHERE status = 'PENDING'
+WHERE processed_at IS NULL
 ORDER BY created_at ASC
 LIMIT 1
 FOR UPDATE SKIP LOCKED
@@ -107,7 +102,7 @@ FOR UPDATE SKIP LOCKED
 
 	rows, err := conn.Query(ctx, q)
 	if err != nil {
-		return fmt.Errorf("cannot query pending webhook data: %w", err)
+		return fmt.Errorf("cannot query unprocessed webhook data: %w", err)
 	}
 
 	data, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[WebhookData])
@@ -123,16 +118,14 @@ FOR UPDATE SKIP LOCKED
 	return nil
 }
 
-func (w *WebhookData) UpdateStatus(
+func (w *WebhookData) UpdateProcessedAt(
 	ctx context.Context,
 	conn pg.Conn,
 	scope Scoper,
 ) error {
 	q := `
 UPDATE webhook_data
-SET
+SET processed_at = @processed_at
-    status = @status,
-    processed_at = @processed_at
 WHERE %s
     AND id = @id
 `
@@ -141,7 +134,6 @@ WHERE %s
 
 	args := pgx.StrictNamedArgs{
 		"id":           w.ID,
-		"status":       w.Status.String(),
 		"processed_at": w.ProcessedAt,
 	}
 	maps.Copy(args, scope.SQLArguments())

pkg/coredata/webhook_data_status.go

@@ -1,68 +0,0 @@
-// Copyright (c) 2025 Probo Inc <hello@getprobo.com>.
-//
-// Permission to use, copy, modify, and/or distribute this software for any
-// purpose with or without fee is hereby granted, provided that the above
-// copyright notice and this permission notice appear in all copies.
-//
-// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
-// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
-// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
-// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
-// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
-// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
-// PERFORMANCE OF THIS SOFTWARE.
-
-package coredata
-
-import (
-	"database/sql/driver"
-	"fmt"
-)
-
-type WebhookDataStatus string
-
-const (
-	WebhookDataStatusPending    WebhookDataStatus = "PENDING"
-	WebhookDataStatusProcessing WebhookDataStatus = "PROCESSING"
-	WebhookDataStatusProcessed  WebhookDataStatus = "PROCESSED"
-	WebhookDataStatusFailed     WebhookDataStatus = "FAILED"
-)
-
-func (s WebhookDataStatus) String() string {
-	return string(s)
-}
-
-func (s WebhookDataStatus) IsValid() bool {
-	switch s {
-	case WebhookDataStatusPending, WebhookDataStatusProcessing, WebhookDataStatusProcessed, WebhookDataStatusFailed:
-		return true
-	}
-	return false
-}
-
-func (s WebhookDataStatus) MarshalText() ([]byte, error) {
-	return []byte(s.String()), nil
-}
-
-func (s *WebhookDataStatus) UnmarshalText(text []byte) error {
-	*s = WebhookDataStatus(text)
-	if !s.IsValid() {
-		return fmt.Errorf("%s is not a valid WebhookDataStatus", string(text))
-	}
-	return nil
-}
-
-func (s *WebhookDataStatus) Scan(value any) error {
-	switch v := value.(type) {
-	case string:
-		return s.UnmarshalText([]byte(v))
-	case []byte:
-		return s.UnmarshalText(v)
-	default:
-		return fmt.Errorf("unsupported type for WebhookDataStatus: %T", value)
-	}
-}
-
-func (s WebhookDataStatus) Value() (driver.Value, error) {
-	return s.String(), nil
-}

pkg/coredata/webhook_event.go

@@ -159,3 +159,38 @@ VALUES (
 
 	return nil
 }
+
+func (w *WebhookEvent) UpdateStatus(
+	ctx context.Context,
+	conn pg.Conn,
+	scope Scoper,
+) error {
+	q := `
+UPDATE webhook_events
+SET
+    status = @status,
+    response = @response
+WHERE %s
+    AND id = @id
+`
+
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{
+		"id":       w.ID,
+		"status":   w.Status,
+		"response": w.Response,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot update webhook event: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}

pkg/coredata/webhook_event_status.go

@@ -22,6 +22,7 @@ import (
 type WebhookEventStatus string
 
 const (
+	WebhookEventStatusPending WebhookEventStatus = "PENDING"
 	WebhookEventStatusSucceeded  WebhookEventStatus = "SUCCEEDED"
 	WebhookEventStatusFailed     WebhookEventStatus = "FAILED"
 )
@@ -32,7 +33,7 @@ func (s WebhookEventStatus) String() string {
 
 func (s WebhookEventStatus) IsValid() bool {
 	switch s {
-	case WebhookEventStatusSucceeded, WebhookEventStatusFailed:
+	case WebhookEventStatusPending, WebhookEventStatusSucceeded, WebhookEventStatusFailed:
 		return true
 	}
 	return false

pkg/server/api/console/v1/schema.graphql

@@ -2320,6 +2320,8 @@ type WebhookConfigurationEdge {
 
 enum WebhookEventStatus
     @goModel(model: "go.probo.inc/probo/pkg/coredata.WebhookEventStatus") {
+    PENDING
+        @goEnum(value: "go.probo.inc/probo/pkg/coredata.WebhookEventStatusPending")
     SUCCEEDED
         @goEnum(value: "go.probo.inc/probo/pkg/coredata.WebhookEventStatusSucceeded")
     FAILED

pkg/server/api/console/v1/schema/schema.go

@@ -12907,6 +12907,8 @@ type WebhookConfigurationEdge {
 
 enum WebhookEventStatus
     @goModel(model: "go.probo.inc/probo/pkg/coredata.WebhookEventStatus") {
+    PENDING
+        @goEnum(value: "go.probo.inc/probo/pkg/coredata.WebhookEventStatusPending")
     SUCCEEDED
         @goEnum(value: "go.probo.inc/probo/pkg/coredata.WebhookEventStatusSucceeded")
     FAILED
@@ -100921,10 +100923,12 @@ func (ec *executionContext) marshalNWebhookEventStatus2goᚗproboᚗincᚋprobo
 
 var (
 	unmarshalNWebhookEventStatus2goᚗproboᚗincᚋproboᚋpkgᚋcoredataᚐWebhookEventStatus = map[string]coredata.WebhookEventStatus{
+		"PENDING":   coredata.WebhookEventStatusPending,
 		"SUCCEEDED": coredata.WebhookEventStatusSucceeded,
 		"FAILED":    coredata.WebhookEventStatusFailed,
 	}
 	marshalNWebhookEventStatus2goᚗproboᚗincᚋproboᚋpkgᚋcoredataᚐWebhookEventStatus = map[coredata.WebhookEventStatus]string{
+		coredata.WebhookEventStatusPending:   "PENDING",
 		coredata.WebhookEventStatusSucceeded: "SUCCEEDED",
 		coredata.WebhookEventStatusFailed:    "FAILED",
 	}

pkg/webhook/data.go

@@ -52,7 +52,6 @@ func InsertEvent(
 		ID:             gid.New(scope.GetTenantID(), coredata.WebhookDataEntityType),
 		OrganizationID: organizationID,
 		EventType:      eventType,
-		Status:         coredata.WebhookDataStatusPending,
 		Data:           raw,
 		CreatedAt:      time.Now(),
 	}

pkg/webhook/sender.go

@@ -61,6 +61,11 @@ type (
 		CacheTTL      time.Duration
 		EncryptionKey cipher.EncryptionKey
 	}
+
+	pendingDelivery struct {
+		Event  *coredata.WebhookEvent
+		Config *coredata.WebhookConfiguration
+	}
 )
 
 const maxResponseBodySize = 64 * 1024 // 64KB
@@ -110,7 +115,7 @@ func (s *Sender) processEvents(ctx context.Context) error {
 	}
 
 	for {
-		webhookData, err := s.claimNextWebhookData(ctx)
+		webhookData, deliveries, err := s.claimNextWebhookData(ctx)
 		if err != nil {
 			if errors.Is(err, coredata.ErrResourceNotFound) {
 				return nil
@@ -118,98 +123,85 @@ func (s *Sender) processEvents(ctx context.Context) error {
 			return fmt.Errorf("cannot claim next webhook data: %w", err)
 		}
 
-		s.processWebhookData(ctx, webhookData)
+		s.processDeliveries(ctx, webhookData, deliveries)
 	}
 }
 
-func (s *Sender) claimNextWebhookData(ctx context.Context) (*coredata.WebhookData, error) {
+func (s *Sender) claimNextWebhookData(ctx context.Context) (*coredata.WebhookData, []pendingDelivery, error) {
 	var webhookData coredata.WebhookData
+	var deliveries []pendingDelivery
 
 	err := s.pg.WithTx(ctx, func(tx pg.Conn) error {
-		if err := webhookData.LoadNextPendingForUpdate(ctx, tx); err != nil {
+		if err := webhookData.LoadNextUnprocessedForUpdate(ctx, tx); err != nil {
-			return fmt.Errorf("cannot load next pending webhook data: %w", err)
+			return fmt.Errorf("cannot load next unprocessed webhook data: %w", err)
 		}
 
 		scope := coredata.NewScopeFromObjectID(webhookData.ID)
 
-		webhookData.Status = coredata.WebhookDataStatusProcessing
+		var configs coredata.WebhookConfigurations
+		if err := configs.LoadMatchingByOrganizationIDAndEventType(
+			ctx, tx, scope, webhookData.OrganizationID, webhookData.EventType,
+		); err != nil {
+			return fmt.Errorf("cannot load matching webhook configurations: %w", err)
+		}
 
-		if err := webhookData.UpdateStatus(ctx, tx, scope); err != nil {
+		now := time.Now()
-			return fmt.Errorf("cannot update webhook data to processing: %w", err)
+
+		for _, config := range configs {
+			event := &coredata.WebhookEvent{
+				ID:                     gid.New(webhookData.ID.TenantID(), coredata.WebhookEventEntityType),
+				WebhookDataID:          webhookData.ID,
+				WebhookConfigurationID: config.ID,
+				Status:                 coredata.WebhookEventStatusPending,
+				CreatedAt:              now,
+			}
+
+			if err := event.Insert(ctx, tx, scope); err != nil {
+				return fmt.Errorf("cannot insert webhook event: %w", err)
+			}
+
+			deliveries = append(deliveries, pendingDelivery{
+				Event:  event,
+				Config: config,
+			})
+		}
+
+		webhookData.ProcessedAt = &now
+		if err := webhookData.UpdateProcessedAt(ctx, tx, scope); err != nil {
+			return fmt.Errorf("cannot update webhook data processed_at: %w", err)
 		}
 
 		return nil
 	})
 
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
-	return &webhookData, nil
+	return &webhookData, deliveries, nil
 }
 
-func (s *Sender) processWebhookData(ctx context.Context, webhookData *coredata.WebhookData) {
+func (s *Sender) processDeliveries(ctx context.Context, webhookData *coredata.WebhookData, deliveries []pendingDelivery) {
-	scope := coredata.NewScopeFromObjectID(webhookData.ID)
+	for _, d := range deliveries {
-
+		s.deliver(ctx, webhookData, d)
-	var configs coredata.WebhookConfigurations
-	err := s.pg.WithConn(ctx, func(conn pg.Conn) error {
-		return configs.LoadMatchingByOrganizationIDAndEventType(
-			ctx, conn, scope, webhookData.OrganizationID, webhookData.EventType,
-		)
-	})
-	if err != nil {
-		s.logger.ErrorCtx(ctx, "cannot load matching webhook configurations",
-			log.Error(err),
-			log.String("webhook_data_id", webhookData.ID.String()),
-		)
-		s.markWebhookData(ctx, webhookData, scope, coredata.WebhookDataStatusFailed)
-		return
-	}
-
-	for _, config := range configs {
-		s.deliverToConfiguration(ctx, webhookData, config, scope)
-	}
-
-	s.markWebhookData(ctx, webhookData, scope, coredata.WebhookDataStatusProcessed)
-}
-
-func (s *Sender) markWebhookData(ctx context.Context, webhookData *coredata.WebhookData, scope coredata.Scoper, status coredata.WebhookDataStatus) {
-	now := time.Now()
-	webhookData.Status = status
-	webhookData.ProcessedAt = &now
-
-	err := s.pg.WithConn(ctx, func(conn pg.Conn) error {
-		return webhookData.UpdateStatus(ctx, conn, scope)
-	})
-	if err != nil {
-		s.logger.ErrorCtx(ctx, "cannot update webhook data status",
-			log.Error(err),
-			log.String("webhook_data_id", webhookData.ID.String()),
-			log.String("target_status", status.String()),
-		)
 	}
 }
 
-func (s *Sender) deliverToConfiguration(
+func (s *Sender) deliver(ctx context.Context, webhookData *coredata.WebhookData, d pendingDelivery) {
-	ctx context.Context,
+	scope := coredata.NewScopeFromObjectID(d.Event.ID)
-	webhookData *coredata.WebhookData,
-	config *coredata.WebhookConfiguration,
-	scope coredata.Scoper,
-) {
-	eventID := gid.New(webhookData.ID.TenantID(), coredata.WebhookEventEntityType)
 
-	signingSecret, err := s.getSigningSecret(config.ID.String(), config.EncryptedSigningSecret)
+	signingSecret, err := s.getSigningSecret(d.Config.ID.String(), d.Config.EncryptedSigningSecret)
 	if err != nil {
 		s.logger.ErrorCtx(ctx, "cannot get signing secret",
 			log.Error(err),
 			log.String("webhook_data_id", webhookData.ID.String()),
-			log.String("configuration_id", config.ID.String()),
+			log.String("configuration_id", d.Config.ID.String()),
 		)
-		s.recordEvent(ctx, eventID, webhookData, config, scope, coredata.WebhookEventStatusFailed, nil)
+		s.updateEventStatus(ctx, d.Event, scope, coredata.WebhookEventStatusFailed, nil)
 		return
 	}
 
-	response, sendErr := s.doHTTPCall(ctx, eventID, config.EndpointURL, webhookData, config.ID, signingSecret)
+	response, sendErr := s.doHTTPCall(ctx, d.Event.ID, d.Config.EndpointURL, webhookData, d.Config.ID, signingSecret)
 
 	eventStatus := coredata.WebhookEventStatusSucceeded
 	if sendErr != nil {
@@ -217,10 +209,33 @@ func (s *Sender) deliverToConfiguration(
 		s.logger.ErrorCtx(ctx, "error delivering webhook",
 			log.Error(sendErr),
 			log.String("webhook_data_id", webhookData.ID.String()),
+			log.String("event_id", d.Event.ID.String()),
 		)
 	}
 
-	s.recordEvent(ctx, eventID, webhookData, config, scope, eventStatus, response)
+	s.updateEventStatus(ctx, d.Event, scope, eventStatus, response)
+}
+
+func (s *Sender) updateEventStatus(
+	ctx context.Context,
+	event *coredata.WebhookEvent,
+	scope coredata.Scoper,
+	status coredata.WebhookEventStatus,
+	response json.RawMessage,
+) {
+	event.Status = status
+	event.Response = response
+
+	err := s.pg.WithConn(ctx, func(conn pg.Conn) error {
+		return event.UpdateStatus(ctx, conn, scope)
+	})
+	if err != nil {
+		s.logger.ErrorCtx(ctx, "cannot update webhook event status",
+			log.Error(err),
+			log.String("event_id", event.ID.String()),
+			log.String("target_status", status.String()),
+		)
+	}
 }
 
 func (s *Sender) getSigningSecret(webhookConfigurationID string, encryptedSigningSecret []byte) (string, error) {
@@ -304,39 +319,6 @@ func (s *Sender) doHTTPCall(
 	}
 }
 
-func (s *Sender) recordEvent(
-	ctx context.Context,
-	eventID gid.GID,
-	webhookData *coredata.WebhookData,
-	config *coredata.WebhookConfiguration,
-	scope coredata.Scoper,
-	status coredata.WebhookEventStatus,
-	response json.RawMessage,
-) {
-	event := coredata.WebhookEvent{
-		ID:                     eventID,
-		WebhookDataID:          webhookData.ID,
-		WebhookConfigurationID: config.ID,
-		Status:                 status,
-		Response:               response,
-		CreatedAt:              time.Now(),
-	}
-
-	err := s.pg.WithConn(ctx, func(conn pg.Conn) error {
-		if err := event.Insert(ctx, conn, scope); err != nil {
-			return fmt.Errorf("cannot insert webhook event: %w", err)
-		}
-		return nil
-	})
-	if err != nil {
-		s.logger.ErrorCtx(ctx, "cannot insert webhook event",
-			log.Error(err),
-			log.String("webhook_data_id", webhookData.ID.String()),
-			log.String("configuration_id", config.ID.String()),
-		)
-	}
-}
-
 func buildResponseJSON(resp *http.Response, body []byte) json.RawMessage {
 	headers := make(map[string]any, len(resp.Header))
 	for k, v := range resp.Header {