package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/url"
	"strings"
	"time"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"github.com/pingcap/agent-git-service/internal/db"
)

const (
	webhookFanoutTargetMax      = 1024
	webhookFanoutStorageMax     = 12 << 20
	webhookFanoutDeliveryMax    = 16 << 20
	webhookOperationEventMax    = 2048
	webhookOperationDeliveryMax = 4096
	webhookOperationManifestMax = 64 << 20
	webhookOperationSnapshotMax = 32 << 20
	webhookFanoutPending        = "pending"
	webhookFanoutMaterialized   = "materialized"
)

type durableWebhookTarget struct {
	WebhookID       uint   `json:"webhook_id"`
	GUID            string `json:"guid"`
	RequestURL      string `json:"request_url,omitempty"`
	RequestInsecure bool   `json:"request_insecure,omitempty"`
	RequestHeaders  string `json:"request_headers"`
	RequestBody     string `json:"request_body"`
	TerminalError   string `json:"terminal_error,omitempty"`
}

func (s *Service) freezeDurableWebhookOperation(ctx context.Context, operationID string, events []RepositoryEvent) ([]RepositoryEvent, error) {
	if s == nil || s.durableWebhookDB == nil {
		return nil, fmt.Errorf("durable webhook database is not initialized")
	}
	if operationID == "" || len(operationID) > 64 || len(events) == 0 || len(events) > webhookOperationEventMax {
		return nil, fmt.Errorf("webhook operation identity or event count is invalid")
	}
	requested := make(map[string]RepositoryEvent, len(events))
	repositoryIDs := make(map[uint]struct{})
	ordered := make([]RepositoryEvent, len(events))
	for index, event := range events {
		if event.RepositoryID == 0 || event.Event == "" || len(event.Event) > 64 || len(event.Action) > 64 || event.EventKey == "" || len(event.EventKey) > 400 ||
			len(event.PayloadJSON) > durableWebhookPayloadMax || event.OperationID != "" && event.OperationID != operationID {
			return nil, fmt.Errorf("webhook operation event is invalid")
		}
		if _, exists := requested[event.EventKey]; exists {
			return nil, fmt.Errorf("webhook operation event is duplicated")
		}
		event.OperationID = operationID
		event.PayloadJSON = append([]byte(nil), event.PayloadJSON...)
		requested[event.EventKey] = event
		ordered[index] = event
		repositoryIDs[event.RepositoryID] = struct{}{}
	}

	frozen := make([]RepositoryEvent, len(events))
	err := s.durableWebhookDB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		fanouts, _, err := freezeDurableWebhookFanoutsTx(tx, operationID, ordered, requested, repositoryIDs)
		if err != nil {
			return err
		}
		for index, event := range ordered {
			fanout := fanouts[event.EventKey]
			frozen[index] = event
			frozen[index].PayloadJSON = append([]byte(nil), fanout.PayloadJSON...)
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return frozen, nil
}

func (s *Service) materializeDurableWebhookOperation(ctx context.Context, operationID string) error {
	return s.durableWebhookDB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		var persisted []db.RepositoryWebhookFanout
		if err := tx.Where("operation_id = ?", operationID).Order("event_key ASC").Limit(webhookOperationEventMax + 1).Find(&persisted).Error; err != nil {
			return err
		}
		if len(persisted) == 0 || len(persisted) > webhookOperationEventMax {
			return fmt.Errorf("durable webhook operation manifest set is invalid")
		}
		fanouts := make(map[string]db.RepositoryWebhookFanout, len(persisted))
		targetsByKey := make(map[string][]durableWebhookTarget, len(persisted))
		for _, fanout := range persisted {
			targets, err := decodeDurableWebhookTargets(fanout, fanout.RepositoryID, fanout.Event, fanout.Action)
			if err != nil {
				return err
			}
			fanouts[fanout.EventKey] = fanout
			targetsByKey[fanout.EventKey] = targets
		}
		if err := validateDurableWebhookOperationBounds(fanouts, targetsByKey); err != nil {
			return err
		}
		return materializeDurableWebhookOperationTx(tx, operationID, fanouts, targetsByKey)
	})
}

