package infrastructure

import (
	"context"
	"database/sql"
	"fmt"
	"strconv"
	"time"

	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/kms"
	"github.com/pkg/errors"
	"github.com/twitchtv/twirp"
	"go.uber.org/zap"

	"code.justin.tv/eventbus/controlplane/infrastructure/routing"
	"code.justin.tv/eventbus/controlplane/infrastructure/rpc"
	"code.justin.tv/eventbus/controlplane/internal/arn"
	"code.justin.tv/eventbus/controlplane/internal/clients/slack"
	"code.justin.tv/eventbus/controlplane/internal/db"
	"code.justin.tv/eventbus/controlplane/internal/environment"
	"code.justin.tv/eventbus/controlplane/internal/ldap"
	"code.justin.tv/eventbus/controlplane/internal/logger"
	"code.justin.tv/eventbus/controlplane/internal/twirperr"
)

type AuthorizedFieldGrantManager interface {
	GrantAuthorizedFieldPublisher(ctx context.Context, iamRoleARN, eventType, environment string) (string, error)
	GrantAuthorizedFieldSubscriber(ctx context.Context, iamRoleARN, eventType, environment, messageName, fieldName string) (string, error)
	Revoke(ctx context.Context, grantID string) error
	FindByGrantID(ctx context.Context, grantID string) (*kms.GrantListEntry, error)
}

type SNSManager interface {
	CreateTopic(ctx context.Context, eventType, environment string) (string, error)
	TopicExists(ctx context.Context, topicARN string) (bool, error)
	AllowAccountPublish(ctx context.Context, topicARN, awsAccountID string) error
}

type InfrastructureService struct {
	Session                     *session.Session
	DB                          db.DB
	AuthorizedFieldGrantManager AuthorizedFieldGrantManager
	RouteConfigActions          *routing.ConfigActions
	SNSManager                  SNSManager

	Slack slack.Slack

	DisableGrants bool

	Logger *logger.Logger
}

func (is *InfrastructureService) RegisterEventDefinitions(ctx context.Context, req *rpc.RegisterEventDefinitionsReq) (*rpc.RegisterEventDefinitionsResp, error) {
	if ldap.User(ctx) != updaterLambdaLDAPUser {
		return nil, twirp.NewError(twirp.NotFound, fmt.Sprintf("must belong to group %s", updaterLambdaLDAPUser))
	}

	environments := environment.ValidEnvironments()
	// Create an event stream for each environment

	// HIGH LEVEL LOGIC AFTER REWORK
	// 1. check if type exists. if not, call CreateWithStreams method and exit
	// 2. if type exists already, call update on the type itself, dont do anything with the streams
	results := make([]*rpc.RegisterEventDefinitionsResp_Result, len(req.EventDefinitions))

	for i, eventDef := range req.EventDefinitions {
		var created bool
		var msgs []string
		var eventStreamsToVerify []*db.EventStream

		eventType, err := is.DB.EventTypeByName(ctx, eventDef.EventType)
		if dbErr, ok := err.(twirperr.DBError); ok && dbErr.EventTypeNotFound() { // need to create the event type and its streams
			eventType = &db.EventType{}
			eventType.Name = eventDef.EventType
			msgs = updateEventTypeFields(eventType, eventDef)
			eventStreamsToVerify, err = is.DB.EventTypeAndStreamsCreate(ctx, eventType, environments)
			if err != nil {
				return nil, err
			}
			created = true
		} else if err != nil { // unexpected error
			return nil, err
		} else { // event type exists and we should update it
			msgs = updateEventTypeFields(eventType, eventDef)
			updateEventTypeInput := &db.EventTypeEditable{
				Name:         eventType.Name,
				Description:  eventType.Description,
				Schema:       eventType.Schema,
				RepoFilePath: eventType.RepoFilePath,
				Deprecated:   eventType.Deprecated,
				LDAPGroup:    eventType.LDAPGroup,
			}
			_, err := is.DB.EventTypeUpdate(ctx, eventType.ID, updateEventTypeInput)
			if err != nil {
				return nil, err
			}
			created = false

			eventStreamsToVerify, err = is.DB.EventStreamsByEventTypeID(ctx, eventType.ID)
			if err != nil {
				return nil, err
			}
		}

		// The following operations are idempotent and unconditionally run on the event streams
		for _, eventStream := range eventStreamsToVerify {
			topicARN, err := is.createEventStreamTopic(ctx, eventStream)
			if err != nil {
				return nil, errors.Wrap(err, "could not process event stream topic")
			}

			eventStream.SNSDetails.SNSTopicARN = topicARN

			authFieldMsgs, err := is.registerAuthorizedFields(ctx, eventStream, eventDef.AuthorizedFields)
			if err != nil {
				return nil, errors.Wrap(err, "could not register authorized fields")
			}
			msgs = append(msgs, authFieldMsgs...)

			err = is.RouteConfigActions.PutEventStreamConfigFile(ctx, *eventStream)
			if err != nil {
				return nil, errors.Wrap(err, "failed at putting S3 config")
			}

		}

		results[i] = &rpc.RegisterEventDefinitionsResp_Result{
			Created:   created,
			Messages:  msgs,
			EventType: eventType.Name,

			// Environment field deprecated in result message
		}
	}

	return &rpc.RegisterEventDefinitionsResp{
		Results: results,
	}, nil
}

