package sales

import (
	"context"
	"fmt"
	"os"
	"testing"
	"time"

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

	"github.com/TaruLab/keluargamart-backend/internal/db"
)

// The candidate predicate is SQL, and the sales harness is a fake store that
// never interprets SQL — a fake-based candidate test would assert nothing about
// which rows the sweep actually selects. So this file is the
// DATABASE_URL-gated pattern from internal/finance/repository_integration_test.go,
// run against a real PostgreSQL in a throwaway schema.
//
// It also executes migration 000102's body rather than assuming it: the column
// and index the candidate query reads are created by the migration here, so a
// migration that failed to add the column fails this test.

const snapSweepInvoicesDDL = `CREATE TABLE invoices (
	id BIGSERIAL PRIMARY KEY,
	invoice_number TEXT NOT NULL,
	sales_order_id BIGINT REFERENCES sales_orders(id),
	customer_id BIGINT NOT NULL,
	invoice_date DATE NOT NULL,
	due_date DATE NOT NULL,
	status TEXT NOT NULL DEFAULT 'unpaid',
	subtotal NUMERIC(15,2) NOT NULL DEFAULT 0,
	tax_amount NUMERIC(15,2) NOT NULL DEFAULT 0,
	total_amount NUMERIC(15,2) NOT NULL DEFAULT 0,
	paid_amount NUMERIC(15,2) NOT NULL DEFAULT 0,
	remaining_amount NUMERIC(15,2) NOT NULL DEFAULT 0,
	snap_order_ref TEXT,
	created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
	updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
)`

// snapSweepInvoicesOnlyDDL is the same table without the sales_orders
// reference, for the tests that do not exercise the candidate predicate (which
// is the only query that joins it).
const snapSweepInvoicesOnlyDDL = `CREATE TABLE invoices (
	id BIGSERIAL PRIMARY KEY,
	invoice_number TEXT NOT NULL,
	sales_order_id BIGINT,
	customer_id BIGINT NOT NULL,
	invoice_date DATE NOT NULL,
	due_date DATE NOT NULL,
	status TEXT NOT NULL DEFAULT 'unpaid',
	subtotal NUMERIC(15,2) NOT NULL DEFAULT 0,
	tax_amount NUMERIC(15,2) NOT NULL DEFAULT 0,
	total_amount NUMERIC(15,2) NOT NULL DEFAULT 0,
	paid_amount NUMERIC(15,2) NOT NULL DEFAULT 0,
	remaining_amount NUMERIC(15,2) NOT NULL DEFAULT 0,
	snap_order_ref TEXT,
	created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
	updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
)`

// scratchRepo builds a real Repository whose every connection sees only a
// throwaway schema, so the test neither reads nor writes the developer's data.
// The schema is created and dropped on the base pool (committed, so it is
// visible to the scratch pool's own connections).
func scratchRepo(t *testing.T, ddl []string) (*Repository, *pgxpool.Pool) {
	t.Helper()
	databaseURL := os.Getenv("DATABASE_URL")
	if databaseURL == "" {
		t.Skip("DATABASE_URL is not set")
	}
	ctx := context.Background()

	base, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	schema := fmt.Sprintf("snap_sweep_probe_%d", time.Now().UnixNano())
	if _, err := base.Exec(ctx, `CREATE SCHEMA `+schema); err != nil {
		base.Close()
		t.Fatal(err)
	}

	cfg, err := pgxpool.ParseConfig(databaseURL)
	if err != nil {
		base.Close()
		t.Fatal(err)
	}
	cfg.ConnConfig.RuntimeParams["search_path"] = schema
	pool, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		base.Close()
		t.Fatal(err)
	}
	t.Cleanup(func() {
		pool.Close()
		if _, err := base.Exec(context.Background(), `DROP SCHEMA IF EXISTS `+schema+` CASCADE`); err != nil {
			t.Logf("drop schema %s: %v", schema, err)
		}
		base.Close()
	})

	for _, stmt := range ddl {
		if _, err := pool.Exec(ctx, stmt); err != nil {
			t.Fatalf("%s: %v", stmt, err)
		}
	}
	return &Repository{pool: pool, q: db.New(pool)}, pool
}

