Class: Webhookdb::Replicator::ConvertkitSubscriberV1

Inherits:
Base
  • Object
show all
Includes:
Appydays::Loggable, ConvertkitV1Mixin
Defined in:
lib/webhookdb/replicator/convertkit_subscriber_v1.rb

Constant Summary

Constants included from ConvertkitV1Mixin

Webhookdb::Replicator::ConvertkitV1Mixin::CONV_FIND_CANCELED_AT

Constants included from DBAdapter::ColumnTypes

DBAdapter::ColumnTypes::BIGINT, DBAdapter::ColumnTypes::BIGINT_ARRAY, DBAdapter::ColumnTypes::BOOLEAN, DBAdapter::ColumnTypes::COLUMN_TYPES, DBAdapter::ColumnTypes::DATE, DBAdapter::ColumnTypes::DECIMAL, DBAdapter::ColumnTypes::DOUBLE, DBAdapter::ColumnTypes::FLOAT, DBAdapter::ColumnTypes::INTEGER, DBAdapter::ColumnTypes::INTEGER_ARRAY, DBAdapter::ColumnTypes::OBJECT, DBAdapter::ColumnTypes::TEXT, DBAdapter::ColumnTypes::TEXT_ARRAY, DBAdapter::ColumnTypes::TIMESTAMP, DBAdapter::ColumnTypes::UUID

Constants inherited from Base

Base::MAX_INDEX_NAME_LENGTH

Instance Attribute Summary

Attributes inherited from Base

#service_integration

Class Method Summary collapse

Instance Method Summary collapse

Methods included from ConvertkitV1Mixin

#_remote_key_column, #_verify_backfill_401_err_msg, #_verify_backfill_err_msg, #_webhook_response, #backfiller_server_error_backoff, #backfiller_server_error_retries, #calculate_backfill_state_machine

Methods inherited from Base

#_any_subscriptions_to_notify?, #_backfill_state_change_fields, #_backfillers, #_clear_backfill_information, #_clear_webook_information, #_coalesce_excluded_on_update, #_enqueue_backfill_jobs, #_extra_index_specs, #_fetch_enrichment, #_find_dependency_candidate, #_notify_dependents, #_parallel_backfill, #_prepare_for_insert, #_publish_rowupsert, #_remote_key_column, #_resource_to_data, #_store_enrichment_body?, #_to_json, #_upsert_webhook, #_verify_backfill_err_msg, #_webhook_response, #_webhook_state_change_fields, #admin_dataset, #backfill, #backfill_not_supported_message, #calculate_and_backfill_state_machine, #calculate_backfill_state_machine, #calculate_dependency_state_machine_step, #calculate_preferred_create_state_machine, chunked_row_update_bounds, #clear_backfill_information, #clear_webhook_information, #create_table, #create_table_modification, #data_column, #dbadapter_table, #denormalized_columns, #descriptor, #dispatch_request_to, #documentation_url, #enqueue_sync_targets, #enrichment_column, #ensure_all_columns, #ensure_all_columns_modification, #find_dependent, #find_dependent!, #indices, #initialize, #on_dependency_webhook_upsert, #preferred_create_state_machine_method, #preprocess_headers_for_logging, #primary_key_column, #process_webhooks_synchronously?, #qualified_table_sequel_identifier, #readonly_dataset, #remote_key_column, #requires_sequence?, #resource_name_plural, #resource_name_singular, #schema_and_table_symbols, #storable_columns, #synchronous_processing_response_body, #timestamp_column, #upsert_has_deps?, #upsert_webhook, #upsert_webhook_body, #verify_backfill_credentials, #webhook_endpoint, #webhook_response

Constructor Details

This class inherits a constructor from Webhookdb::Replicator::Base

Class Method Details

.descriptor ⇒ Webhookdb::Replicator::Descriptor

Returns:



12
13
14
15
16
17
18
19
20
21
22
23
24
25
 
# File 'lib/webhookdb/replicator/convertkit_subscriber_v1.rb', line 12

