// Command importer loads deterministic staged CSV data into PostgreSQL
// (migration-plan Phase 7 items 5-7). Modes:
//
//	dry-run  parse + validate every CSV, report counts; no writes
//	apply    load tables in manifest (FK-safe) order inside one transaction,
//	         preserving exact IDs, then reset identity sequences
//	verify   recompute per-table count / max ID / column sums from the CSVs
//	         and PostgreSQL, print a machine-readable JSON report, exit
//	         non-zero on any mismatch
//
// The staged directory must contain manifest.json:
//
//	{"tables":[{"table":"users","file":"users.csv","id_column":"id",
//	            "sum_columns":["balance"]}]}
//
// Each CSV's header row names the target columns; empty fields load as NULL.
package main

import (
	"context"
	"encoding/csv"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"

	"github.com/jackc/pgx/v5"
	"github.com/shopspring/decimal"

	"github.com/TaruLab/keluargamart-backend/pkg/database"
)

type tableSpec struct {
	Table      string   `json:"table"`
	File       string   `json:"file"`
	IDColumn   string   `json:"id_column"`
	SumColumns []string `json:"sum_columns"`
}

type manifest struct {
	Tables []tableSpec `json:"tables"`
}

type tableReport struct {
	Table    string                     `json:"table"`
	Rows     int64                      `json:"rows"`
	MaxID    *int64                     `json:"max_id,omitempty"`
	Sums     map[string]decimal.Decimal `json:"sums,omitempty"`
}

type report struct {
	Source []tableReport `json:"source"`
	Target []tableReport `json:"target,omitempty"`
	Match  bool          `json:"match"`
}

func main() {
	mode := flag.String("mode", "dry-run", "dry-run | apply | verify")
	dir := flag.String("dir", "", "staged CSV directory containing manifest.json")
	dbURL := flag.String("db", os.Getenv("DATABASE_URL"), "PostgreSQL URL (defaults to $DATABASE_URL)")
	flag.Parse()

	if *dir == "" {
		fatal("usage: importer -mode dry-run|apply|verify -dir <staged-dir> [-db <url>]")
	}
	m, err := loadManifest(*dir)
	if err != nil {
		fatal(err.Error())
	}

	switch *mode {
	case "dry-run":
		rep, err := scanAll(*dir, m)
		if err != nil {
			fatal(err.Error())
		}
		rep.Match = true
		emit(rep)
	case "apply":
		if *dbURL == "" {
			fatal("apply requires -db (or DATABASE_URL); refusing to run without an explicit target")
		}
		if err := apply(*dir, *dbURL, m); err != nil {
			fatal(err.Error())
		}
		fmt.Fprintln(os.Stderr, "apply: OK")
	case "verify":
		if *dbURL == "" {
			fatal("verify requires -db (or DATABASE_URL)")
		}
		rep, err := verify(*dir, *dbURL, m)
		if err != nil {
			fatal(err.Error())
		}
		emit(rep)
		if !rep.Match {
			os.Exit(1)
		}
	default:
		fatal("unknown mode: " + *mode)
	}
}

func loadManifest(dir string) (*manifest, error) {
	b, err := os.ReadFile(filepath.Join(dir, "manifest.json"))
	if err != nil {
		return nil, fmt.Errorf("read manifest: %w", err)
	}
	var m manifest
	if err := json.Unmarshal(b, &m); err != nil {
		return nil, fmt.Errorf("parse manifest: %w", err)
	}
	if len(m.Tables) == 0 {
		return nil, fmt.Errorf("manifest lists no tables")
	}
	return &m, nil
}

// openCSV returns a reader positioned after the header, plus the column names.
func openCSV(dir string, spec tableSpec) (*csv.Reader, []string, io.Closer, error) {
	f, err := os.Open(filepath.Join(dir, spec.File))
	if err != nil {
		return nil, nil, nil, err
	}
	r := csv.NewReader(f)
	r.FieldsPerRecord = -1
	header, err := r.Read()
	if err != nil {
		f.Close()
		return nil, nil, nil, fmt.Errorf("%s: read header: %w", spec.File, err)
	}
	return r, header, f, nil
}

// scanAll validates every CSV and reports source-side stats (no DB access).
func scanAll(dir string, m *manifest) (*report, error) {
	rep := &report{Source: make([]tableReport, 0, len(m.Tables))}
	for _, spec := range m.Tables {
		tr, err := scanTable(dir, spec)
		if err != nil {
			return nil, err
		}
		rep.Source = append(rep.Source, tr)
	}
	return rep, nil
}

func scanTable(dir string, spec tableSpec) (tableReport, error) {
	r, header, closer, err := openCSV(dir, spec)
	if err != nil {
		return tableReport{}, err
	}
	defer closer.Close()

	tr := tableReport{Table: spec.Table, Sums: map[string]decimal.Decimal{}}
	sumIdx := map[int]string{}
	for _, col := range spec.SumColumns {
		for i, h := range header {
			if h == col {
				sumIdx[i] = col
			}
		}
	}
	idIdx := -1
	if spec.IDColumn != "" {
		for i, h := range header {
			if h == spec.IDColumn {
				idIdx = i
			}
		}
	}
	for {
		row, err := r.Read()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			return tableReport{}, fmt.Errorf("%s: %w", spec.File, err)
		}
		if len(row) != len(header) {
			return tableReport{}, fmt.Errorf("%s: row %d has %d fields, want %d", spec.File, tr.Rows+2, len(row), len(header))
		}
		tr.Rows++
		if idIdx >= 0 && row[idIdx] != "" {
			var id int64
			if _, err := fmt.Sscanf(row[idIdx], "%d", &id); err == nil {
				if tr.MaxID == nil || id > *tr.MaxID {
					idCopy := id
					tr.MaxID = &idCopy
				}
			}
		}
		for i, col := range sumIdx {
			if i < len(row) && row[i] != "" {
				d, err := decimal.NewFromString(row[i])
				if err != nil {
					return tableReport{}, fmt.Errorf("%s: column %s row %d: %w", spec.File, col, tr.Rows+1, err)
				}
				tr.Sums[col] = tr.Sums[col].Add(d)
			}
		}
	}
	return tr, nil
}

