Module: Legion::Extensions::MicrosoftTeams::Runners::ApiIngest

Extended by:
Definitions, ApiIngest
Includes:
Helpers::Lex, Helpers::Client, Helpers::GraphCache, Helpers::HighWaterMark, Helpers::PermissionGuard
Included in:
ApiIngest
Defined in:
lib/legion/extensions/microsoft_teams/runners/api_ingest.rb

Constant Summary collapse

MAX_CHAT_PAGES = 
10
CHAT_TYPE_PRIORITY = 
{ 'oneOnOne' => 0, 'group' => 1, 'meeting' => 2 }.freeze

Constants included from Helpers::HighWaterMark

Helpers::HighWaterMark::HWM_TTL

Constants included from Helpers::PermissionGuard

Helpers::PermissionGuard::BACKOFF_SCHEDULE

Instance Method Summary collapse

Methods included from Helpers::GraphCache

#cached_graph_get, #graph_cache_ttl, #graph_user_key, #invalidate_graph_cache

Methods included from Helpers::HighWaterMark

#get_extended_hwm, #get_hwm, #hwm_key, #new_messages, #persist_hwm_as_trace, #restore_hwm_from_traces, #set_extended_hwm, #set_hwm, #update_extended_hwm, #update_hwm_from_messages

Methods included from Helpers::PermissionGuard

#denial_info, #guarded_request, #permission_denied?, #record_denial, #reset_denials!

Methods included from Helpers::Client

#bot_connection, #graph_connection, #oauth_connection, #user_path

Instance Method Details

#ingest_api(token:, top_people: 15, message_depth: 50, skip_bots: true, imprint_active: false) ⇒ Object

rubocop:disable Metrics/AbcSize,Metrics/MethodLength



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/legion/extensions/microsoft_teams/runners/api_ingest.rb', line 25

def ingest_api(token:, top_people: 15, message_depth: 50, skip_bots: true, imprint_active: false, **) # rubocop:disable Metrics/AbcSize,Metrics/MethodLength
  log.debug("ApiIngest#ingest_api top_people=#{top_people} message_depth=#{message_depth}")
  return error_result('lex-memory not loaded') unless memory_available?
  return error_result('no token provided') unless token && !token.empty?

  @fetch_failures = 0
  restore_hwm_from_traces

  people = fetch_top_people(token: token, top: top_people)
  log.debug("ApiIngest: fetched #{people.size} top people")
  return error_result('people endpoint denied or empty') if people.empty?

  chats = fetch_one_on_one_chats(token: token)
  log.debug("ApiIngest: fetched #{chats.size} oneOnOne chats")
  return error_result('no 1:1 chats found') if chats.empty?

  existing_hashes = load_existing_hashes
  conn = graph_connection(token: token)
  chat_index = build_chat_member_index(conn: conn, chats: chats)
  stored = 0
  skipped = 0
  people_ingested = 0
  thread_groups = Hash.new { |h, k| h[k] = [] }
  person_texts = Hash.new { |h, k| h[k] = [] }

  people.each do |person|
    chat = find_chat_for_person_indexed(person: person, chat_index: chat_index)
    unless chat
      log.debug("ApiIngest: no chat match for #{person['displayName']} " \
                "(email=#{person.dig('scoredEmailAddresses', 0, 'address')}, id=#{person['id']})")
      next
    end
    log.info("ApiIngest: matched #{person['displayName']} to chat #{chat['id']}")

    messages = fetch_chat_messages(conn: conn, chat_id: chat['id'], depth: message_depth)
    next if messages.empty?

    msg_stored = 0
    messages.each do |msg|
      next if skip_bots && bot_message_graph?(msg)

      text = extract_body_text(msg)
      next if text.length < 5

      content_hash = msg['id'] || Digest::SHA256.hexdigest(text)[0, 16]
      if existing_hashes.include?(content_hash)
        skipped += 1
        next
      end

      trace_result = store_graph_message(msg, text, person, chat['id'],
                                         content_hash:   content_hash,
                                         imprint_active: imprint_active)
      if trace_result
        stored += 1
        msg_stored += 1
        existing_hashes << content_hash
        thread_groups[chat['id']] << trace_result[:trace_id]
        person_texts[person['displayName']] << text
      else
        skipped += 1
      end
    end

    next unless msg_stored.positive?

    people_ingested += 1
    update_extended_hwm(chat_id: chat['id'],
                        last_message_at: messages.filter_map { |m| m['createdDateTime'] }.max,
                        new_message_count: msg_stored, ingested: true)
  end

  coactivate_thread_traces(thread_groups)
  flush_trace_store if stored.positive?
  apollo_results = publish_to_apollo(person_texts) if stored.positive?

  { result: { stored: stored, skipped: skipped, people_ingested: people_ingested,
              people_found: people.length, chats_found: chats.length,
              fetch_failures: @fetch_failures,
              apollo: apollo_results } }
# rubocop:disable Legion/RescueLogging/NoCapture
# Re-raise unlogged: surface the typed throttle to the caller (the
# ApiIngest actor) so it can defer its next scheduled run by the
# advertised retry_after. The throttle is already logged at the
# middleware/circuit layer; folding it into an error result here
# would hide the one signal the actor needs to stop re-charging
# the shared Graph circuit on its fixed interval.
rescue Errors::Throttled
  raise
  # rubocop:enable Legion/RescueLogging/NoCapture
rescue StandardError => e
  handle_exception(e, level: :error, operation: 'ApiIngest#ingest_api')
  { result: { stored: stored || 0, skipped: skipped || 0,
              fetch_failures: @fetch_failures || 0, error: e.message } }
end