package main

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"

	"github.com/hibiken/asynq"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	amqp "github.com/rabbitmq/amqp091-go"
	"github.com/riverqueue/river"
	"github.com/segmentio/kafka-go"
	"github.com/twmb/franz-go/pkg/kgo"
)

type FetchCommit interface {
	FetchMessage(context.Context) (kafka.Message, error)
	CommitMessages(context.Context, ...kafka.Message) error
}

var _ FetchCommit = (*kafka.Reader)(nil)

// The caller must stop/retry on a work error, not skip to a later offset.
func HandleKafka(ctx context.Context, r FetchCommit, work func([]byte) error) error {
	m, err := r.FetchMessage(ctx)
	if err != nil {
		return err
	}
	if err := work(m.Value); err != nil {
		return err
	}
	return r.CommitMessages(ctx, m)
}

// Error classification / Nack policy belongs to the outer consumer loop.
func HandleAMQP(d amqp.Delivery, work func([]byte) error) error {
	if err := work(d.Body); err != nil {
		return err
	}
	return d.Ack(false)
}

type ackRecorder struct {
	calls    int
	tag      uint64
	multiple bool
	err      error
}

func (a *ackRecorder) Ack(tag uint64, multiple bool) error {
	a.calls++
	a.tag = tag
	a.multiple = multiple
	return a.err
}
func (a *ackRecorder) Nack(uint64, bool, bool) error { return nil }
func (a *ackRecorder) Reject(uint64, bool) error     { return nil }

type InvoiceArgs struct {
	OrderID string `json:"order_id"`
}

func (InvoiceArgs) Kind() string { return "invoice:render" }
func render(id string) (string, error) {
	if id == "" {
		return "", errors.New("missing order_id")
	}
	return "invoice:" + id, nil
}

type InvoiceHandler struct{ Out io.Writer }

func (h InvoiceHandler) ProcessTask(ctx context.Context, t *asynq.Task) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	var a InvoiceArgs
	if err := json.Unmarshal(t.Payload(), &a); err != nil {
		return fmt.Errorf("%w: %v", asynq.SkipRetry, err)
	}
	s, err := render(a.OrderID)
	if err != nil {
		return fmt.Errorf("%w: %v", asynq.SkipRetry, err)
	}
	_, err = fmt.Fprintln(h.Out, s)
	return err
}

var _ asynq.Handler = InvoiceHandler{}

type InvoiceWorker struct {
	river.WorkerDefaults[InvoiceArgs]
	Out io.Writer
}

func (w *InvoiceWorker) Work(ctx context.Context, job *river.Job[InvoiceArgs]) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	s, err := render(job.Args.OrderID)
	if err != nil {
		return river.JobCancel(err)
	}
	_, err = fmt.Fprintln(w.Out, s)
	return err
}

func demo() {
	c, err := kgo.NewClient(kgo.RequiredAcks(kgo.NoAck()))
	if c != nil {
		c.Close()
	}
	fmt.Println("franz invalid acks:", err)
	cfg := kafka.ReaderConfig{Brokers: []string{"127.0.0.1:9092"}, Topic: "orders", GroupID: "billing"}
	fmt.Println("kafka reader config:", cfg.Validate())
	fmt.Println("kafka writer zero acks:", int((&kafka.Writer{}).RequiredAcks))
	a := new(ackRecorder)
	d := amqp.Delivery{Acknowledger: a, DeliveryTag: 7, Body: []byte("event")}
	if err := HandleAMQP(d, func([]byte) error { return nil }); err != nil {
		panic(err)
	}
	fmt.Println("amqp ack:", a.calls, a.tag, a.multiple)
	m := nats.NewMsg("orders.created")
	m.Header.Set(jetstream.MsgIDHeader, "e1")
	jsCfg := jetstream.ConsumerConfig{Durable: "billing", AckPolicy: jetstream.AckExplicitPolicy, MaxDeliver: 3}
	fmt.Println("nats message/config:", m.Subject, m.Header.Get(jetstream.MsgIDHeader), jsCfg.AckPolicy == jetstream.AckExplicitPolicy, jsCfg.MaxDeliver)
	var out bytes.Buffer
	h := InvoiceHandler{Out: &out}
	task := asynq.NewTask("invoice:render", []byte(`{"order_id":"o1"}`), asynq.MaxRetry(3))
	if err := h.ProcessTask(context.Background(), task); err != nil {
		panic(err)
	}
	fmt.Printf("asynq result: %s", out.String())
	err = h.ProcessTask(context.Background(), asynq.NewTask("invoice:render", []byte(`{}`)))
	fmt.Println("asynq invalid skips retry:", errors.Is(err, asynq.SkipRetry))
	out.Reset()
	w := &InvoiceWorker{Out: &out}
	workers := river.NewWorkers()
	river.AddWorker(workers, w)
	if err := w.Work(context.Background(), &river.Job[InvoiceArgs]{Args: InvoiceArgs{OrderID: "o2"}}); err != nil {
		panic(err)
	}
	fmt.Printf("river kind/result: %s %s", (InvoiceArgs{}).Kind(), out.String())
}
func main() { demo() }