package inbox

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"strings"
	"testing"
	"time"

	orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
)

type receiverMarker struct {
	Phase     string `json:"phase"`
	EventID   string `json:"event_id"`
	Duplicate bool   `json:"duplicate"`
}

func receiverClient(t *testing.T) *orbit.Client {
	t.Helper()
	base, path := os.Getenv("ORBIT_TEST_RECEIVER_URL"), os.Getenv("ORBIT_TEST_RECEIVER_CREDENTIALS_FILE")
	if base == "" || path == "" {
		t.Skip("requires isolated real Orbit process and receiver credential")
	}
	if !strings.HasPrefix(base, "http://127.0.0.1:") {
		t.Fatal("receiver process integration requires loopback Orbit")
	}
	raw, err := os.ReadFile(path)
	if err != nil {
		t.Fatal("cannot read test credential")
	}
	var credentials struct {
		Token string `json:"token"`
	}
	if json.Unmarshal(raw, &credentials) != nil || credentials.Token == "" {
		t.Fatal("invalid test credential")
	}
	c, err := orbit.NewClient(base, orbit.Options{Token: credentials.Token})
	if err != nil {
		t.Fatal(err)
	}
	return c
}

// 此入口只存在于测试二进制。处理器使用正式Store和Consumer，故障屏障位于Apply提交后、返回nil前。
func TestReceiverProcessHelper(t *testing.T) {
	mode := os.Getenv("ORBIT_TEST_RECEIVER_CHILD")
	if mode == "" {
		t.Skip("subprocess helper")
	}
	if mode != "hold_after_commit" && mode != "confirm_duplicate" {
		t.Fatal("invalid receiver child mode")
	}
	db, _ := integrationDB(t)
	api := receiverClient(t)
	s, err := New(db, os.Getenv("ORBIT_TEST_RECEIVER_NAME"), os.Getenv("ORBIT_TEST_RECEIVER_QUEUE"))
	if err != nil {
		t.Fatal(err)
	}
	ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
	defer stop()
	consumer, err := orbit.NewEventConsumer(api, orbit.EventConsumerOptions{SubscriptionID: os.Getenv("ORBIT_TEST_RECEIVER_SUBSCRIPTION"), Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, event orbit.ResultEvent) error {
		result, err := s.Apply(work, event)
		if err != nil {
			return err
		}
		marker, _ := json.Marshal(receiverMarker{Phase: "committed", EventID: event.ID, Duplicate: result.Duplicate})
		fmt.Println("ORBIT_RECEIVER_TEST " + string(marker))
		if mode == "hold_after_commit" {
			<-work.Done()
			return work.Err()
		}
		if !result.Duplicate {
			return errors.New("restarted receiver did not find committed inbox")
		}
		return nil
	}})
	if err != nil {
		t.Fatal(err)
	}
	run, cancel := context.WithCancel(ctx)
	defer cancel()
	done := make(chan error, 1)
	go func() { done <- consumer.Run(run) }()
	for {
		select {
		case <-ctx.Done():
			cancel()
			<-done
			t.Fatal("receiver helper timed out")
		case err := <-done:
			if err != nil {
				t.Fatal(err)
			}
			return
		case event, ok := <-consumer.Events():
			if !ok {
				if err := <-done; err != nil {
					t.Fatal(err)
				}
				return
			}
			if event.Type == "acknowledged" {
				marker, _ := json.Marshal(receiverMarker{Phase: "acknowledged", EventID: event.EventID})
				fmt.Println("ORBIT_RECEIVER_TEST " + string(marker))
				cancel()
				if err := <-done; err != nil {
					t.Fatal(err)
				}
				return
			}
		}
	}
}

type receiverChild struct {
	cmd     *exec.Cmd
	done    chan struct{}
	waitErr error
	markers chan receiverMarker
}

