Sincronizzazione tra un'app online e una offline con CloudEvents: l'implementazione in C#

Ultimo articolo della serie sulla sincronizzazione tra un'app online e una offline con CloudEvents 1.0 (evento standardizzato, pattern outbox/inbox, push/pull/ack). Chiudiamo con C#: il server online è una ASP.NET Core Minimal API, il client offline un worker service .NET — lo scenario da cui questa serie di articoli è partita, qui completato con entrambi i lati implementati nello stesso linguaggio.

Riepilogo dell'architettura

Il formato resta CloudEvents 1.0: specversion, id univoco (idempotenza), type, source, time, data. Tre operazioni sul server (push, pull, ack), un outbox lato client per gli eventi in uscita, un registro degli eventi applicati per restare idempotenti su quelli in ingresso.

L'app online: ASP.NET Core Minimal API

Niente controller, niente Entity Framework: le Minimal API introdotte in .NET 6 bastano per tre endpoint, e Microsoft.Data.Sqlite è sufficiente come livello dati senza bisogno di un ORM. Le proprietà C# sono in PascalCase per convenzione del linguaggio, ma [JsonPropertyName] le mappa sui nomi minuscoli richiesti dal formato CloudEvents, in modo che il JSON sul filo resti identico a quello degli altri articoli della serie:

// Program.cs — app online: server di sincronizzazione basato su CloudEvents 1.0 (ASP.NET Core Minimal API)
using System.Text.Json;
using System.Text.Json.Serialization;
using Microsoft.Data.Sqlite;

var builder = WebApplication.CreateBuilder(args);
var app = builder.Build();

var apiKey = Environment.GetEnvironmentVariable("SYNC_API_KEY") ?? "dev-secret";
var connectionString = "Data Source=events.db";

InitializeDatabase(connectionString);

// Autenticazione minimale via API key statica (in produzione: JWT o mTLS)
bool IsAuthorized(HttpRequest request) =>
    request.Headers.TryGetValue("X-Api-Key", out var key) && key == apiKey;

// Il client offline invia gli eventi accumulati mentre non era connesso
app.MapPost("/api/sync/push", async (HttpRequest request) =>
{
    if (!IsAuthorized(request)) return Results.Json(new { error = "unauthorized" }, statusCode: 401);

    var body = await JsonSerializer.DeserializeAsync<PushRequest>(request.Body);
    if (body?.ClientId is null || body.Events is null)
        return Results.Json(new { error = "clientId ed events[] sono obbligatori" }, statusCode: 400);

    var accepted = new List<object>();
    var rejected = new List<object>();

    using var connection = new SqliteConnection(connectionString);
    connection.Open();

    foreach (var ev in body.Events)
    {
        if (!IsValidCloudEvent(ev))
        {
            rejected.Add(new { id = ev.Id, reason = "evento non conforme a CloudEvents 1.0" });
            continue;
        }

        // inserted=false => id già visto in precedenza, scartato per idempotenza
        var inserted = InsertEventIfNew(connection, ev);
        accepted.Add(new { id = ev.Id, inserted });
    }

    return Results.Json(new { accepted, rejected, serverTime = DateTime.UtcNow.ToString("o") });
});

// Il client offline scarica gli eventi generati altrove dopo l'ultima sync
app.MapGet("/api/sync/pull", (HttpRequest request, string clientId) =>
{
    if (!IsAuthorized(request)) return Results.Json(new { error = "unauthorized" }, statusCode: 401);
    if (string.IsNullOrEmpty(clientId)) return Results.Json(new { error = "clientId obbligatorio" }, statusCode: 400);

    using var connection = new SqliteConnection(connectionString);
    connection.Open();

    var since = GetClientCursor(connection, clientId);
    var (events, cursor) = GetEventsSince(connection, since, 200);

    return Results.Json(new { events, cursor });
});

