{-# LANGUAGE OverloadedStrings #-}

-- | Versioned, tracked migrations for job queue schemas, run once in order.
-- History lives in a @schema_migrations@ table inside the target schema.
module Arbiter.Migrations
  ( -- * Registry
    QueueSpec (..)
  , Queue

    -- * Configuration
  , MigrationConfig (..)
  , defaultMigrationConfig
  , validateRegistryNames
  , maxQueueNameBytes

    -- * Tracked Migrations
  , runMigrationsForRegistry
  , runMigrationsTrackedForTables
  , jobQueueMigrationsForTable
  , schemaLevelMigrations
  , AdmissionSeeds (..)
  , noAdmissionSeeds
  , TableAdmission (..)
  , allTableAdmission

    -- * Rate-limit reconciliation
  , conflictingPolicyPrefixes

    -- * Re-exports
  , MigrationResult (..)
  ) where

import Arbiter.Core.Concurrency.Schema
  ( addConcurrencyColumnsSQL
  , createConcurrencyIndexSQL
  , createConcurrencyPoliciesTableSQL
  , createConcurrencyTableSQL
  , createConcurrencyTriggerFunctionsSQL
  , createConcurrencyTriggersSQL
  , upsertConcurrencyPolicyRowSQL
  )
import Arbiter.Core.Concurrency.Spec (ConcurrencyPolicy (..), registryConcurrencyPolicies, registryConcurrencyTables)
import Arbiter.Core.CronSchedule
  ( addLastManualRunColumnSQL
  , addQueueNameColumnSQL
  , addRunRequestedColumnSQL
  , addTimezoneColumnSQL
  , createCronSchedulesTableSQL
  )
import Arbiter.Core.Exceptions (displayEx)
import Arbiter.Core.Gates (addGateMetadataColumnSQL, createGatesTableSQL)
import Arbiter.Core.Job.Schema
  ( SchemaName
  , TableName
  , addClaimSeqColumnSQL
  , addKindColumnSQL
  , addTraceContextColumnSQL
  , cancelNotifyChannel
  , createArchiveCompletedAtIndexSQL
  , createArchiveExpiresAtIndexSQL
  , createArchiveGroupKeyIndexSQL
  , createArchiveJobIdIndexSQL
  , createArchiveParentIdIndexSQL
  , createDLQFailedAtIndexSQL
  , createDLQGroupKeyIndexSQL
  , createDLQParentIdIndexSQL
  , createDedupKeyIndexSQL
  , createEventStreamingFunctionSQL
  , createEventStreamingTriggersSQL
  , createJobQueueArchiveTableSQL
  , createJobQueueDLQTableSQL
  , createJobQueueTableSQL
  , createNotifyFunctionSQL
  , createNotifyTriggerSQL
  , createParentIdIndexSQL
  , createResultsTableSQL
  , createSchemaSQL
  , dropEventStreamingFunctionSQL
  , eventStreamingAdoptedObjectComment
  , eventStreamingDLQTriggerName
  , eventStreamingFunctionName
  , eventStreamingObjectComment
  , eventStreamingObjectCommentPrefix
  , eventStreamingTriggerName
  , jobQueueTable
  , legacyEventStreamingTriggers
  , migrateUngroupedReadySplitIndexesSQL
  , notifyAdoptedObjectComment
  , notifyFunctionName
  , notifyObjectComment
  , notifyObjectCommentPrefix
  , notifyTriggerName
  , pauseNotifyChannel
  , queueTableNames
  , setMaxAttemptsDefaultSQL
  )
import Arbiter.Core.Job.Schema.Groups
  ( createGroupsEmptiedIndexSQL
  , createGroupsTableSQL
  , createGroupsTriggerFunctionsSQL
  , createGroupsTriggersSQL
  , createJobQueueGroupInFlightIndexSQL
  , createJobQueueGroupKeyIndexSQL
  , createJobQueueGroupRetriedIndexSQL
  , createJobQueueGroupedDueIndexSQL
  , migrateGroupsReadyRankingSQL
  )
import Arbiter.Core.Job.Types (RegistryAdmissionPolicies)
import Arbiter.Core.QueueRegistry (Queue, QueueSpec (..), RegistryTables (..))
import Arbiter.Core.Queues (createQueuesTableSQL)
import Arbiter.Core.RateLimit.Schema
  ( PolicyRow (..)
  , addRateLimitColumnsSQL
  , addRateLimitCostColumnSQL
  , alterRateLimitsDurabilitySQL
  , arbiterRateLimitsTableName
  , createRateLimitBucketTriggerFunctionsSQL
  , createRateLimitBucketTriggersSQL
  , createRateLimitPoliciesTableSQL
  , createRateLimitsTableSQL
  , createThrottledIndexSQL
  , toPolicyRow
  , upsertPolicyRowSQL
  )
import Arbiter.Core.RateLimit.Spec
  ( Durability (..)
  , registryRateLimitPolicies
  , registryRateLimitTables
  )
import Arbiter.Core.SchemaTables (sharedArbiterTables)
import Arbiter.Core.Worker
  ( addArchiveForColumnSQL
  , addCancelRequestedAtColumnSQL
  , addClaimedByColumnSQL
  , createWorkersTableSQL
  )
import Control.Exception (SomeAsyncException, SomeException, bracket, fromException, throwIO, try)
import Control.Monad (unless, void, when)
import Data.ByteString (ByteString)
import Data.ByteString qualified as BS
import Data.Foldable (find, traverse_)
import Data.Map.Strict qualified as Map
import Data.Maybe (isJust, listToMaybe)
import Data.Proxy (Proxy (..))
import Data.Set qualified as Set
import Data.Text (Text)
import Data.Text qualified as T
import Data.Text.Encoding (decodeUtf8, encodeUtf8)
import Data.Time (NominalDiffTime)
import Database.PostgreSQL.LibPQ qualified as LibPQ
import Database.PostgreSQL.Simple (Only (..), close, connectPostgreSQL, execute_, query)
import Database.PostgreSQL.Simple qualified as PG
import Database.PostgreSQL.Simple.Internal (withConnection)
import Database.PostgreSQL.Simple.Migration
  ( MigrationCommand (..)
  , MigrationOptions (..)
  , MigrationResult (..)
  , Verbosity (..)
  , defaultOptions
  , runMigrations
  )
import Database.PostgreSQL.Simple.Types (PGArray (..), Query (..))

-- | The desired state reconciled after a schema's tracked migrations run.
data MigrationConfig = MigrationConfig
  { MigrationConfig -> Bool
enableNotifications :: Bool
  -- ^ Whether LISTEN/NOTIFY triggers for reactive job claiming should be
  -- installed. Re-running migrations reconciles existing schemas in either
  -- direction. Default: 'True'.
  , MigrationConfig -> Bool
enableEventStreaming :: Bool
  -- ^ Whether event-streaming triggers for the admin UI should be installed.
  -- When enabled, every INSERT\/UPDATE\/DELETE on job tables fires an enriched
  -- JSON event via @pg_notify@. Re-running migrations with this disabled drops
  -- the triggers and shared function. Default: 'False'.
  , MigrationConfig -> Durability
rateLimitDurability :: Durability
  -- ^ WAL-logging for the rate-limit bucket table in this schema. 'Unlogged'
  -- (default) resets buckets on crash\/failover. 'Durable' preserves them at a
  -- throughput cost.
  , MigrationConfig -> Maybe NominalDiffTime
migrationLockTimeout :: Maybe NominalDiffTime
  -- ^ How many seconds to wait for the schema's migration lock. The lock serializes
  -- replicas that migrate at the same time. 'Nothing' (default) waits indefinitely.
  }
  deriving stock (MigrationConfig -> MigrationConfig -> Bool
(MigrationConfig -> MigrationConfig -> Bool)
-> (MigrationConfig -> MigrationConfig -> Bool)
-> Eq MigrationConfig
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: MigrationConfig -> MigrationConfig -> Bool
== :: MigrationConfig -> MigrationConfig -> Bool
$c/= :: MigrationConfig -> MigrationConfig -> Bool
/= :: MigrationConfig -> MigrationConfig -> Bool
Eq, Int -> MigrationConfig -> ShowS
[MigrationConfig] -> ShowS
MigrationConfig -> String
(Int -> MigrationConfig -> ShowS)
-> (MigrationConfig -> String)
-> ([MigrationConfig] -> ShowS)
-> Show MigrationConfig
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> MigrationConfig -> ShowS
showsPrec :: Int -> MigrationConfig -> ShowS
$cshow :: MigrationConfig -> String
show :: MigrationConfig -> String
$cshowList :: [MigrationConfig] -> ShowS
showList :: [MigrationConfig] -> ShowS
Show)

-- | Notify triggers on, event streaming off, unlogged rate-limit buckets.
defaultMigrationConfig :: MigrationConfig
defaultMigrationConfig :: MigrationConfig
defaultMigrationConfig =
  MigrationConfig
    { enableNotifications :: Bool
enableNotifications = Bool
True
    , enableEventStreaming :: Bool
enableEventStreaming = Bool
False
    , rateLimitDurability :: Durability
rateLimitDurability = Durability
Unlogged
    , migrationLockTimeout :: Maybe NominalDiffTime
migrationLockTimeout = Maybe NominalDiffTime
forall a. Maybe a
Nothing
    }

-- | Migrate every queue in a registry into one schema. The schema itself is created
-- first, outside migration tracking.
--
-- @
-- type AppRegistry =
--   '[ Queue "email_jobs" EmailPayload
--    , Queue "order_jobs" OrderPayload
--    ]
--
-- main :: IO ()
-- main = do
--   result <- runMigrationsForRegistry
--               (Proxy @AppRegistry)
--               "host=localhost dbname=mydb"
--               "arbiter"
--               defaultMigrationConfig
-- @
runMigrationsForRegistry
  :: forall registry
   . ( RegistryAdmissionPolicies registry
     , RegistryTables registry
     )
  => Proxy registry
  -- ^ Proxy for the job payload registry
  -> ByteString
  -- ^ Database connection string
  -> SchemaName
  -- ^ Schema name
  -> MigrationConfig
  -- ^ Migration configuration
  -> IO (MigrationResult String)
  -- ^ Migration results
