// Run after completing the Apache Fluss Flink Quickstart.
//
//	go mod init fluss-go-quickstart
//	go get github.com/pletorco/fluss-go/pkg/fgo@v0.1.0-beta.10
//	go run primary-key-table.go
//
// Set FLUSS_BOOTSTRAP when the Coordinator is not at localhost:9123.
package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math/big"
	"math/rand/v2"
	"os"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

// customerProfile is the JSON-friendly representation printed by this example.
// Decimal values are strings so the display never loses decimal precision.
type customerProfile struct {
	CustomerID     int32  `json:"customer_id"`
	Name           string `json:"name"`
	Membership     string `json:"membership"`
	AccountBalance string `json:"account_balance"`
}

func main() {
	ctx := context.Background()
	log.SetFlags(0)
	const customerID int32 = 999999

	// One client owns the Coordinator and TabletServer connections. Reuse it for
	// all table operations in the same application.
	client, err := fgo.Open(
		ctx,
		fgo.WithBootstrapServers(bootstrapAddress()),
		fgo.WithClientSoftware("quickstart-go", "0.1.0"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := client.Close(); err != nil {
			log.Printf("close Fluss client: %v", err)
		}
	}()

	// Opening the table reads its schema and routing metadata. customer_id is an
	// INT column, so its Go value must be int32.
	profiles, err := client.GetTable(ctx, fgo.TablePath{
		Database: "demo",
		Table:    "customer_profile",
	})
	if err != nil {
		log.Fatal(err)
	}

	// Lookups route directly by primary key rather than scanning every profile.
	lookup, err := client.NewLookuper(ctx, profiles, fgo.WithLookupBatchLimits(100, 4))
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := lookup.Close(); err != nil {
			log.Printf("close lookup client: %v", err)
		}
	}()

	// The Quickstart table has no table.merge-engine, so use the default merge
	// mode. Supplying every column updates the current value for the key.
	writer, err := client.NewUpsertWriter(ctx, profiles, fgo.WithUpsertBatchLimits(1<<20, 500))
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := writer.Close(ctx); err != nil {
			log.Printf("close upsert writer: %v", err)
		}
	}()

	writeProfile := func(profile customerProfile) {
		accountBalance, ok := new(big.Rat).SetString(profile.AccountBalance)
		if !ok {
			log.Fatalf("invalid account_balance: %q", profile.AccountBalance)
		}
		result := writer.Upsert(ctx, fgo.Row{
			profile.CustomerID,
			profile.Name,
			profile.Membership,
			accountBalance,
		}).Await(ctx)
		if result.Err != nil {
			log.Fatal(result.Err)
		}
		if err := writer.Flush(ctx); err != nil {
			log.Fatal(err)
		}
	}

	showProfile := func(step string) {
		// A lookup can report a normal missing key or a real request failure.
		result := lookup.Lookup(ctx, fgo.PrimaryKey{customerID})[0]
		switch {
		case errors.Is(result.Err, fgo.ErrNotFound):
			log.Printf("%s customer %v was not found", step, customerID)
		case result.Err != nil:
			log.Fatal(result.Err)
		default:
			printJSON(step, profileFromRow(result.Row))
		}
	}

	before := customerProfile{
		CustomerID:     customerID,
		Name:           "Quickstart User",
		Membership:     "go-client-before",
		AccountBalance: randomAmount(""),
	}
	after := before
	after.Membership = "go-client-after"
	after.AccountBalance = randomAmount(before.AccountBalance)

	printJSON("[1/4] Initial value to write:", before)
	writeProfile(before)
	showProfile("[2/4] Read the value after the first write:")

	log.Printf("[3/4] Change fields for the same primary key:\nmembership=%s\naccount_balance=%s", after.Membership, after.AccountBalance)
	writeProfile(after)
	showProfile("[4/4] Read the value after the update:")
}

// randomAmount returns a two-decimal amount between 10.00 and 999.99 that is
// different from except. It makes every run show a visible state change.
func randomAmount(except string) string {
	for {
		cents := 1_000 + rand.IntN(99_000)
		amount := fmt.Sprintf("%d.%02d", cents/100, cents%100)
		if amount != except {
			return amount
		}
	}
}

func profileFromRow(row fgo.Row) customerProfile {
	accountBalance, ok := row[3].(*big.Rat)
	if !ok {
		log.Fatalf("unexpected account_balance type: %T", row[3])
	}
	return customerProfile{
		CustomerID:     row[0].(int32),
		Name:           row[1].(string),
		Membership:     row[2].(string),
		AccountBalance: accountBalance.FloatString(2),
	}
}

func printJSON(step string, value customerProfile) {
	encoded, err := json.MarshalIndent(value, "", "  ")
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("%s\n%s", step, encoded)
}

func bootstrapAddress() string {
	if address := os.Getenv("FLUSS_BOOTSTRAP"); address != "" {
		return address
	}
	return "localhost:9123"
}