
CREATE OR REPLACE FUNCTION InitializeServer() RETURNS VOID AS
$$
BEGIN
  /* the pgtask server requries postgres 11 */
  IF (SELECT setting::INT FROM pg_settings WHERE name = 'server_version_num')
    < 110000
  THEN
    RAISE EXCEPTION 'pg_task server requires postgres 11 or greater';
  END IF;

  /* the pgtask server requries dblink */
  CREATE EXTENSION IF NOT EXISTS dblink;

  /* generate a sting that identifies the pgtask server.  That way, when we nuke
   * tasks, we make sure to only target tasks runnign from *this* pgtask
   * instance. The server identification is generated from pgtask databasebase
   * name, server ip, and port. (if those values change the id is not updated
   * however).
   */
  CREATE OR REPLACE FUNCTION ServerIdPhrase() RETURNS TEXT AS
  $sip$
    SELECT format('%s_%s_%s',
      current_database(),
      inet_server_addr(),
      (SELECT setting FROM pg_settings WHERE Name = 'port'));
  $sip$ LANGUAGE SQL;

  CREATE TABLE GlobalControl
  (
    Version TEXT,
    MainLoopSleepSecondsIdle NUMERIC DEFAULT 5.0,
    MainLoopSleepSecondsActive NUMERIC DEFAULT 0.1,
    Debug BOOL DEFAULT false,
    DebugQuery TEXT DEFAULT 'SELECT 0',
    MainProcessPid INT,
    GracefulShutdown BOOL,
    TaskMinMessages TEXT DEFAULT 'WARNING',
    HasPLSH BOOL DEFAULT FALSE,
    ServerId TEXT DEFAULT md5(ServerIdPhrase()),
    StartTime TIMESTAMPTZ,
    HasTDSFDW BOOL DEFAULT FALSE,
    LogMinMessages TEXT DEFAULT 'NOTICE',
    LogErrorVerbosity TEXT DEFAULT 'TERSE',
    IdleTaskTimeoutInterval INTERVAL DEFAULT '1 minute'::INTERVAL,
    IdleTaskTimeoutLastCheck TIMESTAMPTZ,
    AuditInterval INTERVAL DEFAULT '5 minutes'::INTERVAL,
    LastAudit TIMESTAMPTZ
  );
  CREATE INDEX ON GlobalControl((1));

  INSERT INTO GlobalControl VALUES ('1.0');

  DO $q$
  BEGIN
    PERFORM 1 FROM pg_extension WHERE extname = 'plsh';
    IF FOUND
    THEN
      UPDATE GlobalControl SET HasPLSH = true;
    ELSE
      CREATE EXTENSION plsh;
    END IF;
  EXCEPTION WHEN OTHERS THEN
    UPDATE GlobalControl SET HasPLSH = false;

  END;
  $q$;

  DO $q$
  BEGIN
    PERFORM 1 FROM pg_extension WHERE extname = 'tds_fdw';
    IF FOUND
    THEN
      UPDATE GlobalControl SET HasTDSFDW = true;
    ELSE
      CREATE EXTENSION HasTDSFDW;
    END IF;
  EXCEPTION WHEN OTHERS THEN
    UPDATE GlobalControl SET HasTDSFDW = false;

  END;
  $q$;

  CREATE TABLE Environment
  (
    Environment TEXT PRIMARY KEY DEFAULT 'Production'
  );

  CREATE TYPE ServerType_t AS ENUM(
    'POSTGRES',
    'SQL SERVER',
    'WEB SERVICE');

  CREATE TABLE Server
  (
    Server TEXT PRIMARY KEY,

    TaskClientThreads INT,

    ServerType ServerType_t NOT NULL DEFAULT 'POSTGRES'
  );

  CREATE TABLE Database
  (
    Environment TEXT DEFAULT 'Production' REFERENCES Environment
      ON UPDATE CASCADE ON DELETE CASCADE,

    /* usually the name of the database, doesn't have to be */
    Database TEXT,

    /* If Database is distributed, each database must have a unique nodeid */
    NodeId INT DEFAULT 0,

    Server TEXT REFERENCES Server ON DELETE CASCADE ON UPDATE CASCADE,

    ConnectionString TEXT UNIQUE,
    LogMinMessages TEXT DEFAULT 'NOTICE',
    LogErrorVerbosity TEXT DEFAULT 'TERSE',
    PRIMARY KEY(Environment, Database, NodeId)
  );

  CREATE TYPE ConcurrencyModel_t AS ENUM(
    /* Task chain may not be created while another is active. */
    'BLOCK',
    /* Task chain may be created, but the second and later task chains
     * will be held so that only one is active at a time.
     */
    'QUEUE',
    /* Task chain may be created if and only if a task chain is not already
     * queued.
     */
    'QUEUE_ONCE',
    /* The task chain, when created, will merge the created DAG with an
     * in progress dag if when is active.
     */
    'MERGE',

    /* The task chain can always be run regardless of what else is runing */
    'NOBLOCK');

  CREATE TABLE TaskChainConfiguration
  (
    TaskChainConfigurationName TEXT PRIMARY KEY,

    /* Controls the behavior when a task chain is created while another task
     * chain is active with the same name in the same environment. Unnamed
     * task chains work as 'QUEUE' with the special dispensation that if
     * the task chain does not share any task with another active task
     * chains that are pending or in progress, it can be made to run
     * run immediately.
     */
    DefaultConcurrencyModel ConcurrencyModel_t NOT NULL DEFAULT 'BLOCK'
  );

  /*
   * Defines specific tasks.
   */
  CREATE TABLE Task
  (
    Task TEXT PRIMARY KEY,

    /* TaskBody can be null, in which case the task is a NOP. This task is then
     * only in place for depndency purposes or as a container for subtasks
     */
    TaskBody TEXT,
    RunSchedule INT[],
    Database TEXT,
    StartTime TIME,
    EndTime TIME,
    MaxRunTime INTERVAL,
    IsConcurrent BOOLEAN,

    /* Tasks can be a place holder for a task chain.  In that case the TaskBody
     * would be the name of the task chain.  During conversion to
     * TaskChainTask, the structure is flattened out to the task level.
     */
    IsTaskChainConfiguration BOOL DEFAULT FALSE,

    /* gives rough visual ordering */
    TaskOrder INT DEFAULT 0,

    /* If the task has subtasks, does one subtask failure cause a failure of
     * the task?  If there are no subtasks, value is moot.
     */
    SubTasksMustComplete BOOL DEFAULT TRUE,
    LogMinMessages TEXT DEFAULT 'NOTICE',
    LogErrorVerbosity TEXT DEFAULT 'TERSE'

  );

  CREATE INDEX ON Task(Database);

  CREATE TABLE TaskChainConfigurationTask
  (
    TaskChainConfigurationName TEXT REFERENCES TaskChainConfiguration
      ON UPDATE CASCADE ON DELETE CASCADE,
    Task TEXT REFERENCES Task ON UPDATE CASCADE ON DELETE CASCADE,
    PRIMARY KEY (TaskChainConfigurationName, Task)
  );

  /*
   * Subtasks have no dependency except on the parent task.  They are all
   * activated at the same time.
   */
  CREATE TABLE SubTask
  (
    Task TEXT REFERENCES Task ON DELETE CASCADE,
    SubTask TEXT,
    TaskBody TEXT,
    RunSchedule INT[],
    PRIMARY KEY(Task, SubTask)
  );

  /*
   * Holds Dependency details of one tasks on another
   */
  CREATE TABLE TaskDependency
  (
    ChildTask TEXT REFERENCES Task ON UPDATE CASCADE ON DELETE CASCADE,
    ParentTask TEXT REFERENCES Task ON UPDATE CASCADE ON DELETE CASCADE,
    AllNodes BOOL,
    CanRun BOOL,
    PRIMARY KEY(ChildTask, ParentTask)
  );

  /* tracks a chain of tasks that are run */
  CREATE TABLE TaskChain
  (
    TaskChainId BIGSERIAL PRIMARY KEY,
    Environment TEXT REFERENCES Environment
      ON UPDATE CASCADE ON DELETE CASCADE,

    /* set to a named configuration if provided */
    TaskChainConfigurationName TEXT,

    Entered TIMESTAMPTZ DEFAULT now(),

    Started TIMESTAMPTZ DEFAULT now(),
    Finished TIMESTAMPTZ,
    Paused BOOL DEFAULT false,
    PauseStartTime TIMESTAMPTZ,
    PauseEndTime TIMESTAMPTZ,
    DidSucceed BOOL,
    ErrorMessage TEXT,
    PendingAction TEXT CHECK (PendingAction IN ('PAUSE', 'CANCEL'))
  );

  CREATE INDEX ON TaskChain(Environment) WHERE Finished IS NULL;

  CREATE TYPE TaskExecutionState_t AS ENUM
  (
    /* task is still running */
    'RUNNING',

    /* task has completed, or has failed. */
    'FINISHED',

    /* the remote server connect reset */
    'RESET',

    /* task failed to execute */
    'FAILED'
  );

  CREATE TYPE TaskResult_t AS ENUM(
    'SUCCESS_COMPLETION',
    'SUCCESS_EMPTY',
    'SUCCESS_VIA_SUBTASKS',
    'FAILED_VIA_SUBTASKS',
    'FAILED_VIA_DEPENDENCY',
    'FAILED_VIA_PARENT_TASK',
    'FAILED_VIA_RESTART',
    'FAILED_VIA_CANCEL',
    'FAILED_VIA_EXECUTION',
    'FAILED_VIA_TIMEOUT',
    'FAILED_VIA_CONNECTION_RESET');

  /* tracks a specific instance of a task in a chain of tasks */
  CREATE TABLE TaskChainTask
  (
    TaskChainTaskId BIGSERIAL UNIQUE,
    TaskChainId BIGINT REFERENCES TaskChain ON DELETE CASCADE,

    Environment TEXT,
    Database TEXT,
    Task TEXT,
    SubTask TEXT,
    NodeId INT DEFAULT 0,

    Eligible BOOL NOT NULL,

    Started TIMESTAMPTZ,
    Finished TIMESTAMPTZ,
    TaskResult TaskResult_t,
    ErrorMessage TEXT,
    TaskBody TEXT,
    Yielded TIMESTAMPTZ,

    PRIMARY KEY(TaskChainId, Environment, Database, NodeId, Task, SubTask),
    FOREIGN KEY(Environment, Database, NodeId)
      REFERENCES Database ON DELETE CASCADE
  );

  CREATE INDEX ON TaskChainTask(Environment, Database, NodeId);


  /* optimizes the question, what tasks are eligible to be run? */
  CREATE INDEX ON TaskChainTask(
    Environment,
    Database,
    NodeId,
    TaskChainTaskId)
  WHERE
    Eligible
    AND Started IS NULL
    AND Finished IS NULL;

  /* optimizes the question, what tasks are running? */
  CREATE INDEX ON TaskChainTask(TaskChainTaskId)
  WHERE
    Started IS NOT NULL
    AND Finished IS NULL;

  CREATE OR REPLACE FUNCTION IsSubTask(_SubTask TEXT) RETURNS BOOL AS
  $q$
    SELECT _SubTask != '';
  $q$ LANGUAGE SQL IMMUTABLE;

  /* Tracks dblink connections to various databases.
   * The table is unlogged as it's meanliss across server restarts.
   */
  CREATE UNLOGGED TABLE ServerConnection
  (
    Server TEXT,
    ConnectionNumber INT,

    Environment TEXT,
    Database TEXT,
    NodeId INT,
    ConnectionName TEXT,
    Active BOOL,
    TaskChainTaskId BIGINT,
    Cancelling BOOL,
    PRIMARY KEY(Server, ConnectionNumber),

    UNIQUE(ConnectionName),

    UNIQUE(Environment, Database, NodeId, ConnectionNumber)

  );

  CREATE OR REPLACE VIEW vw_Server AS
    SELECT
      s.*,
      count(ConnectionName) FILTER (WHERE Active) AS ActiveConnections
    FROM Server s
    LEFT JOIN ServerConnection USING(Server)
    GROUP BY s.Server;

  CREATE OR REPLACE VIEW vw_DatabaseConnection AS
    SELECT *
    FROM Database
    JOIN ServerConnection USING(Server, Environment, Database, NodeId)
    JOIN Server USING(Server);

  CREATE OR REPLACE VIEW vw_ActiveTasks AS
    SELECT
      Server,
      sc.Environment,
      sc.Database,
      sc.NodeId,
      TaskChainTaskId,
      tct.Task,
      tct.SubTask,
      COALESCE(t.TaskBody, st.TaskBody) AS TaskBody,
      ConnectionNumber
    FROM ServerConnection sc
    JOIN TaskChainTask tct USING(TaskChainTaskId)
    LEFT JOIN Task t ON
      t.Task =  tct.Task AND tct.SubTask = ''
    LEFT JOIN SubTask st ON
      st.Task =  tct.Task
      AND st.SubTask = tct.SubTask
      AND tct.SubTask != ''
    WHERE
      Active
      AND COALESCE(t.TaskBody, st.TaskBody) != '';

    CREATE TABLE ActiveTaskAudit AS
    SELECT 
      now() AS AuditTime,
      vat.*
    FROM vw_ActiveTasks vat
    LIMIT 0;

    CREATE TABLE MasterLog
    (
      MasterLogId BIGSERIAL PRIMARY KEY,
      Happened TIMESTAMPTZ DEFAULT clock_timestamp(),
      Level TEXT,
      Log TEXT
    );


    CREATE TABLE AdministrativeCommand
    (
      Entered TIMESTAMPTZ DEFAULT NOW(),
      Consumed TIMESTAMPTZ,
      Query TEXT,
      Result JSON,
      ErrorMessage TEXT
    );

    CREATE UNIQUE INDEX ON AdministrativeCommand((1));

