background.ts

import { createLogger, LogLevelString } from 'bunyan'
import {
  Db,
  HasuraBackgroundClient,
  MetadataClient,
  Worker
} from './index'
import onDeath from 'death'
import { Logger } from 'ts-log'
import { CustomError } from 'ts-custom-error'
import fs from 'fs-extra'
import { DbConfig } from './typeAliases'
// Todo: Hoist to util package next major version
export class MissingConfig extends CustomError {
  public constructor (message: string) {
    super()
    this.message = message
  }
}

export interface BackgroundConfig {
  db: DbConfig;
  composeProfiles: string;
  hasuraCliPath: string;
  hasuraCliExtPath: string;
  hasuraUri: string;
  loggerMinSeverity: LogLevelString;
  metadataServerUri: string;
  metadataUpdateInterval?: {
    assets: number;
  };
  assetBackfillBatchSize?: number;
}

async function getConfig (): Promise<BackgroundConfig> {
  const env = filterAndTypecastEnvs(process.env)
  if (!env.hasuraCliPath) {
    throw new MissingConfig('HASURA_CLI_PATH env not set')
  }
  if (!env.hasuraCliExtPath) {
    throw new MissingConfig('HASURA_CLI_EXT_PATH env not set')
  }
  if (!env.hasuraUri) {
    throw new MissingConfig('HASURA_URI env not set')
  }
  if (!env.metadataServerUri) {
    throw new MissingConfig('METADATA_SERVER_URI env not set')
  }
  if (!env.postgres.dbFile && !env.postgres.db) {
    throw new MissingConfig('POSTGRES_DB_FILE or POSTGRES_DB env not set')
  }
  if (!env.postgres.host) {
    throw new MissingConfig('POSTGRES_HOST env not set')
  }
  if (!env.postgres.passwordFile && !env.postgres.password) {
    throw new MissingConfig(
      'POSTGRES_PASSWORD_FILE or POSTGRES_PASSWORD env not set'
    )
  }
  if (!env.postgres.port) {
    throw new MissingConfig('POSTGRES_PORT env not set')
  }
  if (!env.postgres.userFile && !env.postgres.user) {
    throw new MissingConfig('POSTGRES_USER_FILE or POSTGRES_USER env not set')
  }
  let db: BackgroundConfig['db']
  try {
    db = {
      database:
        env.postgres.db ||
        (await fs.readFile(env.postgres.dbFile, 'utf8')).toString().trim(),
      host: env.postgres.host,
      password:
        env.postgres.password ||
        (await fs.readFile(env.postgres.passwordFile, 'utf8'))
          .toString()
          .trim(),
      port: env.postgres.port,
      user:
        env.postgres.user ||
        (await fs.readFile(env.postgres.userFile, 'utf8')).toString().trim()
    }
  } catch (error) {
    throw new MissingConfig('Database configuration cannot be read')
  }
  const { postgres, ...selectedEnv } = env
  return {
    ...selectedEnv,
    db,
    loggerMinSeverity: env.loggerMinSeverity || ('info' as LogLevelString)
  }
}

function filterAndTypecastEnvs (env: any) {
  const {
    COMPOSE_PROFILES,
    ASSET_METADATA_UPDATE_INTERVAL,
    ASSET_BACKFILL_BATCH_SIZE,
    HASURA_CLI_PATH,
    HASURA_CLI_EXT_PATH,
    HASURA_URI,
    LOGGER_MIN_SEVERITY,
    METADATA_SERVER_URI,
    POSTGRES_DB,
    POSTGRES_DB_FILE,
    POSTGRES_HOST,
    POSTGRES_PASSWORD,
    POSTGRES_PASSWORD_FILE,
    POSTGRES_PORT,
    POSTGRES_USER,
    POSTGRES_USER_FILE
  } = env as NodeJS.ProcessEnv
  return {
    composeProfiles: COMPOSE_PROFILES,
    hasuraCliPath: HASURA_CLI_PATH,
    hasuraCliExtPath: HASURA_CLI_EXT_PATH,
    hasuraUri: HASURA_URI,
    loggerMinSeverity: LOGGER_MIN_SEVERITY as LogLevelString,
    metadataServerUri: METADATA_SERVER_URI,
    metadataUpdateInterval: {
      assets: ASSET_METADATA_UPDATE_INTERVAL
        ? Number(ASSET_METADATA_UPDATE_INTERVAL)
        : undefined
    },
    assetBackfillBatchSize: ASSET_BACKFILL_BATCH_SIZE
      ? Number(ASSET_BACKFILL_BATCH_SIZE)
      : undefined,
    postgres: {
      db: POSTGRES_DB,
      dbFile: POSTGRES_DB_FILE,
      host: POSTGRES_HOST,
      password: POSTGRES_PASSWORD,
      passwordFile: POSTGRES_PASSWORD_FILE,
      port: POSTGRES_PORT ? Number(POSTGRES_PORT) : undefined,
      user: POSTGRES_USER,
      userFile: POSTGRES_USER_FILE
    }
  }
}

