fix: subscription webhook correctness

- Emit subscription.renewed from scanner when advance_subscription_period
  succeeds (was only in hourly job fallback, which rarely runs)
- Fix subscription.canceled spam: select subs before UPDATE, dispatch
  webhooks only for those rows (was re-dispatching for all historically
  canceled subs every hour)
- Emit subscription.canceled on immediate cancel (at_period_end: false)
  in API handler with "immediate": true in payload
- Remove duplicate subscription.renewed from hourly job since scanner
  now handles it

Author: Kenbak

src/api/subscriptions.rs

@@ -2,6 +2,7 @@ use actix_web::{web, HttpRequest, HttpResponse};
 use sqlx::SqlitePool;
 
 use crate::subscriptions::{self, CreateSubscriptionRequest};
+use crate::config::Config;
 
 pub async fn create(
     req: HttpRequest,
@@ -42,6 +43,7 @@ pub struct CancelBody {
 pub async fn cancel(
     req: HttpRequest,
     pool: web::Data<SqlitePool>,
+    config: web::Data<Config>,
     path: web::Path<String>,
     body: web::Json<CancelBody>,
 ) -> HttpResponse {
@@ -56,7 +58,28 @@ pub async fn cancel(
     match subscriptions::cancel_subscription(pool.get_ref(), &sub_id, &merchant.id, at_period_end)
         .await
     {
-        Ok(Some(sub)) => HttpResponse::Ok().json(sub),
+        Ok(Some(sub)) => {
+            // Dispatch webhook for immediate cancels (at_period_end=false and status is now canceled)
+            // Period-end cancels are handled by the hourly process_renewals job when they actually cancel
+            if !at_period_end && sub.status == "canceled" {
+                let http = reqwest::Client::new();
+                let payload = serde_json::json!({
+                    "subscription_id": sub.id,
+                    "price_id": sub.price_id,
+                    "immediate": true,
+                });
+                let _ = crate::webhooks::dispatch_event(
+                    pool.get_ref(),
+                    &http,
+                    &merchant.id,
+                    "subscription.canceled",
+                    payload,
+                    &config.encryption_key,
+                )
+                .await;
+            }
+            HttpResponse::Ok().json(sub)
+        }
         Ok(None) => {
             HttpResponse::NotFound().json(serde_json::json!({"error": "Subscription not found"}))
         }

src/scanner/mod.rs

@@ -715,6 +715,25 @@ async fn on_invoice_confirmed(
                     new_period_end = %sub.current_period_end,
                     "Subscription advanced on payment confirmation"
                 );
+                // Dispatch subscription.renewed webhook
+                let payload = serde_json::json!({
+                    "subscription_id": sub.id,
+                    "invoice_id": invoice.id,
+                    "new_period_start": sub.current_period_start,
+                    "new_period_end": sub.current_period_end,
+                });
+                if let Err(e) = crate::webhooks::dispatch_event(
+                    pool,
+                    http,
+                    &invoice.merchant_id,
+                    "subscription.renewed",
+                    payload,
+                    &config.encryption_key,
+                )
+                .await
+                {
+                    tracing::error!(sub_id, error = %e, "Failed to dispatch subscription.renewed webhook");
+                }
             }
             Ok(None) => {
                 tracing::warn!(sub_id, "Subscription not found for confirmed invoice");

src/subscriptions/mod.rs

@@ -175,28 +175,33 @@ pub async fn process_renewals(
     let mut actions = 0u32;
 
     // 1. Cancel subscriptions marked for end-of-period cancellation
-    let canceled = sqlx::query(
-        "UPDATE subscriptions SET status = 'canceled'
-         WHERE cancel_at_period_end = 1 AND current_period_end <= ? AND status = 'active'",
-    )
-    .bind(&now_str)
-    .execute(pool)
-    .await?;
+    // First, select the ones we're about to cancel (before updating) so we dispatch webhooks only for these
+    let q = format!(
+        "SELECT {} FROM subscriptions WHERE cancel_at_period_end = 1 AND current_period_end <= ? AND status = 'active'",
+        SUB_COLS
+    );
+    let to_cancel: Vec<Subscription> = sqlx::query_as::<_, Subscription>(&q)
+        .bind(&now_str)
+        .fetch_all(pool)
+        .await?;
+
+    if !to_cancel.is_empty() {
+        // Now update them
+        sqlx::query(
+            "UPDATE subscriptions SET status = 'canceled'
+             WHERE cancel_at_period_end = 1 AND current_period_end <= ? AND status = 'active'",
+        )
+        .bind(&now_str)
+        .execute(pool)
+        .await?;
 
-    if canceled.rows_affected() > 0 {
         tracing::info!(
-            count = canceled.rows_affected(),
+            count = to_cancel.len(),
             "Subscriptions canceled at period end"
         );
-        // Fire webhooks for canceled subscriptions
-        let q = format!(
-            "SELECT {} FROM subscriptions WHERE status = 'canceled' AND cancel_at_period_end = 1",
-            SUB_COLS
-        );
-        let canceled_subs: Vec<Subscription> = sqlx::query_as::<_, Subscription>(&q)
-            .fetch_all(pool)
-            .await?;
-        for sub in &canceled_subs {
+
+        // Fire webhooks only for the subscriptions we just canceled
+        for sub in &to_cancel {
             let payload = serde_json::json!({
                 "subscription_id": sub.id,
                 "price_id": sub.price_id,
@@ -211,7 +216,7 @@ pub async fn process_renewals(
             )
             .await;
         }
-        actions += canceled.rows_affected() as u32;
+        actions += to_cancel.len() as u32;
     }
 
     // 2. Generate draft invoices for active subscriptions approaching period end
@@ -326,6 +331,8 @@ pub async fn process_renewals(
     }
 
     // 3. Advance paid periods (subscriptions past period_end with confirmed invoice)
+    // Note: subscription.renewed webhook is dispatched by the scanner on payment confirmation.
+    // This step is a fallback for edge cases (e.g., server restart during scan).
     let q = format!(
         "SELECT {} FROM subscriptions WHERE status = 'active' AND current_period_end <= ? AND cancel_at_period_end = 0",
         SUB_COLS
@@ -346,21 +353,8 @@ pub async fn process_renewals(
 
                 if let Some((ref status,)) = inv_status {
                     if status == "confirmed" {
-                        if let Some(new_sub) = advance_subscription_period(pool, &sub.id).await? {
-                            let payload = serde_json::json!({
-                                "subscription_id": new_sub.id,
-                                "new_period_start": new_sub.current_period_start,
-                                "new_period_end": new_sub.current_period_end,
-                            });
-                            let _ = crate::webhooks::dispatch_event(
-                                pool,
-                                http,
-                                &sub.merchant_id,
-                                "subscription.renewed",
-                                payload,
-                                encryption_key,
-                            )
-                            .await;
+                        // Just advance the period — webhook already dispatched by scanner
+                        if advance_subscription_period(pool, &sub.id).await?.is_some() {
                             actions += 1;
                         }
                         continue;