tasq/supabase/functions/process_scheduled_notifications/index.ts

import { createClient } from 'npm:@supabase/supabase-js@2'

// Minimal Deno Edge Function to process queued scheduled_notifications
// Environment variables required:
// SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY, SEND_FCM_URL

const corsHeaders = {
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Headers': 'authorization, x-client-info, apikey, content-type',
}

const supabase = createClient(Deno.env.get('SUPABASE_URL')!, Deno.env.get('SUPABASE_SERVICE_ROLE_KEY')!)
const SEND_FCM_URL = Deno.env.get('SEND_FCM_URL')!
const BATCH_SIZE = Number(Deno.env.get('PROCESSOR_BATCH_SIZE') || '50')

// deterministic UUIDv5-like from a name using SHA-1
async function uuidFromName(name: string): Promise<string> {
  const encoder = new TextEncoder()
  const data = encoder.encode(name)
  const hashBuffer = await crypto.subtle.digest('SHA-1', data)
  const hash = new Uint8Array(hashBuffer)
  // use first 16 bytes of SHA-1
  const bytes = hash.slice(0, 16)
  // set version (5) and variant (RFC 4122)
  bytes[6] = (bytes[6] & 0x0f) | 0x50 // version 5
  bytes[8] = (bytes[8] & 0x3f) | 0x80 // variant
  const hex = Array.from(bytes).map(b => b.toString(16).padStart(2, '0')).join('')
  return `${hex.slice(0,8)}-${hex.slice(8,12)}-${hex.slice(12,16)}-${hex.slice(16,20)}-${hex.slice(20,32)}`
}

async function processBatch() {
  const nowIso = new Date().toISOString()
  const { data: rows, error } = await supabase
    .from('scheduled_notifications')
    .select('*')
    .eq('processed', false)
    .lte('scheduled_for', nowIso)
    .order('scheduled_for', { ascending: true })
    .limit(BATCH_SIZE)

  if (error) {
    console.error('Failed to fetch scheduled_notifications', error)
    return
  }
  if (!rows || rows.length === 0) {
    console.log('No scheduled rows to process')
    return
  }

  for (const r of rows) {
    try {
      const scheduleId = r.schedule_id
      const userId = r.user_id
      const notifyType = r.notify_type
      const rowId = r.id

      const notificationId = await uuidFromName(`${scheduleId}-${userId}-${notifyType}`)

      // Attempt to mark idempotent push
      const { data: markData, error: markErr } = await supabase.rpc('try_mark_notification_pushed', { p_notification_id: notificationId })
      if (markErr) {
        console.warn('try_mark_notification_pushed error', markErr)
        // do not mark processed; increment retry
        await supabase.from('scheduled_notifications').update({ retry_count: r.retry_count + 1, last_error: String(markErr) }).eq('id', rowId)
        continue
      }

      if (markData === false) {
        console.log('Notification already pushed, skipping', notificationId)
        await supabase.from('scheduled_notifications').update({ processed: true, processed_at: new Date().toISOString() }).eq('id', rowId)
        continue
      }

      // Prepare message
      let title = ''
      let body = ''
      if (notifyType === 'start_15') {
        title = 'Shift starting soon'
        body = 'Your shift starts in 15 minutes. Don\'t forget to check in.'
      } else if (notifyType === 'end') {
        title = 'Shift ended'
        body = 'Your shift has ended. Please remember to check out if you haven\'t.'
      } else {
        title = 'Shift reminder'
        body = 'Reminder about your shift.'
      }

      // Call send_fcm endpoint to deliver push (reuses existing implementation)
      const payload = {
        user_ids: [userId],
        title,
        body,
        data: {
          notification_id: notificationId,
          schedule_id: scheduleId,
          type: notifyType,
        },
      }

      const res = await fetch(SEND_FCM_URL, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify(payload),
      })

      if (!res.ok) {
        const text = await res.text().catch(() => '')
        console.error('send_fcm failed', res.status, text)
        await supabase.from('scheduled_notifications').update({ retry_count: r.retry_count + 1, last_error: `send_fcm ${res.status}: ${text}` }).eq('id', rowId)
        continue
      }

      // Mark processed
      await supabase.from('scheduled_notifications').update({ processed: true, processed_at: new Date().toISOString() }).eq('id', rowId)
      console.log('Processed scheduled notification', rowId, notificationId)
    } catch (err) {
      console.error('Error processing row', r, err)
      try {
        await supabase.from('scheduled_notifications').update({ retry_count: r.retry_count + 1, last_error: String(err) }).eq('id', r.id)
      } catch (e) {
        console.error('Failed to update retry_count', e)
      }
    }
  }
}

Deno.serve(async (req) => {
  if (req.method === 'OPTIONS') {
    return new Response('ok', { headers: corsHeaders })
  }

  try {
    // Allow manual triggering via POST; also allow GET for quick check
    if (req.method === 'POST' || req.method === 'GET') {
      await processBatch()
      return new Response('ok', { headers: corsHeaders })
    }

    return new Response('method not allowed', { status: 405, headers: corsHeaders })
  } catch (err) {
    console.error('Processor error', err)
    return new Response(JSON.stringify({ error: String(err) }), { status: 500, headers: corsHeaders })
  }
})