Notification Flow in Microservices

 In a microservices architecture, services are decoupled. So if you want to notify a user about an event (like "order processed"), you usually use an asynchronous messaging system rather than calling services directly.

Flow Overview

1. Event happens in a microservice
   Example: OrderService processes an order.

   var orderProcessedEvent = new OrderProcessedEvent { OrderId = 123, UserId = 456 };
   

2. Publish Event to Message Broker

   • Microservice publishes the event to Azure Service Bus topic/queue.

   • Topics allow multiple subscribers, queues deliver to one consumer.

3. Notification Service Subscribes

   • A dedicated Notification Microservice subscribes to the topic/queue.

   • When a message arrives, it triggers notification logic (email, SMS, push).

4. Send Notification to Users

   • Notification Service decides which users should get notified.

   • It uses user preferences/configurations stored in a database.

5. Delivery

   • Email: via SMTP, SendGrid, or Azure Communication Services.

   • Push: via Firebase or Azure Notification Hub.

   • SMS: via Twilio, Azure Communication Services, etc.

---

2. Azure Service Bus Setup

• Queues: point-to-point (one consumer).

• Topics & Subscriptions: publish-subscribe pattern (multiple consumers can get the same event).

Example:

• OrderProcessedTopic

  • Subscription 1 → NotificationService

  • Subscription 2 → AnalyticsService

// Publishing
var client = new ServiceBusClient(connectionString);
var sender = client.CreateSender("OrderProcessedTopic");
await sender.SendMessageAsync(new ServiceBusMessage(JsonSerializer.Serialize(orderProcessedEvent)));

// Subscribing in Notification Service
var processor = client.CreateProcessor("OrderProcessedTopic", "NotificationSubscription");
processor.ProcessMessageAsync += async args => {
    var body = args.Message.Body.ToString();
    var orderEvent = JsonSerializer.Deserialize<OrderProcessedEvent>(body);

    // Send notification logic
    await SendEmailNotification(orderEvent.UserId, "Order Processed");
};

---

3. Email Notification Service

Components:

1. Email Templates – HTML/Plain text templates.

2. SMTP/Email Provider Configuration – SMTP settings (host, port, username, password) or a service like SendGrid/Azure Communication Services.

3. Email Sending Logic – Called by the Notification Service when a message is received.

Sample code using SMTP:

public async Task SendEmailNotification(int userId, string message)
{
    var user = await dbContext.Users.FindAsync(userId);
    if (user == null || string.IsNullOrEmpty(user.Email)) return;

    var smtp = new SmtpClient("smtp.yourmail.com")
    {
        Port = 587,
        Credentials = new NetworkCredential("username", "password"),
        EnableSsl = true,
    };

    var mail = new MailMessage("noreply@yourapp.com", user.Email, "Order Update", message);
    await smtp.SendMailAsync(mail);
}

---

4. Configuring Users for Notifications

1. User Preferences Table

   • Table: UserNotifications

   • Columns: UserId, NotificationType, IsEnabled, Email, PushToken, PhoneNumber

   UserId | NotificationType | IsEnabled | Email           | PushToken | PhoneNumber
   456    | OrderProcessed   | 1         | user@example.com | xyz       | 9999999999
   

2. Check Before Sending

   • Before sending, check if the user wants that type of notification:

   var config = dbContext.UserNotifications
                 .Where(u => u.UserId == userId && u.NotificationType == "OrderProcessed" && u.IsEnabled)
                 .FirstOrDefault();
   if(config != null) SendEmail(...);
   

3. Optional UI – Let users manage preferences in Settings → Notification Preferences.

---

5. Overall Flow Diagram

[OrderService] --(event)--> [Azure Service Bus Topic] --> [NotificationService]
                                                        |--> Email/SMS/Push
                                                        |--> Uses User Preferences from DB

✅ Key Points:

• Use Service Bus for decoupling services.

• Use Notification Service as a central microservice.

• Use user preferences to decide who gets notified.

• Email provider can be SMTP, SendGrid, or Azure Communication Services.

• Optional: extend for SMS, push notifications, or mobile apps.

