Notify users of pending User Tasks

User Tasks is a new resource that contains issues about an integration.
For now, User Tasks are only created to register issues related to auto
enrolling EC2 instances.
Other types of issues will be registered in the future.

Users must proactively list User Tasks in order to understand existing
issues.
For now, this can only be done using `tctl get user_task`.

An UI for this new resource will be created, but users still need to
navigate to that UI to see what issues exist.

This PR creates a new Notification type which is used to explicitly
notify the user that the Integration has pending User Tasks.

When are users notified about pending User Tasks?
When a User Task is created or updated, if its state is OPEN, then a
notification is created.

Given that we can have dozens of UserTasks for a single integration, we
had to ensure only one Notification exists per Integration.
To do that, the current Notification ID is stored in the Integration's
Status field.
This way, when we try to create a new Notification, we ensure the
existing one is deleted to prevent showing two (or more) Notifications
saying "Your integration needs attention".
All of this is protected by a semaphore lock. The DiscoveryService is
highly async and multiple UserTasks can be created in parallel, and the
locks prevents it.

api/proto/teleport/legacy/types/types.proto

@@ -7544,6 +7544,25 @@ message IntegrationV1 {
     (gogoproto.nullable) = false,
     (gogoproto.jsontag) = "spec"
   ];
+
+  // Status has the current state of the integration.
+  IntegrationStatusV1 Status = 3 [
+    (gogoproto.nullable) = false,
+    (gogoproto.jsontag) = "status"
+  ];
+}
+
+// IntegrationStatusV1 contains the current status for a given integration.
+message IntegrationStatusV1 {
+  // PendingUserTasksNotificationID contains the notification ID that indicates that this integration has unresolved user tasks.
+  string PendingUserTasksNotificationID = 1 [(gogoproto.jsontag) = "pending_user_tasks_notification_id,omitempty"];
+  // NeedsAttentionNotificationExpires contains the expiration date for the notification.
+  // Used to ensure new notifications' expiration is the greater between the current notification and the new one.
+  google.protobuf.Timestamp PendingUserTasksNotificationExpires = 2 [
+    (gogoproto.stdtime) = true,
+    (gogoproto.nullable) = true,
+    (gogoproto.jsontag) = "pending_user_tasks_notification_expires,omitempty"
+  ];
 }
 
 // IntegrationSpecV1 contains properties of all the supported integrations.

api/types/constants.go

@@ -1156,6 +1156,12 @@ const (
 	// NotificationUserCreatedWarningSubKind is the subkind for a user-created warning notification.
 	NotificationUserCreatedWarningSubKind = "user-created-warning"
 
+	// NotificationPendingUserTaskIntegrationSubKind is the subkind for a notification that warns the user about pending User Tasks for a given integration.
+	NotificationPendingUserTaskIntegrationSubKind = "pending-user-task-for-integration"
+	// NotificationIntegrationLabel is the label which contains the name of the integration.
+	// To be used with NotificationUserTaskIntegrationSubKind.
+	NotificationIntegrationLabel = "integration-name"
+
 	// NotificationAccessRequestPendingSubKind is the subkind for a notification for an access request pending review.
 	NotificationAccessRequestPendingSubKind = "access-request-pending"
 	// NotificationAccessRequestApprovedSubKind is the subkind for a notification for a user's access request being approved.

api/types/integration.go

@@ -78,6 +78,11 @@ type Integration interface {
 	GetCredentials() PluginCredentials
 	// WithoutCredentials returns a copy without credentials.
 	WithoutCredentials() Integration
+
+	// GetStatus returns the Integration Status.
+	GetStatus() IntegrationStatusV1
+	// SetStatus sets the Integration Status.
+	SetStatus(IntegrationStatusV1)
 }
 
 var _ ResourceWithLabels = (*IntegrationV1)(nil)
@@ -339,6 +344,22 @@ func (ig *IntegrationV1) SetGitHubIntegrationSpec(spec *GitHubIntegrationSpecV1)
 	}
 }
 
+// GetStatus returns the Integration Status.
+func (ig *IntegrationV1) GetStatus() IntegrationStatusV1 {
+	if ig == nil {
+		return IntegrationStatusV1{}
+	}
+	return ig.Status
+}
+
+// SetStatus sets the Integration Status.
+func (ig *IntegrationV1) SetStatus(s IntegrationStatusV1) {
+	if ig == nil {
+		return
+	}
+	ig.Status = s
+}
+
 // Integrations is a list of Integration resources.
 type Integrations []Integration
 
@@ -393,6 +414,7 @@ func (ig *IntegrationV1) UnmarshalJSON(data []byte) error {
 			GitHub      json.RawMessage `json:"github"`
 			Credentials json.RawMessage `json:"credentials"`
 		} `json:"spec"`
+		Status IntegrationStatusV1 `json:"status"`
 	}{}
 
 	err := json.Unmarshal(data, &d)
@@ -401,6 +423,8 @@ func (ig *IntegrationV1) UnmarshalJSON(data []byte) error {
 	}
 
 	integration.ResourceHeader = d.ResourceHeader
+	integration.Status = d.Status
+
 	if len(d.Spec.Credentials) != 0 {
 		var credentials PluginCredentialsV1
 		if err := protojson.Unmarshal(d.Spec.Credentials, protoadapt.MessageV2Of(&credentials)); err != nil {
@@ -468,9 +492,12 @@ func (ig *IntegrationV1) MarshalJSON() ([]byte, error) {
 			GitHub      GitHubIntegrationSpecV1    `json:"github,omitempty"`
 			Credentials json.RawMessage            `json:"credentials,omitempty"`
 		} `json:"spec"`
+		Status IntegrationStatusV1 `json:"status"`
 	}{}
 
 	d.ResourceHeader = ig.ResourceHeader
+	d.Status = ig.Status
+
 	if ig.Spec.Credentials != nil {
 		data, err := protojson.Marshal(protoadapt.MessageV2Of(ig.Spec.Credentials))
 		if err != nil {

api/types/notifications/notifications.go

@@ -0,0 +1,56 @@
+/*
+Copyright 2024 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package notifications
+
+import (
+	"time"
+
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	headerv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/header/v1"
+	notificationsv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/notifications/v1"
+	"github.com/gravitational/teleport/api/types"
+)
+
+// NewPendingUserTasksIntegrationNotification creates a new GlobalNotification for warning the users about pending UserTasks related to an integration.
+func NewPendingUserTasksIntegrationNotification(integrationName string, expires time.Time) *notificationsv1.GlobalNotification {
+	return &notificationsv1.GlobalNotification{
+		Spec: &notificationsv1.GlobalNotificationSpec{
+			Matcher: &notificationsv1.GlobalNotificationSpec_ByPermissions{
+				ByPermissions: &notificationsv1.ByPermissions{
+					RoleConditions: []*types.RoleConditions{{
+						Rules: []types.Rule{{
+							Resources: []string{types.KindIntegration},
+							Verbs:     []string{types.VerbList, types.VerbRead},
+						}},
+					}},
+				},
+			},
+			Notification: &notificationsv1.Notification{
+				Spec:    &notificationsv1.NotificationSpec{},
+				SubKind: types.NotificationPendingUserTaskIntegrationSubKind,
+				Metadata: &headerv1.Metadata{
+					Labels: map[string]string{
+						types.NotificationTitleLabel:       "Your integration needs attention.",
+						types.NotificationIntegrationLabel: integrationName,
+					},
+					Expires: timestamppb.New(expires),
+				},
+			},
+		},
+	}
+}