package bot

import (
	"bufio"
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"strings"
	"time"
)

// A minimal Anthropic Messages API client — just the slice the bot brain
// needs (system blocks with cache markers, adaptive thinking, tools), on
// net/http so the static binary takes no new dependency. Base URL and key
// come from the same env the SDKs read (ANTHROPIC_API_KEY,
// ANTHROPIC_BASE_URL), so tests point it at an httptest server.

const anthropicVersion = "2023-06-01"

type anthropicClient struct {
	key  string
	base string
	http *http.Client
	// record, when set, receives every successful call's usage block — the
	// usage meter's tap (#857). It sits here, not at the call sites, so every
	// model path (reply, classifier, task, research, app build) is counted
	// without each one remembering to.
	record func(model string, u anthUsage)
}

func newAnthropicClient(key string, record func(model string, u anthUsage)) *anthropicClient {
	base := os.Getenv("ANTHROPIC_BASE_URL")
	if base == "" {
		base = "https://api.anthropic.com"
	}
	// Long timeout: a thinking run over a large cached system prompt can
	// legitimately take a couple of minutes.
	return &anthropicClient{key: key, base: base, http: &http.Client{Timeout: 5 * time.Minute}, record: record}
}

type anthBlock struct {
	Type string `json:"type"`
	Text string `json:"text,omitempty"`
	// CacheControl marks the prefix up to this block cacheable (the big system
	// prompt rides every call — see systemBlocks in brain.go).
	CacheControl *anthCache `json:"cache_control,omitempty"`
	// tool_use response fields
	ID    string         `json:"id,omitempty"`
	Name  string         `json:"name,omitempty"`
	Input map[string]any `json:"input,omitempty"`
	// tool_result request fields (the task loop's round-trip, task.go)
	ToolUseID string `json:"tool_use_id,omitempty"`
	Content   any    `json:"content,omitempty"`
	// Extended-thinking blocks. A tool loop echoes the model's content back
	// verbatim as the assistant turn, and the API REQUIRES thinking blocks to
	// come back intact (signature and all) — dropping these fields turns
	// every live thinking+tools run into a hard 400 ("thinking.thinking:
	// Field required"), which scripted test fakes can never surface.
	Thinking  string `json:"thinking,omitempty"`
	Signature string `json:"signature,omitempty"`
	Data      string `json:"data,omitempty"` // redacted_thinking payload
}

// MarshalJSON closes two omitempty traps on the tool-loop echo — both the
// same shape: the API REQUIRES a field the model sometimes returns EMPTY,
// and Go's omitempty drops empty values entirely, turning the echo into a
// hard 400 on the loop's next call.
//
//   - tool_use.input: a tool called with NO arguments (a bare issues.list)
//     echoed with no input at all → "tool_use.input: Field required".
//   - thinking.thinking: adaptive thinking can emit a ZERO-LENGTH thinking
//     block on a continuation turn (seen live from the read_app_source
//     loop) → "thinking.thinking: Field required".
//
// Each marshals with the field explicitly present; every other block shape
// is untouched.
func (b anthBlock) MarshalJSON() ([]byte, error) {
	type alias anthBlock // method-less: plain marshalling, no recursion
	switch {
	case b.Type == "tool_use" && len(b.Input) == 0:
		return json.Marshal(struct {
			alias
			Input map[string]any `json:"input"` // shallower field wins: {} rides
		}{alias(b), map[string]any{}})
	case b.Type == "thinking" && b.Thinking == "":
		return json.Marshal(struct {
			alias
			Thinking string `json:"thinking"` // shallower field wins: "" rides
		}{alias(b), ""})
	}
	return json.Marshal(alias(b))
}

type anthCache struct {
	Type string `json:"type"` // "ephemeral"
}

// anthMessage's content is either a plain string (the conversational path) or
// a []anthBlock (the task loop's tool_use / tool_result turns).
type anthMessage struct {
	Role    string `json:"role"`
	Content any    `json:"content"`
}

type anthTool struct {
	Name        string         `json:"name"`
	Description string         `json:"description"`
	InputSchema map[string]any `json:"input_schema"`
}

type anthRequest struct {
	Model      string         `json:"model"`
	MaxTokens  int            `json:"max_tokens"`
	System     []anthBlock    `json:"system,omitempty"`
	Thinking   map[string]any `json:"thinking,omitempty"`
	Tools      []anthTool     `json:"tools,omitempty"`
	ToolChoice map[string]any `json:"tool_choice,omitempty"`
	Messages   []anthMessage  `json:"messages"`
	Stream     bool           `json:"stream,omitempty"`
}

