//go:build runtime_worker_integration

package runtimeworker

import (
	"context"
	"io"
	"net"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/nexuscrm/audit"
	"github.com/nexuscrm/persistence"
)

// Both outcomes of a lost COMMIT acknowledgement must converge without a new
// retry mechanism: TiDB either committed the effect and consumed the outbox,
// or rolled back and left the durable lease available for recovery.
func TestAuditWorkerRealTiDBRecoversSilentCommit(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Fatal("declare a local TiDB environment for the network fault test")
	}
	for _, committed := range []bool{false, true} {
		name := "before_commit"
		if committed {
			name = "after_commit_before_ack"
		}
		t.Run(name, func(t *testing.T) {
			db, cleanup := openWorkerScratchDB(t)
			t.Cleanup(cleanup)
			control, err := persistence.NewRuntimeWorkOwnershipControl(db)
			if err != nil {
				t.Fatal(err)
			}
			owner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
				OperationID: "activate-audit-network-test", WorkFamily: audit.AuditHistoryFanoutWorkFamily,
				DeploymentID: "audit-network-test", WorkerReleaseDigest: integrationWorkerDigest,
			})
			first := insertAuditOutbox(t, db, "silent-commit")
			port := os.Getenv("TIDB_PORT")
			if port == "" {
				port = "4000"
			}
			proxy := newAuditCommitProxy(t, net.JoinHostPort(os.Getenv("TIDB_HOST"), port), committed)
			host, port, err := net.SplitHostPort(proxy.address)
			if err != nil {
				t.Fatal(err)
			}
			t.Setenv("TIDB_HOST", host)
			t.Setenv("TIDB_PORT", port)
			process, workerDB, err := assembleProduction(context.Background(), integrationWorkerSourceRevision,
				integrationWorkerConfig(owner), integrationPermissionStateFactory(t))
			if err != nil {
				t.Fatal(err)
			}
			t.Cleanup(func() { stopWorker(t, process, workerDB) })
			// Assembly has claimed the initial event. The next COMMIT belongs to
			// its fanout, so the injected fault crosses the real effect boundary.
			proxy.armed.Store(true)
			if err := process.Start(); err != nil {
				t.Fatal(err)
			}
			select {
			case ok := <-proxy.intercepted:
				if !ok {
					t.Fatal("TiDB did not acknowledge the intercepted commit successfully")
				}
			case <-time.After(3 * time.Second):
				t.Fatal("fanout did not reach the COMMIT fault")
			}
			if committed {
				assertAuditCounts(t, db, first, 0, 1)
			} else {
				assertAuditCounts(t, db, first, 1, 0)
			}
			second := insertAuditOutbox(t, db, "after-silent-commit")
			select {
			case <-proxy.disconnected:
			case <-time.After(8 * time.Second):
				t.Fatal("audit worker kept the silent connection beyond its network budget")
			}
			waitForAuditCounts(t, db, first, 0, 1)
			waitForAuditCounts(t, db, second, 0, 1)
			// Stop the consumer before the final independent count. No duplicate
			// effect and no stranded outbox row may survive either fault window.
			ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
			defer cancel()
			if err := process.Stop(ctx); err != nil {
				t.Fatal(err)
			}
			assertAuditCounts(t, db, first, 0, 1)
			assertAuditCounts(t, db, second, 0, 1)
		})
	}
}

type auditCommitProxy struct {
	address      string
	armed        atomic.Bool
	intercepted  chan bool
	disconnected chan struct{}
}

// Forward real unencrypted MySQL packets to TiDB, withholding one COMMIT
// acknowledgement. No SQL results, locks, or durable outcomes are simulated.
func newAuditCommitProxy(t *testing.T, upstreamAddress string, committed bool) *auditCommitProxy {
	return newWorkerSQLResponseProxy(t, upstreamAddress, "COMMIT", committed)
}

// Withhold one response from the actual database. Suppressing the request
// first models a disconnect before the server effect; forwarding it models
// the separate committed-but-unacknowledged outcome.
func newWorkerSQLResponseProxy(t *testing.T, upstreamAddress, statement string, forward bool) *auditCommitProxy {
	t.Helper()
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	proxy := &auditCommitProxy{
		address: listener.Addr().String(), intercepted: make(chan bool, 1), disconnected: make(chan struct{}),
	}
	var workers sync.WaitGroup
	workers.Add(1)
	go func() {
		defer workers.Done()
		for {
			client, err := listener.Accept()
			if err != nil {
				return
			}
			workers.Add(1)
			go func() {
				defer workers.Done()
				defer client.Close()
				upstream, err := net.DialTimeout("tcp", upstreamAddress, time.Second)
				if err != nil {
					return
				}
				_ = client.SetDeadline(time.Now().Add(time.Minute))
				_ = upstream.SetDeadline(time.Now().Add(time.Minute))
				var silent atomic.Bool
				forwarded := make(chan struct{})
				go func() {
					defer close(forwarded)
					for {
						packet, err := readAuditProxyPacket(upstream)
						if err != nil {
							if !silent.Load() {
								_ = client.Close()
							}
							return
						}
						if silent.Load() {
							proxy.intercepted <- len(packet) > 4 && packet[4] != 0xff // server did not return an error
							return
						}
						if _, err := client.Write(packet); err != nil {
							return
						}
					}
				}()
				for {
					packet, err := readAuditProxyPacket(client)
					if err != nil {
						break
					}
					if len(packet) > 5 && packet[4] == 3 && strings.EqualFold(string(packet[5:]), statement) && proxy.armed.CompareAndSwap(true, false) {
						silent.Store(true)
						if !forward {
							_ = upstream.Close() // real rollback on disconnect
							proxy.intercepted <- true
							continue
						}
					}
					if _, err := upstream.Write(packet); err != nil {
						break
					}
				}
				_ = upstream.Close()
				<-forwarded
				if silent.Load() {
					close(proxy.disconnected)
				}
			}()
		}
	}()
	t.Cleanup(func() {
		_ = listener.Close()
		workers.Wait()
	})
	return proxy
}

func readAuditProxyPacket(conn net.Conn) ([]byte, error) {
	header := make([]byte, 4)
	if _, err := io.ReadFull(conn, header); err != nil {
		return nil, err
	}
	size := int(header[0]) | int(header[1])<<8 | int(header[2])<<16
	packet := make([]byte, 4+size)
	copy(packet, header)
	_, err := io.ReadFull(conn, packet[4:])
	return packet, err
}