Refactor pipeline and episode status for use in multiple integrations

app/jobs/publish_feed_job.rb

@@ -12,11 +12,17 @@ def perform(podcast, pub_item)
     # grab the current publishing pipeline.
     return :null if null_publishing_item?(podcast, pub_item)
     return :mismatched if mismatched_publishing_item?(podcast, pub_item)
+
     Rails.logger.info("Starting publishing pipeline via PublishFeedJob", {podcast_id: podcast.id, publishing_queue_item_id: pub_item.id})
 
     PublishingPipelineState.start!(podcast)
-    podcast.feeds.each { |feed| publish_apple(podcast, feed) }
+
+    # Publish each integration for each feed (e.g. apple, megaphone)
+    podcast.feeds.each { |feed| publish_integration(podcast, feed) }
+
+    # After integrations, publish RSS, if appropriate
     podcast.feeds.each { |feed| publish_rss(podcast, feed) }
+
     PublishingPipelineState.complete!(podcast)
   rescue Apple::AssetStateTimeoutError => e
     fail_state(podcast, "apple_timeout", e)
@@ -26,18 +32,17 @@ def perform(podcast, pub_item)
     PublishingPipelineState.settle_remaining!(podcast)
   end
 
-  def publish_apple(podcast, feed)
-    return unless feed.publish_to_apple?
-
-    res = PublishAppleJob.do_perform(podcast.apple_config)
-    PublishingPipelineState.publish_apple!(podcast)
+  def publish_integration(podcast, feed)
+    return unless feed.publish_integration?
+    res = feed.publish_integration!
+    PublishingPipelineState.publish_integration!(podcast)
     res
   rescue => e
-    if podcast.apple_config.sync_blocks_rss
-      fail_state(podcast, "apple", e)
+    if feed.config.sync_blocks_rss
+      fail_state(podcast, feed.integration_type, e)
     else
-      Rails.logger.error("Error publishing to Apple, but continuing to publish RSS", {podcast_id: podcast.id, error: e.message})
-      PublishingPipelineState.error_apple!(podcast)
+      Rails.logger.error("Error publishing to #{feed.integration_type}, but continuing to publish RSS", {podcast_id: podcast.id, error: e.message})
+      PublishingPipelineState.error_integration!(podcast)
     end
   end
 
@@ -51,9 +56,9 @@ def publish_rss(podcast, feed)
 
   def fail_state(podcast, type, error)
     (pipeline_method, log_level) = case type
-    when "apple" then [:error_apple!, :warn]
-    when "rss" then [:error_rss!, :warn]
     when "apple_timeout", "error" then [:error!, :error]
+    when "rss" then [:error_rss!, :warn]
+    else [:error_integration!, :warn]
     end
 
     PublishingPipelineState.public_send(pipeline_method, podcast)

app/models/concerns/apple_delivery.rb

@@ -11,7 +11,6 @@ module AppleDelivery
       class_name: "Apple::PodcastDelivery"
     has_many :apple_podcast_delivery_files, through: :apple_podcast_deliveries, source: :podcast_delivery_files,
       class_name: "Apple::PodcastDeliveryFile"
-    has_many :apple_episode_delivery_statuses, -> { order(created_at: :desc) }, class_name: "Apple::EpisodeDeliveryStatus"
 
     alias_method :podcast_container, :apple_podcast_container
     alias_method :apple_status, :apple_episode_delivery_status
@@ -22,15 +21,19 @@ def publish_to_apple?
   end
 
   def apple_update_delivery_status(attrs)
-    Apple::EpisodeDeliveryStatus.update_status(self, attrs)
+    update_episode_delivery_status(:apple, attrs)
+  end
+
+  def apple_episode_delivery_statuses
+    episode_delivery_statuses.apple
   end
 
   def build_initial_delivery_status
-    Apple::EpisodeDeliveryStatus.default_status(self)
+    Integrations::EpisodeDeliveryStatus.default_status(:apple, self)
   end
 
   def apple_episode_delivery_status
