WebHooks
A webhook is a way for an application to provide callbacks to third-party applications.
When an event occurs, the source application sends an HTTP POST request to a pre-configured external URL, with the event data in the request payload.
This allows external applications to react to events through a standard Web API.
This pattern is used by most large platforms. I will use GitHub (e.g. notify when a new pull request or issue is created) as a template and integrate this functionality into an ASP.NET Core (C#) backend.
Demo requirements
- Create, update and delete webhooks.
- Each WebHook can listen for one or more events.
- WebHooks are processed asynchronously, outside the user request, with a basic retry policy when delivery fails.
- You can view the history of triggered WebHooks.
- You can view the system history (audit log) of when a webhook was created, deleted or updated.
⠀
Note: You can find the full source code of the app, including identity, distributed logging, tracing and monitoring, in the open source GitHub repo:
ER diagram
The ER (entity relationship) diagram shows the basic WebHook database tables and the relationships between them. This is what we need to build in our database.
On the left you see the System Event Model, which stores events such as WebHook Created/Updated/Deleted. This is the system audit log for tracking webhook history in the application.
On the right you see the WebHook database model, which is very simple. Each webhook has multiple headers and a history of triggers.
WebHook structure:
- ID - primary key
- WebHookUrl - the URL that is called when the hook is triggered.
- Secret - a shared secret used to sign each delivery. The receiver computes an HMAC-SHA256 of the request body with the same secret and compares it with the
X-Trouble-Signatureheader, so it only accepts requests that really came from us. The secret itself is never sent. - ContentType - the ContentType of the HTTP message.
- IsActive - a flag to enable/disable sending events.
- HookEvents - the events the hook listens for.
- Headers - additional headers sent with the webhook.
- Records - delivery history.
- LastTrigger - when the webhook was last triggered.
WebHookHistory structure:
- ID - primary key
- WebHookID - associated webhook primary key
- Guid - alternate globally unique key
- HookType - type of event that triggered the hook
- Result - result of the delivery
- StatusCode - response status code
- ResponseBody - body received from the endpoint
- RequestBody - body that was sent
- RequestHeaders - headers sent with the request
- Exception - information about any exception
- Timestamp - date and time of the delivery
WebHookHeader structure:
- ID - primary key
- WebHookID - associated webhook primary key
- Name - name of the header
- Value - value of the header
To turn this diagram into real database tables, I use the code-first approach with EF Core (object-relational mapper) and its design-time tools, which generate DB migrations from C# code.
Stack
The most important parts of this demo:
-
ASP.NET Coreas the backend framework. -
EF Coreas the ORM to map C# objects to database entities, andEF Core Toolsto create database migrations. -
MediatRas in-memory messaging to process commands like CreateWebHook/ProcessWebHook etc. -
Hangfirefor processing hooks in a queue and for basic retry policies. -
PostgreSQLas the database (you can also use in-memory for testing).
NOTE: The full source code is more complex. This article focuses only on the WebHook integration.
Domain layer
The domain layer is a collection of entity objects and their business logic, which we call domain models.
WebHook Domain Model
Since we use the code-first approach, we start by defining the WebHook domain model:
/Src/APIServer/Domain/Enity/WebHook_DB/WebHook.cs
namespace APIServer.Domain.Core.Models.WebHooks {
public class WebHook {
public WebHook() {
this.Headers = new HashSet<WebHookHeader>();
this.HookEvents = new HookEventType[0];
this.Records = new List<WebHookRecord>();
}
/// <summary>
/// Hook DB Id
/// </summary>
public long ID { get; set; }
/// <summary>
/// Webhook endpoint
/// </summary>
public string WebHookUrl { get; set; }
/// <summary>
/// Webhook secret (used to sign deliveries)
/// </summary>
#nullable enable
public string? Secret { get; set; }
#nullable disable
/// <summary>
/// Content Type
/// </summary>
public string ContentType { get; set; }
/// <summary>
/// Active / inactive
/// </summary>
public bool IsActive { get; set; }
/// <summary>
/// Hook Events context
/// </summary>
public HookEventType[] HookEvents { get; set; }
/// <summary>
/// Additional HTTP headers. Will be sent with hook.
/// </summary>
virtual public HashSet<WebHookHeader> Headers { get; set; }
/// <summary>
/// Hook call records history
/// </summary>
virtual public ICollection<WebHookRecord> Records { get; set; }
/// <summary>
/// Timestamp of last hook trigger
/// </summary>
public DateTime? LastTrigger { get; set; }
}
}
Each hook has multiple WebHookRecord entries (delivery history):
/Src/APIServer/Domain/Enity/WebHook_DB/WebHookRecord.cs
namespace APIServer.Domain.Core.Models.WebHooks {
public class WebHookRecord {
public WebHookRecord() { }
/// <summary>
/// Hook record DB Id
/// </summary>
public long ID { get; set; }
/// <summary>
/// Linked Webhook Id
/// </summary>
public long WebHookID { get; set; }
/// <summary>
/// Linked Webhook
/// </summary>
public WebHook WebHook { get; set; }
/// <summary>
/// Unique GUID
/// </summary>
public string Guid { get; set; }
/// <summary>
/// WebHookType
/// </summary>
public HookEventType HookType { get; set; }
/// <summary>
/// Hook result enum
/// </summary>
public RecordResult Result { get; set; }
/// <summary>
/// Response
/// </summary>
public int StatusCode { get; set; }
/// <summary>
/// Response json
/// </summary>
public string ResponseBody { get; set; }
/// <summary>
/// Request json
/// </summary>
public string RequestBody { get; set; }
/// <summary>
/// Request Headers json
/// </summary>
public string RequestHeaders { get; set; }
/// <summary>
/// Exception
/// </summary>
public string Exception { get; set; }
/// <summary>
/// Hook Call Timestamp
/// </summary>
public DateTime Timestamp { get; set; }
}
public enum RecordResult {
undefined = 0,
ok,
parameter_error,
http_error,
dataQueryError
}
}
Each hook can have multiple WebHookHeader entries (custom headers):
/Src/APIServer/Domain/Enity/WebHook_DB/WebHookHeader.cs
namespace APIServer.Domain.Core.Models.WebHooks {
public class WebHookHeader {
/// <summary>
/// Hook header ID
/// </summary>
public long ID { get; set; }
/// <summary>
/// Linked Webhook Id
/// </summary>
public long WebHookID { get; set; }
/// <summary>
/// Linked Webhook
/// </summary>
public WebHook WebHook { get; set; }
/// <summary>
/// Header Name
/// </summary>
public string Name { get; set; }
/// <summary>
/// Header content
/// </summary>
public string Value { get; set; }
/// <summary>
/// Header created time
/// </summary>
public DateTime CreatedTimestamp { get; set; }
}
}
Each webhook can be triggered by multiple events:
/Src/APIServer/Domain/Enity/WebHook_DB/WebHookEvents.cs
namespace APIServer.Domain.Core.Models.WebHooks {
public enum HookEventType {
// Take this as an example, you can implement any event source you like.
hook, // (Hook created, hook deleted ...)
file, // (File uploaded, file deleted)
note, // (Note posted, note updated)
project, // (Project created, project disabled)
milestone // (Milestone created, milestone done etc.)
// You can define your own event types...
}
}
Each event type groups related actions:
/Src/APIServer/Domain/Enity/WebHook_System/HookActions.cs
namespace APIServer.Domain.Core.Models.WebHooks {
// Actions of HookEventType
public enum HookResourceAction {
undefined,
hook_created,
hook_removed,
hook_updated,
// etc...
}
// Actions of ProjectEventType
public enum projectAction {
undefined,
project_created,
project_renamed,
project_archived,
// etc...
}
}
Events Domain model
/Src/APIServer/Domain/Enity/DomainEvents/Event.cs
namespace APIServer.Domain.Core.Models.Events {
public class DomainEvent {
public long ID { get; set; }
#nullable enable
public Guid? ActorID { get; set; }
#nullable disable
public DateTime TimeStamp { get; set; }
public EventType EventType { get; set; }
}
}
where EventType is:
/Src/APIServer/Domain/Enity/DomainEvents/EventType.cs
namespace APIServer.Domain.Core.Models.Events {
public enum EventType {
WebHook,
System,
Project,
}
}
Then we can define our system events:
/Src/APIServer/Domain/Enity/DomainEvents/WebHookCreated.cs
/// <summary>
/// WebHookCreated
/// </summary>
public class WebHookCreated : DomainEvent {
public WebHookCreated() { }
public long WebHookId {get;set;}
// Add any custom props here...
}
// The same for WebHookUpdated and WebHookRemoved
Persistence layer
This layer takes care of storing the application state in the database. It defines the DbContext, database configuration files and migrations.
Make sure you have installed the required packages in your APIServer.Persistence.csproj:
<PackageReference Include="Microsoft.EntityFrameworkCore" Version="5.0.9" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="5.0.9" />
Based on our domain model, the DbContext must contain these collections:
-
WebHooks- stored WebHooks -
WebHooksHistory- WebHook delivery history -
Events- WebHook system logs
Database DbContext:
/Src/APIServer/Persistence/ApiDbContext.cs
namespace APIServer.Persistence {
public class ApiDbContext : DbContext {
public DbSet<WebHook> WebHooks { get; set; }
public DbSet<WebHookRecord> WebHooksHistory { get; set; }
public DbSet<DomainEvent> Events { get; set; }
public ApiDbContext(
DbContextOptions<ApiDbContext> options)
: base(options) { }
protected override void OnModelCreating(ModelBuilder modelBuilder) {
modelBuilder.ApplyConfigurationsFromAssembly(typeof(ApiDbContext).Assembly);
modelBuilder.Entity<WebHookCreated>().ToTable("WebHookCreatedEvent");
modelBuilder.Entity<WebHookRemoved>().ToTable("WebHookRemovedEvent");
modelBuilder.Entity<WebHookUpdated>().ToTable("WebHookUpdatedEvent");
modelBuilder.Entity<WebHook>().HasData(
new WebHook() {
ID = 1,
WebHookUrl = "https://localhost:5015/hookloopback",
IsActive = true,
ContentType = "application/json",
HookEvents=new HookEventType[]{HookEventType.hook}
});
base.OnModelCreating(modelBuilder);
}
}
}
By overriding the OnModelCreating method, we configure the database model:
-
ApplyConfigurationsFromAssemblyloads all entity configurations. They are defined in the/Src/APIServer/Persistence/Configurationfolder as separate classes. - Each event type is mapped to its own table (TPT, table per type).
- Initial data is seeded.
Entity configuration (loaded by ApplyConfigurationsFromAssembly):
/Src/APIServer/Persistence/Configuration/WebHook.cs
namespace APIServer.Persistence {
public class WebHookConfiguration : IEntityTypeConfiguration<WebHook> {
public void Configure(EntityTypeBuilder<WebHook> builder) {
builder.HasKey(e => e.ID);
builder.HasMany(e => e.Headers)
.WithOne(e => e.WebHook)
.HasForeignKey(e => e.WebHookID);
builder.HasMany(e => e.Records)
.WithOne(e => e.WebHook)
.HasForeignKey(e => e.WebHookID);
}
}
}
To generate DB migrations from the DbContext and the configuration files, we use EF Core Tools.
Install them with the CLI:
dotnet tool install --global dotnet-ef
You can verify the installation by running dotnet ef in the terminal. You should see this:
_/\__
---==/ \\
___ ___ |. \|\
| __|| __| | ) \\\
| _| | _| \_/ | //|\\
|___||_| / \\\/\\
Entity Framework Core .NET Command-line Tools 5.0.9
To generate migration files, run this command from Src/APIServer/Persistence:
dotnet ef migrations add Init
This creates the Migrations folder with the migration files. To apply the migrations to your database, run:
dotnet ef database update
⠀
NOTE: Make sure the connection string inMigrationConfig.csis valid. The migration tool needs it to connect to your DB and create the tables.
⠀
To keep the startup of the main project clean, we also define a service collection extension. It registers the DbContext with a single call, services.AddApiDbContext(...);, and keeps the configuration out of Startup.cs.
/Src/APIServer/Persistence/Extensions/AddApiDbContext.cs
namespace APIServer.Persistence.Extensions {
public static partial class ServiceExtension {
public static IServiceCollection AddApiDbContext(
this IServiceCollection serviceCollection,
IConfiguration Configuration, IWebHostEnvironment Environment) {
serviceCollection.AddPooledDbContextFactory<ApiDbContext>(
(s, o) => o
.UseNpgsql(Configuration["ConnectionStrings:ApiDbContext"], option => {
option.EnableRetryOnFailure();
if (Environment.IsDevelopment()) {
o.EnableDetailedErrors();
o.EnableSensitiveDataLogging();
}
}).UseLoggerFactory(s.GetRequiredService<ILoggerFactory>()));
return serviceCollection;
}
}
}
ConnectionStrings:ApiDbContext is defined in Src/APIServer/API/appsettings.json:
{
"ConnectionStrings": {
"ApiDbContext": "Host=localhost;Port=5432;Database=APIServer;Username=postgres;Password=postgres"
}
}
GraphQL resolvers load data in parallel, so multiple queries hit the database at the same time. A regular scoped DbContext is not thread-safe and can't handle that, which is why AddPooledDbContextFactory is used and every operation creates its own context.
Application layer
This layer contains MediatR (CQRS) commands and queries for specific application tasks.
It also contains a GraphQL layer on top of the MediatR logic. GraphQL and MediatR overlap in some concepts, so the DDD rules are not followed 100% and the logic is split between these two parts. Ideally they would be completely separated, but because of the advanced error handling and cursor pagination, some DDD rules are bent here.
This is the distributed trace of the CreateWebHook command. You can see each part we will implement.
Theoretical flow:
Real trace measurement:
WebHook commands and queries
Let's look at the CreateWebHook command as an example:
Src/APIServer/Aplication/CQRS/Commands/Public/Hooks/Create_WebHook.cs
namespace APIServer.Aplication.Commands.WebHooks {
/// <summary>
/// Command for creating webhook
/// </summary>
[Authorize] // <-- Activate Auth check for command
// [Authorize(FieldPolicy = true)] <-- Uncomment to activate Field Auth check for command
public class CreateWebHook : IRequest<CreateWebHookPayload> {
public CreateWebHook() {
this.HookEvents = new HashSet<HookEventType>();
}
/// <summary> Url </summary>
public string WebHookUrl { get; set; }
/// <summary> Secret </summary>
public string? Secret { get; set; }
/// <summary> IsActive </summary>
public bool IsActive { get; set; }
/// <summary> HookEvents </summary>
public HashSet<HookEventType> HookEvents { get; set; }
}
/// <summary>
/// CreateWebHook Validator
/// </summary>
public class CreateWebHookValidator : AbstractValidator<CreateWebHook> {
private readonly IDbContextFactory<ApiDbContext> _factory;
public CreateWebHookValidator(IDbContextFactory<ApiDbContext> factory){
_factory = factory;
RuleFor(e => e.WebHookUrl)
.NotEmpty()
.NotNull();
RuleFor(e => e.WebHookUrl)
.Matches(Common.URI_REGEX)
.WithMessage("Does not match URI expression");
RuleFor(e => e.WebHookUrl)
.MustAsync(BeUniqueByURL)
.WithMessage("Hook endpoint already exists");
RuleFor(e => e.WebHookUrl)
.MustAsync(CheckMaxAllowedHooksCount)
.WithMessage("Maximum number of hooks reached");
RuleFor(e => e.HookEvents)
.NotNull();
}
public async Task<bool> BeUniqueByURL(string url, CancellationToken cancellationToken) {
await using ApiDbContext dbContext =
_factory.CreateDbContext();
return await dbContext.WebHooks.AllAsync(e => e.WebHookUrl != url, cancellationToken);
}
public async Task<bool> CheckMaxAllowedHooksCount(string url, CancellationToken cancellationToken) {
await using ApiDbContext dbContext =
_factory.CreateDbContext();
const long MAX_HOOK_COUNT = 3;
return (await dbContext.WebHooks.CountAsync(cancellationToken)) < MAX_HOOK_COUNT;
}
}
/// <summary>
/// Authorization validators for CreateWebHook
/// </summary>
public class CreateWebHookAuthorizationValidator : AuthorizationValidator<CreateWebHook> {
private readonly IDbContextFactory<ApiDbContext> _factory;
public CreateWebHookAuthorizationValidator(IDbContextFactory<ApiDbContext> factory) {
_factory = factory;
// Add field authorization checks... (use [Authorize(FieldPolicy = true)] to activate)
}
}
/// <summary>
/// ICreateWebHookError
/// </summary>
public interface ICreateWebHookError { }
/// <summary>
/// CreateWebHookPayload
/// </summary>
public class CreateWebHookPayload : BasePayload<CreateWebHookPayload, ICreateWebHookError> {
/// <summary>
/// Created WebHook
/// </summary>
public WebHook hook { get; set; }
}
/// <summary>Handler for <c>CreateWebHook</c> command </summary>
public class CreateWebHookHandler : IRequestHandler<CreateWebHook, CreateWebHookPayload> {
/// <summary>
/// Injected <c>IDbContextFactory<ApiDbContext></c>
/// </summary>
private readonly IDbContextFactory<ApiDbContext> _factory;
/// <summary>
/// Injected <c>IPublisher</c>
/// </summary>
private readonly APIServer.Extensions.IPublisher _publisher;
/// <summary>
/// Injected <c>ICurrentUser</c>
/// </summary>
private readonly ICurrentUser _current;
/// <summary>
/// Main constructor
/// </summary>
public CreateWebHookHandler(
IDbContextFactory<ApiDbContext> factory,
APIServer.Extensions.IPublisher publisher,
ICurrentUser currentuser) {
_factory = factory;
_publisher = publisher;
_current = currentuser;
}
/// <summary>
/// Command handler for <c>CreateWebHook</c>
/// </summary>
public async Task<CreateWebHookPayload> Handle(CreateWebHook request, CancellationToken cancellationToken) {
await using ApiDbContext dbContext =
_factory.CreateDbContext();
WebHook hook = new WebHook {
WebHookUrl = request.WebHookUrl,
Secret = request.Secret,
ContentType = "application/json",
IsActive = request.IsActive,
HookEvents = request.HookEvents != null ? request.HookEvents.Distinct().ToArray() : new HookEventType[0]
};
dbContext.WebHooks.Add(hook);
await dbContext.SaveChangesAsync(cancellationToken);
try {
await _publisher.Publish(new WebHookCreatedNotifi() {
ActivityId = Activity.Current?.Id
}, PublishStrategy.ParallelNoWait, default(CancellationToken));
} catch {
// The hook is already saved, a failed notification must not fail the request
}
var response = CreateWebHookPayload.Success();
response.hook = hook;
return response;
}
}
}
As you can see, the commands (which sit underneath GraphQL) already prepare the request/response objects so they fit well into the GraphQL layer.
This means:
- Validation and authorization checks are automatically turned into validation and authorization errors by MediatR behaviours.
- The handler returns a
payloadbased object that is later consumed by GraphQL. - Each command also defines its own marker interface, such as
ICreateWebHookError.
This is explained in more detail in a previous part, GraphQL Mutation Errors aka 6a. The important thing to remember is that all handlers return a payload object.
On the GraphQL side, we define WebHookMutations:
Src/APIServer/Aplication/Graphql/Mutations/WebHooks.cs
namespace APIServer.Aplication.GraphQL.Mutation {
/// <summary>
/// WebHooks Mutations
/// </summary>
[ExtendObjectType(OperationTypeNames.Mutation)]
public class WebHookMutations {
/// <summary>
/// Create new webhook
/// </summary>
public class CreateWebHookInput {
/// <summary> Url </summary>
public string WebHookUrl { get; set; }
/// <summary> Secret </summary>
public string? Secret { get; set; }
/// <summary> IsActive </summary>
public bool IsActive { get; set; }
/// <summary> HookEvents </summary>
public HookEventType[] HookEvents { get; set; }
}
/// <summary>
/// Create new webhook
/// </summary>
/// <returns></returns>
public async Task<CreateWebHookPayload> CreateWebHook(
CreateWebHookInput request,
[Service] IMediator _mediator) {
return await _mediator.Send(new CreateWebHook() {
WebHookUrl = request.WebHookUrl,
Secret = request.Secret,
IsActive = request.IsActive,
HookEvents = request.HookEvents != null ? new HashSet<HookEventType>(request.HookEvents) : new HashSet<HookEventType>(new HookEventType[0]),
});
}
}
}
As you can see, mutation == command. The GraphQL layer simply passes the input object to MediatR and gets the payload back. It's similar to calling MediatR from an API controller, except GraphQL can do some extra magic before and after processing.
Querying is a little different. If we want to use the built-in pagination and filtering, we need to pass the DbContext to our handler or keep the logic in the GraphQL layer. To keep the demo simple, we query the data directly from the GraphQL layer:
[UseApiDbContextAttribute]
[UsePaging(typeof(WebHookType))]
[UseFiltering]
public IQueryable<GQL_WebHook> Webhooks(
[Service] ICurrentUser current,
[ScopedService] ApiDbContext context) {
if (!current.Exist) {
return null;
}
return context.WebHooks
.AsNoTracking()
.Select(e=> new GQL_WebHook {
ID = e.ID,
WebHookUrl = e.WebHookUrl,
ContentType = e.ContentType,
IsActive = e.IsActive,
LastTrigger = e.LastTrigger,
ListeningEvents = e.HookEvents
});
}
Let's take a quick look at how the mutation type is defined in the GraphQL layer:
namespace APIServer.Aplication.GraphQL.Types {
public class CreateWebHookPayloadType : ObjectType<CreateWebHookPayload> {
protected override void Configure(IObjectTypeDescriptor<CreateWebHookPayload> descriptor) {
descriptor.Field(e => e.hook).Type<WebHookType>().Resolve(context => {
WebHook e = context.Parent<CreateWebHookPayload>().hook;
if (e == null) {
return null;
}
return new GQL_WebHook {
ID = e.ID,
WebHookUrl = e.WebHookUrl,
ContentType = e.ContentType,
IsActive = e.IsActive,
LastTrigger = e.LastTrigger,
ListeningEvents = e.HookEvents
};
});
}
}
public class CreateWebHookErrorUnion : UnionType<ICreateWebHookError> {
protected override void Configure(IUnionTypeDescriptor descriptor) {
descriptor.Type<ValidationErrorType>();
descriptor.Type<UnAuthorisedType>();
descriptor.Type<InternalServerErrorType>();
}
}
}
From the definition, we can see that the MediatR command returns CreateWebHookPayload with the hook field. GraphQL can extend or override any value, so this is the right place to change the response or define how additional data is resolved.
Another important part is CreateWebHookErrorUnion, which defines the 3 error types the command can return as a union. They are available under the payload Error[]. This is also explained in more detail in GraphQL mutation errors aka 6a.
It's a similar story for GraphQL queries. Let's take a look at WebHookType:
namespace APIServer.Aplication.GraphQL.Types {
/// <summary> Graphql WebHookType </summary>
public class WebHookType : ObjectType<GQL_WebHook> {
protected override void Configure(IObjectTypeDescriptor<GQL_WebHook> descriptor) {
descriptor.AsNode().IdField(t => t.ID).NodeResolver((ctx, id) =>
ctx.DataLoader<WebHookByIdDataLoader>().LoadAsync(id, ctx.RequestAborted));
descriptor.Field(t => t.ID).Type<NonNullType<IdType>>();
descriptor.Field("systemid").Type<NonNullType<LongType>>().Resolve((IResolverContext context) => {
return context.Parent<GQL_WebHook>().ID;
});
descriptor.Field(t => t.Records)
.UseDbContext<ApiDbContext>()
.Resolve(async ctx => {
ApiDbContext _context = ctx.Service<ApiDbContext>();
ICurrentUser _current = ctx.Service<ICurrentUser>();
long hook_id = ctx.Parent<GQL_WebHook>().ID;
if (!_current.Exist) {
return null;
}
if (hook_id <= 0) {
return new List<GQL_WebHookRecord>().AsQueryable();
}
return _context.WebHooksHistory
.AsNoTracking()
.Where(e => e.WebHookID == hook_id)
.Select(e => new GQL_WebHookRecord() {
ID = e.ID,
WebHookID = e.WebHookID,
WebHookSystemID = e.WebHookID,
StatusCode = e.StatusCode,
ResponseBody = e.ResponseBody,
RequestBody = e.RequestBody,
TriggerType = e.HookType,
Result = e.Result,
Guid = e.Guid,
RequestHeaders = e.RequestHeaders,
Exception = e.Exception,
Timestamp = e.Timestamp,
}!).OrderByDescending(e => e.Timestamp);
})
.UsePaging<WebHookRecordType>()
.UseFiltering();
}
}
}
This ObjectType uses a DTO object and defines how each field is resolved. Some fields are resolved from the Parent (our top query) and others have custom resolution logic, like the Records field.
Hot Chocolate follows the Relay specification, so the Relay global ID and the Node interface are implemented. The definition is in WebHookType.
This allows us to query any object that implements it with the node query.
descriptor.AsNode().IdField(t => t.ID).NodeResolver((ctx, id) =>
ctx.DataLoader<WebHookByIdDataLoader>().LoadAsync(id, ctx.RequestAborted));
Where DataLoader<WebHookByIdDataLoader> is:
Src/APIServer/Aplication/CQRS/Dataloaders/WebHooks/WebHookById_DataLoader.cs
namespace APIServer.Aplication.GraphQL.DataLoaders {
public class WebHookByIdDataLoader : BatchDataLoader<long, GQL_WebHook> {
/// <summary> Injected <c>ApiDbContext</c> </summary>
private readonly IDbContextFactory<ApiDbContext> _factory;
/// <summary> Injected <c>ICurrentUser</c> </summary>
private readonly ICurrentUser _current;
private readonly SemaphoreSlim _semaphoregate = new SemaphoreSlim(1);
public WebHookByIdDataLoader(
IBatchScheduler scheduler,
IDbContextFactory<ApiDbContext> factory,
ICurrentUser current) : base(scheduler) {
_current = current;
_factory = factory;
}
protected override async Task<IReadOnlyDictionary<long, GQL_WebHook>> LoadBatchAsync(
IReadOnlyList<long> keys,
CancellationToken cancellationToken) {
if (!_current.Exist) {
return new List<GQL_WebHook>().ToDictionary(e => e.ID, null);
}
await using ApiDbContext dbContext =
_factory.CreateDbContext();
await _semaphoregate.WaitAsync();
try {
return await dbContext.WebHooks
.AsNoTracking()
.Where(s => keys.Contains(s.ID))
.Select(e => new GQL_WebHook {
ID = e.ID,
WebHookUrl = e.WebHookUrl,
ContentType = e.ContentType,
IsActive = e.IsActive,
LastTrigger = e.LastTrigger,
ListeningEvents = e.HookEvents
})
.ToDictionaryAsync(t => t.ID, cancellationToken);
} finally {
_semaphoregate.Release();
}
}
}
}
DataLoaders are an optimization technique. They are not GraphQL specific and can be part of the application/business logic outside of GraphQL. We use them to solve the N+1 problem: requests for the same kind of data with different IDs are batched into one query.
The pros and cons of DataLoaders, with benchmarks against the traditional approach, are explained in a separate part, where you can go in depth and see how to design your schema. This part is only a basic introduction to the backend logic, so don't worry if the topic isn't fully clear yet.
WebHook notification and processing
We have seen how a mutation is passed to a MediatR command and how the result comes back. Now let's set up how webhooks are processed. This is internal functionality, so these commands must not be publicly available.
You can find all internal commands in Src/APIServer/Aplication/CQRS/Commands/Internal.
To trigger internal notifications, the MediatR publisher is used:
await _publisher.Publish(new WebHookCreatedNotifi() {
ActivityId = Activity.Current?.Id
}, PublishStrategy.ParallelNoWait, default(CancellationToken));
These notifications are handled within the user's request, so keep them fast: collect only the context data and hand the heavy work to the scheduler (background worker) via Enqueue.
This is the MediatR extension that hands a command over to Hangfire:
public string Enqueue(IRequest request, string parentJobId, JobContinuationOptions continuationOption, string description = null) {
var mediatorSerializedObject = SerializeObject(request, description);
return BackgroundJob.ContinueJobWith(parentJobId, () => _commandsExecutor.ExecuteCommand(mediatorSerializedObject), continuationOption);
}
This internal handler enqueues one delivery job for every active hook that listens to the event:
namespace APIServer.Aplication.Commands.Internall.Hooks {
/// <summary>
/// Command for enqueueing webhooks related to an event
/// </summary>
public class EnqueueRelatedWebHooks : CommandBase {
public HookEventType EventType { get; set; }
public object Event { get; set; }
}
/// <summary>
/// Command handler for <c>EnqueueRelatedWebHooks</c>
/// </summary>
public class EnqueueRelatedWebHooksHandler : IRequestHandler<EnqueueRelatedWebHooks, Unit> {
/// <summary> Injected <c>ApiDbContext</c> </summary>
private readonly IDbContextFactory<ApiDbContext> _factory;
/// <summary> Injected <c>IMediator</c> </summary>
private readonly IMediator _mediator;
/// <summary> Main Constructor </summary>
public EnqueueRelatedWebHooksHandler(
IDbContextFactory<ApiDbContext> factory,
IMediator mediator) {
_factory = factory;
_mediator = mediator;
}
/// <summary> Command handler for <c>EnqueueRelatedWebHooks</c> </summary>
public async Task<Unit> Handle(EnqueueRelatedWebHooks request, CancellationToken cancellationToken) {
if (request == null || request.Event == null) {
throw new ArgumentNullException(nameof(request));
}
await using ApiDbContext dbContext =
_factory.CreateDbContext();
List<long> hook_ids = await dbContext.WebHooks
.AsNoTracking()
.Where(e => e.IsActive && e.HookEvents.Contains(request.EventType))
.Select(e => e.ID)
.ToListAsync(cancellationToken);
foreach (var hook_id in hook_ids) {
// One job per hook, so a failing endpoint is retried on its own
_mediator.Enqueue(new ProcessWebHook() {
HookId = hook_id,
Event = request.Event,
EventType = request.EventType
});
}
return Unit.Value;
}
}
}
And this is the delivery itself. Every attempt is saved to the history, and a failed one throws, so Hangfire retries it (5 attempts with an increasing delay in this demo):
namespace APIServer.Aplication.Commands.Internall.Hooks {
/// <summary> Command for processing WebHook event </summary>
public class ProcessWebHook : CommandBase {
public long HookId { get; set; }
public dynamic Event { get; set; }
public HookEventType EventType { get; set; }
}
/// <summary> Command handler for <c>ProcessWebHook</c> </summary>
public class ProcessWebHookHandler : IRequestHandler<ProcessWebHook, Unit> {
/// <summary> Injected <c>ApiDbContext</c> </summary>
private readonly IDbContextFactory<ApiDbContext> _factory;
/// <summary> Injected <c>IHttpClientFactory</c> </summary>
private readonly IHttpClientFactory _clientFactory;
/// <summary> Main Constructor </summary>
public ProcessWebHookHandler(
IDbContextFactory<ApiDbContext> factory,
IHttpClientFactory httpClient) {
_factory = factory;
_clientFactory = httpClient;
}
/// <summary> Command handler for <c>ProcessWebHook</c> </summary>
public async Task<Unit> Handle(ProcessWebHook request, CancellationToken cancellationToken) {
if (request == null || request.HookId <= 0) {
return Unit.Value; // Nothing to deliver
}
await using ApiDbContext dbContext =
_factory.CreateDbContext();
WebHook hook = await dbContext.WebHooks
.Include(e => e.Headers)
.FirstOrDefaultAsync(e => e.ID == request.HookId, cancellationToken);
if (hook == null || !hook.IsActive || string.IsNullOrWhiteSpace(hook.WebHookUrl)) {
return Unit.Value; // Hook was removed or disabled in the meantime
}
WebHookRecord record = new WebHookRecord() {
WebHookID = hook.ID,
Guid = Guid.NewGuid().ToString(),
HookType = request.EventType,
Timestamp = DateTime.UtcNow
};
var options = new JsonSerializerOptions {
IncludeFields = true,
};
string body = JsonSerializer.Serialize<dynamic>(request.Event, options);
using var message = new HttpRequestMessage(HttpMethod.Post, hook.WebHookUrl) {
Content = new StringContent(body, Encoding.UTF8, "application/json")
};
/* Set Headers */
message.Headers.Add("X-Trouble-Delivery", record.Guid);
message.Headers.Add("X-Trouble-Event", request.EventType.ToString().ToLowerInvariant());
if (!string.IsNullOrWhiteSpace(hook.Secret)) {
// Sign the body, never send the secret itself
message.Headers.Add("X-Trouble-Signature", "sha256=" + Sign(body, hook.Secret));
}
foreach (var header in hook.Headers) {
message.Headers.TryAddWithoutValidation(header.Name, header.Value);
}
record.RequestBody = body;
record.RequestHeaders = JsonSerializer.Serialize(
message.Headers.ToDictionary(h => h.Key, h => h.Value));
var httpClient = _clientFactory.CreateClient();
httpClient.Timeout = TimeSpan.FromSeconds(10);
try {
using var response = await httpClient.SendAsync(message, cancellationToken);
record.StatusCode = (int)response.StatusCode;
record.ResponseBody = await response.Content.ReadAsStringAsync(cancellationToken);
record.Result = response.IsSuccessStatusCode ? RecordResult.ok : RecordResult.http_error;
} catch (Exception ex) {
record.Result = RecordResult.http_error;
record.Exception = ex.ToString();
}
hook.LastTrigger = record.Timestamp;
dbContext.WebHooksHistory.Add(record);
await dbContext.SaveChangesAsync(cancellationToken);
if (record.Result != RecordResult.ok) {
// Throw so Hangfire retries the delivery
throw new HttpRequestException(
$"WebHook {hook.ID} delivery failed (status {record.StatusCode})");
}
return Unit.Value;
}
private static string Sign(string body, string secret) {
using var hmac = new HMACSHA256(Encoding.UTF8.GetBytes(secret));
byte[] hash = hmac.ComputeHash(Encoding.UTF8.GetBytes(body));
return Convert.ToHexString(hash).ToLowerInvariant();
}
}
}
On the receiving side, the signature is checked against the raw request body. Compare it in constant time:
using var hmac = new HMACSHA256(Encoding.UTF8.GetBytes(secret));
string expected = "sha256=" + Convert.ToHexString(
hmac.ComputeHash(Encoding.UTF8.GetBytes(rawBody))).ToLowerInvariant();
bool valid = CryptographicOperations.FixedTimeEquals(
Encoding.UTF8.GetBytes(expected),
Encoding.UTF8.GetBytes(signatureHeader ?? ""));
API Layer
This is where the API microservice is configured. It depends on the application and persistence layers.
The Startup.cs class configures the host: dependency injection and the middleware pipeline. This is the demo configuration:
Src/APIServer/API/Startup.cs
namespace APIServer
{
public class Startup
{
public Startup(IConfiguration configuration, IWebHostEnvironment environment)
{
Configuration = configuration;
Environment = environment;
}
public IWebHostEnvironment Environment { get; }
public IConfiguration Configuration { get; }
// This method gets called by the runtime. Use this method to add services to the container.
public void ConfigureServices(IServiceCollection services)
{
services.AddControllers();
services.AddSwaggerGen(c =>{
c.SwaggerDoc("v1", new OpenApiInfo { Title = "Api", Version = "v1" });
});
services.AddAuth(Configuration);
services.AddDbContext(Configuration,Environment);
services.AddHttpClient();
services.AddHealthChecks();
services.AddGraphql(Environment);
services.AddHttpContextAccessor();
services.AddMemoryCache();
services.AddTelemetry(Configuration,Environment);
services.AddScoped<ICurrentUser, CurrentUser>();
services.AddMediatR();
services.AddScheduler(Configuration);
services.AddSingleton(Serilog.Log.Logger);
}
// This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
public void Configure(
IApplicationBuilder app,
IWebHostEnvironment env,
IServiceProvider serviceProvider,
IServiceScopeFactory scopeFactory)
{
app.UseForwardedHeaders(new ForwardedHeadersOptions{
ForwardedHeaders = ForwardedHeaders.XForwardedFor | ForwardedHeaders.XForwardedProto | ForwardedHeaders.XForwardedHost,
});
app.UseEnsureApiContextCreated(serviceProvider,scopeFactory);
if (env.IsDevelopment()) {
app.UseDeveloperExceptionPage();
app.UseSwagger();
app.UseSwaggerUI(c => c.SwaggerEndpoint("/swagger/v1/swagger.json", "Api v1"));
}
app.UseHealthChecks("/health");
app.UseHttpsRedirection();
app.UseRouting();
app.UseAuthentication();
app.UseAuthorization();
app.UseHangfireServer();
if(env.IsDevelopment()){
app.UseHangfireDashboard("/scheduler");
}
app.UseEndpoints(endpoints =>
{
endpoints.MapControllers()
.RequireAuthorization("ApiCaller");
endpoints.MapGraphQL()
.WithOptions(new GraphQLServerOptions {
EnableSchemaRequests = env.IsDevelopment(),
Tool = { Enable = env.IsDevelopment() },
});
endpoints.MapControllerRoute(
name: "default",
pattern: "{controller}/{action=Index}/{id?}");
});
}
}
}
Logging, tracing, monitoring and GraphQL errors are explained in separate parts, so I won't go into detail here and will only show the most important parts. For the full configuration, see the repo linked at the end of this article.
The GraphQL configuration:
namespace APIServer.Configuration {
public static partial class ServiceExtension {
public static IServiceCollection AddGraphql(
this IServiceCollection serviceCollection, IWebHostEnvironment env) {
serviceCollection.AddGraphQLServer()
.SetPagingOptions(
new PagingOptions { IncludeTotalCount = true, MaxPageSize = 200 })
.ModifyRequestOptions(requestExecutorOptions => {
if (env.IsDevelopment() ||
System.Diagnostics.Debugger.IsAttached) {
requestExecutorOptions.ExecutionTimeout = TimeSpan.FromMinutes(1);
}
requestExecutorOptions.IncludeExceptionDetails = !env.IsProduction();
})
.AddGlobalObjectIdentification()
.AddQueryFieldToMutationPayloads()
.AddFiltering()
.AddSorting()
.AddQueryType<Query>()
.AddTypeExtension<WebHookQueries>()
.AddTypeExtension<UserQueries>()
.AddTypeExtension<SystemQueries>()
.AddMutationType<Mutation>()
.AddTypeExtension<WebHookMutations>()
.BindRuntimeType<DateTime, DateTimeType>()
.BindRuntimeType<int, IntType>()
.AddType<BadRequestType>()
.AddType<InternalServerErrorType>()
.AddType<UnAuthorisedType>()
.AddType<ValidationErrorType>()
.AddType<BaseErrorType>()
.AddType<UserDeactivatedType>()
.AddType<BaseErrorInterfaceType>()
.AddType<WebHookNotFoundType>()
.AddType<WebHookRecordType>()
.AddType<WebHookType>()
.AddType<UserType>()
.AddType<UpdateWebHookUriPayloadType>()
.AddType<UpdateWebHookUriErrorUnion>()
.AddType<UpdateWebHookTriggerEventsPayloadType>()
.AddType<UpdateWebHookTriggerEventsErrorUnion>()
.AddType<UpdateWebHookSecretPayloadType>()
.AddType<UpdateWebHookSecretErrorUnion>()
.AddType<UpdateWebHookPayloadPayloadType>()
.AddType<UpdateWebHookPayloadErrorUnion>()
.AddType<UpdateWebHookActivStatePayloadType>()
.AddType<UpdateWebHookActivStateErrorUnion>()
.AddType<RemoveWebHookPayloadType>()
.AddType<RemoveWebHookErrorUnion>()
.AddType<CreateWebHookPayloadType>()
.AddType<CreateWebHookErrorUnion>()
.AddDataLoader<UserByIdDataLoader>()
.AddDataLoader<WebHookByIdDataLoader>()
.AddDataLoader<WebHookRecordByIdDataLoader>()
.UsePersistedQueryPipeline()
.UseReadPersistedQuery()
.AddReadOnlyFileSystemQueryStorage("./persisted_queries");
return serviceCollection;
}
}
}
MediatR configuration:
namespace APIServer.Configuration {
public static partial class ServiceExtension {
public static IServiceCollection AddMediatR(this IServiceCollection services) {
// Command executor
services.AddMediatR(cfg => cfg.Using<AppMediator>(), typeof(CreateWebHook).GetTypeInfo().Assembly);
services.AddTransient<APIServer.Extensions.IPublisher, APIServer.Extensions.Publisher>();
services.AddValidatorsFromAssembly(typeof(CreateWebHookValidator).GetTypeInfo().Assembly);
services.AddTransient<IRequestHandler<EnqueSaveEvent<WebHookCreated>, Unit>, EnqueSaveEventHandler<WebHookCreated>>();
services.AddTransient<IRequestHandler<EnqueSaveEvent<WebHookUpdated>, Unit>, EnqueSaveEventHandler<WebHookUpdated>>();
services.AddTransient<IRequestHandler<EnqueSaveEvent<WebHookRemoved>, Unit>, EnqueSaveEventHandler<WebHookRemoved>>();
services.AddMediatRSchedulerIntegration();
services.AddTransient(typeof(IPipelineBehavior<,>), typeof(TracingBehaviour<,>));
services.AddTransient(typeof(IPipelineBehavior<,>), typeof(CommandPerformanceBehaviour<,>));
services.AddTransient(typeof(IPipelineBehavior<,>), typeof(ValidationBehaviour<,>));
services.AddTransient(typeof(IPipelineBehavior<,>), typeof(AuthorizationBehaviour<,>));
services.AddTransient(typeof(IPipelineBehavior<,>), typeof(UnhandledExBehaviour<,>));
return services;
}
}
}
Where AppMediator is:
namespace APIServer.Extensions {
public class AppMediator : Mediator {
private Func<IEnumerable<Func<INotification, CancellationToken, Task>>, INotification, CancellationToken, Task> _publishStrategy;
public AppMediator(
ServiceFactory serviceFactory
) : base(serviceFactory) {
}
public AppMediator(
ServiceFactory serviceFactory,
Func<IEnumerable<Func<INotification, CancellationToken, Task>>, INotification, CancellationToken, Task>? publishStrategy
) : base(serviceFactory) {
_publishStrategy = publishStrategy != null ? publishStrategy : SyncStopOnException;
}
public Task<TResponse> Send<TResponse>(ICommandBase<TResponse> request, CancellationToken cancellationToken = default) {
return base.Send<TResponse>(request, cancellationToken);
}
public Task<object?> Send(ICommandBase request, CancellationToken cancellationToken = default) {
return base.Send(request as object, cancellationToken);
}
private static async Task SyncStopOnException(IEnumerable<Func<INotification, CancellationToken, Task>> handlers, INotification notification, CancellationToken cancellationToken) {
foreach (var handler in handlers) {
await handler(notification, cancellationToken).ConfigureAwait(false);
}
}
protected override async Task PublishCore(IEnumerable<Func<INotification, CancellationToken, Task>> allHandlers, INotification notification, CancellationToken cancellationToken = default) {
Activity activity = null;
if (notification is INotificationBase) {
INotificationBase I_base_notify = notification as INotificationBase;
// If there is an activity in the context, use it as the default when ActivityId == null
if (I_base_notify.ActivityId == null
&& Activity.Current != null
&& Activity.Current.Id != null) {
I_base_notify.ActivityId = Activity.Current.Id;
}
activity = Sources.DemoSource.StartActivity(
String.Format("PublishCore: Notification<{0}>", notification.GetType().FullName), ActivityKind.Producer);
// Link the activity to its parent (parent / child relation)
if (I_base_notify.ActivityId != null
&& Activity.Current != null
&& Activity.Current.ParentId == null) {
Activity.Current.SetParentId(I_base_notify.ActivityId);
}
if (Activity.Current != null
&& Activity.Current.ParentId != null) {
Activity.Current.AddTag("Parent Id", Activity.Current.ParentId);
}
} else {
activity = Sources.DemoSource.StartActivity(
String.Format("PublishCore: Notification<{0}>", notification.GetType().FullName), ActivityKind.Producer);
}
try {
Activity.Current?.AddTag("Activity Id", Activity.Current.Id);
if (_publishStrategy != null) {
await _publishStrategy(allHandlers, notification, cancellationToken);
} else {
await base.PublishCore(allHandlers, notification, cancellationToken);
}
} finally {
activity?.Stop();
activity?.Dispose();
}
}
}
}
UI
The React UI is covered separately. This is how the demo looks:
Notes for production
This demo covers the core of the integration. For a production setup, I would also add:
- Outbox: save the event in the same transaction as the change and enqueue it from there, so nothing gets lost if the app crashes between saving and enqueueing.
- SSRF protection: reject webhook URLs that point to localhost or private/internal addresses (10.x, 192.168.x, 169.254.x…).
- Replay protection: include a timestamp in the signed content and reject old deliveries.
- Encrypted secrets: store webhook secrets encrypted (e.g. with ASP.NET Core Data Protection), not as plain text.
Repository
You can find the full source code of the app, including identity, distributed logging, tracing and monitoring, in the open source GitHub repo:
Top comments (1)
Great, thank you