package subscriptions

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/aws/aws-sdk-go/aws/arn"
	"github.com/aws/aws-sdk-go/service/sns/snsiface"
	"github.com/pkg/errors"
	"github.com/twitchtv/twirp"
	"go.uber.org/zap"

	"code.justin.tv/eventbus/controlplane/internal/auditlog"
	"code.justin.tv/eventbus/controlplane/internal/backoff"
	"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/rpc"
)

type SNSManager interface {
	AllowAccountSubscribe(ctx context.Context, topicARN, awsAccountID string) error
	DisallowAccountSubscribe(ctx context.Context, topicARN, awsAccountID string) error
	SubscriptionExists(ctx context.Context, queueARN, subscriptionARN string) (bool, error)
	Subscribe(ctx context.Context, topicARN, queueARN string) (string, error)
	Unsubscribe(ctx context.Context, queueARN, subscriptionARN string) error
}

type SubscriptionsService struct {
	Logger     *logger.Logger
	DB         db.DB
	SNS        snsiface.SNSAPI
	SNSManager SNSManager
}

func (ss *SubscriptionsService) Create(ctx context.Context, req *rpc.CreateSubscriptionReq) (*rpc.CreateSubscriptionResp, error) {
	target, err := ss.target(ctx, req.TargetId)
	if err != nil {
		return nil, err
	}

	base := &auditlog.BaseLog{
		ResourceName: req.EventType + "-" + req.Environment,
		ServiceID:    target.ServiceID,
		DB:           ss.DB,
	}

	wrapAndAuditLog := func(err error, msg string) error {
		err = errors.Wrap(err, msg)
		base.LogSubscriptionCreate(ctx, err, nil)
		return err
	}

	if err := ss.authorize(ctx, target); err != nil {
		base.LogSubscriptionCreate(ctx, err, nil)
		return nil, err
	}

	eventStream, err := ss.DB.EventStreamByNameAndEnvironment(ctx, req.EventType, req.Environment)
	if db.IsNotFoundError(err) {
		err = twirp.NewError(twirp.InvalidArgument, fmt.Sprintf("eventStream (%s,%s) does not exist", req.EventType, req.Environment))
		base.LogSubscriptionCreate(ctx, err, nil)
		return nil, err
	} else if err != nil {
		return nil, wrapAndAuditLog(err, "could not find event stream")
	} else if eventStream.Deprecated {
		err = twirp.NewError(twirp.InvalidArgument, fmt.Sprintf("eventstream (%s, %s) is deprecated", req.EventType, req.Environment))
		base.LogSubscriptionCreate(ctx, err, nil)
		return nil, err
	}

	// For idempotency's sake, check for an existing in-progress subscription object to use.
	sub, err := ss.DB.SubscriptionByEventStreamIDAndSubscriptionTargetID(ctx, eventStream.ID, target.ID)
	if db.IsNotFoundError(err) {
		sub = &db.Subscription{
			SubscriptionTargetID: target.ID,
			EventStreamID:        eventStream.ID,
		}

		subID, err := ss.DB.SubscriptionCreate(ctx, sub)
		if err != nil {
			return nil, wrapAndAuditLog(err, "could not create subscription DB entry")
		}
		sub.ID = subID
	} else if err != nil {
		return nil, wrapAndAuditLog(err, "unknown db error accessing sub")
	}

	eventStreamLease, subscriptionLease, ctx, err := ss.acquireLeases(ctx, eventStream.ID, sub.ID)
	if err != nil {
		return nil, errors.Wrap(err, "could not acquire infrastructure leases")
	}

	defer ss.releaseLeases(eventStreamLease, subscriptionLease)

	var subExists bool
	if sub.SNSSubscriptionARN != "" {
		subExists, err = ss.SNSManager.SubscriptionExists(ctx, eventStream.SNSDetails.SNSTopicARN, sub.SNSSubscriptionARN)
		if err != nil {
			return nil, errors.Wrap(err, "could not check for prior subscription existence")
		}
	}

	if !subExists {
		parsedTargetARN, err := arn.Parse(target.SQSDetails.SQSQueueARN)
		if err != nil {
			return nil, errors.Wrap(err, "could not parse subscription target arn")
		}

		err = ss.SNSManager.AllowAccountSubscribe(ctx, eventStream.SNSDetails.SNSTopicARN, parsedTargetARN.AccountID)
		if err != nil {
			return nil, errors.Wrap(err, "could not grant subscription target subscribe permissions")
		}

		var subscriptionARN string
		err = backoff.RetryWithBackoff(ctx, "sns subscribe", 6, func() error {
			subscriptionARN, err = ss.SNSManager.Subscribe(ctx, eventStream.SNSDetails.SNSTopicARN, target.SQSDetails.SQSQueueARN)
			return err
		})
		if err != nil {
			return nil, errors.Wrap(err, "could not subscribe target to event stream")
		}

		subUpdate := &db.SubscriptionInfraUpdate{
			SNSSubscriptionARN: subscriptionARN,
			Status:             rpc.SubscriptionStatus_SUBSCRIPTION_STATUS_ENABLED.String(),
			Error:              "",
		}

		_, err = ss.DB.SubscriptionUpdateInfra(ctx, subscriptionLease, sub.ID, subUpdate)
		if err != nil {
			return nil, errors.Wrapf(err, "unable to update subscription status for subscription %d", sub.ID)
		}
	}

	base.LogSubscriptionCreate(ctx, nil, sub)

	return &rpc.CreateSubscriptionResp{
		Subscription: subToRPC(sub, eventStream),
	}, nil
}