def self.descriptor
  return Webhookdb::Replicator::Descriptor.new(
    name: "convertkit_subscriber_v1",
    ctor: ->(sint) { Webhookdb::Replicator::ConvertkitSubscriberV1.new(sint) },
    feature_roles: [],
    resource_name_singular: "ConvertKit Subscriber",
    supports_webhooks: true,
    supports_backfill: true,
    description: "Replicate ConvertKit subscribers into a database. " \
                 "This is one of the only ways you can keep track of historical subscriber information " \
                 "with ConvertKit.",
    api_docs_url: "https://developers.convertkit.com/#list-subscribers",
  )
end

Instance Method Details

#_create_webhooks ⇒ Object 



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
 
# File 'lib/webhookdb/replicator/convertkit_subscriber_v1.rb', line 33

def _create_webhooks
  # ConvertKit has made several other webhooks available for the subscriber object, but they all have required
  # parameters that are pks of other objects that webhookdb knows nothing about.

  # first verify that the webhooks don't exist
  url = "https://api.convertkit.com/v3/automations/hooks?api_secret=#{self.service_integration.backfill_secret}"
  response = Webhookdb::Http.get(url, logger: self.logger, timeout: Webhookdb::Convertkit.http_timeout)
  # the data returned here is a list of the existing webhooks
  data = response.parsed_response

  # does the "subscriber.subscriber_activate" exist? if not, create it
  # rubocop:disable Style/GuardClause
  if data.present?
    sub_activate_webhook = data.find do |obj|
      obj.dig("rule", "event", "name") == "subscriber_activate"
    end
  end
  unless sub_activate_webhook.present?
    Webhookdb::Http.post(
      "https://api.convertkit.com/v3/automations/hooks",
      {
        "api_secret" => self.service_integration.backfill_secret,
        "target_url" => self.webhook_endpoint,
        "event" => {"name" => "subscriber.subscriber_activate"},
      },
      logger: self.logger,
      timeout: Webhookdb::Convertkit.http_timeout,
    )
    end

  # does the "subscriber.subscriber_activate" exist? if not, create it
  if data.present?
    sub_unsubscribe_webhook = data.find do |obj|
      obj.dig("rule", "event", "name") == "subscriber_activate"
    end
  end
  unless sub_unsubscribe_webhook.present?
    Webhookdb::Http.post(
      "https://api.convertkit.com/v3/automations/hooks",
      {
        "api_secret" => self.service_integration.backfill_secret,
        "target_url" => self.webhook_endpoint,
        "event" => {"name" => "subscriber.subscriber_unsubscribe"},
      },
      logger: self.logger,
      timeout: Webhookdb::Convertkit.http_timeout,
    )
  end
  # rubocop:enable Style/GuardClause
end

#_denormalized_columns ⇒ Object 



119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/webhookdb/replicator/convertkit_subscriber_v1.rb', line 119

def _denormalized_columns
  return [
    Webhookdb::Replicator::Column.new(
      :canceled_at,
      TIMESTAMP,
      index: true,
      optional: true,
      converter: CONV_FIND_CANCELED_AT,
    ),
    Webhookdb::Replicator::Column.new(:created_at, TIMESTAMP, data_key: "created_at", index: true),
    Webhookdb::Replicator::Column.new(:email_address, TEXT, index: true),
    Webhookdb::Replicator::Column.new(:first_name, TEXT),
    Webhookdb::Replicator::Column.new(:last_name, TEXT, data_key: ["fields", "last_name"], optional: true),
    Webhookdb::Replicator::Column.new(:state, TEXT),
  ]
end

#_fetch_backfill_page(pagination_token, last_backfilled:) ⇒ Object 



161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
 
# File 'lib/webhookdb/replicator/convertkit_subscriber_v1.rb', line 161

