package io.delphiplatform.api.v3.bigtable.reader;


import com.google.cloud.bigtable.data.v2.models.Row;
import com.google.cloud.bigtable.data.v2.models.RowCell;
import com.google.protobuf.ByteString;
import com.google.protobuf.GeneratedMessageV3;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.util.EnumSet;
import java.util.List;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Stream;

import io.delphiplatform.api.v3.bigtable.entity.TrackStream;

import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.ARTIST_ID;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.DEMOGRAPHICS;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.DSP;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.ISRC;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.PLAYLIST_ID;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.PRODUCT_FAMILY_ID;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.PRODUCT_ID;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.PROJECT_NUMBER;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.STREAMS;
import static io.delphiplatform.api.v3.bigtable.reader.BigtableColumns.TRACK_ID;

public abstract class BigtableEntityReader<T> {

    public static final DateTimeFormatter LOCAL_DATE_TIME_FORMAT;

    static {
        LOCAL_DATE_TIME_FORMAT = new DateTimeFormatterBuilder()
            .parseCaseInsensitive()
            .append(DateTimeFormatter.ISO_LOCAL_DATE)
            .appendLiteral(' ')
            .append(DateTimeFormatter.ISO_LOCAL_TIME)
            .toFormatter();
    }

    public static final String meta = "meta";
    public static final String metrics = "metrics";
    public static final String pos = "pos";

    public static enum EntityReadingParams {
        EXPAND_ISRC_BY_TRACK_CURRENT_POSITION
    }

    public abstract Class<T> getParsedClass();

    public abstract T readRow(Row row);

    public Stream<T> readRows(Stream<Row> rowsStream, EnumSet<EntityReadingParams> params) {
        return rowsStream.map(this::readRow);
    }

    public Stream<T> readRows(Stream<Row> rowsStream) {
        return readRows(rowsStream, null);
    }

    protected <P extends GeneratedMessageV3, E extends TrackStream<P, ? extends GeneratedMessageV3>> E readTrackStream(
        E emptyStream, Row row,
        Function<ByteString, P> protoParser, String streamsColumnName, String demographicsColumnName) {
        emptyStream.setDate(readDate(row));
        emptyStream.setDsp(readString(row, meta, DSP));
        emptyStream.setArtistId(readString(row, meta, ARTIST_ID));
        emptyStream.setIsrc(readString(row, meta, ISRC));
        emptyStream.setPlaylistId(readString(row, meta, PLAYLIST_ID));
        emptyStream.setProductId(readString(row, meta, PRODUCT_ID));
        emptyStream.setTrackId(readString(row, meta, TRACK_ID));
        emptyStream.setProjectNumber(readString(row, meta, PROJECT_NUMBER));
        emptyStream.setProductFamilyId(readString(row, meta, PRODUCT_FAMILY_ID));

        readBytes(row, streamsColumnName).ifPresent(bytes -> emptyStream.setStreams(protoParser.apply(bytes)));
        readBytes(row, demographicsColumnName).ifPresent(
            bytes -> emptyStream.setDemographics(protoParser.apply(bytes)));
        return emptyStream;
    }

    protected <P extends GeneratedMessageV3, E extends TrackStream<P, ? extends GeneratedMessageV3>> E readTrackStream(
        E emptyStream, Row row,
        Function<ByteString, P> protoParser) {
        return readTrackStream(emptyStream, row, protoParser, STREAMS, DEMOGRAPHICS);
    }

    protected LocalDate readDate(Row row) {
        return readDate(row, meta, BigtableColumns.DATE);
    }

    protected LocalDate readDate(Row row, String qualifier) {
        return readDate(row, meta, qualifier);
    }

    protected LocalDate readDate(Row row, String columnFamily, String qualifier) {
        String stringValue = readString(row, columnFamily, qualifier);
        return stringValue == null ? null : LocalDate.parse(stringValue);
    }

    protected LocalDateTime readDateTime(Row row, String columnFamily, String qualifier) {
        String stringValue = readString(row, columnFamily, qualifier);

        return stringValue == null ? null : LocalDateTime.parse(stringValue, LOCAL_DATE_TIME_FORMAT);
    }

    protected Long readLong(Row row, String qualifier) {
        return readLong(row, metrics, qualifier);
    }

    protected Long readLong(Row row, String columnFamily, String qualifier) {
        return safeGetCell(row, columnFamily, qualifier)
            .map(cell -> cell.getValue().asReadOnlyByteBuffer().getLong())
            .orElse(null);
    }

    protected Integer readInt(Row row, String columnFamily, String qualifier) {
        return safeGetCell(row, columnFamily, qualifier)
            .map(cell -> cell.getValue().asReadOnlyByteBuffer().getInt())
            .orElse(null);
    }

    protected Integer readIntFromStringColumn(Row row, String columnFamily, String qualifier) {
        String stringVal = readString(row, columnFamily, qualifier);

        return stringVal == null ? null : Integer.valueOf(stringVal);
    }

    protected Boolean readBoolFromStringColumn(Row row, String columnFamily, String qualifier) {
        String stringVal = readString(row, columnFamily, qualifier);

        return stringVal == null ? null : Boolean.valueOf(stringVal);
    }

    protected String readString(Row row, String qualifier) {
        return readString(row, meta, qualifier);
    }

    protected String readString(Row row, String columnFamily, String qualifier) {
        return safeGetCell(row, columnFamily, qualifier)
            .map(cell -> cell.getValue().toStringUtf8())
            .orElse(null);
    }

    protected Optional<ByteString> readBytes(Row row, String qualifier) {
        return readBytes(row, metrics, qualifier);
    }

    private Optional<ByteString> readBytes(Row row, String columnFamily, String qualifier) {
        return safeGetCell(row, columnFamily, qualifier)
            .map(RowCell::getValue);
    }

    private Optional<RowCell> safeGetCell(Row row, String columnFamily, String qualifier) {
        List<RowCell> cells = row.getCells(columnFamily, qualifier);
        return cells.isEmpty() ? Optional.empty() : Optional.of(cells.get(0));
    }


}