func (ss *SubscriptionsService) Delete(ctx context.Context, req *rpc.DeleteSubscriptionReq) (*rpc.DeleteSubscriptionResp, error) {
	target, err := ss.target(ctx, req.TargetId)
	if err != nil {
		return nil, err
	}

	base := &auditlog.BaseLog{
		ResourceName: req.EventType + "-" + req.Environment,
		ServiceID:    target.ServiceID,
		DB:           ss.DB,
	}

	if err := ss.authorize(ctx, target); err != nil {
		base.LogSubscriptionDelete(ctx, err)
		return nil, err
	}

	eventStream, err := ss.DB.EventStreamByNameAndEnvironment(ctx, req.EventType, req.Environment)
	if err != nil {
		err = errors.Wrapf(err, "could not get event stream %q - %q for subscription delete", req.EventType, req.Environment)
		base.LogSubscriptionDelete(ctx, err)
		return nil, err
	}

	sub, err := ss.DB.SubscriptionByEventStreamIDAndSubscriptionTargetID(ctx, eventStream.ID, target.ID)
	if err != nil {
		err = errors.Wrap(err, "could not fetch subscription")
		base.LogSubscriptionDelete(ctx, err)
		return nil, err
	}

	eventStreamLease, subscriptionLease, ctx, err := ss.acquireLeases(ctx, eventStream.ID, sub.ID)
	if err != nil {
		return nil, errors.Wrap(err, "could not acquire infrastructure leases")
	}

	defer ss.releaseLeases(eventStreamLease, subscriptionLease)

	var subExists bool
	if sub.SNSSubscriptionARN != "" {
		subExists, err = ss.SNSManager.SubscriptionExists(ctx, eventStream.SNSDetails.SNSTopicARN, sub.SNSSubscriptionARN)
		if err != nil {
			return nil, errors.Wrap(err, "could not check for prior subscription existence")
		}
	}

	if subExists {
		parsedSubscriptionARN, err := arn.Parse(sub.SNSSubscriptionARN)
		if err != nil {
			return nil, errors.Wrap(err, "could not parse subscription target arn")
		}

		err = ss.SNSManager.Unsubscribe(ctx, target.SQSDetails.SQSQueueARN, sub.SNSSubscriptionARN)
		if err != nil {
			return nil, errors.Wrap(err, "could not unsubscribe")
		}

		err = ss.SNSManager.DisallowAccountSubscribe(ctx, eventStream.SNSDetails.SNSTopicARN, parsedSubscriptionARN.AccountID)
		if err != nil {
			return nil, errors.Wrap(err, "could not revoke subscription target subscribe permissions")
		}

		err = ss.DB.SubscriptionDelete(ctx, subscriptionLease, sub.ID)
		if err != nil {
			return nil, errors.Wrap(err, "could not delete subscription")
		}
	}

	base.LogSubscriptionDelete(ctx, nil)
	return &rpc.DeleteSubscriptionResp{}, nil
}

