diff --git a/cmd/probod/CHANGELOG.md b/cmd/probod/CHANGELOG.md index 4a8895c0e..73b2f8a92 100644 --- a/cmd/probod/CHANGELOG.md +++ b/cmd/probod/CHANGELOG.md @@ -4,6 +4,10 @@ All notable changes to `probod` (the server, including the bundled `@probo/conso ## Unreleased +### Added + +- Webhook deliveries for `*:updated` events now include a top-level `updatedFrom` field alongside `data`, carrying a full snapshot of the entity as it was before the update (e.g. the prior membership role on `user:updated`). The field is omitted for non-update events + ## [0.225.0] - 2026-07-13 ### Added diff --git a/packages/n8n-node/CHANGELOG.md b/packages/n8n-node/CHANGELOG.md index bd552e049..9303ec42f 100644 --- a/packages/n8n-node/CHANGELOG.md +++ b/packages/n8n-node/CHANGELOG.md @@ -4,6 +4,10 @@ All notable changes to the `@probo/n8n-nodes-probo` package will be documented i ## Unreleased +### Added + +- Probo Trigger output for `*:updated` events now includes an `updatedFrom` object alongside `data`, holding the entity snapshot from before the update so workflows can diff old vs new values (for example `{{ $json.updatedFrom.membership.role }}` vs `{{ $json.data.membership.role }}`) + ## [0.201.0] - 2026-07-09 ### Added diff --git a/packages/n8n-node/README.md b/packages/n8n-node/README.md index 72ab0f9bd..7090e0395 100644 --- a/packages/n8n-node/README.md +++ b/packages/n8n-node/README.md @@ -103,6 +103,15 @@ Probo Trigger → Slack 4. **Activate the workflow.** n8n registers a webhook subscription in Probo. When a document version is published, Probo delivers the event and the Slack message is sent. +### Update events carry the previous state + +For `*:updated` events, the payload includes an `updatedFrom` object next to `data`, holding a full snapshot of the entity as it was before the update. This lets a workflow react to what actually changed — for example, only notify when a user's role changes: + +- **Condition:** `{{ $json.data.membership.role !== $json.updatedFrom.membership.role }}` +- **Text:** `Role changed from {{ $json.updatedFrom.membership.role }} to {{ $json.data.membership.role }}` + +`updatedFrom` is present only on update events; it is absent for created, deleted, and other lifecycle events. + ### Alternative: list open tasks on a schedule Use the **Probo** action node without a trigger: diff --git a/pkg/coredata/migrations/20260715T080636Z.sql b/pkg/coredata/migrations/20260715T080636Z.sql new file mode 100644 index 000000000..a39bc8d35 --- /dev/null +++ b/pkg/coredata/migrations/20260715T080636Z.sql @@ -0,0 +1,22 @@ +-- Copyright (c) 2026 Probo Inc . +-- +-- Permission is hereby granted, free of charge, to any person obtaining a copy +-- of this software and associated documentation files (the "Software"), to deal +-- in the Software without restriction, including without limitation the rights +-- to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +-- copies of the Software, and to permit persons to whom the Software is +-- furnished to do so, subject to the following conditions: +-- +-- The above copyright notice and this permission notice shall be included in +-- all copies or substantial portions of the Software. +-- +-- THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +-- IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +-- FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +-- AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +-- LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +-- OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +-- SOFTWARE. + +ALTER TABLE webhook_data + ADD COLUMN updated_from JSONB; diff --git a/pkg/coredata/webhook_data.go b/pkg/coredata/webhook_data.go index 968de0544..7aa3005e8 100644 --- a/pkg/coredata/webhook_data.go +++ b/pkg/coredata/webhook_data.go @@ -39,6 +39,7 @@ type ( OrganizationID gid.GID `db:"organization_id"` EventType WebhookEventType `db:"event_type"` Data json.RawMessage `db:"data"` + UpdatedFrom json.RawMessage `db:"updated_from"` CreatedAt time.Time `db:"created_at"` ProcessedAt *time.Time `db:"processed_at"` } @@ -58,6 +59,7 @@ INSERT INTO webhook_data ( organization_id, event_type, data, + updated_from, created_at ) VALUES ( @@ -66,6 +68,7 @@ VALUES ( @organization_id, @event_type, @data, + @updated_from, @created_at ) ` @@ -76,6 +79,7 @@ VALUES ( "organization_id": w.OrganizationID, "event_type": w.EventType, "data": w.Data, + "updated_from": w.UpdatedFrom, "created_at": w.CreatedAt, } @@ -97,6 +101,7 @@ SELECT organization_id, event_type, data, + updated_from, created_at, processed_at FROM webhook_data diff --git a/pkg/iam/organization_service.go b/pkg/iam/organization_service.go index 8c296d75b..5212c1a85 100644 --- a/pkg/iam/organization_service.go +++ b/pkg/iam/organization_service.go @@ -293,6 +293,8 @@ func (s *OrganizationService) UpdateMembership( } } + previousUser := webhooktypes.NewUser(profile, &membership) + membership.Role = role membership.UpdatedAt = time.Now() @@ -300,7 +302,7 @@ func (s *OrganizationService) UpdateMembership( return fmt.Errorf("cannot update membership: %w", err) } - if err := webhook.InsertData(ctx, tx, scope, organizationID, coredata.WebhookEventTypeUserUpdated, webhooktypes.NewUser(profile, &membership)); err != nil { + if err := webhook.InsertUpdateData(ctx, tx, scope, organizationID, coredata.WebhookEventTypeUserUpdated, webhooktypes.NewUser(profile, &membership), previousUser); err != nil { return fmt.Errorf("cannot insert webhook event: %w", err) } @@ -439,6 +441,8 @@ func (s *OrganizationService) ArchiveUser( return fmt.Errorf("cannot delete requested signatures: %w", err) } + previousUser := webhooktypes.NewUser(&profile, membership) + now := time.Now() if profile.State != coredata.ProfileStateInactive { @@ -455,7 +459,7 @@ func (s *OrganizationService) ArchiveUser( return fmt.Errorf("cannot update membership: %w", err) } - if err := webhook.InsertData(ctx, tx, scope, profile.OrganizationID, coredata.WebhookEventTypeUserUpdated, webhooktypes.NewUser(&profile, membership)); err != nil { + if err := webhook.InsertUpdateData(ctx, tx, scope, profile.OrganizationID, coredata.WebhookEventTypeUserUpdated, webhooktypes.NewUser(&profile, membership), previousUser); err != nil { return fmt.Errorf("cannot insert webhook event: %w", err) } @@ -1099,6 +1103,8 @@ func (s *OrganizationService) UpdateUser(ctx context.Context, req *UpdateUserReq return fmt.Errorf("cannot load profile: %w", err) } + previousProfile := *profile + if profile.Source != coredata.ProfileSourceSCIM { profile.FullName = req.FullName profile.Kind = req.Kind @@ -1130,7 +1136,10 @@ func (s *OrganizationService) UpdateUser(ctx context.Context, req *UpdateUserReq membership := &coredata.Membership{} - var webhookPayload *webhooktypes.User + var ( + webhookPayload *webhooktypes.User + previousUser *webhooktypes.User + ) if err := membership.LoadByIdentityIDAndOrganizationID(ctx, conn, scope, profile.IdentityID, profile.OrganizationID); err != nil { if !errors.Is(err, coredata.ErrResourceNotFound) { @@ -1138,16 +1147,20 @@ func (s *OrganizationService) UpdateUser(ctx context.Context, req *UpdateUserReq } webhookPayload = webhooktypes.NewUser(profile, nil) + previousUser = webhooktypes.NewUser(&previousProfile, nil) } else { webhookPayload = webhooktypes.NewUser(profile, membership) + previousUser = webhooktypes.NewUser(&previousProfile, membership) } - if err := webhook.InsertData(ctx, + if err := webhook.InsertUpdateData( + ctx, conn, scope, profile.OrganizationID, coredata.WebhookEventTypeUserUpdated, webhookPayload, + previousUser, ); err != nil { return fmt.Errorf("cannot insert webhook event: %w", err) } diff --git a/pkg/iam/scim/service.go b/pkg/iam/scim/service.go index 997b9e646..3efb11458 100644 --- a/pkg/iam/scim/service.go +++ b/pkg/iam/scim/service.go @@ -190,6 +190,8 @@ func (s *Service) CreateUser( eventType := coredata.WebhookEventTypeUserUpdated profile = &coredata.MembershipProfile{} + var previousProfile *coredata.MembershipProfile + if err := profile.LoadByIdentityIDAndOrganizationID( ctx, tx, @@ -214,6 +216,9 @@ func (s *Service) CreateUser( *externalIdPtr, config.OrganizationID, ); err == nil { + snapshot := *profile + previousProfile = &snapshot + // Migrate the existing membership to the new identity // so the user's role is preserved. oldIdentityID := profile.IdentityID @@ -270,6 +275,11 @@ func (s *Service) CreateUser( eventType = coredata.WebhookEventTypeUserCreated } } else { + if previousProfile == nil { + snapshot := *profile + previousProfile = &snapshot + } + if profile.Source == coredata.ProfileSourceSCIM { return scimerrors.ScimErrorUniqueness } @@ -340,7 +350,12 @@ func (s *Service) CreateUser( } } - if err := webhook.InsertData(ctx, tx, scope, config.OrganizationID, eventType, webhooktypes.NewUser(profile, membership)); err != nil { + var previousUser any + if eventType == coredata.WebhookEventTypeUserUpdated && previousProfile != nil { + previousUser = webhooktypes.NewUser(previousProfile, membership) + } + + if err := webhook.InsertUpdateData(ctx, tx, scope, config.OrganizationID, eventType, webhooktypes.NewUser(profile, membership), previousUser); err != nil { return fmt.Errorf("cannot insert webhook event: %w", err) } @@ -517,6 +532,10 @@ func (s *Service) updateUser( return fmt.Errorf("cannot load membership: %w", err) } + previousProfile := *profile + previousMembership := *membership + previousUser := webhooktypes.NewUser(&previousProfile, &previousMembership) + shouldReactivate := attrs.Active != nil && *attrs.Active && profile.State == coredata.ProfileStateInactive shouldDeactivate := attrs.Active != nil && !*attrs.Active && profile.State == coredata.ProfileStateActive @@ -785,7 +804,7 @@ func (s *Service) updateUser( } } - if err := webhook.InsertData(ctx, tx, scope, config.OrganizationID, coredata.WebhookEventTypeUserUpdated, webhooktypes.NewUser(profile, membership)); err != nil { + if err := webhook.InsertUpdateData(ctx, tx, scope, config.OrganizationID, coredata.WebhookEventTypeUserUpdated, webhooktypes.NewUser(profile, membership), previousUser); err != nil { return fmt.Errorf("cannot insert webhook event: %w", err) } @@ -931,6 +950,8 @@ func (s *Service) deactivateProfileInTx( return nil } + previousUser := webhooktypes.NewUser(profile, membership) + now := time.Now() profile.State = coredata.ProfileStateInactive profile.UpdatedAt = now @@ -964,7 +985,7 @@ func (s *Service) deactivateProfileInTx( } } - if err := webhook.InsertData(ctx, tx, scope, config.OrganizationID, coredata.WebhookEventTypeUserUpdated, webhooktypes.NewUser(profile, membership)); err != nil { + if err := webhook.InsertUpdateData(ctx, tx, scope, config.OrganizationID, coredata.WebhookEventTypeUserUpdated, webhooktypes.NewUser(profile, membership), previousUser); err != nil { return fmt.Errorf("cannot insert webhook event: %w", err) } diff --git a/pkg/probo/document_approval_service.go b/pkg/probo/document_approval_service.go index c5fc076be..9e4177386 100644 --- a/pkg/probo/document_approval_service.go +++ b/pkg/probo/document_approval_service.go @@ -228,6 +228,7 @@ func (s *DocumentApprovalService) BulkPublishVersions( dv, nil, &requestedQuorum.ID, + nil, ); err != nil { return fmt.Errorf("cannot emit approval quorum requested webhook: %w", err) } @@ -241,6 +242,7 @@ func (s *DocumentApprovalService) BulkPublishVersions( dv, nil, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit version published webhook: %w", err) } @@ -398,6 +400,33 @@ func (s *DocumentApprovalService) Approve( return fmt.Errorf("cannot create electronic signature: %w", err) } + // Snapshot the quorum before mutating it, but only when a + // subscriber can actually receive the resulting + // `...:updated` event. Otherwise the emitter short-circuits + // on the same subscription check and this extra load would + // be wasted work that could still fail the approval. + var updatedFromQuorum any + + subscriptions := coredata.WebhookSubscriptions{} + + hasSubscription, err := subscriptions.ExistsByOrganizationIDAndEventType( + ctx, + tx, + scope, + document.OrganizationID, + coredata.WebhookEventTypeDocumentVersionApprovalQuorumUpdated, + ) + if err != nil { + return fmt.Errorf("cannot check webhook subscriptions for approval quorum: %w", err) + } + + if hasSubscription { + updatedFromQuorum, err = s.svc.Documents.loadDocumentApprovalQuorumForWebhook(ctx, scope, tx, quorum.ID, documentVersion, document) + if err != nil { + return fmt.Errorf("cannot load approval quorum snapshot for webhook: %w", err) + } + } + decision.State = coredata.DocumentVersionApprovalDecisionStateApproved decision.Comment = req.Comment decision.ElectronicSignatureID = &esig.ID @@ -433,6 +462,7 @@ func (s *DocumentApprovalService) Approve( documentVersion, nil, &quorum.ID, + updatedFromQuorum, ); err != nil { return fmt.Errorf("cannot emit approval quorum updated webhook: %w", err) } @@ -531,6 +561,7 @@ func (s *DocumentApprovalService) Reject( documentVersion, nil, &quorum.ID, + nil, ); err != nil { return fmt.Errorf("cannot emit approval quorum rejected webhook: %w", err) } @@ -544,6 +575,7 @@ func (s *DocumentApprovalService) Reject( documentVersion, nil, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit document version rejected webhook: %w", err) } @@ -635,6 +667,7 @@ func (s *DocumentApprovalService) VoidApproval( documentVersion, nil, &quorum.ID, + nil, ); err != nil { return fmt.Errorf("cannot emit approval quorum voided webhook: %w", err) } @@ -998,6 +1031,7 @@ func (s *DocumentApprovalService) maybeApproveQuorum( version, nil, &quorum.ID, + nil, ); err != nil { return fmt.Errorf("cannot emit approval quorum approved webhook: %w", err) } @@ -1011,6 +1045,7 @@ func (s *DocumentApprovalService) maybeApproveQuorum( version, nil, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit document version published webhook: %w", err) } diff --git a/pkg/probo/document_service.go b/pkg/probo/document_service.go index c92bc6306..c701ba669 100644 --- a/pkg/probo/document_service.go +++ b/pkg/probo/document_service.go @@ -628,6 +628,7 @@ func (s *DocumentService) PublishVersion( version, nil, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit document version published webhook: %w", err) } @@ -653,6 +654,7 @@ func (s *DocumentService) PublishVersion( version, nil, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit document version published webhook: %w", err) } @@ -708,6 +710,7 @@ func (s *DocumentService) PublishVersion( dv, nil, &quorum.ID, + nil, ); err != nil { return fmt.Errorf("cannot emit approval quorum requested webhook: %w", err) } @@ -805,7 +808,7 @@ func (s *DocumentService) Create( } } - if err := s.emitDocumentEventInTx(ctx, scope, conn, documentID, coredata.WebhookEventTypeDocumentCreated, nil, nil, nil); err != nil { + if err := s.emitDocumentEventInTx(ctx, scope, conn, documentID, coredata.WebhookEventTypeDocumentCreated, nil, nil, nil, nil); err != nil { return fmt.Errorf("cannot emit document created webhook: %w", err) } @@ -818,6 +821,7 @@ func (s *DocumentService) Create( documentVersion, nil, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit document version created webhook: %w", err) } @@ -993,6 +997,7 @@ func (s *DocumentService) SignDocumentVersionByIdentity( documentVersion, documentVersionSignature, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit document version signature signed webhook: %w", err) } @@ -1122,6 +1127,7 @@ func (s *DocumentService) BulkRequestSignatures( documentVersion, signature, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit signature requested webhook: %w", err) } @@ -1258,6 +1264,7 @@ func (s *DocumentService) RequestSignature( documentVersion, signature, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit document version signature requested webhook: %w", err) } @@ -1378,13 +1385,14 @@ func (s *DocumentService) emitDocumentEventInTx( version *coredata.DocumentVersion, signature *coredata.DocumentVersionSignature, quorumID *gid.GID, + updatedFrom any, ) error { document := &coredata.Document{} if err := document.LoadByID(ctx, tx, scope, documentID); err != nil { return fmt.Errorf("cannot load document for %q webhook: %w", eventType, err) } - return s.emitLoadedDocumentEventInTx(ctx, scope, tx, document, eventType, version, signature, quorumID) + return s.emitLoadedDocumentEventInTx(ctx, scope, tx, document, eventType, version, signature, quorumID, updatedFrom) } func (s *DocumentService) emitLoadedDocumentEventInTx( @@ -1395,6 +1403,7 @@ func (s *DocumentService) emitLoadedDocumentEventInTx( version *coredata.DocumentVersion, signature *coredata.DocumentVersionSignature, quorumID *gid.GID, + updatedFrom any, ) error { subscriptions := coredata.WebhookSubscriptions{} @@ -1423,13 +1432,14 @@ func (s *DocumentService) emitLoadedDocumentEventInTx( payload = webhooktypes.NewDocument(document) } - if err := webhook.InsertData( + if err := webhook.InsertUpdateData( ctx, tx, scope, document.OrganizationID, eventType, payload, + updatedFrom, ); err != nil { return fmt.Errorf("cannot insert %q webhook event: %w", eventType, err) } @@ -1533,7 +1543,7 @@ func (s *DocumentService) SoftDelete( return s.svc.pg.WithTx( ctx, func(ctx context.Context, tx pg.Tx) error { - if err := s.emitDocumentEventInTx(ctx, scope, tx, documentID, coredata.WebhookEventTypeDocumentDeleted, nil, nil, nil); err != nil { + if err := s.emitDocumentEventInTx(ctx, scope, tx, documentID, coredata.WebhookEventTypeDocumentDeleted, nil, nil, nil, nil); err != nil { return fmt.Errorf("cannot emit document deleted webhook: %w", err) } @@ -2134,6 +2144,8 @@ func (s *DocumentService) Update( return &ErrDocumentArchived{} } + previousDocument := *document + if req.TrustCenterVisibility != nil { document.TrustCenterVisibility = *req.TrustCenterVisibility } @@ -2172,8 +2184,13 @@ func (s *DocumentService) Update( versionDeleted := false + var versionPrevious any + if hasVersionChanges { if latestVersion.Status == coredata.DocumentVersionStatusDraft { + previousVersion := *latestVersion + versionPrevious = webhooktypes.NewDocumentVersion(&previousVersion, &previousDocument) + if err := s.updateVersionInTx(ctx, scope, tx, latestVersion, req.Content, req.Classification, req.DocumentType, req.Title); err != nil { return err } @@ -2231,6 +2248,7 @@ func (s *DocumentService) Update( latestVersion, nil, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit document version deleted webhook: %w", err) } @@ -2238,16 +2256,17 @@ func (s *DocumentService) Update( versionEvent := coredata.WebhookEventTypeDocumentVersionUpdated if draftCreated { versionEvent = coredata.WebhookEventTypeDocumentVersionCreated + versionPrevious = nil } - if err := s.emitDocumentEventInTx(ctx, scope, tx, resultVersion.DocumentID, versionEvent, resultVersion, nil, nil); err != nil { + if err := s.emitDocumentEventInTx(ctx, scope, tx, resultVersion.DocumentID, versionEvent, resultVersion, nil, nil, versionPrevious); err != nil { return fmt.Errorf("cannot emit document version webhook: %w", err) } } } if docLevelChanged { - if err := s.emitDocumentEventInTx(ctx, scope, tx, req.DocumentID, coredata.WebhookEventTypeDocumentUpdated, nil, nil, nil); err != nil { + if err := s.emitDocumentEventInTx(ctx, scope, tx, req.DocumentID, coredata.WebhookEventTypeDocumentUpdated, nil, nil, nil, webhooktypes.NewDocument(&previousDocument)); err != nil { return fmt.Errorf("cannot emit document updated webhook: %w", err) } } @@ -2305,6 +2324,7 @@ func (s *DocumentService) DeleteDraft( latestVersion, nil, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit document version deleted webhook: %w", err) } @@ -2365,7 +2385,7 @@ func (s *DocumentService) Archive( return fmt.Errorf("cannot archive document: %w", err) } - if err := s.emitDocumentEventInTx(ctx, scope, tx, documentID, coredata.WebhookEventTypeDocumentArchived, nil, nil, nil); err != nil { + if err := s.emitDocumentEventInTx(ctx, scope, tx, documentID, coredata.WebhookEventTypeDocumentArchived, nil, nil, nil, nil); err != nil { return fmt.Errorf("cannot emit document archived webhook: %w", err) } @@ -2405,7 +2425,7 @@ func (s *DocumentService) Unarchive( return fmt.Errorf("cannot unarchive document: %w", err) } - if err := s.emitDocumentEventInTx(ctx, scope, tx, documentID, coredata.WebhookEventTypeDocumentUnarchived, nil, nil, nil); err != nil { + if err := s.emitDocumentEventInTx(ctx, scope, tx, documentID, coredata.WebhookEventTypeDocumentUnarchived, nil, nil, nil, nil); err != nil { return fmt.Errorf("cannot emit document unarchived webhook: %w", err) } @@ -2466,6 +2486,7 @@ func (s *DocumentService) CancelSignatureRequest( documentVersion, documentVersionSignature, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit document version signature cancelled webhook: %w", err) } diff --git a/pkg/probo/generated_document_service.go b/pkg/probo/generated_document_service.go index c61397636..229ba2263 100644 --- a/pkg/probo/generated_document_service.go +++ b/pkg/probo/generated_document_service.go @@ -3296,7 +3296,7 @@ func (s *GeneratedDocumentService) publishOrRequestApproval( } if isFirstVersion { - if err := s.svc.Documents.emitDocumentEventInTx(ctx, scope, tx, document.ID, coredata.WebhookEventTypeDocumentCreated, nil, nil, nil); err != nil { + if err := s.svc.Documents.emitDocumentEventInTx(ctx, scope, tx, document.ID, coredata.WebhookEventTypeDocumentCreated, nil, nil, nil, nil); err != nil { return fmt.Errorf("cannot emit document created webhook: %w", err) } } @@ -3310,6 +3310,7 @@ func (s *GeneratedDocumentService) publishOrRequestApproval( version, nil, &quorum.ID, + nil, ); err != nil { return fmt.Errorf("cannot emit approval quorum requested webhook: %w", err) } @@ -3332,7 +3333,7 @@ func (s *GeneratedDocumentService) publishOrRequestApproval( } if isFirstVersion { - if err := s.svc.Documents.emitDocumentEventInTx(ctx, scope, tx, document.ID, coredata.WebhookEventTypeDocumentCreated, nil, nil, nil); err != nil { + if err := s.svc.Documents.emitDocumentEventInTx(ctx, scope, tx, document.ID, coredata.WebhookEventTypeDocumentCreated, nil, nil, nil, nil); err != nil { return fmt.Errorf("cannot emit document created webhook: %w", err) } } @@ -3346,6 +3347,7 @@ func (s *GeneratedDocumentService) publishOrRequestApproval( version, nil, nil, + nil, ); err != nil { return fmt.Errorf("cannot emit document version published webhook: %w", err) } diff --git a/pkg/probo/obligation_service.go b/pkg/probo/obligation_service.go index 5db42b878..5f58e9e66 100644 --- a/pkg/probo/obligation_service.go +++ b/pkg/probo/obligation_service.go @@ -198,6 +198,8 @@ func (s *ObligationService) Update( return fmt.Errorf("cannot load obligation: %w", err) } + previousObligation := webhooktypes.NewObligation(obligation) + if req.Area != nil { obligation.Area = *req.Area } @@ -249,7 +251,7 @@ func (s *ObligationService) Update( return fmt.Errorf("cannot update obligation: %w", err) } - if err := webhook.InsertData(ctx, conn, scope, obligation.OrganizationID, coredata.WebhookEventTypeObligationUpdated, webhooktypes.NewObligation(obligation)); err != nil { + if err := webhook.InsertUpdateData(ctx, conn, scope, obligation.OrganizationID, coredata.WebhookEventTypeObligationUpdated, webhooktypes.NewObligation(obligation), previousObligation); err != nil { return fmt.Errorf("cannot insert webhook event: %w", err) } diff --git a/pkg/probo/third_party_service.go b/pkg/probo/third_party_service.go index 545b2c5c1..04df7623e 100644 --- a/pkg/probo/third_party_service.go +++ b/pkg/probo/third_party_service.go @@ -368,6 +368,8 @@ func (s ThirdPartyService) Update( return fmt.Errorf("cannot load thirdParty %q: %w", req.ID, err) } + previousThirdParty := webhooktypes.NewThirdParty(thirdParty) + if req.Name != nil { thirdParty.Name = *req.Name } @@ -478,13 +480,14 @@ func (s ThirdPartyService) Update( return fmt.Errorf("cannot update thirdParty: %w", err) } - if err := webhook.InsertData( + if err := webhook.InsertUpdateData( ctx, conn, scope, thirdParty.OrganizationID, coredata.WebhookEventTypeThirdPartyUpdated, webhooktypes.NewThirdParty(thirdParty), + previousThirdParty, ); err != nil { return fmt.Errorf("cannot insert webhook event: %w", err) } diff --git a/pkg/webhook/data.go b/pkg/webhook/data.go index 0b6bd7c23..a13e3bc7c 100644 --- a/pkg/webhook/data.go +++ b/pkg/webhook/data.go @@ -38,6 +38,7 @@ type Payload struct { EventType string `json:"eventType"` CreatedAt time.Time `json:"createdAt"` Data json.RawMessage `json:"data"` + UpdatedFrom json.RawMessage `json:"updatedFrom,omitempty"` } func InsertData( @@ -47,6 +48,22 @@ func InsertData( organizationID gid.GID, eventType coredata.WebhookEventType, data any, +) error { + if err := InsertUpdateData(ctx, tx, scope, organizationID, eventType, data, nil); err != nil { + return fmt.Errorf("cannot insert webhook data: %w", err) + } + + return nil +} + +func InsertUpdateData( + ctx context.Context, + tx pg.Tx, + scope coredata.Scoper, + organizationID gid.GID, + eventType coredata.WebhookEventType, + data any, + updatedFrom any, ) error { var configs coredata.WebhookSubscriptions @@ -64,11 +81,20 @@ func InsertData( return fmt.Errorf("cannot marshal webhook event data: %w", err) } + var updatedFromRaw json.RawMessage + if updatedFrom != nil { + updatedFromRaw, err = json.Marshal(updatedFrom) + if err != nil { + return fmt.Errorf("cannot marshal webhook event updated-from data: %w", err) + } + } + webhookData := &coredata.WebhookData{ ID: gid.New(scope.GetTenantID(), coredata.WebhookDataEntityType), OrganizationID: organizationID, EventType: eventType, Data: raw, + UpdatedFrom: updatedFromRaw, CreatedAt: time.Now(), } diff --git a/pkg/webhook/data_test.go b/pkg/webhook/data_test.go new file mode 100644 index 000000000..6610964d0 --- /dev/null +++ b/pkg/webhook/data_test.go @@ -0,0 +1,241 @@ +// Copyright (c) 2026 Probo Inc . +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +package webhook_test + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.gearno.de/kit/pg" + "go.probo.inc/probo/internal/test" + "go.probo.inc/probo/pkg/coredata" + "go.probo.inc/probo/pkg/gid" + "go.probo.inc/probo/pkg/webhook" +) + +func TestInsertUpdateData_PersistsUpdatedFromSnapshot(t *testing.T) { + client := test.PGClient(t) + orgID := insertTestOrganization(t, client) + scope := coredata.NewScope(orgID.TenantID()) + + insertTestSubscription(t, client, orgID, coredata.WebhookEventTypeUserUpdated) + + current := map[string]any{"role": "OWNER"} + updatedFrom := map[string]any{"role": "ADMIN"} + + err := client.WithTx(context.Background(), func(ctx context.Context, tx pg.Tx) error { + return webhook.InsertUpdateData( + ctx, + tx, + scope, + orgID, + coredata.WebhookEventTypeUserUpdated, + current, + updatedFrom, + ) + }) + require.NoError(t, err) + + data, updatedFromData := loadWebhookData(t, client, orgID) + + assert.JSONEq(t, `{"role":"OWNER"}`, string(data)) + require.NotNil(t, updatedFromData, "updated_from must be persisted for update events") + assert.JSONEq(t, `{"role":"ADMIN"}`, string(updatedFromData)) +} + +func TestInsertData_StoresNullUpdatedFrom(t *testing.T) { + client := test.PGClient(t) + orgID := insertTestOrganization(t, client) + scope := coredata.NewScope(orgID.TenantID()) + + insertTestSubscription(t, client, orgID, coredata.WebhookEventTypeObligationUpdated) + + err := client.WithTx(context.Background(), func(ctx context.Context, tx pg.Tx) error { + return webhook.InsertData( + ctx, + tx, + scope, + orgID, + coredata.WebhookEventTypeObligationUpdated, + map[string]any{"status": "OPEN"}, + ) + }) + require.NoError(t, err) + + data, updatedFromData := loadWebhookData(t, client, orgID) + + assert.JSONEq(t, `{"status":"OPEN"}`, string(data)) + assert.Nil(t, updatedFromData, "updated_from must be SQL NULL when no previous snapshot is provided") +} + +func TestInsertUpdateData_NoSubscriptionIsNoop(t *testing.T) { + client := test.PGClient(t) + orgID := insertTestOrganization(t, client) + scope := coredata.NewScope(orgID.TenantID()) + + // Subscribe to a different event than the one we emit. + insertTestSubscription(t, client, orgID, coredata.WebhookEventTypeObligationUpdated) + + err := client.WithTx(context.Background(), func(ctx context.Context, tx pg.Tx) error { + return webhook.InsertUpdateData( + ctx, + tx, + scope, + orgID, + coredata.WebhookEventTypeUserUpdated, + map[string]any{"role": "OWNER"}, + map[string]any{"role": "ADMIN"}, + ) + }) + require.NoError(t, err) + + assert.Equal(t, 0, countWebhookData(t, client, orgID), "no webhook_data row should be enqueued without a matching subscription") +} + +func TestPayload_UpdatedFromOmittedWhenAbsent(t *testing.T) { + withUpdatedFrom, err := json.Marshal(webhook.Payload{ + EventType: "user:updated", + Data: json.RawMessage(`{"role":"OWNER"}`), + UpdatedFrom: json.RawMessage(`{"role":"ADMIN"}`), + }) + require.NoError(t, err) + assert.Contains(t, string(withUpdatedFrom), `"updatedFrom":{"role":"ADMIN"}`) + + withoutUpdatedFrom, err := json.Marshal(webhook.Payload{ + EventType: "user:created", + Data: json.RawMessage(`{"role":"OWNER"}`), + }) + require.NoError(t, err) + assert.NotContains(t, string(withoutUpdatedFrom), "updatedFrom") +} + +func insertTestOrganization(t *testing.T, client *pg.Client) gid.GID { + t.Helper() + + tenantID := gid.NewTenantID() + orgID := gid.New(tenantID, coredata.OrganizationEntityType) + now := time.Now() + + err := client.WithConn( + context.Background(), + func(ctx context.Context, conn pg.Querier) error { + _, err := conn.Exec( + ctx, + `INSERT INTO organizations (id, tenant_id, name, created_at, updated_at) VALUES ($1, $2, $3, $4, $5)`, + orgID.String(), + tenantID.String(), + "test-org-"+orgID.String(), + now, + now, + ) + + return err + }, + ) + require.NoError(t, err) + + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + _ = client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error { + _, err := conn.Exec(ctx, "DELETE FROM organizations WHERE id = $1", orgID.String()) + return err + }) + }) + + return orgID +} + +func insertTestSubscription( + t *testing.T, + client *pg.Client, + orgID gid.GID, + events ...coredata.WebhookEventType, +) { + t.Helper() + + now := time.Now() + subscription := coredata.WebhookSubscription{ + ID: gid.New(orgID.TenantID(), coredata.WebhookSubscriptionEntityType), + OrganizationID: orgID, + EndpointURL: "https://example.test/webhook", + SelectedEvents: coredata.WebhookEventTypes(events), + EncryptedSigningSecret: []byte("test-signing-secret"), + CreatedAt: now, + UpdatedAt: now, + } + + err := client.WithTx( + context.Background(), + func(ctx context.Context, tx pg.Tx) error { + return subscription.Insert(ctx, tx, coredata.NewScope(orgID.TenantID())) + }, + ) + require.NoError(t, err) +} + +func loadWebhookData(t *testing.T, client *pg.Client, orgID gid.GID) (json.RawMessage, json.RawMessage) { + t.Helper() + + var ( + data []byte + updatedFrom []byte + ) + + err := client.WithConn( + context.Background(), + func(ctx context.Context, conn pg.Querier) error { + return conn.QueryRow( + ctx, + "SELECT data, updated_from FROM webhook_data WHERE organization_id = $1", + orgID.String(), + ).Scan(&data, &updatedFrom) + }, + ) + require.NoError(t, err) + + return data, updatedFrom +} + +func countWebhookData(t *testing.T, client *pg.Client, orgID gid.GID) int { + t.Helper() + + var count int + + err := client.WithConn( + context.Background(), + func(ctx context.Context, conn pg.Querier) error { + return conn.QueryRow( + ctx, + "SELECT COUNT(*) FROM webhook_data WHERE organization_id = $1", + orgID.String(), + ).Scan(&count) + }, + ) + require.NoError(t, err) + + return count +} diff --git a/pkg/webhook/webhook_worker.go b/pkg/webhook/webhook_worker.go index 4def1998c..6f88d9db9 100644 --- a/pkg/webhook/webhook_worker.go +++ b/pkg/webhook/webhook_worker.go @@ -321,6 +321,7 @@ func (h *webhookHandler) doHTTPCall( EventType: webhookData.EventType.String(), CreatedAt: webhookData.CreatedAt, Data: webhookData.Data, + UpdatedFrom: webhookData.UpdatedFrom, } body, err := json.Marshal(payload)