func (is *InfrastructureService) createEventStreamTopic(ctx context.Context, eventStream *db.EventStream) (string, error) {
	lease, ctx, err := is.DB.EventStreamAcquireLease(ctx, eventStream.ID, 2*time.Minute)
	if err != nil {
		return "", errors.Wrap(err, "could not acquire event stream lease")
	}

	defer func() {
		if err := is.DB.EventStreamReleaseLease(lease); err != nil {
			is.Logger.Error("could not release event streams lease", zap.Error(err))
		}
	}()

	if eventStream.SNSDetails.SNSTopicARN != "" {
		exists, err := is.SNSManager.TopicExists(ctx, eventStream.SNSDetails.SNSTopicARN)
		if err != nil {
			return "", errors.Wrap(err, "could not check if topic exists")
		} else if exists {
			return eventStream.SNSDetails.SNSTopicARN, nil
		}
	}

	topicARN, err := is.SNSManager.CreateTopic(ctx, eventStream.EventType.Name, eventStream.Environment)
	if err != nil {
		return "", errors.Wrap(err, "could not create topic")
	}

	logger.FromContext(ctx).Info("Created sns topic", zap.Object("eventStream", eventStream), zap.String("TopicArn", topicARN))
	eventStreamInfraUpdate := &db.EventStreamInfraUpdate{
		SNSTopicARN: topicARN,
	}

	if _, err = is.DB.EventStreamUpdateInfra(ctx, lease, eventStream.ID, eventStreamInfraUpdate); err != nil {
		return "", errors.Wrap(err, "could not update event stream infra in db")
	}

	if is.Slack != nil {
		slackMessage := fmt.Sprintf("Created SNS topic %s", topicARN)
		if err = is.Slack.SendMessage(ctx, slackMessage); err != nil {
			is.Logger.Warn("could not send Slack message", zap.Error(err))
		}
	}

	return topicARN, nil
}

func (is *InfrastructureService) registerAuthorizedFields(ctx context.Context, eventStream *db.EventStream, authorizedFields []*rpc.AuthorizedField) ([]string, error) {
	msgs := []string{}

	for _, authFieldDef := range authorizedFields {
		_, err := is.DB.AuthorizedFieldByAuthContext(ctx, eventStream.ID, authFieldDef.MessageName, authFieldDef.FieldName)
		if err == nil { // exists
			continue
		} else if twirpErr, ok := err.(twirperr.DBError); !ok || !twirpErr.AuthorizedFieldNotFound() { // unexpected error
			return msgs, errors.Wrap(err, "error checking authorized field existence")
		}

		authField := &db.AuthorizedField{
			EventStreamID: eventStream.ID,
			MessageName:   authFieldDef.MessageName,
			FieldName:     authFieldDef.FieldName,
		}
		_, err = is.DB.AuthorizedFieldCreate(ctx, authField)
		if err != nil {
			return msgs, errors.Wrap(err, "error creating authorized field")
		}
		msgs = append(msgs, fmt.Sprintf("registered authorized field (%s,%s)", authField.MessageName, authField.FieldName))
	}

	return msgs, nil
}

// https://jira.twitch.com/browse/ASYNC-740
func (is *InfrastructureService) AllowAccountsPublish(ctx context.Context, req *rpc.AllowAccountsPublishReq) (*rpc.AllowAccountsPublishResp, error) {
	return nil, errors.New("not implemented")
}

