feat(datasource-pylon): messages / conversation thread (EXT-9)

packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/client.rb

@@ -4,6 +4,12 @@ module ForestAdminDatasourcePylon
   class Client # rubocop:disable Metrics/ClassLength
     MAX_SEARCH_LIMIT = 1000
 
+    # Bounds `collect_pages`, which asks for a whole dataset rather than a
+    # window: the endpoints it reads answer in one response, so reaching this
+    # many pages means the API started paginating on its own and the walk is
+    # spending more of the per-minute budget than the answer is worth.
+    MAX_COLLECTED_PAGES = 10
+
     # `next_cursor` is nil as soon as Pylon stops advertising a next page, so
     # callers never have to know how the absence is spelled on the wire.
     SearchPage = Struct.new(:records, :next_cursor, keyword_init: true)
@@ -29,6 +35,23 @@ def fetch_issue(id)
       fetch_resource('issues', id)
     end
 
+    # The whole conversation of an issue, oldest message first, or nil when the
+    # thread could not be read.
+    #
+    # `limit` is left out on purpose: Pylon then answers with every message in a
+    # single response. Asking for a page would hand back the OLDEST messages and
+    # cut the most recent ones off, which is the half of a conversation nobody
+    # opens a ticket to read.
+    #
+    # The cursor is still followed, defensively: Pylon paginates this endpoint
+    # 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"
+
+      best_effort("fetch_issue_messages(#{issue_id})", default: nil) { must_succeed(path) { collect_pages(path) } }
+    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
@@ -99,6 +122,41 @@ def fetch_all(path, params = {})
       must_succeed(path) { Array(extract_data(connection.get(path, params).body)) }
     end
 
+    # Every record of a cursor-paginated GET, no window asked for and no limit
+    # sent. `CursorWalker` answers the other question — the offset/limit window a
+    # list view asks for — and is not what this needs.
+    #
+    # An empty page and a cursor that does not move both stop the loop: neither
+    # happens today, but a walk driven by a remote value stops on its own terms.
+    def collect_pages(path)
+      records = []
+      cursor = nil
+      pages = 0
+
+      loop do
+        page = to_search_page(connection.get(path, cursor.nil? ? {} : { 'cursor' => cursor }).body)
+        records.concat(page.records)
+        pages += 1
+        break if page.next_cursor.nil? || page.next_cursor == cursor || page.records.empty?
+
+        if pages >= MAX_COLLECTED_PAGES
+          log_pagination_cap(path, pages, records.size)
+          break
+        end
+
+        cursor = page.next_cursor
+      end
+
+      records
+    end
+
+    def log_pagination_cap(path, pages, collected)
+      ForestAdminDatasourcePylon.logger.warn(
+        "[forest_admin_datasource_pylon] Stopped paginating #{path} after #{pages} page(s) / " \
+        "#{collected} record(s); the rest is left out."
+      )
+    end
+
     # The id comes from operator-supplied filter values, so it is escaped before
     # being joined to the path.
     def fetch_resource(resource, id)
@@ -145,6 +203,18 @@ def must_succeed(operation)
       raise APIError, "Pylon API call failed: #{operation}: #{e.class}: #{e.message}"
     end
 
+    # For the calls whose result enriches a page rather than being the page: the
+    # failure is reported and the default returned, so a degraded thread or a
+    # missing enrichment costs the operator a column, not the record they opened.
+    def best_effort(operation, default:)
+      yield
+    rescue StandardError => e
+      ForestAdminDatasourcePylon.logger.warn(
+        "[forest_admin_datasource_pylon] #{operation} failed; degrading: #{e.class}: #{e.message}"
+      )
+      default
+    end
+
     # Builds an APIError preserving the HTTP status and Pylon's own error body so
     # smart actions can show the operator the real reason instead of "failed".
     def api_error(operation, error)

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

@@ -5,6 +5,7 @@ class Issue < BaseCollection
       include RecordSerialization
       include Serializer
       include RelationEmbedder
+      include MessagesEmbedder
 
       # `/issues/search` exposes no sort parameter, so the allow-list is empty and
       # every requested order is reported instead of being silently swallowed.
@@ -19,6 +20,32 @@ class Issue < BaseCollection
       # throttling that would let this cap grow.
       MAX_ID_LOOKUPS = 20
 
+      # The shape of one message inside the `messages` column. Field names follow
+      # the columns of this collection rather than the payload: Pylon spells them
+      # `message_html` and `timestamp`, which would put two conventions in the
+      # same schema for the operator to reconcile.
+      MESSAGE_THREAD_SCHEMA = {
+        'id' => 'String',
+        'body_html' => 'String',
+        'is_private' => 'Boolean',
+        'source' => 'String',
+        'thread_id' => 'String',
+        'file_urls' => 'Json',
+        'created_at' => 'Date',
+        'author_name' => 'String',
+        'author_email' => 'String',
+        'author_avatar_url' => 'String',
+        'author_contact_id' => 'String',
+        'author_user_id' => 'String'
+      }.freeze
+
+      # One thread is one `GET /issues/{id}/messages`, an endpoint allowing 20
+      # requests per minute: a page asking for more threads than this reads the
+      # first ones and reports the rest, for the same reason MAX_ID_LOOKUPS
+      # bounds the primary-key fan-out. Story 9 (EXT-13) owns the throttling
+      # that would let this cap grow.
+      MAX_MESSAGE_EMBEDS = 10
+
       def initialize(datasource, custom_fields: [])
         super(datasource, 'PylonIssue', custom_fields: custom_fields, searchable: true)
       end
