package message

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"path/filepath"
	"testing"
	"time"

	"git.asio.asia/nixevol/NixMsg/internal/app/port"
	"git.asio.asia/nixevol/NixMsg/internal/auth"
	"git.asio.asia/nixevol/NixMsg/internal/protocol"
	"git.asio.asia/nixevol/NixMsg/internal/store"
)

type deliveryEnv struct {
	t     *testing.T
	app   *App
	db    *store.DB
	conns *MemoryConns
	down  *RecordingDownlink
	nowMs int64
}

func openDeliveryEnv(t *testing.T, mutate func(*Limits)) *deliveryEnv {
	t.Helper()
	dir := t.TempDir()
	db, err := store.Open(filepath.Join(dir, "data"), "FULL")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = db.Close() })
	lim := defaultTestLimits()
	lim.GraceSeconds = 60
	lim.AckTimeoutSeconds = 300
	lim.DeliveryWindow = 32
	if mutate != nil {
		mutate(&lim)
	}
	nowMs := int64(1_700_000_000_000)
	conns := NewMemoryConns()
	down := &RecordingDownlink{}
	app := New(db, lim, auth.NewStubHashPool(),
		WithNow(func() time.Time { return time.UnixMilli(nowMs) }),
		WithLocks(auth.NewStubLoginLocks()),
		WithConnRegistry(conns),
		WithDownlink(down),
	)
	return &deliveryEnv{t: t, app: app, db: db, conns: conns, down: down, nowMs: nowMs}
}

func (e *deliveryEnv) setNow(ms int64) {
	e.nowMs = ms
	e.app.nowFn = func() time.Time { return time.UnixMilli(e.nowMs) }
}

func (e *deliveryEnv) online(id string, connID port.ConnID) {
	e.conns.Set(id, LiveConn{ConnID: connID, MaxReceiveBytes: 0, MaxPacketSize: 0, Ready: true})
}

func (e *deliveryEnv) deliveryState(seq int64, endpointID string) (state, reason string) {
	e.t.Helper()
	err := e.db.Read.QueryRow(`SELECT state, reason FROM deliveries WHERE seq=? AND endpoint_id=?`, seq, endpointID).Scan(&state, &reason)
	if err != nil {
		e.t.Fatal(err)
	}
	return
}

func (e *deliveryEnv) msgState(sender, id string) (state, reason string) {
	e.t.Helper()
	err := e.db.Read.QueryRow(`SELECT state, reason FROM messages WHERE sender_id=? AND id=?`, sender, id).Scan(&state, &reason)
	if err != nil {
		e.t.Fatal(err)
	}
	return
}

func (e *deliveryEnv) seqOf(sender, id string) int64 {
	e.t.Helper()
	var seq int64
	if err := e.db.Read.QueryRow(`SELECT seq FROM messages WHERE sender_id=? AND id=?`, sender, id).Scan(&seq); err != nil {
		e.t.Fatal(err)
	}
	return seq
}

func keepTrue() *protocol.OfflineOpts {
	ttl := int64(3600)
	return &protocol.OfflineOpts{Keep: true, TTLSeconds: &ttl}
}

