package agent

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

	"github.com/tidbcloud/AgentTeam/desktop/internal/agent/drivercontract"
)

// subscribeSSE retries setup failures until the first data frame. Once a stream
// has been established, any loss retires the runtime generation because
// OpenCode exposes no replay cursor.
func (d *OpenCodeDriver) subscribeSSE(ctx context.Context, cfg drivercontract.LaunchConfig, ready chan<- openCodeSSEReadyEvent) {
	attempt := 0
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}

		attempt++
		d.emitRuntimeDebug("opencode sse connect attempt", map[string]any{
			"agent_id": cfg.AgentID,
			"attempt":  attempt,
		})
		err := d.connectSSE(ctx, cfg, ready)
		if err != nil {
			if ctx.Err() != nil {
				d.emitRuntimeDebug("opencode sse subscription stopped", map[string]any{
					"agent_id": cfg.AgentID,
					"attempt":  attempt,
					"reason":   ctx.Err(),
				})
				return
			}
			if errors.Is(err, errOpenCodeSSEReplayGap) {
				// OpenCode's SSE endpoint exposes no replay cursor. Reconnecting
				// and accepting server.connected would not recover a lost
				// assistant-parent ACK or idle boundary, and could leave local
				// state permanently busy. Retire this runtime generation so the
				// durable ownership layer can fail closed instead.
				d.mu.Lock()
				serverCancel := d.beginOpenCodeGenerationRetirementLocked()
				d.mu.Unlock()
				d.emitRuntimeDebug("opencode SSE replay gap; retiring runtime generation", map[string]any{
					"agent_id": cfg.AgentID,
					"attempt":  attempt,
					"error":    err,
				})
				if serverCancel != nil {
					serverCancel()
				}
				return
			}
			d.emitRuntimeDebug("opencode sse connection lost; reconnecting", map[string]any{
				"agent_id": cfg.AgentID,
				"attempt":  attempt,
				"error":    err,
			})
			slog.Warn("opencode: SSE connection lost, reconnecting",
				"agent_id", cfg.AgentID,
				"error", err,
			)
			if err := waitOpenCodeRetry(ctx, time.Second, time.Second); err != nil {
				return
			}
			continue
		}
	}
}

// connectSSE establishes a single SSE connection and processes events.
// It uses a dedicated HTTP client without a timeout so the long-lived SSE
// stream is not interrupted.
func (d *OpenCodeDriver) connectSSE(ctx context.Context, cfg drivercontract.LaunchConfig, ready chan<- openCodeSSEReadyEvent) error {
	// Keep runtime mutation closed for the entire disconnected/connecting
	// interval. OpenCode SSE has no replay cursor, so accepting a prompt without
	// a live acknowledgement stream can strand its exact parent/status evidence.
	d.mu.Lock()
	d.sseDeliveryBlocked = true
	d.mu.Unlock()
	defer func() {
		d.mu.Lock()
		d.sseDeliveryBlocked = true
		d.mu.Unlock()
	}()

	startedAt := time.Now()
	d.emitRuntimeDebug("opencode sse connect started", map[string]any{
		"agent_id": cfg.AgentID,
		"path":     "/event",
	})
	req, err := d.newRequest(ctx, http.MethodGet, "/event", nil)
	if err != nil {
		signalOpenCodeSSEError(ready, err)
		d.emitRuntimeDebug("opencode sse request build failed", map[string]any{
			"agent_id": cfg.AgentID,
			"error":    err,
		})
		return err
	}
	req.Header.Set("Accept", "text/event-stream")
	req.Header.Set("Cache-Control", "no-cache")
	req.Header.Set("Connection", "keep-alive")
	d.addDirectoryHeader(req, cfg.WorkDir)

	// Dedicated client without timeout; the request context (sseCtx) handles
	// cancellation when the agent process exits.
	sseClient := &http.Client{Timeout: 0}
	resp, err := sseClient.Do(req)
	if err != nil {
		signalOpenCodeSSEError(ready, err)
		d.emitRuntimeDebug("opencode sse connect failed", map[string]any{
			"agent_id": cfg.AgentID,
			"duration": time.Since(startedAt).Round(time.Millisecond).String(),
			"error":    err,
		})
		return fmt.Errorf("SSE connect: %w", err)
	}
	defer resp.Body.Close()
	d.emitRuntimeDebug("opencode sse response received", map[string]any{
		"agent_id": cfg.AgentID,
		"status":   resp.StatusCode,
		"duration": time.Since(startedAt).Round(time.Millisecond).String(),
	})

	if resp.StatusCode != http.StatusOK {
		err := fmt.Errorf("SSE connect: status %d", resp.StatusCode)
		signalOpenCodeSSEError(ready, err)
		d.emitRuntimeDebug("opencode sse connect status failed", map[string]any{
			"agent_id": cfg.AgentID,
			"status":   resp.StatusCode,
		})
		return err
	}

	sawData, scanErr, contextErr := d.consumeOpenCodeSSEStream(ctx, resp.Body, cfg.AgentID, ready, startedAt)
	if contextErr != nil {
		return contextErr
	}

	if scanErr != nil {
		if !sawData {
			signalOpenCodeSSEError(ready, scanErr)
		}
		d.emitRuntimeDebug("opencode sse scanner failed", map[string]any{
			"agent_id":      cfg.AgentID,
			"saw_data":      sawData,
			"stream_uptime": time.Since(startedAt).Round(time.Millisecond).String(),
			"error":         scanErr,
		})
		if sawData {
			return fmt.Errorf("%w: %v", errOpenCodeSSEReplayGap, scanErr)
		}
		return scanErr
	}
	if !sawData {
		err := fmt.Errorf("SSE stream closed before first data frame")
		signalOpenCodeSSEError(ready, err)
		d.emitRuntimeDebug("opencode sse stream closed before first data", map[string]any{
			"agent_id": cfg.AgentID,
			"duration": time.Since(startedAt).Round(time.Millisecond).String(),
		})
		return err
	}
	d.emitRuntimeDebug("opencode sse stream closed", map[string]any{
		"agent_id":      cfg.AgentID,
		"stream_uptime": time.Since(startedAt).Round(time.Millisecond).String(),
	})
	return errOpenCodeSSEReplayGap
}