END;
$$ LANGUAGE PLPGSQL;

/* initialize the schema if needed, other stuff always. */
DO
$$
BEGIN
  PERFORM 1 FROM GlobalControl;
EXCEPTION WHEN OTHERS THEN
  PERFORM InitializeServer();
END;
$$;


CREATE OR REPLACE FUNCTION DidSucceed(TaskResult_t) RETURNS BOOL AS
$$
  SELECT $1
    IN ('SUCCESS_COMPLETION', 'SUCCESS_EMPTY', 'SUCCESS_VIA_SUBTASKS');
$$ LANGUAGE SQL IMMUTABLE;


CREATE OR REPLACE FUNCTION ShouldCancel(TaskResult_t) RETURNS BOOL AS
$$
  SELECT $1
    IN ('FAILED_VIA_TIMEOUT', 'FAILED_VIA_CANCEL');
$$ LANGUAGE SQL IMMUTABLE;


/* When a task connection is no longer determined to be blocking, it's time to
 * resolve it and capture the error message (if any).
 */
CREATE OR REPLACE FUNCTION ReapTask(
  _ConnectionName TEXT,
  _WantError BOOL DEFAULT true,
  TaskExecutionState OUT TaskExecutionState_t,
  ErrorMessage OUT TEXT) RETURNS RECORD AS
$$
DECLARE
  _Error TEXT;
BEGIN
  /* all errors have to be trapped due to not being able to handle thrown
   * canclled queries. We can check the error message to see if the query
   * succeeded or not.
   */
  PERFORM dblink_get_result(_ConnectionName, FALSE);

  _Error := dblink_error_message(_ConnectionName);

  IF _Error = 'OK'
  THEN
    TaskExecutionState := 'FINISHED';
  ELSE
    TaskExecutionState := 'FAILED';
  END IF;

  ErrorMessage := _Error;

  /* dblink requires an extra call to reset connection state.  This can also
   * fail if the connection is dead.
   */
  PERFORM dblink_get_result(_ConnectionName, FALSE);

  _Error := dblink_error_message(_ConnectionName);

  IF _Error LIKE '%server closed the connection%'
  THEN
    TaskExecutionState := 'RESET';
    ErrorMessage := _Error;
  END IF;

  /* If there is an error on the task, disconnect the connection as a
   * precaution
   */
  IF TaskExecutionState = 'FAILED'
  THEN
    PERFORM Disconnect(_ConnectionName);
  END IF;
END;
$$ LANGUAGE PLPGSQL;



CREATE OR REPLACE FUNCTION CancelTask(_ConnectionName TEXT) RETURNS VOID AS
$$
DECLARE
  s vw_DatabaseConnection;
BEGIN
  SELECT INTO s *
  FROM vw_DatabaseConnection WHERE ConnectionName = _ConnectionName;

  PERFORM Log(format(
    'Cancelling connection %s id %s environment %s server %s node %s',
    _ConnectionName,
    s.TaskChainTaskId,
    s.Environment,
    s.Server,
    s.NodeId));

  PERFORM dblink_cancel_query(_ConnectionName);

  UPDATE ServerConnection SET Cancelling = true
  WHERE ConnectionName = s.ConnectionName;
END;
$$ LANGUAGE PLPGSQL;



/*
 * Is the task still running?
 *
 * XXX: Bizarrely, the only way to determine if the connection has been
 *      reset without blocking is to call 'dblink_get_notify' twice so that
 *      the connection errror is established.  This could probably be fixed
 *      by having dblink_is_busy check the response of PQconsumeInput.
 */
CREATE OR REPLACE FUNCTION TaskRunning(
  _ConnectionName TEXT) RETURNS TaskExecutionState_t AS
$$
DECLARE
  _ErrorMessage TEXT;
BEGIN
  IF dblink_is_busy(_ConnectionName) = 0
  THEN
    RETURN 'FINISHED';
  END IF;

  PERFORM * from dblink_get_notify(_ConnectionName);
  PERFORM * from dblink_get_notify(_ConnectionName);

  _ErrorMessage := dblink_error_message(_ConnectionName);

  IF _ErrorMessage LIKE 'server closed the connection%'
    OR _ErrorMessage LIKE 'connection not open'
  THEN
    PERFORM Log('WARNING', format('connection: %s is dead via %s', 
      _ConnectionName, _ErrorMessage));

    /* connection dead. pull the result while ignoring the error. */
    PERFORM dblink_get_result(_ConnectionName, false);

    /* clear connection status */
    PERFORM dblink_get_result(_ConnectionName);

    RETURN 'RESET';
  END IF;

  RETURN 'RUNNING';
END;
$$ LANGUAGE PLPGSQL;



CREATE OR REPLACE FUNCTION Log(
  TEXT,
  TEXT = NULL,
  _Debug BOOL = false) RETURNS VOID AS
$$
DECLARE
  _Message TEXT;
  _Level TEXT;
BEGIN
  IF $2 = 'DEBUG' AND _Debug IS NOT true
  THEN
    RETURN;
  END IF;

  IF $2 IS NULL
  THEN
    _Message := $1;
    _Level = 'NOTICE';
  ELSEIF $1 = 'DEBUG'
  THEN
    _Message := $2;
    _Level = 'NOTICE';
  ELSE
    _Message := $2;
    _Level = upper($1);
  END IF;

  INSERT INTO MasterLog(Level, Log) VALUES (_Level, _Message);

  IF _Level = 'NOTICE'
  THEN
    RAISE NOTICE '%: %', clock_timestamp()::timestamp(1), _Message;
  ELSIF _Level = 'WARNING'
  THEN
    RAISE WARNING '%: %', clock_timestamp()::timestamp(1), _Message;
  ELSE
    PERFORM * FROM
      dblink_exec('dbname=' || current_database(),
        format(
          $q$INSERT INTO MasterLog(Level, Log) VALUES ('ERROR', %s)$q$,
          quote_literal(_message)));

    RAISE EXCEPTION '%: %', clock_timestamp()::timestamp(1), _Message;
  END IF;
END;
$$ LANGUAGE PLPGSQL;


CREATE OR REPLACE FUNCTION Debug(ANYELEMENT) RETURNS ANYELEMENT AS
$$
BEGIN
  RAISE NOTICE '%', $1;
  RETURN $1;
END;
$$ LANGUAGE PLPGSQL;

CREATE OR REPLACE FUNCTION TaskClientThreads(_Server TEXT) RETURNS INT AS
$$
  SELECT TaskClientThreads FROM Server WHERE Server = _Server;
$$ LANGUAGE SQL STABLE;

/* Connect to client databases, make sure they are all configured */
CREATE OR REPLACE FUNCTION InitializeClientDatabases() RETURNS VOID AS
$$
DECLARE
  s Server;
  d Database;

  _HasClientDatabase BOOL;
  _Version INT;
BEGIN
  /* Initialize Client servers */
  FOR d IN SELECT * FROM Database WHERE Server IN(
    SELECT Server FROM Server WHERE Configured)
  LOOP
    /* test database connection and get version information */
    SELECT v INTO _Version
    FROM dblink(
      d.ConnectionString,
      $q$SELECT setting FROM pg_settings WHERE name = 'server_version_num'$q$)
        AS R(v INT);

    /* if we got this far we are able to connect :-). */
    UPDATE Database SET Version = _Version WHERE Database = d.Database;
  END LOOP;
END;
$$ LANGUAGE PLPGSQL;

/* Can this task chain be run? Return true if it can, false if it can be
 * queue.  If it is not able to be queued, return an error.
 */
CREATE OR REPLACE FUNCTION TaskChainCanStart(
  _TaskChainConfigurationName TEXT,
  _Tasks TEXT[],
  _ConcurrencyModel ConcurrencyModel_t,
  _Environment TEXT) RETURNS BOOL AS
$$
DECLARE
  _Running BOOL;
  _AlreadyQueued BOOL;
BEGIN
  IF _ConcurrencyModel = 'NOBLOCK'
  THEN
    RETURN true;
  END IF;

  IF _ConcurrencyModel = 'BLOCK'
  THEN
    IF _TaskChainConfigurationName IS NULL
    THEN
      SELECT INTO _Running
        EXISTS
        (
          SELECT 1
          FROM TaskChainTask tct
          JOIN TaskChain tc USING(TaskChainId)
          WHERE
            Task = any(_Tasks)
            AND tc.Environment = _Environment
            AND tct.Finished IS NULL
            AND tc.Finished IS NULL
        );
    ELSE
      SELECT INTO _Running
        EXISTS
        (
          SELECT 1
          FROM TaskChain
          WHERE
            TaskChainConfigurationName = _TaskChainConfigurationName
            AND Environment = _Environment
            AND Started IS NOT NULL AND Finished IS NULL
        );
    END IF;

    IF _Running
    THEN
      PERFORM Log('ERROR', format(
        'Can not start task chain configuration %s while already running',
        COALESCE(_TaskChainConfigurationName, '<unnamed>')));
    END IF;

    RETURN true;
  END IF;

  IF _ConcurrencyModel IN ('QUEUE', 'QUEUE_ONCE')
  THEN
    IF _TaskChainConfigurationName IS NULL
    THEN
      SELECT INTO _Running, _AlreadyQueued
        COUNT(DISTINCT TaskChainId) FILTER (WHERE tct.Started IS NOT NULL) > 0,
        COUNT(DISTINCT TaskChainId) FILTER (WHERE tct.Started IS NULL) > 0
      FROM TaskChainTask tct
      JOIN TaskChain tc USING(TaskChainId)
      WHERE
        Task = any(_Tasks)
        AND tc.Environment = _Environment
        AND tct.Finished IS NULL
        AND tc.Finished IS NULL;
    ELSE
      SELECT INTO _Running, _AlreadyQueued
        COUNT(*) FILTER (WHERE Started IS NOT NULL) > 0,
        COUNT(*) FILTER (WHERE Started IS NULL) > 0
      FROM TaskChain
      WHERE
        TaskChainConfigurationName = _TaskChainConfigurationName
        AND Environment = _Environment
        AND Finished IS NULL;
    END IF;

    IF _ConcurrencyModel = 'QUEUE_ONCE' AND _AlreadyQueued
    THEN
      PERFORM Log('ERROR', format(
        'Can not queue task chain configuration %s while already queued',
        COALESCE(_TaskChainConfigurationName, '<unnamed>')));
    END IF;

    /* If it the task is already running, place it on the queue */
    RETURN NOT _Running;
  END IF;

  IF _ConcurrencyModel = 'MERGE'
  THEN
    PERFORM Log('ERROR', 'Unimplemented');
  END IF;
END;
$$ LANGUAGE PLPGSQL;

/* Sets up a task chain based on an array of tasks */
CREATE OR REPLACE FUNCTION CreateTaskChain(
  _Tasks TEXT[], /* these are the tasks that *must* be run */
  _Environment TEXT DEFAULT 'Production',
  _WantParents BOOL DEFAULT true,
  _WantChildren BOOL DEFAULT true,
  _TaskChainConfigurationName TEXT DEFAULT NULL,
  _ForceConcurrencyModel ConcurrencyModel_t DEFAULT NULL,
  _User TEXT DEFAULT 'System') RETURNS BIGINT AS
$$
DECLARE
  r RECORD;
  _TaskChainId BIGINT;
  tcc TaskChainConfiguration;
  _CanStart BOOL;
BEGIN
   PERFORM Log(format(
    'User %s is requesting %s in %s',
    _User,
    'New Task Chain ' || COALESCE(_TaskChainConfigurationName, ''),
     _Environment
    ));

  /* The task chain may or may not be able to be immediately started. */
  _CanStart := TaskChainCanStart(
    _TaskChainConfigurationName,
    _Tasks,
    COALESCE(
      _ForceConcurrencyModel,
      (
          SELECT DefaultConcurrencyModel
          FROM TaskChainConfiguration
          WHERE TaskChainConfigurationName = _TaskChainConfigurationName
      ),
      'QUEUE_ONCE'::ConcurrencyModel_t),
    _Environment);

  INSERT INTO TaskChain(
    Environment,
    TaskChainConfigurationName,
    Started)
  VALUES (
    _Environment,
    _TaskChainConfigurationName,
    CASE WHEN _CanStart THEN clock_timestamp() END)
  RETURNING TaskChainId INTO _TaskChainId;

  /* which tasks are we running? */

  WITH tasks AS
  (
    SELECT Task 
    FROM
    (
      SELECT Task FROM DependentTasks(_Tasks, _WantParents, _WantChildren)
      UNION ALL SELECT unnest(_Tasks) WHERE NOT _WantParents
    ) q
    JOIN Task t USING(Task) 
    WHERE 
      t.RunSchedule IS NULL 
      OR Extract('dow' FROM now()) = ANY(t.RunSchedule)
  )
  INSERT INTO TaskChainTask(
    TaskChainId,
    Environment,
    Database,
    NodeId,
    Task,
    SubTask,
    Eligible)
  SELECT 
    _TaskChainId,
    _Environment,
    Database,
    NodeId,
    Task,

    /* subtask is defined '' by convention, as we can't insert NULL */
    '',
    /*
     * regular tasks are entered eligible if they have no parent dependencies
     * in the chain
     */
    NOT EXISTS (
      SELECT 1 FROM tasks
      JOIN TaskDependency td ON 
        td.ChildTask = t.Task
        AND td.ParentTask = tasks.Task
    )
  FROM Task t
  JOIN Database USING(Database)
  WHERE
    t.Task IN (
      SELECT Task FROM tasks
    )
    AND Environment = _Environment;

  INSERT INTO TaskChainTask(
    TaskChainId,
    Environment,
    Database,
    NodeId,
    Task,
    SubTask,
    Eligible)
  SELECT
    _TaskChainId,
    _Environment,
    Database,
    NodeId,
    Task,
    SubTask,
    false /* subtasks are always intialized false */
  FROM SubTask st
  JOIN Task t USING(Task)
  JOIN Database d USING(Database)
  WHERE
    t.Task IN (
      SELECT Task FROM DependentTasks(_Tasks, _WantParents, _WantChildren)
      UNION ALL SELECT unnest(_Tasks) WHERE NOT _WantParents
    )
    AND (t.RunSchedule IS NULL
      OR Extract('dow' FROM now()) IN (SELECT unnest(t.RunSchedule))
    )
    AND (
      st.RunSchedule iS NULL
      OR Extract ('dow' FROM now()) IN (select unnest(st.RunSchedule))
    )

    AND Environment = _Environment;

  RETURN _TaskChainId;
