claw-email/internal/ai_client/client.go

package ai_client

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"
	"time"

	"thuanle.me/claw-email-bridge/internal/config"
	"thuanle.me/claw-email-bridge/internal/database"
	"thuanle.me/claw-email-bridge/internal/logging"
	"gorm.io/gorm"
)

// Retry backoff schedule: 1s, 5s, 15s.
var retryBackoffs = []time.Duration{1 * time.Second, 5 * time.Second, 15 * time.Second}

// openClawRequest is the payload sent to OpenClaw API.
type openClawRequest struct {
	Input      string            `json:"input"`
	SessionID  string            `json:"session_id"`
	History    []historyEntry    `json:"history,omitempty"`
	CallbackURL string           `json:"callback_url"`
	Metadata   map[string]string `json:"metadata"`
}

type historyEntry struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

// Dispatcher sends tasks to OpenClaw and handles retry logic.
type Dispatcher struct {
	cfg        *config.Config
	db         *gorm.DB
	httpClient *http.Client
}

// NewDispatcher creates a dispatcher with a direct HTTP client (no system proxy).
func NewDispatcher(cfg *config.Config, db *gorm.DB) *Dispatcher {
	return &Dispatcher{
		cfg: cfg,
		db:  db,
		httpClient: &http.Client{
			Timeout: 30 * time.Second,
			Transport: &http.Transport{
				Proxy: nil, // Direct connection, no proxy inheritance.
			},
		},
	}
}

// Dispatch sends a task to OpenClaw with retry.
// This should be called asynchronously (go dispatcher.Dispatch(task)).
func (d *Dispatcher) Dispatch(task *database.Task) {
	log := logging.TaskLogger(task.TaskUUID, task.ThreadID, task.MessageID)

	// Build history from completed tasks in the same thread.
	history, err := d.buildHistory(task.ThreadID)
	if err != nil {
		log.Error("openclaw: failed to build history", "error", err)
		// Continue without history — non-fatal.
	}

	// Build callback URL.
	callbackURL := buildCallbackURL(d.cfg.CallbackBaseURL)

	payload := openClawRequest{
		Input:       task.BodyPlain,
		SessionID:   task.ThreadID,
		History:     history,
		CallbackURL: callbackURL,
		Metadata:    map[string]string{"task_uuid": task.TaskUUID},
	}

	// Update status to AI_PROCESSING before first attempt.
	task.Status = database.StatusAIProcessing
	if err := database.UpdateTask(d.db, task); err != nil {
		log.Error("openclaw: failed to update status to AI_PROCESSING", "error", err)
		return
	}

	// Retry loop: up to 3 attempts.
	for attempt := 1; attempt <= len(retryBackoffs); attempt++ {
		task.AttemptOpenClaw = attempt

		log.Info("openclaw: dispatching",
			"attempt", attempt,
			"status", task.Status,
		)

		err := d.callAPI(payload)
		if err == nil {
			// Success — task stays AI_PROCESSING, waiting for callback.
			log.Info("openclaw: dispatch successful", "attempt", attempt)
			if err := database.UpdateTask(d.db, task); err != nil {
				log.Error("openclaw: failed to save attempt count", "error", err)
			}
			return
		}

		log.Warn("openclaw: dispatch failed",
			"attempt", attempt,
			"error", err,
		)

		// Last attempt — mark as FAILED.
		if attempt == len(retryBackoffs) {
			errMsg := err.Error()
			task.Status = database.StatusFailed
			task.LastError = &errMsg
			if err := database.UpdateTask(d.db, task); err != nil {
				log.Error("openclaw: failed to save FAILED status", "error", err)
			}
			log.Error("openclaw: all retries exhausted",
				"attempt", attempt,
				"status", database.StatusFailed,
				"error", errMsg,
			)
			return
		}

		// Wait before retry.
		backoff := retryBackoffs[attempt-1]
		log.Info("openclaw: retrying after backoff", "backoff", backoff)
		time.Sleep(backoff)
	}
}

func buildCallbackURL(base string) string {
	trimmed := strings.TrimRight(base, "/")
	if trimmed == "" {
		trimmed = "http://localhost"
	}
	return trimmed + "/callback"
}

// callAPI makes a single HTTP POST to OpenClaw.
func (d *Dispatcher) callAPI(payload openClawRequest) error {
	body, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("marshal payload: %w", err)
	}

	req, err := http.NewRequest(http.MethodPost, d.cfg.OpenClawURL, bytes.NewReader(body))
	if err != nil {
		return fmt.Errorf("create request: %w", err)
	}
	req.Header.Set("Content-Type", "application/json")

	if d.cfg.OpenClawAPIKey != "" {
		req.Header.Set("Authorization", "Bearer "+d.cfg.OpenClawAPIKey)
	}

	resp, err := d.httpClient.Do(req)
	if err != nil {
		return fmt.Errorf("http request: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 200 && resp.StatusCode < 300 {
		return nil
	}

	// Read response body for error details (limit to 1KB).
	respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
	return fmt.Errorf("openclaw returned %d: %s", resp.StatusCode, string(respBody))
}

// buildHistory fetches completed tasks in the thread and formats them as history entries.
func (d *Dispatcher) buildHistory(threadID string) ([]historyEntry, error) {
	tasks, err := database.ThreadHistory(d.db, threadID)
	if err != nil {
		return nil, err
	}

	var history []historyEntry
	for _, t := range tasks {
		// User message.
		history = append(history, historyEntry{
			Role:    "user",
			Content: t.BodyPlain,
		})
		// AI response.
		if t.AIResponse != "" {
			history = append(history, historyEntry{
				Role:    "assistant",
				Content: t.AIResponse,
			})
		}
	}

	return history, nil
}