package service_test

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

	"git.sendico.io/tech/sxchnge/notifications/bus"
	"git.sendico.io/tech/sxchnge/notifications/email"
	pb "git.sendico.io/tech/sxchnge/notifications/interface/notifications/v1"
	"git.sendico.io/tech/sxchnge/notifications/model"
	"git.sendico.io/tech/sxchnge/notifications/service"
	"git.sendico.io/tech/sxchnge/notifications/store"
	"git.sendico.io/tech/sxchnge/notifications/templates"
	db "git.sendico.io/tech/sxchnge/shared/database"
	"git.sendico.io/tech/sxchnge/shared/database/repository"
	merrors "git.sendico.io/tech/sxchnge/shared/errors"
	"git.sendico.io/tech/sxchnge/shared/messaging"
	"git.sendico.io/tech/sxchnge/shared/messaging/reliable"
	msgruntime "git.sendico.io/tech/sxchnge/shared/messaging/runtime"
	"github.com/testcontainers/testcontainers-go"
	"github.com/testcontainers/testcontainers-go/modules/mongodb"
	"github.com/testcontainers/testcontainers-go/wait"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type sender struct {
	calls    atomic.Int32
	mu       sync.Mutex
	messages map[string]model.Email
	state    model.State
}

func (s *sender) Send(_ context.Context, id string, message model.Email) email.Outcome {
	s.mu.Lock()
	if s.messages == nil {
		s.messages = map[string]model.Email{}
	}
	s.messages[id] = message
	s.mu.Unlock()
	s.calls.Add(1)
	return email.Outcome{State: s.state, Code: "test", RetryAfter: time.Millisecond}
}
func TestDurableDelivery(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	defer cancel()
	container, err := mongodb.Run(ctx, "mongo:latest")
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = container.Terminate(context.Background()) }()
	uri, err := container.ConnectionString(ctx)
	if err != nil {
		t.Fatal(err)
	}
	conn, err := db.ConnectMongo(zap.NewNop(), &db.Config{Driver: db.Mongo, Settings: map[string]any{"uri": uri, "database": "notifications_test"}})
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Disconnect(context.Background())
	repo, err := store.New(zap.NewNop(), conn.Database())
	if err != nil {
		t.Fatal(err)
	}
	catalog, err := templates.New([]templates.Definition{{ID: "test", Version: "1", DefaultLocale: "en", Parameters: []string{"name"}, Locales: map[string]templates.Content{"en": {Subject: "Hello", HTML: "<p>{{.name}}</p>"}}}})
	if err != nil {
		t.Fatal(err)
	}
	driver := &sender{state: model.Accepted}
	svc := service.New(repo, catalog, driver, "resend", zap.NewNop())
	request := model.Request{Key: "duplicate", UserID: "u", TemplateID: "test", TemplateVersion: "1", Parameters: map[string]string{"name": "A & B"}, Address: "user@example.com"}
	var wg sync.WaitGroup
	ids := make(chan string, 10)
	for range 10 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			n, e := svc.Send(ctx, "product", request)
			if e != nil {
				t.Error(e)
				return
			}
			ids <- n.ID.Hex()
		}()
	}
	wg.Wait()
	close(ids)
	id := ""
	for value := range ids {
		if id != "" && value != id {
			t.Error("idempotency race")
		}
		id = value
	}
	if id == "" {
		t.Fatal("no notification")
	}
	request.Parameters = map[string]string{"name": "Changed"}
	if _, err = svc.Send(ctx, "product", request); !errors.Is(err, merrors.ErrDataConflict) {
		t.Fatal("missing conflict", err)
	}
	if _, err = svc.Get(ctx, "other", id); !errors.Is(err, merrors.ErrNoData) {
		t.Fatal("caller isolation failed", err)
	}
	stored, err := svc.Get(ctx, "product", id)
	if err != nil || stored.Email.HTML != "<p>A &amp; B</p>" {
		t.Fatal(stored, err)
	}
	for range 10 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if _, e := svc.Dispatch(ctx, id); e != nil {
				t.Error(e)
			}
		}()
	}
	wg.Wait()
	if driver.calls.Load() != 1 {
		t.Fatalf("sent %d times", driver.calls.Load())
	}
	stored, err = svc.Get(ctx, "product", id)
	if err != nil || stored.State != model.Accepted || stored.Email.Address != "" {
		t.Fatal("completion/scrubbing failed", stored, err)
	}
	// A stale sending lease is never automatically resent.
	request.Key = "stale"
	stale, err := svc.Send(ctx, "product", request)
	if err != nil {
		t.Fatal(err)
	}
	old := time.Now().Add(-2 * time.Minute)
	if err = repo.DBImp.Patch(ctx, stale.ID, repository.Patch().Set(repository.Field("next_attempt_at"), old)); err != nil {
		t.Fatal(err)
	}
	claimed, err := repo.Claim(ctx, stale.ID.Hex(), old)
	if err != nil {
		t.Fatal(err)
	}
	if err = repo.Recover(ctx, time.Now()); err != nil {
		t.Fatal(err)
	}
	claimed.State = model.Accepted
	if err = repo.Finish(ctx, claimed); !errors.Is(err, merrors.ErrDataConflict) {
		t.Fatal("stale worker overwrote recovery", err)
	}
	stale, err = svc.Get(ctx, "product", stale.ID.Hex())
	if err != nil || stale.State != model.Unknown {
		t.Fatal(stale, err)
	}
	// Retry state persists, and a duplicate dispatch cannot bypass the due time.
	request.Key = "retry"
	retry, err := svc.Send(ctx, "product", request)
	if err != nil {
		t.Fatal(err)
	}
	retryDriver := &sender{state: model.RetryWait}
	retrySvc := service.New(repo, catalog, retryDriver, "resend", zap.NewNop())
	if _, err = retrySvc.Dispatch(ctx, retry.ID.Hex()); err != nil {
		t.Fatal(err)
	}
	if _, err = retrySvc.Dispatch(ctx, retry.ID.Hex()); err != nil {
		t.Fatal(err)
	}
	retry, err = svc.Get(ctx, "product", retry.ID.Hex())
	if err != nil || retry.State != model.RetryWait || retryDriver.calls.Load() != 1 {
		t.Fatal(retry, err)
	}
	// Expire both queued and retrying notifications without a connected broker.
	for _, state := range []model.State{model.Queued, model.RetryWait} {
		request.Key = "offline-expiry-" + string(state)
		expired, e := svc.Send(ctx, "product", request)
		if e != nil {
			t.Fatal(e)
		}
		patch := repository.Patch().Set(repository.Field("state"), state).Set(repository.Field("expires_at"), time.Now().Add(-time.Minute)).Set(repository.Field("next_attempt_at"), time.Now().Add(time.Hour))
		if e = repo.DBImp.Patch(ctx, expired.ID, patch); e != nil {
			t.Fatal(e)
		}
		if e = repo.Recover(ctx, time.Now()); e != nil {
			t.Fatal(e)
		}
		expired, e = svc.Get(ctx, "product", expired.ID.Hex())
		if e != nil || expired.State != model.Expired || expired.Email.Address != "" {
			t.Fatal("offline expiry failed", expired, e)
		}
	}
	// Verify the real JetStream publisher/consumer path with an isolated broker.
	ncContainer, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ContainerRequest: testcontainers.ContainerRequest{Image: "nats:latest", ExposedPorts: []string{"4222/tcp"}, Cmd: []string{"-js"}, WaitingFor: wait.ForListeningPort("4222/tcp")}, Started: true})
	if err != nil {
		t.Fatal(err)
	}
	defer ncContainer.Terminate(context.Background())
	endpoint, err := ncContainer.Endpoint(ctx, "")
	if err != nil {
		t.Fatal(err)
	}
	t.Setenv("TEST_NOTIFICATIONS_NATS", "nats://"+endpoint)
	rt, err := msgruntime.New(zap.NewNop(), &messaging.Config{Driver: messaging.BBNats, Settings: map[string]any{"url_env": "TEST_NOTIFICATIONS_NATS", "broker_name": "test", "max_reconnects": -1, "reconnect_wait": 1}})
	if err != nil {
		t.Fatal(err)
	}
	defer rt.Close()
	broker, err := bus.New(ctx, rt.Durable, rt.DurableAdmin, repo, svc, zap.NewNop(), bus.DefaultConfig())
	if err != nil {
		t.Fatal(err)
	}
	defer broker.Close()
	// Production ops mode validates existing resources without silently rewriting drift.
	queue := bus.DefaultConfig()
	settings := reliable.DurableSettings{Stream: queue.Stream, Provisioning: reliable.DurableProvisioningSettings{Mode: reliable.DurableProvisioningOps, Environment: reliable.DurableEnvironmentProduction}}
	if err = rt.DurableAdmin.EnsureProducerResources(ctx, settings); err != nil {
		t.Fatal(err)
	}
	settings.Stream.MaxAgeHours = 72
	if err = rt.DurableAdmin.EnsureProducerResources(ctx, settings); !errors.Is(err, merrors.ErrDataConflict) {
		t.Fatal("retention drift accepted", err)
	}
	settings.Stream.MaxAgeHours = 48
	if err = rt.DurableAdmin.EnsureProducerResources(ctx, settings); err != nil {
		t.Fatal("ops mode mutated the stream", err)
	}
	canceled, stopAdmin := context.WithCancel(ctx)
	stopAdmin()
	if err = rt.DurableAdmin.EnsureProducerResources(canceled, settings); err == nil {
		t.Fatal("canceled provisioning succeeded")
	}

	client := grpcClient(t, svc)
	rpcCtx := callerContext(ctx, "product")
	input := &pb.SendRequest{IdempotencyKey: "jetstream", UserId: "u", TemplateId: "test", TemplateVersion: "1", Parameters: map[string]string{"name": "End to end & escaped"}, Targets: []*pb.Target{{Destination: &pb.Target_Email{Email: &pb.EmailTarget{Address: "user@example.com"}}}}}
	if _, e := client.Send(ctx, input); status.Code(e) != codes.Unauthenticated {
		t.Fatal("unauthenticated request accepted", e)
	}
	sent, err := client.Send(rpcCtx, input)
	if err != nil {
		t.Fatal(err)
	}
	replay, err := client.Send(rpcCtx, input)
	if err != nil || replay.Notification.Id != sent.Notification.Id {
		t.Fatal("RPC idempotency failed", err)
	}
	if _, e := client.Get(callerContext(ctx, "other"), &pb.GetRequest{Id: sent.Notification.Id}); status.Code(e) != codes.NotFound {
		t.Fatal("RPC caller isolation failed", e)
	}
	if err = broker.PublishDue(ctx); err != nil {
		t.Fatal(err)
	}
	workerCtx, stop := context.WithCancel(ctx)
	done := make(chan struct{})
	go func() { broker.RunWorker(workerCtx); close(done) }()
	defer func() { stop(); <-done }()
	deadline := time.Now().Add(10 * time.Second)
	for {
		result, e := client.Get(rpcCtx, &pb.GetRequest{Id: sent.Notification.Id})
		if e != nil {
			t.Fatal(e)
		}
		if strings.Contains(result.String(), "user@example.com") || strings.Contains(result.String(), "escaped") {
			t.Fatal("response exposes email content")
		}
		if result.Notification.Deliveries[0].State == pb.DeliveryState_DELIVERY_STATE_ACCEPTED {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("JetStream job did not complete", result.Notification.Deliveries[0].State)
		}
		time.Sleep(20 * time.Millisecond)
	}
	driver.mu.Lock()
	delivered := driver.messages[sent.Notification.Id]
	driver.mu.Unlock()
	if delivered.HTML != "<p>End to end &amp; escaped</p>" || delivered.Subject != "Hello" || delivered.Address != "user@example.com" {
		t.Fatal("parameters did not reach driver", delivered)
	}
}