func receiverEnv(values map[string]string) []string {
	out := []string{}
	for _, entry := range os.Environ() {
		key, _, _ := strings.Cut(entry, "=")
		if _, found := values[key]; !found {
			out = append(out, entry)
		}
	}
	for key, value := range values {
		out = append(out, key+"="+value)
	}
	return out
}
func startReceiverChild(t *testing.T, s *Store, sub, mode string) *receiverChild {
	t.Helper()
	binary, err := os.Executable()
	if err != nil {
		t.Fatal(err)
	}
	cmd := exec.Command(binary, "-test.run=^TestReceiverProcessHelper$")
	cmd.Env = receiverEnv(map[string]string{"ORBIT_TEST_RECEIVER_CHILD": mode, "ORBIT_TEST_RECEIVER_NAME": s.consumer, "ORBIT_TEST_RECEIVER_QUEUE": s.queue, "ORBIT_TEST_RECEIVER_SUBSCRIPTION": sub})
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		t.Fatal(err)
	}
	cmd.Stderr = io.Discard
	p := &receiverChild{cmd: cmd, done: make(chan struct{}), markers: make(chan receiverMarker, 8)}
	if err = cmd.Start(); err != nil {
		t.Fatal(err)
	}
	readDone := make(chan struct{})
	go func() {
		defer close(readDone)
		scanner := bufio.NewScanner(stdout)
		for scanner.Scan() {
			line := scanner.Text()
			if !strings.HasPrefix(line, "ORBIT_RECEIVER_TEST ") {
				continue
			}
			var marker receiverMarker
			if json.Unmarshal([]byte(strings.TrimPrefix(line, "ORBIT_RECEIVER_TEST ")), &marker) == nil {
				select {
				case p.markers <- marker:
				default:
				}
			}
		}
	}()
	go func() { <-readDone; p.waitErr = cmd.Wait(); close(p.done) }()
	t.Cleanup(func() {
		select {
		case <-p.done:
			return
		default:
		}
		_ = cmd.Process.Kill()
		select {
		case <-p.done:
		case <-time.After(5 * time.Second):
			t.Error("owned receiver child failed to exit")
		}
	})
	return p
}
func (p *receiverChild) marker(t *testing.T, ctx context.Context, phase string) receiverMarker {
	t.Helper()
	for {
		select {
		case marker := <-p.markers:
			if marker.Phase == phase {
				return marker
			}
		case <-p.done:
			select {
			case marker := <-p.markers:
				if marker.Phase == phase {
					return marker
				}
			default:
			}
			t.Fatal("receiver child exited before expected boundary", p.waitErr)
		case <-ctx.Done():
			t.Fatal("receiver child did not reach " + phase)
		}
	}
}
func (p *receiverChild) kill(t *testing.T) {
	t.Helper()
	select {
	case <-p.done:
		t.Fatal("receiver exited before fault 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("receiver did not terminate by signal")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("receiver SIGKILL was not confirmed")
	}
}

func TestIntegrationReceiverProcessCommitBeforeAck(t *testing.T) {
	db, _ := integrationDB(t)
	api := receiverClient(t)
	ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
	defer stop()
	id := newID(t)
	queue, err := api.CreateQueue(ctx, "queue-"+id, orbit.CreateQueueRequest{Name: "inbox-process-" + id, MaxConcurrency: 1, RatePerSecond: 10})
	if err != nil {
		t.Fatal(err)
	}
	s, err := New(db, "restart-"+id, queue.Queue.ID)
	if err != nil {
		t.Fatal(err)
	}
	created, err := api.CreateSubscription(ctx, "sub-"+id, orbit.CreateSubscriptionRequest{Name: "inbox-process-" + id, QueueID: queue.Queue.ID, Types: []string{"task.cancelled"}, MaxPending: 1, MaxAttempts: 3})
	if err != nil {
		t.Fatal(err)
	}
	task, err := api.SubmitTask(ctx, "task-"+id, orbit.SubmitTaskRequest{QueueID: queue.Queue.ID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)})
	if err != nil {
		t.Fatal(err)
	}
	if _, err = api.CancelTask(ctx, task.Task.ID, "cancel-"+id, orbit.CancelTaskRequest{ExpectedRevision: task.Task.Revision, Reason: "receiver process fault"}); err != nil {
		t.Fatal(err)
	}
	first := startReceiverChild(t, s, created.Subscription.ID, "hold_after_commit")
	committed := first.marker(t, ctx, "committed")
	if committed.Duplicate {
		t.Fatal("first receiver unexpectedly deduplicated")
	}
	view, err := s.View(ctx, task.Task.ID)
	if err != nil || view.AppliedEvents != 1 || inboxCount(t, ctx, s) != 1 {
		t.Fatal("receiver barrier occurred before durable commit", err)
	}
	before, err := api.ListEventDeliveries(ctx, created.Subscription.ID, orbit.DeliveryPageRequest{})
	if err != nil || len(before.Items) != 1 || before.Items[0].State != orbit.DeliveryLeased || before.Items[0].AcknowledgedAt != nil {
		t.Fatal("fault boundary is not before ACK", err)
	}
	first.kill(t)
	second := startReceiverChild(t, s, created.Subscription.ID, "confirm_duplicate")
	duplicate := second.marker(t, ctx, "committed")
	ack := second.marker(t, ctx, "acknowledged")
	if !duplicate.Duplicate || duplicate.EventID != committed.EventID || ack.EventID != committed.EventID {
		t.Fatal("new process did not reconcile original event")
	}
	select {
	case <-second.done:
		if second.waitErr != nil {
			t.Fatal("replacement receiver failed", second.waitErr)
		}
	case <-ctx.Done():
		t.Fatal("replacement receiver did not exit cleanly")
	}
	after, err := s.View(ctx, task.Task.ID)
	if err != nil || after.AppliedEvents != 1 || inboxCount(t, ctx, s) != 1 {
		t.Fatal("business effect repeated after consumer restart", err)
	}
	confirmed, err := api.ListEventDeliveries(ctx, created.Subscription.ID, orbit.DeliveryPageRequest{State: orbit.DeliveryAcked})
	if err != nil || len(confirmed.Items) != 1 || confirmed.Items[0].Attempts != 2 || confirmed.Items[0].ID != before.Items[0].ID {
		t.Fatal("delivery recovery or acknowledgement history wrong", err)
	}
	t.Log("business commit verified before ACK; consumer SIGKILL confirmed, new process used durable Inbox after real lease expiry; one business effect and two delivery attempts")
}