import java.math.BigDecimal;
import java.time.Duration;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

import org.apache.fluss.client.Connection;
import org.apache.fluss.client.ConnectionFactory;
import org.apache.fluss.client.table.Table;
import org.apache.fluss.client.table.scanner.ScanRecord;
import org.apache.fluss.client.table.scanner.log.LogScanner;
import org.apache.fluss.client.table.scanner.log.ScanRecords;
import org.apache.fluss.client.table.writer.AppendWriter;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.row.BinaryString;
import org.apache.fluss.row.Decimal;
import org.apache.fluss.row.GenericRow;
import org.apache.fluss.row.InternalRow;

// Run after completing the Apache Fluss Flink Quickstart.
//
//   MAVEN_OPTS='--add-opens=java.base/java.nio=ALL-UNNAMED' \
//     mvn compile exec:java -Dexec.mainClass=LogTableExample
//
// Set FLUSS_BOOTSTRAP when the Coordinator is not at localhost:9123.
public final class LogTableExample {
    private static final int CUSTOMER_ID = 999999;

    private LogTableExample() {}

    public static void main(String[] args) throws Exception {
        long orderIdBase = System.currentTimeMillis() * 10;
        List<ExampleEvent> eventsToAppend = List.of(
                new ExampleEvent(orderIdBase, CUSTOMER_ID, new BigDecimal("42.15"), LocalDate.of(2026, 8, 1), "low", "Java Client Clerk"),
                new ExampleEvent(orderIdBase + 1, CUSTOMER_ID, new BigDecimal("85.60"), LocalDate.of(2026, 8, 2), "medium", "Java Client Clerk"),
                new ExampleEvent(orderIdBase + 2, CUSTOMER_ID, new BigDecimal("129.90"), LocalDate.of(2026, 8, 3), "high", "Java Client Clerk"),
                new ExampleEvent(orderIdBase + 3, CUSTOMER_ID, new BigDecimal("24.50"), LocalDate.of(2026, 8, 4), "low", "Java Client Clerk"));

        printEvents("[1/3] Queue these four new events:", eventsToAppend);

        Configuration config = new Configuration();
        config.setString("bootstrap.servers", bootstrapAddress());

        try (Connection connection = ConnectionFactory.createConnection(config);
                Table events = connection.getTable(TablePath.of("demo", "order_events"))) {
            AppendWriter writer = events.newAppend().createWriter();
            for (ExampleEvent event : eventsToAppend) {
                writer.append(eventRow(event));
            }
            writer.flush();
            System.out.println("[2/3] Fluss acknowledged the four append requests.");

            List<ExampleEvent> readEvents = readAppendedEvents(events, eventsToAppend);
            printEvents("[3/3] Read the appended events:", readEvents);
        }
    }

    private static List<ExampleEvent> readAppendedEvents(Table events, List<ExampleEvent> expected)
            throws Exception {
        Set<Long> remaining = expected.stream()
                .map(ExampleEvent::orderId)
                .collect(Collectors.toCollection(LinkedHashSet::new));
        List<ExampleEvent> found = new ArrayList<>();
        long deadline = System.nanoTime() + Duration.ofSeconds(15).toNanos();

        try (LogScanner scanner = events.newScan().createLogScanner()) {
            // The Quickstart creates one non-partitioned bucket.
            scanner.subscribeFromBeginning(0);
            while (!remaining.isEmpty() && System.nanoTime() < deadline) {
                ScanRecords records = scanner.poll(Duration.ofSeconds(1));
                for (ScanRecord record : records) {
                    InternalRow row = record.getRow();
                    if (remaining.remove(row.getLong(0))) {
                        found.add(eventFromRow(row));
                    }
                }
            }
        }

        if (!remaining.isEmpty()) {
            throw new IllegalStateException("did not read every appended event: " + remaining);
        }
        return found;
    }

    private static GenericRow eventRow(ExampleEvent event) {
        return GenericRow.of(
                event.orderId(),
                event.customerId(),
                Decimal.fromBigDecimal(event.totalPrice(), 15, 2),
                (int) event.orderedOn().toEpochDay(),
                BinaryString.fromString(event.orderPriority()),
                BinaryString.fromString(event.clerk()));
    }

    private static ExampleEvent eventFromRow(InternalRow row) {
        return new ExampleEvent(
                row.getLong(0),
                row.getInt(1),
                row.getDecimal(2, 15, 2).toBigDecimal(),
                LocalDate.ofEpochDay(row.getInt(3)),
                row.getString(4).toString(),
                row.getString(5).toString());
    }

    private static void printEvents(String step, List<ExampleEvent> events) {
        System.out.println(step);
        System.out.println("[");
        for (int i = 0; i < events.size(); i++) {
            System.out.printf("  %s%s%n", eventJson(events.get(i)), i + 1 == events.size() ? "" : ",");
        }
        System.out.println("]");
    }

    private static String eventJson(ExampleEvent event) {
        return """
                {
                  "order_id": %d,
                  "customer_id": %d,
                  "total_price": "%s",
                  "ordered_on": "%s",
                  "order_priority": "%s",
                  "clerk": "%s"
                }"""
                .formatted(
                        event.orderId(),
                        event.customerId(),
                        event.totalPrice().toPlainString(),
                        event.orderedOn(),
                        event.orderPriority(),
                        event.clerk())
                .replace("\n", "\n  ");
    }

    private static String bootstrapAddress() {
        String address = System.getenv("FLUSS_BOOTSTRAP");
        return address == null || address.isBlank() ? "localhost:9123" : address;
    }

    private record ExampleEvent(
            long orderId,
            int customerId,
            BigDecimal totalPrice,
            LocalDate orderedOn,
            String orderPriority,
            String clerk) {}
}