package host

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"math"
	"strings"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/alerting"
	"github.com/carlosframework/platform/internal/metrics"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
)

// Alert sinks: where a firing or clearing transition from metrics.Evaluate
// goes once alertTick has published the new state to the bucket.
//
// Two sinks, either, both, or neither:
//
//	CARLOS_ALERT_WEBHOOK        the event as JSON, POSTed (httpPostWebhook)
//	CARLOS_ALERT_SNS_TOPIC_ARN  a plain-text subject and message, published
//	                            to that SNS topic with the box's instance role
//
// SNS is here because a webhook cannot reach it: Publish needs a SigV4
// signature, which only the AWS SDK on a box holding the role can make. The
// flagship's topics (carlos-flagship-<region>-alerts) already carry confirmed
// email subscriptions, so this is the path from a disk at 91% to an inbox
// (discussion #192).
//
// Delivery is best effort and never fatal. A sink that fails is logged and
// counted, the other sink still runs, and the tick carries on. Failed sends
// are not rescheduled by a later tick: Evaluate emits an event only on a
// transition, so the next tick has no event to resend. (The SDK itself tries a
// retryable failure up to three times inside one publish, and a failed bucket
// write in alertTick repeats the whole transition next tick, delivery
// included.) The count is what makes a lost send visible, through
// status.AlertDelivery in the box snapshot.
//
// The delivery itself (the webhook post, the SNS publish and its timeout)
// is internal/alerting's, shared with the console; this file formats the
// box's alerts and counts their failures. The same sinks carry the
// operator pages of the moves code (PageOperator, moves spec §13).

// snsPublisher is the one SNS call the sinks make.
type snsPublisher = alerting.Publisher

// alertCounters counts failed sends since the process started, and holds the
// one fact the status loop cannot derive from the env: whether the SNS client
// was actually built. They are separate goroutines, so this is their seam.
type alertCounters struct {
	webhook atomic.Int64
	sns     atomic.Int64
	// snsInit is snsInitUnknown until buildAlertSinks has run for a valid
	// topic ARN, then snsInitOK or snsInitFailed.
	snsInit atomic.Int32
}

const (
	snsInitUnknown int32 = iota
	snsInitOK
	snsInitFailed
)

// processAlertCounters is the process-wide count the metrics loop writes
// and the status loop reads. They are separate goroutines with no other
// shared state, so this is the seam between them.
var processAlertCounters alertCounters

// alertSinks is the configured delivery for one agent: alerting's sinks,
// counting their failures in counts.
type alertSinks struct {
	sinks  alerting.Sinks
	counts *alertCounters
}

// newAlertSinks builds a sink set. An empty webhookURL or topicARN switches
// that sink off; a nil publisher does the same for SNS.
func newAlertSinks(webhookURL string, post func(url string, body []byte) error,
	topicARN string, pub snsPublisher, counts *alertCounters) *alertSinks {
	if counts == nil {
		counts = &processAlertCounters
	}
	return &alertSinks{counts: counts, sinks: alerting.Sinks{WebhookURL: webhookURL, Post: post, TopicARN: topicARN, Pub: pub,
		WebhookFailed: func() { counts.webhook.Add(1) }, SNSFailed: func() { counts.sns.Add(1) }}}
}

// deliver sends one transition to every configured sink. Safe on a nil
// receiver, which means "no sinks".
func (a *alertSinks) deliver(ctx context.Context, rules metrics.Rules, at time.Time, ev metrics.Event) {
	if a == nil {
		return
	}
	if a.sinks.HasWebhook() {
		evBody, err := json.Marshal(ev)
		if err != nil {
			log.Printf("host: alert event %s: marshal: %v", ev.Rule, err)
		} else if err := a.sinks.PostWebhook(evBody); err != nil {
			log.Printf("host: alert webhook post (rule %s): %v", ev.Rule, err)
		}
	}
	if a.sinks.HasSNS() {
		subject, message := formatAlert(rules, at, ev)
		if err := a.sinks.PublishSNS(ctx, subject, message); err != nil {
			log.Printf("host: alert sns publish (rule %s): %v", ev.Rule, err)
		}
	}
}

// PageOperator pages an operator through this agent's sinks: the same
// webhook and SNS topic deliver uses, its failures counted with theirs
// (alerting.Sinks.Page). Safe on a nil receiver, which means "no sinks":
// nil, and one line in the journal.
func (a *alertSinks) PageOperator(ctx context.Context, level, subject, body string) error {
	if a == nil {
		return (*alerting.Sinks)(nil).Page(ctx, level, subject, body)
	}
	return a.sinks.Page(ctx, level, subject, body)
}

