189 lines
7.0 KiB
Go
189 lines
7.0 KiB
Go
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
|
|
// SPDX-FileCopyrightText: 2025-2026 Christian Galo
|
|
|
|
package workflows
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"io"
|
|
"log/slog"
|
|
"testing"
|
|
|
|
"github.com/google/uuid"
|
|
"go.temporal.io/sdk/testsuite"
|
|
)
|
|
|
|
func insertEventRow(t *testing.T, db *sql.DB, eventType, payload string) WebhookEvent {
|
|
t.Helper()
|
|
evt := WebhookEvent{Provider: "stripe", ProviderEventID: "evt_act_" + uuid.New().String()[:12], EventType: eventType}
|
|
if err := db.QueryRowContext(context.Background(),
|
|
`INSERT INTO core.webhook_events (provider, provider_event_id, event_type, payload, status)
|
|
VALUES ($1, $2, $3, $4, 'received') RETURNING id`,
|
|
evt.Provider, evt.ProviderEventID, evt.EventType, payload).Scan(&evt.ID); err != nil {
|
|
t.Fatalf("insert event row: %v", err)
|
|
}
|
|
t.Cleanup(func() {
|
|
_, _ = db.ExecContext(context.Background(), `DELETE FROM core.webhook_events WHERE id = $1`, evt.ID)
|
|
})
|
|
return evt
|
|
}
|
|
|
|
func eventRow(t *testing.T, db *sql.DB, id int64) (status string, retryCount int, errMsg string) {
|
|
t.Helper()
|
|
var msg sql.NullString
|
|
if err := db.QueryRowContext(context.Background(),
|
|
`SELECT status, retry_count, error_message FROM core.webhook_events WHERE id = $1`, id).Scan(&status, &retryCount, &msg); err != nil {
|
|
t.Fatalf("read row %d: %v", id, err)
|
|
}
|
|
return status, retryCount, msg.String
|
|
}
|
|
|
|
// A failed attempt is written to the row before the error goes back to
|
|
// Temporal: the status, the attempt number Temporal reports, and the error.
|
|
// The operator page and the Temporal UI then tell the same story.
|
|
func TestProcessWebhookEventRecordsAFailedAttemptOnTheRow(t *testing.T) {
|
|
db := testDB(t)
|
|
acts := NewWebhookActivities(db, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
// An invoice event without an id fails in the handler, retryably.
|
|
evt := insertEventRow(t, db, "invoice.paid", `{}`)
|
|
|
|
var suite testsuite.WorkflowTestSuite
|
|
env := suite.NewTestActivityEnvironment()
|
|
env.RegisterActivity(acts)
|
|
_, err := env.ExecuteActivity(acts.ProcessWebhookEvent, evt)
|
|
if err == nil {
|
|
t.Fatal("processing an invoice event without an id must fail")
|
|
}
|
|
|
|
status, attempts, msg := eventRow(t, db, evt.ID)
|
|
if status != "failed" || attempts != 1 || msg == "" {
|
|
t.Errorf("row after the failed attempt: status %q, retry_count %d, error %q; want failed/1/<the error>", status, attempts, msg)
|
|
}
|
|
}
|
|
|
|
func TestMarkEventDeadLetter(t *testing.T) {
|
|
db := testDB(t)
|
|
acts := NewWebhookActivities(db, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
evt := insertEventRow(t, db, "invoice.paid", `{}`)
|
|
|
|
if err := acts.MarkEventDeadLetter(context.Background(), evt.ID, "schedule-to-close timeout"); err != nil {
|
|
t.Fatalf("mark dead letter: %v", err)
|
|
}
|
|
status, _, msg := eventRow(t, db, evt.ID)
|
|
if status != "dead_letter" || msg != "schedule-to-close timeout" {
|
|
t.Errorf("row: status %q, error %q; want dead_letter with the error", status, msg)
|
|
}
|
|
}
|
|
|
|
// The boot sweep hands every unfinished Stripe row to its workflow and
|
|
// leaves finished rows and other providers' rows alone. The starter is
|
|
// exercised through a fake client-free path: SweepUnfinishedWebhookEvents
|
|
// takes the Temporal client, so this test covers only the selection by
|
|
// reading what it would start, through the same query.
|
|
func TestSweepSelectsOnlyUnfinishedStripeRows(t *testing.T) {
|
|
db := testDB(t)
|
|
want := map[int64]bool{}
|
|
for _, status := range []string{"received", "processing", "failed"} {
|
|
evt := insertEventRow(t, db, "customer.updated", `{}`)
|
|
if _, err := db.ExecContext(context.Background(), `UPDATE core.webhook_events SET status = $1 WHERE id = $2`, status, evt.ID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
want[evt.ID] = true
|
|
}
|
|
skip := map[int64]bool{}
|
|
for _, status := range []string{"completed", "skipped", "dead_letter"} {
|
|
evt := insertEventRow(t, db, "customer.updated", `{}`)
|
|
if _, err := db.ExecContext(context.Background(), `UPDATE core.webhook_events SET status = $1 WHERE id = $2`, status, evt.ID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
skip[evt.ID] = true
|
|
}
|
|
other := insertEventRow(t, db, "user_created", `{}`)
|
|
if _, err := db.ExecContext(context.Background(), `UPDATE core.webhook_events SET provider = 'discourse' WHERE id = $1`, other.ID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
skip[other.ID] = true
|
|
|
|
rows, err := db.QueryContext(context.Background(), unfinishedStripeEventsSQL)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer rows.Close()
|
|
selected := map[int64]bool{}
|
|
for rows.Next() {
|
|
var e WebhookEvent
|
|
if err := rows.Scan(&e.ID, &e.Provider, &e.ProviderEventID, &e.EventType); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
selected[e.ID] = true
|
|
}
|
|
for id := range want {
|
|
if !selected[id] {
|
|
t.Errorf("unfinished row %d must be swept", id)
|
|
}
|
|
}
|
|
for id := range skip {
|
|
if selected[id] {
|
|
t.Errorf("row %d must not be swept", id)
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestProductCreatedStampsTheMapping pins the created branch of design D1.
|
|
// product.created reaches a mapping the outbox wrote before the stamp
|
|
// existed, so the branch must write the environment the parsed product
|
|
// reports, not leave the row unverified; replaying the same event keeps the
|
|
// flag, because the upsert's COALESCE takes the offered flag over the
|
|
// stored one only when a caller has one to offer.
|
|
func TestProductCreatedStampsTheMapping(t *testing.T) {
|
|
db := testDB(t)
|
|
ctx := context.Background()
|
|
acts := NewWebhookActivities(db, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
|
|
var productID string
|
|
if err := db.QueryRowContext(ctx,
|
|
`INSERT INTO core.products (name, display_category, is_active)
|
|
VALUES ($1, NULL, TRUE) RETURNING product_id`,
|
|
"envstamp-product-"+uuid.New().String()[:8],
|
|
).Scan(&productID); err != nil {
|
|
t.Fatalf("create product: %v", err)
|
|
}
|
|
stripeProductID := "prod_envstamp_" + uuid.New().String()[:12]
|
|
if _, err := db.ExecContext(ctx,
|
|
`INSERT INTO stripe.product_mappings (product_id, stripe_product_id, sync_status, livemode)
|
|
VALUES ($1, $2, 'synced', NULL)`,
|
|
productID, stripeProductID,
|
|
); err != nil {
|
|
t.Fatalf("seed unstamped product mapping: %v", err)
|
|
}
|
|
|
|
payload := `{"id": "` + stripeProductID + `", "livemode": true}`
|
|
evt := insertEventRow(t, db, "product.created", payload)
|
|
if _, err := acts.handleProductEvent(ctx, evt); err != nil {
|
|
t.Fatalf("handleProductEvent: %v", err)
|
|
}
|
|
if flag := productMappingLivemode(t, ctx, db, productID); !flag.Valid || !flag.Bool {
|
|
t.Fatalf("after product.created, livemode = %+v, want true", flag)
|
|
}
|
|
|
|
replay := insertEventRow(t, db, "product.created", payload)
|
|
if _, err := acts.handleProductEvent(ctx, replay); err != nil {
|
|
t.Fatalf("handleProductEvent replay: %v", err)
|
|
}
|
|
if flag := productMappingLivemode(t, ctx, db, productID); !flag.Valid || !flag.Bool {
|
|
t.Fatalf("after replay, livemode = %+v, want true", flag)
|
|
}
|
|
}
|
|
|
|
func productMappingLivemode(t *testing.T, ctx context.Context, db *sql.DB, productID string) sql.NullBool {
|
|
t.Helper()
|
|
var flag sql.NullBool
|
|
if err := db.QueryRowContext(ctx,
|
|
`SELECT livemode FROM stripe.product_mappings WHERE product_id = $1`, productID,
|
|
).Scan(&flag); err != nil {
|
|
t.Fatalf("read product mapping livemode: %v", err)
|
|
}
|
|
return flag
|
|
}
|