🔑 Example: Ocelot API Gateway with JWT Authentication

 

1️⃣ Create Solution Structure

We’ll create 3 projects:

1. AuthService → issues JWT tokens.

2. ProductService → sample microservice.

3. ApiGateway → Ocelot API Gateway.

dotnet new webapi -n AuthService
dotnet new webapi -n ProductService
dotnet new webapi -n ApiGateway

---

2️⃣ Implement AuthService (JWT Token Issuer)

Install NuGet packages

dotnet add package Microsoft.AspNetCore.Authentication.JwtBearer
dotnet add package System.IdentityModel.Tokens.Jwt

Add Token Generation (AuthController.cs)

using Microsoft.AspNetCore.Mvc;
using Microsoft.IdentityModel.Tokens;
using System.IdentityModel.Tokens.Jwt;
using System.Security.Claims;
using System.Text;

[ApiController]
[Route("api/[controller]")]
public class AuthController : ControllerBase
{
    [HttpPost("login")]
    public IActionResult Login(string username, string password)
    {
        // Simple validation (replace with real DB check)
        if (username == "admin" && password == "password")
        {
            var tokenHandler = new JwtSecurityTokenHandler();
            var key = Encoding.ASCII.GetBytes("SuperSecretKeyForJwt123456"); // store securely in secrets manager

            var tokenDescriptor = new SecurityTokenDescriptor
            {
                Subject = new ClaimsIdentity(new[] { new Claim("role", "Admin") }),
                Expires = DateTime.UtcNow.AddMinutes(30),
                SigningCredentials = new SigningCredentials(new SymmetricSecurityKey(key), SecurityAlgorithms.HmacSha256Signature)
            };

            var token = tokenHandler.CreateToken(tokenDescriptor);
            return Ok(new { token = tokenHandler.WriteToken(token) });
        }

        return Unauthorized();
    }
}

---

3️⃣ Implement ProductService (Protected Microservice)

Add a Controller (ProductsController.cs)

using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;

[ApiController]
[Route("api/[controller]")]
public class ProductsController : ControllerBase
{
    [Authorize]
    [HttpGet]
    public IActionResult GetProducts()
    {
        return Ok(new[]
        {
            new { Id = 1, Name = "T-shirt", Price = 499 },
            new { Id = 2, Name = "Jeans", Price = 999 }
        });
    }
}

Configure JWT Authentication in Program.cs

using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.IdentityModel.Tokens;
using System.Text;

var builder = WebApplication.CreateBuilder(args);

var key = Encoding.ASCII.GetBytes("SuperSecretKeyForJwt123456");

builder.Services.AddAuthentication(x =>
{
    x.DefaultAuthenticateScheme = JwtBearerDefaults.AuthenticationScheme;
    x.DefaultChallengeScheme = JwtBearerDefaults.AuthenticationScheme;
})
.AddJwtBearer(x =>
{
    x.RequireHttpsMetadata = false;
    x.SaveToken = true;
    x.TokenValidationParameters = new TokenValidationParameters
    {
        ValidateIssuerSigningKey = true,
        IssuerSigningKey = new SymmetricSecurityKey(key),
        ValidateIssuer = false,
        ValidateAudience = false
    };
});

builder.Services.AddControllers();
var app = builder.Build();

app.UseAuthentication();
app.UseAuthorization();

app.MapControllers();
app.Run();

---

4️⃣ Configure API Gateway (Ocelot)

Install Ocelot

dotnet add package Ocelot
dotnet add package Microsoft.AspNetCore.Authentication.JwtBearer

Add ocelot.json

{
  "Routes": [
    {
      "DownstreamPathTemplate": "/api/products",
      "DownstreamScheme": "http",
      "DownstreamHostAndPorts": [
        { "Host": "localhost", "Port": 5002 }   // ProductService port
      ],
      "UpstreamPathTemplate": "/products",
      "UpstreamHttpMethod": [ "Get" ],
      "AuthenticationOptions": {
        "AuthenticationProviderKey": "TestKey",
        "AllowedScopes": []
      }
    }
  ],
  "GlobalConfiguration": {
    "BaseUrl": "https://localhost:5000"
  }
}