-    apple_episode_delivery_statuses.order(created_at: :desc).first || build_initial_delivery_status
+    episode_delivery_status(:apple) || build_initial_delivery_status
   end
 
   def apple_needs_delivery?

app/models/episode.rb

@@ -8,6 +8,7 @@ class Episode < ApplicationRecord
   include EpisodeFilters
   include EpisodeHasFeeds
   include EpisodeMedia
+  include Integrations::EpisodeIntegrations
   include PublishingStatus
   include TextSanitizer
   include EmbedPlayerHelper
@@ -245,7 +246,7 @@ def copy_media(force = false)
 
   def publish!
     Rails.logger.tagged("Episode#publish!") do
-      apple_mark_for_reupload!
+      feeds.each { |f| f.mark_as_not_delivered!(self) }
       podcast&.publish!
     end
   end

app/models/feed.rb

@@ -70,6 +70,20 @@ def self.enclosure_template_default
     "https://#{ENV["DOVETAIL_HOST"]}{/podcast_id,feed_slug,guid,original_basename}{feed_extension}"
   end
 
+  def mark_as_not_delivered!(episode)
+    # for default / RSS feeds, don't do anything
+    # TODO: we could mark an episode needing to pulished in this RSS feed file
+    #   then later check to see if it is published in the feed yet
+    #   a la "where's my episode?" publish tracking
+  end
+
+  def publish_integration?
+    false
+  end
+
+  def publish_integration!
+  end
+
   def set_defaults
     self.file_name ||= DEFAULT_FILE_NAME
     self.enclosure_template ||= Feed.enclosure_template_default

app/models/feeds/apple_subscription.rb

@@ -25,6 +25,8 @@ class Feeds::AppleSubscription < Feed
   validate :must_be_private
   validate :must_have_token
 
+  alias_method :config, :apple_config
+
   # for soft delete, need a unique slug to be able to make another
   def paranoia_destroy_attributes
     {
@@ -33,6 +35,10 @@ def paranoia_destroy_attributes
     }
   end
 
+  def mark_as_not_delivered!(episode)
+    episode.apple_episode_delivery_status.mark_as_not_delivered!
+  end
+
   def set_defaults
     self.slug ||= DEFAULT_FEED_SLUG
     self.title ||= DEFAULT_TITLE
@@ -113,6 +119,20 @@ def apple?
     true
   end
 
+  def integration_type
+    :apple
+  end
+
+  def publish_integration!
+    if publish_integration?
+      apple_config.build_publisher.publish!
+    end
+  end
+
+  def publish_integration?
+    publish_to_apple?
+  end
+
   def publish_to_apple?
     !!apple_config&.publish_to_apple?
   end

app/models/feeds/megaphone_feed.rb

@@ -1,7 +1,27 @@
 class Feeds::MegaphoneFeed < Feed
   has_one :megaphone_config, class_name: "::Megaphone::Config", inverse_of: :feed
 
+  alias_method :config, :megaphone_config
+
   def self.model_name
     Feed.model_name
   end
+
+  def integration_type
+    :megaphone
+  end
+
+  def publish_integration?
+    megaphone_config&.publish_to_megaphone?
+  end
+
+  def publish_integration!
+    if publish_integration?
+      megaphone_config.build_publisher.publish!
+    end
+  end
+
+  def mark_as_not_delivered!(episode)
+    episode.episode_delivery_statuses.megaphone.first&.mark_as_not_delivered!
+  end
 end

app/models/integrations.rb

@@ -0,0 +1,7 @@
+module Integrations
+  def self.table_name_prefix
+    "integrations_"
+  end
+
+  INTEGRATIONS = %i[apple megaphone]
+end

app/models/integrations/episode_delivery_status.rb