runMigrationsForRegistry :: forall (registry :: JobPayloadRegistry).
(RegistryAdmissionPolicies registry, RegistryTables registry) =>
Proxy registry
-> ByteString
-> Text
-> MigrationConfig
-> IO (MigrationResult String)
runMigrationsForRegistry Proxy registry
proxy ByteString
connStr Text
schemaName MigrationConfig
config = do
  let ccTables :: Map Text Bool
ccTables = [(Text, Bool)] -> Map Text Bool
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList (forall (registry :: JobPayloadRegistry).
RegistryConcurrencyPolicies registry =>
[(Text, Bool)]
registryConcurrencyTables @registry)
      rlTables :: Map Text Bool
rlTables = [(Text, Bool)] -> Map Text Bool
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList (forall (registry :: JobPayloadRegistry).
RegistryRateLimitPolicies registry =>
[(Text, Bool)]
registryRateLimitTables @registry)
      admissionFor :: Text -> TableAdmission
admissionFor Text
table =
        TableAdmission
          { tableConcurrency :: Bool
tableConcurrency = Bool -> Text -> Map Text Bool -> Bool
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault Bool
False Text
table Map Text Bool
ccTables
          , tableRateLimit :: Bool
tableRateLimit = Bool -> Text -> Map Text Bool -> Bool
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault Bool
False Text
table Map Text Bool
rlTables
          }
      tables :: [(Text, TableAdmission)]
tables = [(Text
table, Text -> TableAdmission
admissionFor Text
table) | Text
table <- Proxy registry -> [Text]
forall (registry :: JobPayloadRegistry).
RegistryTables registry =>
Proxy registry -> [Text]
registryTableNames Proxy registry
proxy]
      -- Policies are collected from each payload's 'rateLimitFor' selector.
      seeds :: AdmissionSeeds
seeds =
        AdmissionSeeds
          { seedRateLimitPolicies :: [PolicyRow]
seedRateLimitPolicies = (Policy -> PolicyRow) -> [Policy] -> [PolicyRow]
forall a b. (a -> b) -> [a] -> [b]
map Policy -> PolicyRow
toPolicyRow (Set Policy -> [Policy]
forall a. Set a -> [a]
Set.toList (forall (registry :: JobPayloadRegistry).
RegistryRateLimitPolicies registry =>
Set Policy
registryRateLimitPolicies @registry))
          , seedConcurrencyPolicies :: [ConcurrencyPolicy]
seedConcurrencyPolicies = Set ConcurrencyPolicy -> [ConcurrencyPolicy]
forall a. Set a -> [a]
Set.toList (forall (registry :: JobPayloadRegistry).
RegistryConcurrencyPolicies registry =>
Set ConcurrencyPolicy
registryConcurrencyPolicies @registry)
          , seedDurability :: Durability
seedDurability = MigrationConfig -> Durability
rateLimitDurability MigrationConfig
config
          }
  ByteString
-> Text
-> [(Text, TableAdmission)]
-> MigrationConfig
-> AdmissionSeeds
-> IO (MigrationResult String)
runMigrationsTrackedForTables ByteString
connStr Text
schemaName [(Text, TableAdmission)]
tables MigrationConfig
config AdmissionSeeds
seeds

-- | Admission policy rows to seed after a successful migration.
data AdmissionSeeds = AdmissionSeeds
  { AdmissionSeeds -> [PolicyRow]
seedRateLimitPolicies :: [PolicyRow]
  , AdmissionSeeds -> [ConcurrencyPolicy]
seedConcurrencyPolicies :: [ConcurrencyPolicy]
  , AdmissionSeeds -> Durability
seedDurability :: Durability
  }

-- | Seeds for a deployment with no admission policies.
noAdmissionSeeds :: AdmissionSeeds
noAdmissionSeeds :: AdmissionSeeds
noAdmissionSeeds = [PolicyRow] -> [ConcurrencyPolicy] -> Durability -> AdmissionSeeds
AdmissionSeeds [] [] Durability
Unlogged

-- | Which admission trigger kinds a table's payload declares. Trigger migrations
-- are install-only. A kind removed from a payload keeps its triggers.
data TableAdmission = TableAdmission
  { TableAdmission -> Bool
tableConcurrency :: Bool
  , TableAdmission -> Bool
tableRateLimit :: Bool
  }
  deriving stock (TableAdmission -> TableAdmission -> Bool
(TableAdmission -> TableAdmission -> Bool)
-> (TableAdmission -> TableAdmission -> Bool) -> Eq TableAdmission
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: TableAdmission -> TableAdmission -> Bool
== :: TableAdmission -> TableAdmission -> Bool
$c/= :: TableAdmission -> TableAdmission -> Bool
/= :: TableAdmission -> TableAdmission -> Bool
Eq, Int -> TableAdmission -> ShowS
[TableAdmission] -> ShowS
TableAdmission -> String
(Int -> TableAdmission -> ShowS)
-> (TableAdmission -> String)
-> ([TableAdmission] -> ShowS)
-> Show TableAdmission
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> TableAdmission -> ShowS
showsPrec :: Int -> TableAdmission -> ShowS
$cshow :: TableAdmission -> String
show :: TableAdmission -> String
$cshowList :: [TableAdmission] -> ShowS
showList :: [TableAdmission] -> ShowS
Show)

-- | Install every admission trigger kind.
allTableAdmission :: TableAdmission
allTableAdmission :: TableAdmission
allTableAdmission = Bool -> Bool -> TableAdmission
TableAdmission Bool
True Bool
True

-- | The longest queue name whose generated identifiers survive PostgreSQL's 63-byte
-- truncation distinct. Derived by rendering a probe queue's own DDL at each length.
maxQueueNameBytes :: Int
maxQueueNameBytes :: Int
maxQueueNameBytes = [Int] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length ((Int -> Bool) -> [Int] -> [Int]
forall a. (a -> Bool) -> [a] -> [a]
takeWhile Int -> Bool
identifiersDistinct [Int
1 .. Int
63])
  where
    identifiersDistinct :: Int -> Bool
