From 6de906d0a980c3e0d0086877363de60de8070b83 Mon Sep 17 00:00:00 2001 From: Sacha Al Himdani Date: Fri, 13 Feb 2026 16:23:38 +0100 Subject: [PATCH] Remove status on webhook data Signed-off-by: Sacha Al Himdani --- ...ebhooksSettingsPage_eventsQuery.graphql.ts | 4 +- .../settings/WebhooksSettingsPage.tsx | 3 + pkg/coredata/migrations/20260210T135740Z.sql | 9 +- pkg/coredata/webhook_data.go | 30 ++-- pkg/coredata/webhook_data_status.go | 68 ------- pkg/coredata/webhook_event.go | 35 ++++ pkg/coredata/webhook_event_status.go | 7 +- pkg/server/api/console/v1/schema.graphql | 2 + pkg/server/api/console/v1/schema/schema.go | 4 + pkg/webhook/data.go | 1 - pkg/webhook/sender.go | 170 ++++++++---------- 11 files changed, 138 insertions(+), 195 deletions(-) delete mode 100644 pkg/coredata/webhook_data_status.go diff --git a/apps/console/src/__generated__/core/WebhooksSettingsPage_eventsQuery.graphql.ts b/apps/console/src/__generated__/core/WebhooksSettingsPage_eventsQuery.graphql.ts index 16e304490..fc9c59632 100644 --- a/apps/console/src/__generated__/core/WebhooksSettingsPage_eventsQuery.graphql.ts +++ b/apps/console/src/__generated__/core/WebhooksSettingsPage_eventsQuery.graphql.ts @@ -1,5 +1,5 @@ /** - * @generated SignedSource<> + * @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; diff --git a/apps/console/src/pages/organizations/settings/WebhooksSettingsPage.tsx b/apps/console/src/pages/organizations/settings/WebhooksSettingsPage.tsx index b4a40e187..4c202e515 100644 --- a/apps/console/src/pages/organizations/settings/WebhooksSettingsPage.tsx +++ b/apps/console/src/pages/organizations/settings/WebhooksSettingsPage.tsx @@ -289,6 +289,9 @@ function EventStatusBadge({ status }: { status: string }) { if (status === "SUCCEEDED") { return {__("Succeeded")}; } + if (status === "PENDING") { + return {__("Pending")}; + } return {__("Failed")}; } diff --git a/pkg/coredata/migrations/20260210T135740Z.sql b/pkg/coredata/migrations/20260210T135740Z.sql index 456481f05..961f0e68f 100644 --- a/pkg/coredata/migrations/20260210T135740Z.sql +++ b/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' ); diff --git a/pkg/coredata/webhook_data.go b/pkg/coredata/webhook_data.go index 9317eab39..8e5527662 100644 --- a/pkg/coredata/webhook_data.go +++ b/pkg/coredata/webhook_data.go @@ -29,13 +29,12 @@ import ( type ( WebhookData struct { - 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"` + ID gid.GID `db:"id"` + OrganizationID gid.GID `db:"organization_id"` + EventType WebhookEventType `db:"event_type"` + Data json.RawMessage `db:"data"` + CreatedAt time.Time `db:"created_at"` + ProcessedAt *time.Time `db:"processed_at"` } WebhookDataList []*WebhookData @@ -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 - status = @status, - processed_at = @processed_at +SET 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()) diff --git a/pkg/coredata/webhook_data_status.go b/pkg/coredata/webhook_data_status.go deleted file mode 100644 index f0705443a..000000000 --- a/pkg/coredata/webhook_data_status.go +++ /dev/null @@ -1,68 +0,0 @@ -// Copyright (c) 2025 Probo Inc . -// -// 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 -} diff --git a/pkg/coredata/webhook_event.go b/pkg/coredata/webhook_event.go index 52589491d..a5fcc84c1 100644 --- a/pkg/coredata/webhook_event.go +++ b/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 +} diff --git a/pkg/coredata/webhook_event_status.go b/pkg/coredata/webhook_event_status.go index b65e5483c..88e39bb2f 100644 --- a/pkg/coredata/webhook_event_status.go +++ b/pkg/coredata/webhook_event_status.go @@ -22,8 +22,9 @@ import ( type WebhookEventStatus string const ( - WebhookEventStatusSucceeded WebhookEventStatus = "SUCCEEDED" - WebhookEventStatusFailed WebhookEventStatus = "FAILED" + WebhookEventStatusPending WebhookEventStatus = "PENDING" + WebhookEventStatusSucceeded WebhookEventStatus = "SUCCEEDED" + WebhookEventStatusFailed WebhookEventStatus = "FAILED" ) func (s WebhookEventStatus) String() string { @@ -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 diff --git a/pkg/server/api/console/v1/schema.graphql b/pkg/server/api/console/v1/schema.graphql index cd288d30f..6042afeb7 100644 --- a/pkg/server/api/console/v1/schema.graphql +++ b/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 diff --git a/pkg/server/api/console/v1/schema/schema.go b/pkg/server/api/console/v1/schema/schema.go index 3163ceb46..4d0bb22d0 100644 --- a/pkg/server/api/console/v1/schema/schema.go +++ b/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", } diff --git a/pkg/webhook/data.go b/pkg/webhook/data.go index 285d3baa9..649914178 100644 --- a/pkg/webhook/data.go +++ b/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(), } diff --git a/pkg/webhook/sender.go b/pkg/webhook/sender.go index fc0dd52ad..4a28f61ca 100644 --- a/pkg/webhook/sender.go +++ b/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 { - return fmt.Errorf("cannot load next pending webhook data: %w", err) + if err := webhookData.LoadNextUnprocessedForUpdate(ctx, tx); err != nil { + 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 { - return fmt.Errorf("cannot update webhook data to processing: %w", err) + now := time.Now() + + 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) { - scope := coredata.NewScopeFromObjectID(webhookData.ID) - - 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) processDeliveries(ctx context.Context, webhookData *coredata.WebhookData, deliveries []pendingDelivery) { + for _, d := range deliveries { + s.deliver(ctx, webhookData, d) } } -func (s *Sender) deliverToConfiguration( - ctx context.Context, - webhookData *coredata.WebhookData, - config *coredata.WebhookConfiguration, - scope coredata.Scoper, -) { - eventID := gid.New(webhookData.ID.TenantID(), coredata.WebhookEventEntityType) +func (s *Sender) deliver(ctx context.Context, webhookData *coredata.WebhookData, d pendingDelivery) { + scope := coredata.NewScopeFromObjectID(d.Event.ID) - 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 {