package eventstreams

import (
	"context"

	"github.com/golang/protobuf/ptypes/wrappers"
	"github.com/pkg/errors"
	"github.com/twitchtv/twirp"
	"go.uber.org/zap"

	"code.justin.tv/eventbus/controlplane/internal/clients/servicecatalog"
	"code.justin.tv/eventbus/controlplane/internal/db"
	"code.justin.tv/eventbus/controlplane/internal/ldap"
	"code.justin.tv/eventbus/controlplane/internal/logger"
	"code.justin.tv/eventbus/controlplane/internal/metrics/clients/eventstreamstats"
	"code.justin.tv/eventbus/controlplane/internal/twirperr"
	"code.justin.tv/eventbus/controlplane/rpc"
)

type EventTypesService struct {
	DB db.DB
}

func (s *EventTypesService) List(ctx context.Context, req *rpc.ListEventTypesReq) (*rpc.ListEventTypesResp, error) {
	events, err := s.DB.EventTypes(ctx)
	if err != nil {
		return nil, twirperr.Convert(errors.Wrap(err, "could not fetch event types"))
	}

	if req.LdapGroup == "" {
		return &rpc.ListEventTypesResp{
			EventTypes: eventTypesToRPC(events),
		}, nil
	}

	filteredEvents := make([]*db.EventType, 0)
	for _, et := range events {
		if et.LDAPGroup == req.LdapGroup {
			filteredEvents = append(filteredEvents, et)
		}
	}
	return &rpc.ListEventTypesResp{
		EventTypes: eventTypesToRPC(filteredEvents),
	}, nil
}

func (s *EventTypesService) ListAuthorizedFieldSubscriberGrantsByEventType(ctx context.Context, req *rpc.ListAuthorizedFieldSubscriberGrantsByEventTypeReq) (*rpc.ListAuthorizedFieldSubscriberGrantsByEventTypeResp, error) {
	if req.EventTypeName == "" {
		return nil, twirp.NewError(twirp.InvalidArgument, "No event type name provided")
	}

	et, err := s.DB.EventTypeByName(ctx, req.EventTypeName)
	if err != nil {
		return nil, twirperr.Convert(errors.Wrap(err, "failed to retrieve event type"))
	}

	if !ldap.BelongsToGroup(ctx, et.LDAPGroup) {
		return nil, twirp.NewError(twirp.NotFound, "not found")
	}

	ess, err := s.DB.EventStreamsByEventTypeID(ctx, et.ID)
	if err != nil {
		return nil, twirperr.Convert(errors.Wrap(err, "failed to retrieve event streams"))
	}

	var protos = make([]*rpc.Grant, 0)
	for _, es := range ess {
		afs, err := s.DB.AuthorizedFieldsByEventStreamID(ctx, es.ID)
		if err != nil {
			return nil, twirperr.Convert(errors.Wrap(err, "failed to retrieve authorized fields"))
		}

		for _, af := range afs {
			sgs, err := s.DB.AuthorizedFieldSubscriberGrantsByAuthorizedField(ctx, af.ID)
			if err != nil {
				return nil, twirperr.Convert(errors.Wrap(err, "failed to retrieve authorized field subscriber grants"))
			}

			for _, sg := range sgs {
				ir, err := s.DB.IAMRoleByID(ctx, sg.IAMRoleID)
				if err != nil {
					return nil, twirperr.Convert(errors.Wrap(err, "failed to retrieve iam role"))
				}

				srv, err := s.DB.ServiceByID(ctx, ir.ServiceID)
				if err != nil {
					return nil, twirperr.Convert(errors.Wrap(err, "failed to retrieve service"))
				}

				protos = append(protos, &rpc.Grant{
					IamRoleArn:  ir.ARN,
					ServiceName: srv.Name,
					FieldName:   af.FieldName,
					MessageName: af.MessageName,
					Environment: es.Environment,
				})
			}
		}
	}

	return &rpc.ListAuthorizedFieldSubscriberGrantsByEventTypeResp{
		Grants: protos,
	}, nil
}

