Phase 2: deliveries, status callbacks, dedupe, log subscriber

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.

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"

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 |

app/controllers/smswire/status_callbacks_controller.rb

@@ -0,0 +1,43 @@
+module Smswire
+  # Receives provider delivery status callbacks at
+  # POST <mount>/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

app/models/smswire/application_record.rb

@@ -0,0 +1,5 @@
+module Smswire
+  class ApplicationRecord < ActiveRecord::Base
+    self.abstract_class = true
+  end
+end

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

config/routes.rb

@@ -0,0 +1,3 @@
+Smswire::Engine.routes.draw do
+  post "status/:provider", to: "status_callbacks#create", as: :status_callback
+end

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