package main

import (
	"context"
	"fmt"
	"time"

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/ThreeDotsLabs/watermill/pubsub/gochannel"
)

func main() {
	// Create a logger
	logger := watermill.NewStdLogger(false, false)

	// Create an in-memory pub/sub (good for testing/examples)
	// For production, use Kafka, RabbitMQ, Google Pub/Sub, etc.
	pubSub := gochannel.NewGoChannel(
		gochannel.Config{},
		logger,
	)

	// Subscribe to a topic
	messages, err := pubSub.Subscribe(context.Background(), "user.created")
	if err != nil {
		panic(err)
	}

	// Process messages in a goroutine
	go func() {
		for msg := range messages {
			fmt.Printf("Received: %s\n", string(msg.Payload))

			// Process the message
			processUserCreated(msg)

			// Acknowledge the message
			msg.Ack()
		}
	}()

	// Publish messages
	for i := 0; i < 3; i++ {
		msg := message.NewMessage(
			watermill.NewUUID(),
			[]byte(fmt.Sprintf(`{"user_id": %d, "email": "user%d@example.com"}`, i, i)),
		)

		// Add metadata
		msg.Metadata.Set("timestamp", time.Now().Format(time.RFC3339))
		msg.Metadata.Set("version", "1.0")

		if err := pubSub.Publish("user.created", msg); err != nil {
			panic(err)
		}
		fmt.Printf("Published message: %s\n", msg.UUID)
	}

	// Wait for messages to be processed
	time.Sleep(time.Second)

	// Close the pub/sub
	pubSub.Close()
}

func processUserCreated(msg *message.Message) {
	// Access metadata
	timestamp := msg.Metadata.Get("timestamp")
	version := msg.Metadata.Get("version")

	fmt.Printf("  UUID: %s\n", msg.UUID)
	fmt.Printf("  Metadata: timestamp=%s, version=%s\n", timestamp, version)
	fmt.Printf("  Payload: %s\n", string(msg.Payload))
}

// Watermill Router example (for production use):
//
// router, err := message.NewRouter(message.RouterConfig{}, logger)
//
// router.AddHandler(
//     "user_created_handler",        // handler name
//     "user.created",                // subscribe topic
//     pubSub,                        // subscriber
//     "user.processed",              // publish topic
//     pubSub,                        // publisher
//     func(msg *message.Message) ([]*message.Message, error) {
//         // Process and optionally publish new messages
//         newMsg := message.NewMessage(watermill.NewUUID(), processedPayload)
//         return []*message.Message{newMsg}, nil
//     },
// )
//
// router.AddNoPublisherHandler(
//     "notification_handler",
//     "user.created",
//     pubSub,
//     func(msg *message.Message) error {
//         // Process without publishing
//         return sendNotification(msg)
//     },
// )
//
// router.Run(context.Background())

// Available pub/sub implementations:
// - gochannel (in-memory, for testing)
// - kafka
// - nats
// - rabbitmq (amqp)
// - googlecloud (Google Pub/Sub)
// - redis
// - sql (Postgres, MySQL)
// - bolt (embedded)