END;
$$ LANGUAGE PLPGSQL;

/* non array variant of CreateTaskChain */
CREATE OR REPLACE FUNCTION CreateTaskChainFromTask(
  _Task TEXT,
  _Environment TEXT DEFAULT 'Production',
  _WantParents BOOL DEFAULT true,
  _WantChildren BOOL DEFAULT true,
  _ForceConcurrencyModel ConcurrencyModel_t DEFAULT NULL,
  _User TEXT DEFAULT 'System') RETURNS BIGINT AS
$$
  SELECT CreateTaskChain(
    array[_Task],
    _Environment,
    _WantParents,
    _WantChildren,
    _User)
$$ LANGUAGE SQL;


/* Creates a task chain from a named configuration */
CREATE OR REPLACE FUNCTION CreateTaskChain(
  _TaskChainConfigurationName TEXT,
  _Environment TEXT DEFAULT 'Production',
  _WantParents BOOL DEFAULT true,
  _WantChildren BOOL DEFAULT true,
  _ForceConcurrencyModel ConcurrencyModel_t DEFAULT NULL,
  _User TEXT DEFAULT 'System') RETURNS BIGINT AS
$$
  SELECT CreateTaskChain(
    array(
      SELECT Task
      FROM TaskChainConfigurationTask
      WHERE TaskChainConfigurationName = _TaskChainConfigurationName
    ),
    _Environment,
    _WantParents,
    _WantChildren,
    _TaskChainConfigurationName,
    _ForceConcurrencyModel,
    _User)
  FROM TaskChainConfiguration
  WHERE TaskChainConfigurationName = _TaskChainConfigurationName;
$$ LANGUAGE SQL;


CREATE OR REPLACE FUNCTION FinishTaskChain(
  _TaskChainId BIGINT) RETURNS VOID AS
$$
DECLARE
  r RECORD;
BEGIN
  UPDATE TaskChain SET Finished = clock_timestamp()
  WHERE TaskChainId = _TaskChainId;
END;
$$ LANGUAGE PLPGSQL;


CREATE OR REPLACE FUNCTION DependentTasks(
  _Tasks TEXT[],
  _WantParents BOOL DEFAULT false,
  _WantChildren BOOL DEFAULT true,
  Task OUT TEXT,
  CanRun OUT BOOL,
  AllNodes OUT BOOL) RETURNS SETOF RECORD AS
$$
  WITH RECURSIVE c AS
  (
    SELECT
      t.Task,
      NULL::TEXT AS ParentTask,
      NULL::TEXT AS ChildTask,
      NULL::BOOL AS CanRun,
      NULL::BOOL AS AllNodes
    FROM Task t
    WHERE
      Task = any(_Tasks)
      AND _WantParents
    UNION ALL SELECT
      p.Task,
      d.ParentTask,
      d.ChildTask,
      d.CanRun,
      d.AllNodes
    FROM c
    JOIN TaskDependency d ON d.ChildTask = c.Task
    JOIN Task p ON p.Task = d.ParentTask
  ),
  p AS
  (
    SELECT
      t.Task,
      NULL::TEXT AS ParentTask,
      NULL::TEXT AS ChildTask,
      NULL::BOOL AS CanRun,
      NULL::BOOL AS AllNodes
    FROM Task t
    WHERE
      Task = any(_Tasks)
      AND _WantChildren
    UNION ALL SELECT
      c.Task,
      d.ParentTask,
      d.ChildTask,
      d.CanRun,
      /* went travelling downstream, the all nodes bit carries so that
       * once you've seen it in the tree, it is always true.  The idea is that
       * a cross node dependency, once identified, must always be honored.
       */
      COALESCE(p.AllNodes, false) OR d.AllNodes
    FROM p
    JOIN TaskDependency d ON d.ParentTask = p.Task
    JOIN Task c ON c.Task = d.ChildTask
  )
  SELECT DISTINCT
    Task,
    CanRun,
    AllNodes
  FROM
  (
    SELECT * FROM p
    /* children does not include self */
    WHERE Task NOT IN (SELECT Unnest(_Tasks))
    /* parents includes self */
    UNION ALL SELECT * FROM c
  ) q;
$$ LANGUAGE SQL;


CREATE OR REPLACE FUNCTION DependentTasks(
  _Task TEXT,
  _WantParents BOOL DEFAULT false,
  _WantChildren BOOL DEFAULT true,
  Task OUT TEXT,
  CanRun OUT BOOL,
  AllNodes OUT BOOL) RETURNS SETOF RECORD AS
$$
  SELECT DependentTasks(array[_Task], _WantParents, _WantChildren);
$$ LANGUAGE SQL;


CREATE OR REPLACE FUNCTION Duration(_i INTERVAL) RETURNS TEXT AS
$$
  SELECT
  CASE WHEN _i IS NOT NULL THEN
    format('%sh %s',
      floor(extract('epoch' from _i) / 3600),
      to_char(_i, 'MIm SSs'))
  ELSE NULL END;
$$ LANGUAGE SQL;


/* Internal cancel function. This can only be called by the same process
 * running main() because only the process running dblink connections can
 * cancel them.
 */
CREATE OR REPLACE FUNCTION CancelTaskChain(
  _TaskChainId BIGINT,
  _Pause BOOL) RETURNS VOID AS
$$
DECLARE
  q TEXT;
  n RECORD;
  sc ServerConnection;

  _ConnectionNames TEXT[];

  gc GlobalControl;
BEGIN
  SELECT INTO gc * FROM GlobalControl;

  IF pg_backend_pid() != gc.MainProcessPid
  THEN
    PERFORM Log('ERROR',
      'This function must be called from main process. Call Cancel()');
  END IF;

  PERFORM Log(format('Cancelling task chain id %s for %s',
    _TaskChainId,
    CASE WHEN _Pause THEN 'Pause' ELSE 'Administrative Cancel' END));

  FOR sc IN
    SELECT *
    FROM ServerConnection
    WHERE
      Active
      AND TaskChainTaskId IN (
        SELECT tct.TaskChainTaskId
        FROM TaskChainTask tct
        WHERE TaskChainId = _TaskChainId
      )
  LOOP
    /* connections are now ready to go */
    PERFORM TaskFinished(
      sc.TaskChainTaskId,
      CASE WHEN NOT _Pause THEN 'FAILED_VIA_CANCEL' END::TaskResult_t,
      CASE WHEN NOT _Pause THEN
        'Cancelled while in progress due to administrative cancel' END,
      _Pause);
  END LOOP;
END;
$$ LANGUAGE PLPGSQL;


/* Cancel all running tasks (optionally to a specific chain) and pause
 * execution.  All chains so paused must be so marked.
 *
 * The list of paused task chains will be returned; if a task chain is not
 * running it will not be returned.
 */
CREATE OR REPLACE FUNCTION Pause(
  _TaskChainId BIGINT DEFAULT NULL,
  _User TEXT DEFAULT 'System',
  TaskChainId OUT BIGINT) RETURNS SETOF BIGINT AS
$$
BEGIN
  PERFORM Log(format(
    'User %s is pausing %s',
    _User,
    CASE WHEN _TaskChainId IS NULL
      THEN 'all task chains'
      ELSE 'task chain ' || _TaskChainId
    END));

  RETURN QUERY UPDATE TaskChain tc SET PendingAction = 'PAUSE'
  WHERE
    (tc.TaskChainId = _TaskChainId OR _TaskChainId IS NULL)
    AND Finished IS NULL
  RETURNING tc.TaskChainId;
END;
$$ LANGUAGE PLPGSQL;

/* Resume all paused task chains or a specific paused task chain.
 *
 * The list of resumed task chains will be returned; if a task chain is not
 * considered able to be resumed it will not be returned.
 */
CREATE OR REPLACE FUNCTION Resume(
  _TaskChainId BIGINT DEFAULT NULL,
  _User TEXT DEFAULT 'System',
  TaskChainId OUT BIGINT) RETURNS SETOF BIGINT AS
$$
BEGIN
  PERFORM Log(format(
    'User %s is resuming %s',
    _User,
    CASE WHEN _TaskChainId IS NULL
      THEN 'all task chains'
      ELSE 'task chain ' || _TaskChainId
    END));

  RETURN QUERY UPDATE TaskChain tc SET
    Paused = false,
    PauseEndTime = clock_timestamp()
  WHERE
    (
      tc.TaskChainId = _TaskChainId
      OR _TaskChainId IS NULL
    )
    AND Paused = true
    AND Finished IS NULL
  RETURNING tc.TaskChainId;
END;
$$ LANGUAGE PLPGSQL;


/* Cancel all running task chains or a specific running task chain.
 *
 * The list of cancelled task chains will be returned; if a task chain is not
 * considered able to be cancelled it will not be returned.
 *
 * This will only work correctly if called from the procedure which owns the
 * connections being cancelled.
 */
CREATE OR REPLACE FUNCTION Cancel(
  _TaskChainId BIGINT DEFAULT NULL,
  _User TEXT DEFAULT 'System',
  TaskChainId OUT BIGINT) RETURNS SETOF BIGINT AS
$$
BEGIN
  PERFORM Log(format(
    'User %s is cancelling %s',
    _User,
    CASE WHEN _TaskChainId IS NULL
      THEN 'all task chains'
      ELSE 'task chain ' || _TaskChainId
    END));

  RETURN QUERY UPDATE TaskChain tc SET PendingAction = 'CANCEL'
  WHERE
    (tc.TaskChainId = _TaskChainId OR _TaskChainId IS NULL)
    AND Finished IS NULL
  RETURNING tc.TaskChainId;
END;
$$ LANGUAGE PLPGSQL;

/* Is it appropriate to run the task at this time? */
CREATE OR REPLACE FUNCTION TimeEligible(
  _StartTime TIME,
  _EndTime TIME,
  _CurrentTime TIME DEFAULT now()::TIME) RETURNS BOOL AS
$$
  SELECT
    CASE
      WHEN _StartTime IS NULL AND _EndTime IS NULL
        THEN true
      WHEN _StartTime IS NOT NULL AND _EndTime IS NULL
        THEN _CurrentTime >= _StartTime
      WHEN _StartTime IS NULL AND _EndTime IS NOT NULL
        THEN _CurrentTime <= _EndTime
      WHEN _StartTime = _EndTime
        THEN false
      WHEN _StartTime < _EndTime
        THEN _CurrentTime BETWEEN _StartTime AND _EndTime
      WHEN _StartTime > _EndTime
        THEN _CurrentTime > _StartTime OR _CurrentTime < _EndTime
    END
$$ LANGUAGE SQL IMMUTABLE;




/* Returns task to run so that all dependencies (other tasks, time of day etc)
 * must met.  If there tasks ready for execution, the one with the earliest
 * entered time will be returned.
 *
 * NULL is returned if there is nothing to do.
 */
CREATE OR REPLACE FUNCTION GetTask(
  TaskChainTaskId OUT BIGINT) RETURNS BIGINT AS
$$
BEGIN
  SELECT INTO GetTask.TaskChainTaskId tct.TaskChainTaskId
  FROM TaskChainTask tct
  JOIN Database USING(Environment, Database, NodeId)
  JOIN TaskChain tc USING(TaskChainId)
  JOIN vw_Server USING (Server)
  JOIN Task t USING(Task)
  WHERE
    /* Are we eligible for dependency reasons? */
    Eligible

    AND ActiveConnections < TaskClientThreads

    /* Task chain must be qualifying */
    AND tc.Started IS NOT NULL
    AND tc.Finished IS NULL
    AND NOT tc.Paused

    /* Don't return any tasks for chains with pending administrative action */
    AND tc.PendingAction IS NULL

    /* Is the task eligible to run at this time? */
    AND TimeEligible(t.StartTime, t.EndTime)

    /* The task must not already have started... */
    AND tct.Started IS NULL
    AND tct.Finished IS NULL
  ORDER BY 
    tct.Yielded NULLS FIRST,
    t.TaskOrder,
    tct.TaskChainTaskId LIMIT 1;
