package mysql

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"math"
	"time"

	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/platform/auth"
	"git.yuntuo.hk/k2safe/orbit/platform/events"
	"git.yuntuo.hk/k2safe/orbit/pulse/service"
)

// requireActiveQueue 对创建/恢复工作取得共享锁，与退役排他锁串行；历史读取不使用此检查。
func requireActiveQueue(ctx context.Context, tx *sql.Tx, ns, id core.ID) error {
	var retired *time.Time
	err := tx.QueryRowContext(ctx, `SELECT retired_at FROM pulse_queues WHERE namespace_id=? AND id=? FOR SHARE`, ns, id).Scan(&retired)
	if errors.Is(err, sql.ErrNoRows) {
		return fault.NotFound
	}
	if err != nil {
		return err
	}
	if retired != nil {
		return fault.Retired
	}
	return nil
}
func retirementBusy(message string) error { return fault.New("RESOURCE_BUSY", message) }

// RetireSubscription 锁定提交水位后锁订阅，既不丢未物化结果，也不丢过期交付。
// 与创建共用流锁串行化活动名额；消费只锁订阅，不反向锁流/队列。
func (s *EventStore) RetireSubscription(ctx context.Context, t auth.Token, id core.ID, key string, in events.RetireInput) (events.SubscriptionResult, error) {
	var out events.SubscriptionResult
	err := s.transaction(ctx, t, "events:manage", func(tx *sql.Tx, p auth.Permissions) error {
		if _, err := readSubscription(ctx, tx, t, p, id, false); err != nil {
			return err
		}
		identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "subscription.retire", eventScope(t.CredentialID, id), key, "subscription", id, in)
		if err != nil {
			return err
		}
		if cached != nil {
			if err = json.Unmarshal(cached, &out); err != nil {
				return err
			}
			out.Deduplicated = true
			return nil
		}
		head, retained, err := lockEventStream(ctx, tx, t.NamespaceID)
		if err != nil {
			return err
		}
		before, err := readSubscription(ctx, tx, t, p, id, true)
		if err != nil {
			return err
		}
		if before.RetiredAt != nil {
			return fault.Retired
		}
		if before.Revision != in.ExpectedRevision || before.Revision == math.MaxUint64 {
			return fault.Conflict
		}
		if before.PendingCount != 0 {
			return retirementBusy("subscription has unacknowledged deliveries; drain or resolve expired deliveries first")
		}
		var pending bool
		if err = tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND state IN ('PENDING','LEASED','EXPIRED'))`, t.NamespaceID, id).Scan(&pending); err != nil {
			return err
		}
		if pending {
			return retirementBusy("subscription has unacknowledged deliveries")
		}
		if before.CursorSequence < retained-1 {
			return events.HistoryExpired
		}
		filter, err := json.Marshal(before.Types)
		if err != nil {
			return err
		}
		// 缺失来源事件也拒绝退役，不能将损坏记录误当成不匹配过滤器。
		if err = tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM orbit_result_events r LEFT JOIN orbit_events e ON e.namespace_id=r.namespace_id AND e.id=r.id WHERE r.namespace_id=? AND r.queue_id=? AND r.sequence>? AND r.sequence<=? AND (e.id IS NULL OR JSON_CONTAINS(?,JSON_QUOTE(e.type))))`, t.NamespaceID, before.QueueID, before.CursorSequence, head, string(filter)).Scan(&pending); err != nil {
			return err
		}
		if pending {
			return retirementBusy("subscription has unconsumed results; resume consumption and acknowledge them before retirement")
		}
		if _, err = tx.ExecContext(ctx, `UPDATE orbit_subscriptions SET enabled=FALSE,retired_at=UTC_TIMESTAMP(6),retired_sequence=?,cursor_sequence=?,revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, head, head, t.NamespaceID, id); err != nil {
			return err
		}
		out.Subscription, err = readSubscription(ctx, tx, t, p, id, false)
		if err != nil {
			return err
		}
		if err = appendAudit(ctx, tx, t, identity, id, "subscription", "subscription.retire", in.Reason, before, out.Subscription); err != nil {
			return err
		}
		return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
	})
	return out, err
}

// RetireQueue 按应用额度→身份→队列持锁，拒绝尚未结清的工作和未来工作来源。
// 暂停计划可以保留；其恢复/发布与本操作共用队列锁，退役后无法重新激活。
func (s *TaskStore) RetireQueue(ctx context.Context, t auth.Token, id core.ID, key string, in service.RetireQueueInput) (service.QueueResult, error) {
	var out service.QueueResult
	err := s.transaction(ctx, t, "queues:manage", func(tx *sql.Tx, p auth.Permissions) error {
		if !p.AllowsQueue(id) {
			return fault.NotFound
		}
		if _, err := lockApplicationResources(ctx, tx, t.NamespaceID, false); err != nil {
			return err
		}
		identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "queue.retire", string(id), key, "queue", id, in)
		if err != nil {
			return err
		}
		if cached != nil {
			if err = json.Unmarshal(cached, &out); err != nil {
				return err
			}
			out.Deduplicated = true
			return nil
		}
		if _, _, err = lockQueueAfterApplication(ctx, tx, t.NamespaceID, id); err != nil {
			return err
		}
		before, err := readQueue(ctx, tx, t.NamespaceID, id)
		if err != nil {
			return err
		}
		if before.Revision != in.ExpectedRevision || before.Revision == math.MaxUint64 {
			return fault.Conflict
		}
		if before.UnfinishedTasks != 0 {
			return retirementBusy("queue has unfinished tasks, including tasks needing review")
		}
		checks := []struct {
			query, message string
			args           []any
		}{
			{`SELECT EXISTS(SELECT 1 FROM pulse_tasks WHERE namespace_id=? AND queue_id=? AND state IN ('WAITING','LEASED','RUNNING','NEEDS_REVIEW'))`, "queue has unfinished or uncertain tasks", []any{t.NamespaceID, id}},
			{`SELECT EXISTS(SELECT 1 FROM pulse_attempts a JOIN pulse_tasks t ON t.namespace_id=a.namespace_id AND t.id=a.task_id WHERE t.namespace_id=? AND t.queue_id=? AND a.state IN ('LEASED','RUNNING'))`, "queue has active execution attempts", []any{t.NamespaceID, id}},
			{`SELECT EXISTS(SELECT 1 FROM pulse_permits p JOIN pulse_attempts a ON a.namespace_id=p.namespace_id AND a.id=p.attempt_id JOIN pulse_tasks t ON t.namespace_id=a.namespace_id AND t.id=a.task_id WHERE t.namespace_id=? AND t.queue_id=? AND p.released_at IS NULL)`, "queue has unreleased execution permits", []any{t.NamespaceID, id}},
			{`SELECT EXISTS(SELECT 1 FROM pulse_schedules WHERE namespace_id=? AND queue_id=? AND paused=FALSE)`, "queue has enabled schedules; pause them first", []any{t.NamespaceID, id}},
			{`SELECT EXISTS(SELECT 1 FROM weave_instances i LEFT JOIN weave_version_scopes v ON v.namespace_id=i.namespace_id AND v.definition_version_id=i.definition_version_id WHERE i.namespace_id=? AND i.state IN ('RUNNING','WAITING','CANCELLING','NEEDS_REVIEW') AND (v.definition_version_id IS NULL OR JSON_CONTAINS(v.queue_ids,JSON_QUOTE(?))))`, "queue is referenced by unfinished workflows", []any{t.NamespaceID, string(id)}},
			{`SELECT EXISTS(SELECT 1 FROM orbit_subscriptions WHERE namespace_id=? AND queue_id=? AND retired_at IS NULL)`, "queue has subscriptions; retire them first", []any{t.NamespaceID, id}},
		}
		for _, check := range checks {
			var busy bool
			if err = tx.QueryRowContext(ctx, check.query, check.args...).Scan(&busy); err != nil {
				return err
			}
			if busy {
				return retirementBusy(check.message)
			}
		}
		if _, err = tx.ExecContext(ctx, `UPDATE pulse_queues SET paused=TRUE,retired_at=UTC_TIMESTAMP(6),revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, t.NamespaceID, id); err != nil {
			return err
		}
		out.Queue, err = readQueue(ctx, tx, t.NamespaceID, id)
		if err != nil {
			return err
		}
		if err = appendAudit(ctx, tx, t, identity, id, "queue", "queue.retire", in.Reason, before, out.Queue); err != nil {
			return err
		}
		return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
	})
	return out, err
}