package meshxassets

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/api/daemon"
)

const (
	defaultInterval      = 2 * time.Second
	defaultRequestTTL    = 30 * time.Second
	defaultLeaseDuration = 30 * time.Second
	defaultLimit         = 50
	defaultTokenTTL      = time.Minute
	defaultLeaseOwner    = "agentteam-asset-access-worker"
)

// Notifier is the daemon control-plane surface used by the worker.
type Notifier interface {
	NotifyAssetsAccessChangedByAgent(ctx context.Context, workspaceID string, agentID string, event daemon.AssetsAccessChangedEvent) error
}

// WorkspaceIDSource returns the workspace IDs this process owns. A nil source
// or an empty result means "lease globally", used by all-in-one deployments.
type WorkspaceIDSource func(context.Context) ([]string, error)

// Options configures a MeshX agent-asset access outbox worker.
type Options struct {
	BaseURL       string
	HTTPClient    *http.Client
	Notifier      Notifier
	WorkspaceIDs  WorkspaceIDSource
	Logger        *slog.Logger
	LeaseOwner    string
	Interval      time.Duration
	RequestTTL    time.Duration
	LeaseDuration time.Duration
	TokenTTL      time.Duration
	Limit         int
	Now           func() time.Time
}

// Worker leases MeshX asset-access outbox rows and pushes them to Desktop
// daemons through AgentTeam's existing daemon WebSocket hub.
type Worker struct {
	client       *client
	notifier     Notifier
	workspaceIDs WorkspaceIDSource
	logger       *slog.Logger
	leaseOwner   string
	interval     time.Duration
	lease        time.Duration
	limit        int
	now          func() time.Time
}

// RunResult summarizes one worker pass.
type RunResult struct {
	Leased    int
	Delivered int
	Retried   int
	Failed    int
}

// NewWorker validates dependencies and returns a configured worker.
func NewWorker(options Options) (*Worker, error) {
	baseURL, err := normalizeBaseURL(options.BaseURL)
	if err != nil {
		return nil, err
	}
	if options.Notifier == nil {
		return nil, errors.New("meshx asset access worker notifier is required")
	}
	logger := options.Logger
	if logger == nil {
		logger = slog.Default()
	}
	now := options.Now
	if now == nil {
		now = time.Now
	}
	requestTTL := options.RequestTTL
	if requestTTL <= 0 {
		requestTTL = defaultRequestTTL
	}
	tokenTTL := options.TokenTTL
	if tokenTTL <= 0 {
		tokenTTL = defaultTokenTTL
	}
	httpClient := options.HTTPClient
	if httpClient == nil {
		httpClient = &http.Client{Timeout: requestTTL}
	}
	leaseOwner := strings.TrimSpace(options.LeaseOwner)
	if leaseOwner == "" {
		leaseOwner = defaultLeaseOwner
	}
	interval := options.Interval
	if interval <= 0 {
		interval = defaultInterval
	}
	leaseDuration := options.LeaseDuration
	if leaseDuration <= 0 {
		leaseDuration = defaultLeaseDuration
	}
	limit := options.Limit
	if limit <= 0 {
		limit = defaultLimit
	}
	return &Worker{
		client: &client{
			baseURL:      baseURL,
			httpClient:   httpClient,
			requestTTL:   requestTTL,
			tokenTTL:     tokenTTL,
			tokenSubject: leaseOwner,
			now:          now,
		},
		notifier:     options.Notifier,
		workspaceIDs: options.WorkspaceIDs,
		logger:       logger,
		leaseOwner:   leaseOwner,
		interval:     interval,
		lease:        leaseDuration,
		limit:        limit,
		now:          now,
	}, nil
}

// Run starts the polling agentteam. It performs one immediate pass, then repeats on
// Interval until ctx is cancelled.
func (w *Worker) Run(ctx context.Context) {
	if w == nil {
		return
	}
	w.runAndLog(ctx)
	ticker := time.NewTicker(w.interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			w.runAndLog(ctx)
		}
	}
}

