package main

import (
	"context"
	"flag"
	"fmt"
	"time"

	"github.com/twmb/franz-go/pkg/kadm"
	"github.com/twmb/franz-go/pkg/kgo"
	"github.com/twmb/franz-go/pkg/sasl/plain"
)

func main() {
	ctx := context.Background()

	var password, hostname string
	var replication int
	flag.StringVar(&password, "p", "", "password to connect to kafka")
	flag.StringVar(&hostname, "h", "kafka", "hostname to connect to kafka")
	flag.IntVar(&replication, "r", 3, "number of replication")
	flag.Parse()

	seeds := []string{fmt.Sprintf("%s.zerops:9092", hostname)}
	cl, err := kgo.NewClient(
		kgo.SeedBrokers(seeds...),
		kgo.SASL(plain.Auth{
			User: "zerops",
			Pass: password,
		}.AsMechanism()),
	)
	if err != nil {
		panic(err)
	}
	defer cl.Close()

	adminClient := kadm.NewClient(cl)
	topic, err := adminClient.ListTopics(ctx, "foo")
	if err != nil {
		panic(err)
	}
	// if topic.Error() != nil {
	// 	panic(topic.Error())
	// }

	if !topic.Has("foo") {
		partitions := int32(6)
		if replication == 1 {
			partitions = 1
		}
		resp, err := adminClient.CreateTopic(ctx, partitions, int16(replication), nil, "foo")
		if err != nil {
			panic(err)
		}
		if resp.Err != nil {
			panic(resp.Err)
		}
	}

	go func() {
		for {
			message := time.Now().Format(time.TimeOnly)
			fmt.Printf("%s ==>\n", message)
			// Alternatively, ProduceSync exists to synchronously produce a batch of records.
			record := &kgo.Record{Topic: "foo", Value: []byte(message)}
			if err := cl.ProduceSync(ctx, record).FirstErr(); err != nil {
				fmt.Printf("record had a produce error while synchronously producing: %v\n", err)
			}
			time.Sleep(time.Second)
		}
	}()

	cl.AddConsumeTopics("foo")
	for {
		fetches := cl.PollFetches(ctx)
		if fetches.IsClientClosed() {
			fmt.Println("closed ")
			return
		}
		if errs := fetches.Errors(); len(errs) > 0 {
			// All errors are retried internally when fetching, but non-retriable errors are
			// returned from polls so that users can notice and take action.
			panic(fmt.Sprint(errs))
		}
		// We can iterate through a record iterator...
		iter := fetches.RecordIter()
		for !iter.Done() {
			record := iter.Next()
			fmt.Printf("==> %s\n", string(record.Value))
		}
	}
}