1 files changed, 26 insertions, 4 deletions
diff --git a/watermark_worker/watermark.py b/watermark_worker/watermark.py
index f2b02f0..5d6388b 100644
--- a/watermark_worker/watermark.py
+++ b/watermark_worker/watermark.py
@@ -21,7 +21,7 @@ def connect_to_s3():
                         verify=False) # disable SSL
 
 def connect_to_beanstalkd():
-    return beanstalkc.Connection("localhost", 11300)
+    return beanstalkc.Connection("localhost", 11300, encoding="utf-8")
 
 def watermark(image: PIL.Image.Image) -> PIL.Image.Image:
     w, h = image.size
@@ -76,7 +76,27 @@ def process_image(db_conn: psycopg.Connection, s3_client, image_id: int, s3_key:
     """, [image_id])
     db_conn.commit()
 
-def process_assignment(assignment_id: int):
+def notify_owner(beanstakd_conn: beanstalkc.Connection, db_conn: psycopg.Connection, assignment_id: int):
+    result = db_conn.execute("""
+        SELECT
+            c.owner_id,
+            a.cemetary_plot_id
+        FROM assignments AS a
+        INNER JOIN cemetary_plots AS c ON c.id = a.cemetary_plot_id
+        WHERE a.id = %s;
+    """, [assignment_id]).fetchone()
+    assert result is not None
+    owner_id, cemetary_plot_id = result
+
+    beanstakd_conn.use("notification")
+    body = json.dumps({
+        "userId": owner_id,
+        "message": "Vi har besøgt dit gravsted",
+        "url": f"http://localhost:8000/cemetary_plots/{cemetary_plot_id}",
+    })
+    beanstakd_conn.put(body)
+
+def process_assignment(beanstakd_conn: beanstalkc.Connection, assignment_id: int):
     # Retrieve all UNPROCESSED images from DB.
     # Some images may have already been processed by an earlier, partial job.
     db_conn = connect_to_db()
@@ -101,12 +121,14 @@ def process_assignment(assignment_id: int):
         UPDATE
             assignments
         SET
-            state = 'AWAITING_OWNER_NOTIFICATION' :: assignment_state
+            state = 'AWAITING_DONE' :: assignment_state
         WHERE
             id = %s
     """, [assignment_id])
     db_conn.commit()
 
+    notify_owner(beanstakd_conn, db_conn, assignment_id)
+
 beanstalk = connect_to_beanstalkd()
 beanstalk.watch("watermarking")
 
@@ -117,6 +139,6 @@ while True:
     print(f"!! Got job: {job.body}")
 
     job_body = json.loads(job.body)
-    process_assignment(job_body["assignmentId"])
+    process_assignment(beanstalk, job_body["assignmentId"])
 
     job.delete()