From 0eeefa377d16c324629319c34663fd09906a5e26 Mon Sep 17 00:00:00 2001
From: ruinivist <ruinivist@proton.me>
Date: Tue, 14 Apr 2026 21:54:01 +0000
Subject: [PATCH] add notifier

---
 TASK.md     | 42 +++++++++++++++++++++---------------------
 notifier.py | 44 ++++++++++++++++++++++++++++++++++++++++++++
 worker.py   |  2 +-
 3 files changed, 66 insertions(+), 22 deletions(-)

diff --git a/TASK.md b/TASK.md
index 9334046..5efe065 100644
--- a/TASK.md
+++ b/TASK.md
@@ -104,47 +104,47 @@ Success check:
 
 ## Step 4: Worker (35 min)
 
-- [ ] `worker.py` consumes from `orders.q` with manual ACK.
-- [ ] For each message:
-- [ ] Parse JSON
-- [ ] Simulate processing (`sleep(1)`)
-- [ ] If `should_fail` and `retry_count < 2`, republish same order with `retry_count + 1` to `order.created`, ACK original.
-- [ ] If `should_fail` and `retry_count >= 2`, publish to `order.failed`, ACK original.
-- [ ] Else publish to `order.processed`, ACK original.
+- [x] `worker.py` consumes from `orders.q` with manual ACK.
+- [x] For each message:
+- [x] Parse JSON
+- [x] Simulate processing (`sleep(1)`)
+- [x] If `should_fail` and `retry_count < 2`, republish same order with `retry_count + 1` to `order.created`, ACK original.
+- [x] If `should_fail` and `retry_count >= 2`, publish to `order.failed`, ACK original.
+- [x] Else publish to `order.processed`, ACK original.
 
 Important:
 
-- [ ] Set `prefetch_count=1`.
-- [ ] Use persistent publish for all outgoing messages.
+- [x] Set `prefetch_count=1`.
+- [x] Use persistent publish for all outgoing messages.
 
 Success check:
 
-- [ ] You see logs for retries and final outcomes.
+- [x] You see logs for retries and final outcomes.
 
 ## Step 5: Notifier (15 min)
 
-- [ ] `notifier.py` consumes both `orders.processed.q` and `orders.failed.q`.
-- [ ] Print concise final messages, e.g.:
-- [ ] `ORDER 1001 PROCESSED`
-- [ ] `ORDER 1003 FAILED AFTER RETRIES`
-- [ ] ACK all notifier messages.
+- [x] `notifier.py` consumes both `orders.processed.q` and `orders.failed.q`.
+- [x] Print concise final messages, e.g.:
+- [x] `ORDER 1001 PROCESSED`
+- [x] `ORDER 1003 FAILED AFTER RETRIES`
+- [x] ACK all notifier messages.
 
 Success check:
 
-- [ ] After running producer, notifier shows both success and failure events.
+- [x] After running producer, notifier shows both success and failure events.
 
 ## Step 6: End-to-End Run (10 min)
 
 Run in 3 terminals:
 
-- [ ] Terminal A: `uv run worker.py`
-- [ ] Terminal B: `uv run notifier.py`
-- [ ] Terminal C: `uv run producer.py`
+- [x] Terminal A: `uv run worker.py`
+- [x] Terminal B: `uv run notifier.py`
+- [x] Terminal C: `uv run producer.py`
 
 Expected behavior:
 
-- [ ] Normal orders finish in `orders.processed.q` path.
-- [ ] Failing orders retry 2 times then go `orders.failed.q`.
+- [x] Normal orders finish in `orders.processed.q` path.
+- [x] Failing orders retry 2 times then go `orders.failed.q`.
 
 ## Step 7: Failure Drill (10 min)
 
diff --git a/notifier.py b/notifier.py
index e69de29..3e73ecd 100644
--- a/notifier.py
+++ b/notifier.py
@@ -0,0 +1,44 @@
+from typing import TypedDict
+
+import json
+import pika
+from pika.adapters.blocking_connection import BlockingChannel
+from pika.spec import Basic
+
+conn = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
+ch = conn.channel()
+
+ch.basic_qos(prefetch_count=1)
+
+# hover over on_messae_callback
+# the docs specify signature
+
+
+class OrderMessage(TypedDict):
+    order_id: int
+    user_id: str
+    amount: float
+    should_fail: bool
+    retry_count: int
+
+
+def on_order_processed(channel: BlockingChannel, method: Basic.Deliver, properties, body):
+    order: OrderMessage = json.loads(body)
+    print(f"processed order: {order["order_id"]}")
+    channel.basic_ack(delivery_tag=method.delivery_tag)
+
+
+def on_order_failed(channel: BlockingChannel, method: Basic.Deliver, properties, body):
+    order: OrderMessage = json.loads(body)
+    print(f"failed after retries: {order["order_id"]}")
+    channel.basic_ack(delivery_tag=method.delivery_tag)
+
+
+ch.basic_consume(queue="orders.processed.q",
+                 on_message_callback=on_order_processed, auto_ack=False)
+ch.basic_consume(queue="orders.failed.q",
+                 on_message_callback=on_order_failed, auto_ack=False)
+
+# easy to forget
+print("[notifier] waiting on messages")
+ch.start_consuming()
diff --git a/worker.py b/worker.py
index 1fe9151..eb09e87 100644
--- a/worker.py
+++ b/worker.py
@@ -75,6 +75,6 @@ def on_message(
 
 
 ch.basic_consume(queue="orders.q",
-                 on_message_callback=on_message, auto_ack=False)
+                 on_message_callback=on_message, auto_ack=False)  # the manual ack so this is false
 print("[worker] waiting for messages...")
 ch.start_consuming()