package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	cloudevents "github.com/cloudevents/sdk-go/v2"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"golang.source-fellows.com/seminar/event-sourcing/pkg/model"
	"log"
)

var ships = map[string]*model.Ship{}

func handleEvent(event cloudevents.Event) {

	movement := model.Movement{}
	err := json.Unmarshal(event.Data(), &movement)
	if err != nil {
		log.Printf("error while unmarshalling data: %v", err)
	}

	ship, ok := ships[movement.ShipName]
	if !ok {
		ship = model.NewShip(movement.ShipName, movement.ToPosition)
		ships[movement.ShipName] = ship
	} else {
		ship.UpdatePosition(movement.ToPosition)
	}
	// do something with event.
	fmt.Printf("%+v\n", ship)
}

func main() {
	// The default client is HTTP.
	nc, err := nats.Connect(nats.DefaultURL)
	if err != nil {
		log.Fatalf("failed to create connection, %v", err)
	}
	js, err := jetstream.New(nc)
	if err != nil {
		log.Fatalf("failed to create jetstream connection, %v", err)
	}

	stream, err := js.Stream(context.Background(), "positionStream")
	if err != nil {
		log.Fatalf("failed to create jetstream, %v", err)
	}

	consumer, err := stream.CreateConsumer(context.Background(),
		jetstream.ConsumerConfig{
			Durable:       "cons1",
			AckPolicy:     jetstream.AckExplicitPolicy,
			FilterSubject: "position.new",
		})
	if errors.Is(err, jetstream.ErrConsumerExists) {
		log.Println(err)
		consumer, err = stream.Consumer(context.Background(), "cons1")
	}
	if err != nil {
		log.Fatalf("failed to use consumer, %v", err)
	}

	iter, err := consumer.Messages()
	if err != nil {
		log.Fatalf("failed to use consumer, %v", err)
	}
	for {
		msg, err := iter.Next()
		if err != nil {
			break
		}
		event := cloudevents.Event{}
		err = event.UnmarshalJSON(msg.Data())
		if err != nil {
			log.Println(err)
			return
		}
		handleEvent(event)
		msg.Ack()
	}

	if err != nil {
		log.Fatalf("failed to consume, %v", err)
	}

	log.Println("finished")

}