// apply loads every table in manifest order inside one transaction.
func apply(dir, dbURL string, m *manifest) error {
	ctx := context.Background()
	pool, err := database.New(ctx, dbURL)
	if err != nil {
		return err
	}
	defer pool.Close()

	tx, err := pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx)

	for _, spec := range m.Tables {
		if err := copyTable(ctx, tx, dir, spec); err != nil {
			return fmt.Errorf("%s: %w", spec.Table, err)
		}
		if spec.IDColumn != "" {
			q := fmt.Sprintf(`SELECT setval(pg_get_serial_sequence('%s','%s'), COALESCE((SELECT MAX(%s) FROM %s),1), true)`,
				spec.Table, spec.IDColumn, spec.IDColumn, spec.Table)
			if _, err := tx.Exec(ctx, q); err != nil {
				return fmt.Errorf("%s: reset sequence: %w", spec.Table, err)
			}
		}
	}
	return tx.Commit(ctx)
}

// copyTable streams the CSV (minus header) into COPY ... FROM STDIN in text
// CSV format so PostgreSQL coerces each field into the column type. Empty
// fields become NULL via FORCE_NULL on every column. Identifiers are quoted
// with pgx.Identifier; the manifest is a trusted local config file.
func copyTable(ctx context.Context, tx pgx.Tx, dir string, spec tableSpec) error {
	r, header, closer, err := openCSV(dir, spec)
	if err != nil {
		return err
	}
	defer closer.Close()

	pr, pw := io.Pipe()
	go func() {
		w := csv.NewWriter(pw)
		var werr error
		for {
			row, rerr := r.Read()
			if errors.Is(rerr, io.EOF) {
				break
			}
			if rerr != nil {
				werr = rerr
				break
			}
			if werr = w.Write(row); werr != nil {
				break
			}
		}
		w.Flush()
		if ferr := w.Error(); werr == nil {
			werr = ferr
		}
		pw.CloseWithError(werr)
	}()

	quoted := make([]string, len(header))
	for i, h := range header {
		quoted[i] = pgx.Identifier{h}.Sanitize()
	}
	sql := fmt.Sprintf(`COPY %s (%s) FROM STDIN WITH (FORMAT csv, FORCE_NULL (%s))`,
		pgx.Identifier{spec.Table}.Sanitize(),
		strings.Join(quoted, ","),
		strings.Join(quoted, ","))
	_, err = tx.Conn().PgConn().CopyFrom(ctx, pr, sql)
	return err
}

// verify compares source CSV stats against the target database.
func verify(dir, dbURL string, m *manifest) (*report, error) {
	src, err := scanAll(dir, m)
	if err != nil {
		return nil, err
	}
	ctx := context.Background()
	pool, err := database.New(ctx, dbURL)
	if err != nil {
		return nil, err
	}
	defer pool.Close()

	rep := &report{Source: src.Source, Target: make([]tableReport, 0, len(m.Tables)), Match: true}
	for i, spec := range m.Tables {
		tr := tableReport{Table: spec.Table, Sums: map[string]decimal.Decimal{}}
		if err := pool.QueryRow(ctx,
			fmt.Sprintf(`SELECT count(*) FROM %s`, pgx.Identifier{spec.Table}.Sanitize())).
			Scan(&tr.Rows); err != nil {
			return nil, fmt.Errorf("%s: count: %w", spec.Table, err)
		}
		if spec.IDColumn != "" {
			var maxID *int64
			if err := pool.QueryRow(ctx,
				fmt.Sprintf(`SELECT MAX(%s) FROM %s`, pgx.Identifier{spec.IDColumn}.Sanitize(), pgx.Identifier{spec.Table}.Sanitize())).
				Scan(&maxID); err != nil {
				return nil, fmt.Errorf("%s: max id: %w", spec.Table, err)
			}
			tr.MaxID = maxID
		}
		for _, col := range spec.SumColumns {
			var sum decimal.Decimal
			if err := pool.QueryRow(ctx,
				fmt.Sprintf(`SELECT COALESCE(SUM(%s),0) FROM %s`, pgx.Identifier{col}.Sanitize(), pgx.Identifier{spec.Table}.Sanitize())).
				Scan(&sum); err != nil {
				return nil, fmt.Errorf("%s: sum %s: %w", spec.Table, col, err)
			}
			tr.Sums[col] = sum
		}
		rep.Target = append(rep.Target, tr)
		if !sameReport(src.Source[i], tr) {
			rep.Match = false
		}
	}
	return rep, nil
}

func sameReport(a, b tableReport) bool {
	if a.Rows != b.Rows {
		return false
	}
	if (a.MaxID == nil) != (b.MaxID == nil) {
		return false
	}
	if a.MaxID != nil && *a.MaxID != *b.MaxID {
		return false
	}
	for col, sum := range a.Sums {
		if !b.Sums[col].Equal(sum) {
			return false
		}
	}
	return true
}

func emit(rep *report) {
	enc := json.NewEncoder(os.Stdout)
	enc.SetIndent("", "  ")
	if err := enc.Encode(rep); err != nil {
		fatal(err.Error())
	}
}

func fatal(msg string) {
	fmt.Fprintln(os.Stderr, "importer: "+msg)
	os.Exit(1)
}