Configure Program.cs in ApiGateway

using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.IdentityModel.Tokens;
using Ocelot.DependencyInjection;
using Ocelot.Middleware;
using System.Text;

var builder = WebApplication.CreateBuilder(args);

var key = Encoding.ASCII.GetBytes("SuperSecretKeyForJwt123456");

builder.Services.AddAuthentication(options =>
{
    options.DefaultAuthenticateScheme = JwtBearerDefaults.AuthenticationScheme;
    options.DefaultChallengeScheme = JwtBearerDefaults.AuthenticationScheme;
})
.AddJwtBearer("TestKey", x =>
{
    x.RequireHttpsMetadata = false;
    x.SaveToken = true;
    x.TokenValidationParameters = new TokenValidationParameters
    {
        ValidateIssuerSigningKey = true,
        IssuerSigningKey = new SymmetricSecurityKey(key),
        ValidateIssuer = false,
        ValidateAudience = false
    };
});

builder.Configuration.AddJsonFile("ocelot.json", optional: false, reloadOnChange: true);
builder.Services.AddOcelot();

var app = builder.Build();

app.UseAuthentication();
app.UseAuthorization();
await app.UseOcelot();

app.Run();

---

5️⃣ Test Flow

1. Get Token

   POST https://localhost:5001/api/auth/login
   Body: { "username": "admin", "password": "password" }
   

   Response → { "token": "eyJhbGci..." }

2. Call Product API via Gateway

   GET https://localhost:5000/products
   Authorization: Bearer eyJhbGci...
   

   ✅ Response → [ { "Id": 1, "Name": "T-shirt", "Price": 499 }, ... ]

3. Without Token → 401 Unauthorized.

---

🚀 Summary

• AuthService issues JWT.

• ProductService validates JWT.

• ApiGateway (Ocelot) sits in front, validates tokens, and routes traffic.

How to Implement MicroServices in .Net Core Application- Point-to-point to explanation

 

1) Service boundaries & contracts

• One business capability per service (e.g., Orders, Payments, Catalog). Each owns its own database.
• API surface: REST (JSON) or gRPC for synchronous calls; events (Kafka/RabbitMQ/Azure Service Bus) for async integration.
• Version your APIs (e.g., /v1), keep backward compatible changes, use OpenAPI/Swagger and contract tests (Pact).

2) Resilience patterns (timeouts, retries, circuit breaker, bulkhead)

Use HttpClientFactory + Polly on every outbound call:

builder.Services.AddHttpClient("CatalogClient", c =>

{

    c.BaseAddress = new Uri(config["CatalogUrl"]);

    c.Timeout = TimeSpan.FromSeconds(3);

})

.AddTransientHttpErrorPolicy(p => p.WaitAndRetryAsync(

    retryCount: 3,

    sleepDurationProvider: attempt => TimeSpan.FromMilliseconds(200 * Math.Pow(2, attempt)), // expo backoff

    onRetry: (outcome, ts, attempt, ctx) =>

        logger.LogWarning(outcome.Exception, "Retry {Attempt}", attempt)))

.AddTransientHttpErrorPolicy(p => p.CircuitBreakerAsync(

    handledEventsAllowedBeforeBreaking: 5,

    durationOfBreak: TimeSpan.FromSeconds(30)))

.AddPolicyHandler(Policy.BulkheadAsync<HttpResponseMessage>(maxParallelization: 50, maxQueuingActions: 100));

Rules of thumb

• Always set timeouts on I/O.
• Prefer idempotent operations; send Idempotency-Key headers for POSTs where possible.
• Use cancellation tokens end-to-end.

3) Circuit breaker semantics

• Trip when a dependency is unhealthy; fail fast and return a cached/partial response or friendly error.
• Expose breaker state via metrics and health checks (see §5).

4) Data & database handling (consistency without distributed transactions)

• Database per service (EF Core or Dapper). No cross-service joins.
• Propagate data between services using domain events—not shared tables.
• Ensure atomicity with the Outbox pattern:
1. Write your entity and an OutboxMessage row in the same DB transaction.
2. A background Outbox Publisher reads unpublished rows and emits them to the broker, then marks them sent.

