package serviceactor

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"io"
	"reflect"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/core/identifiers"
	"github.com/tidbcloud/AgentTeam/internal/core/runeffect"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
)

const (
	eventDeliveryRetentionDeleteBatchDefault = 500
	eventDeliveryRetentionDeleteBatchMax     = 1000
)

// EventDelivery records the per-subscription delivery ledger for one service event.
type EventDelivery struct {
	ID                     string
	WorkspaceID            string
	EventSubscriptionID    string
	WebhookReceiptID       string
	ServiceDefinitionID    string
	ServiceInstallationID  string
	ServiceMemberID        string
	ServiceConnectionID    string
	ServiceConnectionRef   string
	ServiceTargetBindingID string
	ServiceKind            string
	SubscriptionKind       string
	TargetKind             string
	TargetRef              string
	DeliveryScopeID        string
	PermissionCheckScopeID string
	OwnerMemberIDSnapshot  string
	AuthorizerMemberID     string
	ExternalResourceType   string
	ExternalResourceKey    string
	ExternalResourceRef    string
	ExternalResourceURL    string
	EventType              string
	EventAction            string
	EventID                string
	EventKey               string
	EventPayload           []byte
	DeliveryPolicy         string
	NotifyPrincipal        bool
	Status                 string
	IdempotencyKey         string
	AttemptCount           int
	AttemptedAt            *time.Time
	CompletedAt            *time.Time
	NextRetryAt            *time.Time
	EffectKind             string
	EffectRef              string
	MessageID              string
	ErrorCode              string
	ErrorDetail            string
	LeaseOwner             string
	LeaseExpiresAt         *time.Time
	CreatedAt              time.Time
	UpdatedAt              time.Time
}

// EventDeliveryReserve describes the immutable service event facts needed to
// reserve a delivery before any AgentTeam-visible effect is written.
type EventDeliveryReserve struct {
	ID                   string
	WorkspaceID          string
	EventSubscriptionID  string
	WebhookReceiptID     string
	EventType            string
	EventAction          string
	EventID              string
	EventKey             string
	EventPayload         []byte
	ExternalResourceType string
	ExternalResourceKey  string
	ExternalResourceRef  string
	ExternalResourceURL  string
	IdempotencyKey       string
	Now                  time.Time
}

// EventDeliveryCompletion records the final effect/error state of a delivery.
type EventDeliveryCompletion struct {
	ID     string
	Status runeffect.Status
	Effect runeffect.Effect
	Error  runeffect.Error
	Now    time.Time
}

// EventDeliveryRetentionFilter selects old service event delivery rows for
// bounded cleanup.
type EventDeliveryRetentionFilter struct {
	ServiceKind      string
	SubscriptionKind string
	Cutoff           time.Time
	Limit            int
}

// ReserveEventDelivery reserves a service event delivery idempotently. Existing
// terminal deliveries are returned as duplicates; failed deliveries are reopened
// for retry with an incremented attempt count.
func (store *TxStore) ReserveEventDelivery(ctx context.Context, params EventDeliveryReserve) (EventDelivery, bool, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return EventDelivery{}, false, err
	}
	return reserveEventDelivery(ctx, execer, params)
}

// CompleteEventDelivery records the final run-effect state for a delivery.
func (store *TxStore) CompleteEventDelivery(ctx context.Context, params EventDeliveryCompletion) error {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return err
	}
	return completeEventDelivery(ctx, execer, params)
}

// LoadEventDeliveryForUpdate locks and loads one delivery row.
func (store *TxStore) LoadEventDeliveryForUpdate(ctx context.Context, workspaceID string, deliveryID string) (EventDelivery, error) {
	queryer, err := queryerFromTxStore(store)
	if err != nil {
		return EventDelivery{}, err
	}
	return loadEventDeliveryForUpdate(ctx, queryer, workspaceID, deliveryID)
}

