Supabase Webhooks: Integrate External Services

-- Enable pg_net extension for HTTP requests
create extension if not exists pg_net;

-- Create webhook log table
create table webhook_logs (
  id uuid default gen_random_uuid() primary key,
  event_type text not null,
  payload jsonb not null,
  response jsonb,
  status int,
  created_at timestamp with time zone default now()
);

-- Webhook function for new user signups
create or replace function notify_new_user()
returns trigger as $$
declare
  webhook_url text := 'https://your-app.com/api/webhooks/new-user';
  payload jsonb;
  request_id bigint;
begin
  -- Prepare payload
  payload := jsonb_build_object(
    'event', 'user.created',
    'timestamp', now(),
    'data', jsonb_build_object(
      'id', new.id,
      'email', new.email,
      'created_at', new.created_at
    )
  );

  -- Make HTTP request
  select net.http_post(
    url := webhook_url,
    headers := jsonb_build_object(
      'Content-Type', 'application/json',
      'Authorization', 'Bearer your-secret-key'
    ),
    body := payload
  ) into request_id;

  -- Log webhook call
  insert into webhook_logs (event_type, payload)
  values ('user.created', payload);

  return new;
end;
$$ language plpgsql security definer;

-- Create trigger
create trigger on_user_created
  after insert on auth.users
  for each row
  execute function notify_new_user();

-- Webhook for post updates
create or replace function notify_post_update()
returns trigger as $$
declare
  webhook_url text := 'https://your-app.com/api/webhooks/post-updated';
  payload jsonb;
begin
  payload := jsonb_build_object(
    'event', 'post.updated',
    'timestamp', now(),
    'old_data', row_to_json(old),
    'new_data', row_to_json(new)
  );

  perform net.http_post(
    url := webhook_url,
    headers := jsonb_build_object(
      'Content-Type', 'application/json',
      'X-Webhook-Secret', current_setting('app.webhook_secret')
    ),
    body := payload
  );

  return new;
end;
$$ language plpgsql security definer;

create trigger on_post_updated
  after update on posts
  for each row
  execute function notify_post_update();

-- Batch webhook for multiple records
create or replace function notify_bulk_delete()
returns trigger as $$
declare
  webhook_url text := 'https://your-app.com/api/webhooks/bulk-delete';
  payload jsonb;
begin
  payload := jsonb_build_object(
    'event', 'posts.deleted',
    'timestamp', now(),
    'count', tg_nargs,
    'deleted_ids', array_agg(old.id)
  );

  perform net.http_post(
    url := webhook_url,
    body := payload
  );

  return old;
end;
$$ language plpgsql security definer;

-- Configure webhook secret
alter database postgres set app.webhook_secret = 'your-secret-key';

// supabase/functions/webhook-handler/index.ts
import { serve } from 'https://deno.land/[email protected]/http/server.ts'
import { createClient } from 'https://esm.sh/@supabase/supabase-js@2'

const WEBHOOK_SECRET = Deno.env.get('WEBHOOK_SECRET')!

interface WebhookPayload {
  event: string
  timestamp: string
  data: any
}

serve(async (req) => {
  try {
    // Verify webhook signature
    const signature = req.headers.get('x-webhook-secret')
    if (signature !== WEBHOOK_SECRET) {
      return new Response(
        JSON.stringify({ error: 'Invalid signature' }),
        { status: 401 }
      )
    }

    const payload: WebhookPayload = await req.json()
    console.log('Webhook received:', payload.event)

    // Route to appropriate handler
    switch (payload.event) {
      case 'user.created':
        await handleUserCreated(payload.data)
        break
      case 'post.updated':
        await handlePostUpdated(payload.data)
        break
      case 'payment.succeeded':
        await handlePaymentSuccess(payload.data)
        break
      default:
        console.log('Unknown event:', payload.event)
    }

    return new Response(
      JSON.stringify({ success: true }),
      { status: 200 }
    )
  } catch (error) {
    console.error('Webhook error:', error)
    return new Response(
      JSON.stringify({ error: error.message }),
      { status: 500 }
    )
  }
})