func (is *InfrastructureService) AllowIAMRolePublish(ctx context.Context, req *rpc.AllowIAMRolePublishReq) (*rpc.AllowIAMRolePublishResp, error) {
	eventStream, err := is.DB.EventStreamByNameAndEnvironment(ctx, req.GetEventType(), req.GetEnvironment())
	if err != nil {
		return nil, dbError(err, "event stream")
	}

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

	log := logger.FromContext(ctx).With(
		zap.String("event_type", eventStream.EventType.Name),
		zap.String("environment", eventStream.Environment),
	)

	iamRole, err := is.DB.IAMRoleByARN(ctx, req.GetIamRole())
	if err != nil {
		return nil, twirperr.Convert(err)
	}

	publications, err := is.DB.PublicationsByEventStreamID(ctx, eventStream.ID)
	if err != nil {
		return nil, errors.Wrap(err, "could not get publications for event stream")
	}

	pubExists := false
	for _, p := range publications {
		if p.IAMRoleID.Valid && p.IAMRoleID.Int64 == int64(iamRole.ID) {
			pubExists = true
			break
		}
	}

	if !pubExists {
		p := &db.Publication{
			EventStreamID: eventStream.ID,
			IAMRoleID: sql.NullInt64{
				Int64: int64(iamRole.ID),
				Valid: true,
			},
		}

		log = log.With(zap.Object("publication", p))

		_, err = is.DB.PublicationCreate(ctx, p)
		if err != nil {
			log.Error(
				"failure adding IAM role to event stream publication",
				zap.Error(err),
			)
			return nil, err
		}
		log.Info("added IAM role to event stream publication")
	}

	// Regardless of whether we added a publication, we check SNS perms and possibly add them.
	awsAccountID, err := arn.AccountID(iamRole.ARN)
	if err != nil {
		return nil, errors.Wrap(err, "could not derive AWS account ID from IAM role arn")
	}

	lease, ctx, err := is.DB.EventStreamAcquireLease(ctx, eventStream.ID, 10*time.Minute)
	if err != nil {
		return nil, errors.Wrap(err, "could not acquire event stream lease")
	}
	defer func() {
		if err := is.DB.EventStreamReleaseLease(lease); err != nil {
			log.Error("could not release event streams lease", zap.Error(err))
		}
	}()

	err = is.SNSManager.AllowAccountPublish(ctx, eventStream.SNSDetails.SNSTopicARN, awsAccountID)
	if err != nil {
		return nil, errors.Wrap(err, "could not grant IAM role publish permissions")
	}

	err = is.createAuthorizedFieldPublisherGrant(ctx, eventStream, iamRole)
	if err != nil {
		return nil, errors.Wrap(err, "could not create publisher authorized field grant")
	}

	return &rpc.AllowIAMRolePublishResp{
		Created: !pubExists,
	}, nil
}

