bb0e54dcee
Every notification depended on Firebase Cloud Messaging, so a missing Firebase project, which is the current state, meant no notification could be delivered on any platform. Web Push and token-based Apple Push are now first class transports alongside FCM, chosen per registered device, and a scheduled Cloudflare Worker drain retries an outbox so a provider outage delays rather than drops a message.
774 lines
26 KiB
TypeScript
774 lines
26 KiB
TypeScript
import { createClient } from '@supabase/supabase-js'
|
|
import { requireUser } from '../_shared/auth.ts'
|
|
import { corsHeaders, handleCorsPreflightRequest } from '../_shared/cors.ts'
|
|
import {
|
|
buildPushNotification,
|
|
getFcmAccessToken,
|
|
isServiceRoleApiKey,
|
|
isServiceRoleAuthorization,
|
|
isUuid,
|
|
parsePushRequest,
|
|
providerIsSupported,
|
|
PushContractError,
|
|
readFcmConfig,
|
|
sendFcmMessage,
|
|
type PushEventType,
|
|
type PushProvider,
|
|
type TeamInviteNotificationContext,
|
|
} from '../_shared/push-contract.ts'
|
|
import { readApnsConfig, sendApnsMessage } from '../_shared/apns.ts'
|
|
import { readWebPushConfig, sendWebPushMessage } from '../_shared/webpush.ts'
|
|
import { createServiceRoleClient } from '../_shared/quota.ts'
|
|
|
|
const JSON_HEADERS = {
|
|
...corsHeaders,
|
|
'Content-Type': 'application/json',
|
|
'Cache-Control': 'no-store',
|
|
Pragma: 'no-cache',
|
|
}
|
|
const REQUEST_LIMIT_BYTES = 4 * 1024
|
|
const SEND_CONCURRENCY = 10
|
|
const DRAIN_CONCURRENCY = 4
|
|
|
|
type DispatchActor =
|
|
| { kind: 'user'; userId: string }
|
|
| { kind: 'system' }
|
|
|
|
interface ResolvedEvent {
|
|
recipientUserIds: string[]
|
|
inviteContext?: TeamInviteNotificationContext
|
|
}
|
|
|
|
interface ReservationResult {
|
|
reserved: boolean
|
|
duplicate: boolean
|
|
attemptId: string
|
|
status: string
|
|
dispatchLeaseToken: string | null
|
|
nextRetryAt: string | null
|
|
}
|
|
|
|
interface LeasedDelivery {
|
|
deliveryId: string
|
|
deliveryLeaseToken: string
|
|
pushTokenId: string
|
|
provider: PushProvider
|
|
registrationId: string
|
|
lastRegisteredAt: string
|
|
}
|
|
|
|
interface DispatchSummary {
|
|
attemptId: string
|
|
status: 'pending' | 'processing' | 'succeeded' | 'partial' | 'failed'
|
|
complete: boolean
|
|
delivered: number
|
|
stale: number
|
|
retryableFailed: number
|
|
permanentFailed: number
|
|
nextRetryAt: string | null
|
|
}
|
|
|
|
interface ClaimedDispatch {
|
|
attemptId: string
|
|
eventType: PushEventType
|
|
resourceId: string
|
|
dispatchLeaseToken: string
|
|
}
|
|
|
|
interface ProcessResult extends DispatchSummary {
|
|
attempted: number
|
|
transientError: string | null
|
|
transientStatus: number | null
|
|
}
|
|
|
|
function jsonResponse(body: Record<string, unknown>, status = 200): Response {
|
|
return new Response(JSON.stringify(body), { status, headers: JSON_HEADERS })
|
|
}
|
|
|
|
function isRecord(value: unknown): value is Record<string, unknown> {
|
|
return value !== null && typeof value === 'object' && !Array.isArray(value)
|
|
}
|
|
|
|
function readEndpointMode(url: URL): { mode: 'dispatch' } | { mode: 'drain'; batchLimit: number } {
|
|
const entries = [...url.searchParams.entries()]
|
|
if (entries.length === 0) return { mode: 'dispatch' }
|
|
if (
|
|
entries.some(([key]) => key !== 'mode' && key !== 'limit')
|
|
|| url.searchParams.getAll('mode').length !== 1
|
|
|| url.searchParams.get('mode') !== 'drain'
|
|
|| url.searchParams.getAll('limit').length > 1
|
|
) {
|
|
throw new PushContractError('invalid_push_request', 400)
|
|
}
|
|
const rawLimit = url.searchParams.get('limit') ?? '20'
|
|
const batchLimit = Number(rawLimit)
|
|
if (!Number.isSafeInteger(batchLimit) || batchLimit < 1 || batchLimit > 100) {
|
|
throw new PushContractError('invalid_batch_limit', 400)
|
|
}
|
|
return { mode: 'drain', batchLimit }
|
|
}
|
|
|
|
async function readJson(req: Request): Promise<unknown> {
|
|
if (req.headers.get('Content-Type')?.split(';', 1)[0].trim().toLowerCase() !== 'application/json') {
|
|
throw new PushContractError('unsupported_media_type', 415)
|
|
}
|
|
const declaredLength = Number(req.headers.get('Content-Length') ?? '0')
|
|
if (Number.isFinite(declaredLength) && declaredLength > REQUEST_LIMIT_BYTES) {
|
|
throw new PushContractError('request_too_large', 413)
|
|
}
|
|
const text = await req.text()
|
|
if (new TextEncoder().encode(text).byteLength > REQUEST_LIMIT_BYTES) {
|
|
throw new PushContractError('request_too_large', 413)
|
|
}
|
|
try {
|
|
return JSON.parse(text)
|
|
} catch {
|
|
throw new PushContractError('invalid_json', 400)
|
|
}
|
|
}
|
|
|
|
function createUserClient(req: Request) {
|
|
return createClient(
|
|
Deno.env.get('SUPABASE_URL') ?? '',
|
|
Deno.env.get('SUPABASE_ANON_KEY') ?? '',
|
|
{
|
|
auth: { persistSession: false, autoRefreshToken: false },
|
|
global: { headers: { Authorization: req.headers.get('Authorization') ?? '' } },
|
|
},
|
|
)
|
|
}
|
|
|
|
async function authenticateDispatcher(req: Request): Promise<DispatchActor> {
|
|
const authorization = req.headers.get('Authorization')
|
|
const legacyServiceBearer = await isServiceRoleAuthorization(
|
|
authorization, Deno.env.get('SUPABASE_SERVICE_ROLE_KEY'),
|
|
)
|
|
const serviceApiKey = await isServiceRoleApiKey(req.headers.get('apikey'))
|
|
if (legacyServiceBearer || serviceApiKey) {
|
|
return { kind: 'system' }
|
|
}
|
|
const user = await requireUser(req)
|
|
return { kind: 'user', userId: user.id }
|
|
}
|
|
|
|
async function resolvePushEvent(
|
|
eventType: PushEventType,
|
|
resourceId: string,
|
|
actor: DispatchActor,
|
|
userClient: ReturnType<typeof createUserClient> | null,
|
|
): Promise<ResolvedEvent> {
|
|
const serviceClient = createServiceRoleClient()
|
|
if (eventType === 'transcription.completed') {
|
|
const { data, error } = await serviceClient
|
|
.from('history')
|
|
.select('id,user_id,status')
|
|
.eq('id', resourceId)
|
|
.maybeSingle()
|
|
if (error) throw new PushContractError('push_resource_lookup_failed', 500)
|
|
if (!data || !isUuid(data.user_id)) {
|
|
throw new PushContractError('push_resource_unavailable', 404)
|
|
}
|
|
if (actor.kind === 'user' && data.user_id !== actor.userId) {
|
|
throw new PushContractError('push_resource_unavailable', 404)
|
|
}
|
|
if (data.status !== 'completed') {
|
|
throw new PushContractError('push_resource_not_ready', 409)
|
|
}
|
|
return { recipientUserIds: [data.user_id] }
|
|
}
|
|
|
|
if (eventType === 'billing.status.changed') {
|
|
const { data, error } = await serviceClient
|
|
.from('subscriptions')
|
|
.select('id,user_id')
|
|
.eq('id', resourceId)
|
|
.maybeSingle()
|
|
if (error) throw new PushContractError('push_resource_lookup_failed', 500)
|
|
if (!data || !isUuid(data.user_id)) {
|
|
throw new PushContractError('push_resource_unavailable', 404)
|
|
}
|
|
if (actor.kind === 'user' && data.user_id !== actor.userId) {
|
|
throw new PushContractError('push_resource_unavailable', 404)
|
|
}
|
|
return { recipientUserIds: [data.user_id] }
|
|
}
|
|
|
|
const { data: invite, error: inviteError } = await serviceClient
|
|
.from('team_invites')
|
|
.select('id,team_id,invited_by,token,accepted_at,expires_at')
|
|
.eq('id', resourceId)
|
|
.maybeSingle()
|
|
if (inviteError) throw new PushContractError('push_resource_lookup_failed', 500)
|
|
if (
|
|
!invite
|
|
|| (actor.kind === 'user' && invite.invited_by !== actor.userId)
|
|
|| invite.accepted_at !== null
|
|
|| typeof invite.expires_at !== 'string'
|
|
|| Date.parse(invite.expires_at) <= Date.now()
|
|
|| typeof invite.token !== 'string'
|
|
|| !isUuid(invite.team_id)
|
|
) {
|
|
throw new PushContractError('push_resource_unavailable', 404)
|
|
}
|
|
|
|
let recipientId: unknown
|
|
if (actor.kind === 'user') {
|
|
if (userClient === null) throw new PushContractError('push_contract_unavailable', 503)
|
|
const result = await userClient.rpc('resolve_team_invite_recipient', { invite_id: resourceId })
|
|
if (result.error) {
|
|
const unavailable = result.error.code === '42883'
|
|
|| result.error.message?.includes('resolve_team_invite_recipient')
|
|
throw new PushContractError(
|
|
unavailable ? 'push_contract_unavailable' : 'push_resource_lookup_failed',
|
|
unavailable ? 503 : 500,
|
|
)
|
|
}
|
|
recipientId = result.data
|
|
} else {
|
|
const result = await serviceClient.rpc('push_dispatch_target_user', {
|
|
target_event_type: eventType,
|
|
target_resource_id: resourceId,
|
|
})
|
|
if (result.error) throw new PushContractError('push_resource_lookup_failed', 500)
|
|
recipientId = result.data
|
|
}
|
|
if (recipientId !== null && !isUuid(recipientId)) {
|
|
throw new PushContractError('push_resource_lookup_failed', 500)
|
|
}
|
|
return {
|
|
recipientUserIds: recipientId === null ? [] : [recipientId],
|
|
inviteContext: { inviteToken: invite.token, teamId: invite.team_id },
|
|
}
|
|
}
|
|
|
|
function normalizeReservation(value: unknown): ReservationResult {
|
|
if (!isRecord(value)) throw new PushContractError('push_contract_invalid', 502)
|
|
const reserved = value.reserved
|
|
const duplicate = value.duplicate
|
|
const rawLeaseToken = value.dispatch_lease_token ?? null
|
|
const leaseToken = typeof rawLeaseToken === 'string' ? rawLeaseToken : null
|
|
const nextRetryAt = value.next_retry_at ?? null
|
|
if (
|
|
typeof reserved !== 'boolean'
|
|
|| typeof duplicate !== 'boolean'
|
|
|| reserved === duplicate
|
|
|| !isUuid(value.attempt_id)
|
|
|| typeof value.status !== 'string'
|
|
|| (reserved && !isUuid(rawLeaseToken))
|
|
|| (!reserved && rawLeaseToken !== null)
|
|
|| (nextRetryAt !== null && (
|
|
typeof nextRetryAt !== 'string' || !Number.isFinite(Date.parse(nextRetryAt))
|
|
))
|
|
) {
|
|
throw new PushContractError('push_contract_invalid', 502)
|
|
}
|
|
return {
|
|
reserved,
|
|
duplicate,
|
|
attemptId: value.attempt_id,
|
|
status: value.status,
|
|
dispatchLeaseToken: leaseToken,
|
|
nextRetryAt,
|
|
}
|
|
}
|
|
|
|
async function reserveDispatch(
|
|
actor: DispatchActor,
|
|
userClient: ReturnType<typeof createUserClient> | null,
|
|
eventType: PushEventType,
|
|
resourceId: string,
|
|
): Promise<ReservationResult> {
|
|
const client = actor.kind === 'system' ? createServiceRoleClient() : userClient
|
|
if (client === null) throw new PushContractError('push_contract_unavailable', 503)
|
|
const functionName = actor.kind === 'system'
|
|
? 'reserve_system_push_dispatch'
|
|
: 'reserve_push_dispatch'
|
|
const { data, error } = await client.rpc(functionName, {
|
|
push_event_type: eventType,
|
|
push_resource_id: resourceId,
|
|
})
|
|
if (error) {
|
|
const message = error.message?.toLowerCase() ?? ''
|
|
if (message.includes('push_rate_limited') || error.code === '54000') {
|
|
throw new PushContractError('push_rate_limited', 429)
|
|
}
|
|
if (message.includes('push_resource_unavailable') || error.code === 'P0002') {
|
|
throw new PushContractError('push_resource_unavailable', 404)
|
|
}
|
|
if (
|
|
message.includes('push_claim_busy')
|
|
|| error.code === '55P03'
|
|
|| error.code === '40001'
|
|
) {
|
|
throw new PushContractError('push_claim_busy', 409)
|
|
}
|
|
if (message.includes('invalid_push_event') || error.code === '22023') {
|
|
throw new PushContractError('unsupported_push_event', 400)
|
|
}
|
|
if (
|
|
error.code === '42883'
|
|
|| message.includes('reserve_push_dispatch')
|
|
|| message.includes('reserve_system_push_dispatch')
|
|
|| message.includes('service_role_required')
|
|
) {
|
|
throw new PushContractError('push_contract_unavailable', 503)
|
|
}
|
|
throw new PushContractError('push_reservation_failed', 500)
|
|
}
|
|
return normalizeReservation(data)
|
|
}
|
|
|
|
function normalizeLeasedDeliveries(value: unknown): LeasedDelivery[] {
|
|
if (!Array.isArray(value)) throw new PushContractError('push_contract_invalid', 502)
|
|
return value.map((row) => {
|
|
if (
|
|
!isRecord(row)
|
|
|| !isUuid(row.delivery_id)
|
|
|| !isUuid(row.delivery_lease_token)
|
|
|| !isUuid(row.push_token_id)
|
|
|| (row.provider !== 'fcm' && row.provider !== 'apns' && row.provider !== 'webpush')
|
|
|| typeof row.registration_id !== 'string'
|
|
|| row.registration_id.length < 20
|
|
|| row.registration_id.length > 4096
|
|
|| /\s/.test(row.registration_id)
|
|
|| typeof row.last_registered_at !== 'string'
|
|
|| !Number.isFinite(Date.parse(row.last_registered_at))
|
|
) {
|
|
throw new PushContractError('push_contract_invalid', 502)
|
|
}
|
|
return {
|
|
deliveryId: row.delivery_id,
|
|
deliveryLeaseToken: row.delivery_lease_token,
|
|
pushTokenId: row.push_token_id,
|
|
provider: row.provider,
|
|
registrationId: row.registration_id,
|
|
lastRegisteredAt: row.last_registered_at,
|
|
}
|
|
})
|
|
}
|
|
|
|
async function leaseDeliveries(
|
|
attemptId: string,
|
|
dispatchLeaseToken: string,
|
|
): Promise<LeasedDelivery[]> {
|
|
const { data, error } = await createServiceRoleClient().rpc('lease_push_deliveries', {
|
|
target_attempt_id: attemptId,
|
|
target_dispatch_lease_token: dispatchLeaseToken,
|
|
})
|
|
if (error) {
|
|
const conflict = error.code === '40001' || error.message?.includes('lease_conflict')
|
|
throw new PushContractError(conflict ? 'push_lease_conflict' : 'push_lease_failed', conflict ? 409 : 500)
|
|
}
|
|
return normalizeLeasedDeliveries(data)
|
|
}
|
|
|
|
async function finalizeDelivery(
|
|
delivery: LeasedDelivery,
|
|
outcome: 'delivered' | 'stale' | 'retryable_failure' | 'permanent_failure',
|
|
errorCode: string | null,
|
|
): Promise<void> {
|
|
const { error } = await createServiceRoleClient().rpc('finalize_push_delivery', {
|
|
target_delivery_id: delivery.deliveryId,
|
|
target_delivery_lease_token: delivery.deliveryLeaseToken,
|
|
delivery_outcome: outcome,
|
|
delivery_error_code: errorCode,
|
|
})
|
|
if (error) {
|
|
const conflict = error.code === '40001' || error.message?.includes('lease_conflict')
|
|
throw new PushContractError(
|
|
conflict ? 'push_delivery_lease_conflict' : 'push_delivery_finalize_failed',
|
|
conflict ? 409 : 500,
|
|
)
|
|
}
|
|
}
|
|
|
|
function normalizeDispatchSummary(value: unknown, attemptId: string): DispatchSummary {
|
|
if (!isRecord(value) || value.attempt_id !== attemptId) {
|
|
throw new PushContractError('push_contract_invalid', 502)
|
|
}
|
|
const status = value.status
|
|
if (
|
|
status !== 'pending' && status !== 'processing' && status !== 'succeeded'
|
|
&& status !== 'partial' && status !== 'failed'
|
|
) {
|
|
throw new PushContractError('push_contract_invalid', 502)
|
|
}
|
|
const complete = value.complete
|
|
if (typeof complete !== 'boolean') throw new PushContractError('push_contract_invalid', 502)
|
|
const count = (key: string): number => {
|
|
const candidate = value[key] ?? 0
|
|
if (typeof candidate !== 'number' || !Number.isSafeInteger(candidate) || candidate < 0) {
|
|
throw new PushContractError('push_contract_invalid', 502)
|
|
}
|
|
return candidate
|
|
}
|
|
const nextRetryAt = value.next_retry_at ?? null
|
|
if (nextRetryAt !== null && (
|
|
typeof nextRetryAt !== 'string' || !Number.isFinite(Date.parse(nextRetryAt))
|
|
)) {
|
|
throw new PushContractError('push_contract_invalid', 502)
|
|
}
|
|
return {
|
|
attemptId,
|
|
status,
|
|
complete,
|
|
delivered: count('delivered'),
|
|
stale: count('stale'),
|
|
retryableFailed: count('retryable_failed'),
|
|
permanentFailed: count('permanent_failed'),
|
|
nextRetryAt,
|
|
}
|
|
}
|
|
|
|
async function finalizeDispatch(
|
|
attemptId: string,
|
|
dispatchLeaseToken: string,
|
|
): Promise<DispatchSummary> {
|
|
const { data, error } = await createServiceRoleClient().rpc('finalize_push_dispatch', {
|
|
target_attempt_id: attemptId,
|
|
target_dispatch_lease_token: dispatchLeaseToken,
|
|
})
|
|
if (error) {
|
|
const conflict = error.code === '40001' || error.message?.includes('lease_conflict')
|
|
throw new PushContractError(
|
|
conflict ? 'push_dispatch_lease_conflict' : 'push_dispatch_finalize_failed',
|
|
conflict ? 409 : 500,
|
|
)
|
|
}
|
|
return normalizeDispatchSummary(data, attemptId)
|
|
}
|
|
|
|
function retryablePushError(error: unknown, fallbackCode: string): { code: string; status: number } {
|
|
if (!(error instanceof PushContractError)) return { code: fallbackCode, status: 502 }
|
|
const permanent = new Set(['invalid_push_payload'])
|
|
return {
|
|
code: permanent.has(error.code) ? 'push_payload_invalid' : error.code,
|
|
status: error.status,
|
|
}
|
|
}
|
|
|
|
async function processAttempt(input: {
|
|
eventType: PushEventType
|
|
resourceId: string
|
|
attemptId: string
|
|
dispatchLeaseToken: string
|
|
inviteContext?: TeamInviteNotificationContext
|
|
}): Promise<ProcessResult> {
|
|
const deliveries = await leaseDeliveries(input.attemptId, input.dispatchLeaseToken)
|
|
if (deliveries.length === 0) {
|
|
const summary = await finalizeDispatch(input.attemptId, input.dispatchLeaseToken)
|
|
return { ...summary, attempted: 0, transientError: null, transientStatus: null }
|
|
}
|
|
|
|
let inviteContext = input.inviteContext
|
|
if (input.eventType === 'team.invite.created' && !inviteContext) {
|
|
const { data: invite, error } = await createServiceRoleClient()
|
|
.from('team_invites')
|
|
.select('team_id,token,accepted_at,expires_at')
|
|
.eq('id', input.resourceId)
|
|
.maybeSingle()
|
|
if (
|
|
error || !invite || invite.accepted_at !== null
|
|
|| typeof invite.expires_at !== 'string' || Date.parse(invite.expires_at) <= Date.now()
|
|
|| !isUuid(invite.team_id) || typeof invite.token !== 'string'
|
|
) {
|
|
await mapWithConcurrency(deliveries, SEND_CONCURRENCY, (delivery) => (
|
|
finalizeDelivery(delivery, 'permanent_failure', 'push_resource_unavailable')
|
|
))
|
|
const summary = await finalizeDispatch(input.attemptId, input.dispatchLeaseToken)
|
|
return { ...summary, attempted: deliveries.length, transientError: null, transientStatus: null }
|
|
}
|
|
inviteContext = { teamId: invite.team_id, inviteToken: invite.token }
|
|
}
|
|
|
|
const unsupported = deliveries.filter((delivery) => !providerIsSupported(delivery.provider))
|
|
await mapWithConcurrency(unsupported, SEND_CONCURRENCY, (delivery) => (
|
|
finalizeDelivery(delivery, 'permanent_failure', 'push_provider_not_supported')
|
|
))
|
|
|
|
let transientError: string | null = null
|
|
let transientStatus: number | null = null
|
|
|
|
const recordTransient = (code: string, status: number): void => {
|
|
transientError ??= code
|
|
transientStatus ??= status
|
|
}
|
|
|
|
const deliveriesFor = (provider: PushProvider): LeasedDelivery[] => (
|
|
deliveries.filter((delivery) => delivery.provider === provider)
|
|
)
|
|
|
|
const failProvider = async (provider: PushProvider, code: string): Promise<void> => {
|
|
await mapWithConcurrency(deliveriesFor(provider), SEND_CONCURRENCY, (delivery) => (
|
|
finalizeDelivery(delivery, 'retryable_failure', code)
|
|
))
|
|
}
|
|
|
|
const sendFor = async (
|
|
provider: PushProvider,
|
|
fallbackCode: string,
|
|
send: (registrationId: string) => Promise<unknown>,
|
|
): Promise<void> => {
|
|
const scoped = deliveriesFor(provider)
|
|
if (scoped.length === 0) return
|
|
await mapWithConcurrency(scoped, SEND_CONCURRENCY, async (delivery) => {
|
|
try {
|
|
await send(delivery.registrationId)
|
|
await finalizeDelivery(delivery, 'delivered', null)
|
|
} catch (error) {
|
|
if (error instanceof PushContractError && error.staleRegistration) {
|
|
await finalizeDelivery(delivery, 'stale', 'registration_stale')
|
|
return
|
|
}
|
|
const normalized = retryablePushError(error, fallbackCode)
|
|
if (normalized.code === 'push_payload_invalid') {
|
|
await finalizeDelivery(delivery, 'permanent_failure', normalized.code)
|
|
} else {
|
|
recordTransient(normalized.code, normalized.status)
|
|
await finalizeDelivery(delivery, 'retryable_failure', normalized.code)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
const buildNotification = (): ReturnType<typeof buildPushNotification> => (
|
|
buildPushNotification(input.eventType, input.resourceId, 'en', inviteContext)
|
|
)
|
|
|
|
if (deliveriesFor('fcm').length > 0) {
|
|
let config: ReturnType<typeof readFcmConfig> | null = null
|
|
let accessToken: string | null = null
|
|
try {
|
|
config = readFcmConfig()
|
|
accessToken = await getFcmAccessToken(config)
|
|
} catch (error) {
|
|
const normalized = retryablePushError(error, 'fcm_send_failed')
|
|
recordTransient(normalized.code, normalized.status)
|
|
await failProvider('fcm', normalized.code)
|
|
}
|
|
if (config && accessToken) {
|
|
await sendFor('fcm', 'fcm_send_failed', (registrationId) => sendFcmMessage(
|
|
registrationId,
|
|
buildNotification(),
|
|
{
|
|
config: config as ReturnType<typeof readFcmConfig>,
|
|
getAccessToken: () => Promise.resolve(accessToken as string),
|
|
},
|
|
))
|
|
}
|
|
}
|
|
|
|
if (deliveriesFor('webpush').length > 0) {
|
|
let webPushConfig: ReturnType<typeof readWebPushConfig> | null = null
|
|
try {
|
|
webPushConfig = readWebPushConfig()
|
|
} catch (error) {
|
|
const normalized = retryablePushError(error, 'webpush_send_failed')
|
|
recordTransient(normalized.code, normalized.status)
|
|
await failProvider('webpush', normalized.code)
|
|
}
|
|
if (webPushConfig) {
|
|
await sendFor('webpush', 'webpush_send_failed', (registrationId) => sendWebPushMessage(
|
|
registrationId,
|
|
buildNotification(),
|
|
{ config: webPushConfig as ReturnType<typeof readWebPushConfig> },
|
|
))
|
|
}
|
|
}
|
|
|
|
if (deliveriesFor('apns').length > 0) {
|
|
let apnsConfig: ReturnType<typeof readApnsConfig> | null = null
|
|
try {
|
|
apnsConfig = readApnsConfig()
|
|
} catch (error) {
|
|
const normalized = retryablePushError(error, 'apns_send_failed')
|
|
recordTransient(normalized.code, normalized.status)
|
|
await failProvider('apns', normalized.code)
|
|
}
|
|
if (apnsConfig) {
|
|
await sendFor('apns', 'apns_send_failed', (registrationId) => sendApnsMessage(
|
|
registrationId,
|
|
buildNotification(),
|
|
{ config: apnsConfig as ReturnType<typeof readApnsConfig> },
|
|
))
|
|
}
|
|
}
|
|
|
|
const summary = await finalizeDispatch(input.attemptId, input.dispatchLeaseToken)
|
|
return {
|
|
...summary,
|
|
attempted: deliveries.length,
|
|
transientError,
|
|
transientStatus,
|
|
}
|
|
}
|
|
|
|
async function mapWithConcurrency<T, R>(
|
|
values: T[],
|
|
concurrency: number,
|
|
action: (value: T) => Promise<R>,
|
|
): Promise<R[]> {
|
|
const results = new Array<R>(values.length)
|
|
let cursor = 0
|
|
const workers = Array.from({ length: Math.min(concurrency, values.length) }, async () => {
|
|
while (cursor < values.length) {
|
|
const index = cursor++
|
|
results[index] = await action(values[index])
|
|
}
|
|
})
|
|
await Promise.all(workers)
|
|
return results
|
|
}
|
|
|
|
function normalizeClaims(value: unknown): ClaimedDispatch[] {
|
|
if (!Array.isArray(value)) throw new PushContractError('push_contract_invalid', 502)
|
|
return value.map((row) => {
|
|
if (
|
|
!isRecord(row)
|
|
|| !isUuid(row.attempt_id)
|
|
|| (row.event_type !== 'transcription.completed'
|
|
&& row.event_type !== 'team.invite.created'
|
|
&& row.event_type !== 'billing.status.changed')
|
|
|| !isUuid(row.resource_id)
|
|
|| !isUuid(row.dispatch_lease_token)
|
|
) {
|
|
throw new PushContractError('push_contract_invalid', 502)
|
|
}
|
|
return {
|
|
attemptId: row.attempt_id,
|
|
eventType: row.event_type,
|
|
resourceId: row.resource_id,
|
|
dispatchLeaseToken: row.dispatch_lease_token,
|
|
}
|
|
})
|
|
}
|
|
|
|
async function drainDueDispatches(batchLimit: number): Promise<Record<string, unknown>> {
|
|
const { data, error } = await createServiceRoleClient().rpc('claim_due_push_dispatches', {
|
|
batch_limit: batchLimit,
|
|
})
|
|
if (error) throw new PushContractError('push_drain_claim_failed', 500)
|
|
const claims = normalizeClaims(data)
|
|
const results = await mapWithConcurrency(claims, DRAIN_CONCURRENCY, async (claim) => {
|
|
try {
|
|
return await processAttempt({
|
|
eventType: claim.eventType,
|
|
resourceId: claim.resourceId,
|
|
attemptId: claim.attemptId,
|
|
dispatchLeaseToken: claim.dispatchLeaseToken,
|
|
})
|
|
} catch (error) {
|
|
return {
|
|
attemptId: claim.attemptId,
|
|
status: 'processing',
|
|
complete: false,
|
|
attempted: 0,
|
|
error: error instanceof PushContractError ? error.code : 'push_drain_failed',
|
|
}
|
|
}
|
|
})
|
|
return {
|
|
claimed: claims.length,
|
|
completed: results.filter((result) => result.complete === true).length,
|
|
pending: results.filter((result) => result.complete !== true).length,
|
|
results,
|
|
}
|
|
}
|
|
|
|
Deno.serve(async (req: Request) => {
|
|
const preflight = handleCorsPreflightRequest(req)
|
|
if (preflight) return preflight
|
|
if (req.method !== 'POST') return jsonResponse({ error: 'method_not_allowed' }, 405)
|
|
|
|
try {
|
|
const actor = await authenticateDispatcher(req)
|
|
const requestUrl = new URL(req.url)
|
|
const endpoint = readEndpointMode(requestUrl)
|
|
if (endpoint.mode === 'drain') {
|
|
if (actor.kind !== 'system') return jsonResponse({ error: 'forbidden' }, 403)
|
|
const body = await readJson(req)
|
|
if (!isRecord(body) || Object.keys(body).length !== 0) {
|
|
throw new PushContractError('invalid_push_request', 400)
|
|
}
|
|
return jsonResponse(await drainDueDispatches(endpoint.batchLimit))
|
|
}
|
|
|
|
const input = parsePushRequest(await readJson(req))
|
|
const userClient = actor.kind === 'user' ? createUserClient(req) : null
|
|
const resolved = await resolvePushEvent(
|
|
input.eventType,
|
|
input.resourceId,
|
|
actor,
|
|
userClient,
|
|
)
|
|
if (resolved.recipientUserIds.length === 0) {
|
|
return jsonResponse({
|
|
event_type: input.eventType,
|
|
resource_id: input.resourceId,
|
|
duplicate: false,
|
|
status: 'succeeded',
|
|
attempted: 0,
|
|
sent: 0,
|
|
stale_removed: 0,
|
|
})
|
|
}
|
|
|
|
const reservation = await reserveDispatch(
|
|
actor,
|
|
userClient,
|
|
input.eventType,
|
|
input.resourceId,
|
|
)
|
|
if (reservation.duplicate || reservation.dispatchLeaseToken === null) {
|
|
return jsonResponse({
|
|
event_type: input.eventType,
|
|
resource_id: input.resourceId,
|
|
duplicate: true,
|
|
status: reservation.status,
|
|
next_retry_at: reservation.nextRetryAt,
|
|
attempted: 0,
|
|
sent: 0,
|
|
stale_removed: 0,
|
|
})
|
|
}
|
|
|
|
const result = await processAttempt({
|
|
eventType: input.eventType,
|
|
resourceId: input.resourceId,
|
|
attemptId: reservation.attemptId,
|
|
dispatchLeaseToken: reservation.dispatchLeaseToken,
|
|
inviteContext: resolved.inviteContext,
|
|
})
|
|
const responseBody = {
|
|
event_type: input.eventType,
|
|
resource_id: input.resourceId,
|
|
duplicate: false,
|
|
status: result.status,
|
|
complete: result.complete,
|
|
attempted: result.attempted,
|
|
sent: result.delivered,
|
|
stale_removed: result.stale,
|
|
retryable_failed: result.retryableFailed,
|
|
permanent_failed: result.permanentFailed,
|
|
next_retry_at: result.nextRetryAt,
|
|
}
|
|
if (result.transientError) {
|
|
return jsonResponse(
|
|
{ error: result.transientError, ...responseBody },
|
|
result.transientStatus ?? 502,
|
|
)
|
|
}
|
|
if (result.status === 'partial' || result.status === 'failed') {
|
|
return jsonResponse({ error: 'push_delivery_failed', ...responseBody }, 502)
|
|
}
|
|
return jsonResponse(responseBody, result.complete ? 200 : 202)
|
|
} catch (error) {
|
|
if (error instanceof PushContractError) {
|
|
return jsonResponse({ error: error.code, message: error.code }, error.status)
|
|
}
|
|
if (error && typeof error === 'object' && 'status' in error && 'message' in error) {
|
|
return jsonResponse({ error: 'unauthorized', message: 'unauthorized' }, 401)
|
|
}
|
|
return jsonResponse({ error: 'push_dispatch_failed', message: 'push_dispatch_failed' }, 500)
|
|
}
|
|
})
|