// RunOnce executes one lease/deliver/ack pass.
func (w *Worker) RunOnce(ctx context.Context) (RunResult, error) {
	if w == nil {
		return RunResult{}, errors.New("meshx asset access worker is nil")
	}
	workspaceIDs, err := w.workspaceFilter(ctx)
	if err != nil {
		return RunResult{}, err
	}
	events, err := w.client.lease(ctx, leaseRequest{
		LeaseOwner:   w.leaseOwner,
		LeaseSeconds: int64(w.lease / time.Second),
		Limit:        w.limit,
		WorkspaceIDs: workspaceIDs,
	})
	if err != nil {
		return RunResult{}, err
	}
	result := RunResult{Leased: len(events)}
	for _, event := range events {
		if err := w.deliver(ctx, event); err != nil {
			updated, failed, markErr := w.client.markFailed(ctx, event.EventID, failedRequest{
				LeaseOwner:    w.leaseOwner,
				LastError:     truncateError(err.Error()),
				NextAttemptAt: w.nextAttemptAt(event).Format(time.RFC3339Nano),
			})
			if markErr != nil {
				return result, markErr
			}
			if !updated {
				return result, fmt.Errorf("meshx asset access event %s failure ack did not update", event.EventID)
			}
			if failed {
				result.Failed++
			} else {
				result.Retried++
			}
			continue
		}
		updated, err := w.client.markDelivered(ctx, event.EventID, deliveredRequest{LeaseOwner: w.leaseOwner})
		if err != nil {
			return result, err
		}
		if !updated {
			return result, fmt.Errorf("meshx asset access event %s delivered ack did not update", event.EventID)
		}
		result.Delivered++
	}
	return result, nil
}

func (w *Worker) runAndLog(ctx context.Context) {
	result, err := w.RunOnce(ctx)
	if err != nil {
		if !errors.Is(err, context.Canceled) {
			w.logger.Warn("meshx asset access outbox pass failed", slog.Any("error", err))
		}
		return
	}
	if result.Leased > 0 {
		w.logger.Info("meshx asset access outbox pass completed",
			slog.Int("leased", result.Leased),
			slog.Int("delivered", result.Delivered),
			slog.Int("retried", result.Retried),
			slog.Int("failed", result.Failed))
	}
}

func (w *Worker) workspaceFilter(ctx context.Context) ([]string, error) {
	if w.workspaceIDs == nil {
		return nil, nil
	}
	ids, err := w.workspaceIDs(ctx)
	if err != nil {
		return nil, fmt.Errorf("meshx asset access workspace filter: %w", err)
	}
	return normalizeStrings(ids), nil
}

func (w *Worker) deliver(ctx context.Context, event accessEvent) error {
	agentIDs := normalizeStrings(event.AgentIDs)
	if len(agentIDs) == 0 {
		return errors.New("meshx asset access event has no agent ids")
	}
	workspaceID := strings.TrimSpace(event.WorkspaceID)
	if workspaceID == "" {
		return errors.New("meshx asset access event has no workspace id")
	}
	payload := daemon.AssetsAccessChangedEvent{
		EventID:    strings.TrimSpace(event.EventID),
		AgentIDs:   agentIDs,
		AssetTypes: normalizeStrings(event.AssetTypes),
		Version:    event.Version,
		OccurredAt: event.occurredAt(),
		Source:     strings.TrimSpace(event.Source),
	}
	var errs []error
	for _, agentID := range agentIDs {
		if err := w.notifier.NotifyAssetsAccessChangedByAgent(ctx, workspaceID, agentID, payload); err != nil {
			errs = append(errs, fmt.Errorf("agent %s: %w", agentID, err))
		}
	}
	return errors.Join(errs...)
}

func (w *Worker) nextAttemptAt(event accessEvent) time.Time {
	attempts := event.Attempts
	if attempts < 0 {
		attempts = 0
	}
	delay := 5 * time.Second
	for i := 0; i < attempts && delay < 5*time.Minute; i++ {
		delay *= 2
	}
	if delay > 5*time.Minute {
		delay = 5 * time.Minute
	}
	return w.now().UTC().Add(delay)
}

type client struct {
	baseURL      *url.URL
	httpClient   *http.Client
	requestTTL   time.Duration
	tokenTTL     time.Duration
	tokenSubject string
	now          func() time.Time
}

type leaseRequest struct {
	LeaseOwner   string   `json:"lease_owner"`
	LeaseSeconds int64    `json:"lease_seconds,omitempty"`
	Limit        int      `json:"limit,omitempty"`
	WorkspaceIDs []string `json:"workspace_ids,omitempty"`
}

type leaseResponse struct {
	Events []accessEvent `json:"events"`
}

type accessEvent struct {
	EventID     string   `json:"event_id"`
	WorkspaceID string   `json:"workspace_id"`
	AgentIDs    []string `json:"agent_ids"`
	AssetTypes  []string `json:"asset_types"`
	Version     int64    `json:"version"`
	OccurredAt  string   `json:"occurred_at"`
	Source      string   `json:"source"`
	Attempts    int      `json:"attempts,omitempty"`
	MaxAttempts int      `json:"max_attempts,omitempty"`
}

type deliveredRequest struct {
	LeaseOwner string `json:"lease_owner"`
}

type failedRequest struct {
	LeaseOwner    string `json:"lease_owner"`
	LastError     string `json:"last_error,omitempty"`
	NextAttemptAt string `json:"next_attempt_at,omitempty"`
}

type ackResponse struct {
	Updated bool `json:"updated"`
	Failed  bool `json:"failed,omitempty"`
}

