Series Laravel & Elasticsearch Thực chiến #5: Đồng bộ Thời gian thực – Khi MySQL và Quái vật chung nhịp đập

1. Phân tích Kiến trúc: Đừng bao giờ bắt API chờ đợi

Cách "ngây thơ" nhất mà nhiều người mới làm là chèn thẳng đoạn code gọi API của Elasticsearch vào ngay bên trong Controller cập nhật giao dịch.

Tại sao cách này là thảm họa? Nếu cụm Elasticsearch đang bị nghẽn mạng và mất 2 giây để phản hồi, API cập nhật trạng thái thẻ của bạn cũng sẽ bị treo đúng 2 giây. Hành khách sẽ phải đứng chờ ở cổng quay (turnstile) chỉ vì hệ thống bận... đi đồng bộ log.

Kiến trúc Enterprise chuẩn mực:

1. Lắng nghe (Observer): Bắt sự kiện khi Model Transaction trong MySQL bị thêm/sửa/xóa.
2. Đẩy vào hàng đợi (Queue): Ném ngay tác vụ đồng bộ vào Redis (chúng ta đã dựng container Queue Worker ở series Docker trước đây) và trả kết quả cho người dùng ngay lập tức.
3. Xử lý ngầm (Worker): Queue Worker ở background sẽ âm thầm lấy tác vụ ra và gọi API của Elasticsearch để cập nhật (Update) hoặc xóa (Delete) document.

2. Bước 1: Tạo Job đồng bộ ngầm (The Worker)

Mở Terminal và tạo một Job mới:

d-artisan make:job SyncTransactionToElastic

Mở file app/Jobs/SyncTransactionToElastic.php. Chúng ta sẽ truyền dữ liệu giao dịch và hành động (save hoặc delete) vào Job này.

namespace App\Jobs;

use Elastic\Elasticsearch\Client;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;

class SyncTransactionToElastic implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $transactionData;
    protected $action;

    public function __construct(array $transactionData, string $action = 'save')
    {
        $this->transactionData = $transactionData;
        $this->action = $action; // 'save' (cho create/update) hoặc 'delete'
    }

    public function handle(Client $elasticsearch)
    {
        $indexName = 'afc_transactions';
        $id = $this->transactionData['id'];

        try {
            if ($this->action === 'delete') {
                // Lệnh XÓA document khỏi ES
                $elasticsearch->delete([
                    'index' => $indexName,
                    'id'    => $id,
                    'client' => ['ignore' => 404] // Bỏ qua lỗi nếu document không tồn tại
                ]);
            } else {
                // Lệnh UPSERT (Update nếu có, Insert nếu chưa)
                $elasticsearch->index([
                    'index' => $indexName,
                    'id'    => $id,
                    'body'  => [
                        'transaction_id' => $this->transactionData['transaction_code'],
                        'card_id'        => $this->transactionData['card_id'],
                        'station_id'     => $this->transactionData['station_id'],
                        'amount'         => (int) $this->transactionData['amount'],
                        'status'         => $this->transactionData['status'],
                        'tapped_at'      => $this->transactionData['created_at'],
                        'error_notes'    => $this->transactionData['error_message'] ?? ''
                    ]
                ]);
            }
        } catch (\Exception $e) {
            // Ném lỗi để Laravel Queue tự động thử lại (Retry)
            throw $e;
        }
    }
}

Lưu ý: Hàm index trong Elastic PHP Client có tính chất "Upsert". Nếu cái id đó chưa có, nó sẽ tạo mới (Insert). Nếu có rồi, nó sẽ đè lên toàn bộ cục data cũ (Update). Rất tiện lợi!

3. Bước 2: Dựng Tai mắt (Model Observer)

Bây giờ, chúng ta cần một "camera giám sát" để tự động kích hoạt cái Job trên mỗi khi có người chạm vào bảng transactions trong Database.

Tạo Observer:

d-artisan make:observer TransactionObserver --model=Transaction

Mở file app/Observers/TransactionObserver.php và nhúng cái Job của chúng ta vào:

namespace App\Observers;

use App\Models\Transaction;
use App\Jobs\SyncTransactionToElastic;

class TransactionObserver
{
    // Chạy khi một giao dịch mới được tạo thành công
    public function created(Transaction $transaction)
    {
        SyncTransactionToElastic::dispatch($transaction->toArray(), 'save');
    }

    // Chạy khi trạng thái hoặc bất kỳ trường nào của giao dịch bị sửa đổi
    public function updated(Transaction $transaction)
    {
        SyncTransactionToElastic::dispatch($transaction->toArray(), 'save');
    }

    // Chạy khi một giao dịch bị xóa khỏi MySQL
    public function deleted(Transaction $transaction)
    {
        // Chỉ gửi ID để Job biết mà xóa bên ES
        SyncTransactionToElastic::dispatch(['id' => $transaction->id], 'delete');
    }
}

4. Bước 3: Đăng ký Observer vào hệ thống

Để Laravel biết sự tồn tại của cái "camera" này, bạn phải đăng ký nó. Mở file app/Providers/EventServiceProvider.php (hoặc AppServiceProvider.php nếu dùng Laravel 11) và thêm vào hàm boot():

use App\Models\Transaction;
use App\Observers\TransactionObserver;

public function boot()
{
    // Gắn Observer vào Model
    Transaction::observe(TransactionObserver::class);
}

5. Tận hưởng thành quả tự động hóa

Từ giờ phút này trở đi, bạn (và bất kỳ đồng nghiệp nào trong team) có thể hoàn toàn quên đi sự tồn tại của Elasticsearch khi viết logic nghiệp vụ thông thường.

Khi bạn chạy một đoạn code như thế này ở bất kỳ đâu (API, Command, Tinker):

$txn = Transaction::find(1);
$txn->status = 'success';
$txn->save(); // Boom!

Ngay lập tức, MySQL lưu dữ liệu mới. Observer updated chớp lấy khoảnh khắc đó, ném data vào hàng đợi Redis (mất 0.001 giây). Ứng dụng đi tiếp một cách mượt mà. Ở phía sau, container queue-worker của bạn thò tay nhặt Job lên và gõ cửa Elasticsearch để cập nhật lại document số 1 thành success. Mọi thứ diễn ra theo thời gian thực (Real-time) và hoàn toàn bất đồng bộ (Asynchronous).

Đến đây, bạn đã sở hữu một luồng lưu chuyển dữ liệu hoàn hảo, không còn lo báo cáo bị lệch hay hệ thống bị nghẽn.

Kiến trúc Model Observer + Redis Queue này chạy rất tốt cho các hệ thống cỡ vừa và lớn. Tuy nhiên, nếu hệ thống Metro sau này mở rộng ra toàn quốc, việc dùng code PHP để bơm dữ liệu có thể không còn là lựa chọn chịu tải tốt nhất. Ở bài học tiếp theo (Trạm cuối), bạn muốn chúng ta khám phá cách dùng siêu công cụ Logstash (để đồng bộ tự động mà không cần viết 1 dòng code nào), hay muốn học cách cắm Kibana vào để vẽ Dashboard biểu đồ trực quan (như Grafana) không?