package main

import (
	"fmt"
	"sync"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

// start docker: docker run --rm -it --hostname my-rabbit -p 15672:15672 -p 5672:5672 rabbitmq:3-management"
// setup consumer_timeout: rabbitmqctl eval 'application:set_env(rabbit, consumer_timeout, 10000).'

func main() {
	rmq, _ := NewRabbitMQ("amqp://guest:guest@localhost:5672/")

	err := rmq.StartConsumeMessage()
	if err != nil {
		fmt.Printf("error start rabbitmq: %v\n", err)
	}

	_ = rmq.Close()
}

type rabbitMQ struct {
	uri                 string
	conn                *amqp.Connection
	connCloseNotifyChan chan *amqp.Error
}

func NewRabbitMQ(uri string) (*rabbitMQ, error) {
	return &rabbitMQ{
		uri: uri,
	}, nil
}

func (r *rabbitMQ) Connect() error {
	conn, err := amqp.Dial(r.uri)
	if err != nil {
		return fmt.Errorf("error dial rabbitmq: %w", err)
	}

	r.conn = conn

	r.connCloseNotifyChan = make(chan *amqp.Error, 1)
	r.conn.NotifyClose(r.connCloseNotifyChan) // note: this channel will be closed when connection closed >> we need to create new channel when reconnect

	return nil
}

func (r *rabbitMQ) StartConsumeMessage() error {
	fmt.Println("Start rabbitmq")
	err := r.Connect()
	if err != nil {
		return fmt.Errorf("error connect rabbitmq: %w", err)
	}

	// thread to listen connection close
	go func() {
		fmt.Println("Start thread listen connection close")
		defer fmt.Println("Stop thread listen connection close")
		for {
			select {
			case err := <-r.connCloseNotifyChan:
				if err != nil {
					fmt.Printf("Received connection event: err  : %v - %v\n", err.Code, err.Reason)
				}
			}
		}
	}()

	ch, err := r.conn.Channel()
	if err != nil {
		return fmt.Errorf("error create channel: %v\n", err)
	}
	fmt.Println("DEBUG: create channel success")

	// create queue

	_, err = ch.QueueDeclare(
		"test_queue",
		false,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		return fmt.Errorf("error declare queue: %w", err)
	}

	// set qos
	err = ch.Qos(1, 0, false)
	if err != nil {
		return fmt.Errorf("error set qos: %w", err)
	}

	internalQueue := make(chan amqp.Delivery, 100)
	wg := sync.WaitGroup{}
	wg.Add(2)

	// thread consume message
	go func() {
		defer wg.Done()
		defer close(internalQueue)
		defer func() {
			fmt.Printf("Stop thread consume message: Sleep 20s \n")
			time.Sleep(time.Second * 20)
		}()

		q, err := ch.QueueDeclare(
			"test_queue",
			false,
			false,
			false,
			false,
			nil,
		)
		if err != nil {
			fmt.Printf("error declare queue: %w", err)
		}

		deliveries, err := ch.Consume(
			q.Name,
			"",
			false,
			false,
			false,
			false,
			nil,
		)
		if err != nil {
			fmt.Printf("error consume message: %w", err)
			return
		}

		for msg := range deliveries {
			internalQueue <- msg
		}

	}()

	// thread process message
	go func() {
		defer wg.Done()
		defer func() {
			fmt.Printf("Stop thread handle message: Sleep 20s \n")
			time.Sleep(time.Second * 20)
		}()

		for msg := range internalQueue {
			fmt.Println("start", string(msg.Body))
			time.Sleep(time.Second * 20)
			fmt.Println("done", string(msg.Body))
			err := msg.Ack(false)
			if err != nil {
				fmt.Printf("error ack message: %v\n", err)
			}
		}
	}()

	wg.Wait()

	return nil
}

func (r *rabbitMQ) Close() error {
	return r.conn.Close()
}