identifiersDistinct Int
len = [Text] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null ((Text -> Text) -> (Text -> Text) -> [Text] -> [Text]
forall params row.
Ord params =>
(row -> Text) -> (row -> params) -> [row] -> [Text]
conflictingPrefixes (Int -> Text -> Text
T.take Int
63) Text -> Text
forall a. a -> a
id (Text -> [Text]
renderedIdentifiers (Int -> Text -> Text
T.replicate Int
len Text
"q")))

-- | All identifiers in a queue's rendered DDL.
renderedIdentifiers :: TableName -> [Text]
renderedIdentifiers :: Text -> [Text]
renderedIdentifiers Text
table =
  (Text -> [Text]) -> [Text] -> [Text]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap Text -> [Text]
quoted ([Text] -> [Text]) -> [Text] -> [Text]
forall a b. (a -> b) -> a -> b
$
    [ByteString -> Text
decodeUtf8 ByteString
body | MigrationScript String
_ ByteString
body <- Text -> Text -> TableAdmission -> [MigrationCommand]
jobQueueMigrationsForTable Text
probeSchema Text
table TableAdmission
allTableAdmission]
      [Text] -> [Text] -> [Text]
forall a. Semigroup a => a -> a -> a
<> [ Text -> Text -> Text
createNotifyFunctionSQL Text
probeSchema Text
table
         , Text -> Text -> Text
createNotifyTriggerSQL Text
probeSchema Text
table
         , Text -> Text -> Text
createEventStreamingTriggersSQL Text
probeSchema Text
table
         ]
  where
    probeSchema :: Text
probeSchema = Text
"arbiter"
    quoted :: Text -> [Text]
quoted = [Text] -> [Text]
forall {a}. [a] -> [a]
everyOther ([Text] -> [Text]) -> (Text -> [Text]) -> Text -> [Text]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. HasCallStack => Text -> Text -> [Text]
Text -> Text -> [Text]
T.splitOn Text
"\""
    everyOther :: [a] -> [a]
everyOther (a
_ : a
name : [a]
rest) = a
name a -> [a] -> [a]
forall a. a -> [a] -> [a]
: [a] -> [a]
everyOther [a]
rest
    everyOther [a]
_ = []

-- | Reject queue names that generate a schema-wide arbiter table, or that generate
-- object or channel names PostgreSQL truncates into each other. The length limit is
-- where two of a queue's generated names collide.
validateRegistryNames :: SchemaName -> [TableName] -> Either Text ()
validateRegistryNames :: Text -> [Text] -> Either Text ()
validateRegistryNames Text
schemaName [Text]
tables
  | Text -> Bool
T.null Text
schemaName = Text -> Either Text ()
forall a b. a -> Either a b
Left Text
"Arbiter schema name must not be empty"
  | Text -> Int
byteLength Text
schemaName Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
63 = Text -> Either Text ()
forall a b. a -> Either a b
Left Text
"Arbiter schema name exceeds PostgreSQL's 63-byte identifier limit"
  | (Text -> Bool) -> [Text] -> Bool
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Bool
any Text -> Bool
T.null [Text]
tables = Text -> Either Text ()
forall a b. a -> Either a b
Left Text
"Arbiter queue name must not be empty"
  | Just Text
table <- (Text -> Bool) -> [Text] -> Maybe Text
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Maybe a
find ((Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
maxQueueNameBytes) (Int -> Bool) -> (Text -> Int) -> Text -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> Int
byteLength) [Text]
tables =
      Text -> Either Text ()
forall a b. a -> Either a b
Left
        ( Text
"Arbiter queue name exceeds the "
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int -> String
forall a. Show a => a -> String
show Int
maxQueueNameBytes)
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"-byte generated-identifier limit: "
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
table
        )
  | Just Text
reserved <- Maybe Text
reservedCollision =
      Text -> Either Text ()
forall a b. a -> Either a b
Left (Text
"Arbiter queue name generates a reserved arbiter table: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
reserved)
  | Just Text
generated <- Maybe Text
generatedCollision =
      Text -> Either Text ()
forall a b. a -> Either a b
Left (Text
"Arbiter queue names generate the same PostgreSQL object: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
generated)
  | Just Text
channel <- Maybe Text
channelCollision =
      Text -> Either Text ()
forall a b. a -> Either a b
Left (Text
"Arbiter queue names generate the same notification channel: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
channel)
  | Bool
otherwise = () -> Either Text ()
forall a b. b -> Either a b
Right ()
  where
    byteLength :: Text -> Int
byteLength = ByteString -> Int
BS.length (ByteString -> Int) -> (Text -> ByteString) -> Text -> Int
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> ByteString
encodeUtf8
    generatedCollision :: Maybe Text
generatedCollision =
      [(Text, Text)] -> Maybe Text
sharedName [(Text
generated, Text
table) | Text
table <- [Text]
tables, Text
generated <- Text -> [Text]
queueTableNames Text
table]
    reservedCollision :: Maybe Text
reservedCollision =
      (Text -> Bool) -> [Text] -> Maybe Text
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Maybe a
find (Text -> Set Text -> Bool
forall a. Ord a => a -> Set a -> Bool
`Set.member` [Text] -> Set Text
forall a. Ord a => [a] -> Set a
Set.fromList ((Text -> [Text]) -> [Text] -> [Text]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap Text -> [Text]
queueTableNames [Text]
tables)) [Text]
sharedArbiterTables
    channelCollision :: Maybe Text
channelCollision =
      [(Text, Text)] -> Maybe Text
sharedName
        [ (Text -> Text -> Text
channel Text
schemaName Text
table, Text
table)
        | Text
table <- [Text]
tables
        , Text -> Text -> Text
channel <- [Text -> Text -> Text
pauseNotifyChannel, Text -> Text -> Text
cancelNotifyChannel]
        ]
    -- The first generated name more than one queue claims.
    sharedName :: [(Text, Text)] -> Maybe Text
sharedName = [Text] -> Maybe Text
forall a. [a] -> Maybe a
listToMaybe ([Text] -> Maybe Text)
-> ([(Text, Text)] -> [Text]) -> [(Text, Text)] -> Maybe Text
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ((Text, Text) -> Text)
-> ((Text, Text) -> Text) -> [(Text, Text)] -> [Text]
forall params row.
Ord params =>
(row -> Text) -> (row -> params) -> [row] -> [Text]
conflictingPrefixes (Text, Text) -> Text
forall a b. (a, b) -> a
fst (Text, Text) -> Text
forall a b. (a, b) -> b
snd

-- | Run migrations for multiple tables within a single schema, seeding the given
-- rate-limit policies. On migration success, reconciles the policy and bucket
-- tables on the same connection. The table list must be the schema's whole queue
-- set. Reconciliation treats an omitted queue as removed and drops its notify and
-- event-streaming objects.
runMigrationsTrackedForTables
  :: ByteString
  -> SchemaName
  -> [(TableName, TableAdmission)]
  -> MigrationConfig
  -> AdmissionSeeds
  -> IO (MigrationResult String)
runMigrationsTrackedForTables :: ByteString
-> Text
-> [(Text, TableAdmission)]
-> MigrationConfig
-> AdmissionSeeds
-> IO (MigrationResult String)
runMigrationsTrackedForTables ByteString
connStr Text
schemaName [(Text, TableAdmission)]
tableNames MigrationConfig
config AdmissionSeeds
seeds =
  case Text -> [Text] -> Either Text ()
validateRegistryNames Text
schemaName (((Text, TableAdmission) -> Text)
-> [(Text, TableAdmission)] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map (Text, TableAdmission) -> Text
forall a b. (a, b) -> a
fst [(Text, TableAdmission)]
tableNames) of
    Left Text
err -> MigrationResult String -> IO (MigrationResult String)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (String -> MigrationResult String
forall a. a -> MigrationResult a
MigrationError (Text -> String
T.unpack Text
err))
    Right () -> IO Connection
-> (Connection -> IO ())
-> (Connection -> IO (MigrationResult String))
-> IO (MigrationResult String)
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
bracket (ByteString -> IO Connection
connectPostgreSQL ByteString
connStr) Connection -> IO ()
close ((Connection -> IO (MigrationResult String))
 -> IO (MigrationResult String))
-> (Connection -> IO (MigrationResult String))
-> IO (MigrationResult String)
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
      Connection
