package distworker

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"rocketgit.ru/rsmon/worker/app/models"
	"rocketgit.ru/rsmon/worker/internal/wire"
)

// runnerWithCreds is the smallest fixture that yields a Runner with a
// credentials block already applied (no DB, no init websocket).
func runnerWithCreds(creds *wire.NotificationCredentials) *Runner {
	r := NewRunner(&Config{MaxConcurrency: 4})
	r.credentialsMu.Lock()
	r.credentials = creds
	r.credentialsMu.Unlock()
	return r
}

// TestExecuteNotification_Email_NoCredentials verifies that the executor
// returns a permanent failure when no SMTP credential is pushed.
func TestExecuteNotification_Email_NoCredentials(t *testing.T) {
	r := runnerWithCreds(nil)
	report := r.ExecuteNotification(context.Background(), models.Task{
		JobID: "job-1",
		Payload: []byte(`{
			"job_id":"job-1","message_id":1,"notification_id":1,
			"method":"email","subject":"x","body_text":"x","body_html":"<p>x</p>",
			"contact":{"id":1,"kind":"email","value":"ops@example.com","name":"ops"},
			"message_kind":"down"
		}`),
	})
	assert.Equal(t, wire.NotificationResultPermanent, report.Status)
	assert.NotNil(t, report.Error)
	assert.Equal(t, wire.NotificationResultPermanent, report.Status, "must be permanent so the producer does not loop")
}

func TestClassifyResult_NilEmailErrorDoesNotPanic(t *testing.T) {
	status, _, text, retry, err := classifyResult(nil, "")
	assert.Equal(t, wire.NotificationResultDelivered, status)
	assert.Empty(t, text)
	assert.Nil(t, retry)
	assert.NoError(t, err)
}

// TestExecuteNotification_UnsupportedMethod covers sms/voice returning
// permanent + unsupported_method.
func TestExecuteNotification_UnsupportedMethod(t *testing.T) {
	r := runnerWithCreds(&wire.NotificationCredentials{
		SMTP:     []wire.SMTPCredential{{Server: "smtp.example.com", Port: 587, Login: "u", Password: "p"}},
		Telegram: []wire.TelegramCredential{{Name: "bot", Token: "123:abc"}},
	})
	for _, method := range []string{"sms", "voice"} {
		t.Run(method, func(t *testing.T) {
			report := r.ExecuteNotification(context.Background(), models.Task{
				JobID: "job-" + method,
				Payload: []byte(`{
					"job_id":"job-` + method + `","message_id":1,"notification_id":1,
					"method":"` + method + `","subject":"x","body_text":"x","body_html":"<p>x</p>",
					"contact":{"id":1,"kind":"` + method + `","value":"x","name":"ops"},
					"message_kind":"down"
				}`),
			})
			assert.Equal(t, wire.NotificationResultPermanent, report.Status)
			require.NotNil(t, report.Error)
			assert.Contains(t, *report.Error, "unsupported_method")
		})
	}
}

// TestExecuteNotification_UnknownMethod classifies unknown methods as
// permanent so we never loop on bad producer output.
func TestExecuteNotification_UnknownMethod(t *testing.T) {
	r := runnerWithCreds(&wire.NotificationCredentials{})
	report := r.ExecuteNotification(context.Background(), models.Task{
		JobID: "job-x",
		Payload: []byte(`{
			"job_id":"job-x","message_id":1,"notification_id":1,
			"method":"pigeon","subject":"x","body_text":"x","body_html":"<p>x</p>",
			"contact":{"id":1,"kind":"pigeon","value":"ops@example.com","name":"ops"},
			"message_kind":"down"
		}`),
	})
	assert.Equal(t, wire.NotificationResultPermanent, report.Status)
}

// TestExecuteNotification_EmptyPayloadPermanent: a malformed task must be
// rejected permanently so the operator can spot it on the admin page.
func TestExecuteNotification_EmptyPayloadPermanent(t *testing.T) {
	r := runnerWithCreds(&wire.NotificationCredentials{
		SMTP: []wire.SMTPCredential{{Server: "smtp.example.com", Port: 587, Login: "u", Password: "p"}},
	})
	report := r.ExecuteNotification(context.Background(), models.Task{JobID: "job-empty"})
	assert.Equal(t, wire.NotificationResultPermanent, report.Status)
	require.NotNil(t, report.Error)
}

