/**
 * Firehose
 */
import { default as Firehose } from "aws-sdk/clients/firehose";
import { IDENTIFIERS } from "constant/Identifiers";
import { inject, injectable } from "inversify";
import "reflect-metadata";
import { IFirehoseGateway } from "repository/IFirehoseGateway";
import { Observable, of } from "rxjs";
import { tag } from "rxjs-spy/operators";
import { concatMap, mapTo } from "rxjs/operators";
import tsLogClass from "ts-log-class";

/**
 * Firehose
 */
@injectable()
@tsLogClass()
export class FireHoseGateway implements IFirehoseGateway {
  private readonly _client: Firehose;

  constructor(@inject(IDENTIFIERS.AwsFirehose) client: Firehose) {
    this._client = client;
  }

  public put(stream: string, data: object): Observable<boolean> {
    return of(1).pipe(
      concatMap(async () =>
        this._client
          .putRecordBatch({
            DeliveryStreamName: stream,
            Records: [{ Data: JSON.stringify(data) }],
          })
          .promise()
      ),
      tag("FireHostGateway | put"),
      mapTo(true)
    );
  }
}