func freezeDurableWebhookFanoutsTx(
	tx *gorm.DB,
	operationID string,
	events []RepositoryEvent,
	requested map[string]RepositoryEvent,
	repositoryIDs map[uint]struct{},
) (map[string]db.RepositoryWebhookFanout, map[string][]durableWebhookTarget, error) {
	var persisted []db.RepositoryWebhookFanout
	if err := tx.Where("operation_id = ?", operationID).Order("event_key ASC").Limit(webhookOperationEventMax + 1).Find(&persisted).Error; err != nil {
		return nil, nil, err
	}
	if len(persisted) > webhookOperationEventMax {
		return nil, nil, fmt.Errorf("durable webhook operation exceeds event limit")
	}
	byKey := make(map[string]db.RepositoryWebhookFanout, len(events))
	targetsByKey := make(map[string][]durableWebhookTarget, len(events))
	for _, fanout := range persisted {
		event, ok := requested[fanout.EventKey]
		if !ok {
			return nil, nil, fmt.Errorf("durable webhook operation event set changed")
		}
		targets, err := decodeDurableWebhookTargets(fanout, event.RepositoryID, event.Event, event.Action)
		if err != nil {
			return nil, nil, err
		}
		byKey[event.EventKey] = fanout
		targetsByKey[event.EventKey] = targets
	}

	missing := make([]RepositoryEvent, 0, len(events)-len(byKey))
	for _, event := range events {
		if _, ok := byKey[event.EventKey]; !ok {
			missing = append(missing, event)
		}
	}
	if len(missing) != 0 {
		ids := make([]uint, 0, len(repositoryIDs))
		for id := range repositoryIDs {
			ids = append(ids, id)
		}
		var hooks []db.Webhook
		if err := tx.Where("repository_id IN ? AND active = ?", ids, true).
			Order("repository_id ASC, id ASC").Limit(webhookOperationDeliveryMax + 1).Find(&hooks).Error; err != nil {
			return nil, nil, err
		}
		if len(hooks) > webhookOperationDeliveryMax {
			return nil, nil, fmt.Errorf("durable webhook operation exceeds delivery row limit")
		}
		hooksByRepository := make(map[uint][]db.Webhook, len(repositoryIDs))
		for _, hook := range hooks {
			hooksByRepository[hook.RepositoryID] = append(hooksByRepository[hook.RepositoryID], hook)
		}
		rows := 0
		for _, targets := range targetsByKey {
			rows += len(targets)
		}
		for _, event := range missing {
			matched := 0
			for _, hook := range hooksByRepository[event.RepositoryID] {
				if webhookSubscribed(hook.EventsJSON, event.Event) {
					matched++
				}
			}
			if matched > webhookFanoutTargetMax {
				return nil, nil, fmt.Errorf("durable webhook fanout exceeds target limit")
			}
			rows += matched
			if rows > webhookOperationDeliveryMax {
				return nil, nil, fmt.Errorf("durable webhook operation exceeds delivery row limit")
			}
		}

		created := make([]db.RepositoryWebhookFanout, 0, len(missing))
		for _, event := range missing {
			targets, err := buildDurableWebhookTargets(event, hooksByRepository[event.RepositoryID])
			if err != nil {
				return nil, nil, err
			}
			encoded, err := json.Marshal(targets)
			if err != nil {
				return nil, nil, err
			}
			if len(event.PayloadJSON)+len(encoded) > webhookFanoutStorageMax {
				return nil, nil, fmt.Errorf("durable webhook fanout exceeds storage limit")
			}
			fanout := db.RepositoryWebhookFanout{
				RepositoryID: event.RepositoryID, OperationID: operationID, EventKey: event.EventKey, Event: event.Event, Action: event.Action,
				PayloadJSON: db.LargeText(event.PayloadJSON), TargetsJSON: db.LargeText(encoded), State: webhookFanoutPending,
			}
			if _, err := decodeDurableWebhookTargets(fanout, event.RepositoryID, event.Event, event.Action); err != nil {
				return nil, nil, err
			}
			created = append(created, fanout)
			targetsByKey[event.EventKey] = targets
			byKey[event.EventKey] = fanout
		}
		if err := validateDurableWebhookOperationBounds(byKey, targetsByKey); err != nil {
			return nil, nil, err
		}
		if err := tx.CreateInBatches(&created, 16).Error; err != nil {
			return nil, nil, err
		}
		for _, fanout := range created {
			byKey[fanout.EventKey] = fanout
		}
	}
	if len(byKey) != len(events) {
		return nil, nil, fmt.Errorf("durable webhook operation event set is incomplete")
	}
	if err := validateDurableWebhookOperationBounds(byKey, targetsByKey); err != nil {
		return nil, nil, err
	}
	return byKey, targetsByKey, nil
}

