package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/glebarez/sqlite"
	"github.com/pingcap/agent-git-service/internal/db"
	"github.com/pingcap/agent-git-service/internal/gitstore"
	"gorm.io/gorm"
)

type failingRepositoryEventSink struct{ err error }

func (sink failingRepositoryEventSink) HandleRepositoryEvent(context.Context, RepositoryEvent) error {
	return sink.err
}

func newDurableWebhookTestService(database *gorm.DB) *Service {
	return &Service{DB: database, durableWebhookDB: database}
}

func markDurableWebhookTestDelivery(delivery *db.HookDelivery, operationID, eventKey string) {
	delivery.OperationID = &operationID
	delivery.OperationEventKey = &eventKey
	delivery.SnapshotVersion = durableWebhookSnapshotVersion
}

func TestOrdinaryWebhookDispatchAndRedeliveryKeepRequestScopedDB(t *testing.T) {
	openDatabase := func(name string) *gorm.DB {
		t.Helper()
		database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), name)), &gorm.Config{})
		if err != nil {
			t.Fatalf("open %s: %v", name, err)
		}
		if err := database.AutoMigrate(&db.Repository{}, &db.Webhook{}, &db.HookDelivery{}); err != nil {
			t.Fatalf("migrate %s: %v", name, err)
		}
		return database
	}
	primary := openDatabase("ordinary-primary.db")
	scoped := openDatabase("ordinary-scoped.db")
	received := make(chan struct{}, 2)
	receiver := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		received <- struct{}{}
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer receiver.Close()
	repository := db.Repository{ID: 41, Name: "repo", FullName: "owner/repo", DefaultBranch: "main"}
	if err := scoped.Create(&repository).Error; err != nil {
		t.Fatalf("create scoped repository: %v", err)
	}
	hook := db.Webhook{RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`, ConfigJSON: fmt.Sprintf(`{"url":%q}`, receiver.URL)}
	if err := scoped.Create(&hook).Error; err != nil {
		t.Fatalf("create scoped hook: %v", err)
	}
	svc := newDurableWebhookTestService(primary)
	serverCtx, cancelServer := context.WithCancel(context.Background())
	svc.Ctx = serverCtx
	defer func() {
		cancelServer()
		svc.Wg.Wait()
	}()
	requestCtx := ContextWithDB(t.Context(), scoped)
	if err := svc.DispatchWebhookEvent(requestCtx, repository.ID, "push", "", map[string]any{"ref": "refs/heads/main"}); err != nil {
		t.Fatalf("dispatch scoped webhook: %v", err)
	}
	select {
	case <-received:
	case <-time.After(time.Second):
		t.Fatal("ordinary worker did not deliver the scoped webhook")
	}
	var first db.HookDelivery
	deadline := time.Now().Add(time.Second)
	for {
		if err := scoped.Where("webhook_id = ?", hook.ID).First(&first).Error; err != nil {
			t.Fatalf("load scoped delivery: %v", err)
		}
		if first.Status == webhookStatusOK {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("scoped delivery status = %q", first.Status)
		}
		time.Sleep(5 * time.Millisecond)
	}
	legacyGUID := strings.Repeat("g", 64)
	if err := scoped.Model(&db.HookDelivery{}).Where("id = ?", first.ID).Update("guid", legacyGUID).Error; err != nil {
		t.Fatalf("set legacy 64-byte GUID: %v", err)
	}
	first.GUID = legacyGUID
	redelivery, err := svc.RedeliverHookDelivery(requestCtx, repository.ID, hook.ID, first.ID)
	if err != nil {
		t.Fatalf("redeliver scoped webhook: %v", err)
	}
	if redelivery.Status != webhookStatusPending {
		t.Fatalf("returned redelivery status = %q, want persisted pre-send value", redelivery.Status)
	}
	if redelivery.GUID != legacyGUID || redelivery.LeaseToken == "" || len(redelivery.LeaseToken) > 64 || redelivery.LeaseToken == legacyGUID {
		t.Fatalf("legacy GUID redelivery identity = guid:%q lease:%q", redelivery.GUID, redelivery.LeaseToken)
	}
	select {
	case <-received:
	case <-time.After(time.Second):
		t.Fatal("ordinary worker did not deliver the scoped redelivery")
	}
	var scopedDeliveries, primaryDeliveries int64
	scoped.Model(&db.HookDelivery{}).Count(&scopedDeliveries)
	primary.Model(&db.HookDelivery{}).Count(&primaryDeliveries)
	if scopedDeliveries != 2 || primaryDeliveries != 0 {
		t.Fatalf("ordinary delivery rows = scoped:%d primary:%d", scopedDeliveries, primaryDeliveries)
	}
	var storedRedelivery db.HookDelivery
	deadline = time.Now().Add(time.Second)
	for {
		err := scoped.First(&storedRedelivery, redelivery.ID).Error
		if err == nil && storedRedelivery.Status == webhookStatusOK {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("stored scoped redelivery = status:%q err:%v", storedRedelivery.Status, err)
		}
		time.Sleep(5 * time.Millisecond)
	}
}

func TestOrdinaryPrimaryDeliveryLeasePreventsDurableScannerDuplicate(t *testing.T) {
	database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "ordinary-lease.db")), &gorm.Config{})
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := database.AutoMigrate(&db.Repository{}, &db.Webhook{}, &db.HookDelivery{}); err != nil {
		t.Fatalf("AutoMigrate: %v", err)
	}
	entered := make(chan struct{})
	release := make(chan struct{}, 1)
	var requests atomic.Int64
	receiver := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		if requests.Add(1) == 1 {
			close(entered)
		}
		<-release
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer receiver.Close()
	repository := db.Repository{ID: 42, Name: "repo", FullName: "owner/repo", DefaultBranch: "main"}
	if err := database.Create(&repository).Error; err != nil {
		t.Fatalf("create repository: %v", err)
	}
	hook := db.Webhook{RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`, ConfigJSON: fmt.Sprintf(`{"url":%q}`, receiver.URL)}
	if err := database.Create(&hook).Error; err != nil {
		t.Fatalf("create hook: %v", err)
	}
	svc := newDurableWebhookTestService(database)
	serverCtx, cancelServer := context.WithCancel(context.Background())
	svc.Ctx = serverCtx
	defer func() {
		cancelServer()
		svc.Wg.Wait()
	}()
	defer func() {
		select {
		case release <- struct{}{}:
		default:
		}
	}()
	if err := svc.DispatchWebhookEvent(t.Context(), repository.ID, "push", "", map[string]any{"ref": "refs/heads/main"}); err != nil {
		t.Fatalf("dispatch webhook: %v", err)
	}
	select {
	case <-entered:
	case <-time.After(time.Second):
		t.Fatal("ordinary delivery did not enter receiver")
	}
	jobs, attempted, err := svc.claimPendingWebhookDeliveries(t.Context(), webhookDeliveryPageSize)
	if err != nil || attempted != 0 || len(jobs) != 0 {
		t.Fatalf("durable scanner claimed direct delivery = jobs:%d attempted:%d err:%v", len(jobs), attempted, err)
	}
	release <- struct{}{}
	deadline := time.Now().Add(time.Second)
	for {
		var delivery db.HookDelivery
		if err := database.First(&delivery, "webhook_id = ?", hook.ID).Error; err != nil {
			t.Fatalf("load delivery: %v", err)
		}
		if delivery.Status == webhookStatusOK {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("delivery status = %q", delivery.Status)
		}
		time.Sleep(5 * time.Millisecond)
	}
	if got := requests.Load(); got != 1 {
		t.Fatalf("ordinary delivery requests = %d, want 1", got)
	}
}

