package orbit

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync/atomic"
	"testing"
	"time"
)

// orbitTestProcess 持有真实子进程，必须 Wait 确认退出才算完成故障注入。
type orbitTestProcess struct {
	cmd         *exec.Cmd
	done        chan struct{}
	waitErr     error
	url, nodeID string
}

func processEnv(overrides map[string]string) []string {
	env := make([]string, 0, len(os.Environ())+len(overrides))
	for _, entry := range os.Environ() {
		key, _, _ := strings.Cut(entry, "=")
		if _, replace := overrides[key]; !replace {
			env = append(env, entry)
		}
	}
	for k, v := range overrides {
		env = append(env, k+"="+v)
	}
	return env
}
func launchTestProcess(t *testing.T, cmd *exec.Cmd) *orbitTestProcess {
	t.Helper()
	p := &orbitTestProcess{cmd: cmd, done: make(chan struct{})}
	if err := cmd.Start(); err != nil {
		t.Fatal(err)
	}
	go func() { p.waitErr = cmd.Wait(); close(p.done) }()
	t.Cleanup(func() {
		select {
		case <-p.done:
			return
		default:
		}
		_ = p.cmd.Process.Kill()
		select {
		case <-p.done:
		case <-time.After(5 * time.Second):
			t.Error("owned child process did not exit")
		}
	})
	return p
}
func (p *orbitTestProcess) kill(t *testing.T) {
	t.Helper()
	select {
	case <-p.done:
		t.Fatal("fault target exited before injection")
	default:
	}
	if err := p.cmd.Process.Kill(); err != nil {
		t.Fatal(err)
	}
	select {
	case <-p.done:
		var exit *exec.ExitError
		if !errors.As(p.waitErr, &exit) || exit.ExitCode() != -1 {
			t.Fatal("owned process was not terminated by a signal")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("killed process did not exit")
	}
}
func startOrbitProcess(t *testing.T) *orbitTestProcess {
	t.Helper()
	binary, dsn := os.Getenv("ORBIT_TEST_SERVER_BINARY"), os.Getenv("ORBIT_TEST_MYSQL_DSN")
	if binary == "" || dsn == "" {
		t.Skip("requires isolated database and compiled Orbit binary")
	}
	if !filepath.IsAbs(binary) || !strings.Contains(dsn, "@tcp(127.0.0.1:") {
		t.Fatal("process faults require disposable local resources")
	}
	cmd := exec.Command(binary)
	cmd.Env = processEnv(map[string]string{"ORBIT_MYSQL_DSN": dsn, "ORBIT_HTTP_ADDR": "127.0.0.1:0"})
	output, err := cmd.StdoutPipe()
	if err != nil {
		t.Fatal(err)
	}
	cmd.Stderr = io.Discard
	address := make(chan [2]string, 1)
	go func() {
		scanner := bufio.NewScanner(output)
		for scanner.Scan() {
			var line struct {
				Address string `json:"address"`
				Node    string `json:"node_id"`
			}
			if json.Unmarshal(scanner.Bytes(), &line) == nil && strings.HasPrefix(line.Address, "127.0.0.1:") && validID(line.Node) {
				select {
				case address <- [2]string{line.Address, line.Node}:
				default:
				}
			}
		}
	}()
	p := launchTestProcess(t, cmd)
	select {
	case value := <-address:
		p.url = "http://" + value[0]
		p.nodeID = value[1]
	case <-p.done:
		t.Fatal("isolated Orbit exited during startup")
	case <-time.After(10 * time.Second):
		t.Fatal("isolated Orbit did not publish its bound address")
	}
	return p
}
func processClient(t *testing.T, p *orbitTestProcess, token string) *Client {
	t.Helper()
	c, err := NewClient(p.url, Options{Token: token})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	if _, err := c.Readiness(ctx); err != nil {
		t.Fatal(err)
	}
	return c
}
func awaitProcessDrop(t *testing.T, ctx context.Context, g *workerFaultGateway, operation string) {
	t.Helper()
	select {
	case got := <-g.dropped:
		if got != operation {
			t.Fatal("wrong process fault boundary")
		}
	case <-ctx.Done():
		t.Fatal("process fault boundary not reached")
	}
}

// 两次在真实数据库提交后丢 ACK 并强杀节点，由另外的真实进程核对原操作。
func TestIntegrationWorkerProcessACKRecovery(t *testing.T) {
	a := startOrbitProcess(t)
	_, queue, token := integrationWorkerClient(t)
	apiA := processClient(t, a, token)
	b := startOrbitProcess(t)
	apiB := processClient(t, b, token)
	if a.nodeID == b.nodeID {
		t.Fatal("server processes reused node identity")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
	defer cancel()
	g := newWorkerFaultGateway(t, a.url)
	g.dropOffline.Store(true)
	g.drop.Store("task.start")
	client, err := NewClient(g.server.URL, Options{Token: token, HTTPClient: g.server.Client()})
	if err != nil {
		t.Fatal(err)
	}
	name := "process.ack." + workerTestID(t)
	entered, release := make(chan WorkerExecution, 2), make(chan struct{})
	var called atomic.Int32
	w, err := NewWorker(client, WorkerOptions{QueueID: queue, Capacity: 1, ReconnectMaxDelay: time.Second, Handlers: []WorkerHandler{runtimeHandler(name, func(work context.Context, e WorkerExecution) (json.RawMessage, error) {
		called.Add(1)
		entered <- e
		select {
		case <-release:
			return json.RawMessage(`{"value":9007199254740993}`), nil
		case <-work.Done():
			return nil, work.Err()
		}
	})}})
	if err != nil {
		t.Fatal(err)
	}
	task := runtimeTask(t, ctx, apiA, queue, name, 30000)
	run, stop := context.WithCancel(ctx)
	defer stop()
	done := make(chan error, 1)
	go func() { done <- w.Run(run) }()
	awaitProcessDrop(t, ctx, g, "task.start")
	if called.Load() != 0 {
		t.Fatal("handler ran before recovering lost start confirmation")
	}
	before, err := apiB.ListTaskAttempts(ctx, task.ID, HistoryPageRequest{})
	if err != nil || len(before.Items) != 1 || before.Items[0].AuthorizedAt == nil {
		t.Fatal("start was not durably committed before kill")
	}
	initial, err := apiB.GetTaskAttempt(ctx, task.ID, before.Items[0].ID)
	if err != nil {
		t.Fatal(err)
	}
	a.kill(t)
	g.target.Store(b.url)
	g.offline.Store(false)
	var execution WorkerExecution
	select {
	case execution = <-entered:
	case err := <-done:
		t.Fatalf("Worker stopped during node recovery: %v", err)
	case <-ctx.Done():
		t.Fatal("handler did not start after failover")
	}
	moved, err := apiB.GetTaskAttempt(ctx, task.ID, execution.AttemptID)
	if err != nil || moved.ID != initial.ID || moved.SessionID == initial.SessionID || moved.WorkerInstanceID != initial.WorkerInstanceID || moved.AuthorizedAt == nil || !moved.AuthorizedAt.Equal(*initial.AuthorizedAt) {
		t.Fatal("node failover changed attempt or start identity")
	}
	g.drop.Store("task.finish")
	close(release)
	awaitProcessDrop(t, ctx, g, "task.finish")
	committed, err := apiB.GetTask(ctx, task.ID)
	if err != nil || committed.State != TaskSucceeded {
		t.Fatal("finish was not committed before second kill")
	}
	b.kill(t)
	c := startOrbitProcess(t)
	apiC := processClient(t, c, token)
	if c.nodeID == a.nodeID || c.nodeID == b.nodeID {
		t.Fatal("replacement reused node identity")
	}
	g.target.Store(c.url)
	g.offline.Store(false)
	for w.Stats().ReportsConfirmed != 1 {
		select {
		case err := <-done:
			t.Fatalf("Worker stopped before saved result confirmation: %v", err)
		case <-ctx.Done():
			t.Fatal("saved result was not recovered")
		case <-time.After(20 * time.Millisecond):
		}
	}
	stop()
	select {
	case err := <-done:
		if err != nil {
			t.Fatal(err)
		}
	case <-ctx.Done():
		t.Fatal("Worker did not drain after failover")
	}
	current, err := apiC.GetTask(ctx, task.ID)
	if err != nil || current.State != TaskSucceeded || current.AttemptCount != 1 || current.Revision != committed.Revision || string(current.Result) != `{"value":9007199254740993}` {
		t.Fatal("lost ACK recovery changed authoritative result")
	}
	attempts, err := apiC.ListTaskAttempts(ctx, task.ID, HistoryPageRequest{})
	if err != nil || len(attempts.Items) != 1 || attempts.Items[0].ID != initial.ID {
		t.Fatal("node crash created duplicate execution")
	}
	evidence, err := apiC.ListAttemptEvidence(ctx, task.ID, initial.ID, HistoryPageRequest{})
	if err != nil || len(evidence.Items) != 0 || called.Load() != 1 || w.Stats().Started != 1 || w.Stats().Reconnects < 2 {
		t.Fatal("node recovery replayed handler or produced false evidence")
	}
}

// 此测试只供父测试启动真实 Worker 子进程，父进程负责强杀与 Wait。
func TestWorkerProcessCrashChild(t *testing.T) {
	if os.Getenv("ORBIT_TEST_WORKER_CHILD") != "1" {
		t.Skip("owned subprocess only")
	}
	api, queue, _ := integrationWorkerClient(t)
	queue = os.Getenv("ORBIT_TEST_WORKER_CHILD_QUEUE")
	if !validID(queue) {
		t.Fatal("invalid child queue")
	}
	name, marker := os.Getenv("ORBIT_TEST_WORKER_HANDLER"), os.Getenv("ORBIT_TEST_WORKER_MARKER")
	if !workerName(name, 128) || !filepath.IsAbs(marker) {
		t.Fatal("invalid child fixture")
	}
	w, err := NewWorker(api, WorkerOptions{QueueID: queue, Capacity: 1, Handlers: []WorkerHandler{runtimeHandler(name, func(ctx context.Context, e WorkerExecution) (json.RawMessage, error) {
		// 文件代表已发起的外部工作标记，不能推断为可安全重试的明确失败。
		file, err := os.OpenFile(marker, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
		if err != nil {
			return nil, err
		}
		_, err = file.WriteString(e.Task.ID)
		syncErr := file.Sync()
		closeErr := file.Close()
		if err != nil || syncErr != nil || closeErr != nil {
			return nil, errors.New("child marker failed")
		}
		<-ctx.Done()
		return nil, ctx.Err()
	})}})
	if err != nil {
		t.Fatal(err)
	}
	if err := w.Run(context.Background()); err != nil {
		t.Fatal(err)
	}
	t.Fatal("crash child unexpectedly drained")
}
func TestIntegrationWorkerProcessCrashUnknown(t *testing.T) {
	a := startOrbitProcess(t)
	_, queue, token := integrationWorkerClient(t)
	apiA := processClient(t, a, token)
	ctx, cancel := context.WithTimeout(context.Background(), 35*time.Second)
	defer cancel()
	dedicated, err := apiA.CreateQueue(ctx, "crash-queue-"+workerTestID(t), CreateQueueRequest{Name: "crash-" + workerTestID(t), MaxConcurrency: 1, RatePerSecond: 10})
	if err != nil {
		t.Fatal(err)
	}
	queue = dedicated.Queue.ID
	name := "process.crash." + workerTestID(t)
	task, err := apiA.SubmitTask(ctx, "process-crash-"+workerTestID(t), SubmitTaskRequest{QueueID: queue, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), TaskTimeoutMS: 30000, Retry: RetryPolicy{MaxAttempts: 3, AttemptTimeoutMS: 6000}, RecoverySafety: RecoveryReconcileOnly})
	if err != nil {
		t.Fatal(err)
	}
	marker := filepath.Join(t.TempDir(), "external-work-started")
	executable, err := os.Executable()
	if err != nil {
		t.Fatal(err)
	}
	cmd := exec.Command(executable, "-test.run=^TestWorkerProcessCrashChild$", "-test.timeout=30s")
	cmd.Env = processEnv(map[string]string{"ORBIT_TEST_WORKER_CHILD": "1", "ORBIT_TEST_WORKER_URL": a.url, "ORBIT_TEST_WORKER_CHILD_QUEUE": queue, "ORBIT_TEST_WORKER_HANDLER": name, "ORBIT_TEST_WORKER_MARKER": marker})
	cmd.Stdout = io.Discard
	cmd.Stderr = io.Discard
	child := launchTestProcess(t, cmd)
	for {
		raw, err := os.ReadFile(marker)
		if err == nil && string(raw) == task.Task.ID {
			break
		}
		if err != nil && !os.IsNotExist(err) {
			t.Fatal(err)
		}
		select {
		case <-child.done:
			t.Fatal("Worker child exited before starting")
		case <-ctx.Done():
			t.Fatal("Worker child did not start")
		case <-time.After(20 * time.Millisecond):
		}
	}
	running, err := apiA.GetTask(ctx, task.Task.ID)
	if err != nil || running.State != TaskRunning {
		t.Fatal("child did not receive durable start authorization")
	}
	child.kill(t)
	a.kill(t)
	b := startOrbitProcess(t)
	apiB := processClient(t, b, token)
	unknown := waitRuntimeState(t, ctx, apiB, task.Task.ID, TaskNeedsReview)
	if unknown.AttemptCount != 1 || unknown.Retry.MaxAttempts != 3 || len(unknown.Result) != 0 {
		t.Fatal("crash recovery reset budget or fabricated success")
	}
	attempts, err := apiB.ListTaskAttempts(ctx, task.Task.ID, HistoryPageRequest{})
	if err != nil || len(attempts.Items) != 1 || attempts.Items[0].State != AttemptUnknown || attempts.Items[0].EndedAt == nil {
		t.Fatal("replacement did not preserve unknown attempt")
	}
	// 相同能力的新 Worker 消费探针，原任务保持核查，验证许可已回收。
	var called atomic.Int32
	w, err := NewWorker(apiB, WorkerOptions{QueueID: queue, Capacity: 1, Handlers: []WorkerHandler{runtimeHandler(name, func(_ context.Context, e WorkerExecution) (json.RawMessage, error) {
		called.Add(1)
		if e.Task.ID == task.Task.ID {
			return nil, errors.New("unknown task was replayed")
		}
		return json.RawMessage(`{"probe":true}`), nil
	})}})
	if err != nil {
		t.Fatal(err)
	}
	probe := runtimeTask(t, ctx, apiB, queue, name, 5000)
	run, stop := context.WithCancel(ctx)
	defer stop()
	done := make(chan error, 1)
	go func() { done <- w.Run(run) }()
	waitRuntimeState(t, ctx, apiB, probe.ID, TaskSucceeded)
	stop()
	select {
	case err := <-done:
		if err != nil {
			t.Fatal(err)
		}
	case <-ctx.Done():
		t.Fatal("replacement Worker did not drain")
	}
	final, err := apiB.GetTask(ctx, task.Task.ID)
	if err != nil || final.State != TaskNeedsReview || final.AttemptCount != 1 || called.Load() != 1 {
		t.Fatal("replacement replayed unknown work")
	}
	raw, err := os.ReadFile(marker)
	if err != nil || string(raw) != task.Task.ID {
		t.Fatal("external work marker lost")
	}
}