func buildDurableWebhookTargets(event RepositoryEvent, hooks []db.Webhook) ([]durableWebhookTarget, error) {
	targets := make([]durableWebhookTarget, 0, len(hooks))
	for _, hook := range hooks {
		if !webhookSubscribed(hook.EventsJSON, event.Event) {
			continue
		}
		guid, err := newWebhookGUID()
		if err != nil {
			return nil, err
		}
		cfg, cfgErr := parseWebhookConfig(hook.ConfigJSON)
		target := durableWebhookTarget{WebhookID: hook.ID, GUID: guid}
		if cfgErr != nil {
			target.TerminalError = cfgErr.Error()
		} else {
			headers, requestBody := buildWebhookRequest(hook.ID, event.Event, guid, event.PayloadJSON, cfg.Secret, cfg.ContentType)
			if len(requestBody) > webhookRequestBodyMax {
				return nil, fmt.Errorf("webhook request body exceeds limit")
			}
			target.RequestURL = cfg.URL
			target.RequestInsecure = webhookConfigAllowsInsecureSSL(cfg.InsecureSSL)
			target.RequestHeaders = marshalWebhookHeaders(headers)
			target.RequestBody = string(requestBody)
		}
		targets = append(targets, target)
	}
	return normalizeDurableWebhookTargets(targets)
}

func validateDurableWebhookOperationBounds(fanouts map[string]db.RepositoryWebhookFanout, targetsByKey map[string][]durableWebhookTarget) error {
	rows := 0
	manifestBytes := 0
	snapshotBytes := 0
	for eventKey, fanout := range fanouts {
		targets := targetsByKey[eventKey]
		rows += len(targets)
		manifestBytes += len(fanout.PayloadJSON) + len(fanout.TargetsJSON)
		for _, target := range targets {
			snapshotBytes += len(target.RequestURL) + len(target.RequestHeaders) + len(target.RequestBody) + len(target.TerminalError)
		}
		if rows > webhookOperationDeliveryMax || manifestBytes > webhookOperationManifestMax || snapshotBytes > webhookOperationSnapshotMax {
			return fmt.Errorf("durable webhook operation exceeds final snapshot limit")
		}
	}
	return nil
}

func decodeDurableWebhookTargets(fanout db.RepositoryWebhookFanout, repoID uint, event, action string) ([]durableWebhookTarget, error) {
	if fanout.RepositoryID != repoID || fanout.OperationID == "" || len(fanout.OperationID) > 64 || fanout.EventKey == "" || len(fanout.EventKey) > 512 ||
		fanout.Event == "" || len(fanout.Event) > 64 || fanout.Event != event || len(fanout.Action) > 64 || fanout.Action != action ||
		len(fanout.PayloadJSON) > durableWebhookPayloadMax || len(fanout.PayloadJSON)+len(fanout.TargetsJSON) > webhookFanoutStorageMax ||
		(fanout.State != webhookFanoutPending && fanout.State != webhookFanoutMaterialized) {
		return nil, fmt.Errorf("durable webhook fanout identity is invalid")
	}
	var targets []durableWebhookTarget
	decoder := json.NewDecoder(strings.NewReader(string(fanout.TargetsJSON)))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&targets); err != nil || len(targets) > webhookFanoutTargetMax {
		return nil, fmt.Errorf("durable webhook fanout targets are invalid")
	}
	var trailing any
	if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
		return nil, fmt.Errorf("durable webhook fanout targets are invalid")
	}
	return normalizeDurableWebhookTargets(targets)
}

