Since lesson 3.5.2, every notice has been getting into the database through the queue tender-import. Now other parts of the Kladde should react to it: the assessment should look at tenders that fit the office, and people who watch certain tenders should get an email. The two know nothing about each other, and the import should know about neither of them.

That is what topics are for. After each import the worker publishes an event to the topic tender-events, and two subscriptions distribute it further: assessment only for selected CPV codes, notification for all. Sessions make sure that the events of one tender are processed in the order in which they arose.

From command to event

ImportTenderNotice is a command to exactly one receiver. What happens afterwards is an event: a tender is new, or it has changed. For that, Kladde.Contracts gets two records and a shared interface:

C#
// src/Kladde.Contracts/TenderEvents.cs
namespace Kladde.Contracts;

/// <summary>What every event about a tender carries.</summary>
public interface ITenderEvent
{
    string NoticeId { get; }

    /// <summary>Counts the tender's versions: 1 on publication, one more with each change.</summary>
    int Version { get; }

    string MainCpv { get; }
}

/// <summary>Event: a tender was imported for the first time.</summary>
public sealed record TenderPublished(string NoticeId, int Version, string Title, string MainCpv, DateTimeOffset? Deadline)
    : ITenderEvent;

/// <summary>Event: an imported tender has changed (title, main CPV code or deadline).</summary>
public sealed record TenderChanged(string NoticeId, int Version, string Title, string MainCpv, DateTimeOffset? Deadline)
    : ITenderEvent;

An event should arise only when something has really happened. If the fetch imports the same notice a second time unchanged, nobody may get an email "tender changed". For that, the upsert from K2 has to say more than "new or not". It compares title, main CPV code and deadline with the stored state and counts a version number up only when something has changed. The tenders table gets a column Version for that (a migration as in K2), and ITenderStore returns a result instead of a bool:

C#
// src/Kladde.Core/Tenders.cs (excerpt)
/// <summary>What an upsert did: created the tender, changed it, or found it unchanged; and the version it has now.</summary>
public sealed record UpsertResult(bool Created, bool Changed, int Version);

/// <summary>Where tenders are kept; the EF Core adapter of K2 implements it.</summary>
public interface ITenderStore
{
    /// <summary>Inserts or updates by notice number; counts the version up only when title, main CPV code or deadline changed.</summary>
    Task<UpsertResult> UpsertAsync(TenderNotice notice, CancellationToken cancellationToken);
}

From that, the TenderImportHandler from lesson 3.5.2 decides which event it publishes:

C#
// src/Kladde.Worker/TenderImportHandler.cs
using Kladde.Contracts;
using Kladde.Core;
using Kladde.Worker.Events;

namespace Kladde.Worker;

/// <summary>Imports one notice: raw file from Blob Storage, an upsert into SQL, then an event if something changed.</summary>
public sealed partial class TenderImportHandler(
    IRawTenderSource source,
    ITenderStore store,
    ITenderEventPublisher events,
    ILogger<TenderImportHandler> logger)
{
    public async Task HandleAsync(ImportTenderNotice command, CancellationToken cancellationToken)
    {
        var json = await source.ReadAsync(command.BlobName, cancellationToken)
            ?? throw new InvalidOperationException($"The raw notice '{command.BlobName}' does not exist.");

        var notice = TedNoticeParser.Parse(json);
        if (notice.NoticeId != command.NoticeId)
        {
            throw new InvalidOperationException($"'{command.BlobName}' holds notice '{notice.NoticeId}', not '{command.NoticeId}'.");
        }

        var result = await store.UpsertAsync(notice, cancellationToken);
        ITenderEvent? tenderEvent = result switch
        {
            { Created: true } => new TenderPublished(notice.NoticeId, result.Version, notice.Title, notice.MainCpv, notice.Deadline),
            { Changed: true } => new TenderChanged(notice.NoticeId, result.Version, notice.Title, notice.MainCpv, notice.Deadline),
            _ => null, // imported again without a change: nothing to tell anyone
        };
        if (tenderEvent is not null)
        {
            await events.PublishAsync(tenderEvent, cancellationToken);
        }

        Imported(logger, notice.NoticeId, result.Version);
    }

    [LoggerMessage(Level = LogLevel.Information, Message = "Tender {NoticeId} imported, version {Version}")]
    private static partial void Imported(ILogger logger, string noticeId, int version);
}

This code contains a bug that you deliberately leave in until lesson 3.5.4 fixes it: the upsert is saved before the event is sent. If sending fails, the processor abandons the message, it comes back, and the upsert now reports "nothing changed". The event is lost. Saving and sending cannot be done in one step; that is exactly what the outbox is for.