func TestOrdinaryScopedPendingRowsAreNotClaimedAsPrimaryDurableWork(t *testing.T) {
	openDatabase := func(name string) *gorm.DB {
		t.Helper()
		database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), name)), &gorm.Config{})
		if err != nil {
			t.Fatalf("open %s: %v", name, err)
		}
		if err := database.AutoMigrate(&db.HookDelivery{}); err != nil {
			t.Fatalf("migrate %s: %v", name, err)
		}
		return database
	}
	primary := openDatabase("scoped-contract-primary.db")
	scoped := openDatabase("scoped-contract-tenant.db")
	expired := time.Now().UTC().Add(-time.Minute)
	row := db.HookDelivery{WebhookID: 1, RepositoryID: 1, GUID: "ordinary-scoped-audit-row", Event: "push",
		Status: webhookStatusPending, RequestURL: "https://example.invalid/hook", RequestHeaders: "{}",
		RequestPayload: db.LargeText(`{}`), NextAttemptAt: expired, LeaseUntil: &expired, LeaseToken: "old-live-process"}
	if err := scoped.Create(&row).Error; err != nil {
		t.Fatalf("create scoped pending audit row: %v", err)
	}

	rebuiltPrimary := newDurableWebhookTestService(primary)
	attempted, err := rebuiltPrimary.processPendingWebhookDeliveryPage(t.Context())
	if err != nil || attempted != 0 {
		t.Fatalf("primary rebuild claimed scoped work = attempted:%d err:%v", attempted, err)
	}
	var retained db.HookDelivery
	if err := scoped.First(&retained, row.ID).Error; err != nil {
		t.Fatalf("reload scoped audit row: %v", err)
	}
	if retained.Status != webhookStatusPending || retained.LeaseToken != row.LeaseToken || retained.LeaseUntil == nil {
		t.Fatalf("primary scanner mutated scoped audit row: %+v", retained)
	}
}

func TestDurableWebhookAuthorityIgnoresReplaceableServiceAndRequestDB(t *testing.T) {
	openDatabase := func(name string) *gorm.DB {
		t.Helper()
		database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), name)), &gorm.Config{})
		if err != nil {
			t.Fatalf("open %s: %v", name, err)
		}
		if err := database.AutoMigrate(
			&db.Repository{}, &db.RepositoryMigrationFreeze{}, &db.RepositoryPushRecovery{},
			&db.RepositoryPushOutbox{}, &db.RepositoryWebhookFanout{}, &db.Webhook{}, &db.HookDelivery{},
		); err != nil {
			t.Fatalf("migrate %s: %v", name, err)
		}
		return database
	}
	primary := openDatabase("primary.db")
	alternate := openDatabase("alternate.db")
	store, err := gitstore.New(t.TempDir())
	if err != nil {
		t.Fatalf("gitstore.New: %v", err)
	}
	svc, err := New(Config{DB: primary, Git: store})
	if err != nil {
		t.Fatalf("service.New: %v", err)
	}

	repository := db.Repository{ID: 31, Name: "repo", FullName: "owner/repo", DefaultBranch: "main"}
	if err := primary.Create(&repository).Error; err != nil {
		t.Fatalf("create primary repository: %v", err)
	}
	if err := alternate.Create(&repository).Error; err != nil {
		t.Fatalf("create alternate repository: %v", err)
	}
	canonicalRequests := make(chan struct{}, 1)
	canonicalReceiver := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		canonicalRequests <- struct{}{}
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer canonicalReceiver.Close()
	alternateRequests := make(chan struct{}, 1)
	alternateReceiver := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		alternateRequests <- struct{}{}
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer alternateReceiver.Close()
	canonicalHook := db.Webhook{RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`, ConfigJSON: fmt.Sprintf(`{"url":%q}`, canonicalReceiver.URL)}
	if err := primary.Create(&canonicalHook).Error; err != nil {
		t.Fatalf("create primary hook: %v", err)
	}
	alternateHook := db.Webhook{RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`, ConfigJSON: fmt.Sprintf(`{"url":%q}`, alternateReceiver.URL)}
	if err := alternate.Create(&alternateHook).Error; err != nil {
		t.Fatalf("create alternate hook: %v", err)
	}
	const operationID = "primary-durable-webhook"
	const eventKey = "push:refs/heads/main"
	outbox := db.RepositoryPushOutbox{RepositoryID: repository.ID, OperationID: operationID, EventKey: "post_push",
		OccurredAt: time.Now().UTC(), RefUpdatesJSON: db.LargeText("[]"), State: "pending", NextAttemptAt: time.Now().UTC().Add(-time.Minute)}
	if err := primary.Create(&outbox).Error; err != nil {
		t.Fatalf("create primary outbox: %v", err)
	}

	var alternateAuthorityAccesses atomic.Int64
	countAlternate := func(*gorm.DB) { alternateAuthorityAccesses.Add(1) }
	callbacks := []struct {
		name     string
		register func(string, func(*gorm.DB)) error
		remove   func(string) error
	}{
		{"query", alternate.Callback().Query().Before("gorm:query").Register, alternate.Callback().Query().Remove},
		{"row", alternate.Callback().Row().Before("gorm:row").Register, alternate.Callback().Row().Remove},
		{"raw", alternate.Callback().Raw().Before("gorm:raw").Register, alternate.Callback().Raw().Remove},
		{"create", alternate.Callback().Create().Before("gorm:create").Register, alternate.Callback().Create().Remove},
		{"update", alternate.Callback().Update().Before("gorm:update").Register, alternate.Callback().Update().Remove},
		{"delete", alternate.Callback().Delete().Before("gorm:delete").Register, alternate.Callback().Delete().Remove},
	}
	for _, callback := range callbacks {
		if err := callback.register("test:durable_webhook_authority_"+callback.name, countAlternate); err != nil {
			t.Fatalf("register alternate %s callback: %v", callback.name, err)
		}
	}

	svc.DB = alternate
	requestCtx := ContextWithDB(t.Context(), alternate)
	if err := svc.DispatchWebhookEventIdempotent(requestCtx, repository.ID, "push", "", map[string]any{"ref": "refs/heads/main"}, operationID, eventKey); err != nil {
		t.Fatalf("dispatch primary durable webhook: %v", err)
	}
	select {
	case <-canonicalRequests:
	case <-time.After(2 * time.Second):
		t.Fatal("primary webhook delivery was not sent")
	}
	svc.Wg.Wait()
	select {
	case <-alternateRequests:
		t.Fatal("durable webhook authority was retargeted to alternate DB")
	default:
	}
	claims, err := svc.ClaimRepositoryPushOutbox(requestCtx, repository.ID, 1)
	if err != nil || len(claims) != 1 {
		t.Fatalf("claim primary outbox = %d, %v", len(claims), err)
	}
	if err := svc.CompleteRepositoryPushOutbox(requestCtx, claims[0]); err != nil {
		t.Fatalf("complete primary outbox: %v", err)
	}
	if got := alternateAuthorityAccesses.Load(); got != 0 {
		t.Fatalf("alternate DB received %d durable webhook authority accesses", got)
	}
	for _, callback := range callbacks {
		if err := callback.remove("test:durable_webhook_authority_" + callback.name); err != nil {
			t.Fatalf("remove alternate %s callback: %v", callback.name, err)
		}
	}

	var primaryDeliveries, canonicalFanouts, canonicalOutboxes int64
	primary.Model(&db.HookDelivery{}).Where("operation_id = ?", operationID).Count(&primaryDeliveries)
	primary.Model(&db.RepositoryWebhookFanout{}).Where("operation_id = ?", operationID).Count(&canonicalFanouts)
	primary.Model(&db.RepositoryPushOutbox{}).Where("operation_id = ?", operationID).Count(&canonicalOutboxes)
	if primaryDeliveries != 1 || canonicalFanouts != 0 || canonicalOutboxes != 0 {
		t.Fatalf("primary durable rows = deliveries:%d fanouts:%d outboxes:%d", primaryDeliveries, canonicalFanouts, canonicalOutboxes)
	}
	var alternateDeliveries, alternateFanouts int64
	alternate.Model(&db.HookDelivery{}).Count(&alternateDeliveries)
	alternate.Model(&db.RepositoryWebhookFanout{}).Count(&alternateFanouts)
	if alternateDeliveries != 0 || alternateFanouts != 0 {
		t.Fatalf("alternate durable rows = deliveries:%d fanouts:%d", alternateDeliveries, alternateFanouts)
	}
}

