package delivery

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/jmoiron/sqlx"

	"github.com/tidbcloud/alertstore/pkg/textutil"
)

// Task kinds. im_topic fans out the rest when it executes.
const (
	KindIMTopic         = "im_topic"
	KindIMForward       = "im_forward"
	KindIMCard          = "im_card"
	KindIMResolvedReply = "im_resolved_reply"
	KindIMFiringReply   = "im_firing_reply" // an occurrence joined an alert that was already posted
	KindIMCardRefresh   = "im_card_refresh" // rewrite the posted card from current state
	KindIMOwner         = "im_owner"        // record who owns the alert, and every later change
	KindIMOwnerNotify   = "im_owner_notify" // reach that owner personally
	KindIMComment       = "im_comment"      // mirror a web/API comment into the topic
	KindIMNotice        = "im_notice"       // announce one stored observation, per channel
	KindEscalation      = "escalation"      // phone-call escalation for page-class alerts
	KindAckReminder     = "ack_reminder"    // nag the acker of an alert that stays unresolved

	// KindAgentDispatch hands one alert to one agent over HTTP. It is in the
	// outbox for the retry, not for the ordering: an agent that is down must not
	// cost the dispatch, and must equally not delay the topic or the page, which
	// are enqueued as its siblings and executed independently.
	KindAgentDispatch = "agent_dispatch"

	// KindAgentVerdictReply replies a finished run's verdict into the thread.
	// Dedup key is the run ID, so a verdict reported twice replies once.
	KindAgentVerdictReply = "agent_verdict_reply"

	// Jira mirror (transitional). These three kinds and pkg/jira are the whole
	// integration: deleting them together removes it.
	KindJiraCreate     = "jira_create"     // create the compatibility issue
	KindJiraComment    = "jira_comment"    // mirror an occurrence change as a comment
	KindJiraTransition = "jira_transition" // move the issue to match alertstore state
)

const (
	StatusPending = "pending"
	StatusDone    = "done"
	StatusFailed  = "failed"
)

// NullableJSON wraps json.RawMessage to handle SQL NULL values.
type NullableJSON json.RawMessage

func (n *NullableJSON) Scan(value interface{}) error {
	if value == nil {
		*n = nil
		return nil
	}
	switch v := value.(type) {
	case []byte:
		// Copy: the driver owns v and reuses it for the next row, so aliasing
		// it corrupts this payload once a batch read (ListDue) advances.
		*n = append((*n)[:0], v...)
	case string:
		*n = NullableJSON(v)
	default:
		return fmt.Errorf("NullableJSON.Scan: unsupported type %T", value)
	}
	return nil
}

// Raw returns the underlying json.RawMessage (nil-safe).
func (n NullableJSON) Raw() json.RawMessage {
	if n == nil {
		return nil
	}
	return json.RawMessage(n)
}

// Task is one unit of outbound delivery work, persisted so it survives
// crashes and is retried until it succeeds or exhausts attempts.
type Task struct {
	ID            int64        `db:"id"`
	AlertID       string       `db:"alert_id"`
	Kind          string       `db:"kind"`
	DedupKey      string       `db:"dedup_key"`
	Payload       NullableJSON `db:"payload"`
	Status        string       `db:"status"`
	Attempts      int          `db:"attempts"`
	NextAttemptAt time.Time    `db:"next_attempt_at"`
	LastError     string       `db:"last_error"`
	CreatedAt     time.Time    `db:"created_at"`
	UpdatedAt     time.Time    `db:"updated_at"`
}

type forwardPayload struct {
	ThreadID  string `json:"thread_id"`
	ChannelID string `json:"channel_id"`
}

type cardPayload struct {
	ChannelID string `json:"channel_id"`
}

type commentPayload struct {
	Actor string `json:"actor"`
	Text  string `json:"text"`
}

// ownerPayload drives one owner delivery. An empty Assignee means "whoever the
// on-call schedule says", which is the alert's initial assignment; a set one is
// a reassignment already committed by the caller. Actor is empty for the former
// and names the human (or Jira) behind the latter.
type ownerPayload struct {
	Assignee string `json:"assignee,omitempty"`
	Actor    string `json:"actor,omitempty"`
	Reason   string `json:"reason,omitempty"`
}

// ownerNotifyPayload carries an owner already resolved by im_owner, so reaching
// them personally neither repeats the schedule lookup nor can disagree with the
// announcement the group saw.
type ownerNotifyPayload struct {
	Assignee string `json:"assignee"`
	UserID   string `json:"user_id"`
}

type instanceResolvedPayload struct {
	InstanceID      int64  `json:"instance_id"`
	DisplayName     string `json:"display_name"`
	Fingerprint     string `json:"fingerprint"`
	RemainingFiring int    `json:"remaining_firing"`
}

// jiraCommentPayload carries text already rendered by whoever enqueued it, so
// the comment a retry writes cannot drift from the one the topic announced.
type jiraCommentPayload struct {
	Text string `json:"text"`
}

