// Copyright 2018 Google LLC

//

// Licensed under the Apache License, Version 2.0 (the "License");

// you may not use this file except in compliance with the License.

// You may obtain a copy of the License at

//

// https://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software

// distributed under the License is distributed on an "AS IS" BASIS,

// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

// See the License for the specific language governing permissions and

// limitations under the License.

import {Transform, TransformOptions} from 'stream';

import {Bytes, Mutation} from './mutation';

import {TableUtils} from './utils/table';

export type Value = string | number | boolean | Uint8Array;

export enum DataEvent {

LAST_ROW_KEY_UPDATE,

}

export interface Chunk {

rowContents: Value;

commitRow: boolean;

resetRow: boolean;

rowKey?: string | Bytes;

familyName?: {value: string};

qualifier?: Qualifier | {value: Value};

timestampMicros?: number | Long;

labels?: string[];

value?: string | Buffer;

valueSize?: number;

}

export interface Data {

chunks: Chunk[];

lastScannedRowKey?: Buffer;

}

export interface Family {

[qualifier: string]: Qualifier[];

}

export interface Qualifier {

value?: string | Buffer;

labels?: string[];

timestamp?: number | Long;

size?: number;

}

export interface Row {

key?: Value;

data?: {[index: string]: Family};

}

export interface TransformErrorProps {

message: string;

chunk: Chunk | null;

}

export interface ChunkPushLastScannedRowData {

eventType: DataEvent.LAST_ROW_KEY_UPDATE;

lastScannedRowKey?: string;

}

export type ChunkPushData = Row | ChunkPushLastScannedRowData;

class TransformError extends Error {

constructor(props: TransformErrorProps) {

super();

this.name = 'TransformError';

this.message = `${props.message}: ${JSON.stringify(props.chunk)}`;

}

}

/**

* Enum for chunk formatter Row state.

* NEW_ROW: initial state or state after commitRow or resetRow

* ROW_IN_PROGRESS: state after first valid chunk without commitRow or resetRow

* CELL_IN_PROGRESS: state when valueSize > 0(partial cell)

*/

export enum RowStateEnum {

NEW_ROW = 1,

ROW_IN_PROGRESS = 2,

CELL_IN_PROGRESS = 3,

}

/**

* ChunkTransformer formats all incoming chunks in to row

* keeps all intermediate state until end of stream.

* Should use new instance for each request.

*/

