Add Stripe event deduplication to prevent duplicate notifications
- Added PROCESSED_EVENTS_FILE to track processed Stripe event IDs
- Added is_event_processed() to check if an event was already handled
- Added mark_event_processed() to record processed events
- Webhook handler now checks for duplicate events before processing
- _process_checkout_session marks event as processed immediately
- Duplicate events now return {'status': 'duplicate'} instead of reprocessing
Fixes phantom notification bug where Stripe retry webhooks caused duplicate Telegram messages.
This commit is contained in:
+53
-3
@@ -30,6 +30,46 @@ from update_ledger import add_donation as ledger_add_donation
|
|||||||
STRIPE_SECRET_KEY = os.environ.get('STRIPE_SECRET_KEY', '')
|
STRIPE_SECRET_KEY = os.environ.get('STRIPE_SECRET_KEY', '')
|
||||||
WEBHOOK_SECRET = os.environ.get('STRIPE_WEBHOOK_SECRET', '') # whsec_... when Stripe is configured
|
WEBHOOK_SECRET = os.environ.get('STRIPE_WEBHOOK_SECRET', '') # whsec_... when Stripe is configured
|
||||||
OUTPUT_DIR = os.environ.get('FJCCV_RECEIPTS_DIR', '/root/fjccv-receipts')
|
OUTPUT_DIR = os.environ.get('FJCCV_RECEIPTS_DIR', '/root/fjccv-receipts')
|
||||||
|
PROCESSED_EVENTS_FILE = os.environ.get('PROCESSED_EVENTS_FILE', '/root/fjccv-receipts/processed_events.json')
|
||||||
|
|
||||||
|
# Thread-safe lock for file operations
|
||||||
|
import threading
|
||||||
|
processed_events_lock = threading.Lock()
|
||||||
|
|
||||||
|
def is_event_processed(event_id: str) -> bool:
|
||||||
|
"""Check if a Stripe event has already been processed."""
|
||||||
|
if not event_id:
|
||||||
|
return False
|
||||||
|
try:
|
||||||
|
with processed_events_lock:
|
||||||
|
if not os.path.exists(PROCESSED_EVENTS_FILE):
|
||||||
|
return False
|
||||||
|
with open(PROCESSED_EVENTS_FILE, 'r') as f:
|
||||||
|
processed = json.load(f)
|
||||||
|
return event_id in processed
|
||||||
|
except (json.JSONDecodeError, IOError) as e:
|
||||||
|
app.logger.warning(f"Error reading processed events file: {e}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
def mark_event_processed(event_id: str) -> None:
|
||||||
|
"""Mark a Stripe event as processed."""
|
||||||
|
if not event_id:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
with processed_events_lock:
|
||||||
|
processed = set()
|
||||||
|
if os.path.exists(PROCESSED_EVENTS_FILE):
|
||||||
|
try:
|
||||||
|
with open(PROCESSED_EVENTS_FILE, 'r') as f:
|
||||||
|
processed = set(json.load(f))
|
||||||
|
except (json.JSONDecodeError, IOError):
|
||||||
|
pass
|
||||||
|
processed.add(event_id)
|
||||||
|
os.makedirs(os.path.dirname(PROCESSED_EVENTS_FILE), exist_ok=True)
|
||||||
|
with open(PROCESSED_EVENTS_FILE, 'w') as f:
|
||||||
|
json.dump(list(processed), f)
|
||||||
|
except IOError as e:
|
||||||
|
app.logger.warning(f"Error writing processed events file: {e}")
|
||||||
|
|
||||||
# Sami's PC for receipt delivery
|
# Sami's PC for receipt delivery
|
||||||
SAMIPC_HOST = os.environ.get('SAMIPC_HOST', 'hello@100.85.236.10')
|
SAMIPC_HOST = os.environ.get('SAMIPC_HOST', 'hello@100.85.236.10')
|
||||||
@@ -287,7 +327,13 @@ def stripe_webhook():
|
|||||||
abort(400, 'Invalid JSON')
|
abort(400, 'Invalid JSON')
|
||||||
|
|
||||||
event_type = event.get('type', '')
|
event_type = event.get('type', '')
|
||||||
app.logger.info(f"Received event: {event_type}")
|
event_id = event.get('id', '')
|
||||||
|
app.logger.info(f"Received event: {event_type} (id: {event_id})")
|
||||||
|
|
||||||
|
# Check for duplicate events
|
||||||
|
if event_id and is_event_processed(event_id):
|
||||||
|
app.logger.info(f"Duplicate event skipped: {event_id}")
|
||||||
|
return jsonify({'received': True, 'status': 'duplicate'})
|
||||||
|
|
||||||
# Handle checkout.session.completed (Stripe Checkout Payment Links)
|
# Handle checkout.session.completed (Stripe Checkout Payment Links)
|
||||||
if event_type == 'checkout.session.completed':
|
if event_type == 'checkout.session.completed':
|
||||||
@@ -322,7 +368,7 @@ def stripe_webhook():
|
|||||||
# This is important because Stripe requires a quick 2xx response
|
# This is important because Stripe requires a quick 2xx response
|
||||||
thread = threading.Thread(
|
thread = threading.Thread(
|
||||||
target=_process_checkout_session,
|
target=_process_checkout_session,
|
||||||
args=(session, donor_name, donor_email, amount, transaction_id, date_str, year)
|
args=(session, event_id, donor_name, donor_email, amount, transaction_id, date_str, year)
|
||||||
)
|
)
|
||||||
thread.start()
|
thread.start()
|
||||||
|
|
||||||
@@ -332,9 +378,13 @@ def stripe_webhook():
|
|||||||
return jsonify({'received': True})
|
return jsonify({'received': True})
|
||||||
|
|
||||||
|
|
||||||
def _process_checkout_session(session, donor_name, donor_email, amount, transaction_id, date_str, year):
|
def _process_checkout_session(session, event_id, donor_name, donor_email, amount, transaction_id, date_str, year):
|
||||||
"""Background processing for checkout.session.completed events."""
|
"""Background processing for checkout.session.completed events."""
|
||||||
try:
|
try:
|
||||||
|
# Mark event as processed immediately to prevent duplicate processing
|
||||||
|
if event_id:
|
||||||
|
mark_event_processed(event_id)
|
||||||
|
|
||||||
app.logger.info(f"Processing donation: {donor_name}, ${amount:.2f}, {date_str}, tx: {transaction_id}")
|
app.logger.info(f"Processing donation: {donor_name}, ${amount:.2f}, {date_str}, tx: {transaction_id}")
|
||||||
|
|
||||||
# Get donation number
|
# Get donation number
|
||||||
|
|||||||
Reference in New Issue
Block a user