func TestExecuteNotification_ExpiredDeadlineDoesNotDeliver(t *testing.T) {
	r := runnerWithCreds(&wire.NotificationCredentials{})
	expired := time.Now().Add(-time.Second)
	report := r.ExecuteNotification(context.Background(), models.Task{JobID: "expired", Deadline: &expired, Payload: []byte(`{"job_id":"expired","message_id":1,"method":"email","contact":{"id":1,"kind":"email"}}`)})
	require.NotNil(t, report.Error)
	assert.Contains(t, *report.Error, "deadline expired")
	assert.Equal(t, wire.NotificationResultPermanent, report.Status)
}

// TestExecuteNotification_DurationPositive: every report carries a non-zero
// duration_ms even when the work happens instantly. This is what the
// notification_deliveries audit row expects.
func TestExecuteNotification_DurationPositive(t *testing.T) {
	r := runnerWithCreds(&wire.NotificationCredentials{})
	report := r.ExecuteNotification(context.Background(), models.Task{
		JobID: "job-d",
		Payload: []byte(`{
			"job_id":"job-d","message_id":1,"notification_id":1,
			"method":"sms","subject":"x","body_text":"x","body_html":"<p>x</p>",
			"contact":{"id":1,"kind":"sms","value":"x","name":"ops"},
			"message_kind":"down"
		}`),
	})
	assert.GreaterOrEqual(t, report.DurationMs, 0)
}

// TestClassifyResult verifies the error -> status mapping.
func TestClassifyResult(t *testing.T) {
	tests := []struct {
		name           string
		err            error
		wantStatus     string
		wantRetryAfter *int
	}{
		{"nil error", nil, wire.NotificationResultDelivered, nil},
		{"permanent provider error", errors.New("550 mailbox not found"), wire.NotificationResultPermanent, nil},
		{"smtp 421 retryable", errors.New("smtp: 421 try again later"), wire.NotificationResultRetryable, intPtr(30)},
		{"smtp 452 retryable", errors.New("452 insufficient storage"), wire.NotificationResultRetryable, intPtr(30)},
		{"network timeout retryable", errors.New("dial tcp: i/o timeout"), wire.NotificationResultRetryable, intPtr(15)},
		{"tls handshake retryable", errors.New("tls: handshake failure"), wire.NotificationResultRetryable, intPtr(15)},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			status, _, errStr, retry, _ := classifyResult(tc.err, "")
			assert.Equal(t, tc.wantStatus, status)
			if tc.err != nil {
				assert.NotEmpty(t, errStr)
			}
			if tc.wantRetryAfter == nil {
				assert.Nil(t, retry)
			} else {
				require.NotNil(t, retry)
				assert.Equal(t, *tc.wantRetryAfter, *retry)
			}
		})
	}
}

// TestWireTelegramToModel_RoundTrip sanity-checks the wire->DB credential
// translation used by the telegram branch.
func TestWireTelegramToModel_RoundTrip(t *testing.T) {
	wire := &wire.TelegramCredential{
		ID:      1,
		Name:    "main-bot",
		BotName: "rsmon_bot",
		Token:   "123456:ABCDEFG",
		APIURL:  "https://api.telegram.org",
	}
	nc := wireTelegramToModel(wire)
	require.NotNil(t, nc)
	assert.Equal(t, models.CredentialKindTelegram, nc.Kind)
	assert.Equal(t, "main-bot", nc.Name)
	require.NotNil(t, nc.BotName)
	assert.Equal(t, "rsmon_bot", *nc.BotName)
	require.NotNil(t, nc.APIURL)
	assert.Equal(t, "https://api.telegram.org", *nc.APIURL)
	got, err := nc.GetSecret()
	require.NoError(t, err)
	assert.Equal(t, "123456:ABCDEFG", got)
}

func TestSelectNotificationCredentialByID(t *testing.T) {
	id := int64(2)
	smtp, ok := selectSMTPCredential([]wire.SMTPCredential{{ID: 1, Name: "first"}, {ID: 2, Name: "second"}}, &id)
	require.True(t, ok)
	assert.Equal(t, "second", smtp.Name)

	tg, ok := selectTelegramCredential([]wire.TelegramCredential{{ID: 1, Name: "first"}, {ID: 2, Name: "second"}}, &id)
	require.True(t, ok)
	assert.Equal(t, "second", tg.Name)

	missingID := int64(3)
	_, ok = selectSMTPCredential([]wire.SMTPCredential{{ID: 1, Name: "first"}}, &missingID)
	assert.False(t, ok)
}