func TestDeliveryStateMachine(t *testing.T) {
	t.Parallel()

	t.Run("F10_grace_within_keeps_pending", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		// offline_since = now → 宽限内
		ctx := context.Background()
		res, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("g1", "bob"))
		if err != nil {
			t.Fatal(err)
		}
		if res.State != StateDispatched {
			t.Fatalf("state=%s", res.State)
		}
		seq := e.seqOf("alice", "g1")
		st, reason := e.deliveryState(seq, "bob")
		if st != DeliveryPending || reason != "" {
			t.Fatalf("got %s/%s", st, reason)
		}
		var exp sql.NullInt64
		_ = e.db.Read.QueryRow(`SELECT expire_at FROM deliveries WHERE seq=?`, seq).Scan(&exp)
		if !exp.Valid || exp.Int64 != e.nowMs+60_000 {
			t.Fatalf("expire_at=%v want %d", exp, e.nowMs+60_000)
		}
	})

	t.Run("F10_grace_exceeded_dropped_offline", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		_ = e.db.Queue.Do(context.Background(), func(tx *sql.Tx) error {
			_, err := tx.Exec(`UPDATE endpoints SET offline_since = ? WHERE id=?`, e.nowMs-120_000, "bob")
			return err
		})
		ctx := context.Background()
		res, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("g2", "bob"))
		if err != nil {
			t.Fatal(err)
		}
		if res.State != StateCompleted {
			t.Fatalf("state=%s", res.State)
		}
		seq := e.seqOf("alice", "g2")
		st, reason := e.deliveryState(seq, "bob")
		if st != DeliveryDropped || reason != ReasonOffline {
			t.Fatalf("got %s/%s", st, reason)
		}
	})

	t.Run("F09_keep_ttl_expire_via_cleanup", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, func(l *Limits) { l.GraceSeconds = 60 })
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		req := baseSend("k1", "bob")
		req.Offline = keepTrue()
		ctx := context.Background()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
			t.Fatal(err)
		}
		seq := e.seqOf("alice", "k1")
		st, _ := e.deliveryState(seq, "bob")
		if st != DeliveryPending {
			t.Fatalf("state=%s", st)
		}
		e.setNow(e.nowMs + 3600_000 + 1)
		if err := e.app.CleanupOnce(ctx, e.nowMs); err != nil {
			t.Fatal(err)
		}
		st, reason := e.deliveryState(seq, "bob")
		if st != DeliveryExpired || reason != ReasonTTL {
			t.Fatalf("got %s/%s", st, reason)
		}
	})

	t.Run("F08_ack_timeout_not_keep_dropped", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, func(l *Limits) { l.AckTimeoutSeconds = 10 })
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		e.online("bob", "c-bob")
		ctx := context.Background()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("a1", "bob")); err != nil {
			t.Fatal(err)
		}
		if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
			t.Fatal(err)
		}
		if e.down.FilterType(protocol.TypeMsg) < 1 {
			t.Fatal("expected msg publish")
		}
		seq := e.seqOf("alice", "a1")
		var pushed sql.NullString
		_ = e.db.Read.QueryRow(`SELECT pushed_conn FROM deliveries WHERE seq=?`, seq).Scan(&pushed)
		if !pushed.Valid {
			t.Fatal("expected pushed")
		}
		e.setNow(e.nowMs + 11_000)
		if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
			t.Fatal(err)
		}
		st, reason := e.deliveryState(seq, "bob")
		if st != DeliveryDropped || reason != ReasonNotAcked {
			t.Fatalf("got %s/%s", st, reason)
		}
	})

	t.Run("F08_ack_timeout_keep_repush_then_expire", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, func(l *Limits) {
			l.AckTimeoutSeconds = 10
		})
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		e.online("bob", "c-bob")
		req := baseSend("a2", "bob")
		ttl := int64(30)
		req.Offline = &protocol.OfflineOpts{Keep: true, TTLSeconds: &ttl}
		ctx := context.Background()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
			t.Fatal(err)
		}
		if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
			t.Fatal(err)
		}
		seq := e.seqOf("alice", "a2")
		// 未过 expire：清标记重推
		e.setNow(e.nowMs + 11_000)
		if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
			t.Fatal(err)
		}
		st, _ := e.deliveryState(seq, "bob")
		if st != DeliveryPending {
			t.Fatalf("want pending got %s", st)
		}
		var pushed sql.NullString
		_ = e.db.Read.QueryRow(`SELECT pushed_conn FROM deliveries WHERE seq=?`, seq).Scan(&pushed)
		// 可能已重推或仍清空后待推
		// 过 expire
		e.setNow(e.nowMs + 30_000)
		_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
			_, err := tx.Exec(`UPDATE deliveries SET pushed_conn=?, pushed_at=? WHERE seq=?`, "c-bob", e.nowMs-11_000, seq)
			return err
		})
		if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
			t.Fatal(err)
		}
		st, reason := e.deliveryState(seq, "bob")
		if st != DeliveryExpired || reason != ReasonTTL {
			t.Fatalf("got %s/%s", st, reason)
		}
	})

	t.Run("F12_scheduled_recall", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		delay := int64(10_000)
		req := baseSend("r1", "bob")
		req.DelayMs = &delay
		ctx := context.Background()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
			t.Fatal(err)
		}
		data, err := e.app.Recall(ctx, "alice", &protocol.Recall{V: 1, Type: protocol.TypeRecall, RID: "1", ID: "r1"})
		if err != nil {
			t.Fatal(err)
		}
		if data.Result != "recalled" || data.Recalled != 0 {
			t.Fatalf("%+v", data)
		}
		st, reason := e.msgState("alice", "r1")
		if st != StateCompleted || reason != ReasonRecalled {
			t.Fatalf("%s/%s", st, reason)
		}
	})

	t.Run("F13_recall_race_partial", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		insertEndpoint(t, e.db, "carol", "", 1, 0)
		ctx := context.Background()
		_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
			_, err := tx.Exec(`INSERT INTO groups(id, name, owner_id, created_at) VALUES('g1','g','alice',?)`, e.nowMs)
			if err != nil {
				return err
			}
			for _, m := range []string{"alice", "bob", "carol"} {
				if _, err := tx.Exec(`INSERT INTO group_members(group_id, endpoint_id, joined_at) VALUES('g1',?,?)`, m, e.nowMs); err != nil {
					return err
				}
			}
			return nil
		})
		req := &protocol.Send{
			V: protocol.Version, Type: protocol.TypeSend, RID: "1", ID: "grp1",
			To:   protocol.Target{Kind: protocol.TargetGroup, ID: "g1"},
			Body: protocol.Body{Enc: protocol.EncUTF8, Data: "hi"},
		}
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
			t.Fatal(err)
		}
		seq := e.seqOf("alice", "grp1")
		// bob 先确认
		if _, err := e.app.Ack(ctx, "bob", &protocol.Ack{V: 1, Type: protocol.TypeAck, RID: "a", From: "alice", ID: "grp1"}); err != nil {
			t.Fatal(err)
		}
		data, err := e.app.Recall(ctx, "alice", &protocol.Recall{V: 1, Type: protocol.TypeRecall, RID: "2", ID: "grp1"})
		if err != nil {
			t.Fatal(err)
		}
		if data.Result != "partial" || data.Accepted != 1 || data.Recalled != 1 {
			t.Fatalf("%+v", data)
		}
		st, _ := e.deliveryState(seq, "carol")
		if st != DeliveryRecalled {
			t.Fatalf("carol=%s", st)
		}
		st, _ = e.deliveryState(seq, "bob")
		if st != DeliveryAccepted {
			t.Fatalf("bob=%s", st)
		}
	})

	t.Run("F13_recall_pending_unpushed", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		ctx := context.Background()
		req := baseSend("r2", "bob")
		req.Offline = keepTrue()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
			t.Fatal(err)
		}
		data, err := e.app.Recall(ctx, "alice", &protocol.Recall{V: 1, Type: protocol.TypeRecall, RID: "1", ID: "r2"})
		if err != nil {
			t.Fatal(err)
		}
		if data.Result != "recalled" || data.Recalled != 1 {
			t.Fatalf("%+v", data)
		}
	})

	t.Run("F14_receipt_when_sender_offline", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		e.online("bob", "c-bob")
		ctx := context.Background()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("rc1", "bob")); err != nil {
			t.Fatal(err)
		}
		if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
			t.Fatal(err)
		}
		if _, err := e.app.Ack(ctx, "bob", &protocol.Ack{V: 1, Type: protocol.TypeAck, RID: "a", From: "alice", ID: "rc1"}); err != nil {
			t.Fatal(err)
		}
		var n int
		if err := e.db.Read.QueryRow(`SELECT COUNT(*) FROM receipts WHERE sender_id=? AND msg_id=? AND state=?`, "alice", "rc1", DeliveryAccepted).Scan(&n); err != nil {
			t.Fatal(err)
		}
		if n != 1 {
			t.Fatalf("receipts=%d", n)
		}
		// alice 上线后能推到回执
		e.online("alice", "c-alice")
		if err := e.app.PushPending(ctx, "alice", "c-alice"); err != nil {
			t.Fatal(err)
		}
		if e.down.FilterType(protocol.TypeReceipt) < 1 {
			t.Fatal("expected receipt push")
		}
	})

	t.Run("F18_body_gone_record_zero_idempotent_kept", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, func(l *Limits) { l.RecordRetentionDays = 0 })
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		e.online("bob", "c-bob")
		ctx := context.Background()
		req := baseSend("f18", "bob")
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
			t.Fatal(err)
		}
		if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
			t.Fatal(err)
		}
		if _, err := e.app.Ack(ctx, "bob", &protocol.Ack{V: 1, Type: protocol.TypeAck, RID: "a", From: "alice", ID: "f18"}); err != nil {
			t.Fatal(err)
		}
		var bodies int
		_ = e.db.Read.QueryRow(`SELECT COUNT(*) FROM message_bodies`).Scan(&bodies)
		if bodies != 0 {
			t.Fatalf("bodies=%d", bodies)
		}
		var msgs int
		_ = e.db.Read.QueryRow(`SELECT COUNT(*) FROM messages WHERE id=?`, "f18").Scan(&msgs)
		if msgs != 0 {
			t.Fatalf("messages=%d want 0", msgs)
		}
		var keys int
		_ = e.db.Read.QueryRow(`SELECT COUNT(*) FROM send_keys WHERE msg_id=?`, "f18").Scan(&keys)
		if keys != 1 {
			t.Fatalf("send_keys=%d", keys)
		}
		// 防重：消息已删 → not_found
		_, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req)
		if protoCode(err) != protocol.CodeNotFound {
			t.Fatalf("want not_found got %v", err)
		}
	})

	t.Run("F11_dispatch_due_after_downtime", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		e.online("bob", "c-bob")
		delay := int64(60_000)
		req := baseSend("due1", "bob")
		req.DelayMs = &delay
		ctx := context.Background()
		res, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req)
		if err != nil {
			t.Fatal(err)
		}
		if res.State != StateScheduled {
			t.Fatalf("%s", res.State)
		}
		e.setNow(e.nowMs + 60_000)
		n, err := e.app.DispatchDue(ctx, e.nowMs, 10)
		if err != nil {
			t.Fatal(err)
		}
		if n != 1 {
			t.Fatalf("dispatched=%d", n)
		}
		st, _ := e.msgState("alice", "due1")
		if st != StateDispatched {
			t.Fatalf("%s", st)
		}
	})

	t.Run("F10_recover_clears_pushed_and_extends_grace", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		e.online("bob", "c-bob")
		ctx := context.Background()
		req := baseSend("rec1", "bob")
		req.Offline = keepTrue()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
			t.Fatal(err)
		}
		if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
			t.Fatal(err)
		}
		seq := e.seqOf("alice", "rec1")
		// 模拟保留期在停机期间已过
		_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
			_, err := tx.Exec(`UPDATE deliveries SET expire_at=?, pushed_conn=? WHERE seq=?`, e.nowMs-1000, "old-conn", seq)
			return err
		})
		_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
			_, err := tx.Exec(`UPDATE endpoints SET online_since=?, offline_since=NULL WHERE id=?`, e.nowMs-5000, "bob")
			return err
		})
		start := e.nowMs + 1_000
		e.setNow(start)
		if err := e.app.RecoverOnStart(ctx); err != nil {
			t.Fatal(err)
		}
		var pushed sql.NullString
		var exp sql.NullInt64
		_ = e.db.Read.QueryRow(`SELECT pushed_conn, expire_at FROM deliveries WHERE seq=?`, seq).Scan(&pushed, &exp)
		if pushed.Valid {
			t.Fatalf("pushed_conn still set: %v", pushed.String)
		}
		wantMin := start + 60_000
		if !exp.Valid || exp.Int64 < wantMin {
			t.Fatalf("expire_at=%v want >= %d", exp, wantMin)
		}
		var offline sql.NullInt64
		_ = e.db.Read.QueryRow(`SELECT offline_since FROM endpoints WHERE id=?`, "bob").Scan(&offline)
		if !offline.Valid || offline.Int64 != start {
			t.Fatalf("offline_since=%v want %d", offline, start)
		}
	})

	t.Run("disabled_target_rejected", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		delay := int64(5_000)
		req := baseSend("dis1", "bob")
		req.DelayMs = &delay
		ctx := context.Background()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
			t.Fatal(err)
		}
		_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
			_, err := tx.Exec(`UPDATE endpoints SET enabled=0 WHERE id=?`, "bob")
			return err
		})
		e.setNow(e.nowMs + 5_000)
		if _, err := e.app.DispatchDue(ctx, e.nowMs, 10); err != nil {
			t.Fatal(err)
		}
		seq := e.seqOf("alice", "dis1")
		st, reason := e.deliveryState(seq, "bob")
		if st != DeliveryRejected || reason != ReasonEndpointDisabled {
			t.Fatalf("%s/%s", st, reason)
		}
	})

	t.Run("queue_full_rejected", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, func(l *Limits) { l.MaxPendingPerReceiver = 1 })
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		e.online("bob", "c-bob")
		ctx := context.Background()
		req1 := baseSend("qf1", "bob")
		req1.Offline = keepTrue()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req1); err != nil {
			t.Fatal(err)
		}
		req2 := baseSend("qf2", "bob")
		req2.Offline = keepTrue()
		res, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req2)
		if err != nil {
			t.Fatal(err)
		}
		if res.State != StateCompleted {
			t.Fatalf("state=%s", res.State)
		}
		seq := e.seqOf("alice", "qf2")
		st, reason := e.deliveryState(seq, "bob")
		if st != DeliveryRejected || reason != ReasonQueueFull {
			t.Fatalf("%s/%s", st, reason)
		}
	})

	t.Run("sender_left_no_recipients", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		ctx := context.Background()
		_ = e.db.Queue.Do(ctx, func(tx *sql.Tx) error {
			_, err := tx.Exec(`INSERT INTO groups(id,name,owner_id,created_at) VALUES('g2','g','alice',?)`, e.nowMs)
			if err != nil {
				return err
			}
			_, err = tx.Exec(`INSERT INTO group_members(group_id,endpoint_id,joined_at) VALUES('g2','alice',?)`, e.nowMs)
			return err
		})
		delay := int64(1000)
		req := &protocol.Send{
			V: protocol.Version, Type: protocol.TypeSend, RID: "1", ID: "nr1",
			To:   protocol.Target{Kind: protocol.TargetGroup, ID: "g2"},
			Body: protocol.Body{Enc: protocol.EncUTF8, Data: "x"}, DelayMs: &delay,
		}
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
			t.Fatal(err)
		}
		e.setNow(e.nowMs + 1000)
		if _, err := e.app.DispatchDue(ctx, e.nowMs, 10); err != nil {
			t.Fatal(err)
		}
		st, reason := e.msgState("alice", "nr1")
		if st != StateCompleted || reason != ReasonNoRecipients {
			t.Fatalf("%s/%s", st, reason)
		}
	})

	t.Run("too_large_rejected", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		e.conns.Set("bob", LiveConn{ConnID: "c-bob", MaxReceiveBytes: 50, Ready: true})
		ctx := context.Background()
		req := baseSend("big1", "bob")
		req.Body.Data = string(make([]byte, 200))
		for i := range req.Body.Data {
			// utf8 valid
			_ = i
		}
		req.Body.Data = "{\"x\":\"" + string(make([]byte, 80)) + "\"}"
		// simpler: long ascii
		b := make([]byte, 80)
		for i := range b {
			b[i] = 'a'
		}
		req.Body.Data = string(b)
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
			t.Fatal(err)
		}
		if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
			t.Fatal(err)
		}
		seq := e.seqOf("alice", "big1")
		st, reason := e.deliveryState(seq, "bob")
		if st != DeliveryRejected || reason != ReasonTooLarge {
			t.Fatalf("%s/%s", st, reason)
		}
	})

	t.Run("on_publish_dropped_clears_mark", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		e.online("bob", "c-bob")
		ctx := context.Background()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("drop1", "bob")); err != nil {
			t.Fatal(err)
		}
		if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
			t.Fatal(err)
		}
		payload := e.down.Snapshots()[0].Payload
		if err := e.app.OnPublishDropped(ctx, "bob", "c-bob", payload); err != nil {
			t.Fatal(err)
		}
		seq := e.seqOf("alice", "drop1")
		var pushed sql.NullString
		_ = e.db.Read.QueryRow(`SELECT pushed_conn FROM deliveries WHERE seq=?`, seq).Scan(&pushed)
		if pushed.Valid {
			t.Fatalf("still pushed %s", pushed.String)
		}
	})

	t.Run("ack_duplicate_and_terminal", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		e.online("bob", "c-bob")
		ctx := context.Background()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("ack1", "bob")); err != nil {
			t.Fatal(err)
		}
		_ = e.app.PushPending(ctx, "bob", "c-bob")
		r1, err := e.app.Ack(ctx, "bob", &protocol.Ack{V: 1, Type: protocol.TypeAck, RID: "1", From: "alice", ID: "ack1"})
		if err != nil {
			t.Fatal(err)
		}
		if r1.Result != DeliveryAccepted {
			t.Fatalf("%+v", r1)
		}
		r2, err := e.app.Ack(ctx, "bob", &protocol.Ack{V: 1, Type: protocol.TypeAck, RID: "2", From: "alice", ID: "ack1"})
		if err != nil {
			t.Fatal(err)
		}
		if r2.Result != DeliveryAccepted {
			t.Fatalf("dup %+v", r2)
		}
	})

	t.Run("disconnect_extends_grace", func(t *testing.T) {
		t.Parallel()
		e := openDeliveryEnv(t, nil)
		insertEndpoint(t, e.db, "alice", "", 1, 0)
		insertEndpoint(t, e.db, "bob", "", 1, 0)
		e.online("bob", "c-bob")
		ctx := context.Background()
		if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("dc1", "bob")); err != nil {
			t.Fatal(err)
		}
		_ = e.app.PushPending(ctx, "bob", "c-bob")
		if err := e.app.OnDisconnect(ctx, "bob", "c-bob", true); err != nil {
			t.Fatal(err)
		}
		e.conns.Clear("bob", "c-bob")
		seq := e.seqOf("alice", "dc1")
		var pushed sql.NullString
		var exp sql.NullInt64
		_ = e.db.Read.QueryRow(`SELECT pushed_conn, expire_at FROM deliveries WHERE seq=?`, seq).Scan(&pushed, &exp)
		if pushed.Valid {
			t.Fatal("pushed should clear")
		}
		if !exp.Valid || exp.Int64 != e.nowMs+60_000 {
			t.Fatalf("expire_at=%v", exp)
		}
	})
}