var snapSweepBaseDDL = []string{
	`CREATE TABLE customers (id BIGSERIAL PRIMARY KEY, code TEXT NOT NULL, name TEXT NOT NULL)`,
	`CREATE TABLE sales_orders (
		id BIGSERIAL PRIMARY KEY,
		so_number TEXT NOT NULL,
		customer_id BIGINT NOT NULL REFERENCES customers(id),
		order_date DATE NOT NULL,
		status TEXT NOT NULL DEFAULT 'draft',
		order_type TEXT NOT NULL DEFAULT 'regular'
	)`,
	snapSweepInvoicesDDL,
	// The status and order_type checks, as the real schema declares them.
	`ALTER TABLE invoices ADD CONSTRAINT invoices_status_check
		CHECK (status IN ('unpaid','partial','paid','overdue','cancelled'))`,
	`ALTER TABLE sales_orders ADD CONSTRAINT sales_orders_order_type_check
		CHECK (order_type IN ('regular','pos','consignment','ecommerce'))`,
}

func TestRepositoryListSnapSweepCandidatesAgainstPostgres(t *testing.T) {
	repo, pool := scratchRepo(t, append(append([]string{}, snapSweepBaseDDL...),
		// Migration 000102's body, verbatim, applied to the pre-migration shape.
		`ALTER TABLE invoices ADD COLUMN IF NOT EXISTS snap_last_checked_at TIMESTAMPTZ`,
		`CREATE INDEX IF NOT EXISTS invoices_snap_sweep_idx
			ON invoices (updated_at)
			WHERE status IN ('unpaid', 'partial') AND snap_order_ref IS NOT NULL`))

	ctx := context.Background()
	if _, err := pool.Exec(ctx, `INSERT INTO customers (code, name) VALUES ('SWEEP-1', 'Sweep Probe')`); err != nil {
		t.Fatal(err)
	}
	var customerID int64
	if err := pool.QueryRow(ctx, `SELECT id FROM customers WHERE code = 'SWEEP-1'`).Scan(&customerID); err != nil {
		t.Fatal(err)
	}

	cases := []struct {
		name          string
		orderStatus   string
		orderType     string
		invStatus     string
		ref           *string
		age           time.Duration
		checkedAgo    *time.Duration
		wantCandidate bool
		why           string
	}{
		{
			name:        "open snap invoice inside the horizon and unthrottled is a candidate",
			orderStatus: "confirmed", orderType: "ecommerce",
			invStatus: "unpaid", ref: strPtr("KM-SWEEP-A-1"), age: time.Hour,
			wantCandidate: true, why: "the happy path",
		},
		{
			name:        "partially paid is still a candidate",
			orderStatus: "confirmed", orderType: "ecommerce",
			invStatus: "partial", ref: strPtr("KM-SWEEP-A-2"), age: time.Hour,
			wantCandidate: true, why: "partial is an open status",
		},
		{
			name:        "paid invoice is not a candidate",
			orderStatus: "confirmed", orderType: "ecommerce",
			invStatus: "paid", ref: strPtr("KM-SWEEP-B-1"), age: time.Hour,
			why: "settled money must never be re-probed",
		},
		{
			name:        "invoice with no snap ref is not a candidate",
			orderStatus: "confirmed", orderType: "ecommerce",
			invStatus: "unpaid", ref: nil, age: time.Hour,
			why: "nothing to ask Midtrans about",
		},
		{
			name:        "invoice older than the 12 hour horizon is not a candidate",
			orderStatus: "confirmed", orderType: "ecommerce",
			invStatus: "unpaid", ref: strPtr("KM-SWEEP-B-2"), age: 13 * time.Hour,
			why: "the horizon drains the set",
		},
		{
			name:        "invoice probed within the throttle window is not a candidate",
			orderStatus: "confirmed", orderType: "ecommerce",
			invStatus: "unpaid", ref: strPtr("KM-SWEEP-B-3"), age: time.Hour,
			checkedAgo: durPtr(5 * time.Minute),
			why:        "the throttle is the point of the column",
		},
		{
			name:        "invoice probed longer ago than the throttle window is a candidate again",
			orderStatus: "confirmed", orderType: "ecommerce",
			invStatus: "unpaid", ref: strPtr("KM-SWEEP-B-4"), age: time.Hour,
			checkedAgo: durPtr(11 * time.Minute), wantCandidate: true,
			why: "the throttle expires",
		},
		{
			name:        "cancelled order's invoice is not a candidate",
			orderStatus: "cancelled", orderType: "ecommerce",
			invStatus: "unpaid", ref: strPtr("KM-SWEEP-B-5"), age: time.Hour,
			why: "written-off debt",
		},
		{
			name:        "non ecommerce order's invoice is not a candidate",
			orderStatus: "confirmed", orderType: "pos",
			invStatus: "unpaid", ref: strPtr("KM-SWEEP-B-6"), age: time.Hour,
			why: "only Snap invoices exist online",
		},
	}

	want := map[string]bool{}
	for i, tc := range cases {
		var orderID int64
		soNumber := fmt.Sprintf("SWEEP-SO-%d", i)
		if err := pool.QueryRow(ctx, `INSERT INTO sales_orders (so_number, customer_id, order_date, status, order_type)
			VALUES ($1, $2, CURRENT_DATE, $3, $4) RETURNING id`,
			soNumber, customerID, tc.orderStatus, tc.orderType).Scan(&orderID); err != nil {
			t.Fatalf("%s: %s: %v", tc.name, tc.why, err)
		}
		number := fmt.Sprintf("SWEEP-INV-%d", i)
		// updated_at carries the age, so the horizon is exercised for real
		// rather than approximated.
		if _, err := pool.Exec(ctx, `INSERT INTO invoices
			(invoice_number, sales_order_id, customer_id, invoice_date, due_date, status,
			 subtotal, tax_amount, total_amount, paid_amount, remaining_amount, snap_order_ref, updated_at)
			VALUES ($1, $2, $3, CURRENT_DATE, CURRENT_DATE, $4, 0, 0, 100, 0, 100, $5,
			        now() - make_interval(secs => $6))`,
			number, orderID, customerID, tc.invStatus, tc.ref, tc.age.Seconds()); err != nil {
			t.Fatalf("%s: %s: %v", tc.name, tc.why, err)
		}
		if tc.checkedAgo != nil {
			if _, err := pool.Exec(ctx, `UPDATE invoices SET snap_last_checked_at = now() - make_interval(secs => $2)
				WHERE invoice_number = $1`, number, tc.checkedAgo.Seconds()); err != nil {
				t.Fatalf("%s: %s: %v", tc.name, tc.why, err)
			}
		}
		if tc.wantCandidate {
			want[number] = true
		}
	}

	got, err := repo.ListSnapSweepCandidates(ctx, 100)
	if err != nil {
		t.Fatal(err)
	}
	gotNames := map[string]bool{}
	for _, c := range got {
		gotNames[c.InvoiceNumber] = true
	}
	for name := range want {
		if !gotNames[name] {
			t.Errorf("candidate %s missing: a real settlement would never be found", name)
		}
	}
	for name := range gotNames {
		if !want[name] {
			t.Errorf("candidate %s returned but must be excluded: the sweep would probe an invoice it must not touch", name)
		}
	}
	if len(got) != len(want) {
		t.Errorf("candidates = %d (%v), want %d (%v)", len(got), gotNames, len(want), want)
	}

	// Oldest activity first: the rows were inserted in order, so ascending ID
	// is ascending updated_at here.
	for i := 1; i < len(got); i++ {
		if got[i-1].ID > got[i].ID {
			t.Errorf("candidates are not ordered by updated_at ASC: id %d came before %d", got[i-1].ID, got[i].ID)
		}
	}

	// The limit is a bound parameter, and a bad caller degrades to the ceiling
	// rather than to a silent no-op.
	one, err := repo.ListSnapSweepCandidates(ctx, 1)
	if err != nil {
		t.Fatal(err)
	}
	if len(one) != 1 {
		t.Errorf("limit 1 returned %d candidates, want 1", len(one))
	}
	for _, limit := range []int{0, -1, snapSweepMaxBatch + 1} {
		capped, err := repo.ListSnapSweepCandidates(ctx, limit)
		if err != nil {
			t.Fatalf("limit %d: %v", limit, err)
		}
		if len(capped) != len(want) {
			t.Errorf("limit %d returned %d candidates, want the clamped %d — a bad caller must not scan unbounded or nothing", limit, len(capped), len(want))
		}
	}
}

