// The package implements a minimal set of Azure Event Hubs functionality.
//
// We could use https://github.com/Azure/azure-event-hubs-go but it's too huge
// and also it has a number of dependencies that aren't desired in the project.
package eventhub

import (
	"context"
	"crypto/rand"
	"crypto/tls"
	"encoding/hex"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/Azure/go-amqp"
)

// Credentials is an evenhub connection string representation.
type Credentials struct {
	Endpoint            string
	SharedAccessKeyName string
	SharedAccessKey     string
	EntityPath          string
}

// ParseConnectionString parses the given connection string into Credentials structure.
func ParseConnectionString(cs string) (*Credentials, error) {
	var c Credentials
	for _, s := range strings.Split(cs, ";") {
		kv := strings.SplitN(s, "=", 2)
		if len(kv) != 2 {
			return nil, errors.New("malformed connection string")
		}

		switch kv[0] {
		case "Endpoint":
			if !strings.HasPrefix(kv[1], "sb://") {
				return nil, errors.New("only sb:// schema supported")
			}
			c.Endpoint = strings.TrimRight(kv[1][5:], "/")
		case "SharedAccessKeyName":
			c.SharedAccessKeyName = kv[1]
		case "SharedAccessKey":
			c.SharedAccessKey = kv[1]
		case "EntityPath":
			c.EntityPath = kv[1]
		}
	}
	return &c, nil
}

// Option is a client configuration option.
type Option func(c *Client)

// WithTLSConfig sets connection TLS configuration.
func WithTLSConfig(tc *tls.Config) Option {
	return func(c *Client) {
		c.opts.TLSConfig = tc
	}
}

// WithSASLPlain configures connection username and password.
func WithSASLPlain(username, password string) Option {
	return func(c *Client) {
		c.opts.SASLType = amqp.SASLTypePlain(username, password)
	}
}

// WithConnOption sets a low-level connection option.
func WithConnOption(key string, value any) Option {
	return func(c *Client) {
		if c.opts.Properties == nil {
			c.opts.Properties = make(map[string]any)
		}
		c.opts.Properties[key] = value
	}
}

// DialContext connects to the named EventHub and returns a client instance
// using the provided context.
func DialContext(ctx context.Context, host, name string, opts ...Option) (*Client, error) {
	c := &Client{name: name}
	for _, opt := range opts {
		opt(c)
	}

	var err error
	c.conn, err = amqp.Dial(ctx, "amqps://"+host, &c.opts)
	if err != nil {
		return nil, err
	}
	return c, nil
}

// Dial connects to the named EventHub and returns a client instance.
// DEPRECATED: Use DialContext.
func Dial(host, name string, opts ...Option) (*Client, error) {
	return DialContext(context.Background(), host, name, opts...)
}

// DialConnectionString dials an EventHub instance using the given connection string
// and the provided context.
func DialConnectionStringContext(ctx context.Context, cs string, opts ...Option) (*Client, error) {
	creds, err := ParseConnectionString(cs)
	if err != nil {
		return nil, err
	}
	return DialContext(ctx, creds.Endpoint, creds.EntityPath, append([]Option{
		WithSASLPlain(creds.SharedAccessKeyName, creds.SharedAccessKey),
	}, opts...)...)
}

// DialConnectionString dials an EventHub instance using the given connection string.
// DEPRECATED: Use DialConnectionStringContext.
func DialConnectionString(cs string, opts ...Option) (*Client, error) {
	return DialConnectionStringContext(context.Background(), cs, opts...)
}

// Client is an EventHub client.
type Client struct {
	name string
	conn *amqp.Conn
	opts amqp.ConnOptions
}

// SubscribeOption is a Subscribe option.
type SubscribeOption func(r *sub)

// WithSubscribeConsumerGroup overrides default consumer group, default is `$Default`.
func WithSubscribeConsumerGroup(name string) SubscribeOption {
	return func(s *sub) {
		s.group = name
	}
}

// WithSubscribeSince requests events that occurred after the given time.
func WithSubscribeSince(t time.Time) SubscribeOption {
	return func(s *sub) {
		if s.receiverOpts.Filters == nil {
			s.receiverOpts.Filters = make([]amqp.LinkFilter, 0)
		}
		s.receiverOpts.Filters = append(s.receiverOpts.Filters, amqp.NewSelectorFilter(fmt.Sprintf("amqp.annotation.x-opt-enqueuedtimeutc > '%d'",
			t.UnixNano()/int64(time.Millisecond))))
	}
}

type sub struct {
	group        string
	sessionOpts  amqp.SessionOptions
	receiverOpts amqp.ReceiverOptions
}