func (s *EventTypesService) ListAuthorizedFieldPublisherGrantsByEventType(ctx context.Context, req *rpc.ListAuthorizedFieldPublisherGrantsByEventTypeReq) (*rpc.ListAuthorizedFieldPublisherGrantsByEventTypeResp, error) {
	if req.EventTypeName == "" {
		return nil, twirperr.Convert(errors.New("no event type provided"))
	}

	et, err := s.DB.EventTypeByName(ctx, req.EventTypeName)
	if err != nil {
		return nil, twirperr.Convert(errors.Wrap(err, "failed to retrieve event type"))
	}

	if !ldap.BelongsToGroup(ctx, et.LDAPGroup) {
		return nil, twirp.NewError(twirp.NotFound, "not found")
	}

	ess, err := s.DB.EventStreamsByEventTypeID(ctx, et.ID)
	if err != nil {
		return nil, twirperr.Convert(errors.Wrap(err, "failed to retrieve event streams"))
	}

	var protos = make([]*rpc.PublisherGrant, 0)
	for _, es := range ess {
		pgs, err := s.DB.AuthorizedFieldPublisherGrantsByEventStreamID(ctx, es.ID)
		if err != nil {
			return nil, twirperr.Convert(errors.Wrap(err, "failed to retrieve authorized fields publisher grants"))
		}

		for _, pg := range pgs {
			ir, err := s.DB.IAMRoleByID(ctx, pg.IAMRoleID)
			if err != nil {
				return nil, twirperr.Convert(errors.Wrap(err, "failed to retrieve iam role"))
			}

			srv, err := s.DB.ServiceByID(ctx, ir.ServiceID)
			if err != nil {
				return nil, twirperr.Convert(errors.Wrap(err, "failed to retrieve service"))
			}

			protos = append(protos, &rpc.PublisherGrant{
				IamRoleArn:  ir.ARN,
				ServiceName: srv.Name,
				Environment: es.Environment,
			})
		}
	}

	return &rpc.ListAuthorizedFieldPublisherGrantsByEventTypeResp{
		Grants: protos,
	}, nil
}

type EventStreamsService struct {
	DB             db.DB
	Stats          eventstreamstats.Fetcher
	ServiceCatalog servicecatalog.Client
}

func (s *EventStreamsService) List(ctx context.Context, req *rpc.ListEventStreamsReq) (*rpc.ListEventStreamsResp, error) {
	var events []*db.EventStream
	var err error
	if req.EventTypeName != "" {
		events, err = s.DB.EventStreamsByName(ctx, req.EventTypeName)
	} else {
		events, err = s.DB.EventStreams(ctx)
	}

	if err != nil {
		return nil, twirperr.Convert(errors.Wrap(err, "could not fetch event streams"))
	}

	authFields, err := s.DB.AuthorizedFields(ctx)
	if err != nil {
		return nil, twirperr.Convert(errors.Wrap(err, "could not fetch event stream authorized fields"))
	}

	authFieldMapping := make(map[int][]*db.AuthorizedField)
	for _, authField := range authFields {
		authFieldMapping[authField.EventStreamID] = append(authFieldMapping[authField.EventStreamID], authField)
	}

	outEvents := make([]*rpc.EventStream, len(events))
	for i, event := range events {
		authFields := authFieldMapping[event.ID]
		if authFields == nil {
			authFields = []*db.AuthorizedField{}
		}
		outEvents[i] = eventStreamToRPC(event, authFields)
	}
	return &rpc.ListEventStreamsResp{
		EventStreams: outEvents,
	}, nil
}

// Get details of a single event stream
func (s *EventStreamsService) Get(ctx context.Context, req *rpc.GetEventStreamReq) (*rpc.EventStream, error) {
	var eventStream *db.EventStream
	var err error
	if eventStream, err = s.loadEventStream(ctx, req.EventType, req.Environment); err != nil {
		return nil, err
	}

	authFields, err := s.DB.AuthorizedFieldsByEventStreamID(ctx, eventStream.ID)
	if err != nil {
		return nil, twirperr.Convert(errors.Wrap(err, "error fetching event stream authorized fields"))
	}

	return eventStreamToRPC(eventStream, authFields), nil
}

