Sincronizzazione tra un'app online e una offline con CloudEvents: l'implementazione in PHP puro

Quinto articolo della serie sulla sincronizzazione tra un'app online e una offline (CloudEvents 1.0, pattern outbox/inbox, push/pull/ack). Dopo Go, Node.js, Laravel e Python, qui togliamo il framework: niente routing, niente ORM, niente client HTTP di libreria. Solo PHP, PDO e cURL, per vedere esattamente cosa fa un framework al posto tuo quando decidi di non usarlo.

Riepilogo dell'architettura

Il formato resta CloudEvents 1.0: specversion, id univoco, 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: persistenza con PDO

<?php

// config.php — configurazione dell'app online
return [
    'db_path' => __DIR__ . '/events.db',
    'api_key' => getenv('API_KEY') ?: 'dev-secret',
];

<?php

// db.php — persistenza dell'app online: event store append-only + cursore per client

function db_connect(string $path): PDO
{
    $pdo = new PDO('sqlite:' . $path);
    $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
    $pdo->exec('PRAGMA journal_mode = WAL');

    $pdo->exec('
        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\'))
        )
    ');

    $pdo->exec('
        CREATE TABLE IF NOT EXISTS client_cursors (
            client_id TEXT PRIMARY KEY,
            last_seq  INTEGER NOT NULL DEFAULT 0
        )
    ');

    return $pdo;
}

// Inserisce l'evento solo se il suo id non è già stato visto (idempotenza)
function insert_event_if_new(PDO $pdo, array $event): bool
{
    $stmt = $pdo->prepare('SELECT 1 FROM events WHERE id = ?');
    $stmt->execute([$event['id']]);
    if ($stmt->fetchColumn()) {
        return false;
    }

    $stmt = $pdo->prepare('
        INSERT INTO events (id, specversion, type, source, time, datacontenttype, data)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    ');
    $stmt->execute([
        $event['id'],
        $event['specversion'],
        $event['type'],
        $event['source'],
        $event['time'],
        $event['datacontenttype'] ?? 'application/json',
        json_encode($event['data'] ?? new stdClass()),
    ]);

    return true;
}

function events_since(PDO $pdo, int $seq, int $limit = 200): array
{
    $stmt = $pdo->prepare('
        SELECT seq, id, specversion, type, source, time, datacontenttype, data
        FROM events WHERE seq > ? ORDER BY seq ASC LIMIT ?
    ');
    $stmt->bindValue(1, $seq, PDO::PARAM_INT);
    $stmt->bindValue(2, $limit, PDO::PARAM_INT);
    $stmt->execute();

    $result = [];
    foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) {
        $result[] = [
            '_seq' => (int) $row['seq'],
            'specversion' => $row['specversion'],
            'id' => $row['id'],
            'type' => $row['type'],
            'source' => $row['source'],
            'time' => $row['time'],
            'datacontenttype' => $row['datacontenttype'],
            'data' => json_decode($row['data'], true),
        ];
    }

    return $result;
}

function get_client_cursor(PDO $pdo, string $clientId): int
{
    $stmt = $pdo->prepare('SELECT last_seq FROM client_cursors WHERE client_id = ?');
    $stmt->execute([$clientId]);
    $value = $stmt->fetchColumn();
    return $value === false ? 0 : (int) $value;
}

function set_client_cursor(PDO $pdo, string $clientId, int $seq): void
{
    $stmt = $pdo->prepare('
        INSERT INTO client_cursors (client_id, last_seq) VALUES (?, ?)
        ON CONFLICT(client_id) DO UPDATE SET last_seq = excluded.last_seq
    ');
    $stmt->execute([$clientId, $seq]);
}

L'app online: il front controller

Senza un framework, il routing è un semplice controllo su metodo HTTP e percorso, fatto a mano dentro un unico script che il server integrato di PHP usa come router:

<?php

// index.php — app online: front controller ed endpoint di sincronizzazione basati su CloudEvents 1.0
declare(strict_types=1);

require __DIR__ . '/db.php';

$config = require __DIR__ . '/config.php';
$pdo = db_connect($config['db_path']);

header('Content-Type: application/json');

// Autenticazione minimale via API key statica (in produzione: JWT o mTLS)
function require_api_key(array $config): bool
{
    $headers = getallheaders();
    $key = $headers['X-Api-Key'] ?? $headers['x-api-key'] ?? '';
    if ($key !== $config['api_key']) {
        http_response_code(401);
        echo json_encode(['error' => 'unauthorized']);
        return false;
    }
    return true;
}

function is_valid_cloud_event($event): bool
{
    return is_array($event)
        && ($event['specversion'] ?? null) === '1.0'
        && !empty($event['id'])
        && !empty($event['type'])
        && !empty($event['source'])
        && !empty($event['time']);
}

$method = $_SERVER['REQUEST_METHOD'];
$path = parse_url($_SERVER['REQUEST_URI'], PHP_URL_PATH);

// Il client offline invia gli eventi accumulati mentre non era connesso
if ($method === 'POST' && $path === '/api/sync/push') {
    if (!require_api_key($config)) exit;

    $body = json_decode(file_get_contents('php://input'), true) ?? [];
    $clientId = $body['clientId'] ?? null;
    $events = $body['events'] ?? null;

    if (!$clientId || !is_array($events)) {
        http_response_code(400);
        echo json_encode(['error' => 'clientId ed events[] sono obbligatori']);
        exit;
    }

    $accepted = [];
    $rejected = [];

    foreach ($events as $event) {
        if (!is_valid_cloud_event($event)) {
            $rejected[] = ['id' => $event['id'] ?? null, 'reason' => 'evento non conforme a CloudEvents 1.0'];
            continue;
        }
        // inserted=false => id già visto in precedenza, scartato per idempotenza
        $inserted = insert_event_if_new($pdo, $event);
        $accepted[] = ['id' => $event['id'], 'inserted' => $inserted];
    }

    echo json_encode([
        'accepted' => $accepted,
        'rejected' => $rejected,
        'serverTime' => gmdate('c'),
    ]);
    exit;
}

// Il client offline scarica gli eventi generati altrove dopo l'ultima sync
if ($method === 'GET' && $path === '/api/sync/pull') {
    if (!require_api_key($config)) exit;

    $clientId = $_GET['clientId'] ?? null;
    if (!$clientId) {
        http_response_code(400);
        echo json_encode(['error' => 'clientId obbligatorio']);
        exit;
    }

    $since = get_client_cursor($pdo, $clientId);
    $events = events_since($pdo, $since);

    $cursor = $since;
    $out = [];
    foreach ($events as $ev) {
        $cursor = $ev['_seq'];
        unset($ev['_seq']);
        $out[] = $ev;
    }

    echo json_encode(['events' => $out, 'cursor' => $cursor]);
    exit;
}

// Il client conferma fino a dove ha applicato gli eventi ricevuti
if ($method === 'POST' && $path === '/api/sync/ack') {
    if (!require_api_key($config)) exit;

    $body = json_decode(file_get_contents('php://input'), true) ?? [];
    $clientId = $body['clientId'] ?? null;
    $cursor = $body['cursor'] ?? null;

    if (!$clientId || !is_int($cursor)) {
        http_response_code(400);
        echo json_encode(['error' => 'clientId e cursor sono obbligatori']);
        exit;
    }

    set_client_cursor($pdo, $clientId, $cursor);
    echo json_encode(['ok' => true]);
    exit;
}

http_response_code(404);
echo json_encode(['error' => 'not found']);

L'app offline: outbox e loop di sync con cURL

Stessa struttura vista negli articoli precedenti, con cURL al posto di un client HTTP di libreria:

<?php

// db.php — persistenza locale dell'app offline: outbox ed eventi applicati

function db_connect(string $path): PDO
{
    $pdo = new PDO('sqlite:' . $path);
    $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
    $pdo->exec('PRAGMA journal_mode = WAL');

    $pdo->exec('
        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
        )
    ');

    $pdo->exec('
        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\'))
        )
    ');

    return $pdo;
}

// Chiamata dal dominio offline ogni volta che avviene un cambiamento locale
function enqueue_event(PDO $pdo, array $event): void
{
    $stmt = $pdo->prepare('
        INSERT INTO outbox (id, type, source, time, data, sent) VALUES (?, ?, ?, ?, ?, 0)
    ');
    $stmt->execute([
        $event['id'],
        $event['type'],
        $event['source'],
        $event['time'],
        json_encode($event['data'] ?? new stdClass()),
    ]);
}

function get_unsent_events(PDO $pdo, int $limit = 200): array
{
    $stmt = $pdo->prepare('
        SELECT id, type, source, time, data FROM outbox WHERE sent = 0 ORDER BY rowid ASC LIMIT ?
    ');
    $stmt->bindValue(1, $limit, PDO::PARAM_INT);
    $stmt->execute();

    $result = [];
    foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) {
        $result[] = [
            'specversion' => '1.0',
            'id' => $row['id'],
            'type' => $row['type'],
            'source' => $row['source'],
            'time' => $row['time'],
            'datacontenttype' => 'application/json',
            'data' => json_decode($row['data'], true),
        ];
    }

    return $result;
}

function mark_sent(PDO $pdo, array $ids): void
{
    $stmt = $pdo->prepare('UPDATE outbox SET sent = 1 WHERE id = ?');
    foreach ($ids as $id) {
        $stmt->execute([$id]);
    }
}

// Applica un evento ricevuto dal server al dominio locale; idempotente sull'id
function apply_incoming_event(PDO $pdo, array $event): void
{
    $stmt = $pdo->prepare('INSERT OR IGNORE INTO applied_events (id, type, data) VALUES (?, ?, ?)');
    $stmt->execute([$event['id'], $event['type'], json_encode($event['data'] ?? new stdClass())]);
    // Qui va la logica reale: leggere $event['type']/$event['data'] e
    // aggiornare l'entità corrispondente nel modello di dominio offline.
}

<?php

// sync.php — app offline: prova a sincronizzarsi con il server a intervalli regolari.
// Se la rete manca, la richiesta cURL fallisce e si riprova al giro successivo: è
// questo che rende l'app "offline-first" invece che "online-required".
declare(strict_types=1);

require __DIR__ . '/db.php';

$baseUrl = getenv('SYNC_SERVER_URL') ?: 'http://localhost:3000';
$apiKey = getenv('API_KEY') ?: 'dev-secret';
$clientId = getenv('CLIENT_ID') ?: 'offline-client-01';
$dbPath = __DIR__ . '/offline.db';

$pdo = db_connect($dbPath);

function http_request(string $method, string $url, string $apiKey, ?array $body = null): array
{
    $ch = curl_init($url);
    $headers = ['X-Api-Key: ' . $apiKey];

    if ($body !== null) {
        $headers[] = 'Content-Type: application/json';
        curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($body));
    }

    curl_setopt($ch, CURLOPT_CUSTOMREQUEST, $method);
    curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);
    curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
    curl_setopt($ch, CURLOPT_TIMEOUT, 10);

    $response = curl_exec($ch);
    if ($response === false) {
        $error = curl_error($ch);
        curl_close($ch);
        throw new RuntimeException("richiesta HTTP fallita: {$error}");
    }

    $status = curl_getinfo($ch, CURLINFO_HTTP_CODE);
    curl_close($ch);

    if ($status >= 400) {
        throw new RuntimeException("richiesta HTTP fallita: HTTP {$status}");
    }

    return json_decode($response, true) ?? [];
}

function try_sync(PDO $pdo, string $baseUrl, string $apiKey, string $clientId): void
{
    // 1) PUSH: invia gli eventi locali accumulati mentre si era offline
    $pending = get_unsent_events($pdo);

    if (!empty($pending)) {
        http_request('POST', "{$baseUrl}/api/sync/push", $apiKey, [
            'clientId' => $clientId,
            'events' => $pending,
        ]);

        mark_sent($pdo, array_column($pending, 'id'));
        fwrite(STDOUT, 'Inviati ' . count($pending) . " eventi al server\n");
    }

    // 2) PULL: scarica gli eventi generati altrove dopo l'ultima sync
    $payload = http_request('GET', "{$baseUrl}/api/sync/pull?clientId=" . urlencode($clientId), $apiKey);

    foreach ($payload['events'] ?? [] as $event) {
        apply_incoming_event($pdo, $event);
    }

    // 3) ACK: conferma al server fino a dove si è applicato
    if (!empty($payload['events'])) {
        http_request('POST', "{$baseUrl}/api/sync/ack", $apiKey, [
            'clientId' => $clientId,
            'cursor' => $payload['cursor'],
        ]);
        fwrite(STDOUT, 'Applicati ' . count($payload['events']) . " eventi ricevuti dal server\n");
    }
}

// Esempio: simula un cambiamento avvenuto nel dominio offline.
// Nella tua app reale questa chiamata va fatta subito dopo ogni
// scrittura locale rilevante.
enqueue_event($pdo, [
    'id' => bin2hex(random_bytes(16)),
    'type' => 'com.gabrieleromanato.offlineapp.record.created',
    'source' => "urn:client:{$clientId}",
    'time' => gmdate('c'),
    'data' => ['recordId' => 42, 'note' => 'Creato mentre offline'],
]);

while (true) {
    try {
        try_sync($pdo, $baseUrl, $apiKey, $clientId);
    } catch (Throwable $e) {
        // Rete assente o server irraggiungibile: comportamento atteso per un
        // client offline-first. Si riprova semplicemente al prossimo giro.
        fwrite(STDERR, "Sync non riuscita (rete assente?): {$e->getMessage()}\n");
    }

    sleep(30);
}

Provarlo end-to-end

cd online-php && API_KEY=dev-secret php -S 0.0.0.0:3000 index.php
# in un altro terminale
cd offline-php && API_KEY=dev-secret php sync.php

Al primo giro il client accoda l'evento di esempio, lo invia con push e lo riceve indietro con pull, applicandolo in modo idempotente grazie a INSERT OR IGNORE. Puoi verificarlo interrogando direttamente events.db con PDO o con il client sqlite3 da riga di comando: l'evento generato offline compare sul server con lo stesso id, e client_cursors registra il cursore avanzato a 1.

Cosa manca per la produzione

Rispetto a un'app costruita su un framework, qui mancano anche pezzi che di solito dai per scontati: gestione degli errori più strutturata (qui è tutto Throwable generico), validazione dell'input più rigorosa dei semplici controlli manuali, un vero autoloading invece di require espliciti man mano che il progetto cresce. Restano valide anche le lacune viste negli articoli precedenti: autenticazione più solida di una API key statica, retry con backoff esponenziale invece del sleep(30) fisso, un filtro lato client sugli eventi con source uguale al proprio, validazione dello schema di data per ogni type, e una strategia esplicita di conflict resolution per le modifiche concorrenti. In produzione, index.php andrebbe servito da PHP-FPM dietro Nginx (il server integrato è solo per lo sviluppo), e sync.php supervisionato da systemd o Supervisor invece che lasciato girare a mano in un terminale.

Nel prossimo articolo vediamo la stessa architettura in Java con Spring Boot.