// PageOperator pages an operator on the sinks the metrics loop armed
// (armPaging): the moves code's paging primitive (moves spec §13). Before
// they are armed, or with none configured, it returns nil and says so once.
func PageOperator(ctx context.Context, level, subject, body string) error {
	return alerting.Page(ctx, level, subject, body)
}

// armAgentAlerts builds the agent's alert sinks from the env and arms
// paging on them (armPaging). runAgent calls it once at startup, before
// anything that can run a moves pass, and hands the same instance to the
// metrics loop, so arming never waits on the metrics store.
func armAgentAlerts(ctx context.Context, getenv func(string) string, counts *alertCounters, newClient snsClientFactory) *alertSinks {
	sinks := buildAlertSinks(ctx, getenv, counts, newClient)
	armPaging(sinks)
	return sinks
}

// armPaging makes a the sinks PageOperator pages on. nil means none.
func armPaging(a *alertSinks) {
	if a == nil {
		alerting.Arm(nil)
		return
	}
	alerting.Arm(&a.sinks)
}

// deliverRelease sends a D1 refusal to every configured sink. The webhook
// body is the record with kind "release_refused"; SNS gets a sentence.
func (a *alertSinks) deliverRelease(ctx context.Context, r release.Refusal) {
	if a == nil {
		return
	}
	if a.sinks.HasWebhook() {
		body, err := json.Marshal(struct {
			Kind string `json:"kind"`
			// The record's own Kind is shadowed by the envelope's "kind"
			// above, so it travels as refusal_kind.
			RefusalKind string `json:"refusal_kind,omitempty"`
			release.Refusal
		}{"release_refused", r.Kind, r})
		if err == nil {
			if err := a.sinks.PostWebhook(body); err != nil {
				log.Printf("host: alert webhook post (release refusal %s): %v", r.Host, err)
			}
		}
	}
	if a.sinks.HasSNS() {
		subject, msg := formatReleaseAlert(r)
		if err := a.sinks.PublishSNS(ctx, subject, msg); err != nil {
			log.Printf("host: alert sns publish (release refusal %s): %v", r.Host, err)
		}
	}
}

// formatReleaseAlert is a refusal's SNS subject and message. A recovery that
// failed reads as one in the subject (spec §2.6): "rolled back" over a box
// whose restored build never came up would send the operator back to bed.
func formatReleaseAlert(r release.Refusal) (subject, msg string) {
	if r.Kind == release.RefusalWrongProgram {
		// Nothing started and nothing was rolled back: the box never left its
		// old build, and "--retry" would be refused the same way. A version's
		// manifest is immutable, so the way out is a NEW version.
		subject = fmt.Sprintf("%s: %s refused, wrong program (still on %s)", r.Host, r.Version, r.RestoredVersion)
		if len(subject) > 100 { // SNS subject limit
			subject = subject[:100]
		}
		msg = fmt.Sprintf("%s refused %s at %s: %s\n\nShip a new version with carlos ship --program-change and promote it, if the change is intended, or promote a different build.",
			r.Host, r.Version, r.RefusedAt.Format(time.RFC3339), r.Reason)
		return subject, msg
	}
	if r.Kind == release.RefusalNotProven {
		// D3's binding check: the same shape as the wrong program. Nothing
		// started and nothing was rolled back, and "--retry" would re-sign
		// the pointer over the same staged bytes.
		still := r.RestoredVersion
		if still == "" {
			still = "nothing"
		}
		subject = fmt.Sprintf("%s: %s refused, staged bytes do not match dev's proof (still on %s)", r.Host, r.Version, still)
		if len(subject) > 100 { // SNS subject limit
			subject = subject[:100]
		}
		msg = fmt.Sprintf("%s refused %s at %s: %s\n\nThe staged bytes do not match dev's proof, so %s was never pointed live. The box is still on %s. Roll the exact bytes dev proved.",
			r.Host, r.Version, r.RefusedAt.Format(time.RFC3339), strings.TrimPrefix(r.Reason, "not proven: "), r.Version, still)
		return subject, msg
	}
	if r.Recovered {
		subject = fmt.Sprintf("%s: %s did not start, rolled back to %s", r.Host, r.Version, r.RestoredVersion)
	} else {
		subject = fmt.Sprintf("%s: %s did not start and the rollback did not come up either", r.Host, r.Version)
	}
	if len(subject) > 100 { // SNS subject limit
		subject = subject[:100]
	}
	msg = fmt.Sprintf("%s refused %s at %s: %s", r.Host, r.Version, r.RefusedAt.Format(time.RFC3339), r.Reason)
	if !r.Recovered {
		msg += "\n\nThe rollback did not come up either: " + recoveryError(r)
	}
	msg += fmt.Sprintf("\n\nRetry with: carlos promote --retry %s <channel>", r.Version)
	return subject, msg
}