func (is *InfrastructureService) CreateAuthorizedFieldSubscriberGrant(ctx context.Context, req *rpc.CreateAuthorizedFieldSubscriberGrantReq) (*rpc.CreateAuthorizedFieldSubscriberGrantResp, error) {
	eventStream, err := is.DB.EventStreamByNameAndEnvironment(ctx, req.EventType, req.Environment)
	if err != nil {
		return nil, dbError(err, "event stream")
	}

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

	fieldInfo := req.GetAuthorizedField()
	if fieldInfo == nil {
		return nil, twirp.NewError(twirp.InvalidArgument, "authorized field cannot be nil")
	}

	authorizedField, err := is.DB.AuthorizedFieldByAuthContext(ctx, eventStream.ID, fieldInfo.GetMessageName(), fieldInfo.GetFieldName())
	if err != nil {
		return nil, dbError(err, "authorized field")
	}

	iamRole, err := is.DB.IAMRoleByARN(ctx, req.IamRole)
	if err != nil {
		return nil, twirperr.Convert(err)
	}

	var grantID string
	if !is.DisableGrants {
		grantID, err = is.AuthorizedFieldGrantManager.GrantAuthorizedFieldSubscriber(
			ctx,
			iamRole.ARN,
			req.GetEventType(),
			req.GetEnvironment(),
			req.GetAuthorizedField().GetMessageName(),
			req.GetAuthorizedField().GetFieldName(),
		)
		if err != nil {
			return nil, errors.Wrap(err, "could not create kms grant")
		}
	}

	rollback := is.rollbackGrantCreationFunc(ctx, grantID)

	grant, err := is.DB.AuthorizedFieldSubscriberGrant(ctx, iamRole.ID, authorizedField.ID)
	if twirpErr, ok := err.(twirperr.DBError); ok && twirpErr.AuthorizedFieldSubscriberGrantNotFound() {
		grant = &db.AuthorizedFieldSubscriberGrant{
			IAMRoleID:         iamRole.ID,
			AuthorizedFieldID: authorizedField.ID,
		}
		i, err := is.DB.AuthorizedFieldSubscriberGrantCreate(ctx, grant)
		if err != nil {
			return nil, rollback(err, "could not create authorized field subscriber grant")
		}
		grant.ID = i
	} else if err != nil {
		return nil, rollback(err, "could not check for grant existence")
	}

	if grant.KMSGrantID == grantID {
		return &rpc.CreateAuthorizedFieldSubscriberGrantResp{}, nil
	}

	lease, leaseCtx, err := is.DB.AuthorizedFieldSubscriberGrantAcquireLease(ctx, grant.ID, 1*time.Minute) // TODO: tune?
	if err != nil {
		return nil, rollback(err, "unable to acquire authorized field subscriber grant lease")
	}

	defer func() {
		err := is.DB.AuthorizedFieldSubscriberGrantReleaseLease(lease)
		if err != nil {
			log := logger.FromContext(ctx)
			log.Warn("could not release authorized field subscriber grant lease", zap.Object("lease", lease), zap.Object("grant", grant))
		}
	}()

	grantUpdate := &db.AuthorizedFieldSubscriberGrantInfraUpdate{
		KMSGrantID: grantID,
	}
	_, err = is.DB.AuthorizedFieldSubscriberGrantUpdateInfra(leaseCtx, lease, grant.ID, grantUpdate)
	if err != nil {
		return nil, rollback(err, "unable to update authorized field subscriber grant")
	}

	return &rpc.CreateAuthorizedFieldSubscriberGrantResp{}, nil
}

func (is *InfrastructureService) CreateAuthorizedFieldPublisherGrant(ctx context.Context, req *rpc.CreateAuthorizedFieldPublisherGrantReq) (*rpc.CreateAuthorizedFieldPublisherGrantResp, error) {
	return nil, twirp.NewError(twirp.BadRoute, "not implemented")
}

func (is *InfrastructureService) ListPublicationsByEventStream(ctx context.Context, req *rpc.ListPublicationsByEventStreamReq) (*rpc.ListPublicationsByEventStreamResp, error) {
	if !ldap.BelongsToGroup(ctx, "team-eventbus") {
		return nil, twirp.NewError(twirp.NotFound, "not found")
	}

	eventStream, err := is.DB.EventStreamByNameAndEnvironment(ctx, req.EventType, req.Environment)
	if err != nil {
		return nil, dbError(err, "event stream")
	}

	eventType, err := is.DB.EventTypeByName(ctx, req.EventType)
	if err != nil {
		return nil, twirperr.Convert(err)
	}

	pubs, err := is.DB.PublicationsByEventStreamID(ctx, eventStream.ID)
	if err != nil {
		return nil, dbError(err, "publication")
	}

	pubsResult := []*rpc.Publication{}
	for _, pub := range pubs {
		var service *db.Service
		var publisherIAMRole string

		if pub.IAMRoleID.Valid {
			iam, err := is.DB.IAMRoleByID(ctx, int(pub.IAMRoleID.Int64))
			if err != nil {
				return nil, twirperr.Convert(err)
			}

			service, err = is.DB.ServiceByID(ctx, iam.ServiceID)
			if err != nil {
				return nil, dbError(err, "service")
			}
			publisherIAMRole = iam.ARN
		} else if pub.AccountID.Valid {
			acct, err := is.DB.AccountByID(ctx, int(pub.AccountID.Int64))
			if err != nil {
				return nil, dbError(err, "account")
			}

			service, err = is.DB.ServiceByID(ctx, acct.ServiceID)
			if err != nil {
				return nil, dbError(err, "service")
			}
			publisherIAMRole = "missing"
		}

		pubSingle := &rpc.Publication{
			EventType:                 eventType.Name,
			Environment:               eventStream.Environment,
			EventDeprecated:           eventType.Deprecated,
			EventDescription:          eventType.Description,
			SnsTopicArn:               eventStream.SNSDetails.SNSTopicARN,
			PublisherIamRole:          publisherIAMRole,
			PublisherServiceName:      service.Name,
			PublisherServiceCatalogId: service.ServiceCatalogID,
			ServiceId:                 strconv.Itoa(service.ID),
		}
		pubsResult = append(pubsResult, pubSingle)
	}

	return &rpc.ListPublicationsByEventStreamResp{
		Publications: pubsResult,
	}, nil
}

