Add updated-from entity snapshot to updated webhooks
Update webhook events now carry a top-level "updatedFrom" field alongside "data", containing a full snapshot of the entity as it was before the update. This lets subscribers diff old vs new state (for example the prior membership role on user:updated) without tracking prior state themselves. It is a complete snapshot with the same shape as "data", not a partial diff, so consumers select whatever fields they need. The field is omitted for non-update events. The webhook_data table gains a nullable updated_from JSONB column, and webhook.InsertUpdateData enqueues both snapshots; InsertData delegates to it with a nil updatedFrom so non-update callers are unaffected. Each *:updated emission site snapshots the entity right after load, before mutation: obligation, third-party, user (org and SCIM flows), document, document-version, and document-version-approval-quorum. The document emit helpers gained an optional updatedFrom argument threaded through to the payload. For document-version-approval-quorum:updated the snapshot requires an extra query, so it is now gated behind the same subscription-existence check the emitter uses: when no subscriber is configured the load is skipped entirely rather than running (and potentially failing the approval) for an event nobody receives. Add integration tests (against a real Postgres, skipped when none is reachable) covering the updated_from round-trip, the SQL NULL behavior when no snapshot is provided, and the no-op when no subscription matches, plus a unit test asserting updatedFrom is omitted from the payload when absent. Document the new field in the probod and n8n changelogs and the n8n README. Signed-off-by: Sacha Al Himdani <sacha@probo.com>
This commit is contained in:
Sacha Al Himdani
@@ -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(),
|
||||
}
|
||||
|
||||
|
||||
241
pkg/webhook/data_test.go
Normal file
241
pkg/webhook/data_test.go
Normal file
@@ -0,0 +1,241 @@
|
||||
// Copyright (c) 2026 Probo Inc <hello@probo.com>.
|
||||
//
|
||||
// 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
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user