Backstage catalog processor for Google Cloud Storage

GcsProcessor.ts

import { Readable } from 'stream';
import { LocationSpec } from '@backstage/catalog-model';
import {
  results,
  CatalogProcessor,
  CatalogProcessorEmit,
} from '@backstage/plugin-catalog-backend';
import { Storage, StorageOptions } from '@google-cloud/storage';
import YAML from 'yaml';

const readToPromise = (stream: Readable): Promise<string> => {
  const chunks: Uint8Array[] = [];
  return new Promise((resolve, reject) => {
    stream.on('data', (chunk: Uint8Array) => {
      chunks.push(chunk);
    });
    stream.on('error', reject);
    stream.on('end', () => {
      resolve(Buffer.concat(chunks).toString());
    });
  });
};

export class StorageReaderImpl {
  private readonly storage: Storage;

  constructor(storage: Storage) {
    this.storage = storage;
  }

  async read(bucket: string, key: string) {
    return await readToPromise(
      this.storage
        .bucket(bucket)
        .file(key)
        .createReadStream(),
    );
  }
}

export interface StorageReader {
  read(bucket: string, key: string): Promise<string>;
}

const parseURL = ({
  target,
}: LocationSpec): { bucket: string; key: string } => {
  const { protocol, host, pathname } = new URL(target);

  if (protocol !== 'gs:') {
    throw new Error(`not a valid GCS URL: ${target}`);
  }

  return {
    bucket: host,
    key: pathname.slice(1),
  };
};

export class GcsProcessor implements CatalogProcessor {
  constructor(private readonly storageReader: StorageReader) {}

  async readLocation(
    location: LocationSpec,
    _optional: boolean,
    emit: CatalogProcessorEmit,
  ): Promise<boolean> {
    if (location.type !== 'gocardless.io/catalog-processors/gcs') {
      return false;
    }

    try {
      const { bucket, key } = parseURL(location);

      const yaml = await this.storageReader.read(bucket, key);

      const documents = YAML.parseAllDocuments(yaml);

      documents
        .map(doc => doc.toJSON())
        .forEach(doc => {
          emit(results.entity(location, doc));
        });
    } catch (error) {
      emit(results.generalError(location, error.message));
      return false;
    }

    return true;
  }
}

export const createProcessor = (options?: StorageOptions): CatalogProcessor => {
  const storage = new Storage(options);
  return new GcsProcessor(new StorageReaderImpl(storage));
};