diff --git a/CHANGELOG.md b/CHANGELOG.md index 952f7b3..835d139 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ All notable changes to this project are documented here. The format follows ### Added +- `Smswire::Engine` with `smswire:install:migrations` and the + `smswire_deliveries` table, following the app's primary key type. +- `Smswire::Delivery` records for every send, with atomic send claims, + forward-only status, and retries that reuse their row. +- Deduplication of identical messages inside `dedupe_window`, returning + `:duplicate`, with optional explicit `idempotency_key`. +- Status callback endpoint with Twilio signature verification and + `sms_status_updated` observer hook. `callbacks_url` builds callback URLs. +- Per-category body storage, with `otp` bodies not stored by default. +- `Smswire::LogSubscriber` logging deliveries, rejections, and status + updates with masked numbers and no bodies. + - `Smswire::Base` messenger classes with `default`, `with(params)`, `.text.erb` templates, layouts, I18n lazy lookup, and URL helpers. - `Smswire::Message` with GSM-7 / UCS-2 encoding detection and segment counting. diff --git a/Gemfile b/Gemfile index 6bbf775..80fa44e 100644 --- a/Gemfile +++ b/Gemfile @@ -7,12 +7,14 @@ if (rails_version = ENV["RAILS_VERSION"]) requirement = "~> #{rails_version}.0" gem "actionpack", requirement gem "activejob", requirement + gem "activerecord", requirement gem "activesupport", requirement gem "railties", requirement # ActiveSupport 7.1 and 8.0 pass to_json options that json 3.0 removed. gem "json", "< 3" if %w[7.1 8.0].include?(rails_version) end +gem "sqlite3", (ENV["RAILS_VERSION"] == "7.1") ? "~> 1.4" : ">= 2.1" gem "rake", "~> 13.0" gem "minitest", "~> 5.25" gem "webmock", "~> 3.23" diff --git a/README.md b/README.md index f2b72f7..ca0fdcd 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ consent and STOP / HELP / START handling, quiet hours, previews, and a [Noticed](https://github.com/excid3/noticed) delivery adapter follow in later phases of the [specification](docs/SPEC.md). -**Status:** pre-release. Phase 1 of the spec is implemented on `main`; the +**Status:** pre-release. Phases 1 and 2 of the spec are implemented; the published `0.0.1.pre` gem only reserves the name. ## Requirements @@ -58,6 +58,38 @@ through interceptors, sent, and reported to observers and `ActiveSupport::Notifications` (`deliver.smswire`, `reject.smswire`). Bodies are never instrumented or logged by provider adapters. +### Deliveries and status callbacks + +``` +bin/rails smswire:install:migrations +bin/rails db:migrate +``` + +```ruby +# config/routes.rb +mount Smswire::Engine => "/smswire" + +# config/initializers/smswire.rb +config.callbacks_url = "https://app.example.com/smswire" +``` + +Every send is recorded in `Smswire::Delivery` with the provider message id, +segments, price, and status. With `callbacks_url` set, Twilio posts status +updates to the engine, which verifies the signature and moves the row +forward: `pending`, `queued`, `sent`, `delivered` (or `undelivered` / +`failed`). Out-of-order callbacks never move a status backwards. + +Identical messages to the same number inside `dedupe_window` (10 minutes by +default) are refused with status `:duplicate`. Pass `idempotency_key:` to +`text` to choose what counts as identical. Bodies are stored unless the +category opts out; `otp` does by default. + +```ruby +result = OrderMessenger.with(order:).shipped(user).deliver_now +result.delivery.status # => "accepted" +Smswire::Delivery.for_number("+15551234567").recent +``` + ### Providers | Name | Use | diff --git a/app/controllers/smswire/status_callbacks_controller.rb b/app/controllers/smswire/status_callbacks_controller.rb new file mode 100644 index 0000000..9098421 --- /dev/null +++ b/app/controllers/smswire/status_callbacks_controller.rb @@ -0,0 +1,43 @@ +module Smswire + # Receives provider delivery status callbacks at + # POST /status/:provider. Always answers 2xx for well-formed, + # signed requests so providers do not retry, even when the delivery is + # unknown. + class StatusCallbacksController < ActionController::Base + skip_forgery_protection + + def create + provider = Providers.resolve(params[:provider]) + return head(:not_found) unless provider.capabilities.include?(:status_callbacks) + + if Smswire.config.verify_callback_signatures + provider.verify_signature!(request, url: Callbacks.request_url(request)) + end + + update = provider.parse_status_callback(request) + payload = {provider: provider.name, provider_id: update.provider_id, status: update.status, + error_code: update.error_code} + ActiveSupport::Notifications.instrument("status_update.smswire", payload) do + delivery = Delivery.apply_status_update(update, provider: provider.name, delivery_id: params[:delivery_id]) + payload[:delivery_id] = delivery&.id + payload[:delivery_status] = delivery&.status + notify_observers(delivery, update) if delivery + end + + head :no_content + rescue ConfigurationError + head :not_found + rescue SignatureError + head :forbidden + end + + private + + def notify_observers(delivery, update) + Smswire.config.observers.each do |observer| + observer = observer.is_a?(String) ? observer.constantize : observer + observer.sms_status_updated(delivery, update) if observer.respond_to?(:sms_status_updated) + end + end + end +end diff --git a/app/models/smswire/application_record.rb b/app/models/smswire/application_record.rb new file mode 100644 index 0000000..be3e86b --- /dev/null +++ b/app/models/smswire/application_record.rb @@ -0,0 +1,5 @@ +module Smswire + class ApplicationRecord < ActiveRecord::Base + self.abstract_class = true + end +end diff --git a/app/models/smswire/delivery.rb b/app/models/smswire/delivery.rb new file mode 100644 index 0000000..ac4c303 --- /dev/null +++ b/app/models/smswire/delivery.rb @@ -0,0 +1,157 @@ +module Smswire + # One outbound SMS, from claim through provider status callbacks. + # + # Status lifecycle: pending, then queued / accepted / sent from the + # provider, then delivered, undelivered, or failed, and optionally read. + # suppressed means an interceptor cancelled it. Status only moves forward, + # so out-of-order callbacks are ignored. + class Delivery < ApplicationRecord + STATUSES = %w[pending suppressed queued accepted sent delivered undelivered failed read].freeze + PROGRESS = { + "pending" => 0, "suppressed" => 0, "queued" => 1, "accepted" => 2, "sent" => 3, + "delivered" => 4, "undelivered" => 4, "failed" => 4, "read" => 5 + }.freeze + RECLAIMABLE = %w[pending failed suppressed].freeze + TERMINAL_FAILURES = %w[undelivered failed].freeze + + # A claim older than this is treated as abandoned by a crashed worker. + CLAIM_TIMEOUT = 2.minutes + + enum :status, STATUSES.index_with(&:itself) + + belongs_to :recipient, polymorphic: true, optional: true + + scope :recent, -> { order(created_at: :desc) } + scope :for_number, ->(number) { where(to_number: number) } + + class << self + # Create or reuse the delivery row for +message+ and claim it for + # sending. Returns nil when an identical message was already sent + # inside the dedupe window, or another worker holds the claim. + def claim_for(message, provider:) + key = Smswire.config.dedupe_window ? message.idempotency_key : nil + + if key && (existing = find_by(idempotency_key: key)) + if existing.created_at < Smswire.config.dedupe_window.ago + where(id: existing.id, idempotency_key: key).update_all(idempotency_key: nil) + else + return existing.claim + end + end + + create!(attributes_for(message, provider:).merge( + idempotency_key: key, status: "pending", attempts: 1, claimed_at: Time.current + )) + rescue ActiveRecord::RecordNotUnique + nil + end + + # Apply a provider status callback. Looks up by delivery id when the + # callback URL carried one, falling back to the provider message id. + def apply_status_update(update, provider:, delivery_id: nil) + delivery = find_by(id: delivery_id, provider: provider.to_s) if delivery_id.present? + if delivery.nil? || (delivery.provider_id.present? && delivery.provider_id != update.provider_id) + delivery = find_by(provider: provider.to_s, provider_id: update.provider_id) + end + delivery&.advance!(update.status, provider_id: update.provider_id, + error_code: update.error_code, error_message: update.error_message) + delivery + end + + def attributes_for(message, provider:) + record = message.recipient&.record + { + messenger: message.messenger, + action: message.action, + to_number: message.to, + from_number: message.from, + messaging_service: message.messaging_service, + recipient: (record if record.is_a?(ActiveRecord::Base) && record.persisted?), + category: message.category.to_s, + body: (message.body if Smswire.config.store_body?(message.category)), + segments: message.segments, + encoding: message.encoding.to_s, + provider: provider.to_s, + metadata: message.metadata.presence + } + end + end + + # Atomically claim a reusable row for another send attempt. + def claim + return nil unless RECLAIMABLE.include?(status) + + now = Time.current + claimed = self.class + .where(id:, status: RECLAIMABLE) + .where("claimed_at IS NULL OR claimed_at < ?", now - CLAIM_TIMEOUT) + .update_all(["claimed_at = ?, attempts = attempts + 1, status = 'pending', " \ + "error_code = NULL, error_message = NULL, failed_at = NULL, updated_at = ?", now, now]) + reload if claimed == 1 + end + + def record_receipt!(receipt, message) + now = Time.current + with_lock do + assign_attributes(final_message_attributes(message)) + assign_attributes(provider_id: receipt.provider_id, segments: receipt.segments || message.segments, + price_amount: receipt.price_amount, price_currency: receipt.price_currency, + sent_at: now, claimed_at: nil) + apply_status(receipt.status.to_s, now) if progresses_to?(receipt.status) + save! + end + end + + def record_failure!(error, message) + update!(final_message_attributes(message).merge( + status: "failed", error_code: error.provider_code, error_message: error.message, + failed_at: Time.current, status_updated_at: Time.current, claimed_at: nil + )) + end + + # Release the claim after a transient error so a retry can reclaim it. + def release!(error) + update!(claimed_at: nil, error_code: error.provider_code, error_message: error.message) + end + + def record_suppression!(message) + update!(final_message_attributes(message).merge(status: "suppressed", claimed_at: nil, status_updated_at: Time.current)) + end + + def advance!(new_status, provider_id: nil, error_code: nil, error_message: nil) + with_lock do + self.provider_id ||= provider_id + if progresses_to?(new_status) + apply_status(new_status.to_s, Time.current) + self.error_code = error_code if error_code.present? + self.error_message = error_message if error_message.present? + end + save! + end + end + + def progresses_to?(new_status) + PROGRESS.fetch(new_status.to_s) > PROGRESS.fetch(status) + end + + private + + def apply_status(new_status, at) + self.status = new_status + self.status_updated_at = at + self.delivered_at = at if new_status == "delivered" + self.failed_at = at if TERMINAL_FAILURES.include?(new_status) + end + + def final_message_attributes(message) + { + to_number: message.to, + from_number: message.from, + messaging_service: message.messaging_service, + body: (message.body if Smswire.config.store_body?(message.category)), + segments: message.segments, + encoding: message.encoding.to_s + } + end + end +end diff --git a/config/routes.rb b/config/routes.rb new file mode 100644 index 0000000..98ed050 --- /dev/null +++ b/config/routes.rb @@ -0,0 +1,3 @@ +Smswire::Engine.routes.draw do + post "status/:provider", to: "status_callbacks#create", as: :status_callback +end diff --git a/db/migrate/20261008000000_create_smswire_deliveries.rb b/db/migrate/20261008000000_create_smswire_deliveries.rb new file mode 100644 index 0000000..2189b21 --- /dev/null +++ b/db/migrate/20261008000000_create_smswire_deliveries.rb @@ -0,0 +1,47 @@ +class CreateSmswireDeliveries < ActiveRecord::Migration[7.1] + def change + primary_key_type, foreign_key_type = primary_and_foreign_key_types + + create_table :smswire_deliveries, id: primary_key_type do |t| + t.string :messenger + t.string :action + t.string :to_number, null: false + t.string :from_number + t.string :messaging_service + t.references :recipient, polymorphic: true, type: foreign_key_type, index: true + t.string :category, null: false + t.text :body + t.integer :segments + t.string :encoding + t.string :provider, null: false + t.string :provider_id + t.string :status, null: false, default: "pending" + t.string :error_code + t.text :error_message + t.string :idempotency_key + t.decimal :price_amount, precision: 12, scale: 5 + t.string :price_currency + t.json :metadata + t.integer :attempts, null: false, default: 0 + t.datetime :claimed_at + t.datetime :sent_at + t.datetime :delivered_at + t.datetime :failed_at + t.datetime :status_updated_at + t.timestamps + end + + add_index :smswire_deliveries, [:provider, :provider_id], unique: true + add_index :smswire_deliveries, :idempotency_key, unique: true + add_index :smswire_deliveries, [:to_number, :created_at] + add_index :smswire_deliveries, [:status, :created_at] + end + + private + + def primary_and_foreign_key_types + config = Rails.configuration.generators + setting = config.options[config.orm][:primary_key_type] + [setting || :primary_key, setting || :bigint] + end +end diff --git a/docs/SPEC.md b/docs/SPEC.md index 4206973..2856ed3 100644 --- a/docs/SPEC.md +++ b/docs/SPEC.md @@ -290,22 +290,25 @@ Three tables, installed by `rails g smswire:install`. | column | type | notes | |---|---|---| -| id | uuid/bigint | | +| id | app default | follows the app's generator `primary_key_type`, like Active Storage | | messenger, action | string | e.g. `OrderMessenger`, `shipped` | -| to, from | string | E.164 | -| recipient_type, recipient_id | string, string | polymorphic, optional | +| to_number, from_number | string | E.164; named to avoid the reserved words `to` and `from` | +| messaging_service | string | | +| recipient_type, recipient_id | string, app key type | polymorphic, optional; set only for persisted Active Record recipients | | category | string | `transactional`, `otp`, `marketing`, custom | | body | text | nullable; `store_bodies` config defaults to true, with a per-category override so OTP bodies can be excluded | | segments, encoding | integer, string | | | provider | string | | | provider_id | string | indexed, unique with provider | -| status | string | normalized vocabulary | +| status | string | `pending`, `suppressed`, then the normalized vocabulary; only moves forward | | error_code, error_message | string, text | provider raw code + normalized message | -| idempotency_key | string | unique index | +| idempotency_key | string | unique index; cleared on the old row when the dedupe window expires | | price_amount, price_currency | decimal, string | if the provider reports it | | metadata | json | app-supplied | | scheduled_at, sent_at, delivered_at, failed_at | datetime | | | attempts | integer | | +| claimed_at | datetime | send claim; released on transient errors, expires after two minutes | +| status_updated_at | datetime | | | created_at, updated_at | datetime | | **smswire_consents** @@ -455,7 +458,9 @@ Smswire.configure do |c| c.body_whitespace = :strip # or :squish, :preserve c.deliver_later_queue = :default # Smswire.register_interceptor / register_observer, per environment - c.callbacks_host = "https://app.example.com" # used to build StatusCallback URLs + c.callbacks_url = "https://app.example.com/smswire" # engine mount URL; builds StatusCallback URLs + c.persist_deliveries = true + c.verify_callback_signatures = true end ``` @@ -521,6 +526,11 @@ next. No phase includes a calendar estimate. - Accept: a delivery row moves `pending → accepted → delivered` from a replayed Twilio callback fixture; duplicate sends inside the window are refused. +- Status: implemented. Rows are claimed atomically before sending, so a + transient-error retry reuses its row and two workers never send the same + claimed message. Callback URLs carry the delivery id to survive the race + between the provider's response and its first callback. Rejected messages + are not persisted; they are reported through `reject.smswire`. ### Phase 3: Consent and inbound keywords @@ -564,13 +574,14 @@ next. No phase includes a calendar estimate. ## 11. Open decisions -These do not block Phase 0 or 1 but should be settled before Phase 2: +Decisions 1 to 3 are settled. The rest should be settled before 1.0. 1. **Name.** `smswire` is recommended; `textable` and `smsable` are the fallbacks. Decide before the pre-release push. -2. **Body storage default.** Spec says `true` with OTP excluded. Privacy-first - alternative is `false` by default. -3. **UUID vs. bigint** primary keys for the engine tables. Spec leans UUID. +2. **Body storage default.** Settled in Phase 2: `store_bodies = true`, with + `categories: { otp: { store_body: false } }` as the shipped default. +3. **UUID vs. bigint** primary keys. Settled in Phase 2: the migration follows + the app's generator `primary_key_type`, so both work without a choice here. 4. **Phone validation**: `phonelib` soft dependency vs. a vendored minimal E.164 validator. 5. **Whether `deliver_now` should be allowed in production** or warn, given diff --git a/lib/smswire.rb b/lib/smswire.rb index e48e604..aa63f3a 100644 --- a/lib/smswire.rb +++ b/lib/smswire.rb @@ -4,13 +4,14 @@ require "active_support/core_ext/string/filters" require "active_support/core_ext/hash/keys" require "active_support/core_ext/class/attribute" +require "active_support/core_ext/numeric/time" require "active_job" loader = Zeitwerk::Loader.for_gem loader.inflector.inflect("http" => "HTTP", "rspec" => "RSpec") loader.ignore( "#{__dir__}/smswire/errors.rb", - "#{__dir__}/smswire/railtie.rb", + "#{__dir__}/smswire/engine.rb", "#{__dir__}/smswire/rspec.rb" ) loader.setup @@ -63,4 +64,4 @@ def unregister_observer(observer) end end -require_relative "smswire/railtie" if defined?(Rails::Railtie) +require_relative "smswire/engine" if defined?(Rails::Railtie) diff --git a/lib/smswire/callbacks.rb b/lib/smswire/callbacks.rb new file mode 100644 index 0000000..a40d325 --- /dev/null +++ b/lib/smswire/callbacks.rb @@ -0,0 +1,24 @@ +module Smswire + # Builds callback URLs from Smswire.config.callbacks_url and reconstructs + # the public URL of an incoming callback for signature checks. + module Callbacks + module_function + + def status_url(provider, delivery = nil) + base = Smswire.config.callbacks_url.presence or return nil + url = "#{base.chomp("/")}/status/#{provider}" + delivery ? "#{url}?delivery_id=#{delivery.id}" : url + end + + # The URL the provider signed. Behind a TLS-terminating proxy the + # request may look like plain HTTP, so when callbacks_url is configured + # its scheme, host, and port are used with the request path. + def request_url(request) + base = Smswire.config.callbacks_url.presence or return request.original_url + uri = URI(base) + origin = "#{uri.scheme}://#{uri.host}" + origin += ":#{uri.port}" unless uri.port == uri.default_port + origin + request.original_fullpath + end + end +end diff --git a/lib/smswire/configuration.rb b/lib/smswire/configuration.rb index f07639e..0c5d25a 100644 --- a/lib/smswire/configuration.rb +++ b/lib/smswire/configuration.rb @@ -40,6 +40,28 @@ class Configuration attr_accessor :interceptors, :observers, :logger + # Record every send in smswire_deliveries. Requires the engine migrations. + attr_accessor :persist_deliveries + + # Identical messages (same messenger, action, recipient, and body, or the + # same explicit idempotency_key) inside this window are refused as + # :duplicate. nil disables deduplication. + attr_accessor :dedupe_window + + # Public base URL where Smswire::Engine is mounted, e.g. + # "https://app.example.com/smswire". When set, providers that support + # status callbacks are told to post to "/status/". + attr_accessor :callbacks_url + + attr_accessor :verify_callback_signatures + + # Store message bodies on deliveries. Overridden per category with + # categories: { otp: { store_body: false } }. + attr_accessor :store_bodies + + # Per-category settings. Phase 2 honors :store_body. + attr_accessor :categories + def initialize @default_provider = nil @providers = {} @@ -54,6 +76,12 @@ def initialize @interceptors = [] @observers = [] @logger = nil + @persist_deliveries = true + @dedupe_window = 10.minutes + @callbacks_url = nil + @verify_callback_signatures = true + @store_bodies = true + @categories = {otp: {store_body: false}} end def phone_validator=(value) @@ -72,6 +100,15 @@ def body_whitespace=(value) @body_whitespace = value end + def category_options(category) + (categories[category.to_s.to_sym] || categories[category.to_s] || {}).to_h.symbolize_keys + end + + def store_body?(category) + setting = category_options(category)[:store_body] + setting.nil? ? store_bodies : setting + end + def provider_options(name) (providers[name.to_sym] || providers[name.to_s] || {}).to_h.symbolize_keys end diff --git a/lib/smswire/railtie.rb b/lib/smswire/engine.rb similarity index 71% rename from lib/smswire/railtie.rb rename to lib/smswire/engine.rb index 60534bf..5a45614 100644 --- a/lib/smswire/railtie.rb +++ b/lib/smswire/engine.rb @@ -1,8 +1,15 @@ -require "rails/railtie" +require "rails/engine" require "abstract_controller/railties/routes_helpers" module Smswire - class Railtie < ::Rails::Railtie + # Mount for status callbacks: + # + # mount Smswire::Engine => "/smswire" + # + # Install the migrations with +bin/rails smswire:install:migrations+. + class Engine < ::Rails::Engine + isolate_namespace Smswire + ENVIRONMENT_PROVIDERS = {development: :log, test: :test}.freeze config.smswire = ActiveSupport::OrderedOptions.new @@ -14,6 +21,10 @@ class Railtie < ::Rails::Railtie Smswire.config.logger ||= ::Rails.logger end + initializer "smswire.log_subscriber" do + Smswire::LogSubscriber.attach_to :smswire + end + initializer "smswire.messenger" do |app| ActiveSupport.on_load(:smswire) do self.view_paths = ["#{::Rails.root}/app/views"] diff --git a/lib/smswire/errors.rb b/lib/smswire/errors.rb index 9f640aa..43fe347 100644 --- a/lib/smswire/errors.rb +++ b/lib/smswire/errors.rb @@ -9,6 +9,9 @@ class ConfigurationError < Error; end # number. Raised while building the message so it surfaces in tests. class UnresolvableRecipient < Error; end + # Raised when a provider callback fails signature verification. + class SignatureError < Error; end + # Base class for errors raised by provider adapters. class DeliveryError < Error attr_reader :provider, :provider_code, :http_status, :raw diff --git a/lib/smswire/log_subscriber.rb b/lib/smswire/log_subscriber.rb new file mode 100644 index 0000000..a7ebda0 --- /dev/null +++ b/lib/smswire/log_subscriber.rb @@ -0,0 +1,48 @@ +module Smswire + # Logs deliveries, rejections, and status updates. Phone numbers are + # masked and bodies are never logged. + class LogSubscriber < ActiveSupport::LogSubscriber + def deliver(event) + payload = event.payload + route = "#{payload[:messenger]}##{payload[:action]} to #{mask(payload[:to])} via #{payload[:provider]}" + if (error = payload[:exception_object]) + warn { "SMS #{route} raised #{error.class}: #{error.message} (#{event.duration.round(1)}ms)" } + else + info do + "SMS #{route}: #{payload[:status]} #{payload[:provider_id]} " \ + "(#{payload[:segments]} segment(s), #{event.duration.round(1)}ms)" + end + end + end + + def reject(event) + payload = event.payload + info { "SMS #{payload[:messenger]}##{payload[:action]} to #{mask(payload[:to])} not sent: #{payload[:reason]}" } + end + + def status_update(event) + payload = event.payload + if payload[:delivery_id] + info do + "SMS status #{payload[:provider]} #{payload[:provider_id]}: #{payload[:status]} " \ + "(delivery #{payload[:delivery_id]} now #{payload[:delivery_status]})" + end + else + warn { "SMS status #{payload[:provider]} #{payload[:provider_id]}: #{payload[:status]} for unknown delivery" } + end + end + + def logger + Smswire.logger + end + + private + + def mask(number) + number = number.to_s + return number if number.length < 8 + + "#{number[0, 2]}#{"*" * (number.length - 6)}#{number[-4..]}" + end + end +end diff --git a/lib/smswire/message.rb b/lib/smswire/message.rb index 409eaad..13fdd46 100644 --- a/lib/smswire/message.rb +++ b/lib/smswire/message.rb @@ -7,6 +7,9 @@ class Message attr_accessor :to, :from, :body, :media_urls, :messaging_service, :provider, :category, :metadata, :validity_period, :messenger, :action, :recipient attr_reader :raw_to + + # Set by the pipeline when Smswire.config.callbacks_url is configured. + attr_accessor :status_callback_url attr_writer :idempotency_key def initialize(to:, body:, from: nil, media_urls: [], messaging_service: nil, provider: nil, @@ -49,6 +52,12 @@ def idempotency_key @idempotency_key || Digest::SHA256.hexdigest([messenger, action, to, body].join("\u0000"))[0, 32] end + # Fix the key to the current content, so interceptors that rewrite the + # body do not change which message this is. + def lock_idempotency_key! + @idempotency_key = idempotency_key + end + # Attributes safe to log or instrument. Never includes the body. def to_log_h {messenger:, action:, to:, from:, messaging_service:, provider:, category:, diff --git a/lib/smswire/pipeline.rb b/lib/smswire/pipeline.rb index 2de3b67..4a20215 100644 --- a/lib/smswire/pipeline.rb +++ b/lib/smswire/pipeline.rb @@ -1,9 +1,12 @@ module Smswire - # Runs a message through normalize, validate, intercept, send, and observe. + # Runs a message through normalize, validate, claim, intercept, send, + # record, and observe. # # Returns a Smswire::Result. Permanent provider errors become a :failed # result. Transient errors and configuration errors propagate so the - # caller (or Smswire::DeliveryJob) can retry or fail loudly. + # caller (or Smswire::DeliveryJob) can retry or fail loudly. When + # persistence is enabled, each claimed message has a Smswire::Delivery row + # that a retry reuses, so a message is never recorded twice. class Pipeline def self.call(message) new(message).call @@ -22,27 +25,44 @@ def call return reject(:rejected_empty_body) if message.body.blank? && message.media_urls.empty? + provider = Providers.resolve(message.provider) + message.lock_idempotency_key! + + delivery = nil + if Smswire.config.persist_deliveries + delivery = Delivery.claim_for(message, provider: provider.name) + return reject(:duplicate) unless delivery + message.status_callback_url ||= Callbacks.status_url(provider.name, delivery) if provider.capabilities.include?(:status_callbacks) + end + run_interceptors - return finish(Result.new(status: :suppressed_by_interceptor, message:)) unless message.perform_deliveries? + unless message.perform_deliveries? + delivery&.record_suppression!(message) + return finish(Result.new(status: :suppressed_by_interceptor, message:, delivery:)) + end - deliver + deliver(provider, delivery) end private - def deliver - provider = Providers.resolve(message.provider) - payload = message.to_log_h.merge(provider: provider.name) + def deliver(provider, delivery) + payload = message.to_log_h.merge(provider: provider.name, delivery_id: delivery&.id) receipt = ActiveSupport::Notifications.instrument("deliver.smswire", payload) do provider.deliver(message).tap do |r| payload[:provider_id] = r.provider_id payload[:status] = r.status end end + delivery&.record_receipt!(receipt, message) status = (receipt.status == :failed) ? :failed : :accepted - finish(Result.new(status:, message:, receipt:)) + finish(Result.new(status:, message:, receipt:, delivery:)) rescue PermanentError => error - finish(Result.new(status: :failed, message:, error:)) + delivery&.record_failure!(error, message) + finish(Result.new(status: :failed, message:, error:, delivery:)) + rescue TransientError => error + delivery&.release!(error) + raise end def reject(status) diff --git a/lib/smswire/providers/base.rb b/lib/smswire/providers/base.rb index 3134438..0b600fa 100644 --- a/lib/smswire/providers/base.rb +++ b/lib/smswire/providers/base.rb @@ -19,6 +19,17 @@ def capabilities Set.new end + # Raise Smswire::SignatureError unless +request+ was signed by the + # provider. +url+ is the public URL the provider posted to. + def verify_signature!(request, url:) + raise NotImplementedError, "#{self.class.name} does not support callbacks" + end + + # Return a Smswire::StatusUpdate for a status callback request. + def parse_status_callback(request) + raise NotImplementedError, "#{self.class.name} does not support status callbacks" + end + private def receipt(**attributes) diff --git a/lib/smswire/providers/twilio.rb b/lib/smswire/providers/twilio.rb index 6f5c505..d64c08b 100644 --- a/lib/smswire/providers/twilio.rb +++ b/lib/smswire/providers/twilio.rb @@ -1,4 +1,5 @@ require "bigdecimal" +require "base64" module Smswire module Providers @@ -53,10 +54,33 @@ def form_for(message) pairs << ["Body", message.body] if message.body.present? message.media_urls.each { |url| pairs << ["MediaUrl", url] } pairs << ["ValidityPeriod", message.validity_period.to_i.to_s] if message.validity_period - pairs << ["StatusCallback", options[:status_callback]] if options[:status_callback] + callback = message.status_callback_url || options[:status_callback] + pairs << ["StatusCallback", callback] if callback pairs end + # Twilio signs callbacks with HMAC-SHA1 over the full URL followed by + # each POST parameter's name and value, sorted by name. + def verify_signature!(request, url:) + signature = request.headers["X-Twilio-Signature"].to_s + data = url + request.request_parameters.sort.map { |key, value| "#{key}#{value}" }.join + expected = Base64.strict_encode64(OpenSSL::HMAC.digest("SHA1", auth_token, data)) + return if signature.present? && ActiveSupport::SecurityUtils.secure_compare(expected, signature) + + raise SignatureError, "Invalid X-Twilio-Signature" + end + + def parse_status_callback(request) + params = request.request_parameters + StatusUpdate.new( + provider_id: params["MessageSid"] || params["SmsSid"], + status: STATUSES.fetch((params["MessageStatus"] || params["SmsStatus"]).to_s, :accepted), + error_code: params["ErrorCode"].presence, + error_message: params["ErrorMessage"].presence, + raw: params.to_h + ) + end + private def handle(response) diff --git a/lib/smswire/result.rb b/lib/smswire/result.rb index 245b5a2..be104a0 100644 --- a/lib/smswire/result.rb +++ b/lib/smswire/result.rb @@ -8,8 +8,11 @@ module Smswire # :rejected_empty_body no body and no media # :suppressed_by_interceptor an interceptor cancelled delivery # :skipped the messenger action did not call +text+ - Result = Data.define(:status, :message, :receipt, :error) do - def initialize(status:, message: nil, receipt: nil, error: nil) + # :duplicate an identical message was sent inside the dedupe window + # + # +delivery+ is the Smswire::Delivery row when persistence is enabled. + Result = Data.define(:status, :message, :receipt, :error, :delivery) do + def initialize(status:, message: nil, receipt: nil, error: nil, delivery: nil) super end @@ -23,6 +26,8 @@ def suppressed? = status == :suppressed_by_interceptor def skipped? = status == :skipped + def duplicate? = status == :duplicate + def provider_id = receipt&.provider_id end end diff --git a/lib/smswire/status_update.rb b/lib/smswire/status_update.rb new file mode 100644 index 0000000..0db6cee --- /dev/null +++ b/lib/smswire/status_update.rb @@ -0,0 +1,8 @@ +module Smswire + # A provider's report that a message changed status, parsed from a callback. + StatusUpdate = Data.define(:provider_id, :status, :error_code, :error_message, :raw) do + def initialize(provider_id:, status:, error_code: nil, error_message: nil, raw: {}) + super + end + end +end diff --git a/smswire.gemspec b/smswire.gemspec index 8da1fdf..a65d275 100644 --- a/smswire.gemspec +++ b/smswire.gemspec @@ -35,6 +35,7 @@ Gem::Specification.new do |spec| rails_requirement = [">= 7.1", "< 9"] spec.add_dependency "actionpack", *rails_requirement spec.add_dependency "activejob", *rails_requirement + spec.add_dependency "activerecord", *rails_requirement spec.add_dependency "activesupport", *rails_requirement spec.add_dependency "railties", *rails_requirement spec.add_dependency "zeitwerk", "~> 2.6" diff --git a/test/delivery_test.rb b/test/delivery_test.rb new file mode 100644 index 0000000..d9c5efa --- /dev/null +++ b/test/delivery_test.rb @@ -0,0 +1,167 @@ +require "test_helper" + +class DeliveryTest < Smswire::TestCase + def send_literal(body = "hi", to: "+14155552671") + OrderMessenger.literal(to, body).deliver_now + end + + test "an accepted send is recorded with its receipt" do + result = send_literal("Your order shipped") + delivery = result.delivery + + assert delivery.persisted? + assert_equal "accepted", delivery.status + assert_equal result.provider_id, delivery.provider_id + assert_equal ["OrderMessenger", "literal"], [delivery.messenger, delivery.action] + assert_equal ["+14155552671", "+15005550006"], [delivery.to_number, delivery.from_number] + assert_equal "Your order shipped", delivery.body + assert_equal [1, "gsm7", "transactional", "test"], [delivery.segments, delivery.encoding, delivery.category, delivery.provider] + assert_equal 1, delivery.attempts + assert_nil delivery.claimed_at + assert delivery.sent_at + end + + test "identical messages inside the dedupe window are refused" do + assert send_literal.accepted? + duplicate = send_literal + assert duplicate.duplicate? + assert_nil duplicate.delivery + assert_equal 1, Smswire::Delivery.count + assert_equal 1, sms_deliveries.size + + assert send_literal("different").accepted? + assert send_literal(to: "+12125550123").accepted? + end + + test "an explicit idempotency key dedupes across different bodies" do + messenger = Class.new(ApplicationMessenger) do + def self.name = "OtpMessenger" + + def code(phone, code) = text(to: phone, body: "Code #{code}", idempotency_key: "otp:#{phone}") + end + assert messenger.code("+14155552671", "1111").deliver_now.accepted? + assert messenger.code("+14155552671", "2222").deliver_now.duplicate? + end + + test "the same message is allowed again after the window" do + travel_to Time.utc(2026, 1, 1, 12) do + send_literal + end + travel_to Time.utc(2026, 1, 1, 12, 11) do + assert send_literal.accepted? + end + assert_equal 2, Smswire::Delivery.count + assert_equal 1, Smswire::Delivery.where.not(idempotency_key: nil).count + end + + test "dedupe can be disabled" do + with_config(dedupe_window: nil) do + 2.times { assert send_literal.accepted? } + end + assert_equal 2, Smswire::Delivery.count + end + + test "a transient failure releases the claim and the retry reuses the row" do + Smswire::Providers::Test.fail_next(Smswire::TransientError.new("timeout", provider_code: "T1")) + assert_raises(Smswire::TransientError) { send_literal } + + delivery = Smswire::Delivery.sole + assert_equal "pending", delivery.status + assert_nil delivery.claimed_at + assert_equal "timeout", delivery.error_message + + result = send_literal + assert result.accepted? + assert_equal delivery.id, result.delivery.id + assert_equal 2, result.delivery.attempts + assert_nil result.delivery.error_message + assert_equal 1, Smswire::Delivery.count + end + + test "a row claimed by another worker is not sent twice until the claim expires" do + travel_to Time.utc(2026, 1, 1, 12) do + Smswire::Providers::Test.fail_next(Smswire::TransientError.new("timeout")) + assert_raises(Smswire::TransientError) { send_literal } + Smswire::Delivery.sole.update!(claimed_at: Time.current) # another worker picked it up + assert send_literal.duplicate? + end + travel_to Time.utc(2026, 1, 1, 12, 3) do + assert send_literal.accepted? + end + assert_equal 1, sms_deliveries.size + end + + test "a permanent failure is recorded and may be retried" do + Smswire::Providers::Test.fail_next(Smswire::PermanentError.new("blocked", reason: :opted_out, provider_code: "21610")) + failed = send_literal + assert failed.failed? + assert_equal ["failed", "21610", "blocked"], [failed.delivery.status, failed.delivery.error_code, failed.delivery.error_message] + assert failed.delivery.failed_at + + retried = send_literal + assert retried.accepted? + assert_equal failed.delivery.id, retried.delivery.id + end + + test "suppressed messages are recorded" do + blocker = Object.new + def blocker.delivering_sms(message) = message.cancel! + Smswire.register_interceptor(blocker) + result = send_literal + assert_equal "suppressed", result.delivery.status + ensure + Smswire.unregister_interceptor(blocker) + end + + test "the stored message reflects interceptor changes" do + rewriter = Object.new + def rewriter.delivering_sms(message) = message.body = "[staging] #{message.body}" + Smswire.register_interceptor(rewriter) + assert_equal "[staging] hi", send_literal.delivery.body + ensure + Smswire.unregister_interceptor(rewriter) + end + + test "bodies are not stored for the otp category or when disabled" do + messenger = Class.new(ApplicationMessenger) do + def self.name = "CodeMessenger" + + def code(phone) = text(to: phone, body: "Code 4321", category: :otp) + end + assert_nil messenger.code("+14155552671").deliver_now.delivery.body + + with_config(store_bodies: false) do + assert_nil send_literal("not kept").delivery.body + end + end + + test "rejections are not persisted" do + send_literal(to: "nope") + assert_equal 0, Smswire::Delivery.count + end + + test "persistence can be disabled" do + with_config(persist_deliveries: false) do + result = send_literal + assert result.accepted? + assert_nil result.delivery + end + assert_equal 0, Smswire::Delivery.count + end + + test "status only moves forward" do + delivery = send_literal.delivery + delivery.advance!(:delivered) + delivery.advance!(:sent) + assert_equal "delivered", delivery.reload.status + assert delivery.delivered_at + + delivery.advance!(:read) + assert_equal "read", delivery.reload.status + end + + test "engine provides the install migrations task" do + Rails.application.load_tasks + assert Rake::Task.task_defined?("smswire:install:migrations") + end +end diff --git a/test/dummy/config/application.rb b/test/dummy/config/application.rb index ab0bf52..84f1904 100644 --- a/test/dummy/config/application.rb +++ b/test/dummy/config/application.rb @@ -1,5 +1,6 @@ require "rails" require "active_job/railtie" +require "active_record/railtie" require "action_controller/railtie" require "smswire" diff --git a/test/dummy/config/database.yml b/test/dummy/config/database.yml new file mode 100644 index 0000000..e5837c7 --- /dev/null +++ b/test/dummy/config/database.yml @@ -0,0 +1,4 @@ +test: + adapter: sqlite3 + database: ":memory:" + pool: 5 diff --git a/test/dummy/config/routes.rb b/test/dummy/config/routes.rb index ae55598..e0ccaa0 100644 --- a/test/dummy/config/routes.rb +++ b/test/dummy/config/routes.rb @@ -1,4 +1,5 @@ Rails.application.routes.draw do default_url_options host: "shop.example" resources :orders, only: :show + mount Smswire::Engine => "/smswire" end diff --git a/test/log_subscriber_test.rb b/test/log_subscriber_test.rb new file mode 100644 index 0000000..6d4efe6 --- /dev/null +++ b/test/log_subscriber_test.rb @@ -0,0 +1,26 @@ +require "test_helper" + +class LogSubscriberTest < Smswire::TestCase + def capture + output = StringIO.new + with_config(logger: Logger.new(output)) { yield } + output.string + end + + test "logs deliveries with masked numbers and no body" do + log = capture { OrderMessenger.literal("+14155552671", "secret 5555").deliver_now } + assert_match(/SMS OrderMessenger#literal to \+1\*{6}2671 via test: accepted SMtest\w+ \(1 segment\(s\)/, log) + refute_includes log, "secret" + refute_includes log, "4155552671" + end + + test "logs rejections and transient errors" do + log = capture do + OrderMessenger.literal("bogus", "x").deliver_now + Smswire::Providers::Test.fail_next(Smswire::TransientError.new("network down")) + assert_raises(Smswire::TransientError) { OrderMessenger.literal("+14155552671", "y").deliver_now } + end + assert_includes log, "not sent: rejected_invalid_number" + assert_includes log, "raised Smswire::TransientError: network down" + end +end diff --git a/test/rspec_matchers_test.rb b/test/rspec_matchers_test.rb index 0b23a9e..4a46b6a 100644 --- a/test/rspec_matchers_test.rb +++ b/test/rspec_matchers_test.rb @@ -20,7 +20,7 @@ class RSpecMatchersTest < Smswire::TestCase end test "deliver_sms with exact counts" do - two = -> { 2.times { OrderMessenger.literal("+14155552671", "x").deliver_now } } + two = -> { 2.times { |n| OrderMessenger.literal("+14155552671", "x#{n}").deliver_now } } expect(&two).to deliver_sms.exactly(2).times assert_raises(NotMet) { expect(&two).to deliver_sms.once } end diff --git a/test/status_callbacks_test.rb b/test/status_callbacks_test.rb new file mode 100644 index 0000000..a2efa1e --- /dev/null +++ b/test/status_callbacks_test.rb @@ -0,0 +1,115 @@ +require "test_helper" +require "base64" + +class StatusCallbacksTest < ActionDispatch::IntegrationTest + include Smswire::TestHelper + include ConfigHelpers + + setup { Smswire::Delivery.delete_all } + + def sign(url, params, token: "secret") + data = url + params.sort.map { |key, value| "#{key}#{value}" }.join + Base64.strict_encode64(OpenSSL::HMAC.digest("SHA1", token, data)) + end + + def callback(path, params, signature: :valid, base: "http://www.example.com") + signature = sign(base + path, params) if signature == :valid + post path, params:, headers: {"X-Twilio-Signature" => signature.to_s} + end + + def send_via_twilio(sid: "SMabc") + stub_request(:post, twilio_url).to_return(twilio_success(sid:, status: "queued")) + with_config(default_provider: :twilio) { OrderMessenger.literal("+14155552671", "Your order shipped").deliver_now } + end + + test "acceptance: a delivery moves pending to queued to sent to delivered from replayed callbacks" do + result = nil + with_config(callbacks_url: "https://app.example/smswire") do + stub = stub_request(:post, twilio_url).with { |request| + URI.decode_www_form(request.body).include?(["StatusCallback", "https://app.example/smswire/status/twilio?delivery_id=#{Smswire::Delivery.last&.id}"]) + }.to_return(twilio_success(sid: "SMabc", status: "queued")) + result = with_config(default_provider: :twilio) { OrderMessenger.literal("+14155552671", "Your order shipped").deliver_now } + assert_requested stub + end + delivery = result.delivery + assert_equal "queued", delivery.reload.status + + path = "/smswire/status/twilio?delivery_id=#{delivery.id}" + with_config(callbacks_url: "https://app.example/smswire") do + callback(path, {"MessageSid" => "SMabc", "MessageStatus" => "sent"}, base: "https://app.example") + assert_response :no_content + assert_equal "sent", delivery.reload.status + + callback(path, {"MessageSid" => "SMabc", "MessageStatus" => "delivered"}, base: "https://app.example") + assert_equal "delivered", delivery.reload.status + assert delivery.delivered_at + + callback(path, {"MessageSid" => "SMabc", "MessageStatus" => "sent"}, base: "https://app.example") + assert_equal "delivered", delivery.reload.status, "late callbacks never move status backwards" + end + end + + test "callbacks without a delivery id are matched by provider message id" do + delivery = send_via_twilio(sid: "SMxyz").delivery + callback("/smswire/status/twilio", {"MessageSid" => "SMxyz", "MessageStatus" => "undelivered", + "ErrorCode" => "30003", "ErrorMessage" => "Unreachable"}) + assert_response :no_content + delivery.reload + assert_equal ["undelivered", "30003", "Unreachable"], [delivery.status, delivery.error_code, delivery.error_message] + assert delivery.failed_at + end + + test "a mismatched delivery id falls back to the provider message id" do + first = send_via_twilio(sid: "SM1").delivery + second = with_config(default_provider: :twilio) do + stub_request(:post, twilio_url).to_return(twilio_success(sid: "SM2")) + OrderMessenger.literal("+14155552671", "another").deliver_now.delivery + end + callback("/smswire/status/twilio?delivery_id=#{first.id}", {"MessageSid" => "SM2", "MessageStatus" => "delivered"}) + assert_equal "queued", first.reload.status + assert_equal "delivered", second.reload.status + end + + test "invalid signatures are refused" do + delivery = send_via_twilio.delivery + callback("/smswire/status/twilio", {"MessageSid" => "SMabc", "MessageStatus" => "delivered"}, signature: "forged") + assert_response :forbidden + callback("/smswire/status/twilio", {"MessageSid" => "SMabc", "MessageStatus" => "delivered"}, signature: nil) + assert_response :forbidden + assert_equal "queued", delivery.reload.status + end + + test "signature checks can be disabled" do + delivery = send_via_twilio.delivery + with_config(verify_callback_signatures: false) do + callback("/smswire/status/twilio", {"MessageSid" => "SMabc", "MessageStatus" => "delivered"}, signature: "x") + end + assert_response :no_content + assert_equal "delivered", delivery.reload.status + end + + test "unknown deliveries are acknowledged" do + callback("/smswire/status/twilio", {"MessageSid" => "SMnope", "MessageStatus" => "delivered"}) + assert_response :no_content + end + + test "unknown providers and providers without callbacks are not found" do + post "/smswire/status/carrier-pigeon", params: {"MessageSid" => "x"} + assert_response :not_found + post "/smswire/status/test", params: {"MessageSid" => "x"} + assert_response :not_found + end + + test "observers are told about status updates" do + delivery = send_via_twilio.delivery + seen = [] + observer = Object.new + observer.define_singleton_method(:delivered_sms) { |_result| } + observer.define_singleton_method(:sms_status_updated) { |record, update| seen << [record.id, update.status] } + Smswire.register_observer(observer) + callback("/smswire/status/twilio", {"MessageSid" => "SMabc", "MessageStatus" => "delivered"}) + assert_equal [[delivery.id, :delivered]], seen + ensure + Smswire.unregister_observer(observer) + end +end diff --git a/test/test_helper.rb b/test/test_helper.rb index 8bb525b..036d77f 100644 --- a/test/test_helper.rb +++ b/test/test_helper.rb @@ -1,6 +1,9 @@ ENV["RAILS_ENV"] = "test" require_relative "dummy/config/environment" + +ActiveRecord::Migration.verbose = false +ActiveRecord::MigrationContext.new(File.expand_path("../db/migrate", __dir__)).migrate require "minitest/autorun" require "webmock/minitest" @@ -32,4 +35,6 @@ def twilio_success(sid: "SM123", status: "queued", segments: "1") class Smswire::TestCase < ActiveSupport::TestCase include Smswire::TestHelper include ConfigHelpers + + setup { Smswire::Delivery.delete_all } end