package main

import (
	"context"
	"encoding/json"
	"errors"
	"testing"

	"github.com/segmentio/kafka-go"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)

func TestBridgeBatchWritesKafkaThenAcks(t *testing.T) {
	env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"}
	payload, err := json.Marshal(env)
	if err != nil {
		t.Fatal(err)
	}
	writer := &recordingBridgeWriter{}
	acked := 0

	err = bridgeBatch(context.Background(), writer, []bridgeMessage{
		{subject: "vehicle.raw.jt808.v1", data: payload, ack: func() error {
			acked++
			return nil
		}},
	}, map[string]string{"vehicle.raw.jt808.v1": "vehicle.raw.jt808.v1"})
	if err != nil {
		t.Fatalf("bridgeBatch() error = %v", err)
	}
	if len(writer.messages) != 1 {
		t.Fatalf("kafka writes = %d, want 1", len(writer.messages))
	}
	if got, want := writer.messages[0].Topic, "vehicle.raw.jt808.v1"; got != want {
		t.Fatalf("topic = %q, want %q", got, want)
	}
	if got, want := string(writer.messages[0].Key), "JT808:13307795425"; got != want {
		t.Fatalf("key = %q, want %q", got, want)
	}
	if acked != 1 {
		t.Fatalf("acks = %d, want 1", acked)
	}
}

func TestBridgeBatchDoesNotAckWhenKafkaFails(t *testing.T) {
	writer := &recordingBridgeWriter{err: errors.New("kafka unavailable")}
	acked := 0

	err := bridgeBatch(context.Background(), writer, []bridgeMessage{
		{subject: "vehicle.event.unified.v1", data: []byte(`{"protocol":"JT808"}`), ack: func() error {
			acked++
			return nil
		}},
	}, map[string]string{"vehicle.event.unified.v1": "vehicle.event.unified.v1"})
	if err == nil {
		t.Fatal("bridgeBatch() error is nil, want kafka error")
	}
	if acked != 0 {
		t.Fatalf("acks = %d, want 0", acked)
	}
}

type recordingBridgeWriter struct {
	messages []kafka.Message
	err      error
}

func (w *recordingBridgeWriter) WriteMessages(_ context.Context, messages ...kafka.Message) error {
	if w.err != nil {
		return w.err
	}
	w.messages = append(w.messages, messages...)
	return nil
}