Loading...
System requirements
Functional:
API design
Here's a breakdown of the essential REST
API
User service:
PUT /auth/v1/login - authenticate a user
{
"login": "user1"
"pass_hash": "90AB787887878F"
}
Response: HTTP 200 and JWT token or HTT P 403 if the credentials are invalid.
PUT /auth/v1/logout/ - log out user
{
"login": "user1"
}
Response: HTTP 200 if login and JWT are valid or HTTP 403 if JWT is invalid or absent in HTTP headers.
POST /auth/v1/register - register new user:
{
"name": "user1"
"email": "[email protected]"
"phone": "0283929382829"
"manager_id": "232323323"
}
Response: HTTP 201 and employee_id of the new user or HTTP 403 if the JWT is invalid or absent in HTTP headers.
GET /user/v1/manager/{employee_id}
Response: HTTP 200 and the manager information:
{
'employee_id': "28382329082"
'name': "user1"
'email': "[email protected]"
"phone": "928329329"
'manager_id': "92839238293"
}
Otherwise, HTTP 403 if JWT is absent in the HTTP header or invalid
Employee shift matching service:
PUT /shift_matching/v1/swap/{employee_id}/{srcshift_id}/{target_shift_id} - try to swap the the shift {src_shift_id} with the shift {target_shift_id} for employee {empoloyee_id}
Response: HTTP 200 and
{ "swap_id": "9829389293",
"status": "IN_PROGRESS"|
"APPROVED_BY_OTHER_EMPLOYEE"|"CANCELED"|"COMPLETED",
"createdAt": "01/22/2025:08:09:01"
}
Otherwise, HTTP 403 if the JWT is invalid or absent in the HTTP headers.
GET /shift_matching/v1/swap/{employee_id}/status/{swap_id} - Response HTTP 200 and the swap status:
{ "swap_id": "1233333"
"status": "IN_PROGRESS" | "COMPELETED"| "APPROVED_BY_EMPLOYEE" | "CANCELED"
"createdAt": "01/10/2025:09:10:23"
}
Otherwise, HTTP 403 if the JWT is invalid or absent in the HTTP headers.
GET /shift_matching/v1/shift/{employee_id}
Response: HTTP 200 and the shift list:
{
[
'shift_id': "1234444",
'employee_id': '23232323'
'start_time': "01/10/2025:09:10:23",
'end_time': "01/11/2025:09:10:23",
'last_swap_date': "01/11/2025:09:10:23",
'status': 'ACTIVE' | 'PENDING' | 'CANCELED'
]
}
Otherwise, HTTP 403 if the JWT is invalid or absent in the HTTP headers.
PUT /shift_matching/v1/swap/manager/{manager_id}/approve/{swap_id} - a manager approves the suggested shift if both employees agree.
Response: HTTP 200 or
Otherwise, HTTP 403 if the JWT is invalid or absent in the HTTP headers.
Employee Shift service:
GET /employee_shift/v1/shift/{employee_id}/shifts?startDate=01/10/2025:09:10:23&endDate=01/10/2025:09:10:23&offset=1&pageSize=10 - Response HTTP 200 and return the list of shifts belonging to the employee:
{ "employee_id":'12223232',
"shifts":[
{ "shift_id" : '2323232'
"startFrom": '01/02/2025:09:10:23',
"endTo": '01/10/2025:09:10:23',
'status': 'ACTIVE' | 'PENDING' | 'CANCELED',
'last_swap_date': '01/10/2025:09:10:23'
} ]
}
Otherwise, HTTP 403 if the JWT is invalid or absent in the HTTP headers.
POST /employee_shift/v1/shift/{employee_id}
Persist new shift for an employee.
body request:
{ 'shift_id' : '1121212121',
'employee_id': '22323232',
'start_date': '01/02/2025:09:10:23',
'end_date': '01/02/2025:09:10:23',
'status': 'ACTIVE' | 'PENDING' | 'CANCELED',
'last_swap_date': '01/02/2025:09:10:23'
}
Response HTTP 201.Otherwise, HTTP 403 if the JWT is invalid or absent in the HTTP headers.
PUT /employee_shift/v1/shift/{employee_id}
Update the shift information for an employee.
The body request:
{ 'shift_id' : '1121212121',
'employee_id': '22323232',
'start_date': '01/02/2025:09:10:23',
'end_date': '01/02/2025:09:10:23',
'status': 'ACTIVE' | 'PENDING' | 'CANCELED',
'last_swap_date': '01/02/2025:09:10:23'
}
Response HTTP 200. Otherwise, HTTP 403 if the JWT is invalid or absent in the HTTP headers.
Notification Service:
Uses the WebSocket to send updates to the client application.
PUT /notification/v1/notify
The request body:
{
'employee_id': '1121221212121',
'message': 'Please confirm the swap',
'url': 'http://blabla.com/callback',
'email': '[email protected]',
'phone': '131232323232',
'expired_at': '01/02/2025:09:10:23'
}
Response: HTTP 200. Otherwise, HTTP 403 if the JWT is invalid or absent in the HTTP headers.
The system is built around a microservices architecture to ensure separation of concerns, independent scalability, and resilience. Communication between services is a mix of synchronous (HTTP/REST, WebSocket) and asynchronous (Kafka) patterns.
Improvements:
UserService, ShiftMatchingService, etc.).POST /register: Register new employees.POST /login: Authenticate users (e.g., with username/password). Upon success, it generates and returns a signed JWT. The JWT should contain claims like user_id, role (e.g., employee, manager), and permissions.POST /logout (Optional): Invalidate the JWT on the server-side if using a blacklist (stored in Redis).GET /shifts/{id}: Get a specific shift.GET /employees/{id}/shifts: Get an employee's shift schedule, including the last 100 swaps (as per the original requirement).SHIFT_SWAP_APPROVED events from the Shift Matching Service and permanently updates the shift ownership in the PostgreSQL database. This service is the ultimate authority on "who is working when."users table: id, name, email, role, etc.shifts table: id, employee_id (foreign key), start_time, end_time, status (SCHEDULED, IN_PROGRESS, CANCELED).shift_swaps_history table: A log of all completed swaps for audit and display purposes.This is the most complex service. Its logic is refined for clarity and robustness.
GET /swap/options: Provides a list of potential shifts an employee can swap for.SCHEDULED shifts from other employees that do not cause scheduling conflicts (e.g., overlapping shifts). The simple "Hungarian algorithm" mention is replaced with a more practical conflict-detection query in the database. Results can be cached in Redis for a short period to reduce load.POST /swap/request: Initiates a swap request.{ "my_shift_id": "123", "requested_shift_id": "456" }my_shift_id and requested_shift_id as part of the key. This prevents race conditions for the same shifts.SCHEDULED and available.PENDING.employee_456).POST /swap/{id}/approve: Allows an employee to accept a swap proposed to them.ACCEPTED_BY_EMPLOYEE.POST /swap/{id}/manager-approve: Manager gives final approval.APPROVED.SHIFT_SWAP_APPROVED containing the details of the swap (the two shift IDs and the two employee IDs).POST /swap/{id}/cancel: Allows either involved employee or a manager to cancel a swap at any state (PENDING, ACCEPTED_BY_EMPLOYEE). Status becomes CANCELED.swap_proposals table: id, requesting_employee_id, requesting_shift_id, receiving_employee_id, receiving_shift_id, status, created_at.SWAP_REQUESTED, SWAP_ACCEPTED, SWAP_APPROVED, SWAP_CANCELED.user_id -> WebSocket session).shift_swap_requestsshift_swap_approvals (consumed by the Employee Shift Service)notification_events (consumed by the Notification Service)lock:shift:{shift_id}.This improved design is more robust, scalable, and clearly defines the data flow and responsibilities of each component.
# Kubernetes Configuration for API Gateway
apiVersion: apps/v1
kind: Deployment
metadata:
name: api-gateway
spec:
replicas: 3
template:
spec:
containers:
- name: gateway
image: nginx-plus/istio-proxy
env:
- name: SHARDING_KEY
value: "employee_id"
Improvements:
Critical Fixes:
class ShiftMatchingService:
def get_swap_recommendations(self, employee_id: str) -> List[Shift]:
"""Get personalized shift swap recommendations"""
# 1. Get employee's upcoming shifts
my_shifts = self.shift_service.get_upcoming_shifts(employee_id)
# 2. Find compatible employees (same manager, similar role, location)
compatible_employees = self.find_compatible_employees(employee_id)
# 3. Get available shifts from compatible employees
candidate_shifts = self.get_available_shifts(
compatible_employees,
exclude_conflicts_with=my_shifts
)
# 4. Score and rank shifts based on multiple factors
scored_shifts = self.score_shifts(my_shifts, candidate_shifts)
return scored_shifts[:20] # Return top 20 recommendations
def score_shifts(self, my_shifts: List, candidate_shifts: List) -> List[ScoredShift]:
"""Score shifts based on multiple factors"""
scores = []
for candidate in candidate_shifts:
score = self.calculate_match_score(my_shifts, candidate)
scores.append(ScoredShift(shift=candidate, score=score))
return sorted(scores, key=lambda x: x.score, reverse=True)
def calculate_match_score(self, my_shifts: List, candidate: Shift) -> float:
"""Calculate match score using practical business rules"""
score = 0.0
# Time proximity (prefer shifts close to original)
time_diff = abs(candidate.start_time - my_shifts[0].start_time)
score += max(0, 1 - (time_diff.total_hours() / 168)) # Normalize to 1 week
# Duration match
if candidate.duration == my_shifts[0].duration:
score += 0.3
# Skill compatibility
if self.has_required_skills(candidate.required_skills):
score += 0.2
# Location preference
if candidate.location == my_shifts[0].location:
score += 0.2
return min(1.0, score) # Cap at 1.0
State Transitions:
-- PostgreSQL table for swap proposals
CREATE TABLE shift_swaps (
id UUID PRIMARY KEY,
requesting_employee_id UUID,
receiving_employee_id UUID,
requested_shift_id UUID,
offered_shift_id UUID,
status SWAP_STATUS, -- 'PENDING', 'ACCEPTED', 'MANAGER_REVIEW', 'COMPLETED', 'CANCELLED'
created_at TIMESTAMPTZ,
updated_at TIMESTAMPTZ,
expires_at TIMESTAMPTZ, -- Redis lock TTL
manager_id UUID
);
class ShiftSwapOrchestrator:
def init(self, redis_client, db_session):
self.redis = redis_client
self.db = db_session
def request_swap(self, request: SwapRequest) -> SwapResult:
# Create lock keys for both shifts
lock_key_1 = f"lock:shift:{request.requested_shift_id}"
lock_key_2 = f"lock:shift:{request.offered_shift_id}"
try:
# Acquire distributed locks with retry logic
acquired = self.redis.set(
lock_key_1, request.requesting_employee_id,
nx=True, ex=3600 # 1 hour TTL
)
if not acquired:
return SwapResult(error="Shift already locked")
# Create swap proposal in database (source of truth)
swap = ShiftSwap(
id=uuid.uuid4(),
requesting_employee_id=request.requesting_employee_id,
requested_shift_id=request.requested_shift_id,
offered_shift_id=request.offered_shift_id,
status='PENDING'
)
self.db.add(swap)
self.db.commit()
# Send notification
self.notification_service.notify_swap_request(swap)
return SwapResult(success=True, swap_id=swap.id)
except Exception as e:
# Release locks on failure
self.redis.delete(lock_key_1)
self.redis.delete(lock_key_2)
raise e
AWS ElastiCache Redis Cluster configuration
ClusterMode: enabled
NumNodeGroups: 3
ReplicasPerNodeGroup: 1
AutoFailover: enabled
DataTiering: false
Key sharding strategy
ShardingKey: "employee_id" # Consistent hashing for user data
Redis Data Structure:
Shift cache (TTL: 5 minutes)
shift_key = f"shift:{shift_id}"
redis.hset(shift_key, mapping={
"employee_id": "emp_123",
"start_time": "2024-01-15T09:00:00Z",
"end_time": "2024-01-15T17:00:00Z",
"status": "SCHEDULED"
})
redis.expire(shift_key, 300)
User session store
session_key = f"session:{user_id}"
redis.setex(session_key, 3600, jwt_token)
Rate limiting
rate_limit_key = f"rate_limit:{user_id}:{action}"
redis.incr(rate_limit_key)
redis.expire(rate_limit_key, 60)
class ShiftSwapClient {
private retryConfig = {
maxRetries: 3,
baseDelay: 1000,
maxDelay: 10000
};
async requestShiftSwap(swapRequest: SwapRequest): Promise<SwapResponse> {
return this.retryableRequest(
() => this.apiClient.post('/swap/request', swapRequest),
this.retryConfig
);
}
private async retryableRequest(
requestFn: () => Promise<any>,
config: RetryConfig
): Promise<any> {
let lastError: Error;
for (let attempt = 0; attempt < config.maxRetries; attempt++) {
try {
return await requestFn();
} catch (error) {
lastError = error;
if (this.isRetryableError(error)) {
const delay = Math.min(
config.baseDelay * Math.pow(2, attempt),
config.maxDelay
);
await this.sleep(delay + Math.random() * 1000); // Jitter
continue;
}
throw error;
}
}
throw lastError;
}
// WebSocket reconnection with exponential backoff
setupWebSocket() {
this.ws = new WebSocket(this.wsUrl);
this.ws.onclose = () => {
const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
setTimeout(() => this.setupWebSocket(), delay);
};
}
}
class NotificationService:
def init(self, redis_client, ws_manager, email_client, sms_client):
self.redis = redis_client
self.ws_manager = ws_manager
self.email_client = email_client
self.sms_client = sms_client
async def notify_swap_request(self, swap: ShiftSwap):
# 1. Immediate WebSocket notification
await self.ws_manager.send_to_user(
swap.receiving_employee_id,
{
"type": "SWAP_REQUEST",
"swap_id": str(swap.id),
"message": "You have a new shift swap request"
}
)
# 2. Fallback to email (async, retryable)
asyncio.create_task(self.send_email_notification(swap))
# 3. Optional SMS for urgent notifications
if self.is_urgent(swap):
asyncio.create_task(self.send_sms_notification(swap))
async def send_email_notification(self, swap: ShiftSwap):
"""Send email with retry logic"""
max_retries = 3
for attempt in range(max_retries):
try:
await self.email_client.send(
to=swap.receiving_employee.email,
subject="Shift Swap Request",
template="swap_request.html",
context={"swap": swap}
)
break
except SMTPException as e:
if attempt == max_retries - 1:
await self.log_failed_notification(swap, str(e))
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: shift-matching-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: shift-matching-service
minReplicas: 3
maxReplicas: 20
metrics:
type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
type: Pods
pods:
metric:
name: requests_per_second
target:
type: AverageValue
averageValue: 100
behavior:
scaleDown:
stabilizationWindowSeconds: 300
policies:
type: Percent
value: 50
periodSeconds: 60
class SMTPClientWithCircuitBreaker:
def init(self):
self.failure_count = 0
self.circuit_state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
self.last_failure_time = None
async def send_email(self, email: Email) -> bool:
if self.circuit_state == "OPEN":
if time.time() - self.last_failure_time > 60: # 60 second timeout
self.circuit_state = "HALF_OPEN"
else:
raise CircuitBreakerOpenError("SMTP service unavailable")
try:
result = await self._actual_send(email)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise e
def _on_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= 5:
self.circuit_state = "OPEN"
def _on_success(self):
self.failure_count = 0
self.circuit_state = "CLOSED"
User service:
Uses Postgres to hold the user information because we need to support strong consistency in SQL storage:
The user table:
employee_id : the unique key string 128 bytes
name: string 128 bytes
email: string 256 bytes
role: string 'employee' or 'manager'
manager_id: string 128 bytes
createAt: timestamp 4 bytes
loggedIn: boolean 1 bytes
LastLogin: timestamp 4 bytes
The employee shift service:
Utilizes Postgres to store the shift information.
The shift table:
shift_id: string 128 bytes, the unique key
the employee_id: string 128 bytes
start_date: timestamp
end_date: timestamp
status: byte
The swap_history table:
swap_id: string 128 bytes
src_emploee_id1: string 128 bytes
src_shift_id: string
target_ employee_id: string 128 bytes
target_shift_id: string 128 bytes
swapped_at: timestamp 4 bytes
Capacity estimation:
So it takes 2 kbytes to store all tables, and it takes 80 megabytes for 1 year, it takes 1 gigabyte for 5 years. (We're going to hold one replication copy)
Choose to share one swap matrix between all the Employee shift matching service instances, whereas it's better to create a structure to partition that by shift_id between the
Employee shift matching service instances.
Need to invent the mechanism to restore the last swap shift information from the client application if the Employee shift matching service is down and loses the information, now it ignores that moment.
The system would be deployed to AWS EKS and the service container would be orchestrated by Kubernetes.
If client applications request the swap history information frequently it gives the high load to Postgres and it may be cached.
If many employees would request changes of the swap shift intensively, it causes the race condition in Redis, so it's better to shard this process.
What are some future improvements you would make? How would you mitigate the failure scenario(s) you described above?
we improve this service in future by spending time on choose other object store.