Bcpsdems 791 schema registry (#9)

* SASL OAUTHCONFIG for Kafka Broker

* SASL OAUTHCONFIG for Kafka Broker

* bcps authorization and user update

* kafka OauthTokenRefreshCallback Handler

* topic name change

* merge updates

* Merged from build branch - cleanup

* Merged from build branch - cleanup

* Minor changes for dev deployments

* Remove IDP options on login page
Enable different scope/consumer groups

* IDP update

* Add some additional logging

* Minor UI tweaks, test removing a completed migration job

* Helm updates and fixes for typos

* Use vanity URL

* Test auto dev deploy

* 2 replicas by default

* consumer retry logic and architecure diagram for edt

* consumer retry logic and architecure diagram for edt

* Update Readme.md

* Update Readme.md

* Update Readme.md

* manually commit offset store

* Changes to helm charts
and minor UI tweaks
Oauth kafka updates

* Minor change to trigger deploy

* Fix issues with secrets being in appsettings and not environment

* Get all refresh values from environment

* Clear launch - secrets updated

* Helm updates

* Update assigned groups for a user

* Generate user create (provision) messages on new topic for JUSTIN
Add SEQ chart and connect to logging for PIDP and EDT

* launch settings - new key

* Updated dockerfile and new xml config

* Changes for helm, dynamic retry topics and notifications.

* Schema aware producer for user modifications
Helm updates

* Library update

* Remove config value
Base URI cannot start with slash if baseurl has additional paths

* Helm updates for vanity URLS

* Add helm values for registry

* Add client id to producer and consumer

Co-authored-by: Stanley Okeke <[email protected]>
Co-authored-by: Stanley Okeke <[email protected]>

Readme.md

@@ -0,0 +1,3 @@
+# Solution Architecture
+
+<img align="center" width="1110" src="/docs/Future State Solution Architecture.drawio (3).png">

backend/service.edt/.dockerignore

@@ -0,0 +1,2 @@
+bin/
+obj/

backend/service.edt/Data/Configuration/FailedEventLogConfiguration.cs

@@ -0,0 +1,13 @@
+namespace edt.service.Data.Configuration;
+
+using edt.service.ServiceEvents.UserAccountCreation.Models;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+public class FailedLogEventConfiguration : IEntityTypeConfiguration<FailedEventLog>
+{
+    public void Configure(EntityTypeBuilder<FailedEventLog> builder)
+    {
+        //builder.Property(e=>e.EventPayload).HasJsonConversion<>
+    }
+}

backend/service.edt/Data/EdtDataStoreDbContext.cs

@@ -12,6 +12,7 @@ public class EdtDataStoreDbContext : DbContext
     public DbSet<EmailLog> EmailLogs { get; set; } = default!;
     public DbSet<IdempotentConsumer> IdempotentConsumers { get; set; } = default!;
     public DbSet<NotificationAckModel> Notifications { get; set; } = default!;
+    public DbSet<FailedEventLog> FailedEventLogs { get; set; } = default!;
 
     public override int SaveChanges()
     {

backend/service.edt/Data/Migrations/20221129204729_FailedEvent.cs

@@ -0,0 +1,37 @@
+using System;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+#nullable disable
+
+namespace edt.service.Data.Migrations
+{
+    public partial class FailedEvent : Migration
+    {
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.CreateTable(
+                name: "FailedEventLogs",
+                schema: "edt",
+                columns: table => new
+                {
+                    EventId = table.Column<string>(type: "nvarchar(450)", nullable: false),
+                    Producer = table.Column<string>(type: "nvarchar(max)", nullable: true),
+                    ConsumerGroupId = table.Column<string>(type: "nvarchar(max)", nullable: true),
+                    ConsumerId = table.Column<string>(type: "nvarchar(max)", nullable: true),
+                    Created = table.Column<DateTime>(type: "datetime2", nullable: false),
+                    Modified = table.Column<DateTime>(type: "datetime2", nullable: false)
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_FailedEventLogs", x => x.EventId);
+                });
+        }
+
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropTable(
+                name: "FailedEventLogs",
+                schema: "edt");
+        }
+    }
+}

backend/service.edt/Data/Migrations/EdtDataStoreDbContextModelSnapshot.cs

@@ -77,6 +77,31 @@ protected override void BuildModel(ModelBuilder modelBuilder)
                     b.ToTable("EmailLog", "edt");
                 });
 