// Il client conferma fino a dove ha applicato gli eventi ricevuti
app.MapPost("/api/sync/ack", async (HttpRequest request) =>
{
    if (!IsAuthorized(request)) return Results.Json(new { error = "unauthorized" }, statusCode: 401);

    var body = await JsonSerializer.DeserializeAsync<AckRequest>(request.Body);
    if (body?.ClientId is null) return Results.Json(new { error = "clientId e cursor sono obbligatori" }, statusCode: 400);

    using var connection = new SqliteConnection(connectionString);
    connection.Open();
    SetClientCursor(connection, body.ClientId, body.Cursor);

    return Results.Json(new { ok = true });
});

app.Run("http://0.0.0.0:3000");

// --- persistenza: event store append-only + cursore per client ---

void InitializeDatabase(string connString)
{
    using var connection = new SqliteConnection(connString);
    connection.Open();
    var command = connection.CreateCommand();
    command.CommandText = @"
        CREATE TABLE IF NOT EXISTS events (
            seq             INTEGER PRIMARY KEY AUTOINCREMENT,
            id              TEXT UNIQUE NOT NULL,
            specversion     TEXT NOT NULL,
            type            TEXT NOT NULL,
            source          TEXT NOT NULL,
            time            TEXT NOT NULL,
            datacontenttype TEXT,
            data            TEXT NOT NULL,
            received_at     TEXT NOT NULL DEFAULT (datetime('now'))
        );

        CREATE TABLE IF NOT EXISTS client_cursors (
            client_id TEXT PRIMARY KEY,
            last_seq  INTEGER NOT NULL DEFAULT 0
        );
    ";
    command.ExecuteNonQuery();
}

bool IsValidCloudEvent(CloudEvent ev) =>
    ev.SpecVersion == "1.0" && !string.IsNullOrEmpty(ev.Id) && !string.IsNullOrEmpty(ev.Type)
    && !string.IsNullOrEmpty(ev.Source) && !string.IsNullOrEmpty(ev.Time);

bool InsertEventIfNew(SqliteConnection connection, CloudEvent ev)
{
    var checkCmd = connection.CreateCommand();
    checkCmd.CommandText = "SELECT 1 FROM events WHERE id = $id";
    checkCmd.Parameters.AddWithValue("$id", ev.Id);
    if (checkCmd.ExecuteScalar() is not null) return false;

    var insertCmd = connection.CreateCommand();
    insertCmd.CommandText = @"
        INSERT INTO events (id, specversion, type, source, time, datacontenttype, data)
        VALUES ($id, $specversion, $type, $source, $time, $datacontenttype, $data)
    ";
    insertCmd.Parameters.AddWithValue("$id", ev.Id);
    insertCmd.Parameters.AddWithValue("$specversion", ev.SpecVersion);
    insertCmd.Parameters.AddWithValue("$type", ev.Type);
    insertCmd.Parameters.AddWithValue("$source", ev.Source);
    insertCmd.Parameters.AddWithValue("$time", ev.Time);
    insertCmd.Parameters.AddWithValue("$datacontenttype", ev.DataContentType ?? "application/json");
    insertCmd.Parameters.AddWithValue("$data", JsonSerializer.Serialize(ev.Data));
    insertCmd.ExecuteNonQuery();

    return true;
}

(List<CloudEvent> events, long cursor) GetEventsSince(SqliteConnection connection, long seq, int limit)
{
    var command = connection.CreateCommand();
    command.CommandText = @"
        SELECT seq, id, specversion, type, source, time, datacontenttype, data
        FROM events WHERE seq > $seq ORDER BY seq ASC LIMIT $limit
    ";
    command.Parameters.AddWithValue("$seq", seq);
    command.Parameters.AddWithValue("$limit", limit);

    var events = new List<CloudEvent>();
    var cursor = seq;

    using var reader = command.ExecuteReader();
    while (reader.Read())
    {
        cursor = reader.GetInt64(0);
        events.Add(new CloudEvent
        {
            SpecVersion = reader.GetString(2),
            Id = reader.GetString(1),
            Type = reader.GetString(3),
            Source = reader.GetString(4),
            Time = reader.GetString(5),
            DataContentType = reader.IsDBNull(6) ? null : reader.GetString(6),
            Data = JsonDocument.Parse(reader.GetString(7)).RootElement
        });
    }

    return (events, cursor);
}

