package mysql

import (
	"context"
	"crypto/subtle"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"slices"
	"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"
	platform "git.yuntuo.hk/k2safe/orbit/platform/model"
)

// materializeResults 持有订阅锁，流和日志只做已提交读，不反向锁生产者。
// 先保存独立交付记录再推进扫描序号；未确认窗口满时停止扫描，保留来源日志。
func materializeResults(ctx context.Context, tx *sql.Tx, ns core.ID, sub *events.Subscription) error {
	if sub.PendingCount >= sub.MaxPending {
		return nil
	}
	var head, retained uint64
	if err := tx.QueryRowContext(ctx, `SELECT last_sequence,retained_from_sequence FROM orbit_event_streams WHERE namespace_id=?`, ns).Scan(&head, &retained); err != nil {
		return err
	}
	if sub.CursorSequence < retained-1 {
		return events.HistoryExpired
	}
	limit := min(uint32(100), sub.MaxPending-sub.PendingCount)
	rows, err := tx.QueryContext(ctx, `SELECT r.id,r.sequence,e.type 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<=? ORDER BY r.sequence LIMIT ?`, ns, sub.QueueID, sub.CursorSequence, head, limit)
	if err != nil {
		return err
	}
	type candidate struct {
		id        core.ID
		sequence  uint64
		eventType string
	}
	batch := make([]candidate, 0, limit)
	for rows.Next() {
		var c candidate
		if err = rows.Scan(&c.id, &c.sequence, &c.eventType); err != nil {
			rows.Close()
			return err
		}
		batch = append(batch, c)
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return err
	}
	cursor := sub.CursorSequence
	pending := sub.PendingCount
	for _, c := range batch {
		cursor = c.sequence
		if !slices.Contains(sub.Types, c.eventType) {
			continue
		}
		id, err := core.NewID()
		if err != nil {
			return err
		}
		if _, err = tx.ExecContext(ctx, `INSERT INTO orbit_deliveries(namespace_id,id,subscription_id,event_id,available_at,attempt_budget) VALUES (?,?,?,?,UTC_TIMESTAMP(6),?)`, ns, id, sub.ID, c.id, sub.MaxAttempts); err != nil {
			return err
		}
		pending++
	}
	if len(batch) < int(limit) {
		cursor = head
	}
	if cursor == sub.CursorSequence && pending == sub.PendingCount {
		return nil
	}
	if _, err = tx.ExecContext(ctx, `UPDATE orbit_subscriptions SET cursor_sequence=?,pending_count=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, cursor, pending, ns, sub.ID); err != nil {
		return err
	}
	sub.CursorSequence, sub.PendingCount = cursor, pending
	return nil
}

// recoverDeliveries 在新领取前有界回收过期租约。耗尽预算仍占积压，禁止静默丢弃。
// 最后一次证明保留；迟到确认只在还没有新交付代次时有效。
func recoverDeliveries(ctx context.Context, tx *sql.Tx, ns, sub core.ID, now time.Time) error {
	_, err := tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state=IF(attempts>=attempt_budget,'EXPIRED','PENDING'),available_at=TIMESTAMPADD(SECOND,LEAST(30,POW(2,LEAST(attempts-1,5))),?),lease_owner=NULL,lease_until=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND subscription_id=? AND state='LEASED' AND lease_until<=?`, now, now, ns, sub, now)
	return err
}

type storedDelivery struct {
	view   events.Delivery
	digest []byte
}