// TestEnqueueNotification_BoundedQueue ensures the notifyQueue provides
// backpressure: the channel capacity is queueCapacity() and EnqueueNotification
// blocks once it is full.
func TestEnqueueNotification_BoundedQueue(t *testing.T) {
	r := NewRunner(&Config{MaxConcurrency: 4})
	r.notifyQueue = make(chan wire.NotificationTask, 4)

	for i := 0; i < 4; i++ {
		require.True(t, r.EnqueueNotification(wire.NotificationTask{JobID: "x"}))
	}
	// Channel is full; a non-blocking send must fail. We cannot truly verify
	// the blocking case in unit tests, so we just assert the depth counter.
	assert.Equal(t, int64(4), r.notifyDepth)
}

// TestExecuteNotification_ReportsJobIDFromPayload verifies the executor
// falls back to the task's JobID when the payload has none.
func TestExecuteNotification_ReportsJobIDFromPayload(t *testing.T) {
	r := runnerWithCreds(nil)
	report := r.ExecuteNotification(context.Background(), models.Task{
		JobID: "outer-job",
		Payload: []byte(`{
			"job_id":"","message_id":1,"notification_id":1,
			"method":"sms","subject":"x","body_text":"x","body_html":"<p>x</p>",
			"contact":{"id":1,"kind":"sms","value":"x","name":"ops"},
			"message_kind":"down"
		}`),
	})
	assert.Equal(t, "outer-job", report.JobID)
}

func TestDelegatedNotificationAuditCoexistsWithSelfcheckWithoutSecrets(t *testing.T) {
	const (
		recipient        = "ops-private@example.com"
		renderedBody     = "private rendered body"
		credentialSecret = "smtp-password-secret"
		hookURL          = "https://hooks.example.com/private"
		authorization    = "Bearer private-authorization"
		providerResponse = "private provider response"
	)
	r := runnerWithCreds(&wire.NotificationCredentials{
		SMTP: []wire.SMTPCredential{{Password: credentialSecret}},
	})
	r.notifyResults = make(chan notifyResultEnvelope, 1)
	r.RecordNotification(&NotificationRow{
		Kind: "email", Channel: notificationChannelSMTP, Subject: "selfcheck", Body: "main API down", OK: true,
	})
	r.executeAndForwardNotification(wire.NotificationTask{
		JobID: "delegated-job", Method: "sms", Subject: authorization,
		BodyText: renderedBody, BodyHTML: providerResponse,
		Contact: wire.NotificationContact{Kind: "sms", Value: recipient, Name: hookURL},
	})

	rows := r.RecentNotifications(10)
	require.Len(t, rows, 2)
	assert.Equal(t, "selfcheck", rows[0].Subject)
	delegated := rows[1]
	assert.Equal(t, "delegated-job", delegated.JobID)
	assert.Equal(t, "sms", delegated.Method)
	assert.Equal(t, wire.NotificationResultPermanent, delegated.Status)
	assert.Empty(t, delegated.Subject)
	assert.Empty(t, delegated.Body)
	assert.Empty(t, delegated.Error)
	assert.False(t, delegated.At.IsZero())

	stored, err := json.Marshal(delegated)
	require.NoError(t, err)
	for _, secret := range []string{recipient, renderedBody, credentialSecret, hookURL, authorization, providerResponse} {
		assert.NotContains(t, string(stored), secret)
	}
}