END;
$$ LANGUAGE PLPGSQL;


CREATE OR REPLACE FUNCTION GetTasks(
    TaskChainTaskId OUT BIGINT,
    _YieldTime INTERVAL DEFAULT '15 minutes'::INTERVAL) RETURNS SETOF BIGINT AS
$$
BEGIN
  RETURN QUERY
  SELECT q.TaskChainTaskId
  FROM
  (
    SELECT
      tct.TaskChainTaskId,
      TaskClientThreads - ActiveConnections AS AvailableThreads,
      row_number() OVER (
        PARTITION BY Server
        ORDER BY COALESCE(Parent.Started, tct.Started)) AS Count
    FROM TaskChainTask tct
    JOIN Database USING(Environment, Database, NodeId)
    JOIN TaskChain tc USING(TaskChainId)
    JOIN vw_Server USING (Server)
    JOIN Task t USING(Task)
    LEFT JOIN TaskChainTask parent ON
      tct.SubTask != ''
      AND parent.Subtask = ''
      AND parent.Task = tct.Task
      AND parent.TaskChainId = tct.TaskChainId
      AND parent.Environment = tct.Environment
      AND parent.Database = tct.Database
      AND parent.NodeId = tct.NodeId
    WHERE
      /* Are we eligible for dependency reasons? */
      tct.Eligible

      /* This is on off switch at server level */
      AND ActiveConnections < TaskClientThreads

      /* Task chain must be qualifying */
      AND tc.Started IS NOT NULL
      AND tc.Finished IS NULL
      AND NOT tc.Paused

      AND (tct.Yielded IS NULL OR tct.Yielded < now() - _YieldTime)

      /* Don't return any tasks for chains with pending administrative action */
      AND tc.PendingAction IS NULL

      /* Is the task eligible to run at this time? */
      AND TimeEligible(t.StartTime, t.EndTime)

      /* The task must not already have started... */
      AND tct.Started IS NULL
      AND tct.Finished IS NULL
  ) q
  WHERE Count <= AvailableThreads;
END;
$$ LANGUAGE PLPGSQL;     



/*
 * Runs a task.  Returns true if generally anything interesting happened;
 * the task was run or there was some kind of failure.  If no connection was
 * available, (this may happen if the server the task is trying to get a
 * connection to is completely down), false is returned to inform the main
 * stored procedure loop to chill out for a bit
 */
CREATE OR REPLACE FUNCTION RunTask(
  _TaskChainTaskId BIGINT) RETURNS BOOL AS
$$
DECLARE
  tct TaskChainTask;
  r INT;
  _ConnectionName TEXT;

  gc GlobalControl;

  _TaskBody TEXT;

  _d RECORD;
  _t RECORD;

  _ErrorMessage TEXT;

BEGIN
  PERFORM Log(format('Starting Task id %s', _TaskChainTaskId));

  SELECT INTO gc * FROM GlobalControl;

  SELECT INTO tct * FROM TaskChainTask
  WHERE TaskChainTaskId = _TaskChainTaskId;

  IF tct.SubTask = ''
  THEN
    SELECT INTO _TaskBody TaskBody FROM Task WHERE Task = tct.Task;
  ELSE
    IF  COALESCE(tct.TaskBody,'') = '' THEN
      SELECT INTO _TaskBody TaskBody
      FROM SubTask
      WHERE
        Task = tct.Task
        AND SubTask = tct.SubTask;
    ELSE
    /* if subtask was dynamically created sub taskbody will be in taskchaintask */
      _TaskBody := tct.TaskBody;
    END IF;
  END IF;

  IF NOT FOUND
  THEN
    PERFORM Log('ERROR', format(
      'Could not resolve Task %s%s for ID %s',
      tct.Task,
      CASE WHEN tct.SubTask != '' THEN '/' || tct.SubTask END,
      _TaskChainTaskId));
  END IF;

  IF tct.TaskChainTaskId IS NULL
  THEN
    PERFORM Log('ERROR', format('Tried to run invalid id %s', _TaskChainTaskId));
  END IF;

  IF _TaskBody = '' AND tct.SubTask = ''
  THEN
    /* Parent task -- NOP.  Just mark started. */
    PERFORM TaskStarted(_TaskChainTaskId, FALSE);

    /* if there are no sub tasks, immediately finish the task */
    IF NOT EXISTS (
      SELECT 1 FROM TaskChainTask s
      WHERE
        s.TaskChainId = tct.TaskChainId
        AND s.Task = tct.Task
        AND s.SubTask != '')
    THEN
      PERFORM TaskFinished(
        tct.TaskChainTaskId,
        'SUCCESS_EMPTY',
        'Auto finishing task with no subtasks');
    END IF;

    RETURN TRUE;
  END IF;

  PERFORM Log(format(
    'Running task %s%s id %s Environment %s Database %s NodeId %s',
    tct.Task,
    CASE WHEN tct.SubTask != '' THEN '/' || tct.SubTask END,
    _TaskChainTaskId,
    tct.Environment,
    tct.Database,
    tct.NodeId));

  /* Get a connection to the database */
  _ConnectionName := DblinkConnection(
    tct.Environment,
    tct.Database,
    tct.NodeId);

  IF _ConnectionName IS NULL
  THEN
    _ErrorMessage := format(
      'Could not get connection to database %s node %s for id %s',
      tct.Database, tct.NodeId, _TaskChainTaskId); 
    PERFORM Log('WARNING', _ErrorMessage);

  /* in case a connection is not available, YIELD the task for later execution by 
   * by appending [PGTASK.YIELD] keyword in the errormessage , this will prevent the 
   * mainloop from getting same task instead of processing other pending eligible tasks
   */
    PERFORM TaskFinished(
      tct.TaskChainTaskId,
      'SUCCESS_EMPTY',
      format('[PGTASK.YIELD] %s', _ErrorMessage));  
    RETURN FALSE;
  END IF;

  UPDATE ServerConnection SET TaskChainTaskId = _TaskChainTaskId
  WHERE
    ConnectionName = _ConnectionName;

  SELECT INTO _d
    LogMinMessages,
    LogErrorVerbosity
  FROM Database d
  WHERE d.Database = tct.Database
    AND d.Environment = tct.Environment
    AND d.NodeId = tct.NodeId
  LIMIT 1;

  SELECT INTO _t
    LogMinMessages,
    LogErrorVerbosity
  FROM Task t
  WHERE t.task = tct.task;

  PERFORM Log('DEBUG', format('Got connection %s', _ConnectionName), gc.Debug);

  /* run it */
  BEGIN
    SELECT INTO r dblink_send_query(
      _ConnectionName,
      FormatTaskQuery(
        CASE WHEN gc.Debug THEN gc.DebugQuery ELSE _TaskBody END,
        tct.Task,
        tct.SubTask,
        _TaskChainTaskId,
        gc.TaskMinMessages,
        COALESCE(_t.LogMinMessages, _d.LogMinMessages, gc.LogMinMessages, 'NOTICE'),
        COALESCE(_t.LogErrorVerbosity, _d.LogErrorVerbosity, gc.LogErrorVerbosity, 'TERSE')));
    IF r = 0
    THEN
      PERFORM Log('WARNING',
        format('Failed to send query for id %s', _TaskChainTaskId));

      RETURN FALSE;
    END IF;

  EXCEPTION WHEN OTHERS THEN
    PERFORM Log('WARNING', format(
      'Got %s while attempting to run task id %s',
      SQLERRM,
      _TaskChainTaskId));

    RETURN FALSE;
  END;

  /* Mark task started */
  PERFORM TaskStarted(_TaskChainTaskId);

  UPDATE ServerConnection SET TaskChainTaskId = _TaskChainTaskId
  WHERE ConnectionName = _ConnectionName;

  RETURN TRUE;

END;
$$ LANGUAGE PLPGSQL;


/*
 * Mark a task as having been started.
 */
CREATE OR REPLACE FUNCTION TaskStarted(
  _TaskChainTaskId BIGINT,
  _MarkConnection BOOL DEFAULT true) RETURNS VOID AS
$$
DECLARE
  gc GlobalControl;
BEGIN
  SELECT INTO gc * FROM GlobalControl;

  UPDATE TaskChainTask SET Started = clock_timestamp()
  WHERE
    TaskChainTaskId = _TaskChainTaskId
    AND Started IS NULL;

  /* Mark the connection the task is attached to as active, but only if this is
   * a task that requires actually doing something (that is, not a task that
   * contains subtasks).
   */
  IF _MarkConnection
  THEN
    UPDATE ServerConnection SET Active = true
    WHERE TaskChainTaskId = _TaskChainTaskId;
  END IF;
END;
$$ LANGUAGE PLPGSQL;


/*
 * Marks a task as finished.  All dependent tasks are cancelled on the node
 * for canRun = false.  Task failures can travel through dependencies across
 * nodes if and only if the can node flag is set.
 *
 * In a pause scenario, the task is not finished, however the connection is
 * released.
 *
 */
CREATE OR REPLACE FUNCTION TasksFinished(
  _TaskChainTaskIds BIGINT[],
  _TaskResult TaskResult_t,
  _ErrorMessage TEXT,
  _Pause BOOL DEFAULT FALSE) RETURNS VOID AS
$$
DECLARE
  _TasksLeft INT;
  _Found BOOL;
  _Finished TIMESTAMPTZ;
  _TaskChainComplete BOOL;
  t TaskChainTask[];
  _FinishedString TEXT;
  /* if a task was yielded by the source for some reason then 
   *  the task is pushed for later execution
   */
  _Yield BOOL := FALSE;

  gc GlobalControl;
  _ActiveTasks BIGINT[];
BEGIN
  SELECT INTO gc * FROM GlobalControl;

  /* save off the tasks to cancel now, so that we can clear the cancel flag
   * before the cancel is thrown (otherwise we get repeated processing of the
   * of cancellation event).
   */
  SELECT INTO _ActiveTasks array_agg(TaskChainTaskId)
  FROM ServerConnection
  WHERE
    TaskChainTaskId = ANY(_TaskChainTaskIds)
    AND Active;

  /* Mark the connection the task is attached to as inactive */
  UPDATE ServerConnection SET Active = false
  WHERE TaskChainTaskId = ANY(_ActiveTasks);

  /* Check if the finished task was actually yielded, if it was 
   * yielded then reset task status so it can be called again
   */
  _Yield := (_ErrorMessage LIKE '%[PGTASK.YIELD]%');
  
  IF _Yield THEN
    /* reset status of Task if it was yielded */
    WITH tct AS
    (
      UPDATE TaskChainTask SET
        Started = NULL,
        Yielded = CASE WHEN _Yield THEN clock_timestamp() END
      WHERE
        TaskChainTaskId = ANY(_TaskChainTaskIds)
        AND TaskResult IS NULL
      RETURNING *
    )
    SELECT INTO t array_agg(tct) FROM tct;
  ELSE 
    WITH tct AS
    (
      UPDATE TaskChainTask SET
        Started = 
          CASE WHEN NOT _Pause THEN COALESCE(Started, clock_timestamp()) END,
        Finished = CASE WHEN NOT _Pause THEN clock_timestamp() END,
        TaskResult = CASE WHEN NOT _Pause  THEN _TaskResult END,
        ErrorMessage = CASE WHEN NOT _Pause  THEN _ErrorMessage END
      WHERE
        TaskChainTaskId = ANY(_TaskChainTaskIds)
        AND TaskResult IS NULL
      RETURNING *
    )
    SELECT INTO t array_agg(tct) FROM tct;
  END IF;

  /* log finshed tasks.  If all tasks were determined to be cancelled for
   * other reasons (typically due to a admin cancel chaining into a query
   * cancel), do nothing.
   */
  IF array_upper(t, 1) > 0
  THEN
    SELECT INTO _FinishedString
      CASE
        WHEN array_upper(t, 1) = 1
          THEN format(
            'Task %s%s id %s Environment %s Database %s Node %s',
            t[1].Task,
            CASE WHEN t[1].SubTask != '' THEN '/' || t[1].SubTask END,
            t[1].TaskChainTaskId,
            t[1].Environment,
            t[1].Database,
            t[1].NodeId)
          ELSE format('%s tasks', array_upper(t, 1))
      END;

    PERFORM Log(format(
      '%s %s Result %s Message %s',
      CASE WHEN _Pause THEN 'Pausing' WHEN _Yield THEN 'Yielding' ELSE 'Finishing' END,
      _FinishedString,
      _TaskResult,
      _ErrorMessage));
  END IF;

  IF ShouldCancel(_TaskResult) OR _Pause
  THEN
    /* There are scenarios where the task may still be running when an attempt
     * is made to finish it e.g. cancelling situations
     */
    PERFORM CancelTask(ConnectionName)

    FROM ServerConnection
    WHERE
      TaskChainTaskId = ANY(_ActiveTasks);
  END IF;

  IF NOT _Pause
  THEN
      /* are there any tasks that have yet to finish? */
    SELECT INTO _TaskChainComplete NOT EXISTS
      (
        SELECT 1 FROM TaskChainTask
        WHERE
          Finished IS NULL
          AND TaskChainId = t[1].TaskChainId
      );

    PERFORM FinishTaskChain(t[1].TaskChainId)
    WHERE _TaskChainComplete;
  END IF;