// GetStats retrieves usage statistics for an event stream
func (s *EventStreamsService) GetStats(ctx context.Context, req *rpc.GetEventStreamStatsReq) (*rpc.EventStreamStats, error) {
	var eventStream *db.EventStream
	var err error
	if eventStream, err = s.loadEventStream(ctx, req.EventType, req.Environment); err != nil {
		return nil, err
	}

	eventStreamStats, err := s.Stats.Fetch(ctx, eventStream.SNSDetails.SNSTopicARN)
	if err != nil {
		return nil, twirperr.Convert(errors.Wrap(err, "could not fetch stats from cloudwatch"))
	}
	rpcStats := statsToRPC(eventStreamStats)

	publishingServices, err := s.DB.PublisherServicesByEventStreamID(ctx, eventStream.ID)
	if err != nil {
		return nil, twirperr.Convert(errors.Wrap(err, "could not determine publishing services for event stream"))
	}
	rpcStats.PublisherCount = uint32(len(publishingServices))
	rpcStats.Publishers = s.genPublisherInfo(ctx, publishingServices)

	subscribers, err := s.DB.SubscriptionsByEventStreamID(ctx, eventStream.ID)
	if err != nil {
		return nil, twirperr.Convert(errors.Wrap(err, "could not determine subscribers for event stream"))
	}
	rpcStats.SubscriberCount = uint32(len(subscribers))

	return rpcStats, nil
}

func (s *EventStreamsService) loadEventStream(ctx context.Context, name, environment string) (*db.EventStream, error) {
	eventStream, err := s.DB.EventStreamByNameAndEnvironment(ctx, name, environment)
	if err != nil {
		if err == db.ErrResourceNotFound {
			return nil, twirp.NotFoundError("Event Stream Not Found")
		}
		return nil, twirperr.Convert(errors.Wrap(err, "could not load event stream"))
	}
	return eventStream, nil
}

func (s *EventStreamsService) genPublisherInfo(ctx context.Context, services []*db.Service) []*rpc.PublisherServiceInfo {
	log := logger.FromContext(ctx)
	info := make([]*rpc.PublisherServiceInfo, 0)
	for _, service := range services {
		serviceName := service.Name
		serviceResp, err := s.ServiceCatalog.Get(ctx, service.ServiceCatalogID)
		if err == servicecatalog.ErrDoesNotExist {
			// without catalog information we have no meaningful way to describe the service to the requester,
			// and it makes the most sense to simply skip the service in question
			continue
		} else if err != nil {
			log.Warn("could not pull information from service catalog", zap.String("service catalog id", service.ServiceCatalogID), zap.Error(err))
		} else {
			serviceName = serviceResp.Name
		}
		info = append(info, &rpc.PublisherServiceInfo{
			Name:              serviceName,
			ServiceCatalogUrl: servicecatalog.FormatURL(service.ServiceCatalogID),
		})
	}
	return info
}

func statsToRPC(stats *eventstreamstats.EventStreamStatistics) *rpc.EventStreamStats {
	rpcStats := &rpc.EventStreamStats{}
	if stats.MaxEventsPerMinute != nil {
		rpcStats.MaxEventsPerMinute = &wrappers.DoubleValue{Value: *stats.MaxEventsPerMinute}
	}
	if stats.MinEventsPerMinute != nil {
		rpcStats.MinEventsPerMinute = &wrappers.DoubleValue{Value: *stats.MinEventsPerMinute}
	}
	if stats.MeanEventsPerMinute != nil {
		rpcStats.MeanEventsPerMinute = &wrappers.DoubleValue{Value: *stats.MeanEventsPerMinute}
	}
	if stats.MeanEventPayloadSize != nil {
		rpcStats.MeanEventPayloadSize = &wrappers.DoubleValue{Value: *stats.MeanEventPayloadSize}
	}
	return rpcStats
}
