package daemon

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/url"
	"strings"
)

// AgentEvent is one event from the daemon's agent bus.
type AgentEvent struct {
	Progress *ToolProgress `json:"progress"`
	Type     string        `json:"type"`
	Session  string        `json:"session"`
	Parent   string        `json:"parent"`
	Name     string        `json:"name"`
	Model    string        `json:"model"`
	From     string        `json:"from"`
	To       string        `json:"to"`
	FromName string        `json:"fromName"`
	Kind     string        `json:"kind"`
	Text     string        `json:"text"`
	CallID   string        `json:"callId"`
	Output   string        `json:"output"`
	Source   string        `json:"source"`
	Depth    int           `json:"depth"`
	Bytes    int           `json:"bytes"`
	Running  bool          `json:"running"`
}

type Member struct {
	Session string `json:"session"`
	Parent  string `json:"parent"`
	Name    string `json:"name"`
	Depth   int    `json:"depth"`
	Closed  bool   `json:"closed"`
}

type ChildResult struct {
	Session Session `json:"session"`
	Member  Member  `json:"member"`
}

type AgentNode struct {
	Parent  *string `json:"parent"`
	Address *string `json:"address"`
	Session Session `json:"session"`
	Name    string  `json:"name"`
	Depth   int     `json:"depth"`
	Running bool    `json:"running"`
	Closed  bool    `json:"closed"`
}

type AgentsSnapshot struct {
	Root  string      `json:"root"`
	Nodes []AgentNode `json:"nodes"`
}

type ChildRequest struct {
	Name string `json:"name"`
	Task string `json:"task"`
}

// StreamAgents delivers the first batch, including an empty readiness batch,
// then nonempty batches until the stream ends or onBatch fails.
// The caller controls cancellation, including any blocking work in onBatch.
func StreamAgents(ctx context.Context, conn *Connection, onBatch func([]AgentEvent) error) error {
	operation := operation{Name: "stream agents", Method: http.MethodGet, Path: "/agents/stream", Policy: readRecovery}
	first := true
	err := scanEventStream(ctx, conn, operation, streamLimits{requireSSE: true, lineBytes: 8 * 1024 * 1024}, func(scanner *bufio.Scanner) error {
		for scanner.Scan() {
			line := scanner.Text()
			if !strings.HasPrefix(line, "data:") {
				continue
			}
			var batch struct {
				Events []json.RawMessage `json:"events"`
			}
			if err := json.Unmarshal([]byte(strings.TrimSpace(line[5:])), &batch); err != nil {
				return streamFailure(StreamProtocol, fmt.Errorf("invalid agent batch: %w", err))
			}
			if batch.Events == nil {
				return streamFailure(StreamProtocol, errors.New("invalid agent batch: missing events"))
			}
			events := make([]AgentEvent, 0, len(batch.Events))
			for _, raw := range batch.Events {
				event, err := decodeAgentEvent(raw)
				if err != nil {
					return streamFailure(StreamProtocol, err)
				}
				if event != nil {
					if event.Type == "overflow" && len(batch.Events) != 1 {
						return streamFailure(StreamProtocol, errors.New("overflow must be the sole agent event"))
					}
					events = append(events, *event)
				}
			}
			if len(events) == 0 && !first {
				continue
			}
			first = false
			if err := onBatch(events); err != nil {
				return streamFailure(StreamTerminal, err)
			}
			if len(events) == 1 && events[0].Type == "overflow" {
				return nil
			}
		}
		return nil
	})
	return classifyStreamFailure(err)
}

func CreateChild(ctx context.Context, conn *Connection, id string, body ChildRequest) (ChildResult, error) {
	var result ChildResult
	err := executeMutation(ctx, conn, operation{Name: "create child", Method: http.MethodPost, Path: sessionPath(id, "/children"), Body: body, Policy: authRecovery}, []int{201}, func(body []byte, _ int) error {
		var err error
		result, err = decodeChild(body)
		return err
	})
	return result, err
}