// recoveryError is why a refusal's recovery failed, never empty for one that
// did: a record written before the outcome was recorded has none to give.
func recoveryError(r release.Refusal) string {
	if r.RecoveryError != "" {
		return r.RecoveryError
	}
	return "outcome not recorded"
}

// parseAlertTopicARN checks CARLOS_ALERT_SNS_TOPIC_ARN and returns the
// region the topic lives in (alerting.ParseTopicARN).
func parseAlertTopicARN(s string) (region string, err error) { return alerting.ParseTopicARN(s) }

// alertDeliveryProbe is the box's answer to "do my alert rules have anywhere
// to go". The webhook and the ARN's shape come from the same env
// buildAlertSinks reads; whether the SNS client was actually built comes from
// counts, which buildAlertSinks fills in. So the snapshot reports the sink the
// agent really holds: a topic ARN that does not parse is not a sink, and one
// whose client failed to build reads SNS=false, SNSUnavailable=true.
func alertDeliveryProbe(getenv func(string) string, counts *alertCounters) *status.AlertDelivery {
	if counts == nil {
		counts = &processAlertCounters
	}
	out := &status.AlertDelivery{
		Webhook:         strings.TrimSpace(getenv("CARLOS_ALERT_WEBHOOK")) != "",
		WebhookFailures: counts.webhook.Load(),
		SNSFailures:     counts.sns.Load(),
	}
	if _, err := parseAlertTopicARN(getenv("CARLOS_ALERT_SNS_TOPIC_ARN")); err == nil {
		// Only an initialised client is a sink. snsInitUnknown (not built
		// yet, or the metrics loop never got there) is neither available
		// nor unavailable: the box has no working SNS sink to report.
		switch counts.snsInit.Load() {
		case snsInitOK:
			out.SNS = true
		case snsInitFailed:
			out.SNSUnavailable = true
		}
	}
	return out
}

// snsClientFactory builds the publisher for a region. It is a parameter so
// tests can make client construction fail without an AWS config.
type snsClientFactory func(ctx context.Context, region string) (snsPublisher, error)

// newSNSClient is the production factory: the default credential chain (the
// box's instance role), aimed at the topic's own region.
func newSNSClient(ctx context.Context, region string) (snsPublisher, error) {
	return alerting.NewSNSClient(ctx, region)
}

// buildAlertSinks reads the sink env and builds the sinks. A bad topic ARN
// is logged once and leaves SNS off; a client that cannot be built does the
// same, and is also recorded in counts (state and one counted failure) so the
// status snapshot reports the sink the agent really has. Neither stops the
// metrics loop. With nothing configured it says so in the journal, because
// that is the state that used to be silent.
func buildAlertSinks(ctx context.Context, getenv func(string) string, counts *alertCounters, newClient snsClientFactory) *alertSinks {
	webhook := strings.TrimSpace(getenv("CARLOS_ALERT_WEBHOOK"))
	topic := strings.TrimSpace(getenv("CARLOS_ALERT_SNS_TOPIC_ARN"))

	var pub snsPublisher
	if topic != "" {
		region, err := parseAlertTopicARN(topic)
		if err != nil {
			log.Printf("agent: alerts: CARLOS_ALERT_SNS_TOPIC_ARN is not usable, SNS sink off: %v", err)
			topic = ""
		} else if p, err := newClient(ctx, region); err != nil {
			log.Printf("agent: alerts: build SNS client, SNS sink off: %v", err)
			counts.sns.Add(1)
			counts.snsInit.Store(snsInitFailed)
			topic = ""
		} else {
			counts.snsInit.Store(snsInitOK)
			pub = p
		}
	}

	switch {
	case webhook == "" && topic == "":
		log.Printf("agent: alerts: NO SINK CONFIGURED (set CARLOS_ALERT_SNS_TOPIC_ARN or CARLOS_ALERT_WEBHOOK); disk, memory and cpu alerts will be published to the bucket and delivered nowhere")
	case webhook != "" && topic != "":
		log.Printf("agent: alerts: delivering to webhook and SNS topic %s", topic)
	case topic != "":
		log.Printf("agent: alerts: delivering to SNS topic %s", topic)
	default:
		log.Printf("agent: alerts: delivering to webhook")
	}
	return newAlertSinks(webhook, httpPostWebhook, topic, pub, counts)
}