// Event is an Event Hub event, simply wraps an AMQP message.
type Event struct {
	*amqp.Message
	recv *amqp.Receiver
}

// Subscribe subscribes to all hub's partitions and registers the given
// handler and blocks until it encounters an error or the context is cancelled.
//
// Non-nil errors returned from fn reject AMQP messages.
func (c *Client) Subscribe(
	ctx context.Context,
	fn func(event *Event) error,
	opts ...SubscribeOption,
) error {
	var s sub
	for _, opt := range opts {
		opt(&s)
	}
	if s.group == "" {
		s.group = "$Default"
	}

	// initialize new session for each subscribe session
	sess, err := c.conn.NewSession(ctx, &s.sessionOpts)
	if err != nil {
		return err
	}
	defer sess.Close(context.Background())

	ids, err := c.getPartitionIDs(ctx, sess)
	if err != nil {
		return err
	}

	// stop all goroutines at return
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	evc := make(chan *Event)
	errc := make(chan error)

	for _, id := range ids {
		addr := fmt.Sprintf("/%s/ConsumerGroups/%s/Partitions/%s", c.name, s.group, id)
		recv, err := sess.NewReceiver(ctx, addr, &s.receiverOpts)
		if err != nil {
			return err
		}

		go func(recv *amqp.Receiver) {
			defer recv.Close(context.Background())
			for {
				msg, err := recv.Receive(ctx, &amqp.ReceiveOptions{})
				if err != nil {
					select {
					case errc <- err:
					case <-ctx.Done():
					}
					return
				}
				select {
				case evc <- &Event{Message: msg, recv: recv}:
				case <-ctx.Done():
				}
			}
		}(recv)
	}

	for {
		select {
		case ev := <-evc:
			if err := fn(ev); err != nil {
				if rerr := ev.recv.RejectMessage(ctx, ev.Message, &amqp.Error{
					Condition:   amqp.ErrCondInternalError,
					Description: err.Error(),
				}); rerr != nil {
					return rerr
				}
				return err
			}
			if err := ev.recv.AcceptMessage(ctx, ev.Message); err != nil {
				return err
			}
		case err := <-errc:
			return err
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

// getPartitionIDs returns partition ids of the hub.
func (c *Client) getPartitionIDs(ctx context.Context, sess *amqp.Session) ([]string, error) {
	replyTo := genID()
	recv, err := sess.NewReceiver(ctx, "$management", &amqp.ReceiverOptions{TargetAddress: replyTo})
	if err != nil {
		return nil, err
	}
	defer recv.Close(context.Background())

	send, err := sess.NewSender(ctx, "$management", &amqp.SenderOptions{SourceAddress: replyTo})
	if err != nil {
		return nil, err
	}
	defer send.Close(context.Background())

	mid := genID()
	if err := send.Send(ctx, &amqp.Message{
		Properties: &amqp.MessageProperties{
			MessageID: mid,
			ReplyTo:   &replyTo,
		},
		ApplicationProperties: map[string]interface{}{
			"operation": "READ",
			"name":      c.name,
			"type":      "com.microsoft:eventhub",
		},
	}, &amqp.SendOptions{}); err != nil {
		return nil, err
	}

	msg, err := recv.Receive(ctx, &amqp.ReceiveOptions{})
	if err != nil {
		return nil, err
	}
	if err = CheckMessageResponse(msg); err != nil {
		return nil, err
	}
	if msg.Properties.CorrelationID != mid {
		return nil, errors.New("message-id mismatch")
	}
	if err := recv.AcceptMessage(ctx, msg); err != nil {
		return nil, err
	}

	val, ok := msg.Value.(map[string]interface{})
	if !ok {
		return nil, errors.New("unable to typecast value")
	}
	ids, ok := val["partition_ids"].([]string)
	if !ok {
		return nil, errors.New("unable to typecast partition_ids")
	}
	return ids, nil
}

// Close closes underlying AMQP connection.
func (c *Client) Close() error {
	return c.conn.Close()
}

// CheckMessageResponse checks for 200 response code otherwise returns an error.
func CheckMessageResponse(msg *amqp.Message) error {
	rc, ok := msg.ApplicationProperties["status-code"].(int32)
	if !ok {
		return errors.New("unable to typecast status-code")
	}
	if rc == 200 {
		return nil
	}
	rd, _ := msg.ApplicationProperties["status-description"].(string)
	return fmt.Errorf("code = %d, description = %q", rc, rd)
}

func genID() string {
	b := make([]byte, 16)
	if _, err := rand.Read(b); err != nil {
		panic(err)
	}
	return hex.EncodeToString(b)
}
