refactor(kafka): use Nais API for topic access grants

internal/issues/issues.go

@@ -46,7 +46,7 @@ func GetAll(ctx context.Context, teamSlug string, issueFilter gql.IssueFilter) (
 			id
 			severity
 			message
-			... on ExternalIngressCriticalVulnerabilityIssue {
+			... on ExternalIngressUrgentVulnerabilityIssue {
 				workload {
 					name
 					__typename
@@ -190,7 +190,7 @@ func GetAll(ctx context.Context, teamSlug string, issueFilter gql.IssueFilter) (
 			i.ResourceType = c.Unleash.GetTypename()
 		case *gql.GetAllIssuesTeamIssuesIssueConnectionNodesVulnerableImageIssue:
 			setWorkloadResource(c.GetWorkload())
-		case *gql.GetAllIssuesTeamIssuesIssueConnectionNodesExternalIngressCriticalVulnerabilityIssue:
+		case *gql.GetAllIssuesTeamIssuesIssueConnectionNodesExternalIngressUrgentVulnerabilityIssue:
 			setWorkloadResource(c.GetWorkload())
 		case *gql.GetAllIssuesTeamIssuesIssueConnectionNodesWorkloadProblemIssue:
 			setWorkloadResource(c.GetWorkload())

internal/kafka/command/flag/flag.go

@@ -2,24 +2,55 @@ package flag
 
 import (
 	"context"
+	"slices"
+	"strings"
 
 	"github.com/nais/cli/internal/flags"
 	"github.com/nais/cli/internal/labels"
 	"github.com/nais/naistrix"
 )
 
-type Kafka struct {
-	*flags.GlobalFlags
-}
-
-type Output string
+type (
+	Output                string
+	CredentialsOutput     string
+	KafkaTopicGrantAccess string
+)
 
-var _ naistrix.FlagAutoCompleter = (*Output)(nil)
+var (
+	_ naistrix.FlagAutoCompleter = (*Output)(nil)
+	_ naistrix.FlagAutoCompleter = (*CredentialsOutput)(nil)
+	_ naistrix.FlagAutoCompleter = (*KafkaTopicGrantAccess)(nil)
+)
 
 func (o *Output) AutoComplete(context.Context, *naistrix.Arguments, string, any) ([]string, string) {
 	return []string{"table", "json"}, "Available output formats."
 }
 
+func (o *CredentialsOutput) AutoComplete(context.Context, *naistrix.Arguments, string, any) ([]string, string) {
+	return []string{"env", "kcat", "java"}, "Available output formats."
+}
+
+func (a *KafkaTopicGrantAccess) AutoComplete(context.Context, *naistrix.Arguments, string, any) ([]string, string) {
+	return []string{"read", "write", "readwrite"}, "Available access levels."
+}
+
+func (a *KafkaTopicGrantAccess) Validate() error {
+	valid := []string{"read", "write", "readwrite"}
+	if a == nil {
+		return naistrix.Errorf("access level is required, must be one of: %s", strings.Join(valid, ", "))
+	}
+
+	if !slices.Contains(valid, string(*a)) {
+		return naistrix.Errorf("invalid access level: %q, must be one of: %s", *a, strings.Join(valid, ", "))
+	}
+
+	return nil
+}
+
+type Kafka struct {
+	*flags.GlobalFlags
+}
+
 type List struct {
 	*Kafka
 	Output Output              `name:"output" short:"o" usage:"Format output (table or json)."`
@@ -28,14 +59,6 @@ type List struct {
 
 func (*List) LabelFacetResource() string { return "kafkaTopics" }
 
-type CredentialsOutput string
-
-var _ naistrix.FlagAutoCompleter = (*CredentialsOutput)(nil)
-
-func (o *CredentialsOutput) AutoComplete(context.Context, *naistrix.Arguments, string, any) ([]string, string) {
-	return []string{"env", "kcat", "java"}, "Available output formats."
-}
-
 type Credentials struct {
 	*Kafka
 	TTL    string            `name:"ttl" usage:"Time-to-live for the credentials (e.g. '1d', '7d'). Maximum 30 days."`
@@ -44,5 +67,5 @@ type Credentials struct {
 
 type GrantAccess struct {
 	*Kafka
-	Access string `name:"access" short:"a" usage:"Access |LEVEL| (readwrite, read and write)."`
+	Access KafkaTopicGrantAccess `name:"access" short:"a" usage:"Access |LEVEL| (readwrite, read and write)."`
 }

internal/kafka/command/grant_access.go

@@ -2,11 +2,12 @@ package command
 
 import (
 	"context"
+	"strings"
 
-	"github.com/nais/cli/internal/aiven"
+	"github.com/nais/cli/internal/kafka"
 	"github.com/nais/cli/internal/kafka/command/flag"
+	"github.com/nais/cli/internal/naisapi/gql"
 	"github.com/nais/cli/internal/validation"
-	nais_kafka "github.com/nais/liberator/pkg/apis/kafka.nais.io/v1"
 	"github.com/nais/naistrix"
 )
 
@@ -24,38 +25,28 @@ func grantAccess(parentFlags *flag.Kafka) *naistrix.Command {
 			{Name: "username"},
 			{Name: "topic"},
 		},
-		ValidateFunc: validation.RequireTeam(grantAccessTopicFlags),
+		ValidateFunc: naistrix.ValidateFuncs(
+			validation.RequireTeamAndEnvironment(grantAccessTopicFlags),
+			func(context.Context, *naistrix.Arguments) error {
+				return grantAccessTopicFlags.Access.Validate()
+			},
+		),
 		RunFunc: func(ctx context.Context, args *naistrix.Arguments, out *naistrix.OutputWriter) error {
-			access := grantAccessTopicFlags.Access
-			namespace := grantAccessTopicFlags.Team
 			topicName := args.Get("topic")
-			username := kafkaApplicationName(args.Get("username"))
-
-			if err := aiven.ValidAclPermission(access); err != nil {
-				return err
-			}
-
-			newAcl := nais_kafka.TopicACL{
-				Team:        namespace,
-				Application: username,
-				Access:      access,
-			}
-			accessResult, err := aiven.GrantAccessToTopic(ctx, namespace, topicName, string(grantAccessTopicFlags.Environment), newAcl)
-			if err != nil {
-				return err
+			subject := kafkaApplicationName(args.Get("username"))
+			grant := gql.KafkaTopicGrantInput{
+				Subject:  subject,
+				TeamName: grantAccessTopicFlags.Team,
+				Access:   gql.KafkaTopicGrantAccess(strings.ToUpper(string(grantAccessTopicFlags.Access))),
 			}
 
-			if accessResult.AlreadyAdded {
-				out.Printf(
-					"ACL entry already exists for '%s/%s' on topic %s/%s.",
-					newAcl.Application, newAcl.Access, namespace, topicName,
-				)
-				return nil
+			if err := kafka.GrantAccessToKafkaTopic(ctx, topicName, grantAccessTopicFlags.Team, grantAccessTopicFlags.Environment, grant); err != nil {
+				return naistrix.Errorf("Unable to grant access: %s", err)
 			}
 
 			out.Printf(
-				"ACL added for '%s', with access '%s' on topic '%s/%s'.",
-				newAcl.Application, newAcl.Access, namespace, topicName,
+				"ACL added for %q, with access %q on topic \"%s/%s\".\n",
+				subject, grantAccessTopicFlags.Access, grantAccessTopicFlags.Team, topicName,
 			)
 			return nil
 		},

internal/kafka/kafka.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"sort"
 
+	"github.com/nais/cli/internal/flags"
 	"github.com/nais/cli/internal/naisapi"
 	"github.com/nais/cli/internal/naisapi/gql"
 )
@@ -68,3 +69,33 @@ func GetTeamTopics(ctx context.Context, team string, environment string, labels
 
 	return ret, nil
 }
+
+func GrantAccessToKafkaTopic(ctx context.Context, topicName, teamSlug string, environmentName flags.Environment, grant gql.KafkaTopicGrantInput) error {
+	_ = `# @genqlient
+		mutation GrantAccessToKafkaTopic(
+			$topicName: String!
+			$teamSlug: Slug!,
+			$environmentName: String!,
+			$grant: KafkaTopicGrantInput!,
+		) {
+			updateKafkaTopic(
+				input: { name: $topicName, teamSlug: $teamSlug, environmentName: $environmentName, addGrants: [$grant] }
+			) {
+				kafkaTopic {
+					id
+				}
+			}
+		}
+	`
+
+	client, err := naisapi.GraphqlClient(ctx)
+	if err != nil {
+		return err
+	}
+
+	if _, err = gql.GrantAccessToKafkaTopic(ctx, client, topicName, teamSlug, string(environmentName), grant); err != nil {
+		return err
+	}
+
+	return nil
+}