Payment Processor

A multi-threaded payment processing system built in Python. The system includes:

• In-process pub/sub event bus with per-subscriber fan-out queues
• Priority queue (min-heap) for ordered payment processing
• Payment publisher that generates random payments with location and type data
• Fraud detection that scores risk and flags high-risk payments
• Payment processor that approves or rejects payments based on fraud risk
• Interactive dashboard for searching and inspecting processed transactions in real time

Requirements

• Python 3.13+

Starting the application

Run using uv:

uv run start

On startup the application:

1. Starts the publisher, which begins generating payments immediately
2. Starts the fraud detector, which begins scoring payments as they arrive
3. Counts down 10 seconds (driven by _MIN_BACKLOG) while showing the live queue depth
4. Starts the processor, dashboard watcher, and store service
5. Drops into an interactive REPL with dashboard and Decimal pre-bound — all services keep running in daemon threads behind it

Press Ctrl+D or Ctrl+C to shut everything down gracefully.

Architecture

PaymentPublisherService
        │  publish("payment.pending")     random payment: user, amount, type, location
        ▼
    EventBus  ──────────────────────────────────────────────────────────────┐
        │  subscribe("payment.pending", FIFO)                               │  subscribe("payment.pending", priority=True)
        ▼                                                                   ▼
FraudDetectionService                                          PaymentProcessorService
  - scores risk (type + location)                               - waits for backlog ≥ 10
  - sets payment.risk in-place                                  - high risk  → status = "failed"
  - writes high-risk to fraud.jsonl                             - other risk → status = "completed"
                                                                │  publish("payment.processed")
                                                                ▼
                                                            EventBus  ──────────────────────┐
                                                                │  subscribe("payment.processed")
                                                                ▼                           ▼
                                                   ProcessedPaymentStoreService      DashboardService
                                                     append JSON line                in-memory index
                                                                ▼                           ▼
                                                   processed_payments.jsonl    search() / fraud() / all()

Components

Component	File	Responsibility
EventBus	services/event_bus.py	In-process pub/sub broker. publish(topic, msg) fans out to every subscriber's independent queue. subscribe(topic, priority=True) creates a priority-ordered subscription backed by PriorityQueueService; priority=False (default) creates a plain FIFO queue. Thread-safe.
PriorityQueueService	services/event_bus.py	Thread-safe min-heap backed by queue.PriorityQueue. Used internally by EventBus when priority=True. Items are ordered by their priority attribute (lower = higher priority). A monotonic counter breaks ties so equal-priority payments are served FIFO.
PaymentPublisherService	services/payment_publisher.py	Daemon thread that generates a random Payment every _PUBLISH_INTERVAL seconds and publishes it to "payment.pending". Each payment is assigned a random user, amount, currency, type, and payment location zip code.
FraudDetectionService	services/fraud_detection_service.py	FIFO subscriber to "payment.pending". Scores each payment using payment type and location mismatch against the user's home zip. Sets payment.risk in-place on the shared object. Writes high-risk payments to fraud.jsonl.
PaymentProcessorService	services/processor/payment_processor_service.py	Priority subscriber to "payment.pending". Waits for a backlog of at least _MIN_BACKLOG before consuming. Rejects (failed) high-risk payments and approves (completed) all others. Publishes results to "payment.processed".
ProcessedPaymentStoreService	services/processor/processed_payment_store_service.py	FIFO subscriber to "payment.processed". Appends each payment as a JSON line to processed_payments.jsonl.
DashboardService	services/dashboard_service.py	FIFO subscriber to "payment.processed". Keeps an in-memory list of all processed payments. Exposes search() with optional AND-combined filters and a fraud() shorthand for high-risk payments.

Fraud risk scoring

Each payment is scored by FraudDetectionService before the processor consumes it. The score combines a payment type factor and a location mismatch factor.

Payment type base score:

Type	Score	Reason
chip	1	EMV chip — hardest to clone
tap / contactless	2	NFC limited-range
swipe	3	Magnetic stripe — easily skimmed
online / manual	4	Card-not-present — highest fraud vector