func normalizeDurableWebhookTargets(targets []durableWebhookTarget) ([]durableWebhookTarget, error) {
	if len(targets) > webhookFanoutTargetMax {
		return nil, fmt.Errorf("durable webhook fanout targets are invalid")
	}
	normalized := make([]durableWebhookTarget, len(targets))
	seen := make(map[uint]bool, len(targets))
	persistedBytes := 0
	for index, target := range targets {
		if target.WebhookID == 0 || target.GUID == "" || len(target.GUID) > 64 || seen[target.WebhookID] {
			return nil, fmt.Errorf("durable webhook fanout target is invalid")
		}
		if target.TerminalError != "" {
			target.TerminalError = boundedWebhookTerminalError(target.TerminalError)
			clearDurableWebhookRequest(&target)
		} else {
			switch {
			case target.RequestURL == "" || len(target.RequestURL) > webhookRequestURLMax:
				target.TerminalError = "webhook request URL exceeds limit or is missing"
			case len(target.RequestHeaders) > webhookRequestHeadersMax:
				target.TerminalError = "webhook request headers exceed limit"
			case len(target.RequestBody) > webhookRequestBodyMax:
				target.TerminalError = "webhook request body exceeds limit"
			default:
				if _, err := url.ParseRequestURI(target.RequestURL); err != nil {
					target.TerminalError = "webhook request URL is invalid"
				} else if _, err := unmarshalWebhookHeaders(target.RequestHeaders); err != nil {
					target.TerminalError = "webhook request headers are invalid"
				}
			}
			if target.TerminalError != "" {
				clearDurableWebhookRequest(&target)
			}
		}
		persistedBytes += len(target.RequestURL) + len(target.RequestHeaders) + len(target.RequestBody) + len(target.TerminalError)
		if persistedBytes > webhookFanoutDeliveryMax {
			return nil, fmt.Errorf("durable webhook delivery set exceeds storage limit")
		}
		seen[target.WebhookID] = true
		normalized[index] = target
	}
	return normalized, nil
}

func boundedWebhookTerminalError(message string) string {
	if len(message) > webhookTerminalErrorMax {
		return message[:webhookTerminalErrorMax]
	}
	return message
}

func clearDurableWebhookRequest(target *durableWebhookTarget) {
	target.RequestURL = ""
	target.RequestInsecure = false
	target.RequestHeaders = "{}"
	target.RequestBody = ""
}

func buildDurableWebhookDeliveries(fanout db.RepositoryWebhookFanout, targets []durableWebhookTarget) ([]db.HookDelivery, error) {
	deliveries := make([]db.HookDelivery, 0, len(targets))
	now := time.Now().UTC()
	persistedBytes := 0
	for _, target := range targets {
		operationID := fanout.OperationID
		operationEventKey := fmt.Sprintf("%s:hook:%d", fanout.EventKey, target.WebhookID)
		persistedBytes += len(target.RequestURL) + len(target.RequestHeaders) + len(target.RequestBody) + len(target.TerminalError)
		if persistedBytes > webhookFanoutDeliveryMax {
			return nil, fmt.Errorf("durable webhook delivery set exceeds storage limit")
		}
		delivery := db.HookDelivery{
			WebhookID: target.WebhookID, RepositoryID: fanout.RepositoryID, GUID: target.GUID,
			OperationID: &operationID, OperationEventKey: &operationEventKey,
			SnapshotVersion: durableWebhookSnapshotVersion,
			Event:           fanout.Event, Action: fanout.Action, Status: webhookStatusPending,
			RequestURL: target.RequestURL, RequestInsecure: target.RequestInsecure,
			RequestHeaders: target.RequestHeaders, RequestPayload: db.LargeText(target.RequestBody), NextAttemptAt: now,
		}
		if target.TerminalError != "" {
			delivery.Status = webhookStatusError
			delivery.LastError = target.TerminalError
			delivery.DeliveredAt = &now
		}
		deliveries = append(deliveries, delivery)
	}
	return deliveries, nil
}