func TestRecallDoesNotWriteReceipt(t *testing.T) {
	t.Parallel()
	e := openDeliveryEnv(t, nil)
	insertEndpoint(t, e.db, "alice", "", 1, 0)
	insertEndpoint(t, e.db, "bob", "", 1, 0)
	ctx := context.Background()
	req := baseSend("nr", "bob")
	req.Offline = keepTrue()
	if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
		t.Fatal(err)
	}
	if _, err := e.app.Recall(ctx, "alice", &protocol.Recall{V: 1, Type: protocol.TypeRecall, RID: "1", ID: "nr"}); err != nil {
		t.Fatal(err)
	}
	var n int
	_ = e.db.Read.QueryRow(`SELECT COUNT(*) FROM receipts WHERE msg_id=?`, "nr").Scan(&n)
	if n != 0 {
		t.Fatalf("receipts=%d", n)
	}
}

func TestPushRevokedOnRecallAfterPush(t *testing.T) {
	t.Parallel()
	e := openDeliveryEnv(t, nil)
	insertEndpoint(t, e.db, "alice", "", 1, 0)
	insertEndpoint(t, e.db, "bob", "", 1, 0)
	e.online("bob", "c-bob")
	ctx := context.Background()
	if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, baseSend("rv1", "bob")); err != nil {
		t.Fatal(err)
	}
	_ = e.app.PushPending(ctx, "bob", "c-bob")
	if _, err := e.app.Recall(ctx, "alice", &protocol.Recall{V: 1, Type: protocol.TypeRecall, RID: "1", ID: "rv1"}); err != nil {
		t.Fatal(err)
	}
	found := false
	for _, p := range e.down.Snapshots() {
		var head struct {
			Type string `json:"type"`
		}
		_ = json.Unmarshal(p.Payload, &head)
		if head.Type == protocol.TypeRevoked {
			found = true
		}
	}
	if !found {
		t.Fatal("expected revoked frame")
	}
}

func TestPushPreservesLargeMetaInteger(t *testing.T) {
	t.Parallel()
	raw := `{"id":12345678901234567890}`
	decoded := decodeMetaJSON(raw)
	n, ok := decoded["id"].(json.Number)
	if !ok || n.String() != "12345678901234567890" {
		t.Fatalf("decode meta=%v", decoded)
	}

	e := openDeliveryEnv(t, nil)
	insertEndpoint(t, e.db, "alice", "", 1, 0)
	insertEndpoint(t, e.db, "bob", "", 1, 0)
	e.online("bob", "c-bob")
	ctx := context.Background()
	req := baseSend("meta-big", "bob")
	req.Meta = map[string]any{"id": json.Number("12345678901234567890")}
	if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
		t.Fatal(err)
	}
	if err := e.app.PushPending(ctx, "bob", "c-bob"); err != nil {
		t.Fatal(err)
	}
	found := false
	for _, p := range e.down.Snapshots() {
		if bytes.Contains(p.Payload, []byte("12345678901234567890")) {
			found = true
			break
		}
	}
	if !found {
		t.Fatalf("push payloads missing large int: %v", e.down.Snapshots())
	}
}