func TestDispatchWebhookEventIdempotentMaterializesOneDelivery(t *testing.T) {
	database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-idempotent.db")), &gorm.Config{})
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := database.AutoMigrate(&db.Repository{}, &db.Webhook{}, &db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
		t.Fatalf("AutoMigrate: %v", err)
	}
	if !database.Migrator().HasIndex(&db.HookDelivery{}, "idx_hook_delivery_claim") {
		t.Fatal("durable hook delivery claim index is missing")
	}
	if !database.Migrator().HasIndex(&db.HookDelivery{}, "idx_hook_delivery_durable_claim") {
		t.Fatal("durable hook delivery provenance claim index is missing")
	}
	if !database.Migrator().HasIndex(&db.RepositoryWebhookFanout{}, "idx_repository_webhook_fanout_operation_event") {
		t.Fatal("durable webhook fanout identity index is missing")
	}
	repository := db.Repository{ID: 1, Name: "repo", FullName: "owner/repo", DefaultBranch: "main"}
	if err := database.Create(&repository).Error; err != nil {
		t.Fatalf("create repository: %v", err)
	}
	hook := db.Webhook{RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`, ConfigJSON: `{}`}
	if err := database.Create(&hook).Error; err != nil {
		t.Fatalf("create webhook: %v", err)
	}
	service := newDurableWebhookTestService(database)
	for index := 0; index < 2; index++ {
		if err := service.DispatchWebhookEventIdempotent(t.Context(), repository.ID, "push", "", map[string]any{"ref": "refs/heads/main"}, "operation-1", "push:refs/heads/main"); err != nil {
			t.Fatalf("DispatchWebhookEventIdempotent(%d): %v", index, err)
		}
	}
	var deliveries []db.HookDelivery
	if err := database.Find(&deliveries).Error; err != nil {
		t.Fatalf("load deliveries: %v", err)
	}
	if len(deliveries) != 1 || deliveries[0].OperationID == nil || *deliveries[0].OperationID != "operation-1" ||
		deliveries[0].SnapshotVersion != durableWebhookSnapshotVersion {
		t.Fatalf("deliveries = %+v", deliveries)
	}
}

func TestDispatchWebhookOperationRejectsRefsTimesSubscribersBeforeAnyFanoutSideEffect(t *testing.T) {
	database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-operation-limit.db")), &gorm.Config{})
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := database.AutoMigrate(&db.Webhook{}, &db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
		t.Fatalf("AutoMigrate: %v", err)
	}
	var requests atomic.Int64
	receiver := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		requests.Add(1)
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer receiver.Close()
	hooks := make([]db.Webhook, webhookFanoutTargetMax)
	for index := range hooks {
		hooks[index] = db.Webhook{
			RepositoryID: 91, Active: true, EventsJSON: `["push"]`, ConfigJSON: fmt.Sprintf(`{"url":%q}`, receiver.URL),
		}
	}
	if err := database.CreateInBatches(&hooks, 128).Error; err != nil {
		t.Fatalf("create hooks: %v", err)
	}
	var queries, creates atomic.Int64
	if err := database.Callback().Query().Before("gorm:query").Register("test:count_operation_limit_queries", func(*gorm.DB) { queries.Add(1) }); err != nil {
		t.Fatalf("register query counter: %v", err)
	}
	if err := database.Callback().Create().Before("gorm:create").Register("test:count_operation_limit_creates", func(*gorm.DB) { creates.Add(1) }); err != nil {
		t.Fatalf("register create counter: %v", err)
	}
	events := make([]RepositoryEvent, webhookFanoutTargetMax)
	for index := range events {
		events[index] = RepositoryEvent{
			RepositoryID: 91, Event: "push", EventKey: fmt.Sprintf("push:refs/heads/extreme-%04d", index), PayloadJSON: []byte(`{"ref":"refs/heads/extreme"}`),
		}
	}
	err = newDurableWebhookTestService(database).DispatchWebhookOperationIdempotent(t.Context(), "operation-extreme-product", events)
	if err == nil || !strings.Contains(err.Error(), "delivery row limit") {
		t.Fatalf("operation product error = %v, want delivery row limit", err)
	}
	if got := queries.Load(); got != 2 {
		t.Fatalf("operation preflight queries = %d, want fixed fanout+hook queries", got)
	}
	if got := creates.Load(); got != 0 {
		t.Fatalf("operation preflight create callbacks = %d, want zero", got)
	}
	var deliveries, fanouts int64
	database.Model(&db.HookDelivery{}).Count(&deliveries)
	database.Model(&db.RepositoryWebhookFanout{}).Count(&fanouts)
	if deliveries != 0 || fanouts != 0 || requests.Load() != 0 {
		t.Fatalf("rejected operation side effects = deliveries:%d fanouts:%d HTTP:%d", deliveries, fanouts, requests.Load())
	}
}

func TestDispatchWebhookOperationRejectsFinalEncodedSnapshotBytesBeforeAnyRow(t *testing.T) {
	database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-operation-bytes.db")), &gorm.Config{})
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := database.AutoMigrate(&db.Webhook{}, &db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
		t.Fatalf("AutoMigrate: %v", err)
	}
	hooks := make([]db.Webhook, 3)
	for index := range hooks {
		hooks[index] = db.Webhook{RepositoryID: 92, Active: true, EventsJSON: `["push"]`, ConfigJSON: `{"url":"https://example.invalid/hook"}`}
	}
	if err := database.Create(&hooks).Error; err != nil {
		t.Fatalf("create hooks: %v", err)
	}
	var queries, creates atomic.Int64
	if err := database.Callback().Query().Before("gorm:query").Register("test:count_operation_bytes_queries", func(*gorm.DB) { queries.Add(1) }); err != nil {
		t.Fatalf("register query counter: %v", err)
	}
	if err := database.Callback().Create().Before("gorm:create").Register("test:count_operation_bytes_creates", func(*gorm.DB) { creates.Add(1) }); err != nil {
		t.Fatalf("register create counter: %v", err)
	}
	payload := []byte(`{"data":"` + strings.Repeat("x", 2800<<10) + `"}`)
	events := make([]RepositoryEvent, 5)
	for index := range events {
		events[index] = RepositoryEvent{RepositoryID: 92, Event: "push", EventKey: fmt.Sprintf("push:bytes-%d", index), PayloadJSON: payload}
	}
	err = newDurableWebhookTestService(database).DispatchWebhookOperationIdempotent(t.Context(), "operation-final-snapshot-bytes", events)
	if err == nil || !strings.Contains(err.Error(), "final snapshot limit") {
		t.Fatalf("operation snapshot error = %v, want final snapshot limit", err)
	}
	if got := queries.Load(); got != 2 {
		t.Fatalf("operation byte preflight queries = %d, want fixed fanout+hook queries", got)
	}
	if got := creates.Load(); got != 0 {
		t.Fatalf("operation byte preflight create callbacks = %d, want zero", got)
	}
	var deliveries, fanouts int64
	database.Model(&db.HookDelivery{}).Count(&deliveries)
	database.Model(&db.RepositoryWebhookFanout{}).Count(&fanouts)
	if deliveries != 0 || fanouts != 0 {
		t.Fatalf("rejected snapshot rows = deliveries:%d fanouts:%d", deliveries, fanouts)
	}
}

func TestDispatchWebhookEventIdempotentReturnsDurableMaterializationErrors(t *testing.T) {
	database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-errors.db")), &gorm.Config{})
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := database.AutoMigrate(&db.Repository{}, &db.Webhook{}, &db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
		t.Fatalf("AutoMigrate: %v", err)
	}
	repository := db.Repository{ID: 2, Name: "repo", FullName: "owner/repo", DefaultBranch: "main"}
	if err := database.Create(&repository).Error; err != nil {
		t.Fatalf("create repository: %v", err)
	}
	hook := db.Webhook{RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`, ConfigJSON: `{"url":"http://127.0.0.1:1"}`}
	if err := database.Create(&hook).Error; err != nil {
		t.Fatalf("create webhook: %v", err)
	}
	materializationErr := errors.New("forced hook materialization failure")
	if err := database.Callback().Create().Before("gorm:create").Register("test:fail_hook_delivery", func(tx *gorm.DB) {
		if tx.Statement != nil && tx.Statement.Table == "hook_deliveries" {
			tx.AddError(materializationErr)
		}
	}); err != nil {
		t.Fatalf("register create callback: %v", err)
	}
	sinkErr := errors.New("forced repository event failure")
	service := newDurableWebhookTestService(database)
	service.RepositoryEvents = failingRepositoryEventSink{err: sinkErr}
	err = service.DispatchWebhookEventIdempotent(t.Context(), repository.ID, "push", "", map[string]any{"ref": "refs/heads/main"}, "operation-errors", "push:refs/heads/main")
	if !errors.Is(err, sinkErr) || !errors.Is(err, materializationErr) {
		t.Fatalf("durable materialization error = %v, want sink and hook failures", err)
	}
}