// MarkInvoicePaid's row count, proved against real rows. The count is the
// entire deliverable: a caller that cannot tell 0 from 1 will write a payment
// row for an invoice that did not move.
func TestRepositoryMarkInvoicePaidRowCountAgainstPostgres(t *testing.T) {
	repo, pool := scratchRepo(t, []string{snapSweepInvoicesOnlyDDL})
	ctx := context.Background()
	amount := decimal.NewFromInt(200)

	insert := func(number, status string) int64 {
		var id int64
		if err := pool.QueryRow(ctx, `INSERT INTO invoices
			(invoice_number, customer_id, invoice_date, due_date, status, total_amount, remaining_amount)
			VALUES ($1, 1, CURRENT_DATE, CURRENT_DATE, $2, 200, $3) RETURNING id`,
			number, status, amount).Scan(&id); err != nil {
			t.Fatal(err)
		}
		return id
	}
	openID := insert("MARKPAID-OPEN", "unpaid")
	cancelledID := insert("MARKPAID-CANCELLED", "cancelled")

	n, err := repo.MarkInvoicePaid(ctx, pool, openID, amount)
	if err != nil {
		t.Fatal(err)
	}
	if n != 1 {
		t.Errorf("settleable invoice: rows = %d, want 1", n)
	}
	// Same row again: the status predicate now refuses it, and the caller MUST
	// be able to see that.
	n, err = repo.MarkInvoicePaid(ctx, pool, openID, amount)
	if err != nil {
		t.Fatal(err)
	}
	if n != 0 {
		t.Errorf("already-paid invoice: rows = %d, want 0 — a caller reading 1 would settle a second time", n)
	}
	n, err = repo.MarkInvoicePaid(ctx, pool, cancelledID, amount)
	if err != nil {
		t.Fatal(err)
	}
	if n != 0 {
		t.Errorf("cancelled invoice: rows = %d, want 0 — a written-off invoice must never be revived", n)
	}
}

