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,61 @@ 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, and the result message names any typed address left out. 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 with Pylon's own reason. Any other failure, a
+timeout included, 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, plus one `GET /users` with
+  `post_as_operator`. An unreadable thread is shown as an error, never as an issue without customer
+  message: both reply types stay offered, and an internal note does not depend on the thread.
+
 ## 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)
@@ -168,12 +178,14 @@ def list_resource(path, limit:, cursor: nil)
     # so the walk costs the single request it always did and only does something
     # the day that changes.
     #
-    # Past the cap the walk is refused rather than truncated, unlike the thread
-    # and the custom fields: `FetchAllCollection` filters, sorts, counts and
-    # groups in memory over what this returns, and calls the result exact
-    # BECAUSE it is every record Pylon holds. Truncated, that claim would stand
-    # over a fraction of the collection -- a count answered short and presented
-    # as exact, which is the one answer this datasource refuses everywhere else.
+    # Past the cap the walk is refused rather than truncated, like the thread a
+    # reply acts on (`fetch_complete_issue_messages`) and unlike the thread a
+    # page shows and the custom fields: `FetchAllCollection` filters, sorts,
+    # counts and groups in memory over what this returns, and calls the result
+    # exact BECAUSE it is every record Pylon holds. Truncated, that claim would
+    # stand over a fraction of the collection -- a count answered short and
+    # presented as exact, which is the one answer this datasource refuses
+    # everywhere else.
     def fetch_all(path, params = {})
       must_succeed(path) { collect_pages(path, params, refuse_past_cap: true) }
     end
@@ -193,7 +205,8 @@ def fetch_all(path, params = {})
     # Only `next_cursor` being nil is Pylon saying there is nothing more. Every
     # other stop leaves records behind, and `refuse_past_cap` is which of the two
     # answers the caller wants to those: a warning and a short list, or a refusal.
-    # See `fetch_all`, whose answer is only correct whole.
+    # See `fetch_all`, whose answer is only correct whole, and
+    # `fetch_complete_issue_messages`, whose caller acts on the latest message.
     #
     # `conn` is what a caller reading on the boot path hands its own connection
     # through: every page of the walk is then bounded like the first.
@@ -241,10 +254,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,142 @@
+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, dropped, refusal = reply_payload(state.target, body, context)
+        return [:error, refusal] if refusal
+
+        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) + left_out(dropped)) do
+          datasource.client.reply_to_issue(state.issue_id, payload)
+        end
+      end
+
+      def reply_payload(target, body, context)
+        payload = { 'body_html' => body, 'message_id' => target.message_id }
+        return [payload, []] unless target.email?
+
+        to, cc, dropped = target.sanitize(to: context.form_values[FormBuilder::TO_LABEL],
+                                          cc: context.form_values[FormBuilder::CC_LABEL])
+        refusal = "At least one recipient (To) is required to reply by email.#{left_out(dropped)}" if to.empty?
+        return [nil, dropped, refusal] if refusal
+
+        [payload.merge('email_info' => { 'to_emails' => to, 'cc_emails' => cc }), dropped]
+      end
+
+      def left_out(dropped)
+        dropped.empty? ? '' : " Left out as addresses of your team: #{dropped.join(", ")}."
+      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