From b344ef2dee89060167d59743a7c7db2f6cd44caa Mon Sep 17 00:00:00 2001 From: Aashish John Date: Wed, 29 Jul 2026 12:06:26 -0400 Subject: [PATCH] feat: allow client to request send after --- migrations/20260729154654-add-send-after.js | 49 +++++++++ .../20260729154654-add-send-after-down.sql | 83 ++++++++++++++ .../sqls/20260729154654-add-send-after-up.sql | 101 ++++++++++++++++++ schema-dump.sql | 27 +++-- src/jobs/process-10dlc-message.spec.ts | 45 ++++++++ src/jobs/process-10dlc-message.ts | 4 +- src/jobs/send-message.spec.ts | 50 ++++++++- src/lib/process-message.ts | 1 + 8 files changed, 345 insertions(+), 15 deletions(-) create mode 100644 migrations/20260729154654-add-send-after.js create mode 100644 migrations/sqls/20260729154654-add-send-after-down.sql create mode 100644 migrations/sqls/20260729154654-add-send-after-up.sql diff --git a/migrations/20260729154654-add-send-after.js b/migrations/20260729154654-add-send-after.js new file mode 100644 index 0000000..097fa95 --- /dev/null +++ b/migrations/20260729154654-add-send-after.js @@ -0,0 +1,49 @@ +'use strict'; + +var dbm; +var type; +var seed; +var fs = require('fs'); +var path = require('path'); +var Promise; + +/** + * We receive the dbmigrate dependency from dbmigrate initially. + * This enables us to not have to rely on NODE_PATH. + */ +exports.setup = function(options, seedLink) { + dbm = options.dbmigrate; + type = dbm.dataType; + seed = seedLink; + Promise = options.Promise; +}; + +exports.up = function(db) { + var filePath = path.join(__dirname, 'sqls', '20260729154654-add-send-after-up.sql'); + return new Promise( function( resolve, reject ) { + fs.readFile(filePath, {encoding: 'utf-8'}, function(err,data){ + if (err) return reject(err); + resolve(data); + }); + }) + .then(function(data) { + return db.runSql(data); + }); +}; + +exports.down = function(db) { + var filePath = path.join(__dirname, 'sqls', '20260729154654-add-send-after-down.sql'); + return new Promise( function( resolve, reject ) { + fs.readFile(filePath, {encoding: 'utf-8'}, function(err,data){ + if (err) return reject(err); + resolve(data); + }); + }) + .then(function(data) { + return db.runSql(data); + }); +}; + +exports._meta = { + "version": 1 +}; diff --git a/migrations/sqls/20260729154654-add-send-after-down.sql b/migrations/sqls/20260729154654-add-send-after-down.sql new file mode 100644 index 0000000..6740926 --- /dev/null +++ b/migrations/sqls/20260729154654-add-send-after-down.sql @@ -0,0 +1,83 @@ +drop function sms.send_message; + +CREATE OR REPLACE FUNCTION sms.send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code DEFAULT NULL::text, send_before timestamp without time zone DEFAULT NULL::timestamp without time zone) RETURNS sms.outbound_messages + LANGUAGE plpgsql SECURITY DEFINER + AS $$ +declare + v_client_id uuid; + v_profile_id uuid; + v_profile_active boolean; + v_contact_zip_code zip_code; + v_estimated_segments integer; + v_result sms.outbound_messages; +begin + select billing.current_client_id() into v_client_id; + + if v_client_id is null then + raise 'Not authorized'; + end if; + + select id, active + from sms.profiles + where client_id = v_client_id + and id = send_message.profile_id + into v_profile_id, v_profile_active; + + if v_profile_id is null then + raise 'Profile % not found – it may not exist, or you may not have access', send_message.profile_id using errcode = 'no_data_found'; + end if; + + if v_profile_active is distinct from true then + raise 'Profile % is inactive', send_message.profile_id; + end if; + + if contact_zip_code is null or contact_zip_code = '' then + select sms.map_area_code_to_zip_code(sms.extract_area_code(send_message.to)) into v_contact_zip_code; + else + select contact_zip_code into v_contact_zip_code; + end if; + + select sms.estimate_segments(body) into v_estimated_segments; + + insert into sms.outbound_messages (profile_id, created_at, to_number, stage, body, media_urls, contact_zip_code, estimated_segments, send_before) + values (send_message.profile_id, date_trunc('second', now()), send_message.to, 'processing', body, media_urls, v_contact_zip_code, v_estimated_segments, send_message.send_before) + returning * + into v_result; + + return v_result; +end; +$$; + +ALTER FUNCTION sms.send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code, send_before timestamp without time zone) OWNER TO postgres; + +GRANT ALL ON FUNCTION sms.send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code, send_before timestamp without time zone) TO client; + +CREATE OR REPLACE FUNCTION sms.tg__trigger_process_message() RETURNS trigger + LANGUAGE plpgsql + AS $$ +declare + v_channel sms.traffic_channel; + v_job json; +begin + select coalesce(channel, 'grey-route'::sms.traffic_channel) + from sms.profiles + where id = NEW.profile_id + into v_channel; + + select row_to_json(NEW) into v_job; + + if v_channel = 'grey-route'::sms.traffic_channel then + perform graphile_worker.add_job(identifier => 'process-grey-route-message', payload => v_job, run_at => null, max_attempts => 5); + elsif v_channel = 'toll-free'::sms.traffic_channel then + perform graphile_worker.add_job(identifier => 'process-toll-free-message', payload => v_job, run_at => null, max_attempts => 5); + elsif v_channel = '10dlc'::sms.traffic_channel then + perform graphile_worker.add_job(identifier => 'process-10dlc-message', payload => v_job, run_at => null, max_attempts => 5); + else + raise 'Unsupported traffic channel %', v_channel; + end if; + + return NEW; +end; +$$; + +alter table sms.outbound_messages drop column send_after; diff --git a/migrations/sqls/20260729154654-add-send-after-up.sql b/migrations/sqls/20260729154654-add-send-after-up.sql new file mode 100644 index 0000000..5e1f431 --- /dev/null +++ b/migrations/sqls/20260729154654-add-send-after-up.sql @@ -0,0 +1,101 @@ +alter table sms.outbound_messages add column send_after timestamp; + +drop function sms.send_message; + +CREATE OR REPLACE FUNCTION sms.send_message( + profile_id uuid, + "to" public.phone_number, + body text, + media_urls public.url[], + contact_zip_code public.zip_code DEFAULT NULL::text, + send_before timestamp without time zone DEFAULT NULL::timestamp without time zone, + send_after timestamp without time zone DEFAULT NULL::timestamp without time zone +) RETURNS sms.outbound_messages + LANGUAGE plpgsql SECURITY DEFINER + AS $$ +declare + v_client_id uuid; + v_profile_id uuid; + v_profile_active boolean; + v_contact_zip_code zip_code; + v_estimated_segments integer; + v_result sms.outbound_messages; +begin + select billing.current_client_id() into v_client_id; + + if v_client_id is null then + raise 'Not authorized'; + end if; + + select id, active + from sms.profiles + where client_id = v_client_id + and id = send_message.profile_id + into v_profile_id, v_profile_active; + + if v_profile_id is null then + raise 'Profile % not found – it may not exist, or you may not have access', send_message.profile_id using errcode = 'no_data_found'; + end if; + + if v_profile_active is distinct from true then + raise 'Profile % is inactive', send_message.profile_id; + end if; + + if send_message.send_after is not null + and send_message.send_before is not null + and send_message.send_after >= send_message.send_before then + raise 'send_after must be before send_before'; + end if; + + if contact_zip_code is null or contact_zip_code = '' then + select sms.map_area_code_to_zip_code(sms.extract_area_code(send_message.to)) into v_contact_zip_code; + else + select contact_zip_code into v_contact_zip_code; + end if; + + select sms.estimate_segments(body) into v_estimated_segments; + + insert into sms.outbound_messages (profile_id, created_at, to_number, stage, body, media_urls, contact_zip_code, estimated_segments, send_before, send_after) + values (send_message.profile_id, date_trunc('second', now()), send_message.to, 'processing', body, media_urls, v_contact_zip_code, v_estimated_segments, send_message.send_before, send_message.send_after) + returning * + into v_result; + + return v_result; +end; +$$; + +ALTER FUNCTION sms.send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code, send_before timestamp without time zone, send_after timestamp without time zone) OWNER TO postgres; + +GRANT ALL ON FUNCTION sms.send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code, send_before timestamp without time zone, send_after timestamp without time zone) TO client; + +CREATE OR REPLACE FUNCTION sms.tg__trigger_process_message() RETURNS trigger + LANGUAGE plpgsql + AS $$ +declare + v_channel sms.traffic_channel; + v_job json; +begin + select coalesce(channel, 'grey-route'::sms.traffic_channel) + from sms.profiles + where id = NEW.profile_id + into v_channel; + + if NEW.send_after is not null and v_channel <> '10dlc'::sms.traffic_channel then + raise 'send_after is only supported for the 10dlc channel'; + end if; + + select row_to_json(NEW) into v_job; + + if v_channel = 'grey-route'::sms.traffic_channel then + perform graphile_worker.add_job(identifier => 'process-grey-route-message', payload => v_job, run_at => null, max_attempts => 5); + elsif v_channel = 'toll-free'::sms.traffic_channel then + perform graphile_worker.add_job(identifier => 'process-toll-free-message', payload => v_job, run_at => null, max_attempts => 5); + elsif v_channel = '10dlc'::sms.traffic_channel then + perform graphile_worker.add_job(identifier => 'process-10dlc-message', payload => v_job, run_at => null, max_attempts => 5); + else + raise 'Unsupported traffic channel %', v_channel; + end if; + + return NEW; +end; +$$; diff --git a/schema-dump.sql b/schema-dump.sql index 05f7c86..8315745 100644 --- a/schema-dump.sql +++ b/schema-dump.sql @@ -1889,7 +1889,8 @@ CREATE TABLE sms.outbound_messages ( media_urls public.url[], estimated_segments integer DEFAULT 1, profile_id uuid, - send_before timestamp without time zone + send_before timestamp without time zone, + send_after timestamp without time zone ) WITH (autovacuum_vacuum_threshold='50000', autovacuum_vacuum_scale_factor='0', autovacuum_vacuum_cost_limit='1000', autovacuum_vacuum_cost_delay='0'); @@ -2369,10 +2370,10 @@ COMMENT ON FUNCTION sms.sell_cordoned_numbers(n_days integer) IS '@omit'; -- --- Name: send_message(uuid, public.phone_number, text, public.url[], public.zip_code, timestamp without time zone); Type: FUNCTION; Schema: sms; Owner: postgres +-- Name: send_message(uuid, public.phone_number, text, public.url[], public.zip_code, timestamp without time zone, timestamp without time zone); Type: FUNCTION; Schema: sms; Owner: postgres -- -CREATE FUNCTION sms.send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code DEFAULT NULL::text, send_before timestamp without time zone DEFAULT NULL::timestamp without time zone) RETURNS sms.outbound_messages +CREATE FUNCTION sms.send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code DEFAULT NULL::text, send_before timestamp without time zone DEFAULT NULL::timestamp without time zone, send_after timestamp without time zone DEFAULT NULL::timestamp without time zone) RETURNS sms.outbound_messages LANGUAGE plpgsql SECURITY DEFINER AS $$ declare @@ -2403,6 +2404,12 @@ begin raise 'Profile % is inactive', send_message.profile_id; end if; + if send_message.send_after is not null + and send_message.send_before is not null + and send_message.send_after >= send_message.send_before then + raise 'send_after must be before send_before'; + end if; + if contact_zip_code is null or contact_zip_code = '' then select sms.map_area_code_to_zip_code(sms.extract_area_code(send_message.to)) into v_contact_zip_code; else @@ -2411,8 +2418,8 @@ begin select sms.estimate_segments(body) into v_estimated_segments; - insert into sms.outbound_messages (profile_id, created_at, to_number, stage, body, media_urls, contact_zip_code, estimated_segments, send_before) - values (send_message.profile_id, date_trunc('second', now()), send_message.to, 'processing', body, media_urls, v_contact_zip_code, v_estimated_segments, send_message.send_before) + insert into sms.outbound_messages (profile_id, created_at, to_number, stage, body, media_urls, contact_zip_code, estimated_segments, send_before, send_after) + values (send_message.profile_id, date_trunc('second', now()), send_message.to, 'processing', body, media_urls, v_contact_zip_code, v_estimated_segments, send_message.send_before, send_message.send_after) returning * into v_result; @@ -2421,7 +2428,7 @@ end; $$; -ALTER FUNCTION sms.send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code, send_before timestamp without time zone) OWNER TO postgres; +ALTER FUNCTION sms.send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code, send_before timestamp without time zone, send_after timestamp without time zone) OWNER TO postgres; -- -- Name: sending_locations; Type: TABLE; Schema: sms; Owner: postgres @@ -2845,6 +2852,10 @@ begin where id = NEW.profile_id into v_channel; + if NEW.send_after is not null and v_channel <> '10dlc'::sms.traffic_channel then + raise 'send_after is only supported for the 10dlc channel'; + end if; + select row_to_json(NEW) into v_job; if v_channel = 'grey-route'::sms.traffic_channel then @@ -5556,10 +5567,10 @@ GRANT SELECT,INSERT,UPDATE ON TABLE lookup.requests TO client; -- --- Name: FUNCTION send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code, send_before timestamp without time zone); Type: ACL; Schema: sms; Owner: postgres +-- Name: FUNCTION send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code, send_before timestamp without time zone, send_after timestamp without time zone); Type: ACL; Schema: sms; Owner: postgres -- -GRANT ALL ON FUNCTION sms.send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code, send_before timestamp without time zone) TO client; +GRANT ALL ON FUNCTION sms.send_message(profile_id uuid, "to" public.phone_number, body text, media_urls public.url[], contact_zip_code public.zip_code, send_before timestamp without time zone, send_after timestamp without time zone) TO client; -- diff --git a/src/jobs/process-10dlc-message.spec.ts b/src/jobs/process-10dlc-message.spec.ts index e7af45a..41e237d 100644 --- a/src/jobs/process-10dlc-message.spec.ts +++ b/src/jobs/process-10dlc-message.spec.ts @@ -111,4 +111,49 @@ describe('process message', () => { expect(m.sending_location_id).not.toBeNull(); expect(m.from_number).toBe(fromNumber); }); + + test('it should carry a requested send_after through to outbound_messages_routing', async () => { + const fromNumber = fakeNumber('877'); + const toNumber = fakeNumber(); + const sendAfter = new Date(Date.now() + 60 * 60 * 1000); + + const m = await withClient(pool, async (client) => { + const { profileId } = await setUpProcessMessage(client, fromNumber); + + const { + rows: [message], + } = await client.query( + 'select id from sms.send_message($1, $2, $3, $4, $5, $6, $7)', + [ + profileId, + toNumber, + faker.hacker.phrase(), + null, + '11238', + null, + sendAfter, + ] + ); + + const foundProcessMessageJob = await findJob( + client, + PROCESS_10DLC_MESSAGE_IDENTIFIER, + 'id', + message.id + ); + + await process10DlcMessage(client, foundProcessMessageJob.payload); + + const { + rows: [result], + } = await client.query( + `select * from sms.outbound_messages_routing where id = $1`, + [message.id] + ); + + return result; + }); + + expect(m.send_after?.getTime()).toBe(sendAfter.getTime()); + }); }); diff --git a/src/jobs/process-10dlc-message.ts b/src/jobs/process-10dlc-message.ts index 28c8f01..a1bae43 100644 --- a/src/jobs/process-10dlc-message.ts +++ b/src/jobs/process-10dlc-message.ts @@ -62,7 +62,7 @@ export const process10DlcMessage: WrappableTask = async ( stage: outbound_message_stages.Queued, from_number: prevMappingRecord.from_number, sending_location_id: prevMappingRecord.sending_location_id, - send_after: null, + send_after: payload.send_after as unknown as Date | null, first_from_to_pair_of_day: firstFromToPairOfDay, }); } else { @@ -94,7 +94,7 @@ export const process10DlcMessage: WrappableTask = async ( stage: outbound_message_stages.Queued, from_number: fromNumber, sending_location_id: sendingLocationId, - send_after: null, + send_after: payload.send_after as unknown as Date | null, first_from_to_pair_of_day: true, }); } diff --git a/src/jobs/send-message.spec.ts b/src/jobs/send-message.spec.ts index dac1c13..b771e08 100644 --- a/src/jobs/send-message.spec.ts +++ b/src/jobs/send-message.spec.ts @@ -53,13 +53,14 @@ const sendAndProcessMessage = async ( body: string, mediaUrls: string[] | null, zip: string, - sendBefore: Date | null + sendBefore: Date | null, + sendAfter: Date | null = null ) => { const { rows: [toProcess], } = await client.query( - `select * from sms.send_message($1, $2, $3, $4, $5, $6)`, - [profileId, toNumber, body, mediaUrls, zip, sendBefore] + `select * from sms.send_message($1, $2, $3, $4, $5, $6, $7)`, + [profileId, toNumber, body, mediaUrls, zip, sendBefore, sendAfter] ); const foundJob = await findJob( @@ -106,7 +107,8 @@ const setupSendMessage = async ( service: Service, myNumber: string, sendBefore: Date | null = null, - mediaUrls: string[] | null = null + mediaUrls: string[] | null = null, + sendAfter: Date | null = null ): Promise => { const sendingAccount = await createSendingAccount(client, { triggers: true, @@ -146,7 +148,8 @@ const setupSendMessage = async ( faker.hacker.phrase(), mediaUrls, '11238', - sendBefore + sendBefore, + sendAfter ); return message.id; @@ -649,4 +652,41 @@ describe('send message', () => { expect(job.flags).toEqual({ 'send-message-mms:global': true }); }); + + test('should reject send_after that is not before send_before', async () => { + await withPgMiddlewares(pool, [autoRollbackMiddleware], async (client) => { + const myNumber = fakeNumber(); + const sendBefore = new Date(); + const sendAfter = new Date(sendBefore.getTime() + 1000 * 60 * 60); + + await expect( + setupSendMessage( + client, + Service.Telnyx, + myNumber, + sendBefore, + null, + sendAfter + ) + ).rejects.toThrow(/send_after must be before send_before/); + }); + }); + + test('should reject send_after for a non-10dlc (grey-route) profile', async () => { + await withPgMiddlewares(pool, [autoRollbackMiddleware], async (client) => { + const myNumber = fakeNumber(); + const sendAfter = new Date(Date.now() + 1000 * 60 * 60); + + await expect( + setupSendMessage( + client, + Service.Telnyx, + myNumber, + null, + null, + sendAfter + ) + ).rejects.toThrow(/send_after is only supported for the 10dlc channel/); + }); + }); }); diff --git a/src/lib/process-message.ts b/src/lib/process-message.ts index da95a92..833c85a 100644 --- a/src/lib/process-message.ts +++ b/src/lib/process-message.ts @@ -20,6 +20,7 @@ export const ProcessMessagePayloadSchema = z profile_id: z.string(), contact_zip_code: z.string().length(5), estimated_segments: z.number().int(), + send_after: z.string().nullable(), }) .required();