package mysql

import (
	"context"
	"crypto/sha256"
	"crypto/subtle"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

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

type executionTask struct {
	view           service.TaskView
	current        *core.ID
	epoch          uint64
	lease          *time.Time
	gate           *core.ID
	maxAttempts    uint32
	attemptTimeout uint64
}
type executionRows struct {
	task    executionTask
	attempt pulse.Attempt
	paused  bool
	session sessionState
}

func readExecutionTask(ctx context.Context, tx *sql.Tx, ns, id core.ID) (executionTask, error) {
	var row executionTask
	view, err := readTask(ctx, tx, ns, id, true)
	if err != nil {
		return row, err
	}
	row.view = view
	err = tx.QueryRowContext(ctx, `SELECT current_attempt_id,lease_epoch,lease_until,gate_id,max_attempts,attempt_timeout_ms FROM pulse_tasks WHERE namespace_id=? AND id=?`, ns, id).Scan(&row.current, &row.epoch, &row.lease, &row.gate, &row.maxAttempts, &row.attemptTimeout)
	return row, err
}
func lockQueue(ctx context.Context, tx *sql.Tx, ns, id core.ID) (paused bool, max uint32, err error) {
	if _, err = lockApplicationResources(ctx, tx, ns, false); err != nil {
		return
	}
	err = tx.QueryRowContext(ctx, "SELECT paused,max_concurrency FROM pulse_queues WHERE namespace_id=? AND id=? FOR UPDATE", ns, id).Scan(&paused, &max)
	if errors.Is(err, sql.ErrNoRows) {
		err = fault.NotFound
	}
	return
}
func lockGate(ctx context.Context, tx *sql.Tx, ns core.ID, id *core.ID) (bool, error) {
	if id == nil {
		return false, nil
	}
	var paused bool
	err := tx.QueryRowContext(ctx, "SELECT paused FROM pulse_execution_gates WHERE namespace_id=? AND id=? FOR SHARE", ns, *id).Scan(&paused)
	return paused, err
}
func minimumTime(first time.Time, others ...time.Time) time.Time {
	for _, other := range others {
		if other.Before(first) {
			first = other
		}
	}
	return first
}
func queueLimit(ctx context.Context, tx *sql.Tx, ns, queue core.ID, max uint32) (core.ID, uint32, error) {
	id, err := core.NewID()
	if err != nil {
		return "", 0, err
	}
	digest := core.Digest(sha256.Sum256([]byte(queue)))
	_, err = tx.ExecContext(ctx, `INSERT INTO pulse_concurrency_limits(namespace_id,id,scope_key,key_digest,max_permits) VALUES (?,?,'queue',?,?) ON DUPLICATE KEY UPDATE id=pulse_concurrency_limits.id`, ns, id, digest, max)
	if err != nil {
		return "", 0, err
	}
	var used uint32
	var configured uint32
	err = tx.QueryRowContext(ctx, `SELECT id,used_permits,max_permits FROM pulse_concurrency_limits WHERE namespace_id=? AND scope_key='queue' AND key_digest=? FOR UPDATE`, ns, digest).Scan(&id, &used, &configured)
	// 降额时保留已有许可；队列上限控制新的领取，存储上限至少容纳在途数。
	target := max
	if used > target {
		target = used
	}
	if err == nil && configured != target {
		_, err = tx.ExecContext(ctx, `UPDATE pulse_concurrency_limits SET max_permits=?,revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, target, ns, id)
	}
	return id, used, err
}
func outstanding(ctx context.Context, tx *sql.Tx, ns, instance core.ID) (uint32, error) {
	var count uint32
	err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_attempts a JOIN pulse_worker_sessions s ON s.namespace_id=a.namespace_id AND s.id=a.session_id WHERE a.namespace_id=? AND s.worker_instance_id=? AND a.state IN ('LEASED','RUNNING')`, ns, instance).Scan(&count)
	return count, err
}

func (s *ExecutionStore) Claim(ctx context.Context, token auth.Token, b worker.Binding) (*worker.Assignment, error) {
	var assignment *worker.Assignment
	err := s.transaction(ctx, token, "workers:consume", func(tx *sql.Tx, p auth.Permissions) error {
		state, err := s.lockSession(ctx, tx, token, b, p)
		if err != nil {
			return err
		}
		if state.draining || state.capacity == 0 {
			return nil
		}
		count, err := outstanding(ctx, tx, token.NamespaceID, b.InstanceID)
		if err != nil {
			return err
		}
		if count >= state.capacity {
			return nil
		}
		paused, max, err := lockQueue(ctx, tx, token.NamespaceID, state.queue)
		if err != nil {
			return err
		}
		if paused {
			return nil
		}
		limit, used, err := queueLimit(ctx, tx, token.NamespaceID, state.queue, max)
		if err != nil {
			return err
		}
		if used >= max {
			return nil
		}
		now, err := dbNow(ctx, tx)
		if err != nil {
			return err
		}
		canClaim, err := queueCanClaim(ctx, tx, token.NamespaceID, state.queue, now)
		if err != nil || !canClaim {
			return err
		}
		args := []any{token.NamespaceID, state.queue}
		matches := make([]string, 0, len(state.capabilities))
		for _, c := range state.capabilities {
			matches = append(matches, "(handler=? AND handler_version=? AND payload_version=?)")
			args = append(args, c.Handler, c.HandlerVersion, c.PayloadVersion)
		}
		if len(matches) == 0 {
			return nil
		}
		keyFilter := keyEligibility
		if !state.concurrencyKeys {
			keyFilter = "concurrency_key_digest IS NULL"
		}
		rows, err := tx.QueryContext(ctx, `SELECT id,gate_id FROM pulse_tasks WHERE namespace_id=? AND queue_id=? AND state='WAITING' AND available_at<=UTC_TIMESTAMP(6) AND deadline_at>UTC_TIMESTAMP(6) AND (start_deadline_at IS NULL OR start_deadline_at>UTC_TIMESTAMP(6)) AND attempt_count<max_attempts AND (gate_id IS NULL OR EXISTS (SELECT 1 FROM pulse_execution_gates g WHERE g.namespace_id=pulse_tasks.namespace_id AND g.id=pulse_tasks.gate_id AND g.paused=FALSE)) AND (`+strings.Join(matches, " OR ")+`) AND `+keyFilter+` ORDER BY dispatch_at,available_at,id LIMIT 16`, args...)
		if err != nil {
			return err
		}
		type candidate struct {
			id   core.ID
			gate *core.ID
		}
		candidates := []candidate{}
		for rows.Next() {
			var c candidate
			if err := rows.Scan(&c.id, &c.gate); err != nil {
				rows.Close()
				return err
			}
			candidates = append(candidates, c)
		}
		err = rows.Err()
		rows.Close()
		if err != nil {
			return err
		}
		for _, candidate := range candidates {
			gated, err := lockGate(ctx, tx, token.NamespaceID, candidate.gate)
			if err != nil {
				return err
			}
			if gated {
				continue
			}
			var locked core.ID
			err = tx.QueryRowContext(ctx, `SELECT id FROM pulse_tasks WHERE namespace_id=? AND id=? FOR UPDATE SKIP LOCKED`, token.NamespaceID, candidate.id).Scan(&locked)
			if errors.Is(err, sql.ErrNoRows) {
				continue
			}
			if err != nil {
				return err
			}
			task, err := readExecutionTask(ctx, tx, token.NamespaceID, locked)
			if err != nil {
				return err
			}
			now, err := dbNow(ctx, tx)
			if err != nil {
				return err
			}
			if !state.expires.After(now) {
				return worker.StaleSession
			}
			if task.view.State != pulse.TaskWaiting || task.view.CancelRequestedAt != nil || task.view.AvailableAt.After(now) || !task.view.DeadlineAt.After(now) || task.view.AttemptCount >= task.maxAttempts || (task.view.StartDeadlineAt != nil && !task.view.StartDeadlineAt.After(now)) {
				continue
			}
			if task.epoch == ^uint64(0) {
				return fault.Conflict
			}
			keyLimitID, concurrencyToken, err := reserveConcurrencyKey(ctx, tx, token.NamespaceID, task.view.ConcurrencyKey)
			if err != nil {
				return err
			}
			if task.view.ConcurrencyKey != "" && concurrencyToken == 0 {
				continue
			}
			attempt, err := core.NewID()
			if err != nil {
				return err
			}
			secret, digest, err := worker.NewSecret()
			if err != nil {
				return err
			}
			until := minimumTime(now.Add(s.options.LeaseDuration), task.view.DeadlineAt, state.expires)
			if task.view.StartDeadlineAt != nil {
				until = minimumTime(until, *task.view.StartDeadlineAt)
			}
			_, err = tx.ExecContext(ctx, `INSERT INTO pulse_attempts(namespace_id,id,task_id,epoch,session_id,token_digest,lease_until,concurrency_token) VALUES (?,?,?,?,?,?,?,?)`, token.NamespaceID, attempt, locked, task.epoch+1, b.SessionID, digest, until, concurrencyToken)
			if err != nil {
				return err
			}
			permit, err := core.NewID()
			if err != nil {
				return err
			}
			_, err = tx.ExecContext(ctx, `INSERT INTO pulse_permits(namespace_id,id,limit_id,attempt_id,lease_until) VALUES (?,?,?,?,?)`, token.NamespaceID, permit, limit, attempt, until)
			if err != nil {
				return err
			}
			if keyLimitID != "" {
				id, err := core.NewID()
				if err != nil {
					return err
				}
				if _, err := tx.ExecContext(ctx, `INSERT INTO pulse_permits(namespace_id,id,limit_id,attempt_id,lease_until) VALUES (?,?,?,?,?)`, token.NamespaceID, id, keyLimitID, attempt, until); err != nil {
					return err
				}
			}
			changed, err := tx.ExecContext(ctx, `UPDATE pulse_concurrency_limits SET used_permits=used_permits+1,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND used_permits<max_permits`, now, token.NamespaceID, limit)
			if err := exactlyOne(changed, err); err != nil {
				return err
			}
			changed, err = tx.ExecContext(ctx, `UPDATE pulse_tasks SET state='LEASED',current_attempt_id=?,lease_epoch=lease_epoch+1,lease_until=?,attempt_count=attempt_count+1,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND state='WAITING' AND revision=?`, attempt, until, now, token.NamespaceID, locked, task.view.Revision)
			if err := exactlyOne(changed, err); err != nil {
				return err
			}
			if err := reserveApplicationAttempt(ctx, tx, token.NamespaceID); err != nil {
				return err
			}
			view, err := readTask(ctx, tx, token.NamespaceID, locked, false)
			if err != nil {
				return err
			}
			if err := emitEvent(ctx, tx, token.NamespaceID, locked, view.Revision, "task", "task.leased", map[string]any{"attempt_id": attempt, "epoch": fmt.Sprint(task.epoch + 1), "session_id": b.SessionID, "lease_until": until}); err != nil {
				return err
			}
			assignment = &worker.Assignment{ConcurrencyToken: concurrencyToken, Task: view, AttemptID: attempt, Epoch: task.epoch + 1, LeaseSecret: secret, LeaseUntil: until, ServerTime: now}
			return nil
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return assignment, nil
}
func exactlyOne(result sql.Result, err error) error {
	if err != nil {
		return err
	}
	n, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if n != 1 {
		return errors.New("execution invariant affected unexpected rows")
	}
	return nil
}

// lockExecution 的跨资源顺序为进程/会话 -> 队列 -> 可选门控 -> 任务 -> 尝试。
// 恢复扫描也从队列锁开始，不反向锁定 Worker 进程。
func (s *ExecutionStore) lockExecution(ctx context.Context, tx *sql.Tx, token auth.Token, b worker.Binding, p auth.Permissions, ref worker.AttemptRef, checkGate bool) (executionRows, error) {
	var out executionRows
	state, err := s.lockSession(ctx, tx, token, b, p)
	if err != nil {
		return out, err
	}
	out.session = state
	var queue core.ID
	var gate *core.ID
	err = tx.QueryRowContext(ctx, `SELECT queue_id,gate_id FROM pulse_tasks WHERE namespace_id=? AND id=?`, token.NamespaceID, ref.TaskID).Scan(&queue, &gate)
	if errors.Is(err, sql.ErrNoRows) {
		return out, fault.NotFound
	}
	if err != nil {
		return out, err
	}
	if queue != state.queue || !p.AllowsQueue(queue) {
		return out, fault.NotFound
	}
	out.paused, _, err = lockQueue(ctx, tx, token.NamespaceID, queue)
	if err != nil {
		return out, err
	}
	if checkGate {
		paused, err := lockGate(ctx, tx, token.NamespaceID, gate)
		if err != nil {
			return out, err
		}
		out.paused = out.paused || paused
	}
	out.task, err = readExecutionTask(ctx, tx, token.NamespaceID, ref.TaskID)
	if err != nil {
		return out, err
	}
	out.attempt, err = readAttempt(ctx, tx, token.NamespaceID, ref.AttemptID)
	if err != nil {
		return out, err
	}
	proof, err := worker.SecretDigest(ref.LeaseSecret)
	if err != nil {
		return out, err
	}
	if out.attempt.TaskID != ref.TaskID || out.attempt.Epoch != ref.Epoch || subtle.ConstantTimeCompare(proof[:], out.attempt.TokenDigest[:]) != 1 {
		return out, fault.Forbidden
	}
	var instance core.ID
	if err := tx.QueryRowContext(ctx, `SELECT worker_instance_id FROM pulse_worker_sessions WHERE namespace_id=? AND id=?`, token.NamespaceID, out.attempt.SessionID).Scan(&instance); err != nil {
		return out, err
	}
	if instance != b.InstanceID {
		return out, fault.Forbidden
	}
	return out, nil
}
func readAttempt(ctx context.Context, tx *sql.Tx, ns, id core.ID) (pulse.Attempt, error) {
	var a pulse.Attempt
	a.NamespaceID = ns
	err := tx.QueryRowContext(ctx, `SELECT id,revision,task_id,epoch,session_id,token_digest,state,lease_until,authorized_at,ended_at,error_code,retryable,result,result_digest,concurrency_token FROM pulse_attempts WHERE namespace_id=? AND id=? FOR UPDATE`, ns, id).Scan(&a.ID, &a.Revision, &a.TaskID, &a.Epoch, &a.SessionID, &a.TokenDigest, &a.State, &a.LeaseUntil, &a.AuthorizedAt, &a.EndedAt, &a.ErrorCode, &a.Retryable, &a.Result, &a.ResultDigest, &a.ConcurrencyToken)
	if errors.Is(err, sql.ErrNoRows) {
		err = fault.NotFound
	}
	return a, err
}
func leaseStatus(rows executionRows, b worker.Binding, now time.Time) worker.LeaseStatus {
	t, a := rows.task, rows.attempt
	until := minimumTime(a.LeaseUntil, rows.session.expires, t.view.DeadlineAt)
	if t.lease != nil {
		until = minimumTime(until, *t.lease)
	}
	active := rows.session.expires.After(now) && t.current != nil && *t.current == a.ID && t.epoch == a.Epoch && a.SessionID == b.SessionID && t.lease != nil && t.lease.After(now) && a.LeaseUntil.After(now) && t.view.DeadlineAt.After(now) && ((t.view.State == pulse.TaskLeased && a.State == pulse.AttemptLeased) || (t.view.State == pulse.TaskRunning && a.State == pulse.AttemptRunning))
	if a.AuthorizedAt != nil {
		deadline := a.AuthorizedAt.Add(time.Duration(t.attemptTimeout) * time.Millisecond)
		active = active && deadline.After(now)
		until = minimumTime(until, deadline)
	}
	return worker.LeaseStatus{CancelRequestedAt: t.view.CancelRequestedAt, Active: active, State: a.State, LeaseUntil: until, ServerTime: now}
}
func attemptIdentity(ref worker.AttemptRef) any {
	return struct {
		Task    core.ID
		Attempt core.ID
		Epoch   string
	}{ref.TaskID, ref.AttemptID, fmt.Sprint(ref.Epoch)}
}
func readReceipt(ctx context.Context, tx *sql.Tx, ns, attempt, message core.ID, operation string, digest core.Digest) (*worker.Receipt, error) {
	var saved core.Digest
	var raw []byte
	err := tx.QueryRowContext(ctx, `SELECT request_digest,response FROM pulse_attempt_receipts WHERE namespace_id=? AND attempt_id=? AND operation=? AND message_id=? FOR UPDATE`, ns, attempt, operation, message).Scan(&saved, &raw)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	if saved != digest {
		return nil, fault.IdempotencyConflict
	}
	var receipt worker.Receipt
	if err := json.Unmarshal(raw, &receipt); err != nil {
		return nil, err
	}
	return &receipt, nil
}
func saveReceipt(ctx context.Context, tx *sql.Tx, ns, attempt, message core.ID, operation string, digest core.Digest, receipt worker.Receipt) error {
	id, err := core.NewID()
	if err != nil {
		return err
	}
	raw, err := json.Marshal(receipt)
	if err != nil {
		return err
	}
	_, err = tx.ExecContext(ctx, `INSERT INTO pulse_attempt_receipts(namespace_id,id,attempt_id,operation,message_id,request_digest,response,expires_at) VALUES (?,?,?,?,?,?,?,DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 90 DAY))`, ns, id, attempt, operation, message, digest, string(raw))
	return err
}
func renewPermits(ctx context.Context, tx *sql.Tx, ns, attempt core.ID, until, now time.Time) error {
	result, err := tx.ExecContext(ctx, `UPDATE pulse_permits SET lease_until=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND attempt_id=? AND released_at IS NULL`, until, now, ns, attempt)
	if err != nil {
		return err
	}
	n, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if n < 1 {
		return errors.New("active attempt has no concurrency permit")
	}
	return nil
}
func releasePermits(ctx context.Context, tx *sql.Tx, ns, attempt core.ID, now time.Time) error {
	rows, err := tx.QueryContext(ctx, `SELECT id,limit_id FROM pulse_permits WHERE namespace_id=? AND attempt_id=? AND released_at IS NULL ORDER BY limit_id FOR UPDATE`, ns, attempt)
	if err != nil {
		return err
	}
	type permit struct{ id, limit core.ID }
	permits := []permit{}
	for rows.Next() {
		var p permit
		if err := rows.Scan(&p.id, &p.limit); err != nil {
			rows.Close()
			return err
		}
		permits = append(permits, p)
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return err
	}
	if len(permits) == 0 {
		return errors.New("active attempt has no permit to release")
	}
	if err := releaseApplicationAttempt(ctx, tx, ns, attempt); err != nil {
		return err
	}
	for _, p := range permits {
		result, err := tx.ExecContext(ctx, `UPDATE pulse_concurrency_limits SET used_permits=used_permits-1,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND used_permits>0`, now, ns, p.limit)
		if err := exactlyOne(result, err); err != nil {
			return err
		}
		result, err = tx.ExecContext(ctx, `UPDATE pulse_permits SET released_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND released_at IS NULL`, now, now, ns, p.id)
		if err := exactlyOne(result, err); err != nil {
			return err
		}
	}
	return nil
}

func (s *ExecutionStore) Start(ctx context.Context, token auth.Token, b worker.Binding, message core.ID, ref worker.AttemptRef) (worker.CommandResult, error) {
	return s.executionCommand(ctx, token, b, message, ref, "start")
}
func (s *ExecutionStore) Renew(ctx context.Context, token auth.Token, b worker.Binding, message core.ID, ref worker.AttemptRef) (worker.CommandResult, error) {
	return s.executionCommand(ctx, token, b, message, ref, "renew")
}
func (s *ExecutionStore) executionCommand(ctx context.Context, token auth.Token, b worker.Binding, message core.ID, ref worker.AttemptRef, operation string) (worker.CommandResult, error) {
	var out worker.CommandResult
	digest, err := requestDigest(attemptIdentity(ref))
	if err != nil {
		return out, err
	}
	err = s.transaction(ctx, token, "workers:consume", func(tx *sql.Tx, p auth.Permissions) error {
		rows, err := s.lockExecution(ctx, tx, token, b, p, ref, operation == "start")
		if err != nil {
			return err
		}
		receipt, err := readReceipt(ctx, tx, token.NamespaceID, ref.AttemptID, message, operation, digest)
		if err != nil {
			return err
		}
		now, err := dbNow(ctx, tx)
		if err != nil {
			return err
		}
		status := leaseStatus(rows, b, now)
		if receipt != nil {
			out = worker.CommandResult{Receipt: *receipt, Lease: status, Deduplicated: true}
			return nil
		}
		if !status.Active {
			return worker.LeaseLost
		}
		task, a := rows.task, rows.attempt
		authorized := a.AuthorizedAt
		if operation == "start" {
			if task.view.CancelRequestedAt != nil {
				return worker.LeaseLost
			}
			if a.State != pulse.AttemptLeased || task.view.State != pulse.TaskLeased {
				return fault.Conflict
			}
			if rows.paused {
				return worker.GateClosed
			}
			if task.view.StartDeadlineAt != nil && !task.view.StartDeadlineAt.After(now) {
				return worker.LeaseLost
			}
			if err := consumeQueueStart(ctx, tx, token.NamespaceID, task.view.QueueID, now); err != nil {
				return err
			}
			authorized = &now
		} else if a.State != pulse.AttemptRunning || authorized == nil {
			return fault.Conflict
		}
		if task.view.CancelRequestedAt != nil {
			// 取消后不延长执行窗口；当前权威结果仍可在原租约内提交。
			receipt := worker.Receipt{Operation: operation, TaskState: task.view.State, AttemptState: a.State, TaskRevision: task.view.Revision, LeaseUntil: status.LeaseUntil, Disposition: "APPLIED"}
			if err := saveReceipt(ctx, tx, token.NamespaceID, a.ID, message, operation, digest, receipt); err != nil {
				return err
			}
			out = worker.CommandResult{Receipt: receipt, Lease: status}
			return nil
		}
		until := minimumTime(now.Add(s.options.LeaseDuration), authorized.Add(time.Duration(task.attemptTimeout)*time.Millisecond), task.view.DeadlineAt, rows.session.expires)
		result, err := tx.ExecContext(ctx, `UPDATE pulse_attempts SET state='RUNNING',authorized_at=?,lease_until=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, authorized, until, now, token.NamespaceID, a.ID, a.Revision)
		if err := exactlyOne(result, err); err != nil {
			return err
		}
		result, err = tx.ExecContext(ctx, `UPDATE pulse_tasks SET state='RUNNING',lease_until=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, until, now, token.NamespaceID, ref.TaskID, task.view.Revision)
		if err := exactlyOne(result, err); err != nil {
			return err
		}
		if err := renewPermits(ctx, tx, token.NamespaceID, a.ID, until, now); err != nil {
			return err
		}
		if operation == "start" {
			if err := emitEvent(ctx, tx, token.NamespaceID, ref.TaskID, task.view.Revision+1, "task", "task.started", map[string]any{"attempt_id": a.ID, "epoch": fmt.Sprint(a.Epoch), "session_id": b.SessionID}); err != nil {
				return err
			}
		}
		receipt = &worker.Receipt{Operation: operation, TaskState: pulse.TaskRunning, AttemptState: pulse.AttemptRunning, TaskRevision: task.view.Revision + 1, LeaseUntil: until, Disposition: "APPLIED"}
		if err := saveReceipt(ctx, tx, token.NamespaceID, a.ID, message, operation, digest, *receipt); err != nil {
			return err
		}
		out = worker.CommandResult{Receipt: *receipt, Lease: worker.LeaseStatus{Active: true, State: pulse.AttemptRunning, LeaseUntil: until, ServerTime: now}}
		return nil
	})
	return out, err
}

func (s *ExecutionStore) Resume(ctx context.Context, token auth.Token, b worker.Binding, ref worker.AttemptRef) (worker.LeaseStatus, error) {
	var out worker.LeaseStatus
	err := s.transaction(ctx, token, "workers:consume", func(tx *sql.Tx, p auth.Permissions) error {
		rows, err := s.lockExecution(ctx, tx, token, b, p, ref, false)
		if err != nil {
			return err
		}
		now, err := dbNow(ctx, tx)
		if err != nil {
			return err
		}
		// 只暂按新会话计算有效性；通过所有期限和当前尝试检查后才迁移归属，绝不延长期限。
		previous := rows.attempt.SessionID
		rows.attempt.SessionID = b.SessionID
		out = leaseStatus(rows, b, now)
		if !out.Active || previous == b.SessionID {
			return nil
		}
		result, err := tx.ExecContext(ctx, `UPDATE pulse_attempts SET session_id=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, b.SessionID, now, token.NamespaceID, ref.AttemptID, rows.attempt.Revision)
		if err := exactlyOne(result, err); err != nil {
			return err
		}
		return emitEvent(ctx, tx, token.NamespaceID, ref.AttemptID, rows.attempt.Revision+1, "attempt", "attempt.resumed", map[string]any{"from_session_id": previous, "to_session_id": b.SessionID, "node_id": s.node})
	})
	return out, err
}