// formatAlert renders one transition as an email: a one-line subject and a
// short plain-text body. The inputs are the event, the rule levels and a
// time, and nothing else, so no URL, ARN or credential can reach the
// message. The subject is ASCII with no newline and under 100 characters,
// which is what SNS accepts.
func formatAlert(rules metrics.Rules, at time.Time, ev metrics.Event) (subject, body string) {
	subjectLine := alertSubject(ev)
	var b strings.Builder
	if ev.Firing {
		fmt.Fprintf(&b, "%s.\n\n", alertSentence(ev))
	} else {
		fmt.Fprintf(&b, "%s.\n\n", alertClearSentence(ev))
	}
	if level, ok := alertLevel(rules, ev.Rule); ok {
		fmt.Fprintf(&b, "the alert level is %s.\n", level)
	}
	if ev.Firing {
		if d := alertSustain(rules, ev.Rule); d > 0 {
			fmt.Fprintf(&b, "it stayed above that level for %d minutes before this alert was raised.\n", int(d.Minutes()))
		}
	}
	fmt.Fprintf(&b, "this was reported at %s.\n", at.UTC().Format("2006-01-02 15:04 MST"))
	if ev.Firing {
		b.WriteString("you will get another email when it clears.\n")
	}
	fmt.Fprintf(&b, "\nrule: %s\nbox: %s\n", ev.Rule, ev.Box)
	return subjectLine, b.String()
}

func alertSubject(ev metrics.Event) string {
	if ev.Firing {
		return "carlos alert: " + alertSentence(ev)
	}
	return "carlos alert cleared: " + alertClearSentence(ev)
}

// alertSentence is the lower-case "what is wrong" clause, shared by the
// subject and the body.
func alertSentence(ev metrics.Event) string {
	switch ev.Rule {
	case "disk":
		return fmt.Sprintf("disk is %s full on %s", pct(ev.Value), ev.Box)
	case "mem":
		return fmt.Sprintf("memory use is %s on %s", pct(ev.Value), ev.Box)
	case "cpu":
		return fmt.Sprintf("cpu use is %s on %s", pct(ev.Value), ev.Box)
	case "lag":
		return fmt.Sprintf("mail relay lag is %d seconds on %s", int64(ev.Value), ev.Box)
	case "geoip":
		return geoipAlertSentence(ev)
	default:
		return fmt.Sprintf("%s is firing on %s", ev.Rule, ev.Box)
	}
}

func alertClearSentence(ev metrics.Event) string {
	switch ev.Rule {
	case "disk":
		return fmt.Sprintf("disk is back to %s full on %s", pct(ev.Value), ev.Box)
	case "mem":
		return fmt.Sprintf("memory use is back to %s on %s", pct(ev.Value), ev.Box)
	case "cpu":
		return fmt.Sprintf("cpu use is back to %s on %s", pct(ev.Value), ev.Box)
	case "lag":
		return fmt.Sprintf("mail relay lag is back to %d seconds on %s", int64(ev.Value), ev.Box)
	case "geoip":
		return fmt.Sprintf("geoip on %s is back to normal", ev.Box)
	default:
		return fmt.Sprintf("%s has cleared on %s", ev.Rule, ev.Box)
	}
}

func alertLevel(r metrics.Rules, rule string) (string, bool) {
	switch rule {
	case "disk":
		return pct(r.DiskFrac), true
	case "mem":
		return pct(r.MemFrac), true
	case "cpu":
		return pct(r.CPUFrac), true
	case "lag":
		return fmt.Sprintf("%d seconds", r.LagSecs), true
	}
	return "", false
}

func alertSustain(r metrics.Rules, rule string) time.Duration {
	switch rule {
	case "mem":
		return r.MemSustain
	case "cpu":
		return r.CPUSustain
	}
	return 0
}

// geoipAlertSentence reads the geoip rule's reason code (geoip spec §10.4).
func geoipAlertSentence(ev metrics.Event) string {
	switch int(ev.Value) {
	case 1:
		return fmt.Sprintf("the geoip database on %s is more than 45 days old", ev.Box)
	case 2:
		return fmt.Sprintf("the GeoLite2 database on %s expires within 9 days", ev.Box)
	case 3:
		return fmt.Sprintf("the geoip database on %s has expired and was removed", ev.Box)
	case 4:
		return fmt.Sprintf("%s has not loaded the published geoip database for over an hour", ev.Box)
	case 5:
		return fmt.Sprintf("the geoip feed has failed 7 times in a row (seen from %s)", ev.Box)
	}
	return fmt.Sprintf("geoip needs attention on %s", ev.Box)
}

// pct renders a 0..1 fraction as a whole percent.
func pct(f float64) string { return fmt.Sprintf("%d%%", int(math.Round(f*100))) }