export class ChunkTransformer extends Transform {

options: TransformOptions;

_destroyed: boolean;

lastRowKey?: Value;

state?: number;

row?: Row;

family?: Family;

qualifiers?: Qualifier[];

qualifier?: Qualifier;

constructor(options: TransformOptions = {}) {

options.objectMode = true; // forcing object mode

super(options);

this.options = options;

this._destroyed = false;

this.lastRowKey = undefined;

this.reset();

}

/**

* called at end of the stream.

* @public

* @param {callback} cb callback will be called with error if there is any uncommitted row

*/

_flush(cb: Function): void {

if (typeof this.row!.key !== 'undefined') {

this.destroy(

new TransformError({

message: 'Response ended with pending row without commit',

chunk: null,

}),

);

return;

}

cb();

}

/**

* transform the readrowsresponse chunks into friendly format. Chunks contain

* 3 properties:

*

* `rowContents` The row contents, this essentially is all data pertaining

* to a single family.

*

* `commitRow` This is a boolean telling us the all previous chunks for this

* row are ok to consume.

*

* `resetRow` This is a boolean telling us that all the previous chunks are to

* be discarded.

*

* @public

*

* @param {object} data readrows response containing array of chunks.

* @param {object} [_encoding] encoding options.

* @param {callback} next callback will be called once data is processed, with error if any error in processing

*/

_transform(data: Data, _encoding: string, next: Function): void {

for (const chunk of data.chunks!) {

switch (this.state) {

case RowStateEnum.NEW_ROW:

this.processNewRow(chunk);

break;

case RowStateEnum.ROW_IN_PROGRESS:

this.processRowInProgress(chunk);

break;

case RowStateEnum.CELL_IN_PROGRESS:

this.processCellInProgress(chunk);

break;

default:

break;

}

if (this._destroyed) {

next();

return;

}

}

if (data.lastScannedRowKey && data.lastScannedRowKey.length > 0) {

this.lastRowKey = Mutation.convertFromBytes(

data.lastScannedRowKey as Bytes,

{

userOptions: this.options,

},

);

/**

* Push an event that will update the lastRowKey in the user stream after

* all rows ahead of this event have reached the user stream. This will

* ensure that a retry excludes the lastScannedRow as this is required

* for the TestReadRows_Retry_LastScannedRow conformance test to pass. It

* is important to use a 'data' event to update the last row key in order

* to allow all the data queued ahead of this event to reach the user

* stream first.

*/

this.push({

eventType: DataEvent.LAST_ROW_KEY_UPDATE,

lastScannedRowKey: this.lastRowKey,

});

}

next();

}

/**

* called when stream is destroyed.

* @public

* @param {error} err error if any

*/

destroy(err?: Error): this {

if (this._destroyed) return this;

this._destroyed = true;

if (err) {

this.emit('error', err);

}

this.emit('close');

return this;

}

/**

* Resets state of formatter

* @private

*/

reset(): void {

this.family = {} as Family;

this.qualifiers = [];

this.qualifier = {};

this.row = {};

this.state = RowStateEnum.NEW_ROW;

}

/**

* sets lastRowkey and calls reset when row is committed.

* @private

*/

commit(): void {

const row = this.row;

this.reset();

this.lastRowKey = row!.key;

}

/**

* Validates valuesize and commitrow in a chunk

* @private

* @param {chunk} chunk chunk to validate for valuesize and commitRow

*/

validateValueSizeAndCommitRow(chunk: Chunk): void {

if (chunk.valueSize! > 0 && chunk.commitRow) {

this.destroy(

new TransformError({

message: 'A row cannot be have a value size and be a commit row',

chunk,

}),

);

}

}

/**

* Validates resetRow condition for chunk

* @private

* @param {chunk} chunk chunk to validate for resetrow

*/

validateResetRow(chunk: Chunk): void {

const containsData =

(chunk.rowKey && chunk.rowKey.length !== 0) ||

chunk.familyName ||

chunk.qualifier ||

(chunk.value && chunk.value.length !== 0) ||

// timestampMicros is an int64 in the protobuf definition,

// which can be either a number or an instance of Long.

// If it's a number...

(typeof chunk.timestampMicros === 'number' &&

chunk.timestampMicros! > 0) ||

// If it's an instance of Long...

(typeof chunk.timestampMicros === 'object' &&

'compare' in chunk.timestampMicros &&

typeof chunk.timestampMicros.compare === 'function' &&

chunk.timestampMicros.compare(0) === 1);

if (chunk.resetRow && containsData) {

this.destroy(

new TransformError({

message: 'A reset should have no data',

chunk,

}),

);

}

}

/**

* Validates state for new row.

* @private

* @param {chunk} chunk chunk to validate

* @param {newRowKey} newRowKey newRowKey of the new row

*/

validateNewRow(chunk: Chunk, newRowKey: string | Buffer): void {

const row = this.row;

const lastRowKey = this.lastRowKey;

let errorMessage: string | undefined;

if (typeof row!.key !== 'undefined') {

errorMessage = 'A new row cannot have existing state';

} else if (

typeof chunk.rowKey === 'undefined' ||

chunk.rowKey.length === 0 ||

newRowKey.length === 0

) {

errorMessage = 'A row key must be set';

} else if (chunk.resetRow) {

errorMessage = 'A new row cannot be reset';

} else if (lastRowKey === newRowKey) {

errorMessage = 'A commit happened but the same key followed';

} else if (

typeof lastRowKey !== 'undefined' &&

TableUtils.lessThanOrEqualTo(newRowKey as string, lastRowKey as string)

) {

errorMessage = 'A row key must be strictly increasing';

} else if (!chunk.familyName) {

errorMessage = 'A family must be set';

} else if (chunk.qualifier === null || chunk.qualifier === undefined) {

errorMessage = 'A column qualifier must be set';

}

if (errorMessage) {

this.destroy(new TransformError({message: errorMessage, chunk}));

return;

}

this.validateValueSizeAndCommitRow(chunk);

}

/**

* Validates state for rowInProgress

* @private

* @param {chunk} chunk chunk to validate

*/

validateRowInProgress(chunk: Chunk): void {

const row = this.row;

if (chunk.rowKey && chunk.rowKey.length) {

const newRowKey = Mutation.convertFromBytes(chunk.rowKey as Bytes, {

userOptions: this.options,

});

const oldRowKey = row!.key || '';

if (

newRowKey &&

chunk.rowKey &&

(newRowKey as string).length !== 0 &&

newRowKey.toString() !== oldRowKey.toString()

) {

this.destroy(

new TransformError({

message: 'A commit is required between row keys',

chunk,

}),

);

return;

}

}

if (

chunk.familyName &&

(chunk.qualifier === null || chunk.qualifier === undefined)

) {

this.destroy(

new TransformError({

message: 'A qualifier must be specified',

chunk,

}),

);

return;

}

this.validateResetRow(chunk);

this.validateValueSizeAndCommitRow(chunk);

}

/**

* Validates chunk for cellInProgress state.

* @private

* @param {chunk} chunk chunk to validate

*/

validateCellInProgress(chunk: Chunk): void {

this.validateResetRow(chunk);

this.validateValueSizeAndCommitRow(chunk);

}

/**

* Moves to next state in processing.

* @private

* @param {chunk} chunk chunk in process

*/

moveToNextState(chunk: Chunk): void {

const row = this.row;

if (chunk.commitRow) {

this.push(row);

this.commit();

this.lastRowKey = row!.key;

} else {

if (chunk.valueSize! > 0) {

this.state = RowStateEnum.CELL_IN_PROGRESS;

} else {

this.state = RowStateEnum.ROW_IN_PROGRESS;

}

}

}

/**

* Process chunk when in NEW_ROW state.

* @private

* @param {chunks} chunk chunk to process

*/

processNewRow(chunk: Chunk): void {

const newRowKey = Mutation.convertFromBytes(chunk.rowKey! as Bytes, {

userOptions: this.options,

}) as string;

this.validateNewRow(chunk, newRowKey);

if (chunk.familyName && chunk.qualifier) {

const row = this.row!;

row.key = newRowKey;

row.data = {};

this.family = row.data[chunk.familyName.value] = {} as Family;

const qualifierName = Mutation.convertFromBytes(

chunk.qualifier.value as Bytes,

{

userOptions: this.options,

},

);

this.qualifiers = this.family[qualifierName as {} as string] = [];

this.qualifier = {

value: Mutation.convertFromBytes(chunk.value! as Bytes, {

userOptions: this.options,

isPossibleNumber: true,

}) as string,

labels: chunk.labels,

timestamp: chunk.timestampMicros,

};

this.qualifiers.push(this.qualifier);

this.moveToNextState(chunk);

}

}

/**

* Process chunk when in ROW_IN_PROGRESS state.

* @private

* @param {chunk} chunk chunk to process

*/

processRowInProgress(chunk: Chunk): void {

this.validateRowInProgress(chunk);

if (chunk.resetRow) {

return this.reset();

}

const row = this.row!;

if (chunk.familyName) {

this.family = row.data![chunk.familyName.value] =

row.data![chunk.familyName.value] || {};

}

if (chunk.qualifier) {

const qualifierName = Mutation.convertFromBytes(

chunk.qualifier.value as Bytes,

{

userOptions: this.options,

},

) as string;

this.qualifiers = this.family![qualifierName] =

this.family![qualifierName] || [];

}

this.qualifier = {

value: Mutation.convertFromBytes(chunk.value! as Bytes, {

userOptions: this.options,

isPossibleNumber: true,

}) as string,

labels: chunk.labels,

timestamp: chunk.timestampMicros,

};

this.qualifiers!.push(this.qualifier);

this.moveToNextState(chunk);

}

/**

* Process chunk when in CELl_IN_PROGRESS state.

* @private

* @param {chunk} chunk chunk to process

*/

processCellInProgress(chunk: Chunk): void {

this.validateCellInProgress(chunk);

if (chunk.resetRow) {

return this.reset();

}

const chunkQualifierValue = Mutation.convertFromBytes(

chunk.value! as Bytes,

{

userOptions: this.options,

},

);

if (

chunkQualifierValue instanceof Buffer &&

this.qualifier!.value instanceof Buffer

) {

this.qualifier!.value = Buffer.concat([

this.qualifier!.value,

chunkQualifierValue,

]);

} else {

(this.qualifier!.value as string) += chunkQualifierValue;

}

this.moveToNextState(chunk);

}

}