-> Text
-> Maybe NominalDiffTime
-> IO (MigrationResult String)
-> IO (MigrationResult String)
withMigrationLock Connection
conn Text
schemaName (MigrationConfig -> Maybe NominalDiffTime
migrationLockTimeout MigrationConfig
config) (IO (MigrationResult String) -> IO (MigrationResult String))
-> IO (MigrationResult String) -> IO (MigrationResult String)
forall a b. (a -> b) -> a -> b
$
        Connection
-> Text
-> [(Text, TableAdmission)]
-> MigrationConfig
-> AdmissionSeeds
-> IO (MigrationResult String)
migrateSchema Connection
conn Text
schemaName [(Text, TableAdmission)]
tableNames MigrationConfig
config AdmissionSeeds
seeds

-- | Hold the schema's migration lock for the whole session. Replicas that migrate at
-- the same time run one after another. A session lock spans the reconciles, which run
-- after the tracked migrations commit.
withMigrationLock
  :: PG.Connection
  -> SchemaName
  -> Maybe NominalDiffTime
  -> IO (MigrationResult String)
  -> IO (MigrationResult String)
withMigrationLock :: Connection
-> Text
-> Maybe NominalDiffTime
-> IO (MigrationResult String)
-> IO (MigrationResult String)
withMigrationLock Connection
conn Text
schemaName Maybe NominalDiffTime
timeout IO (MigrationResult String)
act = IO Bool
-> (Bool -> IO ())
-> (Bool -> IO (MigrationResult String))
-> IO (MigrationResult String)
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
bracket IO Bool
acquire Bool -> IO ()
release Bool -> IO (MigrationResult String)
run
  where
    lockName :: Only Text
lockName = Text -> Only Text
forall a. a -> Only a
Only (Text
"arbiter.migrations:" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
schemaName)
    acquire :: IO Bool
acquire =
      IO () -> IO (Either SqlError ())
forall e a. Exception e => IO a -> IO (Either e a)
try (Connection -> IO () -> IO ()
forall a. Connection -> IO a -> IO a
PG.withTransaction Connection
conn (IO ()
setTimeouts IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO ()
lock)) IO (Either SqlError ())
-> (Either SqlError () -> IO Bool) -> IO Bool
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        Right () -> Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True
        Left SqlError
sqlError
          | Maybe NominalDiffTime -> Bool
forall a. Maybe a -> Bool
isJust Maybe NominalDiffTime
timeout, SqlError -> ByteString
PG.sqlState SqlError
sqlError ByteString -> ByteString -> Bool
forall a. Eq a => a -> a -> Bool
== ByteString
"55P03" -> Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
          | Bool
otherwise -> SqlError -> IO Bool
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO SqlError
sqlError
    lock :: IO ()
lock = IO [Only (Maybe Text)] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (Connection -> Query -> Only Text -> IO [Only (Maybe Text)]
forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query Connection
conn Query
"SELECT pg_advisory_lock(hashtextextended(?, 0))::text" Only Text
lockName :: IO [Only (Maybe Text)])
    -- Pinned for this transaction.
    setTimeouts :: IO ()
setTimeouts = ((Text, Text) -> IO ()) -> [(Text, Text)] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Text, Text) -> IO ()
setLocal [(Text
"lock_timeout", Text -> (NominalDiffTime -> Text) -> Maybe NominalDiffTime -> Text
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Text
"0" NominalDiffTime -> Text
millis Maybe NominalDiffTime
timeout), (Text
"statement_timeout", Text
"0")]
    setLocal :: (Text, Text) -> IO ()
setLocal (Text
name, Text
value) =
      IO [Only Text] -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (Connection -> Query -> (Text, Text) -> IO [Only Text]
forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query Connection
conn Query
"SELECT set_config(?, ?, TRUE)" (Text
name :: Text, Text
value :: Text) :: IO [Only Text])
    millis :: NominalDiffTime -> Text
millis NominalDiffTime
seconds = String -> Text
T.pack (Int -> String
forall a. Show a => a -> String
show (Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
1 (Int -> Int -> Int
forall a. Ord a => a -> a -> a
min Int
maxLockTimeoutMillis (NominalDiffTime -> Int
forall b. Integral b => NominalDiffTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
ceiling (NominalDiffTime
seconds NominalDiffTime -> NominalDiffTime -> NominalDiffTime
forall a. Num a => a -> a -> a
* NominalDiffTime
1000) :: Int))))
    maxLockTimeoutMillis :: Int
maxLockTimeoutMillis = Int
2147483647
    release :: Bool -> IO ()
release Bool
locked =
      Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
locked (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        IO (Either SqlError [Only Bool]) -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO [Only Bool] -> IO (Either SqlError [Only Bool])
forall e a. Exception e => IO a -> IO (Either e a)
try IO [Only Bool]
unlock :: IO (Either PG.SqlError [Only Bool]))
    unlock :: IO [Only Bool]
unlock = Connection -> Query -> Only Text -> IO [Only Bool]
forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query Connection
conn Query
"SELECT pg_advisory_unlock(hashtextextended(?, 0))" Only Text
lockName :: IO [Only Bool]
    run :: Bool -> IO (MigrationResult String)
run Bool
locked
      | Bool
locked = IO (MigrationResult String)
act
      | Bool
otherwise =
          MigrationResult String -> IO (MigrationResult String)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (String -> MigrationResult String
forall a. a -> MigrationResult a
MigrationError (String
"Timed out waiting on the arbiter migration lock for schema " String -> ShowS
forall a. Semigroup a => a -> a -> a
<> Text -> String
T.unpack Text
schemaName))

-- | Apply a schema's tracked migrations and reconcile it, under the migration lock.
migrateSchema
  :: PG.Connection
  -> SchemaName
  -> [(TableName, TableAdmission)]
  -> MigrationConfig
  -> AdmissionSeeds
  -> IO (MigrationResult String)
migrateSchema :: Connection
-> Text
-> [(Text, TableAdmission)]
-> MigrationConfig
-> AdmissionSeeds
-> IO (MigrationResult String)
migrateSchema Connection
conn Text
schemaName [(Text, TableAdmission)]
tableNames MigrationConfig
config (AdmissionSeeds [PolicyRow]
policyRows [ConcurrencyPolicy]
concRows Durability
durability) = do
  Connection -> (Connection -> IO ()) -> IO ()
forall a. Connection -> (Connection -> IO a) -> IO a
withConnection Connection
conn ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
libpqConn ->
    Connection -> IO ()
LibPQ.disableNoticeReporting Connection
libpqConn

  -- A CREATE that fails for want of privilege is fine when the schema exists.
  let schemaSQL :: Query
schemaSQL = ByteString -> Query
Query (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
createSchemaSQL Text
schemaName)
  result <- IO Int64 -> IO (Either SqlError Int64)
forall e a. Exception e => IO a -> IO (Either e a)
try (IO Int64 -> IO (Either SqlError Int64))
-> IO Int64 -> IO (Either SqlError Int64)
forall a b. (a -> b) -> a -> b
$ Connection -> Query -> IO Int64
execute_ Connection
conn Query
schemaSQL
  case result of
    Right Int64
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
    Left (SqlError
createError :: PG.SqlError) -> do
      exists <- Connection -> Text -> IO Bool
schemaExists Connection
conn Text
schemaName
      unless exists $
        ioError
          ( userError $
              "Failed to create schema "
                <> T.unpack schemaName
                <> " and it does not exist. Either grant CREATE privilege on the database"
                <> " or create the schema manually: CREATE SCHEMA "
                <> T.unpack schemaName
                <> ";"
                <> "\nOriginal error: "
                <> show createError
          )

  withConnection conn $ \Connection
libpqConn ->
    Connection -> IO ()
LibPQ.enableNoticeReporting Connection
libpqConn

  let schemaMigrations = Text -> [MigrationCommand]
schemaLevelMigrations Text
schemaName
      tableMigrations = ((Text, TableAdmission) -> [MigrationCommand])
-> [(Text, TableAdmission)] -> [MigrationCommand]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap ((Text -> TableAdmission -> [MigrationCommand])
-> (Text, TableAdmission) -> [MigrationCommand]
forall a b c. (a -> b -> c) -> (a, b) -> c
uncurry (Text -> Text -> TableAdmission -> [MigrationCommand]
jobQueueMigrationsForTable Text
schemaName)) [(Text, TableAdmission)]
tableNames
      migrations = [MigrationCommand]