// DeleteEventDeliveriesBefore deletes old delivery ledger rows in bounded
// batches. Callers should rerun when the returned count reaches the normalized
// limit.
func (store *Store) DeleteEventDeliveriesBefore(ctx context.Context, filter EventDeliveryRetentionFilter) (int64, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return 0, err
	}
	return deleteEventDeliveriesBefore(ctx, execer, filter)
}

func deleteEventDeliveriesBefore(ctx context.Context, execer coredb.QueryExecer, filter EventDeliveryRetentionFilter) (int64, error) {
	serviceKind := trim(filter.ServiceKind)
	subscriptionKind := trim(filter.SubscriptionKind)
	if serviceKind == "" || subscriptionKind == "" {
		return 0, badRequest("service event delivery retention requires service and subscription kind")
	}
	limit := normalizeEventDeliveryRetentionLimit(filter.Limit)
	rows, err := execer.Query(ctx, `
SELECT id
FROM service_event_deliveries
WHERE service_kind = $1
  AND subscription_kind = $2
  AND attempted_at < $3
ORDER BY attempted_at ASC, id ASC
LIMIT $4
`, serviceKind, subscriptionKind, filter.Cutoff.UTC(), limit)
	if err != nil {
		return 0, fmt.Errorf("select old service event deliveries: %w", err)
	}
	defer rows.Close()

	ids := make([]string, 0, limit)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return 0, fmt.Errorf("scan old service event deliveries: %w", err)
		}
		if trimmed := strings.TrimSpace(id); trimmed != "" {
			ids = append(ids, trimmed)
		}
	}
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf("iterate old service event deliveries: %w", err)
	}

	if len(ids) == 0 {
		return 0, nil
	}

	deleteIDsCTE, deleteIDArgs, _, err := coredb.BuildStringValuesCTE("delete_delivery_ids", "id", ids, 1)
	if err != nil {
		return 0, fmt.Errorf("build old service event delivery delete: %w", err)
	}
	result, err := execer.Exec(ctx, fmt.Sprintf(`
WITH %s
DELETE FROM service_event_deliveries
WHERE id IN (SELECT id FROM delete_delivery_ids)
`, deleteIDsCTE), deleteIDArgs...)
	if err != nil {
		return 0, fmt.Errorf("delete old service event deliveries: %w", err)
	}
	return result.RowsAffected(), nil
}

func normalizeEventDeliveryRetentionLimit(limit int) int {
	if limit <= 0 {
		return eventDeliveryRetentionDeleteBatchDefault
	}
	if limit > eventDeliveryRetentionDeleteBatchMax {
		return eventDeliveryRetentionDeleteBatchMax
	}
	return limit
}