func (ss *SubscriptionsService) GetSubscriptionsForTarget(ctx context.Context, req *rpc.GetSubscriptionsForTargetReq) (*rpc.GetSubscriptionsForTargetResp, error) {
	target, err := ss.target(ctx, req.TargetId)
	if err != nil {
		return nil, err
	}

	if authErr := ss.authorize(ctx, target); authErr != nil {
		return nil, authErr
	}

	subs, err := ss.DB.SubscriptionsBySubscriptionTargetID(ctx, target.ID)
	if err != nil {
		return nil, errors.Wrapf(err, "could not fetch subscriptions for target %d", target.ID)
	}

	resultSubs, err := ss.withEvents(ctx, subs)
	if err != nil {
		return nil, errors.Wrap(err, "could not fetch events for subscriptions")
	}

	return &rpc.GetSubscriptionsForTargetResp{
		Subscriptions: resultSubs,
	}, nil
}

func (ss *SubscriptionsService) withEvents(ctx context.Context, subs []*db.Subscription) ([]*rpc.Subscription, error) {
	resultSubs := make([]*rpc.Subscription, len(subs))
	for i, sub := range subs {
		eventStream, err := ss.DB.EventStreamByID(ctx, sub.EventStreamID)
		if err != nil {
			return nil, errors.Wrapf(err, "could not get event stream for subscription conversion for the twirp api")
		}
		resultSubs[i] = subToRPC(sub, eventStream)
	}

	return resultSubs, nil
}

func (ss *SubscriptionsService) target(ctx context.Context, targetIDStr string) (*db.SubscriptionTarget, error) {
	targetID, err := strconv.Atoi(targetIDStr)
	if err != nil {
		return nil, errors.Wrap(err, "could not convert target id in authorize")
	}

	target, err := ss.DB.SubscriptionTargetByID(ctx, targetID)
	if err == db.ErrSubscriptionTargetNotFound {
		return nil, twirp.NewError(twirp.NotFound, err.Error())
	} else if err != nil {
		return nil, err
	}
	return target, nil
}

// authorized returns a twirp error in one of two cases
// 1. returns twirp.PermissionDenied if the requestor is not authorized to perform the action
// 2. returns twirp.NotFound if either the target or service in the authorization resolution chain do not exist
// 3. returns base error (resolved to twirp.InternalServerError) for any unexpected error during lookup
func (ss *SubscriptionsService) authorize(ctx context.Context, target *db.SubscriptionTarget) error {
	service, err := ss.DB.ServiceByID(ctx, target.ServiceID)
	if err == db.ErrServiceNotFound {
		return twirp.NewError(twirp.NotFound, err.Error())
	} else if err != nil {
		return err
	}

	if !ldap.BelongsToGroup(ctx, service.LDAPGroup) {
		return twirp.NewError(twirp.PermissionDenied, "access denied")
	}
	return nil
}

func (ss *SubscriptionsService) acquireLeases(ctx context.Context, eventStreamID, subscriptionID int) (db.AWSLease, db.AWSLease, context.Context, error) {
	eventStreamLease, ctx, err := ss.DB.EventStreamAcquireLease(ctx, eventStreamID, 10*time.Minute)
	if err != nil {
		return nil, nil, nil, errors.Wrapf(err, "could not acquire lease for event stream %d", eventStreamID)
	}

	subscriptionLease, ctx, err := ss.DB.SubscriptionAcquireLease(ctx, subscriptionID, 10*time.Minute)
	if err != nil {
		return nil, nil, nil, errors.Wrapf(err, "could not acquire lease for subscription %d", subscriptionID)
	}

	return eventStreamLease, subscriptionLease, ctx, nil
}

func (ss *SubscriptionsService) releaseLeases(eventStreamLease, subscriptionLease db.AWSLease) {
	if err := ss.DB.SubscriptionReleaseLease(subscriptionLease); err != nil {
		ss.logError("could not release subscription lease", err)
	}
	if err := ss.DB.EventStreamReleaseLease(eventStreamLease); err != nil {
		ss.logError("could not release event streams lease", err)
	}
}

func (ss *SubscriptionsService) logError(msg string, err error) {
	if ss.Logger != nil {
		ss.Logger.Error(msg, zap.Error(err))
	}
}