func (is *InfrastructureService) DeletePublication(ctx context.Context, req *rpc.DeletePublicationReq) (*rpc.DeletePublicationResp, error) {
	return nil, twirp.NewError(twirp.BadRoute, "use DeleteIAMRolePublication instead")
}

func (is *InfrastructureService) DeleteIAMRolePublication(ctx context.Context, req *rpc.DeleteIAMRolePublicationReq) (*rpc.DeleteIAMRolePublicationResp, error) {
	eventStream, err := is.DB.EventStreamByNameAndEnvironment(ctx, req.EventType, req.Environment)
	if err != nil {
		return nil, twirperr.Convert(err)
	}

	// Deleting publications permissions for non-deprecated production events is restricted to EventBus admins
	if environment.IsProductionEnv(req.Environment) && !eventStream.Deprecated && !ldap.BelongsToGroup(ctx, ldap.CheatCodesEnabled) {
		return nil, twirp.NewError(twirp.FailedPrecondition, "cannot delete a production publication for an event stream that is not marked as deprecated")
	}

	iamRole, err := is.DB.IAMRoleByARN(ctx, req.IamRoleArn)
	if err != nil {
		return nil, twirperr.Convert(err)
	}

	authorized, err := is.ownsEventTypeOrIAMRole(ctx, &eventStream.EventType, iamRole)
	if err != nil {
		return nil, twirperr.Convert(err)
	} else if !authorized {
		return nil, twirp.NewError(twirp.NotFound, "not found")
	}

	err = is.deleteAuthorizedFieldPublisherGrant(ctx, eventStream, iamRole)
	if err != nil {
		return nil, twirperr.Convert(err)
	}

	pubs, err := is.DB.PublicationsByEventStreamID(ctx, eventStream.ID)
	if err != nil {
		return nil, dbError(err, "publications")
	}

	for _, pub := range pubs {
		if pub.IAMRoleID.Valid && int(pub.IAMRoleID.Int64) == iamRole.ID {
			err = is.DB.PublicationDelete(ctx, pub.ID)
			if err != nil {
				return nil, twirp.NewError(twirp.Internal, "failed to delete publication")
			}
		}
	}

	return &rpc.DeleteIAMRolePublicationResp{}, nil
}

func (is *InfrastructureService) RevokeAuthorizedFieldGrant(ctx context.Context, req *rpc.RevokeAuthorizedFieldGrantReq) (*rpc.RevokeAuthorizedFieldGrantResp, error) {
	return nil, twirp.NewError(twirp.BadRoute, "disabled; use DeleteAuthorizedFieldSubscriberGrant or DeleteIAMRolePublication instead")
}

func (is *InfrastructureService) DeleteAuthorizedFieldGrant(ctx context.Context, req *rpc.DeleteAuthorizedFieldGrantReq) (*rpc.DeleteAuthorizedFieldGrantResp, error) {
	return nil, twirp.NewError(twirp.BadRoute, "disabled; use DeleteAuthorizedFieldSubscriberGrant or DeleteIAMRolePublication instead")
}

