Peerpay.Internal.Backend 1.1.2

InternalBackendLib SDK

An internal SDK for Peerpay Digital Asset backend services, including Kafka logging and Identity management.

Installation

Add the package to your project:

dotnet add package Peerpay.Internal.Backend

# Install a specific version
dotnet add package Peerpay.Internal.Backend --version 1.0.0

Setup & Dependency Injection

Register the services in your Program.cs or Startup.cs.

1. Identity Service

Use AddIdentityService to register the IIdentityService. You can configure the environment via options.

using Peerpay.Internal.Backend.Extensions;

// For Test Environment
builder.Services.AddIdentityService(options => 
{
    options.IsProd = false; 
});

// For Production Environment
builder.Services.AddIdentityService(options => 
{
    options.IsProd = true;
});

Usage Examples

Identity Service

Inject IIdentityService into your class.

public class MyService
{
    private readonly IIdentityService _identityService;

    public MyService(IIdentityService identityService)
    {
        _identityService = identityService;
    }

    public async Task LoginAsync()
    {
        var request = new AuthorizeRequest 
        {
            Email = "user@example.com",
            Password = "password",
            ApplicationId = "guid"
        };

        var response = await _identityService.AuthorizeAsync(request);

        if (response.Status)
        {
            var token = response.Data.Token;

        }
    }

    public async Task GetInternalTokenAsync()
    {
        var request = new InternalTokenRequest
        {
            ClientId = "client_id",
            ClientSecret = "client_secret",
            AppCode = "app_code"
        };
        
        // Pass the API Key/Token required for the Authorization header
        var response = await _identityService.GetInternalTokenAsync(request, "your-api-key");
    }
}

2. Validation Features

The SDK provides strong validation support against XSS and SQL Injection.

String Extensions

Standalone validation for any string.

using Peerpay.Internal.Backend.Extensions;

string input = "<script>alert('xss')</script>";

// Check if valid
bool isValid = input.XssValidation().IsValid; // false

// Get detailed error
var (valid, message) = input.XssValidation();
if (!valid)
{
    Console.WriteLine(message); // "Input contains potentially malicious XSS content."
}

FluentValidation Integration

Easily integrate into your existing validators.

using FluentValidation;
using Peerpay.Internal.Backend.Validation;

public class CreateUserValidator : AbstractValidator<CreateUserRequest>
{
    public CreateUserValidator()
    {
        RuleFor(x => x.Bio)
            .XssCsrfValidation(); // Checks for XSS, SQLi, and malicious patterns
            
        RuleFor(x => x.Website)
            .MustBeValidUrl();
    }
}

// Register validators from assembly
services.AddValidatorsFromAssembly(typeof(Program).Assembly);

3. Resilient Kafka Producer (AWS MSK)

The SDK includes a resilient Kafka producer with AWS MSK support (SASL/OAuthBearer) and Polly policies (Retry, Circuit Breaker, Timeout).

Configuration

Register the producer in Program.cs.

using Peerpay.Internal.Backend.Extensions;

builder.Services.AddKafkaProducer(options => 
{
    options.BootstrapServers = "b-1.msk-cluster.amazonaws.com:9098";
    
    // Enable MSK specific settings (SASL/SSL + OAuthBearer)
    options.IsMsk = true;
    options.Region = "us-east-1";
    
    // IMPORTANT: AccessKey and SecretKey are MANDATORY for MSK
    options.AccessKey = "YOUR_ACCESS_KEY";
    options.SecretKey = "YOUR_SECRET_KEY";
    
    // Optional: Map internal topic names to actual Kafka topics
    options.Topics = new Dictionary<string, string>
    {
        { "orders", "orders-topic-v1" },
        { "logs", "app-logs-prod" }
    };
    
    // Resilience settings
    options.MessageSendMaxRetries = 3;
    options.Ack = Confluent.Kafka.Acks.All;
});

Producing Messages

Inject IResilientKafkaProducer for robust messaging.

public class OrderService
{
    private readonly IResilientKafkaProducer _producer;

    public OrderService(IResilientKafkaProducer producer)
    {
        _producer = producer;
    }

    public async Task CreateOrderAsync(Order order)
    {
        // policies (retry/circuit-breaker) are applied automatically
        bool success = await _producer.ProduceWithRetryAsync("orders", order);
        
        if (!success)
        {
            // Handle failure after retries exhausted
        }
    }
}

4. Kafka Consumer (Background Service)

The SDK provides an optimized KafkaConsumerBase<T> to simplify consuming messages. It handles:

• MSK Authentication: Automatic OAuth token refresh.
• Deserialization: Automated JSON deserialization using System.Text.Json.
• Resilience: Poison pill skipping (bad JSON/nulls) and error logging.
• Graceful Shutdown: Handles cancellation tokens properly.

Implementation

Inherit from KafkaConsumerBase<TMessage>.

public class OrderProcessedConsumer : KafkaConsumerBase<OrderProcessedEvent>
{
    private readonly IServiceProvider _serviceProvider;

    public OrderProcessedConsumer(
        ILogger<KafkaConsumerBase<OrderProcessedEvent>> logger,
        IOptions<KafkaOptions> kafkaConfig,
        IHostEnvironment env,
        IServiceScopeFactory scopeFactory,
        IServiceProvider serviceProvider)
        : base(logger, kafkaConfig, env, scopeFactory)
    {
        _serviceProvider = serviceProvider;
    }

    // Map to config keys in KafkaOptions.Topics and .ConsumerGroups
    protected override string TopicName => "orders"; 
    protected override string GroupIdName => "order-processors";

    protected override async Task ProcessMessageAsync(OrderProcessedEvent message, CancellationToken ct)
    {
        // Use Scoped Services safely
        using var scope = _scopeFactory.CreateScope();
        var handler = scope.ServiceProvider.GetRequiredService<IOrderHandler>();
        
        await handler.HandleAsync(message, ct);
        
        // Log processed
        _logger.LogInformation("Processed order {OrderId}", message.OrderId);
    }
}

Registration

Register as a Hosted Service.

services.AddHostedService<OrderProcessedConsumer>();

// Ensure Config is valid
services.Configure<KafkaOptions>(options => 
{
    options.Topics["orders"] = "actual-order-topic-v1";
    options.ConsumerGroups["order-processors"] = "group-id-v1";
});