func (d *OpenCodeDriver) consumeOpenCodeSSEStream(
	ctx context.Context,
	body io.Reader,
	agentID string,
	ready chan<- openCodeSSEReadyEvent,
	startedAt time.Time,
) (bool, error, error) {
	scanner := bufio.NewScanner(body)
	scanner.Buffer(make([]byte, 512*1024), 512*1024)

	sawData := false
	for scanner.Scan() {
		select {
		case <-ctx.Done():
			return sawData, nil, ctx.Err()
		default:
		}

		line := scanner.Text()

		// opencode SSE wire format: "data: <json>\n\n"
		// where <json> is {"type":"<event-type>","properties":{...}}
		if !strings.HasPrefix(line, "data: ") {
			continue
		}

		data := strings.TrimPrefix(line, "data: ")
		if !sawData {
			d.emitRuntimeDebug("opencode sse first data received", map[string]any{
				"agent_id":   agentID,
				"duration":   time.Since(startedAt).Round(time.Millisecond).String(),
				"data_bytes": len(data),
			})
			d.mu.Lock()
			d.sseDeliveryBlocked = false
			d.mu.Unlock()
			signalOpenCodeSSEEvent(ready, openCodeSSEReadyEvent{ready: true})
			ready = nil
		}
		sawData = true
		d.handleSSEEvent(data)
	}
	return sawData, scanner.Err(), nil
}

type openCodeSSEReadyEvent struct {
	ready bool
	err   error
}

func waitForOpenCodeSSEReady(
	ctx context.Context,
	ready <-chan openCodeSSEReadyEvent,
	timeout time.Duration,
	process *openCodeServeProcess,
	stderrTail func() string,
) error {
	if ready == nil {
		return nil
	}
	timer := time.NewTimer(timeout)
	defer timer.Stop()
	var lastErr error
	select {
	case ev := <-ready:
		if ev.ready {
			return nil
		}
		if ev.err != nil {
			lastErr = ev.err
		}
	default:
	}
	for {
		select {
		case ev := <-ready:
			if ev.ready {
				return nil
			}
			if ev.err != nil {
				lastErr = ev.err
			}
		case <-ctx.Done():
			return ctx.Err()
		case <-openCodeServeExited(process):
			reason := "serve process exited before SSE subscription became ready: process exited successfully"
			if process != nil && process.waitErr != nil {
				reason = "serve process exited before SSE subscription became ready: " + process.waitErr.Error()
			}
			return openCodeHealthCheckError(reason, openCodeSSELastCheck(lastErr), stderrTail, nil)
		case <-timer.C:
			return openCodeHealthCheckError(fmt.Sprintf("timed out after %s", timeout), openCodeSSELastCheck(lastErr), stderrTail, nil)
		}
	}
}

func openCodeServeExited(process *openCodeServeProcess) <-chan struct{} {
	if process == nil || process.exited == nil {
		return nil
	}
	return process.exited
}

func openCodeSSELastCheck(err error) string {
	if err == nil {
		return ""
	}
	detail := strings.TrimSpace(err.Error())
	if strings.HasPrefix(detail, "SSE connect:") {
		return detail
	}
	return "SSE connect: " + detail
}

func signalOpenCodeSSEError(ready chan<- openCodeSSEReadyEvent, err error) {
	if err == nil {
		return
	}
	signalOpenCodeSSEEvent(ready, openCodeSSEReadyEvent{err: err})
}

func signalOpenCodeSSEEvent(ready chan<- openCodeSSEReadyEvent, ev openCodeSSEReadyEvent) {
	if ready == nil {
		return
	}
	select {
	case ready <- ev:
	default:
	}
}

// newRequest creates an HTTP request with Basic Auth headers.
func (d *OpenCodeDriver) newRequest(ctx context.Context, method, path string, body io.Reader) (*http.Request, error) {
	d.mu.Lock()
	serverURL := d.serverURL
	password := d.password
	d.mu.Unlock()

	url := serverURL + path
	req, err := http.NewRequestWithContext(ctx, method, url, body)
	if err != nil {
		return nil, err
	}

	auth := base64.StdEncoding.EncodeToString([]byte(openCodeServerUsername + ":" + password))
	req.Header.Set("Authorization", "Basic "+auth)

	return req, nil
}

// addDirectoryHeader sets the x-opencode-directory header for workspace routing.
// The opencode server resolves the workspace from this header (or the ?directory
// query param, or process.cwd() as fallback).
func (d *OpenCodeDriver) addDirectoryHeader(req *http.Request, workDir string) {
	if strings.TrimSpace(workDir) != "" {
		req.Header.Set("x-opencode-directory", workDir)
	}
}

// getWorkDir returns the workspace directory under lock.
func (d *OpenCodeDriver) getWorkDir() string {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.workDir
}