Outbox

SprintLib/BaseView.py

@@ -1,6 +1,7 @@
 from typing import Optional
 
 from django.core.handlers.wsgi import WSGIRequest
+from django.db import transaction
 from django.http import HttpResponseRedirect, JsonResponse
 from django.shortcuts import render
 from django.utils import timezone
@@ -27,56 +28,57 @@ class BaseView:
     def as_view(cls):
         @csrf_exempt
         def execute(request):
-            if request.user.is_authenticated:
-                user_info = request.user.userinfo
-                user_info.last_request = timezone.now()
-                user_info.save()
-            c = cls(request)
-            if c.required_login is not None:
-                if c.required_login and not request.user.is_authenticated:
-                    return HttpResponseRedirect("/enter")
-                if c.required_login and not request.user.userinfo.verified:
-                    return HttpResponseRedirect("/set_username")
-                if (
-                    not c.required_login
-                    and request.user.is_authenticated
-                    and request.user.userinfo.verified
-                ):
+            with transaction.atomic():
+                if request.user.is_authenticated:
+                    user_info = request.user.userinfo
+                    user_info.last_request = timezone.now()
+                    user_info.save()
+                c = cls(request)
+                if c.required_login is not None:
+                    if c.required_login and not request.user.is_authenticated:
+                        return HttpResponseRedirect("/enter")
+                    if c.required_login and not request.user.userinfo.verified:
+                        return HttpResponseRedirect("/set_username")
+                    if (
+                        not c.required_login
+                        and request.user.is_authenticated
+                        and request.user.userinfo.verified
+                    ):
+                        return HttpResponseRedirect("/")
+                request_method = request.method.lower()
+                exec("from Main.models import *")
+                context = {}
+                for key in request.GET.keys():
+                    if key.endswith("_id") and key not in cls.fields_except:
+                        model_name = key.rstrip("_id")
+                        setattr(
+                            c,
+                            model_name,
+                            eval(model_name[0].upper() + model_name[1:]).objects.get(
+                                id=int(request.GET[key])
+                            ),
+                        )
+                        context[model_name] = getattr(c, model_name)
+                if "action" in request.POST.keys():
+                    request_method += "_" + request.POST["action"]
+                method = getattr(c, request_method, None)
+                try:
+                    data = c.pre_handle()
+                    if method:
+                        if data is None:
+                            data = method()
+                        if type(data) == str:
+                            return HttpResponseRedirect(data)
+                        if type(data) == dict:
+                            return JsonResponse(data)
+                        if data is not None:
+                            return data
+                    context = {**context, **c.context}
+                    res = render(request, c.view_file, context)
+                    res.headers["X-Frame-Options"] = "ALLOW"
+                    return res
+                except AccessError:
                     return HttpResponseRedirect("/")
-            request_method = request.method.lower()
-            exec("from Main.models import *")
-            context = {}
-            for key in request.GET.keys():
-                if key.endswith("_id") and key not in cls.fields_except:
-                    model_name = key.rstrip("_id")
-                    setattr(
-                        c,
-                        model_name,
-                        eval(model_name[0].upper() + model_name[1:]).objects.get(
-                            id=int(request.GET[key])
-                        ),
-                    )
-                    context[model_name] = getattr(c, model_name)
-            if "action" in request.POST.keys():
-                request_method += "_" + request.POST["action"]
-            method = getattr(c, request_method, None)
-            try:
-                data = c.pre_handle()
-                if method:
-                    if data is None:
-                        data = method()
-                    if type(data) == str:
-                        return HttpResponseRedirect(data)
-                    if type(data) == dict:
-                        return JsonResponse(data)
-                    if data is not None:
-                        return data
-                context = {**context, **c.context}
-                res = render(request, c.view_file, context)
-                res.headers["X-Frame-Options"] = "ALLOW"
-                return res
-            except AccessError:
-                return HttpResponseRedirect("/")
 
         return execute
 

SprintLib/queue.py

@@ -1,48 +1,70 @@
 import json