// Inside your OrderService command handler

await using var tx = db.Database.BeginTransactionAsync();

db.Orders.Add(order);

db.Outbox.Add(new OutboxMessage {

    Type = "OrderPlaced",

    Payload = JsonSerializer.Serialize(new { order.Id, order.Total, order.CustomerId }),

    OccurredOn = DateTimeOffset.UtcNow

});

await db.SaveChangesAsync();

await tx.CommitAsync();

// BackgroundService publisher

protected override async Task ExecuteAsync(CancellationToken ct)

{

    while (!ct.IsCancellationRequested)

    {

        var messages = await db.Outbox

            .Where(m => m.SentOn == null)

            .OrderBy(m => m.OccurredOn)

            .Take(100).ToListAsync(ct);

 

        foreach (var msg in messages)

        {

            await broker.PublishAsync(msg.Type, msg.Payload, ct);

            msg.SentOn = DateTimeOffset.UtcNow;

        }

        await db.SaveChangesAsync(ct);

        await Task.Delay(TimeSpan.FromSeconds(1), ct);

    }

}

• Updating “relative/related” databases: consume those events in other services to update local read models (CQRS). Use Sagas (orchestration or choreography) for multi-step processes (e.g., Order → Reserve Inventory → Process Payment → Arrange Shipping), with timeouts + compensation.

5) Health checks, readiness, and monitoring

Add liveness (process alive) and readiness (deps OK) endpoints:

builder.Services.AddHealthChecks()

    .AddSqlServer(config.GetConnectionString("OrdersDb"))

    .AddRabbitMQ(config["RabbitMq:Connection"])

    .AddUrlGroup(new Uri(config["CatalogUrl"] + "/health"), name: "catalog");

 

app.MapHealthChecks("/health/live", new HealthCheckOptions { Predicate = _ => false });

app.MapHealthChecks("/health/ready", new HealthCheckOptions {

    Predicate = _ => true,

    ResponseWriter = UIResponseWriter.WriteHealthCheckUIResponse

});

• Kubernetes/Containers probe /health/live and gate traffic on /health/ready.
• Add self-diagnostics (thread pool starvation, GC info) via EventCounters exported by OpenTelemetry.

6) Observability (logs, traces, metrics)

Unified, structured telemetry makes bugs easy to fix.

OpenTelemetry (logs + traces + metrics) → backends like Grafana Tempo/Prometheus/Loki, Jaeger, Elastic, or Azure Application Insights.

builder.Services.AddOpenTelemetry()

 .WithTracing(t => t

     .AddAspNetCoreInstrumentation()

     .AddHttpClientInstrumentation()

     .AddSqlClientInstrumentation()

     .AddSource("OrderService")

     .AddOtlpExporter()) // or Azure Monitor exporter

 .WithMetrics(m => m

     .AddAspNetCoreInstrumentation()

     .AddRuntimeInstrumentation()

     .AddHttpClientInstrumentation()

     .AddPrometheusExporter());

 

app.MapPrometheusScrapingEndpoint(); // /metrics

Structured logging (e.g., Serilog → console/JSON + sink of choice):

Log.Logger = new LoggerConfiguration()

    .Enrich.FromLogContext()

    .Enrich.WithCorrelationId()

    .WriteTo.Console(new JsonFormatter())

    .WriteTo.Seq(config["SeqUrl"]) // or Elastic/Azure

    .CreateLogger();

 

builder.Host.UseSerilog();

• Include CorrelationId/TraceId in every log; propagate with middleware:

app.Use(async (ctx, next) =>

{

    var cid = ctx.Request.Headers["X-Correlation-ID"].FirstOrDefault() ?? Guid.NewGuid().ToString();

    ctx.Items["CorrelationId"] = cid;

    ctx.Response.Headers["X-Correlation-ID"] = cid;

    using (LogContext.PushProperty("CorrelationId", cid))

        await next();

});

• Emit domain events and key business metrics (orders placed, failures, latencies, queue depth).

