//go:build failpoint

package meta

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"

	"github.com/pingcap/failpoint"
)

// TestWriteBillingHourCommitFailureLeavesNothing proves the atomicity claim:
// a failure after the ledger/outbox writes but before commit leaves no ledger
// row, no outbox row, and no checkpoint advance; a retry then produces
// exactly one billable representation of the hour.
func TestWriteBillingHourCommitFailureLeavesNothing(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	hour := time.Date(2026, 9, 4, 8, 0, 0, 0, time.UTC)

	rows := []BillingUsageLedgerRow{{
		CostID: "cost-atomic-fp", TenantID: "t1", OrgID: "42",
		Unit: BillingUnitReadOps, PeriodStart: hour, Quantity: 5,
		Labels: map[string]string{"traffic_path": "db9"}, Status: BillingLedgerStatusComplete,
	}}
	outbox := []BillingOutboxRow{{CostID: "cost-atomic-fp", Payload: []byte(`{}`)}}

	var fired atomic.Bool
	err := failpoint.EnableCall("github.com/mem9-ai/drive9/pkg/meta/billingHourBeforeCommit", func(injected *error) {
		if fired.CompareAndSwap(false, true) {
			*injected = errors.New("injected billing hour commit failure")
		}
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = failpoint.Disable("github.com/mem9-ai/drive9/pkg/meta/billingHourBeforeCommit") })

	if err := s.WriteBillingHourWithOutbox(ctx, rows, outbox, hour); err == nil {
		t.Fatal("expected injected failure")
	}
	var ledgerCount, outboxCount int
	if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM billing_usage_ledger WHERE cost_id = 'cost-atomic-fp'`).Scan(&ledgerCount); err != nil {
		t.Fatal(err)
	}
	if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM billing_sqs_outbox WHERE cost_id = 'cost-atomic-fp'`).Scan(&outboxCount); err != nil {
		t.Fatal(err)
	}
	if _, ok, err := s.BillingCollectCheckpoint(ctx); err != nil || ok {
		t.Fatalf("checkpoint advanced on failed close: ok=%v err=%v", ok, err)
	}
	if ledgerCount != 0 || outboxCount != 0 {
		t.Fatalf("ledger=%d outbox=%d after failed close, want 0/0", ledgerCount, outboxCount)
	}

	if err := s.WriteBillingHourWithOutbox(ctx, rows, outbox, hour); err != nil {
		t.Fatalf("retry: %v", err)
	}
	if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM billing_usage_ledger WHERE cost_id = 'cost-atomic-fp'`).Scan(&ledgerCount); err != nil {
		t.Fatal(err)
	}
	if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM billing_sqs_outbox WHERE cost_id = 'cost-atomic-fp'`).Scan(&outboxCount); err != nil {
		t.Fatal(err)
	}
	if ledgerCount != 1 || outboxCount != 1 {
		t.Fatalf("after retry ledger=%d outbox=%d, want exactly 1/1", ledgerCount, outboxCount)
	}
	last, ok, err := s.BillingCollectCheckpoint(ctx)
	if err != nil || !ok || !last.Equal(hour) {
		t.Fatalf("checkpoint = %v/%v/%v, want %v", last, ok, err, hour)
	}
}