schemaMigrations [MigrationCommand] -> [MigrationCommand] -> [MigrationCommand]
forall a. Semigroup a => a -> a -> a
<> [MigrationCommand]
tableMigrations
      migrationTableName = Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text
schemaName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
".schema_migrations"
      options =
        MigrationOptions
defaultOptions
          { optVerbose = Quiet
          , optTableName = migrationTableName
          }

  _ <- runMigrations conn options [MigrationInitialization]
  migrationResult <- runMigrations conn options migrations
  case migrationResult of
    MigrationResult String
MigrationSuccess -> do
      reconciled <-
        IO () -> IO (Either SomeException ())
forall e a. Exception e => IO a -> IO (Either e a)
try (IO () -> IO (Either SomeException ()))
-> IO () -> IO (Either SomeException ())
forall a b. (a -> b) -> a -> b
$ do
          Connection -> Text -> [PolicyRow] -> IO ()
reconcileRateLimitPolicies Connection
conn Text
schemaName [PolicyRow]
policyRows
          Connection -> Text -> [ConcurrencyPolicy] -> IO ()
reconcileConcurrencyPolicies Connection
conn Text
schemaName [ConcurrencyPolicy]
concRows
          Connection -> Text -> Durability -> IO ()
reconcileRateLimitDurability Connection
conn Text
schemaName Durability
durability
          Connection -> Text -> [Text] -> MigrationConfig -> IO ()
reconcileOptionalTriggers Connection
conn Text
schemaName (((Text, TableAdmission) -> Text)
-> [(Text, TableAdmission)] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map (Text, TableAdmission) -> Text
forall a b. (a, b) -> a
fst [(Text, TableAdmission)]
tableNames) MigrationConfig
config
      case reconciled of
        Right () -> MigrationResult String -> IO (MigrationResult String)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure MigrationResult String
forall a. MigrationResult a
MigrationSuccess
        -- A reconcile throws on a conflicting prefix or a failed ALTER. Async exceptions propagate.
        Left (SomeException
exception :: SomeException)
          | Just (SomeAsyncException
_ :: SomeAsyncException) <- SomeException -> Maybe SomeAsyncException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
exception -> SomeException -> IO (MigrationResult String)
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO SomeException
exception
          | Bool
otherwise -> MigrationResult String -> IO (MigrationResult String)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (String -> MigrationResult String
forall a. a -> MigrationResult a
MigrationError (Text -> String
T.unpack (SomeException -> Text
displayEx SomeException
exception)))
    MigrationResult String
other -> MigrationResult String -> IO (MigrationResult String)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure MigrationResult String
other

-- | Upsert each policy's @default_*@ params into the policies table. Operator
-- @override_*@ values stay intact. Idempotent. Overrides and removed prefixes survive
-- a deploy.
reconcileRateLimitPolicies :: PG.Connection -> SchemaName -> [PolicyRow] -> IO ()
reconcileRateLimitPolicies :: Connection -> Text -> [PolicyRow] -> IO ()
reconcileRateLimitPolicies =
  String
-> String
-> (PolicyRow -> Text)
-> (PolicyRow -> String)
-> (Text -> PolicyRow -> Text)
-> Connection
-> Text
-> [PolicyRow]
-> IO ()
forall params row.
Ord params =>
String
-> String
-> (row -> Text)
-> (row -> params)
-> (Text -> row -> Text)
-> Connection
-> Text
-> [row]
-> IO ()
reconcilePolicyRows String
"rate-limit policy" String
"parameters" PolicyRow -> Text
prefixId PolicyRow -> String
policyParamsKey Text -> PolicyRow -> Text
upsertPolicyRowSQL

-- | 'conflictingPrefixes' specialized to policy rows.
conflictingPolicyPrefixes :: [PolicyRow] -> [Text]
conflictingPolicyPrefixes :: [PolicyRow] -> [Text]
conflictingPolicyPrefixes = (PolicyRow -> Text)
-> (PolicyRow -> String) -> [PolicyRow] -> [Text]
forall params row.
Ord params =>
(row -> Text) -> (row -> params) -> [row] -> [Text]
conflictingPrefixes PolicyRow -> Text
prefixId PolicyRow -> String
policyParamsKey

-- | A canonical conflict key for a policy's params, rendered as text. A NaN compares
-- equal to itself.
policyParamsKey :: PolicyRow -> String
policyParamsKey :: PolicyRow -> String
policyParamsKey PolicyRow
row = (Double, Double, Double) -> String
forall a. Show a => a -> String
show (PolicyRow -> Double
maxTokens PolicyRow
row, PolicyRow -> Double
refillAmt PolicyRow
row, PolicyRow -> Double
interval PolicyRow
row)

-- | Upsert each pool's @default_limit@, leaving operator overrides intact. Two pools
-- with the same prefix but different limits fail the migration.
reconcileConcurrencyPolicies :: PG.Connection -> SchemaName -> [ConcurrencyPolicy] -> IO ()
reconcileConcurrencyPolicies :: Connection -> Text -> [ConcurrencyPolicy] -> IO ()
reconcileConcurrencyPolicies =
  String
-> String
-> (ConcurrencyPolicy -> Text)
-> (ConcurrencyPolicy -> Int32)
-> (Text -> ConcurrencyPolicy -> Text)
-> Connection
-> Text
-> [ConcurrencyPolicy]
-> IO ()
forall params row.
Ord params =>
String
-> String
-> (row -> Text)
-> (row -> params)
-> (Text -> row -> Text)
-> Connection
-> Text
-> [row]
-> IO ()
reconcilePolicyRows String
"concurrency pool" String
"limits" ConcurrencyPolicy -> Text
cpPrefix ConcurrencyPolicy -> Int32
cpLimit Text -> ConcurrencyPolicy -> Text
upsertConcurrencyPolicyRowSQL

-- | Upsert each policy row's @default_*@ params. Fails on a prefix that contains the
-- @:@ key separator or is declared with conflicting params. Idempotent. Operator
-- overrides and removed prefixes survive a deploy.
reconcilePolicyRows
  :: (Ord params)
  => String
  -> String
  -> (row -> Text)
  -> (row -> params)
  -> (SchemaName -> row -> Text)
  -> PG.Connection
  -> SchemaName
  -> [row]
  -> IO ()
reconcilePolicyRows :: forall params row.
Ord params =>
String
-> String
-> (row -> Text)
-> (row -> params)
-> (Text -> row -> Text)
-> Connection
-> Text
-> [row]
-> IO ()
reconcilePolicyRows String
label String
noun row -> Text
prefixOf row -> params
paramsOf Text -> row -> Text
upsertSQL Connection
conn Text
schemaName [row]
rows
  | Just Text
prefix <- (Text -> Bool) -> [Text] -> Maybe Text
forall (t :: * -> *) a. Foldable t => (a -> Bool) -> t a -> Maybe a
find (Text -> Text -> Bool
T.isInfixOf Text
":") ((row -> Text) -> [row] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map row -> Text
prefixOf [row]
rows) =
      IOError -> IO ()
forall a. HasCallStack => IOError -> IO a
ioError
        ( String -> IOError
userError (String -> IOError) -> String -> IOError
forall a b. (a -> b) -> a -> b
$
            String
label
              String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
" prefix "
              String -> ShowS
forall a. Semigroup a => a -> a -> a
<> Text -> String
T.unpack Text
prefix
              String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
" contains ':', the key separator, so its keys would alias another prefix's. Use a colon-free prefix."
        )
  | (Text
prefix : [Text]
_) <- (row -> Text) -> (row -> params) -> [row] -> [Text]
forall params row.
Ord params =>
(row -> Text) -> (row -> params) -> [row] -> [Text]
conflictingPrefixes row -> Text
prefixOf row -> params
paramsOf [row]
rows =
      IOError -> IO ()
forall a. HasCallStack => IOError -> IO a
ioError
        ( String -> IOError
userError (String -> IOError) -> String -> IOError
forall a b. (a -> b) -> a -> b
$
            String
label String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
" prefix " String -> ShowS
forall a. Semigroup a => a -> a -> a
<> Text -> String
T.unpack Text
prefix String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
" is declared with conflicting " String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
noun String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
". Give each a unique prefix."
        )
  | Bool