7) API gateway / proxy servers

• Use a Gateway for routing, TLS, auth/ratelimiting, request shaping:
• YARP (in-process .NET reverse proxy) or Ocelot; at the edge you can also use NGINX/Envoy.
• Centralize AuthN/Z (OAuth2/OIDC via Azure AD, Auth0, or Duende IdentityServer), rate limits, request/response size limits, CORS, and header normalization at the gateway.
• Prefer service discovery (K8s DNS) and mTLS in cluster; consider a service mesh (Istio/Linkerd) for uniform retries/mtls/traffic shifting and better telemetry without code changes.

8) Security & configuration

• Never share DBs across services. Grant least privilege per service.
• Secrets via Azure Key Vault / K8s secrets; bind with IOptions<T> and validate on startup.
• Enforce TLS everywhere, JWT validation, rotating keys, and input validation (FluentValidation).

9) Deployment & runtime

• Containerize each service (Docker), build once, promote the same image through Dev → QA → Prod.
• Orchestrate with Kubernetes (AKS) or Azure App Services for simpler cases.
• Blue/Green or Canary deployments; mesh/gateway can split traffic gradually.
• Horizontal scaling via HPA (CPU/RPS/latency). Keep services stateless; externalize state (DB/cache).
• Caching: local memory for tiny TTLs, Redis for shared cache; always set cache keys + expirations.

10) CI/CD and quality gates

• GitHub Actions / Azure DevOps Pipelines:
• Build → unit/integration/contract tests → SCA & SAST → container scan → publish image → deploy via Helm/Bicep/Terraform.
• Block merges without green tests and security checks.
• Spin ephemeral test environments per PR when feasible.

11) Diagnostics & “easy bug fixing”

• Global exception handler → consistent error envelopes with traceId:

app.UseExceptionHandler(builder =>

{

    builder.Run(async ctx =>

    {

        var traceId = Activity.Current?.Id ?? ctx.TraceIdentifier;

        var problem = new { traceId, message = "Unexpected error." };

        ctx.Response.StatusCode = 500;

        await ctx.Response.WriteAsJsonAsync(problem);

    });

});

• Feature flags (Azure App Configuration) for safe rollouts; dark-launch code paths.
• Replayable messages (don’t auto-ack until processed). Keep dead-letter queues and dashboards.
• Synthetic checks (probes that place a tiny test order in lower envs and alert on end-to-end failure).

12) Database migrations & uptime

• Apply EF Core migrations at startup (carefully) or via migration jobs;
• Prefer expand-and-contract: add new columns/tables (expand), release code that writes both, backfill, then remove old fields (contract).
• For read models, permit rebuild from event streams or upstream APIs if corrupted.

13) Monitoring & alerts you’ll actually use

• Golden signals per service: Latency, Traffic, Errors, Saturation.
• Alert on SLIs/SLOs (e.g., p95 < 300ms, error rate < 1%), queue lag, breaker open rate, DB CPU, connection pool exhaustion.
• Route alerts to Teams/Slack with runbooks linked (how to diagnose/rollback).

14) Local/dev ergonomics

• Docker-compose for local broker + DB + dependencies.
• Seed data + test fixtures; make services start with sensible defaults.
• Makefile/PS scripts for common tasks; one-command bootstrap.

---

Minimal reference architecture (quick checklist)

• ASP.NET Core services (+ gRPC if needed)
• HttpClientFactory + Polly (timeouts/retries/circuit breaker/bulkhead)
• Outbox + BackgroundPublisher; events over Kafka/RabbitMQ/Azure Service Bus
• DB per service (EF Core/Dapper) + migrations (expand/contract)
• OpenTelemetry (traces/metrics/logs) → Prometheus/Grafana/Tempo or App Insights
• Serilog structured logging + correlation IDs
• Health checks: /health/live, /health/ready
• API Gateway: YARP/Ocelot; edge NGINX/Envoy; OAuth2/OIDC
• Containerized; AKS/App Service; blue/green or canary
• CI/CD with tests, scans, infrastructure as code
• Feature flags; DLQs; synthetic probes; runbooks