import com.pyranid.Database;
import com.pyranid.DatabaseType;
import com.pyranid.NotificationSupervisor;
import com.pyranid.RetryPolicy;

import javax.sql.DataSource;
import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

/**
 * Console demonstration for Pyranid 4.8.0 and PostgreSQL.
 * Create and seed the feature_flag table from the Notifications documentation first.
 * Supply a session-affine DataSource with capacity for the listener and a query connection.
 */
public final class FeatureFlagListener {
  public record FeatureFlag(String name, Boolean enabled) {}

  private FeatureFlagListener() {}

  public static void run(DataSource dataSource) throws Exception {
    Database database = Database.with(dataSource, DatabaseType.POSTGRESQL)
      .build();
    AtomicReference<Map<String, Boolean>> flags =
      new AtomicReference<>(Map.of());

    Runnable reloadFlags = () -> {
      Map<String, Boolean> snapshot = database.query("""
        SELECT name, enabled
        FROM feature_flag
        """)
        .queryTimeout(Duration.ofSeconds(5))
        .fetchList(FeatureFlag.class)
        .stream()
        .collect(Collectors.toUnmodifiableMap(
          FeatureFlag::name, FeatureFlag::enabled));

      flags.set(snapshot);
      System.out.println("Refreshed feature flags: " + snapshot);
    };

    NotificationSupervisor supervisor = NotificationSupervisor
      .withDatabase(database)
      .channels(Set.of("feature_flag_changed"))
      .reconcile(reloadFlags)
      .pollEvery(Duration.ofSeconds(30))
      .onReconciliationFailure(Throwable::printStackTrace)
      .retry(
        failure -> (failure.getPhase() == NotificationSupervisor.FailurePhase.OPEN
            || failure.getPhase() == NotificationSupervisor.FailurePhase.RECEIVE)
          && failure.getException().getSqlState()
            .map(state -> state.startsWith("08")
              || Set.of("57P01", "57P02", "57P03").contains(state))
            .orElse(false),
        RetryPolicy.Backoff.fullJitter(
          RetryPolicy.Backoff.exponential(
            Duration.ofSeconds(1), Duration.ofMinutes(1))),
        Duration.ofMinutes(1)
      )
      .build();

    try {
      supervisor.start();
      supervisor.firstReady().toCompletableFuture().get(10, TimeUnit.SECONDS);

      System.out.println("new_checkout enabled: "
        + flags.get().getOrDefault("new_checkout", false));
      System.out.println("Ready. Press Enter to stop.");
      System.in.read();
    } finally {
      supervisor.requestShutdown();
      NotificationSupervisor.ShutdownResult result = supervisor.awaitShutdown();
      result.getFailure().ifPresent(Throwable::printStackTrace);
    }
  }
}