func GetAgents(ctx context.Context, conn *Connection, session string) (AgentsSnapshot, error) {
	var result AgentsSnapshot
	err := executeRead(ctx, conn, operation{Name: "get agents", Method: http.MethodGet, Path: "/agents?session=" + url.QueryEscape(session), Policy: readRecovery}, func(data []byte) error {
		fields, err := object(data)
		if err != nil {
			return err
		}
		if err = required(fields, "root", &result.Root); err != nil {
			return err
		}
		var rows []json.RawMessage
		if err = required(fields, "nodes", &rows); err != nil {
			return err
		}
		result.Nodes = make([]AgentNode, 0, len(rows))
		for _, row := range rows {
			fields, err := object(row)
			if err != nil {
				return err
			}
			var node AgentNode
			var sessionData json.RawMessage
			if err = nullable(fields, "parent", &node.Parent); err != nil {
				return err
			}
			if err = nullable(fields, "address", &node.Address); err != nil {
				return err
			}
			if err = required(fields, "session", &sessionData); err != nil {
				return err
			}
			node.Session, err = decodeSession(sessionData)
			if err != nil {
				return err
			}
			if err = required(fields, "name", &node.Name); err != nil {
				return err
			}
			if err = required(fields, "depth", &node.Depth); err != nil {
				return err
			}
			if node.Depth < 0 {
				return fieldError("depth")
			}
			if err = required(fields, "running", &node.Running); err != nil {
				return err
			}
			if err = required(fields, "closed", &node.Closed); err != nil {
				return err
			}
			result.Nodes = append(result.Nodes, node)
		}
		return nil
	})
	return result, err
}

func decodeAgentEvent(raw json.RawMessage) (*AgentEvent, error) {
	var header struct {
		Type string `json:"type"`
	}
	if err := json.Unmarshal(raw, &header); err != nil {
		return nil, fmt.Errorf("invalid agent event: %w", err)
	}
	if header.Type == "" {
		return nil, errors.New("invalid agent event: missing type")
	}
	switch header.Type {
	case "overflow":
		return &AgentEvent{Type: "overflow"}, nil
	case "spawn", "gone", "mail", "running", "text", "thinking", "arguments_delta", "tool_progress", "tool", "user", "message", "error", "interrupted", "progress", "renamed", "closed":
	default:
		return nil, nil
	}
	var event AgentEvent
	if err := json.Unmarshal(raw, &event); err != nil {
		return nil, fmt.Errorf("invalid %s agent event: %w", header.Type, err)
	}
	var fields map[string]json.RawMessage
	if err := json.Unmarshal(raw, &fields); err != nil {
		return nil, err
	}
	required := []string{"session"}
	switch event.Type {
	case "mail":
		required = []string{"to", "bytes", "kind"}
	case "spawn":
		required = append(required, "parent", "name", "model", "depth")
	case "running":
		required = append(required, "running")
	case "text", "thinking", "arguments_delta", "user", "message", "error", "progress":
		required = append(required, "text")
		if event.Type == "arguments_delta" {
			required = append(required, "callId", "name")
		}
	case "tool_progress":
		if err := validateToolProgress(fields["progress"]); err != nil {
			return nil, err
		}
		required = append(required, "progress")
	case "tool":
		required = append(required, "output")
	case "renamed":
		required = append(required, "name")
	}
	for _, name := range required {
		if value, exists := fields[name]; !exists || (string(value) == "null" && name != "progress") {
			return nil, fmt.Errorf("invalid %s agent event: missing %s", event.Type, name)
		}
	}
	return &event, nil
}

func decodeChild(data []byte) (ChildResult, error) {
	fields, err := object(data)
	if err != nil {
		return ChildResult{}, err
	}
	var result ChildResult
	raw, ok := fields["session"]
	if !ok {
		return result, fieldError("session")
	}
	result.Session, err = decodeSession(raw)
	if err != nil {
		return ChildResult{}, err
	}
	member, err := object(fields["member"])
	if err != nil {
		return ChildResult{}, err
	}
	for _, field := range []struct {
		target any
		name   string
	}{{&result.Member.Session, "session"}, {&result.Member.Parent, "parent"}, {&result.Member.Name, "name"}, {&result.Member.Depth, "depth"}, {&result.Member.Closed, "closed"}} {
		if err := required(member, field.name, field.target); err != nil {
			return ChildResult{}, err
		}
	}
	return result, nil
}