// jiraTransitionPayload names the alertstore state the issue should be moved to
// match. The transition name itself comes from config at execution time, so
// renaming a workflow step does not strand tasks already in the outbox.
type jiraTransitionPayload struct {
	State string `json:"state"`
}

// noticePayload is one stored observation, rendered at enqueue and addressed
// to one channel.
//
// It carries its content rather than a reference to the occurrence because a
// notice reports an event, not a state: it says this was observed, at this
// time — and for a customer-facing one, that this is what the customer was
// told. Re-reading the occurrence when the task runs would let a retry minutes
// later announce a condition that has since changed, describing a notification
// that never happened. Same reasoning as jiraCommentPayload.
type noticePayload struct {
	ChannelID string `json:"channel_id"`
	// Mode is the subscription mode that asked for this — event or customer.
	// It picks the rendering, so a channel subscribed to both can tell at a
	// glance which stream a card came from.
	Mode    string `json:"mode"`
	EventID int64  `json:"event_id"`
	Status  string `json:"status"` // firing / resolved
	// ObservationType is first_firing / repeat_firing / resolved. It is the
	// reason this stream exists: an alert shows neither how often an occurrence
	// has re-fired nor that it re-fired at all.
	ObservationType string `json:"observation_type,omitempty"`
	Classification  string `json:"classification,omitempty"`
	// AlertID names the alert this observation landed on, empty when it opened
	// none. It is what keeps the card honest about whether anyone was paged.
	AlertID string `json:"alert_id,omitempty"`
	// MuteRuleID and MuteReason explain the other way an observation opens no
	// alert. Carrying them makes this stream the only place an active mute rule
	// is visible while it is suppressing something.
	MuteRuleID  string    `json:"mute_rule_id,omitempty"`
	MuteReason  string    `json:"mute_reason,omitempty"`
	Alertname   string    `json:"alertname"`
	Severity    string    `json:"severity,omitempty"`
	ClusterID   string    `json:"cluster_id,omitempty"`
	ClusterName string    `json:"cluster_name,omitempty"`
	TenantID    string    `json:"tenant_id,omitempty"`
	Region      string    `json:"region,omitempty"`
	Subject     string    `json:"subject,omitempty"`
	Body        string    `json:"body,omitempty"`
	ClinicURL   string    `json:"clinic_url,omitempty"`
	NotifiedAt  time.Time `json:"notified_at"`
}

// escalationPayload tracks progress through the phone-call escalation
// timeline. Step indexes into the hardcoded sequence in escalation.go; the
// generation distinguishes a fresh escalation from a reset caused by reassign.
type escalationPayload struct {
	Step       int    `json:"step"`
	Generation int    `json:"generation"`
	OncallTeam string `json:"oncall_team"`
}

// agentDispatchPayload names the agent to reach and why. The endpoint is
// resolved at execution rather than carried here, so an agent that moves is
// reachable again on the next retry instead of stranding every task enqueued
// before it moved — the same reasoning that keeps the Jira transition name out
// of jiraTransitionPayload.
type agentDispatchPayload struct {
	Agent       string `json:"agent"`
	TriggerKind string `json:"trigger_kind"`
	// ParentRunID is the run that asked for this one, for a handoff. Nil for
	// anything alertstore started itself.
	ParentRunID *int64 `json:"parent_run_id,omitempty"`
}

// agentVerdictPayload carries a finished run's finding onto the thread. The
// verdict is kept raw so its wording lives in one place (agentVerdictText).
type agentVerdictPayload struct {
	Verdict string `json:"verdict"`
	Summary string `json:"summary"`
}

// ackReminderPayload tracks progress through the ack-reminder ladder in
// ackreminder.go. Step indexes the hardcoded sequence; AckGen is the acking
// alert's AckedAt in unix-milli, which scopes a chain to one ack episode — a
// reassign unacks the alert and a later re-ack starts a fresh chain rather than
// colliding with this one's already-used dedup keys.
type ackReminderPayload struct {
	Step   int   `json:"step"`
	AckGen int64 `json:"ack_gen"`
}

type instanceFiringPayload struct {
	InstanceID  int64  `json:"instance_id"`
	DisplayName string `json:"display_name"`
	Fingerprint string `json:"fingerprint"`
	Detail      string `json:"detail"`
	TotalFiring int    `json:"total_firing"`
}

type TaskStore struct {
	db *sqlx.DB
}

func NewTaskStore(db *sqlx.DB) *TaskStore {
	return &TaskStore{db: db}
}

// enqueueTimeout bounds an outbox write after it is detached from the caller's
// context, so a task still records when the triggering webhook or card-action
// callback has already returned, without letting a stuck DB block forever.
const enqueueTimeout = 5 * time.Second