func (event accessEvent) occurredAt() time.Time {
	raw := strings.TrimSpace(event.OccurredAt)
	if raw == "" {
		return time.Time{}
	}
	parsed, err := time.Parse(time.RFC3339Nano, raw)
	if err != nil {
		return time.Time{}
	}
	return parsed
}

func (c *client) lease(ctx context.Context, request leaseRequest) ([]accessEvent, error) {
	var response leaseResponse
	if err := c.post(ctx, "/api/v1/internal/agent-assets/access-events:lease", request, &response); err != nil {
		return nil, err
	}
	return response.Events, nil
}

func (c *client) markDelivered(ctx context.Context, eventID string, request deliveredRequest) (bool, error) {
	var response ackResponse
	path := "/api/v1/internal/agent-assets/access-events/" + url.PathEscape(strings.TrimSpace(eventID)) + "/delivered"
	if err := c.post(ctx, path, request, &response); err != nil {
		return false, err
	}
	return response.Updated, nil
}

func (c *client) markFailed(ctx context.Context, eventID string, request failedRequest) (bool, bool, error) {
	var response ackResponse
	path := "/api/v1/internal/agent-assets/access-events/" + url.PathEscape(strings.TrimSpace(eventID)) + "/failed"
	if err := c.post(ctx, path, request, &response); err != nil {
		return false, false, err
	}
	return response.Updated, response.Failed, nil
}

func (c *client) post(ctx context.Context, path string, body any, out any) error {
	ctx, cancel := context.WithTimeout(ctx, c.requestTTL)
	defer cancel()
	payload, err := json.Marshal(body)
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.endpoint(path), bytes.NewReader(payload))
	if err != nil {
		return err
	}
	token := c.serviceToken()
	req.Header.Set("Authorization", "Bearer "+token)
	req.Header.Set("X-AgentTeam-Verified-JWT", "Bearer "+token)
	req.Header.Set("Content-Type", "application/json")
	resp, err := c.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		msg, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))
		return fmt.Errorf("meshx asset access %s: status %d: %s", path, resp.StatusCode, strings.TrimSpace(string(msg)))
	}
	if out == nil {
		_, _ = io.Copy(io.Discard, resp.Body)
		return nil
	}
	if err := json.NewDecoder(resp.Body).Decode(out); err != nil {
		return fmt.Errorf("meshx asset access %s: decode response: %w", path, err)
	}
	return nil
}

func (c *client) endpoint(path string) string {
	next := *c.baseURL
	next.Path = strings.TrimRight(next.Path, "/") + "/" + strings.TrimLeft(path, "/")
	return next.String()
}

func (c *client) serviceToken() string {
	now := c.now().UTC()
	subject := strings.TrimSpace(c.tokenSubject)
	if subject == "" {
		subject = defaultLeaseOwner
	}
	claims := map[string]any{
		"iss":                    "agentteam-server",
		"aud":                    "meshx",
		"sub":                    subject,
		"service_actor":          true,
		"automation_permissions": []string{"automation:service"},
		"workspace_ids":          []string{"*"},
		"iat":                    now.Unix(),
		"exp":                    now.Add(c.tokenTTL).Unix(),
	}
	header, _ := json.Marshal(map[string]string{"alg": "none", "typ": "JWT"})
	payload, _ := json.Marshal(claims)
	return base64.RawURLEncoding.EncodeToString(header) + "." + base64.RawURLEncoding.EncodeToString(payload) + "."
}

func normalizeBaseURL(raw string) (*url.URL, error) {
	trimmed := strings.TrimRight(strings.TrimSpace(raw), "/")
	if trimmed == "" {
		return nil, errors.New("meshx asset access base URL is required")
	}
	parsed, err := url.Parse(trimmed)
	if err != nil {
		return nil, fmt.Errorf("parse meshx asset access base URL: %w", err)
	}
	if parsed.Scheme != "http" && parsed.Scheme != "https" {
		return nil, errors.New("meshx asset access base URL must use http(s)")
	}
	if parsed.Host == "" {
		return nil, errors.New("meshx asset access base URL must include host")
	}
	return parsed, nil
}

func normalizeStrings(values []string) []string {
	if len(values) == 0 {
		return nil
	}
	seen := make(map[string]struct{}, len(values))
	out := make([]string, 0, len(values))
	for _, value := range values {
		trimmed := strings.TrimSpace(value)
		if trimmed == "" {
			continue
		}
		if _, ok := seen[trimmed]; ok {
			continue
		}
		seen[trimmed] = struct{}{}
		out = append(out, trimmed)
	}
	return out
}

func truncateError(value string) string {
	value = strings.TrimSpace(value)
	if len(value) <= 512 {
		return value
	}
	return value[:509] + "..."
}