long GetClientCursor(SqliteConnection connection, string clientId)
{
    var command = connection.CreateCommand();
    command.CommandText = "SELECT last_seq FROM client_cursors WHERE client_id = $clientId";
    command.Parameters.AddWithValue("$clientId", clientId);
    var result = command.ExecuteScalar();
    return result is null ? 0 : Convert.ToInt64(result);
}

void SetClientCursor(SqliteConnection connection, string clientId, long seq)
{
    var command = connection.CreateCommand();
    command.CommandText = @"
        INSERT INTO client_cursors (client_id, last_seq) VALUES ($clientId, $seq)
        ON CONFLICT(client_id) DO UPDATE SET last_seq = excluded.last_seq
    ";
    command.Parameters.AddWithValue("$clientId", clientId);
    command.Parameters.AddWithValue("$seq", seq);
    command.ExecuteNonQuery();
}

// --- modelli JSON ---

class CloudEvent
{
    [JsonPropertyName("specversion")]
    public string SpecVersion { get; set; } = "1.0";

    [JsonPropertyName("id")]
    public string Id { get; set; } = default!;

    [JsonPropertyName("type")]
    public string Type { get; set; } = default!;

    [JsonPropertyName("source")]
    public string Source { get; set; } = default!;

    [JsonPropertyName("time")]
    public string Time { get; set; } = default!;

    [JsonPropertyName("datacontenttype")]
    public string? DataContentType { get; set; }

    [JsonPropertyName("data")]
    public JsonElement Data { get; set; }
}

class PushRequest
{
    [JsonPropertyName("clientId")]
    public string? ClientId { get; set; }

    [JsonPropertyName("events")]
    public List<CloudEvent>? Events { get; set; }
}

class AckRequest
{
    [JsonPropertyName("clientId")]
    public string? ClientId { get; set; }

    [JsonPropertyName("cursor")]
    public long Cursor { get; set; }
}

<Project Sdk="Microsoft.NET.Sdk.Web">

  <PropertyGroup>
    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
    <InvariantGlobalization>true</InvariantGlobalization>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="Microsoft.Data.Sqlite" Version="8.0.8" />
  </ItemGroup>

</Project>

L'app offline: outbox locale e worker service

Il client è lo stesso pattern outbox/inbox degli altri articoli, qui come BackgroundService .NET con SQLite via Microsoft.Data.Sqlite:

// CloudEvent.cs — rappresentazione minimale di un evento CloudEvents 1.0 in JSON
using System.Text.Json.Serialization;

namespace OfflineSync;

public class CloudEvent
{
    [JsonPropertyName("specversion")]
    public string SpecVersion { get; set; } = "1.0";

    [JsonPropertyName("id")]
    public string Id { get; set; } = Guid.NewGuid().ToString();

    // Convenzione consigliata: reverse-DNS, per evitare collisioni tra domini/app diverse
    // es. "com.gabrieleromanato.offlineapp.record.created"
    [JsonPropertyName("type")]
    public string Type { get; set; } = default!;

    // Identifica chi ha generato l'evento (es. "urn:client:offline-client-01")
    [JsonPropertyName("source")]
    public string Source { get; set; } = default!;

    [JsonPropertyName("time")]
    public string Time { get; set; } = DateTime.UtcNow.ToString("o");

    [JsonPropertyName("datacontenttype")]
    public string DataContentType { get; set; } = "application/json";

    [JsonPropertyName("data")]
    public object? Data { get; set; }
}

// LocalDb.cs — persistenza locale dell'app offline: outbox (eventi da inviare),
// eventi applicati (inbox) e cursore di sincronizzazione.
using System.Text.Json;
using Microsoft.Data.Sqlite;

