feat: raise event on encode status change (#47)

app/src/components/interfaces/encode/chips/EncodeStatus.tsx

@@ -4,24 +4,46 @@ import type { ChipProps } from '@giantnodes/react'
 import { Chip } from '@giantnodes/react'
 import dayjs from 'dayjs'
 import React from 'react'
-import { graphql, useFragment } from 'react-relay'
+import { graphql, useFragment, useSubscription } from 'react-relay'
 
 const FRAGMENT = graphql`
   fragment EncodeStatusFragment on Encode {
+    id
     status
     failed_at
     cancelled_at
     completed_at
   }
 `
 
+const SUBSCRIPTION = graphql`
+  subscription EncodeStatusSubscription($where: EncodeFilterInput) {
+    encode_status_changed(where: $where) {
+      ...EncodeSpeedFragment
+    }
+  }
+`
+
 type EncodeStatusChipProps = {
   $key: EncodeStatusFragment$key
 }
 
 const EncodeStatus: React.FC<EncodeStatusChipProps> = ({ $key }) => {
   const data = useFragment(FRAGMENT, $key)
 
+  useSubscription({
+    subscription: SUBSCRIPTION,
+    variables: {
+      variables: {
+        where: {
+          id: {
+            eq: data.id,
+          },
+        },
+      },
+    },
+  })
+
   const color = React.useMemo<ChipProps['color']>(() => {
     switch (data.status) {
       case 'COMPLETED':

src/Service.Dashboard/src/Application.Components/Encodes/Events/RaiseEncodeStatusChangedTopic.cs

@@ -0,0 +1,25 @@
+using Giantnodes.Service.Dashboard.Application.Contracts.Encodes.Events;
+using Giantnodes.Service.Dashboard.Domain.Aggregates.Encodes.Repositories;
+using HotChocolate.Subscriptions;
+using MassTransit;
+
+namespace Giantnodes.Service.Dashboard.Application.Components.Encodes.Events;
+
+public class RaiseEncodeStatusChangedTopic : IConsumer<EncodeStatusChangedEvent>
+{
+    private readonly IEncodeRepository _repository;
+    private readonly ITopicEventSender _sender;
+
+    public RaiseEncodeStatusChangedTopic(IEncodeRepository repository, ITopicEventSender sender)
+    {
+        _repository = repository;
+        _sender = sender;
+    }
+
+    public async Task Consume(ConsumeContext<EncodeStatusChangedEvent> context)
+    {
+        var encode = await _repository.SingleAsync(x => x.Id == context.Message.EncodeId);
+
+        await _sender.SendAsync(nameof(EncodeStatusChangedEvent), encode, context.CancellationToken);
+    }
+}

src/Service.Dashboard/src/Application.Contracts/Encodes/Events/EncodeStatusChangedEvent.cs

@@ -0,0 +1,13 @@
+using Giantnodes.Infrastructure.Domain.Events;
+using Giantnodes.Service.Dashboard.Domain.Shared.Enums;
+
+namespace Giantnodes.Service.Dashboard.Application.Contracts.Encodes.Events;
+
+public sealed record EncodeStatusChangedEvent : DomainEvent
+{
+    public required Guid EncodeId { get; init; }
+
+    public required EncodeStatus FromStatus { get; init; }
+
+    public required EncodeStatus ToStatus { get; init; }
+}

src/Service.Dashboard/src/Domain/Aggregates/Encodes/Encode.cs

@@ -16,6 +16,8 @@ public class Encode : AggregateRoot<Guid>, ITimestampableEntity
 {
     private readonly List<EncodeSnapshot> _snapshots = new();
 
+    private EncodeStatus _status;
+
     /// <summary>
     /// The file being encoded.
     /// </summary>
@@ -34,7 +36,17 @@ public class Encode : AggregateRoot<Guid>, ITimestampableEntity
     /// <summary>
     /// The current status of the encoding process.
     /// </summary>
-    public EncodeStatus Status { get; private set; }
+    public EncodeStatus Status
+    {
+        get => _status;
+        private set
+        {
+            if (Status != value)
+                DomainEvents.Add(new EncodeStatusChangedEvent { EncodeId = Id, FromStatus = Status, ToStatus = value });
+
+            _status = value;
+        }
+    }
 
     /// <summary>
     /// The machine performing the encoding.

src/Service.Dashboard/src/HttpApi/Resolvers/Encodes/Subscriptions/EncodeStatusChangedSubscription.cs

@@ -0,0 +1,23 @@
+using Giantnodes.Service.Dashboard.Application.Contracts.Encodes.Events;
+using Giantnodes.Service.Dashboard.Domain.Aggregates.Encodes;
+using Giantnodes.Service.Dashboard.Persistence.DbContexts;
+using Microsoft.EntityFrameworkCore;
+
+namespace Giantnodes.Service.Dashboard.HttpApi.Resolvers.Encodes.Subscriptions;
+
+[ExtendObjectType(OperationTypeNames.Subscription)]
+public class EncodeStatusChangedSubscription
+{
+    [Subscribe]
+    [Topic(nameof(EncodeStatusChangedEvent))]
+    [UseSingleOrDefault]
+    [UseProjection]
+    [UseFiltering]
+    [UseSorting]
+    public IQueryable<Encode> EncodeStatusChanged(
+        [Service] ApplicationDbContext database,
+        [EventMessage] Encode encode)
+    {
+        return database.Encodes.Where(x => x.Id == encode.Id).AsNoTracking();
+    }
+}