otherwise =
      Connection -> IO () -> IO ()
forall a. Connection -> IO a -> IO a
PG.withTransaction Connection
conn (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
        (row -> IO ()) -> [row] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (\row
row -> IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ Connection -> Query -> IO Int64
execute_ Connection
conn (ByteString -> Query
Query (Text -> ByteString
encodeUtf8 (Text -> row -> Text
upsertSQL Text
schemaName row
row)))) [row]
rows

-- | Prefixes declared with more than one distinct parameter set.
conflictingPrefixes :: (Ord params) => (row -> Text) -> (row -> params) -> [row] -> [Text]
conflictingPrefixes :: forall params row.
Ord params =>
(row -> Text) -> (row -> params) -> [row] -> [Text]
conflictingPrefixes row -> Text
prefixOf row -> params
paramsOf [row]
rows =
  Map Text (Set params) -> [Text]
forall k a. Map k a -> [k]
Map.keys ((Set params -> Bool)
-> Map Text (Set params) -> Map Text (Set params)
forall a k. (a -> Bool) -> Map k a -> Map k a
Map.filter ((Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
1) (Int -> Bool) -> (Set params -> Int) -> Set params -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Set params -> Int
forall a. Set a -> Int
Set.size) Map Text (Set params)
paramsByPrefix)
  where
    paramsByPrefix :: Map Text (Set params)
paramsByPrefix =
      (Set params -> Set params -> Set params)
-> [(Text, Set params)] -> Map Text (Set params)
forall k a. Ord k => (a -> a -> a) -> [(k, a)] -> Map k a
Map.fromListWith Set params -> Set params -> Set params
forall a. Ord a => Set a -> Set a -> Set a
Set.union [(row -> Text
prefixOf row
row, params -> Set params
forall a. a -> Set a
Set.singleton (row -> params
paramsOf row
row)) | row
row <- [row]
rows]

-- | Converge the bucket table's WAL persistence to the declared durability. Reads
-- the current @pg_class.relpersistence@ and issues @SET LOGGED@/@SET UNLOGGED@ on a
-- change. The ALTER rewrites the table under @ACCESS EXCLUSIVE@. Token consumes block
-- briefly while a switch runs.
reconcileRateLimitDurability :: PG.Connection -> SchemaName -> Durability -> IO ()
reconcileRateLimitDurability :: Connection -> Text -> Durability -> IO ()
reconcileRateLimitDurability Connection
conn Text
schemaName Durability
durability = do
  rows <-
    Connection -> Query -> (Text, Text) -> IO [Only Text]
forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query
      Connection
conn
      Query
"SELECT relation.relpersistence::text FROM pg_class relation \
      \JOIN pg_namespace namespace ON namespace.oid = relation.relnamespace \
      \WHERE namespace.nspname = ? AND relation.relname = ?"
      (Text
schemaName, Text
arbiterRateLimitsTableName)
  let target = case Durability
durability of
        Durability
Durable -> Text
"p" :: Text
        Durability
Unlogged -> Text
"u"
  case rows of
    (Only Text
current : [Only Text]
_)
      | Text
current Text -> Text -> Bool
forall a. Eq a => a -> a -> Bool
/= Text
target ->
          IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$ Connection -> Query -> IO Int64
execute_ Connection
conn (ByteString -> Query
Query (Text -> ByteString
encodeUtf8 (Durability -> Text -> Text
alterRateLimitsDurabilitySQL Durability
durability Text
schemaName)))
    [Only Text]
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()

-- | Converge the optional triggers after the tracked migrations. This restores what an
-- earlier disabled configuration removed and leaves migration history alone. The sweep
-- is schema-wide. The table list is the schema's whole queue set. An object that
-- belongs to a queue outside it is dropped.
reconcileOptionalTriggers :: PG.Connection -> SchemaName -> [TableName] -> MigrationConfig -> IO ()
reconcileOptionalTriggers :: Connection -> Text -> [Text] -> MigrationConfig -> IO ()
reconcileOptionalTriggers Connection
conn Text
schemaName [Text]
tables MigrationConfig
config =
  Connection -> IO () -> IO ()
forall a. Connection -> IO a -> IO a
PG.withTransaction Connection
conn (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    IO ()
adoptUnmarkedObjects
    if MigrationConfig -> Bool
enableNotifications MigrationConfig
config
      then do
        [Text] -> IO ()
dropNotifyObjectsExcept [Text]
tables
        (Text -> IO ()) -> [Text] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ Text -> IO ()
installNotify [Text]
tables
      else [Text] -> IO ()
dropNotifyObjectsExcept []
    if MigrationConfig -> Bool
enableEventStreaming MigrationConfig
config
      then do
        Text -> IO ()
executeSQL (Text -> Text
createEventStreamingFunctionSQL Text
schemaName)
        [Text] -> IO ()
dropEventTriggersExcept [Text]
eventTables
        (Text -> IO ()) -> [Text] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ Text -> IO ()
installEventStream [Text]
tables
      else do
        [Text] -> IO ()
dropEventTriggersExcept []
        ours <- IO Bool
eventFunctionIsMarked
        -- The function stays while foreign triggers depend on it.
        depended <- eventFunctionHasTriggers
        when (ours && not depended) $ executeSQL (dropEventStreamingFunctionSQL schemaName)
  where
    -- Functions are replaced in place. That takes no lock on the queue table. Triggers
    -- are rebuilt when their marker is off the current version.
    installNotify :: Text -> IO ()
installNotify Text
table = do
      Text -> IO ()
executeSQL (Text -> Text -> Text
createNotifyFunctionSQL Text
schemaName Text
table)
      current <- Text -> Text -> Text -> IO Bool
triggerIsCurrent Text
table (Text -> Text
notifyTriggerName Text
table) Text
notifyObjectComment
      unless current $ executeSQL (createNotifyTriggerSQL schemaName table)

    installEventStream :: Text -> IO ()
installEventStream Text
table = do
      mainCurrent <- Text -> Text -> Text -> IO Bool
triggerIsCurrent Text
table (Text -> Text
eventStreamingTriggerName Text
table) Text
eventStreamingObjectComment
      dlqCurrent <- triggerIsCurrent (table <> "_dlq") (eventStreamingDLQTriggerName table) eventStreamingObjectComment
      unless (mainCurrent && dlqCurrent) $
        executeSQL (createEventStreamingTriggersSQL schemaName table)

    triggerIsCurrent :: Text -> Text -> Text -> IO Bool
triggerIsCurrent Text
table Text
trigger Text
marker =
      Query -> (Text, Text, Text, Text) -> IO Bool
forall params. ToRow params => Query -> params -> IO Bool
exists
        ( Query
"SELECT EXISTS (SELECT 1 "
            Query -> Query -> Query
forall a. Semigroup a => a -> a -> a
<> Query
triggerJoins
            Query -> Query -> Query
forall a. Semigroup a => a -> a -> a
<> Query
"WHERE namespace.nspname = ? AND relation.relname = ? AND trigger.tgname = ? \
               \AND obj_description(trigger.oid, 'pg_trigger') = ?)"
        )
        (Text
schemaName, Text
table, Text
trigger, Text
marker)

    eventFunctionIsMarked :: IO Bool
eventFunctionIsMarked =
      Query -> (Text, Text, Text) -> IO Bool
forall params. ToRow params => Query -> params -> IO Bool
exists
        Query
"SELECT EXISTS (SELECT 1 FROM pg_proc function JOIN pg_namespace namespace ON namespace.oid = function.pronamespace \
        \WHERE namespace.nspname = ? AND function.proname = ? AND function.pronargs = 0 \
        \AND obj_description(function.oid, 'pg_proc') LIKE ?)"
        (Text
schemaName, Text
eventStreamingFunctionName, Text
eventStreamingObjectCommentPrefix Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"%")

    eventFunctionHasTriggers :: IO Bool
eventFunctionHasTriggers =
      Query -> (Text, Text) -> IO Bool
forall params. ToRow params => Query -> params -> IO Bool
exists
        Query
"SELECT EXISTS (SELECT 1 FROM pg_trigger trigger JOIN pg_proc function ON function.oid = trigger.tgfoid \
        \JOIN pg_namespace namespace ON namespace.oid = function.pronamespace \
        \WHERE namespace.nspname = ? AND function.proname = ? AND NOT trigger.tgisinternal)"
        (Text
schemaName, Text
eventStreamingFunctionName)

    exists :: (PG.ToRow params) => Query -> params -> IO Bool
    exists :: forall params. ToRow params => Query -> params -> IO Bool
exists Query
sql params
params = do
      rows <- Connection -> Query -> params -> IO [Only Bool]
forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query Connection
conn Query
sql params
params
      pure (maybe False fromOnly (listToMaybe rows))

    dropTriggersMarked :: Text -> [Text] -> IO ()
dropTriggersMarked Text
marker [Text]
keep =
      Query -> (Text, Text, PGArray Text) -> IO ()
forall params. ToRow params => Query -> params -> IO ()
executeRendered
        ( Query
dropTriggerSelect
            Query -> Query -> Query
forall a. Semigroup a => a -> a -> a
<> Query
triggerJoins
            Query -> Query -> Query
forall a. Semigroup a => a -> a -> a
<> Query
"WHERE namespace.nspname = ? AND obj_description(trigger.oid, 'pg_trigger') LIKE ? \
               \AND NOT (relation.relname::text = ANY(?::text[]))"
        )
        (Text
schemaName, Text
marker Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"%", [Text] -> PGArray Text
forall a. [a] -> PGArray a
PGArray [Text]
keep)

    dropEventTriggersExcept :: [Text] -> IO ()
dropEventTriggersExcept = Text -> [Text] -> IO ()
dropTriggersMarked Text
eventStreamingObjectCommentPrefix

    dropNotifyObjectsExcept :: [Text] -> IO ()
dropNotifyObjectsExcept [Text]
keep = do
      Text -> [Text] -> IO ()
dropTriggersMarked Text
notifyObjectCommentPrefix [Text]
keep
      -- A function that still carries a trigger stays.
      Query -> (Text, Text, PGArray Text) -> IO ()
forall params. ToRow params => Query -> params -> IO ()
executeRendered
        Query
"SELECT format('DROP FUNCTION IF EXISTS %I.%I();', namespace.nspname, function.proname) \
        \FROM pg_proc function \
        \JOIN pg_namespace namespace ON namespace.oid = function.pronamespace \
        \WHERE namespace.nspname = ? AND obj_description(function.oid, 'pg_proc') LIKE ? AND function.pronargs = 0 \
        \AND NOT (function.proname::text = ANY(?::text[])) \
        \AND NOT EXISTS (SELECT 1 FROM pg_trigger trigger WHERE trigger.tgfoid = function.oid AND NOT trigger.tgisinternal)"
        (Text
schemaName, Text
notifyObjectCommentPrefix Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"%", [Text] -> PGArray Text
forall a. [a] -> PGArray a
PGArray ((Text -> Text) -> [Text] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map Text -> Text
notifyFunctionName [Text]
keep))

    -- Objects installed before arbiter stamped markers carry no comment. Adoption
    -- stamps a marker that no install writes. The sweeps match it. The currency probe
    -- rejects it. Adoption covers the names this registry generates. An unmarked
    -- lookalike stays foreign.
    adoptUnmarkedObjects :: IO ()
adoptUnmarkedObjects = do
      Query -> (Text, Text, PGArray Text) -> IO ()
forall params. ToRow params => Query -> params -> IO ()
executeRendered
        Query
"SELECT format('COMMENT ON FUNCTION %I.%I() IS %L;', namespace.nspname, function.proname, ?::text) \
        \FROM pg_proc function \
        \JOIN pg_namespace namespace ON namespace.oid = function.pronamespace \
        \WHERE namespace.nspname = ? AND function.proname = ANY(?::text[]) AND function.pronargs = 0 \
        \AND obj_description(function.oid, 'pg_proc') IS NULL"
        (Text
notifyAdoptedObjectComment, Text
schemaName, [Text] -> PGArray Text
forall a. [a] -> PGArray a
PGArray ((Text -> Text) -> [Text] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map Text -> Text
notifyFunctionName [Text]
tables))
      Text -> [Text] -> [Text] -> IO ()
adoptTriggers Text
notifyAdoptedObjectComment ((Text -> Text) -> [Text] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map Text -> Text
notifyTriggerName [Text]
tables) [Text]
tables
      Text -> [Text] -> [Text] -> IO ()
adoptTriggers Text
eventStreamingAdoptedObjectComment [Text]
eventTriggerNames [Text]
eventTables
      -- The shared function is adopted when one of its triggers is.
      Query -> (Text, Text, Text, Text) -> IO ()
forall params. ToRow params => Query -> params -> IO ()
executeRendered
        Query
"SELECT format('COMMENT ON FUNCTION %I.%I() IS %L;', namespace.nspname, function.proname, ?::text) \
        \FROM pg_proc function \
        \JOIN pg_namespace namespace ON namespace.oid = function.pronamespace \
        \WHERE namespace.nspname = ? AND function.proname = ? AND function.pronargs = 0 \
        \AND obj_description(function.oid, 'pg_proc') IS NULL \
        \AND EXISTS (SELECT 1 FROM pg_trigger trigger WHERE trigger.tgfoid = function.oid AND NOT trigger.tgisinternal \
        \AND obj_description(trigger.oid, 'pg_trigger') LIKE ?)"
        ( Text
eventStreamingAdoptedObjectComment
        , Text
schemaName
        , Text
eventStreamingFunctionName
        , Text
eventStreamingObjectCommentPrefix Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"%"
        )

    adoptTriggers :: Text -> [Text] -> [Text] -> IO ()
adoptTriggers Text
marker [Text]
triggerNames [Text]
relNames =
      Query -> (Text, Text, PGArray Text, PGArray Text) -> IO ()
forall params. ToRow params => Query -> params -> IO ()
executeRendered
        ( Query
commentTriggerSelect
            Query -> Query -> Query
forall a. Semigroup a => a -> a -> a
<> Query
triggerJoins
            Query -> Query -> Query
forall a. Semigroup a => a -> a -> a
<> Query
"WHERE namespace.nspname = ? AND NOT trigger.tgisinternal AND trigger.tgname = ANY(?::text[]) \
               \AND relation.relname = ANY(?::text[]) AND obj_description(trigger.oid, 'pg_trigger') IS NULL"
        )
        (Text
marker, Text
schemaName, [Text] -> PGArray Text
forall a. [a] -> PGArray a
PGArray [Text]
triggerNames, [Text] -> PGArray Text
forall a. [a] -> PGArray a
PGArray [Text]
relNames)

    eventTables :: [Text]
eventTables = (Text -> [Text]) -> [Text] -> [Text]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap (\Text
table -> [Text
table, Text
table Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_dlq"]) [Text]
tables

    eventTriggerNames :: [Text]
eventTriggerNames =
      ((Text, Bool) -> Text) -> [(Text, Bool)] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map (Text, Bool) -> Text
forall a b. (a, b) -> a
fst [(Text, Bool)]
legacyEventStreamingTriggers
        [Text] -> [Text] -> [Text]
forall a. Semigroup a => a -> a -> a
<> (Text -> [Text]) -> [Text] -> [Text]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap (\Text
table -> [Text -> Text
eventStreamingTriggerName Text
table, Text -> Text
eventStreamingDLQTriggerName Text
table]) [Text]
tables

    executeRendered :: (PG.ToRow params) => Query -> params -> IO ()
    executeRendered :: forall params. ToRow params => Query -> params -> IO ()
executeRendered Query
sql params
params = do
      commands <- Connection -> Query -> params -> IO [Only Text]
forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query Connection
conn Query
sql params
params
      traverse_ (\(Only Text
command) -> Text -> IO ()
executeSQL Text
command) commands

    executeSQL :: Text -> IO ()
executeSQL = IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> (Text -> IO Int64) -> Text -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Connection -> Query -> IO Int64
execute_ Connection
conn (Query -> IO Int64) -> (Text -> Query) -> Text -> IO Int64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ByteString -> Query
Query (ByteString -> Query) -> (Text -> ByteString) -> Text -> Query
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Text -> ByteString
encodeUtf8

-- | The catalog joins the trigger sweeps share.
triggerJoins :: Query
triggerJoins :: Query
triggerJoins =
  Query
"FROM pg_trigger trigger \
  \JOIN pg_class relation ON relation.oid = trigger.tgrelid \
  \JOIN pg_namespace namespace ON namespace.oid = relation.relnamespace "

-- | A @DROP TRIGGER@ statement per matched row.
dropTriggerSelect :: Query
dropTriggerSelect :: Query
dropTriggerSelect = Query
"SELECT format('DROP TRIGGER IF EXISTS %I ON %I.%I;', trigger.tgname, namespace.nspname, relation.relname) "

-- | A @COMMENT ON TRIGGER@ statement per matched row, stamping the first parameter.
commentTriggerSelect :: Query
commentTriggerSelect :: Query
commentTriggerSelect =
  Query
"SELECT format('COMMENT ON TRIGGER %I ON %I.%I IS %L;', trigger.tgname, namespace.nspname, relation.relname, ?::text) "

-- | The schema-level migrations, run once per schema. Exposed for the golden suite.
-- 'reconcileOptionalTriggers' owns the optional notification and event-streaming
-- objects.
schemaLevelMigrations :: SchemaName -> [MigrationCommand]
schemaLevelMigrations :: Text -> [MigrationCommand]
schemaLevelMigrations Text
schemaName =
  [ String -> ByteString -> MigrationCommand
MigrationScript String
"create-cron-schedules" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
createCronSchedulesTableSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"cron-schedules-add-timezone" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
addTimezoneColumnSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"cron-schedules-add-queue-name" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
addQueueNameColumnSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"cron-schedules-add-run-requested" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
addRunRequestedColumnSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"cron-schedules-add-last-manual-run" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
addLastManualRunColumnSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"create-arbiter-workers" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
createWorkersTableSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"create-arbiter-queues" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
createQueuesTableSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"create-arbiter-gates" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
createGatesTableSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"arbiter-gates-add-metadata" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
addGateMetadataColumnSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"create-arbiter-rate-limit-policies" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
createRateLimitPoliciesTableSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"create-arbiter-rate-limits" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
createRateLimitsTableSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"create-arbiter-concurrency-policies" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
createConcurrencyPoliciesTableSQL Text
schemaName)
  , String -> ByteString -> MigrationCommand
MigrationScript String
"create-arbiter-concurrency" (Text -> ByteString
encodeUtf8 (Text -> ByteString) -> Text -> ByteString
forall a b. (a -> b) -> a -> b
$ Text -> Text
createConcurrencyTableSQL Text
schemaName)
  ]

-- | One queue's tracked migrations, each under its own version identifier.
-- 'reconcileOptionalTriggers' owns the optional notify and event-streaming objects.
jobQueueMigrationsForTable
  :: SchemaName
  -- ^ Schema name
  -> TableName
  -- ^ Table name
  -> TableAdmission
  -- ^ Which admission trigger kinds to install
  -> [MigrationCommand]
  -- ^ List of migration commands
jobQueueMigrationsForTable :: Text -> Text -> TableAdmission -> [MigrationCommand]
jobQueueMigrationsForTable Text
schemaName Text
tableName TableAdmission
admission =
  let prefix :: String
prefix = Text -> String
T.unpack Text
tableName String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
"-"
      script :: String -> Text -> MigrationCommand
script String
name Text
sql = String -> ByteString -> MigrationCommand
MigrationScript (String
prefix String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
name) (Text -> ByteString
encodeUtf8 Text
sql)

      coreMigrations :: [MigrationCommand]
coreMigrations =
        [ String -> Text -> MigrationCommand
script String
"create-table" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createJobQueueTableSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-dlq-table" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createJobQueueDLQTableSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-dlq-group-key-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createDLQGroupKeyIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-dlq-failed-at-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createDLQFailedAtIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-dedup-key-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createDedupKeyIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-group-key-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createJobQueueGroupKeyIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-parent-id-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createParentIdIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-dlq-parent-id-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createDLQParentIdIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-results-table" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createResultsTableSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-groups-table" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createGroupsTableSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"add-claimed-by-column" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
addClaimedByColumnSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"migrate-ungrouped-ready-split-indexes" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
migrateUngroupedReadySplitIndexesSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"migrate-groups-ready-ranking" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
migrateGroupsReadyRankingSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"add-rate-limit-columns" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
addRateLimitColumnsSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-throttled-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createThrottledIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"add-concurrency-columns" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
addConcurrencyColumnsSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-concurrency-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createConcurrencyIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"add-rate-limit-cost-column" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
addRateLimitCostColumnSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"set-job-queue-fillfactor-100" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$
            Text
"ALTER TABLE " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text -> Text
jobQueueTable Text
schemaName Text
tableName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" SET (fillfactor = 100);"
        , String -> Text -> MigrationCommand
script String
"set-max-attempts-default" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
setMaxAttemptsDefaultSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"add-cancel-requested-at-column" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
addCancelRequestedAtColumnSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"add-archive-for-column" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
addArchiveForColumnSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-archive-table" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createJobQueueArchiveTableSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-archive-completed-at-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createArchiveCompletedAtIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-archive-expires-at-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createArchiveExpiresAtIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-archive-job-id-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createArchiveJobIdIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-archive-parent-id-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createArchiveParentIdIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-archive-group-key-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createArchiveGroupKeyIndexSQL Text
schemaName Text
tableName
        , -- Runs after the archive table exists. It alters that table too.
          String -> Text -> MigrationCommand
script String
"add-trace-context-column" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
addTraceContextColumnSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"add-claim-seq-column" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
addClaimSeqColumnSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-groups-emptied-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createGroupsEmptiedIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-group-retried-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createJobQueueGroupRetriedIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-grouped-due-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createJobQueueGroupedDueIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-group-in-flight-index" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createJobQueueGroupInFlightIndexSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-groups-trigger-functions-v10" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createGroupsTriggerFunctionsSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"create-groups-triggers" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createGroupsTriggersSQL Text
schemaName Text
tableName
        , String -> Text -> MigrationCommand
script String
"add-kind-column" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
addKindColumnSQL Text
schemaName Text
tableName
        ]
      concurrencyTriggers :: [MigrationCommand]
concurrencyTriggers
        | TableAdmission -> Bool
tableConcurrency TableAdmission
admission =
            [ String -> Text -> MigrationCommand
script String
"create-concurrency-trigger-functions" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createConcurrencyTriggerFunctionsSQL Text
schemaName Text
tableName
            , String -> Text -> MigrationCommand
script String
"create-concurrency-triggers" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createConcurrencyTriggersSQL Text
schemaName Text
tableName
            ]
        | Bool
otherwise = []
      rateLimitTriggers :: [MigrationCommand]
rateLimitTriggers
        | TableAdmission -> Bool
tableRateLimit TableAdmission
admission =
            [ String -> Text -> MigrationCommand
script String
"create-rate-limit-bucket-trigger-functions" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createRateLimitBucketTriggerFunctionsSQL Text
schemaName Text
tableName
            , String -> Text -> MigrationCommand
script String
"create-rate-limit-bucket-triggers" (Text -> MigrationCommand) -> Text -> MigrationCommand
forall a b. (a -> b) -> a -> b
$ Text -> Text -> Text
createRateLimitBucketTriggersSQL Text
schemaName Text
tableName
            ]
        | Bool
otherwise = []
   in [MigrationCommand]
coreMigrations [MigrationCommand] -> [MigrationCommand] -> [MigrationCommand]
forall a. Semigroup a => a -> a -> a
<> [MigrationCommand]
concurrencyTriggers [MigrationCommand] -> [MigrationCommand] -> [MigrationCommand]
forall a. Semigroup a => a -> a -> a
<> [MigrationCommand]
rateLimitTriggers

-- | Whether a schema exists.
schemaExists :: PG.Connection -> Text -> IO Bool
schemaExists :: Connection -> Text -> IO Bool
schemaExists Connection
conn Text
schemaName = do
  rows <- Connection -> Query -> Only Text -> IO [Only Int]
forall q r.
(ToRow q, FromRow r) =>
Connection -> Query -> q -> IO [r]
query Connection
conn Query
"SELECT 1 FROM pg_namespace WHERE nspname = ?" (Text -> Only Text
forall a. a -> Only a
Only Text
schemaName) :: IO [Only Int]
  pure (not (null rows))