namespace OfflineSync;

public class LocalDb
{
    private readonly string _connectionString;

    public LocalDb(string dbPath)
    {
        _connectionString = $"Data Source={dbPath}";
        Initialize();
    }

    private SqliteConnection Open()
    {
        var conn = new SqliteConnection(_connectionString);
        conn.Open();
        return conn;
    }

    private void Initialize()
    {
        using var conn = Open();
        var cmd = conn.CreateCommand();
        cmd.CommandText = @"
            CREATE TABLE IF NOT EXISTS outbox (
                id     TEXT PRIMARY KEY,
                type   TEXT NOT NULL,
                source TEXT NOT NULL,
                time   TEXT NOT NULL,
                data   TEXT NOT NULL,
                sent   INTEGER NOT NULL DEFAULT 0
            );

            CREATE TABLE IF NOT EXISTS applied_events (
                id         TEXT PRIMARY KEY,
                type       TEXT NOT NULL,
                data       TEXT NOT NULL,
                applied_at TEXT NOT NULL DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS sync_state (
                key   TEXT PRIMARY KEY,
                value TEXT
            );
        ";
        cmd.ExecuteNonQuery();
    }

    // Chiamato dal dominio offline ogni volta che avviene un cambiamento locale
    // (es. dopo aver salvato un record nel proprio DB applicativo)
    public void EnqueueEvent(CloudEvent ev)
    {
        using var conn = Open();
        var cmd = conn.CreateCommand();
        cmd.CommandText = @"
            INSERT INTO outbox (id, type, source, time, data, sent)
            VALUES ($id, $type, $source, $time, $data, 0)
        ";
        cmd.Parameters.AddWithValue("$id", ev.Id);
        cmd.Parameters.AddWithValue("$type", ev.Type);
        cmd.Parameters.AddWithValue("$source", ev.Source);
        cmd.Parameters.AddWithValue("$time", ev.Time);
        cmd.Parameters.AddWithValue("$data", JsonSerializer.Serialize(ev.Data));
        cmd.ExecuteNonQuery();
    }

    public List<CloudEvent> GetUnsentEvents(int limit = 200)
    {
        var result = new List<CloudEvent>();
        using var conn = Open();
        var cmd = conn.CreateCommand();
        cmd.CommandText = "SELECT id, type, source, time, data FROM outbox WHERE sent = 0 ORDER BY rowid ASC LIMIT $limit";
        cmd.Parameters.AddWithValue("$limit", limit);
        using var reader = cmd.ExecuteReader();
        while (reader.Read())
        {
            result.Add(new CloudEvent
            {
                Id = reader.GetString(0),
                Type = reader.GetString(1),
                Source = reader.GetString(2),
                Time = reader.GetString(3),
                Data = JsonDocument.Parse(reader.GetString(4)).RootElement
            });
        }
        return result;
    }

    public void MarkSent(IEnumerable<string> ids)
    {
        using var conn = Open();
        using var transaction = conn.BeginTransaction();
        foreach (var id in ids)
        {
            var cmd = conn.CreateCommand();
            cmd.Transaction = transaction;
            cmd.CommandText = "UPDATE outbox SET sent = 1 WHERE id = $id";
            cmd.Parameters.AddWithValue("$id", id);
            cmd.ExecuteNonQuery();
        }
        transaction.Commit();
    }

    // Applica un evento ricevuto dal server al dominio locale.
    // "INSERT OR IGNORE" garantisce idempotenza se lo stesso evento arriva due volte.
    public void ApplyIncomingEvent(CloudEvent ev)
    {
        using var conn = Open();
        var cmd = conn.CreateCommand();
        cmd.CommandText = @"
            INSERT OR IGNORE INTO applied_events (id, type, data)
            VALUES ($id, $type, $data)
        ";
        cmd.Parameters.AddWithValue("$id", ev.Id);
        cmd.Parameters.AddWithValue("$type", ev.Type);
        cmd.Parameters.AddWithValue("$data", ev.Data?.ToString() ?? "{}");
        cmd.ExecuteNonQuery();

        // Qui va la logica reale: deserializzare ev.Data in base a ev.Type
        // e aggiornare l'entità corrispondente nel DB applicativo offline
        // (es. inserire/aggiornare un Task, un Cliente, ecc.)
    }

    public long GetCursor()
    {
        using var conn = Open();
        var cmd = conn.CreateCommand();
        cmd.CommandText = "SELECT value FROM sync_state WHERE key = 'cursor'";
        var value = cmd.ExecuteScalar();
        return value is null ? 0 : long.Parse((string)value);
    }

    public void SetCursor(long cursor)
    {
        using var conn = Open();
        var cmd = conn.CreateCommand();
        cmd.CommandText = @"
            INSERT INTO sync_state (key, value) VALUES ('cursor', $value)
            ON CONFLICT(key) DO UPDATE SET value = excluded.value
        ";
        cmd.Parameters.AddWithValue("$value", cursor.ToString());
        cmd.ExecuteNonQuery();
    }
}

// SyncService.cs — servizio in background che tenta la sincronizzazione a intervalli
// regolari. Se la rete non è disponibile fallisce silenziosamente e riprova al giro
// successivo: è questo che rende l'app "offline-first" invece che "online-required".
using System.Net.Http.Json;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace OfflineSync;

public class SyncService : BackgroundService
{
    private readonly LocalDb _db;
    private readonly HttpClient _http;
    private readonly ILogger<SyncService> _logger;
    private readonly string _clientId;
    private readonly TimeSpan _interval;

