Add send_email_notification rule action (#2527)

* Add send_email_notification rule action

Adds a new rule action that emails a digest of transactions matching a
rule. Re-syncs re-apply every active rule to all in-window matches, so a
notification_deliveries table (unique on rule_id + transaction_id) backs
deduplication and a per-rule watermark:

- Rule::ActionExecutor::SendEmailNotification plucks candidate ids, drops
  ones already in notification_deliveries, records the remainder BEFORE
  enqueuing (fail-safe: a crash suppresses rather than double-sends), and
  returns the count of newly-notified transactions.
- RuleEmailNotificationJob loads the rule + transactions and delivers the
  digest via RuleNotificationMailer.
- Creating the action pre-seeds all currently-matching transactions as
  already-delivered (after_create_commit), so the rule only emails about
  transactions appearing after the action exists.

Dedup keys on the DB row id, not provider identity, so a re-ingested
transaction (new id) may re-notify; accepted as benign.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Add view transactions link to rule notification digest email

Include a "View transactions" CTA (APP_DOMAIN/transactions) in both the
HTML and text versions of the rule notification digest email.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Address review feedback on rule email notifications

- Restrict digest delivery to family admins only (drop non-admin fallback)
- Make NotificationDelivery.record_for return only inserted ids and enqueue
  off that result, preventing duplicate digests under concurrent rule runs
- Seed the notification baseline when an existing action is changed to
  send_email_notification, so historical matches are not emailed
- Strengthen tests: assert the transactions CTA/link in the digest and the
  enqueued job's transaction-id args

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Include super_admin owner as rule digest recipient

The admin-only recipient lookup used find_by(role: :admin), which excluded
super_admin owners. A self-hosted family is commonly a single super_admin, so
the digest was silently skipped (NullMail no-op) and no email was sent.

Match the recipient on %w[admin super_admin] (the same pattern used elsewhere,
and consistent with User#admin?), while still excluding regular members/guests.
Add tests for the super_admin recipient and the no-admin skip path.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Add mixed-role digest recipient test

Cover the case where a family has both an admin and a super_admin. The recipient
lookup uses find_by(role: %w[admin super_admin]) with no ORDER BY, so which one
is returned is non-deterministic; the contract is only that the recipient is an
admin-level user (never a member). Assert recipient.admin? rather than a brittle
precedence between the two roles.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Deliver rule digest via deliver_later and order in DB

Use deliver_later so a slow/flaky SMTP connection doesn't tie up the
Sidekiq worker, and push the entry-date sort into the query instead of
materializing and sorting the result set in Ruby.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Replace raw hex colors with email-safe design tokens in digest template

The rule digest email hardcoded Tailwind slate-100/200 hex values for
table borders, which aren't part of this project's design system.
Resolve to the actual border-primary/border-secondary token values and
centralize them as a reusable .email-table class in the mailer layout.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

app/jobs/rule_email_notification_job.rb

@@ -0,0 +1,17 @@
+class RuleEmailNotificationJob < ApplicationJob
+  queue_as :medium_priority
+
+  def perform(rule_id, transaction_ids)
+    rule = Rule.find_by(id: rule_id)
+    return unless rule
+
+    transactions = rule.family.transactions
+                       .where(id: transaction_ids)
+                       .includes(entry: :account)
+                       .references(:entry)
+                       .order("entries.date DESC")
+                       .to_a
+
+    RuleNotificationMailer.digest(rule: rule, transactions: transactions).deliver_later if transactions.any?
+  end
+end

app/mailers/rule_notification_mailer.rb

@@ -0,0 +1,20 @@
+class RuleNotificationMailer < ApplicationMailer
+  def digest(rule:, transactions:)
+    @rule = rule
+    @transactions = transactions
+    @family = rule.family
+    @transactions_url = transactions_url
+
+    # Admins only: the digest contains transaction details, so never widen the
+    # recipient set to a regular member/guest. super_admin is the family owner
+    # and must be included (see User#admin?); a self-hosted family is commonly a
+    # single super_admin. Skip delivery entirely when there is no admin.
+    recipient = @family.users.find_by(role: %w[admin super_admin])
+    return if recipient.nil?
+
+    mail(
+      to: recipient.email,
+      subject: t(".subject", count: transactions.size, product_name: product_name)
+    )
+  end
+end

app/models/notification_delivery.rb

@@ -0,0 +1,35 @@
+class NotificationDelivery < ApplicationRecord
+  belongs_to :rule
+  # The association is named :transaction_record rather than :transaction because
+  # ActiveRecord refuses to define a :transaction association (it would clash with
+  # the built-in #transaction method). The underlying column is still
+  # transaction_id; dedup keys on that column directly.
+  belongs_to :transaction_record, class_name: "Transaction", foreign_key: :transaction_id
+
+  # Records deliveries race-safely in a single insert_all keyed on the unique
+  # (rule_id, transaction_id) index, and returns ONLY the transaction_ids this
+  # call actually inserted. Rows that already exist are skipped by the unique
+  # index (no raise) and excluded from the result, so two concurrent runs that
+  # observe the same candidates each get a disjoint set back — the caller can
+  # enqueue off the return value without re-notifying already-delivered ids.
+  #
+  # Dedup keys on the DB row id (`transaction_id`), NOT provider identity: a
+  # re-ingested transaction gets a new id and may notify again. This is accepted
+  # as benign (see Rule::ActionExecutor::SendEmailNotification).
+  def self.record_for(rule_id:, transaction_ids:)
+    return [] if transaction_ids.blank?
+
+    now = Time.current
+    rows = transaction_ids.map do |transaction_id|
+      { rule_id: rule_id, transaction_id: transaction_id, created_at: now, updated_at: now }
+    end
+
+    result = insert_all(
+      rows,
+      unique_by: :index_notification_deliveries_on_rule_and_transaction,
+      returning: [ :transaction_id ]
+    )
+
+    result.rows.flatten
+  end
+end

app/models/rule.rb

@@ -40,6 +40,14 @@ class Rule < ApplicationRecord
     matching_resources_scope.count
   end
 
+  # Public wrapper around the private matching scope so callers can read the
+  # currently-matching transaction ids WITHOUT running executors (e.g. the
+  # notification baseline pre-seed). Mirrors total_affected_resource_count,
+  # which also reaches matching_resources_scope.
+  def matching_transaction_ids
+    matching_resources_scope.pluck(:id)
+  end
+
   # Creates a categorization rule for the Quick Categorize Wizard.
   # Returns the saved rule, or nil if a duplicate or invalid rule already exists.
   def self.create_from_grouping(family, grouping_key, category, transaction_type: nil)

app/models/rule/action.rb

@@ -3,6 +3,23 @@ class Rule::Action < ApplicationRecord
 
   validates :action_type, presence: true
 
+  # Pre-seed (watermark): when a send_email_notification action is created — on a
+  # new rule OR added to an existing one — record all currently-matching
+  # transactions as already-delivered WITHOUT sending, so the rule only ever
+  # emails about transactions that appear AFTER the action exists.
+  #
+  # Uses after_create_commit (not after_create): nested children persist before
+  # the parent rule commits, and the pre-seed reads the rule's conditions, which
+  # must be committed first.
+  #
+  # after_update_commit covers the edit flow: the action_type select is editable
+  # for persisted actions (see rules_controller#rule_params), so an existing
+  # action can be CHANGED to send_email_notification. Without re-seeding, the
+  # next apply/sync would email every historical match. Guard on the type change
+  # so we only watermark when an action actually becomes email-notify.
+  after_create_commit :seed_notification_baseline
+  after_update_commit :seed_notification_baseline, if: :saved_change_to_action_type?
+
   def apply(resource_scope, ignore_attribute_locks: false, rule_run: nil)
     executor.execute(resource_scope, value: value, ignore_attribute_locks: ignore_attribute_locks, rule_run: rule_run) || 0
   end
@@ -26,4 +43,14 @@ class Rule::Action < ApplicationRecord
   def executor
     rule.registry.get_executor!(action_type)
   end
+
+  private
+    def seed_notification_baseline
+      return unless action_type == "send_email_notification"
+
+      NotificationDelivery.record_for(
+        rule_id: rule_id,
+        transaction_ids: rule.matching_transaction_ids
+      )
+    end
 end

app/models/rule/action_executor/send_email_notification.rb

@@ -0,0 +1,30 @@
+class Rule::ActionExecutor::SendEmailNotification < Rule::ActionExecutor
+  def label
+    "Send email notification"
+  end
+
+  # rule_run is accepted for interface compatibility but unused: the digest email
+  # is fire-and-forget and is not tracked as part of RuleRun accounting.
+  def execute(transaction_scope, value: nil, ignore_attribute_locks: false, rule_run: nil)
+    candidate_ids = transaction_scope.pluck(:id)
+
+    # record_for atomically inserts and returns ONLY the ids this run actually
+    # claimed. We enqueue off that result (not the pre-insert candidate list) so
+    # two concurrent runs over the same matches never enqueue duplicate digests:
+    # whichever loses the unique-index race gets those ids back as empty.
+    #
+    # Recording is the dedup boundary, since re-syncs re-apply every active rule
+    # to all in-window matches (not just newly ingested transactions). If the
+    # process crashes after recording but before delivery, the next run suppresses
+    # these ids rather than re-sending — we would rather miss a digest than spam.
+    new_transaction_ids = NotificationDelivery.record_for(rule_id: rule.id, transaction_ids: candidate_ids)
+
+    return 0 if new_transaction_ids.empty?
+
+    RuleEmailNotificationJob.perform_later(rule.id, new_transaction_ids)
+
+    # Synchronous count of newly-notified transactions. The email itself is
+    # delivered out-of-band by the job and is not part of RuleRun accounting.
+    new_transaction_ids.size
+  end
+end

app/models/rule/registry/transaction_resource.rb

@@ -24,7 +24,8 @@ class Rule::Registry::TransactionResource < Rule::Registry
       Rule::ActionExecutor::SetTransactionName.new(rule),
       Rule::ActionExecutor::SetInvestmentActivityLabel.new(rule),
       Rule::ActionExecutor::ExcludeTransaction.new(rule),
-      Rule::ActionExecutor::SetAsTransferOrPayment.new(rule)
+      Rule::ActionExecutor::SetAsTransferOrPayment.new(rule),
+      Rule::ActionExecutor::SendEmailNotification.new(rule)
     ]
 
     if ai_enabled?

app/views/layouts/mailer.html.erb

@@ -46,6 +46,22 @@
         margin-top: 32px;
         text-align: center;
       }