func TestDispatchWebhookEventIdempotentSkipsDeletedFrozenTargetAfterPartialFailure(t *testing.T) {
	database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-fanout.db")+"?_pragma=foreign_keys(1)"), &gorm.Config{})
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := database.AutoMigrate(&db.Repository{}, &db.Webhook{}, &db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
		t.Fatalf("AutoMigrate: %v", err)
	}
	originalRequests := make(chan string, 2)
	originalA := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		originalRequests <- "a"
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer originalA.Close()
	originalB := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		originalRequests <- "b"
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer originalB.Close()
	replacementRequests := make(chan string, 2)
	replacement := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		replacementRequests <- "replacement"
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer replacement.Close()

	owner := db.User{ID: 21, Login: "owner"}
	if err := database.Create(&owner).Error; err != nil {
		t.Fatalf("create owner: %v", err)
	}
	repository := db.Repository{ID: 21, OwnerID: owner.ID, Name: "repo", FullName: "owner/repo", DefaultBranch: "main"}
	if err := database.Create(&repository).Error; err != nil {
		t.Fatalf("create repository: %v", err)
	}
	const secretA = "fanout-secret-a-must-not-persist"
	const secretB = "fanout-secret-b-must-not-persist"
	hookA := db.Webhook{RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`, ConfigJSON: fmt.Sprintf(`{"url":%q,"secret":%q}`, originalA.URL, secretA)}
	hookB := db.Webhook{RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`, ConfigJSON: fmt.Sprintf(`{"url":%q,"secret":%q}`, originalB.URL, secretB)}
	if err := database.Create(&hookA).Error; err != nil {
		t.Fatalf("create hook A: %v", err)
	}
	if err := database.Create(&hookB).Error; err != nil {
		t.Fatalf("create hook B: %v", err)
	}
	materializationErr := errors.New("forced second-phase materialization failure")
	failMaterialization := true
	if err := database.Callback().Create().Before("gorm:create").Register("test:fail_fanout_materialization", func(tx *gorm.DB) {
		if failMaterialization && tx.Statement != nil && tx.Statement.Table == "hook_deliveries" {
			failMaterialization = false
			tx.AddError(materializationErr)
		}
	}); err != nil {
		t.Fatalf("register materialization failure: %v", err)
	}
	service := newDurableWebhookTestService(database)
	const operationID = "operation-frozen-fanout"
	const eventKey = "push:refs/heads/main"
	err = service.DispatchWebhookEventIdempotent(t.Context(), repository.ID, "push", "", map[string]any{"ref": "refs/heads/main"}, operationID, eventKey)
	if !errors.Is(err, materializationErr) {
		t.Fatalf("initial materialization error = %v", err)
	}
	var deliveryCount int64
	if err := database.Model(&db.HookDelivery{}).Count(&deliveryCount).Error; err != nil || deliveryCount != 0 {
		t.Fatalf("partial delivery rows = %d, %v", deliveryCount, err)
	}
	var frozen db.RepositoryWebhookFanout
	if err := database.Where("operation_id = ? AND event_key = ?", operationID, eventKey).First(&frozen).Error; err != nil {
		t.Fatalf("load frozen fanout: %v", err)
	}
	if frozen.State != webhookFanoutPending {
		t.Fatalf("frozen fanout state = %q", frozen.State)
	}
	if strings.Contains(string(frozen.TargetsJSON), secretA) || strings.Contains(string(frozen.TargetsJSON), secretB) {
		t.Fatalf("fanout manifest persisted a raw webhook secret: %s", frozen.TargetsJSON)
	}
	frozenTargets, err := decodeDurableWebhookTargets(frozen, repository.ID, "push", "")
	if err != nil {
		t.Fatalf("decode frozen targets: %v", err)
	}
	frozenByHook := make(map[uint]durableWebhookTarget, len(frozenTargets))
	for _, target := range frozenTargets {
		frozenByHook[target.WebhookID] = target
	}
	if err := database.Callback().Create().Remove("test:fail_fanout_materialization"); err != nil {
		t.Fatalf("remove materialization failure: %v", err)
	}
	hookA.ConfigJSON = fmt.Sprintf(`{"url":%q}`, replacement.URL)
	if err := database.Save(&hookA).Error; err != nil {
		t.Fatalf("retarget hook A: %v", err)
	}
	if err := database.Delete(&hookB).Error; err != nil {
		t.Fatalf("delete hook B: %v", err)
	}
	hookC := db.Webhook{RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`, ConfigJSON: fmt.Sprintf(`{"url":%q}`, replacement.URL)}
	if err := database.Create(&hookC).Error; err != nil {
		t.Fatalf("create hook C: %v", err)
	}
	liveTargetQueries := 0
	if err := database.Callback().Query().Before("gorm:query").Register("test:count_live_frozen_targets", func(tx *gorm.DB) {
		if tx.Statement != nil && tx.Statement.Table == "webhooks" {
			liveTargetQueries++
		}
	}); err != nil {
		t.Fatalf("register live target query counter: %v", err)
	}

	rebuilt := newDurableWebhookTestService(database)
	if err := rebuilt.DispatchWebhookEventIdempotent(t.Context(), repository.ID, "push", "", map[string]any{"ref": "changed-on-replay"}, operationID, eventKey); err != nil {
		t.Fatalf("replay frozen fanout: %v", err)
	}
	select {
	case name := <-originalRequests:
		if name != "a" {
			t.Fatalf("deleted frozen target delivered to %q", name)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("surviving frozen target was not delivered")
	}
	rebuilt.Wg.Wait()
	if err := database.Callback().Query().Remove("test:count_live_frozen_targets"); err != nil {
		t.Fatalf("remove live target query counter: %v", err)
	}
	if liveTargetQueries != 1 {
		t.Fatalf("live frozen target queries = %d, want one bounded batch", liveTargetQueries)
	}
	select {
	case name := <-originalRequests:
		t.Fatalf("unexpected additional frozen target delivered to %q", name)
	default:
	}
	select {
	case <-replacementRequests:
		t.Fatal("replay reinterpreted the changed hook set/config")
	default:
	}
	var deliveries []db.HookDelivery
	if err := database.Order("webhook_id ASC").Find(&deliveries).Error; err != nil {
		t.Fatalf("load materialized deliveries: %v", err)
	}
	if len(deliveries) != 1 || deliveries[0].WebhookID != hookA.ID || deliveries[0].RequestURL != originalA.URL ||
		!strings.Contains(string(deliveries[0].RequestPayload), "refs/heads/main") || strings.Contains(string(deliveries[0].RequestPayload), "changed-on-replay") {
		t.Fatalf("materialized frozen deliveries = %+v", deliveries)
	}
	for _, delivery := range deliveries {
		target := frozenByHook[delivery.WebhookID]
		if delivery.GUID != target.GUID || delivery.RequestURL != target.RequestURL || delivery.RequestInsecure != target.RequestInsecure ||
			delivery.RequestHeaders != target.RequestHeaders || string(delivery.RequestPayload) != target.RequestBody {
			t.Fatalf("delivery %d changed frozen request bytes: delivery=%+v target=%+v", delivery.ID, delivery, target)
		}
	}
	if err := database.First(&frozen, frozen.ID).Error; err != nil || frozen.State != webhookFanoutMaterialized {
		t.Fatalf("materialized fanout = state:%q err:%v", frozen.State, err)
	}
}

func TestDispatchWebhookEventIdempotentRejectsExpandedFormBody(t *testing.T) {
	database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-form-limit.db")), &gorm.Config{})
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := database.AutoMigrate(&db.Repository{}, &db.Webhook{}, &db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
		t.Fatalf("AutoMigrate: %v", err)
	}
	repository := db.Repository{ID: 4, Name: "repo", FullName: "owner/repo", DefaultBranch: "main"}
	if err := database.Create(&repository).Error; err != nil {
		t.Fatalf("create repository: %v", err)
	}
	hook := db.Webhook{RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`, ConfigJSON: `{"url":"https://example.invalid/hook","content_type":"form"}`}
	if err := database.Create(&hook).Error; err != nil {
		t.Fatalf("create webhook: %v", err)
	}
	service := newDurableWebhookTestService(database)
	err = service.DispatchWebhookEventIdempotent(t.Context(), repository.ID, "push", "", map[string]any{"data": strings.Repeat("é", 2<<20)}, "operation-form-limit", "push:refs/heads/main")
	if err == nil || !strings.Contains(err.Error(), "request body exceeds limit") {
		t.Fatalf("expanded form body error = %v, want request-body limit", err)
	}
	var deliveries int64
	if err := database.Model(&db.HookDelivery{}).Count(&deliveries).Error; err != nil {
		t.Fatalf("count deliveries: %v", err)
	}
	if deliveries != 0 {
		t.Fatalf("expanded form body persisted %d deliveries", deliveries)
	}
}