END;
$$ LANGUAGE PLPGSQL;

CREATE OR REPLACE FUNCTION TaskFinished(
  _TaskChainTaskId BIGINT,
  _TaskResult TaskResult_t,
  _ErrorMessage TEXT,
  _Pause BOOL DEFAULT FALSE) RETURNS VOID AS
$$
  SELECT TasksFinished(
    array[_TaskChainTaskId],
    _TaskResult,
    _ErrorMessage,
    _Pause);
$$ LANGUAGE SQL;

/*
 * Function checks whether the task chain is paused
 */
CREATE OR REPLACE FUNCTION Paused(_TaskChainId BIGINT) RETURNS BOOL AS
$$
BEGIN
  RETURN (SELECT IsPaused FROM TaskChain WHERE TaskChainId = _TaskChainId);
END;
$$ LANGUAGE PLPGSQL;


/* Welp, sometimes things don't work out as we've expected.  Upon start up,
 * check and see if there are any tasks running in any server/database
 * this pgtask instance is coordinating.  If there are, destroy them as the
 * expectation is that they are running untracked.
 */
CREATE OR REPLACE FUNCTION NukeRunningTasks() RETURNS VOID AS
$$
DECLARE
  d Database;
  _q TEXT;
BEGIN
  PERFORM Log('Cleaning possibly bad pgtask connections');
  _q := format($q$
    SELECT pg_terminate_backend(pid)
    FROM pg_stat_activity
    WHERE
      query LIKE '%%%s%%'
      AND pid != pg_backend_pid()
    $q$, (SELECT ServerId FROM GlobalControl));

  FOR d IN
     SELECT * FROM Database
  LOOP
    BEGIN
      PERFORM Log(format(
        'Cleaning pgtask connections for database %s environment %s node %s',
        d.Database,
        d.Environment,
        d.NodeId));

      /* paranoia: kill any running pgtask queries */
      PERFORM * FROM dblink(
      d.ConnectionString,
      _q ) AS R(V TEXT);
    EXCEPTION WHEN OTHERS THEN
      PERFORM Log(format(
        'Got %s when attempting to kill connections on databsae %s'
        ' environment %s node %s',
        SQLERRM,
        d.Database,
        d.Environment,
        d.NodeId));
    END;
  END LOOP;
END;
$$ LANGUAGE PLPGSQL;

CREATE OR REPLACE FUNCTION CreateDblinkConnections() RETURNS VOID AS
$$
BEGIN
  PERFORM Log('Initializing Connections');
  /* There is no confidence in tracked server connections across startup */
  TRUNCATE ServerConnection;

  PERFORM dblink_disconnect(c) FROM unnest(dblink_get_connections()) c;

  INSERT INTO ServerConnection(
    Server,
    ConnectionNumber,
    ConnectionName,
    Active,
    Cancelling)
  SELECT
    Server,
    Number,
    format('%s-%s', Server, Number),
    false,
    false
  FROM
  (
    SELECT
      Server,

      /* SRF in tlist is dodgy but works in this simple case */
      generate_series(1, TaskClientThreads) AS Number
    FROM Server
  ) q;

  PERFORM NukeRunningTasks();
END;
$$ LANGUAGE PLPGSQL;


CREATE OR REPLACE FUNCTION VerifyDblinkConnection(
  _Environment TEXT,
  _Database TEXT,
  _NodeId INT,
  ConnectionName INOUT TEXT) RETURNS TEXT AS
$$
BEGIN
  PERFORM Log(format(
    'Testing connection %s to environment %s database %s NodeId %s',
    ConnectionName,
    _Environment,
    _Database,
    _NodeId));

  PERFORM * FROM dblink(ConnectionName, 'SELECT 0') AS R(v int);

EXCEPTION WHEN OTHERS THEN
  /* uh oh. Try to reconnect. */
  RAISE WARNING
    'Got % when attempting to use connection % to environment % database % nodeid %',
    SQLERRM,
    ConnectionName,
    _Environment,
    _Database,
    _NodeId;

  BEGIN
    PERFORM dblink_disconnect(ConnectionName);

  EXCEPTION WHEN OTHERS THEN
    RAISE WARNING
      'Got % when attempting to disconnect connection % from environment % database % nodeid %',
      SQLERRM,
      ConnectionName,
      _Environment,
      _Database,
      _NodeId;

    BEGIN
      PERFORM dblink_connect(ConnectionName,
        CASE WHEN ServerType = 'POSTGRES'::ServerType_t
          THEN ConnectionString
          ELSE ProxyConnectionString()
        END )
      FROM Database
      JOIN Server USING(Server)
      WHERE
        Environment = _Environment
        AND Database = _Database
        AND NodeId = _NodeId;

      PERFORM Log('WARNING', format(
        'Reestablished connection %s to environment %s database %s nodeid %s',
        ConnectionName,
        _Environment,
        _Database,
        _NodeId));

    EXCEPTION WHEN OTHERS THEN
      PERFORM Log('WARNING', format(
        'Punting: Got %s when attempting to disconnect connection %s '
        'from environment %s database %s nodeid %s',
        SQLERRM,
        ConnectionName,
        _Environment,
        _Database,
        _NodeId));

      /* Could not reconnect.  Mark the connection is not active and punt */
      UPDATE ServerConnection sc SET
        Environment = NULL,
        Database = NULL,
        NodeId = NULL,
        Active = false
      WHERE sc.ConnectionName = VerifyDblinkConnection.ConnectionName;

      ConnectionName := NULL;

      RETURN;
    END;
  END;
END;
$$ LANGUAGE PLPGSQL;



CREATE OR REPLACE FUNCTION SetKeepAlives() RETURNS TRIGGER AS
$$
BEGIN
  /* is this postgres databse? or are keepalives already set? */
  PERFORM 1 FROM Server 
  WHERE 
    Server = NEW.Server
    AND ServerType = 'POSTGRES'
    AND NOT NEW.ConnectionString ~* 'keepalive';
  
  IF NOT FOUND
  THEN
    RETURN NEW;
  END IF;
  
  NEW.ConnectionString := NEW.ConnectionString || 
    ' keepalives_idle=300 keepalives_interval=5 keepalives_count=5';
  
  RETURN NEW;
END;
$$ LANGUAGE PLPGSQL;

CREATE TRIGGER OnDatabaseChange 
  BEFORE INSERT OR UPDATE ON Database FOR EACH ROW
  EXECUTE PROCEDURE SetKeepAlives();

CREATE OR REPLACE FUNCTION ProxyConnectionString() RETURNS TEXT AS
$$
  SELECT format(
    'host=localhost dbname=%s user=%s',
    current_database(),
    current_user)
$$ LANGUAGE SQL;

/* Fetch a connection from the server pool.  Prefer a conenction that is
 * already connected to the database but if necessary reconnect to the
 * database if one is not available already connected.
 *
 * If the connection is already established, Test it. If the test fails,
 * mark the database invalid.
 *
 * Returns connection name.  If no connection is available, returns NULL
 */
CREATE OR REPLACE FUNCTION DblinkConnection(
  _Environment TEXT,
  _Database TEXT,
  _NodeId INT,
  ConnectionName OUT TEXT) RETURNS TEXT AS
$$
DECLARE
  _Connected BOOL;
BEGIN
  /* is there a connection available that is already pinned to the database? */
  SELECT
    sc.ConnectionName
    INTO DblinkConnection.ConnectionName
  FROM ServerConnection sc
  WHERE
    Environment = _Environment
    AND Database = _Database
    AND NodeId = _NodeId
    AND NOT Active
    AND NOT Cancelling
  LIMIT 1;

  IF NOT FOUND
  THEN
    /*
     * no existing connections available. look for free connection on the
     * database's server.
     */
    SELECT
      sc.ConnectionName,
      Database IS NOT NULL
    INTO
      DblinkConnection.ConnectionName,
      _Connected
    FROM ServerConnection sc
    WHERE
      Server = (
        SELECT Server
        FROM Database d
        WHERE
          d.Environment = _Environment
          AND d.Database = _Database
          AND d.NodeId = _NodeId)
      AND NOT Active
      AND NOT Cancelling
    ORDER BY random()
    LIMIT 1;

    IF NOT FOUND
    THEN
      /* no connections are available.  This is not supposed to happen but
       * nothing to do but return NULL and punt.
       */
      PERFORM Log(
        'WARNING',
        format(
          'Not able to acquire connection to environment %s database %s Node %s',
          _Environment,
          _Database,
          _NodeId));

      RETURN;
    END IF;

    BEGIN
      /* There is a connection available that is not attached to the database
       * we want.  If it is connected to another database, disconnect it and
       * connect to the correct database
       */
      IF _Connected
      THEN
        PERFORM dblink_disconnect(ConnectionName);
      END IF;

      PERFORM dblink_connect(ConnectionName,
        CASE WHEN ServerType = 'POSTGRES'::ServerType_t
          THEN ConnectionString
          ELSE ProxyConnectionString()
        END)
      FROM Database
      JOIN Server USING(Server)
      WHERE
        Environment = _Environment
        AND Database = _Database
        AND NodeId = _NodeId;

    EXCEPTION WHEN OTHERS THEN
      PERFORM Log('WARNING', format(
        'Got %s when attempting to reconnect connection %s'
        ' to environment %s database %s node %s',
        SQLERRM,
        ConnectionName,
        _Environment,
        _Database,
        _NodeId));

      UPDATE ServerConnection sc SET
        Environment = NULL,
        Database = NULL,
        NodeId = NULL
      WHERE sc.ConnectionName = DblinkConnection.ConnectionName;

      ConnectionName := NULL;

      RETURN;
    END;
  END IF;

  ConnectionName := VerifyDblinkConnection(
    _Environment,
    _Database,
    _NodeId,
    ConnectionName);

  UPDATE ServerConnection sc SET
    Environment = _Environment,
    Database = _Database,
    NodeId = _NodeId
  WHERE
    sc.ConnectionName = DblinkConnection.ConnectionName
    AND DblinkConnection.ConnectionName IS NOT NULL;
END;
$$ LANGUAGE PLPGSQL;


/* wrap a task with do..execute so that we don't have to worry about columns.
 *
 * For a complicated set of reasons, client_min_messages, not log_min_messages,
 * controls how pgtask logs task output to the local database log.  The default
 * client_min_messags is 'NOTICE', which may tend to be overly verbose by
 * default.  We therefore grab the setting here and invoke it as part of task
 * execution.
 */
DROP FUNCTION IF EXISTS  FormatTaskQuery(TEXT, TEXT, TEXT, BIGINT, TEXT);
CREATE OR REPLACE FUNCTION FormatTaskQuery(
  _TaskBody TEXT,
  _Task TEXT,
  _SubTask TEXT,
  _TaskChainTaskId BIGINT,
  _TaskMinMessages TEXT,
  _TaskLogMinMessage TEXT,
  _TaskLogErrorVerbosity TEXT) RETURNS TEXT AS
$$
  SELECT format(
    E'DO /* running from pgtask instance %s for task %s%s task id [%s] */'
    ' \n$pgtask$\nBEGIN\n'
    '  DISCARD TEMP;\n'
    '  SET LOCAL client_min_messages = %s;'
    '  SET LOCAL log_min_messages = %s;'
    '  SET LOCAL log_error_verbosity = %s;'
    '  EXECUTE\n$pgtaskbody$\n/* Task body starts here */'
    '\n\n%s;\n\n/* Task body ends here */\n$pgtaskbody$;'
    '\nEND;\n$pgtask$',
    (SELECT ServerId FROM GlobalControl),
    _Task,
    CASE WHEN _SubTask != '' THEN '/' || _Subtask END,
    _TaskChainTaskId,
    _TaskMinMessages,
    _TaskLogMinMessage,
    _TaskLogErrorVerbosity,
    _TaskBody);
$$ LANGUAGE SQL IMMUTABLE;


CREATE OR REPLACE FUNCTION Disconnect(_ConnectionName TEXT) RETURNS VOID AS
$$
BEGIN
  UPDATE ServerConnection sc SET
    Environment = NULL,
    Database = NULL,
    NodeId = NULL,
    Active = false,
    Cancelling = false,
    TaskChainTaskId = NULL
  WHERE sc.ConnectionName = _ConnectionName;

  PERFORM dblink_disconnect(_ConnectionName);
END;
$$ LANGUAGE PLPGSQL;


CREATE OR REPLACE FUNCTION ConnectionDead(
  _ConnectionName TEXT) RETURNS VOID AS
$$
BEGIN
  PERFORM Log(format('Marking connection %s dead', _ConnectionName));

  PERFORM TaskFinished(
    TaskChainTaskId,
    'FAILED_VIA_CONNECTION_RESET',
    'Connection to remote server has gone bad')
  FROM ServerConnection
  WHERE ConnectionName = _ConnectionName;

  /* Mark connection bad */
  PERFORM Disconnect(_ConnectionName);
END;
$$ LANGUAGE PLPGSQL;


/* 
 * Check and see if tasks are still running, and, if they are determined to be
 * finished, clean them up.
 */
CREATE OR REPLACE FUNCTION CheckForFinishedTasks(
  DidStuff OUT BOOL) RETURNS BOOL AS