+      .email-table {
+        border-collapse: collapse;
+        text-align: left;
+        width: 100%;
+      }
+      .email-table th {
+        border-bottom: 1px solid rgba(11, 11, 11, 0.15); /* border-primary */
+        padding: 8px;
+      }
+      .email-table td {
+        border-bottom: 1px solid rgba(11, 11, 11, 0.1); /* border-secondary */
+        padding: 8px;
+      }
+      .email-table .text-right {
+        text-align: right;
+      }
     </style>
   </head>
 

app/views/rule_notification_mailer/digest.html.erb

@@ -0,0 +1,28 @@
+<h1><%= t(".heading", count: @transactions.size) %></h1>
+
+<p><%= t(".intro", rule: @rule.name.presence || @rule.primary_condition_title) %></p>
+
+<table class="email-table">
+  <thead>
+    <tr>
+      <th><%= t(".date") %></th>
+      <th><%= t(".name") %></th>
+      <th><%= t(".account") %></th>
+      <th class="text-right"><%= t(".amount") %></th>
+    </tr>
+  </thead>
+  <tbody>
+    <% @transactions.each do |txn| %>
+      <tr>
+        <td><%= I18n.l(txn.entry.date, format: :long) %></td>
+        <td><%= txn.entry.name %></td>
+        <td><%= txn.entry.account.name %></td>
+        <td class="text-right"><%= txn.entry.amount_money.format %></td>
+      </tr>
+    <% end %>
+  </tbody>
+</table>
+
+<p style="margin-top: 16px;">
+  <%= link_to t(".cta"), @transactions_url, class: "button" %>
+</p>