+            modelBuilder.Entity("edt.service.ServiceEvents.UserAccountCreation.Models.FailedEventLog", b =>
+                {
+                    b.Property<string>("EventId")
+                        .HasColumnType("nvarchar(450)");
+
+                    b.Property<string>("ConsumerGroupId")
+                        .HasColumnType("nvarchar(max)");
+
+                    b.Property<string>("ConsumerId")
+                        .HasColumnType("nvarchar(max)");
+
+                    b.Property<DateTime>("Created")
+                        .HasColumnType("datetime2");
+
+                    b.Property<DateTime>("Modified")
+                        .HasColumnType("datetime2");
+
+                    b.Property<string>("Producer")
+                        .HasColumnType("nvarchar(max)");
+
+                    b.HasKey("EventId");
+
+                    b.ToTable("FailedEventLogs", "edt");
+                });
+
             modelBuilder.Entity("edt.service.ServiceEvents.UserAccountCreation.Models.IdempotentConsumer", b =>
                 {
                     b.Property<string>("MessageId")

backend/service.edt/EdtServiceConfiguration.cs

@@ -1,4 +1,7 @@
 namespace edt.service;
+
+using System.Text.Json;
+
 public class EdtServiceConfiguration
 {
     public static bool IsProduction() => EnvironmentName == Environments.Production;
@@ -12,8 +15,12 @@ public class EdtServiceConfiguration
     public KafkaClusterConfiguration KafkaCluster { get; set; } = new();
     public KeycloakConfiguration Keycloak { get; set; } = new();
     public MailServerConfiguration MailServer { get; set; } = new();
+    public RetryPolicyConfiguration RetryPolicy { get; set; } = new();
     public EdtClientConfiguration EdtClient { get; set; } = new();
 
+    public SchemaRegistryConfiguration SchemaRegistry { get; set; } = new();
+
+
     // ------- Configuration Objects -------
 
     public class AddressAutocompleteClientConfiguration
@@ -31,6 +38,32 @@ public class ConnectionStringConfiguration
         public string EdtDataStore { get; set; } = string.Empty;
     }
 
+
+    public class RetryTopicModel
+    {
+        public int RetryCount { get; set; }
+        public int DelayMinutes { get; set; }
+        public bool NotifyUser { get; set; }
+        public bool NotifyOnEachRetry { get; set; }
+        public int Order { get; set; }
+        public string TopicName { get; set; } = string.Empty;
+        public override string ToString() => JsonSerializer.Serialize(this);
+
+    }
+
+    public class SchemaRegistryConfiguration
+    {
+        public string Url { get; set; } = string.Empty;
+        public string ClientId { get; set; } = string.Empty;
+        public string ClientSecret { get; set; } = string.Empty;
+
+    }
+
+    public class RetryPolicyConfiguration
+    {
+        public string? DeadLetterTopic { get; set; }
+        public List<RetryTopicModel> RetryTopics { get; set; } = new List<RetryTopicModel>();
+    }
     public class ChesClientConfiguration
     {
         public bool Enabled { get; set; }
@@ -45,6 +78,8 @@ public class KafkaClusterConfiguration
         public string BootstrapServers { get; set; } = string.Empty;
         public string ConsumerTopicName { get; set; } = string.Empty;
         public string ProducerTopicName { get; set; } = string.Empty;
+        public string UserModificationTopicName { get; set; } = string.Empty;
+        public string UserCreationTopicName { get; set; } = string.Empty;
         public string SaslOauthbearerTokenEndpointUrl { get; set; } = string.Empty;
         public string SaslOauthbearerProducerClientId { get; set; } = string.Empty;
         public string SaslOauthbearerProducerClientSecret { get; set; } = string.Empty;
@@ -55,6 +90,7 @@ public class KafkaClusterConfiguration
         public string SslKeyLocation { get; set; } = string.Empty;
         public string Scope { get; set; } = "openid";
         public string ConsumerGroupId { get; set; } = "accessrequest-consumer-group";
+        public string RetryConsumerGroupId { get; set; } = "accessrequest-retry-consumer-group";
 
 
 

backend/service.edt/Exceptions/EdtServiceException.cs

@@ -0,0 +1,13 @@
+namespace edt.service.Exceptions;
+
+using System.Runtime.Serialization;
+
+[Serializable]
+public class EdtServiceException : Exception
+{
+    public EdtServiceException() : base() { }
+    public EdtServiceException(string message) : base(message ?? "EDT Service not responding") { }
+
+    protected EdtServiceException(SerializationInfo serializationInfo, StreamingContext streamingContext) : base(serializationInfo, streamingContext) { }
+
+}

backend/service.edt/Exceptions/UserOnboardingException.cs

@@ -0,0 +1,11 @@
+namespace edt.service.Exceptions;
+using System.Runtime.Serialization;
+
+[Serializable]
+public class UserOnboardingException : Exception
+{
+    public UserOnboardingException() : base() { }
+    public UserOnboardingException(string message) : base(message ?? "User not a valid Justin User.") { }
+    public UserOnboardingException(string message, Exception innerException) : base(message ?? "UserId not found.", innerException) { }
+    protected UserOnboardingException(SerializationInfo serializationInfo, StreamingContext streamingContext) : base(serializationInfo, streamingContext) { }
+}