func reserveEventDelivery(ctx context.Context, execer coredb.QueryExecer, params EventDeliveryReserve) (EventDelivery, bool, error) {
	workspaceID := trim(params.WorkspaceID)
	subscriptionID := trim(params.EventSubscriptionID)
	idempotencyKey := trim(params.IdempotencyKey)
	eventType := trim(params.EventType)
	eventKey := trim(params.EventKey)
	if workspaceID == "" || subscriptionID == "" || idempotencyKey == "" || eventType == "" || eventKey == "" {
		return EventDelivery{}, false, badRequest("service event delivery workspace, subscription, idempotency, event type, and event key are required")
	}
	if existing, found, err := loadEventDeliveryByIdempotencyKeyForUpdate(ctx, execer, idempotencyKey); err != nil {
		return EventDelivery{}, false, err
	} else if found {
		return reserveExistingEventDelivery(ctx, execer, existing, params)
	}

	deliveryID := trim(params.ID)
	var err error
	if deliveryID == "" {
		deliveryID, err = identifiers.NewEntityID("svced")
		if err != nil {
			return EventDelivery{}, false, fmt.Errorf("generate service event delivery id: %w", err)
		}
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	result, err := execer.Exec(ctx, `
INSERT INTO service_event_deliveries (
    id, workspace_id, service_event_subscription_id, service_webhook_receipt_id,
    service_definition_id, service_installation_id, service_member_id,
    service_connection_id, service_connection_ref, service_target_binding_id,
    service_kind, subscription_kind, target_kind, target_ref,
    delivery_scope_id, permission_check_scope_id, owner_member_id_snapshot,
    authorizer_member_id_snapshot, external_resource_type, external_resource_key,
    external_resource_ref, external_resource_url, event_type, event_action,
    event_id, event_key, event_payload, delivery_policy, notify_principal,
    status, idempotency_key, attempt_count, attempted_at,
    effect_kind, effect_ref, message_id, error_code, error_detail,
    created_at, updated_at
)
SELECT
    $1, subscription.workspace_id, subscription.id, $2,
    subscription.service_definition_id, subscription.service_installation_id,
    subscription.service_member_id, subscription.service_connection_id,
    subscription.service_connection_ref, subscription.service_target_binding_id,
    subscription.service_kind, subscription.subscription_kind,
    subscription.target_kind, subscription.target_ref,
    subscription.delivery_scope_id, subscription.permission_check_scope_id,
    subscription.owner_member_id, subscription.authorizer_member_id,
    COALESCE(NULLIF($3, ''), subscription.external_resource_type),
    COALESCE(NULLIF($4, ''), subscription.external_resource_key),
    COALESCE(NULLIF($5, ''), subscription.external_resource_ref),
    COALESCE(NULLIF($6, ''), subscription.external_resource_url),
    $7, $8, $9, $10, $11,
    subscription.delivery_policy, subscription.notify_principal,
    'running', $12, 1, $13,
    'none', '', NULL, '', '',
    $13, $13
FROM service_event_subscriptions subscription
WHERE subscription.workspace_id = $14
  AND subscription.id = $15
  AND subscription.removed_at IS NULL
`, deliveryID, nullableString(trim(params.WebhookReceiptID)), trim(params.ExternalResourceType), trim(params.ExternalResourceKey), trim(params.ExternalResourceRef), trim(params.ExternalResourceURL),
		eventType, trim(params.EventAction), trim(params.EventID), eventKey, nullableBytes(params.EventPayload),
		idempotencyKey, now, workspaceID, subscriptionID)
	if err != nil {
		if coredb.IsUniqueViolation(err) {
			if existing, found, loadErr := loadEventDeliveryByIdempotencyKeyForUpdate(ctx, execer, idempotencyKey); loadErr != nil {
				return EventDelivery{}, false, loadErr
			} else if found {
				return reserveExistingEventDelivery(ctx, execer, existing, params)
			}
		}
		return EventDelivery{}, false, fmt.Errorf("insert service event delivery: %w", err)
	}
	if result.RowsAffected() == 0 {
		return EventDelivery{}, false, notFound("service event subscription not found")
	}
	delivery, err := loadEventDelivery(ctx, execer, deliveryID)
	return delivery, false, err
}

func reserveExistingEventDelivery(ctx context.Context, execer coredb.QueryExecer, existing EventDelivery, params EventDeliveryReserve) (EventDelivery, bool, error) {
	if !eventDeliveryReplayMatches(existing, params) {
		return EventDelivery{}, false, conflict("service event delivery idempotency key was already used with a different event")
	}
	status, err := runeffect.NormalizeStatus(existing.Status)
	if err != nil {
		return EventDelivery{}, false, err
	}
	if status == runeffect.StatusFailed {
		return reopenEventDeliveryForRetry(ctx, execer, existing.ID, params)
	}
	if status.IsActive() {
		return EventDelivery{}, false, conflict("service event delivery is already processing; retry later")
	}
	return existing, true, nil
}

func eventDeliveryReplayMatches(existing EventDelivery, params EventDeliveryReserve) bool {
	if trim(existing.WorkspaceID) != trim(params.WorkspaceID) ||
		trim(existing.EventSubscriptionID) != trim(params.EventSubscriptionID) ||
		trim(existing.EventType) != trim(params.EventType) ||
		trim(existing.EventAction) != trim(params.EventAction) ||
		trim(existing.EventID) != trim(params.EventID) ||
		trim(existing.EventKey) != trim(params.EventKey) {
		return false
	}
	for _, pair := range [][2]string{
		{existing.ExternalResourceType, params.ExternalResourceType},
		{existing.ExternalResourceKey, params.ExternalResourceKey},
		{existing.ExternalResourceRef, params.ExternalResourceRef},
		{existing.ExternalResourceURL, params.ExternalResourceURL},
	} {
		if requested := trim(pair[1]); requested != "" && trim(pair[0]) != requested {
			return false
		}
	}
	return eventDeliveryJSONEqual(existing.EventPayload, params.EventPayload)
}

func eventDeliveryJSONEqual(left []byte, right []byte) bool {
	left = bytes.TrimSpace(left)
	right = bytes.TrimSpace(right)
	if len(left) == 0 || bytes.Equal(left, []byte("null")) {
		left = []byte("{}")
	}
	if len(right) == 0 || bytes.Equal(right, []byte("null")) {
		right = []byte("{}")
	}
	leftValue, leftOK := decodeEventDeliveryJSON(left)
	rightValue, rightOK := decodeEventDeliveryJSON(right)
	return leftOK && rightOK && reflect.DeepEqual(leftValue, rightValue)
}

func decodeEventDeliveryJSON(payload []byte) (any, bool) {
	decoder := json.NewDecoder(bytes.NewReader(payload))
	decoder.UseNumber()
	var value any
	if err := decoder.Decode(&value); err != nil {
		return nil, false
	}
	var extra any
	if err := decoder.Decode(&extra); err != io.EOF {
		return nil, false
	}
	return value, true
}

func reopenEventDeliveryForRetry(ctx context.Context, execer coredb.QueryExecer, deliveryID string, params EventDeliveryReserve) (EventDelivery, bool, error) {
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	_, err := execer.Exec(ctx, `
UPDATE service_event_deliveries
SET service_webhook_receipt_id = $2,
    event_type = $3,
    event_action = $4,
    event_id = $5,
    event_key = $6,
    event_payload = $7,
    external_resource_type = CASE WHEN $8 <> '' THEN $8 ELSE external_resource_type END,
    external_resource_key = CASE WHEN $9 <> '' THEN $9 ELSE external_resource_key END,
    external_resource_ref = CASE WHEN $10 <> '' THEN $10 ELSE external_resource_ref END,
    external_resource_url = COALESCE(NULLIF($11, ''), external_resource_url),
    status = 'running',
    attempt_count = attempt_count + 1,
    attempted_at = $12,
    completed_at = NULL,
    next_retry_at = NULL,
    effect_kind = 'none',
    effect_ref = '',
    message_id = NULL,
    error_code = '',
    error_detail = '',
    lease_owner = '',
    lease_expires_at = NULL,
    updated_at = $12
WHERE id = $1
`, trim(deliveryID), nullableString(trim(params.WebhookReceiptID)), trim(params.EventType),
		trim(params.EventAction), trim(params.EventID), trim(params.EventKey), nullableBytes(params.EventPayload),
		trim(params.ExternalResourceType), trim(params.ExternalResourceKey), trim(params.ExternalResourceRef),
		trim(params.ExternalResourceURL), now)
	if err != nil {
		return EventDelivery{}, false, fmt.Errorf("reopen service event delivery: %w", err)
	}
	delivery, err := loadEventDelivery(ctx, execer, deliveryID)
	return delivery, false, err
}

func completeEventDelivery(ctx context.Context, execer coredb.QueryExecer, params EventDeliveryCompletion) error {
	deliveryID := trim(params.ID)
	if deliveryID == "" {
		return nil
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	status := params.Status
	if status == "" {
		status = runeffect.StatusSucceeded
	}
	current, err := loadEventDelivery(ctx, execer, deliveryID)
	if err != nil {
		return err
	}
	effect, err := params.Effect.Normalize()
	if err != nil {
		return err
	}
	failure := params.Error.Normalize()
	contract := runeffect.Contract{
		Status:         status,
		IdempotencyKey: current.IdempotencyKey,
		AttemptCount:   current.AttemptCount,
		AttemptedAt:    current.AttemptedAt,
		CompletedAt:    &now,
		Effect:         effect,
		Error:          failure,
	}
	if err := contract.Validate(); err != nil {
		return err
	}
	_, err = execer.Exec(ctx, `
UPDATE service_event_deliveries
SET status = $2,
    completed_at = $3,
    effect_kind = $4,
    effect_ref = $5,
    message_id = $6,
    error_code = $7,
    error_detail = $8,
    lease_owner = '',
    lease_expires_at = NULL,
    updated_at = $3
WHERE id = $1
`, deliveryID, string(status), now, string(effect.Kind), trim(effect.Ref), nullableString(effect.MessageID), failure.Code, failure.Detail)
	if err != nil {
		return fmt.Errorf("complete service event delivery: %w", err)
	}
	return nil
}

func loadEventDelivery(ctx context.Context, queryer coredb.QueryRower, deliveryID string) (EventDelivery, error) {
	row := queryer.QueryRow(ctx, eventDeliverySelectFromSQL("service_event_deliveries", "")+` WHERE id = $1`, trim(deliveryID))
	delivery, err := scanEventDelivery(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return EventDelivery{}, notFound("service event delivery not found")
		}
		return EventDelivery{}, fmt.Errorf("load service event delivery: %w", err)
	}
	return delivery, nil
}

func loadEventDeliveryForUpdate(ctx context.Context, queryer coredb.QueryRower, workspaceID string, deliveryID string) (EventDelivery, error) {
	row := queryer.QueryRow(ctx, eventDeliverySelectFromSQL("service_event_deliveries", "")+`
WHERE workspace_id = $1
  AND id = $2
FOR UPDATE
`, trim(workspaceID), trim(deliveryID))
	delivery, err := scanEventDelivery(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return EventDelivery{}, notFound("service event delivery not found")
		}
		return EventDelivery{}, fmt.Errorf("load service event delivery for update: %w", err)
	}
	return delivery, nil
}