@@ -0,0 +1,46 @@
+module Integrations
+  class EpisodeDeliveryStatus < ApplicationRecord
+    belongs_to :episode, -> { with_deleted }, class_name: "::Episode"
+
+    enum :integration, Integrations::INTEGRATIONS
+
+    def self.update_status(integration, episode, attrs)
+      new_status = (episode.episode_delivery_status(integration)&.dup || default_status(integration, episode))
+      attrs[:integration] = integration
+      new_status.assign_attributes(attrs)
+      new_status.save!
+      episode.episode_delivery_statuses.reset
+      new_status
+    end
+
+    def self.default_status(integration, episode)
+      new(episode: episode, integration: integration)
+    end
+
+    def increment_asset_wait
+      self.class.update_status(integration, episode, asset_processing_attempts: (asset_processing_attempts || 0) + 1)
+    end
+
+    def reset_asset_wait
+      self.class.update_status(integration, episode, asset_processing_attempts: 0)
+    end
+
+    def mark_as_uploaded!
+      self.class.update_status(integration, episode, uploaded: true)
+    end
+
+    def mark_as_not_uploaded!
+      self.class.update_status(integration, episode, uploaded: false)
+    end
+
+    # Whether the media file has been uploaded to the Integration
+    # is a subset of whether the episode has been delivered
+    def mark_as_delivered!
+      self.class.update_status(integration, episode, delivered: true, uploaded: true)
+    end
+
+    def mark_as_not_delivered!
+      self.class.update_status(integration, episode, delivered: false, uploaded: false)
+    end
+  end
+end

app/models/integrations/episode_integrations.rb

@@ -0,0 +1,17 @@
+require "active_support/concern"
+
+module Integrations::EpisodeIntegrations
+  extend ActiveSupport::Concern
+
+  included do
+    has_many :episode_delivery_statuses, -> { order(created_at: :desc) }, class_name: "Integrations::EpisodeDeliveryStatus"
+  end
+
+  def episode_delivery_status(integration)
+    episode_delivery_statuses.order(created_at: :desc).send(integration.intern).first
+  end
+
+  def update_episode_delivery_status(integration, attrs)
+    Integrations::EpisodeDeliveryStatus.update_status(integration, self, attrs)
+  end
+end

app/models/megaphone/config.rb

@@ -6,5 +6,9 @@ class Config < ApplicationRecord
 
     encrypts :token
     encrypts :network_id
+
+    def publish_to_megaphone?
+      valid? && publish_enabled?
+    end
   end
 end

app/models/megaphone/publisher.rb

@@ -0,0 +1,6 @@
+module Megaphone
+  class Publisher < Integrations::Base::Publisher
+    def publish!
+    end
+  end
+end

app/models/publishing_pipeline_state.rb

@@ -48,11 +48,11 @@ class PublishingPipelineState < ApplicationRecord
     :created,
     :started,
     :published_rss,
-    :published_apple,
+    :published_integration,
     :complete,
     :error,
     :expired,
-    :error_apple,
+    :error_integration,
     :error_rss
   ]
 
@@ -146,16 +146,17 @@ def self.publish_rss!(podcast)
     state_transition(podcast, :published_rss)
   end
 
-  def self.publish_apple!(podcast)
-    state_transition(podcast, :published_apple)
+  def self.error_rss!(podcast)
+    state_transition(podcast, :error_rss)
   end
 
-  def self.error_apple!(podcast)
-    state_transition(podcast, :error_apple)
+  # TODO: do something with the integration type?
+  def self.publish_integration!(podcast)
+    state_transition(podcast, :published_integration)
   end
 
-  def self.error_rss!(podcast)
-    state_transition(podcast, :error_rss)
+  def self.error_integration!(podcast)
+    state_transition(podcast, :error_integration)
   end
 
   def self.complete!(podcast)

db/migrate/20231101154016_backfill_media_versions.rb

@@ -2,7 +2,7 @@ class BackfillMediaVersions < ActiveRecord::Migration[7.0]
   def change
     reversible do |dir|
       dir.up do
-        episodes = Episode.where(id: Apple::EpisodeDeliveryStatus.distinct.pluck(:episode_id))
+        episodes = Episode.where(id: Integrations::EpisodeDeliveryStatus.distinct.pluck(:episode_id))
 
         episodes.each do |episode|
           ds = episode.apple_episode_delivery_status