package eventstream

import (
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"sync"
	"time"

	"github.com/donovanhide/eventsource"
)

// Stream handles a connection for receiving Server Sent Events.
// It will try and reconnect if the connection is lost, respecting both
// received retry delays and event id's.
type Stream struct {
	c           *http.Client
	req         *http.Request
	lastEventID string
	retry       time.Duration
	// Events emits the events received by the stream
	Events chan StreamEvent
	Ready  chan bool
	// Errors emits any errors encountered while reading events from the stream.
	// It's mainly for informative purposes - the client isn't required to take any
	// action when an error is encountered. The stream will always attempt to continue,
	// even if that involves reconnecting to the server.
	Errors chan error
	// Logger is a logger that, when set, will be used for logging debug messages
	Logger *log.Logger
	// isClosed is a marker that the stream is/should be closed
	isClosed bool
	// isClosedMutex is a mutex protecting concurrent read/write access of isClosed
	closeMutex sync.Mutex
}

type StreamEvent interface {
	// Id is an identifier that can be used to allow a client to replay
	// missed Events by returning the Last-Event-Id header.
	// Return empty string if not required.
	Id() string
	// The name of the event. Return empty string if not required.
	Event() string
	// The payload of the event.
	Data() string
	Retry() int64
}

type SubscriptionError struct {
	Code    int
	Message string
}

func (e SubscriptionError) Error() string {
	return fmt.Sprintf("%d: %s", e.Code, e.Message)
}

// Subscribe to the Events emitted from the specified url.
// If lastEventId is non-empty it will be sent to the server in case it can replay missed events.
func Subscribe(url, lastEventID string) (*Stream, error) {
	req, err := http.NewRequest("GET", url, http.NoBody)
	if err != nil {
		return nil, err
	}

	return SubscribeWithRequest(lastEventID, req)
}

// SubscribeWithRequest will take an http.Request to setup the stream, allowing custom headers
// to be specified, authentication to be configured, etc.
func SubscribeWithRequest(lastEventID string, request *http.Request) (*Stream, error) {
	return SubscribeWith(lastEventID, http.DefaultClient, request)
}

// SubscribeWith takes a http client and request providing customization over both headers and
// control over the http client settings (timeouts, tls, etc)
func SubscribeWith(lastEventID string, client *http.Client, request *http.Request) (*Stream, error) {
	stream := &Stream{
		c:           client,
		req:         request,
		lastEventID: lastEventID,
		retry:       time.Millisecond * 3000,
		Events:      make(chan StreamEvent),
		Errors:      make(chan error, 10),
		Ready:       make(chan bool),
	}
	stream.c.CheckRedirect = checkRedirect

	r, err := stream.connect()
	if err != nil {
		return nil, err
	}

	go stream.stream(r)

	return stream, nil
}

// Close will close the stream. It is safe for concurrent access and can be called multiple times.
func (stream *Stream) Close() {
	go func() {
		stream.closeMutex.Lock()
		defer stream.closeMutex.Unlock()

		if stream.isClosed {
			return
		}

		stream.isClosed = true
		close(stream.Errors)
		close(stream.Events)
	}()
}

// Go's http package doesn't copy headers across when it encounters
// redirects so we need to do that manually.
func checkRedirect(req *http.Request, via []*http.Request) error {
	if len(via) >= 10 {
		return errors.New("stopped after 10 redirects")
	}

	for k, vv := range via[0].Header {
		for _, v := range vv {
			req.Header.Add(k, v)
		}
	}

	return nil
}

func (stream *Stream) connect() (r io.ReadCloser, err error) {
	var resp *http.Response

	stream.req.Header.Set("Cache-Control", "no-cache")
	stream.req.Header.Set("Accept", "text/event-stream")

	if stream.lastEventID != "" {
		stream.req.Header.Set("Last-Event-ID", stream.lastEventID)
	}

	if resp, err = stream.c.Do(stream.req); err != nil {
		return
	}

	if resp.StatusCode != 200 {
		message, _ := io.ReadAll(resp.Body)
		err = SubscriptionError{
			Code:    resp.StatusCode,
			Message: string(message),
		}
	}

	r = resp.Body

	return
}

func (stream *Stream) stream(r io.ReadCloser) {
	defer r.Close()

	stream.Ready <- true

	// receives events until an error is encountered
	stream.receiveEvents(r)

	// tries to reconnect and start the stream again
	stream.retryRestartStream()
}

func (stream *Stream) receiveEvents(r io.ReadCloser) {
	dec := eventsource.NewDecoder(r)

	for {
		ev, err := dec.Decode()

		stream.closeMutex.Lock()
		if stream.isClosed {
			stream.closeMutex.Unlock()
			return
		}

		if err != nil {
			stream.Errors <- err
			stream.closeMutex.Unlock()

			return
		}

		pub, ok := ev.(StreamEvent)
		if !ok {
			stream.closeMutex.Unlock()
			continue
		}

		if pub.Retry() > 0 {
			stream.retry = time.Duration(pub.Retry()) * time.Millisecond
		}

		if pub.Id() != "" {
			stream.lastEventID = pub.Id()
		}

		stream.Events <- pub
		stream.closeMutex.Unlock()
	}
}

func (stream *Stream) retryRestartStream() {
	backoff := stream.retry

	for {
		if stream.Logger != nil {
			stream.Logger.Printf("Reconnecting in %0.4f secs\n", backoff.Seconds())
		}

		time.Sleep(backoff)

		if stream.isClosed {
			return
		}

		// NOTE: because of the defer we're opening the new connection
		// before closing the old one. Shouldn't be a problem in practice,
		// but something to be aware of.
		r, err := stream.connect()
		if err == nil {
			go stream.stream(r)
			return
		}

		stream.closeMutex.Lock()
		if stream.isClosed {
			stream.closeMutex.Unlock()
			return
		}
		stream.Errors <- err
		stream.closeMutex.Unlock()

		backoff = 10 * time.Second
	}
}