app/views/rule_notification_mailer/digest.text.erb

@@ -0,0 +1,9 @@
+<%= t(".heading", count: @transactions.size) %>
+
+<%= t(".intro", rule: @rule.name.presence || @rule.primary_condition_title) %>
+
+<% @transactions.each do |txn| %>
+- <%= I18n.l(txn.entry.date, format: :long) %> | <%= txn.entry.name %> | <%= txn.entry.account.name %> | <%= txn.entry.amount_money.format %>
+<% end %>
+
+<%= t(".cta") %>: <%= @transactions_url %>

config/locales/mailers/rule_notification_mailer/en.yml

@@ -0,0 +1,16 @@
+---
+en:
+  rule_notification_mailer:
+    digest:
+      subject:
+        one: "1 new transaction matched your rule on %{product_name}"
+        other: "%{count} new transactions matched your rule on %{product_name}"
+      heading:
+        one: "1 new transaction matched your rule"
+        other: "%{count} new transactions matched your rule"
+      intro: 'These transactions matched the rule "%{rule}":'
+      date: "Date"
+      name: "Name"
+      account: "Account"
+      amount: "Amount"
+      cta: "View transactions"

db/migrate/20260627000000_create_notification_deliveries.rb

@@ -0,0 +1,15 @@
+class CreateNotificationDeliveries < ActiveRecord::Migration[7.2]
+  def change
+    create_table :notification_deliveries, id: :uuid do |t|
+      t.references :rule, null: false, foreign_key: { on_delete: :cascade }, type: :uuid
+      # Transactions are deleted and re-ingested frequently during syncs, so we
+      # cascade rather than block their deletion. Dedup keys on this row id.
+      t.references :transaction, null: false, foreign_key: { on_delete: :cascade }, type: :uuid
+
+      t.timestamps
+    end
+
+    add_index :notification_deliveries, [ :rule_id, :transaction_id ],
+      unique: true, name: "index_notification_deliveries_on_rule_and_transaction"
+  end
+end