func (is *InfrastructureService) DeleteAuthorizedFieldSubscriberGrant(ctx context.Context, req *rpc.DeleteAuthorizedFieldSubscriberGrantReq) (*rpc.DeleteAuthorizedFieldSubscriberGrantResp, error) {
	subGrant := req.GetGrant()
	if subGrant == nil {
		return nil, twirp.NewError(twirp.InvalidArgument, "must provide authorized field subscriber grant")
	} else if subGrant.GetAuthorizedField() == nil {
		return nil, twirp.NewError(twirp.InvalidArgument, "must provide authorized field information")
	}

	iam, err := is.DB.IAMRoleByARN(ctx, subGrant.IamRole)
	if err != nil {
		return nil, twirperr.Convert(err)
	}

	eventStream, err := is.DB.EventStreamByNameAndEnvironment(ctx, subGrant.EventType, subGrant.Environment)
	if err != nil {
		return nil, dbError(err, "event stream")
	}

	authorized, err := is.ownsEventTypeOrIAMRole(ctx, &eventStream.EventType, iam)
	if err != nil {
		return nil, twirperr.Convert(err)
	} else if !authorized {
		return nil, twirp.NewError(twirp.NotFound, "not found")
	}

	authedField, err := is.DB.AuthorizedFieldByAuthContext(ctx, eventStream.ID, subGrant.AuthorizedField.MessageName, subGrant.AuthorizedField.FieldName)
	if err != nil {
		return nil, dbError(err, "authorized field")
	}

	grant, err := is.DB.AuthorizedFieldSubscriberGrant(ctx, iam.ID, authedField.ID)
	if err != nil {
		return nil, dbError(err, "authorized field subscriber grant")
	}

	kmsGrantID := grant.KMSGrantID

	if !is.DisableGrants {
		if kmsGrantID == "" {
			return nil, twirp.NewError(twirp.InvalidArgument, "provided grant had no ID associated with it")
		}

		kmsGrant, err := is.AuthorizedFieldGrantManager.FindByGrantID(ctx, grant.KMSGrantID)
		if err != nil {
			return nil, errors.Wrap(err, "could not check for existing authorized field subscriber grant")
		} else if kmsGrant != nil {
			err := is.AuthorizedFieldGrantManager.Revoke(ctx, kmsGrantID)
			if err != nil {
				return nil, twirp.NewError(twirp.Unknown, "could not revoke kms grant: "+err.Error())
			}
		}
	}

	err = is.DB.AuthorizedFieldSubscriberGrantDeleteByID(ctx, grant.ID)
	if err != nil {
		return nil, twirp.NewError(twirp.Internal, "could not delete subscriber grant from db: "+err.Error())
	}

	return &rpc.DeleteAuthorizedFieldSubscriberGrantResp{
		Deleted: true,
	}, nil
}

func (is *InfrastructureService) DeleteAuthorizedField(ctx context.Context, req *rpc.DeleteAuthorizedFieldReq) (*rpc.DeleteAuthorizedFieldResp, error) {
	if !ldap.BelongsToGroup(ctx, "team-eventbus") {
		return nil, twirp.NewError(twirp.NotFound, "not found")
	}

	eventStream, err := is.DB.EventStreamByNameAndEnvironment(ctx, req.EventType, req.Environment)
	if err != nil {
		return nil, dbError(err, "event stream")
	}

	if !eventStream.Deprecated {
		return nil, twirp.NewError(twirp.FailedPrecondition, "cannot delete an authorized field for an event stream that is not marked as deprecated")
	}

	authField, err := is.DB.AuthorizedFieldByAuthContext(ctx, eventStream.ID, req.AuthorizedField.MessageName, req.AuthorizedField.FieldName)
	if err != nil {
		return nil, dbError(err, "authorized field")
	}

	err = is.DB.AuthorizedFieldDeleteByID(ctx, authField.ID)
	if err != nil {
		return nil, twirp.NewError(twirp.Internal, "could not delete authorized field: "+err.Error())
	}

	return &rpc.DeleteAuthorizedFieldResp{Deleted: true}, nil
}

func (is *InfrastructureService) DeleteEventType(ctx context.Context, req *rpc.DeleteEventTypeReq) (*rpc.DeleteEventTypeResp, error) {
	if !ldap.BelongsToGroup(ctx, "team-eventbus") {
		return nil, twirp.NewError(twirp.NotFound, "not found")
	}

	eventType, err := is.DB.EventTypeByName(ctx, req.EventType)
	if err != nil {
		return nil, twirperr.Convert(err)
	}

	if !eventType.Deprecated {
		return nil, twirp.NewError(twirp.FailedPrecondition, "cannot delete a non deprecated event type")
	}

	eventStreams, err := is.DB.EventStreamsByEventTypeID(ctx, eventType.ID)
	if err != nil {
		return nil, dbError(err, "event streams")
	}

	if len(eventStreams) > 0 {
		return nil, twirp.NewError(twirp.FailedPrecondition, "cannot delete an event type while existing event streams are associated with it")
	}

	err = is.DB.EventTypeDeleteByName(ctx, eventType.Name)
	if err != nil {
		return nil, twirp.NewError(twirp.Internal, "could not delete event type from db: "+err.Error())
	}

	return &rpc.DeleteEventTypeResp{Deleted: true}, nil
}

