/*
 * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 */

package io.airbyte.integrations.source.relationaldb.state;

import static io.airbyte.integrations.source.relationaldb.state.StateTestConstants.CURSOR;
import static io.airbyte.integrations.source.relationaldb.state.StateTestConstants.CURSOR_FIELD1;
import static io.airbyte.integrations.source.relationaldb.state.StateTestConstants.CURSOR_FIELD2;
import static io.airbyte.integrations.source.relationaldb.state.StateTestConstants.NAME_NAMESPACE_PAIR1;
import static io.airbyte.integrations.source.relationaldb.state.StateTestConstants.NAME_NAMESPACE_PAIR2;
import static io.airbyte.integrations.source.relationaldb.state.StateTestConstants.getCatalog;
import static io.airbyte.integrations.source.relationaldb.state.StateTestConstants.getState;
import static io.airbyte.integrations.source.relationaldb.state.StateTestConstants.getStream;
import static org.junit.jupiter.api.Assertions.assertEquals;

import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
import io.airbyte.integrations.source.relationaldb.CursorInfo;
import io.airbyte.integrations.source.relationaldb.models.DbStreamState;
import java.util.Collections;
import java.util.Optional;
import org.junit.jupiter.api.Test;

/**
 * Test suite for the {@link CursorManager} class.
 */
public class CursorManagerTest {

  @Test
  void testCreateCursorInfoCatalogAndStateSameCursorField() {
    final CursorManager<DbStreamState> cursorManager = createCursorManager(CURSOR_FIELD1, CURSOR, NAME_NAMESPACE_PAIR1);
    final CursorInfo actual = cursorManager.createCursorInfoForStream(
        NAME_NAMESPACE_PAIR1,
        getState(CURSOR_FIELD1, CURSOR),
        getStream(CURSOR_FIELD1),
        DbStreamState::getCursor,
        DbStreamState::getCursorField);
    assertEquals(new CursorInfo(CURSOR_FIELD1, CURSOR, CURSOR_FIELD1, CURSOR), actual);
  }

  @Test
  void testCreateCursorInfoCatalogAndStateSameCursorFieldButNoCursor() {
    final CursorManager<DbStreamState> cursorManager = createCursorManager(CURSOR_FIELD1, null, NAME_NAMESPACE_PAIR1);
    final CursorInfo actual = cursorManager.createCursorInfoForStream(
        NAME_NAMESPACE_PAIR1,
        getState(CURSOR_FIELD1, null),
        getStream(CURSOR_FIELD1),
        DbStreamState::getCursor,
        DbStreamState::getCursorField);
    assertEquals(new CursorInfo(CURSOR_FIELD1, null, CURSOR_FIELD1, null), actual);
  }

  @Test
  void testCreateCursorInfoCatalogAndStateChangeInCursorFieldName() {
    final CursorManager<DbStreamState> cursorManager = createCursorManager(CURSOR_FIELD1, CURSOR, NAME_NAMESPACE_PAIR1);
    final CursorInfo actual = cursorManager.createCursorInfoForStream(
        NAME_NAMESPACE_PAIR1,
        getState(CURSOR_FIELD1, CURSOR),
        getStream(CURSOR_FIELD2),
        DbStreamState::getCursor,
        DbStreamState::getCursorField);
    assertEquals(new CursorInfo(CURSOR_FIELD1, CURSOR, CURSOR_FIELD2, null), actual);
  }

  @Test
  void testCreateCursorInfoCatalogAndNoState() {
    final CursorManager<DbStreamState> cursorManager = createCursorManager(CURSOR_FIELD1, CURSOR, NAME_NAMESPACE_PAIR1);
    final CursorInfo actual = cursorManager.createCursorInfoForStream(
        NAME_NAMESPACE_PAIR1,
        Optional.empty(),
        getStream(CURSOR_FIELD1),
        DbStreamState::getCursor,
        DbStreamState::getCursorField);
    assertEquals(new CursorInfo(null, null, CURSOR_FIELD1, null), actual);
  }

  @Test
  void testCreateCursorInfoStateAndNoCatalog() {
    final CursorManager<DbStreamState> cursorManager = createCursorManager(CURSOR_FIELD1, CURSOR, NAME_NAMESPACE_PAIR1);
    final CursorInfo actual = cursorManager.createCursorInfoForStream(
        NAME_NAMESPACE_PAIR1,
        getState(CURSOR_FIELD1, CURSOR),
        Optional.empty(),
        DbStreamState::getCursor,
        DbStreamState::getCursorField);
    assertEquals(new CursorInfo(CURSOR_FIELD1, CURSOR, null, null), actual);
  }

  // this is what full refresh looks like.
  @Test
  void testCreateCursorInfoNoCatalogAndNoState() {
    final CursorManager<DbStreamState> cursorManager = createCursorManager(CURSOR_FIELD1, CURSOR, NAME_NAMESPACE_PAIR1);
    final CursorInfo actual = cursorManager.createCursorInfoForStream(
        NAME_NAMESPACE_PAIR1,
        Optional.empty(),
        Optional.empty(),
        DbStreamState::getCursor,
        DbStreamState::getCursorField);
    assertEquals(new CursorInfo(null, null, null, null), actual);
  }

  @Test
  void testCreateCursorInfoStateAndCatalogButNoCursorField() {
    final CursorManager<DbStreamState> cursorManager = createCursorManager(CURSOR_FIELD1, CURSOR, NAME_NAMESPACE_PAIR1);
    final CursorInfo actual = cursorManager.createCursorInfoForStream(
        NAME_NAMESPACE_PAIR1,
        getState(CURSOR_FIELD1, CURSOR),
        getStream(null),
        DbStreamState::getCursor,
        DbStreamState::getCursorField);
    assertEquals(new CursorInfo(CURSOR_FIELD1, CURSOR, null, null), actual);
  }

  @Test
  void testGetters() {
    final CursorManager cursorManager = createCursorManager(CURSOR_FIELD1, CURSOR, NAME_NAMESPACE_PAIR1);
    final CursorInfo actualCursorInfo = new CursorInfo(CURSOR_FIELD1, CURSOR, null, null);

    assertEquals(Optional.of(actualCursorInfo), cursorManager.getCursorInfo(NAME_NAMESPACE_PAIR1));
    assertEquals(Optional.empty(), cursorManager.getCursorField(NAME_NAMESPACE_PAIR1));
    assertEquals(Optional.empty(), cursorManager.getCursor(NAME_NAMESPACE_PAIR1));

    assertEquals(Optional.empty(), cursorManager.getCursorInfo(NAME_NAMESPACE_PAIR2));
    assertEquals(Optional.empty(), cursorManager.getCursorField(NAME_NAMESPACE_PAIR2));
    assertEquals(Optional.empty(), cursorManager.getCursor(NAME_NAMESPACE_PAIR2));
  }

  private CursorManager<DbStreamState> createCursorManager(final String cursorField,
                                                           final String cursor,
                                                           final AirbyteStreamNameNamespacePair nameNamespacePair) {
    final DbStreamState dbStreamState = getState(cursorField, cursor).get();
    return new CursorManager<>(
        getCatalog(cursorField).orElse(null),
        () -> Collections.singleton(dbStreamState),
        DbStreamState::getCursor,
        DbStreamState::getCursorField,
        s -> nameNamespacePair);
  }

}