$$
DECLARE
  r RECORD;
  _DidSucceed BOOL;
  _ErrorMessage TEXT;

  _State TaskExecutionState_t;

  _Cancelling BOOL;
  _Active BOOL;

  /* Controls how cancellations are managed.  Values:
   * DELAYED: continue to loop and want for dblink_get_result 
   *          to return as cancelled and reap query normally.  This is the 
   *          best in terms of behavior, but unfortunately, there are now
   *          some bugs (in dblink), that cause dblink to not be able to
   *          pick up the cancellation and so the connection gets hung up
   *          forever.
   * DISCONNECT: send a cancel, but then disconnect the query without waiting.
   * IMMEDIATE: fire an immediate dblink_get_result, waiting for return.
   *            this is the safest, but can hang the main thread.
   */
  _CancelMode TEXT DEFAULT 'DISCONNECT'; 
BEGIN
  DidStuff := false;

  FOR r IN
    SELECT
      sc.*,
      d.ConnectionString,
      COALESCE(parent.Started, tct.Started)
        + MaxRunTime < clock_timestamp() AS Cancel,
      MaxRunTime,
      tct.Task,
      tct.SubTask
    FROM ServerConnection sc
    JOIN Database d USING(Environment, NodeId, Database)
    JOIN TaskChainTask tct USING(TaskChainTaskId)
    JOIN Task t USING(Task)
    /* subtasks will use parent task start task start time for start time since
     * the maximum runtime is established with the parent task
     */
    LEFT JOIN TaskChainTask parent ON
      tct.SubTask != ''
      AND parent.Subtask = ''
      AND parent.Task = tct.Task
      AND parent.TaskChainId = tct.TaskChainId
      AND parent.Environment = tct.Environment
      AND parent.Database = tct.Database
      AND parent.NodeId = tct.NodeId
    WHERE (Active OR Cancelling)
  LOOP
    SELECT INTO
      _Active,
      _Cancelling
      Active,
      Cancelling
    FROM ServerConnection
    WHERE ConnectionName = r.ConnectionName;

    /*
     * Tasks can fail after the cursor is etablished due to triggers.  Recheck
     * the WHERE condition to ensure we have not alreday handled the connection.
     */
    IF NOT _Active AND NOT _Cancelling
    THEN
      CONTINUE;
    END IF;

    _State := TaskRunning(r.ConnectionName);

    IF _State != 'RUNNING'
    THEN
      DidStuff := true;
    END IF;

    IF _State = 'FINISHED' OR 
      (
        _State = 'RUNNING' 
        AND _Cancelling 
        AND _CancelMode = 'IMMEDIATE'
      )
    THEN
      SELECT INTO _State, _ErrorMessage TaskExecutionState, ErrorMessage
      FROM ReapTask(r.ConnectionName);

      IF _Cancelling
      THEN
        PERFORM Log(format(
          'Reaped cancelling task %s/%s (id %s node %s) on connection %s',
          r.Task,
          r.SubTask,
          r.TaskChainTaskId,
          r.NodeId,
          r.ConnectionName));

        UPDATE ServerConnection SET
          Cancelling = false,
          TaskChainTaskId = NULL
        WHERE ConnectionName = r.ConnectionName;
      ELSE
        PERFORM TaskFinished(
          r.TaskChainTaskId,
          CASE _State
            WHEN 'FINISHED'::TaskExecutionState_t THEN 'SUCCESS_COMPLETION'
            WHEN 'FAILED'::TaskExecutionState_t THEN 'FAILED_VIA_EXECUTION'
          END::TaskResult_t,
          _ErrorMessage)
        WHERE _State IN ('FINISHED', 'FAILED');
      END IF;
    ELSIF _State = 'RUNNING' 
      AND _Cancelling 
      AND _CancelMode = 'DISCONNECT'
    THEN
      PERFORM Log(format(
        'Force closed connection on '
        'cancelling task %s/%s (id %s node %s) on connection %s',
        r.Task,
        r.SubTask,
        r.TaskChainTaskId,
        r.NodeId,
        r.ConnectionName));

      PERFORM Disconnect(r.ConnectionName);
    ELSIF r.Cancel AND NOT _Cancelling
    THEN
      _ErrorMessage := format(
        'Cancelling task due to exceeding maximum run time of %s',
        r.MaxRunTime);

      PERFORM TaskFinished(
        r.TaskChainTaskId,
        'FAILED_VIA_TIMEOUT',
        _ErrorMessage);
    END IF;

    IF _State = 'RESET'
    THEN
      IF _Cancelling 
      THEN
        UPDATE ServerConnection SET
          Cancelling = false,
          TaskChainTaskId = NULL
        WHERE ConnectionName = r.ConnectionName;    
      END IF;

      PERFORM Log(format(
        'Got RESET state while checking connection status of %s',
        r.ConnectionName));

      PERFORM ConnectionDead(r.ConnectionName);
    END IF;
  END LOOP;
END;
$$ LANGUAGE PLPGSQL;


CREATE OR REPLACE FUNCTION CheckForAdmnistrativeActions() RETURNS BOOL AS
$$
DECLARE
  _TaskChains TaskChain[];
  _AdminstrativeQuery TEXT;
  _AdminstrativeResult JSON;
BEGIN
  SELECT INTO _AdminstrativeQuery Query
  FROM AdministrativeCommand
  WHERE Consumed IS NULL;

  IF _AdminstrativeQuery IS NOT NULL
  THEN 
    BEGIN 
      PERFORM Log('WARNING', format('Got administrative command %s',
        _AdminstrativeQuery));

      EXECUTE format('SELECT to_json(array_agg(q)) FROM (%s) q',
        _AdminstrativeQuery) INTO _AdminstrativeResult;

      UPDATE AdministrativeCommand SET
        Consumed = clock_timestamp(),
        Result = _AdminstrativeResult,
        ErrorMessage = NULL;
    EXCEPTION WHEN OTHERS THEN
      PERFORM Log('WARNING', format(
        'Got error %s when running administrative command %s',
        SQLERRM,
        _AdminstrativeQuery));

      UPDATE AdministrativeCommand SET
        Consumed = clock_timestamp(),
        Result = NULL,
        ErrorMessage = SQLERRM;
    END;
  END IF;

  /* gather up the task chains to cancel */
  SELECT INTO _TaskChains array_agg(tc) FROM TaskChain tc
  WHERE
    PendingAction IS NOT NULL
    AND Finished IS NULL;

  IF _TaskChains IS NULL OR array_upper(_TaskChains, 1) = 0
  THEN
    RETURN false;
  END IF;

  /* ... and cancel them. */
  PERFORM CancelTaskChain(TaskChainId, PendingAction = 'PAUSE')
  FROM
  (
    SELECT * FROM unnest(_TaskChains) tc
  ) q;

  UPDATE TaskChain tc
  SET
    PendingAction = NULL,
    Finished = CASE WHEN tc.PendingAction = 'CANCEL' THEN clock_timestamp() END,
    ErrorMessage = CASE WHEN tc.PendingAction = 'CANCEL'
      THEN 'Administrative Cancel' END,
    Paused = tc.PendingAction = 'PAUSE',
    PauseStartTime =
      CASE WHEN tc.PendingAction = 'CANCEL' THEN clock_timestamp() END
  WHERE tc.TaskChainId IN (SELECT TaskChainId FROM unnest(_TaskChains));

  RETURN true;
END;
$$ LANGUAGE PLPGSQL;


/*
 * Load task configurations from a structurd json.  As much as possible,
 * the changes will be reconciled in so that RI deletions do not have to be
 * cascaded.
 *
 * XXX: Should loaded configurations be tagged with some some kind of name
 *      so that downstream adjusttments are simpler to make?  this shold
 *      also facilitate saving independent sections of the configuration that
 *      are managed from different areas.
 *
 * XXX: Can this be allowed to run while any impacted task chains are in
 *      progress?
 *
 * XXX: After any database/server changes are made, it's probably a good idea
 *      to flush and reboot any server connections. What if a connection_string
 *      changes?
 */
CREATE OR REPLACE FUNCTION LoadConfiguration(
  _Configuration JSON) RETURNS VOID AS
$$
DECLARE
  r RECORD;
  t JSONB;
BEGIN
  WITH n AS
  (
    SELECT * FROM json_populate_recordset(
      NULL::Environment,
      (_Configuration)->'environments')
  )
  INSERT INTO Environment
  SELECT * FROM n
  WHERE NOT EXISTS (
      SELECT 1 FROM Environment e2
      WHERE e2.Environment = n.Environment
    );

  /* Reconcile servers from input json. We don't want to blow out the table
   * for RI reasons.  Also try to bring in defaults.
   */

  CREATE TEMP TABLE ServerDefaults (LIKE Server INCLUDING DEFAULTS) ON COMMIT DROP;
  ALTER TABLE ServerDefaults ALTER Server DROP NOT NULL;
  INSERT INTO ServerDefaults DEFAULT VALUES;

  CREATE TEMP TABLE servers (LIKE Server);

  t := to_jsonb((SELECT sd FROM ServerDefaults sd));

  FOR r IN SELECT json_array_elements((_Configuration)->'servers') j
  LOOP
    INSERT INTO servers SELECT * FROM jsonb_populate_record(
      NULL::Server,
       t || jsonb_strip_nulls(r.j::jsonb));
  END LOOP;

  INSERT INTO Server SELECT * FROM servers s
  WHERE NOT EXISTS (
    SELECT 1 FROM Server WHERE Server.Server = s.Server
  );

  UPDATE Server s SET
    TaskClientThreads = s_new.TaskClientThreads,
    ServerType = s_new.ServerType
  FROM Servers s_new
  WHERE s.Server = s_new.Server;

  DELETE FROM TaskChainConfiguration WHERE TaskChainConfigurationName IN (
    SELECT
      json_array_elements(
        _Configuration->'taskChainConfigurations')->>'taskchainconfigurationname'
  );

  /* Reconcile databases */
  CREATE TEMP TABLE databases ON COMMIT DROP AS
    SELECT * FROM json_populate_recordset(
      NULL::Database,
      (_Configuration)->'databases');

  INSERT INTO Database SELECT * FROM databases d
  WHERE NOT EXISTS (
    SELECT 1 FROM Database WHERE
      Database.Environment = d.Environment
      AND Database.Database = d.Database
      AND Database.NodeId = d.NodeId
  );

  UPDATE Database d SET
    Server = d_new.Server,
    ConnectionString = d_new.ConnectionString
  FROM databases d_new
  WHERE
    d.Environment = d_new.Environment
    AND d.Database = d_new.Database
    AND d.NodeId = d_new.NodeId;

  /* Reconcile tasks. */

  /* insert task in such a way as to preserve defaults */
  CREATE TEMP TABLE TaskDefaults (LIKE Task INCLUDING DEFAULTS) ON COMMIT DROP;
  ALTER TABLE TaskDefaults ALTER Task DROP NOT NULL;
  INSERT INTO TaskDefaults DEFAULT VALUES;

  CREATE TEMP TABLE tasks (LIKE Task);

  t := to_jsonb((SELECT td FROM TaskDefaults td));

  FOR r IN SELECT json_array_elements((_Configuration)->'tasks') j
  LOOP
    INSERT INTO tasks SELECT * FROM jsonb_populate_record(
      NULL::Task,
       t || jsonb_strip_nulls(r.j::jsonb));
  END LOOP;

  /* Clean out tasks that were attached to databases we are loading but are
   * no longer attached to them
   */
  DELETE FROM Task
  WHERE
    Database IN (SELECT Database FROM Databases)
    AND Task NOT IN (SELECT Task FROM Tasks);


  INSERT INTO Task SELECT * FROM tasks t
  WHERE NOT EXISTS (
    SELECT 1 FROM Task
    WHERE Task.Task = t.Task
  );

  UPDATE Task SET
    TaskBody = t_new.TaskBody,
    RunSchedule = t_new.RunSchedule,
    Database = t_new.Database,
    StartTime = t_new.StartTime,
    EndTime = t_new.EndTime,
    MaxRunTime = t_new.MaxRunTime,
    IsConcurrent = t_new.IsConcurrent,
    IsTaskChainConfiguration = t_new.IsTaskChainConfiguration,
    TaskOrder = t_new.TaskOrder,
    SubTasksMustComplete = t_new.SubTasksMustComplete
  FROM tasks t_new
  WHERE Task.Task = t_new.Task;

  /* reconcile subtasks */
  CREATE TEMP TABLE subtasks ON COMMIT DROP AS
    SELECT * FROM json_populate_recordset(
      NULL::SubTask,
      (_Configuration)->'subTasks');

  /* Get rid of subtasks that are attached to any task we are inserting
   * but are not in the subtasks being inserted.
   */
  DELETE FROM SubTask
  WHERE
    Task IN (SELECT Task FROM subtasks)
    AND SubTask NOT IN (SELECT SubTask FROM subtasks);

  INSERT INTO SubTask SELECT * FROM subtasks s
  WHERE NOT EXISTS (
    SELECT 1 FROM SubTask
    WHERE SubTask.Task = s.Task
    AND SubTask.SubTask = s.SubTask
  );

  UPDATE SubTask s SET
    TaskBody = s_new.TaskBody,
    RunSchedule = s_new.RunSchedule
  FROM subtasks s_new
  WHERE
    s.Task = s_new.Task
    AND s.SubTask = s_new.SubTask;

  /* Reconcile TaskDependencies */
  CREATE TEMP TABLE taskdependencies ON COMMIT DROP AS SELECT *
    FROM json_populate_recordset(
      NULL::TaskDependency,
      (_Configuration)->'taskDependencies');

  DELETE FROM TaskDependency WHERE
    (
      ParentTask IN (SELECT Task FROM tasks)
      OR ChildTask IN (SELECT Task FROM tasks)
    )
    AND (ParentTask, ChildTask) NOT IN (
      SELECT ParentTask, ChildTask FROM taskdependencies
    );

  INSERT INTO TaskDependency
  SELECT * FROM taskdependencies td
  WHERE NOT EXISTS (
    SELECT 1 FROM TaskDependency
    WHERE
      TaskDependency.ParentTask = td.ParentTask
      AND TaskDependency.ChildTask = td.ChildTask
  );

  UPDATE TaskDependency td SET
    AllNodes = td_new.AllNodes,
    CanRun = td_new.CanRun
  FROM taskdependencies td_new
  WHERE
    td.ChildTask = td_new.ChildTask
    AND td.ParentTask = td_new.ParentTask;

  /* update named task chain configuraitonns.  Since there is no RI on this
   * table, simply dump and reloading should do the tick.
   */
  CREATE TEMP TABLE tcc (LIKE TaskChainConfiguration
    INCLUDING DEFAULTS) ON COMMIT DROP;
  ALTER TABLE tcc ALTER TaskChainConfigurationName DROP NOT NULL;
  INSERT INTO tcc DEFAULT VALUES;

  t := to_jsonb((SELECT tcc FROM tcc));

  FOR r IN SELECT json_array_elements((_Configuration)->'taskChainConfigurations') j
  LOOP
    INSERT INTO TaskChainConfiguration SELECT * FROM jsonb_populate_record(
      NULL::TaskChainConfiguration,
       t || jsonb_strip_nulls(r.j::jsonb));
  END LOOP;

  INSERT INTO TaskChainConfigurationTask SELECT * FROM json_populate_recordset(
    NULL::TaskChainConfigurationTask,
    (_Configuration)->'taskChainConfigurationTasks');