@@ -27,6 +54,7 @@ def list(caller, filter, projection)
         records = fetch_records(caller, filter)
         rows = records.map { |record| project(record, projection) }
         embed_relations(records, rows, projection)
+        embed_messages(records, rows) if want_messages?(projection)
         rows
       end
 

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

@@ -0,0 +1,89 @@
+module ForestAdminDatasourcePylon
+  module Collections
+    class Issue < BaseCollection
+      # The conversation of an issue, embedded as a structured array column the
+      # 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
+      # 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
+      # ask for costs no request at all.
+      module MessagesEmbedder
+        include RecordSerialization
+
+        private
+
+        # The thread is read only when the projection names it. A nil projection
+        # — what a count or an export goes through — asks for the record as
+        # Pylon returns it, and embeds nothing, exactly like RelationEmbedder:
+        # spending one request per row on a path that never asked for the
+        # conversation is the very fan-out MAX_MESSAGE_EMBEDS exists to bound.
+        def want_messages?(projection)
+          Array(projection).map(&:to_s).any? { |p| p == 'messages' || p.start_with?('messages:') }
+        end
+
+        # A row past the cap, and a row whose thread failed to be read, are left
+        # at nil: "unknown", never the empty list, which would read as "this
+        # issue has no message" — the kind of answer that looks complete without
+        # being it.
+        def embed_messages(records, rows)
+          embedded = rows.first(MAX_MESSAGE_EMBEDS)
+          warn_truncated_threads(rows.size) if rows.size > embedded.size
+
+          embedded.each_with_index do |row, index|
+            messages = datasource.client.fetch_issue_messages(records[index]['id'])
+            row['messages'] = messages&.map { |message| serialize_message(message) }
+          end
+        end
+
+        def serialize_message(message)
+          attrs = message.is_a?(Hash) ? message : {}
+
+          {
+            'id' => attrs['id'],
+            'body_html' => attrs['message_html'],
+            'is_private' => attrs['is_private'],
+            'source' => attrs['source'],
+            'thread_id' => attrs['thread_id'],
+            'file_urls' => attrs['file_urls'],
+            'created_at' => attrs['timestamp']
+          }.merge(flatten_author(attrs['author']))
+        end
+
+        # Pylon nests the author's contact and user sides side by side, both
+        # optional and with nothing telling them apart: a message written by an
+        # agent carries `user`, one written by a customer carries `contact`. Both
+        # ids are kept, so a message stays traceable to the PylonContact or
+        # PylonUser record it came from, and the email is taken from whichever
+        # side is there.
+        def flatten_author(author)
+          attrs   = author.is_a?(Hash) ? author : {}
+          contact = attrs['contact']
+          user    = attrs['user']
+
+          {
+            'author_name' => attrs['name'],
+            'author_avatar_url' => attrs['avatar_url'],
+            'author_email' => nested_email(contact) || nested_email(user),
+            'author_contact_id' => nested_id(contact),
+            'author_user_id' => nested_id(user)
+          }
+        end
+
+        def nested_email(value)
+          value['email'] if value.is_a?(Hash)
+        end
+
+        def warn_truncated_threads(asked)
+          ForestAdminDatasourcePylon.logger.warn(
+            "[forest_admin_datasource_pylon] Asked for the message thread of #{asked} issues, reading the first " \
+            "#{MAX_MESSAGE_EMBEDS}: one request per issue would exhaust the rate limit of the agent. " \
+            'Narrow the selection, or take the thread out of the projection, to reach the records past this point.'
+          )
+        end
+      end
+    end
+  end
+end

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

@@ -68,6 +68,23 @@ def define_content_fields
           add_column('customer_portal_visible', 'Boolean')
           add_column('author_unverified', 'Boolean')
           add_column('number_of_touches', 'Number')
+          define_thread_field
+        end
+
+        # The conversation, embedded at read time by MessagesEmbedder. Declared
+        # by hand rather than through `add_column`: its type is the shape of one
+        # message, not a primitive.
+        #
+        # Neither filterable nor sortable — `POST /issues/search` covers no
+        # message field, and the thread is not even part of the payload the
+        # search endpoint returns — and not groupable either: `ColumnSchema`
+        # defaults that flag to true, where the `add_column` of the base passes
+        # false for every Pylon column, a thread being both an array and a value
+        # the pages of a cursor walk do not carry.
+        def define_thread_field
+          add_field('messages', ColumnSchema.new(column_type: [Issue::MESSAGE_THREAD_SCHEMA],
+                                                 filter_operators: [], is_groupable: false,
+                                                 is_read_only: true))
         end
 
         # Flattened from the nested `{id: …}` objects Pylon returns, and kept as

packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/client_spec.rb

@@ -250,6 +250,95 @@ def json(payload, status = 200)
     end
   end
 
+  describe '#fetch_issue_messages' do
+    let(:logger) { instance_double(Logger, warn: nil) }
+
+    def page(records, cursor: nil)
+      body = { 'data' => records }
+      body['pagination'] = { 'cursor' => cursor, 'has_next_page' => true } if cursor
+      json(body)
+    end
+
+    it 'returns the whole thread of an issue' do
+      stub_request(:get, "#{base}/issues/i1/messages").to_return(page([{ 'id' => 'm1' }, { 'id' => 'm2' }]))
+
+      expect(client.fetch_issue_messages('i1')).to eq([{ 'id' => 'm1' }, { 'id' => 'm2' }])
+    end
+
+    # Omitting `limit` is what makes Pylon answer with every message at once;
+    # asking for a page would hand back the oldest ones and cut off the rest.
+    it 'sends no limit, so Pylon answers with every message in one request' do
+      stub_request(:get, "#{base}/issues/i1/messages").to_return(page([]))
+
+      client.fetch_issue_messages('i1')
+
+      expect(WebMock).to have_requested(:get, "#{base}/issues/i1/messages").with(query: {}).once
+    end
+
+    it 'escapes an id that would otherwise alter the request path' do
+      stub_request(:get, "#{base}/issues/..%2Fme/messages").to_return(page([]))
+
+      client.fetch_issue_messages('../me')
+
+      expect(WebMock).to have_requested(:get, "#{base}/issues/..%2Fme/messages")
+    end
+
+    it 'follows the cursor when Pylon paginates the thread anyway' do
+      stub_request(:get, "#{base}/issues/i1/messages").with(query: {})
+                                                      .to_return(page([{ 'id' => 'm1' }], cursor: 'c1'))
+      stub_request(:get, "#{base}/issues/i1/messages").with(query: { 'cursor' => 'c1' })
+                                                      .to_return(page([{ 'id' => 'm2' }]))
+
+      expect(client.fetch_issue_messages('i1')).to eq([{ 'id' => 'm1' }, { 'id' => 'm2' }])
+    end
+
+    it 'stops on a cursor that does not move' do
+      stub_request(:get, "#{base}/issues/i1/messages").with(query: { 'cursor' => 'c1' })
+                                                      .to_return(page([{ 'id' => 'm2' }], cursor: 'c1'))
+      stub_request(:get, "#{base}/issues/i1/messages").with(query: {})
+                                                      .to_return(page([{ 'id' => 'm1' }], cursor: 'c1'))
+
+      expect(client.fetch_issue_messages('i1').size).to eq(2)
+    end
+
+    it 'stops on an empty page' do
+      stub_request(:get, "#{base}/issues/i1/messages").to_return(page([], cursor: 'c1'))
+
+      expect(client.fetch_issue_messages('i1')).to eq([])
+      expect(WebMock).to have_requested(:get, "#{base}/issues/i1/messages").once
+    end
+
+    it 'caps a thread Pylon never stops paginating, and says so' do
+      allow(ForestAdminDatasourcePylon).to receive(:logger).and_return(logger)
+      served = 0
+      stub_request(:get, %r{/issues/i1/messages}).to_return do
+        served += 1
+        page([{ 'id' => "m#{served}" }], cursor: "c#{served}")
+      end
+
+      client.fetch_issue_messages('i1')
+
+      expect(served).to eq(described_class::MAX_COLLECTED_PAGES)
+      expect(logger).to have_received(:warn).with(/Stopped paginating/)
+    end
+
+    # The thread enriches a page rather than being it: a failure costs the
+    # operator the column, not the records they opened.
+    it 'degrades to nil and reports the failure when the thread cannot be read' do
+      allow(ForestAdminDatasourcePylon).to receive(:logger).and_return(logger)
+      stub_request(:get, "#{base}/issues/i1/messages").to_return(json({ 'message' => 'boom' }, 500))
+
+      expect(client.fetch_issue_messages('i1')).to be_nil
+      expect(logger).to have_received(:warn).with(/fetch_issue_messages\(i1\) failed; degrading.*HTTP 500 boom/)
+    end
+
+    it 'degrades on a missing issue rather than raising a 404' do
+      stub_request(:get, "#{base}/issues/nope/messages").to_return(json({ 'message' => 'not found' }, 404))
+
+      expect(client.fetch_issue_messages('nope')).to be_nil
+    end
+  end
+
   describe '#search_accounts' do
     it 'posts the full search envelope and returns the records' do
       stub_request(:post, "#{base}/accounts/search").to_return(json('data' => [{ 'id' => 'a1' }]))