db/schema.rb

@@ -1406,6 +1406,16 @@ ActiveRecord::Schema[7.2].define(version: 2026_07_14_120000) do
     t.index ["user_id"], name: "index_mobile_devices_on_user_id"
   end
 
+  create_table "notification_deliveries", id: :uuid, default: -> { "gen_random_uuid()" }, force: :cascade do |t|
+    t.uuid "rule_id", null: false
+    t.uuid "transaction_id", null: false
+    t.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index ["rule_id", "transaction_id"], name: "index_notification_deliveries_on_rule_and_transaction", unique: true
+    t.index ["rule_id"], name: "index_notification_deliveries_on_rule_id"
+    t.index ["transaction_id"], name: "index_notification_deliveries_on_transaction_id"
+  end
+
   create_table "oauth_access_grants", force: :cascade do |t|
     t.string "resource_owner_id", null: false
     t.bigint "application_id", null: false
@@ -2295,6 +2305,8 @@ ActiveRecord::Schema[7.2].define(version: 2026_07_14_120000) do
   add_foreign_key "mercury_items", "families"
   add_foreign_key "messages", "chats"
   add_foreign_key "mobile_devices", "users"
+  add_foreign_key "notification_deliveries", "rules", on_delete: :cascade
+  add_foreign_key "notification_deliveries", "transactions", on_delete: :cascade
   add_foreign_key "oauth_access_grants", "oauth_applications", column: "application_id"
   add_foreign_key "oauth_access_tokens", "oauth_applications", column: "application_id"
   add_foreign_key "oidc_identities", "users"