// SetInvoiceSnapChecked round-trips through the real column and rides the
// caller's transaction: invisible after a rollback, visible after a commit,
// and never touching updated_at in either case.
func TestRepositorySetInvoiceSnapCheckedRoundTripAgainstPostgres(t *testing.T) {
	repo, pool := scratchRepo(t, []string{
		snapSweepInvoicesOnlyDDL,
		`ALTER TABLE invoices ADD COLUMN IF NOT EXISTS snap_last_checked_at TIMESTAMPTZ`,
	})
	ctx := context.Background()

	// The probe row is inserted in its OWN committed transaction: the stamp's
	// transaction must be the only thing under test, so a rollback of that
	// transaction cannot take the row it stamps with it.
	if _, err := pool.Exec(ctx, `INSERT INTO invoices (invoice_number, customer_id, invoice_date, due_date)
		VALUES ('STAMP-1', 1, CURRENT_DATE, CURRENT_DATE)`); err != nil {
		t.Fatal(err)
	}
	var id int64
	if err := pool.QueryRow(ctx, `SELECT id FROM invoices WHERE invoice_number = 'STAMP-1'`).Scan(&id); err != nil {
		t.Fatal(err)
	}
	beginTx := func() pgx.Tx {
		t.Helper()
		tx, err := pool.Begin(ctx)
		if err != nil {
			t.Fatal(err)
		}
		return tx
	}
	read := func() *time.Time {
		t.Helper()
		var at *time.Time
		if err := pool.QueryRow(ctx, `SELECT snap_last_checked_at FROM invoices WHERE invoice_number = 'STAMP-1'`).Scan(&at); err != nil {
			t.Fatal(err)
		}
		return at
	}
	readUpdated := func() time.Time {
		t.Helper()
		var at time.Time
		if err := pool.QueryRow(ctx, `SELECT updated_at FROM invoices WHERE invoice_number = 'STAMP-1'`).Scan(&at); err != nil {
			t.Fatal(err)
		}
		return at
	}

	// Rollback path: the stamp must vanish with the settle it belonged to.
	txRoll := beginTx()
	if err := repo.SetInvoiceSnapChecked(ctx, txRoll, id); err != nil {
		t.Fatal(err)
	}
	if err := txRoll.Rollback(ctx); err != nil {
		t.Fatal(err)
	}
	afterRoll := read()
	if afterRoll != nil {
		t.Errorf("snap_last_checked_at = %v after rollback, want NULL — the stamp must not outlive a rolled-back settle", afterRoll)
	}

	// Commit path, and updated_at must not move on either path.
	beforeCommit := readUpdated()
	txCommit := beginTx()
	if err := repo.SetInvoiceSnapChecked(ctx, txCommit, id); err != nil {
		t.Fatal(err)
	}
	if err := txCommit.Commit(ctx); err != nil {
		t.Fatal(err)
	}
	afterCommit := read()
	if afterCommit == nil {
		t.Error("snap_last_checked_at is NULL after commit — a committed 'no settlement' would never back off")
	}
	if got := readUpdated(); !got.Equal(beforeCommit) {
		t.Errorf("updated_at moved %v -> %v: a read-only probe must not touch business timestamps", beforeCommit, got)
	}
}

// strPtr already exists in service_test.go (same package); only the duration
// helper is local to this file.
func durPtr(d time.Duration) *time.Duration { return &d }