feat(datasource): reply to a Pylon issue from Forest on the customer's channel [PRD-1873]

packages/forest_admin_datasource_pylon/README.md

@@ -169,6 +169,59 @@ The message body is HTML the operator writes and Pylon delivers to the requester
 requester address is a free-text field. Restrict both actions to the roles that should be able to
 send mail on your organization's behalf.
 
+### `ReplyToIssue`
+
+Answers one issue on the channel the customer used — a Slack issue gets a Slack reply, an email
+issue an email — or posts an internal note on it. Single scope only.
+
+```ruby
+@agent.collection :Ticket do |collection|
+  collection.use(ForestAdminDatasourcePylon::Plugins::ReplyToIssue, {
+    datasource: pylon_datasource,          # required
+    issue_id_field: 'pylon_issue_id',      # omit when the action sits on PylonIssue itself
+    action_name: 'Reply to Pylon issue',   # default
+    allow_internal_note: true,             # default
+    support_emails: ['support@example.com'],
+    post_as_operator: false                # default
+  })
+end
+```
+
+**Which channel.** Pylon delivers a reply into the conversation of the message it names, so the
+action picks that message: the channel is the one of the first customer-visible message, and the
+reply answers the latest customer-visible message on that channel. The form shows the channel
+read-only. An issue with no customer-visible message — one created as an internal note, for
+instance — cannot get a public reply; only the internal note is offered.
+
+**Email recipients.** Pylon does not infer them, so the form pre-fills To and CC from the latest
+email of the conversation. When the customer wrote it, To is its sender only: the addresses the
+customer wrote to include your support inbox. When the team wrote it, To is whom it went to. The
+`support_emails` and the addresses the team wrote from in the thread are always left out, from what
+the operator types as well. Set `support_emails`: before the team's first email in a thread, it is
+the only way to know your own addresses. A reply with no To is refused.
+
+**When Pylon does not confirm.** A 4xx is shown as Pylon words it. Any other failure may come after
+Pylon delivered the message, so the action says so and asks to check the issue before sending again.
+
+**Who signs.** By default the account owning the API token signs every reply, and the success
+message says so. With `post_as_operator: true`, a public reply is signed by the active Pylon user
+whose email matches the Forest user's, primary or secondary address. With no such user, or several,
+the reply is refused and nothing is sent: it never falls back to the token. Internal notes are
+always signed by the token.
+
+Three things to know before granting it.
+
+- **The column is the authority.** With `issue_id_field`, as with `CloseIssue`, an operator who can
+  write that column can reply to any Pylon issue, on your organization's behalf.
+- **`post_as_operator` trusts emails.** The token can sign as any agent of the workspace; the only
+  thing tying a Forest user to a Pylon user is that they share an email. Keep both user lists
+  consistent, and restrict the action to the roles that should answer customers.
+- **The thread is read on every form render**, one `GET /issues/{id}/messages` (120 a minute).
+  A submit costs one or two more: the agent renders the form again before running the action, and
+  a public reply reads the thread once more to answer its latest state. An unreadable thread is
+  shown as an error, never as an issue without customer message; an internal note does not depend
+  on it and stays available.
+
 ## Development
 
 ```bash

packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/client.rb

@@ -53,11 +53,19 @@ def fetch_issue(id)
     # when asked to, so a future default page size stays handled rather than
     # silently truncating the thread.
     def fetch_issue_messages(issue_id)
-      path = "issues/#{Faraday::Utils.escape(issue_id)}/messages"
+      path = issue_messages_path(issue_id)
 
       best_effort("fetch_issue_messages(#{issue_id})", default: nil) { must_succeed(path) { collect_pages(path) } }
     end
 
+    # For a caller acting on the thread: a thread that could not be read, or
+    # only in part, raises instead of reading as a conversation without it.
+    def fetch_complete_issue_messages(issue_id)
+      path = issue_messages_path(issue_id)
+
+      must_succeed(path) { collect_pages(path, refuse_past_cap: true) }
+    end
+
     def search_accounts(limit:, cursor: nil, filter: nil, search_text: nil)
       search_resource('accounts/search', limit: limit, cursor: cursor, filter: filter, search_text: search_text)
     end
@@ -144,6 +152,8 @@ def inspect
 
     private
 
+    def issue_messages_path(issue_id) = "issues/#{Faraday::Utils.escape(issue_id)}/messages"
+
     def search_resource(path, limit:, cursor: nil, filter: nil, search_text: nil)
       body = { 'limit' => clamp_limit(limit) }
       body['cursor']      = cursor unless blank?(cursor)
@@ -241,10 +251,8 @@ def truncation_reason(page, pages, seen)
     def refuse_truncated_walk(path, reason, pages, collected)
       raise APIError,
             "Pylon left the walk of #{path} short after #{pages} page(s) / #{collected} record(s): #{reason}. " \
-            'That endpoint is documented as answering with the whole collection in one response, which is what ' \
-            'PylonUser and PylonTeam filter, sort and count in memory as an exact answer. Refusing rather ' \
-            'than answering over a fraction of the collection: reading it needs cursor pagination, the way ' \
-            'PylonIssue reads its own.'
+            'That endpoint is documented as answering with the whole collection in one response, and its ' \
+            'caller needs all of it. Refusing rather than answering over a fraction of it.'
     end
 
     def log_truncated_walk(path, reason, pages, collected)

packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/client/writes.rb

@@ -29,8 +29,23 @@ def update_team(id, attributes) = patch_resource('teams', id, attributes)
 
       def update_user(id, attributes) = patch_resource('users', id, attributes)
 
+      # `message_id` picks the customer conversation the reply is delivered to.
+      def reply_to_issue(id, attributes) = post_message(issue_path(id, 'reply'), attributes)
+      def create_issue_note(id, attributes) = post_message(issue_path(id, 'note'), attributes)
+
       private
 
+      def issue_path(id, action) = "issues/#{Faraday::Utils.escape(id)}/#{action}"
+
+      # A 2xx means Pylon delivered the message, whatever the body says:
+      # refusing an unexpected body would invite the operator to send it twice.
+      def post_message(path, attributes)
+        must_succeed("create(#{path})") do
+          body = connection.post(path, attributes).body
+          body['data'] if body.is_a?(Hash) && body['data'].is_a?(Hash)
+        end
+      end
+
       def post_resource(resource, attributes)
         operation = "create(#{resource})"
 

packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue.rb

@@ -40,7 +40,10 @@ class Issue < BaseCollection
         'author_email' => 'String',
         'author_avatar_url' => 'String',
         'author_contact_id' => 'String',
-        'author_user_id' => 'String'
+        'author_user_id' => 'String',
+        'email_from' => 'String',
+        'email_to' => 'Json',
+        'email_cc' => 'Json'
       }.freeze
 
       # One thread is one `GET /issues/{id}/messages`, and a thread is the whole

packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue/messages_embedder.rb

@@ -5,7 +5,7 @@ class Issue < BaseCollection
       # way the Zendesk datasource embeds a ticket's comments.
       #
       # Pylon has no way to read the threads of several issues at once, so the
-      # thread costs one request per row — against an endpoint allowing 20 per
+      # thread costs one request per row — against an endpoint allowing 120 per
       # minute. The fan-out is therefore bounded like the primary-key lookups of
       # this collection: truncated with a warning rather than turned into a rate
       # limit error halfway through the page. A relation the projection does not
@@ -49,7 +49,19 @@ def serialize_message(message)
             'thread_id' => attrs['thread_id'],
             'file_urls' => attrs['file_urls'],
             'created_at' => attrs['timestamp']
-          }.merge(flatten_author(attrs['author']))
+          }.merge(flatten_author(attrs['author'])).merge(flatten_email(attrs['email_info']))
+        end
+
+        # Null outside email. `bcc_emails` and the RFC 5322 `message_id` header
+        # are left out: neither is something a reply should be built from.
+        def flatten_email(email_info)
+          attrs = email_info.is_a?(Hash) ? email_info : {}
+
+          {
+            'email_from' => attrs['from_email'],
+            'email_to' => attrs['to_emails'],
+            'email_cc' => attrs['cc_emails']
+          }
         end
 
         # Pylon nests the author's contact and user sides side by side, both

packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/plugins/reply_to_issue.rb

@@ -0,0 +1,134 @@
+require 'cgi'
+
+module ForestAdminDatasourcePylon
+  module Plugins
+    # Answers a Pylon issue on the channel the customer used, or posts an
+    # internal note on it. See the README for the channel rule and who signs.
+    class ReplyToIssue < ForestAdminDatasourceCustomizer::Plugins::Plugin
+      BaseAction      = ForestAdminDatasourceCustomizer::Decorators::Action::BaseAction
+      ActionScope     = ForestAdminDatasourceCustomizer::Decorators::Action::Types::ActionScope
+      ForestException = ForestAdminDatasourceToolkit::Exceptions::ForestException
+
+      NAME = 'Reply to Pylon issue'.freeze
+
+      def run(_datasource_customizer, collection_customizer = nil, options = {})
+        options = {} unless options.is_a?(Hash)
+        datasource = options[:datasource]
+        raise ForestException, 'ReplyToIssue plugin requires :datasource' unless datasource
+        raise ForestException, 'ReplyToIssue plugin requires a collection' unless collection_customizer
+
+        opts = {
+          issue_id_field: options[:issue_id_field],
+          allow_internal_note: options.fetch(:allow_internal_note, true) ? true : false,
+          post_as_operator: options[:post_as_operator] ? true : false,
+          support_emails: normalize_emails(options[:support_emails])
+        }
+
+        collection_customizer.add_action((options[:action_name] || NAME).to_s, build_action(datasource, opts))
+      end
+
+      private
+
+      def normalize_emails(value)
+        Array(value).map { |email| email.to_s.strip.downcase }.reject(&:empty?).uniq
+      end
+
+      def build_action(datasource, opts)
+        BaseAction.new(scope: ActionScope::SINGLE, form: FormBuilder.build(datasource, opts),
+                       &executor(datasource, opts))
+      end
+
+      # A note needs the issue only. A reply reads the thread again rather than
+      # trusting the form: the conversation may have moved since it rendered.
+      def executor(datasource, opts)
+        lambda do |context, result_builder|
+          issue_id = ThreadLookup.issue_id(context, opts)
+          next result_builder.error(message: ThreadLookup.no_issue(opts[:issue_id_field])) if issue_id.nil?
+
+          body = context.form_values[FormBuilder::MESSAGE_LABEL].to_s
+          next result_builder.error(message: 'A message is required.') if blank_html?(body)
+
+          status, message = if FormBuilder.public_reply?(context, opts)
+                              reply(datasource, ThreadLookup.read(datasource, issue_id, opts), body, context, opts)
+                            else
+                              note(datasource, issue_id, body)
+                            end
+          result_builder.public_send(status, message: message)
+        end
+      end
+
+      # An emptied rich-text editor still submits markup such as `<p><br></p>`.
+      def blank_html?(body)
+        return false if body.match?(/<img\b/i)
+
+        text = CGI.unescapeHTML(body.gsub(/<[^>]*>/, ' ').gsub(/&nbsp;/i, ' '))
+        text.gsub(/[[:space:]\u200B-\u200D\u2060\uFEFF]/, '').empty?
+      end
+
+      def reply(datasource, state, body, context, opts)
+        return [:error, state.error] if state.error
+        return [:error, no_target_message(state.issue_id, opts)] if state.target.nil?
+
+        payload = { 'body_html' => body, 'message_id' => state.target.message_id }
+        if state.target.email?
+          excluded = state.target.excluded
+          to = ReplyTarget.unique_addresses(Array(context.form_values[FormBuilder::TO_LABEL]), excluded)
+          return [:error, 'At least one recipient (To) is required to reply by email.'] if to.empty?
+
+          cc = ReplyTarget.unique_addresses(Array(context.form_values[FormBuilder::CC_LABEL]),
+                                            excluded + to.map(&:downcase))
+          payload['email_info'] = { 'to_emails' => to, 'cc_emails' => cc }
+        end
+
+        operator = OperatorLookup.call(datasource, context.caller) if opts[:post_as_operator]
+        return [:error, operator.error] if operator&.error
+
+        payload['user_id'] = operator.user['id'] if operator
+        write(state.issue_id, 'reply', reply_success(state, payload, operator&.user)) do
+          datasource.client.reply_to_issue(state.issue_id, payload)
+        end
+      end
+
+      def note(datasource, issue_id, body)
+        write(issue_id, 'internal note', "Internal note added to Pylon issue #{issue_id}.") do
+          datasource.client.create_issue_note(issue_id, 'body_html' => body)
+        end
+      end
+
+      # A 4xx is Pylon naming what is wrong, handed over as it is. Anything else
+      # may have failed after Pylon delivered the message, so the operator is
+      # told to check before sending it again rather than shown a bare failure.
+      def write(issue_id, what, success)
+        yield
+        [:success, success]
+      rescue APIError => e
+        ForestAdminDatasourcePylon.logger.warn(
+          "[forest_admin_datasource_pylon] #{what} on issue #{issue_id} failed: #{e.message}"
+        )
+        return [:error, e.message] if (400..499).cover?(e.status.to_i)
+
+        [:error, "Pylon did not confirm the #{what} on issue #{issue_id} (#{e.message}). It may have been " \
+                 'delivered: check the issue in Pylon before sending it again.']
+      end
+
+      def no_target_message(issue_id, opts)
+        message = "Pylon issue #{issue_id} has no customer-visible message, so it cannot get a public reply."
+        opts[:allow_internal_note] ? "#{message} Post an internal note instead." : message
+      end
+
+      def reply_success(state, payload, operator)
+        signed = if operator
+                   "signed by #{[operator["name"], operator["email"]].find { |v| !v.to_s.strip.empty? }}"
+                 else
+                   'signed by the account of the Pylon API token'
+                 end
+        if payload['email_info']
+          return "Reply emailed to #{payload["email_info"]["to_emails"].join(", ")} on Pylon issue " \
+                 "#{state.issue_id}, #{signed}."
+        end
+
+        "Reply sent via #{state.target.channel_label} on Pylon issue #{state.issue_id}, #{signed}."
+      end
+    end
+  end
+end

packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/plugins/reply_to_issue/form_builder.rb

@@ -0,0 +1,79 @@
+module ForestAdminDatasourcePylon
+  module Plugins
+    class ReplyToIssue
+      # Every dynamic property of the form reads the same ThreadLookup state,
+      # kept on the render's context so it lives exactly as long as the render.
+      # A WeakMap would let the state be collected mid-render.
+      module FormBuilder
+        FieldType = ForestAdminDatasourceCustomizer::Decorators::Action::Types::FieldType
+
+        TYPE_LABEL    = 'Reply type'.freeze
+        CHANNEL_LABEL = 'Channel'.freeze
+        TO_LABEL      = 'To'.freeze
+        CC_LABEL      = 'CC'.freeze
+        MESSAGE_LABEL = 'Message'.freeze
+
+        PUBLIC_REPLY  = 'Public reply'.freeze
+        INTERNAL_NOTE = 'Internal note'.freeze
+
+        module_function
+
+        def build(datasource, opts)
+          ivar  = :"@forest_admin_datasource_pylon_reply_state_#{opts.object_id}"
+          state = lambda do |context|
+            next context.instance_variable_get(ivar) if context.instance_variable_defined?(ivar)
+
+            context.instance_variable_set(ivar, ThreadLookup.call(datasource, context, opts))
+          end
+
+          fields = []
+          fields << type_field(state) if opts[:allow_internal_note]
+          fields << channel_field(state, opts)
+          fields << recipients_field(TO_LABEL, state, opts, :to)
+          fields << recipients_field(CC_LABEL, state, opts, :cc)
+          fields << message_field
+        end
+
+        def public_reply?(context, opts)
+          !opts[:allow_internal_note] || context.get_form_value(TYPE_LABEL) == PUBLIC_REPLY
+        end
+
+        def type_field(state)
+          { type: FieldType::ENUM, label: TYPE_LABEL, is_required: true,
+            enum_values: ->(context) { state.call(context).target ? [PUBLIC_REPLY, INTERNAL_NOTE] : [INTERNAL_NOTE] },
+            default_value: ->(context) { state.call(context).target ? PUBLIC_REPLY : INTERNAL_NOTE },
+            description: 'A public reply reaches the customer; an internal note is seen by your team only.' }
+        end
+
+        def channel_field(state, opts)
+          { type: FieldType::STRING, label: CHANNEL_LABEL, is_read_only: true,
+            if_condition: ->(context) { public_reply?(context, opts) || state.call(context).target.nil? },
+            value: ->(context) { channel_text(state.call(context), opts) },
+            description: 'Pylon delivers the reply on the channel the customer used. It cannot be changed here.' }
+        end
+
+        def channel_text(state, opts)
+          return state.error if state.error
+          return "Reply sent via #{state.target.channel_label}" if state.target
+
+          no_target = 'This issue has no customer-visible message, so it cannot get a public reply.'
+          opts[:allow_internal_note] ? "#{no_target} Post an internal note instead." : no_target
+        end
+
+        def recipients_field(label, state, opts, side)
+          { type: FieldType::STRING_LIST, label: label, is_required: side == :to,
+            if_condition: lambda { |context|
+              target = state.call(context).target
+              public_reply?(context, opts) && !target.nil? && target.email?
+            },
+            default_value: ->(context) { state.call(context).target&.public_send(side) || [] },
+            description: 'Pre-filled from the latest email of the conversation, support addresses left out.' }
+        end
+
+        def message_field
+          { type: FieldType::STRING, label: MESSAGE_LABEL, widget: 'RichText', is_required: true }
+        end
+      end
+    end
+  end
+end