END;
$$ LANGUAGE PLPGSQL;

/*
 * Saves some or all of the task configuration.
 *
 * XXX: Tagged configurations is not a completely developed concept.
 */
CREATE OR REPLACE FUNCTION SaveTaskConfiguration() RETURNS JSON AS
$$
BEGIN

END;
$$ LANGUAGE PLPGSQL;


CREATE OR REPLACE FUNCTION TaskChainUpdate() RETURNS TRIGGER AS
$$
DECLARE
  _FailedTasks INT;
BEGIN
  IF OLD.Finished IS NULL AND NEW.Finished IS NOT NULL
  THEN
    SELECT INTO _FailedTasks COUNT(*)
    FROM TaskChainTask
    WHERE
      DidSucceed(TaskResult) IS DISTINCT FROM true
      AND TaskChainId = NEW.TaskChainId;

    IF _FailedTasks = 0
    THEN
      NEW.DidSucceed = true;
    ELSE
      NEW.DidSucceed = false;
      NEW.ErrorMessage = format('%s tasks did not complete', _FailedTasks);

    END IF;
  END IF;

  RETURN NEW;
END;
$$ LANGUAGE PLPGSQL;

DROP TRIGGER IF EXISTS OnTaskChainUpdate ON TaskChain;
CREATE TRIGGER OnTaskChainUpdate AFTER UPDATE ON TaskChain
  FOR EACH ROW EXECUTE PROCEDURE TaskChainUpdate();


CREATE OR REPLACE FUNCTION CheckAndFailRelatedSubTasks(
  _SubTask TaskChainTask) RETURNS VOID AS
$$
DECLARE
  _SubTasksMustComplete BOOL;
BEGIN
  /* Task finished normally.  */
  IF DidSucceed(_SubTask.TaskResult)
  THEN
    RETURN;
  END IF;

  /* The task failed. What is the controlling task behavior? */
  SELECT INTO _SubTasksMustComplete SubTasksMustComplete
  FROM Task
  WHERE Task = _SubTask.Task;

  /* all other subtasks running or in queue should be failed if reason for 
   * this check was triggered by 'FAILED_VIA_TIMEOUT'
   */
  IF (NOT _SubTasksMustComplete AND _SubTask.TaskResult != 'FAILED_VIA_TIMEOUT')
  THEN
    /* We don't have to worry about other subtasks */
    RETURN;
  END IF;

  /* uh oh.  A subtask failed. Any other subtasks that are running
   * should be cancelled and the rest of them failed
   */
  PERFORM TasksFinished(
    array_agg(sc.TaskChainTaskId),
    'FAILED_VIA_CANCEL'::TaskResult_t,
    'Cancelled while in progress due to failure of other tasks')
  FROM ServerConnection sc
  WHERE
    Active
    AND TaskChainTaskId IN (
      SELECT tct.TaskChainTaskId
      FROM TaskChainTask tct
      WHERE
        TaskChainId = _SubTask.TaskChainId
        AND NodeId = _Subtask.NodeId
        AND Task = _SubTask.Task
        AND SubTask != _SubTask.SubTask
        AND IsSubTask(_SubTask.SubTask)
    );

  /* mark any yet to be started subtasks as failed. */
  PERFORM TasksFinished(TaskChainTasks, TaskResult, ErrorMessage)
  FROM
  (
    SELECT
      array_agg(tct.TaskChainTaskId) AS TaskChainTasks,
      'FAILED_VIA_SUBTASKS'::TaskResult_t AS TaskResult,
      format(
        'Auto cancelling as task %s/%s (id %s node %s) failed',
        _SubTask.Task,
        _SubTask.SubTask,
        _SubTask.TaskChainTaskId,
        _SubTask.NodeId) AS ErrorMessage
    FROM TaskChainTask tct
    WHERE
      IsSubTask(tct.SubTask)
      AND tct.Task = _SubTask.Task
      AND tct.TaskChainId = _SubTask.TaskChainId
      AND tct.Finished IS NULL
      AND tct.NodeId = _SubTask.NodeId
      AND SubTask != _SubTask.SubTask
    GROUP BY 2, 3
  ) q;
END;
$$ LANGUAGE PLPGSQL;

/*
 * Called when (and only when) the last subtask has completed
 */
CREATE OR REPLACE FUNCTION FinishParentTaskFromSubTask(
  _SubTask TaskChainTask) RETURNS VOID AS
$$
DECLARE
  _ParentTaskChainTaskId BIGINT;
  _ParentFinished TIMESTAMPTZ;

  _Failures INT;
BEGIN
  /* If the parent task is already marked failed due to cascaded dependency
   * failure, we don't need to record everything.  However, if the failure
   * was bubbled up from one or more subtasks failing, the parent task would
   * not have been marked failed yet.
   */
  SELECT
    TaskChainTaskId,
    Finished
    INTO
    _ParentTaskChainTaskId,
    _ParentFinished
  FROM TaskChainTask p
  WHERE
    TaskChainId = _SubTask.TaskChainId
    AND Task = _SubTask.Task
    AND NOT IsSubTask(SubTask)
    AND p.NodeId = _SubTask.NodeId;

  IF _ParentTaskChainTaskId IS NULL
  THEN
    PERFORM Log('ERROR',
      format(
        'Unable to resolve parent task for %',
        to_json(_SubTask)));
  END IF;

  IF _ParentFinished IS NOT NULL
  THEN
    RETURN;
  END IF;

  SELECT count(*) AS FailedTasks INTO _Failures
  FROM TaskChainTask p
  WHERE
    p.TaskChainId = _SubTask.TaskChainId
    AND p.Task = _SubTask.Task
    AND p.NodeId = _SubTask.NodeId
    AND IsSubTask(p.SubTask)
    AND NOT DidSucceed(TaskResult);

  PERFORM TaskFinished(
    tct.TaskChainTaskId,
    CASE WHEN _Failures = 0
      THEN 'SUCCESS_VIA_SUBTASKS'
      ELSE 'FAILED_VIA_SUBTASKS'
    END::TaskResult_t,
    CASE WHEN _Failures > 0
      THEN format('%s subtasks failed', _Failures)
    END)
  FROM TaskChainTask tct
  WHERE
    TaskChainId = _SubTask.TaskChainId
    AND Task = _SubTask.Task
    AND SubTask = ''
    AND tct.NodeId = _SubTask.NodeId
    AND NOT EXISTS (
      SELECT 1 FROM TaskChainTask p
      WHERE
        p.TaskChainId = _SubTask.TaskChainId
        AND p.Task = _SubTask.Task
        AND IsSubTask(p.SubTask)
        AND p.Finished IS NULL
        AND p.NodeId = _SubTask.NodeId
    );
END;
$$ LANGUAGE PLPGSQL;


/*
 * When a task completes, some tasks can become eligible to run as maintained
 * by the 'eligible' flag.  Mark them ready to go
 */
CREATE OR REPLACE FUNCTION SetDependentTasksEligible(
  _Task TaskChainTask) RETURNS VOID AS
$$
BEGIN
  UPDATE TaskChainTask tct SET Eligible = true
  FROM TaskDependency td
  WHERE
    tct.TaskChainId = _Task.TaskChainId
    /* don't mess with subtasks */
    AND tct.SubTask = ''

    /* is task dependent on currently completing task */
    AND tct.Task = td.ChildTask
    AND td.ParentTask = _Task.Task
    /* guard against tasks auto cancelled */
    AND tct.Finished IS NULL

    /* no point in setting it if it is already set */
    AND tct.Eligible IS NOT TRUE

    /* if the dependency is 'node local', we only mark eligibility for the
     * local node, else all nodes; This also applies to the negative
     * dependency check.
     */
    AND (td.AllNodes OR _Task.NodeId = tct.NodeId)

    AND (DidSucceed(_Task.TaskResult) OR td.CanRun)

    /* does the task being examined have any ancestors that are incomplete */
    AND NOT EXISTS
    (
      SELECT 1 FROM TaskChainTask parent
      JOIN TaskDependency td ON parent.Task = td.ParentTask
      WHERE
        parent.TaskChainId = _Task.TaskChainId
        AND td.ChildTask = tct.Task
        AND parent.Finished IS NULL
        AND (td.AllNodes OR parent.NodeId = tct.NodeId)
    );
END;
$$ LANGUAGE PLPGSQL;

CREATE OR REPLACE FUNCTION HandleTaskFailure(
  _Task TaskChainTask) RETURNS VOID AS
$$
BEGIN
  /*
   * The task failed, all direct dependencies of the task must also fail.
   * if the allnodes flag is set, that dependency will additionally cross
   * so that other nodes will fail the task well.
   */
  PERFORM TasksFinished(TaskChainTasks, TaskResult, ErrorMessage)
  FROM
  (
    SELECT
      array_agg(tct.TaskChainTaskId) AS TaskChainTasks,
      'FAILED_VIA_DEPENDENCY'::TaskResult_t AS TaskResult,
      format(
        'Auto cancelling as Task %s (id %s node %s) failed',
        _Task.Task,
        _Task.TaskChainTaskId,
        _Task.NodeId) AS ErrorMessage
    FROM TaskChainTask tct
    JOIN TaskDependency td ON
      td.ParentTask = _Task.Task
      AND td.ChildTask = tct.Task
    WHERE
      td.ChildTask = tct.Task
      AND tct.TaskChainId = _Task.TaskChainId
      AND (tct.NodeId = _Task.NodeId OR td.AllNodes)
      AND tct.Finished IS NULL
      AND tct.SubTask = ''
      AND td.CanRun IS NOT TRUE
    GROUP BY 2, 3
  ) q;

  /* If the parent task has failed via subtasks, there is no point to
   * mark subtasks for failure.
   */
  IF _Task.TaskResult IN ('SUCCESS_VIA_SUBTASKS', 'FAILED_VIA_SUBTASKS')
  THEN
    RETURN;
  END IF;

  /* Subtasks are automatically marked failed if the parent task fails */
  PERFORM TasksFinished(TaskChainTasks, TaskResult, ErrorMessage)
  FROM
  (
    SELECT
      array_agg(tct.TaskChainTaskId) AS TaskChainTasks,
      'FAILED_VIA_PARENT_TASK'::TaskResult_t AS TaskResult,
      format(
        'Auto cancelling as Parent Task %s (id %s node %s) failed',
        _Task.Task,
        _Task.TaskChainTaskId,
        _Task.NodeId) AS ErrorMessage
    FROM TaskChainTask tct
    WHERE
      tct.SubTask != ''
      AND tct.Task = _Task.Task
      AND tct.TaskChainId = _Task.TaskChainId
      AND tct.Finished IS NULL
      AND tct.NodeId = _Task.NodeId
    GROUP BY 2, 3
  ) q;
END;
$$ LANGUAGE PLPGSQL;


CREATE OR REPLACE FUNCTION TaskChainTaskUpdate() RETURNS TRIGGER AS
$$
DECLARE
  d TIMESTAMPTZ;
