We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add user id to metadata related to all data/actions in stripe


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `lego/apps/events/tasks.py`
Content:
```
1 from datetime import timedelta
2 
3 import stripe
4 from django.db import IntegrityError, transaction
5 from django.utils import timezone
6 from redis.exceptions import LockError
7 from structlog import get_logger
8 
9 from lego import celery_app
10 from lego.apps.events import constants
11 from lego.apps.events.exceptions import (
12     EventHasClosed, PoolCounterNotEqualToRegistrationCount, WebhookDidNotFindRegistration
13 )
14 from lego.apps.events.models import Event, Registration
15 from lego.apps.events.notifications import EventPaymentOverdueCreatorNotification
16 from lego.apps.events.serializers.registrations import StripeObjectSerializer
17 from lego.apps.events.websockets import (
18     notify_event_registration, notify_user_payment, notify_user_registration
19 )
20 from lego.apps.feed.registry import get_handler
21 from lego.utils.tasks import AbakusTask
22 
23 log = get_logger()
24 
25 
26 class AsyncRegister(AbakusTask):
27     serializer = 'json'
28     default_retry_delay = 5
29     registration = None
30 
31     def on_failure(self, *args):
32         if self.request.retries == self.max_retries:
33             self.registration.status = constants.FAILURE_REGISTER
34             self.registration.save()
35             notify_user_registration(
36                 constants.SOCKET_REGISTRATION_FAILURE, self.registration,
37                 error_message='Registrering feilet'
38             )
39 
40 
41 class Payment(AbakusTask):
42     serializer = 'json'
43     default_retry_delay = 5
44     registration = None
45 
46     def on_failure(self, return_value, *args):
47         if self.request.retries == self.max_retries:
48             if return_value.json_body:
49                 error = return_value.json_body['error']
50                 self.registration.charge_id = error['charge']
51                 self.registration.charge_status = error['code']
52                 self.registration.save()
53                 notify_user_registration(
54                     constants.SOCKET_PAYMENT_FAILURE, self.registration,
55                     error_message=error['message']
56                 )
57             else:
58                 self.registration.charge_status = constants.PAYMENT_FAILURE
59                 self.registration.save()
60                 notify_user_registration(
61                     constants.SOCKET_PAYMENT_FAILURE, self.registration,
62                     error_message='Payment failed'
63                 )
64 
65 
66 @celery_app.task(base=AsyncRegister, bind=True)
67 def async_register(self, registration_id, logger_context=None):
68     self.setup_logger(logger_context)
69 
70     self.registration = Registration.objects.get(id=registration_id)
71     try:
72         with transaction.atomic():
73             self.registration.event.register(self.registration)
74             transaction.on_commit(lambda: notify_event_registration(
75                 constants.SOCKET_REGISTRATION_SUCCESS, self.registration
76             ))
77         log.info('registration_success', registration_id=self.registration.id)
78     except LockError as e:
79         log.error(
80             'registration_cache_lock_error', exception=e, registration_id=self.registration.id
81         )
82         raise self.retry(exc=e, max_retries=3)
83     except EventHasClosed as e:
84         log.warn(
85             'registration_tried_after_started', exception=e, registration_id=self.registration.id
86         )
87     except (ValueError, IntegrityError) as e:
88         log.error('registration_error', exception=e, registration_id=self.registration.id)
89         raise self.retry(exc=e, max_retries=3)
90 
91 
92 @celery_app.task(serializer='json', bind=True, base=AbakusTask, default_retry_delay=30)
93 def async_unregister(self, registration_id, logger_context=None):
94     self.setup_logger(logger_context)
95 
96     registration = Registration.objects.get(id=registration_id)
97     pool_id = registration.pool_id
98     try:
99         with transaction.atomic():
100             registration.event.unregister(registration)
101             activation_time = registration.event.get_earliest_registration_time(registration.user)
102             transaction.on_commit(lambda: notify_event_registration(
103                 constants.SOCKET_UNREGISTRATION_SUCCESS, registration,
104                 from_pool=pool_id, activation_time=activation_time
105             ))
106         log.info('unregistration_success', registration_id=registration.id)
107     except EventHasClosed as e:
108         log.warn('unregistration_tried_after_started', exception=e, registration_id=registration.id)
109     except IntegrityError as e:
110         log.error('unregistration_error', exception=e, registration_id=registration.id)
111         registration.status = constants.FAILURE_UNREGISTER
112         registration.save()
113         notify_user_registration(
114             constants.SOCKET_UNREGISTRATION_FAILURE, registration,
115             error_message='Avregistrering feilet'
116         )
117 
118 
119 @celery_app.task(base=Payment, bind=True)
120 def async_payment(self, registration_id, token, logger_context=None):
121     self.setup_logger(logger_context)
122 
123     self.registration = Registration.objects.get(id=registration_id)
124     event = self.registration.event
125     try:
126         response = stripe.Charge.create(
127             amount=event.get_price(self.registration.user), currency='NOK', source=token,
128             description=event.slug, metadata={
129                 'EVENT_ID': event.id,
130                 'USER': self.registration.user.full_name,
131                 'EMAIL': self.registration.user.email
132             }
133         )
134         log.info('stripe_payment_success', registration_id=self.registration.id)
135         return response
136     except stripe.error.CardError as e:
137         raise self.retry(exc=e)
138     except stripe.error.InvalidRequestError as e:
139         log.error('invalid_request', exception=e, registration_id=self.registration.id)
140         self.registration.charge_status = e.json_body['error']['type']
141         self.registration.save()
142         notify_user_payment(
143             constants.SOCKET_PAYMENT_FAILURE, self.registration, error_message='Invalid request'
144         )
145     except stripe.error.StripeError as e:
146         log.error('stripe_error', exception=e, registration_id=self.registration.id)
147         raise self.retry(exc=e)
148 
149 
150 @celery_app.task(serializer='json', bind=True, base=AbakusTask)
151 def registration_payment_save(self, result, registration_id, logger_context=None):
152     self.setup_logger(logger_context)
153 
154     try:
155         registration = Registration.objects.get(id=registration_id)
156         registration.charge_id = result['id']
157         registration.charge_amount = result['amount']
158         registration.charge_status = result['status']
159         registration.save()
160         notify_user_payment(
161             constants.SOCKET_PAYMENT_SUCCESS, registration, success_message='Betaling gjennomført'
162         )
163     except IntegrityError as e:
164         log.error('registration_save_error', exception=e, registration_id=registration_id)
165         raise self.retry(exc=e)
166 
167 
168 @celery_app.task(serializer='json', bind=True, base=AbakusTask)
169 def check_for_bump_on_pool_creation_or_expansion(self, event_id, logger_context=None):
170     """Task checking for bumps when event and pools are updated"""
171     self.setup_logger(logger_context)
172 
173     # Event is locked using the instance field "is_ready"
174     event = Event.objects.get(pk=event_id)
175     event.bump_on_pool_creation_or_expansion()
176     event.is_ready = True
177     event.save(update_fields=['is_ready'])
178 
179 
180 @celery_app.task(serializer='json', bind=True, base=AbakusTask)
181 def stripe_webhook_event(self, event_id, event_type, logger_context=None):
182     self.setup_logger(logger_context)
183 
184     if event_type in ['charge.failed', 'charge.refunded', 'charge.succeeded']:
185         event = stripe.Event.retrieve(event_id)
186         serializer = StripeObjectSerializer(data=event.data['object'])
187         serializer.is_valid(raise_exception=True)
188 
189         metadata = serializer.data['metadata']
190         registration = Registration.objects.filter(
191             event_id=metadata['EVENT_ID'], user__email=metadata['EMAIL']
192         ).first()
193         if not registration:
194             log.error('stripe_webhook_error', event_id=event_id, metadata=metadata)
195             raise WebhookDidNotFindRegistration(event_id, metadata)
196         registration.charge_id = serializer.data['id']
197         registration.charge_amount = serializer.data['amount']
198         registration.charge_amount_refunded = serializer.data['amount_refunded']
199         registration.charge_status = serializer.data['status']
200         registration.save()
201         log.info('stripe_webhook_received', event_id=event_id, registration_id=registration.id)
202 
203 
204 @celery_app.task(serializer='json', bind=True, base=AbakusTask)
205 def check_events_for_registrations_with_expired_penalties(self, logger_context=None):
206     self.setup_logger(logger_context)
207 
208     events_ids = Event.objects.filter(start_time__gte=timezone.now()
209                                       ).exclude(registrations=None).values_list('id', flat=True)
210     for event_id in events_ids:
211         with transaction.atomic():
212             locked_event = Event.objects.select_for_update().get(pk=event_id)
213             if locked_event.waiting_registrations.exists():
214                 for pool in locked_event.pools.all():
215                     if pool.is_activated and not pool.is_full:
216                         for i in range(locked_event.waiting_registrations.count()):
217                             locked_event.check_for_bump_or_rebalance(pool)
218                             if pool.is_full:
219                                 break
220 
221 
222 @celery_app.task(serializer='json', bind=True, base=AbakusTask)
223 def bump_waiting_users_to_new_pool(self, logger_context=None):
224     self.setup_logger(logger_context)
225 
226     events_ids = Event.objects.filter(start_time__gte=timezone.now()
227                                       ).exclude(registrations=None).values_list('id', flat=True)
228     for event_id in events_ids:
229         with transaction.atomic():
230             locked_event = Event.objects.select_for_update().get(pk=event_id)
231             if locked_event.waiting_registrations.exists():
232                 for pool in locked_event.pools.all():
233                     if not pool.is_full:
234                         act = pool.activation_date
235                         now = timezone.now()
236                         if not pool.is_activated and act < now + timedelta(minutes=35):
237                             locked_event.early_bump(pool)
238                             log.info('early_bump_executed', event_id=event_id, pool_id=pool.id)
239                         elif pool.is_activated and act > now - timedelta(minutes=35):
240                             log.info('early_bump_executed', event_id=event_id, pool_id=pool.id)
241                             locked_event.early_bump(pool)
242 
243 
244 @celery_app.task(serializer='json', bind=True, base=AbakusTask)
245 def notify_user_when_payment_soon_overdue(self, logger_context=None):
246     self.setup_logger(logger_context)
247 
248     time = timezone.now()
249     events = Event.objects.filter(
250         payment_due_date__range=(time - timedelta(days=2), time + timedelta(days=3)),
251         is_priced=True, use_stripe=True
252     ).exclude(registrations=None).prefetch_related('registrations')
253     for event in events:
254         for registration in event.registrations.exclude(pool=None):
255             if registration.should_notify(time):
256                 log.info(
257                     'registration_notified_overdue_payment', event_id=event.id,
258                     registration_id=registration.id
259                 )
260                 get_handler(Registration).handle_payment_overdue(registration)
261 
262 
263 @celery_app.task(serializer='json', bind=True, base=AbakusTask)
264 def notify_event_creator_when_payment_overdue(self, logger_context=None):
265     self.setup_logger(logger_context)
266 
267     time = timezone.now()
268     events = Event.objects.filter(
269         payment_due_date__lte=time, is_priced=True, use_stripe=True, end_time__gte=time
270     ).exclude(registrations=None).prefetch_related('registrations')
271     for event in events:
272         registrations_due = event.registrations.exclude(pool=None).exclude(
273             charge_status__in=[constants.PAYMENT_MANUAL, constants.PAYMENT_SUCCESS]
274         ).prefetch_related('user')
275         if registrations_due:
276             users = [
277                 {
278                     'name': registration.user.get_full_name(),
279                     'email': registration.user.email
280                 } for registration in registrations_due
281             ]
282             notification = EventPaymentOverdueCreatorNotification(
283                 event.created_by, event=event, users=users
284             )
285             notification.notify()
286             log.info(
287                 'event_creator_notified_of_overdue_payments', event_id=event.id,
288                 creator=event.created_by
289             )
290 
291 
292 @celery_app.task(serializer='json', bind=True, base=AbakusTask)
293 def check_that_pool_counters_match_registration_number(self, logger_context=None):
294     """
295     Task that checks whether pools counters are in sync with number of registrations. We do not
296     enforce this check for events that are merged, hence the merge_time filter, because
297     incrementing the counter decreases the registration performance
298     """
299     self.setup_logger(logger_context)
300 
301     events_ids = Event.objects.filter(
302         start_time__gte=timezone.now(), merge_time__gte=timezone.now()
303     ).values_list('id', flat=True)
304 
305     for event_id in events_ids:
306         with transaction.atomic():
307             locked_event = Event.objects.select_for_update().get(pk=event_id)
308             for pool in locked_event.pools.all():
309                 registration_count = pool.registrations.count()
310                 if pool.counter != registration_count:
311                     log.critical('pool_counter_not_equal_registration_count', pool=pool)
312                     raise PoolCounterNotEqualToRegistrationCount(
313                         pool, registration_count, locked_event
314                     )
315
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/lego/apps/events/tasks.py b/lego/apps/events/tasks.py
--- a/lego/apps/events/tasks.py
+++ b/lego/apps/events/tasks.py
@@ -127,8 +127,9 @@
             amount=event.get_price(self.registration.user), currency='NOK', source=token,
             description=event.slug, metadata={
                 'EVENT_ID': event.id,
+                'USER_ID': self.registration.user.id,
                 'USER': self.registration.user.full_name,
-                'EMAIL': self.registration.user.email
+                'EMAIL': self.registration.user.email,
             }
         )
         log.info('stripe_payment_success', registration_id=self.registration.id)

{"golden_diff": "diff --git a/lego/apps/events/tasks.py b/lego/apps/events/tasks.py\n--- a/lego/apps/events/tasks.py\n+++ b/lego/apps/events/tasks.py\n@@ -127,8 +127,9 @@\n             amount=event.get_price(self.registration.user), currency='NOK', source=token,\n             description=event.slug, metadata={\n                 'EVENT_ID': event.id,\n+                'USER_ID': self.registration.user.id,\n                 'USER': self.registration.user.full_name,\n-                'EMAIL': self.registration.user.email\n+                'EMAIL': self.registration.user.email,\n             }\n         )\n         log.info('stripe_payment_success', registration_id=self.registration.id)\n", "issue": "Add user id to metadata related to all data/actions in stripe\n\n", "before_files": [{"content": "from datetime import timedelta\n\nimport stripe\nfrom django.db import IntegrityError, transaction\nfrom django.utils import timezone\nfrom redis.exceptions import LockError\nfrom structlog import get_logger\n\nfrom lego import celery_app\nfrom lego.apps.events import constants\nfrom lego.apps.events.exceptions import (\n    EventHasClosed, PoolCounterNotEqualToRegistrationCount, WebhookDidNotFindRegistration\n)\nfrom lego.apps.events.models import Event, Registration\nfrom lego.apps.events.notifications import EventPaymentOverdueCreatorNotification\nfrom lego.apps.events.serializers.registrations import StripeObjectSerializer\nfrom lego.apps.events.websockets import (\n    notify_event_registration, notify_user_payment, notify_user_registration\n)\nfrom lego.apps.feed.registry import get_handler\nfrom lego.utils.tasks import AbakusTask\n\nlog = get_logger()\n\n\nclass AsyncRegister(AbakusTask):\n    serializer = 'json'\n    default_retry_delay = 5\n    registration = None\n\n    def on_failure(self, *args):\n        if self.request.retries == self.max_retries:\n            self.registration.status = constants.FAILURE_REGISTER\n            self.registration.save()\n            notify_user_registration(\n                constants.SOCKET_REGISTRATION_FAILURE, self.registration,\n                error_message='Registrering feilet'\n            )\n\n\nclass Payment(AbakusTask):\n    serializer = 'json'\n    default_retry_delay = 5\n    registration = None\n\n    def on_failure(self, return_value, *args):\n        if self.request.retries == self.max_retries:\n            if return_value.json_body:\n                error = return_value.json_body['error']\n                self.registration.charge_id = error['charge']\n                self.registration.charge_status = error['code']\n                self.registration.save()\n                notify_user_registration(\n                    constants.SOCKET_PAYMENT_FAILURE, self.registration,\n                    error_message=error['message']\n                )\n            else:\n                self.registration.charge_status = constants.PAYMENT_FAILURE\n                self.registration.save()\n                notify_user_registration(\n                    constants.SOCKET_PAYMENT_FAILURE, self.registration,\n                    error_message='Payment failed'\n                )\n\n\n@celery_app.task(base=AsyncRegister, bind=True)\ndef async_register(self, registration_id, logger_context=None):\n    self.setup_logger(logger_context)\n\n    self.registration = Registration.objects.get(id=registration_id)\n    try:\n        with transaction.atomic():\n            self.registration.event.register(self.registration)\n            transaction.on_commit(lambda: notify_event_registration(\n                constants.SOCKET_REGISTRATION_SUCCESS, self.registration\n            ))\n        log.info('registration_success', registration_id=self.registration.id)\n    except LockError as e:\n        log.error(\n            'registration_cache_lock_error', exception=e, registration_id=self.registration.id\n        )\n        raise self.retry(exc=e, max_retries=3)\n    except EventHasClosed as e:\n        log.warn(\n            'registration_tried_after_started', exception=e, registration_id=self.registration.id\n        )\n    except (ValueError, IntegrityError) as e:\n        log.error('registration_error', exception=e, registration_id=self.registration.id)\n        raise self.retry(exc=e, max_retries=3)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask, default_retry_delay=30)\ndef async_unregister(self, registration_id, logger_context=None):\n    self.setup_logger(logger_context)\n\n    registration = Registration.objects.get(id=registration_id)\n    pool_id = registration.pool_id\n    try:\n        with transaction.atomic():\n            registration.event.unregister(registration)\n            activation_time = registration.event.get_earliest_registration_time(registration.user)\n            transaction.on_commit(lambda: notify_event_registration(\n                constants.SOCKET_UNREGISTRATION_SUCCESS, registration,\n                from_pool=pool_id, activation_time=activation_time\n            ))\n        log.info('unregistration_success', registration_id=registration.id)\n    except EventHasClosed as e:\n        log.warn('unregistration_tried_after_started', exception=e, registration_id=registration.id)\n    except IntegrityError as e:\n        log.error('unregistration_error', exception=e, registration_id=registration.id)\n        registration.status = constants.FAILURE_UNREGISTER\n        registration.save()\n        notify_user_registration(\n            constants.SOCKET_UNREGISTRATION_FAILURE, registration,\n            error_message='Avregistrering feilet'\n        )\n\n\n@celery_app.task(base=Payment, bind=True)\ndef async_payment(self, registration_id, token, logger_context=None):\n    self.setup_logger(logger_context)\n\n    self.registration = Registration.objects.get(id=registration_id)\n    event = self.registration.event\n    try:\n        response = stripe.Charge.create(\n            amount=event.get_price(self.registration.user), currency='NOK', source=token,\n            description=event.slug, metadata={\n                'EVENT_ID': event.id,\n                'USER': self.registration.user.full_name,\n                'EMAIL': self.registration.user.email\n            }\n        )\n        log.info('stripe_payment_success', registration_id=self.registration.id)\n        return response\n    except stripe.error.CardError as e:\n        raise self.retry(exc=e)\n    except stripe.error.InvalidRequestError as e:\n        log.error('invalid_request', exception=e, registration_id=self.registration.id)\n        self.registration.charge_status = e.json_body['error']['type']\n        self.registration.save()\n        notify_user_payment(\n            constants.SOCKET_PAYMENT_FAILURE, self.registration, error_message='Invalid request'\n        )\n    except stripe.error.StripeError as e:\n        log.error('stripe_error', exception=e, registration_id=self.registration.id)\n        raise self.retry(exc=e)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef registration_payment_save(self, result, registration_id, logger_context=None):\n    self.setup_logger(logger_context)\n\n    try:\n        registration = Registration.objects.get(id=registration_id)\n        registration.charge_id = result['id']\n        registration.charge_amount = result['amount']\n        registration.charge_status = result['status']\n        registration.save()\n        notify_user_payment(\n            constants.SOCKET_PAYMENT_SUCCESS, registration, success_message='Betaling gjennomf\u00f8rt'\n        )\n    except IntegrityError as e:\n        log.error('registration_save_error', exception=e, registration_id=registration_id)\n        raise self.retry(exc=e)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef check_for_bump_on_pool_creation_or_expansion(self, event_id, logger_context=None):\n    \"\"\"Task checking for bumps when event and pools are updated\"\"\"\n    self.setup_logger(logger_context)\n\n    # Event is locked using the instance field \"is_ready\"\n    event = Event.objects.get(pk=event_id)\n    event.bump_on_pool_creation_or_expansion()\n    event.is_ready = True\n    event.save(update_fields=['is_ready'])\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef stripe_webhook_event(self, event_id, event_type, logger_context=None):\n    self.setup_logger(logger_context)\n\n    if event_type in ['charge.failed', 'charge.refunded', 'charge.succeeded']:\n        event = stripe.Event.retrieve(event_id)\n        serializer = StripeObjectSerializer(data=event.data['object'])\n        serializer.is_valid(raise_exception=True)\n\n        metadata = serializer.data['metadata']\n        registration = Registration.objects.filter(\n            event_id=metadata['EVENT_ID'], user__email=metadata['EMAIL']\n        ).first()\n        if not registration:\n            log.error('stripe_webhook_error', event_id=event_id, metadata=metadata)\n            raise WebhookDidNotFindRegistration(event_id, metadata)\n        registration.charge_id = serializer.data['id']\n        registration.charge_amount = serializer.data['amount']\n        registration.charge_amount_refunded = serializer.data['amount_refunded']\n        registration.charge_status = serializer.data['status']\n        registration.save()\n        log.info('stripe_webhook_received', event_id=event_id, registration_id=registration.id)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef check_events_for_registrations_with_expired_penalties(self, logger_context=None):\n    self.setup_logger(logger_context)\n\n    events_ids = Event.objects.filter(start_time__gte=timezone.now()\n                                      ).exclude(registrations=None).values_list('id', flat=True)\n    for event_id in events_ids:\n        with transaction.atomic():\n            locked_event = Event.objects.select_for_update().get(pk=event_id)\n            if locked_event.waiting_registrations.exists():\n                for pool in locked_event.pools.all():\n                    if pool.is_activated and not pool.is_full:\n                        for i in range(locked_event.waiting_registrations.count()):\n                            locked_event.check_for_bump_or_rebalance(pool)\n                            if pool.is_full:\n                                break\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef bump_waiting_users_to_new_pool(self, logger_context=None):\n    self.setup_logger(logger_context)\n\n    events_ids = Event.objects.filter(start_time__gte=timezone.now()\n                                      ).exclude(registrations=None).values_list('id', flat=True)\n    for event_id in events_ids:\n        with transaction.atomic():\n            locked_event = Event.objects.select_for_update().get(pk=event_id)\n            if locked_event.waiting_registrations.exists():\n                for pool in locked_event.pools.all():\n                    if not pool.is_full:\n                        act = pool.activation_date\n                        now = timezone.now()\n                        if not pool.is_activated and act < now + timedelta(minutes=35):\n                            locked_event.early_bump(pool)\n                            log.info('early_bump_executed', event_id=event_id, pool_id=pool.id)\n                        elif pool.is_activated and act > now - timedelta(minutes=35):\n                            log.info('early_bump_executed', event_id=event_id, pool_id=pool.id)\n                            locked_event.early_bump(pool)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef notify_user_when_payment_soon_overdue(self, logger_context=None):\n    self.setup_logger(logger_context)\n\n    time = timezone.now()\n    events = Event.objects.filter(\n        payment_due_date__range=(time - timedelta(days=2), time + timedelta(days=3)),\n        is_priced=True, use_stripe=True\n    ).exclude(registrations=None).prefetch_related('registrations')\n    for event in events:\n        for registration in event.registrations.exclude(pool=None):\n            if registration.should_notify(time):\n                log.info(\n                    'registration_notified_overdue_payment', event_id=event.id,\n                    registration_id=registration.id\n                )\n                get_handler(Registration).handle_payment_overdue(registration)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef notify_event_creator_when_payment_overdue(self, logger_context=None):\n    self.setup_logger(logger_context)\n\n    time = timezone.now()\n    events = Event.objects.filter(\n        payment_due_date__lte=time, is_priced=True, use_stripe=True, end_time__gte=time\n    ).exclude(registrations=None).prefetch_related('registrations')\n    for event in events:\n        registrations_due = event.registrations.exclude(pool=None).exclude(\n            charge_status__in=[constants.PAYMENT_MANUAL, constants.PAYMENT_SUCCESS]\n        ).prefetch_related('user')\n        if registrations_due:\n            users = [\n                {\n                    'name': registration.user.get_full_name(),\n                    'email': registration.user.email\n                } for registration in registrations_due\n            ]\n            notification = EventPaymentOverdueCreatorNotification(\n                event.created_by, event=event, users=users\n            )\n            notification.notify()\n            log.info(\n                'event_creator_notified_of_overdue_payments', event_id=event.id,\n                creator=event.created_by\n            )\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef check_that_pool_counters_match_registration_number(self, logger_context=None):\n    \"\"\"\n    Task that checks whether pools counters are in sync with number of registrations. We do not\n    enforce this check for events that are merged, hence the merge_time filter, because\n    incrementing the counter decreases the registration performance\n    \"\"\"\n    self.setup_logger(logger_context)\n\n    events_ids = Event.objects.filter(\n        start_time__gte=timezone.now(), merge_time__gte=timezone.now()\n    ).values_list('id', flat=True)\n\n    for event_id in events_ids:\n        with transaction.atomic():\n            locked_event = Event.objects.select_for_update().get(pk=event_id)\n            for pool in locked_event.pools.all():\n                registration_count = pool.registrations.count()\n                if pool.counter != registration_count:\n                    log.critical('pool_counter_not_equal_registration_count', pool=pool)\n                    raise PoolCounterNotEqualToRegistrationCount(\n                        pool, registration_count, locked_event\n                    )\n", "path": "lego/apps/events/tasks.py"}], "after_files": [{"content": "from datetime import timedelta\n\nimport stripe\nfrom django.db import IntegrityError, transaction\nfrom django.utils import timezone\nfrom redis.exceptions import LockError\nfrom structlog import get_logger\n\nfrom lego import celery_app\nfrom lego.apps.events import constants\nfrom lego.apps.events.exceptions import (\n    EventHasClosed, PoolCounterNotEqualToRegistrationCount, WebhookDidNotFindRegistration\n)\nfrom lego.apps.events.models import Event, Registration\nfrom lego.apps.events.notifications import EventPaymentOverdueCreatorNotification\nfrom lego.apps.events.serializers.registrations import StripeObjectSerializer\nfrom lego.apps.events.websockets import (\n    notify_event_registration, notify_user_payment, notify_user_registration\n)\nfrom lego.apps.feed.registry import get_handler\nfrom lego.utils.tasks import AbakusTask\n\nlog = get_logger()\n\n\nclass AsyncRegister(AbakusTask):\n    serializer = 'json'\n    default_retry_delay = 5\n    registration = None\n\n    def on_failure(self, *args):\n        if self.request.retries == self.max_retries:\n            self.registration.status = constants.FAILURE_REGISTER\n            self.registration.save()\n            notify_user_registration(\n                constants.SOCKET_REGISTRATION_FAILURE, self.registration,\n                error_message='Registrering feilet'\n            )\n\n\nclass Payment(AbakusTask):\n    serializer = 'json'\n    default_retry_delay = 5\n    registration = None\n\n    def on_failure(self, return_value, *args):\n        if self.request.retries == self.max_retries:\n            if return_value.json_body:\n                error = return_value.json_body['error']\n                self.registration.charge_id = error['charge']\n                self.registration.charge_status = error['code']\n                self.registration.save()\n                notify_user_registration(\n                    constants.SOCKET_PAYMENT_FAILURE, self.registration,\n                    error_message=error['message']\n                )\n            else:\n                self.registration.charge_status = constants.PAYMENT_FAILURE\n                self.registration.save()\n                notify_user_registration(\n                    constants.SOCKET_PAYMENT_FAILURE, self.registration,\n                    error_message='Payment failed'\n                )\n\n\n@celery_app.task(base=AsyncRegister, bind=True)\ndef async_register(self, registration_id, logger_context=None):\n    self.setup_logger(logger_context)\n\n    self.registration = Registration.objects.get(id=registration_id)\n    try:\n        with transaction.atomic():\n            self.registration.event.register(self.registration)\n            transaction.on_commit(lambda: notify_event_registration(\n                constants.SOCKET_REGISTRATION_SUCCESS, self.registration\n            ))\n        log.info('registration_success', registration_id=self.registration.id)\n    except LockError as e:\n        log.error(\n            'registration_cache_lock_error', exception=e, registration_id=self.registration.id\n        )\n        raise self.retry(exc=e, max_retries=3)\n    except EventHasClosed as e:\n        log.warn(\n            'registration_tried_after_started', exception=e, registration_id=self.registration.id\n        )\n    except (ValueError, IntegrityError) as e:\n        log.error('registration_error', exception=e, registration_id=self.registration.id)\n        raise self.retry(exc=e, max_retries=3)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask, default_retry_delay=30)\ndef async_unregister(self, registration_id, logger_context=None):\n    self.setup_logger(logger_context)\n\n    registration = Registration.objects.get(id=registration_id)\n    pool_id = registration.pool_id\n    try:\n        with transaction.atomic():\n            registration.event.unregister(registration)\n            activation_time = registration.event.get_earliest_registration_time(registration.user)\n            transaction.on_commit(lambda: notify_event_registration(\n                constants.SOCKET_UNREGISTRATION_SUCCESS, registration,\n                from_pool=pool_id, activation_time=activation_time\n            ))\n        log.info('unregistration_success', registration_id=registration.id)\n    except EventHasClosed as e:\n        log.warn('unregistration_tried_after_started', exception=e, registration_id=registration.id)\n    except IntegrityError as e:\n        log.error('unregistration_error', exception=e, registration_id=registration.id)\n        registration.status = constants.FAILURE_UNREGISTER\n        registration.save()\n        notify_user_registration(\n            constants.SOCKET_UNREGISTRATION_FAILURE, registration,\n            error_message='Avregistrering feilet'\n        )\n\n\n@celery_app.task(base=Payment, bind=True)\ndef async_payment(self, registration_id, token, logger_context=None):\n    self.setup_logger(logger_context)\n\n    self.registration = Registration.objects.get(id=registration_id)\n    event = self.registration.event\n    try:\n        response = stripe.Charge.create(\n            amount=event.get_price(self.registration.user), currency='NOK', source=token,\n            description=event.slug, metadata={\n                'EVENT_ID': event.id,\n                'USER_ID': self.registration.user.id,\n                'USER': self.registration.user.full_name,\n                'EMAIL': self.registration.user.email,\n            }\n        )\n        log.info('stripe_payment_success', registration_id=self.registration.id)\n        return response\n    except stripe.error.CardError as e:\n        raise self.retry(exc=e)\n    except stripe.error.InvalidRequestError as e:\n        log.error('invalid_request', exception=e, registration_id=self.registration.id)\n        self.registration.charge_status = e.json_body['error']['type']\n        self.registration.save()\n        notify_user_payment(\n            constants.SOCKET_PAYMENT_FAILURE, self.registration, error_message='Invalid request'\n        )\n    except stripe.error.StripeError as e:\n        log.error('stripe_error', exception=e, registration_id=self.registration.id)\n        raise self.retry(exc=e)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef registration_payment_save(self, result, registration_id, logger_context=None):\n    self.setup_logger(logger_context)\n\n    try:\n        registration = Registration.objects.get(id=registration_id)\n        registration.charge_id = result['id']\n        registration.charge_amount = result['amount']\n        registration.charge_status = result['status']\n        registration.save()\n        notify_user_payment(\n            constants.SOCKET_PAYMENT_SUCCESS, registration, success_message='Betaling gjennomf\u00f8rt'\n        )\n    except IntegrityError as e:\n        log.error('registration_save_error', exception=e, registration_id=registration_id)\n        raise self.retry(exc=e)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef check_for_bump_on_pool_creation_or_expansion(self, event_id, logger_context=None):\n    \"\"\"Task checking for bumps when event and pools are updated\"\"\"\n    self.setup_logger(logger_context)\n\n    # Event is locked using the instance field \"is_ready\"\n    event = Event.objects.get(pk=event_id)\n    event.bump_on_pool_creation_or_expansion()\n    event.is_ready = True\n    event.save(update_fields=['is_ready'])\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef stripe_webhook_event(self, event_id, event_type, logger_context=None):\n    self.setup_logger(logger_context)\n\n    if event_type in ['charge.failed', 'charge.refunded', 'charge.succeeded']:\n        event = stripe.Event.retrieve(event_id)\n        serializer = StripeObjectSerializer(data=event.data['object'])\n        serializer.is_valid(raise_exception=True)\n\n        metadata = serializer.data['metadata']\n        registration = Registration.objects.filter(\n            event_id=metadata['EVENT_ID'], user__email=metadata['EMAIL']\n        ).first()\n        if not registration:\n            log.error('stripe_webhook_error', event_id=event_id, metadata=metadata)\n            raise WebhookDidNotFindRegistration(event_id, metadata)\n        registration.charge_id = serializer.data['id']\n        registration.charge_amount = serializer.data['amount']\n        registration.charge_amount_refunded = serializer.data['amount_refunded']\n        registration.charge_status = serializer.data['status']\n        registration.save()\n        log.info('stripe_webhook_received', event_id=event_id, registration_id=registration.id)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef check_events_for_registrations_with_expired_penalties(self, logger_context=None):\n    self.setup_logger(logger_context)\n\n    events_ids = Event.objects.filter(start_time__gte=timezone.now()\n                                      ).exclude(registrations=None).values_list('id', flat=True)\n    for event_id in events_ids:\n        with transaction.atomic():\n            locked_event = Event.objects.select_for_update().get(pk=event_id)\n            if locked_event.waiting_registrations.exists():\n                for pool in locked_event.pools.all():\n                    if pool.is_activated and not pool.is_full:\n                        for i in range(locked_event.waiting_registrations.count()):\n                            locked_event.check_for_bump_or_rebalance(pool)\n                            if pool.is_full:\n                                break\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef bump_waiting_users_to_new_pool(self, logger_context=None):\n    self.setup_logger(logger_context)\n\n    events_ids = Event.objects.filter(start_time__gte=timezone.now()\n                                      ).exclude(registrations=None).values_list('id', flat=True)\n    for event_id in events_ids:\n        with transaction.atomic():\n            locked_event = Event.objects.select_for_update().get(pk=event_id)\n            if locked_event.waiting_registrations.exists():\n                for pool in locked_event.pools.all():\n                    if not pool.is_full:\n                        act = pool.activation_date\n                        now = timezone.now()\n                        if not pool.is_activated and act < now + timedelta(minutes=35):\n                            locked_event.early_bump(pool)\n                            log.info('early_bump_executed', event_id=event_id, pool_id=pool.id)\n                        elif pool.is_activated and act > now - timedelta(minutes=35):\n                            log.info('early_bump_executed', event_id=event_id, pool_id=pool.id)\n                            locked_event.early_bump(pool)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef notify_user_when_payment_soon_overdue(self, logger_context=None):\n    self.setup_logger(logger_context)\n\n    time = timezone.now()\n    events = Event.objects.filter(\n        payment_due_date__range=(time - timedelta(days=2), time + timedelta(days=3)),\n        is_priced=True, use_stripe=True\n    ).exclude(registrations=None).prefetch_related('registrations')\n    for event in events:\n        for registration in event.registrations.exclude(pool=None):\n            if registration.should_notify(time):\n                log.info(\n                    'registration_notified_overdue_payment', event_id=event.id,\n                    registration_id=registration.id\n                )\n                get_handler(Registration).handle_payment_overdue(registration)\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef notify_event_creator_when_payment_overdue(self, logger_context=None):\n    self.setup_logger(logger_context)\n\n    time = timezone.now()\n    events = Event.objects.filter(\n        payment_due_date__lte=time, is_priced=True, use_stripe=True, end_time__gte=time\n    ).exclude(registrations=None).prefetch_related('registrations')\n    for event in events:\n        registrations_due = event.registrations.exclude(pool=None).exclude(\n            charge_status__in=[constants.PAYMENT_MANUAL, constants.PAYMENT_SUCCESS]\n        ).prefetch_related('user')\n        if registrations_due:\n            users = [\n                {\n                    'name': registration.user.get_full_name(),\n                    'email': registration.user.email\n                } for registration in registrations_due\n            ]\n            notification = EventPaymentOverdueCreatorNotification(\n                event.created_by, event=event, users=users\n            )\n            notification.notify()\n            log.info(\n                'event_creator_notified_of_overdue_payments', event_id=event.id,\n                creator=event.created_by\n            )\n\n\n@celery_app.task(serializer='json', bind=True, base=AbakusTask)\ndef check_that_pool_counters_match_registration_number(self, logger_context=None):\n    \"\"\"\n    Task that checks whether pools counters are in sync with number of registrations. We do not\n    enforce this check for events that are merged, hence the merge_time filter, because\n    incrementing the counter decreases the registration performance\n    \"\"\"\n    self.setup_logger(logger_context)\n\n    events_ids = Event.objects.filter(\n        start_time__gte=timezone.now(), merge_time__gte=timezone.now()\n    ).values_list('id', flat=True)\n\n    for event_id in events_ids:\n        with transaction.atomic():\n            locked_event = Event.objects.select_for_update().get(pk=event_id)\n            for pool in locked_event.pools.all():\n                registration_count = pool.registrations.count()\n                if pool.counter != registration_count:\n                    log.critical('pool_counter_not_equal_registration_count', pool=pool)\n                    raise PoolCounterNotEqualToRegistrationCount(\n                        pool, registration_count, locked_event\n                    )\n", "path": "lego/apps/events/tasks.py"}]}