package inbox

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 && strings.EqualFold(strings.TrimSpace(string(packet[5:])), "COMMIT") {
							committing.Store(true)
						}
						if writeMySQLPacket(up, packet) != nil {
							return
						}
					}
				}()
				go func() {
					defer func() { done <- struct{}{} }()
					for {
						packet, err := readMySQLPacket(up)
						if err != nil {
							return
						}
						if committing.Load() && packet[3] == 1 && len(packet) > 4 && packet[4] == 0 && g.armed.CompareAndSwap(true, false) {
							g.dropped <- struct{}{}
							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 TestIntegrationReceiverCommitResponseLoss(t *testing.T) {
	direct, dsn := integrationDB(t)
	cfg, err := mysql.ParseDSN(dsn)
	if err != nil || cfg.TLSConfig != "" {
		t.Fatal("commit fault requires isolated plaintext MySQL")
	}
	gateway := newCommitLossGateway(t, cfg.Addr)
	cfg.Addr = gateway.listener.Addr().String()
	ctx, stop := context.WithTimeout(context.Background(), 15*time.Second)
	defer stop()
	proxied, err := Open(ctx, cfg.FormatDSN())
	if err != nil {
		t.Fatal(err)
	}
	defer proxied.Close()
	s := testStore(t, proxied)
	event := testEvent(t, s.queue)
	gateway.armed.Store(true)
	if _, err = s.Apply(ctx, event); !errors.Is(err, ErrCommitUnconfirmed) {
		t.Fatal("lost COMMIT response acknowledged", err)
	}
	select {
	case <-gateway.dropped:
	case <-ctx.Done():
		t.Fatal("real COMMIT success was not intercepted")
	}
	restarted, err := New(direct, s.consumer, s.queue)
	if err != nil {
		t.Fatal(err)
	}
	view, err := restarted.View(ctx, event.AggregateID)
	if err != nil || view.AppliedEvents != 1 || inboxCount(t, ctx, restarted) != 1 {
		t.Fatal("expected actual commit before response loss", err)
	}
	result, err := restarted.Apply(ctx, event)
	if err != nil || !result.Duplicate || !result.Applied {
		t.Fatal("unknown commit not reconciled from durable identity", err)
	}
	after, err := restarted.View(ctx, event.AggregateID)
	if err != nil || after.AppliedEvents != 1 {
		t.Fatal("unknown commit repeated business effect", err)
	}
	t.Log("proxy observed actual MySQL COMMIT OK and discarded it; caller returned unconfirmed, fresh store deduplicated the already committed event")
}