func (is *InfrastructureService) DeleteEventStreamsForEventType(ctx context.Context, req *rpc.DeleteEventStreamsForEventTypeReq) (*rpc.DeleteEventStreamsForEventTypeResp, error) {
	if !ldap.BelongsToGroup(ctx, "team-eventbus") {
		return nil, twirp.NewError(twirp.NotFound, "not found")
	}

	eventType, err := is.DB.EventTypeByName(ctx, req.EventType)
	if err != nil {
		return nil, twirperr.Convert(err)
	}

	if !eventType.Deprecated {
		return nil, twirp.NewError(twirp.FailedPrecondition, "cannot delete event streams where underlying event type is not deprecated")
	}

	numDeleted, err := is.DB.EventStreamDeleteByEventTypeID(ctx, eventType.ID)
	if err != nil {
		return nil, twirp.NewError(twirp.Internal, "could not delete event stream(s): "+err.Error())
	}

	return &rpc.DeleteEventStreamsForEventTypeResp{
		Deleted:    true,
		NumDeleted: int32(numDeleted),
	}, nil
}

func (is *InfrastructureService) ListAuthorizedFieldSubscriberGrantsByAuthorizedField(ctx context.Context, req *rpc.ListAuthorizedFieldSubscriberGrantsByAuthorizedFieldReq) (*rpc.ListAuthorizedFieldSubscriberGrantsByAuthorizedFieldResp, error) {
	if !ldap.BelongsToGroup(ctx, "team-eventbus") {
		return nil, twirp.NewError(twirp.NotFound, "not found")
	}

	eventStream, err := is.DB.EventStreamByNameAndEnvironment(ctx, req.EventType, req.Environment)
	if err != nil {
		return nil, dbError(err, "event stream")
	}

	authField, err := is.DB.AuthorizedFieldByAuthContext(ctx, eventStream.ID, req.AuthorizedField.MessageName, req.AuthorizedField.FieldName)
	if err != nil {
		return nil, dbError(err, "authorized field")
	}

	grants, err := is.DB.AuthorizedFieldSubscriberGrantsByAuthorizedField(ctx, authField.ID)
	if err != nil {
		return nil, dbError(err, "authorized field subscription grants")
	}

	rpcGrants := []*rpc.AuthorizedFieldSubscriberGrant{}
	for _, grant := range grants {
		iam, err := is.DB.IAMRoleByID(ctx, grant.IAMRoleID)
		if err != nil {
			return nil, twirperr.Convert(err)
		}

		rpcGrant := &rpc.AuthorizedFieldSubscriberGrant{
			AuthorizedField: &rpc.AuthorizedField{
				FieldName:   authField.FieldName,
				MessageName: authField.MessageName,
			},
			Environment: eventStream.Environment,
			EventType:   eventStream.EventType.Name,
			IamRole:     iam.ARN,
			KmsGrantId:  grant.KMSGrantID,
		}
		rpcGrants = append(rpcGrants, rpcGrant)
	}

	return &rpc.ListAuthorizedFieldSubscriberGrantsByAuthorizedFieldResp{
		SubscriberGrants: rpcGrants,
	}, nil
}

func (is *InfrastructureService) createAuthorizedFieldPublisherGrant(ctx context.Context, eventStream *db.EventStream, iamRole *db.IAMRole) error {
	var grantID string
	var err error
	if !is.DisableGrants {
		grantID, err = is.AuthorizedFieldGrantManager.GrantAuthorizedFieldPublisher(
			ctx,
			iamRole.ARN,
			eventStream.EventType.Name,
			eventStream.Environment,
		)
		if err != nil {
			return errors.Wrap(err, "could not create kms grant")
		}
	}

	rollback := is.rollbackGrantCreationFunc(ctx, grantID)

	grant, err := is.DB.AuthorizedFieldPublisherGrant(ctx, iamRole.ID, eventStream.ID)
	if twirpErr, ok := err.(twirperr.DBError); ok && twirpErr.AuthorizedFieldPublisherGrantNotFound() {
		grant = &db.AuthorizedFieldPublisherGrant{
			IAMRoleID:     iamRole.ID,
			EventStreamID: eventStream.ID,
		}
		id, err := is.DB.AuthorizedFieldPublisherGrantCreate(ctx, grant)
		if err != nil {
			return rollback(err, "could not create authorized field publisher grant")
		}
		grant.ID = id
	} else if err != nil {
		return rollback(err, "could not check for grant existence")
	}

	if grant.KMSGrantID == grantID {
		return nil
	}

	lease, leaseCtx, err := is.DB.AuthorizedFieldPublisherGrantAcquireLease(ctx, grant.ID, 1*time.Minute) // TODO: tune?
	if err != nil {
		return rollback(err, "unable to acquire authorized field publisher grant lease")
	}

	defer func() {
		err := is.DB.AuthorizedFieldPublisherGrantReleaseLease(lease)
		if err != nil {
			log := logger.FromContext(ctx)
			log.Warn("could not release authorized field publisher grant lease", zap.Object("lease", lease), zap.Object("grant", grant))
		}
	}()

	grantUpdate := &db.AuthorizedFieldPublisherGrantInfraUpdate{
		KMSGrantID: grantID,
	}
	_, err = is.DB.AuthorizedFieldPublisherGrantUpdateInfra(leaseCtx, lease, grant.ID, grantUpdate)
	if err != nil {
		return rollback(err, "unable to update authorized field subscriber grant")
	}

	return nil
}

