Command Query Responsibility Segregation is a design pattern that separates the operations of reading data (queries) from updating data (commands). In CQRS, commands (create, update, delete) modify the system’s state, while queries only retrieve data without altering it. This separation allows for optimized data models for reads and writes, improves performance, scalability, and security, and enables different data storage mechanisms or representations for the read and write sides.
Press enter or click to view image in full size
Basic CQRS Kafka Diagram
We will use our microservices example application from GitHub to explain CQRS, starting with how the profile/details preview is currently implemented, and then demonstrating how it can be enhanced using the CQRS pattern.
Microservices Setup:
Authentication Service: Manages user authentication and stores basic user information (e.g., username, email).
Authorization Service: Manages user roles and permissions.
Profile Service: Stores detailed user information (e.g., address, company).
Gateway Service: Acts as the entry point for client requests, coordinating between the microservices.
The following is the structure of the current and expected API response — it can be found in Postman collection:
The gateway receives this request, checks authentication and authorization.
If the checks pass, the gateway forwards the request to the profile service.
Code References
ProfileController.php
<?php
namespace App\Http\Controllers;
use App\Http\Requests\ProfileRequest; use App\Http\Resources\ProfileResource; use App\Services\ProfileService; use Illuminate\Http\Request;
class ProfileController extends Controller { public function __construct(protected readonly ProfileService $profileService) {}
public function profile(ProfileRequest $request) { return new ProfileResource($this->profileService->getProfile(auth()->user())); }
public function getProfileById(ProfileRequest $request, int $userId) { $response = $this->profileService->getUserByToken($request->bearerToken(), $userId); $userData = json_decode($response->getBody(), true)['data'];
public function admin(Request $request): string { return 'You can access this!'; } }
ProfileService.php
<?php
namespace App\Services;
use App\Models\Profile; use App\Models\User; use GuzzleHttp\Client;
class ProfileService { protected $authServiceUrl;
public function __construct(private readonly Client $client) { $this->authServiceUrl = config('services.micro-services.authentication'); // URL of the User Authentication Service }
public function getUserByToken(string $token, int $userId) { return $this->client->post("{$this->authServiceUrl}/get-user-by-id-and-verify-jwt", [ 'json' => ['token' => $token, 'user_id' => $userId], ]); }
public function getProfile(User $user): Profile { return $user->profile; }
public function getProfileById(int $userId): Profile { return User::where('user_id', $userId)->firstOrFail()->profile; } }
protected function message() { return 'Profile data retrieved successfully.'; } }
2024_05_30_121259_create_profiles_table.php
<?php
use Illuminate\Database\Migrations\Migration; use Illuminate\Database\Schema\Blueprint; use Illuminate\Support\Facades\Schema;
return new class extends Migration { /** * Run the migrations. */ public function up(): void { Schema::create('profiles', function (Blueprint $table) { $table->id(); $table->foreignId('user_id')->nullable()->constrained('users')->onDelete('set null'); $table->string('address', 255)->nullable(); $table->string('phone_number', 35)->nullable(); $table->date('birthdate')->nullable(); $table->string('profile_picture', 255)->nullable(); $table->text('bio')->nullable(); $table->timestamps(); }); }
/** * Reverse the migrations. */ public function down(): void { Schema::dropIfExists('profiles'); } };
AuthController.php
<?php
namespace App\Http\Controllers;
use App\Http\Requests\GetUserByIdAndVerifyJWTToken; use App\Http\Requests\PasswordRecoveryRequest; use App\Http\Requests\PasswordResetRequest; use App\Http\Requests\PasswordResetWithTokenRequest; use App\Http\Requests\UserLoginRequest; use App\Http\Requests\UserRegisterRequest; use App\Http\Requests\VerifyTokenRequest; use App\Http\Resources\PasswordRecoveryEmailSentResource; use App\Http\Resources\PasswordRecoveryResource; use App\Http\Resources\PasswordRecoverySuccessResource; use App\Http\Resources\UserLoginResource; use App\Http\Resources\UserResource; use App\Models\User; use App\Services\AuthService; use Illuminate\Http\Request; use Illuminate\Support\Facades\Auth;
class AuthController extends Controller { protected $authService;
public function __construct(AuthService $authService) { $this->authService = $authService; }
...
public function getUserByIdAndVerifyJWTRequest(GetUserByIdAndVerifyJWTToken $request) { return new UserResource($this->authService->getUserByIdAndVerifyJWTRequest($request->user_id)); }
... }
AuthService.php
<?php
namespace App\Services;
use App\Mail\PasswordReset; use App\Models\User; use Exception; use Illuminate\Auth\AuthenticationException; use Illuminate\Database\Eloquent\ModelNotFoundException; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Hash; use Illuminate\Support\Facades\Log; use Illuminate\Support\Facades\Mail; use Tymon\JWTAuth\Facades\JWTAuth;
class AuthService { ...
public function getUserByIdAndVerifyJWTRequest(int $user_id): User { $user = User::find($user_id);
if (! $user) { throw new ModelNotFoundException('User not found', 404); }
// Get user by ID and verify JWT return $user; } ... }
0001_01_01_000000_create_users_table.php
<?php
use Illuminate\Database\Migrations\Migration; use Illuminate\Database\Schema\Blueprint; use Illuminate\Support\Facades\Schema;
return new class extends Migration { /** * Run the migrations. */ public function up(): void { Schema::create('users', function (Blueprint $table) { $table->id(); $table->string('username'); $table->string('full_name'); $table->string('email')->unique(); $table->timestamp('email_verified_at')->nullable(); $table->string('password_hash'); $table->rememberToken(); $table->boolean('is_admin')->default(false); $table->timestamps(); });
Data Retrieval Process
Get Milos Kecman’s stories in your inbox
Join Medium for free to get updates from this writer.
To fetch all the required data for a specified user, the profile service takes the user_id (sent as a GET parameter) from the request. It then performs the following actions:
Collects basic user data from the authentication service (e.g., email, whether the user is an admin).
Fetches additional details (address, phone number, birthday, profile picture, etc.) from the profiles table.
The profile service communicates with the authorization service to get the roles for the specified user.
After collecting the combined data, the profile service sends it back to the gateway, which then forwards the data to the user.
CQRS Implementation Workflow
We will separate the reading of profile data into different services and use Kafka for event-based communication between the services
Request flow:
User Request via Gateway: ◦ A user makes a request to view their profile by sending a request to the Gateway Service.
Query Dispatch to Kafka: ◦ The Gateway Service does not directly request all the data from different services. Instead, it sends a query message to a Kafka topic (e.g., user-profile-query). ◦ The query contains the user’s user ID or another unique identifier.
Handling Queries in Each Microservice: ◦ Authentication Service subscribes to the user-profile-query topic, listens for query requests, and retrieves the basic user information (username, email) from its database. It then publishes this data back to a Kafka response topic (e.g., user-profile-response). ◦ Authorization Service subscribes to the same user-profile-query topic, retrieves the user’s roles and permissions, and publishes the result back to the user-profile-response topic. ◦ Profile Service also subscribes to the user-profile-query topic, retrieves the detailed profile information (address, company), and publishes this data to the user-profile-response topic.
Aggregating Data in Gateway: ◦ The Gateway Service subscribes to the user-profile-response topic and listens for the responses from the three services (Authentication, Authorization, and Profile). ◦ Once it has collected all necessary data from the services, the Gateway Service aggregates the information and returns the complete profile to the client.
Step 1: User Sends a Request • Client makes a GET request to GET /profile to the Gateway Service.
Step 2: Gateway Publishes a Kafka Query • Gateway Service publishes this message to Kafka topic user-profile-query.
Step 3: Services Handle the Query • Authentication Service: ◦ Listens to the user-profile-query topic, extracts userId, retrieves basic user data, and responds:
Profile Service: ◦ Listens to the same user-profile-query, retrieves detailed profile info, and responds:
{ "userId": "123456", "service": "Profile", "data": { "address": "123 Main St, City, Country", "company": "Tech Solutions Inc." }, "forRequestId":"0000001" }
◦ Publishes the result to user-profile-response.
Step 4: Gateway Aggregates Responses • The Gateway Service listens to the user-profile-response topic and waits for all services (Authentication, Authorization, Profile) to respond. • It aggregates all the data and forms the complete profile:
Finally, the Gateway sends this aggregated data back to the client as the profile response.
Benefits of Using CQRS and Kafka:
Separation of Concerns: By separating the write and read models, the query model is optimized for reads, which allows more flexibility and scalability.
Asynchronous Communication: Kafka enables loose coupling between services. Each service independently processes and responds to requests at its own pace.
Scalability: Each service (Authentication, Authorization, Profile) can scale independently to handle more requests.
Resilience: If one service is delayed or fails, the system can still function by handling other parts of the profile query.
Potential Issues and Considerations:
Timeouts: You may need to handle situations where one of the services takes too long to respond or fails to respond at all. You can implement a timeout mechanism in the Gateway Service to handle this.
Partial Data: If one service fails, you may return partial data to the client or decide to return an error, depending on your system’s requirements.
Data Consistency: Since this is an asynchronous and distributed approach, you may deal with eventual consistency, especially if one of the services updates the data after the request has been processed.
This CQRS with Kafka architecture allows for a scalable, resilient system where each microservice handles its responsibility independently and asynchronously.