func readDelivery(ctx context.Context, tx *sql.Tx, ns, sub, id core.ID, lock bool) (storedDelivery, error) {
	var out storedDelivery
	q := `SELECT id,subscription_id,event_id,revision,state,available_at,lease_epoch,lease_until,attempts,attempt_budget,acknowledged_at,token_digest,created_at FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND id=?`
	if lock {
		q += ` FOR UPDATE`
	}
	v := &out.view
	err := tx.QueryRowContext(ctx, q, ns, sub, id).Scan(&v.ID, &v.SubscriptionID, &v.EventID, &v.Revision, &v.State, &v.AvailableAt, &v.LeaseEpoch, &v.LeaseUntil, &v.Attempts, &v.AttemptBudget, &v.AcknowledgedAt, &out.digest, &v.CreatedAt)
	if errors.Is(err, sql.ErrNoRows) {
		return out, fault.NotFound
	}
	if err != nil {
		return out, err
	}
	err = tx.QueryRowContext(ctx, `SELECT sequence FROM orbit_result_events WHERE namespace_id=? AND id=?`, ns, v.EventID).Scan(&v.Sequence)
	return out, err
}
func readResultEvent(ctx context.Context, tx *sql.Tx, ns, id core.ID) (events.Event, error) {
	var out events.Event
	var details, snapshot []byte
	err := tx.QueryRowContext(ctx, `SELECT r.id,r.sequence,e.type,e.aggregate_kind,e.aggregate_id,e.aggregate_revision,e.occurred_at,e.payload,r.snapshot 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.id=?`, ns, id).Scan(&out.ID, &out.Sequence, &out.Type, &out.AggregateKind, &out.AggregateID, &out.AggregateRevision, &out.OccurredAt, &details, &snapshot)
	if err != nil {
		return out, err
	}
	out.Details = details
	err = json.Unmarshal(snapshot, &out.Snapshot)
	return out, err
}
func (s *EventStore) GetDelivery(ctx context.Context, t auth.Token, sub, id core.ID) (events.Delivery, error) {
	var out events.Delivery
	err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
		if _, err := readSubscription(ctx, tx, t, p, sub, false); err != nil {
			return err
		}
		d, err := readDelivery(ctx, tx, t.NamespaceID, sub, id, false)
		out = d.view
		return err
	})
	return out, err
}
func (s *EventStore) ClaimDelivery(ctx context.Context, t auth.Token, id core.ID, in events.ClaimInput) (*events.Grant, error) {
	var out *events.Grant
	err := s.transaction(ctx, t, "events:consume", func(tx *sql.Tx, p auth.Permissions) error {
		sub, err := readSubscription(ctx, tx, t, p, id, true)
		if err != nil {
			return err
		}
		if !sub.Enabled {
			return nil
		}
		now, err := dbNow(ctx, tx)
		if err != nil {
			return err
		}
		if err = recoverDeliveries(ctx, tx, t.NamespaceID, id, now); err != nil {
			return err
		}
		if err = materializeResults(ctx, tx, t.NamespaceID, &sub); err != nil {
			return err
		}
		// 物化记录的 available_at 来自写入时的数据库时钟；不能继续用物化前的时间领取。
		now, err = dbNow(ctx, tx)
		if err != nil {
			return err
		}
		var active uint32
		if err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND state='LEASED' AND lease_owner=?`, t.NamespaceID, id, in.ConnectionID).Scan(&active); err != nil {
			return err
		}
		if active >= in.Capacity {
			return nil
		}
		var deliveryID core.ID
		err = tx.QueryRowContext(ctx, `SELECT id FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND state='PENDING' AND available_at<=? ORDER BY available_at,id LIMIT 1 FOR UPDATE`, t.NamespaceID, id, now).Scan(&deliveryID)
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		}
		if err != nil {
			return err
		}
		stored, err := readDelivery(ctx, tx, t.NamespaceID, id, deliveryID, false)
		if err != nil {
			return err
		}
		v := stored.view
		if v.Attempts >= v.AttemptBudget || v.LeaseEpoch == math.MaxUint64 || v.Revision == math.MaxUint64 {
			return fault.Conflict
		}
		secret, digest, err := events.NewSecret()
		if err != nil {
			return err
		}
		until := now.Add(events.LeaseDuration)
		if _, err = tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state='LEASED',lease_owner=?,lease_epoch=lease_epoch+1,lease_until=?,token_digest=?,attempts=attempts+1,revision=revision+1,updated_at=? WHERE namespace_id=? AND subscription_id=? AND id=?`, in.ConnectionID, until, digest, now, t.NamespaceID, id, deliveryID); err != nil {
			return err
		}
		v.State = platform.DeliveryLeased
		v.LeaseEpoch++
		v.LeaseUntil = &until
		v.Attempts++
		v.Revision++
		event, err := readResultEvent(ctx, tx, t.NamespaceID, v.EventID)
		if err != nil {
			return err
		}
		operation, err := core.NewID()
		if err != nil {
			return err
		}
		if err = appendAudit(ctx, tx, t, operation, v.ID, "delivery", "delivery.lease", "grant durable result delivery", nil, map[string]any{"event_id": v.EventID, "epoch": fmt.Sprint(v.LeaseEpoch), "attempts": v.Attempts, "lease_until": until, "connection_id": in.ConnectionID}); err != nil {
			return err
		}
		out = &events.Grant{Delivery: v, Event: event, LeaseSecret: secret}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return out, nil
}
func (s *EventStore) AckDelivery(ctx context.Context, t auth.Token, id core.ID, in events.AckInput) (events.AckResult, error) {
	var out events.AckResult
	digest, err := events.SecretDigest(in.LeaseSecret)
	if err != nil {
		return out, err
	}
	err = s.transaction(ctx, t, "events:consume", func(tx *sql.Tx, p auth.Permissions) error {
		sub, err := readSubscription(ctx, tx, t, p, id, true)
		if err != nil {
			return err
		}
		d, err := readDelivery(ctx, tx, t.NamespaceID, id, in.DeliveryID, true)
		if err != nil {
			return err
		}
		v := d.view
		if v.LeaseEpoch != in.Epoch || subtle.ConstantTimeCompare(d.digest, digest[:]) != 1 {
			return events.LeaseLost
		}
		out = events.AckResult{DeliveryID: v.ID, EventID: v.EventID, Epoch: in.Epoch}
		if v.State == platform.DeliveryAcked {
			if v.AcknowledgedAt == nil {
				return errors.New("acked delivery lacks confirmation timestamp")
			}
			out.AcknowledgedAt = *v.AcknowledgedAt
			out.Deduplicated = true
			return nil
		}
		if sub.PendingCount == 0 || v.Revision == math.MaxUint64 {
			return errors.New("delivery confirmation invariant failed")
		}
		now, err := dbNow(ctx, tx)
		if err != nil {
			return err
		}
		if _, err = tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state='ACKED',lease_owner=NULL,lease_until=NULL,acknowledged_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND subscription_id=? AND id=?`, now, now, t.NamespaceID, id, v.ID); err != nil {
			return err
		}
		if _, err = tx.ExecContext(ctx, `UPDATE orbit_subscriptions SET pending_count=pending_count-1,updated_at=? WHERE namespace_id=? AND id=?`, now, t.NamespaceID, id); err != nil {
			return err
		}
		operation, err := core.NewID()
		if err != nil {
			return err
		}
		if err = appendAudit(ctx, tx, t, operation, v.ID, "delivery", "delivery.ack", "consumer confirmed durable processing", map[string]any{"state": v.State}, map[string]any{"state": platform.DeliveryAcked, "event_id": v.EventID, "epoch": fmt.Sprint(in.Epoch)}); err != nil {
			return err
		}
		out.AcknowledgedAt = now
		return nil
	})
	return out, err
}
func (s *EventStore) RetryDelivery(ctx context.Context, t auth.Token, sub, id core.ID, key string, in events.RetryInput) (events.RetryResult, error) {
	var out events.RetryResult
	err := s.transaction(ctx, t, "events:manage", func(tx *sql.Tx, p auth.Permissions) error {
		identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "delivery.retry", eventScope(t.CredentialID, sub, id), key, "delivery", id, in)
		if err != nil {
			return err
		}
		if _, err = readSubscription(ctx, tx, t, p, sub, true); err != nil {
			return err
		}
		if cached != nil {
			if err = json.Unmarshal(cached, &out); err != nil {
				return err
			}
			out.Deduplicated = true
			return nil
		}
		d, err := readDelivery(ctx, tx, t.NamespaceID, sub, id, true)
		if err != nil {
			return err
		}
		v := d.view
		if v.State != platform.DeliveryExpired || v.Revision != in.ExpectedRevision || v.Revision == math.MaxUint64 || uint64(v.AttemptBudget)+uint64(in.AdditionalAttempts) > 1000000 {
			return fault.Conflict
		}
		if _, err = tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state='PENDING',attempt_budget=attempt_budget+?,available_at=UTC_TIMESTAMP(6),revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND subscription_id=? AND id=?`, in.AdditionalAttempts, t.NamespaceID, sub, id); err != nil {
			return err
		}
		d, err = readDelivery(ctx, tx, t.NamespaceID, sub, id, false)
		if err != nil {
			return err
		}
		out.Delivery = d.view
		if err = appendAudit(ctx, tx, t, identity, id, "delivery", "delivery.retry", in.Reason, v, out.Delivery); err != nil {
			return err
		}
		return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
	})
	return out, err
}