// idempotently remove an authorized field publisher grant
func (is *InfrastructureService) deleteAuthorizedFieldPublisherGrant(ctx context.Context, eventStream *db.EventStream, iamRole *db.IAMRole) error {
	grant, err := is.DB.AuthorizedFieldPublisherGrant(ctx, iamRole.ID, eventStream.ID)
	if dbErr, ok := err.(twirperr.DBError); ok && dbErr.AuthorizedFieldPublisherGrantNotFound() {
		return nil
	} else if err != nil {
		return errors.Wrap(err, "could not fetch authorized field publisher grant")
	}

	if !is.DisableGrants {
		if grant.KMSGrantID == "" {
			return nil
		}

		kmsGrant, err := is.AuthorizedFieldGrantManager.FindByGrantID(ctx, grant.KMSGrantID)
		if err != nil {
			return errors.Wrap(err, "could not check for existing authorized field publisher grant")
		} else if kmsGrant != nil {
			err = is.AuthorizedFieldGrantManager.Revoke(ctx, grant.KMSGrantID)
			if err != nil {
				return errors.Wrap(err, "could not revoke kms grant")
			}
		}
	}

	err = is.DB.AuthorizedFieldPublisherGrantDeleteByID(ctx, grant.ID)
	if err != nil {
		return errors.Wrap(err, "could not delete publisher grant from db")
	}

	return nil
}

func (is *InfrastructureService) rollbackGrantCreationFunc(ctx context.Context, grantID string) func(err error, msg string) error {
	return func(apiErr error, msg string) error {
		if is.DisableGrants {
			return nil
		}

		err := is.AuthorizedFieldGrantManager.Revoke(ctx, grantID)
		if err != nil {
			log := logger.FromContext(ctx)
			log.Error("could not revoke kms grant during creation rollback", zap.Error(err), zap.String("apiError", apiErr.Error()))
		}
		return errors.Wrap(apiErr, msg)
	}
}

func (is *InfrastructureService) ownsEventTypeOrIAMRole(ctx context.Context, eventType *db.EventType, iamRole *db.IAMRole) (bool, error) {
	if ldap.BelongsToGroup(ctx, eventType.LDAPGroup) {
		return true, nil
	}

	service, err := is.DB.ServiceByID(ctx, iamRole.ServiceID)
	if err != nil {
		return false, err
	}

	return ldap.BelongsToGroup(ctx, service.LDAPGroup), nil
}

func updateEventTypeFields(model *db.EventType, eventType *rpc.EventDefinition) (messages []string) {
	// helper func for string case to avoid a lot of boilerplate
	updateString := func(dest *string, src string, message string) {
		if src != "" && src != *dest {
			*dest = src
			messages = append(messages, message)
		}
	}
	updateString(&model.Description, eventType.Description, "updated description")
	updateString(&model.Schema, eventType.Schema, "updated schema")
	updateString(&model.RepoFilePath, eventType.RepoFilepath, "updated repo filepath")
	updateString(&model.LDAPGroup, eventType.LdapGroup, "updated ldap group")

	if model.Deprecated != eventType.GetDeprecated() {
		model.Deprecated = eventType.GetDeprecated()
		messages = append(messages, "updated deprecated")
	}

	return messages
}

func dbError(err error, resourceType string) error {
	if err == db.ErrResourceNotFound {
		return twirp.NewError(twirp.NotFound, resourceType+" not found")
	}
	return errors.Wrap(err, "unable to lookup "+resourceType)
}