    public SyncService(LocalDb db, HttpClient http, ILogger<SyncService> logger, string clientId, TimeSpan interval)
    {
        _db = db;
        _http = http;
        _logger = logger;
        _clientId = clientId;
        _interval = interval;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await TrySyncAsync(stoppingToken);
            }
            catch (HttpRequestException ex)
            {
                // Rete assente o server irraggiungibile: comportamento atteso per un
                // client offline-first. Si riprova semplicemente al prossimo giro.
                _logger.LogWarning("Sync non riuscita (rete assente?): {Message}", ex.Message);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Errore inatteso durante la sincronizzazione");
            }

            try
            {
                await Task.Delay(_interval, stoppingToken);
            }
            catch (TaskCanceledException)
            {
                // shutdown in corso
            }
        }
    }

    public async Task TrySyncAsync(CancellationToken ct)
    {
        // 1) PUSH: invia gli eventi locali accumulati mentre si era offline
        var pending = _db.GetUnsentEvents();
        if (pending.Count > 0)
        {
            var pushResponse = await _http.PostAsJsonAsync("/api/sync/push", new
            {
                clientId = _clientId,
                events = pending
            }, ct);

            pushResponse.EnsureSuccessStatusCode();
            _db.MarkSent(pending.Select(e => e.Id));
            _logger.LogInformation("Inviati {Count} eventi al server", pending.Count);
        }

        // 2) PULL: scarica gli eventi generati altrove dopo l'ultima sync
        var pullResponse = await _http.GetAsync(
            $"/api/sync/pull?clientId={Uri.EscapeDataString(_clientId)}", ct);
        pullResponse.EnsureSuccessStatusCode();

        var payload = await pullResponse.Content.ReadFromJsonAsync<PullResponse>(cancellationToken: ct);
        if (payload is null) return;

        foreach (var ev in payload.Events)
        {
            _db.ApplyIncomingEvent(ev);
        }

        // 3) ACK: conferma al server fino a dove si è applicato, per avanzare il cursore
        if (payload.Events.Count > 0)
        {
            await _http.PostAsJsonAsync("/api/sync/ack", new { clientId = _clientId, cursor = payload.Cursor }, ct);
            _db.SetCursor(payload.Cursor);
            _logger.LogInformation("Applicati {Count} eventi ricevuti dal server", payload.Events.Count);
        }
    }

    private class PullResponse
    {
        public List<CloudEvent> Events { get; set; } = new();
        public long Cursor { get; set; }
    }
}