type anthResponse struct {
	Content    []anthBlock `json:"content"`
	StopReason string      `json:"stop_reason"`
	Usage      anthUsage   `json:"usage"`
}

// anthUsage is the token accounting block every Messages response carries.
// All four counters are kept distinct — cache reads and writes price
// differently from fresh input, so the usage meter (#857) records each as its
// own dimension.
type anthUsage struct {
	InputTokens              int64 `json:"input_tokens"`
	OutputTokens             int64 `json:"output_tokens"`
	CacheCreationInputTokens int64 `json:"cache_creation_input_tokens"`
	CacheReadInputTokens     int64 `json:"cache_read_input_tokens"`
}

// anthHTTPError is a non-200 response — carried as an error so create's retry
// loop can tell a deterministic client error (4xx: bad request/auth) from a
// transient one (5xx/429) worth retrying.
type anthHTTPError struct {
	status int
	body   string
}

func (e *anthHTTPError) Error() string { return fmt.Sprintf("anthropic: %d: %s", e.status, e.body) }

// createAttempts is how many times create tries a request. The long build call
// (large max_tokens + thinking) is a non-streamed request that intermediaries
// sometimes cut mid-response ("unexpected EOF") — almost always transient, so a
// couple of retries makes the builder resilient without masking real failures.
const createAttempts = 3

func (c *anthropicClient) create(req anthRequest) (*anthResponse, error) {
	return c.do(req, false)
}

// createStream makes the request as a server-sent-event stream and assembles the
// same anthResponse from the deltas. Streaming keeps the connection continuously
// active with incremental events, so a long generation isn't cut by an
// idle-connection timeout the way a non-streamed request is — the fix for the
// app builder's recurring "unexpected EOF". Use it for the long build call.
func (c *anthropicClient) createStream(req anthRequest) (*anthResponse, error) {
	return c.do(req, true)
}

func (c *anthropicClient) do(req anthRequest, stream bool) (*anthResponse, error) {
	req.Stream = stream
	body, err := json.Marshal(req)
	if err != nil {
		return nil, err
	}
	var lastErr error
	for attempt := 1; attempt <= createAttempts; attempt++ {
		if attempt > 1 {
			time.Sleep(time.Duration(attempt-1) * time.Second) // 1s, 2s
		}
		var resp *anthResponse
		if stream {
			resp, err = c.doStream(body)
		} else {
			resp, err = c.doRequest(body)
		}
		if err == nil {
			if c.record != nil {
				c.record(req.Model, resp.Usage)
			}
			return resp, nil
		}
		lastErr = err
		// Deterministic client errors (4xx other than 429) won't change on a
		// retry — fail fast. Everything else (transport drops like unexpected
		// EOF, 5xx, 429, a cut stream) is retried.
		var he *anthHTTPError
		if errors.As(err, &he) && he.status < 500 && he.status != http.StatusTooManyRequests {
			return nil, err
		}
	}
	return nil, lastErr
}

func (c *anthropicClient) doRequest(body []byte) (*anthResponse, error) {
	hr, err := http.NewRequest("POST", c.base+"/v1/messages", bytes.NewReader(body))
	if err != nil {
		return nil, err
	}
	hr.Header.Set("Content-Type", "application/json")
	hr.Header.Set("x-api-key", c.key)
	hr.Header.Set("anthropic-version", anthropicVersion)
	resp, err := c.http.Do(hr)
	if err != nil {
		return nil, err // transport error (DNS, TLS, connection reset/EOF before headers)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		msg, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))
		return nil, &anthHTTPError{status: resp.StatusCode, body: string(msg)}
	}
	var out anthResponse
	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
		return nil, err // e.g. unexpected EOF: connection cut mid-response
	}
	return &out, nil
}