Location factor (payment zip vs. user home zip):

Condition	Score
Same zip code	+0
Same first 3 digits (same metro area)	+1
Different region	+2

Risk classification:

Total score	Risk
≤ 2	low
3 – 4	medium
≥ 5	high

High-risk payments are written to fraud.jsonl and rejected by the processor.

Predefined users

Payments are always assigned to one of these 10 users. Their user_ids and home_zips are stable across runs.

Name	Email	Home zip	user_id
Alice Martin	alice.martin@example.com	10001	a1b2c3d4-0001-4000-8000-000000000001
Bob Chen	bob.chen@example.com	94102	a1b2c3d4-0002-4000-8000-000000000002
Clara Novak	clara.novak@example.com	60601	a1b2c3d4-0003-4000-8000-000000000003
David Okafor	david.okafor@example.com	77001	a1b2c3d4-0004-4000-8000-000000000004
Elena Vasquez	elena.vasquez@example.com	85001	a1b2c3d4-0005-4000-8000-000000000005
Frank Müller	frank.muller@example.com	30301	a1b2c3d4-0006-4000-8000-000000000006
Grace Kim	grace.kim@example.com	98101	a1b2c3d4-0007-4000-8000-000000000007
Henry Patel	henry.patel@example.com	02101	a1b2c3d4-0008-4000-8000-000000000008
Isabelle Dubois	isabelle.dubois@example.com	33101	a1b2c3d4-0009-4000-8000-000000000009
James O'Brien	james.obrien@example.com	19101	a1b2c3d4-000a-4000-8000-00000000000a

Priority queue behaviour

The publisher assigns each payment a random priority between 0 and 10. The processor waits for a backlog of at least _MIN_BACKLOG before consuming, so there is always a pool of payments at varying priority levels in the heap. A priority-0 payment inserted after several priority-5 payments will still be dequeued ahead of them.

Output files

File	Contents
processed_payments.jsonl	One JSON object per line for every processed payment
fraud.jsonl	One JSON object per line for every high-risk payment flagged by the fraud detector
payment_processor.log	All service log output at INFO level

To watch logs while the REPL is running, open a second terminal:

# Windows
Get-Content payment_processor.log -Wait

# macOS / Linux
tail -f payment_processor.log

Interactive dashboard

All filters are optional and AND-combined. Calling dashboard.search() with no arguments returns everything (equivalent to dashboard.all()). Results always reflect the latest state — re-run any query to pick up newly processed payments without restarting anything.

Fraud queries

Call	Description
dashboard.fraud()	All high-risk payments
dashboard.search(risk="high")	Equivalent to dashboard.fraud()
dashboard.search(risk="medium")	All medium-risk payments
dashboard.search(risk="low")	All low-risk payments
dashboard.search(risk="high", name="Alice")	High-risk payments for a specific user

User lookups

Call	Description
dashboard.search(name="Alice")	Case-insensitive substring match on name
dashboard.search(name="alice martin")	Full name match (still case-insensitive)
dashboard.search(email="alice.martin@example.com")	Exact email match
dashboard.search(email="@example.com")	Partial email match — all users on the same domain
dashboard.search(user_id="a1b2c3d4-0001-4000-8000-000000000001")	Look up by exact user ID

Payment filters

Call	Description
dashboard.count()	Total number of processed payments seen so far
dashboard.all()	Every processed payment as a list of Payment dataclasses
dashboard.search(status="completed")	All approved payments
dashboard.search(status="failed")	All rejected payments
dashboard.search(currency="USD")	All payments in a given currency
dashboard.search(priority=0)	All payments at the highest priority level
dashboard.search(min_amount=Decimal("1000"))	Payments above a minimum amount
dashboard.search(max_amount=Decimal("5000"))	Payments below a maximum amount
dashboard.search(min_amount=Decimal("1000"), max_amount=Decimal("5000"))	Payments within an amount range