func materializeDurableWebhookOperationTx(
	tx *gorm.DB,
	operationID string,
	fanouts map[string]db.RepositoryWebhookFanout,
	targetsByKey map[string][]durableWebhookTarget,
) error {
	if err := retainLiveDurableWebhookTargetsTx(tx, fanouts, targetsByKey); err != nil {
		return err
	}
	deliveries := make([]db.HookDelivery, 0)
	pendingIDs := make([]uint, 0, len(fanouts))
	for eventKey, fanout := range fanouts {
		if fanout.State != webhookFanoutPending && fanout.State != webhookFanoutMaterialized {
			return fmt.Errorf("durable webhook fanout state is invalid")
		}
		built, err := buildDurableWebhookDeliveries(fanout, targetsByKey[eventKey])
		if err != nil {
			return err
		}
		deliveries = append(deliveries, built...)
		if fanout.State == webhookFanoutPending {
			pendingIDs = append(pendingIDs, fanout.ID)
		}
	}
	if len(deliveries) > webhookOperationDeliveryMax {
		return fmt.Errorf("durable webhook operation exceeds delivery row limit")
	}
	if len(deliveries) != 0 {
		if err := tx.Clauses(clause.OnConflict{DoNothing: true}).CreateInBatches(&deliveries, 16).Error; err != nil {
			return err
		}
		var persisted []db.HookDelivery
		if err := tx.Where("operation_id = ?", operationID).Order("operation_event_key ASC").Limit(webhookOperationDeliveryMax + 1).Find(&persisted).Error; err != nil {
			return err
		}
		if len(persisted) > webhookOperationDeliveryMax {
			return fmt.Errorf("durable webhook operation exceeds delivery row limit")
		}
		if err := verifyDurableWebhookDeliveries(deliveries, persisted); err != nil {
			return err
		}
	}
	if len(pendingIDs) == 0 {
		return nil
	}
	result := tx.Model(&db.RepositoryWebhookFanout{}).
		Where("operation_id = ? AND id IN ? AND state = ?", operationID, pendingIDs, webhookFanoutPending).
		Update("state", webhookFanoutMaterialized)
	if result.Error != nil {
		return result.Error
	}
	if result.RowsAffected != int64(len(pendingIDs)) {
		return fmt.Errorf("durable webhook fanout state changed")
	}
	return nil
}

func retainLiveDurableWebhookTargetsTx(
	tx *gorm.DB,
	fanouts map[string]db.RepositoryWebhookFanout,
	targetsByKey map[string][]durableWebhookTarget,
) error {
	targetIDs := make([]uint, 0)
	for eventKey := range fanouts {
		for _, target := range targetsByKey[eventKey] {
			targetIDs = append(targetIDs, target.WebhookID)
		}
	}
	if len(targetIDs) == 0 {
		return nil
	}

	var hooks []db.Webhook
	if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
		Select("id", "repository_id").
		Where("id IN ?", targetIDs).
		Limit(webhookOperationDeliveryMax + 1).
		Find(&hooks).Error; err != nil {
		return err
	}
	if len(hooks) > webhookOperationDeliveryMax {
		return fmt.Errorf("durable webhook operation exceeds delivery row limit")
	}
	liveRepositories := make(map[uint]uint, len(hooks))
	for _, hook := range hooks {
		liveRepositories[hook.ID] = hook.RepositoryID
	}
	for eventKey, fanout := range fanouts {
		targets := targetsByKey[eventKey]
		live := targets[:0]
		for _, target := range targets {
			if liveRepositories[target.WebhookID] == fanout.RepositoryID {
				live = append(live, target)
			}
		}
		targetsByKey[eventKey] = live
	}
	return nil
}

func verifyDurableWebhookDeliveries(expected, persisted []db.HookDelivery) error {
	if len(expected) != len(persisted) {
		return fmt.Errorf("durable webhook delivery set is incomplete")
	}
	byKey := make(map[string]db.HookDelivery, len(persisted))
	for _, delivery := range persisted {
		if delivery.OperationEventKey == nil {
			return fmt.Errorf("durable webhook delivery identity is invalid")
		}
		byKey[*delivery.OperationEventKey] = delivery
	}
	for _, delivery := range expected {
		stored, ok := byKey[*delivery.OperationEventKey]
		if !ok || stored.WebhookID != delivery.WebhookID || stored.RepositoryID != delivery.RepositoryID || stored.GUID != delivery.GUID ||
			stored.Event != delivery.Event || stored.Action != delivery.Action || stored.RequestURL != delivery.RequestURL ||
			stored.RequestInsecure != delivery.RequestInsecure || stored.RequestHeaders != delivery.RequestHeaders || stored.RequestPayload != delivery.RequestPayload {
			return fmt.Errorf("durable webhook delivery snapshot changed")
		}
	}
	return nil
}