package main

import (
	"context"
	"errors"
	amqp "github.com/rabbitmq/amqp091-go"
	"github.com/segmentio/kafka-go"
	"testing"
)

func Example() {
	demo()
	// Output:
	// franz invalid acks: idempotency requires acks=all
	// kafka reader config: <nil>
	// kafka writer zero acks: 0
	// amqp ack: 1 7 false
	// nats message/config: orders.created e1 true 3
	// asynq result: invoice:o1
	// asynq invalid skips retry: true
	// river kind/result: invoice:render invoice:o2
}

type fakeReader struct {
	commits   int
	commitErr error
}

func (*fakeReader) FetchMessage(context.Context) (kafka.Message, error) {
	return kafka.Message{Topic: "orders", Partition: 0, Offset: 10, Value: []byte("e1")}, nil
}
func (r *fakeReader) CommitMessages(context.Context, ...kafka.Message) error {
	r.commits++
	return r.commitErr
}
func TestKafkaNoCommitOnWorkFailure(t *testing.T) {
	r := new(fakeReader)
	err := HandleKafka(context.Background(), r, func([]byte) error { return errors.New("business failed") })
	if err == nil || r.commits != 0 {
		t.Fatalf("err=%v commits=%d", err, r.commits)
	}
}
func TestKafkaCommitFailureIsReturned(t *testing.T) {
	boom := errors.New("commit failed")
	r := &fakeReader{commitErr: boom}
	if err := HandleKafka(context.Background(), r, func([]byte) error { return nil }); !errors.Is(err, boom) {
		t.Fatal(err)
	}
}
func TestAMQPNoAckOnWorkFailure(t *testing.T) {
	a := new(ackRecorder)
	d := amqp.Delivery{Acknowledger: a, DeliveryTag: 7}
	if err := HandleAMQP(d, func([]byte) error { return errors.New("business failed") }); err == nil {
		t.Fatal("expected failure")
	}
	if a.calls != 0 {
		t.Fatal("acked failed business")
	}
}
func TestAMQPAckFailureIsReturned(t *testing.T) {
	boom := errors.New("ack failed")
	a := &ackRecorder{err: boom}
	if err := HandleAMQP(amqp.Delivery{Acknowledger: a}, func([]byte) error { return nil }); !errors.Is(err, boom) {
		t.Fatal(err)
	}
}