package billing

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
	"github.com/aws/smithy-go"

	"github.com/mem9-ai/drive9/pkg/metrics"
)

// SQSSender sends one message body to the billing queue. *sqsSender
// implements it; tests substitute fakes.
type SQSSender interface {
	SendMessage(ctx context.Context, queueURL, body string) error
}

type sqsSender struct {
	client *sqs.Client
}

// NewSQSSender builds an SQSSender from an AWS config (workload identity;
// no long-lived credentials).
func NewSQSSender(cfg aws.Config) SQSSender {
	return &sqsSender{client: sqs.NewFromConfig(cfg)}
}

func (s *sqsSender) SendMessage(ctx context.Context, queueURL, body string) error {
	_, err := s.client.SendMessage(ctx, &sqs.SendMessageInput{
		QueueUrl:    aws.String(queueURL),
		MessageBody: aws.String(body),
	})
	return err
}

// isRetryableSendError classifies SQS failures: message-shape problems never
// succeed on retry and are marked permanent; throttling (429) and transient
// errors retry with backoff; everything else is retried.
func isRetryableSendError(err error) bool {
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		switch apiErr.ErrorCode() {
		case "InvalidMessageContents", "UnsupportedOperation":
			return false
		}
		return true
	}
	var respErr *awshttp.ResponseError
	if errors.As(err, &respErr) {
		code := respErr.HTTPStatusCode()
		if code == http.StatusTooManyRequests {
			return true
		}
		if code >= 400 && code < 500 {
			return false
		}
	}
	// Network/transport errors are retryable.
	return true
}

// Sender drains the billing SQS outbox with lease-based claiming. The
// database transaction never contains the SQS network call: rows are claimed
// (status=sending), sent, then marked sent or returned to pending.
type Sender struct {
	cfg   Config
	meta  MetaStore
	sqs   SQSSender
	owner string
}

func newSender(cfg Config, store MetaStore, sqsClient SQSSender, owner string) *Sender {
	return &Sender{cfg: cfg.withDefaults(), meta: store, sqs: sqsClient, owner: owner}
}

// sendAttemptTimeout bounds a single SendMessage so a hung connection cannot
// stall the outbox loop while holding a claimed batch.
const sendAttemptTimeout = 30 * time.Second

// backoffForAttempt spaces retryable send retries exponentially: 30s, 60s,
// … capped at 10 minutes. With the default MaxSendAttempts (8) a batch spans
// hours before permanent_failed.
func backoffForAttempt(attempts int) time.Duration {
	if attempts < 0 {
		attempts = 0
	}
	d := 30 * time.Second
	for i := 0; i < attempts && d < 10*time.Minute; i++ {
		d *= 2
	}
	if d > 10*time.Minute {
		d = 10 * time.Minute
	}
	return d
}

// SendOnce claims one batch and attempts delivery. Shadow mode
// (SendEnabled=false) leaves rows pending so the pipeline can be verified
// without emitting messages.
func (s *Sender) SendOnce(ctx context.Context) error {
	if !s.cfg.SendEnabled {
		return nil
	}
	if s.sqs == nil || s.cfg.SQSQueueURL == "" {
		return fmt.Errorf("billing sender enabled but SQS client or queue URL is missing")
	}
	rows, err := s.meta.ClaimBillingOutboxBatch(ctx, s.owner, s.cfg.OutboxLease, s.cfg.OutboxBatchSize)
	if err != nil {
		return fmt.Errorf("claim billing outbox: %w", err)
	}
	for _, row := range rows {
		if ctx.Err() != nil {
			return nil
		}
		sendCtx, cancel := context.WithTimeout(ctx, sendAttemptTimeout)
		sendErr := s.sqs.SendMessage(sendCtx, s.cfg.SQSQueueURL, string(row.Payload))
		cancel()
		if sendErr == nil {
			if err := s.meta.MarkBillingOutboxSent(ctx, row.CostID, s.owner); err != nil {
				metrics.RecordTenantOperationCount("", "billing", "sqs_send", "ack_error")
				return fmt.Errorf("mark billing outbox sent %s: %w", row.CostID, err)
			}
			metrics.RecordTenantOperationCount("", "billing", "sqs_send", "ok")
			continue
		}
		retryable := isRetryableSendError(sendErr)
		exhausted := row.Attempts+1 >= s.cfg.MaxSendAttempts
		var nextAttemptAt time.Time
		result := "retry"
		if !retryable || exhausted {
			result = "permanent_failed"
		} else {
			nextAttemptAt = time.Now().Add(backoffForAttempt(row.Attempts))
		}
		metrics.RecordTenantOperationCount("", "billing", "sqs_send", result)
		if err := s.meta.MarkBillingOutboxFailed(ctx, row.CostID, s.owner, nextAttemptAt, sendErr); err != nil {
			return fmt.Errorf("mark billing outbox failed %s: %w", row.CostID, err)
		}
	}
	return nil
}

// ReportOutboxStats exports outbox depth/age gauges for observability.
func (s *Sender) ReportOutboxStats(ctx context.Context) {
	stats, err := s.meta.BillingOutboxStats(ctx)
	if err != nil {
		metrics.RecordTenantOperationCount("", "billing", "outbox_stats", "error")
		return
	}
	metrics.RecordGauge("billing", "outbox_pending", float64(stats.Pending))
	metrics.RecordGauge("billing", "outbox_sending", float64(stats.Sending))
	metrics.RecordGauge("billing", "outbox_permanent_failed", float64(stats.PermanentFailed))
	metrics.RecordGauge("billing", "outbox_oldest_pending_seconds", stats.OldestPendingAgeSec)
}