package connector_events

import (
	"encoding/json"
	"fmt"
	"reflect"
	"strings"
	"testing"
	"time"

	wq "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
)

func TestIngressBatchSettlePreservesStoredDeadlinePrecision(t *testing.T) {
	for _, precision := range []int{0, 6} {
		for _, path := range []string{"singleton", "fresh_batch", "duplicate_batch"} {
			t.Run(fmt.Sprintf("precision_%d/%s", precision, path), func(t *testing.T) {
				f := newProjectionAdmissionFixture(t)
				if precision == 6 {
					// Canonical columns are whole-second DATETIME. Also exercise a
					// higher-precision stored value: reading must not narrow it.
					for _, ddl := range []string{
						`ALTER TABLE _System_ConnectorEvent MODIFY NextAttemptAt DATETIME(6) NULL`,
						`ALTER TABLE _System_ConnectorWorkItem MODIFY RunAfter DATETIME(6) NOT NULL`,
					} {
						if _, err := f.db.ExecContext(f.ctx, ddl); err != nil {
							t.Fatal(err)
						}
					}
				}
				f.now = f.now.Add(250123 * time.Microsecond)
				first := f.enqueue("precision-first", false)
				var deadline time.Time
				if err := f.db.QueryRowContext(f.ctx, `SELECT NextAttemptAt FROM _System_ConnectorEvent WHERE Id=?`, first).Scan(&deadline); err != nil {
					t.Fatal(err)
				}
				if (precision == 6 && deadline.Nanosecond() != 250123000) || (precision == 0 && deadline.Nanosecond() != 0) {
					t.Fatalf("fixture did not retain precision %d: %s", precision, deadline.Format(time.RFC3339Nano))
				}

				f.now = f.now.Add(10 * time.Millisecond)
				events := []BatchEvent{projectionAdmissionEvent("precision-second")}
				wantDuplicates := []bool{false}
				switch path {
				case "fresh_batch":
					other := projectionAdmissionEvent("precision-other")
					other.Payload = json.RawMessage(`{"objectKind":"contact","sourceID":"other-source"}`)
					events = append(events, other)
					wantDuplicates = append(wantDuplicates, false)
				case "duplicate_batch":
					// A duplicate routes the whole batch through per-event admission,
					// which shares the same settle loader as fresh batch admission.
					events = append(events, projectionAdmissionEvent("precision-first"))
					wantDuplicates = append(wantDuplicates, true)
				}
				ids, duplicates, err := f.svc.EnqueueBatch(f.ctx, projectionAdmissionNamespace, projectionAdmissionPack, "precision-batch", events)
				if err != nil || len(ids) != len(events) || !reflect.DeepEqual(duplicates, wantDuplicates) {
					t.Fatalf("enqueue: ids=%v duplicates=%v err=%v", ids, duplicates, err)
				}
				var eventDeadline, workDeadline time.Time
				if err := f.db.QueryRowContext(f.ctx, `SELECT e.NextAttemptAt,w.RunAfter
 FROM _System_ConnectorEvent e JOIN _System_ConnectorWorkItem w ON w.EventId=e.Id
 WHERE e.Id=?`, ids[0]).Scan(&eventDeadline, &workDeadline); err != nil {
					t.Fatal(err)
				}
				if !eventDeadline.Equal(deadline) || !workDeadline.Equal(deadline) {
					t.Errorf("stored deadline=%s, sibling=%s, work=%s", deadline.Format(time.RFC3339Nano), eventDeadline.Format(time.RFC3339Nano), workDeadline.Format(time.RFC3339Nano))
				}

				claim := wq.ClaimWorkItemInput{
					InstalledPackID: projectionAdmissionPack, OwnerID: "precision-worker", FencingToken: "precision-fence",
					LeaseToken: "precision-lease", LeaseUntil: deadline.Add(time.Minute), Now: deadline.Add(-time.Microsecond),
				}
				work, claimed, err := f.svc.WorkItemRepository().ClaimNextWorkItem(wq.WithDistributedClaim(f.ctx), claim)
				if err != nil || claimed || work != nil {
					t.Fatalf("claim before stored deadline: work=%+v claimed=%v err=%v", work, claimed, err)
				}
				claim.Now = deadline
				work, claimed, err = f.svc.WorkItemRepository().ClaimNextWorkItem(wq.WithDistributedClaim(f.ctx), claim)
				if err != nil || !claimed || work == nil || work.EventID != ids[0] {
					t.Fatalf("claim at stored deadline: work=%+v claimed=%v err=%v", work, claimed, err)
				}
				if got, want := rangeTestIDs(f, work), []string{first, ids[0]}; !reflect.DeepEqual(got, want) {
					t.Fatalf("claimed projection inputs=%v want=%v", got, want)
				}
				rangeTestComplete(f, work, wq.WorkStateProcessed)
			})
		}
	}
}

func TestSourceProjectionActivationRequiresReplayColumn(t *testing.T) {
	f := newProjectionAdmissionFixture(t)
	gates := map[string]func() error{
		"process":    func() error { return RequireSourceProjectionProtocol(f.ctx, f.db) },
		"standalone": func() error { return f.svc.RequireSourceProjectionProtocol(f.ctx, projectionAdmissionPack) },
	}
	for name, gate := range gates {
		if err := gate(); err != nil {
			t.Fatalf("%s rejects canonical schema: %v", name, err)
		}
	}
	// Native projection requires the complete schema before activation. The
	// ordinary WorkItem rolling-schema contract is tested separately.
	if _, err := f.db.ExecContext(f.ctx, `ALTER TABLE _System_ConnectorWorkItem DROP INDEX uniq_connector_work_item_replay_of, DROP COLUMN ReplayOfWorkId`); err != nil {
		t.Fatal(err)
	}
	for name, gate := range gates {
		if err := gate(); err == nil || !strings.Contains(strings.ToLower(err.Error()), "replayofworkid") {
			t.Fatalf("%s must reject missing ReplayOfWorkId before admission: %v", name, err)
		}
	}
}