// Enqueue inserts a task. (alert_id, kind, dedup_key) is unique, so calling
// this repeatedly (webhook retries, re-firing events) is a no-op after the
// first insert — enqueues are idempotent.
func (s *TaskStore) Enqueue(ctx context.Context, alertID, kind, dedupKey string, payload interface{}) error {
	ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), enqueueTimeout)
	defer cancel()
	var payloadJSON interface{}
	if payload != nil {
		b, err := json.Marshal(payload)
		if err != nil {
			return fmt.Errorf("marshal task payload: %w", err)
		}
		payloadJSON = b
	}

	const q = `
		INSERT INTO delivery_task (alert_id, kind, dedup_key, payload)
		VALUES (?, ?, ?, ?)
		ON DUPLICATE KEY UPDATE id = id
	`
	if _, err := s.db.ExecContext(ctx, q, alertID, kind, dedupKey, payloadJSON); err != nil {
		return fmt.Errorf("enqueue delivery task %s/%s: %w", kind, alertID, err)
	}
	return nil
}

// EnqueueAt inserts a task scheduled for a future time. Same idempotency as
// Enqueue: (alert_id, kind, dedup_key) is unique.
func (s *TaskStore) EnqueueAt(ctx context.Context, alertID, kind, dedupKey string, payload interface{}, delay time.Duration) error {
	ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), enqueueTimeout)
	defer cancel()
	var payloadJSON interface{}
	if payload != nil {
		b, err := json.Marshal(payload)
		if err != nil {
			return fmt.Errorf("marshal task payload: %w", err)
		}
		payloadJSON = b
	}

	const q = `
		INSERT INTO delivery_task (alert_id, kind, dedup_key, payload, next_attempt_at)
		VALUES (?, ?, ?, ?, DATE_ADD(NOW(3), INTERVAL ? SECOND))
		ON DUPLICATE KEY UPDATE id = id
	`
	if _, err := s.db.ExecContext(ctx, q, alertID, kind, dedupKey, payloadJSON, int(delay.Seconds())); err != nil {
		return fmt.Errorf("enqueue-at delivery task %s/%s: %w", kind, alertID, err)
	}
	return nil
}

// CancelPending marks all pending tasks of the given kind for an alert as done.
func (s *TaskStore) CancelPending(ctx context.Context, alertID, kind string) (int64, error) {
	ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), enqueueTimeout)
	defer cancel()
	const q = `UPDATE delivery_task SET status = ? WHERE alert_id = ? AND kind = ? AND status = ?`
	res, err := s.db.ExecContext(ctx, q, StatusDone, alertID, kind, StatusPending)
	if err != nil {
		return 0, fmt.Errorf("cancel pending %s tasks for %s: %w", kind, alertID, err)
	}
	return res.RowsAffected()
}

// ListDue returns pending tasks whose next attempt time has passed.
func (s *TaskStore) ListDue(ctx context.Context, limit int) ([]Task, error) {
	const q = `
		SELECT * FROM delivery_task
		WHERE status = ? AND next_attempt_at <= NOW(3)
		ORDER BY next_attempt_at, id
		LIMIT ?
	`
	var tasks []Task
	if err := s.db.SelectContext(ctx, &tasks, q, StatusPending, limit); err != nil {
		return nil, fmt.Errorf("list due tasks: %w", err)
	}
	return tasks, nil
}

// Claim takes ownership of a task by bumping attempts and pushing
// next_attempt_at forward as a lease. Returns false if another worker got
// there first. If the worker crashes mid-execution, the lease expires and
// the task becomes due again.
func (s *TaskStore) Claim(ctx context.Context, id int64, lease time.Duration) (bool, error) {
	const q = `
		UPDATE delivery_task
		SET attempts = attempts + 1, next_attempt_at = DATE_ADD(NOW(3), INTERVAL ? SECOND)
		WHERE id = ? AND status = ? AND next_attempt_at <= NOW(3)
	`
	res, err := s.db.ExecContext(ctx, q, int(lease.Seconds()), id, StatusPending)
	if err != nil {
		return false, fmt.Errorf("claim task %d: %w", id, err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return n == 1, nil
}

func (s *TaskStore) MarkDone(ctx context.Context, id int64) error {
	const q = `UPDATE delivery_task SET status = ?, last_error = '' WHERE id = ?`
	_, err := s.db.ExecContext(ctx, q, StatusDone, id)
	return err
}

func (s *TaskStore) MarkRetry(ctx context.Context, id int64, lastError string, backoff time.Duration) error {
	const q = `
		UPDATE delivery_task
		SET next_attempt_at = DATE_ADD(NOW(3), INTERVAL ? SECOND), last_error = ?
		WHERE id = ? AND status = ?
	`
	_, err := s.db.ExecContext(ctx, q, int(backoff.Seconds()), truncateErr(lastError), id, StatusPending)
	return err
}

func (s *TaskStore) MarkFailed(ctx context.Context, id int64, lastError string) error {
	const q = `UPDATE delivery_task SET status = ?, last_error = ? WHERE id = ?`
	_, err := s.db.ExecContext(ctx, q, StatusFailed, truncateErr(lastError), id)
	return err
}

func (s *TaskStore) CountPending(ctx context.Context) (int, error) {
	var n int
	err := s.db.GetContext(ctx, &n, `SELECT COUNT(*) FROM delivery_task WHERE status = ?`, StatusPending)
	return n, err
}

func truncateErr(s string) string {
	return textutil.Truncate(s, 1000)
}