// doStream issues the request with stream:true and reassembles the response from
// the SSE deltas. It accumulates text and tool_use blocks (the only kinds the
// bot reads back — thinking is streamed too but ignored). A stream cut before
// message_stop returns an error, which the retry loop treats as transient.
func (c *anthropicClient) doStream(body []byte) (*anthResponse, error) {
	hr, err := http.NewRequest("POST", c.base+"/v1/messages", bytes.NewReader(body))
	if err != nil {
		return nil, err
	}
	hr.Header.Set("Content-Type", "application/json")
	hr.Header.Set("x-api-key", c.key)
	hr.Header.Set("anthropic-version", anthropicVersion)
	hr.Header.Set("Accept", "text/event-stream")
	resp, err := c.http.Do(hr)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		msg, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))
		return nil, &anthHTTPError{status: resp.StatusCode, body: string(msg)}
	}

	// One accumulator per content block, indexed as the API indexes them.
	type block struct {
		typ  string
		id   string
		name string
		text strings.Builder
		js   strings.Builder // partial tool_use input JSON
	}
	var blocks []*block
	out := &anthResponse{}

	sc := bufio.NewScanner(resp.Body)
	sc.Buffer(make([]byte, 0, 64*1024), 8*1024*1024) // deltas are small; be generous anyway
	done := false
	for sc.Scan() {
		line := sc.Text()
		if !strings.HasPrefix(line, "data:") {
			continue // skip "event:" lines and blank separators — the data JSON self-describes
		}
		data := strings.TrimSpace(line[len("data:"):])
		if data == "" {
			continue
		}
		var ev struct {
			Type         string `json:"type"`
			Index        int    `json:"index"`
			ContentBlock *struct {
				Type string `json:"type"`
				ID   string `json:"id"`
				Name string `json:"name"`
			} `json:"content_block"`
			Delta *struct {
				Type        string `json:"type"`
				Text        string `json:"text"`
				PartialJSON string `json:"partial_json"`
				StopReason  string `json:"stop_reason"`
			} `json:"delta"`
			// Usage rides message_start (input + cache counters, inside the
			// message envelope) and message_delta (cumulative output tokens,
			// as a sibling of delta) — both are folded into out.Usage.
			Message *struct {
				Usage anthUsage `json:"usage"`
			} `json:"message"`
			Usage *anthUsage `json:"usage"`
			Error *struct {
				Type    string `json:"type"`
				Message string `json:"message"`
			} `json:"error"`
		}
		if json.Unmarshal([]byte(data), &ev) != nil {
			continue
		}
		switch ev.Type {
		case "message_start":
			if ev.Message != nil {
				out.Usage = ev.Message.Usage
			}
		case "content_block_start":
			for len(blocks) <= ev.Index {
				blocks = append(blocks, nil)
			}
			b := &block{}
			if ev.ContentBlock != nil {
				b.typ = ev.ContentBlock.Type
				b.id = ev.ContentBlock.ID
				b.name = ev.ContentBlock.Name
			}
			blocks[ev.Index] = b
		case "content_block_delta":
			if ev.Index >= 0 && ev.Index < len(blocks) && blocks[ev.Index] != nil && ev.Delta != nil {
				switch ev.Delta.Type {
				case "text_delta":
					blocks[ev.Index].text.WriteString(ev.Delta.Text)
				case "input_json_delta":
					blocks[ev.Index].js.WriteString(ev.Delta.PartialJSON)
				}
			}
		case "message_delta":
			if ev.Delta != nil && ev.Delta.StopReason != "" {
				out.StopReason = ev.Delta.StopReason
			}
			// message_delta usage counters are cumulative totals — take the
			// latest non-zero values rather than summing.
			if ev.Usage != nil {
				if ev.Usage.OutputTokens > 0 {
					out.Usage.OutputTokens = ev.Usage.OutputTokens
				}
				if ev.Usage.InputTokens > 0 {
					out.Usage.InputTokens = ev.Usage.InputTokens
				}
				if ev.Usage.CacheCreationInputTokens > 0 {
					out.Usage.CacheCreationInputTokens = ev.Usage.CacheCreationInputTokens
				}
				if ev.Usage.CacheReadInputTokens > 0 {
					out.Usage.CacheReadInputTokens = ev.Usage.CacheReadInputTokens
				}
			}
		case "error":
			if ev.Error != nil {
				return nil, fmt.Errorf("anthropic: stream error: %s: %s", ev.Error.Type, ev.Error.Message)
			}
			return nil, fmt.Errorf("anthropic: stream error")
		case "message_stop":
			done = true
		}
	}
	if err := sc.Err(); err != nil {
		return nil, err // stream cut mid-response — retryable
	}
	if !done {
		return nil, fmt.Errorf("anthropic: stream ended without message_stop") // retryable
	}
	for _, b := range blocks {
		if b == nil {
			continue
		}
		switch b.typ {
		case "text":
			out.Content = append(out.Content, anthBlock{Type: "text", Text: b.text.String()})
		case "tool_use":
			blk := anthBlock{Type: "tool_use", ID: b.id, Name: b.name}
			if b.js.Len() > 0 {
				_ = json.Unmarshal([]byte(b.js.String()), &blk.Input)
			}
			out.Content = append(out.Content, blk)
		}
	}
	return out, nil
}

// text joins the response's text blocks.
func (r *anthResponse) text() string {
	var b bytes.Buffer
	for _, blk := range r.Content {
		if blk.Type == "text" {
			b.WriteString(blk.Text)
		}
	}
	return b.String()
}