test/mailers/rule_notification_mailer_test.rb

@@ -0,0 +1,78 @@
+require "test_helper"
+
+class RuleNotificationMailerTest < ActionMailer::TestCase
+  include EntriesTestHelper
+
+  test "digest" do
+    rule = rules(:one)
+    rule.update!(name: "Coffee rule")
+    family = rule.family
+    admin = family.users.find_by(role: %w[admin super_admin])
+    account = family.accounts.create!(name: "Mailer test", balance: 100, currency: "USD", accountable: Depository.new)
+    txn = create_transaction(date: Date.current, account: account, amount: 100, name: "Coffee").transaction
+
+    mail = RuleNotificationMailer.digest(rule: rule, transactions: [ txn ])
+
+    # The mailer derives the recipient from rule.family, so assert against that
+    # admin explicitly rather than an unrelated fixture.
+    assert_equal [ admin.email ], mail.to
+    assert_equal I18n.t(
+      "rule_notification_mailer.digest.subject",
+      count: 1,
+      product_name: Rails.configuration.x.product_name
+    ), mail.subject
+    assert_match "Coffee", mail.body.encoded
+
+    # The "View transactions" CTA must link to the transactions page in both parts.
+    html = mail.html_part.body.encoded
+    text = mail.text_part.body.encoded
+    assert_match I18n.t("rule_notification_mailer.digest.cta"), html
+    assert_match %r{/transactions}, html
+    assert_match %r{/transactions}, text
+  end
+
+  test "digest is delivered to the super_admin owner when there is no plain admin" do
+    # A self-hosted family is commonly a single super_admin (the owner) with no
+    # :admin user. The owner must still receive the digest.
+    family = Family.create!(name: "Solo owner family", currency: "USD")
+    owner = User.create!(family: family, email: "solo-owner@example.com", password: "password123", role: :super_admin)
+    account = family.accounts.create!(name: "Mailer test", balance: 100, currency: "USD", accountable: Depository.new)
+    txn = create_transaction(date: Date.current, account: account, amount: 100, name: "Coffee").transaction
+    rule = Rule.new(family: family, resource_type: "transaction", name: "Coffee rule")
+
+    mail = RuleNotificationMailer.digest(rule: rule, transactions: [ txn ])
+
+    assert_equal [ owner.email ], mail.to
+  end
+
+  test "digest recipient is an admin-level user when both admin and super_admin exist" do
+    # find_by(role: %w[admin super_admin]) has no ORDER BY, so which of the two
+    # is returned is not deterministic and precedence is intentionally undefined.
+    # The contract is only that the recipient is admin-level, never a member.
+    family = Family.create!(name: "Mixed roles family", currency: "USD")
+    User.create!(family: family, email: "the-admin@example.com", password: "password123", role: :admin)
+    User.create!(family: family, email: "the-super-admin@example.com", password: "password123", role: :super_admin)
+    User.create!(family: family, email: "the-member@example.com", password: "password123", role: :member)
+    account = family.accounts.create!(name: "Mailer test", balance: 100, currency: "USD", accountable: Depository.new)
+    txn = create_transaction(date: Date.current, account: account, amount: 100, name: "Coffee").transaction
+    rule = Rule.new(family: family, resource_type: "transaction", name: "Coffee rule")
+
+    mail = RuleNotificationMailer.digest(rule: rule, transactions: [ txn ])
+
+    recipient = family.users.find_by!(email: mail.to.first)
+    assert recipient.admin?, "expected an admin-level recipient, got role=#{recipient.role}"
+  end
+
+  test "digest is skipped when the family has no admin or super_admin" do
+    # Transaction details must never go to a regular member/guest.
+    family = Family.create!(name: "No-admin family", currency: "USD")
+    User.create!(family: family, email: "member-only@example.com", password: "password123", role: :member)
+    account = family.accounts.create!(name: "Mailer test", balance: 100, currency: "USD", accountable: Depository.new)
+    txn = create_transaction(date: Date.current, account: account, amount: 100, name: "Coffee").transaction
+    rule = Rule.new(family: family, resource_type: "transaction", name: "Coffee rule")
+
+    assert_no_emails do
+      RuleNotificationMailer.digest(rule: rule, transactions: [ txn ]).deliver_now
+    end
+  end
+end

