member-console/internal/integrations/stripe/workflows/webhook_ordering_test.go

// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo

package workflows_test

import (
	"context"
	"database/sql"
	"fmt"
	"io"
	"log/slog"
	"math/rand"
	"testing"
	"time"

	internalstripe "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/store"
	stripewf "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/workflows"
)

// insertTimedEvent records a product event with Stripe's own event time; a
// nil at leaves provider_event_at NULL, as rows from before the column do.
func insertTimedEvent(t *testing.T, db *sql.DB, eventType, objectID string, at *time.Time) stripewf.WebhookEvent {
	t.Helper()
	evt := stripewf.WebhookEvent{Provider: "stripe", ProviderEventID: fmt.Sprintf("evt_order_%d", rand.Int63()), EventType: eventType}
	var ts sql.NullTime
	if at != nil {
		ts = sql.NullTime{Time: *at, Valid: true}
	}
	if err := db.QueryRowContext(context.Background(),
		`INSERT INTO core.webhook_events (provider, provider_event_id, event_type, payload, status, provider_event_at)
		 VALUES ('stripe', $1, $2, $3, 'received', $4) RETURNING id`,
		evt.ProviderEventID, eventType, fmt.Sprintf(`{"id": %q}`, objectID), ts).Scan(&evt.ID); err != nil {
		t.Fatalf("insert event: %v", err)
	}
	t.Cleanup(func() {
		_, _ = db.ExecContext(context.Background(), `DELETE FROM core.webhook_events WHERE id = $1`, evt.ID)
	})
	return evt
}

func eventStatus(t *testing.T, db *sql.DB, id int64) string {
	t.Helper()
	var status string
	if err := db.QueryRowContext(context.Background(), `SELECT status FROM core.webhook_events WHERE id = $1`, id).Scan(&status); err != nil {
		t.Fatalf("read event %d: %v", id, err)
	}
	return status
}

// Stripe does not guarantee delivery order. A product.created that arrives
// after a later product.deleted must not put the mapping back to synced:
// the handler compares the events' own timestamps and skips the stale one.
// A genuinely newer product.created still applies, and an event recorded
// without a timestamp (from before the column existed) is never treated as
// stale.
func TestStaleProductEventIsSkipped(t *testing.T) {
	database := testDB(t)
	ctx := context.Background()
	acts := stripewf.NewWebhookActivities(database, slog.New(slog.NewTextHandler(io.Discard, nil)))

	tx, err := database.BeginTx(ctx, nil)
	if err != nil {
		t.Fatal(err)
	}
	productID, _ := createTestProduct(t, ctx, tx)
	if err := tx.Commit(); err != nil {
		t.Fatal(err)
	}
	stripeProductID := fmt.Sprintf("prod_order_%d", rand.Int63())
	q := internalstripe.New(database)
	if _, err := q.UpsertProductMapping(ctx, internalstripe.UpsertProductMappingParams{
		ProductID: productID, StripeProductID: sql.NullString{String: stripeProductID, Valid: true}, SyncStatus: "synced",
	}); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		_, _ = database.ExecContext(ctx, `DELETE FROM stripe.product_mappings WHERE product_id = $1`, productID)
	})
	mappingStatus := func() string {
		m, err := q.GetProductMappingByProductID(ctx, productID)
		if err != nil {
			t.Fatalf("read mapping: %v", err)
		}
		return m.SyncStatus
	}

	base := time.Now().Add(-time.Hour)
	at := func(d time.Duration) *time.Time { ts := base.Add(d); return &ts }

	deleted := insertTimedEvent(t, database, "product.deleted", stripeProductID, at(10*time.Second))
	if err := acts.ProcessWebhookEvent(ctx, deleted); err != nil {
		t.Fatalf("product.deleted: %v", err)
	}
	if got := mappingStatus(); got != "deleted" {
		t.Fatalf("after product.deleted the mapping must be deleted, got %q", got)
	}

	stale := insertTimedEvent(t, database, "product.created", stripeProductID, at(0))
	if err := acts.ProcessWebhookEvent(ctx, stale); err != nil {
		t.Fatalf("stale product.created: %v", err)
	}
	if got := eventStatus(t, database, stale.ID); got != "skipped" {
		t.Errorf("the stale event must be skipped, got %q", got)
	}
	if got := mappingStatus(); got != "deleted" {
		t.Errorf("the stale created must not resurrect the mapping, got %q", got)
	}

	newer := insertTimedEvent(t, database, "product.created", stripeProductID, at(20*time.Second))
	if err := acts.ProcessWebhookEvent(ctx, newer); err != nil {
		t.Fatalf("newer product.created: %v", err)
	}
	if got := mappingStatus(); got != "synced" {
		t.Errorf("a newer created must apply, got %q", got)
	}

	// Back to deleted by a later event, then an untimed created: applied,
	// because nothing can say it is older.
	later := insertTimedEvent(t, database, "product.deleted", stripeProductID, at(30*time.Second))
	if err := acts.ProcessWebhookEvent(ctx, later); err != nil {
		t.Fatalf("later product.deleted: %v", err)
	}
	untimed := insertTimedEvent(t, database, "product.created", stripeProductID, nil)
	if err := acts.ProcessWebhookEvent(ctx, untimed); err != nil {
		t.Fatalf("untimed product.created: %v", err)
	}
	if got := eventStatus(t, database, untimed.ID); got != "completed" {
		t.Errorf("an event without a provider timestamp must be applied, got %q", got)
	}
	if got := mappingStatus(); got != "synced" {
		t.Errorf("the untimed created must apply, got %q", got)
	}
}