def _fetch_backfill_page(pagination_token, last_backfilled:)
  pagination_token ||= ["subscribed", 1]
  list_being_iterated, page = pagination_token

  url = "https://api.convertkit.com/v3/subscribers?api_secret=#{self.service_integration.backfill_secret}&page=#{page}&sort_order=desc"
  url += "&updated_from=#{last_backfilled.strftime('%FT%TZ')}" if last_backfilled.present?
  url += "&sort_field=cancelled_at" if list_being_iterated == "cancelled"

  response = Webhookdb::Http.get(url, logger: self.logger, timeout: Webhookdb::Convertkit.http_timeout)
  data = response.parsed_response
  current_page = data["page"]
  total_pages = data["total_pages"]
  subs = data["subscribers"]

  if last_backfilled.present?
    earliest_data_created = subs.empty? ? Time.at(0) : subs[-1].fetch("created_at")
    paged_to_already_seen_records = earliest_data_created < last_backfilled

    if paged_to_already_seen_records && list_being_iterated == "subscribed"
      # If we are done backfilling from the 'subscribed' list, we can now iterate cancelled
      return subs, ["cancelled", 1]
    end
    if paged_to_already_seen_records && list_being_iterated == "cancelled"
      # If we are done backfilling from the 'cancelled' list, we are done backfilling
      return subs, nil
    end
  end

  if current_page < total_pages
    # If we still have pages on this list, go to the next one
    return subs, [list_being_iterated, current_page + 1]
  end
  if list_being_iterated == "subscribed"
    # If we are done with the 'subscribed' list, we can now iterate cancelled
    return subs, ["cancelled", 1]
  end
  # Otherwise, we're at the last page of our canceled subscribers list
  return subs, nil
end

#_resource_and_event(request) ⇒ Object 



140
141
142
143
144
 
# File 'lib/webhookdb/replicator/convertkit_subscriber_v1.rb', line 140

def _resource_and_event(request)
  body = request.body
  return body["subscriber"], body if body.key?("subscriber").present?
  return body, nil
end

#_timestamp_column_name ⇒ Object 



136
137
138
 
# File 'lib/webhookdb/replicator/convertkit_subscriber_v1.rb', line 136

def _timestamp_column_name
  return :created_at
end

#_update_where_expr ⇒ Object 



146
147
148
 
# File 'lib/webhookdb/replicator/convertkit_subscriber_v1.rb', line 146

def _update_where_expr
  return self.qualified_table_sequel_identifier[:data] !~ Sequel[:excluded][:data]
end

#_upsert_update_expr(inserting, **_kwargs) ⇒ Object 



150
151
152
153
154
155
156
157
158
159
 
# File 'lib/webhookdb/replicator/convertkit_subscriber_v1.rb', line 150

def _upsert_update_expr(inserting, **_kwargs)
  update = super
  state = inserting.fetch(:state)
  # If the state is active, we want to use canceled_at:nil unconditionally.
  return update if state == "active"
  # If it's inactive, we only want to update canceled_at if it's not already set
  # (coalesce the existing row's canceled_at with the 'time.now' we are passing in).
  self._coalesce_excluded_on_update(update, [:canceled_at])
  return update
end

#calculate_webhook_state_machine ⇒ Object 



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
 
# File 'lib/webhookdb/replicator/convertkit_subscriber_v1.rb', line 84

def calculate_webhook_state_machine
  step = Webhookdb::Replicator::StateMachineStep.new
  step.output = %(
Great! We've created your ConvertKit Subscribers integration.

ConvertKit supports Subscriber webhooks.
You have two options for hooking them up:

A: Create the webhook yourself, so you don't need to provide us an API Secret.
Run this from a shell:

curl -X POST https://api.convertkit.com/v3/automations/hooks
   -H 'Content-Type: application/json'\\
   -d '{ "api_secret": "<your_secret_api_key>",\\
         "target_url": "#{self._webhook_endpoint}",\\
         "event": { "name": "subscriber.subscriber_activate" } }'
curl -X POST https://api.convertkit.com/v3/automations/hooks
   -H 'Content-Type: application/json'\\
   -d '{ "api_secret": "<your_secret_api_key>",\\
         "target_url": "#{self._webhook_endpoint}",\\
         "event": { "name": "subscriber.subscriber_unsubscribe" } }'

B: Use WebhookDB to backfill historical data with your API Secret, and when we do this,
we'll also set up webhooks for new data.
To start backfilling historical data, run this from a shell:

#{self._backfill_command}

Once you have data (you set up the webhooks, or ran the 'backfill' command),
your database will be populated.
#{self._query_help_output}
    )
  return step.completed
end

#process_state_change(field, value) ⇒ Object 



27
28
29
30
31
 
# File 'lib/webhookdb/replicator/convertkit_subscriber_v1.rb', line 27

def process_state_change(field, value)
  step = super
  self._create_webhooks if field == "backfill_secret"
  return step
end