async function handleUserCreated(data: any) {
  console.log('New user:', data.email)
  
  // Send welcome email
  await fetch('https://api.sendgrid.com/v3/mail/send', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${Deno.env.get('SENDGRID_API_KEY')}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      personalizations: [{
        to: [{ email: data.email }],
        subject: 'Welcome!',
      }],
      from: { email: '[email protected]' },
      content: [{
        type: 'text/html',
        value: '<h1>Welcome to our app!</h1>',
      }],
    }),
  })

  // Track in analytics
  await fetch('https://api.segment.com/v1/track', {
    method: 'POST',
    headers: {
      'Authorization': `Basic ${btoa(Deno.env.get('SEGMENT_WRITE_KEY') + ':')}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      userId: data.id,
      event: 'User Signed Up',
      properties: {
        email: data.email,
        created_at: data.created_at,
      },
    }),
  })
}

async function handlePostUpdated(data: any) {
  const { old_data, new_data } = data
  
  // Notify via Slack if post published
  if (!old_data.published && new_data.published) {
    await fetch(Deno.env.get('SLACK_WEBHOOK_URL')!, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        text: `📝 New post published: ${new_data.title}`,
        blocks: [
          {
            type: 'section',
            text: {
              type: 'mrkdwn',
              text: `*${new_data.title}*\n${new_data.content.substring(0, 100)}...`,
            },
          },
        ],
      }),
    })
  }
}

async function handlePaymentSuccess(data: any) {
  const supabase = createClient(
    Deno.env.get('SUPABASE_URL')!,
    Deno.env.get('SUPABASE_SERVICE_ROLE_KEY')!
  )

  // Update user subscription
  await supabase
    .from('subscriptions')
    .update({
      status: 'active',
      current_period_end: data.current_period_end,
    })
    .eq('stripe_customer_id', data.customer)

  console.log('Subscription activated:', data.customer)
}

// supabase/functions/stripe-webhook/index.ts
import { serve } from 'https://deno.land/[email protected]/http/server.ts'
import Stripe from 'https://esm.sh/[email protected]?target=deno'
import { createClient } from 'https://esm.sh/@supabase/supabase-js@2'

const stripe = new Stripe(Deno.env.get('STRIPE_SECRET_KEY')!, {
  apiVersion: '2023-10-16',
  httpClient: Stripe.createFetchHttpClient(),
})

const supabase = createClient(
  Deno.env.get('SUPABASE_URL')!,
  Deno.env.get('SUPABASE_SERVICE_ROLE_KEY')!
)

serve(async (req) => {
  const signature = req.headers.get('stripe-signature')!
  const body = await req.text()

  let event: Stripe.Event

  try {
    // Verify webhook signature
    event = stripe.webhooks.constructEvent(
      body,
      signature,
      Deno.env.get('STRIPE_WEBHOOK_SECRET')!
    )
  } catch (err) {
    console.error('Webhook signature verification failed:', err.message)
    return new Response(
      JSON.stringify({ error: 'Invalid signature' }),
      { status: 400 }
    )
  }

  console.log('Stripe event:', event.type)

  try {
    switch (event.type) {
      case 'checkout.session.completed':
        await handleCheckoutCompleted(event.data.object)
        break

      case 'customer.subscription.created':
      case 'customer.subscription.updated':
        await handleSubscriptionChange(event.data.object)
        break

      case 'customer.subscription.deleted':
        await handleSubscriptionDeleted(event.data.object)
        break

      case 'invoice.payment_succeeded':
        await handlePaymentSucceeded(event.data.object)
        break

      case 'invoice.payment_failed':
        await handlePaymentFailed(event.data.object)
        break

      default:
        console.log('Unhandled event type:', event.type)
    }

    return new Response(JSON.stringify({ received: true }), { status: 200 })
  } catch (error) {
    console.error('Error processing webhook:', error)
    return new Response(
      JSON.stringify({ error: error.message }),
      { status: 500 }
    )
  }
})

async function handleCheckoutCompleted(session: Stripe.Checkout.Session) {
  const userId = session.client_reference_id
  const customerId = session.customer as string

  // Create subscription record
  await supabase.from('subscriptions').insert({
    user_id: userId,
    stripe_customer_id: customerId,
    stripe_subscription_id: session.subscription,
    status: 'active',
    price_id: session.line_items?.data[0]?.price?.id,
  })

  console.log('Checkout completed for user:', userId)
}

async function handleSubscriptionChange(subscription: Stripe.Subscription) {
  await supabase
    .from('subscriptions')
    .upsert({
      stripe_subscription_id: subscription.id,
      stripe_customer_id: subscription.customer as string,
      status: subscription.status,
      price_id: subscription.items.data[0].price.id,
      current_period_start: new Date(subscription.current_period_start * 1000).toISOString(),
      current_period_end: new Date(subscription.current_period_end * 1000).toISOString(),
      cancel_at_period_end: subscription.cancel_at_period_end,
    })

  console.log('Subscription updated:', subscription.id)
}

async function handleSubscriptionDeleted(subscription: Stripe.Subscription) {
  await supabase
    .from('subscriptions')
    .update({ status: 'canceled' })
    .eq('stripe_subscription_id', subscription.id)

  console.log('Subscription canceled:', subscription.id)
}

async function handlePaymentSucceeded(invoice: Stripe.Invoice) {
  // Log payment
  await supabase.from('payments').insert({
    stripe_invoice_id: invoice.id,
    stripe_customer_id: invoice.customer as string,
    amount: invoice.amount_paid,
    currency: invoice.currency,
    status: 'succeeded',
  })

  console.log('Payment succeeded:', invoice.id)
}

async function handlePaymentFailed(invoice: Stripe.Invoice) {
  // Notify user of failed payment
  const { data: subscription } = await supabase
    .from('subscriptions')
    .select('user_id')
    .eq('stripe_customer_id', invoice.customer)
    .single()

  if (subscription) {
    // Send notification
    await supabase.from('notifications').insert({
      user_id: subscription.user_id,
      type: 'payment_failed',
      message: 'Your payment failed. Please update your payment method.',
    })
  }

  console.log('Payment failed:', invoice.id)
}

// Webhook signature verification
import crypto from 'crypto'

function verifyWebhookSignature(
  payload: string,
  signature: string,
  secret: string
): boolean {
  const hmac = crypto.createHmac('sha256', secret)
  const expectedSignature = hmac.update(payload).digest('hex')
  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expectedSignature)
  )
}

// Generate signature for outgoing webhooks
function generateWebhookSignature(payload: string, secret: string): string {
  const hmac = crypto.createHmac('sha256', secret)
  return hmac.update(payload).digest('hex')
}

// Edge Function with signature verification
serve(async (req) => {
  const body = await req.text()
  const signature = req.headers.get('x-webhook-signature')
  const timestamp = req.headers.get('x-webhook-timestamp')

  // Verify timestamp to prevent replay attacks
  const requestTime = parseInt(timestamp || '0')
  const currentTime = Math.floor(Date.now() / 1000)
  const timeDifference = Math.abs(currentTime - requestTime)

  if (timeDifference > 300) { // 5 minutes
    return new Response(
      JSON.stringify({ error: 'Request timestamp too old' }),
      { status: 401 }
    )
  }

  // Verify signature
  const isValid = verifyWebhookSignature(
    `${timestamp}.${body}`,
    signature!,
    Deno.env.get('WEBHOOK_SECRET')!
  )

  if (!isValid) {
    return new Response(
      JSON.stringify({ error: 'Invalid signature' }),
      { status: 401 }
    )
  }

  // Process webhook
  const payload = JSON.parse(body)
  // ... handle payload

  return new Response(JSON.stringify({ success: true }))
})

// IP whitelist verification
const ALLOWED_IPS = [
  '192.168.1.1',
  '10.0.0.1',
]

function verifyIPAddress(req: Request): boolean {
  const clientIP = req.headers.get('x-forwarded-for') || 'unknown'
  return ALLOWED_IPS.includes(clientIP)
}

// Rate limiting for webhooks
const rateLimits = new Map<string, number[]>()

function checkRateLimit(identifier: string, limit = 100, windowMs = 60000): boolean {
  const now = Date.now()
  const requests = rateLimits.get(identifier) || []
  
  // Remove old requests outside window
  const validRequests = requests.filter(time => now - time < windowMs)
  
  if (validRequests.length >= limit) {
    return false
  }
  
  validRequests.push(now)
  rateLimits.set(identifier, validRequests)
  return true
}

-- Webhook retry function with exponential backoff
create or replace function send_webhook_with_retry(
  webhook_url text,
  payload jsonb,
  max_retries int default 3
)
returns void as $$
declare
  retry_count int := 0;
  request_id bigint;
  response_status int;
  backoff_seconds int;
begin
  loop
    -- Calculate exponential backoff: 2^retry_count seconds
    backoff_seconds := power(2, retry_count);
    
    -- Wait before retry (except first attempt)
    if retry_count > 0 then
      perform pg_sleep(backoff_seconds);
    end if;
    
    -- Make HTTP request
    select net.http_post(
      url := webhook_url,
      headers := jsonb_build_object('Content-Type', 'application/json'),
      body := payload,
      timeout_milliseconds := 5000
    ) into request_id;
    
    -- Check response (simplified - actual implementation would query net.http_get_result)
    -- If successful, exit loop
    -- if response_status between 200 and 299 then
    --   exit;
    -- end if;
    
    retry_count := retry_count + 1;
    
    -- Max retries reached
    if retry_count >= max_retries then
      -- Log failure
      insert into webhook_logs (event_type, payload, status)
      values ('webhook.failed', payload, -1);
      exit;
    end if;
  end loop;
end;
$$ language plpgsql;

// TypeScript retry logic for client-side webhooks
async function sendWebhookWithRetry(
  url: string,
  payload: any,
  maxRetries = 3
): Promise<Response> {
  let lastError: Error

  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const response = await fetch(url, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
        },
        body: JSON.stringify(payload),
      })

      // Consider 2xx and 3xx as success
      if (response.ok || response.status < 400) {
        return response
      }

      // Don't retry client errors (4xx)
      if (response.status >= 400 && response.status < 500) {
        throw new Error(`Client error: ${response.status}`)
      }

      // Retry server errors (5xx)
      lastError = new Error(`Server error: ${response.status}`)
    } catch (error) {
      lastError = error as Error
      console.error(`Attempt ${attempt + 1} failed:`, error)
    }

    // Exponential backoff: 2^attempt seconds
    if (attempt < maxRetries - 1) {
      const backoffMs = Math.pow(2, attempt) * 1000
      console.log(`Retrying in ${backoffMs}ms...`)
      await new Promise(resolve => setTimeout(resolve, backoffMs))
    }
  }

  throw new Error(`Webhook failed after ${maxRetries} attempts: ${lastError.message}`)
}

// Dead letter queue for failed webhooks
create table webhook_dlq (
  id uuid default gen_random_uuid() primary key,
  webhook_url text not null,
  payload jsonb not null,
  error_message text,
  attempts int default 0,
  last_attempt_at timestamp with time zone,
  created_at timestamp with time zone default now()
);

-- Function to process failed webhooks from DLQ
create or replace function process_webhook_dlq()
returns void as $$
declare
  webhook record;
begin
  for webhook in 
    select * from webhook_dlq 
    where attempts < 10 
      and (last_attempt_at is null or last_attempt_at < now() - interval '1 hour')
    limit 100
  loop
    begin
      perform send_webhook_with_retry(webhook.webhook_url, webhook.payload);
      
      -- Success - remove from DLQ
      delete from webhook_dlq where id = webhook.id;
    exception when others then
      -- Update failure count
      update webhook_dlq
      set attempts = attempts + 1,
          last_attempt_at = now(),
          error_message = sqlerrm
      where id = webhook.id;
    end;
  end loop;
end;
$$ language plpgsql;

-- Schedule DLQ processing
select cron.schedule(
  'process-webhook-dlq',
  '*/30 * * * *',  -- Every 30 minutes
  $$select process_webhook_dlq()$$
);

-- Webhook monitoring view
create view webhook_stats as
select 
  event_type,
  count(*) as total_calls,
  count(*) filter (where status between 200 and 299) as successful,
  count(*) filter (where status >= 400) as failed,
  avg(extract(epoch from (created_at - lag(created_at) over (partition by event_type order by created_at)))) as avg_interval_seconds
from webhook_logs
where created_at > now() - interval '24 hours'
group by event_type;

-- Query webhook statistics
select * from webhook_stats;

// Dashboard API endpoint
import { serve } from 'https://deno.land/[email protected]/http/server.ts'
import { createClient } from 'https://esm.sh/@supabase/supabase-js@2'

serve(async (req) => {
  const supabase = createClient(
    Deno.env.get('SUPABASE_URL')!,
    Deno.env.get('SUPABASE_SERVICE_ROLE_KEY')!
  )

  // Get webhook statistics
  const { data: stats } = await supabase
    .from('webhook_stats')
    .select('*')

  // Get recent failures
  const { data: failures } = await supabase
    .from('webhook_logs')
    .select('*')
    .gte('status', 400)
    .order('created_at', { ascending: false })
    .limit(10)

  // Get DLQ count
  const { count: dlqCount } = await supabase
    .from('webhook_dlq')
    .select('*', { count: 'exact', head: true })

  return new Response(
    JSON.stringify({
      stats,
      recentFailures: failures,
      dlqCount,
    }),
    {
      headers: { 'Content-Type': 'application/json' },
    }
  )
})

-- Alert on high failure rate
create or replace function check_webhook_health()
returns void as $$
declare
  failure_rate float;
begin
  select 
    count(*) filter (where status >= 400)::float / nullif(count(*), 0)
  into failure_rate
  from webhook_logs
  where created_at > now() - interval '1 hour';
  
  if failure_rate > 0.1 then  -- 10% failure rate
    -- Send alert
    perform net.http_post(
      url := 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL',
      body := jsonb_build_object(
        'text', format('⚠️ Webhook failure rate: %.1f%%', failure_rate * 100)
      )
    );
  end if;
end;
$$ language plpgsql;

-- Run health check every 5 minutes
select cron.schedule(
  'webhook-health-check',
  '*/5 * * * *',
  $$select check_webhook_health()$$
);