func TestDurableWebhookFanoutNormalizesColumnLimitsBeforeCrashReplay(t *testing.T) {
	t.Run("overlong-url", func(t *testing.T) {
		database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-url-limit.db")), &gorm.Config{})
		if err != nil {
			t.Fatalf("open database: %v", err)
		}
		if err := database.AutoMigrate(&db.Repository{}, &db.Webhook{}, &db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
			t.Fatalf("AutoMigrate: %v", err)
		}
		repository := db.Repository{ID: 51, Name: "repo", FullName: "owner/repo", DefaultBranch: "main"}
		if err := database.Create(&repository).Error; err != nil {
			t.Fatalf("create repository: %v", err)
		}
		hook := db.Webhook{RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`,
			ConfigJSON: fmt.Sprintf(`{"url":%q}`, "https://example.invalid/"+strings.Repeat("a", webhookRequestURLMax))}
		if err := database.Create(&hook).Error; err != nil {
			t.Fatalf("create webhook: %v", err)
		}
		materializationErr := errors.New("simulated crash after manifest commit")
		if err := database.Callback().Create().Before("gorm:create").Register("test:fail_after_bounded_manifest", func(tx *gorm.DB) {
			if tx.Statement != nil && tx.Statement.Table == "hook_deliveries" {
				tx.AddError(materializationErr)
			}
		}); err != nil {
			t.Fatalf("register materialization failure: %v", err)
		}
		const operationID = "operation-overlong-url"
		const eventKey = "push:refs/heads/main"
		svc := newDurableWebhookTestService(database)
		err = svc.DispatchWebhookEventIdempotent(t.Context(), repository.ID, "push", "", map[string]any{"ref": "refs/heads/main"}, operationID, eventKey)
		if !errors.Is(err, materializationErr) {
			t.Fatalf("initial materialization error = %v", err)
		}
		var fanout db.RepositoryWebhookFanout
		if err := database.Where("operation_id = ? AND event_key = ?", operationID, eventKey).First(&fanout).Error; err != nil {
			t.Fatalf("load bounded fanout: %v", err)
		}
		targets, err := decodeDurableWebhookTargets(fanout, repository.ID, "push", "")
		if err != nil || len(targets) != 1 {
			t.Fatalf("decode bounded fanout = %d, %v", len(targets), err)
		}
		if targets[0].TerminalError == "" || len(targets[0].TerminalError) > webhookTerminalErrorMax || targets[0].RequestURL != "" {
			t.Fatalf("overlong URL target was not terminally normalized: %+v", targets[0])
		}
		if err := database.Callback().Create().Remove("test:fail_after_bounded_manifest"); err != nil {
			t.Fatalf("remove materialization failure: %v", err)
		}
		rebuilt := newDurableWebhookTestService(database)
		if err := rebuilt.DispatchWebhookEventIdempotent(t.Context(), repository.ID, "push", "", map[string]any{"changed": true}, operationID, eventKey); err != nil {
			t.Fatalf("replay bounded fanout: %v", err)
		}
		rebuilt.Wg.Wait()
		var delivery db.HookDelivery
		if err := database.Where("operation_id = ?", operationID).First(&delivery).Error; err != nil {
			t.Fatalf("load terminal delivery: %v", err)
		}
		if delivery.Status != webhookStatusError || len(delivery.LastError) > webhookTerminalErrorMax || delivery.RequestURL != "" {
			t.Fatalf("terminal delivery = %+v", delivery)
		}
	})

	t.Run("legacy-overlong-terminal-error", func(t *testing.T) {
		database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-error-limit.db")), &gorm.Config{})
		if err != nil {
			t.Fatalf("open database: %v", err)
		}
		if err := database.AutoMigrate(&db.Repository{}, &db.Webhook{}, &db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
			t.Fatalf("AutoMigrate: %v", err)
		}
		const operationID = "operation-overlong-error"
		const eventKey = "push:refs/heads/main"
		repository := db.Repository{ID: 52, Name: "repo", FullName: "owner/repo", DefaultBranch: "main"}
		if err := database.Create(&repository).Error; err != nil {
			t.Fatalf("create repository: %v", err)
		}
		hook := db.Webhook{ID: 1, RepositoryID: repository.ID, Active: true, EventsJSON: `["push"]`}
		if err := database.Create(&hook).Error; err != nil {
			t.Fatalf("create webhook: %v", err)
		}
		encoded, err := json.Marshal([]durableWebhookTarget{{WebhookID: 1, GUID: "stable-guid", TerminalError: strings.Repeat("x", webhookTerminalErrorMax+100)}})
		if err != nil {
			t.Fatalf("encode legacy target: %v", err)
		}
		fanout := db.RepositoryWebhookFanout{RepositoryID: repository.ID, OperationID: operationID, EventKey: eventKey,
			Event: "push", PayloadJSON: db.LargeText(`{"ref":"refs/heads/main"}`), TargetsJSON: db.LargeText(encoded), State: webhookFanoutPending}
		if err := database.Create(&fanout).Error; err != nil {
			t.Fatalf("create legacy fanout: %v", err)
		}
		rebuilt := newDurableWebhookTestService(database)
		if err := rebuilt.DispatchWebhookEventIdempotent(t.Context(), fanout.RepositoryID, "push", "", map[string]any{"changed": true}, operationID, eventKey); err != nil {
			t.Fatalf("replay legacy overlong error: %v", err)
		}
		rebuilt.Wg.Wait()
		var delivery db.HookDelivery
		if err := database.Where("operation_id = ?", operationID).First(&delivery).Error; err != nil {
			t.Fatalf("load normalized legacy delivery: %v", err)
		}
		if delivery.Status != webhookStatusError || len(delivery.LastError) != webhookTerminalErrorMax || delivery.RequestURL != "" {
			t.Fatalf("normalized legacy delivery = %+v", delivery)
		}
	})
}

func TestPendingHookDeliveryRecoversAfterProcessRebuild(t *testing.T) {
	database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-recovery.db")), &gorm.Config{})
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := database.AutoMigrate(&db.Repository{}, &db.Webhook{}, &db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
		t.Fatalf("AutoMigrate: %v", err)
	}
	type receivedRequest struct {
		guid string
		body string
	}
	received := make(chan receivedRequest, 2)
	receiver := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
		body, _ := io.ReadAll(request.Body)
		received <- receivedRequest{guid: request.Header.Get("X-GitHub-Delivery"), body: string(body)}
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer receiver.Close()

	const stableGUID = "crash-window-guid"
	const stableBody = `{"durable":true}`
	headers := make(http.Header)
	headers.Set("Content-Type", "application/json")
	headers.Set("X-GitHub-Delivery", stableGUID)
	expiredLease := time.Now().UTC().Add(-time.Second)
	delivery := db.HookDelivery{WebhookID: 7, RepositoryID: 7, GUID: stableGUID, Event: "push", Status: webhookStatusPending,
		RequestURL: receiver.URL, RequestHeaders: marshalWebhookHeaders(headers), RequestPayload: db.LargeText(stableBody),
		NextAttemptAt: time.Now().UTC().Add(-time.Minute), LeaseToken: "crashed-process-lease", LeaseUntil: &expiredLease}
	markDurableWebhookTestDelivery(&delivery, "operation-crash-window", "push:refs/heads/main")
	if err := database.Create(&delivery).Error; err != nil {
		t.Fatalf("create crash-window delivery: %v", err)
	}
	serverCtx, cancel := context.WithCancel(context.Background())
	rebuilt := newDurableWebhookTestService(database)
	rebuilt.Ctx = serverCtx
	rebuilt.ensureWebhookWorkers()
	select {
	case request := <-received:
		if request.guid != stableGUID || strings.TrimSpace(request.body) != stableBody {
			t.Fatalf("recovered request = guid:%q body:%q", request.guid, request.body)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("rebuilt service did not recover pending delivery")
	}
	deadline := time.Now().Add(2 * time.Second)
	for {
		var stored db.HookDelivery
		if err := database.First(&stored, delivery.ID).Error; err != nil {
			t.Fatalf("reload recovered delivery: %v", err)
		}
		if stored.Status == webhookStatusOK {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("recovered delivery status = %q", stored.Status)
		}
		time.Sleep(10 * time.Millisecond)
	}
	select {
	case duplicate := <-received:
		t.Fatalf("recovered delivery was duplicated before completion: %+v", duplicate)
	default:
	}
	cancel()
	rebuilt.Wg.Wait()
}

func TestPendingHookDeliveryMalformedRowsDoNotStarveValidPages(t *testing.T) {
	for _, malformedCount := range []int{1, webhookDeliveryPageSize} {
		t.Run(fmt.Sprintf("malformed-%d", malformedCount), func(t *testing.T) {
			database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-page.db")), &gorm.Config{})
			if err != nil {
				t.Fatalf("open database: %v", err)
			}
			if err := database.AutoMigrate(&db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
				t.Fatalf("AutoMigrate: %v", err)
			}
			received := make(chan struct{}, 1)
			receiver := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
				received <- struct{}{}
				writer.WriteHeader(http.StatusNoContent)
			}))
			defer receiver.Close()

			now := time.Now().UTC()
			const operationID = "operation-malformed-page"
			for index := 0; index < malformedCount; index++ {
				row := db.HookDelivery{WebhookID: uint(index + 1), RepositoryID: 1, GUID: fmt.Sprintf("malformed-%d", index),
					Event: "push", Status: webhookStatusPending, RequestURL: "://invalid", RequestHeaders: "{}",
					RequestPayload: db.LargeText(`{}`), NextAttemptAt: now.Add(-2 * time.Minute)}
				markDurableWebhookTestDelivery(&row, operationID, fmt.Sprintf("push:malformed:%d", index))
				if err := database.Create(&row).Error; err != nil {
					t.Fatalf("create malformed delivery %d: %v", index, err)
				}
			}
			headers := make(http.Header)
			headers.Set("Content-Type", "application/json")
			valid := db.HookDelivery{WebhookID: 99, RepositoryID: 1, GUID: "valid-after-malformed", Event: "push",
				Status: webhookStatusPending, RequestURL: receiver.URL, RequestHeaders: marshalWebhookHeaders(headers),
				RequestPayload: db.LargeText(`{"valid":true}`), NextAttemptAt: now.Add(-time.Minute)}
			markDurableWebhookTestDelivery(&valid, operationID, "push:valid")
			if err := database.Create(&valid).Error; err != nil {
				t.Fatalf("create valid delivery: %v", err)
			}

			service := newDurableWebhookTestService(database)
			t.Cleanup(func() {
				service.webhookFallbackMu.Lock()
				defer service.webhookFallbackMu.Unlock()
				if service.webhookFallbackTimer != nil {
					service.webhookFallbackTimer.Stop()
				}
			})
			service.kickWebhookDeliveryWorker()
			select {
			case <-received:
			case <-time.After(2 * time.Second):
				t.Fatal("valid delivery was starved by malformed rows")
			}
			service.Wg.Wait()
			if err := database.First(&valid, valid.ID).Error; err != nil {
				t.Fatalf("reload valid delivery: %v", err)
			}
			if valid.Status != webhookStatusOK {
				t.Fatalf("valid delivery status = %q, want ok", valid.Status)
			}
		})
	}
}

func TestPendingHookDeliveryNilLifecycleWakesAtFutureRetry(t *testing.T) {
	database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-delayed.db")), &gorm.Config{})
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := database.AutoMigrate(&db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
		t.Fatalf("AutoMigrate: %v", err)
	}
	received := make(chan struct{}, 1)
	receiver := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		received <- struct{}{}
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer receiver.Close()
	headers := make(http.Header)
	headers.Set("Content-Type", "application/json")
	delivery := db.HookDelivery{WebhookID: 1, RepositoryID: 1, GUID: "future-retry", Event: "push",
		Status: webhookStatusPending, RequestURL: receiver.URL, RequestHeaders: marshalWebhookHeaders(headers),
		RequestPayload: db.LargeText(`{}`), NextAttemptAt: time.Now().UTC().Add(100 * time.Millisecond)}
	markDurableWebhookTestDelivery(&delivery, "operation-future-retry", "push:future")
	if err := database.Create(&delivery).Error; err != nil {
		t.Fatalf("create future delivery: %v", err)
	}
	service := newDurableWebhookTestService(database)
	service.kickWebhookDeliveryWorker()
	select {
	case <-received:
	case <-time.After(2 * time.Second):
		t.Fatal("nil-lifecycle fallback did not wake for future delivery")
	}
	service.Wg.Wait()
}

func TestPendingHookDeliveryNilLifecycleCoalescesConcurrentKicks(t *testing.T) {
	database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-coalesced.db")), &gorm.Config{})
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := database.AutoMigrate(&db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
		t.Fatalf("AutoMigrate: %v", err)
	}
	var active atomic.Int64
	var maximum atomic.Int64
	var delivered atomic.Int64
	receiver := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
		current := active.Add(1)
		for {
			observed := maximum.Load()
			if current <= observed || maximum.CompareAndSwap(observed, current) {
				break
			}
		}
		time.Sleep(10 * time.Millisecond)
		delivered.Add(1)
		active.Add(-1)
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer receiver.Close()
	headers := make(http.Header)
	headers.Set("Content-Type", "application/json")
	now := time.Now().UTC().Add(-time.Minute)
	const deliveryTotal = webhookDeliveryPageSize * 3
	for index := 0; index < deliveryTotal; index++ {
		delivery := db.HookDelivery{WebhookID: uint(index + 1), RepositoryID: 1, GUID: fmt.Sprintf("coalesced-%d", index), Event: "push",
			Status: webhookStatusPending, RequestURL: receiver.URL, RequestHeaders: marshalWebhookHeaders(headers),
			RequestPayload: db.LargeText(`{}`), NextAttemptAt: now}
		markDurableWebhookTestDelivery(&delivery, "operation-coalesced", fmt.Sprintf("push:coalesced:%d", index))
		if err := database.Create(&delivery).Error; err != nil {
			t.Fatalf("create delivery %d: %v", index, err)
		}
	}
	service := newDurableWebhookTestService(database)
	var kicks sync.WaitGroup
	for index := 0; index < 100; index++ {
		kicks.Add(1)
		go func() {
			defer kicks.Done()
			service.kickWebhookDeliveryWorker()
		}()
	}
	kicks.Wait()
	service.Wg.Wait()
	if delivered.Load() != deliveryTotal {
		t.Fatalf("delivered = %d, want %d", delivered.Load(), deliveryTotal)
	}
	if maximum.Load() != 1 {
		t.Fatalf("nil-lifecycle HTTP concurrency = %d, want single fallback worker", maximum.Load())
	}
}

func TestPendingHookDeliveryUnownedRowsRemainUntouchedByRecovery(t *testing.T) {
	database, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "webhook-legacy.db")), &gorm.Config{})
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := database.AutoMigrate(&db.HookDelivery{}, &db.RepositoryWebhookFanout{}); err != nil {
		t.Fatalf("AutoMigrate: %v", err)
	}
	legacy := db.HookDelivery{WebhookID: 1, RepositoryID: 1, GUID: "legacy-pending", Event: "push", Status: webhookStatusPending}
	if err := database.Create(&legacy).Error; err != nil {
		t.Fatalf("create legacy pending delivery: %v", err)
	}
	operationID := "operation-unversioned"
	eventKey := "push:unversioned"
	unversioned := db.HookDelivery{WebhookID: 2, RepositoryID: 1, GUID: "unversioned-pending", Event: "push", Status: webhookStatusPending,
		OperationID: &operationID, OperationEventKey: &eventKey, RequestURL: "https://example.invalid/hook", RequestHeaders: "{}",
		RequestPayload: db.LargeText(`{}`), NextAttemptAt: time.Now().UTC().Add(-time.Minute)}
	if err := database.Create(&unversioned).Error; err != nil {
		t.Fatalf("create unversioned pending delivery: %v", err)
	}
	incompleteOperationID := "operation-incomplete"
	incompleteEventKey := "push:incomplete"
	incomplete := db.HookDelivery{WebhookID: 3, RepositoryID: 1, GUID: "incomplete-pending", Event: "push", Status: webhookStatusPending,
		OperationID: &incompleteOperationID, OperationEventKey: &incompleteEventKey, SnapshotVersion: durableWebhookSnapshotVersion,
		RequestHeaders: "{}", RequestPayload: db.LargeText(`{}`), NextAttemptAt: time.Now().UTC().Add(-time.Minute)}
	if err := database.Create(&incomplete).Error; err != nil {
		t.Fatalf("create incomplete pending delivery: %v", err)
	}
	var before []db.HookDelivery
	if err := database.Order("id ASC").Find(&before).Error; err != nil {
		t.Fatalf("load unowned pending deliveries: %v", err)
	}
	if len(before) != 3 || before[0].SnapshotVersion != 0 || before[0].OperationID != nil || before[0].OperationEventKey != nil ||
		before[1].SnapshotVersion != 0 || before[1].OperationID == nil || before[1].OperationEventKey == nil ||
		before[2].SnapshotVersion != durableWebhookSnapshotVersion || before[2].RequestURL != "" {
		t.Fatalf("unowned deliveries have unexpected durable provenance: %+v", before)
	}

	queryObserved := make(chan struct{}, 1)
	if err := database.Callback().Query().After("gorm:query").Register("test:observe_legacy_recovery_query", func(tx *gorm.DB) {
		if tx.Statement != nil && tx.Statement.Table == "hook_deliveries" {
			select {
			case queryObserved <- struct{}{}:
			default:
			}
		}
	}); err != nil {
		t.Fatalf("register legacy recovery query observer: %v", err)
	}
	var updateCount atomic.Int64
	if err := database.Callback().Update().Before("gorm:update").Register("test:count_legacy_recovery_writes", func(tx *gorm.DB) {
		if tx.Statement != nil && tx.Statement.Table == "hook_deliveries" {
			updateCount.Add(1)
		}
	}); err != nil {
		t.Fatalf("register legacy recovery update counter: %v", err)
	}

	service := newDurableWebhookTestService(database)
	serverCtx, cancel := context.WithCancel(context.Background())
	service.Ctx = serverCtx
	service.ensureWebhookWorkers()
	select {
	case <-queryObserved:
	case <-time.After(2 * time.Second):
		cancel()
		service.Wg.Wait()
		t.Fatal("startup recovery did not inspect the durable claim scope")
	}
	cancel()
	service.Wg.Wait()
	if err := database.Callback().Query().Remove("test:observe_legacy_recovery_query"); err != nil {
		t.Fatalf("remove legacy recovery query observer: %v", err)
	}
	if err := database.Callback().Update().Remove("test:count_legacy_recovery_writes"); err != nil {
		t.Fatalf("remove legacy recovery update counter: %v", err)
	}
	var after []db.HookDelivery
	if err := database.Order("id ASC").Find(&after).Error; err != nil {
		t.Fatalf("reload untouched unowned deliveries: %v", err)
	}
	if updateCount.Load() != 0 || len(after) != len(before) {
		t.Fatalf("unowned deliveries changed during startup recovery: writes=%d before=%+v after=%+v", updateCount.Load(), before, after)
	}
	for index := range before {
		if after[index].Status != before[index].Status || after[index].DeliveredAt != nil || after[index].LastError != before[index].LastError ||
			after[index].LeaseToken != before[index].LeaseToken || after[index].LeaseUntil != nil || after[index].Attempts != before[index].Attempts ||
			!after[index].UpdatedAt.Equal(before[index].UpdatedAt) {
			t.Fatalf("unowned delivery changed during startup recovery: index=%d before=%+v after=%+v", index, before[index], after[index])
		}
	}
}