package outbox

import (
	"context"
	"encoding/binary"
	"errors"
	"io"
	"net"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/go-sql-driver/mysql"
)

// commitLossGateway 只用于无TLS的隔离MySQL回归；捕获COMMIT的真实成功包后丢弃并断连。
// 不伪造SQL结果，也不修改数据库中的提交状态。
type commitLossGateway struct {
	listener    net.Listener
	dropped     chan struct{}
	armed       atomic.Bool
	mu          sync.Mutex
	closing     bool
	connections map[net.Conn]struct{}
	wg          sync.WaitGroup
	accepted    chan struct{}
}

func newCommitLossGateway(t *testing.T, target string) *commitLossGateway {
	t.Helper()
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	g := &commitLossGateway{listener: listener, dropped: make(chan struct{}, 1), connections: map[net.Conn]struct{}{}, accepted: make(chan struct{})}
	go func() {
		defer close(g.accepted)
		for {
			down, err := listener.Accept()
			if err != nil {
				return
			}
			g.wg.Add(1)
			go func() {
				defer g.wg.Done()
				defer down.Close()
				up, err := net.DialTimeout("tcp", target, 3*time.Second)
				if err != nil {
					return
				}
				defer up.Close()
				g.mu.Lock()
				if g.closing {
					g.mu.Unlock()
					return
				}
				g.connections[down] = struct{}{}
				g.connections[up] = struct{}{}
				g.mu.Unlock()
				defer func() { g.mu.Lock(); delete(g.connections, down); delete(g.connections, up); g.mu.Unlock() }()
				var committing atomic.Bool
				done := make(chan struct{}, 2)
				go func() {
					defer func() { done <- struct{}{} }()
					for {
						packet, err := readMySQLPacket(down)
						if err != nil {
							return
						}
						if packet[3] == 0 && len(packet) > 4 && packet[4] == 3 {
							committing.Store(strings.EqualFold(strings.TrimSpace(string(packet[5:])), "COMMIT"))
						}
						if writeMySQLPacket(up, packet) != nil {
							return
						}
					}
				}()
				go func() {
					defer func() { done <- struct{}{} }()
					for {
						packet, err := readMySQLPacket(up)
						if err != nil {
							return
						}
						if packet[3] == 1 && committing.Swap(false) && len(packet) > 4 && packet[4] == 0 && g.armed.CompareAndSwap(true, false) {
							select {
							case g.dropped <- struct{}{}:
							default:
								t.Error("unconsumed COMMIT fault marker")
							}
							return
						}
						if writeMySQLPacket(down, packet) != nil {
							return
						}
					}
				}()
				<-done
				down.Close()
				up.Close()
				<-done
			}()
		}
	}()
	t.Cleanup(func() {
		listener.Close()
		<-g.accepted
		g.mu.Lock()
		g.closing = true
		for c := range g.connections {
			c.Close()
		}
		g.mu.Unlock()
		g.wg.Wait()
	})
	return g
}
func readMySQLPacket(r io.Reader) ([]byte, error) {
	var header [4]byte
	if _, err := io.ReadFull(r, header[:]); err != nil {
		return nil, err
	}
	size := int(binary.LittleEndian.Uint32(header[:]) & 0x00ffffff)
	packet := make([]byte, 4+size)
	copy(packet, header[:])
	_, err := io.ReadFull(r, packet[4:])
	return packet, err
}
func writeMySQLPacket(w io.Writer, b []byte) error {
	for len(b) > 0 {
		n, err := w.Write(b)
		if err != nil {
			return err
		}
		if n == 0 {
			return io.ErrShortWrite
		}
		b = b[n:]
	}
	return nil
}

func TestIntegrationSenderCommitResponseLoss(t *testing.T) {
	direct, dsn := integrationDB(t)
	cfg, err := mysql.ParseDSN(dsn)
	if err != nil || cfg.TLSConfig != "" {
		t.Fatal("fault requires isolated plaintext MySQL")
	}
	gateway := newCommitLossGateway(t, cfg.Addr)
	cfg.Addr = gateway.listener.Addr().String()
	ctx, stop := context.WithTimeout(context.Background(), 30*time.Second)
	defer stop()
	proxied, err := Open(ctx, cfg.FormatDSN())
	if err != nil {
		t.Fatal(err)
	}
	defer proxied.Close()
	for _, phase := range []string{"enqueue", "claim", "confirm"} {
		t.Run(phase, func(t *testing.T) {
			s := testStore(t, proxied)
			fresh, _ := businessStore(direct, s.namespace, s.queue)
			id := testID(t)
			in := testRequest(s.queue)
			if phase == "enqueue" {
				gateway.armed.Store(true)
				if _, err = s.Enqueue(ctx, id, "objects/test", in); !errors.Is(err, ErrCommitUnconfirmed) {
					t.Fatal("enqueue reported confirmed", err)
				}
				dup, e := fresh.Enqueue(ctx, id, "objects/test", in)
				if e != nil || !dup || countRows(t, fresh, "sender_jobs") != 1 {
					t.Fatal("local commit not reconciled", e)
				}
			} else {
				if _, err = s.Enqueue(ctx, id, "objects/test", in); err != nil {
					t.Fatal(err)
				}
				if phase == "claim" {
					gateway.armed.Store(true)
				}
				claim, e := s.claim(ctx)
				if phase == "claim" {
					if !errors.Is(e, ErrCommitUnconfirmed) {
						t.Fatal("claim reported confirmed", e)
					}
					r := status(t, fresh, id)
					if r.State != "SENDING" || r.SendCount != 1 {
						t.Fatal("claim was not actually committed")
					}
					execSQL(t, direct, `UPDATE sender_outbox SET wake_at=TIMESTAMPADD(SECOND,-1,UTC_TIMESTAMP(6)) WHERE namespace_id=? AND operation_id=?`, s.namespace, id)
					r, e = fresh.claim(ctx)
					if e != nil || r.SendCount != 2 || r.IdempotencyKey != "outbox-task-"+id {
						t.Fatal("unknown claim did not recover", e)
					}
				} else {
					if e != nil {
						t.Fatal(e)
					}
					receipt := fakeReceipt(t, s)
					gateway.armed.Store(true)
					if e = s.finish(ctx, claim, "ACCEPTED", "ACCEPTED", 0, &receipt); !errors.Is(e, ErrCommitUnconfirmed) {
						t.Fatal("confirmation reported confirmed", e)
					}
					r := status(t, fresh, id)
					if r.State != "ACCEPTED" || r.TaskID == nil || *r.TaskID != receipt.Task.ID {
						t.Fatal("confirmed commit not visible")
					}
					if _, e = fresh.claim(ctx); !errors.Is(e, ErrIdle) {
						t.Fatal("confirmed item eligible for resend", e)
					}
					var task string
					if e = direct.QueryRowContext(ctx, `SELECT task_id FROM sender_jobs WHERE namespace_id=? AND operation_id=?`, s.namespace, id).Scan(&task); e != nil || task != receipt.Task.ID {
						t.Fatal("business confirmation not atomic", e)
					}
				}
			}
			select {
			case <-gateway.dropped:
			case <-ctx.Done():
				t.Fatal("actual COMMIT OK not intercepted")
			}
		})
	}
	t.Log("real COMMIT OK discarded for business enqueue, sender claim and local acceptance; fresh stores reconciled persisted identity and state")
}