const ASSET_POLL_INTERVAL_MS = 30_000

function startAssetPolling (
  hasuraBackgroundClient: HasuraBackgroundClient,
  worker: Worker,
  dbConfig: DbConfig,
  initialLastSeenId: number,
  logger: Logger
): void {
  let lastSeenId = initialLastSeenId

  const poll = async () => {
    try {
      const { assetIds, nextLastSeenId } = await hasuraBackgroundClient.pollNewAssets(dbConfig, lastSeenId)
      lastSeenId = nextLastSeenId
      if (assetIds.length > 0) {
        await worker.publishInitialMetadataFetch(assetIds)
      }
    } catch (error) {
      logger.error({ module: 'AssetPoller' }, `Asset poll failed: ${error.message}`)
    } finally {
      setTimeout(poll, ASSET_POLL_INTERVAL_MS)
    }
  }

  setTimeout(poll, ASSET_POLL_INTERVAL_MS)
}

;(async function () {
  const config = await getConfig()
  const logger: Logger = createLogger({
    name: 'background',
    level: config.loggerMinSeverity
  })
  try {
    const hasuraBackgroundClient = new HasuraBackgroundClient(
      config.hasuraCliPath,
      config.hasuraCliExtPath,
      config.hasuraUri,
      logger
    )

    const isTokenRegistryEnabled =
      config.composeProfiles &&
      config.composeProfiles.split(',').includes('token-registry')

    const metadataClient = new MetadataClient(isTokenRegistryEnabled, config.metadataServerUri, logger)
    const worker = new Worker(
      hasuraBackgroundClient,
      logger,
      metadataClient,
      config.db,
      {
        metadataUpdateInterval: {
          assets: config.metadataUpdateInterval?.assets
        }
      }
    )
    const db = new Db(config.db, logger)
    let setupStarted = false
    await db.init({
      onDbInit: () => hasuraBackgroundClient.shutdown(),
      onDbSetup: async () => {
        if (setupStarted) return
        setupStarted = true
        try {
          await hasuraBackgroundClient.initialize()
          const lastSeenId = await hasuraBackgroundClient.getMaxMultiAssetId(config.db)
          const backfilledAssetIds = await hasuraBackgroundClient.backfillMissingAssets(config.db, config.assetBackfillBatchSize)
          await worker.initQueue()
          startAssetPolling(hasuraBackgroundClient, worker, config.db, lastSeenId, logger)
          await metadataClient.initialize()
          await worker.start()
          await worker.publishInitialMetadataFetch(backfilledAssetIds)
          hasuraBackgroundClient.getAssetIdsWithoutMetadata(config.db)
            .then(assetIds => worker.syncMissingMetadata(assetIds))
            .then(() => hasuraBackgroundClient.getRecentAssetIdsWithoutMetadata(config.db))
            .then(recentIds => worker.publishInitialMetadataFetch(recentIds))
            .catch(err => logger.error({ module: 'MetadataSync' }, `Metadata sync for existing assets failed: ${err.message}`))
        } catch (error) {
          logger.error(error.message)
          process.exit(1)
        }
      }
    })
    onDeath(async () => {
      await Promise.all([
        hasuraBackgroundClient.shutdown,
        worker.shutdown,
        db.shutdown
      ])
      process.exit(1)
    })
  } catch (error) {
    logger.error('Exiting due to uncaught exception', error.message)
    process.exit(1)
  }
})()