package mysql

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"os"
	"strings"
	"testing"
	"time"

	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"
	pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
	"git.yuntuo.hk/k2safe/orbit/pulse/service"
	"git.yuntuo.hk/k2safe/orbit/pulse/worker"
)

type eventFixture struct {
	*executionFixture
	events *events.Service
	store  *EventStore
}

func newEventFixture(t *testing.T, ctx context.Context, db *sql.DB) *eventFixture {
	f := newExecutionFixture(t, ctx, db)
	var p auth.Permissions
	var raw []byte
	if err := db.QueryRowContext(ctx, `SELECT permissions FROM orbit_credentials WHERE namespace_id=? AND id=?`, f.token.NamespaceID, f.token.CredentialID).Scan(&raw); err != nil {
		t.Fatal(err)
	}
	if err := json.Unmarshal(raw, &p); err != nil {
		t.Fatal(err)
	}
	p.Actions = append(p.Actions, "events:manage", "events:read", "events:consume")
	raw, _ = json.Marshal(p)
	f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(raw), f.token.NamespaceID, f.token.CredentialID)
	store := NewEventStore(db)
	return &eventFixture{f, events.New(store), store}
}
func (f *eventFixture) create(in events.CreateInput) events.Subscription {
	f.t.Helper()
	if in.Name == "" {
		in.Name = "result-" + string(executionID(f.t))
	}
	if in.QueueID == "" {
		in.QueueID = f.resources.QueueID
	}
	out, err := f.events.Create(f.ctx, f.token, string(executionID(f.t)), in)
	if err != nil {
		f.t.Fatal(err)
	}
	return out.Subscription
}
func (f *eventFixture) cancelResult() service.TaskView {
	f.t.Helper()
	task := f.submit(1)
	out, err := f.tasks.CancelTask(f.ctx, f.token, task.ID, string(executionID(f.t)), service.CancelInput{ExpectedRevision: task.Revision, Reason: "create result notification"})
	if err != nil {
		f.t.Fatal(err)
	}
	return out.Task
}
func (f *eventFixture) grant(sub core.ID) *events.Grant {
	f.t.Helper()
	v, err := f.events.Claim(f.ctx, f.token, sub, events.ClaimInput{ConnectionID: executionID(f.t), Capacity: 1})
	if err != nil || v == nil {
		f.t.Fatalf("grant: %v", err)
	}
	return v
}
func (f *eventFixture) ack(sub core.ID, g *events.Grant) events.AckResult {
	f.t.Helper()
	v, err := f.events.Ack(f.ctx, f.token, sub, events.AckInput{DeliveryID: g.Delivery.ID, Epoch: g.Delivery.LeaseEpoch, LeaseSecret: g.LeaseSecret})
	if err != nil {
		f.t.Fatal(err)
	}
	return v
}
func (f *eventFixture) subscription(id core.ID) events.Subscription {
	f.t.Helper()
	v, err := f.events.Get(f.ctx, f.token, id)
	if err != nil {
		f.t.Fatal(err)
	}
	return v
}
func (f *eventFixture) expireDelivery(g *events.Grant) {
	f.exec(`UPDATE orbit_deliveries SET lease_until=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 SECOND) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, g.Delivery.ID)
}
func noEventGrant(t *testing.T, f *eventFixture, sub core.ID) {
	t.Helper()
	g, err := f.events.Claim(f.ctx, f.token, sub, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1})
	if err != nil || g != nil {
		t.Fatalf("unexpected grant: %v %v", g, err)
	}
}

func TestIntegrationResultSubscriptions(t *testing.T) {
	dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
	if dsn == "" {
		t.Skip("requires disposable MySQL")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
	defer cancel()
	db, err := Open(ctx, dsn, 32, 8)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if err = Migrate(ctx, db); err != nil {
		t.Fatal(err)
	}
	zero := uint64(0)
	t.Run("explicit_boundary_filters_and_frozen_result_precision", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		task := f.submit(1)
		ref := retryStart(t, f.executionFixture)
		raw := json.RawMessage(`{"n":9007199254740993,"text":"<&>"}`)
		in := worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: raw}
		message := executionID(t)
		if _, err := f.a.Finish(ctx, f.token, f.binding, message, in); err != nil {
			t.Fatal(err)
		}
		if _, err := f.a.Finish(ctx, f.token, f.binding, message, in); err != nil {
			t.Fatal(err)
		}
		if f.count("orbit_result_events", "1=1") != 1 {
			t.Fatal("duplicate result journal entry")
		}
		recent := f.create(events.CreateInput{})
		if recent.CursorSequence != 1 {
			t.Fatal("new subscription did not capture committed head")
		}
		noEventGrant(t, f, recent.ID)
		history := f.create(events.CreateInput{FromSequence: &zero, Types: []string{"task.succeeded"}})
		g := f.grant(history.ID)
		if g.Event.ID != g.Delivery.EventID || g.Event.Sequence != 1 || g.Event.Snapshot.TaskID != task.ID || string(g.Event.Snapshot.Result) != string(raw) || g.Event.Snapshot.CurrentAttemptID == nil || *g.Event.Snapshot.CurrentAttemptID != ref.AttemptID {
			t.Fatalf("bad snapshot: %+v", g.Event)
		}
		f.exec(`UPDATE pulse_tasks SET result='{"changed":true}' WHERE namespace_id=? AND id=?`, f.token.NamespaceID, task.ID)
		f.ack(history.ID, g)
		replay := f.create(events.CreateInput{FromSequence: &zero})
		g = f.grant(replay.ID)
		if string(g.Event.Snapshot.Result) != string(raw) {
			t.Fatal("snapshot read mutable task result")
		}
		f.ack(replay.ID, g)
		f.cancelResult()
		noEventGrant(t, f, history.ID)
		if f.subscription(history.ID).CursorSequence != 2 {
			t.Fatal("filtered event did not advance cursor")
		}
		g = f.grant(recent.ID)
		if g.Event.Type != "task.cancelled" || g.Event.Sequence != 2 {
			t.Fatal("new subscription missed future result")
		}
		_, err := db.ExecContext(ctx, `UPDATE orbit_result_events SET snapshot='{}' WHERE namespace_id=? AND id=?`, f.token.NamespaceID, g.Event.ID)
		if err == nil {
			t.Fatal("journal is mutable")
		}
	})
	t.Run("create_idempotency_queue_ownership_and_permission_isolation", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		other := newEventFixture(t, ctx, db)
		in := events.CreateInput{Name: "stable-result", QueueID: f.resources.QueueID, FromSequence: &zero}
		a, err := f.events.Create(ctx, f.token, "stable", in)
		if err != nil {
			t.Fatal(err)
		}
		b, err := f.events.Create(ctx, f.token, "stable", in)
		if err != nil || !b.Deduplicated || a.Subscription.ID != b.Subscription.ID {
			t.Fatal("unstable create receipt", err)
		}
		in.MaxPending = 2
		_, err = f.events.Create(ctx, f.token, "stable", in)
		expectWorkerCode(t, err, "IDEMPOTENCY_CONFLICT")
		_, err = f.events.Get(ctx, other.token, a.Subscription.ID)
		expectWorkerCode(t, err, "NOT_FOUND")
		_, err = f.events.Create(ctx, f.token, "cross-queue", events.CreateInput{Name: "cross", QueueID: other.resources.QueueID})
		expectWorkerCode(t, err, "NOT_FOUND")
		secondID := executionID(t)
		secret, digest, err := auth.Issue(f.token.NamespaceID, secondID)
		if err != nil {
			t.Fatal(err)
		}
		p, _ := json.Marshal(auth.Permissions{Actions: []string{"events:read", "events:consume", "events:manage"}, Queues: []string{"*"}})
		f.exec(`INSERT INTO orbit_credentials(namespace_id,id,name,secret_digest,permissions) VALUES (?,?,?, ?,?)`, f.token.NamespaceID, secondID, "second", digest, string(p))
		second, err := auth.Parse(secret)
		if err != nil {
			t.Fatal(err)
		}
		_, err = f.events.Get(ctx, second, a.Subscription.ID)
		expectWorkerCode(t, err, "NOT_FOUND")
		own, err := f.events.Create(ctx, second, "stable", events.CreateInput{Name: "second", QueueID: f.resources.QueueID, FromSequence: &zero})
		if err != nil || own.Subscription.ID == a.Subscription.ID {
			t.Fatal("create receipt leaked across owners", err)
		}
		p, _ = json.Marshal(auth.Permissions{Actions: []string{"events:read", "events:consume", "events:manage"}, Queues: []string{string(executionID(t))}})
		f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(p), f.token.NamespaceID, secondID)
		_, err = f.events.Get(ctx, second, own.Subscription.ID)
		expectWorkerCode(t, err, "NOT_FOUND")
		p, _ = json.Marshal(auth.Permissions{Actions: []string{"tasks:read"}, Queues: []string{"*"}})
		f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(p), f.token.NamespaceID, f.token.CredentialID)
		_, err = f.events.Claim(ctx, f.token, a.Subscription.ID, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1})
		expectWorkerCode(t, err, "PERMISSION_DENIED")
		f.exec(`UPDATE orbit_credentials SET revoked_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, f.token.CredentialID)
		_, err = f.events.Get(ctx, f.token, a.Subscription.ID)
		expectWorkerCode(t, err, "UNAUTHENTICATED")
	})
	t.Run("backpressure_does_not_block_result_commit_or_skip_history", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		sub := f.create(events.CreateInput{MaxPending: 1})
		f.cancelResult()
		first := f.grant(sub.ID)
		for range 3 {
			f.cancelResult()
		}
		noEventGrant(t, f, sub.ID)
		got := f.subscription(sub.ID)
		if got.PendingCount != 1 || got.CursorSequence != 1 || f.count("orbit_result_events", "1=1") != 4 {
			t.Fatal("window violated or producer blocked")
		}
		f.ack(sub.ID, first)
		for seq := uint64(2); seq <= 4; seq++ {
			g := f.grant(sub.ID)
			if g.Event.Sequence != seq {
				t.Fatal("backlog skipped")
			}
			f.ack(sub.ID, g)
		}
		if got = f.subscription(sub.ID); got.PendingCount != 0 || got.CursorSequence != 4 || got.Revision != 1 {
			t.Fatal("invalid drained window or configuration revision changed")
		}
	})
	t.Run("competing_consumers_and_duplicate_acks", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		sub := f.create(events.CreateInput{})
		f.cancelResult()
		type answer struct {
			g   *events.Grant
			err error
		}
		results := make(chan answer, 12)
		for range 12 {
			connection := executionID(t)
			go func() {
				g, err := f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: connection, Capacity: 1})
				results <- answer{g, err}
			}()
		}
		var winner *events.Grant
		for range 12 {
			r := <-results
			if r.err != nil {
				t.Fatal(r.err)
			}
			if r.g != nil {
				if winner != nil {
					t.Fatal("two grants for one unexpired delivery")
				}
				winner = r.g
			}
		}
		if winner == nil {
			t.Fatal("no consumer won")
		}
		acks := make(chan events.AckResult, 12)
		errs := make(chan error, 12)
		for range 12 {
			go func() {
				v, err := f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: winner.Delivery.ID, Epoch: winner.Delivery.LeaseEpoch, LeaseSecret: winner.LeaseSecret})
				acks <- v
				errs <- err
			}()
		}
		fresh := 0
		var ackTime time.Time
		for range 12 {
			v := <-acks
			if err := <-errs; err != nil {
				t.Fatal(err)
			}
			if !v.Deduplicated {
				fresh++
			}
			if ackTime.IsZero() {
				ackTime = v.AcknowledgedAt
			} else if !ackTime.Equal(v.AcknowledgedAt) {
				t.Fatal("ack receipt changed")
			}
		}
		if fresh != 1 || f.subscription(sub.ID).PendingCount != 0 || f.count("orbit_audit_events", "resource_id=? AND action='delivery.ack'", winner.Delivery.ID) != 1 {
			t.Fatal("duplicate ack side effects")
		}
	})
	t.Run("lease_expiry_backoff_latest_proof_and_node_restart", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		sub := f.create(events.CreateInput{MaxAttempts: 2})
		f.cancelResult()
		first := f.grant(sub.ID)
		f.expireDelivery(first)
		// 新建存储实例模拟所有连接内存丢失；权威交付依然来自数据库。
		f.events = events.New(NewEventStore(db))
		noEventGrant(t, f, sub.ID)
		v, err := f.events.Delivery(ctx, f.token, sub.ID, first.Delivery.ID)
		if err != nil || v.State != platform.DeliveryPending || v.Attempts != 1 {
			t.Fatal("lease was not recovered", err)
		}
		f.exec(`UPDATE orbit_deliveries SET available_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, v.ID)
		second := f.grant(sub.ID)
		if second.Event.ID != first.Event.ID || second.Delivery.ID != first.Delivery.ID || second.Delivery.LeaseEpoch != 2 || second.Delivery.Attempts != 2 || second.LeaseSecret == first.LeaseSecret {
			t.Fatal("redelivery identity/proof incorrect")
		}
		_, err = f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: first.Delivery.ID, Epoch: first.Delivery.LeaseEpoch, LeaseSecret: first.LeaseSecret})
		expectWorkerCode(t, err, "DELIVERY_LEASE_LOST")
		f.expireDelivery(second)
		noEventGrant(t, f, sub.ID)
		v, err = f.events.Delivery(ctx, f.token, sub.ID, second.Delivery.ID)
		if err != nil || v.State != platform.DeliveryExpired || f.subscription(sub.ID).PendingCount != 1 {
			t.Fatal("exhaustion discarded event", err)
		}
		// 确认本身不是新的执行授权；最近证明在未被新代次替代时允许迟到确认。
		f.ack(sub.ID, second)
		if !f.ack(sub.ID, second).Deduplicated {
			t.Fatal("late ack replay lost")
		}
	})
	t.Run("manual_retry_adds_budget_without_resetting_attempt_or_epoch", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		sub := f.create(events.CreateInput{MaxAttempts: 1})
		f.cancelResult()
		first := f.grant(sub.ID)
		f.expireDelivery(first)
		noEventGrant(t, f, sub.ID)
		v, err := f.events.Delivery(ctx, f.token, sub.ID, first.Delivery.ID)
		if err != nil {
			t.Fatal(err)
		}
		in := events.RetryInput{ExpectedRevision: v.Revision, Reason: "receiver fixed", AdditionalAttempts: 2}
		out, err := f.events.Retry(ctx, f.token, sub.ID, v.ID, "retry", in)
		if err != nil || out.Delivery.Attempts != 1 || out.Delivery.AttemptBudget != 3 || out.Delivery.LeaseEpoch != 1 {
			t.Fatal("budget reset", err)
		}
		second := f.grant(sub.ID)
		if second.Delivery.Attempts != 2 || second.Delivery.LeaseEpoch != 2 {
			t.Fatal("retry reused attempt")
		}
		again, err := f.events.Retry(ctx, f.token, sub.ID, v.ID, "retry", in)
		if err != nil || !again.Deduplicated || again.Delivery.Revision != out.Delivery.Revision {
			t.Fatal("retry replay changed receipt", err)
		}
		_, err = f.events.Retry(ctx, f.token, sub.ID, v.ID, "another", in)
		expectWorkerCode(t, err, "STATE_CONFLICT")
		_, err = f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: first.Delivery.ID, Epoch: 1, LeaseSecret: first.LeaseSecret})
		expectWorkerCode(t, err, "DELIVERY_LEASE_LOST")
		f.ack(sub.ID, second)
		if f.count("orbit_audit_events", "resource_id=? AND action='delivery.retry'", v.ID) != 1 {
			t.Fatal("manual retry audit duplicated")
		}
	})
	t.Run("pause_blocks_new_delivery_but_allows_inflight_ack", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		sub := f.create(events.CreateInput{})
		f.cancelResult()
		g := f.grant(sub.ID)
		paused, err := f.events.Control(ctx, f.token, sub.ID, "pause", events.ControlInput{ExpectedRevision: 1, Enabled: false, Reason: "receiver maintenance"})
		if err != nil || paused.Subscription.Enabled {
			t.Fatal(err)
		}
		f.cancelResult()
		noEventGrant(t, f, sub.ID)
		f.ack(sub.ID, g)
		_, err = f.events.Control(ctx, f.token, sub.ID, "resume", events.ControlInput{ExpectedRevision: 2, Enabled: true, Reason: "receiver ready"})
		if err != nil {
			t.Fatal(err)
		}
		resumed := f.grant(sub.ID)
		if resumed.Event.Sequence != 2 {
			t.Fatal("resume lost result")
		}
	})
	t.Run("ack_and_materialization_are_transactional_with_audit", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		sub := f.create(events.CreateInput{})
		f.cancelResult()
		f.exec(`CREATE TRIGGER test_delivery_reject BEFORE INSERT ON orbit_deliveries FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected delivery failure'`)
		t.Cleanup(func() {
			f.exec(`DROP TRIGGER IF EXISTS test_delivery_reject`)
			f.exec(`DROP TRIGGER IF EXISTS test_delivery_ack_reject`)
		})
		_, err := f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1})
		if err == nil {
			t.Fatal("expected delivery insertion failure")
		}
		f.exec(`DROP TRIGGER test_delivery_reject`)
		v := f.subscription(sub.ID)
		if v.CursorSequence != 0 || v.PendingCount != 0 || f.count("orbit_deliveries", "subscription_id=?", sub.ID) != 0 {
			t.Fatal("partial materialization escaped rollback")
		}
		g := f.grant(sub.ID)
		f.exec(`CREATE TRIGGER test_delivery_ack_reject BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected audit failure'`)
		_, err = f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: g.Delivery.ID, Epoch: g.Delivery.LeaseEpoch, LeaseSecret: g.LeaseSecret})
		if err == nil {
			t.Fatal("ack escaped audit failure")
		}
		f.exec(`DROP TRIGGER test_delivery_ack_reject`)
		d, err := f.events.Delivery(ctx, f.token, sub.ID, g.Delivery.ID)
		if err != nil || d.State != platform.DeliveryLeased || d.AcknowledgedAt != nil || f.subscription(sub.ID).PendingCount != 1 {
			t.Fatal("ack partial commit", err)
		}
		f.ack(sub.ID, g)
	})
	t.Run("producer_journal_failure_rolls_back_task_event_and_quota", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		task := f.submit(1)
		f.exec(`CREATE TRIGGER test_result_reject BEFORE INSERT ON orbit_result_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected journal failure'`)
		t.Cleanup(func() { f.exec(`DROP TRIGGER IF EXISTS test_result_reject`) })
		_, err := f.tasks.CancelTask(ctx, f.token, task.ID, "cancel", service.CancelInput{ExpectedRevision: task.Revision, Reason: "verify atomic result"})
		if err == nil {
			t.Fatal("expected journal failure")
		}
		f.exec(`DROP TRIGGER test_result_reject`)
		if f.get(task.ID).State != pulse.TaskWaiting || f.count("orbit_events", "aggregate_id=?", task.ID) != 1 || f.count("orbit_event_streams", "1=1") != 0 {
			t.Fatal("result partial commit")
		}
		assertApplicationCounts(t, f.executionFixture, 1, 0, 0)
		if _, err = f.tasks.CancelTask(ctx, f.token, task.ID, "cancel", service.CancelInput{ExpectedRevision: task.Revision, Reason: "verify atomic result"}); err != nil {
			t.Fatal(err)
		}
	})
	t.Run("all_terminal_and_unknown_paths_emit_but_safe_requeue_does_not", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		sub := f.create(events.CreateInput{})
		f.submit(2)
		first := retryStart(t, f.executionFixture)
		retryFinish(t, f.executionFixture, first, "FAILED", true)
		if f.count("orbit_result_events", "1=1") != 0 {
			t.Fatal("safe retry published final result")
		}
		makeRetryDue(f.executionFixture, first.TaskID)
		second := retryStart(t, f.executionFixture)
		retryFinish(t, f.executionFixture, second, "FAILED", false)
		g := f.grant(sub.ID)
		if g.Event.Type != "task.failed" {
			t.Fatal("missing failure result")
		}
		f.ack(sub.ID, g)
		f.submit(1)
		unknown := retryStart(t, f.executionFixture)
		retryFinish(t, f.executionFixture, unknown, "UNKNOWN", false)
		g = f.grant(sub.ID)
		if g.Event.Type != "task.needs_review" || !g.Event.Snapshot.EffectUncertain {
			t.Fatal("missing unknown result")
		}
		f.ack(sub.ID, g)
		f.submit(1)
		lost := retryStart(t, f.executionFixture)
		f.expire(lost)
		if _, err := f.storeB.Recover(ctx, 100); err != nil {
			t.Fatal(err)
		}
		g = f.grant(sub.ID)
		if g.Event.Type != "task.needs_review" {
			t.Fatal("recovery failed to publish result")
		}
	})
	t.Run("namespace_limit_future_cursor_and_history_gap_are_explicit", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		future := uint64(1)
		_, err := f.events.Create(ctx, f.token, "future", events.CreateInput{Name: "future", QueueID: f.resources.QueueID, FromSequence: &future})
		expectWorkerCode(t, err, "INVALID_ARGUMENT")
		sub := f.create(events.CreateInput{})
		f.cancelResult()
		f.exec(`UPDATE orbit_event_streams SET retained_from_sequence=2 WHERE namespace_id=?`, f.token.NamespaceID)
		_, err = f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: executionID(t), Capacity: 1})
		expectWorkerCode(t, err, "EVENT_HISTORY_EXPIRED")
		_, err = f.events.Create(ctx, f.token, "expired", events.CreateInput{Name: "expired", QueueID: f.resources.QueueID, FromSequence: &zero})
		expectWorkerCode(t, err, "EVENT_HISTORY_EXPIRED")
		for i := 1; i < events.MaxSubscriptions; i++ {
			f.create(events.CreateInput{Name: fmt.Sprintf("limit-%d", i)})
		}
		_, err = f.events.Create(ctx, f.token, "overflow", events.CreateInput{Name: "overflow", QueueID: f.resources.QueueID})
		expectWorkerCode(t, err, "SUBSCRIPTION_LIMIT")
	})

	t.Run("locked_subscriber_cannot_block_task_result_commit", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		sub := f.create(events.CreateInput{})
		task := f.submit(1)
		tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
		if err != nil {
			t.Fatal(err)
		}
		defer tx.Rollback()
		var id core.ID
		if err = tx.QueryRowContext(ctx, `SELECT id FROM orbit_subscriptions WHERE namespace_id=? AND id=? FOR UPDATE`, f.token.NamespaceID, sub.ID).Scan(&id); err != nil {
			t.Fatal(err)
		}
		bounded, stop := context.WithTimeout(ctx, 2*time.Second)
		defer stop()
		if _, err = f.tasks.CancelTask(bounded, f.token, task.ID, "while-locked", service.CancelInput{ExpectedRevision: task.Revision, Reason: "subscriber is slow"}); err != nil {
			t.Fatal("producer waited for subscriber lock", err)
		}
		if err = tx.Rollback(); err != nil {
			t.Fatal(err)
		}
		g := f.grant(sub.ID)
		if g.Event.Snapshot.TaskID != task.ID {
			t.Fatal("result lost while subscriber blocked")
		}
	})
	t.Run("connection_capacity_is_database_enforced", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		sub := f.create(events.CreateInput{})
		f.cancelResult()
		f.cancelResult()
		claim := events.ClaimInput{ConnectionID: executionID(t), Capacity: 1}
		first, err := f.events.Claim(ctx, f.token, sub.ID, claim)
		if err != nil || first == nil {
			t.Fatal(err)
		}
		if g, err := f.events.Claim(ctx, f.token, sub.ID, claim); err != nil || g != nil {
			t.Fatal("connection capacity exceeded", err)
		}
		f.ack(sub.ID, first)
		next, err := f.events.Claim(ctx, f.token, sub.ID, claim)
		if err != nil || next == nil || next.Event.ID == first.Event.ID {
			t.Fatal("capacity not released on confirmation", err)
		}
	})
	t.Run("late_ack_races_new_delivery_without_double_confirmation", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		sub := f.create(events.CreateInput{MaxPending: 1, MaxAttempts: 2})
		for range 12 {
			f.cancelResult()
			first := f.grant(sub.ID)
			f.expireDelivery(first)
			noEventGrant(t, f, sub.ID)
			f.exec(`UPDATE orbit_deliveries SET available_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.token.NamespaceID, first.Delivery.ID)
			type claimResult struct {
				grant *events.Grant
				err   error
			}
			claims := make(chan claimResult, 1)
			acks := make(chan error, 1)
			start := make(chan struct{})
			connection := executionID(t)
			go func() {
				<-start
				g, err := f.events.Claim(ctx, f.token, sub.ID, events.ClaimInput{ConnectionID: connection, Capacity: 1})
				claims <- claimResult{g, err}
			}()
			go func() {
				<-start
				_, err := f.events.Ack(ctx, f.token, sub.ID, events.AckInput{DeliveryID: first.Delivery.ID, Epoch: first.Delivery.LeaseEpoch, LeaseSecret: first.LeaseSecret})
				acks <- err
			}()
			close(start)
			c := <-claims
			ackErr := <-acks
			if c.err != nil {
				t.Fatal(c.err)
			}
			if c.grant == nil {
				if ackErr != nil {
					t.Fatal("neither ack nor redelivery won", ackErr)
				}
			} else {
				expectWorkerCode(t, ackErr, "DELIVERY_LEASE_LOST")
				f.ack(sub.ID, c.grant)
			}
			if f.subscription(sub.ID).PendingCount != 0 || f.count("orbit_audit_events", "resource_id=? AND action='delivery.ack'", first.Delivery.ID) != 1 {
				t.Fatal("race duplicated confirmation or leaked window")
			}
		}
	})
	t.Run("proof_is_not_stored_in_audit_or_plaintext_column", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		sub := f.create(events.CreateInput{})
		f.cancelResult()
		g := f.grant(sub.ID)
		f.ack(sub.ID, g)
		var before, after string
		if err := db.QueryRowContext(ctx, "SELECT `before`,`after` FROM orbit_audit_events WHERE namespace_id=? AND resource_id=? AND action='delivery.ack'", f.token.NamespaceID, g.Delivery.ID).Scan(&before, &after); err != nil {
			t.Fatal(err)
		}
		if strings.Contains(before+after, g.LeaseSecret) {
			t.Fatal("delivery proof leaked to audit")
		}
	})
}

func TestIntegrationResultCommitOrder(t *testing.T) {
	dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
	if dsn == "" {
		t.Skip("requires disposable MySQL")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	db, err := Open(ctx, dsn, 16, 4)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if err = Migrate(ctx, db); err != nil {
		t.Fatal(err)
	}
	for _, rollback := range []bool{false, true} {
		t.Run(fmt.Sprintf("first_rollback_%t", rollback), func(t *testing.T) {
			f := newEventFixture(t, ctx, db)
			sub := f.create(events.CreateInput{})
			first, second := f.submit(1), f.submit(1)
			finish := func(tx *sql.Tx, task service.TaskView) error {
				if _, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET state='CANCELLED',completed_at=UTC_TIMESTAMP(6),revision=revision+1 WHERE namespace_id=? AND id=?`, f.token.NamespaceID, task.ID); err != nil {
					return err
				}
				return emitEvent(ctx, tx, f.token.NamespaceID, task.ID, task.Revision+1, "task", "task.cancelled", map[string]any{"state": "CANCELLED"})
			}
			tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
			if err != nil {
				t.Fatal(err)
			}
			defer tx.Rollback()
			if err = finish(tx, first); err != nil {
				t.Fatal(err)
			}
			started := make(chan struct{})
			result := make(chan error, 1)
			go func() {
				result <- WithTx(ctx, db, func(other *sql.Tx) error { close(started); return finish(other, second) })
			}()
			<-started
			select {
			case err := <-result:
				t.Fatalf("later transaction crossed uncommitted stream: %v", err)
			case <-time.After(100 * time.Millisecond):
			}
			// 已提交读在第一事务未提交时看不到较大的序号；消费也不等待生产者流锁。
			noEventGrant(t, f, sub.ID)
			if rollback {
				err = tx.Rollback()
			} else {
				err = tx.Commit()
			}
			if err != nil {
				t.Fatal(err)
			}
			if err = <-result; err != nil {
				t.Fatal(err)
			}
			want := uint64(2)
			if rollback {
				want = 1
			}
			var head uint64
			if err = db.QueryRowContext(ctx, `SELECT last_sequence FROM orbit_event_streams WHERE namespace_id=?`, f.token.NamespaceID).Scan(&head); err != nil || head != want {
				t.Fatal("nontransactional sequence", err)
			}
			for seq := uint64(1); seq <= want; seq++ {
				g := f.grant(sub.ID)
				if g.Event.Sequence != seq {
					t.Fatal("commit order skipped result")
				}
				f.ack(sub.ID, g)
			}
		})
	}
	t.Run("create_new_boundary_waits_for_earlier_producer", func(t *testing.T) {
		f := newEventFixture(t, ctx, db)
		task := f.submit(1)
		tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
		if err != nil {
			t.Fatal(err)
		}
		defer tx.Rollback()
		if _, err = tx.ExecContext(ctx, `UPDATE pulse_tasks SET state='CANCELLED',completed_at=UTC_TIMESTAMP(6),revision=2 WHERE namespace_id=? AND id=?`, f.token.NamespaceID, task.ID); err != nil {
			t.Fatal(err)
		}
		if err = emitEvent(ctx, tx, f.token.NamespaceID, task.ID, 2, "task", "task.cancelled", map[string]any{}); err != nil {
			t.Fatal(err)
		}
		type result struct {
			v   events.SubscriptionResult
			err error
		}
		ch := make(chan result, 1)
		go func() {
			v, err := f.events.Create(ctx, f.token, "boundary", events.CreateInput{Name: "boundary", QueueID: f.resources.QueueID})
			ch <- result{v, err}
		}()
		select {
		case <-ch:
			t.Fatal("subscription creation crossed stream boundary")
		case <-time.After(100 * time.Millisecond):
		}
		if err = tx.Commit(); err != nil {
			t.Fatal(err)
		}
		r := <-ch
		if r.err != nil || r.v.Subscription.CursorSequence != 1 {
			t.Fatal("boundary did not use committed head", r.err)
		}
		noEventGrant(t, f, r.v.Subscription.ID)
	})
}