// Program.cs — bootstrap dell'app offline come worker service .NET
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using OfflineSync;

var clientId = "offline-client-01"; // in produzione: id univoco per postazione/dispositivo
var apiKey = "dev-secret";          // deve combaciare con SYNC_API_KEY del server online
var serverBaseUrl = "http://localhost:3000";

var db = new LocalDb("offline.db");

var http = new HttpClient { BaseAddress = new Uri(serverBaseUrl) };
http.DefaultRequestHeaders.Add("x-api-key", apiKey);

var builder = Host.CreateApplicationBuilder(args);
builder.Services.AddSingleton(db);
builder.Services.AddSingleton(http);
builder.Services.AddHostedService(sp => new SyncService(
    db,
    http,
    sp.GetRequiredService<ILogger<SyncService>>(),
    clientId,
    interval: TimeSpan.FromSeconds(30)
));

var host = builder.Build();

// Esempio: simula un cambiamento avvenuto nel dominio offline.
// Nella tua app reale, questa chiamata andrebbe fatta subito dopo ogni
// scrittura locale rilevante (es. dopo aver salvato un'entità nel tuo DB).
db.EnqueueEvent(new CloudEvent
{
    Type = "com.gabrieleromanato.offlineapp.record.created",
    Source = $"urn:client:{clientId}",
    Data = new { recordId = 42, note = "Creato mentre offline" }
});

await host.RunAsync();

<Project Sdk="Microsoft.NET.Sdk.Worker">

  <PropertyGroup>
    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
    <RootNamespace>OfflineSync</RootNamespace>
    <InvariantGlobalization>true</InvariantGlobalization>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="Microsoft.Data.Sqlite" Version="8.0.8" />
    <PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.0" />
  </ItemGroup>

</Project>

Provarlo

cd online-csharp && dotnet run
# in un altro terminale
cd offline-csharp && dotnet run

Il worker offline accoda subito l'evento di esempio, prova a sincronizzarsi con il server ogni 30 secondi, e se il server non è raggiungibile lo registra come warning senza interrompersi. Anche qui il codice è stato scritto e rivisto con attenzione ma non compilato in questo ambiente, perché l'accesso a NuGet non è disponibile in questo sandbox: prova la build con dotnet build nel tuo ambiente prima di fidartene.

Cosa manca per la produzione

Le stesse lacune emerse in tutti gli articoli della serie restano valide anche qui: autenticazione più solida di una API key statica (JWT con refresh o mTLS), retry con backoff esponenziale invece dell'intervallo fisso di 30 secondi, un filtro lato client sugli eventi con source uguale al proprio per non riapplicarsi da solo ciò che ha generato, validazione dello schema di data per ogni type, e soprattutto una strategia esplicita di conflict resolution se due client modificano lo stesso record mentre erano entrambi offline — qui, come negli altri esempi della serie, si assume implicitamente last-write-wins, che basta solo se i tuoi eventi sono per lo più additivi.

Chiudendo la serie

Sette linguaggi, stessa architettura: un evento CloudEvents 1.0 come contratto comune, un pattern outbox/inbox per disaccoppiare il dominio dalla rete, tre operazioni (push, pull, ack) e un cursore per client per sapere chi ha visto cosa. Cambia la sintassi, cambiano gli strumenti di persistenza idiomatici per ogni stack (file JSON in Go, SQLite via driver nativo in Node.js e Python, Eloquent in Laravel, PDO in PHP puro, JPA in Spring Boot, Microsoft.Data.Sqlite in C#), ma il problema e la soluzione restano gli stessi: se il client può stare offline per la maggior parte del tempo, la sincronizzazione non può essere un dettaglio implementativo aggiunto in un secondo momento — va progettata come processo asincrono fin dall'inizio.