BEGIN
  /* If the task has started, all subtasks (if any) become eligible */
  IF NOT IsSubTask(NEW.SubTask)
    AND OLD.Started IS NULL
    AND NEW.Started IS NOT NULL
  THEN
    UPDATE TaskChainTask SET Eligible = true
    WHERE
      TaskChainId = NEW.TaskChainId
      AND Task = NEW.Task
      AND NodeId = NEW.NodeId
      AND IsSubTask(SubTask);
  END IF;

  /* Is this task newly finishing? if not, punt */
  IF NOT (NEW.Finished IS NOT NULL AND OLD.Finished IS NULL)
  THEN
    RETURN NEW;
  END IF;

  IF NEW.Eligible
  THEN
    /* If the task is finished, clear the eligibility flag */
    NEW.Eligible = false;
  END IF;

  /* If a subtask completes, look for an uncompleted subtask at same level. If
   * none found, mark the task complete as well.
   */
  IF IsSubTask(NEW.SubTask)
  THEN
    PERFORM 1
    FROM TaskChainTask p
    WHERE
      p.TaskChainId = NEW.TaskChainId
      AND p.Task = NEW.Task
      AND p.NodeId = NEW.NodeId
      AND IsSubTask(p.SubTask)
      AND Started IS NOT NULL
      AND Finished IS NULL LIMIT 1;

    IF FOUND
    THEN
      /* There is at least one incomplete subtask.  Did we fail and does that
       * failure requre handling?
       */
      PERFORM CheckAndFailRelatedSubTasks(NEW);
    ELSE
      PERFORM FinishParentTaskFromSubTask(NEW);
    END IF;

    RETURN NEW;
  END IF;


  /* If the task is regular task and completes, there is an important processing
   * descision to be made: failed tasks can (depending on how dependency is
   * configured) cascade the failure to dependent tasks while successfully
   * completed tasks may cause other dependent tasks to beecome eligible to
   * be run.
   */
  IF NOT DidSucceed(NEW.TaskResult)
  THEN
    PERFORM HandleTaskFailure(NEW);
  END IF;

  PERFORM SetDependentTasksEligible(NEW);

  RETURN NEW;
END;
$$ LANGUAGE PLPGSQL;

DROP TRIGGER IF EXISTS OnTaskChainTaskUpdate ON TaskChainTask;
CREATE TRIGGER OnTaskChainTaskUpdate AFTER UPDATE ON TaskChainTask
  FOR EACH ROW EXECUTE PROCEDURE TaskChainTaskUpdate();




CREATE OR REPLACE FUNCTION CleanupAfterRestart() RETURNS VOID AS
$$
BEGIN
  /* This is not the same backend as the one that initialized the dblink
   * connections.  There is no chance of recovery of task execution state.
   */
  IF pg_backend_pid() != (SELECT MainProcessPid FROM GlobalControl)
  THEN
    TRUNCATE ServerConnection;
  END IF;

  /* if the dblink connections are still active, grab any reponses that are
   * still outstanding.   This will only work if we restarted frome the
   * same database session that initiated the tasks (which is a bit of a
   * hail mary).
   */
  BEGIN
    PERFORM CheckForFinishedTasks();
  EXCEPTION WHEN OTHERS THEN
    PERFORM Log('WARNING', format(
      'Attempted to recover old tasks, but was not able to do so via %s',
      SQLERRM));
  END;

  /* at this point, any task that was assume to be running before startup
   * must now be considered failed.  Likewise, any running task chains will
   * be assumed dead.
   *
   * XXX: There may be a more clever way to try to deal with this, but there
   *      are a lot of complex failure modes and nuking everything is by
   *      far the simplest approach.
   */
  ALTER TABLE TaskChainTask DISABLE TRIGGER OnTaskChainTaskUpdate;

  PERFORM TaskFinished(
    tct.TaskChainTaskId,
    'FAILED_VIA_RESTART',
    'Presumed failed due to spurious pgtask restart')
  FROM TaskChainTask tct
  WHERE
    tct.Finished IS NULL
    AND EXISTS (
      SELECT 1 FROM TaskChainTask tct2
      WHERE
        tct2.TaskChainId = tct.TaskChainId
        AND tct2.Started IS NOT NULL
    );

  UPDATE TaskChain tc SET
    Finished = now(),
    DidSucceed = false,
    ErrorMessage = 'Presumed failed due to spurious pgtask restart'
  WHERE
    Finished IS NULL
    AND Started IS NOT NULL
    AND EXISTS (
      SELECT 1 FROM TaskChainTask tct
      WHERE
        tct.TaskChainId = tc.TaskChainId
        AND tct.Started IS NOT NULL
    );

  ALTER TABLE TaskChainTask ENABLE TRIGGER OnTaskChainTaskUpdate;
END;
$$ LANGUAGE PLPGSQL;


CREATE OR REPLACE FUNCTION CheckForTimedOutIdleTasksInternal(
  _TaskChainId BIGINT) RETURNS VOID AS
$$
BEGIN
  SET LOCAL enable_nestloop to false;
  
  PERFORM 
    TasksFinished(
    array_agg(tct.TaskChainTaskId),
    'FAILED_VIA_TIMEOUT',
    'Cancelling idle tasks that have timed out')
  FROM TaskChainTask tct
  JOIN TaskChain tc USING(TaskChainId)
  JOIN Task USING(Task)
  JOIN TaskChainTask parent ON
    tct.SubTask != ''
    AND parent.Subtask = ''
    AND parent.Task = tct.Task
    AND parent.TaskChainId = _TaskChainId
    AND tct.TaskChainId = _TaskChainId
    AND parent.NodeId = tct.NodeId
  WHERE 
    tct.Started IS NULL
    AND tct.Finished IS NULL
    AND parent.Started + MaxRunTime < now();   
END;
$$ LANGUAGE PLPGSQL;    


CREATE OR REPLACE FUNCTION CheckForTimedOutIdleTasks() RETURNS VOID AS
$$
DECLARE
  _TaskChainId BIGINT;
BEGIN
  FOR _TaskChainId IN 
    SELECT TaskChainId 
    FROM TaskChain tc
    WHERE
    tc.Started IS NOT NULL
    AND tc.Finished IS NULL
  LOOP
    PERFORM CheckForTimedOutIdleTasksInternal(_TaskChainId);
  END LOOP;
END;
$$ LANGUAGE PLPGSQL;


CREATE OR REPLACE FUNCTION AuditRunningTasks(
  _AuditInterval INTERVAL,
  _LastAudit TIMESTAMPTZ) RETURNS VOID AS
$$
BEGIN
  IF _LastAudit + _AuditInterval > now()
  THEN
    RETURN;
  END IF;

  INSERT INTO ActiveTaskAudit
  SELECT 
    now(),
    vat.*
  FROM vw_ActiveTasks vat;

  UPDATE GlobalControl SET LastAudit = now();
END;
$$ LANGUAGE PLPGSQL;


CREATE OR REPLACE FUNCTION ProcessTasks() RETURNS BOOL AS
$$
DECLARE
  _DidStuff BOOL;
  _RanTasks BOOL;
  _TaskChainTaskId BIGINT;
  gc GlobalControl;
BEGIN
  SELECT INTO gc * FROM GlobalControl;

  /* record tasks that are running on semi-frequent basis */
  PERFORM AuditRunningTasks(gc.AuditInterval, gc.LastAudit);

  /* check for tasks to cancel, etc */
  _DidStuff := CheckForAdmnistrativeActions();

  /* If there were any cancels/pauses/etc, exit out in order to flush and
   * commit the transaction so that no further processing is done.  This is
   * done to avoid conflicting trigger effects.
   */
  IF _DidStuff
  THEN
    RETURN true;
  END IF;

  /* Are there any tasks that were waiting and timed out? */
  IF COALESCE(gc.IdleTaskTimeoutLastCheck, now()) 
    + gc.IdleTaskTimeoutInterval > now()
  THEN
    PERFORM CheckForTimedOutIdleTasks();
    UPDATE GlobalControl SET IdleTaskTimeoutLastCheck = now();
  END IF;

  PERFORM CheckForTimedOutIdleTasks();

  /* reap tasks that resolved in the background */
  _DidStuff := CheckForFinishedTasks();

  FOR _TaskChainTaskId IN SELECT * FROM GetTasks()
  LOOP
    _RanTasks := RunTask(_TaskChainTaskId);

    _DidStuff := _DidStuff OR _RanTasks;
  END LOOP;

  /* if there is a queued task chain, start it */
  UPDATE TaskChain tc SET Started = clock_timestamp()
  WHERE
    tc.Started IS NULL
    AND tc.TaskChainConfigurationName IS NOT NULL
    AND tc.Finished IS NULL
    AND NOT EXISTS (
      /* is there a queued or running task ahead of me? */
      SELECT 1 FROM TaskChain tc2
      WHERE
        tc2.TaskChainConfigurationName = tc.TaskChainConfigurationName
        AND tc2.Environment = tc.Environment
        AND tc2.TaskChainId < tc.TaskChainId
        AND tc2.Finished IS NULL
    );

  RETURN _DidStuff;
END;
$$ LANGUAGE PLPGSQL;

CREATE OR REPLACE PROCEDURE MainLoop() AS
$$
DECLARE
  _DidStuff BOOL;
  _MessagedNothing BOOL DEFAULT false;
  gc GlobalControl;
BEGIN
  SELECT INTO gc * FROM GlobalControl;

  PERFORM Log('Initializaton complete. Entering main server loop');
  LOOP
    _DidStuff := false;

    SELECT INTO gc * FROM GlobalControl;

    _DidStuff := ProcessTasks();

    /* flush transaction */
    COMMIT;

    IF NOT _DidStuff
    THEN
      IF EXISTS(
        SELECT 1 FROM ServerConnection WHERE Active)
      THEN
        PERFORM pg_sleep(gc.MainLoopSleepSecondsActive);
      ELSE
        IF NOT _MessagedNothing
        THEN
          PERFORM Log('Nothing to do...sleeping');
          _MessagedNothing := true;
        END IF;

        PERFORM pg_sleep(gc.MainLoopSleepSecondsIdle);
      END IF;
    ELSE
      _MessagedNothing := false;
    END IF;
  END LOOP;
END;
$$ LANGUAGE PLPGSQL;



/*
 * Main server process.
 */
CREATE OR REPLACE PROCEDURE Main(
  _ForceRestart BOOL DEFAULT false) AS
$$
DECLARE
  gc GlobalControl;

  _Locked BOOL;
  s Server;
  _TaskChainTaskId BIGINT;

  _MessagedNothing BOOL DEFAULT false;

  tct TaskChainTask;
BEGIN
  /* There must only be one :-D  */
  SELECT INTO _Locked pg_try_advisory_lock(0);

  IF NOT _Locked
  THEN
    IF NOT _ForceRestart
    THEN
      RETURN;
    END IF;

    SELECT INTO gc * FROM GlobalControl;

    PERFORM Log(
      'Attemping to kill old main process and take over (good luck!)');

    PERFORM pg_terminate_backend(gc.MainProcessPid);
    PERFORM pg_advisory_lock(0);
  END IF;

  /* set start time so that uptime can be determined */
  UPDATE GlobalControl SET StartTime = clock_timestamp();

  PERFORM Log('pgtask advisory lock acquired.  Starting Initializaton');
  SELECT INTO gc * FROM GlobalControl;

  /* If the exit was unclean, there may be tasks left in indeterminate state.
   * Do some housekeeping to deal with that.
   */
  IF gc.GracefulShutdown IS DISTINCT FROM TRUE
  THEN
    PERFORM CleanupAfterRestart();
  END IF;

  /* Mark the main process pid so that we can guard future restart operations */
  UPDATE GlobalControl SET
    MainProcessPid = pg_backend_pid(),
    GracefulShutdown = FALSE;

  /* Create our connection table. This is temporary as upon restart there is
   * no reason to believe any connection tracking data is correct.
   */
  PERFORM CreateDblinkConnections();

  CALL MainLoop();
END;
$$ LANGUAGE PLPGSQL;


/*
 * Create subtask based on the subtask json for a given parenttask
 * Used by subtask gnerator tasks to create subtasks at runtime
 */
CREATE OR REPLACE FUNCTION CreateTaskChainSubtask(
  _TaskChainTaskId BIGINT, /* TaskChainTaskId for prep Task */
  _Task TEXT , /* Parent Task name */
  _SubTasks JSONB
)  RETURNS VOID AS
$$
DECLARE
  _TaskChainTaskRec  RECORD;
BEGIN
  /* get valid taskchaintask record */
  SELECT tct.* INTO _TaskChainTaskRec
  FROM TaskChainTask tct
  JOIN TaskChainTask ptct
    ON  tct.taskChainId = ptct.taskChainId
      AND tct.environment = ptct.environment
      AND tct.database = ptct.database
      AND tct.nodeid = ptct.nodeId
  WHERE ptct.TaskChainTaskId = _TaskChainTaskId
    AND tct.Task = _Task
    AND tct.Started IS NULL
    AND tct.Finished IS NULL;

  IF FOUND THEN
    PERFORM Log('Adding Subtasks via subtask generator for TaskChainTaskId:' ||
      _TaskChainTaskId::TEXT);
    INSERT INTO TaskChainTask(
      TaskChainId,
      Environment,
      Database,
      NodeId,
      Task,
      SubTask,
      TaskBody,
      Eligible)
    SELECT
      _TaskChainTaskRec.TaskChainId,
      _TaskChainTaskRec.Environment,
      _TaskChainTaskRec.Database,
      _TaskChainTaskRec.NodeId,
      _TaskChainTaskRec.Task,
      s.SubTask,
      s.TaskBody,
      true
    FROM jsonb_populate_recordset(
      NULL::subTask,
      _SubTasks->'subTasks') s;
   ELSE
    PERFORM Log('WARNING','Failed to Add Subtasks via subtask generator
      for TaskChainTaskId:' || _TaskChainTaskId::TEXT);
   END IF;
END;
$$ LANGUAGE PLPGSQL;