func loadEventDeliveryByIdempotencyKeyForUpdate(ctx context.Context, queryer coredb.QueryRower, idempotencyKey string) (EventDelivery, bool, error) {
	row := queryer.QueryRow(ctx, eventDeliverySelectFromSQL("service_event_deliveries", "")+`
WHERE idempotency_key = $1
FOR UPDATE
`, trim(idempotencyKey))
	delivery, err := scanEventDelivery(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return EventDelivery{}, false, nil
		}
		return EventDelivery{}, false, fmt.Errorf("load service event delivery by idempotency key: %w", err)
	}
	return delivery, true, nil
}

func eventDeliverySelectFromSQL(from string, qualifier string) string {
	columnPrefix := ""
	if qualifier != "" {
		columnPrefix = qualifier + "."
	}
	return `
SELECT ` + eventDeliverySelectColumnsSQL(columnPrefix) + `
FROM ` + from
}

func eventDeliverySelectColumnsSQL(columnPrefix string) string {
	return strings.ReplaceAll(`{{col}}id, {{col}}workspace_id, {{col}}service_event_subscription_id,
       COALESCE({{col}}service_webhook_receipt_id, ''), {{col}}service_definition_id,
       {{col}}service_installation_id, {{col}}service_member_id,
       COALESCE({{col}}service_connection_id, ''), {{col}}service_connection_ref,
       {{col}}service_target_binding_id, {{col}}service_kind, {{col}}subscription_kind, {{col}}target_kind,
       {{col}}target_ref, {{col}}delivery_scope_id, {{col}}permission_check_scope_id,
       {{col}}owner_member_id_snapshot, {{col}}authorizer_member_id_snapshot,
       {{col}}external_resource_type, {{col}}external_resource_key, {{col}}external_resource_ref,
       {{col}}external_resource_url, {{col}}event_type, {{col}}event_action, {{col}}event_id, {{col}}event_key,
       {{col}}event_payload, {{col}}delivery_policy, {{col}}notify_principal, {{col}}status,
       {{col}}idempotency_key, {{col}}attempt_count, {{col}}attempted_at, {{col}}completed_at, {{col}}next_retry_at,
       {{col}}effect_kind, {{col}}effect_ref, COALESCE({{col}}message_id, ''), {{col}}error_code,
       {{col}}error_detail, {{col}}lease_owner, {{col}}lease_expires_at, {{col}}created_at, {{col}}updated_at`, "{{col}}", columnPrefix)
}

