Files
Paul NicolasandGitHub 04517d36e0 fix(connectors): fix webhook idempotency key and tests (#445)
* feat(workflows): handle webhook idempotency in a different way

* feat(adyen): add verify webhook and fix tests

* fix(atlar): ensure plugin functions are tested

* fix(bankingcircle): ensure plugin functions are tested

* feat(column): add verify webhook method and fix tests

* fix(currencycloud): ensure plugin functions are tested

* fix(generic): ensure plugin functions are tested

* feat(mangopay): add verify webhook method and fix tests

* fix(modulr): ensure plugin functions are tested

* fix(moneycorp): ensure plugin functions are tested

* fix(stripe): ensure plugin functions are tested

* feat(wise): add verify webhook method and fix tests

* fix dirty

* fix postgres idempotency key

* fix wrong go libs version

* fix coderabbitai

* webhooks failing verification should stop temporal workflow
2025-05-14 15:27:14 +00:00

84 lines
2.0 KiB
Go

package storage
import (
"context"
"github.com/formancehq/payments/internal/models"
"github.com/uptrace/bun"
)
type webhook struct {
bun.BaseModel `bun:"table:webhooks"`
// Mandatory fields
ID string `bun:"id,pk,type:uuid,notnull"`
ConnectorID models.ConnectorID `bun:"connector_id,type:character varying,notnull"`
// Optional fields
IdempotencyKey *string `bun:"idempotency_key,type:text"`
Headers map[string][]string `bun:"headers,type:json"`
QueryValues map[string][]string `bun:"query_values,type:json"`
Body []byte `bun:"body,type:bytea,nullzero"`
}
func (s *store) WebhooksInsert(ctx context.Context, webhook models.Webhook) error {
toInsert := fromWebhookModels(webhook)
_, err := s.db.NewInsert().
Model(&toInsert).
On("CONFLICT (id) DO NOTHING").
Exec(ctx)
if err != nil {
return e("insert webhook", err)
}
return nil
}
func (s *store) WebhooksGet(ctx context.Context, id string) (models.Webhook, error) {
var w webhook
err := s.db.NewSelect().
Model(&w).
Where("id = ?", id).
Scan(ctx)
if err != nil {
return models.Webhook{}, e("get webhook", err)
}
return toWebhookModels(w), nil
}
func (s *store) WebhooksDeleteFromConnectorID(ctx context.Context, connectorID models.ConnectorID) error {
_, err := s.db.NewDelete().
Model((*webhook)(nil)).
Where("connector_id = ?", connectorID).
Exec(ctx)
if err != nil {
return e("delete webhook", err)
}
return nil
}
func fromWebhookModels(from models.Webhook) webhook {
return webhook{
ID: from.ID,
ConnectorID: from.ConnectorID,
IdempotencyKey: from.IdempotencyKey,
Headers: from.Headers,
QueryValues: from.QueryValues,
Body: from.Body,
}
}
func toWebhookModels(from webhook) models.Webhook {
return models.Webhook{
ID: from.ID,
ConnectorID: from.ConnectorID,
IdempotencyKey: from.IdempotencyKey,
Headers: from.Headers,
QueryValues: from.QueryValues,
Body: from.Body,
}
}