1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397"""Queue service for Redis operations."""
import json
import logging
from datetime import UTC, datetime
from typing import Any
from pydantic import BaseModel
from redis.asyncio import Redis
from ..config import settings
from ..shared.constants.redis_keys import timeout_key
logger = logging.getLogger(__name__)
class QueueService:
"""Service for managing Redis queue operations."""
def __init__(self, redis_client: Redis) -> None:
"""Initialize queue service with Redis client.
Args:
redis_client: Redis async client instance
"""
self.redis = redis_client
self.pii_queue = settings.pii_queue_name
async def queue_pii_job(self, job_id: str, s3_key: str) -> None:
"""
Queue a job for PII scanning.
Args:
job_id: Unique job identifier
s3_key: S3 key where document is stored
"""
payload = {
"job_id": job_id,
"s3_key": s3_key,
"created_at": datetime.now(UTC).isoformat()
}
# Push to queue
await self.redis.lpush(self.pii_queue, json.dumps(payload))
async def check_queue_depth(self) -> int:
"""
Check the current queue depth.
Returns:
Number of items in the PII queue
"""
try:
result = await self.redis.llen(self.pii_queue)
return int(result) if result is not None else 0
except Exception:
return -1
async def check_redis_connection(self) -> bool:
"""
Check if Redis is accessible.
Returns:
True if Redis is accessible, False otherwise
"""
try:
await self.redis.ping()
return True
except Exception:
return False
async def enqueue(self, queue_name: str, payload: BaseModel) -> None:
"""
Add job to specified queue.
Args:
queue_name: Name of the Redis queue
payload: Pydantic model instance to enqueue
Raises:
Exception: If enqueue operation fails
"""
try:
# Serialize Pydantic model to JSON
payload_json = payload.model_dump_json()
# Push to Redis list (LPUSH adds to head)
await self.redis.lpush(queue_name, payload_json)
except Exception as e:
raise Exception(f"Failed to enqueue job to {queue_name}: {str(e)}")
async def dequeue(
self,
queue_name: str,
timeout: int = 5,
model_class: type[BaseModel] | None = None
) -> dict[str, Any] | None:
"""
Pop job from queue with blocking timeout.
Args:
queue_name: Name of the Redis queue
timeout: Blocking timeout in seconds (default: 5)
model_class: Optional Pydantic model class for deserialization
Returns:
Deserialized payload as dict, or None if timeout
Raises:
Exception: If dequeue operation fails
"""
try:
# BRPOP blocks until item available or timeout
result = await self.redis.brpop(queue_name, timeout=timeout)
if result is None:
return None
# result is tuple: (queue_name, value)
_, payload_json = result
# Deserialize JSON
payload_dict: dict[str, Any] = json.loads(payload_json)
# Optionally validate with Pydantic model
if model_class:
validated = model_class(**payload_dict)
return validated.model_dump()
return payload_dict
except json.JSONDecodeError as e:
raise Exception(f"Failed to deserialize queue payload: {str(e)}")
except Exception as e:
raise Exception(f"Failed to dequeue job from {queue_name}: {str(e)}")
async def queue_depth(self, queue_name: str) -> int:
"""
Get current queue depth.
Args:
queue_name: Name of the Redis queue
Returns:
Number of items in queue, or -1 on error
"""
try:
result = await self.redis.llen(queue_name)
return int(result) if result is not None else 0
except Exception:
return -1
async def peek_queue(
self,
queue_name: str,
count: int = 10
) -> list[dict[str, Any]]:
"""
View queued jobs without removing them.
Args:
queue_name: Name of the Redis queue
count: Maximum number of items to peek (default: 10)
Returns:
List of deserialized payloads
Raises:
Exception: If peek operation fails
"""
try:
# LRANGE gets items without removing (0 is start, -1 would be end)
items = await self.redis.lrange(queue_name, 0, count - 1)
# Deserialize each item
payloads: list[dict[str, Any]] = []
for item in items:
try:
payload: dict[str, Any] = json.loads(item)
payloads.append(payload)
except json.JSONDecodeError:
# Skip invalid items
continue
return payloads
except Exception as e:
raise Exception(f"Failed to peek queue {queue_name}: {str(e)}")
async def health_check(self) -> bool:
"""
Check Redis connectivity.
Returns:
True if Redis is healthy, False otherwise
"""
return await self.check_redis_connection()
async def add_to_timeout_tracking(
self,
job_id: str,
expires_at: datetime,
timeout_type: str = "approval"
) -> None:
"""Add job to timeout tracking sorted set.
Uses Redis sorted set (ZADD) where:
- Score: Unix timestamp of expiration deadline
- Member: job_id
This allows efficient range queries to find expired jobs using ZRANGEBYSCORE.
Args:
job_id: Job identifier (UUID)
expires_at: Datetime when approval expires
timeout_type: Type of timeout (default: "approval")
Raises:
Exception: If Redis operation fails
Example:
>>> queue = QueueService(redis_client)
>>> expires = datetime.now(timezone.utc) + timedelta(hours=4)
>>> await queue.add_to_timeout_tracking("abc-123", expires)
"""
try:
# Convert datetime to Unix timestamp
timestamp = expires_at.timestamp()
# Get Redis key for this timeout type
key = timeout_key(timeout_type)
# ZADD adds to sorted set: ZADD key score member
await self.redis.zadd(key, {job_id: timestamp})
logger.debug(
f"Added job {job_id} to timeout tracking "
f"(expires at {expires_at.isoformat()})"
)
except Exception as e:
logger.error(
f"Failed to add job {job_id} to timeout tracking: {str(e)}",
exc_info=True
)
raise Exception(f"Failed to add timeout tracking: {str(e)}")
async def get_expired_timeouts(
self,
timeout_type: str = "approval"
) -> list[tuple[str, float]]:
"""Get jobs with expired approval deadlines.
Queries Redis sorted set for members with scores (timestamps) less than
or equal to current time (calculated internally). Returns job IDs and their
expiration timestamps.
Args:
timeout_type: Type of timeout to check (default: "approval").
Valid values: "approval", "processing", etc.
DO NOT pass a timestamp - this method calculates current time internally.
Returns:
List of tuples: [(job_id, expiration_timestamp), ...]
Empty list if no expired jobs or on error
Raises:
TypeError: If timeout_type is not a string
Example:
>>> queue = QueueService(redis_client)
>>> expired = await queue.get_expired_timeouts()
>>> for job_id, timestamp in expired:
... print(f"Job {job_id} expired at {datetime.fromtimestamp(timestamp)}")
"""
# Type validation - prevent common mistake of passing timestamp
if not isinstance(timeout_type, str):
raise TypeError(
f"timeout_type must be a string (e.g., 'approval'), "
f"got {type(timeout_type).__name__}. "
f"Do not pass a timestamp - this method calculates current time internally."
)
try:
# Get current timestamp
current_time = datetime.now(UTC).timestamp()
# Get Redis key for this timeout type
key = timeout_key(timeout_type)
# ZRANGEBYSCORE returns members with score in range [0, current_time]
# withscores=True returns (member, score) tuples
result = await self.redis.zrangebyscore(
key,
min=0,
max=current_time,
withscores=True
)
# Convert result to list of tuples (job_id, timestamp)
expired_jobs = []
for job_id, timestamp in result:
# Redis returns bytes, decode to string
if isinstance(job_id, bytes):
job_id = job_id.decode('utf-8')
expired_jobs.append((job_id, float(timestamp)))
if expired_jobs:
logger.info(
f"Found {len(expired_jobs)} expired {timeout_type} timeouts"
)
return expired_jobs
except Exception as e:
logger.error(
f"Failed to get expired {timeout_type} timeouts: {str(e)}",
exc_info=True
)
# Return empty list on error (fail-safe)
return []
async def remove_from_timeout_tracking(
self,
job_id: str,
timeout_type: str = "approval"
) -> bool:
"""Remove job from timeout tracking sorted set.
Called when:
- Job is approved/denied (no longer needs timeout tracking)
- Timeout has been processed
- Job is completed or failed
Args:
job_id: Job identifier (UUID)
timeout_type: Type of timeout (default: "approval")
Returns:
bool: True if job was removed, False if job wasn't in set
Example:
>>> queue = QueueService(redis_client)
>>> removed = await queue.remove_from_timeout_tracking("abc-123")
>>> if removed:
... print("Job removed from timeout tracking")
"""
try:
# Get Redis key for this timeout type
key = timeout_key(timeout_type)
# ZREM removes member from sorted set
# Returns number of members removed (0 or 1)
count = await self.redis.zrem(key, job_id)
if count > 0:
logger.debug(
f"Removed job {job_id} from {timeout_type} timeout tracking"
)
return True
else:
logger.debug(
f"Job {job_id} was not in {timeout_type} timeout tracking"
)
return False
except Exception as e:
logger.error(
f"Failed to remove job {job_id} from timeout tracking: {str(e)}",
exc_info=True
)
# Return False on error (idempotent - job effectively not tracked)
return False
async def get_timeout_count(self, timeout_type: str = "approval") -> int:
"""Get count of jobs currently in timeout tracking.
Args:
timeout_type: Type of timeout (default: "approval")
Returns:
int: Number of jobs in timeout tracking, or -1 on error
Example:
>>> queue = QueueService(redis_client)
>>> count = await queue.get_timeout_count()
>>> print(f"{count} jobs awaiting approval")
"""
try:
key = timeout_key(timeout_type)
result = await self.redis.zcard(key)
return int(result) if result is not None else 0
except Exception as e:
logger.error(
f"Failed to get timeout count: {str(e)}",
exc_info=True
)
return -1