mirror of
https://github.com/vernu/textbee.git
synced 2026-09-03 03:29:58 +03:00
feat(api): pace bulk dispatch to the device send rate and bound push freshness
Large sends are now released to the push service in waves of BULK_DISPATCH_WINDOW messages spaced by the device's configured send delay, so the backlog waits in the queue instead of piling up ahead of the phone. Each message records when its wave is due (dispatchDueAt) and the stale-status cron leaves paced messages alone until then. A paced batch stays in processing until every wave has been handed off, and the send response carries estimatedCompletionAt for multi-wave sends. SMS pushes now carry a 72 hour ttl (FCM_SMS_TTL_SECONDS) that never expires before scheduledAt, and the heartbeat probe collapses with a 30 minute ttl. Swagger wording for dispatchedAt no longer implies device receipt. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
isra el
@@ -38,6 +38,17 @@ WEBHOOK_AUTO_DISABLE_MIN_FAILURE_RATE=0.50
|
||||
# SMS Queue Configuration
|
||||
USE_SMS_QUEUE=false
|
||||
REDIS_URL=redis://localhost:6379 # if queue is enabled, redis url is required
|
||||
# Max FCM messages per queue job
|
||||
MAX_SMS_BATCH_SIZE=100
|
||||
# Bulk sends are released in waves of this many messages, spaced by the
|
||||
# device's configured send delay, so pushes never pile up ahead of the phone
|
||||
BULK_DISPATCH_WINDOW=50
|
||||
# Log a warning when a paced batch is projected to take longer than this
|
||||
BULK_DISPATCH_MAX_SPREAD_HOURS=72
|
||||
# How long FCM keeps an SMS push for an offline device before dropping it
|
||||
FCM_SMS_TTL_SECONDS=259200
|
||||
# Lifetime of the hourly heartbeat probe push
|
||||
FCM_HEARTBEAT_TTL_SECONDS=1800
|
||||
|
||||
CLOUDFLARE_TURNSTILE_SECRET_KEY=1x0000000000000000000000000000000AA
|
||||
|
||||
|
||||
+18
-2
@@ -1434,6 +1434,12 @@
|
||||
"type": "number",
|
||||
"description": "Number of recipients in the batch. Queued sends only."
|
||||
},
|
||||
"estimatedCompletionAt": {
|
||||
"type": "string",
|
||||
"format": "date-time",
|
||||
"description": "Projected time the device finishes sending, based on the recipient count and the device send delay setting. Present only when the batch is large enough to be released in waves. The device must stay online for the estimate to hold.",
|
||||
"example": "2026-08-22T14:05:00.000Z"
|
||||
},
|
||||
"successCount": {
|
||||
"type": "number",
|
||||
"description": "Messages pushed to the device. Returned instead of the queue fields when the instance dispatches immediately."
|
||||
@@ -1702,10 +1708,15 @@
|
||||
"type": "string",
|
||||
"description": "When the send was requested. Sent messages only."
|
||||
},
|
||||
"dispatchDueAt": {
|
||||
"format": "date-time",
|
||||
"type": "string",
|
||||
"description": "When the queue is due to hand the message to the push service. Set only for large sends, which are released in waves paced to the device send delay so the phone is never asked to hold more than it can send. Absent for small or immediate sends."
|
||||
},
|
||||
"dispatchedAt": {
|
||||
"format": "date-time",
|
||||
"type": "string",
|
||||
"description": "When the send job reached the device."
|
||||
"description": "When the push service accepted the message for delivery to the device. Does not mean the device has received it; sentAt is the first device-confirmed timestamp."
|
||||
},
|
||||
"sentAt": {
|
||||
"format": "date-time",
|
||||
@@ -1864,10 +1875,15 @@
|
||||
"type": "string",
|
||||
"description": "When the send was requested. Sent messages only."
|
||||
},
|
||||
"dispatchDueAt": {
|
||||
"format": "date-time",
|
||||
"type": "string",
|
||||
"description": "When the queue is due to hand the message to the push service. Set only for large sends, which are released in waves paced to the device send delay so the phone is never asked to hold more than it can send. Absent for small or immediate sends."
|
||||
},
|
||||
"dispatchedAt": {
|
||||
"format": "date-time",
|
||||
"type": "string",
|
||||
"description": "When the send job reached the device."
|
||||
"description": "When the push service accepted the message for delivery to the device. Does not mean the device has received it; sentAt is the first device-confirmed timestamp."
|
||||
},
|
||||
"sentAt": {
|
||||
"format": "date-time",
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
import {
|
||||
DEFAULT_FCM_HEARTBEAT_TTL_SECONDS,
|
||||
DEFAULT_FCM_SMS_TTL_SECONDS,
|
||||
heartbeatAndroidConfig,
|
||||
smsAndroidConfig,
|
||||
} from './fcm-push-options'
|
||||
|
||||
describe('fcm-push-options', () => {
|
||||
const originalEnv = { ...process.env }
|
||||
|
||||
afterEach(() => {
|
||||
process.env = { ...originalEnv }
|
||||
})
|
||||
|
||||
describe('smsAndroidConfig', () => {
|
||||
it('sets a 72 hour ttl in milliseconds by default', () => {
|
||||
expect(smsAndroidConfig()).toEqual({
|
||||
priority: 'high',
|
||||
ttl: DEFAULT_FCM_SMS_TTL_SECONDS * 1000,
|
||||
})
|
||||
expect(DEFAULT_FCM_SMS_TTL_SECONDS).toBe(259200)
|
||||
})
|
||||
|
||||
it('reads FCM_SMS_TTL_SECONDS from the environment', () => {
|
||||
process.env.FCM_SMS_TTL_SECONDS = '3600'
|
||||
|
||||
expect(smsAndroidConfig().ttl).toBe(3_600_000)
|
||||
})
|
||||
|
||||
it('ignores a non-positive or malformed env value', () => {
|
||||
process.env.FCM_SMS_TTL_SECONDS = 'soon'
|
||||
expect(smsAndroidConfig().ttl).toBe(DEFAULT_FCM_SMS_TTL_SECONDS * 1000)
|
||||
|
||||
process.env.FCM_SMS_TTL_SECONDS = '0'
|
||||
expect(smsAndroidConfig().ttl).toBe(DEFAULT_FCM_SMS_TTL_SECONDS * 1000)
|
||||
})
|
||||
|
||||
it('never expires before scheduledAt plus the base ttl', () => {
|
||||
const now = Date.parse('2026-08-22T10:00:00Z')
|
||||
const scheduledAt = '2026-08-23T10:00:00Z'
|
||||
|
||||
const config = smsAndroidConfig(scheduledAt, now)
|
||||
|
||||
const expiresAt = now + (config.ttl as number)
|
||||
expect(expiresAt).toBe(
|
||||
Date.parse(scheduledAt) + DEFAULT_FCM_SMS_TTL_SECONDS * 1000,
|
||||
)
|
||||
})
|
||||
|
||||
it('uses the base ttl when scheduledAt is in the past or invalid', () => {
|
||||
const now = Date.parse('2026-08-22T10:00:00Z')
|
||||
|
||||
expect(smsAndroidConfig('2026-08-21T10:00:00Z', now).ttl).toBe(
|
||||
DEFAULT_FCM_SMS_TTL_SECONDS * 1000,
|
||||
)
|
||||
expect(smsAndroidConfig('not a date', now).ttl).toBe(
|
||||
DEFAULT_FCM_SMS_TTL_SECONDS * 1000,
|
||||
)
|
||||
})
|
||||
|
||||
it('does not set a collapse key on sms pushes', () => {
|
||||
expect(smsAndroidConfig()).not.toHaveProperty('collapseKey')
|
||||
})
|
||||
})
|
||||
|
||||
describe('heartbeatAndroidConfig', () => {
|
||||
it('collapses probes and bounds them to 30 minutes by default', () => {
|
||||
expect(heartbeatAndroidConfig()).toEqual({
|
||||
priority: 'high',
|
||||
ttl: DEFAULT_FCM_HEARTBEAT_TTL_SECONDS * 1000,
|
||||
collapseKey: 'heartbeat_check',
|
||||
})
|
||||
expect(DEFAULT_FCM_HEARTBEAT_TTL_SECONDS).toBe(1800)
|
||||
})
|
||||
|
||||
it('reads FCM_HEARTBEAT_TTL_SECONDS from the environment', () => {
|
||||
process.env.FCM_HEARTBEAT_TTL_SECONDS = '600'
|
||||
|
||||
expect(heartbeatAndroidConfig().ttl).toBe(600_000)
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,49 @@
|
||||
import { AndroidConfig } from 'firebase-admin/messaging'
|
||||
|
||||
// 72 hours. Bounds how long FCM holds an SMS push for a device that is offline
|
||||
// or dozing, instead of the platform default of four weeks.
|
||||
export const DEFAULT_FCM_SMS_TTL_SECONDS = 72 * 3600
|
||||
// 30 minutes. The heartbeat probe is sent hourly, so an older probe is useless.
|
||||
export const DEFAULT_FCM_HEARTBEAT_TTL_SECONDS = 30 * 60
|
||||
|
||||
function readTtlSeconds(envKey: string, fallback: number): number {
|
||||
const value = Number(process.env[envKey])
|
||||
return Number.isFinite(value) && value > 0 ? value : fallback
|
||||
}
|
||||
|
||||
export function fcmSmsTtlSeconds(): number {
|
||||
return readTtlSeconds('FCM_SMS_TTL_SECONDS', DEFAULT_FCM_SMS_TTL_SECONDS)
|
||||
}
|
||||
|
||||
export function fcmHeartbeatTtlSeconds(): number {
|
||||
return readTtlSeconds(
|
||||
'FCM_HEARTBEAT_TTL_SECONDS',
|
||||
DEFAULT_FCM_HEARTBEAT_TTL_SECONDS,
|
||||
)
|
||||
}
|
||||
|
||||
// AndroidConfig.ttl is in milliseconds. The push is only handed to FCM at
|
||||
// scheduledAt, but if it is built early the expiry still lands at or after
|
||||
// scheduledAt plus the base ttl.
|
||||
export function smsAndroidConfig(
|
||||
scheduledAt?: Date | string,
|
||||
now: number = Date.now(),
|
||||
): AndroidConfig {
|
||||
const baseMs = fcmSmsTtlSeconds() * 1000
|
||||
let ttl = baseMs
|
||||
if (scheduledAt) {
|
||||
const scheduledTime = new Date(scheduledAt).getTime()
|
||||
if (Number.isFinite(scheduledTime) && scheduledTime > now) {
|
||||
ttl = scheduledTime - now + baseMs
|
||||
}
|
||||
}
|
||||
return { priority: 'high', ttl }
|
||||
}
|
||||
|
||||
export function heartbeatAndroidConfig(): AndroidConfig {
|
||||
return {
|
||||
priority: 'high',
|
||||
ttl: fcmHeartbeatTtlSeconds() * 1000,
|
||||
collapseKey: 'heartbeat_check',
|
||||
}
|
||||
}
|
||||
@@ -881,7 +881,16 @@ export class RetrieveSMSDTO {
|
||||
@ApiProperty({
|
||||
type: Date,
|
||||
required: false,
|
||||
description: 'When the send job reached the device.',
|
||||
description:
|
||||
'When the queue is due to hand the message to the push service. Set only for large sends, which are released in waves paced to the device send delay so the phone is never asked to hold more than it can send. Absent for small or immediate sends.',
|
||||
})
|
||||
dispatchDueAt?: Date
|
||||
|
||||
@ApiProperty({
|
||||
type: Date,
|
||||
required: false,
|
||||
description:
|
||||
'When the push service accepted the message for delivery to the device. Does not mean the device has received it; sentAt is the first device-confirmed timestamp.',
|
||||
})
|
||||
dispatchedAt?: Date
|
||||
|
||||
@@ -1343,6 +1352,16 @@ export class SendSMSResultDTO {
|
||||
})
|
||||
recipientCount?: number
|
||||
|
||||
@ApiProperty({
|
||||
type: String,
|
||||
required: false,
|
||||
format: 'date-time',
|
||||
description:
|
||||
'Projected time the device finishes sending, based on the recipient count and the device send delay setting. Present only when the batch is large enough to be released in waves. The device must stay online for the estimate to hold.',
|
||||
example: '2026-08-22T14:05:00.000Z',
|
||||
})
|
||||
estimatedCompletionAt?: string
|
||||
|
||||
@ApiProperty({
|
||||
type: Number,
|
||||
required: false,
|
||||
|
||||
@@ -59,6 +59,7 @@ describe('GatewayService', () => {
|
||||
findById: jest.fn(),
|
||||
findByIdAndUpdate: jest.fn(),
|
||||
updateMany: jest.fn(),
|
||||
bulkWrite: jest.fn(),
|
||||
countDocuments: jest.fn(),
|
||||
}
|
||||
|
||||
@@ -879,7 +880,11 @@ describe('GatewayService', () => {
|
||||
|
||||
it('should queue SMS if queue is enabled', async () => {
|
||||
mockSmsQueueService.isQueueEnabled.mockReturnValue(true)
|
||||
mockSmsQueueService.addSendSmsJob.mockResolvedValue(true)
|
||||
mockSmsQueueService.addSendSmsJob.mockResolvedValue({
|
||||
waves: [{ start: 0, end: 1, delayMs: 0 }],
|
||||
sendDelaySeconds: 5,
|
||||
projectedCompletionMs: 5000,
|
||||
})
|
||||
|
||||
const result = await service.sendSMS(mockDeviceId, mockSmsInput)
|
||||
|
||||
@@ -887,6 +892,91 @@ describe('GatewayService', () => {
|
||||
expect(mockSmsQueueService.addSendSmsJob).toHaveBeenCalled()
|
||||
expect(result).toHaveProperty('success', true)
|
||||
expect(result).toHaveProperty('smsBatchId', mockSmsBatch._id)
|
||||
// single immediate wave: no dispatchDueAt stamping, no estimate
|
||||
expect(mockSmsModel.bulkWrite).not.toHaveBeenCalled()
|
||||
expect(result).not.toHaveProperty('estimatedCompletionAt')
|
||||
})
|
||||
|
||||
it('passes the device send delay to the queue and stamps dispatchDueAt per wave', async () => {
|
||||
mockDeviceModel.findById.mockResolvedValue({
|
||||
...mockDevice,
|
||||
smsSendDelaySeconds: 7,
|
||||
})
|
||||
mockSmsQueueService.isQueueEnabled.mockReturnValue(true)
|
||||
mockSmsModel.create
|
||||
.mockResolvedValueOnce({ ...mockSms, _id: 'sms-a' })
|
||||
.mockResolvedValueOnce({ ...mockSms, _id: 'sms-b' })
|
||||
.mockResolvedValueOnce({ ...mockSms, _id: 'sms-c' })
|
||||
mockSmsQueueService.addSendSmsJob.mockResolvedValue({
|
||||
waves: [
|
||||
{ start: 0, end: 2, delayMs: 0 },
|
||||
{ start: 2, end: 3, delayMs: 14_000 },
|
||||
],
|
||||
sendDelaySeconds: 7,
|
||||
projectedCompletionMs: 21_000,
|
||||
})
|
||||
|
||||
const before = Date.now()
|
||||
const result = await service.sendSMS(mockDeviceId, {
|
||||
...mockSmsInput,
|
||||
recipients: ['+15550100', '+15550101', '+15550102'],
|
||||
})
|
||||
|
||||
expect(mockSmsQueueService.addSendSmsJob).toHaveBeenCalledWith(
|
||||
mockDeviceId,
|
||||
expect.any(Array),
|
||||
mockSmsBatch._id,
|
||||
undefined,
|
||||
7,
|
||||
)
|
||||
expect(mockSmsModel.bulkWrite).toHaveBeenCalledTimes(1)
|
||||
const ops = mockSmsModel.bulkWrite.mock.calls[0][0]
|
||||
expect(ops).toHaveLength(2)
|
||||
expect(ops[0].updateMany.filter).toEqual({ _id: { $in: ['sms-a', 'sms-b'] } })
|
||||
expect(ops[1].updateMany.filter).toEqual({ _id: { $in: ['sms-c'] } })
|
||||
const due0 = ops[0].updateMany.update.$set.dispatchDueAt.getTime()
|
||||
const due1 = ops[1].updateMany.update.$set.dispatchDueAt.getTime()
|
||||
expect(due1 - due0).toBe(14_000)
|
||||
expect(due0).toBeGreaterThanOrEqual(before)
|
||||
|
||||
const eta = Date.parse(result.estimatedCompletionAt)
|
||||
expect(eta - due0).toBe(21_000)
|
||||
})
|
||||
|
||||
it('builds every push with a bounded ttl and no collapse key', async () => {
|
||||
mockSmsQueueService.isQueueEnabled.mockReturnValue(true)
|
||||
mockSmsQueueService.addSendSmsJob.mockResolvedValue({
|
||||
waves: [{ start: 0, end: 1, delayMs: 0 }],
|
||||
sendDelaySeconds: 5,
|
||||
projectedCompletionMs: 5000,
|
||||
})
|
||||
|
||||
await service.sendSMS(mockDeviceId, mockSmsInput)
|
||||
|
||||
const [, fcmMessages] = mockSmsQueueService.addSendSmsJob.mock.calls[0]
|
||||
expect(fcmMessages[0].android).toEqual({
|
||||
priority: 'high',
|
||||
ttl: 72 * 3600 * 1000,
|
||||
})
|
||||
})
|
||||
|
||||
it('extends the ttl so a scheduled push never expires before scheduledAt', async () => {
|
||||
mockSmsQueueService.isQueueEnabled.mockReturnValue(true)
|
||||
mockSmsQueueService.addSendSmsJob.mockResolvedValue({
|
||||
waves: [{ start: 0, end: 1, delayMs: 3_600_000 }],
|
||||
sendDelaySeconds: 5,
|
||||
projectedCompletionMs: 3_605_000,
|
||||
})
|
||||
const scheduledAt = new Date(Date.now() + 3_600_000).toISOString()
|
||||
|
||||
await service.sendSMS(mockDeviceId, { ...mockSmsInput, scheduledAt })
|
||||
|
||||
const [, fcmMessages, , delayMs] =
|
||||
mockSmsQueueService.addSendSmsJob.mock.calls[0]
|
||||
expect(delayMs).toBeGreaterThan(3_500_000)
|
||||
expect(fcmMessages[0].android.ttl).toBeGreaterThan(72 * 3600 * 1000)
|
||||
// a single scheduled wave is still stamped so the stale cron waits for it
|
||||
expect(mockSmsModel.bulkWrite).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('should handle queue error properly', async () => {
|
||||
@@ -982,7 +1072,11 @@ describe('GatewayService', () => {
|
||||
|
||||
it('should queue bulk SMS if queue is enabled', async () => {
|
||||
mockSmsQueueService.isQueueEnabled.mockReturnValue(true)
|
||||
mockSmsQueueService.addSendSmsJob.mockResolvedValue(true)
|
||||
mockSmsQueueService.addSendSmsJob.mockResolvedValue({
|
||||
waves: [{ start: 0, end: 2, delayMs: 0 }],
|
||||
sendDelaySeconds: 5,
|
||||
projectedCompletionMs: 10_000,
|
||||
})
|
||||
|
||||
const result = await service.sendBulkSMS(mockDeviceId, mockBulkSmsInput)
|
||||
|
||||
@@ -990,6 +1084,65 @@ describe('GatewayService', () => {
|
||||
expect(mockSmsQueueService.addSendSmsJob).toHaveBeenCalled()
|
||||
expect(result).toHaveProperty('success', true)
|
||||
expect(result).toHaveProperty('smsBatchId', mockSmsBatch._id)
|
||||
expect(result).not.toHaveProperty('estimatedCompletionAt')
|
||||
// the batch is marked processing before the waves are queued
|
||||
expect(mockSmsBatchModel.findByIdAndUpdate).toHaveBeenCalledWith(
|
||||
mockSmsBatch._id,
|
||||
{ $set: { status: 'processing' } },
|
||||
)
|
||||
expect(mockSmsModel.bulkWrite).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('paces a large bulk send per scheduled group and reports the latest estimate', async () => {
|
||||
mockDeviceModel.findById.mockResolvedValue({
|
||||
...mockDevice,
|
||||
smsSendDelaySeconds: 5,
|
||||
})
|
||||
mockSmsQueueService.isQueueEnabled.mockReturnValue(true)
|
||||
let created = 0
|
||||
mockSmsModel.create.mockImplementation(async () => ({
|
||||
...mockSms,
|
||||
_id: `sms-${created++}`,
|
||||
}))
|
||||
mockSmsQueueService.addSendSmsJob
|
||||
.mockResolvedValueOnce({
|
||||
waves: [
|
||||
{ start: 0, end: 2, delayMs: 0 },
|
||||
{ start: 2, end: 3, delayMs: 10_000 },
|
||||
],
|
||||
sendDelaySeconds: 5,
|
||||
projectedCompletionMs: 15_000,
|
||||
})
|
||||
.mockResolvedValueOnce({
|
||||
waves: [{ start: 0, end: 1, delayMs: 60_000 }],
|
||||
sendDelaySeconds: 5,
|
||||
projectedCompletionMs: 65_000,
|
||||
})
|
||||
const scheduledAt = new Date(Date.now() + 60_000).toISOString()
|
||||
|
||||
const result = await service.sendBulkSMS(mockDeviceId, {
|
||||
messageTemplate: 'Hi',
|
||||
messages: [
|
||||
{ message: 'Hi', recipients: ['+15550100', '+15550101', '+15550102'] },
|
||||
{ message: 'Later', recipients: ['+15550103'], scheduledAt },
|
||||
],
|
||||
} as any)
|
||||
|
||||
expect(mockSmsQueueService.addSendSmsJob).toHaveBeenCalledTimes(2)
|
||||
expect(mockSmsQueueService.addSendSmsJob.mock.calls[0][4]).toBe(5)
|
||||
expect(mockSmsQueueService.addSendSmsJob.mock.calls[1][3]).toBeGreaterThan(50_000)
|
||||
|
||||
expect(mockSmsModel.bulkWrite).toHaveBeenCalledTimes(2)
|
||||
const firstOps = mockSmsModel.bulkWrite.mock.calls[0][0]
|
||||
expect(firstOps[0].updateMany.filter).toEqual({ _id: { $in: ['sms-0', 'sms-1'] } })
|
||||
expect(firstOps[1].updateMany.filter).toEqual({ _id: { $in: ['sms-2'] } })
|
||||
const secondOps = mockSmsModel.bulkWrite.mock.calls[1][0]
|
||||
expect(secondOps[0].updateMany.filter).toEqual({ _id: { $in: ['sms-3'] } })
|
||||
|
||||
expect(result.recipientCount).toBe(4)
|
||||
const eta = Date.parse(result.estimatedCompletionAt)
|
||||
const due0 = firstOps[0].updateMany.update.$set.dispatchDueAt.getTime()
|
||||
expect(eta - due0).toBe(65_000)
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@@ -29,6 +29,8 @@ import { normalizeOsFields } from './os-version'
|
||||
import { encodeCursor } from './cursor'
|
||||
import { toDirection, toStoredType } from './message-direction'
|
||||
import { ParsedMessageQuery } from './message-query'
|
||||
import { smsAndroidConfig } from './fcm-push-options'
|
||||
import { DispatchPlan } from './queue/dispatch-pacing'
|
||||
|
||||
@Injectable()
|
||||
export class GatewayService {
|
||||
@@ -411,6 +413,28 @@ export class GatewayService {
|
||||
}
|
||||
}
|
||||
|
||||
// Records when each paced wave is due so the stale-status cron leaves
|
||||
// messages alone while they legitimately wait in the queue.
|
||||
private async stampDispatchDueAt(
|
||||
smsIds: Types.ObjectId[],
|
||||
plan: DispatchPlan,
|
||||
now: number,
|
||||
): Promise<void> {
|
||||
const paced = plan.waves.length > 1 || plan.waves[0]?.delayMs > 0
|
||||
if (!paced || smsIds.length === 0) {
|
||||
return
|
||||
}
|
||||
await this.smsModel.bulkWrite(
|
||||
plan.waves.map((wave) => ({
|
||||
updateMany: {
|
||||
filter: { _id: { $in: smsIds.slice(wave.start, wave.end) } },
|
||||
update: { $set: { dispatchDueAt: new Date(now + wave.delayMs) } },
|
||||
},
|
||||
})),
|
||||
{ ordered: false },
|
||||
)
|
||||
}
|
||||
|
||||
async sendSMS(deviceId: string, smsData: SendSMSInputDTO): Promise<any> {
|
||||
const device = await this.deviceModel.findById(deviceId)
|
||||
|
||||
@@ -492,6 +516,7 @@ export class GatewayService {
|
||||
}
|
||||
|
||||
const fcmMessages: Message[] = []
|
||||
const smsIds: Types.ObjectId[] = []
|
||||
|
||||
for (let recipient of recipients) {
|
||||
recipient = recipient.replace(/\s+/g, "")
|
||||
@@ -528,11 +553,10 @@ export class GatewayService {
|
||||
smsData: stringifiedSMSData,
|
||||
},
|
||||
token: device.fcmToken,
|
||||
android: {
|
||||
priority: 'high',
|
||||
},
|
||||
android: smsAndroidConfig(smsData.scheduledAt),
|
||||
}
|
||||
fcmMessages.push(fcmMessage)
|
||||
smsIds.push(sms._id)
|
||||
}
|
||||
|
||||
// Check if we should use the queue
|
||||
@@ -544,18 +568,26 @@ export class GatewayService {
|
||||
})
|
||||
|
||||
// Add to queue
|
||||
await this.smsQueueService.addSendSmsJob(
|
||||
const queuedAt = Date.now()
|
||||
const plan = await this.smsQueueService.addSendSmsJob(
|
||||
deviceId,
|
||||
fcmMessages,
|
||||
smsBatch._id.toString(),
|
||||
delayMs,
|
||||
device.smsSendDelaySeconds,
|
||||
)
|
||||
await this.stampDispatchDueAt(smsIds, plan, queuedAt)
|
||||
|
||||
return {
|
||||
success: true,
|
||||
message: 'SMS added to queue for processing',
|
||||
smsBatchId: smsBatch._id,
|
||||
recipientCount: recipients.length,
|
||||
...(plan.waves.length > 1 && {
|
||||
estimatedCompletionAt: new Date(
|
||||
queuedAt + plan.projectedCompletionMs,
|
||||
).toISOString(),
|
||||
}),
|
||||
}
|
||||
} catch (e) {
|
||||
// Update batch status to failed
|
||||
@@ -697,13 +729,18 @@ export class GatewayService {
|
||||
})
|
||||
|
||||
// Track FCM messages with their calculated delays for grouping
|
||||
const fcmMessagesWithDelays: Array<{ message: Message; delayMs?: number }> = []
|
||||
const fcmMessagesWithDelays: Array<{
|
||||
message: Message
|
||||
delayMs?: number
|
||||
smsId: Types.ObjectId
|
||||
}> = []
|
||||
const smsDocumentsToInsert: Array<Record<string, any>> = []
|
||||
const smsToFcmMetadata: Array<{
|
||||
recipient: string
|
||||
message: string
|
||||
simSubscriptionId?: number
|
||||
delayMs?: number
|
||||
scheduledAt?: string
|
||||
}> = []
|
||||
|
||||
for (const smsData of messages) {
|
||||
@@ -743,6 +780,7 @@ export class GatewayService {
|
||||
simSubscriptionId: smsData.simSubscriptionId,
|
||||
}),
|
||||
delayMs,
|
||||
scheduledAt: smsData.scheduledAt,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -798,33 +836,54 @@ export class GatewayService {
|
||||
smsData: stringifiedSMSData,
|
||||
},
|
||||
token: device.fcmToken,
|
||||
android: {
|
||||
priority: 'high',
|
||||
},
|
||||
android: smsAndroidConfig(metadata.scheduledAt),
|
||||
}
|
||||
fcmMessagesWithDelays.push({ message: fcmMessage, delayMs: metadata.delayMs })
|
||||
fcmMessagesWithDelays.push({
|
||||
message: fcmMessage,
|
||||
delayMs: metadata.delayMs,
|
||||
smsId: sms._id,
|
||||
})
|
||||
}
|
||||
|
||||
// Check if we should use the queue
|
||||
if (this.smsQueueService.isQueueEnabled()) {
|
||||
try {
|
||||
await this.smsBatchModel.findByIdAndUpdate(smsBatch._id, {
|
||||
$set: { status: 'processing' },
|
||||
})
|
||||
|
||||
// Group messages by delay (undefined delay means immediate, group together)
|
||||
const messagesByDelay = new Map<number | undefined, Message[]>()
|
||||
for (const { message, delayMs } of fcmMessagesWithDelays) {
|
||||
const messagesByDelay = new Map<
|
||||
number | undefined,
|
||||
{ messages: Message[]; smsIds: Types.ObjectId[] }
|
||||
>()
|
||||
for (const { message, delayMs, smsId } of fcmMessagesWithDelays) {
|
||||
const delayKey = delayMs !== undefined ? delayMs : undefined
|
||||
if (!messagesByDelay.has(delayKey)) {
|
||||
messagesByDelay.set(delayKey, [])
|
||||
messagesByDelay.set(delayKey, { messages: [], smsIds: [] })
|
||||
}
|
||||
messagesByDelay.get(delayKey)!.push(message)
|
||||
const group = messagesByDelay.get(delayKey)!
|
||||
group.messages.push(message)
|
||||
group.smsIds.push(smsId)
|
||||
}
|
||||
|
||||
// Queue each group with its respective delay
|
||||
for (const [delayMs, messages] of messagesByDelay.entries()) {
|
||||
await this.smsQueueService.addSendSmsJob(
|
||||
// Queue each group with its respective delay, paced per group
|
||||
const queuedAt = Date.now()
|
||||
let multiWave = false
|
||||
let projectedCompletionMs = 0
|
||||
for (const [delayMs, group] of messagesByDelay.entries()) {
|
||||
const plan = await this.smsQueueService.addSendSmsJob(
|
||||
deviceId,
|
||||
messages,
|
||||
group.messages,
|
||||
smsBatch._id.toString(),
|
||||
delayMs,
|
||||
device.smsSendDelaySeconds,
|
||||
)
|
||||
await this.stampDispatchDueAt(group.smsIds, plan, queuedAt)
|
||||
multiWave = multiWave || plan.waves.length > 1
|
||||
projectedCompletionMs = Math.max(
|
||||
projectedCompletionMs,
|
||||
plan.projectedCompletionMs,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -833,6 +892,11 @@ export class GatewayService {
|
||||
message: 'Bulk SMS added to queue for processing',
|
||||
smsBatchId: smsBatch._id,
|
||||
recipientCount: messages.map((m) => m.recipients).flat().length,
|
||||
...(multiWave && {
|
||||
estimatedCompletionAt: new Date(
|
||||
queuedAt + projectedCompletionMs,
|
||||
).toISOString(),
|
||||
}),
|
||||
}
|
||||
} catch (e) {
|
||||
// Update batch status to failed
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
import { planDispatchWaves, resolveSendDelaySeconds } from './dispatch-pacing'
|
||||
|
||||
describe('planDispatchWaves', () => {
|
||||
it('returns a single wave at the base delay when the batch fits the window', () => {
|
||||
const plan = planDispatchWaves(50, { waveSize: 50, sendDelaySeconds: 5 })
|
||||
|
||||
expect(plan.waves).toEqual([{ start: 0, end: 50, delayMs: 0 }])
|
||||
expect(plan.projectedCompletionMs).toBe(50 * 5000)
|
||||
})
|
||||
|
||||
it('splits 2000 messages at 5s into 40 waves spaced 250s apart', () => {
|
||||
const plan = planDispatchWaves(2000, { waveSize: 50, sendDelaySeconds: 5 })
|
||||
|
||||
expect(plan.waves).toHaveLength(40)
|
||||
expect(plan.waves[0]).toEqual({ start: 0, end: 50, delayMs: 0 })
|
||||
expect(plan.waves[1].delayMs).toBe(250_000)
|
||||
expect(plan.waves[39]).toEqual({
|
||||
start: 1950,
|
||||
end: 2000,
|
||||
delayMs: 39 * 250_000,
|
||||
})
|
||||
expect(plan.projectedCompletionMs).toBe(40 * 250_000)
|
||||
})
|
||||
|
||||
it('covers every message exactly once, including a short final wave', () => {
|
||||
const plan = planDispatchWaves(51, { waveSize: 50, sendDelaySeconds: 5 })
|
||||
|
||||
expect(plan.waves).toEqual([
|
||||
{ start: 0, end: 50, delayMs: 0 },
|
||||
{ start: 50, end: 51, delayMs: 250_000 },
|
||||
])
|
||||
expect(plan.projectedCompletionMs).toBe(250_000 + 5000)
|
||||
|
||||
const seen = new Set<number>()
|
||||
for (const wave of plan.waves) {
|
||||
for (let i = wave.start; i < wave.end; i++) {
|
||||
expect(seen.has(i)).toBe(false)
|
||||
seen.add(i)
|
||||
}
|
||||
}
|
||||
expect(seen.size).toBe(51)
|
||||
})
|
||||
|
||||
it('adds a scheduled base delay to every wave instead of replacing it', () => {
|
||||
const plan = planDispatchWaves(120, {
|
||||
waveSize: 50,
|
||||
sendDelaySeconds: 5,
|
||||
baseDelayMs: 60_000,
|
||||
})
|
||||
|
||||
expect(plan.waves.map((w) => w.delayMs)).toEqual([
|
||||
60_000,
|
||||
310_000,
|
||||
560_000,
|
||||
])
|
||||
})
|
||||
|
||||
it.each([undefined, 0, -3, NaN])(
|
||||
'falls back to the default 5s send delay for %p',
|
||||
(sendDelaySeconds) => {
|
||||
const plan = planDispatchWaves(100, { waveSize: 50, sendDelaySeconds })
|
||||
|
||||
expect(plan.sendDelaySeconds).toBe(5)
|
||||
expect(plan.waves[1].delayMs).toBe(250_000)
|
||||
},
|
||||
)
|
||||
|
||||
it('uses the device delay when it is set', () => {
|
||||
const plan = planDispatchWaves(100, { waveSize: 50, sendDelaySeconds: 10 })
|
||||
|
||||
expect(plan.waves[1].delayMs).toBe(500_000)
|
||||
})
|
||||
|
||||
it('treats a wave size below 1 as 1', () => {
|
||||
const plan = planDispatchWaves(3, { waveSize: 0, sendDelaySeconds: 5 })
|
||||
|
||||
expect(plan.waves.map((w) => [w.start, w.end, w.delayMs])).toEqual([
|
||||
[0, 1, 0],
|
||||
[1, 2, 5000],
|
||||
[2, 3, 10_000],
|
||||
])
|
||||
})
|
||||
|
||||
it('returns no waves for an empty batch', () => {
|
||||
const plan = planDispatchWaves(0, { waveSize: 50, sendDelaySeconds: 5 })
|
||||
|
||||
expect(plan.waves).toEqual([])
|
||||
expect(plan.projectedCompletionMs).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('resolveSendDelaySeconds', () => {
|
||||
it('keeps positive values and defaults everything else', () => {
|
||||
expect(resolveSendDelaySeconds(7)).toBe(7)
|
||||
expect(resolveSendDelaySeconds(0)).toBe(5)
|
||||
expect(resolveSendDelaySeconds(undefined)).toBe(5)
|
||||
expect(resolveSendDelaySeconds(-1)).toBe(5)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,57 @@
|
||||
import { DEFAULT_SMS_SEND_DELAY_SECONDS } from '../schemas/device.schema'
|
||||
|
||||
export const DEFAULT_BULK_DISPATCH_WINDOW = 50
|
||||
export const DEFAULT_BULK_DISPATCH_MAX_SPREAD_HOURS = 72
|
||||
|
||||
export interface DispatchWave {
|
||||
// [start, end) index range into the message list
|
||||
start: number
|
||||
end: number
|
||||
delayMs: number
|
||||
}
|
||||
|
||||
export interface DispatchPlan {
|
||||
waves: DispatchWave[]
|
||||
sendDelaySeconds: number
|
||||
// Time from now until the device is expected to finish the last wave
|
||||
projectedCompletionMs: number
|
||||
}
|
||||
|
||||
export function resolveSendDelaySeconds(sendDelaySeconds?: number): number {
|
||||
const value = Number(sendDelaySeconds)
|
||||
return Number.isFinite(value) && value > 0
|
||||
? value
|
||||
: DEFAULT_SMS_SEND_DELAY_SECONDS
|
||||
}
|
||||
|
||||
// Releases messages in waves spaced by the time the device needs to send the
|
||||
// previous wave, so in-flight pushes never pile up beyond one wave.
|
||||
export function planDispatchWaves(
|
||||
messageCount: number,
|
||||
opts: {
|
||||
waveSize: number
|
||||
sendDelaySeconds?: number
|
||||
baseDelayMs?: number
|
||||
},
|
||||
): DispatchPlan {
|
||||
const waveSize = Math.max(1, Math.floor(Number(opts.waveSize) || 1))
|
||||
const sendDelaySeconds = resolveSendDelaySeconds(opts.sendDelaySeconds)
|
||||
const baseDelayMs = Math.max(0, Number(opts.baseDelayMs) || 0)
|
||||
const waveSpacingMs = waveSize * sendDelaySeconds * 1000
|
||||
|
||||
const waves: DispatchWave[] = []
|
||||
for (let start = 0, i = 0; start < messageCount; start += waveSize, i++) {
|
||||
waves.push({
|
||||
start,
|
||||
end: Math.min(start + waveSize, messageCount),
|
||||
delayMs: baseDelayMs + i * waveSpacingMs,
|
||||
})
|
||||
}
|
||||
|
||||
const last = waves[waves.length - 1]
|
||||
const projectedCompletionMs = last
|
||||
? last.delayMs + (last.end - last.start) * sendDelaySeconds * 1000
|
||||
: baseDelayMs
|
||||
|
||||
return { waves, sendDelaySeconds, projectedCompletionMs }
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
import { resolveBatchStatus } from './sms-queue.processor'
|
||||
|
||||
describe('resolveBatchStatus', () => {
|
||||
it('keeps a paced batch in processing while waves are still queued', () => {
|
||||
expect(
|
||||
resolveBatchStatus({ recipientCount: 2000, successCount: 50, failureCount: 0 }),
|
||||
).toBe('processing')
|
||||
})
|
||||
|
||||
it('reports partial_success while still draining if some pushes failed', () => {
|
||||
expect(
|
||||
resolveBatchStatus({ recipientCount: 2000, successCount: 45, failureCount: 5 }),
|
||||
).toBe('partial_success')
|
||||
})
|
||||
|
||||
it('completes once every recipient was pushed without failures', () => {
|
||||
expect(
|
||||
resolveBatchStatus({ recipientCount: 100, successCount: 100, failureCount: 0 }),
|
||||
).toBe('completed')
|
||||
})
|
||||
|
||||
it('fails when every push failed', () => {
|
||||
expect(
|
||||
resolveBatchStatus({ recipientCount: 100, successCount: 0, failureCount: 100 }),
|
||||
).toBe('failed')
|
||||
})
|
||||
|
||||
it('is partial_success when finished with mixed results', () => {
|
||||
expect(
|
||||
resolveBatchStatus({ recipientCount: 100, successCount: 90, failureCount: 10 }),
|
||||
).toBe('partial_success')
|
||||
})
|
||||
})
|
||||
@@ -39,6 +39,21 @@ function getFcmErrorMessage(error: { code?: string; message?: string } | null |
|
||||
return `${rawPart} — ${FCM_ACTIONABLE_MESSAGE}`
|
||||
}
|
||||
|
||||
// A paced batch is still 'processing' until every wave has been handed to FCM
|
||||
export function resolveBatchStatus(batch: {
|
||||
recipientCount: number
|
||||
successCount: number
|
||||
failureCount: number
|
||||
}): 'processing' | 'completed' | 'partial_success' | 'failed' {
|
||||
const attempted = batch.successCount + batch.failureCount
|
||||
if (attempted < batch.recipientCount) {
|
||||
return batch.failureCount > 0 ? 'partial_success' : 'processing'
|
||||
}
|
||||
if (batch.failureCount === 0) return 'completed'
|
||||
if (batch.successCount === 0) return 'failed'
|
||||
return 'partial_success'
|
||||
}
|
||||
|
||||
@Processor('sms')
|
||||
export class SmsQueueProcessor {
|
||||
private readonly logger = new Logger(SmsQueueProcessor.name)
|
||||
@@ -197,12 +212,7 @@ export class SmsQueueProcessor {
|
||||
{ returnDocument: 'after' },
|
||||
)
|
||||
|
||||
const batchStatus =
|
||||
smsBatch.failureCount === smsBatch.recipientCount
|
||||
? 'failed'
|
||||
: smsBatch.successCount === smsBatch.recipientCount
|
||||
? 'completed'
|
||||
: 'partial_success'
|
||||
const batchStatus = resolveBatchStatus(smsBatch)
|
||||
await this.smsBatchModel.findByIdAndUpdate(smsBatchId, {
|
||||
$set: { status: batchStatus },
|
||||
})
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
import { Test, TestingModule } from '@nestjs/testing'
|
||||
import { getQueueToken } from '@nestjs/bull'
|
||||
import { ConfigService } from '@nestjs/config'
|
||||
import { Logger } from '@nestjs/common'
|
||||
import { Message } from 'firebase-admin/messaging'
|
||||
import { SmsQueueService } from './sms-queue.service'
|
||||
|
||||
function buildMessages(count: number): Message[] {
|
||||
return Array.from({ length: count }, (_, i) => ({
|
||||
token: 'token',
|
||||
data: { smsData: JSON.stringify({ smsId: `sms-${i}` }) },
|
||||
}))
|
||||
}
|
||||
|
||||
describe('SmsQueueService', () => {
|
||||
let service: SmsQueueService
|
||||
const queueAdd = jest.fn().mockResolvedValue(undefined)
|
||||
const config: Record<string, unknown> = {}
|
||||
|
||||
const mockConfigService = {
|
||||
get: jest.fn((key: string, fallback?: unknown) =>
|
||||
key in config ? config[key] : fallback,
|
||||
),
|
||||
}
|
||||
|
||||
async function build(overrides: Record<string, unknown> = {}) {
|
||||
for (const key of Object.keys(config)) delete config[key]
|
||||
Object.assign(config, { USE_SMS_QUEUE: true }, overrides)
|
||||
queueAdd.mockClear()
|
||||
|
||||
const module: TestingModule = await Test.createTestingModule({
|
||||
providers: [
|
||||
SmsQueueService,
|
||||
{ provide: getQueueToken('sms'), useValue: { add: queueAdd } },
|
||||
{ provide: ConfigService, useValue: mockConfigService },
|
||||
],
|
||||
}).compile()
|
||||
|
||||
service = module.get(SmsQueueService)
|
||||
}
|
||||
|
||||
beforeEach(() => build())
|
||||
|
||||
it('reports whether the queue is enabled', async () => {
|
||||
expect(service.isQueueEnabled()).toBe(true)
|
||||
await build({ USE_SMS_QUEUE: false })
|
||||
expect(service.isQueueEnabled()).toBe(false)
|
||||
})
|
||||
|
||||
it('enqueues a small batch as one immediate job, unchanged from before', async () => {
|
||||
const messages = buildMessages(10)
|
||||
|
||||
const plan = await service.addSendSmsJob('device-1', messages, 'batch-1')
|
||||
|
||||
expect(queueAdd).toHaveBeenCalledTimes(1)
|
||||
const [name, data, opts] = queueAdd.mock.calls[0]
|
||||
expect(name).toBe('send-sms')
|
||||
expect(data).toEqual({
|
||||
deviceId: 'device-1',
|
||||
fcmMessages: messages,
|
||||
smsBatchId: 'batch-1',
|
||||
})
|
||||
expect(opts).toMatchObject({ attempts: 1, delay: 0, priority: 1 })
|
||||
expect(plan.waves).toEqual([{ start: 0, end: 10, delayMs: 0 }])
|
||||
})
|
||||
|
||||
it('paces a large batch into waves spaced by the device send delay', async () => {
|
||||
const messages = buildMessages(120)
|
||||
|
||||
const plan = await service.addSendSmsJob(
|
||||
'device-1',
|
||||
messages,
|
||||
'batch-1',
|
||||
undefined,
|
||||
5,
|
||||
)
|
||||
|
||||
expect(queueAdd).toHaveBeenCalledTimes(3)
|
||||
expect(queueAdd.mock.calls.map(([, , opts]) => opts.delay)).toEqual([
|
||||
0, 250_000, 500_000,
|
||||
])
|
||||
expect(queueAdd.mock.calls.map(([, data]) => data.fcmMessages.length)).toEqual(
|
||||
[50, 50, 20],
|
||||
)
|
||||
expect(queueAdd.mock.calls[2][1].fcmMessages[0]).toBe(messages[100])
|
||||
expect(plan.waves).toHaveLength(3)
|
||||
expect(plan.projectedCompletionMs).toBe(500_000 + 20 * 5000)
|
||||
})
|
||||
|
||||
it('adds the scheduled delay as a base under the wave spacing', async () => {
|
||||
await service.addSendSmsJob('device-1', buildMessages(100), 'batch-1', 90_000, 5)
|
||||
|
||||
expect(queueAdd.mock.calls.map(([, , opts]) => opts.delay)).toEqual([
|
||||
90_000, 340_000,
|
||||
])
|
||||
})
|
||||
|
||||
it('falls back to SMS_QUEUE_IMMEDIATE_DELAY_MS as the base when not scheduled', async () => {
|
||||
await build({ SMS_QUEUE_IMMEDIATE_DELAY_MS: 1500 })
|
||||
|
||||
await service.addSendSmsJob('device-1', buildMessages(60), 'batch-1', undefined, 5)
|
||||
|
||||
expect(queueAdd.mock.calls.map(([, , opts]) => opts.delay)).toEqual([
|
||||
1500, 251_500,
|
||||
])
|
||||
})
|
||||
|
||||
it('caps the wave size at MAX_SMS_BATCH_SIZE and honours BULK_DISPATCH_WINDOW', async () => {
|
||||
await build({ MAX_SMS_BATCH_SIZE: 20, BULK_DISPATCH_WINDOW: 50 })
|
||||
await service.addSendSmsJob('device-1', buildMessages(40), 'batch-1', undefined, 5)
|
||||
expect(queueAdd.mock.calls.map(([, data]) => data.fcmMessages.length)).toEqual(
|
||||
[20, 20],
|
||||
)
|
||||
|
||||
await build({ BULK_DISPATCH_WINDOW: 10 })
|
||||
await service.addSendSmsJob('device-1', buildMessages(25), 'batch-1', undefined, 5)
|
||||
expect(queueAdd.mock.calls.map(([, , opts]) => opts.delay)).toEqual([
|
||||
0, 50_000, 100_000,
|
||||
])
|
||||
})
|
||||
|
||||
it('setting the window above the batch size disables pacing', async () => {
|
||||
await build({ MAX_SMS_BATCH_SIZE: 5000, BULK_DISPATCH_WINDOW: 5000 })
|
||||
|
||||
await service.addSendSmsJob('device-1', buildMessages(2000), 'batch-1', undefined, 5)
|
||||
|
||||
expect(queueAdd).toHaveBeenCalledTimes(1)
|
||||
expect(queueAdd.mock.calls[0][2].delay).toBe(0)
|
||||
})
|
||||
|
||||
it('warns but still enqueues every wave when the spread exceeds the cap', async () => {
|
||||
await build({ BULK_DISPATCH_MAX_SPREAD_HOURS: 1 })
|
||||
const warn = jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined)
|
||||
|
||||
await service.addSendSmsJob('device-1', buildMessages(1000), 'batch-1', undefined, 5)
|
||||
|
||||
expect(queueAdd).toHaveBeenCalledTimes(20)
|
||||
expect(warn).toHaveBeenCalledTimes(1)
|
||||
expect(warn.mock.calls[0][0]).toContain('batch-1')
|
||||
warn.mockRestore()
|
||||
})
|
||||
})
|
||||
@@ -3,6 +3,12 @@ import { InjectQueue } from '@nestjs/bull'
|
||||
import { Queue } from 'bull'
|
||||
import { ConfigService } from '@nestjs/config'
|
||||
import { Message } from 'firebase-admin/messaging'
|
||||
import {
|
||||
DEFAULT_BULK_DISPATCH_MAX_SPREAD_HOURS,
|
||||
DEFAULT_BULK_DISPATCH_WINDOW,
|
||||
DispatchPlan,
|
||||
planDispatchWaves,
|
||||
} from './dispatch-pacing'
|
||||
|
||||
@Injectable()
|
||||
export class SmsQueueService {
|
||||
@@ -10,6 +16,8 @@ export class SmsQueueService {
|
||||
private readonly useSmsQueue: boolean
|
||||
private readonly maxSmsBatchSize: number
|
||||
private readonly immediateQueueDelayMs: number
|
||||
private readonly bulkDispatchWindow: number
|
||||
private readonly bulkDispatchMaxSpreadMs: number
|
||||
|
||||
constructor(
|
||||
@InjectQueue('sms') private readonly smsQueue: Queue,
|
||||
@@ -24,6 +32,21 @@ export class SmsQueueService {
|
||||
'SMS_QUEUE_IMMEDIATE_DELAY_MS',
|
||||
0,
|
||||
)
|
||||
this.bulkDispatchWindow = Number(
|
||||
this.configService.get<number>(
|
||||
'BULK_DISPATCH_WINDOW',
|
||||
DEFAULT_BULK_DISPATCH_WINDOW,
|
||||
),
|
||||
)
|
||||
this.bulkDispatchMaxSpreadMs =
|
||||
Number(
|
||||
this.configService.get<number>(
|
||||
'BULK_DISPATCH_MAX_SPREAD_HOURS',
|
||||
DEFAULT_BULK_DISPATCH_MAX_SPREAD_HOURS,
|
||||
),
|
||||
) *
|
||||
3600 *
|
||||
1000
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -33,37 +56,46 @@ export class SmsQueueService {
|
||||
return this.useSmsQueue
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue pushes for one batch. Large batches are released in waves paced
|
||||
* to the device's send delay; the returned plan says when each wave is due.
|
||||
*/
|
||||
async addSendSmsJob(
|
||||
deviceId: string,
|
||||
fcmMessages: Message[],
|
||||
smsBatchId: string,
|
||||
delayMs?: number,
|
||||
) {
|
||||
// this.logger.debug(`Adding send-sms job for batch ${smsBatchId}`)
|
||||
|
||||
// Split messages into batches of max smsBatchSize messages
|
||||
const batches = []
|
||||
for (let i = 0; i < fcmMessages.length; i += this.maxSmsBatchSize) {
|
||||
batches.push(fcmMessages.slice(i, i + this.maxSmsBatchSize))
|
||||
}
|
||||
|
||||
// If delayMs is provided, use it for all batches (scheduled send)
|
||||
sendDelaySeconds?: number,
|
||||
): Promise<DispatchPlan> {
|
||||
// If delayMs is provided, use it as the base for all waves (scheduled send)
|
||||
// Otherwise rely on queue limiter/concurrency and optionally fixed jitter.
|
||||
const useScheduledDelay = delayMs !== undefined && delayMs >= 0
|
||||
const baseDelayMs = useScheduledDelay ? delayMs : this.immediateQueueDelayMs
|
||||
|
||||
for (const batch of batches) {
|
||||
const delay = useScheduledDelay ? delayMs : this.immediateQueueDelayMs
|
||||
const plan = planDispatchWaves(fcmMessages.length, {
|
||||
waveSize: Math.min(this.maxSmsBatchSize, this.bulkDispatchWindow),
|
||||
sendDelaySeconds,
|
||||
baseDelayMs,
|
||||
})
|
||||
|
||||
if (plan.projectedCompletionMs - baseDelayMs > this.bulkDispatchMaxSpreadMs) {
|
||||
this.logger.warn(
|
||||
`Batch ${smsBatchId}: ${fcmMessages.length} messages at ${plan.sendDelaySeconds}s/message are projected to take ${Math.round(plan.projectedCompletionMs / 3600000)}h to dispatch`,
|
||||
)
|
||||
}
|
||||
|
||||
for (const wave of plan.waves) {
|
||||
await this.smsQueue.add(
|
||||
'send-sms',
|
||||
{
|
||||
deviceId,
|
||||
fcmMessages: batch,
|
||||
fcmMessages: fcmMessages.slice(wave.start, wave.end),
|
||||
smsBatchId,
|
||||
},
|
||||
{
|
||||
priority: 1, // TODO: Make this dynamic based on users subscription plan
|
||||
attempts: 1,
|
||||
delay: delay,
|
||||
delay: wave.delayMs,
|
||||
backoff: {
|
||||
type: 'exponential',
|
||||
delay: 5000, // 5 seconds
|
||||
@@ -73,5 +105,7 @@ export class SmsQueueService {
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
return plan
|
||||
}
|
||||
}
|
||||
|
||||
@@ -46,6 +46,11 @@ export class SMS {
|
||||
@Prop({ type: Date })
|
||||
requestedAt: Date
|
||||
|
||||
// When the queue is due to hand this message to the push service; set only
|
||||
// for paced bulk sends that wait server-side before dispatch
|
||||
@Prop({ type: Date })
|
||||
dispatchDueAt?: Date
|
||||
|
||||
@Prop({ type: Date })
|
||||
dispatchedAt: Date
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ import { Model, Types } from 'mongoose'
|
||||
import { Device, DeviceDocument } from '../schemas/device.schema'
|
||||
import * as firebaseAdmin from 'firebase-admin'
|
||||
import { Message } from 'firebase-admin/messaging'
|
||||
import { heartbeatAndroidConfig } from '../fcm-push-options'
|
||||
|
||||
const FCM_BATCH_SIZE = 500
|
||||
|
||||
@@ -93,9 +94,7 @@ export class HeartbeatCheckTask {
|
||||
type: 'heartbeat_check',
|
||||
},
|
||||
token: device.fcmToken,
|
||||
android: {
|
||||
priority: 'high',
|
||||
},
|
||||
android: heartbeatAndroidConfig(),
|
||||
}
|
||||
|
||||
fcmMessages.push(fcmMessage)
|
||||
|
||||
@@ -59,6 +59,30 @@ describe('SmsStatusUpdateTask', () => {
|
||||
},
|
||||
);
|
||||
|
||||
// Paced messages whose wave is not yet due must be left alone; legacy
|
||||
// rows without dispatchDueAt keep today's behavior
|
||||
const pendingFilter = (smsModel.updateMany as jest.Mock).mock.calls[0][0];
|
||||
const cutoff = pendingFilter.requestedAt.$lt as Date;
|
||||
expect(pendingFilter.$or).toEqual([
|
||||
{ dispatchDueAt: { $exists: false } },
|
||||
{ dispatchDueAt: { $lt: cutoff } },
|
||||
]);
|
||||
expect(Date.now() - cutoff.getTime()).toBeGreaterThanOrEqual(20 * 60 * 1000 - 1000);
|
||||
|
||||
// The dispatched sweep keys off dispatchedAt and is unchanged
|
||||
expect(smsModel.updateMany).toHaveBeenCalledWith(
|
||||
{
|
||||
status: 'dispatched',
|
||||
dispatchedAt: expect.any(Object),
|
||||
},
|
||||
{
|
||||
$set: {
|
||||
status: 'unknown',
|
||||
errorMessage: 'Status update timeout - no response from device after dispatch',
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
// Check that SMSBatch model was updated with correct query
|
||||
expect(smsBatchModel.updateMany).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
|
||||
@@ -31,6 +31,11 @@ export class SmsStatusUpdateTask {
|
||||
{
|
||||
status: 'pending',
|
||||
requestedAt: { $lt: twentyMinutesAgo },
|
||||
// Paced messages are not stale until their own wave was due
|
||||
$or: [
|
||||
{ dispatchDueAt: { $exists: false } },
|
||||
{ dispatchDueAt: { $lt: twentyMinutesAgo } },
|
||||
],
|
||||
},
|
||||
{
|
||||
$set: {
|
||||
|
||||
Reference in New Issue
Block a user