func TestDelegatedNotificationAuditRecordsOneTerminalRow(t *testing.T) {
	const secret = "private-notification-secret"
	validTask := func(jobID string) wire.NotificationTask {
		return wire.NotificationTask{
			JobID: jobID, LeaseToken: "lease", Method: "sms", Subject: secret, BodyText: secret,
			Contact: wire.NotificationContact{Value: secret, Name: secret},
		}
	}
	tests := []struct {
		name   string
		invoke func(*Runner)
		method string
	}{
		{
			name:   "normal result",
			invoke: func(r *Runner) { r.executeAndForwardNotification(validTask("normal")) },
			method: "sms",
		},
		{
			name: "invalid deadline",
			invoke: func(r *Runner) {
				task := validTask("invalid-deadline")
				deadline := "not-a-timestamp"
				task.Deadline = &deadline
				r.executeAndForwardNotification(task)
			},
			method: "sms",
		},
		{
			name: "expired deadline",
			invoke: func(r *Runner) {
				task := validTask("expired-deadline")
				deadline := time.Now().Add(-time.Second).Format(time.RFC3339Nano)
				task.Deadline = &deadline
				r.executeAndForwardNotification(task)
			},
			method: "sms",
		},
		{
			name: "malformed payload",
			invoke: func(r *Runner) {
				r.ExecuteNotification(context.Background(), models.Task{JobID: "malformed-payload", Payload: []byte(`{"subject":"` + secret)})
			},
		},
		{
			name: "executor panic",
			invoke: func(r *Runner) {
				r.notificationExecutor = func(context.Context, models.Task) wire.NotificationResultReport {
					panic(secret)
				}
				r.ExecuteNotification(context.Background(), models.Task{JobID: "panic", Payload: []byte(`{}`)})
			},
		},
		{
			name: "malformed envelope",
			invoke: func(r *Runner) {
				r.enqueueTaskMessage(wire.WorkerMessage{TaskEnvelope: &wire.TaskEnvelope{
					Type: "invalid", JobID: "bad-envelope", Notify: ptrNotificationTask(validTask("bad-envelope")),
				}})
			},
			method: "sms",
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			r := runnerWithCreds(&wire.NotificationCredentials{})
			r.notifyResults = make(chan notifyResultEnvelope, 1)
			tc.invoke(r)

			rows := r.RecentNotifications(10)
			require.Len(t, rows, 1)
			assert.Equal(t, wire.NotificationResultPermanent, rows[0].Status)
			assert.Equal(t, tc.method, rows[0].Method)
			stored, err := json.Marshal(rows[0])
			require.NoError(t, err)
			assert.NotContains(t, string(stored), secret)
		})
	}
}

func TestDelegatedNotificationAuditConcurrentRowsAreSecretFree(t *testing.T) {
	const attempts = 32
	const secret = "concurrent-private-secret"
	r := runnerWithCreds(&wire.NotificationCredentials{})
	var wg sync.WaitGroup
	for i := 0; i < attempts; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			r.ExecuteNotification(context.Background(), models.Task{
				JobID:   fmt.Sprintf("job-%d", i),
				Payload: []byte(`{"job_id":"job-` + fmt.Sprint(i) + `","method":"sms","body_text":"` + secret + `","contact":{"value":"` + secret + `"}}`),
			})
		}(i)
	}
	wg.Wait()

	rows := r.RecentNotifications(attempts)
	require.Len(t, rows, attempts)
	for _, row := range rows {
		stored, err := json.Marshal(row)
		require.NoError(t, err)
		assert.NotContains(t, string(stored), secret)
		assert.Equal(t, wire.NotificationResultPermanent, row.Status)
	}
}

func TestInvalidNotificationDeadlineIsNotExecuted(t *testing.T) {
	r := runnerWithCreds(&wire.NotificationCredentials{})
	r.notifyResults = make(chan notifyResultEnvelope, 1)
	called := false
	r.notificationExecutor = func(context.Context, models.Task) wire.NotificationResultReport {
		called = true
		return wire.NotificationResultReport{Status: wire.NotificationResultDelivered}
	}
	deadline := "not-a-timestamp"
	r.executeAndForwardNotification(wire.NotificationTask{
		JobID: "invalid-deadline", LeaseToken: "lease", Method: "sms", Deadline: &deadline,
	})

	assert.False(t, called)
	env := <-r.notifyResults
	assert.Equal(t, wire.NotificationResultPermanent, env.report.Status)
	require.NotNil(t, env.report.Error)
	assert.Equal(t, "invalid notification deadline", *env.report.Error)
	require.Len(t, r.RecentNotifications(10), 1)
}

func ptrNotificationTask(task wire.NotificationTask) *wire.NotificationTask { return &task }

// guard against time import being unused if the above compile-time helpers
// are dropped in a future refactor.
var _ = time.Second