+from json import JSONDecodeError
 
-import django.db
-import django.db.utils
 import pika
-import psycopg2
+from django.db import transaction
+from django.core.exceptions import ObjectDoesNotExist
 from django.core.management import BaseCommand
+from django.utils import timezone
 from pika.adapters.utils.connection_workflow import AMQPConnectorException
 
+from Main.models.outbox_message import OutboxMessage
 from Sprint import settings
 
 
 def send_to_queue(queue_name, payload):
-    with pika.BlockingConnection(
-        pika.ConnectionParameters(
-            host=settings.RABBIT_HOST,
-            port=settings.RABBIT_PORT,
-            credentials=pika.PlainCredentials('guest', settings.RABBIT_PASSWORD)
-        )
-    ) as connection:
-        channel = connection.channel()
-        channel.queue_declare(queue=queue_name)
-        channel.basic_publish(
-            exchange="",
-            routing_key=queue_name,
-            body=json.dumps(payload).encode('utf-8'),
-        )
+    OutboxMessage.objects.create(
+        queue=queue_name,
+        body=payload
+    )
+
+
+class RollbackException(Exception):
+    ...
 
 
 class MessagingSupport(BaseCommand):
     queue_name = None
+    with_transaction = True
 
     def process(self, payload: dict):
         raise NotImplementedError
 
     def consume(self, ch, method, properties, body):
-        data = json.loads(body.decode('utf-8'))
+        try:
+            data = json.loads(body.decode('utf-8'))
+        except JSONDecodeError:
+            print("Message is not JSON decodable")
+            return
         print(f"Got {data}, processing...")
         try:
-            self.process(data)
-            print("Process finished successfully")
-        except (psycopg2.OperationalError, django.db.OperationalError, django.db.utils.OperationalError):
-            print("Failed to connect to database, restarting...")
-            send_to_queue(self.queue_name, data)
-            raise
+            outbox_message = OutboxMessage.objects.get(id=data["id"])
+        except ObjectDoesNotExist:
+            print(f"Message with id {data['id']} does not exist")
+            return
+        if outbox_message.time_processed is not None:
+            print("Message is already processed")
+            return
+        outbox_message.time_received = timezone.now()
+        outbox_message.save()
+        try:
+            if self.with_transaction:
+                with transaction.atomic():
+                    self.process(data["body"])
+                    outbox_message.refresh_from_db()
+                    if outbox_message.time_processed is not None:
+                        print("Message already processed, rolling back")
+                        raise RollbackException("Just for rolling back")
+                    else:
+                        outbox_message.time_processed = timezone.now()
+                        outbox_message.save()
+            else:
+                self.process(data["body"])
+                outbox_message.time_processed = timezone.now()
+                outbox_message.save()
+        except RollbackException:
+            pass
+        print("Process finished successfully")
 
     def handle(self, *args, **options):
         if self.queue_name is None:

SprintLib/utils.py

@@ -4,7 +4,7 @@ from random import choice
 from time import sleep
 
 from django.core.management import BaseCommand
-from minio import Minio
+from minio import Minio, S3Error
 
 from Sprint import settings
 from SprintLib.queue import send_to_queue
@@ -23,8 +23,12 @@ client = Minio(
 
 @lock('write_bytes')
 def write_bytes(data: bytes):
-    obj = client.get_object(BUCKET_NAME, 'meta.txt')
-    num = int(obj.data.decode('utf-8')) + 1
+    try:
+        obj = client.get_object(BUCKET_NAME, 'meta.txt')
+        num = int(obj.data.decode('utf-8')) + 1
+    except S3Error:
+        client.put_object(BUCKET_NAME, 'meta.txt', io.BytesIO(b"1"), 1)
+        num = 1
     b_num = str(num).encode('utf-8')
     client.put_object(BUCKET_NAME, str(num), io.BytesIO(data), len(data))
     client.put_object(BUCKET_NAME, 'meta.txt', io.BytesIO(b_num), len(b_num))