func scanEventDelivery(row coredb.Row) (EventDelivery, error) {
	var delivery EventDelivery
	var attemptCount int64
	var attemptedAt, completedAt, nextRetryAt, leaseExpiresAt sql.NullTime
	var eventPayload []byte
	if err := row.Scan(
		&delivery.ID,
		&delivery.WorkspaceID,
		&delivery.EventSubscriptionID,
		&delivery.WebhookReceiptID,
		&delivery.ServiceDefinitionID,
		&delivery.ServiceInstallationID,
		&delivery.ServiceMemberID,
		&delivery.ServiceConnectionID,
		&delivery.ServiceConnectionRef,
		&delivery.ServiceTargetBindingID,
		&delivery.ServiceKind,
		&delivery.SubscriptionKind,
		&delivery.TargetKind,
		&delivery.TargetRef,
		&delivery.DeliveryScopeID,
		&delivery.PermissionCheckScopeID,
		&delivery.OwnerMemberIDSnapshot,
		&delivery.AuthorizerMemberID,
		&delivery.ExternalResourceType,
		&delivery.ExternalResourceKey,
		&delivery.ExternalResourceRef,
		&delivery.ExternalResourceURL,
		&delivery.EventType,
		&delivery.EventAction,
		&delivery.EventID,
		&delivery.EventKey,
		&eventPayload,
		&delivery.DeliveryPolicy,
		&delivery.NotifyPrincipal,
		&delivery.Status,
		&delivery.IdempotencyKey,
		&attemptCount,
		&attemptedAt,
		&completedAt,
		&nextRetryAt,
		&delivery.EffectKind,
		&delivery.EffectRef,
		&delivery.MessageID,
		&delivery.ErrorCode,
		&delivery.ErrorDetail,
		&delivery.LeaseOwner,
		&leaseExpiresAt,
		&delivery.CreatedAt,
		&delivery.UpdatedAt,
	); err != nil {
		return EventDelivery{}, err
	}
	delivery.ID = trim(delivery.ID)
	delivery.WorkspaceID = trim(delivery.WorkspaceID)
	delivery.EventSubscriptionID = trim(delivery.EventSubscriptionID)
	delivery.WebhookReceiptID = trim(delivery.WebhookReceiptID)
	delivery.ServiceDefinitionID = trim(delivery.ServiceDefinitionID)
	delivery.ServiceInstallationID = trim(delivery.ServiceInstallationID)
	delivery.ServiceMemberID = trim(delivery.ServiceMemberID)
	delivery.ServiceConnectionID = trim(delivery.ServiceConnectionID)
	delivery.ServiceConnectionRef = trim(delivery.ServiceConnectionRef)
	delivery.ServiceTargetBindingID = trim(delivery.ServiceTargetBindingID)
	delivery.ServiceKind = trim(delivery.ServiceKind)
	delivery.SubscriptionKind = trim(delivery.SubscriptionKind)
	delivery.TargetKind = trim(delivery.TargetKind)
	delivery.TargetRef = trim(delivery.TargetRef)
	delivery.Status = trim(delivery.Status)
	delivery.IdempotencyKey = trim(delivery.IdempotencyKey)
	delivery.AttemptCount = int(attemptCount)
	delivery.AttemptedAt = ptrFromNullTime(attemptedAt)
	delivery.CompletedAt = ptrFromNullTime(completedAt)
	delivery.NextRetryAt = ptrFromNullTime(nextRetryAt)
	delivery.LeaseExpiresAt = ptrFromNullTime(leaseExpiresAt)
	delivery.EventPayload = eventPayload
	return delivery, nil
}