test/models/rule/action_executor/send_email_notification_test.rb

@@ -0,0 +1,84 @@
+require "test_helper"
+
+class Rule::ActionExecutor::SendEmailNotificationTest < ActiveSupport::TestCase
+  include EntriesTestHelper, ActiveJob::TestHelper
+
+  setup do
+    @family = families(:dylan_family)
+    @rule = rules(:one)
+    @account = @family.accounts.create!(name: "Notify test", balance: 1000, currency: "USD", accountable: Depository.new)
+    @txn1 = create_transaction(date: Date.current, account: @account, amount: 100, name: "Coffee").transaction
+    @scope = @account.transactions
+  end
+
+  def action
+    Rule::Action.new(rule: @rule, action_type: "send_email_notification")
+  end
+
+  test "enqueues one digest job for new matches, records deliveries, returns integer count" do
+    result = nil
+
+    assert_difference -> { NotificationDelivery.where(rule: @rule).count }, 1 do
+      assert_enqueued_with(job: RuleEmailNotificationJob) do
+        result = action.apply(@scope)
+      end
+    end
+
+    assert_equal 1, result
+    assert_includes NotificationDelivery.where(rule: @rule).pluck(:transaction_id), @txn1.id
+  end
+
+  test "dedup suppresses repeat sends across runs" do
+    action.apply(@scope)
+
+    result = nil
+    assert_no_enqueued_jobs only: RuleEmailNotificationJob do
+      result = action.apply(@scope)
+    end
+
+    assert_equal 0, result
+    assert_equal 1, NotificationDelivery.where(rule: @rule).count
+  end
+
+  test "only newly appearing transactions trigger a job" do
+    action.apply(@scope) # baseline: txn1 recorded + enqueued
+
+    txn2 = create_transaction(date: Date.current, account: @account, amount: 50, name: "Lunch").transaction
+
+    result = nil
+    # Only txn2 (the newly appearing match) may be enqueued — never the already
+    # notified txn1. Asserting the args, not just the job class, locks that down.
+    assert_enqueued_with(job: RuleEmailNotificationJob, args: [ @rule.id, [ txn2.id ] ]) do
+      result = action.apply(@scope)
+    end
+
+    assert_equal 1, result
+    recorded = NotificationDelivery.where(rule: @rule).pluck(:transaction_id)
+    assert_includes recorded, txn2.id
+    assert_equal 2, recorded.size
+  end
+
+  test "pre-seed watermark records existing matches without sending so history never emails" do
+    # after_create_commit does not fire under transactional tests (the wrapping
+    # transaction never commits), so invoke the seeding path directly to verify
+    # the watermark behavior the callback performs in production.
+    rule = @family.rules.create!(
+      resource_type: "transaction",
+      actions_attributes: [ { action_type: "send_email_notification" } ]
+    )
+    seed_action = rule.actions.first
+
+    assert_no_enqueued_jobs only: RuleEmailNotificationJob do
+      seed_action.send(:seed_notification_baseline)
+    end
+
+    # Pre-existing @txn1 is now watermarked, so a subsequent run sends nothing.
+    result = nil
+    assert_no_enqueued_jobs only: RuleEmailNotificationJob do
+      result = Rule::Action.new(rule: rule, action_type: "send_email_notification").apply(@scope)
+    end
+
+    assert_equal 0, result
+    assert_includes NotificationDelivery.where(rule: rule).pluck(:transaction_id), @txn1.id
+  end
+end